322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""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")
|