feat(watcher,api,frontend): archive who writes to watched bots over mtproto

backend/src/utils/env.py

@@ -31,6 +31,12 @@ class TelegramSettings(BaseSettings):
     sessions_dir: str = "sessions"
 
 
+class BotsSettings(BaseSettings):
+    sessions_dir: str = "sessions/bots"
+    backfill_per_chat: int = 0
+    backfill_chats: int = 0
+
+
 class ApiSettings(BaseSettings):
     host: str = "0.0.0.0"  # noqa: S104
     port: int = 8080
@@ -56,6 +62,7 @@ class LogSettings(BaseSettings):
 class Settings(BaseSettings):
     db: DatabaseSettings = Field(default_factory=DatabaseSettings)
     tg: TelegramSettings = Field(default_factory=TelegramSettings)
+    bots: BotsSettings = Field(default_factory=BotsSettings)
     api: ApiSettings = Field(default_factory=ApiSettings)
     auth: AuthSettings = Field(default_factory=AuthSettings)
     storage: StorageSettings = Field(default_factory=StorageSettings)

backend/src/utils/read/bots.py

@@ -0,0 +1,261 @@
+from datetime import UTC, datetime, timedelta
+
+import asyncpg
+
+from utils.read.models import (
+    BotBucket,
+    BotCount,
+    BotHour,
+    BotSender,
+    BotStats,
+    BotSummary,
+    BotUpdateView,
+    BotView,
+)
+
+BOTS_CHANGED_CHANNEL = "watched_bots_changed"
+
+STATUS_PENDING = "pending"
+STATUS_OK = "ok"
+STATUS_UNAUTHORIZED = "unauthorized"
+STATUS_ERROR = "error"
+STATUS_REMOVED = "removed"
+
+_BOT_COLS = """
+SELECT b.bot_id, b.username, b.title, b.enabled, b.status, b.status_detail,
+       b.backfilled_at, b.checked_at, b.added_at,
+       coalesce(u.incoming, 0) AS incoming,
+       coalesce(u.senders, 0) AS senders,
+       u.last_ts
+FROM watched_bots b
+LEFT JOIN LATERAL (
+    SELECT count(*) AS incoming, count(DISTINCT sender_id) AS senders,
+           max(ts) AS last_ts
+    FROM bot_updates WHERE bot_id = b.bot_id AND NOT outgoing
+) u ON true
+"""
+
+_REGISTER = """
+INSERT INTO watched_bots (bot_id, status) VALUES ($1, 'pending')
+ON CONFLICT (bot_id) DO UPDATE SET
+    enabled = true, status = 'pending', status_detail = NULL, updated_at = now()
+"""
+
+_REMOVE = """
+UPDATE watched_bots
+SET enabled = false, status = 'removed', status_detail = NULL, updated_at = now()
+WHERE bot_id = $1 AND enabled
+"""
+
+_SET_STATUS = """
+UPDATE watched_bots SET
+    status = $2,
+    status_detail = $3,
+    username = coalesce($4, username),
+    title = coalesce($5, title),
+    checked_at = now(),
+    updated_at = now()
+WHERE bot_id = $1
+"""
+
+_STORE = """
+INSERT INTO bot_updates
+    (bot_id, kind, chat_id, item_id, ts, source, outgoing, sender_id,
+     sender_username, sender_name, sender_is_bot, chat_type, chat_title,
+     text, media_kind, edited_at, raw)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
+        $16, $17::jsonb)
+ON CONFLICT (bot_id, kind, chat_id, item_id, ts) DO UPDATE SET
+    text = EXCLUDED.text,
+    media_kind = EXCLUDED.media_kind,
+    chat_title = coalesce(EXCLUDED.chat_title, bot_updates.chat_title),
+    sender_username = coalesce(EXCLUDED.sender_username,
+                               bot_updates.sender_username),
+    sender_name = coalesce(EXCLUDED.sender_name, bot_updates.sender_name),
+    edited_at = coalesce(EXCLUDED.edited_at, bot_updates.edited_at),
+    raw = EXCLUDED.raw
+"""
+
+_INCOMING = "bot_id = $1 AND NOT outgoing AND ts >= $2"
+
+_SUMMARY = """
+SELECT count(*) FILTER (WHERE NOT outgoing) AS incoming,
+       count(*) FILTER (WHERE outgoing) AS outgoing,
+       count(DISTINCT sender_id) FILTER (WHERE NOT outgoing) AS senders,
+       count(DISTINCT chat_id) AS chats,
+       min(ts) AS first_ts, max(ts) AS last_ts
+FROM bot_updates WHERE bot_id = $1 AND ts >= $2
+"""
+
+_TIMELINE = f"""
+SELECT time_bucket('1 day', ts) AS bucket, count(*) AS incoming,
+       count(DISTINCT sender_id) AS senders
+FROM bot_updates WHERE {_INCOMING}
+GROUP BY bucket ORDER BY bucket
+"""  # noqa: S608
+
+_KINDS = f"""
+SELECT kind AS key, count(*) AS incoming
+FROM bot_updates WHERE {_INCOMING}
+GROUP BY kind ORDER BY incoming DESC
+"""  # noqa: S608
+
+_MEDIA = f"""
+SELECT media_kind AS key, count(*) AS incoming
+FROM bot_updates WHERE {_INCOMING} AND media_kind IS NOT NULL
+GROUP BY media_kind ORDER BY incoming DESC
+"""  # noqa: S608
+
+_HOURS = f"""
+SELECT EXTRACT(HOUR FROM ts)::int AS hour, count(*) AS incoming
+FROM bot_updates WHERE {_INCOMING}
+GROUP BY hour ORDER BY hour
+"""  # noqa: S608
+
+_SENDERS = f"""
+SELECT sender_id,
+       (array_agg(sender_username ORDER BY ts DESC) FILTER (
+           WHERE sender_username IS NOT NULL))[1] AS username,
+       (array_agg(sender_name ORDER BY ts DESC) FILTER (
+           WHERE sender_name IS NOT NULL))[1] AS name,
+       bool_or(sender_is_bot) AS is_bot,
+       count(*) AS incoming,
+       (array_agg(chat_id ORDER BY ts DESC))[1] AS chat_id,
+       min(ts) AS first_ts, max(ts) AS last_ts
+FROM bot_updates WHERE {_INCOMING} AND sender_id IS NOT NULL
+GROUP BY sender_id ORDER BY incoming DESC LIMIT $3
+"""  # noqa: S608
+
+_VIEW_COLS = (
+    "kind, chat_id, item_id, ts, source, outgoing, sender_id, sender_username, "
+    "sender_name, sender_is_bot, chat_type, chat_title, text, media_kind, edited_at"
+)
+
+
+async def notify_bots_changed(pool: asyncpg.Pool) -> None:
+    await pool.execute("SELECT pg_notify($1, '')", BOTS_CHANGED_CHANNEL)
+
+
+async def list_bots(pool: asyncpg.Pool) -> list[BotView]:
+    rows = await pool.fetch(f"{_BOT_COLS} ORDER BY u.last_ts DESC NULLS LAST")
+    return [BotView(**dict(row)) for row in rows]
+
+
+async def get_bot(pool: asyncpg.Pool, bot_id: int) -> BotView | None:
+    row = await pool.fetchrow(f"{_BOT_COLS} WHERE b.bot_id = $1", bot_id)
+    return BotView(**dict(row)) if row else None
+
+
+async def enabled_bot_ids(pool: asyncpg.Pool) -> set[int]:
+    rows = await pool.fetch(
+        "SELECT bot_id FROM watched_bots WHERE enabled AND status <> $1",
+        STATUS_UNAUTHORIZED,
+    )
+    return {row["bot_id"] for row in rows}
+
+
+async def register_bot(pool: asyncpg.Pool, bot_id: int) -> None:
+    await pool.execute(_REGISTER, bot_id)
+
+
+async def remove_bot(pool: asyncpg.Pool, bot_id: int) -> bool:
+    return await pool.execute(_REMOVE, bot_id) != "UPDATE 0"
+
+
+async def set_status(  # noqa: PLR0913
+    pool: asyncpg.Pool,
+    bot_id: int,
+    status: str,
+    *,
+    detail: str | None = None,
+    username: str | None = None,
+    title: str | None = None,
+) -> None:
+    await pool.execute(_SET_STATUS, bot_id, status, detail, username, title)
+
+
+async def needs_backfill(pool: asyncpg.Pool, bot_id: int) -> bool:
+    return not await pool.fetchval(
+        "SELECT backfilled_at IS NOT NULL FROM watched_bots WHERE bot_id = $1", bot_id
+    )
+
+
+async def mark_backfilled(pool: asyncpg.Pool, bot_id: int) -> None:
+    await pool.execute(
+        "UPDATE watched_bots SET backfilled_at = now(), updated_at = now() "
+        "WHERE bot_id = $1",
+        bot_id,
+    )
+
+
+async def store_update(pool: asyncpg.Pool, values: tuple) -> None:
+    await pool.execute(_STORE, *values)
+
+
+async def store_updates(pool: asyncpg.Pool, values: list[tuple]) -> None:
+    if values:
+        await pool.executemany(_STORE, values)
+
+
+async def stats(
+    pool: asyncpg.Pool, bot_id: int, *, days: int = 90, senders: int = 20
+) -> BotStats | None:
+    bot = await get_bot(pool, bot_id)
+    if bot is None:
+        return None
+    since = datetime.now(UTC) - timedelta(days=days)
+    summary = await pool.fetchrow(_SUMMARY, bot_id, since)
+    timeline = await pool.fetch(_TIMELINE, bot_id, since)
+    kinds = await pool.fetch(_KINDS, bot_id, since)
+    media = await pool.fetch(_MEDIA, bot_id, since)
+    hours = await pool.fetch(_HOURS, bot_id, since)
+    sender_rows = await pool.fetch(_SENDERS, bot_id, since, senders)
+    return BotStats(
+        bot=bot,
+        summary=BotSummary(**dict(summary)),
+        timeline=[BotBucket(**dict(row)) for row in timeline],
+        kinds=[BotCount(**dict(row)) for row in kinds],
+        media=[BotCount(**dict(row)) for row in media],
+        hours=[BotHour(**dict(row)) for row in hours],
+        senders=[BotSender(**dict(row)) for row in sender_rows],
+    )
+
+
+async def list_updates(  # noqa: PLR0913
+    pool: asyncpg.Pool,
+    bot_id: int,
+    *,
+    sender_id: int | None = None,
+    chat_id: int | None = None,
+    incoming_only: bool = True,
+    limit: int = 50,
+    offset: int = 0,
+) -> list[BotUpdateView]:
+    rows = await pool.fetch(
+        f"SELECT {_VIEW_COLS} FROM bot_updates "  # noqa: S608
+        "WHERE bot_id = $1 "
+        "AND ($2::bigint IS NULL OR sender_id = $2) "
+        "AND ($3::bigint IS NULL OR chat_id = $3) "
+        "AND (NOT $4 OR NOT outgoing) "
+        "ORDER BY ts DESC LIMIT $5 OFFSET $6",
+        bot_id,
+        sender_id,
+        chat_id,
+        incoming_only,
+        limit,
+        offset,
+    )
+    return [BotUpdateView(**dict(row)) for row in rows]
+
+
+async def removed_bot_ids(pool: asyncpg.Pool) -> set[int]:
+    rows = await pool.fetch(
+        "SELECT bot_id FROM watched_bots WHERE NOT enabled AND status = $1",
+        STATUS_REMOVED,
+    )
+    return {row["bot_id"] for row in rows}
+
+
+async def known_bot_ids(pool: asyncpg.Pool) -> set[int]:
+    rows = await pool.fetch("SELECT bot_id FROM watched_bots")
+    return {row["bot_id"] for row in rows}

backend/src/utils/read/models.py

@@ -423,3 +423,82 @@ class FileShareHitView(BaseModel):
     ip: str | None
     user_agent: str | None
     counted: bool
+
+
+class BotView(BaseModel):
+    bot_id: int
+    username: str | None
+    title: str | None
+    enabled: bool
+    status: str
+    status_detail: str | None
+    backfilled_at: datetime | None
+    checked_at: datetime | None
+    added_at: datetime
+    incoming: int
+    senders: int
+    last_ts: datetime | None
+
+
+class BotSummary(BaseModel):
+    incoming: int
+    outgoing: int
+    senders: int
+    chats: int
+    first_ts: datetime | None
+    last_ts: datetime | None
+
+
+class BotSender(BaseModel):
+    sender_id: int
+    username: str | None
+    name: str | None
+    is_bot: bool | None
+    incoming: int
+    chat_id: int | None
+    first_ts: datetime
+    last_ts: datetime
+
+
+class BotBucket(BaseModel):
+    bucket: datetime
+    incoming: int
+    senders: int
+
+
+class BotCount(BaseModel):
+    key: str
+    incoming: int
+
+
+class BotHour(BaseModel):
+    hour: int
+    incoming: int
+
+
+class BotStats(BaseModel):
+    bot: BotView
+    summary: BotSummary
+    timeline: list[BotBucket]
+    kinds: list[BotCount]
+    media: list[BotCount]
+    hours: list[BotHour]
+    senders: list[BotSender]
+
+
+class BotUpdateView(BaseModel):
+    kind: str
+    chat_id: int
+    item_id: int
+    ts: datetime
+    source: str
+    outgoing: bool
+    sender_id: int | None
+    sender_username: str | None
+    sender_name: str | None
+    sender_is_bot: bool | None
+    chat_type: str | None
+    chat_title: str | None
+    text: str | None
+    media_kind: str | None
+    edited_at: datetime | None

backend/src/utils/telegram.py

@@ -0,0 +1,2 @@
+API_ID = 2040
+API_HASH = "b18441a1ff607e10a989891a5462e627"