///|
/// Minimal byte-oriented storage boundary for immutable index files.
pub(open) trait Directory {
  fn write(Self, String, Bytes) -> Unit raise PersistenceError
  fn read(Self, String) -> Bytes raise PersistenceError
  fn exists(Self, String) -> Bool
}

///|
/// Storage guarantees exposed by a Directory v2 implementation.
pub(all) struct DirectoryCapabilities {
  atomic_replace : Bool
  durable_sync : Bool
  exclusive_lock : Bool
  read_only_mapping : Bool
} derive(Eq, @debug.Debug)

///|
/// Durable index lifecycle operations layered on the byte-oriented Directory.
pub(open) trait DirectoryV2: Directory {
  fn write_atomic(Self, String, Bytes) -> Unit raise PersistenceError
  fn sync(Self, String) -> Unit raise PersistenceError
  fn remove(Self, String) -> Unit raise PersistenceError
  fn list(Self) -> ReadOnlyArray[String] raise PersistenceError
  fn try_acquire_lock(Self, String) -> Int64? raise PersistenceError
  fn release_lock(Self, String, Int64) -> Unit raise PersistenceError
  fn capabilities(Self) -> DirectoryCapabilities
}

///|
fn validate_directory_name(name : String) -> Unit raise PersistenceError {
  guard name != "" &&
    name != "." &&
    name != ".." &&
    !name.contains("/") &&
    !name.contains("\\") else {
    raise PersistenceError::InvalidName(name)
  }
}

///|
/// In-memory Directory implementation. Reads and writes copy bytes so a
/// caller cannot mutate an already committed snapshot through an alias.
pub struct MemoryDirectory {
  names : Array[String]
  contents : Array[Bytes]
  locks : Array[String]
}

///|
pub fn MemoryDirectory::new() -> MemoryDirectory {
  { names: [], contents: [], locks: [] }
}

///|
pub impl Directory for MemoryDirectory with fn write(self, name, bytes) {
  validate_directory_name(name)
  let snapshot = bytes[:].to_owned()
  match self.names.search_by(existing => existing == name) {
    Some(index) => self.contents[index] = snapshot
    None => {
      self.names.push(name)
      self.contents.push(snapshot)
    }
  }
}

///|
pub impl Directory for MemoryDirectory with fn read(self, name) {
  validate_directory_name(name)
  match self.names.search_by(existing => existing == name) {
    Some(index) => self.contents[index][:].to_owned()
    None => raise PersistenceError::NotFound(name)
  }
}

///|
pub impl Directory for MemoryDirectory with fn exists(self, name) {
  try validate_directory_name(name) catch {
    _ => false
  } noraise {
    _ => self.names.search_by(existing => existing == name) is Some(_)
  }
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn write_atomic(self, name, bytes) {
  Directory::write(self, name, bytes)
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn sync(_self, _name) {
  ()
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn remove(self, name) {
  validate_directory_name(name)
  match self.names.search_by(existing => existing == name) {
    Some(index) => {
      ignore(self.names.remove(index))
      ignore(self.contents.remove(index))
    }
    None => ()
  }
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn list(self) {
  let names = self.names.copy()
  names.sort()
  ReadOnlyArray::from_array(names)
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn try_acquire_lock(self, name) {
  validate_directory_name(name)
  if self.locks.search_by(lock => lock == name) is Some(_) {
    None
  } else {
    self.locks.push(name)
    Some(1L)
  }
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn release_lock(
  self,
  name,
  _token,
) {
  validate_directory_name(name)
  match self.locks.search_by(lock => lock == name) {
    Some(index) => ignore(self.locks.remove(index))
    None => ()
  }
}

///|
pub impl DirectoryV2 for MemoryDirectory with fn capabilities(_self) {
  {
    atomic_replace: true,
    durable_sync: true,
    exclusive_lock: true,
    read_only_mapping: true,
  }
}

///|
/// Filesystem-backed Directory rooted at one directory.
pub struct FsDirectory {
  root : String
}

///|
fn[T] wrap_fs_error(
  operation : () -> T raise @fs.IOError,
) -> T raise PersistenceError {
  operation() catch {
    @fs.IOError::IOError(message) => raise PersistenceError::Io(message)
  }
}

///|
pub fn FsDirectory::new(root : String) -> FsDirectory raise PersistenceError {
  guard root != "" else { raise PersistenceError::InvalidName(root) }
  if @fs.path_exists(root) {
    guard wrap_fs_error(() => @fs.is_dir(root)) else {
      raise PersistenceError::Io("directory root is not a directory: \{root}")
    }
  } else {
    wrap_fs_error(() => @fs.create_dir(root))
  }
  { root, }
}

///|
fn FsDirectory::path(
  self : FsDirectory,
  name : String,
) -> String raise PersistenceError {
  validate_directory_name(name)
  "\{self.root}/\{name}"
}

///|
pub impl Directory for FsDirectory with fn write(self, name, bytes) {
  let path = self.path(name)
  wrap_fs_error(() => @fs.write_bytes_to_file(path, bytes))
}

///|
pub impl Directory for FsDirectory with fn read(self, name) {
  let path = self.path(name)
  if !@fs.path_exists(path) {
    raise PersistenceError::NotFound(name)
  }
  wrap_fs_error(() => @fs.read_file_to_bytes(path))
}

///|
pub impl Directory for FsDirectory with fn exists(self, name) {
  try self.path(name) catch {
    _ => false
  } noraise {
    path => @fs.path_exists(path)
  }
}

///|
pub impl DirectoryV2 for FsDirectory with fn write_atomic(self, name, bytes) {
  let target = self.path(name)
  let temporary_name = ".\{name}.tmp"
  let temporary = self.path(temporary_name)
  wrap_fs_error(() => @fs.write_bytes_to_file(temporary, bytes))
  platform_sync_file(temporary)
  platform_atomic_replace(temporary, target)
  platform_sync_file(target)
}

///|
pub impl DirectoryV2 for FsDirectory with fn sync(self, name) {
  platform_sync_file(self.path(name))
}

///|
pub impl DirectoryV2 for FsDirectory with fn remove(self, name) {
  let path = self.path(name)
  if @fs.path_exists(path) {
    wrap_fs_error(() => @fs.remove_file(path))
  }
}

///|
pub impl DirectoryV2 for FsDirectory with fn list(self) {
  let names = wrap_fs_error(() => @fs.read_dir(self.root))
  names.sort()
  ReadOnlyArray::from_array(names)
}

///|
pub impl DirectoryV2 for FsDirectory with fn try_acquire_lock(self, name) {
  platform_try_acquire_lock(self.path(name))
}

///|
pub impl DirectoryV2 for FsDirectory with fn release_lock(self, name, token) {
  platform_release_lock(self.path(name), token)
}

///|
pub impl DirectoryV2 for FsDirectory with fn capabilities(_self) {
  platform_directory_capabilities()
}

///|
/// Read-only mapping facade. Native builds use durable FsDirectory primitives;
/// portable targets retain the same immutable snapshot semantics.
pub struct MmapDirectory {
  inner : FsDirectory
}

///|
pub fn MmapDirectory::new(
  root : String,
) -> MmapDirectory raise PersistenceError {
  { inner: FsDirectory::new(root) }
}

///|
pub impl Directory for MmapDirectory with fn write(self, name, bytes) {
  Directory::write(self.inner, name, bytes)
}

///|
pub impl Directory for MmapDirectory with fn read(self, name) {
  Directory::read(self.inner, name)
}

///|
pub impl Directory for MmapDirectory with fn exists(self, name) {
  Directory::exists(self.inner, name)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn write_atomic(self, name, bytes) {
  DirectoryV2::write_atomic(self.inner, name, bytes)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn sync(self, name) {
  DirectoryV2::sync(self.inner, name)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn remove(self, name) {
  DirectoryV2::remove(self.inner, name)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn list(self) {
  DirectoryV2::list(self.inner)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn try_acquire_lock(self, name) {
  DirectoryV2::try_acquire_lock(self.inner, name)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn release_lock(self, name, token) {
  DirectoryV2::release_lock(self.inner, name, token)
}

///|
pub impl DirectoryV2 for MmapDirectory with fn capabilities(self) {
  let capabilities = DirectoryV2::capabilities(self.inner)
  {
    atomic_replace: capabilities.atomic_replace,
    durable_sync: capabilities.durable_sync,
    exclusive_lock: capabilities.exclusive_lock,
    read_only_mapping: true,
  }
}