@btravstack/temporal-worker

The worker half of Temporal for @btravstack/core — starting workflows is a client's job, and belongs to @temporal-contract/client. One module providing the runtime, its configuration and its connection; one unit per activity attempt; and a drain that releases the kernel at the kernel's deadline, not Temporal's shutdownForceTime.

📖 Documentation · Reference · API Reference

pnpm add @btravstack/temporal-worker @btravstack/core @btravstack/config @btravstack/di unthrown \
  @temporalio/worker @temporalio/activity @temporalio/common \
  @temporal-contract/worker@^8.0.0-beta.7 @temporal-contract/contract@^8.0.0-beta.7

All nine are peer dependencies — install them. Node >=22.22. This package is ESM-only.

A worked example

import { runMain, Logger } from "@btravstack/core";
import { Provider } from "@btravstack/di";
import {
  ActivityInput,
  TemporalActivities,
  TemporalModule,
} from "@btravstack/temporal-worker";
import { P } from "unthrown";

// The module forked around every activity attempt, SEEDED with the validated
// input on `ActivityInput(contract)`. The tenant the contract carries is
// claimed here once — not at each activity — and the application's vertical
// is composed over it, so `PlaceOrder` is bound before any activity runs.
const ActivityUnit = Module("ActivityUnit")({
  needs: [ActivityInput(contract)],
  imports: [OrderApplicationModule],
  provides: [
    Provider(Tenant)({
      inject: { input: ActivityInput(contract) },
      sync: ({ input }) => TenantId(input.tenantId),
    }),
  ],
  exports: [Tenant, PlaceOrder],
});

// The application's half: its activities, on the starter's own activities
// port, typed by the contract (a worker serves one activities record, so
// there is nothing to name). What the leaves read off `context.unit` is
// declared once, beside `inject`.
const orderActivities = TemporalActivities(contract)({
  unit: { place: PlaceOrder },
  sync: () => ({
    placeOrder: {
      place: ({ errors, context, input }) =>
        context.unit.place
          .execute(input.orderId, input.quantity)
          .mapErrCases((matcher) =>
            matcher
              .with(P.tag("DuplicateOrder"), (error) =>
                errors.OrderAlreadyPlaced({ id: error.id }),
              )
              .with(P.tag("InvalidOrderId"), (error) =>
                errors.InvalidOrderId({ id: error.id }),
              )
              .with(P.tag("InvalidQuantity"), (error) =>
                errors.InvalidQuantity({ id: error.id }),
              ),
          ),
    },
  }),
});

// The composition root: a di module, plus the contract, the activities
// provider and the workflow source — and nothing else to know.
const OrderWorker = TemporalModule("OrderWorker")({
  contract,
  activities: orderActivities,
  workflows: {
    workflowsPath: workflowsPathFromURL(import.meta.url, "./workflows.js"),
  },
  imports: [AppModule],
  unit: { activity: ActivityUnit },
});

await runMain(OrderWorker);

TemporalModule imports the starter — TemporalRuntime, TemporalConfig bound from TEMPORAL_ADDRESS (default 127.0.0.1:7233) and TEMPORAL_NAMESPACE (default default), TemporalConnection opened as a resource of the graph and closed on every exit path — provides the activities and exports the runtime port. A service that will not answer is a modeled TemporalUnreachable, exit 1; the starter calls declareActivitiesHandler itself, inside its own error qualifier, so a contract it cannot satisfy is a startup Err, not a defect. runtimeInfo() reads { taskQueue, namespace } back once the worker is polling.

A worker polling for several workflows can be several slices instead of one record: TemporalWorkflowActivities(contract, key)({ inject: { name: Dep }, unit, fork, sync }) mints a provider for ONE workflow's activities (or a contract-global activity), typed by the key alone, and TemporalActivities(contract)([...]) composes an array of them into the same activities provider TemporalModule takes — the array must cover every key the contract declares, and each piece's own port must still be discharged (provides), since the composed provider's deps are the pieces' ports, not what they close over. A workflow's activities belong to exactly one slice, and the compiler holds only half of that: coverage is checked, injectivity is not, so two slices claiming one key type-check together. It is di's duplicate-provider defect at build once both pieces are discharged as providers in one graph — wire in only one and the other's implementation is silently never registered.

A piece's unit names the ports its activities read off context.unit, resolved out of the per-attempt fork of the module bound on unit.activity — which is seeded with the validated input on ActivityInput(contract), so a unit module derives a tenant from the invocation rather than from an ambient record. The record arm above takes the same unit: — one record for every entry in it — so a worker that has not outgrown one function reaches context.unit without slicing first. The root is checked against what was declared either way: bind a module that does not export one of those ports and TemporalModule refuses it, naming the port.

A piece whose activities carry a different scope from the rest names its own module on fork: its attempts fork THAT module, seeded the same way, instead of unit.activity — a listing activity with no tenant in its input beside a fan-out whose activities each act for one. The piece's unit is then checked against its own module at the piece's call, the module's unmet needs become the root's (named by start), and an override reaches it under the module's name: overridden(root, providers, { unit: { TenantUnit: [...] } }).

Its own test suite needs a Docker daemon: the specs run against the shared temporalio/auto-setup container internal/test-infra starts once per machine and every workspace reuses. Nothing else here does.

Schedules

@btravstack/temporal-worker/schedule exports ensureSchedule — the typed schedule client's create, made idempotent, so a deploy that runs again updates the schedule instead of failing on ScheduleAlreadyExistsError. @temporal-contract/client is an optional peer reached only through that subpath. Why scheduled work costs a cluster here, and when it should not be this stack at all: Run something on a schedule.

Testing

@btravstack/temporal-worker/testing runs a worker's specs against one real Temporal server per test run: temporalServer is the Vitest globalSetup (the entry's default export), and temporalEnvironment(prefix) a file-scoped fixture answering { TEMPORAL_ADDRESS, TEMPORAL_NAMESPACE } for a namespace of the spec file's own. It needs Docker, and vitest and testcontainers as optional peers. See Test a Temporal worker.

Options

TemporalModule(name)({...}) takes the starter's options plus activities and the module lists (imports, provides, exports, needs). It is the only way to compose the starter, which is not exported:

| Option | What it is | | --------------- | ------------------------------------------------------------------------------------------------------------------------------------ | | contract | the temporal-contract contract; the task queue this worker polls is read off it | | activities | the activities provider — what TemporalActivities(contract)(...) returns for this contract | | workflows | where the sandbox's code comes from: { workflowsPath } or { workflowBundle }; required unless the contract declares no workflows | | address | pins the service instead of reading TEMPORAL_ADDRESS | | namespace | pins the namespace instead of reading TEMPORAL_NAMESPACE | | gracePeriod | pins TEMPORAL_GRACE_PERIOD_MS — Temporal's shutdownGraceTime, a Duration (default 10_000 ms) | | forceAfter | pins TEMPORAL_FORCE_AFTER_MS — Temporal's shutdownForceTime (default 15_000 ms); keep it under DRAIN_TIMEOUT_MS | | unit | { activity } — the module the worker forks around every activity it dispatches, seeded with the validated input | | dataConverter | the application's port for the DataConverter (payload codecs) the worker and its clients share |

The full table — required/optional, defaults, and the reasoning — lives on the reference page, which is this list's one detailed home.

The drain

worker.shutdown() stops polling at once, but run() resolves only when the last in-flight activity has finished — on Temporal's own shutdownForceTime. Serving.drain(signal) races that wait against the kernel's deadline and returns at whichever comes first: the kernel gets its thread back on time, the work is reported abandoned, and the worker keeps winding down on Temporal's clock until the process exits. @temporalio/worker exposes no public forced shutdown, so "stop waiting" is the escalation. Result → activity failure is declareActivitiesHandler's and is deliberately not duplicated here. The rest is on the documentation site.

TemporalModule automatically exports publicly available Logger and HealthChecks ports, including whole-module re-exports and providers declared at the root. It keeps private infrastructure hidden and business exports explicit.

License

MIT © Benoit TRAVERS