diff --git a/server/src/internal/balances/handlers/handleTrack.ts b/server/src/internal/balances/handlers/handleTrack.ts index dfb501e63..8281c5a33 100644 --- a/server/src/internal/balances/handlers/handleTrack.ts +++ b/server/src/internal/balances/handlers/handleTrack.ts @@ -4,12 +4,13 @@ import { TrackParamsSchema, TrackQuerySchema, } from "@autumn/shared"; +import { shouldUseRedis } from "@/external/redis/initRedis.js"; import { createRoute } from "@/honoMiddlewares/routeHandler.js"; import { runTrackV2 } from "@/internal/balances/track/runTrackV2.js"; -import { - getTrackEventNameDeductions, - getTrackFeatureDeductions, -} from "@/internal/balances/track/utils/getFeatureDeductions.js"; +import { getTrackFeatureDeductionsForBody } from "@/internal/balances/track/utils/getFeatureDeductions.js"; +import { getQueuedTrackResponse } from "@/internal/balances/track/utils/getQueuedTrackResponse.js"; +import { JobName } from "@/queue/JobName.js"; +import { addTaskToQueue } from "@/queue/queueUtils.js"; export const handleTrack = createRoute({ query: TrackQuerySchema, @@ -21,27 +22,35 @@ export const handleTrack = createRoute({ handler: async (c) => { const body = c.req.valid("json"); const ctx = c.get("ctx"); + const featureDeductions = getTrackFeatureDeductionsForBody({ ctx, body }); - // Build feature deductions - const featureDeductions = body.feature_id - ? getTrackFeatureDeductions({ - ctx, - featureId: body.feature_id, - lock: body.lock, - value: body.value, - }) - : getTrackEventNameDeductions({ - ctx, - eventName: body.event_name!, - value: body.value, - }); + if (!shouldUseRedis()) { + const queueUrl = process.env.TRACK_SQS_QUEUE_URL; + if (!queueUrl) { + throw new Error("TRACK_SQS_QUEUE_URL is not configured"); + } - return c.json( - await runTrackV2({ - ctx, - body, - featureDeductions, - }), - ); + await addTaskToQueue({ + jobName: JobName.Track, + queueUrl, + messageGroupId: `${ctx.org.id}:${ctx.env}:${body.customer_id}`, + messageDeduplicationId: body.idempotency_key, + payload: { + orgId: ctx.org.id, + env: ctx.env, + apiVersion: ctx.apiVersion.value, + body, + }, + }); + + return c.json( + getQueuedTrackResponse({ + ctx, + body, + }), + ); + } + + return c.json(await runTrackV2({ ctx, body, featureDeductions })); }, }); diff --git a/server/src/internal/balances/track/utils/getFeatureDeductions.ts b/server/src/internal/balances/track/utils/getFeatureDeductions.ts index e7463732f..2fbddacce 100644 --- a/server/src/internal/balances/track/utils/getFeatureDeductions.ts +++ b/server/src/internal/balances/track/utils/getFeatureDeductions.ts @@ -2,6 +2,7 @@ import { FeatureNotFoundError, type LockParams, RecaseError, + type TrackParams, } from "@autumn/shared"; import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js"; import type { FeatureDeduction } from "../../utils/types/featureDeduction.js"; @@ -72,3 +73,23 @@ export const getTrackEventNameDeductions = ({ return featureDeductions; }; + +export const getTrackFeatureDeductionsForBody = ({ + ctx, + body, +}: { + ctx: AutumnContext; + body: TrackParams; +}) => + body.feature_id + ? getTrackFeatureDeductions({ + ctx, + featureId: body.feature_id, + lock: body.lock, + value: body.value, + }) + : getTrackEventNameDeductions({ + ctx, + eventName: body.event_name!, + value: body.value, + }); diff --git a/server/src/internal/balances/track/utils/getQueuedTrackResponse.ts b/server/src/internal/balances/track/utils/getQueuedTrackResponse.ts new file mode 100644 index 000000000..e223628dd --- /dev/null +++ b/server/src/internal/balances/track/utils/getQueuedTrackResponse.ts @@ -0,0 +1,36 @@ +import { + AffectedResource, + type ApiVersion, + ApiVersionClass, + applyResponseVersionChanges, + type TrackParams, + type TrackResponseV3, +} from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; + +export const getQueuedTrackResponse = ({ + ctx, + body, + apiVersion, +}: { + ctx: AutumnContext; + body: TrackParams; + apiVersion?: ApiVersion; +}) => + applyResponseVersionChanges({ + input: { + customer_id: body.customer_id, + entity_id: body.entity_id, + event_name: body.event_name, + value: body.value ?? 1, + balance: null, + }, + targetVersion: apiVersion + ? new ApiVersionClass(apiVersion) + : ctx.apiVersion, + resource: AffectedResource.Track, + legacyData: { + feature_id: body.feature_id || body.event_name, + }, + ctx, + }); diff --git a/server/src/queue/JobName.ts b/server/src/queue/JobName.ts index 877d86672..245157e29 100644 --- a/server/src/queue/JobName.ts +++ b/server/src/queue/JobName.ts @@ -19,6 +19,7 @@ export enum JobName { SyncBalanceBatchV2 = "sync-balance-batch-v2", SyncBalanceBatchV3 = "sync-balance-batch-v3", InsertEventBatch = "insert-event-batch", + Track = "track", ClearCreditSystemCustomerCache = "clear-credit-system-customer-cache", diff --git a/server/src/queue/queueUtils.ts b/server/src/queue/queueUtils.ts index e51091c8c..0b5dddd1e 100644 --- a/server/src/queue/queueUtils.ts +++ b/server/src/queue/queueUtils.ts @@ -1,4 +1,10 @@ -import type { AppEnv, EventInsert, Price } from "@autumn/shared"; +import type { + ApiVersion, + AppEnv, + EventInsert, + Price, + TrackParams, +} from "@autumn/shared"; import { SendMessageCommand } from "@aws-sdk/client-sqs"; import { generateId } from "@server/utils/genUtils"; import { isHatchetEnabled } from "@/external/hatchet/initHatchet.js"; @@ -8,6 +14,7 @@ import { } from "@/internal/billing/v2/workflows/verifyCacheConsistency/verifyCacheConsistency.js"; import type { ClearCreditSystemCachePayload } from "@/internal/features/featureActions/runClearCreditSystemCacheTask.js"; import type { GenerateFeatureDisplayPayload } from "@/internal/features/workflows/generateFeatureDisplay.js"; +import { getSqsClient } from "./initSqs.js"; import { JobName } from "./JobName.js"; import type { BatchResetCusEntsPayload, @@ -45,6 +52,12 @@ export interface Payloads { [JobName.InsertEventBatch]: { events: EventInsert[]; }; + [JobName.Track]: { + orgId: string; + env: AppEnv; + apiVersion: ApiVersion; + body: TrackParams; + }; [JobName.ClearCreditSystemCustomerCache]: ClearCreditSystemCachePayload; [JobName.GenerateFeatureDisplay]: GenerateFeatureDisplayPayload; [JobName.SendProductsUpdated]: SendProductsUpdatedPayload; @@ -81,6 +94,7 @@ export const addTaskToQueue = async ({ messageDeduplicationId, generateDeduplicationId, delayMs, + queueUrl, }: { jobName: T; payload: Payloads[T]; @@ -88,13 +102,15 @@ export const addTaskToQueue = async ({ messageDeduplicationId?: string; generateDeduplicationId?: boolean; delayMs?: number; + queueUrl?: string; }) => { - if (process.env.SQS_QUEUE_URL) { - const { getSqsClient, QUEUE_URL } = await import("./initSqs.js"); + const resolvedQueueUrl = queueUrl || process.env.SQS_QUEUE_URL; + + if (resolvedQueueUrl) { const sqsClient = getSqsClient(); // SQS implementation - const isFifoQueue = QUEUE_URL.endsWith(".fifo"); + const isFifoQueue = resolvedQueueUrl.endsWith(".fifo"); const messageId = generateDeduplicationId === false ? undefined : generateId("job"); const message = { @@ -113,7 +129,7 @@ export const addTaskToQueue = async ({ Bun.hash(messageId ?? generateId("dedup")).toString(); const command = new SendMessageCommand({ - QueueUrl: QUEUE_URL, + QueueUrl: resolvedQueueUrl, MessageBody: JSON.stringify(message), ...(delaySeconds && { DelaySeconds: delaySeconds }), // FIFO queues require MessageGroupId. Content-based deduplication uses the body. diff --git a/server/tests/unit/balances/track/handle-track-queue-fallback.test.ts b/server/tests/unit/balances/track/handle-track-queue-fallback.test.ts new file mode 100644 index 000000000..a852d4b56 --- /dev/null +++ b/server/tests/unit/balances/track/handle-track-queue-fallback.test.ts @@ -0,0 +1,146 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { ApiVersion, ApiVersionClass, AppEnv } from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; + +const mockState = { + shouldUseRedis: true, + queueCalls: [] as Record[], + runTrackCalls: [] as Record[], +}; + +mock.module("@/external/redis/initRedis.js", () => ({ + shouldUseRedis: () => mockState.shouldUseRedis, +})); + +mock.module("@/queue/queueUtils.js", () => ({ + addTaskToQueue: async (args: Record) => { + mockState.queueCalls.push(args); + }, +})); + +mock.module("@/internal/balances/track/runTrackV2.js", () => ({ + runTrackV2: async (args: Record) => { + mockState.runTrackCalls.push(args); + return { ok: true }; + }, +})); + +import { handleTrack } from "@/internal/balances/handlers/handleTrack.js"; +import { JobName } from "@/queue/JobName.js"; + +const wrappedHandler = handleTrack[handleTrack.length - 1] as unknown as ( + c: ReturnType, +) => Promise; + +const makeCtx = ({ apiVersion }: { apiVersion: ApiVersion }) => + ({ + org: { + id: "org_123", + }, + env: AppEnv.Sandbox, + apiVersion: new ApiVersionClass(apiVersion), + features: [ + { + id: "messages", + event_names: [], + }, + ], + }) as unknown as AutumnContext; + +const makeContext = ({ + body, + apiVersion, +}: { + body: Record; + apiVersion: ApiVersion; +}) => { + const store = new Map([["ctx", makeCtx({ apiVersion })]]); + + return { + req: { + valid: (target: string) => { + if (target === "json") return body; + return {}; + }, + }, + get: (key: string) => store.get(key), + set: (key: string, value: unknown) => { + store.set(key, value); + }, + json: (data: unknown, status = 200) => + new Response(JSON.stringify(data), { + status, + headers: { + "content-type": "application/json", + }, + }), + }; +}; + +describe("handleTrack queue fallback", () => { + const originalTrackQueueUrl = process.env.TRACK_SQS_QUEUE_URL; + + beforeEach(() => { + mockState.shouldUseRedis = false; + mockState.queueCalls = []; + mockState.runTrackCalls = []; + }); + + afterEach(() => { + process.env.TRACK_SQS_QUEUE_URL = originalTrackQueueUrl; + }); + + test("queues track and returns the legacy success shape when Redis is unavailable", async () => { + process.env.TRACK_SQS_QUEUE_URL = + "https://sqs.eu-west-1.amazonaws.com/123456789012/track-dev.fifo"; + + const response = await wrappedHandler( + makeContext({ + apiVersion: ApiVersion.V1_Beta, + body: { + customer_id: "cus_123", + feature_id: "messages", + value: 2, + idempotency_key: "idem_123", + }, + }), + ); + + expect(mockState.queueCalls).toHaveLength(1); + expect(mockState.runTrackCalls).toHaveLength(0); + expect(mockState.queueCalls[0]).toMatchObject({ + jobName: JobName.Track, + queueUrl: + "https://sqs.eu-west-1.amazonaws.com/123456789012/track-dev.fifo", + messageGroupId: "org_123:sandbox:cus_123", + messageDeduplicationId: "idem_123", + }); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + id: "placeholder", + code: "event_received", + customer_id: "cus_123", + feature_id: "messages", + }); + }); + + test("fails clearly when Redis is unavailable and TRACK_SQS_QUEUE_URL is unset", async () => { + delete process.env.TRACK_SQS_QUEUE_URL; + + await expect( + wrappedHandler( + makeContext({ + apiVersion: ApiVersion.V2_1, + body: { + customer_id: "cus_123", + feature_id: "messages", + }, + }), + ), + ).rejects.toThrow("TRACK_SQS_QUEUE_URL is not configured"); + + expect(mockState.queueCalls).toHaveLength(0); + expect(mockState.runTrackCalls).toHaveLength(0); + }); +}); diff --git a/server/tests/unit/customers/full-customer-cache-redis-gating.test.ts b/server/tests/unit/customers/full-customer-cache-redis-gating.test.ts new file mode 100644 index 000000000..ffa1578c8 --- /dev/null +++ b/server/tests/unit/customers/full-customer-cache-redis-gating.test.ts @@ -0,0 +1,169 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import type { FullCustomer } from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; + +const mockState = { + shouldUseRedis: true, + cached: undefined as FullCustomer | undefined, + dbResult: undefined as FullCustomer | undefined, + hydrated: undefined as FullCustomer | undefined, + created: undefined as FullCustomer | undefined, + cacheReads: 0, + cacheWrites: 0, + dbCalls: 0, + createCalls: 0, + updateDetailsCalls: 0, +}; + +mock.module("@/external/redis/initRedis.js", () => ({ + shouldUseRedis: () => mockState.shouldUseRedis, +})); + +mock.module("@/internal/customers/CusService.js", () => ({ + CusService: { + getFull: async () => { + mockState.dbCalls++; + return mockState.dbResult; + }, + }, +})); + +mock.module( + "@/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.js", + () => ({ + getCachedFullCustomer: async () => { + mockState.cacheReads++; + return mockState.cached; + }, + }), +); + +mock.module( + "@/internal/customers/cusUtils/fullCustomerCacheUtils/setCachedFullCustomer.js", + () => ({ + setCachedFullCustomer: async () => { + mockState.cacheWrites++; + return "OK"; + }, + }), +); + +mock.module("@/internal/customers/cusUtils/getFullCustomerSchedule.js", () => ({ + hydrateFullCustomerSchedule: async ({ + fullCustomer, + }: { + fullCustomer: unknown; + }) => mockState.hydrated ?? fullCustomer, +})); + +mock.module("@/internal/customers/actions/index.js", () => ({ + customerActions: { + createWithDefaults: async () => { + mockState.createCalls++; + return mockState.created; + }, + }, +})); + +mock.module( + "@/internal/entities/handlers/handleCreateEntity/autoCreateEntity.js", + () => ({ + autoCreateEntity: async () => null, + }), +); + +mock.module("@/internal/customers/cusUtils/cusUtils.js", () => ({ + updateCustomerDetails: async () => { + mockState.updateDetailsCalls++; + }, +})); + +import { getOrCreateCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/getOrCreateCachedFullCustomer.js"; +import { getOrSetCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/getOrSetCachedFullCustomer.js"; + +const makeCtx = () => + ({ + skipCache: false, + logger: { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: () => {}, + }, + }) as AutumnContext; + +const makeFullCustomer = (id: string) => + ({ + id, + internal_id: `int_${id}`, + org_id: "test_org", + created_at: Date.now(), + env: "sandbox", + name: null, + email: null, + metadata: {}, + processor: null, + send_email_receipts: false, + customer_products: [], + extra_customer_entitlements: [], + entities: [], + }) as FullCustomer; + +describe("full customer cache Redis gating", () => { + beforeEach(() => { + mockState.shouldUseRedis = true; + mockState.cached = undefined; + mockState.dbResult = undefined; + mockState.hydrated = undefined; + mockState.created = undefined; + mockState.cacheReads = 0; + mockState.cacheWrites = 0; + mockState.dbCalls = 0; + mockState.createCalls = 0; + mockState.updateDetailsCalls = 0; + }); + + test("getOrSetCachedFullCustomer skips Redis read/write when shouldUseRedis is false", async () => { + const dbCustomer = makeFullCustomer("db-customer"); + const hydratedCustomer = makeFullCustomer("hydrated-customer"); + + mockState.shouldUseRedis = false; + mockState.cached = makeFullCustomer("cached-customer"); + mockState.dbResult = dbCustomer; + mockState.hydrated = hydratedCustomer; + + const result = await getOrSetCachedFullCustomer({ + ctx: makeCtx(), + customerId: "db-customer", + source: "unit-test", + }); + + expect(result).toBe(hydratedCustomer); + expect(mockState.cacheReads).toBe(0); + expect(mockState.cacheWrites).toBe(0); + expect(mockState.dbCalls).toBe(1); + }); + + test("getOrCreateCachedFullCustomer still creates when Redis is unavailable and DB misses", async () => { + const createdCustomer = makeFullCustomer("created-customer"); + + mockState.shouldUseRedis = false; + mockState.dbResult = undefined; + mockState.created = createdCustomer; + + const result = await getOrCreateCachedFullCustomer({ + ctx: makeCtx(), + params: { + customer_id: "created-customer", + feature_id: "messages", + }, + source: "unit-test", + }); + + expect(result).toBe(createdCustomer); + expect(mockState.cacheReads).toBe(0); + expect(mockState.cacheWrites).toBe(0); + expect(mockState.dbCalls).toBe(1); + expect(mockState.createCalls).toBe(1); + }); +}); diff --git a/server/tests/unit/queue/queue-utils.test.ts b/server/tests/unit/queue/queue-utils.test.ts new file mode 100644 index 000000000..a27f4aa29 --- /dev/null +++ b/server/tests/unit/queue/queue-utils.test.ts @@ -0,0 +1,60 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { ApiVersion, AppEnv } from "@autumn/shared"; + +const mockState = { + commands: [] as Record[], +}; + +mock.module("@/queue/initSqs.js", () => ({ + QUEUE_URL: "https://sqs.eu-west-1.amazonaws.com/123456789012/primary.fifo", + getSqsClient: () => ({ + send: async (command: { input: Record }) => { + mockState.commands.push(command.input); + }, + }), +})); + +import { JobName } from "@/queue/JobName.js"; +import { addTaskToQueue } from "@/queue/queueUtils.js"; + +describe("addTaskToQueue queue override", () => { + const originalSqsQueueUrl = process.env.SQS_QUEUE_URL; + const originalQueueUrl = process.env.QUEUE_URL; + + beforeEach(() => { + mockState.commands = []; + delete process.env.SQS_QUEUE_URL; + delete process.env.QUEUE_URL; + }); + + afterEach(() => { + process.env.SQS_QUEUE_URL = originalSqsQueueUrl; + process.env.QUEUE_URL = originalQueueUrl; + }); + + test("uses the provided SQS queueUrl override", async () => { + await addTaskToQueue({ + jobName: JobName.Track, + queueUrl: + "https://sqs.eu-west-1.amazonaws.com/123456789012/track-dev.fifo", + messageGroupId: "org:sandbox:cus_123", + messageDeduplicationId: "idem_123", + payload: { + orgId: "org_123", + env: AppEnv.Sandbox, + apiVersion: ApiVersion.V2_1, + body: { + customer_id: "cus_123", + feature_id: "messages", + }, + }, + }); + + expect(mockState.commands).toHaveLength(1); + expect(mockState.commands[0]?.QueueUrl).toBe( + "https://sqs.eu-west-1.amazonaws.com/123456789012/track-dev.fifo", + ); + expect(mockState.commands[0]?.MessageGroupId).toBe("org:sandbox:cus_123"); + expect(mockState.commands[0]?.MessageDeduplicationId).toBe("idem_123"); + }); +});