The Effect RPC API layer

This repo used to speak tRPC. It now speaks Effect v4 RPC. The shape is familiar — typed procedures the client calls like local functions — but the whole thing is built out of Effect's Schema, RpcGroup, Layer, and middleware. That's the point of this page: to show you that the RPC layer is not a framework you memorise, it's just Effect primitives you already know, composed in one predictable way.

The surface lives in packages/api/src/rpc/. Three kinds of file matter:

• *-schema.ts — the contract: request/response/error Schemas.
• handlers/*.ts — the implementation: each procedure as an Effect.
• schema.ts + router.ts — the wiring: contracts and handlers meet, and the whole thing becomes a Layer the Next.js route can run.

The web app never imports a handler. It imports the contract (ApiRpc) and a small client runtime, and gets a fully-typed client for free. Let's walk the whole loop.

1. The contract: *-schema.ts

Every domain owns a schema file whose only job is to define the shapes that cross the wire. Here's the audio domain — packages/api/src/rpc/audio-schema.ts:

import { Schema } from "effect";

const Id = Schema.Number.check(Schema.isInt());

/** The persisted audio row, as returned by reads + mutations. */
export const Audio = Schema.Struct({
  id: Id,
  userId: Schema.NonEmptyString,
  filename: Schema.String,
  path: Schema.String,
  originalName: Schema.String,
  mimeType: Schema.String,
  size: Id,
  /** Length in seconds; null until probed. */
  duration: Schema.NullOr(Id),
  createdAt: Schema.Date,
  updatedAt: Schema.Date,
});
export type Audio = typeof Audio.Type;

Two things to internalise here:

• Audio is a runtime value first, a type second. Schema.Struct builds an actual object that can decode an unknown payload and encode a value back to the wire. type Audio = typeof Audio.Type derives the static type from that runtime schema, so the two can never drift. You never hand-write a parallel type Audio = { id: number; ... } — that's the rule the file header calls out: "Derive types via type X = typeof X.Type so the runtime schema stays authoritative."
• Inputs are separate, narrower schemas. The same file defines CreateAudioInput (what a caller may supply) and AudioIdInput ({ id }). userId is deliberately absent from CreateAudioInput — it comes from the authenticated session on the server, never from the client. Encoding trust boundaries into the schema is how you make "the client can't set its own userId" a compile-time and runtime fact instead of a code-review hope.

Errors are Schemas too

The wire needs to carry failures as well as successes, and in Effect a typed error channel is first-class. packages/api/src/rpc/errors.ts defines the whole error vocabulary as tagged Schemas:

import { Schema } from "effect";

export class Unauthorized extends Schema.TaggedErrorClass<Unauthorized>()("Unauthorized", {
  message: Schema.String,
}) {}

export class NotFound extends Schema.TaggedErrorClass<NotFound>()("NotFound", {
  resource: Schema.String,
  id: Schema.Union([Schema.String, Schema.Number]),
}) {}

// ...Forbidden, Internal...

export const RpcError = Schema.Union([Unauthorized, Forbidden, NotFound, Internal]);
export type RpcError = Schema.Schema.Type<typeof RpcError>;

Schema.TaggedErrorClass gives each error a _tag ("NotFound", …) and makes it serialisable, so the client receives a real typed union — not a stringified message. RpcError is the Schema.Union of the four, and almost every procedure declares error: RpcError. The four members map cleanly to intent: Unauthorized (no/invalid session), Forbidden (logged in, wrong role), NotFound (resource + id), and Internal (an infra/DB fault, redacted before it reaches the client).

2. The procedures: RpcGroup in schema.ts

packages/api/src/rpc/schema.ts is where individual procedures are declared and grouped. It imports the domain schemas and describes each call with Rpc.make(name, { payload, success, error }):

import { Rpc, RpcGroup } from "effect/unstable/rpc";

import { Audio, AudioIdInput, CreateAudioInput } from "./audio-schema";
import { RpcError } from "./errors";

const AudioRpc = RpcGroup.make(
  Rpc.make("list", { success: Schema.Array(Audio), error: RpcError }),
  Rpc.make("get", { payload: AudioIdInput, success: Audio, error: RpcError }),
  Rpc.make("create", { payload: CreateAudioInput, success: Audio, error: RpcError }),
  Rpc.make("delete", { payload: AudioIdInput, success: Audio, error: RpcError }),
  Rpc.make("transcribe", { payload: TranscribeAudioInput, success: Transcript, error: RpcError })
).prefix("audio/");

• Each Rpc.make is one procedure. payload is the request Schema (omit it for no-arg calls like list), success is the response Schema, error the failure Schema. Because these are Schemas, the RPC machinery decodes the payload on the server and the result on the client automatically — there is no separate validation step you can forget.
• .prefix("audio/") namespaces every member, so the procedures become audio/list, audio/get, audio/create, … Those prefixed strings are the keys you'll see everywhere: they're the handler keys, the client method names, and the telemetry span names, all the same string.

A group can also widen its error channel per-procedure. The video group's generate adds a render-specific failure on top of the shared union:

Rpc.make("generate", {
  payload: GenerateVideoInput,
  success: GeneratedVideo,
  // The render seam adds a `RenderError` member on top of the shared
  // `RpcError` union (auth + db failures still flow through here too).
  error: Schema.Union([...RpcError.members, RenderError]),
});

RpcError.members is the array of the union's constituents — spreading it keeps all four shared errors and adds RenderError, so the client's typed error union for video/generate is exactly those five.

Groups compose into one ApiRpc, gated by middleware

At the bottom of schema.ts, every group is merged into a single ApiRpc, and this is where authentication is attached — declaratively, per group:

export const ApiRpc = PublicRpc.merge(
  PrivateRpc,
  TodoRpc.middleware(SessionMiddleware),
  TravelRpc.middleware(SessionMiddleware),
  DealsRpc.middleware(SessionMiddleware),
  DealsServiceRpc.middleware(ServiceTokenMiddleware),
  AudioRpc.middleware(SessionMiddleware)
  // ...plan, telemetry, analytics, sentry, video, i18n...
);

• .middleware(SessionMiddleware) says "every procedure in this group requires a resolved session." The middleware provides a Session service to the handler and can fail the whole call with Unauthorized before the handler ever runs.
• PublicRpc (just health) carries no middleware — it's callable without a session. DealsServiceRpc uses ServiceTokenMiddleware instead: it's the n8n-facing deals/upsertOffers, gated by a shared secret header rather than a user session.

3. The implementation: handlers/

A handler file exports a function that takes its dependencies (repos, services) and returns the slice of procedures it implements. Here's the audio slice — packages/api/src/rpc/handlers/audio.ts:

export const audioHandlers = (audioRepo: AudioRepoShape): AudioSlice => ({
  "audio/list": () =>
    adminHandler("audio/list")(
      Effect.gen(function* () {
        const session = yield* Session;
        return yield* audioRepo.listForUser(session.user.id);
      })
    ),

  "audio/get": (input) =>
    adminHandler("audio/get")(
      Effect.gen(function* () {
        const session = yield* Session;
        const row = yield* audioRepo.getById(input.id, session.user.id);
        if (row === undefined) {
          return yield* new NotFound({ resource: "audio", id: input.id });
        }
        return row;
      })
    ),
  // ...create, delete...
});

Read one handler carefully — the pattern repeats everywhere:

• The key is the prefixed procedure name ("audio/get"), matching the Rpc.make in schema.ts. TypeScript checks the payload and return types against the contract, so a handler that returns the wrong shape won't compile.
• const session = yield* Session pulls the authenticated session out of context. It's there because SessionMiddleware provided it (step 2). Notice the handler never touches headers or cookies — auth already happened.
• A missing row is a typed failure, not a thrown error. return yield* new NotFound({ resource: "audio", id: input.id }) fails the effect with a NotFound — one of the RpcError members — which the client receives as a typed error. You never throw; you fail the channel.
• adminHandler("audio/get")(...) wraps the body. That wrapper is the house style for a whole class of concerns, and it's worth understanding because you'll apply it to every new handler.

What adminHandler actually does

From packages/api/src/rpc/handlers/shared.ts:

export const adminHandler =
  (method: string) =>
  <A, E, R>(body: Effect.Effect<A, E, R>) =>
    Effect.gen(function* () {
      yield* requireAdmin;
      return yield* body.pipe(failureToInternal(method));
    }).pipe(withRpcSpan(method));

It composes three cross-cutting concerns so each handler body stays "just the Session read + repo call + a NotFound guard":

1. requireAdmin reads the Session and fails Forbidden unless the user's role is admin. Because it sits outside failureToInternal, a denial stays a 403 Forbidden, never a 500.
2. failureToInternal(method) maps any untyped repo failure (a raw Drizzle error) to the redacted Internal error — but lets an already-typed NotFound/Internal/Forbidden pass straight through untouched, so your explicit NotFound wins. This is why raw Postgres text (table/column names) never leaks to the client.
3. withRpcSpan(method) wraps the whole thing in an OpenTelemetry rpc.<method> span plus request/error/latency metrics, and annotates the span with the end-user id. The method string is passed once so span name, metric tag, and error-mapping label can't drift apart.

4. The wiring: router.ts turns handlers into a Layer

schema.ts gives you the contract; the handlers are plain functions. router.ts marries them and produces the runnable Layer. The key call is ApiRpc.toLayer:

export const ApiRpcHandlers = ApiRpc.toLayer(
  Effect.gen(function* () {
    const todoRepo = yield* TodoRepo;
    const audioRepo = yield* AudioRepo;
    const videoRepo = yield* VideoRepo;
    // ...every other repo/service...

    return ApiRpc.of({
      health: () => Effect.succeed("ok"),
      ...todoHandlers(todoRepo, travelRepo),
      ...audioHandlers(audioRepo),
      ...videoHandlers(videoRepo, videoRenderer),
      // ...all the other slices...
    });
  })
).pipe(
  Layer.provide(PostHogQuery.layer),
  Layer.provide(SentryQuery.layer),
  Layer.provide(GitHubContents.layer)
);

• ApiRpc.toLayer(...) takes an effect that produces the full handler object (ApiRpc.of({...})) and yields a Layer implementing every procedure in the contract. If you add a procedure to schema.ts but forget its handler, ApiRpc.of won't type-check — the contract forces completeness.
• Dependencies are yielded once, at the top. Each repo (yield* AudioRepo) is resolved from context, then closed over by its handler slice (audioHandlers(audioRepo)). The handlers themselves never reach into Effect context for repos — they receive them as plain arguments, which keeps them trivially unit-testable with a fake repo.
• Spreading the slices (...audioHandlers(audioRepo)) merges each domain's { "audio/list": ..., "audio/get": ... } into one flat handler map keyed by the prefixed names.

Below that, ApiRpcLive provides the real implementations of those dependencies:

export const ApiRpcLive = ApiRpcHandlers.pipe(
  Layer.provide(TodoRepo.layer),
  Layer.provide(AudioRepo.layer),
  Layer.provide(VideoRepo.layer),
  // Default render seam: a "not configured" gate; the app overrides it.
  Layer.provide(VideoRenderer.layer)
);

This is standard Effect layer composition: ApiRpcHandlers requires AudioRepo et al.; ApiRpcLive provides them. A test can instead provide fake repo layers and exercise the exact same handlers.

5. How the middleware resolves a session

The SessionMiddleware referenced in schema.ts is defined in packages/api/src/rpc/middleware.ts. Its job is the boundary decode of the better-auth payload into the Session service:

export const resolveSession = (auth, headers) =>
  Effect.gen(function* () {
    const raw = yield* Effect.tryPromise({
      try: () => auth.getSession({ headers: webHeaders }),
      catch: (cause) => cause,
    }).pipe(
      retryOnStaleConnection,
      Effect.mapError(() => new Unauthorized({ message: "auth lookup failed" }))
    );
    if (!raw) return yield* new Unauthorized({ message: "no session" });
    // Boundary rule: validate the better-auth payload before it becomes the
    // Session service; unknown fields are stripped by the decode.
    const session = yield* Schema.decodeUnknownEffect(SessionSchema)(raw).pipe(
      Effect.mapError(() => new Unauthorized({ message: "malformed session" }))
    );
    return session;
  }).pipe(Effect.withSpan("rpc.auth", { kind: "server" }));

• The raw better-auth result is unknown until Schema.decodeUnknownEffect runs — that decode is the trust boundary. SessionUser (in session.ts) is an allowlist: it declares exactly which fields cross the wire (id, email, name, role, …), and decoding strips everything else — plugin fields, tokens, timestamps. Extending it is the single place to add a field the client legitimately needs.
• retryOnStaleConnection re-attempts only connection-class faults (the Supavisor pooler dropping a socket), so a transient pool blip doesn't collapse to a spurious Unauthorized.

6. The server seam: one Next.js route

All of the above becomes an HTTP endpoint in exactly one place — apps/web/src/app/api/rpc/route.ts. There is no per-procedure route; the whole ApiRpc is served from a single POST:

const ServerLayer = Layer.provideMerge(
  ApiRpcLive,
  Layer.mergeAll(
    SessionMiddlewareLive(auth.api),
    ServiceTokenMiddlewareLive,
    RpcSerialization.layerJson,
    Metric.enableRuntimeMetricsLayer,
    TranslationCacheUpstashLive
  )
);

export async function POST(request: Request) {
  const program = Effect.gen(function* () {
    const httpEffect = yield* RpcServer.toHttpEffect(ApiRpc).pipe(Effect.provide(ServerLayer));
    const response = yield* httpEffect;
    return HttpServerResponse.toWeb(response);
  }).pipe(
    Effect.provideService(HttpServerRequest.HttpServerRequest, HttpServerRequest.fromWeb(request)),
    Effect.scoped
    // ...envelope-level catchCause → 500...
  );
  return runDbLive(program);
}

• RpcServer.toHttpEffect(ApiRpc) builds the server handler from the same contract the client uses. ServerLayer supplies its dependencies: the live handlers (ApiRpcLive), the two middleware implementations (with the real auth.api wired into SessionMiddlewareLive), and JSON serialisation.
• runDbLive(program) runs the whole thing against the shared DB runtime. Because that runtime is read fresh each call, a pool rebuilt after a stale-connection fault takes effect on the next request — the page self-heals on reload.

7. The client: @repo/api + RpcClient.make(ApiRpc)

The web app builds a client runtime from the same ApiRpc. The runtime factory lives in packages/api/src/client/ — makeRpcAtomRuntime wires an HTTP protocol layer with JSON serialisation:

export const makeRpcAtomRuntime = (options: MakeRpcRuntimeOptions) =>
  runtime(buildRpcLayer(options));

const buildRpcLayer = (options) =>
  RpcClient.layerProtocolHttp({ url: options.url }).pipe(
    Layer.provide(httpClientLayer),
    Layer.provide(RpcSerialization.layerJson),
    Layer.provide(requestInitLayer)
  );

Then apps/web/src/utils/rpc.ts builds the actual runtime and derives atoms from it. This is the whole client-side story:

const rpcAtomRuntime = makeRpcAtomRuntime({
  url: "/api/rpc",
  requestInit: { credentials: "include" },
});

export const todosAtom = rpcAtomRuntime.factory.withReactivity(["todo/getAll"])(
  rpcAtomRuntime.atom(
    Effect.gen(function* () {
      const client = yield* RpcClient.make(ApiRpc);
      return yield* client["todo/getAll"]();
    }).pipe(Effect.scoped)
  )
);

• RpcClient.make(ApiRpc) produces a client whose methods are exactly the contract's procedures — client["todo/getAll"](), client["audio/get"]({ id }). The argument and return types come straight from the Rpc.make Schemas, so a wrong payload is a compile error in the UI. There is no generated client, no codegen step — the types flow from the one shared ApiRpc object.
• credentials: "include" sends the auth cookie, which is how SessionMiddleware finds a session on the server.
• withReactivity([...]) / reactivityKeys tie a query atom to the mutation atoms that should refresh it. A createTodoFn tagged with { reactivityKeys: ["todo/getAll"] } invalidates todosAtom on success — the UI re-fetches without any manual cache plumbing.

Putting it together: add a new RPC

The audio slice you just read is the template. To add a procedure end to end:

1. Schema — in the domain's *-schema.ts, define (or reuse) the payload and result Schema.Structs, deriving types with type X = typeof X.Type. Keep server-owned fields (like userId) out of the input schema.
2. Contract — in schema.ts, add a Rpc.make("yourProc", { payload, success, error: RpcError }) to the relevant RpcGroup.make(...) (or a new group with .prefix("yourdomain/")). If it's a new group, merge it into ApiRpc with the right .middleware(...).
3. Handler — in handlers/<domain>.ts, add "yourdomain/yourProc": (input) => adminHandler("yourdomain/yourProc")(Effect.gen(...)) (or rpcHandler for non-admin). Read Session, call the repo, fail with NotFound where a lookup misses.
4. Wire-up — in router.ts, make sure the handler slice is spread into ApiRpc.of({...}) and its repo dependency is both yield*-ed and provided via Layer.provide(YourRepo.layer) in ApiRpcLive.
5. Client — in the app's rpc.ts, add an atom/fn that calls client["yourdomain/yourProc"](...), with the right reactivityKeys.

You never touch the transport, the serialiser, the router plumbing, or the Next route. The contract change flows types to both ends, and the layer wiring forces you to provide what you declared.

See also

• How a request flows — one real call traced end to end: component → RPC client → route → auth → handler → repo → Postgres.
• Architecture — where the RPC boundary sits in the wider system.