Skip to main content

Getting started with Events

There are three steps to working with events in Ductape:

  1. Set up a broker — connect a message provider to your product
  2. Define topics — declare the event streams your services will use
  3. Produce and consume — publish and receive events in your application code

Step 1: Set up a broker​

After running ductape init, open ductape/events.json. It contains a placeholder entry you can replace with your real broker configuration:

[
  {
    "tag": "order-events",
    "name": "Order events",
    "description": "All order-related events",
    "envs": [
      {
        "slug": "dev",
        "type": "REDIS",
        "config": {
          "host": "localhost",
          "port": 6379
        }
      },
      {
        "slug": "prd",
        "type": "KAFKA",
        "config": {
          "brokers": ["kafka.example.com:9092"],
          "clientId": "order-service",
          "groupId": "order-consumers",
          "ssl": true,
          "sasl": {
            "mechanism": "scram-sha-256",
            "username": "prod-user",
            "password": "prod-password"
          }
        }
      }
    ]
  }
]

Then apply it to your product:

ductape apply events

You can have multiple broker entries in the array — apply events syncs all of them.

Environment separation

Use a lightweight provider (Redis, NATS) for development and your production-grade provider (Kafka, RabbitMQ) for staging and production. The broker tag stays the same across environments; only the config inside each envs entry changes.

With the SDK​

import Ductape, { MessageBrokerTypes } from '@ductape/sdk';

const ductape = new Ductape({ accessKey: 'your-access-key' });

await ductape.events.create({
  name: 'Order events',
  tag: 'order-events',
  description: 'All order-related events',
  envs: [
    {
      slug: 'dev',
      type: MessageBrokerTypes.REDIS,
      config: { host: 'localhost', port: 6379 },
    },
    {
      slug: 'prd',
      type: MessageBrokerTypes.KAFKA,
      config: {
        brokers: ['kafka.example.com:9092'],
        clientId: 'order-service',
        groupId: 'order-consumers',
        ssl: true,
      },
    },
  ],
});

Provider config reference​

Kafka

{
  brokers: string[];
  clientId: string;
  groupId?: string;
  ssl?: boolean;
  sasl?: {
    mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
    username: string;
    password: string;
  };
}

RabbitMQ

{
  url: string;   // amqps://user:pass@host/vhost
}

Redis

{
  host: string;
  port: number;
  password?: string;
}

AWS SQS

{
  region: string;
  accessKeyId: string;
  secretAccessKey: string;
}

Azure Service Bus

{
  connectionString: string;  // Azure Service Bus connection string
  queueName: string;         // Default queue name
  namespace?: string;
}

Google Pub/Sub

{
  projectId: string;
  credentials: {
    private_key: string;    // Service account private key
    client_email: string;   // Service account email
    project_id?: string;
    // ... other service account JSON fields
  };
}

NATS

{
  servers: string[];
  token?: string;
  user?: string;
  pass?: string;
  tls?: boolean;
}

Step 2: Define topics​

Topics are the named event streams within a broker. Create one for each distinct event type your system produces or consumes.

The topic tag must be the full event identifier — brokerTag:topicTag:

await ductape.events.topics.create('my-product', {
  name: 'Order created',
  tag: 'order-events:order-created',
  description: 'Emitted when a new order is placed',
  sample: {
    orderId: '12345',
    customerId: 'cust_789',
    items: [{ sku: 'PROD-1', qty: 2, price: 49.99 }],
    total: 99.98,
    createdAt: '2026-07-03T10:00:00Z',
  },
});

await ductape.events.topics.create('my-product', {
  name: 'Order fulfilled',
  tag: 'order-events:order-fulfilled',
  description: 'Emitted when an order has been dispatched',
  sample: {
    orderId: '12345',
    dispatchedAt: '2026-07-03T14:00:00Z',
    trackingRef: 'TRK-9876',
  },
});

The sample field documents the expected payload shape. It is displayed in the Workbench and can be used for validation tooling.

List and inspect topics​

const topics = await ductape.events.topics.list('my-product', 'order-events');

const topic = await ductape.events.topics.fetch('my-product', 'order-events:order-created');
console.log(topic.name, topic.sample);

Step 3: Produce and consume​

Producing events​

Call events.produce() with the product, env, event string, and message payload:

await ductape.events.produce({
  product: 'my-product',
  env: 'prd',
  event: 'order-events:order-created',
  message: {
    orderId: '12345',
    customerId: 'cust_789',
    total: 99.98,
    createdAt: new Date().toISOString(),
  },
});

With user context — attach a session token so the event is traceable to the user who triggered it:

const session = await ductape.sessions.start({
  tag: 'user-session',
  data: { userId: 'u1', email: 'user@example.com' },
});

await ductape.events.produce({
  product: 'my-product',
  env: 'prd',
  event: 'order-events:order-created',
  message: { orderId: '12345', total: 99.98 },
  session: `${session.sessionId}:${session.token}`,
});

Consuming events​

Call events.consume() with the product, env, event string, and a callback. The callback receives each message as it arrives:

await ductape.events.consume({
  product: 'my-product',
  env: 'prd',
  event: 'order-events:order-created',
  callback: async (message) => {
    const { orderId, customerId, total } = message as {
      orderId: string;
      customerId: string;
      total: number;
    };

    await db.orders.update({ id: orderId, status: 'processing' });
    await sendConfirmationEmail(customerId);

    console.log(`Processed order ${orderId} — total £${total}`);
  },
});

If your callback throws, the broker records the failure. Messages that fail repeatedly are moved to the dead-letter queue.

NestJS integration​

In a NestJS service, use the @Events decorators from @ductape/nestjs instead of calling the SDK directly.

Producing​

import { Injectable } from '@nestjs/common';
import { Events } from '@ductape/nestjs';

@Injectable()
export class OrdersService {
  @Events.Produce({ event: 'order-events:order-created' })
  emitOrderCreated(payload: { orderId: string; total: number }) {
    return payload;
  }
}

The product and env defaults come from DuctapeModule.forIntegration(...). Pass them explicitly to override:

@Events.Produce({ event: 'order-events:order-created', product: 'my-product', env: 'prd' })

Consuming​

Use @Events.Consumer to bind a method as a message handler. DuctapeModule wires the subscription automatically at startup — no onModuleInit needed:

import { Injectable } from '@nestjs/common';
import { Events } from '@ductape/nestjs';

@Injectable()
export class OrderConsumerService {
  @Events.Consumer({ event: 'order-events:order-created' })
  async onOrderCreated(message: { orderId: string; total: number }) {
    await this.processOrder(message);
    // Return to acknowledge; throw to nack (triggers retry / DLQ)
  }

  private async processOrder(message: { orderId: string; total: number }) {
    // business logic
  }
}

Scheduling (dispatch)​

Use @Events.Dispatch to schedule a message through a Ductape job. The schedule can be static (fixed in the decorator) or dynamic (returned by the method):

import { Injectable } from '@nestjs/common';
import { Events } from '@ductape/nestjs';

@Injectable()
export class SchedulerService {
  // Static — fires every 24 h:
  @Events.Dispatch({
    broker: 'order-events',
    event: 'order-events:reminder-due',
    schedule: { every: 86400000 },
  })
  scheduleReminder(payload: { message: { orderId: string } }) {
    return payload;
  }

  // Dynamic — schedule derived from data at call time:
  @Events.Dispatch({ broker: 'order-events', event: 'order-events:fulfillment-due' })
  scheduleFulfillment(order: { id: string; expectedAt: number }) {
    return {
      message: { orderId: order.id },
      schedule: { start_at: order.expectedAt },
      retries: 3,
    };
  }
}

When the return value has a message key, the decorator reads message, schedule, and retries from it, overriding the decorator config. Use ctx.sdk.events.dispatch() directly when the broker or event also needs to vary at call time.

Observability​

Query produced and consumed messages, inspect dead letters, and get throughput stats:

// Message history
const { messages } = await ductape.events.messages.query({
  brokerTag: 'order-events',
  topicTag: 'order-created',
  page: 1,
  limit: 20,
});

// Dead letters
const { deadLetters } = await ductape.events.messages.getDeadLetters({
  brokerTag: 'order-events',
});

// Throughput stats
const stats = await ductape.events.messages.getStats({
  brokerTag: 'order-events',
});

Replay a failed event:

import { BrokersService } from '@ductape/sdk';

const brokers = new BrokersService({ access_key: 'your-access-key', env_type: 'prd' });

await brokers.replayEvent({ eventId: 'event-abc123', force: true });
await brokers.reprocessDLQ({ brokerTag: 'order-events' });

Cloud-linked brokers​

If your broker lives in a cloud account already connected to Ductape, use the cloud connection tag instead of providing credentials directly. This works for AWS SQS, GCP Pub/Sub, and Azure Service Bus.

Import an existing resource​

Reference an existing queue or topic by name:

import { MessageBrokerTypes } from '@ductape/sdk';

// AWS SQS — existing queue
await ductape.events.create({
  name: 'Order events',
  tag: 'order-events',
  envs: [{
    slug: 'prd',
    type: MessageBrokerTypes.AWS_SQS,
    config: { cloud: 'prod_aws', queueName: 'order-events', region: 'us-east-1' },
  }],
});

// GCP Pub/Sub — existing topic
await ductape.events.create({
  name: 'Order events',
  tag: 'order-events',
  envs: [{
    slug: 'prd',
    type: MessageBrokerTypes.GOOGLE_PUBSUB,
    config: { cloud: 'gcp_prod', topicName: 'order-events', region: 'us-central1' },
  }],
});

// Azure Service Bus — existing namespace and queue
await ductape.events.create({
  name: 'Order events',
  tag: 'order-events',
  envs: [{
    slug: 'prd',
    type: MessageBrokerTypes.AZURE_SERVICE_BUS,
    config: { cloud: 'prod_azure', namespaceName: 'my-namespace', queueName: 'order-events', region: 'eastus' },
  }],
});

Provision a new resource​

To create a new queue or topic in your cloud account from Ductape, use cloud.resources.provision with the cloud connection tag, then register the result as a broker using the import pattern above.

See Cloud-linked components for the full provision and import workflow with per-provider details.

See also​

  • CLI: ductape apply — declare and sync brokers from ductape/events.json
  • Features — use event topics as steps in a workflow
  • Jobs — schedule recurring produce operations