@eventpilot/sdk

TypeScript/Node.js SDK for EventPilot — durable workflow execution for Kafka and Postgres workloads.

Install

npm install @eventpilot/sdk

Two worker models

| Model | Class | Best for | |-------|-------|----------| | gRPC Worker | Worker | Long-lived Node.js processes, self-hosted EventPilot | | HTTP Push Worker | EventPilotApp | Serverless, Next.js, Cloudflare Workers |

gRPC Worker (self-hosted)

Connects to EventPilot via a persistent gRPC bidirectional stream. Steps execute locally inside the worker process. Auto-reconnects on failure.

import { Worker, GrpcStepError, GrpcDefer } from '@eventpilot/sdk';

const worker = new Worker('localhost:50052');

worker.register({
  name: 'order_processing',   // must match the event name you publish
  version: '1.0.0',
  steps: [
    {
      name: 'validate_order',
      execute: async (state) => {
        const orderId = state.context['order_id'] as string;
        console.log(`Validating order ${orderId}`);
        return { data: { valid: true } };
      },
    },
    {
      name: 'charge_payment',
      execute: async (state) => {
        if (!state.context['valid']) {
          throw new GrpcStepError('order invalid', false);  // non-retryable
        }
        return {
          data: { charged: true },
          sideEffectKey: `charge:${state.context['order_id']}`,  // idempotency
        };
      },
      compensate: async (state) => {
        console.log(`Refunding order ${state.context['order_id']}`);
      },
    },
  ],
});

// Blocks until process exits or stop() is called.
// Reconnects automatically on stream errors.
await worker.start();

Step function

async (state: GrpcWorkflowState) => GrpcStepResult | void

GrpcWorkflowState fields:

| Field | Type | Description | |-------|------|-------------| | workflowId | string | Unique workflow ID | | workflowType | string | The event name that triggered it | | context | Record<string, unknown> | Accumulated values from all previous steps | | triggerPayload | unknown | Original event payload (JSON-parsed) | | retryCount | number | How many times this step has been retried | | stepResults | Record<number, { result? }> | Results of completed steps by index | | hasExecutedSideEffect(key) | function | Check if a side-effect key was already completed |

GrpcStepResult fields:

| Field | Type | Description | |-------|------|-------------| | data | Record<string, unknown> | Merged into context for subsequent steps | | result | unknown | Stored in workflow history for this step | | sideEffectKey | string | Idempotency key for external calls | | defer | { delayMs: number } | Re-execute this step after a delay |

Errors

// Permanent failure — skip retries, go to DLQ
throw new GrpcStepError('card declined', false);

// Transient failure — retry with backoff (default)
throw new Error('connection timeout');

// Defer — re-execute after a delay
throw new GrpcDefer(5000);  // 5 seconds
// or return:
return { defer: { delayMs: 5000 } };

Polling pattern

{
  name: 'wait_for_payment',
  execute: async (state) => {
    const status = await checkPaymentAPI(state.context['payment_id'] as string);
    if (status === 'completed') {
      return { data: { paid: true } };
    }
    return { defer: { delayMs: 5_000 } };  // check again in 5s
  },
}

Handler options

worker.register({
  name: 'my_handler',
  version: '2.0.0',
  type: 'event',            // 'task' (default) or 'event' (pub/sub)
  subscribesTo: ['order.placed'],  // only for type='event'
  maxRetries: 5,
  retryBackoffMs: 500,
  rateLimitScope: 'stripe',
  rateLimitRate: 500,       // tokens/sec across all workers
  storeIntermediateSteps: true,
  steps: [...],
});

Graceful shutdown

const controller = new AbortController();

process.on('SIGTERM', () => controller.abort());

await worker.start(controller.signal);

HTTP Push Worker (serverless / Next.js)

EventPilotApp exposes a single HTTP endpoint. EventPilot calls it per step. Works with Next.js App Router, Express, raw Node.js HTTP, Bun, Deno, and Cloudflare Workers.

import { EventPilotApp, DeferStep, EventPilotError } from '@eventpilot/sdk';

const app = new EventPilotApp({
  signingKey: process.env.EP_SIGNING_SECRET,  // optional HMAC verification
});

app.registerHandler(
  'send_email',
  [
    {
      name: 'build_body',
      run: async (ctx) => ({
        result: { html: buildHTML(ctx.event) },
        contextUpdates: { email_body_ready: true },
      }),
    },
    {
      name: 'send',
      run: async (ctx) => {
        await emailService.send(ctx.steps['build_body']);
        return { result: { sent: true } };
      },
    },
  ],
  { type: 'task', maxRetries: 3 }
);

// ── Next.js App Router ────────────────────────────────────
// app/eventpilot/route.ts
export const { GET, POST } = app.serve();

// ── Express ───────────────────────────────────────────────
expressApp.use('/eventpilot', app.expressHandler());

// ── Raw Node.js ───────────────────────────────────────────
import http from 'node:http';
http.createServer(app.nodeHandler()).listen(3000);

EventPilotClient

Publish events and inspect workflows via the HTTP API.

import { EventPilotClient } from '@eventpilot/sdk';

const ep = new EventPilotClient({
  baseUrl: 'http://localhost:8080',
  apiKey: 'ep_live_...',   // optional — omit for self-hosted with AUTH_ENABLED=false
});

// Publish an event (triggers all matching handlers)
const { workflowId } = await ep.publishEvent('order.placed', {
  order_id: 'ord_123',
  amount: 4999,
});

// Dispatch a point-to-point task, wait for result
const result = await ep.dispatch('generate_invoice', { order_id: 'ord_123' }, {
  waitForResult: true,
  timeoutSeconds: 10,
});

// Inspect a workflow
const state = await ep.getWorkflow(workflowId);

// List workflows
const running = await ep.listWorkflows('RUNNING', 20);

// Dead letter queue
const failed = await ep.listDLQ();
await ep.retryDLQ(failed[0].id);

// System stats
const stats = await ep.getStats();
console.log(`${stats.connectedWorkers} workers, ${stats.queueDepth} queued`);

Running tests

npm test              # unit tests (no server required)
npm run test:integration  # requires EventPilot running at EP_URL

Building

npm run build   # compiles TypeScript + copies proto to dist/