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 new NoteDatasource(c).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 (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

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';
import {NoteDatasource} from '../../db/note-datasource';

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

  async handle(c: Context) {
    const ok = await new NoteDatasource(c).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});
  }
}

// usage: new WeatherDatasource(c).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.