//! SQL schema for registered semantic-rerank sources and detached requests.

pgrx::extension_sql!(
    r#"
CREATE TABLE pgcontext._semantic_rerank_sources (
    rerank_source_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL
        REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    source_name text NOT NULL CHECK (
        pg_catalog.octet_length(source_name) BETWEEN 1 AND 128
        AND pg_catalog.btrim(source_name) <> ''
        AND source_name !~ '[[:cntrl:]]'
    ),
    source_table_oid oid NOT NULL,
    source_schema_name text NOT NULL,
    source_table_name text NOT NULL,
    source_key_attnum int2 NOT NULL CHECK (source_key_attnum > 0),
    source_key_type_oid oid NOT NULL,
    source_key_collation_oid oid NOT NULL DEFAULT 0,
    source_key_type_schema text NOT NULL,
    source_key_type_name text NOT NULL,
    text_column_name text NOT NULL,
    text_attnum int2 NOT NULL CHECK (text_attnum > 0),
    text_type_oid oid NOT NULL,
    text_collation_oid oid NOT NULL DEFAULT 0,
    text_collation_schema text NOT NULL,
    text_collation_name text NOT NULL,
    source_version_column_name text NOT NULL,
    source_version_attnum int2 NOT NULL CHECK (source_version_attnum > 0),
    source_version_type_oid oid NOT NULL,
    content_hash_rule text NOT NULL DEFAULT 'sha256_utf8_v1'
        CHECK (content_hash_rule = 'sha256_utf8_v1'),
    registration_revision bigint NOT NULL DEFAULT 1
        CHECK (registration_revision > 0),
    status text NOT NULL DEFAULT 'ready'
        CHECK (status IN ('ready', 'stale', 'failed')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, source_name)
);

CREATE TABLE pgcontext._semantic_rerank_requests (
    request_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL
        REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    rerank_source_id bigint NOT NULL
        REFERENCES pgcontext._semantic_rerank_sources(rerank_source_id) ON DELETE CASCADE,
    session_user_oid oid NOT NULL,
    source_registration_revision bigint NOT NULL CHECK (source_registration_revision > 0),
    filter_binding_sha256 bytea NOT NULL
        CHECK (pg_catalog.octet_length(filter_binding_sha256) = 32),
    model_name text NOT NULL CHECK (
        pg_catalog.octet_length(model_name) BETWEEN 1 AND 128
        AND pg_catalog.btrim(model_name) <> ''
        AND model_name !~ '[[:cntrl:]]'
    ),
    model_revision bigint NOT NULL CHECK (model_revision > 0),
    query_text text NOT NULL CHECK (
        pg_catalog.octet_length(query_text) BETWEEN 1 AND 65536
        AND query_text !~ E'\\x00'
    ),
    filter_json jsonb,
    failure_policy text NOT NULL
        CHECK (failure_policy IN ('require_reranker', 'allow_fused_fallback')),
    allow_partial boolean NOT NULL DEFAULT false,
    expires_at_micros bigint NOT NULL CHECK (expires_at_micros > 0),
    status text NOT NULL DEFAULT 'prepared'
        CHECK (status IN ('prepared', 'finalized')),
    response_sha256 bytea,
    final_status text CHECK (
        final_status IN ('reranked', 'degraded_reranker', 'partial_reranked')
    ),
    degraded_reason text CHECK (
        degraded_reason IS NULL OR (
            pg_catalog.octet_length(degraded_reason) BETWEEN 1 AND 64
            AND degraded_reason ~ '^[a-z0-9_]+$'
        )
    ),
    final_result jsonb,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    finalized_at timestamptz,
    CHECK ((status = 'prepared') = (final_result IS NULL)),
    CHECK ((status = 'prepared') = (response_sha256 IS NULL)),
    CHECK ((status = 'prepared') = (final_status IS NULL))
);

CREATE TABLE pgcontext._semantic_rerank_candidates (
    request_id bigint NOT NULL
        REFERENCES pgcontext._semantic_rerank_requests(request_id) ON DELETE CASCADE,
    occurrence_id bigint NOT NULL CHECK (occurrence_id > 0),
    point_id bigint NOT NULL CHECK (point_id > 0),
    source_version bigint NOT NULL CHECK (source_version > 0),
    content_digest bytea NOT NULL CHECK (pg_catalog.octet_length(content_digest) = 32),
    fused_rank int4 NOT NULL CHECK (fused_rank > 0),
    fused_score double precision NOT NULL CHECK (
        fused_score > '-Infinity'::double precision
        AND fused_score < 'Infinity'::double precision
    ),
    contributions jsonb NOT NULL CHECK (pg_catalog.jsonb_typeof(contributions) = 'array'),
    metadata jsonb NOT NULL CHECK (pg_catalog.jsonb_typeof(metadata) = 'array'),
    PRIMARY KEY (request_id, occurrence_id),
    UNIQUE (request_id, point_id)
);

REVOKE ALL ON TABLE pgcontext._semantic_rerank_sources FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._semantic_rerank_requests FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._semantic_rerank_candidates FROM PUBLIC;

CREATE VIEW pgcontext._visible_semantic_rerank_sources
WITH (security_barrier = true) AS
SELECT sources.*
  FROM pgcontext._semantic_rerank_sources AS sources
  JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER');

CREATE VIEW pgcontext._visible_semantic_rerank_requests
WITH (security_barrier = true) AS
SELECT requests.*
  FROM pgcontext._semantic_rerank_requests AS requests
 WHERE requests.session_user_oid = SESSION_USER::pg_catalog.regrole::oid;

CREATE VIEW pgcontext._visible_semantic_rerank_candidates
WITH (security_barrier = true) AS
SELECT candidates.*
  FROM pgcontext._semantic_rerank_candidates AS candidates
  JOIN pgcontext._semantic_rerank_requests AS requests USING (request_id)
 WHERE requests.session_user_oid = SESSION_USER::pg_catalog.regrole::oid;

GRANT SELECT ON pgcontext._visible_semantic_rerank_sources TO PUBLIC;
GRANT SELECT ON pgcontext._visible_semantic_rerank_requests TO PUBLIC;
GRANT SELECT ON pgcontext._visible_semantic_rerank_candidates TO PUBLIC;

CREATE FUNCTION pgcontext._register_semantic_rerank_source(
    p_collection_id bigint,
    p_source_name text,
    p_text_column text,
    p_source_version_column text
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_schema text;
    source_table text;
    source_oid oid;
    key_row record;
    text_row record;
    version_row record;
    registered_id bigint;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    SELECT collections.source_schema_name,
           collections.source_table_name,
           source_class.oid
      INTO source_schema, source_table, source_oid
      FROM pgcontext._collections AS collections
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.nspname = collections.source_schema_name
      JOIN pg_catalog.pg_class AS source_class
        ON source_class.relnamespace = source_namespace.oid
       AND source_class.relname = collections.source_table_name
       AND source_class.relkind IN ('r', 'p')
     WHERE collections.collection_id = p_collection_id;
    IF source_oid IS NULL THEN
        RAISE EXCEPTION 'semantic rerank source table is unavailable'
            USING ERRCODE = '42P01';
    END IF;
    IF NOT pg_catalog.has_table_privilege(SESSION_USER, source_oid, 'SELECT') THEN
        RAISE EXCEPTION 'permission denied for semantic rerank source'
            USING ERRCODE = '42501';
    END IF;

    SELECT attribute.attnum,
           attribute.atttypid,
           attribute.attcollation,
           type_namespace.nspname AS type_schema,
           type.typname AS type_name
      INTO key_row
      FROM pg_catalog.pg_attribute AS attribute
      JOIN pg_catalog.pg_type AS type ON type.oid = attribute.atttypid
      JOIN pg_catalog.pg_namespace AS type_namespace ON type_namespace.oid = type.typnamespace
      JOIN pg_catalog.pg_index AS index
        ON index.indrelid = attribute.attrelid
       AND index.indisprimary
       AND index.indisvalid
       AND index.indisready
       AND index.indislive
       AND index.indnkeyatts = 1
       AND index.indnatts = 1
       AND index.indkey[0] = attribute.attnum
     WHERE attribute.attrelid = source_oid
       AND attribute.attname = 'id'
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped;
    IF key_row.attnum IS NULL THEN
        RAISE EXCEPTION 'semantic rerank source requires a single-column primary key named id'
            USING ERRCODE = '55000';
    END IF;

    SELECT attribute.attnum,
           attribute.atttypid,
           attribute.attcollation,
           collation_namespace.nspname AS collation_schema,
           text_collation.collname AS collation_name
      INTO text_row
      FROM pg_catalog.pg_attribute AS attribute
      JOIN pg_catalog.pg_collation AS text_collation
        ON text_collation.oid = attribute.attcollation
      JOIN pg_catalog.pg_namespace AS collation_namespace
        ON collation_namespace.oid = text_collation.collnamespace
     WHERE attribute.attrelid = source_oid
       AND attribute.attname = p_text_column
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped
       AND attribute.atttypid = 'pg_catalog.text'::pg_catalog.regtype;
    IF text_row.attnum IS NULL THEN
        RAISE EXCEPTION 'semantic rerank text column is missing or is not text'
            USING ERRCODE = '42703';
    END IF;

    SELECT attribute.attnum, attribute.atttypid
      INTO version_row
      FROM pg_catalog.pg_attribute AS attribute
     WHERE attribute.attrelid = source_oid
       AND attribute.attname = p_source_version_column
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped
       AND attribute.atttypid = 'pg_catalog.int8'::pg_catalog.regtype;
    IF version_row.attnum IS NULL THEN
        RAISE EXCEPTION 'semantic rerank source version column is missing or is not bigint'
            USING ERRCODE = '42703';
    END IF;

    INSERT INTO pgcontext._semantic_rerank_sources (
        collection_id, source_name, source_table_oid,
        source_schema_name, source_table_name,
        source_key_attnum, source_key_type_oid, source_key_collation_oid,
        source_key_type_schema, source_key_type_name,
        text_column_name, text_attnum, text_type_oid, text_collation_oid,
        text_collation_schema, text_collation_name,
        source_version_column_name, source_version_attnum,
        source_version_type_oid, content_hash_rule, registration_revision, status
    ) VALUES (
        p_collection_id, p_source_name, source_oid,
        source_schema, source_table,
        key_row.attnum, key_row.atttypid, key_row.attcollation,
        key_row.type_schema, key_row.type_name,
        p_text_column, text_row.attnum, text_row.atttypid, text_row.attcollation,
        text_row.collation_schema, text_row.collation_name,
        p_source_version_column, version_row.attnum,
        version_row.atttypid, 'sha256_utf8_v1', 1, 'ready'
    )
    ON CONFLICT (collection_id, source_name) DO UPDATE
       SET source_table_oid = EXCLUDED.source_table_oid,
           source_schema_name = EXCLUDED.source_schema_name,
           source_table_name = EXCLUDED.source_table_name,
           source_key_attnum = EXCLUDED.source_key_attnum,
           source_key_type_oid = EXCLUDED.source_key_type_oid,
           source_key_collation_oid = EXCLUDED.source_key_collation_oid,
           source_key_type_schema = EXCLUDED.source_key_type_schema,
           source_key_type_name = EXCLUDED.source_key_type_name,
           text_column_name = EXCLUDED.text_column_name,
           text_attnum = EXCLUDED.text_attnum,
           text_type_oid = EXCLUDED.text_type_oid,
           text_collation_oid = EXCLUDED.text_collation_oid,
           text_collation_schema = EXCLUDED.text_collation_schema,
           text_collation_name = EXCLUDED.text_collation_name,
           source_version_column_name = EXCLUDED.source_version_column_name,
           source_version_attnum = EXCLUDED.source_version_attnum,
           source_version_type_oid = EXCLUDED.source_version_type_oid,
           content_hash_rule = EXCLUDED.content_hash_rule,
           registration_revision = pgcontext._semantic_rerank_sources.registration_revision +
               CASE WHEN
                   pgcontext._semantic_rerank_sources.source_table_oid IS DISTINCT FROM
                       EXCLUDED.source_table_oid
                   OR pgcontext._semantic_rerank_sources.source_key_attnum IS DISTINCT FROM
                       EXCLUDED.source_key_attnum
                   OR pgcontext._semantic_rerank_sources.source_key_type_oid IS DISTINCT FROM
                       EXCLUDED.source_key_type_oid
                   OR pgcontext._semantic_rerank_sources.source_key_collation_oid IS DISTINCT FROM
                       EXCLUDED.source_key_collation_oid
                   OR pgcontext._semantic_rerank_sources.text_column_name IS DISTINCT FROM
                       EXCLUDED.text_column_name
                   OR pgcontext._semantic_rerank_sources.text_attnum IS DISTINCT FROM
                       EXCLUDED.text_attnum
                   OR pgcontext._semantic_rerank_sources.text_type_oid IS DISTINCT FROM
                       EXCLUDED.text_type_oid
                   OR pgcontext._semantic_rerank_sources.text_collation_oid IS DISTINCT FROM
                       EXCLUDED.text_collation_oid
                   OR pgcontext._semantic_rerank_sources.source_version_column_name IS DISTINCT FROM
                       EXCLUDED.source_version_column_name
                   OR pgcontext._semantic_rerank_sources.source_version_attnum IS DISTINCT FROM
                       EXCLUDED.source_version_attnum
                   OR pgcontext._semantic_rerank_sources.source_version_type_oid IS DISTINCT FROM
                       EXCLUDED.source_version_type_oid
               THEN 1 ELSE 0 END,
           status = 'ready',
           updated_at = pg_catalog.now()
    RETURNING rerank_source_id INTO registered_id;
    RETURN registered_id;
END;
$$;

CREATE FUNCTION pgcontext._refresh_semantic_rerank_source(
    p_collection_id bigint,
    p_source_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    registration pgcontext._semantic_rerank_sources%ROWTYPE;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    SELECT * INTO registration
      FROM pgcontext._semantic_rerank_sources
     WHERE collection_id = p_collection_id
       AND source_name = p_source_name;
    IF NOT FOUND THEN
        RETURN;
    END IF;

    UPDATE pgcontext._semantic_rerank_sources AS sources
       SET source_table_oid = source_class.oid,
           source_key_attnum = key_attribute.attnum,
           source_key_type_oid = key_attribute.atttypid,
           source_key_collation_oid = key_attribute.attcollation,
           source_key_type_schema = key_type_namespace.nspname,
           source_key_type_name = key_type.typname,
           text_attnum = text_attribute.attnum,
           text_type_oid = text_attribute.atttypid,
           text_collation_oid = text_attribute.attcollation,
           source_version_attnum = version_attribute.attnum,
           source_version_type_oid = version_attribute.atttypid,
           registration_revision = sources.registration_revision +
               CASE WHEN source_class.oid IS DISTINCT FROM registration.source_table_oid
                          OR key_attribute.attnum IS DISTINCT FROM registration.source_key_attnum
                          OR key_attribute.atttypid IS DISTINCT FROM registration.source_key_type_oid
                          OR key_attribute.attcollation IS DISTINCT FROM registration.source_key_collation_oid
                          OR text_attribute.attnum IS DISTINCT FROM registration.text_attnum
                          OR text_attribute.atttypid IS DISTINCT FROM registration.text_type_oid
                          OR text_attribute.attcollation IS DISTINCT FROM registration.text_collation_oid
                          OR version_attribute.attnum IS DISTINCT FROM registration.source_version_attnum
                          OR version_attribute.atttypid IS DISTINCT FROM registration.source_version_type_oid
                    THEN 1 ELSE 0 END,
           status = 'ready',
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_namespace AS source_namespace,
           pg_catalog.pg_class AS source_class,
           pg_catalog.pg_attribute AS key_attribute,
           pg_catalog.pg_type AS key_type,
           pg_catalog.pg_namespace AS key_type_namespace,
           pg_catalog.pg_attribute AS text_attribute,
           pg_catalog.pg_collation AS text_collation,
           pg_catalog.pg_namespace AS text_collation_namespace,
           pg_catalog.pg_attribute AS version_attribute
     WHERE sources.rerank_source_id = registration.rerank_source_id
       AND source_namespace.nspname = sources.source_schema_name
       AND source_class.relnamespace = source_namespace.oid
       AND source_class.relname = sources.source_table_name
       AND source_class.relkind IN ('r', 'p')
       AND key_attribute.attrelid = source_class.oid
       AND key_attribute.attname = 'id'
       AND key_attribute.attnum > 0
       AND NOT key_attribute.attisdropped
       AND EXISTS (
           SELECT 1
             FROM pg_catalog.pg_index AS primary_index
            WHERE primary_index.indrelid = source_class.oid
              AND primary_index.indisprimary
              AND primary_index.indisvalid
              AND primary_index.indisready
              AND primary_index.indislive
              AND primary_index.indnkeyatts = 1
              AND primary_index.indnatts = 1
              AND primary_index.indkey[0] = key_attribute.attnum
       )
       AND key_type.oid = key_attribute.atttypid
       AND key_type_namespace.oid = key_type.typnamespace
       AND key_type_namespace.nspname = sources.source_key_type_schema
       AND key_type.typname = sources.source_key_type_name
       AND text_attribute.attrelid = source_class.oid
       AND text_attribute.attname = sources.text_column_name
       AND text_attribute.atttypid = 'pg_catalog.text'::pg_catalog.regtype
       AND text_attribute.attnum > 0
       AND NOT text_attribute.attisdropped
       AND text_collation.oid = text_attribute.attcollation
       AND text_collation_namespace.oid = text_collation.collnamespace
       AND text_collation_namespace.nspname = sources.text_collation_schema
       AND text_collation.collname = sources.text_collation_name
       AND version_attribute.attrelid = source_class.oid
       AND version_attribute.attname = sources.source_version_column_name
       AND version_attribute.atttypid = 'pg_catalog.int8'::pg_catalog.regtype
       AND version_attribute.attnum > 0
       AND NOT version_attribute.attisdropped;

    IF NOT FOUND THEN
        UPDATE pgcontext._semantic_rerank_sources
           SET status = 'stale', updated_at = pg_catalog.now()
         WHERE rerank_source_id = registration.rerank_source_id;
    END IF;
END;
$$;

CREATE FUNCTION pgcontext._insert_semantic_rerank_request(
    p_collection_id bigint,
    p_rerank_source_id bigint,
    p_registration_revision bigint,
    p_filter_binding_sha256 bytea,
    p_model_name text,
    p_model_revision bigint,
    p_query_text text,
    p_filter_json jsonb,
    p_failure_policy text,
    p_allow_partial boolean,
    p_expires_at_micros bigint,
    p_candidates jsonb
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    new_request_id bigint;
    expected_count int4;
    inserted_count int4;
    now_micros bigint;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    now_micros := pg_catalog.floor(
        pg_catalog.extract('epoch', pg_catalog.clock_timestamp()) * 1000000
    )::bigint;
    IF p_model_name IS NULL
       OR p_query_text IS NULL
       OR p_model_revision IS NULL
       OR p_failure_policy IS NULL
       OR p_allow_partial IS NULL
       OR p_expires_at_micros IS NULL
       OR p_candidates IS NULL
       OR p_filter_binding_sha256 IS NULL
       OR pg_catalog.octet_length(p_filter_binding_sha256) <> 32
       OR pg_catalog.octet_length(p_model_name) NOT BETWEEN 1 AND 128
       OR pg_catalog.octet_length(p_query_text) NOT BETWEEN 1 AND 65536
       OR p_model_revision <= 0
       OR p_expires_at_micros <= now_micros
       OR p_expires_at_micros > now_micros + 60000000
       OR p_failure_policy NOT IN ('require_reranker', 'allow_fused_fallback')
       OR pg_catalog.pg_column_size(p_candidates) > 4194304
       OR pg_catalog.octet_length(p_candidates::text) > 4194304
       OR (
           p_filter_json IS NOT NULL
           AND (
               pg_catalog.pg_column_size(p_filter_json) > 262144
               OR pg_catalog.octet_length(p_filter_json::text) > 262144
           )
       )
    THEN
        RAISE EXCEPTION 'semantic rerank request exceeds its storage contract'
            USING ERRCODE = '54000';
    END IF;
    IF NOT EXISTS (
        SELECT 1 FROM pgcontext._semantic_rerank_sources
         WHERE rerank_source_id = p_rerank_source_id
           AND collection_id = p_collection_id
           AND registration_revision = p_registration_revision
           AND status = 'ready'
    ) THEN
        RAISE EXCEPTION 'semantic rerank source registration changed'
            USING ERRCODE = '55000';
    END IF;
    IF pg_catalog.jsonb_typeof(p_candidates) <> 'array' THEN
        RAISE EXCEPTION 'semantic rerank candidates must be an array'
            USING ERRCODE = '22023';
    END IF;
    expected_count := pg_catalog.jsonb_array_length(p_candidates);
    IF expected_count NOT BETWEEN 1 AND 512 THEN
        RAISE EXCEPTION 'semantic rerank candidate count is outside 1..=512'
            USING ERRCODE = '54000';
    END IF;

    INSERT INTO pgcontext._semantic_rerank_requests (
        collection_id, rerank_source_id, session_user_oid,
        source_registration_revision, filter_binding_sha256, model_name, model_revision,
        query_text, filter_json, failure_policy, allow_partial, expires_at_micros
    ) VALUES (
        p_collection_id, p_rerank_source_id, SESSION_USER::pg_catalog.regrole::oid,
        p_registration_revision, p_filter_binding_sha256, p_model_name, p_model_revision,
        p_query_text, p_filter_json, p_failure_policy, p_allow_partial,
        p_expires_at_micros
    ) RETURNING request_id INTO new_request_id;

    INSERT INTO pgcontext._semantic_rerank_candidates (
        request_id, occurrence_id, point_id, source_version, content_digest,
        fused_rank, fused_score, contributions, metadata
    )
    SELECT new_request_id,
           candidate.occurrence_id,
           candidate.point_id,
           candidate.source_version,
           pg_catalog.decode(candidate.content_digest, 'hex'),
           candidate.fused_rank,
           candidate.fused_score,
           candidate.contributions,
           candidate.metadata
      FROM pg_catalog.jsonb_to_recordset(p_candidates) AS candidate(
          occurrence_id bigint,
          point_id bigint,
          source_version bigint,
          content_digest text,
          fused_rank int4,
          fused_score double precision,
          contributions jsonb,
          metadata jsonb
      );
    GET DIAGNOSTICS inserted_count = ROW_COUNT;
    IF inserted_count <> expected_count THEN
        RAISE EXCEPTION 'semantic rerank candidate set is incomplete'
            USING ERRCODE = '22023';
    END IF;
    RETURN new_request_id;
END;
$$;

CREATE FUNCTION pgcontext._finalize_semantic_rerank_request(
    p_request_id bigint,
    p_response_sha256 bytea,
    p_final_status text,
    p_degraded_reason text,
    p_final_result jsonb
)
RETURNS boolean
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    IF p_response_sha256 IS NULL
       OR p_final_status IS NULL
       OR p_final_result IS NULL
       OR pg_catalog.octet_length(p_response_sha256) <> 32
       OR p_final_status NOT IN ('reranked', 'degraded_reranker', 'partial_reranked')
       OR pg_catalog.pg_column_size(p_final_result) > 4194304
       OR (
           p_degraded_reason IS NOT NULL
           AND pg_catalog.octet_length(p_degraded_reason) NOT BETWEEN 1 AND 64
       )
    THEN
        RAISE EXCEPTION 'semantic rerank finalization exceeds its storage contract'
            USING ERRCODE = '54000';
    END IF;
    UPDATE pgcontext._semantic_rerank_requests
       SET status = 'finalized',
           response_sha256 = p_response_sha256,
           final_status = p_final_status,
           degraded_reason = p_degraded_reason,
           final_result = p_final_result,
           finalized_at = pg_catalog.now()
     WHERE request_id = p_request_id
       AND session_user_oid = SESSION_USER::pg_catalog.regrole::oid
       AND status = 'prepared';
    RETURN FOUND;
END;
$$;

CREATE FUNCTION pgcontext._cleanup_semantic_rerank_requests(p_limit int4)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    deleted bigint;
BEGIN
    IF p_limit NOT BETWEEN 1 AND 10000 THEN
        RAISE EXCEPTION 'semantic rerank cleanup limit must be 1..=10000'
            USING ERRCODE = '22023';
    END IF;
    WITH victims AS (
        SELECT request_id
          FROM pgcontext._semantic_rerank_requests
         WHERE session_user_oid = SESSION_USER::pg_catalog.regrole::oid
           AND (
               expires_at_micros <= pg_catalog.floor(
                   pg_catalog.extract('epoch', pg_catalog.clock_timestamp()) * 1000000
               )::bigint
               OR status = 'finalized'
           )
         ORDER BY request_id
         LIMIT p_limit
         FOR UPDATE SKIP LOCKED
    ), deleted_rows AS (
        DELETE FROM pgcontext._semantic_rerank_requests AS requests
         USING victims
         WHERE requests.request_id = victims.request_id
         RETURNING 1
    )
    SELECT pg_catalog.count(*) INTO deleted FROM deleted_rows;
    RETURN deleted;
END;
$$;

SELECT pg_catalog.pg_extension_config_dump('pgcontext._semantic_rerank_sources', '');
SELECT pg_catalog.pg_extension_config_dump(
    'pgcontext._semantic_rerank_sources_rerank_source_id_seq', ''
);
"#,
    name = "create_semantic_rerank_catalog_tables",
    requires = ["create_catalog_tables", "create_lexical_catalog_tables"]
);
