use crate::Sealed;
use std::cell::Cell;
use std::fmt;
use std::sync::{Arc, Mutex};
use std::time::Instant;

// ─── CriticalError ───

/// Internal error type for Critical<T>.
/// Carries operation name and error message.
#[derive(Debug, Clone)]
pub struct CriticalError {
    pub operation: &'static str,
    pub message: String,
}

impl fmt::Display for CriticalError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "[{}] {}", self.operation, self.message)
    }
}

// ─── Token traits ───

/// Unified token trait for concept map generated tokens.
/// Tokens are enum variants — Copy + Debug, carrying operation_name.
pub trait ConceptMapToken: Copy + fmt::Debug {
    fn operation_name(&self) -> &'static str;
}

/// Marker trait for fatal tokens — used with TxScope::propagate.
/// Fatal failures panic inside a transaction, aborting the scope.
pub trait FatalToken: ConceptMapToken {}

/// Marker trait for suppressible tokens — used with suppress/suppress_with.
/// Suppressible failures are logged and swallowed.
pub trait SuppressibleToken: ConceptMapToken {}

// ─── Transaction ───

/// Transaction first-class citizen.
///
/// Each concept map entry generates a Transaction struct with associated
/// Fatal and Suppressible token types. This replaces using raw FatalToken
/// as the TxScope type parameter.
///
/// Implementations are generated by scaffold — manual impl is forbidden (Guardian E302).
pub trait Transaction: Send + Sync + 'static {
    type Fatal: FatalToken + Send + Sync + 'static;
    type Suppressible: SuppressibleToken + Send + Sync + 'static;
    /// Concept map transaction name.
    fn name() -> &'static str;
}

// ─── DbBinding ───

/// Internal database binding for transaction scopes.
///
/// Holds a database handle and a lazily-begun sqlx transaction.
/// The transaction is started on first `scope.db()` call and auto-committed
/// or auto-rolled-back when the scope ends.
pub(crate) struct DbBinding {
    pub(crate) db: crate::db::Db,
    pub(crate) tx: Arc<tokio::sync::Mutex<Option<crate::db::Tx>>>,
}

// ─── TxScope ───

/// Transaction scope — captures the first fatal abort for post-panic inspection.
///
/// Created at the start of a transaction boundary. When `Critical::propagate`
/// fails, it records the abort info here before panicking.
///
/// Abort storage preserves the strong token type `T::Fatal` for exhaustive match
/// in the transaction guard. Sub-transaction scoping uses independent storage
/// with detail forwarding via `propagate()` / `suppress()`.
pub struct TxScope<T: Transaction> {
    abort: Arc<Mutex<Option<(T::Fatal, Sealed<String>)>>>,
    db: Option<DbBinding>,
    trace_id: [u8; 4],
    created_at: Instant,
}

impl<T: Transaction> TxScope<T> {
    /// Create a new transaction scope (no database binding).
    pub fn new() -> Self {
        let uuid = uuid::Uuid::new_v4();
        let bytes = uuid.as_bytes();
        let trace_id = [bytes[0], bytes[1], bytes[2], bytes[3]];
        Self {
            abort: Arc::new(Mutex::new(None)),
            db: None,
            trace_id,
            created_at: Instant::now(),
        }
    }

    /// Create a transaction scope bound to a database.
    ///
    /// The DB transaction is lazily begun on first `scope.db()` call.
    /// Called by framework entry points (tx_guard, run_handler_loop, etc.).
    #[doc(hidden)]
    pub fn new_with_db(db: crate::db::Db) -> Self {
        let uuid = uuid::Uuid::new_v4();
        let bytes = uuid.as_bytes();
        let trace_id = [bytes[0], bytes[1], bytes[2], bytes[3]];
        Self {
            abort: Arc::new(Mutex::new(None)),
            db: Some(DbBinding {
                db,
                tx: Arc::new(tokio::sync::Mutex::new(None)),
            }),
            trace_id,
            created_at: Instant::now(),
        }
    }

    // ─── Trace & Logging ───

    /// Returns the 8-character hex trace ID for this transaction scope.
    pub fn trace_id(&self) -> String {
        format!("{:02x}{:02x}{:02x}{:02x}",
            self.trace_id[0], self.trace_id[1],
            self.trace_id[2], self.trace_id[3])
    }

    /// Returns the raw 4-byte trace ID (for injecting into Critical).
    pub fn raw_trace_id(&self) -> [u8; 4] {
        self.trace_id
    }

    /// Transaction log prefix: `[TX:{name}:{trace_id}]`
    fn tx_prefix(&self) -> String {
        format!("[TX:{}:{}]", T::name(), self.trace_id())
    }

    /// Elapsed milliseconds since scope creation.
    fn elapsed_ms(&self) -> u128 {
        self.created_at.elapsed().as_millis()
    }

    /// Log an info-level message with transaction context.
    ///
    /// Output: `[TX:{tx_name}:{trace_id}] {message}`
    pub fn log(&self, msg: impl fmt::Display) {
        tracing::info!("{} {}", self.tx_prefix(), msg);
    }

    /// Log a warn-level message with transaction context.
    pub fn warn(&self, msg: impl fmt::Display) {
        tracing::warn!("{} {}", self.tx_prefix(), msg);
    }

    /// Log an error-level message with transaction context.
    pub fn error(&self, msg: impl fmt::Display) {
        tracing::error!("{} {}", self.tx_prefix(), msg);
    }

    /// Log a debug-level message with transaction context.
    pub fn debug(&self, msg: impl fmt::Display) {
        tracing::debug!("{} {}", self.tx_prefix(), msg);
    }

    /// Get a scoped database handle bound to this transaction.
    ///
    /// On first call, lazily begins a DB transaction (BEGIN).
    /// Returns `ScopedDb` which provides query methods tied to this scope.
    ///
    /// # Panics
    /// Panics if this scope was not bound to a database (created via `new()` instead of `new_with_db()`).
    /// Panics if the lazy BEGIN fails (database unavailable).
    pub async fn db(&self) -> crate::db::ScopedDb {
        let binding = self.db.as_ref()
            .expect("TxScope::db() called on scope not bound to a database");
        let mut guard = binding.tx.clone().lock_owned().await;
        if guard.is_none() {
            tracing::debug!("{} db begin", self.tx_prefix());
            let tx = binding.db.begin_inner().await
                .expect("TxScope::db() lazy BEGIN failed — database unavailable");
            *guard = Some(tx);
        }
        crate::db::ScopedDb(guard)
    }

    /// Auto-commit the database transaction if one was lazily started.
    ///
    /// Called by framework entry points after handler completes successfully.
    /// On panic (abort), the `Tx` is dropped without commit → auto-rollback by sqlx.
    pub(crate) async fn auto_commit(&self) {
        // Commit primary db
        if let Some(binding) = &self.db {
            let mut guard = binding.tx.lock().await;
            if let Some(tx) = guard.take() {
                if let Err(e) = tx.commit_inner().await {
                    panic!("[TX-COMMIT] auto-commit failed: {}", e);
                }
                tracing::debug!("{} db commit", self.tx_prefix());
            }
        }
    }

    /// Get or create a connection pool for a dynamic data source.
    ///
    /// Public helper for engine code that needs to pass a Db to sub-scopes.
    /// The pool is cached globally — multiple calls with the same (D, param) reuse the pool.
    pub async fn dynamic_pool<D: crate::db::DynamicDao>(param: &str) -> crate::db::Db {
        crate::db::get_or_create_dynamic_pool::<D>(param)
            .await
            .expect("dynamic_pool: failed to get/create pool")
    }
    ///


    /// Run a sub-transaction bound to a dynamic database with suppress semantics.
    ///
    /// Creates an independent sub-scope bound to the resolved database pool.
    /// The sub-scope has full mode-1 experience: `sub.db()` returns a scoped handle,
    /// auto-commit on success, auto-rollback on failure (tx dropped without commit).
    ///
    /// No SAVEPOINT — always an independent transaction (different database).
    pub async fn suppress_with_db<SubTx: Transaction, D: crate::db::DynamicDao, V, Fut>(
        &self,
        parent_token: T::Suppressible,
        reason: &str,
        param: &str,
        f: impl FnOnce(TxScope<SubTx>) -> Fut,
    ) -> Option<V>
    where
        Fut: std::future::Future<Output = V>,
    {
        use futures_util::FutureExt;
        use std::panic::AssertUnwindSafe;

        let db = crate::db::get_or_create_dynamic_pool::<D>(param)
            .await
            .expect("suppress_with_db: failed to get/create dynamic pool");

        let tx_holder = Arc::new(tokio::sync::Mutex::new(None));
        let sub = TxScope::<SubTx> {
            abort: Arc::new(Mutex::new(None)),
            db: Some(DbBinding {
                db,
                tx: tx_holder.clone(),
            }),
            trace_id: self.trace_id,
            created_at: self.created_at,
        };
        let sub_abort = sub.abort.clone();
        tracing::info!("{} suppress_with_db({}) begin (param={})", self.tx_prefix(), SubTx::name(), param);
        let sub_start = std::time::Instant::now();
        match AssertUnwindSafe(f(sub)).catch_unwind().await {
            Ok(v) => {
                // Commit the sub-scope's db transaction
                let mut guard = tx_holder.lock().await;
                if let Some(tx) = guard.take() {
                    if let Err(e) = tx.commit_inner().await {
                        panic!("[TX-COMMIT] suppress_with_db auto-commit failed (param={}): {}", param, e);
                    }
                }
                tracing::info!("{} suppress_with_db({}) ok (param={}) ({}ms)", self.tx_prefix(), SubTx::name(), param, sub_start.elapsed().as_millis());
                Some(v)
            }
            Err(_) => {
                // tx_holder dropped without commit → auto-rollback by sqlx
                if let Some((sub_token, detail)) = sub_abort.lock().unwrap().take() {
                    tracing::warn!(
                        "{} suppress_with_db({}) failed: {} — {}",
                        self.tx_prefix(),
                        SubTx::name(),
                        sub_token.operation_name(),
                        detail,
                    );
                } else {
                    tracing::warn!(
                        "{} suppress_with_db({}) failed: unexpected panic",
                        self.tx_prefix(),
                        SubTx::name(),
                    );
                }
                None
            }
        }
    }

    /// Run a sub-transaction bound to a dynamic database with propagate semantics.
    ///
    /// Creates an independent sub-scope bound to the resolved database pool.
    /// The sub-scope has full mode-1 experience: `sub.db()` returns a scoped handle,
    /// auto-commit on success, abort propagated to parent on failure.
    ///
    /// No SAVEPOINT — always an independent transaction (different database).
    pub async fn propagate_with_db<SubTx: Transaction, D: crate::db::DynamicDao, V, Fut>(
        &self,
        parent_token: T::Fatal,
        param: &str,
        f: impl FnOnce(TxScope<SubTx>) -> Fut,
    ) -> V
    where
        Fut: std::future::Future<Output = V>,
    {
        use futures_util::FutureExt;
        use std::panic::AssertUnwindSafe;

        let db = crate::db::get_or_create_dynamic_pool::<D>(param)
            .await
            .expect("propagate_with_db: failed to get/create dynamic pool");

        let tx_holder = Arc::new(tokio::sync::Mutex::new(None));
        let sub = TxScope::<SubTx> {
            abort: Arc::new(Mutex::new(None)),
            db: Some(DbBinding {
                db,
                tx: tx_holder.clone(),
            }),
            trace_id: self.trace_id,
            created_at: self.created_at,
        };
        let sub_abort = sub.abort.clone();
        tracing::info!("{} propagate_with_db({}) begin (param={})", self.tx_prefix(), SubTx::name(), param);
        let sub_start = std::time::Instant::now();
        match AssertUnwindSafe(f(sub)).catch_unwind().await {
            Ok(v) => {
                // Commit the sub-scope's db transaction
                let mut guard = tx_holder.lock().await;
                if let Some(tx) = guard.take() {
                    if let Err(e) = tx.commit_inner().await {
                        panic!("[TX-COMMIT] propagate_with_db auto-commit failed (param={}): {}", param, e);
                    }
                }
                tracing::info!("{} propagate_with_db({}) ok (param={}) ({}ms)", self.tx_prefix(), SubTx::name(), param, sub_start.elapsed().as_millis());
                v
            }
            Err(payload) => {
                // tx_holder dropped without commit → auto-rollback by sqlx
                let detail = sub_abort.lock().unwrap().take()
                    .map(|(_, d)| d)
                    .unwrap_or_else(|| Sealed::new(
                        "unexpected panic in sub-transaction".to_string()
                    ));
                self.record_abort_detail(parent_token, detail);
                std::panic::resume_unwind(payload);
            }
        }
    }

    /// Record an abort (called by Critical::propagate before panic).
    pub fn record_abort(&self, token: T::Fatal, err: CriticalError) {
        *self.abort.lock().unwrap() = Some((
            token,
            Sealed::new(format!("[{}] {}", err.operation, err.message)),
        ));
    }

    /// Take the recorded abort info (called after catch_unwind).
    ///
    /// Returns `(token, detail_message)` preserving the strong Fatal type
    /// for exhaustive match in transaction guards.
    pub fn take_abort(&self) -> Option<(T::Fatal, Sealed<String>)> {
        self.abort.lock().unwrap().take()
    }

    /// Record an abort with a pre-built detail message (used by sub-transaction forwarding).
    fn record_abort_detail(&self, token: T::Fatal, detail: Sealed<String>) {
        *self.abort.lock().unwrap() = Some((token, detail));
    }

    /// Run a sub-transaction with fatal semantics (async).
    ///
    /// If the parent scope has an active DB transaction, creates a SAVEPOINT
    /// for the sub-transaction. On success, releases the savepoint. On failure,
    /// rolls back to the savepoint and re-panics with parent token.
    ///
    /// If no parent DB transaction, creates an independent sub-scope (no SAVEPOINT).
    pub async fn propagate<SubTx: Transaction, V, Fut>(
        &self,
        parent_token: T::Fatal,
        f: impl FnOnce(TxScope<SubTx>) -> Fut,
    ) -> V
    where
        Fut: std::future::Future<Output = V>,
    {
        use futures_util::FutureExt;
        use std::panic::AssertUnwindSafe;

        // Check if parent has active DB transaction
        let parent_has_tx = if let Some(ref binding) = self.db {
            binding.tx.lock().await.is_some()
        } else {
            false
        };

        if parent_has_tx {
            let binding = self.db.as_ref().unwrap();
            let sp_name = format!("sp_{}", SubTx::name());

            // Take tx from parent, create SAVEPOINT
            let mut tx = binding.tx.lock().await.take().unwrap();
                        tracing::info!("{} propagate({}) begin", self.tx_prefix(), SubTx::name());
            let sub_start = std::time::Instant::now();
if let Err(e) = tx.savepoint(&sp_name).await {
                *binding.tx.lock().await = Some(tx);
                panic!("[SAVEPOINT] failed to create {}: {}", sp_name, e);
            }

            // Create shared tx holder for child scope
            let child_tx = Arc::new(tokio::sync::Mutex::new(Some(tx)));
            let sub = TxScope::<SubTx> {
                abort: Arc::new(Mutex::new(None)),
                db: Some(DbBinding {
                    db: binding.db.clone(),
                    tx: child_tx.clone(),
                }),
                trace_id: self.trace_id,
                created_at: self.created_at,
            };
            let sub_abort = sub.abort.clone();

            match AssertUnwindSafe(f(sub)).catch_unwind().await {
                Ok(v) => {
                    // Success — RELEASE SAVEPOINT, put tx back to parent
                    let mut tx = child_tx.lock().await.take().unwrap();
                    if let Err(e) = tx.release_savepoint(&sp_name).await {
                        *binding.tx.lock().await = Some(tx);
                        panic!("[SAVEPOINT] failed to release {}: {}", sp_name, e);
                    }
                    *binding.tx.lock().await = Some(tx);
                    // Commit sub-scope's dynamic dbs (independent from SAVEPOINT)
                    tracing::info!("{} propagate({}) ok ({}ms)", self.tx_prefix(), SubTx::name(), sub_start.elapsed().as_millis());
                    v
                }
                Err(payload) => {
                    // Failure — ROLLBACK TO SAVEPOINT, put tx back to parent
                    let mut tx = child_tx.lock().await.take().unwrap();
                    let _ = tx.rollback_to_savepoint(&sp_name).await;
                    tracing::debug!("{} propagate({}) savepoint rollback", self.tx_prefix(), SubTx::name());
                    *binding.tx.lock().await = Some(tx);

                    let detail = sub_abort.lock().unwrap().take()
                        .map(|(_, d)| d)
                        .unwrap_or_else(|| Sealed::new(
                            "unexpected panic in sub-transaction".to_string()
                        ));
                    self.record_abort_detail(parent_token, detail);
                    std::panic::resume_unwind(payload);
                }
            }
        } else {
            // No parent DB tx — no SAVEPOINT
            let sub = TxScope::<SubTx> {
                abort: Arc::new(Mutex::new(None)),
                db: None,
                trace_id: self.trace_id,
                created_at: self.created_at,
            };
            let sub_abort = sub.abort.clone();
            tracing::info!("{} propagate({}) begin", self.tx_prefix(), SubTx::name());
            let sub_start = std::time::Instant::now();
            match AssertUnwindSafe(f(sub)).catch_unwind().await {
                Ok(v) => {
                    tracing::info!("{} propagate({}) ok ({}ms)", self.tx_prefix(), SubTx::name(), sub_start.elapsed().as_millis());
                    v
                }
                Err(payload) => {
                    let detail = sub_abort.lock().unwrap().take()
                        .map(|(_, d)| d)
                        .unwrap_or_else(|| Sealed::new(
                            "unexpected panic in sub-transaction".to_string()
                        ));
                    self.record_abort_detail(parent_token, detail);
                    std::panic::resume_unwind(payload);
                }
            }
        }
    }

    /// Run a sub-transaction with suppress semantics (async).
    ///
    /// If the parent scope has an active DB transaction, creates a SAVEPOINT.
    /// On success, releases the savepoint. On failure, rolls back to savepoint
    /// and returns `None`.
    ///
    /// If no parent DB transaction, creates an independent sub-scope (no SAVEPOINT).
    pub async fn suppress<SubTx: Transaction, V, Fut>(
        &self,
        parent_token: T::Suppressible,
        reason: &str,
        f: impl FnOnce(TxScope<SubTx>) -> Fut,
    ) -> Option<V>
    where
        Fut: std::future::Future<Output = V>,
    {
        use futures_util::FutureExt;
        use std::panic::AssertUnwindSafe;

        // Check if parent has active DB transaction
        let parent_has_tx = if let Some(ref binding) = self.db {
            binding.tx.lock().await.is_some()
        } else {
            false
        };

        if parent_has_tx {
            let binding = self.db.as_ref().unwrap();
            let sp_name = format!("sp_{}", SubTx::name());
            tracing::info!("{} suppress({}) begin", self.tx_prefix(), SubTx::name());
            let sub_start = std::time::Instant::now();

            // Take tx from parent, create SAVEPOINT
            let mut tx = binding.tx.lock().await.take().unwrap();
            if let Err(e) = tx.savepoint(&sp_name).await {
                *binding.tx.lock().await = Some(tx);
                tracing::error!("[SAVEPOINT] failed to create {}: {}", sp_name, e);
                return None;
            }

            // Create shared tx holder for child scope
            let child_tx = Arc::new(tokio::sync::Mutex::new(Some(tx)));
            let sub = TxScope::<SubTx> {
                abort: Arc::new(Mutex::new(None)),
                db: Some(DbBinding {
                    db: binding.db.clone(),
                    tx: child_tx.clone(),
                }),
                trace_id: self.trace_id,
                created_at: self.created_at,
            };
            let sub_abort = sub.abort.clone();

            match AssertUnwindSafe(f(sub)).catch_unwind().await {
                Ok(v) => {
                    // Success — RELEASE SAVEPOINT, put tx back
                    let mut tx = child_tx.lock().await.take().unwrap();
                    if let Err(e) = tx.release_savepoint(&sp_name).await {
                        tracing::error!("[SAVEPOINT] failed to release {}: {}", sp_name, e);
                    }
                    *binding.tx.lock().await = Some(tx);
                    // Commit sub-scope's dynamic dbs (independent from SAVEPOINT)
                    tracing::info!("{} suppress({}) ok ({}ms)", self.tx_prefix(), SubTx::name(), sub_start.elapsed().as_millis());
                    Some(v)
                }
                Err(_) => {
                    // Failure — ROLLBACK TO SAVEPOINT, put tx back
                    let mut tx = child_tx.lock().await.take().unwrap();
                    let _ = tx.rollback_to_savepoint(&sp_name).await;
                    *binding.tx.lock().await = Some(tx);
                    tracing::debug!("{} savepoint({}) rollback", self.tx_prefix(), SubTx::name());

                    if let Some((sub_token, detail)) = sub_abort.lock().unwrap().take() {
                        tracing::warn!(
                            "{} suppress({}) failed: {} — {}",
                            self.tx_prefix(),
                            SubTx::name(),
                            sub_token.operation_name(),
                            detail,
                        );
                    } else {
                        tracing::warn!(
                            "{} suppress({}) failed: unexpected panic",
                            self.tx_prefix(),
                            SubTx::name(),
                        );
                    }
                    None
                }
            }
        } else {
            // No parent DB tx — no SAVEPOINT
            let sub = TxScope::<SubTx> {
                abort: Arc::new(Mutex::new(None)),
                db: None,
                trace_id: self.trace_id,
                created_at: self.created_at,
            };
            let sub_abort = sub.abort.clone();
            tracing::info!("{} suppress({}) begin", self.tx_prefix(), SubTx::name());
            let sub_start = std::time::Instant::now();
            match AssertUnwindSafe(f(sub)).catch_unwind().await {
                Ok(v) => {
                    tracing::info!("{} suppress({}) ok ({}ms)", self.tx_prefix(), SubTx::name(), sub_start.elapsed().as_millis());
                    Some(v)
                }
                Err(_) => {
                    if let Some((sub_token, detail)) = sub_abort.lock().unwrap().take() {
                        tracing::warn!(
                            "{} suppress({}) failed: {} — {}",
                            self.tx_prefix(),
                            SubTx::name(),
                            sub_token.operation_name(),
                            detail,
                        );
                    } else {
                        tracing::warn!(
                            "{} suppress({}) failed: unexpected panic",
                            self.tx_prefix(),
                            SubTx::name(),
                        );
                    }
                    None
                }
            }
        }
    }
}

impl<T: Transaction> Default for TxScope<T> {
    fn default() -> Self {
        Self::new()
    }
}

// ─── Core transaction execution ───

/// Transaction execution result — returned by `run_tx_scope`.
pub(crate) enum TxOutcome<V, T: Transaction> {
    /// Handler completed normally. DB transaction auto-committed.
    Completed(V),
    /// Fatal abort: `propagate()` triggered a panic with a known token.
    Aborted(T::Fatal, Sealed<String>),
    /// Non-transactional panic (not from `propagate()`).
    UnexpectedPanic,
}

/// Core transaction execution — **the single source of truth** for:
/// - `catch_unwind` around the handler future
/// - `auto_commit` on success
/// - `take_abort` on panic
///
/// All entry points (tx_guard, tunnel_tx_guard, run_handler_loop, etc.)
/// delegate to this function. Transaction framework logic is written once here.
pub(crate) async fn run_tx_scope<T, V, Fut>(
    scope: &TxScope<T>,
    future: Fut,
) -> TxOutcome<V, T>
where
    T: Transaction,
    Fut: std::future::Future<Output = V>,
{
    use futures_util::FutureExt;
    use std::panic::AssertUnwindSafe;

    tracing::info!("{} begin", scope.tx_prefix());

    let result = AssertUnwindSafe(future)
        .catch_unwind()
        .await;

    match result {
        Ok(value) => {
            scope.auto_commit().await;
            tracing::info!("{} commit ({}ms)", scope.tx_prefix(), scope.elapsed_ms());
            TxOutcome::Completed(value)
        }
        Err(_panic) => {
            match scope.take_abort() {
                Some((token, msg)) => {
                    tracing::error!("{} abort: {}", scope.tx_prefix(), msg);
                    TxOutcome::Aborted(token, msg)
                }
                None => {
                    tracing::error!("{} unexpected panic", scope.tx_prefix());
                    TxOutcome::UnexpectedPanic
                }
            }
        }
    }
}

// ─── Startup transaction ───

/// Result of a startup transaction.
///
/// Startup is a one-shot transaction for initialization phases.
/// The hat creates and manages the `TxScope`; engine never touches `TxScope::new()`.
#[derive(Debug)]
pub(crate) enum StartupResult<V, T: Transaction> {
    /// Transaction completed successfully.
    Ok(V),
    /// Transaction aborted with a known fatal token.
    Aborted(T::Fatal, Sealed<String>),
    /// Transaction panicked without recording a fatal token.
    UnexpectedPanic,
}

/// Run a one-shot startup transaction.
///
/// Creates a `TxScope<T>` internally, passes it to the closure,
/// and catches any panic (fatal abort or unexpected).
///
/// This is the dedicated entry point for initialization-phase transactions.
/// Engine code should use this instead of manually creating `TxScope::new()`.
pub(crate) async fn run_startup<T, V, Fut>(
    f: impl FnOnce(Arc<TxScope<T>>) -> Fut,
) -> StartupResult<V, T>
where
    T: Transaction,
    V: Send,
    Fut: std::future::Future<Output = V> + Send,
{
    let scope = Arc::new(TxScope::<T>::new());
    let scope_clone = scope.clone();
    match run_tx_scope(&scope, f(scope_clone)).await {
        TxOutcome::Completed(v) => StartupResult::Ok(v),
        TxOutcome::Aborted(token, detail) => StartupResult::Aborted(token, detail),
        TxOutcome::UnexpectedPanic => StartupResult::UnexpectedPanic,
    }
}

// ─── Main entry (macro support) ───

/// Internal entry point for `#[mad_hatter::main]` macro expansion.
///
/// **Do not call directly** — use the `#[mad_hatter::main(TxType)]` attribute macro.
///
/// Creates a startup `TxScope`, runs the user function with `catch_unwind`,
/// and exits the process on failure.
#[doc(hidden)]
pub async fn __main_entry<T, Fut>(db: Option<crate::db::Db>, f: impl FnOnce(std::sync::Arc<TxScope<T>>) -> Fut + Send)
where
    T: Transaction,
    T::Fatal: std::fmt::Debug,
    Fut: std::future::Future<Output = ()> + Send,
{
    let scope = match db {
        Some(db) => std::sync::Arc::new(TxScope::<T>::new_with_db(db)),
        None => std::sync::Arc::new(TxScope::<T>::new()),
    };
    let scope_clone = scope.clone();
    match run_tx_scope(&scope, f(scope_clone)).await {
        TxOutcome::Completed(()) => {}
        TxOutcome::Aborted(token, detail) => {
            eprintln!("[FATAL] Startup aborted: {:?} — {}", token, detail);
            std::process::exit(1);
        }
        TxOutcome::UnexpectedPanic => {
            eprintln!("[FATAL] Unexpected panic during startup");
            std::process::exit(1);
        }
    }
}


// ─── CriticalGuard ───

/// Runtime guard that panics on drop if Critical was not properly consumed.
/// This is the last line of defense — catches any escape route that
/// bypasses propagate/suppress/suppress_with (e.g. custom consumer functions).
struct CriticalGuard {
    consumed: Cell<bool>,
}

impl CriticalGuard {
    fn new() -> Self {
        Self { consumed: Cell::new(false) }
    }

    fn mark_consumed(&self) {
        self.consumed.set(true);
    }
}

impl Drop for CriticalGuard {
    fn drop(&mut self) {
        if !self.consumed.get() {
            // Use eprintln to ensure visibility even if logging is not configured
            eprintln!("[CRITICAL-LEAK] Critical<T> dropped without being consumed via propagate/suppress/suppress_with. This is a bug — IO errors may have been silently swallowed.");
            // In debug/test builds, panic to catch immediately
            #[cfg(debug_assertions)]
            panic!("[CRITICAL-LEAK] Critical<T> dropped without consumption");
        }
    }
}

// ─── Critical ───

/// Opaque IO result wrapper. No .ok(), no .unwrap().
///
/// Three exits:
/// - `propagate(scope, token)` → V (panics on error — for fatal IO in transactions)
/// - `suppress(token, reason)` → Option<V> (logs and swallows error)
/// - `suppress_with(token, reason, default)` → V (logs and returns default)
#[must_use]
pub struct Critical<V> {
    inner: Result<V, CriticalError>,
    guard: CriticalGuard,
    tx_context: Option<([u8; 4], &'static str)>,
}

impl<V> Critical<V> {
    /// Construct a success value.
    pub fn ok(value: V) -> Self {
        Critical { inner: Ok(value), guard: CriticalGuard::new(), tx_context: None }
    }

    /// Construct a failure.
    pub fn error(operation: &'static str, message: String) -> Self {
        Critical {
            inner: Err(CriticalError { operation, message }),
            guard: CriticalGuard::new(),
            tx_context: None,
        }
    }

    /// Inject transaction context (trace_id + tx_name) into this Critical.
    /// Called by scope.xxx() methods to tag IO results with their origin.
    pub(crate) fn with_tx_context(mut self, trace_id: [u8; 4], tx_name: &'static str) -> Self {
        self.tx_context = Some((trace_id, tx_name));
        self
    }

    /// Format tx prefix from embedded context, if present.
    fn format_tx_prefix(&self) -> Option<String> {
        self.tx_context.map(|(id, name)| {
            format!("[TX:{}:{:02x}{:02x}{:02x}{:02x}]", name, id[0], id[1], id[2], id[3])
        })
    }

    /// Fatal propagation inside a transaction scope.
    /// On success: returns V.
    /// On error: records abort in scope, then panics.
    pub fn propagate<T: Transaction>(self, scope: &TxScope<T>, token: T::Fatal) -> V {
        self.guard.mark_consumed();
        match self.inner {
            Ok(v) => v,
            Err(err) => {
                scope.record_abort(token, err);
                panic!("tx_abort");
            }
        }
    }

    /// Suppress: on error, log warning and return None.
    pub fn suppress<S: SuppressibleToken>(self, token: S, reason: &str) -> Option<V> {
        self.guard.mark_consumed();
        match self.inner {
            Ok(v) => Some(v),
            Err(ref e) => {
                if let Some(prefix) = self.format_tx_prefix() {
                    tracing::warn!(
                        "{} suppress: {} ({}): {} — {}",
                        prefix,
                        token.operation_name(),
                        reason,
                        e.operation,
                        e.message
                    );
                } else {
                    tracing::warn!(
                        "[SUPPRESSED] {} ({}): {} — {}",
                        token.operation_name(),
                        reason,
                        e.operation,
                        e.message
                    );
                }
                None
            }
        }
    }

    /// Transform the success value, preserving errors.
    ///
    /// Useful in persist methods to chain transformations without consuming Critical.
    pub fn map<U>(self, f: impl FnOnce(V) -> U) -> Critical<U> {
        self.guard.mark_consumed();
        Critical {
            inner: self.inner.map(f),
            guard: CriticalGuard::new(),
            tx_context: self.tx_context,
        }
    }

    /// Chain IO operations. If self is Ok, apply f to get a new Critical.
    /// If self is Err, short-circuit with the original error.
    ///
    /// Useful in persist methods with multiple IO steps:
    /// ```ignore
    /// db.execute(insert_sql).and_then(|()| db.fetch_one(rowid_sql))
    /// ```
    pub fn and_then<U>(self, f: impl FnOnce(V) -> Critical<U>) -> Critical<U> {
        self.guard.mark_consumed();
        let ctx = self.tx_context;
        match self.inner {
            Ok(v) => {
                let result = f(v);
                // If f(v) produced a Critical without tx_context, inherit ours
                if result.tx_context.is_none() && ctx.is_some() {
                    Critical { tx_context: ctx, ..result }
                } else {
                    result
                }
            }
            Err(e) => Critical { inner: Err(e), guard: CriticalGuard::new(), tx_context: ctx },
        }
    }

    /// Suppress with default: on error, log warning and return default value.
    pub fn suppress_with<S: SuppressibleToken>(self, token: S, reason: &str, default: V) -> V {
        self.guard.mark_consumed();
        match self.inner {
            Ok(v) => v,
            Err(ref e) => {
                if let Some(prefix) = self.format_tx_prefix() {
                    tracing::warn!(
                        "{} suppress: {} ({}): {} — {}",
                        prefix,
                        token.operation_name(),
                        reason,
                        e.operation,
                        e.message
                    );
                } else {
                    tracing::warn!(
                        "[SUPPRESSED] {} ({}): {} — {}",
                        token.operation_name(),
                        reason,
                        e.operation,
                        e.message
                    );
                }
                default
            }
        }
    }
}

impl<V: fmt::Debug> fmt::Debug for Critical<V> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match &self.inner {
            Ok(v) => write!(f, "Critical::Ok({:?})", v),
            Err(e) => write!(f, "Critical::Err({})", e),
        }
    }
}

// ─── IntoCritical ───

/// Bridge trait: convert any Result<T, E: Display> into Critical<T>.
pub trait IntoCritical<T> {
    fn into_critical(self, operation: &'static str) -> Critical<T>;
}

impl<T, E: fmt::Display> IntoCritical<T> for Result<T, E> {
    fn into_critical(self, operation: &'static str) -> Critical<T> {
        match self {
            Ok(v) => Critical { inner: Ok(v), guard: CriticalGuard::new(), tx_context: None },
            Err(e) => Critical {
                inner: Err(CriticalError {
                    operation,
                    message: e.to_string(),
                }),
                guard: CriticalGuard::new(),
                tx_context: None,
            },
        }
    }
}

// ─── Tests ───

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

    // Fatal token for transaction tests
    #[derive(Debug, Clone, Copy, PartialEq)]
    enum TestFatal {
        ReadFile,
        Connect,
    }

    impl ConceptMapToken for TestFatal {
        fn operation_name(&self) -> &'static str {
            match self {
                TestFatal::ReadFile => "test::read_file",
                TestFatal::Connect => "test::connect",
            }
        }
    }

    impl FatalToken for TestFatal {}

    // Suppressible token for suppress tests
    #[derive(Debug, Clone, Copy, PartialEq)]
    enum TestSuppressible {
        WriteFile,
        LogEntry,
    }

    impl ConceptMapToken for TestSuppressible {
        fn operation_name(&self) -> &'static str {
            match self {
                TestSuppressible::WriteFile => "test::write_file",
                TestSuppressible::LogEntry => "test::log_entry",
            }
        }
    }

    impl SuppressibleToken for TestSuppressible {}

    // Transaction type for tests
    struct TestTx;
    impl Transaction for TestTx {
        type Fatal = TestFatal;
        type Suppressible = TestSuppressible;
        fn name() -> &'static str { "test" }
    }

    // Sub-transaction fatal token
    #[derive(Debug, Clone, Copy, PartialEq)]
    enum SubFatal {
        DbWrite,
        DbRead,
    }

    impl ConceptMapToken for SubFatal {
        fn operation_name(&self) -> &'static str {
            match self {
                SubFatal::DbWrite => "sub::db_write",
                SubFatal::DbRead => "sub::db_read",
            }
        }
    }

    impl FatalToken for SubFatal {}

    // Sub-transaction suppressible (empty)
    #[derive(Debug, Clone, Copy, PartialEq)]
    enum SubSuppressible {
        Noop,
    }

    impl ConceptMapToken for SubSuppressible {
        fn operation_name(&self) -> &'static str {
            "sub::noop"
        }
    }

    impl SuppressibleToken for SubSuppressible {}

    // Sub-transaction type
    struct SubTx;
    impl Transaction for SubTx {
        type Fatal = SubFatal;
        type Suppressible = SubSuppressible;
        fn name() -> &'static str { "sub_test" }
    }

    // Empty fatal token (suppress-only sub-transaction)
    #[derive(Debug, Clone, Copy, PartialEq)]
    enum EmptyFatal {}

    impl ConceptMapToken for EmptyFatal {
        fn operation_name(&self) -> &'static str {
            match *self {}
        }
    }

    impl FatalToken for EmptyFatal {}

    // Empty suppressible
    #[derive(Debug, Clone, Copy, PartialEq)]
    enum EmptySuppressible {}

    impl ConceptMapToken for EmptySuppressible {
        fn operation_name(&self) -> &'static str {
            match *self {}
        }
    }

    impl SuppressibleToken for EmptySuppressible {}

    // Empty transaction type
    struct EmptyTx;
    impl Transaction for EmptyTx {
        type Fatal = EmptyFatal;
        type Suppressible = EmptySuppressible;
        fn name() -> &'static str { "empty" }
    }

    // ─── propagate tests ───

    #[test]
    fn test_propagate_ok() {
        let scope: TxScope<TestTx> = TxScope::new();
        let c: Critical<i32> = Critical::ok(42);
        let val = c.propagate(&scope, TestFatal::ReadFile);
        assert_eq!(val, 42);
        assert!(scope.take_abort().is_none());
    }

    #[test]
    fn test_propagate_error_panics() {
        let scope: TxScope<TestTx> = TxScope::new();
        let c: Critical<i32> = Critical::error("read", "file not found".into());
        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
            c.propagate(&scope, TestFatal::ReadFile)
        }));
        assert!(result.is_err());
        let (token, detail) = scope.take_abort().unwrap();
        assert_eq!(token, TestFatal::ReadFile);
        assert!(detail.to_string().contains("file not found"));
    }

    #[test]
    fn test_propagate_panic_message() {
        let scope: TxScope<TestTx> = TxScope::new();
        let c: Critical<i32> = Critical::error("op", "err".into());
        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
            c.propagate(&scope, TestFatal::Connect)
        }));
        let panic_val = result.unwrap_err();
        let msg = panic_val.downcast_ref::<&str>().unwrap();
        assert_eq!(*msg, "tx_abort");
    }

    // ─── suppress tests ───

    #[test]
    fn test_suppress_ok() {
        let c: Critical<i32> = Critical::ok(42);
        assert_eq!(c.suppress(TestSuppressible::WriteFile, "optional"), Some(42));
    }

    #[test]
    fn test_suppress_error() {
        let c: Critical<i32> = Critical::error("write", "disk full".into());
        assert_eq!(c.suppress(TestSuppressible::WriteFile, "optional"), None);
    }

    #[test]
    fn test_suppress_with_ok() {
        let c: Critical<i32> = Critical::ok(42);
        assert_eq!(
            c.suppress_with(TestSuppressible::LogEntry, "fallback", 0),
            42
        );
    }

    #[test]
    fn test_suppress_with_error() {
        let c: Critical<i32> = Critical::error("log", "io error".into());
        assert_eq!(
            c.suppress_with(TestSuppressible::LogEntry, "fallback", 0),
            0
        );
    }

    // ─── TxScope tests ───

    #[test]
    fn test_tx_scope_default_empty() {
        let scope: TxScope<TestTx> = TxScope::new();
        assert!(scope.take_abort().is_none());
    }

    #[test]
    fn test_tx_scope_record_and_take() {
        let scope: TxScope<TestTx> = TxScope::new();
        scope.record_abort(
            TestFatal::Connect,
            CriticalError {
                operation: "connect",
                message: "refused".into(),
            },
        );
        let (token, detail) = scope.take_abort().unwrap();
        assert_eq!(token, TestFatal::Connect);
        assert!(detail.to_string().contains("refused"));
        // second take returns None
        assert!(scope.take_abort().is_none());
    }

    #[test]
    fn test_tx_scope_default_trait() {
        let scope: TxScope<TestTx> = TxScope::default();
        assert!(scope.take_abort().is_none());
    }

    // ─── sub-transaction propagate tests (async) ───

    #[tokio::test]
    async fn test_narrow_success() {
        let scope: TxScope<TestTx> = TxScope::new();
        let result = scope.propagate::<SubTx, _, _>(
            TestFatal::ReadFile,
            |sub| async move {
                Critical::ok(42).propagate(&sub, SubFatal::DbWrite)
            },
        ).await;
        assert_eq!(result, 42);
        assert!(scope.take_abort().is_none());
    }

    #[tokio::test]
    async fn test_narrow_sub_fatal_re_records_parent() {
        use futures_util::FutureExt;
        let scope = Arc::new(TxScope::<TestTx>::new());
        let scope2 = scope.clone();

        let result = AssertUnwindSafe(async {
            scope2.propagate::<SubTx, _, _>(
                TestFatal::ReadFile,
                |sub| async move {
                    Critical::<i32>::error("db", "write failed".into())
                        .propagate(&sub, SubFatal::DbWrite)
                },
            ).await
        }).catch_unwind().await;

        assert!(result.is_err());
        // Abort re-recorded under parent token
        let (token, detail) = scope.take_abort().unwrap();
        assert_eq!(token, TestFatal::ReadFile); // parent, not sub
        assert!(detail.to_string().contains("write failed")); // sub detail preserved
    }

    #[tokio::test]
    async fn test_narrow_empty_fatal_compiles() {
        let scope: TxScope<TestTx> = TxScope::new();
        // EmptyFatal has no variants → cannot propagate inside
        let result = scope.propagate::<EmptyTx, _, _>(
            TestFatal::ReadFile,
            |_sub| async { 42 },
        ).await;
        assert_eq!(result, 42);
    }

    // ─── sub-transaction suppress tests (async) ───

    #[tokio::test]
    async fn test_catch_success_returns_some() {
        let scope: TxScope<TestTx> = TxScope::new();
        let result = scope.suppress::<SubTx, _, _>(
            TestSuppressible::WriteFile,
            "sub-tx ok",
            |_sub| async { 42 },
        ).await;
        assert_eq!(result, Some(42));
    }

    #[tokio::test]
    async fn test_catch_suppresses_sub_fatal() {
        let scope: TxScope<TestTx> = TxScope::new();
        let result = scope.suppress::<SubTx, i32, _>(
            TestSuppressible::WriteFile,
            "db failure acceptable",
            |sub| async move {
                Critical::<i32>::error("db", "write failed".into())
                    .propagate(&sub, SubFatal::DbWrite)
            },
        ).await;
        assert_eq!(result, None);
        // Abort was consumed by suppress, not visible to parent
        assert!(scope.take_abort().is_none());
    }

    #[tokio::test]
    async fn test_catch_unexpected_panic_returns_none() {
        let scope: TxScope<TestTx> = TxScope::new();
        let result = scope.suppress::<SubTx, i32, _>(
            TestSuppressible::WriteFile,
            "might panic",
            |_sub| async {
                panic!("not a tx_abort");
            },
        ).await;
        assert_eq!(result, None);
        assert!(scope.take_abort().is_none());
    }

    #[tokio::test]
    async fn test_catch_does_not_affect_parent_scope() {
        let scope: TxScope<TestTx> = TxScope::new();

        // Sub-transaction abort is caught
        let _result = scope.suppress::<SubTx, i32, _>(
            TestSuppressible::WriteFile,
            "ok to fail",
            |sub| async move {
                Critical::<i32>::error("db", "boom".into())
                    .propagate(&sub, SubFatal::DbWrite)
            },
        ).await;

        // Parent scope can still record its own abort
        scope.record_abort(
            TestFatal::Connect,
            CriticalError {
                operation: "connect",
                message: "timeout".into(),
            },
        );
        let (token, _) = scope.take_abort().unwrap();
        assert_eq!(token, TestFatal::Connect);
    }

    // ─── IntoCritical tests ───

    #[test]
    fn test_into_critical_ok() {
        let r: Result<i32, String> = Ok(10);
        let c = r.into_critical("parse");
        assert_eq!(c.suppress(TestSuppressible::WriteFile, "test").unwrap(), 10);
    }

    #[test]
    fn test_into_critical_err() {
        let r: Result<i32, String> = Err("bad input".into());
        let c = r.into_critical("parse");
        assert!(c.suppress(TestSuppressible::WriteFile, "test").is_none());
    }

    // ─── map / and_then tests ───

    #[test]
    fn test_map_ok() {
        let c: Critical<i32> = Critical::ok(42);
        let mapped = c.map(|v| v * 2);
        assert_eq!(mapped.suppress(TestSuppressible::WriteFile, "test"), Some(84));
    }

    #[test]
    fn test_map_error() {
        let c: Critical<i32> = Critical::error("op", "fail".into());
        let mapped = c.map(|v| v * 2);
        assert!(mapped.suppress(TestSuppressible::WriteFile, "test").is_none());
    }

    #[test]
    fn test_and_then_ok_ok() {
        let c: Critical<i32> = Critical::ok(10);
        let chained = c.and_then(|v| Critical::ok(v + 5));
        assert_eq!(chained.suppress(TestSuppressible::WriteFile, "test"), Some(15));
    }

    #[test]
    fn test_and_then_ok_err() {
        let c: Critical<i32> = Critical::ok(10);
        let chained = c.and_then(|_| Critical::<i32>::error("step2", "boom".into()));
        assert!(chained.suppress(TestSuppressible::WriteFile, "test").is_none());
    }

    #[test]
    fn test_and_then_err_short_circuits() {
        let c: Critical<i32> = Critical::error("step1", "fail".into());
        let mut called = false;
        let chained = c.and_then(|_| {
            called = true;
            Critical::ok(99)
        });
        assert!(!called);
        assert!(chained.suppress(TestSuppressible::WriteFile, "test").is_none());
    }

    #[test]
    fn test_map_and_then_chain() {
        let c: Critical<String> = Critical::ok("hello".to_string());
        let result = c
            .map(|s| s.len())
            .and_then(|len| Critical::ok(len * 2));
        assert_eq!(result.suppress(TestSuppressible::WriteFile, "test"), Some(10));
    }

    // ─── Debug tests ───

    #[test]
    fn test_critical_debug_ok() {
        let c: Critical<i32> = Critical::ok(42);
        let debug = format!("{:?}", c);
        assert!(debug.contains("Ok(42)"));
        c.suppress(TestSuppressible::WriteFile, "debug test");
    }

    #[test]
    fn test_critical_debug_err() {
        let c: Critical<i32> = Critical::error("op", "fail".into());
        let debug = format!("{:?}", c);
        assert!(debug.contains("Err"));
        c.suppress(TestSuppressible::WriteFile, "debug test");
    }


    // ─── CriticalGuard tests ───

    #[test]
    fn test_guard_propagate_ok_no_panic() {
        let scope: TxScope<TestTx> = TxScope::new();
        let c: Critical<i32> = Critical::ok(42);
        let val = c.propagate(&scope, TestFatal::ReadFile);
        assert_eq!(val, 42);
        // No panic on drop — consumed via propagate
    }

    #[test]
    fn test_guard_suppress_ok_no_panic() {
        let c: Critical<i32> = Critical::ok(42);
        assert_eq!(c.suppress(TestSuppressible::WriteFile, "test"), Some(42));
        // No panic on drop — consumed via suppress
    }

    #[test]
    fn test_guard_map_chain_no_panic() {
        let c: Critical<i32> = Critical::ok(42);
        let mapped = c.map(|v| v * 2);
        assert_eq!(mapped.suppress(TestSuppressible::WriteFile, "test"), Some(84));
        // No panic — old guard consumed by map, new guard consumed by suppress
    }

    #[test]
    fn test_guard_and_then_chain_no_panic() {
        let c: Critical<i32> = Critical::ok(10);
        let chained = c.and_then(|v| Critical::ok(v + 5));
        assert_eq!(chained.suppress(TestSuppressible::WriteFile, "test"), Some(15));
        // No panic — old guard consumed by and_then, new guard consumed by suppress
    }

    #[test]
    #[should_panic(expected = "CRITICAL-LEAK")]
    fn test_guard_unconsumed_panics() {
        let _c: Critical<i32> = Critical::ok(42);
        // Drop without consumption → panic in debug_assertions
    }

    #[test]
    #[should_panic(expected = "CRITICAL-LEAK")]
    fn test_guard_black_hole_panics() {
        fn black_hole<T>(_: T) {}
        black_hole(Critical::<i32>::ok(42));
        // black_hole consumes by move, but guard not marked → panic on drop
    }

    // ─── Token trait tests ───

    #[test]
    fn test_fatal_token_operation_name() {
        assert_eq!(TestFatal::ReadFile.operation_name(), "test::read_file");
        assert_eq!(TestFatal::Connect.operation_name(), "test::connect");
    }

    #[test]
    fn test_suppressible_token_operation_name() {
        assert_eq!(
            TestSuppressible::WriteFile.operation_name(),
            "test::write_file"
        );
        assert_eq!(
            TestSuppressible::LogEntry.operation_name(),
            "test::log_entry"
        );
    }

    // ─── Startup transaction tests ───

    #[tokio::test]
    async fn test_run_startup_ok() {
        let result = run_startup::<TestTx, _, _>(|_scope| async {
            42
        }).await;
        match result {
            StartupResult::Ok(v) => assert_eq!(v, 42),
            _ => panic!("Expected Ok"),
        }
    }

    #[tokio::test]
    async fn test_run_startup_aborted() {
        let result = run_startup::<TestTx, (), _>(|scope| async move {
            let c = Critical::<i32>::error("startup_op", "disk full".to_string());
            c.propagate(&scope, TestFatal::ReadFile);
        }).await;
        match result {
            StartupResult::Aborted(token, _detail) => {
                assert_eq!(token, TestFatal::ReadFile);
            },
            _ => panic!("Expected Aborted"),
        }
    }

    #[tokio::test]
    async fn test_run_startup_unexpected_panic() {
        let result = run_startup::<TestTx, (), _>(|_scope| async {
            panic!("something unexpected");
        }).await;
        match result {
            StartupResult::UnexpectedPanic => {},
            _ => panic!("Expected UnexpectedPanic"),
        }
    }

    // ─── TxScope DB binding tests ───

    #[tokio::test]
    async fn test_txscope_db_lazy_begin_and_auto_commit() {
        use sqlx::sqlite::SqlitePoolOptions;
        // Setup in-memory db
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::query("CREATE TABLE test_tx (id INTEGER PRIMARY KEY, val TEXT)")
            .execute(&pool)
            .await
            .unwrap();
        let db = crate::db::Db::from_pool(pool.clone());

        // Create scope with db binding
        let scope: TxScope<TestTx> = TxScope::new_with_db(db);

        // Use scope.db() to insert — this lazily begins a transaction
        {
            let mut sdb = scope.db().await;
            sdb.execute(
                sqlx::query("INSERT INTO test_tx (id, val) VALUES (?, ?)")
                    .bind(1i64)
                    .bind("hello"),
            ).await.suppress(TestSuppressible::WriteFile, "test").unwrap();
        }

        // Auto-commit
        scope.auto_commit().await;

        // Verify committed by reading directly from pool
        let row: (String,) = sqlx::query_as("SELECT val FROM test_tx WHERE id = 1")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(row.0, "hello");
    }

    #[tokio::test]
    async fn test_txscope_db_rollback_without_commit() {
        use sqlx::sqlite::SqlitePoolOptions;
        let pool = SqlitePoolOptions::new()
            .max_connections(2)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::query("CREATE TABLE test_tx2 (id INTEGER PRIMARY KEY, val TEXT)")
            .execute(&pool)
            .await
            .unwrap();
        let db = crate::db::Db::from_pool(pool.clone());

        // Create scope with db binding, insert, but DON'T auto_commit
        {
            let scope: TxScope<TestTx> = TxScope::new_with_db(db);
            {
                let mut sdb = scope.db().await;
                sdb.execute(
                    sqlx::query("INSERT INTO test_tx2 (id, val) VALUES (?, ?)")
                        .bind(1i64)
                        .bind("should_rollback"),
                ).await.suppress(TestSuppressible::WriteFile, "test").unwrap();
            }
            // scope dropped without auto_commit → Tx dropped → rollback
        }

        // Verify nothing persisted
        let rows: Vec<(i64,)> = sqlx::query_as("SELECT id FROM test_tx2")
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(rows.len(), 0);
    }

    #[tokio::test]
    #[should_panic(expected = "not bound to a database")]
    async fn test_txscope_db_panics_without_binding() {
        let scope: TxScope<TestTx> = TxScope::new();
        let _sdb = scope.db().await;
    }


    // ─── SAVEPOINT tests ───

    #[tokio::test]
    async fn test_suppress_savepoint_success() {
        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap();
        sqlx::query("CREATE TABLE sp_test (id INTEGER PRIMARY KEY, val TEXT)")
            .execute(&pool).await.unwrap();

        let db = crate::db::Db::from_pool(pool.clone());
        let scope = TxScope::<TestTx>::new_with_db(db);

        // Parent writes
        {
            let mut sdb = scope.db().await;
            sdb.execute(sqlx::query("INSERT INTO sp_test (id, val) VALUES (1, 'parent')"))
                .await.suppress(TestSuppressible::WriteFile, "test").unwrap();
        }

        // Suppress sub-transaction (success)
        let result = scope.suppress::<SubTx, _, _>(
            TestSuppressible::WriteFile, "test",
            |sub| async move {
                let mut sdb = sub.db().await;
                sdb.execute(sqlx::query("INSERT INTO sp_test (id, val) VALUES (2, 'child')"))
                    .await.suppress(SubSuppressible::Noop, "test").unwrap();
                42
            },
        ).await;
        assert_eq!(result, Some(42));

        scope.auto_commit().await;

        // Both rows visible
        let rows: Vec<(i64, String)> = sqlx::query_as("SELECT id, val FROM sp_test ORDER BY id")
            .fetch_all(&pool).await.unwrap();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0], (1, "parent".to_string()));
        assert_eq!(rows[1], (2, "child".to_string()));
    }

    #[tokio::test]
    async fn test_suppress_savepoint_rollback() {
        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap();
        sqlx::query("CREATE TABLE sp_test2 (id INTEGER PRIMARY KEY, val TEXT)")
            .execute(&pool).await.unwrap();

        let db = crate::db::Db::from_pool(pool.clone());
        let scope = TxScope::<TestTx>::new_with_db(db);

        // Parent writes
        {
            let mut sdb = scope.db().await;
            sdb.execute(sqlx::query("INSERT INTO sp_test2 (id, val) VALUES (1, 'parent')"))
                .await.suppress(TestSuppressible::WriteFile, "test").unwrap();
        }

        // Suppress sub-transaction (failure)
        let result = scope.suppress::<SubTx, i32, _>(
            TestSuppressible::WriteFile, "test",
            |sub| async move {
                {
                    let mut sdb = sub.db().await;
                    sdb.execute(sqlx::query("INSERT INTO sp_test2 (id, val) VALUES (2, 'rollback')"))
                        .await.suppress(SubSuppressible::Noop, "test").unwrap();
                }
                let err: Critical<i32> = Critical::error("sub_op", "fail".to_string());
                err.propagate(&sub, SubFatal::DbWrite)
            },
        ).await;
        assert!(result.is_none());

        scope.auto_commit().await;

        // Only parent row (child rolled back via SAVEPOINT)
        let rows: Vec<(i64, String)> = sqlx::query_as("SELECT id, val FROM sp_test2 ORDER BY id")
            .fetch_all(&pool).await.unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0], (1, "parent".to_string()));
    }

    #[tokio::test]
    async fn test_propagate_savepoint_success() {
        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap();
        sqlx::query("CREATE TABLE sp_test3 (id INTEGER PRIMARY KEY, val TEXT)")
            .execute(&pool).await.unwrap();

        let db = crate::db::Db::from_pool(pool.clone());
        let scope = TxScope::<TestTx>::new_with_db(db);

        // Parent writes
        {
            let mut sdb = scope.db().await;
            sdb.execute(sqlx::query("INSERT INTO sp_test3 (id, val) VALUES (1, 'parent')"))
                .await.suppress(TestSuppressible::WriteFile, "test").unwrap();
        }

        // Propagate sub-transaction (success)
        let value: i32 = scope.propagate::<SubTx, _, _>(
            TestFatal::ReadFile,
            |sub| async move {
                let mut sdb = sub.db().await;
                sdb.execute(sqlx::query("INSERT INTO sp_test3 (id, val) VALUES (2, 'child')"))
                    .await.suppress(SubSuppressible::Noop, "test").unwrap();
                99
            },
        ).await;
        assert_eq!(value, 99);

        scope.auto_commit().await;

        let rows: Vec<(i64, String)> = sqlx::query_as("SELECT id, val FROM sp_test3 ORDER BY id")
            .fetch_all(&pool).await.unwrap();
        assert_eq!(rows.len(), 2);
    }

    #[tokio::test]
    async fn test_propagate_savepoint_rollback() {
        use futures_util::FutureExt;

        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap();
        sqlx::query("CREATE TABLE sp_test4 (id INTEGER PRIMARY KEY, val TEXT)")
            .execute(&pool).await.unwrap();

        let db = crate::db::Db::from_pool(pool.clone());
        let scope = Arc::new(TxScope::<TestTx>::new_with_db(db));

        // Parent writes
        {
            let mut sdb = scope.db().await;
            sdb.execute(sqlx::query("INSERT INTO sp_test4 (id, val) VALUES (1, 'parent')"))
                .await.suppress(TestSuppressible::WriteFile, "test").unwrap();
        }

        // Propagate sub-transaction (failure)
        let scope_clone = scope.clone();
        let result = std::panic::AssertUnwindSafe(async {
            scope_clone.propagate::<SubTx, (), _>(
                TestFatal::ReadFile,
                |sub| async move {
                    {
                        let mut sdb = sub.db().await;
                        sdb.execute(sqlx::query("INSERT INTO sp_test4 (id, val) VALUES (2, 'rollback')"))
                            .await.suppress(SubSuppressible::Noop, "test").unwrap();
                    }
                    let err: Critical<()> = Critical::error("sub_op", "fail".to_string());
                    err.propagate(&sub, SubFatal::DbWrite);
                },
            ).await;
        })
        .catch_unwind()
        .await;

        assert!(result.is_err());

        // Parent abort should be recorded
        let abort = scope.take_abort();
        assert!(abort.is_some());

        // Drop scope without commit -> rollback everything
        drop(scope);
        drop(scope_clone);
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let rows: Vec<(i64,)> = sqlx::query_as("SELECT id FROM sp_test4")
            .fetch_all(&pool).await.unwrap();
        assert_eq!(rows.len(), 0);
    }

    #[tokio::test]
    async fn test_suppress_no_db_no_savepoint() {
        let scope = TxScope::<TestTx>::new();

        let result = scope.suppress::<SubTx, _, _>(
            TestSuppressible::WriteFile, "test",
            |_sub| async move { 42 },
        ).await;
        assert_eq!(result, Some(42));

        let result = scope.suppress::<SubTx, i32, _>(
            TestSuppressible::WriteFile, "test",
            |sub| async move {
                let err: Critical<i32> = Critical::error("op", "fail".to_string());
                err.propagate(&sub, SubFatal::DbWrite)
            },
        ).await;
        assert!(result.is_none());
    }


    // ─── Dynamic Data Source Tests ───

    // A test DynamicDao using temp files
    struct TestDynDao;
    impl crate::db::DynamicDao for TestDynDao {
        fn resolve(param: &str) -> crate::db::DsConfig {
            let tmp = std::env::temp_dir().join(format!("mad_hatter_dyn_test_{}", std::process::id()));
            let p = tmp.join(format!("dyn_{}.db", param));
            crate::db::DsConfig::sqlite(p.to_str().unwrap().to_string())
        }
    }

    // Another DynamicDao for testing type isolation
    struct TestDynDao2;
    impl crate::db::DynamicDao for TestDynDao2 {
        fn resolve(param: &str) -> crate::db::DsConfig {
            let tmp = std::env::temp_dir().join(format!("mad_hatter_dyn_test_{}", std::process::id()));
            let p = tmp.join(format!("dyn2_{}.db", param));
            crate::db::DsConfig::sqlite(p.to_str().unwrap().to_string())
        }
    }





}
