pylon_storage/

postgres.rs

use crate::{FieldSpec, SchemaOperation, SchemaPlan, StorageAdapter, StorageError};
use pylon_kernel::AppManifest;

// ---------------------------------------------------------------------------
// Type mapping: manifest field types -> PostgreSQL column types
//
//   string    -> TEXT
//   int       -> INTEGER
//   float     -> DOUBLE PRECISION
//   bool      -> BOOLEAN
//   datetime  -> TIMESTAMPTZ
//   richtext  -> TEXT
//   id(...)   -> TEXT
// ---------------------------------------------------------------------------

fn pg_column_type(field_type: &str) -> &'static str {
    match field_type {
        "string" => "TEXT",
        "int" => "INTEGER",
        "float" => "DOUBLE PRECISION",
        "bool" => "BOOLEAN",
        "datetime" => "TIMESTAMPTZ",
        "richtext" => "TEXT",
        _ if field_type.starts_with("id(") => "TEXT",
        _ => "TEXT",
    }
}

// ---------------------------------------------------------------------------
// Identifier quoting
// ---------------------------------------------------------------------------

/// Quote a SQL identifier, escaping embedded double-quotes by doubling them.
///
/// PostgreSQL standard: `"foo""bar"` represents the identifier `foo"bar`.
pub(crate) fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

/// Public re-export for sibling modules (e.g. `pg_tx_store`) that need
/// to build SQL strings against the same dialect rules. Keeping the
/// crate-private function untouched preserves the existing call sites
/// inside this module.
#[cfg(feature = "postgres-live")]
pub fn quote_ident_pub(name: &str) -> String {
    quote_ident(name)
}

/// Public re-export of [`live::row_to_json`] for sibling modules. Used
/// by `pg_tx_store` so transactional reads return rows in the same
/// JSON shape as the non-transactional path.
#[cfg(feature = "postgres-live")]
pub fn row_to_json_pub(row: &postgres::Row) -> serde_json::Value {
    live::row_to_json(row)
}

/// Public re-export of the filter builder so the in-tx PgTxStore can
/// reuse the same operator surface ($eq, $like, $in, $order, $limit,
/// $offset, $not, $gt, $gte, $lt, $lte) as the non-tx path. Returns
/// `(sql, params)` ready to feed to either `client.query` or
/// `transaction.query`.
#[cfg(feature = "postgres-live")]
pub fn build_query_filtered_sql_pub(
    entity: &str,
    filter: &serde_json::Value,
    valid_columns: &[String],
) -> Result<(String, Vec<JsonParam>), StorageError> {
    live::LivePostgresAdapter::build_query_filtered_sql(entity, filter, valid_columns)
}

/// Public re-export of the aggregate builder. Returns
/// `(sql, params, column_names)` — the column_names list drives the
/// per-row projection in `aggregate_rows_to_json_pub`.
#[cfg(feature = "postgres-live")]
pub fn build_aggregate_sql_pub(
    entity: &str,
    spec: &serde_json::Value,
    valid_columns: &[String],
) -> Result<(String, Vec<JsonParam>, Vec<String>), StorageError> {
    live::LivePostgresAdapter::build_aggregate_sql(entity, spec, valid_columns)
}

/// Public re-export of the post-processing helper that converts raw
/// aggregate `Row`s into the `{ rows: [{...}] }` JSON shape.
#[cfg(feature = "postgres-live")]
pub fn aggregate_rows_to_json_pub(
    rows: &[postgres::Row],
    column_names: &[String],
) -> serde_json::Value {
    live::aggregate_rows_to_json(rows, column_names)
}

// ---------------------------------------------------------------------------
// SQL generation
// ---------------------------------------------------------------------------

/// Generate a Postgres CREATE TABLE statement.
pub fn create_table_sql(entity_name: &str, fields: &[FieldSpec]) -> String {
    let mut columns = vec!["id TEXT PRIMARY KEY NOT NULL".to_string()];

    for field in fields {
        let col_type = pg_column_type(&field.field_type);
        let not_null = if field.optional { "" } else { " NOT NULL" };
        let unique = if field.unique { " UNIQUE" } else { "" };
        columns.push(format!(
            "{} {}{}{}",
            quote_ident(&field.name),
            col_type,
            not_null,
            unique
        ));
    }

    format!(
        "CREATE TABLE IF NOT EXISTS {} ({})",
        quote_ident(entity_name),
        columns.join(", ")
    )
}

/// Generate a Postgres ALTER TABLE ADD COLUMN statement.
/// NOT NULL is omitted on ADD COLUMN to avoid requiring DEFAULT values.
/// Required-ness is tracked in the manifest; enforcement deferred.
pub fn add_column_sql(entity_name: &str, field: &FieldSpec) -> String {
    let col_type = pg_column_type(&field.field_type);
    let unique = if field.unique { " UNIQUE" } else { "" };
    format!(
        "ALTER TABLE {} ADD COLUMN {} {}{}",
        quote_ident(entity_name),
        quote_ident(&field.name),
        col_type,
        unique
    )
}

/// Generate a Postgres CREATE INDEX statement.
pub fn create_index_sql(
    entity_name: &str,
    index_name: &str,
    fields: &[String],
    unique: bool,
) -> String {
    let unique_str = if unique { "UNIQUE " } else { "" };
    let full_index_name = format!("{}_{}", entity_name, index_name);
    let quoted_fields: Vec<String> = fields.iter().map(|f| quote_ident(f)).collect();
    format!(
        "CREATE {}INDEX IF NOT EXISTS {} ON {} ({})",
        unique_str,
        quote_ident(&full_index_name),
        quote_ident(entity_name),
        quoted_fields.join(", ")
    )
}

// ---------------------------------------------------------------------------
// PostgresAdapter — planning-only adapter
// ---------------------------------------------------------------------------

/// A Postgres storage adapter. Currently supports planning only.
/// No live connection — SQL generation and planning from manifest.
pub struct PostgresAdapter;

impl StorageAdapter for PostgresAdapter {
    fn plan_schema(&self, target: &AppManifest) -> Result<SchemaPlan, StorageError> {
        // Plan from empty baseline.
        let mut operations = Vec::new();

        for entity in &target.entities {
            let fields: Vec<FieldSpec> = entity
                .fields
                .iter()
                .map(|f| FieldSpec {
                    name: f.name.clone(),
                    field_type: f.field_type.clone(),
                    optional: f.optional,
                    unique: f.unique,
                })
                .collect();

            operations.push(SchemaOperation::CreateEntity {
                name: entity.name.clone(),
                fields,
            });

            for index in &entity.indexes {
                operations.push(SchemaOperation::AddIndex {
                    entity: entity.name.clone(),
                    name: index.name.clone(),
                    fields: index.fields.clone(),
                    unique: index.unique,
                });
            }
        }

        if operations.is_empty() {
            operations.push(SchemaOperation::Noop);
        }

        Ok(SchemaPlan { operations })
    }

    // apply_schema intentionally not implemented — uses default trait error.
}

/// Generate all SQL statements for a plan, in order.
/// Useful for dry-run preview of what Postgres DDL would be executed.
pub fn plan_to_sql(plan: &SchemaPlan) -> Result<Vec<String>, StorageError> {
    let mut statements = Vec::new();

    for op in &plan.operations {
        match op {
            SchemaOperation::CreateEntity { name, fields } => {
                statements.push(create_table_sql(name, fields));
            }
            SchemaOperation::AddField { entity, field } => {
                statements.push(add_column_sql(entity, field));
            }
            SchemaOperation::AlterField {
                entity,
                previous,
                target,
            } => {
                // Only nullable transitions today. SET / DROP NOT NULL is
                // safe on a populated table when going from required →
                // optional (existing rows already satisfy NOT NULL); the
                // reverse direction (optional → required) succeeds only
                // if every row has a non-null value, which the planner
                // has no way to know — Postgres will fail the migration
                // if it can't and the operator gets a clear error from
                // the apply step.
                if previous.optional && !target.optional {
                    statements.push(format!(
                        "ALTER TABLE {} ALTER COLUMN {} SET NOT NULL",
                        quote_ident(entity),
                        quote_ident(&target.name)
                    ));
                } else if !previous.optional && target.optional {
                    statements.push(format!(
                        "ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL",
                        quote_ident(entity),
                        quote_ident(&target.name)
                    ));
                }
                // Nothing emitted when neither nullable nor type changed
                // — falls through silently. AlterField with no actual
                // shape change shouldn't happen in practice (the planner
                // only emits it on real drift), but guard against
                // emitting empty SQL just in case.
            }
            SchemaOperation::AddIndex {
                entity,
                name,
                fields,
                unique,
            } => {
                statements.push(create_index_sql(entity, name, fields, *unique));
            }
            SchemaOperation::CreateSearchIndex { entity, config } => {
                #[cfg(feature = "postgres-live")]
                {
                    statements.extend(crate::pg_search::create_search_index_sql(entity, config));
                }
                #[cfg(not(feature = "postgres-live"))]
                {
                    let _ = (entity, config);
                    return Err(StorageError {
                        code: "PG_SEARCH_FEATURE_OFF".into(),
                        message: "CreateSearchIndex requires the `postgres-live` feature".into(),
                    });
                }
            }
            SchemaOperation::RemoveSearchIndex { entity } => {
                // Without the original config we don't know which
                // facet/sort indexes were created. Drop the FTS table
                // and the GIN index by their fixed names; per-field
                // facet/sort indexes are dropped automatically by the
                // entity DROP path. Operators removing search via
                // schema diff will see leftover indexes only if the
                // entity table itself still exists — at which point
                // the next CREATE INDEX IF NOT EXISTS catches up.
                //
                // quote_ident on the synthetic table/index names so a
                // malicious entity name with embedded `"` can't break
                // out of the identifier.
                statements.push(format!(
                    "DROP TABLE IF EXISTS {} CASCADE",
                    quote_ident(&format!("_fts_{entity}"))
                ));
                statements.push(format!(
                    "DROP INDEX IF EXISTS {}",
                    quote_ident(&format!("{entity}_fts_gin"))
                ));
            }
            SchemaOperation::Noop => {}
            other => {
                return Err(StorageError {
                    code: "PG_OP_UNSUPPORTED".into(),
                    message: format!("Operation not supported by Postgres adapter: {other:?}"),
                });
            }
        }
    }

    Ok(statements)
}

// ---------------------------------------------------------------------------
// Introspection SQL helpers
//
// These generate the SQL queries that a live Postgres connection would run
// to read the current schema. No connection required — just SQL strings.
// ---------------------------------------------------------------------------

/// SQL to list user tables in the public schema.
pub const INTROSPECT_TABLES_SQL: &str = "\
    SELECT table_name \
    FROM information_schema.tables \
    WHERE table_schema = 'public' \
      AND table_type = 'BASE TABLE' \
      AND table_name NOT LIKE '_pylon_%' \
    ORDER BY table_name";

/// SQL to list columns for a given table.
/// Use with parameter: table_name.
pub const INTROSPECT_COLUMNS_SQL: &str = "\
    SELECT column_name, data_type, is_nullable, \
           (SELECT COUNT(*) FROM information_schema.table_constraints tc \
            JOIN information_schema.key_column_usage kcu \
              ON tc.constraint_name = kcu.constraint_name \
            WHERE tc.table_name = c.table_name \
              AND kcu.column_name = c.column_name \
              AND tc.constraint_type = 'PRIMARY KEY') as is_pk \
    FROM information_schema.columns c \
    WHERE table_schema = 'public' AND table_name = $1 \
    ORDER BY ordinal_position";

/// SQL to list indexes for a given table.
/// Use with parameter: table_name.
pub const INTROSPECT_INDEXES_SQL: &str = "\
    SELECT i.relname as index_name, \
           ix.indisunique as is_unique, \
           array_agg(a.attname ORDER BY array_position(ix.indkey, a.attnum)) as columns \
    FROM pg_index ix \
    JOIN pg_class t ON t.oid = ix.indrelid \
    JOIN pg_class i ON i.oid = ix.indexrelid \
    JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY(ix.indkey) \
    JOIN pg_namespace n ON n.oid = t.relnamespace \
    WHERE n.nspname = 'public' \
      AND t.relname = $1 \
      AND NOT ix.indisprimary \
    GROUP BY i.relname, ix.indisunique \
    ORDER BY i.relname";

/// Plan from a snapshot (reuses the shared plan_from_snapshot).
/// This allows Postgres to plan incrementally once introspection data is available.
pub fn plan_from_snapshot(snapshot: &crate::SchemaSnapshot, target: &AppManifest) -> SchemaPlan {
    crate::plan_from_snapshot(snapshot, target)
}

// ---------------------------------------------------------------------------
// CRUD SQL generation helpers (used by live adapter, testable without a DB)
// ---------------------------------------------------------------------------

/// Generate a lex-sortable, monotonic-ish unique ID.
///
/// Format: 32 hex chars of `as_nanos()` (zero-padded) followed by 8 hex chars
/// of a per-process atomic counter. The counter prevents collisions when two
/// inserts hit the same nanosecond and — critically — keeps order stable: an
/// id minted at the same nanosecond is monotonically greater than the
/// previous one. Width is fixed at 40 chars so lexicographic comparison
/// matches creation order, which is what cursor pagination relies on.
pub fn generate_id() -> String {
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};
    static COUNTER: AtomicU32 = AtomicU32::new(0);
    let ts = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("{ts:032x}{seq:08x}")
}

/// Convert a JSON value to its string representation for use as a SQL parameter.
///
/// Kept for back-compat with callers that need a textual fallback (e.g.
/// human-readable logs). New code should bind through [`JsonParam`] so
/// integers/booleans/nulls reach Postgres in their typed form instead of
/// collapsing to TEXT, which the driver can't coerce into INTEGER /
/// BOOLEAN / TIMESTAMPTZ columns and which silently turns JSON `null`
/// into an empty string for FK columns.
pub fn json_value_to_string(val: &serde_json::Value) -> String {
    match val {
        serde_json::Value::String(s) => s.clone(),
        serde_json::Value::Number(n) => n.to_string(),
        serde_json::Value::Bool(b) => b.to_string(),
        serde_json::Value::Null => String::new(),
        other => other.to_string(),
    }
}

/// A typed wrapper around a JSON scalar that implements
/// [`postgres::types::ToSql`] for INSERT/UPDATE parameters.
///
/// The previous implementation passed every value as `String`, which broke
/// non-text columns (the postgres driver can't bind a string literal into
/// an INTEGER / BOOLEAN / TIMESTAMPTZ slot) and silently turned JSON
/// `null` into the empty string for nullable FKs (so `unlink` left
/// dangling `""` references instead of NULL). `JsonParam` carries the
/// JSON variant tag through to `to_sql` so the driver can pick the
/// correct binary representation per column type.
///
/// JSON arrays/objects collapse to their JSON-string form (TEXT) — the
/// runtime layer doesn't currently model array/object columns at the
/// manifest level on Postgres, so anything that lands here came from
/// caller-supplied prose that's expected to fit into a TEXT column.
#[derive(Debug, Clone, PartialEq)]
pub enum JsonParam {
    Null,
    Text(String),
    Int(i64),
    Float(f64),
    Bool(bool),
}

impl JsonParam {
    /// Lift a `serde_json::Value` into the typed parameter form. Numbers
    /// that fit `i64` go through as Int; everything else goes through as
    /// Float to preserve fractional / large-magnitude values.
    pub fn from_json(val: &serde_json::Value) -> Self {
        match val {
            serde_json::Value::Null => JsonParam::Null,
            serde_json::Value::String(s) => JsonParam::Text(s.clone()),
            serde_json::Value::Bool(b) => JsonParam::Bool(*b),
            serde_json::Value::Number(n) => {
                if let Some(i) = n.as_i64() {
                    JsonParam::Int(i)
                } else if let Some(f) = n.as_f64() {
                    JsonParam::Float(f)
                } else {
                    JsonParam::Text(n.to_string())
                }
            }
            other => JsonParam::Text(other.to_string()),
        }
    }
}

#[cfg(feature = "postgres-live")]
impl postgres::types::ToSql for JsonParam {
    fn to_sql(
        &self,
        ty: &postgres::types::Type,
        out: &mut bytes::BytesMut,
    ) -> Result<postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>> {
        use postgres::types::Type;

        // Null binds as SQL NULL regardless of the column's declared
        // type — the postgres driver treats `IsNull::Yes` as a null
        // value of the requested type.
        if matches!(self, JsonParam::Null) {
            return Ok(postgres::types::IsNull::Yes);
        }

        // Match each JsonParam variant against the COLUMN's declared
        // type so the binary encoding actually fits the target slot.
        // Postgres rejects "binary data of wrong size" if you bind an
        // `i64` (BIGINT, 8 bytes) into an INTEGER (4 bytes) — which is
        // exactly what the previous "everything is a String" path did
        // on every non-TEXT column.
        match (self, ty) {
            (JsonParam::Bool(b), &Type::BOOL) => b.to_sql(ty, out),

            (JsonParam::Int(n), &Type::INT2) => (*n as i16).to_sql(ty, out),
            (JsonParam::Int(n), &Type::INT4) => (*n as i32).to_sql(ty, out),
            (JsonParam::Int(n), &Type::INT8) => n.to_sql(ty, out),
            (JsonParam::Int(n), &Type::FLOAT4) => (*n as f32).to_sql(ty, out),
            (JsonParam::Int(n), &Type::FLOAT8) => (*n as f64).to_sql(ty, out),

            (JsonParam::Float(f), &Type::FLOAT4) => (*f as f32).to_sql(ty, out),
            (JsonParam::Float(f), &Type::FLOAT8) => f.to_sql(ty, out),
            (JsonParam::Float(f), &Type::INT4) => (*f as i32).to_sql(ty, out),
            (JsonParam::Float(f), &Type::INT8) => (*f as i64).to_sql(ty, out),

            (JsonParam::Text(s), &Type::TEXT)
            | (JsonParam::Text(s), &Type::VARCHAR)
            | (JsonParam::Text(s), &Type::BPCHAR)
            | (JsonParam::Text(s), &Type::NAME) => s.to_sql(ty, out),
            (JsonParam::Text(s), &Type::TIMESTAMPTZ) => {
                // The runtime models datetimes as ISO 8601 strings
                // (`pylon_kernel::util::now_iso` shape, plus
                // user-supplied RFC 3339). Postgres's TIMESTAMPTZ
                // binary wire format is `i64` microseconds since
                // 2000-01-01 UTC — NOT the bytes of an ISO string.
                // The previous impl bound via `&str::to_sql(TIMESTAMPTZ, ...)`,
                // which advertised TIMESTAMPTZ format but wrote raw
                // ASCII; Postgres rejected with "incorrect binary
                // data format in bind parameter N". This was the
                // OAuth-callback failure mode on pylon-cloud
                // (User.createdAt). Parse via chrono and let the
                // postgres crate's `with-chrono-0_4` ToSql impl
                // emit the proper binary format.
                let dt = chrono::DateTime::parse_from_rfc3339(s)
                    .map_err(|e| format!("invalid TIMESTAMPTZ string {s:?}: {e}"))?
                    .with_timezone(&chrono::Utc);
                dt.to_sql(ty, out)
            }
            (JsonParam::Text(s), &Type::TIMESTAMP) => {
                // TIMESTAMP (no timezone) — same conversion shape but
                // bind as NaiveDateTime so chrono picks the right
                // binary encoding for the column.
                let dt = chrono::DateTime::parse_from_rfc3339(s)
                    .map_err(|e| format!("invalid TIMESTAMP string {s:?}: {e}"))?
                    .with_timezone(&chrono::Utc)
                    .naive_utc();
                dt.to_sql(ty, out)
            }
            (JsonParam::Text(s), &Type::DATE) => {
                let dt = chrono::DateTime::parse_from_rfc3339(s)
                    .map_err(|e| format!("invalid DATE string {s:?}: {e}"))?
                    .with_timezone(&chrono::Utc)
                    .date_naive();
                dt.to_sql(ty, out)
            }

            // Cross-type fallback: render as text and bind into a TEXT
            // slot, OR error if the target column doesn't accept text.
            // Catches "manifest says INT but caller sent a stringified
            // number" — better to fail loudly than silently coerce.
            (other, _) => {
                let s = match other {
                    JsonParam::Bool(b) => b.to_string(),
                    JsonParam::Int(n) => n.to_string(),
                    JsonParam::Float(f) => f.to_string(),
                    JsonParam::Text(s) => s.clone(),
                    JsonParam::Null => unreachable!(),
                };
                s.to_sql(ty, out)
            }
        }
    }

    fn accepts(_ty: &postgres::types::Type) -> bool {
        // Defer per-variant acceptance to to_sql_checked, which dispatches
        // to the inner type's ToSql impl. Returning `true` here matches
        // the postgres crate's recommended pattern for sum-type wrappers.
        true
    }

    postgres::types::to_sql_checked!();
}

/// Build an INSERT SQL statement and collect typed parameter values.
/// Returns `(sql, params)` where `params[0]` is the generated ID
/// (always `JsonParam::Text`). Subsequent params carry the JSON-typed
/// value so the postgres driver can bind them to typed columns
/// (INTEGER / BOOLEAN / TIMESTAMPTZ / TEXT) and so JSON `null` reaches
/// the database as SQL NULL — the previous string-collapsing path stored
/// `""` for nullable FKs and broke any non-text column.
pub fn build_insert_sql(
    entity: &str,
    data: &serde_json::Value,
) -> Result<(String, Vec<JsonParam>), StorageError> {
    let obj = data.as_object().ok_or_else(|| StorageError {
        code: "PG_INVALID_DATA".into(),
        message: "Insert data must be a JSON object".into(),
    })?;

    // Honor caller-supplied `id` (used by the CRDT path that needs to
    // share the row id with the LoroDoc snapshot key it just wrote).
    // Fall back to a fresh ULID-shaped id when absent. A non-string
    // `id` value is rejected explicitly — silently regenerating would
    // mask schema bugs (e.g. a caller passing an int id) and let the
    // CRDT snapshot key drift from the materialized row id.
    let id = match obj.get("id") {
        None | Some(serde_json::Value::Null) => generate_id(),
        Some(serde_json::Value::String(s)) => s.clone(),
        Some(other) => {
            return Err(StorageError {
                code: "PG_INVALID_ID".into(),
                message: format!(
                    "Insert data carried a non-string `id` value: {other}. Pylon row ids \
                     are always strings (40-char hex). Drop the `id` field to let the \
                     server generate one, or supply a string."
                ),
            });
        }
    };

    let mut col_names = vec!["id".to_string()];
    let mut placeholders = vec!["$1".to_string()];
    let mut values: Vec<JsonParam> = vec![JsonParam::Text(id)];

    let mut i = 0usize;
    for (key, val) in obj {
        if key == "id" {
            // Already emitted via the synthetic first column; skipping
            // here avoids `INSERT ... (id, id, ...)` which Postgres
            // rejects with `duplicate key value`.
            continue;
        }
        col_names.push(quote_ident(key));
        placeholders.push(format!("${}", i + 2));
        values.push(JsonParam::from_json(val));
        i += 1;
    }

    let sql = format!(
        "INSERT INTO {} ({}) VALUES ({})",
        quote_ident(entity),
        col_names.join(", "),
        placeholders.join(", ")
    );

    Ok((sql, values))
}

/// Build an UPDATE SQL statement and collect typed parameter values.
/// Returns `(sql, params)` where `params[0]` is the row ID.
pub fn build_update_sql(
    entity: &str,
    id: &str,
    data: &serde_json::Value,
) -> Result<(String, Vec<JsonParam>), StorageError> {
    let obj = data.as_object().ok_or_else(|| StorageError {
        code: "PG_INVALID_DATA".into(),
        message: "Update data must be a JSON object".into(),
    })?;

    if obj.is_empty() {
        return Err(StorageError {
            code: "PG_INVALID_DATA".into(),
            message: "Update data must contain at least one field".into(),
        });
    }

    let mut set_clauses = Vec::new();
    let mut values: Vec<JsonParam> = vec![JsonParam::Text(id.to_string())];

    let mut i = 0usize;
    for (key, val) in obj {
        if key == "id" {
            // Reject primary-key mutation. Letting `id` into the SET
            // clause lets a client move a row out from under its CRDT
            // sidecar (which is keyed by the original row_id) and its
            // FTS shadow row (FK-bound to the original id). The SQLite
            // path silently drops `id` here too — keep the same
            // shape, but errors so the caller sees the bug.
            return Err(StorageError {
                code: "PG_INVALID_UPDATE".into(),
                message:
                    "Updating the `id` column is not allowed — Pylon row ids are immutable; \
                     drop the field from the patch."
                        .into(),
            });
        }
        set_clauses.push(format!("{} = ${}", quote_ident(key), i + 2));
        values.push(JsonParam::from_json(val));
        i += 1;
    }

    if set_clauses.is_empty() {
        return Err(StorageError {
            code: "PG_INVALID_DATA".into(),
            message: "Update data must contain at least one non-id field".into(),
        });
    }

    let sql = format!(
        "UPDATE {} SET {} WHERE id = $1",
        quote_ident(entity),
        set_clauses.join(", ")
    );

    Ok((sql, values))
}

/// Helper for the existing `Vec<JsonParam>` → `&[&dyn ToSql + Sync]` lift
/// at insert/update/transact call sites. The postgres driver wants a
/// slice of trait objects; this avoids repeating the same map at each site.
#[cfg(feature = "postgres-live")]
fn as_pg_params(values: &[JsonParam]) -> Vec<&(dyn postgres::types::ToSql + Sync)> {
    values
        .iter()
        .map(|v| v as &(dyn postgres::types::ToSql + Sync))
        .collect()
}

// ---------------------------------------------------------------------------
// Live Postgres adapter (requires "postgres-live" feature)
// ---------------------------------------------------------------------------

#[cfg(feature = "postgres-live")]
pub mod live {
    use super::*;
    use crate::{
        ColumnSnapshot, IndexSnapshot, SchemaSnapshot, StorageAdapter, StorageError, TableSnapshot,
    };

    /// SSL parsing result for `parse_pg_url_ssl`. We need both the
    /// cleaned-up URL (libpq-only params stripped) and the boolean
    /// "should we use TLS for this connection" so the connect path
    /// can pick between `NoTls` and `MakeTlsConnector`.
    pub(super) struct PgUrlSsl {
        pub use_tls: bool,
    }

    /// Pre-process a Postgres URL: strip libpq-specific query params
    /// the Rust `postgres` crate's URL parser doesn't accept, and
    /// figure out whether TLS should be enabled for the connection.
    ///
    /// Recognizes:
    ///   - sslmode=disable / prefer / allow → no TLS
    ///   - sslmode=require / verify-ca / verify-full → TLS
    ///   - sslrootcert=system → TLS, use OS CA store (handled by
    ///     `native-tls` automatically; we just record the intent)
    ///   - sslrootcert=<path> → TLS, but the path is currently
    ///     ignored — `native-tls` reads system roots only. Logged
    ///     as a one-time warning so operators see the gap.
    ///
    /// Anything we don't understand is dropped from the URL silently
    /// (defense against future libpq additions confusing the parser).
    pub(super) fn parse_pg_url_ssl(url: &str) -> (String, PgUrlSsl) {
        let (base, query) = match url.find('?') {
            Some(idx) => (&url[..idx], &url[idx + 1..]),
            None => return (url.to_string(), PgUrlSsl { use_tls: false }),
        };

        let mut use_tls = false;
        let mut kept: Vec<String> = Vec::new();
        for pair in query.split('&') {
            if pair.is_empty() {
                continue;
            }
            let (k, v) = match pair.find('=') {
                Some(i) => (&pair[..i], &pair[i + 1..]),
                None => (pair, ""),
            };
            match k {
                "sslmode" => match v.to_ascii_lowercase().as_str() {
                    "disable" | "allow" => {
                        // Caller said "no TLS" — pass through to the
                        // postgres crate, which knows `disable`.
                        kept.push("sslmode=disable".to_string());
                    }
                    "prefer" => {
                        // Postgres crate's default; let it through.
                        kept.push("sslmode=prefer".to_string());
                    }
                    "require" | "verify-ca" | "verify-full" | "" => {
                        // All the "use TLS" modes. Rust crate only
                        // accepts `require`; verify-* would be
                        // rejected. Normalize to `require` and let
                        // our TLS connector handle cert verification.
                        use_tls = true;
                        kept.push("sslmode=require".to_string());
                    }
                    other => {
                        tracing::warn!(
                            "[pg] unknown sslmode='{other}' — defaulting to require + TLS"
                        );
                        use_tls = true;
                        kept.push("sslmode=require".to_string());
                    }
                },
                "sslrootcert" => {
                    // Either "system" (use OS roots — that's what
                    // native-tls does anyway) or a path (which we
                    // can't honor without bringing in openssl). Log
                    // and drop in both cases; TLS still happens.
                    if v != "system" && !v.is_empty() {
                        tracing::warn!(
                            "[pg] sslrootcert={v} ignored — native-tls uses system roots"
                        );
                    }
                    use_tls = true;
                }
                _ => {
                    // Forward everything else (search_path, application_name,
                    // connect_timeout, etc.) so libpq-style URLs that work
                    // for the rest of the world keep working here.
                    kept.push(pair.to_string());
                }
            }
        }

        let cleaned = if kept.is_empty() {
            base.to_string()
        } else {
            format!("{}?{}", base, kept.join("&"))
        };
        (cleaned, PgUrlSsl { use_tls })
    }

    #[cfg(test)]
    mod url_tests {
        use super::parse_pg_url_ssl;

        #[test]
        fn strips_libpq_only_sslmode_verify_full() {
            let (cleaned, ssl) = parse_pg_url_ssl(
                "postgres://u:p@h:5432/db?sslmode=verify-full&sslrootcert=system",
            );
            assert!(ssl.use_tls);
            // verify-full normalized to require; sslrootcert dropped.
            assert_eq!(cleaned, "postgres://u:p@h:5432/db?sslmode=require");
        }

        #[test]
        fn passes_through_disable() {
            let (cleaned, ssl) = parse_pg_url_ssl("postgres://h/db?sslmode=disable");
            assert!(!ssl.use_tls);
            assert_eq!(cleaned, "postgres://h/db?sslmode=disable");
        }

        #[test]
        fn no_query_string_no_tls() {
            let (cleaned, ssl) = parse_pg_url_ssl("postgres://h/db");
            assert!(!ssl.use_tls);
            assert_eq!(cleaned, "postgres://h/db");
        }

        #[test]
        fn unknown_params_pass_through() {
            let (cleaned, _) = parse_pg_url_ssl(
                "postgres://h/db?application_name=pylon&connect_timeout=5",
            );
            assert!(cleaned.contains("application_name=pylon"));
            assert!(cleaned.contains("connect_timeout=5"));
        }

        #[test]
        fn sslrootcert_alone_enables_tls() {
            // Rare but valid — sslrootcert=system implies TLS even
            // without an explicit sslmode.
            let (cleaned, ssl) = parse_pg_url_ssl(
                "postgres://h/db?sslrootcert=system",
            );
            assert!(ssl.use_tls);
            // No sslmode synthesized — the postgres crate defaults
            // to `prefer` which happily upgrades to our TLS connector.
            assert_eq!(cleaned, "postgres://h/db");
        }
    }

    /// A live Postgres adapter with a real database connection.
    pub struct LivePostgresAdapter {
        client: postgres::Client,
    }

    impl LivePostgresAdapter {
        /// Borrow the underlying postgres client mutably. Used by
        /// `PostgresDataStore::with_transaction` to start an
        /// interactive transaction across multiple TS-function
        /// `ctx.db` calls. `pub(crate)` because exposing raw
        /// `&mut Client` outside pylon-storage would let callers
        /// issue arbitrary SQL, bypassing the typed `DataStore`
        /// surface that the rest of the framework relies on.
        pub(crate) fn client_mut(&mut self) -> &mut postgres::Client {
            &mut self.client
        }

        /// Connect to a Postgres database.
        ///
        /// Honors libpq-style URL params the Rust postgres crate doesn't
        /// natively understand:
        ///   - `sslmode=verify-full`, `verify-ca`, `require` → use TLS
        ///     via rustls + the OS trust store. The Rust postgres crate
        ///     only knows `disable`/`prefer`/`require` — it rejects the
        ///     libpq extras with "invalid connection string".
        ///   - `sslrootcert=system` → trust the OS CA store (rustls-
        ///     native-certs reads it via SecurityFramework / SChannel /
        ///     /etc/ssl). `sslrootcert=<path>` is not yet supported and
        ///     falls back to system roots with a warning.
        ///
        /// Strips the libpq-only params from the URL before passing to
        /// the postgres crate's parser, so it doesn't choke. Common
        /// real-world example that was failing pre-fix: Fly Postgres
        /// emits `?sslmode=verify-full&sslrootcert=system` URLs.
        ///
        /// Uses rustls (not native-tls/openssl) so binary builds
        /// cross-compile cleanly on musl + arm64 without needing
        /// OPENSSL_DIR. Pure Rust all the way down.
        pub fn connect(url: &str) -> Result<Self, StorageError> {
            let (cleaned, ssl) = parse_pg_url_ssl(url);
            let result = if ssl.use_tls {
                let mut roots = rustls::RootCertStore::empty();
                let native_certs = rustls_native_certs::load_native_certs();
                for cert in native_certs.certs {
                    let _ = roots.add(cert);
                }
                if !native_certs.errors.is_empty() {
                    tracing::warn!(
                        "[pg] rustls native cert load reported {} non-fatal errors",
                        native_certs.errors.len()
                    );
                }
                let config = rustls::ClientConfig::builder()
                    .with_root_certificates(roots)
                    .with_no_client_auth();
                let tls = tokio_postgres_rustls::MakeRustlsConnect::new(config);
                postgres::Client::connect(&cleaned, tls)
            } else {
                postgres::Client::connect(&cleaned, postgres::NoTls)
            };
            let client = result.map_err(|e| StorageError {
                code: "PG_CONNECT_FAILED".into(),
                message: format!("Failed to connect to Postgres: {e}"),
            })?;
            Ok(Self { client })
        }

        /// Read the current schema from the live database.
        pub fn read_schema(&mut self) -> Result<SchemaSnapshot, StorageError> {
            let table_rows = self
                .client
                .query(INTROSPECT_TABLES_SQL, &[])
                .map_err(pg_err)?;

            let mut tables = Vec::new();
            for row in &table_rows {
                let table_name: String = row.get(0);
                let columns = self.read_columns(&table_name)?;
                let indexes = self.read_indexes(&table_name)?;
                tables.push(TableSnapshot {
                    name: table_name,
                    columns,
                    indexes,
                });
            }

            Ok(SchemaSnapshot { tables })
        }

        fn read_columns(&mut self, table: &str) -> Result<Vec<ColumnSnapshot>, StorageError> {
            let rows = self
                .client
                .query(INTROSPECT_COLUMNS_SQL, &[&table])
                .map_err(pg_err)?;

            let mut columns = Vec::new();
            for row in &rows {
                let name: String = row.get(0);
                let data_type: String = row.get(1);
                let is_nullable: String = row.get(2);
                let is_pk: i64 = row.get(3);
                columns.push(ColumnSnapshot {
                    name,
                    column_type: data_type,
                    notnull: is_nullable == "NO",
                    primary_key: is_pk > 0,
                });
            }
            Ok(columns)
        }

        fn read_indexes(&mut self, table: &str) -> Result<Vec<IndexSnapshot>, StorageError> {
            let rows = self
                .client
                .query(INTROSPECT_INDEXES_SQL, &[&table])
                .map_err(pg_err)?;

            let mut indexes = Vec::new();
            for row in &rows {
                let name: String = row.get(0);
                let unique: bool = row.get(1);
                let columns: Vec<String> = row.get(2);
                indexes.push(IndexSnapshot {
                    name,
                    columns,
                    unique,
                });
            }
            Ok(indexes)
        }

        /// Plan from live database state.
        pub fn plan_from_live(&mut self, target: &AppManifest) -> Result<SchemaPlan, StorageError> {
            let snapshot = self.read_schema()?;
            Ok(crate::plan_from_snapshot(&snapshot, target))
        }
    }

    impl StorageAdapter for LivePostgresAdapter {
        fn plan_schema(&self, _target: &AppManifest) -> Result<SchemaPlan, StorageError> {
            Err(StorageError {
                code: "PG_PLAN_NEEDS_MUTABLE".into(),
                message: "Use plan_from_live() instead for live Postgres planning".into(),
            })
        }

        fn apply_schema(&self, _plan: &SchemaPlan) -> Result<(), StorageError> {
            Err(StorageError {
                code: "PG_APPLY_USE_METHOD".into(),
                message: "Use apply_plan() instead of the trait method for live Postgres".into(),
            })
        }
    }

    impl LivePostgresAdapter {
        /// Apply a schema plan to the live database.
        pub fn apply_plan(&mut self, plan: &SchemaPlan) -> Result<(), StorageError> {
            let statements = plan_to_sql(plan)?;
            for sql in &statements {
                self.client.execute(sql.as_str(), &[]).map_err(pg_err)?;
            }
            Ok(())
        }

        /// Execute a raw SQL statement against the live database. Used by
        /// integration tests for setup/teardown (DROP TABLE, TRUNCATE) —
        /// production code should go through `apply_plan` so changes are
        /// represented in the migration history. Returns the number of
        /// rows affected.
        pub fn exec_raw(&mut self, sql: &str) -> Result<u64, StorageError> {
            self.client.execute(sql, &[]).map_err(pg_err)
        }

        /// Insert a row. Returns the generated ID.
        pub fn insert(
            &mut self,
            entity: &str,
            data: &serde_json::Value,
        ) -> Result<String, StorageError> {
            let (sql, values) = build_insert_sql(entity, data)?;
            // The first param is always the generated ID — extract it before
            // we hand `values` off to the postgres driver as borrowed slices.
            let id = match &values[0] {
                JsonParam::Text(s) => s.clone(),
                _ => {
                    return Err(StorageError {
                        code: "PG_INTERNAL".into(),
                        message: "build_insert_sql produced non-text id param".into(),
                    });
                }
            };
            let params = as_pg_params(&values);
            self.client.execute(sql.as_str(), &params).map_err(pg_err)?;
            Ok(id)
        }

        /// Get a row by ID.
        pub fn get_by_id(
            &mut self,
            entity: &str,
            id: &str,
        ) -> Result<Option<serde_json::Value>, StorageError> {
            let sql = format!("SELECT * FROM {} WHERE id = $1", quote_ident(entity));
            let rows = self.client.query(sql.as_str(), &[&id]).map_err(pg_err)?;

            match rows.first() {
                Some(row) => Ok(Some(row_to_json(row))),
                None => Ok(None),
            }
        }

        /// List all rows from an entity.
        pub fn list(&mut self, entity: &str) -> Result<Vec<serde_json::Value>, StorageError> {
            let sql = format!("SELECT * FROM {}", quote_ident(entity));
            let rows = self.client.query(sql.as_str(), &[]).map_err(pg_err)?;

            Ok(rows.iter().map(row_to_json).collect())
        }

        /// Cursor-paginated list. `after` is the last `id` from the previous
        /// page; the result contains rows with `id > after` (lex order),
        /// limited to `limit`. Used for sync push/pull.
        pub fn list_after(
            &mut self,
            entity: &str,
            after: Option<&str>,
            limit: usize,
        ) -> Result<Vec<serde_json::Value>, StorageError> {
            // Cap limit at a sensible upper bound so a malicious client can't
            // stream the whole table by passing limit=u64::MAX.
            let capped: i64 = limit.min(10_000) as i64;
            let sql = match after {
                Some(_) => format!(
                    "SELECT * FROM {} WHERE id > $1 ORDER BY id ASC LIMIT $2",
                    quote_ident(entity)
                ),
                None => format!(
                    "SELECT * FROM {} ORDER BY id ASC LIMIT $1",
                    quote_ident(entity)
                ),
            };
            let rows = match after {
                Some(cursor) => self
                    .client
                    .query(sql.as_str(), &[&cursor, &capped])
                    .map_err(pg_err)?,
                None => self
                    .client
                    .query(sql.as_str(), &[&capped])
                    .map_err(pg_err)?,
            };
            Ok(rows.iter().map(row_to_json).collect())
        }

        /// Update a row by ID. Returns true if the row was found and updated.
        pub fn update(
            &mut self,
            entity: &str,
            id: &str,
            data: &serde_json::Value,
        ) -> Result<bool, StorageError> {
            let (sql, values) = build_update_sql(entity, id, data)?;
            let params = as_pg_params(&values);
            let rows_affected = self.client.execute(sql.as_str(), &params).map_err(pg_err)?;
            Ok(rows_affected > 0)
        }

        /// Delete a row by ID. Returns true if the row was found and deleted.
        pub fn delete(&mut self, entity: &str, id: &str) -> Result<bool, StorageError> {
            let sql = format!("DELETE FROM {} WHERE id = $1", quote_ident(entity));
            let rows_affected = self.client.execute(sql.as_str(), &[&id]).map_err(pg_err)?;
            Ok(rows_affected > 0)
        }

        /// Look up a row by `field = value`. Caller must validate `field`
        /// against the manifest before calling — we still `quote_ident` it
        /// but won't catch a typo against the entity definition.
        pub fn lookup_field(
            &mut self,
            entity: &str,
            field: &str,
            value: &str,
        ) -> Result<Option<serde_json::Value>, StorageError> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 LIMIT 1",
                quote_ident(entity),
                quote_ident(field),
            );
            let rows = self.client.query(sql.as_str(), &[&value]).map_err(pg_err)?;
            Ok(rows.first().map(row_to_json))
        }

        /// Push a `query_filtered` filter down to a real Postgres `WHERE`.
        ///
        /// Supported operators (parity with the SQLite path):
        /// - Equality (`field: value`)
        /// - `$not`: emits `field != value`
        /// - `$gt` / `$gte` / `$lt` / `$lte`
        /// - `$like`: emits `field LIKE value` (use `%`/`_` wildcards in
        ///   the value; case-sensitive — pass `$ilike` for case-insensitive
        ///   if/when the SQLite side adds it)
        /// - `$in: [..]`: emits `field IN ($1, $2, ...)`
        ///
        /// Top-level meta operators: `$order`, `$limit`, `$offset`.
        ///
        /// `$search` (FTS5 on SQLite) is NOT supported here — Postgres
        /// would need a tsvector column or a generic ILIKE OR-fold across
        /// every text field, neither of which is wired up yet. Returns
        /// `SEARCH_NOT_SUPPORTED` so callers can branch instead of
        /// receiving silently-broad results.
        ///
        /// Anything else is silently ignored (matches the in-memory fallback's
        /// permissive behavior). Field names are validated against `valid_columns`
        /// to prevent SQL injection — pass the entity's column set.
        pub fn query_filtered(
            &mut self,
            entity: &str,
            filter: &serde_json::Value,
            valid_columns: &[String],
        ) -> Result<Vec<serde_json::Value>, StorageError> {
            let (sql, params) = Self::build_query_filtered_sql(entity, filter, valid_columns)?;
            let pg_params = as_pg_params(&params);
            let rows = self
                .client
                .query(sql.as_str(), &pg_params)
                .map_err(pg_err)?;
            Ok(rows.iter().map(row_to_json).collect())
        }

        /// Build the `SELECT ... FROM entity ...` SQL + bound params for
        /// a `query_filtered` request. Pure: takes a manifest's column
        /// list, returns text. Both the live adapter and the in-tx
        /// `PgTxStore` call this so the operator surface ($eq, $like,
        /// $in, $order, $limit, $offset) stays identical regardless of
        /// where the query runs.
        pub(crate) fn build_query_filtered_sql(
            entity: &str,
            filter: &serde_json::Value,
            valid_columns: &[String],
        ) -> Result<(String, Vec<JsonParam>), StorageError> {
            let empty = serde_json::Map::new();
            let obj = filter.as_object().unwrap_or(&empty);

            let validate = |col: &str| -> Result<(), StorageError> {
                if col == "id" || valid_columns.iter().any(|c| c == col) {
                    Ok(())
                } else {
                    Err(StorageError {
                        code: "UNKNOWN_COLUMN".into(),
                        message: format!("Unknown column \"{col}\" on entity \"{entity}\""),
                    })
                }
            };

            let mut where_clauses: Vec<String> = Vec::new();
            let mut order_clause = String::new();
            let mut limit_clause = String::new();
            let mut offset_clause = String::new();
            // Collect (col, op, value) so placeholder numbers can be assigned
            // in a single materialization pass after the parse loop. Values
            // are now JsonParam (typed) instead of String — see `value_to_pg`.
            let mut planned: Vec<(String, String, JsonParam)> = Vec::new();

            for (key, val) in obj {
                match key.as_str() {
                    "$search" => {
                        // PG full-text via the entity's `_fts_<entity>`
                        // shadow table: `id IN (SELECT entity_id FROM
                        // _fts_<E> WHERE tsv @@ plainto_tsquery(...))`.
                        // Mirrors the SQLite path that joins the FTS5
                        // virtual table; the join here is a subquery so
                        // it composes with arbitrary other predicates
                        // (`$gt`, `$in`, etc.) in the same WHERE.
                        let raw = match val {
                            serde_json::Value::String(s) => s.clone(),
                            other => other.to_string(),
                        };
                        // Bind as a normal text param so all the
                        // existing placeholder-numbering and binding
                        // logic applies. The SQL uses the placeholder
                        // inside `plainto_tsquery('english', $N)`.
                        // Positioning logic mirrors the `$in` arm: the
                        // value will land at `planned.len()+1` because
                        // the materialization pass below pushes one
                        // param per planned item in order.
                        let placeholder_n = planned.len() + 1;
                        where_clauses.push(format!(
                            "{}.id IN (SELECT entity_id FROM \"_fts_{entity}\" \
                                       WHERE tsv @@ plainto_tsquery('english', ${placeholder_n}))",
                            quote_ident(entity),
                        ));
                        // Reuse the IN-style sentinel so the
                        // materialization pass below pushes the param
                        // without re-emitting a where_clause for it.
                        planned.push((
                            format!("__search_{}", planned.len()),
                            "__INLINE__".into(),
                            JsonParam::Text(raw),
                        ));
                    }
                    "$order" => {
                        if let Some(ord) = val.as_object() {
                            let mut parts = Vec::new();
                            for (col, dir) in ord {
                                validate(col)?;
                                let d = match dir.as_str().unwrap_or("asc") {
                                    "desc" | "DESC" => "DESC",
                                    _ => "ASC",
                                };
                                parts.push(format!("{} {d}", quote_ident(col)));
                            }
                            if !parts.is_empty() {
                                order_clause = format!(" ORDER BY {}", parts.join(", "));
                            }
                        }
                    }
                    "$limit" => {
                        if let Some(n) = val.as_u64() {
                            limit_clause = format!(" LIMIT {}", n);
                        }
                    }
                    "$offset" => {
                        if let Some(n) = val.as_u64() {
                            offset_clause = format!(" OFFSET {}", n);
                        }
                    }
                    field => {
                        validate(field)?;
                        match val {
                            serde_json::Value::Object(ops) => {
                                for (op, v) in ops {
                                    match op.as_str() {
                                        "$not" => planned.push((
                                            field.into(),
                                            "!=".into(),
                                            value_to_pg(v),
                                        )),
                                        "$gt" => {
                                            planned.push((field.into(), ">".into(), value_to_pg(v)))
                                        }
                                        "$gte" => planned.push((
                                            field.into(),
                                            ">=".into(),
                                            value_to_pg(v),
                                        )),
                                        "$lt" => {
                                            planned.push((field.into(), "<".into(), value_to_pg(v)))
                                        }
                                        "$lte" => planned.push((
                                            field.into(),
                                            "<=".into(),
                                            value_to_pg(v),
                                        )),
                                        "$like" => {
                                            // Wrap in `%...%` to match the
                                            // SQLite path's substring
                                            // semantics. Pre-fix divergence:
                                            // SQLite wrapped, PG forwarded
                                            // literally — `{name: {$like: "ann"}}`
                                            // matched "Joanne" on SQLite but
                                            // nothing on PG. Caller-supplied
                                            // wildcards inside the value still
                                            // work (`%j_n%` etc.) because we
                                            // only add wraps, never strip.
                                            let raw = match v {
                                                serde_json::Value::String(s) => s.clone(),
                                                other => other.to_string(),
                                            };
                                            planned.push((
                                                field.into(),
                                                "LIKE".into(),
                                                JsonParam::Text(format!("%{raw}%")),
                                            ));
                                        }
                                        "$in" => {
                                            if let Some(arr) = v.as_array() {
                                                if arr.is_empty() {
                                                    // `field IN ()` is invalid
                                                    // SQL on PG (and on SQLite
                                                    // too, technically — its
                                                    // path also short-circuits).
                                                    // An empty $in matches
                                                    // nothing; emit a guaranteed-
                                                    // false predicate so the
                                                    // parser doesn't choke and
                                                    // the result set comes back
                                                    // empty.
                                                    where_clauses.push("FALSE".into());
                                                } else {
                                                    let placeholders: Vec<String> = (0..arr.len())
                                                        .map(|i| {
                                                            format!("${}", planned.len() + 1 + i)
                                                        })
                                                        .collect();
                                                    where_clauses.push(format!(
                                                        "{} IN ({})",
                                                        quote_ident(field),
                                                        placeholders.join(", "),
                                                    ));
                                                    for x in arr {
                                                        planned.push((
                                                            format!("__inline_{}", planned.len()),
                                                            "__INLINE__".into(),
                                                            value_to_pg(x),
                                                        ));
                                                    }
                                                }
                                            }
                                        }
                                        _ => {}
                                    }
                                }
                            }
                            _ => planned.push((field.into(), "=".into(), value_to_pg(val))),
                        }
                    }
                }
            }

            // Materialize planned -> SQL + params.
            let mut params: Vec<JsonParam> = Vec::with_capacity(planned.len());
            for (field, op, v) in &planned {
                if op == "__INLINE__" {
                    // Already emitted via the IN-clause path; just push the value.
                    params.push(v.clone());
                } else {
                    let placeholder = format!("${}", params.len() + 1);
                    where_clauses.push(format!("{} {} {}", quote_ident(field), op, placeholder));
                    params.push(v.clone());
                }
            }

            let where_sql = if where_clauses.is_empty() {
                String::new()
            } else {
                format!(" WHERE {}", where_clauses.join(" AND "))
            };
            // Default deterministic order when the caller didn't pass
            // `$order` — matches the SQLite path. Without this,
            // identical queries return rows in different orders across
            // backends, which makes paginated APIs flaky.
            let final_order = if order_clause.is_empty() {
                format!(" ORDER BY {}", quote_ident("id"))
            } else {
                order_clause
            };
            let sql = format!(
                "SELECT * FROM {}{}{}{}{}",
                quote_ident(entity),
                where_sql,
                final_order,
                limit_clause,
                offset_clause,
            );

            Ok((sql, params))
        }

        /// Run a `DataStore::aggregate` spec against Postgres. Mirrors the
        /// SQLite path in `pylon-runtime` — supports `count`, `sum`, `avg`,
        /// `min`, `max`, `countDistinct`, `groupBy` (plain field names or
        /// `{field, bucket: hour|day|week|month|year}` for date bucketing
        /// via `date_trunc`), and a flat-equality `where` filter.
        ///
        /// Spec format (same JSON shape used by the SQLite path):
        /// ```json
        /// { "count": "*",
        ///   "sum": ["amount"],
        ///   "groupBy": [{"field": "createdAt", "bucket": "day"}],
        ///   "where": {"status": "paid"} }
        /// ```
        ///
        /// `valid_columns` is used to validate every field name before it's
        /// quoted into SQL — same pattern as `query_filtered`. Caller (the
        /// `DataStore` impl in this crate) supplies the entity's column set
        /// from the manifest.
        pub fn aggregate(
            &mut self,
            entity: &str,
            spec: &serde_json::Value,
            valid_columns: &[String],
        ) -> Result<serde_json::Value, StorageError> {
            let (sql, params, column_names) =
                Self::build_aggregate_sql(entity, spec, valid_columns)?;
            let pg_params = as_pg_params(&params);
            let rows = self
                .client
                .query(sql.as_str(), &pg_params)
                .map_err(pg_err)?;
            Ok(aggregate_rows_to_json(&rows, &column_names))
        }

        /// Build the aggregate `SELECT` SQL + bound params + the
        /// expected output column names. Pure: takes the entity's
        /// validated column list, returns text. Both the live adapter
        /// and the in-tx `PgTxStore` call this so spec parsing
        /// (validation, bucket vocabulary, where-clause translation)
        /// stays identical regardless of where the query runs.
        pub(crate) fn build_aggregate_sql(
            entity: &str,
            spec: &serde_json::Value,
            valid_columns: &[String],
        ) -> Result<(String, Vec<JsonParam>, Vec<String>), StorageError> {
            let obj = spec.as_object().ok_or_else(|| StorageError {
                code: "INVALID_QUERY".into(),
                message: "aggregate spec must be a JSON object".into(),
            })?;

            let validate = |col: &str| -> Result<(), StorageError> {
                if col == "id" || valid_columns.iter().any(|c| c == col) {
                    Ok(())
                } else {
                    Err(StorageError {
                        code: "UNKNOWN_COLUMN".into(),
                        message: format!("Unknown column \"{col}\" on entity \"{entity}\""),
                    })
                }
            };

            let mut select_parts: Vec<String> = Vec::new();
            let mut result_fields: Vec<String> = Vec::new();

            if let Some(count) = obj.get("count") {
                match count {
                    serde_json::Value::String(s) if s == "*" => {
                        select_parts.push("COUNT(*) AS count".into());
                        result_fields.push("count".into());
                    }
                    serde_json::Value::String(field) => {
                        validate(field)?;
                        let alias = format!("count_{field}");
                        select_parts.push(format!(
                            "COUNT({}) AS {}",
                            quote_ident(field),
                            quote_ident(&alias),
                        ));
                        result_fields.push(alias);
                    }
                    _ => {}
                }
            }

            for (fn_name, prefix) in [
                ("sum", "sum_"),
                ("avg", "avg_"),
                ("min", "min_"),
                ("max", "max_"),
            ] {
                if let Some(fields) = obj.get(fn_name).and_then(|v| v.as_array()) {
                    for field in fields {
                        if let Some(f) = field.as_str() {
                            validate(f)?;
                            let alias = format!("{prefix}{f}");
                            let sql_fn = fn_name.to_uppercase();
                            select_parts.push(format!(
                                "{}({}) AS {}",
                                sql_fn,
                                quote_ident(f),
                                quote_ident(&alias),
                            ));
                            result_fields.push(alias);
                        }
                    }
                }
            }

            if let Some(fields) = obj.get("countDistinct").and_then(|v| v.as_array()) {
                for field in fields {
                    if let Some(f) = field.as_str() {
                        validate(f)?;
                        let alias = format!("count_distinct_{f}");
                        select_parts.push(format!(
                            "COUNT(DISTINCT {}) AS {}",
                            quote_ident(f),
                            quote_ident(&alias),
                        ));
                        result_fields.push(alias);
                    }
                }
            }

            // groupBy: column name or { field, bucket } — same vocabulary as
            // the SQLite path. Buckets translate to Postgres `date_trunc`
            // (SQLite uses `strftime`); both collapse rows to the bucket
            // boundary identically.
            let mut group_by: Vec<String> = Vec::new();
            let mut group_select: Vec<String> = Vec::new();
            let mut group_field_names: Vec<String> = Vec::new();
            if let Some(groups) = obj.get("groupBy").and_then(|v| v.as_array()) {
                for g in groups {
                    if let Some(f) = g.as_str() {
                        validate(f)?;
                        let q = quote_ident(f);
                        group_by.push(q.clone());
                        group_select.push(q);
                        group_field_names.push(f.to_string());
                    } else if let Some(spec) = g.as_object() {
                        let field =
                            spec.get("field").and_then(|v| v.as_str()).ok_or_else(|| {
                                StorageError {
                                    code: "INVALID_QUERY".into(),
                                    message: "groupBy object spec requires `field`".into(),
                                }
                            })?;
                        validate(field)?;
                        let bucket = spec.get("bucket").and_then(|v| v.as_str()).unwrap_or("day");
                        let trunc_unit = match bucket {
                            "hour" | "day" | "week" | "month" | "year" => bucket,
                            _ => {
                                return Err(StorageError {
                                    code: "INVALID_QUERY".into(),
                                    message: format!(
                                        "bucket must be one of hour/day/week/month/year, got {bucket}"
                                    ),
                                });
                            }
                        };
                        let alias = format!("{field}_{bucket}");
                        let expr = format!("date_trunc('{}', {})", trunc_unit, quote_ident(field),);
                        group_by.push(expr.clone());
                        group_select.push(format!("{} AS {}", expr, quote_ident(&alias)));
                        group_field_names.push(alias);
                    }
                }
            }

            let mut full_select = group_select.clone();
            full_select.extend(select_parts.iter().cloned());
            if full_select.is_empty() {
                return Err(StorageError {
                    code: "INVALID_QUERY".into(),
                    message: "aggregate spec must include count/sum/avg/min/max/groupBy".into(),
                });
            }

            let mut where_clauses: Vec<String> = Vec::new();
            let mut params: Vec<JsonParam> = Vec::new();
            if let Some(w) = obj.get("where").and_then(|v| v.as_object()) {
                for (k, v) in w {
                    validate(k)?;
                    let placeholder = format!("${}", params.len() + 1);
                    where_clauses.push(format!("{} = {}", quote_ident(k), placeholder));
                    params.push(value_to_pg(v));
                }
            }
            let where_sql = if where_clauses.is_empty() {
                String::new()
            } else {
                format!(" WHERE {}", where_clauses.join(" AND "))
            };
            let group_sql = if group_by.is_empty() {
                String::new()
            } else {
                format!(" GROUP BY {}", group_by.join(", "))
            };

            let sql = format!(
                "SELECT {} FROM {}{}{}",
                full_select.join(", "),
                quote_ident(entity),
                where_sql,
                group_sql,
            );

            let column_names: Vec<String> = group_field_names
                .iter()
                .chain(result_fields.iter())
                .cloned()
                .collect();

            Ok((sql, params, column_names))
        }
    }

    /// Project rows from an aggregate `SELECT` into the
    /// `{ rows: [{...}] }` JSON shape both the SQLite path and the
    /// PG path return. Pure post-processing — works on rows produced
    /// from either `Client::query` or `Transaction::query`.
    pub fn aggregate_rows_to_json(
        rows: &[postgres::Row],
        column_names: &[String],
    ) -> serde_json::Value {
        let mut out: Vec<serde_json::Value> = Vec::with_capacity(rows.len());
        for row in rows {
            let row_json = row_to_json(row);
            if let serde_json::Value::Object(map) = &row_json {
                let mut filtered = serde_json::Map::new();
                for name in column_names {
                    if let Some(v) = map.get(name) {
                        filtered.insert(name.clone(), v.clone());
                    }
                }
                out.push(serde_json::Value::Object(filtered));
            } else {
                out.push(row_json);
            }
        }
        serde_json::json!({ "rows": out })
    }

    /// Atomic operation describing a single mutation inside [`LivePostgresAdapter::transact`].
    pub enum TxOp<'a> {
        Insert {
            entity: &'a str,
            data: &'a serde_json::Value,
        },
        Update {
            entity: &'a str,
            id: &'a str,
            data: &'a serde_json::Value,
        },
        Delete {
            entity: &'a str,
            id: &'a str,
        },
    }

    /// Result of a single op inside a transaction.
    #[derive(Debug, Clone)]
    pub enum TxResult {
        Inserted(String),
        Updated(bool),
        Deleted(bool),
    }

    impl LivePostgresAdapter {
        /// Run `ops` inside a single Postgres transaction. Either all of them
        /// commit together or none of them do — there is no partial state on
        /// failure. The ROLLBACK happens implicitly when the `Transaction`
        /// guard is dropped without `commit()` being called.
        pub fn transact(&mut self, ops: &[TxOp<'_>]) -> Result<Vec<TxResult>, StorageError> {
            let mut tx = self.client.transaction().map_err(pg_err)?;
            let mut results: Vec<TxResult> = Vec::with_capacity(ops.len());

            for op in ops {
                match op {
                    TxOp::Insert { entity, data } => {
                        let (sql, values) = build_insert_sql(entity, data)?;
                        let id = match &values[0] {
                            JsonParam::Text(s) => s.clone(),
                            _ => {
                                return Err(StorageError {
                                    code: "PG_INTERNAL".into(),
                                    message: "build_insert_sql produced non-text id param".into(),
                                });
                            }
                        };
                        let params = as_pg_params(&values);
                        tx.execute(sql.as_str(), &params).map_err(pg_err)?;
                        results.push(TxResult::Inserted(id));
                    }
                    TxOp::Update { entity, id, data } => {
                        let (sql, values) = build_update_sql(entity, id, data)?;
                        let params = as_pg_params(&values);
                        let n = tx.execute(sql.as_str(), &params).map_err(pg_err)?;
                        results.push(TxResult::Updated(n > 0));
                    }
                    TxOp::Delete { entity, id } => {
                        let sql = format!("DELETE FROM {} WHERE id = $1", quote_ident(entity));
                        let n = tx.execute(sql.as_str(), &[id]).map_err(pg_err)?;
                        results.push(TxResult::Deleted(n > 0));
                    }
                }
            }

            tx.commit().map_err(pg_err)?;
            Ok(results)
        }
    }

    /// Lift a JSON value into a typed Postgres parameter. The previous
    /// implementation collapsed everything to `String`, which silently
    /// stringified ints/bools and turned JSON `null` into `""` for
    /// nullable columns. Forwarding through `JsonParam` keeps the column
    /// type honest and lets callers `unlink` (set FK to NULL) cleanly.
    fn value_to_pg(v: &serde_json::Value) -> JsonParam {
        JsonParam::from_json(v)
    }

    pub(super) fn row_to_json(row: &postgres::Row) -> serde_json::Value {
        use postgres::types::Type;
        let mut obj = serde_json::Map::new();
        for (i, col) in row.columns().iter().enumerate() {
            let name = col.name().to_string();

            // Use `try_get` everywhere — `Row::get` panics on decode mismatch,
            // and a panic in a query handler poisons the connection mutex,
            // taking down all subsequent reads on this datastore. Anything
            // that fails to decode becomes Null with a one-shot warning.
            //
            // Timestamps and the catch-all path explicitly DON'T request
            // `String` — the postgres crate uses binary protocol by default
            // and there's no `FromSql<String>` impl for TIMESTAMPTZ etc. We
            // ask for `Vec<u8>` and lossy-stringify, which works for all
            // text-shaped columns in either protocol.
            let value: serde_json::Value = match *col.type_() {
                Type::BOOL => try_get_or_null::<Option<bool>>(row, i)
                    .flatten()
                    .map(serde_json::Value::Bool)
                    .unwrap_or(serde_json::Value::Null),
                Type::INT2 => try_get_or_null::<Option<i16>>(row, i)
                    .flatten()
                    .map(|v| serde_json::Value::Number(v.into()))
                    .unwrap_or(serde_json::Value::Null),
                Type::INT4 => try_get_or_null::<Option<i32>>(row, i)
                    .flatten()
                    .map(|v| serde_json::Value::Number(v.into()))
                    .unwrap_or(serde_json::Value::Null),
                Type::INT8 => try_get_or_null::<Option<i64>>(row, i)
                    .flatten()
                    .map(|v| serde_json::Value::Number(v.into()))
                    .unwrap_or(serde_json::Value::Null),
                Type::FLOAT4 => try_get_or_null::<Option<f32>>(row, i)
                    .flatten()
                    .and_then(|v| serde_json::Number::from_f64(v as f64))
                    .map(serde_json::Value::Number)
                    .unwrap_or(serde_json::Value::Null),
                Type::FLOAT8 => try_get_or_null::<Option<f64>>(row, i)
                    .flatten()
                    .and_then(serde_json::Number::from_f64)
                    .map(serde_json::Value::Number)
                    .unwrap_or(serde_json::Value::Null),
                Type::JSON | Type::JSONB => try_get_or_null::<Option<serde_json::Value>>(row, i)
                    .flatten()
                    .unwrap_or(serde_json::Value::Null),
                Type::BYTEA => try_get_or_null::<Option<Vec<u8>>>(row, i)
                    .flatten()
                    .map(|b| serde_json::Value::String(b64(&b)))
                    .unwrap_or(serde_json::Value::Null),
                Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::NAME | Type::UNKNOWN => {
                    try_get_or_null::<Option<String>>(row, i)
                        .flatten()
                        .map(serde_json::Value::String)
                        .unwrap_or(serde_json::Value::Null)
                }
                Type::TIMESTAMPTZ => {
                    // Decode via chrono::DateTime<Utc> (postgres's
                    // `with-chrono-0_4` feature provides FromSql) and
                    // re-format as ISO 8601 — the shape pylon's clients
                    // expect (matches `pylon_kernel::util::now_iso`,
                    // so timestamps round-trip with the same surface
                    // across SQLite + PG).
                    try_get_or_null::<Option<chrono::DateTime<chrono::Utc>>>(row, i)
                        .flatten()
                        .map(|dt| {
                            serde_json::Value::String(dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
                        })
                        .unwrap_or(serde_json::Value::Null)
                }
                Type::TIMESTAMP => try_get_or_null::<Option<chrono::NaiveDateTime>>(row, i)
                    .flatten()
                    .map(|dt| {
                        serde_json::Value::String(dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
                    })
                    .unwrap_or(serde_json::Value::Null),
                Type::DATE => try_get_or_null::<Option<chrono::NaiveDate>>(row, i)
                    .flatten()
                    .map(|d| serde_json::Value::String(d.format("%Y-%m-%d").to_string()))
                    .unwrap_or(serde_json::Value::Null),
                _ => {
                    // Last resort: ask Postgres to render anything else as
                    // text via a stringifying decode through Vec<u8>. If even
                    // that fails (rare — Postgres types not implementing the
                    // text format), fall through to Null with a warning.
                    match row.try_get::<_, Option<String>>(i) {
                        Ok(Some(s)) => serde_json::Value::String(s),
                        Ok(None) => serde_json::Value::Null,
                        Err(_) => match row.try_get::<_, Option<Vec<u8>>>(i) {
                            Ok(Some(bytes)) => serde_json::Value::String(
                                String::from_utf8_lossy(&bytes).into_owned(),
                            ),
                            _ => serde_json::Value::Null,
                        },
                    }
                }
            };
            obj.insert(name, value);
        }
        serde_json::Value::Object(obj)
    }

    fn try_get_or_null<'a, T>(row: &'a postgres::Row, i: usize) -> Option<T>
    where
        T: postgres::types::FromSql<'a>,
    {
        match row.try_get::<_, T>(i) {
            Ok(v) => Some(v),
            Err(e) => {
                tracing::warn!(
                    "[postgres] decode failed for column {} ({}): {e}",
                    i,
                    row.columns()[i].name()
                );
                None
            }
        }
    }

    /// Minimal base64 encoder so we don't need another dependency just for
    /// the BYTEA column edge case.
    fn b64(bytes: &[u8]) -> String {
        const TABLE: &[u8; 64] =
            b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
        let mut out = String::with_capacity((bytes.len() + 2) / 3 * 4);
        let chunks = bytes.chunks(3);
        for chunk in chunks {
            let b = [
                chunk.first().copied().unwrap_or(0),
                chunk.get(1).copied().unwrap_or(0),
                chunk.get(2).copied().unwrap_or(0),
            ];
            out.push(TABLE[(b[0] >> 2) as usize] as char);
            out.push(TABLE[((b[0] & 0x03) << 4 | b[1] >> 4) as usize] as char);
            if chunk.len() > 1 {
                out.push(TABLE[((b[1] & 0x0F) << 2 | b[2] >> 6) as usize] as char);
            } else {
                out.push('=');
            }
            if chunk.len() > 2 {
                out.push(TABLE[(b[2] & 0x3F) as usize] as char);
            } else {
                out.push('=');
            }
        }
        out
    }

    fn pg_err(e: postgres::Error) -> StorageError {
        // postgres::Error's Display is intentionally short ("db error",
        // "connection error" etc.) — the actual SQLSTATE / detail lives
        // on the source chain. Walk the chain so the final message has
        // enough signal to debug a failed insert/update without
        // attaching a debugger.
        use std::error::Error;
        let mut detail = format!("{e}");
        let mut src: Option<&dyn Error> = e.source();
        while let Some(s) = src {
            detail.push_str(": ");
            detail.push_str(&format!("{s}"));
            src = s.source();
        }
        StorageError {
            code: "PG_QUERY_FAILED".into(),
            message: format!("Postgres query failed: {detail}"),
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    /// Hand-rolled fixture that matches the snapshots in the tests
    /// below. Decoupled from any example's `pylon.manifest.json` so
    /// changing an example schema doesn't bleed into adapter tests.
    fn test_manifest() -> AppManifest {
        use pylon_kernel::{ManifestEntity, ManifestField, ManifestIndex};
        let f = |name: &str, ty: &str, opt: bool, uniq: bool| ManifestField {
            name: name.into(),
            field_type: ty.into(),
            optional: opt,
            unique: uniq,
            crdt: None,
        };
        AppManifest {
            manifest_version: 1,
            name: "test".into(),
            version: "0.0.0".into(),
            entities: vec![
                ManifestEntity {
                    name: "User".into(),
                    fields: vec![
                        f("email", "string", false, true),
                        f("displayName", "string", false, false),
                        f("createdAt", "datetime", false, false),
                    ],
                    indexes: vec![],
                    relations: vec![],
                    search: None,
                    crdt: true,
                },
                ManifestEntity {
                    name: "Todo".into(),
                    fields: vec![
                        f("title", "string", false, false),
                        f("done", "bool", false, false),
                        f("userId", "id(User)", false, false),
                        f("createdAt", "datetime", false, false),
                    ],
                    indexes: vec![ManifestIndex {
                        name: "by_user".into(),
                        fields: vec!["userId".into()],
                        unique: false,
                    }],
                    relations: vec![],
                    search: None,
                    crdt: true,
                },
            ],
            queries: vec![],
            actions: vec![],
            policies: vec![],
            routes: vec![],
            auth: Default::default(),
        }
    }

    #[test]
    fn pg_type_mapping() {
        assert_eq!(pg_column_type("string"), "TEXT");
        assert_eq!(pg_column_type("int"), "INTEGER");
        assert_eq!(pg_column_type("float"), "DOUBLE PRECISION");
        assert_eq!(pg_column_type("bool"), "BOOLEAN");
        assert_eq!(pg_column_type("datetime"), "TIMESTAMPTZ");
        assert_eq!(pg_column_type("richtext"), "TEXT");
        assert_eq!(pg_column_type("id(User)"), "TEXT");
    }

    #[test]
    fn quote_ident_simple() {
        assert_eq!(quote_ident("User"), "\"User\"");
        assert_eq!(quote_ident("email"), "\"email\"");
    }

    #[test]
    fn quote_ident_escapes_embedded_double_quotes() {
        assert_eq!(quote_ident("col\"name"), "\"col\"\"name\"");
        assert_eq!(quote_ident("a\"b\"c"), "\"a\"\"b\"\"c\"");
    }

    #[test]
    fn create_table_sql_basic() {
        let fields = vec![
            FieldSpec {
                name: "email".into(),
                field_type: "string".into(),
                optional: false,
                unique: true,
            },
            FieldSpec {
                name: "age".into(),
                field_type: "int".into(),
                optional: true,
                unique: false,
            },
        ];
        let sql = create_table_sql("User", &fields);
        assert_eq!(
            sql,
            "CREATE TABLE IF NOT EXISTS \"User\" (id TEXT PRIMARY KEY NOT NULL, \"email\" TEXT NOT NULL UNIQUE, \"age\" INTEGER)"
        );
    }

    #[test]
    fn create_table_sql_escapes_identifiers() {
        let fields = vec![FieldSpec {
            name: "col\"x".into(),
            field_type: "string".into(),
            optional: false,
            unique: false,
        }];
        let sql = create_table_sql("my\"table", &fields);
        assert!(sql.contains("\"my\"\"table\""));
        assert!(sql.contains("\"col\"\"x\""));
    }

    #[test]
    fn create_index_sql_unique() {
        let sql = create_index_sql("User", "by_email", &["email".into()], true);
        assert_eq!(
            sql,
            "CREATE UNIQUE INDEX IF NOT EXISTS \"User_by_email\" ON \"User\" (\"email\")"
        );
    }

    #[test]
    fn create_index_sql_non_unique() {
        let sql = create_index_sql("Todo", "by_user", &["userId".into()], false);
        assert_eq!(
            sql,
            "CREATE INDEX IF NOT EXISTS \"Todo_by_user\" ON \"Todo\" (\"userId\")"
        );
    }

    #[test]
    fn add_column_sql_basic() {
        let field = FieldSpec {
            name: "bio".into(),
            field_type: "string".into(),
            optional: true,
            unique: false,
        };
        let sql = add_column_sql("User", &field);
        assert_eq!(sql, "ALTER TABLE \"User\" ADD COLUMN \"bio\" TEXT");
    }

    #[test]
    fn plan_from_manifest() {
        let adapter = PostgresAdapter;
        let manifest = test_manifest();
        let plan = adapter.plan_schema(&manifest).unwrap();

        // Should have CreateEntity for User and Todo, plus AddIndex for by_user.
        assert!(plan.operations.iter().any(|op| matches!(
            op,
            SchemaOperation::CreateEntity { name, .. } if name == "User"
        )));
        assert!(plan.operations.iter().any(|op| matches!(
            op,
            SchemaOperation::CreateEntity { name, .. } if name == "Todo"
        )));
        assert!(plan.operations.iter().any(|op| matches!(
            op,
            SchemaOperation::AddIndex { entity, name, .. } if entity == "Todo" && name == "by_user"
        )));
    }

    #[test]
    fn plan_to_sql_produces_statements() {
        let adapter = PostgresAdapter;
        let manifest = test_manifest();
        let plan = adapter.plan_schema(&manifest).unwrap();
        let stmts = plan_to_sql(&plan).unwrap();

        // 2 CREATE TABLE (User, Todo) + 1 CREATE INDEX for Todo.by_user
        // + 1 CREATE INDEX for Todo.by_user_done. The Todo manifest also
        // declares a unique by_email index on User which lands as part of
        // the table. Final count: 2 tables + 2 indexes.
        let create_tables = stmts
            .iter()
            .filter(|s| s.starts_with("CREATE TABLE"))
            .count();
        let create_indexes = stmts
            .iter()
            .filter(|s| s.starts_with("CREATE INDEX") || s.starts_with("CREATE UNIQUE INDEX"))
            .count();
        assert_eq!(create_tables, 2);
        assert!(create_indexes >= 1);
        assert!(stmts[0].starts_with("CREATE TABLE"));
        assert!(stmts[1].starts_with("CREATE TABLE"));
    }

    #[test]
    fn plan_to_sql_rejects_unsupported() {
        let plan = SchemaPlan {
            operations: vec![SchemaOperation::RemoveEntity {
                name: "User".into(),
            }],
        };
        let result = plan_to_sql(&plan);
        assert!(result.is_err());
        assert_eq!(result.unwrap_err().code, "PG_OP_UNSUPPORTED");
    }

    #[test]
    fn apply_not_implemented() {
        let adapter = PostgresAdapter;
        let plan = SchemaPlan {
            operations: vec![SchemaOperation::Noop],
        };
        let result = adapter.apply_schema(&plan);
        assert!(result.is_err());
        assert_eq!(result.unwrap_err().code, "APPLY_NOT_IMPLEMENTED");
    }

    #[test]
    fn sql_uses_quoted_identifiers() {
        let fields = vec![FieldSpec {
            name: "createdAt".into(),
            field_type: "datetime".into(),
            optional: false,
            unique: false,
        }];
        let sql = create_table_sql("User", &fields);
        // Postgres identifiers should be quoted for case-sensitivity.
        assert!(sql.contains("\"User\""));
        assert!(sql.contains("\"createdAt\""));
        assert!(sql.contains("TIMESTAMPTZ"));
    }

    // -- Introspection SQL tests --

    #[test]
    fn introspect_sql_constants_are_valid() {
        // Sanity checks that the SQL strings exist and look reasonable.
        assert!(INTROSPECT_TABLES_SQL.contains("information_schema.tables"));
        assert!(INTROSPECT_COLUMNS_SQL.contains("$1"));
        assert!(INTROSPECT_INDEXES_SQL.contains("$1"));
        assert!(INTROSPECT_TABLES_SQL.contains("_pylon_"));
    }

    // -- Plan from snapshot tests --

    #[test]
    fn plan_from_empty_snapshot_creates_all() {
        let snapshot = crate::SchemaSnapshot { tables: vec![] };
        let manifest = test_manifest();
        let plan = plan_from_snapshot(&snapshot, &manifest);

        assert!(plan.operations.iter().any(|op| matches!(
            op,
            SchemaOperation::CreateEntity { name, .. } if name == "User"
        )));
        assert!(plan.operations.iter().any(|op| matches!(
            op,
            SchemaOperation::CreateEntity { name, .. } if name == "Todo"
        )));
        assert!(plan.operations.iter().any(|op| matches!(
            op,
            SchemaOperation::AddIndex { entity, name, .. } if entity == "Todo" && name == "by_user"
        )));
    }

    #[test]
    fn plan_from_full_snapshot_is_noop() {
        let snapshot = crate::SchemaSnapshot {
            tables: vec![
                crate::TableSnapshot {
                    name: "User".into(),
                    columns: vec![
                        crate::ColumnSnapshot {
                            name: "id".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: true,
                        },
                        crate::ColumnSnapshot {
                            name: "email".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "displayName".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "createdAt".into(),
                            column_type: "TIMESTAMPTZ".into(),
                            notnull: true,
                            primary_key: false,
                        },
                    ],
                    indexes: vec![],
                },
                crate::TableSnapshot {
                    name: "Todo".into(),
                    columns: vec![
                        crate::ColumnSnapshot {
                            name: "id".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: true,
                        },
                        crate::ColumnSnapshot {
                            name: "title".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "done".into(),
                            column_type: "BOOLEAN".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "userId".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "createdAt".into(),
                            column_type: "TIMESTAMPTZ".into(),
                            notnull: true,
                            primary_key: false,
                        },
                    ],
                    indexes: vec![crate::IndexSnapshot {
                        name: "Todo_by_user".into(),
                        columns: vec!["userId".into()],
                        unique: false,
                    }],
                },
            ],
        };
        let manifest = test_manifest();
        let plan = plan_from_snapshot(&snapshot, &manifest);
        assert!(plan.is_empty());
    }

    #[test]
    fn plan_detects_missing_column_in_snapshot() {
        let snapshot = crate::SchemaSnapshot {
            tables: vec![
                crate::TableSnapshot {
                    name: "User".into(),
                    columns: vec![
                        crate::ColumnSnapshot {
                            name: "id".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: true,
                        },
                        crate::ColumnSnapshot {
                            name: "email".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        // missing displayName and createdAt
                    ],
                    indexes: vec![],
                },
                crate::TableSnapshot {
                    name: "Todo".into(),
                    columns: vec![
                        crate::ColumnSnapshot {
                            name: "id".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: true,
                        },
                        crate::ColumnSnapshot {
                            name: "title".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "done".into(),
                            column_type: "BOOLEAN".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "userId".into(),
                            column_type: "TEXT".into(),
                            notnull: true,
                            primary_key: false,
                        },
                        crate::ColumnSnapshot {
                            name: "createdAt".into(),
                            column_type: "TIMESTAMPTZ".into(),
                            notnull: true,
                            primary_key: false,
                        },
                    ],
                    indexes: vec![crate::IndexSnapshot {
                        name: "Todo_by_user".into(),
                        columns: vec!["userId".into()],
                        unique: false,
                    }],
                },
            ],
        };
        let manifest = test_manifest();
        let plan = plan_from_snapshot(&snapshot, &manifest);

        let add_fields: Vec<_> = plan
            .operations
            .iter()
            .filter(|op| matches!(op, SchemaOperation::AddField { .. }))
            .collect();
        assert_eq!(add_fields.len(), 2); // displayName + createdAt
    }

    // -- CRUD helper tests (no live database required) --

    #[test]
    fn json_value_to_string_handles_all_types() {
        assert_eq!(
            json_value_to_string(&serde_json::Value::String("hello".into())),
            "hello"
        );
        assert_eq!(json_value_to_string(&serde_json::json!(42)), "42");
        assert_eq!(json_value_to_string(&serde_json::json!(1.5)), "1.5");
        assert_eq!(json_value_to_string(&serde_json::Value::Bool(true)), "true");
        assert_eq!(
            json_value_to_string(&serde_json::Value::Bool(false)),
            "false"
        );
        assert_eq!(json_value_to_string(&serde_json::Value::Null), "");
        // Arrays and objects get their JSON representation.
        assert_eq!(
            json_value_to_string(&serde_json::json!([1, 2, 3])),
            "[1,2,3]"
        );
        assert_eq!(
            json_value_to_string(&serde_json::json!({"a": 1})),
            "{\"a\":1}"
        );
    }

    #[test]
    fn generate_id_returns_hex_string() {
        let id = generate_id();
        assert!(!id.is_empty());
        // Must be valid hex characters.
        assert!(id.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn generate_id_is_unique_across_calls() {
        let id1 = generate_id();
        let id2 = generate_id();
        assert_ne!(id1, id2);
    }

    #[test]
    fn generate_id_is_lex_sortable() {
        // 1000 IDs back-to-back must come out in monotonically increasing
        // lexicographic order. This is what makes cursor pagination correct.
        let mut ids: Vec<String> = (0..1000).map(|_| generate_id()).collect();
        let sorted = {
            let mut s = ids.clone();
            s.sort();
            s
        };
        assert_eq!(ids, sorted, "generate_id must be lex-monotonic");
        // And every id must be the same width (otherwise lex comparison is
        // wrong at width boundaries).
        let len0 = ids[0].len();
        assert!(ids.iter().all(|id| id.len() == len0));
        ids.dedup();
        assert_eq!(ids.len(), 1000, "no collisions in a tight loop");
    }

    #[test]
    fn build_insert_sql_simple() {
        let data = serde_json::json!({
            "email": "alice@example.com",
            "displayName": "Alice"
        });
        let (sql, values) = build_insert_sql("User", &data).unwrap();

        assert!(sql.starts_with("INSERT INTO \"User\""));
        assert!(sql.contains("id"));
        assert!(sql.contains("$1"));
        assert!(sql.contains("$2"));
        assert!(sql.contains("$3"));
        // First value is the generated ID — JsonParam::Text variant.
        match &values[0] {
            JsonParam::Text(s) => assert!(!s.is_empty()),
            other => panic!("expected Text id param, got {other:?}"),
        }
        assert_eq!(values.len(), 3); // id + 2 fields
    }

    #[test]
    fn build_insert_sql_preserves_json_types() {
        let data = serde_json::json!({
            "n": 42,
            "f": 1.5,
            "b": true,
            "s": "hi",
            "z": null,
        });
        let (_sql, values) = build_insert_sql("T", &data).unwrap();
        // values[0] is the id; remaining are in BTreeMap order ("b","f","n","s","z").
        let kinds: Vec<&JsonParam> = values.iter().skip(1).collect();
        assert!(matches!(kinds[0], JsonParam::Bool(true)));
        assert!(matches!(kinds[1], JsonParam::Float(_)));
        assert!(matches!(kinds[2], JsonParam::Int(42)));
        assert!(matches!(kinds[3], JsonParam::Text(_)));
        assert!(matches!(kinds[4], JsonParam::Null));
    }

    #[test]
    fn build_insert_sql_quotes_column_names() {
        let data = serde_json::json!({"createdAt": "2026-01-01"});
        let (sql, _) = build_insert_sql("Todo", &data).unwrap();
        assert!(sql.contains("\"createdAt\""));
        assert!(sql.contains("\"Todo\""));
    }

    #[test]
    fn build_insert_sql_rejects_non_object() {
        let data = serde_json::json!("not an object");
        let result = build_insert_sql("User", &data);
        assert!(result.is_err());
        assert_eq!(result.unwrap_err().code, "PG_INVALID_DATA");
    }

    #[test]
    fn build_update_sql_simple() {
        let data = serde_json::json!({
            "displayName": "Bob",
            "email": "bob@example.com"
        });
        let (sql, values) = build_update_sql("User", "abc123", &data).unwrap();

        assert!(sql.starts_with("UPDATE \"User\" SET"));
        assert!(sql.contains("WHERE id = $1"));
        assert!(sql.contains("$2"));
        assert!(sql.contains("$3"));
        match &values[0] {
            JsonParam::Text(s) => assert_eq!(s, "abc123"),
            other => panic!("expected Text id param, got {other:?}"),
        }
        assert_eq!(values.len(), 3); // id + 2 fields
    }

    #[test]
    fn build_update_sql_quotes_column_names() {
        let data = serde_json::json!({"displayName": "Carol"});
        let (sql, _) = build_update_sql("User", "id1", &data).unwrap();
        assert!(sql.contains("\"displayName\" = $2"));
    }

    #[test]
    fn build_update_sql_rejects_non_object() {
        let data = serde_json::json!(42);
        let result = build_update_sql("User", "id1", &data);
        assert!(result.is_err());
        assert_eq!(result.unwrap_err().code, "PG_INVALID_DATA");
    }

    #[test]
    fn build_update_sql_rejects_empty_object() {
        let data = serde_json::json!({});
        let err = build_update_sql("User", "id1", &data).unwrap_err();
        assert_eq!(err.code, "PG_INVALID_DATA");
        assert!(err.message.contains("at least one field"));
    }
}