-- 014: Streaming slice S5 — chat hub + device pairing -- (docs/design/streaming-architecture.md §3, §6 S5). -- -- Design notes: -- * `live_chat_messages` is keyed by (workspace, channel) — the chat hub's -- aggregate is a virtual channel's live chat, and the seq cursor a client -- resumes with is per channel. Global provider dedupe uses a nonpartitioned -- claim journal; the paging index is (workspace_id, channel_id, seq). -- * Ordering is a hub-assigned `seq` under a row lock on the channel's -- chat_seq row; provider timestamps stay display-only (clocks skew, §3.2). -- * Monthly received_at partitions retain ingress ordering. PostgreSQL needs -- the partition key in unique constraints, so global provider dedupe lives -- in the nonpartitioned live_chat_seen_events journal, claimed in the same -- transaction as the message insert. -- * Devices and pairings carry their own workspace column and direct -- workspace FK so revocation is workspace-scoped; user_id references -- users(id) ON DELETE CASCADE so deleting a user removes their devices. -- -- RLS / NULLIF / sp_app grants follow 012/013 conventions. -- One row per (workspace, channel) holding the chat sequence. FOR UPDATE on -- this row inside the ingest transaction is what makes seq assignment monotonic -- per channel (§3.2 "monotonic seq assigned at hub ingress"). CREATE TABLE live_chat_seq ( workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT, channel_id uuid NOT NULL, seq bigint NOT NULL DEFAULT 0 CHECK (seq >= 0), updated_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (workspace_id, channel_id), FOREIGN KEY (channel_id, workspace_id) REFERENCES virtual_channels (id, workspace_id) ON DELETE CASCADE ON UPDATE RESTRICT ); CREATE TABLE live_chat_messages ( id uuid NOT NULL DEFAULT gen_random_uuid(), workspace_id uuid NOT NULL, channel_id uuid NOT NULL, -- Provider connection that produced the message (capability target). connection_id uuid NOT NULL, provider text NOT NULL CHECK (btrim(provider) <> ''), -- Provider's own message id; the dedupe key with workspace + provider. provider_msg_id text NOT NULL CHECK (btrim(provider_msg_id) <> ''), -- Author author_handle text NOT NULL CHECK (btrim(author_handle) <> ''), author_display_name text, author_avatar_url text, author_badges jsonb NOT NULL DEFAULT '[]'::jsonb CHECK (jsonb_typeof(author_badges) = 'array'), author_color text NOT NULL CHECK (author_color <> ''), body text NOT NULL CHECK (length(body) <= 5000), -- Reply threading (provider message id of the parent, nullable). reply_to text, -- Hub ingress seq bigint NOT NULL CHECK (seq > 0), sent_at timestamptz NOT NULL, received_at timestamptz NOT NULL DEFAULT now(), is_moderated boolean NOT NULL DEFAULT false, source text NOT NULL DEFAULT 'live-chat' CHECK (source IN ('live-chat', 'social-reply', 'system')), created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (id, received_at), UNIQUE (workspace_id, provider, provider_msg_id, received_at), FOREIGN KEY (channel_id, workspace_id) REFERENCES virtual_channels (id, workspace_id) ON DELETE CASCADE ON UPDATE RESTRICT, FOREIGN KEY (connection_id, workspace_id) REFERENCES connections (id, workspace_id) ON DELETE NO ACTION ON UPDATE RESTRICT DEFERRABLE INITIALLY DEFERRED ) PARTITION BY RANGE (received_at); -- Precreate monthly partitions; administrators run this before a new month. CREATE FUNCTION live_chat_ensure_month(p_month date) RETURNS void LANGUAGE plpgsql SET search_path = public AS $$ DECLARE v_start timestamptz := date_trunc('month', p_month::timestamptz); v_table text := 'live_chat_messages_' || to_char(v_start, 'YYYYMM'); BEGIN EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF live_chat_messages FOR VALUES FROM (%L) TO (%L)', v_table, v_start, v_start + interval '1 month'); EXECUTE format('ALTER TABLE %I ENABLE ROW LEVEL SECURITY', v_table); EXECUTE format('ALTER TABLE %I FORCE ROW LEVEL SECURITY', v_table); IF NOT EXISTS (SELECT 1 FROM pg_policies WHERE schemaname='public' AND tablename=v_table AND policyname='tenant_isolation') THEN EXECUTE format('CREATE POLICY tenant_isolation ON %I USING (workspace_id = NULLIF(current_setting(''app.workspace_id'', true), '''')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting(''app.workspace_id'', true), '''')::uuid)', v_table); END IF; END $$; REVOKE ALL ON FUNCTION live_chat_ensure_month(date) FROM PUBLIC; SELECT live_chat_ensure_month((current_date - interval '1 month')::date); SELECT live_chat_ensure_month(current_date); SELECT live_chat_ensure_month((current_date + interval '1 month')::date); CREATE TABLE live_chat_messages_default PARTITION OF live_chat_messages DEFAULT; ALTER TABLE live_chat_messages_default ENABLE ROW LEVEL SECURITY; ALTER TABLE live_chat_messages_default FORCE ROW LEVEL SECURITY; CREATE POLICY tenant_isolation ON live_chat_messages_default USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); CREATE INDEX live_chat_messages_channel_seq ON live_chat_messages (workspace_id, channel_id, seq); CREATE INDEX live_chat_messages_author ON live_chat_messages (workspace_id, channel_id, author_handle); -- Global provider-message/event claim journal, mirroring -- monitor_seen_events (ON CONFLICT DO NOTHING semantics come from the PK). CREATE TABLE live_chat_seen_events ( workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, provider text NOT NULL CHECK (btrim(provider) <> ''), provider_event_id text NOT NULL CHECK (btrim(provider_event_id) <> ''), seen_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (workspace_id, provider, provider_event_id) ); CREATE TRIGGER trg_live_chat_seen_event_ws_immutable BEFORE UPDATE ON live_chat_seen_events FOR EACH ROW EXECUTE FUNCTION reject_workspace_change(); -- Devices: Opticast authenticates with a device token, not a session cookie -- (§3.4). Epoch-gated: revoking bumps the epoch, tokens fail closed. CREATE TABLE devices ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT, user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE ON UPDATE RESTRICT, name text NOT NULL CHECK (btrim(name) <> '' AND length(name) <= 120), scopes text[] NOT NULL DEFAULT '{livechat.view,stream.view}'::text[], epoch bigint NOT NULL DEFAULT 1 CHECK (epoch >= 1), authz_epoch bigint NOT NULL, last_seen_at timestamptz, revoked_at timestamptz, created_at timestamptz NOT NULL DEFAULT now(), UNIQUE (workspace_id, id) ); CREATE INDEX devices_ws_user ON devices (workspace_id, user_id); CREATE INDEX devices_ws_revoked ON devices (workspace_id) WHERE revoked_at IS NULL; -- Single-use, TTL'd pairing codes. `redeemed_at` enforces single use; the -- expiry is checked by the app (a code may expire unredeemed, and a purge of -- expired rows is a plain DELETE). CREATE TABLE device_pairings ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT, user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE ON UPDATE RESTRICT, code text NOT NULL CHECK (code ~ '^[A-Z0-9]{4}-[A-Z0-9]{4}$'), qr_payload text NOT NULL, scopes text[] NOT NULL, expires_at timestamptz NOT NULL, redeemed_at timestamptz, device_id uuid, created_at timestamptz NOT NULL DEFAULT now(), UNIQUE (code), UNIQUE (workspace_id, code), FOREIGN KEY (device_id, workspace_id) REFERENCES devices (id, workspace_id) ON DELETE NO ACTION ON UPDATE RESTRICT DEFERRABLE INITIALLY DEFERRED ); CREATE INDEX device_pairings_ws_open ON device_pairings (workspace_id, code) WHERE redeemed_at IS NULL; CREATE TRIGGER trg_device_ws_immutable BEFORE UPDATE ON devices FOR EACH ROW EXECUTE FUNCTION reject_workspace_change(); CREATE TRIGGER trg_device_pairing_ws_immutable BEFORE UPDATE ON device_pairings FOR EACH ROW EXECUTE FUNCTION reject_workspace_change(); CREATE TRIGGER trg_live_chat_message_ws_immutable BEFORE UPDATE ON live_chat_messages FOR EACH ROW EXECUTE FUNCTION reject_workspace_change(); CREATE TRIGGER trg_live_chat_seq_ws_immutable BEFORE UPDATE ON live_chat_seq FOR EACH ROW EXECUTE FUNCTION reject_workspace_change(); ALTER TABLE live_chat_seq ENABLE ROW LEVEL SECURITY; ALTER TABLE live_chat_messages ENABLE ROW LEVEL SECURITY; ALTER TABLE live_chat_seen_events ENABLE ROW LEVEL SECURITY; ALTER TABLE devices ENABLE ROW LEVEL SECURITY; ALTER TABLE device_pairings ENABLE ROW LEVEL SECURITY; ALTER TABLE live_chat_seq FORCE ROW LEVEL SECURITY; ALTER TABLE live_chat_messages FORCE ROW LEVEL SECURITY; ALTER TABLE live_chat_seen_events FORCE ROW LEVEL SECURITY; ALTER TABLE devices FORCE ROW LEVEL SECURITY; ALTER TABLE device_pairings FORCE ROW LEVEL SECURITY; CREATE POLICY tenant_isolation ON live_chat_seq USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); CREATE POLICY tenant_isolation ON live_chat_messages USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); CREATE POLICY tenant_isolation ON live_chat_seen_events USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); CREATE POLICY tenant_isolation ON devices USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); CREATE POLICY tenant_isolation ON device_pairings USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); GRANT SELECT, INSERT, UPDATE, DELETE ON live_chat_seq, live_chat_messages, live_chat_seen_events, devices, device_pairings TO sp_app; -- Retention (follow-on, noted here so the partition/retention row from the -- work-breakdown finds its anchor): pruning live_chat_seen_events older than -- the journal window, mirroring monitor_prune_seen_events from 012. CREATE FUNCTION live_chat_prune_seen_events(p_before timestamptz) RETURNS integer LANGUAGE sql SECURITY INVOKER SET search_path = public AS $$ WITH removed AS ( DELETE FROM live_chat_seen_events WHERE seen_at < p_before RETURNING 1 ) SELECT count(*) FROM removed $$; REVOKE ALL ON FUNCTION live_chat_prune_seen_events(timestamptz) FROM PUBLIC; GRANT EXECUTE ON FUNCTION live_chat_prune_seen_events(timestamptz) TO sp_app; -- Resolves only a valid short-lived bearer pairing code; no tenant rows exposed. CREATE FUNCTION device_pairing_workspace(p_code text) RETURNS uuid LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$ SELECT workspace_id FROM device_pairings WHERE code=p_code AND expires_at > now() $$; REVOKE ALL ON FUNCTION device_pairing_workspace(text) FROM PUBLIC; GRANT EXECUTE ON FUNCTION device_pairing_workspace(text) TO sp_app; -- Tenant-scoped retention for messages and their dedupe claims. CREATE FUNCTION live_chat_prune(p_workspace uuid, p_before timestamptz) RETURNS integer LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ DECLARE v_count integer; BEGIN PERFORM fn_assert_tenant(p_workspace); DELETE FROM live_chat_messages WHERE workspace_id=p_workspace AND received_at < p_before; GET DIAGNOSTICS v_count = ROW_COUNT; DELETE FROM live_chat_seen_events WHERE workspace_id=p_workspace AND seen_at < p_before; DELETE FROM device_pairings WHERE workspace_id=p_workspace AND expires_at < p_before; RETURN v_count; END $$; REVOKE ALL ON FUNCTION live_chat_prune(uuid,timestamptz) FROM PUBLIC; GRANT EXECUTE ON FUNCTION live_chat_prune(uuid,timestamptz) TO sp_app; CREATE TABLE live_chat_requests ( workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, route text NOT NULL, request_key text NOT NULL, request_hash text NOT NULL, status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','complete','failed')), response jsonb, created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (workspace_id,user_id,route,request_key) ); ALTER TABLE live_chat_requests ENABLE ROW LEVEL SECURITY; ALTER TABLE live_chat_requests FORCE ROW LEVEL SECURITY; CREATE POLICY tenant_isolation ON live_chat_requests USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid) WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); CREATE TRIGGER trg_live_chat_request_ws_immutable BEFORE UPDATE ON live_chat_requests FOR EACH ROW EXECUTE FUNCTION reject_workspace_change(); GRANT SELECT,INSERT,UPDATE,DELETE ON live_chat_requests TO sp_app;