Blame for app/syncer.py

"""The background side of the sync.

* a timer: every SYNC_INTERVAL_MINUTES each connected channel is compared with My posts (app/sync_engine.py);
* a queue for what Telegram tells the bot while it runs: a post deleted / edited / added in a channel is looked at a few
  seconds later (the wait lets a burst of changes settle, and lets the bot finish saving its own posts first);
* service messages ("pinned a message", "channel photo changed" ...) are deleted the moment they appear.

Nothing here holds the bot's one-long-job lock: a running /repost, /shift, /swap ... always wins, and what happens
while it runs is picked up by the next timed check.
"""
from __future__ import annotations

import asyncio
import logging
import time
from typing import Optional

from .servicemsg import RIGHTS_ERRORS, action_label, remove
from .sync_engine import SyncOptions, SyncReport, sync_channel, sync_ids
from .tgutil import esc, peer_of

log = logging.getLogger(__name__)


class Syncer:
    def __init__(self, ctx):
        self.ctx = ctx
        # timings (seconds) - the tests make them small
        self.start_delay = 90.0  # after the bot starts, before the first timed check
        self.debounce = 5.0  # quiet time needed before the queue is worked off
        self.max_debounce = 30.0  # ... but never wait longer than this
        self.lock_wait = 30.0  # a long job runs: look again after this
        self.rt_grace = 10.0  # live changes: posts the bot touched / messages younger than this are judged next round
        self.pause = 0.25  # between two requests
        self.cache_ttl = 60.0  # how long "is this channel connected?" is remembered
        self.tell_every = 86400.0  # the same warning goes to the owners at most this often
        self.tell_restricted_every = 7 * 86400.0  # ... a restriction by Telegram can last for weeks: once a week
        self.max_ids = 300  # more changed ids than this in one channel: look at the whole channel instead
        self.max_tries = 8  # a message that stays "too fresh" is given up after this many rounds
        self.history_seconds = 150.0  # a timed check reads a channel's older posts for this long, then goes on next time
        self._busy = asyncio.Lock()  # one look at a time
        self._edits: dict = {}  # channel id -> message ids edited
        self._news: dict = {}  # channel id -> message ids that are new posts
        self._tries: dict = {}
        self._stamp = 0
        self._wake = asyncio.Event()
        self._tasks: list = []
        self._cache: dict = {}
        self._told: dict = {}
        self.last_reports: dict = {}  # channel id -> its latest SyncReport

    # ----------------------------------------------------------------------------------------- running
    @property
    def enabled(self) -> bool:
        return self.ctx.cfg.sync_interval_minutes > 0

    def start(self) -> None:
        if self._tasks or not self.enabled:
            return
        self._tasks = [
            asyncio.create_task(self._periodic(), name="sync-timer"),
            asyncio.create_task(self._events(), name="sync-live"),
        ]

    async def stop(self) -> None:
        tasks, self._tasks = self._tasks, []
        for t in tasks:
            t.cancel()
        for t in tasks:
            try:
                await t
            except (asyncio.CancelledError, Exception):
                pass

    async def _periodic(self) -> None:
        await asyncio.sleep(self.start_delay)
        while True:
            try:
                await self.run_all()
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception("the timed check of the channels failed")
            await asyncio.sleep(max(5, self.ctx.cfg.sync_interval_minutes) * 60)

    async def _events(self) -> None:
        while True:
            await self._wake.wait()
            self._wake.clear()
            try:
                await self.flush_events()
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception("working off the channel changes failed")

    # ----------------------------------------------------------------------------------------- helpers
    async def channel(self, cid: int):
        """The connected channel with this id, or None (remembered for a minute: busy channels cost no database reads)."""
        now = time.monotonic()
        hit = self._cache.get(cid)
        if hit is not None and now - hit[0] < self.cache_ttl:
            return hit[1]
        ch = await self.ctx.db.get_channel(cid)
        if ch is not None and not ch.active:
            ch = None
        self._cache[cid] = (now, ch)
        return ch

    def forget_cache(self) -> None:
        self._cache.clear()

    def exclusive(self):
        """Held (`async with`) while no look at any channel runs - for a change that must not race with one."""
        return self._busy

    def options(self, **kw) -> SyncOptions:
        ctx = self.ctx
        base = dict(
            delete_services=ctx.cfg.delete_service_messages,
            history=ctx.cfg.import_old_posts,
            ignore=ctx.sync_ignore,
            pause=self.pause,
        )
        base.update(kw)
        return SyncOptions(**base)

    async def tell_owners(self, ch, key: str, text: str, every: Optional[float] = None) -> None:
        """A warning for the owners - the same one (per channel and kind) at most once a day (or every `every` seconds)."""
        now = time.monotonic()
        k = (ch.id, key)
        if now - self._told.get(k, -1e12) < (self.tell_every if every is None else every):
            return
        self._told[k] = now
        for uid in sorted(self.ctx.cfg.owners):
            try:
                await self.ctx.client.send_message(uid, text, link_preview=False)
            except Exception:
                log.warning("could not warn user %s about %s", uid, ch.id, exc_info=True)

    async def _after(self, ch, rep: SyncReport) -> None:
        self.last_reports[ch.id] = rep
        if rep.changed or rep.error or rep.anomaly:
            log.info(
                "channel %s: %s removed, %s updated, %s new, %s service messages deleted%s",
                ch.id, len(rep.deleted), len(rep.edited), len(rep.adopted), rep.services_deleted,
                f" - {rep.error or rep.anomaly}" if (rep.error or rep.anomaly) else "",
            )
        if rep.services_failed and rep.services_error in RIGHTS_ERRORS:
            await self.tell_owners(ch, "rights", self._rights_text(ch, rep.services_error))
        if rep.restricted_channel:
            log.info("channel %s is restricted by Telegram (%s): My posts left untouched", ch.id, rep.restricted_channel)
            text = self._restricted_text(ch, rep.restricted_channel)
            await self.tell_owners(ch, "restricted", text, every=self.tell_restricted_every)
        elif rep.restricted:
            log.info("channel %s: %s post(s) held back by Telegram (%s): left untouched", ch.id, len(rep.restricted), rep.restricted_why)
            text = self._held_back_text(ch, len(rep.restricted), rep.restricted_why)
            await self.tell_owners(ch, "held-back", text, every=self.tell_restricted_every)

    @staticmethod
    def _restricted_text(ch, why: str) -> str:
        return (
            f"🚫 Telegram has restricted <b>{esc(ch.title)}</b> ({esc(why)}). While that lasts I don't change anything in "
            "My posts for this channel - the saved posts stay exactly as they are, whatever the channel shows now."
        )

    @staticmethod
    def _held_back_text(ch, n: int, why: Optional[str]) -> str:
        return (
            f"🚫 Telegram holds back {n} post(s) of <b>{esc(ch.title)}</b> ({esc(why or 'restricted')}). "
            "Their saved copies in My posts stay exactly as they are."
        )

    @staticmethod
    def _rights_text(ch, why: Optional[str]) -> str:
        return (
            f"⚠️ I can't delete the service messages (name changes, “pinned a message” ...) in <b>{esc(ch.title)}</b> "
            f"({esc(why or 'refused')}). Give the bot the <i>Delete messages</i> right there, or set "
            "DELETE_SERVICE_MESSAGES=false to stop trying."
        )

    # ------------------------------------------------------------------------------------ full looks
    async def run_all(
        self, chans: Optional[list] = None, *, explicit: bool = False, history: Optional[bool] = None, progress=None
    ) -> list:
        """Look at the connected channels one after the other. `explicit` (the /sync command): the caller holds the
        one-long-job lock and everything is read; the timed check gives way to any long job instead. `history`: True
        reads the channels' older posts even if IMPORT_OLD_POSTS is off (/sync --history)."""
        ctx = self.ctx
        if chans is None:
            chans = await ctx.db.list_channels()
        busy = (lambda: False) if explicit else ctx.lock.locked
        reports: list = []
        for ch in chans:
            if busy():
                break

            async def prog(rep, ch=ch):
                if progress is not None:
                    await progress(ch, rep)

            try:
                async with self._busy:
                    extra = {} if history is None else {"history": history}
                    opts = self.options(deep=explicit, history_seconds=None if explicit else self.history_seconds, **extra)
                    rep = await sync_channel(ctx.client, ctx.db, ch, opts, busy=busy, progress=prog)
            except asyncio.CancelledError:
                raise
            except Exception as e:
                log.exception("looking at channel %s failed", ch.id)
                rep = SyncReport(channel=ch, error=f"{type(e).__name__}: {str(e)[:120]}")
            await self._after(ch, rep)
            reports.append(rep)
        return reports

    # ----------------------------------------------------------------------------- live: what updates say
    def _touch(self) -> None:
        self._stamp += 1
        self._wake.set()

    def note_new(self, cid: int, msg) -> None:
        """A post appeared in a channel. The bot's own posts are skipped (it saves them itself)."""
        if not self.enabled or self.ctx.lock.locked() or getattr(msg, "out", False):
            return
        self._news.setdefault(cid, set()).add(msg.id)
        self._touch()

    def note_edit(self, cid: int, msg) -> None:
        if not self.enabled or self.ctx.lock.locked():
            return
        self._edits.setdefault(cid, set()).add(msg.id)
        self._touch()

    async def note_deleted(self, cid: int, ids) -> None:
        """Posts were deleted in a channel: they leave My posts at once (nothing to read from Telegram for that)."""
        if not self.enabled:
            return
        ch = await self.channel(cid)
        if ch is None:
            return
        for pending in (self._edits, self._news):
            if cid in pending:
                pending[cid].difference_update(ids)
        n = await self.ctx.db.forget_posts(cid, list(ids))
        if n:
            log.info("%s post(s) were deleted in channel %s: removed from My posts", n, cid)

    async def on_service(self, cid: int, msg) -> None:
        """A service message appeared: delete it."""
        if not self.ctx.cfg.delete_service_messages:
            return
        ch = await self.channel(cid)
        if ch is None:
            return
        out = await remove(self.ctx.client, peer_of(ch), [msg.id])
        if out.error is None:
            log.info("removed the “%s” notice %s from channel %s", action_label(msg), msg.id, cid)
            return
        log.warning("could not delete the “%s” notice %s in channel %s: %s", action_label(msg), msg.id, cid, out.error_name)
        if out.error_name in RIGHTS_ERRORS:
            await self.tell_owners(ch, "rights", self._rights_text(ch, out.error_name))

    def _pending(self) -> bool:
        return bool(self._edits or self._news)

    async def flush_events(self) -> None:
        """Work off the queue once it has been quiet for a moment (and no long job is running)."""
        first = time.monotonic()
        while self._pending():
            stamp = self._stamp
            await asyncio.sleep(self.debounce)
            if stamp != self._stamp and time.monotonic() - first < self.max_debounce:
                continue  # more is still coming in: let it settle
            if self.ctx.lock.locked():
                await asyncio.sleep(self.lock_wait)
                continue
            edits, news = self._edits, self._news
            self._edits, self._news = {}, {}
            await self._process(edits, news)
            first = time.monotonic()

    def _requeue(self, pending: dict, cid: int, mid: int) -> None:
        key = (cid, mid)
        self._tries[key] = self._tries.get(key, 0) + 1
        if self._tries[key] > self.max_tries:
            self._tries.pop(key, None)
            return
        pending.setdefault(cid, set()).add(mid)

    async def _process(self, edits: dict, news: dict) -> None:
        ctx = self.ctx
        for cid in sorted(set(edits) | set(news)):
            ch = await self.channel(cid)
            if ch is None:
                continue
            e_ids, n_ids = set(edits.get(cid, ())), set(news.get(cid, ()))
            opts = self.options(grace=self.rt_grace, history=False)  # live: the older posts wait for the timed check
            try:
                async with self._busy:
                    if len(e_ids | n_ids) > self.max_ids:  # a burst: one look at the whole channel is cheaper
                        rep = await sync_channel(ctx.client, ctx.db, ch, opts, busy=ctx.lock.locked)
                    else:
                        rep = await sync_ids(
                            ctx.client, ctx.db, ch, e_ids | n_ids, opts, adopt_ids=n_ids, busy=ctx.lock.locked
                        )
            except asyncio.CancelledError:
                raise
            except Exception as e:
                log.exception("looking at the changes in channel %s failed", cid)
                rep = SyncReport(channel=ch, error=f"{type(e).__name__}: {str(e)[:120]}")
            await self._after(ch, rep)
            if rep.busy:  # a long job started in the meantime: let the next timed check have it
                continue
            for mid in set(rep.young):
                self._requeue(self._edits, cid, mid)
            for mid in set(rep.young_new):
                self._requeue(self._news, cid, mid)
            for mid in (e_ids | n_ids) - set(rep.young) - set(rep.young_new):
                self._tries.pop((cid, mid), None)