@okeav/audit-client

A standalone, framework-agnostic audit-event client for Node.js: declare an event's intent up front, get automatic emit-on-success / emit-on-failure semantics, and persist through a configurable enrich-and-store pipeline.

Extracted from and generalized out of a proven production audit pattern. This package fixes no event vocabulary and no transport — it ships two required fields (action, outcome) and the mechanism that wraps around them: declare-then-resolve on the producer side, enrich-then-store on the consumer side. You bring your own event shape, your own message transport (or none at all), and your own storage.

Zero required setup beyond a write(event) function or an emit(event) function: no database, no broker, no config file. Everything is in-memory pure functions plus optional Express middleware — install it, wire in whatever transport/storage you already have, and it works. See examples/express-quickstart/ for a complete server you can run in under a minute.

Install

npm install @okeav/audit-client

Express integration is an optional peer dependency — only needed if you use @okeav/audit-client/middleware/express.

Local development

npm test                            # full unit + integration suite (node --test), no setup required
cd examples/express-quickstart && npm install && npm start   # runnable demo server

Core concepts

| Concept | What it means | Who defines it | |---|---|---| | action | What happened (e.g. 'user.create') | Your app — any string | | outcome | How it went (e.g. 'SUCCESS' / 'FAILURE') | Your app — OUTCOMES is a convenience default, not enforced | | emit | How a produced event leaves the process | Your transport — RabbitMQ, Kafka, a webhook, an in-memory queue, anything | | storage | Where a received event ends up | Your adapters — ship a file/JSONL one, bring your own for anything else | | enrich | Transform pipeline run before storage | Your plugins — redact, tag, pseudonymize, compute retention, or write your own |

The package never hardcodes the values above — it takes an AuditClient config (your transport function, your storage adapters, your enrich pipeline) and gives you back publish/pending/persist.

Quick start

import { createAuditClient, OUTCOMES } from '@okeav/audit-client';

const audit = createAuditClient({
  emit: (event) => myMessageBus.publish('audit-events', event),
});

// Explicit publish
await audit.publish({ action: 'user.create', outcome: OUTCOMES.SUCCESS, actor: { id: 'u1' } });

// Declare-then-resolve — the pattern this package is built around
const pending = audit.pending({ action: 'user.create', actor: { id: 'u1' } });
try {
  const user = await createUser();
  await pending.succeed({ target: { id: user.id } });
} catch (err) {
  await pending.fail(err); // outcome: FAILURE, err.message captured into metadata.error
  throw err;
}

Producer side: publish and pending

publish(event) stamps eventId/timestamp, validates that action/outcome are present, and calls your emit function. It's the only place validation happens — persist (below) never re-validates.

pending(event) is the "declare intent up front, resolve later" building block. It's framework-agnostic — hold the returned object on whatever request/operation-scoped place your code already has (an Express req, a Koa ctx, a plain local variable in a script or queue worker):

const pending = audit.pending({ action: 'payment.charge', actor });
// ... later, on whichever path actually happens ...
await pending.succeed({ target, metadata: { amount } });
// or
await pending.fail(someError);

A pending event can only be resolved once — resolving it twice rejects with AuditError(code: 'ALREADY_RESOLVED') rather than silently double-publishing.

Consumer side: persist

persist(event) runs your enrich pipeline (in order, each step gets the previous step's output) and then writes the result to every configured storage adapter:

import { createAuditClient, createFileAdapter, redactFields, pseudonymize, retention } from '@okeav/audit-client';

const sink = createAuditClient({
  storage: [createFileAdapter({ dir: './data/audit' })],
  enrich: [
    redactFields(['metadata.password']),
    pseudonymize('actor.id'),
    retention({ days: 2555 }),
  ],
});

// however your transport delivers events to a consumer process:
subscribe('audit-events', async (event) => {
  await sink.persist(event);
});

Storage adapter failures are aggregated into one AuditError(code: 'STORAGE_WRITE_FAILED') by default (with the individual errors on .cause), or routed to onStorageError(err, event) per failure if you provide one instead of throwing.

Producer and consumer are never auto-bridged

Unlike some pub/sub wrappers, this package does not secretly wire emit to persist for you. If you want a single-process client that does both, wire it yourself:

const audit = createAuditClient({
  emit: async (event) => { await audit.persist(event); },
  storage: [...],
});

If you want a real producer process and a separate consumer process (the common case — e.g. an API service publishing events, and a dedicated worker persisting them), construct two client instances, each configured with only the half it needs, connected by whatever transport you already have.

Storage adapters

Ship one: createFileAdapter({ dir, partition, fileName }) — append-only JSONL, one file per UTC day, one line per event; naturally immutable since nothing in it ever rewrites a line. createMemoryAdapter() is provided for tests/quickstarts.

Write your own by implementing { write(event) } (sync or async):

const postgresAdapter = {
  write: (event) => db.query('INSERT INTO audit_events (data) VALUES ($1)', [event]),
};

Enrich plugins

All four are optional, composable, and generic — none hardcode what "sensitive" or "PII" means, that's the field list you pass in.

import { redactFields, tagFields, pseudonymize, retention } from '@okeav/audit-client';

redactFields(['metadata.token', 'actor.email']);              // replace with '[REDACTED]' (configurable)
tagFields(['ipAddress', 'actor.email'], { as: 'pii' });        // event.pii = true/false
pseudonymize('actor.id', { as: 'actor.pseudonymId' });         // one-way hash, keeps a stable join key
retention({ days: 30 });                                       // event.retainUntil = timestamp + days

retention only computes and writes the field — actually expiring data (a DB TTL index, a cleanup job over the file adapter's output) is your infrastructure's job.

Express middleware

Import from the /middleware/express subpath so the core package never pulls in an Express type dependency for consumers who only need the pure client.

import { createAuditMiddleware } from '@okeav/audit-client/middleware/express';

const { attachAudit, auditErrorHandler } = createAuditMiddleware(audit);

app.use(attachAudit());          // early, before routes
// ...routes...
app.use(auditErrorHandler());    // last, after every route/error middleware

router.post('/users', async (req, res, next) => {
  const pending = req.beginAudit({ action: 'user.create', actor: req.user });
  try {
    const user = await createUser(req.body);
    await pending.succeed({ target: { id: user.id } });
    res.json(user);
  } catch (err) {
    next(err); // auditErrorHandler resolves the pending event as a failure automatically
  }
});

If the handler throws or calls next(err) before pending.succeed() runs, auditErrorHandler automatically resolves every unresolved pending event on that request as a failure — you never have to remember to audit the unhappy path yourself. A request can call req.beginAudit() more than once; each pending event is tracked and resolved independently.

attachAudit() merges a small default request context (correlationId, ip, userAgent, request.{method,path}) under whatever fields you pass to beginAudit(). Override either the context extractor or the property name:

createAuditMiddleware(audit, {
  requestContext: (req) => ({ correlationId: req.id, tenantId: req.tenant?.id }),
  propertyName: 'audit', // req.audit(...) instead of req.beginAudit(...)
});

Errors

Every function that throws/rejects uses the single AuditError type — never shapes an HTTP response itself:

import { AuditError, isAuditError, ERROR_CODES } from '@okeav/audit-client';

app.use((err, req, res, next) => {
  if (isAuditError(err)) return res.status(err.httpStatus).json({ code: err.code, message: err.message });
  next(err);
});

What this package deliberately does NOT do (v1 scope)

• No bundled event vocabulary — no categories, actions, or entity types ship with the package; that's your app's config, never package content.
• No transport — no RabbitMQ/Kafka/Redis client bundled. emit is a plain injected function; bring any transport, or none.
• No auto-bridge between emit and persist — see above; wiring the two together (or keeping them in separate processes) is explicit, not implicit.
• No enforced coverage — like the pattern it's extracted from, whether a given code path calls pending()/publish() at all is up to you; this package doesn't add a lint rule or a route registry that makes omission impossible.
• No database-specific storage adapter beyond the file/JSONL one — write your own { write(event) } for Postgres/Mongo/whatever you use.

Testing

npm test

Full unit-test coverage for every module, plus an integration suite (test/integration.test.js) that exercises the whole round trip — pending event → auto-resolve on success/failure → transport → enrich pipeline → storage — against a generic project-management-style event vocabulary, proving the extracted mechanism reproduces the real-world behavior it was pulled out of without carrying any of that origin's specific fields along.