`sink` — `createStdoutAuditSink`

(no submodules)

Purpose

This module wires the schema, HMAC signer, instruments, and options into a RequestAuditSink implementation that writes JSON-per-line audit records to a LineWriter (default process.stdout). It is the default audit sink shipped with @suaveplan/api-host: container runtimes already redirect stdout into a log pipeline, so operators inherit the records without touching host configuration.

The sink is single-flight: at most one write is on the wire at any time. Subsequent records wait for the slot, but only up to maxQueueMs (default 100ms) before being dropped. Drops are NEVER silent — every drop increments api.audit.dropped_total{sink,reason} and (subject to the once-per-warnThrottleMs throttle) emits a single warning log. Validation, HMAC signing, and serialisation happen on the calling side of the queue so a malformed record is rejected synchronously without consuming the slot.

Features

Implements RequestAuditSink (emit(envelope)) from @suaveplan/api-host/request-pipeline — host accepts the instance directly.
write(record: AuditRecord) overload for admin paths that already hold a canonical record.
Boundary Zod validation — AUDIT_STDOUT.INVALID_RECORD on a bad record.
Bounded queue — AUDIT_STDOUT.QUEUE_FULL on timeout, drop counter
- throttled warning.
HMAC-SHA-256 over canonical bytes appended as _hmac field.
dispose() drains the in-flight write and rejects subsequent writes as closed drops.
Pluggable clock, timers, logger, writer, instruments.

Usage: Basic

import { createStdoutAuditSink } from "@suaveplan/api-audit-stdout/sink";

const sink = createStdoutAuditSink({
    hmacKey: new TextEncoder().encode(process.env["AUDIT_HMAC_KEY"]!),
});
await sink.write({
    ts: new Date().toISOString(),
    type: "admin.plugin.disabled",
    principal: { id: "user_1", scopes: ["admin"] },
    tenant: "tenant_a",
    request_id: "req_1",
    trace_id: null,
    subject: "plugin.exports",
    outcome: "success",
});

Usage: Advanced

import { createStdoutAuditSink } from "@suaveplan/api-audit-stdout/sink";
import { createStdoutAuditInstruments } from "@suaveplan/api-audit-stdout/instruments";
import { MeterProvider } from "@opentelemetry/sdk-metrics";

const meterProvider = new MeterProvider({ readers: [...] });
const sink = createStdoutAuditSink({
    hmacKey,
    writer: {
        async write(line) {
            await myStream.write(line);
        },
    },
    maxQueueMs: 250,
    warnThrottleMs: 30_000,
    instruments: createStdoutAuditInstruments({ meterProvider }),
    logger: myLogger,
});
// Hand into createHost({ audit: sink }) per host wiring.
await sink.emit(envelope);
await sink.dispose();

API Reference

createStdoutAuditSink(options) → StdoutAuditSink.
StdoutAuditSinkOptions — { hmacKey, writer?, clock?, timers?, logger?, instruments?, maxQueueMs?, warnThrottleMs?, sinkLabel? }.
StdoutAuditSink — extends RequestAuditSink; adds write(record) and dispose().
LineWriter — { write(line: string): void | Promise<void> }.
adaptEnvelope(envelope) — exposed for tests and bespoke re-emission flows.

Implementation Notes

The slot reservation uses an in-flight Promise<void> rather than a queue + worker so cancellation flows trivially through GC — there's no worker loop to shut down. acquireSlot races the in-flight write against the timers.setTimer callback; whichever wins clears the loser. The closed flag is checked before validation so a disposed sink rejects fast without spending CPU on schema work. adaptEnvelope is the only place the host's camelCase envelope shape meets the canonical snake_case record — keeping that mapping isolated lets the design-doc wire shape evolve without rippling through unrelated callers.

3.9 KiB Raw Permalink Blame History

sink — createStdoutAuditSink