"""Create the isolated Phase 2A.2 PostgreSQL/PostGIS sidecar schema.

This is a hand-reviewed migration. It contains storage contracts and database
safety guards only; no ingestion, profile engine, Noco adapter, or executor is
implemented here.
"""

from __future__ import annotations

from alembic import op

revision = "0001_create_sidecar_schema"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute("CREATE EXTENSION IF NOT EXISTS postgis")
    op.execute("CREATE SCHEMA osm_lead_source")

    op.execute(
        """
        CREATE TABLE osm_lead_source.source_region (
            region_id text PRIMARY KEY,
            display_name text NOT NULL CHECK (btrim(display_name) <> ''),
            extract_name text NOT NULL CHECK (btrim(extract_name) <> ''),
            active boolean NOT NULL DEFAULT true,
            metadata_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CONSTRAINT source_region_id_nonblank CHECK (btrim(region_id) <> '')
        )
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.source_snapshot (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            region_id text NOT NULL REFERENCES osm_lead_source.source_region(region_id),
            source_url text NOT NULL CHECK (btrim(source_url) <> ''),
            content_sha256 text CHECK (content_sha256 IS NULL OR content_sha256 ~ '^[0-9A-Fa-f]{64}$'),
            size_bytes bigint CHECK (size_bytes IS NULL OR size_bytes > 0),
            etag text,
            last_modified text,
            downloaded_at timestamptz,
            validated_at timestamptz,
            parser_completed_at timestamptz,
            reconciliation_completed_at timestamptz,
            completeness_status text NOT NULL DEFAULT 'unknown'
                CHECK (completeness_status IN ('complete', 'partial', 'failed', 'unknown')),
            extract_metadata_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CONSTRAINT source_snapshot_id_region_key UNIQUE (id, region_id),
            CONSTRAINT source_snapshot_complete_structure CHECK (
                completeness_status <> 'complete'
                OR (content_sha256 IS NOT NULL
                    AND size_bytes IS NOT NULL
                    AND downloaded_at IS NOT NULL
                    AND validated_at IS NOT NULL
                    AND parser_completed_at IS NOT NULL
                    AND reconciliation_completed_at IS NOT NULL)
            ),
            CONSTRAINT source_snapshot_lifecycle_order CHECK (
                (downloaded_at IS NULL OR validated_at IS NULL OR downloaded_at <= validated_at)
                AND (validated_at IS NULL OR parser_completed_at IS NULL
                     OR validated_at <= parser_completed_at)
                AND (parser_completed_at IS NULL OR reconciliation_completed_at IS NULL
                     OR parser_completed_at <= reconciliation_completed_at)
            )
        )
        """
    )
    op.execute(
        """
        CREATE INDEX source_snapshot_region_created_idx
        ON osm_lead_source.source_snapshot (region_id, created_at DESC, downloaded_at DESC)
        """
    )
    op.execute(
        """
        CREATE INDEX source_snapshot_complete_region_idx
        ON osm_lead_source.source_snapshot (region_id, downloaded_at DESC, created_at DESC)
        WHERE completeness_status = 'complete'
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.osm_source_object (
            osm_type text NOT NULL CHECK (osm_type IN ('node', 'way', 'relation')),
            osm_id bigint NOT NULL CHECK (osm_id > 0),
            current_version bigint CHECK (current_version IS NULL OR current_version > 0),
            source_payload_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            source_payload_hash text CHECK (source_payload_hash IS NULL OR source_payload_hash ~ '^[0-9A-Fa-f]{64}$'),
            tags_hash text CHECK (tags_hash IS NULL OR tags_hash ~ '^[0-9A-Fa-f]{64}$'),
            geometry_hash text CHECK (geometry_hash IS NULL OR geometry_hash ~ '^[0-9A-Fa-f]{64}$'),
            geometry geometry(Geometry, 4326),
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (osm_type, osm_id)
        )
        """
    )
    op.execute(
        """
        CREATE INDEX osm_source_object_geometry_gist_idx
        ON osm_lead_source.osm_source_object USING gist (geometry)
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.source_snapshot_object (
            source_snapshot_id bigint NOT NULL
                REFERENCES osm_lead_source.source_snapshot(id),
            osm_type text NOT NULL CHECK (osm_type IN ('node', 'way', 'relation')),
            osm_id bigint NOT NULL CHECK (osm_id > 0),
            object_version bigint NOT NULL CHECK (object_version > 0),
            source_payload_hash text NOT NULL CHECK (source_payload_hash ~ '^[0-9A-Fa-f]{64}$'),
            tags_hash text CHECK (tags_hash IS NULL OR tags_hash ~ '^[0-9A-Fa-f]{64}$'),
            geometry_hash text CHECK (geometry_hash IS NULL OR geometry_hash ~ '^[0-9A-Fa-f]{64}$'),
            observed_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            evidence_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            PRIMARY KEY (source_snapshot_id, osm_type, osm_id),
            FOREIGN KEY (osm_type, osm_id)
                REFERENCES osm_lead_source.osm_source_object(osm_type, osm_id)
        )
        """
    )
    op.execute(
        """
        CREATE INDEX source_snapshot_object_identity_idx
        ON osm_lead_source.source_snapshot_object (osm_type, osm_id, observed_at DESC)
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.osm_source_presence (
            region_id text NOT NULL REFERENCES osm_lead_source.source_region(region_id),
            osm_type text NOT NULL CHECK (osm_type IN ('node', 'way', 'relation')),
            osm_id bigint NOT NULL CHECK (osm_id > 0),
            source_presence_state text NOT NULL CHECK (source_presence_state IN (
                'present', 'missing', 'stale_confirmed', 'remap_candidate', 'deleted_in_osm'
            )),
            first_seen_at timestamptz,
            last_seen_at timestamptz,
            first_missing_at timestamptz,
            consecutive_missing_snapshots integer NOT NULL DEFAULT 0
                CHECK (consecutive_missing_snapshots >= 0),
            last_seen_snapshot_id bigint,
            first_missing_snapshot_id bigint,
            latest_snapshot_id bigint NOT NULL,
            extract_metadata_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            completeness_status text NOT NULL CHECK (completeness_status IN (
                'complete', 'partial', 'failed', 'unknown'
            )),
            boundary_ambiguity boolean NOT NULL DEFAULT false,
            expected_region_evidence_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            updated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (region_id, osm_type, osm_id),
            FOREIGN KEY (osm_type, osm_id)
                REFERENCES osm_lead_source.osm_source_object(osm_type, osm_id),
            FOREIGN KEY (last_seen_snapshot_id, region_id)
                REFERENCES osm_lead_source.source_snapshot(id, region_id),
            FOREIGN KEY (first_missing_snapshot_id, region_id)
                REFERENCES osm_lead_source.source_snapshot(id, region_id),
            FOREIGN KEY (latest_snapshot_id, region_id)
                REFERENCES osm_lead_source.source_snapshot(id, region_id),
            CONSTRAINT presence_present_shape CHECK (
                source_presence_state <> 'present'
                OR (consecutive_missing_snapshots = 0
                    AND first_missing_at IS NULL
                    AND first_missing_snapshot_id IS NULL)
            ),
            CONSTRAINT presence_missing_shape CHECK (
                source_presence_state <> 'missing'
                OR (consecutive_missing_snapshots >= 1
                    AND first_missing_at IS NOT NULL
                    AND first_missing_snapshot_id IS NOT NULL)
            ),
            CONSTRAINT presence_seen_evidence CHECK (
                source_presence_state <> 'stale_confirmed'
                OR (last_seen_at IS NOT NULL
                    AND last_seen_snapshot_id IS NOT NULL
                    AND first_missing_at IS NOT NULL
                    AND first_missing_snapshot_id IS NOT NULL)
            )
        )
        """
    )
    op.execute(
        """
        CREATE INDEX osm_source_presence_state_idx
        ON osm_lead_source.osm_source_presence (region_id, source_presence_state, updated_at DESC)
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.profile_version (
            profile_id text NOT NULL CHECK (btrim(profile_id) <> ''),
            profile_version text NOT NULL CHECK (btrim(profile_version) <> ''),
            lifecycle_state text NOT NULL CHECK (lifecycle_state IN ('draft', 'active', 'retired')),
            definition_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            approval_metadata_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            activated_at timestamptz,
            retired_at timestamptz,
            PRIMARY KEY (profile_id, profile_version),
            CONSTRAINT profile_activation_dates CHECK (
                (lifecycle_state <> 'active' OR activated_at IS NOT NULL)
                AND (lifecycle_state <> 'retired' OR retired_at IS NOT NULL)
            )
        )
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.osm_profile_eligibility (
            region_id text NOT NULL,
            osm_type text NOT NULL CHECK (osm_type IN ('node', 'way', 'relation')),
            osm_id bigint NOT NULL CHECK (osm_id > 0),
            profile_id text NOT NULL,
            profile_version text NOT NULL,
            eligibility_state text NOT NULL CHECK (eligibility_state IN (
                'active', 'out_of_scope', 'excluded', 'pending_review'
            )),
            source_snapshot_id bigint NOT NULL,
            evaluated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            rule_evidence_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            PRIMARY KEY (region_id, osm_type, osm_id, profile_id, profile_version),
            FOREIGN KEY (region_id, osm_type, osm_id)
                REFERENCES osm_lead_source.osm_source_presence(region_id, osm_type, osm_id),
            FOREIGN KEY (source_snapshot_id, region_id)
                REFERENCES osm_lead_source.source_snapshot(id, region_id),
            FOREIGN KEY (profile_id, profile_version)
                REFERENCES osm_lead_source.profile_version(profile_id, profile_version)
        )
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.business_identity (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            normalized_name text,
            normalized_domain text,
            normalized_phone text,
            normalized_address text,
            country text,
            city text,
            geohash text,
            identity_hash text NOT NULL UNIQUE CHECK (btrim(identity_hash) <> ''),
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.noco_lead_mapping (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            noco_table_id text NOT NULL CHECK (btrim(noco_table_id) <> ''),
            noco_record_id bigint NOT NULL CHECK (noco_record_id > 0),
            osm_type text NOT NULL CHECK (osm_type IN ('node', 'way', 'relation')),
            osm_id bigint NOT NULL CHECK (osm_id > 0),
            business_identity_id bigint REFERENCES osm_lead_source.business_identity(id),
            adoption_outcome text NOT NULL CHECK (adoption_outcome IN ('exact', 'high_confidence', 'remapped')),
            adoption_rule text NOT NULL CHECK (btrim(adoption_rule) <> ''),
            confidence numeric CHECK (confidence IS NULL OR confidence BETWEEN 0 AND 1),
            evidence_json jsonb NOT NULL CHECK (evidence_json <> '{}'::jsonb),
            active boolean NOT NULL DEFAULT true,
            first_mapped_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            last_confirmed_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            superseded_at timestamptz,
            superseded_by_mapping_id bigint REFERENCES osm_lead_source.noco_lead_mapping(id),
            FOREIGN KEY (osm_type, osm_id)
                REFERENCES osm_lead_source.osm_source_object(osm_type, osm_id),
            CONSTRAINT noco_mapping_active_shape CHECK (
                (active AND superseded_at IS NULL AND superseded_by_mapping_id IS NULL)
                OR (NOT active AND superseded_at IS NOT NULL)
            ),
            CONSTRAINT noco_mapping_timestamp_order CHECK (
                first_mapped_at <= last_confirmed_at
                AND (superseded_at IS NULL OR superseded_at >= last_confirmed_at)
            ),
            CONSTRAINT noco_mapping_not_self_superseded CHECK (
                superseded_by_mapping_id IS NULL OR superseded_by_mapping_id <> id
            )
        )
        """
    )
    op.execute(
        """
        CREATE UNIQUE INDEX noco_lead_mapping_one_active_noco_row
        ON osm_lead_source.noco_lead_mapping (noco_table_id, noco_record_id)
        WHERE active
        """
    )
    op.execute(
        """
        CREATE UNIQUE INDEX noco_lead_mapping_one_active_source_object
        ON osm_lead_source.noco_lead_mapping (osm_type, osm_id)
        WHERE active
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.reconciliation_run (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            region_id text NOT NULL REFERENCES osm_lead_source.source_region(region_id),
            profile_id text,
            profile_version text,
            source_snapshot_id bigint,
            status text NOT NULL DEFAULT 'planned'
                CHECK (status IN ('planned', 'running', 'completed', 'failed', 'superseded')),
            code_version text NOT NULL CHECK (btrim(code_version) <> ''),
            configuration_hash text NOT NULL CHECK (configuration_hash ~ '^[0-9A-Fa-f]{64}$'),
            started_at timestamptz,
            completed_at timestamptz,
            evidence_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            FOREIGN KEY (profile_id, profile_version)
                REFERENCES osm_lead_source.profile_version(profile_id, profile_version),
            FOREIGN KEY (source_snapshot_id, region_id)
                REFERENCES osm_lead_source.source_snapshot(id, region_id),
            CONSTRAINT run_profile_pair CHECK ((profile_id IS NULL) = (profile_version IS NULL))
        )
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.adoption_candidate (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            noco_table_id text NOT NULL CHECK (btrim(noco_table_id) <> ''),
            noco_record_id bigint NOT NULL CHECK (noco_record_id > 0),
            candidate_osm_type text CHECK (candidate_osm_type IN ('node', 'way', 'relation')),
            candidate_osm_id bigint CHECK (candidate_osm_id IS NULL OR candidate_osm_id > 0),
            result_status text NOT NULL CHECK (result_status IN (
                'ambiguous', 'unmatched', 'rejected', 'collision', 'needs_review', 'accepted'
            )),
            proposed_outcome text CHECK (proposed_outcome IN ('exact', 'high_confidence', 'remapped')),
            confidence numeric CHECK (confidence IS NULL OR confidence BETWEEN 0 AND 1),
            evidence_json jsonb NOT NULL CHECK (evidence_json <> '{}'::jsonb),
            collision_details_json jsonb NOT NULL DEFAULT '{}'::jsonb,
            run_id bigint NOT NULL REFERENCES osm_lead_source.reconciliation_run(id),
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CONSTRAINT candidate_identity_pair CHECK (
                (candidate_osm_type IS NULL) = (candidate_osm_id IS NULL)
            )
        )
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.adoption_candidate_review (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            candidate_id bigint NOT NULL REFERENCES osm_lead_source.adoption_candidate(id),
            review_outcome text NOT NULL CHECK (review_outcome IN (
                'ambiguous', 'unmatched', 'rejected', 'collision', 'needs_review', 'accepted'
            )),
            reviewed_by text NOT NULL CHECK (btrim(reviewed_by) <> ''),
            reviewed_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            review_evidence_json jsonb NOT NULL CHECK (review_evidence_json <> '{}'::jsonb)
        )
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.noco_source_baseline (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            noco_table_id text NOT NULL CHECK (btrim(noco_table_id) <> ''),
            noco_record_id bigint NOT NULL CHECK (noco_record_id > 0),
            mapping_id bigint NOT NULL REFERENCES osm_lead_source.noco_lead_mapping(id),
            source_snapshot_id bigint NOT NULL REFERENCES osm_lead_source.source_snapshot(id),
            previous_baseline_id bigint REFERENCES osm_lead_source.noco_source_baseline(id),
            source_payload_json jsonb NOT NULL,
            source_payload_hash text NOT NULL CHECK (source_payload_hash ~ '^[0-9A-Fa-f]{64}$'),
            raw_tags_json_hash text CHECK (raw_tags_json_hash IS NULL OR raw_tags_json_hash ~ '^[0-9A-Fa-f]{64}$'),
            source_owned_field_hashes jsonb NOT NULL,
            source_owned_field_values jsonb NOT NULL DEFAULT '{}'::jsonb,
            valid_from timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            superseded_at timestamptz,
            code_version text NOT NULL CHECK (btrim(code_version) <> ''),
            rule_version text NOT NULL CHECK (btrim(rule_version) <> ''),
            provenance_evidence_json jsonb NOT NULL CHECK (provenance_evidence_json <> '{}'::jsonb)
        )
        """
    )
    op.execute(
        """
        CREATE UNIQUE INDEX noco_source_baseline_one_current_mapping
        ON osm_lead_source.noco_source_baseline (mapping_id)
        WHERE superseded_at IS NULL
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.reconciliation_action (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            run_id bigint NOT NULL REFERENCES osm_lead_source.reconciliation_run(id),
            action_type text NOT NULL CHECK (action_type IN (
                'INSERT', 'UPDATE', 'NO_CHANGE', 'SOURCE_REMAP', 'MARK_STALE_EXTERNALLY',
                'DELETE_SAFE_STALE', 'SKIP_PROTECTED', 'SKIP_AMBIGUOUS'
            )),
            state text NOT NULL DEFAULT 'planned'
                CHECK (state IN ('planned', 'claimed', 'applied', 'failed', 'superseded')),
            planning_key text NOT NULL UNIQUE CHECK (btrim(planning_key) <> ''),
            idempotency_key text NOT NULL UNIQUE CHECK (btrim(idempotency_key) <> ''),
            noco_table_id text CHECK (noco_table_id IS NULL OR btrim(noco_table_id) <> ''),
            noco_record_id bigint CHECK (noco_record_id IS NULL OR noco_record_id > 0),
            osm_type text CHECK (osm_type IS NULL OR osm_type IN ('node', 'way', 'relation')),
            osm_id bigint CHECK (osm_id IS NULL OR osm_id > 0),
            mapping_id bigint REFERENCES osm_lead_source.noco_lead_mapping(id),
            payload_hash text CHECK (payload_hash IS NULL OR payload_hash ~ '^[0-9A-Fa-f]{64}$'),
            reason_codes_json jsonb NOT NULL CHECK (reason_codes_json <> '{}'::jsonb),
            evidence_json jsonb NOT NULL CHECK (evidence_json <> '{}'::jsonb),
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CONSTRAINT action_identity_pair CHECK ((osm_type IS NULL) = (osm_id IS NULL)),
            CONSTRAINT action_noco_pair CHECK ((noco_table_id IS NULL) = (noco_record_id IS NULL))
        )
        """
    )

    op.execute(
        """
        CREATE TABLE osm_lead_source.reconciliation_report (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            run_id bigint NOT NULL REFERENCES osm_lead_source.reconciliation_run(id),
            report_hash text NOT NULL UNIQUE CHECK (report_hash ~ '^[0-9A-Fa-f]{64}$'),
            report_json jsonb NOT NULL,
            created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    op.execute(
        """
        CREATE TABLE osm_lead_source.osm_lead_tombstone (
            id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
            noco_table_id text NOT NULL CHECK (btrim(noco_table_id) <> ''),
            previous_noco_record_id bigint NOT NULL CHECK (previous_noco_record_id > 0),
            deleted_at timestamptz NOT NULL,
            deleted_by_service_run_id bigint NOT NULL REFERENCES osm_lead_source.reconciliation_run(id),
            deletion_reason text NOT NULL CHECK (btrim(deletion_reason) <> ''),
            prior_osm_identities jsonb NOT NULL,
            normalized_business_identity jsonb NOT NULL,
            normalized_business_identity_hash text NOT NULL CHECK (btrim(normalized_business_identity_hash) <> ''),
            last_known_name text,
            last_known_domain text,
            last_known_phone text,
            last_known_address text,
            last_known_lat double precision CHECK (last_known_lat IS NULL OR last_known_lat BETWEEN -90 AND 90),
            last_known_lon double precision CHECK (last_known_lon IS NULL OR last_known_lon BETWEEN -180 AND 180),
            last_known_country text,
            source_payload_hash text NOT NULL CHECK (source_payload_hash ~ '^[0-9A-Fa-f]{64}$'),
            safety_evidence_json jsonb NOT NULL CHECK (safety_evidence_json <> '{}'::jsonb),
            superseded_at timestamptz,
            superseded_reason text,
            CONSTRAINT tombstone_supersession_pair CHECK (
                (superseded_at IS NULL AND superseded_reason IS NULL)
                OR (superseded_at IS NOT NULL AND superseded_reason IS NOT NULL
                    AND btrim(superseded_reason) <> '')
            )
        )
        """
    )

    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.immutable_source_snapshot()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            RAISE EXCEPTION 'source snapshot evidence is append-only';
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.immutable_snapshot_observation()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            RAISE EXCEPTION 'source snapshot observations are immutable';
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.immutable_candidate_evidence()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            RAISE EXCEPTION 'adoption candidate evidence is immutable';
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.immutable_candidate_review()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            RAISE EXCEPTION 'adoption candidate reviews are immutable';
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.immutable_reconciliation_report()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            RAISE EXCEPTION 'reconciliation reports are immutable';
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.guard_baseline_chain()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        DECLARE
            mapped_table text;
            mapped_record bigint;
            previous_baseline osm_lead_source.noco_source_baseline%ROWTYPE;
        BEGIN
            SELECT noco_table_id, noco_record_id
            INTO mapped_table, mapped_record
            FROM osm_lead_source.noco_lead_mapping
            WHERE id = NEW.mapping_id;
            IF NOT FOUND THEN
                RAISE EXCEPTION 'baseline mapping does not exist';
            END IF;
            IF NEW.noco_table_id IS DISTINCT FROM mapped_table
               OR NEW.noco_record_id IS DISTINCT FROM mapped_record THEN
                RAISE EXCEPTION 'baseline Noco identity must match its mapping';
            END IF;

            SELECT * INTO previous_baseline
            FROM osm_lead_source.noco_source_baseline
            WHERE mapping_id = NEW.mapping_id
            ORDER BY valid_from DESC, id DESC
            LIMIT 1;
            IF previous_baseline.id IS NULL THEN
                IF NEW.previous_baseline_id IS NOT NULL THEN
                    RAISE EXCEPTION 'the first baseline cannot reference a previous baseline';
                END IF;
            ELSE
                IF NEW.previous_baseline_id IS DISTINCT FROM previous_baseline.id THEN
                    RAISE EXCEPTION 'baseline must reference the immediate prior baseline';
                END IF;
                IF previous_baseline.superseded_at IS NULL THEN
                    RAISE EXCEPTION 'the previous baseline must be superseded first';
                END IF;
                IF NEW.valid_from <= previous_baseline.valid_from THEN
                    RAISE EXCEPTION 'baseline valid_from must advance chronologically';
                END IF;
                IF NEW.valid_from < previous_baseline.superseded_at THEN
                    RAISE EXCEPTION 'baseline valid_from cannot precede prior supersession';
                END IF;
            END IF;
            RETURN NEW;
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.guard_mapping_lifecycle()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        DECLARE
            target_table text;
            target_record bigint;
        BEGIN
            IF TG_OP = 'DELETE' THEN
                RAISE EXCEPTION 'accepted mappings cannot be deleted';
            END IF;

            IF TG_OP = 'INSERT' THEN
                IF NEW.superseded_by_mapping_id IS NOT NULL THEN
                    SELECT noco_table_id, noco_record_id
                    INTO target_table, target_record
                    FROM osm_lead_source.noco_lead_mapping
                    WHERE id = NEW.superseded_by_mapping_id;
                IF NOT FOUND
                   OR target_table IS DISTINCT FROM NEW.noco_table_id
                   OR target_record IS DISTINCT FROM NEW.noco_record_id
                   OR NOT (SELECT active FROM osm_lead_source.noco_lead_mapping
                           WHERE id = NEW.superseded_by_mapping_id)
                   OR EXISTS (SELECT 1 FROM osm_lead_source.noco_lead_mapping
                              WHERE id = NEW.superseded_by_mapping_id
                                AND superseded_at IS NOT NULL)
                   OR NEW.superseded_by_mapping_id = NEW.id THEN
                        RAISE EXCEPTION 'superseded mapping must target a different active mapping for the same Noco row';
                    END IF;
                END IF;
                RETURN NEW;
            END IF;

            IF OLD.active THEN
                IF NEW.active THEN
                    IF NEW.superseded_at IS NOT NULL
                       OR NEW.superseded_by_mapping_id IS NOT NULL THEN
                        RAISE EXCEPTION 'active mappings cannot carry supersession fields';
                    END IF;
                    IF NEW.noco_table_id IS DISTINCT FROM OLD.noco_table_id
                       OR NEW.noco_record_id IS DISTINCT FROM OLD.noco_record_id
                       OR NEW.osm_type IS DISTINCT FROM OLD.osm_type
                       OR NEW.osm_id IS DISTINCT FROM OLD.osm_id
                       OR NEW.business_identity_id IS DISTINCT FROM OLD.business_identity_id
                       OR NEW.adoption_outcome IS DISTINCT FROM OLD.adoption_outcome
                       OR NEW.adoption_rule IS DISTINCT FROM OLD.adoption_rule
                       OR NEW.confidence IS DISTINCT FROM OLD.confidence
                       OR NEW.evidence_json IS DISTINCT FROM OLD.evidence_json
                       OR NEW.first_mapped_at IS DISTINCT FROM OLD.first_mapped_at THEN
                        RAISE EXCEPTION 'mapping identity and adoption evidence are immutable';
                    END IF;
                    RETURN NEW;
                END IF;

                IF NEW.superseded_at IS NULL
                   OR NEW.superseded_by_mapping_id IS NOT NULL THEN
                    RAISE EXCEPTION 'deactivation must be one-way and set superseded_at only';
                END IF;
                IF NEW.noco_table_id IS DISTINCT FROM OLD.noco_table_id
                   OR NEW.noco_record_id IS DISTINCT FROM OLD.noco_record_id
                   OR NEW.osm_type IS DISTINCT FROM OLD.osm_type
                   OR NEW.osm_id IS DISTINCT FROM OLD.osm_id
                   OR NEW.business_identity_id IS DISTINCT FROM OLD.business_identity_id
                   OR NEW.adoption_outcome IS DISTINCT FROM OLD.adoption_outcome
                   OR NEW.adoption_rule IS DISTINCT FROM OLD.adoption_rule
                   OR NEW.confidence IS DISTINCT FROM OLD.confidence
                   OR NEW.evidence_json IS DISTINCT FROM OLD.evidence_json
                   OR NEW.first_mapped_at IS DISTINCT FROM OLD.first_mapped_at
                   OR NEW.last_confirmed_at IS DISTINCT FROM OLD.last_confirmed_at THEN
                    RAISE EXCEPTION 'mapping identity and adoption evidence are immutable';
                END IF;
                RETURN NEW;
            END IF;

            IF NEW.active
               OR NEW.superseded_at IS DISTINCT FROM OLD.superseded_at
               OR NEW.noco_table_id IS DISTINCT FROM OLD.noco_table_id
               OR NEW.noco_record_id IS DISTINCT FROM OLD.noco_record_id
               OR NEW.osm_type IS DISTINCT FROM OLD.osm_type
               OR NEW.osm_id IS DISTINCT FROM OLD.osm_id
               OR NEW.business_identity_id IS DISTINCT FROM OLD.business_identity_id
               OR NEW.adoption_outcome IS DISTINCT FROM OLD.adoption_outcome
               OR NEW.adoption_rule IS DISTINCT FROM OLD.adoption_rule
               OR NEW.confidence IS DISTINCT FROM OLD.confidence
               OR NEW.evidence_json IS DISTINCT FROM OLD.evidence_json
               OR NEW.first_mapped_at IS DISTINCT FROM OLD.first_mapped_at
               OR NEW.last_confirmed_at IS DISTINCT FROM OLD.last_confirmed_at THEN
                RAISE EXCEPTION 'superseded mappings are immutable';
            END IF;
            IF NEW.superseded_by_mapping_id IS DISTINCT FROM OLD.superseded_by_mapping_id THEN
                IF OLD.superseded_by_mapping_id IS NOT NULL
                   OR NEW.superseded_by_mapping_id IS NULL THEN
                    RAISE EXCEPTION 'superseded_by_mapping_id may be set only once';
                END IF;
                SELECT noco_table_id, noco_record_id
                INTO target_table, target_record
                FROM osm_lead_source.noco_lead_mapping
                WHERE id = NEW.superseded_by_mapping_id;
                IF NOT FOUND
                   OR target_table IS DISTINCT FROM NEW.noco_table_id
                   OR target_record IS DISTINCT FROM NEW.noco_record_id
                   OR NOT (SELECT active FROM osm_lead_source.noco_lead_mapping
                           WHERE id = NEW.superseded_by_mapping_id)
                   OR EXISTS (SELECT 1 FROM osm_lead_source.noco_lead_mapping
                              WHERE id = NEW.superseded_by_mapping_id
                                AND superseded_at IS NOT NULL)
                   OR NEW.superseded_by_mapping_id = NEW.id THEN
                    RAISE EXCEPTION 'superseded mapping must target a different active mapping for the same Noco row';
                END IF;
            END IF;
            RETURN NEW;
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.guard_baseline_immutability()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            IF TG_OP = 'DELETE' THEN
                RAISE EXCEPTION 'source baselines are append-only';
            END IF;
            IF OLD.superseded_at IS NOT NULL THEN
                RAISE EXCEPTION 'a superseded baseline cannot be changed';
            END IF;
            IF NEW.superseded_at IS DISTINCT FROM OLD.superseded_at THEN
                IF NEW.superseded_at IS NULL
                   OR NEW.superseded_at < OLD.valid_from
                   OR NEW.noco_table_id IS DISTINCT FROM OLD.noco_table_id
                   OR NEW.noco_record_id IS DISTINCT FROM OLD.noco_record_id
                   OR NEW.mapping_id IS DISTINCT FROM OLD.mapping_id
                   OR NEW.source_snapshot_id IS DISTINCT FROM OLD.source_snapshot_id
                   OR NEW.previous_baseline_id IS DISTINCT FROM OLD.previous_baseline_id
                   OR NEW.source_payload_json IS DISTINCT FROM OLD.source_payload_json
                   OR NEW.source_payload_hash IS DISTINCT FROM OLD.source_payload_hash
                   OR NEW.raw_tags_json_hash IS DISTINCT FROM OLD.raw_tags_json_hash
                   OR NEW.source_owned_field_hashes IS DISTINCT FROM OLD.source_owned_field_hashes
                   OR NEW.source_owned_field_values IS DISTINCT FROM OLD.source_owned_field_values
                   OR NEW.valid_from IS DISTINCT FROM OLD.valid_from
                   OR NEW.code_version IS DISTINCT FROM OLD.code_version
                   OR NEW.rule_version IS DISTINCT FROM OLD.rule_version
                   OR NEW.provenance_evidence_json IS DISTINCT FROM OLD.provenance_evidence_json THEN
                    RAISE EXCEPTION 'only one-way superseded_at may change on a current baseline';
                END IF;
                RETURN NEW;
            END IF;
            RAISE EXCEPTION 'source baseline fields are immutable';
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.guard_tombstone_immutability()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        BEGIN
            IF TG_OP = 'DELETE' THEN
                RAISE EXCEPTION 'tombstone evidence is immutable';
            END IF;
            IF OLD.superseded_at IS NOT NULL
               OR NEW.superseded_at IS NULL
               OR NEW.superseded_at <= OLD.deleted_at
               OR NEW.noco_table_id IS DISTINCT FROM OLD.noco_table_id
               OR NEW.previous_noco_record_id IS DISTINCT FROM OLD.previous_noco_record_id
               OR NEW.deleted_at IS DISTINCT FROM OLD.deleted_at
               OR NEW.deleted_by_service_run_id IS DISTINCT FROM OLD.deleted_by_service_run_id
               OR NEW.deletion_reason IS DISTINCT FROM OLD.deletion_reason
               OR NEW.prior_osm_identities IS DISTINCT FROM OLD.prior_osm_identities
               OR NEW.normalized_business_identity IS DISTINCT FROM OLD.normalized_business_identity
               OR NEW.normalized_business_identity_hash IS DISTINCT FROM OLD.normalized_business_identity_hash
               OR NEW.last_known_name IS DISTINCT FROM OLD.last_known_name
               OR NEW.last_known_domain IS DISTINCT FROM OLD.last_known_domain
               OR NEW.last_known_phone IS DISTINCT FROM OLD.last_known_phone
               OR NEW.last_known_address IS DISTINCT FROM OLD.last_known_address
               OR NEW.last_known_lat IS DISTINCT FROM OLD.last_known_lat
               OR NEW.last_known_lon IS DISTINCT FROM OLD.last_known_lon
               OR NEW.last_known_country IS DISTINCT FROM OLD.last_known_country
               OR NEW.source_payload_hash IS DISTINCT FROM OLD.source_payload_hash
               OR NEW.safety_evidence_json IS DISTINCT FROM OLD.safety_evidence_json THEN
                RAISE EXCEPTION 'tombstone is immutable except for one-way supersession';
            END IF;
            RETURN NEW;
        END;
        $$
        """
    )
    op.execute(
        """
        CREATE OR REPLACE FUNCTION osm_lead_source.guard_presence_transition()
        RETURNS trigger LANGUAGE plpgsql SET search_path = osm_lead_source, pg_catalog AS $$
        DECLARE
            latest_region text;
            latest_completeness_status text;
            latest_reconciliation_completed_at timestamptz;
            previous_latest_reconciliation_completed_at timestamptz;
            first_missing_completeness_status text;
            first_missing_reconciliation_completed_at timestamptz;
            latest_observed_at timestamptz;
            last_seen_observed_at timestamptz;
        BEGIN
            SELECT region_id, completeness_status, reconciliation_completed_at
            INTO latest_region, latest_completeness_status, latest_reconciliation_completed_at
            FROM osm_lead_source.source_snapshot WHERE id = NEW.latest_snapshot_id;
            IF NOT FOUND THEN
                RAISE EXCEPTION 'latest presence snapshot does not exist';
            END IF;
            IF latest_region IS DISTINCT FROM NEW.region_id THEN
                RAISE EXCEPTION 'presence snapshot must belong to the same region';
            END IF;
            IF NEW.completeness_status IS DISTINCT FROM latest_completeness_status THEN
                RAISE EXCEPTION 'presence completeness must match its latest snapshot';
            END IF;

            IF NEW.source_presence_state = 'present'
               AND NEW.last_seen_snapshot_id IS NULL
               AND NEW.last_seen_at IS NULL THEN
                SELECT observed_at
                INTO latest_observed_at
                FROM osm_lead_source.source_snapshot_object
                WHERE source_snapshot_id = NEW.latest_snapshot_id
                  AND osm_type = NEW.osm_type AND osm_id = NEW.osm_id;
                IF latest_observed_at IS NULL THEN
                    RAISE EXCEPTION 'present state requires a latest observation';
                END IF;
                NEW.last_seen_snapshot_id := NEW.latest_snapshot_id;
                NEW.last_seen_at := latest_observed_at;
            END IF;

            IF (NEW.last_seen_snapshot_id IS NULL) <> (NEW.last_seen_at IS NULL) THEN
                RAISE EXCEPTION 'last-seen snapshot and timestamp must be supplied together';
            END IF;
            IF NEW.last_seen_snapshot_id IS NOT NULL THEN
                SELECT o.observed_at
                INTO last_seen_observed_at
                FROM osm_lead_source.source_snapshot_object AS o
                JOIN osm_lead_source.source_snapshot AS s
                  ON s.id = o.source_snapshot_id AND s.region_id = NEW.region_id
                WHERE o.source_snapshot_id = NEW.last_seen_snapshot_id
                  AND o.osm_type = NEW.osm_type AND o.osm_id = NEW.osm_id;
                IF last_seen_observed_at IS NULL THEN
                    RAISE EXCEPTION 'last-seen snapshot must contain the object';
                END IF;
                IF NEW.last_seen_at IS DISTINCT FROM last_seen_observed_at THEN
                    RAISE EXCEPTION 'last_seen_at must match the immutable observation timestamp';
                END IF;
            END IF;

            IF NEW.first_missing_snapshot_id IS NOT NULL THEN
                SELECT completeness_status, reconciliation_completed_at
                INTO first_missing_completeness_status, first_missing_reconciliation_completed_at
                FROM osm_lead_source.source_snapshot
                WHERE id = NEW.first_missing_snapshot_id AND region_id = NEW.region_id;
                IF NOT FOUND OR first_missing_completeness_status IS DISTINCT FROM 'complete'
                   OR first_missing_reconciliation_completed_at IS NULL THEN
                    RAISE EXCEPTION 'first-missing snapshot must be complete';
                END IF;
                IF EXISTS (
                    SELECT 1 FROM osm_lead_source.source_snapshot_object
                    WHERE source_snapshot_id = NEW.first_missing_snapshot_id
                      AND osm_type = NEW.osm_type AND osm_id = NEW.osm_id
                ) THEN
                    RAISE EXCEPTION 'first-missing snapshot must not contain the object';
                END IF;
                IF NEW.first_missing_at IS NULL THEN
                    NEW.first_missing_at := first_missing_reconciliation_completed_at;
                ELSIF NEW.first_missing_at IS DISTINCT FROM first_missing_reconciliation_completed_at THEN
                    RAISE EXCEPTION 'first_missing_at must match snapshot reconciliation completion';
                END IF;
            ELSIF NEW.first_missing_at IS NOT NULL THEN
                RAISE EXCEPTION 'first-missing snapshot and timestamp must be supplied together';
            END IF;

            IF NEW.source_presence_state = 'present' THEN
                IF NEW.consecutive_missing_snapshots <> 0
                   OR NEW.first_missing_at IS NOT NULL
                   OR NEW.first_missing_snapshot_id IS NOT NULL
                   OR NEW.last_seen_snapshot_id IS DISTINCT FROM NEW.latest_snapshot_id
                   OR last_seen_observed_at IS NULL
                   OR NEW.last_seen_at IS DISTINCT FROM last_seen_observed_at THEN
                    RAISE EXCEPTION 'present state requires a latest observation and zero missing evidence';
                END IF;
                SELECT observed_at INTO latest_observed_at
                FROM osm_lead_source.source_snapshot_object
                WHERE source_snapshot_id = NEW.latest_snapshot_id
                  AND osm_type = NEW.osm_type AND osm_id = NEW.osm_id;
                IF latest_observed_at IS NULL
                   OR NEW.last_seen_at IS DISTINCT FROM latest_observed_at THEN
                    RAISE EXCEPTION 'present state requires latest immutable observation evidence';
                END IF;
                IF NEW.first_seen_at IS NOT NULL AND NEW.first_seen_at > NEW.last_seen_at THEN
                    RAISE EXCEPTION 'first_seen_at cannot be later than last_seen_at';
                END IF;
            ELSE
                IF latest_completeness_status IS DISTINCT FROM 'complete'
                   OR latest_reconciliation_completed_at IS NULL THEN
                    RAISE EXCEPTION 'latest absence snapshot must be complete with a reconciliation timestamp';
                END IF;
                IF EXISTS (
                    SELECT 1 FROM osm_lead_source.source_snapshot_object
                    WHERE source_snapshot_id = NEW.latest_snapshot_id
                      AND osm_type = NEW.osm_type AND osm_id = NEW.osm_id
                ) THEN
                    RAISE EXCEPTION 'absence state cannot reference a snapshot containing the object';
                END IF;
                IF NEW.first_missing_snapshot_id IS NULL OR NEW.first_missing_at IS NULL THEN
                    RAISE EXCEPTION 'absence state requires first-missing evidence';
                END IF;
            END IF;

            IF TG_OP = 'INSERT' THEN
                IF NEW.source_presence_state = 'present' THEN
                    NULL;
                ELSIF NEW.source_presence_state = 'missing' THEN
                    IF NEW.consecutive_missing_snapshots <> 1
                       OR NEW.first_missing_snapshot_id IS DISTINCT FROM NEW.latest_snapshot_id
                       OR NEW.first_missing_at IS DISTINCT FROM first_missing_reconciliation_completed_at THEN
                        RAISE EXCEPTION 'initial missing state requires exactly one missing snapshot';
                    END IF;
                ELSE
                    RAISE EXCEPTION 'presence must begin as present or missing';
                END IF;
            ELSE
                IF NEW.source_presence_state = 'present' THEN
                    IF OLD.source_presence_state NOT IN (
                        'present', 'missing', 'stale_confirmed', 'remap_candidate', 'deleted_in_osm'
                    ) THEN
                        RAISE EXCEPTION 'invalid transition to present';
                    END IF;
                ELSIF OLD.source_presence_state = 'present'
                      AND NEW.source_presence_state = 'missing' THEN
                    IF NEW.consecutive_missing_snapshots <> 1
                       OR NEW.first_missing_snapshot_id IS DISTINCT FROM NEW.latest_snapshot_id
                       OR NEW.first_missing_at IS DISTINCT FROM first_missing_reconciliation_completed_at
                       OR NEW.latest_snapshot_id = OLD.latest_snapshot_id THEN
                        RAISE EXCEPTION 'first missing transition requires one new complete snapshot';
                    END IF;
                ELSIF OLD.source_presence_state = 'missing'
                      AND NEW.source_presence_state IN ('missing', 'stale_confirmed', 'remap_candidate', 'deleted_in_osm') THEN
                    IF NEW.last_seen_snapshot_id IS DISTINCT FROM OLD.last_seen_snapshot_id
                       OR NEW.last_seen_at IS DISTINCT FROM OLD.last_seen_at
                       OR NEW.first_missing_snapshot_id IS DISTINCT FROM OLD.first_missing_snapshot_id
                       OR NEW.first_missing_at IS DISTINCT FROM OLD.first_missing_at THEN
                        RAISE EXCEPTION 'absence evidence is immutable';
                    END IF;
                    IF NEW.latest_snapshot_id = OLD.latest_snapshot_id THEN
                        IF NEW.consecutive_missing_snapshots <> OLD.consecutive_missing_snapshots
                           OR NEW.source_presence_state <> 'missing' THEN
                            RAISE EXCEPTION 'the same snapshot cannot advance absence evidence';
                        END IF;
                    ELSE
                        SELECT reconciliation_completed_at
                        INTO previous_latest_reconciliation_completed_at
                        FROM osm_lead_source.source_snapshot
                        WHERE id = OLD.latest_snapshot_id;
                        IF NEW.consecutive_missing_snapshots <> OLD.consecutive_missing_snapshots + 1
                           OR previous_latest_reconciliation_completed_at IS NULL
                           OR latest_reconciliation_completed_at IS NULL
                           OR latest_reconciliation_completed_at <= previous_latest_reconciliation_completed_at THEN
                            RAISE EXCEPTION 'missing evidence must advance by one later snapshot';
                        END IF;
                    END IF;
                    IF NEW.source_presence_state IN ('remap_candidate', 'deleted_in_osm')
                       AND NEW.boundary_ambiguity THEN
                        RAISE EXCEPTION 'boundary ambiguity blocks remap or deletion state';
                    END IF;
                ELSE
                    RAISE EXCEPTION 'invalid source presence transition';
                END IF;

                IF NEW.source_presence_state = 'stale_confirmed' THEN
                    IF OLD.source_presence_state <> 'missing'
                       OR NEW.latest_snapshot_id = OLD.latest_snapshot_id
                       OR NEW.consecutive_missing_snapshots <> OLD.consecutive_missing_snapshots + 1
                       OR NEW.consecutive_missing_snapshots < 2
                       OR NEW.first_missing_snapshot_id = NEW.latest_snapshot_id
                       OR NEW.last_seen_snapshot_id IS NULL
                       OR NEW.last_seen_at IS NULL
                       OR NEW.last_seen_at > CURRENT_TIMESTAMP - INTERVAL '60 days'
                       OR NEW.boundary_ambiguity
                       OR NEW.last_seen_at >= NEW.first_missing_at
                       OR latest_reconciliation_completed_at <= first_missing_reconciliation_completed_at THEN
                        RAISE EXCEPTION 'stale confirmation requires two distinct ordered complete absences, old presence evidence, 60 days, and clear boundaries';
                    END IF;
                END IF;
            END IF;
            NEW.updated_at := CURRENT_TIMESTAMP;
            RETURN NEW;
        END;
        $$
        """
    )

    op.execute(
        """
        CREATE TRIGGER source_snapshot_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.source_snapshot
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.immutable_source_snapshot()
        """
    )
    op.execute(
        """
        CREATE TRIGGER source_snapshot_object_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.source_snapshot_object
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.immutable_snapshot_observation()
        """
    )
    op.execute(
        """
        CREATE TRIGGER adoption_candidate_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.adoption_candidate
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.immutable_candidate_evidence()
        """
    )
    op.execute(
        """
        CREATE TRIGGER adoption_candidate_review_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.adoption_candidate_review
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.immutable_candidate_review()
        """
    )
    op.execute(
        """
        CREATE TRIGGER reconciliation_report_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.reconciliation_report
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.immutable_reconciliation_report()
        """
    )
    op.execute(
        """
        CREATE TRIGGER noco_lead_mapping_lifecycle
        BEFORE INSERT OR UPDATE OR DELETE ON osm_lead_source.noco_lead_mapping
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.guard_mapping_lifecycle()
        """
    )
    op.execute(
        """
        CREATE TRIGGER noco_source_baseline_chain
        BEFORE INSERT ON osm_lead_source.noco_source_baseline
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.guard_baseline_chain()
        """
    )
    op.execute(
        """
        CREATE TRIGGER noco_source_baseline_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.noco_source_baseline
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.guard_baseline_immutability()
        """
    )
    op.execute(
        """
        CREATE TRIGGER osm_lead_tombstone_immutable
        BEFORE UPDATE OR DELETE ON osm_lead_source.osm_lead_tombstone
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.guard_tombstone_immutability()
        """
    )
    op.execute(
        """
        CREATE TRIGGER osm_source_presence_transition
        BEFORE INSERT OR UPDATE ON osm_lead_source.osm_source_presence
        FOR EACH ROW EXECUTE FUNCTION osm_lead_source.guard_presence_transition()
        """
    )

    op.execute(
        """
        CREATE VIEW osm_lead_source.v_current_noco_lead_mapping AS
        SELECT * FROM osm_lead_source.noco_lead_mapping WHERE active
        """
    )
    op.execute(
        """
        CREATE VIEW osm_lead_source.v_current_noco_source_baseline AS
        SELECT * FROM osm_lead_source.noco_source_baseline WHERE superseded_at IS NULL
        """
    )
    op.execute(
        """
        CREATE VIEW osm_lead_source.v_latest_complete_source_snapshot AS
        SELECT DISTINCT ON (region_id) *
        FROM osm_lead_source.source_snapshot
        WHERE completeness_status = 'complete'
        ORDER BY region_id, COALESCE(downloaded_at, created_at) DESC, id DESC
        """
    )
    op.execute(
        """
        CREATE VIEW osm_lead_source.v_source_presence_profile_eligibility AS
        SELECT
            p.region_id, p.osm_type, p.osm_id,
            p.source_presence_state,
            p.first_seen_at, p.last_seen_at, p.first_missing_at,
            p.consecutive_missing_snapshots, p.latest_snapshot_id,
            p.completeness_status, p.boundary_ambiguity,
            e.profile_id, e.profile_version, e.eligibility_state,
            e.source_snapshot_id AS eligibility_snapshot_id,
            e.evaluated_at, e.rule_evidence_json
        FROM osm_lead_source.osm_source_presence AS p
        JOIN osm_lead_source.osm_profile_eligibility AS e
          ON e.region_id = p.region_id
         AND e.osm_type = p.osm_type
         AND e.osm_id = p.osm_id
        """
    )


def downgrade() -> None:
    op.execute("DROP VIEW osm_lead_source.v_source_presence_profile_eligibility")
    op.execute("DROP VIEW osm_lead_source.v_latest_complete_source_snapshot")
    op.execute("DROP VIEW osm_lead_source.v_current_noco_source_baseline")
    op.execute("DROP VIEW osm_lead_source.v_current_noco_lead_mapping")

    op.execute("DROP TRIGGER osm_source_presence_transition ON osm_lead_source.osm_source_presence")
    op.execute("DROP TRIGGER osm_lead_tombstone_immutable ON osm_lead_source.osm_lead_tombstone")
    op.execute(
        "DROP TRIGGER noco_source_baseline_immutable ON osm_lead_source.noco_source_baseline"
    )
    op.execute("DROP TRIGGER noco_source_baseline_chain ON osm_lead_source.noco_source_baseline")
    op.execute(
        "DROP TRIGGER reconciliation_report_immutable ON osm_lead_source.reconciliation_report"
    )
    op.execute(
        "DROP TRIGGER adoption_candidate_review_immutable ON osm_lead_source.adoption_candidate_review"
    )
    op.execute("DROP TRIGGER adoption_candidate_immutable ON osm_lead_source.adoption_candidate")
    op.execute(
        "DROP TRIGGER source_snapshot_object_immutable ON osm_lead_source.source_snapshot_object"
    )
    op.execute("DROP TRIGGER source_snapshot_immutable ON osm_lead_source.source_snapshot")
    op.execute("DROP TRIGGER noco_lead_mapping_lifecycle ON osm_lead_source.noco_lead_mapping")

    op.execute("DROP FUNCTION osm_lead_source.guard_presence_transition()")
    op.execute("DROP FUNCTION osm_lead_source.guard_mapping_lifecycle()")
    op.execute("DROP FUNCTION osm_lead_source.guard_tombstone_immutability()")
    op.execute("DROP FUNCTION osm_lead_source.guard_baseline_immutability()")
    op.execute("DROP FUNCTION osm_lead_source.guard_baseline_chain()")
    op.execute("DROP FUNCTION osm_lead_source.immutable_reconciliation_report()")
    op.execute("DROP FUNCTION osm_lead_source.immutable_candidate_review()")
    op.execute("DROP FUNCTION osm_lead_source.immutable_candidate_evidence()")
    op.execute("DROP FUNCTION osm_lead_source.immutable_snapshot_observation()")
    op.execute("DROP FUNCTION osm_lead_source.immutable_source_snapshot()")

    op.execute("DROP TABLE osm_lead_source.osm_lead_tombstone")
    op.execute("DROP TABLE osm_lead_source.reconciliation_report")
    op.execute("DROP TABLE osm_lead_source.reconciliation_action")
    op.execute("DROP TABLE osm_lead_source.noco_source_baseline")
    op.execute("DROP TABLE osm_lead_source.adoption_candidate_review")
    op.execute("DROP TABLE osm_lead_source.adoption_candidate")
    op.execute("DROP TABLE osm_lead_source.reconciliation_run")
    op.execute("DROP TABLE osm_lead_source.noco_lead_mapping")
    op.execute("DROP TABLE osm_lead_source.business_identity")
    op.execute("DROP TABLE osm_lead_source.osm_profile_eligibility")
    op.execute("DROP TABLE osm_lead_source.profile_version")
    op.execute("DROP TABLE osm_lead_source.osm_source_presence")
    op.execute("DROP TABLE osm_lead_source.source_snapshot_object")
    op.execute("DROP TABLE osm_lead_source.osm_source_object")
    op.execute("DROP TABLE osm_lead_source.source_snapshot")
    op.execute("DROP TABLE osm_lead_source.source_region")
    op.execute("DROP SCHEMA osm_lead_source")