Durable Objects

@trpc12/trpc/cloudflare has what a router needs inside a Durable Object. The examples/cloudflare app uses all of it.

Hibernating WebSockets

A Durable Object can accept WebSockets with hibernation, so idle connections don't keep it in memory (or on the bill). experimental_createHibernatingWebSockets() serves a handler's router that way, from the object's fetch, webSocketMessage and webSocketClose:

import { class DurableObject<Env = Cloudflare.Env, Props = {}>DurableObject } from 'cloudflare:workers';
import { function experimental_createHibernatingWebSockets<TPlatform>(handler: FetchHandler<TPlatform>, opts: HibernatingWebSocketsOptions): HibernatingWebSockets<TPlatform>experimental_createHibernatingWebSockets } from '@trpc12/trpc/cloudflare';
import { function createFetchHandler<TRouter extends AnyRouter, TPlatform = undefined>(opts: FetchHandlerOptions<TRouter, TPlatform>): FetchHandler<TPlatform>createFetchHandler } from '@trpc12/trpc/server';

const const handler: FetchHandler<{
    env: Env;
}>handler = createFetchHandler<{
    post: {
        byId: BuildProcedure<InputDef<RootDef<Context, {}>, ZodObject<{
            id: ZodString;
        }, $strip>>, "query", {
            output: Post;
            errors: TrpcError<"NOT_FOUND", {
                id: string;
            }>;
            services: never;
            unmapped: never;
        }>;
        list: BuildProcedure<RootDef<Context, {}>, "query", {
            output: Post[];
            errors: never;
            services: never;
            unmapped: never;
        }>;
        create: BuildProcedure<InputDef<With<RootDef<Context, {}>, {
            ctx: Context;
            ctxAdded: {
                ...;
            };
            inputOut: Unset;
            errors: TrpcError<...>;
            shorts: never;
        }>, ZodObject<...>>, "mutation", {
            ...;
        }>;
    };
}, {
    ...;
}>(opts: FetchHandlerOptions<...>): FetchHandler<...>createFetchHandler<type AppRouter = {
    post: {
        byId: BuildProcedure<InputDef<RootDef<Context, {}>, ZodObject<{
            id: ZodString;
        }, $strip>>, "query", {
            output: Post;
            errors: TrpcError<"NOT_FOUND", {
                id: string;
            }>;
            services: never;
            unmapped: never;
        }>;
        list: BuildProcedure<RootDef<Context, {}>, "query", {
            output: Post[];
            errors: never;
            services: never;
            unmapped: never;
        }>;
        create: BuildProcedure<InputDef<With<RootDef<Context, {}>, {
            ctx: Context;
            ctxAdded: {
                ...;
            };
            inputOut: Unset;
            errors: TrpcError<...>;
            shorts: never;
        }>, ZodObject<...>>, "mutation", {
            ...;
        }>;
    };
}AppRouter, { env: Envenv: Env }>({
  BaseHandlerOptions<{ post: { byId: BuildProcedure<InputDef<RootDef<Context, {}>, ZodObject<{ id: ZodString; }, $strip>>, "query", { output: Post; errors: TrpcError<...>; services: never; unmapped: never; }>; list: BuildProcedure<...>; create: BuildProcedure<...>; }; }>.router: {
    post: {
        byId: BuildProcedure<InputDef<RootDef<Context, {}>, ZodObject<{
            id: ZodString;
        }, $strip>>, "query", {
            output: Post;
            errors: TrpcError<"NOT_FOUND", {
                id: string;
            }>;
            services: never;
            unmapped: never;
        }>;
        list: BuildProcedure<RootDef<Context, {}>, "query", {
            output: Post[];
            errors: never;
            services: never;
            unmapped: never;
        }>;
        create: BuildProcedure<InputDef<With<RootDef<Context, {}>, {
            ctx: Context;
            ctxAdded: {
                ...;
            };
            inputOut: Unset;
            errors: TrpcError<...>;
            shorts: never;
        }>, ZodObject<...>>, "mutation", {
            ...;
        }>;
    };
}router,
  createContext: (opts: CreateContextOpts<{
    env: Env;
}>) => MaybePromise<{
    user: User | null;
}>createContext,
});

export class class RoomRoom extends class DurableObject<Env = Cloudflare.Env, Props = {}>DurableObject<Env> {
  Room.sockets: HibernatingWebSockets<{
    env: Env;
}>sockets = experimental_createHibernatingWebSockets<{
    env: Env;
}>(handler: FetchHandler<{
    env: Env;
}>, opts: HibernatingWebSocketsOptions): HibernatingWebSockets<{
    env: Env;
}>experimental_createHibernatingWebSockets(const handler: FetchHandler<{
    env: Env;
}>handler, {
    HibernatingWebSocketsOptions.state: HibernationStatestate: this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx,
  });

  Room.fetch(request: Request): Response | Promise<Response>fetch(request: Request<unknown, CfProperties<unknown>>request: interface Request<CfHostMetadata = unknown, Cf = CfProperties<CfHostMetadata>>Request) {
    if (request: Request<unknown, CfProperties<unknown>>request.Request<unknown, CfProperties<unknown>>.headers: Headersheaders.Headers.get(name: string): string | nullget('upgrade') !== 'websocket') {
      return const handler: FetchHandler<{
    env: Env;
}>handler.FetchHandler<{ env: Env; }>.fetch: (request: Request, platform: {
    env: Env;
}) => Promise<Response>fetch(request: Request<unknown, CfProperties<unknown>>request, { env: Envenv: this.CloudflareWorkersModule.DurableObject<Env, {}>.env: Envenv });
    }
    return this.Room.sockets: HibernatingWebSockets<{
    env: Env;
}>sockets.HibernatingWebSockets<{ env: Env; }>.upgrade(request: Request, platform: {
    env: Env;
}): Responseupgrade(request: Request<unknown, CfProperties<unknown>>request, { env: Envenv: this.CloudflareWorkersModule.DurableObject<Env, {}>.env: Envenv });
  }
  Room.webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): voidwebSocketMessage(ws: WebSocketws: WebSocket, message: string | ArrayBuffermessage: string | ArrayBuffer) {
    this.Room.sockets: HibernatingWebSockets<{
    env: Env;
}>sockets.HibernatingWebSockets<{ env: Env; }>.message(socket: HibernatableWebSocket, data: string | ArrayBuffer, platform: {
    env: Env;
}): voidmessage(ws: WebSocketws, message: string | ArrayBuffermessage, { env: Envenv: this.CloudflareWorkersModule.DurableObject<Env, {}>.env: Envenv });
  }
  Room.webSocketClose(ws: WebSocket, code: number, reason: string): voidwebSocketClose(ws: WebSocketws: WebSocket, code: numbercode: number, reason: stringreason: string) {
    this.Room.sockets: HibernatingWebSockets<{
    env: Env;
}>sockets.HibernatingWebSockets<{ env: Env; }>.close(socket: HibernatableWebSocket, code?: number, reason?: string): voidclose(ws: WebSocketws, code: numbercode, reason: stringreason);
  }
  Room.webSocketError(ws: WebSocket): voidwebSocketError(ws: WebSocketws: WebSocket) {
    this.Room.sockets: HibernatingWebSockets<{
    env: Env;
}>sockets.HibernatingWebSockets<{ env: Env; }>.close(socket: HibernatableWebSocket, code?: number, reason?: string): voidclose(ws: WebSocketws);
  }
}

When the object wakes, each socket's request URL, its cookie and authorization headers (keepHeaders to choose others) and its connectionParams are rebuilt from the socket's attachment, so createContext sees the same connection. The attachment has a 2 KB limit; past it, the headers are dropped with a console.warn (the connection stays). Browsers can't set WebSocket headers, so same-origin apps authenticate with cookies.
wsLink's keep-alive pings are answered by the runtime (setWebSocketAutoResponse) without waking the object. The handler's websocket.idleTimeoutMs doesn't apply here: a timer would wake it.
A generator or Stream subscription keeps the object awake until it ends. Subscriptions to a hibernatingPublisher don't.
Already created the WebSocketPair yourself? sockets.accept(server, request, platform) instead of upgrade().

Subscriptions that hibernate

Provide the object's publisher with hibernatingPublisher(Posts, { state }). A subscription that returns Posts.subscribe(...) over a hibernating socket is then kept in the socket's attachment (call id, channel, path) instead of a running fiber, and Posts.publish() sends each event straight to the matching sockets, validated by the subscription's .output(). The object sleeps between events.

import {
  function experimental_createHibernatingWebSockets<TPlatform>(handler: FetchHandler<TPlatform>, opts: HibernatingWebSocketsOptions): HibernatingWebSockets<TPlatform>experimental_createHibernatingWebSockets,
  function hibernatingPublisher<Self, Key extends string, TEvents extends PublisherEvents>(publisher: PublisherClass<Self, Key, TEvents>, opts: HibernatingPublisherOptions): Layer<Self>hibernatingPublisher,
} from '@trpc12/trpc/cloudflare';

class class PostsPosts extends const Publisher: {
    Service: <Self, TEvents extends PublisherEvents>() => <const Key extends string>(key: Key) => PublisherClass<Self, Key, TEvents>;
}Publisher.type Service: <Posts, {
    created: Post;
}>() => <const Key extends string>(key: Key) => PublisherClass<Posts, Key, {
    created: Post;
}>Service<class PostsPosts, { created: Postcreated: Post }>()('Posts') {}

const const router: {
    create: BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<{
        id: ZodString;
        title: ZodString;
        author: ZodNullable<ZodString>;
        createdAt: ZodDate;
    }, $strip>>, "mutation", {
        output: {
            id: string;
            title: string;
            author: string | null;
            createdAt: Date;
        };
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}router = {
  create: BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>, "mutation", {
    output: {
        id: string;
        title: string;
        author: string | null;
        createdAt: Date;
    };
    errors: never;
    services: Posts;
    unmapped: never;
}>create: const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure
    .ProcedureBuilder<RootDef<{}, {}>>.input<ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>(schema: ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>): ProcedureBuilder<InputDef<RootDef<{}, {}>, ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>> (+1 overload)input(const PostInput: ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>PostInput)
    .ProcedureBuilder<InputDef<RootDef<{}, {}>, ZodObject<{ id: ZodString; title: ZodString; author: ZodNullable<ZodString>; createdAt: ZodDate; }, $strip>>>.mutation<Effect.Effect<{
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}, never, Posts>>(resolver: (opts: ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>>) => Effect.Effect<{
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}, never, Posts>): BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<...>>, "mutation", {
    ...;
}>mutation((opts: ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>>opts) =>
      import EffectEffect.const as: <string, never, Posts, {
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}>(self: Effect.Effect<string, never, Posts>, value: {
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}) => Effect.Effect<{
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}, never, Posts> (+1 overload)as(class PostsPosts.PublisherClass<Posts, "Posts", { created: Post; }>.publish<"created">(channel: "created", data: Post): Effect.Effect<string, never, Posts>publish('created', opts: ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>>opts.ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{ id: ZodString; title: ZodString; author: ZodNullable<ZodString>; createdAt: ZodDate; }, $strip>>>.input: {
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}input), opts: ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{
    id: ZodString;
    title: ZodString;
    author: ZodNullable<ZodString>;
    createdAt: ZodDate;
}, $strip>>>opts.ResolverOpts<InputDef<RootDef<{}, {}>, ZodObject<{ id: ZodString; title: ZodString; author: ZodNullable<ZodString>; createdAt: ZodDate; }, $strip>>>.input: {
    id: string;
    title: string;
    author: string | null;
    createdAt: Date;
}input),
    ),
  onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
    output: TrackedEnvelope<Post>;
    errors: never;
    services: Posts;
}>onCreated: const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure.ProcedureBuilder<RootDef<{}, {}>>.subscription<Stream<TrackedEnvelope<Post>, never, Posts>>(resolver: (opts: SubscriptionResolverOpts<RootDef<{}, {}>>) => Stream<TrackedEnvelope<Post>, never, Posts>): BuildProcedure<RootDef<{}, {}>, "subscription", {
    output: TrackedEnvelope<Post>;
    errors: never;
    services: Posts;
}>subscription((opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts) =>
    class PostsPosts.PublisherClass<Posts, "Posts", { created: Post; }>.subscribe<"created">(channel: "created", opts?: SubscribeOptions): Stream<TrackedEnvelope<Post>, never, Posts>subscribe('created', { SubscribeOptions.lastEventId?: string | undefinedlastEventId: opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts.SubscriptionResolverOpts<RootDef<{}, {}>>.lastEventId: string | undefinedlastEventId }),
  ),
};

export class class RoomRoom extends class DurableObject<Env = Cloudflare.Env, Props = {}>DurableObject<Env> {
  Room.handler: FetchHandler<undefined>handler = createFetchHandler<{
    create: BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<{
        id: ZodString;
        title: ZodString;
        author: ZodNullable<ZodString>;
        createdAt: ZodDate;
    }, $strip>>, "mutation", {
        output: {
            id: string;
            title: string;
            author: string | null;
            createdAt: Date;
        };
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}, undefined>(opts: FetchHandlerOptions<...>): FetchHandler<...>createFetchHandler({
    BaseHandlerOptions<{ create: BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<{ id: ZodString; title: ZodString; author: ZodNullable<ZodString>; createdAt: ZodDate; }, $strip>>, "mutation", { ...; }>; onCreated: BuildProcedure<...>; }>.router: {
    create: BuildProcedure<InputDef<RootDef<{}, {}>, ZodObject<{
        id: ZodString;
        title: ZodString;
        author: ZodNullable<ZodString>;
        createdAt: ZodDate;
    }, $strip>>, "mutation", {
        output: {
            id: string;
            title: string;
            author: string | null;
            createdAt: Date;
        };
        errors: never;
        services: Posts;
        unmapped: never;
    }>;
    onCreated: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Post>;
        errors: never;
        services: Posts;
    }>;
}router,
    layer: Layer<Posts, any, never>layer: hibernatingPublisher<Posts, "Posts", {
    created: Post;
}>(publisher: PublisherClass<Posts, "Posts", {
    created: Post;
}>, opts: HibernatingPublisherOptions): Layer<Posts, never, never>hibernatingPublisher(class PostsPosts, { HibernatingPublisherOptions.state: HibernationStatestate: this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx, HibernatingPublisherOptions.buffer?: number | undefinedbuffer: 100 }),
  });
  Room.sockets: HibernatingWebSockets<undefined>sockets = experimental_createHibernatingWebSockets<undefined>(handler: FetchHandler<undefined>, opts: HibernatingWebSocketsOptions): HibernatingWebSockets<undefined>experimental_createHibernatingWebSockets(this.Room.handler: FetchHandler<undefined>handler, {
    HibernatingWebSocketsOptions.state: HibernationStatestate: this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx,
  });
  // fetch, webSocketMessage and webSocketClose as above
}

Event ids are timestamps, so they keep growing across wakes. Kept events (buffer) are in memory: a reconnect with lastEventId replays the ones the object still has.
The subscription runs its resolver once, when it starts. Anything the resolver does per event (filtering, mapping the stream) is skipped; filter in the channel name instead ('room:42').
One adapter per object: the latest experimental_createHibernatingWebSockets for a state gets the events.
A subscription that doesn't fit the attachment runs as a normal stream, with a warning. SSE subscribers in the same object get a normal stream too.
A parked subscription keeps its channel, so with a channel per audience Notes.publish('inbox:1', data) only wakes the sockets subscribed to inbox:1.

Durable calls

experimental_durable() needs Effect's BackingPersistence. In an object, back it with the object's storage, so kept results and streamed frames outlive an eviction: Effect's KeyValueStore.makeStringOnly over ctx.storage, under Persistence.layerBackingKvs. examples/cloudflare/src/storage.ts has one, which the example merges with its publisher layer:

  Room.handler: FetchHandler<undefined>handler = createFetchHandler<{
    onMessage: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Message>;
        errors: never;
        services: Messages;
    }>;
}, undefined>(opts: FetchHandlerOptions<{
    onMessage: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Message>;
        errors: never;
        services: Messages;
    }>;
}, undefined>): FetchHandler<undefined>createFetchHandler({
    BaseHandlerOptions<{ onMessage: BuildProcedure<RootDef<{}, {}>, "subscription", { output: TrackedEnvelope<Message>; errors: never; services: Messages; }>; }>.router: {
    onMessage: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: TrackedEnvelope<Message>;
        errors: never;
        services: Messages;
    }>;
}router,
    layer: Layer.Layer<Messages, any, never>layer: import LayerLayer.const mergeAll: <[Layer.Layer<Messages, never, never>, Layer.Layer<Persistence.BackingPersistence, never, never>]>(layers_0: Layer.Layer<Messages, never, never>, layers_1: Layer.Layer<Persistence.BackingPersistence, never, never>) => Layer.Layer<Messages | Persistence.BackingPersistence, never, never>mergeAll(
      hibernatingPublisher<Messages, "Messages", {
    message: Message;
}>(publisher: PublisherClass<Messages, "Messages", {
    message: Message;
}>, opts: HibernatingPublisherOptions): Layer.Layer<Messages, never, never>hibernatingPublisher(class MessagesMessages, { HibernatingPublisherOptions.state: HibernationStatestate: this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx, HibernatingPublisherOptions.buffer?: number | undefinedbuffer: 100 }),
      function durableObjectPersistence(storage: DurableObjectStorage): Layer.Layer<Persistence.BackingPersistence>durableObjectPersistence(this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx.DurableObjectState<{}>.storage: DurableObjectStoragestorage),
    ),
  });

Locks

experimental_durableLocker(env.LOCKS) is an experimental_lock() locker every Worker instance shares: one object per key. Each caller keeps a hibernatable WebSocket to the object while it holds or waits for the lock, so the object isn't billed meanwhile, waiters are told when it's their turn (nothing polls), and a crashed holder's lock goes when its socket drops. The object delegates to experimental_durableLockObject(this.ctx):

import {
  function experimental_durableLocker(namespace: DurableLockNamespace, options?: experimental_DurableLockerOptions): Layer<experimental_Locker>experimental_durableLocker,
  function experimental_durableLockObject(state: DurableLockState): DurableLockObjectexperimental_durableLockObject,
} from '@trpc12/trpc/cloudflare';

export class class LocksLocks extends class DurableObject<Env = Cloudflare.Env, Props = {}>DurableObject<Env> {
  Locks.locks: DurableLockObjectlocks = function experimental_durableLockObject(state: DurableLockState): DurableLockObjectexperimental_durableLockObject(this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx);
  override Locks.fetch(request: Request): Responsefetch(request: Request<unknown, CfProperties<unknown>>request: interface Request<CfHostMetadata = unknown, Cf = CfProperties<CfHostMetadata>>Request) {
    return this.Locks.locks: DurableLockObjectlocks.DurableLockObject.fetch(request: Request): Responsefetch(request: Request<unknown, CfProperties<unknown>>request);
  }
  override Locks.webSocketClose(ws: WebSocket): voidwebSocketClose(ws: WebSocketws: WebSocket) {
    this.Locks.locks: DurableLockObjectlocks.DurableLockObject.close(socket: HibernatableWebSocket): voidclose(ws: WebSocketws);
  }
  override Locks.webSocketError(ws: WebSocket): voidwebSocketError(ws: WebSocketws: WebSocket) {
    this.Locks.locks: DurableLockObjectlocks.DurableLockObject.close(socket: HibernatableWebSocket): voidclose(ws: WebSocketws);
  }
}

// wherever the handler is made
createFetchHandler<{
    health: BuildProcedure<RootDef<{}, {}>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}, undefined>(opts: FetchHandlerOptions<{
    health: BuildProcedure<RootDef<{}, {}>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}, undefined>): FetchHandler<undefined>createFetchHandler({ BaseHandlerOptions<{ health: BuildProcedure<RootDef<{}, {}>, "query", { output: string; errors: never; services: never; unmapped: never; }>; }>.router: {
    health: BuildProcedure<RootDef<{}, {}>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}router, layer?: Layer<never, any, never> | undefinedlayer: function experimental_durableLocker(namespace: DurableLockNamespace, options?: experimental_DurableLockerOptions): Layer<experimental_Locker>experimental_durableLocker(const env: Envenv.Env.LOCKS: DurableObjectNamespace<Locks>LOCKS) });

Bind LOCKS to Locks in wrangler.jsonc (a new_sqlite_classes migration). The example locks a seat across two room objects.

Workers Traces

experimental_workersTracer({ tracing }) is an Effect Tracer that records spans with Workers Traces' custom spans API, with no OpenTelemetry SDK: tRPC's server and client spans and any Effect.withSpan, with their attributes and an error status on failure. Turn on observability.traces in the Wrangler config. Workers nests spans by async context, so pass snapshot: () => AsyncLocalStorage.snapshot() (nodejs_compat) too: then spans nest as Effect nests them, and the spans Workers adds itself (subrequests, storage, other objects) sit under the call that made them. Without it, every span is a child of the request's.

import { const tracing: Tracingtracing } from 'cloudflare:workers';
import { class AsyncLocalStorage<T>AsyncLocalStorage } from 'node:async_hooks';
import { function experimental_workersTracer(options: experimental_WorkersTracerOptions): Layer<never>experimental_workersTracer } from '@trpc12/trpc/cloudflare';

createFetchHandler<{
    health: BuildProcedure<RootDef<{}, {}>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}, undefined>(opts: FetchHandlerOptions<{
    health: BuildProcedure<RootDef<{}, {}>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}, undefined>): FetchHandler<undefined>createFetchHandler({
  BaseHandlerOptions<{ health: BuildProcedure<RootDef<{}, {}>, "query", { output: string; errors: never; services: never; unmapped: never; }>; }>.router: {
    health: BuildProcedure<RootDef<{}, {}>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}router,
  layer?: Layer<never, any, never> | undefinedlayer: function experimental_workersTracer(options: experimental_WorkersTracerOptions): Layer<never>experimental_workersTracer({
    experimental_WorkersTracerOptions.tracing: WorkersTracingtracing,
    experimental_WorkersTracerOptions.snapshot?: (() => AsyncContextRunner) | undefinedsnapshot: () => class AsyncLocalStorage<T>AsyncLocalStorage.AsyncLocalStorage<T>.snapshot(): <R, TArgs extends any[]>(fn: (...args: TArgs) => R, ...args: TArgs) => Rsnapshot(),
  }),
});

Workers spans have no events, so Effect's span events are dropped, and a traceparent from another service doesn't parent the call: its spans join the Worker's own trace. Locally, Wrangler's Local Explorer (e in wrangler dev) shows the spans; the example's smoke test reads them from its trace store.

Signed connection tokens

The Worker in front of the object usually authenticates the user already. connectionToken(schema, { secret }) lets it hand that over: a procedure on the Worker signs claims, the client sends the token as a connection param, and the object's createContext verifies it (HMAC-SHA256 with WebCrypto) instead of looking up the session again:

import { function connectionToken<S extends Schema.Codec<any, any>>(schema: S, opts: ConnectionTokenOptions): ConnectionToken<S["Type"]>connectionToken } from '@trpc12/trpc/cloudflare';

const const RoomToken: ConnectionToken<{
    readonly room: string;
    readonly user: string;
}>RoomToken = connectionToken<Schema.Struct<{
    readonly room: Schema.String;
    readonly user: Schema.String;
}>>(schema: Schema.Struct<{
    readonly room: Schema.String;
    readonly user: Schema.String;
}>, opts: ConnectionTokenOptions): ConnectionToken<{
    readonly room: string;
    readonly user: string;
}>connectionToken(
  import SchemaSchema.function Struct<{
    readonly room: Schema.String;
    readonly user: Schema.String;
}>(fields: {
    readonly room: Schema.String;
    readonly user: Schema.String;
}): Schema.Struct<{
    readonly room: Schema.String;
    readonly user: Schema.String;
}>Struct({ room: Schema.Stringroom: import SchemaSchema.const String: Schema.StringString, user: Schema.Stringuser: import SchemaSchema.const String: Schema.StringString }),
  { ConnectionTokenOptions.secret: string | Redacted<string>secret: const env: Envenv.Env.TOKEN_SECRET: stringTOKEN_SECRET, ConnectionTokenOptions.ttlSeconds?: number | undefinedttlSeconds: 600 },
);

// the Worker's router
const const workerRouter: {
    roomToken: BuildProcedure<InputDef<RootDef<{
        user: User;
    }, {}>, Schema.Struct<{
        readonly room: Schema.String;
    }>>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}workerRouter = {
  roomToken: BuildProcedure<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>, "query", {
    output: string;
    errors: never;
    services: never;
    unmapped: never;
}>roomToken: const t: TrpcRoot<{
    user: User;
}, {}, RootDef<{
    user: User;
}, {}>>t.TrpcRoot<{ user: User; }, {}, RootDef<{ user: User; }, {}>>.procedure: ProcedureBuilder<RootDef<{
    user: User;
}, {}>>procedure
    .ProcedureBuilder<RootDef<{ user: User; }, {}>>.input<Schema.Struct<{
    readonly room: Schema.String;
}>>(schema: Schema.Struct<{
    readonly room: Schema.String;
}>): ProcedureBuilder<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>> (+1 overload)input(import SchemaSchema.function Struct<{
    readonly room: Schema.String;
}>(fields: {
    readonly room: Schema.String;
}): Schema.Struct<{
    readonly room: Schema.String;
}>Struct({ room: Schema.Stringroom: import SchemaSchema.const String: Schema.StringString }))
    .ProcedureBuilder<InputDef<RootDef<{ user: User; }, {}>, Struct<{ readonly room: String; }>>>.query<Effect<string, never, never>>(resolver: (opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>>) => Effect<string, never, never>): BuildProcedure<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>, "query", {
    output: string;
    errors: never;
    services: never;
    unmapped: never;
}>query((opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>>opts) =>
      const RoomToken: ConnectionToken<{
    readonly room: string;
    readonly user: string;
}>RoomToken.ConnectionToken<{ readonly room: string; readonly user: string; }>.sign: (claims: {
    readonly room: string;
    readonly user: string;
}) => Effect<string>sign({ room: stringroom: opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>>opts.ResolverOpts<InputDef<RootDef<{ user: User; }, {}>, Struct<{ readonly room: String; }>>>.input: {
    readonly room: string;
}input.room: stringroom, user: stringuser: opts: ResolverOpts<InputDef<RootDef<{
    user: User;
}, {}>, Schema.Struct<{
    readonly room: Schema.String;
}>>>opts.ResolverOpts<InputDef<RootDef<{ user: User; }, {}>, Struct<{ readonly room: String; }>>>.ctx: {
    user: User;
}ctx.user: Useruser.User.id: stringid }),
    ),
};

// the Durable Object's router and handler
const const roomRouter: {
    inbox: BuildProcedure<RootDef<{
        room: string;
        user: string;
    }, {}>, "subscription", {
        output: TrackedEnvelope<{
            text: string;
        }>;
        errors: never;
        services: Notes;
    }>;
}roomRouter = {
  inbox: BuildProcedure<RootDef<{
    room: string;
    user: string;
}, {}>, "subscription", {
    output: TrackedEnvelope<{
        text: string;
    }>;
    errors: never;
    services: Notes;
}>inbox: const t: TrpcRoot<{
    room: string;
    user: string;
}, {}, RootDef<{
    room: string;
    user: string;
}, {}>>t.TrpcRoot<{ room: string; user: string; }, {}, RootDef<{ room: string; user: string; }, {}>>.procedure: ProcedureBuilder<RootDef<{
    room: string;
    user: string;
}, {}>>procedure.ProcedureBuilder<RootDef<{ room: string; user: string; }, {}>>.subscription<Stream<TrackedEnvelope<{
    text: string;
}>, never, Notes>>(resolver: (opts: SubscriptionResolverOpts<RootDef<{
    room: string;
    user: string;
}, {}>>) => Stream<TrackedEnvelope<{
    text: string;
}>, never, Notes>): BuildProcedure<RootDef<{
    room: string;
    user: string;
}, {}>, "subscription", {
    output: TrackedEnvelope<{
        text: string;
    }>;
    errors: never;
    services: Notes;
}>subscription((opts: SubscriptionResolverOpts<RootDef<{
    room: string;
    user: string;
}, {}>>opts) =>
    class NotesNotes.PublisherClass<Notes, "Notes", { [inbox: `inbox:${string}`]: { text: string; }; }>.subscribe<`inbox:${string}`>(channel: `inbox:${string}`, opts?: SubscribeOptions): Stream<TrackedEnvelope<{
    text: string;
}>, never, Notes>subscribe(`inbox:${opts: SubscriptionResolverOpts<RootDef<{
    room: string;
    user: string;
}, {}>>opts.ResolverOpts<RootDef<{ room: string; user: string; }, {}>>.ctx: {
    room: string;
    user: string;
}ctx.user: stringuser}`, {
      SubscribeOptions.lastEventId?: string | undefinedlastEventId: opts: SubscriptionResolverOpts<RootDef<{
    room: string;
    user: string;
}, {}>>opts.SubscriptionResolverOpts<RootDef<{ room: string; user: string; }, {}>>.lastEventId: string | undefinedlastEventId,
    }),
  ),
};

export class class RoomRoom extends class DurableObject<Env = Cloudflare.Env, Props = {}>DurableObject<Env> {
  Room.handler: FetchHandler<undefined>handler = createFetchHandler<{
    inbox: BuildProcedure<RootDef<{
        room: string;
        user: string;
    }, {}>, "subscription", {
        output: TrackedEnvelope<{
            text: string;
        }>;
        errors: never;
        services: Notes;
    }>;
}, undefined>(opts: FetchHandlerOptions<{
    inbox: BuildProcedure<RootDef<{
        room: string;
        user: string;
    }, {}>, "subscription", {
        output: TrackedEnvelope<{
            text: string;
        }>;
        errors: never;
        services: Notes;
    }>;
}, undefined>): FetchHandler<undefined>createFetchHandler({
    BaseHandlerOptions<{ inbox: BuildProcedure<RootDef<{ room: string; user: string; }, {}>, "subscription", { output: TrackedEnvelope<{ text: string; }>; errors: never; services: Notes; }>; }>.router: {
    inbox: BuildProcedure<RootDef<{
        room: string;
        user: string;
    }, {}>, "subscription", {
        output: TrackedEnvelope<{
            text: string;
        }>;
        errors: never;
        services: Notes;
    }>;
}router: const roomRouter: {
    inbox: BuildProcedure<RootDef<{
        room: string;
        user: string;
    }, {}>, "subscription", {
        output: TrackedEnvelope<{
            text: string;
        }>;
        errors: never;
        services: Notes;
    }>;
}roomRouter,
    createContext: (opts: CreateContextOpts<undefined>) => MaybePromise<{
    room: string;
    user: string;
}>createContext: (opts: CreateContextOpts<undefined>opts) =>
      import EffectEffect.const runPromise: <{
    readonly room: string;
    readonly user: string;
}, TrpcError<string, unknown>>(effect: Effect.Effect<{
    readonly room: string;
    readonly user: string;
}, TrpcError<string, unknown>, never>, options?: Effect.RunOptions | undefined) => Promise<{
    readonly room: string;
    readonly user: string;
}>runPromise(
        const RoomToken: ConnectionToken<{
    readonly room: string;
    readonly user: string;
}>RoomToken.ConnectionToken<{ readonly room: string; readonly user: string; }>.verify: (token: unknown) => Effect.Effect<{
    readonly room: string;
    readonly user: string;
}, TrpcError<string, unknown>, never>verify(opts: CreateContextOpts<undefined>opts.CreateContextOpts<undefined>.info: {
    readonly calls: ReadonlyArray<{
        path: string;
        type: ProcedureType;
    }>;
    readonly signal: AbortSignal;
    readonly transport: Transport;
    readonly connectionParams: Record<string, unknown> | undefined;
}info.connectionParams: Record<string, unknown> | undefinedconnectionParams?.['token']),
      ),
    layer: Layer<Notes, any, never>layer: hibernatingPublisher<Notes, "Notes", {
    [inbox: `inbox:${string}`]: {
        text: string;
    };
}>(publisher: PublisherClass<Notes, "Notes", {
    [inbox: `inbox:${string}`]: {
        text: string;
    };
}>, opts: HibernatingPublisherOptions): Layer<Notes, never, never>hibernatingPublisher(class NotesNotes, { HibernatingPublisherOptions.state: HibernationStatestate: this.CloudflareWorkersModule.DurableObject<Env, {}>.ctx: DurableObjectState<{}>ctx }),
  });
  // sockets, fetch and webSocketMessage as above
}

// the browser
const const link: SocketLink<{
    form: JsonForm;
}>link = wsLink<{
    url: string;
    connectionParams: () => Promise<{
        token: string;
    }>;
}>(opts: {
    url: string;
    connectionParams: () => Promise<{
        token: string;
    }>;
}): SocketLink<{
    form: JsonForm;
}>wsLink({
  url: stringurl,
  connectionParams: () => Promise<{
    token: string;
}>connectionParams: async () => ({
    token: stringtoken: await const client: TrpcClient<{
    roomToken: BuildProcedure<InputDef<RootDef<{
        user: User;
    }, {}>, Struct<{
        readonly room: String;
    }>>, "query", {
        output: string;
        errors: never;
        services: never;
        unmapped: never;
    }>;
}, {}, undefined, LinksTypes<readonly [SocketLink<{
    form: JsonForm;
}>]>>client.roomToken: {
    query(input: {
        readonly room: string;
    }, opts?: CallOptions<{}> | undefined): TrpcPromise<string, CallError<{
        type: "query";
        ctx: {
            user: User;
        };
        meta: {};
        input: {
            readonly room: string;
        };
        output: string;
        errors: InputValidationError<EffectInputIssue<{
            readonly room: string;
        }>>;
        services: never;
    }, JsonForm, never>>;
}roomToken.function query(input: {
    readonly room: string;
}, opts?: CallOptions<{}> | undefined): TrpcPromise<string, CallError<{
    type: "query";
    ctx: {
        user: User;
    };
    meta: {};
    input: {
        readonly room: string;
    };
    output: string;
    errors: InputValidationError<EffectInputIssue<{
        readonly room: string;
    }>>;
    services: never;
}, JsonForm, never>>query({ room: stringroom }),
  }),
});

verify fails with UNAUTHORIZED when the token is missing, malformed, signed with another secret or past ttlSeconds (default an hour). It reads the time from Effect's Clock.
The claims are encoded with the schema and readable by the client; keep secrets out of them.
Expiry is checked whenever createContext runs: on each call, and when a subscription starts. A parked subscription keeps getting events after the token expires, until the socket closes or the client reconnects.