T-HUB/apps/api/migrations/027_intake.sql

-- 027: content intake firehose phase 1 (WP15+ firehose research §1/§2/§4/§5.3,
-- decision record docs/decisions/2026-10-09-intake-posture.md — Tier-1
-- sanctioned sources only, metadata-first).
--
-- NOTE on the number: this slice was specced as "026_intake.sql (next free)"
-- but 025_archive_search and 026_columns_p1 already exist, so intake lands as
-- 027 — the next free slot at authoring time.
--
-- Two tables:
--
--   intake_sources  the per-workspace subscription registry (§4.4 sketch,
--                   tenant-scoped per the phase-1 dedup note in §2.4: dedup
--                   happens WITHIN a workspace; the global item registry +
--                   claims split is the phase-1.5 upgrade).
--   intake_items    the harvested rows. Dedup is CONTENT-ADDRESSED:
--                   UNIQUE (workspace_id, content_hash) where content_hash is
--                   sha256(normalized plain text + media origin identity), so
--                   the same statement arriving via RSS and a Mastodon boost
--                   (or the same URL watched twice) stores once. Storage-level
--                   dedup stays in the S3 content-addressed layer.
--
-- `content` is the P0c safe-content rich-text AST — the ONLY render source;
-- adapters never hand raw provider HTML through (apps/api/src/feeds/
-- feed-normalize.ts funnels every item). `text` is the derived flat text
-- (astToPlainText) that FTS + pg_trgm index.
--
-- FTS mirrors 025's design: 'simple' config (multilingual firehose, no
-- stemming dictionary is "right"), websearch_to_tsquery in the API ORed with
-- pg_trgm similarity for typo tolerance; tsvector maintained by trigger
-- because edits/telegram revisions must reindex.
--
-- RLS / NULLIF / reject_workspace_change / sp_app grants follow the
-- 001/015/026 conventions: no app.workspace_id => NULL => no rows; FORCE
-- binds sp_app too. CREATE EXTENSION requires privileges sp_app lacks; this
-- migration runs as the table owner only (025 note).

CREATE EXTENSION IF NOT EXISTS pg_trgm;

-- ---------------------------------------------------------------------------
-- intake_sources
-- ---------------------------------------------------------------------------
CREATE TABLE intake_sources (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT,
  -- Tier-1 sanctioned intake kinds (decision record). bluesky_keyword is a
  -- stub seam in this slice — the Jetstream consumer is a later slice, so the
  -- scheduler parks those rows in health='unsupported' rather than polling.
  kind text NOT NULL CHECK (kind IN ('rss','mastodon_tag','mastodon_account','bluesky_keyword',
                                     'telegram_channel','peertube_channel','youtube_channel','web_watch')),
  -- Per-kind identity: url (rss/web_watch), tag (mastodon_tag), instance +
  -- accountId (mastodon_account), keywords[] (bluesky_keyword), channelId
  -- (telegram_channel / youtube_channel), handle + instance (peertube_channel).
  -- Validated app-side per kind; the DB only backstops "is an object".
  params jsonb NOT NULL CHECK (jsonb_typeof(params) = 'object'),
  -- sha256 of the canonically serialized params — the stable identity half of
  -- UNIQUE (workspace_id, kind, params_hash). Canonicalization is app-side.
  params_hash text NOT NULL CHECK (params_hash ~ '^[0-9a-f]{64}$'),
  -- 0 = push-only (telegram webhook bridge, bluesky Jetstream): never due.
  poll_interval_sec integer NOT NULL DEFAULT 300 CHECK (poll_interval_sec >= 0),
  -- Adapter carry: etag / lastModified / lastContentHash / lastRemoteRef.
  state jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (jsonb_typeof(state) = 'object'),
  health text NOT NULL DEFAULT 'active'
    CHECK (health IN ('active','backoff','error','paused','unsupported')),
  last_error text,
  last_polled_at timestamptz,
  next_poll_at timestamptz,
  consecutive_failures integer NOT NULL DEFAULT 0 CHECK (consecutive_failures >= 0),
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (workspace_id, id),
  UNIQUE (workspace_id, kind, params_hash)
);

CREATE TRIGGER trg_intake_sources_ws_immutable BEFORE UPDATE ON intake_sources
  FOR EACH ROW EXECUTE FUNCTION reject_workspace_change();

ALTER TABLE intake_sources ENABLE ROW LEVEL SECURITY;
ALTER TABLE intake_sources FORCE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON intake_sources
  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 intake_sources TO sp_app;

-- Due-source scan (the scheduler's loop): workspace + next_poll_at with the
-- non-pollable rows partial-indexed away.
CREATE INDEX intake_sources_due_idx
  ON intake_sources (workspace_id, next_poll_at)
  WHERE health NOT IN ('paused','unsupported');

-- Listing / "which sources feed this item".
CREATE INDEX intake_sources_kind_idx ON intake_sources (workspace_id, kind);

-- ---------------------------------------------------------------------------
-- intake_items
-- ---------------------------------------------------------------------------
CREATE TABLE intake_items (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE ON UPDATE RESTRICT,
  source_id uuid NOT NULL,
  -- Platform-native id for the source (guid, status id, yt:video:…, tg chat:id,
  -- url#hash for web_watch diffs).
  remote_ref text NOT NULL CHECK (btrim(remote_ref) <> '' AND length(remote_ref) <= 2048),
  -- sha256(NFKC + whitespace-collapsed AST plain text + sorted media origin
  -- identity). THE dedup key: UNIQUE (workspace_id, content_hash).
  content_hash text NOT NULL CHECK (content_hash ~ '^[0-9a-f]{64}$'),
  published_at timestamptz NOT NULL,
  -- Flat text derived from `content` at ingest (astToPlainText) — the FTS and
  -- pg_trgm corpus. `text` is non-reserved; kept as the specced column name.
  text text NOT NULL DEFAULT '',
  -- P0c safe-content rich-text AST (doc/paragraph/inline). NEVER raw HTML.
  content jsonb NOT NULL CHECK (jsonb_typeof(content) = 'object'),
  -- [{kind, proxyUrl}…] for pull adapters (feed-normalize proxied media) or
  -- [{kind, storageKey, …}] for the Telegram archive bridge (content-addressed
  -- S3 keys already harvested by apps/api/src/archive).
  media_refs jsonb NOT NULL DEFAULT '[]'::jsonb CHECK (jsonb_typeof(media_refs) = 'array'),
  -- {authorHandle, authorRef, url, title, provider, …kind extras}
  attribution jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (jsonb_typeof(attribution) = 'object'),
  fetched_at timestamptz NOT NULL DEFAULT now(),
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (id, workspace_id),
  -- Content-addressed dedup (spec §2.4 layer 2): first writer wins, every
  -- later delivery of the same content no-ops.
  UNIQUE (workspace_id, content_hash),
  FOREIGN KEY (source_id, workspace_id) REFERENCES intake_sources (id, workspace_id)
    ON DELETE CASCADE ON UPDATE RESTRICT
);

-- Maintained by trigger below; declared here so the column exists for the
-- index + trigger DDL in one place.
ALTER TABLE intake_items ADD COLUMN search_vector tsvector NOT NULL DEFAULT '';

CREATE TRIGGER trg_intake_items_ws_immutable BEFORE UPDATE ON intake_items
  FOR EACH ROW EXECUTE FUNCTION reject_workspace_change();

ALTER TABLE intake_items ENABLE ROW LEVEL SECURITY;
ALTER TABLE intake_items FORCE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON intake_items
  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 intake_items TO sp_app;

-- Per-source listing (GET /intake/items?sourceId=) keyset order.
CREATE INDEX intake_items_source_idx
  ON intake_items (workspace_id, source_id, published_at DESC, id DESC);

-- Whole-workspace newest-first (GET /intake/items without a source filter).
CREATE INDEX intake_items_time_idx
  ON intake_items (workspace_id, published_at DESC, id DESC);

-- ---------------------------------------------------------------------------
-- FTS (025 pattern)
-- ---------------------------------------------------------------------------
-- String-valued attribution keys worth indexing as metadata.
CREATE FUNCTION intake_attribution_text(attribution jsonb)
RETURNS text LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
  SELECT coalesce(string_agg(value, ' '), '')
    FROM jsonb_each_text(attribution)
   WHERE key IN ('authorHandle','authorRef','title','channelTitle','channelUsername',
                 'author','authorSignature','forwardFromChat','forwardFromUser','handle')
$$;

CREATE FUNCTION intake_items_search_vector() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  NEW.search_vector :=
      setweight(to_tsvector('simple', coalesce(NEW.attribution->>'title', '')), 'A')
   || setweight(to_tsvector('simple', intake_attribution_text(NEW.attribution)), 'B')
   || to_tsvector('simple', coalesce(NEW.text, ''));
  RETURN NEW;
END $$;

CREATE TRIGGER trg_intake_items_search
  BEFORE INSERT OR UPDATE OF text, attribution
  ON intake_items FOR EACH ROW
  EXECUTE FUNCTION intake_items_search_vector();

-- Exact/lexical path. fastupdate stays on explicitly: GIN insert cost is the
-- watch item at firehose write rates (research §2.7).
CREATE INDEX intake_items_fts ON intake_items USING gin (search_vector) WITH (fastupdate = on);

-- Fuzzy path (typo tolerance) + ILIKE support on the text corpus.
CREATE INDEX intake_items_text_trgm ON intake_items USING gin (text gin_trgm_ops);