feat(backend): feed lists name, username and photo changes of people and chat renames, merged per day

This commit is contained in:
hh committed 2026-10-19 03:25:43 +00:00
1 parent 9aa8690e33
commit 8e0f828dd4
4 files changed
+352 -12

No files matched your search

@@ -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")
+12 -8
View File
@@ -78,7 +78,8 @@ Typical flows:
joins and leaves, deleted message count, the private chat. joins and leaves, deleted message count, the private chat.
- Forensics: `get_feed` lists what was deleted or edited in Telegram across - 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 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. `get_message_versions` recover the content of one chat or one message.
- Files: `download_media` gets any message attachment (photo, video, voice, - Files: `download_media` gets any message attachment (photo, video, voice,
document, ...). It pulls the file from Telegram if the archive lacks it and 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, limit: int = DEFAULT_LIMIT,
cursor: str | None = None, cursor: str | None = None,
) -> dict[str, Any]: ) -> 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 `kind` picks deleted, edited, profile, messages (deleted and edited) or all;
time; `chat_id` or `chat_kind` (dm, group, channel) narrow it down. Edited `since` keeps items at or after that time; `chat_id` or `chat_kind` (dm,
items carry `before` and `after` texts. A deleted item with group, channel) narrow it down. Edited items carry `before` and `after`
`message_count` above 1 is a bulk deletion in one chat; read its messages texts. A deleted item with `message_count` above 1 is a bulk deletion in
with `get_deleted_messages` for that chat. Pass `next_cursor` back as one chat; read its messages with `get_deleted_messages` for that chat.
`cursor` for the next page. 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: if cursor is not None and re.fullmatch(CURSOR_PATTERN, cursor) is None:
msg = "cursor must be a next_cursor returned by get_feed" msg = "cursor must be a next_cursor returned by get_feed"
+18 -4
View File
@@ -1,3 +1,4 @@
import json
from dataclasses import dataclass from dataclasses import dataclass
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from enum import StrEnum from enum import StrEnum
@@ -24,7 +25,8 @@ _EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
_MICROSECOND = timedelta(microseconds=1) _MICROSECOND = timedelta(microseconds=1)
_ITEM_COLS = ( _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 = """ _VERSIONS = """
@@ -59,8 +61,10 @@ SET seen_at = greatest(feed_seen.seen_at, EXCLUDED.seen_at)
class FeedKind(StrEnum): class FeedKind(StrEnum):
ALL = "all" ALL = "all"
MESSAGES = "messages"
DELETED = "deleted" DELETED = "deleted"
EDITED = "edited" EDITED = "edited"
PROFILE = "profile"
class FeedChatKind(StrEnum): class FeedChatKind(StrEnum):
@@ -91,8 +95,11 @@ def _where(
) -> tuple[str, list[object]]: ) -> tuple[str, list[object]]:
params: list[object] = [account_id] params: list[object] = [account_id]
clauses = ["account_id = $1"] clauses = ["account_id = $1"]
if feed.kind is FeedKind.MESSAGES:
clauses.append("kind <> 'profile'")
exact = feed.kind not in {FeedKind.ALL, FeedKind.MESSAGES}
conditions = ( 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_id = {}", feed.chat_id),
("chat_kind = {}", feed.chat_kind.value if feed.chat_kind else None), ("chat_kind = {}", feed.chat_kind.value if feed.chat_kind else None),
("at >= {}", feed.since), ("at >= {}", feed.since),
@@ -153,7 +160,11 @@ async def _build_items(
messages = await chats.get_messages_at( messages = await chats.get_messages_at(
pool, pool,
account_id, 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) versions = await _versions(pool, account_id, rows)
chat_views = { chat_views = {
@@ -192,6 +203,7 @@ async def _build_items(
after=version["text"] if version else None, after=version["text"] if version else None,
has_before=bool(version and version["has_before"]), has_before=bool(version and version["has_before"]),
edit_date=version["edit_date"] if version else None, edit_date=version["edit_date"] if version else None,
detail=json.loads(row["detail"]) if row["detail"] else None,
) )
) )
return items 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 "SELECT seen_at FROM feed_seen WHERE account_id = $1", account_id
) )
counts = await pool.fetchrow(_UNSEEN, account_id, seen_at) 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( return FeedUnseen(
count=counts["deleted"] + counts["edited"], count=counts["deleted"] + counts["edited"],
deleted=counts["deleted"], deleted=counts["deleted"],
+1
View File
@@ -235,6 +235,7 @@ class FeedItem(BaseModel):
after: str | None = None after: str | None = None
has_before: bool = False has_before: bool = False
edit_date: datetime | None = None edit_date: datetime | None = None
detail: dict[str, list[str | None]] | None = None
class FeedPage(BaseModel): class FeedPage(BaseModel):