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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 00:02:02 +08:00
parent 359fdddcf4
commit 4fa7f76bc1
15 changed files with 488 additions and 1 deletions
@@ -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 () => {
@@ -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<typeof vi.fn>;
+let logSpy: ReturnType<typeof vi.spyOn>;
+
+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();
+  });
+});
@@ -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<string, number>) =>
+  ({
+    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'],
+    );
+  });
+});
@@ -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 });
  }
 }
@@ -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 });
  }
 }
@@ -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<CloudflareEnv> | 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],
+  });
+}