From 3ee3beb4207d2644de5c71a6beb7ee4b0e78f906 Mon Sep 17 00:00:00 2001 From: John Yeo Date: Tue, 30 Dec 2025 17:58:54 -0800 Subject: [PATCH] wip --- server/src/check.ts | 414 ------------------ .../stripe/handleStripeWebhookEvent.ts | 90 ++-- .../honoMiddlewares/refreshCacheMiddleware.ts | 22 + .../deleteCachedApiCustomer.ts | 9 +- .../apiCusCacheUtils/setCachedApiCustomer.ts | 4 +- .../apiCusCacheUtils/setCachedApiInvoices.ts | 1 - server/src/queue/JobName.ts | 1 + server/src/queue/initWorkers.ts | 19 +- .../src/queue/jobs/verifyCacheConsistency.ts | 95 ++++ server/src/queue/queueUtils.ts | 18 +- server/tests/_temp/temp.test.ts | 35 +- server/tests/attach/basic/basic2.test.ts | 1 + server/tests/utils/compare.ts | 3 - .../changes/V1.1_FeaturesArrayToObject.ts | 2 +- 14 files changed, 231 insertions(+), 483 deletions(-) delete mode 100644 server/src/check.ts create mode 100644 server/src/queue/jobs/verifyCacheConsistency.ts diff --git a/server/src/check.ts b/server/src/check.ts deleted file mode 100644 index 74caa4c22..000000000 --- a/server/src/check.ts +++ /dev/null @@ -1,414 +0,0 @@ -// import { config } from "dotenv"; - -// config(); - -// import assert from "node:assert"; -// import { -// AppEnv, -// CusProductStatus, -// cusProductToPrices, -// type Entity, -// type FullCusProduct, -// type FullCustomer, -// type Organization, -// } from "@autumn/shared"; - -// import type Stripe from "stripe"; -// import { initDrizzle } from "@/db/initDrizzle.js"; -// import { createStripeCli } from "@/external/connect/createStripeCli.js"; -// import { getStripeSchedules } from "@/external/stripe/stripeSubUtils.js"; -// import { createSupabaseClient } from "@/external/supabaseUtils.js"; -// import { CusService } from "@/internal/customers/CusService.js"; -// import { OrgService } from "@/internal/orgs/OrgService.js"; -// import { isFreeProduct, isOneOff } from "@/internal/products/productUtils.js"; -// import { notNullish } from "@/utils/genUtils.js"; -// import { -// getAllEntities, -// getAllFullCustomers, -// } from "@/utils/scriptUtils/getAll/getAllAutumnCustomers.js"; -// import { -// getAllStripeSchedules, -// getAllStripeSubscriptions, -// } from "@/utils/scriptUtils/getAll/getAllStripeSubs.js"; -// import { EntityService } from "./internal/api/entities/EntityService.js"; -// import { getRelatedCusPrice } from "./internal/customers/cusProducts/cusEnts/cusEntUtils.js"; -// import { checkCusSubCorrect } from "./utils/checkUtils/checkCustomerCorrect.js"; - -// const { db } = initDrizzle({ maxConnections: 5 }); - -// let orgSlugs = process.env.ORG_SLUGS!.split(","); -// const skipEmails = process.env.SKIP_EMAILS!.split(","); -// const skipIds = [ -// "cus_2tXCCwC6iyiftgA6ndSo1Ubb2dx", -// "DxG668K7uDd0Vahk54YWjvCGVgf2", -// ]; - -// orgSlugs = ["athenahq"]; -// // let customerId = null; -// const customerId = null; - -// const getSingleCustomer = async ({ -// stripeCli, -// customerId, -// orgId, -// env, -// }: { -// stripeCli: Stripe; -// customerId: string; -// orgId: string; -// env: AppEnv; -// }) => { -// const customers = [ -// await CusService.getFull({ -// db, -// idOrInternalId: customerId, -// orgId, -// env, -// }), -// ]; - -// const stripeCusId = customers[0].processor?.id; -// const stripeSubs = stripeCusId -// ? ( -// await stripeCli.subscriptions.list({ -// customer: stripeCusId, -// expand: ["data.discounts.coupon"], -// }) -// ).data -// : []; - -// // const stripeSubs = await getStripeSubs({ -// // stripeCli, -// // subIds: customers[0].customer_products.flatMap( -// // (cp) => cp.subscription_ids || [] -// // ), -// // }); - -// let scheduleIds = customers[0].customer_products.flatMap( -// (cp) => cp.scheduled_ids || [], -// ); - -// scheduleIds = Array.from(new Set(scheduleIds)); - -// const stripeSchedules = await getStripeSchedules({ -// stripeCli, -// scheduleIds, -// }); - -// const entities = await EntityService.list({ -// db, -// internalCustomerId: customers[0].internal_id, -// }); - -// return { customers, stripeSubs, stripeSchedules, entities }; -// }; - -// const checkCustomerCorrect = async ({ -// fullCus, -// subs, -// schedules, -// org, -// entities, -// }: { -// fullCus: FullCustomer; -// subs: Stripe.Subscription[]; -// schedules: Stripe.SubscriptionSchedule[]; -// org: Organization; -// entities: Entity[]; -// }) => { -// if (skipIds.includes(fullCus.internal_id!)) return; - -// if (skipEmails.some((skipEmail) => skipEmail === fullCus.email)) { -// return; -// } - -// fullCus.entities = entities.filter( -// (entity) => entity.internal_customer_id === fullCus.internal_id, -// ); - -// // console.log(`Checking ${fullCus.email} (${fullCus.id})`); -// const cusProducts = fullCus.customer_products; - -// await checkCusSubCorrect({ -// db, -// fullCus, -// subs, -// schedules, -// org, -// env: AppEnv.Live, -// }); - -// for (const cusProduct of cusProducts) { -// if (!cusProduct.subscription_ids) continue; - -// if (cusProduct.status === CusProductStatus.Scheduled) { -// // Check if there's a main product elsewhere -// const mainCusProd = cusProducts.find( -// (cp: FullCusProduct) => -// cp.product.group === cusProduct.product.group && -// cp.id !== cusProduct.id && -// cp.status !== CusProductStatus.Scheduled && -// (cusProduct.internal_entity_id -// ? cusProduct.internal_entity_id === cp.internal_entity_id -// : true), -// ); - -// assert( -// mainCusProd, -// `Found scheduled cus product with no main product (${cusProduct.product.name})`, -// ); -// } - -// if ( -// !cusProduct.product.is_add_on && -// cusProduct.status !== CusProductStatus.Scheduled -// ) { -// const group = cusProduct.product.group; -// const otherCusProd = cusProducts.find( -// (cp: FullCusProduct) => -// cp.product.group === group && -// cp.id !== cusProduct.id && -// !cp.product.is_add_on && -// cp.status !== CusProductStatus.Scheduled && -// cp.internal_entity_id === cusProduct.internal_entity_id, -// ); - -// assert( -// !otherCusProd, -// `found two cus products from the same group: ${otherCusProd?.product.name} and ${cusProduct.product.name}`, -// ); -// } - -// const stripeSubs = subs.filter((sub: any) => -// cusProduct.subscription_ids!.some((id: string) => id === sub.id), -// ); - -// assert( -// stripeSubs.length === cusProduct.subscription_ids!.length, -// "number of stripe subs should be the same as number of subscription ids", -// ); - -// // let subItems = stripeSubs.flatMap((sub: any) => sub.items.data); - -// const prices = cusProductToPrices({ cusProduct }); - -// if ( -// isOneOff(prices) || -// isFreeProduct(prices) || -// cusProduct.status === CusProductStatus.Scheduled -// ) { -// continue; -// } - -// for (const cusEnt of cusProduct.customer_entitlements) { -// const cusPrice = getRelatedCusPrice(cusEnt, cusProduct.customer_prices); - -// if (cusEnt.usage_allowed && !cusPrice) { -// assert.fail( -// `Feature ${cusEnt.feature_id} has usage allowed but no related cus price`, -// ); -// } -// } -// } - -// // Other checks to perform -// }; - -// const checkCustomerHandleError = async ({ -// fullCus, -// subs, -// org, -// schedules, -// entities, -// }: { -// fullCus: FullCustomer; -// subs: Stripe.Subscription[]; -// org: Organization; -// schedules: Stripe.SubscriptionSchedule[]; -// entities: Entity[]; -// }) => { -// try { -// await checkCustomerCorrect({ -// fullCus, -// subs, -// org, -// schedules, -// entities, -// }); - -// return undefined; -// } catch (error: any) { -// return { -// id: fullCus.id, -// name: fullCus.name, -// email: fullCus.email, -// error: error.message, -// }; -// } -// }; - -// export const check = async () => { -// const env = AppEnv.Live; -// const sb = createSupabaseClient(); - -// const today = new Date().toISOString().slice(0, 16); - -// for (const slug of orgSlugs) { -// const org = await OrgService.getBySlug({ -// db, -// slug, -// }); - -// if (!org) { -// console.log(`Org ${slug} not found`); -// continue; -// } - -// const fileName = `errors/${today}-${org.slug}.json`; - -// const stripeCli = createStripeCli({ -// org, -// env, -// }); - -// console.log("--------------------------------"); -// console.log(`Running error check for ${org.name}`); - -// let customers: FullCustomer[] = []; -// let stripeSubs: Stripe.Subscription[] = []; -// let stripeSchedules: Stripe.SubscriptionSchedule[] = []; -// let entities: Entity[] = []; - -// if (customerId) { -// const res = await getSingleCustomer({ -// stripeCli, -// customerId, -// orgId: org.id, -// env, -// }); - -// customers = res.customers; -// stripeSubs = res.stripeSubs; -// entities = res.entities; -// } else { -// const [customersRes, stripeSubsRes, stripeSchedulesRes, entitiesRes] = -// await Promise.all([ -// getAllFullCustomers({ -// db, -// orgId: org.id, -// env, -// }), -// getAllStripeSubscriptions({ -// stripeCli, -// waitForSeconds: 1, -// }), -// getAllStripeSchedules({ -// stripeCli, -// waitForSeconds: 1, -// }), -// getAllEntities({ -// db, -// orgId: org.id, -// env, -// }), -// ]); - -// customers = customersRes; -// stripeSubs = stripeSubsRes.subscriptions; -// stripeSchedules = stripeSchedulesRes.schedules; -// entities = entitiesRes; -// } - -// const batchSize = 1; -// const allErrors = []; -// for (let i = 0; i < customers.length; i += batchSize) { -// const batch = customers.slice(i, i + batchSize); - -// const batchCheck: any = []; -// for (const customer of batch) { -// batchCheck.push( -// checkCustomerHandleError({ -// fullCus: customer, -// subs: stripeSubs, -// schedules: stripeSchedules, -// org, -// entities, -// }), -// ); -// } - -// let results = await Promise.all(batchCheck); -// results = results.filter(notNullish); -// allErrors.push(...results); -// } - -// console.log(`Found ${allErrors.length} errors`); - -// if (allErrors.length > 0 && customers.length > 1) { -// await sb.storage -// .from("autumn") -// .upload(fileName, JSON.stringify(allErrors, null, 2)); - -// if (allErrors.length > 0) { -// const slackBody = { -// text: `Error check for ${org.name}`, -// blocks: [ -// { -// type: "section", -// text: { -// type: "mrkdwn", -// text: `*Error check for ${org.name}*: found ${allErrors.length} errors\nSee results at ${process.env.SUPABASE_URL}/storage/v1/object/public/autumn/${fileName}`, -// }, -// }, -// ], -// }; - -// await fetch(process.env.SLACK_WEBHOOK_URL!, { -// method: "POST", -// body: JSON.stringify(slackBody), -// }); -// } -// } else { -// console.log(allErrors); -// } -// } - -// console.log( -// `COMPLETED ERROR CHECK FOR ${new Date().toISOString().slice(0, 16)}`, -// ); - -// if (process.env.NODE_ENV === "production") { -// const slackBody = { -// text: `Completed error check for ${new Date().toISOString().slice(0, 16)}`, -// blocks: [ -// { -// type: "section", -// text: { -// type: "mrkdwn", -// text: `Error check completed for ${new Date().toISOString().slice(0, 16)}`, -// }, -// }, -// ], -// }; - -// await fetch(process.env.SLACK_WEBHOOK_URL!, { -// method: "POST", -// body: JSON.stringify(slackBody), -// }); -// } -// }; - -// check() -// .catch((error) => { -// console.error(error); -// process.exit(1); -// }) -// .finally(() => { -// process.exit(0); -// }); - -import { initInfisical } from "./external/infisical/initInfisical.js"; - -await initInfisical(); - -await import("./scan/runScan.js"); diff --git a/server/src/external/stripe/handleStripeWebhookEvent.ts b/server/src/external/stripe/handleStripeWebhookEvent.ts index 3cb057057..3b7863a33 100644 --- a/server/src/external/stripe/handleStripeWebhookEvent.ts +++ b/server/src/external/stripe/handleStripeWebhookEvent.ts @@ -1,8 +1,4 @@ -import { - CusExpand, - type FullCustomer, - type Organization, -} from "@autumn/shared"; +import { type Organization } from "@autumn/shared"; import * as Sentry from "@sentry/bun"; import chalk from "chalk"; import { Stripe } from "stripe"; @@ -12,8 +8,6 @@ import { unsetOrgStripeKeys } from "@/internal/orgs/orgUtils.js"; import type { ExtendedRequest } from "@/utils/models/Request.js"; import type { AutumnContext } from "../../honoUtils/HonoEnv.js"; import { deleteCachedApiCustomer } from "../../internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js"; -import { setCachedApiInvoices } from "../../internal/customers/cusUtils/apiCusCacheUtils/setCachedApiInvoices.js"; -import { setCachedApiSubs } from "../../internal/customers/cusUtils/apiCusCacheUtils/setCachedApiSubs.js"; import type { Logger } from "../logtail/logtailUtils.js"; import { getSentryTags } from "../sentry/sentryUtils.js"; import { handleCheckoutSessionCompleted } from "./webhookHandlers/handleCheckoutCompleted.js"; @@ -108,45 +102,53 @@ const handleStripeWebhookRefresh = async ({ return; } - let fullCus: FullCustomer | undefined; - if ( - updateProductEvents.includes(eventType) || - updateInvoiceEvents.includes(eventType) - ) { - fullCus = await CusService.getFull({ - db, - idOrInternalId: cus.id!, - orgId: org.id, - env, - withEntities: true, - withSubs: true, - expand: [CusExpand.Invoices], - }); + logger.info(`Attempting delete cached api customer! ${eventType}`); + await deleteCachedApiCustomer({ + customerId: cus.id!, + orgId: org.id, + env, + source: `handleStripeWebhookRefresh: ${eventType}`, + }); - if (updateProductEvents.includes(eventType)) { - await setCachedApiSubs({ - ctx, - fullCus, - customerId: cus.id!, - }); - } + // let fullCus: FullCustomer | undefined; + // if ( + // updateProductEvents.includes(eventType) || + // updateInvoiceEvents.includes(eventType) + // ) { + // fullCus = await CusService.getFull({ + // db, + // idOrInternalId: cus.id!, + // orgId: org.id, + // env, + // withEntities: true, + // withSubs: true, + // expand: [CusExpand.Invoices], + // }); - if (updateInvoiceEvents.includes(eventType)) { - await setCachedApiInvoices({ - ctx, - fullCus, - customerId: cus.id!, - }); - } - } else { - logger.info(`Attempting delete cached api customer! ${eventType}`); - await deleteCachedApiCustomer({ - customerId: cus.id!, - orgId: org.id, - env, - source: `handleStripeWebhookRefresh: ${eventType}`, - }); - } + // if (updateProductEvents.includes(eventType)) { + // await setCachedApiSubs({ + // ctx, + // fullCus, + // customerId: cus.id!, + // }); + // } + + // if (updateInvoiceEvents.includes(eventType)) { + // await setCachedApiInvoices({ + // ctx, + // fullCus, + // customerId: cus.id!, + // }); + // } + // } else { + // logger.info(`Attempting delete cached api customer! ${eventType}`); + // await deleteCachedApiCustomer({ + // customerId: cus.id!, + // orgId: org.id, + // env, + // source: `handleStripeWebhookRefresh: ${eventType}`, + // }); + // } } }; diff --git a/server/src/honoMiddlewares/refreshCacheMiddleware.ts b/server/src/honoMiddlewares/refreshCacheMiddleware.ts index 4a1692b9e..95636abce 100644 --- a/server/src/honoMiddlewares/refreshCacheMiddleware.ts +++ b/server/src/honoMiddlewares/refreshCacheMiddleware.ts @@ -111,7 +111,29 @@ export const refreshCacheMiddleware = async ( customerId: body.customer_id, orgId: org.id, env: env, + source: "refreshCacheMiddleware", + logger, }); + + // Schedule cache verification job for 1 minute later + // try { + // await addTaskToQueue({ + // jobName: JobName.VerifyCacheConsistency, + // payload: { + // customerId: body.customer_id, + // orgId: org.id, + // env: env, + // source: `post-${pathname.replace("/", "")}`, + // }, + // delayMs: 60000, // 1 minute delay + // }); + // logger.info( + // `Scheduled cache verification for ${body.customer_id} in 60s`, + // ); + // } catch (error) { + // // Don't fail the request if scheduling fails + // logger.error(`Failed to schedule cache verification: ${error}`); + // } } } }; diff --git a/server/src/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.ts b/server/src/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.ts index 11f591bc1..16463c836 100644 --- a/server/src/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.ts +++ b/server/src/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.ts @@ -1,4 +1,7 @@ -import { logger } from "../../../../external/logtail/logtailUtils.js"; +import { + type Logger, + logger as loggerInstance, +} from "../../../../external/logtail/logtailUtils.js"; import { redis } from "../../../../external/redis/initRedis.js"; /** @@ -11,12 +14,16 @@ export const deleteCachedApiCustomer = async ({ orgId, env, source, + logger, }: { customerId: string; orgId: string; env: string; source?: string; + logger?: Logger; }): Promise => { + logger = loggerInstance || loggerInstance; + if (redis.status !== "ready") { logger.warn("❗️ Redis not ready, skipping cache deletion", { data: { diff --git a/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiCustomer.ts b/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiCustomer.ts index 0ffe424b1..e4868ee18 100644 --- a/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiCustomer.ts +++ b/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiCustomer.ts @@ -113,14 +113,14 @@ export const setCachedApiCustomer = async ({ }); if (result === "CACHE_EXISTS") { - logger.debug( + logger.info( `Cache already exists for customer ${customerId}, source: ${source}`, ); return; } if (result === "STALE_WRITE") { - logger.debug( + logger.info( `Stale write blocked for customer ${customerId}, source: ${source}`, ); return; diff --git a/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiInvoices.ts b/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiInvoices.ts index 3225ccf85..a3af7aa1e 100644 --- a/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiInvoices.ts +++ b/server/src/internal/customers/cusUtils/apiCusCacheUtils/setCachedApiInvoices.ts @@ -37,7 +37,6 @@ export const setCachedApiInvoices = async ({ // Then write to Redis await tryRedisWrite(async () => { - // Update customer invoices await redis.setInvoices( JSON.stringify(masterApiInvoices), org.id, diff --git a/server/src/queue/JobName.ts b/server/src/queue/JobName.ts index 838d9d1c2..f54acb823 100644 --- a/server/src/queue/JobName.ts +++ b/server/src/queue/JobName.ts @@ -17,4 +17,5 @@ export enum JobName { InsertEventBatch = "insert-event-batch", ClearCreditSystemCustomerCache = "clear-credit-system-customer-cache", + VerifyCacheConsistency = "verify-cache-consistency", } diff --git a/server/src/queue/initWorkers.ts b/server/src/queue/initWorkers.ts index 63141a45e..24ea3fb6b 100644 --- a/server/src/queue/initWorkers.ts +++ b/server/src/queue/initWorkers.ts @@ -24,6 +24,7 @@ import { setSentryTags } from "../external/sentry/sentryUtils.js"; import { createWorkerContext } from "./createWorkerContext.js"; import { QUEUE_URL, sqs } from "./initSqs.js"; import { JobName } from "./JobName.js"; +import { verifyCacheConsistency } from "./jobs/verifyCacheConsistency.js"; const actionHandlers = [ JobName.HandleProductsUpdated, @@ -145,7 +146,10 @@ const processMessage = async ({ } if (job.name === JobName.SyncBalanceBatchV2) { - if (!ctx) return; + if (!ctx) { + workerLogger.error("No context found for sync balance batch v2 job"); + return; + } await syncItemV2({ ctx, @@ -154,6 +158,19 @@ const processMessage = async ({ return; } + if (job.name === JobName.VerifyCacheConsistency) { + if (!ctx) { + workerLogger.error("No context found for verify cache consistency job"); + return; + } + + await verifyCacheConsistency({ + ctx, + payload: job.data, + }); + return; + } + if (job.name === JobName.InsertEventBatch) { await runInsertEventBatch({ db, diff --git a/server/src/queue/jobs/verifyCacheConsistency.ts b/server/src/queue/jobs/verifyCacheConsistency.ts new file mode 100644 index 000000000..df9a5f3e3 --- /dev/null +++ b/server/src/queue/jobs/verifyCacheConsistency.ts @@ -0,0 +1,95 @@ +import { CusExpand } from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; +import { CusService } from "@/internal/customers/CusService.js"; +import { getCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/getCachedApiCustomer.js"; +import { getApiCustomerBase } from "../../internal/customers/cusUtils/apiCusUtils/getApiCustomerBase"; + +export const verifyCacheConsistency = async ({ + ctx, + payload, +}: { + ctx: AutumnContext; + payload: { + customerId: string; + orgId: string; + env: string; + source: string; + }; +}) => { + const { customerId, source } = payload; + const { db, org, env, logger } = ctx; + + // Get from cache + const { apiCustomer: cachedCustomer } = await getCachedApiCustomer({ + ctx, + customerId, + source: "verify", + }); + + // Get fresh from DB + const fullCus = await CusService.getFull({ + db, + idOrInternalId: customerId, + orgId: org.id, + env, + withEntities: true, + withSubs: true, + expand: [CusExpand.Invoices], + }); + + const { apiCustomer: dbCustomer } = await getApiCustomerBase({ + ctx, + fullCus, + withAutumnId: true, + }); + + // 1. Check if products match + const checkProductsMatch = () => { + for (const subscription of dbCustomer.subscriptions) { + const cachedSubscription = cachedCustomer.subscriptions.find( + (s) => s.plan_id === subscription.plan_id, + ); + + if (!cachedSubscription) return false; + } + + for (const scheduledSubscription of dbCustomer.scheduled_subscriptions) { + const cachedScheduledSubscription = + cachedCustomer.scheduled_subscriptions.find( + (s) => s.plan_id === scheduledSubscription.plan_id, + ); + + if (!cachedScheduledSubscription) return false; + } + }; + + // if (productMismatch) { + // logger.error( + // "[VerifyCacheConsistency] Cache inconsistency detected! Auto-fixing...", + // { + // data: { + // customerId, + // source, + // cached: [...cachedProductIds], + // fresh: [...freshProductIds], + // }, + // }, + // ); + + // // Auto-fix by deleting cache (next read will repopulate) + // await deleteCachedApiCustomer({ + // customerId, + // orgId: org.id, + // env, + // source: "verification-auto-fix", + // }); + + // logger.info( + // `[VerifyCacheConsistency] Cache cleared for ${customerId}, will repopulate on next read`, + // ); + // } else { + // logger.info( + // `[VerifyCacheConsistency] Cache is consistent for ${customerId}`, + // ); + // } +}; diff --git a/server/src/queue/queueUtils.ts b/server/src/queue/queueUtils.ts index 102db797b..c8727a3e0 100644 --- a/server/src/queue/queueUtils.ts +++ b/server/src/queue/queueUtils.ts @@ -27,6 +27,12 @@ export interface Payloads { events: EventInsert[]; }; [JobName.ClearCreditSystemCustomerCache]: ClearCreditSystemCachePayload; + [JobName.VerifyCacheConsistency]: { + customerId: string; + orgId: string; + env: string; + source: string; + }; [key: string]: unknown; } @@ -64,11 +70,13 @@ export const addTaskToQueue = async ({ payload, messageGroupId, messageDeduplicationId, + delayMs, }: { jobName: T; payload: Payloads[T]; messageGroupId?: string; messageDeduplicationId?: string; + delayMs?: number; }) => { await initializeQueue(); @@ -80,9 +88,15 @@ export const addTaskToQueue = async ({ data: payload, }; + // Convert milliseconds to seconds for SQS (max 900 seconds) + const delaySeconds = delayMs + ? Math.min(Math.floor(delayMs / 1000), 900) + : undefined; + const command = new SendMessageCommand({ QueueUrl: sqsQueueUrl!, MessageBody: JSON.stringify(message), + ...(delaySeconds && { DelaySeconds: delaySeconds }), // FIFO queues require MessageGroupId and MessageDeduplicationId ...(isFifoQueue && { MessageGroupId: messageGroupId || generateId("msg"), @@ -94,6 +108,8 @@ export const addTaskToQueue = async ({ await sqsClient.send(command); } else { // BullMQ implementation (ignores messageGroupId) - await bullmqQueue.add(jobName as string, payload); + await bullmqQueue.add(jobName as string, payload, { + delay: delayMs, + }); } }; diff --git a/server/tests/_temp/temp.test.ts b/server/tests/_temp/temp.test.ts index b2f6e9d18..4890a7697 100644 --- a/server/tests/_temp/temp.test.ts +++ b/server/tests/_temp/temp.test.ts @@ -1,16 +1,9 @@ -import { beforeAll, describe, it } from "bun:test"; +import { beforeAll, describe, test } from "bun:test"; import { ApiVersion } from "@autumn/shared"; import { TestFeature } from "@tests/setup/v2Features.js"; -import { expectProductAttached } from "@tests/utils/expectUtils/expectProductAttached"; import ctx from "@tests/utils/testInitUtils/createTestContext.js"; import chalk from "chalk"; import { AutumnInt } from "@/external/autumn/autumnCli.js"; -import { AutumnCliV2 } from "@/external/autumn/autumnCliV2"; -import { - attachAuthenticatePaymentMethod, - attachFailedPaymentMethod, -} from "@/external/stripe/stripeCusUtils"; -import { timeout } from "@/utils/genUtils"; import { constructFeatureItem, constructPrepaidItem, @@ -20,21 +13,21 @@ import { constructRawProduct, } from "@/utils/scriptUtils/createTestProducts.js"; import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js"; +import { attachAuthenticatePaymentMethod } from "../../src/external/stripe/stripeCusUtils"; import { initCustomerV3 } from "../../src/utils/scriptUtils/testUtils/initCustomerV3"; const pro = constructProduct({ type: "pro", items: [ constructFeatureItem({ - featureId: TestFeature.Users, - includedUsage: 10, + featureId: TestFeature.Credits, + includedUsage: 500, }), ], }); const oneOffCredits = constructRawProduct({ id: "one_off_credits", - // isAddOn: true, items: [ constructPrepaidItem({ featureId: TestFeature.Credits, @@ -48,7 +41,7 @@ const oneOffCredits = constructRawProduct({ const testCase = "temp"; -describe(`${chalk.yellowBright("temp: one off credits test")}`, () => { +describe(`${chalk.yellowBright("temp: invoice payment failed for one off credits")}`, () => { const customerId = testCase; const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 }); @@ -71,23 +64,35 @@ describe(`${chalk.yellowBright("temp: one off credits test")}`, () => { product_id: pro.id, }); + await autumnV1.attach({ + customer_id: customerId, + product_id: oneOffCredits.id, + options: [ + { + feature_id: TestFeature.Credits, + quantity: 100, + }, + ], + }); + }); + + test("should handle invoice payment failed for one off credits", async () => { await attachAuthenticatePaymentMethod({ ctx, customerId, }); - await timeout(1000); - const res = await autumnV1.attach({ customer_id: customerId, product_id: oneOffCredits.id, options: [ { feature_id: TestFeature.Credits, - quantity: 2000, + quantity: 250, }, ], }); + console.log(res); }); }); diff --git a/server/tests/attach/basic/basic2.test.ts b/server/tests/attach/basic/basic2.test.ts index 717e317b8..7252c7fb7 100644 --- a/server/tests/attach/basic/basic2.test.ts +++ b/server/tests/attach/basic/basic2.test.ts @@ -93,6 +93,7 @@ describe(`${chalk.yellowBright("basic2: Testing attach monthly add on")}`, () => cusRes: res, }); }); + return; const monthlyQuantity = 500; diff --git a/server/tests/utils/compare.ts b/server/tests/utils/compare.ts index 1bdb2097c..ec0fe1103 100644 --- a/server/tests/utils/compare.ts +++ b/server/tests/utils/compare.ts @@ -103,9 +103,6 @@ export const compareMainProduct = ({ ).toBeDefined(); if (entitlement.allowance_type === AllowanceType.Unlimited) { - // expect(recEntitlement.unlimited).toStrictEqual(true); - // expect(recEntitlement.balance).toStrictEqual(null); - // expect(recEntitlement.used).toStrictEqual(null); expect(recEntitlement).toMatchObject({ unlimited: true, balance: null, diff --git a/shared/api/customers/changes/V1.1_FeaturesArrayToObject.ts b/shared/api/customers/changes/V1.1_FeaturesArrayToObject.ts index 19d6d4154..c02756ab0 100644 --- a/shared/api/customers/changes/V1.1_FeaturesArrayToObject.ts +++ b/shared/api/customers/changes/V1.1_FeaturesArrayToObject.ts @@ -158,7 +158,7 @@ export const V1_1_FeaturesArrayToObject = defineVersionChange({ included_usage: mergedItem.included_usage, next_reset_at: mergedItem.next_reset_at, usage_limit: mergedItem.usage_limit, - unlimited: false, + unlimited: feature.unlimited, // inherit from parent overage_allowed: mergedItem.overage_allowed, } satisfies ApiCusFeatureV2); }