added lazy migrations

package.json

@@ -131,7 +131,7 @@
 		"@aws-sdk/client-sqs": "^3.985.0",
 		"@better-auth/core": "catalog:",
 		"@better-auth/oauth-provider": "catalog:",
-		"@trigger.dev/sdk": "4.4.5",
+		"@trigger.dev/sdk": "4.4.6",
 		"@wooorm/starry-night": "^3.8.0",
 		"ag-charts-react": "^12.3.0",
 		"better-auth": "catalog:",
@@ -143,13 +143,14 @@
 	"devDependencies": {
 		"@better-auth/cli": "^1.4.21",
 		"@biomejs/biome": "^2.2.7",
-		"@trigger.dev/build": "4.4.5",
+		"@trigger.dev/build": "4.4.6",
 		"@types/node": "^24.9.1",
 		"concurrently": "^9.2.1",
 		"dotenv": "^16.6.1",
 		"husky": "^9.1.7",
 		"inquirer": "^12.10.0",
 		"knip": "^6.7.0",
+		"trigger.dev": "4.4.6",
 		"ts-to-zod": "^5.1.0",
 		"turbo": "^2.9.6"
 	}

scripts/dev.ts

@@ -49,6 +49,30 @@ function getEnvVariable(filePath: string, key: string): string | null {
 	return null;
 }
 
+function getPackageDependencyVersion({
+	projectRoot,
+	packageName,
+}: {
+	projectRoot: string;
+	packageName: string;
+}): string {
+	const packageJson = JSON.parse(
+		readFileSync(join(projectRoot, "package.json"), "utf-8"),
+	) as {
+		dependencies?: Record<string, string>;
+		devDependencies?: Record<string, string>;
+	};
+	const version =
+		packageJson.dependencies?.[packageName] ??
+		packageJson.devDependencies?.[packageName];
+
+	if (!version) {
+		throw new Error(`Missing ${packageName} in package.json`);
+	}
+
+	return version;
+}
+
 function killPorts({ ports }: { ports: number[] }) {
 	if (process.platform === "win32") {
 		return;
@@ -136,6 +160,11 @@ async function startDev() {
 
 		// Use cmd on Windows, sh on Unix
 		const isWindows = process.platform === "win32";
+		const triggerDevVersion = getPackageDependencyVersion({
+			projectRoot,
+			packageName: "trigger.dev",
+		});
+		const bunInstallBin = dirname(process.execPath);
 
 		let shellArgs: string[];
 		if (serverOnly) {
@@ -179,8 +208,8 @@ async function startDev() {
 				colors.push("cyan");
 				cmds.push(
 					isWindows
-						? `"bunx trigger.dev@latest dev"`
-						: `"bunx trigger.dev@latest dev"`,
+						? `"bunx trigger.dev@${triggerDevVersion} dev"`
+						: `"bunx trigger.dev@${triggerDevVersion} dev"`,
 				);
 			}
 
@@ -204,12 +233,13 @@ async function startDev() {
 
 		const concurrentlyProc = Bun.spawn(shellArgs, {
 			cwd: projectRoot,
-		env: {
-			...process.env,
-			VITE_PORT: VITE_PORT.toString(),
-			SERVER_PORT: SERVER_PORT.toString(),
-			CHECKOUT_PORT: CHECKOUT_PORT.toString(),
-			VITE_APP_ENV: viteAppEnv,
+			env: {
+				...process.env,
+				VITE_PORT: VITE_PORT.toString(),
+				SERVER_PORT: SERVER_PORT.toString(),
+				CHECKOUT_PORT: CHECKOUT_PORT.toString(),
+				VITE_APP_ENV: viteAppEnv,
+				BUN_INSTALL_BIN: process.env.BUN_INSTALL_BIN ?? bunInstallBin,
 				...(worktreeNum > 1 && {
 					CLIENT_URL: `http://localhost:${VITE_PORT}`,
 					BETTER_AUTH_URL: `http://localhost:${SERVER_PORT}`,

server/src/_luaScriptsV2/deleteFullCustomerCache/deleteFullCustomerCache.lua

@@ -6,11 +6,11 @@
   2. Optionally sets the stale-write guard key (to prevent in-flight requests from writing stale data)
   3. Deletes the cache key and path index key
 
-  All keys are constructed internally from orgId/env/customerId using the
-  prepended key builder functions.
-
   KEYS:
-    [1] cacheKey - used for cluster slot routing only
+    [1] cacheKey
+    [2] testGuardKey
+    [3] guardKey
+    [4] pathIdxKey
 
   ARGV:
     [1] orgId
@@ -33,10 +33,10 @@ local guardTimestamp = ARGV[4]
 local guardTtl = tonumber(ARGV[5])
 local skipGuard = ARGV[6] == "true"
 
-local testGuardKey = build_test_guard_key(org_id, env, customer_id)
-local guardKey = build_guard_key(org_id, env, customer_id)
-local cacheKey = build_full_customer_cache_key(org_id, env, customer_id)
-local pathIdxKey = build_path_index_key(org_id, env, customer_id)
+local cacheKey = KEYS[1]
+local testGuardKey = KEYS[2]
+local guardKey = KEYS[3]
+local pathIdxKey = KEYS[4]
 
 -- Check test guard first (used in race condition tests)
 if redis.call("EXISTS", testGuardKey) == 1 then

server/src/_luaScriptsV2/deleteFullCustomerCache/setFullCustomerCache.lua

@@ -8,11 +8,10 @@
   4. Sets TTL on the cache key
   5. Replaces the path index Hash (DEL + HSET + EXPIRE)
 
-  All keys are constructed internally from orgId/env/customerId using the
-  prepended key builder functions.
-
   KEYS:
-    [1] cacheKey - used for cluster slot routing only
+    [1] guardKey
+    [2] cacheKey
+    [3] pathIdxKey
 
   ARGV:
     [1] orgId
@@ -39,9 +38,9 @@ local serializedData = ARGV[6]
 local overwrite = ARGV[7] == "true"
 local pathIndexJson = ARGV[8]
 
-local guardKey = build_guard_key(org_id, env, customer_id)
-local cacheKey = build_full_customer_cache_key(org_id, env, customer_id)
-local pathIdxKey = build_path_index_key(org_id, env, customer_id)
+local guardKey = KEYS[1]
+local cacheKey = KEYS[2]
+local pathIdxKey = KEYS[3]
 
 -- Check if guard exists (deletion happened recently)
 -- Skip check if either value is nil/null/falsey

server/src/db/initDrizzle.ts

@@ -98,7 +98,7 @@ export const { db: dbCritical, client: clientCritical } = initDrizzle({
 
 // -- General pool: used by all other endpoints --
 export const { db: dbGeneral, client: clientGeneral } = initDrizzle({
-	// connectTimeout: 5,
+	connectTimeout: isProd ? 5 : 30,
 });
 
 // -- Replica pool: used as fallback when primary is degraded --

server/src/external/autumn/autumnCli.ts

@@ -996,6 +996,15 @@ export class AutumnInt {
 				run_id: string;
 			};
 		},
+		lazyRun: async (params: {
+			id: string;
+		}): Promise<{
+			migration_id: string;
+			run_id: string;
+		}> => {
+			const data = await this.post(`/migrations.lazy_run`, params);
+			return data as { migration_id: string; run_id: string };
+		},
 		listRuns: async (params: {
 			migrationId: string;
 		}): Promise<{ list: MigrationRun[] }> => {

server/src/external/redis/initRedisV2.ts

@@ -14,7 +14,7 @@ import {
 } from "./initUtils/redisV2Config.js";
 
 const redisV2Config = getRedisV2ConnectionConfig({
-	cacheV2Url: process.env.CACHE_V2_UPSTASH_URL,
+	cacheV2Url: process.env.CACHE_V2_DRAGONFLY_URL,
 	primaryCacheUrl: process.env.CACHE_URL,
 	currentRegion,
 });

server/src/external/redis/initUtils/createRedisClient.ts

@@ -6,6 +6,15 @@ import { registerRedisCommands } from "./registerRedisCommands.js";
 const REDIS_COMMAND_TIMEOUT_MS =
 	process.env.NODE_ENV === "production" ? 10_000 : 60_000;
 
+const formatRedisEndpoint = ({ cacheUrl }: { cacheUrl: string }) => {
+	try {
+		const url = new URL(cacheUrl);
+		return `${url.protocol}//${url.host}`;
+	} catch {
+		return "<invalid redis url>";
+	}
+};
+
 /** Create a Redis connection for a specific region.
  *  `supportsUpstashShebang` defaults to true; set false for non-Upstash
  *  providers (ElastiCache, Dragonfly, self-hosted) that reject the
@@ -21,6 +30,10 @@ export const createRedisClient = ({
 	supportsUpstashShebang?: boolean;
 	commandTimeout?: number;
 }): Redis => {
+	console.log(
+		`[Redis] ${region}: connecting to ${formatRedisEndpoint({ cacheUrl })}`,
+	);
+
 	const instance = new Redis(cacheUrl, {
 		tls:
 			process.env.CACHE_CERT && !cacheBackupUrl

server/src/external/redis/initUtils/redisTypes.ts

@@ -94,6 +94,9 @@ declare module "ioredis" {
 		): Promise<string>;
 		deleteFullCustomerCache(
 			cacheKey: string,
+			testGuardKey: string,
+			guardKey: string,
+			pathIndexKey: string,
 			orgId: string,
 			env: string,
 			customerId: string,
@@ -102,7 +105,9 @@ declare module "ioredis" {
 			skipGuard: string,
 		): Promise<"SKIPPED" | "DELETED" | "NOT_FOUND">;
 		setFullCustomerCache(
+			guardKey: string,
 			cacheKey: string,
+			pathIndexKey: string,
 			orgId: string,
 			env: string,
 			customerId: string,

server/src/external/redis/initUtils/redisV2Config.ts

@@ -1,6 +1,7 @@
 import type { RedisV2InstanceName } from "@/internal/misc/redisV2Cache/redisV2CacheSchemas.js";
 
-export const REDIS_V2_COMMAND_TIMEOUT_MS = 1_000;
+export const REDIS_V2_COMMAND_TIMEOUT_MS =
+	process.env.NODE_ENV === "production" ? 1_000 : 10_000;
 
 export const getRedisV2ConnectionConfig = ({
 	cacheV2Url,

server/src/external/redis/initUtils/registerRedisCommands.ts

@@ -128,12 +128,12 @@ export const registerRedisCommands = ({
 	});
 
 	redisInstance.defineCommand("deleteFullCustomerCache", {
-		numberOfKeys: 1,
+		numberOfKeys: 4,
 		lua: DELETE_FULL_CUSTOMER_CACHE_SCRIPT,
 	});
 
 	redisInstance.defineCommand("setFullCustomerCache", {
-		numberOfKeys: 1,
+		numberOfKeys: 3,
 		lua: SET_FULL_CUSTOMER_CACHE_SCRIPT,
 	});
 

server/src/honoUtils/HonoEnv.ts

@@ -61,6 +61,13 @@ export type RequestContext = {
 	expand: string[];
 	skipCache: boolean;
 
+	/** True when the context is built by `createTriggerContext` — i.e. we're
+	 *  executing inside a Trigger.dev task. Read by `checkPendingMigrationsForCustomer`
+	 *  to short-circuit: a migration task loads `CusService.getFull` /
+	 *  `getFullSubject` for its target customer, and that load must NOT
+	 *  re-enqueue another migration task. */
+	insideTriggerTask?: boolean;
+
 	extraLogs: Record<string, unknown>;
 
 	fullCustomer?: FullCustomer;

server/src/init.ts

@@ -53,8 +53,6 @@ let shuttingDown = false;
 const init = async ({ startupStartedAt }: { startupStartedAt: number }) => {
 	logger.info(getRedactedDatabaseUrls(), "DB URLs");
 
-	console.log("DB URLs:", getRedactedDatabaseUrls());
-
 	const app = createHonoApp();
 
 	initPgHealthMonitor({ client: clientCritical });

server/src/internal/customers/CusService.ts

@@ -33,6 +33,7 @@ import type { DrizzleCli } from "@/db/initDrizzle.js";
 import { executeWithHealthTracking } from "@/db/pgHealthMonitor.js";
 import type { RepoContext } from "@/db/repoContext.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { checkPendingMigrationsForCustomer } from "@/internal/migrations/v2/lazy/checkPendingMigrationsForCustomer.js";
 import { withSpan } from "../analytics/tracer/spanUtils.js";
 import {
 	getOrgCusProductLimit,
@@ -179,6 +180,10 @@ export class CusService {
 						fullCus,
 						ctx,
 					});
+					await checkPendingMigrationsForCustomer({
+						ctx,
+						fullCustomer: fullCus,
+					});
 				}
 
 				return fullCus;

server/src/internal/customers/cache/fullSubject/actions/getCachedFullSubject.ts

@@ -1,8 +1,13 @@
-import { type FullSubject, normalizedToFullSubject } from "@autumn/shared";
+import {
+	type FullSubject,
+	fullSubjectToFullCustomer,
+	normalizedToFullSubject,
+} from "@autumn/shared";
 import { isRedisMigrationCacheStale } from "@/external/redis/customerRedisRouting.js";
 import { runRedisOp } from "@/external/redis/utils/runRedisOp.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { lazyResetSubjectEntitlements } from "@/internal/customers/actions/resetCustomerEntitlementsV2/lazyResetSubjectEntitlements.js";
+import { checkPendingMigrationsForCustomer } from "@/internal/migrations/v2/lazy/checkPendingMigrationsForCustomer.js";
 import { getFullSubjectRolloutSnapshot } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
 import { isSnapshotCacheStale } from "@/internal/misc/rollouts/rolloutUtils.js";
 import { applyLiveAggregatedBalances } from "../balances/applyLiveAggregatedBalances.js";
@@ -234,6 +239,10 @@ export const getCachedFullSubject = async ({
 
 		const fullSubject = normalizedToFullSubject({ normalized });
 		await lazyResetSubjectEntitlements({ ctx, fullSubject });
+		await checkPendingMigrationsForCustomer({
+			ctx,
+			fullCustomer: fullSubjectToFullCustomer({ fullSubject }),
+		});
 		return { fullSubject, subjectViewEpoch: currentSubjectViewEpoch };
 	} catch (error) {
 		logger.warn(

server/src/internal/customers/cache/fullSubject/fullSubjectCacheModel.ts

@@ -1,12 +1,13 @@
 import {
 	CusProductSchema,
-	CustomerSchema,
 	CustomerPriceSchema,
+	CustomerSchema,
 	EntitlementWithFeatureSchema,
 	EntityAggregationsSchema,
 	EntitySchema,
 	FreeTrialSchema,
 	InvoiceSchema,
+	MigrationItemRunSchema,
 	type NormalizedFullSubject,
 	PriceSchema,
 	ProductSchema,
@@ -62,6 +63,11 @@ export const CachedFullSubjectSchema = z.object({
 
 	entity_aggregations: EntityAggregationsSchema.optional(),
 
+	// `.default([])` makes pre-existing cache entries (written before this
+	// field existed) hole-fill to an empty array via `normalizeFromSchema`.
+	// The empty-array vs empty-object Lua quirk is also handled there.
+	migration_item_runs: z.array(MigrationItemRunSchema).default([]),
+
 	_schemaVersion: z.number().optional(),
 	_cachedAt: z.number(),
 	meteredFeatures: z.array(z.string()),
@@ -117,6 +123,7 @@ export const normalizedToCachedFullSubject = ({
 		subscriptions: normalized.subscriptions,
 		invoices: normalized.invoices,
 		entity_aggregations: normalized.entity_aggregations,
+		migration_item_runs: normalized.migration_item_runs ?? [],
 		_schemaVersion: FULL_SUBJECT_CACHE_SCHEMA_VERSION,
 		_cachedAt: Date.now(),
 		meteredFeatures,
@@ -151,5 +158,6 @@ export const cachedFullSubjectToNormalized = ({
 		subscriptions: cached.subscriptions,
 		invoices: cached.invoices,
 		entity_aggregations: cached.entity_aggregations,
+		migration_item_runs: cached.migration_item_runs ?? [],
 	};
 };

server/src/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.ts

@@ -5,10 +5,13 @@ import {
 } from "@/external/redis/initRedis.js";
 import { invalidateCachedFullSubject } from "@/internal/customers/cache/fullSubject/index.js";
 import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
+import { buildPathIndexKey } from "../../cache/pathIndex/pathIndexConfig.js";
 import {
+	buildFullCustomerCacheGuardKey,
 	buildFullCustomerCacheKey,
 	FULL_CUSTOMER_CACHE_GUARD_TTL_SECONDS,
 } from "./fullCustomerCacheConfig.js";
+import { buildTestFullCustomerCacheGuardKey } from "./testFullCustomerCacheGuard.js";
 
 /**
  * Delete FullCustomer from Redis cache across ALL regions.
@@ -36,6 +39,21 @@ export const deleteCachedFullCustomer = async ({
 		env,
 		customerId,
 	});
+	const testGuardKey = buildTestFullCustomerCacheGuardKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
+	const guardKey = buildFullCustomerCacheGuardKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
+	const pathIndexKey = buildPathIndexKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
 	const regions = getConfiguredRegions();
 	const guardTimestamp = Date.now().toString();
 	const customerLabel = entityId ? `${customerId}:${entityId}` : customerId;
@@ -69,6 +87,9 @@ export const deleteCachedFullCustomer = async ({
 
 			const result = await regionalRedis.deleteFullCustomerCache(
 				cacheKey,
+				testGuardKey,
+				guardKey,
+				pathIndexKey,
 				org.id,
 				env,
 				customerId,

server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.ts

@@ -11,6 +11,7 @@ import { getDbHealth, PgHealth } from "@/db/pgHealthMonitor.js";
 import { isRedisMigrationCacheStale } from "@/external/redis/customerRedisRouting.js";
 import { redis } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { checkPendingMigrationsForCustomer } from "@/internal/migrations/v2/lazy/checkPendingMigrationsForCustomer.js";
 import { isSnapshotCacheStale } from "@/internal/misc/rollouts/rolloutUtils.js";
 import { tryRedisRead } from "@/utils/cacheUtils/cacheUtils.js";
 import { normalizeFromSchema } from "@/utils/cacheUtils/normalizeFromSchema.js";
@@ -235,6 +236,7 @@ export const getCachedFullCustomer = async ({
 	// path to fail because the primary is down.
 	if (getDbHealth() !== PgHealth.Degraded) {
 		await resetCustomerEntitlements({ ctx, fullCus: fullCustomer });
+		await checkPendingMigrationsForCustomer({ ctx, fullCustomer });
 	}
 
 	// Round balance fields to handle floating-point precision from JSON.NUMINCRBY

server/src/internal/customers/cusUtils/fullCustomerCacheUtils/setCachedFullCustomer.ts

@@ -2,9 +2,11 @@ import { type FullCustomer, isBooleanCusEnt } from "@autumn/shared";
 import { redis } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { buildPathIndex } from "@/internal/customers/cache/pathIndex/buildPathIndex.js";
+import { buildPathIndexKey } from "@/internal/customers/cache/pathIndex/pathIndexConfig.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { addToExtraLogs } from "@/utils/logging/addToExtraLogs.js";
 import {
+	buildFullCustomerCacheGuardKey,
 	buildFullCustomerCacheKey,
 	FULL_CUSTOMER_CACHE_TTL_SECONDS,
 } from "./fullCustomerCacheConfig.js";
@@ -97,6 +99,16 @@ export const setCachedFullCustomer = async ({
 		env,
 		customerId,
 	});
+	const guardKey = buildFullCustomerCacheGuardKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
+	const pathIndexKey = buildPathIndexKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
 	const pathIndexEntries = buildPathIndex({
 		fullCustomer: fullCustomerForCache,
 	});
@@ -106,7 +118,9 @@ export const setCachedFullCustomer = async ({
 
 	const result = await tryRedisWrite(async () => {
 		return await redis.setFullCustomerCache(
+			guardKey,
 			cacheKey,
+			pathIndexKey,
 			org.id,
 			env,
 			customerId,

server/src/internal/customers/getFullCusQuery.ts

@@ -366,6 +366,31 @@ export const getFullCusQuery = ({
 		sqlChunks.push(buildInvoicesCTE(!!entityId));
 	}
 
+	// Unconditional CTE for the customer's `migration_item_runs` scoped to
+	// the org's active lazy migrations. Empty in steady state — joins through
+	// `migration_runs` so callers don't have to thread the active list.
+	sqlChunks.push(sql`, `);
+	sqlChunks.push(sql`
+    customer_migration_item_runs AS (
+      SELECT mir.*
+      FROM migration_item_runs mir
+      WHERE mir.item_kind = 'customer'
+        AND mir.dry_run = false
+        AND mir.item_id = (SELECT internal_id FROM customer_record)
+        AND mir.migration_internal_id IN (
+          SELECT mr.migration_internal_id
+          FROM migration_runs mr
+          WHERE mr.org_id = ${orgId}
+            AND mr.env = ${env}
+            AND mr.status IN ('queued', 'running')
+            AND mr.dry_run = false
+            AND mr.lazy_run = true
+        )
+      ORDER BY mir.updated_at DESC NULLS LAST, mir.created_at DESC
+      LIMIT 10
+    )
+  `);
+
 	// Conditionally add events CTE
 	if (withEvents) {
 		sqlChunks.push(sql`, `);
@@ -456,6 +481,13 @@ export const getFullCusQuery = ({
       (SELECT events FROM customer_events) AS events`);
 	}
 
+	selectFieldsChunks.push(sql`,
+    COALESCE(
+      (SELECT json_agg(row_to_json(mir) ORDER BY mir.updated_at DESC NULLS LAST, mir.created_at DESC)
+       FROM customer_migration_item_runs mir),
+      '[]'::json
+    ) AS migration_item_runs`);
+
 	sqlChunks.push(sql`
     SELECT ${sql.join(selectFieldsChunks, sql``)}
     FROM customer_record cr

server/src/internal/customers/repos/getFullSubject/getFullSubject.ts

@@ -1,11 +1,13 @@
 import {
 	type CusProductStatus,
 	type FullSubject,
+	fullSubjectToFullCustomer,
 	type NormalizedFullSubject,
 	normalizedToFullSubject,
 	type SubjectQueryRow,
 } from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { checkPendingMigrationsForCustomer } from "@/internal/migrations/v2/lazy/checkPendingMigrationsForCustomer.js";
 import { lazyResetSubjectEntitlements } from "../../actions/resetCustomerEntitlementsV2/lazyResetSubjectEntitlements.js";
 import { RELEVANT_STATUSES } from "../../cusProducts/CusProductService.js";
 import { getFullSubjectQuery } from "./getFullSubjectQuery.js";
@@ -49,6 +51,10 @@ export async function getFullSubject({
 		allowMissingEntity,
 	});
 	await lazyResetSubjectEntitlements({ ctx, fullSubject });
+	await checkPendingMigrationsForCustomer({
+		ctx,
+		fullCustomer: fullSubjectToFullCustomer({ fullSubject }),
+	});
 	return fullSubject;
 }
 
@@ -92,6 +98,10 @@ export async function getFullSubjectNormalized({
 
 	const fullSubject = normalizedToFullSubject({ normalized });
 	await lazyResetSubjectEntitlements({ ctx, fullSubject, normalized });
+	await checkPendingMigrationsForCustomer({
+		ctx,
+		fullCustomer: fullSubjectToFullCustomer({ fullSubject }),
+	});
 
 	return { normalized, fullSubject };
 }

server/src/internal/customers/repos/getFullSubject/getFullSubjectQuery.ts

@@ -184,6 +184,27 @@ export const getFullSubjectQuery = ({
 			LIMIT 10
 		)`;
 
+	const migrationItemRunsCte = sql`,
+
+		customer_migration_item_runs AS (
+			SELECT mir.*
+			FROM migration_item_runs mir
+			WHERE mir.item_kind = 'customer'
+				AND mir.dry_run = false
+				AND mir.item_id IN (SELECT internal_id FROM subject_customer_records)
+				AND mir.migration_internal_id IN (
+					SELECT mr.migration_internal_id
+					FROM migration_runs mr
+					WHERE mr.org_id = ${orgId}
+						AND mr.env = ${env}
+						AND mr.status IN ('queued', 'running')
+						AND mr.dry_run = false
+						AND mr.lazy_run = true
+				)
+			ORDER BY mir.updated_at DESC NULLS LAST, mir.created_at DESC
+			LIMIT 10
+		)`;
+
 	const subscriptionsSelect = sql`,
 
 			COALESCE(
@@ -208,6 +229,20 @@ export const getFullSubjectQuery = ({
 				'[]'::json
 			) AS invoices`;
 
+	const migrationItemRunsSelect = sql`,
+
+			COALESCE(
+				(
+					SELECT json_agg(
+						row_to_json(mir)
+						ORDER BY mir.updated_at DESC NULLS LAST, mir.created_at DESC
+					)
+					FROM customer_migration_item_runs mir
+					WHERE mir.item_id = scr.internal_id
+				),
+				'[]'::json
+			) AS migration_item_runs`;
+
 	const entitySelect = entityId
 		? sql`,
 
@@ -283,6 +318,7 @@ export const getFullSubjectQuery = ({
 
 		${subscriptionsCte}
 		${invoicesCte}
+		${migrationItemRunsCte}
 		${entityFragments.ctes}
 		,
 
@@ -467,6 +503,7 @@ export const getFullSubjectQuery = ({
 
 			${subscriptionsSelect}
 			${invoicesSelect}
+			${migrationItemRunsSelect}
 			${entitySelect}
 			${entityFragments.selectColumns}
 

server/src/internal/customers/repos/getFullSubject/subjectQueryRowToNormalized.ts

@@ -266,6 +266,7 @@ export const subjectQueryRowToNormalized = ({
 		subscriptions: row.subscriptions ?? [],
 		invoices: row.invoices ?? [],
 		entity_aggregations: entityAggregations,
+		migration_item_runs: row.migration_item_runs ?? [],
 	};
 };
 

server/src/internal/migrations/v2/actions/migrationRun/finishLazyMigrationRun.ts

@@ -0,0 +1,43 @@
+import { MigrationRunStatus } from "@autumn/shared";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { clearOrgCache } from "@/internal/orgs/orgUtils/clearOrgCache.js";
+import { migrationRunRepo } from "../../repos/index.js";
+
+/**
+ * Mark a lazy migration run as terminally done and bust the org's api-key
+ * cache so `ctx.org.pendingMigrations` drops it on the next authed request.
+ *
+ * This is the "done with this lazy migration" hook — after calling, the
+ * customer-fetch hot path stops checking item_runs for this migration.
+ *
+ * Idempotent: if the run is already at a terminal status the update is a
+ * no-op; we still clear the org cache so callers can use this as a forced
+ * reload mechanism.
+ */
+export const finishLazyMigrationRun = async ({
+	ctx,
+	runId,
+	status = MigrationRunStatus.Succeeded,
+	errorMessage,
+}: {
+	ctx: AutumnContext;
+	runId: string;
+	status?: MigrationRunStatus;
+	errorMessage?: string;
+}): Promise<void> => {
+	await migrationRunRepo.update({
+		ctx,
+		internalId: runId,
+		updates: {
+			status,
+			finished_at: Date.now(),
+			...(errorMessage ? { error_message: errorMessage } : {}),
+		},
+	});
+
+	await clearOrgCache({
+		db: ctx.db,
+		orgId: ctx.org.id,
+		env: ctx.env,
+	});
+};

server/src/internal/migrations/v2/actions/migrationRun/index.ts

@@ -1,9 +1,15 @@
+import { finishLazyMigrationRun } from "./finishLazyMigrationRun.js";
 import { withMigrationRunClaim } from "./withMigrationRunClaim.js";
 import { withMigrationRunTracking } from "./withMigrationRunTracking.js";
 
 export const migrationRunActions = {
+	finishLazy: finishLazyMigrationRun,
 	withClaim: withMigrationRunClaim,
 	withTracking: withMigrationRunTracking,
 } as const;
 
-export { withMigrationRunClaim, withMigrationRunTracking };
+export {
+	finishLazyMigrationRun,
+	withMigrationRunClaim,
+	withMigrationRunTracking,
+};

server/src/internal/migrations/v2/lazy/checkPendingMigrationsForCustomer.ts

@@ -0,0 +1,83 @@
+import type { FullCustomer, MigrationItemRunData } from "@autumn/shared";
+import { customerFilterMatchesFullCustomer } from "@autumn/shared/api/customers/utils/match/index.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { runMigrationCustomerTask } from "@/trigger/migrations/runMigrationCustomerTask.js";
+
+/**
+ * For each pending lazy migration on `ctx.org`, decide whether this customer
+ * needs migrating and enqueue a per-customer Trigger.dev task if so.
+ *
+ * Reads `migration_item_runs` from the already-loaded `fullCustomer` (embedded
+ * by the FullSubject / FullCustomer query) — no extra DB roundtrip.
+ *
+ * Fire-and-forget: the helper doesn't wait for the migration to complete.
+ * `executeMigrateCustomerPlan` inside the task busts the customer cache,
+ * so subsequent requests read post-migration state.
+ */
+export const checkPendingMigrationsForCustomer = async ({
+	ctx,
+	fullCustomer,
+}: {
+	ctx: AutumnContext;
+	fullCustomer: Pick<
+		FullCustomer,
+		"id" | "internal_id" | "customer_products" | "migration_item_runs"
+	>;
+}): Promise<void> => {
+	// Short-circuit when we're already inside a Trigger.dev task. A migration
+	// worker loading the customer via `CusService.getFull` would otherwise
+	// re-enter this helper and enqueue another task. Flag is set by
+	// `createTriggerContext`.
+	if (ctx.insideTriggerTask) return;
+
+	const pending = ctx.org.pendingMigrations ?? [];
+	if (pending.length === 0) return;
+
+	const itemRunsByMigrationInternalId = new Map<string, MigrationItemRunData>();
+	for (const itemRun of fullCustomer.migration_item_runs ?? []) {
+		const existing = itemRunsByMigrationInternalId.get(
+			itemRun.migration_internal_id,
+		);
+		if (
+			!existing ||
+			(itemRun.updated_at ?? 0) > (existing.updated_at ?? 0) ||
+			(itemRun.updated_at === existing.updated_at &&
+				itemRun.created_at > existing.created_at)
+		) {
+			itemRunsByMigrationInternalId.set(itemRun.migration_internal_id, itemRun);
+		}
+	}
+
+	for (const pendingMigration of pending) {
+		const { internal_id: migrationRunId, migration } = pendingMigration;
+
+		const matches = customerFilterMatchesFullCustomer({
+			filter: migration.filter?.customer ?? {},
+			fullCustomer,
+		});
+		if (!matches) continue;
+
+		const itemRun = itemRunsByMigrationInternalId.get(migration.internal_id);
+		if (
+			itemRun?.status === "succeeded" ||
+			itemRun?.status === "skipped" ||
+			itemRun?.status === "running"
+		) {
+			continue;
+		}
+
+		await runMigrationCustomerTask.trigger(
+			{
+				orgId: ctx.org.id,
+				env: ctx.env,
+				migrationInternalId: migration.internal_id,
+				migrationRunId,
+				customerInternalId: fullCustomer.internal_id,
+				customerId: fullCustomer.id ?? null,
+			},
+			{
+				concurrencyKey: `${migration.internal_id}:${fullCustomer.internal_id}`,
+			},
+		);
+	}
+};

server/src/internal/migrations/v2/repos/migrationRun/insertMigrationRun.ts

@@ -39,7 +39,7 @@ export const insertMigrationRun = async ({
 			finished_at: null,
 		})
 		.onConflictDoNothing({
-			target: [migrationRuns.org_id, migrationRuns.env],
+			target: [migrationRuns.migration_internal_id],
 			where: sql`${migrationRuns.status} IN ('queued', 'running')`,
 		})
 		.returning();

server/src/internal/misc/redisV2Cache/redisV2CacheSchemas.ts

@@ -4,7 +4,7 @@ export const RedisV2InstanceName = z.enum(["upstash", "redis", "dragonfly"]);
 export type RedisV2InstanceName = z.infer<typeof RedisV2InstanceName>;
 
 export const RedisV2CacheConfigSchema = z.object({
-	activeInstance: RedisV2InstanceName.default("upstash"),
+	activeInstance: RedisV2InstanceName.default("dragonfly"),
 });
 
 export type RedisV2CacheConfig = z.infer<typeof RedisV2CacheConfigSchema>;

server/src/internal/misc/redisV2Cache/redisV2CacheStore.ts

@@ -11,7 +11,7 @@ import {
 const store = createEdgeConfigStore<RedisV2CacheConfig>({
 	s3Key: ADMIN_REDIS_V2_CACHE_CONFIG_KEY,
 	schema: RedisV2CacheConfigSchema,
-	defaultValue: () => ({ activeInstance: "upstash" }),
+	defaultValue: () => ({ activeInstance: "dragonfly" }),
 	pollIntervalMs: ms.seconds(10),
 });
 

server/src/trigger/migrations/runMigrationCustomerTask.ts

@@ -0,0 +1,103 @@
+import { AppEnv } from "@autumn/shared";
+import { task } from "@trigger.dev/sdk/v3";
+import { z } from "zod/v4";
+import { deleteCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.js";
+import { withMigrationItemTracking } from "@/internal/migrations/v2/actions/migrationItem/index.js";
+import { migrationRepo } from "@/internal/migrations/v2/repos/index.js";
+import { migrateCustomer } from "@/internal/migrations/v2/run/migrateCustomer/index.js";
+import { createTriggerContext } from "@/trigger/utils/createTriggerContext.js";
+
+const PayloadSchema = z.object({
+	orgId: z.string(),
+	env: z.enum(AppEnv),
+	migrationInternalId: z.string(),
+	migrationRunId: z.string(),
+	customerInternalId: z.string(),
+	customerId: z.string().nullable(),
+});
+
+export type RunMigrationCustomerPayload = z.infer<typeof PayloadSchema>;
+
+/**
+ * Per-customer lazy migration task. Enqueued by `checkPendingMigrationsForCustomer`
+ * on the customer-fetch path. Claims the `migration_item_runs` row, busts the
+ * customer cache, then runs `migrateCustomer` under the existing tracking machinery.
+ *
+ * Trigger.dev's `concurrencyKey` (set at enqueue time) serializes parallel
+ * requests for the same customer + migration; the server-side claim is the
+ * real authority if another worker raced ahead.
+ */
+export const runMigrationCustomerTask = task({
+	id: "run-migration-customer",
+	maxDuration: 600,
+	run: async (rawPayload: unknown, { ctx: triggerCtx }) => {
+		const {
+			orgId,
+			env,
+			migrationInternalId,
+			migrationRunId,
+			customerInternalId,
+			customerId,
+		} = PayloadSchema.parse(rawPayload);
+
+		const { ctx, logger } = await createTriggerContext({
+			orgId,
+			env,
+			triggerCtx,
+			customerId: customerId ?? customerInternalId,
+		});
+
+		logger.info("run-migration-customer: starting", {
+			data: { migrationInternalId, migrationRunId, customerInternalId },
+		});
+
+		const migration = await migrationRepo.find({
+			ctx,
+			internalId: migrationInternalId,
+		});
+
+		await withMigrationItemTracking({
+			ctx,
+			migrationInternalId,
+			migrationRunId,
+			item: {
+				kind: "customer",
+				internal_id: customerInternalId,
+				id: customerId,
+			},
+			dryRun: false,
+			claimItemRun: true,
+			run: async () => {
+				// Bust the customer cache as soon as we own the claim so in-flight
+				// reads load fresh state and see the `running` item_run.
+				// `deleteCachedFullCustomer` also invalidates the FullSubject cache.
+				const cacheKey = customerId ?? customerInternalId;
+				await deleteCachedFullCustomer({
+					ctx,
+					customerId: cacheKey,
+					source: "runMigrationCustomerTask",
+				});
+
+				const result = await migrateCustomer({
+					ctx,
+					customerId: cacheKey,
+					migration,
+				});
+
+				return {
+					itemPreview: {
+						id: customerId,
+						name: null,
+						email: null,
+					},
+					status: result.status,
+					response: result.response,
+				};
+			},
+		});
+
+		logger.info("run-migration-customer: done", {
+			data: { migrationInternalId, customerInternalId },
+		});
+	},
+});

server/src/trigger/utils/createTriggerContext.ts

@@ -17,10 +17,12 @@ export const createTriggerContext = async ({
 	orgId,
 	env,
 	triggerCtx,
+	customerId,
 }: {
 	orgId: string;
 	env: AppEnv;
 	triggerCtx: TriggerRunContext;
+	customerId?: string;
 }): Promise<{ ctx: AutumnContext; logger: Logger }> => {
 	const logger = addTriggerToLogs({
 		logger: createDualLogger(),
@@ -33,7 +35,7 @@ export const createTriggerContext = async ({
 
 	const ctx = await createWorkerContext({
 		db,
-		payload: { orgId, env, requestId: triggerCtx.run.id },
+		payload: { orgId, env, customerId, requestId: triggerCtx.run.id },
 		logger,
 	});
 
@@ -42,5 +44,7 @@ export const createTriggerContext = async ({
 			`createTriggerContext: failed to build context for org=${orgId} env=${env}`,
 		);
 
+	ctx.insideTriggerTask = true;
+
 	return { ctx, logger };
 };

server/src/utils/logging/initLogger.ts

@@ -78,7 +78,11 @@ const levelNames: Record<number | string, string> = {
 };
 
 /** Bun-friendly pino sink that prints `<ts> <LEVEL> <msg> <extras>`. */
-const createDevLogStream = () =>
+const createDevLogStream = ({
+	trailingNewline = true,
+}: {
+	trailingNewline?: boolean;
+} = {}) =>
 	new Writable({
 		write(chunk, _encoding, callback) {
 			try {
@@ -108,7 +112,7 @@ const createDevLogStream = () =>
 					message += ` ${JSON.stringify(additionalFields, null, 2)}`;
 				}
 
-				const formattedLog = `${colors.gray}${timestamp}${colors.reset} ${levelColor}${colors.bright}${levelName}${colors.reset} ${message}\n`;
+				const formattedLog = `${colors.gray}${timestamp}${colors.reset} ${levelColor}${colors.bright}${levelName}${colors.reset} ${message}${trailingNewline ? "\n" : ""}`;
 
 				process.stdout.write(formattedLog);
 				callback();
@@ -145,7 +149,9 @@ export const initLogger = (options: InitLoggerOptions = {}) => {
 	if (mode === "dual") {
 		streams.push({
 			level: isDevOrTest ? "debug" : "info",
-			stream: isDevOrTest ? createDevLogStream() : process.stdout,
+			stream: isDevOrTest
+				? createDevLogStream({ trailingNewline: false })
+				: process.stdout,
 		});
 		if (process.env.AXIOM_TOKEN) {
 			streams.push({
@@ -208,7 +214,7 @@ export const initLogger = (options: InitLoggerOptions = {}) => {
 				};
 			},
 			formatters: {
-				level: (label: any) => {
+				level: (label: string) => {
 					return {
 						level: label.toUpperCase(),
 					};

server/tests/integration/billing/migrations-v2/lazy/lazy-migration-basic.test.ts

@@ -0,0 +1,249 @@
+/**
+ * TDD test for Phase 2 of lazy migrations: customer-fetch path triggers the
+ * per-customer Trigger.dev task and the customer ends up migrated.
+ *
+ * Contract under test:
+ *   Behavior:
+ *     - POST /migrations.lazy_run starts a lazy-mode migration_run
+ *       (lazy_run=true, status='running')
+ *     - On the next /customers.get for an eligible customer:
+ *       * runMigrationCustomerTask is enqueued with concurrencyKey
+ *         `${migration_internal_id}:${customer.internal_id}`
+ *       * eventually `migration_item_runs` has one `succeeded` row
+ *       * subsequent /customers.get returns post-migration state
+ *         (e.g. the new `dashboard` feature is now present)
+ *   Side effects:
+ *     - executeMigrateCustomerPlan invalidates the customer cache
+ *     - exactly one `migration_item_runs` row per (migration, customer)
+ */
+
+import { expect, test } from "bun:test";
+import {
+	MigrationItemKind,
+	MigrationItemRunStatus,
+	migrationItemRuns,
+} from "@autumn/shared";
+import { TestFeature } from "@tests/setup/v2Features.js";
+import { products } from "@tests/utils/fixtures/products.js";
+import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
+import chalk from "chalk";
+import { and, eq } from "drizzle-orm";
+import { finishLazyMigrationRun } from "@/internal/migrations/v2/actions/migrationRun/finishLazyMigrationRun.js";
+import {
+	countCustomerItemRunRows,
+	getCustomerAndAwaitMigration,
+	getInternalCustomerId,
+	releaseLazyMigrationRun,
+	startLazyMigration,
+	waitForCustomerItemRunStatus,
+} from "./utils/lazyMigrationTestUtils.js";
+
+const timeout = (ms: number) =>
+	new Promise((resolve) => setTimeout(resolve, ms));
+
+test.concurrent(`${chalk.yellowBright("lazy migration: multiple customers on pro fetch + auto-migrate")}`, async () => {
+	const firstCustomerId = "lazy-basic-first";
+	const secondCustomerId = "lazy-basic-second";
+	const thirdCustomerId = "lazy-basic-third";
+	const plan = products.pro({ id: "lazy-basic-pro", items: [] });
+
+	const { autumnV2_2, ctx } = await initScenario({
+		customerId: firstCustomerId,
+		setup: [
+			s.customer({ paymentMethod: "success" }),
+			s.otherCustomers([
+				{ id: secondCustomerId, paymentMethod: "success" },
+				{ id: thirdCustomerId, paymentMethod: "success" },
+			]),
+			s.products({ list: [plan] }),
+		],
+		actions: [
+			s.billing.attach({ productId: plan.id }),
+			s.billing.attach({
+				customerId: secondCustomerId,
+				productId: plan.id,
+			}),
+			s.billing.attach({
+				customerId: thirdCustomerId,
+				productId: plan.id,
+			}),
+		],
+	});
+
+	const { migration, run_id } = await startLazyMigration({
+		autumnV2_2,
+		ctx,
+		id: `${firstCustomerId}-mig`,
+		planId: plan.id,
+	});
+
+	try {
+		const customerIds = [firstCustomerId, secondCustomerId, thirdCustomerId];
+
+		// ── Contract assertion 1: fetch + auto-migrate for each customer ──
+		for (const customerId of customerIds) {
+			const customer = await getCustomerAndAwaitMigration({
+				autumnV2_2,
+				customerId,
+			});
+			expect(customer.flags[TestFeature.Dashboard]).toBeDefined();
+		}
+
+		// ── Contract assertion 2: each customer has exactly one succeeded row ──
+		for (const customerId of customerIds) {
+			const internalCustomerId = await getInternalCustomerId({
+				customerId,
+				ctx,
+			});
+			await waitForCustomerItemRunStatus({
+				ctx,
+				migration,
+				internalCustomerId,
+				status: MigrationItemRunStatus.Succeeded,
+			});
+
+			const rowCount = await countCustomerItemRunRows({
+				ctx,
+				migration,
+				internalCustomerId,
+			});
+			expect(rowCount).toBe(1);
+		}
+	} finally {
+		await releaseLazyMigrationRun({ ctx, runId: run_id });
+	}
+});
+
+test.concurrent(`${chalk.yellowBright("lazy migration: non-matching customer is not migrated")}`, async () => {
+	const matchingCustomerId = "lazy-basic-matching";
+	const mismatchCustomerId = "lazy-basic-mismatch";
+	const proPlan = products.pro({ id: "lazy-basic-target-pro", items: [] });
+	const premiumPlan = products.premium({
+		id: "lazy-basic-other-premium",
+		items: [],
+	});
+
+	const { autumnV2_2, ctx } = await initScenario({
+		customerId: matchingCustomerId,
+		setup: [
+			s.customer({ paymentMethod: "success" }),
+			s.otherCustomers([{ id: mismatchCustomerId, paymentMethod: "success" }]),
+			s.products({ list: [proPlan, premiumPlan] }),
+		],
+		actions: [
+			s.billing.attach({ productId: proPlan.id }),
+			s.billing.attach({
+				customerId: mismatchCustomerId,
+				productId: premiumPlan.id,
+			}),
+		],
+	});
+
+	const { migration, run_id } = await startLazyMigration({
+		autumnV2_2,
+		ctx,
+		id: `${matchingCustomerId}-mig`,
+		planId: proPlan.id,
+	});
+
+	try {
+		// Matching customer migrates as expected.
+		await getCustomerAndAwaitMigration({
+			autumnV2_2,
+			customerId: matchingCustomerId,
+		});
+
+		// Mismatching customer's fetch is the in-memory pre-check site — assert
+		// it neither queues a task nor writes a migration_item_runs row.
+		const mismatchCustomer = await autumnV2_2.customers.get(mismatchCustomerId);
+		expect(mismatchCustomer).toBeDefined();
+
+		const mismatchInternalId = await getInternalCustomerId({
+			customerId: mismatchCustomerId,
+			ctx,
+		});
+		const mismatchRowCount = await countCustomerItemRunRows({
+			ctx,
+			migration,
+			internalCustomerId: mismatchInternalId,
+		});
+		expect(mismatchRowCount).toBe(0);
+	} finally {
+		await releaseLazyMigrationRun({ ctx, runId: run_id });
+	}
+});
+
+test.concurrent(`${chalk.yellowBright("lazy migration: marking the run as done clears the org cache and stops the in-memory check")}`, async () => {
+	const customerId = "lazy-basic-done-clears";
+	const plan = products.pro({ id: "lazy-basic-done-pro", items: [] });
+
+	const { autumnV2_2, ctx } = await initScenario({
+		customerId,
+		setup: [
+			s.customer({ paymentMethod: "success" }),
+			s.products({ list: [plan] }),
+		],
+		actions: [s.billing.attach({ productId: plan.id })],
+	});
+	const internalCustomerId = await getInternalCustomerId({ customerId, ctx });
+
+	const { migration, run_id } = await startLazyMigration({
+		autumnV2_2,
+		ctx,
+		id: `${customerId}-mig`,
+		planId: plan.id,
+	});
+
+	try {
+		// Trigger the migration once and wait for it to land.
+		await getCustomerAndAwaitMigration({ autumnV2_2, customerId });
+		await waitForCustomerItemRunStatus({
+			ctx,
+			migration,
+			internalCustomerId,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		expect(
+			await countCustomerItemRunRows({
+				ctx,
+				migration,
+				internalCustomerId,
+			}),
+		).toBe(1);
+
+		// Finish the lazy run — should mark succeeded + clear org cache.
+		await finishLazyMigrationRun({ ctx, runId: run_id });
+
+		// Manually delete the customer's item_run row so we can detect whether
+		// the helper still queues a task post-completion. If `pendingMigrations`
+		// is correctly empty on the next request, no task should run and the
+		// row should stay deleted.
+		await ctx.db
+			.delete(migrationItemRuns)
+			.where(
+				and(
+					eq(migrationItemRuns.migration_internal_id, migration.internal_id),
+					eq(migrationItemRuns.item_kind, MigrationItemKind.Customer),
+					eq(migrationItemRuns.item_id, internalCustomerId),
+				),
+			);
+
+		// Hit /customers.get a few times and give any queued task time to run.
+		for (let i = 0; i < 5; i++) {
+			await autumnV2_2.customers.get(customerId);
+		}
+		await timeout(3_000);
+
+		// Helper should have short-circuited at `pendingMigrations.length === 0`
+		// → no new `migration_item_runs` row recreated.
+		expect(
+			await countCustomerItemRunRows({
+				ctx,
+				migration,
+				internalCustomerId,
+			}),
+		).toBe(0);
+	} finally {
+		await releaseLazyMigrationRun({ ctx, runId: run_id });
+	}
+});

server/tests/integration/billing/migrations-v2/lazy/lazy-migration-concurrency.test.ts

@@ -0,0 +1,214 @@
+/**
+ * TDD test for Phase 2 lazy migrations — concurrency safety.
+ *
+ * Contract under test:
+ *   - Concurrent /customers.get requests for the same customer during an
+ *     active lazy migration may enqueue multiple Trigger.dev tasks, but
+ *     `migration_item_runs` claim machinery guarantees the migration runs
+ *     exactly once for that customer.
+ *   - The partial unique index on `migration_runs (migration_internal_id)
+ *     WHERE status IN ('queued','running')` rejects a second `lazy_run`
+ *     call for the SAME migration definition while one is already active.
+ *   - Different migration definitions CAN have active lazy runs
+ *     concurrently — correctness lives at the per-customer claim layer.
+ */
+
+import { expect, test } from "bun:test";
+import { ErrCode, MigrationItemRunStatus } from "@autumn/shared";
+import { TestFeature } from "@tests/setup/v2Features.js";
+import { itemsV2 } from "@tests/utils/fixtures/itemsV2.js";
+import { products } from "@tests/utils/fixtures/products.js";
+import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
+import chalk from "chalk";
+import {
+	buildDashboardLazyMigration,
+	countCustomerItemRunRows,
+	getCustomerAndAwaitMigration,
+	getInternalCustomerId,
+	releaseLazyMigrationRun,
+	startLazyMigration,
+	waitForCustomerItemRunStatus,
+} from "./utils/lazyMigrationTestUtils.js";
+
+test.concurrent(`${chalk.yellowBright("lazy migration concurrency: concurrent fetches execute migration exactly once")}`, async () => {
+	const customerId = "lazy-concurrent-fetch";
+	const plan = products.pro({ id: "lazy-concurrent-pro", items: [] });
+
+	const { autumnV2_2, ctx } = await initScenario({
+		customerId,
+		setup: [
+			s.customer({ paymentMethod: "success" }),
+			s.products({ list: [plan] }),
+		],
+		actions: [s.billing.attach({ productId: plan.id })],
+	});
+
+	const { migration, run_id } = await startLazyMigration({
+		autumnV2_2,
+		ctx,
+		id: `${customerId}-mig`,
+		planId: plan.id,
+	});
+
+	try {
+		const concurrency = 10;
+		await Promise.all(
+			Array.from({ length: concurrency }, () =>
+				autumnV2_2.customers.get(customerId),
+			),
+		);
+
+		const internalCustomerId = await getInternalCustomerId({
+			customerId,
+			ctx,
+		});
+		await waitForCustomerItemRunStatus({
+			ctx,
+			migration,
+			internalCustomerId,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+
+		const rowCount = await countCustomerItemRunRows({
+			ctx,
+			migration,
+			internalCustomerId,
+		});
+		expect(rowCount).toBe(1);
+
+		const migrated = await getCustomerAndAwaitMigration({
+			autumnV2_2,
+			customerId,
+		});
+		expect(migrated.flags[TestFeature.Dashboard]).toBeDefined();
+	} finally {
+		await releaseLazyMigrationRun({ ctx, runId: run_id });
+	}
+});
+
+test.concurrent(`${chalk.yellowBright("lazy migration concurrency: starting a second lazy_run on the SAME migration is rejected (409)")}`, async () => {
+	const customerId = "lazy-same-mig-rejected";
+	const plan = products.pro({ id: "lazy-same-mig-pro", items: [] });
+
+	const { autumnV2_2, ctx } = await initScenario({
+		customerId,
+		setup: [
+			s.customer({ paymentMethod: "success" }),
+			s.products({ list: [plan] }),
+		],
+		actions: [s.billing.attach({ productId: plan.id })],
+	});
+
+	const { migration, run_id } = await startLazyMigration({
+		autumnV2_2,
+		ctx,
+		id: `${customerId}-mig`,
+		planId: plan.id,
+	});
+
+	try {
+		// Same migration id → 409 (partial unique index on migration_internal_id).
+		await expect(
+			autumnV2_2.migrationsV2.lazyRun({ id: migration.id }),
+		).rejects.toMatchObject({
+			code: ErrCode.MigrationAlreadyInProgress,
+			message: expect.stringContaining("already running"),
+		});
+	} finally {
+		await releaseLazyMigrationRun({ ctx, runId: run_id });
+	}
+});
+
+test.concurrent(`${chalk.yellowBright("lazy migration concurrency: two different migrations on the same customer can run concurrently")}`, async () => {
+	const customerId = "lazy-two-migs";
+	const plan = products.pro({ id: "lazy-two-migs-pro", items: [] });
+
+	const { autumnV2_2, ctx } = await initScenario({
+		customerId,
+		setup: [
+			s.customer({ paymentMethod: "success" }),
+			s.products({ list: [plan] }),
+		],
+		actions: [s.billing.attach({ productId: plan.id })],
+	});
+	const internalCustomerId = await getInternalCustomerId({ customerId, ctx });
+
+	// First migration — adds Dashboard.
+	const { migration: firstMigration, run_id: firstRunId } =
+		await startLazyMigration({
+			autumnV2_2,
+			ctx,
+			id: `${customerId}-first-mig`,
+			planId: plan.id,
+		});
+
+	// Second migration — adds AdminRights. Starts WHILE first is still active.
+	const secondMigrationDef = await autumnV2_2.migrationsV2.deleteAndCreate({
+		id: `${customerId}-second-mig`,
+		filter: { customer: { plan: { plan_id: plan.id } } },
+		operations: {
+			customer: [
+				{
+					type: "update_plan" as const,
+					plan_filter: { plan_id: plan.id },
+					customize: {
+						add_items: [{ feature_id: TestFeature.AdminRights }],
+					},
+				},
+			],
+		},
+	});
+	const secondRun = await autumnV2_2.migrationsV2.lazyRun({
+		id: secondMigrationDef.id,
+	});
+
+	try {
+		// Customer fetch eventually picks up BOTH migrations — poll in parallel
+		// since each migration is independent and triggers its own task.
+		await Promise.all([
+			getCustomerAndAwaitMigration({
+				autumnV2_2,
+				customerId,
+				featureId: TestFeature.Dashboard,
+			}),
+			getCustomerAndAwaitMigration({
+				autumnV2_2,
+				customerId,
+				featureId: TestFeature.AdminRights,
+			}),
+		]);
+
+		await Promise.all([
+			waitForCustomerItemRunStatus({
+				ctx,
+				migration: firstMigration,
+				internalCustomerId,
+				status: MigrationItemRunStatus.Succeeded,
+			}),
+			waitForCustomerItemRunStatus({
+				ctx,
+				migration: secondMigrationDef,
+				internalCustomerId,
+				status: MigrationItemRunStatus.Succeeded,
+			}),
+		]);
+
+		expect(
+			await countCustomerItemRunRows({
+				ctx,
+				migration: firstMigration,
+				internalCustomerId,
+			}),
+		).toBe(1);
+		expect(
+			await countCustomerItemRunRows({
+				ctx,
+				migration: secondMigrationDef,
+				internalCustomerId,
+			}),
+		).toBe(1);
+	} finally {
+		await releaseLazyMigrationRun({ ctx, runId: firstRunId });
+		await releaseLazyMigrationRun({ ctx, runId: secondRun.run_id });
+	}
+});

server/tests/integration/billing/migrations-v2/lazy/lazy-migration-multi-entity.test.ts

@@ -0,0 +1,187 @@
+/**
+ * TDD test for Phase 2 lazy migrations — multi-entity customers.
+ *
+ * Contract under test:
+ *   - When a customer has multiple entities on the same plan and the
+ *     lazy migration filter matches the customer (not entity-specific),
+ *     concurrent entity-scoped fetches must NOT cause double execution.
+ *   - The migration's `migration_item_runs` row is keyed by customer,
+ *     not entity, so even with N entities only one row should exist
+ *     for the customer after migration.
+ *   - After the migration completes, every entity returns post-migration
+ *     state on subsequent fetches.
+ *   - When entities each have their own (free) attached plan, migrating
+ *     that plan touches BOTH entity-scoped customer_products through one
+ *     customer-level item_run.
+ */
+
+import { expect, test } from "bun:test";
+import {
+	type ApiCustomerV5,
+	MigrationItemRunStatus,
+} from "@autumn/shared";
+import { TestFeature } from "@tests/setup/v2Features.js";
+import { products } from "@tests/utils/fixtures/products.js";
+import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
+import chalk from "chalk";
+import {
+	countCustomerItemRunRows,
+	getCustomerAndAwaitMigration,
+	getInternalCustomerId,
+	releaseLazyMigrationRun,
+	startLazyMigration,
+	waitForCustomerItemRunStatus,
+} from "./utils/lazyMigrationTestUtils.js";
+
+test.concurrent(
+	`${chalk.yellowBright("lazy migration multi-entity: concurrent entity fetches still migrate the customer exactly once")}`,
+	async () => {
+		const customerId = "lazy-multi-entity";
+		const plan = products.pro({ id: "lazy-multi-entity-pro", items: [] });
+
+		const { autumnV2_2, ctx, entities } = await initScenario({
+			customerId,
+			setup: [
+				s.customer({ paymentMethod: "success" }),
+				s.products({ list: [plan] }),
+				s.entities({ count: 2, featureId: TestFeature.Users }),
+			],
+			actions: [s.billing.attach({ productId: plan.id })],
+		});
+
+		expect(entities.length).toBe(2);
+		const [firstEntity, secondEntity] = entities;
+		if (!firstEntity || !secondEntity)
+			throw new Error("expected 2 entities to be generated");
+
+		const { migration, run_id } = await startLazyMigration({
+			autumnV2_2,
+			ctx,
+			id: `${customerId}-mig`,
+			planId: plan.id,
+		});
+
+		try {
+			// Concurrent entity-scoped fetches. Both may enqueue their own
+			// runMigrationCustomerTask — the claim must serialize them.
+			await Promise.all([
+				autumnV2_2.entities.get(customerId, firstEntity.id),
+				autumnV2_2.entities.get(customerId, secondEntity.id),
+			]);
+
+			// Migration eventually completes for the underlying customer.
+			const internalCustomerId = await getInternalCustomerId({
+				customerId,
+				ctx,
+			});
+			await waitForCustomerItemRunStatus({
+				ctx,
+				migration,
+				internalCustomerId,
+				status: MigrationItemRunStatus.Succeeded,
+			});
+
+			// Exactly one item_run row (keyed by customer, not entity).
+			const rowCount = await countCustomerItemRunRows({
+				ctx,
+				migration,
+				internalCustomerId,
+			});
+			expect(rowCount).toBe(1);
+
+			// Subsequent /customers.get returns post-migration state.
+			const customer = await getCustomerAndAwaitMigration({
+				autumnV2_2,
+				customerId,
+			});
+			expect(customer.flags[TestFeature.Dashboard]).toBeDefined();
+		} finally {
+			await releaseLazyMigrationRun({ ctx, runId: run_id });
+		}
+	},
+);
+
+test.concurrent(
+	`${chalk.yellowBright("lazy migration multi-entity: per-entity free plans both migrated under one customer item_run")}`,
+	async () => {
+		const customerId = "lazy-multi-entity-free";
+		const freePlan = products.base({
+			id: "lazy-multi-entity-free-plan",
+			items: [],
+		});
+
+		const { autumnV2_2, ctx, entities } = await initScenario({
+			customerId,
+			setup: [
+				s.customer({ paymentMethod: "success" }),
+				s.products({ list: [freePlan] }),
+				s.entities({ count: 2, featureId: TestFeature.Users }),
+			],
+			actions: [
+				// Attach the free plan once per entity. Each attach produces an
+				// entity-scoped customer_product row (`internal_entity_id` set).
+				s.billing.attach({ productId: freePlan.id, entityIndex: 0 }),
+				s.billing.attach({ productId: freePlan.id, entityIndex: 1 }),
+			],
+		});
+
+		expect(entities.length).toBe(2);
+		const [firstEntity, secondEntity] = entities;
+		if (!firstEntity || !secondEntity)
+			throw new Error("expected 2 entities to be generated");
+
+		const { migration, run_id } = await startLazyMigration({
+			autumnV2_2,
+			ctx,
+			id: `${customerId}-mig`,
+			planId: freePlan.id,
+		});
+
+		try {
+			// Entity-scoped fetches surface entity-scoped customer_products in
+			// FullSubject, so the helper's filter pre-check matches there. The
+			// task itself loads a fresh customer-level view inside
+			// `setupMigrateCustomerContext` (`withEntities: true`) and applies
+			// the operation to BOTH entity-scoped plans in one execution.
+			await Promise.all([
+				autumnV2_2.entities.get<ApiCustomerV5>(customerId, firstEntity.id),
+				autumnV2_2.entities.get<ApiCustomerV5>(customerId, secondEntity.id),
+			]);
+
+			const internalCustomerId = await getInternalCustomerId({
+				customerId,
+				ctx,
+			});
+			await waitForCustomerItemRunStatus({
+				ctx,
+				migration,
+				internalCustomerId,
+				status: MigrationItemRunStatus.Succeeded,
+			});
+
+			// ── A. Migration ran once ────────────────────────────────────────
+			expect(
+				await countCustomerItemRunRows({
+					ctx,
+					migration,
+					internalCustomerId,
+				}),
+			).toBe(1);
+
+			// ── B. Both entity-scoped free plans carry the migrated state ─────
+			const firstEntityView = await autumnV2_2.entities.get<ApiCustomerV5>(
+				customerId,
+				firstEntity.id,
+			);
+			const secondEntityView = await autumnV2_2.entities.get<ApiCustomerV5>(
+				customerId,
+				secondEntity.id,
+			);
+
+			expect(firstEntityView.flags[TestFeature.Dashboard]).toBeDefined();
+			expect(secondEntityView.flags[TestFeature.Dashboard]).toBeDefined();
+		} finally {
+			await releaseLazyMigrationRun({ ctx, runId: run_id });
+		}
+	},
+);

server/tests/integration/billing/migrations-v2/lazy/utils/lazyMigrationTestUtils.ts

@@ -0,0 +1,230 @@
+import { expect } from "bun:test";
+import {
+	type ApiCustomerV5,
+	type Migration,
+	MigrationItemKind,
+	type MigrationItemRun,
+	MigrationItemRunStatus,
+	migrationRuns,
+} from "@autumn/shared";
+import { TestFeature } from "@tests/setup/v2Features.js";
+import { itemsV2 } from "@tests/utils/fixtures/itemsV2.js";
+import type { initScenario } from "@tests/utils/testInitUtils/initScenario.js";
+import { and, eq } from "drizzle-orm";
+import { CusService } from "@/internal/customers/CusService.js";
+import { finishLazyMigrationRun } from "@/internal/migrations/v2/actions/migrationRun/finishLazyMigrationRun.js";
+import { migrationItemRunRepo } from "@/internal/migrations/v2/repos/index.js";
+import { waitForMigrationResult } from "../../utils/runUpdatePlanMigration.js";
+
+type ScenarioCtx = Awaited<ReturnType<typeof initScenario>>["ctx"];
+type AutumnV2_2 = Awaited<ReturnType<typeof initScenario>>["autumnV2_2"];
+
+/** Boilerplate dashboard migration — adds the `TestFeature.Dashboard` boolean
+ *  feature to every customer matching the given plan filter. */
+export const buildDashboardLazyMigration = ({
+	id,
+	planId,
+}: {
+	id: string;
+	planId: string;
+}) => ({
+	id,
+	filter: { customer: { plan: { plan_id: planId } } },
+	operations: {
+		customer: [
+			{
+				type: "update_plan" as const,
+				plan_filter: { plan_id: planId },
+				customize: {
+					add_items: [itemsV2.dashboard()],
+				},
+			},
+		],
+	},
+});
+
+export const getInternalCustomerId = async ({
+	customerId,
+	ctx,
+}: {
+	customerId: string;
+	ctx: ScenarioCtx;
+}) => {
+	const customer = await CusService.get({
+		db: ctx.db,
+		idOrInternalId: customerId,
+		orgId: ctx.org.id,
+		env: ctx.env,
+	});
+	if (!customer) throw new Error(`Expected customer ${customerId}`);
+	return customer.internal_id;
+};
+
+export const getCustomerItemRun = ({
+	ctx,
+	migration,
+	internalCustomerId,
+}: {
+	ctx: ScenarioCtx;
+	migration: Migration;
+	internalCustomerId: string;
+}) =>
+	migrationItemRunRepo.getCustomer({
+		ctx,
+		migrationInternalId: migration.internal_id,
+		internalCustomerId,
+	});
+
+export const waitForCustomerItemRunStatus = async ({
+	ctx,
+	migration,
+	internalCustomerId,
+	status,
+	timeoutMs = 60_000,
+	pollIntervalMs = 1_000,
+}: {
+	ctx: ScenarioCtx;
+	migration: Migration;
+	internalCustomerId: string;
+	status: MigrationItemRunStatus;
+	timeoutMs?: number;
+	pollIntervalMs?: number;
+}) =>
+	waitForMigrationResult({
+		timeoutMs,
+		pollIntervalMs,
+		waitFor: async () => {
+			expect(
+				await getCustomerItemRun({
+					ctx,
+					migration,
+					internalCustomerId,
+				}),
+			).toMatchObject({ status });
+		},
+	});
+
+/** Count `migration_item_runs` rows for a given (migration, customer). The
+ *  partial unique index `migration_item_runs_live_unique` guarantees at most
+ *  one row, so this should always be 0 or 1. */
+export const countCustomerItemRunRows = async ({
+	ctx,
+	migration,
+	internalCustomerId,
+}: {
+	ctx: ScenarioCtx;
+	migration: Migration;
+	internalCustomerId: string;
+}): Promise<number> => {
+	const rows = (await ctx.db.query.migrationItemRuns.findMany({
+		where: (mir, { and, eq }) =>
+			and(
+				eq(mir.migration_internal_id, migration.internal_id),
+				eq(mir.item_kind, MigrationItemKind.Customer),
+				eq(mir.item_id, internalCustomerId),
+				eq(mir.dry_run, false),
+			),
+	})) as MigrationItemRun[];
+	return rows.length;
+};
+
+/** Start a lazy migration. Only one active `migration_runs` row can exist
+ *  per `(org, env)` (partial unique index), so concurrent tests in the same
+ *  file serialize here: if another test holds the claim, we poll until it
+ *  releases (succeeded / failed) then try again. */
+export const startLazyMigration = async ({
+	autumnV2_2,
+	ctx,
+	id,
+	planId,
+	timeoutMs = 120_000,
+	pollIntervalMs = 500,
+}: {
+	autumnV2_2: AutumnV2_2;
+	ctx: ScenarioCtx;
+	id: string;
+	planId: string;
+	timeoutMs?: number;
+	pollIntervalMs?: number;
+}): Promise<{ migration: Migration; run_id: string }> => {
+	const migration = await autumnV2_2.migrationsV2.deleteAndCreate(
+		buildDashboardLazyMigration({ id, planId }),
+	);
+
+	const deadline = Date.now() + timeoutMs;
+	let lastError: unknown;
+	while (Date.now() < deadline) {
+		try {
+			const run = await autumnV2_2.migrationsV2.lazyRun({ id: migration.id });
+			return { migration, run_id: run.run_id };
+		} catch (error) {
+			lastError = error;
+			const code = (error as { code?: string } | undefined)?.code;
+			if (code !== "migration_already_in_progress") throw error;
+			await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
+		}
+	}
+
+	throw new Error(
+		`startLazyMigration: timed out after ${timeoutMs}ms waiting for another run to release. last error: ${
+			lastError instanceof Error ? lastError.message : String(lastError)
+		}`,
+	);
+};
+
+/** Fetch the customer and poll until the migration's added feature shows up.
+ *  Used as the standard “fetch + wait for lazy migration to land” primitive. */
+export const getCustomerAndAwaitMigration = async ({
+	autumnV2_2,
+	customerId,
+	featureId = TestFeature.Dashboard,
+	timeoutMs = 60_000,
+	pollIntervalMs = 1_000,
+}: {
+	autumnV2_2: AutumnV2_2;
+	customerId: string;
+	featureId?: string;
+	timeoutMs?: number;
+	pollIntervalMs?: number;
+}): Promise<ApiCustomerV5> => {
+	let latest: ApiCustomerV5 | undefined;
+	await waitForMigrationResult({
+		timeoutMs,
+		pollIntervalMs,
+		waitFor: async () => {
+			latest = await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
+			expect(latest.flags[featureId]).toBeDefined();
+		},
+	});
+	if (!latest) throw new Error("getCustomerAndAwaitMigration: no response");
+	return latest;
+};
+
+/** Wipe every `migration_runs` row for the test org/env. */
+export const cleanupMigrationRunsForOrg = async ({
+	ctx,
+}: {
+	ctx: ScenarioCtx;
+}): Promise<void> => {
+	await ctx.db
+		.delete(migrationRuns)
+		.where(
+			and(eq(migrationRuns.org_id, ctx.org.id), eq(migrationRuns.env, ctx.env)),
+		);
+};
+
+/** Mark a lazy run as done via the real `finishLazyMigrationRun` action and
+ *  bust the org cache. Use this in `finally` blocks so each test cleans up
+ *  ITS OWN run without stepping on concurrent tests. Idempotent — safe to
+ *  call multiple times. */
+export const releaseLazyMigrationRun = async ({
+	ctx,
+	runId,
+}: {
+	ctx: ScenarioCtx;
+	runId: string;
+}): Promise<void> => {
+	await finishLazyMigrationRun({ ctx, runId });
+};
+
+export { MigrationItemRunStatus };

shared/api/customers/utils/match/customerFilterMatchesFullCustomer.ts

@@ -0,0 +1,65 @@
+import type { FullCustomer } from "../../../../models/cusModels/fullCusModel.js";
+import { customerProductHasActiveStatus } from "../../../../utils/index.js";
+import type { CustomerFilter } from "../../../migrations/filters/customerFilter.js";
+import {
+	arrayFilterMatches,
+	stringMatcherMatches,
+} from "../../../migrations/filters/match/index.js";
+import { planFilterMatchesCustomerProduct } from "../../../products/utils/match/planFilterMatchesCustomerProduct.js";
+
+/**
+ * Predicate: does `filter` match a customer with `customer_products`?
+ *
+ * JS-side mirror of the SQL compiler's `customerRegistry`. Used by the lazy
+ * migration helper to skip non-matching customers without queueing work.
+ * Mirrors the `cp.status IN ACTIVE_STATUSES` ambient predicate baked into
+ * the SQL plan scope — non-active cusProducts are ignored.
+ *
+ * Supports `customer_id` and `plan` (`$some` / `$every` / `$none` and the
+ * implicit-`$some` bare form). `item` sugar throws to make the gap explicit,
+ * matching the convention in `planFilterMatchesCustomerProduct`.
+ */
+export const customerFilterMatchesFullCustomer = ({
+	filter,
+	fullCustomer,
+}: {
+	filter: CustomerFilter;
+	fullCustomer: Pick<FullCustomer, "id" | "customer_products">;
+}): boolean => {
+	if (
+		filter.customer_id !== undefined &&
+		!stringMatcherMatches({
+			matcher: filter.customer_id,
+			value: fullCustomer.id,
+		})
+	) {
+		return false;
+	}
+
+	if (filter.plan !== undefined) {
+		const activeProducts = fullCustomer.customer_products.filter(
+			customerProductHasActiveStatus,
+		);
+		if (
+			!arrayFilterMatches({
+				filter: filter.plan,
+				items: activeProducts,
+				matchesElement: ({ filter: planFilter, item: customerProduct }) =>
+					planFilterMatchesCustomerProduct({
+						filter: planFilter,
+						cusProduct: customerProduct,
+					}),
+			})
+		) {
+			return false;
+		}
+	}
+
+	if (filter.item !== undefined) {
+		throw new Error(
+			"customerFilterMatchesFullCustomer: filter.item not supported in JS matcher yet",
+		);
+	}
+
+	return true;
+};

shared/api/customers/utils/match/index.ts

@@ -0,0 +1 @@
+export * from "./customerFilterMatchesFullCustomer.js";

shared/api/migrations/compiler/filterToIr/fields/index.ts

@@ -1,5 +0,0 @@
-export * from "./makeExistenceParser.js";
-export * from "./parseLeaf.js";
-export * from "./parsePriceExistence.js";
-export * from "./parseRolloverExistence.js";
-export * from "./translateValue.js";

shared/api/migrations/compiler/filterToIr/helpers/index.ts

@@ -1,2 +0,0 @@
-export * from "./isQuantifierWrapper.js";
-export * from "./wrapAnd.js";

shared/api/migrations/compiler/filterToIr/index.ts

@@ -1,2 +0,0 @@
-export * from "./filterToIr.js";
-export * from "./resolutionContext.js";

shared/api/migrations/compiler/filterToIr/navs/index.ts

@@ -1,2 +0,0 @@
-export * from "./parseItemNav.js";
-export * from "./parsePlanNav.js";

shared/api/migrations/compiler/filterToIr/scopes/index.ts

@@ -1,3 +0,0 @@
-export * from "./parseCustomerFilter.js";
-export * from "./parsePlanFilter.js";
-export * from "./parsePlanItemFilter.js";

shared/api/migrations/filters/match/arrayFilterMatches.ts

@@ -0,0 +1,76 @@
+/**
+ * Generic in-memory evaluator for the `arrayFilter` quantifier shape.
+ *
+ * `filter` is either a bare element filter (implicit `$some`) or a
+ * `{ $some?, $every?, $none? }` wrapper. Multiple quantifiers in the
+ * same wrapper are AND'd.
+ *
+ * Reusable across any nav — pass the element matcher as `matchesElement`.
+ */
+export type ArrayQuantifierFilter<ElementFilter> =
+	| ElementFilter
+	| {
+			$some?: ElementFilter;
+			$every?: ElementFilter;
+			$none?: ElementFilter;
+	  };
+
+export const arrayFilterMatches = <Item, ElementFilter>({
+	filter,
+	items,
+	matchesElement,
+}: {
+	filter: ArrayQuantifierFilter<ElementFilter>;
+	items: Item[];
+	matchesElement: ({
+		filter,
+		item,
+	}: {
+		filter: ElementFilter;
+		item: Item;
+	}) => boolean;
+}): boolean => {
+	const wrapped = isQuantifierWrapper<ElementFilter>(filter)
+		? filter
+		: { $some: filter };
+
+	if (
+		wrapped.$some !== undefined &&
+		!items.some((item) =>
+			matchesElement({ filter: wrapped.$some as ElementFilter, item }),
+		)
+	) {
+		return false;
+	}
+
+	if (
+		wrapped.$every !== undefined &&
+		!items.every((item) =>
+			matchesElement({ filter: wrapped.$every as ElementFilter, item }),
+		)
+	) {
+		return false;
+	}
+
+	if (
+		wrapped.$none !== undefined &&
+		items.some((item) =>
+			matchesElement({ filter: wrapped.$none as ElementFilter, item }),
+		)
+	) {
+		return false;
+	}
+
+	return true;
+};
+
+const isQuantifierWrapper = <ElementFilter>(
+	value: ArrayQuantifierFilter<ElementFilter>,
+): value is {
+	$some?: ElementFilter;
+	$every?: ElementFilter;
+	$none?: ElementFilter;
+} => {
+	if (typeof value !== "object" || value === null) return false;
+	return "$some" in value || "$every" in value || "$none" in value;
+};

shared/api/migrations/filters/match/index.ts

@@ -1,3 +1,4 @@
+export * from "./arrayFilterMatches.js";
 export * from "./nullableFieldMatches.js";
 export * from "./numberMatcherMatches.js";
 export * from "./stringMatcherMatches.js";

shared/index.ts

@@ -120,6 +120,7 @@ export * from "./models/genModels/processorSchemas";
 export * from "./models/migrationModels/migrationErrorTable";
 export * from "./models/migrationModels/migrationJobTable";
 export * from "./models/migrationModels/migrationModels";
+export * from "./models/migrationV2Models/migrationItemRunSchema";
 export * from "./models/migrationV2Models/migrationItemRunTable";
 export * from "./models/migrationV2Models/migrationRunTable";
 export * from "./models/migrationV2Models/migrationTable";

shared/models/cusModels/fullCusModel.ts

@@ -7,6 +7,10 @@ import {
 	FullCusProductSchema,
 } from "../cusProductModels/cusProductModels.js";
 import type { Event } from "../eventModels/eventTable.js";
+import {
+	type MigrationItemRunData,
+	MigrationItemRunSchema,
+} from "../migrationV2Models/migrationItemRunSchema.js";
 import type {
 	Schedule,
 	SchedulePhase,
@@ -56,6 +60,10 @@ export const FullCustomerSchema = CustomerSchema.extend({
 		.optional(),
 	invoices: z.array(InvoiceSchema).optional(),
 	schedule: FullCustomerScheduleSchema.optional(),
+	// `.default([])` makes the FullCustomer cache (which uses this schema for
+	// hole-filling via `normalizeFromSchema`) tolerant of entries written
+	// before this field existed. Empty array also matches the SQL default.
+	migration_item_runs: z.array(MigrationItemRunSchema).optional(),
 });
 
 export type FullCustomerSchedule = Schedule & { phases: SchedulePhase[] };
@@ -74,6 +82,7 @@ export type FullCustomer = Customer & {
 	events?: Event[];
 	extra_customer_entitlements: FullCustomerEntitlement[];
 	schedule?: FullCustomerSchedule;
+	migration_item_runs?: MigrationItemRunData[];
 };
 
 export const CustomerWithProductsSchema = CustomerSchema.extend({

shared/models/cusModels/fullSubject/fullSubjectModel.ts

@@ -2,6 +2,7 @@ import { z } from "zod/v4";
 import { FullAggregatedFeatureBalanceSchema } from "../../cusProductModels/cusEntModels/aggregatedCusEnt.js";
 import { FullCustomerEntitlementSchema } from "../../cusProductModels/cusEntModels/cusEntModels.js";
 import { FullCusProductSchema } from "../../cusProductModels/cusProductModels.js";
+import { MigrationItemRunSchema } from "../../migrationV2Models/migrationItemRunSchema.js";
 import { SubscriptionSchema } from "../../subModels/subModels.js";
 import { CustomerSchema } from "../cusModels.js";
 import { EntitySchema } from "../entityModels/entityModels.js";
@@ -38,6 +39,8 @@ export const FullSubjectSchema = z.object({
 	aggregated_subject_flags: z
 		.record(z.string(), AggregatedSubjectFlagSchema)
 		.optional(),
+
+	migration_item_runs: z.array(MigrationItemRunSchema).optional(),
 });
 
 export type FullSubject = z.infer<typeof FullSubjectSchema>;

shared/models/cusModels/fullSubject/normalizedFullSubjectModel.ts

@@ -18,6 +18,7 @@ import {
 	FeatureOptionsSchema,
 } from "../../cusProductModels/cusProductModels.js";
 import type { DbCustomerProduct } from "../../cusProductModels/cusProductTable.js";
+import type { MigrationItemRunData } from "../../migrationV2Models/migrationItemRunSchema.js";
 import type { EntitlementWithFeature } from "../../productModels/entModels/entModels.js";
 import type { DbFreeTrial } from "../../productModels/freeTrialModels/freeTrialTable.js";
 import type { DbPrice } from "../../productModels/priceModels/priceTable.js";
@@ -178,4 +179,9 @@ export type NormalizedFullSubject = {
 	invoices: Invoice[];
 
 	entity_aggregations?: EntityAggregations;
+
+	/** Latest 10 `migration_item_runs` for this customer, scoped to the
+	 *  org's active lazy/background migrations. Empty when no migrations
+	 *  are active for the org. */
+	migration_item_runs?: MigrationItemRunData[];
 };

shared/models/cusModels/fullSubject/subjectQueryRow.ts

@@ -5,6 +5,7 @@ import type { DbRollover } from "../../cusProductModels/cusEntModels/rolloverMod
 import type { DbCustomerPrice } from "../../cusProductModels/cusPriceModels/cusPriceTable.js";
 import type { DbCustomerProduct } from "../../cusProductModels/cusProductTable.js";
 import type { DbFeature } from "../../featureModels/featureTable.js";
+import type { MigrationItemRunData } from "../../migrationV2Models/migrationItemRunSchema.js";
 import type { DbEntitlement } from "../../productModels/entModels/entTable.js";
 import type { DbFreeTrial } from "../../productModels/freeTrialModels/freeTrialTable.js";
 import type { DbPrice } from "../../productModels/priceModels/priceTable.js";
@@ -38,4 +39,5 @@ export type SubjectQueryRow = {
 	subscriptions: Subscription[];
 	invoices?: Invoice[];
 	entity?: Entity;
+	migration_item_runs?: MigrationItemRunData[];
 };

shared/models/migrationV2Models/migrationItemRunSchema.ts

@@ -0,0 +1,25 @@
+import { z } from "zod/v4";
+
+/**
+ * Zod mirror of the `migration_item_runs` row shape. Used for FullSubject /
+ * FullCustomer cache hole-filling and as the canonical schema-derived type
+ * (`MigrationItemRunData`) for embedded `migration_item_runs` on those.
+ *
+ * Looser than the Drizzle table type by design — `timestamp` arrives as a
+ * string after `row_to_json` JSON serialization, and embedded item_runs only
+ * need shallow consumers (the lazy migration helper).
+ */
+export const MigrationItemRunSchema = z.object({
+	migration_item_run_id: z.string(),
+	migration_internal_id: z.string(),
+	migration_run_id: z.string().nullable(),
+	dry_run: z.boolean(),
+	item_kind: z.string(),
+	item_id: z.string(),
+	status: z.enum(["running", "succeeded", "skipped", "failed"]),
+	timestamp: z.union([z.string(), z.date()]).nullish(),
+	created_at: z.number(),
+	updated_at: z.number().nullable(),
+});
+
+export type MigrationItemRunData = z.infer<typeof MigrationItemRunSchema>;

shared/models/migrationV2Models/migrationRunTable.ts

@@ -53,9 +53,11 @@ export const migrationRuns = pgTable(
 			foreignColumns: [organizations.id],
 			name: "migration_runs_org_id_fkey",
 		}).onDelete("cascade"),
-		// Allows historical runs while enforcing one queued/running migration per org/env.
-		uniqueIndex("migration_runs_active_org_env_unique")
-			.on(table.org_id, table.env)
+		// One queued/running run per migration definition. Different migrations
+		// can be active concurrently — correctness lives at the per-customer
+		// `migration_item_runs_live_unique` claim.
+		uniqueIndex("migration_runs_active_per_migration_unique")
+			.on(table.migration_internal_id)
 			.where(sql`${table.status} IN ('queued', 'running')`),
 	],
 );

shared/utils/fullSubjectUtils/fullSubjectToFullCustomer.ts

@@ -13,4 +13,5 @@ export const fullSubjectToFullCustomer = ({
 	extra_customer_entitlements: fullSubject.extra_customer_entitlements,
 	subscriptions: fullSubject.subscriptions,
 	invoices: fullSubject.invoices,
+	migration_item_runs: fullSubject.migration_item_runs,
 });

shared/utils/fullSubjectUtils/normalizedToFullSubject.ts

@@ -397,5 +397,6 @@ export const normalizedToFullSubject = ({
 		...(aggregatedSubjectFlags
 			? { aggregated_subject_flags: aggregatedSubjectFlags }
 			: {}),
+		migration_item_runs: normalized.migration_item_runs ?? [],
 	} as FullSubject;
 };

trigger.config.ts

@@ -8,7 +8,7 @@ import { fetchInfisicalSecretsFromEnv } from "./server/src/external/infisical/fe
 
 export default defineConfig({
 	project: "proj_cwiutfmpdzfcshxevkok",
-	runtime: "node",
+	runtime: "bun",
 	logLevel: "log",
 	maxDuration: 3600,
 	retries: {