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`.
fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

// ---------------------------------------------------------------------------
// 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::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 id = generate_id();
    let obj = data.as_object().ok_or_else(|| StorageError {
        code: "PG_INVALID_DATA".into(),
        message: "Insert data must be a JSON object".into(),
    })?;

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

    for (i, (key, val)) in obj.iter().enumerate() {
        col_names.push(quote_ident(key));
        placeholders.push(format!("${}", i + 2));
        values.push(JsonParam::from_json(val));
    }

    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())];

    for (i, (key, val)) in obj.iter().enumerate() {
        set_clauses.push(format!("{} = ${}", quote_ident(key), i + 2));
        values.push(JsonParam::from_json(val));
    }

    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,
    };

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

    impl LivePostgresAdapter {
        /// Connect to a Postgres database.
        pub fn connect(url: &str) -> Result<Self, StorageError> {
            let client =
                postgres::Client::connect(url, postgres::NoTls).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 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" => {
                        // FTS5 (SQLite) has no portable equivalent in
                        // Postgres without an explicit tsvector column.
                        // Return a clear error rather than silently
                        // ignoring the operator — callers that hit this
                        // need to either branch on backend or define a
                        // tsvector + GIN index in their schema.
                        return Err(StorageError {
                            code: "SEARCH_NOT_SUPPORTED".into(),
                            message: "$search is SQLite-FTS5-only; use a Postgres tsvector column with the storage adapter's full-text path"
                                .into(),
                        });
                    }
                    "$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,
            );

            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())
        }

        /// 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 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 pg_params = as_pg_params(&params);
            let rows = self
                .client
                .query(sql.as_str(), &pg_params)
                .map_err(pg_err)?;

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

            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);
                }
            }
            Ok(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)
    }

    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"));
    }
}