How to Dispatch Events

A guide for platform developers implementing EEP event delivery.

1. Set Up an Event Bus

EEP doesn't mandate a specific message broker. Use Redis Streams, RabbitMQ, Kafka, or any reliable queue.

Publish Event

import { randomUUID } from 'crypto';

await eventBus.publish({
  specversion: '1.0',
  id: randomUUID(),
  source: entity.did,
  type: 'com.example.entity.updated',
  time: new Date().toISOString(),
  datacontenttype: 'application/json',
  eep_version: '0.1',
  data: { field: 'bio', previous: oldBio, current: newBio },
});

2. Implement Webhook Delivery

Deliver Webhook

import { EEPSigner } from '@eep-dev/signer';
import { validateSSRF } from '@eep-dev/validator';

async function deliverWebhook(subscription, event) {
  // SSRF check BEFORE connecting
  await validateSSRF(subscription.delivery_url);

  const signer = new EEPSigner(subscription.delivery_secret);
  const webhookId = `msg_${event.id}`;
  const timestamp = Math.floor(Date.now() / 1000).toString();
  const body = JSON.stringify(event);
  const signature = signer.sign(webhookId, timestamp, body);

  const response = await fetch(subscription.delivery_url, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'webhook-id': webhookId,
      'webhook-timestamp': timestamp,
      'webhook-signature': `v1,${signature}`,
      'EEP-Version': '0.1',
    },
    body,
    redirect: 'error',
    signal: AbortSignal.timeout(10000),
  });

  return response.ok;
}

3. Implement Retry with Exponential Backoff

The specification defines a normative webhook retry schedule (several attempts with increasing delays). Align your production dispatcher with SPECIFICATION.md — the loop below is illustrative only.

Retry Logic

const MAX_RETRIES = 5;
const BASE_DELAY = 1000; // 1 second

async function deliverWithRetry(subscription, event) {
  for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
    try {
      const ok = await deliverWebhook(subscription, event);
      if (ok) return true;
    } catch (err) {
      // Network error
    }
    const delay = BASE_DELAY * Math.pow(2, attempt);
    await new Promise(r => setTimeout(r, delay));
  }
  // After 5 failures, pause subscription
  await pauseSubscription(subscription.id);
  return false;
}

4. Implement SSE Streaming

SSE Endpoint

// Hono framework
app.get('/eep/stream', async (c) => {
  const source = c.req.query('source');
  const stream = new ReadableStream({
    start(controller) {
      const encoder = new TextEncoder();

      eventBus.subscribe(source, (event) => {
        const chunk = `id: ${event.id}\nevent: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`;
        controller.enqueue(encoder.encode(chunk));
      });

      // Heartbeat every 15s
      setInterval(() => {
        const hb = `: heartbeat ${new Date().toISOString()}\n\n`;
        controller.enqueue(encoder.encode(hb));
      }, 15000);
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
});

5. WebSub Intent Verification

Before activating any webhook subscription, verify the subscriber controls the URL by sending a GET challenge.

Challenge Flow

import crypto from 'crypto';

async function verifyIntent(deliveryUrl: string, sourceDid: string) {
  const challenge = crypto.randomBytes(16).toString('hex');
  const url = new URL(deliveryUrl);
  url.searchParams.set('hub.mode', 'subscribe');
  url.searchParams.set('hub.topic', sourceDid);
  url.searchParams.set('hub.challenge', challenge);

  const res = await fetch(url.toString());
  const body = await res.text();

  return res.status === 200 && body.trim() === challenge;
}