Merge pull request #802 from useautumn/feat/lazy-resets

feat/lazy resets

scripts/testScripts/runTestsV2.tsx

@@ -364,8 +364,9 @@ async function runTestFile({
 		// A file is considered failed if:
 		// 1. Any individual test failed, OR
 		// 2. The process exited non-zero (e.g. module import error), OR
-		// 3. Zero tests were found (likely a silent import failure)
-		const isFailed = hasFailures || processExitedNonZero || hasNoTests;
+		// 3. Zero tests were found AND process exited non-zero (likely a silent import failure)
+		// Note: Empty files that run successfully (exit 0) are treated as passed/skipped
+		const isFailed = hasFailures || processExitedNonZero;
 
 		const finalResult: TestFileResult = {
 			file,

server/src/_luaScriptsV2/luaScriptsV2.ts

@@ -8,6 +8,7 @@ const __dirname = dirname(__filename);
 // Path to script folders
 const DEDUCT_DIR = join(__dirname, "deductFromCustomerEntitlements");
 const DELETE_CACHE_DIR = join(__dirname, "deleteFullCustomerCache");
+const RESET_DIR = join(__dirname, "resetCustomerEntitlements");
 
 // ============================================================================
 // HELPER MODULES
@@ -93,3 +94,20 @@ export const BATCH_DELETE_FULL_CUSTOMER_CACHE_SCRIPT = readFileSync(
 	join(DELETE_CACHE_DIR, "batchDeleteFullCustomerCache.lua"),
 	"utf-8",
 );
+
+// ============================================================================
+// RESET CUSTOMER ENTITLEMENTS SCRIPT
+// ============================================================================
+
+const resetMainScript = readFileSync(
+	join(RESET_DIR, "resetCustomerEntitlements.lua"),
+	"utf-8",
+);
+
+/**
+ * Lua script for atomically resetting cusEnt fields in the cached FullCustomer.
+ * Reuses luaUtils helpers for find_entitlement navigation.
+ * Skips if cache doesn't exist or cusEnt already reset (optimistic guard).
+ */
+export const RESET_CUSTOMER_ENTITLEMENTS_SCRIPT = `${LUA_UTILS}
+${resetMainScript}`;

server/src/_luaScriptsV2/resetCustomerEntitlements/resetCustomerEntitlements.lua

@@ -0,0 +1,112 @@
+--[[
+  Lua Script: Reset Customer Entitlements in Redis Cache
+
+  Atomically updates cached cusEnt fields after a DB reset.
+  Skips if cache doesn't exist or if the cusEnt's next_reset_at already
+  equals the new value (same optimistic guard as the SQL function).
+
+  Helper functions prepended via string interpolation from:
+    - luaUtils.lua (find_entitlement, safe_number, is_nil)
+
+  KEYS[1] = FullCustomer cache key
+
+  ARGV[1] = JSON params:
+    {
+      resets: [{
+        cus_ent_id: string,
+        balance: number | null,
+        additional_balance: number | null,
+        adjustment: number,
+        entities: object | null,
+        next_reset_at: number,
+        rollover_insert: { id, cus_ent_id, balance, usage, expires_at, entities } | null
+      }]
+    }
+
+  Returns JSON:
+    { "applied": { "<cus_ent_id>": true }, "skipped": ["id1"] }
+]]
+
+local cache_key = KEYS[1]
+local params = cjson.decode(ARGV[1])
+local resets = params.resets or {}
+
+-- Early return if no resets
+if #resets == 0 then
+  return cjson.encode({ applied = {}, skipped = {} })
+end
+
+-- Check if cache exists
+local key_exists = redis.call('EXISTS', cache_key)
+if key_exists == 0 then
+  return cjson.encode({ applied = {}, skipped = {}, cache_miss = true })
+end
+
+-- Read the full customer structure for entitlement path lookups
+local full_customer_json = redis.call('JSON.GET', cache_key, '.')
+if not full_customer_json then
+  return cjson.encode({ applied = {}, skipped = {}, cache_miss = true })
+end
+
+local full_customer = cjson.decode(full_customer_json)
+
+local applied = {}
+local skipped = {}
+
+for _, reset in ipairs(resets) do
+  local ent_id = reset.cus_ent_id
+  local new_next_reset_at = reset.next_reset_at
+
+  -- Find the cusEnt in the FullCustomer structure
+  local cus_ent, cus_product, ce_idx, cp_idx = find_entitlement(full_customer, ent_id)
+
+  if not cus_ent then
+    table.insert(skipped, ent_id)
+  else
+    -- Build the JSON path to this cusEnt
+    local base_path
+    local is_loose = (cp_idx == nil)
+
+    if is_loose then
+      base_path = '$.extra_customer_entitlements[' .. (ce_idx - 1) .. ']'
+    else
+      base_path = '$.customer_products[' .. (cp_idx - 1) .. '].customer_entitlements[' .. (ce_idx - 1) .. ']'
+    end
+
+    -- Optimistic guard: skip if next_reset_at already equals the new value
+    local current_reset_at = safe_number(cus_ent.next_reset_at)
+    if current_reset_at == new_next_reset_at then
+      table.insert(skipped, ent_id)
+    else
+      -- Apply reset fields using JSON.SET for absolute values
+      if not is_nil(reset.balance) then
+        redis.call('JSON.SET', cache_key, base_path .. '.balance', tostring(reset.balance))
+      end
+
+      if not is_nil(reset.additional_balance) then
+        redis.call('JSON.SET', cache_key, base_path .. '.additional_balance', tostring(reset.additional_balance))
+      end
+
+      redis.call('JSON.SET', cache_key, base_path .. '.adjustment', tostring(reset.adjustment))
+      redis.call('JSON.SET', cache_key, base_path .. '.next_reset_at', tostring(new_next_reset_at))
+
+      -- Set entities if provided (entity-scoped entitlement)
+      if not is_nil(reset.entities) then
+        redis.call('JSON.SET', cache_key, base_path .. '.entities', cjson.encode(reset.entities))
+      end
+
+      -- Increment cache_version
+      redis.call('JSON.NUMINCRBY', cache_key, base_path .. '.cache_version', 1)
+
+      -- Append rollover if provided
+      if not is_nil(reset.rollover_insert) then
+        local rollover_json = cjson.encode(reset.rollover_insert)
+        redis.call('JSON.ARRAPPEND', cache_key, base_path .. '.rollovers', rollover_json)
+      end
+
+      applied[ent_id] = true
+    end
+  end
+end
+
+return cjson.encode({ applied = applied, skipped = skipped })

server/src/db/initializeDatabaseFunctions.ts

@@ -28,6 +28,7 @@ export const initializeDatabaseFunctions = async () => {
 			"performDeduction.sql",
 			"syncBalances.sql",
 			"syncBalancesV2.sql",
+			"resetCusEnts.sql",
 		];
 
 		for (const file of sqlFiles) {

server/src/external/redis/initRedis.ts

@@ -17,6 +17,7 @@ import {
 	BATCH_DELETE_FULL_CUSTOMER_CACHE_SCRIPT,
 	DEDUCT_FROM_CUSTOMER_ENTITLEMENTS_SCRIPT,
 	DELETE_FULL_CUSTOMER_CACHE_SCRIPT,
+	RESET_CUSTOMER_ENTITLEMENTS_SCRIPT,
 	SET_FULL_CUSTOMER_CACHE_SCRIPT,
 } from "../../_luaScriptsV2/luaScriptsV2.js";
 
@@ -179,6 +180,11 @@ const configureRedisInstance = (redisInstance: Redis): Redis => {
 		lua: SET_FULL_CUSTOMER_CACHE_SCRIPT,
 	});
 
+	redisInstance.defineCommand("resetCustomerEntitlements", {
+		numberOfKeys: 1,
+		lua: RESET_CUSTOMER_ENTITLEMENTS_SCRIPT,
+	});
+
 	redisInstance.on("error", (error) => {
 		console.error(`[Redis] Connection error:`, error.message);
 	});
@@ -353,6 +359,10 @@ declare module "ioredis" {
 			serializedData: string,
 			overwrite: string,
 		): Promise<"STALE_WRITE" | "CACHE_EXISTS" | "OK">;
+		resetCustomerEntitlements(
+			cacheKey: string,
+			paramsJson: string,
+		): Promise<string>;
 	}
 }
 

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

@@ -63,10 +63,8 @@ export const resolveRevenuecatResources = async ({
 					customerId,
 				})
 			: CusService.getFull({
-					db,
+					ctx,
 					idOrInternalId: customerId,
-					orgId: org.id,
-					env,
 				}),
 	]);
 

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

@@ -39,10 +39,8 @@ const getAutumnCustomerId = async ({ ctx }: { ctx: StripeWebhookContext }) => {
 	if (!cus) return;
 
 	const fullCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: cus.internal_id,
-		orgId: ctx.org.id,
-		env: ctx.env,
 		withEntities: true,
 		withSubs: true,
 		inStatuses: RELEVANT_STATUSES,

server/src/external/vercel/handlers/handleListBillingPlans.ts

@@ -211,7 +211,8 @@ export const handleListBillingPlansPerInstall = createRoute({
 			integrationConfigurationId?: string;
 			productId?: string;
 		};
-		const { db, org, logger } = c.get("ctx");
+		const ctx = c.get("ctx");
+		const { db, org, logger } = ctx;
 
 		if (!integrationConfigurationId && !productId) {
 			return c.json(
@@ -245,10 +246,8 @@ export const handleListBillingPlansPerInstall = createRoute({
 		}
 
 		const customer = await CusService.getByVercelId({
-			db,
+			ctx,
 			vercelInstallationId: integrationConfigurationId,
-			orgId: org.id,
-			env: env as AppEnv,
 		});
 
 		// Parse metadata from query params

server/src/external/vercel/handlers/handleUpdateBillingPlan.ts

@@ -1,4 +1,4 @@
-import { type AppEnv, CustomerExpand, RecaseError } from "@autumn/shared";
+import { CustomerExpand, RecaseError } from "@autumn/shared";
 import { ErrCode } from "@shared/enums/ErrCode.js";
 import { StatusCodes } from "http-status-codes";
 import { z } from "zod/v4";
@@ -15,17 +15,16 @@ export const handleUpdateVercelBillingPlan = createRoute({
 	}),
 	// assertIdempotence: "Idempotency-Key",
 	handler: async (c) => {
-		const { orgId, env, integrationConfigurationId } = c.req.param();
-		const { db, org, features, logger } = c.get("ctx");
+		const { integrationConfigurationId } = c.req.param();
+		const ctx = c.get("ctx");
+		const { org, logger } = ctx;
 
 		const { billingPlanId } = c.req.valid("json");
 
 		// Get customer by Vercel installation ID (not by customer.id which may differ)
 		const customer = await CusService.getByVercelId({
-			db,
+			ctx,
 			vercelInstallationId: integrationConfigurationId,
-			orgId,
-			env: env as AppEnv,
 			expand: [CustomerExpand.Entities],
 		});
 
@@ -55,7 +54,7 @@ export const handleUpdateVercelBillingPlan = createRoute({
 
 			const stripeCli = createStripeCli({
 				org,
-				env: env as AppEnv,
+				env: ctx.env,
 			});
 
 			const stripeCustomer = await stripeCli.customers.retrieve(
@@ -90,16 +89,12 @@ export const handleUpdateVercelBillingPlan = createRoute({
 			if (!existingSubscription) {
 				// New subscription flow - create installation-level subscription
 				const { product } = await createVercelSubscription({
-					db,
-					org,
-					env: env as AppEnv,
+					ctx,
 					customer,
 					stripeCustomer,
 					stripeCli,
 					integrationConfigurationId,
 					billingPlanId,
-					features,
-					logger,
 					c,
 				});
 

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

@@ -51,10 +51,8 @@ export const handleDeleteInstallation = createRoute({
 
 			// 2. Get customer by Vercel installation ID (customer.id may differ from installation_id)
 			const customer = await CusService.getByVercelId({
-				db,
+				ctx,
 				vercelInstallationId: integrationConfigurationId,
-				orgId,
-				env: ctx.env,
 			});
 
 			// 3. Delete the customer/installation using the actual customer ID

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

@@ -1,7 +1,7 @@
 import {
 	cusProductToProduct,
-	mapToProductV2,
 	type FullCusProduct,
+	mapToProductV2,
 	productV2ToBasePrice,
 } from "@autumn/shared";
 import { createRoute } from "@/honoMiddlewares/routeHandler.js";
@@ -13,13 +13,11 @@ export const handleGetInstallation = createRoute({
 	handler: async (c) => {
 		const ctx = c.get("ctx");
 		const { integrationConfigurationId } = c.req.param();
-		const { db, org } = ctx;
+		const { org } = ctx;
 
 		const customer = await CusService.getByVercelId({
-			db,
+			ctx,
 			vercelInstallationId: integrationConfigurationId,
-			orgId: org.id,
-			env: ctx.env,
 		});
 
 		if (!customer) {

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

@@ -144,10 +144,8 @@ export const handleUpsertInstallation = createRoute({
 
 		if (createdCustomer) {
 			const fullCreatedCustomer = await CusService.getFull({
-				db: ctx.db,
+				ctx,
 				idOrInternalId: createdCustomer.internal_id,
-				orgId: ctx.org.id,
-				env: ctx.env,
 			});
 
 			const installation = {

server/src/external/vercel/handlers/marketplace/handleMarketplaceInvoicePaid.ts

@@ -72,10 +72,8 @@ export const handleMarketplaceInvoicePaid = async ({
 		}
 
 		const customer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: partialCustomer.internal_id,
-			orgId: org.id,
-			env,
 		});
 
 		if (!customer) {

server/src/external/vercel/handlers/marketplace/handleMarketplaceInvoidNotPaid.ts

@@ -65,10 +65,8 @@ export const handleMarketplaceInvoiceNotPaid = async ({
 		}
 
 		const customer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: partialCustomer.internal_id,
-			orgId: org.id,
-			env,
 		});
 
 		if (!customer) {

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

@@ -41,15 +41,14 @@ export const handleCreateResource = createRoute({
 	}),
 	handler: async (c) => {
 		const { orgId, env, integrationConfigurationId } = c.req.param();
-		const { db, org, features, logger } = c.get("ctx");
+		const ctx = c.get("ctx");
+		const { db, org } = ctx;
 		const { productId, name, metadata, billingPlanId } = c.req.valid("json");
 
 		// 1. Get customer by Vercel installation ID (not by customer.id which may differ)
 		const customer = await CusService.getByVercelId({
-			db,
+			ctx,
 			vercelInstallationId: integrationConfigurationId,
-			orgId,
-			env: env as AppEnv,
 			expand: [CustomerExpand.Entities],
 		});
 
@@ -100,16 +99,12 @@ export const handleCreateResource = createRoute({
 				try {
 					// 3. Create subscription (installation-level billing)
 					const { product } = await createVercelSubscription({
-						db: tx as unknown as DrizzleCli,
-						org,
-						env: env as AppEnv,
+						ctx: { ...ctx, db: tx as unknown as DrizzleCli },
 						customer,
 						stripeCustomer,
 						stripeCli,
 						integrationConfigurationId,
 						billingPlanId,
-						features,
-						logger,
 						c,
 						metadata,
 						resourceId,

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

@@ -17,7 +17,8 @@ export const handleDeleteResource = createRoute({
 	handler: async (c) => {
 		const { orgId, env, integrationConfigurationId, resourceId } =
 			c.req.param();
-		const { db, org } = c.get("ctx");
+		const ctx = c.get("ctx");
+		const { db, org } = ctx;
 		const stripeCli = createStripeCli({ org, env: env as AppEnv });
 
 		await VercelResourceService.delete({
@@ -45,10 +46,8 @@ export const handleDeleteResource = createRoute({
 		});
 
 		const customer = await CusService.getByVercelId({
-			db,
+			ctx,
 			vercelInstallationId: integrationConfigurationId,
-			orgId,
-			env: env as AppEnv,
 		});
 
 		customer?.customer_products.forEach(async (x) => {

server/src/external/vercel/handlers/transfers/handleResourceTransfers.ts

@@ -1,5 +1,5 @@
-import { ErrCode } from "@shared/enums/ErrCode.js";
 import { RecaseError } from "@autumn/shared";
+import { ErrCode } from "@shared/enums/ErrCode.js";
 import { StatusCodes } from "http-status-codes";
 import { createRoute } from "@/honoMiddlewares/routeHandler.js";
 

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

@@ -1,19 +1,15 @@
 import {
-	type AppEnv,
-	type Feature,
 	type FullCustomer,
 	type FullProduct,
-	type Organization,
 	RecaseError,
 } from "@autumn/shared";
 import { ErrCode } from "@shared/enums/ErrCode.js";
 import type { Context } from "hono";
 import { StatusCodes } from "http-status-codes";
 import type Stripe from "stripe";
-import type { DrizzleCli } from "@/db/initDrizzle.js";
 import { getCusPaymentMethod } from "@/external/stripe/stripeCusUtils.js";
 import { getStripeSubItems2 } from "@/external/stripe/stripeSubUtils/getStripeSubItems.js";
-import type { HonoEnv } from "@/honoUtils/HonoEnv.js";
+import type { AutumnContext, HonoEnv } from "@/honoUtils/HonoEnv.js";
 import { createStripeSub2 } from "@/internal/customers/attach/attachFunctions/addProductFlow/createStripeSub2.js";
 import { handleFreeProduct } from "@/internal/customers/attach/attachFunctions/addProductFlow/handleAddProduct.js";
 import { CusService } from "@/internal/customers/CusService.js";
@@ -40,34 +36,27 @@ import {
  * Future: handleUpdateBillingPlan will also handle upgrades/downgrades when subscription exists
  */
 export const createVercelSubscription = async ({
-	db,
-	org,
-	env,
+	ctx,
 	customer,
 	stripeCustomer,
 	stripeCli,
 	integrationConfigurationId,
 	billingPlanId,
-	features,
-	logger,
 	c,
 	metadata,
 	resourceId,
 }: {
-	db: DrizzleCli;
-	org: Organization;
-	env: AppEnv;
+	ctx: AutumnContext;
 	customer: FullCustomer;
 	stripeCustomer: Stripe.Customer;
 	stripeCli: Stripe;
 	integrationConfigurationId: string;
 	billingPlanId: string;
-	features: Feature[];
-	logger: any;
 	c: Context<HonoEnv>;
 	metadata?: Record<string, any>;
 	resourceId?: string;
 }): Promise<{ product: FullProduct }> => {
+	const { db, org, env, features, logger } = ctx;
 	// 1. Check for existing non-incomplete subscription (only allow one per installation)
 	const existingSubscription = stripeCustomer.subscriptions?.data.find(
 		(s) =>
@@ -105,10 +94,8 @@ export const createVercelSubscription = async ({
 	}
 
 	const refreshedCustomer = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customer.internal_id,
-		orgId: org.id,
-		env,
 	});
 
 	// 3. Get custom payment method (created in handleUpsertInstallation)

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

@@ -110,10 +110,8 @@ export const handleProductsUpdated = async ({
 
 	const fullProduct: FullProduct = cusProductToProduct({ cusProduct });
 	const fullCus = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: data.customerId || data.internalCustomerId,
-		orgId: org.id,
-		env: env,
 		entityId: cusProduct.internal_entity_id || undefined,
 		allowNotFound: true,
 	});

server/src/internal/analytics/internalHandlers/handleInternalAggregateEvents.ts

@@ -45,10 +45,8 @@ export const handleInternalAggregateEvents = createRoute({
 		} else {
 			// Customer ID provided, fetch customer data
 			customer = await CusService.getFull({
-				db,
+				ctx,
 				idOrInternalId: customer_id,
-				orgId: org.id,
-				env,
 				withSubs: true,
 			});
 

server/src/internal/analytics/internalHandlers/handleInternalListRawEvents.ts

@@ -29,10 +29,8 @@ export const handleInternalListRawEvents = createRoute({
 		} else {
 			// Customer ID provided, fetch customer data
 			customer = await CusService.getFull({
-				db,
+				ctx,
 				idOrInternalId: customer_id,
-				orgId: org.id,
-				env,
 				withSubs: true,
 			});
 

server/src/internal/api/check/getCheckPreview.ts

@@ -41,10 +41,8 @@ export const getCheckPreview = async ({
 
 	const { db, org, env, features: allFeatures } = ctx;
 	const fullCus = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: org.id,
-		env,
 		entityId,
 	});
 

server/src/internal/balances/handlers/handleCreateBalance.ts

@@ -23,10 +23,8 @@ export const handleCreateBalance = createRoute({
 		}
 
 		const fullCustomer = await CusService.getFull({
-			db: ctx.db,
+			ctx,
 			idOrInternalId: customer_id,
-			orgId: org.id,
-			env: env,
 			entityId: entity_id,
 			withEntities: true,
 		});

server/src/internal/balances/handlers/handleListBalances.ts

@@ -21,10 +21,8 @@ export const handleListBalances = createRoute({
 		const { customer_id } = c.req.valid("query");
 
 		const fullCus = await CusService.getFull({
-			db: ctx.db,
+			ctx,
 			idOrInternalId: customer_id,
-			orgId: ctx.org.id,
-			env: ctx.env,
 		});
 
 		if (!fullCus) {

server/src/internal/balances/handlers/handleUpdateBalance.ts

@@ -44,10 +44,8 @@ export const handleUpdateBalance = createRoute({
 			});
 
 			const fullCus = await CusService.getFull({
-				db: ctx.db,
+				ctx,
 				idOrInternalId: params.customer_id,
-				orgId: ctx.org.id,
-				env: ctx.env,
 				entityId: params.entity_id,
 				withEntities: true,
 			});

server/src/internal/balances/setUsage/handleSetUsage.ts

@@ -12,11 +12,9 @@ export const handleSetUsage = createRoute({
 		const ctx = c.get("ctx");
 
 		const fullCustomer = await CusService.getFull({
-			db: ctx.db,
+			ctx,
 			idOrInternalId: body.customer_id,
 			entityId: body.entity_id,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			inStatuses: ACTIVE_STATUSES,
 		});
 

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

@@ -57,10 +57,8 @@ export const executePostgresDeduction = async ({
 	// Need to getOrCreateCustomer here too...
 	if (!fullCustomer) {
 		fullCustomer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customerId,
-			orgId: org.id,
-			env,
 			inStatuses: ACTIVE_STATUSES,
 			entityId,
 			withSubs: true,

server/src/internal/balances/utils/sql/client.ts

@@ -0,0 +1,64 @@
+import type { EntityBalance, Rollover } from "@autumn/shared";
+import { sql } from "drizzle-orm";
+import type { DrizzleCli } from "@/db/initDrizzle.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+
+export type ResetCusEntParam = {
+	cus_ent_id: string;
+	balance: number | null;
+	additional_balance: number | null;
+	adjustment: number;
+	entities: Record<string, EntityBalance> | null;
+	next_reset_at: number;
+	rollover_insert: Pick<
+		Rollover,
+		"id" | "cus_ent_id" | "balance" | "usage" | "expires_at" | "entities"
+	> | null;
+};
+
+export type AppliedCusEntReset = {
+	balance: number;
+	additional_balance: number;
+	adjustment: number;
+	entities: Record<string, EntityBalance> | null;
+	next_reset_at: number;
+	cache_version: number;
+	rollover: Pick<
+		Rollover,
+		"id" | "cus_ent_id" | "balance" | "usage" | "expires_at" | "entities"
+	> | null;
+};
+
+type ResetCusEntsResult = {
+	applied: Record<string, AppliedCusEntReset>;
+	skipped: string[];
+};
+
+/** Calls the `reset_customer_entitlements` PL/pgSQL function atomically. */
+export const resetCusEnts = async ({
+	ctx,
+	resets,
+}: {
+	ctx: AutumnContext;
+	resets: ResetCusEntParam[];
+}): Promise<ResetCusEntsResult> => {
+	const { db } = ctx;
+	if (resets.length === 0) {
+		return { applied: {}, skipped: [] };
+	}
+
+	const result = await db.execute(
+		sql`SELECT * FROM reset_customer_entitlements(${JSON.stringify({
+			resets,
+		})}::jsonb)`,
+	);
+
+	const raw = result[0]?.reset_customer_entitlements as
+		| ResetCusEntsResult
+		| undefined;
+
+	return {
+		applied: raw?.applied ?? {},
+		skipped: raw?.skipped ?? [],
+	};
+};

server/src/internal/balances/utils/sql/resetCusEnts.sql

@@ -0,0 +1,139 @@
+-- Atomically reset customer entitlements that have passed their next_reset_at.
+-- Uses per-row locking + optimistic check: only resets a cusEnt if its
+-- next_reset_at does NOT already equal the new value (prevents double-resets).
+--
+-- Params (JSONB):
+--   resets: array of objects with:
+--     - cus_ent_id: text
+--     - balance: numeric (null if entity-scoped)
+--     - additional_balance: numeric (null if entity-scoped)
+--     - adjustment: numeric
+--     - entities: jsonb (null if non-entity)
+--     - next_reset_at: bigint (new next_reset_at value)
+--     - rollover_insert: jsonb object or null, with fields:
+--         id, cus_ent_id, balance, usage, expires_at, entities
+--
+-- Returns JSONB:
+--   {
+--     "applied": {
+--       "<cus_ent_id>": {
+--         "balance": number,
+--         "additional_balance": number,
+--         "adjustment": number,
+--         "entities": jsonb,
+--         "next_reset_at": number,
+--         "cache_version": number,
+--         "rollover": jsonb or null
+--       }
+--     },
+--     "skipped": ["id1", "id2"]
+--   }
+--
+DROP FUNCTION IF EXISTS reset_customer_entitlements(jsonb);
+
+CREATE FUNCTION reset_customer_entitlements(params jsonb)
+RETURNS jsonb
+LANGUAGE plpgsql
+AS $$
+DECLARE
+  resets_param jsonb := params->'resets';
+
+  reset_obj jsonb;
+  ent_id text;
+  new_balance numeric;
+  new_additional_balance numeric;
+  new_adjustment numeric;
+  new_entities jsonb;
+  new_next_reset_at bigint;
+  rollover_obj jsonb;
+
+  db_next_reset_at bigint;
+  updated_row record;
+
+  applied_json jsonb := '{}'::jsonb;
+  skipped_ids jsonb := '[]'::jsonb;
+BEGIN
+  IF resets_param IS NULL OR jsonb_array_length(resets_param) = 0 THEN
+    RETURN jsonb_build_object('applied', '{}'::jsonb, 'skipped', '[]'::jsonb);
+  END IF;
+
+  FOR reset_obj IN SELECT * FROM jsonb_array_elements(resets_param)
+  LOOP
+    ent_id := reset_obj->>'cus_ent_id';
+    new_balance := (reset_obj->>'balance')::numeric;
+    new_additional_balance := (reset_obj->>'additional_balance')::numeric;
+    new_adjustment := (reset_obj->>'adjustment')::numeric;
+    new_entities := reset_obj->'entities';
+    new_next_reset_at := (reset_obj->>'next_reset_at')::bigint;
+    rollover_obj := reset_obj->'rollover_insert';
+
+    -- Lock and read the single row
+    SELECT ce.next_reset_at INTO db_next_reset_at
+    FROM customer_entitlements ce
+    WHERE ce.id = ent_id
+    FOR UPDATE;
+
+    -- Skip if the row doesn't exist (stale ID from a deleted cusEnt)
+    IF NOT FOUND THEN
+      skipped_ids := skipped_ids || to_jsonb(ent_id);
+      CONTINUE;
+    END IF;
+
+    -- Optimistic lock: skip if next_reset_at already equals the new value
+    IF db_next_reset_at IS NOT DISTINCT FROM new_next_reset_at THEN
+      skipped_ids := skipped_ids || to_jsonb(ent_id);
+      CONTINUE;
+    END IF;
+
+    -- Apply the reset update and capture the updated row
+    UPDATE customer_entitlements ce
+    SET
+      balance = COALESCE(new_balance, ce.balance),
+      additional_balance = COALESCE(new_additional_balance, ce.additional_balance),
+      adjustment = COALESCE(new_adjustment, ce.adjustment),
+      entities = COALESCE(new_entities, ce.entities),
+      next_reset_at = new_next_reset_at,
+      cache_version = COALESCE(ce.cache_version, 0) + 1
+    WHERE ce.id = ent_id
+    RETURNING ce.balance, ce.additional_balance, ce.adjustment, ce.entities,
+              ce.next_reset_at, ce.cache_version
+    INTO updated_row;
+
+    -- Insert rollover row if provided
+    IF rollover_obj IS NOT NULL AND rollover_obj != 'null'::jsonb THEN
+      INSERT INTO rollovers (id, cus_ent_id, balance, usage, expires_at, entities)
+      VALUES (
+        rollover_obj->>'id',
+        rollover_obj->>'cus_ent_id',
+        (rollover_obj->>'balance')::numeric,
+        (rollover_obj->>'usage')::numeric,
+        (rollover_obj->>'expires_at')::numeric,
+        COALESCE(rollover_obj->'entities', '{}'::jsonb)
+      );
+    END IF;
+
+    -- Record the latest state of the updated cusEnt
+    applied_json := jsonb_set(
+      applied_json,
+      ARRAY[ent_id],
+      jsonb_build_object(
+        'balance', updated_row.balance,
+        'additional_balance', updated_row.additional_balance,
+        'adjustment', updated_row.adjustment,
+        'entities', updated_row.entities,
+        'next_reset_at', updated_row.next_reset_at,
+        'cache_version', updated_row.cache_version,
+        'rollover', CASE
+          WHEN rollover_obj IS NOT NULL AND rollover_obj != 'null'::jsonb THEN rollover_obj
+          ELSE NULL
+        END
+      )
+    );
+  END LOOP;
+
+  RETURN jsonb_build_object(
+    'applied', applied_json,
+    'skipped', skipped_ids
+  );
+END;
+$$;

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

@@ -196,14 +196,13 @@ export const syncItemV3 = async ({
 	ctx: AutumnContext;
 	payload: SyncItemV3;
 }): Promise<void> => {
-	const { customerId, orgId, env, region, cusEntIds, rolloverIds } = payload;
+	const { customerId, region, cusEntIds, rolloverIds } = payload;
 	const { db, logger } = ctx;
 
 	const redisInstance = region ? getRegionalRedis(region) : undefined;
 
 	const fullCustomer = await getCachedFullCustomer({
-		orgId,
-		env,
+		ctx,
 		customerId,
 		redisInstance,
 	});

server/src/internal/billing/v2/setup/setupFullCustomerContext.ts

@@ -9,14 +9,11 @@ export const setupFullCustomerContext = async ({
 	ctx: AutumnContext;
 	params: { customer_id: string; entity_id?: string };
 }) => {
-	const { db, org, env } = ctx;
 	const { customer_id: customerId } = params;
 
 	const fullCustomer = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: org.id,
-		env,
 		withSubs: true,
 		withEntities: true,
 		entityId: params.entity_id ?? undefined,

server/src/internal/billing/v2/workflows/sendProductsUpdated/sendProductsUpdated.ts

@@ -42,10 +42,8 @@ export const sendProductsUpdated = async ({
 
 	// Fetch FullCustomer
 	const fullCustomer = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId ?? "",
-		orgId: org.id,
-		env,
 		withEntities: true,
 		withSubs: true,
 		allowNotFound: true,

server/src/internal/billing/v2/workflows/verifyCacheConsistency/verifyCacheConsistency.ts

@@ -79,8 +79,7 @@ verifyCacheConsistency?.task({
 
 			// Get from cache (now using full customer cache)
 			const cachedFullCustomer = await getCachedFullCustomer({
-				orgId: autumnContext.org.id,
-				env: autumnContext.env,
+				ctx: autumnContext,
 				customerId,
 			});
 
@@ -97,10 +96,8 @@ verifyCacheConsistency?.task({
 
 			// Get fresh from DB
 			const fullCus = await CusService.getFull({
-				db,
+				ctx: autumnContext,
 				idOrInternalId: customerId,
-				orgId: autumnContext.org.id,
-				env: autumnContext.env,
 				withEntities: true,
 				withSubs: true,
 				expand: [CustomerExpand.Invoices],

server/src/internal/customers/CusBatchService.ts

@@ -1,36 +1,32 @@
 import {
 	AffectedResource,
 	type ApiCustomerV5,
-	type AppEnv,
 	applyResponseVersionChanges,
 	type CusProductStatus,
 	CustomerExpand,
 	type CustomerLegacyData,
 	type FullCustomer,
 	type ListCustomersV2Params,
-	type Organization,
 	RELEVANT_STATUSES,
 } from "@autumn/shared";
-import type { DrizzleCli } from "@/db/initDrizzle.js";
-import type { RequestContext } from "@/honoUtils/HonoEnv.js";
+import * as Sentry from "@sentry/bun";
+import type { AutumnContext, RequestContext } from "@/honoUtils/HonoEnv.js";
+import { triggerBatchResetCustomerEntitlements } from "./actions/resetCustomerEntitlements/triggerBatchResetCustomerEntitlements.js";
 import { getApiCustomerBase } from "./cusUtils/apiCusUtils/getApiCustomerBase.js";
 import { getPaginatedFullCusQuery } from "./getFullCusQuery.js";
 
 export class CusBatchService {
 	static async getByInternalIds({
-		db,
-		org,
-		env,
+		ctx,
 		internalCustomerIds,
 	}: {
-		db: DrizzleCli;
-		org: Organization;
-		env: AppEnv;
+		ctx: AutumnContext;
 		internalCustomerIds: string[];
 	}) {
+		const { org, env, db } = ctx;
 		const query = getPaginatedFullCusQuery({
-			orgId: org.id,
-			env,
+			orgId: ctx.org.id,
+			env: ctx.env,
 			includeInvoices: true,
 			withEntities: true,
 			withTrialsUsed: false,
@@ -40,8 +36,20 @@ export class CusBatchService {
 			internalCustomerIds,
 		});
 		const results = await db.execute(query);
+		const fullCustomers = results as unknown as FullCustomer[];
 
-		return results as unknown as FullCustomer[];
+		// Fire-and-forget: queue SQS job for any stale entitlement resets
+		triggerBatchResetCustomerEntitlements({
+			ctx,
+			fullCustomers,
+		}).catch((err) => {
+			ctx.logger.error(
+				`[CusBatchService.getByInternalIds] batch reset failed: ${err}`,
+			);
+			Sentry.captureException(err);
+		});
+
+		return fullCustomers;
 	}
 
 	static async getPage({
@@ -75,12 +83,14 @@ export class CusBatchService {
 		});
 		const results = await ctx.db.execute(sqlQuery);
 		const finals = [];
+		const fullCustomers: FullCustomer[] = [];
 
 		for (const result of results) {
 			try {
 				const normalizedCustomer =
 					CusBatchService.normalizeCustomerData(result);
 				const fullCus = normalizedCustomer as FullCustomer;
+				fullCustomers.push(fullCus);
 
 				// Since we already have fullCus from DB, call getApiCustomerBase directly
 				const { apiCustomer: baseCustomer, legacyData } =
@@ -104,10 +114,19 @@ export class CusBatchService {
 
 				finals.push(versionedCustomer);
 			} catch (error) {
-				console.error(`Failed to process customer ${result.id}:`, error);
+				ctx.logger.error(`Failed to process customer ${result.id}: ${error}`);
 			}
 		}
 
+		// Fire-and-forget: queue SQS job for any stale entitlement resets
+		triggerBatchResetCustomerEntitlements({
+			ctx,
+			fullCustomers,
+		}).catch((err) => {
+			ctx.logger.error("[CusBatchService.getPage] batch reset failed:", err);
+			Sentry.captureException(err);
+		});
+
 		return finals;
 	}
 

server/src/internal/customers/CusService.ts

@@ -25,6 +25,7 @@ import {
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { withSpan } from "../analytics/tracer/spanUtils.js";
+import { resetCustomerEntitlements } from "./actions/resetCustomerEntitlements/resetCustomerEntitlements.js";
 import { RELEVANT_STATUSES } from "./cusProducts/CusProductService.js";
 import { getFullCusQuery } from "./getFullCusQuery.js";
 
@@ -32,10 +33,8 @@ import { getFullCusQuery } from "./getFullCusQuery.js";
 
 export class CusService {
 	static async getFull({
-		db,
+		ctx,
 		idOrInternalId,
-		orgId,
-		env,
 		inStatuses = RELEVANT_STATUSES,
 		withEntities = false,
 		entityId,
@@ -44,10 +43,8 @@ export class CusService {
 		allowNotFound = false,
 		withEvents = false,
 	}: {
-		db: DrizzleCli;
+		ctx: AutumnContext;
 		idOrInternalId: string;
-		orgId: string;
-		env: AppEnv;
 		inStatuses?: CusProductStatus[];
 		withEntities?: boolean;
 		entityId?: string;
@@ -56,6 +53,9 @@ export class CusService {
 		allowNotFound?: boolean;
 		withEvents?: boolean;
 	}): Promise<FullCustomer> {
+		const { db, org, env } = ctx;
+		const orgId = org.id;
+
 		const includeInvoices = expand?.includes(CustomerExpand.Invoices) || false;
 		const withTrialsUsed = expand?.includes(CustomerExpand.TrialsUsed) || false;
 
@@ -109,7 +109,15 @@ export class CusService {
 					}
 				}
 
-				return data as FullCustomer;
+				const fullCus = data as FullCustomer;
+
+				// Lazy reset stale entitlements (mutates fullCus in-memory + writes DB)
+				await resetCustomerEntitlements({
+					fullCus,
+					ctx,
+				});
+
+				return fullCus;
 			},
 		});
 	}
@@ -418,25 +426,23 @@ export class CusService {
 	}
 
 	static async getByVercelId({
-		db,
+		ctx,
 		vercelInstallationId,
-		orgId,
-		env,
 		expand,
 	}: {
-		db: DrizzleCli;
+		ctx: AutumnContext;
 		vercelInstallationId: string;
-		orgId: string;
-		env: AppEnv;
 		expand?: (CustomerExpand | EntityExpand)[];
 	}) {
+		const { db, org, env } = ctx;
+
 		// This assumes the "processors" column is a JSONB object that can have a "vercel" object with "installation_id"
 		const results = await db
 			.select()
 			.from(customers as unknown as Table)
 			.where(
 				and(
-					eq(customers.org_id, orgId),
+					eq(customers.org_id, org.id),
 					eq(customers.env, env),
 					// This JSON path works for Postgres jsonb column
 					// Check for 'vercel.installation_id' inside the processors JSONB column
@@ -449,10 +455,8 @@ export class CusService {
 		if (!customer) return null;
 		else {
 			return (await CusService.getFull({
-				db,
+				ctx,
 				idOrInternalId: customer.internal_id,
-				orgId,
-				env,
 				expand,
 			})) as FullCustomer;
 		}

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

@@ -57,20 +57,18 @@ export const executeAutumnCreateCustomerPlan = async ({
 
 	if (error) {
 		if (isUniqueConstraintError(error)) {
-			logger.info(
-				`Customer already exists, returning existing: ${fullCustomer.id || fullCustomer.email}`,
-			);
-			const existingCustomer = await CusService.getFull({
-				db,
-				idOrInternalId: fullCustomer.id || fullCustomer.internal_id,
-				orgId: ctx.org.id,
-				env: ctx.env,
-				withEntities: true,
-				withSubs: true,
-				expand: [CustomerExpand.Invoices],
-			});
-			context.fullCustomer = existingCustomer;
-			return { type: "existing" };
+		logger.info(
+			`Customer already exists, returning existing: ${fullCustomer.id || fullCustomer.email}`,
+		);
+		const existingCustomer = await CusService.getFull({
+			ctx,
+			idOrInternalId: fullCustomer.id || fullCustomer.internal_id,
+			withEntities: true,
+			withSubs: true,
+			expand: [CustomerExpand.Invoices],
+		});
+		context.fullCustomer = existingCustomer;
+		return { type: "existing" };
 		}
 		throw error;
 	}
@@ -80,10 +78,8 @@ export const executeAutumnCreateCustomerPlan = async ({
 			`Customer already exists (claimed or existing): ${fullCustomer.id || fullCustomer.internal_id}`,
 		);
 		const existingCustomer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: fullCustomer.internal_id,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			withEntities: true,
 			withSubs: true,
 			expand: [CustomerExpand.Invoices],

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

@@ -0,0 +1,76 @@
+import type {
+	FullCustomer,
+	FullCustomerEntitlement,
+	Rollover,
+} from "@autumn/shared";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { RolloverService } from "@/internal/customers/cusProducts/cusEnts/cusRollovers/RolloverService.js";
+import type { ProcessResetResult } from "./processReset.js";
+
+/** Find a cusEnt on the FullCustomer by ID. */
+const findCusEnt = ({
+	fullCus,
+	cusEntId,
+}: {
+	fullCus: FullCustomer;
+	cusEntId: string;
+}): FullCustomerEntitlement | null => {
+	for (const cusProduct of fullCus.customer_products) {
+		for (const cusEnt of cusProduct.customer_entitlements) {
+			if (cusEnt.id === cusEntId) return cusEnt;
+		}
+	}
+	for (const cusEnt of fullCus.extra_customer_entitlements || []) {
+		if (cusEnt.id === cusEntId) return cusEnt;
+	}
+	return null;
+};
+
+/**
+ * Applies computed reset values to in-memory FullCustomer for all cusEnts,
+ * and runs rollover max-clearing only for DB-applied (non-skipped) ones.
+ */
+export const applyResetResults = async ({
+	ctx,
+	fullCus,
+	computed,
+	skipped,
+}: {
+	ctx: AutumnContext;
+	fullCus: FullCustomer;
+	computed: Array<{ cusEntId: string; result: ProcessResetResult }>;
+	skipped: string[];
+}): Promise<void> => {
+	const { db } = ctx;
+	const skippedSet = new Set(skipped);
+	const clearingPromises: Promise<Rollover[]>[] = [];
+
+	for (const { cusEntId, result } of computed) {
+		const original = findCusEnt({ fullCus, cusEntId });
+		if (!original) continue;
+
+		const { updates } = result;
+		if (updates.balance !== null) original.balance = updates.balance;
+		if (updates.additional_balance !== null)
+			original.additional_balance = updates.additional_balance;
+		original.adjustment = updates.adjustment;
+		if (updates.entities !== null) original.entities = updates.entities;
+		original.next_reset_at = updates.next_reset_at;
+
+		// Only run rollover clearing for DB-applied entries.
+		// Skipped entries were already cleared by the winning request.
+		if (!skippedSet.has(cusEntId) && result.rolloverInsert) {
+			clearingPromises.push(
+				RolloverService.clearExcessRollovers({
+					db,
+					newRows: result.rolloverInsert.rows,
+					fullCusEnt: original,
+				}),
+			);
+		}
+	}
+
+	if (clearingPromises.length > 0) {
+		await Promise.all(clearingPromises);
+	}
+};

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

@@ -0,0 +1,36 @@
+import { CusProductStatus } from "@autumn/shared";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import type { BatchResetCusEntsPayload } from "@/queue/workflows.js";
+import { CusService } from "../../CusService.js";
+
+/**
+ * SQS worker handler: fetches each FullCustomer via CusService.getFull,
+ * which triggers the lazy reset internally.
+ */
+export const batchResetCustomerEntitlements = async ({
+	ctx,
+	payload,
+}: {
+	ctx: AutumnContext;
+	payload: BatchResetCusEntsPayload;
+}): Promise<void> => {
+	const { resets } = payload;
+
+	if (resets.length === 0) return;
+
+	const BATCH_SIZE = 100;
+
+	for (let i = 0; i < resets.length; i += BATCH_SIZE) {
+		const batch = resets.slice(i, i + BATCH_SIZE);
+
+		await Promise.all(
+			batch.map((reset) =>
+				CusService.getFull({
+					ctx,
+					idOrInternalId: reset.internalCustomerId,
+					inStatuses: [CusProductStatus.Active, CusProductStatus.PastDue],
+				}),
+			),
+		);
+	}
+};

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

@@ -0,0 +1,34 @@
+import { redis } from "@/external/redis/initRedis.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv";
+import type { ResetCusEntParam } from "@/internal/balances/utils/sql/client.js";
+import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js";
+import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
+
+/**
+ * Atomically resets cusEnt fields in the cached FullCustomer blob.
+ * Skips gracefully if the cache doesn't exist or the cusEnt was already reset.
+ * Fire-and-forget — failures are logged but don't propagate.
+ */
+export const executeResetCache = async ({
+	ctx,
+	customerId,
+	resets,
+}: {
+	ctx: AutumnContext;
+	customerId: string;
+	resets: ResetCusEntParam[];
+}): Promise<void> => {
+	if (resets.length === 0) return;
+
+	const { org, env, logger } = ctx;
+
+	const cacheKey = buildFullCustomerCacheKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
+
+	await tryRedisWrite(() =>
+		redis.resetCustomerEntitlements(cacheKey, JSON.stringify({ resets })),
+	);
+};

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

@@ -0,0 +1,37 @@
+import {
+	CusProductStatus,
+	cusEntToCusPrice,
+	type FullCusEntWithFullCusProduct,
+	type FullCustomer,
+	fullCustomerToCustomerEntitlements,
+} from "@autumn/shared";
+
+/** Collects cusEnts from a FullCustomer that need resetting (next_reset_at < now). */
+export const getCusEntsNeedingReset = ({
+	fullCus,
+	now,
+}: {
+	fullCus: FullCustomer;
+	now: number;
+}): FullCusEntWithFullCusProduct[] => {
+	const result: FullCusEntWithFullCusProduct[] = [];
+
+	const cusEnts = fullCustomerToCustomerEntitlements({
+		fullCustomer: fullCus,
+		inStatuses: [CusProductStatus.Active, CusProductStatus.PastDue],
+	});
+
+	for (const cusEnt of cusEnts) {
+		if (!cusEnt.next_reset_at || cusEnt.next_reset_at >= now) continue;
+
+		const cusPrice = cusEntToCusPrice({ cusEnt });
+		if (cusPrice) continue;
+
+		result.push({
+			...cusEnt,
+			customer_product: cusEnt.customer_product,
+		});
+	}
+
+	return result;
+};

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

@@ -0,0 +1,69 @@
+import type {
+	AppEnv,
+	EntInterval,
+	FullCusProduct,
+	Organization,
+} from "@autumn/shared";
+import { UTCDate } from "@date-fns/utc";
+import { getDate, getMonth, setDate } from "date-fns";
+import { createStripeCli } from "@/external/connect/createStripeCli.js";
+import { getNextResetAt } from "@/utils/timeUtils.js";
+
+/** Computes next reset timestamp, adjusting for Stripe billing anchor on edge dates. */
+export const getResetAtUpdate = async ({
+	curResetAt,
+	interval,
+	intervalCount,
+	cusProduct,
+	org,
+	env,
+}: {
+	curResetAt: number;
+	interval: EntInterval;
+	intervalCount: number;
+	cusProduct: FullCusProduct | null;
+	org: Organization;
+	env: AppEnv;
+}): Promise<number> => {
+	const nextResetAt = getNextResetAt({
+		curReset: new UTCDate(curResetAt),
+		interval,
+		intervalCount,
+	});
+
+	if (!cusProduct) return nextResetAt;
+
+	// Only check Stripe anchor on edge dates (28th Feb, 30th of month)
+	const nextResetAtDate = new UTCDate(nextResetAt);
+	const nextResetAtDay = getDate(nextResetAtDate);
+	const nextResetAtMonth = getMonth(nextResetAtDate);
+
+	const shouldCheck =
+		nextResetAtDay === 30 || (nextResetAtDay === 28 && nextResetAtMonth === 2);
+
+	if (!shouldCheck) return nextResetAt;
+
+	if (
+		!cusProduct.subscription_ids ||
+		cusProduct.subscription_ids.length === 0
+	) {
+		return nextResetAt;
+	}
+
+	try {
+		const stripeCli = createStripeCli({ org, env });
+		const subId = cusProduct.subscription_ids[0];
+		const sub = await stripeCli.subscriptions.retrieve(subId);
+
+		const billingCycleAnchor = sub.billing_cycle_anchor * 1000;
+		const billingCycleDay = getDate(new UTCDate(billingCycleAnchor));
+
+		if (billingCycleDay > nextResetAtDay) {
+			return setDate(nextResetAtDate, billingCycleDay).getTime();
+		}
+	} catch (error) {
+		console.log(`[Lazy Reset] WARNING: Failed to check sub anchor: ${error}`);
+	}
+
+	return nextResetAt;
+};

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

@@ -0,0 +1,119 @@
+import {
+	cusEntToOptions,
+	type EntInterval,
+	type EntityBalance,
+	type FullCusEntWithFullCusProduct,
+	type FullCustomerEntitlement,
+	getStartingBalance,
+	isLifetimeEntitlement,
+	isUnlimitedEntitlement,
+	type Rollover,
+} from "@autumn/shared";
+import { logger } from "better-auth";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import { getRolloverUpdates } from "@/internal/customers/cusProducts/cusEnts/cusRollovers/rolloverUtils.js";
+import { getResetBalancesUpdate } from "@/internal/customers/cusProducts/cusEnts/groupByUtils.js";
+import { getResetAtUpdate } from "./getResetAtUpdate.js";
+
+export type ResetUpdates = {
+	balance: number | null;
+	additional_balance: number | null;
+	adjustment: number;
+	entities: Record<string, EntityBalance> | null;
+	next_reset_at: number;
+};
+
+export type ProcessResetResult = {
+	updates: ResetUpdates;
+	rolloverInsert?: { rows: Rollover[]; fullCusEnt: FullCustomerEntitlement };
+};
+
+/** Processes a single cusEnt reset. Returns updates + optional rollover insert, or null if skipped. */
+export const processReset = async ({
+	cusEnt,
+	ctx,
+}: {
+	cusEnt: FullCusEntWithFullCusProduct;
+	ctx: AutumnContext;
+}): Promise<ProcessResetResult | null> => {
+	const ent = cusEnt.entitlement;
+	const cusProduct = cusEnt.customer_product;
+
+	// Unlimited / lifetime cusEnts should never reach here
+	// (getCusEntsNeedingReset filters them out), but guard defensively
+	if (
+		isUnlimitedEntitlement({ entitlement: ent }) ||
+		isLifetimeEntitlement({ entitlement: ent })
+	) {
+		return null;
+	}
+
+	const options = cusEntToOptions({ cusEnt });
+
+	const resetBalance = getStartingBalance({
+		entitlement: cusEnt.entitlement,
+		options,
+		productQuantity: cusProduct?.quantity ?? 1,
+	});
+
+	if (!cusEnt.next_reset_at) {
+		logger.error(
+			`[customerEntitlement processReset] next_reset_at is null, cusEntId: ${cusEnt.id}`,
+		);
+		return null;
+	}
+
+	const { org, env } = ctx;
+
+	// Compute next reset time (with Stripe anchor adjustment on edge dates)
+	const nextResetAt = await getResetAtUpdate({
+		curResetAt: cusEnt.next_reset_at,
+		interval: ent.interval as EntInterval,
+		intervalCount: ent.interval_count,
+		cusProduct,
+		org,
+		env,
+	});
+
+	// Compute rollover before resetting balance
+	const rolloverUpdate = getRolloverUpdates({
+		cusEnt,
+		nextResetAt: cusEnt.next_reset_at,
+	});
+
+	// Compute reset balance update
+	const resetBalanceUpdate = getResetBalancesUpdate({
+		cusEnt,
+		allowance: resetBalance,
+	});
+
+	const updates: ResetUpdates =
+		"entities" in resetBalanceUpdate
+			? {
+					balance: null,
+					additional_balance: null,
+					adjustment: 0,
+					entities: resetBalanceUpdate.entities,
+					next_reset_at: nextResetAt,
+				}
+			: {
+					balance: resetBalanceUpdate.balance,
+					additional_balance: resetBalanceUpdate.additional_balance,
+					adjustment: 0,
+					entities: null,
+					next_reset_at: nextResetAt,
+				};
+
+	let rolloverInsert:
+		| { rows: Rollover[]; fullCusEnt: FullCustomerEntitlement }
+		| undefined;
+
+	if (rolloverUpdate?.toInsert && rolloverUpdate.toInsert.length > 0) {
+		rolloverInsert = {
+			rows: rolloverUpdate.toInsert,
+			fullCusEnt: cusEnt,
+		};
+	}
+
+	return { updates, rolloverInsert };
+};

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

@@ -0,0 +1,116 @@
+import type { FullCustomer } from "@autumn/shared";
+import * as Sentry from "@sentry/bun";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
+import {
+	type ResetCusEntParam,
+	resetCusEnts,
+} from "@/internal/balances/utils/sql/client.js";
+import { applyResetResults } from "./applyResetResults.js";
+import { executeResetCache } from "./executeResetCache.js";
+import { getCusEntsNeedingReset } from "./getCusEntsNeedingReset.js";
+import { type ProcessResetResult, processReset } from "./processReset.js";
+
+/** Maps a processReset result into the JSONB shape for the SQL function. */
+const toResetParam = ({
+	cusEntId,
+	result,
+}: {
+	cusEntId: string;
+	result: ProcessResetResult;
+}): ResetCusEntParam => {
+	const { updates } = result;
+	const firstRollover = result.rolloverInsert?.rows[0] ?? null;
+
+	return {
+		cus_ent_id: cusEntId,
+		balance: updates.balance,
+		additional_balance: updates.additional_balance,
+		adjustment: updates.adjustment,
+		entities: updates.entities,
+		next_reset_at: updates.next_reset_at,
+		rollover_insert: firstRollover,
+	};
+};
+
+/**
+ * Lazily resets customer entitlements that have passed their next_reset_at.
+ * Uses an atomic Postgres function with per-row locking to prevent double-resets.
+ * Mutates the FullCustomer in-memory using the latest DB state from applied resets.
+ * Returns true if any entitlements were reset.
+ */
+export const resetCustomerEntitlements = async ({
+	ctx,
+	fullCus,
+}: {
+	ctx: AutumnContext;
+	fullCus: FullCustomer;
+}): Promise<boolean> => {
+	const now = Date.now();
+
+	const { logger } = ctx;
+	const customerId = fullCus.id || fullCus.internal_id;
+
+	const cusEntsNeedingReset = getCusEntsNeedingReset({ fullCus, now });
+
+	if (cusEntsNeedingReset.length === 0) return false;
+
+	try {
+		logger.info(
+			`[resetCustomerEntitlements] customer=${customerId}, cusEnts needing reset: ${cusEntsNeedingReset.length}`,
+		);
+
+		// 1. Compute all resets (pure computation, no DB writes)
+		const computed: Array<{
+			cusEntId: string;
+			result: ProcessResetResult;
+		}> = [];
+
+		for (const cusEnt of cusEntsNeedingReset) {
+			const result = await processReset({ cusEnt, ctx });
+			if (!result) continue;
+			computed.push({ cusEntId: cusEnt.id, result });
+		}
+
+		if (computed.length === 0) return false;
+
+		// 2. Execute atomic DB writes via Postgres function
+		const resets = computed.map(({ cusEntId, result }) =>
+			toResetParam({ cusEntId, result }),
+		);
+
+		const { applied, skipped } = await resetCusEnts({ ctx, resets });
+
+		logger.info(
+			`[resetCustomerEntitlements] customer=${customerId}, applied: ${Object.keys(applied).length}, skipped: ${skipped.length}`,
+		);
+
+		// 3. Apply computed reset values to in-memory FullCustomer.
+		// Both DB-applied and DB-skipped cusEnts get their in-memory state updated
+		// (skipped means another request already wrote the same values to DB).
+		// Rollover clearing only runs for DB-applied entries.
+		await applyResetResults({ ctx, fullCus, computed, skipped });
+
+		// 4. Update Redis cache atomically (fire-and-forget)
+		// Only needed when we actually wrote to DB — skipped means cache was
+		// already updated by the winning request.
+		if (Object.keys(applied).length > 0) {
+			await executeResetCache({
+				ctx,
+				customerId,
+				resets,
+			});
+
+			logger.info(
+				`[resetCustomerEntitlements] customer=${customerId}, Redis cache updated`,
+			);
+		}
+
+		return true;
+	} catch (error) {
+		logger.error(
+			`[resetCustomerEntitlements] customer=${customerId}, failed: ${error}`,
+		);
+		Sentry.captureException(error);
+		return false;
+	}
+};

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

@@ -0,0 +1,42 @@
+import type { FullCustomer } from "@autumn/shared";
+import type { AutumnContext } from "@/honoUtils/HonoEnv";
+import { type BatchResetCusEntsPayload, workflows } from "@/queue/workflows.js";
+import { getCusEntsNeedingReset } from "./getCusEntsNeedingReset";
+
+/**
+ * Checks a list of FullCustomers for entitlements needing reset,
+ * and queues an SQS job with the cusEnt IDs if any are found.
+ */
+export const triggerBatchResetCustomerEntitlements = async ({
+	ctx,
+	fullCustomers,
+}: {
+	ctx: AutumnContext;
+	fullCustomers: FullCustomer[];
+}): Promise<void> => {
+	const now = Date.now();
+
+	const resets: BatchResetCusEntsPayload["resets"] = [];
+	for (const fullCus of fullCustomers) {
+		const cusEntsNeedingReset = getCusEntsNeedingReset({
+			fullCus,
+			now,
+		});
+
+		if (cusEntsNeedingReset.length === 0) continue;
+
+		resets.push({
+			internalCustomerId: fullCus.internal_id,
+			customerId: fullCus.id ?? "",
+			cusEntIds: cusEntsNeedingReset.map((cusEnt) => cusEnt.id),
+		});
+	}
+
+	if (resets.length === 0) return;
+
+	await workflows.triggerBatchResetCusEnts({
+		orgId: ctx.org.id,
+		env: ctx.env,
+		resets,
+	});
+};

server/src/internal/customers/cusProducts/cusEnts/CusEntitlementService.ts

@@ -1,3 +1,4 @@
+import type { UpdateCustomerEntitlement } from "@autumn/shared";
 import {
 	type AppEnv,
 	type CusProduct,
@@ -15,11 +16,10 @@ import {
 	type InsertCustomerEntitlement,
 	type ResetCusEnt,
 } from "@autumn/shared";
-import { and, eq, gt, isNull, lt, or, sql } from "drizzle-orm";
+import { and, eq, gt, inArray, isNull, lt, or, sql } from "drizzle-orm";
 import { StatusCodes } from "http-status-codes";
 import { buildConflictUpdateColumns } from "@/db/dbUtils.js";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
-import type { UpdateCustomerEntitlement } from "@autumn/shared";
 import RecaseError from "@/utils/errorUtils.js";
 
 export class CusEntService {
@@ -44,6 +44,17 @@ export class CusEntService {
 			});
 	}
 
+	static async getByIds({ db, ids }: { db: DrizzleCli; ids: string[] }) {
+		if (ids.length === 0) return [];
+
+		const data = await db
+			.select()
+			.from(customerEntitlements)
+			.where(inArray(customerEntitlements.id, ids));
+
+		return data as CustomerEntitlement[];
+	}
+
 	static async getByFeature({
 		db,
 		internalFeatureId,

server/src/internal/customers/cusProducts/cusEnts/cusEntUtils/getCusEntByFeature.ts

@@ -1,29 +1,19 @@
-import type {
-	AppEnv,
-	FullCustomerEntitlement,
-	Organization,
-} from "@autumn/shared";
-import type { DrizzleCli } from "../../../../../db/initDrizzle.js";
+import type { FullCustomerEntitlement } from "@autumn/shared";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { CusService } from "../../../CusService.js";
 
 export const getCusEntByFeature = async ({
-	db,
-	org,
-	env,
+	ctx,
 	customerId,
 	featureId,
 }: {
-	db: DrizzleCli;
-	org: Organization;
-	env: AppEnv;
+	ctx: AutumnContext;
 	customerId: string;
 	featureId: string;
 }) => {
 	const fullCus = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: org.id,
-		env,
 	});
 
 	const cusEnts = fullCus?.customer_products

server/src/internal/customers/cusProducts/cusEnts/cusRollovers/RolloverService.ts

@@ -88,7 +88,24 @@ export class RolloverService {
 
 		await db.insert(rollovers).values(rows).returning();
 
-		let curRollovers = [...fullCusEnt.rollovers, ...rows];
+		return RolloverService.clearExcessRollovers({
+			db,
+			newRows: rows,
+			fullCusEnt,
+		});
+	}
+
+	/** Enforces the rollover max cap after new rollovers have been inserted into the DB. */
+	static async clearExcessRollovers({
+		db,
+		newRows,
+		fullCusEnt,
+	}: {
+		db: DrizzleCli;
+		newRows: Rollover[];
+		fullCusEnt: FullCustomerEntitlement;
+	}): Promise<Rollover[]> {
+		const curRollovers = [...fullCusEnt.rollovers, ...newRows];
 
 		const { toDelete, toUpdate } = performMaximumClearing({
 			rows: curRollovers as Rollover[],
@@ -103,17 +120,9 @@ export class RolloverService {
 			await RolloverService.upsert({ db, rows: toUpdate });
 		}
 
-		// Return latest rollovers...?
-		curRollovers = curRollovers.filter((r) => toDelete.includes(r.id));
-		curRollovers = curRollovers.map((r) => {
-			const updatedRow = toUpdate.find((u) => u.id === r.id);
-			if (updatedRow) {
-				return updatedRow;
-			}
-			return r;
-		});
-
-		return curRollovers;
+		return curRollovers
+			.filter((r) => !toDelete.includes(r.id))
+			.map((r) => toUpdate.find((u) => u.id === r.id) ?? r);
 	}
 
 	static async delete({ db, ids }: { db: DrizzleCli; ids: string[] }) {

server/src/internal/customers/cusProducts/cusEnts/groupByUtils.ts

@@ -1,14 +1,17 @@
-import type { FullCustomerEntitlement } from "@autumn/shared";
+import type { EntityBalance, FullCustomerEntitlement } from "@autumn/shared";
 import { notNullish } from "@/utils/genUtils.js";
 
+export type ResetBalancesUpdate =
+	| { entities: Record<string, EntityBalance> }
+	| { balance: number; additional_balance: number; adjustment: number };
+
 export const getResetBalancesUpdate = ({
 	cusEnt,
 	allowance,
 }: {
 	cusEnt: FullCustomerEntitlement;
 	allowance?: number;
-}) => {
-	let update = {};
+}): ResetBalancesUpdate => {
 	const newBalance = notNullish(allowance)
 		? allowance!
 		: cusEnt.entitlement.allowance || 0;
@@ -21,14 +24,12 @@ export const getResetBalancesUpdate = ({
 			newEntities[entityId].balance = newBalance;
 			newEntities[entityId].adjustment = 0;
 		}
-		update = { entities: newEntities };
-	} else {
-		update = {
-			balance: newBalance,
-			additional_balance: 0,
-			adjustment: 0,
-		};
+		return { entities: newEntities };
 	}
 
-	return update;
+	return {
+		balance: newBalance,
+		additional_balance: 0,
+		adjustment: 0,
+	};
 };

server/src/internal/customers/cusUtils/apiCusUtils/getApiCustomerExpand.ts

@@ -37,10 +37,8 @@ export const getApiCustomerExpand = async ({
 
 	if (!fullCus) {
 		fullCus = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customerId || "",
-			orgId: org.id,
-			env,
 			expand: expand as CustomerExpand[],
 			withEntities: expand.includes(CustomerExpand.Entities),
 			withSubs: true,

server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.ts

@@ -2,7 +2,9 @@ import type { FullCustomer } from "@autumn/shared";
 import { Decimal } from "decimal.js";
 import type { Redis } from "ioredis";
 import { redis } from "@/external/redis/initRedis.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { tryRedisRead } from "@/utils/cacheUtils/cacheUtils.js";
+import { resetCustomerEntitlements } from "../../actions/resetCustomerEntitlements/resetCustomerEntitlements.js";
 import { buildFullCustomerCacheKey } from "./fullCustomerCacheConfig.js";
 
 /**
@@ -80,23 +82,26 @@ const roundFullCustomerBalances = (
 };
 
 /**
- * Get FullCustomer from Redis cache
+ * Get FullCustomer from Redis cache. Lazily resets stale entitlements.
  * @returns FullCustomer if found, null if not in cache
  */
 export const getCachedFullCustomer = async ({
-	orgId,
-	env,
+	ctx,
 	customerId,
 	entityId,
 	redisInstance,
 }: {
-	orgId: string;
-	env: string;
+	ctx: AutumnContext;
 	customerId: string;
 	entityId?: string;
 	redisInstance?: Redis;
 }): Promise<FullCustomer | null> => {
-	const cacheKey = buildFullCustomerCacheKey({ orgId, env, customerId });
+	const { org, env } = ctx;
+	const cacheKey = buildFullCustomerCacheKey({
+		orgId: org.id,
+		env,
+		customerId,
+	});
 	const redisClient = redisInstance || redis;
 
 	const cached = await tryRedisRead(
@@ -123,6 +128,9 @@ export const getCachedFullCustomer = async ({
 		fullCustomer.send_email_receipts = false;
 	}
 
+	// Lazy reset stale entitlements (DB + in-memory + cache via Lua)
+	await resetCustomerEntitlements({ ctx, fullCus: fullCustomer });
+
 	// Round balance fields to handle floating-point precision from JSON.NUMINCRBY
 	return roundFullCustomerBalances(fullCustomer);
 };

server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getOrCreateCachedFullCustomer.ts

@@ -1,5 +1,4 @@
 import {
-	type AppEnv,
 	type CheckParams,
 	CustomerExpand,
 	type Entity,
@@ -29,7 +28,7 @@ export const getOrCreateCachedFullCustomer = async ({
 	};
 	source?: string;
 }): Promise<FullCustomer> => {
-	const { org, env, db, skipCache, logger } = ctx;
+	const { skipCache, logger } = ctx;
 	const {
 		customer_id: customerId,
 		customer_data: customerData,
@@ -40,13 +39,12 @@ export const getOrCreateCachedFullCustomer = async ({
 	let fullCustomer: FullCustomer | undefined;
 	const fetchTimeMs = Date.now();
 
-	// 1. Try cache first
+	// 1. Try cache first (getCachedFullCustomer handles lazy reset internally)
 	let setCache = true;
 	if (customerId && !skipCache) {
 		fullCustomer =
 			(await getCachedFullCustomer({
-				orgId: org.id,
-				env,
+				ctx,
 				customerId,
 				entityId,
 			})) ?? undefined;
@@ -57,13 +55,11 @@ export const getOrCreateCachedFullCustomer = async ({
 		}
 	}
 
-	// 2. Try DB if not in cache
+	// 2. Try DB if not in cache (CusService.getFull handles lazy reset internally)
 	if (!fullCustomer && customerId) {
 		fullCustomer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customerId,
-			orgId: org.id,
-			env: env as AppEnv,
 			withEntities: true,
 			withSubs: true,
 			expand: [CustomerExpand.Invoices],
@@ -91,10 +87,8 @@ export const getOrCreateCachedFullCustomer = async ({
 		setCache = true;
 
 		fullCustomer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: fullCustomer.id || fullCustomer.internal_id,
-			orgId: org.id,
-			env: env as AppEnv,
 			withEntities: true,
 			withSubs: true,
 			entityId,

server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getOrSetCachedFullCustomer.ts

@@ -1,5 +1,4 @@
 import {
-	type AppEnv,
 	CustomerExpand,
 	CustomerNotFoundError,
 	EntityNotFoundError,
@@ -25,13 +24,12 @@ export const getOrSetCachedFullCustomer = async ({
 	entityId?: string;
 	source?: string;
 }): Promise<FullCustomer> => {
-	const { org, env, db, skipCache, logger } = ctx;
+	const { skipCache, logger } = ctx;
 
-	// 1. Try cache first
+	// 1. Try cache first (getCachedFullCustomer handles lazy reset internally)
 	if (!skipCache) {
 		const cached = await getCachedFullCustomer({
-			orgId: org.id,
-			env,
+			ctx,
 			customerId,
 		});
 
@@ -63,10 +61,8 @@ export const getOrSetCachedFullCustomer = async ({
 	const fetchTimeMs = Date.now();
 
 	const fullCustomer = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: org.id,
-		env: env as AppEnv,
 		withEntities: true,
 		withSubs: true,
 		expand: [CustomerExpand.Invoices],

server/src/internal/customers/cusUtils/getOrCreateCustomer.ts

@@ -54,10 +54,8 @@ export const getOrCreateCustomer = async ({
 
 	if (!skipGet && customerId) {
 		customer = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customerId,
-			orgId: org.id,
-			env,
 			inStatuses,
 			withEntities,
 			entityId,
@@ -88,10 +86,8 @@ export const getOrCreateCustomer = async ({
 
 		if (updated) {
 			customer = await CusService.getFull({
-				db,
+				ctx,
 				idOrInternalId: customer.id || customer.internal_id,
-				orgId: org.id,
-				env,
 				inStatuses,
 				withEntities,
 				entityId,

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

@@ -40,10 +40,8 @@ export const handleTransferProductV2 = createRoute({
 		}
 
 		const customer = await CusService.getFull({
+			ctx,
 			idOrInternalId: customer_id,
-			orgId: org.id,
-			env,
-			db,
 			withEntities: true,
 		});
 

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

@@ -12,15 +12,13 @@ export const handleUpdateBalancesV2 = createRoute({
 	handler: async (c) => {
 		const ctx = c.get("ctx");
 
-		const { org, env, db, features } = ctx;
+		const { features } = ctx;
 		const { customer_id } = c.req.param();
 		const { balances, entity_id } = c.req.valid("json");
 
 		const fullCus = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customer_id,
-			orgId: org.id,
-			env,
 			entityId: entity_id,
 		});
 

server/src/internal/customers/internalHandlers/handleGetCustomer.ts

@@ -7,13 +7,11 @@ import { CusService } from "@/internal/customers/CusService";
  */
 export const handleGetCustomer = createRoute({
 	handler: async (c) => {
-		const { db, org, env } = c.get("ctx");
+		const ctx = c.get("ctx");
 		const { customer_id } = c.req.param();
 
 		const fullCus = await CusService.getFull({
-			db,
-			orgId: org.id,
-			env,
+			ctx,
 			idOrInternalId: customer_id,
 			withEntities: true,
 			expand: [CustomerExpand.Invoices],

server/src/internal/customers/internalHandlers/handleGetCustomerEvents.ts

@@ -19,13 +19,11 @@ export const handleGetCustomerEvents = createRoute({
 	query: QuerySchema,
 	handler: async (c) => {
 		const ctx = c.get("ctx");
-		const { db, org, env } = ctx;
 		const { customer_id } = c.req.param();
 		const { interval, limit } = c.req.valid("query");
 
 		const customer = await getCachedFullCustomer({
-			orgId: org.id,
-			env,
+			ctx,
 			customerId: customer_id,
 		});
 

server/src/internal/customers/internalHandlers/handleGetCustomerProduct.ts

@@ -22,14 +22,13 @@ export const handleGetCustomerProduct = createRoute({
 		entity_id: z.string().optional(),
 	}),
 	handler: async (c) => {
-		const { db, org, env, features } = c.get("ctx");
+		const ctx = c.get("ctx");
+		const { db, org, env, features } = ctx;
 		const { customer_id, product_id } = c.req.param();
 		const { version, customer_product_id, entity_id } = c.req.valid("query");
 
 		const customer = await CusService.getFull({
-			db,
-			orgId: org.id,
-			env,
+			ctx,
 			idOrInternalId: customer_id,
 			withEntities: true,
 			entityId: entity_id,

server/src/internal/customers/internalHandlers/handleGetFullCustomers.ts

@@ -19,9 +19,11 @@ export const handleGetFullCustomers = createRoute({
 		filters: z.any().optional(),
 	}),
 	handler: async (c) => {
-		const { db, org, env } = c.get("ctx");
+		const ctx = c.get("ctx");
 		const { search, page_size, page, last_item, filters } = c.req.valid("json");
 
+		const { org, env, db } = ctx;
+
 		const { data: customers } = await CusSearchService.search({
 			db,
 			orgId: org.id,
@@ -34,9 +36,7 @@ export const handleGetFullCustomers = createRoute({
 		});
 
 		const fullCustomers = await CusBatchService.getByInternalIds({
-			db,
-			org,
-			env,
+			ctx,
 			internalCustomerIds: customers.map(
 				(customer: Customer) => customer.internal_id,
 			),

server/src/internal/entities/actions/deleteEntity.ts

@@ -32,10 +32,8 @@ export const deleteEntity = async ({
 	const { db, org, env, features, logger } = ctx;
 
 	const fullCus = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: org.id,
-		env,
 		withEntities: true,
 	});
 

server/src/internal/entities/entityUtils/apiEntityUtils/getApiEntityExpand.ts

@@ -27,10 +27,8 @@ export const getApiEntityExpand = async ({
 
 	if (!fullCus) {
 		fullCus = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customerId || "",
-			orgId: org.id,
-			env,
 			entityId,
 		});
 	}

server/src/internal/entities/handlers/handleCreateEntity/autoCreateEntity.ts

@@ -49,10 +49,8 @@ export const autoCreateEntity = async ({
 
 	if (!fullCus) {
 		fullCus = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customerId,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			withEntities: true,
 			entityId,
 		});

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

@@ -25,10 +25,8 @@ export const handleDeleteEntity = createRoute({
 		const { db, org, env, features, logger } = ctx;
 
 		const fullCus = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customer_id,
-			orgId: org.id,
-			env,
 			withEntities: true,
 		});
 

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

@@ -6,13 +6,9 @@ export const handleListEntities = createRoute({
 		const { customer_id } = c.req.param();
 		const ctx = c.get("ctx");
 
-		const { db, org, env } = ctx;
-
 		const fullCus = await CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: customer_id,
-			orgId: org.id,
-			env,
 		});
 
 		return c.json({

server/src/internal/events/handlers/handleExternalAggregateEvents.ts

@@ -36,10 +36,8 @@ export const handleExternalAggregateEvents = createRoute({
 		});
 
 		const customer = await CusService.getFull({
-			db,
-			orgId: org.id,
+			ctx,
 			idOrInternalId: customer_id,
-			env,
 			withSubs: true,
 		});
 

server/src/internal/migrations/migrationSteps/migrateCustomer.ts

@@ -26,8 +26,7 @@ export const migrateCustomer = async ({
 	toProduct: FullProduct;
 	migrationJob?: MigrationJob;
 }) => {
-	const { db, org, env } = ctx;
-	const orgId = org.id;
+	const { db } = ctx;
 
 	// Create customer-specific logger
 	const customerLogger = createMigrationCustomerLogger({
@@ -39,10 +38,8 @@ export const migrateCustomer = async ({
 
 	try {
 		const fullCus = await CusService.getFull({
-			db,
+			ctx: customerCtx,
 			idOrInternalId: customerId,
-			orgId,
-			env,
 			withEntities: true,
 			inStatuses: ACTIVE_STATUSES,
 		});

server/src/internal/misc/components/handlers/handleGetPricingTable.ts

@@ -29,9 +29,7 @@ export const handleGetPricingTable = createRoute({
 					return undefined;
 				}
 				return await CusService.getFull({
-					db,
-					orgId: org.id,
-					env,
+					ctx,
 					idOrInternalId: customerId,
 				});
 			})(),

server/src/internal/products/handlers/handleListPlans.ts

@@ -29,10 +29,8 @@ export const handleListPlans = createRoute({
 			}),
 			customer_id
 				? CusService.getFull({
-						db,
+						ctx,
 						idOrInternalId: customer_id,
-						orgId: org.id,
-						env,
 						entityId: entity_id,
 						withEntities: true,
 						withSubs: true,

server/src/internal/products/handlers/handleListPlans/handleListPlansV2.ts

@@ -29,10 +29,8 @@ export const handleListPlansV2 = createRoute({
 			}),
 			customer_id
 				? CusService.getFull({
-						db,
+						ctx,
 						idOrInternalId: customer_id,
-						orgId: org.id,
-						env,
 						entityId: entity_id,
 						withEntities: true,
 						withSubs: true,

server/src/internal/rewards/referralUtils/triggerFreePaidProduct.ts

@@ -62,18 +62,14 @@ export const triggerFreePaidProduct = async ({
 
 	const [fullReferrer, fullRedeemer] = await Promise.all([
 		CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: referralCode.internal_customer_id,
-			orgId: org.id,
-			env,
 			withEntities: true,
 			withSubs: true,
 		}),
 		CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: redeemer.id!,
-			orgId: org.id,
-			env,
 			withEntities: true,
 			withSubs: true,
 		}),

server/src/internal/rewards/referralUtils/triggerFreeProduct.ts

@@ -77,17 +77,13 @@ export const triggerFreeProduct = async ({
 
 	const [fullReferrer, fullRedeemer] = await Promise.all([
 		CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: referralCode.internal_customer_id,
-			orgId: org.id,
-			env,
 			allowNotFound: true,
 		}),
 		CusService.getFull({
-			db,
+			ctx,
 			idOrInternalId: redeemer.id!,
-			orgId: org.id,
-			env,
 		}),
 	]);
 

server/src/queue/JobName.ts

@@ -22,6 +22,8 @@ export enum JobName {
 
 	ClearCreditSystemCustomerCache = "clear-credit-system-customer-cache",
 
+	BatchResetCusEnts = "batch-reset-cus-ents",
+
 	// Hatchet workflows
 	VerifyCacheConsistency = "verify-cache-consistency",
 }

server/src/queue/processMessage.ts

@@ -8,6 +8,7 @@ import { runInsertEventBatch } from "@/internal/balances/events/runInsertEventBa
 import { syncItemV3 } from "@/internal/balances/utils/sync/syncItemV3.js";
 import { grantCheckoutReward } from "@/internal/billing/v2/workflows/grantCheckoutReward/grantCheckoutReward.js";
 import { sendProductsUpdated } from "@/internal/billing/v2/workflows/sendProductsUpdated/sendProductsUpdated.js";
+import { batchResetCustomerEntitlements } from "@/internal/customers/actions/resetCustomerEntitlements/batchResetCustomerEntitlements.js";
 import { runClearCreditSystemCacheTask } from "@/internal/features/featureActions/runClearCreditSystemCacheTask.js";
 import { generateFeatureDisplay } from "@/internal/features/workflows/generateFeatureDisplay.js";
 import { runMigrationTask } from "@/internal/migrations/runMigrationTask.js";
@@ -183,6 +184,19 @@ export const processMessage = async ({
 				ctx,
 				payload: job.data,
 			});
+			return;
+		}
+
+		if (job.name === JobName.BatchResetCusEnts) {
+			if (!ctx) {
+				workerLogger.error("No context found for batch reset cus ents job");
+				return;
+			}
+			await batchResetCustomerEntitlements({
+				ctx,
+				payload: job.data,
+			});
+			return;
 		}
 	} catch (error) {
 		Sentry.captureException(error);

server/src/queue/queueUtils.ts

@@ -9,7 +9,10 @@ import {
 import type { ClearCreditSystemCachePayload } from "@/internal/features/featureActions/runClearCreditSystemCacheTask.js";
 import type { GenerateFeatureDisplayPayload } from "@/internal/features/workflows/generateFeatureDisplay.js";
 import { JobName } from "./JobName.js";
-import type { SendProductsUpdatedPayload } from "./workflows.js";
+import type {
+	BatchResetCusEntsPayload,
+	SendProductsUpdatedPayload,
+} from "./workflows.js";
 
 export interface Payloads {
 	[JobName.RewardMigration]: {
@@ -45,6 +48,7 @@ export interface Payloads {
 	[JobName.ClearCreditSystemCustomerCache]: ClearCreditSystemCachePayload;
 	[JobName.GenerateFeatureDisplay]: GenerateFeatureDisplayPayload;
 	[JobName.SendProductsUpdated]: SendProductsUpdatedPayload;
+	[JobName.BatchResetCusEnts]: BatchResetCusEntsPayload;
 	[JobName.VerifyCacheConsistency]: {
 		customerId: string;
 		orgId: string;

server/src/queue/workflows.ts

@@ -35,6 +35,16 @@ export type GrantCheckoutRewardPayload = {
 	stripeSubscriptionId?: string;
 };
 
+export type BatchResetCusEntsPayload = {
+	orgId: string;
+	env: string;
+	resets: {
+		internalCustomerId: string;
+		customerId: string;
+		cusEntIds: string[];
+	}[];
+};
+
 // ============ Workflow Registry ============
 
 type WorkflowRunner = "sqs" | "hatchet";
@@ -65,6 +75,11 @@ const workflowRegistry = {
 		jobName: JobName.GrantCheckoutReward,
 		runner: "sqs",
 	} as WorkflowConfig<GrantCheckoutRewardPayload>,
+
+	batchResetCusEnts: {
+		jobName: JobName.BatchResetCusEnts,
+		runner: "sqs",
+	} as WorkflowConfig<BatchResetCusEntsPayload>,
 } as const;
 
 // ============ Type Utilities ============
@@ -131,4 +146,9 @@ export const workflows = {
 		payload: GrantCheckoutRewardPayload,
 		options?: TriggerOptions,
 	) => triggerWorkflow({ name: "grantCheckoutReward", payload, options }),
+
+	triggerBatchResetCusEnts: (
+		payload: BatchResetCusEntsPayload,
+		options?: TriggerOptions,
+	) => triggerWorkflow({ name: "batchResetCusEnts", payload, options }),
 };

server/src/utils/scriptUtils/initCustomer.ts

@@ -1,7 +1,6 @@
 import {
 	type AppEnv,
 	type Customer,
-	type CustomerData,
 	type Organization,
 	ProcessorType,
 } from "@autumn/shared";
@@ -206,80 +205,3 @@ export const attachPaymentMethod = async ({
 		console.log("failed to attach payment method", error);
 	}
 };
-
-// V2 initializes the customer in Stripe, then creates the customer in Autumn
-export const initCustomerV2 = async ({
-	autumn,
-	customerId,
-	customerData,
-	org,
-	env,
-	db,
-	attachPm,
-	withTestClock = true,
-}: {
-	autumn: AutumnInt;
-	customerId: string;
-	customerData?: CustomerData;
-	org: Organization;
-	env: AppEnv;
-	db: DrizzleCli;
-	attachPm?: "success" | "fail";
-	withTestClock?: boolean;
-}) => {
-	const name = customerId;
-	const email = `${customerId}@example.com`;
-	const fingerprint_ = "";
-	const stripeCli = createStripeCli({ org, env });
-
-	let testClockId: string | undefined;
-
-	if (withTestClock) {
-		const testClock = await stripeCli.testHelpers.testClocks.create({
-			frozen_time: Math.floor(Date.now() / 1000),
-		});
-		testClockId = testClock.id;
-	}
-
-	// 1. Create stripe customer
-	const stripeCus = await stripeCli.customers.create({
-		email,
-		name,
-		test_clock: testClockId,
-	});
-
-	// 2. Create customer
-	try {
-		await autumn.customers.delete(customerId);
-	} catch (_error) {}
-
-	await autumn.customers.create({
-		id: customerId,
-		name,
-		email,
-		fingerprint: customerData?.fingerprint || undefined,
-		stripe_id: stripeCus.id,
-		metadata: {},
-	});
-
-	// 3. Attach payment method
-	if (attachPm) {
-		await attachPaymentMethod({
-			stripeCli,
-			stripeCusId: stripeCus.id,
-			type: attachPm,
-		});
-	}
-
-	const customer = await CusService.getFull({
-		db,
-		idOrInternalId: customerId,
-		orgId: org.id,
-		env: env,
-	});
-
-	return {
-		testClockId: testClockId || "",
-		customer,
-	};
-};

server/src/utils/scriptUtils/testUtils/cusTestUtils.ts

@@ -1,28 +1,24 @@
-import type { Organization } from "@autumn/shared";
 import { AppEnv } from "@autumn/shared";
-import type { DrizzleCli } from "@/db/initDrizzle.js";
 import { createStripeCli } from "@/external/connect/createStripeCli.js";
+import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
 import { CusService } from "@/internal/customers/CusService.js";
 import { cusProductToSub } from "@/internal/customers/cusProducts/cusProductUtils/convertCusProduct.js";
 
 export const getCusSub = async ({
-	db,
-	org,
+	ctx,
 	customerId,
 	productId,
 }: {
-	db: DrizzleCli;
-	org: Organization;
+	ctx: AutumnContext;
 	customerId: string;
 	productId: string;
 }) => {
+	const { org } = ctx;
 	const env = AppEnv.Sandbox;
 	const stripeCli = createStripeCli({ org, env });
 	const fullCus = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		env,
-		orgId: org.id,
 	});
 
 	const cusProduct = fullCus.customer_products.find(

server/src/utils/scriptUtils/testUtils/initCustomerV3.ts

@@ -92,12 +92,9 @@ export const initCustomerV3 = async ({
 		});
 	}
 
-	const { db, org, env } = ctx;
 	const customer = await CusService.getFull({
-		db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: org.id,
-		env: env,
 	});
 
 	return {

server/tests/advanced/multiFeature/multiFeature1.test.ts

@@ -6,7 +6,7 @@ import {
 	getUsageCusEnt,
 } from "@tests/utils/cusProductUtils/cusEntSearchUtils.js";
 import { getMainCusProduct } from "@tests/utils/cusProductUtils/cusProductUtils.js";
-import ctx from "@tests/utils/testInitUtils/createTestContext.js";
+import ctx, { type TestContext } from "@tests/utils/testInitUtils/createTestContext.js";
 import chalk from "chalk";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import { AutumnInt } from "@/external/autumn/autumnCli.js";
@@ -64,23 +64,17 @@ const premium = constructProduct({
 });
 
 export const getPrepaidAndUsageCusEnts = async ({
+	ctx,
 	customerId,
-	db,
-	orgId,
-	env,
 	featureId,
 }: {
+	ctx: TestContext;
 	customerId: string;
-	db: DrizzleCli;
-	orgId: string;
-	env: AppEnv;
 	featureId: string;
 }) => {
 	const mainCusProduct = await getMainCusProduct({
+		ctx,
 		customerId,
-		db,
-		orgId,
-		env,
 	});
 
 	const prepaidCusEnt = getPrepaidCusEnt({
@@ -147,10 +141,8 @@ describe(`${chalk.yellowBright(
 		});
 
 		const { prepaidCusEnt, usageCusEnt } = await getPrepaidAndUsageCusEnts({
+			ctx,
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -172,10 +164,8 @@ describe(`${chalk.yellowBright(
 		await timeout(3000);
 
 		const { prepaidCusEnt, usageCusEnt } = await getPrepaidAndUsageCusEnts({
+			ctx,
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -196,10 +186,8 @@ describe(`${chalk.yellowBright(
 		await timeout(2500);
 
 		const { usageCusEnt } = await getPrepaidAndUsageCusEnts({
+			ctx,
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -211,10 +199,8 @@ describe(`${chalk.yellowBright(
 
 		const { prepaidCusEnt, usageCusEnt: newUsageCusEnt } =
 			await getPrepaidAndUsageCusEnts({
+				ctx,
 				customerId,
-				db: ctx.db,
-				orgId: ctx.org.id,
-				env: ctx.env,
 				featureId: TestFeature.Messages,
 			});
 

server/tests/advanced/multiFeature/multiFeature2.test.ts

@@ -56,22 +56,14 @@ const premium = constructProduct({
 
 export const getLifetimeAndUsageCusEnts = async ({
 	customerId,
-	db,
-	orgId,
-	env,
 	featureId,
 }: {
 	customerId: string;
-	db: DrizzleCli;
-	orgId: string;
-	env: AppEnv;
 	featureId: string;
 }) => {
 	const mainCusProduct = await getMainCusProduct({
+		ctx,
 		customerId: customerId,
-		db,
-		orgId,
-		env,
 	});
 
 	const lifetimeCusEnt = getLifetimeFreeCusEnt({
@@ -127,9 +119,6 @@ describe(`${chalk.yellowBright(
 
 		const { lifetimeCusEnt, usageCusEnt } = await getLifetimeAndUsageCusEnts({
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -153,9 +142,6 @@ describe(`${chalk.yellowBright(
 
 		const { lifetimeCusEnt, usageCusEnt } = await getLifetimeAndUsageCusEnts({
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -181,9 +167,6 @@ describe(`${chalk.yellowBright(
 		const { lifetimeCusEnt, usageCusEnt: newUsageCusEnt } =
 			await getLifetimeAndUsageCusEnts({
 				customerId,
-				db: ctx.db,
-				orgId: ctx.org.id,
-				env: ctx.env,
 				featureId: TestFeature.Messages,
 			});
 

server/tests/advanced/multiFeature/multiFeature3.test.ts

@@ -44,22 +44,14 @@ const pro = constructProduct({
 
 export const getLifetimeAndUsageCusEnts = async ({
 	customerId,
-	db,
-	orgId,
-	env,
 	featureId,
 }: {
 	customerId: string;
-	db: DrizzleCli;
-	orgId: string;
-	env: AppEnv;
 	featureId: string;
 }) => {
 	const mainCusProduct = await getMainCusProduct({
+		ctx,
 		customerId,
-		db,
-		orgId,
-		env,
 	});
 
 	const lifetimeCusEnt = getLifetimeFreeCusEnt({
@@ -116,9 +108,6 @@ describe(`${chalk.yellowBright(
 
 		const { lifetimeCusEnt, usageCusEnt } = await getLifetimeAndUsageCusEnts({
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -144,9 +133,6 @@ describe(`${chalk.yellowBright(
 
 		const { lifetimeCusEnt, usageCusEnt } = await getLifetimeAndUsageCusEnts({
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 
@@ -165,9 +151,6 @@ describe(`${chalk.yellowBright(
 
 		const { lifetimeCusEnt, usageCusEnt } = await getLifetimeAndUsageCusEnts({
 			customerId,
-			db: ctx.db,
-			orgId: ctx.org.id,
-			env: ctx.env,
 			featureId: TestFeature.Messages,
 		});
 

server/tests/advanced/referrals/paid/referrals13.test.ts

@@ -274,21 +274,19 @@ describe(`${chalk.yellowBright(
 			expect(actualTotal!).toBeLessThanOrEqual(expectedTotal / 2);
 		}
 
-		const dbCustomers = await Promise.all(
-			[mainCustomerId, redeemer].map((x) =>
-				CusService.getFull({
-					db,
-					idOrInternalId: x,
-					orgId: org.id,
-					env,
-					inStatuses: [
-						CusProductStatus.Active,
-						CusProductStatus.PastDue,
-						CusProductStatus.Expired,
-					],
-				}),
-			),
-		);
+	const dbCustomers = await Promise.all(
+		[mainCustomerId, redeemer].map((x) =>
+			CusService.getFull({
+				ctx,
+				idOrInternalId: x,
+				inStatuses: [
+					CusProductStatus.Active,
+					CusProductStatus.PastDue,
+					CusProductStatus.Expired,
+				],
+			}),
+		),
+	);
 
 		const expectedProducts = [
 			[

server/tests/advanced/referrals/paid/referrals14.test.ts

@@ -306,21 +306,19 @@ describe(`${chalk.yellowBright(
 			expect(premiumInvoice.total).toBeLessThan(premiumPrice);
 		}
 
-		const dbCustomers = await Promise.all(
-			[mainCustomerId, redeemer].map((x) =>
-				CusService.getFull({
-					db,
-					idOrInternalId: x,
-					orgId: org.id,
-					env,
-					inStatuses: [
-						CusProductStatus.Active,
-						CusProductStatus.PastDue,
-						CusProductStatus.Expired,
-					],
-				}),
-			),
-		);
+	const dbCustomers = await Promise.all(
+		[mainCustomerId, redeemer].map((x) =>
+			CusService.getFull({
+				ctx,
+				idOrInternalId: x,
+				inStatuses: [
+					CusProductStatus.Active,
+					CusProductStatus.PastDue,
+					CusProductStatus.Expired,
+				],
+			}),
+		),
+	);
 
 		const expectedProducts = [
 			[

server/tests/advanced/referrals/paid/referrals15.test.ts

@@ -289,21 +289,19 @@ describe(`${chalk.yellowBright(
 			expect(redeemerProInvoice.total).toBe(expectedTotal);
 		}
 
-		const dbCustomers = await Promise.all(
-			[mainCustomerId, redeemer].map((x) =>
-				CusService.getFull({
-					db,
-					idOrInternalId: x,
-					orgId: org.id,
-					env,
-					inStatuses: [
-						CusProductStatus.Active,
-						CusProductStatus.PastDue,
-						CusProductStatus.Expired,
-					],
-				}),
-			),
-		);
+	const dbCustomers = await Promise.all(
+		[mainCustomerId, redeemer].map((x) =>
+			CusService.getFull({
+				ctx,
+				idOrInternalId: x,
+				inStatuses: [
+					CusProductStatus.Active,
+					CusProductStatus.PastDue,
+					CusProductStatus.Expired,
+				],
+			}),
+		),
+	);
 
 		const expectedProducts = [
 			[

server/tests/balances/check/loose/loose-expiry.test.ts

@@ -1,205 +1,200 @@
-import { beforeAll, describe, expect, test } from "bun:test";
+import { expect, test } from "bun:test";
 import {
 	ApiVersion,
 	type CheckResponseV2,
 	ResetInterval,
 } from "@autumn/shared";
 import { TestFeature } from "@tests/setup/v2Features.js";
-import ctx from "@tests/utils/testInitUtils/createTestContext.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 { AutumnInt } from "@/external/autumn/autumnCli.js";
-import { constructFeatureItem } from "@/utils/scriptUtils/constructItem.js";
-import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
-import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
-import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
 
-/**
- * Sleep until a specific epoch time in milliseconds
- */
 function sleepUntil(epochMs: number): Promise<void> {
 	const delay = epochMs - Date.now();
-
-	if (delay <= 0) {
-		return Promise.resolve();
-	}
-
+	if (delay <= 0) return Promise.resolve();
 	return new Promise((resolve) => setTimeout(resolve, delay));
 }
 
-const messagesFeature = constructFeatureItem({
-	featureId: TestFeature.Messages,
-	includedUsage: 100,
-});
-
-const freeProd = constructProduct({
-	type: "free",
-	isDefault: false,
-	items: [messagesFeature],
-});
-
 const testCase = "check-loose-expiry";
 
-describe(`${chalk.yellowBright(`${testCase}: expiring loose entitlement check`)}`, () => {
-	const customerBasic = `${testCase}-basic`;
-	const customerProductMix = `${testCase}-prod`;
-	const customerResetMix = `${testCase}-reset`;
+test.concurrent(chalk.yellowBright(`${testCase}-basic: expiring loose entitlement should be allowed before expiry, then denied after`), async () => {
+	const customerId = `${testCase}-basic`;
+	const messagesItem = items.monthlyMessages({ includedUsage: 100 });
+	const freeProd = products.base({ id: "free", items: [messagesItem] });
 
-	const autumnV2 = new AutumnInt({ version: ApiVersion.V2_0 });
-	const autumnV1 = new AutumnInt({ version: ApiVersion.V1_2 });
-
-	beforeAll(async () => {
-		// Setup products
-		await initProductsV0({
-			ctx,
-			products: [freeProd],
-			prefix: testCase,
-		});
-
-		// Setup customers only
-		await initCustomerV3({
-			ctx,
-			customerId: customerBasic,
-			withTestClock: false,
-		});
-
-		await initCustomerV3({
-			ctx,
-			customerId: customerProductMix,
-			withTestClock: false,
-		});
-
-		await initCustomerV3({
-			ctx,
-			customerId: customerResetMix,
-			withTestClock: false,
-		});
+	const { ctx } = await initScenario({
+		customerId,
+		setup: [s.customer({ testClock: false }), s.products({ list: [freeProd] })],
+		actions: [],
 	});
 
-	test("basic: expiring loose entitlement should be allowed before expiry, then denied after", async () => {
-		const expiresAt = Date.now() + 3000;
+	const autumnV1 = new AutumnInt({
+		version: ApiVersion.V1_2,
+		secretKey: ctx.orgSecretKey,
+	});
+	const autumnV2 = new AutumnInt({
+		version: ApiVersion.V2_0,
+		secretKey: ctx.orgSecretKey,
+	});
 
-		// Create expiring loose entitlement
-		await autumnV1.balances.create({
-			customer_id: customerBasic,
+	const expiresAt = Date.now() + 3000;
+
+	await autumnV1.balances.create({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+		granted_balance: 500,
+		expires_at: expiresAt,
+	});
+
+	const resBefore = (await autumnV2.check({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+	})) as unknown as CheckResponseV2;
+
+	expect(resBefore).toMatchObject({
+		allowed: true,
+		customer_id: customerId,
+		balance: {
+			plan_id: null,
 			feature_id: TestFeature.Messages,
 			granted_balance: 500,
-			expires_at: expiresAt,
-		});
-
-		// Check before expiry
-		const resBefore = (await autumnV2.check({
-			customer_id: customerBasic,
-			feature_id: TestFeature.Messages,
-		})) as unknown as CheckResponseV2;
-
-		expect(resBefore.allowed).toBe(true);
-		expect(resBefore.customer_id).toBe(customerBasic);
-		expect(resBefore.balance).toBeDefined();
-		expect(resBefore.balance?.plan_id).toBeNull();
-		expect(resBefore.balance?.feature_id).toBe(TestFeature.Messages);
-		expect(resBefore.balance?.granted_balance).toBe(500);
-		expect(resBefore.balance?.current_balance).toBe(500);
-		expect(resBefore.balance?.usage).toBe(0);
-		expect(resBefore.balance?.unlimited).toBe(false);
-
-		// Wait until expiry
-		await sleepUntil(expiresAt + 1000);
-
-		// Check after expiry
-		const resAfter = (await autumnV2.check({
-			customer_id: customerBasic,
-			feature_id: TestFeature.Messages,
-		})) as unknown as CheckResponseV2;
-
-		expect(resAfter.allowed).toBe(false);
-		expect(resAfter.balance).toBeNull();
+			current_balance: 500,
+			usage: 0,
+			unlimited: false,
+		},
 	});
 
-	test("product-mix: should combine product and expiring loose ent, then only product after expiry", async () => {
-		const expiresAt = Date.now() + 3000;
+	await sleepUntil(expiresAt + 1000);
 
-		// Attach product with 100 messages
-		await autumnV1.attach({
-			customer_id: customerProductMix,
-			product_id: freeProd.id,
-		});
+	const resAfter = (await autumnV2.check({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+	})) as unknown as CheckResponseV2;
 
-		// Create expiring loose entitlement with 200 messages
-		await autumnV1.balances.create({
-			customer_id: customerProductMix,
-			feature_id: TestFeature.Messages,
-			granted_balance: 200,
-			expires_at: expiresAt,
-		});
+	expect(resAfter).toMatchObject({
+		allowed: false,
+		balance: null,
+	});
+});
 
-		// Check before expiry
-		const resBefore = (await autumnV2.check({
-			customer_id: customerProductMix,
-			feature_id: TestFeature.Messages,
-		})) as unknown as CheckResponseV2;
+test.concurrent(chalk.yellowBright(`${testCase}-product-mix: should combine product and expiring loose ent, then only product after expiry`), async () => {
+	const customerId = `${testCase}-product-mix`;
+	const messagesItem = items.monthlyMessages({ includedUsage: 100 });
+	const freeProd = products.base({ id: "free", items: [messagesItem] });
 
-		expect(resBefore.allowed).toBe(true);
-		expect(resBefore.balance?.granted_balance).toBe(300); // 100 from product + 200 from loose
-		expect(resBefore.balance?.current_balance).toBe(300);
-
-		// Wait until expiry
-		await sleepUntil(expiresAt + 1000);
-
-		// Check after expiry
-		const resAfter = (await autumnV2.check({
-			customer_id: customerProductMix,
-			feature_id: TestFeature.Messages,
-		})) as unknown as CheckResponseV2;
-
-		expect(resAfter.allowed).toBe(true);
-		expect(resAfter.balance?.granted_balance).toBe(100); // Only product balance remains
-		expect(resAfter.balance?.current_balance).toBe(100);
+	const { autumnV1, ctx } = await initScenario({
+		customerId,
+		setup: [s.customer({ testClock: false }), s.products({ list: [freeProd] })],
+		actions: [s.attach({ productId: freeProd.id })],
 	});
 
-	test("reset-mix: should combine expiring and resetting loose ents, then only resetting after expiry", async () => {
-		const expiresAt = Date.now() + 3000;
+	const autumnV2 = new AutumnInt({
+		version: ApiVersion.V2_0,
+		secretKey: ctx.orgSecretKey,
+	});
 
-		// Create expiring loose entitlement
-		await autumnV1.balances.create({
-			customer_id: customerResetMix,
-			feature_id: TestFeature.Messages,
-			granted_balance: 200,
-			expires_at: expiresAt,
-		});
+	const expiresAt = Date.now() + 3000;
 
-		// Create resetting loose entitlement (no expiry)
-		await autumnV1.balances.create({
-			customer_id: customerResetMix,
-			feature_id: TestFeature.Messages,
-			granted_balance: 100,
+	await autumnV1.balances.create({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+		granted_balance: 200,
+		expires_at: expiresAt,
+	});
+
+	const resBefore = (await autumnV2.check({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+	})) as unknown as CheckResponseV2;
+
+	expect(resBefore).toMatchObject({
+		allowed: true,
+		balance: {
+			granted_balance: 300, // 100 from product + 200 from loose
+			current_balance: 300,
+		},
+	});
+
+	await sleepUntil(expiresAt + 1000);
+
+	const resAfter = (await autumnV2.check({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+	})) as unknown as CheckResponseV2;
+
+	expect(resAfter).toMatchObject({
+		allowed: true,
+		balance: {
+			granted_balance: 100, // Only product balance remains
+			current_balance: 100,
+		},
+	});
+});
+
+test.concurrent(chalk.yellowBright(`${testCase}-reset-mix: should combine expiring and resetting loose ents, then only resetting after expiry`), async () => {
+	const customerId = `${testCase}-reset-mix`;
+	const messagesItem = items.monthlyMessages({ includedUsage: 100 });
+	const freeProd = products.base({ id: "free", items: [messagesItem] });
+
+	const { autumnV1, ctx } = await initScenario({
+		customerId,
+		setup: [s.customer({ testClock: false }), s.products({ list: [freeProd] })],
+		actions: [],
+	});
+
+	const autumnV2 = new AutumnInt({
+		version: ApiVersion.V2_0,
+		secretKey: ctx.orgSecretKey,
+	});
+
+	const expiresAt = Date.now() + 3000;
+
+	await autumnV1.balances.create({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+		granted_balance: 200,
+		expires_at: expiresAt,
+	});
+
+	await autumnV1.balances.create({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+		granted_balance: 100,
+		reset: {
+			interval: ResetInterval.Month,
+		},
+	});
+
+	const resBefore = (await autumnV2.check({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+	})) as unknown as CheckResponseV2;
+
+	expect(resBefore).toMatchObject({
+		allowed: true,
+		balance: {
+			granted_balance: 300, // 200 expiring + 100 resetting
+			current_balance: 300,
+		},
+	});
+
+	await sleepUntil(expiresAt + 1000);
+
+	const resAfter = (await autumnV2.check({
+		customer_id: customerId,
+		feature_id: TestFeature.Messages,
+	})) as unknown as CheckResponseV2;
+
+	expect(resAfter).toMatchObject({
+		allowed: true,
+		balance: {
+			granted_balance: 100, // Only resetting balance remains
+			current_balance: 100,
 			reset: {
 				interval: ResetInterval.Month,
 			},
-		});
-
-		// Check before expiry
-		const resBefore = (await autumnV2.check({
-			customer_id: customerResetMix,
-			feature_id: TestFeature.Messages,
-		})) as unknown as CheckResponseV2;
-
-		expect(resBefore.allowed).toBe(true);
-		expect(resBefore.balance?.granted_balance).toBe(300); // 200 expiring + 100 resetting
-		expect(resBefore.balance?.current_balance).toBe(300);
-
-		// Wait until expiry
-		await sleepUntil(expiresAt + 1000);
-
-		// Check after expiry
-		const resAfter = (await autumnV2.check({
-			customer_id: customerResetMix,
-			feature_id: TestFeature.Messages,
-		})) as unknown as CheckResponseV2;
-
-		expect(resAfter.allowed).toBe(true);
-		expect(resAfter.balance?.granted_balance).toBe(100); // Only resetting balance remains
-		expect(resAfter.balance?.current_balance).toBe(100);
-		expect(resAfter.balance?.reset).toBeDefined();
-		expect(resAfter.balance?.reset?.interval).toBe(ResetInterval.Month);
+		},
 	});
 });

server/tests/balances/cron/loose-reset.test.ts

@@ -84,12 +84,10 @@ describe(`${chalk.yellowBright("loose-reset: test getActiveResetPassed for loose
 		// Wait for sync
 		await new Promise((resolve) => setTimeout(resolve, 2000));
 
-		const fullCustomer = await CusService.getFull({
-			db: ctx.db,
-			idOrInternalId: customerId,
-			orgId: ctx.org.id,
-			env: ctx.env,
-		});
+	const fullCustomer = await CusService.getFull({
+		ctx,
+		idOrInternalId: customerId,
+	});
 
 		const cusEnt = await findCustomerEntitlement({
 			ctx,

server/tests/balances/utils/findCustomerEntitlement.ts

@@ -20,10 +20,8 @@ export const findCustomerEntitlement = async ({
 	fullCustomer =
 		fullCustomer ||
 		(await CusService.getFull({
-			db: ctx.db,
+			ctx,
 			idOrInternalId: customerId,
-			orgId: ctx.org.id,
-			env: ctx.env,
 		}));
 
 	const cusEnts = fullCustomerToCustomerEntitlements({

server/tests/balances/utils/getCustomerEntitlement.ts

@@ -12,10 +12,8 @@ export const findCustomerEntitlement = async ({
 	featureId?: string;
 }) => {
 	const fullCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 
 	const cusEnts = fullCustomerToCustomerEntitlements({

server/tests/balances/utils/getCustomerEntitlements.ts

@@ -12,10 +12,8 @@ export const getCustomerEntitlement = async ({
 	featureId?: string;
 }) => {
 	const fullCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 
 	const cusEnts = fullCustomerToCustomerEntitlements({

server/tests/integration/billing/attach/edge-cases/v1-v2-compatibility/prepaid/v2-attach-v1-update-quantity.test.ts

@@ -51,10 +51,8 @@ const getStripePrepaidSubscriptionItem = async ({
 	const stripeCli = createStripeCli({ org: ctx.org, env: ctx.env });
 
 	const fullCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 
 	const stripeCustomerId =

server/tests/integration/billing/legacy/attach/attach-misc.test.ts

@@ -1,5 +1,5 @@
 import { expect, test } from "bun:test";
-import { type AppEnv, ErrCode } from "@autumn/shared";
+import { ErrCode } from "@autumn/shared";
 import { TestFeature } from "@tests/setup/v2Features.js";
 import { getMainCusProduct } from "@tests/utils/cusProductUtils/cusProductUtils.js";
 import { expectProductAttached } from "@tests/utils/expectUtils/expectProductAttached.js";
@@ -87,7 +87,7 @@ test.concurrent(`${chalk.yellowBright("attach-misc: convert collection method fr
 		actions: [],
 	});
 
-	const { db, org, env, stripeCli } = ctx;
+	const { stripeCli } = ctx;
 
 	// Attach with invoice option
 	const res = await autumnV1.attach({
@@ -113,10 +113,8 @@ test.concurrent(`${chalk.yellowBright("attach-misc: convert collection method fr
 	await timeout(10000);
 
 	const cusProduct = await getMainCusProduct({
-		db,
+		ctx,
 		customerId,
-		orgId: org.id,
-		env: env as AppEnv,
 		productGroup: pro.group ?? undefined,
 	});
 

server/tests/integration/billing/legacy/attach/checkout/legacy-checkout-advanced.test.ts

@@ -326,10 +326,8 @@ test.concurrent(`${chalk.yellowBright("legacy-checkout-adv 4: separate subs due
 
 	// Verify separate subscriptions
 	const fullCus = await CusService.getFull({
+		ctx,
 		idOrInternalId: customerId,
-		db,
-		orgId: org.id,
-		env,
 	});
 
 	const cusProducts = fullCus.customer_products;
@@ -392,10 +390,8 @@ test.concurrent(`${chalk.yellowBright("legacy-checkout-adv 4: separate subs due
 
 	// Verify add-on is on entity 2's subscription
 	const fullCusAfterAddOn = await CusService.getFull({
+		ctx,
 		idOrInternalId: customerId,
-		db,
-		orgId: org.id,
-		env,
 	});
 
 	const addOnProd = fullCusAfterAddOn.customer_products.find(

server/tests/integration/billing/legacy/attach/invoice/legacy-attach-invoice-mode-advanced.test.ts

@@ -304,10 +304,8 @@ test.concurrent(`${chalk.yellowBright("legacy-inv-mode-adv 4: separate subs due
 
 	// Verify separate subscriptions
 	const fullCus = await CusService.getFull({
+		ctx,
 		idOrInternalId: customerId,
-		db,
-		orgId: org.id,
-		env,
 	});
 
 	const cusProducts = fullCus.customer_products;

server/tests/integration/billing/legacy/attach/new/legacy-merge-interval.test.ts

@@ -80,8 +80,7 @@ test.concurrent(`${chalk.yellowBright("legacy-merge-interval 1: entity attach mi
 
 	// Verify Stripe subscription period_end matches
 	const sub = await getCusSub({
-		db: ctx.db,
-		org: ctx.org,
+		ctx,
 		customerId,
 		productId: pro.id,
 	});
@@ -154,8 +153,7 @@ test.concurrent(`${chalk.yellowBright("legacy-merge-interval 2: entity attach an
 
 	// Verify Stripe subscription period_end matches
 	const sub = await getCusSub({
-		db: ctx.db,
-		org: ctx.org,
+		ctx,
 		customerId,
 		productId: proAnnual.id,
 	});
@@ -230,8 +228,7 @@ test.concurrent(`${chalk.yellowBright("legacy-merge-interval 3: entity attach an
 
 	// Verify at least one subscription item has matching period_end
 	const sub = await getCusSub({
-		db: ctx.db,
-		org: ctx.org,
+		ctx,
 		customerId,
 		productId: proAnnual.id,
 	});

server/tests/integration/billing/legacy/attach/separate/legacy-separate.test.ts

@@ -99,10 +99,8 @@ test.concurrent(`${chalk.yellowBright("legacy-separate 1: separate subs via invo
 
 	// Verify different subscription IDs per entity
 	const fullCus = await CusService.getFull({
+		ctx,
 		idOrInternalId: customerId,
-		db: ctx.db,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 	const cusProducts = fullCus.customer_products;
 	const entity1Prod = cusProducts.find((cp) => cp.entity_id === entities[0].id);
@@ -241,10 +239,8 @@ test.concurrent(`${chalk.yellowBright("legacy-separate 2: separate subs via forc
 
 	// Verify different subscription IDs per entity
 	let fullCus = await CusService.getFull({
+		ctx,
 		idOrInternalId: customerId,
-		db: ctx.db,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 	let cusProducts = fullCus.customer_products;
 	const entity1Prod = cusProducts.find((cp) => cp.entity_id === entities[0].id);
@@ -305,10 +301,8 @@ test.concurrent(`${chalk.yellowBright("legacy-separate 2: separate subs via forc
 
 	// Verify add-on's sub ID matches entity 2's sub ID
 	fullCus = await CusService.getFull({
+		ctx,
 		idOrInternalId: customerId,
-		db: ctx.db,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 	cusProducts = fullCus.customer_products;
 	const addOnProd = cusProducts.find((cp) => cp.product.id === addOn.id);

server/tests/integration/billing/legacy/attach/upgrade/legacy-upgrade.test.ts

@@ -226,8 +226,7 @@ test.concurrent(`${chalk.yellowBright("legacy-upgrade 2: upgrade monthly to annu
 
 	// Verify Stripe subscription period_end matches checkout preview
 	const sub = await getCusSub({
-		db: ctx.db,
-		org: ctx.org,
+		ctx,
 		customerId,
 		productId: proAnnual.id,
 	});
@@ -297,8 +296,7 @@ test.concurrent(`${chalk.yellowBright("legacy-upgrade 3: upgrade monthly to annu
 
 	// Verify Stripe subscription period_end matches checkout preview
 	const sub = await getCusSub({
-		db: ctx.db,
-		org: ctx.org,
+		ctx,
 		customerId,
 		productId: proAnnual.id,
 	});

server/tests/integration/billing/migrations/migrate-addons.test.ts

@@ -332,9 +332,7 @@ test.concurrent(`${chalk.yellowBright("migrate-addons-3: same add-on attached tw
 	// Verify migrated state
 
 	const fullCustomer = await CusService.getFull({
-		db: ctx.db,
-		orgId: ctx.org.id,
-		env: ctx.env,
+		ctx,
 		idOrInternalId: customerId,
 	});
 

server/tests/integration/billing/stripe-webhooks/invoice-created/invoice-created-discounts.test.ts

@@ -25,10 +25,8 @@ const getStripeInfo = async ({ customerId }: { customerId: string }) => {
 	const stripeCli = createStripeCli({ org: ctx.org, env: ctx.env });
 
 	const fullCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 
 	const stripeCustomerId =

server/tests/integration/billing/stripe-webhooks/invoice-paid/send-email-receipt.test.ts

@@ -64,10 +64,8 @@ test(`${chalk.yellowBright("invoice.paid: sends email receipt when send_email_re
 
 	// Verify the update was applied (both Autumn and Stripe)
 	const updatedCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 	expect(updatedCustomer.send_email_receipts).toBe(true);
 	expect(updatedCustomer.email).toBe(testEmail);
@@ -255,10 +253,8 @@ test(`${chalk.yellowBright("invoice.paid: does NOT send email receipt when custo
 
 	// Verify send_email_receipts was enabled
 	const updatedCustomer = await CusService.getFull({
-		db: ctx.db,
+		ctx,
 		idOrInternalId: customerId,
-		orgId: ctx.org.id,
-		env: ctx.env,
 	});
 	expect(updatedCustomer.send_email_receipts).toBe(true);