Subscribe an Agent to Your Event Stream¶

import crypto from 'node:crypto';
import express from 'express';

const app = express();

// --- Signature verification (Standard Webhooks) ---

function deriveKey(secret: string): Buffer {
  // Centrali secrets are `whsec_`-prefixed and base64url-encoded.
  return Buffer.from(secret.replace(/^whsec_/, ''), 'base64url');
}

function timingSafeEqual(a: string, b: string): boolean {
  const left = Buffer.from(a);
  const right = Buffer.from(b);
  return left.length === right.length && crypto.timingSafeEqual(left, right);
}

function verify(headers: Record<string, unknown>, rawBody: Buffer, secret: string): boolean {
  const id = (headers['webhook-id'] ?? headers['centrali-id']) as string | undefined;
  const ts = (headers['webhook-timestamp'] ?? headers['centrali-timestamp']) as string | undefined;
  const sig = (headers['webhook-signature'] ?? headers['centrali-signature']) as string | undefined;

  if (!id || !ts || !sig) return false;
  if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false; // 5-minute tolerance

  const expected = `v1,${crypto
    .createHmac('sha256', deriveKey(secret))
    .update(`${id}.${ts}.${rawBody.toString('utf8')}`)
    .digest('base64')}`;

  // During secret rotation the header carries multiple space-separated signatures.
  return String(sig)
    .split(/\s+/)
    .filter(Boolean)
    .some((candidate) => timingSafeEqual(candidate, expected));
}

// --- The agent decision step ---

interface CentraliEvent {
  event: string;
  workspaceSlug: string;
  recordSlug?: string;
  recordId?: string;
  data?: Record<string, unknown>;
  timestamp: string;
  correlationId?: string;
}

async function decide(event: CentraliEvent): Promise<void> {
  // Per-event triage. Cheap, deterministic checks first — only escalate to the
  // model (and its cost/latency) for events that actually warrant a decision.
  if (event.event !== 'record_created' || event.recordSlug !== 'orders') return;

  const order = event.data?.data as { total?: number } | undefined;
  if (!order || (order.total ?? 0) < 1000) return; // small orders: no-op

  // High-value order → let the agent decide what to do.
  //
  //   import Anthropic from '@anthropic-ai/sdk';
  //   const client = new Anthropic();
  //   const decision = await client.messages.create({
  //     model: 'claude-sonnet-4-6',
  //     max_tokens: 512,
  //     messages: [{ role: 'user', content: `Triage this order: ${JSON.stringify(order)}` }],
  //   });
  //   // ...then act on `decision`: write back to Centrali, notify a human, call a tool.

  console.log(`[agent] reviewing high-value order ${event.recordId}`);
}

// --- Endpoint: verify, ack fast, decide async ---

const seen = new Set<string>(); // swap for Redis/DB in production

app.post('/webhooks/centrali', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verify(req.headers, req.body, process.env.WEBHOOK_SECRET!)) {
    return res.status(401).send('Invalid signature');
  }

  // Idempotency: Centrali delivers at least once. Dedupe on webhook-id.
  const id = String(req.headers['webhook-id'] ?? req.headers['centrali-id']);
  if (seen.has(id)) return res.status(200).send('Duplicate ignored');
  seen.add(id);

  const event = JSON.parse(req.body.toString('utf8')) as CentraliEvent;

  // Acknowledge immediately; never block delivery on a model call.
  res.status(200).send('OK');

  // Run the decision out of band. A real deployment enqueues here (BullMQ, SQS, …)
  // so a crash mid-decision is retried without re-fetching from Centrali.
  decide(event).catch((err) => console.error('[agent] decide failed', err));
});

app.listen(3000, () => console.log('Agent listener on :3000'));