fix(paykit): harden listen webhook flow

maxktz · maxktz · commit 24545363bddc · 2026-05-08T17:00:22.000+04:00
diff --git a/apps/wh/drizzle.config.ts b/apps/wh/drizzle.config.ts
@@ -4,6 +4,7 @@ import path from "node:path";
 import { defineConfig } from "drizzle-kit";
 
 const d1StateDir = path.resolve(process.cwd(), ".wrangler/state/v3/d1/miniflare-D1DatabaseObject");
+const isGenerateCommand = process.argv.includes("generate");
 
 function resolveLocalSqliteFile(): string {
   const files = fs
@@ -26,6 +27,8 @@ export default defineConfig({
   out: "./migrations",
   schema: "./src/db/schema.ts",
   dbCredentials: {
-    url: resolveLocalSqliteFile(),
+    get url() {
+      return isGenerateCommand ? ":memory:" : resolveLocalSqliteFile();
+    },
   },
 });
diff --git a/apps/wh/package.json b/apps/wh/package.json
@@ -5,7 +5,7 @@
   "type": "module",
   "scripts": {
     "dev": "wrangler dev --local",
-    "deploy": "if [ -z \"$PAYKIT_WEBHOOK_API_BASE_URL\" ]; then printf '%s\n' 'Missing PAYKIT_WEBHOOK_API_BASE_URL'; exit 1; fi; wrangler deploy --var PAYKIT_WEBHOOK_API_BASE_URL:$PAYKIT_WEBHOOK_API_BASE_URL",
+    "deploy": "if [ -z \"$PAYKIT_WEBHOOK_API_BASE_URL\" ]; then printf '%s\n' 'Missing PAYKIT_WEBHOOK_API_BASE_URL'; exit 1; fi; wrangler deploy --var \"PAYKIT_WEBHOOK_API_BASE_URL:$PAYKIT_WEBHOOK_API_BASE_URL\"",
     "db:migrate:local": "wrangler d1 migrations apply paykit-wh --local",
     "db:migrate:remote": "wrangler d1 migrations apply paykit-wh --remote",
     "db:studio": "drizzle-kit studio --config drizzle.config.ts",
diff --git a/apps/wh/src/index.ts b/apps/wh/src/index.ts
@@ -98,6 +98,24 @@ async function getOwnedTunnel(params: {
   return rows[0] ?? null;
 }
 
+function readNumberParam(value: string | undefined, fallback: number): number {
+  if (value === undefined) {
+    return fallback;
+  }
+
+  const parsed = Number(value);
+  return Number.isNaN(parsed) ? fallback : parsed;
+}
+
+function readOptionalNumberParam(value: string | undefined): number | undefined {
+  if (value === undefined) {
+    return undefined;
+  }
+
+  const parsed = Number(value);
+  return Number.isNaN(parsed) ? undefined : parsed;
+}
+
 function buildPullableDeliveryWhere(params: {
   includeFailedBefore?: number;
   retryWindowMs: number;
@@ -142,7 +160,9 @@ async function pruneDeliveries(params: {
   const maxDeliveries = getNumericVar(params.env.MAX_DELIVERIES_PER_TUNNEL, 5000);
   const cutoff = now() - retentionDays * 24 * 60 * 60 * 1000;
 
-  await params.db.delete(delivery).where(lt(delivery.receivedAt, cutoff));
+  await params.db
+    .delete(delivery)
+    .where(and(eq(delivery.tunnelId, params.tunnelId), lt(delivery.receivedAt, cutoff)));
 
   const rows = await params.db
     .select({ count: count() })
@@ -181,9 +201,11 @@ app.post("/api/tunnels/ensure", async (c) => {
     return c.text("providerId, providerAccountId, and environment are required", 400);
   }
 
-  const retryWindowMs = Math.max(0, Number(body.retryWindowMs ?? 0));
+  const retryWindowMs = Math.max(0, readNumberParam(String(body.retryWindowMs ?? "0"), 0));
   const includeFailedBefore =
-    typeof body.includeFailedBefore === "number" ? body.includeFailedBefore : undefined;
+    typeof body.includeFailedBefore === "number" && !Number.isNaN(body.includeFailedBefore)
+      ? body.includeFailedBefore
+      : undefined;
 
   const createIfMissing = body.createIfMissing !== false;
   const existing = await db
@@ -265,9 +287,12 @@ app.get("/api/tunnels/:tunnelId/welcome", async (c) => {
     return c.text("Tunnel not found", 404);
   }
 
-  const retryWindowMs = Math.max(0, Number(c.req.query("retryWindowMs") ?? 0));
-  const includeFailedBeforeRaw = c.req.query("includeFailedBefore");
-  const includeFailedBefore = includeFailedBeforeRaw ? Number(includeFailedBeforeRaw) : undefined;
+  if (current.status === "disabled") {
+    return c.text("Tunnel disabled", 410);
+  }
+
+  const retryWindowMs = Math.max(0, readNumberParam(c.req.query("retryWindowMs"), 0));
+  const includeFailedBefore = readOptionalNumberParam(c.req.query("includeFailedBefore"));
 
   return c.json({
     pendingCount: await getPullableCount(db, {
@@ -292,6 +317,10 @@ app.post("/api/tunnels/:tunnelId/provider-webhook", async (c) => {
     return c.text("Tunnel not found", 404);
   }
 
+  if (current.status === "disabled") {
+    return c.text("Tunnel disabled", 410);
+  }
+
   const body = (await c.req.json()) as { providerWebhookEndpointId?: string };
   if (!body.providerWebhookEndpointId) {
     return c.text("providerWebhookEndpointId is required", 400);
@@ -318,11 +347,14 @@ app.get("/api/tunnels/:tunnelId/pull", async (c) => {
     return c.text("Tunnel not found", 404);
   }
 
-  const limit = clamp(Number(c.req.query("limit") ?? 30), 1, 100);
-  const offset = clamp(Number(c.req.query("offset") ?? 0), 0, 10_000);
-  const retryWindowMs = Math.max(0, Number(c.req.query("retryWindowMs") ?? 0));
-  const includeFailedBeforeRaw = c.req.query("includeFailedBefore");
-  const includeFailedBefore = includeFailedBeforeRaw ? Number(includeFailedBeforeRaw) : undefined;
+  if (current.status === "disabled") {
+    return c.text("Tunnel disabled", 410);
+  }
+
+  const limit = clamp(readNumberParam(c.req.query("limit"), 30), 1, 100);
+  const offset = clamp(readNumberParam(c.req.query("offset"), 0), 0, 10_000);
+  const retryWindowMs = Math.max(0, readNumberParam(c.req.query("retryWindowMs"), 0));
+  const includeFailedBefore = readOptionalNumberParam(c.req.query("includeFailedBefore"));
   const deliveries = await db
     .select()
     .from(delivery)
@@ -371,6 +403,10 @@ app.get("/api/deliveries/:deliveryId", async (c) => {
     return c.text("Delivery not found", 404);
   }
 
+  if (currentTunnel.status === "disabled") {
+    return c.text("Tunnel disabled", 410);
+  }
+
   return c.json({
     body: currentDelivery.body,
     deliveredAt: currentDelivery.deliveredAt,
@@ -405,6 +441,10 @@ app.post("/api/deliveries/:deliveryId/ack", async (c) => {
     return c.text("Delivery not found", 404);
   }
 
+  if (currentTunnel.status === "disabled") {
+    return c.text("Tunnel disabled", 410);
+  }
+
   await db
     .update(delivery)
     .set({ deliveredAt: now(), error: null, failedAt: null })
@@ -436,6 +476,10 @@ app.post("/api/deliveries/:deliveryId/fail", async (c) => {
     return c.text("Delivery not found", 404);
   }
 
+  if (currentTunnel.status === "disabled") {
+    return c.text("Tunnel disabled", 410);
+  }
+
   const body = (await c.req.json()) as { error?: string };
   await db
     .update(delivery)
diff --git a/packages/paykit/src/cli/commands/listen.ts b/packages/paykit/src/cli/commands/listen.ts
@@ -12,6 +12,8 @@ import { capture } from "../utils/telemetry";
 const DEFAULT_CLOUD_BASE_URL = "https://wh.paykit.sh";
 const DEFAULT_URL = "http://localhost:3000";
 const DEFAULT_BATCH_SIZE = 30;
+const DEFAULT_ERROR_BACKOFF_MS = 2_000;
+const MAX_ERROR_BACKOFF_MS = 15_000;
 const DEFAULT_POLL_INTERVAL_MS = 2_000;
 const DEFAULT_RETRY_WINDOW = "5m";
 const REPLAY_HEADER = "x-paykit-cloud-replay";
@@ -447,6 +449,10 @@ async function processPendingDeliveries(params: {
   return { hadDeliveries: true, processedCount: deliveries.length };
 }
 
+function getNextErrorBackoff(currentMs: number): number {
+  return currentMs === 0 ? DEFAULT_ERROR_BACKOFF_MS : Math.min(currentMs * 2, MAX_ERROR_BACKOFF_MS);
+}
+
 async function loadRelayRuntimeContext(params: {
   configPath?: string;
   cwd: string;
@@ -519,31 +525,41 @@ async function listenAction(options: {
   }
 
   let mode: "live" | "replay" = "replay";
+  let errorBackoffMs = 0;
 
   for (;;) {
-    const deliveries = await pullDeliveries({
-      deviceToken,
-      includeFailedBefore: mode === "replay" ? relayStartedAt : undefined,
-      limit: DEFAULT_BATCH_SIZE,
-      retryWindowMs: mode === "replay" ? retryWindowMs : 0,
-      tunnelId: tunnel.tunnelId,
-    });
+    try {
+      const deliveries = await pullDeliveries({
+        deviceToken,
+        includeFailedBefore: mode === "replay" ? relayStartedAt : undefined,
+        limit: DEFAULT_BATCH_SIZE,
+        retryWindowMs: mode === "replay" ? retryWindowMs : 0,
+        tunnelId: tunnel.tunnelId,
+      });
 
-    const result = await processPendingDeliveries({
-      devLogger,
-      deliveries,
-      deviceToken,
-      localWebhookUrl,
-      mode,
-    });
+      const result = await processPendingDeliveries({
+        devLogger,
+        deliveries,
+        deviceToken,
+        localWebhookUrl,
+        mode,
+      });
 
-    if (!result.hadDeliveries && mode === "replay") {
-      devLogger.info("replay complete, listening for new webhooks");
-      mode = "live";
-      continue;
-    }
+      errorBackoffMs = 0;
+
+      if (!result.hadDeliveries && mode === "replay") {
+        devLogger.info("replay complete, listening for new webhooks");
+        mode = "live";
+        continue;
+      }
 
-    await sleep(result.processedCount > 0 ? 250 : DEFAULT_POLL_INTERVAL_MS);
+      await sleep(result.processedCount > 0 ? 250 : DEFAULT_POLL_INTERVAL_MS);
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+      devLogger.warn(`Listen loop failed: ${message}`);
+      errorBackoffMs = getNextErrorBackoff(errorBackoffMs);
+      await sleep(errorBackoffMs);
+    }
   }
 }
 
@@ -573,9 +589,8 @@ async function enableAction(options: { config?: string; cwd: string; url: string
   devLogger.update("Ensuring webhook endpoint");
   const { webhookSecret } = await syncProviderWebhook({ deviceToken, provider, tunnel });
 
-  const localWebhookUrl = buildLocalWebhookUrl(localOrigin, config.options.basePath ?? "/paykit");
+  buildLocalWebhookUrl(localOrigin, config.options.basePath ?? "/paykit");
   devLogger.stop();
-  void localWebhookUrl;
   printEnableSummary(devLogger, {
     account,
     webhookSecret,
diff --git a/packages/paykit/src/cli/utils/device-token.ts b/packages/paykit/src/cli/utils/device-token.ts
@@ -19,14 +19,22 @@ export function getDeviceConfigPath(): string {
 
 export function getOrCreateDeviceToken(): string {
   if (fs.existsSync(DEVICE_CONFIG_PATH)) {
-    const parsed = JSON.parse(fs.readFileSync(DEVICE_CONFIG_PATH, "utf8")) as Partial<DeviceConfig>;
+    let parsed: Partial<DeviceConfig> = {};
+    try {
+      parsed = JSON.parse(fs.readFileSync(DEVICE_CONFIG_PATH, "utf8")) as Partial<DeviceConfig>;
+    } catch {
+      console.warn(`Invalid device token config at ${DEVICE_CONFIG_PATH}, creating a fresh token.`);
+    }
+
     if (typeof parsed.deviceToken === "string" && parsed.deviceToken.length > 0) {
       return parsed.deviceToken;
     }
   }
 
   const deviceToken = generateDeviceToken();
   fs.mkdirSync(path.dirname(DEVICE_CONFIG_PATH), { recursive: true });
-  fs.writeFileSync(DEVICE_CONFIG_PATH, JSON.stringify({ deviceToken }, null, 2) + "\n");
+  fs.writeFileSync(DEVICE_CONFIG_PATH, JSON.stringify({ deviceToken }, null, 2) + "\n", {
+    mode: 0o600,
+  });
   return deviceToken;
 }
diff --git a/packages/paykit/src/webhook/webhook.api.ts b/packages/paykit/src/webhook/webhook.api.ts
@@ -10,7 +10,15 @@ function headersToRecord(headers: Headers): Record<string, string> {
 }
 
 function shouldAllowStaleSignatures(headers: Headers): boolean {
-  return process.env.NODE_ENV !== "production" && headers.get("x-paykit-cloud-replay") === "1";
+  if (headers.get("x-paykit-cloud-replay") !== "1") {
+    return false;
+  }
+
+  return (
+    process.env.PAYKIT_ALLOW_STALE_SIGNATURES === "1" ||
+    process.env.NODE_ENV === "development" ||
+    process.env.NODE_ENV === "test"
+  );
 }
 
 /** Applies an incoming provider webhook payload. */
diff --git a/packages/stripe/src/stripe-provider.ts b/packages/stripe/src/stripe-provider.ts
@@ -189,6 +189,22 @@ function getStripeEnvironment(secretKey: string): string {
   return secretKey.startsWith("sk_test_") || secretKey.startsWith("rk_test_") ? "test" : "live";
 }
 
+function getStripeDisplayName(account: StripeSdk.Account): string {
+  return account.settings?.dashboard?.display_name || account.business_profile?.name || account.id;
+}
+
+function isStripeResourceMissingError(error: unknown): boolean {
+  if (!(error instanceof StripeSdk.errors.StripeError)) {
+    return false;
+  }
+
+  return (
+    error.type === "StripeInvalidRequestError" &&
+    error.code === "resource_missing" &&
+    error.statusCode === 404
+  );
+}
+
 async function retrieveExpandedSubscription(
   client: StripeSdk,
   providerSubscriptionId: string,
@@ -958,8 +974,7 @@ export function createStripeProvider(client: StripeSdk, options: StripeOptions):
 
     async getTunnelAccount() {
       const account = await client.accounts.retrieve();
-      const displayName =
-        account.settings?.dashboard?.display_name || account.business_profile?.name || account.id;
+      const displayName = getStripeDisplayName(account);
       return {
         displayName,
         environment: getStripeEnvironment(options.secretKey),
@@ -980,7 +995,11 @@ export function createStripeProvider(client: StripeSdk, options: StripeOptions):
             endpointId: endpoint.id,
             webhookSecret: options.webhookSecret,
           };
-        } catch {
+        } catch (error) {
+          if (!isStripeResourceMissingError(error)) {
+            throw error;
+          }
+
           // Fall through to create a fresh endpoint when the stored one no longer exists.
         }
       }
@@ -1013,8 +1032,7 @@ export function createStripeProvider(client: StripeSdk, options: StripeOptions):
       const mode = getStripeEnvironment(options.secretKey) === "test" ? "test mode" : "live mode";
       try {
         const account = await client.accounts.retrieve();
-        const displayName =
-          account.settings?.dashboard?.display_name || account.business_profile?.name || account.id;
+        const displayName = getStripeDisplayName(account);
 
         let webhookEndpoints: Array<{ url: string; status: string }> = [];
         try {
