chore: enable/disable slots workers via env (#26621)

* chore: enable/disable slots workers via env * fix: address Cubic AI review feedback - Fix incorrect JSDoc comment for getSerializableContext method - Remove debug console.log statement from slots controller - Fix port suffix condition to preserve original behavior Co-Authored-By: unknown <> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-01-09 15:03:13 +00:00
parent 87a2f855f8
commit 8666efa29d
6 changed files with 125 additions and 81 deletions
@@ -3,18 +3,22 @@ import { getEnv } from "@/env";
 import type { AppConfig } from "./type";

 const loadConfig = (): AppConfig => {
+  const env = getEnv("NODE_ENV", "development");
+  const apiPort = Number(getEnv("API_PORT", "5555"));
+  const apiUrl = getEnv("API_URL", "http://localhost");
+  let portSuffix = "";
+  if (process.env.API_PORT && env === "development") {
+    portSuffix = `:${apiPort}`;
+  }
+
  return {
    env: {
-      type: getEnv("NODE_ENV", "development"),
+      type: env,
    },
    api: {
-      port: Number(getEnv("API_PORT", "5555")),
-      path: getEnv("API_URL", "http://localhost"),
-      url: `${getEnv("API_URL", "http://localhost")}${
-        process.env.API_PORT && getEnv("NODE_ENV", "development") === "development"
-          ? `:${Number(getEnv("API_PORT", "5555"))}`
-          : ""
-      }/v2`,
+      port: apiPort,
+      path: apiUrl,
+      url: `${apiUrl}${portSuffix}/v2`,
      keyPrefix: getEnv("API_KEY_PREFIX", "cal_"),
      licenseKey: getEnv("CALCOM_LICENSE_KEY", ""),
      licenseKeyUrl: getEnv("GET_LICENSE_KEY_URL", "https://console.cal.com/api/license"),
@@ -40,7 +44,9 @@ const loadConfig = (): AppConfig => {
    app: {
      baseUrl: getEnv("WEB_APP_URL", "https://app.cal.com"),
    },
-    e2e: getEnv("IS_E2E", "false") === "true" ? true : false,
+    e2e: getEnv("IS_E2E", "false") === "true",
+    enableSlotsWorkers: getEnv("ENABLE_SLOTS_WORKERS", "true") === "true",
+    slotsWorkerPoolSize: Number(getEnv("SLOTS_WORKER_POOL_SIZE", "4")),
  };
 };

@@ -32,4 +32,6 @@ export type AppConfig = {
    baseUrl: string;
  };
  e2e: boolean;
+  enableSlotsWorkers: boolean;
+  slotsWorkerPoolSize: number;
 };
@@ -38,9 +38,13 @@ export type Environment = {
  DATABASE_WRITE_POOL_MAX: number;
  DATABASE_READ_WORKER_POOL_MAX: number;
  DATABASE_WRITE_WORKER_POOL_MAX: number;
+  ENABLE_SLOTS_WORKERS: string;
+  SLOTS_WORKER_POOL_SIZE: string;
 };

 export const getEnv = <K extends keyof Environment>(key: K, fallback?: Environment[K]): Environment[K] => {
+  // biome-ignore lint/style/noProcessEnv: <config file>
+  // biome-ignore lint/correctness/noProcessGlobal: <config file>
  const value = process.env[key] as Environment[K] | undefined;

  if (value === undefined) {
@@ -1,25 +1,32 @@
-import { TRPC_ERROR_CODE, TRPC_ERROR_MAP, TRPCErrorCode } from "@/filters/trpc-exception.filter";
-import { AvailableSlotsService } from "@/lib/services/available-slots.service";
-import { SlotsOutputService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-output.service";
-import type { RangeSlots, TimeSlots } from "@/modules/slots/slots-2024-04-15/services/slots-output.service";
-import { SlotsWorkerService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-worker.service";
-import { SlotsService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots.service";
-import { Query, Body, Controller, Get, Delete, Post, Req, Res, BadRequestException } from "@nestjs/common";
-import { ConfigService } from "@nestjs/config";
-import { ApiExcludeController as DocsExcludeController } from "@nestjs/swagger";
-import { ApiTags as DocsTags, ApiCreatedResponse, ApiOkResponse, ApiOperation } from "@nestjs/swagger";
-import { Response as ExpressResponse, Request as ExpressRequest } from "express";
-
 import {
  SUCCESS_STATUS,
-  VERSION_2024_06_14,
  VERSION_2024_04_15,
  VERSION_2024_06_11,
+  VERSION_2024_06_14,
  VERSION_2024_08_13,
 } from "@calcom/platform-constants";
 import { TRPCError } from "@calcom/platform-libraries";
-import { RemoveSelectedSlotInput_2024_04_15, ReserveSlotInput_2024_04_15 } from "@calcom/platform-types";
-import { ApiResponse, GetAvailableSlotsInput_2024_04_15 } from "@calcom/platform-types";
+import {
+  ApiResponse,
+  GetAvailableSlotsInput_2024_04_15,
+  RemoveSelectedSlotInput_2024_04_15,
+  ReserveSlotInput_2024_04_15,
+} from "@calcom/platform-types";
+import { BadRequestException, Body, Controller, Delete, Get, Post, Query, Req, Res } from "@nestjs/common";
+import { ConfigService } from "@nestjs/config";
+import {
+  ApiCreatedResponse,
+  ApiOkResponse,
+  ApiOperation,
+  ApiExcludeController as DocsExcludeController,
+} from "@nestjs/swagger";
+import { Request as ExpressRequest, Response as ExpressResponse } from "express";
+import { TRPC_ERROR_CODE, TRPC_ERROR_MAP, TRPCErrorCode } from "@/filters/trpc-exception.filter";
+import { AvailableSlotsService } from "@/lib/services/available-slots.service";
+import { SlotsService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots.service";
+import type { RangeSlots, TimeSlots } from "@/modules/slots/slots-2024-04-15/services/slots-output.service";
+import { SlotsOutputService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-output.service";
+import { SlotsWorkerService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-worker.service";

@Controller({
  path: "/v2/slots",
@@ -166,32 +173,29 @@ export class SlotsController_2024_04_15 {
  ): Promise<ApiResponse<{ slots: TimeSlots["slots"] | RangeSlots["slots"] }>> {
    try {
      const isTeamEvent =
-        query.isTeamEvent === undefined
-          ? await this.slotsService.checkIfIsTeamEvent(query.eventTypeId)
-          : query.isTeamEvent;
+        query.isTeamEvent ?? (await this.slotsService.checkIfIsTeamEvent(query.eventTypeId));

      // Do not use workers in E2E, not supported by TS-JEST
-      const availableSlotsService = this.config.get<boolean>("e2e") ? this.availableSlotsService : null;
+      // Or if explicitly disabled via specific env var
+      const shouldUseAvailableSlotsService =
+        this.config.get<boolean>("e2e") || !this.config.get<boolean>("enableSlotsWorkers");
+      const slotsArgs = {
+        input: {
+          ...query,
+          isTeamEvent,
+        },
+        ctx: {
+          req,
+        },
+      };

-      const availableSlots = availableSlotsService
-        ? await availableSlotsService.getAvailableSlots({
-            input: {
-              ...query,
-              isTeamEvent,
-            },
-            ctx: {
-              req,
-            },
-          })
-        : await this.slotsWorkerService.getAvailableSlotsInWorker({
-            input: {
-              ...query,
-              isTeamEvent,
-            },
-            ctx: {
-              req,
-            },
-          });
+      let availableSlots: TimeSlots;
+
+      if (shouldUseAvailableSlotsService) {
+        availableSlots = await this.availableSlotsService.getAvailableSlots(slotsArgs);
+      } else {
+        availableSlots = await this.slotsWorkerService.getAvailableSlotsInWorker(slotsArgs);
+      }

      const { slots } = await this.slotsOutputService.getOutputSlots(
        availableSlots,
@@ -1,10 +1,9 @@
-import { Injectable, Logger, OnModuleDestroy } from "@nestjs/common";
-import { ConfigService } from "@nestjs/config";
 import * as path from "node:path";
 import { Worker } from "node:worker_threads";

-// Import WorkerOptions type
 import type { GetScheduleOptions } from "@calcom/trpc/server/routers/viewer/slots/types";
+import { Injectable, Logger, OnModuleDestroy } from "@nestjs/common";
+import { ConfigService } from "@nestjs/config";

 import { TimeSlots } from "./slots-output.service";

@@ -43,11 +42,13 @@ export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy {
  private availableWorkers: Worker[] = [];

  constructor(private readonly config: ConfigService) {
-    this.maxWorkers = process.env.SLOTS_WORKER_POOL_SIZE
-      ? parseInt(process.env.SLOTS_WORKER_POOL_SIZE, 10)
-      : 4;
+    this.maxWorkers = this.config.get<number>("slotsWorkerPoolSize") ?? 4;
+    if (!this.config.get<boolean>("enableSlotsWorkers")) {
+      this.maxWorkers = 0;
+    }
+
    // Workers are not initialized in E@E
-    if (!this.config.get<boolean>("e2e")) {
+    if (!this.config.get<boolean>("e2e") && this.maxWorkers > 0) {
      this.initializeWorkerPool();
    }
  }
@@ -108,10 +109,12 @@ export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy {
          this.logger.error(`Error terminating failed worker ${failedWorker.threadId}: ${err?.message}`);
        });
    } catch (error) {
+      let errorMessage = String(error);
+      if (error instanceof Error) {
+        errorMessage = error.message;
+      }
      this.logger.error(
-        `Failed to invoke terminate method on failed worker ${failedWorker.threadId}: ${
-          error instanceof Error ? error.message : String(error)
-        }`
+        `Failed to invoke terminate method on failed worker ${failedWorker.threadId}: ${errorMessage}`
      );
    }

@@ -119,18 +122,45 @@ export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy {
    try {
      this.createNewWorker();
    } catch (error) {
-      this.logger.error(
-        `Failed to create replacement worker after failure: ${
-          error instanceof Error ? error.message : String(error)
-        }`,
-        error instanceof Error ? error.stack : undefined
-      );
+      let errorMessage = String(error);
+      let errorStack: string | undefined;
+
+      if (error instanceof Error) {
+        errorMessage = error.message;
+        errorStack = error.stack;
+      }
+
+      this.logger.error(`Failed to create replacement worker after failure: ${errorMessage}`, errorStack);
    }

    // After a worker fails, process the next task in case there are queued tasks
    this.processNextTask();
  }

+  /**
+   * Converts the request context into a serializable format for worker communication.
+   * @param ctx The context from GetScheduleOptions containing request data.
+   * @returns A serializable context object with cookies and headers, or undefined.
+   */
+  private getSerializableContext(ctx: GetScheduleOptions["ctx"]): WorkerMessage["ctx"] {
+    if (!ctx) {
+      return undefined;
+    }
+
+    let reqData: NonNullable<NonNullable<WorkerMessage["ctx"]>["req"]> | undefined;
+
+    if (ctx.req) {
+      reqData = {
+        cookies: (ctx.req.cookies as Record<string, string>) || {},
+        headers: (ctx.req.headers as Record<string, string>) || {},
+      };
+    }
+
+    return {
+      req: reqData,
+    };
+  }
+
  /**
   * Processes the next task in the queue if there are available workers.
   * Assigns a task to an available worker and sets up 'once' listeners for its specific result.
@@ -146,21 +176,12 @@ export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy {
      }

      // Prepare context for serialization
-      const serializableCtx: WorkerMessage["ctx"] = task.options.ctx
-        ? {
-            req: task.options.ctx.req
-              ? {
-                  cookies: (task.options.ctx.req.cookies as Record<string, string>) || {},
-                  headers: (task.options.ctx.req.headers as Record<string, string>) || {},
-                }
-              : undefined,
-          }
-        : undefined;
+      const serializableCtx = this.getSerializableContext(task.options.ctx);

      try {
        // Use 'once' listeners for task-specific responses and errors.
        // 'once' listeners automatically remove themselves after being invoked, preventing leaks.
-        const messageListener = (result: WorkerResult) => {
+        const messageListener = (result: WorkerResult): void => {
          this.availableWorkers.push(worker); // Return worker to the available pool
          if (result.success) {
            task.resolve(result.data as TimeSlots);
@@ -170,7 +191,7 @@ export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy {
          this.processNextTask(); // Attempt to process the next task
        };

-        const errorListener = (err: Error) => {
+        const errorListener = (err: Error): void => {
          this.availableWorkers.push(worker); // Ensure worker is returned
          task.reject(new Error(`Worker thread error during task execution: ${err.message}`));
          this.processNextTask(); // Attempt to process the next task
@@ -186,11 +207,13 @@ export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy {
      } catch (error) {
        // If posting the message itself fails (e.g., serialization error)
        this.availableWorkers.push(worker); // Ensure worker is returned to pool
-        task.reject(
-          new Error(
-            `Failed to dispatch task to worker: ${error instanceof Error ? error.message : String(error)}`
-          )
-        );
+
+        let errorMessage = String(error);
+        if (error instanceof Error) {
+          errorMessage = error.message;
+        }
+
+        task.reject(new Error(`Failed to dispatch task to worker: ${errorMessage}`));
        this.processNextTask(); // Try to process next task if available
      }
    }
@@ -1,3 +1,6 @@
+/** biome-ignore-all lint/correctness/noProcessGlobal: e2e file */
+/** biome-ignore-all lint/suspicious/noTsIgnore: e2e file */
+/** biome-ignore-all lint/style/noProcessEnv: e2e file */
 import type { Environment } from "@/env";
 import "dotenv/config";

@@ -22,6 +25,8 @@ const env: Partial<Omit<Environment, "NODE_ENV">> = {
  RATE_LIMIT_DEFAULT_LIMIT: 10000,
  RATE_LIMIT_DEFAULT_BLOCK_DURATION_MS: 60000,
  IS_TEAM_BILLING_ENABLED: false,
+  ENABLE_SLOTS_WORKERS: "false",
+  SLOTS_WORKER_POOL_SIZE: "0",
 };

 // eslint-disable-next-line @typescript-eslint/ban-ts-comment
@@ -36,5 +41,5 @@ process.env = {
  CALENDSO_ENCRYPTION_KEY: "22gfxhWUlcKliUeXcu8xNah2+HP/29ZX",
  INTEGRATION_TEST_MODE: "true",
  e2e: "true",
-  SLOTS_CACHE_TTL: "1"
+  SLOTS_CACHE_TTL: "1",
 };