Adapter yazma

Kütüphanenin dünyaya iki kapısı var: tarife ve kullanımı soran getRate, planı yazan commit. İkisi de senin kodundur; referans adapter’lar birer başlangıç noktasıdır.

getRate sözleşmesi

(meter, dims, ctx, at) => { rate, usedSoFar?, carry?, limit? } | null | undefined   // ya da Promise'i

• Tarifenin tamamını dön; kademe seçimini rate() yapar. Ham obje dönebilirsin, kütüphane doğrular (invalid_rate).
• usedSoFar: bu hesabın bu sayaçta, seçtiğin dönemde şimdiye kadarki adedi. Kademeli, volume, package, sabit ücretli ve cumulative tarifelerde zorunlu; eksikse usage_required (kurallar).
• carry: yalnızca cumulative yuvarlamada, bir önceki usage satırının detail.carry’si.
• limit: bu sayacın usedSoFar ile aynı dönemdeki üst sınırı; aşan çağrı quota_exceeded (kota). usedSoFar olmadan usage_required.
• at’i nasıl yorumlayacağın (olay anı mı, hesap anı mı) senin kararın. Dönem sonu gözlemleri at: window.to − 1 ile yapılır, çünkü pencereler yarı açıktır ve window.to bir sonraki dönemi seçer (A98).
• Sayacın kapsamı senin: referans adapter’lar sayaç başına, boyutsuz sayar. Boyut başına kademe istiyorsan getRate’te dims ile say (sqliteAdapter.usedSoFar({ dims }); A100).
• null = fiyat yok → ifMissing. Fırlatma: bir getRate hatası observe’dan istisna olarak çıkar.
• Aynı çağrıda aynı (meter, dims) için bir kez çağrılır. Dört parametrenin dördünü de yaz (A94).

commit sözleşmesi

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

Kural	Neden
Ya hep ya hiç: ledger ve usage satırları tek transaction’da	Yarım yazılmış plan tutarsız bakiye demektir
İdempotent: aynı (refType, refId) ikinci kez gelirse sessizce atla (ledger’da op ile birlikte)	Uygulama retry’ı güvenli olsun
Bakiye yalnızca gerçekten eklenen ledger satırlarıyla hareket eder	Retry bakiyeyi iki kez düşürmesin
Prepaid: balance − reserved < 0 olacaksa InsufficientCredit fırlat, hiçbir şey yazma	Bakiye kapısı yazımla aynı atomik adımda
limit taşıyan yeni bir usage satırı için: satırın dönemindeki sayaç + quantity > limit olacaksa QuotaExceeded fırlat, hiçbir şey yazma	Kota kapısı da yazımla aynı atomik adımda (A107). SQLite: windowOf, DO: periodOf
charge bakiyeyi düşürür (negatifse artırır); hold/extend reserved’ı artırır ve borç sayılır; capture bakiyeyi düşürür; release rezervasyonu serbest bırakır	Kullanılabilir = balance − reserved
Hold başına gerçekten ayrılan tutarı tut: capture rezervasyonu en fazla onu kadar düşürür, release planın tutarını değil onu serbest bırakır (A102)	Farklı tutarla tekrarlanan bir hold no-op yazılır ama yeniden hesaplanmış bir upperBound döner; plana güvenen release rezervasyonu kaydırır
Başka her hata için fırlat	Lib commit_failed döner, cause’ta hata

InsufficientCredit name ile de eşleşir, bu yüzden RPC ve realm sınırlarından geçer. Boş planlar (ledger ve usage boş) commit’e hiç gelmez.

commitContract

pricemeter/testing her commit’in geçmesi gereken sözleşmeyi bir test takımı olarak verir. Kendi adapter’ın için seed (hesaba bakiye ver) ve snapshot (satır sayıları ve kullanılabilir bakiye) yaz:

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 }, …]

Postpaid bir adapter için { prepaid: false } ver; bakiye kontrolleri atlanır.

SQLite

pricemeter/sqlite, node:sqlite, bun:sqlite ve better-sqlite3 ile çalışır (yapısal SqliteLike). migrate, commit, usedSoFar, spent, balance, credit verir.

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

Plan başına tek bir BEGIN IMMEDIATE. Satırlar INSERT … ON CONFLICT DO NOTHING ile yazılır (OR IGNORE değil: o NOT NULL ihlallerini de yutar); bakiye ve rezervasyon yalnızca gerçekten eklenen ledger satırları için hareket eder:

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

Şema: usage_events (UNIQUE (ref_type, ref_id)), ledger_entries (UNIQUE (type, ref_type, ref_id)), holds (hold başına ayrılan tutar), accounts. dims kanonik JSON (sıralı anahtarlar) olarak saklanır, böylece usedSoFar({ dims }) eşitlikle sorgulanabilir.

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: hesap başına bir Durable Object; ledger, bakiye ve usedSoFar sayaçları DO’nun SQLite deposunda.

• applyPlan(storage, plan, { prepaid, periodOf }) her şeyi transactionSync içinde yapar: idempotent ledger, bakiye kapısı, dönem başına sayaçlar ve usage outbox’ı. Fırlatmak her şeyi geri alır.
• doCommit(stub) Worker tarafındaki commit’tir: planı hesabın DO’suna yönlendirir. Hatalar RPC’den güvenilir geçmediği için yetersizlik bir değer olarak döner (commitReply) ve Worker’da yeniden InsufficientCredit olarak fırlatılır.
• Usage satırları aynı transaction’da outbox’a girer; flushOutbox + d1UsageSink onları D1’e idempotent taşır (waitUntil ya da 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;
}

Sayaçlar (meter, period) başınadır, boyutlardan bağımsızdır; dönem periodOf(row) ile seçilir (varsayılan tek ömürlük dönem "*"). Gömülü fiyatlama için bkz. price() ve plan → gömülü mod.

Not

Bellek içi memoryAdapters (/testing) aynı sözleşmeyi uygular ve en kısa referanstır.