feat: sync Loops audiences reliably through Stripe

Author: richiemcilroy

.github/workflows/ci.yml

@@ -93,8 +93,8 @@ jobs:
       - name: Check email library
         run: |
           bun run emails:check
-          bun test scripts/emails scripts/loops/profile.test.ts
-          bun run tsc --noEmit --strict --target ESNext --module ESNext --moduleResolution bundler --resolveJsonModule --types bun --skipLibCheck scripts/loops/*.ts scripts/emails/*.ts emails/*.ts emails/marketing/*.ts
+          bun test scripts/emails scripts/loops/profile.test.ts scripts/loops/lifecycle.test.ts
+          bun run tsc --noEmit --strict --jsx react-jsx --allowImportingTsExtensions --target ESNext --module ESNext --moduleResolution bundler --resolveJsonModule --types bun --skipLibCheck scripts/loops/*.ts scripts/emails/*.ts emails/*.ts emails/marketing/*.ts
 
       - name: Test web React compatibility
         run: >-

apps/web/app/api/cron/sync-loops/route.ts

@@ -0,0 +1,60 @@
+import { timingSafeEqual } from "node:crypto";
+import { runLoopsSync } from "@cap/database/loops/worker";
+import {
+	HttpApi,
+	HttpApiBuilder,
+	HttpApiEndpoint,
+	HttpApiError,
+	HttpApiGroup,
+	HttpServerRequest,
+} from "@effect/platform";
+import { Effect, Layer, Schema } from "effect";
+import { apiToHandler } from "@/lib/server";
+import { customerCopy } from "../../../../../../emails/customer-copy";
+
+class Api extends HttpApi.make("LoopsSyncApi").add(
+	HttpApiGroup.make("root").add(
+		HttpApiEndpoint.get("sync")`/api/cron/sync-loops`
+			.addSuccess(
+				Schema.Struct({
+					enabled: Schema.Boolean,
+					processed: Schema.Number,
+					failed: Schema.Number,
+				}),
+			)
+			.addError(HttpApiError.Unauthorized)
+			.addError(HttpApiError.InternalServerError),
+	),
+) {}
+
+const ApiLive = HttpApiBuilder.api(Api).pipe(
+	Layer.provide(
+		HttpApiBuilder.group(Api, "root", (handlers) =>
+			handlers.handle("sync", () =>
+				Effect.gen(function* () {
+					const request = yield* HttpServerRequest.HttpServerRequest;
+					const secret = process.env.CRON_SECRET;
+					if (!secret) return yield* new HttpApiError.InternalServerError();
+					const expected = Buffer.from(`Bearer ${secret}`);
+					const actual = Buffer.from(request.headers.authorization ?? "");
+					if (
+						actual.length !== expected.length ||
+						!timingSafeEqual(actual, expected)
+					)
+						return yield* new HttpApiError.Unauthorized();
+					const result = yield* Effect.tryPromise({
+						try: () => runLoopsSync(customerCopy),
+						catch: () => new HttpApiError.InternalServerError(),
+					});
+					if (result.failed)
+						return yield* new HttpApiError.InternalServerError();
+					return result;
+				}),
+			),
+		),
+	),
+);
+
+export const GET = apiToHandler(ApiLive);
+export const maxDuration = 120;
+export const dynamic = "force-dynamic";

apps/web/app/api/invite/accept/route.ts

@@ -1,6 +1,7 @@
 import { db } from "@cap/database";
 import { getCurrentUser } from "@cap/database/auth/session";
 import { nanoId } from "@cap/database/helpers";
+import { enqueueLoopsSync } from "@cap/database/loops/queue";
 import {
 	organizationInvites,
 	organizationMembers,
@@ -149,6 +150,7 @@ export async function POST(request: NextRequest) {
 
 			await tx.update(users).set(userUpdate).where(eq(users.id, user.id));
 
+			await enqueueLoopsSync(tx, user.id, true);
 			await tx
 				.delete(organizationInvites)
 				.where(eq(organizationInvites.id, inviteId));

apps/web/app/api/webhooks/stripe/route.ts

@@ -2,6 +2,7 @@ import { db } from "@cap/database";
 import { sendEmail } from "@cap/database/emails/config";
 import { PaymentFailed } from "@cap/database/emails/payment-failed";
 import { nanoId } from "@cap/database/helpers";
+import { enqueueLoopsSync } from "@cap/database/loops/queue";
 import {
 	developerCreditTransactions,
 	signedBaas,
@@ -608,6 +609,7 @@ export const POST = async (req: Request) => {
 						onboarding_completed_at: isOnBoarding ? new Date() : undefined,
 					})
 					.where(eq(users.id, dbUser.id));
+				await enqueueLoopsSync(db(), dbUser.id);
 
 				console.log("Successfully updated user in database");
 
@@ -767,6 +769,7 @@ export const POST = async (req: Request) => {
 						inviteQuota: inviteQuota,
 					})
 					.where(eq(users.id, dbUser.id));
+				await enqueueLoopsSync(db(), dbUser.id);
 
 				console.log(
 					"Successfully updated user in database with new invite quota:",
@@ -942,6 +945,7 @@ export const POST = async (req: Request) => {
 						inviteQuota: 1,
 					})
 					.where(eq(users.id, foundUserId));
+				await enqueueLoopsSync(db(), foundUserId);
 
 				console.log("User updated successfully", {
 					foundUserId,

apps/web/lib/organization-provisioning.ts

@@ -2,6 +2,7 @@ import "server-only";
 
 import { db } from "@cap/database";
 import { nanoId } from "@cap/database/helpers";
+import { enqueueLoopsSync } from "@cap/database/loops/queue";
 import {
 	organizationInvites,
 	organizationMembers,
@@ -49,7 +50,9 @@ export async function provisionOrganizationInvitee({
 		const userId = existingUser?.id ?? User.UserId.make(nanoId());
 
 		if (existingUser) {
-			const userUpdate: Partial<typeof users.$inferInsert> = {};
+			const userUpdate: Partial<typeof users.$inferInsert> = {
+				marketingOrigin: "teammate",
+			};
 
 			if (!existingUser.name) {
 				userUpdate.name = getProvisionedUserName(normalizedEmail);
@@ -68,12 +71,14 @@ export async function provisionOrganizationInvitee({
 			await tx.insert(users).values({
 				id: userId,
 				email: normalizedEmail,
+				marketingOrigin: "teammate",
 				name: getProvisionedUserName(normalizedEmail),
 				activeOrganizationId: organizationId,
 				defaultOrgId: organizationId,
 			});
 		}
 
+		await enqueueLoopsSync(tx, userId);
 		const [existingMember] = await tx
 			.select({
 				id: organizationMembers.id,

apps/web/vercel.json

@@ -1,6 +1,7 @@
 {
 	"$schema": "https://openapi.vercel.sh/vercel.json",
 	"crons": [
+		{ "path": "/api/cron/sync-loops", "schedule": "* * * * *" },
 		{
 			"path": "/api/cron/cleanup-agent-api",
 			"schedule": "17 3 * * *"

emails/CATALOG.md

@@ -13,7 +13,9 @@ These are the locally configured draft journeys. This document is not a live sta
 | [Teammate onboarding](https://app.loops.so/workflows/cmtvpm0ag01it0j37w02o6wav) | teammate | Day 0, Day 3 | 2 |
 | [Former customer follow-up](https://app.loops.so/workflows/cmtvpm8zm01ii0j01cz7d15qr) | former | Day 14 | 1 |
 
-All journeys require global subscription, positive Cap consent, the exact audience, lifecycle enabled and onboarding eligible. These filters continue to apply downstream. Free/former promotional flows additionally exclude teammates and require promotional eligibility. Customer and teammate flows still require marketing consent.
+Completed signups reach Loops through Stripe; SSO uses a small direct fallback. Cap supplies targeting through a durable sync queue, without a separate marketing opt-in step. Existing opt-outs and suppressions take precedence. The integration is not deployed. Historical imports stay held; a new accepted invitation can start teammate help only.
+
+Current draft journeys require global subscription, capConsent=subscribed, the exact audience, lifecycle enabled and onboarding eligible. capConsent is a legacy migration guard, not a separate consent-capture requirement for new signups. These filters continue to apply downstream. Free/former promotional flows additionally exclude teammates and require promotional eligibility.
 
 Teammate history takes priority over paid/free classification. Ambiguous contacts receive no journey. [Audience classification and consent](../scripts/loops/README.md#audience-rules).
 
@@ -112,7 +114,7 @@ flowchart TD
 
 ## Campaign templates
 
-Campaigns are manually scheduled product updates, with no automatic enrollment. Both require subscription, positive consent and their exact audience. The free template also requires promotional eligibility and excludes teammates.
+Campaigns are manually scheduled product updates, with no automatic enrollment. Both require subscription, capConsent=subscribed and their exact audience. The free template also requires promotional eligibility and excludes teammates.
 
 | Campaign | Audience | Subject and source | Loops ID |
 | --- | --- | --- | --- |

emails/QA.md

@@ -43,7 +43,15 @@ The local suite checks consent precedence, suppression preservation, customer an
 
 ## Remaining gate before switching
 
-The deployed Cap signup/purchase/invite/opt-out path through an enrollment dispatcher and into an inbox has not been proven. The reviewed schema is not deployed, new explicit consent capture/enrollment and retry/deduplication dispatch need implementation, and the current sync intentionally keeps contacts held. Freshness expiry and scheduler monitoring are also required: an old `capVerifiedAt` value does not automatically prevent a Loops send.
+The new implementation uses native Stripe imports, a held SSO fallback, and `loops_sync_jobs` for durable enrichment. Local routing tests cover missing Stripe imports, incomplete billing bootstrap, teammate persistence, an existing account accepting a new invite, global/list opt-outs, missing list membership, replay, identity conflicts and imported holds. The full email-script, profile and lifecycle suites passed 54 tests; existing SSO and Stripe subscription suites passed another 83. Database and web TypeScript checks passed.
+
+Five integration checks passed on an isolated PlanetScale branch: transactional rollback, overlapping claims, expired lease recovery, newer work surviving an older completion, and an indexed due query. A full `db:push` attempt hit the repository's unrelated storage prefix-index quoting bug; the exact generated Loops migration was then applied and independently checked on the empty test branch. No production schema or data changed. The owned schema-only test branch was deleted after verification.
+
+The worker was exercised from synthetic database records through the real Loops API, including a simulated HTTP 429 followed by a successful retry, free and SSO teammate classification, a free-to-teammate transition, and a later unsubscribe surviving a billing change. Separate live checks cover imported holds and list removals. Loops omits an API-removed list from the returned map; the runtime now treats missing membership as ineligible and never re-adds lists during updates. All owned test contacts were read back held and globally unsubscribed. These checks sent no email and kept production workflows in Draft.
+
+The native Stripe integration is connected to Cap Software, Inc. and enabled for customer creation and updates only. A new owned live Stripe customer appeared in Loops; two later name changes updated that same contact. Name import and Product updates and tips list assignment were verified, and a global unsubscribe survived the final Stripe update. No payments, subscriptions or emails were created. The test contact remains held and globally unsubscribed. Both saved event mappings were checked after reload. A separate test-mode Stripe customer did not import through this live connection, is not counted as a passing native test, and was deleted after verification.
+
+The deployed Cap signup/purchase/invite path has not yet been proven end to end. Production schema/code deployment, recipient Preference Center list-opt-out preservation across native Stripe updates, the final Bento suppression delta, and scheduler monitoring remain cutover gates. The Cap-worker API list-removal check above does not independently prove native Stripe handling of a recipient list opt-out. The worker retries failed updates, but Loops still uses the last successfully synced values during an outage; `capVerifiedAt` is not a native expiry rule.
 
 Follow the [cutover runbook](../scripts/loops/README.md#before-any-activation). Test that deployed path using owned inboxes, including outages and retries, before requesting production activation. Keep imported history held, preserve all opt-outs, reconcile fresh Bento changes, check Resend overlap and disable duplicate Bento automation only during the approved cutover.
 

packages/database/auth/drizzle-adapter.ts

@@ -1,11 +1,11 @@
 import { createHash } from "node:crypto";
-import { isProSubscription, STRIPE_AVAILABLE, stripe } from "@cap/utils";
+import { STRIPE_AVAILABLE } from "@cap/utils";
 import { type ImageUpload, Organisation, User } from "@cap/web-domain";
 import { and, eq } from "drizzle-orm";
 import type { MySql2Database } from "drizzle-orm/mysql2";
 import type { Adapter } from "next-auth/adapters";
-import type Stripe from "stripe";
 import { nanoId } from "../helpers.ts";
+import { enqueueLoopsSync } from "../loops/queue.ts";
 import {
 	accounts,
 	organizationInvites,
@@ -16,6 +16,7 @@ import {
 	verificationTokens,
 } from "../schema.ts";
 import type { ValidatedSsoIdentity } from "./sso.ts";
+import { provisionStripeCustomer } from "./stripe-customer.ts";
 
 type CreateUserData = Parameters<NonNullable<Adapter["createUser"]>>[0];
 type LinkAccountData = Parameters<NonNullable<Adapter["linkAccount"]>>[0];
@@ -118,6 +119,7 @@ export function DrizzleAdapter(
 						await tx.update(users).set(userUpdate).where(eq(users.id, userId));
 					}
 
+					await enqueueLoopsSync(tx, userId);
 					return;
 				}
 
@@ -157,6 +159,7 @@ export function DrizzleAdapter(
 					await insertUser;
 				}
 
+				await enqueueLoopsSync(tx, userId);
 				if (pendingInvite || ssoIdentity) {
 					return;
 				}
@@ -194,69 +197,7 @@ export function DrizzleAdapter(
 			if (!row) throw new Error("User not found");
 
 			if (STRIPE_AVAILABLE() && !ssoIdentity) {
-				const existingCustomers = await stripe().customers.list({
-					email: normalizedEmail,
-					limit: 1,
-				});
-
-				let customer: Stripe.Customer;
-				if (existingCustomers.data.length > 0 && existingCustomers.data[0]) {
-					customer = existingCustomers.data[0];
-
-					customer = await stripe().customers.update(customer.id, {
-						metadata: {
-							...customer.metadata,
-							userId: row.id,
-						},
-					});
-				} else {
-					customer = await stripe().customers.create({
-						email: normalizedEmail,
-						metadata: {
-							userId: row.id,
-						},
-					});
-				}
-
-				const subscriptions = await stripe().subscriptions.list({
-					customer: customer.id,
-					status: "active",
-					limit: 100,
-				});
-
-				const proSubscriptions = subscriptions.data.filter(isProSubscription);
-				const inviteQuota = proSubscriptions.reduce((total, sub) => {
-					return (
-						total +
-						sub.items.data.reduce(
-							(subTotal, item) => subTotal + (item.quantity || 1),
-							0,
-						)
-					);
-				}, 0);
-
-				const mostRecentSubscription = proSubscriptions[0];
-
-				await db
-					.update(users)
-					.set({
-						stripeCustomerId: customer.id,
-						...(mostRecentSubscription && {
-							stripeSubscriptionId: mostRecentSubscription.id,
-							stripeSubscriptionStatus: mostRecentSubscription.status,
-							inviteQuota: inviteQuota || 1,
-						}),
-					})
-					.where(eq(users.id, row.id));
-
-				const [updatedRow] = await db
-					.select()
-					.from(users)
-					.where(eq(users.id, row.id))
-					.limit(1);
-				if (updatedRow) {
-					row = updatedRow;
-				}
+				row = await provisionStripeCustomer(db, row);
 			}
 
 			return row;
@@ -309,13 +250,16 @@ export function DrizzleAdapter(
 		},
 		async updateUser({ id, image, ...userData }) {
 			if (!id) throw new Error("User not found");
-			await db
-				.update(users)
-				.set({
-					...userData,
-					image: image as ImageUpload.ImageUrlOrKey | null,
-				})
-				.where(eq(users.id, User.UserId.make(id)));
+			await db.transaction(async (tx) => {
+				await tx
+					.update(users)
+					.set({
+						...userData,
+						image: image as ImageUpload.ImageUrlOrKey | null,
+					})
+					.where(eq(users.id, User.UserId.make(id)));
+				await enqueueLoopsSync(tx, User.UserId.make(id));
+			});
 			const rows = await db
 				.select()
 				.from(users)

packages/database/auth/sso.ts

@@ -6,6 +6,7 @@ import { WorkOS } from "@workos-inc/node";
 import { and, eq, isNull } from "drizzle-orm";
 import { nanoId } from "../helpers.ts";
 import { db } from "../index.ts";
+import { enqueueLoopsSync } from "../loops/queue.ts";
 import {
 	accounts,
 	organizationInvites,
@@ -273,5 +274,10 @@ export async function provisionSsoMembership(
 				},
 			})
 			.where(eq(users.id, userId));
+		await enqueueLoopsSync(
+			tx,
+			userId,
+			!member || !user.onboarding_completed_at,
+		);
 	});
 }