///|
/// Write-ahead log, native file layer. Requires `targets: native` in
/// moon.pkg (uses moonbitlang/async/fs).
///
/// Durability protocol:
/// - `add_durable` / `delete_durable`: append one CRC-framed entry to
///   `dir/wal.log` with data fsync, THEN apply it in memory. A crash
///   between the two leaves the entry in the WAL; recovery replays it.
/// - `checkpoint`: serialize the whole index to `segments.msk.tmp`,
///   fsync, atomically rename over `segments.msk`, then truncate the
///   WAL. A crash before the rename leaves the old snapshot + full WAL
///   (replayed on open); a crash after the rename but before truncation
///   replays already-checkpointed entries, which is idempotent because
///   Add replaces by id and Delete is a tombstone.
/// - `open_or_create`: load `segments.msk` if present (corrupt snapshot
///   is an error, never silently dropped), then replay `wal.log` on top.
///   A torn WAL tail replays only the complete prefix.

///|
/// Open the index stored in `dir`, or create a fresh one. Replays the
/// WAL on top of the last checkpoint. The directory is created if missing.
pub async fn SegmentedIndex::open_or_create(
  dir : String,
  schema : Array[FieldSchema],
) -> SegmentedIndex raise StoreError {
  @fs.mkdir(dir, recursive=true, allow_exist=true) catch {
    e => raise StoreError::IoError(e.to_string())
  }
  let snapshot_path = dir + "/segments.msk"
  let has_snapshot = @fs.exists(snapshot_path) catch {
    e => raise StoreError::IoError(e.to_string())
  }
  let seg = if has_snapshot {
    let data = @fs.read_file(snapshot_path) catch {
      e => raise StoreError::IoError(e.to_string())
    }
    SegmentedIndex::from_bytes(data.binary()) catch {
      e => raise StoreError::CorruptIndex(@debug.to_string(e))
    }
  } else {
    SegmentedIndex::new(schema)
  }
  let wal_path = dir + "/wal.log"
  let has_wal = @fs.exists(wal_path) catch {
    e => raise StoreError::IoError(e.to_string())
  }
  if has_wal {
    let data = @fs.read_file(wal_path) catch {
      e => raise StoreError::IoError(e.to_string())
    }
    let entries = wal_decode(data.binary()) catch {
      e => raise StoreError::CorruptIndex(@debug.to_string(e))
    }
    // Schema guard BEFORE any replay: an Add referencing a field absent
    // from the snapshot schema would abort() deep inside Index::add
    // (uncatchable, process dies on open). Fail as a catchable
    // StoreError instead so callers can quarantine the directory.
    let known : @hashmap.HashMap[String, Unit] = @hashmap.HashMap([])
    for field in schema {
      known.set(field.name, ())
    }
    for entry in entries {
      match entry {
        Add(doc) =>
          for (name, _) in doc.fields {
            if !known.contains(name) {
              raise StoreError::CorruptIndex(
                "wal references unknown field: " + name,
              )
            }
          }
        Delete(_) => ()
      }
    }
    for entry in entries {
      match entry {
        Add(doc) => seg.add(doc)
        Delete(id) => ignore(seg.delete(id))
      }
    }
  }
  seg
}

///|
/// Atomically persist the full index state and clear the WAL.
/// Write temp + fsync + rename, then truncate the log.
pub async fn SegmentedIndex::checkpoint(
  self : SegmentedIndex,
  dir : String,
) -> Unit raise StoreError {
  let image = self.to_bytes()
  let tmp = dir + "/segments.msk.tmp"
  let dest = dir + "/segments.msk"
  try {
    @fs.write_file(tmp, image, sync=Data)
    @fs.rename(tmp, dest)
    // Order guarantee between the rename and the WAL truncation is what
    // makes the crash window below safe: without syncing the directory
    // entry, a power loss can persist the truncate before the rename,
    // leaving "old snapshot + empty WAL" — every write since the last
    // checkpoint lost even though checkpoint() returned. Sync the
    // directory now so the rename is durable BEFORE the WAL is emptied.
    // Windows: FlushFileBuffers needs write access on the directory
    // handle; POSIX: fsync on a ReadOnly fd is enough. Without this
    // (wasm/js), skip — those targets have no durability contract here.
    sync_directory(dir)
    // The snapshot now covers everything; the WAL may be discarded.
    // A crash between the dir sync and truncate is safe: replaying the
    // WAL on top of the new snapshot is idempotent.
    @fs.write_file(dir + "/wal.log", b"", sync=Data)
  } catch {
    e => raise StoreError::IoError(e.to_string())
  }
}

///|
// fsync the directory entry itself so a rename inside `dir` is durable
// before subsequent writes. Windows: FlushFileBuffers rejects a
// GENERIC_READ directory handle (Access denied), so open WriteOnly —
// O_WRONLY on a directory succeeds there and OpenExisting carries no
// truncate semantics. POSIX: opening a directory write-only fails with
// EISDIR, so ReadOnly is used; fsync on that fd is exactly the
// directory-entry durability primitive. A failure surfaces to the caller
// as StoreError rather than silently skipping the ordering guarantee.
#cfg(all(target="native", platform="windows"))
async fn sync_directory(dir : String) -> Unit {
  let f = @fs.open(dir, mode=WriteOnly)
  defer f.close()
  f.sync(only_data=false)
}

///|
#cfg(all(target="native", not(platform="windows")))
async fn sync_directory(dir : String) -> Unit {
  let f = @fs.open(dir, mode=ReadOnly)
  defer f.close()
  f.sync(only_data=false)
}

///|
/// Append an Add entry to `dir/wal.log` (fsync) and then apply it in
/// memory. `dir` must be the same directory passed to `open_or_create`.
pub async fn SegmentedIndex::add_durable(
  self : SegmentedIndex,
  dir : String,
  doc : Document,
) -> Unit raise StoreError {
  append_wal_entry(dir, Add(doc))
  self.add(doc)
}

///|
/// Append a Delete entry to `dir/wal.log` (fsync) and then apply it in
/// memory. Returns true if the id existed (same semantics as `delete`).
pub async fn SegmentedIndex::delete_durable(
  self : SegmentedIndex,
  dir : String,
  id : String,
) -> Bool raise StoreError {
  append_wal_entry(dir, Delete(id))
  self.delete(id)
}

///|
async fn append_wal_entry(
  dir : String,
  entry : WalEntry,
) -> Unit raise StoreError {
  let buf : Array[Byte] = []
  // Each append batch carries the header so a fresh file is self-
  // describing; wal_decode skips repeated headers.
  for b in WAL_MAGIC.iter() {
    buf.push(b)
  }
  wal_encode_entry(buf, entry)
  // create_mode=OpenOrCreate is essential: write_file defaults to
  // CreateOrTruncate, which would wipe the log on every append even
  // with append=true (truncation happens at open time).
  @fs.write_file(
    dir + "/wal.log",
    Bytes::from_array(buf[:]),
    create_mode=OpenOrCreate,
    append=true,
    sync=Data,
  ) catch {
    e => raise StoreError::IoError(e.to_string())
  }
}