stella-ops.org/git.stella-ops.org
1
0
Fork 0
Code Issues Pull Requests Actions 2 Releases Wiki Activity

ui progressing

Browse Source
This commit is contained in:
master
2026-02-20 23:32:20 +02:00
parent ca5e7888d6
commit 1ec797d5e8
191 changed files with 32771 additions and 6504 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
devops/compose/postgres-init/01-create-schemas.sql
View File

@@ -12,3 +12,6 @@ CREATE SCHEMA IF NOT EXISTS findings;
CREATE SCHEMA IF NOT EXISTS timeline;
CREATE SCHEMA IF NOT EXISTS doctor;
CREATE SCHEMA IF NOT EXISTS issuer_directory;
CREATE SCHEMA IF NOT EXISTS analytics;
CREATE SCHEMA IF NOT EXISTS scheduler_app;
CREATE SCHEMA IF NOT EXISTS findings_ledger_app;

565
devops/compose/postgres-init/02-findings-ledger-tables.sql Normal file
View File

@@ -0,0 +1,565 @@
-- Findings Ledger: Consolidated init from migrations 001-009
-- Auto-generated for docker-compose postgres-init
-- Creates all tables required by stellaops-findings-ledger-web
-- ============================================================================
-- 001_initial.sql - Bootstrap schema (LEDGER-29-001)
-- ============================================================================
BEGIN;
CREATE TYPE ledger_event_type AS ENUM (
'finding.created',
'finding.status_changed',
'finding.severity_changed',
'finding.tag_updated',
'finding.comment_added',
'finding.assignment_changed',
'finding.accepted_risk',
'finding.remediation_plan_added',
'finding.attachment_added',
'finding.closed'
);
CREATE TYPE ledger_action_type AS ENUM (
'assign',
'comment',
'attach_evidence',
'link_ticket',
'remediation_plan',
'status_change',
'accept_risk',
'reopen',
'close'
);
CREATE TABLE ledger_events (
tenant_id TEXT NOT NULL,
chain_id UUID NOT NULL,
sequence_no BIGINT NOT NULL,
event_id UUID NOT NULL,
event_type ledger_event_type NOT NULL,
policy_version TEXT NOT NULL,
finding_id TEXT NOT NULL,
artifact_id TEXT NOT NULL,
source_run_id UUID,
actor_id TEXT NOT NULL,
actor_type TEXT NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
event_body JSONB NOT NULL,
event_hash CHAR(64) NOT NULL,
previous_hash CHAR(64) NOT NULL,
merkle_leaf_hash CHAR(64) NOT NULL,
CONSTRAINT pk_ledger_events PRIMARY KEY (tenant_id, chain_id, sequence_no),
CONSTRAINT uq_ledger_events_event_id UNIQUE (tenant_id, event_id),
CONSTRAINT uq_ledger_events_chain_hash UNIQUE (tenant_id, chain_id, event_hash),
CONSTRAINT ck_ledger_events_event_hash_hex CHECK (event_hash ~ '^[0-9a-f]{64}$'),
CONSTRAINT ck_ledger_events_previous_hash_hex CHECK (previous_hash ~ '^[0-9a-f]{64}$'),
CONSTRAINT ck_ledger_events_leaf_hash_hex CHECK (merkle_leaf_hash ~ '^[0-9a-f]{64}$'),
CONSTRAINT ck_ledger_events_actor_type CHECK (actor_type IN ('system', 'operator', 'integration'))
) PARTITION BY LIST (tenant_id);
CREATE TABLE ledger_events_default PARTITION OF ledger_events DEFAULT;
CREATE INDEX ix_ledger_events_finding ON ledger_events (tenant_id, finding_id, policy_version);
CREATE INDEX ix_ledger_events_type ON ledger_events (tenant_id, event_type, recorded_at DESC);
CREATE INDEX ix_ledger_events_recorded_at ON ledger_events (tenant_id, recorded_at DESC);
CREATE TABLE ledger_merkle_roots (
tenant_id TEXT NOT NULL,
anchor_id UUID NOT NULL,
window_start TIMESTAMPTZ NOT NULL,
window_end TIMESTAMPTZ NOT NULL,
sequence_start BIGINT NOT NULL,
sequence_end BIGINT NOT NULL,
root_hash CHAR(64) NOT NULL,
leaf_count INTEGER NOT NULL,
anchored_at TIMESTAMPTZ NOT NULL,
anchor_reference TEXT,
CONSTRAINT pk_ledger_merkle_roots PRIMARY KEY (tenant_id, anchor_id),
CONSTRAINT uq_ledger_merkle_root_hash UNIQUE (tenant_id, root_hash),
CONSTRAINT ck_ledger_merkle_root_hash_hex CHECK (root_hash ~ '^[0-9a-f]{64}$')
) PARTITION BY LIST (tenant_id);
CREATE TABLE ledger_merkle_roots_default PARTITION OF ledger_merkle_roots DEFAULT;
CREATE INDEX ix_merkle_sequences ON ledger_merkle_roots (tenant_id, sequence_end DESC);
CREATE TABLE findings_projection (
tenant_id TEXT NOT NULL,
finding_id TEXT NOT NULL,
policy_version TEXT NOT NULL,
status TEXT NOT NULL,
severity NUMERIC(6,3),
labels JSONB NOT NULL DEFAULT '{}'::JSONB,
current_event_id UUID NOT NULL,
explain_ref TEXT,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
cycle_hash CHAR(64) NOT NULL,
CONSTRAINT pk_findings_projection PRIMARY KEY (tenant_id, finding_id, policy_version),
CONSTRAINT ck_findings_projection_cycle_hash_hex CHECK (cycle_hash ~ '^[0-9a-f]{64}$')
) PARTITION BY LIST (tenant_id);
CREATE TABLE findings_projection_default PARTITION OF findings_projection DEFAULT;
CREATE INDEX ix_projection_status ON findings_projection (tenant_id, status, severity DESC);
CREATE INDEX ix_projection_labels_gin ON findings_projection USING GIN (labels JSONB_PATH_OPS);
CREATE TABLE finding_history (
tenant_id TEXT NOT NULL,
finding_id TEXT NOT NULL,
policy_version TEXT NOT NULL,
event_id UUID NOT NULL,
status TEXT NOT NULL,
severity NUMERIC(6,3),
actor_id TEXT NOT NULL,
comment TEXT,
occurred_at TIMESTAMPTZ NOT NULL,
CONSTRAINT pk_finding_history PRIMARY KEY (tenant_id, finding_id, event_id)
) PARTITION BY LIST (tenant_id);
CREATE TABLE finding_history_default PARTITION OF finding_history DEFAULT;
CREATE INDEX ix_finding_history_timeline ON finding_history (tenant_id, finding_id, occurred_at DESC);
CREATE TABLE triage_actions (
tenant_id TEXT NOT NULL,
action_id UUID NOT NULL,
event_id UUID NOT NULL,
finding_id TEXT NOT NULL,
action_type ledger_action_type NOT NULL,
payload JSONB NOT NULL DEFAULT '{}'::JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_by TEXT NOT NULL,
CONSTRAINT pk_triage_actions PRIMARY KEY (tenant_id, action_id)
) PARTITION BY LIST (tenant_id);
CREATE TABLE triage_actions_default PARTITION OF triage_actions DEFAULT;
CREATE INDEX ix_triage_actions_event ON triage_actions (tenant_id, event_id);
CREATE INDEX ix_triage_actions_created_at ON triage_actions (tenant_id, created_at DESC);
COMMIT;
-- ============================================================================
-- 002_projection_offsets.sql - Projection worker checkpoints (LEDGER-29-003)
-- ============================================================================
BEGIN;
CREATE TABLE IF NOT EXISTS ledger_projection_offsets (
worker_id TEXT NOT NULL PRIMARY KEY,
last_recorded_at TIMESTAMPTZ NOT NULL,
last_event_id UUID NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
);
INSERT INTO ledger_projection_offsets (worker_id, last_recorded_at, last_event_id, updated_at)
VALUES (
'default',
'1970-01-01T00:00:00Z',
'00000000-0000-0000-0000-000000000000',
NOW())
ON CONFLICT (worker_id) DO NOTHING;
COMMIT;
-- ============================================================================
-- 002_add_evidence_bundle_ref.sql - Evidence bundle references (LEDGER-OBS-53-001)
-- ============================================================================
ALTER TABLE ledger_events
ADD COLUMN evidence_bundle_ref text NULL;
CREATE INDEX IF NOT EXISTS ix_ledger_events_finding_evidence_ref
ON ledger_events (tenant_id, finding_id, recorded_at DESC)
WHERE evidence_bundle_ref IS NOT NULL;
-- ============================================================================
-- 003_policy_rationale.sql - Policy rationale column (LEDGER-29-004)
-- ============================================================================
BEGIN;
ALTER TABLE findings_projection
ADD COLUMN IF NOT EXISTS policy_rationale JSONB NOT NULL DEFAULT '[]'::JSONB;
ALTER TABLE findings_projection
ALTER COLUMN policy_rationale SET DEFAULT '[]'::JSONB;
UPDATE findings_projection
SET policy_rationale = '[]'::JSONB
WHERE policy_rationale IS NULL;
COMMIT;
-- ============================================================================
-- 004_ledger_attestations.sql - Attestation verification exports (LEDGER-OBS-54-001)
-- ============================================================================
BEGIN;
CREATE TABLE IF NOT EXISTS ledger_attestations (
tenant_id text NOT NULL,
attestation_id uuid NOT NULL,
artifact_id text NOT NULL,
finding_id text NULL,
verification_status text NOT NULL,
verification_time timestamptz NOT NULL,
dsse_digest text NOT NULL,
rekor_entry_id text NULL,
evidence_bundle_ref text NULL,
ledger_event_id uuid NOT NULL,
recorded_at timestamptz NOT NULL,
merkle_leaf_hash text NOT NULL,
root_hash text NOT NULL,
cycle_hash text NOT NULL,
projection_version text NOT NULL
);
ALTER TABLE ledger_attestations
ADD CONSTRAINT pk_ledger_attestations PRIMARY KEY (tenant_id, attestation_id);
CREATE INDEX IF NOT EXISTS ix_ledger_attestations_recorded
ON ledger_attestations (tenant_id, recorded_at, attestation_id);
CREATE INDEX IF NOT EXISTS ix_ledger_attestations_artifact
ON ledger_attestations (tenant_id, artifact_id, recorded_at DESC);
CREATE INDEX IF NOT EXISTS ix_ledger_attestations_finding
ON ledger_attestations (tenant_id, finding_id, recorded_at DESC)
WHERE finding_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_ledger_attestations_status
ON ledger_attestations (tenant_id, verification_status, recorded_at DESC);
COMMIT;
-- ============================================================================
-- 004_risk_fields.sql - Risk scoring fields (LEDGER-RISK-66-001/002)
-- ============================================================================
BEGIN;
ALTER TABLE findings_projection
ADD COLUMN IF NOT EXISTS risk_score NUMERIC(6,3),
ADD COLUMN IF NOT EXISTS risk_severity TEXT,
ADD COLUMN IF NOT EXISTS risk_profile_version TEXT,
ADD COLUMN IF NOT EXISTS risk_explanation_id UUID,
ADD COLUMN IF NOT EXISTS risk_event_sequence BIGINT;
CREATE INDEX IF NOT EXISTS ix_projection_risk ON findings_projection (tenant_id, risk_severity, risk_score DESC);
COMMIT;
-- ============================================================================
-- 005_risk_fields.sql - Risk scoring fields idempotent re-add (LEDGER-RISK-66-001)
-- ============================================================================
BEGIN;
ALTER TABLE findings_projection
ADD COLUMN IF NOT EXISTS risk_score numeric(6,2) NULL,
ADD COLUMN IF NOT EXISTS risk_severity text NULL,
ADD COLUMN IF NOT EXISTS risk_profile_version text NULL,
ADD COLUMN IF NOT EXISTS risk_explanation_id text NULL,
ADD COLUMN IF NOT EXISTS risk_event_sequence bigint NULL;
CREATE INDEX IF NOT EXISTS ix_findings_projection_risk
ON findings_projection (tenant_id, risk_severity, risk_score DESC);
COMMIT;
-- ============================================================================
-- 006_orchestrator_airgap.sql - Export and import provenance (LEDGER-34-101, LEDGER-AIRGAP-56-001)
-- ============================================================================
BEGIN;
CREATE TABLE IF NOT EXISTS orchestrator_exports
(
tenant_id TEXT NOT NULL,
run_id UUID NOT NULL,
job_type TEXT NOT NULL,
artifact_hash TEXT NOT NULL,
policy_hash TEXT NOT NULL,
started_at TIMESTAMPTZ NOT NULL,
completed_at TIMESTAMPTZ,
status TEXT NOT NULL,
manifest_path TEXT,
logs_path TEXT,
merkle_root CHAR(64) NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (tenant_id, run_id)
);
CREATE UNIQUE INDEX IF NOT EXISTS ix_orchestrator_exports_artifact_run
ON orchestrator_exports (tenant_id, artifact_hash, run_id);
CREATE INDEX IF NOT EXISTS ix_orchestrator_exports_artifact
ON orchestrator_exports (tenant_id, artifact_hash);
CREATE TABLE IF NOT EXISTS airgap_imports
(
tenant_id TEXT NOT NULL,
bundle_id TEXT NOT NULL,
mirror_generation TEXT,
merkle_root TEXT NOT NULL,
time_anchor TIMESTAMPTZ NOT NULL,
publisher TEXT,
hash_algorithm TEXT,
contents JSONB,
imported_at TIMESTAMPTZ NOT NULL,
import_operator TEXT,
ledger_event_id UUID,
PRIMARY KEY (tenant_id, bundle_id, time_anchor)
);
CREATE INDEX IF NOT EXISTS ix_airgap_imports_bundle
ON airgap_imports (tenant_id, bundle_id);
CREATE INDEX IF NOT EXISTS ix_airgap_imports_event
ON airgap_imports (tenant_id, ledger_event_id);
COMMIT;
-- ============================================================================
-- 007_enable_rls.sql - Row-Level Security (LEDGER-TEN-48-001-DEV)
-- ============================================================================
BEGIN;
CREATE SCHEMA IF NOT EXISTS findings_ledger_app;
CREATE OR REPLACE FUNCTION findings_ledger_app.require_current_tenant()
RETURNS TEXT
LANGUAGE plpgsql
STABLE
AS $$
DECLARE
tenant_text TEXT;
BEGIN
tenant_text := current_setting('app.current_tenant', true);
IF tenant_text IS NULL OR length(trim(tenant_text)) = 0 THEN
RAISE EXCEPTION 'app.current_tenant is not set for the current session'
USING ERRCODE = 'P0001';
END IF;
RETURN tenant_text;
END;
$$;
COMMENT ON FUNCTION findings_ledger_app.require_current_tenant() IS
'Returns the current tenant ID from session variable, raises exception if not set';
ALTER TABLE ledger_events ENABLE ROW LEVEL SECURITY;
ALTER TABLE ledger_events FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS ledger_events_tenant_isolation ON ledger_events;
CREATE POLICY ledger_events_tenant_isolation
ON ledger_events
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE ledger_merkle_roots ENABLE ROW LEVEL SECURITY;
ALTER TABLE ledger_merkle_roots FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS ledger_merkle_roots_tenant_isolation ON ledger_merkle_roots;
CREATE POLICY ledger_merkle_roots_tenant_isolation
ON ledger_merkle_roots
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE findings_projection ENABLE ROW LEVEL SECURITY;
ALTER TABLE findings_projection FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS findings_projection_tenant_isolation ON findings_projection;
CREATE POLICY findings_projection_tenant_isolation
ON findings_projection
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE finding_history ENABLE ROW LEVEL SECURITY;
ALTER TABLE finding_history FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS finding_history_tenant_isolation ON finding_history;
CREATE POLICY finding_history_tenant_isolation
ON finding_history
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE triage_actions ENABLE ROW LEVEL SECURITY;
ALTER TABLE triage_actions FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS triage_actions_tenant_isolation ON triage_actions;
CREATE POLICY triage_actions_tenant_isolation
ON triage_actions
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE ledger_attestations ENABLE ROW LEVEL SECURITY;
ALTER TABLE ledger_attestations FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS ledger_attestations_tenant_isolation ON ledger_attestations;
CREATE POLICY ledger_attestations_tenant_isolation
ON ledger_attestations
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE orchestrator_exports ENABLE ROW LEVEL SECURITY;
ALTER TABLE orchestrator_exports FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS orchestrator_exports_tenant_isolation ON orchestrator_exports;
CREATE POLICY orchestrator_exports_tenant_isolation
ON orchestrator_exports
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
ALTER TABLE airgap_imports ENABLE ROW LEVEL SECURITY;
ALTER TABLE airgap_imports FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS airgap_imports_tenant_isolation ON airgap_imports;
CREATE POLICY airgap_imports_tenant_isolation
ON airgap_imports
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'findings_ledger_admin') THEN
CREATE ROLE findings_ledger_admin NOLOGIN BYPASSRLS;
END IF;
END;
$$;
COMMENT ON ROLE findings_ledger_admin IS
'Admin role that bypasses RLS for migrations and cross-tenant operations';
COMMIT;
-- ============================================================================
-- 008_attestation_pointers.sql - Finding-to-attestation pointers (LEDGER-ATTEST-73-001)
-- ============================================================================
BEGIN;
CREATE TABLE IF NOT EXISTS ledger_attestation_pointers (
tenant_id text NOT NULL,
pointer_id uuid NOT NULL,
finding_id text NOT NULL,
attestation_type text NOT NULL,
relationship text NOT NULL,
attestation_ref jsonb NOT NULL,
verification_result jsonb NULL,
created_at timestamptz NOT NULL,
created_by text NOT NULL,
metadata jsonb NULL,
ledger_event_id uuid NULL
);
ALTER TABLE ledger_attestation_pointers
ADD CONSTRAINT pk_ledger_attestation_pointers PRIMARY KEY (tenant_id, pointer_id);
CREATE INDEX IF NOT EXISTS ix_ledger_attestation_pointers_finding
ON ledger_attestation_pointers (tenant_id, finding_id, created_at DESC);
CREATE INDEX IF NOT EXISTS ix_ledger_attestation_pointers_digest
ON ledger_attestation_pointers (tenant_id, (attestation_ref->>'digest'));
CREATE INDEX IF NOT EXISTS ix_ledger_attestation_pointers_type
ON ledger_attestation_pointers (tenant_id, attestation_type, created_at DESC);
CREATE INDEX IF NOT EXISTS ix_ledger_attestation_pointers_verified
ON ledger_attestation_pointers (tenant_id, ((verification_result->>'verified')::boolean))
WHERE verification_result IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_ledger_attestation_pointers_signer
ON ledger_attestation_pointers (tenant_id, (attestation_ref->'signer_info'->>'subject'))
WHERE attestation_ref->'signer_info' IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_ledger_attestation_pointers_predicate
ON ledger_attestation_pointers (tenant_id, (attestation_ref->>'predicate_type'))
WHERE attestation_ref->>'predicate_type' IS NOT NULL;
ALTER TABLE ledger_attestation_pointers ENABLE ROW LEVEL SECURITY;
ALTER TABLE ledger_attestation_pointers FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS ledger_attestation_pointers_tenant_isolation ON ledger_attestation_pointers;
CREATE POLICY ledger_attestation_pointers_tenant_isolation
ON ledger_attestation_pointers
FOR ALL
USING (tenant_id = findings_ledger_app.require_current_tenant())
WITH CHECK (tenant_id = findings_ledger_app.require_current_tenant());
COMMENT ON TABLE ledger_attestation_pointers IS
'Links findings to verification reports and attestation envelopes for explainability (LEDGER-ATTEST-73-001)';
COMMIT;
-- ============================================================================
-- 009_snapshots.sql - Ledger snapshots for time-travel
-- ============================================================================
CREATE TABLE IF NOT EXISTS ledger_snapshots (
tenant_id TEXT NOT NULL,
snapshot_id UUID NOT NULL,
label TEXT,
description TEXT,
status TEXT NOT NULL DEFAULT 'Creating',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ,
expires_at TIMESTAMPTZ,
sequence_number BIGINT NOT NULL,
snapshot_timestamp TIMESTAMPTZ NOT NULL,
findings_count BIGINT NOT NULL DEFAULT 0,
vex_statements_count BIGINT NOT NULL DEFAULT 0,
advisories_count BIGINT NOT NULL DEFAULT 0,
sboms_count BIGINT NOT NULL DEFAULT 0,
events_count BIGINT NOT NULL DEFAULT 0,
size_bytes BIGINT NOT NULL DEFAULT 0,
merkle_root TEXT,
dsse_digest TEXT,
metadata JSONB,
include_entity_types JSONB,
sign_requested BOOLEAN NOT NULL DEFAULT FALSE,
PRIMARY KEY (tenant_id, snapshot_id)
);
CREATE INDEX IF NOT EXISTS idx_ledger_snapshots_status
ON ledger_snapshots (tenant_id, status, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_ledger_snapshots_expires
ON ledger_snapshots (expires_at)
WHERE expires_at IS NOT NULL AND status = 'Available';
CREATE INDEX IF NOT EXISTS idx_ledger_snapshots_sequence
ON ledger_snapshots (tenant_id, sequence_number);
CREATE INDEX IF NOT EXISTS idx_ledger_snapshots_label
ON ledger_snapshots (tenant_id, label)
WHERE label IS NOT NULL;
ALTER TABLE ledger_snapshots ENABLE ROW LEVEL SECURITY;
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_policies
WHERE tablename = 'ledger_snapshots'
AND policyname = 'ledger_snapshots_tenant_isolation'
) THEN
CREATE POLICY ledger_snapshots_tenant_isolation ON ledger_snapshots
USING (tenant_id = current_setting('app.tenant_id', true))
WITH CHECK (tenant_id = current_setting('app.tenant_id', true));
END IF;
END $$;
COMMENT ON TABLE ledger_snapshots IS 'Point-in-time snapshots of ledger state for time-travel queries';

690
devops/compose/postgres-init/03-scheduler-tables.sql Normal file
View File

@@ -0,0 +1,690 @@
-- Scheduler: Consolidated init from migrations 001-003
-- Auto-generated for docker-compose postgres-init
-- Creates all tables required by stellaops-scheduler-worker
-- ============================================================================
-- 001_initial_schema.sql - Complete scheduler schema
-- ============================================================================
CREATE SCHEMA IF NOT EXISTS scheduler;
CREATE SCHEMA IF NOT EXISTS scheduler_app;
-- Enum types
DO $$ BEGIN
CREATE TYPE scheduler.job_status AS ENUM (
'pending', 'scheduled', 'leased', 'running',
'succeeded', 'failed', 'canceled', 'timed_out'
);
EXCEPTION WHEN duplicate_object THEN null; END $$;
DO $$ BEGIN
CREATE TYPE scheduler.graph_job_type AS ENUM ('build', 'overlay');
EXCEPTION WHEN duplicate_object THEN NULL; END $$;
DO $$ BEGIN
CREATE TYPE scheduler.graph_job_status AS ENUM ('pending', 'queued', 'running', 'completed', 'failed', 'canceled');
EXCEPTION WHEN duplicate_object THEN NULL; END $$;
DO $$ BEGIN
CREATE TYPE scheduler.run_state AS ENUM ('planning','queued','running','completed','error','cancelled');
EXCEPTION WHEN duplicate_object THEN NULL; END $$;
DO $$ BEGIN
CREATE TYPE scheduler.policy_run_status AS ENUM ('pending','submitted','retrying','failed','completed','cancelled');
EXCEPTION WHEN duplicate_object THEN NULL; END $$;
-- Helper functions
CREATE OR REPLACE FUNCTION scheduler.update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = NOW();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE OR REPLACE FUNCTION scheduler_app.require_current_tenant()
RETURNS TEXT
LANGUAGE plpgsql STABLE SECURITY DEFINER
AS $$
DECLARE
v_tenant TEXT;
BEGIN
v_tenant := current_setting('app.tenant_id', true);
IF v_tenant IS NULL OR v_tenant = '' THEN
RAISE EXCEPTION 'app.tenant_id session variable not set'
USING HINT = 'Set via: SELECT set_config(''app.tenant_id'', ''<tenant>'', false)',
ERRCODE = 'P0001';
END IF;
RETURN v_tenant;
END;
$$;
REVOKE ALL ON FUNCTION scheduler_app.require_current_tenant() FROM PUBLIC;
-- Core tables: jobs
CREATE TABLE IF NOT EXISTS scheduler.jobs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id TEXT NOT NULL,
project_id TEXT,
job_type TEXT NOT NULL,
status scheduler.job_status NOT NULL DEFAULT 'pending',
priority INT NOT NULL DEFAULT 0,
payload JSONB NOT NULL DEFAULT '{}',
payload_digest TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
correlation_id TEXT,
attempt INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 3,
lease_id UUID,
worker_id TEXT,
lease_until TIMESTAMPTZ,
not_before TIMESTAMPTZ,
reason TEXT,
result JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
scheduled_at TIMESTAMPTZ,
leased_at TIMESTAMPTZ,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
created_by TEXT,
UNIQUE(tenant_id, idempotency_key)
);
CREATE INDEX IF NOT EXISTS idx_jobs_tenant_status ON scheduler.jobs(tenant_id, status);
CREATE INDEX IF NOT EXISTS idx_jobs_tenant_type ON scheduler.jobs(tenant_id, job_type);
CREATE INDEX IF NOT EXISTS idx_jobs_scheduled ON scheduler.jobs(tenant_id, status, not_before, priority DESC, created_at)
WHERE status = 'scheduled';
CREATE INDEX IF NOT EXISTS idx_jobs_leased ON scheduler.jobs(tenant_id, status, lease_until)
WHERE status = 'leased';
CREATE INDEX IF NOT EXISTS idx_jobs_project ON scheduler.jobs(tenant_id, project_id);
CREATE INDEX IF NOT EXISTS idx_jobs_correlation ON scheduler.jobs(correlation_id);
-- Triggers table (cron-based job triggers)
CREATE TABLE IF NOT EXISTS scheduler.triggers (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id TEXT NOT NULL,
name TEXT NOT NULL,
description TEXT,
job_type TEXT NOT NULL,
job_payload JSONB NOT NULL DEFAULT '{}',
cron_expression TEXT NOT NULL,
timezone TEXT NOT NULL DEFAULT 'UTC',
enabled BOOLEAN NOT NULL DEFAULT TRUE,
next_fire_at TIMESTAMPTZ,
last_fire_at TIMESTAMPTZ,
last_job_id UUID REFERENCES scheduler.jobs(id),
fire_count BIGINT NOT NULL DEFAULT 0,
misfire_count INT NOT NULL DEFAULT 0,
metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_by TEXT,
UNIQUE(tenant_id, name)
);
CREATE INDEX IF NOT EXISTS idx_triggers_tenant_id ON scheduler.triggers(tenant_id);
CREATE INDEX IF NOT EXISTS idx_triggers_next_fire ON scheduler.triggers(enabled, next_fire_at) WHERE enabled = TRUE;
CREATE INDEX IF NOT EXISTS idx_triggers_job_type ON scheduler.triggers(tenant_id, job_type);
CREATE TRIGGER trg_triggers_updated_at
BEFORE UPDATE ON scheduler.triggers
FOR EACH ROW EXECUTE FUNCTION scheduler.update_updated_at();
-- Workers table (global, NOT RLS-protected)
CREATE TABLE IF NOT EXISTS scheduler.workers (
id TEXT PRIMARY KEY,
tenant_id TEXT,
hostname TEXT NOT NULL,
process_id INT,
job_types TEXT[] NOT NULL DEFAULT '{}',
max_concurrent_jobs INT NOT NULL DEFAULT 1,
current_jobs INT NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'draining', 'stopped')),
last_heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
registered_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
metadata JSONB NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS idx_workers_status ON scheduler.workers(status);
CREATE INDEX IF NOT EXISTS idx_workers_heartbeat ON scheduler.workers(last_heartbeat_at);
CREATE INDEX IF NOT EXISTS idx_workers_tenant ON scheduler.workers(tenant_id);
COMMENT ON TABLE scheduler.workers IS 'Global worker registry. Not RLS-protected - workers serve all tenants.';
-- Distributed locks
CREATE TABLE IF NOT EXISTS scheduler.locks (
lock_key TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
holder_id TEXT NOT NULL,
acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
expires_at TIMESTAMPTZ NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS idx_locks_tenant ON scheduler.locks(tenant_id);
CREATE INDEX IF NOT EXISTS idx_locks_expires ON scheduler.locks(expires_at);
-- Job history
CREATE TABLE IF NOT EXISTS scheduler.job_history (
id BIGSERIAL PRIMARY KEY,
job_id UUID NOT NULL,
tenant_id TEXT NOT NULL,
project_id TEXT,
job_type TEXT NOT NULL,
status scheduler.job_status NOT NULL,
attempt INT NOT NULL,
payload_digest TEXT NOT NULL,
result JSONB,
reason TEXT,
worker_id TEXT,
duration_ms BIGINT,
created_at TIMESTAMPTZ NOT NULL,
completed_at TIMESTAMPTZ NOT NULL,
archived_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_job_history_tenant ON scheduler.job_history(tenant_id);
CREATE INDEX IF NOT EXISTS idx_job_history_job_id ON scheduler.job_history(job_id);
CREATE INDEX IF NOT EXISTS idx_job_history_type ON scheduler.job_history(tenant_id, job_type);
CREATE INDEX IF NOT EXISTS idx_job_history_completed ON scheduler.job_history(tenant_id, completed_at);
-- Metrics table
CREATE TABLE IF NOT EXISTS scheduler.metrics (
id BIGSERIAL PRIMARY KEY,
tenant_id TEXT NOT NULL,
job_type TEXT NOT NULL,
period_start TIMESTAMPTZ NOT NULL,
period_end TIMESTAMPTZ NOT NULL,
jobs_created BIGINT NOT NULL DEFAULT 0,
jobs_completed BIGINT NOT NULL DEFAULT 0,
jobs_failed BIGINT NOT NULL DEFAULT 0,
jobs_timed_out BIGINT NOT NULL DEFAULT 0,
avg_duration_ms BIGINT,
p50_duration_ms BIGINT,
p95_duration_ms BIGINT,
p99_duration_ms BIGINT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE(tenant_id, job_type, period_start)
);
CREATE INDEX IF NOT EXISTS idx_metrics_tenant_period ON scheduler.metrics(tenant_id, period_start);
-- Schedules and runs
CREATE TABLE IF NOT EXISTS scheduler.schedules (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
name TEXT NOT NULL,
description TEXT,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
cron_expression TEXT,
timezone TEXT NOT NULL DEFAULT 'UTC',
mode TEXT NOT NULL CHECK (mode IN ('analysisonly', 'contentrefresh')),
selection JSONB NOT NULL DEFAULT '{}',
only_if JSONB NOT NULL DEFAULT '{}',
notify JSONB NOT NULL DEFAULT '{}',
limits JSONB NOT NULL DEFAULT '{}',
subscribers TEXT[] NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_by TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_by TEXT NOT NULL,
deleted_at TIMESTAMPTZ,
deleted_by TEXT
);
CREATE INDEX IF NOT EXISTS idx_schedules_tenant ON scheduler.schedules(tenant_id) WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_schedules_enabled ON scheduler.schedules(tenant_id, enabled) WHERE deleted_at IS NULL;
CREATE UNIQUE INDEX IF NOT EXISTS uq_schedules_tenant_name_active ON scheduler.schedules(tenant_id, name) WHERE deleted_at IS NULL;
-- Runs table
CREATE TABLE IF NOT EXISTS scheduler.runs (
id TEXT NOT NULL,
tenant_id TEXT NOT NULL,
schedule_id TEXT,
trigger JSONB NOT NULL,
state scheduler.run_state NOT NULL,
stats JSONB NOT NULL,
reason JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
started_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
error TEXT,
deltas JSONB NOT NULL,
retry_of TEXT,
schema_version TEXT,
finding_count INT GENERATED ALWAYS AS (NULLIF((stats->>'findingCount'), '')::int) STORED,
critical_count INT GENERATED ALWAYS AS (NULLIF((stats->>'criticalCount'), '')::int) STORED,
high_count INT GENERATED ALWAYS AS (NULLIF((stats->>'highCount'), '')::int) STORED,
new_finding_count INT GENERATED ALWAYS AS (NULLIF((stats->>'newFindingCount'), '')::int) STORED,
component_count INT GENERATED ALWAYS AS (NULLIF((stats->>'componentCount'), '')::int) STORED,
PRIMARY KEY (tenant_id, id)
);
CREATE INDEX IF NOT EXISTS idx_runs_state ON scheduler.runs(state);
CREATE INDEX IF NOT EXISTS idx_runs_schedule ON scheduler.runs(tenant_id, schedule_id);
CREATE INDEX IF NOT EXISTS idx_runs_created ON scheduler.runs(created_at);
CREATE INDEX IF NOT EXISTS ix_runs_with_findings ON scheduler.runs(tenant_id, created_at DESC) WHERE finding_count > 0;
CREATE INDEX IF NOT EXISTS ix_runs_critical ON scheduler.runs(tenant_id, created_at DESC, critical_count) WHERE critical_count > 0;
CREATE INDEX IF NOT EXISTS ix_runs_summary_cover ON scheduler.runs(tenant_id, state, created_at DESC) INCLUDE (finding_count, critical_count, high_count, new_finding_count);
CREATE INDEX IF NOT EXISTS ix_runs_tenant_findings ON scheduler.runs(tenant_id, finding_count DESC, created_at DESC) WHERE state = 'completed';
-- Impact snapshots
CREATE TABLE IF NOT EXISTS scheduler.impact_snapshots (
snapshot_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
run_id TEXT,
impact JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_impact_snapshots_run ON scheduler.impact_snapshots(run_id);
-- Run summaries
CREATE TABLE IF NOT EXISTS scheduler.run_summaries (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
schedule_id TEXT REFERENCES scheduler.schedules(id),
period_start TIMESTAMPTZ NOT NULL,
period_end TIMESTAMPTZ NOT NULL,
total_runs INT NOT NULL DEFAULT 0,
successful_runs INT NOT NULL DEFAULT 0,
failed_runs INT NOT NULL DEFAULT 0,
cancelled_runs INT NOT NULL DEFAULT 0,
avg_duration_seconds NUMERIC(10,2),
max_duration_seconds INT,
min_duration_seconds INT,
total_findings_detected INT NOT NULL DEFAULT 0,
new_criticals INT NOT NULL DEFAULT 0,
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (tenant_id, schedule_id, period_start)
);
CREATE INDEX IF NOT EXISTS idx_run_summaries_tenant ON scheduler.run_summaries(tenant_id, period_start DESC);
-- Execution logs
CREATE TABLE IF NOT EXISTS scheduler.execution_logs (
id BIGSERIAL PRIMARY KEY,
run_id TEXT NOT NULL,
logged_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
level TEXT NOT NULL,
message TEXT NOT NULL,
logger TEXT,
data JSONB NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS idx_execution_logs_run ON scheduler.execution_logs(run_id);
-- Graph jobs (v2 schema)
CREATE TABLE IF NOT EXISTS scheduler.graph_jobs (
id UUID PRIMARY KEY,
tenant_id TEXT NOT NULL,
type scheduler.graph_job_type NOT NULL,
status scheduler.graph_job_status NOT NULL,
payload JSONB NOT NULL,
correlation_id TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_graph_jobs_tenant_status ON scheduler.graph_jobs(tenant_id, status, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_graph_jobs_tenant_type_status ON scheduler.graph_jobs(tenant_id, type, status, created_at DESC);
CREATE TABLE IF NOT EXISTS scheduler.graph_job_events (
id BIGSERIAL PRIMARY KEY,
job_id UUID NOT NULL REFERENCES scheduler.graph_jobs(id) ON DELETE CASCADE,
tenant_id TEXT NOT NULL,
status scheduler.graph_job_status NOT NULL,
payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_graph_job_events_job ON scheduler.graph_job_events(job_id, created_at DESC);
-- Policy run jobs
CREATE TABLE IF NOT EXISTS scheduler.policy_jobs (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
policy_pack_id TEXT NOT NULL,
policy_version INT,
target_type TEXT NOT NULL,
target_id TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('pending','queued','running','completed','failed','cancelled')),
priority INT NOT NULL DEFAULT 100,
run_id TEXT,
requested_by TEXT,
mode TEXT,
metadata JSONB NOT NULL DEFAULT '{}',
inputs JSONB NOT NULL DEFAULT '{}',
attempt_count INT NOT NULL DEFAULT 0,
max_attempts INT NOT NULL DEFAULT 3,
queued_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
available_at TIMESTAMPTZ,
submitted_at TIMESTAMPTZ,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
cancellation_requested BOOLEAN NOT NULL DEFAULT FALSE,
cancellation_reason TEXT,
cancelled_at TIMESTAMPTZ,
last_attempt_at TIMESTAMPTZ,
last_error TEXT,
lease_owner TEXT,
lease_expires_at TIMESTAMPTZ,
correlation_id TEXT
);
CREATE INDEX IF NOT EXISTS idx_policy_jobs_tenant_status ON scheduler.policy_jobs(tenant_id, status);
CREATE INDEX IF NOT EXISTS idx_policy_jobs_run ON scheduler.policy_jobs(run_id);
CREATE TABLE IF NOT EXISTS scheduler.policy_run_jobs (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
policy_id TEXT NOT NULL,
policy_version INT,
mode TEXT NOT NULL,
priority INT NOT NULL,
priority_rank INT NOT NULL,
run_id TEXT,
requested_by TEXT,
correlation_id TEXT,
metadata JSONB,
inputs JSONB NOT NULL,
queued_at TIMESTAMPTZ,
status scheduler.policy_run_status NOT NULL,
attempt_count INT NOT NULL,
last_attempt_at TIMESTAMPTZ,
last_error TEXT,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
available_at TIMESTAMPTZ NOT NULL,
submitted_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
lease_owner TEXT,
lease_expires_at TIMESTAMPTZ,
cancellation_requested BOOLEAN NOT NULL DEFAULT FALSE,
cancellation_requested_at TIMESTAMPTZ,
cancellation_reason TEXT,
cancelled_at TIMESTAMPTZ,
schema_version TEXT
);
CREATE INDEX IF NOT EXISTS idx_policy_run_jobs_tenant ON scheduler.policy_run_jobs(tenant_id);
CREATE INDEX IF NOT EXISTS idx_policy_run_jobs_status ON scheduler.policy_run_jobs(status);
CREATE INDEX IF NOT EXISTS idx_policy_run_jobs_run ON scheduler.policy_run_jobs(run_id);
CREATE INDEX IF NOT EXISTS idx_policy_run_jobs_policy ON scheduler.policy_run_jobs(tenant_id, policy_id);
-- Partitioned audit table
CREATE TABLE IF NOT EXISTS scheduler.audit (
id BIGSERIAL,
tenant_id TEXT NOT NULL,
user_id UUID,
action TEXT NOT NULL,
resource_type TEXT NOT NULL,
resource_id TEXT,
old_value JSONB,
new_value JSONB,
correlation_id TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);
DO $$
DECLARE
v_start DATE;
v_end DATE;
v_partition_name TEXT;
BEGIN
v_start := date_trunc('month', NOW() - INTERVAL '6 months')::DATE;
WHILE v_start <= date_trunc('month', NOW() + INTERVAL '3 months')::DATE LOOP
v_end := (v_start + INTERVAL '1 month')::DATE;
v_partition_name := 'audit_' || to_char(v_start, 'YYYY_MM');
IF NOT EXISTS (
SELECT 1 FROM pg_class c
JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE n.nspname = 'scheduler' AND c.relname = v_partition_name
) THEN
EXECUTE format(
'CREATE TABLE scheduler.%I PARTITION OF scheduler.audit FOR VALUES FROM (%L) TO (%L)',
v_partition_name, v_start, v_end
);
END IF;
v_start := v_end;
END LOOP;
END $$;
CREATE TABLE IF NOT EXISTS scheduler.audit_default PARTITION OF scheduler.audit DEFAULT;
CREATE INDEX IF NOT EXISTS ix_audit_tenant ON scheduler.audit(tenant_id);
CREATE INDEX IF NOT EXISTS ix_audit_resource ON scheduler.audit(resource_type, resource_id);
CREATE INDEX IF NOT EXISTS ix_audit_correlation ON scheduler.audit(correlation_id) WHERE correlation_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS brin_audit_created ON scheduler.audit USING BRIN(created_at) WITH (pages_per_range = 128);
COMMENT ON TABLE scheduler.audit IS 'Audit log for scheduler operations. Partitioned monthly by created_at for retention management.';
-- Row-Level Security
ALTER TABLE scheduler.schedules ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.schedules FORCE ROW LEVEL SECURITY;
CREATE POLICY schedules_tenant_isolation ON scheduler.schedules FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.runs ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.runs FORCE ROW LEVEL SECURITY;
CREATE POLICY runs_tenant_isolation ON scheduler.runs FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.jobs ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.jobs FORCE ROW LEVEL SECURITY;
CREATE POLICY jobs_tenant_isolation ON scheduler.jobs FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.triggers ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.triggers FORCE ROW LEVEL SECURITY;
CREATE POLICY triggers_tenant_isolation ON scheduler.triggers FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.graph_jobs ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.graph_jobs FORCE ROW LEVEL SECURITY;
CREATE POLICY graph_jobs_tenant_isolation ON scheduler.graph_jobs FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.policy_jobs ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.policy_jobs FORCE ROW LEVEL SECURITY;
CREATE POLICY policy_jobs_tenant_isolation ON scheduler.policy_jobs FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.locks ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.locks FORCE ROW LEVEL SECURITY;
CREATE POLICY locks_tenant_isolation ON scheduler.locks FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.impact_snapshots ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.impact_snapshots FORCE ROW LEVEL SECURITY;
CREATE POLICY impact_snapshots_tenant_isolation ON scheduler.impact_snapshots FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.run_summaries ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.run_summaries FORCE ROW LEVEL SECURITY;
CREATE POLICY run_summaries_tenant_isolation ON scheduler.run_summaries FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.audit ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.audit FORCE ROW LEVEL SECURITY;
CREATE POLICY audit_tenant_isolation ON scheduler.audit FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.job_history ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.job_history FORCE ROW LEVEL SECURITY;
CREATE POLICY job_history_tenant_isolation ON scheduler.job_history FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.metrics ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.metrics FORCE ROW LEVEL SECURITY;
CREATE POLICY metrics_tenant_isolation ON scheduler.metrics FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());
ALTER TABLE scheduler.execution_logs ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.execution_logs FORCE ROW LEVEL SECURITY;
CREATE POLICY execution_logs_tenant_isolation ON scheduler.execution_logs FOR ALL
USING (
run_id IN (SELECT id FROM scheduler.runs WHERE tenant_id = scheduler_app.require_current_tenant())
);
-- Admin bypass role
DO $$
BEGIN
IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'scheduler_admin') THEN
CREATE ROLE scheduler_admin WITH NOLOGIN BYPASSRLS;
END IF;
END
$$;
-- ============================================================================
-- 002_hlc_queue_chain.sql - HLC-ordered scheduler queue with chain linking
-- ============================================================================
CREATE TABLE IF NOT EXISTS scheduler.scheduler_log (
seq_bigint BIGSERIAL PRIMARY KEY,
tenant_id TEXT NOT NULL,
t_hlc TEXT NOT NULL,
partition_key TEXT DEFAULT '',
job_id UUID NOT NULL,
payload_hash BYTEA NOT NULL CHECK (octet_length(payload_hash) = 32),
prev_link BYTEA CHECK (prev_link IS NULL OR octet_length(prev_link) = 32),
link BYTEA NOT NULL CHECK (octet_length(link) = 32),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT uq_scheduler_log_order UNIQUE (tenant_id, t_hlc, partition_key, job_id)
);
CREATE INDEX IF NOT EXISTS idx_scheduler_log_tenant_hlc
ON scheduler.scheduler_log (tenant_id, t_hlc ASC);
CREATE INDEX IF NOT EXISTS idx_scheduler_log_partition
ON scheduler.scheduler_log (tenant_id, partition_key, t_hlc ASC);
CREATE INDEX IF NOT EXISTS idx_scheduler_log_job_id
ON scheduler.scheduler_log (job_id);
CREATE INDEX IF NOT EXISTS idx_scheduler_log_link
ON scheduler.scheduler_log (link);
CREATE INDEX IF NOT EXISTS idx_scheduler_log_created
ON scheduler.scheduler_log (tenant_id, created_at DESC);
CREATE TABLE IF NOT EXISTS scheduler.batch_snapshot (
batch_id UUID PRIMARY KEY,
tenant_id TEXT NOT NULL,
range_start_t TEXT NOT NULL,
range_end_t TEXT NOT NULL,
head_link BYTEA NOT NULL CHECK (octet_length(head_link) = 32),
job_count INT NOT NULL CHECK (job_count >= 0),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
signed_by TEXT,
signature BYTEA,
CONSTRAINT chk_signature_requires_signer CHECK (
(signature IS NULL AND signed_by IS NULL) OR
(signature IS NOT NULL AND signed_by IS NOT NULL)
)
);
CREATE INDEX IF NOT EXISTS idx_batch_snapshot_tenant
ON scheduler.batch_snapshot (tenant_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_batch_snapshot_range
ON scheduler.batch_snapshot (tenant_id, range_start_t, range_end_t);
CREATE TABLE IF NOT EXISTS scheduler.chain_heads (
tenant_id TEXT NOT NULL,
partition_key TEXT NOT NULL DEFAULT '',
last_link BYTEA NOT NULL CHECK (octet_length(last_link) = 32),
last_t_hlc TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (tenant_id, partition_key)
);
CREATE INDEX IF NOT EXISTS idx_chain_heads_updated
ON scheduler.chain_heads (updated_at DESC);
CREATE OR REPLACE FUNCTION scheduler.upsert_chain_head(
p_tenant_id TEXT,
p_partition_key TEXT,
p_new_link BYTEA,
p_new_t_hlc TEXT
)
RETURNS VOID
LANGUAGE plpgsql
AS $$
BEGIN
INSERT INTO scheduler.chain_heads (tenant_id, partition_key, last_link, last_t_hlc, updated_at)
VALUES (p_tenant_id, p_partition_key, p_new_link, p_new_t_hlc, NOW())
ON CONFLICT (tenant_id, partition_key)
DO UPDATE SET
last_link = EXCLUDED.last_link,
last_t_hlc = EXCLUDED.last_t_hlc,
updated_at = EXCLUDED.updated_at
WHERE scheduler.chain_heads.last_t_hlc < EXCLUDED.last_t_hlc;
END;
$$;
-- ============================================================================
-- 003_exception_lifecycle.sql - Exception management tables
-- ============================================================================
DO $$ BEGIN
CREATE TYPE scheduler.exception_state AS ENUM ('pending', 'active', 'expired', 'revoked');
EXCEPTION WHEN duplicate_object THEN NULL; END $$;
CREATE TABLE IF NOT EXISTS scheduler.scheduler_exceptions (
exception_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
policy_id TEXT NOT NULL,
vulnerability_id TEXT NOT NULL,
component_purl TEXT,
state scheduler.exception_state NOT NULL DEFAULT 'pending',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
activation_date TIMESTAMPTZ,
expiration_date TIMESTAMPTZ,
activated_at TIMESTAMPTZ,
expired_at TIMESTAMPTZ,
justification TEXT,
created_by TEXT
);
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_tenant
ON scheduler.scheduler_exceptions(tenant_id);
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_state
ON scheduler.scheduler_exceptions(state);
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_tenant_state
ON scheduler.scheduler_exceptions(tenant_id, state);
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_pending_activation
ON scheduler.scheduler_exceptions(activation_date)
WHERE state = 'pending';
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_active_expiration
ON scheduler.scheduler_exceptions(expiration_date)
WHERE state = 'active';
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_policy
ON scheduler.scheduler_exceptions(tenant_id, policy_id);
CREATE INDEX IF NOT EXISTS idx_scheduler_exceptions_vulnerability
ON scheduler.scheduler_exceptions(tenant_id, vulnerability_id);
ALTER TABLE scheduler.scheduler_exceptions ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler.scheduler_exceptions FORCE ROW LEVEL SECURITY;
CREATE POLICY scheduler_exceptions_tenant_isolation ON scheduler.scheduler_exceptions FOR ALL
USING (tenant_id = scheduler_app.require_current_tenant())
WITH CHECK (tenant_id = scheduler_app.require_current_tenant());