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