Order Temporal worker
examples/order-temporal-worker — the orchestration deployment: the order application owning two journeys, served by @btravstack/temporal-worker.
pnpm turbo run test --filter=@btravstack/example-order-temporal-workerNeeds Docker
The suite runs a real @temporalio/worker Worker against a real Temporal server — one temporalio/auto-setup container shared by the whole repository (internal/test-infra), with a namespace of this spec file's own on it, and the example's own PostgreSQL database on the same server it uses. Nothing is started per workspace and nothing is cleaned up between tests: the namespace isolates the file, a per-test task queue isolates the tests inside it, and a per-test tenant isolates their rows.
It replaced Temporal's time-skipping test server, a 64 MB local binary started per vitest worker. Nothing here ever advanced a clock, so the skippable clock bought nothing a private namespace does not — and the example stopped being the one that needs the network on a cold cache.
Two sagas, two verticals, one queue
order-temporal-contract declares two workflows on the one orders task queue: fulfillOrder, the orders saga this example started with, and chargeOrder, a second saga — a second vertical, since taking the money is not part of placing, reserving or shipping the order. This worker is a modulith of two slices, src/slices/fulfillment/ and src/slices/billing/, one per workflow — the same shape order-api's HTTP controllers use, but with a property order-amqp-worker's two subscriber slices deliberately do not have: each slice here orchestrates a genuinely different thing. FulfillmentSlice imports FulfillmentModule; BillingSlice imports BillingModule. PaymentService is as invisible inside the first as the two fulfillment services are inside the second — they meet only at the root, in the list of slices, never inside either slice's own graph.
Neither imports the orders vertical, and that is the tenancy showing through: PlaceOrder and OrderRepository are built per ATTEMPT, in ActivityUnitModule, over the tenant the attempt's own input names — so fulfillOrder declares them beside inject and reads them off context.unit.
TemporalWorkflowActivities(contract, key) mints one piece per workflow — no port class, no name, since the contract key IS the port's name — and the piece is typed by the ONE workflow it implements: an activity the workflow does not declare is a compile error in that slice's own file, not a defect declareActivitiesHandler reports at startup.
export const chargeOrder = TemporalWorkflowActivities(
orderContract,
"chargeOrder",
)({
inject: { payments: PaymentService },
sync: ({ payments }) => ({
authorizePayment: ({ errors, idempotencyKey, input }) =>
payments
.authorize(input.orderId, input.amount, idempotencyKey)
.map((authorizationId) => ({ authorizationId }))
.mapErrCases((matcher) =>
matcher.with(P.tag("PaymentDeclined"), (error) =>
errors.PaymentDeclined({ id: error.id }),
),
),
capturePayment: ({ idempotencyKey, input }) =>
payments.capture(input.authorizationId, idempotencyKey),
refundPayment: ({ idempotencyKey, input }) =>
payments.refund(input.authorizationId, idempotencyKey),
}),
});All three take an idempotencyKey, and billing is the slice where that earns its place: Temporal runs an activity at least once, so a retry after a timeout the gateway never saw would authorize the card twice. The contract derives each key from the activity's own input —
idempotencyKey: ({ tenantId, orderId }) => `authorize:${tenantId}:${orderId}`,— which makes it stable across a retry, a worker crash, and a fresh execution with the same input.
Stability is the whole property, and it decides what may go in. The key is built only from the fields that identify the operation: the tenant, because every port here is tenant-scoped and an id is only unique within one, and the order. amount is deliberately absent — it describes the charge rather than identifying it, so folding it in would make the key move whenever the description does, which is exactly the retry the key exists to collapse. The sibling mistake is keying on a descriptive field instead of an identifying one: authorize:${tenantId}:${amount} would make one customer's two orders of the same value collide on a single key, and the second would never be charged.
The compensation gets one too, and arguably needs it most: refundPayment's sibling rule is that a repeated compensation must be a no-op, and for a refund the gateway is the only thing that can make it one.
helpers.idempotencyKey is typed string for an activity whose contract declares a key and undefined for one that does not, so reaching for a key nobody declared is a compile error rather than a silent undefined on the wire.
fulfillOrder's own piece is the same activities record this example always had, moved into src/slices/fulfillment/activities.ts unchanged and typed by its own key. The root composes both pieces into the one record the starter needs, keyed by the contract's own workflow names:
export const orderActivities = TemporalActivities(orderContract)([
fulfillOrder,
chargeOrder,
]);
export const OrderTemporalWorker = TemporalModule("OrderTemporalWorker")({
contract: orderContract,
activities: orderActivities,
workflows: {
workflowsPath: workflowsPathFromURL(import.meta.url, "./workflows.js"),
},
imports: [
FulfillmentSlice,
BillingSlice,
OrderPersistenceModule,
observability(),
otel(),
],
// Forked once per activity attempt, after it is invoked and before the body
// runs — which is where the input's `tenantId` becomes the fork's `Tenant`.
unit: { activity: ActivityUnitModule },
// Everything the fork reads out of the application scope.
exports: [Tracer, Logger, OrderDatabase],
});A wiring rule worth stating because the reason isn't obvious: orderActivities's own deps are the two pieces' ports, and di's flatten discovers providers only from a module's imports and provides — never from a provider's own deps. The root must import both FulfillmentSlice and BillingSlice, even though nothing in it names fulfillOrder or chargeOrder directly. Forgetting one still fails to compile: TemporalActivities declares each piece's port as one of its own deps, so a missing import is an undeclared need at the TemporalModule(...) call, refused with the exact port named — pnpm typecheck catches it, not a runtime WiringDefect.
The fulfillment saga
Three forward steps, each an activity calling into the application layer, and two compensations the workflow runs in reverse order of the steps they undo:
place ──▶ reserveStock ──▶ arrangeShipping ──▶ done
▲ ▲ OutOfStock? ▲ ShippingUnavailable?
│ └── cancelPlacement └── releaseStock, then cancelPlacementcontext.saga() owns both halves of that: the LIFO unwind, and the rule for which failures earn one. A declared error is a permanent domain answer, so it compensates. Temporal's own machinery tags (an activity that exhausted its retries unmodeled, or was cancelled) are handed back as-is and re-raised by propagateFailure, and compensation deliberately does not run for them, since a step that died mid-flight left unknown state — so what remains is one triage at the end, re-minting each declared error against context.errors so the client branches on it by name.
export const fulfillOrder = declareWorkflow({
workflowName: "fulfillOrder",
contract: orderContract,
implementation: (context, args) => {
const order = { tenantId: args.tenantId, orderId: args.orderId };
// A saga answers its LAST step's value, and this workflow answers the
// PLACEMENT's — so the first step keeps it and the last hands it back.
let placed!: PlacedOrder;
return propagateFailure(
context
.saga()
.step(
() =>
context.activities
.place({ ...order, quantity: args.quantity })
.tap((placement) => {
placed = placement;
}),
() => context.activities.cancelPlacement(order),
)
.step(
() => context.activities.reserveStock({ ...order, quantity: args.quantity }),
() => context.activities.releaseStock(order),
)
.step(() => context.activities.arrangeShipping(order).map(() => placed))
.run()
.mapErrCases((matcher) =>
matcher
.with({ errorName: "InvalidQuantity" }, (error) =>
context.errors.InvalidQuantity({ id: error.data.id }),
)
.with({ errorName: "InvalidOrderId" }, (error) =>
context.errors.InvalidOrderId({ id: error.data.id }),
)
.with({ errorName: "OrderAlreadyPlaced" }, (error) =>
context.errors.OrderAlreadyPlaced({ id: error.data.id }),
)
.with({ errorName: "OutOfStock" }, (error) =>
context.errors.OutOfStock({ id: error.data.id }),
)
.with({ errorName: "ShippingUnavailable" }, (error) =>
context.errors.ShippingUnavailable({ id: error.data.id }),
)
.with(
P.tag(ACTIVITY_ERROR_TAG),
P.tag(ACTIVITY_CANCELLED_ERROR_TAG),
(error) => error,
),
),
);
},
});The compensations declare no errors: compensation is the saga un-deciding, and a step that could answer "no" would leave it stuck half-done. cancelPlacement absorbs OrderNotFound on purpose — undoing a placement that never landed is the no-op a repeated compensation performs, and an activity Temporal may re-run has to answer the same both times.
The billing saga
The smallest saga in the example that still has a compensation:
authorizePayment ──▶ capturePayment ──▶ done
│ activity failure?
└── refundPaymentexport const chargeOrder = declareWorkflow({
workflowName: "chargeOrder",
contract: orderContract,
implementation: (context, args) =>
propagateFailure(
context.activities
.authorizePayment({
tenantId: args.tenantId,
orderId: args.orderId,
amount: args.amount,
})
.mapErrCases((matcher) =>
matcher
.with({ errorName: "PaymentDeclined" }, (error) =>
context.errors.PaymentDeclined({ id: error.data.id }),
)
.with(
P.tag(ACTIVITY_ERROR_TAG),
P.tag(ACTIVITY_CANCELLED_ERROR_TAG),
(error) => error,
),
)
.flatTap((authorized) =>
context.activities
.capturePayment({
tenantId: args.tenantId,
authorizationId: authorized.authorizationId,
})
.flatMapErrCases((matcher) =>
matcher.with(
P.tag(ACTIVITY_ERROR_TAG),
P.tag(ACTIVITY_CANCELLED_ERROR_TAG),
(error) =>
context.activities
.refundPayment({
tenantId: args.tenantId,
authorizationId: authorized.authorizationId,
})
.flatMap(() => ErrAsync(error)),
),
),
),
),
});authorizePayment's PaymentDeclined is declared nonRetryable in the contract — a refused card is a permanent answer, and asking Temporal to try four more times is the bug that discipline prevents. refundPayment declares no errors at all, for the same reason releaseStock does: un-deciding must not be able to answer no. The compensation only runs on an activity failure — a machinery tag, not a declared one — because capturePayment has no declared error of its own to compensate for; anything it fails with is infrastructure, which is exactly when the money needs to go back.
That is also why this one is hand-written where fulfillOrder uses context.saga(). The saga compensates on a declared error and refuses to on a machinery tag, which is the right default and the wrong one here: this workflow wants the compensation precisely in the case the default excludes. A policy that can be opted out of per workflow would erase the default's value, so the boundary is that the two spellings coexist — reach for the saga when a declared error is what earns the walk-back.
One subtlety worth stealing
An AsyncResult is eager: building a step starts its activity. So a sequence must never construct two steps as siblings — hoist them into consts and the "sequence" runs as a race, silently, with the types still checking out.
Both spellings in workflows.ts avoid it, and neither ever names a step in a const. context.saga() takes thunks, so nothing is built until the saga reaches it — that is fulfillOrder. chargeOrder, which compensates on a machinery tag the saga's policy refuses, sequences with flatTap instead: it runs a failable step, discards its value and passes the original one through, so each step's triage sits at one level of indentation instead of accumulating, and the next step is a callback that cannot start before the previous one settles.
Where a later step needs an earlier step's value rather than just its success, DoAsync().bind(...) is the same idea with an accumulating scope.
The external services
FulfillmentModule provides StockService and ShippingService; BillingModule provides PaymentService — in a real system other teams' APIs, here in-memory stand-ins that always say yes and leave a log line, because what this deployment demonstrates is the orchestration. The fulfillment specs swap in providers that say no; that is where both of fulfillOrder's compensation paths run, against the real application and the real persistence: after a refusal, the spec reads the database through the same repository the saga used and finds the placement gone. ShippingService.arrange is the deployment's one kernel touchpoint:
arrange: (orderId) =>
currentUnit()?.signal.aborted === true
? fromSafePromise(
Promise.reject(
new Error(
`the drain deadline passed before shipping for ${orderId} was arranged`,
),
),
)
: (logger.info("arranged shipping", { orderId }), OkAsync()),An adapter is where reading the ambient record is legitimate, and here it is the only route to the unit's AbortSignal at all: activityUnits calls next() unchanged, so an activity has no parameter to receive one through. Failing as a defect is the point — the platform retries that attempt on another worker, where the contract's ShippingUnavailable is a permanent no and would be the wrong answer to "we ran out of time". Temporal's own Context.current().cancellationSignal is a different clock, firing on shutdownGraceTime; the two are honoured together. See Read the ambient unit from an adapter.
The specs: a namespace as a fixture
test-fixtures.ts builds the it every spec imports. Its server fixture is file-scoped — createNamespace(address, "order-worker") registers a namespace on the shared server and waits for every Temporal service's registry to catch up before handing it over — and its tenant fixture is per test. serve boots, through @btravstack/testing's boot, the same TemporalModule sugar main.ts does, with a per-test task queue and a workflow bundle memoised per spec file — and composes BillingModule beside whichever fulfillment module the test hands it, since billing is never swapped:
const worker = TemporalModule("StubTemporalWorker")({
contract,
activities: orderActivities,
workflows: { workflowBundle },
imports: [module, BillingModule],
provides: [fulfillOrder, chargeOrder],
});provides: [fulfillOrder, chargeOrder] is there for the same wiring reason the root's own imports list both slices: the composed orderActivities's own needs are the two pieces' ports, and nothing else in this graph discharges them.
The stub deployments (fulfilling, outOfStock, noShipping) are each a tapped(rootWith(fulfillment, sink), [OrderRepository]), so a spec reads the database through the very repository the saga used. Five specs: the fulfillment saga fulfills in order; a stock refusal walks the placement back; a shipping refusal releases the reservation and then cancels, in that order; the duplicate the API answers CONFLICT for arrives at the client as OrderAlreadyPlaced, rehydrated by name with its payload intact; and the billing saga answers on the same task queue as the fulfillment one — proving every piece was mounted under its own key:
const { client } = await serve(fulfilling.module);
const charged = client.executeWorkflow("chargeOrder", {
workflowId: "wf-charge-1",
args: { orderId: "0199a1e0-0000-7000-8000-00000000a001", amount: 42 },
});
await expect(charged).toBeOkWith({
authorizationId: "auth-0199a1e0-0000-7000-8000-00000000a001",
});The drain, honouring the kernel's deadline
This is the first transport where Serving.drain is a genuine wait, and the half that lives in the package rather than the example: worker.shutdown() stops polling at once, run() resolves only once the in-flight activity has finished — on Temporal's shutdownForceTime, not the kernel's drainTimeoutMs. The starter races run() against the deadline signal, so an activity that never finishes cannot hold Serving.stop past the kernel's deadline; when the deadline wins the kernel gets its thread back, reports the unit abandoned, and the worker keeps winding down on Temporal's clock until the process exits, since @temporalio/worker exposes no public forced shutdown to escalate to. See Draining, in three beats.
The gate
needs-gate.test-d.ts pins NO RUNTIME — … (the graph without the starter fails to match the sentence intersected onto start's module parameter) and the unmet-need refusal spelled with the temporal() primitive, since the sugar cannot leave the activities out at all:
// @ts-expect-error — UNMET NEED: the module's needs channel carries the activities port, which nothing provides.
const _missingActivities = start(ActivitylessTemporal, options);That second one is the Needs channel, not di's UNSATISFIED DEPENDENCIES dependency gate: start's module parameter accepts only Scope | Env outstanding, so the activities port fails to assign and the diagnostic ends on Type '"TemporalActivities"' is not assignable to type '"@di/Scope"' — the port named, after several lines of the contract expanding.
Dropping one slice's import while still providing the composed activities fails the same way: TemporalActivities carries each piece's port as one of its own deps, so the missing import is an undeclared need at the TemporalModule(...) call — the wiring rule above — refused by di's NeedsGate and, again, by start's module parameter. No spec needs to pin it; pnpm typecheck already does.
Where to go next
- The same application, broadcasting: Order AMQP worker.
- The package:
@btravstack/temporal-worker; the task: Run a Temporal worker.