Skip to content

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.

Terminal window
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
examples/messaging/src/catalog.ts
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().

examples/messaging/src/prices.ts
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):

examples/messaging/src/prices.ts
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;
}
examples/messaging/src/app.ts
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.

examples/messaging/src/app.ts
/** 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;
},
examples/messaging/src/app.ts
/** 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.

examples/messaging/src/app.ts
/**
* 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.

examples/messaging/src/account.ts
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;
}
}
examples/messaging/src/worker.ts
/** 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.