examples/storefront Public
// Fulfilment: for each order.paid event, send the order to the warehouse and record the// shipment. Events arrive at least once and a batch can be redelivered, so everything here is// keyed by the order id: the order's own status, and the Idempotency-Key the warehouse sees.import type { OrderEvent } from "../../../src/events.ts";import type { Order, Settled } from "../../../src/orders.ts";// storefront-api's OrdersService, through the ORDERS service binding.export interface Orders { forFulfilment(orderId: string): Promise<Order | null>; recordShipment(orderId: string, shipmentId: string): Promise<Settled>;}export interface Env { ORDERS: Orders; WAREHOUSE_URL: string; WAREHOUSE_TOKEN: string;}export type Outcome = "shipped" | "skipped";export class WarehouseError extends Error { readonly status: number; constructor(status: number, body: string) { super(`warehouse answered ${status}: ${body}`); this.status = status; }}// Each message is acked or retried on its own, so one bad order does not hold up the batch.// After max_retries (workers/fulfilment/wrangler.jsonc) a message goes to order-events-dlq.export async function fulfilBatch(batch: MessageBatch<OrderEvent>, env: Env, fetchFn: typeof fetch = fetch): Promise<void> { for (const msg of batch.messages) { try { const outcome = await fulfil(msg.body, env, fetchFn); console.log(JSON.stringify({ msg: "order event handled", type: msg.body.type, order: msg.body.order_id, outcome })); msg.ack(); } catch (e) { const delaySeconds = backoff(msg.attempts); console.error(JSON.stringify({ msg: "fulfilment failed", order: msg.body.order_id, attempt: msg.attempts, retry_in: delaySeconds, error: String(e) })); msg.retry({ delaySeconds }); } }}export async function fulfil(event: OrderEvent, env: Env, fetchFn: typeof fetch): Promise<Outcome> { if (event.type !== "order.paid") return "skipped"; const order = await env.ORDERS.forFulfilment(event.order_id); // Already fulfilled (this is a repeat) or cancelled since: nothing to send. if (!order || order.status !== "paid") return "skipped"; const res = await fetchFn(`${env.WAREHOUSE_URL}/v1/shipments`, { method: "POST", headers: { authorization: `Bearer ${env.WAREHOUSE_TOKEN}`, "content-type": "application/json", "idempotency-key": `order-${order.id}`, }, body: JSON.stringify({ reference: order.id, email: order.email, ship_to: order.ship_to, lines: order.items.map((i) => ({ sku: i.product_id, name: i.name, quantity: i.quantity })), }), }); if (!res.ok) throw new WarehouseError(res.status, (await res.text()).slice(0, 300)); const body = (await res.json()) as { shipment_id?: unknown }; if (typeof body.shipment_id !== "string") throw new WarehouseError(res.status, "no shipment_id in the response"); await env.ORDERS.recordShipment(order.id, body.shipment_id); return "shipped";}// 30 s, 1 min, 2 min, ... capped at an hour: long enough to ride out a warehouse outage within// the six retries before the message is dead-lettered.export function backoff(attempts: number): number { return Math.min(30 * 2 ** Math.max(0, attempts - 1), 3600);}