Worker DevKit

The Virtufin.Worker.DevKit NuGet package is the shared base library for all Virtufin workers. It provides a single, idiomatic .NET API for writing workers that runs against the two managed engines:

Engine	DevKit class to extend	Runs in
`DotNetDllEngine`	`WorkerBase<TInput, TOutput>` (or one of its subclasses)	In-process (JIT via hostfxr)
`NativeDllEngine`	`WorkerBase<TInput, TOutput>` (or one of its subclasses), wrapped in `AotNative<W>`	In-process (AOT-compiled, FlatBuffer wire format)

The package is dependency-light: it only references CloudNative.CloudEvents (plus, for the API helpers, Virtufin.Api.Client). The AOT native side also references Google.FlatBuffers.

The package is published from this repository at the same LIBRARY_VERSION as the WorkManager service. See the introduction for the package-publishing model.

Class hierarchy

The hierarchy is generic on the wire codec. The base WorkerBase<TInput, TOutput> takes a codec that owns payload decode/encode; the worker only deals with the typed in/out. The default JSON codec is JsonCloudEventCodec; plug a different codec in for non-JSON formats.

classDiagram
    class IWorker {
        <<interface>>
        +ProcessAsync(CloudEvent) Task~CloudEvent?~
    }

    class WorkerBase~TInput, TOutput~ {
        <<abstract, generic>>
        +Source Uri
        +ResponseEventType string
        +Codec ICloudEventCodec~TInput, TOutput~
        +ProcessAsync(CloudEvent) Task~CloudEvent?~
        +BuildResponse(CloudEvent, TOutput) CloudEvent
        +WithCorrelationId(CloudEvent) CloudEvent
        +WithTimestamp(CloudEvent) CloudEvent
    }
    IWorker <|.. WorkerBase~TInput, TOutput~

    class CommandWorker~TInput, TOutput, TCommand~ {
        <<abstract, generic>>
        +HandleCommandAsync(CloudEvent, TCommand, TInput) Task~TOutput?~
        +ExtractCommand(CloudEvent, TInput) TCommand
    }
    WorkerBase~TInput, TOutput~ <|-- CommandWorker~TInput, TOutput, TCommand~

    class CommandWorker~T~ {
        <<abstract, generic (JSON convenience)>>
        -Codec JsonCloudEventCodec
    }
    CommandWorker~TInput, TOutput, TCommand~ <|-- CommandWorker~T~

    class ApiWorker {
        <<sealed (JSON)>>
        +Api ApiClient?
        +HandleAsync(CloudEvent, JsonNode) Task~JsonObject?~
    }
    WorkerBase~JsonNode, JsonObject~ <|-- ApiWorker

    class ApiCommandWorker~T~ {
        <<sealed (JSON)>>
        +Api ApiClient?
        +HandleCommandAsync(CloudEvent, T, JsonNode) Task~JsonObject?~
        +EnsureClient(host, port)
        +EnsureClientFromCommand(node)
    }
    CommandWorker~T~ <|-- ApiCommandWorker~T~

    class ICloudEventCodec~TInput, TOutput~ {
        <<interface>>
        +ContentType string
        +DecodePayload(CloudEvent) TInput
        +EncodeResponse(CloudEvent, TOutput) (object?, string)
        +EncodeError(CloudEvent, string) TOutput
    }

    class JsonCloudEventCodec {
        +ContentType = "application/json"
    }
    ICloudEventCodec~TInput, TOutput~ <|.. JsonCloudEventCodec

    class AotNative~W~ {
        <<sealed, generic bridge>>
        +CurrentContext WorkerContext?
        +ProcessStatic(host, inBuf, inLen) int
        +FreeResultStatic(result) void
    }
    note for AotNative~W~ "Wraps any IWorker; the same worker<br/>class works under DotNetDllEngine<br/>(JIT) and NativeDllEngine (AOT)."

CommandWorker<T> and the *CommandWorker* variants require the enum identifier names to match the wire-level command strings (case-insensitive). See the CommandWorker<T> section below for the data flow.

Public API (JIT side: `DotNetDllEngine`)

`IWorker` and `WorkerBase<TInput, TOutput>`

The minimal contract a worker must satisfy:

public interface IWorker
{
    Task<CloudEvent?> ProcessAsync(CloudEvent input);
}

WorkerBase<TInput, TOutput> is the abstract, generic base class. The codec — supplied in the constructor — owns the wire-format decode/encode (ICloudEventCodec<TInput, TOutput>). The base class provides response-building helpers, automatic correlation-id propagation, and the canonical CloudEvent extension attribute names.

public abstract class WorkerBase<TInput, TOutput> : IWorker
{
    public const string CorrelationIdAttribute = "correlationid";
    public const string ReplyTopicAttribute     = "replytopic";
    public const string FieldError               = "error";
    public const string DataContentTypeJson      = "application/json";

    protected ICloudEventCodec<TInput, TOutput> Codec { get; }

    protected WorkerBase(
        Uri source,
        string responseEventType,
        ICloudEventCodec<TInput, TOutput> codec) { ... }

    public Task<CloudEvent?> ProcessAsync(CloudEvent input);   // not sealed
    protected abstract Task<TOutput?> ProcessAsync(CloudEvent input, TInput data);

    protected CloudEvent BuildResponse(CloudEvent input, TOutput payload);
    protected CloudEvent BuildResponse(CloudEvent input, object? data, string contentType);
    protected CloudEvent WithCorrelationId(CloudEvent input);
    protected CloudEvent WithTimestamp(CloudEvent input);
}

Response helpers

BuildResponse builds a CloudEvent with the configured Source and ResponseEventType. If the input has a correlationid extension, the reply inherits it automatically (via WithCorrelationId); the WithTimestamp extension is also applied automatically:

sequenceDiagram
    autonumber
    participant Sub as Pub/Sub
    participant WM as WorkManager
    participant W as WorkerBase.ProcessAsync
    participant Codec as JsonCloudEventCodec
    participant H as HandleAsync(CloudEvent, JsonNode)

    Sub->>WM: input CE (with correlationid)
    WM->>W: ProcessAsync(ce)
    W->>Codec: DecodePayload(ce) → node
    Codec-->>W: node
    W->>H: HandleAsync(ce, node)
    H-->>W: JsonObject payload
    W->>W: WithCorrelationId(ce).WithTimestamp()<br/>Type = responseEventType
    W-->>WM: CloudEvent envelope with payload
    WM->>Sub: publish response

If the handler throws, the base class catches and builds an error envelope via codec.EncodeError(...) — the typed error payload (JsonObject shaped like {"command":"error","success":false,"message":"..."} for JSON).

`replytopic` extension

If the input CloudEvent carries a replytopic extension attribute, the WorkManager routes the response to that topic instead of the worker's default reply topic. WorkerBase does not need to do anything — the engine reads this extension and overrides the publish topic.

Worker-instance config, delivered as extension attributes

CreateWorkerRequest.Config (a map<string, string>, immutable after creation) lets a caller give a worker per-instance defaults without process environment variables — which would leak between workers co-loaded in the same WorkManager process. Runtime.Worker.ProcessAsync stamps every Config entry onto each incoming CloudEvent as an extension attribute before dispatching to the engine, using the same generic mechanism as correlationid/replytopic — no codec or ABI changes are needed for a worker to read one.

Convention: read the worker's own JSON payload field first, and fall back to the same-named CloudEvent extension only if the payload doesn't specify a value — the extension is a default, the payload (if present) always wins:

protected override async Task<MyOutput?> ProcessAsync(CloudEvent input, JsonNode payload)
{
    var url = payload.GetString("url") ?? (input["url"] as string);
    ...
}

`CommandWorker<T>` — typed command dispatch

The CommandWorker<T> base class is for workers that handle a finite set of named commands carried in the CloudEvent data payload. The wire format is:

{ "command": "create", "api_host": "...", "data": { ... } }

CommandWorker<T> parses data.command, calls Enum.TryParse<T>(command, ignoreCase: true, out _), and forwards the parsed enum value plus the full JsonNode payload to your override:

sequenceDiagram
    autonumber
    participant Sub as Pub/Sub
    participant WM as WorkManager
    participant CW as CommandWorker.ProcessAsync
    participant Codec as JsonCloudEventCodec
    participant H as HandleCommandAsync

    Sub->>WM: data = {"command": "create", ...}
    WM->>CW: ProcessAsync(ce)
    CW->>Codec: DecodePayload(ce) → node
    Codec-->>CW: node
    CW->>CW: ExtractCommand(ce, node)<br/>node["command"] → "create"<br/>Enum.TryParse<T>("create") → MyCmd.create
    CW->>H: HandleCommandAsync(ce, MyCmd.create, node)
    alt Handler returns JsonObject
        H-->>CW: payload JsonObject
        CW->>Codec: EncodeResponse(ce, payload) → (data, contentType)
        CW->>CW: envelope = WithCorrelationId(ce).WithTimestamp()<br/>Type = responseEventType
        CW-->>WM: response CE
    else Handler returns null
        H-->>CW: null (no reply)
        CW-->>WM: null
    end

Usage:

public enum MyCommand { create, destroy, list }

public class MyWorker : CommandWorker<MyCommand>
{
    public MyWorker(Uri source = new("urn:com.example.myworker"),
                    string responseEventType = "com.example.myworker.response")
        : base(source, responseEventType) { }

    protected override Task<JsonObject?> HandleCommandAsync(
        CloudEvent input, MyCommand command, JsonNode node)
        => command switch
        {
            MyCommand.create  => HandleCreateAsync(input, node),
            MyCommand.destroy => HandleDestroyAsync(input),
            MyCommand.list    => HandleListAsync(input),
            _ => null,
        };
}

The enum identifier names are the wire strings. The same enum drives both the gRPC service definition (proto) and the worker handler, so a rename in one place surfaces as a wire-protocol break.

The handler returns a typed payload (JsonObject), not a CloudEvent. The base class wraps it in the envelope with Id, Type, Source, correlationid extension (auto-applied from input), and time extension (auto-applied). Use the Response(...) helper on CommandWorker<T> to build the standard { "command", "success", "id", "message" } payload:

return Response(input, MyCommand.create.ToString(), true,
                 id: createdId, message: "Created");

For richer payloads, build a JsonObject directly.

`ApiWorker` and `ApiCommandWorker<T>`

ApiWorker : WorkerBase<JsonNode, JsonObject> is the JSON-convenience base for workers that call backend services through the API Gateway without a typed command. Override HandleAsync(CloudEvent, JsonNode) → Task<JsonObject?>.

ApiCommandWorker<T> : CommandWorker<T> combines typed command dispatch with the API client helper. Both inject a pre-configured Virtufin.Api.Client.ApiClient for backend calls through the API Gateway.

public class MyApiWorker : ApiCommandWorker<MyCommand>
{
    public MyApiWorker() : base(
        source: new("urn:com.example.myworker"),
        responseEventType: "com.example.myworker.response",
        apiHost: "localhost",
        apiPort: 5002) { }

    protected override async Task<JsonObject?> HandleCommandAsync(
        CloudEvent input, MyCommand command, JsonNode node)
    {
        if (command == MyCommand.create)
        {
            EnsureClientFromCommand(node);   // or pass apiHost/apiPort in ctor
            var created = await Api!.SomeService.CreateAsync(...);
            return Response(input, command.ToString(), true,
                            id: created.Id, message: "Created");
        }
        return ErrorPayload("Unsupported command");
    }
}

Two API-client construction modes are available:

Eager: pass apiHost and apiPort to the base ctor; Api is ready in HandleCommandAsync.
Lazy: pass only source and responseEventType. Api is null until EnsureClient(host, port) (or EnsureClientFromCommand(node), which reads the api_host field from the decoded JsonNode) is called. Use this when the API host comes from the command payload.

AOT native side (`NativeDllEngine`)

The AOT side is a thin bridge over the same IWorker contract — there is no parallel class hierarchy. Any IWorker implementation (typically extending WorkerBase<TInput, TOutput> or CommandWorker<T>) runs unchanged under DotNetDllEngine (JIT, in-process ALC) and NativeDllEngine (AOT, via the AotNative<W> wrapper).

`AotNative<W>`

AotNative<W> where W : IWorker, new() is the only AOT-side base — a sealed generic bridge that provides the FlatBuffer / native C ABI plumbing needed by NativeDllEngine. The worker's class declares two [UnmanagedCallersOnly] exports (Process, FreeResult) that delegate to AotNative<W>.ProcessStatic and AotNative<W>.FreeResultStatic.

public class MyWorker : CommandWorker<MyCommand>
{
    public MyWorker() : base(new Uri("urn:com.example"), "com.example.response") { }

    protected override async Task<JsonObject?> HandleCommandAsync(
        CloudEvent input, MyCommand command, JsonNode node)
    {
        var context = AotNative<MyWorker>.CurrentContext;
        if (context is null) return ErrorPayload("no AOT context");
        var gateway = new NativeApiGateway(context);
        // ... dispatch on command ...
        return Response(input, command.ToString(), true, message: "ok");
    }

    [UnmanagedCallersOnly(EntryPoint = "Process")]
    public static int Process(IntPtr host, IntPtr inBuf, int inLen,
                              IntPtr outBuf, IntPtr outLen)
    {
        unsafe
        {
            return AotNative<MyWorker>.ProcessStatic(
                host, inBuf, inLen, (IntPtr*)outBuf, (int*)outLen);
        }
    }

    [UnmanagedCallersOnly(EntryPoint = "FreeResult")]
    public static void FreeResult(IntPtr result)
        => AotNative<MyWorker>.FreeResultStatic(result);
}

The same MyWorker class is also usable directly by DotNetDllEngine (the JIT engine calls worker.ProcessAsync(CloudEvent) without going through the AotNative bridge).

`AotNative<W>.CurrentContext`

A static AsyncLocal<WorkerContext?> set by ProcessStatic immediately before the wrapped worker's ProcessAsync is called and cleared in a finally block after it returns. Workers read this to obtain a NativeApiGateway for backend calls under NativeDllEngine. It is null outside a native invocation (i.e. under DotNetDllEngine).

Environment variables

NativeDllEngine.SetIdentity accepts a WorkerIdentity(Id, Topic). The engine sets two reserved env vars that identify the worker to the native side, via the shared IdentityEnvVarScope helper (also used by DotNetDllEngine):

Environment variable	Source	Notes
`VIRTUFIN_WORKER_ID`	`Id`	Always set
`VIRTUFIN_WORKER_TOPIC`	`Topic`	Always set

Values are applied before LoadCodeAsync completes and restored on dispose. There is no custom per-worker environment-variable mechanism — that was removed (it mutated process-wide state, so any two workers co-loaded in the same WorkManager process would silently stomp on each other's values). Per-instance configuration belongs in the triggering CloudEvent's own payload instead — read it from the decoded command JSON, the same way ApiCommandWorker.EnsureClientFromCommand reads api_host.

Per-instance state

Each unique host pointer (one per LoadCodeAsync call) gets its own wrapped W instance, looked up via a ConcurrentDictionary<IntPtr, AotNative<W>>. Multiple workers sharing the same loaded library still get independent state.

sequenceDiagram
    autonumber
    participant WM as WorkManager<br/>(NativeDllEngine)
    participant Lib as Native worker<br/>(lib*.so)
    participant Bridge as AotNative<MyWorker>
    participant Sub as MyWorker.HandleCommandAsync
    participant Ctx as WorkerContext

    WM->>Lib: Process(host, inBuf, inLen, &outBuf, &outLen)
    Note over WM,Lib: inBuf is a CloudEvent FlatBuffer
    Lib->>Bridge: ProcessStatic(host, inBuf, inLen, ...)
    Bridge->>Bridge: CloudEventCodec.DecodeCloudEvent(inBuf) → ce
    Bridge->>Bridge: _instances.GetOrAdd(host) → wrapper
    Bridge->>Bridge: CurrentContext = BuildWorkerContext(host)
    Bridge->>Sub: wrapper.ProcessAsync(ce)
    Sub->>Ctx: ctx.Log("...", level)
    Ctx->>Lib: VirtufinHost.log(host, level, msg)
    Lib-->>Ctx: log goes back to WM via callback
    Sub->>Ctx: gateway.InvokeAsync(...)
    Ctx->>Lib: VirtufinHost.gateway_call(host, ...)
    Lib-->>Ctx: gateway goes back to WM
    Sub-->>Bridge: payload JsonObject (or null)
    Bridge->>Bridge: codec.EncodeResponse(ce, payload) → envelope
    Bridge->>Bridge: WithCorrelationId(ce).WithTimestamp()
    Bridge->>Bridge: CloudEventCodec.EncodeWorkerResponse(envelope) → bytes
    Bridge->>Bridge: Marshal.AllocHGlobal(bytes)<br/>*outBuf = ptr, *outLen = bytes.Length
    Bridge-->>Lib: 0
    Lib-->>WM: outBuf, outLen
    WM->>WM: Marshal.Copy → span → DecodeWorkerResponse
    WM->>Lib: FreeResult(outBuf)
    WM->>WM: publish reply CE

`WorkerContext`

WorkerContext wraps the native VirtufinHost* that the engine hands to the worker. It exposes typed helpers for the three host callbacks:

Log(LogLevel level, string message) — write to the WorkManager log
InvokeGateway(string service, string method, byte[] requestJson) — invoke a backend service through the API Gateway via the host's gateway_call callback
The host's free_response callback is invoked automatically by InvokeGateway after the worker reads the response bytes

Workers read WorkerContext via AotNative<W>.CurrentContext and wrap it in a NativeApiGateway for typed gateway calls. Do not dereference WorkerContext.Host; it is opaque.

`NativeApiGateway`

Thin wrapper over WorkerContext.InvokeGateway. Construct one from the ambient CurrentContext:

var gateway = new NativeApiGateway(AotNative<MyWorker>.CurrentContext!);
var responseBytes = gateway.InvokeAsync("websocketmanager", "Connect", requestBytes);

`CloudEventCodec`

CloudEventCodec is the single source of truth for the CloudEvent ↔ FlatBuffer mapping shared by NativeDllEngine (engine side) and AotNative<W> (AOT-compiled worker side). The wire format is the FlatBuffer binary produced by FlatBufferBuilder.Finish(offset) without a size prefix — ByteBuffer.Wrap(bytes) consumes the bytes.

Method	Direction	Description
`EncodeCloudEvent(CloudEvent) → byte[]`	managed → wire	Encode a managed `CloudEvent` to the FlatBuffer wire format.
`DecodeCloudEvent(byte[]) → CloudEvent`	wire → managed	Decode a `Process` input buffer.
`EncodeWorkerResponse(CloudEvent?) → byte[]`	managed → wire	Encode the `WorkerResponse` returned from `Process`.
`DecodeWorkerResponse(byte[]) → (CloudEvent?, string?)`	wire → managed	Decode the `WorkerResponse`; `(null, "...")` on error.

If the worker's WorkerResponse.error_message is set, the engine surfaces a worker.error lifecycle event and (if no result_event is also set) throws InvalidOperationException. If neither field is set, the worker declines to publish a reply (the engine returns null).

JIT vs AOT — what changes?

The JIT and AOT paths share WorkerBase<TInput, TOutput>, CommandWorker<TInput, TOutput, TCommand>, CommandWorker<T>, ApiWorker, and ApiCommandWorker<T>. Only the plumbing differs:

Concern	`DotNetDllEngine` (JIT)	`NativeDllEngine` (AOT)
Worker class	Plain `IWorker` impl	Plain `IWorker` impl, wrapped in `AotNative<W>`
Bridge	(none — engine calls `ProcessAsync` directly)	`[UnmanagedCallersOnly]` `Process` + `FreeResult` exports on the worker, delegating to `AotNative<W>.ProcessStatic` / `FreeResultStatic`
Native exports	None	Required on the worker class; need `<AllowUnsafeBlocks>true</AllowUnsafeBlocks>` (the bridge uses `IntPtr`/`int` because .NET 10 forbids `out` on `[UnmanagedCallersOnly]`)
Backend gateway	`ApiClient` (gRPC) injected into `ApiWorker`/`ApiCommandWorker<T>`	`NativeApiGateway(context)` constructed from `AotNative<W>.CurrentContext`
Per-instance state	One `IWorker` per load	One `AotNative<W>` (and one wrapped `W`) per host pointer

Wire format (FlatBuffer)

The on-the-wire schema is the FlatBuffer defined in src/Virtufin.Worker.DevKit/Schemas/worker_api.fbs. Three top-level tables:

CloudEvent — the input to Process. Mirrors the CloudEvents 1.0 spec fields (id, type, source, specversion, datacontenttype, dataschema, subject, time, data: [ubyte]) plus an extensions: [Extension] vector carrying (name, value) string pairs. Non-string extension values are JSON-encoded by the engine before being placed in the wire buffer.
WorkerResponse — the output of Process. Exactly one of result_event: CloudEvent or error_message: string is set.
Extension — the (name, value) pair for the extensions vector.

namespace Virtufin.Worker.FlatBuffers;

table CloudEvent { id, type, source, specversion, datacontenttype,
                   dataschema, subject, time, data: [ubyte],
                   extensions: [Extension]; }
table Extension  { name, value: string; }
table WorkerResponse { result_event: CloudEvent; error_message: string; }

The matching generated C# types live in src/Virtufin.Worker.DevKit/Generated/ (CloudEvent.cs, Extension.cs, WorkerResponse.cs) and are checked in.

Regenerating the wire format

# Requires flatc (Google FlatBuffers compiler) on PATH.
flatc --csharp -o src/Virtufin.Worker.DevKit/Generated \
    src/Virtufin.Worker.DevKit/Schemas/worker_api.fbs

Regenerate when the schema changes; do not hand-edit the files under Generated/.

Worker DevKit

Class hierarchy

Public API (JIT side: DotNetDllEngine)

IWorker and WorkerBase<TInput, TOutput>

Response helpers

replytopic extension

Worker-instance config, delivered as extension attributes

CommandWorker<T> — typed command dispatch

ApiWorker and ApiCommandWorker<T>

AOT native side (NativeDllEngine)

AotNative<W>

AotNative<W>.CurrentContext