Build an Event-Driven Backend with Durable Workflows

// services/orders/handler.ts — Durable order processing
import * as restate from "@restatedev/restate-sdk";

interface OrderRequest {
  orderId: string;
  userId: string;
  items: Array<{ productId: string; quantity: number; price: number }>;
  paymentMethodId: string;
}

const orderService = restate.service({
  name: "orders",
  handlers: {
    async processOrder(ctx: restate.Context, order: OrderRequest) {
      const total = order.items.reduce((sum, i) => sum + i.price * i.quantity, 0);

      // Step 1: Reserve inventory (durable — won't re-run on retry)
      const reservation = await ctx.run("reserve-inventory", async () => {
        const res = await fetch("http://inventory:3001/reserve", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ orderId: order.orderId, items: order.items }),
        });
        if (!res.ok) throw new Error("Inventory reservation failed");
        return res.json();
      });

      // Step 2: Charge payment (exactly-once — safe to retry)
      const charge = await ctx.run("charge-payment", async () => {
        const res = await fetch("http://payments:3002/charge", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            orderId: order.orderId,
            userId: order.userId,
            amount: total,
            paymentMethodId: order.paymentMethodId,
            idempotencyKey: order.orderId,  // Stripe won't double-charge
          }),
        });
        if (!res.ok) {
          // Payment failed — release inventory (compensating action)
          await fetch("http://inventory:3001/release", {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ reservationId: reservation.id }),
          });
          throw new Error("Payment failed");
        }
        return res.json();
      });

      // Step 3: Confirm order in database
      await ctx.run("confirm-order", async () => {
        await fetch("http://orders-db:3003/confirm", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            orderId: order.orderId,
            chargeId: charge.chargeId,
            reservationId: reservation.id,
            total,
          }),
        });
      });

      // Step 4: Trigger post-order workflows (async — don't block the order)
      await ctx.run("trigger-post-order", async () => {
        await fetch("http://hatchet:7077/api/v1/events", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            key: "order:completed",
            data: { orderId: order.orderId, userId: order.userId, total, items: order.items },
          }),
        });
      });

      return {
        orderId: order.orderId,
        chargeId: charge.chargeId,
        status: "completed",
      };
    },
  },
});

restate.endpoint().bind(orderService).listen(9080);

// workflows/post-order.ts — Async post-order processing
import Hatchet from "@hatchet-dev/typescript-sdk";

const hatchet = Hatchet.init();

const postOrderWorkflow = hatchet.workflow({
  name: "post-order",
  on: { event: "order:completed" },
});

// Send confirmation email (retries 3x)
postOrderWorkflow.step("send-confirmation", async (ctx) => {
  const { userId, orderId, total } = ctx.input();
  await emailService.send(userId, {
    template: "order-confirmation",
    data: { orderId, total: `$${(total / 100).toFixed(2)}` },
  });
  return { emailSent: true };
}, { retries: 3, timeout: "30s" });

// Add loyalty points
postOrderWorkflow.step("add-loyalty-points", async (ctx) => {
  const { userId, total } = ctx.input();
  const points = Math.floor(total / 100);  // 1 point per dollar
  await loyaltyService.addPoints(userId, points);
  return { pointsAdded: points };
}, { retries: 2, timeout: "15s" });

// Ingest into analytics (runs in parallel with email and loyalty)
postOrderWorkflow.step("track-analytics", async (ctx) => {
  const { orderId, userId, total, items } = ctx.input();

  await fetch("https://api.tinybird.co/v0/events?name=orders", {
    method: "POST",
    headers: { Authorization: `Bearer ${process.env.TINYBIRD_TOKEN}` },
    body: JSON.stringify({
      order_id: orderId,
      user_id: userId,
      total_cents: total,
      item_count: items.length,
      timestamp: new Date().toISOString(),
    }),
  });

  return { tracked: true };
}, { retries: 3, timeout: "10s" });

-- tinybird/datasources/orders.datasource
SCHEMA >
  `order_id` String,
  `user_id` String,
  `total_cents` Int64,
  `item_count` Int32,
  `timestamp` DateTime
ENGINE MergeTree
ENGINE_SORTING_KEY timestamp

-- tinybird/pipes/revenue_dashboard.pipe
NODE hourly_revenue
SQL >
  SELECT
    toStartOfHour(timestamp) AS hour,
    count() AS order_count,
    sum(total_cents) / 100 AS revenue_usd,
    avg(total_cents) / 100 AS avg_order_usd
  FROM orders
  WHERE timestamp >= now() - INTERVAL 24 HOUR
  GROUP BY hour
  ORDER BY hour DESC