From 8b2a07a7d99f77a77394e8323f069f111323af31 Mon Sep 17 00:00:00 2001 From: John Yeo Date: Tue, 21 Apr 2026 00:08:34 +0100 Subject: [PATCH] feat: swappable redis --- server/experiments/explainGetFull.ts | 8 +- .../fullSubjectDeduction/contextUtilsV2.lua | 1 + .../deductFromSubjectBalances.lua | 1 + .../readSubjectBalances.lua | 47 ++-- server/src/cron.ts | 9 + .../src/cron/oneoffCron/runOneOffCleanup.ts | 2 + server/src/cron/productCron/runProductCron.ts | 1 + .../resetCron/resetCustomerEntitlement.ts | 3 + server/src/db/repoContext.ts | 2 + server/src/external/aws/s3/adminS3Config.ts | 2 + server/src/external/redis/initRedisV2.ts | 35 +++ server/src/external/redis/resolveRedisV2.ts | 29 +++ server/src/external/stripe/stripeCusUtils.ts | 2 + .../honoMiddlewares/analyticsMiddleware.ts | 10 + server/src/honoMiddlewares/baseMiddleware.ts | 2 + server/src/honoUtils/HonoEnv.ts | 5 + server/src/init.ts | 1 + server/src/internal/admin/adminRouter.ts | 10 + .../admin/handleGetAdminRedisV2CacheConfig.ts | 18 ++ .../handleUpsertAdminRedisV2CacheConfig.ts | 12 + .../deductionV2/executePostgresDeductionV2.ts | 3 +- .../deductionV2/executeRedisDeductionV2.ts | 21 +- .../syncDeductionUpdatesToFullSubjectCache.ts | 3 +- .../balances/utils/lock/fetchLockReceipt.ts | 2 +- .../lockV2/buildFinalizeLockContextV2.ts | 3 +- .../utils/sync/refreshEntityAggregateCache.ts | 2 +- .../balances/utils/sync/syncItemV4.ts | 3 +- .../resetSubjectCache.ts | 3 +- .../add-product/createFullCusProduct.ts | 2 + .../actions/getCachedFullSubject.ts | 6 +- .../batchInvalidateCachedFullSubjects.ts | 4 +- .../getOrInitFullSubjectViewEpoch.ts | 2 +- .../incrementFullSubjectViewEpoch.ts | 2 +- .../invalidateCustomerEntitlementBalance.ts | 4 +- .../invalidate/invalidateFullSubjectExact.ts | 3 +- .../invalidateSharedBalanceFields.ts | 7 +- .../partial/getCachedPartialFullSubject.ts | 6 +- .../setCachedFullSubject.ts | 3 +- .../actions/updateCachedCustomerData.ts | 3 +- .../actions/updateCachedCustomerProduct.ts | 3 +- .../actions/updateCachedEntityData.ts | 3 +- .../actions/upsertCachedInvoiceV2.ts | 3 +- .../balances/getCachedFeatureBalances.ts | 24 +- .../cache/adjustSubjectBalanceCache.ts | 2 +- .../cache/updateSubjectBalanceCache.ts | 2 +- .../handlers/handleClearCustomerCache.ts | 1 + .../handleDeleteEntity/handleDeleteEntity.ts | 16 +- .../runClearCreditSystemCacheTask.ts | 2 + .../misc/redisV2Cache/redisV2CacheSchemas.ts | 10 + .../misc/redisV2Cache/redisV2CacheStore.ts | 40 ++++ server/src/queue/createWorkerContext.ts | 18 +- server/src/utils/logging/loggerTypes.ts | 2 + .../utils/workerUtils/createAutumnContext.ts | 2 + server/src/workers.ts | 1 + .../_temp/batchDeleteCachedCustomers.test.ts | 18 +- .../track/rollovers/rolloverTestUtils.ts | 1 + .../lock/check-with-lock-expiry.test.ts | 5 +- .../reset/persist-free-overage-on.test.ts | 1 + .../balances/utils/lockUtils/deleteLock.ts | 6 +- .../full-subject-cache-rollout.test.ts | 3 +- .../full-subject-cache-roundtrip.test.ts | 27 ++- .../refresh-cache-routes.test.ts | 17 +- .../balances/track-v3/trackV3Helpers.test.ts | 3 +- .../invalidateCachedFullSubject.test.ts | 21 +- .../utils/cusProductUtils/resetTestUtils.ts | 6 +- .../utils/testInitUtils/createTestContext.ts | 2 + .../views/admin/components/EdgeConfigTab.tsx | 24 ++ .../admin/components/RedisV2CacheDialog.tsx | 214 ++++++++++++++++++ 68 files changed, 601 insertions(+), 158 deletions(-) create mode 100644 server/src/external/redis/resolveRedisV2.ts create mode 100644 server/src/internal/admin/handleGetAdminRedisV2CacheConfig.ts create mode 100644 server/src/internal/admin/handleUpsertAdminRedisV2CacheConfig.ts create mode 100644 server/src/internal/misc/redisV2Cache/redisV2CacheSchemas.ts create mode 100644 server/src/internal/misc/redisV2Cache/redisV2CacheStore.ts create mode 100644 vite/src/views/admin/components/RedisV2CacheDialog.tsx diff --git a/server/experiments/explainGetFull.ts b/server/experiments/explainGetFull.ts index 936733786..955b0c3fb 100644 --- a/server/experiments/explainGetFull.ts +++ b/server/experiments/explainGetFull.ts @@ -1,9 +1,9 @@ import { AppEnv, CusProductStatus } from "@autumn/shared"; import { sql } from "drizzle-orm"; import { - initDrizzle, - prodTestCustomerId, - prodTestOrgId, + initDrizzle, + prodTestCustomerId, + prodTestOrgId, } from "./experimentEnv"; const { getFullCusQuery } = await import( "../src/internal/customers/getFullCusQuery" @@ -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 diff --git a/server/src/_luaScriptsV2/fullSubjectDeduction/contextUtilsV2.lua b/server/src/_luaScriptsV2/fullSubjectDeduction/contextUtilsV2.lua index ef5a1129d..bad7de72b 100644 --- a/server/src/_luaScriptsV2/fullSubjectDeduction/contextUtilsV2.lua +++ b/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 = { diff --git a/server/src/_luaScriptsV2/fullSubjectDeduction/deductFromSubjectBalances.lua b/server/src/_luaScriptsV2/fullSubjectDeduction/deductFromSubjectBalances.lua index 730f3c60c..123ac9ed0 100644 --- a/server/src/_luaScriptsV2/fullSubjectDeduction/deductFromSubjectBalances.lua +++ b/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 diff --git a/server/src/_luaScriptsV2/fullSubjectDeduction/readSubjectBalances.lua b/server/src/_luaScriptsV2/fullSubjectDeduction/readSubjectBalances.lua index 8cf817a05..d4e485815 100644 --- a/server/src/_luaScriptsV2/fullSubjectDeduction/readSubjectBalances.lua +++ b/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,30 +18,29 @@ 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, + customer_entitlement_ids = {}, + } + end - if entries_by_balance_key[balance_key] == nil then - entries_by_balance_key[balance_key] = { - feature_id = feature_id, - customer_entitlement_ids = {}, - } + table.insert( + entries_by_balance_key[balance_key].customer_entitlement_ids, + customer_entitlement_id + ) end - - table.insert( - entries_by_balance_key[balance_key].customer_entitlement_ids, - customer_entitlement_id - ) end end diff --git a/server/src/cron.ts b/server/src/cron.ts index b5124f084..a01f90876 100644 --- a/server/src/cron.ts +++ b/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"); diff --git a/server/src/cron/oneoffCron/runOneOffCleanup.ts b/server/src/cron/oneoffCron/runOneOffCleanup.ts index a7e3303c4..9094c1b02 100644 --- a/server/src/cron/oneoffCron/runOneOffCleanup.ts +++ b/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, diff --git a/server/src/cron/productCron/runProductCron.ts b/server/src/cron/productCron/runProductCron.ts index 197914ca7..2a278a046 100644 --- a/server/src/cron/productCron/runProductCron.ts +++ b/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; diff --git a/server/src/cron/resetCron/resetCustomerEntitlement.ts b/server/src/cron/resetCron/resetCustomerEntitlement.ts index 8194680c2..5a161b257 100644 --- a/server/src/cron/resetCron/resetCustomerEntitlement.ts +++ b/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; }; diff --git a/server/src/db/repoContext.ts b/server/src/db/repoContext.ts index 78cae9a87..229249b0e 100644 --- a/server/src/db/repoContext.ts +++ b/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; } diff --git a/server/src/external/aws/s3/adminS3Config.ts b/server/src/external/aws/s3/adminS3Config.ts index 398d0a316..ee00ac09b 100644 --- a/server/src/external/aws/s3/adminS3Config.ts +++ b/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"; diff --git a/server/src/external/redis/initRedisV2.ts b/server/src/external/redis/initRedisV2.ts index 9116361a0..9e196db49 100644 --- a/server/src/external/redis/initRedisV2.ts +++ b/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> = { + canary: process.env.CACHE_V2_CANARY_URL?.trim() || undefined, + dragonfly: process.env.CACHE_V2_DRAGONFLY_URL?.trim() || undefined, +}; + +const instancePool = new Map(); +const missingUrlWarned = new Set(); + +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 => { if (redisV2 === redis) return; diff --git a/server/src/external/redis/resolveRedisV2.ts b/server/src/external/redis/resolveRedisV2.ts new file mode 100644 index 000000000..256224890 --- /dev/null +++ b/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; +}; diff --git a/server/src/external/stripe/stripeCusUtils.ts b/server/src/external/stripe/stripeCusUtils.ts index ec2b8e88a..ba641e8a7 100644 --- a/server/src/external/stripe/stripeCusUtils.ts +++ b/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({ diff --git a/server/src/honoMiddlewares/analyticsMiddleware.ts b/server/src/honoMiddlewares/analyticsMiddleware.ts index 83932c7d1..47e563340 100644 --- a/server/src/honoMiddlewares/analyticsMiddleware.ts +++ b/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, 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, 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, }, }); diff --git a/server/src/honoMiddlewares/baseMiddleware.ts b/server/src/honoMiddlewares/baseMiddleware.ts index 4fd36fc39..3f683ffc9 100644 --- a/server/src/honoMiddlewares/baseMiddleware.ts +++ b/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, next: Next) => { db, dbGeneral, logger: childLogger, + redisV2: resolveRedisV2(), // Request info id, diff --git a/server/src/honoUtils/HonoEnv.ts b/server/src/honoUtils/HonoEnv.ts index 887ca584a..6722559f4 100644 --- a/server/src/honoUtils/HonoEnv.ts +++ b/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; diff --git a/server/src/init.ts b/server/src/init.ts index bc819970e..6fb5bf225 100644 --- a/server/src/init.ts +++ b/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, diff --git a/server/src/internal/admin/adminRouter.ts b/server/src/internal/admin/adminRouter.ts index e2dddb67f..b54bd59e3 100644 --- a/server/src/internal/admin/adminRouter.ts +++ b/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); diff --git a/server/src/internal/admin/handleGetAdminRedisV2CacheConfig.ts b/server/src/internal/admin/handleGetAdminRedisV2CacheConfig.ts new file mode 100644 index 000000000..92357a8d4 --- /dev/null +++ b/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, + }); + }, +}); diff --git a/server/src/internal/admin/handleUpsertAdminRedisV2CacheConfig.ts b/server/src/internal/admin/handleUpsertAdminRedisV2CacheConfig.ts new file mode 100644 index 000000000..89724e122 --- /dev/null +++ b/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 }); + }, +}); diff --git a/server/src/internal/balances/utils/deductionV2/executePostgresDeductionV2.ts b/server/src/internal/balances/utils/deductionV2/executePostgresDeductionV2.ts index 4903665ae..a59ce9e11 100644 --- a/server/src/internal/balances/utils/deductionV2/executePostgresDeductionV2.ts +++ b/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) { diff --git a/server/src/internal/balances/utils/deductionV2/executeRedisDeductionV2.ts b/server/src/internal/balances/utils/deductionV2/executeRedisDeductionV2.ts index 381c8edf4..a64414de5 100644 --- a/server/src/internal/balances/utils/deductionV2/executeRedisDeductionV2.ts +++ b/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 = {}; + 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( () => diff --git a/server/src/internal/balances/utils/deductionV2/syncDeductionUpdatesToFullSubjectCache.ts b/server/src/internal/balances/utils/deductionV2/syncDeductionUpdatesToFullSubjectCache.ts index f5e2ec02f..1db3cb12e 100644 --- a/server/src/internal/balances/utils/deductionV2/syncDeductionUpdatesToFullSubjectCache.ts +++ b/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 = {}; diff --git a/server/src/internal/balances/utils/lock/fetchLockReceipt.ts b/server/src/internal/balances/utils/lock/fetchLockReceipt.ts index db014fddb..6edc78d1b 100644 --- a/server/src/internal/balances/utils/lock/fetchLockReceipt.ts +++ b/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, diff --git a/server/src/internal/balances/utils/lockV2/buildFinalizeLockContextV2.ts b/server/src/internal/balances/utils/lockV2/buildFinalizeLockContextV2.ts index 66ca88824..9324257ce 100644 --- a/server/src/internal/balances/utils/lockV2/buildFinalizeLockContextV2.ts +++ b/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, diff --git a/server/src/internal/balances/utils/sync/refreshEntityAggregateCache.ts b/server/src/internal/balances/utils/sync/refreshEntityAggregateCache.ts index f596c2a5b..55c54ce07 100644 --- a/server/src/internal/balances/utils/sync/refreshEntityAggregateCache.ts +++ b/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; diff --git a/server/src/internal/balances/utils/sync/syncItemV4.ts b/server/src/internal/balances/utils/sync/syncItemV4.ts index b81568c8b..6987b026c 100644 --- a/server/src/internal/balances/utils/sync/syncItemV4.ts +++ b/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, diff --git a/server/src/internal/customers/actions/resetCustomerEntitlementsV2/resetSubjectCache.ts b/server/src/internal/customers/actions/resetCustomerEntitlementsV2/resetSubjectCache.ts index d1f0a84f9..c64e48d5f 100644 --- a/server/src/internal/customers/actions/resetCustomerEntitlementsV2/resetSubjectCache.ts +++ b/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 = {}; diff --git a/server/src/internal/customers/add-product/createFullCusProduct.ts b/server/src/internal/customers/add-product/createFullCusProduct.ts index eab317f94..a444930e0 100644 --- a/server/src/internal/customers/add-product/createFullCusProduct.ts +++ b/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 ( diff --git a/server/src/internal/customers/cache/fullSubject/actions/getCachedFullSubject.ts b/server/src/internal/customers/cache/fullSubject/actions/getCachedFullSubject.ts index c9c7ac5f3..eebbd524b 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/getCachedFullSubject.ts +++ b/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 => { - 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, diff --git a/server/src/internal/customers/cache/fullSubject/actions/invalidate/batchInvalidateCachedFullSubjects.ts b/server/src/internal/customers/cache/fullSubject/actions/invalidate/batchInvalidateCachedFullSubjects.ts index 2a19d1f6b..092616775 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/invalidate/batchInvalidateCachedFullSubjects.ts +++ b/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; export const batchInvalidateCachedFullSubjects = async ({ customers, featuresByOrgEnv, + redisV2, }: { customers: BatchInvalidateCustomer[]; featuresByOrgEnv: FeaturesByOrgEnv; + redisV2: Redis; }): Promise => { if (customers.length === 0) return 0; diff --git a/server/src/internal/customers/cache/fullSubject/actions/invalidate/getOrInitFullSubjectViewEpoch.ts b/server/src/internal/customers/cache/fullSubject/actions/invalidate/getOrInitFullSubjectViewEpoch.ts index 260fcb976..15f070063 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/invalidate/getOrInitFullSubjectViewEpoch.ts +++ b/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 => { + const { redisV2 } = ctx; const epochKey = buildFullSubjectViewEpochKey({ orgId: ctx.org.id, env: ctx.env, diff --git a/server/src/internal/customers/cache/fullSubject/actions/invalidate/incrementFullSubjectViewEpoch.ts b/server/src/internal/customers/cache/fullSubject/actions/invalidate/incrementFullSubjectViewEpoch.ts index 2a617d0ce..3966d82aa 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/invalidate/incrementFullSubjectViewEpoch.ts +++ b/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 => { + const { redisV2 } = ctx; const epochKey = buildFullSubjectViewEpochKey({ orgId: ctx.org.id, env: ctx.env, diff --git a/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateCustomerEntitlementBalance.ts b/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateCustomerEntitlementBalance.ts index d73c6254d..cdf870580 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateCustomerEntitlementBalance.ts +++ b/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 => { if ( !orgId || diff --git a/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateFullSubjectExact.ts b/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateFullSubjectExact.ts index dfe009379..512ce35e1 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateFullSubjectExact.ts +++ b/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 => { - const { org, env, logger } = ctx; + const { org, env, logger, redisV2 } = ctx; if (!customerId || redisV2.status !== "ready") return; const subjectKey = buildFullSubjectKey({ diff --git a/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateSharedBalanceFields.ts b/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateSharedBalanceFields.ts index 9eee330a3..f26069320 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/invalidate/invalidateSharedBalanceFields.ts +++ b/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 => { + 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(); diff --git a/server/src/internal/customers/cache/fullSubject/actions/partial/getCachedPartialFullSubject.ts b/server/src/internal/customers/cache/fullSubject/actions/partial/getCachedPartialFullSubject.ts index 0689ff7bb..724efed78 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/partial/getCachedPartialFullSubject.ts +++ b/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 => { - 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, diff --git a/server/src/internal/customers/cache/fullSubject/actions/setCachedFullSubject/setCachedFullSubject.ts b/server/src/internal/customers/cache/fullSubject/actions/setCachedFullSubject/setCachedFullSubject.ts index 2418353d9..24023f0b9 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/setCachedFullSubject/setCachedFullSubject.ts +++ b/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 => { - const { logger, org, env } = ctx; + const { logger, org, env, redisV2 } = ctx; const { customerId, entityId } = normalized; const cached = normalizedToCachedFullSubject({ diff --git a/server/src/internal/customers/cache/fullSubject/actions/updateCachedCustomerData.ts b/server/src/internal/customers/cache/fullSubject/actions/updateCachedCustomerData.ts index f9301e31d..046285928 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/updateCachedCustomerData.ts +++ b/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 => { 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, diff --git a/server/src/internal/customers/cache/fullSubject/actions/updateCachedCustomerProduct.ts b/server/src/internal/customers/cache/fullSubject/actions/updateCachedCustomerProduct.ts index 47cb76507..32324599e 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/updateCachedCustomerProduct.ts +++ b/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, diff --git a/server/src/internal/customers/cache/fullSubject/actions/updateCachedEntityData.ts b/server/src/internal/customers/cache/fullSubject/actions/updateCachedEntityData.ts index 81803bb90..028aa165e 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/updateCachedEntityData.ts +++ b/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, diff --git a/server/src/internal/customers/cache/fullSubject/actions/upsertCachedInvoiceV2.ts b/server/src/internal/customers/cache/fullSubject/actions/upsertCachedInvoiceV2.ts index a65bd5315..5ce4d90c6 100644 --- a/server/src/internal/customers/cache/fullSubject/actions/upsertCachedInvoiceV2.ts +++ b/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, diff --git a/server/src/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.ts b/server/src/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.ts index 49579dabf..5998a3a06 100644 --- a/server/src/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.ts +++ b/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 => { + 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; @@ -112,6 +113,7 @@ export const getCachedFeatureBalancesBatch = async ({ }): Promise => { 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, diff --git a/server/src/internal/customers/cusProducts/cusEnts/actions/cache/adjustSubjectBalanceCache.ts b/server/src/internal/customers/cusProducts/cusEnts/actions/cache/adjustSubjectBalanceCache.ts index 773f5ce1d..473f35cab 100644 --- a/server/src/internal/customers/cusProducts/cusEnts/actions/cache/adjustSubjectBalanceCache.ts +++ b/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 => { try { + const { redisV2 } = ctx; const balanceKey = buildSharedFullSubjectBalanceKey({ orgId: ctx.org.id, env: ctx.env, diff --git a/server/src/internal/customers/cusProducts/cusEnts/actions/cache/updateSubjectBalanceCache.ts b/server/src/internal/customers/cusProducts/cusEnts/actions/cache/updateSubjectBalanceCache.ts index 7acaaa06b..77f471dfd 100644 --- a/server/src/internal/customers/cusProducts/cusEnts/actions/cache/updateSubjectBalanceCache.ts +++ b/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, diff --git a/server/src/internal/customers/handlers/handleClearCustomerCache.ts b/server/src/internal/customers/handlers/handleClearCustomerCache.ts index 2791a55db..4b44b54b0 100644 --- a/server/src/internal/customers/handlers/handleClearCustomerCache.ts +++ b/server/src/internal/customers/handlers/handleClearCustomerCache.ts @@ -36,6 +36,7 @@ export const handleClearCustomerCache = createRoute({ await batchInvalidateCachedFullSubjects({ customers: customersToDelete, featuresByOrgEnv, + redisV2: ctx.redisV2, }); } diff --git a/server/src/internal/entities/handlers/handleDeleteEntity/handleDeleteEntity.ts b/server/src/internal/entities/handlers/handleDeleteEntity/handleDeleteEntity.ts index 4737249a4..827e2289b 100644 --- a/server/src/internal/entities/handlers/handleDeleteEntity/handleDeleteEntity.ts +++ b/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, }); diff --git a/server/src/internal/features/featureActions/runClearCreditSystemCacheTask.ts b/server/src/internal/features/featureActions/runClearCreditSystemCacheTask.ts index a2234e0bc..906c7d759 100644 --- a/server/src/internal/features/featureActions/runClearCreditSystemCacheTask.ts +++ b/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; } diff --git a/server/src/internal/misc/redisV2Cache/redisV2CacheSchemas.ts b/server/src/internal/misc/redisV2Cache/redisV2CacheSchemas.ts new file mode 100644 index 000000000..4e80d68b0 --- /dev/null +++ b/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; + +export const RedisV2CacheConfigSchema = z.object({ + activeInstance: RedisV2InstanceName.default("primary"), +}); + +export type RedisV2CacheConfig = z.infer; diff --git a/server/src/internal/misc/redisV2Cache/redisV2CacheStore.ts b/server/src/internal/misc/redisV2Cache/redisV2CacheStore.ts new file mode 100644 index 000000000..1263107fe --- /dev/null +++ b/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({ + 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 }); +}; diff --git a/server/src/queue/createWorkerContext.ts b/server/src/queue/createWorkerContext.ts index 99ec41034..f21b30b38 100644 --- a/server/src/queue/createWorkerContext.ts +++ b/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(), diff --git a/server/src/utils/logging/loggerTypes.ts b/server/src/utils/logging/loggerTypes.ts index 9418e381b..34e75431e 100644 --- a/server/src/utils/logging/loggerTypes.ts +++ b/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 */ diff --git a/server/src/utils/workerUtils/createAutumnContext.ts b/server/src/utils/workerUtils/createAutumnContext.ts index 1fa5e645b..9809964b0 100644 --- a/server/src/utils/workerUtils/createAutumnContext.ts +++ b/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, diff --git a/server/src/workers.ts b/server/src/workers.ts index 3668df86b..49b8bd031 100644 --- a/server/src/workers.ts +++ b/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; diff --git a/server/tests/_temp/batchDeleteCachedCustomers.test.ts b/server/tests/_temp/batchDeleteCachedCustomers.test.ts index 8663a465f..372af75c9 100644 --- a/server/tests/_temp/batchDeleteCachedCustomers.test.ts +++ b/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(customerId, { skip_cache: "true", diff --git a/server/tests/balances/track/rollovers/rolloverTestUtils.ts b/server/tests/balances/track/rollovers/rolloverTestUtils.ts index bb363258c..80ae9122b 100644 --- a/server/tests/balances/track/rollovers/rolloverTestUtils.ts +++ b/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({ diff --git a/server/tests/integration/balances/lock/check-with-lock-expiry.test.ts b/server/tests/integration/balances/lock/check-with-lock-expiry.test.ts index 73b5f3847..a55950532 100644 --- a/server/tests/integration/balances/lock/check-with-lock-expiry.test.ts +++ b/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; diff --git a/server/tests/integration/balances/reset/persist-free-overage-on.test.ts b/server/tests/integration/balances/reset/persist-free-overage-on.test.ts index 74fee25ce..c2acc749d 100644 --- a/server/tests/integration/balances/reset/persist-free-overage-on.test.ts +++ b/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, diff --git a/server/tests/integration/balances/utils/lockUtils/deleteLock.ts b/server/tests/integration/balances/utils/lockUtils/deleteLock.ts index 21d4b9885..2c2c21d47 100644 --- a/server/tests/integration/balances/utils/lockUtils/deleteLock.ts +++ b/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), + ]); }; diff --git a/server/tests/integration/db/full-subject-cache/full-subject-cache-rollout.test.ts b/server/tests/integration/db/full-subject-cache/full-subject-cache-rollout.test.ts index 1dfe3fe13..235c9023b 100644 --- a/server/tests/integration/db/full-subject-cache/full-subject-cache-rollout.test.ts +++ b/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, diff --git a/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts b/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts index 86b46878b..3e04b482c 100644 --- a/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts +++ b/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); diff --git a/server/tests/integration/others/refresh-cache/refresh-cache-routes.test.ts b/server/tests/integration/others/refresh-cache/refresh-cache-routes.test.ts index 4d2a4b675..ff3d567cd 100644 --- a/server/tests/integration/others/refresh-cache/refresh-cache-routes.test.ts +++ b/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); }); }); diff --git a/server/tests/unit/balances/track-v3/trackV3Helpers.test.ts b/server/tests/unit/balances/track-v3/trackV3Helpers.test.ts index 926ebdd55..ad8b2f21b 100644 --- a/server/tests/unit/balances/track-v3/trackV3Helpers.test.ts +++ b/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 => ({ diff --git a/server/tests/unit/full-subject-cache/invalidateCachedFullSubject.test.ts b/server/tests/unit/full-subject-cache/invalidateCachedFullSubject.test.ts index 283f6f641..1eda48b85 100644 --- a/server/tests/unit/full-subject-cache/invalidateCachedFullSubject.test.ts +++ b/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, diff --git a/server/tests/utils/cusProductUtils/resetTestUtils.ts b/server/tests/utils/cusProductUtils/resetTestUtils.ts index e413b0d85..759c4b195 100644 --- a/server/tests/utils/cusProductUtils/resetTestUtils.ts +++ b/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 => { 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, }); } diff --git a/server/tests/utils/testInitUtils/createTestContext.ts b/server/tests/utils/testInitUtils/createTestContext.ts index 5562e3f1b..af24a81ac 100644 --- a/server/tests/utils/testInitUtils/createTestContext.ts +++ b/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"), diff --git a/vite/src/views/admin/components/EdgeConfigTab.tsx b/vite/src/views/admin/components/EdgeConfigTab.tsx index 7d2376dd9..d1ff8e782 100644 --- a/vite/src/views/admin/components/EdgeConfigTab.tsx +++ b/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 (
@@ -108,6 +110,23 @@ export function EdgeConfigTab() { Edit
+ +
+
+
V2 Redis Instance
+
+ Switch the active V2 Redis between primary, canary, and dragonfly + at runtime. +
+
+ +
+ + ); } diff --git a/vite/src/views/admin/components/RedisV2CacheDialog.tsx b/vite/src/views/admin/components/RedisV2CacheDialog.tsx new file mode 100644 index 000000000..719457972 --- /dev/null +++ b/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(DEFAULT_CONFIG); + const [selected, setSelected] = useState("primary"); + + useEffect(() => { + if (!open) return; + + let cancelled = false; + setLoading(true); + + void axiosInstance + .get("/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 ( + + + + V2 Redis Instance + + Switch the active Redis instance used for V2 cache reads and writes. + Change propagates to all servers, workers, and cron jobs within ~10 + seconds. + + + + {loading ? ( +
Loading...
+ ) : ( +
+
+
+ Active Instance +
+ +
+ Currently active:{" "} + + {config.activeInstance} + +
+
+ +
+
+ + {config.configHealthy + ? "Config healthy" + : "Config unavailable"} + + {config.lastSuccessAt && ( + + Last refresh:{" "} + {new Date(config.lastSuccessAt).toLocaleString()} + + )} +
+
+ {config.configConfigured === false + ? "S3 redis v2 cache config is not configured." + : config.error || + "Instance switches propagate within ~10 seconds."} +
+
+
+ )} + + + + + + + + ); +}