From d8dd08d7de863394a33e9f3b51e4680b9ccf015e Mon Sep 17 00:00:00 2001 From: h Date: Mon, 19 Oct 2026 02:26:58 +0000 Subject: [PATCH] feat(backend): read time of my messages from the archive, exact dates and group readers asked from telegram on demand --- .../versions/d7a3c5e9f2b1_message_reads.py | 51 +++++ backend/src/api/mcp/server.py | 18 ++ backend/src/api/realtime.py | 2 + backend/src/api/routers/social.py | 37 +++- .../userbot/modules/jobs/handlers/__init__.py | 2 + .../modules/jobs/handlers/fetch_read_info.py | 124 ++++++++++++ .../modules/read_receipts/repository.py | 52 ++++- backend/src/utils/read/chats.py | 18 +- backend/src/utils/read/models.py | 24 +++ backend/src/utils/read/read_receipts.py | 178 ++++++++++++++++++ 10 files changed, 497 insertions(+), 9 deletions(-) create mode 100644 backend/migrations/versions/d7a3c5e9f2b1_message_reads.py create mode 100644 backend/src/userbot/modules/jobs/handlers/fetch_read_info.py diff --git a/backend/migrations/versions/d7a3c5e9f2b1_message_reads.py b/backend/migrations/versions/d7a3c5e9f2b1_message_reads.py new file mode 100644 index 0000000..b6944a3 --- /dev/null +++ b/backend/migrations/versions/d7a3c5e9f2b1_message_reads.py @@ -0,0 +1,51 @@ +"""message reads + +Revision ID: d7a3c5e9f2b1 +Revises: f8c2d6e1a4b7 +Create Date: 2026-10-19 01:00:00.000000 + +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "d7a3c5e9f2b1" +down_revision: str | None = "f8c2d6e1a4b7" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_READS = """ +CREATE TABLE message_reads ( + account_id integer NOT NULL, + chat_id bigint NOT NULL, + message_id bigint NOT NULL, + reader_id bigint NOT NULL, + read_at timestamptz NOT NULL, + source text NOT NULL, + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (account_id, chat_id, message_id, reader_id) +) +""" + +_FETCHES = """ +CREATE TABLE read_fetches ( + account_id integer NOT NULL, + chat_id bigint NOT NULL, + message_id bigint NOT NULL, + outcome text NOT NULL, + fetched_at timestamptz NOT NULL DEFAULT now(), + retry_at timestamptz NOT NULL, + PRIMARY KEY (account_id, chat_id, message_id) +) +""" + + +def upgrade() -> None: + op.execute(_READS) + op.execute(_FETCHES) + + +def downgrade() -> None: + op.execute("DROP TABLE IF EXISTS read_fetches") + op.execute("DROP TABLE IF EXISTS message_reads") diff --git a/backend/src/api/mcp/server.py b/backend/src/api/mcp/server.py index 2ad9f7c..0b5cb13 100644 --- a/backend/src/api/mcp/server.py +++ b/backend/src/api/mcp/server.py @@ -23,6 +23,7 @@ from utils.files import resolve_mime from utils.read import annotations, chats, media, peers, presence, social, watches from utils.read.accounts import self_user_id, unknown_account_hint from utils.read.models import DEFAULT_LIMIT, Page +from utils.read.read_receipts import read_info from utils.search.models import SearchFilters from utils.search.repository import search_messages from utils.storage import ContentAddressedStorage @@ -295,6 +296,23 @@ async def get_callbacks( return _dump(items) +@mcp.tool +async def get_read_info( + account_id: int, chat_id: int, message_id: int +) -> dict[str, Any] | None: + """When the user's own message was read, from the archive only. + + `observed_at` is when the archive saw the chat read up to this message, + `exact_at` the read date Telegram reported for a private chat, `readers` + who read it in a group and when, `listened_at` when a voice or round + video was played. `outcome` is the last Telegram answer: ok, not_read_yet, + user_privacy (they hide read time), your_privacy (the user hides theirs), + chat_too_big, too_old, unavailable, flood or error. + """ + info = await read_info(await _pool(), account_id, chat_id, message_id) + return info.model_dump(mode="json") if info else None + + @mcp.tool async def presence_history( account_id: int, diff --git a/backend/src/api/realtime.py b/backend/src/api/realtime.py index 2b9803c..5a1001a 100644 --- a/backend/src/api/realtime.py +++ b/backend/src/api/realtime.py @@ -1,6 +1,7 @@ import asyncio import json import logging +from datetime import UTC, datetime from typing import Any import asyncpg @@ -114,6 +115,7 @@ class EventHub: "type": "receipt", "chat_id": event["chat_id"], "read_up_to": event["message_id"], + "read_at": datetime.now(UTC).isoformat(), } return None diff --git a/backend/src/api/routers/social.py b/backend/src/api/routers/social.py index 544a6f1..5bd128f 100644 --- a/backend/src/api/routers/social.py +++ b/backend/src/api/routers/social.py @@ -2,10 +2,13 @@ from typing import Annotated import asyncpg from dishka.integrations.fastapi import DishkaRoute, FromDishka -from fastapi import APIRouter, Query +from fastapi import APIRouter, HTTPException, Query +from pydantic import BaseModel +from utils.jobs import enqueue from utils.read import social -from utils.read.models import CallbackView, LinkView, ReactionView +from utils.read.models import CallbackView, LinkView, ReactionView, ReadInfoView +from utils.read.read_receipts import read_info router = APIRouter( prefix="/api/messages/{chat_id}/{message_id}", @@ -16,6 +19,10 @@ router = APIRouter( AccountId = Annotated[int, Query()] +class RefreshRequest(BaseModel): + account_id: int + + @router.get("/callbacks") async def message_callbacks( pool: FromDishka[asyncpg.Pool], chat_id: int, message_id: int, account_id: AccountId @@ -35,3 +42,29 @@ async def message_links( pool: FromDishka[asyncpg.Pool], chat_id: int, message_id: int, account_id: AccountId ) -> list[LinkView]: return await social.get_links(pool, account_id, chat_id, message_id) + + +@router.get("/read") +async def message_read( + pool: FromDishka[asyncpg.Pool], chat_id: int, message_id: int, account_id: AccountId +) -> ReadInfoView: + info = await read_info(pool, account_id, chat_id, message_id) + if info is None: + raise HTTPException(status_code=404, detail="message not found") + return info + + +@router.post("/read/refresh") +async def refresh_message_read( + pool: FromDishka[asyncpg.Pool], chat_id: int, message_id: int, body: RefreshRequest +) -> dict[str, int | None]: + info = await read_info(pool, body.account_id, chat_id, message_id) + if info is None or not info.can_fetch: + return {"job_id": None} + job_id = await enqueue( + pool, + body.account_id, + "fetch_read_info", + {"chat_id": chat_id, "message_id": message_id}, + ) + return {"job_id": job_id} diff --git a/backend/src/userbot/modules/jobs/handlers/__init__.py b/backend/src/userbot/modules/jobs/handlers/__init__.py index 05b1dc4..a3fe7ca 100644 --- a/backend/src/userbot/modules/jobs/handlers/__init__.py +++ b/backend/src/userbot/modules/jobs/handlers/__init__.py @@ -5,6 +5,7 @@ from userbot.modules.jobs.handlers import ( fetch_avatar, fetch_custom_emoji, fetch_media, + fetch_read_info, search_peers, sync_contacts, sync_dialogs, @@ -18,6 +19,7 @@ __all__ = [ "fetch_avatar", "fetch_custom_emoji", "fetch_media", + "fetch_read_info", "search_peers", "sync_contacts", "sync_dialogs", diff --git a/backend/src/userbot/modules/jobs/handlers/fetch_read_info.py b/backend/src/userbot/modules/jobs/handlers/fetch_read_info.py new file mode 100644 index 0000000..0f186be --- /dev/null +++ b/backend/src/userbot/modules/jobs/handlers/fetch_read_info.py @@ -0,0 +1,124 @@ +from datetime import UTC, datetime, timedelta + +from pyrogram import Client, raw +from pyrogram.errors import ( + ChannelInvalid, + ChannelPrivate, + ChatTooBig, + FloodPremiumWait, + FloodWait, + MessageIdInvalid, + MessageNotReadYet, + MessageTooOld, + MsgIdInvalid, + MsgTooOld, + PeerIdInvalid, + RPCError, + UserPrivacyRestricted, + YourPrivacyRestricted, +) + +from userbot.modules.jobs.context import JobContext +from userbot.modules.jobs.registry import register +from userbot.modules.read_receipts.repository import save_fetch, save_reads +from utils.jobs import enqueue +from utils.logging import logger +from utils.policy.models import ChatKind +from utils.read.peers import get_peers +from utils.read.read_receipts import read_info + +RETRY = timedelta(seconds=60) +LATER = timedelta(days=1) + +Refusal = tuple[tuple[type[Exception], ...], str, timedelta] + +REFUSALS: tuple[Refusal, ...] = ( + ((MessageNotReadYet,), "not_read_yet", RETRY), + ((UserPrivacyRestricted,), "user_privacy", RETRY), + ((YourPrivacyRestricted,), "your_privacy", LATER), + ((ChatTooBig,), "chat_too_big", LATER), + ((MsgTooOld, MessageTooOld), "too_old", LATER), + ( + ( + PeerIdInvalid, + ChannelInvalid, + ChannelPrivate, + MessageIdInvalid, + MsgIdInvalid, + KeyError, + ValueError, + ), + "unavailable", + RETRY, + ), +) + + +def _refusal(exc: Exception) -> tuple[str, timedelta]: + for kinds, outcome, retry in REFUSALS: + if isinstance(exc, kinds): + return outcome, retry + return "error", RETRY + + +def _at(timestamp: int) -> datetime: + return datetime.fromtimestamp(timestamp, UTC) + + +async def ask_telegram( + client: Client, chat_id: int, message_id: int, *, group: bool +) -> list[tuple[int, datetime]]: + peer = await client.resolve_peer(chat_id) + if peer is None: + msg = f"unknown peer {chat_id}" + raise ValueError(msg) + if group: + participants = await client.invoke( + raw.functions.messages.GetMessageReadParticipants( + peer=peer, msg_id=message_id + ) + ) + return [(item.user_id, _at(item.date)) for item in participants] + outbox = await client.invoke( + raw.functions.messages.GetOutboxReadDate(peer=peer, msg_id=message_id) + ) + return [(chat_id, _at(outbox.date))] + + +async def _enrich_unknown(ctx: JobContext, chat_id: int, reader_ids: list[int]) -> None: + known = { + peer.peer_id for peer in await get_peers(ctx.pool, ctx.account_id, reader_ids) + } + if set(reader_ids) - known: + await enqueue(ctx.pool, ctx.account_id, "enrich_chat", {"chat_id": chat_id}) + + +@register("fetch_read_info") +async def fetch_read_info(ctx: JobContext) -> None: + client = ctx.client + if client is None: + return + chat_id = ctx.job.params["chat_id"] + message_id = ctx.job.params["message_id"] + info = await read_info(ctx.pool, ctx.account_id, chat_id, message_id) + if info is None or not info.can_fetch: + return + group = info.chat_kind == ChatKind.GROUP + try: + reads = await ask_telegram(client, chat_id, message_id, group=group) + except (FloodWait, FloodPremiumWait) as exc: + wait = exc.value if isinstance(exc.value, int) else 0 + retry = max(RETRY, timedelta(seconds=wait)) + await save_fetch(ctx.pool, ctx.account_id, chat_id, message_id, "flood", retry) + return + except (RPCError, KeyError, ValueError) as exc: + outcome, retry = _refusal(exc) + if outcome == "error": + logger.warning(f"[yellow]Read info {chat_id}/{message_id}: {exc!r}[/]") + await save_fetch(ctx.pool, ctx.account_id, chat_id, message_id, outcome, retry) + return + source = "participants" if group else "outbox_date" + await save_reads(ctx.pool, ctx.account_id, chat_id, message_id, source, reads) + await save_fetch(ctx.pool, ctx.account_id, chat_id, message_id, "ok", RETRY) + if group and reads: + await _enrich_unknown(ctx, chat_id, [reader_id for reader_id, _ in reads]) diff --git a/backend/src/userbot/modules/read_receipts/repository.py b/backend/src/userbot/modules/read_receipts/repository.py index e03bb7b..aabff2c 100644 --- a/backend/src/userbot/modules/read_receipts/repository.py +++ b/backend/src/userbot/modules/read_receipts/repository.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import datetime, timedelta import asyncpg @@ -9,6 +9,26 @@ VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb) ON CONFLICT (account_id, chat_id, message_id, ts) DO NOTHING """ +_UPSERT_READ = """ +INSERT INTO message_reads + (account_id, chat_id, message_id, reader_id, read_at, source, fetched_at) +VALUES ($1, $2, $3, $4, $5, $6, now()) +ON CONFLICT (account_id, chat_id, message_id, reader_id) DO UPDATE SET + read_at = EXCLUDED.read_at, + source = EXCLUDED.source, + fetched_at = EXCLUDED.fetched_at +""" + +_UPSERT_FETCH = """ +INSERT INTO read_fetches + (account_id, chat_id, message_id, outcome, fetched_at, retry_at) +VALUES ($1, $2, $3, $4, now(), now() + $5::interval) +ON CONFLICT (account_id, chat_id, message_id) DO UPDATE SET + outcome = EXCLUDED.outcome, + fetched_at = EXCLUDED.fetched_at, + retry_at = EXCLUDED.retry_at +""" + async def insert_read( # noqa: PLR0913 pool: asyncpg.Pool, @@ -34,3 +54,33 @@ async def insert_listened( # noqa: PLR0913 await pool.execute( _INSERT, account_id, chat_id, message_id, ts, reader_id, "listened", raw ) + + +async def save_reads( # noqa: PLR0913 + pool: asyncpg.Pool, + account_id: int, + chat_id: int, + message_id: int, + source: str, + reads: list[tuple[int, datetime]], +) -> None: + await pool.executemany( + _UPSERT_READ, + [ + (account_id, chat_id, message_id, reader_id, read_at, source) + for reader_id, read_at in reads + ], + ) + + +async def save_fetch( # noqa: PLR0913 + pool: asyncpg.Pool, + account_id: int, + chat_id: int, + message_id: int, + outcome: str, + retry_after: timedelta, +) -> None: + await pool.execute( + _UPSERT_FETCH, account_id, chat_id, message_id, outcome, retry_after + ) diff --git a/backend/src/utils/read/chats.py b/backend/src/utils/read/chats.py index dca0d53..289a758 100644 --- a/backend/src/utils/read/chats.py +++ b/backend/src/utils/read/chats.py @@ -10,7 +10,7 @@ from utils.read.models import ( MessageView, Page, ) -from utils.read.read_receipts import read_up_to +from utils.read.read_receipts import read_times, read_up_to _MESSAGE_COLS = ( "chat_id, message_id, date, sender_id, text, has_media, is_self_destruct, " @@ -285,12 +285,18 @@ async def _apply_read_status( self_id = await self_user_id(pool, account_id) if self_id is None: return - marker = await read_up_to(pool, account_id, chat_id) - if marker is None: + own = [view for view in views if view.sender_id == self_id] + if not own: return - for view in views: - if view.sender_id == self_id and view.message_id <= marker: - view.read = True + marker = await read_up_to(pool, account_id, chat_id) + times = await read_times( + pool, account_id, chat_id, [(view.message_id, view.date) for view in own] + ) + for view in own: + view.read_at = times.get(view.message_id) + view.read = view.read_at is not None or ( + marker is not None and view.message_id <= marker + ) def _build_album( diff --git a/backend/src/utils/read/models.py b/backend/src/utils/read/models.py index 8cc6f50..b62cbcb 100644 --- a/backend/src/utils/read/models.py +++ b/backend/src/utils/read/models.py @@ -203,6 +203,7 @@ class MessageView(BaseModel): is_sticker: bool = False is_animated_emoji: bool = False read: bool = False + read_at: datetime | None = None class MessageVersionView(BaseModel): @@ -387,6 +388,29 @@ class PeerView(BaseModel): updated_at: datetime +class ReaderView(BaseModel): + peer_id: int + read_at: datetime + peer: PeerView | None + + +class ReadInfoView(BaseModel): + chat_id: int + message_id: int + chat_kind: str + own: bool + voice: bool + observed_at: datetime | None + listened_at: datetime | None + exact_at: datetime | None + readers: list[ReaderView] + read: bool + unread: bool + outcome: str | None + fetched_at: datetime | None + can_fetch: bool + + class PeerHistoryView(BaseModel): observed_at: datetime first_name: str | None diff --git a/backend/src/utils/read/read_receipts.py b/backend/src/utils/read/read_receipts.py index ae9fd38..77e1b7c 100644 --- a/backend/src/utils/read/read_receipts.py +++ b/backend/src/utils/read/read_receipts.py @@ -1,5 +1,89 @@ +from datetime import UTC, datetime, timedelta + import asyncpg +from utils.policy.models import ChatKind +from utils.read.discover import chat_kind +from utils.read.models import ReaderView, ReadInfoView +from utils.read.peers import get_peers + +FETCH_WINDOW = timedelta(days=7) +CLOCK_SKEW = timedelta(minutes=1) +VOICE_KINDS = ["voice", "video_note"] + +_READ_TIMES = """ +SELECT m.id, + COALESCE( + (SELECT min(mr.read_at) FROM message_reads mr + WHERE mr.account_id = $1 AND mr.chat_id = $2 AND mr.message_id = m.id), + CASE WHEN m.date >= $5 THEN ( + SELECT r.ts FROM read_receipts r + WHERE r.account_id = $1 AND r.chat_id = $2 AND r.kind = 'read' + AND r.message_id >= m.id AND r.ts >= m.date - $6::interval + ORDER BY r.ts LIMIT 1 + ) END + ) AS read_at +FROM unnest($3::bigint[], $4::timestamptz[]) AS m(id, date) +""" + +_MESSAGE = """ +SELECT m.sender_id, m.date, a.tg_user_id, a.is_bot, + EXISTS ( + SELECT 1 FROM media md + WHERE md.account_id = m.account_id AND md.chat_id = m.chat_id + AND md.message_id = m.message_id AND md.kind = ANY($4::text[]) + ) AS voice +FROM messages m +JOIN accounts a ON a.account_id = m.account_id +WHERE m.account_id = $1 AND m.chat_id = $2 AND m.message_id = $3 +LIMIT 1 +""" + +_RECEIPTS = """ +SELECT + (SELECT ts FROM read_receipts + WHERE account_id = $1 AND chat_id = $2 AND kind = 'read' + AND message_id >= $3 AND ts >= $4 + ORDER BY ts LIMIT 1) AS observed_at, + (SELECT min(ts) FROM read_receipts + WHERE account_id = $1 AND chat_id = $2 AND kind = 'listened' + AND message_id = $3) AS listened_at, + (SELECT max(message_id) FROM read_receipts + WHERE account_id = $1 AND chat_id = $2 AND kind = 'read') AS watermark +""" + +_READS = """ +SELECT reader_id, read_at, source FROM message_reads +WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 +ORDER BY read_at +""" + +_FETCH = """ +SELECT outcome, fetched_at, retry_at FROM read_fetches +WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 +""" + +_INHERITED = """ +SELECT outcome FROM read_fetches +WHERE account_id = $1 AND retry_at > now() + AND ((chat_id = $2 AND outcome = 'chat_too_big') OR outcome = 'your_privacy') +LIMIT 1 +""" + +_capture_starts: dict[int, datetime] = {} + + +async def capture_start(pool: asyncpg.Pool, account_id: int) -> datetime | None: + cached = _capture_starts.get(account_id) + if cached is not None: + return cached + start = await pool.fetchval( + "SELECT min(ts) FROM read_receipts WHERE account_id = $1", account_id + ) + if start is not None: + _capture_starts[account_id] = start + return start + async def read_up_to(pool: asyncpg.Pool, account_id: int, chat_id: int) -> int | None: return await pool.fetchval( @@ -8,3 +92,97 @@ async def read_up_to(pool: asyncpg.Pool, account_id: int, chat_id: int) -> int | account_id, chat_id, ) + + +async def read_times( + pool: asyncpg.Pool, + account_id: int, + chat_id: int, + messages: list[tuple[int, datetime]], +) -> dict[int, datetime]: + if not messages: + return {} + start = await capture_start(pool, account_id) or datetime.max.replace(tzinfo=UTC) + rows = await pool.fetch( + _READ_TIMES, + account_id, + chat_id, + [message_id for message_id, _ in messages], + [date for _, date in messages], + start, + CLOCK_SKEW, + ) + return {row["id"]: row["read_at"] for row in rows if row["read_at"] is not None} + + +async def read_info( + pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int +) -> ReadInfoView | None: + message = await pool.fetchrow( + _MESSAGE, account_id, chat_id, message_id, VOICE_KINDS + ) + if message is None: + return None + kind = await chat_kind(pool, account_id, chat_id) + date: datetime = message["date"] + receipts = await pool.fetchrow( + _RECEIPTS, account_id, chat_id, message_id, date - CLOCK_SKEW + ) + reads = await pool.fetch(_READS, account_id, chat_id, message_id) + fetch = await pool.fetchrow(_FETCH, account_id, chat_id, message_id) + inherited = await pool.fetchval(_INHERITED, account_id, chat_id) + start = await capture_start(pool, account_id) + + observed_at = receipts["observed_at"] if start and date >= start else None + exact_at = next( + (row["read_at"] for row in reads if row["source"] == "outbox_date"), None + ) + reader_rows = [row for row in reads if row["source"] == "participants"] + known = { + peer.peer_id: peer + for peer in await get_peers( + pool, account_id, [row["reader_id"] for row in reader_rows] + ) + } + readers = [ + ReaderView( + peer_id=row["reader_id"], + read_at=row["read_at"], + peer=known.get(row["reader_id"]), + ) + for row in reader_rows + ] + outcome = fetch["outcome"] if fetch else inherited + + now = datetime.now(UTC) + own = ( + message["sender_id"] is not None + and message["sender_id"] == (message["tg_user_id"]) + ) + applicable = own and kind != ChatKind.CHANNEL and chat_id != message["tg_user_id"] + watermark = receipts["watermark"] + covered = watermark is not None and message_id <= watermark + seen = covered or any((exact_at, readers, receipts["listened_at"])) + evidence = watermark is not None or outcome in {"ok", "not_read_yet"} + due = fetch is None or fetch["retry_at"] <= now + return ReadInfoView( + chat_id=chat_id, + message_id=message_id, + chat_kind=kind.value, + own=own, + voice=message["voice"], + observed_at=observed_at, + listened_at=receipts["listened_at"], + exact_at=exact_at, + readers=readers, + read=applicable and seen, + unread=applicable and not seen and evidence, + outcome=outcome, + fetched_at=fetch["fetched_at"] if fetch else None, + can_fetch=applicable + and not message["is_bot"] + and now - date < FETCH_WINDOW + and exact_at is None + and inherited is None + and due, + )