//! 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_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),
    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._model_versions (
    model_version_id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    collection_id bigint NOT NULL REFERENCES pgcontext._collections(collection_id) ON DELETE CASCADE,
    model_name text NOT NULL CHECK (model_name <> ''),
    model_version text NOT NULL CHECK (model_version <> ''),
    dimensions int4 NOT NULL CHECK (dimensions > 0),
    metric text NOT NULL CHECK (metric IN ('l2', 'inner_product', 'cosine', 'l1', 'hamming', 'jaccard')),
    is_active boolean NOT NULL DEFAULT true,
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, model_name, model_version)
);

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_model_version_id bigint NOT NULL REFERENCES pgcontext._model_versions(model_version_id),
    target_model_version_id bigint NOT NULL REFERENCES pgcontext._model_versions(model_version_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_model_version_id <> target_model_version_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')),
    artifact_name text NOT NULL CHECK (artifact_name <> ''),
    target_name text NOT NULL CHECK (target_name <> ''),
    status text NOT NULL CHECK (
        status IN ('planned', 'running', 'cancel_requested', 'cancelled', '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),
    config_revision bigint,
    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)
);

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') 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;
    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;
    END IF;
    IF NEW.status = '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, 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');

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')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (build_job_id, point_id, delta_sequence)
);

CREATE FUNCTION pgcontext._capture_build_point_delta()
RETURNS trigger
LANGUAGE plpgsql
AS $$
DECLARE
    delta_operation text;
BEGIN
    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)
    SELECT jobs.build_job_id, NEW.point_id, delta_operation
      FROM pgcontext._build_jobs AS jobs
     WHERE jobs.collection_id = NEW.collection_id
       AND jobs.status IN ('running', 'cancel_requested');
    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._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 = 1),
    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)
);

-- 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_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_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._model_versions', '');
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', '');
"#,
    name = "create_catalog_tables",
    requires = ["pgcontext_bootstrap"]
);
