Example: messaging (Cloudflare)
examples/messaging: a Cellonay-style OTP + WhatsApp service. A Cloudflare Worker, one Durable Object per account (ledger, balance, counters), D1 (usage), KV (OTP records), a Queue (late-published prices) and a monthly cron. Prepaid.
bunx vitest run examples/messaging # tests without workerd (a DurableObject shim)cd examples/messaging && bun run dev # wrangler dev| File | Role |
|---|---|
src/catalog.ts |
The shared catalog; no adapters |
src/prices.ts |
Layered price table + route-cost fallback |
src/app.ts |
Worker logic: OTP send / resend / verify / expire, WhatsApp windows, monthly fees |
src/account.ts |
The account DO’s logic (AccountCore), independent of the DurableObject class |
src/do.ts, src/worker.ts |
Thin wrappers Cloudflare instantiates |
Catalog
Section titled “Catalog”import { buildMetering, type Metering, type PricedLineOf, type RefOf } from "pricemeter";import { z } from "zod";
export const Context = z.object({ accountId: z.string(), plan: z.enum(["free", "pro", "enterprise"]),});export type Context = z.infer<typeof Context>;
const country = z.string().length(2);
export const catalog = buildMetering() .context(Context) .refs(["otp_send", "wa_window", "month"]) // pool first: every message, whatever the channel, counts toward the monthly free platform quota .meter("msg/free_pool") .meter("otp/sms", { country }, { feeds: { "msg/free_pool": 1 } }) .meter("otp/whatsapp", { country }, { feeds: { "msg/free_pool": 1 } }) .meter("otp/verify") .meter("wa/window", { country }) .meter("ip/dedicated");
export type Catalog = typeof catalog;type State = Catalog extends Metering<infer S> ? S : never;/** A priced line of this catalog, as `plan.observe(...).lines` returns it. */export type CatalogLine = PricedLineOf<State>;export type CatalogRef = RefOf<State>;The pool (msg/free_pool) is declared first; SMS and WhatsApp feed it at weight 1, so a channel-independent quota of 1,000 free messages a month (B4) is one tiered tariff. The catalog has no adapters: the Worker binds it, the DO only uses it for price().
Prices
Section titled “Prices”export const PRICE_ROWS: PriceVersion[] = [ // SMS: Turkey gets cheaper above 10k/month (B1) { id: "sms-tr-list", meter: "otp/sms", scope: "list", dims: { country: "TR" }, effectiveFrom: 0, rate: { model: "graduated", tiers: [ { from: 0, unitPriceMicroUsd: 20_000 }, { from: 10_000, unitPriceMicroUsd: 15_000 }, ], policy: { tierPeriod: "month" }, }, }, { id: "sms-pro", meter: "otp/sms", scope: "plan:pro", dims: { country: "TR" }, effectiveFrom: 0, rate: flat(18_000) }, { id: "wa-list", meter: "otp/whatsapp", scope: "list", effectiveFrom: 0, rate: flat(10_000) }, { id: "verify-list", meter: "otp/verify", scope: "list", effectiveFrom: 0, rate: flat(5_000) }, // platform fee per message after 1,000 free each month, across channels (B4) { id: "pool-list", meter: "msg/free_pool", scope: "list", effectiveFrom: 0, rate: { model: "graduated", tiers: [ { from: 0, unitPriceMicroUsd: 0 }, { from: 1000, unitPriceMicroUsd: 2_000 }, ], policy: { tierPeriod: "month" }, }, }, // WhatsApp conversation windows, per country, as Meta publishes them { id: "wa-window-tr", meter: "wa/window", scope: "list", dims: { country: "TR" }, effectiveFrom: 0, rate: flat(9_000) }, // Dedicated IP: $50 once per month, whatever else happens (B11) { id: "ip-list", meter: "ip/dedicated", scope: "list", effectiveFrom: 0, rate: { model: "graduated", tiers: [{ from: 0, unitPriceMicroUsd: 0, flatMicroUsd: 50_000_000 }], policy: { tierPeriod: "month" } }, },];Turkish SMS gets cheaper after 10,000 a month (B1), the pro plan has its own row, and a dedicated IP costs $50 once a month (B11). For SMS countries without a table row, the price is the carrier’s cost at at plus a 25% margin (B23):
export function routeRate(routes: readonly RouteCost[], countryCode: string, at: number): Rate | null { const current = routes.filter((r) => r.country === countryCode && r.from <= at).sort((a, b) => b.from - a.from)[0]; if (!current) return null; return { id: `route-${countryCode}-${current.from}`, model: "graduated", tiers: [{ from: 0, unitPriceMicroUsd: Math.ceil(current.costMicroUsd * ROUTE_MARGIN) }] };}
/** Table first, then route cost for SMS. */export function findRate(i: { rows: readonly PriceVersion[]; routes: readonly RouteCost[]; meter: string; dims: Record<string, unknown>; ctx: Context; at: number }): Rate | null { const fromTable = resolveLayered({ rows: i.rows, scopes: scopesFor(i.ctx), dims: i.dims, at: i.at, meter: i.meter }); if (fromTable) return fromTable; if (i.meter === "otp/sms") return routeRate(i.routes, String(i.dims.country), i.at); return null;}Worker: bind and the OTP flow
Section titled “Worker: bind and the OTP flow”const metering = catalog.bind({ getRate: async (meter, dims, ctx, at) => { const rate: Rate | null = findRate({ rows: await deps.priceRows(), routes: await deps.routes(), meter, dims, ctx, at }); if (!rate) return null; const multiTier = rate.tiers.length > 1 || rate.tiers.some((t) => t.flatMicroUsd); return multiTier ? { rate, usedSoFar: await deps.account(ctx.accountId).used(meter, monthKey({ at })) } : { rate }; }, commit: doCommit((id) => deps.account(id)),});usedSoFar is only asked for multi-tier or flat-fee tariffs, and read from the account’s DO for the monthKey({ at }) period.
/** Send: reserve the message segments and one verification in a single two-line hold. */async sendOtp(i: { id: string; ctx: Context; channel: "sms" | "whatsapp"; country: string; text: string; at: number }) { if (sandboxed(i.ctx)) return { ok: true as const, sandbox: true }; const meter = i.channel === "sms" ? "otp/sms" : "otp/whatsapp"; const h = await metering.hold( [ { meter, dims: { country: i.country }, quantity: segments(i.text) }, { meter: "otp/verify", quantity: 1 }, ], { type: "otp_send", id: i.id }, i.ctx, { at: i.at }, ); if (!h.ok) return h; await deps.otps.put({ id: i.id, ctx: i.ctx, channel: i.channel, country: i.country, hold: h.hold, sends: 1 }); return h;},/** Verified: capture what was actually sent plus the verification, release the rest. */async verifyOtp(i: { id: string; delivered: number; at: number }) { const otp = await deps.otps.get(i.id); if (!otp) return { ok: false as const, reason: "unknown_otp" }; const meter = otp.channel === "sms" ? "otp/sms" : "otp/whatsapp"; const c = await metering.capture( otp.hold, [ { meter, quantity: i.delivered }, { meter: "otp/verify", quantity: 1 }, ], otp.ctx, { at: i.at }, ); if (!c.ok) return c; const r = await metering.release(c.hold, otp.ctx, { at: i.at }); if (r.ok) await deps.otps.delete(i.id); return { ...c, released: r.ok };},Sending opens a two-line hold (segments + one verification) stored on the OTP record; a resend extends the same hold; verification captures the segments sent and the verification and releases the rest. An expired OTP pays only for delivered segments.
Embedded pricing and late prices
Section titled “Embedded pricing and late prices”/** * A WhatsApp conversation window closed. Embedded mode: the Worker plans with its view, the account DO * re-prices with its own counters and applies. If the country's price isn't published yet, queue it: * the retry uses the original `at`. */async closeWindow(m: WindowMessage) { if (sandboxed(m.ctx)) return { ok: true as const, sandbox: true }; const planned = await metering.plan.observe("wa/window", { country: m.country }, 1, { type: "wa_window", id: m.windowId }, m.ctx, { at: m.at }); if (!planned.result.ok) { if (planned.result.reason === "no_price") await deps.windowQueue.send(m); return planned.result; } return deps.account(m.ctx.accountId).observeEmbedded(planned.lines, { type: "wa_window", id: m.windowId }, m.ctx);},The Worker prices the lines with plan.observe but does not write them; the account’s DO re-prices them with its own counters and applies them in the same step (embedded mode). If the country’s window price is not published yet, the result is no_price: the message is queued and retried an hour later with the same at.
export class AccountCore implements AccountApi { constructor( private storage: DurableStorageLike, private sink: (rows: readonly UsageRow[]) => Promise<void>, private opts: { prepaid: boolean } = { prepaid: true }, ) { migrateAccount(storage.sql); }
async commit(plan: Plan): Promise<CommitReply> { return commitReply(this.storage, plan, { prepaid: this.opts.prepaid, periodOf }); }
async used(meter: string, period: string): Promise<number> { return accountUsed(this.storage.sql, meter, period); }
async observeEmbedded(lines: CatalogLine[], ref: CatalogRef, ctx: Context): Promise<Result> { // Nothing awaits between reading counters and writing: the DO's input gate makes this atomic. const fresh = lines.map((l) => ({ ...l, usedSoFar: accountUsed(this.storage.sql, l.meter, periodOf({ at: l.at ?? Date.now() })) })); const { result, plan } = catalog.price(fresh, ref, ctx); if (!result.ok) return result; const reply = commitReply(this.storage, plan, { prepaid: this.opts.prepaid, periodOf }); return reply.ok ? result : { ok: false, reason: "insufficient_credit" }; }
async credit(amountMicroUsd: number): Promise<void> { creditAccount(this.storage.sql, amountMicroUsd); }
async state(): Promise<AccountState> { return accountState(this.storage.sql); }
async flush(): Promise<number> { let total = 0; for (let n = -1; n !== 0; total += n) n = await flushOutbox(this.storage.sql, this.sink); return total; }}HTTP mapping
Section titled “HTTP mapping”/** HTTP mapping of results: the status code is the app's decision, not the library's. */function reply(r: { ok: boolean; reason?: string }) { if (r.ok) return json(r); const status = r.reason === "insufficient_credit" ? 402 : r.reason === "no_price" ? 422 : r.reason === "unknown_otp" ? 404 : 400; return json(r, status);}The status code is the application’s decision, not the library’s: insufficient_credit → 402, no_price → 422.