//! SQL schema objects for the pgContext catalog.

pgrx::extension_sql!(
    r#"
CREATE TABLE pgcontext._collections (
    collection_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_name text NOT NULL UNIQUE,
    owner_role oid NOT NULL,
    config_revision bigint NOT NULL DEFAULT 1 CHECK (config_revision > 0),
    source_table_oid oid,
    source_schema_name text,
    source_table_name text,
    strict_mode boolean NOT NULL DEFAULT false,
    max_dimensions int4 CHECK (max_dimensions IS NULL OR max_dimensions > 0),
    max_vectors int4 CHECK (max_vectors IS NULL OR max_vectors > 0),
    max_points bigint CHECK (max_points IS NULL OR max_points > 0),
    max_filter_nodes int4 CHECK (max_filter_nodes IS NULL OR max_filter_nodes > 0),
    max_search_limit int4 CHECK (max_search_limit IS NULL OR max_search_limit > 0),
    max_candidate_budget int4 CHECK (max_candidate_budget IS NULL OR max_candidate_budget > 0),
    query_timeout_ms int4 CHECK (query_timeout_ms IS NULL OR query_timeout_ms > 0),
    max_index_memory_bytes bigint CHECK (
        max_index_memory_bytes IS NULL OR max_index_memory_bytes > 0
    ),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(), updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    CHECK (
        (source_table_oid IS NULL AND source_schema_name IS NULL AND source_table_name IS NULL)
        OR
        (source_table_oid IS NOT NULL AND source_schema_name IS NOT NULL AND source_table_name IS NOT NULL)
    )
);

CREATE TABLE pgcontext._collection_vectors (
    vector_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    vector_name text NOT NULL,
    source_table_oid oid NOT NULL,
    source_schema_name text NOT NULL,
    source_table_name text NOT NULL,
    vector_column_name text NOT NULL,
    vector_attnum int2 NOT NULL,
    hnsw_index_oid oid,
    dimensions int4 NOT NULL CHECK (dimensions > 0),
    metric text NOT NULL CHECK (metric IN ('l2', 'inner_product', 'cosine', 'l1', 'hamming', 'jaccard')),
    hnsw_options jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (pg_catalog.jsonb_typeof(hnsw_options) = 'object'),
    quantization_options jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (pg_catalog.jsonb_typeof(quantization_options) = 'object'),
    status text NOT NULL DEFAULT 'ready' CHECK (status IN ('ready', 'building', 'disabled', 'failed')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(), updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, vector_name),
    UNIQUE (collection_id, vector_column_name)
);

CREATE TABLE pgcontext._collection_sparse_vectors (
    sparse_vector_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    vector_name text NOT NULL,
    source_table_oid oid NOT NULL,
    source_schema_name text NOT NULL,
    source_table_name text NOT NULL,
    vector_column_name text NOT NULL,
    vector_attnum int2 NOT NULL,
    dimensions int4 NOT NULL CHECK (dimensions > 0),
    metric text NOT NULL CHECK (metric IN ('l2', 'inner_product', 'cosine', 'l1', 'hamming', 'jaccard')),
    storage_options jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (pg_catalog.jsonb_typeof(storage_options) = 'object'),
    index_options jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (pg_catalog.jsonb_typeof(index_options) = 'object'),
    status text NOT NULL DEFAULT 'ready' CHECK (status IN ('ready', 'building', 'disabled', 'failed')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(), updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, vector_name),
    UNIQUE (collection_id, vector_column_name)
);

CREATE TABLE pgcontext._collection_points (
    point_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    source_key text NOT NULL,
    deleted_at timestamptz,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, source_key)
);

CREATE TABLE pgcontext._collection_source_revisions (
    collection_id bigint PRIMARY KEY REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    source_version bigint NOT NULL DEFAULT 1 CHECK (source_version > 0),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now()
);

CREATE FUNCTION pgcontext._initialize_collection_source_revision()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    INSERT INTO pgcontext._collection_source_revisions (collection_id)
    VALUES (NEW.collection_id)
    ON CONFLICT (collection_id) DO NOTHING;
    RETURN NEW;
END;
$$;

CREATE TRIGGER pgcontext_initialize_collection_source_revision
AFTER INSERT ON pgcontext._collections
FOR EACH ROW
EXECUTE FUNCTION pgcontext._initialize_collection_source_revision();

CREATE TABLE pgcontext._collection_aliases (
    alias_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    alias_name text NOT NULL UNIQUE,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now()
);

CREATE TABLE pgcontext._collection_payload_columns (
    payload_column_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    filter_key text NOT NULL,
    source_table_oid oid NOT NULL,
    source_schema_name text NOT NULL,
    source_table_name text NOT NULL,
    column_name text NOT NULL,
    column_attnum int2 NOT NULL,
    jsonb_path text[],
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, filter_key),
    UNIQUE NULLS NOT DISTINCT (collection_id, column_name, jsonb_path)
);

CREATE TABLE pgcontext._query_stats (
    query_stat_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    cohort text NOT NULL CHECK (cohort <> ''),
    query_kind text NOT NULL CHECK (query_kind IN ('search', 'search_filtered', 'candidate_recheck', 'hybrid')),
    result_count bigint NOT NULL CHECK (result_count >= 0),
    candidate_count bigint CHECK (candidate_count IS NULL OR candidate_count >= 0),
    rows_rechecked bigint NOT NULL DEFAULT 0 CHECK (rows_rechecked >= 0),
    rows_pruned bigint NOT NULL DEFAULT 0 CHECK (rows_pruned >= 0),
    recall_threshold double precision CHECK (recall_threshold IS NULL OR (recall_threshold >= 0 AND recall_threshold <= 1)),
    recall_achieved double precision CHECK (recall_achieved IS NULL OR (recall_achieved >= 0 AND recall_achieved <= 1)),
    latency_bucket text NOT NULL DEFAULT 'Unspecified' CHECK (latency_bucket IN ('Lt1Ms', 'Lt10Ms', 'Lt100Ms', 'Lt1S', 'Gte1S', 'Unspecified')),
    lifecycle_state text NOT NULL DEFAULT 'Unspecified' CHECK (lifecycle_state IN ('Unspecified', 'Exact', 'Indexed', 'Fallback', 'IndexNotReady', 'IndexCorrupt', 'ArtifactMissing')),
    strategy text NOT NULL DEFAULT 'unspecified' CHECK (
        pg_catalog.octet_length(strategy) BETWEEN 1 AND 64
        AND strategy ~ '^[a-z0-9_]+$'
    ),
    visits bigint NOT NULL DEFAULT 0 CHECK (visits >= 0),
    filter_candidates bigint NOT NULL DEFAULT 0 CHECK (filter_candidates >= 0),
    candidates bigint NOT NULL DEFAULT 0 CHECK (candidates >= 0),
    rechecks bigint NOT NULL DEFAULT 0 CHECK (rechecks >= 0),
    stages bigint NOT NULL DEFAULT 0 CHECK (stages >= 0),
    expansions bigint NOT NULL DEFAULT 0 CHECK (expansions >= 0),
    adaptive_prefix_dimensions int4 CHECK (adaptive_prefix_dimensions > 0),
    adaptive_termination text CHECK (
        adaptive_termination IN (
            'exhaustive', 'empty_corpus', 'candidate_budget',
            'comparison_budget', 'recheck_budget', 'memory_budget',
            'expansion_budget'
        )
    ),
    completion text NOT NULL DEFAULT 'unspecified' CHECK (
        completion IN ('unspecified', 'complete', 'cancelled', 'budget_exhausted', 'error')
    ),
    latency_ms double precision NOT NULL CHECK (latency_ms >= 0),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now()
);

CREATE TABLE pgcontext._embedding_profiles (
    embedding_profile_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    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:]]'
    ),
    source_schema_name text NOT NULL CHECK (pg_catalog.btrim(source_schema_name) <> ''),
    source_table_name text NOT NULL CHECK (pg_catalog.btrim(source_table_name) <> ''),
    source_column_name text NOT NULL CHECK (pg_catalog.btrim(source_column_name) <> ''),
    source_attnum int2 NOT NULL CHECK (source_attnum > 0),
    source_type_name text NOT NULL CHECK (source_type_name IN ('vector', 'halfvec', 'sparsevec', 'bitvec', 'int8vec', 'uint8vec')),
    source_typmod int4 NOT NULL CHECK (source_typmod BETWEEN 1 AND 16000),
    source_version_column_name text,
    source_version_attnum int2,
    embedding_version_column_name text,
    embedding_version_attnum int2,
    hnsw_schema_name text NOT NULL CHECK (pg_catalog.btrim(hnsw_schema_name) <> ''),
    hnsw_index_name text NOT NULL CHECK (pg_catalog.btrim(hnsw_index_name) <> ''),
    hnsw_opclass text NOT NULL CHECK (pg_catalog.btrim(hnsw_opclass) <> ''),
    representation text NOT NULL CHECK (representation IN ('dense', 'half', 'sparse', 'bit', 'int8', 'uint8')),
    dimensions int4 NOT NULL CHECK (dimensions BETWEEN 1 AND 16000),
    normalization text NOT NULL CHECK (normalization IN ('none', 'unit_l2')),
    metric text NOT NULL CHECK (metric IN ('l2', 'inner_product', 'cosine', 'l1', 'hamming', 'jaccard')),
    provider text NOT NULL CHECK (pg_catalog.btrim(provider) <> ''),
    model text NOT NULL CHECK (pg_catalog.btrim(model) <> ''),
    revision text NOT NULL CHECK (pg_catalog.btrim(revision) <> ''),
    input_template text NOT NULL CHECK (pg_catalog.btrim(input_template) <> ''),
    output_template text NOT NULL CHECK (pg_catalog.btrim(output_template) <> ''),
    bit_order text CHECK (bit_order IN ('msb_first', 'lsb_first')),
    byte_order text CHECK (byte_order IN ('msb_first', 'lsb_first')),
    scale double precision CHECK (scale IS NULL OR (scale > 0 AND scale < 'Infinity'::double precision)),
    zero_point int4,
    configuration_hash text NOT NULL CHECK (configuration_hash ~ '^[0-9a-f]{16}$' AND configuration_hash <> '0000000000000000'),
    matryoshka_prefixes int4[],
    -- Only `active` and `draining` serve queries. A profile registers into
    -- `shadow` or `active`; every later move is a validated transition.
    lifecycle text NOT NULL DEFAULT 'active'
        CHECK (lifecycle IN ('shadow', 'active', 'draining', 'retired', 'failed')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, profile_name),
    -- A declared Matryoshka policy must certify 1..=8 prefixes below the full
    -- dimension, and only for a representation and metric whose coordinates
    -- stay interpretable at a cut point. Strict ascent is enforced by the typed
    -- `MatryoshkaPolicy` constructor: a CHECK cannot contain the subquery that
    -- a pairwise comparison would need, and adding an extension-owned function
    -- to a CHECK would entangle dump/restore ordering.
    CHECK (matryoshka_prefixes IS NULL OR (
        pg_catalog.cardinality(matryoshka_prefixes) BETWEEN 1 AND 8
        AND representation IN ('dense', 'half')
        AND metric IN ('l2', 'inner_product', 'cosine')
        AND matryoshka_prefixes[1] > 0
        AND matryoshka_prefixes[pg_catalog.cardinality(matryoshka_prefixes)] < dimensions
    )),
    UNIQUE (collection_id, source_column_name, revision),
    CHECK (source_type_name = CASE representation
        WHEN 'dense' THEN 'vector'
        WHEN 'half' THEN 'halfvec'
        WHEN 'sparse' THEN 'sparsevec'
        WHEN 'bit' THEN 'bitvec'
        WHEN 'int8' THEN 'int8vec'
        WHEN 'uint8' THEN 'uint8vec'
    END),
    CHECK (source_typmod = dimensions),
    CHECK (
        (source_version_column_name IS NULL
            AND source_version_attnum IS NULL
            AND embedding_version_column_name IS NULL
            AND embedding_version_attnum IS NULL)
        OR
        (pg_catalog.btrim(source_version_column_name) <> ''
            AND source_version_attnum > 0
            AND pg_catalog.btrim(embedding_version_column_name) <> ''
            AND embedding_version_attnum > 0
            AND source_version_column_name <> embedding_version_column_name
            AND source_version_attnum <> embedding_version_attnum)
    ),
    CHECK (
        (representation = 'bit'
            AND metric IN ('hamming', 'jaccard')
            AND bit_order IS NOT NULL
            AND byte_order IS NOT NULL
            AND scale IS NULL
            AND zero_point IS NULL)
        OR
        (representation <> 'bit'
            AND metric IN ('l2', 'inner_product', 'cosine', 'l1')
            AND bit_order IS NULL
            AND byte_order IS NULL
            AND (
                (representation = 'int8' AND (
                    (scale IS NULL AND zero_point IS NULL)
                    OR (scale IS NOT NULL AND zero_point IS NOT NULL AND zero_point BETWEEN -128 AND 127)
                ))
                OR
                (representation = 'uint8' AND (
                    (scale IS NULL AND zero_point IS NULL)
                    OR (scale IS NOT NULL AND zero_point IS NOT NULL AND zero_point BETWEEN 0 AND 255)
                ))
                OR
                (representation NOT IN ('int8', 'uint8') AND scale IS NULL AND zero_point IS NULL)
            ))
    )
);

-- An embedding profile's *contract* is immutable: representation, dimensions,
-- metric, templates, bindings, and hash describe a model that already produced
-- stored vectors, so changing one would silently reinterpret existing data.
-- `lifecycle` is different in kind -- it records whether that fixed contract is
-- currently serving -- so it is the one column an update may touch, and only
-- along a legal transition.
--
-- The contract comparison neutralizes `lifecycle` in both row images and then
-- compares the rows as text. Comparing whole rows rather than listing protected
-- columns means a column added later is protected by default; comparing text
-- rather than jsonb also catches changes jsonb normalizes away, such as an
-- array rewritten with different dimension bounds.
--
-- The transition table is duplicated from `context-core::ProfileLifecycle` on
-- purpose: the Rust table governs the SQL-facing API, and this one governs
-- direct catalog writes, so the invariant survives a superuser bypassing the
-- API.
CREATE FUNCTION pgcontext._reject_embedding_profile_mutation()
RETURNS trigger
LANGUAGE plpgsql
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    frozen_old pgcontext._embedding_profiles := OLD;
    frozen_new pgcontext._embedding_profiles := NEW;
BEGIN
    frozen_old.lifecycle := '';
    frozen_new.lifecycle := '';
    IF frozen_new::text IS DISTINCT FROM frozen_old::text THEN
        RAISE EXCEPTION USING
            ERRCODE = '55000',
            MESSAGE = 'embedding profiles are immutable; register a new profile revision';
    END IF;

    IF NEW.lifecycle = OLD.lifecycle THEN
        RAISE EXCEPTION USING
            ERRCODE = '22023',
            MESSAGE = pg_catalog.format(
                'embedding profile cannot move from %s to %s', OLD.lifecycle, NEW.lifecycle
            );
    END IF;

    IF NOT (
        (OLD.lifecycle = 'shadow' AND NEW.lifecycle IN ('active', 'failed', 'retired'))
        OR (OLD.lifecycle = 'active' AND NEW.lifecycle IN ('draining', 'failed'))
        OR (OLD.lifecycle = 'draining' AND NEW.lifecycle IN ('retired', 'active', 'failed'))
        OR (OLD.lifecycle = 'failed' AND NEW.lifecycle IN ('shadow', 'retired'))
    ) THEN
        RAISE EXCEPTION USING
            ERRCODE = '22023',
            MESSAGE = pg_catalog.format(
                'embedding profile cannot move from %s to %s', OLD.lifecycle, NEW.lifecycle
            );
    END IF;

    RETURN NEW;
END;
$$;

-- Catalog write for the one mutable column. Retrieval and registration run
-- SECURITY INVOKER, so a non-superuser collection member holds no privilege on
-- this private table; the write is performed as the extension owner after an
-- explicit owner-role check, exactly like the other catalog writers here.
CREATE FUNCTION pgcontext._set_embedding_profile_lifecycle(
    p_collection_id bigint,
    p_profile_name text,
    p_lifecycle text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    IF NOT EXISTS (
        SELECT 1
          FROM pgcontext._collections
         WHERE collection_id = p_collection_id
           AND pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER')
    ) THEN
        RAISE EXCEPTION 'permission denied for collection %', p_collection_id
            USING ERRCODE = '42501';
    END IF;

    UPDATE pgcontext._embedding_profiles
       SET lifecycle = p_lifecycle
     WHERE collection_id = p_collection_id
       AND profile_name = p_profile_name;
END;
$$;

CREATE TRIGGER embedding_profiles_immutable
BEFORE UPDATE ON pgcontext._embedding_profiles
FOR EACH ROW EXECUTE FUNCTION pgcontext._reject_embedding_profile_mutation();

CREATE TABLE pgcontext._embedding_migrations (
    migration_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    source_embedding_profile_id bigint NOT NULL REFERENCES pgcontext._embedding_profiles(embedding_profile_id),
    target_embedding_profile_id bigint NOT NULL REFERENCES pgcontext._embedding_profiles(embedding_profile_id),
    status text NOT NULL CHECK (status IN ('planned', 'running', 'completed', 'failed')),
    total_points bigint NOT NULL CHECK (total_points >= 0),
    processed_points bigint NOT NULL DEFAULT 0 CHECK (processed_points >= 0),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    CHECK (processed_points <= total_points),
    CHECK (source_embedding_profile_id <> target_embedding_profile_id)
);

CREATE TABLE pgcontext._build_jobs (
    build_job_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    artifact_kind text NOT NULL CHECK (
        artifact_kind IN ('index', 'segment', 'sparse_index', 'mmap', 'certification')
    ),
    artifact_name text NOT NULL CHECK (artifact_name <> ''),
    target_name text NOT NULL CHECK (target_name <> ''),
    job_kind text NOT NULL DEFAULT 'artifact_build' CHECK (
        job_kind IN ('artifact_build', 'compaction', 'projection_backfill', 'certification')
    ),
    status text NOT NULL CHECK (
        status IN ('planned', 'running', 'cancel_requested', 'cancelled', 'validating', 'publishing', 'completed', 'failed', 'abandoned')
    ),
    backend_pid int4,
    backend_identity text,
    attempt int4 NOT NULL DEFAULT 1 CHECK (attempt > 0),
    total_units bigint NOT NULL CHECK (total_units >= 0),
    processed_units bigint NOT NULL DEFAULT 0 CHECK (processed_units >= 0),
    last_source_point_id bigint NOT NULL DEFAULT 0 CHECK (last_source_point_id >= 0),
    source_high_water bigint NOT NULL DEFAULT 0 CHECK (source_high_water >= 0),
    config_revision bigint,
    source_version bigint CHECK (source_version IS NULL OR source_version > 0),
    lease_expires_at timestamptz,
    validation_passed boolean,
    validation_findings int4 CHECK (validation_findings IS NULL OR validation_findings >= 0),
    published_generation bigint CHECK (published_generation IS NULL OR published_generation > 0),
    supervised boolean NOT NULL DEFAULT false,
    cancel_requested boolean NOT NULL DEFAULT false,
    error_message text,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    completed_at timestamptz,
    CHECK (processed_units <= total_units),
    CHECK (NOT supervised OR job_kind IN ('certification', 'compaction'))
);

CREATE FUNCTION pgcontext._reject_build_job_progress_regression()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    IF NEW.processed_units < OLD.processed_units THEN
        RAISE EXCEPTION 'build job progress cannot go backwards: % < %',
            NEW.processed_units,
            OLD.processed_units
            USING ERRCODE = '22023';
    END IF;
    RETURN NEW;
END;
$$;

CREATE TRIGGER pgcontext_build_jobs_no_progress_regression
BEFORE UPDATE OF processed_units ON pgcontext._build_jobs
FOR EACH ROW
EXECUTE FUNCTION pgcontext._reject_build_job_progress_regression();

CREATE FUNCTION pgcontext._enforce_build_job_terminal_state()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    IF NEW.status IN ('running', 'cancel_requested', 'validating', 'publishing') THEN
        IF NEW.backend_pid IS NULL THEN
            RAISE EXCEPTION 'active build job % must record backend_pid',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
        IF NEW.backend_identity IS NULL THEN
            RAISE EXCEPTION 'active build job % must record backend_identity',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
        IF NEW.completed_at IS NOT NULL THEN
            RAISE EXCEPTION 'active build job % cannot retain completed_at',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
        IF NEW.lease_expires_at IS NULL THEN
            RAISE EXCEPTION 'active build job % must record lease_expires_at',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
    END IF;
    IF NEW.status IN ('completed', 'failed', 'cancelled', 'abandoned') THEN
        IF NEW.backend_pid IS NOT NULL OR NEW.backend_identity IS NOT NULL THEN
            RAISE EXCEPTION 'terminal build job % cannot retain backend ownership',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
        IF NEW.completed_at IS NULL THEN
            RAISE EXCEPTION 'terminal build job % must record completed_at',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
        IF NEW.lease_expires_at IS NOT NULL THEN
            RAISE EXCEPTION 'terminal build job % cannot retain a lease',
                NEW.status
                USING ERRCODE = '22023';
        END IF;
    END IF;
    IF NEW.status IN ('validating', 'publishing', 'completed') AND NEW.processed_units <> NEW.total_units THEN
        RAISE EXCEPTION 'completed build job progress must equal total: % <> %',
            NEW.processed_units,
            NEW.total_units
            USING ERRCODE = '22023';
    END IF;
    RETURN NEW;
END;
$$;

CREATE TRIGGER pgcontext_build_jobs_terminal_state
BEFORE INSERT OR UPDATE OF status, backend_pid, backend_identity, lease_expires_at, completed_at, processed_units, total_units
ON pgcontext._build_jobs
FOR EACH ROW
EXECUTE FUNCTION pgcontext._enforce_build_job_terminal_state();

CREATE UNIQUE INDEX pgcontext_build_jobs_one_active_target
    ON pgcontext._build_jobs (collection_id, artifact_kind, artifact_name, target_name)
 WHERE status IN ('planned', 'running', 'cancel_requested', 'validating', 'publishing');

CREATE TABLE pgcontext._build_deltas (
    delta_sequence bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    build_job_id bigint NOT NULL REFERENCES pgcontext._build_jobs(build_job_id) ON DELETE CASCADE,
    point_id bigint NOT NULL REFERENCES pgcontext._collection_points(point_id) ON DELETE CASCADE,
    operation text NOT NULL CHECK (operation IN ('upsert', 'delete')),
    source_version bigint NOT NULL CHECK (source_version > 0),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (build_job_id, source_version, point_id)
);

CREATE FUNCTION pgcontext._capture_build_point_delta()
RETURNS trigger
LANGUAGE plpgsql
AS $$
DECLARE
    delta_operation text;
    next_source_version bigint;
BEGIN
    INSERT INTO pgcontext._collection_source_revisions (
        collection_id, source_version, updated_at
    )
    VALUES (NEW.collection_id, 1, pg_catalog.now())
    ON CONFLICT (collection_id) DO UPDATE
       SET source_version = pgcontext._collection_source_revisions.source_version + 1,
           updated_at = pg_catalog.now()
    RETURNING source_version INTO STRICT next_source_version;
    delta_operation := CASE WHEN NEW.deleted_at IS NULL THEN 'upsert' ELSE 'delete' END;
    INSERT INTO pgcontext._build_deltas (
        build_job_id, point_id, operation, source_version
    )
    SELECT jobs.build_job_id, NEW.point_id, delta_operation, next_source_version
      FROM pgcontext._build_jobs AS jobs
     WHERE jobs.collection_id = NEW.collection_id
       AND (
           jobs.status IN ('running', 'cancel_requested', 'validating', 'publishing')
           OR (jobs.supervised AND jobs.status = 'planned')
       )
    ON CONFLICT (build_job_id, source_version, point_id) DO UPDATE
       SET operation = EXCLUDED.operation;
    RETURN NEW;
END;
$$;

CREATE TRIGGER pgcontext_capture_build_point_delta
AFTER INSERT OR UPDATE OF deleted_at ON pgcontext._collection_points
FOR EACH ROW
EXECUTE FUNCTION pgcontext._capture_build_point_delta();

CREATE TABLE pgcontext._generation_build_rows (
    build_job_id bigint NOT NULL REFERENCES pgcontext._build_jobs(build_job_id) ON DELETE CASCADE,
    point_id bigint NOT NULL,
    source_key text NOT NULL,
    PRIMARY KEY (build_job_id, point_id)
);

CREATE TABLE pgcontext._artifact_segments (
    artifact_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    build_job_id bigint NOT NULL REFERENCES pgcontext._build_jobs(build_job_id) ON DELETE RESTRICT,
    artifact_kind text NOT NULL CHECK (artifact_kind IN ('segment', 'mmap')),
    artifact_name text NOT NULL CHECK (artifact_name <> ''),
    target_name text NOT NULL CHECK (target_name <> ''),
    generation bigint NOT NULL CHECK (generation > 0),
    segment_kind text NOT NULL CHECK (segment_kind IN ('hnsw_graph')),
    format_version int4 NOT NULL CHECK (format_version = 2),
    payload_bytes bigint NOT NULL CHECK (payload_bytes >= 0),
    checksum bigint NOT NULL,
    config_revision bigint,
    relative_path text,
    lifecycle_state text NOT NULL DEFAULT 'validated' CHECK (lifecycle_state IN ('validated', 'file_materialized', 'rebuild_required', 'retired')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    CHECK (lifecycle_state <> 'file_materialized' OR relative_path IS NOT NULL),
    UNIQUE (collection_id, artifact_kind, artifact_name, target_name, generation)
);

CREATE FUNCTION pgcontext.current_vector_config_revision(collection bigint)
RETURNS bigint
LANGUAGE sql
STABLE
AS $$
    SELECT config_revision
      FROM pgcontext._collections
     WHERE collection_id = $1
$$;

CREATE TABLE pgcontext._artifact_reader_pins (
    artifact_id bigint NOT NULL REFERENCES pgcontext._artifact_segments(artifact_id) ON DELETE CASCADE,
    backend_pid int4 NOT NULL,
    backend_identity text NOT NULL,
    pin_count int4 NOT NULL CHECK (pin_count > 0),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    PRIMARY KEY (artifact_id, backend_pid, backend_identity)
);

CREATE TABLE pgcontext._generation_manifests (
    generation bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    build_job_id bigint NOT NULL UNIQUE REFERENCES pgcontext._build_jobs(build_job_id) ON DELETE RESTRICT,
    publication_alias text NOT NULL CHECK (publication_alias <> ''),
    source_version bigint NOT NULL CHECK (source_version > 0),
    config_revision bigint NOT NULL CHECK (config_revision > 0),
    lifecycle_state text NOT NULL DEFAULT 'staged' CHECK (
        lifecycle_state IN ('staged', 'validated', 'published', 'retiring', 'retired', 'rebuild_required')
    ),
    validation_passed boolean,
    validation_findings int4 CHECK (validation_findings IS NULL OR validation_findings >= 0),
    total_payload_bytes bigint NOT NULL DEFAULT 0 CHECK (total_payload_bytes >= 0),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    CHECK ((validation_passed IS NULL) = (validation_findings IS NULL)),
    CHECK (lifecycle_state NOT IN ('validated', 'published', 'retiring', 'retired') OR validation_passed)
);

CREATE TABLE pgcontext._generation_artifacts (
    generation bigint NOT NULL REFERENCES pgcontext._generation_manifests(generation) ON DELETE CASCADE,
    artifact_kind text NOT NULL CHECK (artifact_kind IN (
        'hnsw_segment', 'hnsw_delta', 'hnsw_directory', 'ivf_centroids',
        'ivf_postings', 'quantized_codes', 'chunk_projection',
        'vector_projection', 'graph_projection', 'community_summary',
        'certification_evidence'
    )),
    artifact_name text NOT NULL CHECK (artifact_name <> ''),
    payload_bytes bigint NOT NULL CHECK (payload_bytes >= 0),
    checksum_algorithm text NOT NULL DEFAULT 'pg_hashtextextended_hex_v1' CHECK (
        checksum_algorithm = 'pg_hashtextextended_hex_v1'
    ),
    checksum bigint NOT NULL,
    payload bytea NOT NULL,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    PRIMARY KEY (generation, artifact_kind, artifact_name),
    CONSTRAINT generation_artifact_payload_length_check CHECK (
        payload_bytes = pg_catalog.octet_length(payload)
    ),
    CONSTRAINT generation_artifact_payload_checksum_check CHECK (
        checksum_algorithm = 'pg_hashtextextended_hex_v1'
        AND checksum = pg_catalog.hashtextextended(pg_catalog.encode(payload, 'hex'), 0)
    )
);

CREATE TABLE pgcontext._generation_aliases (
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    publication_alias text NOT NULL CHECK (publication_alias <> ''),
    generation bigint NOT NULL UNIQUE REFERENCES pgcontext._generation_manifests(generation) ON DELETE RESTRICT,
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    PRIMARY KEY (collection_id, publication_alias)
);

CREATE TABLE pgcontext._generation_reader_pins (
    generation bigint NOT NULL REFERENCES pgcontext._generation_manifests(generation) ON DELETE CASCADE,
    backend_pid int4 NOT NULL,
    backend_identity text NOT NULL,
    pin_count int4 NOT NULL CHECK (pin_count > 0),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    PRIMARY KEY (generation, backend_pid, backend_identity)
);

CREATE FUNCTION pgcontext._publish_generation(target_generation bigint)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
    target_collection bigint;
    target_alias text;
    previous_generation bigint;
BEGIN
    SELECT collection_id, publication_alias
      INTO target_collection, target_alias
      FROM pgcontext._generation_manifests
     WHERE generation = target_generation
       AND lifecycle_state IN ('validated', 'published')
       AND validation_passed
     FOR UPDATE;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'generation % is not validated for publication', target_generation
            USING ERRCODE = '55000';
    END IF;
    IF NOT EXISTS (
        SELECT 1 FROM pgcontext._generation_artifacts WHERE generation = target_generation
    ) THEN
        RAISE EXCEPTION 'generation % has no artifact inventory', target_generation
            USING ERRCODE = '55000';
    END IF;

    -- A collection row always exists for a valid manifest and provides a
    -- transaction-scoped serialization point even when an alias has never
    -- been published before.
    PERFORM 1
      FROM pgcontext._collections
     WHERE collection_id = target_collection
     FOR UPDATE;

    SELECT generation
      INTO previous_generation
      FROM pgcontext._generation_aliases
     WHERE collection_id = target_collection
       AND publication_alias = target_alias
     FOR UPDATE;

    IF previous_generation IS NOT NULL AND previous_generation <> target_generation THEN
        DELETE FROM pgcontext._generation_reader_pins AS pins
         WHERE pins.generation = previous_generation
           AND NOT EXISTS (
               SELECT 1
                 FROM pg_catalog.pg_stat_activity AS activity
                WHERE activity.pid = pins.backend_pid
                  AND pg_catalog.to_char(
                          activity.backend_start AT TIME ZONE 'UTC',
                          'YYYY-MM-DD"T"HH24:MI:SS.US'
                      ) = pins.backend_identity
           );
        UPDATE pgcontext._generation_manifests
           SET lifecycle_state = CASE
                   WHEN EXISTS (
                       SELECT 1
                         FROM pgcontext._generation_reader_pins AS pins
                        WHERE pins.generation = previous_generation
                   ) THEN 'retiring'
                   ELSE 'retired'
               END,
               updated_at = pg_catalog.now()
         WHERE generation = previous_generation
           AND lifecycle_state = 'published';
    END IF;

    INSERT INTO pgcontext._generation_aliases (collection_id, publication_alias, generation)
    VALUES (target_collection, target_alias, target_generation)
    ON CONFLICT (collection_id, publication_alias) DO UPDATE
       SET generation = EXCLUDED.generation,
           updated_at = pg_catalog.now();

    UPDATE pgcontext._generation_manifests
       SET lifecycle_state = 'published', updated_at = pg_catalog.now()
     WHERE generation = target_generation;
END;
$$;

CREATE FUNCTION pgcontext._pin_generation(
    target_collection bigint,
    target_alias text
)
RETURNS bigint
LANGUAGE plpgsql
AS $$
DECLARE
    target_generation bigint;
    backend_identity_value text;
BEGIN
    SELECT aliases.generation
      INTO target_generation
      FROM pgcontext._generation_aliases AS aliases
      JOIN pgcontext._generation_manifests AS manifests USING (generation)
     WHERE aliases.collection_id = target_collection
       AND aliases.publication_alias = target_alias
       AND manifests.lifecycle_state = 'published'
     FOR SHARE OF aliases, manifests;
    IF NOT FOUND THEN
        RETURN NULL;
    END IF;
    SELECT pg_catalog.to_char(
               activity.backend_start AT TIME ZONE 'UTC',
               'YYYY-MM-DD"T"HH24:MI:SS.US'
           )
      INTO STRICT backend_identity_value
      FROM pg_catalog.pg_stat_activity AS activity
     WHERE activity.pid = pg_catalog.pg_backend_pid();
    INSERT INTO pgcontext._generation_reader_pins (
        generation, backend_pid, backend_identity, pin_count
    )
    VALUES (
        target_generation, pg_catalog.pg_backend_pid(), backend_identity_value, 1
    )
    ON CONFLICT (generation, backend_pid, backend_identity) DO UPDATE
       SET pin_count = pgcontext._generation_reader_pins.pin_count + 1,
           updated_at = pg_catalog.now();
    RETURN target_generation;
END;
$$;

CREATE FUNCTION pgcontext._unpin_generation(target_generation bigint)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
    backend_identity_value text;
BEGIN
    SELECT pg_catalog.to_char(
               activity.backend_start AT TIME ZONE 'UTC',
               'YYYY-MM-DD"T"HH24:MI:SS.US'
           )
      INTO STRICT backend_identity_value
      FROM pg_catalog.pg_stat_activity AS activity
     WHERE activity.pid = pg_catalog.pg_backend_pid();
    UPDATE pgcontext._generation_reader_pins
       SET pin_count = pin_count - 1,
           updated_at = pg_catalog.now()
     WHERE generation = target_generation
       AND backend_pid = pg_catalog.pg_backend_pid()
       AND backend_identity = backend_identity_value
       AND pin_count > 1;
    IF NOT FOUND THEN
        DELETE FROM pgcontext._generation_reader_pins
         WHERE generation = target_generation
           AND backend_pid = pg_catalog.pg_backend_pid()
           AND backend_identity = backend_identity_value;
        IF NOT FOUND THEN
            RAISE EXCEPTION 'generation % is not pinned by this backend', target_generation
                USING ERRCODE = '55000';
        END IF;
    END IF;
    UPDATE pgcontext._generation_manifests AS manifests
       SET lifecycle_state = 'retired', updated_at = pg_catalog.now()
     WHERE manifests.generation = target_generation
       AND manifests.lifecycle_state = 'retiring'
       AND NOT EXISTS (
           SELECT 1
             FROM pgcontext._generation_reader_pins AS pins
            WHERE pins.generation = manifests.generation
       );
END;
$$;

-- Intentionally unfiltered pre-authorization lookup. Callers need this minimal
-- collection/alias name, owner-role, and source-binding summary to resolve a
-- collection before checking membership. It must not expose source-table
-- identity, source keys, vectors, payload metadata, or operational state.
CREATE VIEW pgcontext._collection_acl AS
SELECT collection_id,
       collection_name,
       owner_role,
       source_table_oid IS NOT NULL AS has_source_table
  FROM pgcontext._collections
UNION ALL
SELECT collections.collection_id,
       aliases.alias_name AS collection_name,
       collections.owner_role,
       collections.source_table_oid IS NOT NULL AS has_source_table
  FROM pgcontext._collection_aliases AS aliases
  JOIN pgcontext._collections AS collections USING (collection_id);

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

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

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

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

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

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

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

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

CREATE VIEW pgcontext._visible_collection_limits
WITH (security_barrier = true) AS
SELECT collection_id,
       strict_mode,
       max_dimensions,
       max_vectors,
       max_points,
       max_filter_nodes,
       max_search_limit,
       max_candidate_budget,
       query_timeout_ms,
       max_index_memory_bytes
  FROM pgcontext._collections
 WHERE pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER');

CREATE VIEW pgcontext._visible_collections
WITH (security_barrier = true) AS
SELECT collection_id,
       collection_name,
       owner_role,
       source_table_oid,
       source_schema_name,
       source_table_name
  FROM pgcontext._collections
 WHERE pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER');

GRANT SELECT ON pgcontext._collection_acl TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_vectors TO PUBLIC;
GRANT SELECT ON pgcontext._visible_embedding_profiles TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_sparse_vectors TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_points TO PUBLIC;
GRANT SELECT ON pgcontext._visible_query_stats TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_payload_columns TO PUBLIC;
GRANT SELECT ON pgcontext._visible_build_jobs TO PUBLIC;
GRANT SELECT ON pgcontext._visible_artifact_segments TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_limits TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collections TO PUBLIC;

-- Drift self-healing writes. Retrieval functions run SECURITY INVOKER so that
-- source-table SELECT/RLS is enforced against the caller, which means a
-- non-superuser collection member holds no direct privilege on the private
-- catalog tables. When a dump/restore or table rewrite changes a source
-- table's oid, the query path detects the drift and refreshes the stored
-- metadata through these SECURITY DEFINER helpers so the write succeeds as the
-- extension owner.
--
-- Security: these helpers accept only identifying keys, never a caller-supplied
-- oid or attnum. Each re-derives the authoritative oid/attnum from pg_catalog
-- using the schema, table, and column names already stored in the private
-- catalog, so a call can only ever set the true current binding of the
-- already-registered source table. This is deliberate: accepting an oid would
-- let a collection member repoint _collections.source_table_oid at an unrelated
-- table they control, and the SECURITY DEFINER payload/backfill paths check
-- privileges by that oid while mutating the real table by name -- a
-- confused-deputy escalation. Each helper also re-checks owner-role membership
-- on SESSION_USER (defence in depth for direct calls) and pins search_path.
CREATE FUNCTION pgcontext._refresh_collection_source_table(
    p_collection_id bigint
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    IF NOT EXISTS (
        SELECT 1
          FROM pgcontext._collections
         WHERE collection_id = p_collection_id
           AND pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER')
    ) THEN
        RAISE EXCEPTION 'permission denied to refresh collection %', p_collection_id
            USING ERRCODE = '42501';
    END IF;

    UPDATE pgcontext._collections AS collections
       SET source_table_oid = source_class.oid,
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS source_class
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.oid = source_class.relnamespace
     WHERE collections.collection_id = p_collection_id
       AND source_namespace.nspname = collections.source_schema_name
       AND source_class.relname = collections.source_table_name
       AND source_class.relkind IN ('r', 'p');
END;
$$;

CREATE FUNCTION pgcontext._refresh_vector_source_binding(
    p_collection_id bigint,
    p_vector_column_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    IF NOT EXISTS (
        SELECT 1
          FROM pgcontext._collections
         WHERE collection_id = p_collection_id
           AND pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER')
    ) THEN
        RAISE EXCEPTION 'permission denied to refresh collection %', p_collection_id
            USING ERRCODE = '42501';
    END IF;

    UPDATE pgcontext._collection_vectors AS vectors
       SET source_table_oid = source_class.oid,
           vector_attnum = source_attribute.attnum,
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS source_class
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.oid = source_class.relnamespace
      JOIN pg_catalog.pg_attribute AS source_attribute
        ON source_attribute.attrelid = source_class.oid
     WHERE vectors.collection_id = p_collection_id
       AND vectors.vector_column_name = p_vector_column_name
       AND source_namespace.nspname = vectors.source_schema_name
       AND source_class.relname = vectors.source_table_name
       AND source_class.relkind IN ('r', 'p')
       AND source_attribute.attname = vectors.vector_column_name
       AND source_attribute.attnum > 0
       AND NOT source_attribute.attisdropped;
END;
$$;

CREATE FUNCTION pgcontext._refresh_sparse_vector_source_binding(
    p_collection_id bigint,
    p_vector_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    IF NOT EXISTS (
        SELECT 1
          FROM pgcontext._collections
         WHERE collection_id = p_collection_id
           AND pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER')
    ) THEN
        RAISE EXCEPTION 'permission denied to refresh collection %', p_collection_id
            USING ERRCODE = '42501';
    END IF;

    UPDATE pgcontext._collection_sparse_vectors AS sparse_vectors
       SET source_table_oid = source_class.oid,
           vector_attnum = source_attribute.attnum,
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS source_class
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.oid = source_class.relnamespace
      JOIN pg_catalog.pg_attribute AS source_attribute
        ON source_attribute.attrelid = source_class.oid
     WHERE sparse_vectors.collection_id = p_collection_id
       AND sparse_vectors.vector_name = p_vector_name
       AND source_namespace.nspname = sparse_vectors.source_schema_name
       AND source_class.relname = sparse_vectors.source_table_name
       AND source_class.relkind IN ('r', 'p')
       AND source_attribute.attname = sparse_vectors.vector_column_name
       AND source_attribute.attnum > 0
       AND NOT source_attribute.attisdropped;
END;
$$;

CREATE FUNCTION pgcontext._refresh_payload_source_bindings(
    p_collection_id bigint
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    IF NOT EXISTS (
        SELECT 1
          FROM pgcontext._collections
         WHERE collection_id = p_collection_id
           AND pg_catalog.pg_has_role(SESSION_USER, owner_role, 'MEMBER')
    ) THEN
        RAISE EXCEPTION 'permission denied to refresh collection %', p_collection_id
            USING ERRCODE = '42501';
    END IF;

    UPDATE pgcontext._collection_payload_columns AS payload_columns
       SET source_table_oid = source_class.oid,
           column_attnum = source_attribute.attnum,
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS source_class
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.oid = source_class.relnamespace
      JOIN pg_catalog.pg_attribute AS source_attribute
        ON source_attribute.attrelid = source_class.oid
     WHERE payload_columns.collection_id = p_collection_id
       AND source_namespace.nspname = payload_columns.source_schema_name
       AND source_class.relname = payload_columns.source_table_name
       AND source_class.relkind IN ('r', 'p')
       AND source_attribute.attname = payload_columns.column_name
       AND source_attribute.attnum > 0
       AND NOT source_attribute.attisdropped;
END;
$$;

SELECT pg_catalog.pg_extension_config_dump('pgcontext._collections', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._collection_vectors', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._collection_sparse_vectors', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._collection_points', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._collection_aliases', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._collection_payload_columns', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._query_stats', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._embedding_profiles', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._embedding_migrations', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._build_jobs', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._build_deltas', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._artifact_segments', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._generation_manifests', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._generation_artifacts', '');
SELECT pg_catalog.pg_extension_config_dump('pgcontext._generation_aliases', '');
"#,
    name = "create_catalog_tables",
    requires = ["pgcontext_bootstrap"]
);
