drizzle 0.1.13

// =============================================================================
// Temp DB Path Generation (file-based test isolation)
// =============================================================================

use std::path::PathBuf;
use std::sync::Once;
use std::sync::atomic::{AtomicU64, Ordering};

const TEST_DB_DIR: &str = "drizzle_rs_tests";

fn ensure_test_db_dir() {
    static INIT: Once = Once::new();
    INIT.call_once(|| {
        // Each process gets its own subdirectory keyed by PID, so concurrent
        // `cargo test` invocations never interfere with each other.
        let dir = std::env::temp_dir()
            .join(TEST_DB_DIR)
            .join(std::process::id().to_string());
        // Clean up any stale files from a previous run with the same PID.
        if dir.exists() {
            let _ = std::fs::remove_dir_all(&dir);
        }
        let _ = std::fs::create_dir_all(&dir);
    });
}

pub fn temp_db_path() -> PathBuf {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    ensure_test_db_dir();
    let id = COUNTER.fetch_add(1, Ordering::Relaxed);
    let pid = std::process::id();
    std::env::temp_dir()
        .join(TEST_DB_DIR)
        .join(pid.to_string())
        .join(format!("test_{}.db", id))
}

// =============================================================================
// Test Failure Report Infrastructure
// =============================================================================

const BOX_WIDTH: usize = 80;
const CONTENT_WIDTH: usize = BOX_WIDTH - 4; // Account for "│ " prefix and " │" suffix

/// Captured SQL statement with optional error
#[derive(Clone, Debug)]
pub struct CapturedStatement {
    pub sql: String,
    pub params: Option<String>,
    pub source: Option<String>,
    pub error: Option<String>,
}

/// Process-global panic hook that prints the SQL trail for the currently
/// running `#[drizzle::test]` function.
///
/// The naive approach — `take_hook` / `set_hook` on every test entry — races
/// under parallel test execution: Test B's `take_hook` can capture Test A's
/// hook, and the saved `__prev_hook` chain ends up pointing at closures owned
/// by other tests. The result is cross-test SQL trails on panic.
///
/// Instead we install one global hook, keyed on a `thread_local!` that each
/// test sets via a RAII guard. The default `#[tokio::test]` runtime is
/// `flavor = "current_thread"`, so the whole test body (including async
/// polling) runs on the thread that constructed the guard. `#[test]` is
/// obviously single-threaded. Either way the hook finds the right trail.
pub mod panic_hook {
    use super::CapturedStatement;
    use std::cell::RefCell;
    use std::sync::{Arc, Mutex, Once};

    type StatementTrail = Arc<Mutex<Vec<CapturedStatement>>>;
    type TrailSlot = RefCell<Option<(String, StatementTrail)>>;

    thread_local! {
        static CURRENT_TRAIL: TrailSlot = const { RefCell::new(None) };
    }

    static INSTALL: Once = Once::new();

    /// Install the process-global panic hook. Idempotent — safe to call from
    /// every test's setup; the `Once` gate means the `take_hook`/`set_hook`
    /// swap happens exactly once for the whole process.
    pub fn install_once() {
        INSTALL.call_once(|| {
            let prev = std::panic::take_hook();
            std::panic::set_hook(Box::new(move |info| {
                let mut buf = String::new();
                CURRENT_TRAIL.with(|cell| {
                    // `try_borrow` (not `borrow`) — if the panic fires while
                    // the cell is borrowed mutably elsewhere we want to fall
                    // through to the default hook rather than double-panic.
                    let Ok(guard) = cell.try_borrow() else { return };
                    let Some((name, arc)) = guard.as_ref() else { return };
                    use std::fmt::Write as _;
                    let _ = writeln!(buf, "[{}] panicked", name);
                    // `try_lock` guards against deadlock if the panic happens
                    // while a `record_sql` / `report` caller holds the mutex.
                    match arc.try_lock() {
                        Ok(stmts) => {
                            if !stmts.is_empty() {
                                let _ = writeln!(buf, "captured statements ({}):", stmts.len());
                                for (i, s) in stmts.iter().enumerate() {
                                    if let Some(src) = &s.source {
                                        let _ = writeln!(buf, "  #{} {}", i + 1, src);
                                    }
                                    let _ = writeln!(buf, "     sql: {}", s.sql);
                                    if let Some(p) = &s.params {
                                        let _ = writeln!(buf, "     params: {}", p);
                                    }
                                    if let Some(e) = &s.error {
                                        let _ = writeln!(buf, "     error: {}", e);
                                    }
                                }
                            }
                        }
                        Err(_) => {
                            let _ = writeln!(
                                buf,
                                "(statements mutex was locked at time of panic; SQL trail unavailable)"
                            );
                        }
                    }
                });
                if !buf.is_empty() {
                    eprintln!("{}", buf);
                }
                prev(info);
            }));
        });
    }

    /// RAII guard that stashes `(test_name, statements)` in the thread-local
    /// on construction and restores the previous value on drop (normal exit
    /// or unwinding). Tests never nest today, but the save-and-restore keeps
    /// us safe if they ever do.
    pub struct TrailGuard {
        prev: Option<(String, StatementTrail)>,
    }

    impl TrailGuard {
        pub fn new(test_name: String, statements: StatementTrail) -> Self {
            let prev =
                CURRENT_TRAIL.with(|cell| cell.borrow_mut().replace((test_name, statements)));
            Self { prev }
        }
    }

    impl Drop for TrailGuard {
        fn drop(&mut self) {
            let prev = self.prev.take();
            CURRENT_TRAIL.with(|cell| {
                *cell.borrow_mut() = prev;
            });
        }
    }
}

/// Calculate display width accounting for special characters
fn display_width(s: &str) -> usize {
    s.chars()
        .map(|c| match c {
            '✓' | '✗' | '→' => 1,
            _ if c.is_ascii() => 1,
            _ => 2,
        })
        .sum()
}

/// Expand tabs to spaces
fn expand_tabs(s: &str) -> String {
    s.replace('\t', "    ")
}

/// Wrap text to fit within a given width
fn wrap_text(text: &str, width: usize) -> Vec<String> {
    let text = expand_tabs(text);
    let mut lines = Vec::new();

    for line in text.lines() {
        if line.is_empty() {
            lines.push(String::new());
            continue;
        }

        if display_width(line) <= width {
            lines.push(line.to_string());
        } else {
            let mut current_line = String::new();
            let mut current_width = 0;

            for word in line.split_inclusive(' ') {
                let word_width = display_width(word);

                if current_width + word_width <= width {
                    current_line.push_str(word);
                    current_width += word_width;
                } else {
                    // Flush current line if non-empty
                    if !current_line.is_empty() {
                        lines.push(current_line.trim_end().to_string());
                        current_line = String::new();
                        current_width = 0;
                    }

                    if word_width <= width {
                        current_line.push_str(word);
                        current_width = word_width;
                    } else {
                        // Word is longer than width, force split
                        let mut chars = word.chars().peekable();
                        while chars.peek().is_some() {
                            let mut chunk = String::new();
                            let mut chunk_width = 0;
                            while let Some(&c) = chars.peek() {
                                let c_width = if c.is_ascii() { 1 } else { 2 };
                                if chunk_width + c_width > width {
                                    break;
                                }
                                chunk.push(chars.next().unwrap());
                                chunk_width += c_width;
                            }
                            if !chunk.is_empty() {
                                lines.push(chunk);
                            }
                        }
                    }
                }
            }

            if !current_line.is_empty() {
                lines.push(current_line.trim_end().to_string());
            }
        }
    }

    if lines.is_empty() {
        lines.push(String::new());
    }

    lines
}

/// Format a line with proper box drawing
fn box_line(content: &str, prefix: &str) -> String {
    let content = expand_tabs(content);
    let prefix_width = display_width(prefix);
    let content_width = display_width(&content);
    let total_used = prefix_width + content_width;
    let padding = CONTENT_WIDTH.saturating_sub(total_used);
    format!("│ {}{}{} │\n", prefix, content, " ".repeat(padding))
}

/// Format a section header
fn section_header(title: &str) -> String {
    // Total inner width is BOX_WIDTH - 2 (for the ├ and ┤)
    // Format: ├─ TITLE ─────...─────┤
    // So: 1 (─) + 1 (space) + title + 1 (space) + remaining dashes = BOX_WIDTH - 2
    let inner_width = BOX_WIDTH - 2;
    let title_width = display_width(title);
    let used = 1 + 1 + title_width + 1; // "─ TITLE "
    let dashes = inner_width.saturating_sub(used);
    format!("├─ {} {}┤\n", title, "─".repeat(dashes))
}

/// Format the top border
fn top_border() -> String {
    format!("╔{}╗\n", "═".repeat(BOX_WIDTH - 2))
}

/// Format the bottom border
fn bottom_border() -> String {
    format!("╚{}╝\n", "═".repeat(BOX_WIDTH - 2))
}

/// Format an empty line within the box
fn empty_box_line() -> String {
    format!("│{}│\n", " ".repeat(BOX_WIDTH - 2))
}

/// Context for generating a structured failure report
pub struct FailureContext<'a> {
    pub driver_name: &'a str,
    pub test_name: &'a str,
    pub error: &'a dyn std::fmt::Display,
    pub expected: Option<&'a str>,
    pub actual: Option<&'a str>,
    pub failed_operation: Option<&'a str>,
    pub schema_ddl: &'a [String],
    pub statements: &'a [CapturedStatement],
}

/// Generate a structured failure report for any driver
pub fn failure_report(ctx: &FailureContext<'_>) -> String {
    let FailureContext {
        driver_name,
        test_name,
        error,
        expected,
        actual,
        failed_operation,
        schema_ddl,
        statements,
    } = ctx;
    let mut report = String::new();

    // Header
    let header = "TEST FAILURE REPORT";
    let header_width = display_width(header);
    let header_padding = (BOX_WIDTH - 2 - header_width) / 2;
    let header_padding_right = BOX_WIDTH - 2 - header_width - header_padding;

    report.push('\n');
    report.push_str(&top_border());
    report.push_str(&format!(
        "║{}{}{}║\n",
        " ".repeat(header_padding),
        header,
        " ".repeat(header_padding_right)
    ));
    report.push_str(&bottom_border());
    report.push('\n');

    // Test identification section
    report.push_str(&top_border());
    report.push_str(&section_header("TEST"));
    let name_lines = wrap_text(test_name, CONTENT_WIDTH - 8);
    for (i, line) in name_lines.iter().enumerate() {
        let prefix = if i == 0 { "Name:   " } else { "        " };
        report.push_str(&box_line(line, prefix));
    }
    report.push_str(&box_line(driver_name, "Driver: "));
    report.push_str(&bottom_border());
    report.push('\n');

    // Error section
    report.push_str(&top_border());
    report.push_str(&section_header("ERROR"));
    let error_text = format!("{}", error);
    let error_lines = wrap_text(&error_text, CONTENT_WIDTH);
    for line in error_lines {
        report.push_str(&box_line(&line, ""));
    }
    report.push_str(&bottom_border());
    report.push('\n');

    // Expected vs Actual (if provided)
    if expected.is_some() || actual.is_some() {
        report.push_str(&top_border());
        report.push_str(&section_header("COMPARISON"));
        if let Some(exp) = expected {
            let exp_lines = wrap_text(exp, CONTENT_WIDTH - 10);
            for (i, line) in exp_lines.iter().enumerate() {
                if i == 0 {
                    report.push_str(&box_line(line, "Expected: "));
                } else {
                    report.push_str(&box_line(line, "          "));
                }
            }
        }
        if let Some(act) = actual {
            let act_lines = wrap_text(act, CONTENT_WIDTH - 10);
            for (i, line) in act_lines.iter().enumerate() {
                if i == 0 {
                    report.push_str(&box_line(line, "Actual:   "));
                } else {
                    report.push_str(&box_line(line, "          "));
                }
            }
        }
        report.push_str(&bottom_border());
        report.push('\n');
    }

    // Failed operation section (if provided, skip when redundant with last statement)
    if let Some(op) = failed_operation {
        let redundant = statements
            .last()
            .and_then(|s| s.source.as_deref())
            .is_some_and(|src| src == *op);
        if !redundant {
            report.push_str(&top_border());
            report.push_str(&section_header("FAILED OPERATION"));
            let op_lines = wrap_text(op, CONTENT_WIDTH - 2);
            for line in op_lines {
                report.push_str(&box_line(&line, "  "));
            }
            report.push_str(&bottom_border());
            report.push('\n');
        }
    }

    // Schema DDL section
    report.push_str(&top_border());
    report.push_str(&section_header("SCHEMA DDL"));
    if schema_ddl.is_empty() {
        report.push_str(&box_line("(no DDL statements captured)", ""));
    } else {
        for (i, ddl) in schema_ddl.iter().enumerate() {
            report.push_str(&box_line(&format!("[{}]", i + 1), ""));
            for line in ddl.lines() {
                let expanded = expand_tabs(line);
                let wrapped = wrap_text(&expanded, CONTENT_WIDTH - 2);
                for wrap_line in wrapped {
                    report.push_str(&box_line(&wrap_line, "  "));
                }
            }
            if i < schema_ddl.len() - 1 {
                report.push_str(&empty_box_line());
            }
        }
    }
    report.push_str(&bottom_border());
    report.push('\n');

    // Executed statements section
    report.push_str(&top_border());
    report.push_str(&section_header("EXECUTED STATEMENTS"));
    if statements.is_empty() {
        report.push_str(&box_line("(no statements executed)", ""));
    } else {
        for (i, stmt) in statements.iter().enumerate() {
            let status = if stmt.error.is_some() { "✗" } else { "✓" };
            report.push_str(&box_line(&format!("[{}]", i + 1), &format!("{} ", status)));

            if let Some(source) = &stmt.source {
                // Show Rust source expression
                for line in source.lines() {
                    let expanded = expand_tabs(line);
                    let wrapped = wrap_text(&expanded, CONTENT_WIDTH - 4);
                    for wrap_line in wrapped {
                        report.push_str(&box_line(&wrap_line, "    "));
                    }
                }
                // Blank line separating source from SQL/params/error
                report.push_str(&empty_box_line());
                // Show generated SQL on next line with arrow
                let sql_display = format!("→ {}", stmt.sql);
                for line in sql_display.lines() {
                    let expanded = expand_tabs(line);
                    let wrapped = wrap_text(&expanded, CONTENT_WIDTH - 4);
                    for wrap_line in wrapped {
                        report.push_str(&box_line(&wrap_line, "    "));
                    }
                }
                if let Some(params) = &stmt.params {
                    let params_display = format!("Params: {}", params);
                    let wrapped = wrap_text(&params_display, CONTENT_WIDTH - 4);
                    for wrap_line in wrapped {
                        report.push_str(&box_line(&wrap_line, "    "));
                    }
                }
                if let Some(err) = &stmt.error {
                    let err_display = format!("Error: {}", err);
                    let wrapped = wrap_text(&err_display, CONTENT_WIDTH - 4);
                    for wrap_line in wrapped {
                        report.push_str(&box_line(&wrap_line, "    "));
                    }
                }
            } else {
                // No source — show SQL directly (DDL or old pattern)
                for line in stmt.sql.lines() {
                    let expanded = expand_tabs(line);
                    let wrapped = wrap_text(&expanded, CONTENT_WIDTH - 4);
                    for wrap_line in wrapped {
                        report.push_str(&box_line(&wrap_line, "    "));
                    }
                }
                if let Some(err) = &stmt.error {
                    let err_display = format!("Error: {}", err);
                    let wrapped = wrap_text(&err_display, CONTENT_WIDTH - 4);
                    for wrap_line in wrapped {
                        report.push_str(&box_line(&wrap_line, "    "));
                    }
                }
            }

            if i < statements.len() - 1 {
                report.push_str(&empty_box_line());
            }
        }
    }
    report.push_str(&bottom_border());
    report.push('\n');

    report
}

/// Test database wrapper that captures execution context for failure reports
pub mod test_db {
    use super::{CapturedStatement, FailureContext, failure_report};
    use std::ops::{Deref, DerefMut};

    use std::path::PathBuf;
    use std::sync::{Arc, Mutex};

    /// Generic test database wrapper
    pub struct TestDb<D> {
        pub db: D,
        pub driver_name: String,
        pub schema_ddl: Vec<String>,
        // `Arc<Mutex<_>>` (not `RefCell`) so the statements trail can be shared
        // into the `set_hook` panic closure, which requires `Send + Sync`.
        pub statements: Arc<Mutex<Vec<CapturedStatement>>>,
        pub db_path: Option<PathBuf>,
    }

    impl<D> Deref for TestDb<D> {
        type Target = D;
        fn deref(&self) -> &Self::Target {
            &self.db
        }
    }

    impl<D> DerefMut for TestDb<D> {
        fn deref_mut(&mut self) -> &mut Self::Target {
            &mut self.db
        }
    }

    impl<D> Drop for TestDb<D> {
        fn drop(&mut self) {
            if let Some(path) = &self.db_path {
                let _ = std::fs::remove_file(path);
                // Clean up WAL/SHM sidecar files
                let path_str = path.to_string_lossy();
                let _ = std::fs::remove_file(format!("{}-wal", path_str));
                let _ = std::fs::remove_file(format!("{}-shm", path_str));
            }
        }
    }

    impl<D> TestDb<D> {
        pub fn new(db: D, driver_name: impl Into<String>, schema_ddl: Vec<String>) -> Self {
            Self {
                db,
                driver_name: driver_name.into(),
                schema_ddl,
                statements: Arc::new(Mutex::new(Vec::new())),
                db_path: None,
            }
        }

        pub fn with_db_path(mut self, path: PathBuf) -> Self {
            self.db_path = Some(path);
            self
        }

        /// Record a SQL statement execution
        pub fn record(&self, sql: impl Into<String>, error: Option<String>) {
            self.statements.lock().unwrap().push(CapturedStatement {
                sql: sql.into(),
                params: None,
                source: None,
                error,
            });
        }

        /// Record a SQL statement with source expression and params
        pub fn record_sql(&self, source: &str, sql: &str, params: &str, error: Option<String>) {
            self.statements.lock().unwrap().push(CapturedStatement {
                sql: sql.into(),
                params: Some(params.into()),
                source: Some(source.into()),
                error,
            });
        }

        /// Generate a failure report
        pub fn report(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            expected: Option<&str>,
            actual: Option<&str>,
            failed_operation: Option<&str>,
        ) -> String {
            let stmts = self.statements.lock().unwrap();
            failure_report(&FailureContext {
                driver_name: &self.driver_name,
                test_name,
                error,
                expected,
                actual,
                failed_operation,
                schema_ddl: &self.schema_ddl,
                statements: &stmts,
            })
        }

        /// Panic with a formatted failure report
        pub fn fail(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            expected: Option<&str>,
            actual: Option<&str>,
        ) -> ! {
            panic!("{}", self.report(test_name, error, expected, actual, None));
        }

        /// Panic with a formatted failure report including the failed operation
        pub fn fail_with_op(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            failed_operation: &str,
        ) -> ! {
            panic!(
                "{}",
                self.report(test_name, error, None, None, Some(failed_operation))
            );
        }
    }
}

// =============================================================================
// Driver-specific setup modules
// =============================================================================

#[cfg(feature = "rusqlite")]
pub mod rusqlite_setup {
    use super::temp_db_path;
    use super::test_db::TestDb;
    use drizzle::sqlite::rusqlite::Drizzle;
    use drizzle_migrations::{Migration, Tracking};
    use rusqlite::Connection;

    pub fn setup_empty() -> TestDb<Drizzle<()>> {
        let db_path = temp_db_path();
        let conn = Connection::open(&db_path).expect("Failed to create database");
        conn.execute_batch("PRAGMA foreign_keys = ON")
            .expect("Failed to enable foreign keys");
        let (db, _) = Drizzle::new(conn, ());
        TestDb::new(db, "rusqlite", Vec::new()).with_db_path(db_path)
    }

    pub fn setup_empty_db<S: Copy + drizzle::core::SQLSchemaImpl>(
        schema: S,
    ) -> (TestDb<Drizzle<S>>, S) {
        let db_path = temp_db_path();
        let conn = Connection::open(&db_path).expect("Failed to create database");
        conn.execute_batch("PRAGMA foreign_keys = ON")
            .expect("Failed to enable foreign keys");
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(conn, schema);
        let test_db = TestDb::new(db, "rusqlite", schema_ddl).with_db_path(db_path);
        (test_db, schema)
    }

    pub fn legacy_tracking_columns(conn: &Connection, table: &str) -> Vec<String> {
        let pragma = format!("SELECT name FROM pragma_table_info('{table}') ORDER BY cid");
        let mut stmt = conn.prepare(&pragma).expect("prepare pragma_table_info");
        stmt.query_map([], |row| row.get::<_, String>(0))
            .expect("query pragma_table_info")
            .collect::<Result<Vec<_>, _>>()
            .expect("collect pragma columns")
    }

    pub fn create_legacy_tracking_table(conn: &Connection, table: &str) {
        conn.execute(
            &format!(
                "CREATE TABLE \"{table}\" (id INTEGER PRIMARY KEY AUTOINCREMENT, hash text NOT NULL, created_at numeric)"
            ),
            [],
        )
        .expect("create legacy tracking table");
    }

    pub fn table_exists(conn: &Connection, table: &str) -> i64 {
        conn.query_row(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name = ?1",
            [table],
            |row| row.get(0),
        )
        .expect("query sqlite_master")
    }

    pub fn setup_db<S: Default + drizzle::core::SQLSchemaImpl + Copy>() -> (TestDb<Drizzle<S>>, S) {
        let db_path = temp_db_path();
        let conn = Connection::open(&db_path).expect("Failed to create database");
        conn.execute_batch("PRAGMA foreign_keys = ON")
            .expect("Failed to enable foreign keys");
        let schema = S::default();
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(conn, schema);
        let migrations = vec![Migration::with_hash(
            "0000_schema_init",
            "schema_init",
            0,
            schema_ddl.clone(),
        )];

        if let Err(e) = db.migrate(&migrations, Tracking::SQLITE) {
            let test_db = TestDb::new(db, "rusqlite", schema_ddl).with_db_path(db_path);
            test_db.fail(
                "schema_creation",
                &e,
                Some("Schema created successfully"),
                None,
            );
        }

        let test_db = TestDb::new(db, "rusqlite", schema_ddl).with_db_path(db_path);
        (test_db, schema)
    }
}

#[cfg(feature = "libsql")]
pub mod libsql_setup {
    use super::temp_db_path;
    use super::test_db::TestDb;
    use drizzle::sqlite::libsql::Drizzle;
    use drizzle_migrations::{Migration, Tracking};
    use libsql::Builder;

    pub async fn setup_empty() -> TestDb<Drizzle<()>> {
        let db_path = temp_db_path();
        let db_path_str = db_path
            .to_str()
            .expect("temporary sqlite path must be valid UTF-8");
        let db = Builder::new_local(db_path_str)
            .build()
            .await
            .expect("build db");
        let conn = db.connect().expect("connect to db");
        conn.execute("PRAGMA foreign_keys = ON", libsql::params![])
            .await
            .expect("Failed to enable foreign keys");
        let (db, _) = Drizzle::new(conn, ());
        TestDb::new(db, "libsql", Vec::new()).with_db_path(db_path)
    }

    pub async fn setup_empty_db<S: Copy + drizzle::core::SQLSchemaImpl>(
        schema: S,
    ) -> (TestDb<Drizzle<S>>, S) {
        let db_path = temp_db_path();
        let db_path_str = db_path
            .to_str()
            .expect("temporary sqlite path must be valid UTF-8");
        let db = Builder::new_local(db_path_str)
            .build()
            .await
            .expect("build db");
        let conn = db.connect().expect("connect to db");
        conn.execute("PRAGMA foreign_keys = ON", libsql::params![])
            .await
            .expect("Failed to enable foreign keys");
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(conn, schema);
        let test_db = TestDb::new(db, "libsql", schema_ddl).with_db_path(db_path);
        (test_db, schema)
    }

    pub async fn legacy_tracking_columns(conn: &libsql::Connection, table: &str) -> Vec<String> {
        let pragma = format!("SELECT name FROM pragma_table_info('{table}') ORDER BY cid");
        let mut rows = conn
            .query(&pragma, ())
            .await
            .expect("query pragma_table_info");
        let mut columns = Vec::new();
        while let Some(row) = rows.next().await.expect("next pragma row") {
            columns.push(row.get::<String>(0).expect("pragma column name"));
        }
        columns
    }

    pub async fn create_legacy_tracking_table(conn: &libsql::Connection, table: &str) {
        conn.execute(
            &format!(
                "CREATE TABLE \"{table}\" (id INTEGER PRIMARY KEY AUTOINCREMENT, hash text NOT NULL, created_at numeric)"
            ),
            (),
        )
        .await
        .expect("create legacy tracking table");
    }

    pub async fn table_exists(conn: &libsql::Connection, table: &str) -> i64 {
        let mut rows = conn
            .query(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name = ?1",
                libsql::params![table],
            )
            .await
            .expect("query sqlite_master");
        let row = rows
            .next()
            .await
            .expect("next sqlite_master row")
            .expect("sqlite_master row");
        row.get::<i64>(0).expect("sqlite_master count")
    }

    pub async fn setup_db<S: Default + drizzle::core::SQLSchemaImpl + Copy>()
    -> (TestDb<Drizzle<S>>, S) {
        let db_path = temp_db_path();
        let db_path_str = db_path
            .to_str()
            .expect("temporary sqlite path must be valid UTF-8");
        let db = Builder::new_local(db_path_str)
            .build()
            .await
            .expect("build db");
        let conn = db.connect().expect("connect to db");
        conn.execute("PRAGMA foreign_keys = ON", libsql::params![])
            .await
            .expect("Failed to enable foreign keys");
        let schema = S::default();
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(conn, schema);
        let migrations = vec![Migration::with_hash(
            "0000_schema_init",
            "schema_init",
            0,
            schema_ddl.clone(),
        )];

        if let Err(e) = db.migrate(&migrations, Tracking::SQLITE).await {
            let test_db = TestDb::new(db, "libsql", schema_ddl).with_db_path(db_path);
            test_db.fail(
                "schema_creation",
                &e,
                Some("Schema created successfully"),
                None,
            );
        }

        let test_db = TestDb::new(db, "libsql", schema_ddl).with_db_path(db_path);
        (test_db, schema)
    }
}

#[cfg(feature = "turso")]
pub mod turso_setup {
    use super::temp_db_path;
    use super::test_db::TestDb;
    use drizzle::sqlite::turso::Drizzle;
    use drizzle_migrations::{Migration, Tracking};
    use turso::Builder;

    pub async fn setup_empty() -> TestDb<Drizzle<()>> {
        let db_path = temp_db_path();
        let db_path_str = db_path
            .to_str()
            .expect("temporary sqlite path must be valid UTF-8");
        let db = Builder::new_local(db_path_str)
            .build()
            .await
            .expect("build db");
        let conn = db.connect().expect("connect to db");
        conn.execute("PRAGMA foreign_keys = ON", turso::params![])
            .await
            .expect("Failed to enable foreign keys");
        let (db, _) = Drizzle::new(conn, ());
        TestDb::new(db, "turso", Vec::new()).with_db_path(db_path)
    }

    pub async fn setup_empty_db<S: Copy + drizzle::core::SQLSchemaImpl>(
        schema: S,
    ) -> (TestDb<Drizzle<S>>, S) {
        let db_path = temp_db_path();
        let db_path_str = db_path
            .to_str()
            .expect("temporary sqlite path must be valid UTF-8");
        let db = Builder::new_local(db_path_str)
            .build()
            .await
            .expect("build db");
        let conn = db.connect().expect("connect to db");
        conn.execute("PRAGMA foreign_keys = ON", turso::params![])
            .await
            .expect("Failed to enable foreign keys");
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(conn, schema);
        let test_db = TestDb::new(db, "turso", schema_ddl).with_db_path(db_path);
        (test_db, schema)
    }

    pub async fn legacy_tracking_columns(conn: &turso::Connection, table: &str) -> Vec<String> {
        let pragma = format!("SELECT name FROM pragma_table_info('{table}') ORDER BY cid");
        let mut rows = conn
            .query(&pragma, ())
            .await
            .expect("query pragma_table_info");
        let mut columns = Vec::new();
        while let Some(row) = rows.next().await.expect("next pragma row") {
            columns.push(row.get::<String>(0).expect("pragma column name"));
        }
        columns
    }

    pub async fn create_legacy_tracking_table(conn: &turso::Connection, table: &str) {
        conn.execute(
            &format!(
                "CREATE TABLE \"{table}\" (id INTEGER PRIMARY KEY AUTOINCREMENT, hash text NOT NULL, created_at numeric)"
            ),
            (),
        )
        .await
        .expect("create legacy tracking table");
    }

    pub async fn table_exists(conn: &turso::Connection, table: &str) -> i64 {
        let mut rows = conn
            .query(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name = ?1",
                turso::params![table],
            )
            .await
            .expect("query sqlite_master");
        let row = rows
            .next()
            .await
            .expect("next sqlite_master row")
            .expect("sqlite_master row");
        row.get::<i64>(0).expect("sqlite_master count")
    }

    pub async fn setup_db<S: Default + drizzle::core::SQLSchemaImpl + Copy>()
    -> (TestDb<Drizzle<S>>, S) {
        let db_path = temp_db_path();
        let db_path_str = db_path
            .to_str()
            .expect("temporary sqlite path must be valid UTF-8");
        let db = Builder::new_local(db_path_str)
            .build()
            .await
            .expect("build db");
        let conn = db.connect().expect("connect to db");
        conn.execute("PRAGMA foreign_keys = ON", turso::params![])
            .await
            .expect("Failed to enable foreign keys");
        let schema = S::default();
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (mut db, schema) = Drizzle::new(conn, schema);
        let migrations = vec![Migration::with_hash(
            "0000_schema_init",
            "schema_init",
            0,
            schema_ddl.clone(),
        )];

        if let Err(e) = db.migrate(&migrations, Tracking::SQLITE).await {
            let test_db = TestDb::new(db, "turso", schema_ddl).with_db_path(db_path);
            test_db.fail(
                "schema_creation",
                &e,
                Some("Schema created successfully"),
                None,
            );
        }

        let test_db = TestDb::new(db, "turso", schema_ddl).with_db_path(db_path);
        (test_db, schema)
    }
}

#[cfg(feature = "postgres-sync")]
pub mod postgres_sync_setup {
    use super::{CapturedStatement, FailureContext, failure_report};
    use drizzle::postgres::sync::Drizzle;
    use drizzle_migrations::{Migration, Tracking};
    use postgres::{Client, NoTls};
    use std::ops::{Deref, DerefMut};
    use std::process::Command;
    use std::sync::Once;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::{Arc, Mutex};
    use std::thread;
    use std::time::Duration;

    static DOCKER_STARTED: Once = Once::new();
    static TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    fn get_database_url() -> String {
        std::env::var("DATABASE_URL").unwrap_or_else(|_| {
            "host=localhost user=postgres password=postgres dbname=drizzle_test".to_string()
        })
    }

    fn ensure_postgres_running() {
        DOCKER_STARTED.call_once(|| {
            let database_url = get_database_url();

            // Try to connect first
            if Client::connect(&database_url, NoTls).is_ok() {
                println!("PostgreSQL already running");
                return;
            }

            println!("Starting PostgreSQL via Docker Compose...");

            // Start docker compose
            let status = Command::new("docker")
                .args(["compose", "up", "-d", "postgres"])
                .status();

            match status {
                Ok(s) if s.success() => {
                    // Wait for PostgreSQL to be ready
                    println!("Waiting for PostgreSQL to be ready...");
                    for i in 0..30 {
                        thread::sleep(Duration::from_secs(1));
                        if Client::connect(&database_url, NoTls).is_ok() {
                            println!("PostgreSQL is ready! (took {}s)", i + 1);
                            return;
                        }
                    }
                    panic!("PostgreSQL failed to start within 30 seconds");
                }
                Ok(_) => {
                    eprintln!("Docker Compose failed. Make sure Docker is running.");
                    eprintln!("You can manually start with: docker compose up -d postgres");
                }
                Err(e) => {
                    eprintln!("Could not run docker compose: {}", e);
                    eprintln!("Make sure Docker is installed and running.");
                }
            }
        });
    }

    /// Generate a unique schema name for this test
    fn generate_schema_name() -> String {
        let counter = TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
        let thread_id = format!("{:?}", thread::current().id());
        // Extract just the number from ThreadId(X)
        let thread_num: String = thread_id.chars().filter(|c| c.is_ascii_digit()).collect();
        format!("test_{}_{}", thread_num, counter)
    }

    /// Wrapper around Drizzle that automatically cleans up its schema on drop.
    pub struct TestDb<S> {
        pub db: Drizzle<S>,
        schema_name: String,
        schema_ddl: Vec<String>,
        // `Arc<Mutex<_>>` (not `RefCell`) so the statements trail can be shared
        // into the `set_hook` panic closure, which requires `Send + Sync`.
        pub statements: Arc<Mutex<Vec<CapturedStatement>>>,
    }

    impl<S> Deref for TestDb<S> {
        type Target = Drizzle<S>;
        fn deref(&self) -> &Self::Target {
            &self.db
        }
    }

    impl<S> DerefMut for TestDb<S> {
        fn deref_mut(&mut self) -> &mut Self::Target {
            &mut self.db
        }
    }

    impl<S> TestDb<S> {
        pub fn schema_name(&self) -> &str {
            &self.schema_name
        }

        pub fn record(&self, sql: impl Into<String>, error: Option<String>) {
            self.statements.lock().unwrap().push(CapturedStatement {
                sql: sql.into(),
                params: None,
                source: None,
                error,
            });
        }

        pub fn record_sql(&self, source: &str, sql: &str, params: &str, error: Option<String>) {
            self.statements.lock().unwrap().push(CapturedStatement {
                sql: sql.into(),
                params: Some(params.into()),
                source: Some(source.into()),
                error,
            });
        }

        pub fn report(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            expected: Option<&str>,
            actual: Option<&str>,
            failed_operation: Option<&str>,
        ) -> String {
            let stmts = self.statements.lock().unwrap();
            failure_report(&FailureContext {
                driver_name: "postgres-sync",
                test_name,
                error,
                expected,
                actual,
                failed_operation,
                schema_ddl: &self.schema_ddl,
                statements: &stmts,
            })
        }

        pub fn fail(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            expected: Option<&str>,
            actual: Option<&str>,
        ) -> ! {
            panic!("{}", self.report(test_name, error, expected, actual, None));
        }

        pub fn fail_with_op(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            failed_operation: &str,
        ) -> ! {
            panic!(
                "{}",
                self.report(test_name, error, None, None, Some(failed_operation))
            );
        }
    }

    impl<S> Drop for TestDb<S> {
        fn drop(&mut self) {
            // Open a new connection to drop the schema (original is owned by Drizzle)
            if let Ok(mut client) = Client::connect(&get_database_url(), NoTls) {
                let drop_sql = format!("DROP SCHEMA IF EXISTS \"{}\" CASCADE", self.schema_name);
                if let Err(e) = client.batch_execute(&drop_sql) {
                    eprintln!("Failed to drop test schema {}: {}", self.schema_name, e);
                }
            }
        }
    }

    pub fn setup_empty_named(schema_name: impl Into<String>) -> TestDb<()> {
        ensure_postgres_running();

        let database_url = get_database_url();
        let schema_name = schema_name.into();

        let mut client =
            Client::connect(&database_url, NoTls).expect("Failed to connect to PostgreSQL");
        let setup_sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE; CREATE SCHEMA \"{}\"",
            schema_name, schema_name
        );
        client
            .batch_execute(&setup_sql)
            .expect("Failed to create test schema");

        let (db, _) = Drizzle::new(client, ());
        TestDb {
            db,
            schema_name,
            schema_ddl: Vec::new(),
            statements: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn setup_empty_named_db<S: Copy + drizzle::core::SQLSchemaImpl>(
        schema_name: impl Into<String>,
        schema: S,
    ) -> (TestDb<S>, S) {
        ensure_postgres_running();

        let database_url = get_database_url();
        let schema_name = schema_name.into();

        let mut client =
            Client::connect(&database_url, NoTls).expect("Failed to connect to PostgreSQL");
        let setup_sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE; CREATE SCHEMA \"{}\"",
            schema_name, schema_name
        );
        client
            .batch_execute(&setup_sql)
            .expect("Failed to create test schema");

        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(client, schema);
        let test_db = TestDb {
            db,
            schema_name,
            schema_ddl,
            statements: Arc::new(Mutex::new(Vec::new())),
        };
        (test_db, schema)
    }

    pub fn legacy_tracking_columns(client: &mut Client, schema: &str, table: &str) -> Vec<String> {
        client
            .query(
                "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position",
                &[&schema, &table],
            )
            .expect("query information_schema.columns")
            .into_iter()
            .map(|row| row.get::<_, String>(0))
            .collect()
    }

    pub fn create_legacy_tracking_table(client: &mut Client, schema: &str, table: &str) {
        client
            .batch_execute(&format!(
                "CREATE TABLE \"{schema}\".\"{table}\" (id SERIAL PRIMARY KEY, hash TEXT NOT NULL, created_at BIGINT)"
            ))
            .expect("create legacy tracking table");
    }

    pub fn table_exists(client: &mut Client, schema: &str, table: &str) -> i64 {
        client
            .query_one(
                "SELECT COUNT(*)::bigint FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2",
                &[&schema, &table],
            )
            .expect("query information_schema.tables")
            .get(0)
    }

    pub fn setup_db<S: Default + drizzle::core::SQLSchemaImpl + Copy>() -> (TestDb<S>, S) {
        // Ensure PostgreSQL is running (auto-starts via Docker if needed)
        ensure_postgres_running();

        let database_url = get_database_url();
        let schema_name = generate_schema_name();

        let mut client =
            Client::connect(&database_url, NoTls).expect("Failed to connect to PostgreSQL");

        // Create isolated schema for this test and set search_path
        let setup_sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE; CREATE SCHEMA \"{}\"; SET search_path TO \"{}\"",
            schema_name, schema_name, schema_name
        );
        client
            .batch_execute(&setup_sql)
            .expect("Failed to create test schema");

        let schema = S::default();
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (mut db, schema) = Drizzle::new(client, schema);

        let migrations = vec![Migration::with_hash(
            "0000_schema_init",
            "schema_init",
            0,
            schema_ddl.clone(),
        )];
        let config = Tracking::POSTGRES.schema(schema_name.clone());

        if let Err(e) = db.migrate(&migrations, config) {
            let test_db = TestDb {
                db,
                schema_name,
                schema_ddl,
                statements: Arc::new(Mutex::new(Vec::new())),
            };
            test_db.fail(
                "schema_creation",
                &e,
                Some("Schema created successfully"),
                None,
            );
        }

        let test_db = TestDb {
            db,
            schema_name,
            schema_ddl,
            statements: Arc::new(Mutex::new(Vec::new())),
        };
        (test_db, schema)
    }
}

#[cfg(feature = "tokio-postgres")]
pub mod tokio_postgres_setup {
    use super::{CapturedStatement, FailureContext, failure_report};
    use drizzle::postgres::tokio::Drizzle;
    use drizzle_migrations::{Migration, Tracking};
    use std::ops::{Deref, DerefMut};
    use std::process::Command;
    use std::sync::Once;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::{Arc, Mutex};
    use std::thread;
    use std::time::Duration;
    use tokio_postgres::NoTls;

    static DOCKER_STARTED: Once = Once::new();
    static TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    fn get_database_url() -> String {
        std::env::var("DATABASE_URL").unwrap_or_else(|_| {
            "host=localhost user=postgres password=postgres dbname=drizzle_test".to_string()
        })
    }

    /// Check if postgres is reachable (runs on a separate thread with its own runtime)
    fn check_postgres_available(database_url: &str) -> bool {
        let url = database_url.to_string();
        thread::spawn(move || {
            let rt = match tokio::runtime::Runtime::new() {
                Ok(rt) => rt,
                Err(_) => return false,
            };
            rt.block_on(async move { tokio_postgres::connect(&url, NoTls).await.is_ok() })
        })
        .join()
        .unwrap_or(false)
    }

    fn ensure_postgres_running() {
        DOCKER_STARTED.call_once(|| {
            let database_url = get_database_url();

            // Try to connect using tokio-postgres on separate thread
            if check_postgres_available(&database_url) {
                println!("PostgreSQL already running");
                return;
            }

            println!("Starting PostgreSQL via Docker Compose...");

            let status = Command::new("docker")
                .args(["compose", "up", "-d", "postgres"])
                .status();

            match status {
                Ok(s) if s.success() => {
                    println!("Waiting for PostgreSQL to be ready...");
                    for i in 0..30 {
                        thread::sleep(Duration::from_secs(1));
                        if check_postgres_available(&database_url) {
                            println!("PostgreSQL is ready! (took {}s)", i + 1);
                            return;
                        }
                    }
                    panic!("PostgreSQL failed to start within 30 seconds");
                }
                Ok(_) => {
                    eprintln!("Docker Compose failed. Make sure Docker is running.");
                    eprintln!("You can manually start with: docker compose up -d postgres");
                }
                Err(e) => {
                    eprintln!("Could not run docker compose: {}", e);
                    eprintln!("Make sure Docker is installed and running.");
                }
            }
        });
    }

    fn generate_schema_name() -> String {
        let counter = TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
        let thread_id = format!("{:?}", thread::current().id());
        let thread_num: String = thread_id.chars().filter(|c| c.is_ascii_digit()).collect();
        format!("test_async_{}_{}", thread_num, counter)
    }

    /// Wrapper around Drizzle that automatically cleans up its schema on drop.
    pub struct TestDb<S> {
        pub db: Drizzle<S>,
        schema_name: String,
        schema_ddl: Vec<String>,
        // `Arc<Mutex<_>>` (not `RefCell`) so the statements trail can be shared
        // into the `set_hook` panic closure, which requires `Send + Sync`.
        pub statements: Arc<Mutex<Vec<CapturedStatement>>>,
    }

    impl<S> Deref for TestDb<S> {
        type Target = Drizzle<S>;
        fn deref(&self) -> &Self::Target {
            &self.db
        }
    }

    impl<S> DerefMut for TestDb<S> {
        fn deref_mut(&mut self) -> &mut Self::Target {
            &mut self.db
        }
    }

    impl<S> TestDb<S> {
        pub fn schema_name(&self) -> &str {
            &self.schema_name
        }

        pub fn record(&self, sql: impl Into<String>, error: Option<String>) {
            self.statements.lock().unwrap().push(CapturedStatement {
                sql: sql.into(),
                params: None,
                source: None,
                error,
            });
        }

        pub fn record_sql(&self, source: &str, sql: &str, params: &str, error: Option<String>) {
            self.statements.lock().unwrap().push(CapturedStatement {
                sql: sql.into(),
                params: Some(params.into()),
                source: Some(source.into()),
                error,
            });
        }

        pub fn report(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            expected: Option<&str>,
            actual: Option<&str>,
            failed_operation: Option<&str>,
        ) -> String {
            let stmts = self.statements.lock().unwrap();
            failure_report(&FailureContext {
                driver_name: "tokio-postgres",
                test_name,
                error,
                expected,
                actual,
                failed_operation,
                schema_ddl: &self.schema_ddl,
                statements: &stmts,
            })
        }

        pub fn fail(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            expected: Option<&str>,
            actual: Option<&str>,
        ) -> ! {
            panic!("{}", self.report(test_name, error, expected, actual, None));
        }

        pub fn fail_with_op(
            &self,
            test_name: &str,
            error: &dyn std::fmt::Display,
            failed_operation: &str,
        ) -> ! {
            panic!(
                "{}",
                self.report(test_name, error, None, None, Some(failed_operation))
            );
        }
    }

    impl<S> Drop for TestDb<S> {
        fn drop(&mut self) {
            let schema_name = self.schema_name.clone();
            let database_url = get_database_url();

            // Spawn a thread with its own tokio runtime for async cleanup
            let _ = thread::spawn(move || {
                let rt = tokio::runtime::Runtime::new().expect("Failed to create cleanup runtime");
                rt.block_on(async move {
                    if let Ok((client, connection)) =
                        tokio_postgres::connect(&database_url, NoTls).await
                    {
                        // Spawn connection handler (fire and forget)
                        tokio::spawn(async move {
                            let _ = connection.await;
                        });

                        let drop_sql = format!("DROP SCHEMA IF EXISTS \"{}\" CASCADE", schema_name);
                        if let Err(e) = client.batch_execute(&drop_sql).await {
                            eprintln!("Failed to drop test schema {}: {}", schema_name, e);
                        }
                    }
                });
            })
            .join();
        }
    }

    pub async fn setup_empty_named(schema_name: impl Into<String>) -> TestDb<()> {
        ensure_postgres_running();

        let database_url = get_database_url();
        let schema_name = schema_name.into();

        let (client, connection) = tokio_postgres::connect(&database_url, NoTls)
            .await
            .expect("Failed to connect to PostgreSQL");

        tokio::spawn(async move {
            if let Err(e) = connection.await {
                eprintln!("PostgreSQL connection error: {}", e);
            }
        });

        let setup_sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE; CREATE SCHEMA \"{}\"",
            schema_name, schema_name
        );
        client
            .batch_execute(&setup_sql)
            .await
            .expect("Failed to create test schema");

        let (db, _) = Drizzle::new(client, ());
        TestDb {
            db,
            schema_name,
            schema_ddl: Vec::new(),
            statements: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub async fn setup_empty_named_db<S: Copy + drizzle::core::SQLSchemaImpl>(
        schema_name: impl Into<String>,
        schema: S,
    ) -> (TestDb<S>, S) {
        ensure_postgres_running();

        let database_url = get_database_url();
        let schema_name = schema_name.into();

        let (client, connection) = tokio_postgres::connect(&database_url, NoTls)
            .await
            .expect("Failed to connect to PostgreSQL");

        tokio::spawn(async move {
            if let Err(e) = connection.await {
                eprintln!("PostgreSQL connection error: {}", e);
            }
        });

        let setup_sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE; CREATE SCHEMA \"{}\"",
            schema_name, schema_name
        );
        client
            .batch_execute(&setup_sql)
            .await
            .expect("Failed to create test schema");

        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (db, schema) = Drizzle::new(client, schema);
        let test_db = TestDb {
            db,
            schema_name,
            schema_ddl,
            statements: Arc::new(Mutex::new(Vec::new())),
        };
        (test_db, schema)
    }

    pub async fn legacy_tracking_columns(
        client: &tokio_postgres::Client,
        schema: &str,
        table: &str,
    ) -> Vec<String> {
        client
            .query(
                "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position",
                &[&schema, &table],
            )
            .await
            .expect("query information_schema.columns")
            .into_iter()
            .map(|row| row.get::<_, String>(0))
            .collect()
    }

    pub async fn create_legacy_tracking_table(
        client: &tokio_postgres::Client,
        schema: &str,
        table: &str,
    ) {
        client
            .batch_execute(&format!(
                "CREATE TABLE \"{schema}\".\"{table}\" (id SERIAL PRIMARY KEY, hash TEXT NOT NULL, created_at BIGINT)"
            ))
            .await
            .expect("create legacy tracking table");
    }

    pub async fn table_exists(client: &tokio_postgres::Client, schema: &str, table: &str) -> i64 {
        client
            .query_one(
                "SELECT COUNT(*)::bigint FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2",
                &[&schema, &table],
            )
            .await
            .expect("query information_schema.tables")
            .get(0)
    }

    pub async fn setup_db<S: Default + drizzle::core::SQLSchemaImpl + Copy>() -> (TestDb<S>, S) {
        // Ensure PostgreSQL is running (auto-starts via Docker if needed)
        ensure_postgres_running();

        let database_url = get_database_url();
        let schema_name = generate_schema_name();

        // Connect using tokio-postgres
        let (client, connection) = tokio_postgres::connect(&database_url, NoTls)
            .await
            .expect("Failed to connect to PostgreSQL");

        // Spawn the connection handler
        tokio::spawn(async move {
            if let Err(e) = connection.await {
                eprintln!("PostgreSQL connection error: {}", e);
            }
        });

        // Create isolated schema for this test and set search_path
        let setup_sql = format!(
            "DROP SCHEMA IF EXISTS \"{}\" CASCADE; CREATE SCHEMA \"{}\"; SET search_path TO \"{}\"",
            schema_name, schema_name, schema_name
        );
        client
            .batch_execute(&setup_sql)
            .await
            .expect("Failed to create test schema");

        let schema = S::default();
        let schema_ddl: Vec<_> = schema
            .create_statements()
            .expect("create statements")
            .collect();
        let (mut db, schema) = Drizzle::new(client, schema);
        let migrations = vec![Migration::with_hash(
            "0000_schema_init",
            "schema_init",
            0,
            schema_ddl.clone(),
        )];
        let config = Tracking::POSTGRES.schema(schema_name.clone());

        if let Err(e) = db.migrate(&migrations, config).await {
            let test_db = TestDb {
                db,
                schema_name,
                schema_ddl,
                statements: Arc::new(Mutex::new(Vec::new())),
            };
            test_db.fail(
                "schema_creation",
                &e,
                Some("Schema created successfully"),
                None,
            );
        }

        let test_db = TestDb {
            db,
            schema_name,
            schema_ddl,
            statements: Arc::new(Mutex::new(Vec::new())),
        };
        (test_db, schema)
    }
}