Publish domain events to queues for asynchronous processing with reliability and delivery guarantees

`send` - Event Publishing

The send primitive enables you to publish domain events to queues for asynchronous processing. It provides reliable, ordered delivery of events with retry mechanisms, priority handling, and delivery guarantees.

Type Signature

function send<T = unknown>(event: PathProxy, data: T, options?: SendOptions): Promise<void>

interface SendOptions {
  priority?: number
  delay?: number
  retry?: {
    maxAttempts?: number
    backoff?: 'linear' | 'exponential'
  }
  source?: string
  traceId?: string
  correlationId?: string
  waitForDelivery?: boolean
}

Basic Usage

Simple Event Publishing

Publish an event with data:

import { send } from 'sdk.do'

// Publish an order created event
send.Order.created({
  id: 'ord_123',
  customerId: 'cus_456',
  total: 100,
  items: [{ productId: 'prod_789', quantity: 2, price: 50 }],
  createdAt: new Date(),
})

console.log('Event queued successfully')

Type-Safe Event Publishing

Use TypeScript generics for type-safe event data:

interface OrderCreatedEvent {
  id: string
  customerId: string
  total: number
  items: Array<{
    productId: string
    quantity: number
    price: number
  }>
  createdAt: Date
}

const orderData: OrderCreatedEvent = {
  id: 'ord_123',
  customerId: 'cus_456',
  total: 100,
  items: [{ productId: 'prod_789', quantity: 2, price: 50 }],
  createdAt: new Date(),
}

send.Order.created<OrderCreatedEvent>(orderData)

Event Publishing with Error Handling

Implement robust error handling for event publishing:

try {
  send.Payment.process({
    orderId: 'ord_123',
    amount: 100,
    customerId: 'cus_456',
  })

  console.log('Payment processing event queued')
} catch (error) {
  console.error('Failed to publish payment event:', error)

  // Fallback: Store for retry
  db.FailedEvent.create({
    type: 'Payment.process',
    data: { orderId: 'ord_123', amount: 100 },
    error: error.message,
    timestamp: new Date(),
  })

  throw error
}

Event Publishing Patterns

Domain Events

Publish events that represent state changes in your domain:

// Order lifecycle events
send.Order.created({
  orderId: 'ord_123',
  customerId: 'cus_456',
  total: 100,
})

send.Order.updated({
  orderId: 'ord_123',
  status: 'processing',
})

send.Order.shipped({
  orderId: 'ord_123',
  trackingNumber: 'TRK123456',
  carrier: 'UPS',
})

send.Order.completed({
  orderId: 'ord_123',
  completedAt: new Date(),
})

// User events
send.User.registered({
  userId: 'usr_789',
  email: '[email protected]',
  registeredAt: new Date(),
})

send.User.email.verified({
  userId: 'usr_789',
  verifiedAt: new Date(),
})

// Payment events
send.Payment.succeeded({
  paymentId: 'pay_321',
  orderId: 'ord_123',
  amount: 100,
  method: 'credit_card',
})

send.Payment.failed({
  paymentId: 'pay_321',
  orderId: 'ord_123',
  reason: 'insufficient_funds',
})

Command Events

Publish events that request actions to be performed:

// Email sending
send.Email({
  to: '[email protected]',
  subject: 'Order Confirmation',
  template: 'order-confirmation',
  data: {
    orderNumber: '12345',
    total: 100,
  },
})

// Notification requests
send.Notification({
  userId: 'usr_789',
  title: 'Payment received',
  message: 'Your payment of $100 has been processed',
  channels: ['email', 'push'],
})

// Data processing
send.Report.generate({
  type: 'sales',
  startDate: new Date('2025-01-01'),
  endDate: new Date('2025-01-31'),
  format: 'pdf',
})

// Background jobs
send.Job.start({
  type: 'data-export',
  userId: 'usr_789',
  format: 'csv',
  filters: { status: 'active' },
})

Integration Events

Publish events to trigger external integrations:

// Webhook events
send.Webhook.trigger({
  url: 'https://example.com/webhooks/order',
  method: 'POST',
  payload: {
    event: 'order.created',
    data: {
      orderId: 'ord_123',
      total: 100,
    },
  },
  headers: {
    'X-Webhook-Secret': 'secret_key',
  },
})

// Stripe integration
send.Stripe.charge.create({
  amount: 10000, // $100.00
  currency: 'usd',
  customerId: 'cus_123',
  description: 'Order payment',
})

// Analytics tracking
send.Analytics.track({
  event: 'purchase_completed',
  userId: 'usr_789',
  properties: {
    orderId: 'ord_123',
    total: 100,
    items: 2,
  },
})

Advanced Options

Priority Handling

Set event priority to control processing order (0-10, default: 5):

// High priority - process first
await send(
  $.Alert.critical,
  {
    message: 'System error detected',
    severity: 'critical',
  },
  {
    priority: 10,
  }
)

// Normal priority
await send(
  $.Order.created,
  {
    orderId: 'ord_123',
  },
  {
    priority: 5, // default
  }
)

// Low priority - process last
await send(
  $.Analytics.track,
  {
    event: 'page_view',
    page: '/home',
  },
  {
    priority: 1,
  }
)

Delayed Delivery

Schedule events for future delivery:

// Send reminder in 24 hours
await send(
  $.Email.send,
  {
    to: '[email protected]',
    subject: 'Complete your order',
    template: 'cart-reminder',
  },
  {
    delay: 24 * 60 * 60 * 1000, // 24 hours in milliseconds
  }
)

// Schedule payment reminder 7 days before due date
const dueDate = new Date('2025-02-15')
const reminderDate = new Date(dueDate.getTime() - 7 * 24 * 60 * 60 * 1000)
const delay = reminderDate.getTime() - Date.now()

await send(
  $.Payment.reminder,
  {
    invoiceId: 'inv_123',
    amount: 500,
    dueDate,
  },
  {
    delay,
  }
)

// Scheduled retry after failure
await send(
  $.Payment.retry,
  {
    paymentId: 'pay_321',
  },
  {
    delay: 60 * 60 * 1000, // 1 hour
  }
)

Retry Configuration

Configure retry behavior for failed event delivery:

// Exponential backoff with 5 attempts
await send(
  $.Payment.process,
  {
    amount: 100,
    customerId: 'cus_456',
  },
  {
    retry: {
      maxAttempts: 5,
      backoff: 'exponential', // 1s, 2s, 4s, 8s, 16s
    },
  }
)

// Linear backoff with 3 attempts
await send(
  $.Email.send,
  {
    to: '[email protected]',
    subject: 'Test',
  },
  {
    retry: {
      maxAttempts: 3,
      backoff: 'linear', // 1s, 2s, 3s
    },
  }
)

// No retry for time-sensitive operations
await send(
  $.Alert.realtime,
  {
    message: 'Live event started',
  },
  {
    retry: {
      maxAttempts: 1, // No retry
    },
  }
)

Distributed Tracing

Track events across system boundaries:

// Generate trace ID for request
const traceId = crypto.randomUUID()

// Pass trace ID through event chain
await send(
  $.Order.created,
  {
    orderId: 'ord_123',
  },
  {
    traceId,
    source: 'order-service',
  }
)

// Handler receives event and continues trace
on($.Order.created, async (order) => {
  await send(
    $.Payment.process,
    {
      orderId: order.orderId,
      amount: order.total,
    },
    {
      traceId: order.traceId, // Pass through
      correlationId: order.orderId, // Link related events
      source: 'payment-service',
    }
  )
})

Wait for Delivery

Block until event is delivered to handlers:

// Fire-and-forget (default)
await send($.Analytics.track, { event: 'page_view' })
console.log('Event queued')

// Wait for delivery confirmation
await send(
  $.Payment.process,
  {
    amount: 100,
  },
  {
    waitForDelivery: true,
  }
)
console.log('Event delivered and processed')

Email Sending Patterns

Transactional Emails

Send important transactional emails with templates:

// Order confirmation
await send($.Email.send, {
  to: '[email protected]',
  subject: 'Your order has been confirmed',
  template: 'order-confirmation',
  data: {
    orderNumber: '12345',
    orderDate: new Date(),
    items: [{ name: 'Product 1', quantity: 2, price: 50 }],
    total: 100,
    shippingAddress: '123 Main St',
  },
})

// Password reset
await send($.Email.send, {
  to: user.email,
  subject: 'Reset your password',
  template: 'password-reset',
  data: {
    name: user.name,
    resetLink: `https://app.example.com/reset?token=${token}`,
    expiresIn: '1 hour',
  },
})

// Invoice
await send($.Email.send, {
  to: customer.email,
  subject: `Invoice #${invoice.number}`,
  template: 'invoice',
  data: {
    invoiceNumber: invoice.number,
    dueDate: invoice.dueDate,
    items: invoice.items,
    total: invoice.total,
  },
  attachments: [
    {
      filename: `invoice-${invoice.number}.pdf`,
      content: invoicePdf,
    },
  ],
})

Bulk Email Campaigns

Send marketing emails with batching:

// Get subscriber list
const subscribers = await db.list('Subscriber', {
  where: { status: 'active', emailVerified: true },
})

// Batch send to avoid rate limits
for (const subscriber of subscribers) {
  await send(
    $.Email.send,
    {
      to: subscriber.email,
      subject: 'New product announcement',
      template: 'product-launch',
      data: {
        name: subscriber.name,
        unsubscribeLink: `https://app.example.com/unsubscribe?id=${subscriber.id}`,
      },
    },
    {
      priority: 3, // Lower priority for marketing
      retry: {
        maxAttempts: 2, // Fewer retries for bulk
      },
    }
  )

  // Track sent
  await db.create('EmailCampaignDelivery', {
    campaignId: 'camp_123',
    subscriberId: subscriber.id,
    sentAt: new Date(),
  })
}

Email with Dynamic Content

Generate email content with AI:

const emailContent = await ai.generate({
  prompt: `Write a personalized welcome email for ${user.name} who just signed up for our ${subscription.plan} plan.`,
  model: 'gpt-5',
  schema: {
    subject: 'string',
    greeting: 'string',
    body: 'string',
    callToAction: 'string',
  },
})

await send($.Email.send, {
  to: user.email,
  subject: emailContent.subject,
  template: 'custom',
  data: {
    greeting: emailContent.greeting,
    body: emailContent.body,
    callToAction: emailContent.callToAction,
  },
})

Webhook Patterns

Outbound Webhooks

Notify external systems of events:

// Order webhook
await send(
  $.Webhook.trigger,
  {
    url: 'https://partner.example.com/webhooks/order',
    method: 'POST',
    payload: {
      event: 'order.created',
      id: 'evt_123',
      data: {
        orderId: 'ord_123',
        customerId: 'cus_456',
        total: 100,
        createdAt: new Date(),
      },
    },
    headers: {
      'Content-Type': 'application/json',
      'X-Webhook-Signature': computeSignature(payload, secret),
    },
  },
  {
    retry: {
      maxAttempts: 5,
      backoff: 'exponential',
    },
  }
)

// Payment webhook
await send($.Webhook.trigger, {
  url: customer.webhookUrl,
  method: 'POST',
  payload: {
    event: 'payment.succeeded',
    data: {
      paymentId: 'pay_321',
      amount: 100,
      status: 'succeeded',
    },
  },
})

Webhook Retry Logic

Handle webhook delivery failures:

on($.Webhook.deliveryFailed, async (webhook) => {
  console.log('Webhook delivery failed:', webhook.url)

  // Store failed webhook
  await db.create('FailedWebhook', {
    url: webhook.url,
    payload: webhook.payload,
    error: webhook.error,
    attempts: webhook.attempts,
    failedAt: new Date(),
  })

  // Retry with exponential backoff
  if (webhook.attempts < 5) {
    await send($.Webhook.trigger, webhook, {
      delay: Math.pow(2, webhook.attempts) * 1000,
      retry: {
        maxAttempts: 5 - webhook.attempts,
      },
    })
  } else {
    // Alert admin after max retries
    await send($.Alert.send, {
      severity: 'warning',
      message: `Webhook delivery failed after ${webhook.attempts} attempts`,
      url: webhook.url,
    })
  }
})

Queue Patterns and Reliability

Guaranteed Delivery

Events are queued reliably with at-least-once delivery:

// Critical business event
await send(
  $.Order.created,
  {
    orderId: 'ord_123',
    customerId: 'cus_456',
    total: 100,
  },
  {
    priority: 9, // High priority
    retry: {
      maxAttempts: 10, // Many retries
      backoff: 'exponential',
    },
    waitForDelivery: true, // Confirm delivery
  }
)

Idempotent Handlers

Design handlers to handle duplicate events:

on($.Order.created, async (order) => {
  // Check if already processed
  const existing = await db.get('Order', order.orderId)
  if (existing) {
    console.log('Order already processed:', order.orderId)
    return // Idempotent - safe to ignore duplicates
  }

  // Process order
  await db.create('Order', {
    id: order.orderId,
    customerId: order.customerId,
    total: order.total,
    createdAt: new Date(),
  })
})

Event Ordering

Events are processed in order within the same queue:

// These events will be processed in order
await send($.Order.created, { orderId: 'ord_123' })
await send($.Order.validated, { orderId: 'ord_123' })
await send($.Order.shipped, { orderId: 'ord_123' })

// Handler processes in order
on($.Order['*'], async (order) => {
  console.log('Processing:', order.eventType, order.orderId)
  // Logs: created, validated, shipped (in order)
})

Batch Event Publishing

Publish multiple events efficiently:

// Publish many events
const events = products.map((product) => ({
  event: $.Product.created,
  data: {
    productId: product.id,
    name: product.name,
    price: product.price,
  },
}))

// Send all events
for (const { event, data } of events) {
  await send(event, data, {
    priority: 5,
  })
}

// Alternative: Use transaction pattern
await db.transaction(async (tx) => {
  for (const product of products) {
    await tx.create('Product', product)
    await send($.Product.created, { productId: product.id })
  }
})

Integration with Other Primitives

Event Publishing with Database Operations

Coordinate events with database changes:

// Create order and publish event atomically
async function createOrder(orderData) {
  // Create in database
  const order = await db.create('Order', {
    customerId: orderData.customerId,
    items: orderData.items,
    total: orderData.total,
    status: 'pending',
    createdAt: new Date(),
  })

  // Publish event
  await send($.Order.created, {
    orderId: order.id,
    customerId: order.customerId,
    total: order.total,
    createdAt: order.createdAt,
  })

  return order
}

// Update order and publish event
async function updateOrderStatus(orderId, status) {
  const order = await db.update('Order', orderId, {
    status,
    updatedAt: new Date(),
  })

  await send($.Order.statusChanged, {
    orderId: order.id,
    oldStatus: order.previousStatus,
    newStatus: status,
    changedAt: order.updatedAt,
  })

  return order
}

Event Publishing with AI Operations

Generate and publish AI-enhanced content:

// Generate product description with AI
const product = await db.get('Product', productId)

const description = await ai.generate({
  prompt: `Write a compelling product description for: ${product.name}`,
  model: 'gpt-5',
  schema: {
    title: 'string',
    description: 'string',
    features: 'array',
    benefits: 'array',
  },
})

// Update product
await db.update('Product', productId, {
  description: description.description,
  features: description.features,
})

// Publish update event
await send($.Product.updated, {
  productId,
  changes: ['description', 'features'],
  updatedBy: 'ai-content-generator',
})

Event Chains

Create complex workflows with event chains:

// Step 1: User registers
await send($.User.registered, {
  userId: 'usr_789',
  email: '[email protected]',
})

// Step 2: Handler sends verification
on($.User.registered, async (user) => {
  const token = crypto.randomUUID()

  await db.create('VerificationToken', {
    userId: user.userId,
    token,
    expiresAt: new Date(Date.now() + 24 * 60 * 60 * 1000),
  })

  await send($.Email.send, {
    to: user.email,
    template: 'email-verification',
    data: { token },
  })

  await send($.User.verification.emailSent, {
    userId: user.userId,
    sentAt: new Date(),
  })
})

// Step 3: User verifies email
await send($.User.email.verified, {
  userId: 'usr_789',
  verifiedAt: new Date(),
})

// Step 4: Handler activates account
on($.User.email.verified, async (user) => {
  await db.update('User', user.userId, {
    emailVerified: true,
    status: 'active',
  })

  await send($.User.activated, {
    userId: user.userId,
    activatedAt: new Date(),
  })

  await send($.Email.send, {
    to: user.email,
    template: 'welcome',
    data: { userId: user.userId },
  })
})

Real-World Examples

Example 1: E-commerce Order Processing

Complete order fulfillment workflow:

// Customer places order
async function placeOrder(customerId, items) {
  // Create order
  const order = await db.create('Order', {
    customerId,
    items,
    total: items.reduce((sum, item) => sum + item.price * item.quantity, 0),
    status: 'pending',
    createdAt: new Date(),
  })

  // Publish order created event
  await send(
    $.Order.created,
    {
      orderId: order.id,
      customerId: order.customerId,
      items: order.items,
      total: order.total,
    },
    {
      priority: 8, // High priority
      traceId: crypto.randomUUID(),
    }
  )

  return order
}

// Validate order
on($.Order.created, async (order) => {
  let valid = true
  const errors = []

  // Check inventory
  for (const item of order.items) {
    const product = await db.get('Product', item.productId)
    if (!product || product.inventory < item.quantity) {
      valid = false
      errors.push(`Insufficient inventory for ${item.productId}`)
    }
  }

  if (valid) {
    await send(
      $.Order.validated,
      {
        orderId: order.orderId,
      },
      {
        traceId: order.traceId,
      }
    )
  } else {
    await send(
      $.Order.validationFailed,
      {
        orderId: order.orderId,
        errors,
      },
      {
        traceId: order.traceId,
        priority: 9,
      }
    )
  }
})

// Process payment
on($.Order.validated, async (order) => {
  await send(
    $.Payment.process,
    {
      orderId: order.orderId,
      amount: order.total,
      customerId: order.customerId,
    },
    {
      traceId: order.traceId,
      retry: {
        maxAttempts: 5,
        backoff: 'exponential',
      },
    }
  )
})

// Fulfill order
on($.Payment.succeeded, async (payment) => {
  // Reserve inventory
  const order = await db.get('Order', payment.orderId)
  for (const item of order.items) {
    await db.update('Product', item.productId, {
      inventory: (product) => product.inventory - item.quantity,
    })
  }

  // Update order status
  await db.update('Order', payment.orderId, {
    status: 'processing',
    paidAt: new Date(),
  })

  // Request fulfillment
  await send(
    $.Order.fulfillmentRequested,
    {
      orderId: payment.orderId,
    },
    {
      traceId: payment.traceId,
    }
  )

  // Send confirmation email
  await send($.Email.send, {
    to: order.customerEmail,
    subject: 'Order confirmed',
    template: 'order-confirmation',
    data: { order },
  })
})

Example 2: Multi-Channel Notification System

Send notifications across multiple channels:

async function sendNotification(userId, notification) {
  // Get user preferences
  const user = await db.get('User', userId)
  const preferences = await db.get('NotificationPreferences', userId)

  // Determine channels
  const channels = []
  if (preferences.email) channels.push('email')
  if (preferences.sms) channels.push('sms')
  if (preferences.push) channels.push('push')

  // Publish notification request
  await send(
    $.Notification.send,
    {
      userId,
      title: notification.title,
      message: notification.message,
      channels,
      priority: notification.priority || 5,
    },
    {
      traceId: crypto.randomUUID(),
    }
  )
}

// Handle notification delivery
on($.Notification.send, async (notification) => {
  const user = await db.get('User', notification.userId)

  // Send to each channel
  for (const channel of notification.channels) {
    if (channel === 'email') {
      await send(
        $.Email.send,
        {
          to: user.email,
          subject: notification.title,
          body: notification.message,
        },
        {
          traceId: notification.traceId,
          correlationId: notification.userId,
        }
      )
    }

    if (channel === 'sms') {
      await send(
        $.SMS.send,
        {
          to: user.phone,
          message: notification.message,
        },
        {
          traceId: notification.traceId,
          correlationId: notification.userId,
        }
      )
    }

    if (channel === 'push') {
      await send(
        $.Push.send,
        {
          userId: user.id,
          title: notification.title,
          body: notification.message,
        },
        {
          traceId: notification.traceId,
          correlationId: notification.userId,
        }
      )
    }
  }

  // Log notification
  await db.create('NotificationLog', {
    userId: notification.userId,
    channels: notification.channels,
    sentAt: new Date(),
  })
})

Example 3: Asynchronous Job Processing

Process long-running jobs in the background:

// Request data export
async function requestDataExport(userId, options) {
  const job = await db.create('Job', {
    userId,
    type: 'data-export',
    status: 'queued',
    options,
    createdAt: new Date(),
  })

  await send(
    $.Job.queued,
    {
      jobId: job.id,
      userId,
      type: 'data-export',
      options,
    },
    {
      priority: 6,
      traceId: crypto.randomUUID(),
    }
  )

  return job
}

// Process job
on($.Job.queued, async (job) => {
  await db.update('Job', job.jobId, {
    status: 'processing',
    startedAt: new Date(),
  })

  await send(
    $.Job.started,
    {
      jobId: job.jobId,
      startedAt: new Date(),
    },
    {
      traceId: job.traceId,
    }
  )

  try {
    // Perform export
    const data = await db.list('Order', {
      where: { userId: job.userId },
    })

    const csv = convertToCSV(data)

    // Store result
    const file = await db.create('File', {
      userId: job.userId,
      content: csv,
      filename: `export-${job.jobId}.csv`,
      createdAt: new Date(),
    })

    await db.update('Job', job.jobId, {
      status: 'completed',
      completedAt: new Date(),
      resultId: file.id,
    })

    await send(
      $.Job.completed,
      {
        jobId: job.jobId,
        resultId: file.id,
        completedAt: new Date(),
      },
      {
        traceId: job.traceId,
      }
    )

    // Notify user
    await send($.Email.send, {
      to: job.userEmail,
      subject: 'Your export is ready',
      template: 'export-ready',
      data: {
        downloadUrl: `https://app.example.com/downloads/${file.id}`,
      },
    })
  } catch (error) {
    await db.update('Job', job.jobId, {
      status: 'failed',
      error: error.message,
      failedAt: new Date(),
    })

    await send(
      $.Job.failed,
      {
        jobId: job.jobId,
        error: error.message,
        failedAt: new Date(),
      },
      {
        traceId: job.traceId,
        priority: 9,
      }
    )
  }
})

Example 4: Event Sourcing Pattern

Implement event sourcing for audit trail:

// All state changes are events
async function updateUserProfile(userId, changes) {
  // Publish events for each change
  for (const [field, value] of Object.entries(changes)) {
    await send($.User.profile.updated, {
      userId,
      field,
      oldValue: changes[`old_${field}`],
      newValue: value,
      timestamp: new Date(),
    })
  }
}

// Rebuild state from events
on($.User.profile.updated, async (event) => {
  // Store event in event store
  await db.create('UserProfileEvent', {
    userId: event.userId,
    field: event.field,
    oldValue: event.oldValue,
    newValue: event.newValue,
    timestamp: event.timestamp,
    version: event.version,
  })

  // Update current state
  await db.update('UserProfile', event.userId, {
    [event.field]: event.newValue,
    updatedAt: event.timestamp,
  })

  // Rebuild projections
  await send($.Projection.rebuild, {
    entity: 'UserProfile',
    id: event.userId,
  })
})

Performance Considerations

Event Publishing Performance

Events are published asynchronously for performance:

// Fast - doesn't wait for processing
await send($.Analytics.track, { event: 'page_view' })
console.log('Event queued immediately')

// Slower - waits for delivery
await send(
  $.Payment.process,
  { amount: 100 },
  {
    waitForDelivery: true,
  }
)
console.log('Event processed')

Batch Publishing

Optimize for bulk event publishing:

// Inefficient - publishes one at a time
for (const order of orders) {
  await send($.Order.created, { orderId: order.id })
}

// Better - publish concurrently
await Promise.all(orders.map((order) => send($.Order.created, { orderId: order.id })))

// Best - use lower priority for bulk
for (const order of orders) {
  await send(
    $.Order.created,
    { orderId: order.id },
    {
      priority: 3, // Lower priority for bulk
    }
  )
}

Rate Limiting

Control event publishing rate:

async function publishWithRateLimit(events, maxPerSecond) {
  const delay = 1000 / maxPerSecond

  for (const event of events) {
    await send(event.type, event.data)
    await new Promise((resolve) => setTimeout(resolve, delay))
  }
}

// Publish 10 events per second
await publishWithRateLimit(events, 10)

Authentication Requirements

Event publishing requires authentication because it creates side effects.

Readonly Mode

Anonymous users cannot publish events:

// ❌ Requires authentication
await send($.Order.created, { orderId: 'ord_123' })

Authenticated Mode

With valid API key, event publishing works:

import { configure, send } from 'sdk.do'

configure({
  apiUrl: 'https://api.do',
  apiKey: process.env.SDK_DO_API_KEY,
})

// ✅ Works with authentication
await send($.Order.created, { orderId: 'ord_123' })

Common Pitfalls

Missing Error Handling

Always handle publishing errors:

// ❌ Bad: Errors crash the application
await send($.Order.created, orderData)

// ✅ Good: Handle errors gracefully
try {
  await send($.Order.created, orderData)
} catch (error) {
  console.error('Failed to publish event:', error)
  // Store for retry or alert admin
}

Publishing Too Much Data

Keep event payloads small:

// ❌ Bad: Publishing entire object
await send($.Order.created, {
  ...order,
  customer: { ...fullCustomerObject },
  items: items.map((item) => ({ ...fullProductObject })),
})

// ✅ Good: Only publish IDs
await send($.Order.created, {
  orderId: order.id,
  customerId: order.customerId,
  itemIds: items.map((item) => item.id),
  total: order.total,
})

Synchronous Dependencies

Don't wait for non-critical operations:

// ❌ Bad: Blocks on email sending
await send($.Email.send, emailData, { waitForDelivery: true })

// ✅ Good: Fire and forget
await send($.Email.send, emailData)
console.log('Email queued')

Delivery Guarantees

At-Least-Once Delivery

Events are delivered at least once to all handlers:

Events persist in queue until processed
Failed deliveries are retried automatically
Handlers may receive duplicate events

Ordering Guarantees

Events are processed in order within the same queue:

Order preserved for events on same semantic path
Cross-path ordering is not guaranteed
Use correlation IDs to track related events

Reliability

The event system provides strong reliability:

Events survive service restarts
Automatic retry with exponential backoff
Dead letter queue for failed events
Monitoring and alerting for failures

on - Subscribe to events published by send
every - Schedule recurring event publishing
db - Coordinate events with database operations
ai - Generate content for events

Best Practices

Use semantic patterns: Follow $.Subject.predicate.Object convention
Keep payloads small: Only send necessary data, use IDs for references
Handle errors: Always wrap send in try-catch
Set appropriate priority: Use high priority for critical events
Add trace IDs: Track events across system boundaries
Design idempotent handlers: Handle duplicate events gracefully
Use correlation IDs: Link related events in workflows
Batch when possible: Publish multiple events efficiently
Monitor delivery: Track failed events and retry rates
Document events: Maintain event schema documentation

Next Steps

Learn about on to handle published events
Explore every for scheduled publishing
Review Examples for real-world patterns
Check Authentication for security

send - Event Publishing

On this page