@geekmidas/constructs

A comprehensive framework for building type-safe HTTP endpoints, cloud functions, scheduled tasks, and event subscribers with AWS Lambda and Hono support.

Installation

pnpm add @geekmidas/constructs

Features

• ✅ Declared infrastructure — a database, bucket, cache, or credential is one statement
• ✅ .dependsOn() — one edge derives the environment, the client, and the cloud access
• ✅ Fluent endpoint builder pattern
• ✅ Full TypeScript type inference
• ✅ StandardSchema validation (Zod, Valibot, etc.)
• ✅ AWS Lambda adapter support (API Gateway v1/v2)
• ✅ Hono framework adapter
• ✅ Service dependency injection
• ✅ Built-in error handling
• ✅ Session and authorization management
• ✅ Structured logging
• ✅ Rate limiting
• ✅ Audit logging
• ✅ Row Level Security (RLS) with PostgreSQL
• ✅ Response handling (cookies, headers, status codes)
• ✅ Event publishing
• ✅ Event subscribers (topic fan-out) and queue workers (point-to-point)
• ✅ Testing utilities

Package Exports

Export	Description
/	Core types and utilities
/endpoints	EndpointFactory (what api.get() and api.database() return) and types
/worker	Worker — the process with no port; builds crons, subscribers, queues and functions
/topic	Topic — a declared event contract; its publisher is its service
/functions	The Function construct (built with worker.input(…) / worker.dependsOn(…))
/crons	The Cron construct (built with worker.cron(…)) and AWSScheduledFunction
/subscribers	The Subscriber construct (built with worker.topic(…)) — topic fan-out
/queue	The Queue construct (built with worker.queue(…)) — one queue, one consumer
/types	Type definitions
/hono	Hono framework adapter (HonoEndpoint)
/aws	AWS Lambda adaptors (API Gateway v1/v2, AWSLambdaFunction, AWSLambdaSubscriber, AWSLambdaQueue, AWSScheduledFunction)
/testing	Testing utilities (TestEndpointAdaptor, TestFunctionAdaptor, TestSubscriberAdaptor, TestQueueAdaptor)
/construct	The Construct interface, Consumable, and Declaration types
/database/kysely	KyselyDatabase — a declared Postgres database, typed by its schema
/object-storage	ObjectStorage — a declared bucket
/file-server	FileServer — a domain that serves a bucket's objects
/cache	Cache — a declared cache
/credential	Credential — a third-party credential with a shape
/encryption	Encryption — a key that encrypts what the application stores
/external-api	ExternalApi — an HTTP API somebody else runs, faked in tests
/email	Email — declared outbound mail
/rest-api	RestApi — an API surface
/websocket-api	WebSocketApi — a surface that holds long-lived connections
/site	StaticSite — a declared static site
/auth	BetterAuth — a declared auth server
/telemetry	Telemetry — what the application emits (traces and logs)

Service integrations moved: the tRPC and Middy middlewares now live in @geekmidas/services (@geekmidas/services/trpc, @geekmidas/services/middy) since they depend only on @geekmidas/services, not on the constructs.

Constructs

A construct is one declaration that stands for a piece of infrastructure. It is simultaneously an infrastructure requirement at build time, a runtime capability, and something other code can consume — which is why the interface has exactly three members:

interface Construct<TName extends string = string, TClient = never> {
  readonly id: TName;
  declare(): Declaration[];
  readonly service: [TClient] extends [never] ? never : Service<TName, TClient>;
}

From that one declaration, three different owners derive three different things:

Derived	By whom	When
The function's environment	the framework	gkm build
Its runtime client	the framework	first request
The infrastructure — a container locally, a cloud resource deployed	the target adapter	gkm dev / gkm deploy

The resource constructs

import { KyselyDatabase } from '@geekmidas/constructs/database/kysely';
import { ObjectStorage } from '@geekmidas/constructs/object-storage';
import { Cache } from '@geekmidas/constructs/cache';
import { Credential } from '@geekmidas/constructs/credential';

export const database = new KyselyDatabase<Database, 'Orders'>('Orders');
export const uploads = new ObjectStorage('Uploads', { versioned: true });
export const cache = new Cache('Sessions');
export const stripe = new Credential('Stripe', {
  schema: z.object({ secretKey: z.string(), webhookSecret: z.string() }),
});

Each canonicalises its id — uploads and Uploads are one construct, not two that collide — and derives its env key from it: UPLOADS_CONNECTION_STRING, SESSIONS_URL, ORDERS_URL (plus ORDERS_OWNER_URL for the DDL role). The key the target publishes and the key the client reads come from the same field, so they cannot drift.

Construct	Import	The client it hands back
KyselyDatabase	/database/kysely	Kysely<DB>, typed by your schema
ObjectStorage	/object-storage	StorageClient — write, delete, presign
FileServer	/file-server	a StorageClient superset — plus url() and signedUrl()
Cache	/cache	CacheClient
Credential	/credential	the parsed, validated value — no await at the call site
RandomPassword	/random-password	the password, a string — generated once per stage
Encryption	/encryption	a Cipher — encrypt, decrypt, index, reencrypt
ExternalApi	/external-api	whatever its client builds — the real API, or its fake in tests and gkm dev --fake
Email	/email	an EmailClient typed by your templates
Topic	/topic	an EventPublisher typed to its events
Queue	worker.queue(…)	an EventPublisher typed to its message
RestApi	/rest-api	its public URL, a string — API_URL
StaticSite	/site	its public URL, a string — WEB_URL
BetterAuth	/auth	an AuthClient — api.getSession({ headers }), over HTTP to the auth app

A FileServer serves the paths it calls open without a signature, and everything else only through signedUrl(). Paths whose keys are written once can also be immutable, so the edge tells every client to keep them forever:

import { FileServer } from '@geekmidas/constructs/file-server';

export const assets = new FileServer('Assets', {
  open: ['brand/**', 'avatars/*'],
  immutable: ['brand/logos/*'], // a subset of `open`, or `ImmutableNotOpen`
});

A file server answers on the name it was given: assets.<domain> here, and its address is ASSETS_URL (its bucket's connection is ASSETS_CONNECTION_STRING). An app or site that would answer on the same name is refused with FileServerHostTaken. On a stage whose buckets the s3 provider creates, the file server answers on <fileserver>.<domain> through CloudFront, with that header on the immutable paths — see Providers.

A site's or a mobile app's plain .dependsOn([assets]) is to the file server's public face: the bundle gets VITE_ASSETS_URL (NEXT_PUBLIC_, EXPO_PUBLIC_) and nothing that writes to the bucket. Its assets client serves (url(), openUrl(), processedUrl()), and a bucket method throws FileServerReadOnly. A full-stack site whose server half uploads or presigns depends on assets.server instead, which is the bucket, its credentials and ASSETS_URL. A backend's plain edge is already that. See Edges and Environment.

An ObjectStorage splits the same way. A site's or a mobile app's plain .dependsOn([documents]) gets where the bucket signs its links, DOCUMENTS_SIGNED_ORIGIN (inlined as <PREFIX>DOCUMENTS_SIGNED_ORIGIN), and no connection or credentials: its documents client's methods throw ObjectStorageReadOnly. A server half that uploads or presigns depends on documents.server, the connection and the AWS key pair as well. A backend's plain edge is already the full one, and a backend cannot depend on .server.

Next images. A Next site builds images.remotePatterns from the workspace rather than a hand-kept list:

// apps/web/next.config.ts
import { imageRemotePatterns } from '@geekmidas/cli/next';

export default {
  images: { remotePatterns: [...imageRemotePatterns() /* , third parties */] },
};

It returns one pattern for each file server the site has an edge to, and one for the host each bucket it reaches signs its links on, for the stage being built. Under gkm dev those are the local file server host and MinIO's address. Without gkm's environment (a plain next build) it returns [].

What it reads is not Next-specific:

• <ID>_SIGNED_ORIGIN: where a bucket's presigned links point (https://<bucket>.s3.<region>.amazonaws.com, an S3-compatible endpoint's host, MinIO locally). It is derived from the bucket's connection, carries no credentials, and is inlined into a site or app with an edge to the bucket or to a file server over it. s3Url.signedOrigin() in @geekmidas/storage computes it.
• <PREFIX>GKM_IMAGE_ORIGINS: the site's file-server and signed-link origins as one comma-separated list, passed by gkm dev, compose site builds (an ARG in the site's Dockerfile) and the AWS target. A Vite or Expo site can feed it to a CSP.

On a deployed stage, both come from the bucket's stored connection. A compose stack's MinIO is the origin only for the buckets it serves; a bucket on a provider (deploy.objects.<stage>: { provider: 's3' }) signs on S3. A site image built for a stage that has not resolved them yet (a CI build ahead of the stage's first deploy) fails with SiteBuildValueUnresolved, naming the key, rather than ship without its own asset host. Deploy the stage first.

A file server's own URL is derived wherever the stage's domain decides it: over the stack's MinIO, or behind the CDN the s3 provider puts in front of a bucket, it is https://<label>.<domain> (assets.<domain> for new FileServer('Assets', …)). A site built right after domains.<stage> changes inlines the new domain's host, even though the stored copy is rewritten only by the deploy that follows the build. A URL set by hand (an 'external' bucket) is kept, and gkm compose warns when it differs from the one the stage's domain gives. On S3 with cdn: false it is the bucket's own endpoint, which only the deploy knows.

A file server can also make derivatives of its open objects on read — a thumbnail at /i/256/avatars/7.png — with assets.process('/i/:width/*'). See File Server Processors.

A Credential's value is the stage's <ID>_CREDENTIALS — STRIPE_CREDENTIALS here — set with gkm secrets:add. On the test stage, and under gkm dev --fake, it can come from a fake beside the external APIs' instead, which no deployed stage reads:

// test/fakes/stripe.ts
import { fake } from '@geekmidas/constructs/credential';
import type { stripe } from '../../constructs/stripe';

export default fake.credential<typeof stripe>({
  secretKey: 'sk_test_fake',
  webhookSecret: 'whsec_fake',
});

A test stage set up fresh stores no credentials, so gkm test refuses a credential with neither a fake nor a stored value before any test runs (CredentialHasNoTestValue, naming the key and the file to create). See Testing — a credential's test value.

A RandomPassword is the other way round: a secret nobody issues, so nobody sets it either.

import { RandomPassword } from '@geekmidas/constructs/random-password';

export const adminPassword = new RandomPassword('AdminPassword');

Its key is its name — ADMIN_PASSWORD — and a handler that depends on it gets the value as a string. The first deploy of a stage generates it and keeps it, a server stage in its secrets and an AWS stage in its state, so every later deploy reads the same value back. One set by hand (gkm secrets:set, sst secret set) is kept rather than replaced. gkm dev and gkm test derive it from the project, the stage and the id, so it needs no fake.

It is 43 letters, digits, - and _ unless you say otherwise — for a system with password rules of its own:

export const routerPassword = new RandomPassword('RouterPassword', {
  length: 24,
  symbols: '!@#$%', // drawn beside letters and digits; '' for none
});

When there are symbols, at least one is always in the password. Every stage draws it the same way, so a password that passes the rules in gkm dev passes them deployed. A deployed stage keeps the password it generated even if the format changes later.

WebSocketApi — long-lived connections

A surface like RestApi, in a container of its own, whose routes are dispatched on a frame's action rather than a method and path. Frames are API Gateway's shape — { "action": "chat.say", "id": 1, "data": … } — and the connection's lifecycle runs connect() and disconnect():

// constructs/realtime.ts
import { WebSocketApi } from '@geekmidas/constructs/websocket-api';

export const realtime = new WebSocketApi('Realtime', {
  path: 'apps/realtime',
  replicas: 2, // processes on one machine; see "WebSocket surfaces" in the compose guide
  messages: {
    // what it sends its sockets, by action
    announcement: z.object({ text: z.string() }),
    'chat.said': z.object({ from: z.string(), text: z.string() }),
  },
  logger,
}).auth(auth);

// apps/realtime/routes/chat.ts
export const connect = realtime.connect().handle(({ session, query }) => {
  if (!session) throw new UnauthorizedError(); // refuses the upgrade with 401
  return { userId: session.user.id };          // the connection's `data` from here on
});

export const say = realtime
  .route('chat.say')
  .body(z.object({ text: z.string() }))
  .handle(async ({ body, connection, sockets }) => {
    await sockets.broadcast(
      'chat.said',
      { from: String(connection.data.userId), text: body.text },
      { except: [connection.id] },
    );
    return { sent: true }; // sent back as { action: 'chat.say', id, data }
  });

• A reply is whatever the handler returns, echoed with the frame's id. A frame that is not JSON, names no route, or fails its schema is answered with { action: '$error', id, data: { message, issues? } } and the socket stays open. $default answers any action nothing else does.
• What it sends is declared. A message goes out as { action, data }, the shape frames arrive in. messages names each action with the schema its data must pass, and types the publisher by it: broadcast('anouncement', …) or data of the wrong shape doesn't compile. The schema checks the data before anything is sent too, so a bad message throws in the sender (UnknownWebSocketMessage, InvalidWebSocketMessage) rather than reaching every client. A surface without messages sends any action and any data, unchecked.
• Anything else reaches a socket through the surface's publisher..dependsOn([realtime]) on an endpoint, a cron or a subscriber puts services.realtime in its handler: send(ids, action, data), broadcast(action, data, { except }) and close(ids, code, reason). A connection id names the process that holds it, so a send goes to that process alone; between processes the publisher is Redis pub/sub (REALTIME_PUBLISHER_CONNECTION_STRING), which gkm dev and gkm compose run for you.
• Browsers are checked against the graph. An upgrade carrying an Origin outside REALTIME_TRUSTED_ORIGINS — read off what declares an edge to the surface, as a RestApi's CORS origins are — is refused with 403. A client with no Origin (a server, a load generator) is not a browser. The origin of the surface's own address, REALTIME_URL, is always trusted too — both https://realtime.example.com and wss://realtime.example.com for wss://realtime.example.com — because React Native's WebSocket sends the socket's own origin as its Origin, so a native app connects without being listed.
• Liveness is the server's: a ping every heartbeat seconds (30), and a socket that missed the last one is dropped. One that stops reading while 4 MB is queued for it is closed with 1013. On SIGTERM every socket is closed with 1012 — reconnect, and land on a process that is still up.
• Limits bound what one client may ask of it — see below.
• A site that declares an edge to the surface gets its URL alone, never the publisher: whoever holds that string can write to every socket.

Limits — limits

Every upgrade runs the connect handler, usually a session lookup, and every frame is parsed, validated and answered. Without a bound, one script can open handshakes as fast as the edge accepts them, one session can hold any number of sockets, and one socket can send as fast as it likes. A surface limits all three by default:

Limit	Per	Default	Past it
handshakesPerMinute	client address	120	the upgrade is answered 429 with Retry-After, before the connect handler runs
inFlightHandshakes	client address	16	429 — handshakes still waiting on the connect handler
connectionsPerIp	client address	256	429
messagesPerSecond + burst	socket	50, burst 100	the frame is dropped and the socket is sent one $error; if it keeps sending before its budget refills, it is closed with 1008
maxConnections (with maxConnectionsPer)	identity	16	overflow: 'refuse': 429; 'closeOldest': the new socket is let in and the identity's oldest are closed with 1008

export const realtime = new WebSocketApi('Realtime', {
  path: 'apps/realtime',
  limits: {
    messagesPerSecond: 20,
    burst: 40,
    // who a connection is: read from what connect() returned
    maxConnectionsPer: (data) => data.userId as string,
    maxConnections: 5,
    overflow: 'closeOldest', // a new tab wins; 'refuse' (default) turns it away
    connectionsPerIp: false, // lift one: every client behind one NAT
  },
  requireOrigin: true, // 403 for an upgrade with no Origin
}).auth(auth);

• Across replicas. The counts per address and per identity live where the publisher does: on Redis (REALTIME_PUBLISHER_CONNECTION_STRING) every replica counts in one place, so connectionsPerIp: 10 is ten across the surface, not ten per process. A socket's place is renewed with each heartbeat and lapses three heartbeats after a process that died without giving it back. With closeOldest, the oldest socket is closed on whichever replica holds it. If Redis can't be reached, the limit is logged and not applied rather than refusing everyone.
• The client address is the TCP peer's — or, from a private peer (the stack's edge), the last X-Forwarded-For entry, the one the edge wrote. Traefik and Caddy both replace what an outside client sends there, and a client that reaches the port directly can't pick its own bucket.
• The edge enforces the per-address limits too. The surface declares handshakesPerMinute and connectionsPerIp, and the shared Traefik renders them as rateLimit and inFlightReq middlewares on its route, so a flood is answered 429 before it reaches a replica. Caddy ships no rate limiter; behind it, the surface's own count is what holds.
• No Origin. Browsers and native clients always send one, so an upgrade without it is a server or a script. requireOrigin: true refuses those with 403. It's off by default because a backend may hold a socket too; the limits apply either way.
• A surface that only pushes, whose clients write over HTTP, declares no routes and sets a small maxPayload (say 1024): every inbound frame is then refused before it's parsed.

Flat frames — frame: { discriminator }

Clients that already parse another shape — a released mobile build can't be changed in place — usually route on a flat key rather than { action, data }. Name that key and the surface speaks it both ways:

export const realtime = new WebSocketApi('Realtime', {
  path: 'apps/realtime',
  frame: { discriminator: 'type' }, // absent: { action, data }
  messages: {
    'chat.message.created': z.object({
      chatId: z.string(),
      message: ChatMessage,
    }),
    'scan.result': z.object({ ok: z.boolean() }),
  },
});

// in a handler that .dependsOn([realtime]), or a route's `sockets`
await services.realtime.send(ids, 'chat.message.created', { chatId, message });
// wire: {"type":"chat.message.created","chatId":"…","message":{…}}

• Outbound is { [discriminator]: action, ...data }, with data checked by the action's schema first, as in the default shape. send(ids, action, data) and broadcast(action, data) are typed the same way in both.
• Inbound frames are routed on the same key, and every other field is the route's body: { "type": "chat.say", "text": "hi" } reaches route('chat.say') with body { text: 'hi' }. A reply is flat too — { "type": "chat.say", ...returned } — so a handler returns an object.
• $error uses the same shape: { "type": "$error", "message", "issues"? }.
• A flat frame carries no id. Every field but the discriminator is data, so a reply is paired by its action, not an echoed id.
• Refused when the surface is built. A flat frame needs object data, so new WebSocketApi(…) throws WebSocketMessageNotAnObject for a message whose schema isn't an object (a union of objects is fine), and WebSocketMessageDeclaresDiscriminator for one that has a field named the discriminator itself. A surface without messages checks at send time instead: WebSocketDataNotAnObject, WebSocketDataHasDiscriminator.
• realtime.frame is the shape ({ discriminator: 'type' }, or {}), for a client of the surface to decode by.

One core, served by a runtime

A surface is defined once and served by an adaptor, as an endpoint is by Hono, API Gateway or endpoint.request(). realtime.core({ envParser, sink, modules }) is the part with no socket in it — a WebSocketCore that, given a connection id, runs the handlers:

• connect({ connectionId, headers, query, session? }) → accepted with the connection's data, or refused with the status of the HttpError the handler threw (500 for anything else). Without session, the surface's .auth() verifies the headers.
• message(connection, raw, write) decodes by frame, checks the route's body, runs the handler and writes the reply or $error with write.
• disconnect(connection, { code, reason }).

Each takes { services } to hand in what it would otherwise register. The sink is where the handlers' sockets send: an object with send(ids, payload), broadcast(payload, except?) and close(ids, code, reason), given frames already encoded and checked against messages.

realtime.server() is the Node adaptor over it. It owns what only a real socket has: the upgrade, the Origin check, connection ids, the heartbeat and backpressure. Its sink is a TransportSink: its own sockets directly, the rest over Redis.

Two more adaptors are for tests. route.request() calls the core directly, and surface.testPublisher() is a publisher over a recording sink (see Testing WebSocket routes). createMswWebSocketHandlers hands MSW-intercepted sockets to the core (see WebSocket clients). surface.publisher(sink) is the surface's typed publisher over a sink of an adaptor's own.

On AWS: API Gateway

The same surface deploys to an SST stage unchanged. gkm build --provider aws writes one Lambda per route under .gkm/aws/websocket/, each serving only its own route through AmazonApiGatewayWebSocket (from @geekmidas/constructs/aws-websocket). It also writes $connect and $disconnect, even when the surface declares no handler for them. fromManifest in @geekmidas/cloud then creates an API Gateway WebSocket API and mounts each Lambda on its route key. API Gateway picks the route from $request.body.action, or from $request.body.<discriminator> when the surface sets frame.discriminator, so a frame reaches the route message() would send it to on Node.

	compose (Node)	AWS (API Gateway)
who holds the socket	the surface's process	API Gateway
$connect	the upgrade: Origin check, then connect()	a Lambda: Origin check, then connect(). A refusal answers with its status
connection.data	in memory, beside the socket	a DynamoDB connection table the surface owns
send / close	over Redis, to the process holding the socket	postToConnection / deleteConnection
broadcast	every process	every id in the connection table
<ID>_PUBLISHER_CONNECTION_STRING	redis://…	apigateway://<region>?endpoint=…&table=…

services.realtime works from any Lambda that declared the edge (.dependsOn([realtime])). The link hands that Lambda the apigateway:// string, execute-api:ManageConnections on the API, and access to the table. A connection that has gone (410) is dropped from the table without an error, the same way the Redis publisher drops a socket no process holds.

The connection table is a DynamoDB table that @geekmidas/cloud creates next to the API. It is keyed by connection id, and rows expire through a TTL a little after API Gateway's two-hour limit on a connection. It isn't the Cache construct, because a cache has no way to list its keys and a broadcast needs every id. A surface on AWS declares nothing extra for it.

This target's limits come from API Gateway, not from the surface's config:

• Connection lifetime: at most 2 hours, and 10 minutes idle. API Gateway sends its own pings, so heartbeat doesn't apply. A client reconnects as it would after a restart.
• Frame size: 128 KB per message, sent in frames of up to 32 KB. This replaces maxPayload.
• Close codes: close(ids, code, reason) ends the connections, but API Gateway closes with a code of its own, so the client doesn't see code or reason.
• Sending from $connect: a $connect handler can't send to the connection it is accepting, because the connection doesn't exist until the handler returns. Send from the first route the client calls instead.

.dependsOn() — the one primitive

Every builder takes it: endpoints, functions, crons, queue workers, and subscribers.

export const createAvatar = api
  .post('/avatars')
  .dependsOn([uploads, stripe])
  .handle(async ({ services }) => {
    // Both exist, and both type, because of the edge above.
    services.uploads.getUploadURL({ path, contentType, contentLength });
    services.stripe.secretKey;
  });

The edge records what a handler consumes. The framework derives its environment and its runtime binding; the deploy target separately derives cloud access from the same edge. Permissions are not a framework concept and are not in the manifest — what an edge implies on a given cloud is the adapter's business.

.dependsOn() takes constructs only. A Service ({ serviceName, register }) does not match the shape, which is what keeps environment sniffing confined to .services().

Code that builds a link — an email's sign-in link, a payment provider's callback — depends on the site or the API it points at, rather than reading a key nothing declares:

export const digest = worker
  .cron('rate(1 day)')
  .dependsOn([web, api])
  .handle(async ({ services }) => {
    const inbox = new URL('/inbox', services.web); // WEB_URL
    const callback = new URL('/payments/callback', services.api); // API_URL
  });

A RestApi or a StaticSite hands back its public URL — <ID>_URL, the same address on every command and stage, which a browser opens and another app's server calls through the edge. The edge is what puts the key in the worker's environment, under gkm dev and gkm compose alike.

A construct that owns no client — a Cron, a Subscriber — types its service as never, so depending on one is a compile error rather than a stub that throws at runtime. You cannot call a queue worker; you send to its queue.

The auth server, and what its callers hold

BetterAuth declares an app of its own (path), and that app is the only process the server runs in: its generated entry calls auth.server(), which reads AUTH_SECRET, opens the auth tenant and builds the mailer its options use.

Everywhere else — the API, a worker, a feature test's in-process endpoints — .dependsOn([auth]) hands the handler an AuthClient: the auth app's URL and nothing more. Its one call keeps Better Auth's shape, so a handler written against the server reads the same:

export const getUser = router
  .get('/users/:id')
  .dependsOn([auth])
  .handle(async ({ services, header }) => {
    // GET <AUTH_URL><basePath>/get-session, with this request's cookie
    const session = await services.auth.api.getSession({
      headers: { cookie: header('cookie') ?? '' },
    });
    if (!session) throw new UnauthorizedError('Not signed in');
    // session.user, session.session — Better Auth's own types
  });

• Only cookie, authorization and origin are forwarded, with the active trace context and the client's address — see below.
• The client's address (the request's x-forwarded-for) travels as x-gkm-client-ip, never as x-forwarded-for. The auth server reads it first when it rate-limits (advanced.ipAddress.ipAddressHeaders is ['x-gkm-client-ip', …yours or 'x-forwarded-for']), so each user still has a bucket of their own. Sent as a forwarding header, the call would look like outside traffic, and the auth server's span would start a trace of its own instead of continuing the API's.
• x-gkm-client-ip is reserved by gkm. The session check goes to the auth server's public address, through the stack's edge — Caddy, the shared Traefik, the gkm dev edge — with the stack's token, x-gkm-edge-token (GKM_EDGE_TOKEN, derived from the stage's seed and handed to every process). The edge removes the header from every request without it, and the auth server ignores the header unless the request carries its own token or comes from an internal caller (no Origin, no forwarding header, a private or loopback peer), so nobody outside can choose the address they are rate-limited by. Don't send it from your own code.
• No session — Better Auth's null, or a 401 — is null.
• A server that answers with any other failure throws SessionCheckFailed (with its status), and one that cannot be reached throws AuthServerUnreachable (with the url it tried). Neither is read as "signed out".
• getSession is all the client does. Anything else Better Auth's server api offers — signing in, listing sessions, an admin plugin's calls — is a request to the auth app from wherever the user is (its browser client), or an endpoint of the auth app itself, not a call another app makes in-process.

So an app that calls the auth server is given AUTH_URL — on the compose network, http://auth:<port> — and never AUTH_SECRET, the auth tenant's URL or its mail keys. In a feature test the auth app is served in-process at its URL, so the client's request goes where it would deployed.

The .auth(auth) a surface declares (session(async ({ auth }) => auth.getSession())) makes the same call.

An API somebody else runs

Polar, Stripe, a payment gateway: nothing to provision, but an address that changes by stage and credentials the provider issued for each one.

// constructs/polar.ts
import { ExternalApi } from '@geekmidas/constructs/external-api';

export const polar = new ExternalApi('Polar', {
  url: 'https://www.polaraccesslink.com',
  credentials: z.object({ clientId: z.string(), clientSecret: z.string() }),
  client: ({ url, credentials }) => new PolarClient(url, credentials),
});

.dependsOn([polar])
.handle(async ({ services }) => services.polar.exchangeCode(code))

It provides two keys, POLAR_URL and POLAR_CREDENTIALS:

	Deployed, and gkm dev	gkm dev --fake and gkm test
POLAR_URL	url, for the stage	the fake
POLAR_CREDENTIALS	the stage's secret	the fake's credentials

Fakes are opt-in. Plain gkm dev calls the real API — its sandbox, say — at its url for the local stage, with the local stage's own credentials. gkm test always uses the fakes, and gkm dev --fake uses them too. Nothing that acts on a stage (gkm setup, secrets:*, deploy) reads one.

url is one URL or one per stage. A string answers every stage. A record is keyed by stage name, with default for any stage not listed — for a provider whose sandbox lives somewhere else:

url: {
  prod: 'https://www.payfast.co.za',
  default: 'https://sandbox.payfast.co.za',
},

Credentials are one JSON value per stage:

gkm secrets:set POLAR_CREDENTIALS '{"clientId":"…","clientSecret":"…"}' --stage prod

Both are checked at deploy: a stage the url record does not name, with no default, fails with NoUrlForStage, and a stage with no POLAR_CREDENTIALS fails with MissingSuppliedSecret, naming the command above. The JSON itself is validated against the schema when the process starts (MalformedCredential).

The fake

The construct never names its fake — so no bundler can carry a fake, or the responses it answers from, into a deployed build. It lives at test/fakes/<id>.ts, found by name the way test/factories/<database>.ts is, and default-exports one of two things.

An app — a working implementation of the API, served by gkm in-process through MSW in a feature test and on an allocated port by gkm dev --fake:

// test/fakes/polar.ts
import { fake } from '@geekmidas/constructs/external-api';
import type { polar } from '../../constructs/polar';

export default fake.app<typeof polar>(
  new Hono().post('/v3/users', (c) => c.json({ 'polar-user-id': 1 })),
  { credentials: { clientId: 'fake', clientSecret: 'fake' } },
);

Or an image the provider publishes, run as a container. port is the one it listens on inside; the host port is allocated like every other container's:

// test/fakes/stripe.ts
export default fake.image<typeof stripe>('stripe/stripe-mock', {
  port: 12111,
  credentials: { secretKey: 'sk_test_fake', webhookSecret: 'whsec_fake' },
});

<typeof polar> checks the fake's credentials against the construct's schema. An external API with no fake fails gkm test and gkm dev --fake with NoFake, naming the file to create. Writing a fake, asserting on what it received, and how long its state lives are in the testing guide.

Encrypting what you store

import { Encryption } from '@geekmidas/constructs/encryption';

export const pii = new Encryption('Pii');

export const createUser = router
  .post('/users')
  .dependsOn([pii])
  .body(z.object({ email: z.email() }))
  .handle(async ({ body, db, services }) => {
    await db
      .insertInto('users')
      .values({
        email: await services.pii.encrypt(body.email),
        // A blind index: the same email always gives the same string, so the
        // encrypted column can still be found without decrypting it.
        emailIndex: await services.pii.index(body.email),
      })
      .execute();
  });

// …and to look one up
db.selectFrom('users').where('emailIndex', '=', await services.pii.index(email));

The app names no cipher and holds no key. Pii provides one value, PII_URL, and its scheme picks the backend:

Where	What holds the key
gkm dev, gkm test	an aes256gcm:// keyring derived from the project and stage — nothing to set
A server stage (Dokploy)	an aes256gcm:// keyring generated into the stage's secrets on its first deploy
AWS	a KMS key (envelope encryption, rotated yearly by KMS) and a KMS HMAC key for the index — granted only to functions that .dependsOn([pii])

Every ciphertext names the key that wrote it (gkm1.k2.…), and is bound to the construct that wrote it — a value copied into another Encryption's column does not decrypt. On a server stage, rotation is three steps:

gkm encryption:rotate Pii --stage production   # new writes use k2; k1 still opens
# redeploy, then move what is stored onto k2:
#   for each row: row.email = await services.pii.reencrypt(row.email)
gkm encryption:retire Pii k1 --stage production

Retiring is never automatic — nothing but the sweep knows a key is unused — and a value still under an old key logs a warning the first time a process decrypts it. The index key does not rotate: a rotated index key would make every stored index miss. On AWS neither command applies; KMS keeps every version it rotated through, so nothing is ever stranded.

The database, and what comes off it

export const database = new KyselyDatabase<Database, 'Orders'>('Orders');

export const replica = database.reader();                    // OrdersReader
export const cache = database.cache();                       // cached *in* this database
export const acme = database.schema<TenantDB, 'Acme'>('Acme'); // its own role + URL
database.owner;                                              // the DDL role

database.cache()

A cache backed by this database — entries are rows in a table inside it, created in this database's schema and reached by its role:

database.cache();                          // OrdersCache, table `cache`
database.cache('Sessions');                // a different id
database.cache('Sessions', { table: 'kv' }); // a different table

It is the stronger of the two forms. new Cache('Sessions') says the app caches and leaves where to the deployment; this says it caches here, which is a fact about the application rather than about a stage — so the declaration wins over the deploy target, and a target that puts caches in Upstash does not move it.

What that buys beyond being explicit: the table's schema and the role that reaches it come from this database rather than from a second convention, and the backend never has to guess which database "the database" meant in an app that declares two.

The client is still a CacheClient — a key/value store, not a query builder, so this returns a Cache and not a KyselyDatabase. And the table is DDL, so whatever applies DDL creates it: a handler's role may not create anything, which is the same reason the Postgres driver does not create it lazily.

.database(database) on a factory puts the client in the handler context as db, and the execution wrapper is what opens the transaction and applies the RLS context:

export const router = api.database(database);

export const listOrders = router
  .get('/orders')
  .handle(async ({ db }) => db.selectFrom('orders').selectAll().execute());

.owner is a Service rather than a construct, deliberately: .dependsOn() takes constructs, so a handler cannot ask for DDL rights at all. A migrator asks for it by name.

Discovery

Point constructs: './src/constructs/**/*.ts' at them in gkm.config.ts. One glob, every kind — a declared ObjectStorage has no kind to be listed under, so a glob per kind could never find it. Discovery inspects every export of every matching module and asks one structural question: does it have an id, and can it declare()?

Basic Usage

Connections you can trace

Every connection a database opens says who it is, and every query run inside a request says what ran it — so a database at max_connections, or a slow query in the log, points at the code responsible.

• application_name is the process: the Lambda function's name, or the surface's id on a server (Api). It is a fallback, so PGAPPNAME or ?application_name= in the URL still win.

  select application_name, count(*) from pg_stat_activity group by 1;

• Query tags. Inside an endpoint, subscriber, queue or cron, each query ends in a sqlcommenter comment:

  select … from "orders" where "id" = $1 /*operation='GET /orders/{id}',request_id='7f2a…',traceparent='00-4bf9…-00f0…-01'*/

  Inside a trace, traceparent is the query's own span, in W3C form — so a slow query in the log, or a row of pg_stat_activity, is one search away from its trace. Outside a request and a trace, a query is left as written. It appears in pg_stat_activity.query and in the server's slow-query and auto_explain logs, and costs no round trip; pg_stat_statements ignores comments, so its grouping is unchanged. Turn it off with new KyselyDatabase('Orders', { queryTags: false }).

• An idle connection the server ends — idle_session_timeout, a failover — is logged and replaced, never an uncaught error that takes the process down.

• Shutdown. A production server stops taking requests on SIGTERM, lets in-flight ones finish, then closes every pool (runShutdownHooks from @geekmidas/constructs), so a rolling deploy does not leave the old task holding connections. It exits by GKM_SHUTDOWN_TIMEOUT_MS (8000 by default — under Docker's 10s stop timeout) even if a request has not finished, with exit code 1 so the forced stop is visible.

Creating an Endpoint

import { api } from '../constructs/api';
import { z } from 'zod';

export const getUserEndpoint = api
  .get('/users/:id')
  .params(z.object({
    id: z.string().uuid(),
  }))
  .output(z.object({
    id: z.string(),
    name: z.string(),
    email: z.string(),
  }))
  .handle(async ({ params }) => {
    const user = await db.users.findById(params.id);
    if (!user) {
      throw createError.notFound('User not found');
    }
    return user;
  });

Input Validation

Endpoints support three types of input validation:

const createUserEndpoint = api
  .post('/users')
  .body(z.object({
    name: z.string(),
    email: z.email(),
  }))
  .query(z.object({
    sendEmail: z.boolean().optional(),
  }))
  .handle(async ({ body, query }) => {
    const user = await createUser(body);

    if (query.sendEmail) {
      await sendWelcomeEmail(user.email);
    }

    return user;
  });

Error Handling

import { createError } from '@geekmidas/errors';

// Built-in HTTP errors
throw createError.badRequest('Invalid input');
throw createError.unauthorized('Not authenticated');
throw createError.forbidden('Not authorized');
throw createError.notFound('Resource not found');
throw createError.conflict('Resource already exists');
throw createError.internalServerError('Something went wrong');

Services — the escape hatch

Reach for a construct first. A third-party API is an ExternalApi, a key somebody issued you is a Credential, and a database is a KyselyDatabase — each is one declaration, and each goes in .dependsOn([...]):

export const database = new KyselyDatabase<Database, 'Orders'>('Orders');

const endpoint = api
  .get('/data')
  .dependsOn([database])
  .handle(async ({ services }) =>
    services.database.selectFrom('orders').selectAll().execute(),
  );

A hand-written Service ({ serviceName, register }) is what the build cannot see into: it has to run the service against a sniffer to learn which env keys it touches, and it records no edge, so a deploy target cannot grant it anything. The builders still accept one through .services() for code that has not moved yet; new code should not need it.

Advanced Usage

Request Context

The endpoint handler receives a context object that includes all request data, services, logger, headers, cookies, and session information.

Reading Cookies

Access incoming request cookies using the cookie function:

const endpoint = api
  .get('/dashboard')
  .handle(async ({ cookie }) => {
    const sessionId = cookie('session');
    const theme = cookie('theme') || 'light';

    if (!sessionId) {
      throw createError.unauthorized('Session cookie missing');
    }

    return {
      sessionId,
      theme,
    };
  });

Using Cookies for Authentication

Combine cookie reading with session management by calling .session() on a branch — here the app's router (api.database(database)):

// The surface was declared with `.auth(auth)`; `auth` here is that construct,
// verifying this request's headers — the session cookie included.
const sessionRouter = router.session(async ({ auth }) => {
  const session = await auth.getSession();
  if (!session) {
    throw new ForbiddenError('No active session');
  }

  return session;
});

// Endpoints created from sessionRouter automatically have session available
const profileEndpoint = sessionRouter
  .get('/profile')
  .handle(async ({ session }) => {
    return {
      userId: session.userId,
      email: session.email,
    };
  });

Response Handling

The endpoint handler receives two parameters: the context object and a response builder. The response builder allows you to set cookies, custom headers, and override status codes.

Setting Cookies

import { api } from '../constructs/api';
import { z } from 'zod';

const loginEndpoint = api
  .post('/auth/login')
  .body(z.object({
    email: z.email(),
    password: z.string(),
  }))
  .output(z.object({
    id: z.string(),
    email: z.string(),
  }))
  .handle(async ({ body }, response) => {
    const user = await authenticateUser(body);

    // Set authentication cookie
    return response
      .cookie('session', user.sessionToken, {
        httpOnly: true,
        secure: true,
        sameSite: 'strict',
        maxAge: 60 * 60 * 24 * 7, // 7 days in seconds
        path: '/',
      })
      .send(user);
  });

Cookie Options:

• domain: Cookie domain
• path: Cookie path (default: '/')
• expires: Expiration date
• maxAge: Maximum age in seconds
• httpOnly: Prevent JavaScript access
• secure: Only send over HTTPS
• sameSite: CSRF protection ('strict' | 'lax' | 'none')

Deleting Cookies

const logoutEndpoint = api
  .post('/auth/logout')
  .output(z.object({ success: z.boolean() }))
  .handle(async (ctx, response) => {
    // Delete the session cookie
    return response
      .deleteCookie('session', { path: '/' })
      .send({ success: true });
  });

Custom Headers

Set custom response headers for cache control, content disposition, or custom metadata:

const downloadEndpoint = api
  .get('/files/:id/download')
  .params(z.object({ id: z.string() }))
  .handle(async ({ params }, response) => {
    const file = await getFile(params.id);

    return response
      .header('Content-Disposition', `attachment; filename="${file.name}"`)
      .header('X-File-Size', file.size.toString())
      .header('Cache-Control', 'private, max-age=3600')
      .send(file.data);
  });

Dynamic Status Codes

Override the default status code (200) or the status set in the builder:

import { SuccessStatus } from '@geekmidas/constructs/endpoints';

const createEndpoint = api
  .post('/users')
  .body(z.object({
    name: z.string(),
    email: z.email(),
  }))
  .output(z.object({
    id: z.string(),
    name: z.string(),
    email: z.string(),
  }))
  .handle(async ({ body }, response) => {
    const user = await createUser(body);

    // Return 201 Created with Location header
    return response
      .status(SuccessStatus.Created)
      .header('Location', `/users/${user.id}`)
      .send(user);
  });

Available Success Status Codes:

• SuccessStatus.OK - 200
• SuccessStatus.Created - 201
• SuccessStatus.Accepted - 202
• SuccessStatus.NoContent - 204
• SuccessStatus.ResetContent - 205
• SuccessStatus.PartialContent - 206

Default Headers

Set headers that apply to all responses from an endpoint:

const apiEndpoint = api
  .get('/api/data')
  .header('X-API-Version', '1.0')
  .headers({
    'Cache-Control': 'no-cache',
    'X-Custom-Header': 'value',
  })
  .output(dataSchema)
  .handle(async () => {
    return await getData();
  });

Simple Responses (No Modifications)

If you don't need to modify the response, simply return the data directly:

const getEndpoint = api
  .get('/users/:id')
  .params(z.object({ id: z.string() }))
  .output(userSchema)
  .handle(async ({ params }) => {
    // No response parameter needed if not using it
    return await getUser(params.id);
  });

Complete Example

Combining multiple request and response features:

const uploadEndpoint = api
  .post('/files/upload')
  .body(z.object({
    file: z.string(),
    name: z.string(),
  }))
  .output(z.object({
    id: z.string(),
    url: z.string(),
  }))
  .handle(async ({ body, cookie, logger }, response) => {
    // Read existing preference cookie
    const preferredFormat = cookie('format') || 'standard';

    const file = await uploadFile(body, preferredFormat);

    logger.info({ fileId: file.id }, 'File uploaded successfully');

    // Set multiple cookies and headers in response
    return response
      .status(SuccessStatus.Created)
      .header('Location', `/files/${file.id}`)
      .header('X-File-Id', file.id)
      .cookie('last-upload', file.id, {
        maxAge: 60 * 60, // 1 hour
        httpOnly: true,
      })
      .cookie('upload-count', String(getUploadCount() + 1), {
        maxAge: 60 * 60 * 24 * 365, // 1 year
      })
      .send({
        id: file.id,
        url: file.url,
      });
  });

Direct uploads

.upload(bucket, …) makes an endpoint hand the browser a signed form that uploads one file straight to the bucket — the file never passes through your server. The endpoint decides where (key, resolved from the session, body and params); the form's policy decides what: the storage refuses a file over maxSize, or of a type accept does not allow.

import { uploads } from '../constructs/storage';

export const avatarUpload = router
  .post('/me/avatar')
  .upload(uploads, {
    key: ({ session }) => `avatars/${session.userId}.png`,
    maxSize: 5 * 1024 * 1024,
    accept: ['image/png'],
  })
  .handle(async ({ upload }) => upload); // { url, fields, key, expiresAt }

The browser posts every field of fields, then the file last as file, to url as multipart/form-data.

• The bucket is a dependency, as with .dependsOn([uploads]): services.uploads is in the handler and the manifest records the edge.

• The form is the output — typed, and documented in OpenAPI — unless an .output() came first; then the handler returns what that schema says.

• .upload() is the last step before .handle(). The key is resolved from the body, params and session, so they are declared first.

• accept takes media types (image/png) and whole types (image/*). A signed policy fixes one type or one prefix, never "one of these", so with several, contentType says which this upload is — and a type not in accept is refused with a 415 (UploadContentTypeRefused):

  .body(z.object({ contentType: z.enum(['application/pdf', 'image/png']) }))
  .upload(uploads, {
    key: ({ session }) => `documents/${session.userId}/${crypto.randomUUID()}`,
    maxSize: 20 * 1024 * 1024,
    accept: ['application/pdf', 'image/png'],
    contentType: ({ body }) => body.contentType,
  })

• expiresIn is in seconds — 300 by default, at most seven days.

Options that could never be honoured fail when the module loads, not on the first request: UploadMaxSizeInvalid, UploadExpiryInvalid, UploadAcceptInvalid, UploadContentTypeUnresolved.

The endpoint never learns that the upload finished — the browser talks to the storage, not to it. That arrives as the bucket's object.created event, to a worker subscribed to the bucket (#273).

HTML forms and redirects

A server-rendered page needs two things an API rarely does: a body posted by an HTML <form>, and a redirect after the post. Both work on every adaptor — a Hono server, gkm dev, API Gateway and a feature test.

The body is read by its Content-Type into the same .body() schema:

Content-Type	What the schema is handed
application/json, application/*+json	the parsed JSON
application/x-www-form-urlencoded, multipart/form-data	an object of fields; a repeated field (or one named field[]) is an array, a file part a File
text/*	the text
anything else	nothing — the request is answered 415 Unsupported Media Type

A form field is always a string, so coerce what isn't (z.coerce.number()). A body that fails the schema is the same 422 whatever it was sent as.

response.redirect(url, status = 303) sets Location and the status and sends no body. A redirect is not checked against .output(), so a page endpoint can redirect instead of rendering. Since a redirect has no output, an endpoint's declared .event()s and .audit()s are not sent for it. .status() and response.status() also take a 3xx (RedirectStatus: 301, 302, 303, 307, 308).

import { z } from 'zod';

// GET /login — the page
export const loginPage = api
  .get('/login')
  .query(z.object({ error: z.string().optional() }))
  .output(z.string())
  .responseType('text/html')
  .handle(async ({ query }) => `<!doctype html>
    <form method="post" action="/login">
      ${query.error ? '<p>Wrong email or password</p>' : ''}
      <input name="email" type="email" required>
      <input name="password" type="password" required>
      <button>Sign in</button>
    </form>`);

// POST /login — the form posts application/x-www-form-urlencoded
export const login = router
  .post('/login')
  .body(z.object({ email: z.email(), password: z.string() }))
  .output(z.string())
  .responseType('text/html')
  .handle(async ({ body, db }, response) => {
    const session = await signIn(db, body.email, body.password);
    if (!session) return response.redirect('/login?error=1');

    // Post/Redirect/Get: a reload of the next page doesn't post again
    return response
      .cookie('session', session.token, {
        httpOnly: true,
        secure: true,
        sameSite: 'lax',
        path: '/',
      })
      .redirect('/dashboard');
  });

In a feature test, the browser follows a redirect as a browser does; pass redirect: 'manual' to see it instead:

it('signs in', async ({ browser }) => {
  const response = await browser.fetch(`${API_URL}/login`, {
    method: 'POST',
    headers: { 'content-type': 'application/x-www-form-urlencoded' },
    body: new URLSearchParams({ email, password }).toString(),
    redirect: 'manual',
  });

  expect(response.status).toBe(303);
  expect(response.headers.get('location')).toBe('/dashboard');
});

The OpenAPI document and the generated client still describe the body as JSON: a form is for a browser, and JSON is what a typed client sends.

Authorization and Sessions

.session() is called on the factory to create a session-enabled router. The session callback receives header, cookie, services, auth (the construct the surface named with .auth(…)), and db (when a database is configured). Throw an error to reject unauthorized requests.

import { api } from '../constructs/api';
import { ForbiddenError } from '@geekmidas/errors';

// Branch from the surface with the database, then add a session.
// `auth` is the construct the surface named with `.auth(auth)`.
const sessionRouter = api
  .database(database)
  .session(async ({ auth, db }) => {
    const session = await auth.getSession();
    if (!session) {
      throw new ForbiddenError('No active session');
    }

    return session;
  });

// Endpoints created from sessionRouter have session in their context
const protectedEndpoint = sessionRouter
  .get('/protected')
  .handle(async ({ session }) => {
    return { userId: session.id };
  });

Authorization

.authorize() adds an authorization check that runs after session extraction and input validation, but before the handler. It can be set on the factory (applies to all endpoints) or on individual endpoints. The callback returns a boolean or Promise<boolean> — returning false results in a 401 Unauthorized response.

The authorize context includes session, services, logger, header, cookie, and validated request input (body, query, params) when schemas are defined:

// Factory-level: applies to all endpoints created from this factory
const adminRouter = sessionRouter
  .authorize(({ session }) => session.role === 'admin');

const adminEndpoint = adminRouter
  .post('/admin/action')
  .handle(async ({ session }) => {
    return { status: 'ok' };
  });

// Endpoint-level: authorize based on request input
const updateEndpoint = sessionRouter
  .post('/orgs/:orgId/members')
  .params(z.object({ orgId: z.string() }))
  .body(z.object({ userId: z.string(), role: z.string() }))
  .authorize(({ session, body, params }) => {
    // Check the user has permission for this org and role assignment
    return session.orgs.includes(params.orgId) && body.role !== 'owner';
  })
  .handle(async ({ body, params }) => {
    return { ok: true };
  });

// Async authorization with service lookups
const resourceEndpoint = sessionRouter
  .get('/resources/:id')
  .params(z.object({ id: z.string() }))
  .authorize(async ({ session, params, services }) => {
    return await services.permissions.canAccess(session.userId, params.id);
  })
  .handle(async ({ params, services }) => {
    return await services.resources.getById(params.id);
  });

Rate Limiting

import { InMemoryCache } from '@geekmidas/cache/memory';

const rateLimitedEndpoint = api
  .post('/api/messages')
  .rateLimit({
    limit: 10,
    windowMs: 60000, // 1 minute
    cache: new InMemoryCache(),
  })
  .body(messageSchema)
  .handle(async ({ body }) => {
    return await sendMessage(body);
  });

Audit Logging

Endpoints support declarative and manual audit logging via integration with @geekmidas/audit. Audits can be recorded automatically after a handler completes or manually inside the handler, and they are flushed atomically within the same database transaction when possible.

Setting Up Audit Storage

Define an audit storage service and attach it to an endpoint with .auditor():

import { api } from '../constructs/api';
import { KyselyAuditStorage } from '@geekmidas/audit/kysely';
import type { Service } from '@geekmidas/services';
import type { AuditableAction } from '@geekmidas/audit';

// Define type-safe audit actions
type AppAuditAction =
  | AuditableAction<'user.created', { userId: string; email: string }>
  | AuditableAction<'user.updated', { userId: string; changes: string[] }>
  | AuditableAction<'user.deleted', { userId: string }>;

// Create audit storage service
const auditStorageService = {
  serviceName: 'auditStorage' as const,
  async register(envParser) {
    return new KyselyAuditStorage<Database>({
      db: kyselyDb,
      tableName: 'audit_logs',
    });
  },
} satisfies Service<'auditStorage', KyselyAuditStorage<Database>>;

const endpoint = api
  .post('/users')
  .auditor(auditStorageService)
  .handle(async ({ auditor }) => {
    // auditor is now available in the handler context
    return { id: '123' };
  });

Actor Extraction

Use .actor() to identify who performed the action. The extractor receives the request context and returns an AuditActor:

const endpoint = api
  .post('/users')
  .auditor(auditStorageService)
  .actor(({ session }) => ({
    id: session.sub,
    type: 'user',
    data: { email: session.email },
  }))
  .handle(async ({ auditor }) => {
    // auditor.actor is { id: session.sub, type: 'user', ... }
    return { id: '123' };
  });

The actor extractor can also be async and has access to services, session, header, cookie, and logger.

Declarative Audits

Use .audit() to define audits that fire automatically after the handler returns successfully. Each audit receives the handler's response to extract the payload:

import { z } from 'zod';

const createUserEndpoint = api
  .post('/users')
  .auditor(auditStorageService)
  .actor(({ session }) => ({ id: session.sub, type: 'user' }))
  .body(z.object({ name: z.string(), email: z.string() }))
  .output(z.object({ id: z.string(), email: z.string() }))
  .audit([
    {
      type: 'user.created',
      payload: (response) => ({
        userId: response.id,
        email: response.email,
      }),
      entityId: (response) => response.id,
      table: 'users',
    },
  ])
  .handle(async ({ body }) => {
    const user = await createUser(body);
    return user;
  });

Audit definition fields:

Field	Type	Description
type	string	The audit action type (must match a defined AuditableAction)
payload	(response) => object	Extracts the payload from the handler response
when	(response) => boolean	Optional condition — skips the audit if it returns false
entityId	(response) => string	Optional entity identifier for querying
table	string	Optional table name for querying

Conditional Audits

Use the when clause to only record audits under certain conditions:

const updateUserEndpoint = api
  .patch('/users/:id')
  .auditor(auditStorageService)
  .actor(({ session }) => ({ id: session.sub, type: 'user' }))
  .output(userResponseSchema)
  .audit([
    {
      type: 'user.updated',
      payload: (response) => ({
        userId: response.id,
        changes: response.changedFields,
      }),
      when: (response) => response.changedFields.length > 0,
    },
  ])
  .handle(async ({ body, params }) => {
    return await updateUser(params.id, body);
  });

Manual Audits in Handlers

When you need more control, call ctx.auditor directly inside the handler. This is useful for auditing intermediate steps or conditional logic:

const transferEndpoint = api
  .post('/transfers')
  .auditor(auditStorageService)
  .actor(({ session }) => ({ id: session.sub, type: 'user' }))
  .handle(async ({ body, auditor }) => {
    const result = await processTransfer(body);

    auditor.audit('transfer.completed', {
      transferId: result.id,
      amount: result.amount,
    });

    if (result.flagged) {
      auditor.audit('transfer.flagged', {
        transferId: result.id,
        reason: result.flagReason,
      });
    }

    return result;
  });

Manual audits are buffered in memory and flushed together with any declarative audits when the handler completes.

Factory-Level Defaults

Set .auditor() and .actor() on a branch of the surface's factory so every endpoint built from it inherits the configuration:

import { api } from '../constructs/api';

const router = api
  .database(database)
  .session(extractSession)
  .authorizer('jwt')
  .auditor(auditStorageService)
  .actor(({ session }) => ({
    id: session.sub,
    type: 'user',
  }));

// All endpoints inherit auditor and actor
const createUser = router
  .post('/users')
  .audit([{
    type: 'user.created',
    payload: (response) => ({
      userId: response.id,
      email: response.email,
    }),
  }])
  .handle(async ({ body }) => {
    return await createUser(body);
  });

const deleteUser = router
  .delete('/users/:id')
  .handle(async ({ params, auditor }) => {
    await removeUser(params.id);
    auditor.audit('user.deleted', { userId: params.id });
    return { success: true };
  });

Transaction Coordination

When the audit storage uses the same database as the endpoint (e.g., both use the same Kysely instance), audits are flushed inside the same database transaction. This guarantees atomicity — if the handler fails, both the data changes and the audit records are rolled back.

const router = api
  .database(database)
  .auditor(auditStorageService)
  .actor(({ session }) => ({ id: session.sub, type: 'user' }));

const endpoint = router
  .post('/users')
  .output(userSchema)
  .audit([{
    type: 'user.created',
    payload: (response) => ({
      userId: response.id,
      email: response.email,
    }),
  }])
  .handle(async ({ body, db }) => {
    // db is a transaction — both the insert and audit write
    // happen atomically in the same transaction
    const user = await db
      .insertInto('users')
      .values(body)
      .returningAll()
      .executeTakeFirstOrThrow();

    return user;
  });

TIP

When .database() is configured and the audit storage's databaseServiceName matches the endpoint's database, the framework automatically wraps the handler and the audit flush in a single transaction. No extra configuration is needed.

Row Level Security (RLS)

Endpoints support PostgreSQL Row Level Security via the .rls() method. When configured, the handler receives a db parameter — a transaction with PostgreSQL session variables set — so RLS policies can filter rows automatically.

Setting Up RLS on the Factory

Configure RLS once on the factory so all endpoints inherit it:

import { api } from '../constructs/api';

const router = api
  .database(database)
  .session(extractSession)
  .authorizer('jwt')
  .rls({
    extractor: ({ session }) => ({
      user_id: session.sub,
      tenant_id: session.tenantId,
    }),
    prefix: 'app', // optional, default: 'app'
  });

The .database() call tells the factory which service provides the database connection. The .rls() extractor receives the request context (session, services, headers, cookies, logger) and returns key-value pairs that become PostgreSQL session variables (e.g. app.user_id).

Using db in Handlers

When RLS is configured, handlers receive a db parameter — a transaction with the RLS context applied:

const listOrders = api
  .get('/orders')
  .handle(async ({ db }) => {
    // db is a transaction with app.user_id and app.tenant_id set
    // PostgreSQL policies using current_setting('app.user_id') filter rows automatically
    return db
      .selectFrom('orders')
      .selectAll()
      .execute();
  });

Important

Always use db from the handler context when RLS is configured. Using services.database directly bypasses the RLS transaction and PostgreSQL session variables won't be set.

// Wrong - bypasses RLS
.handle(async ({ services }) => {
  return services.database.selectFrom('orders').execute();
});

// Correct - uses RLS transaction
.handle(async ({ db }) => {
  return db.selectFrom('orders').execute();
});

Bypassing RLS for Specific Endpoints

For admin endpoints that need unrestricted access, bypass the factory-level RLS:

// Using .rls(false)
const adminOrders = api
  .get('/admin/orders')
  .rls(false)
  .handle(async ({ services }) => {
    return services.database
      .selectFrom('orders')
      .selectAll()
      .execute();
  });

// Or using .rlsBypass()
const adminUsers = api
  .get('/admin/users')
  .rlsBypass()
  .handle(async ({ services }) => {
    return services.database
      .selectFrom('users')
      .selectAll()
      .execute();
  });

Per-Endpoint RLS

You can also configure RLS on individual endpoints instead of (or in addition to) the factory:

import { api } from '../constructs/api';

const endpoint = api
  .get('/orders')
  .database(database)
  .rls({
    extractor: ({ session, header }) => ({
      user_id: session.userId,
      ip_address: header('x-forwarded-for'),
    }),
  })
  .handle(async ({ db }) => {
    return db
      .selectFrom('orders')
      .selectAll()
      .execute();
  });

PostgreSQL Policy Example

The session variables set by the RLS extractor are consumed by PostgreSQL policies:

-- Enable RLS on the table
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;

-- Create policies using session variables
CREATE POLICY tenant_isolation ON orders
  USING (tenant_id = current_setting('app.tenant_id', true));

CREATE POLICY user_access ON orders
  USING (user_id = current_setting('app.user_id', true));

Combining RLS with Other Handler Context

The db parameter coexists with services, logger, session, and other context:

const endpoint = api
  .get('/orders')
  .handle(async ({ db, services, logger, session }) => {
    const orders = await db
      .selectFrom('orders')
      .selectAll()
      .execute();

    logger.info({ count: orders.length }, 'Fetched orders');

    return orders;
  });

Event Publishing

Events go to a topic, and a topic is a construct: it declares the event contract, and its publisher is its service.

// constructs/topics.ts
import { Topic } from '@geekmidas/constructs/topic';

export const orders = new Topic('Orders', {
  events: {
    'order.created': z.object({ orderId: z.string(), total: z.number() }),
    'order.cancelled': z.object({ orderId: z.string() }),
  },
});

An endpoint publishes with .event(topic, …). The event is sent once the handler has succeeded, with a payload built from what it returned:

import { orders } from '../constructs/topics';
import { router } from './router';

export const createOrder = router
  .post('/orders')
  .body(orderSchema)
  .output(orderResponseSchema)
  .event(orders, {
    type: 'order.created',
    payload: (order) => ({ orderId: order.id, total: order.total }),
  })
  .handle(async ({ body, db }) =>
    db.insertInto('orders').values(body).returningAll().executeTakeFirstOrThrow(),
  );

type is checked against the topic's events and payload against that event's schema, so an event cannot be published in one shape and read in another. Pass when: (output) => boolean to publish only for some outputs. A failed publish is logged, never thrown: the handler has already succeeded, and an event that could not be delivered does not turn its response into an error.

Several topics. .event() is repeatable, and each call names its own topic. Two calls to two topics publish to each, through each topic's own publisher — there is no single shared one to configure:

export const cancelOrder = router
  .post('/orders/:id/cancel')
  .output(orderResponseSchema)
  .event(orders, {
    type: 'order.cancelled',
    payload: (order) => ({ orderId: order.id }),
  })
  .event(audit, {
    type: 'audit.recorded',
    payload: (order) => ({ subject: `order:${order.id}` }),
  })
  .handle(async ({ params, db }) => { /* … */ });

Publishing from the handler. .event(orders, …) also puts the topic's publisher in the handler as services.orders — exactly what .dependsOn([orders]) does — for an event the handler decides on itself:

export const createTransfer = router
  .post('/transfers')
  .dependsOn([transfers])
  .handle(async ({ body, services }) => {
    const result = await processTransfer(body);

    if (result.flagged) {
      await services.transfers.publish([
        { type: 'transfer.flagged', payload: { transferId: result.id } },
      ]);
    }

    return result;
  });

The service key is the topic's id, uncapitalised: Orders is services.orders. Depending on a topic is what gives a handler ORDERS_PUBLISHER_CONNECTION_STRING; the publisher reads it and picks its transport from the protocol — pgboss:// on a server target, sns:// on AWS (locally, against the AWS emulator).

Functions and crons take .event(topic, …) the same way. Publishing is never granted to a branch or a surface: only the routes that name the topic can publish to it.

Event Subscribers

A subscriber is built from the Worker that runs it, and binds to a topic rather than depending on it. Its handler is handed the topic's event types and no publisher; the runtime that feeds it reaches the topic through the topic's own connection string.

// constructs/worker.ts
import { Worker } from '@geekmidas/constructs/worker';

export const worker = new Worker('Jobs', { logger }).database(database);

The worker's .database(database) is the default database for everything built from it — subscribers, queues, crons and functions all receive it as db, typed from the construct. One that works against another database names its own with .database(other), which replaces the worker's for that runnable alone: its db, and the manifest edge a deploy grants it from.

Basic Subscriber

import { users } from '../constructs/topics';
import { worker } from '../constructs/worker';

export const onUserCreated = worker
  .topic(users)
  .subscribe(['user.created'])
  .handle(async ({ events, logger }) => {
    for (const event of events) {
      logger.info({ userId: event.payload.userId }, 'Processing new user');
    }
  });

.topic() comes first because what comes first decides what is built: a topic makes a subscriber, where .dependsOn() first would make a plain function.

With Database Access

The worker's database is already there as db:

export const onUserCreated = worker
  .topic(users)
  .subscribe(['user.created'])
  .handle(async ({ events, db }) => {
    await db
      .insertInto('profiles')
      .values(events.map((event) => ({ userId: event.payload.userId })))
      .execute();
  });

.database(other) gives one subscriber a different one, and retypes db:

export const trackSignup = worker
  .topic(users)
  .database(analytics)
  .subscribe(['user.created'])
  .handle(async ({ events, db }) => {
    // db is analytics' client, not the worker's
  });

With Dependencies

export const onOrderCreated = worker
  .topic(orders)
  .dependsOn([database, email])
  .subscribe(['order.created'])
  .handle(async ({ events, services, logger }) => {
    for (const event of events) {
      const order = await services.database
        .selectFrom('orders')
        .where('id', '=', event.payload.orderId)
        .selectAll()
        .executeTakeFirstOrThrow();

      await services.email.sendTemplate('orderConfirmation', {
        to: order.customerEmail,
        subject: 'Your order',
        props: { orderId: order.id, total: order.total },
      });

      logger.info({ orderId: order.id }, 'Order confirmation sent');
    }
  });

Subscribing to Multiple Events

export const onUserEvents = worker
  .topic(users)
  .subscribe(['user.created', 'user.updated'])
  .handle(async ({ events }) => {
    for (const event of events) {
      if (event.type === 'user.created') {
        // TypeScript narrows payload to the topic's 'user.created' schema
        await indexNewUser(event.payload.userId);
      } else {
        await reindexUser(event.payload.userId);
      }
    }
  });

Publishing Follow-up Events

A subscriber that emits follow-ups depends on the topic it emits to — the only reason it would need a publisher at all:

export const onUserCreated = worker
  .topic(users)
  .dependsOn([notifications])
  .subscribe(['user.created'])
  .handle(async ({ events, services }) => {
    await services.notifications.publish(
      events.map((event) => ({
        type: 'notification.sent',
        payload: { userId: event.payload.userId, kind: 'welcome' },
      })),
    );
  });

Configuration Options

Method	Description
worker.topic(topic)	The topic to bind to — types the events and records the binding
.subscribe(events)	Event type(s) to listen for (string or array)
.dependsOn([...])	Constructs whose clients the handler gets in services
.output(schema)	Validate the return value with a StandardSchema
.timeout(ms)	Set execution timeout in milliseconds
.handle(fn)	Define the handler and build the Subscriber

The logger comes from the worker.

Handler Context

{
  events: Array<{ type; payload }>;  // A batch, typed by the topic's contract
  services: ServiceRecord;           // From .dependsOn([...])
  logger: Logger;                    // The worker's
}

How Subscribers Run

Development (gkm dev) and server builds: Subscribers run in-process beside the server, each through the <ID>_PUBLISHER_CONNECTION_STRING of the topic it is bound to — there is no shared subscriber string. How a subscriber is fed follows that string:

• sns:// — pushed. The server mounts POST /__gkm/subscribers/<exportName>, subscribed to the topic with a filter policy on the type attribute listing the events from .subscribe([...]). SNS fans out, one subscription per subscriber. Against SNS itself the deploy makes that subscription and hands the server its ARN in GKM_SNS_SUBSCRIPTIONS; against an emulator the server subscribes itself once listening. The route is SnsPushSubscriberAdaptor (from @geekmidas/constructs/subscribers), which reads each notification as an SNS Lambda event — the same parsing, services and error handling as deployed, without Lambda's middy wrapper, so a server needs no @middy/core. Confirmations are confirmed automatically — through the API, as the topic's owner, against SNS itself — and signatures are verified except against an emulator. A notification naming a subscription other than the recorded one is refused. GKM_SUBSCRIBER_PUSH_URL is the base an emulator pushes to (by default http://host.docker.internal:<port>).
• pgboss://, rabbitmq:// — polled. On pg-boss each subscriber drains a queue of its own, <topic>/<exportName>, so every subscriber sees every event and replicas of one subscriber share it.

gkm dev --no-subscribers runs none of them. See Events: Dev Server.

Production (AWS Lambda): Each subscriber is compiled into a Lambda handler via AWSLambdaSubscriber, which parses SQS/SNS records, filters to the subscribed types, and invokes the handler. The build records the binding on the topic's manifest entry.

import { AWSLambdaSubscriber } from '@geekmidas/constructs/aws';
import { EnvironmentParser } from '@geekmidas/envkit';
import { onUserCreated } from './subscribers/userEvents';

const envParser = new EnvironmentParser(process.env);
const adaptor = new AWSLambdaSubscriber(envParser, onUserCreated);

export const handler = adaptor.handler;

Testing Subscribers

In a featureTest, a subscriber runs whenever a request publishes an event it named — delivered in-process, in the test's transaction, after the topic's schema has accepted the payload (see Background work, end to end). So the usual test drives the endpoint and asserts what the subscriber did.

To run one on its own — a redelivery, an event no endpoint publishes yet — subscriber() hands it events directly, with the test's services:

import { it } from '#test';

it('writes each user a notification', async ({ subscriber, db }) => {
  await subscriber(onUserCreated).invoke({
    events: [
      {
        type: 'user.created',
        payload: { userId: '123', email: 'ada@example.com', name: 'Ada' },
      },
    ],
  });

  // Assert on what the handler did — a row, a mail in the mailbox, …
});

TestSubscriberAdaptor from @geekmidas/constructs/testing does the same outside a feature test: new TestSubscriberAdaptor(onUserCreated).invoke({ events }).

Project Configuration

Subscribers are discovered by the one constructs glob in gkm.config.ts, like every other construct:

constructs: [
  './constructs/**/*.ts',
  './apps/*/{endpoints,queues,subscribers,crons,functions}/**/*.ts',
],

Queues

A queue is point-to-point work: a queue and its single consumer, declared as one construct. Where a subscriber is topic fan-out — any number of them, each filtering a stream by event type — a queue drains every message of its one typed message and hands it to exactly one handler. Reach for a queue for jobs (send an order to be fulfilled), and a topic for domain events (tell everyone a user was created).

Basic Queue

import { worker } from '../constructs/worker';
import { z } from 'zod';

export const orderJobs = worker
  .queue('OrderJobs')
  .message(z.object({ orderId: z.string() }))
  .handle(async ({ messages, logger }) => {
    for (const { orderId } of messages) {
      logger.info({ orderId }, 'Fulfilling order');
    }
  });

One construct because a queue has exactly one consumer: nothing can attach a second, or forget to attach the first.

With Database Access

A queue built from a worker with .database(database) gets that database as db; .database(other) replaces it for this queue:

export const orderJobs = worker
  .queue('OrderJobs')
  .message(z.object({ orderId: z.string() }))
  .handle(async ({ messages, db }) => {
    for (const { orderId } of messages) {
      await db
        .updateTable('orders')
        .set({ status: 'fulfilled' })
        .where('id', '=', orderId)
        .execute();
    }
  });

export const reportJobs = worker
  .queue('ReportJobs')
  .database(analytics)
  .message(z.object({ reportId: z.string() }))
  .handle(async ({ messages, db }) => { /* db is analytics' client */ });

Every runtime hands it over the same way — the server's poller, the Lambda adaptor, and TestQueueAdaptor, which also takes a db in the request to stand in for it (a transaction, say).

With Dependencies

.dependsOn() names constructs. The edge is what the manifest records, and what a deploy target reads to grant this consumer exactly what it named:

export const orderJobs = worker
  .queue('OrderJobs')
  .dependsOn([database])
  .message(z.object({ orderId: z.string() }))
  .handle(async ({ messages, services }) => {
    for (const { orderId } of messages) {
      await fulfil(services.database, orderId);
    }
  });

Sending to a Queue

A producer sends to a queue by depending on it. The queue's service is its publisher, typed to its message:

import { orderJobs } from '../queues/orderJobs';

export const createOrder = router
  .post('/orders')
  .body(z.object({ sku: z.string() }))
  .dependsOn([orderJobs])
  .handle(async ({ body, services }) => {
    const orderId = crypto.randomUUID();
    await services.orderJobs.publish([
      { type: 'OrderJobs', payload: { orderId } },
    ]);
    return { orderId };
  });

type is the queue's name exactly as written — it is the wire type the consumer subscribes to, so it is never rewritten. The service key is the canonical id, uncapitalised (services.orderJobs).

The publisher reads ORDER_JOBS_PUBLISHER_CONNECTION_STRING and selects its transport from the protocol — pgboss:// locally, sqs:// deployed — so the same code publishes to Postgres in dev and SQS in prod. Each queue gets its own key, given to the constructs that depend on it and to the server that runs its consumer.

Configuration Options

Method	Description
worker.queue(name)	Queue name — the wire type, and the source of its id and env keys
.message(schema)	The typed message payload
.dependsOn([...])	Constructs whose clients the handler gets in services
.timeout(ms)	Handler timeout (default 30000)
.batchSize(n)	SQS event-source batch size (deployed)
.fifo()	Mark the queue as FIFO (deployed)

How Queues Run

A queue is not an HTTP route — it's background work, and it runs in three modes:

Development (gkm dev): The CLI generates a poller that runs in-process alongside the Hono server. Each queue's consumer polls its own queue through the queue's own ORDER_JOBS_PUBLISHER_CONNECTION_STRING — the string its producers publish on — so it is always the queue they reach: pg-boss by default, or the SQS queue on the local AWS emulator for an AWS target. Queues are polled on every transport (SQS cannot push), and --no-subscribers leaves them running. No Lambda required locally.

Production (AWS Lambda): Each queue's consumer is compiled into a Lambda handler via AWSLambdaQueue, subscribed to its SQS queue. The handler unwraps each record's payload, validates it against message, and hands the batch to the handler. It uses SQS partial-batch responses, so a record that fails validation (or a handler error) is retried without re-processing the rest.

import { AWSLambdaQueue } from '@geekmidas/constructs/aws';
import { EnvironmentParser } from '@geekmidas/envkit';
import { orderJobs } from './queues/orderJobs';

const envParser = new EnvironmentParser(process.env);
const adaptor = new AWSLambdaQueue(envParser, orderJobs);

export const handler = adaptor.handler;

Production (Server): With gkm build --provider server, queues are included in the generated server and poll using the configured connection string (same as dev).

Testing Queues

In a featureTest, what a request enqueues is delivered to the queue's consumer before the response comes back — in-process, in the test's transaction, after the queue's message schema accepted it. published(queue) still records it:

import { it } from '#test';

it('enqueues the order and fulfils it', async ({ browser, published }) => {
  const { orderId } = await browser.api.post('/orders', { body: { sku: 'A1' } });
  expect(published(orderJobs)).toEqual([
    { type: 'OrderJobs', payload: { orderId } },
  ]);

  // The consumer has already run: assert what it did (order fulfilled, …).
});

queue(orderJobs).invoke({ messages }) runs the consumer on its own — a redelivery, say — and delivers whatever it publishes.

TestQueueAdaptor from @geekmidas/constructs/testing runs a consumer outside a feature test: new TestQueueAdaptor(orderJobs).invoke({ messages }).

Cron Jobs

A cron is a scheduled function, built from the worker that runs it: worker.cron(schedule). It supports dependencies, input/output schemas, logging, event publishing, and database access. On AWS it is an EventBridge rule; on a server the worker schedules it in Postgres, which is why the worker takes .database(database) — and that database is every cron's db unless the cron names its own.

Basic Cron

import { worker } from '../constructs/worker';

export const dailyCleanup = worker
  .cron('rate(1 day)')
  .handle(async ({ logger }) => {
    logger.info('Running daily cleanup');
    await cleanupExpiredSessions();
    return { cleaned: true };
  });

Schedule Expressions

Crons accept two schedule formats:

Rate expressions — run at a fixed interval:

worker.cron('rate(5 minutes)')
worker.cron('rate(1 hour)')
worker.cron('rate(7 days)')

Cron expressions — run on a specific schedule (minute, hour, day, month, weekday):

// Every day at midnight
worker.cron('cron(0 0 * * *)')

// Every Monday at 9am
worker.cron('cron(0 9 * * MON)')

// Every 15 minutes during business hours on weekdays
worker.cron('cron(*/15 9-17 * * MON-FRI)')

// First day of every month at noon
worker.cron('cron(0 12 1 * *)')

With Dependencies

export const syncCron = worker
  .cron('rate(30 minutes)')
  .dependsOn([database, cache])
  .handle(async ({ services, logger }) => {
    const staleRecords = await services.database
      .selectFrom('records')
      .where('updated_at', '<', new Date(Date.now() - 3600000))
      .selectAll()
      .execute();

    for (const record of staleRecords) {
      await services.sessions.delete(`record:${record.id}`);
    }

    logger.info({ count: staleRecords.length }, 'Cache invalidated');
    return { invalidated: staleRecords.length };
  });

With Input and Output Schemas

import { z } from 'zod';

export const reportCron = worker
  .cron('cron(0 6 * * MON)')
  .input(z.object({
    reportType: z.enum(['daily', 'weekly', 'monthly']),
  }))
  .output(z.object({
    generatedAt: z.string(),
    rowCount: z.number(),
  }))
  .handle(async ({ input }) => {
    const report = await generateReport(input.reportType);
    return {
      generatedAt: new Date().toISOString(),
      rowCount: report.rows.length,
    };
  });

With Database Access

The worker's database is the cron's db without naming it again; .database(other) overrides it for one cron, and the schedule stays in the worker's:

export const archiveCron = worker
  .cron('cron(0 2 * * *)')
  .handle(async ({ db, logger }) => {
    const cutoff = new Date(Date.now() - 90 * 24 * 3600000); // 90 days

    const result = await db
      .deleteFrom('logs')
      .where('created_at', '<', cutoff)
      .executeTakeFirst();

    logger.info({ deleted: result.numDeletedRows }, 'Archived old logs');
    return { deleted: Number(result.numDeletedRows) };
  });

With Event Publishing

.event(topic, …) publishes after each successful run. For one event per row, publish through the topic's service, which .event() or .dependsOn([topic]) puts in the handler:

export const reminderCron = worker
  .cron('rate(1 hour)')
  .dependsOn([database, reminders])
  .output(z.object({ notified: z.number() }))
  .event(reminders, {
    type: 'reminders.swept',
    payload: (result) => ({ count: result.notified }),
  })
  .handle(async ({ services }) => {
    const users = await services.database
      .selectFrom('users')
      .where('reminder_due', '<', new Date())
      .selectAll()
      .execute();

    await services.reminders.publish(
      users.map((user) => ({ type: 'reminder.due', payload: { userId: user.id } })),
    );

    return { notified: users.length };
  });

Configuration Options

Method	Description
worker.cron(expression)	The cron or rate schedule expression
.input(schema)	Validate the input payload with a StandardSchema
.output(schema)	Validate the return value with a StandardSchema
.dependsOn([...])	Constructs whose clients the handler gets in services
.database(database)	The database db is, in place of the worker's
.event(topic, { type, payload, when? })	Publish to a topic after each successful run
.timeout(ms)	Set the execution timeout in milliseconds (default: 30000)
.memorySize(mb)	Set the memory allocation in MB (AWS Lambda)
.handle(fn)	Define the handler and build the Cron

Project Configuration

Crons are found by the same constructs glob as everything else in gkm.config.ts — there is no per-kind glob.

AWS Lambda Deployment

When the project deploys with SST (deploy: { default: 'sst' }), gkm build compiles each cron into its own Lambda handler under <app>/.gkm/aws/crons/, with its schedule expression in the manifest at .gkm/manifest/aws.ts. fromManifest from @geekmidas/cloud/sst turns it into an EventBridge rule; another IaC tool can read the same manifest.

Telemetry

Telemetry declares what a process emits — never where it goes, which is the stage's deploy.telemetry (see the Telemetry guide):

import { Telemetry } from '@geekmidas/constructs/telemetry';

export const telemetry = new Telemetry('Telemetry', {
  ignorePaths: ['/health', '/ready'],        // no span for these, on every surface
  attributes: { 'service.namespace': 'shop' }, // on every span and log record
});

// passed like the logger — each one is an edge to the node
new RestApi('Api', { path: 'apps/api', defaultAuthorizer: 'none', logger, telemetry });
new Worker('Jobs', { logger, telemetry });
new BetterAuth('Auth', { path: 'apps/auth', database: authDb, telemetry });
new StaticSite('Web', { path: 'apps/web', telemetry });

// one route says something of its own
router.get('/health').telemetry({ ignore: true }).handle(() => ({ ok: true }));
router
  .get('/orders/:id')
  .telemetry({ attributes: { 'app.area': 'orders' } })
  .handle(…);

The node provides OpenTelemetry's own keys — OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_HEADERS, OTEL_SERVICE_NAME, OTEL_TRACES_SAMPLER and OTEL_TRACES_SAMPLER_ARG (TELEMETRY_KEYS in @geekmidas/manifest) — to each process with an edge, and none of them to a site's public values. A process with the edge is built with the SDK, and needs @geekmidas/telescope and its OpenTelemetry peers; one without it loads nothing.

The old Telemetry interface the Lambda adaptors take (onRequestStart/onRequestEnd/onRequestError) is still exported from the package root under that name; the construct is @geekmidas/constructs/telemetry.

Tracing

The constructs record their own spans through the global OpenTelemetry tracer, so they are a no-op unless a provider is registered (the entry of a process given the Telemetry construct registers one when OTEL_EXPORTER_OTLP_ENDPOINT is set):

Span	Kind	Attributes
<operation> <table> (e.g. select orders), or db.query	CLIENT	db.system (postgresql), db.name, db.operation, db.sql.table, server.address, server.port, and db.statement
cron <name>	INTERNAL, a root	gkm.cron.name, gkm.cron.schedule
<Api>.<method> for an ExternalApi client	INTERNAL	gkm.external_api.name, gkm.external_api.method, server.address
<queue> process, <topic> process	CONSUMER	see @geekmidas/events

• Database queries get a span each from the pool a KyselyDatabase opens: pg is bundled into a production entry, where instrumentation cannot hook it, so the client is wrapped explicitly, as the query tags are. The statement text is recorded only for a parameterized query — values passed separately, as Kysely always does — and truncated to 2048 characters. Parameter values never are.
• Each worker job — a queue message, a subscriber's event — runs inside a CONSUMER span that continues the trace of the request that published it, on a server and in a Lambda alike.
• Each cron run is a trace of its own.
• ExternalApi hands handlers its client wrapped so each method call is a span named for the API. The outbound fetch it makes is traced already, and becomes that span's child.

Deployment

Hono Integration

import { HonoEndpoint } from '@geekmidas/constructs/hono';
import { Hono } from 'hono';
import { EnvironmentParser } from '@geekmidas/envkit';
import { ConsoleLogger } from '@geekmidas/logger/console';

const app = new Hono();
const logger = new ConsoleLogger();
const envParser = new EnvironmentParser(process.env);

await HonoEndpoint.fromRoutes(
  ['./src/endpoints/**/*.ts'],
  envParser,
  app,
  logger,
  process.cwd(),
  {
    docsPath: '/__docs',
    openApiOptions: {
      title: 'My API',
      version: '1.0.0',
    },
  }
);

export default app;

AWS Lambda (API Gateway v2)

import { AmazonApiGatewayV2Endpoint } from '@geekmidas/constructs/aws';
import { EnvironmentParser } from '@geekmidas/envkit';
import { getUserEndpoint } from './endpoints/users';

const envParser = new EnvironmentParser(process.env);
const adaptor = new AmazonApiGatewayV2Endpoint(envParser, getUserEndpoint);

export const handler = adaptor.handler;

AWS Lambda (API Gateway v1)

import { AmazonApiGatewayV1Endpoint } from '@geekmidas/constructs/aws';
import { EnvironmentParser } from '@geekmidas/envkit';
import { getUserEndpoint } from './endpoints/users';

const envParser = new EnvironmentParser(process.env);
const adaptor = new AmazonApiGatewayV1Endpoint(envParser, getUserEndpoint);

export const handler = adaptor.handler;

tRPC Integration

The tRPC middlewares now live in @geekmidas/services/trpc. See the services package docs.

Testing

Testing Endpoints

import { TestEndpointAdaptor } from '@geekmidas/constructs/testing';
import { describe, it, expect } from 'vitest';

describe('Login Endpoint', () => {
  it('should set session cookie on successful login', async () => {
    const adaptor = new TestEndpointAdaptor(loginEndpoint);

    const result = await adaptor.request({
      body: {
        email: 'user@example.com',
        password: 'password123',
      },
      headers: {
        'content-type': 'application/json',
      },
      services: {},
    });

    // Check response data
    expect(result.data).toMatchObject({
      id: expect.any(String),
      email: 'user@example.com',
    });

    // Check response metadata
    expect(result.metadata.cookies?.has('session')).toBe(true);
    const sessionCookie = result.metadata.cookies?.get('session');
    expect(sessionCookie?.options).toMatchObject({
      httpOnly: true,
      secure: true,
      sameSite: 'strict',
    });
  });

  it('should read and use request cookies', async () => {
    const adaptor = new TestEndpointAdaptor(profileEndpoint);

    const result = await adaptor.request({
      headers: {
        cookie: 'session=abc123; theme=dark',
      },
      services: {},
    });

    expect(result.data).toMatchObject({
      userId: expect.any(String),
      preferences: {
        theme: 'dark',
      },
    });
  });

  it('should delete session cookie on logout', async () => {
    const adaptor = new TestEndpointAdaptor(logoutEndpoint);

    const result = await adaptor.request({
      headers: {},
      services: {},
    });

    expect(result.data.success).toBe(true);

    const sessionCookie = result.metadata.cookies?.get('session');
    expect(sessionCookie?.value).toBe('');
    expect(sessionCookie?.options?.maxAge).toBe(0);
  });

  it('should return 201 status code for resource creation', async () => {
    const adaptor = new TestEndpointAdaptor(createUserEndpoint);

    const result = await adaptor.request({
      body: {
        name: 'John Doe',
        email: 'john@example.com',
      },
      headers: {},
      services: {},
    });

    expect(result.metadata.status).toBe(201);
    expect(result.metadata.headers?.['Location']).toBe(`/users/${result.data.id}`);
  });

  it('should set custom headers', async () => {
    const adaptor = new TestEndpointAdaptor(downloadEndpoint);

    const result = await adaptor.request({
      params: { id: 'file-123' },
      headers: {},
      services: {},
    });

    expect(result.metadata.headers).toMatchObject({
      'Content-Disposition': expect.stringContaining('attachment'),
      'Cache-Control': 'private, max-age=3600',
    });
  });
});

Test Result Structure:

When using response.send(), the test adaptor returns:

{
  data: T,              // The actual response data
  metadata: {
    headers?: Record<string, string>,
    cookies?: Map<string, { value: string, options?: CookieOptions }>,
    status?: SuccessStatus,
  }
}

Testing Simple Responses (without metadata):

For endpoints that don't use the response builder, the test adaptor returns just the data:

describe('Simple Endpoint', () => {
  it('should return user data', async () => {
    const adaptor = new TestEndpointAdaptor(getUserEndpoint);

    const result = await adaptor.request({
      params: { id: 'user-123' },
      headers: {},
      services: {},
    });

    // Direct data access (no metadata wrapper)
    expect(result).toMatchObject({
      id: 'user-123',
      name: expect.any(String),
    });
  });
});

Capturing Published Events

A topic is a service under its own id, so a recorder passed in services stands in for its publisher and captures what .event(users, …) sent:

const published: { type: string; payload: unknown }[] = [];
const users = {
  async publish(messages: { type: string; payload: unknown }[]) {
    published.push(...messages);
  },
};

await new TestEndpointAdaptor(createUser).request({
  body: { name: 'Ada', email: 'ada@example.com' },
  services: { users },
  headers: { host: 'example.com' },
});

expect(published).toEqual([
  { type: 'user.created', payload: expect.objectContaining({ name: 'Ada' }) },
]);

In a featureTest, published(users) does this for you.

Testing WebSocket routes

A WebSocketApi route is tested the way an endpoint is: by calling it. route.request() runs its handler through the surface's core — no server, no socket, no auth server.

// connect: resolves with the connection's `data`, or rejects with the
// HttpError that refused it (an UnauthorizedError stays a 401)
const data = await connect.request({
  connectionId: 'a.1',              // optional; `test.<n>` when absent
  headers: { cookie: 'session=…' }, // what `.auth()` verifies without a session
  query: { room: 'kitchen' },
  session: { user: { email } },     // what `.auth()` would have resolved, or null
  services: { database: trx },      // overrides; the rest are registered
});

// a message route: `body` is encoded as a frame and checked by the route's
// `body` schema; resolves with the handler's return
const reply = await say.request({
  connection: { id: 'a.1', data },
  body: { text: 'hi' },
  services: { database: trx },
});

// …and every frame written back — each `reply()` and the reply itself
const { returned, frames } = await say.fullRequest({ body: { text: 'hi' } });

await disconnect.request({ connection: { id: 'a.1', data }, code: 1000 });

A body the schema refuses rejects with InvalidWebSocketRequest (its issues are the schema's). What the handler threw rejects as it is. A disconnect handler's error, which the core only logs, rejects too. request() is typed by the route: body by its schema's input, services by its .dependsOn(), and a connect route resolves with its handler's return.

Recording what was sent

surface.testPublisher() returns a publisher of the type services.<surface> is, which records instead of sending. Hand it to an endpoint that .dependsOn([realtime]), or to a route as sockets:

const realtime = surface.testPublisher();

await new TestEndpointAdaptor(sendMessage).request({
  body: { to: ['a.1'], text: 'hi' },
  services: { realtime },
  headers: {},
});

expect(realtime.sent).toEqual([
  {
    ids: ['a.1'],
    action: 'chat.message.created',
    data: { chatId: 'c1', text: 'hi' },
    // the frame a client parses: flat here, under `frame.discriminator`
    frame: { type: 'chat.message.created', chatId: 'c1', text: 'hi' },
  },
]);
realtime.broadcasts; // { action, data, except, frame }[]
realtime.closed;     // { ids, code, reason }[]
realtime.clear();

Each send is checked by the declared messages and encoded by frame, as deployed: an undeclared action or invalid data rejects (UnknownWebSocketMessage, InvalidWebSocketMessage), and data is what the schema returned. In a featureTest, sockets(realtime) is the recorder every handler in the test was handed.

Frontend Integration Testing with MSW

Use createMswHandlers to test frontend code against real backend endpoints — with full validation, authorization, and session handling — without running an HTTP server.

How It Works

1. The API app exports its endpoint constructs via @myapp/api/endpoints
2. The web app creates MSW handlers from those endpoints
3. Each test registers an isolated context (services, database transaction, etc.)
4. MSW intercepts fetch requests and routes them through Hono's app.request()

Setup

API app — export endpoints from package.json:

{
  "exports": {
    "./endpoints": "./.gkm/server/endpoints.ts"
  }
}

The endpoints array is auto-generated by the CLI when running gkm dev or gkm build.

Web app — create the MSW server:

// apps/web/src/test/msw-server.ts
import { createMswHandlers, TEST_CONTEXT_HEADER } from '@geekmidas/constructs/testing';
import { endpoints } from '@myapp/api/endpoints';
import { setupServer } from 'msw/node';

const { handlers, registerContext } = createMswHandlers(endpoints, {
  baseURL: 'http://localhost:3000',
});

export const mswServer = setupServer(...handlers);
export { registerContext, TEST_CONTEXT_HEADER };

Writing Tests

Each test registers its own context with registerContext(). The context ID is sent via the x-test-context-id header so concurrent tests are isolated.

// apps/web/src/__tests__/users.spec.tsx
import { createApi } from '@myapp/client/api';
import { render, screen, waitFor } from '@testing-library/react';
import { afterAll, afterEach, beforeAll, expect, it } from 'vitest';
import { mswServer, registerContext, TEST_CONTEXT_HEADER } from '../test/msw-server';

beforeAll(() => mswServer.listen({ onUnhandledRequest: 'bypass' }));
afterEach(() => mswServer.resetHandlers());
afterAll(() => mswServer.close());

it('should render users list', async ({ db }) => {
  const contextId = crypto.randomUUID();

  registerContext(contextId, {
    database: db,  // transaction-wrapped db for test isolation
    services: {
      auth: mockAuth,
      storage: mockStorage,
    },
  });

  const api = createApi({
    baseURL: 'http://localhost:3000',
    headers: { [TEST_CONTEXT_HEADER]: contextId },
  });

  const users = await api('GET /users');
  expect(users).toBeDefined();
});

Context Isolation

Each call to registerContext() creates an isolated ServiceDiscovery instance. This means:

• Database transactions — pass a transaction-wrapped db so test data rolls back automatically
• Service overrides — each test can provide different mock services
• Concurrent safety — tests running in parallel each get their own context via the header

registerContext(contextId, {
  // Services keyed by serviceName
  services: {
    auth: mockAuthService,
    storage: mockStorageService,
    notification: mockNotificationService,
    // A topic is a service under its own name, so a recorder here
    // captures what endpoints publish to it with .event(users, …)
    users: usersRecorder,
  },
  // Database instance (from .database() on the factory)
  database: transactionDb,
  // Audit storage instance (from .auditor() on the factory)
  auditorStorage: mockAuditStorage,
});

API

Export	Description
createMswHandlers(endpoints, options)	Creates MSW handlers from endpoint constructs. Returns { handlers, registerContext }
TEST_CONTEXT_HEADER	The header name (x-test-context-id) used to identify test contexts

CreateMswHandlersOptions:

Property	Type	Description
baseURL	string	Base URL the client fetches from (e.g., 'http://localhost:3000')

RegisterContextOptions:

Property	Type	Description
services	Record<string, unknown>	Service instances keyed by serviceName
database	unknown	Database instance (required when endpoints use .database())
auditorStorage	unknown	Audit storage instance (required when endpoints use .auditor())

WebSocket clients

createMswWebSocketHandlers does the same for a WebSocketApi: MSW's ws.link() intercepts the client's WebSocket, and each connection is handed to the surface's real routes. Client code — its reconnects, its decoding by frame — runs unchanged, and an endpoint's push reaches it in the same process:

import {
  createMswWebSocketHandlers,
  TEST_CONTEXT_PARAM,
} from '@geekmidas/constructs/testing';

const ws = createMswWebSocketHandlers(realtime, [routes], {
  url: 'wss://realtime.test',
});
server.use(...ws.handlers);

it('shows a new message', async () => {
  const id = crypto.randomUUID();
  ws.registerContext(id, { session: { user: { email } }, services: { database: trx } });
  const socket = new WebSocket(`wss://realtime.test/?${TEST_CONTEXT_PARAM}=${id}`);

  // endpoint → in-process channel → the surface → MSW → socket.onmessage
  await new TestEndpointAdaptor(sendMessage).request({
    services: { realtime: ws.publisher },
    // …
  });
});

• The context travels as a query parameter, since a browser WebSocket cannot set headers. The adaptor strips it before connect() sees query.
• There is no handshake, so the session is the context's (no context is no session). A refused connection is closed with 4000 + status, so a 401 is 4401 (REFUSED_CLOSE_BASE). A context nobody registered closes it with 1011.
• ws.publisher sends to these connections. ws.connectionString (a memory:// channel of its own) is the same channel as <ID>_PUBLISHER_CONNECTION_STRING, for code that registers services.<surface> itself. ws.connections lists the open ones.

The Origin check, the handshake headers and the heartbeat need a real upgrade, and stay covered by the Node adaptor's own tests.

OpenAPI Documentation

Endpoints automatically generate OpenAPI 3.1 documentation:

import { Endpoint } from '@geekmidas/constructs/endpoints';

const schema = await Endpoint.buildOpenApiSchema(
  [getUserEndpoint, createUserEndpoint],
  {
    title: 'User API',
    version: '1.0.0',
    description: 'API for managing users',
  }
);

The Hono adapter automatically serves OpenAPI docs at /docs (configurable).