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