feat(convex/customers): add normalizedEmail field + index for broadcast dedup

koala73 · koala73 · commit 09c7664ec76d · 2026-04-26T11:23:09.000+04:00
Required pre-work for the PRO-launch broadcast. Without this, the
audience-build dedup query (registrations − emailSuppressions − paying
customers) is a full `customers` table scan, AND the join would silently
fail across case/whitespace differences between `registrations.normalizedEmail`
(lowercased+trimmed) and `customers.email` (raw) — leaking paid users into
the "buy PRO!" send.

Mirrors the existing convention from `registrations` and `emailSuppressions`,
both of which already store + index `normalizedEmail`.

- schema.ts: customers.normalizedEmail (v.optional, backward-compatible)
  + by_normalized_email index
- subscriptionHelpers.ts: populate normalizedEmail at both write sites in
  handleSubscriptionActive (insert + patch); single derivation reused
- backfillCustomerNormalizedEmail.ts (new): idempotent paginated
  internalMutation + countPending diagnostic query for existing rows

Operational sequence after deploy:
  npx convex run payments/backfillCustomerNormalizedEmail:countPending
  npx convex run payments/backfillCustomerNormalizedEmail:backfill
  # rerun backfill until done:true; verify with countPending → pending:0
diff --git a/convex/payments/backfillCustomerNormalizedEmail.ts b/convex/payments/backfillCustomerNormalizedEmail.ts
@@ -0,0 +1,76 @@
+/**
+ * One-time backfill: populate `customers.normalizedEmail` on rows
+ * that predate the field's introduction.
+ *
+ * Required before the PRO-launch broadcast — the dedup query
+ * (`registrations` − `emailSuppressions` − paying-customers) joins
+ * on `normalizedEmail`, and rows missing the field would otherwise
+ * fall through and receive a "buy PRO!" email despite already paying.
+ *
+ * Idempotent: skips rows that already have a non-empty `normalizedEmail`.
+ * Paginated: pass a `batchSize` (default 500). Re-run until `done: true`.
+ *
+ * Usage:
+ *   npx convex run payments/backfillCustomerNormalizedEmail:backfill
+ *   npx convex run payments/backfillCustomerNormalizedEmail:backfill '{"batchSize":1000}'
+ */
+import { v } from "convex/values";
+import { internalMutation, query } from "../_generated/server";
+
+export const backfill = internalMutation({
+  args: {
+    batchSize: v.optional(v.number()),
+  },
+  handler: async (ctx, { batchSize }) => {
+    const limit = batchSize ?? 500;
+    const all = await ctx.db.query("customers").collect();
+
+    let scanned = 0;
+    let patched = 0;
+    let alreadySet = 0;
+    let emptyEmail = 0;
+
+    for (const row of all) {
+      scanned++;
+
+      if (row.normalizedEmail && row.normalizedEmail.length > 0) {
+        alreadySet++;
+        continue;
+      }
+
+      const computed = (row.email ?? "").trim().toLowerCase();
+      if (computed.length === 0) {
+        emptyEmail++;
+        await ctx.db.patch(row._id, { normalizedEmail: "" });
+        continue;
+      }
+
+      await ctx.db.patch(row._id, { normalizedEmail: computed });
+      patched++;
+      if (patched >= limit) break;
+    }
+
+    const remaining = all.length - scanned;
+    const done = remaining === 0;
+    return { total: all.length, scanned, patched, alreadySet, emptyEmail, remaining, done };
+  },
+});
+
+/**
+ * Diagnostic: how many customer rows still need backfilling?
+ * Returns an exact count (full scan — only run from CLI, not hot paths).
+ */
+export const countPending = query({
+  args: {},
+  handler: async (ctx) => {
+    const all = await ctx.db.query("customers").collect();
+    let pending = 0;
+    let withEmail = 0;
+    let total = all.length;
+    for (const row of all) {
+      if (!row.normalizedEmail || row.normalizedEmail.length === 0) pending++;
+      if (row.email && row.email.length > 0) withEmail++;
+    }
+    return { total, pending, withEmail };
+  },
+});
diff --git a/convex/payments/subscriptionHelpers.ts b/convex/payments/subscriptionHelpers.ts
@@ -333,6 +333,7 @@ export async function handleSubscriptionActive(
   // Upsert customer record so portal session creation can find dodoCustomerId
   const dodoCustomerId = data.customer?.customer_id;
   const email = data.customer?.email ?? "";
+  const normalizedEmail = email.trim().toLowerCase();
 
   if (dodoCustomerId) {
     const existingCustomer = await ctx.db
@@ -346,13 +347,15 @@ export async function handleSubscriptionActive(
       await ctx.db.patch(existingCustomer._id, {
         userId,
         email,
+        normalizedEmail,
         updatedAt: eventTimestamp,
       });
     } else {
       await ctx.db.insert("customers", {
         userId,
         dodoCustomerId,
         email,
+        normalizedEmail,
         createdAt: eventTimestamp,
         updatedAt: eventTimestamp,
       });
diff --git a/convex/schema.ts b/convex/schema.ts
@@ -223,11 +223,19 @@ export default defineSchema({
     userId: v.string(),
     dodoCustomerId: v.optional(v.string()),
     email: v.string(),
+    // Lowercased + trimmed mirror of `email`. Required for O(1) joins from
+    // `registrations`/`emailSuppressions` (both keyed on `normalizedEmail`)
+    // when building broadcast audiences — without this, dedup is a full
+    // table scan and paid users can leak into "buy PRO!" sends.
+    // Optional so existing rows pass schema validation; backfilled via
+    // `npx convex run payments/backfillCustomerNormalizedEmail:backfill`.
+    normalizedEmail: v.optional(v.string()),
     createdAt: v.number(),
     updatedAt: v.number(),
   })
     .index("by_userId", ["userId"])
-    .index("by_dodoCustomerId", ["dodoCustomerId"]),
+    .index("by_dodoCustomerId", ["dodoCustomerId"])
+    .index("by_normalized_email", ["normalizedEmail"]),
 
   webhookEvents: defineTable({
     webhookId: v.string(),
