Automated sync from GitHub

.gitattributes

@@ -1,8 +1,8 @@
-*.jpeg filter=lfs diff=lfs merge=lfs -text
-*.png filter=lfs diff=lfs merge=lfs -text
-*.ttf filter=lfs diff=lfs merge=lfs -text
 *.otf filter=lfs diff=lfs merge=lfs -text
 *.webp filter=lfs diff=lfs merge=lfs -text
 *.mp4 filter=lfs diff=lfs merge=lfs -text
 *.mp3 filter=lfs diff=lfs merge=lfs -text
 *.jpg filter=lfs diff=lfs merge=lfs -text
+*.jpeg filter=lfs diff=lfs merge=lfs -text
+*.png filter=lfs diff=lfs merge=lfs -text
+*.ttf filter=lfs diff=lfs merge=lfs -text

bot/app.py

@@ -24,6 +24,8 @@ from . import fasttransfer as transfer
 from . import errorlog, media, mediainfo, panels, richtext, speed, sysinfo
 from .config import Config
 from .db import MAX_FORMATS, Database
+from .dbflow import PREFIX as DBFLOW_PREFIX
+from .dbflow import DbFlow
 from .errors import FormatError
 from .formats import (
     FIELD_KINDS,
@@ -40,7 +42,7 @@ from .parser import compute_new_name, missing_variables, parse_caption, parse_in
 from .pipeline import PROBE_EXTS, Renamer, embedded_thumb, original_name_of, source_ext
 from .queue import Job, JobQueue
 from .sequence import SeqItem, SequenceState, analyse, format_report
-from .settings_ops import CLEAR_WORDS, META_COMMANDS, apply_setting, parse_channel
+from .settings_ops import CLEAR_WORDS, META_COMMANDS, apply_setting, parse_channel, parse_format_channel
 from .telegraph import Telegraph, TelegraphError
 from .utils import clip, esc, human_size, human_uptime
 
@@ -223,6 +225,10 @@ class App:
         self._tasks: Set["asyncio.Future[Any]"] = set()
         self.renamer = Renamer(client, cfg, db)
         self.queue = JobQueue(self.renamer.process, workers=cfg.max_jobs, on_cancelled=self._on_job_cancelled, settle=QUEUE_SETTLE)
+        # uploading a manually renamed episode to the Database Channel (and having it published) needs both chats
+        self.dbflow: Optional[DbFlow] = DbFlow(client, cfg) if cfg.upload_enabled else None
+        if self.dbflow is not None:
+            self.renamer.on_delivered = self.dbflow.offer
         self._warned: Set[int] = set()
         self._username: Optional[str] = None  # the bot's own @username, for the links on shared-format cards
         self._inline_on = False  # is inline mode on for the bot?  See _check_inline
@@ -247,11 +253,34 @@ class App:
                 await self.client.get_input_entity(self.cfg.log_channel)
             except Exception as exc:  # noqa: BLE001
                 log.warning("cannot access LOG_CHANNEL %s: %s (add the bot there as an admin)", self.cfg.log_channel, exc)
+        await self._check_upload_chats()
         self.queue.start()
 
+    async def _check_upload_chats(self) -> None:
+        """Say in the log what the upload to the Database Channel cannot reach (or that it is switched off)."""
+        cfg = self.cfg
+        if self.dbflow is None:
+            if cfg.db_channel or cfg.comm_chat:
+                log.warning(
+                    "uploading to the Database Channel is off: it needs both DB_CHANNEL (%s) and COMMUNICATION_CHAT (%s)",
+                    cfg.db_channel or "missing", cfg.comm_chat or "missing",
+                )
+            return
+        log.info(
+            "uploading to the Database Channel is on: Database Channel %s, communication chat %s, channel post bot %s",
+            cfg.db_channel, cfg.comm_chat, cfg.post_bot_id,
+        )
+        for name, ref in (("DB_CHANNEL", cfg.db_channel), ("COMMUNICATION_CHAT", cfg.comm_chat)):
+            try:
+                await self.client.get_input_entity(ref)
+            except Exception as exc:  # noqa: BLE001
+                log.warning("cannot access %s %s: %s (add the bot there as an admin)", name, ref, exc)
+
     async def shutdown(self) -> None:
         for task in list(self._tasks):
             task.cancel()
+        if self.dbflow is not None:
+            await self.dbflow.shutdown()
         await self.queue.stop()
 
     async def _set_commands(self) -> None:
@@ -342,6 +371,11 @@ class App:
         add(self._guard(self.on_text), events.NewMessage(incoming=True, func=self._wants_text))
         add(self._guard(self.on_callback), events.CallbackQuery())
         add(self._guard_inline(self.on_inline), events.InlineQuery())
+        if self.dbflow is not None:  # the channel post bot's answers in the communication chat (not a user of this bot: no guard)
+            add(
+                self.dbflow.on_chat_message,
+                events.NewMessage(incoming=True, chats=[self.cfg.comm_chat], from_users=[self.cfg.post_bot_id]),
+            )
         add(self._on_stranger, events.NewMessage(incoming=True, func=lambda e: e.is_private and not self.is_allowed(e.sender_id)))
 
     async def _on_stranger(self, event: Any) -> None:
@@ -1508,6 +1542,13 @@ class App:
             await self._cb_rename(event, uid, parts)
             return
 
+        if head == DBFLOW_PREFIX:
+            if self.dbflow is None:
+                await event.answer("Uploading to the Database Channel is switched off (DB_CHANNEL and COMMUNICATION_CHAT).", alert=True)
+            else:
+                await self.dbflow.on_callback(event, parts)
+            return
+
         if head == "sq":
             await self._cb_sequence(event, uid, parts)
             return
@@ -1615,7 +1656,7 @@ class App:
             active = current is not None and current["id"] == fmt["id"]
         if share_inline is None:
             share_inline = await self._check_inline()
-        return panels.format_view(fmt, share_inline=share_inline, active=active)
+        return panels.format_view(fmt, share_inline=share_inline, active=active, upload_on=self.cfg.upload_enabled)
 
     async def _show_format(self, event: Any, uid: int, fid: int, note: str = "") -> bool:
         fmt = await self.db.get_format(uid, fid)
@@ -1673,6 +1714,15 @@ class App:
         if action in ("v", "cx"):
             await self._show_format(event, uid, fid)
             await event.answer()
+        elif action == "in" and len(parts) == 4 and parts[3] == "channel":
+            if fmt["kind"] != "manual" or not self.cfg.upload_enabled:
+                await event.answer("Channel is an option of manual formats, while uploading to the Database Channel is set up.", alert=True)
+                return
+            await self._format_prompt(event, uid, "channel", fid)
+        elif action == "chc":
+            await self.db.update_format(uid, fid, channel=None)
+            await self._show_format(event, uid, fid)
+            await event.answer("Channel cleared.")
         elif action == "in" and len(parts) == 4 and parts[3] in (*FIELD_KINDS, "name"):
             await self._format_prompt(event, uid, parts[3], fid)
         elif action == "th":
@@ -1769,6 +1819,12 @@ class App:
                     msg = f"✅ Renamed to <b>{esc(value)}</b>."
                 else:
                     msg = value
+            elif field == "channel":
+                ok, channel, msg = parse_format_channel(text)
+                if ok:
+                    if await self.db.update_format(uid, p.fid, channel=channel) is None:
+                        raise LookupError
+                    msg = f"✅ Channel saved: <code>{channel}</code>" if channel else "✅ Channel cleared."
             else:
                 ok, msg = await apply_setting(FormatSettings(self.db, p.fid), uid, FIELD_KINDS[field], text)
         except FormatError as exc:

bot/config.py

@@ -10,6 +10,9 @@ class ConfigError(RuntimeError):
     pass
 
 
+DEFAULT_POST_BOT_ID = 8950016484  # the channel post bot: the only account whose answers in the communication chat count
+
+
 def _parse_ids(value: Optional[str]) -> Tuple[int, ...]:
     ids = []
     for chunk in (value or "").replace(";", ",").replace(" ", ",").split(","):
@@ -33,6 +36,27 @@ def _get_int(env: Mapping[str, str], name: str, default: int) -> int:
         raise ConfigError(f"{name} must be an integer, got {raw!r}") from exc
 
 
+def _env_any(env: Mapping[str, str], name: str) -> str:
+    """A variable by name whatever its capital letters (``DB_Channel``, ``DB_CHANNEL`` and ``db_channel`` are one setting)."""
+    raw = env.get(name)
+    if raw is None:
+        wanted = name.lower()
+        raw = next((v for k, v in env.items() if k.lower() == wanted), "")
+    return (raw or "").strip()
+
+
+def _get_chat(env: Mapping[str, str], name: str) -> Optional[int]:
+    """A chat id such as ``-1001234567890`` (unset or empty: ``None``; a bare channel id gets its ``-100`` in front)."""
+    raw = _env_any(env, name)
+    if not raw or raw.lower() in {"0", "off", "none", "false", "no"}:
+        return None
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise ConfigError(f"{name} must be a numeric chat id like -1001234567890, got {raw!r}") from exc
+    return -1_000_000_000_000 - value if value > 0 else value
+
+
 def _get_bool(env: Mapping[str, str], name: str, default: bool) -> bool:
     raw = env.get(name, "").strip().lower()
     if not raw:
@@ -58,11 +82,19 @@ class Config:
     fast_transfer: bool
     progress_interval: float
     stream_transfer: bool = True  # big files go straight from the download into the upload instead of through the disk
+    db_channel: Optional[int] = None  # the Database Channel a manually renamed episode can be uploaded to
+    comm_chat: Optional[int] = None  # the chat where this bot and the channel post bot talk to each other
+    post_bot_id: int = DEFAULT_POST_BOT_ID
 
     @property
     def max_file_bytes(self) -> int:
         return self.max_file_mb * 1024 * 1024
 
+    @property
+    def upload_enabled(self) -> bool:
+        """Uploading to the Database Channel (and having the post published) needs both chats."""
+        return bool(self.db_channel and self.comm_chat)
+
 
 def load_config(env: Optional[Mapping[str, str]] = None) -> Config:
     env = os.environ if env is None else env
@@ -79,6 +111,14 @@ def load_config(env: Optional[Mapping[str, str]] = None) -> Config:
         except ValueError as exc:
             raise ConfigError("LOG_CHANNEL must be a numeric chat id like -1001234567890") from exc
 
+    db_channel = _get_chat(env, "DB_CHANNEL")
+    comm_chat = _get_chat(env, "COMMUNICATION_CHAT")
+    post_bot_raw = _env_any(env, "CHANNEL_POST_BOT_ID")
+    try:
+        post_bot_id = int(post_bot_raw) if post_bot_raw else DEFAULT_POST_BOT_ID
+    except ValueError as exc:
+        raise ConfigError(f"CHANNEL_POST_BOT_ID must be a numeric user id, got {post_bot_raw!r}") from exc
+
     connections = max(1, min(_get_int(env, "TRANSFER_CONNECTIONS", 8), 16))
     max_jobs = max(1, min(_get_int(env, "MAX_CONCURRENT_JOBS", 1), 4))
 
@@ -99,4 +139,7 @@ def load_config(env: Optional[Mapping[str, str]] = None) -> Config:
         fast_transfer=_get_bool(env, "FAST_TRANSFER", True),
         progress_interval=max(2.0, float(env.get("PROGRESS_INTERVAL", "4") or 4)),
         stream_transfer=_get_bool(env, "STREAM_RENAME", True),
+        db_channel=db_channel,
+        comm_chat=comm_chat,
+        post_bot_id=post_bot_id,
     )

bot/db.py

@@ -59,6 +59,8 @@ DEFAULT_FORMAT: Dict[str, Any] = {
     "meta": {"title": "", "author": "", "artist": "", "audio": "", "subtitle": "", "video": ""},
     "keep_audio": [],
     "keep_subtitle": [],
+    # a manual format can name the channel its episodes are published in (see dbflow); None: no post is made
+    "channel": None,
 }
 
 
@@ -165,6 +167,9 @@ def _merge_format(stored: Dict[str, Any]) -> Dict[str, Any]:
     for key in ("keep_audio", "keep_subtitle"):
         if isinstance(stored.get(key), list):
             merged[key] = [str(t) for t in stored[key]]
+    channel = stored.get("channel")
+    if merged["kind"] == "manual" and isinstance(channel, int) and not isinstance(channel, bool):
+        merged["channel"] = channel  # only a manual format has one: an auto format renames files that nobody uploads on
     return merged
 
 
@@ -389,7 +394,7 @@ class Database:
         return _format_dict(fid, name, raw, False)
 
     async def update_format(self, uid: int, fid: int, **changes: Any) -> Optional[Dict[str, Any]]:
-        """Change ``name``, ``template``, ``caption``, ``keep_audio``, ``keep_subtitle`` or merge a ``meta`` dict; ``None`` if it is gone."""
+        """Change ``name``, ``template``, ``caption``, ``keep_audio``, ``keep_subtitle``, ``channel`` or merge a ``meta`` dict; ``None`` if it is gone."""
         async with self._sessions() as session:
             row = await session.get(Format, fid)
             if row is None or row.user_id != uid:
@@ -402,6 +407,8 @@ class Database:
                     row.name = value
                 elif key == "meta":
                     current["meta"].update(value)
+                elif key == "channel":
+                    current["channel"] = value if current["kind"] == "manual" else None
                 elif key in DEFAULT_FORMAT:
                     current[key] = value
                 else:
@@ -478,7 +485,8 @@ class Database:
 
         The copy is of the same kind as the original unless ``kind`` says otherwise: that is how a user copies an auto format
         to the manual ones and the other way round (``owner_id`` is then the user themselves).  ``None`` when that format is
-        gone; ``FormatError("limit")`` when ``uid`` already has ``MAX_FORMATS`` of that kind.
+        gone; ``FormatError("limit")`` when ``uid`` already has ``MAX_FORMATS`` of that kind.  The Channel of a manual format is
+        kept only by its owner's own manual copy.
         """
         if kind is not None and kind not in FORMAT_KINDS:
             raise KeyError(kind)
@@ -491,7 +499,10 @@ class Database:
             if len(names) >= MAX_FORMATS:
                 raise FormatError("limit")
             name = _free_name(source.name, names, name_max)
-            data = _format_json({**_merge_format(_stored(source.data)), "kind": target})
+            copied = _merge_format(_stored(source.data))
+            if uid != owner_id or target != "manual":
+                copied["channel"] = None  # somebody else's copy (or an auto one) does not publish in the original's channel
+            data = _format_json({**copied, "kind": target})
             row = Format(user_id=uid, name=name, data=data, thumb=source.thumb)
             session.add(row)
             await session.flush()

bot/dbflow.py

@@ -0,0 +1,382 @@
+"""Upload a manually renamed episode to the Database Channel, and have the channel post bot publish it.
+
+All of this exists only while DB_CHANNEL and COMMUNICATION_CHAT are both set.
+
+1. After /rename with a manual format has sent the renamed episode, ``DbFlow.offer`` asks (Yes / No) whether to upload it.
+2. Yes: the message becomes "Processing…" and the episode is copied into the Database Channel on Telegram's side (forwarded
+   without the "Forwarded from" line): nothing is downloaded or uploaded again.
+3. When the format has a Channel, "File uploaded to the Database Channel" is then written in the communication chat, with the
+   episode number, that channel id, the file name and the id of the copy.  The channel post bot answers there, replying to it:
+   "message received", then "post published successfully" or "publication failed: <reason>".
+4. Those answers are read here (``on_chat_message``) and shown in the Processing message.  No answer at all: the channel post
+   bot is down.
+
+A format without a Channel is only uploaded: no notice is written and nothing is published.
+
+Nothing about a running upload is stored: the buttons carry what is needed, and a restart in the middle just leaves the
+Processing message as it was.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+import secrets
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional, Set, Tuple
+
+from telethon import errors
+
+from . import panels
+from .config import Config
+from .parser import parse_info, split_ext
+from .utils import clip, esc
+
+log = logging.getLogger(__name__)
+
+PREFIX = "dbu"  # callback data: dbu:y:<chat>:<message>:<channel or 0> (Yes), dbu:n (No)
+ACK_WAIT = 60.0  # seconds the channel post bot has to say "message received" before it counts as down
+RESULT_WAIT = 900.0  # ... and then to say whether the post was published (it waits at least 10 s for the Share URL button)
+NOTICE_HEAD = "File uploaded to the Database Channel"
+
+_FAILED = re.compile(r"(?i)publication\s+failed\s*[:\-–]?[ \t]*(.*)")
+_PUBLISHED = re.compile(r"(?i)post\s+published\s+successfully")
+_RECEIVED = re.compile(r"(?i)message\s+received")
+
+
+def episode_of(file_name: str) -> str:
+    """The episode number of a file name, as written in it (``05``); empty when it has none."""
+    return parse_info(split_ext(file_name or "")[0]).episode
+
+
+def line_of(label: str, text: str) -> str:
+    """The value of a ``Label: value`` line of a message; empty when there is no such line."""
+    m = re.search(rf"(?im)^[ \t]*{label}[ \t]*:[ \t]*(.+?)[ \t]*$", text or "")
+    return m.group(1).strip() if m else ""
+
+
+def classify(text: Optional[str]) -> Tuple[str, str]:
+    """What a message of the channel post bot says: ``("fail", reason)``, ``("ok", "")``, ``("ack", "")`` or ``("other", "")``."""
+    text = text or ""
+    failed = _FAILED.search(text)
+    if failed:
+        return "fail", " ".join(failed.group(1).split())
+    if _PUBLISHED.search(text):
+        return "ok", ""
+    if _RECEIVED.search(text):
+        return "ack", ""
+    return "other", ""
+
+
+def notice_text(*, episode: str, channel: int, file: str, post_id: Optional[int], ref: str) -> str:
+    """The message for the communication chat.  The channel post bot reads the Episode and the Channel from it."""
+    lines = [NOTICE_HEAD]
+    if episode:
+        lines.append(f"Episode: {episode}")
+    lines.append(f"Channel: {channel}")
+    lines.append(f"File: {file}")
+    if post_id:
+        lines.append(f"Post: {post_id}")
+    lines.append(f"Ref: {ref}")
+    return "\n".join(lines)
+
+
+def reply_target(event: Any) -> Optional[int]:
+    """The id of the message an incoming message replies to."""
+    message = getattr(event, "message", None)
+    return getattr(message, "reply_to_msg_id", None) or getattr(event, "reply_to_msg_id", None) or None
+
+
+def _name(name: str) -> str:
+    return f"<code>{esc(clip(name, 120))}</code>" if name else ""
+
+
+def _why(exc: BaseException) -> str:
+    return esc(clip(f"{type(exc).__name__}: {exc}", 250))
+
+
+def offer_text(name: str, episode: str, channel: Optional[int]) -> str:
+    lines = ["📤 <b>Upload this episode to the Database Channel?</b>", _name(name)]
+    if channel:
+        lines.append(f"🎞 Episode: <b>{esc(episode)}</b>" if episode else "⚠️ No episode number found in the file name (the post uses it).")
+        lines.append(f"📣 Channel: <code>{channel}</code>: the channel post bot posts it there.")
+    else:
+        lines.append("This format has no Channel, so the episode is only uploaded: no post is made.")
+    return "\n".join(line for line in lines if line)
+
+
+def processing_text(name: str, *steps: str) -> str:
+    return "\n".join(line for line in ["⏳ <b>Processing…</b>", _name(name), *steps] if line)
+
+
+def published_text(name: str, reply: str, channel: int) -> str:
+    where = line_of("channel", reply) or str(channel)
+    link = line_of("post", reply)
+    lines = ["✅ <b>Post published successfully</b>", _name(name), f"📣 {esc(clip(where, 80))}"]
+    if link:
+        lines.append(esc(clip(link, 200)))
+    return "\n".join(line for line in lines if line)
+
+
+def failed_text(name: str, reason: str) -> str:
+    lines = ["❌ <b>Publication failed</b>", _name(name), esc(clip(reason or "No reason was given.", 600)), "The episode is in the Database Channel."]
+    return "\n".join(line for line in lines if line)
+
+
+def down_text(name: str, seconds: float) -> str:
+    return "\n".join(
+        [
+            "⚠️ <b>The channel post bot is down or not responding</b>",
+            _name(name),
+            f"It did not answer in the communication chat within {int(seconds)} seconds. "
+            "The episode is in the Database Channel, but nothing was published.",
+        ]
+    )
+
+
+def silent_text(name: str, seconds: float) -> str:
+    return "\n".join(
+        [
+            "⚠️ <b>No word from the channel post bot</b>",
+            _name(name),
+            f"It got the message but did not say whether the post was published in {int(seconds // 60)} minutes. Check the channel.",
+        ]
+    )
+
+
+@dataclass
+class Flow:
+    """An upload that waits for the channel post bot's answers."""
+
+    ref: str  # in the notice and echoed in every answer: how the answers find their way back
+    inbox: "asyncio.Queue[Tuple[str, str]]" = field(default_factory=asyncio.Queue)  # (ack | ok | fail, the message)
+    notice_id: int = 0  # the notice in the communication chat; an answer is a reply to it
+
+
+class DbFlow:
+    def __init__(self, client: Any, cfg: Config, *, ack_wait: float = ACK_WAIT, result_wait: float = RESULT_WAIT):
+        self.client = client
+        self.cfg = cfg
+        self.ack_wait = ack_wait
+        self.result_wait = result_wait
+        self.flows: Dict[str, Flow] = {}
+        self._busy: Set[Tuple[int, int]] = set()  # (chat, offer message) of the uploads under way: a second tap does nothing
+        self._tasks: Set["asyncio.Future[Any]"] = set()
+
+    # ------------------------------------------------------------------ the offer
+    async def offer(self, job: Any, target: int, sent: Any) -> None:
+        """Ask whether the episode that was just sent (``sent``, in ``target``) goes to the Database Channel too."""
+        sent_id = getattr(sent, "id", None)
+        if not isinstance(sent_id, int):
+            return  # nothing to point at
+        channel = (job.fmt or {}).get("channel")
+        rows = [[("✅ Yes", f"{PREFIX}:y:{target}:{sent_id}:{channel or 0}"), ("❌ No", f"{PREFIX}:n")]]
+        await self.client.send_message(
+            job.chat_id, offer_text(job.new_name, episode_of(job.new_name), channel), buttons=panels.to_buttons(rows),
+            parse_mode="html", link_preview=False, reply_to=sent_id if target == job.chat_id else job.msg_id,
+        )
+
+    async def on_callback(self, event: Any, parts: list) -> None:
+        action = parts[1] if len(parts) > 1 else ""
+        if action == "n":
+            await event.answer()
+            await event.edit("👍 Okay, it is not uploaded to the Database Channel.", buttons=None, parse_mode="html")
+            return
+        try:
+            if action != "y" or len(parts) != 5:
+                raise ValueError(parts)
+            source_chat, source_id, channel = int(parts[2]), int(parts[3]), int(parts[4])
+        except ValueError:
+            await event.answer()
+            return
+        key = (event.chat_id, event.message_id)
+        if key in self._busy:
+            await event.answer("I'm already on it…")
+            return
+        self._busy.add(key)
+        try:
+            await event.answer()
+            try:
+                await event.edit(processing_text(""), buttons=None, parse_mode="html")
+            except errors.MessageNotModifiedError:
+                pass
+        except BaseException:
+            self._busy.discard(key)
+            raise
+        self._spawn(self._run(key, source_chat, source_id, channel or None))
+
+    # --------------------------------------------------------------- the upload
+    async def _run(self, key: Tuple[int, int], source_chat: int, source_id: int, channel: Optional[int]) -> None:
+        chat_id, msg_id = key
+        flow: Optional[Flow] = None
+        name = ""
+        try:
+            msg = await self.client.get_messages(source_chat, ids=source_id)
+            if msg is None or getattr(msg, "file", None) is None:
+                await self._show(
+                    chat_id, msg_id, "❌ <b>That episode is not there anymore</b>\nIt was deleted from the chat, so there is nothing to upload."
+                )
+                return
+            name = getattr(msg.file, "name", None) or ""
+            episode = episode_of(name)
+
+            # ---- 1. the Database Channel
+            try:
+                copy = await self._copy(msg, source_chat)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # noqa: BLE001
+                log.warning("could not upload %s to the Database Channel: %s: %s", name, type(exc).__name__, exc)
+                retry = [[("🔁 Try again", f"{PREFIX}:y:{source_chat}:{source_id}:{channel or 0}"), ("✖ Close", "x:close")]]
+                await self._show(
+                    chat_id, msg_id,
+                    f"❌ <b>The upload to the Database Channel failed</b>\n{_name(name)}\n{_why(exc)}\n"
+                    "Check that the bot is an admin of the Database Channel and can post there.",
+                    retry,
+                )
+                return
+            if channel is None:
+                await self._show(
+                    chat_id, msg_id,
+                    f"✅ <b>Uploaded to the Database Channel</b>\n{_name(name)}\nThis format has no Channel, so no post is made.",
+                )
+                return
+
+            # ---- 2. the communication chat
+            flow = Flow(ref=secrets.token_hex(3))
+            self.flows[flow.ref] = flow  # before the notice goes out, so that no answer can come back too early
+            uploaded = "✅ Uploaded to the Database Channel"
+            await self._show(chat_id, msg_id, processing_text(name, uploaded, "⏳ Waiting for the channel post bot…"))
+            try:
+                note = await self.client.send_message(
+                    self.cfg.comm_chat,
+                    notice_text(episode=episode, channel=channel, file=name, post_id=_id_of(copy), ref=flow.ref),
+                    parse_mode=None, link_preview=False,
+                )
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # noqa: BLE001
+                log.warning("could not write in the communication chat: %s: %s", type(exc).__name__, exc)
+                await self._show(
+                    chat_id, msg_id,
+                    f"⚠️ <b>Uploaded, but I could not write in the communication chat</b>\n{_name(name)}\n{_why(exc)}\n"
+                    "Nothing was published. Is the bot a member of that chat?",
+                )
+                return
+            flow.notice_id = _id_of(note) or 0
+
+            # ---- 3. what the channel post bot says
+            verdict, reply = await self._answers(flow, chat_id, msg_id, name, uploaded)
+            if verdict == "ok":
+                text = published_text(name, reply, channel)
+            elif verdict == "fail":
+                text = failed_text(name, classify(reply)[1])
+            elif verdict == "down":
+                text = down_text(name, self.ack_wait)
+            else:
+                text = silent_text(name, self.result_wait)
+            await self._show(chat_id, msg_id, text)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            log.exception("the upload to the Database Channel went wrong")
+            await self._show(chat_id, msg_id, f"❌ <b>Something went wrong</b>\n{_name(name)}\n{_why(exc)}")
+        finally:
+            self._busy.discard(key)
+            if flow is not None:
+                self.flows.pop(flow.ref, None)
+
+    async def _copy(self, msg: Any, source_chat: int) -> Any:
+        """The episode in the Database Channel, copied on Telegram's side: forwarded without the author, so nothing is downloaded or
+        uploaded.  When Telegram refuses that (a chat that forbids forwarding), the same file is sent again by its id, which is as quick."""
+        try:
+            return await self.client.forward_messages(self.cfg.db_channel, msg.id, from_peer=source_chat, drop_author=True)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            log.info("forwarding to the Database Channel was refused (%s: %s); sending the file by its id", type(exc).__name__, exc)
+        return await self.client.send_file(
+            self.cfg.db_channel, msg.media, caption=getattr(msg, "message", "") or "",
+            formatting_entities=getattr(msg, "entities", None) or None, parse_mode=None,
+        )
+
+    async def _answers(self, flow: Flow, chat_id: int, msg_id: int, name: str, uploaded: str) -> Tuple[str, str]:
+        """Wait for the channel post bot: ``("ok" | "fail", its message)``, ``("down", "")`` when it never acknowledged the notice,
+        ``("silent", "")`` when it acknowledged it but never said how it went."""
+        loop = asyncio.get_running_loop()
+        acked = False
+        deadline = loop.time() + self.ack_wait
+        while True:
+            try:
+                kind, text = await asyncio.wait_for(flow.inbox.get(), timeout=max(0.01, deadline - loop.time()))
+            except asyncio.TimeoutError:
+                return ("silent" if acked else "down"), ""
+            if kind != "ack":
+                return kind, text
+            if not acked:
+                acked = True
+                deadline = loop.time() + self.result_wait
+                await self._show(chat_id, msg_id, processing_text(name, uploaded, "📥 The channel post bot got it and is publishing…"))
+
+    async def _show(self, chat_id: int, msg_id: int, text: str, rows: Optional[list] = None) -> None:
+        try:
+            await self.client.edit_message(
+                chat_id, msg_id, text, buttons=panels.to_buttons(rows) if rows else None, parse_mode="html", link_preview=False
+            )
+        except errors.MessageNotModifiedError:
+            pass
+        except Exception as exc:  # noqa: BLE001
+            log.warning("could not update the upload message: %s: %s", type(exc).__name__, exc)
+
+    # ------------------------------------------------- the channel post bot's answers
+    async def on_chat_message(self, event: Any) -> None:
+        """A message of the channel post bot in the communication chat."""
+        try:
+            if getattr(event, "chat_id", None) != self.cfg.comm_chat or getattr(event, "sender_id", None) != self.cfg.post_bot_id:
+                return  # (the event filter does this already)
+            text = getattr(event, "raw_text", None) or ""
+            kind = classify(text)[0]
+            if kind == "other":
+                return
+            flow = self._flow_for(event, text)
+            if flow is not None:
+                flow.inbox.put_nowait((kind, text))
+        except Exception:  # noqa: BLE001
+            log.exception("could not read a message in the communication chat")
+
+    def _flow_for(self, event: Any, text: str) -> Optional[Flow]:
+        """The upload an answer belongs to: the one whose Ref it repeats, or whose notice it replies to."""
+        ref = line_of("ref", text)
+        if ref and ref in self.flows:
+            return self.flows[ref]
+        target = reply_target(event)
+        if target:
+            return next((f for f in self.flows.values() if f.notice_id == target), None)
+        if len(self.flows) == 1:  # a plain message, and only one upload waiting for an answer
+            return next(iter(self.flows.values()))
+        return None
+
+    # ---------------------------------------------------------------- plumbing
+    def _spawn(self, coro: Any) -> "asyncio.Future[Any]":
+        task = asyncio.ensure_future(coro)
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+        return task
+
+    async def idle(self) -> None:
+        """Wait until no upload is under way (the tests use this)."""
+        while self._tasks:
+            await asyncio.gather(*list(self._tasks), return_exceptions=True)
+
+    async def shutdown(self) -> None:
+        for task in list(self._tasks):
+            task.cancel()
+        await asyncio.gather(*list(self._tasks), return_exceptions=True)
+        self._tasks.clear()
+
+
+def _id_of(sent: Any) -> Optional[int]:
+    if isinstance(sent, (list, tuple)):
+        sent = sent[0] if sent else None
+    value = getattr(sent, "id", None)
+    return value if isinstance(value, int) else None

bot/panels.py

@@ -346,13 +346,17 @@ def _format_lines(fmt: Dict[str, Any], *, thumb_yes: str, thumb_no: str) -> List
     return lines
 
 
-def format_view(fmt: Dict[str, Any], *, share_inline: bool = False, active: bool = False) -> Panel:
+def format_view(fmt: Dict[str, Any], *, share_inline: bool = False, active: bool = False, upload_on: bool = False) -> Panel:
     """A format's screen.  ``active``: it is the auto format in use.  ``share_inline``: the Share format button opens the chat
     picker straight away; that only works while inline mode is on for the bot (Telegram refuses the message otherwise), so
-    without it the button explains how."""
+    without it the button explains how.  ``upload_on``: uploading to the Database Channel is set up, so a manual format shows
+    its Channel option."""
     kind = fmt.get("kind", "manual")
+    has_channel_option = upload_on and kind == "manual"
     lines = [f"<b>🗂 {esc(clip(fmt['name'], 30))}</b> · {KIND_ICONS[kind]} {KIND_LABELS[kind]}", ""]
     lines += _format_lines(fmt, thumb_yes="✅ saved", thumb_no=THUMB_NONE)
+    if has_channel_option:
+        lines.append(f"{DOT} <b>Channel</b> ▸ " + (_code(str(fmt["channel"])) if fmt.get("channel") else "— (optional)"))
     if kind == "manual":
         lines += ["", "Use it with /rename. Mode, file type and the rest come from your normal settings."]
     elif active:
@@ -375,6 +379,13 @@ def format_view(fmt: Dict[str, Any], *, share_inline: bool = False, active: bool
         [("Title", f"fm:in:{fid}:title"), ("Author", f"fm:in:{fid}:author"), ("Artist", f"fm:in:{fid}:artist")],
         [("Audio", f"fm:in:{fid}:audio"), ("Subtitle", f"fm:in:{fid}:subtitle"), ("Video", f"fm:in:{fid}:video")],
         [("Keep Audio", f"fm:in:{fid}:keep_audio"), ("Keep Subtitle", f"fm:in:{fid}:keep_subtitle")],
+    ]
+    if has_channel_option:
+        channel_row: Row = [("Channel", f"fm:in:{fid}:channel")]
+        if fmt.get("channel"):
+            channel_row.append(("🗑 Clear Channel", f"fm:chc:{fid}"))
+        rows.append(channel_row)
+    rows += [
         [("🏷 Apply default metadata", f"fm:dm:{fid}")],
         [(f"📋 Copy to {other_kind(kind)} formats", f"fm:cp:{fid}")],
     ]
@@ -612,6 +623,11 @@ FORMAT_PROMPTS: Dict[str, str] = {
     "title": PROMPTS["mt_title"], "author": PROMPTS["mt_author"], "artist": PROMPTS["mt_artist"],
     "audio": PROMPTS["mt_audio"], "subtitle": PROMPTS["mt_subtitle"], "video": PROMPTS["mt_video"],
     "keep_audio": PROMPTS["keep_audio"], "keep_subtitle": PROMPTS["keep_subtitle"],
+    "channel": (
+        "<b>Send the id of the channel</b> this format's episodes are published in, like <code>-1001234567890</code>.\n\n"
+        "After /rename I can upload the episode to the Database Channel; the channel post bot then posts it in this channel.\n"
+        "It is optional: send <code>off</code> to clear it (the episode is then only uploaded, nothing is posted)."
+    ),
 }
 
 

bot/pipeline.py

@@ -7,7 +7,7 @@ import shutil
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
 
 from telethon import errors, types
 
@@ -112,6 +112,9 @@ class Renamer:
         self.client = client
         self.cfg = cfg
         self.db = db
+        # called as ``await on_delivered(job, chat, sent)`` once a file renamed with a manual format is in its chat (the offer
+        # to upload it to the Database Channel, see dbflow); ``None``: nothing to do
+        self.on_delivered: Optional[Callable[[Job, int, Any], Awaitable[None]]] = None
 
     # ------------------------------------------------------------------
     async def _fresh_message(self, job: Job) -> Any:
@@ -260,11 +263,11 @@ class Renamer:
         if target == job.chat_id:
             kwargs["reply_to"] = job.msg_id
         try:
-            await self.client.send_file(target, input_file, **kwargs)
+            sent = await self.client.send_file(target, input_file, **kwargs)
         except (errors.EntityBoundsInvalidError, errors.MediaCaptionTooLongError, errors.MessageTooLongError):
             # the caption template produced something Telegram rejects: send the file without a caption
             kwargs.update(caption=None)
-            await self.client.send_file(target, input_file, **kwargs)
+            sent = await self.client.send_file(target, input_file, **kwargs)
             warnings.append("The caption was rejected by Telegram (check your caption template), so it was left out.")
 
         await self.db.record_job(job.user_id, final_size)
@@ -275,6 +278,16 @@ class Renamer:
         await reporter.finish(
             f"✅ <b>Done</b> in {human_duration(took)}\n<code>{esc(clip(job.new_name, 120))}</code>\n{human_size(final_size)}{rates}{where}{note}"
         )
+        await self._after_delivery(job, target, sent)
+
+    async def _after_delivery(self, job: Job, target: int, sent: Any) -> None:
+        """A file renamed with a manual format may go on to the Database Channel: the offer comes after the file itself."""
+        if self.on_delivered is None or job.fmt is None or job.fmt.get("kind") != "manual":
+            return
+        try:
+            await self.on_delivered(job, target, sent)
+        except Exception:  # noqa: BLE001 - the rename is done whatever happens here
+            log.exception("could not offer the upload to the Database Channel")
 
     async def _run(self, job: Job, s: Dict[str, Any], work: Path, reporter: ProgressReporter, started: float) -> None:
         client, cfg = self.client, self.cfg

bot/settings_ops.py

@@ -136,3 +136,28 @@ def parse_channel(text: str) -> Tuple[bool, Optional[int], str]:
     if re.fullmatch(r"-?\d{5,20}", text):
         return True, int(text), ""
     return False, None, "❌ Send the numeric chat id, like <code>-1001234567890</code> (not @username)."
+
+
+_CHANNEL_LINK = re.compile(r"(?:https?://)?(?:www\.)?t\.me/c/(\d{5,15})(?:/\d+){0,2}/?", re.I)
+CHANNEL_ID_HINT = "❌ Send the channel id, like <code>-1001234567890</code> (a channel id starts with -100), or <code>off</code> to clear it."
+
+
+def parse_format_channel(text: str) -> Tuple[bool, Optional[int], str]:
+    """The Channel option of a manual format: ``-1001234567890`` -> that id, ``off`` -> ``None``.
+
+    A bare channel id (``1234567890``) and a link to a post of a private channel (``t.me/c/1234567890/5``) work too: both get
+    the ``-100`` in front that Telegram puts on channel ids.
+    """
+    text = (text or "").strip()
+    if _is_clear(text):
+        return True, None, ""
+    link = _CHANNEL_LINK.fullmatch(text)
+    if link:
+        return True, -1_000_000_000_000 - int(link.group(1)), ""
+    if re.fullmatch(r"-?\d{5,20}", text):
+        value = int(text)
+        if value > 0:
+            return True, -1_000_000_000_000 - value, ""
+        if value <= -1_000_000_000_001:
+            return True, value, ""
+    return False, None, CHANNEL_ID_HINT

tests/test_dbflow.py

@@ -0,0 +1,711 @@
+"""Uploading a manually renamed episode to the Database Channel and having the channel post bot publish it:
+the two chats in the settings, the Channel option of a manual format, the offer, the upload, the notice in the
+communication chat and the answers of the channel post bot."""
+import asyncio
+import re
+import time
+from types import SimpleNamespace
+
+import pytest
+from telethon import errors
+
+from bot import panels
+from bot.app import App
+from bot.config import ConfigError, load_config
+from bot.db import Database, Format
+from bot.dbflow import classify, episode_of, line_of, notice_text, reply_target
+from bot.formats import default_auto_format
+from bot.pipeline import Renamer
+from bot.queue import Job
+from bot.settings_ops import parse_format_channel
+
+from .appkit import ADMIN_CFG, OWNER, STRANGER, click, data_of, env_for, say
+from .tgfakes import FakeClient, FakeEvent, dispatch, document_message, flat_buttons
+
+DB = -1002000000001  # the Database Channel
+COMM = -1003000000002  # the communication chat
+POST_BOT = 8950016484
+SHOW = -1004000000003  # the channel a format publishes in
+NAME = "[S01E05] Tokyo Revengers [1080p].mkv"
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def up(tmp_path):
+    """Like ``ctx``, with uploading to the Database Channel switched on (and short waits)."""
+    cfg = load_config(env_for(tmp_path, DB_Channel=str(DB), Communication_chat=str(COMM)))
+    db = Database(f"sqlite+aiosqlite:///{tmp_path}/app.db")
+    await db.init()
+    client = FakeClient()
+    app = App(client, cfg, db)
+    app.dbflow.ack_wait, app.dbflow.result_wait = 2.0, 4.0
+    app._register()
+    yield SimpleNamespace(app=app, client=client, db=db, cfg=cfg, tmp=tmp_path, flow=app.dbflow)
+    await app.shutdown()
+    await db.close()
+
+
+def notices(up):
+    """What was written in the communication chat: ``[(message id, text, kwargs)]``."""
+    return [(2000 + n, text, kw) for n, (entity, text, kw) in enumerate(up.client.sent, 1) if entity == COMM]
+
+
+async def wait_for(predicate, timeout=3.0):
+    end = time.monotonic() + timeout
+    while not predicate():
+        assert time.monotonic() < end, "timed out"
+        await asyncio.sleep(0.005)
+
+
+async def the_notice(up, nth=1):
+    await wait_for(lambda: len(notices(up)) >= nth)
+    return notices(up)[nth - 1]
+
+
+def ref_of(text):
+    return re.search(r"(?m)^Ref: (\w+)$", text).group(1)
+
+
+def last_edit(up):
+    return up.client.edited[-1][1]
+
+
+async def post_bot_says(up, text, *, reply_to=None, sender=POST_BOT, chat=COMM):
+    ev = FakeEvent(sender, text, chat=chat, private=False, reply_to_id=reply_to)
+    await dispatch(up.client, ev)
+    return ev
+
+
+async def start_upload(up, *, channel=SHOW, name=NAME, mid=555, offer_id=77, caption=""):
+    """The renamed episode is in the user's chat, and the offer is answered with Yes."""
+    up.client.messages[mid] = document_message(name, 10 * 1024 * 1024, mid=mid, caption=caption)
+    return await click(up, OWNER, f"dbu:y:{OWNER}:{mid}:{channel or 0}", msg_id=offer_id)
+
+
+def manual_job(name=NAME, channel=SHOW, kind="manual", chat=OWNER):
+    fmt = {"id": 5, "name": "One Piece", "kind": kind, "channel": channel}
+    return Job(
+        id=1, user_id=OWNER, chat_id=chat, msg_id=40, file_name="original.mkv", size=1, new_name=name, fmt=fmt,
+        status_msg=SimpleNamespace(id=99),
+    )
+
+
+# ---------------------------------------------------------------------------
+# settings
+# ---------------------------------------------------------------------------
+
+
+def test_both_chats_switch_uploading_on(tmp_path):
+    cfg = load_config(env_for(tmp_path, DB_CHANNEL=str(DB), COMMUNICATION_CHAT=str(COMM)))
+    assert (cfg.db_channel, cfg.comm_chat, cfg.upload_enabled, cfg.post_bot_id) == (DB, COMM, True, 8950016484)
+
+
+@pytest.mark.parametrize("db_name, comm_name", [("DB_CHANNEL", "COMMUNICATION_CHAT"), ("db_channel", "communication_chat"), ("DB_Channel", "Communication_chat")])
+def test_the_names_may_be_written_in_any_case(tmp_path, db_name, comm_name):
+    cfg = load_config(env_for(tmp_path, **{db_name: str(DB), comm_name: str(COMM)}))
+    assert (cfg.db_channel, cfg.comm_chat, cfg.upload_enabled) == (DB, COMM, True)
+
+
+@pytest.mark.parametrize("extra", [{"DB_CHANNEL": str(DB)}, {"COMMUNICATION_CHAT": str(COMM)}, {}])
+def test_one_chat_alone_does_not_switch_it_on(tmp_path, extra):
+    assert load_config(env_for(tmp_path, **extra)).upload_enabled is False
+
+
+def test_a_bare_channel_id_gets_its_minus_100(tmp_path):
+    cfg = load_config(env_for(tmp_path, DB_CHANNEL="2000000001", COMMUNICATION_CHAT="3000000002"))
+    assert (cfg.db_channel, cfg.comm_chat) == (-1002000000001, -1003000000002)
+
+
+@pytest.mark.parametrize("name", ["DB_CHANNEL", "COMMUNICATION_CHAT", "CHANNEL_POST_BOT_ID"])
+def test_an_id_that_is_not_a_number_is_a_configuration_error(tmp_path, name):
+    with pytest.raises(ConfigError, match=name):
+        load_config(env_for(tmp_path, **{name: "@somewhere"}))
+
+
+@pytest.mark.parametrize("value", ["", "0", "off", "OFF", "none"])
+def test_empty_or_off_means_not_set(tmp_path, value):
+    cfg = load_config(env_for(tmp_path, DB_CHANNEL=value, COMMUNICATION_CHAT=value))
+    assert cfg.db_channel is None and cfg.comm_chat is None and cfg.upload_enabled is False
+
+
+def test_the_post_bot_id_can_be_changed(tmp_path):
+    assert load_config(env_for(tmp_path, CHANNEL_POST_BOT_ID="12345")).post_bot_id == 12345
+
+
+async def test_without_both_chats_nothing_is_offered_and_the_buttons_say_so(ctx):
+    assert ctx.app.dbflow is None and ctx.app.renamer.on_delivered is None
+    cb = await click(ctx, OWNER, f"dbu:y:{OWNER}:555:{SHOW}")
+    assert cb.alerts and "switched off" in cb.alerts[0] and not ctx.client.forwarded and not ctx.client.sent
+
+
+async def test_with_both_chats_the_offer_is_hooked_into_the_pipeline(up):
+    assert up.app.renamer.on_delivered == up.flow.offer
+
+
+# ---------------------------------------------------------------------------
+# the Channel option of a manual format
+# ---------------------------------------------------------------------------
+
+
+async def test_a_manual_format_keeps_its_channel(ctx):
+    fmt = await ctx.db.create_format(OWNER, "One Piece")
+    assert fmt["channel"] is None
+    assert (await ctx.db.update_format(OWNER, fmt["id"], channel=SHOW))["channel"] == SHOW
+    assert (await ctx.db.get_format(OWNER, fmt["id"]))["channel"] == SHOW
+    assert [f["channel"] for f in await ctx.db.list_formats(OWNER, "manual")] == [SHOW]
+    assert (await ctx.db.update_format(OWNER, fmt["id"], channel=None))["channel"] is None
+
+
+async def test_an_auto_format_has_no_channel(ctx):
+    fmt = await ctx.db.create_format(OWNER, "Auto One", kind="auto")
+    assert (await ctx.db.update_format(OWNER, fmt["id"], channel=SHOW))["channel"] is None
+
+
+async def test_formats_saved_before_the_option_have_no_channel(ctx):
+    async with ctx.db._sessions() as session:
+        session.add(Format(user_id=OWNER, name="Old", data='{"template": "{title}", "kind": "manual"}'))
+        session.add(Format(user_id=OWNER, name="Odd", data='{"kind": "manual", "channel": "-100123"}'))  # not a number: ignored
+        await session.commit()
+    assert [f["channel"] for f in await ctx.db.list_formats(OWNER)] == [None, None]
+
+
+async def test_a_copy_keeps_the_channel_only_for_the_owners_own_manual_copy(ctx):
+    fmt = await ctx.db.create_format(OWNER, "One Piece")
+    await ctx.db.update_format(OWNER, fmt["id"], channel=SHOW)
+    auto = await ctx.db.copy_format(OWNER, OWNER, fmt["id"], kind="auto")
+    assert auto["kind"] == "auto" and auto["channel"] is None
+    again = await ctx.db.copy_format(OWNER, OWNER, fmt["id"])
+    assert again["kind"] == "manual" and again["channel"] == SHOW
+    theirs = await ctx.db.copy_format(ADMIN_CFG, OWNER, fmt["id"])  # somebody else adds it from a share card
+    assert theirs["kind"] == "manual" and theirs["channel"] is None
+    back = await ctx.db.copy_format(OWNER, OWNER, auto["id"], kind="manual")
+    assert back["channel"] is None
+    assert (await ctx.db.get_format(OWNER, fmt["id"]))["channel"] == SHOW  # the original is untouched
+
+
+@pytest.mark.parametrize(
+    "text, expected",
+    [
+        ("-1001234567890", -1001234567890),
+        ("  -1001234567890 ", -1001234567890),
+        ("1234567890", -1001234567890),
+        ("https://t.me/c/1234567890/5", -1001234567890),
+        ("t.me/c/1234567890", -1001234567890),
+        ("off", None), ("None", None), ("-", None), ("clear", None),
+    ],
+)
+def test_a_channel_id_is_read(text, expected):
+    assert parse_format_channel(text) == (True, expected, "")
+
+
+@pytest.mark.parametrize("text", ["", "abc", "@mychannel", "https://t.me/mychannel", "-12345", "123", "-1001234567890x", "12 34567890"])
+def test_anything_else_is_refused_with_the_way_to_write_it(text):
+    ok, value, message = parse_format_channel(text)
+    assert not ok and value is None and "-1001234567890" in message
+
+
+async def test_without_the_two_chats_the_format_has_no_channel_option(ctx):
+    fmt = await ctx.db.create_format(OWNER, "One Piece")
+    text, buttons = (await click(ctx, OWNER, f"fm:v:{fmt['id']}")).edits[-1]
+    assert "Channel" not in text and not [d for d in data_of(buttons) if "channel" in d]
+    cb = await click(ctx, OWNER, f"fm:in:{fmt['id']}:channel")
+    assert cb.alerts and OWNER not in ctx.app.pending
+
+
+async def test_a_manual_format_shows_the_channel_option_and_an_auto_format_does_not(up):
+    fmt = await up.db.create_format(OWNER, "One Piece")
+    i = fmt["id"]
+    text, buttons = (await click(up, OWNER, f"fm:v:{i}")).edits[-1]
+    assert "<b>Channel</b> ▸ — (optional)" in text
+    assert f"fm:in:{i}:channel" in data_of(buttons) and f"fm:chc:{i}" not in data_of(buttons)
+    auto = await up.db.create_format(OWNER, "Auto One", kind="auto")
+    text, buttons = (await click(up, OWNER, f"fm:v:{auto['id']}")).edits[-1]
+    assert "Channel" not in text and f"fm:in:{auto['id']}:channel" not in data_of(buttons)
+    cb = await click(up, OWNER, f"fm:in:{auto['id']}:channel")
+    assert cb.alerts and OWNER not in up.app.pending
+
+
+async def test_the_channel_is_typed_in_checked_and_cleared(up):
+    fmt = await up.db.create_format(OWNER, "One Piece")
+    i = fmt["id"]
+    cb = await click(up, OWNER, f"fm:in:{i}:channel", msg_id=710)
+    assert cb.edits[-1][0] == panels.FORMAT_PROMPTS["channel"] and up.app.pending[OWNER].kind == "fmt:channel"
+    assert data_of(cb.edits[-1][1]) == [f"fm:cx:{i}"]
+    ev = await say(up, OWNER, "@not_an_id")
+    assert "❌" in ev.texts[0] and "Try again" in ev.texts[0] and OWNER in up.app.pending
+    ev = await say(up, OWNER, "1234567890")  # a bare channel id
+    assert ev.deleted and OWNER not in up.app.pending
+    assert (await up.db.get_format(OWNER, i))["channel"] == -1001234567890
+    chat, text, kw = up.client.edited[-1]
+    assert kw["message"] == 710 and "Channel saved" in text and "<code>-1001234567890</code>" in text
+    text, buttons = (await click(up, OWNER, f"fm:v:{i}")).edits[-1]
+    assert "<b>Channel</b> ▸ <code>-1001234567890</code>" in text and f"fm:chc:{i}" in data_of(buttons)
+    cb = await click(up, OWNER, f"fm:chc:{i}")  # the button clears it
+    assert (await up.db.get_format(OWNER, i))["channel"] is None and "— (optional)" in cb.edits[-1][0]
+    await up.db.update_format(OWNER, i, channel=SHOW)
+    await click(up, OWNER, f"fm:in:{i}:channel", msg_id=711)
+    await say(up, OWNER, "off")  # and so does typing off
+    assert (await up.db.get_format(OWNER, i))["channel"] is None
+
+
+async def test_the_channel_of_a_deleted_format_is_handled(up):
+    fmt = await up.db.create_format(OWNER, "One Piece")
+    await click(up, OWNER, f"fm:in:{fmt['id']}:channel")
+    await up.db.delete_format(OWNER, fmt["id"])
+    ev = await say(up, OWNER, "-1001234567890")
+    assert "no longer exists" in ev.texts[0] and OWNER not in up.app.pending
+
+
+async def test_the_format_screen_with_a_channel_still_fits_a_caption(up):
+    fmt = await up.db.create_format(OWNER, "N" * 30, template="T" * 200)
+    await up.db.update_format(OWNER, fmt["id"], channel=-1009999999999, caption="C" * 700)
+    text, _rows = panels.format_view(await up.db.get_format(OWNER, fmt["id"]), upload_on=True)
+    assert len(re.sub(r"<[^>]+>", "", text)) < 1024
+
+
+def test_a_shared_format_card_does_not_show_the_channel():
+    fmt = {**default_auto_format(), "kind": "manual", "name": "One Piece", "id": 5, "channel": SHOW}
+    text, _rows = panels.shared_format(fmt, "token")
+    assert str(SHOW) not in text and "Channel" not in text
+
+
+# ---------------------------------------------------------------------------
+# the messages
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "name, episode",
+    [
+        (NAME, "05"), ("One Piece - E1147.mkv", "1147"), ("Black Clover - S2 E05.mkv", "05"),
+        ("Tokyo Revengers Movie.mkv", ""), ("", ""),
+    ],
+)
+def test_the_episode_number_is_read_from_the_renamed_file(name, episode):
+    assert episode_of(name) == episode
+
+
+def test_the_notice_has_what_the_post_bot_needs():
+    text = notice_text(episode="05", channel=SHOW, file=NAME, post_id=4001, ref="ab12cd")
+    assert text.splitlines() == [
+        "File uploaded to the Database Channel", "Episode: 05", f"Channel: {SHOW}", f"File: {NAME}", "Post: 4001", "Ref: ab12cd"
+    ]
+    bare = notice_text(episode="", channel=SHOW, file=NAME, post_id=None, ref="ab12cd")
+    assert "Episode:" not in bare and "Post:" not in bare and f"Channel: {SHOW}" in bare
+
+
+def test_the_label_lines_of_a_message_are_found():
+    text = "✅ Post published successfully\nChannel: Black Clover English\nPost: https://t.me/c/1/2\nRef: ab12cd"
+    assert line_of("channel", text) == "Black Clover English" and line_of("post", text) == "https://t.me/c/1/2"
+    assert line_of("ref", text) == "ab12cd" and line_of("episode", text) == "" and line_of("ref", None) == ""
+
+
+@pytest.mark.parametrize(
+    "text, expected",
+    [
+        ("📥 Message received - checking the Database Channel…\nRef: ab12", ("ack", "")),
+        ("message received", ("ack", "")),
+        ("✅ Post published successfully\nChannel: X", ("ok", "")),
+        ("POST PUBLISHED SUCCESSFULLY", ("ok", "")),
+        ("❌ Publication failed: no button found\nFile: a.mkv\nRef: ab12", ("fail", "no button found")),
+        ("publication failed:   many   spaces  here", ("fail", "many spaces here")),
+        ("Publication failed", ("fail", "")),
+        ("❌ Publication failed: no post published successfully", ("fail", "no post published successfully")),  # a failure wins
+        ("hello", ("other", "")), ("", ("other", "")), (None, ("other", "")),
+    ],
+)
+def test_what_the_post_bot_says_is_told_apart(text, expected):
+    assert classify(text) == expected
+
+
+def test_the_reply_target_is_read_from_the_message():
+    assert reply_target(FakeEvent(POST_BOT, "x", reply_to_id=7)) == 7
+    assert reply_target(FakeEvent(POST_BOT, "x")) is None
+
+
+# ---------------------------------------------------------------------------
+# the offer after /rename
+# ---------------------------------------------------------------------------
+
+
+async def test_the_offer_asks_yes_or_no_after_the_episode(up):
+    await up.app.dbflow.offer(manual_job(), OWNER, SimpleNamespace(id=555))
+    (entity, text, kw), = up.client.sent
+    assert entity == OWNER and kw["reply_to"] == 555 and kw["parse_mode"] == "html"
+    assert "Upload this episode to the Database Channel?" in text and NAME in text
+    assert "Episode: <b>05</b>" in text and f"<code>{SHOW}</code>" in text
+    assert flat_buttons(kw["buttons"]) == [("✅ Yes", f"dbu:y:{OWNER}:555:{SHOW}"), ("❌ No", "dbu:n")]
+
+
+def test_the_buttons_fit_telegrams_limit_even_for_the_biggest_ids():
+    data = f"dbu:y:{-1009999999999}:{2147483647}:{-1009999999999}"
+    assert len(data.encode()) <= 64
+
+
+async def test_the_offer_for_a_format_without_a_channel_says_it_is_only_uploaded(up):
+    await up.app.dbflow.offer(manual_job(channel=None), OWNER, SimpleNamespace(id=555))
+    (_entity, text, kw), = up.client.sent
+    assert "has no Channel" in text and "no post is made" in text and "Episode:" not in text
+    assert data_of(kw["buttons"])[0] == f"dbu:y:{OWNER}:555:0"
+
+
+async def test_the_offer_warns_when_the_file_name_has_no_episode_number(up):
+    await up.app.dbflow.offer(manual_job(name="Tokyo Revengers Movie.mkv"), OWNER, SimpleNamespace(id=555))
+    assert "No episode number found" in up.client.sent[0][1]
+
+
+async def test_a_file_sent_to_a_channel_is_still_offered_in_the_chat_of_the_user(up):
+    await up.app.dbflow.offer(manual_job(), SHOW, SimpleNamespace(id=555))
+    (entity, _text, kw), = up.client.sent
+    assert entity == OWNER and kw["reply_to"] == 40  # the file the user sent, not the one in the channel
+    assert f"dbu:y:{SHOW}:555:{SHOW}" in data_of(kw["buttons"])
+
+
+async def test_nothing_is_offered_without_a_message_to_point_at(up):
+    await up.app.dbflow.offer(manual_job(), OWNER, None)
+    assert not up.client.sent
+
+
+class Reporter:
+    def __init__(self, log):
+        self.log = log
+
+    async def finish(self, text):
+        self.log.append("finished")
+
+
+async def deliver(up, job, hook, target_channel=None, log=None):
+    """Run the delivery step of a rename; ``log`` says in what order the file, the Done message and the hook came."""
+    log = [] if log is None else log
+    renamer = Renamer(up.client, up.cfg, up.db)
+    renamer.on_delivered = hook
+    if target_channel:
+        await up.db.update_settings(OWNER, target_channel=target_channel)
+    original_send = up.client.send_file
+
+    async def send_file(*args, **kw):
+        log.append("sent")
+        return await original_send(*args, **kw)
+
+    up.client.send_file = send_file
+    await renamer._deliver(
+        job, await up.db.get_settings(OWNER), document_message(NAME, 10), "input", caption=None, attrs=[], as_document=True,
+        thumb=None, final_size=10, warnings=[], rates="", started=time.monotonic(), reporter=Reporter(log),
+    )
+    return log
+
+
+async def test_a_manual_rename_is_offered_after_the_file_and_the_done_message(up):
+    log, calls = [], []
+
+    async def hook(job, target, sent):
+        log.append("offer")
+        calls.append((job.id, target, sent.id))
+
+    job = manual_job()
+    await deliver(up, job, hook, log=log)
+    assert job.state == "done" and calls == [(1, OWNER, 3001)] and log == ["sent", "finished", "offer"]
+
+
+@pytest.mark.parametrize("kind", ["auto", None])
+async def test_auto_renames_are_not_offered(up, kind):
+    called = []
+
+    async def hook(*args):
+        called.append(args)
+
+    job = manual_job(kind=kind or "auto")
+    if kind is None:
+        job.fmt = None  # an auto rename of someone who has no auto format
+    await deliver(up, job, hook)
+    assert called == [] and job.state == "done"
+
+
+async def test_a_failing_offer_does_not_spoil_the_rename(up):
+    async def hook(*args):
+        raise RuntimeError("telegram said no")
+
+    job = manual_job()
+    log = await deliver(up, job, hook)
+    assert job.state == "done" and log == ["sent", "finished"]
+
+
+async def test_without_the_hook_nothing_happens(up):
+    job = manual_job()
+    assert await deliver(up, job, None) == ["sent", "finished"] and job.state == "done"
+
+
+async def test_a_file_sent_to_the_target_channel_is_offered_with_that_channel_as_its_chat(up):
+    seen = []
+
+    async def hook(job, target, sent):
+        seen.append(target)
+
+    await deliver(up, manual_job(), hook, target_channel=SHOW)
+    assert seen == [SHOW]
+
+
+# ---------------------------------------------------------------------------
+# Yes: upload, tell the communication chat, read the answers
+# ---------------------------------------------------------------------------
+
+
+async def test_yes_copies_the_episode_and_tells_the_communication_chat(up):
+    cb = await start_upload(up, caption="<b>x</b>")
+    assert cb.edits[-1][0].startswith("⏳ <b>Processing…</b>") and cb.edits[-1][1] is None and cb.answers
+    nid, text, kw = await the_notice(up)
+    # copied inside Telegram: nothing was downloaded, nothing was uploaded again
+    assert up.client.forwarded == [(DB, 555, OWNER, {"drop_author": True})]
+    assert up.client.downloads == [] and up.client.files == []
+    ref = ref_of(text)
+    assert text.splitlines() == [
+        "File uploaded to the Database Channel", "Episode: 05", f"Channel: {SHOW}", f"File: {NAME}", "Post: 4001", f"Ref: {ref}"
+    ]
+    assert kw["parse_mode"] is None  # plain text: nothing in a file name can be taken for formatting
+
+
+async def test_the_whole_story_from_yes_to_published(up):
+    await start_upload(up)
+    nid, text, _ = await the_notice(up)
+    ref = ref_of(text)
+    await post_bot_says(up, f"📥 Message received - checking the Database Channel…\nRef: {ref}", reply_to=nid)
+    await wait_for(lambda: len(up.client.edited) >= 2)
+    assert "channel post bot got it" in last_edit(up) and "Waiting" not in last_edit(up)
+    done = (
+        "✅ Post published successfully\nChannel: Black Clover English\nPost: https://t.me/c/1/2\n"
+        f"Button: Download Episode 05 → https://t.me/x\nEpisode: 05\nFile: {NAME}\nRef: {ref}"
+    )
+    await post_bot_says(up, done, reply_to=nid)
+    await up.flow.idle()
+    texts = [t for _e, t, _k in up.client.edited]
+    assert len(texts) == 3
+    assert "Waiting for the channel post bot" in texts[0] and "Uploaded to the Database Channel" in texts[0]
+    assert "Post published successfully" in texts[2] and "Black Clover English" in texts[2] and "https://t.me/c/1/2" in texts[2]
+    assert NAME in texts[2]
+    assert {kw["message"] for _e, _t, kw in up.client.edited} == {77} and {e for e, _t, _k in up.client.edited} == {OWNER}
+    assert all(kw["buttons"] is None for _e, _t, kw in up.client.edited)
+    assert up.flow.flows == {} and up.flow._busy == set()
+
+
+async def test_a_failure_of_the_post_bot_is_shown_with_its_reason(up):
+    await start_upload(up)
+    nid, text, _ = await the_notice(up)
+    await post_bot_says(up, f"📥 Message received - checking the Database Channel…\nRef: {ref_of(text)}", reply_to=nid)
+    reason = "The latest post of the Database Channel is the file “Other.mkv”, not the one that was reported - nothing was published."
+    await post_bot_says(up, f"❌ Publication failed: {reason}\nFile: {NAME}\nRef: {ref_of(text)}", reply_to=nid)
+    await up.flow.idle()
+    final = last_edit(up)
+    assert final.startswith("❌ <b>Publication failed</b>") and reason in final and NAME in final
+
+
+async def test_names_and_reasons_are_escaped_in_the_edits_but_not_in_the_notice(up):
+    await start_upload(up, name="[S01E05] Tom & Jerry <HD>.mkv")
+    nid, text, _ = await the_notice(up)
+    assert "File: [S01E05] Tom & Jerry <HD>.mkv" in text
+    await post_bot_says(up, "❌ Publication failed: bad <b>thing</b> & more\nRef: " + ref_of(text), reply_to=nid)
+    await up.flow.idle()
+    final = last_edit(up)
+    assert "Tom &amp; Jerry &lt;HD&gt;" in final and "bad &lt;b&gt;thing&lt;/b&gt; &amp; more" in final
+
+
+async def test_no_answer_at_all_means_the_channel_post_bot_is_down(up):
+    up.flow.ack_wait = 0.1
+    await start_upload(up)
+    await up.flow.idle()
+    final = last_edit(up)
+    assert "down or not responding" in final and "The episode is in the Database Channel" in final and "nothing was published" in final
+    assert up.flow.flows == {} and up.flow._busy == set()
+
+
+async def test_an_acknowledged_notice_that_is_never_answered_is_not_called_down(up):
+    up.flow.ack_wait, up.flow.result_wait = 2.0, 0.15
+    await start_upload(up)
+    nid, text, _ = await the_notice(up)
+    await post_bot_says(up, f"📥 Message received - checking the Database Channel…\nRef: {ref_of(text)}", reply_to=nid)
+    await up.flow.idle()
+    final = last_edit(up)
+    assert "No word from the channel post bot" in final and "down" not in final
+
+
+async def test_an_answer_without_the_acknowledgement_still_counts(up):
+    await start_upload(up)
+    nid, text, _ = await the_notice(up)
+    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(text)}", reply_to=nid)
+    await up.flow.idle()
+    assert "Post published successfully" in last_edit(up)
+
+
+async def test_answers_are_matched_by_the_ref_alone_or_by_the_reply_alone(up):
+    await start_upload(up, offer_id=71, mid=601)
+    nid, text, _ = await the_notice(up)
+    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(text)}")  # not a reply: the Ref says whose it is
+    await up.flow.idle()
+    assert "Post published successfully" in last_edit(up)
+
+    await start_upload(up, offer_id=72, mid=602)
+    nid, text, _ = await the_notice(up, 2)
+    await post_bot_says(up, "❌ Publication failed: nope", reply_to=nid)  # no Ref: the reply says whose it is
+    await up.flow.idle()
+    assert "Publication failed" in last_edit(up) and "nope" in last_edit(up)
+
+
+async def test_with_two_uploads_waiting_the_ref_tells_whose_answer_it_is(up):
+    up.client.messages[601] = document_message("[S01E05] A [1080p].mkv", 10, mid=601)
+    up.client.messages[602] = document_message("[S01E06] A [1080p].mkv", 10, mid=602)
+    await click(up, OWNER, f"dbu:y:{OWNER}:601:{SHOW}", msg_id=71)
+    await click(up, OWNER, f"dbu:y:{OWNER}:602:{SHOW}", msg_id=72)
+    first, second = await the_notice(up, 1), await the_notice(up, 2)
+    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(second[1])}")  # not a reply, and two are waiting
+    await wait_for(lambda: any(kw["message"] == 72 and "published successfully" in t for _e, t, kw in up.client.edited))
+    assert not any(kw["message"] == 71 and "published successfully" in t for _e, t, kw in up.client.edited)
+    assert len(up.flow.flows) == 1 and ref_of(first[1]) in up.flow.flows  # the first is still waiting
+    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(first[1])}")
+    await up.flow.idle()
+
+
+async def test_a_plain_message_counts_when_only_one_upload_is_waiting(up):
+    await start_upload(up)
+    await the_notice(up)
+    await post_bot_says(up, "✅ Post published successfully")
+    await up.flow.idle()
+    assert "Post published successfully" in last_edit(up)
+
+
+async def test_other_senders_other_chats_other_replies_and_other_words_are_ignored(up):
+    await start_upload(up)
+    nid, text, _ = await the_notice(up)
+    ref = ref_of(text)
+    ok = f"✅ Post published successfully\nRef: {ref}"
+    await post_bot_says(up, ok, reply_to=nid, sender=STRANGER)  # somebody else
+    await post_bot_says(up, ok, reply_to=nid, chat=-1005000000005)  # another chat
+    await post_bot_says(up, f"hello there\nRef: {ref}", reply_to=nid)  # not an answer
+    await post_bot_says(up, "✅ Post published successfully", reply_to=1)  # a reply to something else
+    await asyncio.sleep(0.05)
+    assert up.flow.flows[ref].inbox.empty() and len(up.client.edited) == 1
+    await post_bot_says(up, ok, reply_to=nid)
+    await up.flow.idle()
+    assert "Post published successfully" in last_edit(up)
+
+
+async def test_two_uploads_at_once_each_get_their_own_answer(up):
+    up.client.messages[601] = document_message("[S01E05] A [1080p].mkv", 10, mid=601)
+    up.client.messages[602] = document_message("[S01E06] A [1080p].mkv", 10, mid=602)
+    await click(up, OWNER, f"dbu:y:{OWNER}:601:{SHOW}", msg_id=71)
+    await click(up, OWNER, f"dbu:y:{OWNER}:602:{SHOW}", msg_id=72)
+    first, second = await the_notice(up, 1), await the_notice(up, 2)
+    assert "Episode: 05" in first[1] and "Episode: 06" in second[1] and ref_of(first[1]) != ref_of(second[1])
+    await post_bot_says(up, f"❌ Publication failed: nope\nRef: {ref_of(second[1])}", reply_to=second[0])  # out of order
+    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(first[1])}", reply_to=first[0])
+    await up.flow.idle()
+    last = {kw["message"]: text for _e, text, kw in up.client.edited}
+    assert "Post published successfully" in last[71] and "Publication failed" in last[72]
+
+
+async def test_a_format_without_a_channel_is_only_uploaded(up):
+    await start_upload(up, channel=None)
+    await up.flow.idle()
+    assert len(up.client.forwarded) == 1 and not notices(up)
+    final = last_edit(up)
+    assert final.startswith("✅ <b>Uploaded to the Database Channel</b>") and "has no Channel" in final and NAME in final
+    assert up.flow.flows == {}
+
+
+async def test_when_forwarding_is_refused_the_file_is_sent_again_by_its_id(up):
+    up.client.forward_errors.append(errors.ChatForwardsRestrictedError(None))
+    await start_upload(up, caption="Some caption")
+    nid, text, _ = await the_notice(up)
+    assert up.client.forwarded == []
+    (entity, file, kw), = up.client.files
+    assert entity == DB and file is up.client.messages[555].media  # the file itself is not sent again, only its id
+    assert kw["caption"] == "Some caption" and kw["formatting_entities"] is None and kw["parse_mode"] is None
+    assert "Post: 3001" in text and up.client.downloads == []
+
+
+async def test_a_failed_upload_can_be_tried_again(up):
+    up.client.forward_errors.append(RuntimeError("no rights"))
+    up.client.fail_send_file = True
+    await start_upload(up)
+    await up.flow.idle()
+    chat, text, kw = up.client.edited[-1]
+    assert "The upload to the Database Channel failed" in text and "RuntimeError: cannot send file" in text and "admin" in text
+    assert flat_buttons(kw["buttons"]) == [("🔁 Try again", f"dbu:y:{OWNER}:555:{SHOW}"), ("✖ Close", "x:close")]
+    assert not notices(up) and up.flow._busy == set()  # nothing was announced
+    up.client.fail_send_file = False
+    await click(up, OWNER, f"dbu:y:{OWNER}:555:{SHOW}", msg_id=77)
+    await the_notice(up)
+    assert len(up.client.forwarded) == 1
+
+
+async def test_an_episode_that_was_deleted_is_reported(up):
+    await click(up, OWNER, f"dbu:y:{OWNER}:999:{SHOW}", msg_id=77)
+    await up.flow.idle()
+    assert "not there anymore" in last_edit(up) and not up.client.forwarded and not notices(up)
+
+
+async def test_a_second_tap_while_it_is_working_does_nothing(up):
+    await start_upload(up)
+    again = await click(up, OWNER, f"dbu:y:{OWNER}:555:{SHOW}", msg_id=77)
+    assert again.toasts == ["I'm already on it…"] and again.edits == []
+    await the_notice(up)
+    await asyncio.sleep(0.05)
+    assert len(up.client.forwarded) == 1 and len(notices(up)) == 1
+
+
+async def test_no_does_not_upload(up):
+    cb = await click(up, OWNER, "dbu:n", msg_id=77)
+    assert "not uploaded" in cb.edits[-1][0] and cb.edits[-1][1] is None and cb.answers
+    assert not up.client.forwarded and not up.client.sent
+
+
+async def test_a_notice_that_cannot_be_sent_is_reported(up):
+    up.client.message_errors[COMM] = RuntimeError("chat not found")
+    await start_upload(up)
+    await up.flow.idle()
+    final = last_edit(up)
+    assert "could not write in the communication chat" in final and "chat not found" in final and "Uploaded" in final
+    assert len(up.client.forwarded) == 1 and up.flow.flows == {} and up.flow._busy == set()
+
+
+async def test_an_unexpected_error_is_shown_instead_of_leaving_processing_on_the_screen(up):
+    async def broken(*args, **kw):
+        raise RuntimeError("boom")
+
+    up.client.get_messages = broken
+    await start_upload(up)
+    await up.flow.idle()
+    assert "Something went wrong" in last_edit(up) and "RuntimeError: boom" in last_edit(up) and up.flow._busy == set()
+
+
+async def test_a_message_that_was_deleted_meanwhile_does_not_break_the_flow(up):
+    async def gone(*args, **kw):
+        raise errors.MessageIdInvalidError(None)
+
+    up.client.edit_message = gone
+    await start_upload(up)
+    nid, text, _ = await the_notice(up)
+    await post_bot_says(up, f"✅ Post published successfully\nRef: {ref_of(text)}", reply_to=nid)
+    await up.flow.idle()
+    assert up.flow.flows == {} and up.flow._busy == set()
+
+
+async def test_shutdown_stops_the_waiting(up):
+    await start_upload(up)
+    await the_notice(up)
+    await up.app.shutdown()
+    assert up.flow._tasks == set() and up.flow.flows == {} and up.flow._busy == set()
+
+
+async def test_strangers_cannot_press_the_buttons(up):
+    up.client.messages[555] = document_message(NAME, 10, mid=555)
+    await click(up, STRANGER, f"dbu:y:{STRANGER}:555:{SHOW}")
+    assert not up.client.forwarded and not up.flow._tasks
+
+
+@pytest.mark.parametrize("data", ["dbu", "dbu:y", "dbu:y:1:2", "dbu:y:a:b:c", "dbu:z", "dbu:y:1:2:3:4"])
+async def test_a_broken_button_does_nothing(up, data):
+    cb = await click(up, OWNER, data)
+    assert cb.answers and not up.client.forwarded and up.flow._busy == set() and not up.flow._tasks

tests/tgfakes.py

@@ -81,8 +81,9 @@ class FakeMessage:
     def __init__(self, text: str = "", *, mid: int = 1, sender: int = 1, file: Any = None, video: bool = False,
                  audio: bool = False, photo: bool = False, sticker: bool = False, gif: bool = False,
                  voice: bool = False, video_note: bool = False, fwd: bool = False, out: bool = False,
-                 entities: Any = None, thumbs: Any = None):
+                 entities: Any = None, thumbs: Any = None, reply_to_id: Optional[int] = None):
         self.id = mid
+        self.reply_to_msg_id = reply_to_id
         self.message = text
         self.raw_text = text
         self.entities = entities
@@ -111,8 +112,10 @@ class FakeEvent:
 
     def __init__(self, sender: int, text: str = "", *, message: Optional[FakeMessage] = None, reply: Any = None,
                  mid: int = 1, chat: Optional[int] = None, private: bool = True, photo_bytes: Optional[bytes] = None,
-                 entities: Any = None):
-        self.message = message if message is not None else FakeMessage(text, mid=mid, sender=sender, entities=entities)
+                 entities: Any = None, reply_to_id: Optional[int] = None):
+        self.message = message if message is not None else FakeMessage(
+            text, mid=mid, sender=sender, entities=entities, reply_to_id=reply_to_id
+        )
         self.message.sender_id = sender
         self.sender_id = sender
         self.chat_id = sender if chat is None else chat
@@ -229,6 +232,9 @@ class FakeClient:
         self.inline_placeholder: Optional[str] = "Pick a format"  # what the bot's profile says; None = inline mode is off
         self.server_inline = True  # False: Telegram refuses messages that carry a share button
         self.me_calls = 0
+        self.forwarded: List[tuple] = []  # forward_messages(entity, messages, from_peer, **kw)
+        self.forward_errors: List[Exception] = []  # raised by forward_messages(), one per call
+        self.message_errors: dict = {}  # entity -> exception that send_message() raises for it
 
     async def get_me(self) -> SimpleNamespace:
         self.me_calls += 1
@@ -246,8 +252,16 @@ class FakeClient:
             raise ValueError(f"Could not find the input entity for {entity!r}")
         return SimpleNamespace(id=entity)
 
+    async def forward_messages(self, entity: Any, messages: Any, from_peer: Any = None, **kw: Any) -> SimpleNamespace:
+        if self.forward_errors:
+            raise self.forward_errors.pop(0)
+        self.forwarded.append((entity, messages, from_peer, kw))
+        return SimpleNamespace(id=4000 + len(self.forwarded))
+
     async def send_message(self, entity: Any, text: str = "", **kw: Any) -> SimpleNamespace:
         refuse_share_button(self, kw.get("buttons"))
+        if entity in self.message_errors:
+            raise self.message_errors[entity]
         self.sent.append((entity, text, kw))
         return SimpleNamespace(id=2000 + len(self.sent))