Create a database adapter

When to write a database adapter

Every Hot Updater database runs on one storage engine. Core (Bundles, Releases, Release Catalogs and Channels) and server plugins such as insights() and apiKeys() declare tables and indexes, and the engine turns their reads and transactions into a few calls on an adapter. The adapter stores rows and reads index ranges with its backend's native features. It knows nothing about Bundles, Releases or plugins, so a database adapter is small: the postgres provider is about 30 lines.

Write one when no existing adapter reaches your backend. Database adapters cover Supabase, Cloudflare D1, Firestore and DynamoDB, and a custom server can use Kysely, Drizzle, Prisma or MongoDB.

1. Choose how to build it: a database that speaks PostgreSQL, MySQL or SQLite needs only a SQL executor.
2. Wrap it as a database that the CLI, the Console and createHotUpdater take.
3. Create the schema with a migration.
4. Test the adapter with the suites every official database runs, then publish it.

Install

npm install @hot-updater/plugin-core
npm install @hot-updater/server @hot-updater/test-utils vitest --save-dev

@hot-updater/plugin-core exports the adapter contract, the helpers that implement it, and the EngineDatabase type that a provider returns. A published adapter lists it in peerDependencies, so the server and the adapter share one copy. The tests run the adapter behind @hot-updater/server. The examples run PostgreSQL through pg; install its driver and types to follow them, or use your own driver in its place:

npm install pg
npm install @types/pg --save-dev

How the pieces fit

createHotUpdater, CLI, console
  core and plugins     tables, indexes, typed reads and transactions
  storage engine       guards, retries, references, aggregates, schema fence
  database adapter     get, query, write, fits
  your backend

A provider's factory returns an EngineDatabase, { name, adapter, dispose? }. hot-updater.config.ts, the console and createHotUpdater all take it as database. The engine keeps references, cascades, counters and retries itself and expresses them as a few guarded ops, so an adapter implements none of them. The database adapter contract defines those ops.

Choose how to build it

A SQL executor

createSqlAdapter({ executor }) compiles every read and write to SQL for PostgreSQL, MySQL or SQLite, and runs it through an executor for your driver:

interface SqlExecutor {
  readonly dialect: "postgresql" | "mysql" | "sqlite";
  /** Runs one statement: `{ sql, params }` to `{ rows, changes }`. */
  execute(statement: SqlStatement): Promise<SqlResult>;
  /** Commits when `fn` resolves and rolls back when it throws. */
  transaction<T>(fn: (connection: SqlConnection) => Promise<T>): Promise<T>;
  /** Runs statements as one atomic batch, for drivers without transactions. */
  batch?(statements: readonly SqlStatement[]): Promise<readonly SqlResult[]>;
}

A complete provider over pg:

import {
  createEngineDatabase,
  createSqlAdapter,
  type EngineDatabase,
  migrateCoreSchema,
  type PluginTables,
  type SqlConnection,
  type SqlExecutor,
} from "@hot-updater/plugin-core";
import pg, { type PoolClient, type PoolConfig } from "pg";

/** Runs the SQL core's statements on a pool; a transaction holds one client. */
export const myExecutor = (pool: pg.Pool): SqlExecutor => {
  const over = (runner: pg.Pool | PoolClient): SqlConnection => ({
    execute: async ({ sql, params }) => {
      const result = await runner.query(sql, [...params]);
      return { rows: result.rows, changes: result.rowCount ?? 0 };
    },
  });
  return {
    dialect: "postgresql",
    ...over(pool),
    async transaction(fn) {
      const client = await pool.connect();
      try {
        await client.query("BEGIN");
        const result = await fn(over(client));
        await client.query("COMMIT");
        return result;
      } catch (error) {
        await client.query("ROLLBACK").catch(() => undefined);
        throw error;
      } finally {
        client.release();
      }
    },
  };
};

/** Hot Updater's database on PostgreSQL through `pg`. */
export const myDatabase = (config: PoolConfig): EngineDatabase => {
  const pool = new pg.Pool(config);
  return {
    ...createEngineDatabase({
      name: "myDatabase",
      adapter: createSqlAdapter({ executor: myExecutor(pool) }),
    }),
    dispose: () => pool.end(),
  };
};

/**
 * Creates the tables of core and `plugins`, the plugins the server runs, then
 * writes the settings rows the schema fence checks.
 */
export const migrateMyDatabase = async (
  config: PoolConfig,
  plugins: readonly PluginTables[] = [],
) => {
  const pool = new pg.Pool(config);
  try {
    await migrateCoreSchema(
      createSqlAdapter({ executor: myExecutor(pool) }),
      "myDatabase",
      plugins,
    );
  } finally {
    await pool.end();
  }
};

• execute returns the rows of a SELECT and, as changes, the number of rows an INSERT, UPDATE or DELETE changed. The core reads changes to tell whether a guarded update applied.
• transaction runs fn on one connection. On SQLite, begin it with BEGIN IMMEDIATE and run statements one at a time.
• Keep the driver's error codes on what you throw (code, errno, sqlState, also on cause). The core maps a unique violation to failedOp, and a serialization failure, deadlock, lock timeout or busy database to retry, by SQLSTATE, MySQL and SQLite codes.
• Statements use $1, $2… on PostgreSQL and ? elsewhere. JSON and multi-valued values are bound as JSON text, cast to jsonb on PostgreSQL, and booleans as 0 or 1 outside PostgreSQL.
• A driver with a Kysely dialect needs no executor of its own: kyselyExecutor(db, dialect) from @hot-updater/server/adapters/kysely is one, and the postgres provider is createSqlAdapter over it.

An executor without interactive transactions, such as D1 or a database RPC, provides batch instead. Each write is then one atomic batch that records the first failed guard in the _hu_write table (WRITE_GUARD_TABLE) and applies its changes only when no guard failed. migrations.apply creates that table with the others.

adapter.migrations.apply creates tables with createTableStatements(dialect, tables, tablePrefix): strings with binary collation (COLLATE "C" on PostgreSQL, utf8mb4_0900_bin or ascii_bin on MySQL), whole numbers as bigint, and a table <table>__<index> for each index over a multi-valued column.

A key-value store

createKvAdapter({ store, tablePrefix }) implements the contract over a store with strongly consistent point reads, sorted partitions and atomic conditional writes, such as DynamoDB or Firestore:

interface KeyValueStore {
  readonly id: string;
  /** The most items and bytes one write takes, per item, and per key. */
  readonly limits: {
    readonly items: number;
    readonly bytes: number;
    readonly itemBytes?: number;
    readonly keyBytes?: { readonly pk: number; readonly sk: number };
  };
  /** Strongly consistent point reads, in the order of `keys`. */
  get(keys: readonly KvKey[]): Promise<readonly (StoredRow | null)[]>;
  /** One native page of one partition's sort-key range. */
  query(
    range: KvRange,
  ): Promise<{ readonly items: readonly KvItem[]; readonly more: boolean }>;
  /** Applies every op or none. */
  write(ops: readonly KvOp[]): Promise<WriteResult>;
  /** Deletes items with no condition and not atomically, a batch at a time. */
  deleteConsumed?(keys: readonly KvKey[]): Promise<void>;
  /** Prepares the store, such as creating its table. */
  readonly migrations?: { apply(): Promise<void> };
  dispose?(): Promise<void>;
}

• An item is addressed by a partition key pk and a sort key sk, both strings. Sort keys compare by UTF-8 bytes.
• query({ pk, gte, lt, order, limit, after }) returns one page of the partition's items from gte up to lt, past after, and more when the range may go on. An empty page ends the range.
• write(ops) applies put, delete, check and add, an increment that creates a missing item from init. Every op's condition (the item exists, does not exist, or has _v equal to v) sees the state before the write. Return failedOp for an op whose condition failed and retry for a transient conflict, as an adapter's write does.
• If the store throws when its table is missing, give that error code: "42P01", as the DynamoDB store does, so the schema fence reports a missing schema.

The helper stores a row as one item at pk = <table>, and an index in key order reads those items. Every other index adds an item per entry in its own partition, holding a copy of the row, written in the same write. A unique entry is an item that one row at a time may hold. fits counts items and bytes against limits. The DynamoDB store and the Firestore store are complete implementations. Wrap the adapter as in myDatabase, with createKvAdapter({ store }) in place of createSqlAdapter.

The adapter contract directly

A backend with its own transactions and secondary indexes can implement DatabaseAdapter directly: keep each table's rows by key, create each index as declared, and read an index range in indexOrderColumns order. The database adapter contract lists every rule, and createMemoryAdapter() implements them all as the reference. The MongoDB adapter is a production one: a collection per table keyed by _id, guards as conditional updates, and one transaction per write. While you develop, wrap your adapter in verifyAdapter(adapter). It fails reads that return rows for other keys, rows outside the range or out of order, or more than limit rows, and writes that touch a row twice or that fits refuses.

Document-store adapters

An adapter over a document store, which is neither SQL nor key-value, reads the physical schema through four names from @hot-updater/plugin-core, as the MongoDB adapter does:

Adapters over an ORM

An adapter over an ORM's database, as the Kysely, Drizzle, and Prisma adapters are, runs core's SQL through createSqlAdapter and gives its database a migrator from @hot-updater/plugin-core, which its createMigrator returns:

• createEngineSqlMigrator({ executor, adapterName, schema, settings, tablePrefix }) when Hot Updater creates the tables: it applies the generated SQL schema when a settings row is missing or different, as kyselyAdapter's does.
• createSettingsMigrator({ executor, adapterName, settings, applyTables, fixups }) when the ORM applies the DDL itself, as Drizzle's and Prisma's do: fixups sets what the ORM cannot declare, then it writes the settings rows. applyTables says how the ORM creates the tables, for the error when they are missing.

Both refuse a v0 database. The ORM's own schema generator renders each table from sqlTableShapes and quotes names with quoteSql; see Create the schema.

Wrap it as a database

createEngineDatabase({ name, adapter }) returns the EngineDatabase a provider's factory returns, as myDatabase does. It puts the adapter behind the schema fence and forwards the adapter's dispose. Add your own dispose when the factory opens a client. When the adapter has migrations, the database also carries the migration that hot-updater db migrate runs.

Pass the database to the CLI and the server, with the same plugins. Partial configuration:

import { apiKeys, insights } from "hot-updater/plugins";
import { defineConfig } from "hot-updater";

import { myDatabase } from "./myDatabase";

export default defineConfig({
  // build: ...,
  // storage: ...,
  database: myDatabase({ connectionString: process.env.DATABASE_URL }),
  plugins: [insights(), apiKeys()],
});

import { createHotUpdater } from "@hot-updater/server";
import { apiKeys, insights } from "@hot-updater/server/plugins";

import { myDatabase } from "./myDatabase";

export const hotUpdater = createHotUpdater({
  database: myDatabase({ connectionString: process.env.DATABASE_URL }),
  // storage: s3Storage({ ... }),
  plugins: [insights(), apiKeys()],
});

Measure what each call reads

meterReads(database) from @hot-updater/plugin-core meters any database on the storage engine, yours or a provider's, so you can see what each server call costs a store that bills by read. Pass the metered database to createHotUpdater, then run a call through measureReads:

import { meterReads } from "@hot-updater/plugin-core";
import { createHotUpdater } from "@hot-updater/server";

import { myDatabase } from "./myDatabase";

const database = meterReads(
  myDatabase({ connectionString: process.env.DATABASE_URL }),
);
const hotUpdater = createHotUpdater({
  database,
  storage, // The adapter hot-updater.config.ts uploads with, such as s3Storage(...).
  clientAccess: "public",
});

const { result, adapter, engine } = await database.measureReads(() =>
  hotUpdater.core.listBundles({ limit: 20 }),
);

adapter counts what your adapter returned: get calls and the keys they asked for, and query calls and the rows they returned. engine counts the reads core or a plugin made and the rows they got back, where an aggregate row counts once however many shard rows the adapter read. Aggregate batching's reads count too: a compaction or a flush that runs during the call, including a memory-mode timer flush that lands in it, since a store bills those like any other read. A process's first schema-fence read and its retention passes belong to no call and are not counted. Every engine on the database reports to the same meter.

createEngine(database, { plugins }) runs that storage engine without a server: engine.core holds core's tables and engine.database(plugin) a plugin's, behind the plugins' settings rows on a fenced database, as createHotUpdater runs them. Tooling and tests use it to run a plugin's tables on your database, and createPluginTestHarness from @hot-updater/test-utils is built on it. Wrap the adapter in verifyAdapter to check every call against the contract as it runs.

Create the schema

Migrations write settings rows to private_hot_updater_settings last, after every table exists: schema.engine ("1"), schema.core ("1.0.0"), and schema.<id> for each plugin the server runs, holding its schemaVersion, such as schema.insights and schema.apiKeys. Before a process's first read or write, the schema fence reads core's rows in one batch, and a server reads its plugins' rows in another. A missing or different row throws HotUpdaterSchemaMigrationRequiredError, which names the setting, and the server's handlers answer 503. A read that fails because the settings table is missing counts as a missing row when the error says so: PostgreSQL 42P01 or 42703, MySQL ER_NO_SUCH_TABLE or ER_BAD_FIELD_ERROR, or a message such as no such table. Any other failure, such as a refused connection, is thrown as it is.

Export a migration with your factory, as migrateMyDatabase does. migrateCoreSchema(adapter, name, plugins) passes core's tables and those of plugins to adapter.migrations.apply, then writes their settings rows. Running it again changes nothing. Run it from a script, a deploy job or your provider's setup command with the plugins the server runs, as migrateDynamoDB and migrateFirebaseDatabase do for their providers.

hot-updater db migrate runs the same migration from the file that exports the server's hotUpdater, with the server's plugins; see Server Plugins. It refuses a database whose adapter has no migrations.

A SQL provider can check in its schema instead. generateEngineSql(dialect, schema, settings, { tablePrefix }) from @hot-updater/plugin-core returns the statements: tables and indexes, then the settings rows, in SETTINGS_TABLE. Pass coreSchema and coreSettings for core alone, or the schema and settings of toolingTargetOf(plugins) for core and those plugins. The postgres provider generates its checked-in SQL for core, and D1 its Worker migration and sql/bundles.sql for core and the managed Worker's plugins; a test fails when a file differs from the generator. Add createTableStatements(dialect, [WRITE_GUARD_TABLE], tablePrefix) when the executor has batch. Use the same tablePrefix for schema generation, the write guard table, and createSqlAdapter. When other tooling creates the tables, writeSchemaSettings(adapter, name, settings) writes only the settings rows; that tooling creates SETTINGS_TABLE with the rest, and an ORM's schema generator renders each table from sqlTableShapes and quotes names with quoteSql.

A provider whose schema is applied as migration files adds generateSchema(version, name, target) to its database, as supabaseDatabase and the REST d1Database do. hot-updater db generate calls it with "latest" and the server's target, { schema, settings }: its toolingTargetOf(plugins), or coreTarget without plugins. It writes the returned code to its relative path. When the adapter also has migrations, as D1's does, hot-updater db migrate applies the same schema directly; without them, as with Supabase, it points to db generate.

Test the adapter

@hot-updater/test-utils publishes the suites that every official database runs. Register all three:

The examples expect TEST_DATABASE_URL to name an empty PostgreSQL database. Each suite creates its own tables, and the specs drop or empty them through this helper:

import { isMultiIndex, type PhysicalTable } from "@hot-updater/plugin-core";

/** The quoted tables the SQL core creates for `tables`, index tables included. */
export const tableNames = (tables: readonly PhysicalTable[], prefix = "") =>
  tables
    .flatMap((table) => [
      table.name,
      ...table.indexes
        .filter((index) => isMultiIndex(table, index))
        .map((index) => `${table.name}__${index.name}`),
    ])
    .map((name) => `"${prefix}${name}"`)
    .join(", ");

Adapter conformance

import { createSqlAdapter } from "@hot-updater/plugin-core";
import { setupDatabaseAdapterConformanceSuite } from "@hot-updater/test-utils";
import pg from "pg";
import { afterAll } from "vitest";

import { myExecutor } from "./myDatabase";
import { tableNames } from "./testTables";

const pool = new pg.Pool({ connectionString: process.env.TEST_DATABASE_URL });
afterAll(() => pool.end());

setupDatabaseAdapterConformanceSuite({
  name: "myDatabase",
  createAdapter: async ({ tables }) => {
    const adapter = createSqlAdapter({ executor: myExecutor(pool) });
    await adapter.migrations!.apply(tables);
    return {
      adapter,
      cleanup: async () => {
        await pool.query(`DROP TABLE ${tableNames(tables)}`);
      },
    };
  },
});

The cases cover value round-trips, point reads in key order, each op's guard, all-or-nothing writes with a failure at every position, unique indexes and the op they name, UTF-8 byte order, cursor paging without gaps or repeats, multi-valued indexes, full pages under capped native pages, one winner among concurrent writers, no lost increments, write-skew rejection, and over-limit writes refused before sending. Every call also passes through verifyAdapter.

A key-value store takes nativePageSize as its page size. Its maxOps is the number of conformance inserts one write fits: the DynamoDB spec sets 33, since an insert there is 3 items and a transaction takes 100.

Read budgets

import { createSqlAdapter } from "@hot-updater/plugin-core";
import {
  postgresRowsExamined,
  setupReadBudgetTestSuite,
} from "@hot-updater/test-utils";
import pg from "pg";
import { afterAll } from "vitest";

import { myExecutor } from "./myDatabase";
import { tableNames } from "./testTables";

const pool = new pg.Pool({ connectionString: process.env.TEST_DATABASE_URL });
afterAll(() => pool.end());

setupReadBudgetTestSuite({
  name: "myDatabase",
  createAdapter: async ({ tables }) => {
    // Each read is explained on one session of the pool.
    const session = await pool.connect();
    const reads = postgresRowsExamined(
      async (sql, params) => (await session.query(sql, [...params])).rows,
    );
    const adapter = createSqlAdapter({
      executor: reads.wrap(myExecutor(pool)),
      tablePrefix: "budget_",
    });
    await adapter.migrations!.apply(tables);
    return {
      adapter,
      examined: reads.examined,
      cleanup: async () => {
        session.release();
        await pool.query(`DROP TABLE ${tableNames(tables, "budget_")}`);
      },
    };
  },
});

The suite runs core and the server's insights(), apiKeys(), and remoteConfig() on createMeasuredDatabase(adapter, plugins, { now, storage }) from @hot-updater/test-utils: createHotUpdater over meterReads({ name, adapter: verifyAdapter(adapter) }), so every adapter call is checked and measured, without the schema fence. now is the plugins' clock. It loads @hot-updater/server, which your project installs for the tests. The suite seeds Bundles, Releases, Channels, an API key, Insights events, and Remote Config versions, caps native pages at two rows, and checks each API's reads exactly, at the adapter and at the engine. For example, the update check is one point read of a catalog row, and artifact resolution is one batch read of two Bundles plus one patch read.

The server over HTTP

import { toolingTargetOf } from "@hot-updater/plugin-core";
import { createHotUpdater } from "@hot-updater/server";
import { createInsightsModel, insights } from "@hot-updater/server/plugins";
import {
  createReleaseCatalogTestStorage,
  insightsTestSuite,
  setupDatabaseTestSuite,
  startHttpTestServer,
} from "@hot-updater/test-utils";
import pg from "pg";

import { migrateMyDatabase, myDatabase } from "./myDatabase";
import { tableNames } from "./testTables";

const config = { connectionString: process.env.TEST_DATABASE_URL };
const pool = new pg.Pool(config);
/** The plugins the suite's server runs, whose tables the migration adds. */
const plugins = [insights()];

setupDatabaseTestSuite({
  name: "myDatabase",
  migrate: () => migrateMyDatabase(config, plugins),
  createDatabase: () => myDatabase(config),
  // Every table but the settings, which the schema fence checks.
  reset: async () => {
    await pool.query(
      `TRUNCATE ${tableNames(toolingTargetOf(plugins).schema.tables)}`,
    );
  },
  dispose: async (database) => {
    await database.dispose?.();
    await pool.end();
  },
  createHttpClient: (options) =>
    startHttpTestServer(
      createHotUpdater({ ...options, plugins, clientAccess: "public" })
        .handlers,
    ),
  plugins: [
    insightsTestSuite({
      // The plugin's API on your database, as the server assembles it.
      createModel: (database) =>
        createInsightsModel(
          createHotUpdater({
            database,
            storage: createReleaseCatalogTestStorage(),
            plugins,
            clientAccess: "public",
          }).api.insights,
        ),
    }),
  ],
});

The suite sends real requests to the admin API and the client routes. It runs core's operations, the Bundle admin API, the Release Catalog contract, and each listed plugin's suite. The Release Catalog scenarios pass each catalog they receive to the production client selector and carry a simulated device from its built-in bundle to OTA A, OTA B, back to A, and back to built-in. Other scenarios cover deleted Releases, the native minimum, cohort changes and crashed updates.

Run the specs with Vitest:

TEST_DATABASE_URL=postgres://localhost:5432/hot_updater_test npx vitest run

Test a running server

setupReleaseCatalogTestSuite runs the Release Catalog scenarios against a server you started, through its framework routing and authentication:

import {
  createHttpTestClient,
  setupReleaseCatalogTestSuite,
} from "@hot-updater/test-utils";

setupReleaseCatalogTestSuite({
  getClient: () =>
    createHttpTestClient({
      clientBaseUrl: "http://127.0.0.1:3000/hot-updater",
      adminBaseUrl: "http://127.0.0.1:3000/hot-updater/admin",
      adminHeaders: { Authorization: `Bearer ${process.env.ADMIN_TOKEN}` },
    }),
});

Start the server on a disposable database, with createReleaseCatalogTestStorage() from @hot-updater/test-utils/node in its storage. That entry does not load Vitest, so a separate server process can import it. Add clientHeaders when the server's client routes need a key.

Publish the adapter

Publish the factory and its migration from one package, as dynamoDB and migrateDynamoDB ship together. List @hot-updater/plugin-core in peerDependencies, so the adapter runs on the project's copy of the engine, and keep the three suites in your CI. See Test and publish for the package layout, naming, and what to document.