Writing adapters

The library has two doors to the world: getRate, which asks for the tariff and usage, and commit, which writes the plan. Both are your code; the reference adapters are starting points.

The getRate contract

(meter, dims, ctx, at) => { rate, usedSoFar?, carry?, limit? } | null | undefined   // or a Promise of it

• Return the whole tariff; rate() picks the tier. Raw objects are fine; the library validates them (invalid_rate).
• usedSoFar: units this account used on this meter in the period you choose, so far. Required for tiered, volume, package, flat-fee and cumulative tariffs; missing → usage_required (rules).
• carry: cumulative rounding only, the previous usage row’s detail.carry.
• limit: this meter’s cap in the same period as usedSoFar; a call beyond it returns quota_exceeded (quotas). Without usedSoFar: usage_required.
• How to read at (event time or computation time) is your decision. Period-end observations use at: window.to − 1, because windows are half-open and window.to selects the next period (A98).
• Counter scope is yours: the reference adapters count per meter, without dimensions. For tiers per dimension value, count with dims in getRate (sqliteAdapter.usedSoFar({ dims }); A100).
• null = no price → ifMissing. A throw from getRate surfaces as an exception from observe.
• Called once per (meter, dims) per call. Declare all four parameters (A94).

The commit contract

(plan: Plan) => void | Promise<void>

Rule	Why
All or nothing: ledger and usage rows in one transaction	A half-written plan is an inconsistent balance
Idempotent: skip a repeated (refType, refId) silently (together with op on the ledger)	Application retries stay safe
The balance moves only for ledger rows actually inserted	A retry must not debit twice
Prepaid: if balance − reserved would go below 0, throw InsufficientCredit and write nothing	The balance gate is in the same atomic step as the write
For a new usage row with a limit: if the counter for the row’s period + quantity would exceed it, throw QuotaExceeded and write nothing	The quota gate is in the same atomic step too (A107). SQLite: windowOf, DO: periodOf
charge lowers the balance (raises it if negative); hold/extend raise reserved and count as debits; capture lowers the balance; release frees the reservation	Available = balance − reserved
Track what is actually reserved per hold: capture lowers the reservation by at most that, release frees that rather than the plan’s amount (A102)	A hold retried with a different amount is written as a no-op but returns a recomputed upperBound; a release trusting the plan would shift the reservation
Throw for anything else	The library returns commit_failed with the error in cause

InsufficientCredit is also matched by name, so it crosses RPC and realm boundaries. Empty plans (no ledger and no usage rows) never reach commit.

commitContract

pricemeter/testing ships the contract every commit must satisfy as a test suite. For your adapter, write seed (give an account a balance) and snapshot (row counts and available balance):

docs/src/snippets/adapters.test.ts

const report = await commitContract({
  make: () => {
    const a = sqliteAdapter({ db: new DatabaseSync(":memory:"), prepaid: true });
    a.migrate();
    return {
      commit: a.commit,
      seed: (account, amount) => a.credit(account, amount),
      snapshot: (account) => ({
        usage: a.usedSoFar({ account, meter: "contract/meter" }) / 3, // the sample row has quantity 3
        ledger: a.spent({ account }) === 0 ? 0 : 1,
        available: a.balance(account).available,
      }),
    };
  },
});
report.ok; // true
report.checks; // [{ name: "repeated (refType, refId) is a silent no-op", ok: true }, …]

For a postpaid adapter pass { prepaid: false }; the balance checks are skipped.

SQLite

pricemeter/sqlite works with node:sqlite, bun:sqlite and better-sqlite3 (a structural SqliteLike). It provides migrate, commit, usedSoFar, spent, balance and credit.

docs/src/snippets/adapters.test.ts

import { DatabaseSync } from "node:sqlite";
import { buildMetering } from "pricemeter";
import { monthWindow } from "pricemeter/calendar";
import { sqliteAdapter } from "pricemeter/sqlite";

const db = new DatabaseSync(":memory:"); // node:sqlite here; bun:sqlite and better-sqlite3 fit too
const store = sqliteAdapter({ db, prepaid: true });
store.migrate();
store.credit("acme", 5_000_000); // $5

const metering = buildMetering()
  .refs(["api_call"])
  .meter("api/requests")
  .getRate(async (meter, _dims, ctx, at) => {
    const { from, to } = monthWindow({ at }); // usedSoFar = this calendar month
    return {
      rate: { model: "graduated", tiers: [{ from: 0, unitPriceMicroUsd: 0 }, { from: 1_000, unitPriceMicroUsd: 1_000 }] },
      usedSoFar: store.usedSoFar({ account: ctx.accountId, meter, from, to }),
    };
  })
  .commit(store.commit);

One BEGIN IMMEDIATE per plan. Rows are written with INSERT … ON CONFLICT DO NOTHING (not OR IGNORE, which also swallows NOT NULL violations); balance and reservations move only for ledger rows actually inserted:

packages/core/src/sqlite/index.ts

const commit: Commit = (plan: Plan) => {
  const q = s();
  db.exec("BEGIN IMMEDIATE");
  try {
    const moves = new Map<string, { balance: number; reserved: number; debit: number }>();
    const move = (a: string) => {
      let m = moves.get(a);
      if (!m) moves.set(a, (m = { balance: 0, reserved: 0, debit: 0 }));
      return m;
    };
    for (const op of plan.ledger) {
      const holdId = op.op === "charge" ? null : op.holdId;
      const r = q.ledger.run(`${op.op}:${op.refType}:${op.refId}`, op.account, op.op, op.amount, holdId, op.refType, op.refId, op.at);
      if (Number(r.changes) === 0) continue; // retry: already written
      const m = move(op.account);
      const outstanding = op.op === "charge" ? 0 : Number((q.holdGet.get(op.holdId) as { r: number | bigint } | undefined)?.r ?? 0);
      switch (op.op) {
        case "charge":
          m.balance -= op.amount;
          m.debit += op.amount;
          break;
        case "hold":
        case "extend":
          m.reserved += op.amount;
          m.debit += op.amount;
          q.holdAdd.run(op.holdId, op.account, op.amount);
          break;
        case "capture": {
          const freed = Math.max(0, Math.min(op.amount, outstanding));
          m.balance -= op.amount;
          m.reserved -= freed;
          q.holdAdd.run(op.holdId, op.account, -freed);
          break;
        }
        case "release":
          // frees what is actually reserved, whatever the (possibly retried) plan computed
          m.reserved -= outstanding;
          q.holdAdd.run(op.holdId, op.account, -outstanding);
          q.fixAmount.run(outstanding, `${op.op}:${op.refType}:${op.refId}`);
          break;
      }
    }
    for (const [account, m] of moves) {
      q.ensure.run(account);
      q.move.run(m.balance, m.reserved, account);
      if (o.prepaid && m.debit > 0) {
        const b = balance(account);
        if (b.available < 0) throw new InsufficientCredit(`account ${account} is short by ${-b.available}`, { account, available: b.available + m.debit, required: m.debit });
      }
    }
    for (const u of plan.usage) {
      if (u.limit !== undefined && !q.usageSeen.get(u.refType, u.refId)) {
        // counted inside the transaction, including rows this plan inserted before this one
        const w = o.windowOf?.(u) ?? { from: Number.MIN_SAFE_INTEGER, to: Number.MAX_SAFE_INTEGER };
        const used = Number((q.usedIn.get(u.account, u.meter, w.from, w.to) as { n: number | bigint }).n);
        if (used + u.quantity > u.limit) throw new QuotaExceeded({ meter: u.meter, limit: u.limit, used, requested: u.quantity, account: u.account });
      }
      q.usage.run(
        `${u.refType}:${u.refId}`,
        u.account,
        u.meter,
        canonicalJson(u.dims),
        u.quantity,
        u.amount,
        u.rateId ?? null,
        JSON.stringify(u.detail),
        u.refType,
        u.refId,
        u.at,
      );
    }
    db.exec("COMMIT");
  } catch (e) {
    db.exec("ROLLBACK");
    throw e;
  }
};

Schema: usage_events (UNIQUE (ref_type, ref_id)), ledger_entries (UNIQUE (type, ref_type, ref_id)), holds (the amount reserved per hold), accounts. dims are stored as canonical JSON (sorted keys), so usedSoFar({ dims }) can compare by equality.

packages/core/src/sqlite/index.ts

export function schema(prefix = ""): string {
  return `
CREATE TABLE IF NOT EXISTS ${prefix}usage_events (
  id TEXT PRIMARY KEY, account_id TEXT NOT NULL, meter TEXT NOT NULL, dims TEXT NOT NULL, quantity INTEGER NOT NULL,
  amount_micro_usd INTEGER NOT NULL, rate_id TEXT, detail TEXT NOT NULL, ref_type TEXT NOT NULL, ref_id TEXT NOT NULL,
  at INTEGER NOT NULL, UNIQUE (ref_type, ref_id));
CREATE INDEX IF NOT EXISTS ${prefix}usage_events_account_meter_at ON ${prefix}usage_events (account_id, meter, at);
CREATE TABLE IF NOT EXISTS ${prefix}ledger_entries (
  id TEXT PRIMARY KEY, account_id TEXT NOT NULL, type TEXT NOT NULL, amount_micro_usd INTEGER NOT NULL, hold_id TEXT,
  ref_type TEXT NOT NULL, ref_id TEXT NOT NULL, at INTEGER NOT NULL, UNIQUE (type, ref_type, ref_id));
CREATE INDEX IF NOT EXISTS ${prefix}ledger_entries_account_at ON ${prefix}ledger_entries (account_id, at);
CREATE TABLE IF NOT EXISTS ${prefix}holds (
  id TEXT PRIMARY KEY, account_id TEXT NOT NULL, reserved_micro_usd INTEGER NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS ${prefix}accounts (
  id TEXT PRIMARY KEY, balance_micro_usd INTEGER NOT NULL DEFAULT 0, reserved_micro_usd INTEGER NOT NULL DEFAULT 0);
`;
}

Cloudflare

pricemeter/cloudflare: one Durable Object per account; the ledger, balance and usedSoFar counters live in the DO’s SQLite storage.

• applyPlan(storage, plan, { prepaid, periodOf }) does everything inside transactionSync: idempotent ledger, balance gate, per-period counters and the usage outbox. A throw rolls it all back.
• doCommit(stub) is the Worker-side commit: it routes the plan to the account’s DO. Errors do not cross RPC reliably, so insufficiency comes back as a value (commitReply) and is rethrown as InsufficientCredit in the Worker.
• Usage rows enter the outbox in the same transaction; flushOutbox + d1UsageSink ship them to D1 idempotently (waitUntil or an alarm).

packages/core/src/cloudflare/index.ts

/**
 * Applies a plan atomically inside the account's DO. Repeated `(refType, refId)` rows are skipped.
 * Throws `InsufficientCredit` (nothing written) when prepaid and short.
 */
export function applyPlan(storage: DurableStorageLike, plan: Plan, opts: ApplyOptions = {}): void {
  const { sql } = storage;
  const periodOf = opts.periodOf ?? (() => "*");
  storage.transactionSync(() => {
    let balance = 0;
    let reserved = 0;
    let debit = 0;
    for (const op of plan.ledger) {
      const seen = sql.exec(`SELECT 1 AS x FROM pm_ledger WHERE type = ? AND ref_type = ? AND ref_id = ?`, op.op, op.refType, op.refId).toArray();
      if (seen.length) continue;
      sql.exec(
        `INSERT INTO pm_ledger (type, ref_type, ref_id, amount, hold_id, at) VALUES (?, ?, ?, ?, ?, ?)`,
        op.op,
        op.refType,
        op.refId,
        op.amount,
        op.op === "charge" ? null : op.holdId,
        op.at,
      );
      const outstanding =
        op.op === "charge" ? 0 : Number((sql.exec(`SELECT reserved FROM pm_holds WHERE hold_id = ?`, op.holdId).toArray()[0] as { reserved?: number } | undefined)?.reserved ?? 0);
      const moveHold = (d: number) =>
        sql.exec(
          `INSERT INTO pm_holds (hold_id, reserved) VALUES (?, ?) ON CONFLICT (hold_id) DO UPDATE SET reserved = reserved + excluded.reserved`,
          (op as { holdId: string }).holdId,
          d,
        );
      switch (op.op) {
        case "charge":
          balance -= op.amount;
          debit += op.amount;
          break;
        case "hold":
        case "extend":
          reserved += op.amount;
          debit += op.amount;
          moveHold(op.amount);
          break;
        case "capture": {
          const freed = Math.max(0, Math.min(op.amount, outstanding));
          balance -= op.amount;
          reserved -= freed;
          moveHold(-freed);
          break;
        }
        case "release":
          // frees what is actually reserved, whatever the (possibly retried) plan computed
          reserved -= outstanding;
          moveHold(-outstanding);
          sql.exec(`UPDATE pm_ledger SET amount = ? WHERE type = ? AND ref_type = ? AND ref_id = ?`, outstanding, op.op, op.refType, op.refId);
          break;
      }
    }
    if (balance || reserved) sql.exec(`UPDATE pm_account SET balance = balance + ?, reserved = reserved + ? WHERE id = 1`, balance, reserved);
    if (opts.prepaid && debit > 0) {
      const s = accountState(sql);
      // throwing inside transactionSync rolls everything back
      if (s.available < 0) throw new InsufficientCredit(`short by ${-s.available}`, { available: s.available + debit, required: debit });
    }
    for (const row of plan.usage) {
      const seen = sql.exec(`SELECT 1 AS x FROM pm_usage_seen WHERE ref_type = ? AND ref_id = ?`, row.refType, row.refId).toArray();
      if (seen.length) continue;
      sql.exec(`INSERT INTO pm_usage_seen (ref_type, ref_id) VALUES (?, ?)`, row.refType, row.refId);
      if (row.limit !== undefined) {
        const used = accountUsed(sql, row.meter, periodOf(row));
        // throwing inside transactionSync rolls the whole plan back
        if (used + row.quantity > row.limit) throw new QuotaExceeded({ meter: row.meter, limit: row.limit, used, requested: row.quantity, account: row.account });
      }
      if (row.quantity) {
        sql.exec(
          `INSERT INTO pm_counters (meter, period, used) VALUES (?, ?, ?) ON CONFLICT (meter, period) DO UPDATE SET used = used + excluded.used`,
          row.meter,
          periodOf(row),
          row.quantity,
        );
      }
      sql.exec(`INSERT INTO pm_outbox (row) VALUES (?)`, JSON.stringify(row));
    }
  });
}

packages/core/src/cloudflare/index.ts

/** Wraps `applyPlan` for an RPC method: returns a `CommitReply` instead of throwing `InsufficientCredit`. */
export function commitReply(storage: DurableStorageLike, plan: Plan, opts?: ApplyOptions): CommitReply {
  try {
    applyPlan(storage, plan, opts);
    return { ok: true };
  } catch (e) {
    if (e instanceof InsufficientCredit || (e as Error)?.name === "InsufficientCredit")
      return { ok: false, reason: "insufficient_credit", message: (e as Error).message };
    if (isQuotaExceeded(e) && e.details) return { ok: false, reason: "quota_exceeded", message: (e as unknown as Error).message, details: e.details };
    throw e;
  }
}

/**
 * The Worker-side `commit`: routes the plan to its account's DO. `stub(accountId)` returns something
 * with `commit(plan): Promise<CommitReply>` (e.g. `env.ACCOUNT.get(env.ACCOUNT.idFromName(id))`).
 */
export function doCommit(stub: (accountId: string) => { commit(plan: Plan): Promise<CommitReply> }): Commit {
  return async (plan) => {
    const accounts = new Set([...plan.ledger.map((o) => o.account), ...plan.usage.map((u) => u.account)]);
    if (accounts.size > 1) throw new Error("doCommit: a plan must belong to one account");
    const [account] = accounts;
    if (account === undefined) return;
    const reply = await stub(account).commit(plan);
    if (!reply.ok) throw reply.reason === "quota_exceeded" ? new QuotaExceeded(reply.details) : new InsufficientCredit(reply.message);
  };
}

packages/core/src/cloudflare/index.ts

/** Writes usage rows to D1 in one batch; duplicates are ignored, so retries are safe. */
export function d1UsageSink(db: D1Like): (rows: readonly UsageRow[]) => Promise<void> {
  return async (rows) => {
    if (!rows.length) return;
    const stmt = db.prepare(
      `INSERT INTO usage_events (id, account_id, meter, dims, quantity, amount_micro_usd, rate_id, detail, ref_type, ref_id, at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING`,
    );
    await db.batch(
      rows.map((u) =>
        stmt.bind(`${u.refType}:${u.refId}`, u.account, u.meter, canonicalJson(u.dims), u.quantity, u.amount, u.rateId ?? null, JSON.stringify(u.detail), u.refType, u.refId, u.at),
      ),
    );
  };
}

/** Moves one outbox batch to D1. Call from `alarm()` or `ctx.waitUntil`. Returns the number of rows shipped. */
export async function flushOutbox(sql: SqlStorage, sink: (rows: readonly UsageRow[]) => Promise<void>, limit = 100): Promise<number> {
  const { rows, upTo } = readOutbox(sql, limit);
  if (!rows.length) return 0;
  await sink(rows);
  ackOutbox(sql, upTo);
  return rows.length;
}

Counters are per (meter, period), independent of dimensions; the period comes from periodOf(row) (default: one lifetime period "*"). For embedded pricing see price() and the plan → embedded mode.

Note

The in-memory memoryAdapters (/testing) implements the same contract and is the shortest reference.