0pt/sql/007_ingestion_plane.sql

-- 0pt ingestion plane (P060-P068).
--
-- Adds the durable state the Source Registry and the ingestion runner need
-- and that sql/002_source_registry.sql did NOT carry:
--
--   * opt_source_authorization  -- P060 "authorization": the recorded decision
--     that this tenant may use this source at all. Separate from rights
--     (license/purpose) on purpose: rights say WHAT the data may be used for,
--     authorization says WHO signed off that 0pt may connect. Both are
--     required to activate.
--   * opt_source_schedule       -- P060 "schedule": ingestion mode + cadence.
--   * opt_source_schemas        -- P060 "schema" + P063 schema evolution:
--     one row per (source_id, schema_version), plus the compatibility policy
--     that governs the NEXT version.
--   * opt_ingestion_checkpoints -- P060 "checkpoint" + P062: the durable
--     resume point per (source_id, partition_key). This is the table that
--     makes "restart resumes without silent gaps" true rather than aspirational.
--   * opt_ingestion_ledger      -- P068: idempotency ledger. One row per
--     (source_id, record_key, payload hash); a repeated backfill that
--     re-delivers the same record is recognised, not re-emitted.
--   * opt_quality_quarantine    -- P064: where envelopes go when the declared
--     enforcement policy says "quarantine" instead of "block".
--   * opt_lineage_edges         -- P067: the LOCAL, QUERYABLE projection of
--     the OpenLineage events 0pt already emits to AUD0. AUD0 remains the
--     system of record; this table exists because "emitted to AUD0" is not
--     the same thing as "queryable from 0pt", and the P067 DoD asks for the
--     latter.
--
-- Every statement is IF NOT EXISTS / idempotent so it can be applied to an
-- already-migrated database, matching sql/002_source_registry.sql's contract
-- (tests/p0/*.realinfra.test.ts apply the files themselves).
--
-- Ordering: requires sql/002_source_registry.sql (opt_sources) to have run.

-- ── P060 owner + the pre-activation status ────────────────────────────────
-- `created_by` (002) records who typed the registration; `owner_id` records
-- who is ACCOUNTABLE for the source afterwards. The activation gate requires
-- the latter, so they cannot be collapsed.
ALTER TABLE opt_sources ADD COLUMN IF NOT EXISTS owner_id TEXT;
CREATE INDEX IF NOT EXISTS opt_sources_owner_idx ON opt_sources (owner_id);

-- New sources are created 'pending' (see src/repositories/*-source-repository.ts).
-- The column default in 002 is 'active'; it is changed here so a plain INSERT
-- that omits `status` -- from any client, including psql -- cannot produce an
-- unclassified-but-active source. Rows that are already 'active' are left
-- exactly as they are: this migration does not retroactively deactivate
-- anything an operator has already approved.
ALTER TABLE opt_sources ALTER COLUMN status SET DEFAULT 'pending';

-- ── P060 authorization ────────────────────────────────────────────────────
-- One row per source. `basis` is the legal/contractual ground; `granted_by`
-- is the accountable actor id (never "system"); `evidence_ref` points at the
-- AUD0 evidence or 1p/contract artifact. `expires_at` is honoured by the
-- activation gate -- an expired authorization does not classify a source.
CREATE TABLE IF NOT EXISTS opt_source_authorization (
  source_id TEXT PRIMARY KEY REFERENCES opt_sources (source_id),
  tenant_id TEXT NOT NULL,
  basis TEXT NOT NULL,            -- contract | public_register | consent | legitimate_interest | internal
  granted_by TEXT NOT NULL,       -- accountable actor id
  granted_at TIMESTAMPTZ NOT NULL,
  expires_at TIMESTAMPTZ,
  evidence_ref TEXT,              -- aud0 evidence ref / contract artifact ref
  notes TEXT,
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS opt_source_authorization_tenant_idx ON opt_source_authorization (tenant_id);

-- ── P060 schedule ─────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS opt_source_schedule (
  source_id TEXT PRIMARY KEY REFERENCES opt_sources (source_id),
  tenant_id TEXT NOT NULL,
  mode TEXT NOT NULL,             -- batch | stream | cdc
  interval_seconds INT,           -- batch/cdc poll cadence; NULL for continuous streams
  cron TEXT,                      -- optional cron expression (documented, not executed here)
  enabled BOOLEAN NOT NULL DEFAULT false,
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS opt_source_schedule_tenant_idx ON opt_source_schedule (tenant_id);

-- ── P060 schema + P063 schema evolution ───────────────────────────────────
-- `fields` is the record schema as JSON: {"<field>": {"type": "...",
-- "required": true|false}}. `compatibility` governs what the NEXT registered
-- version is allowed to change (see src/schema/compatibility.ts).
CREATE TABLE IF NOT EXISTS opt_source_schemas (
  source_id TEXT NOT NULL REFERENCES opt_sources (source_id),
  schema_version INT NOT NULL,
  tenant_id TEXT NOT NULL,
  fields JSONB NOT NULL,
  compatibility TEXT NOT NULL,    -- backward | forward | full | none
  registered_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  registered_by TEXT,
  PRIMARY KEY (source_id, schema_version)
);
CREATE INDEX IF NOT EXISTS opt_source_schemas_tenant_idx ON opt_source_schemas (tenant_id);

-- ── P060 checkpoint + P062 durable resume ─────────────────────────────────
-- `cursor` is opaque to 0pt -- the connector defines it (an offset, an LSN,
-- a page token). `sequence` is 0pt's own monotonic envelope counter for the
-- partition, used to detect gaps on resume.
CREATE TABLE IF NOT EXISTS opt_ingestion_checkpoints (
  source_id TEXT NOT NULL REFERENCES opt_sources (source_id),
  partition_key TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  cursor TEXT,
  sequence BIGINT NOT NULL DEFAULT 0,
  records_committed BIGINT NOT NULL DEFAULT 0,
  committed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (source_id, partition_key)
);
CREATE INDEX IF NOT EXISTS opt_ingestion_checkpoints_tenant_idx ON opt_ingestion_checkpoints (tenant_id);

-- ── P068 backfill idempotency ledger ──────────────────────────────────────
-- A backfill/replay that re-delivers a record 0pt has already accepted must
-- be a no-op, not a duplicate. `payload_hash` distinguishes "same record,
-- same content" (skip) from "same record, new content" (an update, emitted).
CREATE TABLE IF NOT EXISTS opt_ingestion_ledger (
  source_id TEXT NOT NULL REFERENCES opt_sources (source_id),
  record_key TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  payload_hash TEXT NOT NULL,
  first_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  delivery_count BIGINT NOT NULL DEFAULT 1,
  PRIMARY KEY (source_id, record_key)
);
CREATE INDEX IF NOT EXISTS opt_ingestion_ledger_tenant_idx ON opt_ingestion_ledger (tenant_id);

-- ── P064 quarantine ───────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS opt_quality_quarantine (
  quarantine_id TEXT PRIMARY KEY,
  source_id TEXT NOT NULL REFERENCES opt_sources (source_id),
  tenant_id TEXT NOT NULL,
  run_id TEXT NOT NULL,
  record_key TEXT NOT NULL,
  violations JSONB NOT NULL,      -- [{expectation_type, kwargs, detail}]
  envelope JSONB NOT NULL,
  quarantined_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS opt_quality_quarantine_source_idx ON opt_quality_quarantine (source_id, quarantined_at DESC);
CREATE INDEX IF NOT EXISTS opt_quality_quarantine_tenant_idx ON opt_quality_quarantine (tenant_id);

-- ── P067 queryable lineage projection ─────────────────────────────────────
-- One row per (run, input, output) edge. `edge_id` is deterministic
-- (sha256 of the tuple) so a re-emitted idempotent run does not duplicate
-- edges -- the same dedup property LineageEmitter relies on in AUD0.
CREATE TABLE IF NOT EXISTS opt_lineage_edges (
  edge_id TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  run_id TEXT NOT NULL,
  capability_id TEXT NOT NULL,
  input_namespace TEXT NOT NULL,
  input_name TEXT NOT NULL,
  input_ref TEXT,
  output_namespace TEXT NOT NULL,
  output_name TEXT NOT NULL,
  output_ref TEXT,
  event_time TIMESTAMPTZ NOT NULL,
  correlation_id TEXT
);
CREATE INDEX IF NOT EXISTS opt_lineage_edges_output_idx ON opt_lineage_edges (tenant_id, output_name);
CREATE INDEX IF NOT EXISTS opt_lineage_edges_input_idx ON opt_lineage_edges (tenant_id, input_name);
CREATE INDEX IF NOT EXISTS opt_lineage_edges_run_idx ON opt_lineage_edges (tenant_id, run_id);