feat(backend): read time of my messages from the archive, exact dates and group readers asked from telegram on demand

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,

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
 

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}

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",

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])

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
+    )

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(

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

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,
+    )