diff --git a/backend/migrations/versions/b3d5f7a9c2e4_feed.py b/backend/migrations/versions/b3d5f7a9c2e4_feed.py new file mode 100644 index 0000000..2cc903e --- /dev/null +++ b/backend/migrations/versions/b3d5f7a9c2e4_feed.py @@ -0,0 +1,198 @@ +"""feed of deleted and edited messages + +Revision ID: b3d5f7a9c2e4 +Revises: a2c4e6f8b1d3 +Create Date: 2026-10-19 02:40:00.000000 + +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "b3d5f7a9c2e4" +down_revision: str | None = "a2c4e6f8b1d3" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_ITEMS = """ +CREATE TABLE feed_items ( + id bigserial PRIMARY KEY, + account_id integer NOT NULL, + kind text NOT NULL, + chat_id bigint NOT NULL, + chat_kind text NOT NULL, + message_id bigint NOT NULL, + message_date timestamptz, + at timestamptz NOT NULL, + since timestamptz NOT NULL, + message_count integer NOT NULL DEFAULT 1 +) +""" + +_SEEN = """ +CREATE TABLE feed_seen ( + account_id integer PRIMARY KEY, + seen_at timestamptz NOT NULL +) +""" + +_CHAT_KIND = """ +CREATE FUNCTION feed_chat_kind(chat_id bigint, chat_type text) +RETURNS text LANGUAGE sql IMMUTABLE AS $$ + SELECT CASE + WHEN chat_id > 0 THEN 'dm' + WHEN chat_type = 'ChatType.CHANNEL' THEN 'channel' + ELSE 'group' END +$$ +""" + +_MESSAGE_DATE = """ +CREATE FUNCTION feed_message_date(stamp text) +RETURNS timestamptz LANGUAGE sql STABLE AS $$ + SELECT CASE + WHEN stamp ~ '^\\d{4}-\\d{2}-\\d{2}[ T]\\d{2}:\\d{2}:\\d{2}(\\.\\d+)?$' + THEN stamp::timestamp AT TIME ZONE 'UTC' END +$$ +""" + +_ON_DELETE = """ +CREATE FUNCTION feed_on_delete() RETURNS trigger LANGUAGE plpgsql AS $$ +DECLARE + item feed_items%ROWTYPE; +BEGIN + SELECT * INTO item FROM feed_items + WHERE account_id = NEW.account_id AND chat_id = NEW.chat_id + AND kind = 'deleted' + ORDER BY at DESC, id DESC LIMIT 1; + IF FOUND AND NEW.deleted_at BETWEEN item.since - interval '5 seconds' + AND item.at + interval '5 seconds' THEN + IF (NEW.date, NEW.message_id) > (item.message_date, item.message_id) THEN + item.message_id := NEW.message_id; + item.message_date := NEW.date; + END IF; + UPDATE feed_items SET + message_count = message_count + 1, + at = greatest(at, NEW.deleted_at), + since = least(since, NEW.deleted_at), + message_id = item.message_id, + message_date = item.message_date + WHERE id = item.id; + ELSE + INSERT INTO feed_items + (account_id, kind, chat_id, chat_kind, message_id, message_date, + at, since) + VALUES (NEW.account_id, 'deleted', NEW.chat_id, + feed_chat_kind(NEW.chat_id, NEW.raw->'chat'->>'type'), + NEW.message_id, NEW.date, NEW.deleted_at, NEW.deleted_at); + END IF; + RETURN NULL; +END; +$$ +""" + +_ON_VERSION = """ +CREATE FUNCTION feed_on_version() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.edit_date IS NULL AND NOT EXISTS ( + SELECT 1 FROM message_versions v + WHERE v.account_id = NEW.account_id AND v.chat_id = NEW.chat_id + AND v.message_id = NEW.message_id AND v.observed_at < NEW.observed_at + ) THEN + RETURN NULL; + END IF; + INSERT INTO feed_items + (account_id, kind, chat_id, chat_kind, message_id, message_date, at, since) + VALUES (NEW.account_id, 'edited', NEW.chat_id, + feed_chat_kind(NEW.chat_id, NEW.raw->'chat'->>'type'), + NEW.message_id, feed_message_date(NEW.raw->>'date'), + NEW.observed_at, NEW.observed_at); + RETURN NULL; +END; +$$ +""" + +_BACKFILL = """ +WITH d AS ( + SELECT account_id, chat_id, message_id, date, deleted_at, + raw->'chat'->>'type' AS chat_type, + deleted_at - lag(deleted_at) OVER ( + PARTITION BY account_id, chat_id ORDER BY deleted_at, message_id + ) AS gap + FROM messages WHERE deleted_at IS NOT NULL +), b AS ( + SELECT *, count(*) FILTER (WHERE gap IS NULL OR gap > interval '5 seconds') + OVER (PARTITION BY account_id, chat_id ORDER BY deleted_at, message_id) + AS burst + FROM d +), deleted AS ( + SELECT account_id, 'deleted' AS kind, chat_id, + feed_chat_kind(chat_id, + (array_agg(chat_type ORDER BY date DESC, message_id DESC))[1]) + AS chat_kind, + (array_agg(message_id ORDER BY date DESC, message_id DESC))[1] + AS message_id, + max(date) AS message_date, max(deleted_at) AS at, + min(deleted_at) AS since, count(*)::integer AS message_count + FROM b GROUP BY account_id, chat_id, burst +), edited AS ( + SELECT v.account_id, 'edited' AS kind, v.chat_id, + feed_chat_kind(v.chat_id, v.raw->'chat'->>'type') AS chat_kind, + v.message_id, feed_message_date(v.raw->>'date') AS message_date, + v.observed_at AS at, v.observed_at AS since, 1 AS message_count + FROM message_versions v + WHERE v.edit_date IS NOT NULL OR EXISTS ( + SELECT 1 FROM message_versions p + WHERE p.account_id = v.account_id AND p.chat_id = v.chat_id + AND p.message_id = v.message_id AND p.observed_at < v.observed_at + ) +) +INSERT INTO feed_items + (account_id, kind, chat_id, chat_kind, message_id, message_date, at, since, + message_count) +SELECT * FROM (SELECT * FROM deleted UNION ALL SELECT * FROM edited) items +ORDER BY at, chat_id, message_id +""" + + +def upgrade() -> None: + op.execute("SET LOCAL jit = off") + op.execute(_ITEMS) + op.execute(_SEEN) + op.execute(_CHAT_KIND) + op.execute(_MESSAGE_DATE) + op.execute(_ON_DELETE) + op.execute(_ON_VERSION) + op.execute( + "CREATE TRIGGER messages_feed_deleted AFTER UPDATE OF deleted_at " + "ON messages FOR EACH ROW " + "WHEN (OLD.deleted_at IS NULL AND NEW.deleted_at IS NOT NULL) " + "EXECUTE FUNCTION feed_on_delete()" + ) + op.execute( + "CREATE TRIGGER message_versions_feed AFTER INSERT ON message_versions " + "FOR EACH ROW EXECUTE FUNCTION feed_on_version()" + ) + op.execute(_BACKFILL) + op.execute( + "CREATE INDEX ix_feed_items_recent ON feed_items (account_id, at DESC, id DESC)" + ) + op.execute( + "CREATE INDEX ix_feed_items_chat ON feed_items " + "(account_id, chat_id, at DESC, id DESC)" + ) + op.execute( + "INSERT INTO feed_seen (account_id, seen_at) SELECT account_id, now() " + "FROM accounts" + ) + + +def downgrade() -> None: + op.execute("DROP TRIGGER IF EXISTS message_versions_feed ON message_versions") + op.execute("DROP TRIGGER IF EXISTS messages_feed_deleted ON messages") + op.execute("DROP FUNCTION IF EXISTS feed_on_version()") + op.execute("DROP FUNCTION IF EXISTS feed_on_delete()") + op.execute("DROP FUNCTION IF EXISTS feed_message_date(text)") + op.execute("DROP FUNCTION IF EXISTS feed_chat_kind(bigint, text)") + op.execute("DROP TABLE IF EXISTS feed_seen") + op.execute("DROP TABLE IF EXISTS feed_items") diff --git a/backend/src/api/app.py b/backend/src/api/app.py index 7d34589..96e84ba 100644 --- a/backend/src/api/app.py +++ b/backend/src/api/app.py @@ -25,6 +25,7 @@ from api.routers import ( custom_emoji, discover, events, + feed, files, folders, media, @@ -96,6 +97,7 @@ app.include_router(folders.router) app.include_router(backfill.router) app.include_router(search.router) app.include_router(chats.router) +app.include_router(feed.router) app.include_router(media.router) app.include_router(avatars.router) app.include_router(custom_emoji.router) diff --git a/backend/src/api/routers/feed.py b/backend/src/api/routers/feed.py new file mode 100644 index 0000000..c1e514e --- /dev/null +++ b/backend/src/api/routers/feed.py @@ -0,0 +1,70 @@ +from datetime import datetime +from typing import Annotated + +import asyncpg +from dishka.integrations.fastapi import DishkaRoute, FromDishka +from fastapi import APIRouter, HTTPException, Query +from pydantic import BaseModel + +from utils.read import feed +from utils.read.feed import CURSOR_PATTERN, FeedChatKind, FeedFilter, FeedKind +from utils.read.models import DEFAULT_LIMIT, FeedMessagesPage, FeedPage, FeedUnseen + +router = APIRouter(prefix="/api/feed", tags=["feed"], route_class=DishkaRoute) + +AccountId = Annotated[int, Query()] +Limit = Annotated[int, Query(ge=1)] +Cursor = Annotated[str | None, Query(pattern=CURSOR_PATTERN)] + + +class SeenRequest(BaseModel): + account_id: int + at: datetime | None = None + + +@router.get("") +async def list_feed( + pool: FromDishka[asyncpg.Pool], + account_id: AccountId, + kind: Annotated[FeedKind, Query()] = FeedKind.ALL, + chat_id: Annotated[int | None, Query()] = None, + chat_kind: Annotated[FeedChatKind | None, Query()] = None, + since: Annotated[datetime | None, Query()] = None, + cursor: Cursor = None, + limit: Limit = DEFAULT_LIMIT, +) -> FeedPage: + return await feed.list_feed( + pool, + account_id, + FeedFilter(kind=kind, chat_id=chat_id, chat_kind=chat_kind, since=since), + limit=limit, + cursor=cursor, + ) + + +@router.get("/unseen") +async def feed_unseen( + pool: FromDishka[asyncpg.Pool], account_id: AccountId +) -> FeedUnseen: + return await feed.unseen(pool, account_id) + + +@router.post("/seen") +async def feed_seen(pool: FromDishka[asyncpg.Pool], body: SeenRequest) -> FeedUnseen: + return await feed.mark_seen(pool, body.account_id, body.at) + + +@router.get("/{item_id}/messages") +async def feed_item_messages( + pool: FromDishka[asyncpg.Pool], + item_id: int, + account_id: AccountId, + cursor: Cursor = None, + limit: Limit = DEFAULT_LIMIT, +) -> FeedMessagesPage: + page = await feed.item_messages( + pool, account_id, item_id, limit=limit, cursor=cursor + ) + if page is None: + raise HTTPException(status_code=404, detail="Feed item not found") + return page diff --git a/backend/src/utils/read/chats.py b/backend/src/utils/read/chats.py index 289a758..b27de98 100644 --- a/backend/src/utils/read/chats.py +++ b/backend/src/utils/read/chats.py @@ -1,3 +1,5 @@ +from datetime import datetime + import asyncpg from utils.policy.models import FolderSpec @@ -331,6 +333,12 @@ async def get_deleted_messages( f"ORDER BY deleted_at DESC LIMIT ${len(params) - 1} OFFSET ${len(params)}", *params, ) + return await _single_views(pool, account_id, rows) + + +async def _single_views( + pool: asyncpg.Pool, account_id: int, rows: list[asyncpg.Record] +) -> list[MessageView]: media_by_key = await _media_map(pool, account_id, rows) views: list[MessageView] = [] for row in rows: @@ -341,6 +349,67 @@ async def get_deleted_messages( return views +async def _rows_at( + pool: asyncpg.Pool, account_id: int, keys: list[tuple[int, int, datetime]] +) -> list[asyncpg.Record]: + if not keys: + return [] + params: list[object] = [account_id] + clauses: list[str] = [] + for key in keys: + params.extend(key) + n = len(params) + clauses.append( + f"(chat_id = ${n - 2} AND message_id = ${n - 1} AND date = ${n})" + ) + return await pool.fetch( + f"SELECT {_MESSAGE_COLS} FROM messages " # noqa: S608 + f"WHERE account_id = $1 AND ({' OR '.join(clauses)})", + *params, + ) + + +async def get_messages_at( + pool: asyncpg.Pool, account_id: int, keys: list[tuple[int, int, datetime | None]] +) -> dict[tuple[int, int], MessageView]: + dated = list( + dict.fromkeys((chat, message, date) for chat, message, date in keys if date) + ) + rows = await _rows_at(pool, account_id, dated) + views = { + (view.chat_id, view.message_id): view + for view in await _single_views(pool, account_id, rows) + } + for chat_id, message_id, _ in keys: + if (chat_id, message_id) not in views: + view = await get_message(pool, account_id, chat_id, message_id) + if view is not None: + views[chat_id, message_id] = view + return views + + +async def get_deleted_between( # noqa: PLR0913 + pool: asyncpg.Pool, + account_id: int, + chat_id: int, + since: datetime, + until: datetime, + limit: int, + before: tuple[datetime, int] | None = None, +) -> list[MessageView]: + params: list[object] = [account_id, chat_id, since, until, limit] + where = "account_id = $1 AND chat_id = $2 AND deleted_at BETWEEN $3 AND $4" + if before is not None: + params.extend(before) + where += " AND (date, message_id) < ($6, $7)" + rows = await pool.fetch( + f"SELECT {_MESSAGE_COLS} FROM messages WHERE {where} " # noqa: S608 + "ORDER BY date DESC, message_id DESC LIMIT $5", + *params, + ) + return await _single_views(pool, account_id, rows) + + async def get_message_versions( pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int ) -> list[MessageVersionView]: diff --git a/backend/src/utils/read/feed.py b/backend/src/utils/read/feed.py new file mode 100644 index 0000000..cfd9bfe --- /dev/null +++ b/backend/src/utils/read/feed.py @@ -0,0 +1,276 @@ +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from enum import StrEnum + +import asyncpg + +from utils.read import chats, peers +from utils.read.accounts import self_user_id +from utils.read.models import ( + DEFAULT_LIMIT, + MAX_LIMIT, + ChatListItem, + FeedChat, + FeedItem, + FeedMessagesPage, + FeedPage, + FeedUnseen, + MessageView, + PeerView, +) + +CURSOR_PATTERN = r"^\d+\.\d+$" +_EPOCH = datetime(1970, 1, 1, tzinfo=UTC) +_MICROSECOND = timedelta(microseconds=1) + +_ITEM_COLS = ( + "id, kind, chat_id, chat_kind, message_id, message_date, at, since, message_count" +) + +_VERSIONS = """ +SELECT v.chat_id, v.message_id, v.observed_at, v.edit_date, v.text, + p.text AS before, p.observed_at IS NOT NULL AS has_before +FROM unnest($2::bigint[], $3::bigint[], $4::timestamptz[]) + AS k(chat_id, message_id, observed_at) +JOIN message_versions v + ON v.account_id = $1 AND v.chat_id = k.chat_id + AND v.message_id = k.message_id AND v.observed_at = k.observed_at +LEFT JOIN LATERAL ( + SELECT text, observed_at FROM message_versions p + WHERE p.account_id = $1 AND p.chat_id = v.chat_id + AND p.message_id = v.message_id AND p.observed_at < v.observed_at + ORDER BY p.observed_at DESC LIMIT 1 +) p ON true +""" + +_UNSEEN = """ +SELECT count(*) FILTER (WHERE kind = 'deleted') AS deleted, + count(*) FILTER (WHERE kind = 'edited') AS edited +FROM feed_items +WHERE account_id = $1 AND at > coalesce($2, '-infinity'::timestamptz) +""" + +_MARK_SEEN = """ +INSERT INTO feed_seen (account_id, seen_at) VALUES ($1, coalesce($2, now())) +ON CONFLICT (account_id) DO UPDATE +SET seen_at = greatest(feed_seen.seen_at, EXCLUDED.seen_at) +""" + + +class FeedKind(StrEnum): + ALL = "all" + DELETED = "deleted" + EDITED = "edited" + + +class FeedChatKind(StrEnum): + DM = "dm" + GROUP = "group" + CHANNEL = "channel" + + +@dataclass(slots=True) +class FeedFilter: + kind: FeedKind = FeedKind.ALL + chat_id: int | None = None + chat_kind: FeedChatKind | None = None + since: datetime | None = None + + +def encode_cursor(at: datetime, key: int) -> str: + return f"{(at - _EPOCH) // _MICROSECOND}.{key}" + + +def decode_cursor(cursor: str) -> tuple[datetime, int]: + micros, _, key = cursor.partition(".") + return _EPOCH + int(micros) * _MICROSECOND, int(key) + + +def _where( + account_id: int, feed: FeedFilter, cursor: str | None +) -> tuple[str, list[object]]: + params: list[object] = [account_id] + clauses = ["account_id = $1"] + conditions = ( + ("kind = {}", None if feed.kind is FeedKind.ALL else feed.kind.value), + ("chat_id = {}", feed.chat_id), + ("chat_kind = {}", feed.chat_kind.value if feed.chat_kind else None), + ("at >= {}", feed.since), + ) + for template, value in conditions: + if value is not None: + params.append(value) + clauses.append(template.format(f"${len(params)}")) + if cursor is not None: + params.extend(decode_cursor(cursor)) + clauses.append(f"(at, id) < (${len(params) - 1}, ${len(params)})") + return " AND ".join(clauses), params + + +def _peer_name(peer: PeerView) -> str | None: + name = " ".join(part for part in (peer.first_name, peer.last_name) if part) + return name or peer.username + + +async def _versions( + pool: asyncpg.Pool, account_id: int, rows: list[asyncpg.Record] +) -> dict[tuple[int, int, datetime], asyncpg.Record]: + edited = [row for row in rows if row["kind"] == FeedKind.EDITED] + if not edited: + return {} + found = await pool.fetch( + _VERSIONS, + account_id, + [row["chat_id"] for row in edited], + [row["message_id"] for row in edited], + [row["at"] for row in edited], + ) + return { + (row["chat_id"], row["message_id"], row["observed_at"]): row for row in found + } + + +async def _sender_names( + pool: asyncpg.Pool, + account_id: int, + messages: dict[tuple[int, int], MessageView], + chat_views: dict[int, ChatListItem], +) -> dict[int, str | None]: + ids = {m.sender_id for m in messages.values() if m.sender_id is not None} + found = await peers.get_peers(pool, account_id, list(ids)) + names: dict[int, str | None] = { + chat_id: chat.title for chat_id, chat in chat_views.items() + } + names.update({peer.peer_id: _peer_name(peer) for peer in found}) + return names + + +async def _build_items( + pool: asyncpg.Pool, account_id: int, rows: list[asyncpg.Record] +) -> list[FeedItem]: + if not rows: + return [] + messages = await chats.get_messages_at( + pool, + account_id, + [(row["chat_id"], row["message_id"], row["message_date"]) for row in rows], + ) + versions = await _versions(pool, account_id, rows) + chat_views = { + chat.chat_id: chat + for chat in await chats.get_chats( + pool, account_id, list({row["chat_id"] for row in rows}) + ) + } + names = await _sender_names(pool, account_id, messages, chat_views) + self_id = await self_user_id(pool, account_id) + items: list[FeedItem] = [] + for row in rows: + message = messages.get((row["chat_id"], row["message_id"])) + chat = chat_views.get(row["chat_id"]) + version = versions.get((row["chat_id"], row["message_id"], row["at"])) + sender_id = message.sender_id if message else None + items.append( + FeedItem( + id=row["id"], + kind=row["kind"], + at=row["at"], + since=row["since"], + message_count=row["message_count"], + chat=FeedChat( + chat_id=row["chat_id"], + title=chat.title if chat else None, + kind=row["chat_kind"], + has_avatar=chat.has_avatar if chat else False, + ), + message=message, + sender_id=sender_id, + sender_name=names.get(sender_id) if sender_id is not None else None, + outgoing=self_id is not None and sender_id == self_id, + before=version["before"] if version else None, + after=version["text"] if version else None, + has_before=bool(version and version["has_before"]), + edit_date=version["edit_date"] if version else None, + ) + ) + return items + + +async def list_feed( + pool: asyncpg.Pool, + account_id: int, + feed: FeedFilter, + *, + limit: int = DEFAULT_LIMIT, + cursor: str | None = None, +) -> FeedPage: + size = max(1, min(limit, MAX_LIMIT)) + where, params = _where(account_id, feed, cursor) + params.append(size + 1) + rows = await pool.fetch( + f"SELECT {_ITEM_COLS} FROM feed_items WHERE {where} " # noqa: S608 + f"ORDER BY at DESC, id DESC LIMIT ${len(params)}", + *params, + ) + page = rows[:size] + more = len(rows) > size + return FeedPage( + items=await _build_items(pool, account_id, page), + next_cursor=encode_cursor(page[-1]["at"], page[-1]["id"]) if more else None, + ) + + +async def unseen(pool: asyncpg.Pool, account_id: int) -> FeedUnseen: + seen_at = await pool.fetchval( + "SELECT seen_at FROM feed_seen WHERE account_id = $1", account_id + ) + counts = await pool.fetchrow(_UNSEEN, account_id, seen_at) + latest = await list_feed(pool, account_id, FeedFilter(), limit=1) + return FeedUnseen( + count=counts["deleted"] + counts["edited"], + deleted=counts["deleted"], + edited=counts["edited"], + seen_at=seen_at, + latest=latest.items[0] if latest.items else None, + ) + + +async def mark_seen( + pool: asyncpg.Pool, account_id: int, at: datetime | None +) -> FeedUnseen: + await pool.execute(_MARK_SEEN, account_id, at) + return await unseen(pool, account_id) + + +async def item_messages( + pool: asyncpg.Pool, + account_id: int, + item_id: int, + *, + limit: int = DEFAULT_LIMIT, + cursor: str | None = None, +) -> FeedMessagesPage | None: + row = await pool.fetchrow( + "SELECT chat_id, since, at FROM feed_items " + "WHERE id = $1 AND account_id = $2 AND kind = 'deleted'", + item_id, + account_id, + ) + if row is None: + return None + size = max(1, min(limit, MAX_LIMIT)) + views = await chats.get_deleted_between( + pool, + account_id, + row["chat_id"], + row["since"], + row["at"], + size + 1, + decode_cursor(cursor) if cursor else None, + ) + page = views[:size] + more = len(views) > size + return FeedMessagesPage( + messages=page, + next_cursor=encode_cursor(page[-1].date, page[-1].message_id) if more else None, + ) diff --git a/backend/src/utils/read/models.py b/backend/src/utils/read/models.py index 41e8789..3d53871 100644 --- a/backend/src/utils/read/models.py +++ b/backend/src/utils/read/models.py @@ -212,6 +212,48 @@ class MessageVersionView(BaseModel): text: str | None +class FeedChat(BaseModel): + chat_id: int + title: str | None + kind: str + has_avatar: bool + + +class FeedItem(BaseModel): + id: int + kind: str + at: datetime + since: datetime + message_count: int + chat: FeedChat + message: MessageView | None + sender_id: int | None + sender_name: str | None + outgoing: bool + before: str | None = None + after: str | None = None + has_before: bool = False + edit_date: datetime | None = None + + +class FeedPage(BaseModel): + items: list[FeedItem] + next_cursor: str | None + + +class FeedUnseen(BaseModel): + count: int + deleted: int + edited: int + seen_at: datetime | None + latest: FeedItem | None + + +class FeedMessagesPage(BaseModel): + messages: list[MessageView] + next_cursor: str | None + + class MediaView(BaseModel): id: int account_id: int