@crossdelta/cloudevents

Type-safe event-driven microservices with NATS and Zod validation, using the CloudEvents specification.

                         NATS JetStream
                        ┌──────────────┐
  ┌──────────────┐      │              │      ┌──────────────┐
  │    Service   │      │   Stream:    │      │   Service    │
  │   (publish)  │─────▶│   ORDERS     │─────▶│  (consume)   │
  └──────────────┘      │              │      └──────────────┘
                        └──────────────┘
        │                                            │
        │  publish(...)                              │  handleEvent(...)
        ▼                                            ▼
  ┌──────────────┐                            ┌──────────────┐
  │ { orderId,   │                            │ Zod schema   │
  │   total }    │                            │ validates    │
  └──────────────┘                            └──────────────┘

bun add @crossdelta/cloudevents zod@4

Prerequisites: A running NATS server with JetStream enabled.

Note: Requires Zod v4 for full TypeScript support.

Quick Start

Note: Services never create streams. Streams must exist before services consume from them:

• Development: Use NATS CLI or Platform SDK (pf dev auto-creates from contracts)
• Production: Use infrastructure-as-code (Pulumi, Terraform) or the Stream Setup function below

1. Create an event handler (src/events/order-created.handler.ts):

import { handleEvent } from '@crossdelta/cloudevents'
import { z } from 'zod'

// Export schema for mock generation
export const OrderCreatedSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

// Export type for use in use-cases
export type OrderCreatedEvent = z.infer<typeof OrderCreatedSchema>

export default handleEvent(
  {
    schema: OrderCreatedSchema,
    type: 'order.created',
  },
  async (data) => {
    console.log(`New order: ${data.orderId}, total: ${data.total}`)
  },
)

2. Start consuming:

import { consumeJetStreams } from '@crossdelta/cloudevents'

consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
})

3. Publish from another service:

import { publish } from '@crossdelta/cloudevents'

await publish('order.created', { orderId: 'ord_123', total: 99.99 })
// String API auto-pluralizes the domain token: subject → 'orders.created'
// Use createContract({ type, channel, schema }) for explicit subject control (no pluralization)

That's it. Handlers are auto-discovered, validated with Zod, and messages persist in JetStream.

Stream creation is handled by your development environment or infrastructure — not by services:

• Development: NATS CLI, @crossdelta/platform-sdk, or manual setup
• Production: See Stream Setup below for ensureJetStreams() usage in infrastructure code

Why use this?

Core Concepts

Event Type vs. Event Data

Important distinction:

• Event Type (order.created): Lives in the CloudEvent envelope (ce.type). Used for routing and handler matching.
• Event Data ({ orderId, total }): The actual payload. Does not include the type.

const Schema = z.object({
  orderId: z.string(), 
})

export default handleEvent(
  {
    schema: Schema,
    type: 'order.created',
  },
  async (data) => { ... }
)

Handlers

Drop a *.handler.ts file anywhere — it's auto-registered:

// src/events/user-signup.handler.ts
import { z } from 'zod'

const UserSignupSchema = z.object({ 
  email: z.email(),
  name: z.string(),
})

// Export type for use in use-cases
export type UserSignupEvent = z.infer<typeof UserSignupSchema>

export default handleEvent(
  {
    schema: UserSignupSchema,
    type: 'user.signup',
  },
  async (data) => {
    await sendWelcomeEmail(data.email)
  },
)

Publishing

await publish('order.created', orderData)

Stream Setup

For infrastructure use only - Services should never call this function.

Create streams during infrastructure setup (Pulumi, deployment scripts):

import { ensureJetStreams } from '@crossdelta/cloudevents'

await ensureJetStreams({
  streams: [
    {
      stream: 'ORDERS',
      subjects: ['order.>'],  // singular — matches contract subjects (order.created, order.completed, …)
      config: {
        maxAge: 7 * 24 * 60 * 60 * 1000, // 7 days retention
        replicas: 3,                       // For HA
      },
    },
  ],
})

Consuming

import { consumeJetStreams } from '@crossdelta/cloudevents'

// JetStream (recommended) — persistent, retries, exactly-once
consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'billing',
  discover: './src/events/**/*.handler.ts',
})

// Core NATS — fire-and-forget, simpler
await consumeNatsEvents({
  subject: 'notifications.*',
  consumerName: 'my-service',
  discover: './src/events/**/*.handler.ts',
})

// NATS Core Request-Reply — synchronous validation feedback
// Consumer returns result, publisher awaits it
import { request } from '@crossdelta/cloudevents'
const result = await request('call.received', payload)
// result: { accepted: true, data: { ... } } | { accepted: false, error: '...' }

Configuration

Environment Variables

NATS_URL=nats://localhost:4222
NATS_USER=myuser        # optional
NATS_PASSWORD=mypass    # optional

Optional: Google Cloud Pub/Sub

To use Google Cloud Pub/Sub instead of NATS, install the optional dependency:

bun add @google-cloud/pubsub
# or
npm install @google-cloud/pubsub

Then use the publishToPubSub function:

import { publishToPubSub } from '@crossdelta/cloudevents'

await publishToPubSub('order.created', { orderId: 'ord_123', total: 99.99 })

Note: @google-cloud/pubsub is not bundled with this library to keep the package size small. Install it only if you need Google Cloud Pub/Sub support.

Stream Options

await ensureJetStreams({
  streams: [
    {
      // Required
      stream: 'ORDERS',
      subjects: ['order.>'],  // singular — matches contract subjects

      // Optional
      config: {
        maxAge: 7 * 24 * 60 * 60 * 1000,  // Message retention (ms)
        maxBytes: 1024 * 1024 * 1024,       // Reserved at creation, not on write
        replicas: 1,                        // Replication factor
        storage: 'file',                    // 'file' or 'memory'
        retention: 'limits',                // 'limits', 'interest', or 'workqueue'
      },
    },
  ],
  // Optional
  servers: 'nats://localhost:4222',
})

Consumer Options

consumeJetStreams({
  // Required
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',

  // Optional
  servers: 'nats://localhost:4222',
  filterSubjects: ['order.created'],  // Filter specific subjects
  maxDeliver: 5,                       // Retry attempts
  ackWait: 30_000,                     // Timeout per attempt (ms)
  quarantineTopic: 'dlq',              // For poison messages
})

Advanced Features

import { createContract } from '@crossdelta/cloudevents'
import { z } from 'zod'

export const OrdersCreatedContract = createContract({
  type: 'order.created',
  channel: {
    stream: 'ORDERS',
    // subject defaults to 'order.created' if omitted
  },
  schema: z.object({
    orderId: z.string(),
    total: z.number(),
  }),
})

interface ChannelConfig {
  stream: string    // JetStream stream name (e.g., 'ORDERS')
  subject?: string  // NATS subject (optional override)
}

// subject = 'order.created' (from type)
createContract({ type: 'order.created', channel: { stream: 'ORDERS' } })

// subject = 'orders.v1.created' (explicit override)
createContract({ type: 'order.created', channel: { stream: 'ORDERS', subject: 'orders.v1.created' } })

// order.created → ORDERS stream
export const OrdersCreatedContract = createContract({
  type: 'order.created',
  channel: { stream: 'ORDERS' },
  schema: OrderCreatedSchema,
})

// order.updated → ORDERS stream (same stream!)
export const OrdersUpdatedContract = createContract({
  type: 'order.updated',
  channel: { stream: 'ORDERS' },
  schema: OrderUpdatedSchema,
})

import { collectStreamDefinitions } from './helpers/streams'
import { ensureJetStreams } from '@crossdelta/cloudevents'

// Collect from contracts
const streams = collectStreamDefinitions()
// [{ stream: 'ORDERS', subjects: ['order.created', 'order.updated'] }]

// Materialize with policies
await ensureJetStreams({
  streams: streams.map(({ stream, subjects }) => ({
    stream,
    subjects,
    config: {
      maxAge: 14 * 24 * 60 * 60 * 1000, // 14 days
      replicas: 3,                       // High availability
      storage: 'file',                   // Persistent
    },
  })),
})

// Legacy contract (still works)
export const LegacyContract = createContract({
  type: 'legacy.event',
  schema: LegacySchema,
  // No channel - handler still works
})

// packages/contracts/src/events/orders/created.ts
import { createContract } from '@crossdelta/cloudevents'
import { z } from 'zod'

export const OrderCreatedSchema = z.object({
  orderId: z.string(),
  total: z.number(),
})

export type OrderCreatedData = z.infer<typeof OrderCreatedSchema>

export const OrdersCreatedContract = createContract({
  type: 'order.created',
  schema: OrderCreatedSchema,
})

import { handleEvent } from '@crossdelta/cloudevents'
import { OrdersCreatedContract } from '@my-org/contracts'

export default handleEvent(
  OrdersCreatedContract,
  async (data) => {
    // data is typed as OrderCreatedData
    console.log(data.orderId)
  },
)

export default handleEvent({
  type: 'order.created',
  schema: OrderSchema,
  tenantId: 'tenant-a',  // Only process tenant-a events
}, async (data) => { ... })

import { handleEvent } from '@crossdelta/cloudevents'
import { z } from 'zod'

export default handleEvent(
  {
    type: '*',
    schema: z.record(z.string(), z.unknown()),
  },
  async (data, context) => {
    await persistAuditEvent({
      eventType: context.eventType,
      source: context.source,
      data,
    })
  },
)

export default handleEvent({
  type: 'order.created',
  schema: OrderSchema,
  match: (event) => event.data.total > 100,  // Only high-value orders
}, async (data) => { ... })

const redisStore = {
  has: (id) => redis.exists(`idem:${id}`),
  add: (id, ttl) => redis.set(`idem:${id}`, '1', 'PX', ttl),
}

await consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
  idempotencyStore: redisStore,
})

await consumeJetStreams({
  streams: ['ORDERS'],
  consumer: 'my-service',
  discover: './src/events/**/*.handler.ts',
  quarantineTopic: 'events.quarantine',  // For malformed messages
  errorTopic: 'events.errors',            // For handler errors
})

import { Hono } from 'hono'
import { cloudEvents } from '@crossdelta/cloudevents'

const app = new Hono()
app.use('/events', cloudEvents({ discover: 'src/events/**/*.handler.ts' }))

API

Distributed Tracing

When @crossdelta/telemetry is installed (optional peer dependency), trace context is propagated automatically across NATS:

• Publish: The current W3C traceparent is injected as a CloudEvent extension field.
• Consume: The traceparent extension is extracted and restored in AsyncLocalStorage before the handler runs.

This enables end-to-end distributed traces in Grafana:

Storefront → API Gateway → NATS → Orders Service → NATS → Audit Service
                └── all spans share the same traceId ──┘

No configuration required. If @crossdelta/telemetry is not installed, CloudEvents work normally without trace propagation.

License

MIT © crossdelta