Streaming

sensored is built streaming-first. The streaming engine processes continuous text streams without buffering the entire input, making it suitable for real-time pipelines, large files, and network sources.

Basic streaming

const redactor = createRedactor({ presets: ["pii"], rules: {} });

async function* asyncChunks() {
  yield "Contact me at john@";
  yield "example.com for details.";
}

const stream = redactor.stream(asyncChunks());
for await (const event of stream) {
  if (event.type === "text") {
    process.stdout.write(event.text);
  }
}
// Output: "Contact me at [EMAIL_1] for details."

StreamEvent

The stream yields three event types:

type StreamEvent =
  | { type: "text"; text: string }
  | { type: "detection"; group: InspectionGroup }
  | { type: "complete"; map?: RestorationMap };

text

Redacted text segments. Concatenate these to build the full output.

detection

An InspectionGroup with absolute offsets into the original stream. Only emitted when report: true is set in StreamOptions.

const stream = redactor.stream(asyncChunks(), { report: true });
for await (const event of stream) {
  if (event.type === "detection") {
    console.log(event.group.start, event.group.end, event.group.replacement);
  }
}

complete

Emitted once at the end of the stream. When restoration is enabled, includes the RestorationMap:

const stream = redactor.stream(asyncChunks(), { restore: true });
for await (const event of stream) {
  if (event.type === "complete") {
    console.log(event.map);
    // { "[EMAIL_1]": "john@example.com" }
  }
}

StreamOptions

interface StreamOptions {
  signal?: AbortSignal;
  report?: boolean;
  restore?: boolean;
}

signal

An AbortSignal to cancel the stream. If aborted, a CANCELLED error is thrown. The stream checks the signal between chunks.

const controller = new AbortController();

setTimeout(() => controller.abort(), 5000);

const stream = redactor.stream(asyncChunks(), { signal: controller.signal });

report

When true, detection events are emitted with InspectionGroup data containing absolute offsets and match details.

restore

When true, the complete event includes a RestorationMap. This is separate from the config.restore setting — you can use restoration in streaming mode independently.

How it works

The streaming engine:

1. Buffers incoming chunks into an internal buffer
2. Calculates safe flush points based on each detector's stream metadata (maxMatchLength, context windows, boundary lookaround)
3. Adjusts for grapheme clusters to avoid splitting multi-codepoint characters
4. Calls the detection engine on the safe portion up to the flush point
5. Emits redacted text and optional detection events
6. Retains context behind the flush point for detectors that need lookbehind

Limitations

• Buffer limit: The internal buffer is capped at 65,536 UTF-16 code units. If the buffer exceeds this (e.g., a single chunk larger than 64 KiB with no flush point), a BUFFER_LIMIT error is thrown.
• Stream metadata required: Every active detector must declare stream metadata. If any active rule lacks stream support, a STREAM_UNSUPPORTED error is thrown at stream creation time.
• Source errors: If the async iterable throws, it's wrapped in a SOURCE_FAILURE error.