@devindex/api-kit

Building blocks for backend services. Factories are inert until start() and never select a driver from the environment.

Installation

npm install @devindex/api-kit

Node >=22. The memory drivers need no external service. The Redis-backed drivers load their optional peers only on first use:

npm install bullmq ioredis

./errors

DomainError and its subtypes carry a code and details. The HTTP status is optional: the kit's own codes are mapped by the HTTP layer, so an error thrown in a queue or CLI need not know about HTTP.

throw new ConflictError('email already registered', {
  details: [{ field: 'email' }],
});

An app defines its own errors by extending DomainError with a code of its own. STATUS_BY_CODE cannot know that code, so declare the status where the error is defined — otherwise every app error answers 400:

class PaymentDeclinedError extends DomainError {
  constructor(message = 'Payment declined', options = {}) {
    super(message, { ...options, code: 'PAYMENT_DECLINED', status: 402 });
  }
}

status is set only when given, so an error that never meets HTTP carries no trace of it. Subtypes pin their code but not their status, so new NotFoundError('order archived', { status: 410 }) works.

| Error | Code | |---|---| | ValidationError | VALIDATION_ERROR | | AuthError | UNAUTHORIZED | | ForbiddenError | FORBIDDEN | | NotFoundError | NOT_FOUND | | MethodNotAllowedError | METHOD_NOT_ALLOWED | | ConflictError | CONFLICT | | LimitError | LIMIT_REACHED | | PayloadError | PAYLOAD_TOO_LARGE | | TooManyRequestsError | TOO_MANY_REQUESTS | | UnavailableError | UNAVAILABLE | | DomainError | DOMAIN_ERROR |

Use isDomainError(error) instead of instanceof. The brand crosses multiple installed copies of the package and stays out of serialized responses and logs.

./http

createApp() builds the full Fastify stack and returns it without listening, so tests drive it with app.inject() and starting the server stays the entrypoint's job. Every failure — a DomainError, a schema rejection, an unexpected throw — leaves through one envelope:

import { createApp } from '@devindex/api-kit/http';

const app = await createApp({
  logger,
  routes: async (instance) => {
    instance.get('/orders/:id', async (req) => orders.find(req.params.id));
  },
});

await app.listen({ port: 3000 });

A ConflictError('email already registered') becomes:

{ "error": { "code": "CONFLICT", "message": "email already registered", "details": [], "requestId": "…" } }

The status is the error's own status when it has one, otherwise STATUS_BY_CODE[code], otherwise 400. Unclassified errors are logged and masked as a 500 that never leaks the original message. createApp options:

| Option | Default | Purpose | |---|---|---| | logger | none | A base pino instance; the kit types the lines itself (see below). Omitted disables Fastify logging | | context | none | A ./context store; omitted disables async context | | cors | on (origin: true) | @fastify/cors options; pass false to disable | | helmet | on (CSP off) | @fastify/helmet options; pass false to disable | | routes | none | The app's route plugin, registered last | | plugins | [] | Extra plugins Function or [Function, options], in order | | requestProperties | {} | Request decorators — the app's own vocabulary | | captureRawBody | false | Keep the exact bytes on req.rawBody for webhook signatures | | ajvPlugins / ajvOptions | [] / {} | ajv plugins and merged custom options | | genReqId | kit default | Correlation id strategy | | fastify | {} | Merged last into the Fastify constructor options |

Inbound x-request-id is reused only when it is a valid UUID, otherwise a fresh v4 is generated; the id is always echoed back. schema.js ships ODM-agnostic JSON-Schema helpers — objectSchema, stringSchema, pageQuery, pageResponse, email, dateTime, dateKey, clock (see Pagination).

CORS and security headers

CORS is on by default with origin: true, reflecting the caller's origin, because a browser-facing API almost always needs it. It registers before the routes.

// Default: reflects any origin, no cookies.
await createApp({ routes });

// Server-to-server or same-origin app: turn it off.
await createApp({ cors: false, routes });

Footgun: the default origin: true must not be combined with credentials: true — that lets any site read an authenticated response. A cookie/credentialed API must override origin with an explicit allowlist, which also makes non-listed origins get no CORS header at all:

await createApp({ cors: { origin: ['https://app.example.com'], credentials: true }, routes });

helmet is on by default with contentSecurityPolicy: false. CSP is a browser directive for rendered documents — inert on JSON responses, and its default breaks any HTML tooling bolted onto the API (Swagger, GraphQL playground, HTML error pages). The other headers (X-Content-Type-Options: nosniff, frameguard, HSTS…) stay on.

// Default: security headers on, CSP off.
await createApp({ routes });

// Turn it off entirely.
await createApp({ helmet: false, routes });

// An endpoint serving HTML re-enables CSP with its own policy.
await createApp({ helmet: { contentSecurityPolicy: { useDefaults: true } }, routes });

Anything else — rate limits, compression — still goes through plugins, which registers after these and before the routes.

Pagination

Offset pagination in two shapes, { items, hasMore } and { items, hasMore, total }, over one rule: the query fetches limit + 1 rows and paginate slices the extra one off. The kit never runs the query — skip/limit belong to Mongoose, offset/limit to a SQL builder, a cursor to an upstream API — so it only does the arithmetic and the shape, and works with all of them.

import { paginate } from '@devindex/api-kit/http';

const rows = await Order.find(filter).skip(offset).limit(limit + 1);
return paginate(rows, limit);        // { items, hasMore }

A resource that really needs the count passes it as the third argument, and gets total in the response:

const [rows, total] = await Promise.all([
  Order.find(filter).skip(offset).limit(limit + 1),
  Order.countDocuments(filter),
]);
return paginate(rows, limit, total); // { items, hasMore, total }

hasMore always comes from the extra row, never from total. Deriving it from the count would make it depend on two queries that can disagree — a document inserted between them, and the flag promises a page that is not there.

pageQuery and pageResponse are the two ends of the contract. The response schema is not optional decoration: Fastify strips whatever it does not declare, so a missing hasMore there silently disappears from a correct payload.

import { pageQuery, pageResponse } from '@devindex/api-kit/http';

instance.get('/orders', {
  schema: {
    querystring: pageQuery({ maxLimit: 50 }),
    response: { 200: pageResponse(orderSchema) },
  },
}, async (req) => orders.list(req.query));

pageQuery applies the defaults (limit 20, offset 0), so the handler always reads two integers. pageResponse(items, { total: true }) adds total to the schema — pass it wherever paginate gets a count.

./context

An isolated AsyncLocalStorage store, owned by the service, so two services in one process never leak each other's request metadata. It is opt-in: pass it to createApp({ context }) and the HTTP layer propagates requestId/correlationId below itself, readable in the service layer without threading them through every call.

import { createContextStore, serializableContext } from '@devindex/api-kit/context';

const context = createContextStore();
const app = await createApp({ context, routes });

// deeper in a use case:
const log = context.logger(logger); // child logger bound to the correlation id

serializableContext(context.get()) keeps only the correlation fields, which is what should cross a queue boundary into an event or job.

./log

createLogger() builds a pino instance shaped for later analysis: JSON to stdout, secret keys redacted, and every line ready to carry a type discriminator so a log store can split request, event, job and integration lines apart. pino is an optional peer — install it (and pino-pretty for local pretty-printing) only when you use this module:

npm install pino

import { createLogger, LOG_TYPE, withType } from '@devindex/api-kit/log';

const logger = createLogger({
  level: process.env.LOG_LEVEL ?? 'info',
  base: { service: 'orders' },
  context, // the ./context store — stamps requestId/correlationId on every line
  pretty: process.env.NODE_ENV !== 'production',
});

// Category a scope once; every line from it inherits the type and bindings.
withType(logger, LOG_TYPE.INTEGRATION, { provider: 'stripe' })
  .info({ durationMs, status }, 'charge created');

LOG_TYPE is the closed vocabulary for the type field — request, event, job, schedule, integration, lifecycle. Filtering then reads naturally: type:integration AND level>=50 is every integration error. Pass the same instance — the base one, never a withType child — to createApp({ logger }): the HTTP layer is the one place a single logger emits two categories, so the kit types them itself. What the Fastify instance says (Server listening at…, plugin warnings) is lifecycle; what a request says (incoming request, request completed, the error envelope) is request.

Secrets are redacted at logger creation, never at the call site: DEFAULT_REDACT_PATHS covers password, token, authorization, cookie and friends across three nesting levels. Logging { err } runs pino's error serializer by default, yielding type/message/stack.

Transports

transport is passed straight through to pino, so any target or fan-out works — a file, a service, or several at once. It takes precedence over pretty:

// One line to two sinks: pretty on the console, JSON to a file.
const logger = createLogger({
  transport: {
    targets: [
      { target: 'pino-pretty', options: { destination: 1 } },
      { target: 'pino/file', options: { destination: './logs/app.log' }, level: 'warn' },
    ],
  },
});

In production, prefer the default JSON on stdout and let your collector (Datadog agent, Vector, Fluent Bit…) ship it — skip pretty there. A transport cannot combine with a destination stream; passing both throws.

./events

Publish/subscribe with fan-out: one published event is delivered to every named subscriber independently. Unlike a job, an event has no single consumer — a publisher does not know or wait for who reacts.

import { createEventBus } from '@devindex/api-kit/events';

const bus = createEventBus({
  driver: 'bullmq',
  redisUrl,
  prefix: 'billing',
  logger,
  defaults: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1_000 },
  },
});

bus.subscribe('user.registered', 'send-welcome', async ({ userId }, { key, log }) => {
  await mailer.welcome(userId, { idempotencyKey: key });
  log.info({ userId }, 'welcome sent');
}, { concurrency: 5 });

bus.subscribe('user.registered', 'provision-workspace', async ({ userId }) => {
  await workspaces.provision(userId);
});

await bus.start();
await bus.publish('user.registered', { userId }, {
  key: `user:${userId}`,
  delay: 0,
});
await bus.stop();

Subscribers must be declared before start(). A subscriber has a stable name that is unique per event; declaring several subscribers on the same event is how fan-out happens. Bus defaults can be overridden per subscriber with attempts, backoff and concurrency; a publish only carries the required logical key and an optional delay.

The handler context is:

{ event, subscriber, eventId, key, attempt, attemptsLeft, signal, log }

signal aborts when stop() begins draining, per subscriber delivery; a long handler should observe it.

publish() returns { eventId, event, key, deliveries }, where deliveries lists one { subscriber, deliveryId } per subscriber reached. Publishing an event with no subscribers is a valid no-op that returns an empty deliveries list.

Identity

Each subscriber is an independent stream: the same event and key deliver once per subscriber while that delivery is waiting, delayed, active or retrying, and the identity is released after success or final failure. Two subscribers of the same event never share a queue and never collapse each other's deliveries, so one subscriber failing and retrying never blocks another.

Handlers must remain idempotent. Each subscriber gets at-least-once delivery: a process can finish the external effect and die before acknowledging the delivery.

Driver guarantees

| | memory | bullmq | |---|---|---| | External service | None | Redis | | Multiple replicas | One private bus per replica | One distributed queue per subscriber | | Survives restart | No | Yes | | Stalled redelivery | No | Yes | | Deduplication scope | Process | Cluster | | Fan-out isolation | Per process | One durable queue per subscriber |

The bullmq driver keeps one BullMQ queue and worker per (event, subscriber) pair, so each subscriber is a durable consumer group. Every replica declares the same subscribers; BullMQ routes each delivery to one worker within a subscriber, with no elected leader. The memory driver is for development, tests and single-process workloads.

./jobs

Background work with one deliberately small contract:

import { createJobQueue } from '@devindex/api-kit/jobs';

const jobs = createJobQueue({
  driver: 'bullmq',
  redisUrl,
  prefix: 'billing',
  logger,
  defaults: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1_000 },
  },
});

jobs.define('send-receipt', async ({ orderId }, { key, log }) => {
  await mailer.send(orderId, { idempotencyKey: key });
  log.info({ orderId }, 'receipt sent');
}, { concurrency: 5 });

await jobs.start();
await jobs.enqueue('send-receipt', { orderId }, {
  key: `receipt:${orderId}`,
  delay: 0,
});
await jobs.stop();

Jobs must be declared before start() and only declared names can be enqueued. Queue defaults can be overridden by a definition with attempts, backoff and concurrency; an enqueue only carries the required logical key and an optional delay.

The handler context is:

{ name, jobId, key, attempt, attemptsLeft, signal, log }

signal aborts when stop() begins draining; a long handler should observe it.

idle() resolves when no job is delayed, queued or running. Both drivers expose the same behavior, so tests and local tools do not need to know which backend is active.

Identity

The same name and key produce one job while that job is waiting, delayed, active or retrying. The identity is released after success or final failure. Permanent idempotency belongs in the database that commits the external effect.

Handlers must remain idempotent. A durable queue provides at-least-once delivery: a process can finish the external effect and die before acknowledging the job.

Driver guarantees

| | memory | bullmq | |---|---|---| | External service | None | Redis | | Multiple replicas | One private queue per replica | One distributed queue | | Survives restart | No | Yes | | Stalled redelivery | No | Yes | | Deduplication scope | Process | Cluster |

The memory driver is for development, tests and basic single-process workloads. It can run in many replicas, but each replica processes only the jobs enqueued into that process.

./schedule

Cron is independent from jobs. A schedule can execute a use case directly or enqueue a job by closing over a queue owned by the application.

import { createSchedule } from '@devindex/api-kit/schedule';

const schedule = createSchedule({
  driver: 'bullmq',
  redisUrl,
  prefix: 'billing',
  logger,
});

schedule.define(
  'daily-settlement',
  { pattern: '0 0 3 * * *', timeZone: 'America/Sao_Paulo' },
  async ({ signal, log }) => settle({ signal, log }),
);

await schedule.start();

The pattern supports the six-field cron form, with seconds first; the standard five-field form works too. The handler receives:

{ name, signal, log }

list() returns local declarations. remove(name) stops a local clock or explicitly removes the corresponding BullMQ Job Scheduler. stop() never removes Redis schedulers because one replica cannot know whether another still serves them.

Driver guarantees

| | memory | bullmq | |---|---|---| | Clock | Croner in each process | BullMQ Job Scheduler | | Multiple replicas | One run per replica | One cluster run per occurrence | | Survives restart | No | Yes | | Overlap of the same schedule | Skipped per process | Globally serialized | | Window with no replicas | Missed | One delayed run remains |

Every Redis replica upserts the same scheduler and starts an equivalent Worker. There is no elected leader; BullMQ coordinates which Worker receives each occurrence. A new occurrence is produced when the previous one starts, so global concurrency serializes slow runs rather than overlapping them.

./cache

Key/value cache with one contract over two stores:

import { createCache } from '@devindex/api-kit/cache';

const cache = createCache({
  driver: 'redis',
  redisUrl,
  prefix: 'billing',
  ttl: 300_000,
  logger,
});

await cache.start();

const plan = await cache.wrap(`plan:${planId}`, () => plans.findById(planId));

await cache.set(`quote:${userId}`, quote, { ttl: 60_000 });
const quote = await cache.get(`quote:${userId}`);
const known = await cache.has(`quote:${userId}`);

await plans.update(planId, changes);
await cache.delete(`plan:${planId}`);

await cache.stop();

Every key is namespaced by prefix, so plan:42 is stored as billing:plan:42. ttl is in milliseconds and 0 never expires; the cache default is overridable on each write.

Values must be JSON-serializable. undefined reports a miss and is never stored — writing it throws — while null is a value like any other, so get() returns null and has() returns true for it. Both drivers keep the serialized form, so the object the memory driver returns can be mutated without corrupting what is cached.

wrap() returns the cached value or runs the loader once, stores the result and returns it; a loader that resolves undefined is not cached. Concurrent misses on the same key run one loader each — the cache does not collapse them.

Failures

A cache read or write is an optimization and never fails a request: with the store unreachable, get() and has() report a miss and set() logs a warning, so the caller falls back to its source. delete() is the exception and rejects. An invalidation that did not happen keeps serving stale data until the TTL expires, which only the caller can weigh — .catch(() => {}) says so explicitly.

Driver guarantees

| | memory | redis | |---|---|---| | External service | None | Redis | | Scope | One private cache per replica | One cache shared by every replica | | Survives restart | No | Yes | | Expiration | On read | Owned by Redis | | Size bound | None | Redis eviction policy |

The memory driver frees an entry only when it is read after expiring, so a key written and never read again occupies memory until stop(). Keep it for development, tests and small derived values.

./env

createEnvReader() reads process.env and collects what is wrong instead of failing on the first problem, so a misconfigured deploy reports every missing or malformed variable in one boot rather than one per restart. It does not decide how to fail: issues() hands the list back and the service merges it with its own cross-field rules.

import { createEnvReader } from '@devindex/api-kit/env';

const env = createEnvReader();

export const config = Object.freeze({
  port: env.int('PORT', { fallback: 3000 }),
  mongoUri: env.str('MONGO_URI', { required: true }),
  driver: env.oneOf('MESSAGING_DRIVER', ['memory', 'bullmq'], { fallback: 'memory' }),
});

export function assertConfig() {
  const issues = env.issues();
  if (issues.length === 0) return;

  for (const issue of issues) console.error(`config: ${issue}`);
  process.exit(1);
}

str, int and oneOf take fallback (default null) and required (default false), and read an empty string as an absent value — a variable left blank in a .env is not a value. A rejected variable still returns its fallback, so the config object finishes building and issues() reports everything in one pass.

A reader owns its own list, so config split across several modules is just the reader passed to each one, and a test builds its own with createEnvReader({ PORT: '3000' }) without touching the process.

./runtime

onShutdown wires SIGINT/SIGTERM to a teardown callback and exits — the one lifecycle step services forget. It owns only the signal, running once and the exit code; the order of teardown and which components to stop stay yours, so any subset is just the calls you put in the callback.

import { onShutdown } from '@devindex/api-kit/runtime';

onShutdown(async () => {
  await app.close();               // stop accepting HTTP first
  await Promise.allSettled([       // then drain the consumers behind it
    jobs.stop({ timeoutMs: 10_000 }),
    bus.stop({ timeoutMs: 10_000 }),
    schedule.stop(),
  ]);
}, { logger });

A second signal arriving mid-drain is a no-op. A close that throws exits 1 after logging; one that hangs past timeoutMs (default 10_000) force-exits 1 so a stuck drain cannot wedge the process. signals defaults to ['SIGINT', 'SIGTERM'].

onFatalError covers the other exit: an uncaught exception or an unhandled rejection is logged as fatal and the process exits 1.

import { onFatalError } from '@devindex/api-kit/runtime';

onFatalError({ logger });

It deliberately does not run the shutdown callback. After an uncaught throw the process state is undefined, and a teardown running over it can hang or corrupt what it touches — exiting fast leaves the restart to the supervisor. The logger is flushed first, so a pretty transport writing from a worker thread does not lose the line that explains the crash.

Tests

The default suite exercises every memory path and skips integration tests when Redis is absent:

npm test -w @devindex/api-kit

Redis is mandatory in the integration command and in CI:

REDIS_URL=redis://127.0.0.1:6379 npm run test:redis -w @devindex/api-kit

License

MIT