///|
/// The outcome of one asynchronous call to `Statement::step_async`.
///
/// `Done(changes=None)` means SQLite classified the statement as read-only.
/// For a potentially writing statement, `Done(changes=Some(count))` carries
/// the connection's direct row-change count captured before its mutex was
/// released.
pub(all) enum StepResult {
  Row
  Done(changes~ : Int64?)
} derive(Debug, Eq)

///|
// Translate I/O failures at each public operation's boundary, covering job
// construction, completion, and disposal. Cancellation remains an async error.
fn async_io_error(error : @os_error.OSError) -> SqliteError {
  sqlite_error(SQLITE_IOERR, "sqlite3: asynchronous operation failed: \{error}")
}

///|
/// Open a database without blocking the MoonBit event loop.
///
/// Later asynchronous operations on the connection execute in submission order.
pub async fn Connection::open_async(filename : String) -> Connection {
  try {
    if filename.contains_code_unit(0x0000) {
      raise sqlite_misuse("sqlite3: filename contains a NUL code unit")
    }
    let executor = create_executor()
    let mut executor_owned = true
    defer (if executor_owned { @ffi.sqlite3_executor_release(executor) })
    let flags = SQLITE_OPEN_READWRITE |
      SQLITE_OPEN_CREATE |
      SQLITE_OPEN_FULLMUTEX
    let submit_rescode = Ref(SQLITE_OK)
    let job = @ffi.sqlite3_open_job(
      executor,
      @utf8.encode(filename),
      flags,
      submit_rescode,
    )
    defer job.release()
    if submit_rescode.val != SQLITE_OK {
      raise sqlite_error(
        submit_rescode.val,
        "sqlite3: failed to submit asynchronous open",
      )
    }
    @async.protect_from_cancel(() => job.wait())
    let (rescode, extended_rescode, message) = job.result()
    if rescode != SQLITE_OK {
      raise sqlite_error(
        extended_rescode,
        if message == "" {
          "sqlite3: asynchronous open failed"
        } else {
          message
        },
      )
    }
    if @async.is_being_cancelled() {
      @async.protect_from_cancel(() => job.discard())
      @async.pause()
    }
    let database = job.take_database()
    executor_owned = false
    Connection::from_open_database(database, Some(executor))
  } catch {
    @os_error.OSError(_) as error => raise async_io_error(error)
    error => raise error
  }
}

///|
/// Prepare one SQL statement without blocking the MoonBit event loop.
///
/// Jobs for one connection execute in submission order. Cancellation waits for an
/// in-flight prepare and finalizes its result before returning control to the
/// caller.
pub async fn Connection::prepare_async(
  self : Connection,
  sql : String,
) -> Statement {
  try {
    if sql.contains_code_unit(0x0000) {
      raise sqlite_misuse("sqlite3: SQL contains a NUL code unit")
    }
    let (database, executor) = self.begin_job()
    defer self.finish_job()
    let submit_rescode = Ref(SQLITE_OK)
    // Job construction reserves cleanup before SQLite can produce a resource.
    // Keep that resource with its job until validation and cancellation checks
    // finish, so rejection can dispose of it without publishing a raw handle.
    let job = @ffi.sqlite3_prepare_job(executor, database, sql, submit_rescode)
    defer job.release()
    if submit_rescode.val != SQLITE_OK {
      raise sqlite_error(
        submit_rescode.val,
        "sqlite3: failed to submit asynchronous prepare",
      )
    }
    @async.protect_from_cancel(() => job.wait())
    let (rescode, extended_rescode, message, tail_offset) = job.result()
    if rescode != SQLITE_OK {
      raise sqlite_error(
        extended_rescode,
        if message == "" {
          "sqlite3: asynchronous prepare failed"
        } else {
          message
        },
      )
    }
    if tail_offset < 0 {
      raise sqlite_misuse("sqlite3: SQL does not contain a statement")
    }
    let invalid_tail = tail_offset > sql.length() ||
      !sql_tail_is_ignorable(sql, tail_offset)
    let cancelled = @async.is_being_cancelled()
    if invalid_tail || cancelled {
      @async.protect_from_cancel(() => job.discard())
    }
    if invalid_tail {
      raise sqlite_misuse("sqlite3: SQL contains more than one statement")
    }
    if cancelled {
      @async.pause()
    }
    let statement = job.take_statement()
    {
      handle: Some(statement),
      step_in_flight: Ref(false),
      has_row: false,
      column_types: None,
      connection: self,
    }
  } catch {
    @os_error.OSError(_) as error => raise async_io_error(error)
    error => raise error
  }
}

///|
/// Execute one step without blocking the MoonBit event loop.
///
/// Jobs for one connection execute in submission order. Operations on different
/// connections may run concurrently.
///
/// Synchronous operations on other statements may run while this step is in
/// flight, but can block behind SQLite's serialized connection access. The
/// statement being stepped rejects synchronous use until the step completes.
///
/// Cancellation waits for an in-flight SQLite step to finish before returning
/// the connection and statement to the caller.
pub async fn Statement::step_async(self : Statement) -> StepResult {
  try {
    let connection = self.connection
    let result = @async.protect_from_cancel() <| () => {
      let statement = self.get_handle()
      self.step_in_flight.val = true
      defer {
        self.step_in_flight.val = false
      }
      let (database, executor) = connection.begin_job()
      defer connection.finish_job()
      self.has_row = false
      self.column_types = None
      let submit_rescode = Ref(SQLITE_OK)
      let job = @ffi.sqlite3_step_job(
        executor, database, statement, submit_rescode,
      )
      defer job.release()
      if submit_rescode.val != SQLITE_OK {
        raise sqlite_error(
          submit_rescode.val,
          "sqlite3: failed to submit asynchronous operation",
        )
      }
      job.wait()
      let (rescode, extended_rescode, message, changes) = job.result()
      match rescode {
        SQLITE_ROW => {
          self.has_row = true
          Row
        }
        SQLITE_DONE => Done(changes~)
        _ =>
          raise sqlite_error(
            extended_rescode,
            if message == "" {
              "sqlite3: asynchronous operation failed"
            } else {
              message
            },
          )
      }
    }
    if @async.is_being_cancelled() {
      @async.pause()
    }
    result
  } catch {
    @os_error.OSError(_) as error => raise async_io_error(error)
    error => raise error
  }
}