///|
/// Main database handle.
pub struct DB {
  dir : String
  keydir : Keydir
  mut active_file_id : Int
  mut active_file_offset : Int
  max_file_size : Int
  mut logical_clock : Int64
}

///|
/// Helper to check if a file name ends with ".data".
fn ends_with_data(name : String) -> Bool {
  let len = name.length()
  if len > 5 {
    name[len - 5:] == ".data"
  } else {
    false
  }
}

///|
/// Helper to check if a string starts with a prefix.
fn starts_with(s : String, prefix : String) -> Bool {
  let len = s.length()
  let plen = prefix.length()
  if len >= plen {
    s[:plen] == prefix[:]
  } else {
    false
  }
}

///|
/// Reject keys that would be interpreted as internal transaction records.
fn validate_user_key(key : String) -> Unit raise MoonKVError {
  if starts_with(key, "__tx__:") || starts_with(key, "__tx_commit__:") {
    raise MoonKVError::InvalidRecord(
      "Key uses a reserved transaction namespace: \{key}",
    )
  }
}

///|
/// Helper to parse a 32-bit signed integer.
fn parse_int(s : String) -> Int raise MoonKVError {
  let mut val = 0
  let len = s.length()
  if len == 0 {
    raise MoonKVError::InvalidRecord("Empty integer string")
  }
  let mut start = 0
  let is_neg = if s[0] == '-' {
    start = 1
    true
  } else {
    false
  }
  for i = start; i < len; i = i + 1 {
    let ch = s[i]
    if ch >= '0' && ch <= '9' {
      let digit = ch.to_int() - '0'.to_int()
      val = val * 10 + digit
    } else {
      raise MoonKVError::InvalidRecord("Invalid digit in integer string: \{s}")
    }
  }
  if is_neg {
    -val
  } else {
    val
  }
}

///|
/// Helper to parse a 64-bit signed integer.
fn parse_int64(s : String) -> Int64 raise MoonKVError {
  let mut val = 0L
  let len = s.length()
  if len == 0 {
    raise MoonKVError::InvalidRecord("Empty integer string")
  }
  let mut start = 0
  let is_neg = if s[0] == '-' {
    start = 1
    true
  } else {
    false
  }
  for i = start; i < len; i = i + 1 {
    let ch = s[i]
    if ch >= '0' && ch <= '9' {
      let digit = (ch.to_int() - '0'.to_int()).to_int64()
      val = val * 10L + digit
    } else {
      raise MoonKVError::InvalidRecord("Invalid digit in integer string: \{s}")
    }
  }
  if is_neg {
    -val
  } else {
    val
  }
}

///|
/// Helper to parse the key of a transaction record: "__tx__::"
fn parse_tx_key(key : String) -> (Int64, String) raise MoonKVError {
  let s = key[7:]
  let len = s.length()
  let mut colon_idx = -1
  for i = 0; i < len; i = i + 1 {
    if s[i] == ':' {
      colon_idx = i
      break
    }
  }
  if colon_idx == -1 {
    raise MoonKVError::InvalidRecord("Invalid tx key format: \{key}")
  }
  let tx_id_str = s[:colon_idx].to_owned()
  let actual_key = s[colon_idx + 1:].to_owned()
  let tx_id = parse_int64(tx_id_str)
  (tx_id, actual_key)
}

///|
/// Helper to parse the key of a transaction commit record: "__tx_commit__:"
fn parse_commit_key(key : String) -> Int64 raise MoonKVError {
  let tx_id_str = key[14:].to_owned()
  parse_int64(tx_id_str)
}

///| Opens a database instance in the specified directory.

///|
/// Rebuilds the in-memory index from existing data files.
pub fn DB::open(dir : String, max_file_size : Int) -> DB raise MoonKVError {
  if max_file_size <= 0 {
    raise MoonKVError::InvalidRecord("max_file_size must be greater than zero")
  }

  try {
    if @fs.path_exists(dir) == false {
      @fs.create_dir(dir)
    }
  } catch {
    _ =>
      raise MoonKVError::IOError("Failed to create database directory \{dir}")
  }

  let keydir = Keydir::new()

  let files = @fs.read_dir(dir) catch {
    _ => raise MoonKVError::IOError("Failed to read database directory \{dir}")
  }

  let data_file_ids = Array::new()
  for file in files {
    if ends_with_data(file) {
      let id_str = file[:file.length() - 5].to_owned()
      let id = parse_int(id_str)
      data_file_ids.push(id)
    }
  }

  data_file_ids.sort()

  let mut active_file_id = 0
  let mut active_file_offset = 0
  let mut max_ts = 0L

  // uncommitted_txs maps transaction ID to records written in that transaction
  let uncommitted_txs : Map[Int64, Array[LogRecord]] = Map([])

  for id in data_file_ids {
    let hint_path = dir + "/" + id.to_string() + ".hint"
    let has_hint = @fs.path_exists(hint_path)

    if has_hint {
      let hint_records = read_hint_records_from_file(hint_path)
      for hint_rec in hint_records {
        let key_str = bytes_to_string(hint_rec.key)
        if hint_rec.val_sz < 0 {
          keydir.delete(key_str)
        } else if hint_rec.expires_at > 0L && hint_rec.expires_at <= max_ts {
          keydir.delete(key_str)
        } else {
          let entry = KeyEntry::new(
            id,
            hint_rec.val_sz,
            hint_rec.value_pos,
            hint_rec.timestamp,
            hint_rec.expires_at,
          )
          keydir.put(key_str, entry)
        }
        if hint_rec.timestamp > max_ts {
          max_ts = hint_rec.timestamp
        }
      }
    } else {
      let path = dir + "/" + id.to_string() + ".data"
      let records = read_records_from_file(path)
      for log_rec in records {
        if log_rec.record.timestamp > max_ts {
          max_ts = log_rec.record.timestamp
        }

        let key_str = bytes_to_string(log_rec.record.key)
        if starts_with(key_str, "__tx__:") {
          let (tx_id, _) = parse_tx_key(key_str)
          match uncommitted_txs.get(tx_id) {
            Some(arr) => arr.push(log_rec)
            None => uncommitted_txs.set(tx_id, [log_rec])
          }
        } else if starts_with(key_str, "__tx_commit__:") {
          let tx_id = parse_commit_key(key_str)
          match uncommitted_txs.get(tx_id) {
            Some(recs) => {
              for rec in recs {
                let tx_k = bytes_to_string(rec.record.key)
                let (_, actual_k) = parse_tx_key(tx_k)
                if rec.record.is_tombstone {
                  keydir.delete(actual_k)
                } else if rec.record.expires_at > 0L &&
                  rec.record.expires_at <= max_ts {
                  keydir.delete(actual_k)
                } else {
                  let entry = KeyEntry::new(
                    id,
                    rec.record.value.length(),
                    rec.offset.to_int64(),
                    rec.record.timestamp,
                    rec.record.expires_at,
                  )
                  keydir.put(actual_k, entry)
                }
              }
              uncommitted_txs.remove(tx_id)
            }
            None => ()
          }
          // Normal record
        } else if log_rec.record.is_tombstone {
          keydir.delete(key_str)
        } else if log_rec.record.expires_at > 0L &&
          log_rec.record.expires_at <= max_ts {
          keydir.delete(key_str)
        } else {
          let entry = KeyEntry::new(
            id,
            log_rec.record.value.length(),
            log_rec.offset.to_int64(),
            log_rec.record.timestamp,
            log_rec.record.expires_at,
          )
          keydir.put(key_str, entry)
        }
      }
    }
    active_file_id = id
  }

  if data_file_ids.length() > 0 {
    let path = dir + "/" + active_file_id.to_string() + ".data"
    try {
      if @fs.path_exists(path) {
        let file_bytes = @fs.read_file_to_bytes(path)
        active_file_offset = file_bytes.length()
      }
    } catch {
      _ => ()
    }
  }

  let logical_clock = max_ts + 1L

  {
    dir,
    keydir,
    active_file_id,
    active_file_offset,
    max_file_size,
    logical_clock,
  }
}

///|
/// Writes a key-value pair to the database.
pub fn DB::put(
  self : DB,
  key : String,
  value : Bytes,
) -> Unit raise MoonKVError {
  validate_user_key(key)
  let timestamp = self.logical_clock
  self.logical_clock = self.logical_clock + 1L

  let record = Record::new(string_to_bytes(key), value, timestamp)
  let record_bytes = record.serialize()
  let record_len = record_bytes.length()

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

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

  let entry = KeyEntry::new(
    self.active_file_id,
    value.length(),
    self.active_file_offset.to_int64(),
    timestamp,
    0L,
  )
  self.keydir.put(key, entry)

  self.active_file_offset = self.active_file_offset + record_len
}

///|
/// Retrieves a value from the database by its key.
pub fn DB::get(self : DB, key : String) -> Bytes raise MoonKVError {
  let entry = match self.keydir.get(key) {
    Some(e) => e
    None => raise MoonKVError::KeyNotFound("Key not found: \{key}")
  }

  // Check if expired
  if entry.expires_at > 0L && self.logical_clock >= entry.expires_at {
    self.keydir.delete(key) // Lazy delete from memory
    raise MoonKVError::KeyNotFound("Key not found (expired): \{key}")
  }

  let path = self.dir + "/" + entry.file_id.to_string() + ".data"
  let file_bytes = @fs.read_file_to_bytes(path) catch {
    _ => raise MoonKVError::IOError("Failed to read data file: \{path}")
  }

  let start = entry.value_pos.to_int()
  let file_len = file_bytes.length()
  if start + 28 > file_len {
    raise MoonKVError::CorruptedDatabase(
      "Record header out of bounds for key: \{key}",
    )
  }

  // Extract and decode header to find full record size
  let header_arr = Array::make(28, (0).to_byte())
  for i = 0; i < 28; i = i + 1 {
    header_arr[i] = file_bytes[start + i]
  }
  let header = Bytes::from_array(header_arr)
  let (_, _, _, key_sz, val_sz) = decode_header(header)

  let val_len = if val_sz < 0 { 0 } else { val_sz }
  let record_sz = 28 + key_sz + val_len

  if start + record_sz > file_len {
    raise MoonKVError::CorruptedDatabase(
      "Full record out of bounds for key: \{key}",
    )
  }

  // Extract and deserialize full record
  let record_arr = Array::make(record_sz, (0).to_byte())
  for i = 0; i < record_sz; i = i + 1 {
    record_arr[i] = file_bytes[start + i]
  }
  let record_bytes = Bytes::from_array(record_arr)
  let record = deserialize_record(record_bytes)

  if record.is_tombstone {
    raise MoonKVError::KeyNotFound("Key not found (tombstone): \{key}")
  }

  record.value
}

///|
/// Deletes a key-value pair from the database by writing a tombstone.
pub fn DB::delete(self : DB, key : String) -> Unit raise MoonKVError {
  validate_user_key(key)
  let entry = match self.keydir.get(key) {
    Some(e) => e
    None => raise MoonKVError::KeyNotFound("Key not found: \{key}")
  }

  // Check if already expired
  if entry.expires_at > 0L && self.logical_clock >= entry.expires_at {
    self.keydir.delete(key)
    raise MoonKVError::KeyNotFound("Key not found: \{key}")
  }

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

  let record = Record::new_tombstone(string_to_bytes(key), timestamp)
  let record_bytes = record.serialize()
  let record_len = record_bytes.length()

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

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

  self.keydir.delete(key)
  self.active_file_offset = self.active_file_offset + record_len
}

///|
/// Helper to close/clean resources (currently a no-op).
pub fn DB::close(self : DB) -> Unit {
  let _ = self
}