282 lines
9.5 KiB
PL/PgSQL
282 lines
9.5 KiB
PL/PgSQL
-- Durable metadata-only monthly report publication for ARR2.
|
|
-- PostgreSQL 15+. Migrations 008-011 remain immutable.
|
|
|
|
BEGIN;
|
|
|
|
DO $$
|
|
BEGIN
|
|
IF current_database() <> 'booking_test' THEN
|
|
RAISE EXCEPTION
|
|
'ARR monthly publication migration is allowed only in booking_test';
|
|
END IF;
|
|
IF to_regclass('ingestion.artifacts') IS NULL
|
|
OR to_regclass('ingestion.outbox_events') IS NULL
|
|
OR to_regclass('finance.daily_versions') IS NULL THEN
|
|
RAISE EXCEPTION
|
|
'ARR migrations 008 through 011 must be applied first';
|
|
END IF;
|
|
IF to_regnamespace('reporting') IS NOT NULL THEN
|
|
RAISE EXCEPTION
|
|
'ARR monthly publication migration is already applied';
|
|
END IF;
|
|
IF NOT EXISTS (
|
|
SELECT 1
|
|
FROM pg_constraint
|
|
WHERE conrelid = 'ingestion.artifacts'::regclass
|
|
AND conname = 'artifacts_storage_provider_check'
|
|
) THEN
|
|
RAISE EXCEPTION
|
|
'ingestion.artifacts storage-provider constraint is missing';
|
|
END IF;
|
|
END;
|
|
$$;
|
|
|
|
ALTER TABLE ingestion.artifacts
|
|
DROP CONSTRAINT artifacts_storage_provider_check;
|
|
|
|
ALTER TABLE ingestion.artifacts
|
|
ADD CONSTRAINT artifacts_storage_provider_check CHECK (
|
|
storage_provider IN ('oss', 's3', 'local_fixture', 'local')
|
|
);
|
|
|
|
CREATE SCHEMA reporting;
|
|
|
|
COMMENT ON SCHEMA reporting IS
|
|
'Metadata-only publication lifecycle and lineage for derived ARR reports; business rows remain in Finance facts.';
|
|
|
|
CREATE TABLE reporting.monthly_runs (
|
|
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
|
period_start date NOT NULL,
|
|
as_of_date date NOT NULL,
|
|
version_no integer NOT NULL CHECK (version_no > 0),
|
|
source_snapshot_sha256 character(64) NOT NULL CHECK (
|
|
source_snapshot_sha256 ~ '^[0-9a-f]{64}$'
|
|
),
|
|
report_status text NOT NULL CHECK (report_status IN (
|
|
'reserved', 'active', 'superseded', 'failed'
|
|
)),
|
|
processor_version text NOT NULL CHECK (btrim(processor_version) <> ''),
|
|
rule_set_sha256 character(64) NOT NULL CHECK (
|
|
rule_set_sha256 ~ '^[0-9a-f]{64}$'
|
|
),
|
|
result_schema_version text NOT NULL CHECK (
|
|
btrim(result_schema_version) <> ''
|
|
),
|
|
row_count integer NOT NULL CHECK (row_count >= 0),
|
|
channel_count integer NOT NULL CHECK (channel_count >= 5),
|
|
workbook_artifact_id bigint REFERENCES ingestion.artifacts(id),
|
|
result_artifact_id bigint REFERENCES ingestion.artifacts(id),
|
|
semantic_sha256 character(64) CHECK (
|
|
semantic_sha256 IS NULL OR semantic_sha256 ~ '^[0-9a-f]{64}$'
|
|
),
|
|
failure_code text,
|
|
failure_message text,
|
|
created_at timestamptz NOT NULL DEFAULT now(),
|
|
updated_at timestamptz NOT NULL DEFAULT now(),
|
|
published_at timestamptz,
|
|
superseded_at timestamptz,
|
|
failed_at timestamptz,
|
|
CONSTRAINT monthly_runs_period_shape CHECK (
|
|
period_start = date_trunc('month', period_start)::date
|
|
AND as_of_date >= period_start
|
|
AND as_of_date < (period_start + interval '1 month')::date
|
|
),
|
|
CONSTRAINT monthly_runs_period_version_unique
|
|
UNIQUE (period_start, version_no),
|
|
CONSTRAINT monthly_runs_snapshot_unique
|
|
UNIQUE (period_start, source_snapshot_sha256),
|
|
CONSTRAINT monthly_runs_status_shape CHECK (
|
|
(
|
|
report_status = 'reserved'
|
|
AND workbook_artifact_id IS NULL
|
|
AND result_artifact_id IS NULL
|
|
AND semantic_sha256 IS NULL
|
|
AND failure_code IS NULL
|
|
AND failure_message IS NULL
|
|
AND published_at IS NULL
|
|
AND superseded_at IS NULL
|
|
AND failed_at IS NULL
|
|
)
|
|
OR (
|
|
report_status = 'active'
|
|
AND workbook_artifact_id IS NOT NULL
|
|
AND result_artifact_id IS NOT NULL
|
|
AND semantic_sha256 IS NOT NULL
|
|
AND failure_code IS NULL
|
|
AND failure_message IS NULL
|
|
AND published_at IS NOT NULL
|
|
AND superseded_at IS NULL
|
|
AND failed_at IS NULL
|
|
)
|
|
OR (
|
|
report_status = 'superseded'
|
|
AND workbook_artifact_id IS NOT NULL
|
|
AND result_artifact_id IS NOT NULL
|
|
AND semantic_sha256 IS NOT NULL
|
|
AND failure_code IS NULL
|
|
AND failure_message IS NULL
|
|
AND published_at IS NOT NULL
|
|
AND superseded_at IS NOT NULL
|
|
AND failed_at IS NULL
|
|
)
|
|
OR (
|
|
report_status = 'failed'
|
|
AND workbook_artifact_id IS NULL
|
|
AND result_artifact_id IS NULL
|
|
AND semantic_sha256 IS NULL
|
|
AND failure_code IS NOT NULL
|
|
AND btrim(failure_code) <> ''
|
|
AND published_at IS NULL
|
|
AND superseded_at IS NULL
|
|
AND failed_at IS NOT NULL
|
|
)
|
|
)
|
|
);
|
|
|
|
CREATE UNIQUE INDEX monthly_runs_one_active_period_idx
|
|
ON reporting.monthly_runs (period_start)
|
|
WHERE report_status = 'active';
|
|
|
|
CREATE INDEX monthly_runs_period_history_idx
|
|
ON reporting.monthly_runs (period_start, version_no DESC);
|
|
|
|
CREATE TABLE reporting.monthly_run_daily_versions (
|
|
report_id bigint NOT NULL REFERENCES reporting.monthly_runs(id),
|
|
business_date date NOT NULL,
|
|
daily_version_id bigint NOT NULL,
|
|
PRIMARY KEY (report_id, business_date),
|
|
CONSTRAINT monthly_run_daily_versions_version_unique
|
|
UNIQUE (report_id, daily_version_id),
|
|
CONSTRAINT monthly_run_daily_versions_version_fk
|
|
FOREIGN KEY (daily_version_id, business_date)
|
|
REFERENCES finance.daily_versions(id, business_date)
|
|
);
|
|
|
|
CREATE TABLE reporting.monthly_channel_manifest (
|
|
report_id bigint NOT NULL REFERENCES reporting.monthly_runs(id),
|
|
worksheet text NOT NULL CHECK (
|
|
btrim(worksheet) <> ''
|
|
AND char_length(worksheet) <= 31
|
|
AND worksheet !~ '[:\\/?*\[\]]'
|
|
),
|
|
worksheet_order integer NOT NULL CHECK (worksheet_order > 0),
|
|
row_count integer NOT NULL CHECK (row_count >= 0),
|
|
PRIMARY KEY (report_id, worksheet),
|
|
CONSTRAINT monthly_channel_manifest_order_unique
|
|
UNIQUE (report_id, worksheet_order)
|
|
);
|
|
|
|
CREATE OR REPLACE FUNCTION reporting.validate_monthly_run_publication()
|
|
RETURNS trigger
|
|
LANGUAGE plpgsql
|
|
AS $$
|
|
DECLARE
|
|
workbook_kind text;
|
|
workbook_provider text;
|
|
result_kind text;
|
|
result_provider text;
|
|
manifest_count integer;
|
|
manifest_rows bigint;
|
|
manifest_min integer;
|
|
manifest_max integer;
|
|
lineage_count integer;
|
|
BEGIN
|
|
IF NEW.report_status NOT IN ('active', 'superseded') THEN
|
|
RETURN NEW;
|
|
END IF;
|
|
|
|
SELECT artifact_kind, storage_provider
|
|
INTO workbook_kind, workbook_provider
|
|
FROM ingestion.artifacts
|
|
WHERE id = NEW.workbook_artifact_id;
|
|
|
|
SELECT artifact_kind, storage_provider
|
|
INTO result_kind, result_provider
|
|
FROM ingestion.artifacts
|
|
WHERE id = NEW.result_artifact_id;
|
|
|
|
IF workbook_kind IS DISTINCT FROM 'monthly_xlsx'
|
|
OR workbook_provider IS DISTINCT FROM 'local'
|
|
OR result_kind IS DISTINCT FROM 'result_json'
|
|
OR result_provider IS DISTINCT FROM 'local' THEN
|
|
RAISE EXCEPTION
|
|
'published monthly run must reference controlled local monthly_xlsx and result_json artifacts';
|
|
END IF;
|
|
|
|
SELECT count(*), COALESCE(sum(row_count), 0), min(worksheet_order), max(worksheet_order)
|
|
INTO manifest_count, manifest_rows, manifest_min, manifest_max
|
|
FROM reporting.monthly_channel_manifest
|
|
WHERE report_id = NEW.id;
|
|
|
|
IF manifest_count <> NEW.channel_count
|
|
OR manifest_rows <> NEW.row_count
|
|
OR manifest_min <> 1
|
|
OR manifest_max <> manifest_count THEN
|
|
RAISE EXCEPTION
|
|
'published monthly run requires a continuous reconciled channel manifest';
|
|
END IF;
|
|
|
|
SELECT count(*)
|
|
INTO lineage_count
|
|
FROM reporting.monthly_run_daily_versions
|
|
WHERE report_id = NEW.id;
|
|
|
|
IF lineage_count < 1 THEN
|
|
RAISE EXCEPTION
|
|
'published monthly run requires daily-version lineage';
|
|
END IF;
|
|
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
|
|
CREATE TRIGGER monthly_runs_publication_guard
|
|
BEFORE INSERT OR UPDATE ON reporting.monthly_runs
|
|
FOR EACH ROW
|
|
EXECUTE FUNCTION reporting.validate_monthly_run_publication();
|
|
|
|
CREATE OR REPLACE FUNCTION reporting.protect_published_monthly_child()
|
|
RETURNS trigger
|
|
LANGUAGE plpgsql
|
|
AS $$
|
|
DECLARE
|
|
parent_id bigint;
|
|
BEGIN
|
|
parent_id := COALESCE(OLD.report_id, NEW.report_id);
|
|
IF EXISTS (
|
|
SELECT 1
|
|
FROM reporting.monthly_runs AS run
|
|
WHERE run.id = parent_id
|
|
AND run.report_status IN ('active', 'superseded')
|
|
) THEN
|
|
RAISE EXCEPTION
|
|
'published monthly report lineage and manifest are immutable';
|
|
END IF;
|
|
RETURN COALESCE(NEW, OLD);
|
|
END;
|
|
$$;
|
|
|
|
CREATE TRIGGER monthly_run_daily_versions_immutable
|
|
BEFORE UPDATE OR DELETE ON reporting.monthly_run_daily_versions
|
|
FOR EACH ROW
|
|
EXECUTE FUNCTION reporting.protect_published_monthly_child();
|
|
|
|
CREATE TRIGGER monthly_channel_manifest_immutable
|
|
BEFORE UPDATE OR DELETE ON reporting.monthly_channel_manifest
|
|
FOR EACH ROW
|
|
EXECUTE FUNCTION reporting.protect_published_monthly_child();
|
|
|
|
COMMENT ON TABLE reporting.monthly_runs IS
|
|
'Durable publication versions and artifact identities only; no guest-bearing monthly business rows are duplicated.';
|
|
|
|
COMMENT ON COLUMN reporting.monthly_runs.as_of_date IS
|
|
'Greatest ARRIVAL included in this publication; never derived from an upload filename or wall-clock date.';
|
|
|
|
COMMENT ON TABLE reporting.monthly_run_daily_versions IS
|
|
'Immutable current Finance daily-version pins used to build one monthly publication.';
|
|
|
|
COMMENT ON TABLE reporting.monthly_channel_manifest IS
|
|
'Privacy-minimized worksheet order and row-count manifest for one monthly publication.';
|
|
|
|
COMMIT;
|