comi-logic/cfw-autumn
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: comments and tests

Browse Source
This commit is contained in:
John Yeo
2026-04-20 12:33:44 +01:00
parent 59640a95c7
commit 7e820d4591
53 changed files with 2243 additions and 1288 deletions
Download Patch File Download Diff File Expand all files Collapse all files

378
AGENTS.md
View File

@@ -1,5 +1,50 @@
<!-- Generated by ai-sync. Edit ai/rules/ instead. -->
# Scope Cache Refresh Changes Safely
When changing cache-refresh behavior for API routes:
- Prefer editing `server/src/honoMiddlewares/refreshCacheConfigs.ts` to add or remove route entries.
- Do **not** remove or bypass logic in `server/src/honoMiddlewares/refreshCacheMiddleware.ts` unless the user explicitly asks to change middleware behavior globally.
- If the request is ambiguous, ask whether they want a route-level config change or a global middleware behavior change.
# Cache Version Increment Policy
Treat `cache_version` as a DB-side stale-sync guard for `syncItemV4`, not a general cache mutation counter.
## Increment cache_version only when needed
Increment only for DB updates that must not be overwritten by stale sync payloads read from cache (for example lifecycle or billing transitions).
## Do not increment on cache-side/runtime patch paths
For runtime balance/reset/cache patch flows, do not bump `cache_version` in cache writers or Lua update scripts.
Examples:
- FullSubject cache patch helpers (`updateSubjectBalanceCache`, reset/deduction cache patch sync helpers)
- FullSubject Lua cache update scripts (`updateSubjectBalances`)
- Update-balance runtime paths that patch Redis and then sync
## Call-site rule for CusEntService.update
When using `CusEntService.update(...)` in runtime FullSubject cache patch flows, set `incrementCacheVersion` explicitly.
- Use `incrementCacheVersion: false` for routine balance/reset/adjustment updates that are mirrored to Redis.
- Use `incrementCacheVersion: true` only for intentional DB-side stale-write protection transitions.
## Why
Incorrect version bumps create `CACHE_VERSION_MISMATCH` conflicts in `syncItemV4`, causing repeated invalidation and stale/lost update behavior.
## Legacy exception (review-required)
There is a legacy-compatibility exception in the adjust-balance flow:
- `adjustBalanceDbAndCache` currently uses `CusEntService.increment/decrement`, which increments `cache_version`.
- Treat this as a reviewable legacy action, not a pattern to copy into new runtime/cache patch paths.
- Any new or refactored runtime balance/reset/cache patch code should continue following this policy and avoid adding new cache-version bumps by default.
## Project Context System
Projects maintain state in `.context/<project>/` folders across sessions. Tasks are optional parallel workstreams within a project.
@@ -45,336 +90,3 @@ Always REWRITE STATUS.md completely rather than append.
### Task handoff
When a task is being handed to another agent, ensure the task's STATUS.md is up to date -- it's the handoff artifact.
# scripts-v2 Conventions
## File naming
- Always kebab-case: `standardize-stripe-state.ts`, not `standardizeStripeState.ts`
- Scripts go in `runs/{org-name}/`: e.g. `runs/mintlify/fix-prices.ts`
- **Never use `index.ts`** for business logic — name files after what the function does
## Imports
- `script` / `step`: from `"../lib"` or `"../../lib"` (relative to script)
- Types/enums: from `"@autumn/shared"` (AppEnv, FullProduct, Customer, etc.)
- Server services: from `"@autumn/server/src/internal/..."` (CusService, ProductService, etc.)
- Lib utilities: from `"@autumn-cloud/lib/..."` (createScriptContext, search helpers)
## Function signatures
- Always use named object params: `fn({ db, orgId })` not `fn(db, orgId)`
- This applies to all functions, even single-argument ones
## Three categories of functions
### 1. Steps (`step()`)
Named phases of work. Declare their own `id` and typed input/output. Called via `s.run()`.
```typescript
import { step, type ScriptContext } from "../../../lib";
export const standardizeBasePrices = step({
id: "standardize-base-prices",
run: async ({ ctx, group, priceCache }: {
ctx: ScriptContext;
group: SubscriptionGroup;
priceCache: StripePriceCache;
}): Promise<BasePriceMismatch[]> => {
// business logic — returns typed data
return mismatches;
},
});
```
### 2. Orchestrators (steps that compose other steps)
Their `run` function takes `s` in addition to `ctx` + payload. They call `s.run()` to compose sub-steps.
```typescript
export const processCustomer = step({
id: "process-customer",
run: async ({ s, ctx, customerId, priceCache }: {
s: ScriptUtils;
ctx: ScriptContext;
customerId: string;
priceCache: StripePriceCache;
}) => {
const groups = await s.run(loadSubscriptionGroups, { customerId });
const result = await s.run(standardizeBasePrices, { group, priceCache });
return { customerId, status: "ok", ...result };
},
});
```
### 3. Utilities (plain functions)
Helpers that aren't meaningful workflow phases. Regular async functions, no `step()`.
Examples: `formatSubscriptionUpdate`, `listStripeSubscriptions`, `buildCacheKey`.
## Params rules
- `dryRun` lives on `ctx.dryRun` — never pass as a separate param
- `data` is accessed via `s.data` in orchestrators only — never threaded to leaf steps
- If a step needs more than `ctx` + 3 business params, it's doing too much
## Result convention (Trigger.dev-style ok pattern)
- Every function returns typed data — never mutates caller's objects
- Fallible steps return a discriminated union: `{ ok: true; ... } | { ok: false; reason: string }`
- The caller checks `result.ok` — just like Trigger.dev's `triggerAndWait()` result
- Infallible steps return their typed data directly (e.g. `BasePriceMismatch[]`)
## Script structure
- `dryRun: true` by default. Flip to `false` only when ready to mutate
- All tunables go in the `params` block — never use CLI args or scattered top-of-file constants
- Use `s.step()` for inline phases, `s.run(step, payload)` for named step functions
- Use `s.batch()` for iterating over items — return objects from fn to collect results
- Batch fn receives `{ item, ctx, s }` — use `s.run()` inside batch for sub-steps
- Use `output: "csv"` on batch to auto-write results
- Use `checkpoint: true` on batch for resume across restarts
- Always use `ctx.logger.info()` — never `console.log()`
## Large scripts — folder organization
For non-trivial work, create a folder under `runs/{org-name}/` and split into three subfolders:
```
runs/mintlify/standardize-stripe/
standardize-stripe-state.ts # entrypoint (script)
config.ts # constants
orchestrators/ # steps that compose other steps via s.run()
process-customer.ts
process-subscription-group.ts
steps/ # leaf steps (pure business logic, no s)
load-customer-ids.ts
load-exclude-list.ts
load-subscription-groups.ts
guards.ts
check-customer-flags.ts
standardize-base-prices.ts
standardize-usage-prices.ts
update-stripe-subscription.ts
cancel-monthly-companion.ts
utils/ # plain functions (no step(), not logged)
find-or-create-matching-stripe-price.ts
format-subscription-update.ts
format-customer-result.ts
```
**Folder rules:**
- **`orchestrators/`** — takes `{ s, ctx, ... }`, calls `s.run()` to compose steps
- **`steps/`** — takes `{ ctx, ... }`, returns typed data, no `s`
- **`utils/`** — plain functions, no `step()`, no logging
- Root: only the entrypoint and config
**Entrypoint pattern:**
```typescript
run: async ({ s, ctx }) => {
const customerIds = await s.step({
id: "load customers",
fn: () => loadCustomerIds({ ctx }),
});
await s.batch({
id: "standardize",
items: customerIds,
output: "csv",
fn: async ({ item: customerId, s: batchS }) => {
return batchS.run(processCustomer, { customerId, priceCache });
},
});
},
```
## Export pattern (for Trigger.dev compatibility)
```typescript
const myScript = await script({ ... });
export default myScript;
if (import.meta.main) await myScript.run();
```
# scripts-v2 API
## script()
Self-executing entry point. The file IS the execution — hit ctrl+enter to run via `./run.sh`.
```typescript
import { script } from "../lib";
import { AppEnv } from "@autumn/shared";
const myScript = await script({
id: "org-name/script-name",
org: "autumn_org_id",
env: AppEnv.Sandbox,
dryRun: true,
loadProducts: true,
description: "What this does",
params: {
concurrency: 5,
limit: 1 as number | null,
only: null as string[] | null,
},
run: async ({ s, ctx }) => {
// s = script utilities (step, batch, run, log, data)
// ctx = data and services (extends AutumnContext)
},
});
export default myScript;
if (import.meta.main) await myScript.run();
```
## step()
Define a named step with typed input/output. Steps declare their own `id` — the caller just provides the business payload.
```typescript
import { step, type ScriptContext } from "../lib";
export const loadCustomerGroups = step({
id: "load-customer-groups",
run: async ({ ctx, customerId }: {
ctx: ScriptContext;
customerId: string;
}) => {
// business logic
return { fullCustomer, groups };
},
});
```
Orchestrator steps (compose sub-steps) take `s` in addition to `ctx`:
```typescript
export const processCustomer = step({
id: "process-customer",
run: async ({ s, ctx, customerId }: {
s: ScriptUtils;
ctx: ScriptContext;
customerId: string;
}) => {
const { groups } = await s.run(loadCustomerGroups, { customerId });
// ... compose more steps
},
});
```
## s.run()
Execute a step, auto-injecting `ctx` and `s`. The caller only passes business-specific payload.
```typescript
// step defined elsewhere with id and typed run function
const result = await s.run(standardizeBasePrices, { group, priceCache });
// ^^ knows its own id ^^ only business payload
```
TypeScript strips `ctx` and `s` from the call-site — you get autocomplete on just the business params and a typed return value.
## ctx (ScriptContext)
Extends `AutumnContext` — pass directly to any server function.
| Field | Type | Description |
|-------|------|-------------|
| `ctx.db` | DrizzleCli | Postgres database client |
| `ctx.stripe` | Stripe | Stripe client (env/Connect-aware) |
| `ctx.org` | Organization | Loaded org object |
| `ctx.env` | AppEnv | Live or Sandbox |
| `ctx.features` | Feature[] | All features for this org |
| `ctx.products` | FullProduct[] | All products (empty if loadProducts: false) |
| `ctx.dryRun` | boolean | Whether this is a dry run |
| `ctx.params` | P | Typed params from your params block |
| `ctx.logger` | Logger | Pino-based logger (info/warn/error/debug/child) |
## s (ScriptUtils)
| Field | Type | Description |
|-------|------|-------------|
| `s.step` | function | Inline named section with auto-logging |
| `s.batch` | function | Batch processor with output, checkpoint, buffered logging |
| `s.run` | function | Execute a `step()` with auto-injected ctx/s |
| `s.log` | ScriptLog | Logger with table/json/section formatting |
| `s.data` | ScriptData | Data layer: store, list, write, read |
## s.step()
For inline phases (not defined as separate `step()` functions):
```typescript
const data = await s.step({
id: "load customers",
fn: async () => CusService.getByOrg({ db: ctx.db, orgId: ctx.org.id, env: ctx.env }),
});
```
## s.batch()
Batch fn receives `{ item, ctx, s }` — use `s.run()` inside for sub-steps:
```typescript
await s.batch({
id: "process-customers",
items: customerIds,
output: "csv",
checkpoint: true,
fn: async ({ item: customerId, ctx, s }) => {
return s.run(processCustomer, { customerId, priceCache });
},
concurrency: 5,
limit: 10,
onError: "continue",
});
```
### BatchResult
```typescript
const result = await s.batch({ ... });
result.processed // number of items successfully processed
result.skipped // number skipped (checkpoint or skipIf)
result.errors // number of errors (when onError: "continue")
result.rows // collected return values from fn
result.duration // total ms
```
## s.data (ScriptData)
Data dir lives at `runs/<script-id>/data/` with four subdirs:
- `inputs/` — curated inputs (git-committed)
- `outputs/` — script results (git-committed)
- `state/` — checkpoints, resume tracking (gitignored)
- `logs/` — append-only audit trail (gitignored)
```typescript
const finished = s.data.store<boolean>("state/finished");
finished.has("cus_123");
finished.set("cus_123", true);
finished.flush();
const results = s.data.list<AuditRow>("outputs/audit-results");
results.push({ customerId: "cus_123", status: "ok" });
s.data.write("outputs/report.csv", csvContent);
const config = s.data.read("inputs/exclude-orgs.csv");
```
## params pattern
All tunables in one typed block. Edit and re-run — no CLI args.
```typescript
params: {
concurrency: 5,
limit: 1 as number | null,
only: null as string[] | null,
},
```
Batch automatically inherits `concurrency`, `limit`, `only` from params.
## Running
```bash
./run.sh runs/org-name/my-script.ts
LOCAL_OVERRIDE=true ./run.sh runs/org-name/my-script.ts
```

4
server/src/_luaScriptsV2/fullSubject/updateCustomerDataV2.lua
View File

@@ -36,6 +36,10 @@ if not current_raw then
end
local cached = cjson.decode(current_raw)
if not cached or not cached.customer then
return cjson.encode({ success = false, cache_miss = true })
end
local updated_fields = {}
for field_name, field_value in pairs(updates) do

16
server/src/_luaScriptsV2/fullSubjectDeduction/deductFromSubjectBalances.lua
View File

@@ -76,18 +76,16 @@ local lock = params.lock
local unwind_value = params.unwind_value
local lock_receipt_key = params.lock_receipt_key
local empty_logs = cjson.decode('[]')
-- Initialize context with in-memory state from Redis
if #customer_entitlement_deductions == 0 then
return cjson.encode({
updates = {},
rollover_updates = {},
modified_customer_entitlement_ids = empty_logs,
mutation_logs = empty_logs,
modified_customer_entitlement_ids = new_empty_array(),
mutation_logs = new_empty_array(),
remaining = 0,
error = cjson.null,
logs = empty_logs,
logs = new_empty_array(),
})
end
@@ -103,8 +101,8 @@ if #(context.missing_customer_entitlement_ids or {}) > 0 then
error = 'SUBJECT_BALANCE_NOT_FOUND',
updates = {},
rollover_updates = {},
modified_customer_entitlement_ids = empty_logs,
mutation_logs = empty_logs,
modified_customer_entitlement_ids = new_empty_array(),
mutation_logs = new_empty_array(),
remaining = 0,
logs = context.logs,
missing_customer_entitlement_ids = context.missing_customer_entitlement_ids,
@@ -125,7 +123,7 @@ if not is_nil(unwind_value) and safe_number(unwind_value) > 0 then
error = unwind_result.error,
updates = {},
rollover_updates = {},
modified_customer_entitlement_ids = empty_logs,
modified_customer_entitlement_ids = new_empty_array(),
mutation_logs = context.mutation_logs or cjson.decode('[]'),
remaining = 0,
logs = context.logs,
@@ -203,7 +201,7 @@ if remaining_amount > 0 and overage_behaviour == 'reject' then
feature_id = feature_id,
remaining = remaining_amount,
updates = {},
modified_customer_entitlement_ids = empty_logs,
modified_customer_entitlement_ids = new_empty_array(),
mutation_logs = mutation_logs,
logs = context.logs
})

4
server/src/_luaScriptsV2/luaUtils.lua
View File

@@ -42,3 +42,7 @@ local function sorted_keys(tbl)
table.sort(keys)
return keys
end
local function new_empty_array()
return cjson.decode('[]')
end

2
server/src/external/redis/initUtils/createRedisClient.ts vendored
View File

@@ -1,5 +1,5 @@
import { Redis } from "ioredis";
import { instrumentRedis } from "../../../utils/otel/instrumentRedis.js";
import { instrumentRedis } from "../otel/instrumentRedis.js";
import { cacheBackupUrl } from "./redisConfig.js";
import { registerRedisCommands } from "./registerRedisCommands.js";

52
server/src/external/redis/otel/emitRedisSlowLog.ts vendored Normal file
View File

@@ -0,0 +1,52 @@
import { logger } from "@/external/logtail/logtailUtils.js";
import { addRedisToLogs } from "@/utils/logging/addContextToLogs.js";
import type { RedisKeyContext } from "./parseRedisKeyContext.js";
import type { ResolvedThresholds } from "./redisSlowlogConfig.js";
import { shouldEmitSlowLog } from "./slowLogRateLimit.js";
/**
* Emits a structured `redis_slow_command` log for commands that exceeded
* their severe threshold. Rate-limited per-operation to avoid flooding.
*/
export const emitRedisSlowLog = ({
operation,
durationMs,
thresholds,
keyContext,
region,
key,
}: {
operation: string;
durationMs: number;
thresholds: ResolvedThresholds;
keyContext: RedisKeyContext;
region?: string;
key?: string;
}): void => {
try {
if (!shouldEmitSlowLog({ operation })) return;
const slowMs = thresholds.slowMs;
const breachRatio = slowMs > 0 ? durationMs / slowMs : 0;
addRedisToLogs({
logger,
redisData: {
operation,
duration_ms: durationMs,
slow_ms: slowMs,
base_slow_ms: thresholds.baseSlowMs,
region_baseline_ms: thresholds.regionBaselineMs,
severe_ms: thresholds.severeMs,
breach_ratio: breachRatio,
region,
key,
org_id: keyContext.orgId,
customer_id: keyContext.customerId,
entity_id: keyContext.entityId,
},
}).warn({ type: "redis_slow_command" }, "Redis slow command");
} catch {
// swallow — telemetry must never break app commands
}
};

194
server/src/utils/otel/instrumentRedis.ts → server/src/external/redis/otel/instrumentRedis.ts vendored
View File

@@ -7,7 +7,16 @@ import {
trace,
} from "@opentelemetry/api";
import type { Command, Redis } from "ioredis";
import { otelConfig } from "./otelConfig.js";
import { otelConfig } from "@/utils/otel/otelConfig.js";
import { emitRedisSlowLog } from "./emitRedisSlowLog.js";
import {
parseRedisKeyContext,
type RedisKeyContext,
} from "./parseRedisKeyContext.js";
import {
type ResolvedThresholds,
resolveThresholds,
} from "./redisSlowlogConfig.js";
const TRACER_NAME = "autumn.redis";
const INSTRUMENTED = new WeakSet<object>();
@@ -27,14 +36,61 @@ const SKIP_COMMANDS = new Set([
"disconnect",
"cluster",
"command",
// SCRIPT LOAD / FLUSH / EXISTS fires rarely on connection bootstrap and
// can spike into 100s of ms — noise that swamps the slowlog.
"script",
// Custom Lua commands are traced via defineCommand wrapper,
// so skip the underlying evalsha/eval to avoid double-spanning.
"evalsha",
"eval",
]);
type SpanContext = {
span: Span;
startedAt: number;
thresholds: ResolvedThresholds;
keyContext: RedisKeyContext;
operation: string;
region?: string;
key?: string;
};
/** Ends a span safely — never throws. */
const finalizeSpan = ({ span, error }: { span: Span; error?: unknown }) => {
const finalizeSpan = ({
spanCtx,
error,
}: {
spanCtx: SpanContext;
error?: unknown;
}) => {
const { span, startedAt, thresholds, keyContext, operation, region, key } =
spanCtx;
try {
const durationMs = performance.now() - startedAt;
span.setAttribute("db.redis.duration_ms", durationMs);
if (durationMs > thresholds.slowMs) {
span.setAttribute("db.redis.slow", true);
span.setAttribute(
"db.redis.breach_ratio",
thresholds.slowMs > 0 ? durationMs / thresholds.slowMs : 0,
);
}
if (durationMs > thresholds.severeMs) {
emitRedisSlowLog({
operation,
durationMs,
thresholds,
keyContext,
region,
key,
});
}
} catch {
// swallow — never mask application results
}
try {
if (error) {
span.setStatus({ code: SpanStatusCode.ERROR });
@@ -54,18 +110,65 @@ const finalizeSpan = ({ span, error }: { span: Span; error?: unknown }) => {
/**
* Extracts the first key argument from a Redis command's args.
* For most commands, args[0] is the key. Returns undefined for
* For most commands, args[0] is the key. Custom Lua commands registered
* without `numberOfKeys` (e.g. `setCachedFullSubject`) pass the key count
* as args[0] and the first real key at args[1]. Returns undefined for
* keyless commands or evalsha (where args[0] is a SHA hash).
*/
const extractKey = ({ args }: { args: unknown[] }): string | undefined => {
export const extractKey = ({
args,
}: {
args: unknown[];
}): string | undefined => {
if (args.length === 0) return undefined;
const firstArg = args[0];
if (typeof firstArg === "string") return firstArg;
if (Buffer.isBuffer(firstArg)) return firstArg.toString("utf8");
const candidate = typeof args[0] === "number" ? args[1] : args[0];
if (typeof candidate === "string") return candidate;
if (Buffer.isBuffer(candidate)) return candidate.toString("utf8");
return undefined;
};
/** Truncates a key for `db.statement` (Axiom trace attribute size limit). */
const truncateKey = (key: string): string =>
key.length > 200 ? `${key.slice(0, 200)}...` : key;
/**
* Applies SLO threshold + key-context attributes to a fresh span.
* Isolated in a try/catch so partial-enrichment failures don't prevent
* the command from running.
*/
const enrichSpan = ({
span,
operation,
region,
key,
}: {
span: Span;
operation: string;
region?: string;
key?: string;
}): {
thresholds: ResolvedThresholds;
keyContext: RedisKeyContext;
} => {
const thresholds = resolveThresholds({ operation, redisRegion: region });
span.setAttribute("db.redis.slow_ms", thresholds.slowMs);
span.setAttribute("db.redis.base_slow_ms", thresholds.baseSlowMs);
span.setAttribute("db.redis.region_baseline_ms", thresholds.regionBaselineMs);
span.setAttribute("db.redis.severe_ms", thresholds.severeMs);
const keyContext = parseRedisKeyContext({ key });
if (keyContext.orgId) span.setAttribute("db.redis.org_id", keyContext.orgId);
if (keyContext.customerId)
span.setAttribute("db.redis.customer_id", keyContext.customerId);
if (keyContext.entityId)
span.setAttribute("db.redis.entity_id", keyContext.entityId);
if (keyContext.generation)
span.setAttribute("db.redis.cache_gen", keyContext.generation);
return { thresholds, keyContext };
};
/** Wraps a custom command method created by defineCommand with a traced version. */
const wrapCustomCommand = ({
redis,
@@ -86,19 +189,39 @@ const wrapCustomCommand = ({
// biome-ignore lint: dynamic property assignment for custom redis commands
(redis as any)[name] = function (this: Redis, ...args: unknown[]) {
let span: Span;
let spanCtx: SpanContext;
try {
span = tracer.startSpan(`redis.${name}`, {
const span = tracer.startSpan(`redis.${name}`, {
kind: SpanKind.CLIENT,
});
span.setAttribute("db.system", "redis");
span.setAttribute("db.operation", name);
if (region) span.setAttribute("db.redis.region", region);
const key = extractKey({ args });
if (key) span.setAttribute("db.statement", truncateKey(key));
const { thresholds, keyContext } = enrichSpan({
span,
operation: name,
region,
key,
});
spanCtx = {
span,
startedAt: performance.now(),
thresholds,
keyContext,
operation: name,
region,
key,
};
} catch {
return original.apply(this, args);
}
const activeContext = trace.setSpan(context.active(), span);
const activeContext = trace.setSpan(context.active(), spanCtx.span);
try {
const result = context.with(activeContext, () =>
@@ -108,20 +231,20 @@ const wrapCustomCommand = ({
if (result && typeof (result as Promise<unknown>).then === "function") {
return (result as Promise<unknown>).then(
(val) => {
finalizeSpan({ span });
finalizeSpan({ spanCtx });
return val;
},
(err) => {
finalizeSpan({ span, error: err });
finalizeSpan({ spanCtx, error: err });
throw err;
},
);
}
finalizeSpan({ span });
finalizeSpan({ spanCtx });
return result;
} catch (err) {
finalizeSpan({ span, error: err });
finalizeSpan({ spanCtx, error: err });
throw err;
}
};
@@ -167,32 +290,39 @@ export const instrumentRedis = ({
return originalSendCommand.call(this, command, stream);
}
let span: Span;
let spanCtx: SpanContext;
try {
span = tracer.startSpan(`redis.${commandName}`, {
const span = tracer.startSpan(`redis.${commandName}`, {
kind: SpanKind.CLIENT,
});
span.setAttribute("db.system", "redis");
span.setAttribute("db.operation", commandName.toUpperCase());
if (region) span.setAttribute("db.redis.region", region);
if (region) {
span.setAttribute("db.redis.region", region);
}
const key = extractKey({ args: command?.args ?? [] });
if (key) span.setAttribute("db.statement", truncateKey(key));
const key = extractKey({
args: command?.args ?? [],
const { thresholds, keyContext } = enrichSpan({
span,
operation: commandName,
region,
key,
});
if (key) {
span.setAttribute(
"db.statement",
key.length > 200 ? `${key.slice(0, 200)}...` : key,
);
}
spanCtx = {
span,
startedAt: performance.now(),
thresholds,
keyContext,
operation: commandName,
region,
key,
};
} catch {
return originalSendCommand.call(this, command, stream);
}
const activeContext = trace.setSpan(context.active(), span);
const activeContext = trace.setSpan(context.active(), spanCtx.span);
try {
const result = context.with(activeContext, () =>
@@ -202,20 +332,20 @@ export const instrumentRedis = ({
if (result && typeof (result as Promise<unknown>).then === "function") {
return (result as Promise<unknown>).then(
(val) => {
finalizeSpan({ span });
finalizeSpan({ spanCtx });
return val;
},
(err) => {
finalizeSpan({ span, error: err });
finalizeSpan({ spanCtx, error: err });
throw err;
},
);
}
finalizeSpan({ span });
finalizeSpan({ spanCtx });
return result;
} catch (err) {
finalizeSpan({ span, error: err });
finalizeSpan({ spanCtx, error: err });
throw err;
}
};

109
server/src/external/redis/otel/parseRedisKeyContext.ts vendored Normal file
View File

@@ -0,0 +1,109 @@
/**
* Parses business context (org_id, customer_id, entity_id) out of the
* Redis key patterns used by the app.
*
* Supported patterns:
* - `{orgId}:env:customer:version:customerId`
* - `{orgId}:env:customer:version:customerId:entity:entityId`
* - `{orgId}:env:customer:version:customerId:balances:featureId[...]`
* - `{orgId}:env:customer_guard:customerId`
* - `{orgId}:env:test_cache_delete_guard:customerId`
* - `{orgId}:env:fullcustomer:version:customerId`
* - `{orgId}:env:fullcustomer:guard:customerId`
* - `{orgId}:env:fullcustomer:pathidx:customerId`
* - `{customerId}:orgId:env:full_subject[:...]` (FullSubject V2 — hash-tagged
* on customerId, orgId is the first unbraced segment; covers base,
* shared-balances, and view-epoch shapes)
* - `{customerId}:orgId:env:entity:entityId:full_subject` (FullSubject V2
* entity variant)
*
* Any parsing failure is swallowed — callers get `{}` back.
*/
export type RedisCacheGeneration = "v1" | "v2";
export type RedisKeyContext = {
orgId?: string;
customerId?: string;
entityId?: string;
/**
* Which cache shape this key belongs to. `v2` = FullSubject cache,
* `v1` = legacy per-customer / fullcustomer cache. Undefined when the
* key doesn't match any known shape.
*/
generation?: RedisCacheGeneration;
};
const stripHashTag = (segment: string): string => {
if (segment.startsWith("{") && segment.endsWith("}")) {
return segment.slice(1, -1);
}
return segment;
};
export const parseRedisKeyContext = ({
key,
}: {
key?: string;
}): RedisKeyContext => {
if (!key) return {};
try {
const parts = key.split(":");
if (parts.length < 3) return {};
const first = parts[0];
// FullSubject V2 keys — hash-tagged on customerId, orgId is the
// first unbraced segment. Shapes:
// {customerId}:orgId:env:full_subject[:...]
// {customerId}:orgId:env:entity:entityId:full_subject
if (
first.startsWith("{") &&
first.endsWith("}") &&
parts.includes("full_subject")
) {
const customerId = stripHashTag(first);
const orgId = parts[1];
const entityId =
parts[3] === "entity" && parts[4] && parts[5] === "full_subject"
? parts[4]
: undefined;
return { customerId, orgId, entityId, generation: "v2" };
}
// {orgId}:env:<kind>:...
if (first.startsWith("{") && first.endsWith("}")) {
const orgId = stripHashTag(first);
const kind = parts[2];
if (kind === "fullcustomer") {
// {orgId}:env:fullcustomer:(version|guard|pathidx):customerId
const customerId = parts[4];
return { orgId, customerId, generation: "v1" };
}
if (kind === "customer") {
// {orgId}:env:customer:version:customerId[:entity:entityId][...]
const customerId = parts[4];
let entityId: string | undefined;
const entityIdx = parts.indexOf("entity", 5);
if (entityIdx !== -1 && parts[entityIdx + 1]) {
entityId = parts[entityIdx + 1];
}
return { orgId, customerId, entityId, generation: "v1" };
}
if (kind === "customer_guard" || kind === "test_cache_delete_guard") {
return { orgId, customerId: parts[3], generation: "v1" };
}
// Unknown kind, but orgId is still useful.
return { orgId, generation: "v1" };
}
return {};
} catch {
return {};
}
};

167
server/src/external/redis/otel/redisSlowlogConfig.ts vendored Normal file
View File

@@ -0,0 +1,167 @@
/**
* Redis slowlog thresholds (per-operation SLOs).
*
* Scope: the V2 FullSubject cache surface only. Commands outside this set
* fall through to `DEFAULT_THRESHOLD`, which is intentionally lax so it
* doesn't flap on untuned ops.
*
* Two independent lookups compose into an effective threshold:
* 1. Per-operation base threshold ("how long should this command take
* once it gets there?" — service time + same-region network)
* 2. Per-region additive baseline ("how much extra network cost do I
* expect when hitting Redis in this region?")
*
* effective slowMs = base.slowMs + regionBaselineMs
*
* See `redis-slow-command-investigation.md` for the Axiom query templates
* used to tune these numbers.
*/
export type RedisThresholdConfig = {
operation: string;
slowMs: number;
severeMs: number;
};
export type RegionBaselineConfig = {
region: string;
baselineMs: number;
};
const threshold = ({
operation,
slowMs,
severeMs,
}: RedisThresholdConfig): RedisThresholdConfig => ({
operation,
slowMs,
severeMs,
});
const regionBaseline = ({
region,
baselineMs,
}: RegionBaselineConfig): RegionBaselineConfig => ({
region,
baselineMs,
});
/**
* Fallback for any operation not in `REDIS_THRESHOLDS`. Kept lax so that
* random non-V2 Redis calls don't flood slow-command logs.
*/
const DEFAULT_THRESHOLD: RedisThresholdConfig = {
operation: "__default__",
slowMs: 100,
severeMs: 500,
};
const DEFAULT_REGION_BASELINE_MS = 0;
/**
* Operation names match:
* - lowercase built-in command names (e.g. "get", "hmget")
* - exact `defineCommand` name for custom Lua commands (case sensitive)
*
* Built-in thresholds below are tuned for the V2 FullSubject cache path
* but apply globally wherever these commands run — a slow GET anywhere
* is still worth catching.
*/
export const REDIS_THRESHOLDS: RedisThresholdConfig[] = [
// --- Built-in commands used by the V2 FullSubject cache layer ---
// (server/src/internal/customers/cache/fullSubject/)
threshold({ operation: "get", slowMs: 15, severeMs: 100 }),
threshold({ operation: "set", slowMs: 20, severeMs: 100 }),
threshold({ operation: "hmget", slowMs: 25, severeMs: 150 }),
threshold({ operation: "hdel", slowMs: 20, severeMs: 100 }),
threshold({ operation: "unlink", slowMs: 20, severeMs: 100 }),
threshold({ operation: "incr", slowMs: 15, severeMs: 100 }),
threshold({ operation: "expire", slowMs: 15, severeMs: 100 }),
// --- V2 FullSubject Lua commands ---
// (server/src/_luaScriptsV2/fullSubject/, fullSubjectDeduction/)
threshold({ operation: "setCachedFullSubject", slowMs: 50, severeMs: 300 }),
threshold({ operation: "adjustSubjectBalance", slowMs: 50, severeMs: 300 }),
threshold({
operation: "updateFullSubjectCustomerDataV2",
slowMs: 50,
severeMs: 300,
}),
threshold({
operation: "updateFullSubjectEntityDataV2",
slowMs: 50,
severeMs: 300,
}),
threshold({
operation: "updateFullSubjectCustomerProductV2",
slowMs: 50,
severeMs: 300,
}),
threshold({
operation: "upsertInvoiceInFullSubjectV2",
slowMs: 50,
severeMs: 300,
}),
threshold({ operation: "updateSubjectBalances", slowMs: 75, severeMs: 400 }),
threshold({
operation: "deductFromSubjectBalances",
slowMs: 75,
severeMs: 400,
}),
threshold({ operation: "claimLockReceipt", slowMs: 30, severeMs: 200 }),
];
/**
* Seed values are guesses; adjust after observing real p50 latencies in
* Axiom (see Query E in `redis-slow-command-investigation.md`).
*/
export const REGION_BASELINES: RegionBaselineConfig[] = [
regionBaseline({ region: "us-east-2", baselineMs: 0 }),
regionBaseline({ region: "us-west-2", baselineMs: 70 }),
];
export const getRedisThresholdConfig = ({
operation,
}: {
operation: string;
}): RedisThresholdConfig =>
REDIS_THRESHOLDS.find((config) => config.operation === operation) ??
DEFAULT_THRESHOLD;
export const getRegionBaselineMs = ({
region,
}: {
region?: string;
}): number => {
if (!region) return DEFAULT_REGION_BASELINE_MS;
return (
REGION_BASELINES.find((config) => config.region === region)?.baselineMs ??
DEFAULT_REGION_BASELINE_MS
);
};
export type ResolvedThresholds = {
slowMs: number;
severeMs: number;
baseSlowMs: number;
baseSevereMs: number;
regionBaselineMs: number;
};
export const resolveThresholds = ({
operation,
redisRegion,
}: {
operation: string;
redisRegion?: string;
}): ResolvedThresholds => {
const base = getRedisThresholdConfig({ operation });
const regionBaselineMs = getRegionBaselineMs({ region: redisRegion });
return {
baseSlowMs: base.slowMs,
baseSevereMs: base.severeMs,
slowMs: base.slowMs + regionBaselineMs,
severeMs: base.severeMs + regionBaselineMs,
regionBaselineMs,
};
};

31
server/src/external/redis/otel/slowLogRateLimit.ts vendored Normal file
View File

@@ -0,0 +1,31 @@
/**
* Token-bucket rate limiter for `redis_slow_command` logs.
*
* Prevents flooding Axiom when a slow operation is happening in a hot loop.
* Per-operation bucket: up to MAX_PER_MINUTE events per operation per minute.
*
* Process-local (in-memory); does not coordinate across instances, which is
* fine — the aggregate otel spans still capture everything for percentiles.
*/
const MAX_PER_MINUTE = 10;
const WINDOW_MS = 60_000;
type Bucket = { count: number; resetAt: number };
const buckets = new Map<string, Bucket>();
export const shouldEmitSlowLog = ({
operation,
}: {
operation: string;
}): boolean => {
const now = Date.now();
const bucket = buckets.get(operation);
if (!bucket || bucket.resetAt < now) {
buckets.set(operation, { count: 1, resetAt: now + WINDOW_MS });
return true;
}
if (bucket.count >= MAX_PER_MINUTE) return false;
bucket.count++;
return true;
};

2
server/src/internal/balances/utils/deductionV2/executePostgresDeductionV2.ts
View File

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

6
server/src/internal/balances/utils/deductionV2/executeRedisDeductionV2.ts
View File

@@ -4,7 +4,8 @@ import {
fullSubjectToFullCustomer,
} from "@autumn/shared";
import type { Redis } from "ioredis";
import { currentRegion, redis } from "@/external/redis/initRedis.js";
import { currentRegion } from "@/external/redis/initRedis.js";
import { redisV2 } from "@/external/redis/initRedisV2.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { triggerAutoTopUp } from "@/internal/balances/autoTopUp/triggerAutoTopUp.js";
import { fireTrackWebhooks } from "@/internal/balances/trackWebhooks/fireTrackWebhooks.js";
@@ -150,7 +151,8 @@ export const executeRedisDeductionV2 = async ({
lock_receipt_key: lockReceiptKey ?? null,
};
const targetRedis = redisInstance ?? redis;
const targetRedis = redisInstance ?? redisV2;
const result = await tryRedisWrite(
() =>
targetRedis.deductFromSubjectBalances(

13
server/src/internal/balances/utils/lock/saveLockReceipt.ts
View File

@@ -1,4 +1,5 @@
import { ErrCode, InternalError, RecaseError } from "@autumn/shared";
import type { Redis } from "ioredis";
import { currentRegion, redis } from "@/external/redis/initRedis.js";
import type { MutationLogItem } from "@/internal/balances/utils/types/mutationLogItem.js";
import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
@@ -9,6 +10,7 @@ export const saveLockReceipt = async ({
featureId,
entityId,
items,
redisInstance,
}: {
lock: {
lock_id?: string;
@@ -22,9 +24,11 @@ export const saveLockReceipt = async ({
featureId: string;
entityId?: string;
items: MutationLogItem[];
redisInstance?: Redis;
}) => {
// Check if a lock receipt already exists before writing
const existing = await redis.call("EXISTS", lock.redis_receipt_key);
const targetRedis = redisInstance ?? redis;
const existing = await targetRedis.call("EXISTS", lock.redis_receipt_key);
if (existing === 1) {
throw new RecaseError({
message: "A lock with this ID already exists",
@@ -35,7 +39,7 @@ export const saveLockReceipt = async ({
const result = await tryRedisWrite(
() =>
redis.call(
targetRedis.call(
"JSON.SET",
lock.redis_receipt_key,
"$",
@@ -52,10 +56,11 @@ export const saveLockReceipt = async ({
items,
}),
) as Promise<"OK" | null>,
targetRedis,
);
if (result === "OK") {
await redis.expireat(lock.redis_receipt_key, lock.ttl_at);
await targetRedis.expireat(lock.redis_receipt_key, lock.ttl_at);
return;
}

4
server/src/internal/balances/utils/sync/syncItemV4.ts
View File

@@ -7,7 +7,7 @@ import {
} from "@autumn/shared";
import { sql } from "drizzle-orm";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { invalidateCachedFullSubject } from "@/internal/customers/cache/fullSubject/actions/invalidate/invalidateFullSubject.js";
import { deleteCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.js";
import { getCachedFeatureBalance } from "@/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.js";
import { refreshEntityAggregateCache } from "./refreshEntityAggregateCache.js";
@@ -49,7 +49,7 @@ const handleSyncPostgresError = async ({
`[SYNC V4] (${customerId}) Sync conflict detected: ${code}, cus_ent: ${cusEntId}. Invalidating cache.`,
);
await invalidateCachedFullSubject({
await deleteCachedFullCustomer({
ctx,
customerId,
entityId,

18
server/src/internal/customers/actions/update/updateCustomer.ts
View File

@@ -18,6 +18,7 @@ import type { AutumnContext } from "@/honoUtils/HonoEnv";
import { triggerAutoTopUpsOnEnabled } from "@/internal/balances/autoTopUp/triggerAutoTopUpsOnEnabled";
import { CusService } from "@/internal/customers/CusService";
import { invalidateCachedFullSubject } from "../../cache/fullSubject/actions/invalidate/invalidateFullSubject";
import { updateCachedCustomerData as updateCachedCustomerDataV2 } from "../../cache/fullSubject/actions/updateCachedCustomerData";
import { updateCachedCustomerData } from "../../cusUtils/fullCustomerCacheUtils/updateCachedCustomerData";
import { getApiCustomerByRollout } from "../getApiCustomerByRollout";
@@ -181,11 +182,18 @@ export const updateCustomer = async ({
});
}
await updateCachedCustomerData({
ctx,
customerId: originalCustomerId,
updates: updateData,
});
await Promise.all([
updateCachedCustomerData({
ctx,
customerId: originalCustomerId,
updates: updateData,
}),
updateCachedCustomerDataV2({
ctx,
customerId: originalCustomerId,
updates: updateData,
}),
]);
ctx.skipCache = true;
const resolvedCustomerId = newCustomerId ?? customerId;

18
server/src/internal/customers/actions/updateCustomerData.ts
View File

@@ -3,6 +3,7 @@ import { z } from "zod/v4";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { CusService } from "@/internal/customers/CusService.js";
import { updateCachedCustomerData } from "@/internal/customers/cache/fullSubject/index.js";
import { updateCachedCustomerData as updateCachedCustomerDataV1 } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/updateCachedCustomerData.js";
export const updateCustomerData = async ({
ctx,
@@ -50,11 +51,18 @@ export const updateCustomerData = async ({
Object.assign(fullSubject.customer, updates);
await updateCachedCustomerData({
ctx,
customerId: idOrInternalId,
updates,
});
await Promise.all([
updateCachedCustomerData({
ctx,
customerId: idOrInternalId,
updates,
}),
updateCachedCustomerDataV1({
ctx,
customerId: idOrInternalId,
updates,
}),
]);
return true;
};

7
server/src/internal/customers/cache/fullSubject/actions/getOrCreateCachedFullSubject.ts vendored
View File

@@ -79,12 +79,17 @@ export const getOrCreateCachedFullSubject = async ({
});
}
await updateCustomerData({
const customerDataUpdated = await updateCustomerData({
ctx,
fullSubject,
customerData,
});
if (customerDataUpdated && normalizedResult) {
normalizedResult.normalized.customer = fullSubject.customer;
normalizedResult.fullSubject.customer = fullSubject.customer;
}
if (entityId && !fullSubject.entity) {
const newEntity = await autoCreateEntity({
ctx,

12
server/src/internal/customers/cache/fullSubject/actions/updateCachedEntityData.ts vendored
View File

@@ -1,6 +1,7 @@
import type { Entity } from "@autumn/shared";
import { redisV2 } from "@/external/redis/initRedisV2.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { updateEntityInCache } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/updateEntityInCache.js";
import { tryRedisRead, tryRedisWrite } from "@/utils/cacheUtils/cacheUtils.js";
import { logAlertEvent } from "@/utils/logging/logAlertEvent.js";
import { buildFullSubjectKey } from "../builders/buildFullSubjectKey.js";
@@ -25,6 +26,17 @@ export const updateCachedEntityData = async ({
}): Promise<void> => {
if (Object.keys(updates).length === 0) return;
updateEntityInCache({
ctx,
customerId,
idOrInternalId: entityId,
updates,
}).catch((error) => {
ctx.logger.error(
`[updateCachedEntityData] V1 cache update failed for ${customerId}:${entityId}: ${error}`,
);
});
const { org, env, logger } = ctx;
const subjectKey = buildFullSubjectKey({
orgId: org.id,

23
server/src/internal/customers/cache/fullSubject/balances/getCachedFeatureBalances.ts vendored
View File

@@ -3,7 +3,11 @@ import { redisV2 } from "@/external/redis/initRedisV2.js";
import { tryRedisRead } from "@/utils/cacheUtils/cacheUtils.js";
import { buildSharedFullSubjectBalanceKey } from "../builders/buildSharedFullSubjectBalanceKey.js";
import { AGGREGATED_BALANCE_FIELD } from "../config/fullSubjectCacheConfig.js";
import { sanitizeCachedSubjectBalance } from "../sanitize/index.js";
import { roundSubjectBalance } from "../roundCacheBalance.js";
import {
sanitizeCachedAggregatedFeatureBalance,
sanitizeCachedSubjectBalance,
} from "../sanitize/index.js";
export type FeatureBalanceResult = {
featureId: string;
@@ -77,8 +81,10 @@ export const getCachedFeatureBalance = async ({
try {
const parsedBalance = JSON.parse(entryJson) as SubjectBalance;
balances.push(
sanitizeCachedSubjectBalance({
subjectBalance: parsedBalance,
roundSubjectBalance({
subjectBalance: sanitizeCachedSubjectBalance({
subjectBalance: parsedBalance,
}),
}),
);
} catch {
@@ -142,7 +148,10 @@ export const getCachedFeatureBalancesBatch = async ({
const aggregatedJson = allValues.pop() ?? null;
if (aggregatedJson) {
try {
aggregated = JSON.parse(aggregatedJson) as AggregatedFeatureBalance;
const parsed = JSON.parse(aggregatedJson) as AggregatedFeatureBalance;
aggregated = sanitizeCachedAggregatedFeatureBalance({
aggregated: parsed,
});
} catch {
// Malformed _aggregated is non-fatal; fall back to subject string value
}
@@ -160,7 +169,11 @@ export const getCachedFeatureBalancesBatch = async ({
try {
const parsedBalance = JSON.parse(entryJson) as SubjectBalance;
balances.push(
sanitizeCachedSubjectBalance({ subjectBalance: parsedBalance }),
roundSubjectBalance({
subjectBalance: sanitizeCachedSubjectBalance({
subjectBalance: parsedBalance,
}),
}),
);
} catch {
return undefined;

55
server/src/internal/customers/cache/fullSubject/fullSubjectCacheModel.ts vendored
View File

@@ -1,4 +1,18 @@
import type { NormalizedFullSubject } from "@autumn/shared";
import {
CusProductSchema,
CustomerSchema,
EntitlementWithFeatureSchema,
EntityAggregationsSchema,
EntitySchema,
FreeTrialSchema,
InvoiceSchema,
type NormalizedFullSubject,
PriceSchema,
ProductSchema,
SubjectFlagSchema,
SubscriptionSchema,
} from "@autumn/shared";
import { z } from "zod/v4";
export type CachedFullSubject = Omit<
NormalizedFullSubject,
@@ -10,6 +24,45 @@ export type CachedFullSubject = Omit<
subjectViewEpoch: number;
};
/**
* Schema mirror of `CachedFullSubject` used by the cache-hole-filling walker
* ({@link normalizeFromSchema}) to locate nullable positions in cached
* payloads. Not a validator — runtime type above is the source of truth.
*
* Sub-shapes intentionally reuse existing shared schemas (CusProductSchema
* mirrors DbCustomerProduct, etc.). Mismatches at non-nullable positions
* are harmless: the walker only fills undefined at nullable positions and
* passes through unknown keys.
*/
export const CachedFullSubjectSchema = z.object({
subjectType: z.enum(["customer", "entity"]),
customerId: z.string(),
internalCustomerId: z.string(),
entityId: z.string().optional(),
internalEntityId: z.string().optional(),
customer: CustomerSchema,
entity: EntitySchema.optional(),
customer_products: z.array(CusProductSchema),
flags: z.record(z.string(), SubjectFlagSchema),
products: z.array(ProductSchema),
entitlements: z.array(EntitlementWithFeatureSchema),
prices: z.array(PriceSchema),
free_trials: z.array(FreeTrialSchema),
subscriptions: z.array(SubscriptionSchema),
invoices: z.array(InvoiceSchema),
entity_aggregations: EntityAggregationsSchema.optional(),
_cachedAt: z.number(),
meteredFeatures: z.array(z.string()),
customerEntitlementIdsByFeatureId: z.record(z.string(), z.array(z.string())),
subjectViewEpoch: z.number(),
});
export const normalizedToCachedFullSubject = ({
normalized,
subjectViewEpoch,

53
server/src/internal/customers/cache/fullSubject/roundCacheBalance.ts vendored Normal file
View File

@@ -0,0 +1,53 @@
import type { SubjectBalance } from "@autumn/shared";
import { Decimal } from "decimal.js";
/**
* Round a number to avoid floating-point precision issues from Lua 5.1 double arithmetic.
* Uses Decimal.js toDecimalPlaces(10) — enough precision while eliminating float drift.
*/
export const roundCacheBalance = (
value: number | null | undefined,
): number => {
if (value === null || value === undefined) return 0;
return new Decimal(value).toDecimalPlaces(10).toNumber();
};
/**
* Round all balance-related numeric fields on a SubjectBalance in-place.
* Handles top-level fields, entity-scoped balances, and rollover balances.
*/
export const roundSubjectBalance = ({
subjectBalance,
}: {
subjectBalance: SubjectBalance;
}): SubjectBalance => {
subjectBalance.balance = roundCacheBalance(subjectBalance.balance);
if (subjectBalance.adjustment !== null && subjectBalance.adjustment !== undefined)
subjectBalance.adjustment = roundCacheBalance(subjectBalance.adjustment);
if (subjectBalance.additional_balance !== null && subjectBalance.additional_balance !== undefined)
subjectBalance.additional_balance = roundCacheBalance(subjectBalance.additional_balance);
if (subjectBalance.entities && typeof subjectBalance.entities === "object") {
for (const entityId of Object.keys(subjectBalance.entities)) {
const entityData = subjectBalance.entities[entityId];
if (!entityData || typeof entityData !== "object") continue;
if (entityData.balance !== null && entityData.balance !== undefined)
entityData.balance = roundCacheBalance(entityData.balance);
if (entityData.adjustment !== null && entityData.adjustment !== undefined)
entityData.adjustment = roundCacheBalance(entityData.adjustment);
}
}
if (subjectBalance.rollovers && Array.isArray(subjectBalance.rollovers)) {
for (const rollover of subjectBalance.rollovers) {
if (rollover.balance !== null && rollover.balance !== undefined)
rollover.balance = roundCacheBalance(rollover.balance);
}
}
return subjectBalance;
};

1
server/src/internal/customers/cache/fullSubject/sanitize/index.ts vendored
View File

@@ -1,2 +1,3 @@
export { sanitizeCachedAggregatedFeatureBalance } from "./sanitizeCachedAggregatedFeatureBalance.js";
export { sanitizeCachedFullSubject } from "./sanitizeCachedFullSubject.js";
export { sanitizeCachedSubjectBalance } from "./sanitizeCachedSubjectBalance.js";

129
server/src/internal/customers/cache/fullSubject/sanitize/normalizeFromSchema.ts vendored Normal file
View File

@@ -0,0 +1,129 @@
import { z } from "zod/v4";
/**
* Normalize cached JSON against a Zod schema to repair Upstash cjson quirks.
*
* Upstash's Lua cjson (Go-backed) differs from standard Redis Lua:
* - JSON `null` decodes to Lua `nil` (dropped on re-encode).
* - Empty JS objects/arrays both become `{}` after a Lua round-trip.
*
* Walks the schema and fills holes at nullable positions, and swaps
* empty-object/empty-array when the schema says otherwise. Never throws,
* never re-validates — this is a cleanup pass, not a `.parse()`.
*/
type ZodSchema = z.core.$ZodType;
const isPlainObject = (value: unknown): value is Record<string, unknown> =>
!!value && typeof value === "object" && !Array.isArray(value);
const isEmptyObject = (value: unknown): boolean =>
isPlainObject(value) && Object.keys(value).length === 0;
const isEmptyArray = (value: unknown): boolean =>
Array.isArray(value) && value.length === 0;
/**
* True if the schema's wrapper chain contains `ZodNullable`. Walks through
* `ZodOptional` / `ZodDefault` / `ZodPipe` so `.nullish()` (optional→nullable)
* and `.nullable().default(x)` both report as nullable.
*/
const isNullable = (schema: ZodSchema): boolean => {
if (schema instanceof z.ZodNullable) return true;
if (schema instanceof z.ZodOptional) return isNullable(schema._def.innerType);
if (schema instanceof z.ZodDefault) return isNullable(schema._def.innerType);
if (schema instanceof z.ZodPipe)
return isNullable(schema._def.in) || isNullable(schema._def.out);
return false;
};
/**
* Strip wrapper layers to reveal the payload-shaping schema
* (`ZodObject` / `ZodArray` / `ZodRecord` / leaf).
*/
const unwrapSchema = (schema: ZodSchema): ZodSchema => {
if (schema instanceof z.ZodNullable)
return unwrapSchema(schema._def.innerType);
if (schema instanceof z.ZodOptional)
return unwrapSchema(schema._def.innerType);
if (schema instanceof z.ZodDefault)
return unwrapSchema(schema._def.innerType);
if (schema instanceof z.ZodPipe) return unwrapSchema(schema._def.in);
return schema;
};
/**
* If `undefined` hits a `ZodDefault` anywhere in the wrapper chain, return
* its default value. v4 exposes `defaultValue` as a direct value, not a thunk.
*/
const applyDefaultIfUndefined = (
schema: ZodSchema,
value: unknown,
): { applied: true; value: unknown } | { applied: false } => {
if (value !== undefined) return { applied: false };
if (schema instanceof z.ZodDefault) {
return { applied: true, value: schema._def.defaultValue };
}
if (schema instanceof z.ZodOptional)
return applyDefaultIfUndefined(schema._def.innerType, value);
if (schema instanceof z.ZodNullable)
return applyDefaultIfUndefined(schema._def.innerType, value);
if (schema instanceof z.ZodPipe)
return applyDefaultIfUndefined(schema._def.in, value);
return { applied: false };
};
const normalize = (schema: ZodSchema, data: unknown): unknown => {
if (data === undefined) {
const defaultResult = applyDefaultIfUndefined(schema, data);
if (defaultResult.applied)
return normalize(unwrapSchema(schema), defaultResult.value);
if (isNullable(schema)) return null;
return data;
}
const unwrapped = unwrapSchema(schema);
if (unwrapped instanceof z.ZodObject) {
if (!isPlainObject(data)) return data;
const shape = unwrapped._def.shape;
const normalized: Record<string, unknown> = { ...data };
for (const key of Object.keys(shape)) {
normalized[key] = normalize(shape[key], normalized[key]);
}
return normalized;
}
if (unwrapped instanceof z.ZodArray) {
if (isEmptyObject(data)) return [];
if (!Array.isArray(data)) return data;
const element = unwrapped._def.element;
return data.map((item) => normalize(element, item));
}
if (unwrapped instanceof z.ZodRecord) {
if (isEmptyArray(data)) return {};
if (!isPlainObject(data)) return data;
const valueType = unwrapped._def.valueType;
const normalized: Record<string, unknown> = {};
for (const key of Object.keys(data)) {
normalized[key] = normalize(valueType, data[key]);
}
return normalized;
}
return data;
};
export const normalizeFromSchema = <T>({
schema,
data,
}: {
schema: ZodSchema;
data: unknown;
}): T => normalize(schema, data) as T;

71
server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCacheShapeUtils.ts vendored
View File

@@ -1,71 +0,0 @@
/**
* Shape spec describes which fields should be arrays, records, or nested objects
* so the recursive sanitizer can coerce malformed Redis/Lua JSON payloads.
*
* "array" -> coerce non-arrays to []
* "record" -> coerce non-objects to {}
* "nullable_record" -> coerce non-objects to null (for optional record fields)
* ShapeSpec -> recurse into object fields
* { items: ShapeSpec } -> coerce field to array, then recurse each element
*/
export interface ShapeSpec {
[key: string]: FieldRule;
}
export type FieldRule =
| "array"
| "record"
| "nullable_record"
| ShapeSpec
| { items: ShapeSpec };
const isPlainObject = (value: unknown): value is Record<string, unknown> =>
!!value && typeof value === "object" && !Array.isArray(value);
const coerceArray = (value: unknown): unknown[] =>
Array.isArray(value) ? value : [];
const coerceRecord = (value: unknown): Record<string, unknown> =>
isPlainObject(value) ? value : {};
const coerceNullableRecord = (
value: unknown,
): Record<string, unknown> | null => (isPlainObject(value) ? value : null);
/**
* Recursively sanitizes an object against a shape spec.
* Only touches fields that appear in the spec; all other fields pass through.
*/
export const sanitizeShape = <T>({
value,
spec,
}: {
value: unknown;
spec: ShapeSpec;
}): T => {
if (!isPlainObject(value)) return {} as T;
const result = { ...value } as Record<string, unknown>;
for (const [key, rule] of Object.entries(spec)) {
const fieldValue = result[key];
if (rule === "array") {
result[key] = coerceArray(fieldValue);
} else if (rule === "record") {
result[key] = coerceRecord(fieldValue);
} else if (rule === "nullable_record") {
result[key] = coerceNullableRecord(fieldValue);
} else if (isPlainObject(rule) && "items" in rule) {
const itemsSpec = (rule as { items: ShapeSpec }).items;
const arr = coerceArray(fieldValue);
result[key] = arr.map((item) =>
sanitizeShape({ value: item, spec: itemsSpec }),
);
} else if (isPlainObject(rule) && isPlainObject(fieldValue)) {
result[key] = sanitizeShape({ value: fieldValue, spec: rule as ShapeSpec });
}
}
return result as T;
};

24
server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedAggregatedFeatureBalance.ts vendored Normal file
View File

@@ -0,0 +1,24 @@
import {
type AggregatedFeatureBalance,
AggregatedFeatureBalanceSchema,
} from "@autumn/shared";
import { normalizeFromSchema } from "./normalizeFromSchema.js";
/**
* Repair an aggregated feature balance read from Redis (the `_aggregated`
* hash field) so Upstash cjson null-drops and empty-collection swaps are
* reversed before the value reaches downstream consumers.
*
* Walks against `AggregatedFeatureBalanceSchema` because the cache stores
* the non-Full (no embedded feature) shape; see
* `setSharedFullSubjectBalances.ts`.
*/
export const sanitizeCachedAggregatedFeatureBalance = ({
aggregated,
}: {
aggregated: AggregatedFeatureBalance;
}): AggregatedFeatureBalance =>
normalizeFromSchema<AggregatedFeatureBalance>({
schema: AggregatedFeatureBalanceSchema,
data: aggregated,
});

85
server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.ts vendored
View File

@@ -1,79 +1,20 @@
import type { CachedFullSubject } from "../fullSubjectCacheModel.js";
import { type ShapeSpec, sanitizeShape } from "./sanitizeCacheShapeUtils.js";
const featureShapeSpec: ShapeSpec = {
event_names: "array",
};
const entitlementCatalogShapeSpec: ShapeSpec = {
feature: featureShapeSpec,
};
const priceConfigShapeSpec: ShapeSpec = {
usage_tiers: "array",
};
const priceShapeSpec: ShapeSpec = {
config: priceConfigShapeSpec,
};
const customerShapeSpec: ShapeSpec = {
auto_topups: "array",
spend_limits: "array",
usage_alerts: "array",
overage_allowed: "array",
};
const entityShapeSpec: ShapeSpec = {
spend_limits: "array",
usage_alerts: "array",
overage_allowed: "array",
};
const customerProductShapeSpec: ShapeSpec = {
options: "array",
subscription_ids: "array",
scheduled_ids: "array",
};
const subscriptionShapeSpec: ShapeSpec = {
usage_features: "array",
};
const invoiceShapeSpec: ShapeSpec = {
product_ids: "array",
internal_product_ids: "array",
discounts: "array",
items: "array",
};
const entityAggregationsShapeSpec: ShapeSpec = {
aggregated_customer_products: { items: customerProductShapeSpec },
aggregated_customer_entitlements: "array",
};
const cachedFullSubjectShapeSpec: ShapeSpec = {
customer: customerShapeSpec,
entity: entityShapeSpec,
customer_products: { items: customerProductShapeSpec },
products: "array",
entitlements: { items: entitlementCatalogShapeSpec },
prices: { items: priceShapeSpec },
free_trials: "array",
subscriptions: { items: subscriptionShapeSpec },
invoices: { items: invoiceShapeSpec },
flags: "record",
meteredFeatures: "array",
customerEntitlementIdsByFeatureId: "record",
entity_aggregations: entityAggregationsShapeSpec,
};
import {
type CachedFullSubject,
CachedFullSubjectSchema,
} from "../fullSubjectCacheModel.js";
import { normalizeFromSchema } from "./normalizeFromSchema.js";
/**
* Repair a `CachedFullSubject` read from Redis so Upstash cjson null-drops
* and empty-collection swaps are reversed before the value reaches downstream
* consumers.
*/
export const sanitizeCachedFullSubject = ({
cachedFullSubject,
}: {
cachedFullSubject: CachedFullSubject;
}): CachedFullSubject =>
sanitizeShape<CachedFullSubject>({
value: cachedFullSubject,
spec: cachedFullSubjectShapeSpec,
normalizeFromSchema<CachedFullSubject>({
schema: CachedFullSubjectSchema,
data: cachedFullSubject,
});

47
server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedSubjectBalance.ts vendored
View File

@@ -1,44 +1,17 @@
import type { SubjectBalance } from "@autumn/shared";
import { type ShapeSpec, sanitizeShape } from "./sanitizeCacheShapeUtils.js";
const featureShapeSpec: ShapeSpec = {
event_names: "array",
};
const entitlementShapeSpec: ShapeSpec = {
feature: featureShapeSpec,
};
const priceConfigShapeSpec: ShapeSpec = {
usage_tiers: "array",
};
const priceShapeSpec: ShapeSpec = {
config: priceConfigShapeSpec,
};
const customerPriceShapeSpec: ShapeSpec = {
price: priceShapeSpec,
};
const rolloverShapeSpec: ShapeSpec = {
entities: "record",
};
const subjectBalanceShapeSpec: ShapeSpec = {
replaceables: "array",
rollovers: { items: rolloverShapeSpec },
entities: "nullable_record",
entitlement: entitlementShapeSpec,
customerPrice: customerPriceShapeSpec,
};
import { type SubjectBalance, SubjectBalanceSchema } from "@autumn/shared";
import { normalizeFromSchema } from "./normalizeFromSchema.js";
/**
* Repair a `SubjectBalance` read from Redis so Upstash cjson null-drops and
* empty-collection swaps are reversed before the value reaches downstream
* Zod validators (e.g. the webhook / API response schemas).
*/
export const sanitizeCachedSubjectBalance = ({
subjectBalance,
}: {
subjectBalance: SubjectBalance;
}): SubjectBalance =>
sanitizeShape<SubjectBalance>({
value: subjectBalance,
spec: subjectBalanceShapeSpec,
normalizeFromSchema<SubjectBalance>({
schema: SubjectBalanceSchema,
data: subjectBalance,
});

100
server/src/internal/customers/cusUtils/apiCusUtils/getApiCustomerExpandV2.ts Normal file
View File

@@ -0,0 +1,100 @@
import {
ApiBaseEntitySchema,
type ApiCusExpand,
CustomerExpand,
type FullCusProduct,
type FullSubject,
filterExpand,
fullSubjectToFullCustomer,
} from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { EntityService } from "@/internal/api/entities/EntityService.js";
import { getCusPaymentMethodRes } from "../cusResponseUtils/getCusPaymentMethodRes.js";
import { getCusReferrals } from "../cusResponseUtils/getCusReferrals.js";
import { getCusRewards } from "../cusResponseUtils/getCusRewards.js";
import { getCusTrialsUsed } from "../cusResponseUtils/getCusTrialsUsed.js";
/**
* Build expand fields directly from a FullSubject — no extra FullCustomer fetch.
* For entities, queries the DB directly (capped at 1000).
*/
export const getApiCustomerExpandV2 = async ({
ctx,
fullSubject,
}: {
ctx: AutumnContext;
fullSubject: FullSubject;
}): Promise<ApiCusExpand> => {
const { expand } = ctx;
const filteredExpand = filterExpand({
expand,
filter: [
CustomerExpand.BalancesFeature,
CustomerExpand.FlagsFeature,
CustomerExpand.SubscriptionsPlan,
CustomerExpand.Invoices,
],
});
if (filteredExpand.length === 0) return {};
const cusExpand = expand as CustomerExpand[];
const fullCus = fullSubjectToFullCustomer({ fullSubject });
const subIds = fullSubject.customer_products.flatMap(
(cp: FullCusProduct) => cp.subscription_ids || [],
);
const getApiEntities = async () => {
if (!cusExpand.includes(CustomerExpand.Entities)) return undefined;
const entities = await EntityService.list({
db: ctx.db,
internalCustomerId: fullCus.internal_id,
});
return entities.map((e) =>
ApiBaseEntitySchema.parse({
...e,
customer_id: fullSubject.customer.id,
}),
);
};
const [entities, rewards, referrals, paymentMethod, trialsUsed] =
await Promise.all([
getApiEntities(),
getCusRewards({
org: ctx.org,
env: ctx.env,
fullCus,
subIds,
expand: cusExpand,
}),
getCusReferrals({
db: ctx.db,
fullCus,
expand: cusExpand,
}),
getCusPaymentMethodRes({
org: ctx.org,
env: ctx.env,
fullCus,
expand: cusExpand,
}),
getCusTrialsUsed({
ctx,
fullCus,
expand: cusExpand,
}),
]);
return {
trials_used: trialsUsed ?? undefined,
entities: entities ?? undefined,
rewards: rewards ?? undefined,
referrals: referrals ?? undefined,
payment_method: paymentMethod ?? undefined,
};
};

18
server/src/internal/customers/cusUtils/cusUtils.ts
View File

@@ -20,6 +20,7 @@ import {
import RecaseError from "@/utils/errorUtils.js";
import { notNullish, nullish } from "@/utils/genUtils.js";
import type { AutumnContext } from "../../../honoUtils/HonoEnv.js";
import { updateCachedCustomerData as updateCachedCustomerDataV2 } from "../cache/fullSubject/actions/updateCachedCustomerData.js";
import { updateCachedCustomerData } from "./fullCustomerCacheUtils/updateCachedCustomerData.js";
export const updateCustomerDetails = async ({
@@ -68,11 +69,18 @@ export const updateCustomerDetails = async ({
fullCustomer = { ...fullCustomer, ...updates };
await updateCachedCustomerData({
ctx,
customerId: idOrInternalId,
updates,
});
await Promise.all([
updateCachedCustomerData({
ctx,
customerId: idOrInternalId,
updates,
}),
updateCachedCustomerDataV2({
ctx,
customerId: idOrInternalId,
updates,
}),
]);
return true;
}

51
server/src/internal/customers/cusUtils/getApiCustomerV2/getApiBalance/getApiBalanceV2.ts
View File

@@ -38,6 +38,7 @@ import {
getEmptyApiBalanceV2,
mergeAggregatedBalanceIntoApiBalanceV2,
} from "./apiBalanceV2Utils.js";
import { roundApiBalance } from "./roundApiBalance.js";
const getApiBalanceBreakdownItemV2 = ({
fullSubject,
@@ -219,29 +220,29 @@ export const getApiBalanceV2 = ({
entityId,
});
return {
data: mergeAggregatedBalanceIntoApiBalanceV2({
apiBalance: {
object: "balance",
feature_id: feature.id,
feature: apiFeature,
granted: new Decimal(totalGranted).add(totalRolloverGranted).toNumber(),
remaining: new Decimal(totalRemaining)
.add(totalRolloverBalance)
.add(totalUnused)
.toNumber(),
usage: new Decimal(totalUsage)
.add(totalRolloverUsage)
.sub(totalUnused)
.toNumber(),
unlimited,
overage_allowed: usageAllowed ?? false,
max_purchase: totalMaxPurchase,
next_reset_at: nextResetAt,
breakdown: breakdownItems,
rollovers: totalRollovers,
},
aggregatedFeatureBalance,
}),
};
const merged = mergeAggregatedBalanceIntoApiBalanceV2({
apiBalance: {
object: "balance",
feature_id: feature.id,
feature: apiFeature,
granted: new Decimal(totalGranted).add(totalRolloverGranted).toNumber(),
remaining: new Decimal(totalRemaining)
.add(totalRolloverBalance)
.add(totalUnused)
.toNumber(),
usage: new Decimal(totalUsage)
.add(totalRolloverUsage)
.sub(totalUnused)
.toNumber(),
unlimited,
overage_allowed: usageAllowed ?? false,
max_purchase: totalMaxPurchase,
next_reset_at: nextResetAt,
breakdown: breakdownItems,
rollovers: totalRollovers,
},
aggregatedFeatureBalance,
});
return { data: roundApiBalance({ apiBalance: merged }) };
};

27
server/src/internal/customers/cusUtils/getApiCustomerV2/getApiBalance/roundApiBalance.ts Normal file
View File

@@ -0,0 +1,27 @@
import type { ApiBalanceV1 } from "@autumn/shared";
import { roundCacheBalance } from "@/internal/customers/cache/fullSubject/roundCacheBalance.js";
/**
* Round all numeric balance fields on an ApiBalanceV1 to eliminate
* floating-point drift from Lua 5.1 double arithmetic.
*/
export const roundApiBalance = ({
apiBalance,
}: {
apiBalance: ApiBalanceV1;
}): ApiBalanceV1 => {
apiBalance.granted = roundCacheBalance(apiBalance.granted);
apiBalance.remaining = roundCacheBalance(apiBalance.remaining);
apiBalance.usage = roundCacheBalance(apiBalance.usage);
if (apiBalance.breakdown) {
for (const item of apiBalance.breakdown) {
item.included_grant = roundCacheBalance(item.included_grant);
item.prepaid_grant = roundCacheBalance(item.prepaid_grant);
item.remaining = roundCacheBalance(item.remaining);
item.usage = roundCacheBalance(item.usage);
}
}
return apiBalance;
};

6
server/src/internal/customers/cusUtils/getApiCustomerV2/getApiCustomerV2.ts
View File

@@ -6,7 +6,7 @@ import {
type FullSubject,
} from "@autumn/shared";
import type { RequestContext } from "@/honoUtils/HonoEnv.js";
import { getApiCustomerExpand } from "../apiCusUtils/getApiCustomerExpand.js";
import { getApiCustomerExpandV2 } from "../apiCusUtils/getApiCustomerExpandV2.js";
import { getApiCustomerBaseV2 } from "./getApiCustomerBaseV2.js";
/**
@@ -36,9 +36,9 @@ export const getApiCustomerV2 = async ({
: undefined,
};
const apiCustomerExpand = await getApiCustomerExpand({
const apiCustomerExpand = await getApiCustomerExpandV2({
ctx,
customerId: fullSubject.customer.id || fullSubject.customer.internal_id,
fullSubject,
});
const apiCustomer: ApiCustomerV5 = {

19
server/src/internal/entities/handlers/handleCreateEntity/autoCreateEntity.ts
View File

@@ -8,6 +8,7 @@ import {
} from "@autumn/shared";
import { isUniqueConstraintError } from "@/db/dbUtils.js";
import { EntityService } from "@/internal/api/entities/EntityService.js";
import { invalidateCachedFullSubject } from "@/internal/customers/cache/fullSubject/actions/invalidate/invalidateFullSubject.js";
import { upsertEntityInCache } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/appendEntityToCache.js";
import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
@@ -123,13 +124,19 @@ export const autoCreateEntity = async ({
}
}
// Add/update entity in full customer cache
if (entity) {
await upsertEntityInCache({
ctx,
customerId,
entity,
});
await Promise.all([
upsertEntityInCache({
ctx,
customerId,
entity,
}),
invalidateCachedFullSubject({
ctx,
customerId,
source: "autoCreateEntity",
}),
]);
}
return entity;

4
server/src/internal/misc/edgeConfig/edgeConfigStore.ts
View File

@@ -26,7 +26,9 @@ export const createEdgeConfigStore = <T>({
s3Key,
schema,
defaultValue,
pollIntervalMs = ms.seconds(10),
pollIntervalMs = process.env.NODE_ENV === "development"
? ms.seconds(1)
: ms.seconds(10),
s3Client: injectedS3Client,
}: {
s3Key: string;

1
server/src/utils/cacheUtils/cacheUtils.ts
View File

@@ -60,6 +60,7 @@ export const tryRedisWrite = async <T>(
}
const result = await operation();
return (result === undefined ? true : result) as T extends void
? true
: T | null;

11
server/src/utils/logging/addContextToLogs.ts
View File

@@ -1,6 +1,7 @@
import type { Logger } from "@/external/logtail/logtailUtils.js";
import type {
LogAppContext,
LogRedisData,
LogRequestContext,
LogStripeEventContext,
LogWorkflowContext,
@@ -46,6 +47,16 @@ export const addWorkflowToLogs = ({
return logger.child({ context: { workflow: workflowContext } });
};
export const addRedisToLogs = ({
logger,
redisData,
}: {
logger: Logger;
redisData: LogRedisData;
}): Logger => {
return logger.child({ context: { data: redisData } });
};
export const addExtrasToLogs = ({
logger,
extras,

16
server/src/utils/logging/loggerTypes.ts
View File

@@ -47,6 +47,22 @@ export type LogWorkflowContext = {
name: string; // workflow / job name
};
/** Redis slow-command context - goes under context.redis.data (map field) */
export type LogRedisData = {
operation: string;
duration_ms: number;
slow_ms: number;
base_slow_ms: number;
region_baseline_ms: number;
severe_ms: number;
breach_ratio: number;
region?: string;
key?: string;
org_id?: string;
customer_id?: string;
entity_id?: string;
};
export type AlertSeverity = "warning" | "error" | "critical";
export type AlertCategory =

19
server/tests/_groups/temp.ts
View File

@@ -5,19 +5,10 @@ export const temp: TestGroup = {
description: "Failed tests to triage and fix",
tier: "domain",
paths: [
"balances/track/breakdown/track-entity-breakdown1.test.ts",
"balances/track/breakdown/track-breakdown4.test.ts",
"balances/track/breakdown/track-breakdown3.test.ts",
"balances/track/legacy/track-legacy2.test.ts",
"balances/track/legacy/track-legacy3.test.ts",
"balances/check/send-event/send-event1.test.ts",
"balances/track/paid-allocated/track-paid-allocated7.test.ts",
"balances/check/loose/loose-expiry.test.ts",
"balances/check/loose/loose-expiry-cross-version.test.ts",
"integration/balances/track/allocated-invoice/allocated-invoice-advances.test.ts",
"integration/balances/track/basic/track-negative.test.ts",
"integration/balances/reset/get-customer-reset.test.ts",
"integration/balances/lock/check-with-lock-concurrent-stress.test.ts",
"integration/balances/check/spend-limit/check-customer-spend-limit.test.ts",
"integration/balances/track/basic/track-event-name.test.ts",
"integration/balances/track/basic/track-credit-system.test.ts",
"integration/balances/auto-topup/auto-topup-edge-cases.test.ts",
"integration/balances/reset/persist-free-overage-on.test.ts",
"integration/balances/lock/check-with-lock-expiry.test.ts",
],
};

28
server/tests/clearMasterOrg.ts
View File

@@ -8,6 +8,7 @@ import { AppEnv } from "@autumn/shared";
import chalk from "chalk";
import { redis } from "@/external/redis/initRedis.js";
import { redisV2 } from "@/external/redis/initRedisV2.js";
import { clearOrg } from "./utils/setup/clearOrg.js";
import { setupOrg } from "./utils/setup/setupOrg.js";
@@ -33,20 +34,33 @@ export const clearMasterOrg = async () => {
env: AppEnv.Sandbox,
});
// Flush cache if not pointed at regional Redis
const redisUrl = process.env.REDIS_URL ?? process.env.BUN_REDIS_URL ?? "";
const normalizedRedisUrl = redisUrl.toLowerCase();
const isRegionalRedisUrl = normalizedRedisUrl.includes(
"redis-17710.mc1716-0.us",
);
const isRegionalRedisUrl = (url: string | undefined) =>
(url ?? "").toLowerCase().includes("redis-17710.mc1716-0.us");
if (!isRegionalRedisUrl) await redis.flushall();
// Flush primary cache if not pointed at regional Redis
const redisUrl = process.env.REDIS_URL ?? process.env.BUN_REDIS_URL ?? "";
if (!isRegionalRedisUrl(redisUrl)) await redis.flushall();
else
console.log(
chalk.yellow(
"\n⚠️ Skipping redis flush (regional Redis URL detected).\n",
),
);
// Flush v2 cache (CACHE_V2_URL) if it's a distinct, non-regional connection
const cacheV2Url = process.env.CACHE_V2_URL?.trim();
if (redisV2 !== redis && cacheV2Url) {
if (!isRegionalRedisUrl(cacheV2Url)) {
await redisV2.flushall();
console.log(chalk.green("✅ Cleared CACHE_V2_URL redis.\n"));
} else {
console.log(
chalk.yellow(
"\n⚠️ Skipping CACHE_V2_URL flush (regional Redis URL detected).\n",
),
);
}
}
console.log(chalk.green("\n✅ Master org setup complete!\n"));
} catch (error) {
console.error(chalk.red("\n❌ Error:"), error);

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

@@ -11,8 +11,10 @@ import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
import chalk from "chalk";
import { addSeconds } from "date-fns";
import { redis } from "@/external/redis/initRedis";
import { redisV2 } from "@/external/redis/initRedisV2";
import { expireLock } from "@/internal/balances/finalizeLock/expireLock";
import { buildLockReceiptKey } from "@/internal/balances/utils/lock/buildLockReceiptKey";
import { fetchLockReceipt } from "@/internal/balances/utils/lock/fetchLockReceipt";
import { timeout } from "@/utils/genUtils";
import { getCustomerEvents } from "../utils/events/getCustomerEvents";
@@ -200,13 +202,10 @@ test.concurrent(`${chalk.yellowBright("check-lock-expiry-4: no expires_at sets T
lock: { enabled: true, lock_id: customerId },
});
const lockReceiptKey = buildLockReceiptKey({
orgId: ctx.org.id,
env: ctx.env,
lockKey: Bun.hash(customerId).toString(),
});
const { lockReceiptKey, source } = await fetchLockReceipt({ ctx, lockId: customerId });
const redisInstance = source === "redis_v2" ? redisV2 : redis;
const expireAt = await redis.expiretime(lockReceiptKey);
const expireAt = await redisInstance.expiretime(lockReceiptKey);
const expectedTtl = beforeCheck + 24 * 60 * 60;
// TTL should be within 5s of now + 1 day
@@ -240,13 +239,10 @@ test.concurrent(`${chalk.yellowBright("check-lock-expiry-5: expires_at set, TTL
lock: { enabled: true, lock_id: customerId, expires_at: expiresAt },
});
const lockReceiptKey = buildLockReceiptKey({
orgId: ctx.org.id,
env: ctx.env,
lockKey: Bun.hash(customerId).toString(),
});
const { lockReceiptKey, source } = await fetchLockReceipt({ ctx, lockId: customerId });
const redisInstance = source === "redis_v2" ? redisV2 : redis;
const expireAt = await redis.expiretime(lockReceiptKey);
const expireAt = await redisInstance.expiretime(lockReceiptKey);
const expectedTtl = Math.ceil(expiresAt / 1000) + 60 * 60;
// TTL should be within 5s of expires_at + 1 hour

12
server/tests/integration/balances/track/basic/track-credit-system.test.ts
View File

@@ -8,6 +8,7 @@ import { timeout } from "@tests/utils/genUtils.js";
import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
import chalk from "chalk";
import { Decimal } from "decimal.js";
import { roundCacheBalance } from "@/internal/customers/cache/fullSubject/roundCacheBalance";
import { getCreditCost } from "@/internal/features/creditSystemUtils.js";
// ═══════════════════════════════════════════════════════════════════
@@ -392,10 +393,17 @@ test.concurrent(`${chalk.yellowBright("track-credit-system4: test deduction with
balance: 0,
usage: 60,
});
expect(customer2.features[TestFeature.Credits].balance).toBe(
const creditsBalance = roundCacheBalance(
customer2.features[TestFeature.Credits].balance,
);
const credits2Balance = roundCacheBalance(
customer2.features[TestFeature.Credits2].balance,
);
expect(creditsBalance).toBe(
new Decimal(creditsBefore).sub(creditCostAction1).toNumber(),
);
expect(customer2.features[TestFeature.Credits2].balance).toBe(
expect(credits2Balance).toBe(
new Decimal(credits2Before).sub(creditCostAction3).toNumber(),
);

17
server/tests/integration/balances/track/basic/track-event-name.test.ts
View File

@@ -8,6 +8,7 @@ import { timeout } from "@tests/utils/genUtils.js";
import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
import chalk from "chalk";
import { Decimal } from "decimal.js";
import { roundCacheBalance } from "@/internal/customers/cache/fullSubject/roundCacheBalance";
// ═══════════════════════════════════════════════════════════════════
// TRACK-EVENT-NAME1: Track using event_name instead of feature_id (single feature)
@@ -109,13 +110,19 @@ test.concurrent(`${chalk.yellowBright("track-event-name2: track with event_name
expect(trackRes.value).toBe(deductValue);
expect(trackRes.balance).toBeNull();
expect(trackRes.balances).toBeDefined();
expect(trackRes.balances?.[TestFeature.Action1]?.current_balance).toBe(
expectedAction1Balance,
const currentBalance1 = roundCacheBalance(
trackRes.balances?.[TestFeature.Action1]?.current_balance,
);
expect(trackRes.balances?.[TestFeature.Action1]?.usage).toBe(deductValue);
expect(trackRes.balances?.[TestFeature.Action3]?.current_balance).toBe(
expectedAction3Balance,
expect(currentBalance1).toBe(expectedAction1Balance);
expect(
roundCacheBalance(trackRes.balances?.[TestFeature.Action1]?.usage),
).toBe(deductValue);
const currentBalance = roundCacheBalance(
trackRes.balances?.[TestFeature.Action3]?.current_balance,
);
expect(currentBalance).toBe(expectedAction3Balance);
expect(trackRes.balances?.[TestFeature.Action3]?.usage).toBe(deductValue);
const customer = await autumnV1.customers.get<ApiCustomerV3>(customerId);

2
server/tests/integration/billing/stripe-webhooks/invoice-created/invoice-created-entity-consumable.test.ts
View File

@@ -65,7 +65,7 @@ test.concurrent(`${chalk.yellowBright("invoice.created entity: regular renewal -
],
actions: [
s.attach({ productId: pro.id, entityIndex: 0 }),
s.track({ featureId: TestFeature.Messages, value: 500 }),
s.track({ featureId: TestFeature.Messages, value: 500, entityIndex: 0 }),
],
});

3
server/tests/integration/crud/customers/get-customer.test.ts
View File

@@ -315,13 +315,14 @@ test.concurrent(`${chalk.yellowBright("get-customer: customer cache updates afte
keepInternalFields: true,
},
);
ApiCustomerV5Schema.parse(afterAttach);
expectBalanceCorrect({
customer: afterAttach,
featureId: TestFeature.Credits,
remaining: 200,
usage: 0,
planId: entityProd.id,
// planId: null,
});
});

287
server/tests/integration/others/rollout/rollout-track-transition.test.ts Normal file
View File

@@ -0,0 +1,287 @@
import { test } from "bun:test";
import type { ApiCustomerV5 } from "@autumn/shared";
import { expectBalanceCorrect } from "@tests/integration/utils/expectBalanceCorrect";
import { TestFeature } from "@tests/setup/v2Features.js";
import { items } from "@tests/utils/fixtures/items.js";
import { products } from "@tests/utils/fixtures/products.js";
import { timeout } from "@tests/utils/genUtils.js";
import {
cleanupOrgRollout,
removeCachedAtField,
setOrgRolloutPercent,
} from "@tests/utils/rolloutTestUtils.js";
import { initScenario, s } from "@tests/utils/testInitUtils/initScenario.js";
import chalk from "chalk";
const testCase = "rollout-track-transition";
test(
`${chalk.yellowBright(`${testCase}: track across v1 → v2 → v1 rollout transitions`)}`,
async () => {
const monthlyMessages = items.monthlyMessages({ includedUsage: 100 });
const freeProd = products.base({
id: "free",
items: [monthlyMessages],
});
const customerId = `${testCase}`;
const { autumnV2_2, ctx } = await initScenario({
customerId,
setup: [
s.customer({ testClock: false }),
s.products({ list: [freeProd] }),
],
actions: [s.attach({ productId: freeProd.id })],
});
const orgId = ctx.org.id;
// ── Phase 1: rollout at 0% (v1 path) ──────────────────────────────
await setOrgRolloutPercent({ orgId, percent: 0 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterTrack1 =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterTrack1,
featureId: TestFeature.Messages,
remaining: 90,
});
// ── Phase 2: rollout to 100% (v2 path) ────────────────────────────
await setOrgRolloutPercent({ orgId, percent: 100 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterTrack2 =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterTrack2,
featureId: TestFeature.Messages,
remaining: 80,
});
// ── Phase 3: rollout back to 0% (v1 path) ─────────────────────────
await setOrgRolloutPercent({ orgId, percent: 0 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterTrack3 =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterTrack3,
featureId: TestFeature.Messages,
remaining: 70,
});
// ── Cleanup ────────────────────────────────────────────────────────
await cleanupOrgRollout({ orgId });
},
{ timeout: 120_000 },
);
// ─────────────────────────────────────────────────────────────────────────────
// Case A: v1 track -> remove _cachedAt (legacy) -> v2 track -> rollback v1 track
// Verifies that missing _cachedAt triggers conservative staleness eviction,
// and that v2 deductions are preserved when rolling back to v1.
// ─────────────────────────────────────────────────────────────────────────────
test(
`${chalk.yellowBright("rollout-staleness-a: legacy _cachedAt removal across v1 → v2 → v1")}`,
async () => {
const monthlyMessages = items.monthlyMessages({ includedUsage: 100 });
const freeProd = products.base({
id: "free",
items: [monthlyMessages],
});
const customerId = "rollout-staleness-a";
const { autumnV2_2, ctx } = await initScenario({
customerId,
setup: [
s.customer({ testClock: false }),
s.products({ list: [freeProd] }),
],
actions: [s.attach({ productId: freeProd.id })],
});
const orgId = ctx.org.id;
// ── Phase 1: track on v1 (rollout 0%) ─────────────────────────────
await setOrgRolloutPercent({ orgId, percent: 0 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterV1Track =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterV1Track,
featureId: TestFeature.Messages,
remaining: 90,
});
// ── Strip _cachedAt to simulate legacy cache entry ────────────────
await removeCachedAtField({
orgId: ctx.org.id,
env: ctx.env,
customerId,
});
// ── Phase 2: roll to 100%, track on v2 ───────────────────────────
await setOrgRolloutPercent({ orgId, percent: 100 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterV2Track =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterV2Track,
featureId: TestFeature.Messages,
remaining: 80,
});
// ── Phase 3: rollback to 0%, track on v1 ─────────────────────────
await setOrgRolloutPercent({ orgId, percent: 0 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterRollback =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterRollback,
featureId: TestFeature.Messages,
remaining: 70,
});
await cleanupOrgRollout({ orgId });
},
{ timeout: 120_000 },
);
// ─────────────────────────────────────────────────────────────────────────────
// Case B: v2 track -> rollback v1 track -> roll forward v2 track
// Verifies full round-trip: no deductions lost across v2 → v1 → v2.
// ─────────────────────────────────────────────────────────────────────────────
test(
`${chalk.yellowBright("rollout-staleness-b: v2 → v1 → v2 round-trip preserves balances")}`,
async () => {
const monthlyMessages = items.monthlyMessages({ includedUsage: 100 });
const freeProd = products.base({
id: "free",
items: [monthlyMessages],
});
const customerId = "rollout-staleness-b";
const { autumnV2_2, ctx } = await initScenario({
customerId,
setup: [
s.customer({ testClock: false }),
s.products({ list: [freeProd] }),
],
actions: [s.attach({ productId: freeProd.id })],
});
const orgId = ctx.org.id;
// ── Phase 1: track on v2 (rollout 100%) ──────────────────────────
await setOrgRolloutPercent({ orgId, percent: 100 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterV2Track =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterV2Track,
featureId: TestFeature.Messages,
remaining: 90,
});
// ── Phase 2: rollback to 0%, track on v1 ─────────────────────────
await setOrgRolloutPercent({ orgId, percent: 0 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterV1Track =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterV1Track,
featureId: TestFeature.Messages,
remaining: 80,
});
// ── Phase 3: roll forward to 100%, track on v2 ───────────────────
await setOrgRolloutPercent({ orgId, percent: 100 });
await autumnV2_2.track({
customer_id: customerId,
feature_id: TestFeature.Messages,
value: 10,
});
await timeout(2000);
const customerAfterRoundTrip =
await autumnV2_2.customers.get<ApiCustomerV5>(customerId);
expectBalanceCorrect({
customer: customerAfterRoundTrip,
featureId: TestFeature.Messages,
remaining: 70,
});
await cleanupOrgRollout({ orgId });
},
{ timeout: 120_000 },
);

7
server/tests/integration/utils/expectBalanceCorrect.ts
View File

@@ -8,6 +8,9 @@ import {
type ResetInterval,
} from "@autumn/shared";
const roundTo8Dp = (value: number) =>
Math.round(value * 1e8) / 1e8;
type BucketExpectation = {
included_grant?: number;
prepaid_grant?: number;
@@ -46,14 +49,14 @@ export const expectBalanceCorrect = ({
}) => {
const balance = customer.balances[featureId];
expect(balance).toBeDefined();
expect(balance.remaining).toBe(remaining);
expect(roundTo8Dp(balance.remaining)).toBe(roundTo8Dp(remaining));
if (typeof planId !== "undefined") {
expect(balance.breakdown?.[0]?.plan_id ?? null).toBe(planId);
}
if (typeof usage !== "undefined") {
expect(balance.usage).toBe(usage);
expect(roundTo8Dp(balance.usage)).toBe(roundTo8Dp(usage));
}
if (typeof nextResetAt !== "undefined") {

95
server/tests/references/getCustomerV1Response.json
View File

@@ -1,10 +1,10 @@
{
"id": "get-cus-multi-ent-v2",
"created_at": 1776094834653,
"created_at": 1776681093029,
"name": "get-cus-multi-ent-v2",
"email": "get-cus-multi-ent-v2@example.com",
"fingerprint": null,
"stripe_id": "cus_UKRUjVsG2UM9eG",
"stripe_id": "cus_UMz5mvPwIP5PS3",
"env": "sandbox",
"metadata": {},
"send_email_receipts": false,
@@ -15,25 +15,13 @@
"group": "get-cus-multi-ent-v2",
"status": "active",
"canceled_at": null,
"started_at": 1776094833000,
"started_at": 1776681092000,
"is_default": false,
"is_add_on": false,
"version": 1,
"current_period_start": 1776094833000,
"current_period_end": 1778686833000,
"current_period_start": 1776681092000,
"current_period_end": 1779273092000,
"items": [
{
"type": "price",
"feature_id": null,
"feature": null,
"interval": "month",
"interval_count": 1,
"price": 20,
"display": {
"primary_text": "$20",
"secondary_text": "per month"
}
},
{
"type": "feature",
"feature_id": "dashboard",
@@ -83,45 +71,14 @@
"group": "get-cus-multi-ent-v2",
"status": "active",
"canceled_at": null,
"started_at": 1776094847266,
"started_at": 1776681113920,
"is_default": false,
"is_add_on": false,
"version": 1,
"current_period_start": null,
"current_period_end": null,
"items": [
{
"type": "feature",
"feature_id": "credits",
"feature_type": "single_use",
"feature": {
"id": "credits",
"name": "Credits",
"type": "credit_system",
"display": {
"singular": "Credits",
"plural": "Credits"
},
"credit_schema": [
{
"metered_feature_id": "action1",
"credit_cost": 0.2
},
{
"metered_feature_id": "action2",
"credit_cost": 0.6
}
]
},
"included_usage": 200,
"interval": "month",
"reset_usage_when_enabled": true,
"display": {
"primary_text": "200 Credits"
}
}
],
"quantity": 2
"items": [],
"quantity": 1
}
],
"features": {
@@ -135,7 +92,7 @@
"balance": 90,
"usage": 10,
"included_usage": 100,
"next_reset_at": 1778686833000,
"next_reset_at": 1779273092000,
"overage_allowed": false,
"breakdown": [
{
@@ -144,7 +101,7 @@
"balance": 90,
"usage": 10,
"included_usage": 100,
"next_reset_at": 1778686833000,
"next_reset_at": 1779273092000,
"expires_at": null,
"overage_allowed": false
}
@@ -154,36 +111,14 @@
"id": "credits",
"type": "single_use",
"name": "Credits",
"interval": "month",
"interval": null,
"interval_count": 1,
"unlimited": false,
"balance": 400,
"usage": 0,
"included_usage": 400,
"next_reset_at": 1778686833000,
"next_reset_at": null,
"overage_allowed": false,
"breakdown": [
{
"interval": "month",
"interval_count": 1,
"balance": 200,
"usage": 0,
"included_usage": 200,
"next_reset_at": 1778686833000,
"expires_at": null,
"overage_allowed": false
},
{
"interval": "month",
"interval_count": 1,
"balance": 200,
"usage": 0,
"included_usage": 200,
"next_reset_at": 1778686833000,
"expires_at": null,
"overage_allowed": false
}
],
"credit_schema": [
{
"feature_id": "action1",
@@ -226,12 +161,12 @@
"product_ids": [
"cus-lvl_get-cus-multi-ent-v2"
],
"stripe_id": "in_1TLmbJ6GVhMpZYlMf5hdUuSY",
"stripe_id": "in_1TOF756GVhH8pLcDfKcFgsHu",
"status": "paid",
"total": 20,
"currency": "usd",
"created_at": 1776094833000,
"hosted_invoice_url": "http://localhost:8080/invoices/hosted_invoice_url/inv_3CJDaKtrWTXrrHak4MK0tm8cNRC"
"created_at": 1776681092000,
"hosted_invoice_url": "http://localhost:8080/invoices/hosted_invoice_url/inv_3CcNrtgUERTv8MnmpqncpHXIwRk"
}
]
}

84
server/tests/references/getCustomerV2_1Response.json
View File

@@ -2,16 +2,16 @@
"id": "get-cus-multi-ent-v2",
"name": "get-cus-multi-ent-v2",
"email": "get-cus-multi-ent-v2@example.com",
"created_at": 1776094834653,
"created_at": 1776681093029,
"fingerprint": null,
"stripe_id": "cus_UKRUjVsG2UM9eG",
"stripe_id": "cus_UMz5mvPwIP5PS3",
"env": "sandbox",
"metadata": {},
"send_email_receipts": false,
"billing_controls": {},
"subscriptions": [
{
"id": "cus_prod_3CJDZn10ncaVu1b77eOQxGOW9vf",
"id": "cus_prod_3CcNrGiJGlmGt6kWD0DmRWURUQw",
"plan_id": "cus-lvl_get-cus-multi-ent-v2",
"auto_enable": false,
"add_on": false,
@@ -20,13 +20,13 @@
"canceled_at": null,
"expires_at": null,
"trial_ends_at": null,
"started_at": 1776094833000,
"current_period_start": 1776094833000,
"current_period_end": 1778686833000,
"started_at": 1776681092000,
"current_period_start": 1776681092000,
"current_period_end": 1779273092000,
"quantity": 1
},
{
"id": "cus_prod_3CJDb647kuMvZHiYtyNTRTlhHSz",
"id": "cus_prod_3CcNtXszM7noCNIdjl9H6gs8hdj",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"auto_enable": false,
"add_on": false,
@@ -35,22 +35,7 @@
"canceled_at": null,
"expires_at": null,
"trial_ends_at": null,
"started_at": 1776094847266,
"current_period_start": null,
"current_period_end": null,
"quantity": 1
},
{
"id": "cus_prod_3CJDcGaCgzfwk14a1kkKrbyPu0y",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"auto_enable": false,
"add_on": false,
"status": "active",
"past_due": false,
"canceled_at": null,
"expires_at": null,
"trial_ends_at": null,
"started_at": 1776094856266,
"started_at": 1776681113920,
"current_period_start": null,
"current_period_end": null,
"quantity": 1
@@ -67,11 +52,11 @@
"unlimited": false,
"overage_allowed": false,
"max_purchase": null,
"next_reset_at": 1778686833000,
"next_reset_at": 1779273092000,
"breakdown": [
{
"object": "balance_breakdown",
"id": "cus_ent_3CJDZl7yzFLTUepkmclQasbbtrX",
"id": "cus_ent_3CcNrCvPuPFhzyhShn1g64uZeJ6",
"plan_id": "cus-lvl_get-cus-multi-ent-v2",
"included_grant": 100,
"prepaid_grant": 0,
@@ -80,7 +65,7 @@
"unlimited": false,
"reset": {
"interval": "month",
"resets_at": 1778686833000
"resets_at": 1779273092000
},
"price": null,
"expires_at": null,
@@ -97,49 +82,14 @@
"unlimited": false,
"overage_allowed": false,
"max_purchase": null,
"next_reset_at": 1778686833000,
"breakdown": [
{
"object": "balance_breakdown",
"id": "cus_ent_3CJDb8zdtPX9UORevJDRpsr7DAv",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"included_grant": 200,
"prepaid_grant": 0,
"remaining": 200,
"usage": 0,
"unlimited": false,
"reset": {
"interval": "month",
"resets_at": 1778686833000
},
"price": null,
"expires_at": null,
"overage": 0
},
{
"object": "balance_breakdown",
"id": "cus_ent_3CJDcJJ8aUXI3YPYBEe3bDuqRiZ",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"included_grant": 200,
"prepaid_grant": 0,
"remaining": 200,
"usage": 0,
"unlimited": false,
"reset": {
"interval": "month",
"resets_at": 1778686833000
},
"price": null,
"expires_at": null,
"overage": 0
}
]
"next_reset_at": null,
"breakdown": []
}
},
"flags": {
"dashboard": {
"object": "flag",
"id": "cus_ent_3CJDZmPcB7RaKXcOnXYPu0R1N4p",
"id": "cus_ent_3CcNrCFvKwLC3y0XoUiLazztjwy",
"plan_id": "cus-lvl_get-cus-multi-ent-v2",
"expires_at": null,
"feature_id": "dashboard"
@@ -150,12 +100,12 @@
"plan_ids": [
"cus-lvl_get-cus-multi-ent-v2"
],
"stripe_id": "in_1TLmbJ6GVhMpZYlMf5hdUuSY",
"stripe_id": "in_1TOF756GVhH8pLcDfKcFgsHu",
"status": "paid",
"total": 20,
"currency": "usd",
"created_at": 1776094833000,
"hosted_invoice_url": "http://localhost:8080/invoices/hosted_invoice_url/inv_3CJDaKtrWTXrrHak4MK0tm8cNRC"
"created_at": 1776681092000,
"hosted_invoice_url": "http://localhost:8080/invoices/hosted_invoice_url/inv_3CcNrtgUERTv8MnmpqncpHXIwRk"
}
]
}

84
server/tests/references/getCustomerV2_2Response.json
View File

@@ -2,16 +2,16 @@
"id": "get-cus-multi-ent-v2",
"name": "get-cus-multi-ent-v2",
"email": "get-cus-multi-ent-v2@example.com",
"created_at": 1776094834653,
"created_at": 1776681093029,
"fingerprint": null,
"stripe_id": "cus_UKRUjVsG2UM9eG",
"stripe_id": "cus_UMz5mvPwIP5PS3",
"env": "sandbox",
"metadata": {},
"send_email_receipts": false,
"billing_controls": {},
"subscriptions": [
{
"id": "cus_prod_3CJDZn10ncaVu1b77eOQxGOW9vf",
"id": "cus_prod_3CcNrGiJGlmGt6kWD0DmRWURUQw",
"plan_id": "cus-lvl_get-cus-multi-ent-v2",
"auto_enable": false,
"add_on": false,
@@ -20,13 +20,13 @@
"canceled_at": null,
"expires_at": null,
"trial_ends_at": null,
"started_at": 1776094833000,
"current_period_start": 1776094833000,
"current_period_end": 1778686833000,
"started_at": 1776681092000,
"current_period_start": 1776681092000,
"current_period_end": 1779273092000,
"quantity": 1
},
{
"id": "cus_prod_3CJDb647kuMvZHiYtyNTRTlhHSz",
"id": "cus_prod_3CcNtXszM7noCNIdjl9H6gs8hdj",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"auto_enable": false,
"add_on": false,
@@ -35,22 +35,7 @@
"canceled_at": null,
"expires_at": null,
"trial_ends_at": null,
"started_at": 1776094847266,
"current_period_start": null,
"current_period_end": null,
"quantity": 1
},
{
"id": "cus_prod_3CJDcGaCgzfwk14a1kkKrbyPu0y",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"auto_enable": false,
"add_on": false,
"status": "active",
"past_due": false,
"canceled_at": null,
"expires_at": null,
"trial_ends_at": null,
"started_at": 1776094856266,
"started_at": 1776681113920,
"current_period_start": null,
"current_period_end": null,
"quantity": 1
@@ -67,11 +52,11 @@
"unlimited": false,
"overage_allowed": false,
"max_purchase": null,
"next_reset_at": 1778686833000,
"next_reset_at": 1779273092000,
"breakdown": [
{
"object": "balance_breakdown",
"id": "cus_ent_3CJDZl7yzFLTUepkmclQasbbtrX",
"id": "cus_ent_3CcNrCvPuPFhzyhShn1g64uZeJ6",
"plan_id": "cus-lvl_get-cus-multi-ent-v2",
"included_grant": 100,
"prepaid_grant": 0,
@@ -80,7 +65,7 @@
"unlimited": false,
"reset": {
"interval": "month",
"resets_at": 1778686833000
"resets_at": 1779273092000
},
"price": null,
"expires_at": null,
@@ -97,49 +82,14 @@
"unlimited": false,
"overage_allowed": false,
"max_purchase": null,
"next_reset_at": 1778686833000,
"breakdown": [
{
"object": "balance_breakdown",
"id": "cus_ent_3CJDb8zdtPX9UORevJDRpsr7DAv",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"included_grant": 200,
"prepaid_grant": 0,
"remaining": 200,
"usage": 0,
"unlimited": false,
"reset": {
"interval": "month",
"resets_at": 1778686833000
},
"price": null,
"expires_at": null,
"overage": 0
},
{
"object": "balance_breakdown",
"id": "cus_ent_3CJDcJJ8aUXI3YPYBEe3bDuqRiZ",
"plan_id": "ent-prod_get-cus-multi-ent-v2",
"included_grant": 200,
"prepaid_grant": 0,
"remaining": 200,
"usage": 0,
"unlimited": false,
"reset": {
"interval": "month",
"resets_at": 1778686833000
},
"price": null,
"expires_at": null,
"overage": 0
}
]
"next_reset_at": null,
"breakdown": []
}
},
"flags": {
"dashboard": {
"object": "flag",
"id": "cus_ent_3CJDZmPcB7RaKXcOnXYPu0R1N4p",
"id": "cus_ent_3CcNrCFvKwLC3y0XoUiLazztjwy",
"plan_id": "cus-lvl_get-cus-multi-ent-v2",
"expires_at": null,
"feature_id": "dashboard"
@@ -150,12 +100,12 @@
"plan_ids": [
"cus-lvl_get-cus-multi-ent-v2"
],
"stripe_id": "in_1TLmbJ6GVhMpZYlMf5hdUuSY",
"stripe_id": "in_1TOF756GVhH8pLcDfKcFgsHu",
"status": "paid",
"total": 20,
"currency": "usd",
"created_at": 1776094833000,
"hosted_invoice_url": "http://localhost:8080/invoices/hosted_invoice_url/inv_3CJDaKtrWTXrrHak4MK0tm8cNRC"
"created_at": 1776681092000,
"hosted_invoice_url": "http://localhost:8080/invoices/hosted_invoice_url/inv_3CcNrtgUERTv8MnmpqncpHXIwRk"
}
]
}

767
server/tests/unit/full-subject-cache/sanitizeCachedPayloads.test.ts
View File

@@ -1,93 +1,146 @@
import { describe, expect, test } from "bun:test";
import { AppEnv, type SubjectBalance } from "@autumn/shared";
import {
type AggregatedFeatureBalanceSchema,
AppEnv,
type SubjectBalance,
} from "@autumn/shared";
import { z } from "zod/v4";
import type { CachedFullSubject } from "@/internal/customers/cache/fullSubject/fullSubjectCacheModel.js";
import { normalizeFromSchema } from "@/internal/customers/cache/fullSubject/sanitize/normalizeFromSchema.js";
import { sanitizeCachedAggregatedFeatureBalance } from "@/internal/customers/cache/fullSubject/sanitize/sanitizeCachedAggregatedFeatureBalance.js";
import { sanitizeCachedFullSubject } from "@/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.js";
import { sanitizeCachedSubjectBalance } from "@/internal/customers/cache/fullSubject/sanitize/sanitizeCachedSubjectBalance.js";
import { sanitizeShape } from "@/internal/customers/cache/fullSubject/sanitize/sanitizeCacheShapeUtils.js";
describe("sanitizeShape (recursive core)", () => {
test("should coerce non-array to [] for 'array' rule", () => {
const result = sanitizeShape({
value: { items: {} },
spec: { items: "array" },
describe("normalizeFromSchema (core walker)", () => {
test("fills undefined at nullable position with null", () => {
const schema = z.object({ expires_at: z.number().nullable() });
const result = normalizeFromSchema<{ expires_at: number | null }>({
schema,
data: {},
});
expect(result).toEqual({ items: [] });
expect(result.expires_at).toBeNull();
});
test("should preserve valid arrays", () => {
const result = sanitizeShape({
value: { items: [1, 2, 3] },
spec: { items: "array" },
test("treats .nullish() as nullable (undefined -> null)", () => {
const schema = z.object({ x: z.number().nullish() });
const result = normalizeFromSchema<{ x: number | null | undefined }>({
schema,
data: {},
});
expect(result).toEqual({ items: [1, 2, 3] });
expect(result.x).toBeNull();
});
test("should coerce non-object to {} for 'record' rule", () => {
const result = sanitizeShape({
value: { flags: [] },
spec: { flags: "record" },
test("leaves pure .optional() undefined as undefined", () => {
const schema = z.object({ x: z.number().optional() });
const result = normalizeFromSchema<{ x: number | undefined }>({
schema,
data: {},
});
expect(result).toEqual({ flags: {} });
expect(result.x).toBeUndefined();
});
test("should coerce non-object to null for 'nullable_record' rule", () => {
const result = sanitizeShape({
value: { entities: [] },
spec: { entities: "nullable_record" },
});
expect(result).toEqual({ entities: null });
test("applies ZodDefault when data is undefined", () => {
const schema = z.object({ x: z.number().default(42) });
const result = normalizeFromSchema<{ x: number }>({ schema, data: {} });
expect(result.x).toBe(42);
});
test("should preserve null for 'nullable_record' rule", () => {
const result = sanitizeShape({
value: { entities: null },
spec: { entities: "nullable_record" },
test("ZodDefault wins over nullable wrapper", () => {
const schema = z.object({ adjustment: z.number().nullable().default(0) });
const result = normalizeFromSchema<{ adjustment: number | null }>({
schema,
data: {},
});
expect(result).toEqual({ entities: null });
expect(result.adjustment).toBe(0);
});
test("should recurse into nested object specs", () => {
const result = sanitizeShape({
value: { feature: { event_names: {} } },
spec: { feature: { event_names: "array" } },
test("coerces empty object {} to [] when schema says array", () => {
const schema = z.object({ items: z.array(z.string()) });
const result = normalizeFromSchema<{ items: string[] }>({
schema,
data: { items: {} },
});
expect(result).toEqual({ feature: { event_names: [] } });
expect(result.items).toEqual([]);
});
test("should handle { items: spec } for array-of-objects", () => {
const result = sanitizeShape({
value: { rollovers: [{}, { entities: "bad" }] },
spec: { rollovers: { items: { entities: "record" } } },
test("coerces empty array [] to {} when schema says record", () => {
const schema = z.object({
flags: z.record(z.string(), z.boolean()),
});
expect(result).toEqual({
rollovers: [{ entities: {} }, { entities: {} }],
const result = normalizeFromSchema<{ flags: Record<string, boolean> }>({
schema,
data: { flags: [] },
});
expect(result.flags).toEqual({});
});
test("should coerce then recurse for { items: spec } when field is not an array", () => {
const result = sanitizeShape({
value: { rollovers: {} },
spec: { rollovers: { items: { entities: "record" } } },
test("recurses into nested arrays (filling null inside array items)", () => {
const schema = z.object({
items: z.array(z.object({ id: z.string(), due: z.number().nullable() })),
});
expect(result).toEqual({ rollovers: [] });
const result = normalizeFromSchema<{
items: Array<{ id: string; due: number | null }>;
}>({
schema,
data: { items: [{ id: "a" }, { id: "b", due: 5 }] },
});
expect(result.items[0].due).toBeNull();
expect(result.items[1].due).toBe(5);
});
test("should leave unspecified fields untouched", () => {
const result = sanitizeShape({
value: { name: "test", items: {} },
spec: { items: "array" },
test("recurses into records (filling null inside record values)", () => {
const schema = z.object({
flags: z.record(
z.string(),
z.object({ expiresAt: z.number().nullable() }),
),
});
expect(result).toEqual({ name: "test", items: [] });
const result = normalizeFromSchema<{
flags: Record<string, { expiresAt: number | null }>;
}>({
schema,
data: { flags: { a: {} } },
});
expect(result.flags.a.expiresAt).toBeNull();
});
test("should return {} for non-object input", () => {
const result = sanitizeShape({ value: "not_an_object", spec: {} });
expect(result).toEqual({});
test("preserves unknown keys not covered by schema", () => {
const schema = z.object({ known: z.string() });
const result = normalizeFromSchema<Record<string, unknown>>({
schema,
data: { known: "x", unknown: "preserved" },
});
expect(result.unknown).toBe("preserved");
});
test("returns null for explicit null passthrough", () => {
const schema = z.object({ x: z.number().nullable() });
const result = normalizeFromSchema<{ x: number | null }>({
schema,
data: { x: null },
});
expect(result.x).toBeNull();
});
test("does not touch present non-nullable fields", () => {
const schema = z.object({ name: z.string() });
const result = normalizeFromSchema<{ name: string }>({
schema,
data: { name: "abc" },
});
expect(result.name).toBe("abc");
});
test("never throws on schema-mismatched data", () => {
const schema = z.object({ x: z.number().nullable() });
expect(() =>
normalizeFromSchema({ schema, data: "not an object" }),
).not.toThrow();
});
});
describe("sanitizeCachedSubjectBalance", () => {
const buildMalformedSubjectBalance = (): unknown => ({
const buildSubjectBalance = (): unknown => ({
id: "cus_ent_1",
customer_product_id: "cp_1",
entitlement_id: "ent_1",
@@ -95,19 +148,13 @@ describe("sanitizeCachedSubjectBalance", () => {
internal_entity_id: null,
internal_feature_id: "feat_int_1",
feature_id: "messages",
unlimited: false,
balance: 100,
adjustment: 0,
additional_balance: 0,
usage_allowed: true,
next_reset_at: null,
expires_at: null,
external_id: null,
cache_version: 1,
entities: null,
created_at: 1000,
customer_id: "cus_1",
rollovers: {},
entities: [],
replaceables: [],
rollovers: [],
entitlement: {
id: "ent_1",
created_at: 1,
@@ -125,426 +172,332 @@ describe("sanitizeCachedSubjectBalance", () => {
type: "metered",
config: { usage_type: "single", schema: {} },
archived: false,
event_names: {},
event_names: [],
display: null,
},
},
customerPrice: {
id: "cp_1",
internal_customer_id: "cus_int_1",
customer_product_id: "cp_1",
created_at: 1,
price_id: "price_1",
price: {
id: "price_1",
internal_product_id: "prod_int_1",
billing_type: null,
tier_behavior: null,
config: {
type: "usage",
bill_when: "end_of_period",
internal_feature_id: "feat_int_1",
feature_id: "messages",
usage_tiers: {},
interval: "month",
},
entitlement_id: null,
proration_config: null,
},
},
customerPrice: null,
customerProductOptions: null,
customerProductQuantity: 1,
isEntityLevel: false,
});
test("should coerce rollovers from {} to []", () => {
const malformed = buildMalformedSubjectBalance() as SubjectBalance;
test("fills dropped nullable scalars (Upstash null-drop repair)", () => {
const malformed = buildSubjectBalance() as Record<string, unknown>;
// Remove fields that Upstash Lua cjson would have dropped when they were null
delete malformed.expires_at;
delete malformed.next_reset_at;
delete malformed.unlimited;
delete malformed.usage_allowed;
delete malformed.external_id;
delete malformed.cache_version;
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed,
subjectBalance: malformed as unknown as SubjectBalance,
});
expect(result.expires_at).toBeNull();
expect(result.next_reset_at).toBeNull();
expect(result.unlimited).toBeNull();
expect(result.usage_allowed).toBeNull();
expect(result.external_id).toBeNull();
// cache_version has .default(0) in CustomerEntitlementSchema; default wins
expect(result.cache_version).toBe(0);
});
test("coerces rollovers from {} to []", () => {
const malformed = buildSubjectBalance() as Record<string, unknown>;
malformed.rollovers = {};
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed as unknown as SubjectBalance,
});
expect(Array.isArray(result.rollovers)).toBe(true);
expect(result.rollovers).toEqual([]);
});
test("should coerce entities from [] to null", () => {
const malformed = buildMalformedSubjectBalance() as SubjectBalance;
test("fills dropped expires_at inside rollovers array items", () => {
const malformed = buildSubjectBalance() as Record<string, unknown>;
malformed.rollovers = [
{ id: "r1", cus_ent_id: "ce1", balance: 50, usage: 0, entities: {} },
];
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed,
subjectBalance: malformed as unknown as SubjectBalance,
});
expect(result.entities).toBeNull();
expect(result.rollovers.length).toBe(1);
expect(result.rollovers[0].expires_at).toBeNull();
});
test("should coerce entitlement.feature.event_names from {} to []", () => {
const malformed = buildMalformedSubjectBalance() as SubjectBalance;
test("passes through helper fields not on the schema (isEntityLevel, customerProductQuantity)", () => {
const malformed = buildSubjectBalance() as Record<string, unknown>;
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed,
subjectBalance: malformed as unknown as SubjectBalance,
});
expect(Array.isArray(result.entitlement.feature.event_names)).toBe(true);
expect(result.entitlement.feature.event_names).toEqual([]);
expect(result.isEntityLevel).toBe(false);
expect(result.customerProductQuantity).toBe(1);
});
test("should coerce customerPrice.price.config.usage_tiers from {} to []", () => {
const malformed = buildMalformedSubjectBalance() as SubjectBalance;
test("preserves valid scalar fields untouched", () => {
const malformed = buildSubjectBalance() as Record<string, unknown>;
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed,
});
const config = result.customerPrice?.price?.config as Record<
string,
unknown
>;
expect(Array.isArray(config?.usage_tiers)).toBe(true);
});
test("should preserve valid fields untouched", () => {
const malformed = buildMalformedSubjectBalance() as SubjectBalance;
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed,
subjectBalance: malformed as unknown as SubjectBalance,
});
expect(result.id).toBe("cus_ent_1");
expect(result.balance).toBe(100);
expect(result.feature_id).toBe("messages");
expect(result.entitlement.feature.name).toBe("Messages");
});
test("should handle rollovers array with nested entities coercion", () => {
const malformed = buildMalformedSubjectBalance() as SubjectBalance;
(malformed as unknown as Record<string, unknown>).rollovers = [
{ id: "r1", cus_ent_id: "ce1", balance: 50, usage: 0, entities: "bad" },
];
const result = sanitizeCachedSubjectBalance({
subjectBalance: malformed,
});
expect(Array.isArray(result.rollovers)).toBe(true);
expect(result.rollovers.length).toBe(1);
expect(result.rollovers[0].entities).toEqual({});
});
});
describe("sanitizeCachedFullSubject", () => {
const buildMalformedCachedFullSubject = (): unknown => ({
const buildCachedFullSubject = (): unknown => ({
subjectType: "customer",
customerId: "cus_1",
internalCustomerId: "cus_int_1",
_cachedAt: Date.now(),
subjectViewEpoch: 1,
meteredFeatures: {},
meteredFeatures: [],
customerEntitlementIdsByFeatureId: {},
customer: {
id: "cus_1",
internal_id: "cus_int_1",
org_id: "org_1",
env: AppEnv.Live,
created_at: 1,
name: "Test",
email: null,
fingerprint: null,
processor: null,
processors: null,
metadata: {},
send_email_receipts: false,
auto_topups: {},
spend_limits: {},
usage_alerts: {},
overage_allowed: {},
},
entity: {
id: "ent_1",
org_id: "org_1",
created_at: 1,
internal_id: "ent_int_1",
internal_customer_id: "cus_int_1",
env: "live",
name: null,
deleted: false,
feature_id: "messages",
internal_feature_id: "feat_int_1",
spend_limits: {},
usage_alerts: {},
overage_allowed: {},
},
customer_products: {},
products: {},
entitlements: [
customer_products: [],
products: [],
entitlements: [],
prices: [],
free_trials: [],
subscriptions: [],
invoices: [],
flags: {},
});
test("fills customer.email / customer.name dropped by Upstash", () => {
const malformed = buildCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({ cachedFullSubject: malformed });
expect(result.customer.email).toBeNull();
expect(result.customer.name).toBeNull();
expect(result.customer.fingerprint).toBeNull();
});
test("fills entity.name / entity.id dropped by Upstash", () => {
const malformed = buildCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({ cachedFullSubject: malformed });
expect(result.entity?.name).toBeNull();
expect(result.entity?.id).toBeNull();
});
test("coerces subscriptions from {} to [] (Upstash empty-table encoding)", () => {
const malformed = buildCachedFullSubject() as unknown as Record<
string,
unknown
>;
malformed.subscriptions = {};
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed as unknown as CachedFullSubject,
});
expect(Array.isArray(result.subscriptions)).toBe(true);
expect(result.subscriptions).toEqual([]);
});
test("coerces flags from [] to {} and fills nullable scalars on flag entries", () => {
const malformed = buildCachedFullSubject() as unknown as Record<
string,
unknown
>;
malformed.flags = {
seat: {
featureId: "seat",
internalFeatureId: "if_seat",
entitlementId: "e_seat",
customerEntitlementId: "ce_seat",
internalCustomerId: "cus_int_1",
// customerProductId, internalEntityId, expiresAt, externalId all dropped
},
};
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed as unknown as CachedFullSubject,
});
expect(Array.isArray(result.flags)).toBe(false);
const seat = (result.flags as Record<string, Record<string, unknown>>).seat;
expect(seat.customerProductId).toBeNull();
expect(seat.internalEntityId).toBeNull();
expect(seat.expiresAt).toBeNull();
expect(seat.externalId).toBeNull();
});
test("fills dropped canceled_at / ended_at inside customer_products array", () => {
const malformed = buildCachedFullSubject() as unknown as Record<
string,
unknown
>;
malformed.customer_products = [
{
id: "ent_1",
id: "cp_1",
internal_product_id: "ip_1",
product_id: "p_1",
internal_customer_id: "cus_int_1",
created_at: 1,
internal_feature_id: "feat_int_1",
internal_product_id: "prod_int_1",
status: "active",
canceled: false,
starts_at: 1,
options: [],
collection_method: "charge_automatically",
api_semver: null,
is_custom: false,
interval_count: 1,
feature: {
internal_id: "feat_int_1",
org_id: "org_1",
created_at: 1,
env: "sandbox",
id: "messages",
name: "Messages",
type: "metered",
config: {},
archived: false,
event_names: {},
display: null,
},
billing_version: "v1",
external_id: null,
// trial_ends_at / canceled_at / ended_at / billing_cycle_anchor_resets_at
// / free_trial_id dropped
},
],
prices: [
];
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed as unknown as CachedFullSubject,
});
const cp = result.customer_products[0] as unknown as Record<
string,
unknown
>;
expect(cp.trial_ends_at).toBeNull();
expect(cp.canceled_at).toBeNull();
expect(cp.ended_at).toBeNull();
expect(cp.billing_cycle_anchor_resets_at).toBeNull();
expect(cp.free_trial_id).toBeNull();
});
test("fills dropped canceled / current_period_start on subscriptions", () => {
const malformed = buildCachedFullSubject() as unknown as Record<
string,
unknown
>;
malformed.subscriptions = [
{
id: "price_1",
internal_product_id: "prod_int_1",
billing_type: null,
tier_behavior: null,
config: {
type: "usage",
bill_when: "end_of_period",
internal_feature_id: "feat_int_1",
feature_id: "messages",
usage_tiers: {},
interval: "month",
},
entitlement_id: null,
proration_config: null,
id: "sub_1",
created_at: 1,
usage_features: [],
org_id: "org_1",
env: "live",
// stripe_id, stripe_schedule_id, current_period_start/end dropped
},
],
free_trials: {},
subscriptions: {},
invoices: [
];
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed as unknown as CachedFullSubject,
});
expect(result.subscriptions[0].stripe_id).toBeNull();
expect(result.subscriptions[0].stripe_schedule_id).toBeNull();
expect(result.subscriptions[0].current_period_start).toBeNull();
expect(result.subscriptions[0].current_period_end).toBeNull();
});
test("fills dropped hosted_invoice_url / internal_entity_id on invoices", () => {
const malformed = buildCachedFullSubject() as unknown as Record<
string,
unknown
>;
malformed.invoices = [
{
id: "inv_1",
created_at: 1,
internal_customer_id: "cus_int_1",
internal_entity_id: null,
product_ids: {},
internal_product_ids: {},
product_ids: [],
internal_product_ids: [],
stripe_id: "in_1",
status: "paid",
hosted_invoice_url: null,
total: 100,
currency: "usd",
discounts: {},
items: {},
},
],
flags: [],
entity_aggregations: {
aggregated_customer_products: [
{
id: "acp_1",
internal_product_id: "prod_int_1",
product_id: "prod_1",
internal_customer_id: "cus_int_1",
created_at: 1,
status: "active",
canceled: false,
starts_at: 1,
options: {},
collection_method: "charge_automatically",
quantity: 1,
api_semver: null,
is_custom: false,
billing_version: "v1",
external_id: null,
subscription_ids: {},
scheduled_ids: {},
},
],
aggregated_customer_entitlements: {},
},
});
test("should coerce top-level arrays from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.customer_products)).toBe(true);
expect(result.customer_products).toEqual([]);
expect(Array.isArray(result.products)).toBe(true);
expect(result.products).toEqual([]);
expect(Array.isArray(result.free_trials)).toBe(true);
expect(result.free_trials).toEqual([]);
});
test("should coerce meteredFeatures from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.meteredFeatures)).toBe(true);
expect(result.meteredFeatures).toEqual([]);
});
test("should coerce flags from [] to {}", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.flags)).toBe(false);
expect(typeof result.flags).toBe("object");
expect(result.flags).toEqual({});
});
test("should coerce customer.auto_topups from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.customer.auto_topups)).toBe(true);
});
test("should coerce customer.spend_limits from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.customer.spend_limits)).toBe(true);
});
test("should coerce customer.usage_alerts from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.customer.usage_alerts)).toBe(true);
});
test("should coerce customer.overage_allowed from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.customer.overage_allowed)).toBe(true);
});
test("should coerce entity.spend_limits from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.entity?.spend_limits)).toBe(true);
});
test("should coerce entity.usage_alerts from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.entity?.usage_alerts)).toBe(true);
});
test("should coerce entity.overage_allowed from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.entity?.overage_allowed)).toBe(true);
});
test("should coerce entitlements[].feature.event_names from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
const entitlements = result.entitlements as Array<{
feature: { event_names: unknown };
}>;
expect(entitlements.length).toBe(1);
expect(Array.isArray(entitlements[0].feature.event_names)).toBe(true);
expect(entitlements[0].feature.event_names).toEqual([]);
});
test("should coerce prices[].config.usage_tiers from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
const prices = result.prices as Array<{
config: { usage_tiers: unknown };
}>;
expect(prices.length).toBe(1);
expect(Array.isArray(prices[0].config.usage_tiers)).toBe(true);
});
test("should coerce invoices[].product_ids from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(result.invoices.length).toBe(1);
expect(Array.isArray(result.invoices[0].product_ids)).toBe(true);
});
test("should coerce invoices[].internal_product_ids from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.invoices[0].internal_product_ids)).toBe(true);
});
test("should coerce invoices[].discounts from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.invoices[0].discounts)).toBe(true);
});
test("should coerce invoices[].items from {} to []", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
expect(Array.isArray(result.invoices[0].items)).toBe(true);
});
test("should coerce entity_aggregations nested arrays", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
});
const entityAgg = result.entity_aggregations;
expect(entityAgg).toBeDefined();
expect(Array.isArray(entityAgg?.aggregated_customer_entitlements)).toBe(
true,
);
const products = entityAgg?.aggregated_customer_products ?? [];
expect(Array.isArray(products)).toBe(true);
expect(products.length).toBe(1);
expect(Array.isArray(products[0].options)).toBe(true);
expect(Array.isArray(products[0].subscription_ids)).toBe(true);
expect(Array.isArray(products[0].scheduled_ids)).toBe(true);
});
test("should coerce subscriptions from {} to [] and recurse usage_features", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
(malformed as unknown as Record<string, unknown>).subscriptions = [
{
id: "sub_1",
stripe_id: null,
stripe_schedule_id: null,
created_at: 1,
usage_features: {},
org_id: "org_1",
current_period_start: null,
current_period_end: null,
env: "live",
discounts: [],
items: [],
// hosted_invoice_url, internal_entity_id dropped
},
];
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
cachedFullSubject: malformed as unknown as CachedFullSubject,
});
expect(Array.isArray(result.subscriptions)).toBe(true);
expect(result.subscriptions.length).toBe(1);
expect(Array.isArray(result.subscriptions[0].usage_features)).toBe(true);
expect(result.invoices[0].hosted_invoice_url).toBeNull();
expect(result.invoices[0].internal_entity_id).toBeNull();
});
test("should preserve scalar fields untouched", () => {
const malformed = buildMalformedCachedFullSubject() as CachedFullSubject;
test("preserves unknown top-level keys not on the schema", () => {
const malformed = buildCachedFullSubject() as unknown as Record<
string,
unknown
>;
malformed._futureField = "future";
const result = sanitizeCachedFullSubject({
cachedFullSubject: malformed,
cachedFullSubject: malformed as unknown as CachedFullSubject,
});
expect((result as unknown as Record<string, unknown>)._futureField).toBe(
"future",
);
});
test("preserves scalar cache metadata fields", () => {
const malformed = buildCachedFullSubject() as CachedFullSubject;
const result = sanitizeCachedFullSubject({ cachedFullSubject: malformed });
expect(result.customerId).toBe("cus_1");
expect(result.subjectViewEpoch).toBe(1);
expect(result.customer.name).toBe("Test");
});
});
describe("sanitizeCachedAggregatedFeatureBalance", () => {
test("fills undefined at nullable positions for AggregatedFeatureBalance", () => {
// AggregatedFeatureBalance entities is .nullish() — undefined should
// become null after walker pass.
const malformed = {
api_id: "f1",
internal_feature_id: "if_1",
internal_customer_id: "cus_int_1",
feature_id: "messages",
allowance_total: 100,
balance: 50,
adjustment: 0,
additional_balance: 0,
unlimited: false,
usage_allowed: false,
entity_count: 0,
// entities dropped by Upstash null-strip
};
const result = sanitizeCachedAggregatedFeatureBalance({
aggregated: malformed as unknown as z.infer<
typeof AggregatedFeatureBalanceSchema
>,
});
expect(result.entities).toBeNull();
});
test("applies schema defaults for rollover_balance / rollover_usage", () => {
const malformed = {
api_id: "f1",
internal_feature_id: "if_1",
internal_customer_id: "cus_int_1",
feature_id: "messages",
allowance_total: 100,
balance: 50,
adjustment: 0,
additional_balance: 0,
unlimited: false,
usage_allowed: false,
entity_count: 0,
entities: null,
};
const result = sanitizeCachedAggregatedFeatureBalance({
aggregated: malformed as unknown as z.infer<
typeof AggregatedFeatureBalanceSchema
>,
});
// rollover_balance has .default(0)
expect(result.rollover_balance).toBe(0);
expect(result.rollover_usage).toBe(0);
expect(result.prepaid_grant_from_options).toBe(0);
});
});

214
server/tests/unit/redis-otel/parseRedisKeyContext.test.ts Normal file
View File

@@ -0,0 +1,214 @@
import { describe, expect, test } from "bun:test";
import { extractKey } from "@/external/redis/otel/instrumentRedis.js";
import { parseRedisKeyContext } from "@/external/redis/otel/parseRedisKeyContext.js";
import {
buildFullSubjectBalanceKey,
buildFullSubjectKey,
buildFullSubjectViewEpochKey,
buildSharedFullSubjectBalanceKey,
} from "@/internal/customers/cache/fullSubject/index.js";
describe("parseRedisKeyContext - FullSubject V2", () => {
test("parses base subject key", () => {
const key = buildFullSubjectKey({
orgId: "org_abc",
env: "sandbox",
customerId: "cus_1",
});
expect(parseRedisKeyContext({ key })).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: undefined,
generation: "v2",
});
});
test("parses entity-variant subject key", () => {
const key = buildFullSubjectKey({
orgId: "org_abc",
env: "sandbox",
customerId: "cus_1",
entityId: "ent_42",
});
expect(parseRedisKeyContext({ key })).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: "ent_42",
generation: "v2",
});
});
test("parses shared-balance key", () => {
const key = buildSharedFullSubjectBalanceKey({
orgId: "org_abc",
env: "sandbox",
customerId: "cus_1",
featureId: "messages",
});
expect(parseRedisKeyContext({ key })).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: undefined,
generation: "v2",
});
});
test("parses per-feature balance key", () => {
const key = buildFullSubjectBalanceKey({
orgId: "org_abc",
env: "sandbox",
customerId: "cus_1",
featureId: "messages",
});
expect(parseRedisKeyContext({ key })).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: undefined,
generation: "v2",
});
});
test("parses view-epoch key", () => {
const key = buildFullSubjectViewEpochKey({
orgId: "org_abc",
env: "sandbox",
customerId: "cus_1",
});
expect(parseRedisKeyContext({ key })).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: undefined,
generation: "v2",
});
});
test("does not mis-extract customerId as orgId for entity keys with hyphenated ids", () => {
const key = buildFullSubjectKey({
orgId: "org_2sWv2S8LJ9iaTjLI6UtNsfL88Kt",
env: "sandbox",
customerId: "track-rollover2",
entityId: "1",
});
const parsed = parseRedisKeyContext({ key });
expect(parsed.orgId).toBe("org_2sWv2S8LJ9iaTjLI6UtNsfL88Kt");
expect(parsed.customerId).toBe("track-rollover2");
expect(parsed.entityId).toBe("1");
expect(parsed.generation).toBe("v2");
});
});
describe("parseRedisKeyContext - V1 shapes (regression)", () => {
test("parses {orgId}:env:customer:version:customerId", () => {
expect(
parseRedisKeyContext({
key: "{org_abc}:sandbox:customer:v1:cus_1",
}),
).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: undefined,
generation: "v1",
});
});
test("parses {orgId}:env:customer:...:entity:entityId", () => {
expect(
parseRedisKeyContext({
key: "{org_abc}:sandbox:customer:v1:cus_1:entity:ent_42",
}),
).toEqual({
orgId: "org_abc",
customerId: "cus_1",
entityId: "ent_42",
generation: "v1",
});
});
test("parses {orgId}:env:fullcustomer:version:customerId", () => {
expect(
parseRedisKeyContext({
key: "{org_abc}:sandbox:fullcustomer:v1:cus_1",
}),
).toEqual({
orgId: "org_abc",
customerId: "cus_1",
generation: "v1",
});
});
test("parses {orgId}:env:customer_guard:customerId", () => {
expect(
parseRedisKeyContext({
key: "{org_abc}:sandbox:customer_guard:cus_1",
}),
).toEqual({
orgId: "org_abc",
customerId: "cus_1",
generation: "v1",
});
});
test("parses {orgId}:env:test_cache_delete_guard:customerId", () => {
expect(
parseRedisKeyContext({
key: "{org_abc}:sandbox:test_cache_delete_guard:cus_1",
}),
).toEqual({
orgId: "org_abc",
customerId: "cus_1",
generation: "v1",
});
});
test("tags unknown-kind hash-tagged keys as v1", () => {
expect(
parseRedisKeyContext({
key: "{org_abc}:sandbox:some_other_kind:cus_1",
}),
).toEqual({
orgId: "org_abc",
generation: "v1",
});
});
});
describe("extractKey - numeric-first-arg custom commands", () => {
test("uses args[1] as key when args[0] is the key count", () => {
expect(
extractKey({
args: [3, "{cus_1}:org_abc:sandbox:full_subject", "k2", "k3", "arg"],
}),
).toBe("{cus_1}:org_abc:sandbox:full_subject");
});
test("uses args[0] when it is a string (commands with numberOfKeys)", () => {
expect(
extractKey({
args: ["{cus_1}:org_abc:sandbox:full_subject", "arg1", "arg2"],
}),
).toBe("{cus_1}:org_abc:sandbox:full_subject");
});
test("falls back to undefined when args[1] is not a key", () => {
expect(extractKey({ args: [0] })).toBeUndefined();
expect(extractKey({ args: [] })).toBeUndefined();
});
test("handles Buffer keys", () => {
expect(
extractKey({
args: [1, Buffer.from("{cus}:org:env:full_subject", "utf8")],
}),
).toBe("{cus}:org:env:full_subject");
});
});
describe("parseRedisKeyContext - edge cases", () => {
test("returns empty for undefined key", () => {
expect(parseRedisKeyContext({ key: undefined })).toEqual({});
});
test("returns empty for short key", () => {
expect(parseRedisKeyContext({ key: "foo:bar" })).toEqual({});
});
});

56
server/tests/utils/rolloutTestUtils.ts Normal file
View File

@@ -0,0 +1,56 @@
import { redis } from "@/external/redis/initRedis.js";
import { buildFullCustomerCacheKey } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/fullCustomerCacheConfig.js";
import { FULL_SUBJECT_ROLLOUT_ID } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js";
import {
removeRolloutOrg,
updateRolloutPercent,
} from "@/internal/misc/rollouts/rolloutConfigStore.js";
import { timeout } from "@tests/utils/genUtils.js";
const POLL_SETTLE_MS = 3000;
/**
* Sets the v2-cache rollout percentage for a specific org and waits
* for the server's edge config poll to pick up the change.
*/
export const setOrgRolloutPercent = async ({
orgId,
percent,
}: {
orgId: string;
percent: number;
}) => {
await updateRolloutPercent({
rolloutId: FULL_SUBJECT_ROLLOUT_ID,
orgId,
percent,
});
await timeout(POLL_SETTLE_MS);
};
/**
* Strips _cachedAt from the v1 FullCustomer cache entry to simulate
* a legacy cache entry without a timestamp.
*/
export const removeCachedAtField = async ({
orgId,
env,
customerId,
}: {
orgId: string;
env: string;
customerId: string;
}) => {
const cacheKey = buildFullCustomerCacheKey({ orgId, env, customerId });
await redis.call("JSON.DEL", cacheKey, "$._cachedAt");
};
/**
* Removes the org-level rollout override (cleanup after test).
*/
export const cleanupOrgRollout = async ({ orgId }: { orgId: string }) => {
await removeRolloutOrg({
rolloutId: FULL_SUBJECT_ROLLOUT_ID,
orgId,
});
};

58
shared/models/cusModels/fullSubject/normalizedFullSubjectModel.ts
View File

@@ -1,10 +1,22 @@
import type { AggregatedFeatureBalance } from "../../cusProductModels/cusEntModels/aggregatedCusEnt.js";
import type { EntityBalance } from "../../cusProductModels/cusEntModels/cusEntModels.js";
import { z } from "zod/v4";
import {
type AggregatedFeatureBalance,
AggregatedFeatureBalanceSchema,
} from "../../cusProductModels/cusEntModels/aggregatedCusEnt.js";
import {
type EntityBalance,
FullCustomerEntitlementSchema,
} from "../../cusProductModels/cusEntModels/cusEntModels.js";
import type { Replaceable } from "../../cusProductModels/cusEntModels/replaceableTable.js";
import type { DbRollover } from "../../cusProductModels/cusEntModels/rolloverModels/rolloverTable.js";
import type { FullCustomerPrice } from "../../cusProductModels/cusPriceModels/cusPriceModels.js";
import { FullCustomerPriceSchema } from "../../cusProductModels/cusPriceModels/cusPriceModels.js";
import type { DbCustomerPrice } from "../../cusProductModels/cusPriceModels/cusPriceTable.js";
import type { FeatureOptions } from "../../cusProductModels/cusProductModels.js";
import {
CusProductSchema,
type FeatureOptions,
FeatureOptionsSchema,
} from "../../cusProductModels/cusProductModels.js";
import type { DbCustomerProduct } from "../../cusProductModels/cusProductTable.js";
import type { EntitlementWithFeature } from "../../productModels/entModels/entModels.js";
import type { DbFreeTrial } from "../../productModels/freeTrialModels/freeTrialTable.js";
@@ -16,6 +28,22 @@ import type { Entity } from "../entityModels/entityModels.js";
import type { Invoice } from "../invoiceModels/invoiceModels.js";
import type { SubjectType } from "./fullSubjectModel.js";
/**
* Schema mirror of the `SubjectFlag` shape. Used by the cached-payload
* schema walker to know where nullable positions are.
*/
export const SubjectFlagSchema = z.object({
featureId: z.string(),
internalFeatureId: z.string(),
entitlementId: z.string(),
customerEntitlementId: z.string(),
customerProductId: z.string().nullable(),
internalCustomerId: z.string(),
internalEntityId: z.string().nullable(),
expiresAt: z.number().nullable(),
externalId: z.string().nullable(),
});
export type SubjectFlag = {
featureId: string;
internalFeatureId: string;
@@ -28,6 +56,21 @@ export type SubjectFlag = {
externalId: string | null;
};
/**
* Schema mirror of `SubjectBalance`. Extends `FullCustomerEntitlementSchema`
* with the helper fields attached during normalization (customerPrice,
* customerProductOptions, customerProductQuantity, isEntityLevel).
*
* Used by the cache-hole-filling walker; this is not a validator — the
* runtime `SubjectBalance` type below is the source of truth.
*/
export const SubjectBalanceSchema = FullCustomerEntitlementSchema.extend({
customerPrice: FullCustomerPriceSchema.nullable(),
customerProductOptions: FeatureOptionsSchema.nullable(),
customerProductQuantity: z.number(),
isEntityLevel: z.boolean(),
});
export type SubjectBalance = {
id: string;
customer_product_id: string | null;
@@ -58,6 +101,15 @@ export type SubjectBalance = {
isEntityLevel: boolean;
};
/**
* Schema mirror of `EntityAggregations`. Reuses `CusProductSchema` as the Zod
* mirror of `DbCustomerProduct` for the aggregated customer products array.
*/
export const EntityAggregationsSchema = z.object({
aggregated_customer_products: z.array(CusProductSchema),
aggregated_customer_entitlements: z.array(AggregatedFeatureBalanceSchema),
});
export type EntityAggregations = {
aggregated_customer_products: DbCustomerProduct[];
aggregated_customer_entitlements: AggregatedFeatureBalance[];