# @t4h.framework/fs

File system claim for reading and writing files within activities, plus the
workflow-facing file activities: `split` fans a large file out into one child
workflow per slice, `join` folds the batch results back into a single file,
`read` brings a small byte window (a header, a trailer) into the workflow as a
string, and `write` turns content the workflow produced into a file.

## Claim

### FileSystemClaim

Abstract dependency for file system operations. An activity declares this claim when it needs to read or write files. The runtime provides a concrete implementation (e.g. local disk, S3, GCS).

```typescript
import { FileSystemClaim } from '@t4h.framework/fs'
```

### Interface

```typescript
abstract class FileSystemClaim {
  public abstract write(
    path: string,
    readable: Readable | Buffer,
    contentType?: string,
  ): Binary | Promise<Binary>
  public abstract read(path: string): Binary | Promise<Binary>

  public abstract slice(
    binary: Binary,
    range: { start: number; end: number },
    encoding?: BufferEncoding,
  ): Binary | Promise<Binary>
}
```

### Usage inside an activity

```typescript
import { SyncActivity, Claim } from '@t4h.framework/core'
import { FileSystemClaim } from '@t4h.framework/fs'

class SaveReportActivity extends SyncActivity<
  [data: Buffer],
  { path: string },
  any
> {
  private readonly claims = new Claim({
    fs: FileSystemClaim,
  })

  public async toInput(data: Buffer) {
    return { path: 'reports/output.pdf' }
  }

  public async run(input: { path: string }) {
    const binary = await this.claims.fs.write(input.path, Buffer.from('...'))
    return { contentType: binary.contentType }
  }

  public toOutput(output: any) {
    return output
  }
}
```

### Runtime provider

The runtime must supply a concrete implementation:

```typescript
{ provide: FileSystemClaim, value: new LocalFileSystemClaimImpl() }
```

## File read

### read

Reads a byte window of a `Binary` into the workflow as a decoded string. The
output is recorded in the workflow history (replay never re-reads the file),
which is why reads are capped at 1 MiB: this is the primitive for headers,
trailers and single records — a data file flows through `split`, never through
`read`.

```typescript
import { read } from '@t4h.framework/fs'

const header = await read(file, { range: { start: 0, end: 240 } })

const trailer = await read(file, {
  range: { start: file.contentLength - 240, end: file.contentLength },
})
```

Options: `range` (default: the whole binary, subject to the same cap) and
`encoding` (default `'utf8'`).

The activity tries `FileSystemClaim.slice` first — on a file-backed binary a
trailer read is a ranged download, whatever the file size. Binaries the claim
cannot slice (e.g. an SFTP reference) fall back to streaming: the stream is
consumed up to the window and destroyed as soon as it is filled, so a header
read pulls the first chunks only, whatever the file size. The one expensive
case is a trailing window over a non-sliceable binary — the stream must
traverse the whole file to reach it.

Reading a slice is relative to the slice: `read(slice, { range: { start: 0,
end: 10 } })` returns the first 10 bytes of the slice, not of the underlying
file — slice ranges compose.

## File write

### write

Stores content produced inside the workflow and returns its `Binary` — the only
supported way to obtain a `Binary` from a string or a `Buffer`:

```typescript
import { write } from '@t4h.framework/fs'

const file = await write(JSON.stringify(state, null, 2), {
  contentType: 'application/json',
})

await sftp.put('/out/state.json', file)
```

Options: `contentType` (default: inferred from the filename, then
`application/octet-stream`), `filename` (default: the activity id) and
`encoding` for a string content (default `'utf8'`; a `Buffer` always travels as
base64).

**Never implement `Binary` in an App.** Passing a value to a primitive makes it
the input of a durable activity, and that input is serialized to storage before
the activity runs — a `Binary` holding bytes in memory has nothing to serialize
into, and the activity may run in another process after a replay. A hand-rolled
class can pass locally (the dev server keeps values in memory) and fail once
deployed.

`write` takes content already in workflow memory — a state file, a report, a
return file. To build a large file out of many records, use `split` + `join`,
which stream through storage.

## File split

### split

Splits a file into slices, starts one child workflow per slice as
a single batch, and suspends the parent until every child is terminal —
regardless of each child succeeding or failing (a rejected child never
rejects the batch):

```typescript
import { split, Split } from '@t4h.framework/fs'

const processRecord = new Workflow(
  { id: 'process-record', schema: Type.String() },
  async (record: string) => {
    // record is the raw slice content
  },
)

const main = new Workflow({ id: 'main', schema: fileSchema }, async (file: Binary) => {
  const batch = await split(file, processRecord, {
    delimiter: Split.line(),
  })

  // this line only runs once the whole batch is terminal;
  // batch: WorkflowBatchRef — { id: string, count: number }
})
```

The source is read once, as a stream:

1. The source is read once, worker side: cut into records, filtered by
   `where` when given, grouped by `every`, and each slice goes into the batch
   as a string as soon as it is complete. Nothing collects the slices, and a
   slice may take up to 3 MiB: past that the split fails
   while it reads, at the byte the slice starts.
2. The runtime reads that stream once and stores the batch as it goes; no copy
   of the file is written and no slice is read back by range.
3. The activity output is just the batch reference (`{ id, count }`), so the
   parent history stays O(1) whatever the slice count. Per-child results are
   never aggregated: they live with each child run, to be inspected
   individually through the batch id.

Child workflows declare their real input type (e.g. `Type.String()`); how the
bytes reach them is a runtime concern.

### Delimiters

The cut is defined by the dev through a serializable delimiter spec. When the
delimiter is found at `[p, p + length)`, the previous slice ends at
`p + keep.previous` and the next slice starts at `p + length - keep.next`; the
bytes in between are discarded. The defaults discard the whole delimiter
(classic line semantics).

```typescript
Split.line() // cut at every '\n', delimiter discarded
Split.char(';') // cut at any byte sequence
Split.char(';', { previous: 1 }) // keep the ';' with the previous slice
Split.char('\n10', { next: 2 }) // positional files, see below
```

The last form targets positional files where a logical record spans multiple
physical lines and a new record only starts when a line begins with a marker
(the record type code):

```
10JOHN;123\n20detail-a\n20detail-b\n10MARY;456\n20detail-c
└──────────── record 1 ───────────┘└─────── record 2 ──────┘
```

With `Split.char('\n10', { next: 2 })` the newline between records is
discarded, and the `10` marker — which is data, not separator — stays at the
start of the next slice.

### Options

```typescript
await split(file, workflow, {
  delimiter: Split.line(), // required: where to cut
  every: 10_000, // cut only every N occurrences (groups records)
  trim: { start: 241, end: 241 }, // bytes cut off each end before scanning
  skipEmpty: false, // keep zero-length slices (default: skipped)
  encoding: 'latin1', // delimiter bytes + slice decoding (default: utf8)
  timeoutInMs: 3_600_000, // max ms to await the batch; rejects on expiry
})
```

- **`every`** groups N records per child. With one million lines, `every:
10_000` starts 100 children instead of one million; each child receives a
  raw block with its inner delimiters untouched and iterates locally. `keep`
  arithmetic only applies at actual cut points.
- **`trim`** cuts a fixed number of bytes off each end of the file before
  scanning — a positional header/trailer, which may not be aligned with the
  record delimiter (a record-start delimiter like `Split.char('\n10')` fuses
  the trailer into the last record's slice, so counting records would drop a
  valid one along with it). Trimmed content never becomes a slice nor counts
  toward `every`; fetch it with `read`. Getting the byte counts right is the
  caller's business — positional layouts make them constants:

  ```typescript
  const header = await read(file, { range: { start: 0, end: 240 } })

  const batch = await split(file, processInvoice, {
    delimiter: Split.char('\n10', { next: 2 }),
    trim: { start: 241, end: 241 }, // header + '\n', trailer + '\n'
  })
  ```

- **`skipEmpty`** (default `true`) drops zero-length slices — consecutive
  delimiters, or a file starting/ending with one — so they never become child
  workflows. Turn it off when the slice index must align 1:1 with the record
  position in the source file. "Empty" means zero bytes: a whitespace-only
  line is a regular slice.
- **`encoding`** converts the delimiter to bytes and decodes each slice into
  the string its child receives. Offsets are always computed over raw bytes,
  so multi-byte encodings never skew the cuts.
- **`timeoutInMs`** is the maximum time in milliseconds to await the batch. On
  expiry the activity rejects (reason `timeout`) instead of suspending
  forever — the safety net for a child that never reaches a terminal state.

## File join

The inverse of `split`: folds the results of a completed batch into a single
file. Outputs stream in one at a time through `WorkflowBatchClaim` and flow
straight into `FileSystemClaim.write` with backpressure, so memory stays
bounded whatever the batch size — only the resulting binary reaches the
workflow history.

```typescript
import { join, split, Split } from '@t4h.framework/fs'

const batch = await split(file, processRecord, { delimiter: Split.line() })

const csv = await join(batch, {
  format: 'csv',
  columns: [
    { key: 'id', header: 'Código' },
    { key: 'value', header: 'Valor (R$)' },
    'date', // shorthand: key === header
  ],
})
// csv: Binary
```

### Formats

- **`lines`** — concatenates raw child outputs; the child owns the line
  format. Options: `header`, `separator` (default `'\n'`).
- **`entries`** — NDJSON dump of `{ index, status, output }`, one entry per
  line. The format-agnostic view of the batch, rejected children included by
  default.
- **`csv`** — projects object outputs into RFC 4180 escaped columns. Options:
  `columns` (required; each entry is `{ key, header? }` where `key` selects
  the value on the child output — a dot-path like `'meta.total'` reaches
  nested fields, matching the `where` semantics; a literal top-level key that
  contains dots wins over the nested path — and `header` is the label on the
  header row — a plain string is shorthand for `{ key }`), `delimiter`
  (default `','`), `header` (default `true`; emits the column labels as the
  first row).

### Status filter

`status` selects which child results make it into the file, instead of an
error policy: it defaults to `['fulfilled']` for `lines` and `csv`, and to
both statuses for `entries`. Filtering is silent by design — a file of
rejected children for reprocessing is one call away:

```typescript
const failures = await join(batch, { format: 'entries', status: ['rejected'] })
```

### Where filter

`where` narrows results with a [sift](https://github.com/crcn/sift.js)
(MongoDB-style) query over each entry's output, applied after the `status`
filter. The accepted subset is deliberate — only declarative, JSON-only
operators: `$eq`, `$ne`, `$lt`, `$lte`, `$gt`, `$gte`, `$in`, `$nin`,
`$exists`, `$not`, `$and`, `$or`, `$nor`, plus plain equality and dot-paths
for nested fields:

```typescript
const relevant = await join(batch, {
  format: 'csv',
  columns: ['id', 'total'],
  where: {
    valid: true, // plain strict equality
    total: { $gt: 100 }, // operators
    'meta.situacao': 'aprovada', // dot-path into nested output
    $or: [{ ref: 'a' }, { ref: 'b' }],
  },
})
```

`join` also accepts the child output type as a generic — top-level keys then
narrow to its fields, with values typed by the field (dot-paths are accepted
without narrowing):

```typescript
type InvoiceResult = { id: string; valid: boolean; total: number }

await join<InvoiceResult>(batch, {
  format: 'csv',
  columns: ['id'],
  where: { valid: true, total: { $gt: 100 } }, // { valid: 'yes' } rejects
})
```

Anything that cannot serialize losslessly into the workflow history is
rejected eagerly at `toInput` with a `TypeError`: functions (including
`$where`), `RegExp` values (including `$regex`), `undefined`, and operators
outside the subset — a dropped closure would make the replay filter
differently than the first execution. For arbitrary predicates, write a
custom activity in project code (the function then lives in versioned code,
not in history).

Two MongoDB semantics worth knowing: array fields match by containment
(`{ tags: 'a' }` matches `tags: ['a', 'b']`), and equality against an object
value compares deeply (never by reference). Comparisons never coerce types.

Filtering runs inside the existing streaming loop, over outputs that are
already hydrated one at a time — the query compiles once and adds one
predicate call per entry, no extra serialization and no change to the memory
profile. It is applied app-side: every child output is still fetched and
parsed; `where` decides what enters the file, not what leaves the database.