Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
236 lines
13 KiB
PL/PgSQL
236 lines
13 KiB
PL/PgSQL
-- 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;
|