Workflow Engine — Multi-Step Orchestration for Bun

validate ──→ reserve stock ──→ charge payment ──→ send confirmation
                  ↑                    ↑
            compensate:           compensate:
            release stock         refund payment

import { Workflow, Engine } from 'bunqueue/workflow';
// Define a workflow
const orderFlow = new Workflow('order-pipeline')
  .step('validate', async (ctx) => {
    const { orderId, amount } = ctx.input as { orderId: string; amount: number };
    if (amount <= 0) throw new Error('Invalid amount');
    return { orderId, validated: true };
  })
  .step('charge', async (ctx) => {
    const { orderId } = ctx.steps['validate'] as { orderId: string };
    const { amount } = ctx.input as { amount: number };
    const txId = await payments.charge(orderId, amount);
    return { transactionId: txId };
  }, {
    compensate: async () => {
      // Runs automatically if a later step fails
      await payments.refund();
    },
  })
  .step('confirm', async (ctx) => {
    const { transactionId } = ctx.steps['charge'] as { transactionId: string };
    await mailer.send('order-confirm', { txId: transactionId });
    return { emailSent: true, transactionId };
  });
// Create engine and run
const engine = new Engine({ embedded: true });
engine.register(orderFlow);
const run = await engine.start('order-pipeline', {
  orderId: 'ORD-1',
  amount: 99.99,
});
// Check status
const exec = engine.getExecution(run.id);
console.log(exec.state);  // 'running' | 'completed' | 'failed' | 'waiting' | 'compensating'

const flow = new Workflow('data-pipeline')
  .step('extract', async (ctx) => {
    const { source } = ctx.input as { source: string };
    const rawData = await fetchFromSource(source);
    return { records: rawData.length, data: rawData };
  })
  .step('transform', async (ctx) => {
    // Access previous step results via ctx.steps
    const { data } = ctx.steps['extract'] as { data: RawRecord[] };
    const cleaned = data.filter(r => r.valid).map(normalize);
    return { cleaned, dropped: data.length - cleaned.length };
  })
  .step('load', async (ctx) => {
    const { cleaned } = ctx.steps['transform'] as { cleaned: CleanRecord[] };
    await db.insertBatch('analytics', cleaned);
    // Access original input too
    return { loaded: cleaned.length, source: (ctx.input as { source: string }).source };
  });

const flow = new Workflow('money-transfer')
  .step('debit-source', async (ctx) => {
    const { from, amount } = ctx.input as { from: string; to: string; amount: number };
    await accounts.debit(from, amount);
    return { debited: true, account: from, amount };
  }, {
    compensate: async (ctx) => {
      // Undo: credit back the source account
      const { from, amount } = ctx.input as { from: string; to: string; amount: number };
      await accounts.credit(from, amount);
      console.log('Rolled back: source account credited');
    },
  })
  .step('credit-target', async (ctx) => {
    const { to, amount } = ctx.input as { from: string; to: string; amount: number };
    await accounts.credit(to, amount);
    return { credited: true, account: to, amount };
  }, {
    compensate: async (ctx) => {
      // Undo: debit back the target account
      const { to, amount } = ctx.input as { from: string; to: string; amount: number };
      await accounts.debit(to, amount);
      console.log('Rolled back: target account debited');
    },
  })
  .step('send-receipt', async () => {
    throw new Error('Email service down');
    // → Compensation runs automatically in reverse:
    //   1. credit-target compensate (debit target)
    //   2. debit-source compensate (credit source)
  });

const flow = new Workflow('support-ticket')
  .step('classify', async (ctx) => {
    const { message, plan } = ctx.input as { message: string; plan: string };
    const sentiment = await analyzeSentiment(message);
    const priority = plan === 'enterprise' ? 'high' : sentiment < 0 ? 'medium' : 'low';
    return { priority };
  })
  .branch((ctx) => (ctx.steps['classify'] as { priority: string }).priority)
  .path('high', (w) =>
    w.step('assign-senior', async (ctx) => {
      const agent = await roster.getAvailable('senior');
      await slack.notify(agent, 'Urgent ticket assigned');
      return { assignedTo: agent.name, sla: '1h' };
    })
  )
  .path('medium', (w) =>
    w.step('assign-regular', async (ctx) => {
      const agent = await roster.getAvailable('regular');
      return { assignedTo: agent.name, sla: '4h' };
    })
  )
  .path('low', (w) =>
    w.step('auto-reply', async (ctx) => {
      await mailer.sendTemplate('auto-reply', ctx.input);
      return { assignedTo: 'bot', sla: '24h' };
    })
  )
  .step('log-ticket', async (ctx) => {
    // This step always runs, regardless of which branch was taken
    await auditLog.write('ticket-created', { executionId: ctx.executionId });
    return { logged: true };
  });

const flow = new Workflow('content-publishing')
  .step('draft', async (ctx) => {
    const { title, body } = ctx.input as { title: string; body: string };
    const draft = await cms.createDraft({ title, body });
    await slack.notify('#editorial', `New draft "${title}" ready for review`);
    return { draftId: draft.id, previewUrl: draft.previewUrl };
  })
  .waitFor('editorial-review')
  .step('publish-or-reject', async (ctx) => {
    const review = ctx.signals['editorial-review'] as {
      approved: boolean;
      editor: string;
      notes?: string;
    };
    const { draftId } = ctx.steps['draft'] as { draftId: string };
    if (!review.approved) {
      await cms.reject(draftId, review.notes);
      return { status: 'rejected', editor: review.editor };
    }
    const published = await cms.publish(draftId);
    return { status: 'published', url: published.url, editor: review.editor };
  });
// Start the workflow
const run = await engine.start('content-publishing', {
  title: 'Announcing Workflow Engine',
  body: '...',
});
// The execution pauses at 'editorial-review' with state: 'waiting'
// Your app can show a UI, send a Slack button, expose an API endpoint, etc.
// When the editor makes a decision (could be minutes, hours, or days later):
await engine.signal(run.id, 'editorial-review', {
  approved: true,
  editor: 'alice@company.com',
  notes: 'Great article, ship it!',
});
// → Execution resumes from 'publish-or-reject'

const flow = new Workflow('api-aggregation')
  .step('fetch-primary', async () => {
    const res = await fetch('https://api.primary.com/data');
    return await res.json();
  }, { timeout: 5000 })  // 5 second timeout
  .step('fetch-secondary', async () => {
    const res = await fetch('https://api.secondary.com/data');
    return await res.json();
  }, { timeout: 10000 })  // 10 second timeout
  .step('merge', async (ctx) => {
    const primary = ctx.steps['fetch-primary'];
    const secondary = ctx.steps['fetch-secondary'];
    return { ...primary, ...secondary };
  });

const flow = new Workflow('resilient-pipeline')
  .step('call-api', async () => {
    const res = await fetch('https://api.external.com/data');
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return await res.json();
  }, {
    retry: 5,       // Max 5 attempts (default: 3)
    timeout: 10000,  // 10s per attempt
  })
  .step('process', async (ctx) => {
    const data = ctx.steps['call-api'] as ApiResponse;
    return { processed: true };
  });

const flow = new Workflow('data-enrichment')
  .step('fetch-user', async (ctx) => {
    const { userId } = ctx.input as { userId: string };
    return await db.users.find(userId);
  })
  .parallel((w) => w
    .step('fetch-orders', async (ctx) => {
      const { userId } = ctx.input as { userId: string };
      return await db.orders.findByUser(userId);
    })
    .step('fetch-preferences', async (ctx) => {
      const { userId } = ctx.input as { userId: string };
      return await db.preferences.get(userId);
    })
    .step('fetch-activity', async (ctx) => {
      const { userId } = ctx.input as { userId: string };
      return await analytics.getRecent(userId);
    })
  )
  .step('merge', async (ctx) => {
    // All parallel step results are available
    const orders = ctx.steps['fetch-orders'];
    const prefs = ctx.steps['fetch-preferences'];
    const activity = ctx.steps['fetch-activity'];
    return { profile: { orders, prefs, activity } };
  });

const flow = new Workflow('time-limited-approval')
  .step('submit', async (ctx) => {
    const { amount } = ctx.input as { amount: number };
    await slack.notify('#approvals', `Expense $${amount} needs review`);
    return { submitted: true };
  })
  .waitFor('manager-approval', { timeout: 86400000 }) // 24 hours
  .step('process', async (ctx) => {
    const decision = ctx.signals['manager-approval'] as { approved: boolean };
    return { status: decision.approved ? 'paid' : 'rejected' };
  });

const paymentFlow = new Workflow('payment')
  .step('validate-card', async (ctx) => {
    const { cardToken, amount } = ctx.input as { cardToken: string; amount: number };
    return { valid: true, amount };
  })
  .step('charge', async (ctx) => {
    return { txId: `tx_${Date.now()}` };
  }, {
    compensate: async () => { await payments.refund(); },
  });
const orderFlow = new Workflow('order')
  .step('create-order', async (ctx) => {
    const { amount } = ctx.input as { amount: number; cardToken: string };
    return { orderId: `ORD-${Date.now()}`, total: amount };
  })
  .subWorkflow('payment', (ctx) => ({
    // Map parent context to child input
    cardToken: (ctx.input as { cardToken: string }).cardToken,
    amount: (ctx.steps['create-order'] as { total: number }).total,
  }))
  .step('confirm', async (ctx) => {
    // Child results available under 'sub:<workflow-name>'
    const paymentResult = ctx.steps['sub:payment'] as Record<string, unknown>;
    return { confirmed: true, payment: paymentResult };
  });
const engine = new Engine({ embedded: true });
engine.register(paymentFlow); // Register child first
engine.register(orderFlow);
await engine.start('order', { amount: 99, cardToken: 'tok_abc' });

const engine = new Engine({ embedded: true });
// Listen to specific event types
engine.on('workflow:started', (event) => {
  console.log(`Workflow ${event.workflowName} started: ${event.executionId}`);
});
engine.on('step:completed', (event) => {
  const { stepName } = event as StepEvent;
  console.log(`Step ${stepName} completed in ${event.executionId}`);
});
engine.on('workflow:failed', (event) => {
  alerting.send(`Workflow ${event.workflowName} failed: ${event.executionId}`);
});
// Listen to ALL events
engine.onAny((event) => {
  metrics.increment(`workflow.${event.type}`, {
    workflow: event.workflowName,
  });
});
// Or pass onEvent in constructor
const engine2 = new Engine({
  embedded: true,
  onEvent: (event) => logger.info(event),
});

const engine = new Engine({ embedded: true });
// Delete old completed/failed executions (older than 7 days)
const deleted = engine.cleanup(7 * 24 * 60 * 60 * 1000);
console.log(`Deleted ${deleted} executions`);
// Or selectively clean only completed executions
engine.cleanup(7 * 24 * 60 * 60 * 1000, ['completed']);
// Archive instead of delete (moves to archive table)
const archived = engine.archive(30 * 24 * 60 * 60 * 1000); // 30 days
console.log(`Archived ${archived} executions`);
// Check archive count
console.log(`Total archived: ${engine.getArchivedCount()}`);

// doUntil: retry sending until delivery confirmed
const flow = new Workflow('delivery')
  .doUntil(
    (ctx) => (ctx.steps['send'] as { delivered: boolean })?.delivered === true,
    (w) => w.step('send', async (ctx) => {
      const result = await deliveryService.attempt(ctx.input);
      return { delivered: result.success };
    }),
    { maxIterations: 10 } // safety limit (default: 100)
  );
// doWhile: process items while queue has items
const batchFlow = new Workflow('batch')
  .doWhile(
    (ctx) => {
      const remaining = (ctx.steps['process'] as { remaining: number })?.remaining ?? 10;
      return remaining > 0;
    },
    (w) => w.step('process', async (ctx) => {
      const batch = await fetchNextBatch();
      await processBatch(batch);
      return { remaining: await getQueueSize() };
    }),
  );

const flow = new Workflow<{ userIds: string[] }>('notify-all')
  .forEach(
    (ctx) => (ctx.input as { userIds: string[] }).userIds, // items extractor
    'notify',                                               // step name
    async (ctx) => {
      const userId = ctx.steps.__item as string;   // current item
      const index = ctx.steps.__index as number;    // current index (0-based)
      await sendNotification(userId);
      return { notified: userId };
    },
    { retry: 3, timeout: 10_000 }  // standard step options
  );

const flow = new Workflow('etl')
  .step('fetch', async () => {
    const records = await db.query('SELECT * FROM events');
    return { records };
  })
  .map('aggregate', (ctx) => {
    const { records } = ctx.steps['fetch'] as { records: Event[] };
    return {
      total: records.length,
      byType: Object.groupBy(records, r => r.type),
    };
  })
  .step('store', async (ctx) => {
    const agg = ctx.steps['aggregate'] as AggregatedData;
    await analytics.insert(agg);
    return { stored: true };
  });

import { z } from 'zod';
const OrderInput = z.object({
  orderId: z.string(),
  amount: z.number().positive(),
});
const ChargeResult = z.object({
  transactionId: z.string(),
  charged: z.number(),
});
const flow = new Workflow('validated-order')
  .step('validate', async (ctx) => {
    const { orderId } = ctx.input as { orderId: string; amount: number };
    return { orderId, validated: true };
  }, {
    inputSchema: OrderInput,   // validates ctx.input before handler runs
  })
  .step('charge', async (ctx) => {
    return { transactionId: 'tx_123', charged: 99.99 };
  }, {
    outputSchema: ChargeResult, // validates return value after handler runs
  });

const run = await engine.start('order-pipeline', { orderId: 'ORD-1' });
// Subscribe to all events for this execution
const unsubscribe = engine.subscribe(run.id, (event) => {
  console.log(`[${event.type}]`, event);
  if (event.type === 'workflow:completed') {
    console.log('Order pipeline finished!');
  }
});
// Later: stop listening
unsubscribe();

// Embedded mode — everything in-process, no server needed
const engine = new Engine({ embedded: true });
// Embedded with SQLite persistence
const engine = new Engine({
  embedded: true,
  dataPath: './data/workflows.db',
});
// TCP mode — connects to a running bunqueue server
const engine = new Engine({
  connection: { host: 'localhost', port: 6789 },
});
// All options
const engine = new Engine({
  embedded: true,              // Use embedded mode (default: false)
  connection: { port: 6789 },  // TCP server connection (mutually exclusive with embedded)
  dataPath: './data/wf.db',    // SQLite persistence path
  concurrency: 10,             // Max parallel step executions (default: 5)
  queueName: '__wf:steps',     // Internal queue name (default: '__wf:steps')
  onEvent: (event) => {},      // Global event listener (optional)
});

const exec = engine.getExecution(run.id);
exec.id;            // 'wf_abc123' — unique execution ID
exec.workflowName;  // 'order-pipeline'
exec.state;         // 'running' | 'completed' | 'failed' | 'waiting' | 'compensating'
exec.input;         // { orderId: 'ORD-1', amount: 99.99 }
exec.steps;         // Step-by-step status and results:
// {
//   'validate': { status: 'completed', result: { orderId: 'ORD-1', validated: true } },
//   'charge':   { status: 'completed', result: { transactionId: 'tx_abc' } },
//   'confirm':  { status: 'running' }
// }
exec.signals;       // { 'manager-approval': { approved: true } }
exec.createdAt;     // 1712700000000
exec.updatedAt;     // 1712700005000

const orderFlow = new Workflow<{ orderId: string; items: Item[]; amount: number }>('process-order')
  .step('validate-order', async (ctx) => {
    const { orderId, items, amount } = ctx.input as OrderInput;
    if (items.length === 0) throw new Error('Empty cart');
    if (amount <= 0) throw new Error('Invalid amount');
    // Check all items are in catalog
    for (const item of items) {
      const exists = await catalog.exists(item.sku);
      if (!exists) throw new Error(`Unknown SKU: ${item.sku}`);
    }
    return { orderId, itemCount: items.length, amount };
  })
  .step('reserve-inventory', async (ctx) => {
    const { items } = ctx.input as OrderInput;
    const reservationId = await inventory.reserveBatch(items);
    return { reservationId };
  }, {
    retry: 3, // Retry on transient inventory service errors
    compensate: async (ctx) => {
      const { reservationId } = ctx.steps['reserve-inventory'] as { reservationId: string };
      await inventory.releaseBatch(reservationId);
    },
  })
  .step('process-payment', async (ctx) => {
    const { amount, orderId } = ctx.steps['validate-order'] as ValidatedOrder;
    const charge = await stripe.charges.create({
      amount: Math.round(amount * 100),
      currency: 'usd',
      metadata: { orderId },
    });
    return { chargeId: charge.id, receiptUrl: charge.receipt_url };
  }, {
    retry: 5,     // Payment gateway can be flaky
    timeout: 15000, // 15s timeout per attempt
    compensate: async (ctx) => {
      const { chargeId } = ctx.steps['process-payment'] as { chargeId: string };
      await stripe.refunds.create({ charge: chargeId });
    },
  })
  .step('create-shipment', async (ctx) => {
    const { orderId, items } = ctx.input as OrderInput;
    const { reservationId } = ctx.steps['reserve-inventory'] as { reservationId: string };
    const shipment = await shipping.create({ orderId, items, reservationId });
    return { trackingNumber: shipment.tracking, carrier: shipment.carrier };
  })
  .parallel((w) => w
    .step('send-confirmation', async (ctx) => {
      const payment = ctx.steps['process-payment'] as { chargeId: string; receiptUrl: string };
      const shipment = ctx.steps['create-shipment'] as { trackingNumber: string; carrier: string };
      const { email } = ctx.input as { email: string };
      await mailer.send('order-confirmation', {
        to: email,
        receiptUrl: payment.receiptUrl,
        tracking: shipment.trackingNumber,
      });
      return { emailSent: true };
    })
    .step('notify-warehouse', async (ctx) => {
      const { reservationId } = ctx.steps['reserve-inventory'] as { reservationId: string };
      await warehouse.notifyShipment(reservationId);
      return { warehouseNotified: true };
    })
    .step('update-analytics', async (ctx) => {
      const { amount, orderId } = ctx.steps['validate-order'] as ValidatedOrder;
      await analytics.trackPurchase({ orderId, amount });
      return { tracked: true };
    })
  );

const deployFlow = new Workflow('deploy-pipeline')
  .step('build', async (ctx) => {
    const { repo, branch, commitSha } = ctx.input as DeployInput;
    const build = await ci.triggerBuild({ repo, branch, commitSha });
    await ci.waitForBuild(build.id); // Polls until complete
    return { buildId: build.id, artifact: build.artifactUrl, duration: build.durationMs };
  })
  .step('run-tests', async (ctx) => {
    const { buildId } = ctx.steps['build'] as { buildId: string };
    const results = await ci.runTestSuite(buildId, {
      suites: ['unit', 'integration', 'e2e'],
      parallel: true,
    });
    if (results.failed > 0) {
      throw new Error(`${results.failed}/${results.total} tests failed`);
    }
    return { passed: results.passed, coverage: results.coverage };
  })
  .step('deploy-staging', async (ctx) => {
    const { artifact } = ctx.steps['build'] as { artifact: string };
    await k8s.deploy('staging', artifact);
    const healthCheck = await k8s.waitForHealthy('staging', 60000);
    // Notify the team
    await slack.send('#deploys', {
      text: `Staging deploy ready for review`,
      url: `https://staging.example.com`,
    });
    return { env: 'staging', healthy: healthCheck.ok };
  }, {
    compensate: async () => {
      // Rollback staging to previous version
      await k8s.rollback('staging');
    },
  })
  .waitFor('production-approval', { timeout: 48 * 60 * 60 * 1000 }) // 48h timeout
  .step('deploy-production', async (ctx) => {
    const approval = ctx.signals['production-approval'] as {
      approver: string;
      strategy: 'rolling' | 'blue-green' | 'canary';
    };
    const { artifact } = ctx.steps['build'] as { artifact: string };
    // Deploy with the approved strategy
    await k8s.deploy('production', artifact, { strategy: approval.strategy });
    await k8s.waitForHealthy('production', 120000);
    await slack.send('#deploys', {
      text: `Production deploy complete (${approval.strategy})`,
      approvedBy: approval.approver,
    });
    return {
      env: 'production',
      approvedBy: approval.approver,
      strategy: approval.strategy,
    };
  }, {
    compensate: async () => {
      await k8s.rollback('production');
      await slack.send('#deploys', { text: 'Production rolled back!' });
    },
  });
// Usage
const run = await engine.start('deploy-pipeline', {
  repo: 'myapp',
  branch: 'release/v2.5',
  commitSha: 'abc123f',
});
// After QA on staging (hours/days later):
await engine.signal(run.id, 'production-approval', {
  approver: 'cto@company.com',
  strategy: 'canary',
});

const kycFlow = new Workflow('kyc-onboarding')
  .step('create-account', async (ctx) => {
    const { email, name, country } = ctx.input as OnboardingInput;
    const user = await db.users.create({ email, name, country, status: 'pending' });
    return { userId: user.id };
  }, {
    compensate: async (ctx) => {
      // Delete the account if verification fails
      const { userId } = ctx.steps['create-account'] as { userId: string };
      await db.users.delete(userId);
    },
  })
  .step('risk-assessment', async (ctx) => {
    const { country, email, ip } = ctx.input as OnboardingInput;
    const score = await riskEngine.assess({ country, email, ip });
    return {
      score,
      riskLevel: score > 80 ? 'low' : score > 50 ? 'medium' : 'high',
    };
  })
  .branch((ctx) => (ctx.steps['risk-assessment'] as { riskLevel: string }).riskLevel)
  .path('low', (w) =>
    w.step('auto-approve', async () => {
      return { approved: true, method: 'automatic', verifiedAt: Date.now() };
    })
  )
  .path('medium', (w) =>
    w.step('request-documents', async (ctx) => {
      const { userId } = ctx.steps['create-account'] as { userId: string };
      await mailer.send('document-request', { userId });
      return { documentsRequested: true };
    })
    .waitFor('documents-uploaded')
    .step('verify-documents', async (ctx) => {
      const docs = ctx.signals['documents-uploaded'] as { files: string[] };
      const verification = await docVerification.check(docs.files);
      if (!verification.passed) {
        throw new Error(`Document verification failed: ${verification.reason}`);
      }
      return { approved: true, method: 'document-review', verifiedAt: Date.now() };
    })
  )
  .path('high', (w) =>
    w.step('flag-compliance', async (ctx) => {
      const { userId } = ctx.steps['create-account'] as { userId: string };
      const { score } = ctx.steps['risk-assessment'] as { score: number };
      await complianceQueue.assign({ userId, riskScore: score });
      return { flagged: true };
    })
    .waitFor('compliance-decision')
    .step('apply-compliance-decision', async (ctx) => {
      const decision = ctx.signals['compliance-decision'] as {
        approved: boolean;
        reviewer: string;
        notes: string;
      };
      if (!decision.approved) {
        throw new Error(`Rejected by compliance: ${decision.notes}`);
      }
      return { approved: true, method: 'compliance-review', reviewer: decision.reviewer };
    })
  )
  .step('activate-account', async (ctx) => {
    const { userId } = ctx.steps['create-account'] as { userId: string };
    await db.users.update(userId, { status: 'active', activatedAt: Date.now() });
    await mailer.send('welcome', { userId });
    return { activated: true };
  });

const etlFlow = new Workflow('daily-etl')
  .step('extract', async (ctx) => {
    const { date, sources } = ctx.input as { date: string; sources: string[] };
    const records: Record[] = [];
    for (const source of sources) {
      const data = await dataLake.query(source, { date });
      records.push(...data);
    }
    return { totalRecords: records.length, sources: sources.length, data: records };
  })
  .step('transform', async (ctx) => {
    const { data } = ctx.steps['extract'] as { data: Record[] };
    const cleaned = data
      .filter(r => r.timestamp && r.value !== null)
      .map(r => ({ ...r, value: normalize(r.value), processedAt: Date.now() }));
    const dropped = data.length - cleaned.length;
    return { cleanedRecords: cleaned.length, droppedRecords: dropped, data: cleaned };
  })
  .step('load', async (ctx) => {
    const { data } = ctx.steps['transform'] as { data: CleanRecord[] };
    const extract = ctx.steps['extract'] as { totalRecords: number; sources: number };
    const transform = ctx.steps['transform'] as { cleanedRecords: number; droppedRecords: number };
    // Batch insert
    const batches = chunk(data, 1000);
    for (const batch of batches) {
      await warehouse.insertBatch('analytics_events', batch);
    }
    return {
      pipeline: 'daily-etl',
      date: (ctx.input as { date: string }).date,
      metrics: {
        sourcesProcessed: extract.sources,
        rawRecords: extract.totalRecords,
        cleanedRecords: transform.cleanedRecords,
        droppedRecords: transform.droppedRecords,
        loadedRecords: data.length,
      },
    };
  });

const engine = new Engine({ embedded: true, dataPath: './data/etl.db' });
// Observability: track step durations and failures
engine.on('step:started', (e) => {
  const { stepName } = e as StepEvent;
  metrics.startTimer(`etl.${stepName}.duration`);
});
engine.on('step:completed', (e) => {
  const { stepName } = e as StepEvent;
  metrics.stopTimer(`etl.${stepName}.duration`);
  metrics.increment('etl.steps.completed');
});
engine.on('step:retry', (e) => {
  const { stepName, attempt, error } = e as StepEvent;
  logger.warn(`Retrying ${stepName}, attempt ${attempt}: ${error}`);
});
engine.on('workflow:failed', (e) => {
  alerting.pagerduty(`ETL pipeline failed: ${e.executionId}`);
});
engine.register(etlFlow);
// Run daily via cron
await engine.start('daily-etl', { date: '2026-04-10', sources: ['clickstream', 'transactions'] });
// Cleanup: archive completed runs older than 30 days, delete archived after 90 days
engine.archive(30 * 24 * 60 * 60 * 1000, ['completed']);
engine.cleanup(90 * 24 * 60 * 60 * 1000);

import { z } from 'zod';
const InvoiceInput = z.object({
  invoiceIds: z.array(z.string()),
  batchId: z.string(),
});
const invoiceFlow = new Workflow<{ invoiceIds: string[]; batchId: string }>('process-invoices')
  // Validate input with Zod schema
  .step('init', async (ctx) => {
    return { count: (ctx.input as { invoiceIds: string[] }).invoiceIds.length };
  }, { inputSchema: InvoiceInput })
  // Process each invoice
  .forEach(
    (ctx) => (ctx.input as { invoiceIds: string[] }).invoiceIds,
    'process-invoice',
    async (ctx) => {
      const invoiceId = ctx.steps.__item as string;
      const result = await billingService.process(invoiceId);
      return { invoiceId, amount: result.amount, status: result.status };
    },
    { retry: 3, timeout: 15_000 }
  )
  // Aggregate all invoice results
  .map('summary', (ctx) => {
    const results: { amount: number; status: string }[] = [];
    let i = 0;
    while (ctx.steps[`process-invoice:${i}`]) {
      results.push(ctx.steps[`process-invoice:${i}`] as { amount: number; status: string });
      i++;
    }
    const total = results.reduce((sum, r) => sum + r.amount, 0);
    const failed = results.filter(r => r.status === 'failed').length;
    return { total, processed: results.length, failed };
  })
  // Send summary report
  .step('report', async (ctx) => {
    const summary = ctx.steps['summary'] as { total: number; processed: number; failed: number };
    await notificationService.send({
      channel: '#billing',
      text: `Batch complete: ${summary.processed} invoices, $${summary.total} total, ${summary.failed} failures`,
    });
    return { reported: true };
  });

const deployFlow = new Workflow<{ deployId: string }>('wait-deploy')
  .step('trigger', async (ctx) => {
    const id = (ctx.input as { deployId: string }).deployId;
    await cloudProvider.triggerDeploy(id);
    return { deployId: id };
  })
  .doUntil(
    (ctx) => (ctx.steps['check-status'] as { ready: boolean })?.ready === true,
    (w) => w.step('check-status', async (ctx) => {
      const id = (ctx.steps['trigger'] as { deployId: string }).deployId;
      const status = await cloudProvider.getDeployStatus(id);
      // Simulate wait between polls
      await new Promise((r) => setTimeout(r, 5000));
      return { ready: status === 'running', status };
    }, { retry: 1, timeout: 30_000 }),
    { maxIterations: 60 } // max 5 minutes of polling
  )
  .step('verify', async (ctx) => {
    const id = (ctx.steps['trigger'] as { deployId: string }).deployId;
    const health = await cloudProvider.healthCheck(id);
    return { healthy: health.ok };
  });