//! SQL catalogs and fenced internal helpers for automatic document chunking.

#[allow(
    unused_imports,
    reason = "pgrx consumes this symbol in the SQL dependency graph"
)]
use crate::document_chunking::document_chunk_source_key_visible;

pgrx::extension_sql!(
    r#"
CREATE TABLE pgcontext._chunking_profiles (
    chunking_profile_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    owner_role oid NOT NULL,
    profile_name text NOT NULL CHECK (pg_catalog.octet_length(profile_name) BETWEEN 1 AND 128
        AND pg_catalog.btrim(profile_name) <> '' AND profile_name !~ '[[:cntrl:]]'),
    profile_revision bigint NOT NULL DEFAULT 1 CHECK (profile_revision > 0),
    parser_revision text NOT NULL CHECK (
        parser_revision IN ('plain_text_v1', 'markdown_v1', 'html_v1')),
    tokenizer_revision text NOT NULL DEFAULT 'unicode_words_v1'
        CHECK (tokenizer_revision = 'unicode_words_v1'),
    target_tokens int4 NOT NULL CHECK (target_tokens BETWEEN 1 AND 512),
    max_tokens int4 NOT NULL CHECK (max_tokens BETWEEN target_tokens AND 512),
    min_tokens int4 NOT NULL CHECK (min_tokens BETWEEN 1 AND target_tokens),
    overlap_tokens int4 NOT NULL CHECK (
        overlap_tokens BETWEEN 0 AND 64 AND overlap_tokens < target_tokens),
    max_document_bytes bigint NOT NULL CHECK (max_document_bytes BETWEEN 1 AND 8388608),
    include_structure_context boolean NOT NULL DEFAULT false,
    configuration_sha256 bytea NOT NULL CHECK (
        pg_catalog.octet_length(configuration_sha256) = 32),
    status text NOT NULL DEFAULT 'ready'
        CHECK (status IN ('ready', 'retired')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    retired_at timestamptz,
    UNIQUE (owner_role, profile_name),
    CHECK ((status = 'retired') = (retired_at IS NOT NULL))
);
CREATE TABLE pgcontext._chunking_profile_aliases (
    chunking_profile_alias_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    owner_role oid NOT NULL,
    alias_name text NOT NULL CHECK (pg_catalog.octet_length(alias_name) BETWEEN 1 AND 128
        AND pg_catalog.btrim(alias_name) <> '' AND alias_name !~ '[[:cntrl:]]'),
    chunking_profile_id bigint NOT NULL
        REFERENCES pgcontext._chunking_profiles(chunking_profile_id),
    shadow_chunking_profile_id bigint
        REFERENCES pgcontext._chunking_profiles(chunking_profile_id),
    alias_revision bigint NOT NULL DEFAULT 1 CHECK (alias_revision > 0),
    status text NOT NULL DEFAULT 'ready' CHECK (status IN ('ready','retired')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (owner_role, alias_name)
);
CREATE TABLE pgcontext._chunking_profile_alias_history (
    chunking_profile_alias_id bigint NOT NULL
        REFERENCES pgcontext._chunking_profile_aliases(chunking_profile_alias_id)
        ON DELETE CASCADE,
    alias_revision bigint NOT NULL CHECK (alias_revision > 0),
    chunking_profile_id bigint NOT NULL
        REFERENCES pgcontext._chunking_profiles(chunking_profile_id),
    promoted_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    PRIMARY KEY (chunking_profile_alias_id, alias_revision)
);
CREATE TABLE pgcontext._document_sources (
    document_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:]]'),
    rerank_source_id bigint NOT NULL
        REFERENCES pgcontext._semantic_rerank_sources(rerank_source_id) ON DELETE CASCADE,
    rerank_registration_revision bigint NOT NULL CHECK (rerank_registration_revision > 0),
    chunking_profile_alias_id bigint NOT NULL
        REFERENCES pgcontext._chunking_profile_aliases(chunking_profile_alias_id),
    projection_table_oid oid NOT NULL,
    projection_schema_name text NOT NULL,
    projection_table_name text NOT NULL,
    projection_column_attnums int2[] NOT NULL CHECK (
        pg_catalog.cardinality(projection_column_attnums) = 28),
    projection_column_type_oids oid[] NOT NULL CHECK (
        pg_catalog.cardinality(projection_column_type_oids) = 28),
    projection_column_collation_oids oid[] NOT NULL CHECK (
        pg_catalog.cardinality(projection_column_collation_oids) = 28),
    registration_system_identifier bigint NOT NULL,
    registration_database_oid oid NOT NULL,
    registration_revision bigint NOT NULL DEFAULT 1 CHECK (registration_revision > 0),
    status text NOT NULL DEFAULT 'ready'
        CHECK (status IN ('ready', 'stale', 'retired', '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._document_chunk_generations (
    generation_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    document_source_id bigint NOT NULL
        REFERENCES pgcontext._document_sources(document_source_id) ON DELETE CASCADE,
    source_key text NOT NULL CHECK (pg_catalog.octet_length(source_key) BETWEEN 1 AND 1024),
    source_version bigint NOT NULL CHECK (source_version > 0),
    source_sha256 bytea NOT NULL CHECK (pg_catalog.octet_length(source_sha256) = 32),
    chunking_profile_id bigint NOT NULL
        REFERENCES pgcontext._chunking_profiles(chunking_profile_id),
    source_registration_revision bigint NOT NULL CHECK (source_registration_revision > 0),
    status text NOT NULL DEFAULT 'queued' CHECK (
        status IN ('queued', 'leased', 'parsing', 'chunking', 'embedding',
                   'validating', 'publishing', 'ready', 'cancel_requested',
                   'cancelled', 'failed', 'superseded', 'retired')
    ),
    prior_generation_id bigint
        REFERENCES pgcontext._document_chunk_generations(generation_id),
    chunk_count int4,
    token_count bigint,
    staging_bytes bigint,
    publication_sha256 bytea,
    projection_sha256 bytea,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    published_at timestamptz,
    UNIQUE (document_source_id, source_key, source_version, chunking_profile_id,
        source_registration_revision),
    CHECK (publication_sha256 IS NULL OR pg_catalog.octet_length(publication_sha256) = 32),
    CHECK (projection_sha256 IS NULL OR pg_catalog.octet_length(projection_sha256) = 32),
    CHECK ((status IN ('ready', 'retired')) = (published_at IS NOT NULL))
);
CREATE TABLE pgcontext._document_chunk_jobs (
    job_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    generation_id bigint NOT NULL UNIQUE
        REFERENCES pgcontext._document_chunk_generations(generation_id) ON DELETE CASCADE,
    status text NOT NULL DEFAULT 'queued' CHECK (
        status IN ('queued', 'leased', 'parsing', 'chunking', 'embedding',
                   'validating', 'publishing', 'ready', 'cancel_requested',
                   'cancelled', 'failed', 'superseded', 'retired')
    ),
    attempt int4 NOT NULL DEFAULT 0 CHECK (attempt BETWEEN 0 AND 3),
    lease_worker text,
    lease_token bigint NOT NULL DEFAULT 0 CHECK (lease_token >= 0),
    lease_expires_at timestamptz,
    processed_units bigint NOT NULL DEFAULT 0 CHECK (processed_units >= 0),
    total_units bigint NOT NULL DEFAULT 1 CHECK (total_units > 0),
    error_code text CHECK (error_code IS NULL OR (
        pg_catalog.octet_length(error_code) BETWEEN 1 AND 64
        AND error_code ~ '^[a-z0-9_]+$')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    CHECK ((lease_worker IS NULL) = (lease_expires_at IS NULL))
);
CREATE TABLE pgcontext._document_chunk_staging (
    job_id bigint PRIMARY KEY
        REFERENCES pgcontext._document_chunk_jobs(job_id) ON DELETE CASCADE,
    lease_token bigint NOT NULL CHECK (lease_token > 0),
    response_json jsonb NOT NULL,
    response_sha256 bytea NOT NULL CHECK (pg_catalog.octet_length(response_sha256) = 32),
    chunk_count int4 NOT NULL CHECK (chunk_count BETWEEN 0 AND 16384),
    token_count bigint NOT NULL CHECK (token_count >= 0),
    staging_bytes bigint NOT NULL CHECK (staging_bytes BETWEEN 1 AND 33554432),
    staged_at timestamptz NOT NULL DEFAULT pg_catalog.now()
);
CREATE TABLE pgcontext._current_document_chunk_generations (
    document_source_id bigint NOT NULL
        REFERENCES pgcontext._document_sources(document_source_id) ON DELETE CASCADE,
    source_key text NOT NULL,
    chunking_profile_id bigint NOT NULL
        REFERENCES pgcontext._chunking_profiles(chunking_profile_id),
    generation_id bigint NOT NULL
        REFERENCES pgcontext._document_chunk_generations(generation_id),
    publication_revision bigint NOT NULL DEFAULT 1 CHECK (publication_revision > 0),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    PRIMARY KEY (document_source_id, source_key, chunking_profile_id)
);
CREATE TABLE pgcontext._document_embedding_jobs (
    generation_id bigint NOT NULL
        REFERENCES pgcontext._document_chunk_generations(generation_id) ON DELETE CASCADE,
    occurrence_id bigint NOT NULL CHECK (occurrence_id > 0),
    content_hash bigint NOT NULL,
    embedding_profile_id bigint NOT NULL DEFAULT 0 CHECK (embedding_profile_id >= 0),
    status text NOT NULL DEFAULT 'ready' CHECK (status IN ('queued', 'ready', 'failed')),
    fake_embedding jsonb NOT NULL CHECK (pg_catalog.jsonb_typeof(fake_embedding) = 'array'),
    PRIMARY KEY (generation_id, occurrence_id, embedding_profile_id)
);
REVOKE ALL ON TABLE pgcontext._chunking_profiles FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._chunking_profile_aliases FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._chunking_profile_alias_history FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._document_sources FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._document_chunk_generations FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._document_chunk_jobs FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._document_chunk_staging FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._current_document_chunk_generations FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._document_embedding_jobs FROM PUBLIC;
CREATE VIEW pgcontext._visible_chunking_profiles
WITH (security_barrier = true) AS
SELECT profiles.*
  FROM pgcontext._chunking_profiles AS profiles
 WHERE pg_catalog.pg_has_role(SESSION_USER, profiles.owner_role, 'MEMBER');
CREATE VIEW pgcontext._visible_chunking_profile_aliases
WITH (security_barrier = true) AS
SELECT aliases.*
  FROM pgcontext._chunking_profile_aliases AS aliases
 WHERE pg_catalog.pg_has_role(SESSION_USER, aliases.owner_role, 'MEMBER');
CREATE VIEW pgcontext._visible_document_sources
WITH (security_barrier = true) AS
SELECT sources.document_source_id, sources.collection_id, sources.source_name,
       sources.rerank_source_id, sources.rerank_registration_revision,
       aliases.chunking_profile_id, sources.chunking_profile_alias_id,
       sources.projection_table_oid, sources.projection_schema_name,
       sources.projection_table_name, sources.projection_column_attnums,
       sources.projection_column_type_oids, sources.projection_column_collation_oids,
       sources.registration_revision, sources.status, sources.created_at,
       sources.updated_at
  FROM pgcontext._document_sources AS sources
  JOIN pgcontext._chunking_profile_aliases AS aliases USING (chunking_profile_alias_id)
  JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER');
CREATE VIEW pgcontext._visible_document_chunk_generations
WITH (security_barrier = true) AS
SELECT generations.*
  FROM pgcontext._document_chunk_generations AS generations
 JOIN pgcontext._document_sources AS sources USING (document_source_id)
 JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
   AND pgcontext._document_chunk_source_key_visible(
           generations.document_source_id, generations.source_key
       );
CREATE VIEW pgcontext._visible_document_chunk_jobs
WITH (security_barrier = true) AS
SELECT jobs.job_id, jobs.generation_id, jobs.status, jobs.attempt,
       jobs.processed_units, jobs.total_units, jobs.error_code,
       jobs.created_at, jobs.updated_at,
       generations.document_source_id, generations.source_version,
       generations.chunking_profile_id
  FROM pgcontext._document_chunk_jobs AS jobs
  JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
 JOIN pgcontext._document_sources AS sources USING (document_source_id)
 JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
   AND pgcontext._document_chunk_source_key_visible(
           generations.document_source_id, generations.source_key
       );
CREATE VIEW pgcontext._visible_current_document_chunk_generations
WITH (security_barrier = true) AS
SELECT aliases.*, sources.collection_id, sources.source_name,
       sources.projection_table_oid, sources.projection_schema_name,
       sources.projection_table_name
  FROM pgcontext._current_document_chunk_generations AS aliases
 JOIN pgcontext._document_sources AS sources USING (document_source_id)
 JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
   AND pgcontext._document_chunk_source_key_visible(
           aliases.document_source_id, aliases.source_key
       );
CREATE VIEW pgcontext._visible_document_chunk_staging
WITH (security_barrier = true) AS
SELECT staging.job_id, staging.chunk_count, staging.token_count,
       staging.staging_bytes, staging.staged_at
  FROM pgcontext._document_chunk_staging AS staging
  JOIN pgcontext._document_chunk_jobs AS jobs USING (job_id)
  JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
 JOIN pgcontext._document_sources AS sources USING (document_source_id)
 JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
   AND pgcontext._document_chunk_source_key_visible(
           generations.document_source_id, generations.source_key
       );
CREATE VIEW pgcontext._visible_document_embedding_jobs
WITH (security_barrier = true) AS
SELECT embedding.*
  FROM pgcontext._document_embedding_jobs AS embedding
  JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
 JOIN pgcontext._document_sources AS sources USING (document_source_id)
 JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
   AND pgcontext._document_chunk_source_key_visible(
           generations.document_source_id, generations.source_key
       );
GRANT SELECT ON pgcontext._visible_chunking_profiles,
    pgcontext._visible_chunking_profile_aliases,
    pgcontext._visible_document_sources,
    pgcontext._visible_document_chunk_generations,
    pgcontext._visible_document_chunk_jobs,
    pgcontext._visible_current_document_chunk_generations,
    pgcontext._visible_document_chunk_staging,
    pgcontext._visible_document_embedding_jobs TO PUBLIC;
CREATE FUNCTION pgcontext._register_document_source(
    p_collection_id bigint,
    p_source_name text,
    p_rerank_source_id bigint,
    p_chunking_profile_id bigint,
    p_chunking_profile_alias_id bigint,
    p_projection_table_oid oid,
    p_projection_schema_name text,
    p_projection_table_name text,
    p_projection_column_attnums int2[],
    p_projection_column_type_oids oid[],
    p_projection_column_collation_oids bigint[]
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_id bigint;
    rerank_revision bigint;
    changed boolean;
    projection_collation_oids oid[];
    alias_profile_id bigint;
    origin_system_identifier bigint;
    origin_database_oid oid;
BEGIN
    PERFORM pgcontext._consume_document_chunk_permit(
        4, p_collection_id, p_rerank_source_id
    );
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    SELECT pg_catalog.array_agg(value::oid ORDER BY ordinal)
      INTO projection_collation_oids
      FROM pg_catalog.unnest(p_projection_column_collation_oids)
           WITH ORDINALITY AS collations(value, ordinal);
    SELECT system_identifier INTO origin_system_identifier
      FROM pg_catalog.pg_control_system();
    SELECT oid INTO origin_database_oid
      FROM pg_catalog.pg_database WHERE datname = pg_catalog.current_database();
    SELECT registration_revision INTO rerank_revision
      FROM pgcontext._semantic_rerank_sources
     WHERE rerank_source_id = p_rerank_source_id
       AND collection_id = p_collection_id
       AND status = 'ready';
    IF rerank_revision IS NULL THEN
        RAISE EXCEPTION 'semantic rerank source does not match the document collection'
            USING ERRCODE = '42704';
    END IF;
    SELECT aliases.chunking_profile_id INTO alias_profile_id
      FROM pgcontext._chunking_profile_aliases AS aliases
      JOIN pgcontext._chunking_profiles AS profiles USING (chunking_profile_id)
     WHERE aliases.chunking_profile_alias_id = p_chunking_profile_alias_id
       AND aliases.status = 'ready' AND profiles.status = 'ready'
       AND pg_catalog.pg_has_role(SESSION_USER, aliases.owner_role, 'MEMBER')
     FOR SHARE OF aliases;
    IF alias_profile_id IS DISTINCT FROM p_chunking_profile_id THEN
        RAISE EXCEPTION 'chunking profile does not exist or is not ready'
            USING ERRCODE = '42704';
    END IF;
    PERFORM pg_catalog.pg_advisory_xact_lock(pg_catalog.hashtextextended(
        p_collection_id::text || E'\x1fdocument\x1f' || p_source_name, 0
    ));
    PERFORM 1
      FROM pgcontext._document_sources AS existing
     WHERE existing.collection_id = p_collection_id
       AND existing.source_name = p_source_name
     FOR UPDATE;
    SELECT EXISTS (
        SELECT 1 FROM pgcontext._document_sources AS existing
         WHERE existing.collection_id = p_collection_id
           AND existing.source_name = p_source_name
           AND (
               existing.rerank_source_id IS DISTINCT FROM p_rerank_source_id
               OR existing.rerank_registration_revision IS DISTINCT FROM rerank_revision
               OR existing.chunking_profile_alias_id IS DISTINCT FROM p_chunking_profile_alias_id
               OR existing.projection_table_oid IS DISTINCT FROM p_projection_table_oid
               OR existing.projection_schema_name IS DISTINCT FROM p_projection_schema_name
               OR existing.projection_table_name IS DISTINCT FROM p_projection_table_name
               OR existing.projection_column_attnums IS DISTINCT FROM p_projection_column_attnums
               OR existing.projection_column_type_oids IS DISTINCT FROM p_projection_column_type_oids
               OR existing.projection_column_collation_oids IS DISTINCT FROM projection_collation_oids
           )
    ) INTO changed;
    INSERT INTO pgcontext._document_sources (
        collection_id, source_name, rerank_source_id, rerank_registration_revision,
        chunking_profile_alias_id,
        projection_table_oid, projection_schema_name, projection_table_name,
        projection_column_attnums, projection_column_type_oids,
        projection_column_collation_oids,
        registration_system_identifier, registration_database_oid
    ) VALUES (
        p_collection_id, p_source_name, p_rerank_source_id, rerank_revision,
        p_chunking_profile_alias_id,
        p_projection_table_oid, p_projection_schema_name, p_projection_table_name,
        p_projection_column_attnums, p_projection_column_type_oids,
        projection_collation_oids, origin_system_identifier, origin_database_oid
    )
    ON CONFLICT (collection_id, source_name) DO UPDATE
       SET rerank_source_id = EXCLUDED.rerank_source_id,
           rerank_registration_revision = EXCLUDED.rerank_registration_revision,
           chunking_profile_alias_id = EXCLUDED.chunking_profile_alias_id,
           projection_table_oid = EXCLUDED.projection_table_oid,
           projection_schema_name = EXCLUDED.projection_schema_name,
           projection_table_name = EXCLUDED.projection_table_name,
           projection_column_attnums = EXCLUDED.projection_column_attnums,
           projection_column_type_oids = EXCLUDED.projection_column_type_oids,
           projection_column_collation_oids = EXCLUDED.projection_column_collation_oids,
           registration_system_identifier = EXCLUDED.registration_system_identifier,
           registration_database_oid = EXCLUDED.registration_database_oid,
           registration_revision = pgcontext._document_sources.registration_revision +
               CASE WHEN changed THEN 1 ELSE 0 END,
           status = 'ready', updated_at = pg_catalog.now()
    RETURNING document_source_id INTO source_id;
    IF changed THEN
        PERFORM 1
          FROM pgcontext._document_chunk_jobs AS jobs
          JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
         WHERE generations.document_source_id = source_id
         ORDER BY jobs.job_id FOR UPDATE OF jobs;
        UPDATE pgcontext._document_chunk_generations
           SET status = 'retired'
         WHERE document_source_id = source_id AND status = 'ready';
        UPDATE pgcontext._document_chunk_jobs AS jobs
           SET status = CASE
                   WHEN generations.status = 'retired' THEN 'retired'
                   ELSE 'superseded'
               END,
               lease_worker = NULL, lease_expires_at = NULL,
               updated_at = pg_catalog.now()
          FROM pgcontext._document_chunk_generations AS generations
         WHERE jobs.generation_id = generations.generation_id
           AND generations.document_source_id = source_id
           AND jobs.status NOT IN ('retired','superseded');
        UPDATE pgcontext._document_chunk_generations
           SET status = 'superseded'
         WHERE document_source_id = source_id
           AND status NOT IN ('ready','retired','superseded');
        DELETE FROM pgcontext._document_chunk_staging AS staging
         USING pgcontext._document_chunk_jobs AS jobs,
               pgcontext._document_chunk_generations AS generations
         WHERE staging.job_id = jobs.job_id AND jobs.generation_id = generations.generation_id
           AND generations.document_source_id = source_id
           AND generations.status = 'superseded';
        DELETE FROM pgcontext._current_document_chunk_generations
         WHERE document_source_id = source_id;
    END IF;
    RETURN source_id;
END;
$$;
CREATE FUNCTION pgcontext._refresh_document_chunk_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._document_sources%ROWTYPE;
    current_projection_oid oid;
    current_projection_attnums int2[];
    current_projection_type_oids oid[];
    current_projection_collation_oids oid[];
    current_rerank_revision bigint;
    identity_changed boolean;
    current_system_identifier bigint;
    current_database_oid oid;
    logical_restore boolean;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    PERFORM pgcontext._refresh_semantic_rerank_source(p_collection_id, p_source_name);
    PERFORM pg_catalog.pg_advisory_xact_lock(pg_catalog.hashtextextended(
        p_collection_id::text || E'\x1fdocument\x1f' || p_source_name, 0
    ));
    SELECT * INTO registration
      FROM pgcontext._document_sources
     WHERE collection_id = p_collection_id AND source_name = p_source_name
     FOR UPDATE;
    IF NOT FOUND THEN
        RETURN;
    END IF;
    SELECT system_identifier INTO current_system_identifier
      FROM pg_catalog.pg_control_system();
    SELECT oid INTO current_database_oid
      FROM pg_catalog.pg_database WHERE datname = pg_catalog.current_database();
    logical_restore := registration.registration_system_identifier IS DISTINCT FROM
            current_system_identifier
        OR registration.registration_database_oid IS DISTINCT FROM current_database_oid;
    SELECT rerank.registration_revision
      INTO current_rerank_revision
      FROM pgcontext._semantic_rerank_sources AS rerank
     WHERE rerank.rerank_source_id = registration.rerank_source_id
       AND rerank.collection_id = p_collection_id
       AND rerank.status = 'ready';
    SELECT relation.oid
      INTO current_projection_oid
      FROM pg_catalog.pg_namespace AS namespace
      JOIN pg_catalog.pg_class AS relation ON relation.relnamespace = namespace.oid
     WHERE namespace.nspname = registration.projection_schema_name
       AND relation.relname = registration.projection_table_name
       AND relation.relkind IN ('r', 'p');
    IF current_projection_oid IS NOT NULL THEN
        SELECT pg_catalog.array_agg(attribute.attnum ORDER BY required.ordinal),
               pg_catalog.array_agg(attribute.atttypid ORDER BY required.ordinal),
               pg_catalog.array_agg(attribute.attcollation ORDER BY required.ordinal)
          INTO current_projection_attnums, current_projection_type_oids,
               current_projection_collation_oids
          FROM pg_catalog.unnest(ARRAY[
              'document_source_id','source_key','source_version','source_sha256',
              'profile_revision','generation_id','occurrence_id','ordinal',
              'original_text','retrieval_text','start_byte','end_byte','start_char',
              'end_char','token_count','structure_kind','structure_path',
              'page_number','region',
              'parent_occurrence_id','previous_occurrence_id','next_occurrence_id',
              'content_hash','context_prefix','context_prefix_hash','fake_embedding',
              'ready','provenance'
          ]::text[]) WITH ORDINALITY AS required(column_name, ordinal)
          LEFT JOIN pg_catalog.pg_attribute AS attribute
            ON attribute.attrelid = current_projection_oid
           AND attribute.attname = required.column_name
           AND attribute.attnum > 0 AND NOT attribute.attisdropped;
    END IF;
    IF current_rerank_revision IS NULL OR current_projection_oid IS NULL
       OR current_projection_attnums IS NULL
       OR pg_catalog.cardinality(current_projection_attnums) <> 28
       OR (NOT logical_restore AND (
           current_projection_attnums IS DISTINCT FROM registration.projection_column_attnums
           OR current_projection_type_oids IS DISTINCT FROM registration.projection_column_type_oids
           OR current_projection_collation_oids IS DISTINCT FROM
              registration.projection_column_collation_oids
       )) THEN
        UPDATE pgcontext._document_sources
           SET status = 'stale', updated_at = pg_catalog.now()
         WHERE document_source_id = registration.document_source_id;
        RETURN;
    END IF;
    identity_changed := NOT logical_restore AND (
        current_projection_oid IS DISTINCT FROM registration.projection_table_oid
        OR current_rerank_revision IS DISTINCT FROM registration.rerank_registration_revision
    );
    UPDATE pgcontext._document_sources AS sources
       SET projection_table_oid = current_projection_oid,
           rerank_registration_revision = current_rerank_revision,
           projection_column_attnums = CASE WHEN logical_restore
               THEN current_projection_attnums ELSE sources.projection_column_attnums END,
           projection_column_type_oids = CASE WHEN logical_restore
               THEN current_projection_type_oids ELSE sources.projection_column_type_oids END,
           projection_column_collation_oids = CASE WHEN logical_restore
               THEN current_projection_collation_oids
               ELSE sources.projection_column_collation_oids END,
           registration_system_identifier = current_system_identifier,
           registration_database_oid = current_database_oid,
           registration_revision = sources.registration_revision +
               CASE WHEN identity_changed THEN 1 ELSE 0 END,
           status = 'ready', updated_at = pg_catalog.now()
     WHERE sources.document_source_id = registration.document_source_id;
    IF identity_changed THEN
        PERFORM 1
          FROM pgcontext._document_chunk_jobs AS jobs
          JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
         WHERE generations.document_source_id = registration.document_source_id
         ORDER BY jobs.job_id FOR UPDATE OF jobs;
        UPDATE pgcontext._document_chunk_generations
           SET status = CASE WHEN status = 'ready' THEN 'retired' ELSE 'superseded' END
         WHERE document_source_id = registration.document_source_id
           AND status NOT IN ('retired','superseded');
        UPDATE pgcontext._document_chunk_jobs AS jobs
           SET status = CASE
                   WHEN generations.status = 'retired' THEN 'retired'
                   ELSE 'superseded'
               END,
               lease_worker = NULL, lease_expires_at = NULL,
               updated_at = pg_catalog.now()
          FROM pgcontext._document_chunk_generations AS generations
         WHERE jobs.generation_id = generations.generation_id
           AND generations.document_source_id = registration.document_source_id
           AND jobs.status NOT IN ('retired','superseded');
        DELETE FROM pgcontext._document_chunk_staging AS staging
         USING pgcontext._document_chunk_jobs AS jobs,
               pgcontext._document_chunk_generations AS generations
         WHERE staging.job_id = jobs.job_id
           AND jobs.generation_id = generations.generation_id
           AND generations.document_source_id = registration.document_source_id;
        DELETE FROM pgcontext._current_document_chunk_generations
         WHERE document_source_id = registration.document_source_id;
    END IF;
END;
$$;
CREATE FUNCTION pgcontext._enqueue_document_chunk_job(
    p_document_source_id bigint,
    p_source_key text,
    p_source_version bigint,
    p_source_sha256 bytea,
    p_chunking_profile_id bigint
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_row pgcontext._document_sources%ROWTYPE;
    v_generation_id bigint;
    v_job_id bigint;
    prior_id bigint;
    current_profile_id bigint;
BEGIN
    PERFORM pgcontext._consume_document_chunk_permit(
        11, p_document_source_id, p_source_version
    );
    SELECT * INTO source_row FROM pgcontext._document_sources
     WHERE document_source_id = p_document_source_id AND status = 'ready'
     FOR SHARE;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'document source does not exist or is not ready'
            USING ERRCODE = '42704';
    END IF;
    SELECT CASE
               WHEN p_chunking_profile_id = aliases.chunking_profile_id
                 OR p_chunking_profile_id = aliases.shadow_chunking_profile_id
               THEN p_chunking_profile_id
           END INTO current_profile_id
      FROM pgcontext._chunking_profile_aliases AS aliases
     WHERE aliases.chunking_profile_alias_id = source_row.chunking_profile_alias_id
       AND aliases.status = 'ready'
     FOR SHARE;
    IF current_profile_id IS NULL THEN
        RAISE EXCEPTION 'chunking profile alias is unavailable' USING ERRCODE = '42704';
    END IF;
    PERFORM pgcontext._require_collection_owner(source_row.collection_id);
    SELECT aliases.generation_id INTO prior_id
      FROM pgcontext._current_document_chunk_generations AS aliases
     WHERE aliases.document_source_id = p_document_source_id
       AND aliases.source_key = p_source_key
       AND aliases.chunking_profile_id = current_profile_id;
    PERFORM 1
      FROM pgcontext._document_chunk_jobs AS jobs
      JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
     WHERE generations.document_source_id = p_document_source_id
       AND generations.source_key = p_source_key
       AND generations.chunking_profile_id = current_profile_id
     ORDER BY jobs.job_id FOR UPDATE OF jobs;
    UPDATE pgcontext._document_chunk_generations
       SET status = 'superseded'
     WHERE document_source_id = p_document_source_id
       AND source_key = p_source_key
       AND chunking_profile_id = current_profile_id
       AND source_version < p_source_version
       AND status NOT IN ('ready', 'retired', 'superseded');
    UPDATE pgcontext._document_chunk_jobs AS jobs
       SET status = 'superseded', lease_worker = NULL, lease_expires_at = NULL,
           updated_at = pg_catalog.now()
      FROM pgcontext._document_chunk_generations AS generations
     WHERE jobs.generation_id = generations.generation_id
       AND generations.document_source_id = p_document_source_id
       AND generations.source_key = p_source_key
       AND generations.chunking_profile_id = current_profile_id
       AND generations.source_version < p_source_version
       AND jobs.status NOT IN ('ready', 'retired', 'superseded');
    INSERT INTO pgcontext._document_chunk_generations (
        document_source_id, source_key, source_version, source_sha256,
        chunking_profile_id, source_registration_revision, prior_generation_id
    ) VALUES (
        p_document_source_id, p_source_key, p_source_version, p_source_sha256,
        current_profile_id, source_row.registration_revision, prior_id
    )
    ON CONFLICT (
        document_source_id, source_key, source_version, chunking_profile_id,
        source_registration_revision
    )
    DO UPDATE SET source_sha256 = pgcontext._document_chunk_generations.source_sha256
      WHERE pgcontext._document_chunk_generations.source_sha256 = EXCLUDED.source_sha256
        AND pgcontext._document_chunk_generations.source_registration_revision =
            EXCLUDED.source_registration_revision
    RETURNING pgcontext._document_chunk_generations.generation_id INTO v_generation_id;
    IF v_generation_id IS NULL THEN
        RAISE EXCEPTION 'document source version identity changed'
            USING ERRCODE = '55000';
    END IF;
    INSERT INTO pgcontext._document_chunk_jobs (generation_id)
    VALUES (v_generation_id)
    ON CONFLICT (generation_id) DO UPDATE SET updated_at = pg_catalog.now()
    RETURNING pgcontext._document_chunk_jobs.job_id INTO v_job_id;
    RETURN v_job_id;
END;
$$;
CREATE FUNCTION pgcontext._claim_document_chunk_jobs(
    p_limit int4,
    p_lease_millis int4,
    p_worker_id text
)
RETURNS TABLE(job_id bigint, lease_token bigint)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    PERFORM pgcontext._consume_document_chunk_permit(
        12, p_limit::bigint, p_lease_millis::bigint
    );
    IF p_limit NOT BETWEEN 1 AND 256 OR p_lease_millis NOT BETWEEN 1 AND 60000
       OR pg_catalog.octet_length(p_worker_id) NOT BETWEEN 1 AND 128
       OR p_worker_id ~ '[[:cntrl:]]' THEN
        RAISE EXCEPTION 'invalid document chunk claim bounds' USING ERRCODE = '22023';
    END IF;
    WITH exhausted_ids AS (
        SELECT jobs.job_id
          FROM pgcontext._document_chunk_jobs AS jobs
          JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
          JOIN pgcontext._document_sources AS sources USING (document_source_id)
          JOIN pgcontext._collections AS collections USING (collection_id)
         WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
           AND (jobs.status = 'cancel_requested' OR jobs.attempt >= 3)
           AND jobs.status IN (
               'leased','parsing','chunking','embedding','validating','publishing',
               'cancel_requested'
           )
           AND jobs.lease_expires_at <= pg_catalog.clock_timestamp()
         ORDER BY jobs.updated_at, jobs.job_id
         LIMIT p_limit
         FOR UPDATE OF jobs SKIP LOCKED
    ), exhausted AS (
        UPDATE pgcontext._document_chunk_jobs AS jobs
           SET status = CASE
                   WHEN jobs.status = 'cancel_requested' THEN 'cancelled'
                   ELSE 'failed'
               END,
               lease_worker = NULL, lease_expires_at = NULL,
               error_code = CASE
                   WHEN jobs.status = 'cancel_requested' THEN 'cancelled'
                   ELSE 'attempts_exhausted'
               END,
               updated_at = pg_catalog.now()
          FROM exhausted_ids
         WHERE jobs.job_id = exhausted_ids.job_id
        RETURNING jobs.job_id, jobs.generation_id, jobs.status
    ), updated_generations AS (
        UPDATE pgcontext._document_chunk_generations AS generations
           SET status = exhausted.status
          FROM exhausted
         WHERE generations.generation_id = exhausted.generation_id
        RETURNING exhausted.job_id
    )
    DELETE FROM pgcontext._document_chunk_staging AS staging
     USING updated_generations
    WHERE staging.job_id = updated_generations.job_id;
    WITH mismatched_ids AS (
        SELECT jobs.job_id
          FROM pgcontext._document_chunk_jobs AS jobs
          JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
          JOIN pgcontext._document_sources AS sources USING (document_source_id)
          JOIN pgcontext._chunking_profile_aliases AS aliases
            USING (chunking_profile_alias_id)
          JOIN pgcontext._collections AS collections USING (collection_id)
         WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
           AND generations.chunking_profile_id NOT IN (
               aliases.chunking_profile_id,
               COALESCE(aliases.shadow_chunking_profile_id, aliases.chunking_profile_id)
           )
           AND jobs.status NOT IN ('ready','retired','superseded','cancelled','failed')
         ORDER BY jobs.updated_at, jobs.job_id
         LIMIT p_limit
         FOR SHARE OF aliases
         FOR UPDATE OF jobs SKIP LOCKED
    ), mismatched AS (
        UPDATE pgcontext._document_chunk_jobs AS jobs
           SET status = 'superseded', lease_worker = NULL, lease_expires_at = NULL,
               updated_at = pg_catalog.now()
          FROM mismatched_ids
         WHERE jobs.job_id = mismatched_ids.job_id
        RETURNING jobs.job_id, jobs.generation_id
    ), superseded_generations AS (
        UPDATE pgcontext._document_chunk_generations AS generations
           SET status = 'superseded'
          FROM mismatched
         WHERE generations.generation_id = mismatched.generation_id
        RETURNING mismatched.job_id
    )
    DELETE FROM pgcontext._document_chunk_staging AS staging
     USING superseded_generations
    WHERE staging.job_id = superseded_generations.job_id;
    /* Cleanup is bounded by p_limit; later claims continue the work. */
    RETURN QUERY
    WITH candidates AS (
        SELECT jobs.job_id
          FROM pgcontext._document_chunk_jobs AS jobs
          JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
          JOIN pgcontext._document_sources AS sources USING (document_source_id)
          JOIN pgcontext._chunking_profile_aliases AS aliases
            USING (chunking_profile_alias_id)
          JOIN pgcontext._collections AS collections USING (collection_id)
         WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER')
           AND jobs.attempt < 3
           AND generations.chunking_profile_id IN (
               aliases.chunking_profile_id,
               COALESCE(aliases.shadow_chunking_profile_id, aliases.chunking_profile_id)
           )
           AND (
               jobs.status = 'queued'
               OR (jobs.status IN ('leased','parsing','chunking','embedding','validating','publishing')
                   AND jobs.lease_expires_at <= pg_catalog.clock_timestamp())
           )
         ORDER BY jobs.updated_at, jobs.job_id
         LIMIT p_limit
         FOR SHARE OF sources, aliases
         FOR UPDATE OF jobs SKIP LOCKED
    ), claimed AS (
        UPDATE pgcontext._document_chunk_jobs AS jobs
           SET status = 'leased',
               attempt = jobs.attempt + 1,
               lease_worker = p_worker_id,
               lease_token = jobs.lease_token + 1,
               lease_expires_at = pg_catalog.clock_timestamp()
                   + pg_catalog.make_interval(secs => p_lease_millis::double precision / 1000.0),
               processed_units = 0,
               error_code = NULL,
               updated_at = pg_catalog.now()
          FROM candidates
         WHERE jobs.job_id = candidates.job_id
        RETURNING jobs.job_id, jobs.lease_token
    ), reset_generations AS (
        UPDATE pgcontext._document_chunk_generations AS generations
           SET status = 'leased'
          FROM pgcontext._document_chunk_jobs AS jobs, claimed
         WHERE jobs.job_id = claimed.job_id
           AND generations.generation_id = jobs.generation_id
        RETURNING claimed.job_id, claimed.lease_token
    )
    SELECT reset_generations.job_id, reset_generations.lease_token
      FROM reset_generations ORDER BY reset_generations.job_id;
END;
$$;
CREATE FUNCTION pgcontext._stage_document_chunk_response(
    p_job_id bigint,
    p_lease_token bigint,
    p_response_json jsonb,
    p_response_sha256 bytea,
    p_chunk_count int4,
    p_token_count bigint,
    p_staging_bytes bigint
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    job_row pgcontext._document_chunk_jobs%ROWTYPE;
    collection_id bigint;
BEGIN
    PERFORM pgcontext._consume_document_chunk_permit(
        1, p_job_id, p_lease_token
    );
    SELECT sources.collection_id INTO collection_id
      FROM pgcontext._document_chunk_jobs AS jobs
      JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
      JOIN pgcontext._document_sources AS sources USING (document_source_id)
     WHERE jobs.job_id = p_job_id
     FOR SHARE OF sources;
    SELECT * INTO job_row FROM pgcontext._document_chunk_jobs WHERE job_id = p_job_id FOR UPDATE;
    IF NOT FOUND OR job_row.lease_token <> p_lease_token
       OR job_row.lease_expires_at <= pg_catalog.clock_timestamp()
       OR job_row.status IN ('cancel_requested','cancelled','failed','superseded','retired','ready') THEN
        RAISE EXCEPTION 'stale document chunk job lease' USING ERRCODE = '40001';
    END IF;
    PERFORM pgcontext._require_collection_owner(collection_id);
    INSERT INTO pgcontext._document_chunk_staging (
        job_id, lease_token, response_json, response_sha256,
        chunk_count, token_count, staging_bytes
    ) VALUES (
        p_job_id, p_lease_token, p_response_json, p_response_sha256,
        p_chunk_count, p_token_count, p_staging_bytes
    )
    ON CONFLICT (job_id) DO UPDATE
       SET lease_token = EXCLUDED.lease_token,
           response_json = EXCLUDED.response_json,
           response_sha256 = EXCLUDED.response_sha256,
           chunk_count = EXCLUDED.chunk_count,
           token_count = EXCLUDED.token_count,
           staging_bytes = EXCLUDED.staging_bytes,
           staged_at = pg_catalog.now();
    UPDATE pgcontext._document_chunk_jobs
       SET status = 'validating', processed_units = total_units, updated_at = pg_catalog.now()
     WHERE job_id = p_job_id;
    UPDATE pgcontext._document_chunk_generations AS generations
       SET status = 'validating'
      FROM pgcontext._document_chunk_jobs AS jobs
     WHERE jobs.job_id = p_job_id AND generations.generation_id = jobs.generation_id;
END;
$$;
CREATE FUNCTION pgcontext._complete_document_chunk_publication(
    p_job_id bigint,
    p_lease_token bigint,
    p_response_sha256 bytea,
    p_projection_sha256 bytea,
    p_chunk_count int4,
    p_token_count bigint,
    p_staging_bytes bigint
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    job_row pgcontext._document_chunk_jobs%ROWTYPE;
    generation_row pgcontext._document_chunk_generations%ROWTYPE;
    source_row pgcontext._document_sources%ROWTYPE;
    embedding_count bigint;
BEGIN
    PERFORM pgcontext._consume_document_chunk_permit(
        2, p_job_id, p_lease_token
    );
    SELECT sources.* INTO source_row
      FROM pgcontext._document_chunk_jobs AS jobs
      JOIN pgcontext._document_chunk_generations AS generations USING (generation_id)
      JOIN pgcontext._document_sources AS sources USING (document_source_id)
     WHERE jobs.job_id = p_job_id
     FOR SHARE OF sources;
    SELECT * INTO job_row FROM pgcontext._document_chunk_jobs
     WHERE job_id = p_job_id FOR UPDATE;
    IF FOUND AND job_row.lease_token = p_lease_token AND job_row.status = 'ready' THEN
        SELECT * INTO generation_row FROM pgcontext._document_chunk_generations
         WHERE generation_id = job_row.generation_id FOR UPDATE;
        IF generation_row.status = 'ready'
           AND generation_row.publication_sha256 = p_response_sha256
           AND generation_row.projection_sha256 = p_projection_sha256
           AND generation_row.chunk_count = p_chunk_count
           AND generation_row.token_count = p_token_count
           AND generation_row.staging_bytes = p_staging_bytes THEN
            RETURN generation_row.generation_id;
        END IF;
        RAISE EXCEPTION 'document chunk publication replay differs from ready generation'
            USING ERRCODE = '55000';
    END IF;
    IF NOT FOUND OR job_row.lease_token <> p_lease_token
       OR job_row.lease_expires_at <= pg_catalog.clock_timestamp()
       OR job_row.status NOT IN ('validating','publishing') THEN
        RAISE EXCEPTION 'stale document chunk job lease' USING ERRCODE = '40001';
    END IF;
    SELECT * INTO generation_row FROM pgcontext._document_chunk_generations
     WHERE generation_id = job_row.generation_id FOR UPDATE;
    PERFORM pgcontext._require_collection_owner(source_row.collection_id);
    IF source_row.registration_revision <> generation_row.source_registration_revision
       OR source_row.status <> 'ready' THEN
        RAISE EXCEPTION 'document source registration changed during chunking'
            USING ERRCODE = '55000';
    END IF;
    UPDATE pgcontext._document_chunk_jobs
       SET status = 'publishing', updated_at = pg_catalog.now()
     WHERE job_id = p_job_id;
    UPDATE pgcontext._document_chunk_generations
       SET status = 'publishing'
     WHERE generation_id = generation_row.generation_id;
    INSERT INTO pgcontext._document_embedding_jobs (
        generation_id, occurrence_id, content_hash, embedding_profile_id,
        status, fake_embedding
    )
    SELECT generation_row.generation_id, chunk.occurrence_id,
           chunk.content_hash, 0, 'ready',
           pg_catalog.jsonb_build_array(
               ((chunk.content_hash & 65535)::double precision / 65535.0),
               (((chunk.content_hash >> 16) & 65535)::double precision / 65535.0),
               (((chunk.content_hash >> 32) & 65535)::double precision / 65535.0)
           )
      FROM pgcontext._document_chunk_staging AS staging,
           pg_catalog.jsonb_to_recordset(staging.response_json->'chunks') AS chunk(
               occurrence_id bigint, content_hash bigint
           )
     WHERE staging.job_id = p_job_id AND staging.lease_token = p_lease_token
    ON CONFLICT (generation_id, occurrence_id, embedding_profile_id) DO UPDATE
       SET status = 'ready', fake_embedding = EXCLUDED.fake_embedding
     WHERE pgcontext._document_embedding_jobs.content_hash = EXCLUDED.content_hash;
    GET DIAGNOSTICS embedding_count = ROW_COUNT;
    IF embedding_count <> p_chunk_count THEN
        RAISE EXCEPTION 'document embedding generation is incomplete'
            USING ERRCODE = '55000';
    END IF;
    INSERT INTO pgcontext._current_document_chunk_generations (
        document_source_id, source_key, chunking_profile_id, generation_id
    ) VALUES (
        generation_row.document_source_id, generation_row.source_key,
        generation_row.chunking_profile_id, generation_row.generation_id
    )
    ON CONFLICT (document_source_id, source_key, chunking_profile_id) DO UPDATE
       SET generation_id = EXCLUDED.generation_id,
           publication_revision =
               pgcontext._current_document_chunk_generations.publication_revision + 1,
           updated_at = pg_catalog.now();
    UPDATE pgcontext._document_chunk_generations
       SET status = 'retired'
     WHERE generation_id = generation_row.prior_generation_id
       AND status = 'ready';
    UPDATE pgcontext._document_chunk_jobs AS jobs
       SET status = 'retired', updated_at = pg_catalog.now()
      FROM pgcontext._document_chunk_generations AS generations
     WHERE jobs.generation_id = generations.generation_id
       AND generations.generation_id = generation_row.prior_generation_id
       AND generations.status = 'retired';
    UPDATE pgcontext._document_chunk_generations
       SET status = 'ready', chunk_count = p_chunk_count,
           token_count = p_token_count, staging_bytes = p_staging_bytes,
           publication_sha256 = p_response_sha256,
           projection_sha256 = p_projection_sha256,
           published_at = pg_catalog.now()
     WHERE generation_id = generation_row.generation_id;
    UPDATE pgcontext._document_chunk_jobs
       SET status = 'ready', lease_worker = NULL, lease_expires_at = NULL,
           updated_at = pg_catalog.now()
     WHERE job_id = p_job_id;
    DELETE FROM pgcontext._document_chunk_staging WHERE job_id = p_job_id;
    RETURN generation_row.generation_id;
END;
$$;
"#,
    name = "create_document_chunking_catalog_tables",
    requires = [
        document_chunk_source_key_visible,
        "create_semantic_rerank_catalog_tables"
    ]
);
