Skip to content

TypeScript workers

@virtufin/worker-devkit (npm, from the Gitea registry under the @virtufin scope) is the TypeScript worker devkit, the mirror of the C# Virtufin.Worker.DevKit and the Python virtufin-worker-devkit. The WorkManager runs TypeScript workers in its TypeScript engine (WorkManager 0.38.1+): one node subprocess per worker, the same lifecycle, lease, outbox, dead letters and input cursors as every engine (engines).

The contract

A worker is an ES module. It exports any of Start(ctx), Process(event) and Stop(), or default-exports a Worker (an object, or a class the host constructs once) with start, process and stop. Every entry point is optional and may be async.

import type { WorkerContext, WorkerEvent } from "@virtufin/worker-devkit";

let ctx: WorkerContext;

export async function Start(context: WorkerContext): Promise<void> {
    ctx = context;                       // once per activation, before the first event
}

export function Process(event: WorkerEvent) {
    return { type: "com.virtufin.example.greeted", source: "urn:example", data: { hello: event.data } };
}

export async function Stop(): Promise<void> {}   // ctx.stopping is aborted already
  • Start before Process. Every host (the Runtime, the local WorkManager) calls start once per activation before the first event and stop after the last; a stopped worker started again gets a new context.
  • Events are the dict form every subprocess worker sees: id, source, specversion, type and every other attribute as a top-level key, text data in data, binary data base64 in data_base64, timestamps as RFC 3339 text. The worker's config fills the attributes an event leaves out, its secrets are set over every input and stripped from every output.
  • Outputs: process returns an event, a list (published in order, each to its own publishtopic, else its type) or nothing. A non-string data is JSON, a Uint8Array binary. A throw dead-letters the input once to dlq.<topic>.
  • The context (WorkerContext): workerId, config, secrets, stopping (an AbortSignal, aborted the moment the host stops the worker or loses it), publish(events) (routed like returned outputs, from start, process, stop or the worker's own timers), reportStatus(state, detail), state (the worker's own keys: get, save with an optional etag, delete, keys) and getSecrets(). A refused call rejects with HostRefused (the worker is stopping or no longer owned), a stale etag with StateConflict, a state value holding a secret with a TypeError. Both error classes match by name, so instanceof holds for the engine's errors too.
  • Syntax: the engine runs TypeScript with Node's type stripping, so use erasable syntax only (tsc --erasableSyntaxOnly): no enum, namespace or constructor parameter properties. The devkit itself is compiled JavaScript.
  • Dependencies: a module's bare imports are installed at load at their latest version (@virtufin/* from the private registry). For third-party pins, ship an npm package (application/x-npm, npm pack): its main is the worker, its dependencies are installed, every @virtufin/* one at its latest.

Bases

WorkerBase is the TypeScript WorkerBase/BehaviorWorkerBase request layer: the worker type's names from the naming library (workerTypeNames: source urn:com.virtufin:worker:<type>, responses com.virtufin.<type>.response on wrk.<type>.response, errors on wrk.<type>.error, a replytopic changes the topic), response/error envelopes with correlationid, and redelivery tolerance (inputs arrive at least once since WorkManager 0.38): an input whose source and id the instance already took is dropped (SeenEvents, 100 000 / 10 minutes), and every output gets a derived id (derivedEventId, the UUIDv5 of the C# DerivedEventId, pinned by conformance/redelivery.v1.json), so a fresh activation that processes a replayed input again republishes the same ids. CommandWorker dispatches {command, ...} JSON inputs to handleCommand.

import { CommandWorker, type WorkerEvent } from "@virtufin/worker-devkit";

class Greeter extends CommandWorker {
    constructor() {
        super("greeter");
    }

    handleCommand(input: WorkerEvent, command: string, data: Record<string, unknown>) {
        return command === "greet" ? this.response(input, command, true, { message: `hello ${data.name}` }) : undefined;
    }
}

export default new Greeter();

Running locally

LocalWorkManager runs TypeScript workers in-process with the cluster's rules (the local WorkManager, async): deploy(worker, topics, {config, secrets}) starts it, publish(topic, events) drains everything the events cause before it resolves, observe(topic) is a hot stream (rxjs-interoperable), stop/start/remove drive the lifecycle, run(worker, inputs) yields one worker's outputs without routing. wire is the engine's codec and eventRules the per-event rules (prepareInput, finishOutputs, route, deadLetter).

import { LocalWorkManager, type WorkerEvent } from "@virtufin/worker-devkit";
import * as greeter from "./greeter.mts";

await using wm = new LocalWorkManager();
await wm.deploy(greeter, ["x.examples.greeter.in"]);
const responses: WorkerEvent[] = [];
wm.observe("wrk.greeter.response").subscribe((e) => responses.push(e));
await wm.publish("x.examples.greeter.in", { id: "1", source: "urn:test", data: JSON.stringify({ command: "greet", name: "ts" }) });

The conformance vectors run on it in the package's tests (test/conformance.test.ts): the 20 worker-event-rule vectors and the codec cases with expect_python (the TypeScript engine shares the Python engine's dict-form wire path) and the 7 lifecycle vectors; the same vectors run on the real TypeScript engine in workmanager's Runtime tests.

Deploying

Through the gateway, as any worker: CreateWorker with mime_type text/x-typescript and the module as code_source.content (or an npm tarball as application/x-npm), then StartWorker. The examples repo's typescript/examples/typescript-worker.ts deploys a greeter and per-pair market-data consumers this way.