feat: swappable redis

server/experiments/explainGetFull.ts

@@ -34,7 +34,7 @@ const main = async () => {
 		withTrialsUsed: false,
 		withSubs: true,
 		withEvents: false,
-		cusProductLimit: 15,
+		cusProductLimit: 10,
 	});
 
 	// Run the actual query to measure wall-clock time

server/src/_luaScriptsV2/fullSubjectDeduction/contextUtilsV2.lua

@@ -24,6 +24,7 @@ local function init_context(params)
     env = params.env,
     customer_id = params.customer_id,
     customer_entitlement_deductions = params.customer_entitlement_deductions,
+    balance_keys_by_feature_id = params.balance_keys_by_feature_id,
   })
 
   local context = {

server/src/_luaScriptsV2/fullSubjectDeduction/deductFromSubjectBalances.lua

@@ -94,6 +94,7 @@ local context = init_context({
   env = env,
   customer_id = customer_id,
   customer_entitlement_deductions = customer_entitlement_deductions,
+  balance_keys_by_feature_id = params.balance_keys_by_feature_id,
 })
 
 if #(context.missing_customer_entitlement_ids or {}) > 0 then

server/src/_luaScriptsV2/fullSubjectDeduction/readSubjectBalances.lua

@@ -1,19 +1,11 @@
 -- ============================================================================
 -- READ SUBJECT BALANCES
--- Functions to read shared SubjectBalance payloads from Redis hashes
+-- Functions to read shared SubjectBalance payloads from Redis hashes.
+-- Balance keys are built by the TS caller and passed in via
+-- params.balance_keys_by_feature_id to keep this script portable across
+-- Redis/Dragonfly cluster modes (no key construction inside Lua).
 -- ============================================================================
 
-local function build_shared_subject_balance_key(params)
-  return '{'
-      .. params.customer_id
-      .. '}:'
-      .. params.org_id
-      .. ':'
-      .. params.env
-      .. ':full_subject:shared_balances:'
-      .. params.feature_id
-end
-
 local function decode_subject_balance(raw_value)
   local decoded = safe_decode(raw_value)
   if type(decoded) ~= 'table' then
@@ -26,19 +18,17 @@ local function read_subject_balances(params)
   local balances_by_id = {}
   local missing_customer_entitlement_ids = {}
   local entries_by_balance_key = {}
+  local balance_keys_by_feature_id = safe_table(params.balance_keys_by_feature_id)
 
   for _, ent_obj in ipairs(params.customer_entitlement_deductions or {}) do
     local customer_entitlement_id = ent_obj.customer_entitlement_id
     local feature_id = ent_obj.feature_id
 
     if customer_entitlement_id and feature_id then
-      local balance_key = build_shared_subject_balance_key({
-        org_id = params.org_id,
-        env = params.env,
-        customer_id = params.customer_id,
-        feature_id = feature_id,
-      })
-
+      local balance_key = balance_keys_by_feature_id[feature_id]
+      if not balance_key then
+        table.insert(missing_customer_entitlement_ids, customer_entitlement_id)
+      else
         if entries_by_balance_key[balance_key] == nil then
           entries_by_balance_key[balance_key] = {
             feature_id = feature_id,
@@ -52,6 +42,7 @@ local function read_subject_balances(params)
         )
       end
     end
+  end
 
   for balance_key, entry in pairs(entries_by_balance_key) do
     local hmget_args = { 'HMGET', balance_key }

server/src/cron.ts

@@ -4,4 +4,13 @@ await initInfisical();
 const { warmupRegionalRedis } = await import("./external/redis/initRedis.js");
 await warmupRegionalRedis();
 
+// Edge config modules self-register on import (cron reads redis-v2-cache
+// so resolveRedisV2 picks the right instance on each ctx build).
+await import("./internal/misc/redisV2Cache/redisV2CacheStore.js");
+const { logger } = await import("./external/logtail/logtailUtils.js");
+const { startAllEdgeConfigPolling } = await import(
+	"./internal/misc/edgeConfig/edgeConfigRegistry.js"
+);
+await startAllEdgeConfigPolling({ logger });
+
 await import("./cron/cronInit.js");

server/src/cron/oneoffCron/runOneOffCleanup.ts

@@ -1,5 +1,6 @@
 import { type AppEnv, CusProductStatus } from "@autumn/shared";
 import type { RepoContext } from "@/db/repoContext.js";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.js";
 import { getOneOffCustomerProductsToCleanup } from "@/internal/customers/cusProducts/actions/cleanupOneOff/getOneOffToCleanup.js";
 import { batchUpdateCustomerProducts } from "@/internal/customers/cusProducts/repos/batchUpdateCustomerProducts.js";
 import type { CronContext } from "../utils/CronContext.js";
@@ -63,6 +64,7 @@ export const runOneOffCleanup = async ({ ctx }: { ctx: CronContext }) => {
 				},
 				env: group.env,
 				logger: logger,
+				redisV2: resolveRedisV2(),
 			};
 			await batchUpdateCustomerProducts({
 				ctx: repoContext,

server/src/cron/productCron/runProductCron.ts

@@ -76,6 +76,7 @@ export const runProductCron = async ({
 					await batchInvalidateCachedFullSubjects({
 						customers: customersToDelete,
 						featuresByOrgEnv,
+						redisV2: ctx.redisV2,
 					});
 					console.log(`Expired ${rows.length} customer products`);
 					continue;

server/src/cron/resetCron/resetCustomerEntitlement.ts

@@ -8,6 +8,7 @@ import {
 import { UTCDate } from "@date-fns/utc";
 import { format } from "date-fns";
 import type { RepoContext } from "@/db/repoContext";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.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";
@@ -42,6 +43,7 @@ const resetCustomerEntitlementInDb = async ({
 		},
 		env: cusEnt.customer.env,
 		customerId: cusEnt.customer_id ?? "",
+		redisV2: resolveRedisV2(),
 	};
 
 	try {
@@ -205,6 +207,7 @@ export const resetCustomerEntitlement = async ({
 		customerId: cusEnt.customer_id ?? "",
 		featureId: cusEnt.entitlement.feature.id,
 		customerEntitlementId: cusEnt.id,
+		redisV2: resolveRedisV2(),
 	});
 	return result;
 };

server/src/db/repoContext.ts

@@ -1,4 +1,5 @@
 import type { AppEnv } from "@autumn/shared";
+import type { Redis } from "ioredis";
 import type { Logger } from "@/external/logtail/logtailUtils";
 import type { DrizzleCli } from "./initDrizzle.js";
 
@@ -10,5 +11,6 @@ export interface RepoContext {
 
 	db: DrizzleCli;
 	logger: Logger;
+	redisV2: Redis;
 	customerId?: string;
 }

server/src/external/aws/s3/adminS3Config.ts

@@ -4,6 +4,8 @@ export const ADMIN_FEATURE_FLAGS_CONFIG_KEY = "admin/feature-flags-config.json";
 export const ADMIN_CUSTOMER_BLOCK_CONFIG_KEY =
 	"admin/customer-block-config.json";
 export const ADMIN_ORG_LIMITS_CONFIG_KEY = "admin/org-limits-config.json";
+export const ADMIN_REDIS_V2_CACHE_CONFIG_KEY =
+	"admin/redis-v2-cache-config.json";
 
 const bucket = process.env.S3_BUCKET || "autumn-prod-server";
 const region = process.env.S3_REGION || "us-east-2";

server/src/external/redis/initRedisV2.ts

@@ -1,4 +1,6 @@
 import type { Redis } from "ioredis";
+import { logger } from "@/external/logtail/logtailUtils.js";
+import type { RedisV2InstanceName } from "@/internal/misc/redisV2Cache/redisV2CacheSchemas.js";
 import {
 	createRedisConnection,
 	currentRegion,
@@ -17,6 +19,39 @@ export const redisV2: Redis =
 			})
 		: redis;
 
+const alternateInstanceUrls: Partial<Record<RedisV2InstanceName, string>> = {
+	canary: process.env.CACHE_V2_CANARY_URL?.trim() || undefined,
+	dragonfly: process.env.CACHE_V2_DRAGONFLY_URL?.trim() || undefined,
+};
+
+const instancePool = new Map<RedisV2InstanceName, Redis>();
+const missingUrlWarned = new Set<RedisV2InstanceName>();
+
+export const getAlternateRedisV2Instance = (
+	name: RedisV2InstanceName,
+): Redis | null => {
+	const cacheUrl = alternateInstanceUrls[name];
+	if (!cacheUrl) {
+		if (!missingUrlWarned.has(name)) {
+			missingUrlWarned.add(name);
+			logger.warn(
+				`[resolveRedisV2] activeInstance=${name} but URL is not set; falling back to primary`,
+			);
+		}
+		return null;
+	}
+
+	const existing = instancePool.get(name);
+	if (existing) return existing;
+
+	const instance = createRedisConnection({
+		cacheUrl,
+		region: `${currentRegion}:v2:${name}`,
+	});
+	instancePool.set(name, instance);
+	return instance;
+};
+
 export const warmupRedisV2 = async (): Promise<void> => {
 	if (redisV2 === redis) return;
 

server/src/external/redis/resolveRedisV2.ts

@@ -0,0 +1,29 @@
+import type { Redis } from "ioredis";
+import { logger } from "@/external/logtail/logtailUtils.js";
+import type { RedisV2InstanceName } from "@/internal/misc/redisV2Cache/redisV2CacheSchemas.js";
+import { getActiveRedisV2Instance } from "@/internal/misc/redisV2Cache/redisV2CacheStore.js";
+import {
+	getAlternateRedisV2Instance,
+	redisV2 as redisV2Primary,
+} from "./initRedisV2.js";
+
+let lastLoggedInstance: RedisV2InstanceName | null = null;
+
+/** Returns the active V2 Redis instance, selected by the redis-v2-cache edge config.
+ *  Called by every ctx-building middleware/worker — request-path code reads
+ *  ctx.redisV2 rather than calling this. */
+export const resolveRedisV2 = (): Redis => {
+	const activeInstance = getActiveRedisV2Instance();
+
+	if (activeInstance !== lastLoggedInstance) {
+		logger.info(
+			`[resolveRedisV2] activeInstance switched to "${activeInstance}"`,
+		);
+		lastLoggedInstance = activeInstance;
+	}
+
+	if (activeInstance === "primary") return redisV2Primary;
+
+	const alternate = getAlternateRedisV2Instance(activeInstance);
+	return alternate ?? redisV2Primary;
+};

server/src/external/stripe/stripeCusUtils.ts

@@ -10,6 +10,7 @@ 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 { resolveRedisV2 } from "@/external/redis/resolveRedisV2.js";
 import { createStripeCustomer } from "@/external/stripe/customers";
 import { CusService } from "@/internal/customers/CusService.js";
 
@@ -149,6 +150,7 @@ export const attachPmToCus = async ({
 			org,
 			env,
 			logger: logger,
+			redisV2: resolveRedisV2(),
 		};
 
 		await CusService.update({

server/src/honoMiddlewares/analyticsMiddleware.ts

@@ -1,5 +1,6 @@
 import type { Context, Next } from "hono";
 import type { HonoEnv } from "@/honoUtils/HonoEnv.js";
+import { isFullSubjectRolloutEnabled } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
 import { addAppContextToLogs } from "@/utils/logging/addContextToLogs";
 import { logRequestResult } from "./requestLogging/logRequestResult.js";
 
@@ -80,6 +81,11 @@ export const analyticsMiddleware = async (c: Context<HonoEnv>, next: Next) => {
 	const customerId = ctx.customerId;
 	const entityId = ctx.entityId;
 
+	const fullSubjectBucket =
+		customerId && ctx.rolloutSnapshot?.customerBucket !== undefined
+			? (ctx.rolloutSnapshot.customerBucket ?? undefined)
+			: undefined;
+
 	ctx.logger = addAppContextToLogs({
 		logger: ctx.logger,
 		appContext: {
@@ -92,6 +98,10 @@ export const analyticsMiddleware = async (c: Context<HonoEnv>, next: Next) => {
 			user_id: ctx.userId || undefined,
 			user_email: ctx.user?.email || undefined,
 			api_version: ctx.apiVersion?.semver,
+			full_subject_bucket: fullSubjectBucket ?? undefined,
+			full_subject_rollout_enabled: customerId
+				? isFullSubjectRolloutEnabled({ ctx })
+				: undefined,
 		},
 	});
 

server/src/honoMiddlewares/baseMiddleware.ts

@@ -9,6 +9,7 @@ import {
 import type { Context, Next } from "hono";
 import { db, dbGeneral } from "@/db/initDrizzle.js";
 import { logger } from "@/external/logtail/logtailUtils.js";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.js";
 import type { HonoEnv } from "@/honoUtils/HonoEnv.js";
 import { generateId } from "@/utils/genUtils.js";
 import { addRequestToLogs } from "@/utils/logging/addContextToLogs";
@@ -71,6 +72,7 @@ export const baseMiddleware = async (c: Context<HonoEnv>, next: Next) => {
 		db,
 		dbGeneral,
 		logger: childLogger,
+		redisV2: resolveRedisV2(),
 
 		// Request info
 		id,

server/src/honoUtils/HonoEnv.ts

@@ -7,6 +7,7 @@ import type {
 	Organization,
 } from "@autumn/shared";
 import type { User } from "better-auth";
+import type { Redis } from "ioredis";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import type { Logger } from "@/external/logtail/logtailUtils.js";
 import type { OidcClaims } from "@/external/vercel/misc/vercelAuth.js";
@@ -34,6 +35,10 @@ export type RequestContext = {
 	db: DrizzleCli;
 	dbGeneral: DrizzleCli;
 	logger: Logger;
+	/** V2 Redis instance for this request. Populated by every ctx-building
+	 *  middleware/worker via resolveRedisV2. Never import the singleton directly
+	 *  in request-path code. */
+	redisV2: Redis;
 
 	// Info
 	id: string;

server/src/init.ts

@@ -23,6 +23,7 @@ import "./internal/misc/featureFlags/featureFlagStore.js";
 import "./internal/misc/customerBlocks/customerBlockStore.js";
 import "./internal/misc/edgeConfig/orgLimitsStore.js";
 import "./internal/misc/stripeSync/stripeSyncStore.js";
+import "./internal/misc/redisV2Cache/redisV2CacheStore.js";
 import { closeStripeSyncEngine } from "@autumn/stripe-sync";
 import {
 	startRedisMonitor,

server/src/internal/admin/adminRouter.ts

@@ -5,6 +5,7 @@ import { handleGetAdminFeatureFlagsConfig } from "./handleGetAdminFeatureFlagsCo
 import { handleGetAdminOrgLimitsConfig } from "./handleGetAdminOrgLimitsConfig";
 import { handleGetAdminOrgRequestBlock } from "./handleGetAdminOrgRequestBlock";
 import { handleGetAdminRequestBlockConfig } from "./handleGetAdminRequestBlockConfig";
+import { handleGetAdminRedisV2CacheConfig } from "./handleGetAdminRedisV2CacheConfig";
 import { handleGetAdminStripeSyncConfig } from "./handleGetAdminStripeSyncConfig";
 
 import { handleGetInvoiceLineItems } from "./handleGetInvoiceLineItems";
@@ -18,6 +19,7 @@ import { handleUpsertAdminFeatureFlagsConfig } from "./handleUpsertAdminFeatureF
 import { handleUpsertAdminOrgLimitsConfig } from "./handleUpsertAdminOrgLimitsConfig";
 import { handleUpsertAdminOrgRequestBlock } from "./handleUpsertAdminOrgRequestBlock";
 import { handleUpsertAdminRequestBlockConfig } from "./handleUpsertAdminRequestBlockConfig";
+import { handleUpsertAdminRedisV2CacheConfig } from "./handleUpsertAdminRedisV2CacheConfig";
 import { handleUpsertAdminStripeSyncConfig } from "./handleUpsertAdminStripeSyncConfig";
 import { handleDeleteRolloutOrg } from "./rollouts/handleDeleteRolloutOrg";
 import { handleGetRollouts } from "./rollouts/handleGetRollouts";
@@ -67,6 +69,14 @@ honoAdminRouter.put(
 	"/stripe-sync-config",
 	...handleUpsertAdminStripeSyncConfig,
 );
+honoAdminRouter.get(
+	"/redis-v2-cache-config",
+	...handleGetAdminRedisV2CacheConfig,
+);
+honoAdminRouter.put(
+	"/redis-v2-cache-config",
+	...handleUpsertAdminRedisV2CacheConfig,
+);
 honoAdminRouter.get("/org-member", ...handleGetOrgMember);
 honoAdminRouter.get("/master-stripe-account", ...handleGetMasterStripeAccount);
 honoAdminRouter.get("/oauth-clients", ...handleListOAuthClients);

server/src/internal/admin/handleGetAdminRedisV2CacheConfig.ts

@@ -0,0 +1,18 @@
+import { createRoute } from "@/honoMiddlewares/routeHandler.js";
+import {
+	getActiveRedisV2Instance,
+	getRedisV2CacheStatus,
+} from "@/internal/misc/redisV2Cache/redisV2CacheStore.js";
+
+export const handleGetAdminRedisV2CacheConfig = createRoute({
+	handler: async (c) => {
+		const status = getRedisV2CacheStatus();
+		return c.json({
+			activeInstance: getActiveRedisV2Instance(),
+			configHealthy: status.healthy,
+			configConfigured: status.configured,
+			lastSuccessAt: status.lastSuccessAt ?? null,
+			error: status.error ?? null,
+		});
+	},
+});

server/src/internal/admin/handleUpsertAdminRedisV2CacheConfig.ts

@@ -0,0 +1,12 @@
+import { createRoute } from "@/honoMiddlewares/routeHandler.js";
+import { RedisV2CacheConfigSchema } from "@/internal/misc/redisV2Cache/redisV2CacheSchemas.js";
+import { updateActiveRedisV2Instance } from "@/internal/misc/redisV2Cache/redisV2CacheStore.js";
+
+export const handleUpsertAdminRedisV2CacheConfig = createRoute({
+	body: RedisV2CacheConfigSchema,
+	handler: async (c) => {
+		const { activeInstance } = c.req.valid("json");
+		await updateActiveRedisV2Instance({ activeInstance });
+		return c.json({ success: true, activeInstance });
+	},
+});

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

@@ -11,7 +11,6 @@ import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { triggerAutoTopUp } from "@/internal/balances/autoTopUp/triggerAutoTopUp.js";
 import { fireTrackWebhooks } from "@/internal/balances/trackWebhooks/fireTrackWebhooks.js";
 import { createAllocatedInvoice } from "@/internal/balances/utils/allocatedInvoice/createAllocatedInvoice.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import { saveLockReceipt } from "@/internal/balances/utils/lock/saveLockReceipt.js";
 import type { DeductionOptions } from "../types/deductionTypes.js";
 import type { DeductionUpdate } from "../types/deductionUpdate.js";
@@ -232,7 +231,7 @@ export const executePostgresDeductionV2 = async ({
 						featureId: feature.id,
 						entityId,
 						items: mutation_logs ?? [],
-						redisInstance: redisV2,
+						redisInstance: ctx.redisV2,
 					});
 				}
 			} catch (error) {

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

@@ -5,12 +5,12 @@ import {
 } from "@autumn/shared";
 import type { Redis } from "ioredis";
 import { currentRegion } from "@/external/redis/initRedis.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { triggerAutoTopUp } from "@/internal/balances/autoTopUp/triggerAutoTopUp.js";
 import { fireTrackWebhooks } from "@/internal/balances/trackWebhooks/fireTrackWebhooks.js";
 import { createAllocatedInvoice } from "@/internal/balances/utils/allocatedInvoice/createAllocatedInvoice.js";
 import { buildFullSubjectKey } from "@/internal/customers/cache/fullSubject/builders/buildFullSubjectKey.js";
+import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import type { DeductionOptions } from "../types/deductionTypes.js";
 import type { DeductionUpdate } from "../types/deductionUpdate.js";
@@ -125,11 +125,28 @@ export const executeRedisDeductionV2 = async ({
 			continue;
 		}
 
+		const balanceKeysByFeatureId: Record<string, string> = {};
+		for (const deductionEntry of customerEntitlementDeductions) {
+			const targetFeatureId =
+				(deductionEntry as { feature_id?: string }).feature_id ?? feature.id;
+			if (!balanceKeysByFeatureId[targetFeatureId]) {
+				balanceKeysByFeatureId[targetFeatureId] = buildSharedFullSubjectBalanceKey(
+					{
+						orgId: org.id,
+						env,
+						customerId,
+						featureId: targetFeatureId,
+					},
+				);
+			}
+		}
+
 		const luaParams = {
 			org_id: org.id,
 			env,
 			customer_id: customerId,
 			customer_entitlement_deductions: customerEntitlementDeductions,
+			balance_keys_by_feature_id: balanceKeysByFeatureId,
 			spend_limit_by_feature_id: spendLimitByFeatureId ?? null,
 			usage_based_cus_ent_ids_by_feature_id:
 				usageBasedCusEntIdsByFeatureId ?? null,
@@ -151,7 +168,7 @@ export const executeRedisDeductionV2 = async ({
 			lock_receipt_key: lockReceiptKey ?? null,
 		};
 
-		const targetRedis = redisInstance ?? redisV2;
+		const targetRedis = redisInstance ?? ctx.redisV2;
 
 		const result = await tryRedisWrite(
 			() =>

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

@@ -1,5 +1,4 @@
 import type { EntityRolloverBalance, FullSubject } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 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";
@@ -75,7 +74,7 @@ export const syncDeductionUpdatesToFullSubjectCache = async ({
 				customerEntitlement.next_reset_at ?? null;
 		}
 
-		const { org, env } = ctx;
+		const { org, env, redisV2 } = ctx;
 
 		// Group updates by featureId and build per-feature update arrays
 		const updatesByFeatureId: Record<string, SubjectBalanceUpdate[]> = {};

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

@@ -1,6 +1,5 @@
 import { ErrCode, RecaseError } from "@autumn/shared";
 import { redis } from "@/external/redis/initRedis.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv";
 import type { MutationLogItem } from "@/internal/balances/utils/types/mutationLogItem.js";
 import { tryRedisRead } from "@/utils/cacheUtils/cacheUtils.js";
@@ -46,6 +45,7 @@ export const fetchLockReceipt = async ({
 	ctx: AutumnContext;
 	lockId: string;
 }) => {
+	const { redisV2 } = ctx;
 	const hashedKey = Bun.hash(lockId).toString();
 	const lockReceiptKey = buildLockReceiptKey({
 		orgId: ctx.org.id,

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

@@ -1,7 +1,6 @@
 import type { Feature, FullSubject } from "@autumn/shared";
 import { type FinalizeLockParamsV0, findFeatureById } from "@autumn/shared";
 import type { Redis } from "ioredis";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import type { LockReceipt } from "@/internal/balances/utils/lock/fetchLockReceipt.js";
 import {
@@ -62,7 +61,7 @@ export const buildFinalizeLockContextV2 = async ({
 	return {
 		receipt,
 		lockReceiptKey,
-		redisInstance: redisV2,
+		redisInstance: ctx.redisV2,
 		fullSubject,
 		feature,
 		lockValue,

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

@@ -1,5 +1,4 @@
 import type { AppEnv } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 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";
@@ -34,6 +33,7 @@ export const refreshEntityAggregateCache = async ({
 		if (aggregated.length === 0) return;
 
 		const affectedFeatureIds = new Set(featureIds);
+		const { redisV2 } = ctx;
 		const pipeline = redisV2.pipeline();
 		let writeCount = 0;
 

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

@@ -148,8 +148,7 @@ export const syncItemV4 = async ({
 		modifiedCusEntIdsByFeatureId,
 	)) {
 		const result = await getCachedFeatureBalance({
-			orgId,
-			env,
+			ctx,
 			customerId,
 			featureId,
 			customerEntitlementIds,

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

@@ -1,4 +1,3 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 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";
@@ -45,7 +44,7 @@ export const resetSubjectCache = async ({
 	if (resets.length === 0) return;
 
 	try {
-		const { org, env } = ctx;
+		const { org, env, redisV2 } = ctx;
 
 		const updatesByFeatureId: Record<string, SubjectBalanceUpdate[]> = {};
 

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

@@ -21,6 +21,7 @@ import {
 } from "@autumn/shared";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import type { RepoContext } from "@/db/repoContext.js";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.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";
@@ -362,6 +363,7 @@ export const createFullCusProduct = async ({
 		},
 		env: customer.env,
 		logger,
+		redisV2: resolveRedisV2(),
 	};
 
 	if (

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

@@ -1,5 +1,4 @@
 import { type FullSubject, normalizedToFullSubject } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { lazyResetSubjectEntitlements } from "@/internal/customers/actions/resetCustomerEntitlementsV2/lazyResetSubjectEntitlements.js";
 import { getFullSubjectRolloutSnapshot } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
@@ -28,7 +27,7 @@ export const getCachedFullSubject = async ({
 	entityId?: string;
 	source?: string;
 }): Promise<FullSubject | undefined> => {
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 	const subjectKey = buildFullSubjectKey({
 		orgId: org.id,
 		env,
@@ -97,8 +96,7 @@ export const getCachedFullSubject = async ({
 
 	const isCustomerSubject = !entityId;
 	const balances = await getCachedFeatureBalancesBatch({
-		orgId: org.id,
-		env,
+		ctx,
 		customerId,
 		featureIds: cached.meteredFeatures,
 		customerEntitlementIdsByFeatureId: cached.customerEntitlementIdsByFeatureId,

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

@@ -1,5 +1,5 @@
 import type { AppEnv, Feature } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
+import type { Redis } from "ioredis";
 import { batchDeleteCachedFullCustomers } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.js";
 import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
@@ -22,9 +22,11 @@ type FeaturesByOrgEnv = Record<string, Feature[]>;
 export const batchInvalidateCachedFullSubjects = async ({
 	customers,
 	featuresByOrgEnv,
+	redisV2,
 }: {
 	customers: BatchInvalidateCustomer[];
 	featuresByOrgEnv: FeaturesByOrgEnv;
+	redisV2: Redis;
 }): Promise<number> => {
 	if (customers.length === 0) return 0;
 

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

@@ -1,4 +1,3 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildFullSubjectViewEpochKey } from "../../builders/buildFullSubjectViewEpochKey.js";
@@ -11,6 +10,7 @@ export const getOrInitFullSubjectViewEpoch = async ({
 	ctx: AutumnContext;
 	customerId: string;
 }): Promise<number> => {
+	const { redisV2 } = ctx;
 	const epochKey = buildFullSubjectViewEpochKey({
 		orgId: ctx.org.id,
 		env: ctx.env,

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

@@ -1,4 +1,3 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildFullSubjectViewEpochKey } from "../../builders/buildFullSubjectViewEpochKey.js";
@@ -11,6 +10,7 @@ export const incrementFullSubjectViewEpoch = async ({
 	ctx: AutumnContext;
 	customerId: string;
 }): Promise<number | null> => {
+	const { redisV2 } = ctx;
 	const epochKey = buildFullSubjectViewEpochKey({
 		orgId: ctx.org.id,
 		env: ctx.env,

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

@@ -1,4 +1,4 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
+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";
@@ -9,12 +9,14 @@ export const invalidateCustomerEntitlementBalance = async ({
 	customerId,
 	featureId,
 	customerEntitlementId,
+	redisV2,
 }: {
 	orgId: string;
 	env: string;
 	customerId: string;
 	featureId: string;
 	customerEntitlementId: string;
+	redisV2: Redis;
 }): Promise<void> => {
 	if (
 		!orgId ||

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

@@ -1,4 +1,3 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
@@ -14,7 +13,7 @@ export const invalidateCachedFullSubjectExact = async ({
 	ctx: AutumnContext;
 	source?: string;
 }): Promise<void> => {
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 	if (!customerId || redisV2.status !== "ready") return;
 
 	const subjectKey = buildFullSubjectKey({

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

@@ -1,4 +1,3 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
@@ -22,9 +21,9 @@ export const invalidateSharedBalanceFields = async ({
 	ctx: AutumnContext;
 	customerId: string;
 }): Promise<void> => {
+	const { org, env, redisV2 } = ctx;
 	if (!customerId || redisV2.status !== "ready") return;
 
-	const { org, env } = ctx;
 	const subjectKey = buildFullSubjectKey({ orgId: org.id, env, customerId });
 
 	const cachedRaw = await tryRedisRead(() => redisV2.get(subjectKey), redisV2);
@@ -46,7 +45,7 @@ async function deleteFieldsFromManifest({
 	customerId: string;
 	cachedRaw: string;
 }) {
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 
 	let manifest: CachedFullSubject;
 	try {
@@ -93,7 +92,7 @@ async function deleteAllBalanceKeys({
 	ctx: AutumnContext;
 	customerId: string;
 }) {
-	const { org, env, features, logger } = ctx;
+	const { org, env, features, logger, redisV2 } = ctx;
 	if (features.length === 0) return;
 
 	const pipeline = redisV2.pipeline();

server/src/internal/customers/cache/fullSubject/actions/partial/getCachedPartialFullSubject.ts

@@ -1,6 +1,5 @@
 import type { FullSubject } from "@autumn/shared";
 import { normalizedToFullSubject } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { lazyResetSubjectEntitlements } from "@/internal/customers/actions/resetCustomerEntitlementsV2/lazyResetSubjectEntitlements.js";
 import { getFullSubjectRolloutSnapshot } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
@@ -51,7 +50,7 @@ export const getCachedPartialFullSubject = async ({
 	featureIds: string[];
 	source?: string;
 }): Promise<FullSubject | undefined> => {
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 	const subjectKey = buildFullSubjectKey({
 		orgId: org.id,
 		env,
@@ -124,8 +123,7 @@ export const getCachedPartialFullSubject = async ({
 
 	const isCustomerSubject = !entityId;
 	const featureBalances = await getCachedFeatureBalancesBatch({
-		orgId: org.id,
-		env,
+		ctx,
 		customerId,
 		featureIds: meteredFeatureIdsToFetch,
 		customerEntitlementIdsByFeatureId: cached.customerEntitlementIdsByFeatureId,

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

@@ -1,5 +1,4 @@
 import type { NormalizedFullSubject } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildFullSubjectKey } from "../../builders/buildFullSubjectKey.js";
@@ -20,7 +19,7 @@ export const setCachedFullSubject = async ({
 	normalized: NormalizedFullSubject;
 	fetchedSubjectViewEpoch: number;
 }): Promise<SetCachedFullSubjectResult> => {
-	const { logger, org, env } = ctx;
+	const { logger, org, env, redisV2 } = ctx;
 	const { customerId, entityId } = normalized;
 
 	const cached = normalizedToCachedFullSubject({

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

@@ -1,5 +1,4 @@
 import type { Customer } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { logAlertEvent } from "@/utils/logging/logAlertEvent.js";
@@ -21,7 +20,7 @@ export const updateCachedCustomerData = async ({
 }): Promise<void> => {
 	if (Object.keys(updates).length === 0) return;
 
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 	const subjectKey = buildFullSubjectKey({
 		orgId: org.id,
 		env,

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

@@ -1,5 +1,4 @@
 import type { InsertCustomerProduct } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { updateCachedCustomerProduct } from "@/internal/customers/cusProducts/actions/cache/updateCachedCustomerProduct.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
@@ -48,7 +47,7 @@ export const updateCachedCustomerProductV2 = async ({
 		}
 
 		if (Object.keys(updates).length === 0) return null;
-		const { org, env, logger } = ctx;
+		const { org, env, logger, redisV2 } = ctx;
 		const subjectKey = buildFullSubjectKey({
 			orgId: org.id,
 			env,

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

@@ -1,5 +1,4 @@
 import type { Entity } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { updateEntityInCache } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/updateEntityInCache.js";
 import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
@@ -37,7 +36,7 @@ export const updateCachedEntityData = async ({
 		);
 	});
 
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 	const subjectKey = buildFullSubjectKey({
 		orgId: org.id,
 		env,

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

@@ -1,5 +1,4 @@
 import type { Invoice } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
 import { logAlertEvent } from "@/utils/logging/logAlertEvent.js";
@@ -38,7 +37,7 @@ export const upsertCachedInvoiceV2 = async ({
 		return null;
 	}
 
-	const { org, env, logger } = ctx;
+	const { org, env, logger, redisV2 } = ctx;
 	const subjectKey = buildFullSubjectKey({
 		orgId: org.id,
 		env,

server/src/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.ts

@@ -1,5 +1,5 @@
 import type { AggregatedFeatureBalance, SubjectBalance } from "@autumn/shared";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisRead } from "@/utils/cacheUtils/cacheUtils.js";
 import { buildSharedFullSubjectBalanceKey } from "../builders/buildSharedFullSubjectBalanceKey.js";
 import { AGGREGATED_BALANCE_FIELD } from "../config/fullSubjectCacheConfig.js";
@@ -16,12 +16,15 @@ export type FeatureBalanceResult = {
 };
 
 const readFeatureBalancesFromMaster = async ({
+	ctx,
 	balanceKey,
 	customerEntitlementIds,
 }: {
+	ctx: AutumnContext;
 	balanceKey: string;
 	customerEntitlementIds: string[];
 }): Promise<(string | null)[] | null> => {
+	const { redisV2 } = ctx;
 	const multi = redisV2.multi();
 	multi.hmget(balanceKey, ...customerEntitlementIds);
 	const multiResults = await multi.exec();
@@ -34,22 +37,21 @@ const readFeatureBalancesFromMaster = async ({
 };
 
 export const getCachedFeatureBalance = async ({
-	orgId,
-	env,
+	ctx,
 	customerId,
 	featureId,
 	customerEntitlementIds,
 	readMaster = false,
 }: {
-	orgId: string;
-	env: string;
+	ctx: AutumnContext;
 	customerId: string;
 	featureId: string;
 	customerEntitlementIds: string[];
 	readMaster?: boolean;
 }): Promise<FeatureBalanceResult | undefined> => {
+	const { org, env, redisV2 } = ctx;
 	const balanceKey = buildSharedFullSubjectBalanceKey({
-		orgId,
+		orgId: org.id,
 		env,
 		customerId,
 		featureId,
@@ -63,6 +65,7 @@ export const getCachedFeatureBalance = async ({
 		? await tryRedisRead(
 				() =>
 					readFeatureBalancesFromMaster({
+						ctx,
 						balanceKey,
 						customerEntitlementIds,
 					}),
@@ -96,15 +99,13 @@ export const getCachedFeatureBalance = async ({
 };
 
 export const getCachedFeatureBalancesBatch = async ({
-	orgId,
-	env,
+	ctx,
 	customerId,
 	featureIds,
 	customerEntitlementIdsByFeatureId,
 	includeAggregated = false,
 }: {
-	orgId: string;
-	env: string;
+	ctx: AutumnContext;
 	customerId: string;
 	featureIds: string[];
 	customerEntitlementIdsByFeatureId: Record<string, string[]>;
@@ -112,6 +113,7 @@ export const getCachedFeatureBalancesBatch = async ({
 }): Promise<FeatureBalanceResult[] | undefined> => {
 	if (featureIds.length === 0) return [];
 
+	const { org, env, redisV2 } = ctx;
 	const pipeline = redisV2.pipeline();
 	for (const featureId of featureIds) {
 		const customerEntitlementIds =
@@ -121,7 +123,7 @@ export const getCachedFeatureBalancesBatch = async ({
 			: customerEntitlementIds;
 		pipeline.hmget(
 			buildSharedFullSubjectBalanceKey({
-				orgId,
+				orgId: org.id,
 				env,
 				customerId,
 				featureId,

server/src/internal/customers/cusProducts/cusEnts/actions/cache/adjustSubjectBalanceCache.ts

@@ -1,5 +1,4 @@
 import type { RepoContext } from "@/db/repoContext.js";
-import { redisV2 } from "@/external/redis/initRedisV2.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";
@@ -24,6 +23,7 @@ export const adjustSubjectBalanceCache = async ({
 	delta: number;
 }): Promise<AdjustSubjectBalanceCacheResult | null> => {
 	try {
+		const { redisV2 } = ctx;
 		const balanceKey = buildSharedFullSubjectBalanceKey({
 			orgId: ctx.org.id,
 			env: ctx.env,

server/src/internal/customers/cusProducts/cusEnts/actions/cache/updateSubjectBalanceCache.ts

@@ -1,4 +1,3 @@
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import type { RepoContext } from "@/db/repoContext.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";
@@ -23,6 +22,7 @@ export const updateSubjectBalanceCache = async ({
 		next_reset_at?: number | null;
 	};
 }) => {
+	const { redisV2 } = ctx;
 	const balanceKey = buildSharedFullSubjectBalanceKey({
 		orgId: ctx.org.id,
 		env: ctx.env,

server/src/internal/customers/handlers/handleClearCustomerCache.ts

@@ -36,6 +36,7 @@ export const handleClearCustomerCache = createRoute({
 			await batchInvalidateCachedFullSubjects({
 				customers: customersToDelete,
 				featuresByOrgEnv,
+				redisV2: ctx.redisV2,
 			});
 		}
 

server/src/internal/entities/handlers/handleDeleteEntity/handleDeleteEntity.ts

@@ -107,13 +107,7 @@ export const handleDeleteEntity = createRoute({
 				}
 
 				await CusEntService.update({
-					ctx: {
-						db,
-						logger,
-						org,
-						env,
-						customerId: customer_id,
-					},
+					ctx: { ...ctx, customerId: customer_id },
 					id: linkedCusEnt.id,
 					updates: {
 						entities: newEntities,
@@ -123,13 +117,7 @@ export const handleDeleteEntity = createRoute({
 
 			if (!replaceable) {
 				await CusEntService.increment({
-					ctx: {
-						db,
-						logger,
-						org,
-						env,
-						customerId: customer_id,
-					},
+					ctx: { ...ctx, customerId: customer_id },
 					id: mainCusEnt.id,
 					amount: 1,
 				});

server/src/internal/features/featureActions/runClearCreditSystemCacheTask.ts

@@ -8,6 +8,7 @@ import {
 } from "@autumn/shared";
 import { and, asc, count, eq, gt, inArray } from "drizzle-orm";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.js";
 import { batchInvalidateCachedFullSubjects } from "@/internal/customers/cache/fullSubject/actions/invalidate/batchInvalidateCachedFullSubjects.js";
 import { OrgService } from "@/internal/orgs/OrgService.js";
 import type { Logger } from "../../../external/logtail/logtailUtils";
@@ -151,6 +152,7 @@ export const runClearCreditSystemCacheTask = async ({
 			const deleted = await batchInvalidateCachedFullSubjects({
 				customers: customersToDelete,
 				featuresByOrgEnv,
+				redisV2: resolveRedisV2(),
 			});
 			totalDeleted += deleted;
 		}

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

@@ -0,0 +1,10 @@
+import { z } from "zod/v4";
+
+export const RedisV2InstanceName = z.enum(["primary", "canary", "dragonfly"]);
+export type RedisV2InstanceName = z.infer<typeof RedisV2InstanceName>;
+
+export const RedisV2CacheConfigSchema = z.object({
+	activeInstance: RedisV2InstanceName.default("primary"),
+});
+
+export type RedisV2CacheConfig = z.infer<typeof RedisV2CacheConfigSchema>;

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

@@ -0,0 +1,40 @@
+import { ms } from "@autumn/shared";
+import { ADMIN_REDIS_V2_CACHE_CONFIG_KEY } from "@/external/aws/s3/adminS3Config.js";
+import { registerEdgeConfig } from "@/internal/misc/edgeConfig/edgeConfigRegistry.js";
+import { createEdgeConfigStore } from "@/internal/misc/edgeConfig/edgeConfigStore.js";
+import {
+	type RedisV2CacheConfig,
+	RedisV2CacheConfigSchema,
+	type RedisV2InstanceName,
+} from "./redisV2CacheSchemas.js";
+
+const store = createEdgeConfigStore<RedisV2CacheConfig>({
+	s3Key: ADMIN_REDIS_V2_CACHE_CONFIG_KEY,
+	schema: RedisV2CacheConfigSchema,
+	defaultValue: () => ({ activeInstance: "primary" }),
+	pollIntervalMs: ms.seconds(10),
+});
+
+registerEdgeConfig({ store });
+
+export const getActiveRedisV2Instance = (): RedisV2InstanceName =>
+	store.get().activeInstance;
+
+export const getRedisV2CacheStatus = () => store.getStatus();
+
+export const updateActiveRedisV2Instance = async ({
+	activeInstance,
+}: {
+	activeInstance: RedisV2InstanceName;
+}) => {
+	await store.writeToSource({ config: { activeInstance } });
+};
+
+/** Test-only: override the in-memory active instance without writing to S3. */
+export const _setActiveRedisV2InstanceForTesting = ({
+	activeInstance,
+}: {
+	activeInstance: RedisV2InstanceName;
+}) => {
+	store._setRuntimeConfigForTesting({ activeInstance });
+};

server/src/queue/createWorkerContext.ts

@@ -2,6 +2,7 @@ import { type AppEnv, AuthType, createdAtToVersion } from "@autumn/shared";
 import { addAppContextToLogs } from "@/utils/logging/addContextToLogs.js";
 import type { DrizzleCli } from "../db/initDrizzle.js";
 import type { Logger } from "../external/logtail/logtailUtils.js";
+import { resolveRedisV2 } from "../external/redis/resolveRedisV2.js";
 import type { AutumnContext } from "../honoUtils/HonoEnv.js";
 import { computeRolloutSnapshot } from "../internal/misc/rollouts/rolloutUtils.js";
 import { OrgService } from "../internal/orgs/OrgService.js";
@@ -39,6 +40,11 @@ export const createWorkerContext = async ({
 		createdAt: org.created_at ?? Date.now(),
 	});
 
+	const rolloutSnapshot = computeRolloutSnapshot({
+		orgId: org.id,
+		customerId,
+	});
+
 	const workerLogger = addAppContextToLogs({
 		logger: logger,
 		appContext: {
@@ -48,14 +54,15 @@ export const createWorkerContext = async ({
 			env: env,
 			auth_type: AuthType.Worker,
 			api_version: apiVersion.semver,
+			full_subject_bucket: customerId
+				? (rolloutSnapshot.customerBucket ?? undefined)
+				: undefined,
+			full_subject_rollout_enabled: customerId
+				? rolloutSnapshot.enabled
+				: undefined,
 		},
 	});
 
-	const rolloutSnapshot = computeRolloutSnapshot({
-		orgId: org.id,
-		customerId,
-	});
-
 	const ctx: AutumnContext = {
 		org,
 		env,
@@ -65,6 +72,7 @@ export const createWorkerContext = async ({
 		db,
 		dbGeneral: db,
 		logger: workerLogger,
+		redisV2: resolveRedisV2(),
 
 		id: generateId("job"),
 		timestamp: Date.now(),

server/src/utils/logging/loggerTypes.ts

@@ -31,6 +31,8 @@ export type LogAppContext = {
 	user_id?: string;
 	user_email?: string;
 	api_version: string;
+	full_subject_bucket?: number;
+	full_subject_rollout_enabled?: boolean;
 };
 
 /** Stripe webhook event context */

server/src/utils/workerUtils/createAutumnContext.ts

@@ -8,6 +8,7 @@ import {
 
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import type { Logger } from "@/external/logtail/logtailUtils.js";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { computeRolloutSnapshot } from "@/internal/misc/rollouts/rolloutUtils.js";
 import { OrgService } from "@/internal/orgs/OrgService.js";
@@ -57,6 +58,7 @@ export const createWorkerAutumnContext = async ({
 		db,
 		dbGeneral: db,
 		logger,
+		redisV2: resolveRedisV2(),
 		expand: [],
 
 		id: workerId,

server/src/workers.ts

@@ -9,6 +9,7 @@ import {
 } from "./internal/misc/edgeConfig/edgeConfigRegistry.js";
 import "./internal/misc/requestBlocks/requestBlockStore.js";
 import "./internal/misc/rollouts/rolloutConfigStore.js";
+import "./internal/misc/redisV2Cache/redisV2CacheStore.js";
 
 // Number of worker processes (defaults to CPU cores)
 const NUM_PROCESSES = process.env.NODE_ENV === "development" ? 3 : 4;

server/tests/_temp/batchDeleteCachedCustomers.test.ts

@@ -67,12 +67,12 @@ test.concurrent(`${chalk.yellowBright("batchDeleteCachedFullCustomers: leaves fu
 		featureId: TestFeature.Messages,
 	});
 
-	expect(await redisV2.exists(primarySubjectKey)).toBe(1);
-	expect(await redisV2.exists(otherSubjectKey)).toBe(1);
-	expect(await redisV2.exists(sharedBalanceKey)).toBe(1);
-	expect(await redisV2.exists(otherSharedBalanceKey)).toBe(1);
+	expect(await ctx.redisV2.exists(primarySubjectKey)).toBe(1);
+	expect(await ctx.redisV2.exists(otherSubjectKey)).toBe(1);
+	expect(await ctx.redisV2.exists(sharedBalanceKey)).toBe(1);
+	expect(await ctx.redisV2.exists(otherSharedBalanceKey)).toBe(1);
 
-	await redisV2.unlink(otherSubjectKey);
+	await ctx.redisV2.unlink(otherSubjectKey);
 
 	await batchDeleteCachedFullCustomers({
 		customers: [
@@ -81,10 +81,10 @@ test.concurrent(`${chalk.yellowBright("batchDeleteCachedFullCustomers: leaves fu
 		],
 	});
 
-	expect(await redisV2.exists(primarySubjectKey)).toBe(1);
-	expect(await redisV2.exists(otherSubjectKey)).toBe(0);
-	expect(await redisV2.exists(sharedBalanceKey)).toBe(1);
-	expect(await redisV2.exists(otherSharedBalanceKey)).toBe(1);
+	expect(await ctx.redisV2.exists(primarySubjectKey)).toBe(1);
+	expect(await ctx.redisV2.exists(otherSubjectKey)).toBe(0);
+	expect(await ctx.redisV2.exists(sharedBalanceKey)).toBe(1);
+	expect(await ctx.redisV2.exists(otherSharedBalanceKey)).toBe(1);
 
 	const primaryFromDb = await autumnV2.customers.get<ApiCustomer>(customerId, {
 		skip_cache: "true",

server/tests/balances/track/rollovers/rolloverTestUtils.ts

@@ -54,6 +54,7 @@ export const resetAndGetCusEnt = async ({
 			customerId: customer.id ?? "",
 			featureId,
 			customerEntitlementId: resetCusEnt.id,
+			redisV2: ctx.redisV2,
 		});
 
 		await clearCusEntsFromCache({

server/tests/integration/balances/lock/check-with-lock-expiry.test.ts

@@ -11,7 +11,6 @@ import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
 import chalk from "chalk";
 import { addSeconds } from "date-fns";
 import { redis } from "@/external/redis/initRedis";
-import { redisV2 } from "@/external/redis/initRedisV2";
 import { expireLock } from "@/internal/balances/finalizeLock/expireLock";
 import { buildLockReceiptKey } from "@/internal/balances/utils/lock/buildLockReceiptKey";
 import { fetchLockReceipt } from "@/internal/balances/utils/lock/fetchLockReceipt";
@@ -203,7 +202,7 @@ test.concurrent(`${chalk.yellowBright("check-lock-expiry-4: no expires_at sets T
 	});
 
 	const { lockReceiptKey, source } = await fetchLockReceipt({ ctx, lockId: customerId });
-	const redisInstance = source === "redis_v2" ? redisV2 : redis;
+	const redisInstance = source === "redis_v2" ? ctx.redisV2 : redis;
 
 	const expireAt = await redisInstance.expiretime(lockReceiptKey);
 	const expectedTtl = beforeCheck + 24 * 60 * 60;
@@ -240,7 +239,7 @@ test.concurrent(`${chalk.yellowBright("check-lock-expiry-5: expires_at set, TTL
 	});
 
 	const { lockReceiptKey, source } = await fetchLockReceipt({ ctx, lockId: customerId });
-	const redisInstance = source === "redis_v2" ? redisV2 : redis;
+	const redisInstance = source === "redis_v2" ? ctx.redisV2 : redis;
 
 	const expireAt = await redisInstance.expiretime(lockReceiptKey);
 	const expectedTtl = Math.ceil(expiresAt / 1000) + 60 * 60;

server/tests/integration/balances/reset/persist-free-overage-on.test.ts

@@ -156,6 +156,7 @@ test.concurrent(`${chalk.yellowBright("persist overage ON (cache): lazy reset de
 			customerEntitlementId: cusEnt!.id,
 			field: "balance",
 			value: -50,
+			redisV2: ctx.redisV2,
 		});
 		await expireCusEntForReset({
 			ctx,

server/tests/integration/balances/utils/lockUtils/deleteLock.ts

@@ -1,6 +1,5 @@
 import type { TestContext } from "@tests/utils/testInitUtils/createTestContext.js";
 import { redis } from "@/external/redis/initRedis.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import { buildLockReceiptKey } from "@/internal/balances/utils/lock/buildLockReceiptKey.js";
 
 export const deleteLock = async ({
@@ -17,5 +16,8 @@ export const deleteLock = async ({
 		lockKey: hashedKey,
 	});
 
-	await Promise.all([redis.del(redisReceiptKey), redisV2.del(redisReceiptKey)]);
+	await Promise.all([
+		redis.del(redisReceiptKey),
+		ctx.redisV2.del(redisReceiptKey),
+	]);
 };

server/tests/integration/db/full-subject-cache/full-subject-cache-rollout.test.ts

@@ -1,7 +1,6 @@
 import { describe, expect, test } from "bun:test";
 import ctx from "@tests/utils/testInitUtils/createTestContext.js";
 import chalk from "chalk";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import {
 	buildFullSubjectKey,
 	getCachedFullSubject,
@@ -78,7 +77,7 @@ describe(`${chalk.yellowBright("fullSubject cache rollout staleness")}`, () => {
 
 				expect(cached).toBeUndefined();
 
-				const subjectExists = await redisV2.get(
+				const subjectExists = await ctx.redisV2.get(
 					buildFullSubjectKey({
 						orgId: ctx.org.id,
 						env: ctx.env,

server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts

@@ -2,7 +2,6 @@ import { afterEach, describe, expect, test } from "bun:test";
 import { normalizedToFullSubject } from "@autumn/shared";
 import ctx from "@tests/utils/testInitUtils/createTestContext.js";
 import chalk from "chalk";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import { getOrInitFullSubjectViewEpoch } from "@/internal/customers/cache/fullSubject/actions/invalidate/getOrInitFullSubjectViewEpoch.js";
 import type { CachedFullSubject } from "@/internal/customers/cache/fullSubject/fullSubjectCacheModel.js";
 import {
@@ -36,7 +35,7 @@ const cleanupKeys = async ({
 		customerId,
 		entityId,
 	});
-	const subjectRaw = (await redisV2.get(subjectKey)) as string | null;
+	const subjectRaw = (await ctx.redisV2.get(subjectKey)) as string | null;
 	const keys = [
 		subjectKey,
 		buildFullSubjectViewEpochKey({
@@ -60,7 +59,7 @@ const cleanupKeys = async ({
 		}
 	}
 
-	await redisV2.del(...keys);
+	await ctx.redisV2.del(...keys);
 };
 
 afterEach(async () => {});
@@ -78,7 +77,7 @@ const getSharedBalanceHash = async ({
 	customerId: string;
 	featureId: string;
 }) =>
-	redisV2.hgetall(
+	ctx.redisV2.hgetall(
 		buildSharedFullSubjectBalanceKey({
 			orgId: ctx.org.id,
 			env: ctx.env,
@@ -113,7 +112,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 				});
 				expect(result).toBe("OK");
 
-				const subjectRaw = await redisV2.get(
+				const subjectRaw = await ctx.redisV2.get(
 					buildFullSubjectKey({
 						orgId: ctx.org.id,
 						env: ctx.env,
@@ -172,7 +171,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 					}),
 				});
 				expect(result).toBe("OK");
-				const subjectRaw = await redisV2.get(
+				const subjectRaw = await ctx.redisV2.get(
 					buildFullSubjectKey({
 						orgId: ctx.org.id,
 						env: ctx.env,
@@ -265,7 +264,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 
 				expect(Object.keys(hashBeforeEntityWrite).length).toBeGreaterThan(0);
 				expect(
-					await redisV2.exists(
+					await ctx.redisV2.exists(
 						`{${customerId}}:${ctx.org.id}:${ctx.env}:full_subject:shared_balances`,
 					),
 				).toBe(0);
@@ -296,7 +295,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 				}
 
 				expect(
-					await redisV2.exists(
+					await ctx.redisV2.exists(
 						`{${customerId}}:${ctx.org.id}:${ctx.env}:full_subject:shared_balances`,
 					),
 				).toBe(0);
@@ -334,7 +333,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 				});
 				expect(result).toBe("OK");
 
-				await redisV2.incr(
+				await ctx.redisV2.incr(
 					buildFullSubjectViewEpochKey({
 						orgId: ctx.org.id,
 						env: ctx.env,
@@ -396,10 +395,10 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 					env: ctx.env,
 					customerId: scenario.ids.customerId,
 				});
-				const subjectRaw = (await redisV2.get(subjectKey)) as string | null;
+				const subjectRaw = (await ctx.redisV2.get(subjectKey)) as string | null;
 				const subject = JSON.parse(subjectRaw!) as CachedFullSubject;
 
-				await redisV2.del(
+				await ctx.redisV2.del(
 					buildSharedFullSubjectBalanceKey({
 						orgId: ctx.org.id,
 						env: ctx.env,
@@ -444,7 +443,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 				});
 				expect(result).toBe("OK");
 
-				await redisV2.del(
+				await ctx.redisV2.del(
 					buildFullSubjectKey({
 						orgId: ctx.org.id,
 						env: ctx.env,
@@ -534,7 +533,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 					env: ctx.env,
 					customerId: scenario.ids.customerId,
 				});
-				const firstSubjectRaw = (await redisV2.get(subjectKey)) as
+				const firstSubjectRaw = (await ctx.redisV2.get(subjectKey)) as
 					| string
 					| null;
 				expect(firstSubjectRaw).toBeDefined();
@@ -548,7 +547,7 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => {
 				});
 				expect(secondResult).toBe("CACHE_EXISTS");
 
-				const secondSubjectRaw = (await redisV2.get(subjectKey)) as
+				const secondSubjectRaw = (await ctx.redisV2.get(subjectKey)) as
 					| string
 					| null;
 				expect(secondSubjectRaw).toEqual(firstSubjectRaw);

server/tests/integration/others/refresh-cache/refresh-cache-routes.test.ts

@@ -2,7 +2,6 @@ import { afterAll, beforeAll, describe, expect, test } from "bun:test";
 import type { TestContext } from "@tests/utils/testInitUtils/createTestContext.js";
 import { Hono } from "hono";
 import { redis } from "@/external/redis/initRedis.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import { baseMiddleware } from "@/honoMiddlewares/baseMiddleware.js";
 import {
 	REFRESH_CACHE_ROUTE_CONFIGS,
@@ -174,9 +173,9 @@ describeDb("refreshCacheMiddleware routes", () => {
 	});
 
 	afterAll(async () => {
-		const customerKeys = await redisV2.keys(`{${scenario.ids.customerId}}:*`);
+		const customerKeys = await ctx.redisV2.keys(`{${scenario.ids.customerId}}:*`);
 		if (customerKeys.length > 0) {
-			await redisV2.unlink(...customerKeys);
+			await ctx.redisV2.unlink(...customerKeys);
 		}
 		const oldCacheKey = buildFullCustomerCacheKey({
 			orgId: ctx.org.id,
@@ -242,8 +241,8 @@ describeDb("refreshCacheMiddleware routes", () => {
 		});
 
 		expect(await redis.exists(oldCacheKey)).toBe(0);
-		expect(await redisV2.exists(customerSubjectKey)).toBe(0);
-		expect(await redisV2.get(epochKey)).toBe("1");
+		expect(await ctx.redisV2.exists(customerSubjectKey)).toBe(0);
+		expect(await ctx.redisV2.get(epochKey)).toBe("1");
 
 		if (touchedEntityId) {
 			const touchedEntityKey = buildFullSubjectKey({
@@ -252,17 +251,17 @@ describeDb("refreshCacheMiddleware routes", () => {
 				customerId: scenario.ids.customerId,
 				entityId: touchedEntityId,
 			});
-			expect(await redisV2.exists(touchedEntityKey)).toBe(0);
+			expect(await ctx.redisV2.exists(touchedEntityKey)).toBe(0);
 
 			const untouchedEntityKey =
 				touchedEntityId === scenario.ids.entityIds[0]
 					? entityBSubjectKey
 					: entityASubjectKey;
-			expect(await redisV2.exists(untouchedEntityKey)).toBe(1);
+			expect(await ctx.redisV2.exists(untouchedEntityKey)).toBe(1);
 			return;
 		}
 
-		expect(await redisV2.exists(entityASubjectKey)).toBe(1);
-		expect(await redisV2.exists(entityBSubjectKey)).toBe(1);
+		expect(await ctx.redisV2.exists(entityASubjectKey)).toBe(1);
+		expect(await ctx.redisV2.exists(entityBSubjectKey)).toBe(1);
 	});
 });

server/tests/unit/balances/track-v3/trackV3Helpers.test.ts

@@ -191,7 +191,8 @@ const buildCtx = (): AutumnContext =>
 		expand: [],
 		skipCache: false,
 		extraLogs: {},
-	}) as AutumnContext;
+		redisV2: {} as never,
+	}) as unknown as AutumnContext;
 
 const buildFeatureDeduction = (): FeatureDeduction =>
 	({

server/tests/unit/full-subject-cache/invalidateCachedFullSubject.test.ts

@@ -7,7 +7,6 @@ import {
 	test,
 } from "bun:test";
 import type { TestContext } from "@tests/utils/testInitUtils/createTestContext.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import {
 	buildFullSubjectKey,
 	buildFullSubjectViewEpochKey,
@@ -58,9 +57,9 @@ describeDb("invalidateCachedFullSubject", () => {
 	});
 
 	afterEach(async () => {
-		const customerKeys = await redisV2.keys(`{${scenario.ids.customerId}}:*`);
+		const customerKeys = await ctx.redisV2.keys(`{${scenario.ids.customerId}}:*`);
 		if (customerKeys.length > 0) {
-			await redisV2.unlink(...customerKeys);
+			await ctx.redisV2.unlink(...customerKeys);
 		}
 
 		await cleanupFullSubjectScenario({ ctx, scenario });
@@ -91,11 +90,11 @@ describeDb("invalidateCachedFullSubject", () => {
 			source: "invalidateCachedFullSubjectTest",
 		});
 
-		expect(await redisV2.exists(customerKey)).toBe(0);
-		expect(await redisV2.exists(entityAKey)).toBe(0);
-		expect(await redisV2.exists(entityBKey)).toBe(1);
+		expect(await ctx.redisV2.exists(customerKey)).toBe(0);
+		expect(await ctx.redisV2.exists(entityAKey)).toBe(0);
+		expect(await ctx.redisV2.exists(entityBKey)).toBe(1);
 		expect(
-			await redisV2.get(
+			await ctx.redisV2.get(
 				buildFullSubjectViewEpochKey({
 					orgId: ctx.org.id,
 					env: ctx.env,
@@ -130,9 +129,9 @@ describeDb("invalidateCachedFullSubject", () => {
 			source: "invalidateCachedFullSubjectTest",
 		});
 
-		expect(await redisV2.exists(entityAKey)).toBe(1);
-		expect(await redisV2.exists(entityBKey)).toBe(1);
-		expect(await redisV2.get(epochKey)).toBe("1");
+		expect(await ctx.redisV2.exists(entityAKey)).toBe(1);
+		expect(await ctx.redisV2.exists(entityBKey)).toBe(1);
+		expect(await ctx.redisV2.get(epochKey)).toBe("1");
 	});
 
 	test("sibling entity cache becomes stale after direct entity invalidation", async () => {
@@ -150,7 +149,7 @@ describeDb("invalidateCachedFullSubject", () => {
 			source: "invalidateCachedFullSubjectTest",
 		});
 
-		expect(await redisV2.exists(entityBKey)).toBe(1);
+		expect(await ctx.redisV2.exists(entityBKey)).toBe(1);
 
 		const cachedEntityB = await getCachedFullSubject({
 			ctx,

server/tests/utils/cusProductUtils/resetTestUtils.ts

@@ -6,8 +6,8 @@ import {
 import { findCustomerEntitlement } from "@tests/balances/utils/findCustomerEntitlement.js";
 import type { TestContext } from "@tests/utils/testInitUtils/createTestContext.js";
 import { eq } from "drizzle-orm";
+import type { Redis } from "ioredis";
 import { redis } from "@/external/redis/initRedis.js";
-import { redisV2 } from "@/external/redis/initRedisV2.js";
 import { CusService } from "@/internal/customers/CusService.js";
 import { buildSharedFullSubjectBalanceKey } from "@/internal/customers/cache/fullSubject/builders/buildSharedFullSubjectBalanceKey.js";
 import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js";
@@ -77,6 +77,7 @@ export const setCachedSubjectBalanceField = async ({
 	customerEntitlementId,
 	field,
 	value,
+	redisV2,
 }: {
 	orgId: string;
 	env: string;
@@ -85,6 +86,7 @@ export const setCachedSubjectBalanceField = async ({
 	customerEntitlementId: string;
 	field: string;
 	value: number | string | null;
+	redisV2: Redis;
 }): Promise<void> => {
 	const balanceKey = buildSharedFullSubjectBalanceKey({
 		orgId,
@@ -160,6 +162,7 @@ export const expireCusEntForReset = async ({
 		customerEntitlementId: cusEnt.id,
 		field: "next_reset_at",
 		value: pastTime,
+		redisV2: ctx.redisV2,
 	});
 
 	return cusEnt;
@@ -221,6 +224,7 @@ export const expireAllCusEntsForReset = async ({
 			customerEntitlementId: cusEnt.id,
 			field: "next_reset_at",
 			value: pastTime,
+			redisV2: ctx.redisV2,
 		});
 	}
 

server/tests/utils/testInitUtils/createTestContext.ts

@@ -13,6 +13,7 @@ import {
 import type Stripe from "stripe";
 import { type DrizzleCli, initDrizzle } from "@/db/initDrizzle.js";
 import { createStripeCli } from "@/external/connect/createStripeCli.js";
+import { resolveRedisV2 } from "@/external/redis/resolveRedisV2.js";
 import { FeatureService } from "@/internal/features/FeatureService.js";
 import { OrgService } from "@/internal/orgs/OrgService.js";
 import {
@@ -73,6 +74,7 @@ export const createTestContext = async () => {
 		dbGeneral: db,
 		features,
 		logger,
+		redisV2: resolveRedisV2(),
 		orgSecretKey,
 
 		id: generateId("test"),

vite/src/views/admin/components/EdgeConfigTab.tsx

@@ -5,6 +5,7 @@ import { EdgeConfigDialog } from "./EdgeConfigDialog";
 import { FeatureFlagsDialog } from "./FeatureFlagsDialog";
 import { OrgLimitsDialog } from "./OrgLimitsDialog";
 import { RawEdgeConfigDialog } from "./RawEdgeConfigDialog";
+import { RedisV2CacheDialog } from "./RedisV2CacheDialog";
 import { StripeSyncDialog } from "./StripeSyncDialog";
 
 export function EdgeConfigTab() {
@@ -14,6 +15,7 @@ export function EdgeConfigTab() {
 	const [customerBlockOpen, setCustomerBlockOpen] = useState(false);
 	const [orgLimitsOpen, setOrgLimitsOpen] = useState(false);
 	const [stripeSyncOpen, setStripeSyncOpen] = useState(false);
+	const [redisV2CacheOpen, setRedisV2CacheOpen] = useState(false);
 
 	return (
 		<div className="flex flex-col gap-4">
@@ -108,6 +110,23 @@ export function EdgeConfigTab() {
 						Edit
 					</Button>
 				</div>
+
+				<div className="flex items-center justify-between border-t border-border p-4 last:border-b-0">
+					<div className="flex flex-col gap-0.5">
+						<div className="text-sm font-medium text-t1">V2 Redis Instance</div>
+						<div className="text-xs text-t3">
+							Switch the active V2 Redis between primary, canary, and dragonfly
+							at runtime.
+						</div>
+					</div>
+					<Button
+						variant="primary"
+						size="sm"
+						onClick={() => setRedisV2CacheOpen(true)}
+					>
+						Edit
+					</Button>
+				</div>
 			</div>
 
 			<FeatureFlagsDialog
@@ -139,6 +158,11 @@ export function EdgeConfigTab() {
 				open={stripeSyncOpen}
 				onOpenChange={setStripeSyncOpen}
 			/>
+
+			<RedisV2CacheDialog
+				open={redisV2CacheOpen}
+				onOpenChange={setRedisV2CacheOpen}
+			/>
 		</div>
 	);
 }

vite/src/views/admin/components/RedisV2CacheDialog.tsx

@@ -0,0 +1,214 @@
+import { useEffect, useState } from "react";
+import { toast } from "sonner";
+import { Badge } from "@/components/v2/badges/Badge";
+import { Button } from "@/components/v2/buttons/Button";
+import {
+	Dialog,
+	DialogContent,
+	DialogDescription,
+	DialogFooter,
+	DialogHeader,
+	DialogTitle,
+} from "@/components/v2/dialogs/Dialog";
+import {
+	Select,
+	SelectContent,
+	SelectItem,
+	SelectTrigger,
+	SelectValue,
+} from "@/components/v2/selects/Select";
+import { useAxiosInstance } from "@/services/useAxiosInstance";
+import { getBackendErr } from "@/utils/genUtils";
+
+const INSTANCE_OPTIONS = [
+	{
+		value: "primary",
+		label: "Primary",
+		description: "CACHE_V2_URL (or CACHE_URL fallback)",
+	},
+	{
+		value: "canary",
+		label: "Canary",
+		description: "CACHE_V2_CANARY_URL",
+	},
+	{
+		value: "dragonfly",
+		label: "Dragonfly",
+		description: "CACHE_V2_DRAGONFLY_URL",
+	},
+] as const;
+
+type InstanceName = (typeof INSTANCE_OPTIONS)[number]["value"];
+
+type RedisV2CacheConfig = {
+	activeInstance: InstanceName;
+	configHealthy: boolean;
+	configConfigured: boolean;
+	lastSuccessAt: string | null;
+	error: string | null;
+};
+
+const DEFAULT_CONFIG: RedisV2CacheConfig = {
+	activeInstance: "primary",
+	configHealthy: false,
+	configConfigured: false,
+	lastSuccessAt: null,
+	error: null,
+};
+
+export function RedisV2CacheDialog({
+	open,
+	onOpenChange,
+}: {
+	open: boolean;
+	onOpenChange: (open: boolean) => void;
+}) {
+	const axiosInstance = useAxiosInstance();
+	const [loading, setLoading] = useState(false);
+	const [saving, setSaving] = useState(false);
+	const [config, setConfig] = useState<RedisV2CacheConfig>(DEFAULT_CONFIG);
+	const [selected, setSelected] = useState<InstanceName>("primary");
+
+	useEffect(() => {
+		if (!open) return;
+
+		let cancelled = false;
+		setLoading(true);
+
+		void axiosInstance
+			.get<RedisV2CacheConfig>("/admin/redis-v2-cache-config")
+			.then(({ data }) => {
+				if (cancelled) return;
+				const merged = { ...DEFAULT_CONFIG, ...data };
+				setConfig(merged);
+				setSelected(merged.activeInstance);
+			})
+			.catch((error) => {
+				if (!cancelled) {
+					toast.error(
+						getBackendErr(error, "Failed to load redis v2 cache config"),
+					);
+				}
+			})
+			.finally(() => {
+				if (!cancelled) setLoading(false);
+			});
+
+		return () => {
+			cancelled = true;
+		};
+	}, [axiosInstance, open]);
+
+	const handleSave = async () => {
+		setSaving(true);
+		try {
+			await axiosInstance.put("/admin/redis-v2-cache-config", {
+				activeInstance: selected,
+			});
+			toast.success(`Active V2 Redis set to "${selected}"`);
+			onOpenChange(false);
+		} catch (error) {
+			toast.error(getBackendErr(error, "Failed to save redis v2 cache config"));
+		} finally {
+			setSaving(false);
+		}
+	};
+
+	const dirty = selected !== config.activeInstance;
+
+	return (
+		<Dialog open={open} onOpenChange={onOpenChange}>
+			<DialogContent className="max-w-xl bg-card">
+				<DialogHeader>
+					<DialogTitle>V2 Redis Instance</DialogTitle>
+					<DialogDescription>
+						Switch the active Redis instance used for V2 cache reads and writes.
+						Change propagates to all servers, workers, and cron jobs within ~10
+						seconds.
+					</DialogDescription>
+				</DialogHeader>
+
+				{loading ? (
+					<div className="py-8 text-center text-sm text-t3">Loading...</div>
+				) : (
+					<div className="flex flex-col gap-4">
+						<div className="flex flex-col gap-2">
+							<div className="text-xs font-medium uppercase tracking-wide text-t3">
+								Active Instance
+							</div>
+							<Select
+								value={selected}
+								onValueChange={(value) => setSelected(value as InstanceName)}
+							>
+								<SelectTrigger>
+									<SelectValue />
+								</SelectTrigger>
+								<SelectContent>
+									{INSTANCE_OPTIONS.map((option) => (
+										<SelectItem key={option.value} value={option.value}>
+											<div className="flex flex-col">
+												<span className="text-sm text-t1">{option.label}</span>
+												<span className="text-xs text-t3">
+													{option.description}
+												</span>
+											</div>
+										</SelectItem>
+									))}
+								</SelectContent>
+							</Select>
+							<div className="text-xs text-t3">
+								Currently active:{" "}
+								<span className="font-mono text-t1">
+									{config.activeInstance}
+								</span>
+							</div>
+						</div>
+
+						<div className="rounded-lg border border-border p-3 text-xs text-t3">
+							<div className="mb-2 flex items-center gap-2">
+								<Badge
+									variant="muted"
+									className={
+										config.configHealthy
+											? "border-emerald-200 bg-emerald-50 text-emerald-700"
+											: "border-amber-200 bg-amber-50 text-amber-700"
+									}
+								>
+									{config.configHealthy
+										? "Config healthy"
+										: "Config unavailable"}
+								</Badge>
+								{config.lastSuccessAt && (
+									<span>
+										Last refresh:{" "}
+										{new Date(config.lastSuccessAt).toLocaleString()}
+									</span>
+								)}
+							</div>
+							<div>
+								{config.configConfigured === false
+									? "S3 redis v2 cache config is not configured."
+									: config.error ||
+										"Instance switches propagate within ~10 seconds."}
+							</div>
+						</div>
+					</div>
+				)}
+
+				<DialogFooter>
+					<Button variant="secondary" onClick={() => onOpenChange(false)}>
+						Cancel
+					</Button>
+					<Button
+						variant="primary"
+						onClick={handleSave}
+						isLoading={saving}
+						disabled={loading || !dirty}
+					>
+						Save
+					</Button>
+				</DialogFooter>
+			</DialogContent>
+		</Dialog>
+	);
+}