//// An embedded, crash-safe key-value store for Gleam.
////
//// trove stores data in an append-only B+ tree on disk. Every write appends
//// new nodes and creates a new root. Old data is never overwritten, which
//// gives you crash safety, zero-cost MVCC snapshots, and single-writer /
//// multiple-reader concurrency backed by an OTP actor.
////
//// ## Quick Start
////
//// ```gleam
//// import gleam/string
//// import trove
//// import trove/codec
////
//// let config = trove.Config(
////   path: "./my_db",
////   key_codec: codec.string(),
////   value_codec: codec.string(),
////   key_compare: string.compare,
////   auto_compact: trove.AutoCompact(min_dirt: 1000, min_dirt_factor: 0.25),
////   auto_file_sync: trove.AutoSync,
////   call_timeout: 5000,
//// )
////
//// let assert Ok(db) = trove.open(config)
//// trove.put(db, key: "language", value: "gleam")
//// let assert Ok("gleam") = trove.get(db, key: "language")
//// trove.close(db)
//// ```

import exception.{type Exception}
import gleam/dynamic.{type Dynamic}
import gleam/erlang/atom
import gleam/erlang/process
import gleam/erlang/reference
import gleam/list
import gleam/option
import gleam/order
import gleam/result
import gleam/yielder
import trove/codec
import trove/internal/db
import trove/internal/snapshot
import trove/internal/tx
import trove/range

const reserved_default_keyspace_name = "__trove_default__"

@external(erlang, "erlang", "raise")
fn erlang_raise(
  class: atom.Atom,
  reason: Dynamic,
  stacktrace: List(Dynamic),
) -> a

fn reraise(ex: Exception) -> a {
  let #(class, reason) = case ex {
    exception.Errored(reason) -> #("error", reason)
    exception.Thrown(reason) -> #("throw", reason)
    exception.Exited(reason) -> #("exit", reason)
  }
  erlang_raise(atom.create(class), reason, [])
}

/// Errors that can occur when opening a database.
pub type OpenError {
  /// The database directory could not be created or accessed.
  DirectoryError(detail: String)
  /// The store file could not be opened or its header could not be recovered.
  StoreError(detail: String)
  /// The database path is already open by another actor on this node.
  LockError(detail: String)
  /// The OTP actor failed to start.
  ActorStartError
}

/// Controls automatic compaction behavior. When auto-compaction is enabled,
/// compaction triggers after a write if both `min_dirt` and `min_dirt_factor`
/// thresholds are exceeded simultaneously.
///
/// The recommended default is `AutoCompact(min_dirt: 1000, min_dirt_factor:
/// 0.25)`, which keeps disk usage bounded without compacting too aggressively.
///
/// **Note:** Auto-compaction runs synchronously inside the database actor.
/// While compaction is in progress, all other operations (reads, writes,
/// snapshots) are queued and may time out on large databases. For
/// latency-sensitive workloads, prefer `NoAutoCompact` and call `compact`
/// manually from a separate process with an appropriate timeout.
pub type AutoCompact {
  /// Enable auto-compaction. `min_dirt` is the minimum number of mutation
  /// operations (inserts, updates, and deletes each add one to the dirt count)
  /// and `min_dirt_factor` is the minimum dirt ratio (0.0 to 1.0). Both must
  /// be exceeded for compaction to trigger.
  AutoCompact(min_dirt: Int, min_dirt_factor: Float)
  /// Disable auto-compaction. Compaction can still be triggered manually
  /// with `compact`.
  NoAutoCompact
}

/// Controls whether writes are automatically fsynced to disk.
pub type FileSync {
  /// Automatically fsync after every write for maximum durability.
  AutoSync
  /// Do not fsync automatically. Use `file_sync` to flush manually.
  ManualSync
}

/// Database configuration passed to `open`.
///
/// `path` is the directory where store files are kept (created if needed).
///
/// `key_codec` and `value_codec` control how keys and values are serialized
/// to bytes. Both must satisfy `decode(encode(v)) == Ok(v)` and produce the
/// same output for the same input.
///
/// `key_compare` is a total order over keys: deterministic, antisymmetric,
/// and transitive. It must agree with `key_codec`, so keys that compare
/// as `Eq` encode to identical bytes.
///
/// `auto_compact` toggles automatic compaction after writes. `auto_file_sync`
/// controls whether writes are automatically fsynced.
///
/// `call_timeout` is the number of milliseconds to wait for actor responses;
/// 5000 is a reasonable starting point.
pub type Config(k, v) {
  Config(
    path: String,
    key_codec: codec.Codec(k),
    value_codec: codec.Codec(v),
    key_compare: fn(k, k) -> order.Order,
    auto_compact: AutoCompact,
    auto_file_sync: FileSync,
    call_timeout: Int,
  )
}

/// An open database handle. Parameterized by key type `k` and value type `v`.
pub opaque type Db(k, v) {
  Db(subject: process.Subject(db.Message(k, v)), call_timeout: Int)
}

/// A handle to a named keyspace. Obtained via `trove.keyspace(...)`, which
/// registers the keyspace's codecs and comparator with the database so later
/// operations (`put_in`, `get_in`, compaction) can operate on it.
pub opaque type Keyspace(k, v) {
  Keyspace(
    name: String,
    key_codec: codec.Codec(k),
    value_codec: codec.Codec(v),
    key_compare: fn(k, k) -> order.Order,
  )
}

/// Obtain a typed handle to a named keyspace. First use of a name registers it
/// in this session; later uses update the codecs.
///
/// **Panics** if `name` collides with the reserved default-keyspace sentinel.
///
/// **Codec trust model.** Passing codecs that don't match those previously
/// used for the same keyspace is undefined behavior: reads will likely
/// produce garbage values or panics. Keep the
/// `(key_codec, value_codec, key_compare)` tuple stable across opens for a
/// given keyspace name. Matches the trust model of `Config.key_codec` and
/// `Config.value_codec`.
///
/// ```gleam
/// let users =
///   trove.keyspace(
///     db,
///     name: "users",
///     key_codec: codec.string(),
///     value_codec: codec.string(),
///     key_compare: string.compare,
///   )
/// ```
pub fn keyspace(
  db db: Db(_, _),
  name name: String,
  key_codec key_codec: codec.Codec(k),
  value_codec value_codec: codec.Codec(v),
  key_compare key_compare: fn(k, k) -> order.Order,
) -> Keyspace(k, v) {
  let assert True = name != reserved_default_keyspace_name
  db.register_keyspace(
    subject: db.subject,
    name: name,
    byte_compare: adapt_compare(key_codec, key_compare),
    timeout: db.call_timeout,
  )
  Keyspace(
    name: name,
    key_codec: key_codec,
    value_codec: value_codec,
    key_compare: key_compare,
  )
}

fn adapt_compare(
  key_codec: codec.Codec(k),
  key_compare: fn(k, k) -> order.Order,
) -> fn(BitArray, BitArray) -> order.Order {
  fn(a: BitArray, b: BitArray) {
    let assert Ok(decoded_a) = key_codec.decode(a)
    let assert Ok(decoded_b) = key_codec.decode(b)
    key_compare(decoded_a, decoded_b)
  }
}

/// List the names of every keyspace currently registered on this database.
///
/// Returns names in sorted order. Includes every keyspace that has been
/// opened with `trove.keyspace(...)` in this session, plus every keyspace
/// that was persisted in the store file (even if not yet registered in this
/// session). Reading or writing a persisted-but-unregistered keyspace
/// without first calling `trove.keyspace(...)` panics.
///
/// ```gleam
/// let names = trove.list_keyspaces(db)
/// ```
pub fn list_keyspaces(db db: Db(_, _)) -> List(String) {
  db.list_keyspaces(subject: db.subject, timeout: db.call_timeout)
}

/// Insert or update a key-value pair in a named keyspace.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.put_in(db, keyspace: users, key: "alice", value: "admin")
/// ```
pub fn put_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
  value value: v,
) -> Nil {
  db.put_in(
    subject: db.subject,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
    value_bytes: keyspace.value_codec.encode(value),
    timeout: db.call_timeout,
  )
}

/// Look up a key in a named keyspace. Returns `Ok(value)` if found,
/// `Error(Nil)` if the key does not exist.
///
/// **Panics** on store I/O or decode errors (e.g. file corruption, codec
/// mismatch against on-disk bytes).
///
/// ```gleam
/// let assert Ok("admin") = trove.get_in(db, keyspace: users, key: "alice")
/// ```
pub fn get_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Result(v, Nil) {
  use value_bytes <- result.try(db.get_in(
    subject: db.subject,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
    timeout: db.call_timeout,
  ))
  keyspace.value_codec.decode(value_bytes)
}

/// Remove a key from a named keyspace. No error if the key does not exist.
///
/// **Panics** on store I/O errors.
///
/// ```gleam
/// trove.delete_in(db, keyspace: users, key: "alice")
/// ```
pub fn delete_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Nil {
  db.delete_in(
    subject: db.subject,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
    timeout: db.call_timeout,
  )
}

/// Check whether a key exists in a named keyspace.
///
/// **Panics** on store I/O errors.
///
/// ```gleam
/// let assert True = trove.has_key_in(db, keyspace: users, key: "alice")
/// ```
pub fn has_key_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Bool {
  db.has_key_in(
    subject: db.subject,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
    timeout: db.call_timeout,
  )
}

/// Returns the number of live entries in a named keyspace.
///
/// ```gleam
/// let n = trove.size_in(db, keyspace: users)
/// ```
pub fn size_in(db db: Db(_, _), keyspace keyspace: Keyspace(_, _)) -> Int {
  db.size_in(subject: db.subject, name: keyspace.name, timeout: db.call_timeout)
}

/// Atomically insert multiple key-value pairs into a named keyspace. A
/// single header write covers the entire batch.
///
/// **Panics** on store I/O errors.
///
/// ```gleam
/// trove.put_multi_in(
///   db,
///   keyspace: users,
///   entries: [#("alice", "admin"), #("bob", "member")],
/// )
/// ```
pub fn put_multi_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  entries entries: List(#(k, v)),
) -> Nil {
  put_and_delete_multi_in(db:, keyspace:, puts: entries, deletes: [])
}

/// Atomically delete multiple keys from a named keyspace.
///
/// **Panics** on store I/O errors.
///
/// ```gleam
/// trove.delete_multi_in(db, keyspace: users, keys: ["alice", "bob"])
/// ```
pub fn delete_multi_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  keys keys: List(k),
) -> Nil {
  put_and_delete_multi_in(db:, keyspace:, puts: [], deletes: keys)
}

/// Atomically insert and delete entries in a named keyspace under a single
/// header write. Puts are applied first, then deletes.
///
/// **Panics** on store I/O errors.
///
/// ```gleam
/// trove.put_and_delete_multi_in(
///   db,
///   keyspace: users,
///   puts: [#("bob", "admin")],
///   deletes: ["alice"],
/// )
/// ```
pub fn put_and_delete_multi_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  puts puts: List(#(k, v)),
  deletes deletes: List(k),
) -> Nil {
  let encoded_puts =
    list.map(puts, fn(pair) {
      #(keyspace.key_codec.encode(pair.0), keyspace.value_codec.encode(pair.1))
    })
  let encoded_deletes = list.map(deletes, keyspace.key_codec.encode)
  db.put_and_delete_multi_in(
    subject: db.subject,
    name: keyspace.name,
    puts: encoded_puts,
    deletes: encoded_deletes,
    timeout: db.call_timeout,
  )
}

/// Open a database at the configured path. Creates the directory if it does
/// not exist. If a store file already exists, recovers the tree from the
/// latest valid header.
///
/// ```gleam
/// let config = trove.Config(
///   path: "./my_db",
///   key_codec: codec.string(),
///   value_codec: codec.string(),
///   key_compare: string.compare,
///   auto_compact: trove.AutoCompact(min_dirt: 1000, min_dirt_factor: 0.25),
///   auto_file_sync: trove.AutoSync,
///   call_timeout: 5000,
/// )
/// let assert Ok(db) = trove.open(config)
/// ```
pub fn open(config: Config(k, v)) -> Result(Db(k, v), OpenError) {
  db.open(
    path: config.path,
    key_codec: config.key_codec,
    value_codec: config.value_codec,
    key_compare: config.key_compare,
    auto_file_sync: to_internal_file_sync(config.auto_file_sync),
    auto_compact: to_internal_auto_compact(config.auto_compact),
    call_timeout: config.call_timeout,
  )
  |> result.map(Db(_, config.call_timeout))
  |> result.map_error(map_open_error)
}

/// Close the database and release the file handle and path lock. Does not
/// fsync. If using `ManualSync`, call `file_sync` before closing to ensure
/// durability. The `Db` handle must not be used after calling this.
///
/// **Panics** if the store file handle cannot be closed.
///
/// ```gleam
/// trove.close(db)
/// ```
pub fn close(db: Db(k, v)) -> Nil {
  db.close(subject: db.subject, timeout: db.call_timeout)
}

/// Trigger a manual compaction. Rebuilds the store file keeping only live
/// entries, resetting the dirt factor to zero. Returns `Ok(Nil)` on success
/// or `Error(reason)` if compaction failed. On failure the database remains
/// functional with the original store file.
///
/// The timeout is separate from `call_timeout` because compaction can take
/// much longer than normal operations.
///
/// ```gleam
/// let assert Ok(Nil) = trove.compact(db, timeout: 60_000)
/// ```
pub fn compact(db: Db(k, v), timeout timeout: Int) -> Result(Nil, String) {
  db.compact(subject: db.subject, timeout: timeout)
}

/// Returns the number of live entries in the database.
///
/// ```gleam
/// let count = trove.size(db)
/// ```
pub fn size(db: Db(k, v)) -> Int {
  db.size(subject: db.subject, timeout: db.call_timeout)
}

/// Returns `True` if the database contains no entries.
///
/// ```gleam
/// let empty = trove.is_empty(db)
/// ```
pub fn is_empty(db: Db(k, v)) -> Bool {
  size(db) == 0
}

/// Returns the current dirt factor: a float between 0.0 and 1.0 that
/// approximates how much of the store file is occupied by superseded data.
/// Overwrites and deletes increment the dirt counter because they write new
/// nodes that make old ones unreachable. New inserts do not increment dirt
/// since they don't supersede existing data. The formula is
/// `dirt / (1 + size + dirt)`; the `+1` ensures the result is always
/// well-defined, even for an empty tree. The value approaches but never
/// reaches 1.0. Higher values mean more wasted space that compaction would
/// reclaim.
///
/// ```gleam
/// let df = trove.dirt_factor(db)
/// ```
pub fn dirt_factor(db: Db(k, v)) -> Float {
  db.dirt_factor(subject: db.subject, timeout: db.call_timeout)
}

/// Force an fsync of the store file to disk. Useful when `auto_file_sync`
/// is set to `ManualSync` and you want to control when data is flushed.
///
/// **Panics** if the fsync system call fails.
///
/// ```gleam
/// let config = trove.Config(..config, auto_file_sync: trove.ManualSync)
/// let assert Ok(db) = trove.open(config)
/// trove.put(db, key: "hello", value: "world")
/// trove.file_sync(db)
/// ```
pub fn file_sync(db: Db(k, v)) -> Nil {
  db.file_sync(subject: db.subject, timeout: db.call_timeout)
}

/// Change the auto-compaction setting at runtime.
///
/// ```gleam
/// trove.set_auto_compact(db, trove.AutoCompact(min_dirt: 1000, min_dirt_factor: 0.25))
/// ```
pub fn set_auto_compact(db: Db(k, v), setting setting: AutoCompact) -> Nil {
  db.set_auto_compact(
    subject: db.subject,
    setting: to_internal_auto_compact(setting),
    timeout: db.call_timeout,
  )
}

/// Look up a key. Returns `Ok(value)` if found, `Error(Nil)` if the
/// key does not exist.
///
/// **Panics** on store I/O or decode errors (e.g. file corruption).
///
/// ```gleam
/// let assert Ok("world") = trove.get(db, key: "hello")
/// ```
pub fn get(db: Db(k, v), key key: k) -> Result(v, Nil) {
  db.get(subject: db.subject, key: key, timeout: db.call_timeout)
}

/// Insert or update a key-value pair.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.put(db, key: "hello", value: "world")
/// ```
pub fn put(db: Db(k, v), key key: k, value value: v) -> Nil {
  db.put(subject: db.subject, key: key, value: value, timeout: db.call_timeout)
}

/// Remove a key. No error if the key does not exist.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.delete(db, key: "hello")
/// ```
pub fn delete(db: Db(k, v), key key: k) -> Nil {
  db.delete(subject: db.subject, key: key, timeout: db.call_timeout)
}

/// Check whether a key exists in the database.
///
/// **Panics** on store I/O or decode errors (e.g. file corruption).
///
/// ```gleam
/// let assert True = trove.has_key(db, key: "hello")
/// ```
pub fn has_key(db: Db(k, v), key key: k) -> Bool {
  db.has_key(subject: db.subject, key: key, timeout: db.call_timeout)
}

/// Atomically insert multiple key-value pairs. A single header write covers
/// the entire batch.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.put_multi(db, entries: [#("a", "1"), #("b", "2")])
/// ```
pub fn put_multi(db: Db(k, v), entries entries: List(#(k, v))) -> Nil {
  db.put_multi(subject: db.subject, entries: entries, timeout: db.call_timeout)
}

/// Atomically delete multiple keys.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.delete_multi(db, keys: ["a", "b"])
/// ```
pub fn delete_multi(db: Db(k, v), keys keys: List(k)) -> Nil {
  db.delete_multi(subject: db.subject, keys: keys, timeout: db.call_timeout)
}

/// Atomically insert and delete entries in a single operation. Puts are
/// applied first, then deletes, all under a single header write.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.put_and_delete_multi(
///   db,
///   puts: [#("new_key", "value")],
///   deletes: ["old_key"],
/// )
/// ```
pub fn put_and_delete_multi(
  db: Db(k, v),
  puts puts: List(#(k, v)),
  deletes deletes: List(k),
) -> Nil {
  db.put_and_delete_multi(
    subject: db.subject,
    puts: puts,
    deletes: deletes,
    timeout: db.call_timeout,
  )
}

/// A transaction handle for reading and writing within a transaction.
pub type Tx(k, v) =
  tx.Tx(k, v)

/// The result of a transaction callback. Return `Commit` to apply the
/// transaction's writes, or `Cancel` to discard them.
pub type TransactionResult(k, v, a) {
  /// Apply the transaction's writes and return the result value.
  Commit(tx: Tx(k, v), result: a)
  /// Discard the transaction's writes and return the result value.
  Cancel(result: a)
}

/// A point-in-time snapshot handle for consistent reads.
pub type Snapshot(k, v) =
  snapshot.Snapshot(k, v)

/// Run an atomic transaction. The callback receives a `Tx` handle and must
/// return `Commit(tx:, result: value)` to apply writes or
/// `Cancel(result: value)` to discard. The transaction holds exclusive
/// write access for its duration.
///
/// The `timeout` parameter (in milliseconds) controls how long the caller
/// waits for the transaction to complete, including queue wait time and
/// callback execution. Choose a value appropriate for your workload.
/// Queued operations or auto-compaction may delay the start, and a
/// long-running callback consumes the remaining budget.
///
/// **Important:** The callback runs inside the database actor. Do not call
/// any `trove` functions (such as `get`, `put`, `compact`, etc.) on the
/// same `Db` handle from within the callback; this will deadlock the actor
/// until the call timeout fires. Use the `Tx` handle (`tx_get`, `tx_put`,
/// `tx_delete`) for all reads and writes inside the transaction.
///
/// **Panics** if the `Commit` variant contains a stale or replaced `Tx`
/// handle (e.g. the original handle instead of the latest one returned by
/// `tx_put`/`tx_delete`).
///
/// **Non-escaping:** The `Tx` handle is only valid inside the callback.
/// Do not store it in a variable, send it to another process, or return it.
/// Using a `Tx` after the callback returns will panic or produce undefined
/// behavior.
///
/// **Timeout semantics:** If the timeout fires while the callback is still
/// executing, the caller panics but the actor continues running the callback
/// to completion. This means writes may be durably committed even though the
/// caller observes a timeout failure. Choose a timeout that accommodates your
/// expected callback duration and any queued operations ahead of it.
///
/// ```gleam
/// let result = trove.transaction(db, timeout: 5000, callback: fn(tx) {
///   let tx = trove.tx_put(tx, key: "key", value: "value")
///   trove.Commit(tx:, result: "done")
/// })
/// ```
pub fn transaction(
  db: Db(k, v),
  timeout timeout: Int,
  callback callback: fn(Tx(k, v)) -> TransactionResult(k, v, a),
) -> a {
  let result_subject = process.new_subject()
  let token = reference.new()

  let run = fn(transaction) {
    let nonce_subject = process.new_subject()
    let transaction = tx.set_token(tx: transaction, token: token)
    let transaction =
      tx.set_nonce_tracker(tx: transaction, tracker: option.Some(nonce_subject))
    case exception.rescue(fn() { callback(transaction) }) {
      Ok(Commit(tx_inner, value)) -> {
        case
          exception.rescue(fn() {
            let assert True = tx.token(tx: tx_inner) == token
            let latest_nonce = drain_latest(nonce_subject)
            case latest_nonce {
              option.None -> Nil
              option.Some(expected) -> {
                let assert True = tx.nonce(tx: tx_inner) == expected
                Nil
              }
            }
          })
        {
          Ok(Nil) -> {
            process.send(result_subject, Ok(value))
            db.CommitOutcome(
              tx.get_tree(tx: tx_inner),
              tx.get_other_trees(tx: tx_inner),
            )
          }
          Error(ex) -> {
            process.send(result_subject, Error(ex))
            db.CancelOutcome
          }
        }
      }
      Ok(Cancel(value)) -> {
        process.send(result_subject, Ok(value))
        db.CancelOutcome
      }
      Error(ex) -> {
        process.send(result_subject, Error(ex))
        db.CancelOutcome
      }
    }
  }

  // The `run` closure sends to `result_subject` before the actor sends its
  // own reply, so by the time `db.transaction` (a blocking call) returns the
  // message is guaranteed to be in our mailbox.  A 0-ms receive is safe here.
  db.transaction(subject: db.subject, timeout: timeout, run: run)
  let assert Ok(result) = process.receive(result_subject, 0)
  case result {
    Ok(value) -> value
    Error(ex) -> reraise(ex)
  }
}

fn drain_latest(subject: process.Subject(a)) -> option.Option(a) {
  drain_latest_loop(subject, option.None)
}

fn drain_latest_loop(
  subject: process.Subject(a),
  acc: option.Option(a),
) -> option.Option(a) {
  case process.receive(subject, 0) {
    Ok(value) -> drain_latest_loop(subject, option.Some(value))
    Error(Nil) -> acc
  }
}

/// Read a key within a transaction. Sees writes made earlier in the same
/// transaction. Returns `Error(Nil)` if the key does not exist.
///
/// **Panics** on store I/O or decode errors (e.g. file corruption).
///
/// ```gleam
/// trove.transaction(db, timeout: 5000, callback: fn(tx) {
///   let assert Ok(current) = trove.tx_get(tx, key: "counter")
///   let tx = trove.tx_put(tx, key: "counter", value: current <> "!")
///   trove.Commit(tx:, result: Nil)
/// })
/// ```
pub fn tx_get(tx tx: Tx(k, v), key key: k) -> Result(v, Nil) {
  tx.get(tx: tx, key: key)
}

/// Write a key-value pair within a transaction. Returns the updated `Tx`.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.transaction(db, timeout: 5000, callback: fn(tx) {
///   let tx = trove.tx_put(tx, key: "greeting", value: "hello")
///   trove.Commit(tx:, result: Nil)
/// })
/// ```
pub fn tx_put(tx tx: Tx(k, v), key key: k, value value: v) -> Tx(k, v) {
  tx.put(tx: tx, key: key, value: value)
}

/// Delete a key within a transaction. Returns the updated `Tx`.
///
/// **Panics** on store I/O errors (e.g. disk full, file corruption).
///
/// ```gleam
/// trove.transaction(db, timeout: 5000, callback: fn(tx) {
///   let tx = trove.tx_delete(tx, key: "old_key")
///   trove.Commit(tx:, result: Nil)
/// })
/// ```
pub fn tx_delete(tx tx: Tx(k, v), key key: k) -> Tx(k, v) {
  tx.delete(tx: tx, key: key)
}

/// Check whether a key exists within a transaction. Sees writes made
/// earlier in the same transaction.
///
/// **Panics** on store I/O or decode errors (e.g. file corruption).
///
/// ```gleam
/// trove.transaction(db, timeout: 5000, callback: fn(tx) {
///   let exists = trove.tx_has_key(tx, key: "counter")
///   trove.Commit(tx:, result: exists)
/// })
/// ```
pub fn tx_has_key(tx tx: Tx(k, v), key key: k) -> Bool {
  tx_get(tx: tx, key: key) |> result.is_ok
}

/// Look up a key in a named keyspace within a transaction. Sees writes made
/// earlier in the same transaction.
///
/// **Panics** if the keyspace has not been registered in this session via
/// `trove.keyspace(...)`, or on store I/O or decode errors.
pub fn tx_get_in(
  tx tx: Tx(k_default, v_default),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Result(v, Nil) {
  use value_bytes <- result.try(tx.get_in(
    tx: tx,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
  ))
  keyspace.value_codec.decode(value_bytes)
}

/// Insert or update a key-value pair in a named keyspace within a
/// transaction. Returns the updated `Tx`.
///
/// **Panics** if the keyspace has not been registered in this session via
/// `trove.keyspace(...)`, or on store I/O errors.
pub fn tx_put_in(
  tx tx: Tx(k_default, v_default),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
  value value: v,
) -> Tx(k_default, v_default) {
  tx.put_in(
    tx: tx,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
    value_bytes: keyspace.value_codec.encode(value),
  )
}

/// Delete a key from a named keyspace within a transaction.
///
/// **Panics** if the keyspace has not been registered in this session via
/// `trove.keyspace(...)`, or on store I/O errors.
pub fn tx_delete_in(
  tx tx: Tx(k_default, v_default),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Tx(k_default, v_default) {
  tx.delete_in(
    tx: tx,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
  )
}

/// Check whether a key exists in a named keyspace within a transaction.
///
/// **Panics** if the keyspace has not been registered in this session via
/// `trove.keyspace(...)`, or on store I/O or decode errors.
pub fn tx_has_key_in(
  tx tx: Tx(k_default, v_default),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Bool {
  tx_get_in(tx: tx, keyspace: keyspace, key: key) |> result.is_ok
}

/// Run a callback with a point-in-time snapshot. The snapshot sees the state
/// of the database at the moment it was acquired; subsequent writes are
/// invisible to it.
///
/// **Non-escaping:** The `Snapshot` handle is only valid inside the callback.
/// Do not store it in a variable, send it to another process, or return it.
/// Using a `Snapshot` after the callback returns will panic or produce
/// undefined behavior because the underlying file handle is closed on exit.
///
/// **Panics** if the snapshot file handle cannot be opened.
///
/// ```gleam
/// let result = trove.with_snapshot(db, fn(snap) {
///   trove.snapshot_get(snapshot: snap, key: "my_key")
/// })
/// // result: Result(String, Nil)
/// ```
pub fn with_snapshot(
  db: Db(k, v),
  callback callback: fn(Snapshot(k, v)) -> a,
) -> a {
  let assert Ok(snap) =
    db.acquire_snapshot(subject: db.subject, timeout: db.call_timeout)
  use <- exception.defer(fn() { snapshot.close(snap) })
  callback(snap)
}

/// Look up a key in a snapshot. Returns `Error(Nil)` if the key does
/// not exist.
///
/// **Panics** on store read or decode errors (e.g. file corruption).
///
/// ```gleam
/// trove.with_snapshot(db, fn(snap) {
///   let assert Ok(value) = trove.snapshot_get(snapshot: snap, key: "my_key")
///   value
/// })
/// ```
pub fn snapshot_get(
  snapshot snapshot: Snapshot(k, v),
  key key: k,
) -> Result(v, Nil) {
  snapshot.get(snapshot: snapshot, key: key)
}

/// Iterate over entries in a snapshot within optional key bounds.
/// Returns a lazy `Yielder` that streams entries from disk on demand,
/// reading only one leaf node at a time.
///
/// The yielder holds a reference to the snapshot's file handle, so it
/// must be consumed before the snapshot is closed. For large ranges,
/// prefer this over `range` to avoid loading all entries into memory.
///
/// **Panics** on store read or decode errors during iteration
/// (e.g. file corruption).
///
/// Use `range.Inclusive(key)` or `range.Exclusive(key)` for bounds,
/// or `option.None` for unbounded. Use `range.Forward` or `range.Reverse`
/// for direction.
///
/// ```gleam
/// import gleam/option.{None, Some}
/// import gleam/yielder
/// import trove/range
///
/// let entries = trove.with_snapshot(db, fn(snap) {
///   let y = trove.snapshot_range(
///     snapshot: snap,
///     min: Some(range.Inclusive("a")),
///     max: None,
///     direction: range.Forward,
///   )
///   yielder.to_list(y)
/// })
/// ```
pub fn snapshot_range(
  snapshot snapshot: Snapshot(k, v),
  min min: option.Option(range.Bound(k)),
  max max: option.Option(range.Bound(k)),
  direction direction: range.Direction,
) -> yielder.Yielder(#(k, v)) {
  snapshot.range(snapshot: snapshot, min: min, max: max, direction: direction)
}

/// Iterate over entries in the database within optional key bounds.
/// Returns a `List` of key-value pairs.
///
/// For large result sets, use `with_snapshot` and `snapshot_range` instead
/// to stream entries lazily without loading them all at once.
///
/// **Panics** if the snapshot file handle cannot be opened, or on store
/// read/decode errors during iteration.
///
/// Use `range.Inclusive(key)` or `range.Exclusive(key)` for bounds,
/// or `option.None` for unbounded. Use `range.Forward` or `range.Reverse`
/// for direction.
///
/// ```gleam
/// import gleam/option.{Some}
/// import trove/range
///
/// let results =
///   trove.range(
///     db,
///     min: Some(range.Inclusive("a")),
///     max: Some(range.Exclusive("z")),
///     direction: range.Forward,
///   )
/// ```
pub fn range(
  db db: Db(k, v),
  min min: option.Option(range.Bound(k)),
  max max: option.Option(range.Bound(k)),
  direction direction: range.Direction,
) -> List(#(k, v)) {
  let assert Ok(snap) =
    db.acquire_snapshot(subject: db.subject, timeout: db.call_timeout)
  use <- exception.defer(fn() { snapshot.close(snap) })
  snapshot.range(snapshot: snap, min: min, max: max, direction: direction)
  |> yielder.to_list()
}

/// Look up a key in a named keyspace within a snapshot.
///
/// **Panics** if the keyspace was not registered before the snapshot was
/// acquired, or on store read or decode errors.
///
/// ```gleam
/// trove.with_snapshot(db, fn(snap) {
///   trove.snapshot_get_in(snap, keyspace: users, key: "alice")
/// })
/// ```
pub fn snapshot_get_in(
  snapshot snapshot: Snapshot(_, _),
  keyspace keyspace: Keyspace(k, v),
  key key: k,
) -> Result(v, Nil) {
  use value_bytes <- result.try(snapshot.get_in(
    snapshot: snapshot,
    name: keyspace.name,
    key_bytes: keyspace.key_codec.encode(key),
  ))
  keyspace.value_codec.decode(value_bytes)
}

/// Iterate over entries in a named keyspace within a snapshot. Returns a
/// lazy `Yielder` streaming entries from disk.
///
/// The yielder holds a reference to the snapshot's file handle; consume it
/// before the snapshot closes.
///
/// **Panics** if the keyspace was not registered before the snapshot was
/// acquired, or on store read or decode errors during iteration.
///
/// ```gleam
/// trove.with_snapshot(db, fn(snap) {
///   trove.snapshot_range_in(
///     snap,
///     keyspace: users,
///     min: Some(range.Inclusive("a")),
///     max: None,
///     direction: range.Forward,
///   )
///   |> yielder.to_list
/// })
/// ```
pub fn snapshot_range_in(
  snapshot snapshot: Snapshot(_, _),
  keyspace keyspace: Keyspace(k, v),
  min min: option.Option(range.Bound(k)),
  max max: option.Option(range.Bound(k)),
  direction direction: range.Direction,
) -> yielder.Yielder(#(k, v)) {
  snapshot.range_in(
    snapshot: snapshot,
    name: keyspace.name,
    min: encode_bound(min, keyspace.key_codec),
    max: encode_bound(max, keyspace.key_codec),
    direction: direction,
  )
  |> yielder.map(fn(pair) {
    let assert Ok(k) = keyspace.key_codec.decode(pair.0)
    let assert Ok(v) = keyspace.value_codec.decode(pair.1)
    #(k, v)
  })
}

fn encode_bound(
  bound: option.Option(range.Bound(k)),
  key_codec: codec.Codec(k),
) -> option.Option(range.Bound(BitArray)) {
  option.map(bound, fn(b) {
    case b {
      range.Inclusive(v) -> range.Inclusive(key_codec.encode(v))
      range.Exclusive(v) -> range.Exclusive(key_codec.encode(v))
    }
  })
}

/// Iterate over entries in a named keyspace within optional key bounds.
/// Returns a `List` of key-value pairs. For large result sets, use
/// `with_snapshot` and `snapshot_range_in` instead.
///
/// **Panics** if the keyspace has not been registered in this session via
/// `trove.keyspace(...)`, if the snapshot file handle cannot be opened,
/// or on store read or decode errors during iteration.
///
/// ```gleam
/// let results = trove.range_in(
///   db,
///   keyspace: users,
///   min: Some(range.Inclusive("a")),
///   max: Some(range.Exclusive("z")),
///   direction: range.Forward,
/// )
/// ```
pub fn range_in(
  db db: Db(_, _),
  keyspace keyspace: Keyspace(k, v),
  min min: option.Option(range.Bound(k)),
  max max: option.Option(range.Bound(k)),
  direction direction: range.Direction,
) -> List(#(k, v)) {
  let assert Ok(snap) =
    db.acquire_snapshot(subject: db.subject, timeout: db.call_timeout)
  use <- exception.defer(fn() { snapshot.close(snap) })
  snapshot_range_in(
    snapshot: snap,
    keyspace: keyspace,
    min: min,
    max: max,
    direction: direction,
  )
  |> yielder.to_list
}

fn map_open_error(error: db.OpenError) -> OpenError {
  case error {
    db.DirectoryError(reason) -> DirectoryError(reason)
    db.StoreError(reason) -> StoreError(reason)
    db.LockError(reason) -> LockError(reason)
    db.ActorStartError -> ActorStartError
  }
}

fn to_internal_auto_compact(setting: AutoCompact) -> db.AutoCompact {
  case setting {
    AutoCompact(min_dirt:, min_dirt_factor:) ->
      db.AutoCompact(min_dirt:, min_dirt_factor:)
    NoAutoCompact -> db.NoAutoCompact
  }
}

fn to_internal_file_sync(setting: FileSync) -> db.FileSync {
  case setting {
    AutoSync -> db.AutoSync
    ManualSync -> db.ManualSync
  }
}