From 4469938f4706431209293e5170ceee539eb17c4e Mon Sep 17 00:00:00 2001 From: h Date: Mon, 19 Oct 2026 02:17:57 +0000 Subject: [PATCH] feat(backend): watches fire on new, edited and deleted messages and on coming online, legacy keyword included --- backend/src/userbot/handlers/deletes.py | 2 + backend/src/userbot/handlers/edits.py | 2 + backend/src/userbot/handlers/messages.py | 3 +- backend/src/userbot/handlers/presence.py | 8 +- .../src/userbot/handlers/raw/transcribed.py | 2 +- backend/src/userbot/modules/watches/cache.py | 190 ++++++++++++++---- .../src/userbot/modules/watches/evaluator.py | 114 +++++++++-- .../src/userbot/modules/watches/repository.py | 48 +++++ backend/src/userbot/modules/watches/seen.py | 16 ++ backend/src/utils/watch_rules.py | 145 +++++++++++++ backend/src/watcher/registry.py | 15 +- backend/src/watcher/runner.py | 16 +- 12 files changed, 501 insertions(+), 60 deletions(-) create mode 100644 backend/src/userbot/modules/watches/repository.py create mode 100644 backend/src/userbot/modules/watches/seen.py create mode 100644 backend/src/utils/watch_rules.py diff --git a/backend/src/userbot/handlers/deletes.py b/backend/src/userbot/handlers/deletes.py index 42dfdc8..3676c33 100644 --- a/backend/src/userbot/handlers/deletes.py +++ b/backend/src/userbot/handlers/deletes.py @@ -25,9 +25,11 @@ async def on_deleted_messages(client: PyroClient, messages: list[Message]) -> No ctx.pending_deletes.remember(chat_id, ids) if box: await repository.mark_deleted_box(ctx.pool, ctx.account_id, box) + await ctx.watches.on_deleted(None, box) await notify_bg_event(ctx.pool, "delete", ctx.account_id, message_ids=box) for chat_id, ids in channels.items(): await repository.mark_deleted_channel(ctx.pool, ctx.account_id, chat_id, ids) + await ctx.watches.on_deleted(chat_id, ids) await notify_bg_event( ctx.pool, "delete", ctx.account_id, chat_id=chat_id, message_ids=ids ) diff --git a/backend/src/userbot/handlers/edits.py b/backend/src/userbot/handlers/edits.py index 833b63d..f200735 100644 --- a/backend/src/userbot/handlers/edits.py +++ b/backend/src/userbot/handlers/edits.py @@ -5,6 +5,7 @@ from userbot.modules.capture.chat_meta import meta_from_chat from userbot.modules.capture.message import message_row from userbot.modules.client import PyroClient from userbot.modules.media import capture_media +from userbot.modules.watches.seen import seen_message from utils.events import notify_bg_event @@ -22,6 +23,7 @@ async def on_edited_message(client: PyroClient, message: Message) -> None: chat = message.chat chat_id = chat.id or 0 meta = meta_from_chat(chat, ctx.contacts.ids) + await ctx.watches.on_edit(seen_message(chat_id, message)) toggles = ctx.resolve(meta) if not toggles.track_edits_deletes: return diff --git a/backend/src/userbot/handlers/messages.py b/backend/src/userbot/handlers/messages.py index 5e2bd53..854a574 100644 --- a/backend/src/userbot/handlers/messages.py +++ b/backend/src/userbot/handlers/messages.py @@ -5,6 +5,7 @@ from userbot.modules.capture.chat_meta import meta_from_chat from userbot.modules.client import PyroClient from userbot.modules.stt import is_transcribable from userbot.modules.stt.gate import safe_transcribe +from userbot.modules.watches.seen import seen_message from utils.events import notify_bg_event @@ -20,7 +21,7 @@ async def on_message(client: PyroClient, message: Message) -> None: ): return meta = meta_from_chat(message.chat, ctx.contacts.ids) - await ctx.watches.on_text(meta.chat_id, message.id, message.text or message.caption) + await ctx.watches.on_message(seen_message(meta.chat_id, message)) toggles = ctx.resolve(meta) if not toggles.messages: return diff --git a/backend/src/userbot/handlers/presence.py b/backend/src/userbot/handlers/presence.py index a41be89..7a20e9a 100644 --- a/backend/src/userbot/handlers/presence.py +++ b/backend/src/userbot/handlers/presence.py @@ -13,19 +13,21 @@ async def on_user_status(client: PyroClient, user: User) -> None: ctx = client.capture if ctx is None or user.status is None: return + at = datetime.now(UTC) + status = user.status.name.lower() if ctx.resolve_peer(user.id, is_bot=bool(user.is_bot)).presence: await repository.insert_status( ctx.pool, ctx.account_id, user.id, - datetime.now(UTC), - user.status.name.lower(), + at, + status, user.last_online_date, user.next_offline_date, raw_json(user.raw), ) await notify_bg_event(ctx.pool, "presence", ctx.account_id, chat_id=user.id) - await ctx.watches.on_status(user.id, is_online=user.status.name.lower() == "online") + await ctx.watches.on_status(user.id, status, at, user.last_online_date) handlers = on_user_status.handlers # ty: ignore[unresolved-attribute] diff --git a/backend/src/userbot/handlers/raw/transcribed.py b/backend/src/userbot/handlers/raw/transcribed.py index 335743b..f202621 100644 --- a/backend/src/userbot/handlers/raw/transcribed.py +++ b/backend/src/userbot/handlers/raw/transcribed.py @@ -20,4 +20,4 @@ async def handle( await repository.set_extracted_text( ctx.pool, ctx.account_id, meta.chat_id, update.msg_id, update.text ) - await ctx.watches.on_text(meta.chat_id, update.msg_id, update.text) + await ctx.watches.on_transcript(meta.chat_id, update.msg_id, update.text) diff --git a/backend/src/userbot/modules/watches/cache.py b/backend/src/userbot/modules/watches/cache.py index 145eaf6..965dc26 100644 --- a/backend/src/userbot/modules/watches/cache.py +++ b/backend/src/userbot/modules/watches/cache.py @@ -1,60 +1,182 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from datetime import datetime, timedelta +from typing import TYPE_CHECKING, Any +from userbot.modules.watches import repository from userbot.modules.watches.evaluator import ( - KIND_KEYWORD, - KIND_PEER_ONLINE, + Rules, + Seen, + build_rules, + match_chat_scope, match_keyword, - match_peer_online, + match_new_message, ) from utils.logging import logger from utils.read import watches as watches_read +from utils.read.accounts import self_user_id if TYPE_CHECKING: + from collections.abc import Coroutine + import asyncpg - from utils.read.models import WatchView +ONLINE_QUIET = timedelta(minutes=15) +PREVIEW_CHARS = 500 + + +def _message_payload(event: str, seen: Seen) -> dict[str, Any]: + payload: dict[str, Any] = { + "event": event, + "chat_id": seen.chat_id, + "message_id": seen.message_id, + "sender_id": seen.sender_id, + "text": (seen.text or "")[:PREVIEW_CHARS] or None, + } + if seen.date is not None: + payload["date"] = seen.date.isoformat() + return payload class WatchCache: def __init__(self, pool: asyncpg.Pool, account_id: int) -> None: self._pool = pool self._account_id = account_id - self.watches: list[WatchView] = [] - self._online: set[int] = set() + self.rules = Rules() + self._self_id: int | None = None + self._last_seen: dict[int, datetime] = {} async def refresh(self) -> None: rows = await watches_read.list_watches(self._pool, self._account_id) - self.watches = [watch for watch in rows if watch.enabled] - logger.info(f"[green]Watches cached:[/] {len(self.watches)}") + self.rules = build_rules(rows) + self._self_id = await self_user_id(self._pool, self._account_id) + logger.info(f"[green]Watches cached:[/] {len(self.rules)}") - async def on_text(self, chat_id: int, message_id: int, text: str | None) -> None: - if not text: - return - for watch in self.watches: - if watch.kind == KIND_KEYWORD and match_keyword( - text, chat_id, watch.params - ): - await watches_read.insert_alert( - self._pool, - self._account_id, - watch.id, - {"chat_id": chat_id, "message_id": message_id}, - dedup_key=f"{chat_id}:{message_id}", + def _own(self, seen: Seen) -> bool: + return seen.outgoing or ( + self._self_id is not None and seen.sender_id == self._self_id + ) + + async def _guard(self, work: Coroutine[Any, Any, None]) -> None: + try: + await work + except Exception: + logger.exception("[red]Watch evaluation failed[/]") + + async def _fire(self, watch_id: int, payload: dict[str, Any], key: str) -> None: + await watches_read.insert_alert( + self._pool, self._account_id, watch_id, payload, dedup_key=key + ) + + async def _keyword(self, event: str, seen: Seen) -> None: + for rule in self.rules.keyword: + if match_keyword(rule, seen): + await self._fire( + rule.watch_id, + _message_payload(event, seen), + f"{seen.chat_id}:{seen.message_id}", ) - async def on_status(self, peer_id: int, *, is_online: bool) -> None: - if not is_online: - self._online.discard(peer_id) + async def on_message(self, seen: Seen) -> None: + if self._own(seen): return - if peer_id in self._online: - return - self._online.add(peer_id) - for watch in self.watches: - if watch.kind == KIND_PEER_ONLINE and match_peer_online( - peer_id, watch.params - ): - await watches_read.insert_alert( - self._pool, self._account_id, watch.id, {"peer_id": peer_id} + await self._guard(self._on_message(seen)) + + async def _on_message(self, seen: Seen) -> None: + await self._keyword("message", seen) + for watch_id, params in self.rules.new_message: + if match_new_message(params, seen): + await self._fire( + watch_id, + _message_payload("message", seen), + f"{seen.chat_id}:{seen.message_id}", ) + + async def on_edit(self, seen: Seen) -> None: + if self._own(seen): + return + await self._guard(self._on_edit(seen)) + + async def _on_edit(self, seen: Seen) -> None: + await self._keyword("edit", seen) + if seen.edit_date is None: + return + stamp = int(seen.edit_date.timestamp()) + for watch_id, params in self.rules.edited: + if match_chat_scope(params, seen): + await self._fire( + watch_id, + _message_payload("edit", seen), + f"{seen.chat_id}:{seen.message_id}:{stamp}", + ) + + async def on_transcript(self, chat_id: int, message_id: int, text: str) -> None: + if self.rules.keyword: + await self._guard(self._on_transcript(chat_id, message_id, text)) + + async def _on_transcript(self, chat_id: int, message_id: int, text: str) -> None: + sender_id = await repository.message_sender( + self._pool, self._account_id, chat_id, message_id + ) + seen = Seen(chat_id, message_id, sender_id, text) + if not self._own(seen): + await self._keyword("transcript", seen) + + async def on_deleted(self, chat_id: int | None, message_ids: list[int]) -> None: + if self.rules.deleted and message_ids: + await self._guard(self._on_deleted(chat_id, message_ids)) + + async def _on_deleted(self, chat_id: int | None, message_ids: list[int]) -> None: + rows = await repository.deleted_messages( + self._pool, self._account_id, chat_id, message_ids + ) + for row in rows: + seen = Seen( + row["chat_id"], + row["message_id"], + row["sender_id"], + row["text"], + date=row["date"], + ) + if self._own(seen): + continue + for watch_id, params in self.rules.deleted: + if match_chat_scope(params, seen): + await self._fire( + watch_id, + _message_payload("delete", seen), + f"{seen.chat_id}:{seen.message_id}", + ) + + async def on_status( + self, peer_id: int, status: str, at: datetime, last_online: datetime | None + ) -> None: + watch_ids = [ + watch_id + for watch_id, params in self.rules.online + if params.peer_id == peer_id + ] + if not watch_ids: + return + if status != "online": + self._last_seen[peer_id] = last_online or at + return + await self._guard(self._on_online(peer_id, at, watch_ids)) + + async def _on_online( + self, peer_id: int, at: datetime, watch_ids: list[int] + ) -> None: + stored = await repository.last_seen_before( + self._pool, self._account_id, peer_id, at + ) + known = [value for value in (stored, self._last_seen.get(peer_id)) if value] + previous = max(known, default=None) + self._last_seen[peer_id] = at + if previous is not None and at - previous < ONLINE_QUIET: + return + session = int((previous or at).timestamp()) + payload: dict[str, Any] = {"event": "online", "peer_id": peer_id} + if previous is not None: + payload["since"] = previous.isoformat() + for watch_id in watch_ids: + await self._fire(watch_id, payload, f"{peer_id}:{session}") diff --git a/backend/src/userbot/modules/watches/evaluator.py b/backend/src/userbot/modules/watches/evaluator.py index 7abf0e9..84d5ac1 100644 --- a/backend/src/userbot/modules/watches/evaluator.py +++ b/backend/src/userbot/modules/watches/evaluator.py @@ -1,24 +1,104 @@ import re -from typing import Any +from dataclasses import dataclass, field +from datetime import datetime -KIND_KEYWORD = "keyword" -KIND_PEER_ONLINE = "peer_online" +from utils.logging import logger +from utils.read.models import WatchView +from utils.watch_rules import ( + MAX_TEXT, + DeletedParams, + EditedParams, + KeywordParams, + NewMessageParams, + PeerOnlineParams, + WatchParamsError, + compile_pattern, + parse_params, +) -def match_keyword(text: str, chat_id: int, params: dict[str, Any]) -> bool: - target_chat = params.get("chat_id") - if target_chat is not None and target_chat != chat_id: - return False - pattern = params.get("pattern") - if not pattern: - return False - if params.get("regex"): +@dataclass(frozen=True, slots=True) +class Seen: + chat_id: int + message_id: int + sender_id: int | None + text: str | None + outgoing: bool = False + date: datetime | None = None + edit_date: datetime | None = None + + +@dataclass(frozen=True, slots=True) +class KeywordRule: + watch_id: int + params: KeywordParams + regex: re.Pattern[str] | None + + +@dataclass(slots=True) +class Rules: + keyword: list[KeywordRule] = field(default_factory=list) + new_message: list[tuple[int, NewMessageParams]] = field(default_factory=list) + deleted: list[tuple[int, DeletedParams]] = field(default_factory=list) + edited: list[tuple[int, EditedParams]] = field(default_factory=list) + online: list[tuple[int, PeerOnlineParams]] = field(default_factory=list) + + def __len__(self) -> int: + return ( + len(self.keyword) + + len(self.new_message) + + len(self.deleted) + + len(self.edited) + + len(self.online) + ) + + +def build_rules(watches: list[WatchView]) -> Rules: + rules = Rules() + for watch in watches: + if not watch.enabled: + continue try: - return re.search(pattern, text) is not None - except re.error: - return False - return pattern.casefold() in text.casefold() + params = parse_params(watch.kind, watch.params) + except WatchParamsError as exc: + logger.warning(f"[yellow]Watch {watch.id} skipped:[/] {exc}") + continue + match params: + case KeywordParams(): + regex = compile_pattern(params.pattern) if params.regex else None + rules.keyword.append(KeywordRule(watch.id, params, regex)) + case NewMessageParams(): + rules.new_message.append((watch.id, params)) + case DeletedParams(): + rules.deleted.append((watch.id, params)) + case EditedParams(): + rules.edited.append((watch.id, params)) + case PeerOnlineParams(): + rules.online.append((watch.id, params)) + return rules -def match_peer_online(peer_id: int, params: dict[str, Any]) -> bool: - return params.get("peer_id") == peer_id +def _in_scope(chat_id: int | None, sender_id: int | None, seen: Seen) -> bool: + if chat_id is not None and chat_id != seen.chat_id: + return False + return sender_id is None or sender_id == seen.sender_id + + +def match_keyword(rule: KeywordRule, seen: Seen) -> bool: + params = rule.params + if not seen.text or not _in_scope(params.chat_id, params.sender_id, seen): + return False + sample = seen.text[:MAX_TEXT] + if rule.regex is not None: + return rule.regex.search(sample) is not None + return params.pattern.casefold() in sample.casefold() + + +def match_new_message(params: NewMessageParams, seen: Seen) -> bool: + return _in_scope(params.chat_id, params.sender_id, seen) + + +def match_chat_scope(params: DeletedParams | EditedParams, seen: Seen) -> bool: + if params.chat_id is None: + return seen.chat_id > 0 + return params.chat_id == seen.chat_id diff --git a/backend/src/userbot/modules/watches/repository.py b/backend/src/userbot/modules/watches/repository.py new file mode 100644 index 0000000..229f5e6 --- /dev/null +++ b/backend/src/userbot/modules/watches/repository.py @@ -0,0 +1,48 @@ +from datetime import datetime + +import asyncpg + +from userbot.modules.capture.repository import CHANNEL_ID_THRESHOLD + +_DELETED = """ +SELECT DISTINCT ON (chat_id, message_id) chat_id, message_id, sender_id, text, date +FROM messages +WHERE account_id = $1 AND message_id = ANY($2::bigint[]) + AND deleted_at IS NOT NULL AND {scope} +ORDER BY chat_id, message_id, date DESC +""" + +_SENDER = """ +SELECT sender_id FROM messages +WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 +ORDER BY date DESC LIMIT 1 +""" + +_LAST_SEEN = """ +SELECT CASE WHEN status = 'online' THEN ts ELSE coalesce(last_online_date, ts) END +FROM presence +WHERE account_id = $1 AND peer_id = $2 AND ts < $3 +ORDER BY ts DESC LIMIT 1 +""" + + +async def deleted_messages( + pool: asyncpg.Pool, account_id: int, chat_id: int | None, message_ids: list[int] +) -> list[asyncpg.Record]: + if chat_id is None: + query = _DELETED.format(scope="chat_id > $3") + return await pool.fetch(query, account_id, message_ids, CHANNEL_ID_THRESHOLD) + query = _DELETED.format(scope="chat_id = $3") + return await pool.fetch(query, account_id, message_ids, chat_id) + + +async def message_sender( + pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int +) -> int | None: + return await pool.fetchval(_SENDER, account_id, chat_id, message_id) + + +async def last_seen_before( + pool: asyncpg.Pool, account_id: int, peer_id: int, at: datetime +) -> datetime | None: + return await pool.fetchval(_LAST_SEEN, account_id, peer_id, at) diff --git a/backend/src/userbot/modules/watches/seen.py b/backend/src/userbot/modules/watches/seen.py new file mode 100644 index 0000000..07782b9 --- /dev/null +++ b/backend/src/userbot/modules/watches/seen.py @@ -0,0 +1,16 @@ +from pyrogram.types import Message + +from userbot.modules.capture.message import sender_id +from userbot.modules.watches.evaluator import Seen + + +def seen_message(chat_id: int, message: Message) -> Seen: + return Seen( + chat_id=chat_id, + message_id=message.id, + sender_id=sender_id(message), + text=message.text or message.caption, + outgoing=bool(message.outgoing), + date=message.date, + edit_date=message.edit_date, + ) diff --git a/backend/src/utils/watch_rules.py b/backend/src/utils/watch_rules.py new file mode 100644 index 0000000..614a892 --- /dev/null +++ b/backend/src/utils/watch_rules.py @@ -0,0 +1,145 @@ +import re +from collections.abc import Iterable, Iterator +from importlib import import_module +from typing import Any, Literal, Self, get_args + +from pydantic import BaseModel, Field, ValidationError, model_validator + +MAX_PATTERN = 200 +MAX_TEXT = 4096 +MAX_NAME = 120 + +WatchKind = Literal[ + "keyword", "new_message", "message_deleted", "message_edited", "peer_online" +] +WATCH_KINDS: tuple[str, ...] = get_args(WatchKind) + +sre_parser = import_module("re._parser") +_REPEATS = {sre_parser.MAX_REPEAT, sre_parser.MIN_REPEAT, sre_parser.POSSESSIVE_REPEAT} + + +class WatchParamsError(ValueError): + pass + + +def _subpatterns(value: object) -> Iterator[Any]: + if isinstance(value, sre_parser.SubPattern): + yield value + elif isinstance(value, tuple | list): + for item in value: + yield from _subpatterns(item) + + +def _repeat_depth(items: Iterable[tuple[Any, Any]]) -> int: + depth = 0 + for op, value in items: + if op in _REPEATS: + low, high, body = value + depth = max(depth, _repeat_depth(body) + (high > max(low, 1))) + else: + inner = (_repeat_depth(sub) for sub in _subpatterns(value)) + depth = max(depth, *inner, 0) + return depth + + +def compile_pattern(pattern: str) -> re.Pattern[str]: + try: + compiled = re.compile(pattern, re.IGNORECASE) + except re.error as exc: + msg = f"invalid regex: {exc}" + raise ValueError(msg) from exc + if _repeat_depth(sre_parser.parse(pattern)) > 1: + msg = "regex repeats inside a repeat, like (a+)+, which can hang" + raise ValueError(msg) + return compiled + + +class KeywordParams(BaseModel): + pattern: str = Field(min_length=1, max_length=MAX_PATTERN) + regex: bool = False + chat_id: int | None = None + sender_id: int | None = None + + @model_validator(mode="before") + @classmethod + def _legacy_keyword(cls, data: object) -> object: + if isinstance(data, dict) and "pattern" not in data and "keyword" in data: + return {**data, "pattern": data["keyword"]} + return data + + @model_validator(mode="after") + def _valid_pattern(self) -> Self: + if not self.pattern.strip(): + msg = "pattern is blank" + raise ValueError(msg) + if self.regex: + compile_pattern(self.pattern) + return self + + +class NewMessageParams(BaseModel): + chat_id: int | None = None + sender_id: int | None = None + + @model_validator(mode="after") + def _scoped(self) -> Self: + if self.chat_id is None and self.sender_id is None: + msg = "set chat_id, sender_id or both" + raise ValueError(msg) + return self + + +class DeletedParams(BaseModel): + chat_id: int | None = None + + +class EditedParams(BaseModel): + chat_id: int | None = None + + +class PeerOnlineParams(BaseModel): + peer_id: int + + +WatchParams = ( + KeywordParams | NewMessageParams | DeletedParams | EditedParams | PeerOnlineParams +) + +PARAMS_BY_KIND: dict[str, type[WatchParams]] = { + "keyword": KeywordParams, + "new_message": NewMessageParams, + "message_deleted": DeletedParams, + "message_edited": EditedParams, + "peer_online": PeerOnlineParams, +} + + +def _describe(exc: ValidationError) -> str: + parts = [] + for error in exc.errors(): + where = ".".join(str(part) for part in error["loc"]) + message = error["msg"].removeprefix("Value error, ") + parts.append(f"{where}: {message}" if where else message) + return "; ".join(parts) + + +def parse_params(kind: str, params: dict[str, Any]) -> WatchParams: + model = PARAMS_BY_KIND.get(kind) + if model is None: + msg = f"unknown kind {kind!r}, expected one of {', '.join(WATCH_KINDS)}" + raise WatchParamsError(msg) + try: + return model.model_validate(params) + except ValidationError as exc: + raise WatchParamsError(_describe(exc)) from exc + + +def normalize_params(kind: str, params: dict[str, Any]) -> dict[str, Any]: + return parse_params(kind, params).model_dump(exclude_none=True) + + +def clean_name(name: str | None) -> str | None: + if name is None: + return None + name = name.strip()[:MAX_NAME] + return name or None diff --git a/backend/src/watcher/registry.py b/backend/src/watcher/registry.py index fe7a1e1..8e01163 100644 --- a/backend/src/watcher/registry.py +++ b/backend/src/watcher/registry.py @@ -15,6 +15,7 @@ from userbot.modules.capture import capture_message from userbot.modules.capture.chat_meta import meta_from_chat from userbot.modules.client import PyroClient from userbot.modules.jobs import JobConsumer +from userbot.modules.watches.seen import seen_message from utils.events import notify_bg_event from utils.jobs import enqueue from utils.logging import logger @@ -51,9 +52,11 @@ def _handlers(pool: asyncpg.Pool) -> tuple: if ctx is None or message.empty or message.chat is None or message.date is None: return meta = meta_from_chat(message.chat, ctx.contacts.ids) - await ctx.watches.on_text( - meta.chat_id, message.id, message.text or message.caption - ) + seen = seen_message(meta.chat_id, message) + if message.edit_date is None: + await ctx.watches.on_message(seen) + else: + await ctx.watches.on_edit(seen) toggles = ctx.resolve(meta) if not toggles.messages: return @@ -115,6 +118,12 @@ class BotRegistry: for bot_id in list(self._running): await self._stop(bot_id) + async def refresh_watches(self) -> None: + for running in list(self._running.values()): + capture = getattr(running.client, "capture", None) + if capture is not None: + await capture.watches.refresh() + async def _adopt_dropped_in(self) -> None: unknown = tokens.stored_bot_ids() - await bots.known_bot_ids(self._pool) for bot_id in unknown: diff --git a/backend/src/watcher/runner.py b/backend/src/watcher/runner.py index 6068a3d..63bb3ac 100644 --- a/backend/src/watcher/runner.py +++ b/backend/src/watcher/runner.py @@ -9,6 +9,7 @@ from dependencies.container import container from utils.listen import Listener from utils.logging import logger, setup_logging from utils.read.bots import BOTS_CHANGED_CHANNEL +from utils.read.watches import WATCHES_CHANGED_CHANNEL from utils.storage import ContentAddressedStorage from watcher.registry import BotRegistry @@ -32,7 +33,20 @@ def _change_listener(registry: BotRegistry, tasks: set[asyncio.Task]) -> Listene tasks.add(task) task.add_done_callback(tasks.discard) - return Listener("watcher", {BOTS_CHANGED_CHANNEL: on_change}, lambda: on_change("")) + def on_watch(_payload: str) -> None: + task = asyncio.create_task(registry.refresh_watches()) + tasks.add(task) + task.add_done_callback(tasks.discard) + + def on_reconnect() -> None: + on_change("") + on_watch("") + + return Listener( + "watcher", + {BOTS_CHANGED_CHANNEL: on_change, WATCHES_CHANGED_CHANNEL: on_watch}, + on_reconnect, + ) async def runner() -> None: