diff --git a/ai b/ai index a04ded379..e66bb8079 160000 --- a/ai +++ b/ai @@ -1 +1 @@ -Subproject commit a04ded379f43b3c8a23d7cc33eb125cce557500b +Subproject commit e66bb80796bf8387e4fbcc550f9d520eaa182f6e diff --git a/server/src/internal/balances/track/v3/runTrackV3.ts b/server/src/internal/balances/track/v3/runTrackV3.ts index b39dc7437..a9291195e 100644 --- a/server/src/internal/balances/track/v3/runTrackV3.ts +++ b/server/src/internal/balances/track/v3/runTrackV3.ts @@ -13,6 +13,7 @@ import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; import { getOrCreateCachedFullSubject } from "@/internal/customers/cache/fullSubject/actions/getOrCreateCachedFullSubject.js"; import { getOrSetCachedFullSubject } from "@/internal/customers/cache/fullSubject/actions/getOrSetCachedFullSubject.js"; import type { FeatureDeduction } from "../../utils/types/featureDeduction.js"; +import { handleEventIdempotencyKey } from "../utils/handleEventIdempotencyKey.js"; import { runRedisTrackV3 } from "./runRedisTrackV3.js"; import { getTrackIdempotencyKey } from "./trackIdempotencyKey.js"; @@ -64,6 +65,13 @@ export const runTrackV3 = async ({ body, }); + if (body.idempotency_key) { + await handleEventIdempotencyKey({ + ctx, + body, + }); + } + const redisIdempotencyKey = getTrackIdempotencyKey({ ctx }); const response: TrackResponseV3 = await runRedisTrackV3({ diff --git a/server/src/internal/customers/cache/fullSubject/sanitize/normalizeFromSchema.ts b/server/src/internal/customers/cache/fullSubject/sanitize/normalizeFromSchema.ts index 9ba0ea4ea..b4b9a8be3 100644 --- a/server/src/internal/customers/cache/fullSubject/sanitize/normalizeFromSchema.ts +++ b/server/src/internal/customers/cache/fullSubject/sanitize/normalizeFromSchema.ts @@ -87,10 +87,13 @@ const normalize = (schema: ZodSchema, data: unknown): unknown => { const unwrapped = unwrapSchema(schema); if (unwrapped instanceof z.ZodObject) { - if (!isPlainObject(data)) return data; + // Upstash Lua cjson collapses empty `{}` to `[]`; treat as empty object + // so nested defaults (e.g. ProductConfigSchema) still get applied. + const objectData = isEmptyArray(data) ? {} : data; + if (!isPlainObject(objectData)) return data; const shape = unwrapped._def.shape; - const normalized: Record = { ...data }; + const normalized: Record = { ...objectData }; for (const key of Object.keys(shape)) { normalized[key] = normalize(shape[key], normalized[key]); } diff --git a/server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.ts b/server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.ts index 6918deb1f..ec19f73f8 100644 --- a/server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.ts +++ b/server/src/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.ts @@ -13,8 +13,20 @@ export const sanitizeCachedFullSubject = ({ cachedFullSubject, }: { cachedFullSubject: CachedFullSubject; -}): CachedFullSubject => - normalizeFromSchema({ +}): CachedFullSubject => { + const normalized = normalizeFromSchema({ schema: CachedFullSubjectSchema, data: cachedFullSubject, }); + + // Safeguard for new product fields: Upstash Lua cjson collapses `{}` to `[]`, + // and pre-existing cache entries may not have these fields at all. + for (const product of normalized.products ?? []) { + const productAsRecord = product as { config?: unknown }; + if (!productAsRecord.config || Array.isArray(productAsRecord.config)) { + productAsRecord.config = {}; + } + } + + return normalized; +}; diff --git a/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.ts b/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.ts index f591aefae..e200f217d 100644 --- a/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.ts +++ b/server/src/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.ts @@ -196,6 +196,16 @@ export const getCachedFullCustomer = async ({ fullCustomer.send_email_receipts = false; } + // Safeguard for new product fields: Upstash Lua cjson collapses `{}` to `[]`, + // and pre-existing cache entries may not have these fields at all. + for (const cusProduct of fullCustomer.customer_products ?? []) { + if (!cusProduct.product) continue; + const product = cusProduct.product as { config?: unknown }; + if (!product.config || Array.isArray(product.config)) { + product.config = {}; + } + } + fullCustomer.invoices = deduplicateFullCustomerInvoices(fullCustomer); fullCustomer.customer_products = filterExpiredCustomerProducts(fullCustomer); diff --git a/server/src/utils/cacheUtils/normalizeFromSchema.ts b/server/src/utils/cacheUtils/normalizeFromSchema.ts index 0571cf41f..3b79a1dbf 100644 --- a/server/src/utils/cacheUtils/normalizeFromSchema.ts +++ b/server/src/utils/cacheUtils/normalizeFromSchema.ts @@ -109,13 +109,21 @@ export const normalizeFromSchema = ({ } if (type === "object") { - if (!data || typeof data !== "object" || Array.isArray(data)) { + // Upstash Lua cjson collapses empty `{}` to `[]`; treat as empty object + // so nested defaults still get applied on round-trip. + const objectData = + Array.isArray(data) && data.length === 0 ? {} : data; + if ( + !objectData || + typeof objectData !== "object" || + Array.isArray(objectData) + ) { return data as T; } const shape = (unwrapped as any)._def.shape; const normalized: Record = { - ...(data as Record), + ...(objectData as Record), }; for (const key in shape) { diff --git a/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts b/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts index a241bf280..a37601ef7 100644 --- a/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts +++ b/server/tests/integration/db/full-subject-cache/full-subject-cache-roundtrip.test.ts @@ -1,9 +1,10 @@ import { afterEach, describe, expect, test } from "bun:test"; -import { normalizedToFullSubject } from "@autumn/shared"; +import { AppEnv, normalizedToFullSubject } from "@autumn/shared"; import ctx from "@tests/utils/testInitUtils/createTestContext.js"; import chalk from "chalk"; import { getOrInitFullSubjectViewEpoch } from "@/internal/customers/cache/fullSubject/actions/invalidate/getOrInitFullSubjectViewEpoch.js"; import type { CachedFullSubject } from "@/internal/customers/cache/fullSubject/fullSubjectCacheModel.js"; +import { sanitizeCachedFullSubject } from "@/internal/customers/cache/fullSubject/sanitize/sanitizeCachedFullSubject.js"; import { buildFullSubjectKey, buildFullSubjectViewEpochKey, @@ -556,4 +557,81 @@ describe(`${chalk.yellowBright("fullSubject cache roundtrip")}`, () => { }, }); }); + + // ───────────────────────────────────────────────────────────────────────── + // Commit ce100dbf — cache sanitization regression + // + // Upstash's Lua cjson collapses empty `{}` to `[]` on round-trip. Without + // the sanitizer, a cached `product.config` written as `{}` comes back as + // `[]` and downstream consumers throw "Expected object, received array". + // ───────────────────────────────────────────────────────────────────────── + test(`${chalk.yellowBright("redis round-trip — cached product.config:[] is sanitized to {} on read")}`, async () => { + const planId = "plan_sanitize_redis_roundtrip"; + const cacheKey = `tests:cache-sanitize-roundtrip:${planId}:${Date.now()}`; + const malformedCacheEntry: unknown = { + subjectType: "customer", + customerId: "cus_sanitize_redis", + internalCustomerId: "cus_int_sanitize_redis", + _cachedAt: Date.now(), + subjectViewEpoch: 0, + meteredFeatures: [], + customerEntitlementIdsByFeatureId: {}, + customer: { + internal_id: "cus_int_sanitize_redis", + org_id: ctx.org.id, + env: AppEnv.Live, + created_at: 1, + }, + customer_products: [], + products: [ + { + id: planId, + internal_id: `ip_${planId}`, + name: planId, + group: `grp_${planId}`, + created_at: 1, + env: ctx.env, + org_id: ctx.org.id, + is_add_on: false, + is_default: false, + version: 1, + archived: false, + config: [], // <-- this is what Upstash hands back for `{}` + }, + ], + entitlements: [], + prices: [], + free_trials: [], + subscriptions: [], + invoices: [], + flags: {}, + }; + + try { + await ctx.redisV2.set(cacheKey, JSON.stringify(malformedCacheEntry)); + const raw = await ctx.redisV2.get(cacheKey); + expect(raw).toBeDefined(); + const parsed = JSON.parse(raw as string) as CachedFullSubject; + + // Pre-condition: the on-wire payload really does contain the bug + // shape we're guarding against. + expect( + Array.isArray( + (parsed.products[0] as unknown as { config: unknown }).config, + ), + ).toBe(true); + + const sanitized = sanitizeCachedFullSubject({ + cachedFullSubject: parsed, + }); + const product = sanitized.products[0] as unknown as { + config: { ignore_past_due: boolean }; + }; + + expect(Array.isArray(product.config)).toBe(false); + expect(product.config.ignore_past_due).toBe(false); + } finally { + await ctx.redisV2.del(cacheKey); + } + }); }); diff --git a/server/tests/unit/full-subject-cache/sanitizeCachedPayloads.test.ts b/server/tests/unit/full-subject-cache/sanitizeCachedPayloads.test.ts index fdd5f89cf..eeccf028a 100644 --- a/server/tests/unit/full-subject-cache/sanitizeCachedPayloads.test.ts +++ b/server/tests/unit/full-subject-cache/sanitizeCachedPayloads.test.ts @@ -2,14 +2,19 @@ import { describe, expect, test } from "bun:test"; import { type AggregatedFeatureBalanceSchema, AppEnv, + ProductSchema, type SubjectBalance, } from "@autumn/shared"; import { z } from "zod/v4"; -import type { CachedFullSubject } from "@/internal/customers/cache/fullSubject/fullSubjectCacheModel.js"; +import { + type CachedFullSubject, + CachedFullSubjectSchema, +} 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 { normalizeFromSchema as normalizeFromSchemaCacheUtils } from "@/utils/cacheUtils/normalizeFromSchema.js"; describe("normalizeFromSchema (core walker)", () => { test("fills undefined at nullable position with null", () => { @@ -501,3 +506,400 @@ describe("sanitizeCachedAggregatedFeatureBalance", () => { expect(result.prepaid_grant_from_options).toBe(0); }); }); + +// ═══════════════════════════════════════════════════════════════════════════════ +// Cache sanitization regression — commit ce100dbf +// +// Upstash's Lua cjson collapses empty `{}` to `[]`, and pre-existing cache +// entries that pre-date a new schema field will be missing it entirely. The +// sanitizer must: +// 1. coerce empty arrays back to objects when the schema says "object" +// (so nested defaults like `ProductConfigSchema.ignore_past_due=false` +// are re-applied), +// 2. defensively coerce `product.config` to `{}` when it's an array or +// missing (belt-and-suspenders for downstream consumers), +// 3. never throw "Expected object, received array" for any nested object +// shape regardless of what fields are added in the future. +// ═══════════════════════════════════════════════════════════════════════════════ + +const buildBaseCachedFullSubjectForSanitize = (): CachedFullSubject => + ({ + subjectType: "customer", + customerId: "cus_sanitize", + internalCustomerId: "cus_int_sanitize", + _cachedAt: Date.now(), + subjectViewEpoch: 0, + meteredFeatures: [], + customerEntitlementIdsByFeatureId: {}, + customer: { + internal_id: "cus_int_sanitize", + org_id: "org_sanitize", + env: AppEnv.Live, + created_at: 1, + }, + customer_products: [], + products: [], + entitlements: [], + prices: [], + free_trials: [], + subscriptions: [], + invoices: [], + flags: {}, + }) as unknown as CachedFullSubject; + +const buildBaseProductForSanitize = (planId = "plan_sanitize") => + ({ + // Just enough fields to look like a Product; the sanitizer only + // inspects `config` for this regression. + id: planId, + internal_id: `ip_${planId}`, + name: planId, + group: `grp_${planId}`, + created_at: 1, + env: AppEnv.Live, + org_id: "org_sanitize", + is_add_on: false, + is_default: false, + version: 1, + archived: false, + }) as Record; + +describe("sanitizeCachedFullSubject — product.config (commit ce100dbf)", () => { + test("coerces product.config: [] -> {} and re-applies ignore_past_due default", () => { + // Simulates what Upstash hands back: ignore_past_due defaulted to {}, + // cjson encoded it as [], so on read we get `config: []`. The walker + // must rebuild it as `{ ignore_past_due: false }`. + const malformed = buildBaseCachedFullSubjectForSanitize() as unknown as Record< + string, + unknown + >; + malformed.products = [ + { + ...buildBaseProductForSanitize("plan_sanitize_empty_array_config"), + config: [], + }, + ]; + + const result = sanitizeCachedFullSubject({ + cachedFullSubject: malformed as unknown as CachedFullSubject, + }); + + const product = result.products[0] as unknown as { + config: { ignore_past_due?: boolean }; + }; + + // Layer 1 (walker): config is an object, not an array. + expect(Array.isArray(product.config)).toBe(false); + expect(product.config).toBeDefined(); + expect(typeof product.config).toBe("object"); + + // Layer 2 (walker default-application): nested ZodDefault hydrated. + // If this fails the walker isn't recursing into the rebuilt object. + expect(product.config.ignore_past_due).toBe(false); + }); + + test("fills missing product.config entirely (pre-field-existed cache entries)", () => { + // Pre-existing cache entries written before `config` existed simply + // don't have the field — the belt-and-suspenders block in + // sanitizeCachedFullSubject must inject {}. + const malformed = buildBaseCachedFullSubjectForSanitize() as unknown as Record< + string, + unknown + >; + malformed.products = [buildBaseProductForSanitize("plan_sanitize_missing_config")]; + + const result = sanitizeCachedFullSubject({ + cachedFullSubject: malformed as unknown as CachedFullSubject, + }); + + const product = result.products[0] as unknown as { + config: Record; + }; + expect(product.config).toBeDefined(); + expect(Array.isArray(product.config)).toBe(false); + expect(typeof product.config).toBe("object"); + }); + + test("preserves explicit ignore_past_due=true (no over-correction)", () => { + // If Upstash cjson encoded `{ ignore_past_due: true }` faithfully (only + // fully-empty objects collapse), the value must survive untouched. + const malformed = buildBaseCachedFullSubjectForSanitize() as unknown as Record< + string, + unknown + >; + malformed.products = [ + { + ...buildBaseProductForSanitize("plan_sanitize_preserve_true"), + config: { ignore_past_due: true }, + }, + ]; + + const result = sanitizeCachedFullSubject({ + cachedFullSubject: malformed as unknown as CachedFullSubject, + }); + + const product = result.products[0] as unknown as { + config: { ignore_past_due: boolean }; + }; + expect(product.config.ignore_past_due).toBe(true); + }); +}); + +describe("normalizeFromSchema — empty-array-as-object regression (commit ce100dbf)", () => { + test("cacheUtils walker rebuilds ZodObject from empty array on nested products", () => { + // Direct exercise of the cacheUtils walker — used by + // getCachedFullCustomer. The structural fix from commit ce100dbf: + // nested `config: []` must become `config: {}` so downstream + // "Expected object, received array" errors stop firing. (This walker + // does NOT re-apply ZodDefault values; the belt-and-suspenders block + // in getCachedFullCustomer handles defaults for product.config.) + const TestSchema = z.object({ + products: z.array(ProductSchema), + }); + + const result = normalizeFromSchemaCacheUtils<{ + products: Array<{ config: unknown }>; + }>({ + schema: TestSchema as unknown as z.ZodTypeAny, + data: { + products: [ + { + ...buildBaseProductForSanitize("plan_sanitize_cacheutils"), + config: [], + }, + ], + }, + }); + + const product = result.products[0]!; + expect(Array.isArray(product.config)).toBe(false); + expect(typeof product.config).toBe("object"); + expect(product.config).not.toBeNull(); + }); + + test("full-subject walker rebuilds ZodObject from empty array on nested products", () => { + const result = normalizeFromSchema<{ + products: Array<{ config: { ignore_past_due: boolean } }>; + }>({ + schema: CachedFullSubjectSchema, + data: { + ...(buildBaseCachedFullSubjectForSanitize() as unknown as Record< + string, + unknown + >), + products: [ + { + ...buildBaseProductForSanitize("plan_sanitize_fullsubject"), + config: [], + }, + ], + }, + }); + + const product = result.products[0]!; + expect(Array.isArray(product.config)).toBe(false); + expect(product.config.ignore_past_due).toBe(false); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// Generic fuzz: mutate every field of a sample subject across multiple +// "Upstash-style" corruption modes and assert sanitization always returns a +// stable, downstream-consumable shape (no thrown errors, no array-shaped +// values at object positions). This is the regression net for any future +// field added to ProductSchema or peer schemas — new fields get fuzzed for +// free on the next test run. +// ───────────────────────────────────────────────────────────────────────────── + +type SanitizerCorruptionMode = + | "object_to_empty_array" + | "array_to_empty_object" + | "drop_field" + | "set_null" + | "set_undefined"; + +const ALL_SANITIZER_CORRUPTION_MODES: SanitizerCorruptionMode[] = [ + "object_to_empty_array", + "array_to_empty_object", + "drop_field", + "set_null", + "set_undefined", +]; + +const corruptValueForSanitizer = ( + value: unknown, + mode: SanitizerCorruptionMode, +): unknown => { + switch (mode) { + case "object_to_empty_array": + // Only objects can collapse to [] in Upstash cjson. + return value && typeof value === "object" && !Array.isArray(value) + ? [] + : value; + case "array_to_empty_object": + return Array.isArray(value) ? {} : value; + case "drop_field": + return undefined; + case "set_null": + return null; + case "set_undefined": + return undefined; + } +}; + +/** + * Walk `data` shape-blind and return a new structure where every leaf has had + * `corruptValueForSanitizer(_, mode)` applied. Recurses into objects and + * arrays so nested fields (like `products[*].config`) get hit. Generic — never + * references `config` or `ignore_past_due` directly. + */ +const corruptAllFieldsForSanitizer = ( + data: unknown, + mode: SanitizerCorruptionMode, + depth = 0, +): unknown => { + if (depth > 4) return data; // safety against pathological cycles + if (Array.isArray(data)) { + return data.map((item) => + corruptAllFieldsForSanitizer(item, mode, depth + 1), + ); + } + if (data && typeof data === "object") { + const out: Record = {}; + for (const [key, value] of Object.entries(data)) { + const recursed = corruptAllFieldsForSanitizer(value, mode, depth + 1); + out[key] = corruptValueForSanitizer(recursed, mode); + } + return out; + } + return data; +}; + +const assertNoArrayShapedObjects = ( + value: unknown, + schema: z.ZodTypeAny, + path = "$", +): void => { + // Unwrap optional/nullable/default chains to the inner shape-bearing schema. + let unwrapped: z.ZodTypeAny = schema; + while ( + unwrapped instanceof z.ZodOptional || + unwrapped instanceof z.ZodNullable || + unwrapped instanceof z.ZodDefault + ) { + unwrapped = (unwrapped as unknown as { _def: { innerType: z.ZodTypeAny } }) + ._def.innerType; + } + + if (unwrapped instanceof z.ZodObject) { + // Critical assertion: the object position must NOT carry an Array + // payload — that's exactly the "Expected object, received array" + // breakage commit ce100dbf is fixing. + if (Array.isArray(value)) { + throw new Error( + `Sanitizer left an array at object position ${path} (caller would see "Expected object, received array")`, + ); + } + if (value && typeof value === "object") { + const shape = (unwrapped as unknown as { _def: { shape: Record } })._def.shape; + for (const [key, childSchema] of Object.entries(shape)) { + assertNoArrayShapedObjects( + (value as Record)[key], + childSchema, + `${path}.${key}`, + ); + } + } + return; + } + + if (unwrapped instanceof z.ZodArray) { + if (Array.isArray(value)) { + const element = (unwrapped as unknown as { _def: { element: z.ZodTypeAny } })._def.element; + value.forEach((item, idx) => + assertNoArrayShapedObjects(item, element, `${path}[${idx}]`), + ); + } + return; + } +}; + +describe("sanitizeCachedFullSubject — generic fuzz (regression net for new fields)", () => { + test("every Upstash corruption mode is recovered without leaving array payloads at object positions", () => { + // Build a "fully-populated" subject so the walker has every nested + // shape to traverse. New fields on any nested schema get fuzzed for + // free on the next test run. + const populated = buildBaseCachedFullSubjectForSanitize() as unknown as Record< + string, + unknown + >; + populated.products = [ + { + ...buildBaseProductForSanitize("plan_sanitize_fuzz_1"), + config: { ignore_past_due: true }, + }, + { + ...buildBaseProductForSanitize("plan_sanitize_fuzz_2"), + config: { ignore_past_due: false }, + }, + ]; + + for (const mode of ALL_SANITIZER_CORRUPTION_MODES) { + // We corrupt only the *interior* of products[i] — the field-shape + // bug commit ce100dbf addresses. The top-level container shape + // (products is an array, the subject is an object, etc.) is what + // the cache layer guarantees on write, so corrupting it is + // outside the bug class. Mutating each product's interior is + // sufficient to fuzz every nested field of ProductSchema — + // including `config` and any future field added to it. + const corruptedProducts = ( + populated.products as Array> + ).map((p, i) => { + const corruptedProduct = corruptAllFieldsForSanitizer( + p, + mode, + ) as Record; + // Restore the structural identifier fields so the sanitizer + // has a stable record to work on. (The bug isn't about + // missing IDs.) + return { + ...corruptedProduct, + id: `plan_sanitize_fuzz_${i + 1}_${mode}`, + internal_id: `ip_plan_sanitize_fuzz_${i + 1}_${mode}`, + }; + }); + + const corrupted: Record = { + ...populated, + products: corruptedProducts, + }; + + let sanitized: CachedFullSubject; + expect(() => { + sanitized = sanitizeCachedFullSubject({ + cachedFullSubject: corrupted as unknown as CachedFullSubject, + }); + }).not.toThrow(); + + // Walk the full schema and prove no object position is left holding + // an array payload (the exact symptom of the bug). + expect(() => + assertNoArrayShapedObjects( + sanitized!, + CachedFullSubjectSchema as unknown as z.ZodTypeAny, + ), + ).not.toThrow(); + + // Spot-check the specific known target of commit ce100dbf: + // `product.config` is always an object after sanitization, + // regardless of corruption mode. + for (const product of (sanitized!.products ?? []) as Array<{ + config: unknown; + }>) { + expect(Array.isArray(product.config)).toBe(false); + expect(typeof product.config).toBe("object"); + expect(product.config).not.toBeNull(); + } + } + }); +});