pg_tviews 0.1.0-beta.24

use super::uncascaded::Uncascaded;
use crate::cascade_path;
use crate::config::UncascadedPolicy;
use crate::error::{TViewError, TViewResult};
use crate::schema::{
    TViewSchema, analyzer::analyze_dependencies, direct_map::extract_direct_column_map,
    inference::infer_schema,
};
use crate::utils::{log_debug, quote_identifier};
use pgrx::datum::DatumWithOid;
use pgrx::pg_sys::Oid;
use pgrx::prelude::*;

/// Resolve the target schema for creating TVIEW objects.
///
/// Uses `current_schema()` to respect the active `search_path`, matching
/// standard `PostgreSQL` convention for unqualified DDL statements.
pub(crate) fn current_schema() -> TViewResult<String> {
    crate::utils::spi_get_string("SELECT current_schema()::text")
        .map_err(|e| TViewError::CatalogError {
            operation: "Get current schema".to_string(),
            pg_error: e.to_string(),
        })?
        .ok_or_else(|| TViewError::CatalogError {
            operation: "Get current schema".to_string(),
            pg_error: "current_schema() returned NULL (no schema in search_path?)".to_string(),
        })
}

/// Expand `SELECT * FROM [schema.]source` to an explicit column list.
///
/// When the SELECT is just `SELECT * FROM …`, `pg_tviews` cannot infer the
/// Trinity schema (pk_*, id, data columns) from the wildcard at parse time.
/// This function detects that pattern and expands it by querying
/// `information_schema.columns` for the source view/table's actual columns,
/// preserving their declaration order.
///
/// Returns the original SQL unchanged if it is not a simple `SELECT *`.
///
/// # Errors
/// Returns error only if the `information_schema` query itself fails.
/// A missing source or empty column list silently returns the original SQL
/// so the caller can fall through to the normal error path.
fn expand_select_star_if_needed(select_sql: &str) -> TViewResult<String> {
    let trimmed = select_sql.trim();
    let lower = trimmed.to_lowercase();

    // Must start with SELECT
    let after_kw = lower.strip_prefix("select").unwrap_or("").trim_start();

    // Must have * immediately after SELECT (not SELECT DISTINCT * or SELECT t.*)
    let after_star = match after_kw.strip_prefix('*') {
        Some(rest) => rest.trim_start(),
        None => return Ok(select_sql.to_string()),
    };

    // The token after * must be FROM (no other clauses like WHERE before FROM)
    let after_from = match after_star.strip_prefix("from") {
        Some(rest) if rest.chars().next().is_none_or(|c| c.is_ascii_whitespace()) => {
            rest.trim_start()
        }
        _ => return Ok(select_sql.to_string()),
    };

    // Extract the source name: everything after FROM up to whitespace/semicolon
    let source_qualified = after_from
        .trim_end_matches(';')
        .trim()
        .split_ascii_whitespace()
        .next()
        .unwrap_or("");

    if source_qualified.is_empty() {
        return Ok(select_sql.to_string());
    }

    // Parse optional schema qualifier: "schema.table" or just "table"
    let (schema_name, table_name) = match source_qualified.split_once('.') {
        Some((s, t)) => (
            Some(s.trim_matches('"').to_string()),
            t.trim_matches('"').to_string(),
        ),
        None => (None, source_qualified.trim_matches('"').to_string()),
    };

    // Query information_schema.columns for the column names in order
    let columns: Vec<String> = if let Some(ref schema) = schema_name {
        // SAFETY: DatumWithOid::new wraps PostgreSQL datum pointers for SPI parameter passing.
        // The OID parameter ensures correct type handling in PostgreSQL. Validated strings
        // from table/schema names are passed as text OID parameters.
        let args = vec![
            unsafe {
                pgrx::datum::DatumWithOid::new(
                    table_name.as_str(),
                    pgrx::prelude::PgOid::BuiltIn(pgrx::prelude::PgBuiltInOids::TEXTOID).value(),
                )
            },
            unsafe {
                pgrx::datum::DatumWithOid::new(
                    schema.as_str(),
                    pgrx::prelude::PgOid::BuiltIn(pgrx::prelude::PgBuiltInOids::TEXTOID).value(),
                )
            },
        ];
        pgrx::prelude::Spi::connect(|client| {
            let rows = client.select(
                "SELECT column_name::text \
                 FROM information_schema.columns \
                 WHERE table_name = $1 AND table_schema = $2 \
                 ORDER BY ordinal_position",
                None,
                &args,
            )?;
            let mut result = Vec::new();
            for row in rows {
                if let Some(col) =
                    row[1]
                        .value::<String>()
                        .map_err(|e| TViewError::CatalogError {
                            operation: "expand_select_star: read column_name".to_string(),
                            pg_error: format!("{e:?}"),
                        })?
                {
                    result.push(col);
                }
            }
            Ok(result)
        })
        .map_err(|e: pgrx::spi::Error| TViewError::SpiError {
            query: "expand_select_star: information_schema query".to_string(),
            error: e.to_string(),
        })?
    } else {
        let args = vec![unsafe {
            pgrx::datum::DatumWithOid::new(
                table_name.as_str(),
                pgrx::prelude::PgOid::BuiltIn(pgrx::prelude::PgBuiltInOids::TEXTOID).value(),
            )
        }];
        pgrx::prelude::Spi::connect(|client| {
            let rows = client.select(
                "SELECT column_name::text \
                 FROM information_schema.columns \
                 WHERE table_name = $1 \
                 ORDER BY ordinal_position",
                None,
                &args,
            )?;
            let mut result = Vec::new();
            for row in rows {
                if let Some(col) =
                    row[1]
                        .value::<String>()
                        .map_err(|e| TViewError::CatalogError {
                            operation: "expand_select_star: read column_name".to_string(),
                            pg_error: format!("{e:?}"),
                        })?
                {
                    result.push(col);
                }
            }
            Ok(result)
        })
        .map_err(|e: pgrx::spi::Error| TViewError::SpiError {
            query: "expand_select_star: information_schema query".to_string(),
            error: e.to_string(),
        })?
    };

    if columns.is_empty() {
        // Source not found or no columns — fall through to normal path
        return Ok(select_sql.to_string());
    }

    // Build explicit SELECT preserving original source reference (with schema prefix)
    let col_list = columns.join(", ");
    Ok(format!("SELECT {col_list} FROM {source_qualified}"))
}

/// Storage of a TVIEW's table (issue #134): its persistence, its fillfactor, and
/// whether `data` has a GIN index.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Storage {
    pub logged: bool,
    pub fillfactor: i32,
    pub data_gin_index: bool,
}

impl Storage {
    /// What a new TVIEW gets unless told otherwise: `pg_tviews.unlogged_by_default`,
    /// `pg_tviews.fillfactor` and `pg_tviews.data_gin_index`.
    #[must_use]
    pub fn from_settings() -> Self {
        Self {
            logged: !crate::config::unlogged_by_default(),
            fillfactor: crate::config::fillfactor(),
            data_gin_index: crate::config::data_gin_index(),
        }
    }
}

/// Create a TVIEW in `schema_name` with the given storage, as an aggregate TVIEW
/// when `group_keys` is given, and return the number of rows it was populated
/// with (issue #134).
///
/// Steps: normalize and analyze the definition; create the backing view
/// `v_<entity>` and the table `tv_<entity>`; populate it; register it; install
/// triggers on its base tables.
///
/// # Errors
/// Returns an error if the TVIEW exists, the definition is invalid, or creation fails.
///
/// `policy` is the `uncascaded_policy` to store (a rebuilt TVIEW keeps its own);
/// `None` reads `pg_tviews.uncascaded_policy`.
pub(crate) fn create_tview_in(
    tview_name: &str,
    select_sql: &str,
    schema_name: &str,
    group_keys: Option<&super::aggregate::GroupKeys>,
    storage: Storage,
    policy: Option<UncascadedPolicy>,
) -> TViewResult<u64> {
    let policy = policy.unwrap_or_else(crate::config::uncascaded_policy);
    create_tview_inner(
        tview_name,
        select_sql,
        schema_name,
        group_keys,
        storage,
        policy,
    )
}

/// The creation pipeline's normalization of a definition: `SELECT *` expanded to
/// its columns, and a raw SELECT rewritten to the `pk_<entity>, id, data` shape.
/// Its output normalizes to itself.
///
/// # Errors
/// Returns an error if the definition cannot be analyzed.
pub(crate) fn normalize_definition(
    entity_name: &str,
    select_sql: &str,
) -> TViewResult<(String, TViewSchema)> {
    let select_sql = expand_select_star_if_needed(select_sql)?;
    let schema = infer_schema(&select_sql)?;
    if schema.entity_name.is_none() {
        transform_raw_select_to_tview(entity_name, &select_sql)
    } else {
        Ok((select_sql, schema))
    }
}

fn create_tview_inner(
    tview_name: &str,
    select_sql: &str,
    schema_name: &str,
    group_keys: Option<&super::aggregate::GroupKeys>,
    storage: Storage,
    policy: UncascadedPolicy,
) -> TViewResult<u64> {
    crate::revision::check();
    log_debug!(
        "create_tview start for '{}' in schema '{}'",
        tview_name,
        schema_name
    );
    // Calls that register, change or drop one entity run one after the other.
    super::lock_entity(tview_name.strip_prefix("tv_").unwrap_or(tview_name))?;

    // Step 1: Check if TVIEW already exists
    let exists = tview_exists(tview_name)?;
    if exists {
        return Err(TViewError::TViewAlreadyExists {
            name: tview_name.to_string(),
        });
    }

    // Step 1.5: Extract entity name from tview_name
    // Support both "tv_entity" and just "entity" formats
    let entity_name = tview_name
        .strip_prefix("tv_")
        .map_or(tview_name, |stripped| stripped);

    // Step 2: Expand SELECT * so infer_schema recognises the Trinity Pattern
    // (pk_*, id, data) even for `CREATE TABLE tv_foo AS SELECT * FROM v_foo_base`,
    // and rewrite a raw SELECT to that shape.
    let (final_select_sql, final_schema) = normalize_definition(entity_name, select_sql)?;

    let entity_name =
        final_schema
            .entity_name
            .as_ref()
            .ok_or_else(|| TViewError::RequiredColumnMissing {
                column_name: format!(
                    "pk_{}",
                    tview_name.strip_prefix("tv_").unwrap_or(tview_name)
                ),
                context: "pg_tviews requires a Trinity Pattern primary key column named \
                      \"pk_<entity>\" (e.g., pk_user, pk_post)"
                    .to_string(),
            })?;

    // Validate entity_name inferred from the SELECT to prevent SQL injection
    // (tview_name is validated at the pg_extern boundary, but entity_name comes
    // from infer_schema and could contain metacharacters if the user crafts a
    // malicious column alias like pk_evil'injection).
    crate::validation::validate_sql_identifier(entity_name, "entity_name")?;

    if group_keys.is_some() {
        super::aggregate::validate_definition(&final_select_sql, entity_name).map_err(
            |reason| TViewError::InvalidInput {
                parameter: "aggregate definition".to_string(),
                reason,
            },
        )?;
    }

    // Derive the canonical materialized-table name: always tv_<entity>.
    // This normalises both calling conventions:
    //   pg_tviews_create('post', ...)   → tv_post
    //   pg_tviews_create('tv_post', ...) → tv_post
    let tv_table_name = format!("tv_{entity_name}");

    let schema_name = schema_name.to_string();

    // Reject WITH RECURSIVE up front (issue #51): cascade paths cannot be tracked
    // through a recursive CTE, so creating one would leave a tview that silently
    // refreshes incompletely.
    if crate::sql_parser::has_recursive_cte(&final_select_sql) {
        return Err(TViewError::InvalidInput {
            parameter: "tview definition".to_string(),
            reason: format!(
                "TVIEW '{tv_table_name}' uses WITH RECURSIVE, which pg_tviews does not support: \
                 cascade paths cannot be tracked through a recursive CTE, so the tview would \
                 refresh incompletely. Rewrite the definition without recursion."
            ),
        });
    }

    // Step 3: Create backing view v_<entity>
    let view_name = format!("v_{entity_name}");
    create_backing_view(&view_name, &final_select_sql, &schema_name)?;

    // Step 4: Find base table dependencies, and how a write to each maps to keys,
    // from the view's query tree (ADR 0157), with the column that names the
    // TVIEW's rows (ADR 0169); and the cascade paths of its local tables (for an
    // aggregate TVIEW, one per declared group key, issue #58).
    // Pass schema_name so the view OID lookup searches in the correct schema even when
    // current_schema() resolves to a different schema due to the database search_path.
    let dep_graph = crate::dependency::find_base_tables(&view_name, Some(&schema_name))?;
    let Derivation {
        lineage,
        key_mappings,
        cascade_paths,
    } = derive(
        entity_name,
        &final_select_sql,
        &final_schema,
        group_keys,
        &dep_graph.base_tables,
        &schema_name,
    )?;

    // Step 5: Create materialized table tv_<entity>, keyed on the identity.
    create_materialized_table(
        &tv_table_name,
        &final_schema,
        &schema_name,
        &lineage.identity.name,
        storage,
    )?;

    // Step 6: Populate initial data
    let rows = populate_initial_data(&tv_table_name, &view_name, &schema_name)?;

    // Step 6.6: Reject a tview that can never be incrementally refreshed (issue #49).
    // A refresh is enqueued for this entity only if either a `tb_<entity>` base table
    // exists (its row trigger resolves the entity by stripping `tb_`) or a cascade
    // path routes some base-table change to it. With neither, the tview would silently
    // never refresh — and, when built directly on a `tb_` base table that already
    // backs another entity (e.g. `order_summary` over `tb_order`), shadow that
    // sibling. Aggregate/summary tviews without a `tb_<entity>` stay valid as long as
    // their joins produce cascade paths. This runs before metadata registration and
    // trigger installation, and any objects created above roll back with the ERROR,
    // so a rejected create leaves the incumbent tview untouched.
    let root_table = format!("tb_{entity_name}");
    if group_keys.is_none()
        && !cascade_paths
            .iter()
            .any(|p| p.source_table != root_table && (!p.root || lineage.is_union()))
        && !lineage.has_mapped()
        && !entity_base_table_exists(entity_name, &schema_name)?
    {
        return Err(TViewError::InvalidInput {
            parameter: "tview definition".to_string(),
            reason: format!(
                "TVIEW '{tv_table_name}' (entity '{entity_name}') can never be refreshed: \
                 there is no base table 'tb_{entity_name}', and no cascade path routes any \
                 base-table change to it. pg_tviews maintains a tview either directly (a \
                 pk_<entity> primary key over a tb_<entity> base table) or via cascade \
                 paths from its joined base tables. Rename the primary-key column to match \
                 an existing base table, or ensure the definition joins the base tables it \
                 derives from. Creating it as-is would leave a permanently stale tview and \
                 can silently shadow a correctly-named tview on the same base table."
            ),
        });
    }

    // Step 6.7: Base tables whose writes no cascade reaches (issues #157, #158),
    // reported under the policy; `error` aborts here, and the objects created above
    // roll back with it.
    let uncascaded = Uncascaded {
        tables: uncascaded_tables(&lineage),
        policy,
    };
    super::uncascaded::report(
        &crate::utils::qualified_relname_from_oid(relation_oid(&schema_name, &tv_table_name)?)?,
        &uncascaded.tables,
        policy,
    )?;

    // Step 7: Register metadata (with cascade paths)
    register_metadata(
        entity_name,
        &view_name,
        &tv_table_name,
        &final_select_sql,
        &final_schema,
        &cascade_paths,
        &schema_name,
        group_keys,
        &uncascaded,
        &key_mappings,
        &lineage,
        false,
    )?;

    // Step 8: Install triggers on base tables, as their lineage needs them.
    if dep_graph.base_tables.is_empty() {
        warning!("No base table dependencies found for {}", tv_table_name);
    } else {
        crate::dependency::install_triggers(
            &crate::dependency::trigger_plan(&dep_graph.base_tables, &lineage)?,
            entity_name,
        )?;
    }

    // Invalidate caches since new TVIEW was created
    crate::queue::cache::invalidate_all_caches();

    // Buffer and flush audit entry immediately (we're in SPI context)
    crate::audit::log_create(entity_name, &final_select_sql);
    if let Err(e) = crate::audit::flush_audit_buffer() {
        warning!("Failed to flush audit after CREATE: {}", e);
    }

    Ok(rows)
}

/// Re-derive and replace the metadata of an existing TVIEW from `definition`,
/// with the same analysis as `create_tview_in`, and return the base tables its
/// backing view reads. Used when a column rename has changed the text that
/// defines `v_<entity>`, and by `pg_tviews_reregister()`; the relations themselves
/// are unchanged.
///
/// # Errors
/// Returns an error if the definition cannot be analyzed or the catalog update fails.
pub fn reregister_metadata(
    entity_name: &str,
    schema_name: &str,
    definition: &str,
) -> TViewResult<crate::dependency::TriggerPlan> {
    let schema = infer_schema(definition)?;
    let view_name = format!("v_{entity_name}");
    let dep_graph = crate::dependency::find_base_tables(&view_name, Some(schema_name))?;
    let group_keys = stored_group_keys(entity_name)?;
    // The stored policy holds; an `error` TVIEW whose set is no longer empty
    // aborts the re-registration (and the ALTER that caused it).
    let policy = crate::catalog::TviewMeta::load_by_entity(entity_name)
        .map_err(|e| TViewError::CatalogError {
            operation: format!("Read the uncascaded policy of tv_{entity_name}"),
            pg_error: e.to_string(),
        })?
        .map_or_else(crate::config::uncascaded_policy, |m| m.uncascaded_policy);
    let Derivation {
        lineage,
        key_mappings,
        cascade_paths,
    } = derive(
        entity_name,
        definition,
        &schema,
        group_keys.as_ref(),
        &dep_graph.base_tables,
        schema_name,
    )?;
    key_table_on_identity(
        schema_name,
        &format!("tv_{entity_name}"),
        &lineage.identity.name,
    )?;
    let uncascaded = Uncascaded {
        tables: uncascaded_tables(&lineage),
        policy,
    };
    super::uncascaded::report(
        &crate::utils::qualified_relname_from_oid(relation_oid(
            schema_name,
            &format!("tv_{entity_name}"),
        )?)?,
        &uncascaded.tables,
        policy,
    )?;
    register_metadata(
        entity_name,
        &view_name,
        &format!("tv_{entity_name}"),
        definition,
        &schema,
        &cascade_paths,
        schema_name,
        group_keys.as_ref(),
        &uncascaded,
        &key_mappings,
        &lineage,
        true,
    )?;
    crate::queue::cache::invalidate_all_caches();
    crate::queue::cache::invalidate_all_caches();
    crate::dependency::trigger_plan(&dep_graph.base_tables, &lineage)
}

/// Re-derive `entity`'s metadata from its stored definition and make its
/// base-table triggers match what that definition reads (issue #137).
///
/// # Errors
/// Returns an error if the TVIEW is not registered, the caller does not own it,
/// or the definition cannot be analyzed.
pub fn reregister_tview(entity: &str) -> TViewResult<()> {
    super::lock_entity(entity)?;
    let meta = crate::catalog::TviewMeta::load_by_entity(entity)?.ok_or_else(|| {
        TViewError::MetadataNotFound {
            entity: entity.to_string(),
        }
    })?;
    crate::owner::require_owner(meta.tview_oid, &format!("tv_{entity}"))?;
    let (definition, schema_name) =
        Spi::connect(|client| {
            // SAFETY: the datum borrows `entity`, which outlives the select.
            let args = [unsafe {
                DatumWithOid::new(entity, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value())
            }];
            client
                .select(
                    &format!(
                        "SELECT m.definition, n.nspname::text \
                     FROM {} m \
                     JOIN pg_catalog.pg_class c ON c.oid = m.table_oid \
                     JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
                     WHERE m.entity = $1",
                        crate::utils::meta_table()
                    ),
                    None,
                    &args,
                )?
                .first()
                .get_two::<String, String>()
        })
        .map_err(|e| TViewError::CatalogError {
            operation: format!("Read the definition of TVIEW {entity}"),
            pg_error: e.to_string(),
        })?;
    let (Some(definition), Some(schema_name)) = (definition, schema_name) else {
        return Err(TViewError::MetadataNotFound {
            entity: entity.to_string(),
        });
    };
    let plan = reregister_metadata(entity, &schema_name, &definition)?;
    crate::dependency::sync_entity_triggers(&plan, entity)?;
    let _owner = crate::owner::AsOwner::of_extension()?;
    Spi::run_with_args(
        &format!(
            "UPDATE {} SET needs_reregister = false WHERE entity = $1",
            crate::utils::meta_table()
        ),
        &[unsafe { DatumWithOid::new(entity, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) }],
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Clear needs_reregister of TVIEW {entity}"),
        pg_error: e.to_string(),
    })
}

/// The `group_keys` of an aggregate TVIEW (issue #58), `None` for any other.
pub(crate) fn stored_group_keys(
    entity_name: &str,
) -> TViewResult<Option<super::aggregate::GroupKeys>> {
    let stored: Option<pgrx::JsonB> = Spi::get_one_with_args(
        &format!(
            "SELECT group_keys FROM {} WHERE entity = $1",
            crate::utils::meta_table()
        ),
        &[unsafe {
            DatumWithOid::new(entity_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value())
        }],
    )
    .map_err(|e| TViewError::CatalogError {
        operation: "Read group_keys".to_string(),
        pg_error: e.to_string(),
    })?;
    Ok(stored.and_then(|j| serde_json::from_value(j.0).ok()))
}

/// What registration derives from a definition: how a write to each base table
/// maps to keys (stored as `key_mappings`), and the cascade paths the row trigger
/// follows for local tables.
struct Derivation {
    lineage: crate::lineage::Lineage,
    key_mappings: serde_json::Value,
    cascade_paths: Vec<cascade_path::CascadePath>,
}

/// Analyze `v_<entity>` (ADR 0157) and derive the cascade paths of its local
/// tables, or for an aggregate TVIEW one per declared group key (issue #58).
fn derive(
    entity_name: &str,
    definition: &str,
    schema: &TViewSchema,
    group_keys: Option<&super::aggregate::GroupKeys>,
    base_tables: &[pg_sys::Oid],
    schema_name: &str,
) -> TViewResult<Derivation> {
    let view_oid = relation_oid(schema_name, &format!("v_{entity_name}"))?;
    let mut embeds: Vec<String> = aggregate_embeds(definition, entity_name)?
        .into_keys()
        .collect();
    embeds.extend(
        schema
            .fk_columns
            .iter()
            .filter_map(|c| c.strip_prefix("fk_").map(str::to_string)),
    );
    let lineage = crate::lineage::analyze(entity_name, view_oid, base_tables, &embeds)?;
    let cascade_paths = match group_keys {
        Some(keys) => super::aggregate::cascade_paths(entity_name, keys, base_tables, schema_name)?,
        None => local_cascade_paths(entity_name, definition, schema, &lineage),
    };
    let mut key_mappings = lineage.to_json();
    add_fanout_patches(&mut key_mappings, entity_name, definition, schema, &lineage);
    Ok(Derivation {
        lineage,
        key_mappings,
        cascade_paths,
    })
}

/// A `mapped` table linked to `tb_<entity>` by one equality onto a column the
/// TVIEW projects gets the fan-out patch of issue #120: an UPDATE of its columns
/// copied into `data` is written into every TVIEW row with that column's value.
fn add_fanout_patches(
    key_mappings: &mut serde_json::Value,
    entity_name: &str,
    definition: &str,
    schema: &TViewSchema,
    lineage: &crate::lineage::Lineage,
) {
    let root = format!("tb_{entity_name}");
    // A DISTINCT ON TVIEW's row shows its group's winner: writing one source row
    // into it could write a loser's values.
    if lineage.identity.kind == crate::lineage::IdentityKind::DistinctOn {
        return;
    }
    let Some(entries) = key_mappings.as_array_mut() else {
        return;
    };
    for table in &lineage.tables {
        let Some((own, root_col)) = &table.hop else {
            continue;
        };
        // The shape `fanout_patch` reads: one hop into tb_<entity> on `root_col`.
        let path = cascade_path::CascadePath {
            source_oid: pg_sys::Oid::from(table.relid),
            source_table: table.relname.clone(),
            entity_name: entity_name.to_string(),
            initial_col: own.clone(),
            hops: vec![cascade_path::CascadeHop {
                table_oid: pg_sys::Oid::INVALID,
                table_name: root.clone(),
                lookup_col: root_col.clone(),
                carry_col: format!("pk_{entity_name}"),
            }],
            unresolvable: false,
            source_columns: table.columns.iter().map(|(name, _)| name.clone()).collect(),
            fanout: None,
            root: false,
            initial_attnum: None,
        };
        let Some(fanout) = fanout_patch(definition, entity_name, schema, &path, lineage) else {
            continue;
        };
        if let Some(entry) = entries
            .iter_mut()
            .find(|e| e["relid"].as_u64() == Some(u64::from(table.relid)))
        {
            entry["key_col"] = own.clone().into();
            entry["fanout"] = serde_json::to_value(&fanout).unwrap_or_default();
        }
    }
}

/// One cascade path per local table: the key is the table's `column`, read off the
/// changed row; an UPDATE that touches none of the columns the TVIEW reads is
/// skipped. The table holding the key (of each UNION branch) gets one too, marked
/// `root`: its own rows are the TVIEW's rows (ADR 0169).
fn local_cascade_paths(
    entity_name: &str,
    definition: &str,
    schema: &TViewSchema,
    lineage: &crate::lineage::Lineage,
) -> Vec<cascade_path::CascadePath> {
    lineage
        .tables
        .iter()
        .filter_map(|t| match &t.kind {
            crate::lineage::TableKind::Local(column) => {
                let mut path = cascade_path::CascadePath {
                    source_oid: pg_sys::Oid::from(t.relid),
                    source_table: t.relname.clone(),
                    entity_name: entity_name.to_string(),
                    initial_col: column.clone(),
                    hops: Vec::new(),
                    unresolvable: false,
                    source_columns: t.columns.iter().map(|(name, _)| name.clone()).collect(),
                    fanout: None,
                    root: t.root,
                    initial_attnum: t.columns.iter().find(|(n, _)| n == column).map(|(_, a)| *a),
                };
                if !t.root && lineage.identity.kind == crate::lineage::IdentityKind::Pk {
                    path.fanout = fanout_patch(definition, entity_name, schema, &path, lineage);
                }
                Some(path)
            }
            _ => None,
        })
        .collect()
}

/// Make a TVIEW's table keyed on its identity at re-registration (ADR 0169): drop
/// the unique index on `pk_<entity>` a DISTINCT ON TVIEW of beta.22 had (#164), and
/// add the primary key a table created without one lacks (a DISTINCT ON key named
/// `identifier`, `fk_*` or `*_id`). A primary key on another column is refused.
fn key_table_on_identity(schema_name: &str, tview_name: &str, identity: &str) -> TViewResult<()> {
    let table = relation_oid(schema_name, tview_name)?;
    let qualified = crate::utils::qualified_relname_from_oid(table)?;
    let catalog = |e: pgrx::spi::Error| TViewError::CatalogError {
        operation: format!("Read the keys of {qualified}"),
        pg_error: e.to_string(),
    };
    let pk_unique = index_name(tview_name, "pk_unique");
    // SAFETY: the datums borrow values that outlive each call.
    let leftover = Spi::get_one_with_args::<bool>(
        "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_index i \
                        JOIN pg_catalog.pg_class c ON c.oid = i.indexrelid \
                        WHERE i.indrelid = $1 AND c.relname = $2)",
        &[
            unsafe { DatumWithOid::new(table, PgOid::BuiltIn(PgBuiltInOids::OIDOID).value()) },
            unsafe {
                DatumWithOid::new(
                    pk_unique.as_str(),
                    PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
                )
            },
        ],
    )
    .map_err(catalog)?;
    if leftover == Some(true) {
        let sql = format!(
            "DROP INDEX {}.{}",
            quote_identifier(schema_name),
            quote_identifier(&pk_unique)
        );
        crate::utils::spi_run_ddl(&sql).map_err(|e| TViewError::SpiError {
            query: sql,
            error: e,
        })?;
    }
    let key = Spi::get_one_with_args::<Vec<String>>(
        "SELECT pg_catalog.array_agg(a.attname::text ORDER BY a.attnum) \
         FROM pg_catalog.pg_index i \
         JOIN pg_catalog.pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY (i.indkey) \
         WHERE i.indrelid = $1 AND i.indisprimary",
        // SAFETY: a plain OID datum.
        &[unsafe { DatumWithOid::new(table, PgOid::BuiltIn(PgBuiltInOids::OIDOID).value()) }],
    )
    .map_err(catalog)?
    .unwrap_or_default();
    match key.as_slice() {
        [column] if column == identity => Ok(()),
        [] => {
            let sql = format!(
                "ALTER TABLE {qualified} ADD PRIMARY KEY ({})",
                quote_identifier(identity)
            );
            crate::utils::spi_run_ddl(&sql).map_err(|e| TViewError::SpiError {
                query: sql,
                error: e,
            })
        }
        _ => Err(TViewError::InvalidInput {
            parameter: "tview definition".to_string(),
            reason: format!(
                "{qualified} is keyed on ({}), but its rows are named by {identity}: \
                 pg_tviews_create_or_replace() with the same query rebuilds it",
                key.join(", ")
            ),
        }),
    }
}

/// The tables of `lineage` no cascade reaches, as the policy reports them.
fn uncascaded_tables(lineage: &crate::lineage::Lineage) -> Vec<super::uncascaded::UncascadedTable> {
    lineage
        .all_keys()
        .into_iter()
        .map(|(relid, name, reason)| super::uncascaded::UncascadedTable {
            oid: pg_sys::Oid::from(relid),
            name,
            reason,
        })
        .collect()
}

/// The OID of relation `schema.name`.
fn relation_oid(schema: &str, name: &str) -> TViewResult<pg_sys::Oid> {
    Spi::get_one_with_args::<pg_sys::Oid>(
        "SELECT c.oid FROM pg_catalog.pg_class c \
         JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
         WHERE n.nspname = $1 AND c.relname = $2",
        &[
            unsafe { DatumWithOid::new(schema, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
            unsafe { DatumWithOid::new(name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
        ],
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Look up {schema}.{name}"),
        pg_error: e.to_string(),
    })?
    .ok_or_else(|| TViewError::CatalogError {
        operation: format!("Look up {schema}.{name}"),
        pg_error: "relation not found".to_string(),
    })
}

/// The one-statement fan-out patch of a cascade path (issue #120): set when an
/// UPDATE of the source row can be written into every target row it reaches
/// instead of recomputing each.
///
/// Requires a path of one hop into `tb_<entity>` landing on its pk, a target row
/// found through a projected, indexed `fk_*` column, and source columns copied
/// unchanged into top-level `data` keys and used nowhere else in the definition.
/// Anything else (`None`, or a column left out of `fields`) keeps the recompute.
fn fanout_patch(
    select_sql: &str,
    entity_name: &str,
    schema: &TViewSchema,
    path: &cascade_path::CascadePath,
    lineage: &crate::lineage::Lineage,
) -> Option<cascade_path::FanoutPatch> {
    let [hop] = path.hops.as_slice() else {
        return None;
    };
    let own_table = format!("tb_{entity_name}");
    if hop.table_name != own_table
        || hop.carry_col != format!("pk_{entity_name}")
        || path.source_columns.is_empty()
        || lineage.set_operation
    {
        return None;
    }
    let lookup_col = crate::sql_parser::output_column_for(select_sql, &own_table, &hop.lookup_col)?;
    // Never copy a virtual generated column or an input of one (#179).
    let virtual_reads: Vec<&String> = lineage
        .tables
        .iter()
        .filter(|t| pg_sys::Oid::from(t.relid) == path.source_oid)
        .flat_map(|t| &t.virtual_reads)
        .collect();
    if !schema.fk_columns.contains(&lookup_col) {
        return None;
    }
    let qualifier = crate::sql_parser::table_qualifier(select_sql, &path.source_table)?;
    let outside_data = crate::schema::direct_map::columns_referenced_outside_data(select_sql)?;
    let fields: Vec<(String, String)> =
        crate::schema::direct_map::extract_joined_column_map(select_sql, &qualifier)
            .into_iter()
            .filter(|(col, _)| {
                !virtual_reads.contains(&col)
                    && *col != path.initial_col
                    && path.source_columns.contains(col)
                    && !outside_data.contains(&col.to_lowercase())
            })
            .collect();
    (!fields.is_empty()).then_some(cascade_path::FanoutPatch { lookup_col, fields })
}

/// Re-resolve the relation OIDs stored inside serialized cascade paths against
/// the current catalog, using the same relname → OID map that creation built
/// from the backing view's base tables.
///
/// Cascade paths carry raw OIDs inside JSON text, which `pg_dump` copies
/// verbatim. After a restore those OIDs name nothing (or an unrelated
/// relation), so `pg_tview_meta`'s insert trigger calls this to rebind them.
/// For a freshly created TVIEW the result is identical to the input. A path
/// whose table can no longer be found is marked `unresolvable` (full-refresh
/// fallback) rather than left pointing at a stale OID.
pub fn rebind_cascade_paths(view_oid: Oid, cascade_paths: &[String]) -> TViewResult<Vec<String>> {
    if cascade_paths.is_empty() {
        return Ok(Vec::new());
    }

    let args =
        [unsafe { DatumWithOid::new(view_oid, PgOid::BuiltIn(PgBuiltInOids::OIDOID).value()) }];
    let (view_name, schema_name) = Spi::get_two_with_args::<String, String>(
        "SELECT c.relname::text, n.nspname::text FROM pg_class c \
         JOIN pg_namespace n ON c.relnamespace = n.oid WHERE c.oid = $1",
        &args,
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Resolve backing view {view_oid:?}"),
        pg_error: e.to_string(),
    })?;
    let (Some(view_name), Some(schema_name)) = (view_name, schema_name) else {
        return Err(TViewError::CatalogError {
            operation: format!("Resolve backing view {view_oid:?}"),
            pg_error: "view not found".to_string(),
        });
    };

    let dep_graph = crate::dependency::find_base_tables(&view_name, Some(&schema_name))?;
    let oid_map = build_oid_name_map(&dep_graph.base_tables)?;

    cascade_paths
        .iter()
        .map(|json| {
            let mut path: cascade_path::CascadePath =
                serde_json::from_str(json).map_err(|e| TViewError::CatalogError {
                    operation: "Parse cascade path".to_string(),
                    pg_error: e.to_string(),
                })?;
            match oid_map.get(&path.source_table) {
                Some(oid) => path.source_oid = *oid,
                None => path.unresolvable = true,
            }
            for hop in &mut path.hops {
                match oid_map.get(&hop.table_name) {
                    Some(oid) => hop.table_oid = *oid,
                    None => path.unresolvable = true,
                }
            }
            serde_json::to_string(&path).map_err(|e| TViewError::CatalogError {
                operation: "Serialize cascade path".to_string(),
                pg_error: e.to_string(),
            })
        })
        .collect()
}

/// Build a map from table name → OID for a set of base table OIDs.
fn build_oid_name_map(
    oids: &[pg_sys::Oid],
) -> TViewResult<std::collections::HashMap<String, pg_sys::Oid>> {
    use std::collections::HashMap;

    if oids.is_empty() {
        return Ok(HashMap::new());
    }

    let oid_list = oids
        .iter()
        .map(|o| o.to_u32().to_string())
        .collect::<Vec<_>>()
        .join(",");

    let query = format!("SELECT oid, relname::text FROM pg_class WHERE oid IN ({oid_list})");

    let mut map = HashMap::new();
    Spi::connect(|client| {
        let rows = client.select(&query, None, &[])?;
        for row in rows {
            let oid: pg_sys::Oid = row["oid"].value()?.unwrap_or(pg_sys::Oid::INVALID);
            let name: String = row["relname"].value()?.unwrap_or_default();
            map.insert(name, oid);
        }
        Ok::<_, spi::Error>(())
    })?;

    Ok(map)
}

/// Columns of `source_table` that the backing view `v_<entity>` depends on,
/// read from `PostgreSQL`'s own column-level `pg_depend` records. This is the
/// exact set of source columns whose change can alter a target tview row —
/// it correctly accounts for expressions, `SELECT *` expansion, and repeated
/// joins, with none of the fragility of parsing the SELECT text.
///
/// Returns an empty vec on any error or when the view has no *direct* column
/// dependency on `source_table` (e.g. a multi-hop cascade whose backing view
/// references an intermediate view, not the leaf table). The caller treats an
/// empty result as "unknown ⇒ always refresh", so a miss is never unsafe.
pub(crate) fn view_source_columns(schema: &str, entity: &str, source_oid: Oid) -> Vec<String> {
    // The view and schema names are bound as text parameters (no in-band SQL
    // quoting); the source table is matched by the OID the cascade path resolved,
    // so a joined table in another schema is found too.
    const QUERY: &str = "SELECT a.attname::text AS col \
         FROM pg_depend d \
         JOIN pg_rewrite r ON r.oid = d.objid \
         JOIN pg_class v ON v.oid = r.ev_class \
         JOIN pg_attribute a ON a.attrelid = d.refobjid AND a.attnum = d.refobjsubid \
         WHERE v.relname = $1 AND v.relnamespace = $2::regnamespace \
           AND d.refobjid = $3 AND d.refobjsubid > 0";
    let view_name = format!("v_{entity}");
    let mut cols = Vec::new();
    let result = Spi::connect(|client| {
        // SAFETY: DatumWithOid::new wraps datum pointers for SPI parameter passing;
        // view_name/schema outlive this select call.
        let text = PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value();
        let args = vec![
            unsafe { DatumWithOid::new(view_name.as_str(), text) },
            unsafe { DatumWithOid::new(schema, text) },
            unsafe { DatumWithOid::new(source_oid, PgOid::BuiltIn(PgBuiltInOids::OIDOID).value()) },
        ];
        let rows = client.select(QUERY, None, &args)?;
        for row in rows {
            if let Ok(Some(name)) = row["col"].value::<String>() {
                cols.push(name);
            }
        }
        Ok::<_, spi::Error>(())
    });
    if let Err(e) = result {
        notice!(
            "view_source_columns({view_name}, {source_oid:?}): {e} — cascade will always refresh"
        );
        return Vec::new();
    }
    cols
}

/// Check if a TVIEW already exists
fn tview_exists(tview_name: &str) -> TViewResult<bool> {
    let entity_name = tview_name.trim_start_matches("tv_");
    let args = vec![unsafe {
        DatumWithOid::new(entity_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value())
    }];

    Spi::get_one_with_args::<bool>(
        &format!(
            "SELECT COUNT(*) > 0 FROM {} WHERE entity = $1",
            crate::utils::meta_table()
        ),
        &args,
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Check TVIEW exists: {tview_name}"),
        pg_error: format!("{e:?}"),
    })
    .map(|opt| opt.unwrap_or(false))
}

/// Does a `tb_<entity>` base table exist for this entity?
///
/// Used by the issue #49 refreshability guard: a `tb_<entity>` base table means the
/// entity is reachable directly (its row trigger resolves the entity by stripping
/// `tb_`). The table is resolved in the tview's target schema first, then via the
/// active `search_path`, so both explicit-schema and `current_schema()` callers are
/// handled.
fn entity_base_table_exists(entity_name: &str, schema_name: &str) -> TViewResult<bool> {
    let tb_name = format!("tb_{entity_name}");
    let qualified = format!(
        "{}.{}",
        quote_identifier(schema_name),
        quote_identifier(&tb_name)
    );

    Spi::get_one_with_args::<bool>(
        "SELECT COALESCE(to_regclass($1), to_regclass($2)) IS NOT NULL",
        &[
            unsafe {
                DatumWithOid::new(
                    qualified.as_str(),
                    PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
                )
            },
            unsafe {
                DatumWithOid::new(
                    tb_name.as_str(),
                    PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
                )
            },
        ],
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Check base table exists for entity '{entity_name}'"),
        pg_error: format!("{e:?}"),
    })
    .map(|opt| opt.unwrap_or(false))
}

/// Create the backing view that contains the user's SELECT definition
fn create_backing_view(view_name: &str, select_sql: &str, schema_name: &str) -> TViewResult<()> {
    let qi_schema = quote_identifier(schema_name);
    let qi_view = quote_identifier(view_name);
    let create_view_sql = format!("CREATE VIEW {qi_schema}.{qi_view} AS {select_sql}");

    log_debug!(
        "create_backing_view START - schema='{}', view='{}', sql_len={}",
        schema_name,
        view_name,
        create_view_sql.len()
    );

    match crate::utils::spi_run_ddl(&create_view_sql) {
        Ok(()) => {
            // Log successful spi_run_ddl
            log_debug!("spi_run_ddl SUCCEEDED for {}.{}", schema_name, view_name);
        }
        Err(e) => {
            // Log spi_run_ddl failure
            log_debug!(
                "spi_run_ddl FAILED - {}.{} - error: {}",
                schema_name,
                view_name,
                e
            );
            // Note: error!() macro diverges, so this return is unreachable but needed for type checking
            return Err(TViewError::SpiError {
                query: create_view_sql.clone(),
                error: e,
            });
        }
    }

    // Log before verification check
    log_debug!(
        "checking if view exists - schema='{}', view='{}' in pg_class",
        schema_name,
        view_name
    );

    // Verify the view was created (schema-qualified to avoid false positives across schemas)
    let check_args = vec![
        unsafe { DatumWithOid::new(view_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
        unsafe { DatumWithOid::new(schema_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
    ];
    let exists = match Spi::get_one_with_args::<i32>(
        "SELECT 1 FROM pg_class c \
         JOIN pg_namespace n ON c.relnamespace = n.oid \
         WHERE c.relname = $1 AND n.nspname = $2 AND c.relkind = 'v'",
        &check_args,
    ) {
        Ok(result) => {
            if result.is_some() {
                // Log successful verification
                log_debug!(
                    "VERIFIED - backing view {}.{} exists in pg_class",
                    schema_name,
                    view_name
                );
                true
            } else {
                // Log verification failure
                log_debug!(
                    "VERIFICATION FAILED - backing view {}.{} not found in pg_class after spi_run_ddl",
                    schema_name,
                    view_name
                );
                false
            }
        }
        Err(e) => {
            // Log verification query failure
            log_debug!(
                "verification query FAILED - could not check pg_class: {}",
                e
            );
            // Note: error!() macro diverges, so this return is unreachable but needed for type checking
            return Err(TViewError::SpiError {
                query: format!("Check view {schema_name}.{view_name} exists"),
                error: e.to_string(),
            });
        }
    };

    if !exists {
        return Err(TViewError::CatalogError {
            operation: format!("Create view {schema_name}.{view_name}"),
            pg_error: "View was not created (CREATE VIEW succeeded but view missing from pg_class)"
                .to_string(),
        });
    }

    Ok(())
}

/// The OID of view `schema.view`.
fn view_oid(schema: &str, view: &str) -> TViewResult<pg_sys::Oid> {
    let qualified = format!("{}.{}", quote_identifier(schema), quote_identifier(view));
    // SAFETY: the text datum borrows `qualified`, which outlives the select.
    Spi::get_one_with_args::<pg_sys::Oid>(
        "SELECT pg_catalog.to_regclass($1)::pg_catalog.oid",
        &[unsafe {
            DatumWithOid::new(
                qualified.as_str(),
                PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
            )
        }],
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Find view {qualified}"),
        pg_error: e.to_string(),
    })?
    .ok_or_else(|| TViewError::CatalogError {
        operation: format!("Find view {qualified}"),
        pg_error: "the view does not exist".to_string(),
    })
}

/// Create the materialized table with proper schema inferred from the backing view,
/// with its primary key on the TVIEW's identity column (ADR 0169).
fn create_materialized_table(
    tview_name: &str,
    schema: &TViewSchema,
    schema_name: &str,
    identity: &str,
    storage: Storage,
) -> TViewResult<()> {
    let qi_schema = quote_identifier(schema_name);
    let qi_tview = quote_identifier(tview_name);
    // `<name> <type>`, with PRIMARY KEY on the identity column.
    let key = |name: &str| if name == identity { " PRIMARY KEY" } else { "" };

    // Build column definitions based on inferred schema
    let mut columns = Vec::new();

    // pk_<entity>: the row identity, or a plain column (DISTINCT ON another key)
    if let Some(pk) = &schema.pk_column {
        columns.push(format!("{} BIGINT{}", quote_identifier(pk), key(pk)));
    }

    // ID column (Trinity identifier)
    if let Some(id) = &schema.id_column {
        let not_null = if id == identity { key(id) } else { " NOT NULL" };
        columns.push(format!("{} UUID{not_null}", quote_identifier(id)));
    }

    // Every other column takes the backing view's type, typmod included and
    // schema-qualified (an enum, a domain, a composite, `numeric(6,2)`, `bit(4)`).
    // The convention columns above keep their fixed types: the key is BIGINT,
    // `id` UUID, `data` JSONB and `fk_*` BIGINT whatever the view computes them as.
    let entity = tview_name.strip_prefix("tv_").unwrap_or(tview_name);
    let view_types: std::collections::HashMap<String, String> =
        crate::utils::column_types(view_oid(schema_name, &format!("v_{entity}"))?)?
            .into_iter()
            .collect();
    let view_type = |col: &str, fallback: &str| {
        view_types
            .get(col)
            .cloned()
            .unwrap_or_else(|| fallback.to_string())
    };

    // Identifier column (optional Trinity identifier)
    if let Some(identifier) = &schema.identifier_column {
        columns.push(format!(
            "{} {}{}",
            quote_identifier(identifier),
            view_type(identifier, "TEXT"),
            key(identifier)
        ));
    }

    // Data column (JSONB read model)
    if let Some(data) = &schema.data_column {
        columns.push(format!("{} JSONB", quote_identifier(data)));
    }

    // Foreign key columns (for lineage tracking)
    for fk in &schema.fk_columns {
        columns.push(format!("{} BIGINT{}", quote_identifier(fk), key(fk)));
    }

    // UUID foreign key columns (for filtering): a column named `*_id` may be TEXT.
    for uuid_fk in &schema.uuid_fk_columns {
        columns.push(format!(
            "{} {}{}",
            quote_identifier(uuid_fk),
            view_type(uuid_fk, "UUID"),
            key(uuid_fk)
        ));
    }

    for (col_name, col_type) in &schema.additional_columns_with_types {
        columns.push(format!(
            "{} {}{}",
            quote_identifier(col_name),
            view_type(col_name, col_type),
            key(col_name)
        ));
    }
    if !columns.iter().any(|c| c.ends_with(" PRIMARY KEY")) {
        return Err(TViewError::InvalidInput {
            parameter: "tview definition".to_string(),
            reason: format!("{tview_name} has no column {identity} to key its rows on"),
        });
    }

    // Add timestamps for tracking
    columns.push("created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()".to_string());
    columns.push("updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()".to_string());

    let columns_sql = columns.join(",\n    ");

    let unlogged_keyword = if storage.logged { "" } else { "UNLOGGED " };
    let with = storage_clause(storage.fillfactor);
    let create_table_sql = format!(
        "CREATE {unlogged_keyword}TABLE {qi_schema}.{qi_tview} (\n    {columns_sql}\n){with}"
    );

    crate::utils::spi_run_ddl(&create_table_sql).map_err(|e| TViewError::SpiError {
        query: create_table_sql,
        error: e,
    })?;

    // Create indexes for performance
    create_tview_indexes(tview_name, schema, schema_name, storage.data_gin_index)?;

    Ok(())
}

/// Deterministic index name `idx_<tview>_<suffix>`, fitted to 63 bytes by
/// [`crate::utils::fit_identifier`].
pub(crate) fn index_name(tview_name: &str, suffix: &str) -> String {
    crate::utils::fit_identifier(format!("idx_{tview_name}_{suffix}"))
}

/// DDL for the required propagation index `(fk, pk)` on a TVIEW.
///
/// Cascade propagation (`src/propagate.rs`) looks up parent rows with
/// `SELECT fk, pk FROM tv WHERE fk = ANY($1)`; this index makes that lookup
/// index-only instead of a scan of the whole TVIEW.
pub(crate) fn propagation_index_ddl(
    schema_name: &str,
    tview_name: &str,
    fk: &str,
    pk: &str,
) -> String {
    index_ddl(
        schema_name,
        tview_name,
        &format!("{fk}_{pk}"),
        "",
        &[fk, pk],
    )
}

/// `CREATE INDEX IF NOT EXISTS idx_<tview>_<suffix> ON schema.tview <method>(cols)`.
pub(crate) fn index_ddl(
    schema_name: &str,
    tview_name: &str,
    suffix: &str,
    method: &str,
    columns: &[&str],
) -> String {
    let cols = columns
        .iter()
        .map(|c| quote_identifier(c))
        .collect::<Vec<_>>()
        .join(", ");
    format!(
        "CREATE INDEX IF NOT EXISTS {} ON {}.{} {method}({cols})",
        quote_identifier(&index_name(tview_name, suffix)),
        quote_identifier(schema_name),
        quote_identifier(tview_name),
    )
}

/// Create indexes on the materialized table for optimal query performance
fn create_tview_indexes(
    tview_name: &str,
    schema: &TViewSchema,
    schema_name: &str,
    data_gin: bool,
) -> TViewResult<()> {
    let ddl = tview_index_ddl(tview_name, schema, schema_name, data_gin);
    for create_idx in ddl {
        crate::utils::spi_run_ddl(&create_idx).map_err(|e| TViewError::SpiError {
            query: create_idx.clone(),
            error: e,
        })?;
    }
    Ok(())
}

/// Names of the indexes `pg_tviews` creates on `tview_name` for `schema` (the
/// `data` GIN index included) and for the columns joined to the aggregate TVIEWs it
/// embeds: the indexes a rebuild does not carry over as a user's (issue #134).
pub(crate) fn managed_index_names(
    tview_name: &str,
    schema: &TViewSchema,
    embed_columns: &[String],
) -> std::collections::HashSet<String> {
    let mut names = std::collections::HashSet::new();
    if let Some(id) = &schema.id_column {
        names.insert(index_name(tview_name, id));
    }
    for uuid_fk in &schema.uuid_fk_columns {
        names.insert(index_name(tview_name, uuid_fk));
    }
    if let Some(pk) = &schema.pk_column {
        for column in schema.fk_columns.iter().chain(embed_columns) {
            if column != pk {
                names.insert(index_name(tview_name, &format!("{column}_{pk}")));
            }
        }
    }
    if let Some(data) = &schema.data_column {
        names.insert(index_name(tview_name, &format!("{data}_gin")));
    }
    names
}

/// DDL for every index a new TVIEW gets.
///
/// HOT invariant: refreshes rewrite `data` and `updated_at`, so neither is indexed
/// (the `data` GIN only when `data_gin` is explicitly requested). An index on a
/// rewritten column makes every refresh a non-HOT update: new entries in every
/// index, a dead tuple needing index cleanup, and a cleared visibility-map bit.
fn tview_index_ddl(
    tview_name: &str,
    schema: &TViewSchema,
    schema_name: &str,
    data_gin: bool,
) -> Vec<String> {
    let mut ddl = Vec::new();

    // Trinity identifier and UUID foreign keys (filtering by public id)
    if let Some(id) = &schema.id_column {
        ddl.push(index_ddl(schema_name, tview_name, id, "", &[id]));
    }
    for uuid_fk in &schema.uuid_fk_columns {
        ddl.push(index_ddl(schema_name, tview_name, uuid_fk, "", &[uuid_fk]));
    }

    // Required propagation indexes (see `propagation_index_ddl`)
    if let Some(pk) = &schema.pk_column {
        for fk in schema.fk_columns.iter().filter(|fk| *fk != pk) {
            ddl.push(propagation_index_ddl(schema_name, tview_name, fk, pk));
        }
    }

    // Opt-in (pg_tviews.data_gin_index): top-level containment queries on data
    if data_gin && let Some(data) = &schema.data_column {
        ddl.push(index_ddl(
            schema_name,
            tview_name,
            &format!("{data}_gin"),
            "USING GIN ",
            &[data],
        ));
    }

    ddl
}

/// `WITH (fillfactor = N)` for a new TVIEW table; empty at 100 (the heap default),
/// so opting out yields the same DDL as before the setting existed.
fn storage_clause(fillfactor: i32) -> String {
    if fillfactor < 100 {
        format!(" WITH (fillfactor = {fillfactor})")
    } else {
        String::new()
    }
}

/// Populate the materialized table with initial data from the backing view, and
/// return the number of rows.
fn populate_initial_data(tview_name: &str, view_name: &str, schema_name: &str) -> TViewResult<u64> {
    // Get actual column names from the backing view (like pg_tviews_refresh does)
    // This ensures consistency and handles any discrepancies between inferred schema and actual view
    let view_oid = Spi::get_one::<Oid>(&format!(
        "SELECT c.oid FROM pg_class c JOIN pg_namespace n ON c.relnamespace = n.oid \
         WHERE c.relname::text = '{view_name}' AND n.nspname::text = '{schema_name}' AND c.relkind = 'v'"
    ))?
    .ok_or_else(|| TViewError::CatalogError {
        operation: format!("Find view {view_name} in schema {schema_name}"),
        pg_error: "View not found".to_string(),
    })?;

    let view_columns = crate::utils::get_view_columns_by_oid(view_oid)?;

    if view_columns.is_empty() {
        return Err(TViewError::CatalogError {
            operation: format!("Get columns for view {view_name}"),
            pg_error: "View has no selectable columns".to_string(),
        });
    }

    // Use the actual view columns for both insert and select
    let insert_columns = view_columns;

    let qi_schema = quote_identifier(schema_name);
    let qi_tview = quote_identifier(tview_name);
    let qi_view = quote_identifier(view_name);
    let col_list = insert_columns
        .iter()
        .map(|c| quote_identifier(c))
        .collect::<Vec<_>>()
        .join(", ");

    let insert_sql = format!(
        "INSERT INTO {qi_schema}.{qi_tview} ({col_list}) \
         SELECT {col_list} FROM {qi_schema}.{qi_view}"
    );

    let rows = Spi::connect_mut(|client| client.update(&insert_sql, None, &[]).map(|t| t.len()))
        .map_err(|e| TViewError::SpiError {
            query: insert_sql,
            error: e.to_string(),
        })?;

    Ok(rows as u64)
}

/// Quote a string for use in a `PostgreSQL` array literal.
///
/// Empty strings and strings containing special characters must be double-quoted
/// to avoid producing invalid array literals like `'{,}'`.
fn pg_array_elem(s: &str) -> String {
    if s.is_empty() || s.contains([',', '"', '\\', '{', '}', ' ']) {
        format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
    } else {
        s.to_string()
    }
}

/// Register the TVIEW in metadata tables
#[allow(clippy::too_many_arguments)] // Reason: all args are distinct registration fields with no natural grouping
fn register_metadata(
    entity_name: &str,
    view_name: &str,
    tview_name: &str,
    definition_sql: &str,
    schema: &TViewSchema,
    cascade_paths: &[cascade_path::CascadePath],
    schema_name: &str,
    group_keys: Option<&super::aggregate::GroupKeys>,
    uncascaded: &Uncascaded,
    key_mappings: &serde_json::Value,
    lineage: &crate::lineage::Lineage,
    replace: bool,
) -> TViewResult<()> {
    let identity = &lineage.identity;
    // A set operation (UNION, INTERSECT, EXCEPT): its rows are recomputed, never
    // patched.
    let is_union = lineage.set_operation;

    // Analyze dependencies to populate type/path/match_key info
    let dep_infos = analyze_dependencies(definition_sql, &schema.fk_columns);

    // Aggregate TVIEWs embedded through a join on their key (issue #126).
    let aggregate_embeds = aggregate_embeds(definition_sql, entity_name)?;
    create_embed_lookup_indexes(&aggregate_embeds, schema, tview_name, schema_name)?;

    // Extract the direct-patch column→key map (issue #56): base columns that map
    // identity-style to top-level keys of this entity's own `data` object. Empty
    // ⇒ the direct-patch fast path never engages for this entity.
    // A virtual generated column and its inputs are never patched from the row
    // (#179): the trigger sees the column as NULL, and a patched input would leave
    // the column computed from it stale.
    let root_table = format!("tb_{entity_name}");
    let virtual_reads: Vec<&String> = lineage
        .tables
        .iter()
        .filter(|t| t.relname == root_table)
        .flat_map(|t| &t.virtual_reads)
        .collect();
    let direct_map: Vec<(String, String)> = extract_direct_column_map(definition_sql, &root_table)
        .into_iter()
        .filter(|(col, _)| !virtual_reads.contains(&col))
        .collect();
    let direct_map_columns = direct_map
        .iter()
        .map(|(col, _)| pg_array_elem(col))
        .collect::<Vec<_>>()
        .join(",");
    let direct_map_keys = direct_map
        .iter()
        .map(|(_, key)| pg_array_elem(key))
        .collect::<Vec<_>>()
        .join(",");

    // Serialize schema information (quoted for safe PostgreSQL array literals)
    let fk_columns = schema
        .fk_columns
        .iter()
        .map(|s| pg_array_elem(s))
        .collect::<Vec<_>>()
        .join(",");
    let uuid_fk_columns = schema
        .uuid_fk_columns
        .iter()
        .map(|s| pg_array_elem(s))
        .collect::<Vec<_>>()
        .join(",");

    // Serialize dependency types
    let dep_types = dep_infos
        .iter()
        .map(|d| pg_array_elem(d.dep_type.as_str()))
        .collect::<Vec<_>>()
        .join(",");

    // Serialize dependency paths (TEXT[] format, empty string for None)
    let dep_paths = dep_infos
        .iter()
        .map(|d| {
            pg_array_elem(
                &d.jsonb_path
                    .as_ref()
                    .map_or_else(String::new, |path| path.join(".")),
            )
        })
        .collect::<Vec<_>>()
        .join(",");

    // Serialize array match keys (empty string for None)
    let array_keys = dep_infos
        .iter()
        .map(|d| pg_array_elem(&d.array_match_key.clone().unwrap_or_default()))
        .collect::<Vec<_>>()
        .join(",");

    // Serialize cascade paths as a TEXT[] array of JSON strings
    let cascade_paths_str = cascade_paths
        .iter()
        .map(|path| {
            let json = serde_json::to_string(path).expect("Failed to serialize cascade path");
            pg_array_elem(&json)
        })
        .collect::<Vec<_>>()
        .join(",");
    let cascade_paths_literal = format!("'{{{cascade_paths_str}}}'");

    // Get OIDs for the created objects (schema-qualified, parameterized to prevent injection)
    let view_oid_args = vec![
        unsafe { DatumWithOid::new(view_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
        unsafe { DatumWithOid::new(schema_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
    ];
    let view_oid_result = Spi::get_one_with_args::<pg_sys::Oid>(
        "SELECT c.oid FROM pg_class c \
         JOIN pg_namespace n ON c.relnamespace = n.oid \
         WHERE c.relname = $1 AND n.nspname = $2 AND c.relkind = 'v'",
        &view_oid_args,
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Get OID for view {schema_name}.{view_name}"),
        pg_error: e.to_string(),
    })?;

    let table_oid_args = vec![
        unsafe { DatumWithOid::new(tview_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
        unsafe { DatumWithOid::new(schema_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
    ];
    let table_oid_result = Spi::get_one_with_args::<pg_sys::Oid>(
        "SELECT c.oid FROM pg_class c \
         JOIN pg_namespace n ON c.relnamespace = n.oid \
         WHERE c.relname = $1 AND n.nspname = $2 AND c.relkind = 'r'",
        &table_oid_args,
    )
    .map_err(|e| TViewError::CatalogError {
        operation: format!("Get OID for table {schema_name}.{tview_name}"),
        pg_error: e.to_string(),
    })?;

    let view_oid = view_oid_result.ok_or_else(|| TViewError::CatalogError {
        operation: format!("Find view {schema_name}.{view_name}"),
        pg_error: "View OID not found".to_string(),
    })?;

    let table_oid = table_oid_result.ok_or_else(|| TViewError::CatalogError {
        operation: format!("Find table {schema_name}.{tview_name}"),
        pg_error: "Table OID not found".to_string(),
    })?;

    // A re-registration (after a column rename, or by pg_tviews_reregister)
    // replaces every derived column and keeps created_at, graphql_typename and
    // needs_reregister: only pg_tviews_reregister, which also re-installs the
    // triggers, clears the flag.
    let on_conflict = if replace {
        "ON CONFLICT (entity) DO UPDATE SET \
            view_oid = EXCLUDED.view_oid, table_oid = EXCLUDED.table_oid, \
            definition = EXCLUDED.definition, cascade_paths = EXCLUDED.cascade_paths, \
            fk_columns = EXCLUDED.fk_columns, uuid_fk_columns = EXCLUDED.uuid_fk_columns, \
            dependency_types = EXCLUDED.dependency_types, \
            dependency_paths = EXCLUDED.dependency_paths, \
            array_match_keys = EXCLUDED.array_match_keys, \
            distinct_on_keys = '{}', distinct_on_output_keys = '{}', \
            direct_map_columns = EXCLUDED.direct_map_columns, \
            direct_map_keys = EXCLUDED.direct_map_keys, is_union = EXCLUDED.is_union, \
            group_keys = EXCLUDED.group_keys, aggregate_embeds = EXCLUDED.aggregate_embeds, \
            uncascaded_oids = EXCLUDED.uncascaded_oids, key_mappings = EXCLUDED.key_mappings, \
            identity = EXCLUDED.identity"
    } else {
        "ON CONFLICT (entity) DO NOTHING"
    };

    // Insert metadata record (entity + definition parameterized; OIDs and array literals are safe internal values)
    let meta = crate::utils::meta_table();
    let insert_meta_sql = format!(
        "INSERT INTO {meta} (
            entity,
            view_oid,
            table_oid,
            definition,
            cascade_paths,
            fk_columns,
            uuid_fk_columns,
            dependency_types,
            dependency_paths,
            array_match_keys,
            direct_map_columns,
            direct_map_keys,
            is_union,
            group_keys,
            aggregate_embeds,
            uncascaded_oids,
            uncascaded_policy,
            key_mappings,
            identity
        ) VALUES ($1, {}, {}, $2, {}, '{{{}}}', '{{{}}}', '{{{}}}', '{{{}}}', '{{{}}}', '{{{}}}', '{{{}}}', {}, $3, $4, $5::pg_catalog.oid[]::pg_catalog.regclass[], $6, $7,
                  pg_catalog.jsonb_build_object('kind', $8::pg_catalog.text, 'columns',
                      pg_catalog.jsonb_build_array(pg_catalog.jsonb_build_object(
                          'name', $9::pg_catalog.text,
                          'type', pg_catalog.format_type($10::pg_catalog.oid, NULL)))))
        {on_conflict}",
        view_oid.to_u32(),
        table_oid.to_u32(),
        cascade_paths_literal,
        fk_columns,
        uuid_fk_columns,
        dep_types,
        dep_paths,
        array_keys,
        direct_map_columns,
        direct_map_keys,
        is_union
    );

    let group_keys_json =
        group_keys.map(|keys| pgrx::JsonB(serde_json::to_value(keys).unwrap_or_default()));
    let args = [
        unsafe { DatumWithOid::new(entity_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value()) },
        unsafe {
            DatumWithOid::new(
                definition_sql,
                PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                group_keys_json,
                PgOid::BuiltIn(PgBuiltInOids::JSONBOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                pgrx::JsonB(serde_json::to_value(&aggregate_embeds).unwrap_or_default()),
                PgOid::BuiltIn(PgBuiltInOids::JSONBOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                uncascaded.oids(),
                PgOid::BuiltIn(PgBuiltInOids::OIDARRAYOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                uncascaded.policy.as_str(),
                PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                pgrx::JsonB(key_mappings.clone()),
                PgOid::BuiltIn(PgBuiltInOids::JSONBOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                identity.kind.name(),
                PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                identity.name.as_str(),
                PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
            )
        },
        unsafe {
            DatumWithOid::new(
                pg_sys::Oid::from(identity.type_oid),
                PgOid::BuiltIn(PgBuiltInOids::OIDOID).value(),
            )
        },
    ];
    // The catalog is written as the extension's owner; the caller's right to
    // change this TVIEW was checked before (issue #134).
    let _owner = crate::owner::AsOwner::of_extension()?;
    Spi::run_with_args(&insert_meta_sql, &args).map_err(|e| TViewError::SpiError {
        query: insert_meta_sql,
        error: e.to_string(),
    })?;

    Ok(())
}

/// The aggregate TVIEWs (issue #58) a definition embeds, each mapped to the output
/// column that carries the value joined to the aggregate's `pk_<aggregate>`
/// (issue #126). An aggregate has no `fk_<aggregate>` column to propagate by, so a
/// change to group `k` refreshes the rows whose column equals `k`.
///
/// # Errors
/// Rejects a definition that reads an aggregate TVIEW without projecting the
/// column joined to its key: such a TVIEW could never be refreshed when the
/// aggregate changes.
fn aggregate_embeds(
    definition_sql: &str,
    entity_name: &str,
) -> TViewResult<std::collections::BTreeMap<String, String>> {
    let aggregates: Vec<String> = Spi::connect(|client| {
        let args = [unsafe {
            DatumWithOid::new(entity_name, PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value())
        }];
        client
            .select(
                &format!(
                    "SELECT entity FROM {} WHERE group_keys IS NOT NULL AND entity <> $1 \
                     ORDER BY entity",
                    crate::utils::meta_table()
                ),
                None,
                &args,
            )?
            .map(|row| row["entity"].value::<String>())
            .filter_map(Result::transpose)
            .collect::<Result<_, _>>()
    })
    .map_err(|e| TViewError::CatalogError {
        operation: "List aggregate TVIEWs".to_string(),
        pg_error: e.to_string(),
    })?;
    if aggregates.is_empty() {
        return Ok(std::collections::BTreeMap::new());
    }

    let lookups =
        crate::sql_parser::embed_lookup_columns(definition_sql, &aggregates).map_err(|reason| {
            TViewError::InvalidInput {
                parameter: "tview definition".to_string(),
                reason,
            }
        })?;
    let mut embeds = std::collections::BTreeMap::new();
    for (aggregate, column) in lookups {
        let Some(column) = column else {
            return Err(TViewError::InvalidInput {
                parameter: "tview definition".to_string(),
                reason: format!(
                    "TVIEW 'tv_{entity_name}' reads aggregate TVIEW '{aggregate}' but no output \
                     column carries the value it is joined to on pk_{aggregate}, so a change to \
                     a '{aggregate}' group could not be routed to the rows embedding it. Join \
                     v_{aggregate} with an equality on its key in the FROM clause (e.g. `LEFT \
                     JOIN v_{aggregate} a ON a.pk_{aggregate} = t.pk_{entity_name}`) and \
                     project the other side of that equality."
                ),
            });
        };
        embeds.insert(aggregate, column);
    }
    Ok(embeds)
}

/// Index each aggregate-embed lookup column that is neither the TVIEW's primary
/// key nor already indexed as an `fk_*` propagation column, so propagation from
/// the aggregate does not scan the whole TVIEW.
fn create_embed_lookup_indexes(
    embeds: &std::collections::BTreeMap<String, String>,
    schema: &TViewSchema,
    tview_name: &str,
    schema_name: &str,
) -> TViewResult<()> {
    let Some(pk) = &schema.pk_column else {
        return Ok(());
    };
    let columns: std::collections::BTreeSet<&String> = embeds
        .values()
        .filter(|c| *c != pk && !schema.fk_columns.contains(c))
        .collect();
    for column in columns {
        let ddl = propagation_index_ddl(schema_name, tview_name, column, pk);
        crate::utils::spi_run_ddl(&ddl).map_err(|e| TViewError::SpiError {
            query: ddl.clone(),
            error: e,
        })?;
    }
    Ok(())
}

/// Transform a raw SELECT statement into TVIEW format
///
/// Takes a simple SELECT like "SELECT id, name, price FROM `tb_product`"
/// and transforms it into a proper TVIEW format with:
/// - pk_<entity> column (generated from the source table's primary key or id column)
/// - id column (UUID, generated from the source table's primary key)
/// - data column (JSONB with all fields)
///
/// This creates a "prepared view" that wraps the raw SELECT with TVIEW conventions.
fn transform_raw_select_to_tview(
    entity_name: &str,
    select_sql: &str,
) -> TViewResult<(String, TViewSchema)> {
    // Create a temporary view to analyze the raw SELECT
    let temp_view_name = format!("_temp_raw_{entity_name}");
    let qi_temp_view = quote_identifier(&temp_view_name);

    // First, create temp view to analyze columns
    let create_temp = format!("CREATE TEMP VIEW {qi_temp_view} AS {select_sql}");

    crate::utils::spi_run_ddl(&create_temp).map_err(|e| TViewError::SpiError {
        query: create_temp.clone(),
        error: e,
    })?;

    // Get columns from temp view (parameterized lookup)
    // Cast to text to avoid sql_identifier domain type issues
    let get_columns_sql = "SELECT column_name::text, data_type::text
         FROM information_schema.columns
         WHERE table_name = $1
         ORDER BY ordinal_position";

    let temp_view_args = vec![unsafe {
        DatumWithOid::new(
            temp_view_name.as_str(),
            PgOid::BuiltIn(PgBuiltInOids::TEXTOID).value(),
        )
    }];
    let columns: Vec<(String, String)> = Spi::connect(|client| {
        let rows = client.select(get_columns_sql, None, &temp_view_args)?;
        let mut result = Vec::new();
        for row in rows {
            let col_name: String = row[1].value()?.ok_or_else(|| {
                spi::Error::from(crate::TViewError::SpiError {
                    query: get_columns_sql.to_string(),
                    error: "column name is NULL".to_string(),
                })
            })?;
            let data_type: String = row[2].value()?.ok_or_else(|| {
                spi::Error::from(crate::TViewError::SpiError {
                    query: get_columns_sql.to_string(),
                    error: "data type is NULL".to_string(),
                })
            })?;
            result.push((col_name, data_type));
        }
        Ok(result)
    })
    .map_err(|e: spi::Error| TViewError::CatalogError {
        operation: "Get columns from temp view".to_string(),
        pg_error: format!("{e:?}"),
    })?;

    // Drop temp view
    crate::utils::spi_run_ddl(&format!("DROP VIEW {qi_temp_view}")).ok();

    // Find primary key column using this priority order:
    // 1. Column named exactly 'pk' (original source PK, most reliable)
    // 2. Column that is integer/serial type (natural database PKs)
    // 3. Column named 'id' (may be UUID identifier, fallback only)
    // This avoids incorrectly selecting a UUID 'id' column when an actual 'pk' exists.
    let pk_source_col = columns
        .iter()
        .find(|(name, _)| name == "pk")
        .or_else(|| {
            columns
                .iter()
                .find(|(_, typ)| typ.contains("int") || typ.contains("serial"))
        })
        .or_else(|| columns.iter().find(|(name, _)| name == "id"))
        .map(|(name, _)| name.clone())
        .ok_or_else(|| TViewError::InvalidSelectStatement {
            sql: select_sql.to_string(),
            reason: "No suitable primary key column found (need 'pk', an integer column, or 'id')"
                .to_string(),
        })?;

    // Build explicit column lists for clarity and control

    // 1. Build the source column list (from the subquery)
    let _source_columns: Vec<String> = columns
        .iter()
        .map(|(name, _)| format!("source.{name}"))
        .collect();

    // 2. Build JSONB data column pairs explicitly
    let data_columns: Vec<String> = columns
        .iter()
        .map(|(name, _)| format!("'{name}', source.{name}"))
        .collect();

    // 3. Generate transformed SELECT with explicit column references
    // This makes it clear exactly what's being selected and how it's transformed
    let transformed_select = format!(
        "SELECT
            source.{} AS pk_{},
            gen_random_uuid() AS id,
            jsonb_build_object({}) AS data
        FROM ({}) AS source",
        pk_source_col,
        entity_name,
        data_columns.join(", "),
        select_sql
    );

    // Infer schema from transformed SELECT
    let schema = infer_schema(&transformed_select)?;

    Ok((transformed_select, schema))
}

#[cfg(any(test, feature = "pg_test"))]
#[pgrx::pg_schema]
mod tests {
    use pgrx::prelude::*;

    // ── Unit tests for index set / storage (no database required) ──────────────

    fn post_schema() -> crate::schema::TViewSchema {
        crate::schema::TViewSchema {
            pk_column: Some("pk_post".to_string()),
            id_column: Some("id".to_string()),
            data_column: Some("data".to_string()),
            fk_columns: vec!["fk_user".to_string()],
            uuid_fk_columns: vec!["user_id".to_string()],
            ..Default::default()
        }
    }

    #[test]
    fn test_index_ddl_never_indexes_rewritten_columns_by_default() {
        let ddl = super::tview_index_ddl("tv_post", &post_schema(), "public", false);
        assert_eq!(ddl.len(), 3, "{ddl:?}");
        for stmt in &ddl {
            assert!(!stmt.contains("\"data\""), "indexes data: {stmt}");
            assert!(!stmt.contains("updated_at"), "indexes updated_at: {stmt}");
            assert!(!stmt.contains("GIN"), "creates a GIN: {stmt}");
        }
    }

    #[test]
    fn test_index_ddl_gin_only_when_requested() {
        let ddl = super::tview_index_ddl("tv_post", &post_schema(), "public", true);
        assert_eq!(
            ddl.iter()
                .filter(|s| s.contains("USING GIN (\"data\")"))
                .count(),
            1
        );
        assert!(ddl.iter().all(|s| !s.contains("updated_at")));
    }

    #[test]
    fn test_storage_clause() {
        assert_eq!(super::storage_clause(85), " WITH (fillfactor = 85)");
        assert_eq!(super::storage_clause(100), "");
    }

    // ── Unit tests for index naming (no database required) ─────────────────────

    #[test]
    fn test_index_name_short_is_verbatim() {
        assert_eq!(
            super::index_name("tv_post", "fk_user_pk_post"),
            "idx_tv_post_fk_user_pk_post"
        );
    }

    #[test]
    fn test_index_name_long_fits_and_stays_unique() {
        let entity = "a".repeat(60);
        let a = super::index_name(&format!("tv_{entity}"), "fk_left_pk_x");
        let b = super::index_name(&format!("tv_{entity}"), "fk_right_pk_x");
        assert_eq!(a.len(), crate::utils::MAX_IDENTIFIER_BYTES);
        assert_ne!(a, b);
        assert_eq!(
            a,
            super::index_name(&format!("tv_{entity}"), "fk_left_pk_x")
        );
    }

    #[test]
    fn test_index_name_truncates_on_char_boundary() {
        let name = super::index_name(&format!("tv_{}", "é".repeat(40)), "fk_x_pk_y");
        assert!(name.len() <= crate::utils::MAX_IDENTIFIER_BYTES);
    }

    #[test]
    fn test_propagation_index_ddl() {
        assert_eq!(
            super::propagation_index_ddl("public", "tv_post", "fk_user", "pk_post"),
            "CREATE INDEX IF NOT EXISTS \"idx_tv_post_fk_user_pk_post\" \
             ON \"public\".\"tv_post\" (\"fk_user\", \"pk_post\")"
        );
    }

    // ── Integration tests requiring database access ───────────────────────────

    #[test]
    fn test_tview_exists_non_existent() {
        // Compile-time check only — live DB tests use #[pg_test] below
    }

    /// TVIEW objects are created in the schema that is first in `search_path`,
    /// not hardcoded to public.
    #[pg_test]
    fn test_create_tview_respects_search_path() {
        Spi::run("CREATE SCHEMA tview_test_ns").unwrap();
        Spi::run("SET search_path TO tview_test_ns, public").unwrap();
        Spi::run("CREATE TABLE tb_item (pk_item BIGSERIAL PRIMARY KEY, name TEXT)").unwrap();
        Spi::run("INSERT INTO tb_item VALUES (1, 'Widget')").unwrap();

        Spi::run(
            "SELECT pg_tviews_create('item', $$
            SELECT pk_item, jsonb_build_object('name', name) AS data
            FROM tb_item
        $$)",
        )
        .unwrap();

        // tv_item must be in the target schema
        let in_target = Spi::get_one::<bool>(
            "SELECT COUNT(*) > 0 FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_item' AND n.nspname = 'tview_test_ns'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(in_target, "tv_item should be in tview_test_ns, not public");

        // tv_item must NOT leak into public
        let in_public = Spi::get_one::<bool>(
            "SELECT COUNT(*) > 0 FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_item' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(!in_public, "tv_item must not be created in public schema");

        // The backing view v_item must be in the same schema
        let view_in_target = Spi::get_one::<bool>(
            "SELECT COUNT(*) > 0 FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'v_item' AND n.nspname = 'tview_test_ns'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(view_in_target, "v_item should be in tview_test_ns");
    }

    /// With the default `search_path`, objects still land in public (regression guard).
    #[pg_test]
    fn test_create_tview_defaults_to_public() {
        Spi::run("SET search_path TO public").unwrap();
        Spi::run("CREATE TABLE tb_gadget (pk_gadget BIGSERIAL PRIMARY KEY, label TEXT)").unwrap();
        Spi::run("INSERT INTO tb_gadget VALUES (1, 'Gizmo')").unwrap();

        Spi::run(
            "SELECT pg_tviews_create('gadget', $$
            SELECT pk_gadget, jsonb_build_object('label', label) AS data
            FROM tb_gadget
        $$)",
        )
        .unwrap();

        let in_public = Spi::get_one::<bool>(
            "SELECT COUNT(*) > 0 FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_gadget' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(
            in_public,
            "tv_gadget should be in public with default search_path"
        );
    }

    /// Test CTAS (CREATE TABLE AS SELECT) with preexisting data.
    /// This reproduces the bug where initial population fails.
    #[pg_test]
    fn test_ctas_with_preexisting_data() {
        Spi::run("SET search_path TO public").unwrap();

        // Create base table with data
        Spi::run("CREATE TABLE tb_ctas_test (pk_test BIGSERIAL PRIMARY KEY, name TEXT)").unwrap();
        Spi::run("INSERT INTO tb_ctas_test VALUES (1, 'Alice'), (2, 'Bob')").unwrap();

        // Do CTAS - this should create a TVIEW with the existing data
        Spi::run(
            "CREATE TABLE tv_ctas_test AS
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_ctas_test",
        )
        .unwrap();

        // Check that TVIEW was created
        let tview_exists = Spi::get_one::<bool>(
            "SELECT COUNT(*) > 0 FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_ctas_test' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(tview_exists, "tv_ctas_test should exist");

        // Check that it has the initial data (this is where the bug manifests)
        let row_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_ctas_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            row_count, 2,
            "tv_ctas_test should have 2 rows from initial population"
        );

        // Check specific data
        let alice_exists = Spi::get_one::<bool>(
            "SELECT COUNT(*) > 0 FROM tv_ctas_test WHERE data->>'name' = 'Alice'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(alice_exists, "Alice should be in the TVIEW");
    }

    /// Test that TVIEW tables respect the `unlogged_by_default` GUC.
    #[pg_test]
    fn test_tview_unlogged_guc_control() {
        Spi::run("SET search_path TO public").unwrap();

        // Test with GUC set to true (default)
        Spi::run("SET pg_tviews.unlogged_by_default TO true").unwrap();

        // Create base table
        Spi::run("CREATE TABLE tb_guc_test1 (pk_test BIGSERIAL PRIMARY KEY, name TEXT)").unwrap();

        // Create TVIEW
        Spi::run(
            "SELECT pg_tviews_create('guc_test1', $$
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_guc_test1
        $$)",
        )
        .unwrap();

        // Check that the TVIEW table is UNLOGGED
        let is_unlogged = Spi::get_one::<bool>(
            "SELECT c.relpersistence = 'u' FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_guc_test1' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(
            is_unlogged,
            "tv_guc_test1 should be UNLOGGED when GUC is true"
        );

        // Test with GUC set to false
        Spi::run("SET pg_tviews.unlogged_by_default TO false").unwrap();

        // Create another base table
        Spi::run("CREATE TABLE tb_guc_test2 (pk_test BIGSERIAL PRIMARY KEY, name TEXT)").unwrap();

        // Create another TVIEW
        Spi::run(
            "SELECT pg_tviews_create('guc_test2', $$
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_guc_test2
        $$)",
        )
        .unwrap();

        // Check that this TVIEW table is LOGGED
        let is_logged = Spi::get_one::<bool>(
            "SELECT c.relpersistence = 'p' FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_guc_test2' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(is_logged, "tv_guc_test2 should be LOGGED when GUC is false");

        // Reset GUC to default
        Spi::run("RESET pg_tviews.unlogged_by_default").unwrap();
    }

    /// Test ALTER TABLE SET UNLOGGED/LOGGED on TVIEWs.
    #[pg_test]
    fn test_alter_tview_unlogged_logged() {
        Spi::run("SET search_path TO public").unwrap();

        // Create base table with data
        Spi::run("CREATE TABLE tb_alter_test (pk_test BIGSERIAL PRIMARY KEY, name TEXT)").unwrap();
        Spi::run("INSERT INTO tb_alter_test VALUES (1, 'Alice'), (2, 'Bob')").unwrap();

        // Create TVIEW as LOGGED first
        Spi::run("SET pg_tviews.unlogged_by_default TO false").unwrap();
        Spi::run(
            "SELECT pg_tviews_create('alter_test', $$
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_alter_test
        $$)",
        )
        .unwrap();

        // Verify TVIEW is initially LOGGED
        let is_logged = Spi::get_one::<bool>(
            "SELECT c.relpersistence = 'p' FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_alter_test' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(is_logged, "tv_alter_test should initially be LOGGED");

        // ALTER TABLE to UNLOGGED
        Spi::run("ALTER TABLE tv_alter_test SET UNLOGGED").unwrap();

        // Verify it's now UNLOGGED
        let is_unlogged = Spi::get_one::<bool>(
            "SELECT c.relpersistence = 'u' FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_alter_test' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(
            is_unlogged,
            "tv_alter_test should be UNLOGGED after ALTER TABLE"
        );

        // ALTER TABLE back to LOGGED
        Spi::run("ALTER TABLE tv_alter_test SET LOGGED").unwrap();

        // Verify it's now LOGGED again
        let is_logged_again = Spi::get_one::<bool>(
            "SELECT c.relpersistence = 'p' FROM pg_class c \
             JOIN pg_namespace n ON c.relnamespace = n.oid \
             WHERE c.relname = 'tv_alter_test' AND n.nspname = 'public'",
        )
        .unwrap()
        .unwrap_or(false);
        assert!(
            is_logged_again,
            "tv_alter_test should be LOGGED again after ALTER TABLE"
        );

        // Reset GUC
        Spi::run("RESET pg_tviews.unlogged_by_default").unwrap();
    }

    /// Test data integrity during ALTER TABLE UNLOGGED/LOGGED operations.
    #[pg_test]
    fn test_alter_tview_data_integrity() {
        Spi::run("SET search_path TO public").unwrap();

        // Create base table with data
        Spi::run("CREATE TABLE tb_integrity_test (pk_test BIGSERIAL PRIMARY KEY, name TEXT)")
            .unwrap();
        Spi::run("INSERT INTO tb_integrity_test VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie')")
            .unwrap();

        // Create TVIEW as LOGGED
        Spi::run("SET pg_tviews.unlogged_by_default TO false").unwrap();
        Spi::run(
            "SELECT pg_tviews_create('integrity_test', $$
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_integrity_test
        $$)",
        )
        .unwrap();

        // Verify data is present
        let initial_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_integrity_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(initial_count, 3, "TVIEW should have 3 rows initially");

        // ALTER TABLE from LOGGED to UNLOGGED - data should be preserved
        Spi::run("ALTER TABLE tv_integrity_test SET UNLOGGED").unwrap();

        let after_unlogged_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_integrity_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            after_unlogged_count, 3,
            "Data should be preserved when converting LOGGED to UNLOGGED"
        );

        // ALTER TABLE from UNLOGGED to LOGGED - data is preserved (PostgreSQL behavior)
        Spi::run("ALTER TABLE tv_integrity_test SET LOGGED").unwrap();

        let after_logged_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_integrity_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            after_logged_count, 3,
            "Data should be preserved when converting UNLOGGED to LOGGED"
        );

        // Reset GUC
        Spi::run("RESET pg_tviews.unlogged_by_default").unwrap();
    }

    /// Test detection of post-crash empty UNLOGGED table.
    #[pg_test]
    fn test_detect_post_crash_empty_tview() {
        Spi::run("SET search_path TO public").unwrap();

        // Create base table with data
        Spi::run("CREATE TABLE tb_crash_test (pk_test BIGSERIAL PRIMARY KEY, name TEXT)").unwrap();
        Spi::run("INSERT INTO tb_crash_test VALUES (1, 'Alice'), (2, 'Bob')").unwrap();

        // Create TVIEW
        Spi::run(
            "SELECT pg_tviews_create('crash_test', $$
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_crash_test
        $$)",
        )
        .unwrap();

        // Verify TVIEW has data
        let initial_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_crash_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(initial_count, 2, "TVIEW should have 2 rows initially");

        // Before truncation, should not detect crash
        let crash_before = crate::lifecycle::detect_post_crash_truncation("crash_test").unwrap();
        assert!(!crash_before, "Should not detect crash when table has data");

        // Simulate post-crash truncation (UNLOGGED table behavior)
        Spi::run("TRUNCATE TABLE tv_crash_test").unwrap();

        // Verify table is now empty
        let after_truncate_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_crash_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            after_truncate_count, 0,
            "TVIEW should be empty after truncate"
        );

        // Should now detect crash (table empty but view has data)
        let crash_detected = crate::lifecycle::detect_post_crash_truncation("crash_test").unwrap();
        assert!(
            crash_detected,
            "Should detect crash when UNLOGGED table is empty but view has data"
        );

        // Verify backing view still has data
        let view_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM v_crash_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            view_count, 2,
            "Backing view should still have data after table truncate"
        );
    }

    /// Test automatic recovery after crash detection.
    #[pg_test]
    fn test_auto_recover_after_crash() {
        Spi::run("SET search_path TO public").unwrap();

        // Create base table with data
        Spi::run("CREATE TABLE tb_recover_test (pk_test BIGSERIAL PRIMARY KEY, name TEXT)")
            .unwrap();
        Spi::run("INSERT INTO tb_recover_test VALUES (1, 'Alice'), (2, 'Bob')").unwrap();

        // Create TVIEW
        Spi::run(
            "SELECT pg_tviews_create('recover_test', $$
            SELECT pk_test, jsonb_build_object('name', name) AS data
            FROM tb_recover_test
        $$)",
        )
        .unwrap();

        // Verify TVIEW has data initially
        let initial_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_recover_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(initial_count, 2, "TVIEW should have 2 rows initially");

        // Simulate post-crash truncation
        Spi::run("TRUNCATE TABLE tv_recover_test").unwrap();

        // Verify table is now empty
        let after_truncate_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_recover_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            after_truncate_count, 0,
            "TVIEW should be empty after truncate"
        );

        // Call auto-recovery function
        let recovery_performed =
            Spi::get_one::<bool>("SELECT pg_tviews_recover_after_crash('recover_test')")
                .unwrap()
                .unwrap_or(false);
        assert!(
            recovery_performed,
            "Recovery should be performed when crash is detected"
        );

        // Verify TVIEW has data again after recovery
        let after_recovery_count = Spi::get_one::<i64>("SELECT COUNT(*) FROM tv_recover_test")
            .unwrap()
            .unwrap_or(0);
        assert_eq!(
            after_recovery_count, 2,
            "TVIEW should have 2 rows after recovery"
        );

        // Call recovery again - should return false (no recovery needed)
        let second_recovery =
            Spi::get_one::<bool>("SELECT pg_tviews_recover_after_crash('recover_test')")
                .unwrap()
                .unwrap_or(true);
        assert!(
            !second_recovery,
            "Second recovery call should return false when no crash detected"
        );
    }
}