"""Create the initial RAG schema and vector index.

Revision ID: 0001_initial_schema
Revises: None
Create Date: 2026-07-12
"""

from collections.abc import Sequence

from alembic import op

revision: str = "0001_initial_schema"
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    op.execute(
        """
        DO $migration$
        BEGIN
          IF NOT EXISTS (
            SELECT 1 FROM pg_extension WHERE extname = 'vector'
          ) THEN
            RAISE EXCEPTION
              'pgvector extension is required; run the PostgreSQL bootstrap first';
          END IF;

          IF NOT EXISTS (
            SELECT 1 FROM pg_namespace WHERE nspname = 'rag'
          ) THEN
            RAISE EXCEPTION
              'rag schema is required; run the PostgreSQL bootstrap first';
          END IF;
        END
        $migration$;
        """
    )

    op.execute(
        """
        CREATE TABLE rag.knowledge_bases (
          id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          name text NOT NULL,
          description text,
          created_at timestamptz NOT NULL DEFAULT now(),
          updated_at timestamptz NOT NULL DEFAULT now(),
          CONSTRAINT knowledge_bases_name_nonempty
            CHECK (btrim(name) <> '')
        );
        """
    )

    op.execute(
        """
        CREATE TABLE rag.access_scopes (
          id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          knowledge_base_id uuid NOT NULL,
          name text NOT NULL,
          created_at timestamptz NOT NULL DEFAULT now(),
          CONSTRAINT access_scopes_knowledge_base_fk
            FOREIGN KEY (knowledge_base_id)
            REFERENCES rag.knowledge_bases (id)
            ON DELETE CASCADE,
          CONSTRAINT access_scopes_name_nonempty
            CHECK (btrim(name) <> ''),
          CONSTRAINT access_scopes_knowledge_base_name_key
            UNIQUE (knowledge_base_id, name),
          CONSTRAINT access_scopes_knowledge_base_id_key
            UNIQUE (knowledge_base_id, id)
        );
        """
    )

    op.execute(
        """
        CREATE TABLE rag.documents (
          id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          knowledge_base_id uuid NOT NULL,
          access_scope_id uuid NOT NULL,
          raw_sha256 char(64) NOT NULL,
          filename text NOT NULL,
          storage_key text NOT NULL,
          mime_type text NOT NULL,
          status text NOT NULL DEFAULT 'UPLOADED',
          active_version_id uuid,
          deleted_at timestamptz,
          created_at timestamptz NOT NULL DEFAULT now(),
          updated_at timestamptz NOT NULL DEFAULT now(),
          CONSTRAINT documents_knowledge_base_fk
            FOREIGN KEY (knowledge_base_id)
            REFERENCES rag.knowledge_bases (id)
            ON DELETE CASCADE,
          CONSTRAINT documents_access_scope_fk
            FOREIGN KEY (knowledge_base_id, access_scope_id)
            REFERENCES rag.access_scopes (knowledge_base_id, id)
            ON DELETE RESTRICT,
          CONSTRAINT documents_hash_format
            CHECK (raw_sha256 ~ '^[0-9a-f]{64}$'),
          CONSTRAINT documents_filename_nonempty
            CHECK (btrim(filename) <> ''),
          CONSTRAINT documents_storage_key_nonempty
            CHECK (btrim(storage_key) <> ''),
          CONSTRAINT documents_mime_type_nonempty
            CHECK (btrim(mime_type) <> ''),
          CONSTRAINT documents_status_valid
            CHECK (status IN (
              'UPLOADED', 'QUARANTINED_LOCAL_REVIEW', 'LOCAL_PARSING',
              'LOCAL_OCR_REQUIRED', 'LOCAL_OCR_RUNNING', 'LOCAL_NORMALIZING',
              'LOCAL_CHUNKING_AND_REDACTION',
              'LOCAL_PARSED_PENDING_CLOUD_REVIEW', 'CLOUD_APPROVED',
              'REJECTED', 'EMBEDDING', 'INDEXING', 'VALIDATING',
              'READY', 'FAILED', 'DELETED'
            )),
          CONSTRAINT documents_deleted_state_consistent
            CHECK (
              (status = 'DELETED' AND deleted_at IS NOT NULL AND active_version_id IS NULL)
              OR (status <> 'DELETED' AND deleted_at IS NULL)
            ),
          CONSTRAINT documents_ready_requires_active_version
            CHECK (status <> 'READY' OR active_version_id IS NOT NULL),
          CONSTRAINT documents_knowledge_base_hash_key
            UNIQUE (knowledge_base_id, raw_sha256),
          CONSTRAINT documents_knowledge_base_id_key
            UNIQUE (knowledge_base_id, id)
        );
        """
    )

    op.execute(
        """
        CREATE TABLE rag.document_versions (
          id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          document_id uuid NOT NULL,
          parser_profile_hash char(64) NOT NULL,
          ocr_profile_hash char(64),
          normalization_profile_hash char(64) NOT NULL,
          chunk_profile_hash char(64) NOT NULL,
          cloud_policy_id text NOT NULL,
          outbound_manifest_sha256 char(64),
          review_state text NOT NULL DEFAULT 'QUARANTINED_LOCAL_REVIEW',
          cloud_approved_at timestamptz,
          cloud_approved_by text,
          embedding_profile_hash char(64),
          status text NOT NULL DEFAULT 'PENDING',
          expected_chunk_count integer,
          error_code text,
          created_at timestamptz NOT NULL DEFAULT now(),
          completed_at timestamptz,
          CONSTRAINT document_versions_document_fk
            FOREIGN KEY (document_id)
            REFERENCES rag.documents (id)
            ON DELETE CASCADE,
          CONSTRAINT document_versions_parser_hash_format
            CHECK (parser_profile_hash ~ '^[0-9a-f]{64}$'),
          CONSTRAINT document_versions_ocr_hash_format
            CHECK (ocr_profile_hash IS NULL OR ocr_profile_hash ~ '^[0-9a-f]{64}$'),
          CONSTRAINT document_versions_normalization_hash_format
            CHECK (normalization_profile_hash ~ '^[0-9a-f]{64}$'),
          CONSTRAINT document_versions_chunk_hash_format
            CHECK (chunk_profile_hash ~ '^[0-9a-f]{64}$'),
          CONSTRAINT document_versions_embedding_hash_format
            CHECK (
              embedding_profile_hash IS NULL
              OR embedding_profile_hash ~ '^[0-9a-f]{64}$'
            ),
          CONSTRAINT document_versions_cloud_policy_nonempty
            CHECK (btrim(cloud_policy_id) <> ''),
          CONSTRAINT document_versions_manifest_hash_format
            CHECK (
              outbound_manifest_sha256 IS NULL
              OR outbound_manifest_sha256 ~ '^[0-9a-f]{64}$'
            ),
          CONSTRAINT document_versions_review_state_valid
            CHECK (review_state IN (
              'QUARANTINED_LOCAL_REVIEW',
              'LOCAL_PARSED_PENDING_CLOUD_REVIEW',
              'CLOUD_APPROVED',
              'REJECTED'
            )),
          CONSTRAINT document_versions_cloud_approval_bound
            CHECK (
              (
                review_state = 'CLOUD_APPROVED'
                AND outbound_manifest_sha256 IS NOT NULL
                AND embedding_profile_hash IS NOT NULL
                AND cloud_approved_at IS NOT NULL
                AND cloud_approved_by IS NOT NULL
                AND btrim(cloud_approved_by) <> ''
              )
              OR (
                review_state <> 'CLOUD_APPROVED'
                AND cloud_approved_at IS NULL
                AND cloud_approved_by IS NULL
              )
            ),
          CONSTRAINT document_versions_status_valid
            CHECK (status IN ('PENDING', 'PROCESSING', 'READY', 'FAILED')),
          CONSTRAINT document_versions_chunk_count_valid
            CHECK (expected_chunk_count IS NULL OR expected_chunk_count >= 0),
          CONSTRAINT document_versions_completion_consistent
            CHECK (
              (status IN ('READY', 'FAILED') AND completed_at IS NOT NULL)
              OR (status IN ('PENDING', 'PROCESSING') AND completed_at IS NULL)
            ),
          CONSTRAINT document_versions_document_id_key
            UNIQUE (document_id, id),
          CONSTRAINT document_versions_approval_binding_key
            UNIQUE (
              id,
              review_state,
              outbound_manifest_sha256,
              embedding_profile_hash
            )
        );
        """
    )

    op.execute(
        """
        ALTER TABLE rag.documents
          ADD CONSTRAINT documents_active_version_fk
          FOREIGN KEY (id, active_version_id)
          REFERENCES rag.document_versions (document_id, id)
          DEFERRABLE INITIALLY DEFERRED;
        """
    )

    op.execute(
        """
        CREATE TABLE rag.outbound_manifest_items (
          document_version_id uuid NOT NULL,
          ordinal integer NOT NULL,
          outbound_manifest_sha256 char(64) NOT NULL,
          cloud_text_sha256 char(64) NOT NULL,
          embedding_text_sha256 char(64) NOT NULL,
          created_at timestamptz NOT NULL DEFAULT now(),
          CONSTRAINT outbound_manifest_items_version_fk
            FOREIGN KEY (document_version_id)
            REFERENCES rag.document_versions (id)
            ON DELETE CASCADE,
          CONSTRAINT outbound_manifest_items_ordinal_valid
            CHECK (ordinal >= 0),
          CONSTRAINT outbound_manifest_items_hashes_valid
            CHECK (
              outbound_manifest_sha256 ~ '^[0-9a-f]{64}$'
              AND cloud_text_sha256 ~ '^[0-9a-f]{64}$'
              AND embedding_text_sha256 ~ '^[0-9a-f]{64}$'
            ),
          CONSTRAINT outbound_manifest_items_primary_key
            PRIMARY KEY (document_version_id, ordinal),
          CONSTRAINT outbound_manifest_items_chunk_binding_key
            UNIQUE (
              document_version_id,
              ordinal,
              outbound_manifest_sha256,
              cloud_text_sha256,
              embedding_text_sha256
            )
        );
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.guard_approved_manifest_item_mutation()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        BEGIN
          IF (
            TG_OP = 'INSERT'
            AND EXISTS (
              SELECT 1 FROM rag.document_versions AS version
              WHERE version.id = NEW.document_version_id
                AND version.review_state = 'CLOUD_APPROVED'
            )
          ) OR (
            TG_OP = 'DELETE'
            AND EXISTS (
              SELECT 1 FROM rag.document_versions AS version
              WHERE version.id = OLD.document_version_id
                AND version.review_state = 'CLOUD_APPROVED'
            )
          ) OR (
            TG_OP = 'UPDATE'
            AND EXISTS (
              SELECT 1 FROM rag.document_versions AS version
              WHERE version.id IN (OLD.document_version_id, NEW.document_version_id)
                AND version.review_state = 'CLOUD_APPROVED'
            )
          ) THEN
            RAISE EXCEPTION
              'approved outbound manifest items are immutable; revoke approval first'
              USING ERRCODE = '23514';
          END IF;
          IF TG_OP = 'DELETE' THEN
            RETURN OLD;
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )
    op.execute(
        """
        CREATE TRIGGER outbound_manifest_items_immutable_after_approval
        BEFORE INSERT OR UPDATE OR DELETE
        ON rag.outbound_manifest_items
        FOR EACH ROW
        EXECUTE FUNCTION rag.guard_approved_manifest_item_mutation();
        """
    )

    op.execute(
        """
        CREATE TABLE rag.chunks (
          id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          knowledge_base_id uuid NOT NULL,
          document_id uuid NOT NULL,
          document_version_id uuid NOT NULL,
          access_scope_id uuid NOT NULL,
          ordinal integer NOT NULL,
          display_text text NOT NULL,
          cloud_text text NOT NULL,
          cloud_text_sha256 char(64) NOT NULL,
          embedding_prefix text NOT NULL,
          embedding_text text NOT NULL,
          embedding_text_sha256 char(64) NOT NULL,
          embedded_text_sha256 char(64),
          embedding_profile_hash char(64),
          outbound_manifest_sha256 char(64),
          token_count integer NOT NULL,
          page_start integer,
          page_end integer,
          section_path jsonb NOT NULL DEFAULT '[]'::jsonb,
          metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
          embedding_model text NOT NULL,
          embedding_dimension smallint NOT NULL DEFAULT 1024,
          embedding vector(1024),
          approval_status text NOT NULL DEFAULT 'QUARANTINED_LOCAL_REVIEW',
          index_status text NOT NULL DEFAULT 'PENDING',
          searchable boolean NOT NULL DEFAULT false,
          deleted_at timestamptz,
          created_at timestamptz NOT NULL DEFAULT now(),
          updated_at timestamptz NOT NULL DEFAULT now(),
          CONSTRAINT chunks_document_fk
            FOREIGN KEY (knowledge_base_id, document_id)
            REFERENCES rag.documents (knowledge_base_id, id)
            ON DELETE CASCADE,
          CONSTRAINT chunks_document_version_fk
            FOREIGN KEY (document_id, document_version_id)
            REFERENCES rag.document_versions (document_id, id)
            ON DELETE CASCADE,
          CONSTRAINT chunks_access_scope_fk
            FOREIGN KEY (knowledge_base_id, access_scope_id)
            REFERENCES rag.access_scopes (knowledge_base_id, id)
            ON DELETE RESTRICT,
          CONSTRAINT chunks_approval_binding_fk
            FOREIGN KEY (
              document_version_id,
              approval_status,
              outbound_manifest_sha256,
              embedding_profile_hash
            )
            REFERENCES rag.document_versions (
              id,
              review_state,
              outbound_manifest_sha256,
              embedding_profile_hash
            )
            DEFERRABLE INITIALLY DEFERRED,
          CONSTRAINT chunks_manifest_item_binding_fk
            FOREIGN KEY (
              document_version_id,
              ordinal,
              outbound_manifest_sha256,
              cloud_text_sha256,
              embedding_text_sha256
            )
            REFERENCES rag.outbound_manifest_items (
              document_version_id,
              ordinal,
              outbound_manifest_sha256,
              cloud_text_sha256,
              embedding_text_sha256
            )
            DEFERRABLE INITIALLY DEFERRED,
          CONSTRAINT chunks_ordinal_valid
            CHECK (ordinal >= 0),
          CONSTRAINT chunks_text_nonempty
            CHECK (
              btrim(display_text) <> ''
              AND btrim(cloud_text) <> ''
              AND btrim(embedding_prefix) <> ''
              AND btrim(embedding_text) <> ''
            ),
          CONSTRAINT chunks_cloud_text_hash_matches
            CHECK (
              cloud_text_sha256 = encode(
                sha256(convert_to(cloud_text, 'UTF8')),
                'hex'
              )
            ),
          CONSTRAINT chunks_embedding_input_bound_to_cloud_text
            CHECK (embedding_text = embedding_prefix || cloud_text),
          CONSTRAINT chunks_embedding_text_hash_matches
            CHECK (
              embedding_text_sha256 = encode(
                sha256(convert_to(embedding_text, 'UTF8')),
                'hex'
              )
            ),
          CONSTRAINT chunks_embedded_text_hash_format
            CHECK (
              embedded_text_sha256 IS NULL
              OR embedded_text_sha256 ~ '^[0-9a-f]{64}$'
            ),
          CONSTRAINT chunks_embedding_profile_hash_format
            CHECK (
              embedding_profile_hash IS NULL
              OR embedding_profile_hash ~ '^[0-9a-f]{64}$'
            ),
          CONSTRAINT chunks_manifest_hash_format
            CHECK (
              outbound_manifest_sha256 IS NULL
              OR outbound_manifest_sha256 ~ '^[0-9a-f]{64}$'
            ),
          CONSTRAINT chunks_token_count_valid
            CHECK (token_count > 0),
          CONSTRAINT chunks_pages_valid
            CHECK (
              (page_start IS NULL AND page_end IS NULL)
              OR (
                page_start IS NOT NULL
                AND page_end IS NOT NULL
                AND page_start > 0
                AND page_end >= page_start
              )
            ),
          CONSTRAINT chunks_section_path_array
            CHECK (jsonb_typeof(section_path) = 'array'),
          CONSTRAINT chunks_metadata_object
            CHECK (jsonb_typeof(metadata) = 'object'),
          CONSTRAINT chunks_embedding_model_nonempty
            CHECK (btrim(embedding_model) <> ''),
          CONSTRAINT chunks_embedding_dimension_1024
            CHECK (embedding_dimension = 1024),
          CONSTRAINT chunks_vector_dimension_1024
            CHECK (embedding IS NULL OR vector_dims(embedding) = 1024),
          CONSTRAINT chunks_approval_status_valid
            CHECK (approval_status IN (
              'QUARANTINED_LOCAL_REVIEW',
              'LOCAL_PARSED_PENDING_CLOUD_REVIEW',
              'CLOUD_APPROVED',
              'REJECTED'
            )),
          CONSTRAINT chunks_cloud_approval_has_complete_binding
            CHECK (
              approval_status <> 'CLOUD_APPROVED'
              OR (
                outbound_manifest_sha256 IS NOT NULL
                AND embedding_profile_hash IS NOT NULL
              )
            ),
          CONSTRAINT chunks_index_status_valid
            CHECK (index_status IN ('PENDING', 'EMBEDDING', 'READY', 'FAILED')),
          CONSTRAINT chunks_ready_index_has_current_embedding
            CHECK (
              index_status <> 'READY'
              OR (
                embedding IS NOT NULL
                AND embedded_text_sha256 IS NOT NULL
                AND embedded_text_sha256 = embedding_text_sha256
                AND embedding_profile_hash IS NOT NULL
              )
            ),
          CONSTRAINT chunks_searchable_requires_ready_approved_embedding
            CHECK (
              NOT searchable
              OR (
                embedding IS NOT NULL
                AND approval_status = 'CLOUD_APPROVED'
                AND index_status = 'READY'
                AND embedded_text_sha256 IS NOT NULL
                AND embedded_text_sha256 = embedding_text_sha256
                AND embedding_profile_hash IS NOT NULL
                AND outbound_manifest_sha256 IS NOT NULL
                AND deleted_at IS NULL
              )
            ),
          CONSTRAINT chunks_document_version_ordinal_key
            UNIQUE (document_version_id, ordinal)
        );
        """
    )

    op.execute(
        """
        COMMENT ON COLUMN rag.chunks.display_text IS
          'Local-only normalized source text; never implied safe for cloud egress.';
        """
    )
    op.execute(
        """
        COMMENT ON COLUMN rag.chunks.cloud_text IS
          'Redacted outbound text covered by the document-version approval manifest.';
        """
    )
    op.execute(
        """
        COMMENT ON COLUMN rag.chunks.embedding_prefix IS
          'Provider document prefix selected by the approved embedding profile.';
        """
    )
    op.execute(
        """
        COMMENT ON COLUMN rag.chunks.embedding_text IS
          'Exact provider input, database-enforced as embedding_prefix || cloud_text.';
        """
    )

    op.execute(
        """
        CREATE INDEX chunks_embedding_hnsw
          ON rag.chunks
          USING hnsw (embedding vector_cosine_ops)
          WHERE searchable;
        """
    )
    op.execute(
        """
        CREATE INDEX chunks_access_filter
          ON rag.chunks (knowledge_base_id, access_scope_id)
          WHERE searchable;
        """
    )
    op.execute(
        """
        CREATE INDEX chunks_document_lookup
          ON rag.chunks (document_id, document_version_id);
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.guard_approved_chunk_mutation()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        BEGIN
          IF OLD.approval_status = 'CLOUD_APPROVED'
             AND NEW.approval_status = 'CLOUD_APPROVED'
             AND (
               NEW.knowledge_base_id IS DISTINCT FROM OLD.knowledge_base_id
               OR NEW.document_id IS DISTINCT FROM OLD.document_id
               OR NEW.document_version_id IS DISTINCT FROM OLD.document_version_id
               OR NEW.access_scope_id IS DISTINCT FROM OLD.access_scope_id
               OR NEW.ordinal IS DISTINCT FROM OLD.ordinal
               OR NEW.cloud_text IS DISTINCT FROM OLD.cloud_text
               OR NEW.cloud_text_sha256 IS DISTINCT FROM OLD.cloud_text_sha256
               OR NEW.embedding_prefix IS DISTINCT FROM OLD.embedding_prefix
               OR NEW.embedding_text IS DISTINCT FROM OLD.embedding_text
               OR NEW.embedding_text_sha256 IS DISTINCT FROM OLD.embedding_text_sha256
               OR NEW.outbound_manifest_sha256
                    IS DISTINCT FROM OLD.outbound_manifest_sha256
               OR NEW.section_path IS DISTINCT FROM OLD.section_path
               OR NEW.metadata IS DISTINCT FROM OLD.metadata
               OR NEW.page_start IS DISTINCT FROM OLD.page_start
               OR NEW.page_end IS DISTINCT FROM OLD.page_end
             ) THEN
            RAISE EXCEPTION
              'approved outbound chunk fields are immutable; revoke version approval first'
              USING ERRCODE = '23514';
          END IF;

          IF OLD.index_status = 'READY'
             AND NEW.index_status = 'READY'
             AND (
               NEW.embedded_text_sha256 IS DISTINCT FROM OLD.embedded_text_sha256
               OR NEW.embedding_profile_hash IS DISTINCT FROM OLD.embedding_profile_hash
             ) THEN
            RAISE EXCEPTION
              'ready embedding is immutable; leave READY and searchable state before replacing it'
              USING ERRCODE = '23514';
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.guard_ready_embedding_update()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        BEGIN
          IF OLD.index_status = 'READY' AND NEW.index_status = 'READY' THEN
            RAISE EXCEPTION
              'ready vector is immutable; set index_status to EMBEDDING before replacement'
              USING ERRCODE = '23514';
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )
    op.execute(
        """
        CREATE TRIGGER chunks_guard_ready_vector_update
        BEFORE UPDATE OF embedding
        ON rag.chunks
        FOR EACH ROW
        EXECUTE FUNCTION rag.guard_ready_embedding_update();
        """
    )
    op.execute(
        """
        CREATE TRIGGER chunks_guard_approved_mutation
        BEFORE UPDATE OF
          knowledge_base_id,
          document_id,
          document_version_id,
          access_scope_id,
          ordinal,
          cloud_text,
          cloud_text_sha256,
          embedding_prefix,
          embedding_text,
          embedding_text_sha256,
          outbound_manifest_sha256,
          section_path,
          metadata,
          page_start,
          page_end,
          embedding,
          embedded_text_sha256,
          embedding_profile_hash,
          approval_status,
          index_status
        ON rag.chunks
        FOR EACH ROW
        EXECUTE FUNCTION rag.guard_approved_chunk_mutation();
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.enforce_chunk_searchability()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        BEGIN
          IF NEW.searchable AND NOT EXISTS (
            SELECT 1
            FROM rag.documents AS document
            JOIN rag.document_versions AS version
              ON version.id = NEW.document_version_id
             AND version.document_id = document.id
            WHERE document.id = NEW.document_id
              AND document.knowledge_base_id = NEW.knowledge_base_id
              AND document.active_version_id = NEW.document_version_id
              AND document.status = 'READY'
              AND document.deleted_at IS NULL
              AND version.status = 'READY'
              AND version.review_state = 'CLOUD_APPROVED'
              AND version.outbound_manifest_sha256 = NEW.outbound_manifest_sha256
              AND version.embedding_profile_hash = NEW.embedding_profile_hash
          ) THEN
            RAISE EXCEPTION
              'searchable chunk must belong to the active READY approved version'
              USING ERRCODE = '23514';
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )
    op.execute(
        """
        CREATE TRIGGER chunks_enforce_active_search_projection
        BEFORE INSERT OR UPDATE OF
          searchable,
          knowledge_base_id,
          document_id,
          document_version_id,
          approval_status,
          outbound_manifest_sha256,
          embedding_profile_hash,
          index_status
        ON rag.chunks
        FOR EACH ROW
        EXECUTE FUNCTION rag.enforce_chunk_searchability();
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.enforce_document_activation()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        BEGIN
          IF NEW.active_version_id IS NOT NULL AND NOT EXISTS (
            SELECT 1
            FROM rag.document_versions AS version
            WHERE version.id = NEW.active_version_id
              AND version.document_id = NEW.id
              AND version.status = 'READY'
              AND version.review_state = 'CLOUD_APPROVED'
          ) THEN
            RAISE EXCEPTION
              'active document version must be READY and CLOUD_APPROVED'
              USING ERRCODE = '23514';
          END IF;

          IF NEW.active_version_id IS DISTINCT FROM OLD.active_version_id
             OR NEW.status IS DISTINCT FROM OLD.status
             OR NEW.deleted_at IS DISTINCT FROM OLD.deleted_at THEN
            UPDATE rag.chunks
               SET searchable = false,
                   updated_at = now()
             WHERE document_id = OLD.id
               AND (
                 NEW.active_version_id IS NULL
                 OR document_version_id <> NEW.active_version_id
                 OR NEW.status <> 'READY'
                 OR NEW.deleted_at IS NOT NULL
               );
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )
    op.execute(
        """
        CREATE TRIGGER documents_enforce_activation
        BEFORE UPDATE OF active_version_id, status, deleted_at
        ON rag.documents
        FOR EACH ROW
        EXECUTE FUNCTION rag.enforce_document_activation();
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.revoke_cloud_approval_on_binding_change()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        DECLARE
          revoked_state text;
        BEGIN
          IF OLD.review_state = 'CLOUD_APPROVED'
             AND (
               NEW.review_state IS DISTINCT FROM OLD.review_state
               OR NEW.parser_profile_hash IS DISTINCT FROM OLD.parser_profile_hash
               OR NEW.ocr_profile_hash IS DISTINCT FROM OLD.ocr_profile_hash
               OR NEW.normalization_profile_hash
                    IS DISTINCT FROM OLD.normalization_profile_hash
               OR NEW.chunk_profile_hash IS DISTINCT FROM OLD.chunk_profile_hash
               OR NEW.cloud_policy_id IS DISTINCT FROM OLD.cloud_policy_id
               OR NEW.embedding_profile_hash
                    IS DISTINCT FROM OLD.embedding_profile_hash
               OR NEW.outbound_manifest_sha256
                    IS DISTINCT FROM OLD.outbound_manifest_sha256
             ) THEN
            revoked_state := CASE
              WHEN NEW.review_state = 'CLOUD_APPROVED'
                THEN 'LOCAL_PARSED_PENDING_CLOUD_REVIEW'
              ELSE NEW.review_state
            END;

            UPDATE rag.chunks
               SET searchable = false,
                   approval_status = revoked_state,
                   outbound_manifest_sha256 = NULL,
                   index_status = 'PENDING',
                   embedding = NULL,
                   embedded_text_sha256 = NULL,
                   embedding_profile_hash = NULL,
                   updated_at = now()
             WHERE document_version_id = OLD.id
               AND approval_status = 'CLOUD_APPROVED';

            UPDATE rag.documents
               SET active_version_id = NULL,
                   status = CASE
                     WHEN revoked_state = 'REJECTED' THEN 'REJECTED'
                     ELSE 'LOCAL_PARSED_PENDING_CLOUD_REVIEW'
                   END,
                   updated_at = now()
             WHERE active_version_id = OLD.id;

            NEW.review_state := revoked_state;
            NEW.cloud_approved_at := NULL;
            NEW.cloud_approved_by := NULL;
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )

    op.execute(
        """
        CREATE FUNCTION rag.deactivate_chunks_when_version_not_ready()
        RETURNS trigger
        LANGUAGE plpgsql
        SECURITY INVOKER
        SET search_path = pg_catalog, rag
        AS $function$
        BEGIN
          IF OLD.status = 'READY' AND NEW.status <> 'READY' THEN
            UPDATE rag.chunks
               SET searchable = false,
                   index_status = CASE
                     WHEN NEW.status = 'FAILED' THEN 'FAILED'
                     ELSE index_status
                   END,
                   updated_at = now()
             WHERE document_version_id = OLD.id;

            UPDATE rag.documents
               SET active_version_id = NULL,
                   status = CASE
                     WHEN NEW.status = 'FAILED' THEN 'FAILED'
                     ELSE 'VALIDATING'
                   END,
                   updated_at = now()
             WHERE active_version_id = OLD.id;
          END IF;
          RETURN NEW;
        END
        $function$;
        """
    )
    op.execute(
        """
        CREATE TRIGGER document_versions_deactivate_when_not_ready
        BEFORE UPDATE OF status
        ON rag.document_versions
        FOR EACH ROW
        EXECUTE FUNCTION rag.deactivate_chunks_when_version_not_ready();
        """
    )
    op.execute(
        """
        CREATE TRIGGER document_versions_revoke_cloud_approval
        BEFORE UPDATE OF
          review_state,
          parser_profile_hash,
          ocr_profile_hash,
          normalization_profile_hash,
          chunk_profile_hash,
          cloud_policy_id,
          embedding_profile_hash,
          outbound_manifest_sha256
        ON rag.document_versions
        FOR EACH ROW
        EXECUTE FUNCTION rag.revoke_cloud_approval_on_binding_change();
        """
    )

    op.execute(
        """
        CREATE TABLE rag.background_jobs (
          id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          job_type text NOT NULL,
          required_capability text NOT NULL,
          resource_type text NOT NULL,
          resource_id uuid NOT NULL,
          idempotency_key text NOT NULL,
          payload jsonb NOT NULL DEFAULT '{}'::jsonb,
          stage text NOT NULL DEFAULT 'PENDING',
          status text NOT NULL DEFAULT 'QUEUED',
          progress smallint NOT NULL DEFAULT 0,
          priority integer NOT NULL DEFAULT 0,
          attempt integer NOT NULL DEFAULT 0,
          max_attempts integer NOT NULL DEFAULT 3,
          run_after timestamptz NOT NULL DEFAULT now(),
          lease_owner text,
          lease_token uuid,
          lease_until timestamptz,
          last_error_code text,
          last_error_message text,
          created_at timestamptz NOT NULL DEFAULT now(),
          updated_at timestamptz NOT NULL DEFAULT now(),
          finished_at timestamptz,
          CONSTRAINT background_jobs_job_type_nonempty
            CHECK (btrim(job_type) <> ''),
          CONSTRAINT background_jobs_capability_nonempty
            CHECK (btrim(required_capability) <> ''),
          CONSTRAINT background_jobs_resource_type_nonempty
            CHECK (btrim(resource_type) <> ''),
          CONSTRAINT background_jobs_idempotency_key_nonempty
            CHECK (btrim(idempotency_key) <> ''),
          CONSTRAINT background_jobs_payload_object
            CHECK (jsonb_typeof(payload) = 'object'),
          CONSTRAINT background_jobs_stage_nonempty
            CHECK (btrim(stage) <> ''),
          CONSTRAINT background_jobs_status_valid
            CHECK (status IN ('QUEUED', 'RUNNING', 'SUCCEEDED', 'FAILED', 'CANCELLED')),
          CONSTRAINT background_jobs_progress_valid
            CHECK (progress BETWEEN 0 AND 100),
          CONSTRAINT background_jobs_attempts_valid
            CHECK (
              attempt >= 0
              AND max_attempts > 0
              AND attempt <= max_attempts
            ),
          CONSTRAINT background_jobs_lease_consistent
            CHECK (
              (
                status = 'RUNNING'
                AND lease_owner IS NOT NULL
                AND btrim(lease_owner) <> ''
                AND lease_token IS NOT NULL
                AND lease_until IS NOT NULL
              )
              OR (
                status <> 'RUNNING'
                AND lease_owner IS NULL
                AND lease_token IS NULL
                AND lease_until IS NULL
              )
            ),
          CONSTRAINT background_jobs_terminal_time_consistent
            CHECK (
              (
                status IN ('SUCCEEDED', 'FAILED', 'CANCELLED')
                AND finished_at IS NOT NULL
              )
              OR (
                status IN ('QUEUED', 'RUNNING')
                AND finished_at IS NULL
              )
            ),
          CONSTRAINT background_jobs_success_progress_complete
            CHECK (status <> 'SUCCEEDED' OR progress = 100),
          CONSTRAINT background_jobs_error_message_bounded
            CHECK (last_error_message IS NULL OR length(last_error_message) <= 2000),
          CONSTRAINT background_jobs_type_idempotency_key
            UNIQUE (job_type, idempotency_key)
        );
        """
    )
    op.execute(
        """
        CREATE INDEX background_jobs_claim_queued
          ON rag.background_jobs (
            required_capability,
            priority DESC,
            run_after,
            created_at
          )
          WHERE status = 'QUEUED';
        """
    )
    op.execute(
        """
        CREATE INDEX background_jobs_reap_expired
          ON rag.background_jobs (lease_until, required_capability)
          WHERE status = 'RUNNING';
        """
    )
    op.execute(
        """
        CREATE INDEX background_jobs_resource_lookup
          ON rag.background_jobs (resource_type, resource_id, created_at DESC);
        """
    )


def downgrade() -> None:
    op.execute(
        """
        ALTER TABLE rag.documents
          DROP CONSTRAINT IF EXISTS documents_active_version_fk;
        """
    )
    op.execute("DROP TABLE IF EXISTS rag.background_jobs;")
    op.execute("DROP TABLE IF EXISTS rag.chunks;")
    op.execute("DROP FUNCTION IF EXISTS rag.guard_approved_chunk_mutation();")
    op.execute("DROP FUNCTION IF EXISTS rag.guard_ready_embedding_update();")
    op.execute("DROP FUNCTION IF EXISTS rag.enforce_chunk_searchability();")
    op.execute("DROP TABLE IF EXISTS rag.outbound_manifest_items;")
    op.execute("DROP FUNCTION IF EXISTS rag.guard_approved_manifest_item_mutation();")
    op.execute("DROP TABLE IF EXISTS rag.document_versions;")
    op.execute("DROP FUNCTION IF EXISTS rag.revoke_cloud_approval_on_binding_change();")
    op.execute("DROP FUNCTION IF EXISTS rag.deactivate_chunks_when_version_not_ready();")
    op.execute("DROP TABLE IF EXISTS rag.documents;")
    op.execute("DROP FUNCTION IF EXISTS rag.enforce_document_activation();")
    op.execute("DROP TABLE IF EXISTS rag.access_scopes;")
    op.execute(
        """
        DROP TABLE IF EXISTS rag.knowledge_bases;
        """
    )