@vampgg/utils

Low-level utilities for building bebop/tempo-based applications. Provides the shared wire schema, transport channel/router implementations, error types, loggers, and async iteration primitives used across the vamp.

• bebop — binary serialization library for fast, compact data interchange
• tempo — typed RPC layer built on top of bebop

Development

vp install       # install dependencies
vp test          # run tests
vp run build         # regenerate schema then build (runs: bebopc build && vp pack)

src/bebop.ts is auto-generated by bebopc from the .bop schema files. Do not edit it by hand. Run bebopc build (or vp run build) to regenerate after changing schema/*.bop.

Schema (schema/)

Two bebop types define the wire format for all RPC in this framework.

Error (struct — schema/error.bop)

Structured error type carried inside tempo error responses.

| Field | Type | Description | | --------- | --------------------- | -------------------------------------------- | | code | uint32 | TempoStatusCode value | | tag | string | Machine-readable error tag (see ErrorTags) | | stack | string | Stack trace | | msg | string | Human-readable message | | details | map[string, string] | Arbitrary key-value context |

Message (message — schema/message.bop)

The universal envelope for all RPC calls. Fields are optional (bebop message type — missing fields are simply absent on the wire).

| Field | Type | Description | | --------------- | -------- | --------------------------------------- | | methodId | uint32 | Numeric identifier for the RPC method | | messageId | guid | UUID correlating request/response | | timestamp | date | When the message was created | | data | byte[] | Bebop-encoded request or response body | | deadline | date | Absolute deadline for the call | | status | byte | TempoStatusCode (OK, CANCELLED, etc.) | | msg | string | Error message (on error responses) | | authorization | string | Auth header value | | credential | string | Serialized credential for storage | | headers | string | HTTP-style metadata header string |

Subpath Exports

@vampgg/utils/bebop — Generated Schema Bindings

Auto-generated. Exports Error, Message, and BEBOP_SCHEMA.

Each type exports as both an interface and a factory/static object (the bebop pattern):

import { Message, Error } from "@vampgg/utils/bebop";

// Encode
const bytes = Message.encode({ methodId: 1, messageId: crypto.randomUUID(), data: payload });

// Decode
const msg = Message.decode(bytes);

// Factory (adds .encode() method to instance)
const record = Message({ methodId: 1, messageId: "..." });
record.encode(); // Uint8Array

@vampgg/utils/error — Typed Errors

import { SystemError, ErrorTags } from "@vampgg/utils/error";

SystemError<Tag> extends TempoError from @tempojs/common. Wraps a bebop Error struct. Use this as the base for all thrown errors in the framework.

throw new SystemError(ErrorTags.Validation, {
  code: TempoStatusCode.INVALID_ARGUMENT,
  msg: "field 'name' is required",
  details: new Map([["field", "name"]]),
});

// Convert an unknown caught error into a SystemError
const err = SystemError.generic(caughtError, ErrorTags.Unknown);

ErrorTags enum — the set of recognized error tags:

| Tag | Meaning | | --------------------- | ------------------------------------ | | PgOutage | PostgreSQL connectivity failure | | RedisOutage | Redis connectivity failure | | Unknown | Unclassified error (default) | | NeedPermissions | Missing required permissions | | Validation | Input validation failure | | Authorization | Auth check failed | | GameStateGeneration | Game state generation failure | | InvalidPrompt | AI prompt rejected | | UnknownAIProvider | Unrecognized AI provider | | FailedToStart | Service failed to initialize | | Unimplemented | Method not yet implemented | | IncompatibleVersion | Client/server version mismatch | | PingUnacceptable | Ping latency out of acceptable range |

@vampgg/utils/context-logger — Base Logger

import { ContextLogger } from "@vampgg/utils/context-logger";

Extends ConsoleLogger from @tempojs/common. Adds structured bindings (key-value context attached to every log line) and methods for deriving child loggers.

const logger = new ContextLogger("my-service", TempoLogLevel.Info, undefined, { env: "prod" });

// Derive a child logger with additional bindings
const child = logger.clone("request-handler", false, { requestId: "abc" });

// Derive a child with a correlation ID
const correlated = logger.correlate("db-query", correlationId, false, { table: "users" });

ContextLogger is the base class — extend it (or use DebugLogger / PinoLogger) rather than instantiating it directly in most cases.

@vampgg/utils/debug-logger — Debug-Namespace Logger

import { DebugLogger } from "@vampgg/utils/debug-logger";

Uses the debug package. Each log level gets its own namespace: <sourceName>:trace, <sourceName>:info, etc. Enable output with the DEBUG environment variable.

const logger = new DebugLogger("my-service");
// Enable with: DEBUG=my-service:* node ...

logger.info("server started", { port: 3000 });
logger.error("db failed", { query }, err);

const child = logger.clone("auth", false, { userId: "123" });

@vampgg/utils/pino-logger — Pino Structured Logger

import { PinoLogger } from "@vampgg/utils/pino-logger";
import pino from "pino";

Wraps a pino instance. Pass an existing pino.Logger to integrate with your existing pino setup (transports, formatters, etc.).

const root = pino({ level: "info" });
const logger = new PinoLogger("my-service", TempoLogLevel.Info, root, { env: "prod" });

logger.info("started");
logger.warn("slow query", { duration: 450 });

const child = logger.clone("auth");

@vampgg/utils/async-queue — Bounded Async Queue

A single-consumer async queue that backs both iterators below. A backing ring buffer (no Array.shift()), a high-water mark with an explicit overflow policy, and a real error channel — so a slow consumer can't grow memory without bound and a transport error surfaces at the for await instead of looking like a clean end-of-stream.

import { AsyncQueue, AsyncQueueOverflowError } from "@vampgg/utils/async-queue";

const q = new AsyncQueue<number>(
  { highWaterMark: 1024, overflow: "error" }, // or "drop-latest"
  () => cleanup(), // optional onDispose, called on close/return/throw
);

q.push(1); // returns false if dropped (overflow: "drop-latest")
q.close(); // graceful end → consumer sees { done: true }
q.fail(new Error("transport died")); // error channel → thrown at the consumer

for await (const n of q) {
  /* … */
}

overflow: "error" makes push past highWaterMark raise AsyncQueueOverflowError at the consumer; "drop-latest" silently drops instead.

@vampgg/utils/create-event-iterator — Push-Based Async Generator

import { createEventIterator } from "@vampgg/utils/create-event-iterator";
import type { Subscriber, Context, CleanupFn } from "@vampgg/utils/create-event-iterator";

Converts any push-based event source (EventEmitter, WebSocket messages, browser events) into an AsyncGenerator. The subscriber receives { emit, cancel, error } (error(e) surfaces a failure at the consumer instead of a clean end) and returns an optional cleanup function.

const stream = createEventIterator<MyEvent>(({ emit, cancel }) => {
  const handler = (ev: MyEvent) => emit(ev);
  emitter.on("data", handler);

  return () => {
    emitter.off("data", handler);
  };
});

for await (const event of stream) {
  // process event
}

Call cancel() from within the subscriber (e.g. on a "close" event) to end iteration. Remaining buffered events are yielded before the generator returns.

@vampgg/utils/create-duplex-iterator — Bidirectional Async Stream

import { createDuplexIterator } from "@vampgg/utils/create-duplex-iterator";
import type { Writer } from "@vampgg/utils/create-duplex-iterator";

Combines an outgoing AsyncGenerator<BebopRecord> with an incoming event source. For each outgoing record it calls emitter, then yields the next incoming event. Used internally by duplex-stream RPC methods.

const stream = createDuplexIterator<IncomingRecord>(
  outgoingGenerator, // AsyncGenerator<BebopRecord> — client sends these
  (record) => ws.send(record.encode()), // called for each outgoing record
  ({ emit, cancel }) => {
    // subscriber for incoming records
    ws.onmessage = (ev) => emit(decode(ev.data));
    return () => {
      ws.onmessage = null;
    };
  },
);

for await (const incoming of stream) {
  // interleaved with outgoing sends
}

@vampgg/utils/ws-channel — WebSocket Tempo Client Channel

import { TempoWSChannel } from "@vampgg/utils/ws-channel";
import type { TempoWSChannelOptions } from "@vampgg/utils/ws-channel";

A BaseChannel implementation that transports tempo RPC calls over a WebSocket connection (using the websocket-ts library with auto-reconnect support). The wire format is Message encoded as bebop binary.

// Connect to a new WebSocket
const channel = TempoWSChannel.forAddress("wss://api.example.com/rpc", {
  maxRetries: 3,
});

// Wrap an existing WebSocket
const channel = TempoWSChannel.forWebSocket(existingWs);

await channel.waitForOpen();

// Used by tempo-generated client stubs — not called directly
// channel.startUnary(request, context, method, options)
// channel.startServerStream(...)
// channel.startClientStream(...)
// channel.startDuplexStream(...)

All four tempo method types (unary, server-stream, client-stream, duplex) are supported with deadline and retry policy propagation.

Full client round-trip. In practice you connect, obtain a tempo-generated client stub from the channel, and call RPC methods directly — the channel handles encoding, correlation, and streaming:

import { TempoWSChannel } from "@vampgg/utils/ws-channel";
import { RpcClient, MutationScope } from "./bebop"; // your tempo-generated client + types

const channel = TempoWSChannel.forAddress("wss://host/v1/game?ns=room1", { maxRetries: 3 });
await channel.waitForOpen();

const client = channel.getClient(RpcClient);

const spawned = await client.spawn(entity); // unary
for await (const scope of await client.observe(MutationScope({}))) {
  // server-stream: each frame is a decoded record
}

channel.close();

@vampgg/utils/ws-router — WebSocket Tempo Server Router

import { TempoWsRouter } from "@vampgg/utils/ws-router";

A BaseRouter that dispatches incoming ArrayBuffer WebSocket messages to registered tempo service methods. Extend or instantiate with a ServiceRegistry and optional AuthInterceptor.

const router = new TempoWsRouter<MyContext>(logger, registry, configuration, authInterceptor);
router.useHooks(hookRegistry);

// In your WebSocket message handler:
ws.onmessage = async (ev) => {
  const response = Message({});
  await router.process(ev.data as ArrayBuffer, response, [ws, myContext]);
};

Server-stream and duplex-stream methods must be implemented as async function* generators in the service. Streams send successive Message frames until a CANCELLED status frame closes the stream.

@vampgg/utils/worker-channel — Bun Worker Tempo Client Channel

import { TempoWorkerChannel } from "@vampgg/utils/worker-channel";

A BaseChannel that communicates with a Bun Worker using postMessage / onmessage. Messages are transferred as Uint8Array with zero-copy buffer transfer.

const channel = new TempoWorkerChannel("./my-worker.ts");
await channel.ready; // resolves when the worker sends its first message

// channel.startUnary(...) / startServerStream(...) etc.

The worker property exposes the underlying Bun.Worker instance if you need direct access.

@vampgg/utils/worker-router — Bun Worker Tempo Server Router

import { TempoWorkerRouter } from "@vampgg/utils/worker-router";

Runs inside a Bun Worker (declare var self: Bun.Worker). Dispatches incoming Buffer messages to registered tempo service methods, writing responses back via postMessage.

// Inside worker.ts
const router = new TempoWorkerRouter<MyEnv>(logger, registry);

self.onmessage = async (ev: MessageEvent<Uint8Array>) => {
  const response = Message({});
  await router.process(ev.data, response, myEnv);
};

// Signal ready to the channel
postMessage(new Uint8Array(0));

Supports client-stream cancellation: a CANCELLED status message from the client triggers generator cleanup and stops the stream.

@vampgg/utils/extension-channel — Browser Extension Channel (Client)

import { TempoExtensionChannel } from "@vampgg/utils/extension-channel";

A BaseChannel for browser extensions. Sends messages via browser.runtime.sendMessage and receives responses via browser.runtime.onMessage. Uses webextension-polyfill for cross-browser compatibility.

const channel = new TempoExtensionChannel({ logger });
// channel.startUnary(...) etc. — used by tempo-generated client stubs

Messages are encoded as Array<number> (bebop bytes as a plain array) to satisfy browser extension message serialization constraints.

@vampgg/utils/extension-router — Browser Extension Router (Server)

import { TempoExtensionRouter, ExtensionBaseRouter } from "@vampgg/utils/extension-router";
import { TempoExtensionRouterConfiguration } from "@vampgg/utils/extension-router";

Runs in the extension background script. Dispatches incoming Message objects (decoded from browser.runtime.onMessage) to registered tempo service methods and sends responses back via tabs.sendMessage.

const router = new TempoExtensionRouter<{ sender: Runtime.MessageSender }>(
  logger,
  registry,
  new TempoRouterConfiguration(),
);

browser.runtime.onMessage.addListener(async (raw, sender) => {
  const msg = Message(Message.decode(new Uint8Array(raw as number[])));
  const response = Message({});
  await router.process(msg, response, { sender });
});

@vampgg/utils/memory-storage — In-Memory Credential Storage

import { MemoryStorageStrategy } from "@vampgg/utils/memory-storage";

Implements CredentialStorage from @tempojs/client using a Map. Suitable for testing or short-lived processes where persistence is not required.

const storage = new MemoryStorageStrategy();
await storage.storeCredential("session", credential);
const cred = await storage.getCredential("session");
await storage.removeCredential("session");

@vampgg/utils/sync-storage — WebExtension Sync Storage

import { SyncStorageStrategy } from "@vampgg/utils/sync-storage";

Implements CredentialStorage backed by browser.storage.sync (via webextension-polyfill). Use in browser extension contexts where credentials should persist across browser sessions and sync across devices.

const storage = new SyncStorageStrategy();
await storage.storeCredential("auth", credential);

Key Patterns

Adding a New Transport

To add a new transport (e.g., SharedWorker, BroadcastChannel):

1. Client side — extend BaseChannel from @tempojs/client. Implement startUnary, startClientStream, startServerStream, startDuplexStream. Encode requests as Message binary (Message.encode(msg)), decode responses with Message.decode(bytes). Use createEventIterator for server-stream and createDuplexIterator for duplex-stream.

2. Server side — extend BaseRouter from @tempojs/server. Implement process(rawRequest, responseMessage, env). Decode the incoming buffer with Message.decode(...), dispatch by methodId via this.registry.getMethod(methodId), encode responses with Message.encode(response).

Schema Regeneration

After editing schema/error.bop or schema/message.bop, regenerate the TypeScript bindings:

bebopc build
# or
vp run build  # runs bebopc build then packs the library

Never manually edit src/bebop.ts.

Wire Protocol

All transports use the same protocol:

1. Client creates a Message with methodId, messageId (UUID), data (bebop-encoded request body), optional deadline and authorization.
2. Server decodes the Message, routes by methodId, executes the method, and replies with a Message containing the same messageId, status: OK, and data (bebop-encoded response body).
3. For streams, the server sends multiple frames with status: OK and terminates with status: CANCELLED and an empty data.
4. Errors are returned as a Message with a non-OK status and a human-readable msg.