///|
/// 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
}