nuxt-nats

NATS JetStream integration for Nuxt. Server-side publish, typed consumers, KV and Object Store — wired into Nitro's lifecycle with zero boilerplate.

Status: Beta — production-validated since June 2026. Running in multi-replica Docker Swarm deployments with JWT+NKey auth, JetStream publish, ephemeral consumers, and 5+ KV buckets under real traffic. API is stable; breaking changes unlikely before 1.0.

• ✨  Release Notes

Features

• JetStream publish with automatic JSON encoding, retry, per-message deduplication (Nats-Msg-Id), typed tracing headers, and custom NATS message headers
• Pull consumers via defineNatsConsumer() or declared in nuxt.config, with opt-in durable provisioning, ackWait heartbeats, configurable backoff, and dead-letter routing
• Dead-letter handling via defineDeadLetterConsumer() — NATS has no dead-letter queue, so this consumes max-deliver and terminated advisories from a stream and recovers the original message
• Ephemeral consumers via useEphemeralConsumer() — request-scoped consumers with timeout, disconnect cleanup, and per-message error isolation (ideal for SSE endpoints)
• Connection lifecycle hooks via useNatsHooks() — attach onConnectError, onReconnect (once per outage), onDisconnect, onCredentialsRefreshed and onCredentialsError callbacks for alerting and metrics
• Synadia Cloud — nats: { synadia: true } connects to Synadia Cloud's TLS / WebSocket endpoints with a .creds file (NUXT_NATS_CREDS / NUXT_NATS_CREDS_FILE, re-read on every reconnect)
• Rotating credentials via nats.credentials — fetch NATS creds from Infisical (Kubernetes, AWS, GCP, Azure, OIDC or client-secret identity), the Synadia Control Plane, or your own defineNatsCredentialsProvider(); refreshed ahead of expiry and applied without a restart
• Synadia Control Plane client via useSynadiaCloud() — users, creds, nkey rotation, revocations, streams and KV buckets, typed
• nuxt-nats-rotate CLI — issues fresh creds on a schedule and writes them to Infisical, a file or your own store
• KV buckets via useKV(bucket) — cached per process
• Object Store via useObj(bucket) — stream large blobs through Nitro handlers
• Agent Fabric via defineNatsAgent() / useAgents() — expose the server as a discoverable AI agent on the NATS bus or call other agents, on the Synadia Agent Protocol (streaming, mid-stream human-in-the-loop, heartbeats)
• Stream auto-provisioning on startup (opt-in per stream, with 'update' mode for config reconciliation)
• Health endpoint at /api/_nats/health — connection status, auth method, RTT, JetStream account stats, registered agents, and (opt-in) credential-provider status
• Typed subjects — augment NatsEvents to get end-to-end type safety on jsPublish
• Graceful shutdown — stops agents and consumers, then drains the connection on SIGTERM/SIGINT (works around nitrojs/nitro#4015)
• Bun-ready — auto-detects Bun runtime and uses WebSocket transport

Requirements

• Nuxt >= 4.0.0 (tested on 4.6)
• Node.js ^20.19.0 || >= 22.12.0 (or Bun)
• NATS Server >= 2.10 with JetStream enabled

Setup

npm install nuxt-nats

Add the module to nuxt.config.ts:

export default defineNuxtConfig({
  modules: ['nuxt-nats'],

  nats: {
    servers: ['nats://localhost:4222'],
  },
})

Credentials and server URLs can be overridden at runtime via environment variables:

NUXT_NATS_SERVERS=nats://prod.example.com:4222
NUXT_NATS_TOKEN=your-auth-token

Usage

All server utilities are auto-imported inside server/ — no manual imports needed.

Publish to JetStream

// server/api/orders.post.ts
export default defineEventHandler(async (event) => {
  const body = await readBody(event)
  const traceId = getRequestHeader(event, 'x-trace-id') ?? crypto.randomUUID()

  await jsPublish('orders.created', {
    id: body.id,
    total: body.total,
  }, {
    msgId: body.id,                          // deduplication key
    traceId,                                 // sets X-Trace-Id header
    correlationId: traceId,                  // sets X-Correlation-Id header
  })

  return { ok: true }
})

Core publish (fire-and-forget)

// No PubAck, no durability — use for metrics or ephemeral events
corePublish('metrics.pageview', { path: '/home' })

KV Store

export default defineEventHandler(async (event) => {
  const kv = await useKV('sessions')

  await kv.put('user:123', JSON.stringify({ role: 'admin' }))

  const entry = await kv.get('user:123')
  return JSON.parse(entry?.string() ?? 'null')
})

Object Store

// Upload — putBlob() takes bytes; put() takes a ReadableStream<Uint8Array>
export default defineEventHandler(async (event) => {
  const data = await readRawBody(event, false) // Buffer, not a string
  const obs = await useObj('uploads')
  await obs.putBlob({ name: 'report.pdf' }, data ? new Uint8Array(data) : null)
  return { ok: true }
})

// Download
export default defineEventHandler(async () => {
  const obs = await useObj('uploads')
  const entry = await obs.get('report.pdf')
  return entry?.data   // ReadableStream
})

Ephemeral consumers (SSE / request-scoped)

For SSE endpoints that wait for a single matching event, use useEphemeralConsumer(). It creates an ordered, ephemeral JetStream consumer scoped to the request and handles timeout and client-disconnect cleanup automatically.

// server/api/orders/[id]/status.get.ts
export default defineEventHandler(async (event) => {
  const id = getRouterParam(event, 'id')!
  const stream = createEventStream(event)

  const handle = await useEphemeralConsumer({
    stream: 'ORDERS',
    filterSubjects: ['orders.*.shipped'],
    timeoutMs: 30_000,
    async onMessage(msg) {
      const payload = JSON.parse(new TextDecoder().decode(msg.data))
      if (payload.id !== id) return false          // not our order — keep waiting
      msg.ack()
      await stream.push({ event: 'shipped', data: JSON.stringify(payload) })
      await stream.close()
      return true                                  // done — stop the consumer
    },
    onTimeout: async () => {
      await stream.push({ event: 'timeout', data: '{}' })
      await stream.close()
    },
  })

  stream.onClosed(() => handle.stop())             // client disconnected
  return stream.send()
})

Connection lifecycle hooks

Register callbacks for NATS connection events — useful for alerting and metrics:

// server/plugins/nats-hooks.ts
export default defineNitroPlugin(() => {
  useNatsHooks({
    onConnectError: (err) => logger.error('NATS connect failed', err),
    onReconnect: (server) => metrics.increment('nats.reconnect', { server }),
    onDisconnect: (server) => logger.warn('NATS disconnected', { server }),
  })
})

Multiple useNatsHooks() calls accumulate — all registered callbacks are called in order. Hook errors are isolated and never affect the module.

Consumers

Workers run only when NUXT_NATS_WORKERS=true. This prevents long-lived consumers from starting in serverless or stateless environments. Without it the consumer logs a skip warning and the app publishes but consumes nothing.

Declare consumers in a Nitro server plugin. Nitro auto-registers server/plugins/**; it does not scan server/workers/, so a file there is never imported and the consumer never registers.

// server/plugins/billing.ts
export default defineNitroPlugin(() => {
  defineNatsConsumer({
    stream: 'ORDERS',
    durable: 'billing',
    filterSubjects: ['orders.created'],
    ackWait: 30_000,
    maxDeliver: 5,
    deadLetterSubject: 'orders.dlq',

    async handler(msg, payload) {
      await processBillingEvent(payload)
      msg.ack()
    },
  })
})

The durable has to exist

defineNatsConsumer binds to a durable, it does not create one by default, because a consumer's config is server-side state that usually belongs in IaC. Pass provision: 'startup' when you want the module to create it from the declared config instead:

defineNatsConsumer({
  stream: 'ORDERS',
  durable: 'billing',
  filterSubjects: ['orders.created'],
  provision: 'startup',   // create it if missing; default is 'never'
  async handler(msg) { msg.ack() },
})

Under the default provision: 'never', a missing durable is reported once with an actionable error rather than retried silently.

filterSubjects, ackPolicy, ackWait, maxDeliver and backoff describe the durable and are written to it only when this call creates it; ackWait, maxDeliver and backoff also drive the module's own heartbeat, dead-letter and nak timing on every run. When the durable already exists, a declared filterSubjects that disagrees with the live one is reported as a mismatch: binding cannot change a server-side filter, so the consumer receives the live set.

The same options can be declared in nuxt.config instead; see Declarative consumers. Either way, start the process that should consume with the flag set:

NUXT_NATS_WORKERS=true node .output/server/index.mjs

Tip: For production, run the Nuxt server (publisher) and a separate worker process (consumers) as distinct deployments. Workers need a persistent Node.js or Bun runtime — not serverless.

Declarative consumers in nuxt.config

As an alternative to defineNatsConsumer(), declare consumers in config. The module compiles them into a Nitro plugin at build time, statically importing each handler, so this works with a bundled server and with files under server/workers/ that Nitro does not scan.

// nuxt.config.ts
nats: {
  consumers: [
    {
      stream: 'ORDERS',
      durable: 'billing',
      filterSubjects: ['orders.created'],
      ackPolicy: 'explicit',
      ackWait: 30_000,
      maxDeliver: 5,
      deadLetterSubject: 'orders.dlq',
      provision: 'startup',
      handler: 'workers/billing',   // relative to server/, or absolute
    },
  ],
}

// server/workers/billing.ts
import type { JsMsg } from '@nats-io/jetstream'

export default async function (msg: JsMsg, payload: unknown) {
  await processBillingEvent(payload)
  msg.ack()
}

A missing stream, durable or handler fails the build, as does declaring two consumers on the same durable. Consumers still start only when NUXT_NATS_WORKERS=true.

Dead-letter handling

NATS has no dead-letter queue. Not in any released server, and not in 2.15-RC. When a message exhausts max_deliver the server drops it and publishes an advisory, and that advisory is the only signal you get.

deadLetterSubject on a consumer covers the common case: on the maxDeliver-th delivery this module republishes the message to that subject, then terminates it with a reason, which the server records in a MSG_TERMINATED advisory. If the republish still fails after jsPublish's retries, the failure is logged and the message is terminated anyway, so that advisory is its only record. The durable's own max_deliver must be at least maxDeliver (or unlimited), otherwise the server gives up first and the republish never happens; under provision: 'startup' the module creates the durable with max_deliver set to maxDeliver. For everything else, including messages that died on consumers you do not own, consume the advisories directly:

// nuxt.config.ts — capture advisories into a stream you own
nats: {
  streams: [
    {
      name: 'JS_ADVISORY',
      subjects: [
        '$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>',
        '$JS.EVENT.ADVISORY.CONSUMER.MSG_TERMINATED.>',
      ],
      retention: 'limits',
      storage: 'file',
      maxAge: '30d',
      provision: 'startup',
    },
  ],
}

// server/plugins/dead-letter.ts
export default defineNitroPlugin(() => {
  defineDeadLetterConsumer({
    stream: 'JS_ADVISORY',
    durable: 'dead-letter-handler',
    provision: 'startup',
    async onDeadLetter(event, msg) {
      // event.message is the ORIGINAL, fetched by sequence from event.stream
      await recordPoisonMessage({
        kind: event.kind,               // 'max_deliver' | 'terminated'
        stream: event.stream,
        consumer: event.consumer,
        seq: event.streamSeq,
        reason: event.reason,           // the msg.term(reason) string, when there was one
        body: event.message?.data,
      })
      msg.ack()
    },
  })
})

Two mistakes this exists to avoid. Do not use jsm.advisories(): it subscribes to $JS.EVENT.ADVISORY.>, the whole-account firehose, which includes an audit advisory published on every JetStream API response. And do not use a plain core subscription: advisories are fire-and-forget, so anything published while your subscriber is redeploying is gone permanently. Consuming them from a stream is what makes the failure path as durable as the happy path.

The advisory carries a stream sequence, not the message, so the original is fetched by sequence. That can legitimately return nothing if the message has already aged out of its stream, in which case event.message is null and the handler still runs.

MaxDeliverAdvisory and TerminatedAdvisory are exported. The client types Advisory.data as unknown and ships no payload types, so these are declared here. Note max_deliver has no consumer_seq; only terminated does.

Agent Fabric (Synadia Agent Protocol)

Expose the server as a discoverable AI agent, or call other agents on the bus. Like consumers, defineNatsAgent runs only when NUXT_NATS_WORKERS=true.

// server/plugins/assistant.ts — host an agent
export default defineNitroPlugin(() => {
  defineNatsAgent({
    agent: 'nuxt-assistant', owner: 'acme', name: 'web-1',
    async onPrompt(envelope, response) {
      for await (const token of llm.stream(envelope.prompt)) {
        await response.send(token)   // stream chunks back
      }
    },
  })
})

// server/api/ask.post.ts — call an agent
export default defineEventHandler(async (event) => {
  const { prompt } = await readBody(event)
  const [agent] = await useAgents().discover()
  if (!agent) return { error: 'no agents on the fabric' }

  let text = ''
  for await (const msg of await agent.prompt(prompt)) {
    if (msg.type === 'response') text += msg.text
  }
  return { response: text }
})

See the Agent Fabric guide for mid-stream human-in-the-loop, controller endpoints, and lifecycle details.

Synadia Cloud

// nuxt.config.ts
export default defineNuxtConfig({
  modules: ['nuxt-nats'],
  nats: { synadia: true }, // or { region: 'eu' } — fills in the TLS and WebSocket endpoints
})

NUXT_NATS_CREDS_FILE=/run/secrets/nats.creds   # or NUXT_NATS_CREDS="$(base64 < app.creds)"

Every stream you provision needs maxBytes on Synadia Cloud, and each instance holds one connection against your plan's limit. See the Synadia Cloud guide.

Rotating credentials

Short-lived creds can be fetched and refreshed at runtime instead of baked into env:

// nuxt.config.ts — Infisical with the pod's Kubernetes identity: no stored secret in the app
nats: {
  synadia: true,
  credentials: {
    provider: 'infisical',
    infisical: {
      projectId: '<project id>',
      environment: 'prod',
      secretName: 'NATS_CREDS',
      auth: { method: 'kubernetes', identityId: '<machine identity id>' },
    },
  },
}

The module fetches before connecting, refreshes 20% of the JWT lifetime early and reconnects when the creds change. provider: 'synadia' issues creds from the Control Plane instead, and provider: 'custom' loads your own:

// server/nats/credentials.ts (nats.credentials.customProvider: 'nats/credentials.ts')
export default defineNatsCredentialsProvider({
  name: 'vault',
  async fetch({ signal }) {
    return { creds: await readCredsFromVault(signal) } // or { userJwt, nkeySeed }
  },
})

Keep the store fresh with the bundled rotator, run on a schedule with a token scoped to one NATS user:

npx -p nuxt-nats nuxt-nats-rotate --user-id <nats user id> --min-remaining 12h --verify

See Credential providers and rotation for every auth method, the rotator options, and Kubernetes CronJob / GitHub Actions recipes.

Synadia Control Plane

// server/api/admin/revoke.post.ts — token via NUXT_NATS_SYNADIA_API_TOKEN (scope it to what the app needs)
export default defineEventHandler(async (event) => {
  const { userId } = await readBody(event)
  const cloud = useSynadiaCloud()
  const user = await cloud.natsUsers.get(userId)
  await cloud.natsUsers.revoke(user.account.id, user.user_public_key)
  return { revoked: user.user_public_key }
})

Typed subjects

Augment the NatsEvents interface to get full type safety across all jsPublish calls. Put the declaration in a .d.ts file under server/ (or shared/ if app code needs the types too), starting with the import type line:

// server/types/nats.d.ts
import type {} from 'nuxt-nats'

declare module 'nuxt-nats' {
  interface NatsEvents {
    'orders.created': { id: string; total: number }
    'user.registered': { id: string; email: string }
    'invoice.paid': { invoiceId: string; amount: number }
  }
}

Now jsPublish is typed per subject:

await jsPublish('orders.created', { id: '123', total: 99.99 })  // ✅
await jsPublish('orders.created', { id: '123', foo: 'bar' })    // ✗ type error
await jsPublish('metrics.pageview', { path: '/home' })          // ✅ not declared, so untyped

Undeclared subjects and string-typed subjects still compile, so you can adopt this one subject at a time. The Typed Events guide explains why the file location and that import matter.

Configuration

export default defineNuxtConfig({
  nats: {
    // Synadia Cloud: fills in servers/wsServers for the region. true = geo-routed global
    synadia: false,

    // TCP servers. Default: ['nats://localhost:4222']
    servers: ['nats://localhost:4222'],

    // WebSocket servers for the WebSocket transport (falls back to `servers` when empty)
    wsServers: ['wss://nats.example.com'],

    // 'auto' | 'tcp' | 'ws' — default 'auto': WebSocket when running on Bun, TCP otherwise
    transport: 'auto',

    // Connection name in server reports. Default: 'nuxt-nats@<hostname>:<pid>'
    name: '',

    // Auth — set credentials with env vars (values here are written into the build output).
    // The first match wins:
    // creds > credsFile > userJwt + nkeySeed > userJwt > nkeySeed > token > user/pass > anonymous
    creds: '',
    credsFile: '',
    userJwt: '',
    nkeySeed: '',
    token: '',
    user: '',
    pass: '',

    // TLS: caFile for server TLS; add certFile + keyFile for mTLS
    tls: {
      caFile: '/etc/ssl/certs/nats-ca.pem',
    },

    // -1 = infinite reconnects (default)
    maxReconnectAttempts: -1,

    // JetStream domain and API prefix for multi-tenant setups
    jsDomain: '',
    jsApiPrefix: '',

    // Stream definitions, created on boot when provision is 'startup' or 'update'
    streams: [
      {
        name: 'ORDERS',
        subjects: ['orders.>'],
        retention: 'limits',      // 'limits' | 'workqueue' | 'interest'
        storage: 'file',          // 'file' | 'memory'
        replicas: 1,
        maxAge: '7d',             // Go-style durations: '30s', '5m', '2h', '7d'
        maxBytes: 1_073_741_824,  // 1 GB
        duplicateWindow: '5m',
        placement: { tags: ['geo:europe'] }, // optional; tags or cluster
        provision: 'startup',     // 'startup' | 'update' | 'never' (default: 'never')
      },
    ],

    // Consumers compiled into a generated Nitro plugin (see Declarative consumers above)
    consumers: [],

    // Fetch and rotate credentials at runtime instead of the static settings above:
    // 'static' (default) | 'infisical' | 'synadia' | 'custom'. See docs/guides/credentials-rotation.md
    credentials: { provider: 'static' },

    // Synadia Control Plane API for useSynadiaCloud(); token via NUXT_NATS_SYNADIA_API_TOKEN
    synadiaApi: { url: 'https://cloud.synadia.com/api' },

    health: {
      enabled: true,
      endpoint: '/api/_nats/health',
      details: false, // true adds credential-provider status (public endpoint: off by default)
    },
  },
})

Environment variables

All runtimeConfig.nats.* values can be overridden at runtime. Prefix with NUXT_NATS_:

Authentication

The module selects an auth method based on which credentials are set, in this order:

1. Creds — creds (NUXT_NATS_CREDS, file contents raw or base64), then credsFile (NUXT_NATS_CREDS_FILE, a path). The standard format for Synadia Cloud and nsc generate creds. A creds file is re-read on every reconnect, so rotating it needs no restart. See the Synadia Cloud guide.
2. JWT + NKey (production) — when both userJwt and nkeySeed are set, the module uses jwtAuthenticator(jwt, seed) from @nats-io/nats-core. This is the standard for NATS servers configured with the JWT resolver (nsc operator/account/user hierarchy). The JWT is sent during CONNECT; the NKey seed is used to sign the server's nonce to prove possession of the private key.
3. JWT (unsigned or bearer) — when userJwt is set without nkeySeed, uses jwtAuthenticator(jwt). The JWT is sent unsigned — usable only against servers explicitly configured to accept unsigned JWTs, such as when identity is pinned out-of-band by operator policy or in test environments.
4. NKey only (dev) — when only nkeySeed is set, uses nkeyAuthenticator(seed). For static NKey-based servers without a JWT resolver.
5. Token — when only token is set.
6. User / pass — when only user (and optionally pass) is set.
7. Anonymous — when none of the above are set.

With nats.credentials.provider set to infisical, synadia or custom, a credentials provider supplies the JWT and seed instead, and this list does not apply. See Credential providers.

JWT Auth (production)

Generate a user JWT and NKey seed with nsc (nsc generate creds) and pass them via env vars — never commit them to source:

NUXT_NATS_USER_JWT='eyJ0eXAiOiJqd3Q...'  # full user JWT
NUXT_NATS_NKEY_SEED='SUACSP3ZI...'       # matching user NKey seed (omit for unsigned JWT)

On startup the module checks the JWT's exp claim and logs a warning if it expires within 24 hours, or an error if it is already expired. Connection-status errors that mention Authorization, Permissions Violation or Authentication Expired are logged with an AUTH ERROR prefix so they are distinguishable from network errors; see Auth errors for the exact format. See the NATS JWT guide for chain-of-trust details.

Health endpoint

GET /api/_nats/health

{
  "connected": true,
  "status": "ok",
  "server": "nats://localhost:4222",
  "auth": { "mode": "creds-file" },
  "rttMs": 1,
  "jetstream": {
    "available": true,
    "streams": 3,
    "consumers": 7,
    "memory": 0,
    "storage": 204800
  }
}

auth.mode is the method in use (creds, creds-file, jwt-nkey, jwt, nkey, token, user-pass, anonymous or provider:<name>), never an identity or secret. With health: { details: true } and a credentials provider active, auth also carries the provider status and seconds to expiry. When agents are registered in the process, the response also carries an agents array, and a process with no connection returns { "connected": false, "status": "disconnected" }. See the API reference.

Architecture notes

NATS lives on the server

The module only injects server-side utilities. Browser composables for NATS are intentionally excluded — credentials, JetStream consumers, and reconnect logic are server concerns.

Browser → Nuxt Server API → NATS / JetStream

Serverless vs worker mode

For Cloudflare Workers, set transport: 'ws' and configure wsServers. Publish-only; consumers are not supported on edge runtimes.

Stream provisioning

Use 'never' in production with external IaC. Use 'startup' for local dev where you want idempotent creation. Use 'update' when you need the app to own the authoritative stream config (e.g. the stream is shared and the app is responsible for keeping subjects up to date).

Contribution

# Install dependencies
npm install

# Generate type stubs and prepare playground
npm run dev:prepare

# Start dev server (requires NATS on localhost:4222)
npm run dev

# Lint and type check (module + playground)
npm run lint
npm run test:types

# Unit tests; integration tests need Docker (Testcontainers)
npm test
npm run test:integration

# Build
npm run prepack

See CONTRIBUTING.md for the pull request workflow and release steps.