refactor: unify cache and integration flows

packages/sdk/src/lib/sdks.ts

@@ -249,7 +249,10 @@ export class ClientSDK {
 
     return retry(
       async () => {
-        const req = await this.#hooks.beforeRequest(context, request.clone());
+        const req = await this.#hooks.beforeRequest(
+          context,
+          request.clone() as Request,
+        );
         await logRequest(this.#logger, req).catch((e) =>
           this.#logger?.log("Failed to log request:", e)
         );

server/experiments/benchmarkV2CacheFlows.ts

@@ -425,7 +425,11 @@ const main = async () => {
 	});
 
 	const { db } = initDrizzle();
-	const redisV2 = resolveRedisV2();
+	const workerEnv = process.env as unknown as Env;
+	const redisV2 = resolveRedisV2({
+		env: workerEnv,
+		customerId: CONFIG.customerId,
+	});
 	await warmupRedisV2();
 	const ctx = makeCtx({ redisV2, db });
 

server/experiments/benchmarkV2CacheRedis.ts

@@ -355,7 +355,11 @@ const main = async () => {
 	});
 
 	const { db } = initDrizzle();
-	const redisV2 = resolveRedisV2();
+	const workerEnv = process.env as unknown as Env;
+	const redisV2 = resolveRedisV2({
+		env: workerEnv,
+		customerId: CONFIG.customerId,
+	});
 	await warmupRedisV2();
 
 	const ctx = makeCtx({ redisV2, db });

server/src/cron/resetCron/resetCustomerEntitlement.ts

@@ -8,8 +8,6 @@ import {
 import { UTCDate } from "@date-fns/utc";
 import { format } from "date-fns";
 import type { RepoContext } from "@/db/repoContext";
-import { resolveCustomerRedisRouting } from "@/external/redis/customerRedisRouting.js";
-import type { OrgWithRedisConfig } from "@/external/redis/orgRedisPool.js";
 import { invalidateCustomerEntitlementBalance } from "@/internal/customers/cache/fullSubject/actions/invalidate/invalidateCustomerEntitlementBalance.js";
 import { CusEntService } from "@/internal/customers/cusProducts/cusEnts/CusEntitlementService";
 import { getRelatedCusPrice } from "@/internal/customers/cusProducts/cusEnts/cusEntUtils.js";
@@ -18,6 +16,7 @@ import { getRolloverUpdates } from "@/internal/customers/cusProducts/cusEnts/cus
 import { getResetBalancesUpdate } from "@/internal/customers/cusProducts/cusEnts/groupByUtils";
 import { CusPriceService } from "@/internal/customers/cusProducts/cusPrices/CusPriceService.js";
 import { getEntOptions } from "@/internal/products/prices/priceUtils.js";
+import { createDisabledRedis } from "@/utils/disabledRedis.js";
 import { getNextResetAt } from "@/utils/timeUtils.js";
 import type { CronContext } from "../utils/CronContext";
 import { getStripeSubscriptionAnchor } from "./getStripeSubscriptionAnchor";
@@ -27,21 +26,15 @@ const shortDurations = [EntInterval.Minute, EntInterval.Hour, EntInterval.Day];
 
 const resetCustomerEntitlementInDb = async ({
 	ctx,
-	org,
 	cusEnt,
 	updatedCusEnts,
 	persistFreeOverage = false,
 }: {
 	ctx: CronContext;
-	org: OrgWithRedisConfig;
 	cusEnt: ResetCusEnt;
 	updatedCusEnts: ResetCusEnt[];
 	persistFreeOverage?: boolean;
 }) => {
-	const redisRouting = resolveCustomerRedisRouting({
-		org,
-		customerId: cusEnt.customer_id ?? "",
-	});
 	const repoContext: RepoContext = {
 		db: ctx.db,
 		logger: ctx.logger,
@@ -50,7 +43,8 @@ const resetCustomerEntitlementInDb = async ({
 		},
 		env: cusEnt.customer.env,
 		customerId: cusEnt.customer_id ?? "",
-		redisV2: redisRouting.redis,
+		redisV2: createDisabledRedis(),
+		cacheStore: ctx.cacheStore,
 	};
 
 	try {
@@ -199,30 +193,23 @@ export const resetCustomerEntitlement = async ({
 	persistFreeOverage = false,
 }: {
 	ctx: CronContext;
-	org?: OrgWithRedisConfig;
+	org?: unknown;
 	cusEnt: ResetCusEnt;
 	updatedCusEnts: ResetCusEnt[];
 	persistFreeOverage?: boolean;
 }) => {
-	const routingOrg = org ?? { id: cusEnt.customer.org_id, redis_config: null };
-	const redisRouting = resolveCustomerRedisRouting({
-		org: routingOrg,
-		customerId: cusEnt.customer_id ?? "",
-	});
+	void org;
 	const result = await resetCustomerEntitlementInDb({
 		ctx,
-		org: routingOrg,
 		cusEnt,
 		updatedCusEnts,
 		persistFreeOverage,
 	});
 	await invalidateCustomerEntitlementBalance({
+		ctx,
 		orgId: cusEnt.customer.org_id,
 		env: cusEnt.customer.env,
 		customerId: cusEnt.customer_id ?? "",
-		featureId: cusEnt.entitlement.feature.id,
-		customerEntitlementId: cusEnt.id,
-		redisV2: redisRouting.redis,
 	});
 	return result;
 };

server/src/cron/runCronTick.ts

@@ -1,6 +1,7 @@
 import { db, initDrizzleModules } from "../db/initDrizzle.js";
 import { createLogger } from "../external/logtail/logtailUtils.js";
 import type { Logger } from "../external/logtail/logtailUtils.js";
+import { resolveRequestCacheStore } from "../external/storage/cache/resolveRequestCacheStore.js";
 import { setAllEdgeConfigEnvs } from "../internal/misc/edgeConfig/edgeConfigRegistry.js";
 import {
 	describeSlotGate,
@@ -70,6 +71,7 @@ export const runCloudflareScheduledCron = async (env: Env) => {
 		ctx: {
 			db,
 			logger,
+			cacheStore: resolveRequestCacheStore({ env }),
 			workerEnv: env,
 		},
 	});

server/src/cron/utils/CronContext.ts

@@ -1,8 +1,10 @@
 import type { DrizzleCli } from "../../db/initDrizzle";
+import type { CacheStore } from "../../external/storage/cache/index.js";
 import type { Logger } from "../../external/logtail/logtailUtils";
 
 export interface CronContext {
 	db: DrizzleCli;
 	logger: Logger;
+	cacheStore?: CacheStore;
 	workerEnv?: Env;
 }

server/src/db/repoContext.ts

@@ -1,5 +1,6 @@
 import type { AppEnv } from "@autumn/shared";
 import type { Logger } from "@/external/logtail/logtailUtils";
+import type { CacheStore } from "@/external/storage/cache/index.js";
 import type { LegacyRedisClient } from "@/utils/legacyRedisClient.js";
 import type { DrizzleCli } from "./initDrizzle.js";
 
@@ -12,5 +13,6 @@ export interface RepoContext {
 	db: DrizzleCli;
 	logger: Logger;
 	redisV2: LegacyRedisClient;
+	cacheStore?: CacheStore;
 	customerId?: string;
 }

server/src/db/shed503OnTransientError.ts

@@ -1,5 +1,4 @@
 import { RecaseError } from "@autumn/shared";
-import { isTransientRedisError } from "@/external/redis/utils/isTransientRedisError.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { isTransientDbError } from "./dbUtils.js";
 
@@ -15,9 +14,7 @@ export const shed503OnTransientError = async <T>({
 	try {
 		return await run();
 	} catch (error) {
-		if (!(isTransientDbError({ error }) || isTransientRedisError({ error }))) {
-			throw error;
-		}
+		if (!isTransientDbError({ error })) throw error;
 		ctx.logger.warn(`[${source}] transient DB error, shedding with 503`, {
 			type: `${source}_fail_open`,
 			error,

server/src/external/autumn/autumnCli.ts

@@ -122,7 +122,10 @@ export class AutumnInt {
 		}
 	}
 
-	async get(path: string, headers?: Record<string, string>) {
+	async get<T = any>(
+		path: string,
+		headers?: Record<string, string>,
+	): Promise<T> {
 		const response = await fetch(`${this.baseUrl}${path}`, {
 			headers: { ...this.headers, ...headers },
 		});
@@ -152,10 +155,14 @@ export class AutumnInt {
 			});
 		}
 
-		return response.json();
+		return (await response.json()) as T;
 	}
 
-	async post(path: string, body: any, headers?: Record<string, string>) {
+	async post<T = any>(
+		path: string,
+		body: any,
+		headers?: Record<string, string>,
+	): Promise<T> {
 		const response = await fetch(`${this.baseUrl}${path}`, {
 			method: "POST",
 			headers: { ...this.headers, ...headers },
@@ -196,7 +203,7 @@ export class AutumnInt {
 			});
 		}
 
-		return response.json();
+		return (await response.json()) as T;
 	}
 	async patch(path: string, body: any) {
 		const response = await fetch(`${this.baseUrl}${path}`, {
@@ -423,7 +430,7 @@ export class AutumnInt {
 				headers["x-strip-internal"] = "false";
 			}
 
-			const data = await this.get(
+			const data = await this.get<any>(
 				`/customers?${new URLSearchParams(listParams as Record<string, string>).toString()}`,
 				Object.keys(headers).length > 0 ? headers : undefined,
 			);
@@ -493,7 +500,7 @@ export class AutumnInt {
 					finalParams.with_autumn_id ? "true" : "false",
 				);
 			}
-			const data = await this.get(
+			const data = await this.get<T>(
 				`/customers/${customerId}?${queryParams.toString()}`,
 				Object.keys(headers).length > 0 ? headers : undefined,
 			);
@@ -654,7 +661,9 @@ export class AutumnInt {
 		},
 
 		list: async (customerId: string): Promise<ApiEntityV0[]> => {
-			const data = await this.get(`/customers/${customerId}/entities`);
+				const data = await this.get<ApiEntityV0[]>(
+					`/customers/${customerId}/entities`,
+				);
 			return data;
 		},
 
@@ -788,7 +797,7 @@ export class AutumnInt {
 			internalId: string;
 			reward: any;
 		}) => {
-			const data = await this.post(
+			const data = await this.post<any>(
 				`/rewards/${internalId}?legacyStripe=true`,
 				reward,
 			);
@@ -969,7 +978,7 @@ export class AutumnInt {
 			queryParams.append("skip_cache", "true");
 		}
 
-		const data = await this.post(
+		const data = await this.post<T>(
 			`/check?${queryParams.toString()}`,
 			params,
 			headers,
@@ -1255,7 +1264,10 @@ export class AutumnInt {
 			params: TInput,
 			{ timeout }: { timeout?: number } = {},
 		): Promise<TResponse> => {
-			const data = await this.post(`/billing.create_schedule`, params);
+				const data = await this.post<TResponse>(
+					`/billing.create_schedule`,
+					params,
+				);
 
 			const concurrency = Number(this._env?.TEST_FILE_CONCURRENCY || "0");
 			const defaultTimeout = concurrency > 1 ? 5000 : 4000;

server/src/external/autumn/autumnRpcCli.ts

@@ -52,7 +52,7 @@ export class AutumnRpcCli {
 		return path.startsWith("/") ? path : `/${path}`;
 	}
 
-	async post(path: string, body: any) {
+	async post<T = unknown>(path: string, body: any): Promise<T> {
 		const response = await fetch(`${this.baseUrl}${this.resolvePath(path)}`, {
 			method: "POST",
 			headers: this.headers,
@@ -84,7 +84,7 @@ export class AutumnRpcCli {
 			});
 		}
 
-		return response.json();
+		return (await response.json()) as T;
 	}
 
 	rpc = {

server/src/external/redis/redisLockStore.ts

@@ -1,6 +1,5 @@
-import type { Redis } from "ioredis";
 import { registerDefaultLockStoreFactory } from "./defaultLockStore.js";
-import { redis } from "./initRedis.js";
+import type { LegacyRedisClient } from "@/utils/legacyRedisClient.js";
 import type {
 	LockStore,
 	LockStoreAcquireResult,
@@ -13,7 +12,7 @@ type StoredLockData = {
 };
 
 export class RedisLockStore implements LockStore {
-	constructor(private readonly redisClient: Redis = redis) {}
+	constructor(private readonly redisClient: LegacyRedisClient) {}
 
 	async acquire({
 		lockKey,
@@ -72,7 +71,11 @@ export class RedisLockStore implements LockStore {
 	}
 }
 
-const redisLockStore = new RedisLockStore();
-registerDefaultLockStoreFactory(() => redisLockStore);
+export const registerRedisLockStoreFactory = (
+	redisClient: LegacyRedisClient,
+): void => {
+	const redisLockStore = new RedisLockStore(redisClient);
+	registerDefaultLockStoreFactory(() => redisLockStore);
+};
 
 export { getDefaultLockStore } from "./defaultLockStore.js";

server/src/external/redis/registerRedisRequestAdapters.ts

@@ -8,6 +8,7 @@ import {
 	getRegionalRedis,
 	redis,
 } from "./initRedis.js";
+import { shouldUseRedisV2 } from "./initUtils/redisV2Availability.js";
 import { registerRampDestinationClientFactory } from "@/internal/misc/cacheV2Ramp/index.js";
 import { getOrgRedis, removeOrgRedis } from "./orgRedisPool.js";
 import { getRedisV2LockReceiptCandidates } from "./orgRedisUtils/orgRedisMigrationUtils.js";
@@ -19,11 +20,15 @@ import { registerCustomerRedisContextRouter } from "./requestCustomerRedisRoutin
 import { registerOrgRedisLifecycle } from "./requestOrgRedisLifecycle.js";
 import { registerRequestRedisTargetsForCustomer } from "./requestRedisTargetsForCustomer.js";
 import { registerRequestRedisV2Resolver } from "./requestRedisV2Resolver.js";
+import { registerRedisFailOpenAvailabilityResolver } from "./requestRedisFailOpenAvailability.js";
 import { resolveRedisV2 } from "./resolveRedisV2.js";
+import { registerRedisLockStoreFactory } from "./redisLockStore.js";
 
 registerRequestRedisV2Resolver(resolveRedisV2);
+registerRedisFailOpenAvailabilityResolver(shouldUseRedisV2);
 registerCustomerRedisContextRouter(getCtxWithCustomerRedis);
 registerRequestRedisTargetsForCustomer(getRedisTargetsForCustomer);
+registerRedisLockStoreFactory(redis);
 registerLockReceiptCandidateResolver(getRedisV2LockReceiptCandidates);
 registerLegacyLockReceiptRedisResolver(({ region }) =>
 	region && region !== currentRegion ? getRegionalRedis(region) : redis,

server/src/external/redis/requestRedisFailOpenAvailability.ts

@@ -0,0 +1,13 @@
+type RedisFailOpenAvailabilityResolver = () => boolean;
+
+let redisFailOpenAvailabilityResolver: RedisFailOpenAvailabilityResolver =
+	() => false;
+
+export const registerRedisFailOpenAvailabilityResolver = (
+	resolver: RedisFailOpenAvailabilityResolver,
+): void => {
+	redisFailOpenAvailabilityResolver = resolver;
+};
+
+export const shouldUseRequestRedisFailOpen = (): boolean =>
+	redisFailOpenAvailabilityResolver();

server/src/external/redis/resolveRequestLockStore.ts

@@ -8,20 +8,30 @@ type LockStoreEnv = Env & {
 	USE_DO_LOCK?: string;
 };
 
+type RequestLockStoreResolution = {
+	backend: "durable_object" | "default";
+	durableObjectEnabled: boolean;
+	durableObjectBound: boolean;
+};
+
 const isTruthyEnvFlag = (value: string | undefined): boolean =>
 	value === "true" || value === "1";
 
-export const getRequestLockStoreResolution = (env: Env) => {
+export const getRequestLockStoreResolution = (
+	env: Env,
+): RequestLockStoreResolution => {
 	const lockEnv = env as LockStoreEnv;
 	const durableObjectEnabled = isTruthyEnvFlag(lockEnv.USE_DO_LOCK);
 	const durableObjectBound = !!lockEnv.LOCK_DO;
 
 	return {
 		backend:
-			durableObjectEnabled && durableObjectBound ? "durable_object" : "redis",
+			durableObjectEnabled && durableObjectBound
+				? "durable_object"
+				: "default",
 		durableObjectEnabled,
 		durableObjectBound,
-	} as const;
+	};
 };
 
 export const resolveRequestLockStore = (env: Env): LockStore => {

server/src/external/redis/utils/withRedisFailOpen.ts

@@ -1,5 +1,5 @@
 import { isTransientDbError } from "@/db/dbUtils.js";
-import { shouldUseRedisV2 } from "@/external/redis/initUtils/redisV2Availability.js";
+import { shouldUseRequestRedisFailOpen } from "../requestRedisFailOpenAvailability.js";
 import { RedisUnavailableError } from "./errors.js";
 import { isTransientRedisError } from "./isTransientRedisError.js";
 
@@ -17,7 +17,7 @@ export const withRedisFailOpen = async <T>({
 	alsoFailOpen?: (error: unknown) => boolean;
 }): Promise<T> => {
 	try {
-		if (!shouldUseRedisV2()) {
+		if (!shouldUseRequestRedisFailOpen()) {
 			throw new RedisUnavailableError({ source, reason: "not_ready" });
 		}
 

server/src/external/revenueCat/handlers/handleGetRevenuecatProducts.ts

@@ -18,7 +18,12 @@ export const handleGetRevenueCatProducts = createRoute({
 		}
 
 		const projectId = getRevenuecatProjectId({ revenueCatConfig, env });
-		const accessToken = await getRevenuecatAccessToken({ db, org, env });
+		const accessToken = await getRevenuecatAccessToken({
+			db,
+			org,
+			env,
+			workerEnv: c.env,
+		});
 
 		if (!projectId || !accessToken) {
 			return c.json({ products: [] }, 404);

server/src/external/revenueCat/handlers/handleGetRevenuecatProjects.ts

@@ -15,7 +15,12 @@ export const handleGetRevenueCatProjects = createRoute({
 			return c.json({ projects: [] }, 404);
 		}
 
-		const accessToken = await getRevenuecatAccessToken({ db, org, env });
+		const accessToken = await getRevenuecatAccessToken({
+			db,
+			org,
+			env,
+			workerEnv: c.env,
+		});
 
 		if (!accessToken) {
 			return c.json({ projects: [] }, 404);
@@ -35,7 +40,12 @@ export const handleCreateRevenueCatProject = createRoute({
 		const { db, org, env } = c.get("ctx");
 		const { name } = c.req.valid("json");
 
-		const accessToken = await getRevenuecatAccessToken({ db, org, env });
+		const accessToken = await getRevenuecatAccessToken({
+			db,
+			org,
+			env,
+			workerEnv: c.env,
+		});
 		if (!accessToken) {
 			throw new RecaseError({
 				message: "Connect RevenueCat via OAuth before creating a project",

server/src/external/revenueCat/handlers/handlePreflightRevenueCatSync.ts

@@ -104,7 +104,12 @@ export const handlePreflightRevenueCatSync = createRoute({
 		if (!revenueCatConfig) return c.json({ items: [] });
 
 		const projectId = getRevenuecatProjectId({ revenueCatConfig, env });
-		const accessToken = await getRevenuecatAccessToken({ db, org, env });
+		const accessToken = await getRevenuecatAccessToken({
+			db,
+			org,
+			env,
+			workerEnv: c.env,
+		});
 		if (!projectId || !accessToken) return c.json({ items: [] });
 
 		const rcCli = initRevenuecatCli({ projectId, accessToken });

server/src/external/revenueCat/misc/getRevenuecatAccessToken.ts

@@ -11,7 +11,7 @@ import { decryptData, encryptData } from "@/utils/encryptUtils.js";
 
 const TOKEN_EXPIRY_SKEW_MS = 60_000;
 
-const getOAuthConfigForEnv = (env: Env) => ({
+const getOAuthConfigForEnv = ({
 	revenueCatConfig,
 	env,
 }: {
@@ -58,20 +58,22 @@ const refreshAndPersistTokens = async ({
 	db,
 	org,
 	env,
+	workerEnv,
 	oauthConfig,
 }: {
 	db: DrizzleCli;
 	org: Organization;
 	env: AppEnv;
+	workerEnv: Env;
 	oauthConfig: RevenueCatOAuthConfig;
 }): Promise<string> => {
-	const refreshToken = decryptData(oauthConfig.refresh_token);
-	const tokens = await refreshRcTokens({ refreshToken });
+	const refreshToken = decryptData(oauthConfig.refresh_token, workerEnv);
+	const tokens = await refreshRcTokens({ refreshToken, env: workerEnv });
 
 	const refreshedOAuthConfig: RevenueCatOAuthConfig = {
 		...oauthConfig,
-		access_token: encryptData(tokens.accessToken()),
-		refresh_token: encryptData(tokens.refreshToken()),
+		access_token: encryptData(tokens.accessToken(), workerEnv),
+		refresh_token: encryptData(tokens.refreshToken(), workerEnv),
 		expires_at: tokens.accessTokenExpiresAt().getTime(),
 		...(tokens.hasScopes() ? { scope: tokens.scopes().join(" ") } : {}),
 	};
@@ -90,27 +92,31 @@ export const refreshRevenuecatOAuthAccessToken = async ({
 	db,
 	org,
 	env,
+	workerEnv,
 }: {
 	db: DrizzleCli;
 	org: Organization;
 	env: AppEnv;
+	workerEnv: Env;
 }): Promise<string | null> => {
 	const oauthConfig = getOAuthConfigForEnv({
 		revenueCatConfig: org.processor_configs?.revenuecat ?? {},
 		env,
 	});
 	if (!oauthConfig) return null;
-	return refreshAndPersistTokens({ db, org, env, oauthConfig });
+	return refreshAndPersistTokens({ db, org, env, workerEnv, oauthConfig });
 };
 
 export const getRevenuecatAccessToken = async ({
 	db,
 	org,
 	env,
+	workerEnv,
 }: {
 	db: DrizzleCli;
 	org: Organization;
 	env: AppEnv;
+	workerEnv: Env;
 }): Promise<string | null> => {
 	const revenueCatConfig = org.processor_configs?.revenuecat;
 	if (!revenueCatConfig) return null;
@@ -119,10 +125,10 @@ export const getRevenuecatAccessToken = async ({
 
 	if (oauthConfig) {
 		if (isOAuthAccessTokenValid(oauthConfig)) {
-			return decryptData(oauthConfig.access_token);
+			return decryptData(oauthConfig.access_token, workerEnv);
 		}
 
-		return refreshAndPersistTokens({ db, org, env, oauthConfig });
+		return refreshAndPersistTokens({ db, org, env, workerEnv, oauthConfig });
 	}
 
 	const apiKey =
@@ -130,7 +136,7 @@ export const getRevenuecatAccessToken = async ({
 			? revenueCatConfig.api_key
 			: revenueCatConfig.sandbox_api_key;
 
-	return apiKey ? decryptData(apiKey) : null;
+	return apiKey ? decryptData(apiKey, workerEnv) : null;
 };
 
 export const getRevenuecatProjectId = ({

server/src/external/revenueCat/sync/syncRevenueCatProducts.ts

@@ -269,6 +269,14 @@ export const syncProductsToRevenueCat = async ({
 	productIds: string[];
 }): Promise<ProductSyncResult[]> => {
 	const { db, org, env } = ctx;
+	const workerEnv = ctx.workerEnv;
+	if (!workerEnv) {
+		throw new RecaseError({
+			message: "Worker env is required to sync RevenueCat products",
+			code: ErrCode.InvalidRequest,
+			statusCode: StatusCodes.INTERNAL_SERVER_ERROR,
+		});
+	}
 
 	const revenueCatConfig = org.processor_configs?.revenuecat;
 	if (!revenueCatConfig || !isRevenueCatPushEnabled({ revenueCatConfig, env })) {
@@ -280,7 +288,12 @@ export const syncProductsToRevenueCat = async ({
 	}
 
 	const projectId = getRevenuecatProjectId({ revenueCatConfig, env });
-	const accessToken = await getRevenuecatAccessToken({ db, org, env });
+	const accessToken = await getRevenuecatAccessToken({
+		db,
+		org,
+		env,
+		workerEnv,
+	});
 	if (!projectId || !accessToken) {
 		throw new RecaseError({
 			message: "RevenueCat is not fully configured (missing project or token)",

server/src/external/stripe/handleStripeWebhookEvent.ts

@@ -30,7 +30,7 @@ export const handleStripeWebhookEvent = async (
 	c: Context<StripeWebhookHonoEnv>,
 ) => {
 	const ctx = c.get("ctx") as StripeWebhookContext;
-	const { db, logger, org, env, stripeEvent } = ctx;
+	const { db, logger, org, env, stripeEvent, workerEnv } = ctx;
 	const event = stripeEvent;
 
 	try {
@@ -122,7 +122,7 @@ export const handleStripeWebhookEvent = async (
 		}
 
 		if (
-			env.NODE_ENV === "development" &&
+				workerEnv?.NODE_ENV === "development" &&
 			error instanceof Error &&
 			error.message.includes("No stripe account linked to organization")
 		) {

server/src/external/stripe/stripeCusUtils.ts

@@ -10,10 +10,9 @@ import { StatusCodes } from "http-status-codes";
 import type { Stripe } from "stripe";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import { createStripeCli } from "@/external/connect/createStripeCli.js";
-import { getCustomerRedisRoutingId } from "@/external/redis/customerRedisRoutingInfo.js";
-import { resolveRequestRedisV2 } from "@/external/redis/requestRedisV2Resolver.js";
 import { createStripeCustomer } from "@/external/stripe/customers";
 import { CusService } from "@/internal/customers/CusService.js";
+import { createDisabledRedis } from "@/utils/disabledRedis.js";
 
 import type { TestContext } from "../../../tests/utils/testInitUtils/createTestContext";
 import { createLogger } from "../logtail/logtailUtils";
@@ -154,10 +153,7 @@ export const attachPmToCus = async ({
 			org,
 			env,
 			logger: createLogger(effectiveWorkerEnv),
-			redisV2: resolveRequestRedisV2({
-				env: effectiveWorkerEnv,
-				customerId: getCustomerRedisRoutingId({ customer }),
-			}),
+			redisV2: createDisabledRedis(),
 		};
 
 		await CusService.update({

server/src/external/stripe/stripeOnboardingUtils.ts

@@ -17,10 +17,11 @@ export const createWebhookEndpoint = async (
 	apiKey: string,
 	env: AppEnv,
 	orgId: string,
+	workerEnv: Env,
 ) => {
 	const stripe = new Stripe(apiKey);
 
-	const webhookBaseUrl = env.STRIPE_WEBHOOK_URL || env.SERVER_URL;
+	const webhookBaseUrl = workerEnv.STRIPE_WEBHOOK_URL || workerEnv.SERVER_URL;
 
 	if (!webhookBaseUrl) {
 		throw new RecaseError({

server/src/external/stripe/webhookHandlers/handleStripeInvoiceCreated/setupInvoiceCreatedContext.ts

@@ -95,7 +95,7 @@ export const setupInvoiceCreatedContext = async ({
 		await customerProductActions.expiredCache.getAndMerge({
 			customerProducts: currentCustomerProducts,
 			stripeSubscriptionId,
-			cacheStore: ctx.cacheStore,
+			cacheStore: ctx.cacheStore!,
 		});
 
 	const scheduledCustomerProducts = fullCustomer.customer_products.filter(

server/src/external/stripe/webhookHandlers/handleStripeInvoiceFinalized/setupInvoiceFinalizedContext.ts

@@ -93,7 +93,7 @@ export const setupInvoiceFinalizedContext = async ({
 		await customerProductActions.expiredCache.getAndMerge({
 			customerProducts: currentCustomerProducts,
 			stripeSubscriptionId,
-			cacheStore: ctx.cacheStore,
+			cacheStore: ctx.cacheStore!,
 		});
 
 	if (customerProducts.length === 0) {

server/src/external/stripe/webhookHandlers/handleStripeInvoiceFinalized/tasks/processVercelInvoice.ts

@@ -24,6 +24,7 @@ export const processVercelInvoice = async ({
 	stripeSubscription: Stripe.Subscription | null;
 }): Promise<void> => {
 	const { stripeCli, org, env, db, fullCustomer } = ctx;
+	const workerEnv = ctx.workerEnv;
 	let { logger } = ctx;
 
 	if (stripeInvoice.amount_due <= 0) {
@@ -33,6 +34,10 @@ export const processVercelInvoice = async ({
 	if (!fullCustomer) {
 		return;
 	}
+	if (!workerEnv) {
+		logger.error("[vercel] missing Worker env while processing invoice");
+		return;
+	}
 
 	const invoiceMetadata = stripeInvoice.metadata as Record<
 		string,
@@ -107,6 +112,7 @@ export const processVercelInvoice = async ({
 			invoice: stripeInvoice,
 			customer: fullCustomer,
 			product,
+			env: workerEnv,
 			testOptions: ctx.testOptions,
 		});
 
@@ -118,6 +124,7 @@ export const processVercelInvoice = async ({
 			org,
 			features,
 			logger,
+			env: workerEnv,
 			testOptions: ctx.testOptions,
 		});
 	} catch (error) {

server/src/external/stripe/webhookHandlers/handleStripeInvoicePaid/setupStripeInvoicePaidContext.ts

@@ -56,7 +56,7 @@ export const setupStripeInvoicePaidContext = async ({
 		customerProducts = await customerProductActions.expiredCache.getAndMerge({
 			customerProducts,
 			stripeSubscriptionId,
-			cacheStore: ctx.cacheStore,
+			cacheStore: ctx.cacheStore!,
 		});
 
 		fullCustomer.customer_products = customerProducts;

server/src/external/stripe/webhookHandlers/handleStripeSubscriptionDeleted/tasks/expireAndActivateCustomerProducts.ts

@@ -91,6 +91,6 @@ export const expireAndActivateCustomerProducts = async ({
 	await customerProductActions.expiredCache.set({
 		stripeSubscriptionId: stripeSubscription.id,
 		customerProducts: expiredCustomerProducts,
-		cacheStore: ctx.cacheStore,
+			cacheStore: ctx.cacheStore!,
 	});
 };

server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleSchedulePhaseChanges/expireEndedCustomerProducts.ts

@@ -68,7 +68,7 @@ export const expireEndedCustomerProducts = async ({
 		await customerProductActions.expiredCache.set({
 			stripeSubscriptionId: stripeSubscription.id,
 			customerProducts: expiredCustomerProducts,
-			cacheStore: ctx.cacheStore,
+				cacheStore: ctx.cacheStore!,
 		});
 	}
 };

server/src/external/stripe/webhookHandlers/handleStripeSubscriptionUpdated/tasks/handleSchedulePhaseChanges/releaseScheduleIfLastPhase.ts

@@ -16,7 +16,7 @@ export const releaseScheduleIfLastPhase = async ({
 	ctx: StripeWebhookContext;
 	eventContext: StripeSubscriptionUpdatedContext;
 }): Promise<boolean> => {
-	const { db, org, env, logger } = ctx;
+	const { db, org, env, logger, workerEnv } = ctx;
 	const { stripeSubscription, nowMs } = eventContext;
 
 	const stripeSubscriptionSchedule = stripeSubscription.schedule;
@@ -73,7 +73,7 @@ export const releaseScheduleIfLastPhase = async ({
 		return true;
 	} catch (error: unknown) {
 		if (error instanceof Error) {
-			if (env.NODE_ENV === "development") {
+				if (workerEnv?.NODE_ENV === "development") {
 				logger.warn(
 					`[handleSchedulePhaseChanges] failed to release schedule: ${error.message}`,
 				);

server/src/external/stripe/webhookMiddlewares/stripeConnectSeederMiddleware.ts

@@ -31,12 +31,13 @@ export const stripeConnectSeederMiddleware = async (
 ) => {
 	const ctx = c.get("ctx") as StripeWebhookContext;
 	const { db, logger } = ctx;
-	const { env } = c.req.param() as { env: AppEnv };
+	const { env: appEnv } = c.req.param() as { env: AppEnv };
+	const workerEnv = c.env;
 
 	// Step 1: Initialize master stripe client
 	let masterStripe: Stripe;
 	try {
-		masterStripe = initMasterStripe();
+		masterStripe = initMasterStripe(workerEnv, { env: appEnv });
 	} catch (error) {
 		logger.error(`Failed to initialize master stripe client ${error}`);
 		return c.json({ error: "Failed to initialize stripe client" }, 500);
@@ -46,7 +47,8 @@ export const stripeConnectSeederMiddleware = async (
 	const webhookSecret = await getStripeWebhookSecret({
 		db,
 		orgId: c.req.query("org_id"),
-		env,
+		env: workerEnv,
+		appEnv,
 	});
 
 	// Step 3: Verify webhook signature
@@ -54,8 +56,8 @@ export const stripeConnectSeederMiddleware = async (
 	const signature = c.req.header("stripe-signature") || "";
 
 	const skipVerify =
-		env.STRIPE_WEBHOOK_SKIP_VERIFY === "true" &&
-		env.NODE_ENV !== "production";
+		workerEnv.STRIPE_WEBHOOK_SKIP_VERIFY === "true" &&
+		workerEnv.NODE_ENV !== "production";
 
 	let event: Stripe.Event;
 	if (skipVerify) {
@@ -78,7 +80,7 @@ export const stripeConnectSeederMiddleware = async (
 			);
 		} catch (err: unknown) {
 			const message = err instanceof Error ? err.message : String(err);
-			if (env.NODE_ENV !== "development") {
+			if (workerEnv.NODE_ENV !== "development") {
 				logger.warn(`Webhook verification error: ${message}`);
 			}
 			return c.json({ error: message }, 400);
@@ -114,7 +116,7 @@ export const stripeConnectSeederMiddleware = async (
 			return c.json({ error: "Failed to resolve org for Stripe webhook" }, 500);
 		}
 
-		if (env.NODE_ENV !== "development") {
+		if (workerEnv.NODE_ENV !== "development") {
 			logger.error(
 				`Account ID ${accountId} not linked to any org, skipping Stripe webhook`,
 			);
@@ -128,10 +130,11 @@ export const stripeConnectSeederMiddleware = async (
 	// Step 5: Set up context
 	ctx.org = org;
 	ctx.features = features;
-	ctx.env = env;
+	ctx.env = appEnv;
+	ctx.workerEnv = workerEnv;
 	ctx.authType = AuthType.Stripe;
 	ctx.stripeEvent = event;
-	ctx.stripeCli = createStripeCli({ org, env });
+	ctx.stripeCli = createStripeCli({ org, env: appEnv, workerEnv });
 
 	await next();
 };

server/src/external/stripe/webhookMiddlewares/stripeLegacySeederMiddleware.ts

@@ -24,13 +24,17 @@ export const stripeLegacySeederMiddleware = async (
 ) => {
 	const ctx = c.get("ctx") as StripeWebhookContext;
 	const { db, logger } = ctx;
-	const { orgId, env } = c.req.param() as { orgId: string; env: AppEnv };
+	const { orgId, env: appEnv } = c.req.param() as {
+		orgId: string;
+		env: AppEnv;
+	};
+	const workerEnv = c.env;
 
 	// Step 1: Get org and features
 	const data = await OrgService.getWithFeatures({
 		db,
 		orgId,
-		env,
+		env: appEnv,
 		allowNotFound: true,
 	});
 
@@ -41,10 +45,10 @@ export const stripeLegacySeederMiddleware = async (
 	const { org, features } = data;
 
 	// Step 2: Check if org is connected to Stripe
-	if (!isStripeConnected({ org, env })) {
-		logger.info(`Org ${orgId} and env ${env} is not connected to stripe`);
+	if (!isStripeConnected({ org, env: appEnv })) {
+		logger.info(`Org ${orgId} and env ${appEnv} is not connected to stripe`);
 		return c.json(
-			{ message: `Org ${orgId} and env ${env} is not connected to stripe` },
+			{ message: `Org ${orgId} and env ${appEnv} is not connected to stripe` },
 			200,
 		);
 	}
@@ -54,8 +58,8 @@ export const stripeLegacySeederMiddleware = async (
 	const signature = c.req.header("stripe-signature") || "";
 
 	const skipVerify =
-		env.STRIPE_WEBHOOK_SKIP_VERIFY === "true" &&
-		env.NODE_ENV !== "production";
+		workerEnv.STRIPE_WEBHOOK_SKIP_VERIFY === "true" &&
+		workerEnv.NODE_ENV !== "production";
 
 	let event: Stripe.Event;
 	if (skipVerify) {
@@ -73,7 +77,7 @@ export const stripeLegacySeederMiddleware = async (
 		}
 	} else {
 		try {
-			const webhookSecret = getStripeWebhookSecret(org, env);
+			const webhookSecret = getStripeWebhookSecret(org, appEnv, workerEnv);
 			event = await Stripe.webhooks.constructEventAsync(
 				rawBody,
 				signature,
@@ -91,10 +95,11 @@ export const stripeLegacySeederMiddleware = async (
 	// Step 4: Set up context
 	ctx.org = org;
 	ctx.features = features;
-	ctx.env = env;
+	ctx.env = appEnv;
+	ctx.workerEnv = workerEnv;
 	ctx.authType = AuthType.Stripe;
 	ctx.stripeEvent = event;
-	ctx.stripeCli = createStripeCli({ org, env });
+	ctx.stripeCli = createStripeCli({ org, env: appEnv, workerEnv });
 
 	await next();
 };

server/src/external/tinybird/migrations/migrationItemEventsDataSource.ts

@@ -13,7 +13,7 @@ export const getTinybirdApiUrl = (env: Env) => env.TINYBIRD_US_EAST_API_URL;
 export const getTinybirdToken = (env: Env) => env.TINYBIRD_US_EAST_TOKEN;
 
 /** Tinybird client for migration item events. Null until initMigrationTinybird(env) is called. */
-export let migrationTinybird: Tinybird | null = null;
+export let migrationTinybird: any = null;
 
 export const initMigrationTinybird = (env: Env) => {
 	const apiUrl = env.TINYBIRD_US_EAST_API_URL;

server/src/external/tinybird/tinybirdUtils.ts

@@ -9,12 +9,11 @@ export type TinybirdConfig = {
 export let tinybirdConfig: TinybirdConfig | null = null;
 
 export const initTinybirdConfig = (env: Env) => {
-
 	tinybirdConfig =
-		TINYBIRD_API_URL && TINYBIRD_TOKEN
+		env.TINYBIRD_API_URL && env.TINYBIRD_TOKEN
 			? {
-					baseUrl: TINYBIRD_API_URL,
-					token: TINYBIRD_TOKEN,
+					baseUrl: env.TINYBIRD_API_URL,
+					token: env.TINYBIRD_TOKEN,
 				}
 			: null;
 };

server/src/external/vercel/handlers/installations/handleDeleteInstallation.ts

@@ -32,7 +32,7 @@ export const handleDeleteInstallation = createRoute({
 						orgId,
 						env: ctx.env,
 					}),
-					sendCustomSvixEvent({
+					sendCustomSvixEvent(c.env)({
 						appId:
 							org.processor_configs?.vercel?.svix?.[
 								ctx.env === AppEnv.Live ? "live_id" : "sandbox_id"

server/src/external/vercel/handlers/installations/handleUpsertInstallation.ts

@@ -31,6 +31,7 @@ export const handleUpsertInstallation = createRoute({
 				token,
 				org: ctx.org,
 				env: ctx.env,
+				workerEnv: c.env,
 				testOptions,
 			});
 

server/src/external/vercel/handlers/resources/handleCreateResource.ts

@@ -273,7 +273,7 @@ export const handleCreateResource = createRoute({
 				);
 			}
 
-			await sendCustomSvixEvent({
+			await sendCustomSvixEvent(c.env)({
 				appId:
 					org.processor_configs?.vercel?.svix?.[
 						env === AppEnv.Live ? "live_id" : "sandbox_id"

server/src/external/vercel/handlers/resources/handleDeleteResource.ts

@@ -42,7 +42,7 @@ export const handleDeleteResource = createRoute({
 		}
 
 		try {
-			await sendCustomSvixEvent({
+			await sendCustomSvixEvent(c.env)({
 				appId:
 					org.processor_configs?.vercel?.svix?.[
 						env === AppEnv.Live ? "live_id" : "sandbox_id"

server/src/external/vercel/handlers/resources/handleRotateResourceSecret.ts

@@ -44,7 +44,7 @@ export const handleRotateResourceSecret = createRoute({
 			);
 		}
 
-		await sendCustomSvixEvent({
+		await sendCustomSvixEvent(c.env)({
 			appId:
 				org.processor_configs?.vercel?.svix?.[
 					env === AppEnv.Live ? "live_id" : "sandbox_id"

server/src/external/vercel/misc/vercelAuth.ts

@@ -35,14 +35,16 @@ const synthesizeTestClaims = ({
 	token,
 	org,
 	env,
+	workerEnv,
 	testOptions,
 }: {
 	token: string;
 	org: Organization;
 	env: AppEnv;
+	workerEnv: Env;
 	testOptions?: VercelOidcTestOptions;
 }): OidcClaims | null => {
-	if (env.NODE_ENV === "production") return null;
+	if (workerEnv.NODE_ENV === "production") return null;
 	if (testOptions?.allowVercelTestOidc !== true) return null;
 	if (!token.startsWith(TEST_OIDC_PREFIX)) return null;
 
@@ -69,14 +71,22 @@ export async function verifyToken({
 	token,
 	org,
 	env,
+	workerEnv,
 	testOptions,
 }: {
 	token: string;
 	org: Organization;
 	env: AppEnv;
+	workerEnv: Env;
 	testOptions?: VercelOidcTestOptions;
 }): Promise<OidcClaims> {
-	const testClaims = synthesizeTestClaims({ token, org, env, testOptions });
+	const testClaims = synthesizeTestClaims({
+		token,
+		org,
+		env,
+		workerEnv,
+		testOptions,
+	});
 	if (testClaims) return testClaims;
 
 	try {
@@ -245,7 +255,13 @@ export const vercelOidcAuthMiddleware = async (c: any, next: any) => {
 	// Verify JWT using JWKS
 	let claims: OidcClaims;
 	try {
-		claims = await verifyToken({ token, org, env, testOptions });
+		claims = await verifyToken({
+			token,
+			org,
+			env,
+			workerEnv: c.env,
+			testOptions,
+		});
 	} catch (error: any) {
 		logCaughtError({
 			logger,

server/src/external/vercel/misc/vercelInvoicing.ts

@@ -47,17 +47,19 @@ export const submitBillingDataToVercel = async ({
 	invoice,
 	customer,
 	product,
+	env,
 	testOptions,
 }: {
 	installationId: string;
 	invoice: Stripe.Invoice;
 	customer: Customer;
 	product: FullProduct;
+	env: Env;
 	testOptions?: VercelSdkTestOptions;
 }) => {
 	const vercel = new Vercel({
 		bearerToken: customer.processors?.vercel?.access_token,
-		serverURL: getVercelSdkServerURL(testOptions),
+		serverURL: getVercelSdkServerURL({ env, testOptions }),
 	});
 
 	const firstLineItem = invoice.lines.data[0];
@@ -120,6 +122,7 @@ export const submitInvoiceToVercel = async ({
 	org,
 	features,
 	logger,
+	env,
 	testOptions,
 }: {
 	installationId: string;
@@ -129,11 +132,12 @@ export const submitInvoiceToVercel = async ({
 	org: Organization;
 	features: Feature[];
 	logger?: Logger;
+	env: Env;
 	testOptions?: VercelSdkTestOptions;
 }) => {
 	const vercel = new Vercel({
 		bearerToken: customer.processors?.vercel?.access_token,
-		serverURL: getVercelSdkServerURL(testOptions),
+		serverURL: getVercelSdkServerURL({ env, testOptions }),
 	});
 
 	const price = productV2ToBasePrice({ product: mapToProductV2({ product }) });

server/src/external/vercel/vercelTestApiRouter.ts

@@ -31,9 +31,11 @@ type CapturedCall = {
 export const recordVercelTestCapture = async ({
 	call,
 	cacheStore,
+	env,
 }: {
 	call: CapturedCall;
 	cacheStore?: CacheStore;
+	env?: Env;
 }) => {
 	const key = captureCacheKey(call.installationId);
 	if (cacheStore) {
@@ -44,6 +46,7 @@ export const recordVercelTestCapture = async ({
 		return;
 	}
 
+	if (!env) throw new Error("Worker env is required without cacheStore");
 	const redis = resolveRequestRedisV2({ env });
 	await redis.rpush(key, JSON.stringify(call));
 	await redis.expire(key, CAPTURE_TTL_SECONDS);
@@ -52,13 +55,16 @@ export const recordVercelTestCapture = async ({
 export const getVercelTestCaptures = async ({
 	installationId,
 	cacheStore,
+	env,
 }: {
 	installationId: string;
 	cacheStore?: CacheStore;
+	env?: Env;
 }): Promise<CapturedCall[]> => {
 	const key = captureCacheKey(installationId);
 	if (cacheStore) return (await cacheStore.getJson<CapturedCall[]>(key)) || [];
 
+	if (!env) throw new Error("Worker env is required without cacheStore");
 	const redis = resolveRequestRedisV2({ env });
 	const raw = await redis.lrange(key, 0, -1);
 	return raw
@@ -75,9 +81,11 @@ export const getVercelTestCaptures = async ({
 export const clearVercelTestCaptures = async ({
 	installationId,
 	cacheStore,
+	env,
 }: {
 	installationId: string;
 	cacheStore?: CacheStore;
+	env?: Env;
 }): Promise<void> => {
 	const key = captureCacheKey(installationId);
 	if (cacheStore) {
@@ -85,6 +93,7 @@ export const clearVercelTestCaptures = async ({
 		return;
 	}
 
+	if (!env) throw new Error("Worker env is required without cacheStore");
 	const redis = resolveRequestRedisV2({ env });
 	await redis.del(key);
 };
@@ -113,6 +122,7 @@ vercelTestApiRouter.post(
 		const installationId = c.req.param("integrationConfigurationId");
 		const body = await parseJsonOrEmpty(c, "submitBillingData");
 		await recordVercelTestCapture({
+			env: c.env,
 			cacheStore: c.get("ctx")?.cacheStore,
 			call: {
 				method: "POST",
@@ -139,6 +149,7 @@ vercelTestApiRouter.post(
 		const body = await parseJsonOrEmpty(c, "submitInvoice");
 		const externalId = (body as { externalId?: string })?.externalId;
 		await recordVercelTestCapture({
+			env: c.env,
 			cacheStore: c.get("ctx")?.cacheStore,
 			call: {
 				method: "POST",
@@ -171,6 +182,7 @@ vercelTestApiRouter.get("/__captures/:installationId", async (c) => {
 	const installationId = c.req.param("installationId");
 	const captures = await getVercelTestCaptures({
 		installationId,
+		env: c.env,
 		cacheStore: c.get("ctx")?.cacheStore,
 	});
 	return c.json({ captures }, 200);
@@ -181,6 +193,7 @@ vercelTestApiRouter.delete("/__captures/:installationId", async (c) => {
 	const installationId = c.req.param("installationId");
 	await clearVercelTestCaptures({
 		installationId,
+		env: c.env,
 		cacheStore: c.get("ctx")?.cacheStore,
 	});
 	return c.json({ cleared: true }, 200);

server/src/external/vercel/vercelWebhookRouter.ts

@@ -159,7 +159,7 @@ vercelWebhookRouter.post(
 					return c.json({ received: true }, 200);
 
 				default:
-					await sendCustomSvixEvent({
+					await sendCustomSvixEvent(c.env)({
 						appId:
 							org.processor_configs?.vercel?.svix?.[
 								env === AppEnv.Live ? "live_id" : "sandbox_id"

server/src/init.ts

@@ -43,7 +43,6 @@ import {
 } from "./external/redis/initRedis.js";
 import { ensureRedisV2 } from "./external/redis/initRedisV2.js";
 import "./external/redis/registerRedisRequestAdapters.js";
-import "./external/redis/redisLockStore.js";
 import { primeRedisMonitor } from "./external/redis/initUtils/redisAvailability.js";
 import {
 	primeRedisV2Monitor,
@@ -58,8 +57,6 @@ import { initSentry } from "./sentry.js";
 import { configureTrigger } from "./trigger/configureTrigger.js";
 import { checkEnvVars } from "./utils/initUtils.js";
 import { startMemoryMonitor } from "./utils/memoryMonitor.js";
-import "./internal/misc/idempotency/redisIdempotencyStore.js";
-import "./internal/misc/rateLimiter/rateLimitRedisStore.js";
 
 let shuttingDown = false;
 

server/src/instrumentation.ts

@@ -35,7 +35,7 @@ export const initTelemetry = (env: Env) => {
 	const exportProcessor = new BatchSpanProcessor(traceExporter, {
 		scheduledDelayMillis: isDev ? 1000 : 5000,
 	});
-	const filteredExportProcessor = new FilteringSpanProcessor(exportProcessor);
+	const filteredExportProcessor = new FilteringSpanProcessor(exportProcessor, env);
 	const metricReader = env.AXIOM_METRICS_DATASET
 		? new PeriodicExportingMetricReader({
 				exporter: new OTLPMetricExporter({

server/src/internal/admin/handleAdminCacheV2Ramp.ts

@@ -83,7 +83,7 @@ export const handleUpsertAdminCacheV2Ramp = createRoute({
 		// snapshot (which can lag in multi-instance deployments).
 		const wasConfigured = !!getCacheV2RampConfig();
 		await upsertCacheV2RampConnection({
-			connectionString: encryptData(connectionString),
+			connectionString: encryptData(connectionString, c.env),
 			url: redisUrl.host,
 		});
 

server/src/internal/admin/handleGetMasterStripeAccount.ts

@@ -9,7 +9,7 @@ export const handleGetMasterStripeAccount = createRoute({
 		const { env, logger } = ctx;
 
 		try {
-			const masterStripe = initMasterStripe({ env });
+			const masterStripe = initMasterStripe(c.env, { env });
 			const account = await masterStripe.accounts.retrieve();
 
 			return c.json({

server/src/internal/analytics/handlers/handleProductsUpdated.ts

@@ -63,11 +63,15 @@ export const addProductsUpdatedWebhookTask = async ({
 	// Build action
 
 	try {
+		if (!ctx?.workerEnv) {
+			throw new Error("Cloudflare Worker env is required to enqueue products updated webhook task");
+		}
 		ctx?.logger.info(
 			`[addProductsUpdatedWebhookTask] Sending webhook for product ${cusProduct.product.name}, scenario: ${scenario}`,
 		);
 		await addTaskToQueue({
 			jobName: JobName.HandleProductsUpdated,
+			env: ctx.workerEnv,
 			payload: {
 				reqCtx: ctx ? parseCtxForAction({ ctx }) : undefined,
 				internalCustomerId,

server/src/internal/auth/oauth/handleOAuthConsentWithEnv.ts

@@ -123,7 +123,9 @@ const jsonOAuthError = ({ error }: { error: RecaseError }) =>
 
 export const handleOAuthConsentWithEnv = async (c: Context<HonoEnv>) => {
 	const auth = createAuth(c.env);
-	const { contentType, fields } = await parseRequestFields(c.req.raw.clone());
+	const { contentType, fields } = await parseRequestFields(
+		c.req.raw.clone() as Request,
+	);
 	const clientId = getClientIdFromFields(fields);
 	const redirectUri = getRedirectUriFromFields(fields);
 	const env = parseEnv(fields.env);

server/src/internal/auth/oauth/handleOAuthTokenWithApiKey.ts

@@ -111,7 +111,9 @@ const jsonTokenResponse = ({
 
 export const handleOAuthTokenWithApiKey = async (c: Context<HonoEnv>) => {
 	const auth = createAuth(c.env);
-	const resource = await getResourceFromOAuthTokenRequest(c.req.raw.clone());
+	const resource = await getResourceFromOAuthTokenRequest(
+		c.req.raw.clone() as Request,
+	);
 	const response = await auth.handler(c.req.raw);
 	if (!response.ok) return response;
 

server/src/internal/auth/oauth/oauthRouter.ts

@@ -2,7 +2,7 @@ import {
 	oauthProviderAuthServerMetadata,
 	oauthProviderOpenIdConfigMetadata,
 } from "@better-auth/oauth-provider";
-import { type Context, Hono } from "hono";
+import { type Context, Hono, type Next } from "hono";
 import { rateLimiter } from "hono-rate-limiter";
 import type { HonoEnv } from "@/honoUtils/HonoEnv.js";
 import { createAuth } from "@/utils/auth.js";
@@ -20,12 +20,18 @@ const getClientLookupRateLimitKey = (c: Context<HonoEnv>) =>
 	c.req.header("cf-connecting-ip") ??
 	"unknown";
 
-const oauthClientLookupLimiter = rateLimiter<HonoEnv>({
-	windowMs: 60 * 1000,
-	limit: (c) => (c.env.NODE_ENV === "development" ? 1000 : 60),
-	standardHeaders: "draft-6",
-	keyGenerator: getClientLookupRateLimitKey,
-});
+let oauthClientLookupRateLimiter: ReturnType<
+	typeof rateLimiter<HonoEnv>
+> | null = null;
+const oauthClientLookupLimiter = (c: Context<HonoEnv>, next: Next) => {
+	oauthClientLookupRateLimiter ??= rateLimiter<HonoEnv>({
+		windowMs: 60 * 1000,
+		limit: (ctx) => (ctx.env.NODE_ENV === "development" ? 1000 : 60),
+		standardHeaders: "draft-6",
+		keyGenerator: getClientLookupRateLimitKey,
+	});
+	return oauthClientLookupRateLimiter(c, next);
+};
 
 oauthRouter.get("/api/auth/.well-known/openid-configuration", (c) => {
 	const auth = createAuth(c.env);

server/src/internal/balances/events/EventBatchingManager.ts

@@ -52,6 +52,7 @@ class BatchingManager {
 		await addTaskToQueue({
 			jobName: JobName.InsertEventBatch,
 			payload: { events: eventItems },
+			env,
 		});
 
 		await sendEventsToTinybird({

server/src/internal/balances/finalizeLock/insertFinalizeLockEvent.ts

@@ -27,5 +27,5 @@ export const insertFinalizeLockEvent = ({
 		entityId: receipt.entity_id ?? undefined,
 	});
 
-	globalEventBatchingManager.addEvent(event);
+	globalEventBatchingManager.addEvent(event, ctx.workerEnv!);
 };

server/src/internal/balances/finalizeLock/insertFinalizeLockEventV2.ts

@@ -27,5 +27,5 @@ export const insertFinalizeLockEventV2 = ({
 		entityId: receipt.entity_id ?? undefined,
 	});
 
-	globalEventBatchingManager.addEvent(event);
+	globalEventBatchingManager.addEvent(event, ctx.workerEnv!);
 };

server/src/internal/balances/finalizeLock/runFinalizeLock.ts

@@ -10,7 +10,7 @@ import { isFullSubjectRolloutEnabled } from "@/internal/misc/rollouts/fullSubjec
 import { addToExtraLogs } from "@/utils/logging/addToExtraLogs.js";
 import { buildFinalizeLockContext } from "./buildFinalizeLockContext.js";
 import { runFinalizeLockV2 } from "./runFinalizeLockV2.js";
-import { runRedisFinalizeLock } from "./runRedisFinalizeLock.js";
+import { runPostgresFinalizeLock } from "./runPostgresFinalizeLock.js";
 
 type RunFinalizeLockArgs = {
 	workerEnv?: Env;
@@ -92,7 +92,7 @@ const runFinalizeLockInner = async ({
 		return { success: true };
 	}
 
-	await runRedisFinalizeLock({ ctx, finalizeLockContext, redisInstance });
+	await runPostgresFinalizeLock({ ctx, finalizeLockContext });
 
 	await deleteLockReceipt({ lockReceiptKey, redisInstance });
 

server/src/internal/balances/finalizeLock/runFinalizeLockV2.ts

@@ -11,7 +11,7 @@ import { cancelLockExpiry } from "@/internal/balances/utils/lock/cancelLockExpir
 import type { LockReceipt } from "@/internal/balances/utils/lock/fetchLockReceipt.js";
 import { buildFinalizeLockContextV2 } from "@/internal/balances/utils/lockV2/buildFinalizeLockContextV2.js";
 import { deleteLockReceiptV2 } from "@/internal/balances/utils/lockV2/deleteLockReceiptV2.js";
-import { runRedisFinalizeLockV2 } from "./runRedisFinalizeLockV2.js";
+import { runPostgresFinalizeLockV2 } from "./runPostgresFinalizeLockV2.js";
 
 /**
  * V2 finalize. Receives the receipt + claim outcome from the dispatcher
@@ -72,7 +72,7 @@ export const runFinalizeLockV2 = async ({
 		return { success: true };
 	}
 
-	await runRedisFinalizeLockV2({ ctx, finalizeLockContext });
+	await runPostgresFinalizeLockV2({ ctx, finalizeLockContext });
 	await deleteLockReceiptV2({ lockReceiptKey, redisInstance });
 
 	return { success: true };

server/src/internal/balances/finalizeLock/runRedisFinalizeLock.ts

@@ -1,5 +1,4 @@
 import type { Redis } from "ioredis";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { executeRedisDeduction } from "@/internal/balances/utils/deduction/executeRedisDeduction.js";
 import { deductionUpdatesToModifiedIds } from "@/internal/balances/utils/sync/deductionUpdatesToModifiedIds.js";
@@ -9,6 +8,8 @@ import type { FinalizeLockContext } from "./buildFinalizeLockContext.js";
 import { insertFinalizeLockEvent } from "./insertFinalizeLockEvent.js";
 import { runPostgresFinalizeLock } from "./runPostgresFinalizeLock.js";
 
+const DEFAULT_SYNC_REGION = "us-west-2";
+
 export const runRedisFinalizeLock = async ({
 	ctx,
 	finalizeLockContext,
@@ -55,13 +56,14 @@ export const runRedisFinalizeLock = async ({
 
 	if (modifiedCusEntIds.length > 0 || rolloverIds.length > 0) {
 		ctx.logger.info(`[QUEUE SYNC] (${receipt.customer_id})`);
+		const syncRegion = ctx.workerEnv?.AWS_REGION || DEFAULT_SYNC_REGION;
 		globalSyncBatchingManagerV2.addSyncItem({
 			customerId: receipt.customer_id,
 			orgId: ctx.org.id,
 			env: ctx.env,
 			cusEntIds: modifiedCusEntIds,
 			rolloverIds,
-			region: currentRegion,
+			region: syncRegion,
 		});
 	}
 

server/src/internal/balances/finalizeLock/runRedisFinalizeLockV2.ts

@@ -1,4 +1,3 @@
-import { currentRegion } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { executeRedisDeductionV2 } from "@/internal/balances/utils/deductionV2/executeRedisDeductionV2.js";
 import type { FinalizeLockContextV2 } from "@/internal/balances/utils/lockV2/buildFinalizeLockContextV2.js";
@@ -8,6 +7,8 @@ import { RedisDeductionError } from "@/internal/balances/utils/types/redisDeduct
 import { insertFinalizeLockEventV2 } from "./insertFinalizeLockEventV2.js";
 import { runPostgresFinalizeLockV2 } from "./runPostgresFinalizeLockV2.js";
 
+const DEFAULT_SYNC_REGION = "us-west-2";
+
 export const runRedisFinalizeLockV2 = async ({
 	ctx,
 	finalizeLockContext,
@@ -55,13 +56,14 @@ export const runRedisFinalizeLockV2 = async ({
 		rolloverIds.length > 0 ||
 		usageWindowUpdates.length > 0
 	) {
+		const syncRegion = ctx.workerEnv?.AWS_REGION || DEFAULT_SYNC_REGION;
 		globalSyncBatchingManagerV3.addSyncItem({
 			customerId: receipt.customer_id,
 			orgId: ctx.org.id,
 			env: ctx.env,
 			cusEntIds: modifiedCusEntIds,
 			rolloverIds,
-			region: currentRegion,
+			region: syncRegion,
 			entityId: receipt.entity_id ?? undefined,
 			modifiedCusEntIdsByFeatureId,
 			usageWindowUpdates,

server/src/internal/balances/track/runAsyncTrack.ts

@@ -12,8 +12,8 @@ export const runAsyncTrack = async ({
 	ctx: AutumnContext;
 	body: TrackParams;
 }): Promise<void> => {
-	const workerEnv = ctx.workerEnv ?? (process.env as unknown as Env);
-	if (!workerEnv.TRACK_ASYNC_QUEUE) {
+	const workerEnv = ctx.workerEnv;
+	if (!workerEnv?.TRACK_ASYNC_QUEUE) {
 		ctx.logger.error(
 			"[track] async=true requested but no async track queue is configured",
 		);

server/src/internal/balances/track/runBatchTrack.ts

@@ -15,8 +15,8 @@ export const runBatchTrack = async ({
 	ctx: AutumnContext;
 	body: BatchTrackParams;
 }): Promise<void> => {
-	const workerEnv = ctx.workerEnv ?? (process.env as unknown as Env);
-	if (!workerEnv.TRACK_ASYNC_QUEUE) {
+	const workerEnv = ctx.workerEnv;
+	if (!workerEnv?.TRACK_ASYNC_QUEUE) {
 		ctx.logger.error(
 			"[track] batch track requested but no async track queue is configured",
 		);

server/src/internal/balances/track/runTrackV2.ts

@@ -8,12 +8,10 @@ import {
 	type TrackParams,
 	type TrackResponseV3,
 } from "@autumn/shared";
-import { getOrSetCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/getOrSetCachedFullCustomer.js";
 import type { AutumnContext } from "../../../honoUtils/HonoEnv.js";
-import { getOrCreateCachedFullCustomer } from "../../customers/cusUtils/fullCustomerCacheUtils/getOrCreateCachedFullCustomer.js";
 import type { FeatureDeduction } from "../utils/types/featureDeduction.js";
 import { handleEventIdempotencyKey } from "./utils/handleEventIdempotencyKey.js";
-import { runRedisTrack } from "./utils/runRedisTrack.js";
+import { runPostgresTrack } from "./utils/runPostgresTrack.js";
 
 export const runTrackV2 = async ({
 	ctx,
@@ -36,21 +34,6 @@ export const runTrackV2 = async ({
 		});
 	}
 
-	// 1. Get full customer from cache or DB
-	const { customer_id, entity_id } = body;
-	const fullCustomer = ctx.apiVersion.gte(ApiVersion.V2_1)
-		? await getOrSetCachedFullCustomer({
-				ctx,
-				customerId: customer_id,
-				entityId: entity_id,
-				source: "getCheckData",
-			})
-		: await getOrCreateCachedFullCustomer({
-				ctx,
-				params: body,
-				source: "runTrackV2",
-			});
-
 	// If idempotency key is provided, insert event first and skip insertion later
 	if (body.idempotency_key) {
 		await handleEventIdempotencyKey({
@@ -59,12 +42,9 @@ export const runTrackV2 = async ({
 		});
 	}
 
-	// Try Redis deduction - returns TrackResponseV3 (with ApiBalanceV1)
-	const response: TrackResponseV3 = await runRedisTrack({
+	const response: TrackResponseV3 = await runPostgresTrack({
 		ctx,
-		fullCustomer,
 		featureDeductions,
-		overageBehavior: body.overage_behavior || "cap",
 		body,
 	});
 

server/src/internal/balances/track/utils/queueTrack.ts

@@ -19,10 +19,8 @@ export const queueTrack = async ({
 	messageDeduplicationId?: string;
 }) => {
 	try {
-		const workerEnv = ctx.workerEnv ?? (process.env as unknown as Env);
-		const hasCloudflareQueue =
-			!!workerEnv.TRACK_QUEUE || !!workerEnv.TRACK_ASYNC_QUEUE;
-		if (!hasCloudflareQueue) {
+		const workerEnv = ctx.workerEnv;
+		if (!workerEnv || (!workerEnv.TRACK_QUEUE && !workerEnv.TRACK_ASYNC_QUEUE)) {
 			ctx.logger.warn(
 				"[track] Redis unavailable and TRACK_QUEUE is unbound; falling back to synchronous track",
 			);

server/src/internal/balances/track/utils/runPostgresTrack.ts

@@ -54,7 +54,7 @@ export const runPostgresTrack = async ({
 	const { fullCus, updates } = result;
 
 	// Insert event directly into database
-	if (!body.skip_event && !body.idempotency_key && fullCus) {
+	if (!body.skip_event && fullCus) {
 		const eventInfo = buildEventInfo(body);
 		const event = initEvent({
 			ctx,
@@ -66,7 +66,7 @@ export const runPostgresTrack = async ({
 		});
 
 		// await EventService.insert({ db: ctx.db, event });
-		globalEventBatchingManager.addEvent(event);
+		globalEventBatchingManager.addEvent(event, ctx.workerEnv!);
 	}
 
 	// Build response using unified deductionToTrackResponse

server/src/internal/balances/track/utils/runRedisTrack.ts

@@ -4,7 +4,6 @@ import type {
 	TrackResponseV3,
 } from "@autumn/shared";
 import { tryCatch } from "@autumn/shared";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { globalEventBatchingManager } from "../../events/EventBatchingManager.js";
 import { buildEventInfo, initEvent } from "../../events/initEvent.js";
@@ -18,6 +17,8 @@ import type { RolloverUpdate } from "../../utils/types/rolloverUpdate.js";
 import { buildAiCreditCostProperty } from "./buildAiCreditCostProperty.js";
 import { handleRedisTrackError } from "./handleRedisTrackError.js";
 
+const DEFAULT_SYNC_REGION = "us-west-2";
+
 const aiCreditCostEntries = ({
 	updates,
 	fullCustomer,
@@ -58,13 +59,14 @@ const queueSyncItem = ({
 	if (modifiedCusEntIds.length === 0 && rolloverIds.length === 0) return;
 
 	ctx.logger.info(`[QUEUE SYNC] (${body.customer_id})`);
+	const syncRegion = ctx.workerEnv?.AWS_REGION || DEFAULT_SYNC_REGION;
 	globalSyncBatchingManagerV2.addSyncItem({
 		customerId: body.customer_id,
 		orgId: ctx.org.id,
 		env: ctx.env,
 		cusEntIds: modifiedCusEntIds,
 		rolloverIds,
-		region: currentRegion,
+		region: syncRegion,
 	});
 };
 
@@ -89,8 +91,9 @@ const queueEvent = ({
 			internalEntityId: fullCustomer.entity?.internal_id,
 			customerId: body.customer_id,
 			entityId: body.entity_id,
-		}),
-	);
+			}),
+			ctx.workerEnv!,
+		);
 };
 
 /**

server/src/internal/balances/track/v3/runPostgresTrackV3.ts

@@ -7,11 +7,9 @@ import {
 	initEvent,
 } from "@/internal/balances/events/initEvent.js";
 import { resolveInternalProductIdForEvent } from "../../events/resolveInternalProductIdForEvent.js";
-import {
-	deductionToTrackResponseV2,
-	executePostgresDeductionV2,
-	projectMutationLogsToTrackDeductionsV2,
-} from "@/internal/balances/utils/deductionV2/index.js";
+import { deductionToTrackResponseV2 } from "@/internal/balances/utils/deductionV2/deductionToTrackResponseV2.js";
+import { executePostgresDeductionV2 } from "@/internal/balances/utils/deductionV2/executePostgresDeductionV2.js";
+import { projectMutationLogsToTrackDeductionsV2 } from "@/internal/balances/utils/deductionV2/projectMutationLogsToTrackDeductionsV2.js";
 import type { FeatureDeduction } from "../../utils/types/featureDeduction.js";
 import { handlePostgresTrackError } from "../utils/handlePostgresTrackError.js";
 
@@ -59,7 +57,7 @@ export const runPostgresTrackV3 = async ({
 		mutationLogs,
 	});
 
-	if (!body.skip_event && !body.idempotency_key) {
+	if (!body.skip_event) {
 		const eventInfo = buildEventInfo(body);
 		const event = initEvent({
 			ctx,
@@ -72,7 +70,7 @@ export const runPostgresTrackV3 = async ({
 			deductions,
 		});
 
-		globalEventBatchingManager.addEvent(event);
+		globalEventBatchingManager.addEvent(event, ctx.workerEnv!);
 	}
 
 	const { balance, balances } = await deductionToTrackResponseV2({

server/src/internal/balances/track/v3/runRedisTrackV3.ts

@@ -89,8 +89,9 @@ const queueEvent = ({
 			entityId: body.entity_id,
 			deductions,
 			internalProductId,
-		}),
-	);
+			}),
+			ctx.workerEnv!,
+		);
 };
 
 export const runRedisTrackV3 = async ({

server/src/internal/balances/track/v3/runTrackV3.ts

@@ -14,8 +14,7 @@ import { getOrCreateCachedFullSubject } from "@/internal/customers/cache/fullSub
 import { getOrSetCachedFullSubject } from "@/internal/customers/cache/fullSubject/actions/getOrSetCachedFullSubject.js";
 import type { FeatureDeduction } from "../../utils/types/featureDeduction.js";
 import { handleEventIdempotencyKey } from "../utils/handleEventIdempotencyKey.js";
-import { runRedisTrackV3 } from "./runRedisTrackV3.js";
-import { getTrackIdempotencyKey } from "./trackIdempotencyKey.js";
+import { runPostgresTrackV3 } from "./runPostgresTrackV3.js";
 
 const getTrackFullSubject = async ({
 	ctx,
@@ -72,15 +71,11 @@ export const runTrackV3 = async ({
 		});
 	}
 
-	const redisIdempotencyKey = getTrackIdempotencyKey({ ctx });
-
-	const response: TrackResponseV3 = await runRedisTrackV3({
+	const response: TrackResponseV3 = await runPostgresTrackV3({
 		ctx,
 		fullSubject,
 		featureDeductions,
-		overageBehavior: body.overage_behavior || "cap",
 		body,
-		idempotencyKey: redisIdempotencyKey,
 	});
 
 	return applyResponseVersionChanges<TrackResponseV3>({

server/src/internal/balances/updateBalance/runRedisUpdateBalanceV2.ts

@@ -1,19 +1,12 @@
 import type { CustomerEntitlementFilters, FullCustomer } from "@autumn/shared";
-import { tryCatch } from "@autumn/shared";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { executePostgresDeduction } from "../utils/deduction/executePostgresDeduction.js";
-import { executeRedisDeduction } from "../utils/deduction/executeRedisDeduction.js";
-import { deductionUpdatesToModifiedIds } from "../utils/sync/deductionUpdatesToModifiedIds.js";
-import { syncItemV3 } from "../utils/sync/syncItemV3.js";
 import type { DeductionOptions } from "../utils/types/deductionTypes.js";
 import type { FeatureDeduction } from "../utils/types/featureDeduction.js";
-import { RedisDeductionError } from "../utils/types/redisDeductionError.js";
 
 /**
- * Updates balance in Redis using featureDeductions with targetBalance.
- * Falls back to Postgres if Redis fails with recoverable errors.
- * Syncs to Postgres after successful Redis update.
+ * Updates balance using the DB-backed deduction path.
+ * Kept under the legacy filename while callers are migrated incrementally.
  */
 export const runRedisUpdateBalanceV2 = async ({
 	ctx,
@@ -26,7 +19,6 @@ export const runRedisUpdateBalanceV2 = async ({
 	featureDeductions: FeatureDeduction[];
 	customerEntitlementFilters?: CustomerEntitlementFilters;
 }) => {
-	const { org, env } = ctx;
 	const customerId = fullCustomer.id || fullCustomer.internal_id;
 	const entityId = fullCustomer.entity?.id ?? undefined;
 
@@ -36,59 +28,12 @@ export const runRedisUpdateBalanceV2 = async ({
 		alterGrantedBalance: false,
 	};
 
-	const { data: result, error } = await tryCatch(
-		executeRedisDeduction({
-			ctx,
-			fullCustomer,
-			entityId,
-			deductions: featureDeductions,
-			deductionOptions,
-		}),
-	);
-
-	// Handle errors
-	if (error) {
-		if (error instanceof RedisDeductionError && error.shouldFallback()) {
-			// Fallback to Postgres for recoverable errors
-
-			ctx.logger.info(
-				`[runRedisUpdateBalanceV2] Falling back to Postgres (${error.code})`,
-			);
-
-			await executePostgresDeduction({
-				ctx,
-				fullCustomer,
-				customerId,
-				entityId,
-				deductions: featureDeductions,
-				options: deductionOptions,
-			});
-
-			return;
-		}
-
-		throw error;
-	}
-
-	const { updates, rolloverUpdates } = result;
-
-	const modifiedCusEntIds = deductionUpdatesToModifiedIds({ updates });
-	const modifiedRolloverIds = Object.keys(rolloverUpdates);
-
-	if (modifiedCusEntIds.length > 0 || modifiedRolloverIds.length > 0) {
-		await syncItemV3({
-			payload: {
-				customerId,
-				orgId: org.id,
-				env,
-				cusEntIds: modifiedCusEntIds,
-				rolloverIds: modifiedRolloverIds,
-				region: currentRegion,
-				timestamp: Date.now(),
-			},
-			ctx,
-		});
-	}
-
-	return result;
+	return executePostgresDeduction({
+		ctx,
+		fullCustomer,
+		customerId,
+		entityId,
+		deductions: featureDeductions,
+		options: deductionOptions,
+	});
 };

server/src/internal/balances/updateBalance/v2/updateRemainingV2.ts

@@ -2,15 +2,12 @@ import {
 	FeatureNotFoundError,
 	type FullSubject,
 	notNullish,
-	tryCatch,
 	type UpdateBalanceParamsV0,
 } from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
-import { executeRedisDeductionV2 } from "@/internal/balances/utils/deductionV2/executeRedisDeductionV2.js";
-import { syncItemV4 } from "@/internal/balances/utils/sync/syncItemV4.js";
+import { executePostgresDeductionV2 } from "@/internal/balances/utils/deductionV2/executePostgresDeductionV2.js";
 import { buildCustomerEntitlementFilters } from "../../utils/buildCustomerEntitlementFilters.js";
 import type { FeatureDeduction } from "../../utils/types/featureDeduction.js";
-import { handleUpdateBalanceDeductionErrorV2 } from "./handleUpdateBalanceDeductionErrorV2.js";
 
 /** Updates remaining balance using the FullSubject cache path. */
 export const updateRemainingV2 = async ({
@@ -43,54 +40,16 @@ export const updateRemainingV2 = async ({
 
 	const entityId = fullSubject.entityId;
 
-	const { data: result, error } = await tryCatch(
-		executeRedisDeductionV2({
-			ctx,
-			fullSubject,
-			entityId,
-			deductions: featureDeductions,
-			deductionOptions: {
-				overageBehaviour: "allow",
-				customerEntitlementFilters,
-				alterGrantedBalance: false,
-			},
-		}),
-	);
-
-	if (error) {
-		return handleUpdateBalanceDeductionErrorV2({
-			ctx,
-			error,
-			fullSubject,
-			featureDeductions,
+	return executePostgresDeductionV2({
+		ctx,
+		fullSubject,
+		customerId: fullSubject.customerId,
+		entityId,
+		deductions: featureDeductions,
+		options: {
+			overageBehaviour: "allow",
 			customerEntitlementFilters,
-		});
-	}
-
-	const { rolloverUpdates, modifiedCusEntIdsByFeatureId, usageWindowUpdates } =
-		result;
-	const cusEntIds = Object.values(modifiedCusEntIdsByFeatureId).flat();
-	const rolloverIds = Object.keys(rolloverUpdates);
-
-	if (
-		cusEntIds.length > 0 ||
-		rolloverIds.length > 0 ||
-		usageWindowUpdates.length > 0
-	) {
-		await syncItemV4({
-			ctx,
-			payload: {
-				customerId: fullSubject.customerId,
-				orgId: ctx.org.id,
-				env: ctx.env,
-				timestamp: Date.now(),
-				rolloverIds,
-				entityId: fullSubject.entityId,
-				modifiedCusEntIdsByFeatureId,
-				usageWindowUpdates,
-			},
-		});
-	}
-
-	return result;
+			alterGrantedBalance: false,
+		},
+	});
 };

server/src/internal/balances/updateBalance/v2/updateUsageV2.ts

@@ -6,16 +6,13 @@ import {
 	type FullSubject,
 	fullSubjectToCustomerEntitlements,
 	nullish,
-	tryCatch,
 	type UpdateBalanceParamsV0,
 } from "@autumn/shared";
 import { Decimal } from "decimal.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
-import { executeRedisDeductionV2 } from "@/internal/balances/utils/deductionV2/executeRedisDeductionV2.js";
-import { syncItemV4 } from "@/internal/balances/utils/sync/syncItemV4.js";
+import { executePostgresDeductionV2 } from "@/internal/balances/utils/deductionV2/executePostgresDeductionV2.js";
 import { buildCustomerEntitlementFilters } from "../../utils/buildCustomerEntitlementFilters.js";
 import type { FeatureDeduction } from "../../utils/types/featureDeduction.js";
-import { handleUpdateBalanceDeductionErrorV2 } from "./handleUpdateBalanceDeductionErrorV2.js";
 
 const getUpdateUsageTargetBalance = ({
 	fullSubject,
@@ -87,54 +84,16 @@ export const updateUsageV2 = async ({
 		},
 	];
 
-	const { data: result, error } = await tryCatch(
-		executeRedisDeductionV2({
-			ctx,
-			fullSubject,
-			entityId,
-			deductions: featureDeductions,
-			deductionOptions: {
-				overageBehaviour: "allow",
-				customerEntitlementFilters,
-				alterGrantedBalance: false,
-			},
-		}),
-	);
-
-	if (error) {
-		return handleUpdateBalanceDeductionErrorV2({
-			ctx,
-			error,
-			fullSubject,
-			featureDeductions,
+	return executePostgresDeductionV2({
+		ctx,
+		fullSubject,
+		customerId: fullSubject.customerId,
+		entityId,
+		deductions: featureDeductions,
+		options: {
+			overageBehaviour: "allow",
 			customerEntitlementFilters,
-		});
-	}
-
-	const { rolloverUpdates, modifiedCusEntIdsByFeatureId, usageWindowUpdates } =
-		result;
-	const cusEntIds = Object.values(modifiedCusEntIdsByFeatureId).flat();
-	const rolloverIds = Object.keys(rolloverUpdates);
-
-	if (
-		cusEntIds.length > 0 ||
-		rolloverIds.length > 0 ||
-		usageWindowUpdates.length > 0
-	) {
-		await syncItemV4({
-			ctx,
-			payload: {
-				customerId: fullSubject.customerId,
-				orgId: ctx.org.id,
-				env: ctx.env,
-				timestamp: Date.now(),
-				rolloverIds,
-				entityId: fullSubject.entityId,
-				modifiedCusEntIdsByFeatureId,
-				usageWindowUpdates,
-			},
-		});
-	}
-
-	return result;
+			alterGrantedBalance: false,
+		},
+	});
 };

server/src/internal/balances/utils/deduction/computeCreditCosts.ts

@@ -11,14 +11,14 @@ export type CreditCostLookup = (entitlementId: string) => number;
 export const computeCreditCosts = ({
 	cusEnts,
 	deduction,
-	env
+	env,
 }: {
 	cusEnts: FullCusEntWithFullCusProduct[];
 	deduction: FeatureDeduction;
-	env: Env
+	env: Env;
 }): CreditCostLookup => {
 	const costMap = new Map<string, number>();
-	const logger = createLogger(env)
+	const logger = createLogger(env);
 	for (const ce of cusEnts) {
 		// Token cost is USD: 1:1 on its own ent; parents apply their ratio to it.
 		if (

server/src/internal/balances/utils/deduction/executeDeductionCache.ts

@@ -1,6 +1,9 @@
 import type { EntityRolloverBalance, FullCustomer } from "@autumn/shared";
-import { redis } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import {
+	hasNonRedisSnapshotCacheStore,
+	invalidateCustomerCacheStoreSnapshots,
+} from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
 import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import type { DeductionUpdate } from "../types/deductionUpdate.js";
@@ -15,8 +18,8 @@ export interface RolloverOverwrite {
 }
 
 /**
- * Atomically updates cusEnt fields in the cached FullCustomer blob after a
- * Postgres deduction. Uses the unified updateCustomerEntitlements Lua script.
+ * Keeps customer snapshots fresh after a Postgres deduction. CacheStore-backed
+ * snapshots are invalidated; legacy Redis snapshots still use the Lua patch.
  * Fire-and-forget -- failures are logged but don't propagate.
  */
 export const syncCustomerEntitlementUpdatesToCache = async ({
@@ -50,7 +53,16 @@ export const syncCustomerEntitlementUpdatesToCache = async ({
 		);
 		if (cusEntIds.length === 0) return;
 
-		const { org, env } = ctx;
+		if (hasNonRedisSnapshotCacheStore({ ctx })) {
+			await invalidateCustomerCacheStoreSnapshots({
+				ctx,
+				customerId,
+				source: "syncCustomerEntitlementUpdatesToCache",
+			});
+			return;
+		}
+
+		const { org, env, redisV2 } = ctx;
 
 		const cacheKey = buildFullCustomerCacheKey({
 			orgId: org.id,
@@ -89,8 +101,14 @@ export const syncCustomerEntitlementUpdatesToCache = async ({
 			};
 		});
 
-		await tryRedisWrite(() =>
-			redis.updateCustomerEntitlements(cacheKey, JSON.stringify({ updates })),
+		await tryRedisWrite(
+			ctx.workerEnv ?? ({} as Env),
+			() =>
+				redisV2.updateCustomerEntitlements(
+					cacheKey,
+					JSON.stringify({ updates }),
+				),
+			redisV2,
 		);
 	} catch (error) {
 		ctx.logger.error(

server/src/internal/balances/utils/deduction/executePostgresDeduction.ts

@@ -51,6 +51,7 @@ export const executePostgresDeduction = async ({
 	mutationLogs: MutationLogItem[];
 }> => {
 	const { db, org, env } = ctx;
+	const deductionRegion = ctx.workerEnv?.AWS_REGION || "us-west-2";
 
 	ctx.logger.info(
 		`executing postgres deduction, deductions: ${JSON.stringify(
@@ -128,6 +129,8 @@ export const executePostgresDeduction = async ({
 						entityId,
 						items: [],
 						overrideLockValue: toDeduct,
+						redisInstance: ctx.redisV2,
+						region: deductionRegion,
 					});
 				}
 				continue;
@@ -225,6 +228,8 @@ export const executePostgresDeduction = async ({
 						featureId: feature.id,
 						entityId,
 						items: mutation_logs ?? [],
+						redisInstance: ctx.redisV2,
+						region: deductionRegion,
 					});
 				}
 			} catch (error) {

server/src/internal/balances/utils/deduction/executeRedisDeduction.ts

@@ -2,14 +2,13 @@ import type {
 	FullCusEntWithFullCusProduct,
 	FullCustomer,
 } from "@autumn/shared";
-import type { Redis } from "ioredis";
-import { currentRegion, redis } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { triggerAutoTopUp } from "@/internal/balances/autoTopUp/triggerAutoTopUp.js";
 import { handlePaidAllocatedCusEnt } from "@/internal/balances/utils/paidAllocatedFeature/handlePaidAllocatedCusEnt.js";
 import { rollbackDeduction } from "@/internal/balances/utils/paidAllocatedFeature/rollbackDeduction.js";
 import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+import type { LegacyRedisClient } from "@/utils/legacyRedisClient.js";
 import { fireTrackWebhooks } from "../../trackWebhooks/fireTrackWebhooks.js";
 import { saveLockReceipt } from "../lock/saveLockReceipt.js";
 import type { DeductionOptions } from "../types/deductionTypes.js";
@@ -29,6 +28,8 @@ import { mutationLogsToFeatures } from "./mutationLogsToFeatures.js";
 import { prepareDeductionOptions } from "./prepareDeductionOptions.js";
 import { prepareFeatureDeduction } from "./prepareFeatureDeduction.js";
 
+const DEFAULT_DEDUCTION_REGION = "us-west-2";
+
 export const executeRedisDeduction = async ({
 	ctx,
 	entityId,
@@ -42,7 +43,7 @@ export const executeRedisDeduction = async ({
 	deductions: FeatureDeduction[];
 	fullCustomer: FullCustomer;
 	deductionOptions?: DeductionOptions;
-	redisInstance?: Redis;
+	redisInstance?: LegacyRedisClient;
 }): Promise<{
 	oldFullCus: FullCustomer;
 	fullCus: FullCustomer | undefined;
@@ -51,6 +52,8 @@ export const executeRedisDeduction = async ({
 	mutationLogs: MutationLogItem[];
 }> => {
 	const { org, env } = ctx;
+	const targetRedis = redisInstance ?? ctx.redisV2;
+	const deductionRegion = ctx.workerEnv?.AWS_REGION || DEFAULT_DEDUCTION_REGION;
 	const oldFullCus = structuredClone(fullCustomer);
 
 	const options = prepareDeductionOptions({
@@ -125,7 +128,8 @@ export const executeRedisDeduction = async ({
 					entityId,
 					items: [],
 					overrideLockValue: toDeduct,
-					redisInstance,
+					redisInstance: targetRedis,
+					region: deductionRegion,
 				});
 			}
 			continue;
@@ -151,7 +155,7 @@ export const executeRedisDeduction = async ({
 			lock: preparedLock
 				? {
 						...preparedLock,
-						region: currentRegion,
+						region: deductionRegion,
 					}
 				: null,
 
@@ -160,14 +164,13 @@ export const executeRedisDeduction = async ({
 			lock_receipt_key: lockReceiptKey ?? null,
 		};
 
-		const targetRedis = redisInstance ?? redis;
 		const result = await tryRedisWrite(
 			() =>
 				targetRedis.deductFromCustomerEntitlements(
 					cacheKey,
 					JSON.stringify(luaParams),
 				),
-			redisInstance,
+			targetRedis,
 		);
 
 		if (!result) {

server/src/internal/balances/utils/deduction/prepareFeatureDeduction.ts

@@ -101,7 +101,11 @@ export const prepareFeatureDeduction = ({
 			.map((ce) => ce.entitlement.feature.id),
 	);
 
-	const getCreditCostForEnt = computeCreditCosts({ cusEnts, deduction });
+	const getCreditCostForEnt = computeCreditCosts({
+		cusEnts,
+		deduction,
+		env: ctx.workerEnv!,
+	});
 
 	// Build input for each customer entitlement
 	const customerEntitlementDeductions: CustomerEntitlementDeduction[] =

server/src/internal/balances/utils/deductionV2/executePostgresDeductionV2.ts

@@ -29,6 +29,8 @@ import { prepareFeatureDeductionV2 } from "./prepareFeatureDeductionV2.js";
 import { rollbackDeductionV2 } from "./rollbackDeductionV2.js";
 import { syncDeductionUpdatesToFullSubjectCache } from "./syncDeductionUpdatesToFullSubjectCache.js";
 
+const DEFAULT_DEDUCTION_REGION = "us-west-2";
+
 interface RolloverOverwrite {
 	id: string;
 	cus_ent_id: string;
@@ -59,6 +61,7 @@ export const executePostgresDeductionV2 = async ({
 	modifiedCusEntIdsByFeatureId: Record<string, string[]>;
 }> => {
 	const { db, org, env } = ctx;
+	const deductionRegion = ctx.workerEnv?.AWS_REGION || DEFAULT_DEDUCTION_REGION;
 
 	ctx.logger.info(
 		`executing postgres deduction v2, deductions: ${JSON.stringify(
@@ -132,6 +135,7 @@ export const executePostgresDeductionV2 = async ({
 						items: [],
 						overrideLockValue: toDeduct,
 						redisInstance: ctx.redisV2,
+						region: deductionRegion,
 					});
 				}
 				const unlimitedPlanLog = buildUnlimitedPlanMutationLog({
@@ -263,6 +267,7 @@ export const executePostgresDeductionV2 = async ({
 						entityId,
 						items: mutation_logs ?? [],
 						redisInstance: ctx.redisV2,
+						region: deductionRegion,
 					});
 				}
 			} catch (error) {

server/src/internal/balances/utils/deductionV2/executeRedisDeductionV2.ts

@@ -6,7 +6,6 @@ import {
 	notNullish,
 } from "@autumn/shared";
 import type { Redis } from "ioredis";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { triggerAutoTopUp } from "@/internal/balances/autoTopUp/triggerAutoTopUp.js";
 import {
@@ -43,6 +42,8 @@ import { prepareDeductionOptionsV2 } from "./prepareDeductionOptionsV2.js";
 import { prepareFeatureDeductionV2 } from "./prepareFeatureDeductionV2.js";
 import { rollbackDeductionV2 } from "./rollbackDeductionV2.js";
 
+const DEFAULT_DEDUCTION_REGION = "us-west-2";
+
 export const executeRedisDeductionV2 = async ({
 	ctx,
 	fullSubject,
@@ -72,6 +73,7 @@ export const executeRedisDeductionV2 = async ({
 	usageWindowMutations: UsageWindowMutation[];
 }> => {
 	workerEnv ??= ctx.workerEnv ?? ({} as Env);
+	const deductionRegion = workerEnv.AWS_REGION || DEFAULT_DEDUCTION_REGION;
 	const { org, env } = ctx;
 	const oldFullSubject = structuredClone(fullSubject);
 
@@ -162,6 +164,7 @@ export const executeRedisDeductionV2 = async ({
 					items: [],
 					overrideLockValue: toDeduct,
 					redisInstance: redisInstance ?? ctx.redisV2,
+					region: deductionRegion,
 				});
 			}
 			const unlimitedPlanLog = buildUnlimitedPlanMutationLog({
@@ -231,7 +234,7 @@ export const executeRedisDeductionV2 = async ({
 			lock: preparedLock
 				? {
 					...preparedLock,
-					region: currentRegion,
+					region: deductionRegion,
 				}
 				: null,
 			unwind_value: unwindValue ?? null,

server/src/internal/balances/utils/deductionV2/prepareFeatureDeductionV2.ts

@@ -161,6 +161,7 @@ export const prepareFeatureDeductionV2 = ({
 	const getCreditCostForEnt = computeCreditCosts({
 		cusEnts: customerEntitlements,
 		deduction,
+		env: ctx.workerEnv!,
 	});
 
 	const customerEntitlementDeductions: CustomerEntitlementDeduction[] =

server/src/internal/balances/utils/deductionV2/syncDeductionUpdatesToFullSubjectCache.ts

@@ -2,6 +2,10 @@ import type { EntityRolloverBalance, FullSubject } from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
 import { FULL_SUBJECT_CACHE_TTL_SECONDS } from "@/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.js";
+import {
+	hasNonRedisSnapshotCacheStore,
+	invalidateCustomerCacheStoreSnapshots,
+} from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import type { DeductionUpdate } from "../types/deductionUpdate.js";
 
@@ -110,9 +114,19 @@ export const syncDeductionUpdatesToFullSubjectCache = async ({
 			}
 		}
 
-		if (Object.keys(updatesByFeatureId).length === 0) return;
+			if (Object.keys(updatesByFeatureId).length === 0) return;
 
-		const pipeline = redisV2.pipeline();
+			if (hasNonRedisSnapshotCacheStore({ ctx })) {
+				await invalidateCustomerCacheStoreSnapshots({
+					ctx,
+					customerId,
+					entityId: fullSubject.entityId,
+					source: "syncDeductionUpdatesToFullSubjectCache",
+				});
+				return;
+			}
+
+			const pipeline = redisV2.pipeline();
 		for (const [featureId, updates] of Object.entries(updatesByFeatureId)) {
 			const balanceKey = buildSharedFullSubjectBalanceKey({
 				orgId: org.id,

server/src/internal/balances/utils/lock/saveLockReceipt.ts

@@ -1,8 +1,9 @@
 import { ErrCode, InternalError, RecaseError } from "@autumn/shared";
-import type { Redis } from "ioredis";
-import { currentRegion, redis } from "@/external/redis/initRedis.js";
 import type { MutationLogItem } from "@/internal/balances/utils/types/mutationLogItem.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+import type { LegacyRedisClient } from "@/utils/legacyRedisClient.js";
+
+const DEFAULT_LOCK_RECEIPT_REGION = "us-west-2";
 
 export const saveLockReceipt = async ({
 	lock,
@@ -12,6 +13,7 @@ export const saveLockReceipt = async ({
 	items,
 	overrideLockValue,
 	redisInstance,
+	region = DEFAULT_LOCK_RECEIPT_REGION,
 }: {
 	lock: {
 		lock_id?: string;
@@ -26,9 +28,10 @@ export const saveLockReceipt = async ({
 	entityId?: string;
 	items: MutationLogItem[];
 	overrideLockValue?: number;
-	redisInstance?: Redis;
+	redisInstance: LegacyRedisClient;
+	region?: string;
 }) => {
-	const targetRedis = redisInstance ?? redis;
+	const targetRedis = redisInstance;
 
 	const existing = await targetRedis.call("EXISTS", lock.redis_receipt_key);
 	if (existing === 1) {
@@ -49,7 +52,7 @@ export const saveLockReceipt = async ({
 					lock_id: lock.lock_id ?? null,
 					hashed_key: lock.hashed_key ?? null,
 					status: "pending",
-					region: currentRegion,
+					region,
 					customer_id: customerId,
 					feature_id: featureId,
 					entity_id: entityId ?? null,

server/src/internal/balances/utils/lockV2/saveLockReceiptV2.ts

@@ -1,9 +1,10 @@
 import { ErrCode, RecaseError } from "@autumn/shared";
-import type { Redis } from "ioredis";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import type { MutationLogItem } from "@/internal/balances/utils/types/mutationLogItem.js";
+import type { LegacyRedisClient } from "@/utils/legacyRedisClient.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 
+const DEFAULT_LOCK_RECEIPT_REGION = "us-west-2";
+
 /**
  * V2 save-lock-receipt. Stores the receipt as a plain JSON string via a single
  * `SET key value NX EXAT ttl_at` call — one Redis round trip instead of the
@@ -18,6 +19,7 @@ export const saveLockReceiptV2 = async ({
 	items,
 	overrideLockValue,
 	redisInstance,
+	region = DEFAULT_LOCK_RECEIPT_REGION,
 }: {
 	lock: {
 		lock_id?: string;
@@ -32,13 +34,14 @@ export const saveLockReceiptV2 = async ({
 	entityId?: string;
 	items: MutationLogItem[];
 	overrideLockValue?: number;
-	redisInstance: Redis;
+	redisInstance: LegacyRedisClient;
+	region?: string;
 }) => {
 	const payload = JSON.stringify({
 		lock_id: lock.lock_id ?? null,
 		hashed_key: lock.hashed_key ?? null,
 		status: "pending",
-		region: currentRegion,
+		region,
 		customer_id: customerId,
 		feature_id: featureId,
 		entity_id: entityId ?? null,

server/src/internal/balances/utils/refreshEntityAggregate/refreshEntityAggregateCache.ts

@@ -1,12 +1,16 @@
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
 import { AGGREGATED_BALANCE_FIELD } from "@/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.js";
+import {
+	hasNonRedisSnapshotCacheStore,
+	invalidateCustomerCacheStoreSnapshots,
+} from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
 import { getEntityAggregateForSync } from "@/internal/customers/repos/getFullSubject/getEntityAggregateForSync.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 
 /**
- * After DB sync, recompute entity aggregation from the now-authoritative DB
- * and HSET `_aggregated` on the affected balance hashes.
+ * After DB sync, keep entity aggregation snapshots fresh. CacheStore-backed
+ * snapshots are invalidated; legacy Redis balance hashes refresh `_aggregated`.
  */
 export const refreshEntityAggregateCache = async ({
 	ctx,
@@ -27,6 +31,15 @@ export const refreshEntityAggregateCache = async ({
 
 		if (featureIds.length === 0) return;
 
+		if (hasNonRedisSnapshotCacheStore({ ctx })) {
+			await invalidateCustomerCacheStoreSnapshots({
+				ctx,
+				customerId,
+				source: "refreshEntityAggregateCache",
+			});
+			return;
+		}
+
 		const { redisV2 } = ctx;
 
 		// Only refresh features whose balance hash already has `_aggregated`.

server/src/internal/balances/utils/sync/SyncBatchingManagerV2.ts

@@ -1,9 +1,10 @@
 import type { AppEnv } from "@autumn/shared";
 import { createLogger } from "@/external/logtail/logtailUtils.js";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import { JobName } from "@/queue/JobName.js";
 import { addTaskToQueue } from "@/queue/queueUtils.js";
 
+const DEFAULT_SYNC_REGION = "us-west-2";
+
 interface CustomerBatchContext {
 	customerId: string;
 	orgId: string;
@@ -167,7 +168,7 @@ export class SyncBatchingManagerV2 {
 				customerId,
 				orgId,
 				env,
-				region: region || currentRegion,
+				region: region || this.env?.AWS_REGION || DEFAULT_SYNC_REGION,
 				timestamp: Date.now(),
 				cusEntIds: new Set(),
 				rolloverIds: new Set(),

server/src/internal/balances/utils/sync/SyncBatchingManagerV3.ts

@@ -1,10 +1,11 @@
 import type { AppEnv } from "@autumn/shared";
 import { createLogger } from "@/external/logtail/logtailUtils.js";
-import { currentRegion } from "@/external/redis/initRedis.js";
 import { JobName } from "@/queue/JobName.js";
 import { addTaskToQueue } from "@/queue/queueUtils.js";
 import type { UsageWindowUpdate } from "../types/usageWindowUpdate.js";
 
+const DEFAULT_SYNC_REGION = "us-west-2";
+
 interface CustomerBatchContext {
 	customerId: string;
 	orgId: string;
@@ -191,7 +192,7 @@ export class SyncBatchingManagerV3 {
 				customerId,
 				orgId,
 				env,
-				region: region || currentRegion,
+				region: region || this.env?.AWS_REGION || DEFAULT_SYNC_REGION,
 				timestamp: Date.now(),
 				cusEntIds: new Set(),
 				rolloverIds: new Set(),

server/src/internal/balances/utils/sync/syncItemV3.ts

@@ -7,7 +7,6 @@ import {
 	tryCatch,
 } from "@autumn/shared";
 import { sql } from "drizzle-orm";
-import { getRegionalRedis } from "@/external/redis/initRedis.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { deleteCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.js";
 import { getCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.js";
@@ -196,15 +195,12 @@ export const syncItemV3 = async ({
 	ctx: AutumnContext;
 	payload: SyncItemV3;
 }): Promise<void> => {
-	const { customerId, region, cusEntIds, rolloverIds } = payload;
+	const { customerId, cusEntIds, rolloverIds } = payload;
 	const { db, logger } = ctx;
 
-	const redisInstance = region ? getRegionalRedis(region) : undefined;
-
 	const fullCustomer = await getCachedFullCustomer({
 		ctx,
 		customerId,
-		redisInstance,
 		skipRolloutCheck: true,
 	});
 

server/src/internal/billing/v2/providers/stripe/actionBuilders/evaluateStripeBillingPlan.ts

@@ -79,6 +79,7 @@ export const evaluateStripeBillingPlan = async ({
 		ctx,
 		billingContext,
 		stripeSubscriptionAction,
+		env: ctx.workerEnv!,
 	});
 
 	const stripeRefundAction = await buildStripeRefundAction({

server/src/internal/checkouts/middleware/checkoutMiddleware.ts

@@ -24,12 +24,17 @@ import { checkoutRepo } from "../repos/checkoutRepo";
  * Rate limiter: 10 requests per minute per checkout ID.
  * Prevents enumeration attacks on checkout URLs.
  */
-export const checkoutRateLimiter = rateLimiter<HonoEnv>({
-	windowMs: 60 * 1000, // 1 minute
-	limit: 10,
-	standardHeaders: "draft-6",
-	keyGenerator: (c) => c.req.param("checkout_id") ?? "unknown",
-});
+let checkoutRateLimiterInstance: ReturnType<typeof rateLimiter<HonoEnv>> | null =
+	null;
+export const checkoutRateLimiter = (c: Context<HonoEnv>, next: Next) => {
+	checkoutRateLimiterInstance ??= rateLimiter<HonoEnv>({
+		windowMs: 60 * 1000, // 1 minute
+		limit: 10,
+		standardHeaders: "draft-6",
+		keyGenerator: (ctx) => ctx.req.param("checkout_id") ?? "unknown",
+	});
+	return checkoutRateLimiterInstance(c, next);
+};
 
 // Extend HonoEnv to include checkout in context
 declare module "hono" {

server/src/internal/customers/actions/createWithDefaults/execute/executeAutumnCreateCustomerPlan.ts

@@ -103,9 +103,10 @@ export const executeAutumnCreateCustomerPlan = async ({
 		originalFullCustomer: context.fullCustomer,
 	});
 
-	if (ctx.authType === AuthType.SecretKey) {
-		await captureOrgEvent({
-			orgId: ctx.org.id,
+		if (ctx.authType === AuthType.SecretKey) {
+			await captureOrgEvent({
+				env: ctx.workerEnv!,
+				orgId: ctx.org.id,
 			event: "customer_created_via_api",
 			properties: {
 				org_slug: ctx.org.slug,

server/src/internal/customers/actions/resetCustomerEntitlements/executeResetCache.ts

@@ -1,21 +1,16 @@
 import type { AutumnContext } from "@/honoUtils/HonoEnv";
 import type { ResetCusEntParam } from "@/internal/balances/utils/sql/client.js";
-import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js";
-import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+import { invalidateCustomerCacheStoreSnapshots } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
 import type { RolloverClearingInfo } from "./applyResetResults.js";
 
 /**
- * Atomically resets cusEnt fields in the cached FullCustomer blob.
- * Uses the unified updateCustomerEntitlements Lua script.
- * Skips gracefully if the cache doesn't exist or the cusEnt was already reset.
- * Fire-and-forget -- failures are logged but don't propagate.
+ * Invalidates cached customer snapshots after entitlement resets. Reset values
+ * are persisted in DB and snapshots are rebuilt from DB on the next read.
  */
 export const executeResetCache = async ({
 	ctx,
 	customerId,
 	resets,
-	oldNextResetAts,
-	clearingMap,
 }: {
 	ctx: AutumnContext;
 	customerId: string;
@@ -25,44 +20,9 @@ export const executeResetCache = async ({
 }): Promise<void> => {
 	if (resets.length === 0) return;
 
-	const { org, env, redisV2 } = ctx;
-
-	const cacheKey = buildFullCustomerCacheKey({
-		orgId: org.id,
-		env,
+	await invalidateCustomerCacheStoreSnapshots({
+		ctx,
 		customerId,
-	});
-
-	const updates = resets.map((r) => {
-		const clearing = clearingMap[r.cus_ent_id];
-
-		return {
-			cus_ent_id: r.cus_ent_id,
-			balance: r.balance,
-			additional_balance: r.additional_balance,
-			adjustment: r.adjustment,
-			entities: r.entities,
-			next_reset_at: r.next_reset_at,
-			expected_next_reset_at: oldNextResetAts[r.cus_ent_id] ?? null,
-			rollover_insert: r.rollover_insert,
-			rollover_overwrites:
-				clearing && clearing.overwrites.length > 0 ? clearing.overwrites : null,
-			rollover_delete_ids:
-				clearing && clearing.deletedIds.length > 0 ? clearing.deletedIds : null,
-			new_replaceables: null,
-			deleted_replaceable_ids: null,
-		};
-	});
-
-	await tryRedisWrite(
-		ctx.workerEnv ?? ({} as Env),
-		() =>
-			redisV2.updateCustomerEntitlements(
-				cacheKey,
-				JSON.stringify({ updates }),
-			),
-		redisV2,
-	).catch((error) => {
-		ctx.logger.warn(`[executeResetCache] Redis cache update skipped: ${error}`);
+		source: "executeResetCache",
 	});
 };

server/src/internal/customers/actions/resetCustomerEntitlementsV2/resetSubjectCache.ts

@@ -1,38 +1,16 @@
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import type { ResetCusEntParam } from "@/internal/balances/utils/sql/client.js";
-import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
-import { FULL_SUBJECT_CACHE_TTL_SECONDS } from "@/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.js";
-import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+import { invalidateCustomerCacheStoreSnapshots } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
 import type { RolloverClearingInfo } from "../resetCustomerEntitlements/applyResetResults.js";
 
-interface SubjectBalanceUpdate {
-	cus_ent_id: string;
-	balance: number | null;
-	additional_balance: number | null;
-	adjustment: number | null;
-	entities: Record<string, unknown> | null;
-	next_reset_at: number | null;
-	expected_next_reset_at: number | null;
-	rollover_insert: unknown | null;
-	rollover_overwrites: unknown[] | null;
-	rollover_delete_ids: string[] | null;
-	new_replaceables: unknown[] | null;
-	deleted_replaceable_ids: string[] | null;
-}
-
 /**
- * Patches shared FullSubject balance hashes after a lazy reset.
- * Groups updates by feature_id and pipelines one updateSubjectBalances call per feature.
- * Fire-and-forget -- failures are logged but don't propagate.
- * Does not mutate cache_version in cache; version bumps are DB lifecycle concerns.
+ * Invalidates subject snapshots after a lazy reset. Reset values are persisted
+ * in DB; CacheStore snapshots are rebuilt from DB on the next read.
  */
 export const resetSubjectCache = async ({
 	ctx,
 	customerId,
 	resets,
-	oldNextResetAts,
-	clearingMap,
-	customerEntitlementFeatureIds,
 }: {
 	ctx: AutumnContext;
 	customerId: string;
@@ -44,81 +22,11 @@ export const resetSubjectCache = async ({
 	if (resets.length === 0) return;
 
 	try {
-		const { org, env, redisV2 } = ctx;
-
-		const updatesByFeatureId: Record<string, SubjectBalanceUpdate[]> = {};
-
-		for (const reset of resets) {
-			const featureId = customerEntitlementFeatureIds[reset.cus_ent_id];
-			if (!featureId) continue;
-
-			const clearing = clearingMap[reset.cus_ent_id];
-
-			const update: SubjectBalanceUpdate = {
-				cus_ent_id: reset.cus_ent_id,
-				balance: reset.balance,
-				additional_balance: reset.additional_balance,
-				adjustment: reset.adjustment,
-				entities: reset.entities,
-				next_reset_at: reset.next_reset_at,
-				expected_next_reset_at: oldNextResetAts[reset.cus_ent_id] ?? null,
-				rollover_insert: reset.rollover_insert,
-				rollover_overwrites:
-					clearing && clearing.overwrites.length > 0
-						? clearing.overwrites
-						: null,
-				rollover_delete_ids:
-					clearing && clearing.deletedIds.length > 0
-						? clearing.deletedIds
-						: null,
-				new_replaceables: null,
-				deleted_replaceable_ids: null,
-			};
-
-			if (!updatesByFeatureId[featureId]) {
-				updatesByFeatureId[featureId] = [];
-			}
-			updatesByFeatureId[featureId].push(update);
-		}
-
-		if (Object.keys(updatesByFeatureId).length === 0) return;
-
-		const pipeline = redisV2.pipeline();
-		for (const [featureId, updates] of Object.entries(updatesByFeatureId)) {
-			const balanceKey = buildSharedFullSubjectBalanceKey({
-				orgId: org.id,
-				env,
-				customerId,
-				featureId,
-			});
-			pipeline.updateSubjectBalances(
-				balanceKey,
-				JSON.stringify({
-					ttl_seconds: FULL_SUBJECT_CACHE_TTL_SECONDS,
-					updates,
-				}),
-			);
-		}
-
-		const pipelineResults = await tryRedisWrite(() => pipeline.exec(), redisV2);
-
-		if (pipelineResults) {
-			for (const [, resultRaw] of pipelineResults) {
-				if (typeof resultRaw !== "string") continue;
-				try {
-					const parsed = JSON.parse(resultRaw) as {
-						applied?: Record<string, boolean>;
-						skipped?: string[];
-						logs?: string[];
-					};
-					if (parsed.logs && parsed.logs.length > 0) {
-						ctx.logger.debug(
-							`[resetSubjectCache] Lua logs:\n${parsed.logs.join("\n")}`,
-						);
-					}
-				} catch {}
-			}
-		}
+		await invalidateCustomerCacheStoreSnapshots({
+			ctx,
+			customerId,
+			source: "resetSubjectCache",
+		});
 	} catch (error) {
 		ctx.logger.error(
 			`[resetSubjectCache] customer=${customerId}, failed: ${error}`,

server/src/internal/customers/actions/resetUsageWindows/rollUsageWindowsCache.ts

@@ -1,20 +1,15 @@
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
-import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
-import { FULL_SUBJECT_CACHE_TTL_SECONDS } from "@/internal/customers/cache/fullSubject/config/fullSubjectCacheConfig.js";
-import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+import { invalidateCustomerCacheStoreSnapshots } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
 import type { UsageWindowRoll } from "./computeUsageWindowRolls.js";
 
 /**
- * Atomically patches rolled counters into each affected feature's
- * '_usage_windows' field (one rollUsageWindows Lua call per feature,
- * pipelined). Fire-and-forget -- reads and the deduction script both derive
- * a closed window as 0, so a missed patch only delays the persisted roll.
+ * Invalidates cached customer snapshots after usage windows roll. The rolled
+ * window values are persisted in DB and snapshots are rebuilt on the next read.
  */
 export const rollUsageWindowsCache = async ({
 	ctx,
 	customerId,
 	rolls,
-	now,
 }: {
 	ctx: AutumnContext;
 	customerId: string;
@@ -24,34 +19,11 @@ export const rollUsageWindowsCache = async ({
 	if (rolls.length === 0) return;
 
 	try {
-		const { org, env, redisV2 } = ctx;
-
-		const rollsByFeatureId: Record<string, UsageWindowRoll[]> = {};
-		for (const roll of rolls) {
-			const featureRolls = rollsByFeatureId[roll.feature_id] ?? [];
-			featureRolls.push(roll);
-			rollsByFeatureId[roll.feature_id] = featureRolls;
-		}
-
-		const pipeline = redisV2.pipeline();
-		for (const [featureId, featureRolls] of Object.entries(rollsByFeatureId)) {
-			const balanceKey = buildSharedFullSubjectBalanceKey({
-				orgId: org.id,
-				env,
-				customerId,
-				featureId,
-			});
-			pipeline.rollUsageWindows(
-				balanceKey,
-				JSON.stringify({
-					now,
-					ttl_seconds: FULL_SUBJECT_CACHE_TTL_SECONDS,
-					rolls: featureRolls,
-				}),
-			);
-		}
-
-		await tryRedisWrite(() => pipeline.exec(), redisV2);
+		await invalidateCustomerCacheStoreSnapshots({
+			ctx,
+			customerId,
+			source: "rollUsageWindowsCache",
+		});
 	} catch (error) {
 		ctx.logger.error(
 			`[rollUsageWindowsCache] customer=${customerId}, failed: ${error}`,

server/src/internal/customers/add-product/createFullCusProduct.ts

@@ -21,7 +21,6 @@ import {
 } from "@autumn/shared";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import type { RepoContext } from "@/db/repoContext.js";
-import { createDisabledRedis } from "@/external/redis/disabledRedis.js";
 import { addProductsUpdatedWebhookTask } from "@/internal/analytics/handlers/handleProductsUpdated.js";
 import { triggerVerifyCacheConsistency } from "@/internal/billing/v2/workflows/verifyCacheConsistency/triggerVerifyCacheConsistency.js";
 import { searchCusProducts } from "@/internal/customers/cusProducts/cusProductUtils.js";
@@ -29,6 +28,7 @@ import { getEntRelatedPrice } from "@/internal/products/entitlements/entitlement
 import { freeTrialToStripeTimestamp } from "@/internal/products/free-trials/freeTrialUtils.js";
 import { getEntOptions } from "@/internal/products/prices/priceUtils.js";
 import { isFreeProduct, isOneOff } from "@/internal/products/productUtils.js";
+import { createDisabledRedis } from "@/utils/disabledRedis.js";
 import { generateId, notNullish, nullish } from "@/utils/genUtils.js";
 import type { InsertCusProductParams } from "../cusProducts/AttachParams.js";
 import { CusProductService } from "../cusProducts/CusProductService.js";

server/src/internal/customers/cache/cacheRoutingStaleness.ts

@@ -0,0 +1,24 @@
+import type { OrgRedisConfig } from "@autumn/shared";
+
+const getCustomerRoutingBucket = (customerId: string): number =>
+	Number(BigInt(Bun.hash(customerId)) % 100n);
+
+export const isCacheRoutingMigrationStale = ({
+	cachedAt,
+	customerId,
+	redisConfig,
+}: {
+	cachedAt?: number;
+	customerId?: string;
+	redisConfig?: OrgRedisConfig | null;
+}): boolean => {
+	if (!redisConfig?.migrationChangedAt) return false;
+	if (!customerId) return false;
+	if (cachedAt === undefined) return false;
+	if (cachedAt >= redisConfig.migrationChangedAt) return false;
+
+	const bucket = getCustomerRoutingBucket(customerId);
+	const wasOnDedicated = bucket < redisConfig.previousMigrationPercent;
+	const isOnDedicated = bucket < redisConfig.migrationPercent;
+	return wasOnDedicated !== isOnDedicated;
+};

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

@@ -1,6 +1,6 @@
 import type { Invoice } from "@autumn/shared";
-import { isRedisMigrationCacheStale } from "@/external/redis/customerRedisRoutingInfo.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { isCacheRoutingMigrationStale } from "@/internal/customers/cache/cacheRoutingStaleness.js";
 import { getFullSubjectRolloutSnapshot } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
 import { isSnapshotCacheStale } from "@/internal/misc/rollouts/rolloutUtils.js";
 import { buildFullSubjectKey } from "../builders/buildFullSubjectKey.js";
@@ -88,14 +88,14 @@ export const getFullSubjectBlobFromCacheStore = async ({
 		}
 
 		if (
-			isRedisMigrationCacheStale({
+			isCacheRoutingMigrationStale({
 				cachedAt: sanitized._cachedAt,
 				customerId,
 				redisConfig: ctx.org.redis_config,
 			})
 		) {
 			ctx.logger.warn(
-				`[getFullSubjectBlobFromCacheStore] Stale Redis migration cache for ${customerId}${entityId ? `:${entityId}` : ""}, evicting`,
+				`[getFullSubjectBlobFromCacheStore] Stale cache routing migration snapshot for ${customerId}${entityId ? `:${entityId}` : ""}, evicting`,
 			);
 			await ctx.cacheStore.delete(subjectKey);
 			return undefined;

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

@@ -25,7 +25,7 @@ export const getOrCreateCachedFullSubject = async ({
 	source?: string;
 }): Promise<FullSubject> => {
 	const { skipCache, logger } = ctx;
-	const useRedis = !skipCache;
+	const useCache = !skipCache;
 	const {
 		customer_id: customerId,
 		customer_data: customerData,
@@ -38,9 +38,7 @@ export const getOrCreateCachedFullSubject = async ({
 	let setCache = true;
 	let fetchedSubjectViewEpoch = 0;
 
-	if (customerId && useRedis) {
-		// Pipeline inside getCachedFullSubject already fetches the epoch,
-		// so we reuse it on miss instead of a second round trip.
+	if (customerId && useCache) {
 		const cachedResult = await getCachedFullSubject({
 			ctx,
 			customerId,
@@ -117,7 +115,7 @@ export const getOrCreateCachedFullSubject = async ({
 		}
 	}
 
-	if (useRedis && setCache) {
+	if (useCache && setCache) {
 		if (!normalizedResult) {
 			normalizedResult = await getFullSubjectNormalized({
 				ctx,

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

@@ -23,13 +23,11 @@ export const getOrSetCachedFullSubject = async ({
 	staleWhileRevalidate?: boolean;
 }): Promise<FullSubject> => {
 	const { skipCache, logger } = ctx;
-	const useRedis = !skipCache;
+	const useCache = !skipCache;
 
 	let fetchedSubjectViewEpoch = 0;
 
-	if (useRedis) {
-		// The pipeline inside getCachedFullSubject already fetches + refreshes
-		// the epoch, so we reuse it on miss instead of a second round trip.
+	if (useCache) {
 		const { fullSubject: cached, subjectViewEpoch } =
 			await getCachedFullSubject({
 				ctx,
@@ -65,7 +63,7 @@ export const getOrSetCachedFullSubject = async ({
 
 	const { normalized, fullSubject } = result;
 
-	if (useRedis) {
+	if (useCache) {
 		await setCachedFullSubject({
 			ctx,
 			normalized,

server/src/internal/customers/cache/fullSubject/actions/invalidate/batchInvalidateCachedFullSubjects.ts

@@ -1,16 +1,7 @@
-import type { AppEnv, Feature } from "@autumn/shared";
-import type { Redis } from "ioredis";
-import { batchDeleteCachedFullCustomers } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.js";
+import type { AppEnv } from "@autumn/shared";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
-import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+import { batchDeleteCachedFullCustomers } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.js";
 import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
-import { buildFullSubjectOrgEnvKey } from "../../builders/buildFullSubjectOrgEnvKey.js";
-import { buildFullSubjectViewEpochKey } from "../../builders/buildFullSubjectViewEpochKey.js";
-import { buildSharedFullSubjectBalanceKey } from "../../builders/buildSharedFullSubjectBalanceKey.js";
-import { FULL_SUBJECT_EPOCH_TTL_SECONDS } from "../../config/fullSubjectCacheConfig.js";
-import type { CachedFullSubject } from "../../fullSubjectCacheModel.js";
-
-const PIPELINE_BATCH_SIZE = 1000;
 
 type BatchInvalidateCustomer = {
 	orgId: string;
@@ -18,133 +9,59 @@ type BatchInvalidateCustomer = {
 	customerId: string;
 };
 
-type FeaturesByOrgEnv = Record<string, Feature[]>;
+const CACHE_STORE_DELETE_BATCH_SIZE = 1000;
 
-const batchInvalidateCachedFullSubjectsOnRedis = async ({
+const batchDeleteFullSubjectSnapshots = async ({
+	ctx,
 	customers,
-	featuresByOrgEnv,
-	redisV2,
-	workerEnv,
 }: {
+	ctx?: Pick<AutumnContext, "cacheStore" | "logger">;
 	customers: BatchInvalidateCustomer[];
-	featuresByOrgEnv: FeaturesByOrgEnv;
-	redisV2: Redis;
-	workerEnv: Env;
-}): Promise<void> => {
-	if (customers.length === 0 || redisV2.status !== "ready") return;
+}): Promise<number> => {
+	if (customers.length === 0) return 0;
 
+	const cacheStore = ctx?.cacheStore;
+	if (!cacheStore || cacheStore.metadata.backend === "redis") {
+		ctx?.logger?.warn(
+			`[batchInvalidateCachedFullSubjects] no CacheStore available, customers (${customers.length})`,
+		);
+		return 0;
+	}
+
+	let deleted = 0;
 	for (
 		let offset = 0;
 		offset < customers.length;
-		offset += PIPELINE_BATCH_SIZE
+		offset += CACHE_STORE_DELETE_BATCH_SIZE
 	) {
-		const batch = customers.slice(offset, offset + PIPELINE_BATCH_SIZE);
-		const readPipeline = redisV2.pipeline();
-
-		for (const { orgId, env, customerId } of batch) {
-			if (!customerId) continue;
-
-			const subjectKey = buildFullSubjectKey({ orgId, env, customerId });
-			readPipeline.get(subjectKey);
-		}
-
-		const readResults = await tryRedisRead(
-			workerEnv,
-			() => readPipeline.exec(),
-			redisV2,
+		const batch = customers.slice(offset, offset + CACHE_STORE_DELETE_BATCH_SIZE);
+		await cacheStore.deleteMany(
+			batch.map(({ orgId, env, customerId }) =>
+				buildFullSubjectKey({ orgId, env, customerId }),
+			),
 		);
-		if (!readResults) continue;
-
-		const writePipeline = redisV2.pipeline();
-
-		for (let index = 0; index < batch.length; index++) {
-			const customer = batch[index];
-			if (!customer?.customerId) continue;
-
-			const { orgId, env, customerId } = customer;
-			const subjectKey = buildFullSubjectKey({ orgId, env, customerId });
-			const epochKey = buildFullSubjectViewEpochKey({ orgId, env, customerId });
-			const subjectTuple = readResults[index];
-			const cachedRaw =
-				(subjectTuple?.[1] as string | null | undefined) ?? null;
-
-			let featureIds: string[] = [];
-			if (cachedRaw) {
-				try {
-					const manifest = JSON.parse(cachedRaw) as CachedFullSubject;
-					featureIds = manifest.meteredFeatures ?? [];
-				} catch {
-					featureIds = [];
-				}
-			}
-
-			if (featureIds.length === 0) {
-				const orgFeatures =
-					featuresByOrgEnv[buildFullSubjectOrgEnvKey({ orgId, env })] ?? [];
-				featureIds = orgFeatures.map((feature) => feature.id);
-			}
-
-			for (const featureId of new Set(featureIds)) {
-				writePipeline.unlink(
-					buildSharedFullSubjectBalanceKey({
-						orgId,
-						env,
-						customerId,
-						featureId,
-					}),
-				);
-			}
-
-			writePipeline.unlink(subjectKey);
-			writePipeline.incr(epochKey);
-			writePipeline.expire(epochKey, FULL_SUBJECT_EPOCH_TTL_SECONDS);
-		}
-
-		await tryRedisWrite(workerEnv, () => writePipeline.exec(), redisV2);
+		deleted += batch.length;
 	}
+
+	ctx.logger?.info(
+		`[batchInvalidateCachedFullSubjects] cacheStore deleted ${deleted} subject snapshots`,
+	);
+	return deleted;
 };
 
 export const batchInvalidateCachedFullSubjects = async ({
 	ctx,
 	customers,
-	featuresByOrgEnv,
-	getRedisTargetsForCustomer,
 }: {
-	ctx?: AutumnContext;
+	ctx?: Pick<AutumnContext, "cacheStore" | "logger">;
 	customers: BatchInvalidateCustomer[];
-	featuresByOrgEnv: FeaturesByOrgEnv;
-	getRedisTargetsForCustomer: ({
-		customer,
-	}: {
-		customer: BatchInvalidateCustomer;
-	}) => Redis[];
 }): Promise<number> => {
 	if (customers.length === 0) return 0;
 
-	const deleted = await batchDeleteCachedFullCustomers({ ctx, customers });
-	const workerEnv = ctx?.workerEnv ?? ({} as Env);
+	const [fullCustomerDeleted, fullSubjectDeleted] = await Promise.all([
+		batchDeleteCachedFullCustomers({ ctx, customers }),
+		batchDeleteFullSubjectSnapshots({ ctx, customers }),
+	]);
 
-	const customersByRedis = new Map<Redis, BatchInvalidateCustomer[]>();
-	for (const customer of customers) {
-		for (const targetRedis of new Set(
-			getRedisTargetsForCustomer({ customer }),
-		)) {
-			const existing = customersByRedis.get(targetRedis) ?? [];
-			existing.push(customer);
-			customersByRedis.set(targetRedis, existing);
-		}
-	}
-
-	await Promise.all(
-		[...customersByRedis.entries()].map(([targetRedis, redisCustomers]) =>
-			batchInvalidateCachedFullSubjectsOnRedis({
-				customers: redisCustomers,
-				featuresByOrgEnv,
-				redisV2: targetRedis,
-				workerEnv,
-			}),
-		),
-	);
-
-	return deleted;
+	return Math.max(fullCustomerDeleted, fullSubjectDeleted);
 };

server/src/internal/customers/cache/fullSubject/actions/invalidate/getOrInitFullSubjectViewEpoch.ts

@@ -1,39 +1,9 @@
-import { runRedisOp } from "@/external/redis/utils/runRedisOp.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
-import { buildFullSubjectViewEpochKey } from "../../builders/buildFullSubjectViewEpochKey.js";
-import { FULL_SUBJECT_EPOCH_TTL_SECONDS } from "../../config/fullSubjectCacheConfig.js";
 
 export const getOrInitFullSubjectViewEpoch = async ({
-	ctx,
-	customerId,
+	ctx: _ctx,
+	customerId: _customerId,
 }: {
 	ctx: AutumnContext;
 	customerId: string;
-}): Promise<number> => {
-	const { redisV2 } = ctx;
-	const epochKey = buildFullSubjectViewEpochKey({
-		orgId: ctx.org.id,
-		env: ctx.env,
-		customerId,
-	});
-
-	// GETEX reads the epoch and refreshes its TTL in one round trip.
-	const currentEpoch = await runRedisOp({
-		operation: () =>
-			redisV2.getex(epochKey, "EX", FULL_SUBJECT_EPOCH_TTL_SECONDS),
-		source: "getOrInitFullSubjectViewEpoch:getex",
-		redisInstance: redisV2,
-	});
-	if (currentEpoch !== null && currentEpoch !== undefined) {
-		const parsedEpoch = Number.parseInt(currentEpoch, 10);
-		return Number.isNaN(parsedEpoch) ? 0 : parsedEpoch;
-	}
-
-	await runRedisOp({
-		operation: () =>
-			redisV2.set(epochKey, "0", "EX", FULL_SUBJECT_EPOCH_TTL_SECONDS),
-		source: "getOrInitFullSubjectViewEpoch:init",
-		redisInstance: redisV2,
-	});
-	return 0;
-};
+}): Promise<number> => 0;

server/src/internal/customers/cache/fullSubject/actions/invalidate/incrementFullSubjectViewEpoch.ts

@@ -1,7 +1,4 @@
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
-import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
-import { buildFullSubjectViewEpochKey } from "../../builders/buildFullSubjectViewEpochKey.js";
-import { FULL_SUBJECT_EPOCH_TTL_SECONDS } from "../../config/fullSubjectCacheConfig.js";
 
 export const incrementFullSubjectViewEpoch = async ({
 	ctx,
@@ -10,18 +7,8 @@ export const incrementFullSubjectViewEpoch = async ({
 	ctx: AutumnContext;
 	customerId: string;
 }): Promise<number | null> => {
-	const { redisV2 } = ctx;
-	const epochKey = buildFullSubjectViewEpochKey({
-		orgId: ctx.org.id,
-		env: ctx.env,
-		customerId,
-	});
-
-	const nextEpoch = await tryRedisWrite(() => redisV2.incr(epochKey), redisV2);
-	if (nextEpoch === null || nextEpoch === undefined) return null;
-	await tryRedisWrite(
-		() => redisV2.expire(epochKey, FULL_SUBJECT_EPOCH_TTL_SECONDS),
-		redisV2,
+	ctx.logger.debug(
+		`[incrementFullSubjectViewEpoch] CacheStore snapshots do not use Redis epoch state for ${customerId}`,
 	);
-	return nextEpoch;
+	return null;
 };

server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateCustomerEntitlementBalance.ts

@@ -1,44 +1,34 @@
-import type { Redis } from "ioredis";
-import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
-import { buildSharedFullSubjectBalanceKey } from "../../builders/buildSharedFullSubjectBalanceKey.js";
-import { AGGREGATED_BALANCE_FIELD } from "../../config/fullSubjectCacheConfig.js";
+import type { AppEnv } from "@autumn/shared";
+import type { Logger } from "@/external/logtail/logtailUtils.js";
+import type { CacheStore } from "@/external/storage/cache/index.js";
+import { invalidateCustomerCacheStoreSnapshots } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/invalidateCustomerCacheStoreSnapshots.js";
+
+type InvalidateCustomerEntitlementBalanceArgs = {
+	ctx: {
+		cacheStore?: CacheStore;
+		logger: Logger;
+	};
+	orgId: string;
+	env: AppEnv;
+	customerId: string;
+};
 
 export const invalidateCustomerEntitlementBalance = async ({
+	ctx,
 	orgId,
 	env,
 	customerId,
-	featureId,
-	customerEntitlementId,
-	redisV2,
-}: {
-	orgId: string;
-	env: string;
-	customerId: string;
-	featureId: string;
-	customerEntitlementId: string;
-	redisV2: Redis;
-}): Promise<void> => {
-	if (
-		!orgId ||
-		!env ||
-		!customerId ||
-		!featureId ||
-		!customerEntitlementId ||
-		redisV2.status !== "ready"
-	) {
-		return;
-	}
+}: InvalidateCustomerEntitlementBalanceArgs): Promise<void> => {
+	if (!orgId || !env || !customerId) return;
 
-	const balanceKey = buildSharedFullSubjectBalanceKey({
-		orgId,
-		env,
+	await invalidateCustomerCacheStoreSnapshots({
+		ctx: {
+			org: { id: orgId },
+			env,
+			logger: ctx.logger,
+			cacheStore: ctx.cacheStore,
+		},
 		customerId,
-		featureId,
+		source: "invalidateCustomerEntitlementBalance",
 	});
-
-	await tryRedisWrite(
-		() =>
-			redisV2.hdel(balanceKey, customerEntitlementId, AGGREGATED_BALANCE_FIELD),
-		redisV2,
-	);
 };