// 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);}