///|
/// WriteBatch stores a sequence of database operations to be committed atomically.
pub struct WriteBatch {
  db : DB
  // (key, value, ttl, has_ttl); None value means Delete.
  ops : Array[(String, Bytes?, Int64, Bool)]
}

///|
/// Creates a new WriteBatch for the database.
pub fn DB::new_write_batch(self : DB) -> WriteBatch {
  { db: self, ops: Array::new() }
}

///|
/// Adds a Put operation to the batch.
pub fn WriteBatch::put(self : WriteBatch, key : String, value : Bytes) -> Unit {
  self.ops.push((key, Some(value), 0L, false))
}

///|
/// Adds a Put operation with TTL to the batch.
pub fn WriteBatch::put_ttl(
  self : WriteBatch,
  key : String,
  value : Bytes,
  ttl : Int64,
) -> Unit {
  self.ops.push((key, Some(value), ttl, true))
}

///|
/// Adds a Delete operation to the batch.
pub fn WriteBatch::delete(self : WriteBatch, key : String) -> Unit {
  self.ops.push((key, None, 0L, false))
}

///|
/// Commits all operations in the batch atomically to the WAL log.
pub fn WriteBatch::commit(self : WriteBatch) -> Unit raise MoonKVError {
  if self.ops.length() == 0 {
    return
  }

  // Validate the complete batch before writing its first record. This keeps
  // invalid input from creating a partially committed transaction.
  for op in self.ops {
    let (key, _, ttl, has_ttl) = op
    validate_user_key(key)
    if has_ttl && ttl <= 0L {
      raise MoonKVError::InvalidRecord("ttl must be greater than zero")
    }
  }

  // 1. Allocate a unique transaction ID from the logical clock
  let tx_id = self.db.logical_clock
  self.db.logical_clock = self.db.logical_clock + 1L

  let entries = Array::new()

  // 2. Write all operations prefixed with transaction markers
  for op in self.ops {
    let (key, val_opt, ttl, has_ttl) = op
    let tx_key = "__tx__:" + tx_id.to_string() + ":" + key

    let timestamp = self.db.logical_clock
    self.db.logical_clock = self.db.logical_clock + 1L

    let record = match val_opt {
      Some(val) =>
        if has_ttl {
          Record::new_ttl(string_to_bytes(tx_key), val, timestamp, ttl)
        } else {
          Record::new(string_to_bytes(tx_key), val, timestamp)
        }
      None => Record::new_tombstone(string_to_bytes(tx_key), timestamp)
    }

    let record_bytes = record.serialize()
    let record_len = record_bytes.length()

    // Roll file if size limit exceeded
    if self.db.active_file_offset + record_len > self.db.max_file_size {
      self.db.active_file_id = self.db.active_file_id + 1
      self.db.active_file_offset = 0
    }

    let path = self.db.dir + "/" + self.db.active_file_id.to_string() + ".data"
    append_bytes_to_file(path, record_bytes)

    // Buffer KeyEntry metadata
    let is_put = match val_opt {
      Some(_) => true
      None => false
    }
    let val_len = match val_opt {
      Some(v) => v.length()
      None => 0
    }
    let expires_at = if has_ttl { timestamp + ttl } else { 0L }

    let entry = KeyEntry::new(
      self.db.active_file_id,
      val_len,
      self.db.active_file_offset.to_int64(),
      timestamp,
      expires_at,
    )
    entries.push((key, is_put, entry))

    self.db.active_file_offset = self.db.active_file_offset + record_len
  }

  // 3. Write the commit marker record to make the transaction active
  let commit_key = "__tx_commit__:" + tx_id.to_string()
  let timestamp = self.db.logical_clock
  self.db.logical_clock = self.db.logical_clock + 1L

  let commit_record = Record::new(
    string_to_bytes(commit_key),
    Bytes::make(0, (0).to_byte()),
    timestamp,
  )
  let commit_bytes = commit_record.serialize()
  let commit_len = commit_bytes.length()

  if self.db.active_file_offset + commit_len > self.db.max_file_size {
    self.db.active_file_id = self.db.active_file_id + 1
    self.db.active_file_offset = 0
  }

  let path = self.db.dir + "/" + self.db.active_file_id.to_string() + ".data"
  append_bytes_to_file(path, commit_bytes)
  self.db.active_file_offset = self.db.active_file_offset + commit_len

  // 4. Safely update the in-memory Keydir now that it's committed on disk
  for ent in entries {
    let (key, is_put, entry) = ent
    if is_put {
      self.db.keydir.put(key, entry)
    } else {
      self.db.keydir.delete(key)
    }
  }
}