Fs.FsLayer

Definitions

def accessTime(f: String): Result[IoError, Int64] \ IO Source

def allowGlob(patterns: Nel[String], path: String): Result[IoError, String] Source

Resolves path to an absolute, normalized form and checks that it matches at least one of the given glob patterns. Returns Ok(resolvedPath) if allowed, or Err(PermissionDenied) if the path does not match any pattern.

def allowList(allowedDirs: Nel[String], path: String): Result[IoError, String] Source

Resolves path to an absolute, normalized form and checks that it falls within at least one of the allowedDirs. Returns Ok(resolvedPath) if allowed, or Err(PermissionDenied) if the path is outside every allowed directory.

def append(data: { str = String }, f: String): Result[IoError, Unit] \ IO Source

def appendBytes(data: Vector[Int8], f: String): Result[IoError, Unit] \ IO Source

def appendLines(data: { lines = List[String] }, f: String): Result[IoError, Unit] \ IO Source

def atomicWrite(file: String, doWrite: String -> Result[IoError, Unit] \ ef, doMoveAtomic: String -> (String -> Result[IoError, Unit] \ ef), doDelete: String -> Result[IoError, Unit] \ ef): Result[IoError, Unit] \ ef

Source

Performs an atomic write to file by writing to a temporary sibling path first, then atomically moving it into place.

doWrite(tmp) writes data to the temporary path.
doMoveAtomic(tmp, file) atomically moves the temp file to the target.
doDelete(tmp) cleans up the temp file on failure.

The caller supplies these as effect operations so that this helper works with any effect (FileWrite, FileSystem, etc.).

def backup(file: String, suffix: String, doCopy: String -> (String -> Result[IoError, Unit] \ ef), doAction: Unit -> Result[IoError, a] \ ef): Result[IoError, a] \ ef

Source

Performs a backup of file to file + suffix before executing doAction.

doCopy(file, bak) copies the file to the backup path.
If the copy succeeds, doAction() is executed.
If the copy fails with NotFound, the file doesn't exist yet, so doAction() proceeds without backup.
If the copy fails with any other error, the operation is aborted.

def bytesToHex(v: Vector[Int8]): String Source

Converts a vector of bytes to a lowercase hex string.

def checkTransferLimit(maxSize: Size, actualSize: Size): Result[IoError, Unit] Source

Checks that actualSize does not exceed maxSize. Returns Ok(()) if within the limit, or Err(TransferLimitExceeded) otherwise.

def checksumAfterDelete(file: String, sidecar: String, deleteResult: Result[IoError, Unit], doDeleteSidecar: String -> Result[IoError, Unit] \ ef): Result[IoError, Unit] \ ef

Source

Deletes the checksum sidecar after a successful delete.

If the sidecar does not exist, the delete result is returned unchanged. Sidecar delete errors are discarded.

def checksumAfterMove(src: String, dst: String, sidecar: String, moveResult: Result[IoError, Unit], doMoveSidecar: String -> (String -> Result[IoError, Unit] \ ef)): Result[IoError, Unit] \ ef

Source

Moves the checksum sidecar after a successful moveWith.

If the source sidecar does not exist, the move result is returned unchanged. Sidecar move errors are discarded.

def checksumAfterWrite(file: String, doWrite: Unit -> Result[IoError, Unit] \ ef1, doUpdateSidecar: String -> Result[IoError, Unit] \ ef2): Result[IoError, Unit] \ ef1 + ef2

Source

Performs doWrite, then computes and writes a checksum sidecar for file.

doUpdateSidecar(file) should read the file bytes, hash them, and write the hex digest to the sidecar path.

def checksumOnRead(file: String, sidecar: String, computeHex: Vector[Int8] -> String, readResult: Result[IoError, a], doReadBytes: String -> Result[IoError, Vector[Int8]] \ ef, doReadSidecar: String -> Result[IoError, String] \ ef): Result[IoError, a] \ ef

Source

Verifies a read or readLines result against a checksum sidecar.

Since the result type (String or List[String]) does not preserve raw bytes, doReadBytes is called to obtain the on-disk bytes for hashing. If no sidecar exists, the result passes through.

def checksumOnReadBytes(file: String, sidecar: String, computeHex: Vector[Int8] -> String, readResult: Result[IoError, Vector[Int8]], doReadSidecar: String -> Result[IoError, String] \ ef): Result[IoError, Vector[Int8]] \ ef

Source

Verifies a readBytes result against a checksum sidecar.

The returned bytes are hashed via computeHex and compared against the sidecar. If no sidecar exists, the result passes through.

def chroot(chrootDir: String, path: String): Result[IoError, String] Source

Resolves path against chrootDir and validates that the resolved result stays within the chroot directory. Symlinks are resolved to their real filesystem targets before the boundary check.

Returns Ok(resolvedPath) if the path is within bounds, or Err(PermissionDenied) if the path escapes the chroot.

def conflictCheck(file: String, snapshots: MutMap[String, Int64, r], getModTime: String -> Result[IoError, Int64] \ ef): Result[IoError, Unit] \ ef + r

Source

Checks whether file was modified externally since last snapshot.

No snapshot → Ok() (file was never read, nothing to conflict with).
Mod time matches snapshot → Ok().
Mod time differs → Err(Conflict(...)).
File deleted (NotFound) → Ok() (file gone, proceed).
Other getModTime error → propagated.

def copy(src: { src = String }, dst: String): Result[IoError, Unit] \ IO Source

def copyWith(src: { src = String }, dst: String, opts: Set[CopyOption]): Result[IoError, Unit] \ IO Source

def creationTime(f: String): Result[IoError, Int64] \ IO Source

def delete(f: String): Result[IoError, Unit] \ IO Source

def denyGlob(patterns: Nel[String], path: String): Result[IoError, String] Source

Resolves path to an absolute, normalized form and checks that it does not match any of the given glob patterns. Returns Ok(resolvedPath) if permitted, or Err(PermissionDenied) if the path matches a denied pattern.

def denyList(deniedDirs: Nel[String], path: String): Result[IoError, String] Source

Resolves path to an absolute, normalized form and checks that it does not fall within any of the deniedDirs. Returns Ok(resolvedPath) if permitted, or Err(PermissionDenied) if the path is within a denied directory.

def ensureParentDirs(file: String, doMkDirs: String -> Result[IoError, Unit] \ ef, doAction: Unit -> Result[IoError, a] \ ef): Result[IoError, a] \ ef

Source

Ensures the parent directory of file exists before executing doAction.

If file has no parent (bare filename), doAction() proceeds directly.
If doMkDirs(parent) succeeds, doAction() proceeds.
If doMkDirs(parent) fails, the error is returned and doAction is not performed.

def estimateStringBytes(s: String): Size Source

Estimates the UTF-8 byte size of a string by multiplying the character count by 4 (the maximum bytes per character in UTF-8).

def estimateStringListBytes(l: List[String]): Size Source

Estimates the UTF-8 byte size of a list of strings by multiplying the total character count by 4 (the maximum bytes per character in UTF-8).

def exists(f: String): Result[IoError, Bool] \ IO Source

def glob(base: String, pattern: String): Result[IoError, List[String]] \ IO Source

def isDirectory(f: String): Result[IoError, Bool] \ IO Source

def isExecutable(f: String): Result[IoError, Bool] \ IO Source

def isReadable(f: String): Result[IoError, Bool] \ IO Source

def isRegularFile(f: String): Result[IoError, Bool] \ IO Source

def isSymbolicLink(f: String): Result[IoError, Bool] \ IO Source

def isWritable(f: String): Result[IoError, Bool] \ IO Source

def list(f: String): Result[IoError, List[String]] \ IO Source

def logPost(op: String, path: String, describe: a -> String, result: Result[IoError, a]): Unit \ Logger Source

Logs a post-operation message: Debug on success, Warn on error.

def logPost2(op: String, src: String, dst: String, result: Result[IoError, Unit]): Unit \ Logger Source

Logs a post-operation message for two-path operations (copy/move).

def logPre(op: String, path: String): Unit \ Logger Source

Logs a pre-operation message at Debug level.

def logPre2(op: String, src: String, dst: String): Unit \ Logger Source

Logs a pre-operation message for two-path operations (copy/move) at Debug level.

def mkDir(d: String): Result[IoError, Unit] \ IO Source

def mkDirs(d: String): Result[IoError, Unit] \ IO Source

def mkTempDir(prefix: String): Result[IoError, String] \ IO Source

def modificationTime(f: String): Result[IoError, Int64] \ IO Source

def move(src: { src = String }, dst: String): Result[IoError, Unit] \ IO Source

def moveWith(src: { src = String }, dst: String, opts: Set[MoveOption]): Result[IoError, Unit] \ IO Source

def parentDir(path: String): Option[String] Source

Returns the parent directory of path, or None if the path has no parent (e.g., a bare filename like "test.txt").

def read(f: String): Result[IoError, String] \ IO Source

def readBytes(f: String): Result[IoError, Vector[Int8]] \ IO Source

def readLines(f: String): Result[IoError, List[String]] \ IO Source

def resolve(baseDir: String, path: String): String Source

Resolves path against baseDir using Java's Path.resolve semantics:

If path is absolute, it is returned as-is.
If path is relative, it is joined to baseDir.

def resolveRealPath(path: String): Result[IoError, String] Source

Resolves path to its real filesystem target by following all symlinks. Returns Ok(realPath) if the path exists, or an appropriate IoError if the path is invalid, does not exist, or cannot be resolved.

def resolveRealPathOrSelf(path: String): String Source

Resolves path to its real filesystem target by following all symlinks. Falls back to the original path if resolution fails (e.g., the file does not exist yet). Useful for write/create operations where the target may not exist.

def rotateIfNeeded(file: String, maxSize: Size, maxFiles: Int32, doSize: String -> Result[IoError, Size] \ ef1, doMove: String -> (String -> Result[IoError, Unit] \ ef2), doDelete: String -> Result[IoError, Unit] \ ef2, doAction: Unit -> Result[IoError, a] \ ef2): Result[IoError, a] \ ef1 + ef2

Source

Rotates file if its size meets or exceeds maxSize, then executes doAction.

Rotation shifts existing numbered files (file.1 … file.{maxFiles}) upward by one, deletes the oldest (file.{maxFiles}), and moves file to file.1. The doAction then writes to the now-absent original path.

If the file does not exist (NotFound), rotation is skipped and doAction proceeds directly (append will create it).

def size(f: String): Result[IoError, Size] \ IO Source

def snapshotModTime(file: String, snapshots: MutMap[String, Int64, r], getModTime: String -> Result[IoError, Int64] \ ef): Unit \ ef + r

Source

Snapshots the modification time of file into snapshots.

If getModTime fails (file deleted, etc.), silently skips.

def truncate(f: String): Result[IoError, Unit] \ IO Source

def updateSidecar(file: String, sidecar: String, computeHex: Vector[Int8] -> String, doReadBytes: String -> Result[IoError, Vector[Int8]] \ ef1, doWriteSidecar: String -> (String -> Result[IoError, Unit] \ ef2)): Result[IoError, Unit] \ ef1 + ef2

Source

Reads the bytes of file, hashes them with computeHex, and writes the hex digest to the sidecar path file + sidecar.

def verifyChecksum(file: String, sidecar: String, actual: String, doReadSidecar: String -> Result[IoError, String] \ ef): Result[IoError, Unit] \ ef

Source

Verifies actual (a hex digest string) against the checksum stored in the sidecar file file + sidecar.

If the sidecar does not exist (NotFound), returns Ok(()).
If the sidecar exists and matches, returns Ok(()).
If the sidecar exists and does not match, returns Err(IoError(ChecksumMismatch, ...)).

def write(data: { str = String }, f: String): Result[IoError, Unit] \ IO Source

def writeBytes(data: Vector[Int8], f: String): Result[IoError, Unit] \ IO Source

def writeLines(data: { lines = List[String] }, f: String): Result[IoError, Unit] \ IO Source