feat(backend,frontend,mcp): saved messages take text, files, voices and archived messages, the one write to telegram

backend/src/api/app.py

@@ -34,6 +34,7 @@ from api.routers import (
     policy,
     presence,
     profile,
+    saved,
     search,
     shares,
     social,
@@ -113,6 +114,7 @@ app.include_router(annotations.router)
 app.include_router(watches.router)
 app.include_router(webhooks.router)
 app.include_router(shares.router)
+app.include_router(saved.router)
 app.include_router(storage.router)
 app.include_router(files.router)
 app.include_router(capture.router)

backend/src/api/mcp/server.py

@@ -19,6 +19,7 @@ from api.mcp.download import (
     share_media,
 )
 from api.mcp.format import build_transcript, load_notes, load_photos, resolve_names
+from api.saved import FetchError, fetch_file, forward, send_new
 from api.transcribe import request_transcript
 from dependencies.container import container
 from utils.files import resolve_mime
@@ -46,7 +47,8 @@ from utils.watch_rules import WatchKind, WatchParamsError, clean_name, normalize
 
 INSTRUCTIONS = """\
 beavergram archives Telegram data (chats, messages, media, presence, stories,
-peer history) into Postgres and exposes it read-only over these tools.
+peer history) into Postgres and exposes it over these tools, read-only except
+for the few writes listed at the end.
 
 Account scoping:
 - Every tool takes `account_id`. It selects which archived Telegram account to
@@ -101,8 +103,10 @@ Watched bots:
   own `account_id` and its chats, messages and media read through the same
   tools as a user account. Nothing bot-specific to call.
 
-Writes: everything is read-only except `set_watch` and the share links
-`download_media` creates (visible and revocable in the web UI).
+Writes: everything is read-only except `set_watch`, the share links
+`download_media` creates (visible and revocable in the web UI), and
+`send_to_saved`, the one tool that sends to Telegram: only to the user's own
+Saved Messages.
 
 Paging: list tools accept `limit` and `offset`; `search_messages_tool` returns
 `next_cursor` to pass back as `cursor`.
@@ -392,6 +396,49 @@ async def transcribe_message(
     return result.model_dump(mode="json", exclude_none=True)
 
 
+@mcp.tool
+async def send_to_saved(
+    account_id: int,
+    text: str | None = None,
+    url: str | None = None,
+    voice: bool = False,
+    chat_id: int | None = None,
+    message_id: int | None = None,
+) -> dict[str, Any]:
+    """Send something to the account's own Saved Messages.
+
+    The only tool that writes to Telegram, and only to the user's own chat.
+    - `chat_id` + `message_id`: an archived message. It is forwarded while it
+      still exists in Telegram; a deleted, self-destructing or unforwardable
+      one goes as a copy from the archive. `mode` says which.
+    - `text` and/or `url`: a new message. The server fetches `url` (http or
+      https, up to 200 MB) and sends it as a file with `text` as the caption;
+      `voice=true` sends it as a voice message exactly as given, so pass
+      OGG/Opus. A local file can skip the URL: `POST /api/saved` (multipart
+      `account_id`, `text`, `voice`, `file`) with this server's bearer.
+    Returns `status` (`done`, `pending`, `failed` with `error`) and the
+    `chat_id`/`message_id` of the new message. Any audio can be transcribed
+    through Telegram Premium: send it with `voice=true`, then call
+    `transcribe_message` on the returned ids.
+    """
+    pool = await _pool()
+    if chat_id is not None and message_id is not None:
+        result = await forward(pool, account_id, chat_id, message_id)
+        return result.model_dump(mode="json", exclude_none=True)
+    upload = None
+    if url is not None:
+        try:
+            upload = await fetch_file(url)
+        except FetchError as exc:
+            raise ToolError(str(exc)) from exc
+    if upload is None and not text:
+        msg = "pass text, url, or chat_id with message_id"
+        raise ToolError(msg)
+    storage = await container.get(ContentAddressedStorage)
+    result = await send_new(pool, storage, account_id, text, upload, voice=voice)
+    return result.model_dump(mode="json", exclude_none=True)
+
+
 @mcp.tool
 async def get_callbacks(
     account_id: int, chat_id: int, message_id: int

backend/src/api/routers/saved.py

@@ -0,0 +1,44 @@
+from typing import Annotated
+
+import asyncpg
+from dishka.integrations.fastapi import DishkaRoute, FromDishka
+from fastapi import APIRouter, File, Form, HTTPException, UploadFile, status
+from pydantic import BaseModel
+
+from api.saved import MAX_UPLOAD_BYTES, SavedResult, forward, send_new
+from utils.storage import ContentAddressedStorage
+
+router = APIRouter(prefix="/api/saved", tags=["saved"], route_class=DishkaRoute)
+
+
+class ForwardRequest(BaseModel):
+    account_id: int
+    chat_id: int
+    message_id: int
+
+
+@router.post("")
+async def send_to_saved(
+    pool: FromDishka[asyncpg.Pool],
+    storage: FromDishka[ContentAddressedStorage],
+    account_id: Annotated[int, Form()],
+    text: Annotated[str | None, Form()] = None,
+    voice: Annotated[bool, Form()] = False,
+    file: Annotated[UploadFile | None, File()] = None,
+) -> SavedResult:
+    upload = None
+    if file is not None:
+        if file.size is not None and file.size > MAX_UPLOAD_BYTES:
+            raise HTTPException(status.HTTP_413_CONTENT_TOO_LARGE, "file too large")
+        data = await file.read()
+        upload = (data, file.filename or "file", file.content_type)
+    if upload is None and not text:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "text or file is required")
+    return await send_new(pool, storage, account_id, text, upload, voice=voice)
+
+
+@router.post("/forward")
+async def forward_to_saved(
+    pool: FromDishka[asyncpg.Pool], body: ForwardRequest
+) -> SavedResult:
+    return await forward(pool, body.account_id, body.chat_id, body.message_id)

backend/src/api/saved.py

@@ -0,0 +1,131 @@
+import asyncio
+import json
+from pathlib import Path
+from typing import Any, Literal
+from urllib.parse import urlsplit
+
+import asyncpg
+import httpx2
+from pydantic import BaseModel
+
+from utils.jobs import enqueue
+from utils.read import accounts
+from utils.storage import ContentAddressedStorage
+
+MAX_UPLOAD_BYTES = 200 * 1024 * 1024
+WAIT_SECONDS = 90.0
+FETCH_TIMEOUT_SECONDS = 120.0
+_POLL_SECONDS = 0.5
+
+SavedMode = Literal["sent", "forwarded", "copied"]
+
+
+class SavedResult(BaseModel):
+    status: Literal["done", "pending", "failed"]
+    chat_id: int | None = None
+    message_id: int | None = None
+    mode: SavedMode | None = None
+    error: str | None = None
+    job_id: int | None = None
+
+
+class FetchError(Exception):
+    pass
+
+
+async def fetch_file(url: str) -> tuple[bytes, str, str | None]:
+    if urlsplit(url).scheme not in ("http", "https"):
+        msg = "url must be http or https"
+        raise FetchError(msg)
+    chunks: list[bytes] = []
+    size = 0
+    try:
+        async with (
+            httpx2.AsyncClient(
+                timeout=FETCH_TIMEOUT_SECONDS, follow_redirects=True
+            ) as client,
+            client.stream("GET", url) as response,
+        ):
+            response.raise_for_status()
+            async for chunk in response.aiter_bytes():
+                size += len(chunk)
+                if size > MAX_UPLOAD_BYTES:
+                    msg = "file is over 200 MB"
+                    raise FetchError(msg)
+                chunks.append(chunk)
+    except httpx2.HTTPError as exc:
+        msg = f"could not fetch {url}: {exc}"
+        raise FetchError(msg) from exc
+    name = Path(urlsplit(str(response.url)).path).name or "file"
+    mime = response.headers.get("content-type", "").split(";")[0].strip() or None
+    return b"".join(chunks), name, mime
+
+
+async def _refusal(pool: asyncpg.Pool, account_id: int) -> SavedResult | None:
+    account = await accounts.get_account(pool, account_id)
+    if account is None or account.is_bot:
+        return SavedResult(status="failed", error="bot_account")
+    return None
+
+
+async def _outcome(pool: asyncpg.Pool, job_id: int, mode: SavedMode) -> SavedResult:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + WAIT_SECONDS
+    while loop.time() < deadline:
+        await asyncio.sleep(_POLL_SECONDS)
+        row = await pool.fetchrow(
+            "SELECT status, error, progress FROM jobs WHERE id = $1", job_id
+        )
+        if row is None or row["status"] in ("failed", "canceled"):
+            error = (row and row["error"]) or "failed"
+            return SavedResult(status="failed", error=error, job_id=job_id)
+        if row["status"] == "done":
+            sent: dict[str, Any] = json.loads(row["progress"])
+            return SavedResult(
+                status="done",
+                chat_id=sent.get("chat_id"),
+                message_id=sent.get("message_id"),
+                mode=sent.get("mode", mode),
+                job_id=job_id,
+            )
+    return SavedResult(status="pending", job_id=job_id)
+
+
+async def send_new(  # noqa: PLR0913
+    pool: asyncpg.Pool,
+    storage: ContentAddressedStorage,
+    account_id: int,
+    text: str | None,
+    file: tuple[bytes, str, str | None] | None,
+    *,
+    voice: bool = False,
+) -> SavedResult:
+    refusal = await _refusal(pool, account_id)
+    if refusal is not None:
+        return refusal
+    params: dict[str, Any] = {"text": text}
+    if file is not None:
+        data, name, mime = file
+        params |= {
+            "upload": storage.stage(data),
+            "file_name": name,
+            "mime": mime,
+            "voice": voice,
+        }
+    job_id = await enqueue(pool, account_id, "send_saved", params)
+    return await _outcome(pool, job_id, "sent")
+
+
+async def forward(
+    pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int
+) -> SavedResult:
+    refusal = await _refusal(pool, account_id)
+    if refusal is not None:
+        return refusal
+    job_id = await enqueue(
+        pool,
+        account_id,
+        "forward_saved",
+        {"chat_id": chat_id, "message_id": message_id},
+    )
+    return await _outcome(pool, job_id, "forwarded")

backend/src/userbot/handlers/messages.py

@@ -1,12 +1,7 @@
 from pyrogram.types import Message
 
-from userbot.modules.capture import capture_message
-from userbot.modules.capture.chat_meta import meta_from_chat
+from userbot.modules.capture.ingest import ingest_message
 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
 
 
 @PyroClient.on_message()
@@ -20,21 +15,7 @@ async def on_message(client: PyroClient, message: Message) -> None:
         or message.date is None
     ):
         return
-    meta = meta_from_chat(message.chat, ctx.contacts.ids)
-    await ctx.watches.on_message(seen_message(meta.chat_id, message))
-    toggles = ctx.resolve(meta)
-    if not toggles.messages:
-        return
-    await capture_message(client, message, ctx, toggles)
-    await notify_bg_event(
-        ctx.pool, "message", ctx.account_id, chat_id=meta.chat_id, message_id=message.id
-    )
-    if (
-        toggles.stt
-        and is_transcribable(message)
-        and (message.outgoing or message.unread_media is False)
-    ):
-        await safe_transcribe(client, ctx, meta.chat_id, message.id)
+    await ingest_message(client, ctx, message)
 
 
 handlers = on_message.handlers  # ty: ignore[unresolved-attribute]

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

@@ -0,0 +1,30 @@
+from pyrogram import Client
+from pyrogram.types import Message
+
+from userbot.modules.capture.chat_meta import meta_from_chat
+from userbot.modules.capture.context import CaptureContext
+from userbot.modules.capture.message import capture_message
+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
+
+
+async def ingest_message(client: Client, ctx: CaptureContext, message: Message) -> None:
+    if message.chat is None:
+        return
+    meta = meta_from_chat(message.chat, ctx.contacts.ids)
+    await ctx.watches.on_message(seen_message(meta.chat_id, message))
+    toggles = ctx.resolve(meta)
+    if not toggles.messages:
+        return
+    await capture_message(client, message, ctx, toggles)
+    await notify_bg_event(
+        ctx.pool, "message", ctx.account_id, chat_id=meta.chat_id, message_id=message.id
+    )
+    if (
+        toggles.stt
+        and is_transcribable(message)
+        and (message.outgoing or message.unread_media is False)
+    ):
+        await safe_transcribe(client, ctx, meta.chat_id, message.id)

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

@@ -7,6 +7,7 @@ from userbot.modules.jobs.handlers import (
     fetch_media,
     fetch_read_info,
     search_peers,
+    send_saved,
     sync_contacts,
     sync_dialogs,
     transcribe,
@@ -21,6 +22,7 @@ __all__ = [
     "fetch_media",
     "fetch_read_info",
     "search_peers",
+    "send_saved",
     "sync_contacts",
     "sync_dialogs",
     "transcribe",

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

@@ -0,0 +1,32 @@
+from userbot.modules.jobs.context import JobContext
+from userbot.modules.jobs.registry import register
+from userbot.modules.saved import forward, send_new
+
+
+@register("send_saved")
+async def send_saved(ctx: JobContext) -> None:
+    client = ctx.client
+    capture = getattr(client, "capture", None)
+    if client is None or capture is None:
+        return
+    sent = await send_new(client, capture, ctx.job.params)
+    await ctx.report_progress(
+        {"chat_id": sent.chat.id if sent.chat else None, "message_id": sent.id}
+    )
+
+
+@register("forward_saved")
+async def forward_saved(ctx: JobContext) -> None:
+    client = ctx.client
+    capture = getattr(client, "capture", None)
+    if client is None or capture is None:
+        return
+    params = ctx.job.params
+    sent, mode = await forward(client, capture, params["chat_id"], params["message_id"])
+    await ctx.report_progress(
+        {
+            "chat_id": sent.chat.id if sent.chat else None,
+            "message_id": sent.id,
+            "mode": mode,
+        }
+    )

backend/src/userbot/modules/saved.py

@@ -0,0 +1,158 @@
+from io import BytesIO
+from typing import Any
+
+from pyrogram import Client, enums
+from pyrogram.errors import FloodPremiumWait, FloodWait, RPCError
+from pyrogram.types import Message, MessageEntity
+
+from userbot.modules.capture.context import CaptureContext
+from userbot.modules.capture.ingest import ingest_message
+from utils.files import media_file_name
+from utils.logging import logger
+from utils.read import chats, media
+from utils.read.models import EntityView, MessageView
+
+SAVED = "me"
+PLAIN = enums.ParseMode.DISABLED
+PHOTO_MIMES = ("image/jpeg", "image/png", "image/webp")
+
+
+class SavedError(Exception):
+    pass
+
+
+def _file(data: bytes, name: str) -> BytesIO:
+    file = BytesIO(data)
+    file.name = name
+    return file
+
+
+def _entity(view: EntityView) -> MessageEntity | None:
+    kind = enums.MessageEntityType.__members__.get(view.type.upper())
+    if kind is None or kind is enums.MessageEntityType.TEXT_MENTION:
+        return None
+    return MessageEntity(
+        type=kind,
+        offset=view.offset,
+        length=view.length,
+        url=view.url,
+        language=view.language,
+        custom_emoji_id=view.custom_emoji_id,
+    )
+
+
+def _entities(view: MessageView) -> list[MessageEntity] | None:
+    found = [entity for entity in map(_entity, view.entities) if entity is not None]
+    return found or None
+
+
+async def _send_file(  # noqa: PLR0911
+    client: Client,
+    kind: str,
+    file: BytesIO,
+    caption: str | None,
+    entities: list[MessageEntity] | None,
+) -> Message | None:
+    text: dict[str, Any] = {
+        "caption": caption or "",
+        "caption_entities": entities,
+        "parse_mode": PLAIN,
+    }
+    match kind:
+        case "photo":
+            return await client.send_photo(SAVED, file, **text)
+        case "video":
+            return await client.send_video(SAVED, file, **text)
+        case "voice":
+            return await client.send_voice(SAVED, file, **text)
+        case "audio":
+            return await client.send_audio(SAVED, file, **text)
+        case "animation":
+            return await client.send_animation(SAVED, file, **text)
+        case "video_note":
+            return await client.send_video_note(SAVED, file)
+        case "sticker":
+            return await client.send_sticker(SAVED, file)
+    return await client.send_document(SAVED, file, file_name=file.name, **text)
+
+
+def _upload_kind(mime: str | None, *, voice: bool) -> str:
+    if voice:
+        return "voice"
+    if mime in PHOTO_MIMES:
+        return "photo"
+    if mime and mime.startswith("video/"):
+        return "video"
+    return "document"
+
+
+async def send_new(
+    client: Client, ctx: CaptureContext, params: dict[str, Any]
+) -> Message:
+    text = params.get("text") or None
+    upload = params.get("upload")
+    if upload is None:
+        if not text:
+            msg = "nothing_to_send"
+            raise SavedError(msg)
+        sent = await client.send_message(SAVED, text, parse_mode=PLAIN)
+    else:
+        data = ctx.storage.staged(upload).read_bytes()
+        kind = _upload_kind(params.get("mime"), voice=bool(params.get("voice")))
+        sent = await _send_file(
+            client, kind, _file(data, params.get("file_name") or upload), text, None
+        )
+        ctx.storage.unstage(upload)
+    return await _keep(client, ctx, sent)
+
+
+async def _copy(
+    client: Client, ctx: CaptureContext, view: MessageView
+) -> Message | None:
+    entities = _entities(view)
+    item = None
+    if view.has_media and view.web_page is None:
+        item = await media.get_message_media(
+            ctx.pool, ctx.account_id, view.chat_id, view.message_id
+        )
+    if item is None:
+        if not view.text:
+            msg = "nothing_to_copy"
+            raise SavedError(msg)
+        return await client.send_message(
+            SAVED, view.text, entities=entities, parse_mode=PLAIN
+        )
+    key = item.storage_key
+    if not (item.downloaded and key and ctx.storage.exists(key)):
+        msg = "media_missing"
+        raise SavedError(msg)
+    name = media_file_name(item.kind, item.mime, view.message_id, item.file_name)
+    return await _send_file(
+        client, item.kind, _file(ctx.storage.get(key), name), view.text, entities
+    )
+
+
+async def forward(
+    client: Client, ctx: CaptureContext, chat_id: int, message_id: int
+) -> tuple[Message, str]:
+    view = await chats.get_message(ctx.pool, ctx.account_id, chat_id, message_id)
+    if view is None:
+        msg = "not_archived"
+        raise SavedError(msg)
+    if view.deleted_at is None and not view.is_self_destruct:
+        try:
+            sent = await client.forward_messages(SAVED, chat_id, message_id)
+            return await _keep(client, ctx, sent), "forwarded"
+        except (FloodWait, FloodPremiumWait):
+            raise
+        except RPCError as exc:
+            logger.info(f"[yellow]Forward {chat_id}/{message_id} refused: {exc}[/]")
+    return await _keep(client, ctx, await _copy(client, ctx, view)), "copied"
+
+
+async def _keep(client: Client, ctx: CaptureContext, sent: Message | None) -> Message:
+    if sent is None:
+        msg = "not_sent"
+        raise SavedError(msg)
+    await ingest_message(client, ctx, sent)
+    return sent

backend/src/utils/storage.py

@@ -1,7 +1,9 @@
 import hashlib
 from pathlib import Path
+from uuid import uuid4
 
 TRASH_DIR = ".trash"
+UPLOADS_DIR = ".uploads"
 
 
 class ContentAddressedStorage:
@@ -34,6 +36,19 @@ class ContentAddressedStorage:
     def url(self, key: str) -> str:
         return str(self._path(key))
 
+    def staged(self, name: str) -> Path:
+        return self._root / UPLOADS_DIR / Path(name).name
+
+    def stage(self, data: bytes) -> str:
+        name = uuid4().hex
+        path = self.staged(name)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+        return name
+
+    def unstage(self, name: str) -> None:
+        self.staged(name).unlink(missing_ok=True)
+
     def _trash_path(self, key: str) -> Path:
         return self._root / TRASH_DIR / key