//! SQL schema objects for registered PostgreSQL-native lexical and fuzzy sources.
//!
//! The base tables are private. Membership-filtered security-barrier views
//! expose registration metadata to collection members, and every identity
//! column stores both a stable name and the resolved OID so a dump/restore
//! cycle can refresh OIDs without trusting them.

pgrx::extension_sql!(
    r#"
CREATE TABLE pgcontext._collection_lexical_sources (
    lexical_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,
    source_table_oid oid NOT NULL,
    source_schema_name text NOT NULL,
    source_table_name text NOT NULL,
    document_mode text NOT NULL
        CHECK (document_mode IN ('fields', 'stored_vector')),
    stored_vector_column_name text,
    stored_vector_attnum int2,
    configuration_oid oid NOT NULL,
    configuration_schema_name text NOT NULL,
    configuration_name text NOT NULL,
    ranker text NOT NULL CHECK (ranker IN ('ts_rank', 'ts_rank_cd')),
    normalization int4 NOT NULL DEFAULT 0
        CHECK (normalization BETWEEN 0 AND 63),
    rank_weight_d real NOT NULL DEFAULT 0.1
        CHECK (rank_weight_d BETWEEN 0.0 AND 1.0),
    rank_weight_c real NOT NULL DEFAULT 0.2
        CHECK (rank_weight_c BETWEEN 0.0 AND 1.0),
    rank_weight_b real NOT NULL DEFAULT 0.4
        CHECK (rank_weight_b BETWEEN 0.0 AND 1.0),
    rank_weight_a real NOT NULL DEFAULT 1.0
        CHECK (rank_weight_a BETWEEN 0.0 AND 1.0),
    row_tsquery_name text,
    row_tsquery_column_name text,
    row_tsquery_attnum int2,
    index_oid oid,
    index_name text,
    index_am_name text CHECK (index_am_name IN ('gin', 'gist')),
    index_definition text,
    index_is_lossy boolean NOT NULL DEFAULT false,
    registration_revision bigint NOT NULL DEFAULT 1
        CHECK (registration_revision > 0),
    status text NOT NULL DEFAULT 'ready'
        CHECK (status IN ('ready', 'stale', 'failed')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, source_name),
    CHECK ((document_mode = 'stored_vector') = (stored_vector_column_name IS NOT NULL)),
    CHECK ((stored_vector_column_name IS NULL) = (stored_vector_attnum IS NULL)),
    CHECK ((row_tsquery_name IS NULL) = (row_tsquery_column_name IS NULL)),
    CHECK ((row_tsquery_column_name IS NULL) = (row_tsquery_attnum IS NULL)),
    CHECK ((index_oid IS NULL) = (index_name IS NULL)),
    CHECK ((index_oid IS NULL) = (index_am_name IS NULL)),
    CHECK ((index_oid IS NULL) = (index_definition IS NULL))
);

CREATE TABLE pgcontext._collection_lexical_fields (
    lexical_source_id bigint NOT NULL
        REFERENCES pgcontext._collection_lexical_sources(lexical_source_id)
        ON DELETE CASCADE,
    field_ordinal int4 NOT NULL CHECK (field_ordinal > 0),
    column_name text NOT NULL,
    column_attnum int2 NOT NULL,
    column_type_oid oid NOT NULL,
    column_collation_oid oid NOT NULL DEFAULT 0,
    json_path text[],
    weight text NOT NULL DEFAULT 'D' CHECK (weight IN ('A', 'B', 'C', 'D')),
    PRIMARY KEY (lexical_source_id, field_ordinal),
    CHECK (json_path IS NULL OR pg_catalog.cardinality(json_path) BETWEEN 1 AND 16)
);

CREATE TABLE pgcontext._collection_fuzzy_sources (
    fuzzy_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,
    source_table_oid oid NOT NULL,
    source_schema_name text NOT NULL,
    source_table_name text NOT NULL,
    text_column_name text NOT NULL,
    text_attnum int2 NOT NULL,
    text_type_oid oid NOT NULL,
    text_collation_oid oid NOT NULL DEFAULT 0,
    trgm_extension_oid oid NOT NULL,
    trgm_schema_name text NOT NULL,
    index_oid oid,
    index_name text,
    index_am_name text CHECK (index_am_name IN ('gin', 'gist')),
    index_definition text,
    registration_revision bigint NOT NULL DEFAULT 1
        CHECK (registration_revision > 0),
    status text NOT NULL DEFAULT 'ready'
        CHECK (status IN ('ready', 'stale', 'failed')),
    created_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(),
    UNIQUE (collection_id, source_name),
    CHECK ((index_oid IS NULL) = (index_name IS NULL)),
    CHECK ((index_oid IS NULL) = (index_am_name IS NULL)),
    CHECK ((index_oid IS NULL) = (index_definition IS NULL))
);

REVOKE ALL ON TABLE pgcontext._collection_lexical_sources FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._collection_lexical_fields FROM PUBLIC;
REVOKE ALL ON TABLE pgcontext._collection_fuzzy_sources FROM PUBLIC;

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

CREATE VIEW pgcontext._visible_collection_lexical_fields
WITH (security_barrier = true) AS
SELECT fields.*
  FROM pgcontext._collection_lexical_fields AS fields
  JOIN pgcontext._collection_lexical_sources AS sources USING (lexical_source_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_collection_fuzzy_sources
WITH (security_barrier = true) AS
SELECT sources.*
  FROM pgcontext._collection_fuzzy_sources AS sources
  JOIN pgcontext._collections AS collections USING (collection_id)
 WHERE pg_catalog.pg_has_role(SESSION_USER, collections.owner_role, 'MEMBER');

GRANT SELECT ON pgcontext._visible_collection_lexical_sources TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_lexical_fields TO PUBLIC;
GRANT SELECT ON pgcontext._visible_collection_fuzzy_sources TO PUBLIC;

-- Catalog writes. The SQL-facing lexical functions run SECURITY INVOKER so the
-- caller's source-table SELECT, RLS, and index-creation privileges apply. These
-- helpers perform the private-catalog write as the extension owner.
--
-- Security: no helper accepts a caller-supplied oid or attnum. Every oid is
-- re-derived from pg_catalog using the names the caller already proved access
-- to, each helper re-checks owner-role membership on SESSION_USER, and each
-- pins search_path.
CREATE FUNCTION pgcontext._require_collection_owner(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 for collection %', p_collection_id
            USING ERRCODE = '42501';
    END IF;
END;
$$;

CREATE FUNCTION pgcontext._register_lexical_source(
    p_collection_id bigint,
    p_source_name text,
    p_document_mode text,
    p_stored_vector_column text,
    p_configuration_schema text,
    p_configuration_name text,
    p_ranker text,
    p_normalization int4,
    p_rank_weights real[],
    p_columns text[],
    p_weights text[],
    p_json_paths text[]
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_schema text;
    source_table text;
    source_oid oid;
    configuration_oid oid;
    stored_attnum int2;
    new_source_id bigint;
    ordinal int4;
    column_name text;
    path_text text;
    path_components text[];
    attribute_row record;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);

    SELECT collections.source_schema_name, collections.source_table_name, source_class.oid
      INTO source_schema, source_table, source_oid
      FROM pgcontext._collections AS collections
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.nspname = collections.source_schema_name
      JOIN pg_catalog.pg_class AS source_class
        ON source_class.relnamespace = source_namespace.oid
       AND source_class.relname = collections.source_table_name
       AND source_class.relkind IN ('r', 'p')
     WHERE collections.collection_id = p_collection_id;
    IF source_oid IS NULL THEN
        RAISE EXCEPTION 'collection % has no resolvable source table', p_collection_id
            USING ERRCODE = '42P01';
    END IF;
    IF NOT pg_catalog.has_table_privilege(SESSION_USER, source_oid, 'SELECT') THEN
        RAISE EXCEPTION 'permission denied for source table %.%', source_schema, source_table
            USING ERRCODE = '42501';
    END IF;

    SELECT configuration.oid
      INTO configuration_oid
      FROM pg_catalog.pg_ts_config AS configuration
      JOIN pg_catalog.pg_namespace AS namespace
        ON namespace.oid = configuration.cfgnamespace
     WHERE namespace.nspname = p_configuration_schema
       AND configuration.cfgname = p_configuration_name;
    IF configuration_oid IS NULL THEN
        RAISE EXCEPTION 'text search configuration does not exist: %.%',
            p_configuration_schema, p_configuration_name
            USING ERRCODE = '42704';
    END IF;

    IF p_document_mode = 'stored_vector' THEN
        SELECT attribute.attnum
          INTO stored_attnum
          FROM pg_catalog.pg_attribute AS attribute
         WHERE attribute.attrelid = source_oid
           AND attribute.attname = p_stored_vector_column
           AND attribute.attnum > 0
           AND NOT attribute.attisdropped
           AND pg_catalog.format_type(attribute.atttypid, NULL) = 'tsvector';
        IF stored_attnum IS NULL THEN
            RAISE EXCEPTION 'stored lexical document column is missing or is not tsvector: %',
                p_stored_vector_column
                USING ERRCODE = '42703';
        END IF;
    END IF;

    INSERT INTO pgcontext._collection_lexical_sources (
        collection_id, source_name, source_table_oid, source_schema_name,
        source_table_name, document_mode, stored_vector_column_name,
        stored_vector_attnum, configuration_oid, configuration_schema_name,
        configuration_name, ranker, normalization,
        rank_weight_d, rank_weight_c, rank_weight_b, rank_weight_a
    )
    VALUES (
        p_collection_id, p_source_name, source_oid, source_schema,
        source_table, p_document_mode,
        CASE WHEN p_document_mode = 'stored_vector' THEN p_stored_vector_column END,
        stored_attnum, configuration_oid, p_configuration_schema,
        p_configuration_name, p_ranker, p_normalization,
        p_rank_weights[1], p_rank_weights[2], p_rank_weights[3], p_rank_weights[4]
    )
    RETURNING lexical_source_id INTO new_source_id;

    IF p_document_mode = 'fields' THEN
        FOR ordinal IN 1 .. pg_catalog.cardinality(p_columns) LOOP
            column_name := p_columns[ordinal];
            SELECT attribute.attnum, attribute.atttypid, attribute.attcollation
              INTO attribute_row
              FROM pg_catalog.pg_attribute AS attribute
             WHERE attribute.attrelid = source_oid
               AND attribute.attname = column_name
               AND attribute.attnum > 0
               AND NOT attribute.attisdropped;
            IF NOT FOUND THEN
                RAISE EXCEPTION 'lexical field column is missing: %', column_name
                    USING ERRCODE = '42703';
            END IF;
            path_text := p_json_paths[ordinal];
            IF path_text IS NULL OR path_text = '' THEN
                path_components := NULL;
            ELSE
                path_components := pg_catalog.string_to_array(path_text, '.');
                IF pg_catalog.format_type(attribute_row.atttypid, NULL)
                       NOT IN ('json', 'jsonb') THEN
                    RAISE EXCEPTION 'lexical JSON path requires a json or jsonb column: %',
                        column_name
                        USING ERRCODE = '42804';
                END IF;
            END IF;
            INSERT INTO pgcontext._collection_lexical_fields (
                lexical_source_id, field_ordinal, column_name, column_attnum,
                column_type_oid, column_collation_oid, json_path, weight
            )
            VALUES (
                new_source_id, ordinal, column_name, attribute_row.attnum,
                attribute_row.atttypid, attribute_row.attcollation,
                path_components, pg_catalog.upper(p_weights[ordinal])
            );
        END LOOP;
    END IF;

    RETURN new_source_id;
END;
$$;

CREATE FUNCTION pgcontext._register_lexical_tsquery(
    p_collection_id bigint,
    p_source_name text,
    p_tsquery_name text,
    p_column_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_oid oid;
    column_attnum int2;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    SELECT source_table_oid INTO source_oid
      FROM pgcontext._collection_lexical_sources
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
    IF source_oid IS NULL THEN
        RAISE EXCEPTION 'lexical source is not registered: %', p_source_name
            USING ERRCODE = '42704';
    END IF;
    SELECT attribute.attnum INTO column_attnum
      FROM pg_catalog.pg_attribute AS attribute
     WHERE attribute.attrelid = source_oid
       AND attribute.attname = p_column_name
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped
       AND pg_catalog.format_type(attribute.atttypid, NULL) = 'tsquery';
    IF column_attnum IS NULL THEN
        RAISE EXCEPTION 'registered row tsquery column is missing or is not tsquery: %',
            p_column_name
            USING ERRCODE = '42703';
    END IF;
    UPDATE pgcontext._collection_lexical_sources
       SET row_tsquery_name = p_tsquery_name,
           row_tsquery_column_name = p_column_name,
           row_tsquery_attnum = column_attnum,
           registration_revision = registration_revision + 1,
           updated_at = pg_catalog.now()
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
END;
$$;

CREATE FUNCTION pgcontext._attach_lexical_index(
    p_collection_id bigint,
    p_source_name text,
    p_index_schema text,
    p_index_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_oid oid;
    index_row record;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    SELECT source_table_oid INTO source_oid
      FROM pgcontext._collection_lexical_sources
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
    IF source_oid IS NULL THEN
        RAISE EXCEPTION 'lexical source is not registered: %', p_source_name
            USING ERRCODE = '42704';
    END IF;

    SELECT index_class.oid AS index_oid,
           access_method.amname AS amname,
           index.indisvalid AS indisvalid,
           index.indislive AS indislive,
           index.indpred IS NULL AS is_full,
           index.indrelid AS indrelid,
           pg_catalog.pg_get_indexdef(index.indexrelid) AS definition
      INTO index_row
      FROM pg_catalog.pg_class AS index_class
      JOIN pg_catalog.pg_namespace AS index_namespace
        ON index_namespace.oid = index_class.relnamespace
      JOIN pg_catalog.pg_index AS index
        ON index.indexrelid = index_class.oid
      JOIN pg_catalog.pg_am AS access_method
        ON access_method.oid = index_class.relam
     WHERE index_namespace.nspname = p_index_schema
       AND index_class.relname = p_index_name;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'index does not exist: %.%', p_index_schema, p_index_name
            USING ERRCODE = '42704';
    END IF;
    IF index_row.indrelid <> source_oid THEN
        RAISE EXCEPTION 'index %.% is not defined on the registered source relation',
            p_index_schema, p_index_name
            USING ERRCODE = '42809';
    END IF;
    IF index_row.amname NOT IN ('gin', 'gist') THEN
        RAISE EXCEPTION 'lexical index access method must be gin or gist, found %',
            index_row.amname
            USING ERRCODE = '0A000';
    END IF;
    IF NOT index_row.is_full THEN
        RAISE EXCEPTION 'lexical index must not be partial: %.%', p_index_schema, p_index_name
            USING ERRCODE = '0A000';
    END IF;
    IF NOT index_row.indisvalid OR NOT index_row.indislive THEN
        RAISE EXCEPTION 'lexical index is not valid and live: %.%', p_index_schema, p_index_name
            USING ERRCODE = '55000';
    END IF;

    UPDATE pgcontext._collection_lexical_sources
       SET index_oid = index_row.index_oid,
           index_name = p_index_name,
           index_am_name = index_row.amname,
           index_definition = index_row.definition,
           index_is_lossy = (index_row.amname = 'gist'),
           registration_revision = registration_revision + 1,
           updated_at = pg_catalog.now()
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
END;
$$;

CREATE FUNCTION pgcontext._detach_lexical_index(
    p_collection_id bigint,
    p_source_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    UPDATE pgcontext._collection_lexical_sources
       SET index_oid = NULL,
           index_name = NULL,
           index_am_name = NULL,
           index_definition = NULL,
           index_is_lossy = false,
           registration_revision = registration_revision + 1,
           updated_at = pg_catalog.now()
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
END;
$$;

CREATE FUNCTION pgcontext._drop_lexical_source(
    p_collection_id bigint,
    p_source_name text
)
RETURNS boolean
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    removed bigint;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    DELETE FROM pgcontext._collection_lexical_sources
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
    GET DIAGNOSTICS removed = ROW_COUNT;
    RETURN removed > 0;
END;
$$;

CREATE FUNCTION pgcontext._register_fuzzy_source(
    p_collection_id bigint,
    p_source_name text,
    p_column_name text
)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_schema text;
    source_table text;
    source_oid oid;
    attribute_row record;
    extension_row record;
    new_source_id bigint;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);

    SELECT collections.source_schema_name, collections.source_table_name, source_class.oid
      INTO source_schema, source_table, source_oid
      FROM pgcontext._collections AS collections
      JOIN pg_catalog.pg_namespace AS source_namespace
        ON source_namespace.nspname = collections.source_schema_name
      JOIN pg_catalog.pg_class AS source_class
        ON source_class.relnamespace = source_namespace.oid
       AND source_class.relname = collections.source_table_name
       AND source_class.relkind IN ('r', 'p')
     WHERE collections.collection_id = p_collection_id;
    IF source_oid IS NULL THEN
        RAISE EXCEPTION 'collection % has no resolvable source table', p_collection_id
            USING ERRCODE = '42P01';
    END IF;
    IF NOT pg_catalog.has_table_privilege(SESSION_USER, source_oid, 'SELECT') THEN
        RAISE EXCEPTION 'permission denied for source table %.%', source_schema, source_table
            USING ERRCODE = '42501';
    END IF;

    SELECT extension.oid AS extension_oid, namespace.nspname AS schema_name
      INTO extension_row
      FROM pg_catalog.pg_extension AS extension
      JOIN pg_catalog.pg_namespace AS namespace
        ON namespace.oid = extension.extnamespace
     WHERE extension.extname = 'pg_trgm';
    IF NOT FOUND THEN
        RAISE EXCEPTION 'fuzzy sources require the pg_trgm extension'
            USING ERRCODE = '42704';
    END IF;

    SELECT attribute.attnum, attribute.atttypid, attribute.attcollation
      INTO attribute_row
      FROM pg_catalog.pg_attribute AS attribute
     WHERE attribute.attrelid = source_oid
       AND attribute.attname = p_column_name
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'fuzzy source text column is missing: %', p_column_name
            USING ERRCODE = '42703';
    END IF;
    IF attribute_row.atttypid <> 'pg_catalog.text'::pg_catalog.regtype THEN
        RAISE EXCEPTION 'fuzzy source column must be text: %', p_column_name
            USING ERRCODE = '42804';
    END IF;

    INSERT INTO pgcontext._collection_fuzzy_sources (
        collection_id, source_name, source_table_oid, source_schema_name,
        source_table_name, text_column_name, text_attnum, text_type_oid,
        text_collation_oid,
        trgm_extension_oid, trgm_schema_name
    )
    VALUES (
        p_collection_id, p_source_name, source_oid, source_schema, source_table,
        p_column_name, attribute_row.attnum, attribute_row.atttypid,
        attribute_row.attcollation,
        extension_row.extension_oid, extension_row.schema_name
    )
    RETURNING fuzzy_source_id INTO new_source_id;
    RETURN new_source_id;
END;
$$;

CREATE FUNCTION pgcontext._attach_fuzzy_index(
    p_collection_id bigint,
    p_source_name text,
    p_index_schema text,
    p_index_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    source_oid oid;
    index_row record;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    SELECT source_table_oid INTO source_oid
      FROM pgcontext._collection_fuzzy_sources
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
    IF source_oid IS NULL THEN
        RAISE EXCEPTION 'fuzzy source is not registered: %', p_source_name
            USING ERRCODE = '42704';
    END IF;

    SELECT index_class.oid AS index_oid,
           access_method.amname AS amname,
           index.indisvalid AS indisvalid,
           index.indislive AS indislive,
           index.indpred IS NULL AS is_full,
           index.indrelid AS indrelid,
           pg_catalog.pg_get_indexdef(index.indexrelid) AS definition
      INTO index_row
      FROM pg_catalog.pg_class AS index_class
      JOIN pg_catalog.pg_namespace AS index_namespace
        ON index_namespace.oid = index_class.relnamespace
      JOIN pg_catalog.pg_index AS index
        ON index.indexrelid = index_class.oid
      JOIN pg_catalog.pg_am AS access_method
        ON access_method.oid = index_class.relam
     WHERE index_namespace.nspname = p_index_schema
       AND index_class.relname = p_index_name;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'index does not exist: %.%', p_index_schema, p_index_name
            USING ERRCODE = '42704';
    END IF;
    IF index_row.indrelid <> source_oid THEN
        RAISE EXCEPTION 'index %.% is not defined on the registered source relation',
            p_index_schema, p_index_name
            USING ERRCODE = '42809';
    END IF;
    IF index_row.amname NOT IN ('gin', 'gist') THEN
        RAISE EXCEPTION 'fuzzy index access method must be gin or gist, found %',
            index_row.amname
            USING ERRCODE = '0A000';
    END IF;
    IF NOT index_row.is_full THEN
        RAISE EXCEPTION 'fuzzy index must not be partial: %.%', p_index_schema, p_index_name
            USING ERRCODE = '0A000';
    END IF;
    IF NOT index_row.indisvalid OR NOT index_row.indislive THEN
        RAISE EXCEPTION 'fuzzy index is not valid and live: %.%', p_index_schema, p_index_name
            USING ERRCODE = '55000';
    END IF;

    UPDATE pgcontext._collection_fuzzy_sources
       SET index_oid = index_row.index_oid,
           index_name = p_index_name,
           index_am_name = index_row.amname,
           index_definition = index_row.definition,
           registration_revision = registration_revision + 1,
           updated_at = pg_catalog.now()
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
END;
$$;

CREATE FUNCTION pgcontext._detach_fuzzy_index(
    p_collection_id bigint,
    p_source_name text
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    UPDATE pgcontext._collection_fuzzy_sources
       SET index_oid = NULL,
           index_name = NULL,
           index_am_name = NULL,
           index_definition = NULL,
           registration_revision = registration_revision + 1,
           updated_at = pg_catalog.now()
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
END;
$$;

CREATE FUNCTION pgcontext._drop_fuzzy_source(
    p_collection_id bigint,
    p_source_name text
)
RETURNS boolean
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    removed bigint;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);
    DELETE FROM pgcontext._collection_fuzzy_sources
     WHERE collection_id = p_collection_id AND source_name = p_source_name;
    GET DIAGNOSTICS removed = ROW_COUNT;
    RETURN removed > 0;
END;
$$;

-- Dump/restore OID refresh. Every identity is re-derived from the stable names
-- already stored in the private catalog. A row whose stable names no longer
-- resolve to a compatible object is marked stale rather than silently rebound,
-- and a row an operator has marked failed is never reactivated by a refresh.
CREATE FUNCTION pgcontext._refresh_lexical_catalog_oids(p_collection_id bigint)
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pgcontext
AS $$
DECLARE
    refreshed bigint := 0;
    touched bigint;
BEGIN
    PERFORM pgcontext._require_collection_owner(p_collection_id);

    UPDATE pgcontext._collection_lexical_sources AS sources
       SET source_table_oid = source_class.oid,
           configuration_oid = configuration.oid,
           status = 'ready',
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS source_class,
           pg_catalog.pg_namespace AS source_namespace,
           pg_catalog.pg_ts_config AS configuration,
           pg_catalog.pg_namespace AS configuration_namespace
     WHERE sources.collection_id = p_collection_id
       AND sources.status <> 'failed'
       AND source_namespace.oid = source_class.relnamespace
       AND source_namespace.nspname = sources.source_schema_name
       AND source_class.relname = sources.source_table_name
       AND source_class.relkind IN ('r', 'p')
       AND configuration_namespace.oid = configuration.cfgnamespace
       AND configuration_namespace.nspname = sources.configuration_schema_name
       AND configuration.cfgname = sources.configuration_name;
    GET DIAGNOSTICS touched = ROW_COUNT;
    refreshed := refreshed + touched;

    UPDATE pgcontext._collection_lexical_sources AS sources
       SET stored_vector_attnum = attribute.attnum
      FROM pg_catalog.pg_attribute AS attribute
     WHERE sources.collection_id = p_collection_id
       AND sources.stored_vector_column_name IS NOT NULL
       AND attribute.attrelid = sources.source_table_oid
       AND attribute.attname = sources.stored_vector_column_name
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped;

    UPDATE pgcontext._collection_lexical_fields AS fields
       SET column_attnum = attribute.attnum,
           column_type_oid = attribute.atttypid,
           column_collation_oid = attribute.attcollation
      FROM pgcontext._collection_lexical_sources AS sources,
           pg_catalog.pg_attribute AS attribute
     WHERE fields.lexical_source_id = sources.lexical_source_id
       AND sources.collection_id = p_collection_id
       AND attribute.attrelid = sources.source_table_oid
       AND attribute.attname = fields.column_name
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped;

    UPDATE pgcontext._collection_lexical_sources AS sources
       SET index_oid = index_class.oid,
           index_definition = pg_catalog.pg_get_indexdef(index_class.oid),
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS index_class
      JOIN pg_catalog.pg_namespace AS index_namespace
        ON index_namespace.oid = index_class.relnamespace
     WHERE sources.collection_id = p_collection_id
       AND sources.index_name IS NOT NULL
       AND index_class.relname = sources.index_name
       AND index_namespace.nspname = sources.source_schema_name
       AND index_class.relkind = 'i';

    UPDATE pgcontext._collection_fuzzy_sources AS sources
       SET source_table_oid = source_class.oid,
           text_attnum = attribute.attnum,
           text_type_oid = attribute.atttypid,
           text_collation_oid = attribute.attcollation,
           trgm_extension_oid = extension.oid,
           trgm_schema_name = extension_namespace.nspname,
           status = 'ready',
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS source_class,
           pg_catalog.pg_namespace AS source_namespace,
           pg_catalog.pg_attribute AS attribute,
           pg_catalog.pg_extension AS extension,
           pg_catalog.pg_namespace AS extension_namespace
     WHERE source_namespace.oid = source_class.relnamespace
       AND attribute.attrelid = source_class.oid
       AND extension.extname = 'pg_trgm'
       AND extension_namespace.oid = extension.extnamespace
       AND sources.collection_id = p_collection_id
       AND source_namespace.nspname = sources.source_schema_name
       AND source_class.relname = sources.source_table_name
       AND source_class.relkind IN ('r', 'p')
       AND attribute.attname = sources.text_column_name
       AND attribute.attnum > 0
       AND NOT attribute.attisdropped
       AND attribute.atttypid = 'pg_catalog.text'::pg_catalog.regtype
       AND sources.status <> 'failed';
    GET DIAGNOSTICS touched = ROW_COUNT;
    refreshed := refreshed + touched;

    UPDATE pgcontext._collection_fuzzy_sources AS sources
       SET index_oid = index_class.oid,
           index_definition = pg_catalog.pg_get_indexdef(index_class.oid),
           updated_at = pg_catalog.now()
      FROM pg_catalog.pg_class AS index_class
      JOIN pg_catalog.pg_namespace AS index_namespace
        ON index_namespace.oid = index_class.relnamespace
     WHERE sources.collection_id = p_collection_id
       AND sources.index_name IS NOT NULL
       AND index_class.relname = sources.index_name
       AND index_namespace.nspname = sources.source_schema_name
       AND index_class.relkind = 'i';

    -- Anything that did not re-resolve is marked stale so query preparation
    -- fails closed instead of trusting an OID the restore invalidated.
    UPDATE pgcontext._collection_lexical_sources AS sources
       SET status = 'stale',
           updated_at = pg_catalog.now()
     WHERE sources.collection_id = p_collection_id
       AND sources.status = 'ready'
       AND NOT EXISTS (
           SELECT 1
             FROM pg_catalog.pg_class AS source_class
             JOIN pg_catalog.pg_namespace AS source_namespace
               ON source_namespace.oid = source_class.relnamespace
            WHERE source_class.oid = sources.source_table_oid
              AND source_namespace.nspname = sources.source_schema_name
              AND source_class.relname = sources.source_table_name
              AND source_class.relkind IN ('r', 'p')
       );

    UPDATE pgcontext._collection_fuzzy_sources AS sources
       SET status = 'stale',
           updated_at = pg_catalog.now()
     WHERE sources.collection_id = p_collection_id
       AND sources.status = 'ready'
       AND NOT EXISTS (
           SELECT 1
             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 attribute
               ON attribute.attrelid = source_class.oid
              AND attribute.attname = sources.text_column_name
              AND attribute.attnum > 0
              AND NOT attribute.attisdropped
              AND attribute.atttypid = 'pg_catalog.text'::pg_catalog.regtype
            WHERE source_class.oid = sources.source_table_oid
              AND source_namespace.nspname = sources.source_schema_name
              AND source_class.relname = sources.source_table_name
              AND source_class.relkind IN ('r', 'p')
       );

    RETURN refreshed;
END;
$$;

SELECT pg_catalog.pg_extension_config_dump(
    'pgcontext._collection_lexical_sources',
    ''
);
SELECT pg_catalog.pg_extension_config_dump(
    'pgcontext._collection_lexical_sources_lexical_source_id_seq',
    ''
);
SELECT pg_catalog.pg_extension_config_dump(
    'pgcontext._collection_lexical_fields',
    ''
);
SELECT pg_catalog.pg_extension_config_dump(
    'pgcontext._collection_fuzzy_sources',
    ''
);
SELECT pg_catalog.pg_extension_config_dump(
    'pgcontext._collection_fuzzy_sources_fuzzy_source_id_seq',
    ''
);
"#,
    name = "create_lexical_catalog_tables",
    requires = ["create_catalog_tables"]
);
