diff --git a/apps/api/v2/src/config/app.ts b/apps/api/v2/src/config/app.ts index a19f3b94a1..a32a7d5854 100644 --- a/apps/api/v2/src/config/app.ts +++ b/apps/api/v2/src/config/app.ts @@ -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")), }; }; diff --git a/apps/api/v2/src/config/type.ts b/apps/api/v2/src/config/type.ts index 102ad23739..0b6cd7d0c0 100644 --- a/apps/api/v2/src/config/type.ts +++ b/apps/api/v2/src/config/type.ts @@ -32,4 +32,6 @@ export type AppConfig = { baseUrl: string; }; e2e: boolean; + enableSlotsWorkers: boolean; + slotsWorkerPoolSize: number; }; diff --git a/apps/api/v2/src/env.ts b/apps/api/v2/src/env.ts index be3cc6b143..c25a9cd480 100644 --- a/apps/api/v2/src/env.ts +++ b/apps/api/v2/src/env.ts @@ -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 = (key: K, fallback?: Environment[K]): Environment[K] => { + // biome-ignore lint/style/noProcessEnv: + // biome-ignore lint/correctness/noProcessGlobal: const value = process.env[key] as Environment[K] | undefined; if (value === undefined) { diff --git a/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts b/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts index 2c83a8dfe1..a33ddb4adc 100644 --- a/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts +++ b/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts @@ -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> { 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("e2e") ? this.availableSlotsService : null; + // Or if explicitly disabled via specific env var + const shouldUseAvailableSlotsService = + this.config.get("e2e") || !this.config.get("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, diff --git a/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts b/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts index b409c5fa90..dc9612b3ce 100644 --- a/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts +++ b/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts @@ -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("slotsWorkerPoolSize") ?? 4; + if (!this.config.get("enableSlotsWorkers")) { + this.maxWorkers = 0; + } + // Workers are not initialized in E@E - if (!this.config.get("e2e")) { + if (!this.config.get("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["req"]> | undefined; + + if (ctx.req) { + reqData = { + cookies: (ctx.req.cookies as Record) || {}, + headers: (ctx.req.headers as Record) || {}, + }; + } + + 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) || {}, - headers: (task.options.ctx.req.headers as Record) || {}, - } - : 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 } } diff --git a/apps/api/v2/test/setEnvVars.ts b/apps/api/v2/test/setEnvVars.ts index aac391b80c..84dd6648af 100644 --- a/apps/api/v2/test/setEnvVars.ts +++ b/apps/api/v2/test/setEnvVars.ts @@ -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> = { 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", };