resolved merge conflicts

This commit is contained in:
John Yeo
2026-01-12 08:58:12 +00:00
211 changed files with 13828 additions and 1353 deletions

781
bun.lock

File diff suppressed because it is too large Load Diff

View File

@@ -7,12 +7,12 @@ BUN_PARALLEL_COMPACT \
'server/tests/merged/downgrade' \
'server/tests/merged/separate' \
'server/tests/merged/add' \
# 'server/tests/merged/group' \
# 'server/tests/merged/prepaid' \
# 'server/tests/merged/upgrade' \
# 'server/tests/merged/addOn' \
# 'server/tests/merged/trial' \
# 'server/tests/core/cancel' \
'server/tests/merged/group' \
'server/tests/merged/prepaid' \
'server/tests/merged/upgrade' \
'server/tests/merged/addOn' \
'server/tests/merged/trial' \
'server/tests/core/cancel' \
--max=6 \

View File

@@ -136,17 +136,6 @@ const handleShortDurationCusEnt = async ({
`Reseting short cus ent (${cusEnt.feature_id}) [${ent.interval}], customer: ${cusEnt.customer_id}, org: ${cusEnt.customer.org_id}`,
);
// const org = await OrgService.get({
// db,
// orgId: cusEnt.customer.org_id,
// });
// await deleteCachedApiCustomer({
// customerId: cusEnt.customer.id!,
// orgId: org.id,
// env: cusEnt.customer.env,
// });
updatedCusEnts.push(newCusEnt);
return newCusEnt;

View File

@@ -126,10 +126,10 @@ export class AutumnInt {
return response.json();
}
async post(path: string, body: any) {
async post(path: string, body: any, headers?: Record<string, string>) {
const response = await fetch(`${this.baseUrl}${path}`, {
method: "POST",
headers: this.headers,
headers: { ...this.headers, ...headers },
body: JSON.stringify(body),
});
@@ -252,13 +252,13 @@ export class AutumnInt {
return data;
}
async attach(params: AttachBodyV0) {
async attach(params: AttachBodyV0, headers?: Record<string, string>) {
// const data = await this.post(`/attach`, {
// customer_id: customerId,
// product_id: productId,
// options: toSnakeCase(options),
// });
const data = await this.post(`/attach`, params);
const data = await this.post(`/attach`, params, headers);
return data;
}
@@ -434,6 +434,22 @@ export class AutumnInt {
});
return data;
},
setBalance: async ({
customerId,
balances,
entityId,
}: {
customerId: string;
balances: Array<{ feature_id: string; balance: number }>;
entityId?: string;
}) => {
const data = await this.post(`/customers/${customerId}/balances`, {
balances,
entity_id: entityId,
});
return data;
},
};
entities = {

View File

@@ -27,6 +27,9 @@ if (!process.env.CACHE_URL) {
export const REGION_US_EAST_2 = "us-east-2";
export const REGION_US_WEST_2 = "us-west-2";
// All configured regions
export const ALL_REGIONS = [REGION_US_EAST_2, REGION_US_WEST_2] as const;
// Current region this instance is running in
export const currentRegion = process.env.AWS_REGION || REGION_US_WEST_2;
@@ -36,6 +39,11 @@ const regionToCacheUrl: Record<string, string | undefined> = {
[REGION_US_WEST_2]: process.env.CACHE_URL, // Default/us-west-2 URL
};
/** Get all regions that have configured cache URLs */
export const getConfiguredRegions = (): string[] => {
return ALL_REGIONS.filter((region) => regionToCacheUrl[region]);
};
/** Configure a Redis instance with custom commands */
const configureRedisInstance = (redisInstance: Redis): Redis => {
const batchDeductionScript = getBatchDeductionScript();

View File

@@ -7,13 +7,14 @@ import {
RecaseError,
} from "@shared/index";
import { RCMappingService } from "@/external/revenueCat/misc/RCMappingService";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { CusService } from "@/internal/customers/CusService";
import { ProductService } from "@/internal/products/ProductService";
import { getOrCreateCustomer } from "../../../internal/customers/cusUtils/getOrCreateCustomer";
/**
* Resolves a RevenueCat product ID to an Autumn product and fetches the customer.
* Also sets ctx.customerId for cache invalidation by the refresh middleware.
* Throws if product mapping, product, customer is not found, or customer has non-RevenueCat products.
*/
export const resolveRevenuecatResources = async ({
@@ -22,7 +23,7 @@ export const resolveRevenuecatResources = async ({
customerId,
autoCreateCustomer = false,
}: {
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
revenuecatProductId: string;
customerId: string;
autoCreateCustomer?: boolean;
@@ -58,15 +59,15 @@ export const resolveRevenuecatResources = async ({
}),
autoCreateCustomer
? getOrCreateCustomer({
ctx,
customerId,
})
ctx,
customerId,
})
: CusService.getFull({
db,
idOrInternalId: customerId,
orgId: org.id,
env,
}),
db,
idOrInternalId: customerId,
orgId: org.id,
env,
}),
]);
// If the customer has a product from a different processor than RevenueCat and it has no subscriptions, throw an error
@@ -74,7 +75,7 @@ export const resolveRevenuecatResources = async ({
customer.customer_products.some(
(cp) =>
cp.processor?.type !== ProcessorType.RevenueCat &&
((cp.subscription_ids?.length ?? 0) !== 0),
(cp.subscription_ids?.length ?? 0) !== 0,
)
) {
throw new RecaseError({
@@ -84,8 +85,12 @@ export const resolveRevenuecatResources = async ({
}
const cusProducts = customer.customer_products.filter(
(cp) => (cp.processor?.type === ProcessorType.RevenueCat || cp.product.is_default),
(cp) =>
cp.processor?.type === ProcessorType.RevenueCat || cp.product.is_default,
);
// Set customer ID in context for cache refresh middleware
ctx.customerId = customer.id ?? "";
return { product, customer, cusProducts };
};

View File

@@ -3,11 +3,11 @@ import type { Organization } from "@shared/models/orgModels/orgTable";
import chalk from "chalk";
import type { Context, Next } from "hono";
import type { Logger } from "@/external/logtail/logtailUtils";
import type { HonoEnv } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookHonoEnv } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { OrgService } from "@/internal/orgs/OrgService";
export const revenuecatSeederMiddleware = async (
c: Context<HonoEnv>,
c: Context<RevenueCatWebhookHonoEnv>,
next: Next,
) => {
const { orgId, env } = c.req.param();
@@ -53,7 +53,7 @@ export const logRevCatWebhook = ({
};
export const revenuecatLogMiddleware = async (
c: Context<HonoEnv>,
c: Context<RevenueCatWebhookHonoEnv>,
next: Next,
) => {
const { logger, org } = c.get("ctx");

View File

@@ -9,7 +9,6 @@ import type {
WebhookUnCancellation,
} from "@puzzmo/revenue-cat-webhook-types";
import { type Context, Hono } from "hono";
import type { HonoEnv } from "@/honoUtils/HonoEnv";
import { getRevenuecatWebhookSecret } from "./misc/getRevenuecatWebhookSecret";
import {
revenuecatLogMiddleware,
@@ -22,14 +21,17 @@ import { handleExpiration } from "./webhookHandlers/handleRevenuecatExpiration";
import { handleInitialPurchase } from "./webhookHandlers/handleRevenuecatInitialPurchase";
import { handleNonRenewingPurchase } from "./webhookHandlers/handleRevenuecatNonRenewingPurchase";
import { handleUncancellation } from "./webhookHandlers/handleRevenuecatUncancellation";
import type { RevenueCatWebhookHonoEnv } from "./webhookMiddlewares/revenuecatWebhookContext";
import { revenuecatWebhookRefreshMiddleware } from "./webhookMiddlewares/revenuecatWebhookRefreshMiddleware";
export const revenuecatWebhookRouter = new Hono<HonoEnv>();
export const revenuecatWebhookRouter = new Hono<RevenueCatWebhookHonoEnv>();
revenuecatWebhookRouter.post(
"/:orgId/:env",
revenuecatSeederMiddleware,
revenuecatLogMiddleware,
async (c: Context<HonoEnv>) => {
revenuecatWebhookRefreshMiddleware,
async (c: Context<RevenueCatWebhookHonoEnv>) => {
const ctx = c.get("ctx");
const { logger, org, env } = ctx;
const Authorization = c.req.header("Authorization");
@@ -46,6 +48,9 @@ revenuecatWebhookRouter.post(
return c.json({ error: "Unauthorized" }, 401);
}
// Set event type in context for logging in refresh middleware
ctx.revenuecatEventType = body.event.type;
switch (body.event.type) {
case "INITIAL_PURCHASE":
await handleInitialPurchase({

View File

@@ -8,11 +8,10 @@ import {
} from "@shared/index";
import { createStripeCli } from "@/external/connect/createStripeCli";
import { resolveRevenuecatResources } from "@/external/revenueCat/misc/resolveRevenuecatResources";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { createFullCusProduct } from "@/internal/customers/add-product/createFullCusProduct";
import { CusProductService } from "@/internal/customers/cusProducts/CusProductService";
import { getExistingCusProducts } from "@/internal/customers/cusProducts/cusProductUtils/getExistingCusProducts";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
import {
attachToInsertParams,
isProductUpgrade,
@@ -24,7 +23,7 @@ export const handleRenewal = async ({
ctx,
}: {
event: WebhookRenewal;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, org, env, logger, features } = ctx;
const { product_id, app_user_id } = event;
@@ -63,12 +62,6 @@ export const handleRenewal = async ({
},
});
logger.info(`Marked past due product as active: ${curSameProduct.id}`);
await deleteCachedApiCustomer({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleRenewal: ${product.id}`,
});
return { success: true };
}
@@ -116,13 +109,6 @@ export const handleRenewal = async ({
});
logger.info(`Reactivated cus_product: ${curSameProduct.id}`);
await deleteCachedApiCustomer({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleRenewal: ${product.id}`,
});
return { success: true };
}
@@ -157,12 +143,5 @@ export const handleRenewal = async ({
`Created cus_product for ${product.id} with scenario: ${scenario} (renewal)`,
);
await deleteCachedApiCustomer({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleRenewal: ${product.id}`,
});
return { success: true };
};

View File

@@ -2,13 +2,12 @@ import type { WebhookBillingIssue } from "@puzzmo/revenue-cat-webhook-types";
import { RecaseError } from "@shared/api/errors/base/RecaseError";
import { ErrCode } from "@shared/enums/ErrCode";
import { CusProductStatus } from "@shared/models/cusProductModels/cusProductEnums";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import {
ACTIVE_STATUSES,
CusProductService,
} from "@/internal/customers/cusProducts/CusProductService";
import { getExistingCusProducts } from "@/internal/customers/cusProducts/cusProductUtils/getExistingCusProducts";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
import { resolveRevenuecatResources } from "../misc/resolveRevenuecatResources";
export const handleBillingIssue = async ({
@@ -16,12 +15,12 @@ export const handleBillingIssue = async ({
ctx,
}: {
event: WebhookBillingIssue;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, logger, org, env } = ctx;
const { db, logger } = ctx;
const { product_id, app_user_id } = event;
const { product, customer, cusProducts } = await resolveRevenuecatResources({
const { product, cusProducts } = await resolveRevenuecatResources({
ctx,
revenuecatProductId: product_id,
customerId: app_user_id,
@@ -56,13 +55,6 @@ export const handleBillingIssue = async ({
},
});
await deleteCachedApiCustomer({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleRevenuecatBillingIssue: ${product.id}`,
});
return { success: true };
}

View File

@@ -1,19 +1,18 @@
import type { WebhookCancellation } from "@puzzmo/revenue-cat-webhook-types";
import { ErrCode, RecaseError } from "@shared/index";
import { resolveRevenuecatResources } from "@/external/revenueCat/misc/resolveRevenuecatResources";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { CusProductService } from "@/internal/customers/cusProducts/CusProductService";
import { getExistingCusProducts } from "@/internal/customers/cusProducts/cusProductUtils/getExistingCusProducts";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
export const handleCancellation = async ({
event,
ctx,
}: {
event: WebhookCancellation;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, org, env, logger } = ctx;
const { db, logger } = ctx;
const { product_id, original_app_user_id, app_user_id, expiration_at_ms } =
event;
@@ -49,11 +48,4 @@ export const handleCancellation = async ({
logger.info(
`Marked cus_product ${curSameProduct.id} as cancelled, will expire at ${expiration_at_ms}`,
);
await deleteCachedApiCustomer({
customerId: original_app_user_id ?? app_user_id,
orgId: org.id,
env,
source: `handleRevenuecatCancellation: ${product.id}`,
});
};

View File

@@ -1,11 +1,10 @@
import type { WebhookExpiration } from "@puzzmo/revenue-cat-webhook-types";
import { CusProductStatus, ErrCode, RecaseError } from "@shared/index";
import { resolveRevenuecatResources } from "@/external/revenueCat/misc/resolveRevenuecatResources";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { CusProductService } from "@/internal/customers/cusProducts/CusProductService";
import { activateDefaultProduct } from "@/internal/customers/cusProducts/cusProductUtils";
import { getExistingCusProducts } from "@/internal/customers/cusProducts/cusProductUtils/getExistingCusProducts";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
import { isOneOff } from "@/internal/products/productUtils";
export const handleExpiration = async ({
@@ -13,9 +12,9 @@ export const handleExpiration = async ({
ctx,
}: {
event: WebhookExpiration;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, org, env, logger } = ctx;
const { db, logger } = ctx;
const { product_id, original_app_user_id, app_user_id } = event;
const { product, customer, cusProducts } = await resolveRevenuecatResources({
@@ -62,11 +61,4 @@ export const handleExpiration = async ({
curCusProduct: curSameProduct,
});
}
await deleteCachedApiCustomer({
customerId: event.original_app_user_id ?? event.app_user_id,
orgId: org.id,
env,
source: `handleRevenuecatExpiration: ${product.id}`,
});
};

View File

@@ -9,11 +9,10 @@ import {
} from "@shared/index";
import { createStripeCli } from "@/external/connect/createStripeCli";
import { resolveRevenuecatResources } from "@/external/revenueCat/misc/resolveRevenuecatResources";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { createFullCusProduct } from "@/internal/customers/add-product/createFullCusProduct";
import { CusProductService } from "@/internal/customers/cusProducts/CusProductService";
import { getExistingCusProducts } from "@/internal/customers/cusProducts/cusProductUtils/getExistingCusProducts";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
import {
attachToInsertParams,
isProductUpgrade,
@@ -25,7 +24,7 @@ export const handleInitialPurchase = async ({
ctx,
}: {
event: WebhookInitialPurchase;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, org, env, logger, features } = ctx;
const { product_id, app_user_id } = event;
@@ -115,11 +114,4 @@ export const handleInitialPurchase = async ({
logger.info(
`Created cus_product for ${product.id} with scenario: ${scenario}`,
);
await deleteCachedApiCustomer({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleRevenuecatInitialPurchase: ${product.id}`,
});
};

View File

@@ -7,9 +7,8 @@ import {
} from "@shared/index";
import { createStripeCli } from "@/external/connect/createStripeCli";
import { resolveRevenuecatResources } from "@/external/revenueCat/misc/resolveRevenuecatResources";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { createFullCusProduct } from "@/internal/customers/add-product/createFullCusProduct";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
import { attachToInsertParams } from "@/internal/products/productUtils";
import { oneOffOrAddOn } from "@/internal/products/productUtils/classifyProduct";
@@ -18,7 +17,7 @@ export const handleNonRenewingPurchase = async ({
ctx,
}: {
event: WebhookNonRenewingPurchase;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, org, env, logger, features } = ctx;
@@ -69,11 +68,4 @@ export const handleNonRenewingPurchase = async ({
logger.info(
`Created cus_product for ${product.id} with scenario: ${scenario}`,
);
await deleteCachedApiCustomer({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleRevenuecatNonRenewingPurchase: ${product.id}`,
});
};

View File

@@ -6,18 +6,17 @@ import {
RecaseError,
} from "@shared/index";
import { resolveRevenuecatResources } from "@/external/revenueCat/misc/resolveRevenuecatResources";
import type { AutumnContext } from "@/honoUtils/HonoEnv";
import type { RevenueCatWebhookContext } from "@/external/revenueCat/webhookMiddlewares/revenuecatWebhookContext";
import { CusProductService } from "@/internal/customers/cusProducts/CusProductService";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
export const handleUncancellation = async ({
event,
ctx,
}: {
event: WebhookUnCancellation;
ctx: AutumnContext;
ctx: RevenueCatWebhookContext;
}) => {
const { db, org, env } = ctx;
const { db } = ctx;
const { product_id, original_app_user_id, app_user_id } = event;
const { product, cusProducts } = await resolveRevenuecatResources({
@@ -43,13 +42,6 @@ export const handleUncancellation = async ({
status: CusProductStatus.Active,
},
});
await deleteCachedApiCustomer({
customerId: original_app_user_id ?? app_user_id,
orgId: org.id,
env,
source: `handleRevenuecatUncancellation: ${product.id}`,
});
} else {
throw new RecaseError({
message: "Cus product not found",

View File

@@ -0,0 +1,14 @@
import type { AutumnContext, HonoEnv } from "@/honoUtils/HonoEnv.js";
export type RevenueCatWebhookContext = AutumnContext & {
/** Customer ID to invalidate cache for after handler completes */
customerId?: string;
/** Event type for logging purposes */
revenuecatEventType?: string;
};
export type RevenueCatWebhookHonoEnv = Omit<HonoEnv, "Variables"> & {
Variables: {
ctx: RevenueCatWebhookContext;
};
};

View File

@@ -0,0 +1,46 @@
import type { Context, Next } from "hono";
import { deleteCachedApiCustomer } from "@/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
import type { RevenueCatWebhookHonoEnv } from "./revenuecatWebhookContext.js";
/**
* Middleware that refreshes customer cache after RevenueCat webhook handlers complete.
* Handlers must set `ctx.revenuecatCustomerId` for the cache to be invalidated.
*/
export const revenuecatWebhookRefreshMiddleware = async (
c: Context<RevenueCatWebhookHonoEnv>,
next: Next,
) => {
// Run the main handler first
await next();
// Post-processing: refresh cache
const ctx = c.get("ctx");
const { logger, org, env, customerId, revenuecatEventType } = ctx;
if (!customerId) {
logger.warn(
"RevenueCat webhook: No customer ID set in context, skipping cache refresh",
);
return;
}
try {
logger.info(
`Attempting delete cached api customer! RevenueCat ${revenuecatEventType}`,
);
await deleteCachedApiCustomer({
customerId,
orgId: org.id,
env,
source: `revenuecatWebhookRefreshMiddleware: ${revenuecatEventType}`,
logger,
});
} catch (error) {
logger.error(`RevenueCat webhook, error refreshing cache: ${error}`, {
error: {
message: error instanceof Error ? error.message : String(error),
},
});
}
};

View File

@@ -32,12 +32,14 @@ export const getSentryTags = ({
messageId,
path,
method,
alert = false,
}: {
ctx: AutumnContext;
customerId?: string;
messageId?: string;
path?: string;
method?: string;
alert?: boolean;
}) => {
if (!ctx) return;
return {
@@ -46,11 +48,12 @@ export const getSentryTags = ({
env: ctx.env || "unknown",
auth_type: ctx.authType,
request_id: ctx.id || "",
customer_id: customerId,
customer_id: customerId || ctx.customerId,
message_id: messageId,
path: path,
method: method,
email: ctx.user?.email,
alert: alert ? "true" : "false",
};
};

View File

@@ -32,9 +32,10 @@ export const searchStripeMeter = async ({
}) => {
const allStripeMeters = [];
let hasMore = true;
let startingAfter;
let startingAfter: string | undefined;
const start = performance.now();
// Get max 200 meters
while (hasMore) {
const response: any = await stripeCli.billing.meters.list({
limit: 100,

View File

@@ -132,7 +132,8 @@ export const findPriceInStripeItems = ({
if (subItem) {
itemMatch =
config.stripe_price_id === subItem.price?.id ||
config.stripe_product_id === subItem.price?.product;
config.stripe_product_id === subItem.price?.product ||
config.stripe_empty_price_id === subItem.price?.id;
}
if (lineItem) {

View File

@@ -14,7 +14,6 @@ import { cusProductToSub } from "@/internal/customers/cusProducts/cusProductUtil
import { FeatureService } from "@/internal/features/FeatureService.js";
import { isFixedPrice } from "@/internal/products/prices/priceUtils/usagePriceUtils/classifyUsagePrice.js";
import { getBillingType } from "@/internal/products/prices/priceUtils.js";
import { deleteCachedApiCustomer } from "../../../../internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
import {
getFullStripeInvoice,
invoiceToSubId,
@@ -133,15 +132,6 @@ export const sendUsageAndReset = async ({
handled.push(handledPrepaid);
}
}
if (handled.some((h) => Boolean(h))) {
await deleteCachedApiCustomer({
customerId: customer.id!,
orgId: org.id,
env,
source: `handleInvoiceCreated: ${invoice.id}`,
});
}
};
export const handleInvoiceCreated = async ({

View File

@@ -7,7 +7,6 @@ import { handleUpgradeFlow } from "@server/internal/customers/attach/attachFunct
import { attachParamsToCurCusProduct } from "@server/internal/customers/attach/attachUtils/convertAttachParams";
import { getDefaultAttachConfig } from "@server/internal/customers/attach/attachUtils/getAttachConfig";
import type { AttachParams } from "@server/internal/customers/cusProducts/AttachParams";
import { deleteCachedApiCustomer } from "@server/internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer";
import { MetadataService } from "@server/internal/metadata/MetadataService";
import type Stripe from "stripe";
@@ -64,10 +63,4 @@ export const handleInvoiceActionRequiredCompleted = async ({
db: ctx.db,
id: metadata.id,
});
await deleteCachedApiCustomer({
customerId: attachParams.customer.id || "",
orgId: attachParams.org.id,
env: attachParams.customer.env,
});
};

View File

@@ -5,7 +5,6 @@ import type { AttachParams } from "@/internal/customers/cusProducts/AttachParams
import { attachToInsertParams } from "@/internal/products/productUtils.js";
import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
import { CusService } from "../../../../internal/customers/CusService.js";
import { deleteCachedApiCustomer } from "../../../../internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
import { MetadataService } from "../../../../internal/metadata/MetadataService.js";
export const handleInvoiceCheckoutPaid = async ({
@@ -71,10 +70,4 @@ export const handleInvoiceCheckoutPaid = async ({
customerId = customer?.id;
}
await deleteCachedApiCustomer({
customerId: customerId || "",
orgId: org.id,
env,
});
};

View File

@@ -12,7 +12,6 @@ import { isFreeProduct, isOneOff } from "@/internal/products/productUtils.js";
import { notNullish } from "@/utils/genUtils.js";
import { getStripeNow } from "@/utils/scriptUtils/testClockUtils.js";
import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
import { deleteCachedApiCustomer } from "../../../../internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
export const handleSchedulePhaseCompleted = async ({
ctx,
@@ -100,14 +99,6 @@ export const handleSchedulePhaseCompleted = async ({
});
}
}
// Maybe activate default product?
await deleteCachedApiCustomer({
customerId: cusProduct.customer?.id || "",
orgId: org.id,
env,
source: "handleSchedulePhaseCompleted",
});
}
}

View File

@@ -6,7 +6,6 @@ import {
import type Stripe from "stripe";
import { addProductsUpdatedWebhookTask } from "@/internal/analytics/handlers/handleProductsUpdated.js";
import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
import { deleteCachedApiCustomer } from "../../../../internal/customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
export const isSubPastDue = ({
previousAttributes,
@@ -70,13 +69,4 @@ export const handleSubPastDue = async ({
});
}
}
const customerId = updatedCusProducts[0].customer?.id;
if (customerId) {
await deleteCachedApiCustomer({
customerId,
orgId: org.id,
env,
});
}
};

View File

@@ -1,9 +1,5 @@
import { CusExpand, type FullCustomer } from "@autumn/shared";
import type { Context, Next } from "hono";
import { CusService } from "@/internal/customers/CusService.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 { StripeWebhookHonoEnv } from "./stripeWebhookContext.js";
const updateProductEvents = ["customer.subscription.updated"];
@@ -34,7 +30,7 @@ export const stripeWebhookRefreshMiddleware = async (
// Post-processing: refresh cache
const ctx = c.get("ctx");
const { db, logger, org, env, stripeEvent } = ctx;
const { logger, org, env, stripeEvent } = ctx;
if (!stripeEvent) return;
@@ -69,46 +65,13 @@ export const stripeWebhookRefreshMiddleware = async (
}
logger.info(`Attempting delete cached api customer! ${eventType}`);
let fullCus: FullCustomer | undefined;
if (
updateProductEvents.includes(eventType) ||
updateInvoiceEvents.includes(eventType)
) {
fullCus = await CusService.getFull({
db,
idOrInternalId: customer.id!,
orgId: org.id,
env,
withEntities: true,
withSubs: true,
expand: [CusExpand.Invoices],
});
if (updateProductEvents.includes(eventType)) {
await setCachedApiSubs({
ctx,
fullCus,
customerId: customer.id!,
});
}
if (updateInvoiceEvents.includes(eventType)) {
await setCachedApiInvoices({
ctx,
fullCus,
customerId: customer.id!,
});
}
} else {
logger.info(`Attempting delete cached api customer! ${eventType}`);
await deleteCachedApiCustomer({
customerId: customer.id!,
orgId: org.id,
env,
source: `stripeWebhookRefreshMiddleware: ${eventType}`,
});
}
await deleteCachedApiCustomer({
customerId: customer.id!,
orgId: org.id,
env,
source: `stripeWebhookRefreshMiddleware: ${eventType}`,
logger,
});
}
} catch (error) {
logger.error(`Stripe webhook, error refreshing cache: ${error}`, {

View File

@@ -147,6 +147,8 @@ export const analyticsMiddleware = async (c: Context<HonoEnv>, next: Next) => {
user_id: ctx.userId || null,
};
ctx.customerId = customerId;
// Update logger with enriched context
ctx.logger = ctx.logger.child({
context: {

View File

@@ -0,0 +1,36 @@
import { ErrCode, RecaseError } from "@autumn/shared";
import type { Context, Next } from "hono";
import { redis } from "@/external/redis/initRedis.js";
import type { HonoEnv } from "@/honoUtils/HonoEnv.js";
import { tryRedisWrite } from "@/utils/cacheUtils/cacheUtils";
/**
* Middleware that checks for idempotence in a request
*/
export const idempotencyMiddleware = async (
c: Context<HonoEnv>,
next: Next,
) => {
const headers = c.req.header();
const ctx = c.get("ctx");
const idempotencyKey =
headers["idempotency-key"] || headers["Idempotency-Key"];
if (idempotencyKey) {
const redisKey = `${ctx.org.id}:${ctx.env}:idempotency:${idempotencyKey}`;
// Use SET NX (set if not exists) for atomic check-and-set to prevent race conditions
const wasSet = await tryRedisWrite(() => {
return redis.set(redisKey, "1", "PX", 1000 * 60 * 60 * 24, "NX"); // 24 hours, only set if not exists
});
if (!wasSet) {
throw new RecaseError({
message: `Another request with idempotency key ${idempotencyKey} has already been received`,
code: ErrCode.DuplicateIdempotencyKey,
statusCode: 409,
});
}
}
await next();
};

View File

@@ -82,6 +82,12 @@ export const rateLimitMiddleware = async (c: Context<HonoEnv>, next: Next) => {
const ctx = c.get("ctx");
try {
if (
process.env.NODE_ENV === "development" &&
ctx.org?.id === process.env.TESTS_ORG_ID
) {
return await next();
}
// 1. Determine rate limit type based on endpoint
const rateLimitType = getRateLimitType(c);

View File

@@ -95,13 +95,13 @@ export const refreshCacheMiddleware = async (
customerId,
orgId: org.id,
env: env,
logger,
source: `refreshCacheMiddleware, url: ${pathname}`,
});
await deleteCachedFullCustomer({
customerId,
orgId: org.id,
env: env,
ctx,
source: "refreshCacheMiddleware",
logger,
});
}
return;
@@ -128,10 +128,8 @@ export const refreshCacheMiddleware = async (
});
await deleteCachedFullCustomer({
customerId: body.customer_id,
orgId: org.id,
env: env,
ctx,
source: "refreshCacheMiddleware",
logger,
});
}
}

View File

@@ -39,6 +39,7 @@ export type RequestContext = {
skipCacheDeletion?: boolean;
// Optional (should be populated in Stripe customer?)
customerId?: string;
};
export type AutumnContext = RequestContext;

View File

@@ -1,9 +1,9 @@
// Suppress BullMQ eviction policy warnings BEFORE any imports
// Skip OpenTelemetry instrumentation in development for faster startup
await import("./sentry.js");
if (process.env.NODE_ENV !== "development") {
await import("./instrumentation.js");
await import("./sentry.js");
}
import cluster from "node:cluster";

View File

@@ -43,6 +43,8 @@ const ALLOWED_HEADERS = [
"If-None-Match",
"If-Modified-Since",
"If-Unmodified-Since",
"idempotency-key",
"Idempotency-Key",
];
export const createHonoApp = () => {

View File

@@ -73,9 +73,11 @@ export const handleUpdateBalance = createRoute({
});
await deleteCachedApiCustomer({
logger: ctx.logger,
orgId: ctx.org.id,
env: ctx.env,
customerId: params.customer_id,
source: `handleUpdateBalance, updating next_reset_at`,
});
}

View File

@@ -381,6 +381,8 @@ export const runDeductionTx = async (
customerId: fullCus.id ?? "",
orgId: ctx.org.id,
env: ctx.env,
logger: ctx.logger,
source: `runDeductionTx, refreshing cache`,
});
}

View File

@@ -111,6 +111,7 @@ export const updateGrantedBalance = async ({
env: ctx.env,
customerId: fullCus.id ?? "",
source: "updateGrantedBalance",
logger: ctx.logger,
});
// // Update Redis cache directly (avoids clearing cache which causes race conditions)

View File

@@ -159,6 +159,7 @@ export class SyncBatchingManager {
payload: {
orgId: item.orgId,
env: item.env,
customerId: item.customerId,
item, // Single item
},
messageGroupId: customerId, // FIFO ordering per customer

View File

@@ -14,18 +14,9 @@ export const runSyncBalanceBatch = async ({
payload,
}: {
ctx?: AutumnContext;
payload: SyncBatchPayload | any; // Allow any for backwards compatibility
payload: SyncBatchPayload;
}) => {
// Backwards compatibility: handle both old (items array) and new (item object) formats
let item = payload.item;
// Old format: { items: [...] }
if (!item && payload.items && Array.isArray(payload.items)) {
item = payload.items[0];
console.warn(
"⚠️ Received old format sync message with items array, using first item",
);
}
const item = payload.item;
if (!item || !ctx) {
console.warn("⚠️ No sync item provided");

View File

@@ -11,7 +11,6 @@ import { getRegionalRedis } from "@/external/redis/initRedis.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { deleteCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.js";
import { getCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/getCachedFullCustomer.js";
import type { Logger } from "../../../../external/logtail/logtailUtils";
const SYNC_CONFLICT_CODES = {
ResetAtMismatch: "RESET_AT_MISMATCH",
@@ -24,15 +23,11 @@ const SYNC_CONFLICT_CODES = {
const handleSyncPostgresError = async ({
error,
customerId,
orgId,
env,
logger,
ctx,
}: {
error: Error;
customerId: string;
orgId: string;
env: string;
logger: Logger;
ctx: AutumnContext;
}): Promise<boolean> => {
const message = error.message || "";
const isConflict =
@@ -50,17 +45,15 @@ const handleSyncPostgresError = async ({
const cusEntMatch = message.match(/cus_ent_id:(\S+)/);
const cusEntId = cusEntMatch?.[1];
logger.warn(
ctx.logger.warn(
`[SYNC V3] (${customerId}) Sync conflict detected: ${code}, cus_ent: ${cusEntId}. Clearing cache.`,
);
// Clear the stale cache
await deleteCachedFullCustomer({
customerId,
orgId,
env,
ctx,
source: `sync-conflict-${code}`,
logger,
});
return true;
@@ -247,9 +240,7 @@ export const syncItemV3 = async ({
await handleSyncPostgresError({
error,
customerId,
orgId,
env,
logger,
ctx,
});
return;
}

View File

@@ -72,6 +72,7 @@ export const initCusProduct = ({
trialEndsAt,
subscriptionStatus,
canceledAt,
endedAt,
createdAt,
collectionMethod,
subscriptionIds,
@@ -95,6 +96,7 @@ export const initCusProduct = ({
trialEndsAt?: number | null;
subscriptionStatus?: CusProductStatus;
canceledAt?: number | null;
endedAt?: number | null;
createdAt?: number | null;
collectionMethod?: CollectionMethod;
subscriptionIds?: string[];
@@ -121,7 +123,7 @@ export const initCusProduct = ({
product_id: product.id,
created_at: createdAt || Date.now(),
canceled: notNullish(canceledAt),
ended_at: endedAt,
status: subscriptionStatus
? subscriptionStatus
: isFuture
@@ -280,6 +282,7 @@ export const createFullCusProduct = async ({
trialEndsAt,
subscriptionStatus,
canceledAt = null,
endedAt,
createdAt = null,
subscriptionIds = [],
subscriptionScheduleIds = [],
@@ -302,6 +305,7 @@ export const createFullCusProduct = async ({
trialEndsAt?: number;
subscriptionStatus?: CusProductStatus;
canceledAt?: number | null;
endedAt?: number | null;
createdAt?: number | null;
subscriptionIds?: string[];
subscriptionScheduleIds?: string[];
@@ -464,6 +468,7 @@ export const createFullCusProduct = async ({
entityId: attachParams.entityId,
apiVersion: attachParams.apiVersion,
quantity: productOptions?.quantity ?? undefined,
endedAt,
});
if (!isOneOff(prices) && !product.is_add_on) {
@@ -541,7 +546,7 @@ export const createFullCusProduct = async ({
}
await queueVerifyCacheConsistencyWorkflow({
newCustomerProductId: cusProdId,
newCustomerProduct: fullCusProduct,
previousFullCustomer: attachParams.customer as FullCustomer,
logger,
source: "createFullCusProduct",

View File

@@ -192,7 +192,7 @@ export const updateOneTimeCusProduct = async ({
});
await queueVerifyCacheConsistencyWorkflow({
newCustomerProductId: existingCusProduct.id,
newCustomerProduct: existingCusProduct,
previousFullCustomer: attachParams.customer as FullCustomer,
logger,
source: "updateOneTimeCusProduct",

View File

@@ -109,7 +109,6 @@ export const handlePaidProduct = async ({
itemSet: newItemSet,
config,
branch,
fromCreate: true,
});
sub = updatedSub;

View File

@@ -107,8 +107,6 @@ export const handleMultiAttachFlow = async ({
curSub: curSub!,
itemSet,
branch,
// fromCreate: attachParams.products.length === 0, // just for now, if no products, it comes from cancel product...
fromCreate: true, // just for now, if no products, it comes from cancel product...
});
// TODO: Add these missing functions or remove if not needed

View File

@@ -42,11 +42,13 @@ export const handleUpgradeFlow = async ({
attachParams,
config,
branch,
fromMigration = false,
}: {
ctx: AutumnContext;
attachParams: AttachParams;
config: AttachConfig;
branch: AttachBranch;
fromMigration?: boolean;
}) => {
const curCusProduct = attachParamsToCurCusProduct({ attachParams });
@@ -84,7 +86,8 @@ export const handleUpgradeFlow = async ({
if (
product.is_add_on ||
branch === AttachBranch.NewVersion ||
branch === AttachBranch.SameCustomEnts
branch === AttachBranch.SameCustomEnts ||
fromMigration
)
continue;
@@ -136,18 +139,8 @@ export const handleUpgradeFlow = async ({
curSub: curSub,
itemSet,
branch,
fromCreate: attachParams.products.length === 0, // just for now, if no products, it comes from cancel product...
});
// // Renew sub
// console.log("Sub is canceled!", isStripeSubscriptionCanceled({ sub: res.updatedSub }));
// if (isStripeSubscriptionCanceled({ sub: res.updatedSub })) {
// await attachParams.stripeCli.subscriptions.update(res.updatedSub.id, {
// cancel_at_period_end: false,
// cancel_at: null,
// });
// }
if (res?.latestInvoice) {
logger.info(`UPGRADE FLOW: inserting invoice ${res.latestInvoice.id}`);
await insertInvoiceFromAttach({
@@ -175,6 +168,9 @@ export const handleUpgradeFlow = async ({
config,
schedule,
curSub,
removeCusProducts:
fromMigration && curCusProduct ? [curCusProduct!] : undefined,
addNewProducts: !(fromMigration && curCusProduct?.canceled_at),
});
}
@@ -219,12 +215,18 @@ export const handleUpgradeFlow = async ({
const anchorToUnix = sub ? getEarliestPeriodEnd({ sub }) * 1000 : undefined;
let canceledAt: number | undefined;
let endedAt: number | undefined;
if (sub && isStripeSubscriptionCanceled({ sub })) {
canceledAt = sub.canceled_at
? sub.canceled_at * 1000
: curCusProduct?.canceled_at || undefined;
}
if (fromMigration && curCusProduct?.canceled_at) {
canceledAt = curCusProduct.canceled_at;
endedAt = curCusProduct.ended_at ?? undefined;
}
await createFullCusProduct({
db,
attachParams: attachToInsertParams(
@@ -238,6 +240,7 @@ export const handleUpgradeFlow = async ({
anchorToUnix: anchorToUnix,
scenario: AttachScenario.Upgrade,
canceledAt: canceledAt,
endedAt: endedAt,
subscriptionStatus:
sub?.status === "past_due" ? CusProductStatus.PastDue : undefined,
logger,
@@ -251,25 +254,4 @@ export const handleUpgradeFlow = async ({
? attachToInvoiceResponse({ invoice: latestInvoice || undefined })
: undefined,
});
// if (res) {
// if (req.apiVersion.gte(ApiVersion.V1_1)) {
// res.status(200).json(
// AttachResultSchema.parse({
// customer_id: attachParams.customer.id,
// product_ids: attachParams.products.map((p) => p.id),
// invoice: attachParams.invoiceOnly
// ? attachToInvoiceResponse({ invoice: latestInvoice || undefined })
// : undefined,
// code: "updated_product_successfully",
// message: `Successfully updated product`,
// }),
// );
// } else {
// res.status(200).json({
// success: true,
// message: `Successfully updated product`,
// });
// }
// }
};

View File

@@ -20,6 +20,7 @@ export const handleUpgradeFlowSchedule = async ({
curSub,
removeCusProducts,
fromAddProduct = false,
addNewProducts = true,
}: {
ctx: AutumnContext;
attachParams: AttachParams;
@@ -28,6 +29,7 @@ export const handleUpgradeFlowSchedule = async ({
curSub: Stripe.Subscription;
removeCusProducts?: FullCusProduct[];
fromAddProduct?: boolean;
addNewProducts?: boolean;
}) => {
const { logger } = ctx;
@@ -57,8 +59,11 @@ export const handleUpgradeFlowSchedule = async ({
config,
billingPeriodEnd: schedule?.phases?.[nextPhaseIndex]?.start_date,
removeCusProducts,
addNewProducts,
});
// await logPhases({ phases: newItems.phases, db: ctx.db });
// Should release schedule...
const newCurPhaseIndex = getCurrentPhaseIndex({
schedule: { phases: newItems.phases } as any,

View File

@@ -1,5 +1,5 @@
import {
type AttachBranch,
AttachBranch,
type AttachConfig,
type AttachReplaceable,
ProrationBehavior,
@@ -31,7 +31,6 @@ export const updateStripeSub2 = async ({
curSub,
itemSet,
branch,
fromCreate = false,
}: {
ctx: AutumnContext;
attachParams: AttachParams;
@@ -39,7 +38,6 @@ export const updateStripeSub2 = async ({
curSub: Stripe.Subscription;
itemSet: ItemSet;
branch: AttachBranch;
fromCreate?: boolean;
}) => {
const { db, logger } = ctx;
@@ -93,9 +91,14 @@ export const updateStripeSub2 = async ({
// cancel_at_period_end: false,
// TODO: will error if sub managed by a schedule
cancel_at_period_end: isStripeSubscriptionCanceled({ sub: curSub })
? false
: undefined,
cancel_at_period_end:
isStripeSubscriptionCanceled({ sub: curSub }) &&
!(
branch === AttachBranch.SameCustomEnts ||
branch === AttachBranch.NewVersion
)
? false
: undefined,
});
let latestInvoice = updatedSub.latest_invoice as Stripe.Invoice | null;
@@ -109,13 +112,6 @@ export const updateStripeSub2 = async ({
};
}
// if (fromCreate) {
// return {
// updatedSub,
// latestInvoice: updatedSub.latest_invoice as Stripe.Invoice,
// };
// }
const curCusProduct = attachParamsToCurCusProduct({ attachParams });
const cusEntIds: string[] = [];
const replaceables: AttachReplaceable[] = [];

View File

@@ -200,15 +200,8 @@ const computeUpdatedScheduleItems = async ({
removeCusProducts ||
getCusProductsToRemove({ attachParams, includeScheduled: true });
// console.log(
// "REMOVING CUS PRODUCTS:",
// cusProductsToRemove?.map((cp) => `${cp.product.id} (E: ${cp.entity_id})`)
// );
const allCusProducts = attachParams.customer.customer_products;
// console.log("New schedule items:");
// logScheduleItems({ items: newScheduleItems, cusProducts: allCusProducts });
for (const cusProduct of cusProductsToRemove) {
newScheduleItems = await removeCusProductFromScheduleItems({
curScheduleItems: baseCurScheduleItems,
@@ -221,9 +214,6 @@ const computeUpdatedScheduleItems = async ({
});
}
// console.log("New schedule items after removing cus products:");
// logScheduleItems({ items: newScheduleItems, cusProducts: allCusProducts });
return newScheduleItems;
};
@@ -236,6 +226,7 @@ export const paramsToScheduleItems = async ({
config,
removeCusProducts,
billingPeriodEnd,
addNewProducts = true,
}: {
ctx: AutumnContext;
sub?: Stripe.Subscription;
@@ -244,11 +235,18 @@ export const paramsToScheduleItems = async ({
config: AttachConfig;
removeCusProducts?: FullCusProduct[];
billingPeriodEnd?: number;
addNewProducts?: boolean;
}) => {
const itemSet = await getStripeSubItems2({
attachParams,
config,
});
const itemSet = addNewProducts
? await getStripeSubItems2({
attachParams,
config,
})
: {
subItems: [],
invoiceItems: [],
usageFeatures: [],
};
let phaseIndex = -1;

View File

@@ -39,6 +39,11 @@ export const getCusProductsToRemove = ({
? products
: [cusProductToProduct({ cusProduct: attachParams.cusProduct! })];
console.log(
"Getting customer products to remove, internal entity ID:",
attachParams.internalEntityId,
);
for (const product of prods) {
// Get cur main and cur same
const { curMainProduct, curSameProduct, curScheduledProduct } =
@@ -120,10 +125,9 @@ export const paramsToSubItems = async ({
? removeCusProducts!
: getCusProductsToRemove({ attachParams });
// console.log(
// "Cus products to remove:",
// cusProductsToRemove.map((cp) => cp.product.name),
// );
ctx.logger.info(
`[paramsToSubItems] Removing cus products: ${cusProductsToRemove.map((cp) => `${cp.product.id} (E: ${cp.entity_id})`).join(", ")}`,
);
const newSubItems = mergeNewSubItems({
itemSet,
@@ -187,6 +191,10 @@ export const paramsToSubItems = async ({
continue;
}
if (printRemoveLogs) {
console.log(`Deleting consumable sub item ${existingSubItem.id}`);
}
newSubItems.push({
id: existingSubItem.id,
deleted: true,

View File

@@ -1,11 +1,16 @@
import * as Sentry from "@sentry/bun";
import { redis } from "@/external/redis/initRedis.js";
import {
getConfiguredRegions,
getRegionalRedis,
redis,
} from "@/external/redis/initRedis.js";
import { CACHE_CUSTOMER_VERSIONS } from "../../../../_luaScripts/cacheConfig";
import { batchDeleteCachedFullCustomers } from "../fullCustomerCacheUtils/batchDeleteCachedFullCustomers";
/**
* Batch delete multiple customer caches in one Redis operation
* Much more efficient than calling deleteCachedApiCustomer multiple times
* Batch delete multiple customer caches in one Redis operation across ALL regions.
* This ensures cache consistency and prevents race conditions where
* a stale cache in another region could be read after deletion.
* @param customers Array of {orgId, env, customerId} to delete
* @returns Number of keys deleted
*/
@@ -30,53 +35,78 @@ export const batchDeleteCachedCustomers = async ({
return 0;
}
// Group customers by orgId to avoid Redis Cluster hash slot errors
// All keys in a Lua script must be in the same hash slot (same {orgId})
const customersByOrg = new Map<string, typeof customers>();
for (const customer of customers) {
const key = customer.orgId;
if (!customersByOrg.has(key)) {
customersByOrg.set(key, []);
}
customersByOrg.get(key)?.push(customer);
}
const regions = getConfiguredRegions();
try {
await batchDeleteCachedFullCustomers({
customers,
source: `batchDeleteCachedCustomers, deleting ${customers.length} customers`,
});
// Delete from all regions in parallel
const regionPromises = regions.map(async (region) => {
const regionalRedis = getRegionalRedis(region);
if (regionalRedis.status !== "ready") {
console.warn(
`Redis not ready for region ${region}, skipping batch delete`,
);
return { region, deletedCount: 0, skipped: true };
}
// Use pipeline to batch all org deletions into one network round trip
const pipeline = regionalRedis.pipeline();
for (const orgCustomers of customersByOrg.values()) {
pipeline.batchDeleteCustomers(
CACHE_CUSTOMER_VERSIONS.LATEST,
JSON.stringify(orgCustomers),
);
}
const results = await pipeline.exec();
// Sum up all deleted counts for this region
let regionDeleted = 0;
if (results) {
for (const [error, result] of results) {
if (error) {
console.error(
`Error in pipeline batch delete for region ${region}:`,
error,
);
throw error;
}
regionDeleted += result as number;
}
}
return { region, deletedCount: regionDeleted, skipped: false };
});
// Group customers by orgId to avoid Redis Cluster hash slot errors
// All keys in a Lua script must be in the same hash slot (same {orgId})
const customersByOrg = new Map<string, typeof customers>();
const regionResults = await Promise.all(regionPromises);
for (const customer of customers) {
const key = customer.orgId;
if (!customersByOrg.has(key)) {
customersByOrg.set(key, []);
}
customersByOrg.get(key)?.push(customer);
}
// Use pipeline to batch all org deletions into one network round trip
const pipeline = redis.pipeline();
for (const orgCustomers of customersByOrg.values()) {
pipeline.batchDeleteCustomers(
CACHE_CUSTOMER_VERSIONS.LATEST,
JSON.stringify(orgCustomers),
);
pipeline.batchDeleteCustomers(
CACHE_CUSTOMER_VERSIONS.PREVIOUS,
JSON.stringify(orgCustomers),
);
}
const results = await pipeline.exec();
// Sum up all deleted counts
let totalDeleted = 0;
if (results) {
for (const [error, result] of results) {
if (error) {
console.error("Error in pipeline batch delete:", error);
throw error;
}
totalDeleted += result as number;
}
}
const totalDeleted = regionResults.reduce(
(sum, r) => sum + r.deletedCount,
0,
);
const regionsSummary = regionResults
.map((r) => `${r.region}: ${r.skipped ? "skipped" : r.deletedCount}`)
.join(", ");
console.log(
`Batch deleted ${totalDeleted} cache keys for ${customers.length} customers across ${customersByOrg.size} orgs`,
`[batchDeleteCache] customers: ${customers.length}, keys: ${totalDeleted}, regions: ${regionsSummary}`,
);
return totalDeleted;

View File

@@ -1,30 +1,27 @@
import { CACHE_CUSTOMER_VERSIONS } from "../../../../_luaScripts/cacheConfig.js";
import {
type Logger,
logger as loggerInstance,
} from "../../../../external/logtail/logtailUtils.js";
import { redis } from "../../../../external/redis/initRedis.js";
getConfiguredRegions,
getRegionalRedis,
redis,
} from "../../../../external/redis/initRedis.js";
import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
import { deleteCachedFullCustomer } from "../fullCustomerCacheUtils/deleteCachedFullCustomer.js";
/**
* Delete all cached ApiCustomer data from Redis
* This includes the base customer key and all related feature/breakdown/rollover keys
* Also deletes all associated entity caches atomically using Lua script
* Delete all cached ApiCustomer data from Redis across ALL regions.
* This ensures cache consistency and prevents race conditions where
* a stale cache in another region could be read after deletion.
*/
export const deleteCachedApiCustomer = async ({
customerId,
orgId,
env,
ctx,
source,
logger,
}: {
customerId: string;
orgId: string;
env: string;
ctx: AutumnContext;
source?: string;
logger?: Logger;
}): Promise<void> => {
logger = loggerInstance || loggerInstance;
const { org, env, logger } = ctx;
if (redis.status !== "ready") {
logger.warn("❗️ Redis not ready, skipping cache deletion", {
@@ -38,34 +35,48 @@ export const deleteCachedApiCustomer = async ({
if (!customerId) return;
const regions = getConfiguredRegions();
try {
const [deletedCount, deletedCountV1_2_0] = await Promise.all([
redis.deleteCustomer(
await deleteCachedFullCustomer({
ctx,
customerId,
source,
});
// Delete from all regions in parallel to avoid race conditions
const deletePromises = regions.map(async (region) => {
const regionalRedis = getRegionalRedis(region);
// Check if this regional instance is ready
if (regionalRedis.status !== "ready") {
logger?.warn(`Redis not ready for region ${region}, skipping`, {
data: { status: regionalRedis.status, customerId, region },
});
return { region, deletedCount: 0, skipped: true };
}
const deletedCount = await regionalRedis.deleteCustomer(
CACHE_CUSTOMER_VERSIONS.LATEST,
orgId,
org.id,
env,
customerId,
),
redis.deleteCustomer(
CACHE_CUSTOMER_VERSIONS.PREVIOUS,
orgId,
env,
customerId,
),
deleteCachedFullCustomer({
customerId,
orgId,
env,
source,
logger,
}),
]);
);
return { region, deletedCount, skipped: false };
});
const results = await Promise.all(deletePromises);
const totalDeleted = results.reduce(
(sum, r) => sum + (r.deletedCount || 0),
0,
);
const regionsSummary = results
.map((r) => `${r.region}: ${r.skipped ? "skipped" : r.deletedCount}`)
.join(", ");
logger.info(
`Deleted ${deletedCount} cache keys for customer ${customerId}, source: ${source}`,
);
logger.info(
`Deleted ${deletedCountV1_2_0} cache keys (v1.2.0) for customer ${customerId}, source: ${source}`,
`Deleted cache keys for customer ${customerId}. Source: ${source}, keys: ${totalDeleted}, regions: ${regions.length} (${regionsSummary})`,
);
} catch (error) {
logger.error(`Error deleting customer with entities: ${error}`);

View File

@@ -85,7 +85,6 @@ export const getApiSubscription = async ({
plan: apiPlan,
plan_id: fullProduct.id,
plan_version: fullProduct.version,
add_on: fullProduct.is_add_on,
default: fullProduct.is_default,

View File

@@ -28,7 +28,6 @@ import {
import { getDefaultAttachConfig } from "../attach/attachUtils/getAttachConfig.js";
import { CusService } from "../CusService.js";
import { initStripeCusAndProducts } from "../handlers/handleCreateCustomer.js";
import { deleteCachedApiCustomer } from "./apiCusCacheUtils/deleteCachedApiCustomer.js";
export const getGroupToDefaultProd = async ({
defaultProds,
@@ -215,12 +214,5 @@ export const createNewCustomer = async ({
}
}
// // Clear the customer cache here
// await deleteCachedApiCustomer({
// customerId: newCustomer.id || newCustomer.internal_id,
// orgId: ctx.org.id,
// env: ctx.env,
// });
return newCustomer;
};

View File

@@ -1,8 +1,5 @@
import {
type Logger,
logger as loggerInstance,
} from "@/external/logtail/logtailUtils.js";
import { redis } from "@/external/redis/initRedis.js";
import type { AutumnContext } from "../../../../honoUtils/HonoEnv.js";
import {
buildFullCustomerCacheGuardKey,
buildFullCustomerCacheKey,
@@ -16,21 +13,17 @@ import { buildTestFullCustomerCacheGuardKey } from "./testFullCustomerCacheGuard
*/
export const deleteCachedFullCustomer = async ({
customerId,
orgId,
env,
ctx,
source,
logger,
}: {
customerId: string;
orgId: string;
env: string;
ctx: AutumnContext;
source?: string;
logger?: Logger;
}): Promise<void> => {
const log = logger || loggerInstance;
const { org, env, logger } = ctx;
if (redis.status !== "ready") {
log.warn(
logger.warn(
`[deleteCachedFullCustomer] Redis not ready, skipping deletion for ${customerId}`,
);
return;
@@ -39,12 +32,20 @@ export const deleteCachedFullCustomer = async ({
if (!customerId) return;
const testGuardKey = buildTestFullCustomerCacheGuardKey({
orgId,
orgId: org.id,
env,
customerId,
});
const cacheKey = buildFullCustomerCacheKey({
orgId: org.id,
env,
customerId,
});
const guardKey = buildFullCustomerCacheGuardKey({
orgId: org.id,
env,
customerId,
});
const cacheKey = buildFullCustomerCacheKey({ orgId, env, customerId });
const guardKey = buildFullCustomerCacheGuardKey({ orgId, env, customerId });
try {
const guardTimestamp = Date.now().toString();
@@ -58,20 +59,20 @@ export const deleteCachedFullCustomer = async ({
);
if (result === "SKIPPED") {
log.info(
logger.info(
`[deleteCachedFullCustomer] Test guard exists, skipping deletion for ${customerId}`,
);
} else if (result === "DELETED") {
log.info(
logger.info(
`[deleteCachedFullCustomer] Deleted cache for ${customerId}, source: ${source}`,
);
} else {
log.debug(
logger.debug(
`[deleteCachedFullCustomer] Cache key didn't exist for ${customerId}, source: ${source}`,
);
}
} catch (error) {
log.error(`[deleteCachedFullCustomer] Error: ${error}`);
logger.error(`[deleteCachedFullCustomer] Error: ${error}`);
throw error;
}
};

View File

@@ -12,9 +12,9 @@ import { autoCreateEntity } from "../../entities/handlers/handleCreateEntity/aut
import { CusService } from "../CusService.js";
import { handleCreateCustomer } from "../handlers/handleCreateCustomer.js";
import { getApiCustomerBase } from "./apiCusUtils/getApiCustomerBase.js";
import { updateCustomerDetails } from "./cusUtils.js";
import { deleteCachedFullCustomer } from "./fullCustomerCacheUtils/deleteCachedFullCustomer.js";
import { getOrSetCachedFullCustomer } from "./fullCustomerCacheUtils/getOrSetCachedFullCustomer.js";
import { updateCustomerDetails } from "./cusUtils.js";
export const getOrCreateApiCustomer = async ({
ctx,
@@ -159,8 +159,7 @@ export const getOrCreateApiCustomer = async ({
if (updated) {
await deleteCachedFullCustomer({
customerId: apiCustomer.id || "",
orgId: ctx.org.id,
env: ctx.env,
ctx,
source: "getOrCreateApiCustomer",
});
const fullCus = await getOrSetCachedFullCustomer({
@@ -196,8 +195,7 @@ export const getOrCreateApiCustomer = async ({
await deleteCachedFullCustomer({
customerId,
orgId: ctx.org.id,
env: ctx.env,
ctx,
source: "getOrCreateApiCustomer",
});

View File

@@ -17,6 +17,8 @@ export const handleClearCustomerCache = createRoute({
customerId: customer_id,
orgId: ctx.org.id,
env: ctx.env,
source: `handleClearCustomerCache, deleting single customer cache`,
logger: ctx.logger,
});
}

View File

@@ -65,6 +65,8 @@ export const deleteCusById = async ({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `deleteCusById, deleting customer cache`,
logger: ctx.logger,
});
return response;
@@ -130,6 +132,8 @@ export const handleDeleteCustomerV2 = createRoute({
customerId: customer.id ?? "",
orgId: org.id,
env,
source: `handleDeleteCustomerV2, deleting customer cache`,
logger: ctx.logger,
});
return c.json(response);

View File

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

View File

@@ -71,6 +71,7 @@ export const handleUpdateCusEntitlementV2 = createRoute({
env: ctx.env,
customerId: customer_id,
source: "handleUpdateBalance",
logger: ctx.logger,
});
return c.json({ success: true });

View File

@@ -59,6 +59,8 @@ export const getCachedApiEntity = async ({
entityId,
source: "getCachedApiEntity",
});
fullCus.entity = fullCus.entities.find((e) => e.id === entityId);
}
const entity = fullCus.entity;

View File

@@ -1,19 +1,18 @@
import {
AppEnv,
migrationErrors,
MigrationJob,
type AppEnv,
ErrCode,
type MigrationJob,
MigrationJobStep,
migrationErrors,
migrationJobs,
} from "@autumn/shared";
import { DrizzleCli } from "@/db/initDrizzle.js";
import { migrationJobs } from "@autumn/shared";
import RecaseError from "@/utils/errorUtils.js";
import { ErrCode } from "@autumn/shared";
import { and, eq, ne } from "drizzle-orm";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import RecaseError from "@/utils/errorUtils.js";
export class MigrationService {
static async createJob({ db, data }: { db: DrizzleCli; data: MigrationJob }) {
let result = await db.insert(migrationJobs).values(data).returning();
const result = await db.insert(migrationJobs).values(data).returning();
if (result.length === 0) {
throw new RecaseError({
@@ -34,7 +33,7 @@ export class MigrationService {
migrationJobId: string;
updates: any;
}) {
let results = await db
const results = await db
.update(migrationJobs)
.set({
...updates,
@@ -51,7 +50,7 @@ export class MigrationService {
}
static async getJob({ db, id }: { db: DrizzleCli; id: string }) {
let job = await db.query.migrationJobs.findFirst({
const job = await db.query.migrationJobs.findFirst({
where: eq(migrationJobs.id, id),
});
@@ -74,7 +73,7 @@ export class MigrationService {
orgId: string;
env: AppEnv;
}) {
let jobs = await db.query.migrationJobs.findMany({
const jobs = await db.query.migrationJobs.findMany({
where: and(
eq(migrationJobs.org_id, orgId),
eq(migrationJobs.env, env),
@@ -87,7 +86,7 @@ export class MigrationService {
}
static async insertError({ db, data }: { db: DrizzleCli; data: any }) {
let result = await db.insert(migrationErrors).values(data).returning();
const result = await db.insert(migrationErrors).values(data).returning();
if (result.length === 0) {
throw new RecaseError({
@@ -106,7 +105,7 @@ export class MigrationService {
db: DrizzleCli;
migrationJobId: string;
}) {
let errors = await db.query.migrationErrors.findMany({
const errors = await db.query.migrationErrors.findMany({
where: eq(migrationErrors.migration_job_id, migrationJobId),
with: {
customer: true,

View File

@@ -1,43 +1,43 @@
import {
type AppEnv,
type Feature,
ACTIVE_STATUSES,
type FullCusProduct,
type FullProduct,
type MigrationJob,
type Organization,
ProcessorType,
} from "@autumn/shared";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { createStripeCli } from "@/external/connect/createStripeCli.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { CusService } from "@/internal/customers/CusService.js";
import type { ExtendedRequest } from "@/utils/models/Request.js";
import type { Logger } from "../../../external/logtail/logtailUtils.js";
import { CusProductService } from "../../customers/cusProducts/CusProductService.js";
import { createMigrationCustomerLogger } from "../migrationUtils/createMigrationCustomerLogger.js";
import { migrateRevenueCatCustomer } from "./migrateRevenuecatCustomer.js";
import { migrateStripeCustomer } from "./migrateStripeCustomer.js";
export const migrateCustomer = async ({
db,
ctx,
customerId,
org,
logger,
env,
orgId,
fromProduct,
toProduct,
features,
migrationJob,
}: {
db: DrizzleCli;
ctx: AutumnContext;
customerId: string;
org: Organization;
env: AppEnv;
orgId: string;
fromProduct: FullProduct;
toProduct: FullProduct;
logger: Logger;
features: Feature[];
migrationJob?: MigrationJob;
}) => {
const { db, org, env } = ctx;
const orgId = org.id;
// Create customer-specific logger
const customerLogger = createMigrationCustomerLogger({
ctx,
customerId,
migrationJobId: migrationJob?.id,
});
const customerCtx: AutumnContext = { ...ctx, logger: customerLogger };
try {
const stripeCli = createStripeCli({ org, env });
const fullCus = await CusService.getFull({
@@ -46,57 +46,59 @@ export const migrateCustomer = async ({
orgId,
env,
withEntities: true,
inStatuses: ACTIVE_STATUSES,
});
// 1. Build req object
const req = {
db,
orgId,
env,
org,
features,
logger,
timestamp: Date.now(),
} as ExtendedRequest;
const cusProducts = fullCus.customer_products;
const filteredCusProducts = cusProducts.filter(
(cp: FullCusProduct) =>
cp.product.internal_id === fromProduct.internal_id,
);
for (const cusProduct of filteredCusProducts) {
customerLogger.debug(
`Filtered customer products ${filteredCusProducts.length}`,
);
for (let i = 0; i < filteredCusProducts.length; i++) {
const cusProduct = filteredCusProducts[i];
if (cusProduct.processor?.type === ProcessorType.RevenueCat) {
await migrateRevenueCatCustomer({
req,
ctx: customerCtx,
fullCus,
cusProduct,
toProduct,
customerId,
orgId,
env,
});
} else {
await migrateStripeCustomer({
req,
ctx: customerCtx,
stripeCli,
fullCus,
cusProduct,
toProduct,
fromProduct,
customerId,
orgId,
env,
});
}
// If not last, refresh full customer with new cusProducts
if (i < filteredCusProducts.length - 1) {
const latestCusProducts = await CusProductService.list({
db,
internalCustomerId: fullCus.internal_id,
inStatuses: ACTIVE_STATUSES,
});
fullCus.customer_products = latestCusProducts;
}
}
return true;
} catch (error: any) {
logger.error(
} catch (error) {
customerLogger.error(
`Migration failed for customer ${customerId}, job id: ${migrationJob?.id}`,
);
logger.error(error);
customerLogger.error(error);
return false;
}

View File

@@ -4,35 +4,32 @@ dotenv.config();
import {
type Customer,
type Feature,
type FullProduct,
type MigrationJob,
MigrationJobStep,
} from "@autumn/shared";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { createStripeCli } from "@/external/connect/createStripeCli.js";
import { createStripePriceIFNotExist } from "@/external/stripe/createStripePrice/createStripePrice.js";
import { OrgService } from "@/internal/orgs/OrgService.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { MigrationService } from "../MigrationService.js";
import { migrateCustomer } from "./migrateCustomer.js";
export const migrateCustomers = async ({
db,
ctx,
migrationJob,
fromProduct,
toProduct,
logger,
customers,
features,
}: {
db: DrizzleCli;
ctx: AutumnContext;
migrationJob: MigrationJob;
fromProduct: FullProduct;
toProduct: FullProduct;
logger: any;
customers: Customer[];
features: Feature[];
}) => {
const { db, logger, org } = ctx;
const { env } = migrationJob;
await MigrationService.updateJob({
db,
migrationJobId: migrationJob.id,
@@ -41,14 +38,6 @@ export const migrateCustomers = async ({
},
});
let batchCount = 0;
const { org_id: orgId, env } = migrationJob;
const org = await OrgService.get({
db,
orgId,
});
// Create stripe prices if they don't exist
const stripeCli = createStripeCli({ org, env });
const batchCreate = [];
@@ -77,16 +66,11 @@ export const migrateCustomers = async ({
if (!customer.id) continue;
batchPromises.push(
migrateCustomer({
db,
migrationJob,
customerId: customer.id!,
org,
logger,
env,
orgId,
ctx,
customerId: customer.id,
fromProduct,
toProduct,
features,
migrationJob,
}),
);
}
@@ -130,28 +114,10 @@ export const migrateCustomers = async ({
},
},
});
batchCount++;
}
// Get number of errors
const migrationDetails: any = {};
// try {
// let errors = await MigrationService.getErrors({
// db,
// migrationJobId: migrationJob.id,
// });
// migrationDetails.num_errors = errors!.length;
// migrationDetails.failed_customers = errors!.map(
// (e: any) => `${e.customer.id} - ${e.customer.name}`,
// );
// } catch (error) {
// migrationDetails.failed_to_get_errors = true;
// migrationDetails.error = error;
// logger.error("Failed to get migration errors");
// logger.error(error);
// }
const migrationDetails = {};
const curMigrationJob = await MigrationService.getJob({
db,
@@ -169,10 +135,4 @@ export const migrateCustomers = async ({
},
},
});
// await sendMigrationEmail({
// db,
// migrationJobId: migrationJob.id,
// org,
// });
};

View File

@@ -1,5 +1,4 @@
import {
type AppEnv,
AttachScenario,
CusProductStatus,
type FullCusProduct,
@@ -8,30 +7,27 @@ import {
ProcessorType,
} from "@autumn/shared";
import { createStripeCli } from "@/external/connect/createStripeCli.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { createFullCusProduct } from "@/internal/customers/add-product/createFullCusProduct.js";
import { CusProductService } from "@/internal/customers/cusProducts/CusProductService.js";
import { attachToInsertParams } from "@/internal/products/productUtils.js";
import type { ExtendedRequest } from "@/utils/models/Request.js";
import { deleteCachedApiCustomer } from "../../customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
export const migrateRevenueCatCustomer = async ({
req,
ctx,
fullCus,
cusProduct,
toProduct,
customerId,
orgId,
env,
}: {
req: ExtendedRequest;
ctx: AutumnContext;
fullCus: FullCustomer;
cusProduct: FullCusProduct;
toProduct: FullProduct;
customerId: string;
orgId: string;
env: AppEnv;
}) => {
const { logger } = req;
const { db, logger, org, env, features } = ctx;
fullCus.customer_products = fullCus.customer_products.filter(
(cp) => cp.processor?.type === ProcessorType.RevenueCat,
);
@@ -50,7 +46,7 @@ export const migrateRevenueCatCustomer = async ({
});
await CusProductService.update({
db: req.db,
db,
cusProductId: cusProduct.id,
updates: {
status: CusProductStatus.Expired,
@@ -59,29 +55,11 @@ export const migrateRevenueCatCustomer = async ({
});
const createdAtToPass = cusProduct.created_at;
const startsAtToPass = cusProduct.starts_at;
const anchorToPass = cusProduct.created_at;
// // Debug: Log what we're passing to createFullCusProduct
// logger.info(`[RC Migration] Passing to createFullCusProduct:`, {
// createdAt: createdAtToPass,
// createdAt_date: createdAtToPass
// ? new Date(createdAtToPass).toISOString()
// : null,
// startsAt: startsAtToPass,
// startsAt_date: startsAtToPass
// ? new Date(startsAtToPass).toISOString()
// : null,
// anchorToUnix: anchorToPass,
// anchorToUnix_date: anchorToPass
// ? new Date(anchorToPass).toISOString()
// : null,
// createdAt_type: typeof createdAtToPass,
// });
await createFullCusProduct({
db: req.db,
logger: req.logger,
db,
logger,
scenario: AttachScenario.New,
processorType: ProcessorType.RevenueCat,
// Preserve the original created_at, starts_at, and billing cycle anchor
@@ -95,14 +73,14 @@ export const migrateRevenueCatCustomer = async ({
prices: toProduct.prices,
entitlements: toProduct.entitlements,
entities: fullCus.entities || [],
org: req.org,
stripeCli: createStripeCli({ org: req.org, env: req.env }),
org,
stripeCli: createStripeCli({ org, env }),
paymentMethod: null,
freeTrial: null,
optionsList: cusProduct.options || [],
cusProducts: fullCus.customer_products,
replaceables: [],
features: req.features,
features,
fromMigration: true,
},
toProduct,
@@ -111,7 +89,9 @@ export const migrateRevenueCatCustomer = async ({
await deleteCachedApiCustomer({
customerId,
orgId,
orgId: org.id,
env,
logger,
source: `migrateRevenueCatCustomer, deleting customer cache`,
});
};

View File

@@ -1,39 +1,31 @@
import {
type AppEnv,
type FullCusProduct,
type FullCustomer,
type FullProduct,
} from "@autumn/shared";
import type { FullCusProduct, FullCustomer, FullProduct } from "@autumn/shared";
import type { Stripe } from "stripe";
import type { ExtendedRequest } from "@/utils/models/Request.js";
import type { AutumnContext } from "../../../honoUtils/HonoEnv.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { deleteCachedApiCustomer } from "../../customers/cusUtils/apiCusCacheUtils/deleteCachedApiCustomer.js";
import { migrationToAttachParams } from "../migrationUtils/migrationToAttachParams.js";
import { runMigrationAttach } from "../migrationUtils/runMigrationAttach.js";
export const migrateStripeCustomer = async ({
req,
ctx,
stripeCli,
fullCus,
cusProduct,
toProduct,
fromProduct,
customerId,
orgId,
env,
}: {
req: ExtendedRequest;
ctx: AutumnContext;
stripeCli: Stripe;
fullCus: FullCustomer;
cusProduct: FullCusProduct;
toProduct: FullProduct;
fromProduct: FullProduct;
customerId: string;
orgId: string;
env: AppEnv;
}) => {
const { org, env, logger } = ctx;
const attachParams = await migrationToAttachParams({
req,
ctx,
stripeCli,
customer: fullCus,
cusProduct,
@@ -41,14 +33,16 @@ export const migrateStripeCustomer = async ({
});
await runMigrationAttach({
ctx: req as unknown as AutumnContext,
ctx,
attachParams,
fromProduct,
});
await deleteCachedApiCustomer({
customerId,
orgId,
orgId: org.id,
env,
logger,
source: `migrateStripeCustomer, deleting customer cache`,
});
};

View File

@@ -0,0 +1,28 @@
import { AuthType } from "@autumn/shared";
import type { Logger } from "@/external/logtail/logtailUtils.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
export const createMigrationCustomerLogger = ({
ctx,
customerId,
migrationJobId,
}: {
ctx: AutumnContext;
customerId: string;
migrationJobId?: string;
}): Logger => {
const { logger, org, env } = ctx;
return logger.child({
context: {
context: {
migration_job_id: migrationJobId,
org_id: org.id,
org_slug: org.slug,
customer_id: customerId,
env: env,
authType: AuthType.Worker,
},
},
});
};

View File

@@ -1,38 +1,35 @@
import type { FullCusProduct, FullCustomer, FullProduct } from "@autumn/shared";
import type Stripe from "stripe";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { getStripeCusData } from "@/internal/customers/attach/attachUtils/attachParams/attachParamsUtils/getStripeCusData.js";
import type { AttachParams } from "@/internal/customers/cusProducts/AttachParams.js";
import type { ExtendedRequest } from "@/utils/models/Request.js";
import type { AutumnContext } from "../../../honoUtils/HonoEnv";
export const migrationToAttachParams = async ({
req,
ctx,
stripeCli,
customer,
cusProduct,
newProduct,
}: {
req: ExtendedRequest;
ctx: AutumnContext;
stripeCli: Stripe;
customer: FullCustomer;
cusProduct: FullCusProduct;
newProduct: FullProduct;
}): Promise<AttachParams> => {
const { org, logger } = req;
const { db, org, env, logger, features } = ctx;
const internalEntityId = cusProduct.internal_entity_id || undefined;
const { stripeCus, paymentMethod, now } = await getStripeCusData({
stripeCli,
db: req.db,
db,
org,
env: req.env,
env,
customer,
logger,
allowNoStripe: true,
});
const ctx = req as unknown as AutumnContext;
const attachParams: AttachParams = {
stripeCli,
stripeCus,
@@ -50,8 +47,11 @@ export const migrationToAttachParams = async ({
req: ctx,
org,
entities: customer.entities,
features: ctx.features,
features,
internalEntityId,
entityId:
customer.entities?.find((e) => e.internal_id === internalEntityId)?.id ||
undefined,
cusProducts: customer.customer_products,
// Others

View File

@@ -67,7 +67,7 @@ export const runMigrationAttach = async ({
const customer = attachParams.customer;
logger.info(`--------------------------------`);
logger.info(
`Running migration for ${customer.id}, function: ${attachFunction}`,
`Running migration for ${customer.id} (E: ${attachParams.entityId || "N/A"}), function: ${attachFunction}`,
);
let sameCustomBranch: AttachBranch | undefined;
@@ -97,6 +97,7 @@ export const runMigrationAttach = async ({
ctx,
attachParams,
config,
fromMigration: true,
branch:
sameCustomBranch === AttachBranch.SameCustomEnts
? AttachBranch.SameCustomEnts

View File

@@ -1,26 +1,27 @@
/** biome-ignore-all lint/suspicious/noExplicitAny: ok */
import { MigrationJobStep } from "@autumn/shared";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { FeatureService } from "../features/FeatureService.js";
import { type AppEnv, MigrationJobStep } from "@autumn/shared";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { ProductService } from "../products/ProductService.js";
import { MigrationService } from "./MigrationService.js";
import { getMigrationCustomers } from "./migrationSteps/getMigrationCustomers.js";
import { migrateCustomers } from "./migrationSteps/migrateCustomers.js";
export interface MigrationTaskPayload {
migrationJobId: string;
orgId: string;
env: AppEnv;
}
export const runMigrationTask = async ({
db,
ctx,
payload,
logger,
}: {
db: DrizzleCli;
payload: any;
logger: any;
ctx: AutumnContext;
payload: MigrationTaskPayload;
}) => {
const { db, logger } = ctx;
const { migrationJobId } = payload;
try {
logger.info(`Running migration task, ID: ${migrationJobId}`);
const migrationJob = await MigrationService.getJob({
db,
id: migrationJobId,
@@ -66,26 +67,18 @@ export const runMigrationTask = async ({
fromProduct,
});
const features = await FeatureService.list({
db,
orgId,
env,
});
logger.info(`Job ${migrationJobId} | Found ${customers?.length} customers`);
logger.info(
`Running migration for org ${ctx.org.id}, from ${fromProduct.name} to ${toProduct.name}`,
);
// STEP 2: MIGRATE CUSTOMERS..
await migrateCustomers({
db,
ctx,
migrationJob,
fromProduct,
toProduct,
customers,
logger,
features,
});
// await new Promise((resolve) => setTimeout(resolve, 10000));
} catch (error) {
logger.error(`Migration failed: ${migrationJobId}`);
logger.error(error);

View File

@@ -147,6 +147,8 @@ export const handleMigrateProductV2 = createRoute({
jobName: JobName.Migration,
payload: {
migrationJobId: migrationJob.id,
orgId: org.id,
env,
},
});

View File

@@ -150,6 +150,8 @@ export const triggerFreeProduct = async ({
customerId: fullRedeemer.id!,
orgId: org.id,
env,
logger,
source: `triggerFreeProduct, deleting redeemer cache`,
});
}
@@ -168,6 +170,8 @@ export const triggerFreeProduct = async ({
customerId: fullReferrer.id!,
orgId: org.id,
env,
logger,
source: `triggerFreeProduct, deleting referrer cache`,
});
logger.info(`✅ Added ${fullProduct.name} to referrer`);
}

View File

@@ -89,6 +89,8 @@ const handleRefreshCache = async (req: any, res: any) => {
customerId,
orgId: req.org.id,
env: req.env,
logger: logger,
source: `refreshCacheMiddleware, url: ${req.originalUrl} (express)`,
});
}
@@ -105,6 +107,7 @@ const handleRefreshCache = async (req: any, res: any) => {
customerId: req.body.customer_id,
orgId: req.org.id,
env: req.env,
logger: logger,
});
}
};

View File

@@ -1,4 +1,4 @@
import { type Job, Worker } from "bullmq";
import { type ConnectionOptions, type Job, Worker } from "bullmq";
import type { Logger } from "pino";
import { type DrizzleCli, initDrizzle } from "@/db/initDrizzle.js";
import { logger } from "@/external/logtail/logtailUtils.js";
@@ -42,6 +42,7 @@ const initWorker = ({ id, db }: { id: number; db: DrizzleCli }) => {
const ctx = await createWorkerContext({
db,
logger: workerLogger,
payload: job.data,
});
try {
@@ -64,10 +65,13 @@ const initWorker = ({ id, db }: { id: number; db: DrizzleCli }) => {
}
if (job.name === JobName.Migration) {
if (!ctx) {
workerLogger.error("No context found for migration job");
return;
}
await runMigrationTask({
db,
ctx,
payload: job.data,
logger: workerLogger,
});
return;
}
@@ -125,7 +129,7 @@ const initWorker = ({ id, db }: { id: number; db: DrizzleCli }) => {
}
},
{
connection: workerRedis,
connection: workerRedis as ConnectionOptions,
concurrency: 1,
removeOnComplete: {
count: 0,

View File

@@ -7,24 +7,27 @@ import { generateId } from "../utils/genUtils.js";
export const createWorkerContext = async ({
db,
orgId,
env,
// features,
payload,
logger,
workflowId,
}: {
db: DrizzleCli;
orgId?: string;
env?: AppEnv;
// features: Feature[];
payload: {
orgId?: string;
env?: AppEnv;
customerId?: string;
};
logger: Logger;
workflowId?: string;
}) => {
const { orgId, env, customerId } = payload;
if (!orgId || !env) return;
// Fetch org with features once for all items
const orgData = await OrgService.getWithFeatures({
db,
orgId,
env: env as AppEnv,
env,
});
if (!orgData) {
@@ -36,8 +39,10 @@ export const createWorkerContext = async ({
const workerLogger = logger.child({
context: {
context: {
workflow_id: workflowId,
org_id: org?.id,
org_slug: org?.slug,
customer_id: customerId,
env: env,
authType: AuthType.Worker,
},
@@ -51,12 +56,12 @@ export const createWorkerContext = async ({
db,
logger: workerLogger,
id: generateId("job"),
id: workflowId || generateId("job"),
timestamp: Date.now(),
isPublic: false,
authType: AuthType.Unknown,
apiVersion: createdAtToVersion({ createdAt: org.created_at! }),
clickhouseClient: null as any,
clickhouseClient: undefined,
expand: [],
skipCache: true,
};

View File

@@ -36,14 +36,19 @@ export const createWorkflowTask = <TInput extends BaseWorkflowInput, TOutput>({
const workflowName = hatchetCtx.workflowName();
const taskName = hatchetCtx.taskName();
const name = `${workflowName}/${taskName}`;
const workflowMetadata = hatchetCtx.additionalMetadata();
const autumnContext = await createWorkerContext({
db,
orgId,
env,
payload: input,
logger,
workflowId: workflowMetadata.workflowId,
});
autumnContext?.logger.info(
`[${workflowName}] Running for customer ${customerId}, orgId: ${orgId}`,
);
if (!autumnContext) {
const error = new Error(
`[${name}] Failed to create worker context for org: ${orgId}`,

View File

@@ -1,4 +1,5 @@
import {
CusProductStatus,
cusProductsToCusEnts,
isBooleanCusEnt,
isContUseFeature,
@@ -33,7 +34,7 @@ export const checkForMisingBalance = async ({
(cp) => cp.id === newCustomerProductId,
);
if (!cusProduct) return;
if (!cusProduct || cusProduct.status === CusProductStatus.Scheduled) return;
const cusEnts = cusProductsToCusEnts({
cusProducts: [cusProduct],
@@ -94,12 +95,13 @@ export const checkForMisingBalance = async ({
const threshold = grantedBalanceIncrease.mul(0.995);
if (grantedBalanceIncrease.gt(0) && usageIncrease.gte(threshold)) {
const errMessage = `[RACE CONDITION DETECTED] Usage increase (${usageIncrease}), granted balance increase (${grantedBalanceIncrease.toNumber()}), feature (${feature.name}), customer (${fullCustomer.id})`;
const errMessage = `[RACE CONDITION] Usage increase (${usageIncrease}), granted balance increase (${grantedBalanceIncrease.toNumber()}), feature (${feature.name}), customer (${fullCustomer.id}), product: ${cusProduct.product?.name}`;
Sentry.captureException(new Error(errMessage), {
Sentry.captureException(errMessage, {
tags: getSentryTags({
ctx,
customerId: fullCustomer.id || "",
alert: true,
}),
});
@@ -111,6 +113,8 @@ export const checkForMisingBalance = async ({
newUsage: newUsage.toNumber(),
previousTotalBalance: previousTotalBalance.toNumber(),
newTotalBalance: newTotalBalance.toNumber(),
previousCustomer: previousApiCustomer,
newCustomer: newApiCustomer,
},
});
}

View File

@@ -1,30 +1,45 @@
import type { FullCustomer } from "../../../../../shared";
import {
cusProductToPrices,
type FullCusProduct,
type FullCustomer,
isFreeProduct,
} from "@autumn/shared";
import type { Logger } from "../../../external/logtail/logtailUtils";
import { generateId } from "../../../utils/genUtils";
import { JobName } from "../../JobName";
import { runHatchetWorkflow } from "../../queueUtils";
export const queueVerifyCacheConsistencyWorkflow = async ({
newCustomerProductId,
newCustomerProduct,
previousFullCustomer,
logger,
source,
}: {
newCustomerProductId: string;
newCustomerProduct: FullCusProduct;
previousFullCustomer: FullCustomer;
logger: Logger;
source: string;
}) => {
const workflowId = generateId("workflow");
logger.info(
`[${source}] Scheduling verify cache workflow for customer ${previousFullCustomer.id || previousFullCustomer.internal_id}`,
`[${source}] Scheduling verify cache workflow for customer ${previousFullCustomer.id || previousFullCustomer.internal_id}, workflowId: ${workflowId}`,
);
try {
// 1. Check if new customer product is not free
const newPrices = cusProductToPrices({ cusProduct: newCustomerProduct });
if (isFreeProduct({ prices: newPrices })) return;
await runHatchetWorkflow({
workflowName: JobName.VerifyCacheConsistency,
metadata: {
workflowId,
customerId: previousFullCustomer.id ?? "",
},
payload: {
orgId: previousFullCustomer.org_id,
env: previousFullCustomer.env,
customerId: previousFullCustomer.id || previousFullCustomer.internal_id,
newCustomerProductId,
newCustomerProductId: newCustomerProduct.id,
source,
previousFullCustomer: JSON.stringify(previousFullCustomer), // is there a better approach to this...?
},

View File

@@ -3,6 +3,7 @@ import * as Sentry from "@sentry/bun";
import { db } from "@/db/initDrizzle.js";
import { hatchet } from "@/external/hatchet/initHatchet.js";
import { getSentryTags } from "@/external/sentry/sentryUtils.js";
import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
import { CusService } from "@/internal/customers/CusService.js";
import { getApiCustomerBase } from "@/internal/customers/cusUtils/apiCusUtils/getApiCustomerBase.js";
import { deleteCachedFullCustomer } from "@/internal/customers/cusUtils/fullCustomerCacheUtils/deleteCachedFullCustomer.js";
@@ -38,9 +39,11 @@ export const verifyCacheConsistencyWorkflow = hatchet?.workflow<
// Check if subscriptions match
const checkSubscriptionsMatch = ({
ctx,
dbCustomer,
cachedCustomer,
}: {
ctx: AutumnContext;
dbCustomer: ApiCustomer;
cachedCustomer: ApiCustomer;
}): { success: boolean; message: string } => {
@@ -124,10 +127,14 @@ verifyCacheConsistencyWorkflow?.task({
withAutumnId: true,
});
autumnContext.logger.info(`DB CUSTOMER`, { data: dbCustomer });
autumnContext.logger.info(`CACHED CUSTOMER`, { data: cachedCustomer });
const {
success: subscriptionsMatch,
message: subscriptionsMatchMessage,
} = checkSubscriptionsMatch({
ctx: autumnContext,
dbCustomer,
cachedCustomer,
});
@@ -139,16 +146,15 @@ verifyCacheConsistencyWorkflow?.task({
await deleteCachedFullCustomer({
customerId,
orgId: autumnContext.org.id,
env: autumnContext.env,
ctx: autumnContext,
source: "verifyCacheConsistency",
logger: autumnContext.logger,
});
Sentry.captureException(new Error(subscriptionsMatchMessage), {
tags: getSentryTags({
ctx: autumnContext,
customerId,
alert: true,
}),
});
}

View File

@@ -86,15 +86,6 @@ const processMessage = async ({
return;
}
if (job.name === JobName.Migration) {
await runMigrationTask({
db,
payload: job.data,
logger: workerLogger,
});
return;
}
if (job.name === JobName.ClearCreditSystemCustomerCache) {
await runClearCreditSystemCacheTask({
db,
@@ -107,8 +98,7 @@ const processMessage = async ({
// Jobs below need worker context
const ctx = await createWorkerContext({
db,
orgId: job.data.orgId,
env: job.data.env,
payload: job.data,
logger: workerLogger,
});
@@ -119,6 +109,15 @@ const processMessage = async ({
});
}
if (job.name === JobName.Migration) {
if (!ctx) {
workerLogger.error("No context found for migration job");
return;
}
await runMigrationTask({ ctx, payload: job.data });
return;
}
if (actionHandlers.includes(job.name as JobName)) {
// Note: action handlers need BullMQ queue for nested jobs
// This will need to be refactored when migrating action handlers to SQS

View File

@@ -143,10 +143,12 @@ const hatchetWorkflows = {
*/
export const runHatchetWorkflow = async <T extends keyof HatchetPayloads>({
workflowName,
metadata,
payload,
delayMs,
}: {
workflowName: T;
metadata?: Record<string, string>;
payload: HatchetPayloads[T];
/** Delay in milliseconds before the workflow runs */
delayMs?: number;
@@ -162,7 +164,11 @@ export const runHatchetWorkflow = async <T extends keyof HatchetPayloads>({
if (delayMs) {
// workflow.delay() takes duration in seconds
const delaySeconds = Math.floor(delayMs / 1000);
await workflow.delay(delaySeconds, payload);
await workflow.delay(delaySeconds, payload, {
additionalMetadata: {
...(metadata ?? {}),
},
});
} else {
await workflow.run(payload);
}

View File

@@ -5,6 +5,7 @@ import { configsRouter } from "@/internal/configs/configsRouter.js";
import { eventsRouter } from "@/internal/events/eventsRouter.js";
import { analyticsMiddleware } from "../honoMiddlewares/analyticsMiddleware.js";
import { apiVersionMiddleware } from "../honoMiddlewares/apiVersionMiddleware.js";
import { idempotencyMiddleware } from "../honoMiddlewares/idempotencyMiddleware.js";
import { orgConfigMiddleware } from "../honoMiddlewares/orgConfigMiddleware.js";
import { queryMiddleware } from "../honoMiddlewares/queryMiddleware.js";
import { rateLimitMiddleware } from "../honoMiddlewares/rateLimitMiddleware.js";
@@ -14,14 +15,14 @@ import type { HonoEnv } from "../honoUtils/HonoEnv.js";
import {
redemptionRouter,
referralRouter,
} from "../internal/api/rewards/referralRouter.js";
import { balancesRouter } from "../internal/balances/balancesRouter.js";
import { billingRouter } from "../internal/billing/billingRouter.js";
import { cusRouter } from "../internal/customers/cusRouter.js";
import { entityRouter } from "../internal/entities/entityRouter.js";
import { featureRouter } from "../internal/features/featureRouter.js";
import { honoOrgRouter } from "../internal/orgs/orgRouter.js";
import { platformBetaRouter } from "../internal/platform/platformBeta/platformBetaRouter.js";
} from "../internal/api/rewards/referralRouter";
import { balancesRouter } from "../internal/balances/balancesRouter";
import { billingRouter } from "../internal/billing/billingRouter";
import { cusRouter } from "../internal/customers/cusRouter";
import { entityRouter } from "../internal/entities/entityRouter";
import { featureRouter } from "../internal/features/featureRouter";
import { honoOrgRouter } from "../internal/orgs/orgRouter";
import { platformBetaRouter } from "../internal/platform/platformBeta/platformBetaRouter";
import {
honoProductBetaRouter,
honoProductRouter,
@@ -37,6 +38,7 @@ apiRouter.use("*", analyticsMiddleware);
apiRouter.use("*", rateLimitMiddleware);
apiRouter.use("*", refreshCacheMiddleware);
apiRouter.use("*", queryMiddleware());
apiRouter.use("*", idempotencyMiddleware);
apiRouter.route("", billingRouter);
apiRouter.route("", balancesRouter);

View File

@@ -54,6 +54,7 @@ export const createSharedProducts = async ({
customerId: customer.id ?? "",
orgId: ctx.org.id,
env: ctx.env,
logger: ctx.logger,
}),
);
}

View File

@@ -72,7 +72,16 @@ describe(`${chalk.yellowBright("temp: invoice payment failed for one off credits
customer_id: customerId,
product_id: free.id,
});
});
test("should handle invoice payment failed for one off credits", async () => {});
await autumnV1.attach({
customer_id: customerId,
product_id: oneOffCredits.id,
options: [
{
feature_id: TestFeature.Credits,
quantity: 25,
},
],
});
});
});

View File

@@ -0,0 +1,177 @@
import { beforeAll, describe, expect, test } from "bun:test";
import {
type AppEnv,
CusProductStatus,
type Organization,
} from "@autumn/shared";
import { defaultApiVersion } from "@tests/constants.js";
import { TestFeature } from "@tests/setup/v2Features.js";
import { attachAndExpectCorrect } from "@tests/utils/expectUtils/expectAttach.js";
import { expectDowngradeCorrect } from "@tests/utils/expectUtils/expectScheduleUtils.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import type Stripe from "stripe";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { timeout } from "@/utils/genUtils.js";
import { constructArrearItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
import { expectSubToBeCorrect } from "../../merged/mergeUtils/expectSubCorrect.js";
import { expectProductAttached } from "../../utils/expectUtils/expectProductAttached.js";
const testCase = "migrations6";
const wordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 1000,
});
const pro = constructProduct({
id: "pro",
items: [wordsItem],
type: "pro",
isDefault: false,
});
const premiumWordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 5000,
});
const premium = constructProduct({
id: "premium",
items: [premiumWordsItem],
type: "premium",
isDefault: false,
});
const updatedPremiumWordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 10000,
});
describe(`${chalk.yellowBright(`${testCase}: Testing migration for premium v1 -> premium v2 after downgrade to pro`)}`, () => {
const customerId = testCase;
const autumn: AutumnInt = new AutumnInt({ version: defaultApiVersion });
let db: DrizzleCli, org: Organization, env: AppEnv;
let stripeCli: Stripe;
beforeAll(async () => {
db = ctx.db;
org = ctx.org;
env = ctx.env;
stripeCli = ctx.stripeCli;
await initProductsV0({
ctx,
products: [pro, premium],
prefix: testCase,
customerId,
});
await initCustomerV3({
ctx,
customerId,
customerData: {},
attachPm: "success",
withTestClock: true,
});
});
test("should attach premium product", async () => {
await attachAndExpectCorrect({
autumn,
customerId,
product: premium,
stripeCli,
db,
org,
env,
});
const customer = await autumn.customers.get(customerId);
const wordsBalance = customer.features[TestFeature.Words].balance;
expect(wordsBalance).toBe(5000);
});
test("should downgrade to pro", async () => {
await expectDowngradeCorrect({
autumn,
customerId,
curProduct: premium,
newProduct: pro,
stripeCli,
db,
org,
env,
});
// Customer should still have premium active (with scheduled pro)
const customer = await autumn.customers.get(customerId);
const wordsBalance = customer.features[TestFeature.Words].balance;
expect(wordsBalance).toBe(5000);
});
test("should update premium product to new version", async () => {
await autumn.products.update(premium.id, {
items: [updatedPremiumWordsItem],
});
});
test("should migrate premium v1 to premium v2 after downgrade", async () => {
const wordsUsage = 2000;
await autumn.track({
customer_id: customerId,
value: wordsUsage,
feature_id: TestFeature.Words,
});
await timeout(4000);
// Verify usage was tracked correctly before migration
const customerBeforeMigration = await autumn.customers.get(customerId);
expect(customerBeforeMigration.features[TestFeature.Words].balance).toBe(
5000 - wordsUsage,
);
// Create updated premium with version 2
const premiumV2 = { ...premium, version: 2 };
premiumV2.items = [updatedPremiumWordsItem];
await autumn.migrate({
from_product_id: premium.id,
to_product_id: premiumV2.id,
from_version: premium.version,
to_version: premiumV2.version,
});
await timeout(10000);
// 1. Check that premium v2 active, pro scheduled
const customer = await autumn.customers.get(customerId);
expectProductAttached({
customer,
product: premiumV2,
status: CusProductStatus.Active,
});
expectProductAttached({
customer,
product: pro,
status: CusProductStatus.Scheduled,
});
await expectSubToBeCorrect({
db,
customerId,
org,
env,
});
// After migration, customer should have premium v2 with new included usage
const wordsBalance = customer.features[TestFeature.Words].balance;
// New included usage is 10000, minus 2000 used = 8000
expect(wordsBalance).toBe(10000 - wordsUsage);
});
});

View File

@@ -0,0 +1,143 @@
import { beforeAll, describe, expect, test } from "bun:test";
import {
type AppEnv,
CusProductStatus,
type Organization,
} from "@autumn/shared";
import { defaultApiVersion } from "@tests/constants.js";
import { TestFeature } from "@tests/setup/v2Features.js";
import { attachAndExpectCorrect } from "@tests/utils/expectUtils/expectAttach.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import type Stripe from "stripe";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { timeout } from "@/utils/genUtils.js";
import { constructArrearItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
import { expectSubToBeCorrect } from "../../merged/mergeUtils/expectSubCorrect.js";
import { expectProductAttached } from "../../utils/expectUtils/expectProductAttached.js";
const testCase = "migrations7";
const premiumWordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 5000,
});
const premium = constructProduct({
id: "premium",
items: [premiumWordsItem],
type: "premium",
isDefault: false,
});
const updatedPremiumWordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 10000,
});
describe(`${chalk.yellowBright(`${testCase}: Testing migration for premium v1 -> premium v2 after cancellation at cycle end`)}`, () => {
const customerId = testCase;
const autumn: AutumnInt = new AutumnInt({ version: defaultApiVersion });
let db: DrizzleCli, org: Organization, env: AppEnv;
let stripeCli: Stripe;
beforeAll(async () => {
db = ctx.db;
org = ctx.org;
env = ctx.env;
stripeCli = ctx.stripeCli;
await initProductsV0({
ctx,
products: [premium],
prefix: testCase,
customerId,
});
await initCustomerV3({
ctx,
customerId,
customerData: {},
attachPm: "success",
withTestClock: true,
});
});
test("should attach premium product", async () => {
await attachAndExpectCorrect({
autumn,
customerId,
product: premium,
stripeCli,
db,
org,
env,
});
const customer = await autumn.customers.get(customerId);
const wordsBalance = customer.features[TestFeature.Words].balance;
expect(wordsBalance).toBe(5000);
});
test("should cancel premium at cycle end", async () => {
await autumn.cancel({
customer_id: customerId,
product_id: premium.id,
cancel_immediately: false,
});
const customer = await autumn.customers.get(customerId);
expectProductAttached({
customer,
product: premium,
status: CusProductStatus.Active,
isCanceled: true,
});
});
test("should update premium product to new version", async () => {
await autumn.products.update(premium.id, {
items: [updatedPremiumWordsItem],
});
});
test("should migrate premium v1 to premium v2 and preserve cancellation", async () => {
const premiumV2 = { ...premium, version: 2 };
premiumV2.items = [updatedPremiumWordsItem];
await autumn.migrate({
from_product_id: premium.id,
to_product_id: premiumV2.id,
from_version: premium.version,
to_version: premiumV2.version,
});
await timeout(10000);
// Check that premium v2 is active but still cancelling
const customer = await autumn.customers.get(customerId);
expectProductAttached({
customer,
product: premiumV2,
status: CusProductStatus.Active,
isCanceled: true,
});
// Subscription should be cancelled at period end
await expectSubToBeCorrect({
db,
customerId,
org,
env,
shouldBeCanceled: true,
});
// Should have updated included usage
const wordsBalance = customer.features[TestFeature.Words].balance;
expect(wordsBalance).toBe(10000);
});
});

View File

@@ -0,0 +1,199 @@
import { beforeAll, describe, expect, test } from "bun:test";
import {
type AppEnv,
CusProductStatus,
type Organization,
} from "@autumn/shared";
import { defaultApiVersion } from "@tests/constants.js";
import { TestFeature } from "@tests/setup/v2Features.js";
import { attachAndExpectCorrect } from "@tests/utils/expectUtils/expectAttach.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import type Stripe from "stripe";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { timeout } from "@/utils/genUtils.js";
import { constructArrearItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
import { expectSubToBeCorrect } from "../../merged/mergeUtils/expectSubCorrect.js";
import { expectProductAttached } from "../../utils/expectUtils/expectProductAttached.js";
const testCase = "migrations8";
const premiumWordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 5000,
});
const premium = constructProduct({
id: "premium",
items: [premiumWordsItem],
type: "premium",
isDefault: false,
});
const updatedPremiumWordsItem = constructArrearItem({
featureId: TestFeature.Words,
includedUsage: 10000,
});
const entities = [
{
id: "1",
name: "Entity 1",
feature_id: TestFeature.Users,
},
{
id: "2",
name: "Entity 2",
feature_id: TestFeature.Users,
},
];
describe(`${chalk.yellowBright(`${testCase}: Testing migration with entities - one cancelled, one active`)}`, () => {
const customerId = testCase;
const autumn: AutumnInt = new AutumnInt({ version: defaultApiVersion });
let db: DrizzleCli, org: Organization, env: AppEnv;
let stripeCli: Stripe;
const entity1 = entities[0];
const entity2 = entities[1];
beforeAll(async () => {
db = ctx.db;
org = ctx.org;
env = ctx.env;
stripeCli = ctx.stripeCli;
await initProductsV0({
ctx,
products: [premium],
prefix: testCase,
customerId,
});
await initCustomerV3({
ctx,
customerId,
customerData: {},
attachPm: "success",
withTestClock: true,
});
});
test("should create entities", async () => {
await autumn.entities.create(customerId, entities);
});
test("should attach premium product to entity 1", async () => {
await attachAndExpectCorrect({
autumn,
customerId,
product: premium,
stripeCli,
db,
org,
env,
entityId: entity1.id,
numSubs: 1,
});
});
test("should attach premium product to entity 2", async () => {
await attachAndExpectCorrect({
autumn,
customerId,
product: premium,
stripeCli,
db,
org,
env,
entityId: entity2.id,
numSubs: 2,
});
});
test("should cancel premium on entity 1", async () => {
await autumn.cancel({
customer_id: customerId,
product_id: premium.id,
entity_id: entity1.id,
cancel_immediately: false,
});
await timeout(3000);
const entity = await autumn.entities.get(customerId, entity1.id);
expectProductAttached({
customer: entity,
product: premium,
status: CusProductStatus.Active,
isCanceled: true,
});
// Entity 2 should still be active and not cancelled
const entity2Res = await autumn.entities.get(customerId, entity2.id);
expectProductAttached({
customer: entity2Res,
product: premium,
status: CusProductStatus.Active,
});
expect(entity2Res.products[0].canceled_at).toBeFalsy();
});
test("should update premium product to new version", async () => {
await autumn.products.update(premium.id, {
items: [updatedPremiumWordsItem],
});
});
test("should migrate premium v1 to premium v2", async () => {
const premiumV2 = { ...premium, version: 2 };
premiumV2.items = [updatedPremiumWordsItem];
await autumn.migrate({
from_product_id: premium.id,
to_product_id: premiumV2.id,
from_version: premium.version,
to_version: premiumV2.version,
});
await timeout(10000);
// Entity 1 should have premium v2, still cancelling
const entity1Res = await autumn.entities.get(customerId, entity1.id);
expectProductAttached({
customer: entity1Res,
product: premiumV2,
status: CusProductStatus.Active,
isCanceled: true,
});
// Entity 1 should have updated included usage
const entity1Words = entity1Res.features[TestFeature.Words].balance;
expect(entity1Words).toBe(10000);
// Entity 2 should have premium v2, active and NOT cancelled
const entity2Res = await autumn.entities.get(customerId, entity2.id);
expectProductAttached({
customer: entity2Res,
product: premiumV2,
status: CusProductStatus.Active,
});
expect(entity2Res.products[0].canceled_at).toBeFalsy();
// Entity 2 should have updated included usage
const entity2Words = entity2Res.features[TestFeature.Words].balance;
expect(entity2Words).toBe(10000);
await expectSubToBeCorrect({
db,
customerId,
org,
env,
});
});
});

View File

@@ -1,98 +1,98 @@
import { beforeAll, describe, expect, it } from "bun:test";
import { ApiVersion } from "@autumn/shared";
import { TestFeature } from "@tests/setup/v2Features.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import AutumnError, { AutumnInt } from "@/external/autumn/autumnCli.js";
import { constructFeatureItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
// import { beforeAll, describe, expect, it } from "bun:test";
// import { ApiVersion } from "@autumn/shared";
// import { TestFeature } from "@tests/setup/v2Features.js";
// import ctx from "@tests/utils/testInitUtils/createTestContext.js";
// import chalk from "chalk";
// import AutumnError, { AutumnInt } from "@/external/autumn/autumnCli.js";
// import { constructFeatureItem } from "@/utils/scriptUtils/constructItem.js";
// import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
// import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
// import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
const pro = constructProduct({
type: "pro",
items: [
constructFeatureItem({
featureId: TestFeature.Messages,
includedUsage: 100,
}),
],
});
// const pro = constructProduct({
// type: "pro",
// items: [
// constructFeatureItem({
// featureId: TestFeature.Messages,
// includedUsage: 100,
// }),
// ],
// });
const premium = constructProduct({
type: "premium",
items: [
constructFeatureItem({
featureId: TestFeature.Messages,
includedUsage: 300,
}),
],
});
// const premium = constructProduct({
// type: "premium",
// items: [
// constructFeatureItem({
// featureId: TestFeature.Messages,
// includedUsage: 300,
// }),
// ],
// });
const testCase = "attach-misc4";
// const testCase = "attach-misc4";
describe(`${chalk.yellowBright("attach-misc4: rate limit test")}`, () => {
const customerId = testCase;
const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
// describe(`${chalk.yellowBright("attach-misc4: rate limit test")}`, () => {
// const customerId = testCase;
// const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
beforeAll(async () => {
await initCustomerV3({
ctx,
customerId,
withTestClock: true,
});
// beforeAll(async () => {
// await initCustomerV3({
// ctx,
// customerId,
// withTestClock: true,
// });
await initCustomerV3({
ctx,
customerId: `${customerId}-2`,
withTestClock: true,
});
// await initCustomerV3({
// ctx,
// customerId: `${customerId}-2`,
// withTestClock: true,
// });
await initProductsV0({
ctx,
products: [pro, premium],
prefix: testCase,
});
});
// await initProductsV0({
// ctx,
// products: [pro, premium],
// prefix: testCase,
// });
// });
it("should run 6 attaches for one customer and hit rate limit, then allow attach for second customer", async () => {
const customer1Id = customerId;
const customer2Id = `${customerId}-2`;
// it("should run 6 attaches for one customer and hit rate limit, then allow attach for second customer", async () => {
// const customer1Id = customerId;
// const customer2Id = `${customerId}-2`;
// Run 6 sequential attaches for customer 1 (limit is 5 per minute)
const customer1Results: PromiseSettledResult<unknown>[] = [];
for (let i = 0; i < 6; i++) {
try {
const result = await autumnV1.attach({
customer_id: customer1Id,
product_id: pro.id,
});
customer1Results.push({ status: "fulfilled", value: result });
} catch (error) {
customer1Results.push({ status: "rejected", reason: error });
}
}
// // Run 6 sequential attaches for customer 1 (limit is 5 per minute)
// const customer1Results: PromiseSettledResult<unknown>[] = [];
// for (let i = 0; i < 6; i++) {
// try {
// const result = await autumnV1.attach({
// customer_id: customer1Id,
// product_id: pro.id,
// });
// customer1Results.push({ status: "fulfilled", value: result });
// } catch (error) {
// customer1Results.push({ status: "rejected", reason: error });
// }
// }
// Count rate limit errors for customer 1
const rateLimitErrors = customer1Results.filter(
(r) =>
r.status === "rejected" &&
r.reason instanceof AutumnError &&
r.reason.code === "rate_limit_exceeded",
);
// // Count rate limit errors for customer 1
// const rateLimitErrors = customer1Results.filter(
// (r) =>
// r.status === "rejected" &&
// r.reason instanceof AutumnError &&
// r.reason.code === "rate_limit_exceeded",
// );
// Exactly 1 should have been rate limited (6 requests with limit of 5)
expect(rateLimitErrors.length).toBe(1);
console.log("Customer 1: 1 out of 6 requests was rate limited");
// // Exactly 1 should have been rate limited (6 requests with limit of 5)
// expect(rateLimitErrors.length).toBe(1);
// console.log("Customer 1: 1 out of 6 requests was rate limited");
// Run attach for customer 2 - should NOT be rate limited (different customer = different rate limit key)
const customer2Result = await autumnV1.attach({
customer_id: customer2Id,
product_id: pro.id,
});
// // Run attach for customer 2 - should NOT be rate limited (different customer = different rate limit key)
// const customer2Result = await autumnV1.attach({
// customer_id: customer2Id,
// product_id: pro.id,
// });
// Customer 2's attach should succeed without rate limit error
expect(customer2Result).toBeDefined();
console.log("Customer 2: attach succeeded without rate limit");
});
});
// // Customer 2's attach should succeed without rate limit error
// expect(customer2Result).toBeDefined();
// console.log("Customer 2: attach succeeded without rate limit");
// });
// });

View File

@@ -0,0 +1,93 @@
import { beforeAll, describe, expect, test } from "bun:test";
import { ApiVersion, ErrCode } from "@autumn/shared";
import { TestFeature } from "@tests/setup/v2Features.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import AutumnError, { AutumnInt } from "@/external/autumn/autumnCli.js";
import { generateId } from "@/utils/genUtils";
import { constructFeatureItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
const pro = constructProduct({
type: "pro",
items: [
constructFeatureItem({
featureId: TestFeature.Credits,
includedUsage: 500,
}),
],
});
const testCase = "others10";
describe(`${chalk.yellowBright(`${testCase}/idempotency: idempotency key already exists`)}`, () => {
const customerId = testCase;
const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
const idempotencyKey = generateId("it");
let results: PromiseSettledResult<
Awaited<ReturnType<typeof autumnV1.attach>>
>[];
beforeAll(async () => {
await initCustomerV3({
ctx,
customerId,
withTestClock: true,
attachPm: "success",
});
await initProductsV0({
ctx,
products: [pro],
prefix: testCase,
});
results = await Promise.allSettled([
autumnV1.attach(
{
customer_id: customerId,
product_id: pro.id,
},
{
"idempotency-key": idempotencyKey,
},
),
autumnV1.attach(
{
customer_id: customerId,
product_id: pro.id,
},
{
"idempotency-key": idempotencyKey,
},
),
]);
});
test("should reject duplicate idempotency key with 409", async () => {
// Exactly one request should succeed
const fulfilled = results.filter((r) => r.status === "fulfilled");
const rejected = results.filter((r) => r.status === "rejected");
expect(fulfilled).toHaveLength(1);
expect(rejected).toHaveLength(1);
// The successful request should have attached the product
const successResult = fulfilled[0] as PromiseFulfilledResult<
Awaited<ReturnType<typeof autumnV1.attach>>
>;
expect(successResult.value.success).toBe(true);
expect(successResult.value.customer_id).toBe(customerId);
expect(successResult.value.product_ids).toContain(pro.id);
// The rejected request should have the duplicate idempotency key error
const rejectedResult = rejected[0] as PromiseRejectedResult;
expect(rejectedResult.reason).toBeInstanceOf(AutumnError);
expect((rejectedResult.reason as AutumnError).code).toBe(
ErrCode.DuplicateIdempotencyKey,
);
});
});

View File

@@ -1,12 +1,12 @@
import { beforeAll, describe, test } from "bun:test";
import { type AppEnv, LegacyVersion, type Organization } from "@autumn/shared";
import chalk from "chalk";
import { addWeeks } from "date-fns";
import type Stripe from "stripe";
import { TestFeature } from "@tests/setup/v2Features.js";
import { attachAndExpectCorrect } from "@tests/utils/expectUtils/expectAttach.js";
import { advanceTestClock } from "@tests/utils/stripeUtils.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import { addWeeks } from "date-fns";
import type Stripe from "stripe";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { constructArrearItem } from "@/utils/scriptUtils/constructItem.js";

View File

@@ -1,6 +1,7 @@
import { beforeAll, describe, expect, test } from "bun:test";
import {
ApiVersion,
type CheckResponseV1,
type CheckResponseV2,
ProductItemInterval,
} from "@autumn/shared";
@@ -49,6 +50,7 @@ const testCase = "check-breakdown2";
describe(`${chalk.yellowBright("check-breakdown2: monthly + lifetime messages = 2 breakdown items")}`, () => {
const customerId = testCase;
const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
const autumnV2: AutumnInt = new AutumnInt({ version: ApiVersion.V2_0 });
beforeAll(async () => {
@@ -130,4 +132,49 @@ describe(`${chalk.yellowBright("check-breakdown2: monthly + lifetime messages =
},
});
});
test("v1.2: should have correct parent and 2 breakdown items", async () => {
const res = (await autumnV1.check<CheckResponseV1>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV1;
// Parent balance should sum both products
expect(res.included_usage).toBe(700); // 500 + 200
expect(res.balance).toBe(700);
expect(res.usage).toBe(0);
// Should have 2 breakdown items
expect(res.breakdown).toHaveLength(2);
});
test("v1.2: breakdown items should have correct values", async () => {
const res = (await autumnV1.check<CheckResponseV1>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV1;
console.log("Response:", res);
const monthlyBreakdown = res.breakdown?.find(
(b) => b.included_usage === 500,
);
const lifetimeBreakdown = res.breakdown?.find(
(b) => b.included_usage === 200,
);
expect(monthlyBreakdown).toMatchObject({
included_usage: 500,
balance: 500,
usage: 0,
interval: "month",
});
expect(lifetimeBreakdown).toMatchObject({
included_usage: 200,
balance: 200,
usage: 0,
interval: "lifetime",
});
});
});

View File

@@ -0,0 +1,164 @@
import { beforeAll, describe, expect, test } from "bun:test";
import {
ApiVersion,
type CheckResponseV1,
type CheckResponseV2,
ProductItemInterval,
} from "@autumn/shared";
import { TestFeature } from "@tests/setup/v2Features.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { constructFeatureItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
/**
* Test: Monthly messages + Lifetime messages under the SAME product
* Expected: breakdown array with 2 items (different reset intervals, same product)
*/
const monthlyMessages = constructFeatureItem({
featureId: TestFeature.Messages,
includedUsage: 500,
interval: ProductItemInterval.Month,
});
const lifetimeMessages = constructFeatureItem({
featureId: TestFeature.Messages,
includedUsage: 200,
interval: null, // lifetime
});
const combinedProd = constructProduct({
type: "free",
id: "combined-prod",
isDefault: false,
items: [monthlyMessages, lifetimeMessages],
});
const testCase = "check-breakdown6";
describe(`${chalk.yellowBright("check-breakdown6: monthly + lifetime messages in same product = 2 breakdown items")}`, () => {
const customerId = testCase;
const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
const autumnV2: AutumnInt = new AutumnInt({ version: ApiVersion.V2_0 });
beforeAll(async () => {
await initCustomerV3({
ctx,
customerId,
withTestClock: false,
});
await initProductsV0({
ctx,
products: [combinedProd],
prefix: testCase,
});
await autumnV2.attach({
customer_id: customerId,
product_id: combinedProd.id,
});
});
test("v2: should have correct parent balance and 2 breakdown items", async () => {
const res = (await autumnV2.check<CheckResponseV2>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV2;
// Parent balance should sum both items
expect(res.balance).toMatchObject({
granted_balance: 700, // 500 + 200
current_balance: 700,
usage: 0,
plan_id: combinedProd.id, // Same product for both
});
// Should have 2 breakdown items with unique IDs
const breakdown = res.balance?.breakdown;
expect(breakdown).toHaveLength(2);
expect(new Set(breakdown?.map((b) => b.id)).size).toBe(2);
});
test("v2: breakdown items should have correct values", async () => {
const res = (await autumnV2.check<CheckResponseV2>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV2;
const monthlyBreakdown = res.balance?.breakdown?.find(
(b) => b.granted_balance === 500,
);
const lifetimeBreakdown = res.balance?.breakdown?.find(
(b) => b.granted_balance === 200,
);
expect(monthlyBreakdown).toMatchObject({
granted_balance: 500,
current_balance: 500,
usage: 0,
plan_id: combinedProd.id,
reset: {
interval: "month",
},
});
expect(lifetimeBreakdown).toMatchObject({
granted_balance: 200,
current_balance: 200,
usage: 0,
plan_id: combinedProd.id,
reset: {
interval: "one_off",
resets_at: null,
},
});
});
test("v1.2: should have correct parent and 2 breakdown items", async () => {
const res = (await autumnV1.check<CheckResponseV1>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV1;
// Parent balance should sum both items
expect(res.included_usage).toBe(700); // 500 + 200
expect(res.balance).toBe(700);
expect(res.usage).toBe(0);
// Should have 2 breakdown items
expect(res.breakdown).toHaveLength(2);
});
test("v1.2: breakdown items should have correct values", async () => {
const res = (await autumnV1.check<CheckResponseV1>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV1;
const monthlyBreakdown = res.breakdown?.find(
(b) => b.included_usage === 500,
);
const lifetimeBreakdown = res.breakdown?.find(
(b) => b.included_usage === 200,
);
expect(monthlyBreakdown).toMatchObject({
included_usage: 500,
balance: 500,
usage: 0,
interval: "month",
});
expect(lifetimeBreakdown).toMatchObject({
included_usage: 200,
balance: 200,
usage: 0,
interval: "lifetime",
});
});
});

View File

@@ -0,0 +1,189 @@
import { beforeAll, describe, expect, test } from "bun:test";
import {
ApiVersion,
type CheckResponseV1,
type CheckResponseV2,
ProductItemInterval,
} from "@autumn/shared";
import { TestFeature } from "@tests/setup/v2Features.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import {
constructFeatureItem,
constructPrepaidItem,
} from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
/**
* Test: Free product with monthly messages + One-off top-up product for messages
* Expected: breakdown array with 2 items (monthly + one_off/lifetime)
*/
const freeMonthlyMessages = constructFeatureItem({
featureId: TestFeature.Messages,
includedUsage: 500,
interval: ProductItemInterval.Month,
});
const topUpMessages = constructPrepaidItem({
featureId: TestFeature.Messages,
billingUnits: 100,
price: 10,
isOneOff: true, // One-off top-up
});
const freeProd = constructProduct({
type: "free",
id: "free-prod",
isDefault: false,
items: [freeMonthlyMessages],
});
const topUpProd = constructProduct({
type: "free",
id: "topup-prod",
isAddOn: true,
isDefault: false,
items: [topUpMessages],
});
const testCase = "check-breakdown7";
describe(`${chalk.yellowBright("check-breakdown7: free monthly messages + one-off top-up = 2 breakdown items")}`, () => {
const customerId = testCase;
const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
const autumnV2: AutumnInt = new AutumnInt({ version: ApiVersion.V2_0 });
beforeAll(async () => {
await initCustomerV3({
ctx,
customerId,
attachPm: "success", // Required for prepaid
withTestClock: false,
});
await initProductsV0({
ctx,
products: [freeProd, topUpProd],
prefix: testCase,
});
// Attach free product first
await autumnV2.attach({
customer_id: customerId,
product_id: freeProd.id,
});
// Attach top-up product with quantity
// Note: quantity is the actual number of messages, NOT multiplied by billing_units
await autumnV2.attach({
customer_id: customerId,
product_id: topUpProd.id,
options: [
{
feature_id: TestFeature.Messages,
quantity: 200, // 200 messages directly
},
],
});
});
test("v2: should have correct parent balance and 2 breakdown items", async () => {
const res = (await autumnV2.check<CheckResponseV2>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV2;
// Parent balance should sum both products: 500 (free) + 200 (top-up)
expect(res.balance).toMatchObject({
granted_balance: 500,
current_balance: 700,
purchased_balance: 200,
usage: 0,
plan_id: null, // null when multiple products
});
// Should have 2 breakdown items with unique IDs
const breakdown = res.balance?.breakdown;
expect(breakdown).toHaveLength(2);
expect(new Set(breakdown?.map((b) => b.id)).size).toBe(2);
});
test("v2: breakdown items should have correct values", async () => {
const res = (await autumnV2.check<CheckResponseV2>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV2;
const freeBreakdown = res.balance?.breakdown?.find(
(b) => b.plan_id === freeProd.id,
);
const topUpBreakdown = res.balance?.breakdown?.find(
(b) => b.plan_id === topUpProd.id,
);
expect(freeBreakdown).toMatchObject({
granted_balance: 500,
current_balance: 500,
usage: 0,
plan_id: freeProd.id,
reset: {
interval: "month",
},
});
expect(topUpBreakdown).toMatchObject({
granted_balance: 0,
current_balance: 200,
purchased_balance: 200,
usage: 0,
plan_id: topUpProd.id,
reset: {
interval: "one_off",
resets_at: null,
},
});
});
test("v1.2: should have correct parent and 2 breakdown items", async () => {
const res = (await autumnV1.check<CheckResponseV1>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV1;
// Parent balance should sum both products
expect(res.included_usage).toBe(700); // 500 + 200
expect(res.balance).toBe(700);
expect(res.usage).toBe(0);
// Should have 2 breakdown items
expect(res.breakdown).toHaveLength(2);
});
test("v1.2: breakdown items should have correct values", async () => {
const res = (await autumnV1.check<CheckResponseV1>({
customer_id: customerId,
feature_id: TestFeature.Messages,
})) as unknown as CheckResponseV1;
const freeBreakdown = res.breakdown?.[0];
const topUpBreakdown = res.breakdown?.[1];
expect(freeBreakdown).toMatchObject({
included_usage: 500,
balance: 500,
usage: 0,
interval: "month",
});
expect(topUpBreakdown).toMatchObject({
included_usage: 200,
balance: 200,
usage: 0,
interval: "lifetime", // one_off becomes lifetime in v1.2
});
});
});

View File

@@ -167,6 +167,7 @@ describe(`${chalk.yellowBright("track-race-condition1: sync should not wipe out
env: ctx.env,
customerId: customerId,
source: "test-setup",
logger: ctx.logger,
});
// 1. Check that credits weren't wiped out

View File

@@ -0,0 +1,74 @@
import { beforeAll, describe, expect, test } from "bun:test";
import { ApiVersion } from "@autumn/shared";
import { TestFeature } from "@tests/setup/v2Features.js";
import ctx from "@tests/utils/testInitUtils/createTestContext.js";
import chalk from "chalk";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { constructFeatureItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
import { initProductsV0 } from "@/utils/scriptUtils/testUtils/initProductsV0.js";
const pro = constructProduct({
type: "pro",
items: [
constructFeatureItem({
featureId: TestFeature.Credits,
includedUsage: 500,
}),
],
});
const testCase = "legacy-balance-update1";
describe(`${chalk.yellowBright("legacy-balance-update1: allow updating balances for an entity")}`, () => {
const customerId = testCase;
const entityId = `${testCase}-user-1`;
const autumnV1: AutumnInt = new AutumnInt({ version: ApiVersion.V1_2 });
beforeAll(async () => {
await initCustomerV3({
ctx,
customerId,
withTestClock: true,
attachPm: "success",
});
await autumnV1.entities.create(customerId, [
{
id: entityId,
name: "User 1",
feature_id: TestFeature.Credits,
},
]);
await initProductsV0({
ctx,
products: [pro],
prefix: testCase,
});
await autumnV1.attach({
customer_id: customerId,
entity_id: entityId,
product_id: pro.id,
});
});
test("should allow updating balances for an entity", async () => {
await autumnV1.customers.setBalance({
customerId: customerId,
entityId: entityId,
balances: [
{
feature_id: TestFeature.Credits,
balance: 100,
},
],
});
const entity = await autumnV1.entities.get(customerId, entityId);
expect(entity.features.credits.balance).toBe(100);
});
});

View File

@@ -102,6 +102,8 @@ describe(`${chalk.yellowBright("mergedAdd1: Testing merged subs, with track")}`,
org,
env,
});
await timeout(3000);
});
test("should track usage and have correct invoice end of month", async () => {

View File

@@ -83,7 +83,11 @@ const compareActualItems = async ({
expect(actualItem).toBeDefined();
if (actualItem?.quantity !== (expectedItem as any).quantity) {
// Treat 0 and undefined as equivalent for quantity
const actualQty = actualItem?.quantity ?? 0;
const expectedQty = (expectedItem as any).quantity ?? 0;
if (actualQty !== expectedQty) {
if (phaseStartsAt) {
console.log(`Phase starts at: ${formatUnixToDateTime(phaseStartsAt)}`);
}
@@ -100,9 +104,7 @@ const compareActualItems = async ({
items: expectedItems,
});
console.log(
`Item quantity mismatch: ${actualItem?.quantity} !== ${expectedItem.quantity}`,
);
console.log(`Item quantity mismatch: ${actualQty} !== ${expectedQty}`);
const price = await PriceService.getByStripeId({
db,
@@ -118,7 +120,7 @@ const compareActualItems = async ({
console.log("--------------------------------");
}
expect(actualItem?.quantity).toBe((expectedItem as any).quantity);
expect(actualQty).toBe(expectedQty);
}
expect(actualItems.length).toBe(expectedItems.length);
@@ -299,7 +301,7 @@ export const expectSubToBeCorrect = async ({
org,
options,
existingUsage,
withEntity: !!entityId,
withEntity: !!cusProduct.internal_entity_id,
isCheckout: false,
apiVersion,
productOptions: cusProduct.quantity

View File

@@ -113,7 +113,7 @@ describe(`${chalk.yellowBright(`${testCase}: Testing separate subscriptions beca
});
const subIds: string[] = [];
test("should attach pro product", async () => {
test("should attach pro product", async () => {
for (const op of ops) {
const res = await autumn.attach({
customer_id: customerId,

View File

@@ -17,6 +17,7 @@ import { addWeeks } from "date-fns";
import type { Stripe } from "stripe";
import type { DrizzleCli } from "@/db/initDrizzle.js";
import { AutumnInt } from "@/external/autumn/autumnCli.js";
import { timeout } from "@/utils/genUtils";
import { constructArrearItem } from "@/utils/scriptUtils/constructItem.js";
import { constructProduct } from "@/utils/scriptUtils/createTestProducts.js";
import { initCustomerV3 } from "@/utils/scriptUtils/testUtils/initCustomerV3.js";
@@ -129,6 +130,7 @@ describe(`${chalk.yellowBright("mergedUpgrade1: Testing merged subs, upgrade 1 &
const entity2Val = 300000;
test("should advance test clock and upgrade entity 1 to premium, and have correct invoice", async () => {
await timeout(3000);
await autumn.track({
customer_id: customerId,
feature_id: TestFeature.Words,

View File

@@ -8,8 +8,9 @@ import {
ApiEventsListResponseSchema,
EVENTS_LIST_EXAMPLE,
} from "@api/events/list/eventsListResponse.js";
import type { ZodOpenApiPathsObject } from "zod-openapi";
export const eventsOpenApi = {
export const eventsOpenApi: ZodOpenApiPathsObject = {
"/events/list": {
post: {
summary: "List Events",

View File

@@ -8,8 +8,9 @@ import {
ApiEventsListResponseSchema,
EVENTS_LIST_EXAMPLE,
} from "@api/events/list/eventsListResponse.js";
import type { ZodOpenApiPathsObject } from "zod-openapi";
export const eventsOpenApi = {
export const eventsOpenApi: ZodOpenApiPathsObject = {
"/events/list": {
post: {
summary: "List Events",

View File

@@ -11,4 +11,7 @@ export const UpdateBalancesParamsSchema = z.object({
}),
}),
),
entity_id: z.string().optional().meta({
description: "The ID of the entity to update balance for.",
}),
});

Some files were not shown because too many files have changed in this diff Show More