From 8e0f828dd45f00690841e29f748ca6cd62f06889 Mon Sep 17 00:00:00 2001 From: h Date: Mon, 19 Oct 2026 03:25:43 +0000 Subject: [PATCH] feat(backend): feed lists name, username and photo changes of people and chat renames, merged per day --- .../versions/c5e7a9b1d3f6_feed_profiles.py | 321 ++++++++++++++++++ backend/src/api/mcp/server.py | 20 +- backend/src/utils/read/feed.py | 22 +- backend/src/utils/read/models.py | 1 + 4 files changed, 352 insertions(+), 12 deletions(-) create mode 100644 backend/migrations/versions/c5e7a9b1d3f6_feed_profiles.py diff --git a/backend/migrations/versions/c5e7a9b1d3f6_feed_profiles.py b/backend/migrations/versions/c5e7a9b1d3f6_feed_profiles.py new file mode 100644 index 0000000..05f9d2c --- /dev/null +++ b/backend/migrations/versions/c5e7a9b1d3f6_feed_profiles.py @@ -0,0 +1,321 @@ +"""feed of profile changes and chat renames + +Revision ID: c5e7a9b1d3f6 +Revises: b3d5f7a9c2e4 +Create Date: 2026-10-19 04:00:00.000000 + +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "c5e7a9b1d3f6" +down_revision: str | None = "b3d5f7a9c2e4" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_PEER_VALUE = """ +CREATE FUNCTION feed_peer_value( + field text, first_name text, last_name text, username text, photo text +) RETURNS text LANGUAGE sql IMMUTABLE AS $$ + SELECT CASE field + WHEN 'name' THEN NULLIF(trim(concat_ws(' ', first_name, last_name)), '') + WHEN 'username' THEN NULLIF(username, '') + WHEN 'photo' THEN NULLIF(photo, '') END +$$ +""" + +_PEER_SCOPE = """ +CREATE FUNCTION feed_peer_in_scope(account integer, peer bigint) +RETURNS boolean LANGUAGE sql STABLE AS $$ + SELECT EXISTS ( + SELECT 1 FROM dialogs WHERE account_id = account AND chat_id = peer + ) OR EXISTS ( + SELECT 1 FROM peers + WHERE account_id = account AND peer_id = peer + AND (raw->>'is_contact' = 'true' OR raw->>'contact' = 'true') + ) +$$ +""" + +_MERGE_DETAIL = """ +CREATE FUNCTION feed_merge_detail(base jsonb, incoming jsonb) +RETURNS jsonb LANGUAGE sql IMMUTABLE AS $$ + SELECT coalesce(jsonb_object_agg(key, pair), '{}'::jsonb) + FROM ( + SELECT key, jsonb_build_array( + coalesce(base->key->0, incoming->key->0), + coalesce(incoming->key->1, base->key->1)) AS pair + FROM ( + SELECT jsonb_object_keys(coalesce(base, '{}'::jsonb)) AS key + UNION + SELECT jsonb_object_keys(incoming) + ) keys + ) merged + WHERE pair->0 IS DISTINCT FROM pair->1 +$$ +""" + +_UPSERT = """ +CREATE FUNCTION feed_profile_upsert( + account integer, chat bigint, kind_of_chat text, message bigint, + changed_at timestamptz, changes jsonb +) RETURNS void LANGUAGE plpgsql AS $$ +DECLARE + item feed_items%ROWTYPE; + merged jsonb; +BEGIN + SELECT * INTO item FROM feed_items + WHERE account_id = account AND chat_id = chat AND kind = 'profile' + ORDER BY at DESC, id DESC LIMIT 1; + IF FOUND AND changed_at BETWEEN item.since - interval '1 day' + AND item.at + interval '1 day' THEN + merged := feed_merge_detail(item.detail, changes); + IF merged = '{}'::jsonb THEN + DELETE FROM feed_items WHERE id = item.id; + ELSE + UPDATE feed_items SET + detail = merged, + message_count = message_count + + (SELECT count(*) FROM jsonb_object_keys(changes)), + at = greatest(at, changed_at), + since = least(since, changed_at), + message_id = greatest(message_id, message) + WHERE id = item.id; + END IF; + ELSE + INSERT INTO feed_items + (account_id, kind, chat_id, chat_kind, message_id, at, since, + message_count, detail) + VALUES (account, 'profile', chat, kind_of_chat, message, changed_at, + changed_at, + (SELECT count(*) FROM jsonb_object_keys(changes)), changes); + END IF; +END; +$$ +""" + +_ON_PEER = """ +CREATE FUNCTION feed_on_peer_change() RETURNS trigger LANGUAGE plpgsql AS $$ +DECLARE + changes jsonb := '{}'::jsonb; + field text; + fresh text; + old text; +BEGIN + IF NEW.peer_id <= 0 OR NOT feed_peer_in_scope(NEW.account_id, NEW.peer_id) THEN + RETURN NULL; + END IF; + FOREACH field IN ARRAY ARRAY['name', 'username', 'photo'] LOOP + fresh := feed_peer_value(field, NEW.first_name, NEW.last_name, + NEW.username, NEW.photo_unique_id); + CONTINUE WHEN fresh IS NULL; + SELECT value INTO old FROM ( + SELECT feed_peer_value(field, h.first_name, h.last_name, h.username, + h.photo_unique_id) AS value, h.observed_at + FROM peer_history h + WHERE h.account_id = NEW.account_id AND h.peer_id = NEW.peer_id + AND h.observed_at < NEW.observed_at + ) earlier + WHERE value IS NOT NULL + ORDER BY observed_at DESC LIMIT 1; + CONTINUE WHEN old IS NULL OR old = fresh; + CONTINUE WHEN EXISTS ( + SELECT 1 FROM peer_history h + WHERE h.account_id = NEW.account_id AND h.peer_id = NEW.peer_id + AND h.observed_at < NEW.observed_at + AND feed_peer_value(field, h.first_name, h.last_name, h.username, + h.photo_unique_id) = fresh + ); + changes := changes || jsonb_build_object(field, jsonb_build_array(old, fresh)); + END LOOP; + IF changes <> '{}'::jsonb THEN + PERFORM feed_profile_upsert(NEW.account_id, NEW.peer_id, 'dm', 0, + NEW.observed_at, changes); + END IF; + RETURN NULL; +END; +$$ +""" + +_ON_TITLE = """ +CREATE FUNCTION feed_on_chat_title() RETURNS trigger LANGUAGE plpgsql AS $$ +DECLARE + previous text; +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM dialogs + WHERE account_id = NEW.account_id AND chat_id = NEW.chat_id + ) THEN + RETURN NULL; + END IF; + SELECT title INTO previous FROM chat_history + WHERE account_id = NEW.account_id AND chat_id = NEW.chat_id + AND event IN ('meta', 'title') AND title IS NOT NULL + AND (ts, message_id) < (NEW.ts, NEW.message_id) + ORDER BY ts DESC, message_id DESC LIMIT 1; + IF previous IS NULL OR previous = NEW.title THEN + RETURN NULL; + END IF; + PERFORM feed_profile_upsert( + NEW.account_id, NEW.chat_id, + feed_chat_kind(NEW.chat_id, NEW.raw->'chat'->>'type'), + CASE WHEN NEW.event = 'title' THEN NEW.message_id ELSE 0 END, + NEW.ts, jsonb_build_object('title', jsonb_build_array(previous, NEW.title))); + RETURN NULL; +END; +$$ +""" + +_BACKFILL_PEERS = """ +WITH scope AS ( + SELECT account_id::integer AS account_id, chat_id AS peer_id + FROM dialogs WHERE chat_id > 0 + UNION + SELECT account_id, peer_id FROM peers + WHERE raw->>'is_contact' = 'true' OR raw->>'contact' = 'true' +), firsts AS ( + SELECT h.account_id, h.peer_id, f.field, f.value, + min(h.observed_at) AS at + FROM peer_history h + JOIN scope s ON s.account_id = h.account_id AND s.peer_id = h.peer_id + CROSS JOIN LATERAL ( + SELECT field, feed_peer_value(field, h.first_name, h.last_name, + h.username, h.photo_unique_id) AS value + FROM unnest(ARRAY['name', 'username', 'photo']) AS field + ) f + WHERE f.value IS NOT NULL + GROUP BY 1, 2, 3, 4 +), ranked AS ( + SELECT *, row_number() OVER ( + PARTITION BY account_id, peer_id, field ORDER BY at + ) AS position + FROM firsts +), changes AS ( + SELECT r.account_id, r.peer_id, r.field, r.value AS fresh, r.at, o.value AS old + FROM ranked r + CROSS JOIN LATERAL ( + SELECT value FROM ( + SELECT feed_peer_value(r.field, h.first_name, h.last_name, + h.username, h.photo_unique_id) AS value, + h.observed_at + FROM peer_history h + WHERE h.account_id = r.account_id AND h.peer_id = r.peer_id + AND h.observed_at < r.at + ) earlier + WHERE value IS NOT NULL + ORDER BY observed_at DESC LIMIT 1 + ) o + WHERE r.position > 1 AND r.at > now() - interval '90 days' + AND o.value <> r.value +), gaps AS ( + SELECT *, at - lag(at) OVER (PARTITION BY account_id, peer_id ORDER BY at) + AS gap + FROM changes +), bursts AS ( + SELECT *, count(*) FILTER (WHERE gap IS NULL OR gap > interval '1 day') + OVER (PARTITION BY account_id, peer_id ORDER BY at) AS burst + FROM gaps +), fields AS ( + SELECT account_id, peer_id, burst, field, + (array_agg(old ORDER BY at))[1] AS old, + (array_agg(fresh ORDER BY at DESC))[1] AS fresh, + count(*) AS changes, min(at) AS since, max(at) AS at + FROM bursts GROUP BY 1, 2, 3, 4 +) +INSERT INTO feed_items + (account_id, kind, chat_id, chat_kind, message_id, at, since, + message_count, detail) +SELECT account_id, 'profile', peer_id, 'dm', 0, max(at), min(since), + sum(changes)::integer, + jsonb_object_agg(field, jsonb_build_array(old, fresh)) + FILTER (WHERE old <> fresh) +FROM fields +GROUP BY account_id, peer_id, burst +HAVING bool_or(old <> fresh) +ORDER BY max(at) +""" + +_BACKFILL_CHATS = """ +WITH titles AS ( + SELECT account_id, chat_id, message_id, event, ts, title, + raw->'chat'->>'type' AS chat_type, + lag(title) OVER ( + PARTITION BY account_id, chat_id ORDER BY ts, message_id + ) AS previous + FROM chat_history + WHERE event IN ('meta', 'title') AND title IS NOT NULL +), changes AS ( + SELECT * FROM titles t + WHERE previous IS NOT NULL AND previous <> title + AND ts > now() - interval '90 days' + AND EXISTS ( + SELECT 1 FROM dialogs d + WHERE d.account_id = t.account_id AND d.chat_id = t.chat_id + ) +), gaps AS ( + SELECT *, ts - lag(ts) OVER (PARTITION BY account_id, chat_id ORDER BY ts) + AS gap + FROM changes +), bursts AS ( + SELECT *, count(*) FILTER (WHERE gap IS NULL OR gap > interval '1 day') + OVER (PARTITION BY account_id, chat_id ORDER BY ts) AS burst + FROM gaps +), grouped AS ( + SELECT account_id, chat_id, burst, + (array_agg(chat_type ORDER BY ts DESC))[1] AS chat_type, + max(CASE WHEN event = 'title' THEN message_id ELSE 0 END) AS message_id, + (array_agg(previous ORDER BY ts))[1] AS old, + (array_agg(title ORDER BY ts DESC))[1] AS fresh, + count(*) AS changes, min(ts) AS since, max(ts) AS at + FROM bursts GROUP BY 1, 2, 3 +) +INSERT INTO feed_items + (account_id, kind, chat_id, chat_kind, message_id, at, since, + message_count, detail) +SELECT account_id, 'profile', chat_id, feed_chat_kind(chat_id, chat_type), + message_id, at, since, changes::integer, + jsonb_build_object('title', jsonb_build_array(old, fresh)) +FROM grouped +WHERE old <> fresh +ORDER BY at +""" + + +def upgrade() -> None: + op.execute("SET LOCAL jit = off") + op.execute("ALTER TABLE feed_items ADD COLUMN detail jsonb") + op.execute(_PEER_VALUE) + op.execute(_PEER_SCOPE) + op.execute(_MERGE_DETAIL) + op.execute(_UPSERT) + op.execute(_ON_PEER) + op.execute(_ON_TITLE) + op.execute( + "CREATE TRIGGER peer_history_feed AFTER INSERT ON peer_history " + "FOR EACH ROW EXECUTE FUNCTION feed_on_peer_change()" + ) + op.execute( + "CREATE TRIGGER chat_history_feed AFTER INSERT ON chat_history " + "FOR EACH ROW WHEN (NEW.event IN ('meta', 'title') AND NEW.title IS NOT NULL) " + "EXECUTE FUNCTION feed_on_chat_title()" + ) + op.execute(_BACKFILL_PEERS) + op.execute(_BACKFILL_CHATS) + + +def downgrade() -> None: + op.execute("DROP TRIGGER IF EXISTS chat_history_feed ON chat_history") + op.execute("DROP TRIGGER IF EXISTS peer_history_feed ON peer_history") + op.execute("DROP FUNCTION IF EXISTS feed_on_chat_title()") + op.execute("DROP FUNCTION IF EXISTS feed_on_peer_change()") + op.execute( + "DROP FUNCTION IF EXISTS " + "feed_profile_upsert(integer, bigint, text, bigint, timestamptz, jsonb)" + ) + op.execute("DROP FUNCTION IF EXISTS feed_merge_detail(jsonb, jsonb)") + op.execute("DROP FUNCTION IF EXISTS feed_peer_in_scope(integer, bigint)") + op.execute("DROP FUNCTION IF EXISTS feed_peer_value(text, text, text, text, text)") + op.execute("DELETE FROM feed_items WHERE kind = 'profile'") + op.execute("ALTER TABLE feed_items DROP COLUMN detail") diff --git a/backend/src/api/mcp/server.py b/backend/src/api/mcp/server.py index 326f13c..e302d23 100644 --- a/backend/src/api/mcp/server.py +++ b/backend/src/api/mcp/server.py @@ -78,7 +78,8 @@ Typical flows: joins and leaves, deleted message count, the private chat. - Forensics: `get_feed` lists what was deleted or edited in Telegram across all chats, newest first, with the text before and after each edit; a bulk - deletion is one item with `message_count`. `get_deleted_messages` and + deletion is one item with `message_count`. Its `profile` kind lists name, + username and photo changes of people and renames of chats. `get_deleted_messages` and `get_message_versions` recover the content of one chat or one message. - Files: `download_media` gets any message attachment (photo, video, voice, document, ...). It pulls the file from Telegram if the archive lacks it and @@ -270,14 +271,17 @@ async def get_feed( limit: int = DEFAULT_LIMIT, cursor: str | None = None, ) -> dict[str, Any]: - """Deleted and edited messages across chats, newest first. + """Deleted and edited messages and profile changes across chats, newest first. - `kind` picks deleted, edited or all; `since` keeps items at or after that - time; `chat_id` or `chat_kind` (dm, group, channel) narrow it down. Edited - items carry `before` and `after` texts. A deleted item with - `message_count` above 1 is a bulk deletion in one chat; read its messages - with `get_deleted_messages` for that chat. Pass `next_cursor` back as - `cursor` for the next page. + `kind` picks deleted, edited, profile, messages (deleted and edited) or all; + `since` keeps items at or after that time; `chat_id` or `chat_kind` (dm, + group, channel) narrow it down. Edited items carry `before` and `after` + texts. A deleted item with `message_count` above 1 is a bulk deletion in + one chat; read its messages with `get_deleted_messages` for that chat. + Profile items cover people in private chats or contacts and chats in the + dialog list: `detail` maps `name`, `username`, `photo` (unique file ids) or + `title` to `[old, new]`, changes a day apart or closer are merged. Pass + `next_cursor` back as `cursor` for the next page. """ if cursor is not None and re.fullmatch(CURSOR_PATTERN, cursor) is None: msg = "cursor must be a next_cursor returned by get_feed" diff --git a/backend/src/utils/read/feed.py b/backend/src/utils/read/feed.py index 83f9dfd..96cf01b 100644 --- a/backend/src/utils/read/feed.py +++ b/backend/src/utils/read/feed.py @@ -1,3 +1,4 @@ +import json from dataclasses import dataclass from datetime import UTC, datetime, timedelta from enum import StrEnum @@ -24,7 +25,8 @@ _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" + "id, kind, chat_id, chat_kind, message_id, message_date, at, since, " + "message_count, detail" ) _VERSIONS = """ @@ -59,8 +61,10 @@ SET seen_at = greatest(feed_seen.seen_at, EXCLUDED.seen_at) class FeedKind(StrEnum): ALL = "all" + MESSAGES = "messages" DELETED = "deleted" EDITED = "edited" + PROFILE = "profile" class FeedChatKind(StrEnum): @@ -91,8 +95,11 @@ def _where( ) -> tuple[str, list[object]]: params: list[object] = [account_id] clauses = ["account_id = $1"] + if feed.kind is FeedKind.MESSAGES: + clauses.append("kind <> 'profile'") + exact = feed.kind not in {FeedKind.ALL, FeedKind.MESSAGES} conditions = ( - ("kind = {}", None if feed.kind is FeedKind.ALL else feed.kind.value), + ("kind = {}", feed.kind.value if exact else None), ("chat_id = {}", feed.chat_id), ("chat_kind = {}", feed.chat_kind.value if feed.chat_kind else None), ("at >= {}", feed.since), @@ -153,7 +160,11 @@ async def _build_items( messages = await chats.get_messages_at( pool, account_id, - [(row["chat_id"], row["message_id"], row["message_date"]) for row in rows], + [ + (row["chat_id"], row["message_id"], row["message_date"]) + for row in rows + if row["kind"] != FeedKind.PROFILE + ], ) versions = await _versions(pool, account_id, rows) chat_views = { @@ -192,6 +203,7 @@ async def _build_items( after=version["text"] if version else None, has_before=bool(version and version["has_before"]), edit_date=version["edit_date"] if version else None, + detail=json.loads(row["detail"]) if row["detail"] else None, ) ) return items @@ -226,7 +238,9 @@ async def unseen(pool: asyncpg.Pool, account_id: int) -> FeedUnseen: "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) + latest = await list_feed( + pool, account_id, FeedFilter(kind=FeedKind.MESSAGES), limit=1 + ) return FeedUnseen( count=counts["deleted"] + counts["edited"], deleted=counts["deleted"], diff --git a/backend/src/utils/read/models.py b/backend/src/utils/read/models.py index 5033d8d..794e6ef 100644 --- a/backend/src/utils/read/models.py +++ b/backend/src/utils/read/models.py @@ -235,6 +235,7 @@ class FeedItem(BaseModel): after: str | None = None has_before: bool = False edit_date: datetime | None = None + detail: dict[str, list[str | None]] | None = None class FeedPage(BaseModel):