///|
/// Open and exclusively lock the stable lock file for one persistent database.
async fn acquire_database_writer_lock(
  database_path : String,
) -> @fs.File raise DatabaseError {
  let lock_path = "\{database_path}/LOCK"
  let lock = @fs.open(
    lock_path,
    mode=@fs.Mode::ReadWrite,
    create_mode=@fs.CreateMode::OpenOrCreate,
  ) catch {
    _ =>
      raise DatabaseError::IoError(
        file=lock_path,
        operation="open",
        reason="failed to open database lock file",
      )
  }
  let acquired = lock.try_lock(@fs.Lock::Exclusive) catch {
    _ => {
      lock.close()
      raise DatabaseError::IoError(
        file=lock_path,
        operation="lock",
        reason="failed to acquire database writer lock",
      )
    }
  }
  if !acquired {
    lock.close()
    raise DatabaseError::WriterLockOccupied(path=database_path)
  }
  lock
}

///|
async fn sync_directory_metadata_with_mode(
  directory_path : String,
  mode : @fs.Mode,
) -> Unit raise DatabaseError {
  let directory = @fs.open(directory_path, mode~) catch {
    _ =>
      raise DatabaseError::IoError(
        file=directory_path,
        operation="directory-sync-open",
        reason="failed to open directory for metadata synchronization",
      )
  }
  defer directory.close()
  directory.sync(only_data=false) catch {
    _ =>
      raise DatabaseError::IoError(
        file=directory_path,
        operation="directory-sync",
        reason="failed to synchronize directory metadata",
      )
  }
}

///|
#cfg(platform="windows")
async fn sync_directory_metadata(
  directory_path : String,
) -> Unit raise DatabaseError {
  sync_directory_metadata_with_mode(directory_path, @fs.Mode::WriteOnly)
}

///|
#cfg(not(platform="windows"))
async fn sync_directory_metadata(
  directory_path : String,
) -> Unit raise DatabaseError {
  sync_directory_metadata_with_mode(directory_path, @fs.Mode::ReadOnly)
}

///|
/// Initialize a persistent database directory and hold its writer lock. The
/// immediate parent directory must already exist.
pub async fn Database::init(path : String) -> Database raise DatabaseError {
  let wal_path = "\{path}/wal.bin"
  let wal_exists = @fs.exists(wal_path) catch {
    _ =>
      raise DatabaseError::IoError(
        file=wal_path,
        operation="stat",
        reason="failed to inspect transaction log path",
      )
  }
  if wal_exists {
    raise DatabaseError::DatabaseAlreadyInitialized(path~)
  }
  let directory_exists = @fs.exists(path) catch {
    _ =>
      raise DatabaseError::IoError(
        file=path,
        operation="stat",
        reason="failed to inspect database directory",
      )
  }
  if !directory_exists {
    @fs.mkdir(path, recursive=false) catch {
      _ =>
        raise DatabaseError::IoError(
          file=path,
          operation="mkdir",
          reason="failed to create database directory",
        )
    }
  }
  let writer_lock = acquire_database_writer_lock(path)
  let wal = @fs.open(
    wal_path,
    mode=@fs.Mode::WriteOnly,
    sync=@fs.SyncMode::NoSync,
    create_mode=@fs.CreateMode::CreateNew,
  ) catch {
    _ => {
      writer_lock.unlock()
      writer_lock.close()
      raise DatabaseError::IoError(
        file=wal_path,
        operation="create",
        reason="failed to create the transaction log",
      )
    }
  }
  wal.sync(only_data=false) catch {
    _ => {
      wal.close()
      writer_lock.unlock()
      writer_lock.close()
      raise DatabaseError::IoError(
        file=wal_path,
        operation="sync",
        reason="failed to synchronize the new transaction log",
      )
    }
  }
  wal.close()
  errdefer {
    writer_lock.unlock()
    writer_lock.close()
  }
  sync_directory_metadata(path)
  sync_directory_metadata("\{path}/..")
  {
    state: Ref::Ref({
      committed: GraphState::new(),
      active: None,
      closed: false,
      wal_path: Some(wal_path),
      transaction_id: 0UL,
      recovery_required: false,
      writer_lock: Some(writer_lock),
    }),
  }
}

///|
/// Open an initialized persistent database, replay its WAL and hold the writer lock.
pub async fn Database::open(path : String) -> Database raise DatabaseError {
  let wal_path = "\{path}/wal.bin"
  let wal_exists = @fs.exists(wal_path) catch {
    _ =>
      raise DatabaseError::IoError(
        file=wal_path,
        operation="stat",
        reason="failed to inspect transaction log path",
      )
  }
  if !wal_exists {
    raise DatabaseError::DatabaseNotInitialized(path~)
  }
  let writer_lock = acquire_database_writer_lock(path)
  errdefer {
    writer_lock.unlock()
    writer_lock.close()
  }
  let bytes = read_wal_file(wal_path)
  let (committed, transaction_id, clean_log) = wal_recover(bytes, file=wal_path)
  if clean_log.length() != bytes.length() {
    wal_repair_tail(wal_path, clean_log, path)
  }
  {
    state: Ref::Ref({
      committed,
      active: None,
      closed: false,
      wal_path: Some(wal_path),
      transaction_id,
      recovery_required: false,
      writer_lock: Some(writer_lock),
    }),
  }
}

///|
/// Replace an incomplete final WAL tail with the already validated record prefix.
async fn wal_repair_tail(
  path : String,
  valid_log : Bytes,
  database_path : String,
) -> Unit raise DatabaseError {
  let temporary_path = "\{path}.repair.tmp"
  @fs.write_file(
    temporary_path,
    valid_log,
    create_mode=@fs.CreateMode::CreateOrTruncate,
    sync=@fs.SyncMode::Full,
  ) catch {
    _ =>
      raise DatabaseError::IoError(
        file=temporary_path,
        operation="repair-write",
        reason="failed to write validated transaction log prefix",
      )
  }
  @fs.rename(temporary_path, path, replace=true) catch {
    _ =>
      raise DatabaseError::IoError(
        file=path,
        operation="repair-rename",
        reason="failed to replace transaction log after tail recovery",
      )
  }
  sync_directory_metadata(database_path)
}

///|
/// Read a WAL whose byte length fits the in-memory record scanner.
async fn read_wal_file(path : String) -> Bytes raise DatabaseError {
  let file = @fs.open(path, mode=@fs.Mode::ReadOnly) catch {
    _ =>
      raise DatabaseError::IoError(
        file=path,
        operation="open",
        reason="failed to open transaction log for recovery",
      )
  }
  let size = file.size() catch {
    _ => {
      file.close()
      raise DatabaseError::IoError(
        file=path,
        operation="size",
        reason="failed to read transaction log size",
      )
    }
  }
  if size < 0L || size > 2147483647L {
    file.close()
    raise DatabaseError::IoError(
      file=path,
      operation="read",
      reason="transaction log exceeds the supported in-memory size",
    )
  }
  let bytes = if size == 0L {
    Bytes::make(0, 0)
  } else {
    file.read_exactly_at(size.to_int(), position=0L) catch {
      _ => {
        file.close()
        raise DatabaseError::IoError(
          file=path,
          operation="read",
          reason="failed to read transaction log",
        )
      }
    }
  }
  file.close()
  bytes
}