examples/storefront Public
Send paid orders to the warehouse from a fulfilment queue consumer
storefront-fulfilment consumes order-events, creates a shipment with the order id as idempotency key and records it through OrdersService. Each message is acked or retried on its own with a doubling backoff, and lands in order-events-dlq after six retries.
Changed files
15 files changed338 additions7 deletionsagainst 5324453
- Added.rebasely/architecture/flows/order-fulfilment.yaml+19−0
- Changed.rebasely/architecture/system.yaml+11−1
- Changed.rebasely/kb/concepts/checkout.md+2−2
- Added.rebasely/kb/concepts/fulfilment.md+33−0
- Changed.rebasely/kb/concepts/order-events.md+2−2
- Added.rebasely/kb/decisions/0007-at-least-once-with-idempotency-keys.md+26−0
- Changed.rebasely/kb/index.md+2−0
- ChangedREADME.md+4−1
- Addedmigrations/0004_shipments.sql+3−0
- Changedsrc/index.ts+9−1
- Changedsrc/orders.ts+10−0
- Addedtest/fulfilment.test.ts+110−0
- Addedworkers/fulfilment/src/fulfil.ts+78−0
- Addedworkers/fulfilment/src/index.ts+8−0
- Addedworkers/fulfilment/wrangler.jsonc+21−0
Added.rebasely/architecture/flows/order-fulfilment.yaml+19−0
| @@ -0,0 +1,19 @@ | ||
| +title: Fulfilling a paid order | ||
| +description: >- | ||
| + From Stripe's payment_intent.succeeded to a shipment at the warehouse. The webhook marks the | ||
| + order paid through OrdersService, which announces it on order-events. fulfilment picks the event | ||
| + up, reads the order, creates the shipment with the order id as idempotency key, and records it. | ||
| +actors: [warehouse] | ||
| +steps: | ||
| + - { from: stripe, to: stripe-webhook, label: "payment_intent.succeeded, signed" } | ||
| + - { from: stripe-webhook, to: storefront-api, label: "OrdersService.markPaid(order, intent)" } | ||
| + - { from: storefront-api, to: db, label: mark the order paid } | ||
| + - { from: storefront-api, to: order-events, label: send order.paid } | ||
| + - { from: order-events, to: fulfilment, label: a batch of up to 10 events } | ||
| + - { from: fulfilment, to: storefront-api, label: OrdersService.forFulfilment(order) } | ||
| + - { from: storefront-api, to: db, label: read the order and its lines } | ||
| + - { from: fulfilment, to: warehouse, label: "POST /v1/shipments, Idempotency-Key order-<id>" } | ||
| + - { from: warehouse, to: fulfilment, label: the shipment id } | ||
| + - { from: fulfilment, to: storefront-api, label: "OrdersService.recordShipment(order, shipment)" } | ||
| + - { from: storefront-api, to: db, label: mark the order fulfilled } | ||
| + - { from: fulfilment, to: order-events, label: "ack, or retry with backoff" } | ||
Changed.rebasely/architecture/system.yaml+11−1
| @@ -21,6 +21,15 @@ | ||
| talks_to: | ||
| - { to: storefront-api, via: rpc } | ||
| - { to: stripe, via: webhooks } | ||
| + fulfilment: | ||
| + kind: worker | ||
| + description: >- | ||
| + Worker storefront-fulfilment (workers/fulfilment). Consumes order-events, sends each paid | ||
| + order to the warehouse's shipments API and records the shipment through OrdersService. | ||
| + Failed messages retry with backoff and end in order-events-dlq after six attempts. | ||
| + owns: [workers/fulfilment/**] | ||
| + talks_to: | ||
| + - { to: storefront-api, via: rpc } | ||
| db: | ||
| kind: d1 | ||
| description: >- | ||
| @@ -38,7 +47,8 @@ | ||
| kind: queue | ||
| description: >- | ||
| Queue order-events: order.paid and order.cancelled, sent by OrdersService when an order | ||
| - changes state. At least once, so consumers dedupe by order id. | ||
| + changes state. Consumed by fulfilment (queues.consumers in its wrangler.jsonc). At least | ||
| + once, so consumers dedupe by order id. | ||
| binding: ORDER_EVENTS | ||
| stripe: | ||
| kind: external | ||
Changed.rebasely/kb/concepts/checkout.md+2−2
| @@ -8,7 +8,7 @@ | ||
| - src/orders.ts | ||
| - src/payments.ts | ||
| - migrations/0003_orders.sql | ||
| -reviewed: 2026-10-06 | ||
| +reviewed: 2026-10-07 | ||
| --- | ||
| # Checkout | ||
| @@ -37,6 +37,6 @@ | ||
| | `pending_payment` | Created at checkout; stock is held for it | | ||
| | `paid` | Stripe reported the payment succeeded ([[payment-webhook]]) | | ||
| | `cancelled` | Not paid; the stock went back in the same batch | | ||
| -| `fulfilled` | Handed to the warehouse | | ||
| +| `fulfilled` | Handed to the warehouse ([[fulfilment]]) | | ||
| State changes after checkout are in `src/orders.ts` (`markPaid`, `cancelOrder`) and only move forward from `pending_payment`, so repeating one is harmless. Each change is announced on the queue ([[order-events]]). `GET /orders/:id` returns the order. The id is a random UUID that only the shopper who checked out has. | ||
Added.rebasely/kb/concepts/fulfilment.md+33−0
| @@ -0,0 +1,33 @@ | ||
| +--- | ||
| +id: fulfilment | ||
| +type: concept | ||
| +owner: justin | ||
| +component: fulfilment | ||
| +describes: | ||
| + - workers/fulfilment/src/** | ||
| + - migrations/0004_shipments.sql | ||
| +reviewed: 2026-10-07 | ||
| +--- | ||
| +# Fulfilment | ||
| + | ||
| +The `storefront-fulfilment` Worker turns paid orders into shipments. It consumes the `order-events` queue ([[order-events]]) and talks to two things: storefront-api's OrdersService over the `ORDERS` service binding, and the warehouse's HTTP API at `WAREHOUSE_URL`. | ||
| + | ||
| +## Handling an event | ||
| + | ||
| +For each `order.paid` message: | ||
| + | ||
| +1. `OrdersService.forFulfilment(order)` returns the current order. If it is not `paid` any more (already fulfilled, which means this message is a repeat, or cancelled), the message is acked and nothing else happens. | ||
| +2. `POST {WAREHOUSE_URL}/v1/shipments` with the address and the lines, and `Idempotency-Key: order-<id>`. | ||
| +3. `OrdersService.recordShipment(order, shipment_id)` moves the order to `fulfilled` and stores the shipment id. | ||
| + | ||
| +Every other event type is acked and ignored. The steps are drawn in the order fulfilment flow on the Architecture tab. | ||
| + | ||
| +## Failures | ||
| + | ||
| +Each message in a batch is acked or retried on its own, so one bad order does not hold back the rest. Any error (the warehouse answering 4xx or 5xx, a timeout, OrdersService failing) retries the message after `backoff(attempts)`: 30 seconds, doubling, at most an hour. After six retries the queue moves the message to `order-events-dlq`. Messages there are orders a person has to look at; the logged error says why each one failed. | ||
| + | ||
| +Why repeating step 2 is safe is in [[0007-at-least-once-with-idempotency-keys]]. | ||
| + | ||
| +## Running it locally | ||
| + | ||
| +`npx wrangler dev -c workers/fulfilment/wrangler.jsonc` with `WAREHOUSE_URL` and `WAREHOUSE_TOKEN` in `.dev.vars` next to that config. With storefront-api running in another terminal, wrangler connects the service binding and the local queue. | ||
Changed.rebasely/kb/concepts/order-events.md+2−2
| @@ -6,7 +6,7 @@ | ||
| describes: | ||
| - src/events.ts | ||
| - src/orders.ts | ||
| -reviewed: 2026-10-06 | ||
| +reviewed: 2026-10-07 | ||
| --- | ||
| # Order events | ||
| @@ -29,4 +29,4 @@ | ||
| The price is duplicates: a consumer can see `order.paid` for the same order more than once, and must act on it once. The order id is the dedupe key. | ||
| -Messages are small and carry no order details. A consumer that needs the address or the lines asks OrdersService for them when it handles the message, so it always works from the current order. | ||
| +The consumer today is [[fulfilment]]. Messages are small and carry no order details. A consumer that needs the address or the lines asks OrdersService for them when it handles the message, so it always works from the current order. | ||
Added.rebasely/kb/decisions/0007-at-least-once-with-idempotency-keys.md+26−0
| @@ -0,0 +1,26 @@ | ||
| +--- | ||
| +id: 0007-at-least-once-with-idempotency-keys | ||
| +type: decision | ||
| +owner: justin | ||
| +status: accepted | ||
| +reviewed: 2026-10-07 | ||
| +--- | ||
| +# 0007 At-least-once delivery, made safe with the order id | ||
| + | ||
| +## Context | ||
| + | ||
| +Between a payment and a shipment there are three hops that can each be repeated: Stripe redelivers webhooks, the queue redelivers messages, and fulfilment retries the warehouse call. Exactly-once delivery is not on offer from any of them. | ||
| + | ||
| +## Decision | ||
| + | ||
| +Accept repeats everywhere and make every step idempotent by order id: | ||
| + | ||
| +- order state changes only move forward from a given status ([[payment-webhook]]), so a repeat changes nothing; | ||
| +- `order.paid` may be sent more than once ([[order-events]]); | ||
| +- fulfilment checks the order's status before acting and sends `Idempotency-Key: order-<id>` to the warehouse, which returns the existing shipment for a key it has seen ([[fulfilment]]). | ||
| + | ||
| +## Consequences | ||
| + | ||
| +- No dedupe table and no distributed locks. | ||
| +- Two copies of the same event handled at the same moment can both reach the warehouse. The idempotency key makes the second call return the first shipment, and `recordShipment` only applies once. | ||
| +- This depends on the warehouse honoring idempotency keys for at least as long as our retries last (about two hours). Theirs is 24 hours. | ||
Changed.rebasely/kb/index.md+2−0
| @@ -10,6 +10,7 @@ | ||
| - [[checkout]]: how a cart becomes an order, and how stock and payment are handled. | ||
| - [[payment-webhook]]: how Stripe's events mark orders paid or cancelled. | ||
| - [[order-events]]: the queue that announces paid and cancelled orders. | ||
| +- [[fulfilment]]: how paid orders become shipments at the warehouse. | ||
| ## Decisions | ||
| @@ -19,6 +20,7 @@ | ||
| - [[0004-take-stock-at-checkout]]: carts check stock, checkout takes it. | ||
| - [[0005-stripe-payment-intents]]: payments are Stripe PaymentIntents. | ||
| - [[0006-one-writer-for-the-database]]: other Workers change orders through OrdersService. | ||
| +- [[0007-at-least-once-with-idempotency-keys]]: repeats are expected and harmless. | ||
| ## Guides | ||
ChangedREADME.md+4−1
| @@ -2,10 +2,13 @@ | ||
| The API behind a small online shop, running on Cloudflare Workers. Products, carts and orders live in D1, product images in R2, and payments go through Stripe. | ||
| -There are two Workers: | ||
| +There are three Workers: | ||
| - `storefront-api` (`src/`) is the public API and the only Worker that writes to the database. | ||
| - `storefront-stripe-webhook` (`workers/stripe-webhook/`) receives Stripe's events and settles orders through storefront-api's `OrdersService` entrypoint. | ||
| +- `storefront-fulfilment` (`workers/fulfilment/`) consumes the `order-events` queue and sends paid orders to the warehouse. | ||
| + | ||
| +The order fulfilment flow on the Architecture tab follows a payment through all three. | ||
| It is a plain fetch handler with a small router, no framework, and the tests run the real SQL migrations on Node's built-in SQLite, so `npm test` needs nothing but Node 24. | ||
Addedmigrations/0004_shipments.sql+3−0
| @@ -0,0 +1,3 @@ | ||
| +-- The warehouse's shipment for a fulfilled order. | ||
| +ALTER TABLE orders ADD COLUMN shipment_id TEXT; | ||
| +ALTER TABLE orders ADD COLUMN fulfilled_at INTEGER; | ||
Changedsrc/index.ts+9−1
| @@ -1,7 +1,7 @@ | ||
| import { WorkerEntrypoint } from "cloudflare:workers"; | ||
| import { createApp } from "./app.ts"; | ||
| import type { Env } from "./env.ts"; | ||
| -import { cancelOrder, markPaid, type Settled } from "./orders.ts"; | ||
| +import { cancelOrder, getOrder, markPaid, recordShipment, type Order, type Settled } from "./orders.ts"; | ||
| // The calls the shop's other Workers make through a service binding (entrypoint OrdersService). | ||
| // storefront-api stays the only Worker that writes to the database; see | ||
| @@ -13,6 +13,14 @@ | ||
| cancel(orderId: string): Promise<Settled> { | ||
| return cancelOrder(this.env.DB, this.env.ORDER_EVENTS, orderId, Date.now()); | ||
| + } | ||
| + | ||
| + forFulfilment(orderId: string): Promise<Order | null> { | ||
| + return getOrder(this.env.DB, orderId); | ||
| + } | ||
| + | ||
| + recordShipment(orderId: string, shipmentId: string): Promise<Settled> { | ||
| + return recordShipment(this.env.DB, orderId, shipmentId, Date.now()); | ||
| } | ||
| } | ||
Changedsrc/orders.ts+10−0
| @@ -102,3 +102,13 @@ | ||
| if (!status) throw new Error(`no order ${orderId}`); | ||
| return { order_id: orderId, status, changed }; | ||
| } | ||
| + | ||
| +// Records the warehouse's shipment for a paid order. A second call for the same order changes | ||
| +// nothing and reports the order as it is. | ||
| +export async function recordShipment(db: D1Database, orderId: string, shipmentId: string, now: number): Promise<Settled> { | ||
| + const r = await db | ||
| + .prepare("UPDATE orders SET status = 'fulfilled', shipment_id = ?, fulfilled_at = ?, updated_at = ? WHERE id = ? AND status = 'paid'") | ||
| + .bind(shipmentId, now, now, orderId) | ||
| + .run(); | ||
| + return settled(db, orderId, r.meta.changes > 0); | ||
| +} | ||
Addedtest/fulfilment.test.ts+110−0
| @@ -0,0 +1,110 @@ | ||
| +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]); | ||
| +}); | ||
Addedworkers/fulfilment/src/fulfil.ts+78−0
| @@ -0,0 +1,78 @@ | ||
| +// 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); | ||
| +} | ||
Addedworkers/fulfilment/src/index.ts+8−0
| @@ -0,0 +1,8 @@ | ||
| +import type { OrderEvent } from "../../../src/events.ts"; | ||
| +import { fulfilBatch, type Env } from "./fulfil.ts"; | ||
| + | ||
| +export default { | ||
| + queue(batch, env) { | ||
| + return fulfilBatch(batch, env); | ||
| + }, | ||
| +} satisfies ExportedHandler<Env, OrderEvent>; | ||
Addedworkers/fulfilment/wrangler.jsonc+21−0
| @@ -0,0 +1,21 @@ | ||
| +// storefront-fulfilment: consumes order-events and hands paid orders to the warehouse. | ||
| +// Secrets: WAREHOUSE_URL and WAREHOUSE_TOKEN (npx wrangler secret put ...). | ||
| +{ | ||
| + "$schema": "../../node_modules/wrangler/config-schema.json", | ||
| + "name": "storefront-fulfilment", | ||
| + "main": "src/index.ts", | ||
| + "compatibility_date": "2026-09-01", | ||
| + "observability": { "enabled": true }, | ||
| + "queues": { | ||
| + "consumers": [ | ||
| + { | ||
| + "queue": "order-events", | ||
| + "max_batch_size": 10, | ||
| + "max_batch_timeout": 5, | ||
| + "max_retries": 6, | ||
| + "dead_letter_queue": "order-events-dlq" | ||
| + } | ||
| + ] | ||
| + }, | ||
| + "services": [{ "binding": "ORDERS", "service": "storefront-api", "entrypoint": "OrdersService" }] | ||
| +} | ||
Checks
No checks reported for this commit.
Runs
No runs for this commit.