suaveplan/packages/observability/monitor/logger/src/sink/sink.md

sink

(no submodules)

Purpose

The sink layer defines the Sink interface — the pluggable output primitive for @suaveplan/logger — and provides three concrete implementations: ConsoleSink for human-readable terminal output during development, JsonSink for line-delimited JSON (NDJSON) suitable for log aggregators, and BufferedSink for asynchronous, bounded-queue writes with configurable overflow policy and AbortSignal-aware flush. The MultiSink utility routes a single record to multiple sinks in parallel.

By decoupling the logger from any specific I/O implementation, the same AbstractLogger can write to a terminal, a file, a remote endpoint, or an in-memory buffer without code changes — only the sink configuration differs.

Features

• Sink interface: write(record), optional flush(signal?), optional close()
• ConsoleSink — ANSI colour per level, ISO timestamp prefix, optional tag and metadata formatting
• JsonSink — serialises LogRecord to a single JSON line, passes to a pluggable writer callback (default: process.stdout.write)
• BufferedSink — bounded in-memory queue; "drop" policy silently discards overflow; "block" policy awaits drain; flush(AbortSignal?) drains with cancellation support
• MultiSink — fan-out to multiple sinks concurrently
• isSink() type guard; flushSink() / closeSink() helpers that no-op for sinks lacking those methods

Usage: Basic Example

import { JsonSink } from "@suaveplan/logger/sink";

const sink = new JsonSink({ writer: (line) => process.stdout.write(line + "\n") });
sink.write(record); // {"ts":"...","level":"info","msg":"hello","ctx":{}}

Usage: Advanced Example

import { BufferedSink } from "@suaveplan/logger/sink";
import { JsonSink } from "@suaveplan/logger/sink";

const json = new JsonSink({ writer: appendToFile });
const buf = new BufferedSink({
  sink: json,
  maxQueueSize: 500,
  overflowPolicy: "drop",
});

// In a long-running process, flush on shutdown:
process.on("SIGTERM", async () => {
  const ac = new AbortController();
  setTimeout(() => ac.abort(), 5_000); // 5 s timeout
  await buf.flush(ac.signal);
  await buf.close();
});

API Reference

interface Sink

interface Sink {
  write(record: LogRecord): void | Promise<void>;
  flush?(signal?: AbortSignal): Promise<void>;
  close?(): Promise<void>;
}

class ConsoleSink implements Sink

Options: { colors?, timestamp?, includeMetadata?, includeTags? }.

class JsonSink implements Sink

Options: { writer? }. Throws LoggerError(LOGGER.SERIALIZE) if JSON.stringify fails.

class BufferedSink implements Sink

Options: { sink, maxQueueSize?, overflowPolicy? }. Property .dropped counts overflow-dropped records.

class MultiSink implements Sink

Constructor: new MultiSink(sinks: Sink[]). Writes and flushes all sinks concurrently.

isSink(value): value is Sink

Type guard.

flushSink(sink, signal?) / closeSink(sink)

Helpers that call sink.flush(signal) / sink.close() only if the method exists.

Implementation Notes

BufferedSink does not start a background interval — it flushes lazily on flush() or close(). This keeps the class dependency-free (no Clock needed) and avoids surprises in test environments. The "block" overflow policy uses a simple Promise queue to suspend write() callers until the next flush iteration frees capacity.