Blame for app/repost_engine.py

"""Repost a whole channel in order.

Every post is copied to the end of the channel (silently), in the same order as the original:
  1. a post that needs no change (no buttons, no link to swap) is copied by Telegram itself
     (messages.forwardMessages with drop_author): text, formatting, quotes, media and albums stay exact;
  2. a post with buttons, or with links to swap, is created again in ONE request that carries its final
     text, formatting and buttons - exactly how the bot publishes a post of its own - and is then read back
     to make sure the buttons really are there (a copy that is edited afterwards can end up without them);
  3. if the channel forbids forwarding, every post is created like in 2.
Links can be swapped on the way, the new posts are registered in the database, and the old posts are removed
later in a separate step. Nothing here deletes anything unless asked to (delete_copies).

What a plain copy would lose is put back as well:
  * a post that answers another post is posted as an answer to the COPY of that post (a reply can only be set
    when a post is created, so such a post is always created again instead of copied);
  * a link to another post of the channel (t.me/name/115) is pointed at the copy of that post - at once when the
    copy already exists, afterwards (finalize_repost) for posts that are copied later;
  * a post that was pinned has its copy pinned at the end.

A post that TELEGRAM restricts (a copyright strike: the whole channel is "unavailable due to copyright infringement",
or the post carries a restriction mark, or Telegram's apps show "This message couldn't be displayed on your device due
to copyright infringement" instead of the post) is not copied as Telegram hands it out now. /shift rebuilds such a post
from what My posts saved of it (text, formatting, buttons, media) - for a restricted CHANNEL every post that has a saved
copy, however intact Telegram's version looks. A post without a saved copy is copied as Telegram hands it out when that
is the post itself, and skipped (and reported) when it is only Telegram's notice. Telegram's notice is never posted, from
whichever source. /repost leaves a post whose content Telegram holds back alone.
Every copy of a restricted source is read back: if Telegram itself hides the new post, that is reported (copy_loop).
"""
from __future__ import annotations

import asyncio
import copy
import logging
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Awaitable, Callable, Optional

from telethon import errors, functions, types, utils

from .buttons_sync import ButtonsNotShown, ensure_markup, link_urls
from .crosslinks import CrossResult, relink_after_repost
from .linkswap import compute_message_changes, url_only_markup
from .postlinks import PostLinks, relink_message
from .replace_engine import CHUNK, FALLBACK_EMPTY_BATCHES, fetch_messages, is_skippable, preview_flag
from .restrictions import PLACEHOLDER_NOTE, is_placeholder_text, message_restriction, probe_channel
from .richmsg import Rich, entry as rich_entry, is_rich_message, plain_of, rich_from_received, rich_of, with_rich
from .tgutil import (
    FATAL,
    build_markup,
    classify_media,
    de_entities,
    edit_raw,
    flood_retry,
    media_from_ref,
    media_ref,
    peer_of,
    ser_entities,
    ser_markup,
)

log = logging.getLogger(__name__)

ABORT_AFTER = 8  # consecutive posts that could not be copied -> stop


@dataclass
class RepostOptions:
    old: Optional[str] = None
    new: Optional[str] = None
    include_typed: bool = True
    include_posts: bool = False
    last: Optional[int] = None  # trial run: only the newest N post ids

    @property
    def swaps(self) -> bool:
        return bool(self.old and self.new)


@dataclass
class Final:
    """What a copy of a post should look like."""

    text: str
    entities: list
    markup: Optional[types.ReplyInlineMarkup]
    preview: Optional[bool]
    text_changed: bool
    n_buttons: int = 0
    n_links: int = 0
    n_typed: int = 0
    dropped: int = 0  # other bots' buttons that are not copied
    n_post_links: int = 0  # links to other posts of the same channel (whatever their new id will be)
    n_relinked: int = 0  # of those, how many were already pointed at a copy
    rich: Optional[Rich] = None  # the post is a rich message: this is what the copy is made from (text / entities are empty)


def rich_for(src) -> Optional[Rich]:
    """The rich message that `src` is - a live post, or a stand-in made from a saved post - or None for an ordinary post."""
    saved = getattr(src, "rich_saved", None)
    if saved is not None:
        return saved
    rm = getattr(src, "rich_message", None)
    return rich_from_received(rm) if rm is not None else None


def final_state(src, opts: RepostOptions, links: Optional[PostLinks] = None) -> Final:
    """What the copy of `src` should look like. `links` (optional) points links to other posts of the channel at
    the copies that exist so far (links.ids: old message id -> new one)."""
    change = None
    if opts.swaps:
        change, _ = compute_message_changes(
            src, opts.old, opts.new, include_typed=opts.include_typed, include_posts=opts.include_posts
        )
    rich = rich_for(src)
    text = change.new_text if change else (src.message or "")
    entities = list(change.new_entities) if change else list(src.entities or [])
    if rich is not None:  # a rich message's text is blocks: nothing of it is changed here, only its buttons
        text, entities = "", []
    # text mentions need an access hash to be re-sent; they are rare in channel posts
    entities = [e for e in entities if not isinstance(e, types.MessageEntityMentionName)]
    markup, dropped = url_only_markup(
        src.reply_markup, opts.old if opts.swaps else None, opts.new if opts.swaps else None, opts.include_posts
    )
    n_post_links = n_relinked = 0
    text_changed = bool(change and change.entities_changed)
    if links is not None:
        n_post_links = links.count(text, entities, markup)
        if n_post_links and links.ids:
            text, entities, n_typed, n_hyper = links.remap_text(text, entities)
            markup, n_buttons = links.remap_markup(markup)
            n_relinked = n_typed + n_hyper + n_buttons
            text_changed = text_changed or bool(n_typed or n_hyper)
    # Send exactly what is saved: go through the stored form that "My posts" also uses to build its edits.
    markup = build_markup(ser_markup(markup)) if markup is not None else None
    entities = de_entities(ser_entities(entities))
    return Final(
        text=text,
        entities=entities,
        markup=markup,
        preview=preview_flag(src),
        text_changed=text_changed,
        n_buttons=change.n_buttons if change else 0,
        n_links=change.n_links if change else 0,
        n_typed=change.n_typed if change else 0,
        dropped=dropped,
        n_post_links=n_post_links,
        n_relinked=n_relinked,
        rich=rich,
    )


# ---------------------------------------------------------------------------- replies and pins
class ReplyParentMissing(RuntimeError):
    """A post answers another post whose copy does not exist (yet): copying it now would lose the answer."""


def reply_parent(msg) -> Optional[int]:
    """The message id a post answers (in the same channel), or None."""
    h = getattr(msg, "reply_to", None)
    if h is None or getattr(h, "reply_to_peer_id", None) is not None or getattr(h, "forum_topic", None):
        return None
    return getattr(h, "reply_to_msg_id", None) or None


def new_parent(msg, idmap: dict, seen: set) -> Optional[int]:
    """The copy of the post `msg` answers. None when it answers nothing, or the post it answered is gone / outside
    the range being copied. Raises ReplyParentMissing when that post was read in this run but has no copy."""
    pid = reply_parent(msg)
    if pid is None:
        return None
    if pid in idmap:
        return idmap[pid]
    if pid in seen:
        raise ReplyParentMissing(f"post {msg.id} answers post {pid}, which has not been copied")
    return None


def new_messages_from(result) -> list:
    """The messages created by a send / forward request, oldest first."""
    out = []
    for u in getattr(result, "updates", None) or []:
        if isinstance(u, (types.UpdateNewChannelMessage, types.UpdateNewMessage)):
            out.append(u.message)
    out.sort(key=lambda m: m.id)
    return out


# ------------------------------------------------------ posts Telegram holds back (copyright strike ...)
SavedLookup = Callable[[list], Awaitable[dict]]  # message ids -> {message id: the saved post of My posts}


def saved_lookup(db, channel_id: int) -> SavedLookup:
    """How the copy loop and the plan look up what My posts saved of the posts of one channel."""

    async def lookup(ids: list) -> dict:
        return {p.message_id: p for p in await db.posts_at(channel_id, ids)}

    return lookup


def usable_row(row) -> bool:
    """A saved post that can take the place of a restricted post: it has something to post, and what it says is not
    Telegram's notice (a saved copy that an older version made from a restricted post can hold the notice)."""
    if row is None:
        return False
    text = row.text or ""
    if is_placeholder_text(text):
        return False
    return bool(text.strip() or row.media_file_id)


def copy_source(unit: list, rows: dict, *, restricted: Optional[str] = None, use_saved: bool = False) -> tuple:
    """Where the copy of this post (or album) comes from -> (source, why).

      "live"  - as Telegram hands the post out: nothing restricts it, or Telegram restricts it but still hands out the
                post itself and there is no saved copy to use instead;
      "saved" - from what My posts saved of it: Telegram restricts the channel (`restricted`) or the post (a copyright
                strike) and My posts has every part of it. Only when `use_saved` (/shift) - the saved copy wins even when
                Telegram's version still looks intact, because that is the owner's rule for a struck channel;
      "skip"  - not copied: what Telegram hands out is its notice - or nothing - and My posts can't take over.

    `why` is Telegram's own wording about the restriction, None when nothing restricts the post."""
    marked = next((w for w in (message_restriction(m) for m in unit) if w), None)
    why = marked or restricted
    if why is None:
        return "live", None
    if use_saved and all(usable_row(rows.get(m.id)) for m in unit):
        return "saved", why
    for m in unit:
        live = m.message or ""
        if is_placeholder_text(live) or (
            not live.strip() and getattr(m, "media", None) is None and rich_for(m) is None
        ):
            return "skip", why
    return "live", why


def stand_in(m, row):
    """A copy of the live message `m` that shows what My posts saved for it - text, formatting, buttons, media. It takes
    the place of a post Telegram holds back (the live message keeps what Telegram still tells: id, album, reply)."""
    s = copy.copy(m)
    s.message = row.text or ""
    s.entities = de_entities(row.entities)
    s.rich_message = None  # what it shows is what My posts saved ...
    s.rich_saved = rich_of(row)  # ... which is a rich message when the saved post is one
    s.reply_markup = build_markup(row.buttons)
    s.media = None
    s.saved_ref = None  # the saved media reference (resend_saved may need it again without its file reference)
    if row.media_file_id:
        try:
            s.media = media_from_ref(row.media_file_id)
            s.saved_ref = row.media_file_id
        except ValueError:  # a reference that can't be read: the post goes out without media
            log.warning("the saved media of post %s can't be read", m.id)
    if s.media is None and row.link_preview:
        s.media = types.MessageMediaWebPage(webpage=types.WebPageEmpty(id=0))
    return s


def stand_ins(unit: list, rows: dict) -> Optional[list]:
    """The saved copies that stand in for a restricted post, or None when My posts does not have all of them (or one of
    them has nothing to post, or holds only Telegram's notice)."""
    out = []
    for m in unit:
        row = rows.get(m.id)
        if not usable_row(row):
            return None
        out.append(stand_in(m, row))
    return out


FILE_MEDIA = (types.MessageMediaPhoto, types.MessageMediaDocument)  # a photo, video, document, voice, sticker ...


def has_file(m) -> bool:
    """The post carries a file. A link preview, a poll, a location or a contact is not a file."""
    return isinstance(getattr(m, "media", None), FILE_MEDIA)


def without_files(unit: list) -> tuple:
    """/shift --no-media: the posts of `unit` as they look without their files -> (those that still have a text, the
    number of files left out). A post that was only a file has nothing left to post and is dropped."""
    out, lost = [], 0
    for m in unit:
        c = m
        if has_file(m):
            c = copy.copy(m)
            c.media, c.saved_ref, c.invert_media = None, None, None
            lost += 1
        if (c.message or "").strip() or rich_for(c) is not None:
            out.append(c)
    return out, lost


async def single_posts(units: AsyncIterator[list]) -> AsyncIterator[list]:
    """/shift --no-media: an album can't be rebuilt without its pictures, so every message of it is a post of its own."""
    async for unit in units:
        for m in unit:
            yield [m]


def saved_message(row, channel_id: int):
    """A post built from its saved copy alone, for a channel that can't be read at all. Only the saved parts exist: no
    album, no reply, no pin."""
    base = types.Message(
        id=row.message_id, peer_id=types.PeerChannel(channel_id), date=row.sent_at or row.created_at, message=""
    )
    return stand_in(base, row)


def saved_rows_between(rows: list, first: Optional[int], last: Optional[int]) -> list:
    """The saved posts (oldest first, one per message id) that can be posted - they have something to post that is not
    Telegram's notice - between two message ids."""
    out, seen = [], set()
    for row in sorted(rows, key=lambda r: (r.message_id or 0, r.id)):
        mid = row.message_id
        if mid is None or mid in seen or (first and mid < first) or (last and mid > last):
            continue
        if usable_row(row):
            seen.add(mid)
            out.append(row)
    return out


async def saved_units(db, channel_id: int, first: int, last: int) -> AsyncIterator[list]:
    """The saved posts of a channel between two message ids, oldest first - each one a unit of its own."""
    for row in saved_rows_between(await db.sent_posts_from(channel_id, first), first, last):
        yield [saved_message(row, channel_id)]


# ------------------------------------------------------------------------------- planning
@dataclass
class RepostPlan:
    channel: Any
    first: int = 1
    last: int = 0  # newest existing message id when the plan was made
    scanned: int = 0
    units: int = 0  # posts (an album counts once)
    messages: int = 0
    albums: int = 0
    text: int = 0
    media: int = 0
    polls: int = 0
    other: int = 0
    service: int = 0
    link_posts: int = 0
    n_buttons: int = 0
    n_links: int = 0
    n_typed: int = 0
    dropped_posts: int = 0
    post_link_posts: int = 0  # posts with a link to another post of this channel
    post_links: int = 0
    pinned: int = 0  # pinned posts (their copies get pinned)
    replies: int = 0  # posts that answer another post
    held_saved: int = 0  # restricted posts that are copied from what My posts saved of them (counted in units)
    held_skipped: int = 0  # restricted posts that can't be copied: only Telegram's notice is left (not counted in units)
    live_restricted: int = 0  # restricted posts without a saved copy that are copied as Telegram hands them out
    no_media: bool = False  # /shift --no-media: the files are left out
    files_left_out: int = 0  # ... posts that are copied as text without their file (counted in units)
    file_only: int = 0  # ... posts that are only a file: nothing to copy (not counted in units)
    held_why: Optional[str] = None  # what Telegram says about them
    channel_restricted: Optional[str] = None  # what Telegram says when it restricts the whole channel
    from_db: bool = False  # the channel can't be read: the plan (and the copy) is made from My posts alone
    unreadable: Optional[str] = None  # ... because of this
    partial: bool = False
    error: Optional[str] = None

    def eta_seconds(self, delay: float) -> int:
        return int(self.units * (delay + 1.0))


def tally(plan: RepostPlan, m, opts: RepostOptions, links: PostLinks) -> None:
    """Count one message (as it will be copied) into the plan."""
    plan.messages += 1
    media = m.media
    if media is None or isinstance(media, types.MessageMediaWebPage):
        plan.text += 1
    elif isinstance(media, (types.MessageMediaPhoto, types.MessageMediaDocument)):
        plan.media += 1
    elif isinstance(media, types.MessageMediaPoll):
        plan.polls += 1
    else:
        plan.other += 1
    f = final_state(m, opts, links)
    if f.n_buttons or f.n_links or f.n_typed:
        plan.link_posts += 1
        plan.n_buttons += f.n_buttons
        plan.n_links += f.n_links
        plan.n_typed += f.n_typed
    if f.dropped:
        plan.dropped_posts += 1
    if f.n_post_links:
        plan.post_link_posts += 1
        plan.post_links += f.n_post_links
    if getattr(m, "pinned", False):
        plan.pinned += 1
    if reply_parent(m) is not None:
        plan.replies += 1


def plan_from_saved(
    ch, rows: list, opts: RepostOptions, *, first_id=None, last_id=None, unreadable=None, no_media: bool = False
) -> RepostPlan:
    """The plan of a copy made from My posts alone (the channel can't be read): one post per saved message."""
    plan = RepostPlan(
        channel=ch, partial=bool(first_id or last_id), from_db=True, unreadable=unreadable, no_media=no_media
    )
    links = PostLinks.for_channel(ch)
    chosen = saved_rows_between(rows, first_id, last_id)
    for row in chosen:
        m = saved_message(row, ch.id)
        if no_media:
            kept, lost = without_files([m])
            if not kept:
                plan.file_only += 1
                continue
            m, plan.files_left_out = kept[0], plan.files_left_out + lost
        plan.units += 1
        tally(plan, m, opts, links)
    if chosen:
        plan.first, plan.last = chosen[0].message_id, chosen[-1].message_id
        plan.scanned = plan.last
    else:
        plan.first, plan.last = first_id or 1, last_id or 0
    if not plan.units:
        plan.error = "There is nothing to copy."
        if plan.file_only:
            plan.error = "Every saved post in this range is only a file, so without files there is nothing to copy."
    return plan


async def find_last_id(client, peer, top: int) -> Optional[int]:
    """Newest existing message id, searching backwards from the channel counter."""
    a = top
    for _ in range(300):
        lo = max(1, a - CHUNK + 1)
        msgs = await fetch_messages(client, peer, list(range(lo, a + 1)))
        ids = [m.id for m in msgs if m is not None and not isinstance(m, types.MessageEmpty)]
        if ids:
            return max(ids)
        if lo == 1:
            return None
        a = lo - 1
    return None


async def plan_repost(
    client,
    ch,
    opts: RepostOptions,
    *,
    progress: Optional[Callable[[RepostPlan], Awaitable[None]]] = None,
    first_id: Optional[int] = None,
    last_id: Optional[int] = None,
    saved: Optional[SavedLookup] = None,
    use_saved: bool = False,
    no_media: bool = False,
) -> RepostPlan:
    """Read the channel and count what a copy would do. first_id / last_id limit the range (used by /shift).
    `saved` looks up My posts' copies of the channel's posts; with `use_saved` a post Telegram holds back is counted as
    the saved post it will be copied from (/shift), without it such a post is left out (/repost). With `no_media` the
    files are left out of the copies (/shift --no-media): a post that is only a file has nothing left to copy."""
    peer = peer_of(ch)
    plan = RepostPlan(channel=ch, partial=bool(opts.last), no_media=no_media)
    try:
        top, plan.channel_restricted, _ = await probe_channel(client, peer)
    except Exception as e:  # the counter is only a help: without it the ids are read until nothing more turns up
        log.info("could not read channel counter: %s", e)
        top = None
    first = 1
    if opts.last:
        if top is None:
            plan.error = "I can't tell which post is the newest, so --last can't be used here."
            return plan
        newest = await find_last_id(client, peer, top)
        if newest is None:
            plan.error = "The channel has no posts."
            return plan
        first, top = max(1, newest - opts.last + 1), newest
    if first_id:
        first = max(first, first_id)
        plan.partial = True
    if last_id:
        top = last_id if top is None else min(top, last_id)
        plan.partial = True
    plan.first = first
    links = PostLinks.for_channel(ch)
    a, empty_run, prev_gid, max_seen = first, 0, None, 0
    while True:
        if top is not None and a > top:
            break
        hi = a + CHUNK - 1 if top is None else min(a + CHUNK - 1, top)
        found = False
        rows = await saved(list(range(a, hi + 1))) if saved else {}
        for m in await fetch_messages(client, peer, list(range(a, hi + 1))):
            if m is None or isinstance(m, types.MessageEmpty):
                continue
            found = True
            max_seen = max(max_seen, m.id)
            if isinstance(m, types.MessageService):
                plan.service += 1
                continue
            gid = getattr(m, "grouped_id", None)
            new_unit = no_media or gid is None or gid != prev_gid  # without files an album is a post per message
            row = rows.get(m.id)
            # a struck channel changes how /shift copies (My posts first); /repost copies as before, post by post
            source, why = copy_source(
                [m], rows, restricted=plan.channel_restricted if use_saved else None, use_saved=use_saved
            )
            if why:
                plan.held_why = plan.held_why or why
            if source == "skip":
                plan.held_skipped += 1 if new_unit else 0  # nothing to copy it from
                prev_gid = gid
                continue
            if source == "saved":
                m = stand_in(m, row)  # counted as the saved post it will be copied from
            if no_media:
                kept, lost = without_files([m])
                if not kept:  # nothing but a file: nothing to copy without it
                    plan.file_only += 1
                    prev_gid = gid
                    continue
                m, plan.files_left_out = kept[0], plan.files_left_out + lost
            if source == "saved":
                plan.held_saved += 1 if new_unit else 0
            elif why:
                plan.live_restricted += 1 if new_unit else 0
            if new_unit:
                plan.units += 1
                if gid is not None and not no_media:
                    plan.albums += 1
            prev_gid = gid
            tally(plan, m, opts, links)
        plan.scanned = hi
        if top is None:
            empty_run = 0 if found else empty_run + 1
            if empty_run >= FALLBACK_EMPTY_BATCHES:
                break
        if progress:
            await progress(plan)
        a = hi + 1
    plan.last = max_seen
    if not plan.units:
        plan.error = "There is nothing to copy."
        if plan.held_skipped:
            plan.error = (
                "Telegram holds back every post in this range (it shows a notice instead of the post) and My posts has "
                "no saved copy of them, so there is nothing to copy."
            )
        elif plan.file_only:
            plan.error = "Every post in this range is only a file, so without files there is nothing to copy."
    return plan


# ---------------------------------------------------------------------------------- copying
async def iter_units(client, peer, first: int, last: int) -> AsyncIterator[list]:
    """The posts between two ids, oldest first. An album is one unit (a list of messages)."""
    group: list = []
    a = first
    while a <= last:
        hi = min(a + CHUNK - 1, last)
        for m in await fetch_messages(client, peer, list(range(a, hi + 1))):
            if is_skippable(m):
                continue
            gid = getattr(m, "grouped_id", None)
            if group and gid != group[0].grouped_id:
                yield group
                group = []
            group.append(m)
            if gid is None:
                yield group
                group = []
        a = hi + 1
    if group:
        yield group


async def delete_ids(client, peer, ids: list) -> None:
    if not ids:
        return
    try:
        await flood_retry(lambda: client(functions.channels.DeleteMessagesRequest(channel=peer, id=list(ids))))
    except errors.RPCError as e:
        log.warning("could not remove messages %s: %s", ids, type(e).__name__)


async def forward_unit(client, src_peer, unit: list, dst_peer=None) -> list:
    """Let Telegram copy the posts (no "forwarded from" header). `dst_peer` defaults to the same channel."""
    ids = [m.id for m in unit]
    to_peer = dst_peer if dst_peer is not None else src_peer
    result = await flood_retry(
        lambda: client(
            functions.messages.ForwardMessagesRequest(
                from_peer=src_peer, id=ids, to_peer=to_peer, silent=True, drop_author=True
            )
        )
    )
    return new_messages_from(result)


def _inverted(m) -> Optional[bool]:
    return True if getattr(m, "invert_media", False) else None


def reply_header(reply_to: Optional[int]):
    """The `reply_to` argument of the send requests: answer message `reply_to` of the same channel."""
    return types.InputReplyToMessage(reply_to_msg_id=reply_to) if reply_to else None


async def resend_unit(client, peer, unit: list, finals: list, reply_to: Optional[int] = None) -> list:
    """Rebuild a post from its parts (used when copying is not allowed, or the copy must differ from the original).
    `reply_to`: message id (in the channel `peer`) the new post answers."""
    reply = reply_header(reply_to)
    sender = None
    if len(unit) == 1 and getattr(finals[0], "rich", None) is not None:
        rich_final = finals[0]  # a rich message: Telegram builds it again from its blocks (or its HTML / Markdown)

        async def send_rich_final():
            return await with_rich(
                rich_final.rich,
                lambda inp: client(
                    functions.messages.SendMessageRequest(
                        peer=peer, message="", rich_message=inp, reply_markup=rich_final.markup, silent=True, reply_to=reply
                    )
                ),
            )

        sender = send_rich_final
    elif len(unit) > 1:
        singles = [
            types.InputSingleMedia(media=utils.get_input_media(m.media), message=f.text, entities=f.entities or None)
            for m, f in zip(unit, finals)
        ]

        def make():
            return functions.messages.SendMultiMediaRequest(
                peer=peer, multi_media=singles, silent=True, invert_media=_inverted(unit[0]), reply_to=reply
            )

    else:
        m, f = unit[0], finals[0]
        if m.media is None or isinstance(m.media, types.MessageMediaWebPage):

            def make():
                return functions.messages.SendMessageRequest(
                    peer=peer,
                    message=f.text,
                    entities=f.entities or None,
                    reply_markup=f.markup,
                    no_webpage=True if f.preview is False else None,
                    invert_media=_inverted(m),
                    silent=True,
                    reply_to=reply,
                )

        else:
            media = utils.get_input_media(m.media)

            def make():
                return functions.messages.SendMediaRequest(
                    peer=peer,
                    media=media,
                    message=f.text,
                    entities=f.entities or None,
                    reply_markup=f.markup,
                    invert_media=_inverted(m),
                    silent=True,
                    reply_to=reply,
                )

    result = await flood_retry(sender if sender is not None else (lambda: client(make())))
    news = new_messages_from(result)
    if len(news) != len(unit):
        await delete_ids(client, peer, [n.id for n in news])
        raise RuntimeError(f"Telegram created {len(news)} messages instead of {len(unit)}")
    return news


@dataclass
class _State:
    mode: str = "forward"  # becomes "resend" when the channel forbids forwarding
    forwarded: int = 0
    rebuilt: int = 0
    repaired: int = 0  # new posts whose buttons had to be set again after posting
    replies: int = 0  # new posts created as an answer to the copy of the post the original answered
    reply_lost: int = 0  # answers that could only be copied without the reply
    from_saved: int = 0  # posts Telegram holds back that were rebuilt from their saved copy in My posts
    media_lost: list = field(default_factory=list)  # old ids of those that had to go out without their media


def is_pure(src, f: Final, reply_to: Optional[int] = None) -> bool:
    """True when a server-side copy is already the final post: nothing to swap, no keyboard to rebuild and no
    reply to set (a reply can only be given when a post is created)."""
    return not f.text_changed and src.reply_markup is None and not reply_to


def _has_media(unit: list) -> bool:
    return any(m.media is not None and not isinstance(m.media, types.MessageMediaWebPage) for m in unit)


def _without_reference(unit: list) -> list:
    """The same saved posts with the file reference of their media left out (it may have expired)."""
    out = []
    for m in unit:
        b = copy.copy(m)
        if getattr(m, "saved_ref", None):
            b.media = media_from_ref(m.saved_ref, blank_reference=True)
        out.append(b)
    return out


async def resend_saved(client, dst_peer, unit: list, finals: list, st: _State, reply_to: Optional[int] = None) -> list:
    """Create the copies of posts that Telegram holds back, from the saved posts that stand in for them (stand_in).
    Their media is sent from the saved reference: when Telegram says the reference is out of date it is tried without
    it, and a single post whose text is saved goes out without its media when Telegram still refuses the file
    (a file taken down for copyright is refused) - counted in st.media_lost."""
    failure: Optional[BaseException] = None
    for attempt in (unit, None):
        if attempt is None:
            if failure is None or not _has_media(unit):
                break
            attempt = _without_reference(unit)
        try:
            return await resend_unit(client, dst_peer, attempt, finals, reply_to)
        except (TypeError, ValueError, errors.RPCError) as e:
            if isinstance(e, errors.RPCError) and type(e).__name__ in FATAL:
                raise
            failure = failure or e
            if not _has_media(unit):
                raise
    if len(unit) == 1 and (finals[0].text or "").strip():
        unit[0].media = None  # what goes out now has no media, and neither has its entry in My posts
        news = await resend_unit(client, dst_peer, unit, finals, reply_to)
        st.media_lost.append(unit[0].id)
        return news
    raise failure  # type: ignore[misc]


async def copy_unit(
    client,
    peer,
    unit: list,
    finals: list,
    st: _State,
    reply_to: Optional[int] = None,
    dst_peer=None,
    *,
    from_saved: bool = False,
    no_forward: bool = False,
) -> list:
    """Create the copy of one post (or album) at the end of `dst_peer` (default: the same channel `peer`).
    `from_saved`: the posts in `unit` are saved copies standing in for posts Telegram restricts - never forwarded.
    `no_forward`: Telegram restricts these posts and refuses to copy them itself: they are rebuilt from their parts."""
    dst = dst_peer if dst_peer is not None else peer
    if from_saved:
        news = await resend_saved(client, dst, unit, finals, st, reply_to)
        st.rebuilt += 1
        st.from_saved += 1
        if reply_to:
            st.replies += 1
        return news
    pure = all(is_pure(m, f, reply_to) for m, f in zip(unit, finals))
    if st.mode == "forward" and pure and not no_forward:
        try:
            news = await forward_unit(client, peer, unit, dst)
            if len(news) == len(unit):
                st.forwarded += 1
                return news
            await delete_ids(client, dst, [n.id for n in news])
        except errors.ChatForwardsRestrictedError:
            st.mode = "resend"
            log.info("this channel restricts forwarding - rebuilding posts from their parts instead")
        except errors.RPCError as e:
            if type(e).__name__ in FATAL:
                raise
            log.info("copy failed (%s) - rebuilding the post", type(e).__name__)
    try:
        news = await resend_unit(client, dst, unit, finals, reply_to)
    except (TypeError, ValueError, errors.RPCError) as e:
        # A kind of media a post cannot be rebuilt from (or a reply Telegram refuses). If only the reply made a
        # rebuild necessary, a plain copy by Telegram still works - without the reply.
        if isinstance(e, errors.RPCError) and type(e).__name__ in FATAL:
            raise
        if reply_to and st.mode == "forward" and not no_forward and all(is_pure(m, f) for m, f in zip(unit, finals)):
            news = await forward_unit(client, peer, unit, dst)
            if len(news) != len(unit):
                await delete_ids(client, dst, [n.id for n in news])
                raise
            st.reply_lost += 1
            st.forwarded += 1
            return news
        raise
    st.rebuilt += 1
    if reply_to:
        st.replies += 1
    return news


async def settle_buttons(client, peer, finals: list, news: list, st: _State, *, pause: float) -> None:
    """Read the new posts back: each must show the link buttons it should. Repairs it, or removes the copies."""
    for f, new in zip(finals, news):
        if not link_urls(f.markup):
            continue
        try:
            how = await ensure_markup(client, peer, new.id, f.markup, pause=pause)
        except errors.RPCError as e:
            await delete_ids(client, peer, [n.id for n in news])
            if type(e).__name__ in FATAL:
                raise
            raise ButtonsNotShown(f"could not set the buttons ({type(e).__name__})") from e
        except (ButtonsNotShown, LookupError) as e:
            await delete_ids(client, peer, [n.id for n in news])
            raise ButtonsNotShown(str(e)) from e
        if how != "fine":
            st.repaired += 1


@dataclass
class FinalizeResult:
    """The finishing touches after the copies exist: links to other posts, pins."""

    checked: int = 0  # copies read back
    relinked: int = 0  # copies whose links to other posts now point at the new copies
    links: int = 0  # links changed in total
    missing: int = 0  # copies that do not exist any more
    failed: list = field(default_factory=list)  # [(message id, error name)]
    pin_wanted: int = 0  # copies of pinned posts that still had to be pinned
    pinned: int = 0
    pin_error: Optional[str] = None
    aborted: Optional[str] = None
    stopped: bool = False


@dataclass
class RepostResult:
    copied_units: int = 0
    copied_messages: int = 0
    skipped_done: int = 0
    failed: list = field(default_factory=list)  # [(first old id, error name)]
    forwarded: int = 0
    rebuilt: int = 0
    repaired: int = 0
    replies: int = 0  # copies created as an answer to the copy of the post the original answered
    reply_lost: int = 0  # answers that could only be copied without the reply
    reply_dropped: int = 0  # answers to a post that no longer exists (nothing to answer)
    from_saved: int = 0  # restricted posts that were copied from their saved copy in My posts
    media_lost: list = field(default_factory=list)  # ... of which these (old ids) went out without their media
    held_back: list = field(default_factory=list)  # posts that were not copied: only Telegram's notice was left (old id)
    held_why: Optional[str] = None  # what Telegram says about the restricted posts
    live_restricted: int = 0  # restricted posts without a saved copy, copied as Telegram hands them out
    no_media: bool = False  # /shift --no-media: no file was copied
    files_left_out: int = 0  # ... posts that were copied as text, without their file
    file_only: list = field(default_factory=list)  # ... posts that are only a file: nothing to copy (old id)
    used_saved: bool = False  # the run could take restricted posts from My posts (/shift)
    from_db: bool = False  # the channel can't be read: every post was copied from My posts alone
    checked: int = 0  # new posts that were read back to see whether Telegram itself hides them
    hidden: list = field(default_factory=list)  # [(old id, had media)] new posts Telegram itself hides (restriction mark)
    hidden_why: Optional[str] = None  # what Telegram says about them
    dst_restricted: Optional[str] = None  # what Telegram says about the destination channel (when copies were hidden)
    hidden_stop: bool = False  # the run stopped by itself: Telegram hides every copy it read back
    aborted: Optional[str] = None
    stopped: bool = False
    first_new: Optional[int] = None
    last_new: Optional[int] = None
    phase: str = "copy"  # copy | finish | profile | links
    final: Optional[FinalizeResult] = None
    final_error: Optional[str] = None
    cross: Optional[CrossResult] = None  # links to the old posts in ALL connected channels, pointed at the copies
    cross_error: Optional[str] = None
    profile: Any = None  # /shift -all: what happened to the name, description and photo (profile_copy.ProfileResult)
    profile_error: Optional[str] = None


def _rows(unit: list, finals: list, news: list) -> list:
    rows = []
    for src, f, new in zip(unit, finals, news):
        kind = classify_media(src)
        ref = (media_ref(new) or media_ref(src)) if kind else None
        rows.append(
            {
                "old_id": src.id,
                "new_id": new.id,
                "post": {
                    "text": plain_of(f.rich) if f.rich is not None else f.text,
                    "entities": [rich_entry(f.rich)] if f.rich is not None else ser_entities(f.entities),
                    "buttons": ser_markup(f.markup),
                    "media_kind": kind,
                    "media_file_id": ref,
                    "link_preview": bool(f.preview),
                },
            }
        )
    return rows


CHECK_FIRST = 3  # the first copies of any run are read back ...
CHECK_EVERY = 10  # ... then every tenth (all of them when Telegram restricts the source)
STOP_IF_HIDDEN = 3  # /shift stops when Telegram hides this many new copies and not one of those read back is shown


async def check_copies(client, dst_peer, unit: list, news: list, res: RepostResult) -> None:
    """Read the new posts back: does TELEGRAM itself hide them (a restriction mark, or its notice as the text)? That
    is not the bot copying a notice - Telegram restricts the new post as well, usually because of a file or a link in
    it. What is found goes into `res` (checked, hidden, hidden_why). Never fails the copy: it is only a look."""
    try:
        got = await fetch_messages(client, dst_peer, [n.id for n in news])
    except Exception as e:  # a look that fails is not a failed copy (the post exists and is registered already)
        log.info("could not read the new posts back: %s", type(e).__name__)
        return
    for old, new in zip(unit, got):
        if is_skippable(new):
            continue
        res.checked += 1
        why = message_restriction(new)
        if why:
            res.hidden.append((old.id, _has_media([old])))
            res.hidden_why = res.hidden_why or why


@dataclass
class CopyJob:
    """Everything the copy loop needs to know; /repost and /shift each fill it in their own way."""

    src_peer: Any
    dst_peer: Any  # the same as src_peer for /repost
    opts: RepostOptions
    links: PostLinks  # links.ids is `idmap`: old message id -> id of its copy, grows as we go
    first_id: int
    last_id: int
    done: set  # old ids that already have a copy
    idmap: dict
    record: Callable[[list], Awaitable[None]]  # saves [{old_id, new_id, post}] (and registers the posts)
    saved: Optional[SavedLookup] = None  # what My posts saved of the source's posts
    use_saved: bool = False  # a post Telegram holds back is rebuilt from its saved copy (otherwise it is left out)
    units: Optional[Callable[[], AsyncIterator[list]]] = None  # where the posts come from (default: read the channel)
    from_db: bool = False  # `units` are saved posts: they are all rebuilt, none is read from Telegram
    restricted: Optional[str] = None  # what Telegram says about the source channel itself (a copyright strike), if anything
    no_media: bool = False  # the copies get no files: text, formatting and buttons only (/shift --no-media)
    stop_if_hidden: int = 0  # stop when this many copies were read back and Telegram hides every one (0: never)


async def copy_loop(
    client,
    job: CopyJob,
    *,
    delay: float = 1.2,
    progress: Optional[Callable[[RepostResult], Awaitable[None]]] = None,
    should_stop: Optional[Callable[[], bool]] = None,
    settle_pause: float = 0.7,
) -> RepostResult:
    """Copy the posts first_id..last_id of `job.src_peer` to the end of `job.dst_peer`, in order.
    Safe to run again to continue: posts in `job.done` are skipped."""
    src, dst = job.src_peer, job.dst_peer
    seen: set = set()  # old ids read in this run (a reply to one of them needs its copy)
    res, st, streak = RepostResult(used_saved=job.use_saved or job.from_db, from_db=job.from_db), _State(), 0
    res.no_media = job.no_media
    units = job.units() if job.units is not None else iter_units(client, src, job.first_id, job.last_id)
    if job.no_media:
        units = single_posts(units)
    async for unit in units:
        if should_stop and should_stop():
            res.stopped = True
            break
        ids = [m.id for m in unit]
        seen.update(ids)
        if all(i in job.done for i in ids):
            res.skipped_done += 1
            continue
        rows = await job.saved(ids) if (job.saved and not job.from_db) else {}
        if job.from_db:
            source, why = "saved", None
        else:
            source, why = copy_source(unit, rows, restricted=job.restricted, use_saved=job.use_saved)
        if why:  # Telegram restricts this post (or the whole channel): what it hands out now is not trusted
            res.held_why = res.held_why or why
        from_saved = job.from_db  # saved posts already
        standing = None
        if source == "saved" and not job.from_db:
            standing = stand_ins(unit, rows)
            if standing is None:
                source = "skip"
        if source == "skip":
            log.info("post %s: Telegram shows only its notice (%s) and My posts has no saved copy to use", ids[0], why)
            res.held_back.append(ids[0])
            continue
        if standing is not None:
            unit, from_saved = standing, True
        no_forward, left_out = bool(why), 0
        if job.no_media:  # the copy is the text with its formatting and buttons; its file stays where it is
            unit, left_out = without_files(unit)
            if not unit:  # nothing but a file: there is nothing to post without it, and nothing waits for it
                res.file_only.append(ids[0])
                seen.difference_update(ids)
                continue
            no_forward = no_forward or bool(left_out)  # a copy by Telegram would carry the file
        finals = [final_state(m, job.opts, job.links) for m in unit]
        if any(is_placeholder_text(f.text) for f in finals):  # Telegram's notice is never posted, whatever it came from
            log.info("post %s: its text is Telegram's notice - not copied", ids[0])
            res.held_back.append(ids[0])
            res.held_why = res.held_why or PLACEHOLDER_NOTE
            continue
        try:
            reply_to = new_parent(unit[0], job.idmap, seen)
            news = await copy_unit(
                client, src, unit, finals, st, reply_to, dst, from_saved=from_saved, no_forward=no_forward
            )
            await settle_buttons(client, dst, finals, news, st, pause=settle_pause)
        except Exception as e:  # one bad post must not stop the others
            name = type(e).__name__
            log.warning("copying %s failed: %s %s", ids, name, e)
            res.failed.append((ids[0], name))
            streak += 1
            if name in FATAL or streak >= ABORT_AFTER:
                res.aborted = name
                break
            await asyncio.sleep(delay)
            continue
        try:
            await job.record(_rows(unit, finals, news))
        except Exception as e:  # keep the channel and the database consistent
            log.exception("could not save the copy of %s", ids)
            await delete_ids(client, dst, [n.id for n in news])
            res.failed.append((ids[0], type(e).__name__))
            res.aborted = type(e).__name__
            break
        streak = 0
        job.done.update(ids)
        for old, new in zip(unit, news):
            job.idmap[old.id] = new.id
        if reply_parent(unit[0]) is not None and reply_to is None:
            res.reply_dropped += 1
        if why and not from_saved:
            res.live_restricted += 1
        res.files_left_out += left_out
        if why or res.copied_units < CHECK_FIRST or res.copied_units % CHECK_EVERY == 0:
            await check_copies(client, dst, unit, news, res)
        res.copied_units += 1
        res.copied_messages += len(unit)
        res.first_new = res.first_new or news[0].id
        res.last_new = news[-1].id
        if progress:
            await progress(res)
        if job.stop_if_hidden and res.checked >= job.stop_if_hidden and len(res.hidden) == res.checked:
            log.info("Telegram hides every copy read back (%s): stopping", res.checked)
            res.stopped = res.hidden_stop = True  # more copies would only be more posts that nobody can see
            break
        await asyncio.sleep(delay)
    if res.hidden:  # Telegram hides copies: is the destination itself restricted too?
        try:
            _top, res.dst_restricted, _err = await probe_channel(client, dst)
        except Exception:  # only an extra for the report
            log.debug("could not ask Telegram about the destination channel", exc_info=True)
    res.forwarded, res.rebuilt, res.repaired = st.forwarded, st.rebuilt, st.repaired
    res.replies, res.reply_lost = st.replies, st.reply_lost
    res.from_saved, res.media_lost = st.from_saved, st.media_lost
    return res


async def run_repost(
    client,
    db,
    ch,
    mig,
    user_id: int,
    *,
    delay: float = 1.2,
    progress: Optional[Callable[[RepostResult], Awaitable[None]]] = None,
    should_stop: Optional[Callable[[], bool]] = None,
    settle_pause: float = 0.7,
) -> RepostResult:
    """Copy the posts of migration `mig` (ids first_id..last_id) in order; safe to run again to continue.

    When every post is copied (not stopped, no fatal error) the finishing touches follow: links to posts that were
    copied later are pointed at their copies, the copies of pinned posts are pinned, and then every connected channel
    is searched for links to the old posts (crosslinks.relink_after_repost)."""
    opts = RepostOptions(
        old=mig.old_username, new=mig.new_username, include_typed=mig.include_typed, include_posts=mig.include_posts
    )
    peer = peer_of(ch)
    idmap = dict(await db.migration_pairs(mid=mig.id, alive_only=False))
    job = CopyJob(
        src_peer=peer,
        dst_peer=peer,
        opts=opts,
        links=PostLinks.for_channel(ch, idmap),
        first_id=mig.first_id,
        last_id=mig.last_id,
        done=await db.migration_done_old_ids(mig.id),
        idmap=idmap,
        record=lambda rows: db.record_repost(mig.id, ch.id, rows, user_id),
        saved=saved_lookup(db, ch.id),  # a post Telegram holds back is left alone (and so is not deleted)
    )
    res = await copy_loop(client, job, delay=delay, progress=progress, should_stop=should_stop, settle_pause=settle_pause)
    if not res.stopped and not res.aborted:
        res.phase = "finish"
        if progress:
            await progress(res)
        try:
            res.final = await finalize_repost(
                client, db, ch, mig, delay=delay, settle_pause=settle_pause, should_stop=should_stop
            )
        except Exception as e:  # the copies are fine; only the finishing touches failed
            log.exception("finishing the repost (links, pins) failed")
            res.final_error = type(e).__name__
        if not (res.final is not None and res.final.stopped):
            res.phase = "links"
            res.cross = CrossResult()

            async def watch(_cross) -> None:
                if progress:
                    await progress(res)

            if progress:
                await progress(res)
            try:
                await relink_after_repost(
                    client, db, ch, mig, delay=delay, settle_pause=settle_pause, progress=watch,
                    should_stop=should_stop, result=res.cross,
                )
            except Exception as e:  # the copies are fine; only the links in the other channels were not updated
                log.exception("pointing the links of the other channels at the copies failed")
                res.cross_error = type(e).__name__
    return res


# --------------------------------------------------------------------- links to copies, pins
async def relink_copies(
    client,
    db,
    ch,
    ids: dict,
    targets: list,
    *,
    delay: float = 1.2,
    settle_pause: float = 0.7,
    dry_run: bool = False,
    progress: Optional[Callable[[FinalizeResult], Awaitable[None]]] = None,
    should_stop: Optional[Callable[[], bool]] = None,
    links: Optional[PostLinks] = None,
) -> FinalizeResult:
    """Point every link to a post of this channel at the copy of that post (`ids`: old message id -> new one).

    `targets` are the message ids of `ch` to look at (the copies). Idempotent: a link that already points at a copy
    is left alone, because only OLD ids are replaced. With dry_run nothing is edited, the result only counts.
    `links` (optional) says which links to follow; the default is "links to posts of `ch` itself" (/repost)."""
    res = FinalizeResult()
    peer = peer_of(ch)
    links = links if links is not None else PostLinks.for_channel(ch, ids)
    targets = sorted(set(targets))
    streak, last_name = 0, None
    for i in range(0, len(targets), CHUNK):
        chunk = targets[i : i + CHUNK]
        for mid, m in zip(chunk, await fetch_messages(client, peer, chunk)):
            if is_skippable(m):
                res.missing += 1
                continue
            res.checked += 1
            rl = relink_message(m, links)
            if rl is None:
                continue
            if dry_run:
                res.relinked += 1
                res.links += rl.links
                continue
            if should_stop and should_stop():
                res.stopped = True
                return res
            rich = is_rich_message(m)  # a rich message's text is blocks: only its buttons change, the text is not sent again
            try:
                if rich:
                    await flood_retry(lambda: edit_raw(client, peer, m.id, markup=rl.markup))
                else:
                    await flood_retry(
                        lambda: edit_raw(
                            client, peer, m.id, text=rl.text, entities=rl.entities, markup=rl.markup, preview=preview_flag(m)
                        )
                    )
                if link_urls(rl.markup):
                    await ensure_markup(client, peer, m.id, rl.markup, pause=settle_pause)
            except errors.MessageNotModifiedError:
                continue
            except Exception as e:  # one bad post must not stop the others
                name = type(e).__name__
                log.warning("could not update the links of %s: %s %s", m.id, name, e)
                res.failed.append((m.id, name))
                streak = streak + 1 if name == last_name else 1
                last_name = name
                if name in FATAL or streak >= ABORT_AFTER:
                    res.aborted = name
                    return res
                continue
            streak, last_name = 0, None
            res.relinked += 1
            res.links += rl.links
            try:
                saved = dict(buttons=ser_markup(rl.markup))
                if not rich:
                    saved.update(text=rl.text, entities=ser_entities(rl.entities))
                await db.update_post_by_message(ch.id, m.id, **saved)
            except Exception:  # the channel is right; the saved copy will be corrected by the next sync
                log.warning("could not save the new links of %s", m.id, exc_info=True)
            if progress:
                await progress(res)
            await asyncio.sleep(delay)
    return res


async def pin_copies(client, ch, pairs: list, res: FinalizeResult, *, delay: float = 1.2) -> None:
    """Pin the copy of every post that is pinned among the originals (oldest first, so the newest pin is on top).

    The "pinned a message" notices Telegram adds to the channel are deleted again right away."""
    peer = peer_of(ch)
    pairs = sorted(pairs)
    todo: list = []
    for i in range(0, len(pairs), CHUNK):
        chunk = pairs[i : i + CHUNK]
        olds = await fetch_messages(client, peer, [o for o, _ in chunk])
        news = await fetch_messages(client, peer, [n for _, n in chunk])
        for (_, new_id), src, cur in zip(chunk, olds, news):
            if is_skippable(src) or is_skippable(cur):
                continue
            if getattr(src, "pinned", False) and not getattr(cur, "pinned", False):
                todo.append(new_id)
    res.pin_wanted = len(todo)
    for new_id in todo:
        try:
            out = await flood_retry(
                lambda: client(functions.messages.UpdatePinnedMessageRequest(peer=peer, id=new_id, silent=True))
            )
        except errors.MessageNotModifiedError:
            continue  # already pinned
        except errors.RPCError as e:
            res.pin_error = type(e).__name__
            return
        res.pinned += 1
        await delete_ids(client, peer, [m.id for m in new_messages_from(out) if isinstance(m, types.MessageService)])
        await asyncio.sleep(delay)


async def finalize_repost(
    client,
    db,
    ch,
    mig,
    *,
    delay: float = 1.2,
    settle_pause: float = 0.7,
    progress: Optional[Callable[[FinalizeResult], Awaitable[None]]] = None,
    should_stop: Optional[Callable[[], bool]] = None,
) -> FinalizeResult:
    """After the copies exist: links that point at old posts get the copies' ids, copies of pinned posts get pinned."""
    pairs = sorted(await db.migration_pairs(mid=mig.id))
    res = await relink_copies(
        client, db, ch, dict(pairs), [n for _, n in pairs], delay=delay, settle_pause=settle_pause,
        progress=progress, should_stop=should_stop,
    )
    if not res.stopped and not res.aborted:
        await pin_copies(client, ch, pairs, res, delay=delay)
    return res


# ----------------------------------------------------------------------------- deleting
@dataclass
class DeleteResult:
    deleted: int = 0
    by_userbot: int = 0  # of those, how many only the userbot could delete
    remaining: int = 0
    blocked: bool = False  # some messages are still there although Telegram did not report an error
    error: Optional[str] = None  # the bot's Telegram error that stopped the run
    fallback_error: Optional[str] = None  # why the userbot could not help
    tried_userbot: bool = False


async def gone_ids(client, peer, ids: list) -> set:
    """Which of `ids` no longer exist in the channel."""
    msgs = await fetch_messages(client, peer, ids)
    return {i for i, m in zip(ids, msgs) if m is None or isinstance(m, types.MessageEmpty)}


@dataclass
class BatchOutcome:
    gone: set = field(default_factory=set)  # message ids that no longer exist afterwards
    bot_error: Optional[str] = None
    by_userbot: int = 0
    fallback_error: Optional[str] = None
    tried_userbot: bool = False


async def delete_batch(client, peer, ids: list, fallback: Optional[Callable[[list], Awaitable[Any]]] = None) -> BatchOutcome:
    """Delete up to 100 messages: the bot tries first, the result is checked by reading the messages back, and
    whatever is still there goes to `fallback(ids)` (the userbot), followed by a second check."""
    out = BatchOutcome()
    try:
        await flood_retry(lambda: client(functions.channels.DeleteMessagesRequest(channel=peer, id=list(ids))))
    except errors.RPCError as e:
        out.bot_error = type(e).__name__
    out.gone = await gone_ids(client, peer, ids)
    left = [x for x in ids if x not in out.gone]
    if left and fallback is not None:
        out.tried_userbot = True
        try:
            await fallback(left)
        except Exception as e:  # a userbot problem must not hide what the bot already did
            log.warning("the userbot could not delete %s message(s): %s", len(left), e)
            out.fallback_error = str(e) or type(e).__name__
        again = await gone_ids(client, peer, left)
        out.by_userbot = len(again)
        out.gone |= again
    return out


async def delete_pending(
    client,
    peer,
    todo: list,
    mark: Callable[[list], Awaitable[None]],
    *,
    delay: float = 0.5,
    progress: Optional[Callable[[DeleteResult], Awaitable[None]]] = None,
    fallback: Optional[Callable[[list], Awaitable[Any]]] = None,
) -> DeleteResult:
    """Delete the messages `todo` (oldest first, 100 per request). `mark(ids)` records the ones that are gone.

    The bot tries first and the result is always checked by reading the messages back. Whatever the bot could not
    delete (Telegram may refuse old posts) goes to `fallback(ids)` - the userbot - if there is one, and is checked
    again. A batch the bot cannot delete does not stop the run: newer posts may still be deletable."""
    res = DeleteResult()
    use_fallback = fallback
    for i in range(0, len(todo), 100):
        ids = todo[i : i + 100]
        out = await delete_batch(client, peer, ids, use_fallback)
        res.tried_userbot = res.tried_userbot or out.tried_userbot
        res.by_userbot += out.by_userbot
        if out.fallback_error:
            res.fallback_error = out.fallback_error
            use_fallback = None  # it will not work for the next batches either
        left = [x for x in ids if x not in out.gone]
        if out.gone:
            await mark(sorted(out.gone))
            res.deleted += len(out.gone)
        if left and out.bot_error:
            res.error = res.error or out.bot_error  # remembered for the report
            if out.bot_error in FATAL and use_fallback is None:
                break  # e.g. no "Delete messages" right: the next batch would fail the same way
        if progress:
            await progress(res)
        await asyncio.sleep(delay)
    return res


async def delete_copies(
    client,
    db,
    ch,
    mig,
    which: str,
    *,
    delay: float = 0.5,
    progress: Optional[Callable[[DeleteResult], Awaitable[None]]] = None,
    fallback: Optional[Callable[[list], Awaitable[Any]]] = None,
) -> DeleteResult:
    """which='old': remove the original posts; which='new': remove the copies (undo). Oldest first, 100 per request."""
    todo = await db.migration_pending_ids(mig.id, which, 1_000_000)

    async def mark(ids: list) -> None:
        await db.mark_migration_deleted(mig.id, ch.id, which, ids)

    res = await delete_pending(client, peer_of(ch), todo, mark, delay=delay, progress=progress, fallback=fallback)
    counts = await db.migration_counts(mig.id)
    res.remaining = counts["old_left"] if which == "old" else counts["new_left"]
    res.blocked = res.remaining > 0 and not res.error
    return res