updated filter to have a pre-filter dsl plan step

server/src/internal/migrations/v2/filters/customers/buildCustomerSelect.ts

@@ -1,6 +1,6 @@
 import type { CustomerFilter, MigrationItemRunStatus } from "@autumn/shared";
-import { compileFilter } from "@autumn/shared/api/migrations/compiler/compileFilter.js";
 import type { ResolutionContext } from "@autumn/shared/api/migrations/compiler/filterToIr/resolutionContext.js";
+import { buildCustomerCandidateQuery } from "@autumn/shared/api/migrations/filters/planner/buildCustomerCandidateQuery.js";
 import { type SQL, sql } from "drizzle-orm";
 import { rawWithParamsToDrizzle } from "../rawWithParamsToDrizzle.js";
 
@@ -26,10 +26,22 @@ export type CustomerQueryArgs = {
 	search?: string;
 };
 
-const compileWhere = ({ orgId, env, filter, ctx }: CustomerQueryArgs): SQL =>
-	rawWithParamsToDrizzle(
-		compileFilter({ filter, ctx, ambient: { orgId, env } }),
-	);
+const compileCustomerCandidate = ({
+	orgId,
+	env,
+	filter,
+	ctx,
+}: CustomerQueryArgs): { source: SQL; where: SQL } => {
+	const candidate = buildCustomerCandidateQuery({
+		filter,
+		ctx,
+		ambient: { orgId, env },
+	});
+	return {
+		source: rawWithParamsToDrizzle(candidate.source),
+		where: rawWithParamsToDrizzle(candidate.where),
+	};
+};
 
 export type CustomerCheckpointExclusion = {
 	migrationInternalId: string;
@@ -204,12 +216,12 @@ export const buildCustomerSelect = ({
 	limit?: number;
 	afterInternalId?: string;
 }): SQL => {
-	const where = compileWhere({ orgId, env, filter, ctx });
+	const candidate = compileCustomerCandidate({ orgId, env, filter, ctx });
 	const limitClause = limit !== undefined ? sql`LIMIT ${limit}` : sql``;
 	return sql`
 		SELECT c.internal_id, c.id, c.name, c.email
-		FROM customers c
-		WHERE (${where}) ${buildCommonWhere({ checkpoint, search, afterInternalId })}
+		FROM ${candidate.source}
+		WHERE (${candidate.where}) ${buildCommonWhere({ checkpoint, search, afterInternalId })}
 		ORDER BY c.internal_id DESC
 		${limitClause}
 	`;
@@ -224,11 +236,11 @@ export const buildCustomerCount = ({
 	checkpoint,
 	search,
 }: CustomerQueryArgs): SQL => {
-	const where = compileWhere({ orgId, env, filter, ctx });
+	const candidate = compileCustomerCandidate({ orgId, env, filter, ctx });
 	return sql`
 		SELECT COUNT(*)::bigint AS count
-		FROM customers c
-		WHERE (${where}) ${buildCommonWhere({ checkpoint, search })}
+		FROM ${candidate.source}
+		WHERE (${candidate.where}) ${buildCommonWhere({ checkpoint, search })}
 	`;
 };
 
@@ -258,7 +270,7 @@ export const buildProcessedPreviewSelect = ({
 	limit?: number;
 	afterInternalId?: string;
 }): SQL => {
-	const where = compileWhere({ orgId, env, filter, ctx });
+	const candidate = compileCustomerCandidate({ orgId, env, filter, ctx });
 	const processed = buildProcessedIn(includeProcessed);
 	const limitClause = limit !== undefined ? sql`LIMIT ${limit}` : sql``;
 	const mode = getExecutionFilterMode(includeProcessed);
@@ -276,8 +288,8 @@ export const buildProcessedPreviewSelect = ({
 	if (mode === "not_run_only") {
 		return sql`
 			SELECT c.internal_id, c.id, c.name, c.email
-			FROM customers c
-			WHERE (${where}) ${buildCommonWhere({ checkpoint, search, afterInternalId, includeProcessed })}
+			FROM ${candidate.source}
+			WHERE (${candidate.where}) ${buildCommonWhere({ checkpoint, search, afterInternalId, includeProcessed })}
 			ORDER BY c.internal_id DESC
 			${limitClause}
 		`;
@@ -287,8 +299,8 @@ export const buildProcessedPreviewSelect = ({
 		SELECT u.internal_id, u.id, u.name, u.email
 		FROM (
 			SELECT c.internal_id, c.id, c.name, c.email
-			FROM customers c
-			WHERE (${where}) ${buildCommonWhere({ checkpoint, search, afterInternalId, includeProcessed })}
+			FROM ${candidate.source}
+			WHERE (${candidate.where}) ${buildCommonWhere({ checkpoint, search, afterInternalId, includeProcessed })}
 			UNION
 			SELECT c.internal_id, c.id, c.name, c.email
 			FROM customers c
@@ -308,7 +320,7 @@ export const buildProcessedPreviewCount = ({
 	search,
 	includeProcessed,
 }: ProcessedPreviewArgs): SQL => {
-	const where = compileWhere({ orgId, env, filter, ctx });
+	const candidate = compileCustomerCandidate({ orgId, env, filter, ctx });
 	const processed = buildProcessedIn(includeProcessed);
 	const mode = getExecutionFilterMode(includeProcessed);
 
@@ -323,8 +335,8 @@ export const buildProcessedPreviewCount = ({
 	if (mode === "not_run_only") {
 		return sql`
 			SELECT COUNT(*)::bigint AS count
-			FROM customers c
-			WHERE (${where}) ${buildCommonWhere({ checkpoint, search, includeProcessed })}
+			FROM ${candidate.source}
+			WHERE (${candidate.where}) ${buildCommonWhere({ checkpoint, search, includeProcessed })}
 		`;
 	}
 
@@ -332,8 +344,8 @@ export const buildProcessedPreviewCount = ({
 		SELECT COUNT(*)::bigint AS count
 		FROM (
 			SELECT c.internal_id
-			FROM customers c
-			WHERE (${where}) ${buildCommonWhere({ checkpoint, search, includeProcessed })}
+			FROM ${candidate.source}
+			WHERE (${candidate.where}) ${buildCommonWhere({ checkpoint, search, includeProcessed })}
 			UNION
 			SELECT c.internal_id
 			FROM customers c

server/tests/integration/billing/migrations-v2/filter-planner/customer-filter-planner-parity.test.ts

@@ -0,0 +1,727 @@
+/**
+ * TDD coverage for migration filter planning preserving customer selection.
+ *
+ * Red-failure mode (pre-planner guardrail):
+ *  - An optimized access path could return a narrower customer set than the
+ *    existing fallback compiler once wrapper filters are applied.
+ *
+ * Green-success criteria:
+ *  - Planned and fallback SQL return the same customers, and migration
+ *    wrappers (processed rows, checkpointing, search, cursoring) preserve
+ *    their existing semantics.
+ */
+
+import { expect, test } from "bun:test";
+import {
+	CusProductStatus,
+	customerProducts,
+	customers,
+	MigrationItemKind,
+	MigrationItemRunStatus,
+	migrationItemRuns,
+	migrations,
+	products as productsTable,
+	type CustomerFilter,
+} from "@autumn/shared";
+import { compileFilter } from "@autumn/shared/api/migrations/compiler/compileFilter.js";
+import chalk from "chalk";
+import { sql, type SQL } from "drizzle-orm";
+import {
+	buildCustomerCount,
+	buildCustomerSelect,
+	buildProcessedPreviewCount,
+	buildProcessedPreviewSelect,
+	type CustomerQueryArgs,
+	type IncludeProcessed,
+} from "@/internal/migrations/v2/filters/customers/buildCustomerSelect.js";
+import { rawWithParamsToDrizzle } from "@/internal/migrations/v2/filters/rawWithParamsToDrizzle.js";
+import { initScenario } from "@tests/utils/testInitUtils/initScenario.js";
+
+const CREATED_AT = 1_780_000_000_000;
+const sorted = (values: string[]) => [...values].sort();
+
+type TestCtx = Awaited<ReturnType<typeof initScenario>>["ctx"];
+type TestDb = TestCtx["db"];
+
+type SeededFixture = {
+	ctx: TestCtx;
+	prefix: string;
+	migrationInternalId: string;
+	migrationRunId: string;
+	otherDryRunId: string;
+	customerIds: {
+		active: string;
+		scheduled: string;
+		pastDue: string;
+		duplicateProducts: string;
+		expired: string;
+		pro: string;
+		otherEnv: string;
+	};
+	args: CustomerQueryArgs;
+};
+
+const executeCustomerIds = async ({
+	db,
+	query,
+}: {
+	db: TestDb;
+	query: SQL;
+}) => {
+	const rows = (await db.execute(query)) as Array<{ id: string }>;
+	return rows.map((row) => row.id);
+};
+
+const executeCount = async ({ db, query }: { db: TestDb; query: SQL }) => {
+	const [{ count }] = (await db.execute(query)) as Array<{
+		count: bigint | number;
+	}>;
+	return Number(count);
+};
+
+const cleanupSeededRows = async ({
+	db,
+	prefix,
+}: {
+	db: TestDb;
+	prefix: string;
+}) => {
+	const pattern = `${prefix}-%`;
+	await db.execute(
+		sql`DELETE FROM migration_item_runs WHERE migration_internal_id LIKE ${pattern}`,
+	);
+	await db.execute(sql`DELETE FROM migrations WHERE internal_id LIKE ${pattern}`);
+	await db.execute(sql`DELETE FROM customer_products WHERE id LIKE ${pattern}`);
+	await db.execute(sql`DELETE FROM customers WHERE internal_id LIKE ${pattern}`);
+	await db.execute(sql`DELETE FROM products WHERE internal_id LIKE ${pattern}`);
+};
+
+const buildFallbackCustomerSelect = ({
+	orgId,
+	env,
+	filter,
+	ctx,
+}: CustomerQueryArgs): SQL => {
+	const where = rawWithParamsToDrizzle(
+		compileFilter({ filter, ctx, ambient: { orgId, env } }),
+	);
+	return sql`
+		SELECT c.internal_id, c.id, c.name, c.email
+		FROM customers c
+		WHERE (${where})
+		ORDER BY c.internal_id DESC
+	`;
+};
+
+const seedPlannerFixture = async (prefix: string): Promise<SeededFixture> => {
+	const targetPlanId = `${prefix}-enterprise`;
+	const otherPlanId = `${prefix}-pro`;
+	const otherEnv = "live";
+	const { ctx } = await initScenario({ setup: [], actions: [] });
+
+	const customerIds = {
+		active: `${prefix}-active`,
+		scheduled: `${prefix}-scheduled`,
+		pastDue: `${prefix}-past-due`,
+		duplicateProducts: `${prefix}-duplicate-products`,
+		expired: `${prefix}-expired`,
+		pro: `${prefix}-pro`,
+		otherEnv: `${prefix}-other-env`,
+	};
+
+	await cleanupSeededRows({ db: ctx.db, prefix });
+	await ctx.db.insert(productsTable).values([
+		{
+			internal_id: `${prefix}-prod-enterprise-v1`,
+			id: targetPlanId,
+			name: "Enterprise v1",
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+			version: 1,
+		},
+		{
+			internal_id: `${prefix}-prod-enterprise-v2`,
+			id: targetPlanId,
+			name: "Enterprise v2",
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+			version: 2,
+		},
+		{
+			internal_id: `${prefix}-prod-pro`,
+			id: otherPlanId,
+			name: "Pro",
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+			version: 1,
+		},
+		{
+			internal_id: `${prefix}-prod-enterprise-other-env`,
+			id: targetPlanId,
+			name: "Enterprise other env",
+			org_id: ctx.org.id,
+			env: otherEnv,
+			created_at: CREATED_AT,
+			version: 1,
+		},
+	]);
+	await ctx.db.insert(customers).values([
+		{
+			internal_id: `${prefix}-cus-active`,
+			id: customerIds.active,
+			name: "Alpha Active Enterprise",
+			email: `${prefix}-active@example.com`,
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+		},
+		{
+			internal_id: `${prefix}-cus-scheduled`,
+			id: customerIds.scheduled,
+			name: "Bravo Scheduled Enterprise",
+			email: `${prefix}-scheduled@example.com`,
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+		},
+		{
+			internal_id: `${prefix}-cus-past-due`,
+			id: customerIds.pastDue,
+			name: "Charlie Past Due Enterprise",
+			email: `${prefix}-past-due@example.com`,
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+		},
+		{
+			internal_id: `${prefix}-cus-duplicate-products`,
+			id: customerIds.duplicateProducts,
+			name: "Delta Duplicate Enterprise Products",
+			email: `${prefix}-duplicate-products@example.com`,
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+		},
+		{
+			internal_id: `${prefix}-cus-expired`,
+			id: customerIds.expired,
+			name: "Echo Expired Enterprise",
+			email: `${prefix}-expired@example.com`,
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+		},
+		{
+			internal_id: `${prefix}-cus-pro`,
+			id: customerIds.pro,
+			name: "Foxtrot Pro",
+			email: `${prefix}-pro@example.com`,
+			org_id: ctx.org.id,
+			env: ctx.env,
+			created_at: CREATED_AT,
+		},
+		{
+			internal_id: `${prefix}-cus-other-env`,
+			id: customerIds.otherEnv,
+			name: "Golf Other Env Enterprise",
+			email: `${prefix}-other-env@example.com`,
+			org_id: ctx.org.id,
+			env: otherEnv,
+			created_at: CREATED_AT,
+		},
+	]);
+	await ctx.db.insert(customerProducts).values([
+		{
+			id: `${prefix}-cp-active`,
+			internal_customer_id: `${prefix}-cus-active`,
+			internal_product_id: `${prefix}-prod-enterprise-v1`,
+			product_id: targetPlanId,
+			status: CusProductStatus.Active,
+		},
+		{
+			id: `${prefix}-cp-scheduled`,
+			internal_customer_id: `${prefix}-cus-scheduled`,
+			internal_product_id: `${prefix}-prod-enterprise-v1`,
+			product_id: targetPlanId,
+			status: CusProductStatus.Scheduled,
+		},
+		{
+			id: `${prefix}-cp-past-due`,
+			internal_customer_id: `${prefix}-cus-past-due`,
+			internal_product_id: `${prefix}-prod-enterprise-v1`,
+			product_id: targetPlanId,
+			status: CusProductStatus.PastDue,
+		},
+		{
+			id: `${prefix}-cp-duplicate-v1`,
+			internal_customer_id: `${prefix}-cus-duplicate-products`,
+			internal_product_id: `${prefix}-prod-enterprise-v1`,
+			product_id: targetPlanId,
+			status: CusProductStatus.Active,
+		},
+		{
+			id: `${prefix}-cp-duplicate-v2`,
+			internal_customer_id: `${prefix}-cus-duplicate-products`,
+			internal_product_id: `${prefix}-prod-enterprise-v2`,
+			product_id: targetPlanId,
+			status: CusProductStatus.Scheduled,
+		},
+		{
+			id: `${prefix}-cp-expired`,
+			internal_customer_id: `${prefix}-cus-expired`,
+			internal_product_id: `${prefix}-prod-enterprise-v1`,
+			product_id: targetPlanId,
+			status: CusProductStatus.Expired,
+		},
+		{
+			id: `${prefix}-cp-pro`,
+			internal_customer_id: `${prefix}-cus-pro`,
+			internal_product_id: `${prefix}-prod-pro`,
+			product_id: otherPlanId,
+			status: CusProductStatus.Active,
+		},
+		{
+			id: `${prefix}-cp-other-env`,
+			internal_customer_id: `${prefix}-cus-other-env`,
+			internal_product_id: `${prefix}-prod-enterprise-other-env`,
+			product_id: targetPlanId,
+			status: CusProductStatus.Active,
+		},
+	]);
+
+	const migrationInternalId = `${prefix}-migration`;
+	const migrationRunId = `${prefix}-run`;
+	await ctx.db.insert(migrations).values({
+		internal_id: migrationInternalId,
+		id: `${prefix}-migration`,
+		org_id: ctx.org.id,
+		env: ctx.env,
+		filter: { customer: { plan: { plan_id: targetPlanId } } },
+		created_at: CREATED_AT,
+	});
+
+	return {
+		ctx,
+		prefix,
+		migrationInternalId,
+		migrationRunId,
+		otherDryRunId: `${prefix}-other-dry-run`,
+		customerIds,
+		args: {
+			orgId: ctx.org.id,
+			env: ctx.env,
+			filter: { plan: { plan_id: targetPlanId } },
+			ctx: { features: ctx.features },
+		},
+	};
+};
+
+const withSeededFixture = async (
+	prefix: string,
+	run: (fixture: SeededFixture) => Promise<void>,
+) => {
+	const fixture = await seedPlannerFixture(prefix);
+	try {
+		await run(fixture);
+	} finally {
+		await cleanupSeededRows({ db: fixture.ctx.db, prefix });
+	}
+};
+
+const insertItemRun = async ({
+	db,
+	migrationInternalId,
+	migrationRunId,
+	itemId,
+	status,
+	dryRun = false,
+}: {
+	db: TestDb;
+	migrationInternalId: string;
+	migrationRunId: string;
+	itemId: string;
+	status: MigrationItemRunStatus;
+	dryRun?: boolean;
+}) => {
+	await db.insert(migrationItemRuns).values({
+		migration_item_run_id: `${migrationInternalId}-${migrationRunId}-${itemId}-${status}-${dryRun ? "dry" : "live"}`,
+		migration_internal_id: migrationInternalId,
+		migration_run_id: migrationRunId,
+		dry_run: dryRun,
+		item_kind: MigrationItemKind.Customer,
+		item_id: itemId,
+		status,
+		created_at: CREATED_AT,
+		updated_at: CREATED_AT,
+	});
+};
+
+const includeProcessed = (
+	fixture: SeededFixture,
+	executionFilter?: IncludeProcessed["executionFilter"],
+): IncludeProcessed => ({
+	migrationInternalId: fixture.migrationInternalId,
+	executionFilter,
+});
+
+test(`${chalk.yellowBright("migration filter planner: plan_id access path matches fallback customer set")}`, async () => {
+	await withSeededFixture("planner-parity-base", async (fixture) => {
+		const plannedIds = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect(fixture.args),
+		});
+		const fallbackIds = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildFallbackCustomerSelect(fixture.args),
+		});
+
+		expect(sorted(plannedIds)).toEqual(sorted(fallbackIds));
+		expect(sorted(plannedIds)).toEqual(
+			sorted([
+				fixture.customerIds.active,
+				fixture.customerIds.scheduled,
+				fixture.customerIds.pastDue,
+				fixture.customerIds.duplicateProducts,
+			]),
+		);
+		expect(new Set(plannedIds).size).toBe(plannedIds.length);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: includeProcessed unions stale processed rows once")}`, async () => {
+	await withSeededFixture("planner-parity-processed-union", async (fixture) => {
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-pro`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-active`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+
+		const ids = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildProcessedPreviewSelect({
+				...fixture.args,
+				includeProcessed: includeProcessed(fixture),
+			}),
+		});
+		const count = await executeCount({
+			db: fixture.ctx.db,
+			query: buildProcessedPreviewCount({
+				...fixture.args,
+				includeProcessed: includeProcessed(fixture),
+			}),
+		});
+
+		expect(sorted(ids)).toEqual(
+			sorted([
+				fixture.customerIds.active,
+				fixture.customerIds.scheduled,
+				fixture.customerIds.pastDue,
+				fixture.customerIds.duplicateProducts,
+				fixture.customerIds.pro,
+			]),
+		);
+		expect(new Set(ids).size).toBe(ids.length);
+		expect(count).toBe(5);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: explicit processed statuses ignore current filter")}`, async () => {
+	await withSeededFixture("planner-parity-explicit-status", async (fixture) => {
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-active`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-pro`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-scheduled`,
+			status: MigrationItemRunStatus.Failed,
+		});
+
+		const ids = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildProcessedPreviewSelect({
+				...fixture.args,
+				includeProcessed: includeProcessed(fixture, {
+					statuses: [MigrationItemRunStatus.Succeeded],
+				}),
+			}),
+		});
+
+		expect(sorted(ids)).toEqual(
+			sorted([fixture.customerIds.active, fixture.customerIds.pro]),
+		);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: not_run excludes any processed customer")}`, async () => {
+	await withSeededFixture("planner-parity-not-run", async (fixture) => {
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-active`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-scheduled`,
+			status: MigrationItemRunStatus.Failed,
+		});
+
+		const ids = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildProcessedPreviewSelect({
+				...fixture.args,
+				includeProcessed: includeProcessed(fixture, { statuses: ["not_run"] }),
+			}),
+		});
+
+		expect(sorted(ids)).toEqual(
+			sorted([
+				fixture.customerIds.pastDue,
+				fixture.customerIds.duplicateProducts,
+			]),
+		);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: mixed statuses include succeeded stale rows and matching not-run rows")}`, async () => {
+	await withSeededFixture("planner-parity-mixed-status", async (fixture) => {
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-pro`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-scheduled`,
+			status: MigrationItemRunStatus.Failed,
+		});
+
+		const ids = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildProcessedPreviewSelect({
+				...fixture.args,
+				includeProcessed: includeProcessed(fixture, {
+					statuses: [MigrationItemRunStatus.Succeeded, "not_run"],
+				}),
+			}),
+		});
+
+		expect(sorted(ids)).toEqual(
+			sorted([
+				fixture.customerIds.active,
+				fixture.customerIds.pastDue,
+				fixture.customerIds.duplicateProducts,
+				fixture.customerIds.pro,
+			]),
+		);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: checkpoint excludes completed items from run selection")}`, async () => {
+	await withSeededFixture("planner-parity-checkpoint", async (fixture) => {
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-active`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-scheduled`,
+			status: MigrationItemRunStatus.Failed,
+		});
+
+		const idsWithoutRetry = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({
+				...fixture.args,
+				checkpoint: {
+					migrationInternalId: fixture.migrationInternalId,
+					migrationRunId: fixture.migrationRunId,
+					dryRun: false,
+					excludedStatuses: [
+						MigrationItemRunStatus.Running,
+						MigrationItemRunStatus.Succeeded,
+						MigrationItemRunStatus.Skipped,
+						MigrationItemRunStatus.Failed,
+					],
+				},
+			}),
+		});
+		const idsWithRetry = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({
+				...fixture.args,
+				checkpoint: {
+					migrationInternalId: fixture.migrationInternalId,
+					migrationRunId: fixture.migrationRunId,
+					dryRun: false,
+					excludedStatuses: [
+						MigrationItemRunStatus.Running,
+						MigrationItemRunStatus.Succeeded,
+						MigrationItemRunStatus.Skipped,
+					],
+				},
+			}),
+		});
+
+		expect(sorted(idsWithoutRetry)).toEqual(
+			sorted([
+				fixture.customerIds.pastDue,
+				fixture.customerIds.duplicateProducts,
+			]),
+		);
+		expect(sorted(idsWithRetry)).toEqual(
+			sorted([
+				fixture.customerIds.scheduled,
+				fixture.customerIds.pastDue,
+				fixture.customerIds.duplicateProducts,
+			]),
+		);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: dry-run checkpoint scopes same run differently from other dry runs")}`, async () => {
+	await withSeededFixture("planner-parity-dry-checkpoint", async (fixture) => {
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-active`,
+			status: MigrationItemRunStatus.Succeeded,
+			dryRun: true,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.otherDryRunId,
+			itemId: `${fixture.prefix}-cus-scheduled`,
+			status: MigrationItemRunStatus.Succeeded,
+			dryRun: true,
+		});
+		await insertItemRun({
+			db: fixture.ctx.db,
+			migrationInternalId: fixture.migrationInternalId,
+			migrationRunId: fixture.migrationRunId,
+			itemId: `${fixture.prefix}-cus-past-due`,
+			status: MigrationItemRunStatus.Succeeded,
+		});
+
+		const ids = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({
+				...fixture.args,
+				checkpoint: {
+					migrationInternalId: fixture.migrationInternalId,
+					migrationRunId: fixture.migrationRunId,
+					dryRun: true,
+					excludedStatuses: [MigrationItemRunStatus.Succeeded],
+				},
+			}),
+		});
+
+		expect(sorted(ids)).toEqual(
+			sorted([
+				fixture.customerIds.scheduled,
+				fixture.customerIds.duplicateProducts,
+			]),
+		);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: search and customer_id narrowing remain residual filters")}`, async () => {
+	await withSeededFixture("planner-parity-search-only", async (fixture) => {
+		const searchIds = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({
+				...fixture.args,
+				search: "scheduled@example.com",
+			}),
+		});
+		const onlyIds = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({
+				...fixture.args,
+				filter: {
+					...fixture.args.filter,
+					customer_id: { $in: [fixture.customerIds.pastDue] },
+				},
+			}),
+		});
+
+		expect(searchIds).toEqual([fixture.customerIds.scheduled]);
+		expect(onlyIds).toEqual([fixture.customerIds.pastDue]);
+	});
+});
+
+test(`${chalk.yellowBright("migration filter planner: cursor pagination is stable and complete")}`, async () => {
+	await withSeededFixture("planner-parity-pagination", async (fixture) => {
+		const firstPage = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({ ...fixture.args, limit: 2 }),
+		});
+		const secondPage = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect({
+				...fixture.args,
+				limit: 10,
+				afterInternalId: `${fixture.prefix}-cus-past-due`,
+			}),
+		});
+		const allIds = await executeCustomerIds({
+			db: fixture.ctx.db,
+			query: buildCustomerSelect(fixture.args),
+		});
+
+		expect(firstPage).toEqual([
+			fixture.customerIds.scheduled,
+			fixture.customerIds.pastDue,
+		]);
+		expect(secondPage).toEqual([
+			fixture.customerIds.duplicateProducts,
+			fixture.customerIds.active,
+		]);
+		expect(sorted([...firstPage, ...secondPage])).toEqual(sorted(allIds));
+		expect(
+			await executeCount({
+				db: fixture.ctx.db,
+				query: buildCustomerCount(fixture.args),
+			}),
+		).toBe(4);
+	});
+});

server/tests/unit/compiler/customer/basic.test.ts

@@ -11,8 +11,8 @@ const ctx = contexts.create({ features });
 const ambient = { orgId: "org_test", env: "live" };
 
 const ROOT_AMBIENT = "c.org_id = ? AND c.env = ?";
-const PLAN_AMBIENT = "cp.status IN (?, ?)";
-const PLAN_AMBIENT_PARAMS = ["active", "past_due"];
+const PLAN_AMBIENT = "cp.status IN (?, ?, ?)";
+const PLAN_AMBIENT_PARAMS = ["active", "past_due", "scheduled"];
 
 const normalize = (sql: string) =>
 	sql.replace(/\s+/g, " ").replace(/\(\s+/g, "(").replace(/\s+\)/g, ")").trim();

server/tests/unit/compiler/customer/derived-and-or.test.ts

@@ -11,8 +11,8 @@ const ctx = contexts.create({ features });
 const ambient = { orgId: "org_test", env: "live" };
 
 const ROOT_AMBIENT = "c.org_id = ? AND c.env = ?";
-const PLAN_AMBIENT = "cp.status IN (?, ?)";
-const PLAN_AMBIENT_PARAMS = ["active", "past_due"];
+const PLAN_AMBIENT = "cp.status IN (?, ?, ?)";
+const PLAN_AMBIENT_PARAMS = ["active", "past_due", "scheduled"];
 
 const BASE_PRICE_EXISTS = [
 	"(SELECT base_cpr.id FROM customer_prices base_cpr",

server/tests/unit/compiler/customer/nested-item.test.ts

@@ -11,8 +11,8 @@ const ctx = contexts.create({ features });
 const ambient = { orgId: "org_test", env: "live" };
 
 const ROOT_AMBIENT = "c.org_id = ? AND c.env = ?";
-const PLAN_AMBIENT = "cp.status IN (?, ?)";
-const PLAN_AMBIENT_PARAMS = ["active", "past_due"];
+const PLAN_AMBIENT = "cp.status IN (?, ?, ?)";
+const PLAN_AMBIENT_PARAMS = ["active", "past_due", "scheduled"];
 
 const ITEM_FROM = [
 	"customer_entitlements ce",

server/tests/unit/compiler/customer/planner.test.ts

@@ -0,0 +1,253 @@
+import { describe, expect, test } from "bun:test";
+import type { Feature } from "@autumn/shared";
+import { compileFilter } from "@autumn/shared/api/migrations/compiler/compileFilter.js";
+import { buildCustomerCandidateQuery } from "@autumn/shared/api/migrations/filters/planner/buildCustomerCandidateQuery.js";
+import type { CustomerFilter } from "@autumn/shared/api/migrations/filters/customerFilter.js";
+import { contexts } from "@tests/utils/fixtures/db/contexts";
+
+const features: Feature[] = [
+	{ id: "credits", internal_id: "fea_credits_internal" } as Feature,
+];
+
+const ctx = contexts.create({ features });
+const ambient = { orgId: "org_test", env: "live" };
+const RELEVANT_STATUS_PARAMS = ["active", "past_due", "scheduled"];
+
+const normalize = (sql: string) =>
+	sql.replace(/\s+/g, " ").replace(/\(\s+/g, "(").replace(/\s+\)/g, ")").trim();
+
+const buildCandidate = (filter: CustomerFilter) =>
+	buildCustomerCandidateQuery({
+		filter,
+		ctx: { features: ctx.features },
+		ambient,
+	});
+
+const expectFallbackWhereParity = (filter: CustomerFilter) => {
+	const candidate = buildCandidate(filter);
+	const fallback = compileFilter({
+		filter,
+		ctx: { features: ctx.features },
+		ambient,
+	});
+
+	expect(normalize(candidate.where.sql)).toBe(normalize(fallback.sql));
+	expect(candidate.where.params).toEqual(fallback.params);
+	return candidate;
+};
+
+describe("customer filter planner", () => {
+	test("plan.plan_id eq uses a products-driven candidate source", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: { plan_id: "enterprise" },
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).toBe(
+			normalize(`
+				(WITH plan_products AS MATERIALIZED (
+				SELECT p.internal_id FROM products p
+				WHERE p.org_id = ? AND p.env = ?
+				AND p.id = ?
+				) SELECT DISTINCT c.internal_id, c.id, c.name, c.email, c.org_id, c.env
+				FROM plan_products pp
+				JOIN customer_products cp ON cp.internal_product_id = pp.internal_id
+				JOIN customers c ON c.internal_id = cp.internal_customer_id
+				WHERE cp.status IN (?, ?, ?)
+					AND c.org_id = ?
+					AND c.env = ?) c
+			`),
+		);
+		expect(candidate.source.params).toEqual([
+			"org_test",
+			"live",
+			"enterprise",
+			...RELEVANT_STATUS_PARAMS,
+			"org_test",
+			"live",
+		]);
+	});
+
+	test("plan.plan_id in uses the same candidate path", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: { plan_id: { $in: ["enterprise", "pro"] } },
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).toContain("p.id IN (?, ?)");
+		expect(candidate.source.params).toEqual([
+			"org_test",
+			"live",
+			"enterprise",
+			"pro",
+			...RELEVANT_STATUS_PARAMS,
+			"org_test",
+			"live",
+		]);
+	});
+
+	test("compound filters use plan_id as a candidate and keep fallback semantics", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				plan_id: "enterprise",
+				item: { feature_id: "credits" },
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).toContain("p.id = ?");
+		expect(normalize(candidate.where.sql)).toContain("e.internal_feature_id = ?");
+	});
+
+	test("plan_id + version keeps version as a residual fallback predicate", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				plan_id: "enterprise",
+				version: 2,
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).toContain("p.id = ?");
+		expect(normalize(candidate.source.sql)).not.toContain("p.version = ?");
+		expect(normalize(candidate.where.sql)).toContain(
+			"(p.id = ? AND p.version = ?)",
+		);
+		expect(candidate.where.params).toEqual([
+			"org_test",
+			"live",
+			...RELEVANT_STATUS_PARAMS,
+			"enterprise",
+			2,
+		]);
+	});
+
+	test("plan_id + custom keeps customer-product custom state as a residual predicate", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				plan_id: "enterprise",
+				custom: false,
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).not.toContain("cp.is_custom = ?");
+		expect(normalize(candidate.where.sql)).toContain(
+			"(p.id = ? AND cp.is_custom = ?)",
+		);
+		expect(candidate.where.params).toEqual([
+			"org_test",
+			"live",
+			...RELEVANT_STATUS_PARAMS,
+			"enterprise",
+			false,
+		]);
+	});
+
+	test("plan_id + price keeps base-price existence as a residual predicate", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				plan_id: "enterprise",
+				price: { $ne: null },
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).not.toContain("base_cpr.id");
+		expect(normalize(candidate.where.sql)).toContain("base_cpr.id");
+		expect(normalize(candidate.where.sql)).toContain("IS NOT NULL");
+	});
+
+	test("plan_id + paid/recurring derived filters remain residual predicates", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				plan_id: "enterprise",
+				paid: true,
+				recurring: true,
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).not.toContain("customer_prices");
+		expect(normalize(candidate.where.sql)).toContain("customer_prices cpr");
+		expect(normalize(candidate.where.sql)).toContain(
+			"pr.config->>'interval' <> 'one_off'",
+		);
+	});
+
+	test("plan_id + item rollover keeps entitlement rollover as a residual predicate", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				plan_id: "enterprise",
+				item: { rollover: { $ne: null } },
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({
+			kind: "planned",
+			id: "plan.plan_id",
+		});
+		expect(normalize(candidate.source.sql)).not.toContain("e.rollover");
+		expect(normalize(candidate.where.sql)).toContain("e.rollover IS NOT NULL");
+	});
+
+	test("top-level item rollover falls back until an entitlement access path exists", () => {
+		const candidate = expectFallbackWhereParity({
+			item: { rollover: { $ne: null } },
+		});
+
+		expect(candidate.accessPath).toEqual({ kind: "fallback" });
+		expect(normalize(candidate.source.sql)).toBe("customers c");
+		expect(normalize(candidate.where.sql)).toContain("e.rollover IS NOT NULL");
+	});
+
+	test("plan_id inside an OR falls back to avoid dropping other branches", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: {
+				$or: [{ plan_id: "enterprise" }, { paid: true }],
+			},
+		});
+
+		expect(candidate.accessPath).toEqual({ kind: "fallback" });
+		expect(normalize(candidate.source.sql)).toBe("customers c");
+	});
+
+	test("negative plan quantifiers fall back", () => {
+		const candidate = expectFallbackWhereParity({
+			plan: { $none: { plan_id: "enterprise" } },
+		});
+
+		expect(candidate.accessPath).toEqual({ kind: "fallback" });
+		expect(normalize(candidate.source.sql)).toBe("customers c");
+	});
+
+	test("direct customer filters remain customer-rooted", () => {
+		const candidate = expectFallbackWhereParity({
+			customer_id: "cus_123",
+		});
+
+		expect(candidate.accessPath).toEqual({ kind: "fallback" });
+		expect(normalize(candidate.source.sql)).toBe("customers c");
+	});
+});

server/tests/unit/compiler/customer/rollover-existence.test.ts

@@ -11,8 +11,8 @@ const ctx = contexts.create({ features });
 const ambient = { orgId: "org_test", env: "live" };
 
 const ROOT_AMBIENT = "c.org_id = ? AND c.env = ?";
-const PLAN_AMBIENT = "cp.status IN (?, ?)";
-const PLAN_AMBIENT_PARAMS = ["active", "past_due"];
+const PLAN_AMBIENT = "cp.status IN (?, ?, ?)";
+const PLAN_AMBIENT_PARAMS = ["active", "past_due", "scheduled"];
 
 const ITEM_FROM = [
 	"customer_entitlements ce",

server/tests/unit/compiler/plan/version.test.ts

@@ -20,8 +20,8 @@ const ctx = contexts.create({ features: [] });
 const ambient = { orgId: "org_test", env: "live" };
 
 const ROOT_AMBIENT = "c.org_id = ? AND c.env = ?";
-const PLAN_AMBIENT = "cp.status IN (?, ?)";
-const PLAN_AMBIENT_PARAMS = ["active", "past_due"];
+const PLAN_AMBIENT = "cp.status IN (?, ?, ?)";
+const PLAN_AMBIENT_PARAMS = ["active", "past_due", "scheduled"];
 const PLAN_ROOT_AMBIENT = "p.org_id = ? AND p.env = ?";
 
 const normalize = (sql: string) =>

server/tests/unit/migrations-v2/compiler/customer-select-planner.test.ts

@@ -0,0 +1,62 @@
+import { describe, expect, test } from "bun:test";
+import {
+	buildCustomerCount,
+	buildCustomerSelect,
+} from "@/internal/migrations/v2/filters/customers/buildCustomerSelect.js";
+import { PgDialect } from "drizzle-orm/pg-core";
+
+const dialect = new PgDialect();
+const ctx = { features: [] };
+
+const normalize = (sql: string) => sql.replace(/\s+/g, " ").trim();
+
+describe("migration customer select planner wiring", () => {
+	test("plan_id filters use a planned customer source plus fallback predicate", () => {
+		const query = buildCustomerCount({
+			orgId: "org_test",
+			env: "live",
+			filter: { plan: { plan_id: "enterprise" } },
+			ctx,
+		});
+		const { sql, params } = dialect.sqlToQuery(query);
+
+		expect(normalize(sql)).toContain(
+			"FROM (WITH plan_products AS MATERIALIZED",
+		);
+		expect(normalize(sql)).toContain("SELECT p.internal_id FROM products p");
+		expect(normalize(sql)).toContain(
+			"AND EXISTS (SELECT 1 FROM customer_products cp JOIN products p",
+		);
+		expect(params).toEqual([
+			"org_test",
+			"live",
+			"enterprise",
+			"active",
+			"past_due",
+			"scheduled",
+			"org_test",
+			"live",
+			"org_test",
+			"live",
+			"active",
+			"past_due",
+			"scheduled",
+			"enterprise",
+		]);
+	});
+
+	test("non-planned filters keep the customer root source", () => {
+		const query = buildCustomerSelect({
+			orgId: "org_test",
+			env: "live",
+			filter: { customer_id: "cus_123" },
+			ctx,
+			limit: 10,
+		});
+		const { sql, params } = dialect.sqlToQuery(query);
+
+		expect(normalize(sql)).toContain("FROM customers c");
+		expect(normalize(sql)).not.toContain("FROM (SELECT DISTINCT");
+		expect(params).toEqual(["org_test", "live", "cus_123", 10]);
+	});
+});

shared/api/migrations/filters/planner/accessPaths/planPlanIdAccessPath.ts

@@ -0,0 +1,59 @@
+import { RELEVANT_STATUSES } from "../../../../../utils/cusProductUtils/cusProductConstants.js";
+import type { IRLeaf } from "../../../compiler/ir/irTypes.js";
+import type { CustomerAccessPath } from "../types.js";
+
+export type PlanIdConstraint = Pick<IRLeaf, "op" | "value"> & {
+	field: "plan_id";
+	op: "eq" | "in";
+};
+
+export const planPlanIdAccessPath: CustomerAccessPath<PlanIdConstraint> = {
+	id: "plan.plan_id",
+	buildSource: ({ constraint, ambient }) => {
+		const params: unknown[] = [];
+		const orgId = ambient.orgId;
+		const env = ambient.env;
+		if (orgId === undefined) throw new Error("Missing ambient orgId");
+		if (env === undefined) throw new Error("Missing ambient env");
+
+		params.push(orgId, env);
+		const planPredicate =
+			constraint.op === "eq"
+				? buildEqPredicate(constraint.value, params)
+				: buildInPredicate(constraint.value, params);
+		params.push(...RELEVANT_STATUSES, orgId, env);
+		const statusPlaceholders = RELEVANT_STATUSES.map(() => "?").join(", ");
+
+		return {
+			sql: [
+				"(WITH plan_products AS MATERIALIZED (",
+				"SELECT p.internal_id FROM products p",
+				"WHERE p.org_id = ? AND p.env = ?",
+				`AND ${planPredicate}`,
+				") SELECT DISTINCT c.internal_id, c.id, c.name, c.email, c.org_id, c.env",
+				"FROM plan_products pp",
+				"JOIN customer_products cp ON cp.internal_product_id = pp.internal_id",
+				"JOIN customers c ON c.internal_id = cp.internal_customer_id",
+				`WHERE cp.status IN (${statusPlaceholders})`,
+				"AND c.org_id = ?",
+				"AND c.env = ?) c",
+			].join(" "),
+			params,
+		};
+	},
+};
+
+const buildEqPredicate = (value: PlanIdConstraint["value"], params: unknown[]) => {
+	if (typeof value !== "string")
+		throw new Error("plan.plan_id eq access path requires a string value");
+	params.push(value);
+	return "p.id = ?";
+};
+
+const buildInPredicate = (value: PlanIdConstraint["value"], params: unknown[]) => {
+	if (!Array.isArray(value) || value.some((v) => typeof v !== "string"))
+		throw new Error("plan.plan_id in access path requires string values");
+	if (value.length === 0) return "FALSE";
+	params.push(...value);
+	return `p.id IN (${value.map(() => "?").join(", ")})`;
+};

shared/api/migrations/filters/planner/buildCustomerCandidateQuery.ts

@@ -0,0 +1,50 @@
+import type { CustomerFilter } from "../customerFilter.js";
+import { filterToIr } from "../../compiler/filterToIr/filterToIr.js";
+import type { ResolutionContext } from "../../compiler/filterToIr/resolutionContext.js";
+import {
+	type AmbientContext,
+	irToSql,
+} from "../../compiler/irToSql/irToSql.js";
+import { customerRegistry } from "../../compiler/registry/customerRegistry.js";
+import { planPlanIdAccessPath } from "./accessPaths/planPlanIdAccessPath.js";
+import { chooseCustomerAccessPath } from "./chooseCustomerAccessPath.js";
+import type { CustomerCandidateQuery } from "./types.js";
+
+export const buildCustomerCandidateQuery = ({
+	filter,
+	ctx,
+	ambient,
+}: {
+	filter: CustomerFilter;
+	ctx: ResolutionContext;
+	ambient: AmbientContext;
+}): CustomerCandidateQuery => {
+	const ir = filterToIr({ filter, ctx });
+	const fallbackWhere = irToSql({ ir, root: customerRegistry, ambient });
+	const accessPath = chooseCustomerAccessPath(ir);
+
+	if (!accessPath) {
+		return {
+			source: { sql: "customers c", params: [] },
+			where: fallbackWhere,
+			accessPath: { kind: "fallback" },
+		};
+	}
+
+	if (accessPath.id === "plan.plan_id") {
+		return {
+			source: planPlanIdAccessPath.buildSource({
+				constraint: accessPath.constraint,
+				ambient,
+			}),
+			where: fallbackWhere,
+			accessPath: { kind: "planned", id: accessPath.id },
+		};
+	}
+
+	return {
+		source: { sql: "customers c", params: [] },
+		where: fallbackWhere,
+		accessPath: { kind: "fallback" },
+	};
+};

shared/api/migrations/filters/planner/chooseCustomerAccessPath.ts

@@ -0,0 +1,60 @@
+import type { IRLeaf, IRNav, IRNode } from "../../compiler/ir/irTypes.js";
+import type { PlanIdConstraint } from "./accessPaths/planPlanIdAccessPath.js";
+
+export type ChosenCustomerAccessPath = {
+	id: "plan.plan_id";
+	constraint: PlanIdConstraint;
+};
+
+export const chooseCustomerAccessPath = (
+	ir: IRNode,
+): ChosenCustomerAccessPath | undefined => {
+	const planNav = findNecessaryPlanNav(ir);
+	if (!planNav) return undefined;
+
+	const planIdLeaf = findNecessaryPlanIdLeaf(planNav.child);
+	if (!planIdLeaf) return undefined;
+
+	return {
+		id: "plan.plan_id",
+		constraint: {
+			field: "plan_id",
+			op: planIdLeaf.op,
+			value: planIdLeaf.value,
+		},
+	};
+};
+
+const findNecessaryPlanNav = (node: IRNode): IRNav | undefined => {
+	const children = node.kind === "and" ? node.children : [node];
+	return children.find(
+		(child): child is IRNav =>
+			child.kind === "nav" &&
+			child.name === "plan" &&
+			child.quantifier === "some",
+	);
+};
+
+const findNecessaryPlanIdLeaf = (node: IRNode): PlanIdConstraint | undefined => {
+	const children = node.kind === "and" ? node.children : [node];
+	const leaf = children.find(
+		(child): child is IRLeaf =>
+			child.kind === "leaf" &&
+			child.field === "plan_id" &&
+			(child.op === "eq" || child.op === "in"),
+	);
+
+	if (!leaf) return undefined;
+	if (leaf.op === "eq" && typeof leaf.value === "string") {
+		return { field: "plan_id", op: "eq", value: leaf.value };
+	}
+	if (
+		leaf.op === "in" &&
+		Array.isArray(leaf.value) &&
+		leaf.value.length > 0 &&
+		leaf.value.every((value) => typeof value === "string")
+	) {
+		return { field: "plan_id", op: "in", value: leaf.value };
+	}
+	return undefined;
+};

shared/api/migrations/filters/planner/types.ts

@@ -0,0 +1,21 @@
+import type { CompiledSql } from "../../compiler/irToSql/irToSql.js";
+
+export type CustomerAccessPathId = "plan.plan_id";
+
+export type CustomerCandidateQuery = {
+	/** SQL source for FROM. It must expose a `c` alias with customer columns. */
+	source: CompiledSql;
+	/** Final customer predicate. Planned paths keep the fallback predicate here. */
+	where: CompiledSql;
+	accessPath:
+		| { kind: "fallback" }
+		| { kind: "planned"; id: CustomerAccessPathId };
+};
+
+export type CustomerAccessPath<TConstraint> = {
+	id: CustomerAccessPathId;
+	buildSource: (args: {
+		constraint: TConstraint;
+		ambient: Record<string, unknown>;
+	}) => CompiledSql;
+};