From 4fa7f76bc16b4df31cfc23908faaea251597d369 Mon Sep 17 00:00:00 2001 From: Huang Xin Date: Mon, 22 Jun 2026 00:02:02 +0800 Subject: [PATCH] feat(payment): observability for store subscription webhooks (#4704) * feat(payment): observability for store subscription webhooks Add monitoring for the App Store / Google Play webhooks so store-side subscription changes are observable on Cloudflare, where stored Workers Logs are head-sampled at 1% and would miss almost all low-volume webhook events. - Add iap/telemetry.ts: every webhook invocation emits a structured log line (streamed in full by `wrangler tail`) and a Cloudflare Analytics Engine data point (100% capture, independent of log sampling). Writes no-op off the Worker runtime, mirroring the getCloudflareContext guard in deepl/translate.ts. - Instrument both webhook routes to record outcome (handled, skipped, rejected, error), notification type, status, reason, and latency on every return path. - Add GET /api/cron/iap-reconcile: a CRON_SECRET-protected sweep that counts drift (rows still active while their store expiry has passed = a missed webhook) in both IAP tables and records a reconcile metric. Detection-only; never mutates state. - Add the IAP_WEBHOOK_AE Analytics Engine binding to wrangler.toml. New configuration: CRON_SECRET (reconcile auth) and an iap_webhooks Analytics Engine dataset bound as IAP_WEBHOOK_AE. The reconcile route is triggered on a schedule (a Cloudflare Cron Trigger worker that fetches the URL, or any external scheduler) with an Authorization bearer header. Co-Authored-By: Claude Opus 4.8 (1M context) * refactor(payment): move reconciliation to a dedicated cron Worker Replace the public CRON_SECRET-protected /api/cron/iap-reconcile route with a dedicated Cloudflare Cron Worker. A Cron Trigger invokes the worker's scheduled() handler directly, so there is no public HTTP surface and no shared request secret to manage - the strongest option on Cloudflare (OpenNext's generated worker only exports `fetch`, so the main worker cannot host a scheduled() handler). - Add workers/iap-reconcile: a self-contained worker (own package.json, tsconfig, wrangler.toml) matching the existing workers/send-email convention, registered in pnpm-workspace.yaml. Hourly Cron Trigger; reads the IAP tables via the Supabase service role and records a drift metric to the shared iap_webhooks Analytics Engine dataset. - Reconcile logic lives in workers/iap-reconcile/src/reconcile.ts and is unit-tested from the app suite. - Remove the public route and its test; drop the now-unused recordIapReconcile from iap/telemetry.ts (webhook telemetry unchanged). Configuration: set SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY as secrets on the worker and deploy it with `wrangler deploy` from its directory. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- .../app/api/iap-notifications-routes.test.ts | 21 ++++- .../libs/payment/iap-telemetry.test.ts | 80 +++++++++++++++++++ .../__tests__/workers/iap-reconcile.test.ts | 72 +++++++++++++++++ .../src/app/api/apple/notifications/route.ts | 29 +++++++ .../src/app/api/google/notifications/route.ts | 35 ++++++++ .../src/libs/payment/iap/telemetry.ts | 65 +++++++++++++++ .../workers/iap-reconcile/.gitignore | 1 + .../workers/iap-reconcile/package.json | 19 +++++ .../workers/iap-reconcile/src/index.ts | 46 +++++++++++ .../workers/iap-reconcile/src/reconcile.ts | 56 +++++++++++++ .../workers/iap-reconcile/tsconfig.json | 16 ++++ .../workers/iap-reconcile/wrangler.toml | 25 ++++++ apps/readest-app/wrangler.toml | 7 ++ pnpm-lock.yaml | 16 ++++ pnpm-workspace.yaml | 1 + 15 files changed, 488 insertions(+), 1 deletion(-) create mode 100644 apps/readest-app/src/__tests__/libs/payment/iap-telemetry.test.ts create mode 100644 apps/readest-app/src/__tests__/workers/iap-reconcile.test.ts create mode 100644 apps/readest-app/src/libs/payment/iap/telemetry.ts create mode 100644 apps/readest-app/workers/iap-reconcile/.gitignore create mode 100644 apps/readest-app/workers/iap-reconcile/package.json create mode 100644 apps/readest-app/workers/iap-reconcile/src/index.ts create mode 100644 apps/readest-app/workers/iap-reconcile/src/reconcile.ts create mode 100644 apps/readest-app/workers/iap-reconcile/tsconfig.json create mode 100644 apps/readest-app/workers/iap-reconcile/wrangler.toml diff --git a/apps/readest-app/src/__tests__/app/api/iap-notifications-routes.test.ts b/apps/readest-app/src/__tests__/app/api/iap-notifications-routes.test.ts index 79e6cedc..e025dcb0 100644 --- a/apps/readest-app/src/__tests__/app/api/iap-notifications-routes.test.ts +++ b/apps/readest-app/src/__tests__/app/api/iap-notifications-routes.test.ts @@ -16,6 +16,11 @@ vi.mock('@/libs/payment/iap/google/notifications', () => ({ handleGoogleNotification: hooks.handleGoogleNotification, })); +const telemetry = vi.hoisted(() => ({ recordIapWebhook: vi.fn() })); +vi.mock('@/libs/payment/iap/telemetry', () => ({ + recordIapWebhook: telemetry.recordIapWebhook, +})); + import { POST as applePOST } from '@/app/api/apple/notifications/route'; import { POST as googlePOST } from '@/app/api/google/notifications/route'; @@ -29,6 +34,7 @@ const jsonReq = (url: string, body: unknown) => beforeEach(() => { hooks.handleAppleNotification.mockReset(); hooks.handleGoogleNotification.mockReset(); + telemetry.recordIapWebhook.mockReset(); hooks.handleAppleNotification.mockResolvedValue({ handled: true, status: 'active' }); hooks.handleGoogleNotification.mockResolvedValue({ handled: true, status: 'active' }); delete process.env['GOOGLE_RTDN_VERIFICATION_TOKEN']; @@ -39,14 +45,24 @@ describe('POST /api/apple/notifications', () => { const res = await applePOST(jsonReq('https://web.readest.com/api/apple/notifications', {})); expect(res.status).toBe(400); expect(hooks.handleAppleNotification).not.toHaveBeenCalled(); + expect(telemetry.recordIapWebhook).toHaveBeenCalledWith( + expect.objectContaining({ + provider: 'apple', + outcome: 'rejected', + reason: 'missing_payload', + }), + ); }); - it('processes a signed payload', async () => { + it('processes a signed payload and records the outcome', async () => { const res = await applePOST( jsonReq('https://web.readest.com/api/apple/notifications', { signedPayload: 'jws' }), ); expect(res.status).toBe(200); expect(hooks.handleAppleNotification).toHaveBeenCalledWith('jws'); + expect(telemetry.recordIapWebhook).toHaveBeenCalledWith( + expect.objectContaining({ provider: 'apple', outcome: 'handled' }), + ); }); it('returns 500 when processing throws so Apple retries', async () => { @@ -68,6 +84,9 @@ describe('POST /api/google/notifications', () => { ); expect(res.status).toBe(401); expect(hooks.handleGoogleNotification).not.toHaveBeenCalled(); + expect(telemetry.recordIapWebhook).toHaveBeenCalledWith( + expect.objectContaining({ provider: 'google', outcome: 'rejected', reason: 'invalid_token' }), + ); }); it('processes a message when the token matches', async () => { diff --git a/apps/readest-app/src/__tests__/libs/payment/iap-telemetry.test.ts b/apps/readest-app/src/__tests__/libs/payment/iap-telemetry.test.ts new file mode 100644 index 00000000..eb5caa57 --- /dev/null +++ b/apps/readest-app/src/__tests__/libs/payment/iap-telemetry.test.ts @@ -0,0 +1,80 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; + +// IAP webhook telemetry: a structured log line (for `wrangler tail`) plus a +// Cloudflare Analytics Engine data point (100% capture, independent of the +// head-sampled Workers Logs). Must no-op gracefully off the Worker runtime. + +const cf = vi.hoisted(() => ({ getCloudflareContext: vi.fn() })); +vi.mock('@opennextjs/cloudflare', () => ({ getCloudflareContext: cf.getCloudflareContext })); + +import { recordIapWebhook } from '@/libs/payment/iap/telemetry'; + +type DataPoint = { indexes?: string[]; blobs?: (string | null)[]; doubles?: number[] }; + +let writeDataPoint: ReturnType; +let logSpy: ReturnType; + +const firstDataPoint = (): DataPoint => writeDataPoint.mock.calls[0]![0] as DataPoint; + +beforeEach(() => { + writeDataPoint = vi.fn(); + cf.getCloudflareContext.mockReturnValue({ env: { IAP_WEBHOOK_AE: { writeDataPoint } } }); + logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); +}); + +afterEach(() => { + logSpy.mockRestore(); + cf.getCloudflareContext.mockReset(); +}); + +describe('recordIapWebhook', () => { + it('writes an Analytics Engine data point indexed by provider', () => { + recordIapWebhook({ + provider: 'apple', + outcome: 'handled', + notificationType: 'DID_RENEW', + status: 'active', + durationMs: 12, + }); + + expect(writeDataPoint).toHaveBeenCalledTimes(1); + const point = firstDataPoint(); + expect(point.indexes).toEqual(['apple']); + expect(point.blobs).toEqual(['webhook', 'apple', 'handled', 'DID_RENEW', 'active', '']); + expect(point.doubles).toEqual([12]); + }); + + it('emits a tagged structured log line', () => { + recordIapWebhook({ provider: 'google', outcome: 'error', reason: 'db down', durationMs: 5 }); + + const logged = JSON.parse(logSpy.mock.calls.at(-1)![0] as string); + expect(logged).toMatchObject({ + tag: 'iap-webhook', + kind: 'webhook', + provider: 'google', + outcome: 'error', + reason: 'db down', + }); + }); + + it('does not throw and still logs outside the Worker runtime', () => { + cf.getCloudflareContext.mockImplementation(() => { + throw new Error('no cloudflare context'); + }); + + expect(() => + recordIapWebhook({ provider: 'apple', outcome: 'handled', durationMs: 1 }), + ).not.toThrow(); + expect(logSpy).toHaveBeenCalled(); + expect(writeDataPoint).not.toHaveBeenCalled(); + }); + + it('no-ops the data point when the AE binding is absent', () => { + cf.getCloudflareContext.mockReturnValue({ env: {} }); + + expect(() => + recordIapWebhook({ provider: 'apple', outcome: 'handled', durationMs: 1 }), + ).not.toThrow(); + expect(writeDataPoint).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/readest-app/src/__tests__/workers/iap-reconcile.test.ts b/apps/readest-app/src/__tests__/workers/iap-reconcile.test.ts new file mode 100644 index 00000000..788d3184 --- /dev/null +++ b/apps/readest-app/src/__tests__/workers/iap-reconcile.test.ts @@ -0,0 +1,72 @@ +import { describe, it, expect } from 'vitest'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +// Unit tests for the dedicated reconcile cron Worker's pure logic. The Worker +// itself (workers/iap-reconcile) is a self-contained Cloudflare project; only +// its runtime-agnostic helpers are exercised here. +import { + buildReconcileDataPoint, + countSubscriptionDrift, +} from '../../../workers/iap-reconcile/src/reconcile'; + +const makeSupabase = (counts: Record) => + ({ + from: (table: string) => ({ + select: () => ({ + eq: () => ({ + lt: () => Promise.resolve({ count: counts[table] ?? 0, error: null }), + }), + }), + }), + }) as unknown as SupabaseClient; + +describe('countSubscriptionDrift', () => { + it('counts active-but-expired rows in both store tables', async () => { + const supabase = makeSupabase({ + apple_iap_subscriptions: 2, + google_iap_subscriptions: 1, + }); + + await expect(countSubscriptionDrift(supabase)).resolves.toEqual({ + appleDrift: 2, + googleDrift: 1, + }); + }); + + it('reports zero drift when the stores are clean', async () => { + await expect(countSubscriptionDrift(makeSupabase({}))).resolves.toEqual({ + appleDrift: 0, + googleDrift: 0, + }); + }); + + it('throws when a query fails', async () => { + const supabase = { + from: () => ({ + select: () => ({ + eq: () => ({ + lt: () => Promise.resolve({ count: null, error: { message: 'boom' } }), + }), + }), + }), + } as unknown as SupabaseClient; + + await expect(countSubscriptionDrift(supabase)).rejects.toThrow(/boom/); + }); +}); + +describe('buildReconcileDataPoint', () => { + it('flags drift and records counts as doubles', () => { + expect(buildReconcileDataPoint({ appleDrift: 2, googleDrift: 1, durationMs: 30 })).toEqual({ + indexes: ['reconcile'], + blobs: ['reconcile', 'drift'], + doubles: [2, 1, 30], + }); + }); + + it('flags ok when there is no drift', () => { + expect(buildReconcileDataPoint({ appleDrift: 0, googleDrift: 0, durationMs: 9 }).blobs).toEqual( + ['reconcile', 'ok'], + ); + }); +}); diff --git a/apps/readest-app/src/app/api/apple/notifications/route.ts b/apps/readest-app/src/app/api/apple/notifications/route.ts index 3237b109..aeeaf684 100644 --- a/apps/readest-app/src/app/api/apple/notifications/route.ts +++ b/apps/readest-app/src/app/api/apple/notifications/route.ts @@ -1,29 +1,58 @@ import { NextResponse } from 'next/server'; import { handleAppleNotification } from '@/libs/payment/iap/apple/notifications'; +import { recordIapWebhook } from '@/libs/payment/iap/telemetry'; // App Store Server Notifications V2 endpoint. Configure this URL in App Store // Connect (Production and Sandbox). The payload is signed by Apple and verified // inside `handleAppleNotification`, so no separate authentication is needed. export async function POST(request: Request) { + const startedAt = Date.now(); + let signedPayload: unknown; try { const body = await request.json(); signedPayload = body?.signedPayload; } catch { + recordIapWebhook({ + provider: 'apple', + outcome: 'rejected', + reason: 'invalid_body', + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ error: 'Invalid request body' }, { status: 400 }); } if (typeof signedPayload !== 'string' || !signedPayload) { + recordIapWebhook({ + provider: 'apple', + outcome: 'rejected', + reason: 'missing_payload', + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ error: 'Missing signedPayload' }, { status: 400 }); } try { const result = await handleAppleNotification(signedPayload); + recordIapWebhook({ + provider: 'apple', + outcome: result.handled ? 'handled' : 'skipped', + notificationType: result.notificationType, + status: result.status, + reason: result.reason, + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ received: true, ...result }); } catch (error) { // Respond 500 so Apple retries transient failures (e.g. a database hiccup). const message = error instanceof Error ? error.message : 'Unknown error'; console.error('Apple notification error:', message); + recordIapWebhook({ + provider: 'apple', + outcome: 'error', + reason: message, + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ error: message }, { status: 500 }); } } diff --git a/apps/readest-app/src/app/api/google/notifications/route.ts b/apps/readest-app/src/app/api/google/notifications/route.ts index 3599ebbd..1e00326f 100644 --- a/apps/readest-app/src/app/api/google/notifications/route.ts +++ b/apps/readest-app/src/app/api/google/notifications/route.ts @@ -1,15 +1,24 @@ import { NextResponse } from 'next/server'; import { handleGoogleNotification } from '@/libs/payment/iap/google/notifications'; +import { recordIapWebhook } from '@/libs/payment/iap/telemetry'; // Google Play Real-Time Developer Notifications (RTDN) endpoint, delivered via a // Cloud Pub/Sub push subscription. The push URL must include the shared secret // (`?token=...`) matching GOOGLE_RTDN_VERIFICATION_TOKEN; the notification state // itself is re-verified against the Play Developer API in the handler. export async function POST(request: Request) { + const startedAt = Date.now(); + const expectedToken = process.env['GOOGLE_RTDN_VERIFICATION_TOKEN']; if (expectedToken) { const token = new URL(request.url).searchParams.get('token'); if (token !== expectedToken) { + recordIapWebhook({ + provider: 'google', + outcome: 'rejected', + reason: 'invalid_token', + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ error: 'Invalid verification token' }, { status: 401 }); } } else { @@ -21,22 +30,48 @@ export async function POST(request: Request) { const body = await request.json(); messageData = body?.message?.data; } catch { + recordIapWebhook({ + provider: 'google', + outcome: 'rejected', + reason: 'invalid_body', + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ error: 'Invalid request body' }, { status: 400 }); } // Pub/Sub may deliver an empty message (e.g. during endpoint validation). // Acknowledge it so it is not retried. if (typeof messageData !== 'string' || !messageData) { + recordIapWebhook({ + provider: 'google', + outcome: 'skipped', + reason: 'empty_message', + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ received: true, handled: false, reason: 'empty_message' }); } try { const result = await handleGoogleNotification(messageData); + recordIapWebhook({ + provider: 'google', + outcome: result.handled ? 'handled' : 'skipped', + notificationType: result.notificationType, + status: result.status, + reason: result.reason, + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ received: true, ...result }); } catch (error) { // Respond 500 so Pub/Sub retries transient failures. const message = error instanceof Error ? error.message : 'Unknown error'; console.error('Google notification error:', message); + recordIapWebhook({ + provider: 'google', + outcome: 'error', + reason: message, + durationMs: Date.now() - startedAt, + }); return NextResponse.json({ error: message }, { status: 500 }); } } diff --git a/apps/readest-app/src/libs/payment/iap/telemetry.ts b/apps/readest-app/src/libs/payment/iap/telemetry.ts new file mode 100644 index 00000000..dd3602b1 --- /dev/null +++ b/apps/readest-app/src/libs/payment/iap/telemetry.ts @@ -0,0 +1,65 @@ +import { getCloudflareContext } from '@opennextjs/cloudflare'; + +// Minimal local typing for the Analytics Engine binding (the project does not +// depend on @cloudflare/workers-types). Mirrors the pattern in +// `src/pages/api/deepl/translate.ts` for the KV binding. +interface AnalyticsEngineDataset { + writeDataPoint(event: { + indexes?: string[]; + blobs?: (string | null)[]; + doubles?: number[]; + }): void; +} + +interface CloudflareEnv { + IAP_WEBHOOK_AE?: AnalyticsEngineDataset; +} + +const TELEMETRY_TAG = 'iap-webhook'; + +export type IapWebhookOutcome = 'handled' | 'skipped' | 'rejected' | 'error'; + +export interface IapWebhookEvent { + provider: 'apple' | 'google'; + outcome: IapWebhookOutcome; + notificationType?: string | number; + status?: string; + reason?: string; + durationMs: number; +} + +const getAnalyticsEngine = (): AnalyticsEngineDataset | undefined => { + try { + const env = getCloudflareContext().env as Partial | undefined; + return env?.IAP_WEBHOOK_AE; + } catch { + // getCloudflareContext throws outside the Worker runtime (local dev / tests). + return undefined; + } +}; + +/** + * Record the outcome of an App Store / Google Play webhook invocation. + * + * Emits two signals: + * - a structured log line, streamed in full by `wrangler tail` (stored Workers + * Logs are head-sampled, so they are unreliable for low-volume webhooks); + * - a Cloudflare Analytics Engine data point for 100% capture and dashboards, + * which no-ops when the binding is absent (local dev / preview). + */ +export function recordIapWebhook(event: IapWebhookEvent): void { + console.log(JSON.stringify({ tag: TELEMETRY_TAG, kind: 'webhook', ...event })); + + getAnalyticsEngine()?.writeDataPoint({ + indexes: [event.provider], + blobs: [ + 'webhook', + event.provider, + event.outcome, + String(event.notificationType ?? ''), + event.status ?? '', + event.reason ?? '', + ], + doubles: [event.durationMs], + }); +} diff --git a/apps/readest-app/workers/iap-reconcile/.gitignore b/apps/readest-app/workers/iap-reconcile/.gitignore new file mode 100644 index 00000000..3c3629e6 --- /dev/null +++ b/apps/readest-app/workers/iap-reconcile/.gitignore @@ -0,0 +1 @@ +node_modules diff --git a/apps/readest-app/workers/iap-reconcile/package.json b/apps/readest-app/workers/iap-reconcile/package.json new file mode 100644 index 00000000..6cd105a7 --- /dev/null +++ b/apps/readest-app/workers/iap-reconcile/package.json @@ -0,0 +1,19 @@ +{ + "name": "readest-iap-reconcile-worker", + "version": "0.1.0", + "private": true, + "description": "Cloudflare Cron Worker that reconciles IAP subscription drift (missed App Store / Google Play webhooks)", + "scripts": { + "deploy": "wrangler deploy", + "dev": "wrangler dev", + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@supabase/supabase-js": "^2.45.0" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.0.0", + "typescript": "^5.0.0", + "wrangler": "^4.0.0" + } +} diff --git a/apps/readest-app/workers/iap-reconcile/src/index.ts b/apps/readest-app/workers/iap-reconcile/src/index.ts new file mode 100644 index 00000000..7a8202de --- /dev/null +++ b/apps/readest-app/workers/iap-reconcile/src/index.ts @@ -0,0 +1,46 @@ +import { createClient } from '@supabase/supabase-js'; +import { buildReconcileDataPoint, countSubscriptionDrift } from './reconcile'; + +// Dedicated Cloudflare Cron Worker for IAP subscription reconciliation. Invoked +// by a Cron Trigger (see wrangler.toml) through the `scheduled()` handler, so it +// has NO public HTTP surface and needs no shared request secret. It reads the +// IAP tables directly and records a drift metric to the shared `iap_webhooks` +// Analytics Engine dataset. + +interface Env { + SUPABASE_URL: string; + SUPABASE_SERVICE_ROLE_KEY: string; + IAP_WEBHOOK_AE?: AnalyticsEngineDataset; +} + +export default { + async scheduled(_controller, env, _ctx): Promise { + const startedAt = Date.now(); + const supabase = createClient(env.SUPABASE_URL, env.SUPABASE_SERVICE_ROLE_KEY, { + auth: { persistSession: false, autoRefreshToken: false, detectSessionInUrl: false }, + }); + + try { + const { appleDrift, googleDrift } = await countSubscriptionDrift(supabase); + const durationMs = Date.now() - startedAt; + + // Mirrors the webhook log tag so both stream together under `wrangler tail`. + console.log( + JSON.stringify({ + tag: 'iap-webhook', + kind: 'reconcile', + appleDrift, + googleDrift, + durationMs, + }), + ); + env.IAP_WEBHOOK_AE?.writeDataPoint( + buildReconcileDataPoint({ appleDrift, googleDrift, durationMs }), + ); + } catch (error) { + // Rethrow so Cloudflare records a failed scheduled invocation. + console.error('IAP reconcile failed:', error instanceof Error ? error.message : error); + throw error; + } + }, +} satisfies ExportedHandler; diff --git a/apps/readest-app/workers/iap-reconcile/src/reconcile.ts b/apps/readest-app/workers/iap-reconcile/src/reconcile.ts new file mode 100644 index 00000000..9df33ba2 --- /dev/null +++ b/apps/readest-app/workers/iap-reconcile/src/reconcile.ts @@ -0,0 +1,56 @@ +import type { SupabaseClient } from '@supabase/supabase-js'; + +// Pure reconciliation logic, unit-tested from the app suite +// (src/__tests__/workers/iap-reconcile.test.ts). Kept separate from index.ts so +// it has no dependency on the Worker runtime globals. + +export interface SubscriptionDrift { + appleDrift: number; + googleDrift: number; +} + +const IAP_SUBSCRIPTION_TABLES = { + apple: 'apple_iap_subscriptions', + google: 'google_iap_subscriptions', +} as const; + +const countDrift = async (supabase: SupabaseClient, table: string): Promise => { + // Drift = a subscription still marked active in our DB whose store expiry has + // already passed, i.e. a renewal/expiry webhook we never received. + const { count, error } = await supabase + .from(table) + .select('*', { count: 'exact', head: true }) + .eq('status', 'active') + .lt('expires_date', new Date().toISOString()); + + if (error) { + throw new Error(`Failed to count drift in ${table}: ${error.message}`); + } + return count ?? 0; +}; + +export async function countSubscriptionDrift(supabase: SupabaseClient): Promise { + const [appleDrift, googleDrift] = await Promise.all([ + countDrift(supabase, IAP_SUBSCRIPTION_TABLES.apple), + countDrift(supabase, IAP_SUBSCRIPTION_TABLES.google), + ]); + return { appleDrift, googleDrift }; +} + +export interface ReconcileDataPoint { + indexes: string[]; + blobs: string[]; + doubles: number[]; +} + +/** Build the Analytics Engine data point for a reconciliation sweep. Writes to + * the same `iap_webhooks` dataset the main worker uses for webhook events. */ +export function buildReconcileDataPoint( + event: SubscriptionDrift & { durationMs: number }, +): ReconcileDataPoint { + return { + indexes: ['reconcile'], + blobs: ['reconcile', event.appleDrift + event.googleDrift > 0 ? 'drift' : 'ok'], + doubles: [event.appleDrift, event.googleDrift, event.durationMs], + }; +} diff --git a/apps/readest-app/workers/iap-reconcile/tsconfig.json b/apps/readest-app/workers/iap-reconcile/tsconfig.json new file mode 100644 index 00000000..5d0fd6bf --- /dev/null +++ b/apps/readest-app/workers/iap-reconcile/tsconfig.json @@ -0,0 +1,16 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "Bundler", + "lib": ["ES2022"], + "types": ["@cloudflare/workers-types"], + "strict": true, + "noUnusedLocals": true, + "noUnusedParameters": true, + "noEmit": true, + "skipLibCheck": true, + "esModuleInterop": true + }, + "include": ["src/**/*.ts"] +} diff --git a/apps/readest-app/workers/iap-reconcile/wrangler.toml b/apps/readest-app/workers/iap-reconcile/wrangler.toml new file mode 100644 index 00000000..f0dd7e16 --- /dev/null +++ b/apps/readest-app/workers/iap-reconcile/wrangler.toml @@ -0,0 +1,25 @@ +name = "readest-iap-reconcile" +main = "src/index.ts" +compatibility_date = "2026-04-10" +compatibility_flags = ["nodejs_compat"] +workers_dev = false + +# Hourly drift sweep. Catches subscription state changes that never arrived as a +# webhook (a row still marked active while its store expiry has passed). Invokes +# the worker's scheduled() handler directly — no public HTTP endpoint. +[triggers] +crons = ["7 * * * *"] + +# Low volume; capture every run (the main worker stays head-sampled at 1%). +[observability] +enabled = true +head_sampling_rate = 1 + +# Same dataset the main worker writes webhook events to. +[[analytics_engine_datasets]] +binding = "IAP_WEBHOOK_AE" +dataset = "iap_webhooks" + +# Required secrets (do not commit). Set from this directory with: +# wrangler secret put SUPABASE_URL +# wrangler secret put SUPABASE_SERVICE_ROLE_KEY diff --git a/apps/readest-app/wrangler.toml b/apps/readest-app/wrangler.toml index df3c0097..cebafcd8 100644 --- a/apps/readest-app/wrangler.toml +++ b/apps/readest-app/wrangler.toml @@ -40,3 +40,10 @@ id = "fa6e3eb1988a424bbf3c952e01650ced" [[r2_buckets]] binding = "NEXT_INC_CACHE_R2_BUCKET" bucket_name = "readest-next-inc-cache" + +# Per-event metrics for the App Store / Google Play subscription webhooks and the +# reconciliation sweep. 100% capture, independent of the head-sampled Workers +# Logs above. Queryable via the Analytics Engine SQL API. +[[analytics_engine_datasets]] +binding = "IAP_WEBHOOK_AE" +dataset = "iap_webhooks" diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 49193e5e..d30cefce 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -624,6 +624,22 @@ importers: specifier: ^5.1.4 version: 5.1.4(webpack@5.106.2) + apps/readest-app/workers/iap-reconcile: + dependencies: + '@supabase/supabase-js': + specifier: ^2.45.0 + version: 2.105.4 + devDependencies: + '@cloudflare/workers-types': + specifier: ^4.0.0 + version: 4.20260518.1 + typescript: + specifier: ^5.0.0 + version: 5.9.3 + wrangler: + specifier: ^4.0.0 + version: 4.90.0(@cloudflare/workers-types@4.20260518.1) + apps/readest-app/workers/send-email: dependencies: '@supabase/supabase-js': diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index dcd8b3e4..b1cb6aed 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -1,6 +1,7 @@ packages: - apps/* - apps/readest-app/workers/send-email + - apps/readest-app/workers/iap-reconcile - apps/readest-app/extensions/* - packages/foliate-js