feat: monthly-proration-tasker (#26870)

* feat: monthly-proration-taskerh

* remove cronjob from tasker implementation

* feat: add DI for MonthlyProrationService and TriggerDevLoggerServiceModule

- Create TriggerDevLoggerServiceModule for DI injection of TriggerDevLogger
- Add tokens for TriggerDevLogger in shared.tokens.ts
- Create MonthlyProrationService DI module and container
- Update processMonthlyProrationBatch.ts to use DI container
- Use redactError in Tasker.ts to avoid logging sensitive information

Co-Authored-By: sean@cal.com <Sean@brydon.io>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

packages/features/di/shared/services/triggerDevLogger.service.ts

@@ -0,0 +1,29 @@
+import { createModule } from "@evyweb/ioctopus";
+
+import { bindModuleToClassOnToken, type ModuleLoader } from "@calcom/features/di/di";
+import { TriggerDevLogger } from "@calcom/lib/triggerDevLogger";
+
+import { SHARED_TOKENS } from "../shared.tokens";
+
+const thisModule = createModule();
+const token = SHARED_TOKENS.TRIGGER_DEV_LOGGER;
+const moduleToken = SHARED_TOKENS.TRIGGER_DEV_LOGGER_MODULE;
+
+class TriggerDevLoggerService extends TriggerDevLogger {
+  constructor() {
+    super();
+  }
+}
+
+const loadModule = bindModuleToClassOnToken({
+  module: thisModule,
+  moduleToken,
+  token,
+  classs: TriggerDevLoggerService,
+  depsMap: {},
+});
+
+export const moduleLoader = {
+  token: SHARED_TOKENS.TRIGGER_DEV_LOGGER,
+  loadModule,
+} satisfies ModuleLoader;

packages/features/di/shared/shared.tokens.ts

@@ -3,4 +3,6 @@ export const SHARED_TOKENS = {
   TASKER: Symbol("ITasker"),
   LOGGER: Symbol("ILogger"),
   LOGGER_MODULE: Symbol("ILoggerModule"),
+  TRIGGER_DEV_LOGGER: Symbol("ITriggerDevLogger"),
+  TRIGGER_DEV_LOGGER_MODULE: Symbol("ITriggerDevLoggerModule"),
 } as const;

packages/features/ee/billing/di/containers/MonthlyProrationService.ts

@@ -0,0 +1,12 @@
+import { createContainer } from "@calcom/features/di/di";
+import { MonthlyProrationService } from "@calcom/features/ee/billing/service/proration/MonthlyProrationService";
+
+import { monthlyProrationServiceModuleLoader } from "../modules/MonthlyProrationService";
+import { DI_TOKENS } from "../tokens";
+
+const container = createContainer();
+
+export function getMonthlyProrationService(): MonthlyProrationService {
+  monthlyProrationServiceModuleLoader.loadModule(container);
+  return container.get<MonthlyProrationService>(DI_TOKENS.MONTHLY_PRORATION_SERVICE);
+}

packages/features/ee/billing/di/modules/MonthlyProrationService.ts

@@ -0,0 +1,24 @@
+import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di";
+import { moduleLoader as triggerDevLoggerServiceModule } from "@calcom/features/di/shared/services/triggerDevLogger.service";
+import { MonthlyProrationService } from "@calcom/features/ee/billing/service/proration/MonthlyProrationService";
+
+import { DI_TOKENS } from "../tokens";
+
+const thisModule = createModule();
+const token = DI_TOKENS.MONTHLY_PRORATION_SERVICE;
+const moduleToken = DI_TOKENS.MONTHLY_PRORATION_SERVICE_MODULE;
+
+const loadModule = bindModuleToClassOnToken({
+  module: thisModule,
+  moduleToken,
+  token,
+  classs: MonthlyProrationService,
+  depsMap: {
+    logger: triggerDevLoggerServiceModule,
+  },
+});
+
+export const monthlyProrationServiceModuleLoader = {
+  token,
+  loadModule,
+} satisfies ModuleLoader;

packages/features/ee/billing/di/tasker/MonthlyProrationSyncTasker.module.ts

@@ -0,0 +1,21 @@
+import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di";
+import { moduleLoader as loggerServiceModule } from "@calcom/features/di/shared/services/logger.service";
+import { MonthlyProrationSyncTasker } from "@calcom/features/ee/billing/service/proration/tasker/MonthlyProrationSyncTasker";
+
+import { MONTHLY_PRORATION_TASKER_DI_TOKENS } from "./tokens";
+
+const thisModule = createModule();
+const token = MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_SYNC_TASKER;
+const moduleToken = MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_SYNC_TASKER_MODULE;
+const loadModule = bindModuleToClassOnToken({
+  module: thisModule,
+  moduleToken,
+  token,
+  classs: MonthlyProrationSyncTasker,
+  dep: loggerServiceModule,
+});
+
+export const moduleLoader = {
+  token,
+  loadModule,
+} satisfies ModuleLoader;

packages/features/ee/billing/di/tasker/MonthlyProrationTasker.container.ts

@@ -0,0 +1,13 @@
+import { createContainer } from "@calcom/features/di/di";
+
+import type { MonthlyProrationTasker } from "@calcom/features/ee/billing/service/proration/tasker/MonthlyProrationTasker";
+
+import { moduleLoader as monthlyProrationTaskerModule } from "./MonthlyProrationTasker.module";
+import { MONTHLY_PRORATION_TASKER_DI_TOKENS } from "./tokens";
+
+const container = createContainer();
+
+export function getMonthlyProrationTasker(): MonthlyProrationTasker {
+  monthlyProrationTaskerModule.loadModule(container);
+  return container.get<MonthlyProrationTasker>(MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_TASKER);
+}

packages/features/ee/billing/di/tasker/MonthlyProrationTasker.module.ts

@@ -0,0 +1,27 @@
+import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di";
+import { moduleLoader as loggerServiceModule } from "@calcom/features/di/shared/services/logger.service";
+import { MonthlyProrationTasker } from "@calcom/features/ee/billing/service/proration/tasker/MonthlyProrationTasker";
+
+import { moduleLoader as monthlyProrationSyncTaskerModule } from "./MonthlyProrationSyncTasker.module";
+import { moduleLoader as monthlyProrationTriggerTaskerModule } from "./MonthlyProrationTriggerDevTasker.module";
+import { MONTHLY_PRORATION_TASKER_DI_TOKENS } from "./tokens";
+
+const thisModule = createModule();
+const token = MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_TASKER;
+const moduleToken = MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_TASKER_MODULE;
+const loadModule = bindModuleToClassOnToken({
+  module: thisModule,
+  moduleToken,
+  token,
+  classs: MonthlyProrationTasker,
+  depsMap: {
+    logger: loggerServiceModule,
+    asyncTasker: monthlyProrationTriggerTaskerModule,
+    syncTasker: monthlyProrationSyncTaskerModule,
+  },
+});
+
+export const moduleLoader = {
+  token,
+  loadModule,
+} satisfies ModuleLoader;

packages/features/ee/billing/di/tasker/MonthlyProrationTriggerDevTasker.module.ts

@@ -0,0 +1,23 @@
+import { bindModuleToClassOnToken, createModule, type ModuleLoader } from "@calcom/features/di/di";
+import { moduleLoader as loggerServiceModule } from "@calcom/features/di/shared/services/logger.service";
+import { MonthlyProrationTriggerDevTasker } from "@calcom/features/ee/billing/service/proration/tasker/MonthlyProrationTriggerDevTasker";
+
+import { MONTHLY_PRORATION_TASKER_DI_TOKENS } from "./tokens";
+
+const thisModule = createModule();
+const token = MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_TRIGGER_TASKER;
+const moduleToken = MONTHLY_PRORATION_TASKER_DI_TOKENS.MONTHLY_PRORATION_TRIGGER_TASKER_MODULE;
+const loadModule = bindModuleToClassOnToken({
+  module: thisModule,
+  moduleToken,
+  token,
+  classs: MonthlyProrationTriggerDevTasker,
+  depsMap: {
+    logger: loggerServiceModule,
+  },
+});
+
+export const moduleLoader = {
+  token,
+  loadModule,
+} satisfies ModuleLoader;

packages/features/ee/billing/di/tasker/tokens.ts

@@ -0,0 +1,8 @@
+export const MONTHLY_PRORATION_TASKER_DI_TOKENS = {
+  MONTHLY_PRORATION_TASKER: Symbol("MonthlyProrationTasker"),
+  MONTHLY_PRORATION_TASKER_MODULE: Symbol("MonthlyProrationTaskerModule"),
+  MONTHLY_PRORATION_SYNC_TASKER: Symbol("MonthlyProrationSyncTasker"),
+  MONTHLY_PRORATION_SYNC_TASKER_MODULE: Symbol("MonthlyProrationSyncTaskerModule"),
+  MONTHLY_PRORATION_TRIGGER_TASKER: Symbol("MonthlyProrationTriggerTasker"),
+  MONTHLY_PRORATION_TRIGGER_TASKER_MODULE: Symbol("MonthlyProrationTriggerTaskerModule"),
+};

packages/features/ee/billing/di/tokens.ts

@@ -9,4 +9,6 @@ export const DI_TOKENS = {
   TEAM_BILLING_DATA_REPOSITORY_MODULE: Symbol("TeamBillingDataRepositoryModule"),
   TEAM_BILLING_SERVICE_FACTORY: Symbol("TeamBillingServiceFactory"),
   TEAM_BILLING_SERVICE_FACTORY_MODULE: Symbol("TeamBillingServiceFactoryModule"),
+  MONTHLY_PRORATION_SERVICE: Symbol("MonthlyProrationService"),
+  MONTHLY_PRORATION_SERVICE_MODULE: Symbol("MonthlyProrationServiceModule"),
 };

packages/features/ee/billing/lib/month-key.ts

@@ -0,0 +1,5 @@
+export function formatMonthKey(date: Date): string {
+  const year = date.getUTCFullYear();
+  const month = String(date.getUTCMonth() + 1).padStart(2, "0");
+  return `${year}-${month}`;
+}

packages/features/ee/billing/service/proration/MonthlyProrationService.ts

@@ -1,4 +1,5 @@
 import stripe from "@calcom/features/ee/payments/server/stripe";
+import type { ISimpleLogger } from "@calcom/features/di/shared/services/logger.service";
 import logger from "@calcom/lib/logger";
 import type { Logger } from "tslog";
 import { MonthlyProrationRepository } from "../../repository/proration/MonthlyProrationRepository";
@@ -25,14 +26,27 @@ interface ProcessMonthlyProrationsParams {
   teamIds?: number[];
 }
 
+export interface MonthlyProrationServiceDeps {
+  logger: ISimpleLogger;
+}
+
 export class MonthlyProrationService {
-  private logger: Logger<unknown>;
+  private logger: ISimpleLogger;
   private teamRepository: MonthlyProrationTeamRepository;
   private prorationRepository: MonthlyProrationRepository;
   private billingService: IBillingProviderService;
 
-  constructor(customLogger?: Logger<unknown>, billingService?: IBillingProviderService) {
-    this.logger = customLogger || log;
+  constructor(deps: MonthlyProrationServiceDeps);
+  constructor(customLogger?: Logger<unknown>, billingService?: IBillingProviderService);
+  constructor(
+    depsOrLogger?: MonthlyProrationServiceDeps | Logger<unknown>,
+    billingService?: IBillingProviderService
+  ) {
+    if (depsOrLogger && typeof depsOrLogger === "object" && "logger" in depsOrLogger) {
+      this.logger = depsOrLogger.logger;
+    } else {
+      this.logger = (depsOrLogger as Logger<unknown>) || log;
+    }
     this.teamRepository = new MonthlyProrationTeamRepository();
     this.prorationRepository = new MonthlyProrationRepository();
     this.billingService = billingService || new StripeBillingService(stripe);

packages/features/ee/billing/service/proration/tasker/MonthlyProrationSyncTasker.ts

@@ -0,0 +1,16 @@
+import { nanoid } from "nanoid";
+import type { Logger } from "tslog";
+
+import { MonthlyProrationService } from "../MonthlyProrationService";
+import type { IMonthlyProrationTasker } from "./types";
+
+export class MonthlyProrationSyncTasker implements IMonthlyProrationTasker {
+  constructor(private readonly logger: Logger<unknown>) {}
+
+  async processBatch(payload: Parameters<IMonthlyProrationTasker["processBatch"]>[0]) {
+    const runId = `sync_${nanoid(10)}`;
+    const prorationService = new MonthlyProrationService(this.logger);
+    await prorationService.processMonthlyProrations(payload);
+    return { runId };
+  }
+}

packages/features/ee/billing/service/proration/tasker/MonthlyProrationTasker.ts

@@ -0,0 +1,22 @@
+import { Tasker } from "@calcom/lib/tasker/Tasker";
+import type { Logger } from "tslog";
+
+import type { MonthlyProrationSyncTasker } from "./MonthlyProrationSyncTasker";
+import type { MonthlyProrationTriggerDevTasker } from "./MonthlyProrationTriggerDevTasker";
+import type { IMonthlyProrationTasker, MonthlyProrationBatchPayload } from "./types";
+
+export interface MonthlyProrationTaskerDependencies {
+  asyncTasker: MonthlyProrationTriggerDevTasker;
+  syncTasker: MonthlyProrationSyncTasker;
+  logger: Logger<unknown>;
+}
+
+export class MonthlyProrationTasker extends Tasker<IMonthlyProrationTasker> {
+  constructor(dependencies: MonthlyProrationTaskerDependencies) {
+    super(dependencies);
+  }
+
+  async processBatch(payload: MonthlyProrationBatchPayload): Promise<{ runId: string }> {
+    return await this.dispatch("processBatch", payload);
+  }
+}

packages/features/ee/billing/service/proration/tasker/MonthlyProrationTriggerDevTasker.ts

@@ -0,0 +1,13 @@
+import type { ITaskerDependencies } from "@calcom/lib/tasker/types";
+
+import type { IMonthlyProrationTasker } from "./types";
+
+export class MonthlyProrationTriggerDevTasker implements IMonthlyProrationTasker {
+  constructor(public readonly dependencies: ITaskerDependencies) {}
+
+  async processBatch(payload: Parameters<IMonthlyProrationTasker["processBatch"]>[0]) {
+    const { processMonthlyProrationBatch } = await import("./trigger/processMonthlyProrationBatch");
+    const handle = await processMonthlyProrationBatch.trigger(payload);
+    return { runId: handle.id };
+  }
+}

packages/features/ee/billing/service/proration/tasker/constants.ts

@@ -0,0 +1 @@
+export const MONTHLY_PRORATION_BATCH_SIZE = 10;

packages/features/ee/billing/service/proration/tasker/trigger/config.ts

@@ -0,0 +1,20 @@
+import { queue, type schemaTask } from "@trigger.dev/sdk";
+
+type MonthlyProrationTaskConfig = Pick<Parameters<typeof schemaTask>[0], "machine" | "retry" | "queue">;
+
+export const monthlyProrationQueue = queue({
+  name: "monthly-proration",
+  concurrencyLimit: 5,
+});
+
+export const monthlyProrationTaskConfig: MonthlyProrationTaskConfig = {
+  queue: monthlyProrationQueue,
+  machine: "small-2x",
+  retry: {
+    maxAttempts: 3,
+    factor: 2,
+    minTimeoutInMs: 1_000,
+    maxTimeoutInMs: 30_000,
+    randomize: true,
+  },
+};

packages/features/ee/billing/service/proration/tasker/trigger/processMonthlyProrationBatch.ts

@@ -0,0 +1,22 @@
+import { schemaTask } from "@trigger.dev/sdk";
+
+import { monthlyProrationTaskConfig } from "./config";
+import { monthlyProrationBatchSchema } from "./schema";
+
+export const processMonthlyProrationBatch = schemaTask({
+  id: "billing.monthly-proration.batch",
+  ...monthlyProrationTaskConfig,
+  schema: monthlyProrationBatchSchema,
+  run: async (payload) => {
+    const { getMonthlyProrationService } = await import(
+      "@calcom/features/ee/billing/di/containers/MonthlyProrationService"
+    );
+
+    const prorationService = getMonthlyProrationService();
+
+    await prorationService.processMonthlyProrations({
+      monthKey: payload.monthKey,
+      teamIds: payload.teamIds,
+    });
+  },
+});

packages/features/ee/billing/service/proration/tasker/trigger/schema.ts

@@ -0,0 +1,10 @@
+import { z } from "zod";
+
+import { MONTHLY_PRORATION_BATCH_SIZE } from "../constants";
+
+const monthKeyRegex = /^\d{4}-(0[1-9]|1[0-2])$/;
+
+export const monthlyProrationBatchSchema = z.object({
+  monthKey: z.string().regex(monthKeyRegex),
+  teamIds: z.array(z.number()).min(1).max(MONTHLY_PRORATION_BATCH_SIZE),
+});

packages/features/ee/billing/service/proration/tasker/types.ts

@@ -0,0 +1,8 @@
+export type MonthlyProrationBatchPayload = {
+  monthKey: string;
+  teamIds: number[];
+};
+
+export interface IMonthlyProrationTasker {
+  processBatch(payload: MonthlyProrationBatchPayload): Promise<{ runId: string }>;
+}

packages/features/trigger.config.ts

@@ -1,3 +1,4 @@
+import process from "node:process";
 import { syncVercelEnvVars } from "@trigger.dev/build/extensions/core";
 import { defineConfig } from "@trigger.dev/sdk";
 import dotEnv from "dotenv";
@@ -15,7 +16,7 @@ export default defineConfig({
   project: process.env.TRIGGER_DEV_PROJECT_REF ?? "", // e.g., "proj_abc123"
 
   // Directories containing your tasks
-  dirs: ["./bookings/lib/tasker/trigger/notifications"], // Customize based on your project structure
+  dirs: ["./bookings/lib/tasker/trigger/notifications", "./ee/billing/service/proration/tasker/trigger"], // Customize based on your project structure
 
   // Retry configuration
   retries: {

packages/lib/tasker/Tasker.ts

@@ -1,7 +1,9 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
-import { configure } from "@trigger.dev/sdk";
 
+import process from "node:process";
+import { configure } from "@trigger.dev/sdk";
 import { ENABLE_ASYNC_TASKER } from "../constants";
+import { redactError } from "../redactError";
 import type { ILogger } from "./types";
 
 const isAsyncTaskerEnabled =
@@ -53,9 +55,13 @@ export abstract class Tasker<T> {
       const method = this.asyncTasker[taskName] as (...args: any[]) => any;
       return await method.apply(this.asyncTasker, args);
     } catch (err) {
+      const taskerLabel = isAsyncTaskerEnabled ? "AsyncTasker" : "SyncTasker";
+      const baseUrlInfo = isAsyncTaskerEnabled
+        ? ` (baseURL: ${process.env.TRIGGER_API_URL ?? "unknown"})`
+        : "";
       this.logger.error(
-        `${isAsyncTaskerEnabled ? "AsyncTasker" : "SyncTasker"} failed for '${String(taskName)}'.`,
-        (err as Error)?.message ?? "ERROR MESSAGE UNAVAILABLE"
+        `${taskerLabel} failed for '${String(taskName)}'.${baseUrlInfo}`,
+        this.getErrorDetails(err)
       );
 
       if (this.asyncTasker === this.syncTasker) {
@@ -68,12 +74,22 @@ export abstract class Tasker<T> {
         const fallbackMethod = this.syncTasker[taskName] as (...args: any[]) => any;
         return await fallbackMethod.apply(this.syncTasker, args);
       } catch (err) {
-        this.logger.error(
-          `SyncTasker failed for '${String(taskName)}'.`,
-          (err as Error)?.message ?? "ERROR MESSAGE UNAVAILABLE"
-        );
+        this.logger.error(`SyncTasker failed for '${String(taskName)}'.`, this.getErrorDetails(err));
         throw err;
       }
     }
   }
+
+  private getErrorDetails(err: unknown) {
+    const redactedError = redactError(err);
+    if (redactedError instanceof Error) {
+      return {
+        name: redactedError.name,
+        message: redactedError.message,
+        stack: redactedError.stack,
+      };
+    }
+
+    return { message: String(redactedError) };
+  }
 }