@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

bash

pnpm add @geekmidas/constructs

Features

✅ 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`	HTTP endpoint builder (`e`, `EndpointFactory`) and types
`/functions`	Cloud function builder (`f`)
`/crons`	Scheduled task builder (`c`) and AWS adaptor (`AWSScheduledFunction`)
`/subscribers`	Event subscriber builder (`s`) — topic fan-out
`/queue`	Queue worker builder (`q`) — point-to-point queue + single 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`)

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.

Basic Usage

Creating an Endpoint

typescript

import { e } from '@geekmidas/constructs/endpoints';
import { z } from 'zod';

export const getUserEndpoint = e
  .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:

typescript

const createUserEndpoint = e
  .post('/users')
  .body(z.object({
    name: z.string(),
    email: z.string().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

typescript

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');

Service Pattern

Define services as object literals with dependency injection:

typescript

import type { Service } from '@geekmidas/constructs';
import type { EnvironmentParser } from '@geekmidas/envkit';

const databaseService = {
  serviceName: 'database' as const,
  async register(envParser: EnvironmentParser<{}>) {
    const config = envParser.create((get) => ({
      url: get('DATABASE_URL').string()
    })).parse();

    const db = await createConnection(config.url);
    return db;
  }
} satisfies Service<'database', Database>;

// Use in endpoint
const endpoint = e
  .get('/data')
  .services([databaseService])
  .handle(async ({ services }) => {
    const db = services.database;
    return await db.query('...');
  });

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:

typescript

const endpoint = e
  .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 the factory:

typescript

const sessionRouter = e
  .services([AuthService])
  .session(async ({ cookie, services }) => {
    const sessionToken = cookie('session');
    if (!sessionToken) {
      throw new ForbiddenError('No active session');
    }

    return await services.auth.verifySession(sessionToken);
  });

// 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

typescript

import { e } from '@geekmidas/constructs/endpoints';
import { z } from 'zod';

const loginEndpoint = e
  .post('/auth/login')
  .body(z.object({
    email: z.string().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

typescript

const logoutEndpoint = e
  .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:

typescript

const downloadEndpoint = e
  .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:

typescript

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

const createEndpoint = e
  .post('/users')
  .body(z.object({
    name: z.string(),
    email: z.string().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:

typescript

const apiEndpoint = e
  .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:

typescript

const getEndpoint = e
  .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:

typescript

const uploadEndpoint = e
  .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,
      });
  });

Authorization and Sessions

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

typescript

import { e } from '@geekmidas/constructs/endpoints';
import { ForbiddenError } from '@geekmidas/errors';

// Create a factory with services and database
const r = e
  .services([AuthService])
  .database(DatabaseService);

// Create a session-enabled router
const sessionRouter = r
  .session(async ({ header, services, db, cookie }) => {
    const token = cookie('session_token') || header('authorization');
    if (!token) {
      throw new ForbiddenError('No active session');
    }

    return await services.auth.verifyToken(token);
  });

// 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:

typescript

// 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' };
  });

typescript

// 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 };
  });

typescript

// 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

typescript

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

const rateLimitedEndpoint = e
  .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():

typescript

import { e } from '@geekmidas/constructs/endpoints';
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 = e
  .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:

typescript

const endpoint = e
  .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:

typescript

import { z } from 'zod';

const createUserEndpoint = e
  .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:

typescript

const updateUserEndpoint = e
  .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:

typescript

const transferEndpoint = e
  .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 an EndpointFactory so all endpoints inherit the configuration:

typescript

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

const api = new EndpointFactory()
  .services([databaseService, auditStorageService])
  .session(extractSession)
  .authorizer('jwt')
  .auditor(auditStorageService)
  .actor(({ session }) => ({
    id: session.sub,
    type: 'user',
  }));

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

const deleteUser = api
  .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.

typescript

const api = new EndpointFactory()
  .services([databaseService, auditStorageService])
  .database(databaseService)
  .auditor(auditStorageService)
  .actor(({ session }) => ({ id: session.sub, type: 'user' }));

const endpoint = api
  .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 service, the framework automatically wraps the handler and 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:

typescript

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

const api = new EndpointFactory()
  .services([databaseService])
  .database(databaseService)
  .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:

typescript

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.

typescript

// 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:

typescript

// 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:

typescript

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

const endpoint = e
  .get('/orders')
  .services([databaseService])
  .database(databaseService)
  .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:

sql

-- 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:

typescript

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

Endpoints can declare events that are published automatically after the handler returns:

typescript

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

const createOrder = e
  .post('/orders')
  .services([databaseService])
  .publisher(eventPublisherService)
  .body(orderSchema)
  .output(orderResponseSchema)
  .events([
    {
      type: 'order.created',
      payload: (response) => ({
        orderId: response.id,
        total: response.total,
      }),
    },
  ])
  .handle(async ({ body, services }) => {
    return await services.database
      .insertInto('orders')
      .values(body)
      .returningAll()
      .executeTakeFirstOrThrow();
  });

You can also publish events manually inside the handler using the publish function:

typescript

const endpoint = e
  .post('/transfers')
  .publisher(eventPublisherService)
  .handle(async ({ body, publish }) => {
    const result = await processTransfer(body);

    await publish('transfer.completed', {
      transferId: result.id,
      amount: result.amount,
    });

    return result;
  });

The .publisher() method accepts an event publisher service that implements the @geekmidas/events publisher interface. See Defining a Publisher Service below for how to create one.

Defining a Publisher Service

The publisher service follows the standard service pattern. It wraps a Publisher from @geekmidas/events and uses EVENT_PUBLISHER_CONNECTION_STRING to connect to the correct backend:

typescript

import type { Service } from '@geekmidas/services';
import type { EventPublisher, PublishableMessage } from '@geekmidas/events';
import { Publisher } from '@geekmidas/events';

// 1. Define your event types
type AppEvents =
  | PublishableMessage<'user.created', { userId: string; email: string }>
  | PublishableMessage<'order.placed', { orderId: string; total: number }>
  | PublishableMessage<'notification.sent', { userId: string; type: string }>;

// 2. Create the publisher service
const eventPublisherService = {
  serviceName: 'eventPublisher' as const,
  async register(envParser) {
    const config = envParser.create((get) => ({
      connectionString: get('EVENT_PUBLISHER_CONNECTION_STRING').string(),
    })).parse();

    return Publisher.fromConnectionString<AppEvents>(config.connectionString);
  },
} satisfies Service<'eventPublisher', EventPublisher<AppEvents>>;

// 3. Use with endpoints
const createUser = e
  .post('/users')
  .publisher(eventPublisherService)
  .output(userSchema)
  .event({
    type: 'user.created',
    payload: (response) => ({ userId: response.id, email: response.email }),
  })
  .handle(async ({ body }) => {
    return await createUser(body);
  });

The connection string protocol determines which backend is used (pgboss://, rabbitmq://, sns://, sqs://, basic://). When using services.events in your workspace config, the CLI auto-generates this env var.

Factory-Level Publisher

Set a default publisher on the factory so all endpoints inherit it:

typescript

const api = new EndpointFactory()
  .services([databaseService])
  .publisher(eventPublisherService);

// All endpoints can use .event() without specifying .publisher()
const createOrder = api
  .post('/orders')
  .event({
    type: 'order.placed',
    payload: (response) => ({ orderId: response.id, total: response.total }),
  })
  .handle(async ({ body }) => { /* ... */ });

Event Subscribers

The s builder creates event subscribers that react to published events. Subscribers receive batches of events and can use services, logging, and even publish follow-up events.

Basic Subscriber

typescript

import { s } from '@geekmidas/constructs/subscribers';

export const onUserCreated = s
  .subscribe('user.created')
  .handle(async ({ events, logger }) => {
    for (const event of events) {
      logger.info({ userId: event.payload.userId }, 'Processing new user');
      await sendWelcomeEmail(event.payload.email);
    }
  });

With Services

typescript

export const onOrderPlaced = s
  .services([databaseService, emailService])
  .subscribe('order.placed')
  .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.send('order-confirmation', {
        to: order.customerEmail,
        props: { orderId: order.id, total: order.total },
      });

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

Subscribing to Multiple Events

typescript

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

With Publisher (Chaining Events)

Subscribers can publish follow-up events using a publisher service:

typescript

export const onUserCreated = s
  .publisher(eventPublisherService)
  .subscribe('user.created')
  .handle(async ({ events, publish }) => {
    for (const event of events) {
      await createUserProfile(event.payload.userId);

      // Publish a follow-up event
      await publish('notification.sent', {
        userId: event.payload.userId,
        type: 'welcome',
      });
    }
  });

Configuration Options

Method	Description
`.subscribe(events)`	Event type(s) to listen for (string or string array)
`.services(services)`	Inject services into the handler context
`.publisher(service)`	Set publisher service (provides `publish` in context)
`.output(schema)`	Validate the return value with a StandardSchema
`.logger(logger)`	Set a custom logger instance
`.timeout(ms)`	Set execution timeout in milliseconds
`.handle(fn)`	Define the handler function and build the `Subscriber` instance

Handler Context

The handler receives:

typescript

{
  events: Array<{ type: string; payload: T }>;  // Batch of events (type-safe)
  services: ServiceRecord;                       // Registered services
  logger: Logger;                                // Logger instance
  publish: (type, payload) => Promise<void>;     // If .publisher() is set
}

How Subscribers Run

Development (gkm dev): The CLI scans your routes for exported Subscriber instances, generates polling code, and starts listening on server startup. It uses EVENT_SUBSCRIBER_CONNECTION_STRING to connect. See Events: Dev Server for details.

Production (AWS Lambda): Each subscriber is compiled into a Lambda handler via AWSLambdaSubscriber. The handler parses SQS/SNS events, filters to subscribed types, and invokes the handler. Configure event source mappings in your IaC tool.

typescript

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;

Production (Server): When building with gkm build --provider server, subscribers are included in the generated server and poll using the configured connection string.

Testing Subscribers

typescript

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

describe('onUserCreated subscriber', () => {
  it('should send welcome email', async () => {
    const adaptor = new TestSubscriberAdaptor(onUserCreated);

    const result = await adaptor.invoke({
      events: [
        {
          type: 'user.created',
          payload: { userId: '123', email: 'test@example.com' },
        },
      ],
    });

    // Assert side effects (email sent, etc.)
  });
});

Project Configuration

typescript

import { defineConfig } from '@geekmidas/cli/config';

export default defineConfig({
  routes: './src/endpoints/**/*.ts',
  subscribers: './src/subscribers/**/*.ts',
  envParser: './src/config/env',
  logger: './src/logger',
});

TIP

Subscribers can also be co-located with endpoints in the same route files. The CLI discovers all exported Subscriber instances regardless of file location.

Queues

A queue is point-to-point work distribution: a queue and its single consumer. Where a subscriber (s) is topic fan-out — many subscribers each filtering a stream by subscribedEvents — a queue (q) drains every message of its one typed message and hands it to exactly one handler. Reach for q for job/task processing (send an order to be fulfilled), and s for broadcasting domain events (notify everyone a user was created).

Basic Queue

typescript

import { q } from '@geekmidas/constructs/queue';
import { z } from 'zod';

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

With Services

Services are an array (the same shape endpoints/subscribers use) and are sniffed for the env vars they require — those flow into the manifest so infra provisions them:

typescript

export const ordersQueue = q
  .queue('orders')
  .services([databaseService])
  .message(z.object({ orderId: z.string() }))
  .handle(async ({ messages, services }) => {
    for (const { orderId } of messages) {
      await services.database.fulfil(orderId);
    }
  });

Sending to a Queue (the auto-publisher)

Every queue exposes a publisher — a ready-to-inject Service typed to the queue's message. Any endpoint, function, or other worker connects to the queue by dropping it into .services([...]):

typescript

import { e } from '@geekmidas/constructs/endpoints';
import { ordersQueue } from './queues/orders';

export const createOrder = e
  .post('/orders')
  .body(z.object({ sku: z.string() }))
  .services([ordersQueue.publisher]) // producer side
  .handle(async ({ body, services }) => {
    const orderId = crypto.randomUUID();
    // serviceName is `<name>Publisher` → here `ordersPublisher`
    await services.ordersPublisher.publish([
      { type: 'orders', payload: { orderId } },
    ]);
    return { orderId };
  });

The publisher reads ORDERS_PUBLISHER_CONNECTION_STRING and selects its transport from the URL protocol — pgboss:// locally, sqs:// when deployed — so the same code publishes to Postgres in dev and SQS in prod. Because it's a Service, the connection-string requirement is sniffed into the manifest and infra links exactly that queue with least privilege. Multiple queues never collide: each gets its own <NAME>_PUBLISHER_CONNECTION_STRING.

Configuration Options

Method	Description
`.queue(name)`	Queue name — drives the infra queue and its `<NAME>_*` env vars
`.message(schema)`	The typed message (job) payload
`.services([...])`	Services available to the handler (sniffed for env vars)
`.timeout(ms)`	Handler timeout (default `30000`)
`.batchSize(n)`	SQS event-source batch size (deployed)
`.fifo()`	Mark the queue as FIFO (deployed)
`.logger(logger)`	Custom logger

How Queues Run

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

Development (gkm dev): The CLI generates a setupQueues() pg-boss poller that runs in-process alongside the Hono server. Each queue subscribes by its name on the shared EVENT_SUBSCRIBER_CONNECTION_STRING (pg-boss routes by name), so a producer publishing with pgboss:// reaches it. No SQS or Lambda required locally.

Production (AWS Lambda): Each queue is compiled into a Lambda handler via AWSLambdaQueue, backed by an SQS event-source mapping. 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.

typescript

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

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

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

typescript

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

describe('ordersQueue', () => {
  it('fulfils each order in the batch', async () => {
    const adaptor = new TestQueueAdaptor(ordersQueue);

    const result = await adaptor.invoke({
      messages: [{ orderId: '1' }, { orderId: '2' }],
    });

    // Assert side effects (orders fulfilled, etc.)
  });
});

Project Configuration

typescript

import { defineConfig } from '@geekmidas/cli/config';

export default defineConfig({
  routes: './src/endpoints/**/*.ts',
  subscribers: './src/subscribers/**/*.ts',
  queues: './src/queues/**/*.ts',
  envParser: './src/config/env',
  logger: './src/logger',
});

Cron Jobs

The c builder creates scheduled tasks that run on a cron schedule. Cron jobs extend the same function builder as cloud functions, so they support services, input/output schemas, logging, event publishing, and database access.

Basic Cron

typescript

import { c } from '@geekmidas/constructs/crons';

export const dailyCleanup = c
  .schedule('cron(0 0 * * *)')
  .handle(async ({ logger }) => {
    logger.info('Running daily cleanup');
    await cleanupExpiredSessions();
    return { cleaned: true };
  });

Schedule Expressions

.schedule() takes a cron(…) in either of two forms, told apart by the number of fields:

	Unix (5 fields)	AWS (6 fields)
Fields	`min hour day-of-month month day-of-week`	the same, then `year` (1970–2199)
Day-of-week	`0`–`7` (SUN is `0` or `7`), or `SUN`–`SAT`	`1`–`7` (SUN is `1`), or `SUN`–`SAT`
`?`	not allowed	required in exactly one of the two day fields
Both day fields restricted	allowed: runs on either	not allowed
`L`, `W`, `#`	not allowed	`L`, `L-n`, `LW`, `15W` in day-of-month; `5L`, `6#3` in day-of-week

Within each field, both forms accept *, values (a leading zero is fine), ranges a-b, steps /n and comma lists. Months take JAN–DEC.

typescript

// Unix
c.schedule('cron(0 0 * * *)')               // every day at midnight
c.schedule('cron(0 9 * * MON)')             // every Monday at 9am
c.schedule('cron(*/15 9-17 * * MON-FRI)')   // every 15 minutes, business hours

// AWS
c.schedule('cron(0 12 1 * ? *)')            // first of the month at noon
c.schedule('cron(0 9 ? * 6#3 *)')           // third Friday of the month at 9am
c.schedule('cron(0 18 L * ? *)')            // last day of the month at 6pm

Both forms are checked at compile time, so cron(0 25 * * *) or a ? in a Unix expression doesn't type-check. A schedule built at runtime can be cast to ScheduleExpression; gkm build and gkm dev validate every schedule and fail, listing each invalid cron with the field and token that failed.

Where it runs

The manifest always records the AWS form, since EventBridge reads only that, so gkm build converts a Unix expression:

Unix	AWS
—	a year of `*` is appended
both day fields `*`	day-of-week becomes `?`
one day field restricted	the other becomes `?`
both day fields restricted	approximate: day-of-week is kept and day-of-month becomes `?`, so the cron runs on fewer days. The build warns.
day-of-week numbers `0`–`7`	shifted to `1`–`7`. Names are unchanged.

toUnixCron from @geekmidas/manifest goes the other way, for runners that read Unix cron: ? becomes *, day-of-week numbers shift back, L, nL and n#k are kept, and a specific year is dropped with a warning. W, LW and L-n have no Unix equivalent and fail.

cronPeriod from @geekmidas/manifest gives the longest gap between runs of the AWS form, in seconds: the period a heartbeat should expect. A weekdays-only cron's period is the Friday-to-Monday gap, so it doesn't alert at the weekend.

typescript

import { cronPeriod } from '@geekmidas/manifest';

cronPeriod('cron(0 9 * * MON-FRI)'); // 259200, three days

Rate expressions

rate(…) isn't supported. Write the cron it stands for:

Instead of	Write
`rate(5 minutes)`	`cron(/5 * * *)`
`rate(1 hour)`	`cron(0 * * * *)`
`rate(6 hours)`	`cron(0 /6 * *)`
`rate(1 day)`	`cron(0 0 * * *)`

A rate that doesn't divide its unit, such as rate(7 hours), has no cron equivalent: choose the times it should run.

Uptime

.uptime() reports each run to a heartbeat, so a cron that stops running is noticed: the heartbeat service alerts when no ping arrives in time.

typescript

import { c } from '@geekmidas/constructs/crons';
import { heartbeat } from '@geekmidas/constructs/monitors';

export const nightlyReport = c
  .schedule('cron(0 8 * * MON-FRI)')
  .uptime(heartbeat.betterStack('MONITORING_URL'))
  .handle(async () => {
    // ...
  });

At runtime the cron requests the URL in MONITORING_URL when it succeeds, and {url}/fail when it fails. If the variable is unset, or the service is down, the run goes on without it.

gkm build records it on the cron's manifest entry, with the period the heartbeat should expect: the longest gap between runs, so this weekdays-only cron expects the three-day weekend gap.

typescript

uptime: { type: 'heartbeat', provider: 'betterStack', env: 'MONITORING_URL', period: 259200 }

Infrastructure creates the heartbeat from that entry and sets its URL in MONITORING_URL.

.uptime() is for crons only. A function or endpoint runs when it's called, so there's no time a ping is due, and its failures are already logged and traced where they happen.

With Services

typescript

export const onOrderPlaced = s
  .services([databaseService, emailService])
  .subscribe('order.placed')
  .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.send('order-confirmation', {
        to: order.customerEmail,
        props: { orderId: order.id, total: order.total },
      });

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

Subscribing to Multiple Events

typescript

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

With Publisher (Chaining Events)

Subscribers can publish follow-up events using a publisher service:

typescript

export const onUserCreated = s
  .publisher(eventPublisherService)
  .subscribe('user.created')
  .handle(async ({ events, publish }) => {
    for (const event of events) {
      await createUserProfile(event.payload.userId);

      // Publish a follow-up event
      await publish('notification.sent', {
        userId: event.payload.userId,
        type: 'welcome',
      });
    }
  });

Configuration Options

Method	Description
`.subscribe(events)`	Event type(s) to listen for (string or string array)
`.services(services)`	Inject services into the handler context
`.publisher(service)`	Set publisher service (provides `publish` in context)
`.output(schema)`	Validate the return value with a StandardSchema
`.logger(logger)`	Set a custom logger instance
`.timeout(ms)`	Set execution timeout in milliseconds
`.handle(fn)`	Define the handler function and build the `Subscriber` instance

Handler Context

The handler receives:

typescript

{
  events: Array<{ type: string; payload: T }>;  // Batch of events (type-safe)
  services: ServiceRecord;                       // Registered services
  logger: Logger;                                // Logger instance
  publish: (type, payload) => Promise<void>;     // If .publisher() is set
}

How Subscribers Run

typescript

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;

Production (Server): When building with gkm build --provider server, subscribers are included in the generated server and poll using the configured connection string.

Testing Subscribers

typescript

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

describe('onUserCreated subscriber', () => {
  it('should send welcome email', async () => {
    const adaptor = new TestSubscriberAdaptor(onUserCreated);

    const result = await adaptor.invoke({
      events: [
        {
          type: 'user.created',
          payload: { userId: '123', email: 'test@example.com' },
        },
      ],
    });

    // Assert side effects (email sent, etc.)
  });
});

Project Configuration

typescript

import { defineConfig } from '@geekmidas/cli/config';

export default defineConfig({
  routes: './src/endpoints/**/*.ts',
  subscribers: './src/subscribers/**/*.ts',
  envParser: './src/config/env',
  logger: './src/logger',
});

TIP

Subscribers can also be co-located with endpoints in the same route files. The CLI discovers all exported Subscriber instances regardless of file location.

Queues

Basic Queue

typescript

import { q } from '@geekmidas/constructs/queue';
import { z } from 'zod';

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

With Services

Services are an array (the same shape endpoints/subscribers use) and are sniffed for the env vars they require — those flow into the manifest so infra provisions them:

typescript

export const ordersQueue = q
  .queue('orders')
  .services([databaseService])
  .message(z.object({ orderId: z.string() }))
  .handle(async ({ messages, services }) => {
    for (const { orderId } of messages) {
      await services.database.fulfil(orderId);
    }
  });

Sending to a Queue (the auto-publisher)

Every queue exposes a publisher — a ready-to-inject Service typed to the queue's message. Any endpoint, function, or other worker connects to the queue by dropping it into .services([...]):

typescript

import { e } from '@geekmidas/constructs/endpoints';
import { ordersQueue } from './queues/orders';

export const createOrder = e
  .post('/orders')
  .body(z.object({ sku: z.string() }))
  .services([ordersQueue.publisher]) // producer side
  .handle(async ({ body, services }) => {
    const orderId = crypto.randomUUID();
    // serviceName is `<name>Publisher` → here `ordersPublisher`
    await services.ordersPublisher.publish([
      { type: 'orders', payload: { orderId } },
    ]);
    return { orderId };
  });

Configuration Options

Method	Description
`.queue(name)`	Queue name — drives the infra queue and its `<NAME>_*` env vars
`.message(schema)`	The typed message (job) payload
`.services([...])`	Services available to the handler (sniffed for env vars)
`.timeout(ms)`	Handler timeout (default `30000`)
`.batchSize(n)`	SQS event-source batch size (deployed)
`.fifo()`	Mark the queue as FIFO (deployed)
`.logger(logger)`	Custom logger

How Queues Run

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

typescript

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

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

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

typescript

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

describe('ordersQueue', () => {
  it('fulfils each order in the batch', async () => {
    const adaptor = new TestQueueAdaptor(ordersQueue);

    const result = await adaptor.invoke({
      messages: [{ orderId: '1' }, { orderId: '2' }],
    });

    // Assert side effects (orders fulfilled, etc.)
  });
});

Project Configuration

typescript

import { defineConfig } from '@geekmidas/cli/config';

export default defineConfig({
  routes: './src/endpoints/**/*.ts',
  subscribers: './src/subscribers/**/*.ts',
  queues: './src/queues/**/*.ts',
  envParser: './src/config/env',
  logger: './src/logger',
});

Cron Jobs

Basic Cron

typescript

import { c } from '@geekmidas/constructs/crons';

export const dailyCleanup = c
  .schedule('cron(0 0 * * *)')
  .handle(async ({ logger }) => {
    logger.info('Running daily cleanup');
    await cleanupExpiredSessions();
    return { cleaned: true };
  });

Schedule Expressions

.schedule() takes a cron(…) in either of two forms, told apart by the number of fields:

	Unix (5 fields)	AWS (6 fields)
Fields	`min hour day-of-month month day-of-week`	the same, then `year` (1970–2199)
Day-of-week	`0`–`7` (SUN is `0` or `7`), or `SUN`–`SAT`	`1`–`7` (SUN is `1`), or `SUN`–`SAT`
`?`	not allowed	required in exactly one of the two day fields
Both day fields restricted	allowed: runs on either	not allowed
`L`, `W`, `#`	not allowed	`L`, `L-n`, `LW`, `15W` in day-of-month; `5L`, `6#3` in day-of-week

Within each field, both forms accept *, values (a leading zero is fine), ranges a-b, steps /n and comma lists. Months take JAN–DEC.

typescript

// Unix
c.schedule('cron(0 0 * * *)')               // every day at midnight
c.schedule('cron(0 9 * * MON)')             // every Monday at 9am
c.schedule('cron(*/15 9-17 * * MON-FRI)')   // every 15 minutes, business hours

// AWS
c.schedule('cron(0 12 1 * ? *)')            // first of the month at noon
c.schedule('cron(0 9 ? * 6#3 *)')           // third Friday of the month at 9am
c.schedule('cron(0 18 L * ? *)')            // last day of the month at 6pm

Where it runs

The manifest always records the AWS form, since EventBridge reads only that, so gkm build converts a Unix expression:

Unix	AWS
—	a year of `*` is appended
both day fields `*`	day-of-week becomes `?`
one day field restricted	the other becomes `?`
both day fields restricted	approximate: day-of-week is kept and day-of-month becomes `?`, so the cron runs on fewer days. The build warns.
day-of-week numbers `0`–`7`	shifted to `1`–`7`. Names are unchanged.

typescript

import { cronPeriod } from '@geekmidas/manifest';

cronPeriod('cron(0 9 * * MON-FRI)'); // 259200, three days

Rate expressions

rate(…) isn't supported. Write the cron it stands for:

Instead of	Write
`rate(5 minutes)`	`cron(/5 * * *)`
`rate(1 hour)`	`cron(0 * * * *)`
`rate(6 hours)`	`cron(0 /6 * *)`
`rate(1 day)`	`cron(0 0 * * *)`

A rate that doesn't divide its unit, such as rate(7 hours), has no cron equivalent: choose the times it should run.

Uptime

.uptime() reports each run to a heartbeat, so a cron that stops running is noticed: the heartbeat service alerts when no ping arrives in time.

typescript

import { c } from '@geekmidas/constructs/crons';
import { heartbeat } from '@geekmidas/constructs/monitors';

export const nightlyReport = c
  .schedule('cron(0 8 * * MON-FRI)')
  .uptime(heartbeat.betterStack('MONITORING_URL'))
  .handle(async () => {
    // ...
  });

At runtime the cron requests the URL in MONITORING_URL when it succeeds, and {url}/fail when it fails. If the variable is unset, or the service is down, the run goes on without it.

gkm build records it on the cron's manifest entry, with the period the heartbeat should expect: the longest gap between runs, so this weekdays-only cron expects the three-day weekend gap.

typescript

uptime: { type: 'heartbeat', provider: 'betterStack', env: 'MONITORING_URL', period: 259200 }

Infrastructure creates the heartbeat from that entry and sets its URL in MONITORING_URL.

.uptime() is for crons only. A function or endpoint runs when it's called, so there's no time a ping is due, and its failures are already logged and traced where they happen.

With Services

typescript

import { c } from '@geekmidas/constructs/crons';

export const syncCron = c
  .services([databaseService, cacheService])
  .schedule('cron(*/30 * * * *)')
  .handle(async ({ services, logger }) => {
    const db = services.database;
    const cache = services.cache;

    const staleRecords = await db
      .selectFrom('records')
      .where('updated_at', '<', new Date(Date.now() - 3600000))
      .selectAll()
      .execute();

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

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

With Input and Output Schemas

typescript

import { c } from '@geekmidas/constructs/crons';
import { z } from 'zod';

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

With Database Access

typescript

import { c } from '@geekmidas/constructs/crons';

export const archiveCron = c
  .services([databaseService])
  .database(databaseService)
  .schedule('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

typescript

import { c } from '@geekmidas/constructs/crons';

export const reminderCron = c
  .services([databaseService])
  .publisher(eventPublisherService)
  .schedule('cron(0 * * * *)')
  .handle(async ({ services, publish }) => {
    const users = await services.database
      .selectFrom('users')
      .where('reminder_due', '<', new Date())
      .selectAll()
      .execute();

    for (const user of users) {
      await publish('reminder.due', { userId: user.id });
    }

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

Configuration Options

Method	Description
`.schedule(expression)`	Set the cron schedule, Unix or AWS form
`.input(schema)`	Validate the input payload with a StandardSchema
`.output(schema)`	Validate the return value with a StandardSchema
`.services(services)`	Inject services into the handler context
`.database(service)`	Set the database service (provides `db` in context)
`.publisher(service)`	Set the event publisher service (provides `publish` in context)
`.logger(logger)`	Set a custom logger instance
`.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 function and build the `Cron` instance

Project Configuration

typescript

import { defineConfig } from '@geekmidas/cli/config';

export default defineConfig({
  routes: './src/endpoints/**/*.ts',
  crons: './src/crons/**/*.ts',    // glob pattern for cron files
  envParser: './src/config/env',
  logger: './src/logger',
});

AWS Lambda Deployment

When building with gkm build --provider aws-lambda, each cron is compiled into a separate Lambda handler with its schedule expression included in the build manifest. Use the manifest to configure EventBridge rules in your IaC tool (SST, CDK, Terraform, etc.).

Deployment

Hono Integration

typescript

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)

typescript

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)

typescript

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

typescript

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:

typescript

{
  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:

typescript

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),
    });
  });
});

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

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

Setup

API app — export endpoints from package.json:

json

{
  "exports": {
    "./client": "./.gkm/openapi.ts",
    "./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:

typescript

// 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.

typescript

// apps/web/src/__tests__/users.spec.tsx
import { createApi } from '@myapp/api/client';
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

typescript

registerContext(contextId, {
  // Services keyed by serviceName
  services: {
    auth: mockAuthService,
    storage: mockStorageService,
    notification: mockNotificationService,
  },
  // Database instance (from .database() on the factory)
  database: transactionDb,
  // Event publisher service definition (from .publisher() on the factory)
  publisher: mockPublisherService,
  // 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()`)
`publisher`	`Service`	Event publisher service (required when endpoints use `.publisher()`)
`auditorStorage`	`unknown`	Audit storage instance (required when endpoints use `.auditor()`)

OpenAPI Documentation

Endpoints automatically generate OpenAPI 3.1 documentation:

typescript

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).

@geekmidas/constructs ​

Installation ​

Features ​

Package Exports ​

Basic Usage ​

Creating an Endpoint ​

Input Validation ​

Error Handling ​

Service Pattern ​

Advanced Usage ​

Request Context ​

Reading Cookies ​

Using Cookies for Authentication ​

Response Handling ​

Setting Cookies ​

Deleting Cookies ​

Custom Headers ​

Dynamic Status Codes ​

Default Headers ​

Simple Responses (No Modifications) ​

Complete Example ​

Authorization and Sessions ​

Authorization ​

Rate Limiting ​

Audit Logging ​

Setting Up Audit Storage ​

Actor Extraction ​

Declarative Audits ​

Conditional Audits ​

Manual Audits in Handlers ​

Factory-Level Defaults ​

Transaction Coordination ​

Row Level Security (RLS) ​

Setting Up RLS on the Factory ​

Using db in Handlers ​

Bypassing RLS for Specific Endpoints ​

Per-Endpoint RLS ​

PostgreSQL Policy Example ​

Combining RLS with Other Handler Context ​

Event Publishing ​

Defining a Publisher Service ​

Factory-Level Publisher ​

Event Subscribers ​

Basic Subscriber ​

With Services ​

Subscribing to Multiple Events ​

With Publisher (Chaining Events) ​

Configuration Options ​

Handler Context ​

How Subscribers Run ​

Testing Subscribers ​

Project Configuration ​

Queues ​

Basic Queue ​

With Services ​

Sending to a Queue (the auto-publisher) ​

Configuration Options ​

How Queues Run ​

Testing Queues ​

Project Configuration ​

Cron Jobs ​

Basic Cron ​

Schedule Expressions ​

Where it runs ​

Rate expressions ​

Uptime ​

With Services ​

Subscribing to Multiple Events ​

With Publisher (Chaining Events) ​

Configuration Options ​

Handler Context ​

How Subscribers Run ​

Testing Subscribers ​

Project Configuration ​

Queues ​

Basic Queue ​

With Services ​

Sending to a Queue (the auto-publisher) ​

Configuration Options ​

How Queues Run ​

@geekmidas/constructs

Installation

Features

Package Exports

Basic Usage

Creating an Endpoint

Input Validation

Error Handling

Service Pattern

Advanced Usage

Request Context

Reading Cookies

Using Cookies for Authentication

Response Handling

Setting Cookies

Deleting Cookies

Custom Headers

Dynamic Status Codes

Default Headers

Simple Responses (No Modifications)

Complete Example

Authorization and Sessions

Authorization

Rate Limiting

Audit Logging

Setting Up Audit Storage

Actor Extraction

Declarative Audits

Conditional Audits

Manual Audits in Handlers

Factory-Level Defaults

Transaction Coordination

Row Level Security (RLS)

Setting Up RLS on the Factory

Using `db` in Handlers

Bypassing RLS for Specific Endpoints

Per-Endpoint RLS

PostgreSQL Policy Example

Combining RLS with Other Handler Context

Event Publishing

Defining a Publisher Service

Factory-Level Publisher

Event Subscribers

Basic Subscriber

With Services

Subscribing to Multiple Events

With Publisher (Chaining Events)

Configuration Options

Handler Context

How Subscribers Run

Testing Subscribers

Project Configuration

Queues

Basic Queue

With Services

Sending to a Queue (the auto-publisher)

Configuration Options

How Queues Run

Testing Queues

Project Configuration

Cron Jobs

Basic Cron

Schedule Expressions

Where it runs

Rate expressions

Uptime

With Services

Subscribing to Multiple Events

With Publisher (Chaining Events)

Configuration Options

Handler Context

How Subscribers Run

Testing Subscribers

Project Configuration

Queues

Basic Queue

With Services

Sending to a Queue (the auto-publisher)

Configuration Options

How Queues Run