Merge branch 'useautumn:main' into fix/github-ready

2025-06-16 17:57:42 +02:00
parent a9e34f2d3a a1cd466f31
commit 544f641f8d
34 changed files with 234 additions and 166 deletions
--- a/package-lock.json
+++ b/package-lock.json
@@ -10802,7 +10802,7 @@
        "@types/cors": "^2.8.17",
        "@types/express": "^5.0.0",
        "ai": "^4.3.10",
-        "autumn-js": "^0.0.46",
+        "autumn-js": "^0.0.64",
        "body-parser": "^1.20.3",
        "bullmq": "^5.31.1",
        "chai": "^5.1.2",
@@ -10854,12 +10854,13 @@
      }
    },
    "server/node_modules/autumn-js": {
-      "version": "0.0.46",
-      "resolved": "https://registry.npmjs.org/autumn-js/-/autumn-js-0.0.46.tgz",
-      "integrity": "sha512-5RDn1l+4XMtYQIvjASi1und+SrtCiAZ61CGajptrmySxx2xxrvsBFlnIfwq9CeO4/oOLTMcuuW3qJCAh9/a5RA==",
+      "version": "0.0.64",
+      "resolved": "https://registry.npmjs.org/autumn-js/-/autumn-js-0.0.64.tgz",
+      "integrity": "sha512-Fa5lr9A0ywYNcbny/dQBRKSGaqnTuUvqtiBegHrr5Z3wCw9A/Z2LRH/f8AqHYQSFOXVkS763ngLlGLxfJpYgQQ==",
      "license": "MIT",
      "dependencies": {
-        "rou3": "^0.6.1"
+        "rou3": "^0.6.1",
+        "swr": "^2.3.3"
      },
      "peerDependencies": {
        "@tanstack/react-query": "^5.76.1",
--- a/server/package.json
+++ b/server/package.json
@@ -45,7 +45,7 @@
    "@types/cors": "^2.8.17",
    "@types/express": "^5.0.0",
    "ai": "^4.3.10",
-    "autumn-js": "^0.0.46",
+    "autumn-js": "^0.0.64",
    "body-parser": "^1.20.3",
    "bullmq": "^5.31.1",
    "chai": "^5.1.2",
--- a/server/src/db/initDrizzle.ts
+++ b/server/src/db/initDrizzle.ts
@@ -5,8 +5,11 @@ import postgres from "postgres";
 import { drizzle } from "drizzle-orm/postgres-js";
 import { schemas } from "@autumn/shared";

-export const initDrizzle = () => {
-  const client = postgres(process.env.DATABASE_URL!);
+export const initDrizzle = (params?: { maxConnections?: number }) => {
+  let maxConnections = params?.maxConnections || 10;
+  const client = postgres(process.env.DATABASE_URL!, {
+    max: maxConnections,
+  });

  const db = drizzle(client, {
    schema: schemas,
--- a/server/src/external/stripe/priceToStripeItem/priceToStripeItem.ts
+++ b/server/src/external/stripe/priceToStripeItem/priceToStripeItem.ts
@@ -10,7 +10,6 @@ import {
  FeatureOptions,
  FixedPriceConfig,
  FullProduct,
-  InsertReplaceable,
  Organization,
  UsagePriceConfig,
 } from "@autumn/shared";
--- a/server/src/external/stripe/stripeSubUtils/createStripeSub.ts
+++ b/server/src/external/stripe/stripeSubUtils/createStripeSub.ts
@@ -26,7 +26,6 @@ export const createStripeSub = async ({
  invoiceOnly = false,
  anchorToUnix,
  itemSet,
-  shouldPreview = false,
  now,
 }: {
  db: DrizzleCli;
@@ -37,7 +36,6 @@ export const createStripeSub = async ({
  invoiceOnly?: boolean;
  anchorToUnix?: number;
  itemSet: ItemSet;
-  shouldPreview?: boolean;
  now?: number;
 }) => {
  let paymentMethod = await getCusPaymentMethod({
--- a/server/src/external/stripe/webhookHandlers/handleCusDiscountDeleted.ts
+++ b/server/src/external/stripe/webhookHandlers/handleCusDiscountDeleted.ts
@@ -35,58 +35,61 @@ export async function handleCusDiscountDeleted({
    return;
  }

-  // Check if any redemptions available, and apply to customer if so
-  let redemptions = await RewardRedemptionService.getUnappliedRedemptions({
-    db,
-    internalCustomerId: customer.internal_id,
-  });
-
-  if (redemptions.length == 0) {
-    return;
-  }
-
-  let redemption = redemptions[0];
-  let reward = redemption.reward_program.reward;
-
-  // Apply redemption to customer
-  let stripeCli = createStripeCli({
-    org,
-    env,
-  });
-
-  let stripeCus = (await stripeCli.customers.retrieve(
-    discount.customer,
-  )) as Stripe.Customer;
-
-  if (stripeCus && notNullish(stripeCus.discount)) {
-    logger.info(
-      `discount.deleted: stripe customer ${discount.customer} already has a discount`,
-    );
-    return;
-  }
-
-  // Send response first...?
  res.status(200).json({ message: "OK" });
+  return;

-  if (notNullish(stripeCus.test_clock)) {
-    // Time out for test clock to complete
-    await timeout(5000);
-  }
+  // // Check if any redemptions available, and apply to customer if so
+  // let redemptions = await RewardRedemptionService.getUnappliedRedemptions({
+  //   db,
+  //   internalCustomerId: customer.internal_id,
+  // });

-  await stripeCli.customers.update(discount.customer, {
-    coupon: reward.internal_id,
-  });
+  // if (redemptions.length == 0) {
+  //   return;
+  // }

-  await RewardRedemptionService.update({
-    db,
-    id: redemption.id,
-    updates: {
-      applied: true,
-    },
-  });
+  // let redemption = redemptions[0];
+  // let reward = redemption.reward_program.reward;

-  logger.info(
-    `discount.deleted: applied reward ${reward.name} on customer ${customer.name} (${customer.id})`,
-  );
-  logger.info(`Redemption ID: ${redemption.id}`);
+  // // Apply redemption to customer
+  // let stripeCli = createStripeCli({
+  //   org,
+  //   env,
+  // });
+
+  // let stripeCus = (await stripeCli.customers.retrieve(
+  //   discount.customer,
+  // )) as Stripe.Customer;
+
+  // if (stripeCus && notNullish(stripeCus.discount)) {
+  //   logger.info(
+  //     `discount.deleted: stripe customer ${discount.customer} already has a discount`,
+  //   );
+  //   return;
+  // }
+
+  // // Send response first...?
+  // res.status(200).json({ message: "OK" });
+
+  // if (notNullish(stripeCus.test_clock)) {
+  //   // Time out for test clock to complete
+  //   await timeout(5000);
+  // }
+
+  // await stripeCli.customers.update(discount.customer, {
+  //   coupon: reward.internal_id,
+  // });
+
+  // await RewardRedemptionService.update({
+  //   db,
+  //   id: redemption.id,
+  //   updates: {
+  //     applied: true,
+  //   },
+  // });
+
+  // logger.info(
+  //   `discount.deleted: applied reward ${reward.name} on customer ${customer.name} (${customer.id})`,
+  // );
+  // logger.info(`Redemption ID: ${redemption.id}`);
 }
--- a/server/src/external/stripe/webhookHandlers/handleSubUpdated.ts
+++ b/server/src/external/stripe/webhookHandlers/handleSubUpdated.ts
@@ -38,22 +38,7 @@ export const handleSubscriptionUpdated = async ({
  previousAttributes: any;
  logger: any;
 }) => {
-  const lockKey = `sub_updated_${subscription.id}`;
-
-  // Handle syncing status
-  let stripeCli = createStripeCli({
-    org,
-    env,
-  });
-  let fullSub = await stripeCli.subscriptions.retrieve(subscription.id);
-
-  let subStatusMap: {
-    [key: string]: CusProductStatus;
-  } = {
-    trialing: CusProductStatus.Active,
-    active: CusProductStatus.Active,
-    past_due: CusProductStatus.PastDue,
-  };
+  // const lockKey = `sub_updated_${subscription.id}`;

  // Get cus products by stripe sub id
  const cusProducts = await CusProductService.getByStripeSubId({
@@ -71,37 +56,52 @@ export const handleSubscriptionUpdated = async ({
    return;
  }

-  // Create a lock to prevent race conditions
-  let lockAcquired = false;
-  try {
-    let attempts = 0;
+  // // Create a lock to prevent race conditions
+  // let lockAcquired = false;
+  // try {
+  //   let attempts = 0;

-    while (!lockAcquired && attempts < 3) {
-      lockAcquired = await getWebhookLock({ lockKey, logger });
-      if (!lockAcquired) {
-        attempts++;
-        console.log(
-          `sub.updated: failed to acquire lock for ${subscription.id}, attempt ${attempts}`,
-        );
-        if (attempts < 3) {
-          await new Promise((resolve) => setTimeout(resolve, 1000));
-        }
-      } else {
-        break;
-      }
-    }
-  } catch (error) {
-    logger.error("lock error, setting lockAcquired to true");
-    lockAcquired = true;
-  }
+  //   while (!lockAcquired && attempts < 3) {
+  //     lockAcquired = await getWebhookLock({ lockKey, logger });
+  //     if (!lockAcquired) {
+  //       attempts++;
+  //       console.log(
+  //         `sub.updated: failed to acquire lock for ${subscription.id}, attempt ${attempts}`,
+  //       );
+  //       if (attempts < 3) {
+  //         await new Promise((resolve) => setTimeout(resolve, 1000));
+  //       }
+  //     } else {
+  //       break;
+  //     }
+  //   }
+  // } catch (error) {
+  //   logger.error("lock error, setting lockAcquired to true");
+  //   lockAcquired = true;
+  // }

-  if (!lockAcquired) {
-    throw new RecaseError({
-      message: `Failed to acquire lock for stripe webhook, sub.updated.`,
-      code: ErrCode.InvalidRequest,
-      statusCode: 400,
-    });
-  }
+  // if (!lockAcquired) {
+  //   throw new RecaseError({
+  //     message: `Failed to acquire lock for stripe webhook, sub.updated.`,
+  //     code: ErrCode.InvalidRequest,
+  //     statusCode: 400,
+  //   });
+  // }
+
+  // Handle syncing status
+  let stripeCli = createStripeCli({
+    org,
+    env,
+  });
+  let fullSub = await stripeCli.subscriptions.retrieve(subscription.id);
+
+  let subStatusMap: {
+    [key: string]: CusProductStatus;
+  } = {
+    trialing: CusProductStatus.Active,
+    active: CusProductStatus.Active,
+    past_due: CusProductStatus.PastDue,
+  };

  // 1. Fetch subscription
  const updatedCusProducts = await CusProductService.updateByStripeSubId({
@@ -179,5 +179,5 @@ export const handleSubscriptionUpdated = async ({
    }
  }

-  await releaseWebhookLock({ lockKey, logger });
+  // await releaseWebhookLock({ lockKey, logger });
 };
--- a/server/src/index.ts
+++ b/server/src/index.ts
@@ -1,6 +1,7 @@
 import { config } from "dotenv";
 config();

+import http from "http";
 import cluster from "cluster";
 import os from "os";
 import mainRouter from "./internal/mainRouter.js";
@@ -18,12 +19,10 @@ import {
  createLogtail,
  createLogtailAll,
 } from "./external/logtail/logtailUtils.js";
-import { format } from "date-fns";
 import { CacheManager } from "./external/caching/CacheManager.js";
 import { initDrizzle } from "./db/initDrizzle.js";
 import { createPosthogCli } from "./external/posthog/createPosthogCli.js";
-import pg from "pg";
-import http from "http";
+
 import { generateId } from "./utils/genUtils.js";
 import { subscribeToOrgUpdates } from "./external/supabase/subscribeToOrgUpdates.js";

@@ -32,6 +31,8 @@ if (!process.env.DATABASE_URL) {
  process.exit(1);
 }

+const { db, client } = initDrizzle({ maxConnections: 10 });
+
 const init = async () => {
  const app = express();
  const logger = initLogger();
@@ -43,7 +44,6 @@ const init = async () => {
  await CacheManager.getInstance();

  const supabaseClient = createSupabaseClient();
-  const { db } = initDrizzle();

  // Optional services
  const logtailAll = createLogtailAll();
@@ -127,13 +127,15 @@ const init = async () => {

 if (process.env.NODE_ENV === "development") {
  init();
+  registerShutdownHandlers();
 } else {
  let numCPUs = os.cpus().length;

  if (cluster.isPrimary) {
    console.log(`Master ${process.pid} is running`);
    console.log("Number of CPUs", numCPUs);
-    let numWorkers = Math.min(numCPUs, 3);
+
+    let numWorkers = 8;

    for (let i = 0; i < numWorkers; i++) {
      cluster.fork();
@@ -152,5 +154,24 @@ if (process.env.NODE_ENV === "development") {
    });
  } else {
    init();
+    registerShutdownHandlers();
+  }
+}
+
+function registerShutdownHandlers() {
+  process.on("SIGTERM", gracefulShutdown);
+  process.on("SIGINT", gracefulShutdown);
+  // Do NOT use process.on("exit", ...) for async cleanup!
+}
+
+async function gracefulShutdown() {
+  console.log("Shutting down worker, closing DB connections...");
+  try {
+    await client.end();
+    console.log("DB connection closed. Exiting process.");
+    process.exit(0);
+  } catch (err) {
+    console.error("Error closing DB connection:", err);
+    process.exit(1);
  }
 }
--- a/server/src/internal/api/apiRouter.ts
+++ b/server/src/internal/api/apiRouter.ts
@@ -19,7 +19,7 @@ import { componentRouter } from "./components/componentRouter.js";
 import { analyticsMiddleware } from "@/middleware/analyticsMiddleware.js";

 import rewardRouter from "./rewards/rewardRouter.js";
-import expireRouter from "./customers/products/expireRouter.js";
+import expireRouter from "../customers/expire/expireRouter.js";

 const apiRouter = Router();

--- a/server/src/internal/customers/add-product/createFullCusProduct.ts
+++ b/server/src/internal/customers/add-product/createFullCusProduct.ts
@@ -366,7 +366,6 @@ export const createFullCusProduct = async ({
  for (const entitlement of entitlements) {
    const options = getEntOptions(optionsList, entitlement);
    const relatedPrice = getEntRelatedPrice(entitlement, prices);
-    const now = attachParams.now || Date.now();

    const cusEnt: any = initCusEntitlement({
      entitlement,
@@ -384,7 +383,7 @@ export const createFullCusProduct = async ({
      carryExistingUsages,
      curCusProduct: curCusProduct as FullCusProduct,
      replaceables: attachReplaceables,
-      now,
+      now: attachParams.now,
    });

    cusEnts.push(cusEnt);
--- a/server/src/internal/customers/attach/attachFunctions/upgradeDiffIntFlow/updateCurSchedules.ts
+++ b/server/src/internal/customers/attach/attachFunctions/upgradeDiffIntFlow/updateCurSchedules.ts
@@ -43,10 +43,7 @@ export const updateCurSchedules = async ({

    // If schedule has passed, skip this step.
    let phase = schedule.phases.length > 0 ? schedule.phases[0] : null;
-    let now = await getStripeNow({
-      stripeCli,
-      testClockId: schedule.test_clock as string,
-    });
+    let now = attachParams.now || Date.now();

    if (phase && phase.start_date * 1000 < now) {
      logger.info("Note: Schedule has passed, skipping");
--- a/server/src/internal/customers/attach/attachFunctions/upgradeDiffIntFlow/updateSubsDiffInt.ts
+++ b/server/src/internal/customers/attach/attachFunctions/upgradeDiffIntFlow/updateSubsDiffInt.ts
@@ -48,6 +48,8 @@ export const updateSubsDiffInt = async ({
    itemSet: firstItemSet,
  });

+  // throw new Error("Stop");
+
  let trialEnd = config.disableTrial
    ? undefined
    : freeTrialToStripeTimestamp({
@@ -87,6 +89,7 @@ export const updateSubsDiffInt = async ({
  const newInvoiceIds = latestInvoice ? [latestInvoice.id] : [];

  // 4. Update current sub schedules if exist...
+  logger.info("1.3 Updating current sub schedules");
  await updateCurSchedules({
    db,
    stripeCli,
@@ -119,7 +122,7 @@ export const updateSubsDiffInt = async ({
      itemSet,
      invoiceOnly: attachParams.invoiceOnly || false,
      freeTrial: attachParams.freeTrial,
-      anchorToUnix: updatedSub!.current_period_end! * 1000,
+      // anchorToUnix: updatedSub!.current_period_end! * 1000,
      now: attachParams.now,
    });

--- a/server/src/internal/customers/attach/attachFunctions/upgradeSameIntFlow/updateSubsSameInt.ts
+++ b/server/src/internal/customers/attach/attachFunctions/upgradeSameIntFlow/updateSubsSameInt.ts
@@ -34,17 +34,17 @@ export const updateSubsByInt = async ({

  attachParams.replaceables = replaceables;

-  logger.info(`Cont use items`);
-  logger.info(
-    `New items: `,
-    newItems.map(
-      (item) => `${item.description} | Amount: ${item.amount || item.price}`,
-    ),
-  );
-  logger.info(
-    "Replaceables: ",
-    replaceables.map((r) => `${r.ent.feature_id}`),
-  );
+  // logger.info(`Cont use items`);
+  // logger.info(
+  //   `New items: `,
+  //   newItems.map(
+  //     (item) => `${item.description} | Amount: ${item.amount || item.price}`,
+  //   ),
+  // );
+  // logger.info(
+  //   "Replaceables: ",
+  //   replaceables.map((r) => `${r.ent.feature_id}`),
+  // );

  const itemSets = await getStripeSubItems({ attachParams });
  const invoices: Stripe.Invoice[] = [];
--- a/server/src/internal/customers/attach/attachUtils/attachParams/attachParamsUtils/getStripeCusData.ts
+++ b/server/src/internal/customers/attach/attachUtils/attachParams/attachParamsUtils/getStripeCusData.ts
@@ -19,7 +19,9 @@ export const getStripeCusData = async ({
  let stripeCusData = stripeCus as Stripe.Customer;
  let testClock =
    stripeCusData.test_clock as Stripe.TestHelpers.TestClock | null;
-  let now = testClock ? testClock.frozen_time * 1000 : Date.now();
+
+  // let now = testClock ? testClock.frozen_time * 1000 : Date.now();
+  let now = testClock ? testClock.frozen_time * 1000 : undefined;

  let paymentMethod = stripeCusData.invoice_settings
    ?.default_payment_method as Stripe.PaymentMethod | null;
--- a/server/src/internal/customers/attach/attachUtils/attachParams/getAttachParams.ts
+++ b/server/src/internal/customers/attach/attachUtils/attachParams/getAttachParams.ts
@@ -1,4 +1,3 @@
-import { listCusPaymentMethods } from "@/external/stripe/stripeCusUtils.js";
 import { ExtendedRequest } from "@/utils/models/Request.js";
 import { AttachBody } from "../../models/AttachBody.js";
 import { processAttachBody } from "./processAttachBody.js";
--- a/server/src/internal/customers/attach/attachUtils/attachParams/processAttachBody.ts
+++ b/server/src/internal/customers/attach/attachUtils/attachParams/processAttachBody.ts
@@ -199,8 +199,6 @@ const getPricesAndEnts = async ({
    productId: product.id,
  });

-  const prodIsMain = isMainProduct({ product: products[0], prices });
-
  return {
    optionsList: mapOptionsList({
      optionsInput: optionsInput || [],
--- a/server/src/internal/customers/attach/attachUtils/getContUseItems/getContUseDowngradeItems.ts
+++ b/server/src/internal/customers/attach/attachUtils/getContUseItems/getContUseDowngradeItems.ts
@@ -34,7 +34,6 @@ export const getContUseDowngradeItems = async ({
  proration?: Proration;
  logger: any;
 }) => {
-  let now = attachParams.now || Date.now();
  let prevInvoiceItem = curItem;
  let prevBalance = prevCusEnt.entitlement.allowance! - curUsage;
  const product = attachParamsToProduct({ attachParams });
@@ -66,7 +65,7 @@ export const getContUseDowngradeItems = async ({
      usage: newUsage,
      prodName: product.name,
      proration,
-      now,
+      now: attachParams.now,
      allowNegative: false,
    });

@@ -85,7 +84,7 @@ export const getContUseDowngradeItems = async ({
    usage: newUsage,
    prodName: product.name,
    proration,
-    now,
+    now: attachParams.now,
  });

  let numReplaceables = newUsage - prevUsage;
--- a/server/src/internal/customers/attach/attachUtils/getContUseItems/getContUseInvoiceItems.ts
+++ b/server/src/internal/customers/attach/attachUtils/getContUseItems/getContUseInvoiceItems.ts
@@ -116,8 +116,6 @@ export const getContUseInvoiceItems = async ({
  attachParams: AttachParams;
  logger: any;
 }) => {
-  const now = attachParams.now || Date.now();
-
  const cusPrices = cusProduct ? cusProduct.customer_prices : [];
  const cusEnts = cusProduct ? cusProduct.customer_entitlements : [];

@@ -127,7 +125,6 @@ export const getContUseInvoiceItems = async ({
    ? await getCurContUseItems({
        stripeSubs,
        attachParams,
-        now,
      })
    : [];

--- a/server/src/internal/customers/attach/attachUtils/getContUseItems/getContUseUpgradeItems.ts
+++ b/server/src/internal/customers/attach/attachUtils/getContUseItems/getContUseUpgradeItems.ts
@@ -36,7 +36,6 @@ export const getContUseUpgradeItems = async ({
  proration?: Proration;
  logger: any;
 }) => {
-  let now = attachParams.now || Date.now();
  let prevInvoiceItem = curItem;
  let prevBalance = prevCusEnt.entitlement.allowance! - curUsage;
  let newBalance = ent.allowance! - curUsage;
@@ -70,7 +69,7 @@ export const getContUseUpgradeItems = async ({
    usage: newUsage,
    prodName: product.name,
    proration,
-    now,
+    now: attachParams.now,
  });

  const featureName = usageToFeatureName({
--- a/server/src/internal/customers/cusProducts/cusProductUtils.ts
+++ b/server/src/internal/customers/cusProducts/cusProductUtils.ts
@@ -50,6 +50,7 @@ export const cancelCusProductSubscriptions = async ({
  excludeIds,
  expireImmediately = true,
  logger,
+  prorate = true,
 }: {
  cusProduct: FullCusProduct;
  org: Organization;
@@ -57,6 +58,7 @@ export const cancelCusProductSubscriptions = async ({
  excludeIds?: string[];
  expireImmediately?: boolean;
  logger: any;
+  prorate?: boolean;
 }) => {
  // 1. Cancel all subscriptions
  const stripeCli = createStripeCli({
@@ -81,7 +83,9 @@ export const cancelCusProductSubscriptions = async ({

    try {
      if (expireImmediately) {
-        await stripeCli.subscriptions.cancel(subId);
+        await stripeCli.subscriptions.cancel(subId, {
+          prorate: prorate,
+        });
      } else {
        await stripeCli.subscriptions.update(subId, {
          cancel_at: latestSubEnd || undefined,
--- a/server/src/internal/api/customers/products/expireRouter.ts
+++ b/server/src/internal/api/customers/products/expireRouter.ts
@@ -4,7 +4,7 @@ import RecaseError from "@/utils/errorUtils.js";
 import { routeHandler } from "@/utils/routerUtils.js";
 import { CusProductStatus, ErrCode, FullCusProduct } from "@autumn/shared";
 import { Router } from "express";
-import { expireCusProduct } from "../../../customers/handlers/handleCusProductExpired.js";
+import { expireCusProduct } from "../handlers/handleCusProductExpired.js";

 const expireRouter = Router();

@@ -18,6 +18,7 @@ expireRouter.post("", async (req, res) =>
      let { customer_id, product_id, entity_id, cancel_immediately } = req.body;

      let expireImmediately = cancel_immediately || false;
+      let prorate = true;

      let [customer, org] = await Promise.all([
        CusService.getFull({
@@ -66,6 +67,7 @@ expireRouter.post("", async (req, res) =>
          logger,
          customer,
          expireImmediately,
+          prorate,
        });
      }

--- a/server/src/internal/customers/handlers/handleCusProductExpired.ts
+++ b/server/src/internal/customers/handlers/handleCusProductExpired.ts
@@ -78,6 +78,7 @@ export const expireCusProduct = async ({
  logger,
  customer,
  expireImmediately = true,
+  prorate,
 }: {
  req: ExtendedRequest;
  db: DrizzleCli;
@@ -88,6 +89,7 @@ export const expireCusProduct = async ({
  logger: any;
  customer: Customer;
  expireImmediately: boolean;
+  prorate: boolean;
 }) => {
  logger.info("--------------------------------");
  logger.info(
@@ -186,6 +188,7 @@ export const expireCusProduct = async ({
    org,
    env,
    logger,
+    prorate,
  });

  if (!cancelled) {
@@ -245,6 +248,7 @@ export const handleCusProductExpired = async (req: any, res: any) => {
      logger: req.logtail,
      customer: cusProduct.customer!,
      expireImmediately: true,
+      prorate: true,
    });

    res.status(200).json({ message: "Product expired" });
--- a/server/src/internal/invoices/previewItemUtils/getCurContUseItems.ts
+++ b/server/src/internal/invoices/previewItemUtils/getCurContUseItems.ts
@@ -29,11 +29,9 @@ import { getPriceEntitlement } from "@/internal/products/prices/priceUtils.js";
 export const getCurContUseItems = async ({
  stripeSubs,
  attachParams,
-  now,
 }: {
  stripeSubs: Stripe.Subscription[];
  attachParams: AttachParams;
-  now: number;
 }) => {
  const { features } = attachParams;
  const { curMainProduct } = attachParamToCusProducts({ attachParams });
@@ -42,6 +40,7 @@ export const getCurContUseItems = async ({
  const curEnts = cusProductToEnts({ cusProduct: curCusProduct });

  let items: PreviewLineItem[] = [];
+  let now = attachParams.now || Date.now();

  for (const sub of stripeSubs) {
    for (const item of sub.items.data) {
--- a/server/src/internal/products/prices/billingIntervalUtils.ts
+++ b/server/src/internal/products/prices/billingIntervalUtils.ts
@@ -1,6 +1,8 @@
 import { BillingInterval } from "@autumn/shared";
 import {
+  addMinutes,
  addMonths,
+  addSeconds,
  addYears,
  differenceInSeconds,
  getDate,
@@ -91,6 +93,8 @@ export const getAlignedIntervalUnix = ({
  now?: number;
  alwaysReturn?: boolean;
 }) => {
+  // alignWithUnix = addSeconds(alignWithUnix, 20).getTime();
+
  let nextCycleAnchorUnix = alignWithUnix;

  now = now || Date.now();
@@ -109,6 +113,8 @@ export const getAlignedIntervalUnix = ({
      interval,
    );

+    // console.log("Subtracted unix:", formatUnixToDateTime(subtractedUnix));
+
    if (subtractedUnix <= now) {
      break;
    }
@@ -126,12 +132,17 @@ export const getAlignedIntervalUnix = ({
  // console.log("Next cycle anchor:", formatUnixToDateTime(nextCycleAnchorUnix));
  // console.log("--------------------------------");

-  if (
-    differenceInSeconds(
-      new Date(naturalBillingDate),
-      new Date(nextCycleAnchorUnix),
-    ) < 60
-  ) {
+  let anchorAndNaturalDiff = differenceInSeconds(
+    naturalBillingDate,
+    nextCycleAnchorUnix,
+  );
+
+  // For insurance, also means you can't set billing cycle anchor to a minute in the future...
+  let anchorAndNowDiff = Math.abs(
+    differenceInSeconds(now, nextCycleAnchorUnix),
+  );
+
+  if (anchorAndNaturalDiff < 60 || anchorAndNowDiff < 20) {
    if (alwaysReturn) {
      return naturalBillingDate;
    } else {
--- a/server/src/queue/QueueManager.ts
+++ b/server/src/queue/QueueManager.ts
@@ -33,6 +33,11 @@ export class QueueManager {
  }) {
    // 1. Connect to redis

+    if (useBackup && !process.env.REDIS_BACKUP_URL) {
+      console.warn(`REDIS_BACKUP_URL not set, using main redis`);
+      useBackup = false;
+    }
+
    const redisUrl = useBackup
      ? process.env.REDIS_BACKUP_URL
      : process.env.REDIS_URL;
@@ -47,7 +52,7 @@ export class QueueManager {
      console.log(
        `Redis connection error (${useBackup ? "backup" : "main"}): ${
          error.message
-        }`
+        }`,
      );

      if (!keepConnection) {
--- a/server/src/queue/workersInit.ts
+++ b/server/src/queue/workersInit.ts
@@ -13,13 +13,15 @@ import { DrizzleCli, initDrizzle } from "@/db/initDrizzle.js";
 import { acquireLock, getRedisConnection, releaseLock } from "./lockUtils.js";
 import { runActionHandlerTask } from "@/internal/analytics/runActionHandlerTask.js";

-const NUM_WORKERS = 5;
+const NUM_WORKERS = 15;

 const actionHandlers = [
  JobName.HandleProductsUpdated,
  JobName.HandleCustomerCreated,
 ];

+const { db, client } = initDrizzle({ maxConnections: 20 });
+
 const initWorker = ({
  id,
  queue,
@@ -190,7 +192,6 @@ export const initWorkers = async () => {
  const backupQueue = await QueueManager.getQueue({ useBackup: true });
  await CacheManager.getInstance();
  const logtail = createLogtail();
-  const { db, client } = initDrizzle();

  for (let i = 0; i < NUM_WORKERS; i++) {
    workers.push(
--- a/server/src/utils/scriptUtils/createTestProducts.ts
+++ b/server/src/utils/scriptUtils/createTestProducts.ts
@@ -7,20 +7,12 @@ import {
  AppEnv,
  BillingInterval,
  CreateFreeTrialSchema,
-  Feature,
  FeatureUsageType,
  FreeTrialDuration,
-  Product,
  ProductItem,
  ProductV2,
 } from "@autumn/shared";
-import {
-  constructArrearItem,
-  constructArrearProratedItem,
-} from "./constructItem.js";
-import { constructPrepaidItem } from "./constructItem.js";
 import { keyToTitle } from "../genUtils.js";
-import { FeatureService } from "@/internal/features/FeatureService.js";
 import { constructPriceItem } from "@/internal/products/product-items/productItemUtils.js";

 export enum TestFeatureType {
--- a/server/src/utils/scriptUtils/logUtils/logSubItems.ts
+++ b/server/src/utils/scriptUtils/logUtils/logSubItems.ts
@@ -0,0 +1,20 @@
+import Stripe from "stripe";
+import {
+  stripeToAutumnInterval,
+  subItemToAutumnInterval,
+} from "tests/utils/stripeUtils.js";
+
+export const logSubItems = (sub: Stripe.Subscription) => {
+  for (const item of sub.items.data) {
+    let isMetered = item.price.recurring?.usage_type === "metered";
+    let isTiered = item.price.billing_scheme === "tiered";
+
+    if (isMetered) {
+      console.log(`Usage price`);
+    } else {
+      let price = item.price.unit_amount! / 100;
+      let interval = subItemToAutumnInterval(item);
+      console.log(`${price} / ${interval}`);
+    }
+  }
+};
--- a/server/src/workers.ts
+++ b/server/src/workers.ts
@@ -7,3 +7,5 @@ const init = async () => {
 };

 init();
+
+//
--- a/server/tests/attach/upgrade/upgrade6.ts
+++ b/server/tests/attach/upgrade/upgrade6.ts
@@ -133,6 +133,7 @@ describe(`${chalk.yellowBright(`${testCase}: Testing failed upgrades`)}`, () =>
      errMessage: "Failed to update subscription. Your card was declined.",
    });

+    await timeout(4000);
    let customer = await autumn.customers.get(customerId);

    expectProductAttached({
--- a/server/tests/contUse/update/updateContUse5.ts
+++ b/server/tests/contUse/update/updateContUse5.ts
@@ -142,6 +142,7 @@ describe(`${chalk.yellowBright(`contUse/${testCase}: Testing update contUse incl
      advanceTo: addWeeks(curUnix, 2).getTime(),
      waitForSeconds: 5,
    });
+    return;

    await attachAndExpectCorrect({
      autumn,
--- a/server/tests/utils/expectUtils/expectSubUtils.ts
+++ b/server/tests/utils/expectUtils/expectSubUtils.ts
@@ -224,9 +224,10 @@ export const expectSubItemsCorrect = async ({

  for (const sub of subs.slice(1)) {
    let dateOfAnchor = getDate(sub.current_period_end * 1000);
-    expect(dateOfAnchor).to.equal(
+    expect(dateOfAnchor).to.approximately(
      firstDate,
-      `subscription anchors are the same`,
+      5000,
+      `subscription anchors are the same, +/- 5s`,
    );
  }

--- a/server/tests/utils/stripeUtils.ts
+++ b/server/tests/utils/stripeUtils.ts
@@ -467,3 +467,10 @@ export const stripeToAutumnInterval = ({
    return BillingInterval.Year;
  }
 };
+
+export const subItemToAutumnInterval = (item: Stripe.SubscriptionItem) => {
+  return stripeToAutumnInterval({
+    interval: item.price.recurring?.interval!,
+    intervalCount: item.price.recurring?.interval_count!,
+  });
+};
--- a/vite/src/views/customers/customer/product/components/AttachModal.tsx
+++ b/vite/src/views/customers/customer/product/components/AttachModal.tsx
@@ -138,7 +138,7 @@ export const AttachModal = ({
        optionsInput: options,
        attachState,
        useInvoice,
-        successUrl: `${import.meta.env.VITE_PUBLIC_FRONTEND_URL}${redirectUrl}`,
+        successUrl: `${import.meta.env.VITE_FRONTEND_URL}${redirectUrl}`,
        version,
      });