163 lines
9.0 KiB
SQL
163 lines
9.0 KiB
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);
|