@emeryld/rrroutes-server-data

Opinionated query, mutation, and keyset-feed services for RRRoutes resources, plus exposeRRRoutesResource — one declaration that mounts a service on an Express router and wires its writes to socket broadcasts.

The package is ORM-agnostic. Every service is written against a small driver interface you implement once per model, so the feed engine, cursor format, and service pipeline are reusable whether you run Prisma, Drizzle, Kysely, or raw SQL.

Installation

pnpm add @emeryld/rrroutes-server-data

Prerequisites

• zod ^4.0.0

@emeryld/rrroutes-contract is a dependency. @emeryld/rrroutes-server is an optional peer needed only by the ./expose entry point — the data services import without it.

Entry points

| Import path | Purpose | Runtime | Status | Additional requirements | | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- | ------- | ------ | ------------------------ | | @emeryld/rrroutes-server-data | ORM-agnostic query, mutation, and keyset-feed services written against a driver interface rather than an ORM. | node | stable | — | | @emeryld/rrroutes-server-data/expose | Mounts a built service onto an Express router and a socket connection in one declaration, deriving controllers from the service's own route keys. | node | stable | @emeryld/rrroutes-server |

Compatibility

Node only, ESM + CJS. Feeds require a keyset sorter whose last column is unique; a driver that cannot cheaply count still works and reports total: 0. The Prisma adapter deliberately lives in the consuming app rather than here.

Quick start

1. Bind the package to your context

Every service is curried on your application's Ctx, so it is stated once.

import { createRRRoutesData } from '@emeryld/rrroutes-server-data'
import { BadRequestError, NotFoundError } from '@/errors/ApiError'

export const data = createRRRoutesData<Ctx>({
  errors: {
    badRequest: (message) => new BadRequestError(message),
    notFound: (message) => new NotFoundError(message),
  },
  logger: (ctx) => ctx.logger,
  profiler: (ctx) => ctx.profiler,
  cursorViewer: (ctx) => ctx.bearer.userId,
  debug: process.env.NODE_ENV !== 'production',
})

Only errors is required. logger and profiler fall back to no-ops, so observability is opt-in and never changes behavior.

2. Implement a driver

One driver per model. The generic parameters carry your ORM's own types through untouched, so a Prisma adapter stays exactly as type-safe as hand-written Prisma calls.

import type { RRRoutesDataDriver } from '@emeryld/rrroutes-server-data'
import type { Prisma } from '@prisma'

export const postDriver: RRRoutesDataDriver<
  Ctx,
  Prisma.PostWhereInput,
  Prisma.PostSelect,
  Prisma.PostOrderByWithRelationInput[],
  PostRow,
  Prisma.PostCreateInput,
  Prisma.PostUpdateInput & { id: string },
  { id: string },
  PostRow
> = {
  findMany: (ctx, { where, select, orderBy, take, skip }) =>
    ctx.db.post.findMany({ where, select, orderBy, take, skip }),
  findFirst: (ctx, { where, select }) =>
    ctx.db.post.findFirst({ where, select }),
  count: (ctx, { where }) => ctx.db.post.count({ where }),
  aggregate: (ctx, args) => ctx.db.post.aggregate(args as never),
  deleteMany: async (ctx, { where }) => ({
    deleted: (await ctx.db.post.deleteMany({ where })).count,
  }),
  create: (ctx, input) => ctx.db.post.create({ data: input }),
  update: (ctx, { id, ...data }) => ctx.db.post.update({ where: { id }, data }),
  delete: (ctx, { id }) => ctx.db.post.delete({ where: { id } }),

  // The one piece of query-language syntax the feed engine needs.
  and: (clauses) => ({ AND: clauses }),
}

count, aggregate, and deleteMany are optional. A feed over a source that cannot cheaply count still works — it reports total: 0.

3. Build a feed

import { defineKeysetSorter } from '@emeryld/rrroutes-server-data'

const operators = {
  and: (clauses) => ({ AND: clauses }),
  or: (clauses) => ({ OR: clauses }),
  compare: (field, op, value) => ({
    [field]: op === 'eq' ? value : { [op]: value },
  }),
}

export const postFeed = data
  .createFeedService('posts', postDriver)
  .where<PostFeedQuery>((ctx, query) => ({ authorId: query.authorId }))
  .extract<PostRow>(() => ({ id: true, title: true, createdAt: true }))
  .sortBy({
    recent: defineKeysetSorter({
      // Most significant first. The last column MUST be unique.
      columns: [
        {
          field: 'createdAt',
          get: (row) => row.createdAt,
          deserialize: (v) => new Date(v),
        },
        { field: 'id', get: (row) => row.id },
      ],
      operators,
    }),
  })
  .specs({
    defaultSortBy: 'recent',
    defaultSortDir: 'desc',
    maxLimit: 50,
    anchorWhere: (id) => ({ id }),
  })

4. Build a service and expose it

const posts = data
  .createRRRoutesService({
    name: 'post',
    query: postQueryService,
    feeds: { recent: postFeed },
    repository: postDriver,
    idFromRecord: (record) => record.id,
  })
  .build({
    routes,
    keys: {
      create: 'POST /posts',
      get: 'GET /posts/:id',
      update: 'PATCH /posts/:id',
      delete: 'DELETE /posts/:id',
      feeds: { recent: 'GET /posts/feed' },
    },
    id: {
      fromParams: (args) => args.params.id,
      fromFeedItem: (item) => item.id,
    },
    process: {
      create: {
        preProcess: ({ ctx, body }) => ({ ...body, authorId: ctx.userId }),
      },
      update: { preProcess: ({ id, body }) => ({ id, ...body }) },
      delete: { preProcess: ({ id }) => ({ id }) },
      query: {
        preProcess: ({ ctx, queryService, id }) => queryService.byId(ctx, id),
      },
      feeds: {
        recent: {
          preProcess: ({ ctx, queryService, itemId }) =>
            queryService.byId(ctx, itemId),
        },
      },
    },
  })

exposeRRRoutesResource({
  service: posts,
  server, // from createRRRoute(router, ...)
  realtime: {
    connection, // from createSocketConnections(io, ...)
    on: {
      create: ({ id, out }) => ({
        eventName: 'post:created',
        payload: out,
        toRooms: [`post:${id}`, 'posts:all'],
      }),
    },
  },
})

Detailed usage

The write pipeline

Every write runs the same five stages:

guards -> preProcess -> repository -> canonical read -> events

The canonical read is the point of the design. After a create or update, the service re-reads the resource through the GET route's own query.preProcess rather than shaping the repository record directly. That is what makes a create response byte-identical to a later fetch of the same resource, so a client can seed its cache from a mutation result without drift.

delete inverts the order: it resolves the public payload before removing the row (nothing to read afterwards), then returns it with delete: true.

Hooks by stage:

| Hook | Runs | Typical use | | ------------ | ------------------------ | -------------------------------------------- | | guards | before anything else | authorization, precondition checks | | preProcess | after guards | map the request into repository input | | events | after the canonical read | outbound notifications, audit, cache busting |

Guards and events run sequentially in array order, awaiting each. A throwing guard aborts the operation.

id, and why idFromRecord sits on the seed

Identity extraction is split across two places on purpose:

• idFromRecord is on the seed, next to the repository.
• fromParams / fromFeedItem / fromPut are on build, next to the routes.

That split is not cosmetic. idFromRecord is the sole inference site for TId, and on the seed the record type is already fixed. Declared alongside the route keys it would resolve to unknown, because every other position depends on route types still being inferred in the same pass.

update uses the post-write identity (idFromRecord(record)) for the canonical read and every event, so a repository that relocates or re-keys a record still reports the right id.

Feed pagination

Three mutually exclusive modes per request:

| Mode | Request | Notes | | -------- | ---------------------------------- | ------------------------------- | | Forward | cursor + direction: 'next' | The default | | Backward | cursor + direction: 'previous' | Rows are returned in feed order | | Anchored | anchorItemId | Centers a page on one row |

Anchored pages balance the budget around the anchor and spend any slack from a short side on the other, so an anchor at the head or tail still returns a full page.

Combining anchorItemId with a cursor, or passing direction without a cursor, is rejected as a bad request.

Cursors

Cursors are base64url { v, feed, context, sortBy, dir, direction, value }. The context is a hash of the feed name, cursorVersion, the non-pagination query, the viewer (cursorViewer), and any cursorContext you add.

A cursor is therefore rejected when replayed against a different query, a different viewer, a different ordering, or a different feed. This is a correctness feature, not an inconvenience: a cursor is a position in a specific result set, and reusing it elsewhere silently skips or repeats rows.

Consequences worth knowing:

• Renaming a feed invalidates every outstanding cursor. The name is in the payload.
• Changing what cursorViewer returns invalidates that viewer's cursors.
• Bump cursorVersion deliberately when you change what where produces for the same query.

The v2 payload shape is a wire format. Do not reorder or rename its fields.

Sorters

defineKeysetSorter generates the lexicographic keyset filter for you:

(c1 > C1) OR (c1 = C1 AND c2 > C2) OR (c1 = C1 AND c2 = C2 AND c3 > C3)

Writing that by hand is where feed bugs live — one wrong clause silently drops or duplicates rows across a page boundary.

The final column must be unique across the feed. Without a unique tiebreaker, rows sharing every ordering value can be skipped or repeated.

defineOffsetSorter covers orderings that cannot express a keyset boundary (relevance ranking, seeded shuffles). Feeds using it reject anchored and previous-page requests, because neither can be emulated with an offset.

Access gates

createQueryService wraps a bag of read functions so an access gate runs before each call:

const postQueryService = data.createQueryService(
  'post',
  {
    byId: (ctx, id) => ctx.db.post.findUniqueOrThrow({ where: { id } }),
  },
  async ({ ctx, id }) => ctx.can('read', 'post', id),
)

The gate's id is read from the call's second argument — either the value itself or its .id. Calls with no discoverable id pass through ungated; those are list/aggregate reads whose own where clause is expected to scope results.

service.gate({ ctx, id }) runs it explicitly from a guard or a custom read.

exposeRRRoutesResource

Generates a controller for every route key the service declares, then registers them. It handles three things you would otherwise get wrong by hand:

1. Response envelope. Services return the unwrapped payload; RRRoutes leaves declare { out, meta }. Generated controllers re-wrap.
2. Registration order. Express matches in registration order, so GET /posts/:id registered before GET /posts/feed swallows the feed route. Controllers are sorted by specificity — fewer path parameters first — regardless of the order your keys are declared in.
3. Broadcast isolation. A socket emit that throws is routed to onError, never to the HTTP response. The write already succeeded and the client is owed its answer.

Feed routes map onto the envelope as { out: items, meta: { …pagination } }, which is where createCursorPagination on the client looks for nextCursor and previousCursor. Feed routes must therefore declare an outputMetaSchema — the contract's default meta is an optional string and will reject an object.

resource('/posts').sub(
  resource('/feed')
    .get({
      outputSchema: z.array(PostOut),
      outputMetaSchema: z.object({
        total: z.number(),
        hasNext: z.boolean(),
        hasPrevious: z.boolean(),
        nextCursor: z.string().optional(),
        previousCursor: z.string().optional(),
        anchorItemId: z.string().optional(),
      }),
      feed: true,
    })
    .done(),
)

Override with feedResponse if your clients expect a different shape.

Rooms mirror the client

realtime.on returns the rooms a write broadcasts to. Those names must match what the client's toRooms mapper subscribes to in @emeryld/rrroutes-client's socketed-route helper. Keeping the two readable side by side is the point — a resource's server rooms and client rooms are the same vocabulary.

put (upsert)

Declaring a put key adds a service.put that dispatches by payload:

| Payload | Dispatches to | | ---------------------- | ------------- | | { delete: true, id } | delete | | { id, ... } | update | | no id | create |

delete: true without an id is a bad request. Results carry a delete boolean so a client can branch without a second lookup.

Testing

createMemoryDriver is exported for testing your services without a database:

import { createMemoryDriver } from '@emeryld/rrroutes-server-data'

const driver = createMemoryDriver<Ctx, PostRow>({ rows: seed })

Its where-language is {AND}, {OR}, {field, op, value}, or {} for match-all. It throws on any other clause rather than matching everything — a silent widen is how a mis-shaped anchorWhere hides.

Edge cases and notes

• The default anchorWhere emits { id: anchorItemId }. Supply your own for composite keys, a non-id primary key, or any driver whose where-language does not accept that shape.
• fromFeedItem is required only when the resource declares feeds. Omitting it on a feed-bearing resource throws a clear error at call time.
• Feeds request limit + 1 rows to detect further pages, so a driver's take must be honored exactly.
• hasPrevious/hasNext on the side you did not page toward is inferred from the presence of a cursor, not from a probe query.
• includeCallerStackInError stitches the issuing frame onto driver errors, so an async failure names the feed that caused it.

Full-stack guide

This package's chapters of the full-stack guide — the data layer behind the API set up in @emeryld/rrroutes-server.

server-data is written against a small driver interface rather than an ORM, so the feed engine, the cursor format and the write pipeline are reusable whether the storage underneath is Prisma, Drizzle, Kysely or SQL. and is the only piece of query syntax the engine needs to know: it is how a caller's filter is intersected with a keyset boundary.

import type {
  RRRoutesReadDriver,
  RepositoryConfig,
} from '@emeryld/rrroutes-server-data'
import type { Prisma, Post } from '@prisma/client'

import { prisma } from '../db'
import type { ApiCtx } from '../server'

export const postReads: RRRoutesReadDriver<
  ApiCtx,
  Prisma.PostWhereInput,
  Prisma.PostSelect,
  Prisma.PostOrderByWithRelationInput[],
  Post
> = {
  findMany: (_ctx, { where, select, orderBy, take, skip }) =>
    prisma.post.findMany({ where, select, orderBy, take, skip }),
  findFirst: (_ctx, { where, select }) =>
    prisma.post.findFirst({ where, select }),
  count: (_ctx, { where }) => prisma.post.count({ where }),
  and: (clauses) => ({ AND: clauses }),
}

export const postWrites: RepositoryConfig<
  ApiCtx,
  Prisma.PostCreateInput,
  { id: string; data: Prisma.PostUpdateInput },
  { id: string },
  Post
> = {
  create: (_ctx, input) => prisma.post.create({ data: input }),
  update: (_ctx, { id, data }) => prisma.post.update({ where: { id }, data }),
  delete: (_ctx, { id }) => prisma.post.delete({ where: { id } }),
}

Note — Reads and writes are separate on purpose A resource can be read-only, and the mutation service is generic over its own input types rather than over the read model's.

Reference — @emeryld/rrroutes-server-data

createRRRoutesData fixes the context type and the error constructors once. On top of it, a feed service is declared as a chain — filter, select, order, page — and the resource service pairs that with the write pipeline: guards, preProcess, and events, per operation. build is where the route keys arrive and everything becomes typed against the contract.

import {
  createRRRoutesData,
  defineKeysetSorter,
} from '@emeryld/rrroutes-server-data'

import { registry } from '@app/shared/contract/posts.routes'
import type { ApiCtx } from '../server'
import { postReads, postWrites } from './posts.driver'

export const data = createRRRoutesData<ApiCtx>({
  errors: {
    badRequest: (message) => new BadRequestError(message),
    notFound: (message) => new NotFoundError(message),
  },
  logger,
  profiler,
})

/** Newest first, tie-broken by id so page boundaries cannot skip a row. */
const recentSorter = defineKeysetSorter({
  columns: [
    { field: 'createdAt', get: (row) => row.createdAt },
    { field: 'id', get: (row) => row.id },
  ],
  operators: {
    and: (clauses) => ({ AND: clauses }),
    or: (clauses) => ({ OR: clauses }),
    compare: (field, op, value) => ({ [field]: { [op]: value } }),
  },
})

const recentFeed = data
  .createFeedService('posts.recent', postReads)
  .where((ctx, query) => ({
    deletedAt: null,
    ...(query.tag ? { tags: { has: query.tag } } : {}),
    OR: [{ visibility: 'public' }, { authorId: ctx.userId }],
  }))
  .extract(() => ({
    id: true,
    authorId: true,
    title: true,
    body: true,
    createdAt: true,
  }))
  .sortBy({ recent: recentSorter })
  .specs({ maxLimit: 50, defaultSortBy: 'recent', defaultSortDir: 'desc' })

export const postsService = data
  .createRRRoutesService({
    name: 'post',
    query: {
      byId: async (ctx: ApiCtx, id: string) => {
        const post = await postReads.findFirst(ctx, { where: { id } })
        if (!post) throw new NotFoundError(`No post ${id}`)
        return post
      },
    },
    feeds: { recent: recentFeed },
    repository: postWrites,
    idFromRecord: (post) => post.id,
  })
  .build({
    routes: registry.byKey,
    keys: {
      create: 'POST /v1/posts',
      get: 'GET /v1/posts/:postId',
      update: 'PATCH /v1/posts/:postId',
      delete: 'DELETE /v1/posts/:postId',
      feeds: { recent: 'GET /v1/posts/feed' },
    },
    accessGate: async ({ ctx, id, operation }) => {
      if (operation === 'get') return true
      return (await ownerOf(id)) === ctx.userId
    },
    id: {
      fromParams: ({ params }) => params.postId,
      fromFeedItem: (post) => post.id,
    },
    process: {
      create: {
        guards: [({ ctx }) => assertNotSuspended(ctx.userId)],
        preProcess: ({ ctx, body }) => ({ ...body, authorId: ctx.userId }),
        events: [({ out }) => search.index(out)],
      },
      update: {
        preProcess: ({ id, body }) => ({ id, data: body }),
      },
      delete: {
        preProcess: ({ id }) => ({ id }),
      },
      feeds: {
        recent: {
          preProcess: ({ ctx, queryService, itemId }) =>
            queryService.byId(ctx, itemId),
          // One unreadable row drops out of the page instead of failing it.
          onItemError: ({ feed, error }) =>
            logger.warn('feed item dropped', { feed, error }),
        },
      },
    },
  })

Note — Why the last keyset column must be unique Rows sharing every ordering value can be skipped or repeated at a page boundary. The final column — a primary key, normally — is what makes the cursor total.

Note — A partial page beats an empty screen By default a feed page is all-or-nothing: any item failing to hydrate fails the request. For feeds that mix in rows the viewer may have partly lost access to, onItemError drops the item instead. total and the cursors are deliberately left untouched, because they describe the underlying query the next page must be fetched against.

Note — Ordering that has no keyset Relevance ranking and seeded shuffles cannot express a keyset boundary. defineOffsetSorter compiles those to offset pagination, and the cursor format absorbs the difference.

Reference — @emeryld/rrroutes-server-data

The last connection in the loop. Controllers are derived from the service's own route keys — so adding a feed or renaming a key cannot leave a stale controller behind — and the realtime hooks turn each write into the event the clients in groups 3 and 4 are already reducing into their caches.

import { exposeRRRoutesResource } from '@emeryld/rrroutes-server-data/expose'

import { server } from '../server'
import { connection } from '../sockets'
import { postsService } from './posts.service'

exposeRRRoutesResource({
  service: postsService,
  server,
  realtime: {
    connection,
    on: {
      create: ({ out }) => ({
        eventName: 'post:created',
        payload: out,
        toRooms: ['posts:all'],
      }),
      update: ({ id, out }) => ({
        eventName: 'post:updated',
        payload: out,
        toRooms: [`post:${id}`, 'posts:all'],
      }),
      delete: ({ id }) => ({
        eventName: 'post:deleted',
        payload: { id, socketDelete: true },
        toRooms: [`post:${id}`, 'posts:all'],
      }),
    },
    // A failed broadcast must never fail the HTTP response that caused it.
    onError: ({ operation, error }) =>
      logger.error('broadcast failed', { operation, error }),
  },
})

import { exposeRRRoutesResource } from '@emeryld/rrroutes-server-data/expose'

/**
 * Deriving controllers from route keys mounts *every* declared key. A resource
 * that declares full CRUD but is only meant to be readable names the closed
 * roles instead of quietly omitting a controller.
 */
exposeRRRoutesResource({
  service: activitiesService,
  server,
  disable: ['create', 'update', 'delete'],
})

Note — Rooms match, by construction The room strings here are the same ones toRooms derives on the client in step 2.3. That correspondence is the one thing in this guide that no type checks — which is why both sides derive them from the record's id rather than writing them at call sites.

Note — Route ordering is handled Express matches in registration order, so GET /v1/posts/:postId registered first would swallow GET /v1/posts/feed. Static paths are sorted ahead of parameterised siblings regardless of the order the keys were declared in.

Note — Disabled is not absent A disabled role stays in the contract and appears in the inspector marked as such, but mounts no handler — requests fall through to a 404. A role that matches nothing throws, because the failure this guards against is a write meant to be closed being open.

Reference — @emeryld/rrroutes-server-data/expose

Scripts

pnpm --filter @emeryld/rrroutes-server-data test
pnpm --filter @emeryld/rrroutes-server-data typecheck
pnpm --filter @emeryld/rrroutes-server-data build