///|
/// Failures that come from the database rather than from aya.
pub(all) suberror DbError {
  /// The driver could not talk to the database at all.
  ConnectionFailed(String)
  /// The database rejected or failed a statement.
  QueryFailed(sql~ : String, message~ : String)
  /// `one` found no row where exactly one was required.
  NotFound(sql~ : String)
  /// `one` found more than one row.
  TooManyRows(sql~ : String, got~ : Int)
  /// A nested transaction failed and the enclosing body carried on anyway.
  /// The outermost scope rolled back rather than commit half the work, and
  /// reports the failure that made it impossible.
  RollbackOnly(cause~ : Error)
} derive(Debug)

///|
/// What running a statement needs: text, parameters, and the dialect they were
/// built for. No transaction control.
///
/// This is deliberately the smaller half of what a database connection can do.
/// Everything aya hands to the body of a transaction is an `Executor` and
/// nothing more, so a repository holding one cannot commit or roll back the
/// transaction it is running inside.
pub(open) trait Executor {
  /// Run a statement and return its rows, each already in projection order.
  ///
  /// `columns` is how many values a row must come back with. A driver whose
  /// binding reports the result width can ignore it; one whose binding does
  /// not — SQLite's — needs telling where the row ends.
  async fn query(Self, String, Array[SqlValue], columns~ : Int) -> Array[
    Array[SqlValue],
  ] raise DbError
  /// Run a statement and return how many rows it touched.
  async fn execute(Self, String, Array[SqlValue]) -> Int raise DbError
  /// Which placeholder syntax this executor expects.
  fn dialect(Self) -> Dialect
}

///|
/// What aya needs from a database binding: an `Executor` that can also
/// bracket a transaction.
///
/// Everything above this trait is dialect-agnostic: aya builds SQL text and
/// an ordered parameter list, and a driver is whatever can run that pair. A
/// driver is also the connection — pooling, if wanted, belongs outside.
///
/// The three transaction statements are separated out from `Executor` because
/// they are a capability, not a detail: `Tx` is the only thing in aya that
/// sends them, and nothing aya hands to user code implements this trait.
pub(open) trait Driver: Executor {
  async fn begin(Self) -> Unit raise DbError
  async fn commit(Self) -> Unit raise DbError
  async fn rollback(Self) -> Unit raise DbError
}

///|
/// A driver plus how deep into nested transactions it currently is.
///
/// Build one per connection, at wiring time, and hand *it* to every repository
/// rather than handing them the driver. That shared depth is the whole point:
/// a repository that opens a transaction of its own joins the enclosing one
/// instead of asking the database to `BEGIN` twice.
///
/// ```moonbit nocheck
/// let db = @aya.Tx::new(driver)
/// let tickets = SqlTickets::new(db)
/// let users = SqlUsers::new(db)
/// @aya.transaction(db, _ => {
///   tickets.save(a) // both saves land in one transaction
///   users.save(b)
/// })
/// ```
///
/// A `Tx` is not safe to share between concurrently running tasks: the depth
/// it counts is a property of one connection's position in one call stack.
pub struct Tx[D] {
  db : D
  /// How many `transaction` scopes are currently open. Zero means no
  /// transaction is in flight and the next one has to `BEGIN`.
  mut depth : Int
  /// What a nested scope failed with, if one did. Set once and only cleared
  /// when the outermost scope finishes, so the first failure is the one
  /// reported.
  mut failure : Error?
}

///|
/// Wrap a driver so transactions over it can nest.
pub fn[D] Tx::new(db : D) -> Tx[D] {
  { db, depth: 0, failure: None }
}

///|
/// The driver underneath, for the things aya does not model.
///
/// Reaching for this inside a `transaction` body means talking to the same
/// connection the transaction is open on, which is usually what is wanted —
/// and also means `commit` and `rollback` are in reach again, which is not.
pub fn[D] Tx::driver(self : Tx[D]) -> D {
  self.db
}

///|
/// Whether a transaction is currently open on this connection.
pub fn[D] Tx::is_open(self : Tx[D]) -> Bool {
  self.depth > 0
}

///|
pub impl[D : Driver] Executor for Tx[D] with fn query(
  self,
  sql,
  params,
  columns~,
) {
  Executor::query(self.db, sql, params, columns~)
}

///|
pub impl[D : Driver] Executor for Tx[D] with fn execute(self, sql, params) {
  Executor::execute(self.db, sql, params)
}

///|
pub impl[D : Driver] Executor for Tx[D] with fn dialect(self) {
  Executor::dialect(self.db)
}

///|
/// Run `body` inside a transaction, committing on success and rolling back on
/// any failure.
///
/// Written as a combinator rather than exposed as begin/commit/rollback so
/// that an early `raise` in the middle of `body` cannot leave a transaction
/// open. MoonBit's error effect does the work a transaction monad would do
/// elsewhere: `body` is an ordinary function that may raise.
///
/// Calls nest. Only the outermost scope sends `BEGIN` and `COMMIT`; an inner
/// one just runs its body, so `A.save()` and `B.save()` land in one
/// transaction when something wrapped them in another. What an inner scope
/// cannot do is fail by itself: with no savepoint there is no way to undo only
/// its writes, so its failure marks the whole transaction rollback-only and
/// the outermost scope raises `RollbackOnly` rather than commit half the work.
pub async fn[D : Driver, A] transaction(
  tx : Tx[D],
  body : async (Tx[D]) -> A,
) -> A {
  if tx.depth > 0 {
    tx.join(body)
  } else {
    tx.outermost(body)
  }
}

///|
/// Run `body` as part of the transaction that is already open.
///
/// No statement goes out either side of it. The depth is still counted, so a
/// scope nested three deep knows it is not the one that has to commit.
// The depth bookkeeping belongs in `defer`, which runs on both ways out. What
// is left in `catch` is not cleanup: it reads the failure that poisons the
// transaction, and `errdefer` cannot see the error. `fragile_catch_all`
// matches on the shape either way, so it is silenced for this function alone.
#warnings("-fragile_catch_all")
async fn[D, A] Tx::join(self : Tx[D], body : async (Tx[D]) -> A) -> A {
  self.depth += 1
  defer {
    self.depth -= 1
  }
  body(self) catch {
    e => {
      // Keep the first failure: it is the one that made the transaction
      // uncommittable, whatever happened after it.
      if self.failure is None {
        self.failure = Some(e)
      }
      raise e
    }
  }
}

///|
/// Open a transaction, run `body`, and close it exactly once.
async fn[D : Driver, A] Tx::outermost(
  self : Tx[D],
  body : async (Tx[D]) -> A,
) -> A {
  Driver::begin(self.db)
  self.depth = 1
  self.failure = None
  let result = {
    errdefer {
      self.finish() |> ignore
      self.rollback_quietly()
    }
    body(self)
  }
  // Deciding out here rather than inside the block above is what keeps the two
  // outcomes apart: the `errdefer` covers the body and nothing else, so a
  // `raise` from below it cannot roll back twice.
  match self.finish() {
    // The body swallowed a nested scope's failure and returned anyway.
    // Committing now would write whichever half of the work survived.
    Some(cause) => {
      self.rollback_quietly()
      raise RollbackOnly(cause~)
    }
    None => {
      errdefer self.rollback_quietly()
      Driver::commit(self.db)
      result
    }
  }
}

///|
/// Close the outermost scope, handing back whatever poisoned it, if anything.
fn[D] Tx::finish(self : Tx[D]) -> Error? {
  self.depth = 0
  let failure = self.failure
  self.failure = None
  failure
}

///|
/// Roll back, ignoring a failure to do so.
///
/// A failed rollback must not replace the error that caused it: the original
/// failure is the one worth reporting.
async fn[D : Driver] Tx::rollback_quietly(self : Tx[D]) -> Unit {
  Driver::rollback(self.db) catch {
    _ => ()
  }
}

///|
/// Build the statement for this executor and run it.
///
/// The text comes back alongside the rows because `one` names the statement in
/// the error it raises.
async fn[E : Executor, C, A] Query::fetch(
  self : Query[C, A],
  db : E,
) -> (String, Array[Array[SqlValue]]) {
  let (sql, params) = self.to_sql(dialect=Executor::dialect(db))
  let rows = Executor::query(db, sql, params, columns=self.projection.length())
  (sql, rows)
}

///|
/// Run the query and decode every row.
pub async fn[E : Executor, C, A] Query::run(
  self : Query[C, A],
  db : E,
) -> Array[A] {
  let (_, rows) = self.fetch(db)
  rows.map(row => (self.decode)(row[:]))
}

///|
/// Run the query and decode the single row it must return.
pub async fn[E : Executor, C, A] Query::one(self : Query[C, A], db : E) -> A {
  let (sql, rows) = self.fetch(db)
  match rows {
    [] => raise NotFound(sql~)
    [row] => (self.decode)(row[:])
    _ => raise TooManyRows(sql~, got=rows.length())
  }
}

///|
/// Run the query and decode the first row, if there is one.
pub async fn[E : Executor, C, A] Query::first(self : Query[C, A], db : E) -> A? {
  let (_, rows) = self.fetch(db)
  match rows {
    [] => None
    [row, ..] => Some((self.decode)(row[:]))
  }
}

///|
/// Run the insert, returning how many rows were added.
pub async fn[E : Executor] Insert::run(self : Insert, db : E) -> Int {
  let (sql, params) = self.to_sql(dialect=Executor::dialect(db))
  Executor::execute(db, sql, params)
}

///|
/// Run the update, returning how many rows were changed.
pub async fn[E : Executor, C] Update::run(self : Update[C], db : E) -> Int {
  let (sql, params) = self.to_sql(dialect=Executor::dialect(db))
  Executor::execute(db, sql, params)
}

///|
/// Run the delete, returning how many rows were removed.
pub async fn[E : Executor, C] Delete::run(self : Delete[C], db : E) -> Int {
  let (sql, params) = self.to_sql(dialect=Executor::dialect(db))
  Executor::execute(db, sql, params)
}