Preview

Preview Feature — This feature is currently in preview and under active development. APIs and functionality may change. We recommend testing thoroughly before using in production.

Message Brokers

Message brokers let you build event-driven applications by sending and receiving messages between services. Ductape supports multiple providers through a unified interface exposed as ductape.events.

Architecture: brokers vs topics

A broker is a connection configuration — it holds credentials and per-environment settings for one message provider. It has no topics of its own. Topics are added separately after the broker is registered and there is no limit on how many you can create per broker.

Concept	What it is
Broker	Connection config (credentials + per-env settings). One broker per logical provider setup.
Topic	A named event stream within a broker. Added separately. One broker can hold unlimited topics.
Event	A message produced to or consumed from a topic. Always referenced as brokerTag:topicTag.

Quick Example

Events use the format brokerTag:topicTag (e.g. order-events:order-created):

TypeScript

import Ductape from '@ductape/sdk';

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

// Produce
await ductape.events.produce({
  event: 'order-events:order-created',
  message: { orderId: '123', amount: 99.99 },
});

// Consume
await ductape.events.consume({
  event: 'order-events:order-created',
  callback: async (message) => {
    console.log('Received:', message);
  },
});

Supported Providers

Provider	Type constant	Best For
Kafka	MessageBrokerTypes.KAFKA	High-throughput distributed streaming
RabbitMQ	MessageBrokerTypes.RABBITMQ	Flexible routing, reliable delivery
Redis	MessageBrokerTypes.REDIS	Simple pub/sub, low latency
AWS SQS	MessageBrokerTypes.AWS_SQS	Serverless managed queues
Azure Service Bus	MessageBrokerTypes.AZURE_SERVICE_BUS	Azure managed queues
Google Pub/Sub	MessageBrokerTypes.GOOGLE_PUBSUB	GCP managed messaging
NATS	MessageBrokerTypes.NATS	Lightweight, high-performance messaging

---

Producing Messages

TypeScript

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

// Returns { success: true, process_id: '...' }

With session (user context)

TypeScript

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

await ductape.events.produce({
  event: 'order-events:order-created',
  message: { orderId: '123', total: 99.99 },
  session: `${session.sessionId}:${session.token}`,
});

---

Consuming Messages

TypeScript

await ductape.events.consume({
  event: 'order-events:order-created',
  callback: async (message) => {
    console.log('Received order:', message);
    await processNewOrder(message);
  },
});

---

Listing and Fetching Brokers and Topics

TypeScript

// List all brokers for a product
const brokers = await ductape.events.list('my-product');

// Fetch a single broker
const broker = await ductape.events.fetch('my-product', 'order-events');
console.log('Environments:', broker.envs?.map((e) => e.slug));

// List topics for a broker
const topics = await ductape.events.topics.list('my-product', 'order-events');

// Fetch a single topic (full event string)
const topic = await ductape.events.topics.fetch('my-product', 'order-events:order-created');
console.log('Sample:', topic.sample);

---

Creating a Broker

A broker holds per-environment credentials. Topics are always created separately — see Managing Topics.

Self-hosted providers

TypeScript

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

await ductape.events.create({
  name: 'Order Events',
  tag: 'order-events',
  description: 'Handles all order-related messages',
  envs: [
    {
      slug: 'prd',
      type: MessageBrokerTypes.KAFKA,
      config: {
        brokers: ['kafka-prod.example.com:9092'],
        clientId: 'order-service',
        groupId: 'order-consumers',
        ssl: true,
        sasl: {
          mechanism: 'scram-sha-256',
          username: 'prod-user',
          password: 'prod-password',
        },
      },
    },
    {
      slug: 'dev',
      type: MessageBrokerTypes.REDIS,
      config: { host: 'localhost', port: 6379 },
    },
  ],
});

Provider config reference

Kafka

TypeScript

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

RabbitMQ

TypeScript

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

Redis

TypeScript

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

AWS SQS

TypeScript

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

Google Pub/Sub

TypeScript

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

Azure Service Bus

TypeScript

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

NATS

TypeScript

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

---

Cloud-linked brokers

Provision or import queues/topics from a workspace cloud connection. Add cloud: '<connection-tag>' to the env config instead of direct credentials.

Import an existing resource

TypeScript

// AWS SQS
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
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
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/topic in your cloud account, use cloud.resources.provision with the cloud connection tag. Once provisioned, register it as a broker using the import pattern above.

See Cloud-linked components for the full provision and import workflow.

---

Managing Topics

The topic tag must be the full event identifier: brokerTag:topicTag. A broker can have unlimited topics.

Create a topic

TypeScript

await ductape.events.topics.create('my-product', {
  name: 'Order Created',
  tag: 'order-events:order-created',
  description: 'Emitted when an order is created',
  sample: {
    orderId: '12345',
    customerId: 'cust_789',
    total: 99.99,
    createdAt: '2024-01-15T10:30:00Z',
  },
});

// Add more topics to the same broker — no limit
await ductape.events.topics.create('my-product', {
  name: 'Order Fulfilled',
  tag: 'order-events:order-fulfilled',
  sample: { orderId: '12345', dispatchedAt: '2024-01-15T14:00:00Z' },
});

For AWS SQS, each topic maps to a separate queue per environment via queueUrls:

TypeScript

await ductape.events.topics.create('my-product', {
  name: 'Order Created',
  tag: 'order-events:order-created',
  queueUrls: [
    { env_slug: 'prd', url: 'https://sqs.us-east-1.amazonaws.com/123/orders-prd' },
    { env_slug: 'dev', url: 'https://sqs.us-east-1.amazonaws.com/123/orders-dev' },
  ],
  sample: { orderId: '12345' },
});

Update a topic

TypeScript

await ductape.events.topics.update('my-product', 'order-events:order-created', {
  description: 'Updated description',
  sample: { orderId: '12345', status: 'pending' },
});

---

Event format

Events always use brokerTag:topicTag:

Event	Broker tag	Topic tag
order-events:order-created	order-events	order-created
order-events:payment-processed	order-events	payment-processed
notifications:user-alerts	notifications	user-alerts

---

Message tracking

TypeScript

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

const { producers } = await ductape.events.messages.getProducers({ brokerTag: 'order-events' });
const { consumers } = await ductape.events.messages.getConsumers({ brokerTag: 'order-events' });
const { deadLetters } = await ductape.events.messages.getDeadLetters({ brokerTag: 'order-events' });
const stats = await ductape.events.messages.getStats({ brokerTag: 'order-events' });

---

Advanced: BrokersService (replay, DLQ, idempotency)

Method	Description
publish(options)	Same as events.produce()
subscribe(options)	Same as events.consume()
getBrokers(product)	Same as events.list(product)
getBroker(product, brokerTag)	Same as events.fetch(product, brokerTag)
getTopics(product, brokerTag)	Same as events.topics.list(product, brokerTag)
getTopic(product, event)	Same as events.topics.fetch(product, event)
getEvents(options)	Event history with filters
replayEvent(options)	Replay a failed or successful event
reprocessDLQ(options)	Reprocess dead-letter queue
publishIdempotent(options)	Publish with idempotency key

TypeScript

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

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

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

---

Error handling

TypeScript

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

try {
  await ductape.events.produce({ event: '...', message: {} });
} catch (error) {
  if (error instanceof BrokerError) {
    console.log('Code:', error.code);
    console.log('Message:', error.message);
  }
}

Common codes: BROKER_NOT_FOUND, BROKER_ENV_NOT_FOUND, TOPIC_NOT_FOUND, PUBLISH_FAILED, SUBSCRIBE_FAILED, SESSION_INVALID.

---

See also

• Events — same API, additional detail on provider setup and getting started
• Cloud-linked components — provision or link AWS SQS, GCP Pub/Sub, Azure Service Bus
• Features — use broker topics as steps in a feature workflow
• Jobs — schedule recurring produce operations