Blame for app/sync_engine.py

"""Keeps My posts in step with the channels: what admins do outside the bot is noticed and recorded.

What a look at a channel does
  * every post the bot has saved is read back from Telegram: a post that is gone is removed here, a post that was edited
    gets its saved copy updated (text, formatting, buttons, media);
  * messages that appeared since the last look are saved as new posts (source "adopted");
  * the posts that were in the channel BEFORE the bot was added are read into My posts too, oldest first, up to the
    newest message that existed at the first look. Where the reading stopped is saved, so a long channel takes a few
    looks. Posts the owner told the bot to forget are skipped for good;
  * service messages ("channel name changed", "pinned a message", video chat ...) found on the way are deleted.

Safety rules
  * posts the bot touched a moment ago, and messages that appeared a moment ago, are left for the next look (the bot may
    still be saving them);
  * a saved copy is only changed if nobody changed it since it was read (optimistic check in the database);
  * buttons that are missing in the channel are never removed from the saved copy (Telegram sometimes hides a keyboard
    for a while - the saved buttons are what "Check buttons" restores from);
  * if EVERY saved post of a channel looks deleted, the bot first makes sure it can read the channel at all: when it
    still sees other messages there (or the channel's event counter moved by at least that many events, which is what
    really deleting them looks like) the posts are really gone and leave My posts; only when it sees nothing at all
    - more likely the bot lost access - is nothing removed;
  * what TELEGRAM does to a channel is not an admin's edit: a channel it restricts (a copyright strike turns every post
    into "This message couldn't be displayed on your device due to copyright infringement"), a single post it holds
    back, a notice shown instead of a post, or many posts that suddenly all show one and the same text are left exactly
    as saved - nothing is edited, removed or added for them.
"""
from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Optional

from telethon import errors, types

from .db import as_utc, utcnow
from .replace_engine import fetch_messages
from .repost_engine import delete_batch
from .restrictions import message_restriction, norm_text, probe_channel
from .richmsg import Rich, entry as rich_entry, is_rich_message, plain_of_message, rich_from_received, rich_of, with_fingerprint
from .tgutil import classify_media, explain_rpc, media_from_ref, media_info, peer_of, ser_entities, ser_markup

log = logging.getLogger(__name__)

CHUNK = 100  # Telegram returns at most 100 messages per request
GRACE = 120.0  # seconds a post or message must be old before it is judged
PAUSE = 0.25  # between two requests of a scan
SLACK_CAP = 300  # a quick look reads this many ids beyond the most likely end of the new messages
DEEP_EVERY = 24 * 3600  # the automatic looks read every possible id at least this often
MIN_FOR_ANOMALY = 5
MASS_SAME = 5  # this many posts that suddenly show the very same new text are not edits by a person
MAX_PROBE = 300  # requests spent finding the newest message the first time (30 000 ids)


# ----------------------------------------------------------------------------------------------- options
@dataclass
class SyncOptions:
    deep: bool = False  # read every possible id for new messages, not only the likely ones
    delete_services: bool = True
    adopt: bool = True  # save new posts of others
    history: bool = True  # also read the posts from before the bot was added into My posts
    history_seconds: Optional[float] = None  # ... for at most this long per look (None: as long as it takes)
    grace: float = GRACE
    pause: float = PAUSE
    ignore: Any = field(default_factory=set)  # (channel id, message id) of posts the owner told the bot to forget


@dataclass
class SyncReport:
    channel: Any
    phase: str = ""
    checked: int = 0  # saved posts compared with the channel
    deleted: list = field(default_factory=list)  # message ids that were gone: removed from My posts
    edited: list = field(default_factory=list)  # message ids whose saved copy was brought up to date
    refreshed: int = 0  # saved copies corrected for trifles (a trailing space ...)
    adopted: list = field(default_factory=list)  # message ids of new posts that were saved
    imported: int = 0  # older posts (from before the bot was added) that were added to My posts
    history_unsupported: int = 0  # older messages My posts can't hold (polls, stickers ...)
    history_top: int = 0  # the newest message that counts as older (0: the older posts were not looked at)
    history_to: int = 0  # the older posts have been read up to this message id
    history_done: bool = False  # ... and that was all of them
    unsupported: int = 0  # new messages of a kind My posts can't show (polls, stickers ...)
    buttons_missing: list = field(default_factory=list)  # saved buttons that the channel shows no keyboard for
    services_deleted: int = 0
    services_failed: int = 0
    services_error: Optional[str] = None  # name of the refusal
    restricted: list = field(default_factory=list)  # message ids Telegram holds back (copyright ...): left as saved
    restricted_why: Optional[str] = None  # what Telegram says about them
    restricted_channel: Optional[str] = None  # the whole channel is restricted: nothing was looked at
    young: list = field(default_factory=list)  # saved posts the bot touched a moment ago: judge them later
    young_new: list = field(default_factory=list)  # new messages that appeared a moment ago: save them later
    changed_meanwhile: int = 0  # saved posts that were changed through the bot while we looked (left alone)
    first_look: bool = False
    busy: bool = False  # a long job started: the look was cut short
    anomaly: Optional[str] = None
    error: Optional[str] = None
    scanned_to: int = 0

    @property
    def changed(self) -> bool:
        return bool(self.deleted or self.edited or self.adopted or self.imported or self.services_deleted)


# ------------------------------------------------------------------------------------- comparing a post
def age(dt, now) -> float:
    """Seconds since `dt` (never negative); unknown = very old."""
    if dt is None:
        return float("inf")
    return max(0.0, (now - as_utc(dt)).total_seconds())


def media_key(media) -> Optional[tuple]:
    """Which photo / file a message carries (its id - the file reference changes from fetch to fetch)."""
    if isinstance(media, types.MessageMediaPhoto) and getattr(media, "photo", None):
        return ("photo", media.photo.id)
    if isinstance(media, types.MessageMediaDocument) and getattr(media, "document", None):
        return ("doc", media.document.id)
    return None


def stored_media_key(ref: Optional[str]) -> Optional[tuple]:
    if not ref:
        return None
    try:
        return media_key(media_from_ref(ref))
    except Exception:
        return ("unreadable", ref[:40])  # differs from anything real: the saved reference gets replaced


def live_rich(m) -> Optional[Rich]:
    """The rich message the live message is (as blocks), or None for an ordinary post."""
    rm = getattr(m, "rich_message", None)
    return rich_from_received(rm) if rm is not None else None


def live_fields(m) -> dict:
    """The saved-post fields of a message as it is in the channel now."""
    kind, ref = media_info(m)
    rich = live_rich(m)
    return {
        "text": plain_of_message(m) if rich is not None else (m.message or ""),
        "entities": [rich_entry(rich)] if rich is not None else ser_entities(m.entities),
        "buttons": ser_markup(m.reply_markup),
        "media_kind": kind,
        "media_file_id": ref,
    }


def message_supported(m) -> bool:
    """Something My posts can show and edit: text, a rich message, or a photo / video / GIF / audio / file."""
    return bool((m.message or "").strip()) or classify_media(m) is not None or is_rich_message(m)


def adopt_fields(m) -> dict:
    f = live_fields(m)
    f["link_preview"] = isinstance(getattr(m, "media", None), types.MessageMediaWebPage)
    return f


# ------------------------------------------------------------------- what Telegram itself does to a channel
MASS_NOTE = "many posts suddenly show one and the same text - that is Telegram's notice, not an edit"


def split_mass_text(edits: list) -> tuple:
    """(edits to save, edits to hold back). Five or more saved posts that had five different texts and now all show the
    very same text at once: nobody edits like that - Telegram does it to a channel it restricts."""
    groups: dict = {}
    for item in edits:
        new = norm_text(item[1].fields.get("text"))
        if new:
            groups.setdefault(new, []).append(item)
    bad: set = set()
    for items in groups.values():
        if len(items) >= MASS_SAME and len({norm_text(p.text) for p, _ in items}) >= MASS_SAME:
            bad.update(id(i) for i in items)
    return [i for i in edits if id(i) not in bad], [i for i in edits if id(i) in bad]


def hold_back(rep: "SyncReport", ids, why: Optional[str]) -> None:
    rep.restricted.extend(ids)
    rep.restricted_why = rep.restricted_why or why


@dataclass
class Diff:
    fields: dict = field(default_factory=dict)  # what to save
    buttons_missing: bool = False
    quiet: bool = False  # only a trifle differs (surrounding spaces): fix it without counting it as an edit


def diff_post(p, m) -> Diff:
    """What differs between the saved post `p` and the live message `m`."""
    live = live_fields(m)
    fields: dict = {}
    saved_rich, now_rich = rich_of(p), live_rich(m) if is_rich_message(m) else None
    first_look = False
    if saved_rich is None and now_rich is None:
        if (p.text or "") != live["text"]:
            fields["text"] = live["text"]
        if list(p.entities or []) != live["entities"]:
            fields["entities"] = live["entities"]
    elif now_rich is not None:
        # A rich message is compared by what Telegram shows of it (a fingerprint of its blocks), never by its text: the
        # saved copy keeps the source the bot sent. A post that became rich in the channel is taken over as its blocks.
        if saved_rich is None or (saved_rich.fp is not None and saved_rich.fp != now_rich.fp):
            fields["text"], fields["entities"] = live["text"], live["entities"]
        elif saved_rich.fp is None:  # the bot's own rich post, seen for the first time: remember what Telegram shows
            fields["entities"] = [rich_entry(with_fingerprint(saved_rich, now_rich.fp))]
            first_look = True
    elif live["text"].strip() or live["media_kind"] is not None:
        # the channel shows an ordinary post where a rich message was saved (an empty answer is not trusted)
        fields["text"], fields["entities"] = live["text"], live["entities"]
    # media: replaced media is taken over; media that seems to have vanished is not (a message can't lose its media)
    if live["media_kind"] is not None:
        if p.media_kind != live["media_kind"] or stored_media_key(p.media_file_id) != media_key(m.media):
            fields["media_kind"], fields["media_file_id"] = live["media_kind"], live["media_file_id"]
    missing = False
    if live["buttons"]:
        if list(p.buttons or []) != live["buttons"]:
            fields["buttons"] = live["buttons"]
    elif p.buttons:
        missing = True
    quiet = (set(fields) == {"text"} and (p.text or "").strip() == live["text"].strip()) or (
        first_look and set(fields) == {"entities"}
    )
    return Diff(fields, missing, quiet)


# --------------------------------------------------------------------------------------- talking to Telegram
def is_gone(m) -> bool:
    return m is None or isinstance(m, types.MessageEmpty)


async def find_top(client, peer, pts: int, floor: int, *, pause: float = PAUSE) -> int:
    """The newest message id that exists. Only the event counter is known (never lower), so look downwards from it;
    `floor` (the newest saved post) is where the looking can stop."""
    hi = pts
    for _ in range(MAX_PROBE):
        if hi <= floor:
            break
        lo = max(floor + 1, hi - CHUNK + 1)
        msgs = await fetch_messages(client, peer, list(range(lo, hi + 1)))
        alive = [m.id for m in msgs if not is_gone(m)]
        if alive:
            return max(alive)
        hi = lo - 1
        await asyncio.sleep(pause)
    return max(floor, hi)


async def sees_live_messages(client, peer, pts: Optional[int], *, pause: float = PAUSE, tries: int = MAX_PROBE) -> bool:
    """True when the bot can read at least one message of the channel (a post or a service message). Then a saved post
    that is missing from the channel really is gone - the bot is not just blind. Looks downwards from the event counter."""
    if not pts or pts < 1:
        return False
    hi = pts
    for _ in range(tries):
        if hi < 1:
            return False
        lo = max(1, hi - CHUNK + 1)
        msgs = await fetch_messages(client, peer, list(range(lo, hi + 1)))
        if any(not is_gone(m) for m in msgs):
            return True
        hi = lo - 1
        await asyncio.sleep(pause)
    return False


async def delete_services(client, peer, ids: list, rep: SyncReport) -> None:
    """Delete service messages and count what happened (the bot only: Telegram may refuse the old ones)."""
    ids = sorted(set(ids))
    for i in range(0, len(ids), CHUNK):
        part = ids[i : i + CHUNK]
        out = await delete_batch(client, peer, part)
        rep.services_deleted += len(out.gone)
        left = len(part) - len(out.gone)
        if left:
            rep.services_failed += left
            rep.services_error = rep.services_error or out.bot_error or "NotDeleted"


# ------------------------------------------------------------------------------------- the three steps
async def _save_edits(db, rep: SyncReport, edits: list, busy) -> None:
    for p, d in edits:
        if busy():
            rep.busy = True
            return
        if await db.sync_update_post(p.id, p.updated_at, d.fields):
            if d.quiet:
                rep.refreshed += 1
            else:
                rep.edited.append(p.message_id)
        else:
            rep.changed_meanwhile += 1


async def _adopt(db, ch, rep: SyncReport, fresh: list, opts: SyncOptions, busy, now) -> None:
    """Save messages the bot does not know (ascending ids); too fresh ones are left for the next look."""
    known: set = set()
    ids = [m.id for m in fresh]
    for i in range(0, len(ids), 500):
        known |= {p.message_id for p in await db.posts_at(ch.id, ids[i : i + 500])}
    forgotten = await db.ignored_ids(ch.id) if fresh else set()
    todo = []
    for m in fresh:
        if m.id in known or m.id in forgotten or (ch.id, m.id) in opts.ignore:
            continue
        why = message_restriction(m)
        if why:  # Telegram holds it back: what it shows now is not what the post says
            hold_back(rep, [m.id], why)
        elif age(m.date, now) < opts.grace:
            rep.young_new.append(m.id)
        elif not message_supported(m):
            rep.unsupported += 1
        else:
            todo.append(m)
    for m in todo:
        if busy():
            rep.busy = True
            return
        _, created = await db.adopt_message(ch.id, m.id, adopt_fields(m), sent_at=m.date)
        if created:
            rep.adopted.append(m.id)


async def _compare_saved(
    client, db, ch, rep: SyncReport, opts: SyncOptions, busy, progress, moved=None, pts: Optional[int] = None
) -> None:
    """Every saved post against the channel. `moved`: how far the channel's event counter has come since the last look
    (None: unknown) - deleting posts is an event each, which tells "everything was deleted" from "I can't see anything".
    `pts`: the channel's event counter now - where the bot looks for other messages to prove that it can read the channel."""
    rep.phase = "comparing the saved posts"
    peer = peer_of(ch)
    posts = await db.sent_posts(ch.id)  # the saved copies are read BEFORE the channel is, never after
    now = utcnow()
    judge = []
    for p in posts:
        if min(age(p.updated_at, now), age(p.sent_at, now)) < opts.grace:
            rep.young.append(p.message_id)
        else:
            judge.append(p)
    rep.checked = len(judge)
    by_id = {p.message_id: p for p in judge}
    ids = sorted(by_id)
    gone, edits = [], []
    for i in range(0, len(ids), CHUNK):
        if busy():
            rep.busy = True
            return
        chunk = ids[i : i + CHUNK]
        msgs = await fetch_messages(client, peer, chunk)
        for mid, m in zip(chunk, msgs):
            p = by_id[mid]
            if is_gone(m):
                gone.append(p)
            elif isinstance(m, types.MessageService):
                continue  # not a post any more (cannot happen: ids are never reused)
            else:
                why = message_restriction(m, p.text)
                if why:  # Telegram's doing, not an admin's: the saved copy stays exactly as it is
                    hold_back(rep, [mid], why)
                    continue
                d = diff_post(p, m)
                if d.buttons_missing:
                    rep.buttons_missing.append(mid)
                if d.fields:
                    edits.append((p, d))
        rep.scanned_to = chunk[-1]
        if progress:
            await progress(rep)
        await asyncio.sleep(opts.pause)
    edits, held = split_mass_text(edits)
    if held:
        hold_back(rep, [p.message_id for p, _ in held], MASS_NOTE)
    if gone:
        really = moved is not None and moved >= len(gone)  # at least one event for each post that vanished
        everything = len(ids) >= MIN_FOR_ANOMALY and len(gone) == len(ids)
        if everything and not really:
            # No earlier look to compare with (or too few events): is the bot blind, or are the posts really gone? If it
            # can read other messages of the channel it is not blind.
            if busy():
                rep.busy = True
                return
            really = await sees_live_messages(client, peer, pts, pause=opts.pause)
        if everything and not really:
            rep.anomaly = (
                f"All {len(ids)} saved posts look deleted, and I can't read any other message of the channel either, "
                "so none was removed here - check that the bot is still an admin of the channel."
            )
        else:
            gone_ids = [p.message_id for p in gone]
            await db.forget_posts(ch.id, gone_ids)
            rep.deleted.extend(gone_ids)
    await _save_edits(db, rep, edits, busy)


async def _look_for_new(client, db, ch, rep: SyncReport, opts: SyncOptions, state, pts: int, busy, progress) -> None:
    """Messages that appeared since the last look: new posts are saved, service messages deleted."""
    rep.phase = "looking for new posts"
    peer = peer_of(ch)
    now = utcnow()
    if state is None:  # the first look: only note where the channel stands
        floor = await db.max_saved_id(ch.id)
        top = await find_top(client, peer, pts, floor, pause=opts.pause)
        rep.first_look, rep.scanned_to = True, top
        await db.save_sync(ch.id, last_top=top, last_pts=pts, deferred=0, deep=True)
        return
    deep = opts.deep or age(state.last_deep, now) > DEEP_EVERY
    if pts == state.last_pts and not state.deferred and not deep:
        return  # not a single event since the last look: a quick look would read exactly what it read then
    delta = max(0, pts - state.last_pts)
    slack = min(max(0, state.last_pts - state.last_top), SLACK_CAP)
    soft = pts if deep else min(pts, state.last_top + delta + slack)
    a, highest, had_live = state.last_top + 1, state.last_top, True
    fresh, services = [], []
    while a <= pts:
        if a > soft and not had_live:
            break  # past the likely end and the last stretch was empty
        if busy():
            rep.busy = True
            return
        hi = min(a + CHUNK - 1, pts)
        msgs = await fetch_messages(client, peer, list(range(a, hi + 1)))
        had_live = False
        for m in msgs:
            if is_gone(m):
                continue
            had_live = True
            highest = max(highest, m.id)
            if isinstance(m, types.MessageService):
                services.append(m)
            else:
                fresh.append(m)
        rep.scanned_to = hi
        if progress:
            await progress(rep)
        a = hi + 1
        await asyncio.sleep(opts.pause)
    services_ids = [m.id for m in services]
    if opts.delete_services and services_ids:
        await delete_services(client, peer, services_ids, rep)
    if opts.adopt and fresh:
        await _adopt(db, ch, rep, fresh, opts, busy, now)
        if rep.busy:
            return
    new_top = highest
    if rep.young_new:  # these have to be looked at again: don't move the mark past them
        new_top = min(new_top, min(rep.young_new) - 1)
    await db.save_sync(
        ch.id, last_top=max(state.last_top, new_top), last_pts=pts, deferred=len(rep.young_new), deep=deep
    )


async def _import_history(client, db, ch, rep: SyncReport, opts: SyncOptions, busy, progress, state=None) -> None:
    """The posts that were in the channel before the bot was added are read into My posts: from message 1 up to the newest
    message that existed at the first look (what came after is the business of _look_for_new). The place where the
    reading stopped is saved - a long channel takes a few looks, and a restart loses nothing. `state`: where the look
    before this one stopped (None at the very first look: the mark that look has just set is used)."""
    hist = await db.get_history(ch.id)
    if hist is None:
        base = state if state is not None else await db.get_sync(ch.id)
        if base is None:
            return
        hist = await db.start_history(ch.id, base.last_top)
    if hist.done:
        return
    rep.phase = "reading the older posts"
    peer = peer_of(ch)
    began = time.monotonic()
    forgotten = await db.ignored_ids(ch.id)
    rep.history_top = hist.top
    a = hist.next_id
    rep.history_to = a - 1
    while a <= hist.top:
        if busy():
            rep.busy = True
            return
        if opts.history_seconds is not None and time.monotonic() - began > opts.history_seconds:
            return  # the next look goes on from here
        hi = min(a + CHUNK - 1, hist.top)
        msgs = await fetch_messages(client, peer, list(range(a, hi + 1)))
        now = utcnow()
        stop = None
        todo = []
        # posts the bot has saved already are none of the reading's business: the comparison of the saved posts looks
        # after them (and counts the ones Telegram holds back there - not a second time here)
        have = {p.message_id for p in await db.posts_at(ch.id, [m.id for m in msgs if not is_gone(m)])}
        for m in msgs:
            if is_gone(m) or isinstance(m, types.MessageService):
                continue
            if age(m.date, now) < opts.grace:  # the bot may still be saving it: the next look reads it
                stop = m.id
                break
            if m.id in forgotten or m.id in have or (ch.id, m.id) in opts.ignore:
                continue
            why = message_restriction(m)
            if why:
                hold_back(rep, [m.id], why)
            elif not message_supported(m):
                rep.history_unsupported += 1
            else:
                todo.append(m)
        made = await db.adopt_many(ch.id, [(m.id, adopt_fields(m), m.date) for m in todo])  # (skips what is saved)
        rep.imported += len(made)
        a = hi + 1 if stop is None else stop
        rep.history_done = stop is None and a > hist.top
        await db.save_history(ch.id, next_id=a, imported=len(made), done=rep.history_done)
        rep.history_to = rep.scanned_to = a - 1
        if stop is not None:
            return
        if progress:
            await progress(rep)
        await asyncio.sleep(opts.pause)
    rep.history_done = True  # (also the case when the channel had nothing to read)


# ----------------------------------------------------------------------------------------- entry points
async def sync_channel(client, db, ch, opts: Optional[SyncOptions] = None, *, busy=None, progress=None) -> SyncReport:
    """One full look at a channel (see the module text). Never raises Telegram's refusals: they end up in `report.error`."""
    opts = opts or SyncOptions()
    busy = busy or (lambda: False)
    rep = SyncReport(channel=ch)
    pts, restricted, err = await probe_channel(client, peer_of(ch))
    if pts is None:
        rep.error = err
        return rep
    if restricted:  # Telegram restricted the channel (a copyright strike ...): My posts stay untouched
        rep.restricted_channel = restricted
        return rep
    try:
        state = await db.get_sync(ch.id)
        moved = None if state is None else max(0, pts - state.last_pts)
        await _compare_saved(client, db, ch, rep, opts, busy, progress, moved, pts)
        if not rep.busy:
            await _look_for_new(client, db, ch, rep, opts, state, pts, busy, progress)
        if not rep.busy and opts.history:
            await _import_history(client, db, ch, rep, opts, busy, progress, state)
    except errors.RPCError as e:
        rep.error = explain_rpc(e)
    return rep


async def sync_ids(client, db, ch, ids, opts: Optional[SyncOptions] = None, *, adopt_ids=(), busy=None) -> SyncReport:
    """A look at these messages only - the ones an update told about. `adopt_ids`: the ones that are new posts, so a
    message the bot does not know may be saved (an edited message the bot does not know is history: left alone)."""
    opts = opts or SyncOptions()
    busy = busy or (lambda: False)
    rep = SyncReport(channel=ch)
    peer = peer_of(ch)
    ids = sorted(set(ids))
    adopt_set = set(adopt_ids)
    try:
        rows = {p.message_id: p for p in await db.posts_at(ch.id, ids)}  # saved copies first, the channel second
        now = utcnow()
        gone, edits, fresh, services = [], [], [], []
        for i in range(0, len(ids), CHUNK):
            if busy():
                rep.busy = True
                return rep
            chunk = ids[i : i + CHUNK]
            msgs = await fetch_messages(client, peer, chunk)
            for mid, m in zip(chunk, msgs):
                row = rows.get(mid)
                recent = row is not None and min(age(row.updated_at, now), age(row.sent_at, now)) < opts.grace
                if is_gone(m):
                    if row is not None and not recent:
                        gone.append(row)
                elif isinstance(m, types.MessageService):
                    services.append(mid)
                elif row is not None:
                    why = message_restriction(m, row.text)
                    if why:  # Telegram's doing, not an admin's: the saved copy stays exactly as it is
                        hold_back(rep, [mid], why)
                        continue
                    if recent:
                        rep.young.append(mid)
                        continue
                    d = diff_post(row, m)
                    if d.buttons_missing:
                        rep.buttons_missing.append(mid)
                    if d.fields:
                        edits.append((row, d))
                elif mid in adopt_set:
                    fresh.append(m)
        if opts.delete_services and services:
            await delete_services(client, peer, services, rep)
        edits, held = split_mass_text(edits)
        if held:
            hold_back(rep, [p.message_id for p, _ in held], MASS_NOTE)
        if gone:
            gone_ids = [p.message_id for p in gone]
            await db.forget_posts(ch.id, gone_ids)
            rep.deleted.extend(gone_ids)
        await _save_edits(db, rep, edits, busy)
        if opts.adopt and fresh and not rep.busy:
            await _adopt(db, ch, rep, fresh, opts, busy, now)
    except errors.RPCError as e:
        rep.error = explain_rpc(e)
    return rep