import { internalMutation, internalQuery, type MutationCtx, type QueryCtx, } from "../_generated/server"; import type { Doc } from "../_generated/dataModel"; import { v } from "convex/values"; import { handleSubscriptionActive, handleSubscriptionRenewed, handleSubscriptionOnHold, handleSubscriptionCancelled, handleSubscriptionPlanChanged, handleSubscriptionExpired, handleSubscriptionUpdated, handlePaymentOrRefundEvent, handleDisputeEvent, } from "./subscriptionHelpers"; const MAX_FAILURE_MESSAGE_LENGTH = 1000; const MAX_FAILURE_DATA_KEYS = 100; const MAX_DIAGNOSTIC_ROWS = 250; const GLOBAL_FAILURE_SUMMARY_KEY = "global" as const; type EventTypeCount = { eventType: string; count: number }; function asRecord(value: unknown): Record | null { return value !== null && typeof value === "object" && !Array.isArray(value) ? (value as Record) : null; } function boundedString(value: unknown, maxLength: number): string | undefined { return typeof value === "string" && value.length > 0 ? value.slice(0, maxLength) : undefined; } /** Keep failure rows useful to operators without copying payload values. */ function sanitizeFailureMessage(value: string): string { return value .replace(/[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}/g, "[redacted-email]") .replace(/\b(bearer|token|secret|api[_-]?key|password)\s*[:=]\s*\S+/gi, "$1=[redacted]") .slice(0, MAX_FAILURE_MESSAGE_LENGTH); } function extractFailureContext(rawPayload: unknown) { const payload = asRecord(rawPayload); const data = asRecord(payload?.data); const customer = asRecord(data?.customer); return { dodoSubscriptionId: boundedString(data?.subscription_id, 200), dodoPaymentId: boundedString(data?.payment_id, 200), dodoCustomerId: boundedString(customer?.customer_id, 200), dataKeys: data ? Object.keys(data).sort().slice(0, MAX_FAILURE_DATA_KEYS) : [], }; } function adjustEventTypeCount( counts: EventTypeCount[], eventType: string, delta: number, ): EventTypeCount[] { const next = counts.map((entry) => ({ ...entry })); const entry = next.find((candidate) => candidate.eventType === eventType); if (entry) { entry.count += delta; } else if (delta > 0) { next.push({ eventType, count: delta }); } return next .filter((candidate) => candidate.count > 0) .sort((a, b) => a.eventType.localeCompare(b.eventType)); } async function readFailureSummaryLock(ctx: MutationCtx) { const summary = await ctx.db .query("paymentWebhookFailureSummary") .withIndex("by_key", (q) => q.eq("key", GLOBAL_FAILURE_SUMMARY_KEY)) .first(); if (!summary) { throw new Error( "Dodo webhook failure summary is not seeded; run payments/webhookMutations:_seedFailureSummary", ); } return summary; } async function updateFailureSummary( ctx: MutationCtx, summary: Doc<"paymentWebhookFailureSummary">, existing: Doc<"paymentWebhookFailures"> | null, eventType: string, updatedAt: number, ) { let unresolvedCount = summary.unresolvedCount; let eventTypes = summary.eventTypes; if (existing?.unresolved) { // A repeated failure normally keeps the same event type. Handle a // provider correction defensively so the aggregate remains truthful. if (existing.eventType !== eventType) { eventTypes = adjustEventTypeCount(eventTypes, existing.eventType, -1); eventTypes = adjustEventTypeCount(eventTypes, eventType, 1); } } else { unresolvedCount += 1; eventTypes = adjustEventTypeCount(eventTypes, eventType, 1); } await ctx.db.patch(summary._id, { unresolvedCount, eventTypes, updatedAt, }); } async function getUnresolvedFailureSummary(ctx: QueryCtx) { const rows = await ctx.db .query("paymentWebhookFailureSummary") .withIndex("by_key", (q) => q.eq("key", "global")) // A deploy seed and the daily self-heal can both observe an empty index // and insert concurrently. Operational reads must remain available until // the next seed pass removes the extra row. .first(); return { unresolvedCount: rows?.unresolvedCount ?? 0, eventTypes: rows?.eventTypes.map((entry) => entry.eventType) ?? [], }; } export const recordWebhookFailure = internalMutation({ args: { webhookId: v.string(), eventType: v.string(), rawPayload: v.any(), timestamp: v.number(), receivedAt: v.optional(v.number()), errorKind: v.string(), errorMessage: v.string(), }, handler: async (ctx, args) => { // This pre-seeded summary document is also the OCC serialization point // for the failure row + aggregate update. Index reads do not serialize a // brand-new webhookId race on their own, so the lock must exist before // either mutation can insert the first row for an ID. const summary = await readFailureSummaryLock(ctx); const receivedAt = args.receivedAt ?? Date.now(); const context = extractFailureContext(args.rawPayload); const existing = await ctx.db .query("paymentWebhookFailures") .withIndex("by_webhookId", (q) => q.eq("webhookId", args.webhookId)) .first(); const attemptCount = (existing?.attemptCount ?? 0) + 1; const failure = { webhookId: args.webhookId, eventType: args.eventType, ...context, errorKind: sanitizeFailureMessage(args.errorKind), errorMessage: sanitizeFailureMessage(args.errorMessage), eventTimestamp: args.timestamp, lastSeenAt: receivedAt, attemptCount, unresolved: true, // A retry after manual resolution is a fresh unresolved incident, so // clear the old resolution marker while retaining receivedAt/history. resolvedAt: undefined, resolvedBy: undefined, resolutionNote: undefined, }; if (existing) { await ctx.db.patch(existing._id, failure); } else { await ctx.db.insert("paymentWebhookFailures", { ...failure, receivedAt, }); } await updateFailureSummary(ctx, summary, existing, args.eventType, receivedAt); const currentSummary = await getUnresolvedFailureSummary(ctx); return { isNew: !existing, attemptCount, errorKind: failure.errorKind, errorMessage: failure.errorMessage, ...currentSummary, }; }, }); async function removeUnresolvedFailureFromSummary( ctx: MutationCtx, summary: Doc<"paymentWebhookFailureSummary">, existing: Doc<"paymentWebhookFailures">, updatedAt: number, ) { if (!existing.unresolved) { return; } await ctx.db.patch(summary._id, { unresolvedCount: Math.max(0, summary.unresolvedCount - 1), eventTypes: adjustEventTypeCount(summary.eventTypes, existing.eventType, -1), updatedAt, }); } /** * Idempotently creates the aggregate/serialization document used by the * failure lifecycle. This runs after Convex deploy and is also safe to rerun. */ export const _seedFailureSummary = internalMutation({ args: {}, handler: async (ctx): Promise<{ seeded: number; deduped: number }> => { const existing = await ctx.db .query("paymentWebhookFailureSummary") .withIndex("by_key", (q) => q.eq("key", GLOBAL_FAILURE_SUMMARY_KEY)) .collect(); if (existing.length > 0) { // Concurrent first seeds can both insert because an empty index range is // not a document-backed OCC lock. All operational reads use the oldest // row via `.first()`, so retain that authority and remove later extras. existing.sort((a, b) => a._creationTime - b._creationTime); let deduped = 0; for (let index = 1; index < existing.length; index++) { const extra = existing[index]; if (extra !== undefined) { await ctx.db.delete(extra._id); deduped += 1; } } return { seeded: 0, deduped }; } await ctx.db.insert("paymentWebhookFailureSummary", { key: GLOBAL_FAILURE_SUMMARY_KEY, unresolvedCount: 0, eventTypes: [], updatedAt: Date.now(), }); return { seeded: 1, deduped: 0 }; }, }); /** * Emits one grouped Convex auto-Sentry event after a failure row commits. * Convex forwards structured console errors to Sentry; using a non-throwing * mutation keeps the provider retry path and scheduled-function bookkeeping * independent of the ops signal. */ export const reportDodoWebhookFailure = internalMutation({ args: { webhookId: v.string(), eventType: v.string(), errorKind: v.string(), errorMessage: v.string(), attemptCount: v.number(), unresolvedCount: v.number(), eventTypes: v.array(v.string()), }, handler: async (_ctx, args) => { // sentry-coverage-ok: Convex auto-Sentry forwards this structured ops // signal. It is intentionally non-throwing so the scheduler does not // retry or create an unhandled test/deployment failure. console.error( `[webhook] Dodo processing failure (webhookId=${args.webhookId}, eventType=${args.eventType}, errorKind=${args.errorKind}, attemptCount=${args.attemptCount}, unresolvedCount=${args.unresolvedCount}, affectedEventTypes=${args.eventTypes.join(",")}): ${sanitizeFailureMessage(args.errorMessage)}`, ); }, }); /** * Shared resolution transition for manual resolution and provider-retry * recovery: one timestamp marks the failure row resolved and rebalances the * aggregate. Callers keep their own lookup and missing-row policy. */ async function applyWebhookFailureResolution( ctx: MutationCtx, summary: Doc<"paymentWebhookFailureSummary">, existing: Doc<"paymentWebhookFailures">, attribution: { resolvedBy: string; resolutionNote?: string }, ) { const resolvedAt = Date.now(); await ctx.db.patch(existing._id, { unresolved: false, resolvedAt, resolvedBy: attribution.resolvedBy, resolutionNote: attribution.resolutionNote, }); await removeUnresolvedFailureFromSummary(ctx, summary, existing, resolvedAt); } export const resolveWebhookFailure = internalMutation({ args: { webhookId: v.string(), resolvedBy: v.string(), resolutionNote: v.optional(v.string()), }, handler: async (ctx, args) => { const summary = await readFailureSummaryLock(ctx); const existing = await ctx.db .query("paymentWebhookFailures") .withIndex("by_webhookId", (q) => q.eq("webhookId", args.webhookId)) .first(); if (!existing) { throw new Error(`No Dodo webhook failure found for ${args.webhookId}`); } await applyWebhookFailureResolution(ctx, summary, existing, { resolvedBy: args.resolvedBy.slice(0, 200), resolutionNote: args.resolutionNote ? sanitizeFailureMessage(args.resolutionNote) : undefined, }); }, }); /** Clear a transient incident when the same provider delivery later succeeds. */ export const markWebhookFailureRecovered = internalMutation({ args: { webhookId: v.string() }, handler: async (ctx, { webhookId }) => { const existing = await ctx.db .query("paymentWebhookFailures") .withIndex("by_webhookId", (q) => q.eq("webhookId", webhookId)) .first(); if (!existing?.unresolved) { return; } const summary = await readFailureSummaryLock(ctx); await applyWebhookFailureResolution(ctx, summary, existing, { resolvedBy: "dodo-retry", resolutionNote: "Processed successfully on provider retry", }); }, }); export const listUnresolvedWebhookFailures = internalQuery({ args: { limit: v.optional(v.number()), }, handler: async (ctx, args) => { const limit = Math.max( 1, Math.min(args.limit ?? 100, MAX_DIAGNOSTIC_ROWS), ); return await ctx.db .query("paymentWebhookFailures") .withIndex("by_unresolved_lastSeenAt", (q) => q.eq("unresolved", true)) .order("desc") .take(limit); }, }); export const getWebhookFailureDiagnostics = internalQuery({ args: { limit: v.optional(v.number()), }, handler: async (ctx, args) => { const limit = Math.max( 1, Math.min(args.limit ?? 100, MAX_DIAGNOSTIC_ROWS), ); const summary = await ctx.db .query("paymentWebhookFailureSummary") .withIndex("by_key", (q) => q.eq("key", "global")) .first(); const failures = await ctx.db .query("paymentWebhookFailures") .withIndex("by_unresolved_lastSeenAt", (q) => q.eq("unresolved", true)) .order("desc") .take(limit); return { unresolvedCount: summary?.unresolvedCount ?? 0, eventTypes: summary?.eventTypes ?? [], failures, }; }, }); /** * Idempotent webhook event processor. * * Receives parsed webhook data from the HTTP action handler, * deduplicates by webhook-id, records the event, and dispatches * to event-type-specific handlers from subscriptionHelpers. * * On handler failure, the error is thrown so Convex rolls back the * transaction. The HTTP handler records a sanitized failure projection in a * separate mutation before returning 500, which triggers Dodo's retry. */ export const processWebhookEvent = internalMutation({ args: { webhookId: v.string(), eventType: v.string(), rawPayload: v.any(), timestamp: v.number(), }, handler: async (ctx, args) => { // 1. Idempotency check: skip only if already successfully processed. // Failed events are deleted so the retry can re-process cleanly. const existing = await ctx.db .query("webhookEvents") .withIndex("by_webhookId", (q) => q.eq("webhookId", args.webhookId)) .first(); if (existing) { // Records are only inserted after successful processing (see step 3 below). // If the handler throws, Convex rolls back the transaction and no record // is written. So `existing` always has status "processed" — it's a true // duplicate we can safely skip. console.warn(`[webhook] Duplicate webhook ${args.webhookId}, already processed — skipping`); return; } // 2. Dispatch to event-type-specific handlers. // Errors propagate (throw) so Convex rolls back the entire transaction, // preventing partial writes (e.g., subscription without entitlements). // The HTTP handler catches thrown errors and returns 500 to trigger retries. const data = args.rawPayload.data; // Minimum shape guard — throw so Convex rolls back and returns 500, // causing Dodo to retry instead of silently dropping the event. // The HTTP action records a durable dead-letter projection after this // mutation rejects, outside this transaction, so permanent schema // mismatches remain repairable after Dodo exhausts its retries. if (!data || typeof data !== 'object') { throw new Error( `[webhook] rawPayload.data is missing or not an object (eventType=${args.eventType}, webhookId=${args.webhookId})`, ); } const subscriptionEvents = [ "subscription.active", "subscription.renewed", "subscription.on_hold", "subscription.cancelled", "subscription.plan_changed", "subscription.expired", // PR 3 (post-launch-stabilization): Dodo's docs list `subscription.updated` // as a real-time-sync event for any subscription field change. Handler // dispatches by the payload's `status` field to reuse our existing // lifecycle logic AND respect the paid-through-cancellation policy. "subscription.updated", ] as const; if (subscriptionEvents.includes(args.eventType as typeof subscriptionEvents[number]) && !(data as Record).subscription_id) { throw new Error( `[webhook] Missing subscription_id for subscription event (eventType=${args.eventType}, webhookId=${args.webhookId}, dataKeys=${Object.keys(data as object).join(",")})`, ); } switch (args.eventType) { case "subscription.active": await handleSubscriptionActive(ctx, data, args.timestamp); break; case "subscription.renewed": await handleSubscriptionRenewed(ctx, data, args.timestamp); break; case "subscription.on_hold": await handleSubscriptionOnHold(ctx, data, args.timestamp); break; case "subscription.cancelled": await handleSubscriptionCancelled(ctx, data, args.timestamp); break; case "subscription.plan_changed": await handleSubscriptionPlanChanged(ctx, data, args.timestamp); break; case "subscription.expired": await handleSubscriptionExpired(ctx, data, args.timestamp); break; case "subscription.updated": await handleSubscriptionUpdated(ctx, data, args.timestamp); break; case "payment.succeeded": case "payment.failed": // `payment.processing` is Dodo's only non-terminal payment event; its // payload `data.status` carries the 3DS/SCA-pending `requires_customer_action` // state. `payment.cancelled` is terminal-but-uncharged. Persisting these // gives the app a pending-payment signal for duplicate-prevention (#4438) // and reconciliation (#4439); previously they fell through to `default` // and were dropped while the payment sat in "Requires customer action" on // Dodo's dashboard. (`payment.requires_customer_action` is NOT a Dodo event // type — see derivePaymentEventStatus in subscriptionHelpers.ts.) case "payment.processing": case "payment.cancelled": case "refund.succeeded": case "refund.failed": await handlePaymentOrRefundEvent(ctx, data, args.eventType, args.timestamp); break; case "dispute.opened": case "dispute.won": case "dispute.lost": case "dispute.closed": await handleDisputeEvent(ctx, data, args.eventType, args.timestamp); break; default: // Loud signal for `subscription.*` additions (so a future Dodo event // type doesn't silently no-op). Other unhandled events remain a warn. if (typeof args.eventType === "string" && args.eventType.startsWith("subscription.")) { console.error( `[webhook] Unhandled subscription.* event type: ${args.eventType} — needs a dedicated handler in subscriptionHelpers.ts`, ); } else { console.warn(`[webhook] Unhandled event type: ${args.eventType}`); } } // 3. Record the event AFTER successful processing. // If the handler threw, we never reach here — the transaction rolls back // and Dodo retries. Only successful events are recorded for idempotency. await ctx.db.insert("webhookEvents", { webhookId: args.webhookId, eventType: args.eventType, rawPayload: args.rawPayload, processedAt: Date.now(), status: "processed", }); }, });