Drizzle D1 datasources

Datasources are small classes that own data access for one resource. They are bound to the Hono context of a request, get a request-scoped logger and, when a KV binding exists, a cache.

Import them from the @ltv/cwb/datasources entry point. The whole entry point, AbstractDatasource included, requires the optional peer drizzle-orm 0.45.x.

Class	Use
`AbstractDatasource`	Base class: context, logger, `withCache`, `clear`, lifecycle methods
`DrizzleD1Datasource<TSchema>`	Drizzle ORM on a D1 binding, with `batch()` and `healthCheck()`
`DrizzleD1TenantDatasource<TSchema>`	Adds tenant scoping; see Multi-tenancy

All three are abstract: extend them and declare a unique readonly name.

A D1 datasource

src/db/schema.ts
import {integer, sqliteTable, text} from 'drizzle-orm/sqlite-core';

export const notes = sqliteTable('notes', {
  id: integer('id').primaryKey({autoIncrement: true}),
  title: text('title').notNull(),
});

export const auditLog = sqliteTable('audit_log', {
  id: integer('id').primaryKey({autoIncrement: true}),
  noteId: integer('note_id').notNull(),
  action: text('action').notNull(),
});

export type Note = typeof notes.$inferSelect;

src/db/note-datasource.ts
import {eq} from 'drizzle-orm';
import type {Context} from 'hono';
import {DrizzleD1Datasource} from '@ltv/cwb/datasources';
import * as schema from './schema';

export class NoteDatasource extends DrizzleD1Datasource<typeof schema> {
  readonly name = 'notes'; // tags logs, default cache namespace

  constructor(c: Context) {
    super(c, {schema});
  }

  create(title: string): Promise<schema.Note> {
    return this.db.insert(schema.notes).values({title}).returning().get();
  }

  findById(id: number): Promise<schema.Note | null> {
    return this.withCache(
      'byId',
      {id},
      async () =>
        (await this.db.query.notes.findFirst({where: eq(schema.notes.id, id)})) ?? null,
      {ttl: 60}
    );
  }
}

in a route handler
const note = await c.var.ds.notes.findById(params.id);
if (!note) throw ApiError.notFound('Note');
return c.respond(note);

One instance per request, unique names

Create datasources from the request's context (the registry does this for you, or new NoteDatasource(c)); don't share an instance across requests. The name tags logs and is the default cache namespace, so two datasources with the same name would share, and clear, each other's cache entries.

Options

super(c, options) accepts DrizzleD1DatasourceOptions<TSchema>:

Option	Default	Description
`schema`	none	Your tables and relations. Enables the typed relational API `db.query.*`
`binding`	`'DB'`	Name of the D1 binding on `c.env`
`logQueries`	`false`	Log SQL at `debug` level through the datasource logger. Bound parameters are omitted in production

Members

Member	Description
`db`	Typed `DrizzleD1Database<TSchema>`, created on first access. Throws `D1 binding "DB" not found in environment` when the binding is missing
`batch([...queries])`	Runs several statements atomically in one D1 round trip, with a typed result tuple
`healthCheck()`	Runs `select 1`; resolves to `true` or `false` and logs failures
`clear(options?)`	Removes this datasource's cache entries within the KV operation budget
`name` (abstract)	Unique datasource name
`context`	The Hono context passed to the constructor
`logger` (protected)	Child of the request logger tagged `{component: 'datasource', name}`
`withCache(key, params, fetch, options?)` (protected)	Cached call; see below
`cacheBinding` (protected getter)	KV binding name, default `'KV'`
`cacheNamespace` (protected getter)	Cache namespace, default `name`
`createQueryLogger()` (protected)	The Drizzle logger used by `logQueries`
`init()`, `connect()`, `disconnect()`	Lifecycle hooks with default implementations that only log at debug level

Datasource registry (`c.var.ds`)

Available since @ltv/cwb 0.2.0.

createApp({datasources}) exposes one datasource per name on c.var.ds, in app-wide middleware, module middleware and handlers.

src/datasources.ts
import {defineDatasources, type InferDatasources} from '@ltv/cwb';
import {NoteDatasource} from './db/note-datasource';
import {TenantNoteDatasource} from './db/tenant-note-datasource';

export const datasources = defineDatasources({
  notes: (c) => new NoteDatasource(c),
  tenantNotes: (c) => new TenantNoteDatasource(c),
});

// Types of c.var.ds are inferred from the factories
declare module '@ltv/cwb' {
  interface DatasourceRegistry extends InferDatasources<typeof datasources> {}
}

src/index.ts
import {datasources} from './datasources';

const app = createApp({
  datasources, // c.var.ds.notes, c.var.ds.tenantNotes
  modules: [notes, tenantNotes],
  openApi: (hono) => fromHono(hono),
});

Reading datasources

c.var.ds and c.get('ds') return the same registry (c.var is Hono's shorthand for c.get), and both are fully typed:

// property access
await c.var.ds.tasks.remove(params.projectId, params.taskId);

// c.get with destructuring (rename to avoid clashing with other variables)
const {tasks: taskDs} = c.get('ds');
await taskDs.assertProject(params.projectId);
await taskDs.remove(params.projectId, params.taskId);

Destructuring creates what you name

const {tasks} = c.get('ds') reads tasks, so that datasource is created right away. It is still created once per request, and names you don't destructure are not created. Destructuring an unregistered name throws the "not registered" error; spreading ({...c.var.ds}) creates every datasource.

How it behaves:

Behavior	Details
Lazy	A factory runs the first time a request reads its name. Requests that never use a datasource never create it.
Per request	The instance is reused for the rest of the request and never shared with another request: concurrent requests share an isolate and may belong to different tenants.
Always registered	`c.var.ds` is set right after request logging, before `/health` and your `middleware`, even without `datasources`.
Unknown names	Reading a name without a factory throws `Datasource "x" is not registered. Add a factory to createApp({datasources}).`
Failing factories	The error goes through the normal error envelope (for example `TENANT_REQUIRED`, 403) and isn't cached, so the next access runs the factory again.
Dependencies	Factories and datasource methods can read other datasources through `c.var.ds`. A circular dependency throws a clear error.
Shape	Read-only and enumerable (`Object.keys(c.var.ds)`); safe to pass through `await`.
Logging	`JSON.stringify(c.var)` and logging `c.var` are safe: the registry serializes as `{datasources: [...names]}` without creating anything. Spreading it (`{...c.var.ds}`) creates every datasource.
Reserved names	`then`, `toJSON`, `asymmetricMatch` and `Object.prototype` names (`constructor`, `toString`, `valueOf`, ...) can't be datasource names: `createApp` (or the middleware) throws a `TypeError`.

Keep factories synchronous

The registry caches whatever a factory returns. An async factory returns a promise, and a rejected promise stays cached for the rest of the request instead of being retried. Construct the datasource synchronously and do async work in its methods.

Typing the registry

DatasourceRegistry is an empty interface that you augment once. The inferred style above needs no type list. To list the types by hand, declare them and type the factories as DatasourceFactories; a missing or mistyped factory then fails to compile:

src/datasources.ts
import type {DatasourceFactories} from '@ltv/cwb';

declare module '@ltv/cwb' {
  interface DatasourceRegistry {
    notes: NoteDatasource;
  }
}

export const datasources: DatasourceFactories = {
  notes: (c) => new NoteDatasource(c),
};

One augmentation per TypeScript program

Declarations merge across every file in a program, so declare DatasourceRegistry in one place. If two apps (or an app and its tests with their own registry) live in one repository, give each its own tsconfig.json.

Type caveats:

datasources is optional in AppConfig, so forgetting it after augmenting DatasourceRegistry compiles and fails at runtime, with the "not registered" error, on first use.
c.var.ds is typed on every Hono Context, but it exists at runtime only on createApp apps or apps using datasourcesMiddleware.
The variable name is ds; don't declare a context variable of your own with that name.

Without `createApp`

On a plain Hono app or sub-app, install the middleware yourself:

import {datasourcesMiddleware} from '@ltv/cwb';

app.use('*', datasourcesMiddleware(datasources)); // sets c.var.ds

Constructing a datasource directly (new NoteDatasource(c)) still works, for example to pass an explicit tenant in a background job. See Multi-tenancy.

Atomic writes with `batch`

D1 has no interactive transactions. batch() sends statements in one round trip and applies them atomically:

src/db/note-datasource.ts
async rename(id: number, title: string): Promise<schema.Note[]> {
  const [, renamed] = await this.batch([
    this.db.insert(schema.auditLog).values({noteId: id, action: 'rename'}),
    this.db.update(schema.notes).set({title}).where(eq(schema.notes.id, id)).returning(),
  ]);
  return renamed;
}

Health checks

src/modules/health/db-health.route.ts
import {OpenAPIRoute} from 'chanfana';
import type {Context} from 'hono';
import {z} from 'zod';
import {jsonSuccess} from '@ltv/cwb';

export class DbHealth extends OpenAPIRoute {
  schema = {
    tags: ['Health'],
    responses: jsonSuccess('Database health', z.object({db: z.boolean()})),
  };

  async handle(c: Context) {
    const ok = await c.var.ds.notes.healthCheck();
    return c.respond({db: ok}, ok ? 200 : 503);
  }
}

Query logging

constructor(c: Context) {
  super(c, {schema, logQueries: true});
}

Queries are logged at debug, so set LOG_LEVEL=debug to see them. Each line carries requestId, component: 'datasource', the datasource name and query; params are included only outside production, because bound values can contain personal data.

Caching with `withCache`

withCache(cacheKey, params, fetchFn, options?) memoizes fetchFn in KV when the binding named by cacheBinding exists on c.env, and simply calls fetchFn when it doesn't. It uses a KVCache with:

entries under cacheNamespace (default: name), unless options.namespace is given (entries in a custom namespace are not removed by clear());
background writes kept alive with executionCtx.waitUntil;
cache logs through the datasource logger.

Options are CacheOptions<T>: ttl, namespace and revive. Values follow the JSON-only contract; pass revive to restore dates or class instances on hits.

import {DrizzleD1Datasource} from '@ltv/cwb/datasources';

class ProductDatasource extends DrizzleD1Datasource<typeof schema> {
  readonly name = 'products';

  constructor(c: Context) {
    super(c, {schema, binding: 'CATALOG_DB'});
  }

  protected override get cacheBinding(): string {
    return 'CACHE'; // reads c.env.CACHE instead of c.env.KV
  }
}

clear(options?) deletes the entries in cacheNamespace within the KV operation budget and returns {deleted, complete, cursor, failed}. Without a KV binding it returns {deleted: 0, complete: true, failed: false}. See Invalidation for continuing an incomplete clear.

Other data sources

Only a D1 datasource is provided. For other stores, extend AbstractDatasource to reuse the logger, cache and lifecycle methods:

import {AbstractDatasource} from '@ltv/cwb/datasources';

export class WeatherDatasource extends AbstractDatasource {
  readonly name = 'weather';

  forecast(city: string): Promise<unknown> {
    return this.withCache('forecast', {city}, async () => {
      this.logger.debug({city}, 'Fetching forecast');
      const res = await fetch(`https://weather.example.com/v1/forecast?city=${encodeURIComponent(city)}`);
      return res.json();
    }, {ttl: 600});
  }
}

// register it: defineDatasources({weather: (c) => new WeatherDatasource(c)})
// usage in a handler: await c.var.ds.weather.forecast('Hanoi')

The BaseDatasource interface (name, context, and optional init, connect, disconnect, healthCheck, clear) describes the same shape for classes that don't extend the base class.

A D1 datasource​

Options​

Members​

Datasource registry (c.var.ds)​

Reading datasources​

Typing the registry​

Without createApp​

Atomic writes with batch​

Health checks​

Query logging​

Caching with withCache​

Other data sources​