From 0fde605077e7346d3f1e5fa7677afce98855a81b Mon Sep 17 00:00:00 2001 From: John Yeo Date: Tue, 24 Mar 2026 19:54:19 +0000 Subject: [PATCH] fix: batch delete customers --- .../batchDeleteFullCustomerCache.lua | 46 ------ server/src/_luaScriptsV2/luaScriptsV2.ts | 9 -- server/src/external/redis/initRedis.ts | 11 -- .../batchDeleteCachedFullCustomers.ts | 153 +++++++++++++----- .../_temp/batchDeleteCachedCustomers.test.ts | 87 ++++++++++ 5 files changed, 197 insertions(+), 109 deletions(-) delete mode 100644 server/src/_luaScriptsV2/deleteFullCustomerCache/batchDeleteFullCustomerCache.lua create mode 100644 server/tests/_temp/batchDeleteCachedCustomers.test.ts diff --git a/server/src/_luaScriptsV2/deleteFullCustomerCache/batchDeleteFullCustomerCache.lua b/server/src/_luaScriptsV2/deleteFullCustomerCache/batchDeleteFullCustomerCache.lua deleted file mode 100644 index 5dba9975d..000000000 --- a/server/src/_luaScriptsV2/deleteFullCustomerCache/batchDeleteFullCustomerCache.lua +++ /dev/null @@ -1,46 +0,0 @@ ---[[ - Batch delete multiple FullCustomer caches from Redis. - - For each customer, atomically: - 1. Checks if test guard exists (skip that customer if so) - 2. Sets the stale-write guard key - 3. Deletes the cache key - - KEYS: none (all keys passed via ARGV to support variable number of customers) - - ARGV: - [1] guardTimestamp - timestamp for all guards - [2] guardTtl - TTL in seconds for guard keys - [3] customersJson - JSON array of {testGuardKey, guardKey, cacheKey} objects - - Returns: - JSON object: { deleted: number, skipped: number } -]] - -local guardTimestamp = ARGV[1] -local guardTtl = tonumber(ARGV[2]) -local customersJson = ARGV[3] - -local customers = cjson.decode(customersJson) -local deleted = 0 -local skipped = 0 - -for _, customer in ipairs(customers) do - local testGuardKey = customer.testGuardKey - local guardKey = customer.guardKey - local cacheKey = customer.cacheKey - - -- Check test guard first - if redis.call("EXISTS", testGuardKey) == 1 then - skipped = skipped + 1 - else - -- Set stale-write guard and delete cache - redis.call("SET", guardKey, guardTimestamp, "EX", guardTtl) - local wasDeleted = redis.call("DEL", cacheKey) - if wasDeleted > 0 then - deleted = deleted + 1 - end - end -end - -return cjson.encode({ deleted = deleted, skipped = skipped }) diff --git a/server/src/_luaScriptsV2/luaScriptsV2.ts b/server/src/_luaScriptsV2/luaScriptsV2.ts index 0bb431aa5..a31e4d5f6 100644 --- a/server/src/_luaScriptsV2/luaScriptsV2.ts +++ b/server/src/_luaScriptsV2/luaScriptsV2.ts @@ -165,15 +165,6 @@ const setFullCustomerCacheScript = readFileSync( export const SET_FULL_CUSTOMER_CACHE_SCRIPT = `${FULL_CUSTOMER_KEY_BUILDERS} ${setFullCustomerCacheScript}`; -/** - * Lua script for batch deleting multiple FullCustomer caches from Redis. - * For each customer: checks test guard, sets stale-write guard, deletes cache. - */ -export const BATCH_DELETE_FULL_CUSTOMER_CACHE_SCRIPT = readFileSync( - join(DELETE_CACHE_DIR, "batchDeleteFullCustomerCache.lua"), - "utf-8", -); - // ============================================================================ // RESET CUSTOMER ENTITLEMENTS SCRIPT (deprecated — kept for backward compat) // ============================================================================ diff --git a/server/src/external/redis/initRedis.ts b/server/src/external/redis/initRedis.ts index 3962fd001..d90a2f447 100644 --- a/server/src/external/redis/initRedis.ts +++ b/server/src/external/redis/initRedis.ts @@ -16,7 +16,6 @@ import { import { ADJUST_CUSTOMER_ENTITLEMENT_BALANCE_SCRIPT, APPEND_ENTITY_TO_CUSTOMER_SCRIPT, - BATCH_DELETE_FULL_CUSTOMER_CACHE_SCRIPT, CLAIM_LOCK_RECEIPT_SCRIPT, DEDUCT_FROM_CUSTOMER_ENTITLEMENTS_SCRIPT, DELETE_FULL_CUSTOMER_CACHE_SCRIPT, @@ -179,11 +178,6 @@ const configureRedisInstance = (redisInstance: Redis): Redis => { lua: DELETE_FULL_CUSTOMER_CACHE_SCRIPT, }); - redisInstance.defineCommand("batchDeleteFullCustomerCache", { - numberOfKeys: 0, - lua: BATCH_DELETE_FULL_CUSTOMER_CACHE_SCRIPT, - }); - redisInstance.defineCommand("setFullCustomerCache", { numberOfKeys: 1, lua: SET_FULL_CUSTOMER_CACHE_SCRIPT, @@ -417,11 +411,6 @@ declare module "ioredis" { guardTtl: string, skipGuard: string, ): Promise<"SKIPPED" | "DELETED" | "NOT_FOUND">; - batchDeleteFullCustomerCache( - guardTimestamp: string, - guardTtl: string, - customersJson: string, - ): Promise; setFullCustomerCache( cacheKey: string, orgId: string, diff --git a/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.ts b/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.ts index 86565b3b8..a05bc8e7c 100644 --- a/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.ts +++ b/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/batchDeleteCachedFullCustomers.ts @@ -1,11 +1,10 @@ -import { - type Logger, - logger as loggerInstance, -} from "@/external/logtail/logtailUtils.js"; +import type { Redis } from "ioredis"; +import type { Logger } from "@/external/logtail/logtailUtils.js"; import { getConfiguredRegions, getRegionalRedis, } from "@/external/redis/initRedis.js"; +import { buildPathIndexKey } from "@/internal/customers/cache/pathIndex/pathIndexConfig.js"; import { buildFullCustomerCacheGuardKey, buildFullCustomerCacheKey, @@ -19,21 +18,111 @@ type CustomerToDelete = { customerId: string; }; +const isProductionNode = process.env.NODE_ENV === "production"; + +/** + * Per org: all keys share `{orgId}` so Redis Cluster stays in one slot per pipeline. + */ +const deleteFullCustomerCacheRowsForOrg = async ({ + regionalRedis, + orgCustomers, + guardTimestamp, +}: { + regionalRedis: Redis; + orgCustomers: CustomerToDelete[]; + guardTimestamp: string; +}): Promise<{ deleted: number; skipped: number }> => { + let skipped = 0; + let customersToProcess = orgCustomers; + + if (!isProductionNode) { + const existsPipeline = regionalRedis.pipeline(); + for (const customer of orgCustomers) { + existsPipeline.exists( + buildTestFullCustomerCacheGuardKey({ + orgId: customer.orgId, + env: customer.env, + customerId: customer.customerId, + }), + ); + } + const existsResults = await existsPipeline.exec(); + if (!existsResults) return { deleted: 0, skipped: 0 }; + + const allowed: CustomerToDelete[] = []; + for (let index = 0; index < orgCustomers.length; index++) { + const tuple = existsResults[index]; + if (!tuple) + throw new Error( + "batchDeleteCachedFullCustomers: missing EXISTS result", + ); + const [error, existsCount] = tuple; + if (error) throw error; + if (existsCount === 1) { + skipped += 1; + continue; + } + allowed.push(orgCustomers[index]!); + } + customersToProcess = allowed; + } + + if (customersToProcess.length === 0) return { deleted: 0, skipped }; + + const pipeline = regionalRedis.pipeline(); + for (const customer of customersToProcess) { + const { orgId, env, customerId } = customer; + const guardKey = buildFullCustomerCacheGuardKey({ orgId, env, customerId }); + const cacheKey = buildFullCustomerCacheKey({ orgId, env, customerId }); + const pathIndexKey = buildPathIndexKey({ orgId, env, customerId }); + pipeline.set( + guardKey, + guardTimestamp, + "EX", + FULL_CUSTOMER_CACHE_GUARD_TTL_SECONDS, + ); + pipeline.unlink(cacheKey); + pipeline.unlink(pathIndexKey); + } + + const deleteResults = await pipeline.exec(); + if (!deleteResults) return { deleted: 0, skipped }; + + let deleted = 0; + for ( + let customerIndex = 0; + customerIndex < customersToProcess.length; + customerIndex++ + ) { + const baseIndex = customerIndex * 3; + for (let commandOffset = 0; commandOffset < 3; commandOffset++) { + const tuple = deleteResults[baseIndex + commandOffset]; + if (!tuple) + throw new Error( + "batchDeleteCachedFullCustomers: missing pipeline result", + ); + const [error] = tuple; + if (error) throw error; + } + const unlinkCacheTuple = deleteResults[baseIndex + 1]; + const unlinkCount = unlinkCacheTuple![1] as number; + if (unlinkCount > 0) deleted += 1; + } + + return { deleted, skipped }; +}; + /** * Batch delete multiple FullCustomer caches across ALL regions. */ export const batchDeleteCachedFullCustomers = async ({ customers, - logger, }: { customers: CustomerToDelete[]; logger?: Logger; }): Promise => { - const log = logger || loggerInstance; - if (customers.length === 0) return 0; - // Group customers by orgId for Redis Cluster slot consistency const customersByOrg = new Map(); for (const customer of customers) { const existing = customersByOrg.get(customer.orgId) || []; @@ -44,22 +133,6 @@ export const batchDeleteCachedFullCustomers = async ({ const regions = getConfiguredRegions(); const guardTimestamp = Date.now().toString(); - // Build customers data once (shared across all regions) - const customersDataByOrg = new Map(); - for (const [orgId, orgCustomers] of customersByOrg) { - const customersData = orgCustomers.map(({ env, customerId }) => ({ - testGuardKey: buildTestFullCustomerCacheGuardKey({ - orgId, - env, - customerId, - }), - guardKey: buildFullCustomerCacheGuardKey({ orgId, env, customerId }), - cacheKey: buildFullCustomerCacheKey({ orgId, env, customerId }), - })); - customersDataByOrg.set(orgId, customersData); - } - - // Delete from all regions in parallel const regionPromises = regions.map(async (region) => { const regionalRedis = getRegionalRedis(region); @@ -68,32 +141,26 @@ export const batchDeleteCachedFullCustomers = async ({ return 0; } - const pipeline = regionalRedis.pipeline(); - for (const customersData of customersDataByOrg.values()) { - pipeline.batchDeleteFullCustomerCache( - guardTimestamp, - FULL_CUSTOMER_CACHE_GUARD_TTL_SECONDS.toString(), - JSON.stringify(customersData), - ); - } - - const results = await pipeline.exec(); let deleted = 0; - - if (results) { - for (const [error, resultJson] of results) { - if (error) throw error; - const result = JSON.parse(resultJson as string) as { deleted: number }; - deleted += result.deleted; - } + let skipped = 0; + for (const orgCustomers of customersByOrg.values()) { + const orgResult = await deleteFullCustomerCacheRowsForOrg({ + regionalRedis, + orgCustomers, + guardTimestamp, + }); + deleted += orgResult.deleted; + skipped += orgResult.skipped; } + const skipSuffix = + !isProductionNode && skipped > 0 ? `, skipped_test_guard ${skipped}` : ""; console.info( - `[batchDeleteCachedFullCustomers] ${region}: deleted ${deleted} keys, customers (${customers.length}), orgs (${customersByOrg.size})`, + `[batchDeleteCachedFullCustomers] ${region}: unlinked ${deleted} cache keys, customers (${customers.length}), orgs (${customersByOrg.size})${skipSuffix}`, ); return deleted; }); const regionDeleted = await Promise.all(regionPromises); - return regionDeleted.reduce((sum, d) => sum + d, 0); + return regionDeleted.reduce((sum, count) => sum + count, 0); }; diff --git a/server/tests/_temp/batchDeleteCachedCustomers.test.ts b/server/tests/_temp/batchDeleteCachedCustomers.test.ts new file mode 100644 index 000000000..914378987 --- /dev/null +++ b/server/tests/_temp/batchDeleteCachedCustomers.test.ts @@ -0,0 +1,87 @@ +import { expect, test } from "bun:test"; +import type { ApiCustomer } from "@autumn/shared"; +import { TestFeature } from "@tests/setup/v2Features.js"; +import { items } from "@tests/utils/fixtures/items.js"; +import { products } from "@tests/utils/fixtures/products.js"; +import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js"; +import chalk from "chalk"; +import { redis } from "@/external/redis/initRedis.js"; +import { buildPathIndexKey } from "@/internal/customers/cache/pathIndex/pathIndexConfig.js"; +import { batchDeleteCachedCustomers } from "@/internal/customers/cusUtils/apiCusCacheUtils/batchDeleteCachedCustomers.js"; +import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js"; + +test.concurrent(`${chalk.yellowBright("batchDeleteCachedCustomers: clears full customer cache + path index after V2 get")}`, async () => { + test.skipIf(redis.status !== "ready"); + + const messagesItem = items.monthlyMessages({ includedUsage: 50 }); + const freeProd = products.base({ id: "free", items: [messagesItem] }); + const otherCustomerLabel = "batch-del-cache-other"; + + const { customerId, autumnV2, ctx, otherCustomers } = await initScenario({ + customerId: "batch-del-cache-primary", + setup: [ + s.customer({ testClock: false }), + s.products({ list: [freeProd] }), + s.otherCustomers([{ id: otherCustomerLabel, paymentMethod: "success" }]), + ], + actions: [ + s.attach({ productId: freeProd.id }), + s.attach({ + productId: freeProd.id, + customerId: otherCustomerLabel, + }), + ], + }); + + const otherEntry = otherCustomers.get(otherCustomerLabel); + if (!otherEntry) throw new Error("other customer not initialized"); + + await autumnV2.customers.get(customerId); + await autumnV2.customers.get(otherEntry.id); + + const orgId = ctx.org.id; + const env = ctx.env; + + const primaryFullKey = buildFullCustomerCacheKey({ + orgId, + env, + customerId, + }); + const otherFullKey = buildFullCustomerCacheKey({ + orgId, + env, + customerId: otherEntry.id, + }); + const primaryPathKey = buildPathIndexKey({ + orgId, + env, + customerId, + }); + const otherPathKey = buildPathIndexKey({ + orgId, + env, + customerId: otherEntry.id, + }); + + expect(await redis.call("EXISTS", primaryFullKey)).toBe(1); + expect(await redis.call("EXISTS", otherFullKey)).toBe(1); + expect(await redis.call("EXISTS", primaryPathKey)).toBe(1); + expect(await redis.call("EXISTS", otherPathKey)).toBe(1); + + await batchDeleteCachedCustomers({ + customers: [ + { orgId, env, customerId }, + { orgId, env, customerId: otherEntry.id }, + ], + }); + + expect(await redis.call("EXISTS", primaryFullKey)).toBe(0); + expect(await redis.call("EXISTS", otherFullKey)).toBe(0); + expect(await redis.call("EXISTS", primaryPathKey)).toBe(0); + expect(await redis.call("EXISTS", otherPathKey)).toBe(0); + + const primaryFromDb = await autumnV2.customers.get(customerId, { + skip_cache: "true", + }); + expect(primaryFromDb.balances[TestFeature.Messages]).toBeDefined(); +});