//! SQLite database access via an async worker.
//!
//! The application talks to SQLite through a single
//! request/response channel. A dedicated OS thread owns the
//! `rusqlite::Connection` (which is `Send` but `!Sync` and uses
//! a synchronous API), receives `Request` messages, and replies
//! on per-request `oneshot` channels.
//!
//! This shape was chosen up front in Phase 2 of the parser/DB
//! iteration so that B3 (query timeout/cancellation) and U1
//! (snapshot capture) drop in without an architectural refactor.
//!
//! ## STRICT and foreign keys
//!
//! Per ADR-0002, every table is created with the `STRICT`
//! keyword and the connection-level `PRAGMA foreign_keys` is
//! enabled at open time.
//!
//! ## Error handling
//!
//! Database errors flow through `DbError`, which carries a
//! coarse `kind` to support the future friendly-error layer
//! (H1). For now `friendly_message()` is a passthrough; when H1
//! lands the body of that method becomes the translation table.

use std::fmt::Write as _;
use std::path::Path;
use std::thread;

use rusqlite::Connection;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, info, warn};

use crate::dsl::action::ReferentialAction;
use crate::dsl::command::{RelationshipSelector, RowFilter};
use crate::dsl::ColumnSpec;
use crate::dsl::shortid;
use crate::dsl::types::Type;
use crate::dsl::value::{Bound, Value, ValueError};
use crate::persistence::{
    CellValue, ColumnSchema, Persistence, PersistenceError, RelationshipSchema, SchemaSnapshot,
    TableSchema, TableSnapshot,
};

/// Inbox capacity. The worker is fast enough that this rarely
/// matters; `64` is a generous head-room for bursts.
const REQUEST_CHANNEL_CAPACITY: usize = 64;

/// In-process handle for the database. Cheap to clone.
#[derive(Debug, Clone)]
pub struct Database {
    inbox: mpsc::Sender<Request>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TableDescription {
    pub name: String,
    pub columns: Vec<ColumnDescription>,
    /// Relationships where *this* table is the child (holds the
    /// FK column referencing another table).
    pub outbound_relationships: Vec<RelationshipEnd>,
    /// Relationships where *this* table is the parent (some
    /// other table's column references one of ours).
    pub inbound_relationships: Vec<RelationshipEnd>,
}

/// One end of a relationship as seen from the table being
/// described.
///
/// Used for both outbound (this table is the child, holding the
/// FK column) and inbound (this table is the parent being
/// referenced) sides; the field meanings flip per side.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RelationshipEnd {
    /// User-facing name of the relationship (auto-generated or
    /// user-supplied at creation time).
    pub name: String,
    /// The other table involved.
    pub other_table: String,
    /// The column on the other table.
    pub other_column: String,
    /// The column on *this* table.
    pub local_column: String,
    pub on_delete: ReferentialAction,
    pub on_update: ReferentialAction,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ColumnDescription {
    pub name: String,
    /// The user-facing type the column was declared as, recovered
    /// from our internal column-metadata table. Always populated
    /// for tables created through the DSL. `None` only for the
    /// edge case of a foreign-attached database whose tables we
    /// did not create — not achievable in the current flow.
    pub user_type: Option<Type>,
    /// The SQLite-side type as reported by `PRAGMA table_info`
    /// (e.g. `INTEGER`, `TEXT`). Kept for diagnostics and as a
    /// fall-back when `user_type` is not available; the UI
    /// prefers `user_type` when rendering.
    pub sqlite_type: String,
    pub notnull: bool,
    pub primary_key: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum DbError {
    #[error("database error: {message}")]
    Sqlite { message: String, kind: SqliteErrorKind },
    #[error("operation not supported: {0}")]
    Unsupported(String),
    #[error("invalid value: {0}")]
    InvalidValue(String),
    #[error("could not {operation} `{path}`: {message}")]
    PersistenceFatal {
        operation: &'static str,
        path: std::path::PathBuf,
        message: String,
    },
    #[error("database worker is no longer available")]
    WorkerGone,
    #[error("io error: {0}")]
    Io(String),
}

/// Result of a query / show-data call (schema-less display rows).
/// `None` cells render as NULL; `Some(s)` renders as the string.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DataResult {
    pub table_name: String,
    pub columns: Vec<String>,
    pub rows: Vec<Vec<Option<String>>>,
}

/// Outcome of a successful INSERT — a count plus the new row(s)
/// fetched immediately after so the user can see what landed
/// (auto-filled IDs, generated shortids, etc.).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InsertResult {
    pub rows_affected: usize,
    pub data: DataResult,
}

/// Outcome of a successful UPDATE — a count plus the rows that
/// matched (and were updated). Captured by rowid so that even an
/// UPDATE which changes the WHERE-target column still finds the
/// post-update rows.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpdateResult {
    pub rows_affected: usize,
    pub data: DataResult,
}

/// Outcome of a successful DELETE — the directly-deleted-row
/// count plus any cascade effects observed in child tables.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeleteResult {
    pub rows_affected: usize,
    pub cascade: Vec<CascadeEffect>,
}

/// One observed change in a child table caused by referential
/// action on a parent-side DELETE. Detected by row-count diffing
/// the child table immediately before and after the delete.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CascadeEffect {
    pub relationship_name: String,
    pub child_table: String,
    pub rows_changed: i64,
    pub action: ReferentialAction,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SqliteErrorKind {
    /// `UNIQUE` constraint, including duplicate primary key.
    UniqueViolation,
    /// Referenced or operated-on table does not exist.
    NoSuchTable,
    /// Operated-on column does not exist.
    NoSuchColumn,
    /// Object (table, index, etc.) already exists.
    AlreadyExists,
    /// Catch-all.
    Other,
}

impl DbError {
    /// Placeholder for the H1 friendly-error layer. Today this
    /// returns the same string as [`std::fmt::Display`]; when H1
    /// lands the body becomes the translation logic and
    /// callsites do not need to change.
    #[must_use]
    pub fn friendly_message(&self) -> String {
        self.to_string()
    }

    fn from_rusqlite(err: rusqlite::Error) -> Self {
        let message = err.to_string();
        let kind = classify_sqlite_error(&err, &message);
        Self::Sqlite { message, kind }
    }

    fn from_persistence(err: PersistenceError) -> Self {
        Self::PersistenceFatal {
            operation: err.operation(),
            path: err.path().to_path_buf(),
            message: err.to_string(),
        }
    }

    /// Whether this error means the application cannot
    /// continue and must quit (per ADR-0015 §8). The runtime
    /// surfaces these as fatal banners.
    #[must_use]
    pub const fn is_fatal(&self) -> bool {
        matches!(self, Self::PersistenceFatal { .. })
    }
}

fn classify_sqlite_error(err: &rusqlite::Error, message: &str) -> SqliteErrorKind {
    use rusqlite::ErrorCode;
    if let rusqlite::Error::SqliteFailure(code, _) = err
        && code.code == ErrorCode::ConstraintViolation
    {
        return SqliteErrorKind::UniqueViolation;
    }
    let lowered = message.to_ascii_lowercase();
    if lowered.contains("no such table") {
        SqliteErrorKind::NoSuchTable
    } else if lowered.contains("no such column") {
        SqliteErrorKind::NoSuchColumn
    } else if lowered.contains("already exists") {
        SqliteErrorKind::AlreadyExists
    } else {
        SqliteErrorKind::Other
    }
}

/// Internal request type — kept private so the channel protocol
/// is not part of the public API.
#[derive(Debug)]
enum Request {
    CreateTable {
        name: String,
        columns: Vec<ColumnSpec>,
        primary_key: Vec<String>,
        source: Option<String>,
        reply: oneshot::Sender<Result<TableDescription, DbError>>,
    },
    DropTable {
        name: String,
        source: Option<String>,
        reply: oneshot::Sender<Result<(), DbError>>,
    },
    AddColumn {
        table: String,
        column: String,
        ty: Type,
        source: Option<String>,
        reply: oneshot::Sender<Result<TableDescription, DbError>>,
    },
    ListTables {
        reply: oneshot::Sender<Result<Vec<String>, DbError>>,
    },
    DescribeTable {
        name: String,
        source: Option<String>,
        reply: oneshot::Sender<Result<TableDescription, DbError>>,
    },
    AddRelationship {
        name: Option<String>,
        parent_table: String,
        parent_column: String,
        child_table: String,
        child_column: String,
        on_delete: ReferentialAction,
        on_update: ReferentialAction,
        create_fk: bool,
        source: Option<String>,
        reply: oneshot::Sender<Result<TableDescription, DbError>>,
    },
    DropRelationship {
        selector: RelationshipSelector,
        source: Option<String>,
        reply: oneshot::Sender<Result<Option<TableDescription>, DbError>>,
    },
    Insert {
        table: String,
        columns: Option<Vec<String>>,
        values: Vec<Value>,
        source: Option<String>,
        reply: oneshot::Sender<Result<InsertResult, DbError>>,
    },
    Update {
        table: String,
        assignments: Vec<(String, Value)>,
        filter: RowFilter,
        source: Option<String>,
        reply: oneshot::Sender<Result<UpdateResult, DbError>>,
    },
    Delete {
        table: String,
        filter: RowFilter,
        source: Option<String>,
        reply: oneshot::Sender<Result<DeleteResult, DbError>>,
    },
    QueryData {
        table: String,
        source: Option<String>,
        reply: oneshot::Sender<Result<DataResult, DbError>>,
    },
}

impl Database {
    /// Open a database without per-command persistence.
    ///
    /// The path may be a filesystem location or `":memory:"`
    /// for an ephemeral in-memory database. The connection is
    /// moved onto a dedicated worker thread. With no
    /// persistence handle, the YAML/CSV/`history.log` writes
    /// are skipped — useful for unit tests that exercise the
    /// SQLite layer in isolation.
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, DbError> {
        Self::open_inner(path, None)
    }

    /// Open a database with per-command persistence wired in
    /// (ADR-0015 §6). Every successful user-issued mutation
    /// writes through to `project.yaml`, the affected
    /// `data/<table>.csv` files, and `history.log` *before*
    /// the SQLite tx commits.
    pub fn open_with_persistence<P: AsRef<Path>>(
        path: P,
        persistence: Persistence,
    ) -> Result<Self, DbError> {
        Self::open_inner(path, Some(persistence))
    }

    fn open_inner<P: AsRef<Path>>(
        path: P,
        persistence: Option<Persistence>,
    ) -> Result<Self, DbError> {
        let path_display = path.as_ref().to_string_lossy().into_owned();
        let conn = match path.as_ref().to_str() {
            Some(":memory:") => Connection::open_in_memory(),
            _ => Connection::open(path.as_ref()),
        }
        .map_err(DbError::from_rusqlite)?;
        info!(path = %path_display, "opened database");
        configure_connection(&conn).map_err(DbError::from_rusqlite)?;

        let (tx, rx) = mpsc::channel::<Request>(REQUEST_CHANNEL_CAPACITY);
        thread::Builder::new()
            .name("rdbms-db-worker".to_string())
            .spawn(move || worker_loop(conn, persistence, rx))
            .map_err(|e| DbError::Io(e.to_string()))?;

        Ok(Self { inbox: tx })
    }

    pub async fn create_table(
        &self,
        name: String,
        columns: Vec<ColumnSpec>,
        primary_key: Vec<String>,
        source: Option<String>,
    ) -> Result<TableDescription, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::CreateTable {
            name,
            columns,
            primary_key,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn drop_table(&self, name: String, source: Option<String>) -> Result<(), DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::DropTable { name, source, reply }).await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn add_column(
        &self,
        table: String,
        column: String,
        ty: Type,
        source: Option<String>,
    ) -> Result<TableDescription, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::AddColumn {
            table,
            column,
            ty,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn list_tables(&self) -> Result<Vec<String>, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::ListTables { reply }).await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn describe_table(
        &self,
        name: String,
        source: Option<String>,
    ) -> Result<TableDescription, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::DescribeTable {
            name,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn add_relationship(
        &self,
        name: Option<String>,
        parent_table: String,
        parent_column: String,
        child_table: String,
        child_column: String,
        on_delete: ReferentialAction,
        on_update: ReferentialAction,
        create_fk: bool,
        source: Option<String>,
    ) -> Result<TableDescription, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::AddRelationship {
            name,
            parent_table,
            parent_column,
            child_table,
            child_column,
            on_delete,
            on_update,
            create_fk,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn drop_relationship(
        &self,
        selector: RelationshipSelector,
        source: Option<String>,
    ) -> Result<Option<TableDescription>, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::DropRelationship {
            selector,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn insert(
        &self,
        table: String,
        columns: Option<Vec<String>>,
        values: Vec<Value>,
        source: Option<String>,
    ) -> Result<InsertResult, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::Insert {
            table,
            columns,
            values,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn update(
        &self,
        table: String,
        assignments: Vec<(String, Value)>,
        filter: RowFilter,
        source: Option<String>,
    ) -> Result<UpdateResult, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::Update {
            table,
            assignments,
            filter,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn delete(
        &self,
        table: String,
        filter: RowFilter,
        source: Option<String>,
    ) -> Result<DeleteResult, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::Delete {
            table,
            filter,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    pub async fn query_data(
        &self,
        table: String,
        source: Option<String>,
    ) -> Result<DataResult, DbError> {
        let (reply, recv) = oneshot::channel();
        self.send(Request::QueryData {
            table,
            source,
            reply,
        })
        .await?;
        recv.await.map_err(|_| DbError::WorkerGone)?
    }

    async fn send(&self, req: Request) -> Result<(), DbError> {
        self.inbox.send(req).await.map_err(|_| DbError::WorkerGone)
    }
}

/// Internal table tracking the user-facing type each column was
/// declared with. The structure view consults this so users see
/// the names they typed (`serial`, `date`) rather than SQLite's
/// erased forms (`INTEGER`, `TEXT`) — closing a Q3 / ADR-0005
/// promise. Track 2's project file format is the long-term home
/// for this metadata; this table is the in-database mirror that
/// makes round-trip rendering work today.
const META_TABLE: &str = "__rdbms_playground_columns";
const REL_TABLE: &str = "__rdbms_playground_relationships";
/// Single-row key/value table that carries project metadata
/// the YAML serializer needs to round-trip — currently
/// `created_at`. Created on first connect and only ever
/// written by us; the user never touches it directly.
const META_PROJECT_TABLE: &str = "__rdbms_playground_meta";

fn configure_connection(conn: &Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(&format!(
        "PRAGMA foreign_keys = ON;\n\
         CREATE TABLE IF NOT EXISTS {META_TABLE} (\n\
             table_name TEXT NOT NULL,\n\
             column_name TEXT NOT NULL,\n\
             user_type TEXT NOT NULL,\n\
             PRIMARY KEY (table_name, column_name)\n\
         ) STRICT;\n\
         CREATE TABLE IF NOT EXISTS {REL_TABLE} (\n\
             name TEXT NOT NULL UNIQUE,\n\
             parent_table TEXT NOT NULL,\n\
             parent_column TEXT NOT NULL,\n\
             child_table TEXT NOT NULL,\n\
             child_column TEXT NOT NULL,\n\
             on_delete TEXT NOT NULL,\n\
             on_update TEXT NOT NULL,\n\
             PRIMARY KEY (child_table, child_column)\n\
         ) STRICT;\n\
         CREATE TABLE IF NOT EXISTS {META_PROJECT_TABLE} (\n\
             key TEXT NOT NULL PRIMARY KEY,\n\
             value TEXT NOT NULL\n\
         ) STRICT;"
    ))?;
    // Seed `created_at` once. INSERT OR IGNORE keeps the value
    // stable across re-opens of the same database.
    conn.execute(
        &format!(
            "INSERT OR IGNORE INTO {META_PROJECT_TABLE} (key, value) VALUES ('created_at', ?1)"
        ),
        [iso8601_now()],
    )?;
    Ok(())
}

/// Current UTC time as ISO-8601 with second precision.
/// Duplicates the helper in `persistence::history` rather
/// than depending on it, since this code path runs at
/// connection setup before the rest of the worker is up.
fn iso8601_now() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let secs = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    let day_secs = secs.rem_euclid(86_400);
    let h = day_secs / 3600;
    let m = (day_secs % 3600) / 60;
    let s = day_secs % 60;
    let days = secs.div_euclid(86_400);
    let z = days + 719_468;
    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
    let doe = (z - era * 146_097) as u64;
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe as i64 + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m_cal = if mp < 10 { mp + 3 } else { mp - 9 };
    let y_cal = if m_cal <= 2 { y + 1 } else { y };
    format!("{y_cal:04}-{m_cal:02}-{d:02}T{h:02}:{m:02}:{s:02}Z")
}

fn worker_loop(
    conn: Connection,
    persistence: Option<Persistence>,
    mut rx: mpsc::Receiver<Request>,
) {
    debug!("db worker started");
    while let Some(req) = rx.blocking_recv() {
        handle_request(&conn, persistence.as_ref(), req);
    }
    debug!("db worker exiting");
}

fn handle_request(conn: &Connection, persistence: Option<&Persistence>, req: Request) {
    match req {
        Request::CreateTable {
            name,
            columns,
            primary_key,
            source,
            reply,
        } => {
            let _ = reply.send(do_create_table(
                conn,
                persistence,
                source.as_deref(),
                &name,
                &columns,
                &primary_key,
            ));
        }
        Request::DropTable {
            name,
            source,
            reply,
        } => {
            let _ = reply.send(do_drop_table(conn, persistence, source.as_deref(), &name));
        }
        Request::AddColumn {
            table,
            column,
            ty,
            source,
            reply,
        } => {
            let _ = reply.send(do_add_column(
                conn,
                persistence,
                source.as_deref(),
                &table,
                &column,
                ty,
            ));
        }
        Request::ListTables { reply } => {
            let _ = reply.send(do_list_tables(conn));
        }
        Request::DescribeTable {
            name,
            source,
            reply,
        } => {
            let _ = reply.send(do_describe_table_request(
                conn,
                persistence,
                source.as_deref(),
                &name,
            ));
        }
        Request::AddRelationship {
            name,
            parent_table,
            parent_column,
            child_table,
            child_column,
            on_delete,
            on_update,
            create_fk,
            source,
            reply,
        } => {
            let _ = reply.send(do_add_relationship(
                conn,
                persistence,
                source.as_deref(),
                name.as_deref(),
                &parent_table,
                &parent_column,
                &child_table,
                &child_column,
                on_delete,
                on_update,
                create_fk,
            ));
        }
        Request::DropRelationship {
            selector,
            source,
            reply,
        } => {
            let _ = reply.send(do_drop_relationship(
                conn,
                persistence,
                source.as_deref(),
                &selector,
            ));
        }
        Request::Insert {
            table,
            columns,
            values,
            source,
            reply,
        } => {
            let _ = reply.send(do_insert(
                conn,
                persistence,
                source.as_deref(),
                &table,
                columns.as_deref(),
                &values,
            ));
        }
        Request::Update {
            table,
            assignments,
            filter,
            source,
            reply,
        } => {
            let _ = reply.send(do_update(
                conn,
                persistence,
                source.as_deref(),
                &table,
                &assignments,
                &filter,
            ));
        }
        Request::Delete {
            table,
            filter,
            source,
            reply,
        } => {
            let _ = reply.send(do_delete(
                conn,
                persistence,
                source.as_deref(),
                &table,
                &filter,
            ));
        }
        Request::QueryData {
            table,
            source,
            reply,
        } => {
            let _ = reply.send(do_query_data_request(
                conn,
                persistence,
                source.as_deref(),
                &table,
            ));
        }
    }
}

/// Set of changes a mutation made, used by the post-mutation
/// persistence phase to know which text targets need refreshing
/// (ADR-0015 §6).
#[derive(Debug, Default)]
struct Changes {
    /// Schema (tables / columns / relationships) was modified —
    /// `project.yaml` needs to be rewritten.
    schema_dirty: bool,
    /// Tables whose row data changed and whose CSVs need to be
    /// re-emitted from the post-mutation state.
    rewritten_tables: Vec<String>,
    /// Tables that were dropped — their CSVs should be removed.
    deleted_tables: Vec<String>,
}

/// Drive the post-mutation persistence phase: write the YAML
/// schema, rewrite affected CSVs, append `history.log`. Called
/// before `tx.commit()` so a failure here causes the SQLite
/// transaction to roll back automatically (the `Drop` impl on
/// `Transaction` rolls back on drop).
///
/// Read-only requests (no schema change, no row writes, no
/// drops) still use this to append `history.log` if `source`
/// is set; they pass an empty `Changes`.
fn finalize_persistence(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    changes: &Changes,
) -> Result<(), DbError> {
    let Some(p) = persistence else {
        return Ok(());
    };
    if changes.schema_dirty {
        let schema = read_schema_snapshot(conn)?;
        p.write_schema(&schema).map_err(DbError::from_persistence)?;
    }
    for table in &changes.rewritten_tables {
        if let Some(snapshot) = read_table_snapshot(conn, table)? {
            p.write_table_data(&snapshot)
                .map_err(DbError::from_persistence)?;
        }
    }
    for table in &changes.deleted_tables {
        p.delete_table_data(table)
            .map_err(DbError::from_persistence)?;
    }
    if let Some(text) = source {
        p.append_history(text)
            .map_err(DbError::from_persistence)?;
    }
    Ok(())
}

/// Read the full user-facing schema (tables, columns,
/// relationships, project metadata) from the database. Reads
/// committed *or* in-tx state because SQLite presents the
/// same connection's writes back through the same connection.
fn read_schema_snapshot(conn: &Connection) -> Result<SchemaSnapshot, DbError> {
    let table_names = do_list_tables(conn)?;
    let mut tables: Vec<TableSchema> = Vec::with_capacity(table_names.len());
    for name in &table_names {
        let read = read_schema(conn, name)?;
        let columns: Vec<ColumnSchema> = read
            .columns
            .iter()
            .map(|c| ColumnSchema {
                name: c.name.clone(),
                // user_type is always populated for tables we
                // created; the fallback is defensive.
                user_type: c.user_type.unwrap_or(Type::Text),
            })
            .collect();
        tables.push(TableSchema {
            name: name.clone(),
            primary_key: read.primary_key.clone(),
            columns,
        });
    }

    let relationships = read_all_relationships(conn)?;
    let created_at = read_project_created_at(conn)?;
    Ok(SchemaSnapshot {
        created_at,
        tables,
        relationships,
    })
}

fn read_all_relationships(conn: &Connection) -> Result<Vec<RelationshipSchema>, DbError> {
    let mut stmt = conn
        .prepare(&format!(
            "SELECT name, parent_table, parent_column, child_table, child_column, \
                    on_delete, on_update \
             FROM {REL_TABLE} \
             ORDER BY rowid"
        ))
        .map_err(DbError::from_rusqlite)?;
    let rows = stmt
        .query_map([], |row| {
            Ok(RelationshipSchema {
                name: row.get(0)?,
                parent_table: row.get(1)?,
                parent_column: row.get(2)?,
                child_table: row.get(3)?,
                child_column: row.get(4)?,
                on_delete: parse_action_from_sqlite(row.get::<_, String>(5)?.as_str()),
                on_update: parse_action_from_sqlite(row.get::<_, String>(6)?.as_str()),
            })
        })
        .map_err(DbError::from_rusqlite)?;
    let mut out = Vec::new();
    for row in rows {
        out.push(row.map_err(DbError::from_rusqlite)?);
    }
    Ok(out)
}

fn read_project_created_at(conn: &Connection) -> Result<String, DbError> {
    let value: Option<String> = conn
        .query_row(
            &format!("SELECT value FROM {META_PROJECT_TABLE} WHERE key = 'created_at'"),
            [],
            |row| row.get(0),
        )
        .or_else(|e| match e {
            rusqlite::Error::QueryReturnedNoRows => Ok(None),
            other => Err(other),
        })
        .map_err(DbError::from_rusqlite)?;
    Ok(value.unwrap_or_else(|| "1970-01-01T00:00:00Z".to_string()))
}

/// Read a single table's full row data, returning `None` if
/// the table no longer exists (e.g. a recent `drop_table`).
fn read_table_snapshot(
    conn: &Connection,
    table: &str,
) -> Result<Option<TableSnapshot>, DbError> {
    if !user_table_exists(conn, table)? {
        return Ok(None);
    }
    let read = read_schema(conn, table)?;
    let columns: Vec<ColumnSchema> = read
        .columns
        .iter()
        .map(|c| ColumnSchema {
            name: c.name.clone(),
            user_type: c.user_type.unwrap_or(Type::Text),
        })
        .collect();
    let column_idents: Vec<String> = read
        .columns
        .iter()
        .map(|c| quote_ident(&c.name))
        .collect();
    let sql = format!(
        "SELECT {} FROM {} ORDER BY rowid",
        column_idents.join(", "),
        quote_ident(table),
    );
    let mut stmt = conn.prepare(&sql).map_err(DbError::from_rusqlite)?;
    let column_count = read.columns.len();
    let mut rows: Vec<Vec<CellValue>> = Vec::new();
    let mut iter = stmt.query([]).map_err(DbError::from_rusqlite)?;
    while let Some(row) = iter.next().map_err(DbError::from_rusqlite)? {
        let mut record: Vec<CellValue> = Vec::with_capacity(column_count);
        for i in 0..column_count {
            record.push(row_value_to_cell(row, i)?);
        }
        rows.push(record);
    }
    Ok(Some(TableSnapshot {
        name: table.to_string(),
        columns,
        rows,
    }))
}

fn user_table_exists(conn: &Connection, table: &str) -> Result<bool, DbError> {
    let count: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_schema \
             WHERE type = 'table' AND name = ?1 \
               AND substr(name, 1, 8) != '__rdbms_'",
            [table],
            |row| row.get(0),
        )
        .map_err(DbError::from_rusqlite)?;
    Ok(count > 0)
}

fn row_value_to_cell(row: &rusqlite::Row<'_>, idx: usize) -> Result<CellValue, DbError> {
    use rusqlite::types::ValueRef;
    let v = row.get_ref(idx).map_err(DbError::from_rusqlite)?;
    Ok(match v {
        ValueRef::Null => CellValue::Null,
        ValueRef::Integer(n) => CellValue::Integer(n),
        ValueRef::Real(f) => CellValue::Real(f),
        ValueRef::Text(bytes) => {
            CellValue::Text(String::from_utf8_lossy(bytes).into_owned())
        }
        ValueRef::Blob(bytes) => CellValue::Blob(bytes.to_vec()),
    })
}

/// Quote an identifier for safe inclusion in DDL. Doubles any
/// embedded double-quotes per SQL convention.
fn quote_ident(name: &str) -> String {
    let mut out = String::with_capacity(name.len() + 2);
    out.push('"');
    for c in name.chars() {
        if c == '"' {
            out.push_str("\"\"");
        } else {
            out.push(c);
        }
    }
    out.push('"');
    out
}

fn do_create_table(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    name: &str,
    columns: &[ColumnSpec],
    primary_key: &[String],
) -> Result<TableDescription, DbError> {
    if columns.is_empty() {
        // SQLite requires at least one column. The DSL grammar
        // already prevents this, but defending here too keeps
        // the executor honest if anyone synthesises a Command
        // directly (tests, future scripting).
        return Err(DbError::Unsupported(
            "tables need at least one column".to_string(),
        ));
    }

    // Generate the column list. For a single-column PK we inline
    // `PRIMARY KEY` on the column itself, which is required for
    // SQLite STRICT tables to give an `INTEGER PRIMARY KEY`
    // column its rowid-alias semantics. For compound PKs (or
    // when the single PK is on a non-first column) we emit a
    // table-level constraint.
    let single_inline_pk = primary_key.len() == 1 && columns.len() == 1
        && primary_key[0] == columns[0].name;

    let mut column_clauses: Vec<String> = Vec::with_capacity(columns.len());
    for col in columns {
        let mut clause = format!(
            "{ident} {sqlite_type}",
            ident = quote_ident(&col.name),
            sqlite_type = col.ty.sqlite_strict_type(),
        );
        if single_inline_pk {
            clause.push_str(" PRIMARY KEY");
        }
        column_clauses.push(clause);
    }

    let mut ddl = format!(
        "CREATE TABLE {ident} ({columns}",
        ident = quote_ident(name),
        columns = column_clauses.join(", "),
    );

    if !single_inline_pk && !primary_key.is_empty() {
        let pk_idents: Vec<String> = primary_key.iter().map(|n| quote_ident(n)).collect();
        ddl.push_str(", PRIMARY KEY (");
        ddl.push_str(&pk_idents.join(", "));
        ddl.push(')');
    }

    ddl.push_str(") STRICT;");
    debug!(ddl = %ddl, "create_table");

    // Wrap the table-creation DDL and the metadata inserts in a
    // single transaction so they commit atomically — if either
    // step fails, neither side persists.
    let tx = conn
        .unchecked_transaction()
        .map_err(DbError::from_rusqlite)?;
    tx.execute_batch(&ddl).map_err(DbError::from_rusqlite)?;
    {
        let mut stmt = tx
            .prepare(&format!(
                "INSERT INTO {META_TABLE} (table_name, column_name, user_type) \
                 VALUES (?1, ?2, ?3);"
            ))
            .map_err(DbError::from_rusqlite)?;
        for col in columns {
            stmt.execute([name, col.name.as_str(), col.ty.keyword()])
                .map_err(DbError::from_rusqlite)?;
        }
    }
    let description = do_describe_table(conn, name)?;
    let changes = Changes {
        schema_dirty: true,
        rewritten_tables: vec![name.to_string()],
        ..Changes::default()
    };
    finalize_persistence(conn, persistence, source, &changes)?;
    tx.commit().map_err(DbError::from_rusqlite)?;
    Ok(description)
}

fn do_drop_table(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    name: &str,
) -> Result<(), DbError> {
    // Refuse the drop while any *other* table still has a
    // relationship pointing at this one — dropping the parent
    // would leave dangling FK constraints in the children. The
    // user is told which relationships to drop first.
    let inbound = read_relationships_inbound(conn, name)?;
    if !inbound.is_empty() {
        let names: Vec<&str> = inbound.iter().map(|r| r.name.as_str()).collect();
        return Err(DbError::Unsupported(format!(
            "cannot drop `{name}`: it is referenced by relationship(s) [{}]. \
             Drop those relationships first.",
            names.join(", ")
        )));
    }

    let ddl = format!("DROP TABLE {ident};", ident = quote_ident(name));
    debug!(ddl = %ddl, "drop_table");
    let tx = conn
        .unchecked_transaction()
        .map_err(DbError::from_rusqlite)?;
    tx.execute_batch(&ddl).map_err(DbError::from_rusqlite)?;
    tx.execute(
        &format!("DELETE FROM {META_TABLE} WHERE table_name = ?1;"),
        [name],
    )
    .map_err(DbError::from_rusqlite)?;
    // Outbound relationships are gone with the table — clean
    // their metadata too.
    tx.execute(
        &format!("DELETE FROM {REL_TABLE} WHERE child_table = ?1;"),
        [name],
    )
    .map_err(DbError::from_rusqlite)?;
    let changes = Changes {
        schema_dirty: true,
        rewritten_tables: Vec::new(),
        deleted_tables: vec![name.to_string()],
    };
    finalize_persistence(conn, persistence, source, &changes)?;
    tx.commit().map_err(DbError::from_rusqlite)?;
    Ok(())
}

fn do_add_column(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    table: &str,
    column: &str,
    ty: Type,
) -> Result<TableDescription, DbError> {
    if ty == Type::Serial {
        return Err(DbError::Unsupported(
            "the 'serial' type carries auto-increment primary-key semantics \
             that SQLite's ALTER TABLE ADD COLUMN cannot apply. Specify \
             `serial` at create-table time via `with pk` instead."
                .to_string(),
        ));
    }
    let mut ddl = String::new();
    write!(
        ddl,
        "ALTER TABLE {tbl} ADD COLUMN {col} {sqlite_type}{extra};",
        tbl = quote_ident(table),
        col = quote_ident(column),
        sqlite_type = ty.sqlite_strict_type(),
        extra = ty.sqlite_strict_extra(),
    )
    .expect("write to String never fails");
    debug!(ddl = %ddl, "add_column");
    let tx = conn
        .unchecked_transaction()
        .map_err(DbError::from_rusqlite)?;
    tx.execute_batch(&ddl).map_err(DbError::from_rusqlite)?;
    tx.execute(
        &format!(
            "INSERT INTO {META_TABLE} (table_name, column_name, user_type) \
             VALUES (?1, ?2, ?3);"
        ),
        [table, column, ty.keyword()],
    )
    .map_err(DbError::from_rusqlite)?;
    let description = do_describe_table(conn, table)?;
    let changes = Changes {
        schema_dirty: true,
        rewritten_tables: vec![table.to_string()],
        ..Changes::default()
    };
    finalize_persistence(conn, persistence, source, &changes)?;
    tx.commit().map_err(DbError::from_rusqlite)?;
    Ok(description)
}

fn do_list_tables(conn: &Connection) -> Result<Vec<String>, DbError> {
    let mut stmt = conn
        .prepare(
            "SELECT name FROM sqlite_schema \
             WHERE type = 'table' \
               AND name NOT LIKE 'sqlite_%' \
               AND substr(name, 1, 8) != '__rdbms_' \
             ORDER BY name;",
        )
        .map_err(DbError::from_rusqlite)?;
    let rows = stmt
        .query_map([], |row| row.get::<_, String>(0))
        .map_err(DbError::from_rusqlite)?;
    let mut out = Vec::new();
    for row in rows {
        out.push(row.map_err(DbError::from_rusqlite)?);
    }
    Ok(out)
}

/// Internal full schema of a table, sufficient to regenerate
/// its `CREATE TABLE` statement during the rebuild dance.
#[derive(Debug, Clone)]
struct ReadSchema {
    columns: Vec<ReadColumn>,
    primary_key: Vec<String>,
    foreign_keys: Vec<ReadForeignKey>,
}

#[derive(Debug, Clone)]
struct ReadColumn {
    name: String,
    sqlite_type: String,
    notnull: bool,
    primary_key: bool,
    user_type: Option<Type>,
}

#[derive(Debug, Clone)]
struct ReadForeignKey {
    parent_table: String,
    parent_column: String,
    child_column: String,
    on_delete: ReferentialAction,
    on_update: ReferentialAction,
}

fn read_schema(conn: &Connection, table: &str) -> Result<ReadSchema, DbError> {
    // Columns + PK from pragma_table_info, joined with our user-type metadata.
    let mut col_stmt = conn
        .prepare(&format!(
            "SELECT pti.name, pti.type, pti.\"notnull\", pti.pk, m.user_type \
             FROM pragma_table_info(?1) AS pti \
             LEFT JOIN {META_TABLE} AS m \
               ON m.table_name = ?1 AND m.column_name = pti.name \
             ORDER BY pti.cid;"
        ))
        .map_err(DbError::from_rusqlite)?;
    let rows = col_stmt
        .query_map([table], |row| {
            let user_type_kw: Option<String> = row.get(4)?;
            let user_type = user_type_kw.and_then(|kw| kw.parse::<Type>().ok());
            Ok(ReadColumn {
                name: row.get(0)?,
                sqlite_type: row.get(1)?,
                notnull: row.get::<_, i64>(2)? != 0,
                primary_key: row.get::<_, i64>(3)? != 0,
                user_type,
            })
        })
        .map_err(DbError::from_rusqlite)?;
    let mut columns = Vec::new();
    for row in rows {
        columns.push(row.map_err(DbError::from_rusqlite)?);
    }
    if columns.is_empty() {
        return Err(DbError::Sqlite {
            message: format!("no such table: {table}"),
            kind: SqliteErrorKind::NoSuchTable,
        });
    }
    let primary_key: Vec<String> = columns
        .iter()
        .filter(|c| c.primary_key)
        .map(|c| c.name.clone())
        .collect();

    // Foreign keys from pragma_foreign_key_list.
    let mut fk_stmt = conn
        .prepare(
            "SELECT \"table\", \"from\", \"to\", on_delete, on_update \
             FROM pragma_foreign_key_list(?1) \
             ORDER BY id, seq;",
        )
        .map_err(DbError::from_rusqlite)?;
    let fk_rows = fk_stmt
        .query_map([table], |row| {
            let on_delete_str: String = row.get(3)?;
            let on_update_str: String = row.get(4)?;
            Ok(ReadForeignKey {
                parent_table: row.get(0)?,
                child_column: row.get(1)?,
                parent_column: row.get(2)?,
                on_delete: parse_action_from_sqlite(&on_delete_str),
                on_update: parse_action_from_sqlite(&on_update_str),
            })
        })
        .map_err(DbError::from_rusqlite)?;
    let mut foreign_keys = Vec::new();
    for row in fk_rows {
        foreign_keys.push(row.map_err(DbError::from_rusqlite)?);
    }

    Ok(ReadSchema {
        columns,
        primary_key,
        foreign_keys,
    })
}

fn parse_action_from_sqlite(s: &str) -> ReferentialAction {
    // SQLite stores the action keywords in upper-case form
    // ("CASCADE", "SET NULL", "NO ACTION", "RESTRICT").
    s.parse::<ReferentialAction>()
        .unwrap_or(ReferentialAction::NoAction)
}

/// Generate the CREATE TABLE DDL from a `ReadSchema`. Used during
/// the rebuild dance.
fn schema_to_ddl(table: &str, schema: &ReadSchema) -> String {
    let mut clauses: Vec<String> = Vec::new();

    // The single-column-INTEGER-PK case must be inline (`PRIMARY
    // KEY` on the column itself) so SQLite gives it rowid-alias
    // semantics. Compound or non-INTEGER PKs go to a table-level
    // constraint.
    let single_inline_pk = schema.primary_key.len() == 1
        && !schema.columns.is_empty()
        && schema.columns[0].primary_key
        && schema.primary_key[0] == schema.columns[0].name;

    for col in &schema.columns {
        let mut clause = format!(
            "{ident} {sqlite_type}",
            ident = quote_ident(&col.name),
            sqlite_type = col.sqlite_type,
        );
        if col.notnull {
            clause.push_str(" NOT NULL");
        }
        if single_inline_pk && col.primary_key {
            clause.push_str(" PRIMARY KEY");
        }
        clauses.push(clause);
    }

    if !single_inline_pk && !schema.primary_key.is_empty() {
        let pk_idents: Vec<String> = schema.primary_key.iter().map(|n| quote_ident(n)).collect();
        clauses.push(format!("PRIMARY KEY ({})", pk_idents.join(", ")));
    }

    for fk in &schema.foreign_keys {
        clauses.push(format!(
            "FOREIGN KEY ({child}) REFERENCES {parent_table}({parent_col}) \
             ON DELETE {od} ON UPDATE {ou}",
            child = quote_ident(&fk.child_column),
            parent_table = quote_ident(&fk.parent_table),
            parent_col = quote_ident(&fk.parent_column),
            od = fk.on_delete.sql_clause(),
            ou = fk.on_update.sql_clause(),
        ));
    }

    format!(
        "CREATE TABLE {ident} ({clauses}) STRICT;",
        ident = quote_ident(table),
        clauses = clauses.join(", "),
    )
}

/// The rebuild-table dance. Replaces `table` with a fresh table
/// constructed from `new_schema`, copying data column-by-name.
/// The caller is responsible for any metadata-table updates that
/// need to happen alongside the rebuild — those are passed in as
/// a closure and run inside the same transaction.
///
/// Per the official SQLite ALTER-via-rebuild recipe: foreign-key
/// enforcement must be temporarily disabled at session level,
/// not just deferred, because the connection-level pragma is a
/// no-op inside transactions.
fn rebuild_table<F>(
    conn: &Connection,
    table: &str,
    old_schema: &ReadSchema,
    new_schema: &ReadSchema,
    metadata_updates: F,
) -> Result<(), DbError>
where
    F: FnOnce(&rusqlite::Transaction<'_>) -> Result<(), DbError>,
{
    // foreign_keys=OFF must be set *outside* a transaction.
    conn.execute_batch("PRAGMA foreign_keys = OFF;")
        .map_err(DbError::from_rusqlite)?;

    let result = (|| -> Result<(), DbError> {
        let tx = conn
            .unchecked_transaction()
            .map_err(DbError::from_rusqlite)?;

        let temp_name = format!("__rdbms_rebuild_{table}");
        // Defensive: drop any leftover rebuild table from a
        // previous failed attempt.
        tx.execute_batch(&format!(
            "DROP TABLE IF EXISTS {ident};",
            ident = quote_ident(&temp_name)
        ))
        .map_err(DbError::from_rusqlite)?;

        let create_temp = schema_to_ddl(&temp_name, new_schema);
        tx.execute_batch(&create_temp)
            .map_err(DbError::from_rusqlite)?;

        // Copy data: only for columns that exist on both sides.
        // Auto-created columns (e.g. via --create-fk) are absent
        // from the old table, so they remain NULL in the new one.
        let copy_cols: Vec<&str> = new_schema
            .columns
            .iter()
            .filter(|c| old_schema.columns.iter().any(|oc| oc.name == c.name))
            .map(|c| c.name.as_str())
            .collect();
        if !copy_cols.is_empty() {
            let cols_csv = copy_cols
                .iter()
                .map(|c| quote_ident(c))
                .collect::<Vec<_>>()
                .join(", ");
            let copy_sql = format!(
                "INSERT INTO {temp} ({cols}) SELECT {cols} FROM {orig};",
                temp = quote_ident(&temp_name),
                cols = cols_csv,
                orig = quote_ident(table),
            );
            tx.execute_batch(&copy_sql)
                .map_err(DbError::from_rusqlite)?;
        }

        tx.execute_batch(&format!(
            "DROP TABLE {ident};",
            ident = quote_ident(table)
        ))
        .map_err(DbError::from_rusqlite)?;
        tx.execute_batch(&format!(
            "ALTER TABLE {temp} RENAME TO {final_name};",
            temp = quote_ident(&temp_name),
            final_name = quote_ident(table),
        ))
        .map_err(DbError::from_rusqlite)?;

        metadata_updates(&tx)?;

        // Verify referential integrity before committing. Any
        // returned rows mean a FK violation persisted.
        let mut check = tx
            .prepare("PRAGMA foreign_key_check;")
            .map_err(DbError::from_rusqlite)?;
        let mut rows = check.query([]).map_err(DbError::from_rusqlite)?;
        if let Some(_row) = rows.next().map_err(DbError::from_rusqlite)? {
            return Err(DbError::Sqlite {
                message: format!(
                    "foreign-key check failed after rebuild of `{table}`; \
                     existing data violates the new constraint"
                ),
                kind: SqliteErrorKind::Other,
            });
        }
        drop(rows);
        drop(check);

        tx.commit().map_err(DbError::from_rusqlite)?;
        Ok(())
    })();

    // Always re-enable foreign_keys, even on error.
    let pragma_result = conn
        .execute_batch("PRAGMA foreign_keys = ON;")
        .map_err(DbError::from_rusqlite);

    result.and(pragma_result)
}

#[allow(clippy::too_many_arguments)]
fn do_add_relationship(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    name: Option<&str>,
    parent_table: &str,
    parent_column: &str,
    child_table: &str,
    child_column: &str,
    on_delete: ReferentialAction,
    on_update: ReferentialAction,
    create_fk: bool,
) -> Result<TableDescription, DbError> {
    // 1. Read parent schema; verify the referenced column is a PK.
    let parent_schema = read_schema(conn, parent_table)?;
    let parent_col = parent_schema
        .columns
        .iter()
        .find(|c| c.name == parent_column)
        .ok_or_else(|| DbError::Sqlite {
            message: format!("no such column: {parent_table}.{parent_column}"),
            kind: SqliteErrorKind::NoSuchColumn,
        })?;
    if !parent_col.primary_key {
        return Err(DbError::Unsupported(format!(
            "column `{parent_table}.{parent_column}` is not a primary key. \
             Foreign keys must reference a primary key (UNIQUE-target FKs \
             land in a later iteration)."
        )));
    }

    // 2. Read child schema; verify the FK column or auto-create.
    let mut child_schema = read_schema(conn, child_table)?;
    let needs_create_column = child_schema
        .columns
        .iter()
        .all(|c| c.name != child_column);
    if needs_create_column && !create_fk {
        return Err(DbError::Unsupported(format!(
            "column `{child_table}.{child_column}` does not exist. \
             Add it first, or use `--create-fk` to create it automatically."
        )));
    }

    // 3. Determine child column type. Either the existing one's
    //    user_type, or the parent's fk_target_type for auto-create.
    let parent_user_type = parent_col.user_type.ok_or_else(|| DbError::Unsupported(
        "parent column has no user type metadata".to_string(),
    ))?;
    let expected_child_type = parent_user_type.fk_target_type();

    if needs_create_column {
        // Synthesise the column row for the new schema.
        child_schema.columns.push(ReadColumn {
            name: child_column.to_string(),
            sqlite_type: expected_child_type.sqlite_strict_type().to_string(),
            notnull: false,
            primary_key: false,
            user_type: Some(expected_child_type),
        });
    } else {
        // Validate type compatibility against the existing column.
        let child_col = child_schema
            .columns
            .iter()
            .find(|c| c.name == child_column)
            .expect("checked above");
        let actual = child_col.user_type.ok_or_else(|| DbError::Unsupported(
            "child column has no user type metadata".to_string(),
        ))?;
        if actual != expected_child_type {
            return Err(DbError::Unsupported(format!(
                "type mismatch: `{child_table}.{child_column}` is `{actual}` but \
                 a foreign key referencing `{parent_table}.{parent_column}` \
                 (`{parent_user_type}`) requires `{expected_child_type}`. \
                 Either change the column type, or pick a different FK column."
            )));
        }
    }

    // 4. Determine relationship name (auto-gen or supplied) and
    //    check uniqueness against the metadata table.
    let resolved_name = name.map_or_else(
        // Auto-name follows the user-typed `from <Parent>.<col>
        // to <Child>.<col>` direction so the name reads as the
        // grammar reads — see ADR-0013.
        || format!("{parent_table}_{parent_column}_to_{child_table}_{child_column}"),
        ToString::to_string,
    );
    let collision: i64 = conn
        .query_row(
            &format!("SELECT COUNT(*) FROM {REL_TABLE} WHERE name = ?1;"),
            [&resolved_name],
            |row| row.get(0),
        )
        .map_err(DbError::from_rusqlite)?;
    if collision > 0 {
        return Err(DbError::Unsupported(format!(
            "a relationship named `{resolved_name}` already exists. \
             Pick a different name or drop the existing one first."
        )));
    }

    // 5. Build the new schema with the FK appended.
    let mut new_schema = child_schema.clone();
    new_schema.foreign_keys.push(ReadForeignKey {
        parent_table: parent_table.to_string(),
        parent_column: parent_column.to_string(),
        child_column: child_column.to_string(),
        on_delete,
        on_update,
    });

    // 6. Rebuild, with metadata updates inside the transaction.
    let on_delete_kw = on_delete.keyword();
    let on_update_kw = on_update.keyword();
    let column_user_type_kw = expected_child_type.keyword();
    let resolved_name_for_meta = resolved_name.as_str();
    rebuild_table(conn, child_table, &child_schema, &new_schema, |tx| {
        if needs_create_column {
            tx.execute(
                &format!(
                    "INSERT INTO {META_TABLE} (table_name, column_name, user_type) \
                     VALUES (?1, ?2, ?3);"
                ),
                [child_table, child_column, column_user_type_kw],
            )
            .map_err(DbError::from_rusqlite)?;
        }
        tx.execute(
            &format!(
                "INSERT INTO {REL_TABLE} \
                   (name, parent_table, parent_column, child_table, child_column, on_delete, on_update) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7);"
            ),
            [
                resolved_name_for_meta,
                parent_table,
                parent_column,
                child_table,
                child_column,
                on_delete_kw,
                on_update_kw,
            ],
        )
        .map_err(DbError::from_rusqlite)?;
        // Persistence runs inside the same tx so a write
        // failure rolls back both the schema and the metadata
        // (commit-db-last per ADR-0015 §6).
        let changes = Changes {
            schema_dirty: true,
            // The child table was rebuilt — its CSV needs to
            // be re-emitted from the new state too, in case
            // the FK column was newly created.
            rewritten_tables: vec![child_table.to_string()],
            ..Changes::default()
        };
        finalize_persistence(tx, persistence, source, &changes)?;
        Ok(())
    })?;

    // Return the parent (1-side) description so the user sees
    // the new relationship via the inbound section — that's the
    // perspective that matches the `from <Parent>.col to ...`
    // direction of the command.
    do_describe_table(conn, parent_table)
}

fn do_drop_relationship(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    selector: &RelationshipSelector,
) -> Result<Option<TableDescription>, DbError> {
    // Resolve to a single relationship row.
    let resolved: Option<(String, String, String, String, String)> = match selector {
        RelationshipSelector::Named { name } => conn
            .query_row(
                &format!(
                    "SELECT name, parent_table, parent_column, child_table, child_column \
                     FROM {REL_TABLE} WHERE name = ?1;"
                ),
                [name],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)),
            )
            .ok(),
        RelationshipSelector::Endpoints {
            parent_table,
            parent_column,
            child_table,
            child_column,
        } => conn
            .query_row(
                &format!(
                    "SELECT name, parent_table, parent_column, child_table, child_column \
                     FROM {REL_TABLE} \
                     WHERE parent_table = ?1 AND parent_column = ?2 \
                       AND child_table = ?3 AND child_column = ?4;"
                ),
                [parent_table, parent_column, child_table, child_column],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)),
            )
            .ok(),
    };
    let (rel_name, parent_table, _parent_column, child_table, child_column) =
        resolved.ok_or_else(|| DbError::Sqlite {
            message: format!("no such relationship: {selector}"),
            kind: SqliteErrorKind::Other,
        })?;

    // Read child schema; build new schema without the FK.
    let old_schema = read_schema(conn, &child_table)?;
    let mut new_schema = old_schema.clone();
    new_schema.foreign_keys.retain(|fk| fk.child_column != child_column);

    let child_table_for_persist = child_table.clone();
    rebuild_table(conn, &child_table, &old_schema, &new_schema, |tx| {
        tx.execute(
            &format!("DELETE FROM {REL_TABLE} WHERE name = ?1;"),
            [rel_name.as_str()],
        )
        .map_err(DbError::from_rusqlite)?;
        let changes = Changes {
            schema_dirty: true,
            rewritten_tables: vec![child_table_for_persist.clone()],
            ..Changes::default()
        };
        finalize_persistence(tx, persistence, source, &changes)?;
        Ok(())
    })?;

    // Show the parent (1-side) afterwards — same direction as
    // the command's `from <Parent> to ...` reading.
    Ok(Some(do_describe_table(conn, &parent_table)?))
}

/// Read-only wrapper around `do_describe_table` that runs an
/// auxiliary `history.log` append for user-issued
/// `show table` commands.
fn do_describe_table_request(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    name: &str,
) -> Result<TableDescription, DbError> {
    let description = do_describe_table(conn, name)?;
    if let (Some(p), Some(text)) = (persistence, source) {
        p.append_history(text)
            .map_err(DbError::from_persistence)?;
    }
    Ok(description)
}

fn do_describe_table(conn: &Connection, name: &str) -> Result<TableDescription, DbError> {
    // `pragma_table_info` is a table-valued function in modern
    // SQLite; using it as a SELECT lets us bind the table name
    // via ? rather than splicing it into a PRAGMA statement.
    // We LEFT JOIN our metadata table to recover the user-facing
    // type each column was declared as.
    let mut stmt = conn
        .prepare(&format!(
            "SELECT pti.name, pti.type, pti.\"notnull\", pti.pk, m.user_type \
             FROM pragma_table_info(?1) AS pti \
             LEFT JOIN {META_TABLE} AS m \
               ON m.table_name = ?1 AND m.column_name = pti.name \
             ORDER BY pti.cid;"
        ))
        .map_err(DbError::from_rusqlite)?;
    let rows = stmt
        .query_map([name], |row| {
            let user_type_kw: Option<String> = row.get(4)?;
            let user_type = user_type_kw.and_then(|kw| kw.parse::<Type>().ok());
            Ok(ColumnDescription {
                name: row.get(0)?,
                user_type,
                sqlite_type: row.get(1)?,
                notnull: row.get::<_, i64>(2)? != 0,
                primary_key: row.get::<_, i64>(3)? != 0,
            })
        })
        .map_err(DbError::from_rusqlite)?;
    let mut columns = Vec::new();
    for row in rows {
        columns.push(row.map_err(DbError::from_rusqlite)?);
    }
    if columns.is_empty() {
        // pragma_table_info returns no rows for a non-existent
        // table, which we surface as a NoSuchTable error so
        // describe_table is not silently empty.
        warn!(name, "describe_table: no columns (table missing?)");
        return Err(DbError::Sqlite {
            message: format!("no such table: {name}"),
            kind: SqliteErrorKind::NoSuchTable,
        });
    }

    let outbound_relationships = read_relationships_outbound(conn, name)?;
    let inbound_relationships = read_relationships_inbound(conn, name)?;

    Ok(TableDescription {
        name: name.to_string(),
        columns,
        outbound_relationships,
        inbound_relationships,
    })
}

// --- Data operations (C5) -----------------------------------

fn impl_value_for(
    schema: &ReadSchema,
    column: &str,
    value: &Value,
) -> Result<Bound, DbError> {
    let col = schema
        .columns
        .iter()
        .find(|c| c.name == column)
        .ok_or_else(|| DbError::Sqlite {
            message: format!("no such column: {column}"),
            kind: SqliteErrorKind::NoSuchColumn,
        })?;
    let ty = col.user_type.ok_or_else(|| {
        DbError::Unsupported(format!(
            "column `{column}` has no user-type metadata; cannot validate value"
        ))
    })?;
    value
        .bind_for_column(column, ty)
        .map_err(|e: ValueError| DbError::InvalidValue(e.to_string()))
}

fn bound_to_sqlite_value(b: &Bound) -> rusqlite::types::Value {
    use rusqlite::types::Value as V;
    match b {
        Bound::Integer(i) => V::Integer(*i),
        Bound::Real(r) => V::Real(*r),
        Bound::Text(s) => V::Text(s.clone()),
        Bound::Null => V::Null,
    }
}

fn enrich_fk_message(conn: &Connection, table: &str, base: String) -> String {
    // SQLite tells us "FOREIGN KEY constraint failed" without
    // saying which constraint. We list both:
    //   - outbound FKs (this table is child) — relevant for
    //     INSERT and UPDATE that try to set a non-matching FK
    //     value;
    //   - inbound FKs (this table is parent) — relevant for
    //     DELETE / UPDATE on the parent side that violate a
    //     RESTRICT or NO ACTION constraint.
    // The user reads both lists and recognises the relevant
    // direction from the operation they ran. Full H1 would
    // pinpoint the offending row.
    let outbound = read_relationships_outbound(conn, table).unwrap_or_default();
    let inbound = read_relationships_inbound(conn, table).unwrap_or_default();
    if outbound.is_empty() && inbound.is_empty() {
        return base;
    }
    let mut msg = base;
    if !outbound.is_empty() {
        msg.push_str(". Foreign keys on this table (relevant for INSERT/UPDATE):");
        for r in outbound {
            msg.push_str(&format!(
                "\n  - {}.{} → {}.{}",
                table, r.local_column, r.other_table, r.other_column
            ));
        }
    }
    if !inbound.is_empty() {
        msg.push_str(
            "\nThis table is referenced by (relevant for DELETE/UPDATE on parent side):",
        );
        for r in inbound {
            msg.push_str(&format!(
                "\n  - {}.{} → {}.{}  (on delete {})",
                r.other_table, r.other_column, table, r.local_column, r.on_delete
            ));
        }
    }
    msg.push_str(
        "\nCheck that referenced values exist (for INSERT/UPDATE) or that no children \
         reference these rows (for DELETE/UPDATE on the parent side).",
    );
    msg
}

fn execute_with_fk_enrichment(
    conn: &Connection,
    table: &str,
    sql: &str,
    params: &[rusqlite::types::Value],
) -> Result<usize, DbError> {
    let result = conn.execute(sql, rusqlite::params_from_iter(params.iter()));
    match result {
        Ok(n) => Ok(n),
        Err(e) => {
            let mut db_err = DbError::from_rusqlite(e);
            if let DbError::Sqlite { message, kind } = &db_err {
                let lower = message.to_ascii_lowercase();
                if lower.contains("foreign key") {
                    db_err = DbError::Sqlite {
                        message: enrich_fk_message(conn, table, message.clone()),
                        kind: *kind,
                    };
                }
            }
            Err(db_err)
        }
    }
}

/// Fetch a small `DataResult` containing only the rows whose
/// rowids appear in `rowids`. Used so writes can show only the
/// rows they touched rather than the whole table.
fn query_rows_by_rowid(
    conn: &Connection,
    table: &str,
    rowids: &[i64],
) -> Result<DataResult, DbError> {
    let schema = read_schema(conn, table)?;
    let column_names: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
    let column_types: Vec<Option<Type>> =
        schema.columns.iter().map(|c| c.user_type).collect();

    if rowids.is_empty() {
        return Ok(DataResult {
            table_name: table.to_string(),
            columns: column_names,
            rows: Vec::new(),
        });
    }

    let cols_csv = column_names
        .iter()
        .map(|c| quote_ident(c))
        .collect::<Vec<_>>()
        .join(", ");
    let placeholders = (1..=rowids.len())
        .map(|i| format!("?{i}"))
        .collect::<Vec<_>>()
        .join(", ");
    let sql = format!(
        "SELECT {cols} FROM {ident} WHERE rowid IN ({placeholders});",
        cols = cols_csv,
        ident = quote_ident(table),
    );
    let mut stmt = conn.prepare(&sql).map_err(DbError::from_rusqlite)?;
    let params: Vec<rusqlite::types::Value> = rowids
        .iter()
        .map(|id| rusqlite::types::Value::Integer(*id))
        .collect();
    let rows_iter = stmt
        .query_map(rusqlite::params_from_iter(params.iter()), |row| {
            let mut cells: Vec<rusqlite::types::Value> =
                Vec::with_capacity(column_names.len());
            for i in 0..column_names.len() {
                cells.push(row.get(i)?);
            }
            Ok(cells)
        })
        .map_err(DbError::from_rusqlite)?;
    let mut rows: Vec<Vec<Option<String>>> = Vec::new();
    for r in rows_iter {
        let cells = r.map_err(DbError::from_rusqlite)?;
        let formatted: Vec<Option<String>> = cells
            .into_iter()
            .zip(column_types.iter())
            .map(|(v, ty)| format_cell(v, *ty))
            .collect();
        rows.push(formatted);
    }
    Ok(DataResult {
        table_name: table.to_string(),
        columns: column_names,
        rows,
    })
}

fn count_rows(conn: &Connection, table: &str) -> Result<i64, DbError> {
    let sql = format!("SELECT COUNT(*) FROM {ident};", ident = quote_ident(table));
    conn.query_row(&sql, [], |row| row.get::<_, i64>(0))
        .map_err(DbError::from_rusqlite)
}

fn do_insert(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    table: &str,
    user_columns: Option<&[String]>,
    user_values: &[Value],
) -> Result<InsertResult, DbError> {
    let schema = read_schema(conn, table)?;

    // Resolve which columns the user is providing values for.
    let user_cols: Vec<String> = match user_columns {
        Some(cols) => cols.to_vec(),
        None => {
            // Short form: every non-auto-generated column in
            // schema declaration order. Serial and shortid both
            // get auto-filled below.
            schema
                .columns
                .iter()
                .filter(|c| !matches!(c.user_type, Some(Type::Serial) | Some(Type::ShortId)))
                .map(|c| c.name.clone())
                .collect()
        }
    };

    if user_cols.len() != user_values.len() {
        return Err(DbError::InvalidValue(format!(
            "expected {} value(s), got {}",
            user_cols.len(),
            user_values.len()
        )));
    }

    let mut bindings: Vec<(String, Bound)> = Vec::with_capacity(user_cols.len());
    for (col_name, value) in user_cols.iter().zip(user_values.iter()) {
        let bound = impl_value_for(&schema, col_name, value)?;
        bindings.push((col_name.clone(), bound));
    }

    // Auto-fill any shortid columns the user didn't list.
    let provided: std::collections::HashSet<String> =
        bindings.iter().map(|(c, _)| c.clone()).collect();
    for c in &schema.columns {
        if c.user_type == Some(Type::ShortId) && !provided.contains(&c.name) {
            bindings.push((c.name.clone(), Bound::Text(shortid::generate())));
        }
    }

    if bindings.is_empty() {
        return Err(DbError::InvalidValue(
            "INSERT requires at least one column value".to_string(),
        ));
    }

    let cols_csv = bindings
        .iter()
        .map(|(c, _)| quote_ident(c))
        .collect::<Vec<_>>()
        .join(", ");
    let placeholders = (1..=bindings.len())
        .map(|i| format!("?{i}"))
        .collect::<Vec<_>>()
        .join(", ");
    let sql = format!(
        "INSERT INTO {ident} ({cols_csv}) VALUES ({placeholders});",
        ident = quote_ident(table),
    );
    debug!(sql = %sql, "insert");
    let params: Vec<rusqlite::types::Value> =
        bindings.iter().map(|(_, b)| bound_to_sqlite_value(b)).collect();
    let tx = conn
        .unchecked_transaction()
        .map_err(DbError::from_rusqlite)?;
    let rows_affected = execute_with_fk_enrichment(conn, table, &sql, &params)?;
    let new_rowid = conn.last_insert_rowid();
    let data = query_rows_by_rowid(conn, table, &[new_rowid])?;
    let changes = Changes {
        schema_dirty: false,
        rewritten_tables: vec![table.to_string()],
        ..Changes::default()
    };
    finalize_persistence(conn, persistence, source, &changes)?;
    tx.commit().map_err(DbError::from_rusqlite)?;
    Ok(InsertResult {
        rows_affected,
        data,
    })
}

fn do_update(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    table: &str,
    assignments: &[(String, Value)],
    filter: &RowFilter,
) -> Result<UpdateResult, DbError> {
    if assignments.is_empty() {
        return Err(DbError::InvalidValue(
            "UPDATE requires at least one assignment".to_string(),
        ));
    }
    let schema = read_schema(conn, table)?;

    // Capture rowids of matching rows up front so we can fetch
    // the updated rows even if the UPDATE changed the WHERE column.
    let rowids = match filter {
        RowFilter::AllRows => select_all_rowids(conn, table)?,
        RowFilter::Where { column, value } => {
            let bound = impl_value_for(&schema, column, value)?;
            let mut stmt = conn
                .prepare(&format!(
                    "SELECT rowid FROM {ident} WHERE {col} = ?1;",
                    ident = quote_ident(table),
                    col = quote_ident(column),
                ))
                .map_err(DbError::from_rusqlite)?;
            let bound_param = bound_to_sqlite_value(&bound);
            let rows = stmt
                .query_map([&bound_param], |row| row.get::<_, i64>(0))
                .map_err(DbError::from_rusqlite)?;
            let mut ids = Vec::new();
            for r in rows {
                ids.push(r.map_err(DbError::from_rusqlite)?);
            }
            ids
        }
    };

    let mut params: Vec<rusqlite::types::Value> = Vec::new();
    let mut set_clauses: Vec<String> = Vec::with_capacity(assignments.len());
    for (col, value) in assignments {
        let bound = impl_value_for(&schema, col, value)?;
        set_clauses.push(format!(
            "{col_id} = ?{n}",
            col_id = quote_ident(col),
            n = params.len() + 1
        ));
        params.push(bound_to_sqlite_value(&bound));
    }

    let where_sql = match filter {
        RowFilter::AllRows => String::new(),
        RowFilter::Where { column, value } => {
            let bound = impl_value_for(&schema, column, value)?;
            params.push(bound_to_sqlite_value(&bound));
            format!(
                " WHERE {col} = ?{n}",
                col = quote_ident(column),
                n = params.len()
            )
        }
    };

    let sql = format!(
        "UPDATE {ident} SET {sets}{where_sql};",
        ident = quote_ident(table),
        sets = set_clauses.join(", "),
    );
    debug!(sql = %sql, "update");
    let tx = conn
        .unchecked_transaction()
        .map_err(DbError::from_rusqlite)?;
    let rows_affected = execute_with_fk_enrichment(conn, table, &sql, &params)?;
    let data = query_rows_by_rowid(conn, table, &rowids)?;
    let changes = Changes {
        schema_dirty: false,
        rewritten_tables: vec![table.to_string()],
        ..Changes::default()
    };
    finalize_persistence(conn, persistence, source, &changes)?;
    tx.commit().map_err(DbError::from_rusqlite)?;
    Ok(UpdateResult {
        rows_affected,
        data,
    })
}

fn select_all_rowids(conn: &Connection, table: &str) -> Result<Vec<i64>, DbError> {
    let mut stmt = conn
        .prepare(&format!(
            "SELECT rowid FROM {ident};",
            ident = quote_ident(table)
        ))
        .map_err(DbError::from_rusqlite)?;
    let rows = stmt
        .query_map([], |row| row.get::<_, i64>(0))
        .map_err(DbError::from_rusqlite)?;
    let mut ids = Vec::new();
    for r in rows {
        ids.push(r.map_err(DbError::from_rusqlite)?);
    }
    Ok(ids)
}

fn do_delete(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    table: &str,
    filter: &RowFilter,
) -> Result<DeleteResult, DbError> {
    let schema = read_schema(conn, table)?;

    // Snapshot child-table row counts before the delete so we
    // can detect cascade effects via diffing afterwards. ON
    // UPDATE CASCADE does not change row counts and so is not
    // detected here — it would need value-level diffing, which
    // a future iteration can add.
    let inbound = read_relationships_inbound(conn, table)?;
    let mut before_counts: Vec<(String, i64)> = Vec::with_capacity(inbound.len());
    for r in &inbound {
        before_counts.push((r.other_table.clone(), count_rows(conn, &r.other_table)?));
    }

    let mut params: Vec<rusqlite::types::Value> = Vec::new();
    let where_sql = match filter {
        RowFilter::AllRows => String::new(),
        RowFilter::Where { column, value } => {
            let bound = impl_value_for(&schema, column, value)?;
            params.push(bound_to_sqlite_value(&bound));
            format!(
                " WHERE {col} = ?{n}",
                col = quote_ident(column),
                n = params.len()
            )
        }
    };
    let sql = format!(
        "DELETE FROM {ident}{where_sql};",
        ident = quote_ident(table),
    );
    debug!(sql = %sql, "delete");
    let tx = conn
        .unchecked_transaction()
        .map_err(DbError::from_rusqlite)?;
    let rows_affected = execute_with_fk_enrichment(conn, table, &sql, &params)?;

    // Compare child-table counts after the delete; non-zero
    // diffs are cascade effects. We also collect cascaded
    // tables so the persistence phase rewrites their CSVs too.
    let mut cascade: Vec<CascadeEffect> = Vec::new();
    let mut rewritten_tables: Vec<String> = vec![table.to_string()];
    for (rel, (_child_table, before_count)) in inbound.iter().zip(before_counts.iter()) {
        let after_count = count_rows(conn, &rel.other_table)?;
        let diff = before_count - after_count;
        if diff > 0 {
            cascade.push(CascadeEffect {
                relationship_name: rel.name.clone(),
                child_table: rel.other_table.clone(),
                rows_changed: diff,
                action: rel.on_delete,
            });
            rewritten_tables.push(rel.other_table.clone());
        }
    }

    let changes = Changes {
        schema_dirty: false,
        rewritten_tables,
        ..Changes::default()
    };
    finalize_persistence(conn, persistence, source, &changes)?;
    tx.commit().map_err(DbError::from_rusqlite)?;

    Ok(DeleteResult {
        rows_affected,
        cascade,
    })
}

/// Read-only wrapper that adds the `history.log` append for
/// `show data` user commands.
fn do_query_data_request(
    conn: &Connection,
    persistence: Option<&Persistence>,
    source: Option<&str>,
    table: &str,
) -> Result<DataResult, DbError> {
    let data = do_query_data(conn, table)?;
    if let (Some(p), Some(text)) = (persistence, source) {
        p.append_history(text)
            .map_err(DbError::from_persistence)?;
    }
    Ok(data)
}

fn do_query_data(conn: &Connection, table: &str) -> Result<DataResult, DbError> {
    let schema = read_schema(conn, table)?;
    let column_names: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
    let column_types: Vec<Option<Type>> =
        schema.columns.iter().map(|c| c.user_type).collect();

    let cols_csv = column_names
        .iter()
        .map(|c| quote_ident(c))
        .collect::<Vec<_>>()
        .join(", ");
    let sql = format!(
        "SELECT {cols} FROM {ident};",
        cols = cols_csv,
        ident = quote_ident(table),
    );
    debug!(sql = %sql, "query_data");
    let mut stmt = conn.prepare(&sql).map_err(DbError::from_rusqlite)?;
    let rows_iter = stmt
        .query_map([], |row| {
            let mut cells: Vec<rusqlite::types::Value> = Vec::with_capacity(column_names.len());
            for i in 0..column_names.len() {
                let v: rusqlite::types::Value = row.get(i)?;
                cells.push(v);
            }
            Ok(cells)
        })
        .map_err(DbError::from_rusqlite)?;
    let mut rows: Vec<Vec<Option<String>>> = Vec::new();
    for r in rows_iter {
        let cells = r.map_err(DbError::from_rusqlite)?;
        let formatted: Vec<Option<String>> = cells
            .into_iter()
            .zip(column_types.iter())
            .map(|(v, ty)| format_cell(v, *ty))
            .collect();
        rows.push(formatted);
    }
    Ok(DataResult {
        table_name: table.to_string(),
        columns: column_names,
        rows,
    })
}

fn format_cell(value: rusqlite::types::Value, ty: Option<Type>) -> Option<String> {
    use rusqlite::types::Value as V;
    match value {
        V::Null => None,
        V::Integer(i) => Some(if matches!(ty, Some(Type::Bool)) {
            (if i == 0 { "false" } else { "true" }).to_string()
        } else {
            i.to_string()
        }),
        V::Real(r) => Some(format!("{r}")),
        V::Text(s) => Some(s),
        V::Blob(b) => Some(format!("<blob {} bytes>", b.len())),
    }
}

fn read_relationships_outbound(
    conn: &Connection,
    table: &str,
) -> Result<Vec<RelationshipEnd>, DbError> {
    let mut stmt = conn
        .prepare(&format!(
            "SELECT name, parent_table, parent_column, child_column, on_delete, on_update \
             FROM {REL_TABLE} \
             WHERE child_table = ?1 \
             ORDER BY name;"
        ))
        .map_err(DbError::from_rusqlite)?;
    let rows = stmt
        .query_map([table], |row| {
            let on_delete: String = row.get(4)?;
            let on_update: String = row.get(5)?;
            Ok(RelationshipEnd {
                name: row.get(0)?,
                other_table: row.get(1)?,
                other_column: row.get(2)?,
                local_column: row.get(3)?,
                on_delete: on_delete
                    .parse::<ReferentialAction>()
                    .unwrap_or(ReferentialAction::NoAction),
                on_update: on_update
                    .parse::<ReferentialAction>()
                    .unwrap_or(ReferentialAction::NoAction),
            })
        })
        .map_err(DbError::from_rusqlite)?;
    let mut out = Vec::new();
    for row in rows {
        out.push(row.map_err(DbError::from_rusqlite)?);
    }
    Ok(out)
}

fn read_relationships_inbound(
    conn: &Connection,
    table: &str,
) -> Result<Vec<RelationshipEnd>, DbError> {
    let mut stmt = conn
        .prepare(&format!(
            "SELECT name, child_table, child_column, parent_column, on_delete, on_update \
             FROM {REL_TABLE} \
             WHERE parent_table = ?1 \
             ORDER BY name;"
        ))
        .map_err(DbError::from_rusqlite)?;
    let rows = stmt
        .query_map([table], |row| {
            let on_delete: String = row.get(4)?;
            let on_update: String = row.get(5)?;
            Ok(RelationshipEnd {
                name: row.get(0)?,
                other_table: row.get(1)?,
                other_column: row.get(2)?,
                local_column: row.get(3)?,
                on_delete: on_delete
                    .parse::<ReferentialAction>()
                    .unwrap_or(ReferentialAction::NoAction),
                on_update: on_update
                    .parse::<ReferentialAction>()
                    .unwrap_or(ReferentialAction::NoAction),
            })
        })
        .map_err(DbError::from_rusqlite)?;
    let mut out = Vec::new();
    for row in rows {
        out.push(row.map_err(DbError::from_rusqlite)?);
    }
    Ok(out)
}

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

    fn db() -> Database {
        Database::open(":memory:").expect("open in-memory")
    }

    fn col(name: &str, ty: Type) -> ColumnSpec {
        ColumnSpec {
            name: name.to_string(),
            ty,
        }
    }

    /// Convenience: a `serial`-PK table with a single `id` column.
    async fn make_id_table(db: &Database, name: &str) -> TableDescription {
        db.create_table(
            name.to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .expect("create table")
    }

    #[tokio::test]
    async fn open_in_memory_succeeds() {
        let _ = db();
    }

    #[tokio::test]
    async fn create_table_with_serial_pk_appears_in_list() {
        let db = db();
        make_id_table(&db, "Customers").await;
        let tables = db.list_tables().await.unwrap();
        assert_eq!(tables, vec!["Customers".to_string()]);
    }

    #[tokio::test]
    async fn create_table_with_serial_pk_describes_correctly() {
        let db = db();
        let desc = make_id_table(&db, "Customers").await;
        assert_eq!(desc.name, "Customers");
        assert_eq!(desc.columns.len(), 1);
        let id = &desc.columns[0];
        assert_eq!(id.name, "id");
        assert!(id.primary_key);
        assert_eq!(id.user_type, Some(Type::Serial));
        assert_eq!(id.sqlite_type.to_uppercase(), "INTEGER");
    }

    #[tokio::test]
    async fn create_table_with_text_pk_works() {
        let db = db();
        let desc = db
            .create_table(
                "Customers".to_string(),
                vec![col("email", Type::Text)],
                vec!["email".to_string()],
            None)
            .await
            .unwrap();
        assert_eq!(desc.columns.len(), 1);
        assert_eq!(desc.columns[0].name, "email");
        assert_eq!(desc.columns[0].user_type, Some(Type::Text));
        assert_eq!(desc.columns[0].sqlite_type.to_uppercase(), "TEXT");
        assert!(desc.columns[0].primary_key);
    }

    #[tokio::test]
    async fn create_table_with_compound_pk_works() {
        let db = db();
        let desc = db
            .create_table(
                "OrderLines".to_string(),
                vec![col("order_id", Type::Int), col("product_id", Type::Int)],
                vec!["order_id".to_string(), "product_id".to_string()],
            None)
            .await
            .unwrap();
        assert_eq!(desc.columns.len(), 2);
        assert!(desc.columns.iter().all(|c| c.primary_key));
    }

    #[tokio::test]
    async fn create_table_with_pedagogically_unusual_pk_type_still_works() {
        // The grammar lets users try anything; the DB layer just
        // does what they ask.
        let db = db();
        let desc = db
            .create_table(
                "T".to_string(),
                vec![col("flag", Type::Bool)],
                vec!["flag".to_string()],
            None)
            .await
            .unwrap();
        assert!(desc.columns[0].primary_key);
    }

    #[tokio::test]
    async fn create_table_rejects_zero_columns() {
        let db = db();
        let err = db
            .create_table("T".to_string(), Vec::new(), Vec::new(), None)
            .await
            .unwrap_err();
        assert!(matches!(err, DbError::Unsupported(_)), "got {err:?}");
    }

    #[tokio::test]
    async fn drop_table_removes_it_from_list() {
        let db = db();
        make_id_table(&db, "T").await;
        db.drop_table("T".to_string(), None).await.unwrap();
        let tables = db.list_tables().await.unwrap();
        assert!(tables.is_empty());
    }

    #[tokio::test]
    async fn add_column_appends_to_existing_table() {
        let db = db();
        make_id_table(&db, "Customers").await;
        let desc = db
            .add_column("Customers".to_string(), "Name".to_string(), Type::Text, None)
            .await
            .unwrap();
        let names: Vec<_> = desc.columns.iter().map(|c| c.name.as_str()).collect();
        assert_eq!(names, vec!["id", "Name"]);
        let name_col = desc.columns.iter().find(|c| c.name == "Name").unwrap();
        assert_eq!(name_col.user_type, Some(Type::Text));
        assert_eq!(name_col.sqlite_type.to_uppercase(), "TEXT");
    }

    #[tokio::test]
    async fn user_facing_types_round_trip_through_metadata() {
        let db = db();
        // Create with a serial PK and add columns of every type
        // that would otherwise be erased by SQLite (date,
        // datetime, decimal — all backed by TEXT).
        make_id_table(&db, "T").await;
        for ty in [Type::Date, Type::DateTime, Type::Decimal, Type::ShortId] {
            db.add_column("T".to_string(), format!("c_{ty}"), ty, None)
                .await
                .unwrap();
        }
        let desc = db.describe_table("T".to_string(), None).await.unwrap();
        let id_col = desc.columns.iter().find(|c| c.name == "id").unwrap();
        assert_eq!(id_col.user_type, Some(Type::Serial));
        for ty in [Type::Date, Type::DateTime, Type::Decimal, Type::ShortId] {
            let col_name = format!("c_{ty}");
            let c = desc.columns.iter().find(|c| c.name == col_name).unwrap();
            assert_eq!(c.user_type, Some(ty), "mismatch for {col_name}");
        }
    }

    #[tokio::test]
    async fn list_tables_excludes_internal_metadata_table() {
        let db = db();
        make_id_table(&db, "Visible").await;
        let tables = db.list_tables().await.unwrap();
        assert_eq!(tables, vec!["Visible".to_string()]);
        // Metadata table is present in the underlying schema but
        // hidden from list_tables.
    }

    #[tokio::test]
    async fn drop_table_clears_metadata_so_recreate_starts_fresh() {
        let db = db();
        // Create with a date column.
        db.create_table(
            "T".to_string(),
            vec![col("when", Type::Date)],
            vec!["when".to_string()],
        None)
        .await
        .unwrap();
        let before = db.describe_table("T".to_string(), None).await.unwrap();
        assert_eq!(before.columns[0].user_type, Some(Type::Date));

        // Drop it.
        db.drop_table("T".to_string(), None).await.unwrap();

        // Recreate with a different type for the same-named column;
        // the metadata for the new table must reflect the new type
        // (i.e. metadata from the previous incarnation must not
        // bleed through).
        db.create_table(
            "T".to_string(),
            vec![col("when", Type::DateTime)],
            vec!["when".to_string()],
        None)
        .await
        .unwrap();
        let after = db.describe_table("T".to_string(), None).await.unwrap();
        assert_eq!(after.columns[0].user_type, Some(Type::DateTime));
    }

    #[tokio::test]
    async fn add_column_for_each_value_type() {
        let db = db();
        make_id_table(&db, "T").await;
        for ty in [Type::Text, Type::Int, Type::Real, Type::Bool, Type::ShortId] {
            let col_name = format!("c_{ty}");
            db.add_column("T".to_string(), col_name.clone(), ty, None)
                .await
                .unwrap_or_else(|e| panic!("type {ty} failed: {e}"));
        }
        let desc = db.describe_table("T".to_string(), None).await.unwrap();
        // 5 user columns + the id PK column.
        assert_eq!(desc.columns.len(), 6);
    }

    #[tokio::test]
    async fn add_column_rejects_serial_with_unsupported_error() {
        let db = db();
        make_id_table(&db, "T").await;
        let err = db
            .add_column("T".to_string(), "id2".to_string(), Type::Serial, None)
            .await
            .unwrap_err();
        assert!(matches!(err, DbError::Unsupported(_)), "got {err:?}");
    }

    #[tokio::test]
    async fn create_table_duplicate_returns_already_exists() {
        let db = db();
        make_id_table(&db, "T").await;
        let err = db
            .create_table(
                "T".to_string(),
                vec![col("id", Type::Serial)],
                vec!["id".to_string()],
            None)
            .await
            .unwrap_err();
        match err {
            DbError::Sqlite { kind, .. } => assert_eq!(kind, SqliteErrorKind::AlreadyExists),
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn drop_nonexistent_table_returns_no_such_table() {
        let db = db();
        let err = db.drop_table("Ghost".to_string(), None).await.unwrap_err();
        match err {
            DbError::Sqlite { kind, .. } => assert_eq!(kind, SqliteErrorKind::NoSuchTable),
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn add_column_to_missing_table_returns_no_such_table() {
        let db = db();
        let err = db
            .add_column("Ghost".to_string(), "x".to_string(), Type::Text, None)
            .await
            .unwrap_err();
        match err {
            DbError::Sqlite { kind, .. } => assert_eq!(kind, SqliteErrorKind::NoSuchTable),
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn describe_missing_table_returns_no_such_table() {
        let db = db();
        let err = db.describe_table("Ghost".to_string(), None).await.unwrap_err();
        match err {
            DbError::Sqlite { kind, .. } => assert_eq!(kind, SqliteErrorKind::NoSuchTable),
            other => panic!("unexpected error: {other:?}"),
        }
    }

    // --- Relationship tests ---

    async fn customers_orders_setup(db: &Database) {
        db.create_table(
            "Customers".to_string(),
            vec![col("id", Type::Serial), col("Name", Type::Text)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.add_column("Orders".to_string(), "CustId".to_string(), Type::Int, None)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn add_relationship_appears_outbound_on_child() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        let orders = db.describe_table("Orders".to_string(), None).await.unwrap();
        assert_eq!(orders.outbound_relationships.len(), 1);
        let rel = &orders.outbound_relationships[0];
        assert_eq!(rel.local_column, "CustId");
        assert_eq!(rel.other_table, "Customers");
        assert_eq!(rel.other_column, "id");
        assert_eq!(rel.name, "Customers_id_to_Orders_CustId");
    }

    #[tokio::test]
    async fn add_relationship_appears_inbound_on_parent() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        let customers = db.describe_table("Customers".to_string(), None).await.unwrap();
        assert_eq!(customers.inbound_relationships.len(), 1);
        let rel = &customers.inbound_relationships[0];
        assert_eq!(rel.local_column, "id");
        assert_eq!(rel.other_table, "Orders");
        assert_eq!(rel.other_column, "CustId");
    }

    #[tokio::test]
    async fn add_relationship_with_user_supplied_name() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            Some("cust_orders".to_string()),
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::Cascade,
            ReferentialAction::SetNull,
            false,
        None)
        .await
        .unwrap();
        let orders = db.describe_table("Orders".to_string(), None).await.unwrap();
        let rel = &orders.outbound_relationships[0];
        assert_eq!(rel.name, "cust_orders");
        assert_eq!(rel.on_delete, ReferentialAction::Cascade);
        assert_eq!(rel.on_update, ReferentialAction::SetNull);
    }

    #[tokio::test]
    async fn add_relationship_with_create_fk_creates_the_column() {
        let db = db();
        // Create child *without* the FK column.
        db.create_table(
            "Customers".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();

        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            true, // --create-fk
            None,
        )
        .await
        .unwrap();
        let orders = db.describe_table("Orders".to_string(), None).await.unwrap();
        // The auto-created FK column has user_type Int (Serial's
        // fk_target_type), not Serial.
        let cust = orders
            .columns
            .iter()
            .find(|c| c.name == "CustId")
            .expect("FK column auto-created");
        assert_eq!(cust.user_type, Some(Type::Int));
        assert_eq!(orders.outbound_relationships.len(), 1);
    }

    #[tokio::test]
    async fn add_relationship_without_create_fk_errors_when_column_missing() {
        let db = db();
        db.create_table(
            "Customers".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        let err = db
            .add_relationship(
                None,
                "Customers".to_string(),
                "id".to_string(),
                "Orders".to_string(),
                "CustId".to_string(),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
            None)
            .await
            .unwrap_err();
        assert!(matches!(err, DbError::Unsupported(_)), "got {err:?}");
    }

    #[tokio::test]
    async fn add_relationship_with_type_mismatch_errors_with_advice() {
        let db = db();
        db.create_table(
            "Customers".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        // Wrong type — text instead of int.
        db.add_column("Orders".to_string(), "CustId".to_string(), Type::Text, None)
            .await
            .unwrap();

        let err = db
            .add_relationship(
                None,
                "Customers".to_string(),
                "id".to_string(),
                "Orders".to_string(),
                "CustId".to_string(),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
            None)
            .await
            .unwrap_err();
        match err {
            DbError::Unsupported(msg) => {
                assert!(msg.contains("type mismatch"), "{msg}");
                assert!(msg.contains("text") && msg.contains("int"), "{msg}");
            }
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn add_relationship_against_non_pk_errors() {
        let db = db();
        // Customers has a non-PK column we'll try to point at.
        db.create_table(
            "Customers".to_string(),
            vec![col("id", Type::Serial), col("Name", Type::Text)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.add_column("Orders".to_string(), "CustName".to_string(), Type::Text, None)
            .await
            .unwrap();
        let err = db
            .add_relationship(
                None,
                "Customers".to_string(),
                "Name".to_string(),
                "Orders".to_string(),
                "CustName".to_string(),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
            None)
            .await
            .unwrap_err();
        match err {
            DbError::Unsupported(msg) => assert!(msg.contains("not a primary key"), "{msg}"),
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn drop_relationship_by_name_clears_both_sides() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            Some("cust_orders".to_string()),
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        db.drop_relationship(RelationshipSelector::Named {
            name: "cust_orders".to_string(),
        }, None)
        .await
        .unwrap();
        let orders = db.describe_table("Orders".to_string(), None).await.unwrap();
        let customers = db.describe_table("Customers".to_string(), None).await.unwrap();
        assert!(orders.outbound_relationships.is_empty());
        assert!(customers.inbound_relationships.is_empty());
    }

    #[tokio::test]
    async fn drop_relationship_by_endpoints_works() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        db.drop_relationship(RelationshipSelector::Endpoints {
            parent_table: "Customers".to_string(),
            parent_column: "id".to_string(),
            child_table: "Orders".to_string(),
            child_column: "CustId".to_string(),
        }, None)
        .await
        .unwrap();
        let orders = db.describe_table("Orders".to_string(), None).await.unwrap();
        assert!(orders.outbound_relationships.is_empty());
    }

    #[tokio::test]
    async fn drop_table_with_inbound_relationship_errors() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        let err = db.drop_table("Customers".to_string(), None).await.unwrap_err();
        match err {
            DbError::Unsupported(msg) => {
                assert!(msg.contains("referenced by"), "{msg}");
            }
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn drop_child_table_cleans_relationship_metadata() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        // Dropping the child is allowed (no inbound relationships
        // on Orders) and cleans the metadata.
        db.drop_table("Orders".to_string(), None).await.unwrap();
        let customers = db.describe_table("Customers".to_string(), None).await.unwrap();
        assert!(customers.inbound_relationships.is_empty());
    }

    #[tokio::test]
    async fn add_relationship_with_duplicate_name_errors() {
        let db = db();
        customers_orders_setup(&db).await;
        db.add_column("Orders".to_string(), "OtherCust".to_string(), Type::Int, None)
            .await
            .unwrap();
        db.add_relationship(
            Some("dup".to_string()),
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        let err = db
            .add_relationship(
                Some("dup".to_string()),
                "Customers".to_string(),
                "id".to_string(),
                "Orders".to_string(),
                "OtherCust".to_string(),
                ReferentialAction::NoAction,
                ReferentialAction::NoAction,
                false,
            None)
            .await
            .unwrap_err();
        match err {
            DbError::Unsupported(msg) => assert!(msg.contains("already exists"), "{msg}"),
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn rebuild_preserves_existing_data_through_relationship_change() {
        let db = db();
        customers_orders_setup(&db).await;
        // Direct INSERT through the underlying connection isn't
        // exposed via the public API yet (C5). The point of this
        // test is to verify the rebuild itself works on populated
        // tables — we add a column with a default-able operation
        // and then add a relationship to ensure the table survives.
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::Cascade,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        // After the rebuild, the original columns are still
        // present with the right user types, and any extra
        // metadata (Name on Customers) survives.
        let customers = db.describe_table("Customers".to_string(), None).await.unwrap();
        let names: Vec<&str> = customers.columns.iter().map(|c| c.name.as_str()).collect();
        assert_eq!(names, vec!["id", "Name"]);
        let name_col = customers.columns.iter().find(|c| c.name == "Name").unwrap();
        assert_eq!(name_col.user_type, Some(Type::Text));
    }

    // --- Data operations (C5) ---

    async fn customers_table(db: &Database) {
        db.create_table(
            "Customers".to_string(),
            vec![col("id", Type::Serial), col("Name", Type::Text)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn insert_short_form_skips_serial_column() {
        let db = db();
        customers_table(&db).await;
        let result = db
            .insert(
                "Customers".to_string(),
                None,
                vec![Value::Text("Alice".to_string())],
            None)
            .await
            .unwrap();
        assert_eq!(result.rows_affected, 1);
        // The InsertResult itself carries the just-inserted row.
        assert_eq!(result.data.rows.len(), 1);
        assert_eq!(result.data.rows[0][1], Some("Alice".to_string()));
        let data = db.query_data("Customers".to_string(), None).await.unwrap();
        assert_eq!(data.columns, vec!["id".to_string(), "Name".to_string()]);
        assert_eq!(data.rows.len(), 1);
        assert_eq!(data.rows[0][1], Some("Alice".to_string()));
        // id was auto-filled by SQLite.
        assert_eq!(data.rows[0][0], Some("1".to_string()));
    }

    #[tokio::test]
    async fn insert_short_form_auto_generates_shortid() {
        let db = db();
        db.create_table(
            "Tags".to_string(),
            vec![col("id", Type::ShortId), col("Label", Type::Text)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.insert(
            "Tags".to_string(),
            None,
            vec![Value::Text("database".to_string())],
        None)
        .await
        .unwrap();
        let data = db.query_data("Tags".to_string(), None).await.unwrap();
        let id = data.rows[0][0].as_ref().expect("auto-generated id");
        assert!(
            id.len() >= 10 && id.len() <= 12,
            "expected a base58 shortid, got {id}"
        );
    }

    #[tokio::test]
    async fn insert_explicit_columns_uses_user_values() {
        let db = db();
        customers_table(&db).await;
        db.insert(
            "Customers".to_string(),
            Some(vec!["id".to_string(), "Name".to_string()]),
            vec![Value::Number("99".to_string()), Value::Text("Bob".to_string())],
        None)
        .await
        .unwrap();
        let data = db.query_data("Customers".to_string(), None).await.unwrap();
        assert_eq!(data.rows[0][0], Some("99".to_string()));
        assert_eq!(data.rows[0][1], Some("Bob".to_string()));
    }

    #[tokio::test]
    async fn insert_with_type_mismatch_returns_invalid_value() {
        let db = db();
        customers_table(&db).await;
        // 42 is a number, but Name expects text.
        let err = db
            .insert(
                "Customers".to_string(),
                Some(vec!["Name".to_string()]),
                vec![Value::Number("42".to_string())],
            None)
            .await
            .unwrap_err();
        assert!(matches!(err, DbError::InvalidValue(_)), "got {err:?}");
    }

    #[tokio::test]
    async fn update_with_where_affects_only_matching_rows() {
        let db = db();
        customers_table(&db).await;
        for name in ["Alice", "Bob"] {
            db.insert(
                "Customers".to_string(),
                None,
                vec![Value::Text(name.to_string())],
            None)
            .await
            .unwrap();
        }
        let result = db
            .update(
                "Customers".to_string(),
                vec![("Name".to_string(), Value::Text("Alicia".to_string()))],
                RowFilter::Where {
                    column: "id".to_string(),
                    value: Value::Number("1".to_string()),
                },
            None)
            .await
            .unwrap();
        assert_eq!(result.rows_affected, 1);
        // The UpdateResult contains only the updated rows.
        assert_eq!(result.data.rows.len(), 1);
        assert_eq!(result.data.rows[0][1], Some("Alicia".to_string()));
        let data = db.query_data("Customers".to_string(), None).await.unwrap();
        assert_eq!(data.rows[0][1], Some("Alicia".to_string()));
        assert_eq!(data.rows[1][1], Some("Bob".to_string()));
    }

    #[tokio::test]
    async fn update_with_all_rows_affects_everything() {
        let db = db();
        customers_table(&db).await;
        for name in ["Alice", "Bob", "Carol"] {
            db.insert(
                "Customers".to_string(),
                None,
                vec![Value::Text(name.to_string())],
            None)
            .await
            .unwrap();
        }
        let result = db
            .update(
                "Customers".to_string(),
                vec![("Name".to_string(), Value::Text("X".to_string()))],
                RowFilter::AllRows,
            None)
            .await
            .unwrap();
        assert_eq!(result.rows_affected, 3);
        assert_eq!(result.data.rows.len(), 3);
    }

    #[tokio::test]
    async fn delete_with_where_removes_matching_rows() {
        let db = db();
        customers_table(&db).await;
        for name in ["Alice", "Bob"] {
            db.insert(
                "Customers".to_string(),
                None,
                vec![Value::Text(name.to_string())],
            None)
            .await
            .unwrap();
        }
        let result = db
            .delete(
                "Customers".to_string(),
                RowFilter::Where {
                    column: "id".to_string(),
                    value: Value::Number("1".to_string()),
                },
            None)
            .await
            .unwrap();
        assert_eq!(result.rows_affected, 1);
        assert!(result.cascade.is_empty(), "no children to cascade to");
        let data = db.query_data("Customers".to_string(), None).await.unwrap();
        assert_eq!(data.rows.len(), 1);
        assert_eq!(data.rows[0][1], Some("Bob".to_string()));
    }

    #[tokio::test]
    async fn fk_violation_message_lists_outbound_relationships() {
        let db = db();
        // Two-table setup with FK.
        customers_table(&db).await;
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial), col("CustId", Type::Int)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();

        // Try to insert an Order pointing at a nonexistent Customer.
        let err = db
            .insert(
                "Orders".to_string(),
                Some(vec!["CustId".to_string()]),
                vec![Value::Number("999".to_string())],
            None)
            .await
            .unwrap_err();
        match err {
            DbError::Sqlite { message, .. } => {
                assert!(
                    message.to_ascii_lowercase().contains("foreign key"),
                    "{message}"
                );
                assert!(
                    message.contains("Orders.CustId → Customers.id"),
                    "FK enrichment should list the FK: {message}"
                );
            }
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn cascade_delete_propagates_to_children() {
        let db = db();
        customers_table(&db).await;
        db.create_table(
            "Orders".to_string(),
            vec![col("id", Type::Serial), col("CustId", Type::Int)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.add_relationship(
            None,
            "Customers".to_string(),
            "id".to_string(),
            "Orders".to_string(),
            "CustId".to_string(),
            ReferentialAction::Cascade,
            ReferentialAction::NoAction,
            false,
        None)
        .await
        .unwrap();
        db.insert(
            "Customers".to_string(),
            None,
            vec![Value::Text("Alice".to_string())],
        None)
        .await
        .unwrap();
        db.insert(
            "Orders".to_string(),
            Some(vec!["CustId".to_string()]),
            vec![Value::Number("1".to_string())],
        None)
        .await
        .unwrap();
        // Delete Alice — cascades to Orders.
        db.delete(
            "Customers".to_string(),
            RowFilter::Where {
                column: "id".to_string(),
                value: Value::Number("1".to_string()),
            },
        None)
        .await
        .unwrap();
        let orders = db.query_data("Orders".to_string(), None).await.unwrap();
        assert!(orders.rows.is_empty(), "child rows should be cascaded");
    }

    #[tokio::test]
    async fn query_data_renders_bools_as_words() {
        let db = db();
        db.create_table(
            "Flags".to_string(),
            vec![col("id", Type::Serial), col("Active", Type::Bool)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.insert(
            "Flags".to_string(),
            None,
            vec![Value::Bool(true)],
        None)
        .await
        .unwrap();
        db.insert(
            "Flags".to_string(),
            None,
            vec![Value::Bool(false)],
        None)
        .await
        .unwrap();
        let data = db.query_data("Flags".to_string(), None).await.unwrap();
        assert_eq!(data.rows[0][1], Some("true".to_string()));
        assert_eq!(data.rows[1][1], Some("false".to_string()));
    }

    #[tokio::test]
    async fn query_data_renders_null_as_none() {
        let db = db();
        db.create_table(
            "T".to_string(),
            vec![col("id", Type::Serial), col("note", Type::Text)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        db.insert(
            "T".to_string(),
            None,
            vec![Value::Null],
        None)
        .await
        .unwrap();
        let data = db.query_data("T".to_string(), None).await.unwrap();
        assert_eq!(data.rows[0][1], None);
    }

    #[tokio::test]
    async fn quoted_table_names_round_trip() {
        let db = db();
        // Identifier with internal whitespace would not parse via the DSL
        // today, but the DB layer should still handle it correctly.
        db.create_table(
            "Order Lines".to_string(),
            vec![col("id", Type::Serial)],
            vec!["id".to_string()],
        None)
        .await
        .unwrap();
        let tables = db.list_tables().await.unwrap();
        assert_eq!(tables, vec!["Order Lines".to_string()]);
        let desc = db.describe_table("Order Lines".to_string(), None).await.unwrap();
        assert_eq!(desc.name, "Order Lines");
    }
}