fix(userbot): mark deletions per row to stay under timescale decompression limit

backend/src/api/mcp/server.py

@@ -4,13 +4,17 @@ from typing import Any
 
 import asyncpg
 from fastmcp import FastMCP
+from fastmcp.exceptions import ToolError
+from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
+from fastmcp.tools.base import ToolResult
+from mcp.types import CallToolRequestParams
 from pydantic import BaseModel
 
 from api.mcp.format import build_transcript, load_notes, load_photos, resolve_names
 from dependencies.container import container
 from utils.jobs import enqueue
 from utils.read import annotations, chats, media, peers, presence, social, watches
-from utils.read.accounts import self_user_id
+from utils.read.accounts import self_user_id, unknown_account_hint
 from utils.read.models import DEFAULT_LIMIT, Page
 from utils.search.models import SearchFilters
 from utils.search.repository import search_messages
@@ -23,6 +27,9 @@ peer history) into Postgres and exposes it read-only over these tools.
 Account scoping:
 - Every tool takes `account_id`. It selects which archived Telegram account to
   read. Unless the user names a different one, always pass `account_id=1`.
+- `account_id` is beavergram's own small serial (1, 2, ...), never the archived
+  account's Telegram user id. Passing a Telegram id returns an error naming the
+  right `account_id`.
 
 Identifiers:
 - `chat_id` and `peer_id` are Telegram IDs (negative for groups/channels,
@@ -63,6 +70,25 @@ async def _pool() -> asyncpg.Pool:
     return await container.get(asyncpg.Pool)
 
 
+class AccountGuard(Middleware):
+    """Reject tool calls scoped to an account_id that does not exist."""
+
+    async def on_call_tool(
+        self,
+        context: MiddlewareContext[CallToolRequestParams],
+        call_next: CallNext[CallToolRequestParams, ToolResult],
+    ) -> ToolResult:
+        account_id = (context.message.arguments or {}).get("account_id")
+        if isinstance(account_id, int) and not isinstance(account_id, bool):
+            hint = await unknown_account_hint(await _pool(), account_id)
+            if hint is not None:
+                raise ToolError(hint)
+        return await call_next(context)
+
+
+mcp.add_middleware(AccountGuard())
+
+
 def _dump(items: Sequence[BaseModel]) -> list[dict[str, Any]]:
     return [item.model_dump(mode="json") for item in items]
 

backend/src/userbot/handlers/edits.py

@@ -5,6 +5,7 @@ from userbot.modules.capture import repository
 from userbot.modules.capture.chat_meta import meta_from_chat
 from userbot.modules.capture.message import sender_id
 from userbot.modules.media import capture_media, media_unique_id, self_destruct_ttl
+from userbot.modules.raw_json import raw_json
 from utils.events import notify_bg_event
 
 
@@ -33,7 +34,7 @@ async def on_edited_message(client: PyroClient, message: Message) -> None:
         message.date,
         sender_id(message),
         message.text or message.caption,
-        str(message),
+        raw_json(message),
         message.edit_date,
         media_unique_id(message),
         has_media=message.media is not None,

backend/src/userbot/handlers/presence.py

@@ -4,6 +4,7 @@ from pyrogram.types import User
 
 from userbot import PyroClient
 from userbot.modules.presence import repository
+from userbot.modules.raw_json import raw_json
 from utils.events import notify_bg_event
 
 
@@ -20,7 +21,7 @@ async def on_user_status(client: PyroClient, user: User) -> None:
         user.status.name.lower(),
         user.last_online_date,
         user.next_offline_date,
-        str(user.raw),
+        raw_json(user.raw),
     )
     await ctx.watches.on_status(user.id, is_online=user.status.name.lower() == "online")
     await notify_bg_event(ctx.pool, "presence", ctx.account_id, chat_id=user.id)

backend/src/userbot/handlers/raw/profiles.py

@@ -7,6 +7,7 @@ from userbot.modules.avatars import capture_avatar
 from userbot.modules.profiles import active_username, snapshot_from_user
 from userbot.modules.profiles.parse import ProfileFields
 from userbot.modules.profiles.repository import get_peer, write_profile
+from userbot.modules.raw_json import raw_json
 
 HANDLES = (raw.types.UpdateUserName, raw.types.UpdateUser, raw.types.UpdateUserPhone)
 
@@ -24,7 +25,9 @@ async def _handle_user(
     current = await get_peer(ctx.pool, ctx.account_id, update.user_id)
     if current == fields:
         return
-    await write_profile(ctx.pool, ctx.account_id, update.user_id, fields, str(raw_user))
+    await write_profile(
+        ctx.pool, ctx.account_id, update.user_id, fields, raw_json(raw_user)
+    )
     changed_photo = current is None or current.photo_unique_id != photo_unique_id
     if photo_file_id and photo_unique_id and changed_photo:
         await capture_avatar(
@@ -34,7 +37,7 @@ async def _handle_user(
             "peer",
             photo_file_id,
             photo_unique_id,
-            str(raw_user),
+            raw_json(raw_user),
         )
 
 
@@ -65,9 +68,9 @@ async def handle(
                 "last_name": update.last_name or None,
                 "username": active_username(update.usernames),
             },
-            str(update),
+            raw_json(update),
         )
     elif isinstance(update, raw.types.UpdateUserPhone):
         await _merge_partial(
-            client, update.user_id, {"phone": update.phone or None}, str(update)
+            client, update.user_id, {"phone": update.phone or None}, raw_json(update)
         )

backend/src/userbot/handlers/raw/read_contents.py

@@ -4,6 +4,7 @@ from pyrogram import raw, utils
 
 from userbot import PyroClient
 from userbot.modules.capture.chat_meta import meta_from_chat_id
+from userbot.modules.raw_json import raw_json
 from userbot.modules.read_receipts import repository as receipts
 from userbot.modules.stt import repository
 from userbot.modules.stt.gate import safe_transcribe
@@ -41,7 +42,7 @@ async def handle(
                 cand_chat_id,
                 datetime.now(UTC),
                 message_id,
-                str(update),
+                raw_json(update),
             )
             continue
         if not untranscribed:

backend/src/userbot/handlers/raw/read_receipts.py

@@ -3,6 +3,7 @@ from datetime import UTC, datetime
 from pyrogram import raw, utils
 
 from userbot import PyroClient
+from userbot.modules.raw_json import raw_json
 from userbot.modules.read_receipts import repository
 from utils.events import notify_bg_event
 
@@ -23,7 +24,7 @@ async def handle(
         chat_id,
         datetime.now(UTC),
         update.max_id,
-        str(update),
+        raw_json(update),
     )
     await notify_bg_event(
         ctx.pool, "receipt", ctx.account_id, chat_id=chat_id, message_id=update.max_id

backend/src/userbot/modules/capture/identity.py

@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING
 from userbot.modules.capture.chat_meta import chat_kind
 from userbot.modules.profiles.parse import ProfileFields, snapshot_from_high_level
 from userbot.modules.profiles.repository import get_peer, write_profile
+from userbot.modules.raw_json import raw_json
 from utils.policy.models import ChatKind
 
 if TYPE_CHECKING:
@@ -59,7 +60,7 @@ async def _capture_peer(message: Message, ctx: CaptureContext) -> None:
     fields, photo_file_id, photo_unique_id = snapshot_from_high_level(user)
     if not await ctx.peer_identity.changed(ctx.pool, ctx.account_id, user.id, fields):
         return
-    await write_profile(ctx.pool, ctx.account_id, user.id, fields, str(user))
+    await write_profile(ctx.pool, ctx.account_id, user.id, fields, raw_json(user))
     if photo_file_id and photo_unique_id:
         from userbot.modules.avatars import note_avatar  # noqa: PLC0415
 
@@ -95,7 +96,7 @@ async def _capture_chat(message: Message, ctx: CaptureContext) -> None:
         photo_unique_id,
         None,
         message.date,
-        str(message),
+        raw_json(message),
     )
     if photo_file_id and photo_unique_id:
         from userbot.modules.avatars import note_avatar  # noqa: PLC0415

backend/src/userbot/modules/capture/message.py

@@ -6,6 +6,7 @@ from userbot.modules.capture.context import CaptureContext
 from userbot.modules.links import extract_links
 from userbot.modules.links import repository as links_repository
 from userbot.modules.media import capture_media, self_destruct_ttl
+from userbot.modules.raw_json import raw_json
 from utils.policy.models import CaptureToggles
 
 
@@ -47,7 +48,7 @@ async def capture_message(
         message.date,
         sender_id(message),
         message.text or message.caption,
-        str(message),
+        raw_json(message),
         has_media=message.media is not None,
         is_self_destruct=self_destruct_ttl(message) is not None,
     )
@@ -68,5 +69,5 @@ async def capture_message(
     links = extract_links(message)
     if links:
         await links_repository.insert_links(
-            ctx.pool, ctx.account_id, chat_id, message.id, links, str(message)
+            ctx.pool, ctx.account_id, chat_id, message.id, links, raw_json(message)
         )

backend/src/userbot/modules/capture/repository.py

@@ -30,6 +30,16 @@ ON CONFLICT (account_id, chat_id, message_id, date) DO UPDATE SET
     edited_at = now()
 """
 
+_FIND_DELETED = """
+SELECT chat_id, message_id, date FROM messages
+WHERE account_id = $1 AND message_id = ANY($2::bigint[]) AND deleted_at IS NULL
+"""
+
+_MARK_DELETED = """
+UPDATE messages SET deleted_at = now()
+WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 AND date = $4
+"""
+
 _INSERT_VERSION = """
 INSERT INTO message_versions
     (account_id, chat_id, message_id, observed_at, edit_date, text, raw)
@@ -128,30 +138,34 @@ async def max_message_id(
     )
 
 
+async def _mark_deleted(
+    pool: asyncpg.Pool, account_id: int, rows: list[asyncpg.Record]
+) -> None:
+    for row in rows:
+        await pool.execute(
+            _MARK_DELETED, account_id, row["chat_id"], row["message_id"], row["date"]
+        )
+
+
 async def mark_deleted_box(
     pool: asyncpg.Pool, account_id: int, message_ids: list[int]
 ) -> None:
-    await pool.execute(
+    rows = await pool.fetch(
-        "UPDATE messages SET deleted_at = now() "
+        f"{_FIND_DELETED} AND chat_id > $3",
-        "WHERE account_id = $1 AND message_id = ANY($2::bigint[]) "
-        "AND chat_id > $3 AND deleted_at IS NULL",
         account_id,
         message_ids,
         CHANNEL_ID_THRESHOLD,
     )
+    await _mark_deleted(pool, account_id, rows)
 
 
 async def mark_deleted_channel(
     pool: asyncpg.Pool, account_id: int, chat_id: int, message_ids: list[int]
 ) -> None:
-    await pool.execute(
+    rows = await pool.fetch(
-        "UPDATE messages SET deleted_at = now() "
+        f"{_FIND_DELETED} AND chat_id = $3", account_id, message_ids, chat_id
-        "WHERE account_id = $1 AND chat_id = $2 AND message_id = ANY($3::bigint[]) "
-        "AND deleted_at IS NULL",
-        account_id,
-        chat_id,
-        message_ids,
     )
+    await _mark_deleted(pool, account_id, rows)
 
 
 async def add_version(  # noqa: PLR0913

backend/src/userbot/modules/groups/service.py

@@ -5,6 +5,7 @@ from userbot.modules.avatars import capture_avatar
 from userbot.modules.capture.context import CaptureContext
 from userbot.modules.capture.message import sender_id
 from userbot.modules.groups import repository
+from userbot.modules.raw_json import raw_json
 
 
 async def capture_service(
@@ -28,7 +29,7 @@ async def capture_service(
             None,
             actor,
             ts,
-            str(message),
+            raw_json(message),
         )
     elif service is enums.MessageServiceType.NEW_CHAT_PHOTO:
         photo = message.new_chat_photo
@@ -43,11 +44,17 @@ async def capture_service(
             unique_id,
             actor,
             ts,
-            str(message),
+            raw_json(message),
         )
         if photo is not None and unique_id is not None:
             await capture_avatar(
-                client, ctx, chat_id, "chat", photo.file_id, unique_id, str(message)
+                client,
+                ctx,
+                chat_id,
+                "chat",
+                photo.file_id,
+                unique_id,
+                raw_json(message),
             )
     elif service is enums.MessageServiceType.DELETE_CHAT_PHOTO:
         await repository.insert_chat_history(
@@ -60,7 +67,7 @@ async def capture_service(
             None,
             actor,
             ts,
-            str(message),
+            raw_json(message),
         )
     elif service is enums.MessageServiceType.NEW_CHAT_MEMBERS:
         for member in message.new_chat_members or []:
@@ -73,7 +80,7 @@ async def capture_service(
                 "join",
                 actor,
                 ts,
-                str(message),
+                raw_json(message),
             )
     elif service is enums.MessageServiceType.LEFT_CHAT_MEMBER:
         member = message.left_chat_member
@@ -87,5 +94,5 @@ async def capture_service(
                 "leave",
                 actor,
                 ts,
-                str(message),
+                raw_json(message),
             )

backend/src/userbot/modules/jobs/handlers/backfill_stories.py

@@ -8,6 +8,7 @@ from userbot.modules.capture.context import CaptureContext
 from userbot.modules.jobs.context import JobContext
 from userbot.modules.jobs.registry import register
 from userbot.modules.stories.service import save_story
+from utils.logging import logger
 
 SAVE_EVERY = 10
 
@@ -38,6 +39,9 @@ async def _drain(
             raise
         except RPCError:
             continue
+        except Exception:
+            logger.exception(f"Failed to save story {story.id} from {name}")
+            continue
         saved += 1
         if saved % SAVE_EVERY == 0:
             await ctx.report_progress({"saved": saved, "source": name})

backend/src/userbot/modules/jobs/handlers/enrich_chat.py

@@ -11,6 +11,7 @@ from userbot.modules.jobs.context import JobContext
 from userbot.modules.jobs.registry import register
 from userbot.modules.profiles.parse import snapshot_from_high_level
 from userbot.modules.profiles.repository import write_profile
+from userbot.modules.raw_json import raw_json
 
 MEMBER_CAP = 200
 
@@ -26,7 +27,7 @@ LIMIT 100
 
 async def _save_user(ctx: CaptureContext, user: User) -> None:
     fields, photo_file_id, photo_unique_id = snapshot_from_high_level(user)
-    await write_profile(ctx.pool, ctx.account_id, user.id, fields, str(user))
+    await write_profile(ctx.pool, ctx.account_id, user.id, fields, raw_json(user))
     if photo_file_id and photo_unique_id:
         await note_avatar(
             ctx.pool, ctx.account_id, user.id, "peer", photo_unique_id, photo_file_id
@@ -48,7 +49,7 @@ async def _enrich_chat_meta(client: Client, ctx: CaptureContext, chat_id: int) -
         photo_unique_id,
         None,
         datetime.now(UTC),
-        str(chat),
+        raw_json(chat),
     )
     if photo_file_id and photo_unique_id:
         await note_avatar(

backend/src/userbot/modules/jobs/handlers/sync_contacts.py

@@ -5,6 +5,7 @@ from userbot.modules.jobs.context import JobContext
 from userbot.modules.jobs.registry import register
 from userbot.modules.profiles.parse import snapshot_from_high_level
 from userbot.modules.profiles.repository import write_profile
+from userbot.modules.raw_json import raw_json
 
 
 @register("sync_contacts")
@@ -21,7 +22,7 @@ async def sync_contacts(ctx: JobContext) -> None:
         if not isinstance(user, User):
             continue
         fields, photo_file_id, photo_unique_id = snapshot_from_high_level(user)
-        await write_profile(ctx.pool, ctx.account_id, user.id, fields, str(user))
+        await write_profile(ctx.pool, ctx.account_id, user.id, fields, raw_json(user))
         if photo_file_id and photo_unique_id:
             await note_avatar(
                 ctx.pool,

backend/src/userbot/modules/jobs/handlers/sync_dialogs.py

@@ -9,6 +9,8 @@ from userbot.modules.jobs.registry import register
 from userbot.modules.profiles.parse import snapshot_from_high_level
 from userbot.modules.profiles.repository import write_profile
 from userbot.modules.profiles.snapshots import save_group, save_private
+from userbot.modules.raw_json import raw_json
+from utils.logging import logger
 
 SAVE_EVERY = 100
 USERS_BATCH = 200
@@ -31,7 +33,9 @@ async def _enrich_users(client: Client, ctx: CaptureContext, ids: list[int]) ->
             if not isinstance(user, User):
                 continue
             fields, photo_file_id, photo_unique_id = snapshot_from_high_level(user)
-            await write_profile(ctx.pool, ctx.account_id, user.id, fields, str(user))
+            await write_profile(
+                ctx.pool, ctx.account_id, user.id, fields, raw_json(user)
+            )
             if photo_file_id and photo_unique_id:
                 await note_avatar(
                     ctx.pool,
@@ -66,6 +70,8 @@ async def sync_dialogs(ctx: JobContext) -> None:
                 await save_group(capture, chat)
         except (BadRequest, Forbidden):
             pass
+        except Exception:
+            logger.exception(f"Failed to snapshot dialog {chat_id}")
         await ctx.pool.execute(_UPSERT_DIALOG, ctx.account_id, chat_id)
         processed += 1
         if processed % SAVE_EVERY == 0:

backend/src/userbot/modules/profiles/snapshots.py

@@ -7,12 +7,13 @@ from userbot.modules.capture.context import CaptureContext
 from userbot.modules.groups.repository import insert_chat_history
 from userbot.modules.profiles.parse import snapshot_from_chat
 from userbot.modules.profiles.repository import write_profile
+from userbot.modules.raw_json import raw_json
 
 
 async def save_private(ctx: CaptureContext, chat: Chat) -> bool:
     chat_id = chat.id or 0
     fields, photo_file_id, photo_unique_id = snapshot_from_chat(chat)
-    await write_profile(ctx.pool, ctx.account_id, chat_id, fields, str(chat))
+    await write_profile(ctx.pool, ctx.account_id, chat_id, fields, raw_json(chat))
     if photo_file_id and photo_unique_id:
         await note_avatar(
             ctx.pool, ctx.account_id, chat_id, "peer", photo_unique_id, photo_file_id
@@ -35,7 +36,7 @@ async def save_group(ctx: CaptureContext, chat: Chat) -> None:
         photo_unique_id,
         None,
         datetime.now(UTC),
-        str(chat),
+        raw_json(chat),
     )
     if photo_file_id and photo_unique_id:
         await note_avatar(

backend/src/userbot/modules/raw_json.py

@@ -0,0 +1,8 @@
+from pyrogram.raw.core import TLObject
+from pyrogram.types import Object
+
+EMPTY_JSON = "{}"
+
+
+def raw_json(obj: Object | TLObject | None) -> str:
+    return EMPTY_JSON if obj is None else str(obj)

backend/src/userbot/modules/stories/service.py

@@ -3,6 +3,7 @@ from pyrogram.types import Story
 
 from userbot.modules.capture.context import CaptureContext
 from userbot.modules.download import download_bytes
+from userbot.modules.raw_json import raw_json
 from userbot.modules.stories import repository
 
 
@@ -40,7 +41,7 @@ async def save_story(client: Client, capture: CaptureContext, story: Story) -> N
         storage_key,
         file_size,
         story.views,
-        str(story.raw),
+        raw_json(story.raw),
         pinned=bool(story.pinned),
         deleted=bool(story.deleted),
         downloaded=downloaded,

backend/src/utils/read/accounts.py

@@ -45,6 +45,35 @@ async def self_user_id(pool: asyncpg.Pool, account_id: int) -> int | None:
     return tg_user_id
 
 
+_known_ids: set[int] = set()
+
+
+def _account_label(row: asyncpg.Record) -> str:
+    named = f", {row['label']}" if row["label"] else ""
+    return f"account_id={row['account_id']} (Telegram id {row['tg_user_id']}{named})"
+
+
+async def unknown_account_hint(pool: asyncpg.Pool, account_id: int) -> str | None:
+    """Explain an account_id that matches no account, or None if it is valid."""
+    if account_id in _known_ids:
+        return None
+    rows = await pool.fetch(
+        "SELECT account_id, tg_user_id, label FROM accounts ORDER BY account_id"
+    )
+    _known_ids.update(row["account_id"] for row in rows)
+    if account_id in _known_ids:
+        return None
+    known = "; ".join(_account_label(row) for row in rows) or "none"
+    mistaken = next((row for row in rows if row["tg_user_id"] == account_id), None)
+    if mistaken is not None:
+        return (
+            f"{account_id} is a Telegram user id, not an account_id. In beavergram "
+            f"that account is account_id={mistaken['account_id']}. "
+            f"Retry with account_id={mistaken['account_id']}."
+        )
+    return f"No account with account_id={account_id}. Archived accounts: {known}."
+
+
 async def list_accounts(pool: asyncpg.Pool) -> list[AccountView]:
     rows = await pool.fetch(
         f"SELECT {_ACCOUNT_COLS} FROM accounts ORDER BY account_id"  # noqa: S608