Skip to content

Commit 81b502c

Browse files
petervartanianpetervartanian
committed
2
1 parent a5b5167 commit 81b502c

File tree

10 files changed

+373
-0
lines changed

10 files changed

+373
-0
lines changed

‎warehouse/postgres/fixtures/opa_release_gate_input.json‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
{
2+
"run_id": "sample_run",
3+
"payload_root_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
4+
"required_artifacts": [
5+
"run_manifest.json",
6+
"warehouse/warehouse.duckdb"
7+
]
8+
}

‎warehouse/postgres/fixtures/sample_manifest.json‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
{
2+
"run_id": "sample_run",
3+
"payload_root_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
4+
"manifest": {
5+
"schema_version": "1",
6+
"artifacts": []
7+
}
8+
}

‎warehouse/postgres/functions/001_hashing.sql‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
CREATE OR REPLACE FUNCTION omphalos.sha256_text(p_text TEXT)
2+
RETURNS TEXT
3+
LANGUAGE plpgsql
4+
AS $$
5+
DECLARE
6+
v BYTEA;
7+
BEGIN
8+
v := digest(coalesce(p_text, ''), 'sha256');
9+
RETURN encode(v, 'hex');
10+
END;
11+
$$;
12+
13+
CREATE OR REPLACE FUNCTION omphalos.sha256_jsonb(p_json JSONB)
14+
RETURNS TEXT
15+
LANGUAGE plpgsql
16+
AS $$
17+
BEGIN
18+
RETURN omphalos.sha256_text(coalesce(p_json::TEXT, 'null'));
19+
END;
20+
$$;

‎warehouse/postgres/functions/002_normalization.sql‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
CREATE OR REPLACE FUNCTION omphalos.normalize_text(p_text TEXT)
2+
RETURNS TEXT
3+
LANGUAGE plpgsql
4+
AS $$
5+
DECLARE
6+
t TEXT;
7+
BEGIN
8+
t := upper(trim(coalesce(p_text, '')));
9+
t := regexp_replace(t, '\s+', ' ', 'g');
10+
RETURN nullif(t, '');
11+
END;
12+
$$;
13+
14+
CREATE OR REPLACE FUNCTION omphalos.normalize_country(p_country TEXT)
15+
RETURNS TEXT
16+
LANGUAGE plpgsql
17+
AS $$
18+
DECLARE
19+
c TEXT;
20+
BEGIN
21+
c := omphalos.normalize_text(p_country);
22+
IF c IS NULL THEN
23+
RETURN NULL;
24+
END IF;
25+
IF length(c) = 2 THEN
26+
RETURN c;
27+
END IF;
28+
RETURN c;
29+
END;
30+
$$;
31+
32+
CREATE OR REPLACE FUNCTION omphalos.normalize_hs(p_hs TEXT)
33+
RETURNS TEXT
34+
LANGUAGE plpgsql
35+
AS $$
36+
DECLARE
37+
h TEXT;
38+
BEGIN
39+
h := regexp_replace(coalesce(p_hs, ''), '[^0-9]', '', 'g');
40+
IF length(h) < 2 THEN
41+
RETURN NULL;
42+
END IF;
43+
RETURN h;
44+
END;
45+
$$;

‎warehouse/postgres/functions/003_review_queue.sql‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
CREATE OR REPLACE FUNCTION omphalos.score_severity(p_score DOUBLE PRECISION, p_value DOUBLE PRECISION)
2+
RETURNS INTEGER
3+
LANGUAGE plpgsql
4+
AS $$
5+
DECLARE
6+
s DOUBLE PRECISION;
7+
v DOUBLE PRECISION;
8+
sev INTEGER;
9+
BEGIN
10+
s := coalesce(p_score, 0.0);
11+
v := coalesce(p_value, 0.0);
12+
sev := 0;
13+
IF s >= 0.90 THEN
14+
sev := sev + 5;
15+
ELSIF s >= 0.75 THEN
16+
sev := sev + 3;
17+
ELSIF s >= 0.60 THEN
18+
sev := sev + 2;
19+
ELSIF s >= 0.45 THEN
20+
sev := sev + 1;
21+
END IF;
22+
23+
IF v >= 1000000 THEN
24+
sev := sev + 3;
25+
ELSIF v >= 250000 THEN
26+
sev := sev + 2;
27+
ELSIF v >= 50000 THEN
28+
sev := sev + 1;
29+
END IF;
30+
31+
RETURN least(10, greatest(0, sev));
32+
END;
33+
$$;
34+
35+
CREATE OR REPLACE PROCEDURE omphalos.refresh_review_queue(p_run_id TEXT)
36+
LANGUAGE plpgsql
37+
AS $$
38+
BEGIN
39+
DELETE FROM omphalos.review_queue WHERE run_id = p_run_id;
40+
41+
INSERT INTO omphalos.review_queue (run_id, shipment_id, entity_id, review_status, severity, rationale)
42+
SELECT
43+
m.run_id,
44+
m.shipment_id,
45+
m.entity_id,
46+
CASE
47+
WHEN m.status = 'review' THEN 'needs_review'
48+
WHEN m.status = 'matched' THEN 'auto_match'
49+
ELSE 'unmatched'
50+
END AS review_status,
51+
omphalos.score_severity(m.score, t.value_usd) AS severity,
52+
coalesce(m.explanation, '')
53+
FROM omphalos.entity_matches m
54+
JOIN omphalos.trade_feed t ON t.shipment_id = m.shipment_id
55+
WHERE m.run_id = p_run_id
56+
AND m.status IN ('review', 'matched');
57+
END;
58+
$$;

‎warehouse/postgres/migrations/000_apply_all.sql‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
\set ON_ERROR_STOP on
2+
\i migrations/001_schema.sql
3+
\i functions/001_hashing.sql
4+
\i functions/002_normalization.sql
5+
\i functions/003_review_queue.sql
6+
\i triggers/001_audit_triggers.sql
7+
\i views/001_views.sql
8+
\i policies/001_rls.sql

‎warehouse/postgres/migrations/001_schema.sql‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
CREATE SCHEMA IF NOT EXISTS omphalos;
2+
3+
CREATE EXTENSION IF NOT EXISTS pgcrypto;
4+
5+
CREATE TABLE IF NOT EXISTS omphalos.run_manifest (
6+
run_id TEXT PRIMARY KEY,
7+
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
8+
clock_seed TEXT,
9+
payload_root_sha256 TEXT,
10+
manifest JSONB
11+
);
12+
13+
CREATE TABLE IF NOT EXISTS omphalos.trade_feed (
14+
shipment_id TEXT PRIMARY KEY,
15+
exporter_name TEXT,
16+
importer_name TEXT,
17+
exporter_country TEXT,
18+
importer_country TEXT,
19+
country TEXT,
20+
hs_code TEXT,
21+
hs2 TEXT GENERATED ALWAYS AS (substring(hs_code from 1 for 2)) STORED,
22+
hs4 TEXT GENERATED ALWAYS AS (substring(hs_code from 1 for 4)) STORED,
23+
hs6 TEXT GENERATED ALWAYS AS (substring(hs_code from 1 for 6)) STORED,
24+
value_usd DOUBLE PRECISION,
25+
ship_date DATE,
26+
ingested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
27+
run_id TEXT REFERENCES omphalos.run_manifest(run_id)
28+
);
29+
30+
CREATE INDEX IF NOT EXISTS trade_feed_hs2_idx ON omphalos.trade_feed(hs2);
31+
CREATE INDEX IF NOT EXISTS trade_feed_hs4_idx ON omphalos.trade_feed(hs4);
32+
CREATE INDEX IF NOT EXISTS trade_feed_hs6_idx ON omphalos.trade_feed(hs6);
33+
CREATE INDEX IF NOT EXISTS trade_feed_ship_date_idx ON omphalos.trade_feed(ship_date);
34+
CREATE INDEX IF NOT EXISTS trade_feed_run_idx ON omphalos.trade_feed(run_id);
35+
36+
CREATE TABLE IF NOT EXISTS omphalos.registry (
37+
entity_id TEXT PRIMARY KEY,
38+
entity_name TEXT,
39+
country TEXT,
40+
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
41+
run_id TEXT REFERENCES omphalos.run_manifest(run_id)
42+
);
43+
44+
CREATE INDEX IF NOT EXISTS registry_country_idx ON omphalos.registry(country);
45+
CREATE INDEX IF NOT EXISTS registry_run_idx ON omphalos.registry(run_id);
46+
47+
CREATE TABLE IF NOT EXISTS omphalos.entity_matches (
48+
shipment_id TEXT REFERENCES omphalos.trade_feed(shipment_id),
49+
entity_id TEXT REFERENCES omphalos.registry(entity_id),
50+
score DOUBLE PRECISION,
51+
status TEXT,
52+
explanation TEXT,
53+
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
54+
run_id TEXT REFERENCES omphalos.run_manifest(run_id),
55+
PRIMARY KEY (shipment_id, entity_id)
56+
);
57+
58+
CREATE INDEX IF NOT EXISTS entity_matches_status_idx ON omphalos.entity_matches(status);
59+
CREATE INDEX IF NOT EXISTS entity_matches_run_idx ON omphalos.entity_matches(run_id);
60+
61+
CREATE TABLE IF NOT EXISTS omphalos.entity_scores (
62+
entity_id TEXT REFERENCES omphalos.registry(entity_id),
63+
entity_name TEXT,
64+
country TEXT,
65+
shipment_count INTEGER,
66+
total_value_usd DOUBLE PRECISION,
67+
chokepoint_score DOUBLE PRECISION,
68+
computed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
69+
run_id TEXT REFERENCES omphalos.run_manifest(run_id),
70+
PRIMARY KEY (entity_id, run_id)
71+
);
72+
73+
CREATE INDEX IF NOT EXISTS entity_scores_run_idx ON omphalos.entity_scores(run_id);
74+
CREATE INDEX IF NOT EXISTS entity_scores_score_idx ON omphalos.entity_scores(chokepoint_score);
75+
76+
CREATE TABLE IF NOT EXISTS omphalos.review_queue (
77+
run_id TEXT REFERENCES omphalos.run_manifest(run_id),
78+
shipment_id TEXT REFERENCES omphalos.trade_feed(shipment_id),
79+
entity_id TEXT REFERENCES omphalos.registry(entity_id),
80+
review_status TEXT NOT NULL,
81+
severity INTEGER NOT NULL,
82+
rationale TEXT,
83+
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
84+
PRIMARY KEY (run_id, shipment_id, entity_id)
85+
);
86+
87+
CREATE INDEX IF NOT EXISTS review_queue_run_idx ON omphalos.review_queue(run_id);
88+
CREATE INDEX IF NOT EXISTS review_queue_status_idx ON omphalos.review_queue(review_status);
89+
CREATE INDEX IF NOT EXISTS review_queue_severity_idx ON omphalos.review_queue(severity);
90+
91+
CREATE TABLE IF NOT EXISTS omphalos.audit_event (
92+
audit_id BIGSERIAL PRIMARY KEY,
93+
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
94+
actor TEXT,
95+
action TEXT NOT NULL,
96+
table_name TEXT NOT NULL,
97+
row_pk TEXT,
98+
before_row JSONB,
99+
after_row JSONB,
100+
run_id TEXT
101+
);
102+
103+
CREATE INDEX IF NOT EXISTS audit_event_table_idx ON omphalos.audit_event(table_name);
104+
CREATE INDEX IF NOT EXISTS audit_event_run_idx ON omphalos.audit_event(run_id);
105+
106+
CREATE TABLE IF NOT EXISTS omphalos.release_attestation (
107+
attestation_id BIGSERIAL PRIMARY KEY,
108+
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
109+
run_id TEXT REFERENCES omphalos.run_manifest(run_id),
110+
payload_root_sha256 TEXT NOT NULL,
111+
policy_bundle_sha256 TEXT,
112+
attestation JSONB NOT NULL
113+
);
114+
115+
CREATE INDEX IF NOT EXISTS release_attestation_run_idx ON omphalos.release_attestation(run_id);

‎warehouse/postgres/policies/001_rls.sql‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
ALTER TABLE omphalos.review_queue ENABLE ROW LEVEL SECURITY;
2+
3+
DO $$
4+
BEGIN
5+
IF NOT EXISTS (
6+
SELECT 1 FROM pg_roles WHERE rolname = 'omphalos_reviewer'
7+
) THEN
8+
CREATE ROLE omphalos_reviewer;
9+
END IF;
10+
END;
11+
$$;
12+
13+
DROP POLICY IF EXISTS review_queue_reviewer_select ON omphalos.review_queue;
14+
CREATE POLICY review_queue_reviewer_select ON omphalos.review_queue
15+
FOR SELECT TO omphalos_reviewer
16+
USING (true);

‎warehouse/postgres/triggers/001_audit_triggers.sql‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
CREATE OR REPLACE FUNCTION omphalos.audit_row_change()
2+
RETURNS TRIGGER
3+
LANGUAGE plpgsql
4+
AS $$
5+
DECLARE
6+
pk TEXT;
7+
action TEXT;
8+
BEGIN
9+
action := TG_OP;
10+
pk := NULL;
11+
12+
IF action = 'INSERT' THEN
13+
INSERT INTO omphalos.audit_event(action, table_name, row_pk, after_row, run_id)
14+
VALUES (action, TG_TABLE_SCHEMA || '.' || TG_TABLE_NAME, pk, to_jsonb(NEW), coalesce(NEW.run_id, NULL));
15+
RETURN NEW;
16+
ELSIF action = 'UPDATE' THEN
17+
INSERT INTO omphalos.audit_event(action, table_name, row_pk, before_row, after_row, run_id)
18+
VALUES (action, TG_TABLE_SCHEMA || '.' || TG_TABLE_NAME, pk, to_jsonb(OLD), to_jsonb(NEW), coalesce(NEW.run_id, coalesce(OLD.run_id, NULL)));
19+
RETURN NEW;
20+
ELSIF action = 'DELETE' THEN
21+
INSERT INTO omphalos.audit_event(action, table_name, row_pk, before_row, run_id)
22+
VALUES (action, TG_TABLE_SCHEMA || '.' || TG_TABLE_NAME, pk, to_jsonb(OLD), coalesce(OLD.run_id, NULL));
23+
RETURN OLD;
24+
END IF;
25+
26+
RETURN NULL;
27+
END;
28+
$$;
29+
30+
DROP TRIGGER IF EXISTS trg_audit_trade_feed ON omphalos.trade_feed;
31+
CREATE TRIGGER trg_audit_trade_feed
32+
AFTER INSERT OR UPDATE OR DELETE ON omphalos.trade_feed
33+
FOR EACH ROW EXECUTE FUNCTION omphalos.audit_row_change();
34+
35+
DROP TRIGGER IF EXISTS trg_audit_registry ON omphalos.registry;
36+
CREATE TRIGGER trg_audit_registry
37+
AFTER INSERT OR UPDATE OR DELETE ON omphalos.registry
38+
FOR EACH ROW EXECUTE FUNCTION omphalos.audit_row_change();
39+
40+
DROP TRIGGER IF EXISTS trg_audit_entity_matches ON omphalos.entity_matches;
41+
CREATE TRIGGER trg_audit_entity_matches
42+
AFTER INSERT OR UPDATE OR DELETE ON omphalos.entity_matches
43+
FOR EACH ROW EXECUTE FUNCTION omphalos.audit_row_change();
44+
45+
DROP TRIGGER IF EXISTS trg_audit_entity_scores ON omphalos.entity_scores;
46+
CREATE TRIGGER trg_audit_entity_scores
47+
AFTER INSERT OR UPDATE OR DELETE ON omphalos.entity_scores
48+
FOR EACH ROW EXECUTE FUNCTION omphalos.audit_row_change();

‎warehouse/postgres/views/001_views.sql‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
CREATE OR REPLACE VIEW omphalos.v_trade_enriched AS
2+
SELECT
3+
shipment_id,
4+
exporter_name,
5+
importer_name,
6+
exporter_country,
7+
importer_country,
8+
hs_code,
9+
hs2,
10+
hs4,
11+
hs6,
12+
value_usd,
13+
ship_date,
14+
run_id
15+
FROM omphalos.trade_feed;
16+
17+
CREATE OR REPLACE VIEW omphalos.v_entity_scores_ranked AS
18+
SELECT
19+
entity_id,
20+
entity_name,
21+
country,
22+
shipment_count,
23+
total_value_usd,
24+
chokepoint_score,
25+
dense_rank() OVER (PARTITION BY run_id ORDER BY chokepoint_score DESC, total_value_usd DESC) AS risk_rank,
26+
run_id
27+
FROM omphalos.entity_scores;
28+
29+
CREATE OR REPLACE VIEW omphalos.v_review_queue_flat AS
30+
SELECT
31+
q.run_id,
32+
q.shipment_id,
33+
q.entity_id,
34+
r.entity_name,
35+
q.review_status,
36+
q.severity,
37+
q.rationale,
38+
t.exporter_name,
39+
t.importer_name,
40+
t.exporter_country,
41+
t.importer_country,
42+
t.hs_code,
43+
t.value_usd,
44+
t.ship_date
45+
FROM omphalos.review_queue q
46+
JOIN omphalos.registry r ON r.entity_id = q.entity_id
47+
JOIN omphalos.trade_feed t ON t.shipment_id = q.shipment_id;

0 commit comments

Comments
 (0)