-- Realtime-Grundlage: monotoner Cursor je Conversation + Änderungs-Log.
--
-- Warum ein Zähler auf `conversations` und nicht BIGSERIAL auf `messages`:
-- Eine Sequenz vergibt die Nummer beim INSERT, sichtbar wird die Zeile aber erst
-- beim COMMIT. Diese beiden Reihenfolgen können auseinanderlaufen — Transaktion A
-- holt 100, B holt 101 und committet zuerst; ein Leser bei `since = 101` verliert
-- Nachricht 100 für immer. Der Zähler hier wird per UPDATE gezogen und nimmt damit
-- den Zeilen-Lock der Conversation: der zweite Schreiber blockiert bis zum COMMIT
-- des ersten, also ist Commit-Reihenfolge == seq-Reihenfolge.

ALTER TABLE conversations ADD COLUMN seq BIGINT NOT NULL DEFAULT 0;

-- Das Änderungs-Log. Eine Zeile je Mutation — diese Zeile *ist* der Event, der
-- über den WebSocket geht. Deckt auch ab, was keine eigene Nachrichtenzeile hat
-- (Reactions, Attachments, später Presence und Read-Receipts).
CREATE TABLE conversation_events (
    conversation_id UUID        NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    seq             BIGINT      NOT NULL,
    kind            TEXT        NOT NULL,
    target_id       UUID        NOT NULL,
    at              TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (conversation_id, seq)
);
-- Der Primärschlüssel bedient bereits `WHERE conversation_id = $1 AND seq > $2
-- ORDER BY seq` — der Backfill braucht keinen weiteren Index.

-- Unveränderlicher Sortierschlüssel einer Nachricht: die seq ihres
-- Erzeugungs-Events. Damit verliert `created_at` seine Sortierrolle — es ist NOW(),
-- also der Transaktionsstart, weder eindeutig noch commit-geordnet.
ALTER TABLE messages ADD COLUMN seq BIGINT;

-- Bestand nachziehen: nach created_at durchnummerieren, dann den Conversation-
-- Zähler auf das Maximum setzen.
WITH nummeriert AS (
    SELECT id, row_number() OVER (PARTITION BY conversation_id ORDER BY created_at, id) AS n
    FROM messages
)
UPDATE messages m
SET    seq = nummeriert.n
FROM   nummeriert
WHERE  m.id = nummeriert.id;

UPDATE conversations c
SET    seq = COALESCE((SELECT max(m.seq) FROM messages m WHERE m.conversation_id = c.id), 0);

ALTER TABLE messages ALTER COLUMN seq SET NOT NULL;
CREATE UNIQUE INDEX messages_conversation_seq_idx ON messages(conversation_id, seq);

-- Für Bestandsnachrichten werden bewusst KEINE Log-Zeilen nachgetragen: ein Client
-- ohne Cursor bekommt `resync` und lädt über den Historien-Endpunkt, nie über das
-- Log. Das Log braucht also keine Vergangenheit, nur Zukunft.

-- Der Signalpfad. Der Trigger enthält keine Fachlogik — er feuert nur. Damit kann
-- kein künftiger Schreibweg das Benachrichtigen vergessen, und weil NOTIFY
-- transaktional ist, feuert er erst beim COMMIT, also genau dann, wenn die Zeile
-- auch sichtbar wird.
CREATE FUNCTION notify_conversation_event() RETURNS TRIGGER AS $$
BEGIN
    PERFORM pg_notify(
        'conversation_changed',
        json_build_object('conversation', NEW.conversation_id, 'seq', NEW.seq)::text
    );
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER conversation_events_notify
    AFTER INSERT ON conversation_events
    FOR EACH ROW EXECUTE FUNCTION notify_conversation_event();