import assert from "node:assert/strict";import { test } from "node:test";import type { OrderEvent } from "../src/events.ts";import { cancelOrder, getOrder, markPaid, recordShipment } from "../src/orders.ts";import { backoff, fulfilBatch, type Env } from "../workers/fulfilment/src/fulfil.ts";import { NOW, setup } from "./helpers/app.ts";const CHECKOUT = { email: "ada@park.test", ship_to: { name: "Ada Park", line1: "12 Alder St", city: "Portland", postal_code: "97205", country: "US" } };type Sent = { url: string; headers: Headers; body: Record<string, unknown> };async function paidOrder(warehouse: (sent: Sent) => Response = () => Response.json({ shipment_id: "shp_1001" }, { status: 201 })) {  const s = setup();  await s.addProduct("wool-socks", { name: "Wool socks", price_cents: 1800, stock: 3 });  const cartId = (await s.call("POST", "/carts")).body.cart.id;  await s.call("PUT", `/carts/${cartId}/items/wool-socks`, { quantity: 2 });  const order = (await s.call("POST", `/carts/${cartId}/checkout`, CHECKOUT)).body.order;  await markPaid(s.env.DB, s.events, order.id, `pi_${order.id}`, NOW);  const sent: Sent[] = [];  const fetchFn = (async (input: RequestInfo | URL, init?: RequestInit) => {    const one = { url: String(input), headers: new Headers(init?.headers), body: JSON.parse(String(init?.body)) };    sent.push(one);    return warehouse(one);  }) as typeof fetch;  const env: Env = {    WAREHOUSE_URL: "https://warehouse.test",    WAREHOUSE_TOKEN: "wh_test",    ORDERS: {      forFulfilment: (id) => getOrder(s.env.DB, id),      recordShipment: (id, shipment) => recordShipment(s.env.DB, id, shipment, NOW),    },  };  return { ...s, order, sent, run: (...events: OrderEvent[]) => runBatch(events, env, fetchFn) };}async function runBatch(events: OrderEvent[], env: Env, fetchFn: typeof fetch) {  const results: { acked: boolean; retryDelay?: number }[] = [];  const messages = events.map((body, i) => {    const r: { acked: boolean; retryDelay?: number } = { acked: false };    results.push(r);    return {      id: `msg-${i}`,      timestamp: new Date(NOW),      body,      attempts: 1,      ack: () => void (r.acked = true),      retry: (o?: { delaySeconds?: number }) => void (r.retryDelay = o?.delaySeconds),    };  });  await fulfilBatch({ queue: "order-events", messages } as unknown as MessageBatch<OrderEvent>, env, fetchFn);  return results;}const paid = (order_id: string): OrderEvent => ({ type: "order.paid", order_id, at: new Date(NOW).toISOString() });test("a paid order goes to the warehouse and is marked fulfilled", async () => {  const s = await paidOrder();  const [r] = await s.run(paid(s.order.id));  assert.equal(r!.acked, true);  assert.equal(s.sent.length, 1);  assert.equal(s.sent[0]!.url, "https://warehouse.test/v1/shipments");  assert.equal(s.sent[0]!.headers.get("idempotency-key"), `order-${s.order.id}`);  assert.deepEqual(s.sent[0]!.body.lines, [{ sku: "wool-socks", name: "Wool socks", quantity: 2 }]);  assert.equal((await s.call("GET", `/orders/${s.order.id}`)).body.order.status, "fulfilled");});test("a repeated order.paid is acked without a second shipment", async () => {  const s = await paidOrder();  await s.run(paid(s.order.id));  const [r] = await s.run(paid(s.order.id));  assert.equal(r!.acked, true);  assert.equal(s.sent.length, 1);});test("a warehouse error retries the message with a backoff", async () => {  const s = await paidOrder(() => new Response("upstream unavailable", { status: 503 }));  const [r] = await s.run(paid(s.order.id));  assert.equal(r!.acked, false);  assert.equal(r!.retryDelay, 30);  assert.equal((await s.call("GET", `/orders/${s.order.id}`)).body.order.status, "paid");});test("one failing order does not hold up the others in the batch", async () => {  const s = await paidOrder();  const results = await s.run(paid("no-such-order"), paid(s.order.id));  assert.deepEqual(results.map((r) => r.acked), [true, true], "an unknown order is skipped, not retried");  assert.equal(s.sent.length, 1);});test("cancelled orders and other events are skipped", async () => {  const s = setup();  await s.addProduct("wool-socks", { stock: 3 });  const cartId = (await s.call("POST", "/carts")).body.cart.id;  await s.call("PUT", `/carts/${cartId}/items/wool-socks`, { quantity: 1 });  const order = (await s.call("POST", `/carts/${cartId}/checkout`, CHECKOUT)).body.order;  await cancelOrder(s.env.DB, s.events, order.id, NOW);  const env: Env = {    WAREHOUSE_URL: "https://warehouse.test",    WAREHOUSE_TOKEN: "wh_test",    ORDERS: { forFulfilment: (id) => getOrder(s.env.DB, id), recordShipment: (id, sh) => recordShipment(s.env.DB, id, sh, NOW) },  };  const neverCalled = (async () => assert.fail("the warehouse was called")) as typeof fetch;  const results = await runBatch([paid(order.id), { type: "order.cancelled", order_id: order.id, at: new Date(NOW).toISOString() }], env, neverCalled);  assert.deepEqual(results.map((r) => r.acked), [true, true]);});test("backoff doubles from 30 seconds up to an hour", () => {  assert.deepEqual([1, 2, 3, 4, 8, 20].map(backoff), [30, 60, 120, 240, 3600, 3600]);});