Public: anyone can read this; the crew can change it.
A daily sweep reconciles every paid entitlement with Stripe, so a missed webhook heals #11310
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/11247-a-daily-sweep-reconciles-every-paid-enti → main 3 commits fork 2 threads · 2 unresolved
5 files
changed +160
−54
src/billing/sync.ts added +82
| @@ -0,0 +1,82 @@ | ||
| 1 | /** | |
| 2 | * Write an entitlement from the subscription as Stripe has it now (#11245), | |
| 3 | * and the daily sweep that does it for every subscription the forge holds | |
| 4 | * (#11247). The webhook and the sweep share `syncSubscription`, so a missed | |
| 5 | * hook heals the same way a delivered one writes. Reads and writes for one | |
| 6 | * subscription run one at a time, so the last write is always from the last | |
| 7 | * read. Stripe unreachable or refusing leaves the entitlement as it was and | |
| 8 | * is audited: nothing here ever downgrades on a failed read. | |
| 9 | */ | |
| 10 | ||
| 11 | import type { App } from "../app.ts"; | |
| 12 | import { listChildrenNow } from "../actions/spawn.ts"; | |
| 13 | import { openKey } from "./connection.ts"; | |
| 14 | import { getStored, putEntitlement } from "./store.ts"; | |
| 15 | import { stripeClient, type StripeError } from "./stripe.ts"; | |
| 16 | import { entitlementFrom } from "./subscription.ts"; | |
| 17 | ||
| 18 | const fetches = new WeakMap<App, typeof fetch>(); | |
| 19 | /** Replace the fetch this app reads Stripe with; for tests. */ | |
| 20 | export function setStripeFetch(app: App, f: typeof fetch | undefined): void { | |
| 21 | if (f === undefined) fetches.delete(app); | |
| 22 | else fetches.set(app, f); | |
| 23 | } | |
| 24 | ||
| 25 | const chains = new Map<string, Promise<void>>(); | |
| 26 | /** Run `job` after every earlier job for the same key finished, so read and write are one critical section. */ | |
| 27 | function serialised(key: string, job: () => Promise<void>): Promise<void> { | |
| 28 | const run = (chains.get(key) ?? Promise.resolve()).then(job, job); | |
| 29 | const tail = run.catch(() => {}); | |
| 30 | chains.set(key, tail); | |
| 31 | void tail.then(() => { | |
| 32 | if (chains.get(key) === tail) chains.delete(key); | |
| 33 | }); | |
| 34 | return run; | |
| 35 | } | |
| 36 | ||
| 37 | /** Re-read one subscription and write its entitlement; the failure kind when Stripe could not be read, else undefined. */ | |
| 38 | export async function syncSubscription(app: App, id: string | undefined): Promise<StripeError["kind"] | undefined> { | |
| 39 | const opened = openKey(app.db, app.config.dataDir, app.bus, undefined); | |
| 40 | if (id === undefined || opened === undefined) return undefined; | |
| 41 | let failed: StripeError["kind"] | undefined; | |
| 42 | await serialised(id, async () => { | |
| 43 | const read = await stripeClient(opened.key, fetches.get(app) ?? fetch).getSubscription(id); | |
| 44 | if (!read.ok) { | |
| 45 | failed = read.error.kind; | |
| 46 | app.bus.append({ kind: "billing.sync_failed", payload: { subscription: id, reason: read.error.kind } }); | |
| 47 | return; | |
| 48 | } | |
| 49 | const put = entitlementFrom(app.db, read.value); | |
| 50 | if (put === undefined) return; | |
| 51 | const size = listChildrenNow(app).find((c) => c.name === put.forge)?.size ?? getStored(app.db, put.forge)?.plan ?? "small"; | |
| 52 | putEntitlement(app.db, put, size, "stripe"); | |
| 53 | }); | |
| 54 | return failed; | |
| 55 | } | |
| 56 | ||
| 57 | /** | |
| 58 | * Every entitlement bound to a subscription, re-read and rewritten. An | |
| 59 | * unreachable Stripe ends the sweep with one audit row (`billing.sweep_skipped`) | |
| 60 | * instead of one failure per subscription; entitlements stay as they were. | |
| 61 | */ | |
| 62 | export async function sweepEntitlements(app: App): Promise<{ read: number; skipped: boolean }> { | |
| 63 | const ids = app.db | |
| 64 | .query<{ subscription_id: string }, []>("SELECT subscription_id FROM entitlement WHERE subscription_id != '' AND provider = 'stripe' ORDER BY forge") | |
| 65 | .all() | |
| 66 | .map((r) => r.subscription_id); | |
| 67 | let read = 0; | |
| 68 | for (const id of ids) { | |
| 69 | let failed: StripeError["kind"] | undefined; | |
| 70 | try { | |
| 71 | failed = await syncSubscription(app, id); | |
| 72 | } catch { | |
| 73 | failed = "unreachable"; | |
| 74 | } | |
| 75 | if (failed === "unreachable") { | |
| 76 | app.bus.append({ kind: "billing.sweep_skipped", payload: { reason: "unreachable", read } }); | |
| 77 | return { read, skipped: true }; | |
| 78 | } | |
| 79 | if (failed === undefined) read += 1; | |
| 80 | } | |
| 81 | return { read, skipped: false }; | |
| 82 | } |
src/http/stripe_hook.ts +4 −49
| @@ -13,12 +13,10 @@ | ||
| 13 | 13 | */ |
| 14 | 14 | |
| 15 | 15 | import type { App } from "../app.ts"; |
| 16 | import { listChildrenNow } from "../actions/spawn.ts"; | |
| 17 | 16 | import { bindCheckout } from "../billing/checkout_completed.ts"; |
| 18 | import { openKey, openWebhookSecret } from "../billing/connection.ts"; | |
| 19 | import { getStored, putEntitlement } from "../billing/store.ts"; | |
| 20 | import { stripeClient } from "../billing/stripe.ts"; | |
| 21 | import { entitlementFrom, subscriptionIdOf } from "../billing/subscription.ts"; | |
| 17 | import { openWebhookSecret } from "../billing/connection.ts"; | |
| 18 | import { syncSubscription } from "../billing/sync.ts"; | |
| 19 | import { subscriptionIdOf } from "../billing/subscription.ts"; | |
| 22 | 20 | import { verifyStripeSignature } from "../billing/signature.ts"; |
| 23 | 21 | import { RateLimiter } from "../core/ratelimit.ts"; |
| 24 | 22 | import type { Ctx } from "./context.ts"; |
| @@ -54,49 +52,6 @@ export function setStripeHookCaps(app: App, perClient: number, wholeForge: numbe | ||
| 54 | 52 | bucketsOf(app, perClient, wholeForge); |
| 55 | 53 | } |
| 56 | 54 | |
| 57 | const fetches = new WeakMap<App, typeof fetch>(); | |
| 58 | /** Replace the fetch this app's hook reads Stripe with; for tests. */ | |
| 59 | export function setStripeHookFetch(app: App, f: typeof fetch | undefined): void { | |
| 60 | if (f === undefined) fetches.delete(app); | |
| 61 | else fetches.set(app, f); | |
| 62 | } | |
| 63 | ||
| 64 | const chains = new Map<string, Promise<void>>(); | |
| 65 | /** Run `job` after every earlier job for the same key finished, so read and write are one critical section. */ | |
| 66 | function serialised(key: string, job: () => Promise<void>): Promise<void> { | |
| 67 | const run = (chains.get(key) ?? Promise.resolve()).then(job, job); | |
| 68 | const tail = run.catch(() => {}); | |
| 69 | chains.set(key, tail); | |
| 70 | void tail.then(() => { | |
| 71 | if (chains.get(key) === tail) chains.delete(key); | |
| 72 | }); | |
| 73 | return run; | |
| 74 | } | |
| 75 | ||
| 76 | /** | |
| 77 | * Write the entitlement from the subscription as Stripe has it now, never from | |
| 78 | * the event's body. Reads and writes for one subscription run one at a time, so | |
| 79 | * the last write is always from the last read. Stripe unreachable or refusing | |
| 80 | * leaves the entitlement as it was (and is audited): the hook still answers | |
| 81 | * 200, so Stripe does not retry into a stale read. | |
| 82 | */ | |
| 83 | async function syncSubscription(app: App, event: unknown): Promise<void> { | |
| 84 | const id = subscriptionIdOf(event); | |
| 85 | const opened = openKey(app.db, app.config.dataDir, app.bus, undefined); | |
| 86 | if (id === undefined || opened === undefined) return; | |
| 87 | await serialised(id, async () => { | |
| 88 | const read = await stripeClient(opened.key, fetches.get(app) ?? fetch).getSubscription(id); | |
| 89 | if (!read.ok) { | |
| 90 | app.bus.append({ kind: "billing.sync_failed", payload: { subscription: id, reason: read.error.kind } }); | |
| 91 | return; | |
| 92 | } | |
| 93 | const put = entitlementFrom(app.db, read.value); | |
| 94 | if (put === undefined) return; | |
| 95 | const size = listChildrenNow(app).find((c) => c.name === put.forge)?.size ?? getStored(app.db, put.forge)?.plan ?? "small"; | |
| 96 | putEntitlement(app.db, put, size, "stripe"); | |
| 97 | }); | |
| 98 | } | |
| 99 | ||
| 100 | 55 | const answer = (status: number, headers?: Record<string, string>): Response => |
| 101 | 56 | new Response(null, { status, headers: { "cache-control": "no-store", ...headers } }); |
| 102 | 57 | |
| @@ -158,7 +113,7 @@ export async function stripeHook(ctx: Ctx): Promise<Response> { | ||
| 158 | 113 | })(); |
| 159 | 114 | if (fresh && (type.startsWith("customer.subscription.") || type.startsWith("invoice."))) { |
| 160 | 115 | try { |
| 161 | await syncSubscription(app, event); | |
| 116 | await syncSubscription(app, subscriptionIdOf(event)); | |
| 162 | 117 | } catch (cause) { |
| 163 | 118 | // never let a read of Stripe turn into a retry storm; the audit says it failed |
| 164 | 119 | app.bus.append({ kind: "billing.sync_failed", payload: { event: e.id, reason: cause instanceof Error ? cause.name : "error" } }); |
src/main.ts +5
| @@ -4,6 +4,7 @@ | ||
| 4 | 4 | * environment-only — a handful of variables, all with sane defaults. |
| 5 | 5 | */ |
| 6 | 6 | |
| 7 | import { sweepEntitlements } from "./billing/sync.ts"; | |
| 7 | 8 | import { sweepStarts } from "./actions/start_sweep.ts"; |
| 8 | 9 | import { scanNightly } from "./checks/scan.ts"; |
| 9 | 10 | import { nightlyDue } from "./checks/scan_store.ts"; |
| @@ -697,6 +698,9 @@ const staleTimer = setInterval(() => void during("asks", tickAsks), 120_000); | ||
| 697 | 698 | // archived after a month (`actions/start_sweep.ts`, #10912): hourly, and on boot. |
| 698 | 699 | const startsTimer = setInterval(() => during("starts", () => void sweepStarts(app)), 3_600_000); |
| 699 | 700 | during("starts", () => void sweepStarts(app)); |
| 701 | // Every paid entitlement is re-read from Stripe once a day, so a missed webhook heals (#11247). | |
| 702 | const entitlementTimer = setInterval(() => void during("entitlements", () => sweepEntitlements(app)), 86_400_000); | |
| 703 | void during("entitlements", () => sweepEntitlements(app)); | |
| 700 | 704 | void during("asks", tickAsks); |
| 701 | 705 | |
| 702 | 706 | // Every start is counted, and one that follows a process that never said |
| @@ -925,6 +929,7 @@ function drain(signal: NodeJS.Signals): Promise<unknown> { | ||
| 925 | 929 | clearInterval(syncTimer); |
| 926 | 930 | clearInterval(staleTimer); |
| 927 | 931 | clearInterval(startsTimer); |
| 932 | clearInterval(entitlementTimer); | |
| 928 | 933 | clearInterval(followTimer); |
| 929 | 934 | stopClogTick(); |
| 930 | 935 | stopKeeper(); |
test/stripe-hook.test.ts +6 −5
| @@ -8,7 +8,8 @@ import { createApp, type App } from "../src/app.ts"; | ||
| 8 | 8 | import { sealKey, sealWebhook } from "../src/billing/connection.ts"; |
| 9 | 9 | import { serve, type Servers } from "../src/http/server.ts"; |
| 10 | 10 | import { getStored } from "../src/billing/store.ts"; |
| 11 | import { refusedCount, setStripeHookCaps, setStripeHookFetch } from "../src/http/stripe_hook.ts"; | |
| 11 | import { refusedCount, setStripeHookCaps } from "../src/http/stripe_hook.ts"; | |
| 12 | import { setStripeFetch } from "../src/billing/sync.ts"; | |
| 12 | 13 | |
| 13 | 14 | const secret = "whsec_hook_test_secret"; |
| 14 | 15 | let dir: string; |
| @@ -145,13 +146,13 @@ describe("POST /hooks/stripe", () => { | ||
| 145 | 146 | JSON.stringify({ id, type, data: { object: type.startsWith("invoice.") ? { id: "in_1", subscription: "sub_1", status } : { id: "sub_1", status } } }); |
| 146 | 147 | const deliver = async (body: string) => (await post(body, await sign(body))).status; |
| 147 | 148 | beforeAll(() => { |
| 148 | setStripeHookFetch(app, (async (url: string) => { | |
| 149 | setStripeFetch(app, (async (url: string) => { | |
| 149 | 150 | calls.push(String(url)); |
| 150 | 151 | if (down) throw new Error("offline"); |
| 151 | 152 | return new Response(JSON.stringify(sub()), { status: 200 }); |
| 152 | 153 | }) as unknown as typeof fetch); |
| 153 | 154 | }); |
| 154 | afterAll(() => setStripeHookFetch(app, undefined)); | |
| 155 | afterAll(() => setStripeFetch(app, undefined)); | |
| 155 | 156 | |
| 156 | 157 | test("events in reverse order end on what Stripe says now, not on the event bodies", async () => { |
| 157 | 158 | calls = []; |
| @@ -190,7 +191,7 @@ describe("POST /hooks/stripe", () => { | ||
| 190 | 191 | |
| 191 | 192 | test("deliveries for one subscription are serialised: a slow older read cannot land last", async () => { |
| 192 | 193 | const gates: Array<(status: string) => void> = []; |
| 193 | setStripeHookFetch(app, (async () => | |
| 194 | setStripeFetch(app, (async () => | |
| 194 | 195 | new Promise<Response>((resolve) => { |
| 195 | 196 | gates.push((status) => resolve(new Response(JSON.stringify({ ...sub(), status }), { status: 200 }))); |
| 196 | 197 | })) as unknown as typeof fetch); |
| @@ -208,7 +209,7 @@ describe("POST /hooks/stripe", () => { | ||
| 208 | 209 | }); |
| 209 | 210 | |
| 210 | 211 | test("a failed read is audited", async () => { |
| 211 | setStripeHookFetch(app, (async () => { throw new Error("offline"); }) as unknown as typeof fetch); | |
| 212 | setStripeFetch(app, (async () => { throw new Error("offline"); }) as unknown as typeof fetch); | |
| 212 | 213 | expect(await deliver(ev("evt_f1", "invoice.paid", "paid"))).toBe(200); |
| 213 | 214 | const n = app.db.query<{ n: number }, []>("SELECT count(*) n FROM event WHERE kind = 'billing.sync_failed'").get()?.n ?? 0; |
| 214 | 215 | expect(n).toBeGreaterThan(0); |
test/stripe-sweep.test.ts added +63
| @@ -0,0 +1,63 @@ | ||
| 1 | // The daily sweep re-reads every subscribed entitlement from Stripe (#11247). | |
| 2 | import { afterAll, beforeAll, describe, expect, test } from "bun:test"; | |
| 3 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 4 | import { tmpdir } from "node:os"; | |
| 5 | import { join } from "node:path"; | |
| 6 | import { createApp, type App } from "../src/app.ts"; | |
| 7 | import { sealKey } from "../src/billing/connection.ts"; | |
| 8 | import { getStored, putEntitlement } from "../src/billing/store.ts"; | |
| 9 | import { setStripeFetch, sweepEntitlements } from "../src/billing/sync.ts"; | |
| 10 | ||
| 11 | let dir: string; | |
| 12 | let app: App; | |
| 13 | beforeAll(async () => { | |
| 14 | dir = await mkdtemp(join(tmpdir(), "stripe-sweep-")); | |
| 15 | app = createApp({ dataDir: dir, port: 0 }); | |
| 16 | sealKey(app.db, dir, "sk_test_abcdefgh12345678"); | |
| 17 | // bound to a subscription, but its webhook was never delivered: the row still says free | |
| 18 | putEntitlement(app.db, { forge: "acme", plan: "free", status: "none", source: "stripe", provider: "stripe", customerId: "cus_1", subscriptionId: "sub_1" }, "small", "test"); | |
| 19 | putEntitlement(app.db, { forge: "beta", plan: "free", status: "none", source: "stripe", provider: "stripe", customerId: "cus_2", subscriptionId: "sub_2" }, "small", "test"); | |
| 20 | }); | |
| 21 | afterAll(async () => { | |
| 22 | setStripeFetch(app, undefined); | |
| 23 | await rm(dir, { recursive: true, force: true }); | |
| 24 | }); | |
| 25 | const count = (kind: string) => app.db.query<{ n: number }, [string]>("SELECT count(*) n FROM event WHERE kind = ?1").get(kind)?.n ?? 0; | |
| 26 | ||
| 27 | describe("sweepEntitlements", () => { | |
outdated
This test seeds one subscription, so it would pass even if the sweep did not stop early and audited a failure per subscription. Seed at least two Stripe-bound entitlements, and assert that the fetch was called once and that billing.sweep_skipped is 1. Add a case where Stripe refuses (a 401 or 404 response): the entitlement is unchanged, billing.sync_failed is audited, and the sweep carries on to the next subscription. |
||
| 28 | test("an unreachable Stripe skips the sweep, says so once, and changes nothing", async () => { | |
| 29 | let calls = 0; | |
| 30 | setStripeFetch(app, (async () => { calls += 1; throw new Error("offline"); }) as unknown as typeof fetch); | |
| 31 | const before = [getStored(app.db, "acme"), getStored(app.db, "beta")]; | |
| 32 | expect(await sweepEntitlements(app)).toEqual({ read: 0, skipped: true }); | |
| 33 | expect(calls).toBe(1); // stopped at the first failure, not one read per subscription | |
| 34 | expect(count("billing.sweep_skipped")).toBe(1); | |
| 35 | expect([getStored(app.db, "acme"), getStored(app.db, "beta")]).toEqual(before as never); | |
| 36 | }); | |
| 37 | ||
| 38 | test("a refused read leaves that entitlement, is audited, and the sweep carries on", async () => { | |
| 39 | const calls: string[] = []; | |
| 40 | const failedBefore = count("billing.sync_failed"); | |
| 41 | setStripeFetch(app, (async (url: string) => { | |
| 42 | calls.push(String(url)); | |
| 43 | if (String(url).endsWith("sub_1")) return new Response(JSON.stringify({ error: { type: "invalid_request_error", code: "resource_missing" } }), { status: 404 }); | |
| 44 | return new Response(JSON.stringify({ id: "sub_2", status: "active", customer: "cus_2", metadata: { forge: "beta", plan: "medium" } }), { status: 200 }); | |
| 45 | }) as unknown as typeof fetch); | |
| 46 | expect(await sweepEntitlements(app)).toEqual({ read: 1, skipped: false }); | |
| 47 | expect(calls.length).toBe(2); | |
| 48 | expect(count("billing.sync_failed") - failedBefore).toBe(1); | |
| 49 | expect(getStored(app.db, "acme")).toMatchObject({ plan: "free", status: "none" }); | |
| 50 | expect(getStored(app.db, "beta")).toMatchObject({ plan: "medium", status: "active" }); | |
| 51 | }); | |
| 52 | ||
| 53 | test("an entitlement whose webhook never came is healed from Stripe", async () => { | |
| 54 | const calls: string[] = []; | |
| 55 | setStripeFetch(app, (async (url: string) => { | |
| 56 | calls.push(String(url)); | |
| 57 | return new Response(JSON.stringify({ id: String(url).split("/").pop(), status: "active", customer: "cus_1", metadata: { forge: String(url).endsWith("sub_1") ? "acme" : "beta", plan: "medium" }, items: { data: [{ current_period_end: 1900000000 }] } }), { status: 200 }); | |
| 58 | }) as unknown as typeof fetch); | |
| 59 | expect(await sweepEntitlements(app)).toEqual({ read: 2, skipped: false }); | |
| 60 | expect(calls).toContain("https://api.stripe.com/v1/subscriptions/sub_1"); | |
| 61 | expect(getStored(app.db, "acme")).toMatchObject({ plan: "medium", status: "active", periodEnd: 1900000000, source: "stripe" }); | |
| 62 | }); | |
| 63 | }); |
new above
offline
A refused read (kind other than unreachable) still increments
read. Count only successful reads, otherwise the returned count overstates what was healed.