diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda.driver.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda.driver.ts index f8ec03710c9..2151fc370cb 100644 --- a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda.driver.ts +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda.driver.ts @@ -1,34 +1,10 @@ -import { createHash } from 'crypto'; -import * as fs from 'fs/promises'; -import { join, resolve } from 'path'; - import { - CreateFunctionCommand, - type CreateFunctionCommandInput, - DeleteFunctionCommand, - DeleteLayerVersionCommand, - GetFunctionCommand, - GetFunctionCommandOutput, InvokeCommand, type InvokeCommandInput, - Lambda, - type LambdaClientConfig, - ListLayerVersionsCommand, - type ListLayerVersionsCommandInput, LogType, - PublishLayerVersionCommand, ResourceNotFoundException, - TagResourceCommand, - UpdateFunctionCodeCommand, - UpdateFunctionConfigurationCommand, - waitUntilFunctionActiveV2, - waitUntilFunctionUpdatedV2, } from '@aws-sdk/client-lambda'; -import { PutObjectCommand, S3Client } from '@aws-sdk/client-s3'; -import { AssumeRoleCommand, STSClient } from '@aws-sdk/client-sts'; -import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; import { Logger } from '@nestjs/common'; -import { isDefined } from 'twenty-shared/utils'; import { type LogicFunctionDriver, @@ -39,24 +15,19 @@ import { type LogicFunctionTranspileResult, } from 'src/engine/core-modules/logic-function/logic-function-drivers/interfaces/logic-function-driver.interface'; -import { isNonEmptyString } from '@sniptt/guards'; -import { ASSET_PATH } from 'src/constants/assets-path'; -import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; -import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; -import { COMMON_LAYER_DEPENDENCIES_DIRNAME } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/common-layer-dependencies-dirname'; -import { copyBuilder } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-builder'; -import { copyCommonLayerDependencies } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-common-layer-dependencies'; -import { copyExecutor } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-executor'; -import { copyYarnInstall } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-yarn-install'; -import { createZipFile } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/create-zip-file'; -import { TemporaryDirManager } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/temporary-dir-manager'; import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; -import { type SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service'; -import { LogicFunctionExecutionStatus } from 'src/engine/metadata-modules/logic-function/dtos/logic-function-execution-result.dto'; import { - LogicFunctionExecutionMode, - LogicFunctionRuntime, -} from 'src/engine/metadata-modules/logic-function/logic-function.entity'; + type LambdaDriverExecutorPayload, + type LambdaDriverOptions, + LambdaExecutionPhase, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type'; +import { LambdaAwsClientService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service'; +import { LambdaExecutorManagerService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-executor-manager.service'; +import { LambdaLayerManagerService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-layer-manager.service'; +import { LambdaToolFunctionsService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-tool-functions.service'; +import { parseLambdaLogResult } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/parse-lambda-log-result.util'; +import { LogicFunctionExecutionStatus } from 'src/engine/metadata-modules/logic-function/dtos/logic-function-execution-result.dto'; +import { LogicFunctionExecutionMode } from 'src/engine/metadata-modules/logic-function/logic-function.entity'; import { LogicFunctionException, LogicFunctionExceptionCode, @@ -64,1112 +35,74 @@ import { import { type FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type'; import { isLogicFunctionReadyForPrebuiltInstall } from 'src/engine/metadata-modules/logic-function/utils/is-logic-function-ready-for-prebuilt-install.util'; -enum LambdaExecutionPhase { - BUILD = 'build', - FETCH_CODE = 'fetch-code', - INVOKE = 'invoke', -} - -const UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS = 60; -const CREDENTIALS_DURATION_IN_SECONDS = 60 * 60; // 1h -const YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS = 300; -const YARN_INSTALL_LAMBDA_MEMORY_MB = 1024; -const COMMON_LAYER_NAME_PREFIX = 'twenty-common-layer'; -const BUILDER_LAMBDA_TIMEOUT_SECONDS = 60; -const BUILDER_LAMBDA_MEMORY_MB = 512; -const EXECUTOR_LAMBDA_MEMORY_MB = 512; -const LAMBDA_EPHEMERAL_STORAGE_MB = 4096; -const YARN_INSTALL_HANDLER_PATH = resolve( - __dirname, - join( - ASSET_PATH, - 'engine/core-modules/logic-function/logic-function-drivers/constants/yarn-install/index.mjs', - ), -); - -const BUILDER_HANDLER_PATH = resolve( - __dirname, - join( - ASSET_PATH, - 'engine/core-modules/logic-function/logic-function-drivers/constants/builder/index.mjs', - ), -); - -const LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG = 'twenty:bundle-checksum'; - -const PREBUILT_BUNDLE_FILE_NAME = 'prebuilt-logic-function.mjs'; - -const PREBUILT_INSTALL_LOCK_TTL_MS = 180_000; -const PREBUILT_INSTALL_LOCK_RETRY_MS = 1_000; -const PREBUILT_INSTALL_LOCK_MAX_RETRIES = 180; - -type LambdaDriverExecutorPayload = { - code?: string; - params: object; - env: Record; - handlerName: string; -}; - -export type YarnInstallLambdaPayload = { - action: 'createLayer'; - packageJson: string; - yarnLock: string; - presignedUploadUrl: string; -}; - -export type YarnInstallLambdaResult = { - success: boolean; -}; - -export type BuilderLambdaPayload = { - action: 'transpile'; - sourceCode: string; - sourceFileName: string; - builtFileName: string; -}; - -export type BuilderLambdaResult = { - builtCode: string; -}; - -export interface LambdaDriverOptions extends LambdaClientConfig { - logicFunctionResourceService: LogicFunctionResourceService; - sdkClientArchiveService: SdkClientArchiveService; - cacheLockService: CacheLockService; - region: string; - lambdaRole: string; - subhostingRole?: string; - layerBucket: string; - layerBucketRegion: string; -} +export { + type LambdaDriverOptions, + type BuilderLambdaPayload, + type BuilderLambdaResult, + type YarnInstallLambdaPayload, + type YarnInstallLambdaResult, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type'; export class LambdaDriver implements LogicFunctionDriver { private readonly logger = new Logger(LambdaDriver.name); - private lambdaClient: Lambda | undefined; - private assumeRoleCredentials: - | { accessKeyId: string; secretAccessKey: string; sessionToken: string } - | undefined; - private credentialsExpiry: Date | null = null; - private readonly options: LambdaDriverOptions; + + private readonly awsClient: LambdaAwsClientService; + private readonly toolFunctions: LambdaToolFunctionsService; + private readonly layerManager: LambdaLayerManagerService; + private readonly executorManager: LambdaExecutorManagerService; private readonly logicFunctionResourceService: LogicFunctionResourceService; - private readonly sdkClientArchiveService: SdkClientArchiveService; - private readonly cacheLockService: CacheLockService; constructor(options: LambdaDriverOptions) { - this.options = options; - this.lambdaClient = undefined; this.logicFunctionResourceService = options.logicFunctionResourceService; - this.sdkClientArchiveService = options.sdkClientArchiveService; - this.cacheLockService = options.cacheLockService; - } - - private areAssumeRoleCredentialsExpired(): boolean { - return ( - !isDefined(this.assumeRoleCredentials) || - (isDefined(this.credentialsExpiry) && - new Date() >= this.credentialsExpiry) + this.awsClient = new LambdaAwsClientService(options); + this.toolFunctions = new LambdaToolFunctionsService( + options, + this.awsClient, ); - } - - private async refreshAssumeRoleCredentials() { - const stsClient = new STSClient({ region: this.options.region }); - - const assumeRoleCommand = new AssumeRoleCommand({ - RoleArn: this.options.subhostingRole, - RoleSessionName: 'LambdaSession', - DurationSeconds: CREDENTIALS_DURATION_IN_SECONDS, - }); - - const { Credentials } = await stsClient.send(assumeRoleCommand); - - if ( - !isDefined(Credentials) || - !isDefined(Credentials.AccessKeyId) || - !isDefined(Credentials.SecretAccessKey) || - !isDefined(Credentials.SessionToken) - ) { - throw new Error('Failed to assume role'); - } - - this.assumeRoleCredentials = { - accessKeyId: Credentials.AccessKeyId, - secretAccessKey: Credentials.SecretAccessKey, - sessionToken: Credentials.SessionToken, - }; - - this.credentialsExpiry = new Date( - Date.now() + (CREDENTIALS_DURATION_IN_SECONDS - 60 * 5) * 1000, + this.layerManager = new LambdaLayerManagerService( + options, + this.awsClient, + this.toolFunctions, + options.logicFunctionResourceService, + options.sdkClientArchiveService, ); - - this.lambdaClient = undefined; - } - - private async getAssumeRoleCredentials() { - if (this.areAssumeRoleCredentialsExpired()) { - await this.refreshAssumeRoleCredentials(); - } - - return this.assumeRoleCredentials!; - } - - private async getLambdaClient() { - if ( - !isDefined(this.lambdaClient) || - (isDefined(this.options.subhostingRole) && - this.areAssumeRoleCredentialsExpired()) - ) { - this.lambdaClient = new Lambda({ - ...this.options, - ...(isDefined(this.options.subhostingRole) && { - credentials: await this.getAssumeRoleCredentials(), - }), - }); - } - - return this.lambdaClient; - } - - private async generatePresignedUploadUrl( - s3Key: string, - expiresIn: number = 300, - ): Promise { - const s3Client = new S3Client({ - region: this.options.layerBucketRegion, - credentials: isDefined(this.options.subhostingRole) - ? await this.getAssumeRoleCredentials() - : this.options.credentials, - }); - - const putCommand = new PutObjectCommand({ - Bucket: this.options.layerBucket, - Key: s3Key, - ContentType: 'application/zip', - }); - - return getSignedUrl(s3Client, putCommand, { expiresIn }); - } - - private async waitFunctionActive( - functionName: string, - maxWaitTime: number = UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS, - ) { - await waitUntilFunctionActiveV2( - { client: await this.getLambdaClient(), maxWaitTime }, - { FunctionName: functionName }, + this.executorManager = new LambdaExecutorManagerService( + options, + this.awsClient, + this.layerManager, + options.cacheLockService, + options.logicFunctionResourceService, ); } - private async waitFunctionUpdated( - functionName: string, - maxWaitTime: number = UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS, - ) { - await waitUntilFunctionUpdatedV2( - { client: await this.getLambdaClient(), maxWaitTime }, - { FunctionName: functionName }, - ); - } - - private getDepsLayerName(flatApplication: FlatApplication): string { - const checksum = flatApplication.yarnLockChecksum ?? 'default'; - - return `deps-${checksum}`; - } - - private getSdkLayerName({ - workspaceId, - applicationUniversalIdentifier, - }: { - workspaceId: string; - applicationUniversalIdentifier: string; - }): string { - return `sdk-${workspaceId}-${applicationUniversalIdentifier}`; - } - - private yarnInstallFunctionName: string | undefined; - private builderFunctionName: string | undefined; - private commonLayerName: string | undefined; - - private async getCommonLayerName(): Promise { - if (isDefined(this.commonLayerName)) { - return this.commonLayerName; - } - - const [packageJson, yarnLock] = await Promise.all([ - fs.readFile( - join(COMMON_LAYER_DEPENDENCIES_DIRNAME, 'package.json'), - 'utf-8', - ), - fs.readFile( - join(COMMON_LAYER_DEPENDENCIES_DIRNAME, 'yarn.lock'), - 'utf-8', - ), - ]); - - const checksum = createHash('sha256') - .update(packageJson) - .update(yarnLock) - .digest('hex') - .slice(0, 12); - - this.commonLayerName = `${COMMON_LAYER_NAME_PREFIX}-${checksum}`; - - return this.commonLayerName; - } - - private async getYarnInstallFunctionName(): Promise { - if (isDefined(this.yarnInstallFunctionName)) { - return this.yarnInstallFunctionName; - } - - const handlerContent = await fs.readFile( - YARN_INSTALL_HANDLER_PATH, - 'utf-8', - ); - const checksum = createHash('sha256') - .update(handlerContent) - .digest('hex') - .slice(0, 12); - - this.yarnInstallFunctionName = `twenty-yarn-install-${checksum}`; - - return this.yarnInstallFunctionName; - } - - private async getBuilderFunctionName(): Promise { - if (isDefined(this.builderFunctionName)) { - return this.builderFunctionName; - } - - const handlerContent = await fs.readFile(BUILDER_HANDLER_PATH, 'utf-8'); - const checksum = createHash('sha256') - .update(handlerContent) - .digest('hex') - .slice(0, 12); - - this.builderFunctionName = `twenty-builder-${checksum}`; - - return this.builderFunctionName; - } - - private async ensureCommonLayerExists(): Promise { - const commonLayerName = await this.getCommonLayerName(); - const existingArn = await this.getExistingLayerArn(commonLayerName); - - if (isDefined(existingArn)) { - return existingArn; - } - - const temporaryDirManager = new TemporaryDirManager(); - const { sourceTemporaryDir, lambdaZipPath } = - await temporaryDirManager.init(); - - try { - await copyCommonLayerDependencies(sourceTemporaryDir); - - await createZipFile(sourceTemporaryDir, lambdaZipPath); - - const lambdaClient = await this.getLambdaClient(); - - const result = await lambdaClient.send( - new PublishLayerVersionCommand({ - LayerName: commonLayerName, - Content: { ZipFile: await fs.readFile(lambdaZipPath) }, - CompatibleRuntimes: [ - LogicFunctionRuntime.NODE18, - LogicFunctionRuntime.NODE22, - ], - }), - ); - - if (!result.LayerVersionArn) { - throw new Error( - 'PublishLayerVersion did not return a LayerVersionArn for common layer', - ); - } - - return result.LayerVersionArn; - } finally { - await temporaryDirManager.clean(); - } - } - - private async ensureYarnInstallLambdaExists(): Promise { - const yarnInstallFunctionName = await this.getYarnInstallFunctionName(); - const lambdaClient = await this.getLambdaClient(); - - try { - await lambdaClient.send( - new GetFunctionCommand({ FunctionName: yarnInstallFunctionName }), - ); - - return; - } catch (error) { - if (!(error instanceof ResourceNotFoundException)) { - throw error; - } - } - - const commonLayerArn = await this.ensureCommonLayerExists(); - - const temporaryDirManager = new TemporaryDirManager(); - const { sourceTemporaryDir, lambdaZipPath } = - await temporaryDirManager.init(); - - try { - await copyYarnInstall(sourceTemporaryDir); - - await createZipFile(sourceTemporaryDir, lambdaZipPath); - - const params: CreateFunctionCommandInput = { - Code: { - ZipFile: await fs.readFile(lambdaZipPath), - }, - FunctionName: yarnInstallFunctionName, - Layers: [commonLayerArn], - Handler: 'index.handler', - Role: this.options.lambdaRole, - Runtime: LogicFunctionRuntime.NODE22, - Timeout: YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS, - MemorySize: YARN_INSTALL_LAMBDA_MEMORY_MB, - EphemeralStorage: { Size: LAMBDA_EPHEMERAL_STORAGE_MB }, - }; - - await lambdaClient.send(new CreateFunctionCommand(params)); - } finally { - await temporaryDirManager.clean(); - } - - await this.waitFunctionActive(yarnInstallFunctionName); - } - - private async invokeYarnInstallLambda( - payload: YarnInstallLambdaPayload, - ): Promise { - const lambdaClient = await this.getLambdaClient(); - - const yarnInstallFunctionName = await this.getYarnInstallFunctionName(); - - const result = await lambdaClient.send( - new InvokeCommand({ - FunctionName: yarnInstallFunctionName, - Payload: JSON.stringify(payload), - LogType: LogType.Tail, - }), - { - abortSignal: AbortSignal.timeout( - YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS * 1000, - ), - }, - ); - - if (result.FunctionError) { - const parsedResult = result.Payload - ? JSON.parse(result.Payload.transformToString()) - : {}; - - throw new LogicFunctionException( - `Yarn install Lambda failed: ${JSON.stringify(parsedResult)}`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_CREATE_FAILED, - ); - } - - const parsedResult: YarnInstallLambdaResult = result.Payload - ? JSON.parse(result.Payload.transformToString()) - : {}; - - if (!parsedResult.success) { - throw new Error('Yarn install Lambda did not report success'); - } - - return parsedResult; - } - - private async ensureBuilderLambdaExists(): Promise { - const builderFunctionName = await this.getBuilderFunctionName(); - const lambdaClient = await this.getLambdaClient(); - - try { - await lambdaClient.send( - new GetFunctionCommand({ FunctionName: builderFunctionName }), - ); - - return; - } catch (error) { - if (!(error instanceof ResourceNotFoundException)) { - throw error; - } - } - - const commonLayerArn = await this.ensureCommonLayerExists(); - - const temporaryDirManager = new TemporaryDirManager(); - const { sourceTemporaryDir, lambdaZipPath } = - await temporaryDirManager.init(); - - try { - await copyBuilder(sourceTemporaryDir); - - await createZipFile(sourceTemporaryDir, lambdaZipPath); - - const params: CreateFunctionCommandInput = { - Code: { - ZipFile: await fs.readFile(lambdaZipPath), - }, - FunctionName: builderFunctionName, - Layers: [commonLayerArn], - Handler: 'index.handler', - Role: this.options.lambdaRole, - Runtime: LogicFunctionRuntime.NODE22, - Timeout: BUILDER_LAMBDA_TIMEOUT_SECONDS, - MemorySize: BUILDER_LAMBDA_MEMORY_MB, - EphemeralStorage: { Size: LAMBDA_EPHEMERAL_STORAGE_MB }, - }; - - await lambdaClient.send(new CreateFunctionCommand(params)); - } finally { - await temporaryDirManager.clean(); - } - - await this.waitFunctionActive(builderFunctionName); - } - async transpile({ sourceCode, sourceFileName, builtFileName, }: LogicFunctionTranspileParams): Promise { - await this.ensureBuilderLambdaExists(); - - const lambdaClient = await this.getLambdaClient(); - const builderFunctionName = await this.getBuilderFunctionName(); - - const payload: BuilderLambdaPayload = { - action: 'transpile', + const { builtCode } = await this.toolFunctions.transpile({ sourceCode, sourceFileName, builtFileName, - }; - - const result = await lambdaClient.send( - new InvokeCommand({ - FunctionName: builderFunctionName, - Payload: JSON.stringify(payload), - LogType: LogType.Tail, - }), - { - abortSignal: AbortSignal.timeout(BUILDER_LAMBDA_TIMEOUT_SECONDS * 1000), - }, - ); - - if (result.FunctionError) { - const parsedResult = result.Payload - ? JSON.parse(result.Payload.transformToString()) - : {}; - - const userCompilationErrorRegex = /^Build failed with \d+ error/; - - const isUserCompilationError = - isNonEmptyString(parsedResult?.errorMessage) && - userCompilationErrorRegex.test(parsedResult.errorMessage); - - if (isUserCompilationError) { - throw new LogicFunctionException( - `Function code compilation failed: ${parsedResult.errorMessage}`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_COMPILATION_FAILED, - ); - } - - throw new LogicFunctionException( - `Builder Lambda failed: ${JSON.stringify(parsedResult)}`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_CREATE_FAILED, - ); - } - - const parsedResult: BuilderLambdaResult = result.Payload - ? JSON.parse(result.Payload.transformToString()) - : {}; - - if (!parsedResult.builtCode) { - throw new Error('Builder Lambda did not return builtCode'); - } - - return { builtCode: parsedResult.builtCode }; - } - - private async getExistingLayerArn( - layerName: string, - ): Promise { - const listLayerParams: ListLayerVersionsCommandInput = { - LayerName: layerName, - MaxItems: 1, - }; - - const listLayerResult = await ( - await this.getLambdaClient() - ).send(new ListLayerVersionsCommand(listLayerParams)); - - return listLayerResult.LayerVersions?.[0]?.LayerVersionArn; - } - - private async publishLayer({ - layerName, - zipBuffer, - }: { - layerName: string; - zipBuffer: Buffer; - }): Promise { - const result = await ( - await this.getLambdaClient() - ).send( - new PublishLayerVersionCommand({ - LayerName: layerName, - Content: { ZipFile: zipBuffer }, - CompatibleRuntimes: [ - LogicFunctionRuntime.NODE18, - LogicFunctionRuntime.NODE22, - ], - }), - ); - - if (!isDefined(result.LayerVersionArn)) { - throw new Error('New layer version ARN is undefined'); - } - - return result.LayerVersionArn; - } - - private async getDependencyContents( - flatApplication: FlatApplication, - applicationUniversalIdentifier: string, - ): Promise<{ packageJson: string; yarnLock: string }> { - const temporaryDirManager = new TemporaryDirManager(); - const { sourceTemporaryDir } = await temporaryDirManager.init(); - - try { - await this.logicFunctionResourceService.copyDependenciesInMemory({ - applicationUniversalIdentifier, - workspaceId: flatApplication.workspaceId, - inMemoryFolderPath: sourceTemporaryDir, - }); - - const [packageJson, yarnLock] = await Promise.all([ - fs.readFile(`${sourceTemporaryDir}/package.json`, 'utf-8'), - fs.readFile(`${sourceTemporaryDir}/yarn.lock`, 'utf-8'), - ]); - - return { packageJson, yarnLock }; - } finally { - await temporaryDirManager.clean(); - } - } - - private async createLayerIfNotExist({ - flatApplication, - applicationUniversalIdentifier, - }: { - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise { - const layerName = this.getDepsLayerName(flatApplication); - - const existingArn = await this.getExistingLayerArn(layerName); - - if (isDefined(existingArn)) { - return; - } - - const { packageJson, yarnLock } = await this.getDependencyContents( - flatApplication, - applicationUniversalIdentifier, - ); - - await this.ensureYarnInstallLambdaExists(); - - const s3Key = `lambda-layers/${layerName}.zip`; - - const presignedUploadUrl = await this.generatePresignedUploadUrl(s3Key); - - await this.invokeYarnInstallLambda({ - action: 'createLayer', - packageJson, - yarnLock, - presignedUploadUrl, }); - const bucket = this.options.layerBucket; - - const lambdaClient = await this.getLambdaClient(); - - const publishResult = await lambdaClient.send( - new PublishLayerVersionCommand({ - LayerName: layerName, - Content: { S3Bucket: bucket, S3Key: s3Key }, - CompatibleRuntimes: [ - LogicFunctionRuntime.NODE18, - LogicFunctionRuntime.NODE22, - ], - }), - ); - - if (!publishResult.LayerVersionArn) { - throw new Error( - `PublishLayerVersion did not return a LayerVersionArn for layer '${layerName}'`, - ); - } + return { builtCode }; } - private async getLayerArn({ - flatApplication, - applicationUniversalIdentifier, - }: { - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise { - const layerName = this.getDepsLayerName(flatApplication); - - const existingArn = await this.getExistingLayerArn(layerName); - - if (isDefined(existingArn)) { - return existingArn; - } - - await this.createLayerIfNotExist({ - flatApplication, - applicationUniversalIdentifier, - }); - - const newArn = await this.getExistingLayerArn(layerName); - - if (!isDefined(newArn)) { - throw new Error( - `Layer '${layerName}' was not created by the yarn install Lambda`, - ); - } - - return newArn; + async delete(flatLogicFunction: FlatLogicFunction): Promise { + await this.executorManager.delete(flatLogicFunction); } - private async ensureSdkLayer({ - flatApplication, - applicationUniversalIdentifier, - }: { - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise { - const layerName = this.getSdkLayerName({ - workspaceId: flatApplication.workspaceId, - applicationUniversalIdentifier, - }); - - if (!flatApplication.isSdkLayerStale) { - const existingArn = await this.getExistingLayerArn(layerName); - - if (isDefined(existingArn)) { - return existingArn; - } - } - - await this.deleteAllLayerVersions({ - lambdaClient: await this.getLambdaClient(), - layerName, - }); - - const sdkArchiveBuffer = - await this.sdkClientArchiveService.downloadArchiveBuffer({ - workspaceId: flatApplication.workspaceId, - applicationId: flatApplication.id, - applicationUniversalIdentifier, - }); - - const zipBuffer = await this.reprefixZipEntries({ - sourceBuffer: sdkArchiveBuffer, - prefix: 'nodejs/node_modules/twenty-client-sdk', - }); - - const arn = await this.publishLayer({ layerName, zipBuffer }); - - await this.sdkClientArchiveService.markSdkLayerFresh({ - applicationId: flatApplication.id, - workspaceId: flatApplication.workspaceId, - }); - - return arn; + async installPrebuiltBundle( + params: LogicFunctionInstallPrebuiltBundleParams, + ): Promise { + await this.executorManager.installPrebuiltBundle(params); } - // Re-wraps zip entries under a new prefix path without extracting to disk. - private async reprefixZipEntries({ - sourceBuffer, - prefix, - }: { - sourceBuffer: Buffer; - prefix: string; - }): Promise { - const { default: unzipper } = await import('unzipper'); - const archiver = (await import('archiver')).default; - - const directory = await unzipper.Open.buffer(sourceBuffer); - const archive = archiver('zip', { zlib: { level: 9 } }); - - const chunks: Buffer[] = []; - - archive.on('data', (chunk: Buffer) => chunks.push(chunk)); - - for (const entry of directory.files) { - if (entry.type === 'Directory') { - continue; - } - - archive.append(entry.stream(), { - name: `${prefix}/${entry.path}`, - }); - } - - await new Promise((resolve, reject) => { - archive.on('end', resolve); - archive.on('error', reject); - void archive.finalize(); - }); - - return Buffer.concat(chunks); - } - - private async getLambdaExecutor(flatLogicFunction: FlatLogicFunction) { - try { - const getFunctionCommand: GetFunctionCommand = new GetFunctionCommand({ - FunctionName: flatLogicFunction.id, - }); - - return await (await this.getLambdaClient()).send(getFunctionCommand); - } catch (error) { - if (!(error instanceof ResourceNotFoundException)) { - throw error; - } - } - } - - async delete(flatLogicFunction: FlatLogicFunction) { - const lambdaExecutor = await this.getLambdaExecutor(flatLogicFunction); - - if (isDefined(lambdaExecutor)) { - const deleteFunctionCommand = new DeleteFunctionCommand({ - FunctionName: flatLogicFunction.id, - }); - - await (await this.getLambdaClient()).send(deleteFunctionCommand); - } - } - - private async deleteAllLayerVersions({ - lambdaClient, - layerName, - }: { - lambdaClient: Lambda; - layerName: string; - }): Promise { - let marker: string | undefined; - - do { - const listResult = await lambdaClient.send( - new ListLayerVersionsCommand({ - LayerName: layerName, - MaxItems: 50, - Marker: marker, - }), - ); - - const versions = listResult.LayerVersions ?? []; - - await Promise.all( - versions.map((version) => - lambdaClient.send( - new DeleteLayerVersionCommand({ - LayerName: layerName, - VersionNumber: version.Version, - }), - ), - ), - ); - - marker = listResult.NextMarker; - } while (isDefined(marker)); - } - - private hasExpectedLayers({ - lambdaExecutor, - flatApplication, - applicationUniversalIdentifier, - }: { - lambdaExecutor: GetFunctionCommandOutput; - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): boolean { - const layers = lambdaExecutor.Configuration?.Layers; - - if (!isDefined(layers) || layers.length !== 2) { - return false; - } - - const depsLayerName = this.getDepsLayerName(flatApplication); - const sdkLayerName = this.getSdkLayerName({ - workspaceId: flatApplication.workspaceId, - applicationUniversalIdentifier, - }); - - return ( - layers.some((layer) => layer.Arn?.includes(depsLayerName)) && - layers.some((layer) => layer.Arn?.includes(sdkLayerName)) - ); - } - - private async updateLambdaExecutorConfiguration({ - flatLogicFunction, - depsLayerArn, - sdkLayerArn, - }: { - flatLogicFunction: FlatLogicFunction; - depsLayerArn: string; - sdkLayerArn: string; - }) { - const lambdaClient = await this.getLambdaClient(); - - await lambdaClient.send( - new UpdateFunctionConfigurationCommand({ - FunctionName: flatLogicFunction.id, - Layers: [depsLayerArn, sdkLayerArn], - Runtime: flatLogicFunction.runtime, - Timeout: 900, - MemorySize: EXECUTOR_LAMBDA_MEMORY_MB, - }), - ); - } - - private async buildLambdaExecutor({ - flatLogicFunction, - flatApplication, - applicationUniversalIdentifier, - }: { - flatLogicFunction: FlatLogicFunction; - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }) { - const buildArgs = { - flatLogicFunction, - flatApplication, - applicationUniversalIdentifier, - }; - - const { canSkip } = await this.checkLambdaExecutorBuildStatus(buildArgs); - - if (canSkip) { - return; - } - - const buildLockTtlMs = 120_000; - const buildLockRetryMs = 500; - const buildLockMaxRetries = 240; - - await this.cacheLockService.withLock( - async () => { - // Need to check again inside the lock in case lock was not acquired immediately. - const { canSkip, lambdaExecutor } = - await this.checkLambdaExecutorBuildStatus(buildArgs); - - if (canSkip) { - return; - } - - await this.ensureLambdaExecutor({ ...buildArgs, lambdaExecutor }); - }, - `lambda-build:${flatLogicFunction.id}`, - { - ttl: buildLockTtlMs, - ms: buildLockRetryMs, - maxRetries: buildLockMaxRetries, - }, - ); - } - - private async checkLambdaExecutorBuildStatus({ - flatLogicFunction, - flatApplication, - applicationUniversalIdentifier, - }: { - flatLogicFunction: FlatLogicFunction; - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise<{ - canSkip: boolean; - lambdaExecutor: GetFunctionCommandOutput | undefined; - }> { - const lambdaExecutor = await this.getLambdaExecutor(flatLogicFunction); - - const isActive = lambdaExecutor?.Configuration?.State === 'Active'; - - const canSkip = - isDefined(lambdaExecutor) && - isActive && - !flatApplication.isSdkLayerStale && - this.hasExpectedLayers({ - lambdaExecutor, - flatApplication, - applicationUniversalIdentifier, - }); - - return { canSkip, lambdaExecutor }; - } - - private async ensureLambdaExecutor({ - flatLogicFunction, - flatApplication, - applicationUniversalIdentifier, - lambdaExecutor, - }: { - flatLogicFunction: FlatLogicFunction; - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - lambdaExecutor: GetFunctionCommandOutput | undefined; - }) { - let depsLayerArn: string; - - try { - depsLayerArn = await this.getLayerArn({ - flatApplication, - applicationUniversalIdentifier, - }); - } catch (error) { - this.logger.error( - `Failed to get dependency layer for function ${flatLogicFunction.id}: ${error instanceof Error ? error.message : String(error)}`, - error instanceof Error ? error.stack : undefined, - ); - throw new LogicFunctionException( - `Failed to get dependency layer for function '${flatLogicFunction.id}': ${error instanceof Error ? error.message : 'Unknown error'}`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_LAYER_BUILD_FAILED, - ); - } - - let sdkLayerArn: string; - - try { - sdkLayerArn = await this.ensureSdkLayer({ - flatApplication, - applicationUniversalIdentifier, - }); - } catch (error) { - this.logger.error( - `Failed to get SDK layer for function ${flatLogicFunction.id}: ${error instanceof Error ? error.message : String(error)}`, - error instanceof Error ? error.stack : undefined, - ); - throw new LogicFunctionException( - `Failed to get SDK layer for function '${flatLogicFunction.id}': ${error instanceof Error ? error.message : 'Unknown error'}`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_LAYER_BUILD_FAILED, - ); - } - - if (!isDefined(lambdaExecutor)) { - await this.createLambdaExecutor({ - flatLogicFunction, - depsLayerArn, - sdkLayerArn, - }); - await this.waitFunctionActive(flatLogicFunction.id); - - return; - } - - await this.updateLambdaExecutorConfiguration({ - flatLogicFunction, - depsLayerArn, - sdkLayerArn, - }); - await this.waitFunctionUpdated(flatLogicFunction.id); - } - - private async createLambdaExecutor({ - flatLogicFunction, - depsLayerArn, - sdkLayerArn, - }: { - flatLogicFunction: FlatLogicFunction; - depsLayerArn: string; - sdkLayerArn: string; - }) { - const temporaryDirManager = new TemporaryDirManager(); - - const { sourceTemporaryDir, lambdaZipPath } = - await temporaryDirManager.init(); - - try { - await copyExecutor(sourceTemporaryDir); - - await createZipFile(sourceTemporaryDir, lambdaZipPath); - - // SDK layer listed last so it overwrites the stub twenty-client-sdk - // from the deps layer (later layers take precedence in /opt merge). - const params: CreateFunctionCommandInput = { - Code: { - ZipFile: await fs.readFile(lambdaZipPath), - }, - FunctionName: flatLogicFunction.id, - Layers: [depsLayerArn, sdkLayerArn], - Handler: 'index.handler', - Role: this.options.lambdaRole, - Runtime: flatLogicFunction.runtime, - Timeout: 900, - MemorySize: EXECUTOR_LAMBDA_MEMORY_MB, - EphemeralStorage: { Size: LAMBDA_EPHEMERAL_STORAGE_MB }, - }; - - const lambdaClient = await this.getLambdaClient(); - - await lambdaClient.send(new CreateFunctionCommand(params)); - } finally { - await temporaryDirManager.clean(); - } - } - - private parseLambdaLogResult(logResult: string | undefined): { - logs: string; - initDurationMs: string | null; - billedDurationMs: string | null; - reportDurationMs: string | null; - coldStart: boolean; - } { - if (!logResult) { - return { - logs: '', - initDurationMs: null, - billedDurationMs: null, - reportDurationMs: null, - coldStart: false, - }; - } - - const decoded = Buffer.from(logResult, 'base64').toString('utf8'); - - const initDurationMs = - decoded.match(/Init Duration:\s*([\d.]+)\s*ms/i)?.[1] ?? null; - const billedDurationMs = - decoded.match(/Billed Duration:\s*([\d.]+)\s*ms/i)?.[1] ?? null; - const reportDurationMs = - decoded.match(/\bDuration:\s*([\d.]+)\s*ms/i)?.[1] ?? null; - - const logs = decoded - .split('\t') - .join(' ') - .replace(/^(START|END|REPORT).*\n?/gm, '') - .replace( - /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z) [a-f0-9-]+ INFO /gm, - '$1 INFO ', - ) - .trim(); - - return { - logs, - initDurationMs, - billedDurationMs, - reportDurationMs, - coldStart: initDurationMs !== null, - }; + async getInstalledBundleChecksum( + flatLogicFunction: FlatLogicFunction, + ): Promise { + return this.executorManager.getInstalledBundleChecksum(flatLogicFunction); } async execute({ @@ -1200,7 +133,7 @@ export class LambdaDriver implements LogicFunctionDriver { try { const buildStart = Date.now(); - await this.buildLambdaExecutor({ + await this.executorManager.buildExecutor({ flatLogicFunction, flatApplication, applicationUniversalIdentifier, @@ -1238,7 +171,7 @@ export class LambdaDriver implements LogicFunctionDriver { }; const command = new InvokeCommand(params); - const lambdaClient = await this.getLambdaClient(); + const lambdaClient = await this.awsClient.getLambdaClient(); const invokeStart = Date.now(); const result = await lambdaClient.send(command, { @@ -1256,7 +189,7 @@ export class LambdaDriver implements LogicFunctionDriver { billedDurationMs, reportDurationMs, coldStart, - } = this.parseLambdaLogResult(result.LogResult); + } = parseLambdaLogResult(result.LogResult); const duration = Date.now() - invokeFlowStart; @@ -1296,9 +229,9 @@ export class LambdaDriver implements LogicFunctionDriver { } if (error instanceof Error && error.name === 'TimeoutError') { - const executor = await this.getLambdaExecutor(flatLogicFunction).catch( - () => undefined, - ); + const executor = await this.executorManager + .getLambdaExecutor(flatLogicFunction) + .catch(() => undefined); const functionState = executor?.Configuration?.State ?? 'unknown'; throw new LogicFunctionException( @@ -1317,109 +250,4 @@ export class LambdaDriver implements LogicFunctionDriver { ); } } - - private getPrebuiltInstallLockKey(flatLogicFunction: FlatLogicFunction) { - return `lambda-install:${flatLogicFunction.id}`; - } - - async installPrebuiltBundle({ - flatLogicFunction, - flatApplication, - applicationUniversalIdentifier, - }: LogicFunctionInstallPrebuiltBundleParams): Promise { - if (!isNonEmptyString(flatLogicFunction.checksum)) { - throw new LogicFunctionException( - `Cannot install prebuilt bundle for function '${flatLogicFunction.id}' without a checksum`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED, - ); - } - - const checksum = flatLogicFunction.checksum; - - await this.buildLambdaExecutor({ - flatLogicFunction, - flatApplication, - applicationUniversalIdentifier, - }); - - await this.cacheLockService.withLock( - async () => { - const compiledCode = - await this.logicFunctionResourceService.getBuiltCode({ - workspaceId: flatLogicFunction.workspaceId, - applicationUniversalIdentifier, - builtHandlerPath: flatLogicFunction.builtHandlerPath, - }); - - const temporaryDirManager = new TemporaryDirManager(); - const { sourceTemporaryDir, lambdaZipPath } = - await temporaryDirManager.init(); - - try { - await copyExecutor(sourceTemporaryDir); - await fs.writeFile( - join(sourceTemporaryDir, PREBUILT_BUNDLE_FILE_NAME), - compiledCode, - 'utf8', - ); - await createZipFile(sourceTemporaryDir, lambdaZipPath); - - const lambdaClient = await this.getLambdaClient(); - - const updateResult = await lambdaClient.send( - new UpdateFunctionCodeCommand({ - FunctionName: flatLogicFunction.id, - ZipFile: await fs.readFile(lambdaZipPath), - }), - ); - - await this.waitFunctionUpdated(flatLogicFunction.id); - - const functionArn = updateResult.FunctionArn; - - if (!isNonEmptyString(functionArn)) { - throw new LogicFunctionException( - `UpdateFunctionCode did not return a FunctionArn for '${flatLogicFunction.id}'`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED, - ); - } - - await lambdaClient.send( - new TagResourceCommand({ - Resource: functionArn, - Tags: { - [LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG]: checksum, - }, - }), - ); - } catch (error) { - this.logger.error( - `Failed to install prebuilt bundle for function ${flatLogicFunction.id}: ${error instanceof Error ? error.message : String(error)}`, - error instanceof Error ? error.stack : undefined, - ); - throw error; - } finally { - await temporaryDirManager.clean(); - } - }, - this.getPrebuiltInstallLockKey(flatLogicFunction), - { - ttl: PREBUILT_INSTALL_LOCK_TTL_MS, - ms: PREBUILT_INSTALL_LOCK_RETRY_MS, - maxRetries: PREBUILT_INSTALL_LOCK_MAX_RETRIES, - }, - ); - } - - async getInstalledBundleChecksum( - flatLogicFunction: FlatLogicFunction, - ): Promise { - const lambdaExecutor = await this.getLambdaExecutor(flatLogicFunction); - - if (!isDefined(lambdaExecutor)) { - return null; - } - - return lambdaExecutor.Tags?.[LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG] ?? null; - } } diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant.ts new file mode 100644 index 00000000000..e479414f00e --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant.ts @@ -0,0 +1,43 @@ +import { join, resolve } from 'path'; + +import { ASSET_PATH } from 'src/constants/assets-path'; + +export const UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS = 60; +export const CREDENTIALS_DURATION_IN_SECONDS = 60 * 60; // 1h + +export const YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS = 300; +export const YARN_INSTALL_LAMBDA_MEMORY_MB = 1024; +export const BUILDER_LAMBDA_TIMEOUT_SECONDS = 60; +export const BUILDER_LAMBDA_MEMORY_MB = 512; +export const EXECUTOR_LAMBDA_MEMORY_MB = 512; +export const EXECUTOR_LAMBDA_TIMEOUT_SECONDS = 900; +export const LAMBDA_EPHEMERAL_STORAGE_MB = 4096; + +export const COMMON_LAYER_NAME_PREFIX = 'twenty-common-layer'; +export const YARN_INSTALL_FUNCTION_NAME_PREFIX = 'twenty-yarn-install'; +export const BUILDER_FUNCTION_NAME_PREFIX = 'twenty-builder'; + +export const SDK_LAYER_PREFIX_IN_ZIP = 'nodejs/node_modules/twenty-client-sdk'; + +export const LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG = 'twenty:bundle-checksum'; +export const PREBUILT_BUNDLE_FILE_NAME = 'prebuilt-logic-function.mjs'; + +export const PREBUILT_INSTALL_LOCK_TTL_MS = 180_000; +export const PREBUILT_INSTALL_LOCK_RETRY_MS = 1_000; +export const PREBUILT_INSTALL_LOCK_MAX_RETRIES = 180; + +export const YARN_INSTALL_HANDLER_PATH = resolve( + __dirname, + join( + ASSET_PATH, + 'engine/core-modules/logic-function/logic-function-drivers/constants/yarn-install/index.mjs', + ), +); + +export const BUILDER_HANDLER_PATH = resolve( + __dirname, + join( + ASSET_PATH, + 'engine/core-modules/logic-function/logic-function-drivers/constants/builder/index.mjs', + ), +); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service.ts new file mode 100644 index 00000000000..009b6174ad1 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service.ts @@ -0,0 +1,145 @@ +import { + Lambda, + ListLayerVersionsCommand, + waitUntilFunctionActiveV2, + waitUntilFunctionUpdatedV2, +} from '@aws-sdk/client-lambda'; +import { PutObjectCommand, S3Client } from '@aws-sdk/client-s3'; +import { AssumeRoleCommand, STSClient } from '@aws-sdk/client-sts'; +import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; +import { isDefined } from 'twenty-shared/utils'; + +import { + CREDENTIALS_DURATION_IN_SECONDS, + UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant'; +import { type LambdaDriverOptions } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type'; + +export class LambdaAwsClientService { + private lambdaClient: Lambda | undefined; + private assumeRoleCredentials: + | { accessKeyId: string; secretAccessKey: string; sessionToken: string } + | undefined; + private credentialsExpiry: Date | null = null; + + constructor(private readonly options: LambdaDriverOptions) {} + + async getLambdaClient() { + if ( + !isDefined(this.lambdaClient) || + (isDefined(this.options.subhostingRole) && + this.areAssumeRoleCredentialsExpired()) + ) { + this.lambdaClient = new Lambda({ + ...this.options, + ...(isDefined(this.options.subhostingRole) && { + credentials: await this.getAssumeRoleCredentials(), + }), + }); + } + + return this.lambdaClient; + } + + async generatePresignedUploadUrl( + s3Key: string, + expiresIn: number = 300, + ): Promise { + const s3Client = new S3Client({ + region: this.options.layerBucketRegion, + credentials: isDefined(this.options.subhostingRole) + ? await this.getAssumeRoleCredentials() + : this.options.credentials, + }); + + const putCommand = new PutObjectCommand({ + Bucket: this.options.layerBucket, + Key: s3Key, + ContentType: 'application/zip', + }); + + return getSignedUrl(s3Client, putCommand, { expiresIn }); + } + + async waitFunctionActive( + functionName: string, + maxWaitTime: number = UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS, + ): Promise { + await waitUntilFunctionActiveV2( + { client: await this.getLambdaClient(), maxWaitTime }, + { FunctionName: functionName }, + ); + } + + async waitFunctionUpdated( + functionName: string, + maxWaitTime: number = UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS, + ): Promise { + await waitUntilFunctionUpdatedV2( + { client: await this.getLambdaClient(), maxWaitTime }, + { FunctionName: functionName }, + ); + } + + async getExistingLayerArn(layerName: string): Promise { + const lambdaClient = await this.getLambdaClient(); + + const listLayerResult = await lambdaClient.send( + new ListLayerVersionsCommand({ + LayerName: layerName, + MaxItems: 1, + }), + ); + + return listLayerResult.LayerVersions?.[0]?.LayerVersionArn; + } + + private areAssumeRoleCredentialsExpired(): boolean { + return ( + !isDefined(this.assumeRoleCredentials) || + (isDefined(this.credentialsExpiry) && + new Date() >= this.credentialsExpiry) + ); + } + + private async refreshAssumeRoleCredentials() { + const stsClient = new STSClient({ region: this.options.region }); + + const assumeRoleCommand = new AssumeRoleCommand({ + RoleArn: this.options.subhostingRole, + RoleSessionName: 'LambdaSession', + DurationSeconds: CREDENTIALS_DURATION_IN_SECONDS, + }); + + const { Credentials } = await stsClient.send(assumeRoleCommand); + + if ( + !isDefined(Credentials) || + !isDefined(Credentials.AccessKeyId) || + !isDefined(Credentials.SecretAccessKey) || + !isDefined(Credentials.SessionToken) + ) { + throw new Error('Failed to assume role'); + } + + this.assumeRoleCredentials = { + accessKeyId: Credentials.AccessKeyId, + secretAccessKey: Credentials.SecretAccessKey, + sessionToken: Credentials.SessionToken, + }; + + this.credentialsExpiry = new Date( + Date.now() + (CREDENTIALS_DURATION_IN_SECONDS - 60 * 5) * 1000, + ); + + this.lambdaClient = undefined; + } + + private async getAssumeRoleCredentials() { + if (this.areAssumeRoleCredentialsExpired()) { + await this.refreshAssumeRoleCredentials(); + } + + return this.assumeRoleCredentials!; + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-executor-manager.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-executor-manager.service.ts new file mode 100644 index 00000000000..834d77792d5 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-executor-manager.service.ts @@ -0,0 +1,369 @@ +import * as fs from 'fs/promises'; +import { join } from 'path'; + +import { + CreateFunctionCommand, + type CreateFunctionCommandInput, + DeleteFunctionCommand, + GetFunctionCommand, + type GetFunctionCommandOutput, + ResourceNotFoundException, + TagResourceCommand, + UpdateFunctionCodeCommand, + UpdateFunctionConfigurationCommand, +} from '@aws-sdk/client-lambda'; +import { Logger } from '@nestjs/common'; +import { isNonEmptyString } from '@sniptt/guards'; +import { isDefined } from 'twenty-shared/utils'; + +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; +import { + EXECUTOR_LAMBDA_MEMORY_MB, + EXECUTOR_LAMBDA_TIMEOUT_SECONDS, + LAMBDA_EPHEMERAL_STORAGE_MB, + LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG, + PREBUILT_BUNDLE_FILE_NAME, + PREBUILT_INSTALL_LOCK_MAX_RETRIES, + PREBUILT_INSTALL_LOCK_RETRY_MS, + PREBUILT_INSTALL_LOCK_TTL_MS, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant'; +import { type LambdaDriverOptions } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type'; +import { type LambdaAwsClientService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service'; +import { type LambdaLayerManagerService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-layer-manager.service'; +import { copyExecutor } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-executor'; +import { createZipFile } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/create-zip-file'; +import { TemporaryDirManager } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/temporary-dir-manager'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; +import { + LogicFunctionException, + LogicFunctionExceptionCode, +} from 'src/engine/metadata-modules/logic-function/logic-function.exception'; +import { type FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type'; + +type ExecutorBuildContext = { + flatLogicFunction: FlatLogicFunction; + flatApplication: FlatApplication; + applicationUniversalIdentifier: string; +}; + +export class LambdaExecutorManagerService { + private readonly logger = new Logger(LambdaExecutorManagerService.name); + + constructor( + private readonly options: Pick, + private readonly awsClient: LambdaAwsClientService, + private readonly layerManager: LambdaLayerManagerService, + private readonly cacheLockService: CacheLockService, + private readonly logicFunctionResourceService: LogicFunctionResourceService, + ) {} + + async getLambdaExecutor( + flatLogicFunction: FlatLogicFunction, + ): Promise { + const lambdaClient = await this.awsClient.getLambdaClient(); + + try { + return await lambdaClient.send( + new GetFunctionCommand({ FunctionName: flatLogicFunction.id }), + ); + } catch (error) { + if (error instanceof ResourceNotFoundException) { + return undefined; + } + + throw error; + } + } + + async delete(flatLogicFunction: FlatLogicFunction): Promise { + const lambdaExecutor = await this.getLambdaExecutor(flatLogicFunction); + + if (!isDefined(lambdaExecutor)) { + return; + } + + const lambdaClient = await this.awsClient.getLambdaClient(); + + await lambdaClient.send( + new DeleteFunctionCommand({ FunctionName: flatLogicFunction.id }), + ); + } + + async buildExecutor(context: ExecutorBuildContext): Promise { + const { canSkip } = await this.checkBuildStatus(context); + + if (canSkip) { + return; + } + + const buildLockTtlMs = 120_000; + const buildLockRetryMs = 500; + const buildLockMaxRetries = 240; + + await this.cacheLockService.withLock( + async () => { + // Need to check again inside the lock in case lock was not acquired immediately. + const { canSkip: canSkipAfterLock, lambdaExecutor } = + await this.checkBuildStatus(context); + + if (canSkipAfterLock) { + return; + } + + await this.ensureExecutor({ ...context, lambdaExecutor }); + }, + `lambda-build:${context.flatLogicFunction.id}`, + { + ttl: buildLockTtlMs, + ms: buildLockRetryMs, + maxRetries: buildLockMaxRetries, + }, + ); + } + + async installPrebuiltBundle(context: ExecutorBuildContext): Promise { + const { flatLogicFunction, applicationUniversalIdentifier } = context; + + if (!isNonEmptyString(flatLogicFunction.checksum)) { + throw new LogicFunctionException( + `Cannot install prebuilt bundle for function '${flatLogicFunction.id}' without a checksum`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED, + ); + } + + const checksum = flatLogicFunction.checksum; + + await this.buildExecutor(context); + + await this.cacheLockService.withLock( + async () => { + const compiledCode = + await this.logicFunctionResourceService.getBuiltCode({ + workspaceId: flatLogicFunction.workspaceId, + applicationUniversalIdentifier, + builtHandlerPath: flatLogicFunction.builtHandlerPath, + }); + + const temporaryDirManager = new TemporaryDirManager(); + const { sourceTemporaryDir, lambdaZipPath } = + await temporaryDirManager.init(); + + try { + await copyExecutor(sourceTemporaryDir); + await fs.writeFile( + join(sourceTemporaryDir, PREBUILT_BUNDLE_FILE_NAME), + compiledCode, + 'utf8', + ); + await createZipFile(sourceTemporaryDir, lambdaZipPath); + + const lambdaClient = await this.awsClient.getLambdaClient(); + + const updateResult = await lambdaClient.send( + new UpdateFunctionCodeCommand({ + FunctionName: flatLogicFunction.id, + ZipFile: await fs.readFile(lambdaZipPath), + }), + ); + + await this.awsClient.waitFunctionUpdated(flatLogicFunction.id); + + const functionArn = updateResult.FunctionArn; + + if (!isNonEmptyString(functionArn)) { + throw new LogicFunctionException( + `UpdateFunctionCode did not return a FunctionArn for '${flatLogicFunction.id}'`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED, + ); + } + + await lambdaClient.send( + new TagResourceCommand({ + Resource: functionArn, + Tags: { + [LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG]: checksum, + }, + }), + ); + } catch (error) { + this.logger.error( + `Failed to install prebuilt bundle for function ${flatLogicFunction.id}: ${error instanceof Error ? error.message : String(error)}`, + error instanceof Error ? error.stack : undefined, + ); + throw error; + } finally { + await temporaryDirManager.clean(); + } + }, + `lambda-install:${flatLogicFunction.id}`, + { + ttl: PREBUILT_INSTALL_LOCK_TTL_MS, + ms: PREBUILT_INSTALL_LOCK_RETRY_MS, + maxRetries: PREBUILT_INSTALL_LOCK_MAX_RETRIES, + }, + ); + } + + async getInstalledBundleChecksum( + flatLogicFunction: FlatLogicFunction, + ): Promise { + const lambdaExecutor = await this.getLambdaExecutor(flatLogicFunction); + + if (!isDefined(lambdaExecutor)) { + return null; + } + + return lambdaExecutor.Tags?.[LAMBDA_PREBUILT_BUNDLE_CHECKSUM_TAG] ?? null; + } + + private async checkBuildStatus(context: ExecutorBuildContext): Promise<{ + canSkip: boolean; + lambdaExecutor: GetFunctionCommandOutput | undefined; + }> { + const { flatApplication, applicationUniversalIdentifier } = context; + const lambdaExecutor = await this.getLambdaExecutor( + context.flatLogicFunction, + ); + + const isActive = lambdaExecutor?.Configuration?.State === 'Active'; + + const canSkip = + isDefined(lambdaExecutor) && + isActive && + !flatApplication.isSdkLayerStale && + this.layerManager.hasExpectedLayers({ + lambdaExecutor, + flatApplication, + applicationUniversalIdentifier, + }); + + return { canSkip, lambdaExecutor }; + } + + private async ensureExecutor({ + flatLogicFunction, + flatApplication, + applicationUniversalIdentifier, + lambdaExecutor, + }: ExecutorBuildContext & { + lambdaExecutor: GetFunctionCommandOutput | undefined; + }): Promise { + let depsLayerArn: string; + + try { + depsLayerArn = await this.layerManager.ensureDepsLayer({ + flatApplication, + applicationUniversalIdentifier, + }); + } catch (error) { + this.logger.error( + `Failed to get dependency layer for function ${flatLogicFunction.id}: ${error instanceof Error ? error.message : String(error)}`, + error instanceof Error ? error.stack : undefined, + ); + throw new LogicFunctionException( + `Failed to get dependency layer for function '${flatLogicFunction.id}': ${error instanceof Error ? error.message : 'Unknown error'}`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_LAYER_BUILD_FAILED, + ); + } + + let sdkLayerArn: string; + + try { + sdkLayerArn = await this.layerManager.ensureSdkLayer({ + flatApplication, + applicationUniversalIdentifier, + }); + } catch (error) { + this.logger.error( + `Failed to get SDK layer for function ${flatLogicFunction.id}: ${error instanceof Error ? error.message : String(error)}`, + error instanceof Error ? error.stack : undefined, + ); + throw new LogicFunctionException( + `Failed to get SDK layer for function '${flatLogicFunction.id}': ${error instanceof Error ? error.message : 'Unknown error'}`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_LAYER_BUILD_FAILED, + ); + } + + if (!isDefined(lambdaExecutor)) { + await this.createExecutor({ + flatLogicFunction, + depsLayerArn, + sdkLayerArn, + }); + await this.awsClient.waitFunctionActive(flatLogicFunction.id); + + return; + } + + await this.updateExecutorConfiguration({ + flatLogicFunction, + depsLayerArn, + sdkLayerArn, + }); + await this.awsClient.waitFunctionUpdated(flatLogicFunction.id); + } + + private async updateExecutorConfiguration({ + flatLogicFunction, + depsLayerArn, + sdkLayerArn, + }: { + flatLogicFunction: FlatLogicFunction; + depsLayerArn: string; + sdkLayerArn: string; + }): Promise { + const lambdaClient = await this.awsClient.getLambdaClient(); + + await lambdaClient.send( + new UpdateFunctionConfigurationCommand({ + FunctionName: flatLogicFunction.id, + Layers: [depsLayerArn, sdkLayerArn], + Runtime: flatLogicFunction.runtime, + Timeout: EXECUTOR_LAMBDA_TIMEOUT_SECONDS, + MemorySize: EXECUTOR_LAMBDA_MEMORY_MB, + }), + ); + } + + private async createExecutor({ + flatLogicFunction, + depsLayerArn, + sdkLayerArn, + }: { + flatLogicFunction: FlatLogicFunction; + depsLayerArn: string; + sdkLayerArn: string; + }): Promise { + const temporaryDirManager = new TemporaryDirManager(); + const { sourceTemporaryDir, lambdaZipPath } = + await temporaryDirManager.init(); + + try { + await copyExecutor(sourceTemporaryDir); + await createZipFile(sourceTemporaryDir, lambdaZipPath); + + // SDK layer listed last so it overwrites the stub twenty-client-sdk + // from the deps layer (later layers take precedence in /opt merge). + const params: CreateFunctionCommandInput = { + Code: { + ZipFile: await fs.readFile(lambdaZipPath), + }, + FunctionName: flatLogicFunction.id, + Layers: [depsLayerArn, sdkLayerArn], + Handler: 'index.handler', + Role: this.options.lambdaRole, + Runtime: flatLogicFunction.runtime, + Timeout: EXECUTOR_LAMBDA_TIMEOUT_SECONDS, + MemorySize: EXECUTOR_LAMBDA_MEMORY_MB, + EphemeralStorage: { Size: LAMBDA_EPHEMERAL_STORAGE_MB }, + }; + + const lambdaClient = await this.awsClient.getLambdaClient(); + + await lambdaClient.send(new CreateFunctionCommand(params)); + } finally { + await temporaryDirManager.clean(); + } + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-layer-manager.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-layer-manager.service.ts new file mode 100644 index 00000000000..b346aeea7ce --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-layer-manager.service.ts @@ -0,0 +1,253 @@ +import * as fs from 'fs/promises'; + +import { + DeleteLayerVersionCommand, + type GetFunctionCommandOutput, + ListLayerVersionsCommand, + PublishLayerVersionCommand, +} from '@aws-sdk/client-lambda'; +import { isDefined } from 'twenty-shared/utils'; + +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { SDK_LAYER_PREFIX_IN_ZIP } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant'; +import { type LambdaDriverOptions } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type'; +import { type LambdaAwsClientService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service'; +import { type LambdaToolFunctionsService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-tool-functions.service'; +import { getLambdaDepsLayerName } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-deps-layer-name.util'; +import { getLambdaSdkLayerName } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-sdk-layer-name.util'; +import { reprefixLambdaZipEntries } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/reprefix-lambda-zip-entries.util'; +import { TemporaryDirManager } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/temporary-dir-manager'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; +import { type SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service'; +import { LogicFunctionRuntime } from 'src/engine/metadata-modules/logic-function/logic-function.entity'; + +type LayerAppContext = { + flatApplication: FlatApplication; + applicationUniversalIdentifier: string; +}; + +export class LambdaLayerManagerService { + constructor( + private readonly options: Pick, + private readonly awsClient: LambdaAwsClientService, + private readonly toolFunctions: LambdaToolFunctionsService, + private readonly logicFunctionResourceService: LogicFunctionResourceService, + private readonly sdkClientArchiveService: SdkClientArchiveService, + ) {} + + async ensureDepsLayer(context: LayerAppContext): Promise { + const layerName = getLambdaDepsLayerName(context.flatApplication); + + const existingArn = await this.awsClient.getExistingLayerArn(layerName); + + if (isDefined(existingArn)) { + return existingArn; + } + + await this.createDepsLayer({ ...context, layerName }); + + const newArn = await this.awsClient.getExistingLayerArn(layerName); + + if (!isDefined(newArn)) { + throw new Error( + `Layer '${layerName}' was not created by the yarn install Lambda`, + ); + } + + return newArn; + } + + async ensureSdkLayer(context: LayerAppContext): Promise { + const { flatApplication, applicationUniversalIdentifier } = context; + const layerName = getLambdaSdkLayerName({ + workspaceId: flatApplication.workspaceId, + applicationUniversalIdentifier, + }); + + if (!flatApplication.isSdkLayerStale) { + const existingArn = await this.awsClient.getExistingLayerArn(layerName); + + if (isDefined(existingArn)) { + return existingArn; + } + } + + await this.deleteAllLayerVersions(layerName); + + const sdkArchiveBuffer = + await this.sdkClientArchiveService.downloadArchiveBuffer({ + workspaceId: flatApplication.workspaceId, + applicationId: flatApplication.id, + applicationUniversalIdentifier, + }); + + const zipBuffer = await reprefixLambdaZipEntries({ + sourceBuffer: sdkArchiveBuffer, + prefix: SDK_LAYER_PREFIX_IN_ZIP, + }); + + const arn = await this.publishLayer({ layerName, zipBuffer }); + + await this.sdkClientArchiveService.markSdkLayerFresh({ + applicationId: flatApplication.id, + workspaceId: flatApplication.workspaceId, + }); + + return arn; + } + + hasExpectedLayers({ + lambdaExecutor, + flatApplication, + applicationUniversalIdentifier, + }: LayerAppContext & { + lambdaExecutor: GetFunctionCommandOutput; + }): boolean { + const layers = lambdaExecutor.Configuration?.Layers; + + if (!isDefined(layers) || layers.length !== 2) { + return false; + } + + const depsLayerName = getLambdaDepsLayerName(flatApplication); + const sdkLayerName = getLambdaSdkLayerName({ + workspaceId: flatApplication.workspaceId, + applicationUniversalIdentifier, + }); + + return ( + layers.some((layer) => layer.Arn?.includes(depsLayerName)) && + layers.some((layer) => layer.Arn?.includes(sdkLayerName)) + ); + } + + private async createDepsLayer({ + flatApplication, + applicationUniversalIdentifier, + layerName, + }: LayerAppContext & { layerName: string }): Promise { + const existingArn = await this.awsClient.getExistingLayerArn(layerName); + + if (isDefined(existingArn)) { + return; + } + + const { packageJson, yarnLock } = await this.getDependencyContents({ + flatApplication, + applicationUniversalIdentifier, + }); + + const s3Key = `lambda-layers/${layerName}.zip`; + const presignedUploadUrl = + await this.awsClient.generatePresignedUploadUrl(s3Key); + + await this.toolFunctions.runYarnInstallCreateLayer({ + packageJson, + yarnLock, + presignedUploadUrl, + }); + + const lambdaClient = await this.awsClient.getLambdaClient(); + const publishResult = await lambdaClient.send( + new PublishLayerVersionCommand({ + LayerName: layerName, + Content: { + S3Bucket: this.options.layerBucket, + S3Key: s3Key, + }, + CompatibleRuntimes: [ + LogicFunctionRuntime.NODE18, + LogicFunctionRuntime.NODE22, + ], + }), + ); + + if (!publishResult.LayerVersionArn) { + throw new Error( + `PublishLayerVersion did not return a LayerVersionArn for layer '${layerName}'`, + ); + } + } + + private async getDependencyContents({ + flatApplication, + applicationUniversalIdentifier, + }: LayerAppContext): Promise<{ packageJson: string; yarnLock: string }> { + const temporaryDirManager = new TemporaryDirManager(); + const { sourceTemporaryDir } = await temporaryDirManager.init(); + + try { + await this.logicFunctionResourceService.copyDependenciesInMemory({ + applicationUniversalIdentifier, + workspaceId: flatApplication.workspaceId, + inMemoryFolderPath: sourceTemporaryDir, + }); + + const [packageJson, yarnLock] = await Promise.all([ + fs.readFile(`${sourceTemporaryDir}/package.json`, 'utf-8'), + fs.readFile(`${sourceTemporaryDir}/yarn.lock`, 'utf-8'), + ]); + + return { packageJson, yarnLock }; + } finally { + await temporaryDirManager.clean(); + } + } + + private async publishLayer({ + layerName, + zipBuffer, + }: { + layerName: string; + zipBuffer: Buffer; + }): Promise { + const lambdaClient = await this.awsClient.getLambdaClient(); + + const result = await lambdaClient.send( + new PublishLayerVersionCommand({ + LayerName: layerName, + Content: { ZipFile: zipBuffer }, + CompatibleRuntimes: [ + LogicFunctionRuntime.NODE18, + LogicFunctionRuntime.NODE22, + ], + }), + ); + + if (!isDefined(result.LayerVersionArn)) { + throw new Error('New layer version ARN is undefined'); + } + + return result.LayerVersionArn; + } + + private async deleteAllLayerVersions(layerName: string): Promise { + const lambdaClient = await this.awsClient.getLambdaClient(); + let marker: string | undefined; + + do { + const listResult = await lambdaClient.send( + new ListLayerVersionsCommand({ + LayerName: layerName, + MaxItems: 50, + Marker: marker, + }), + ); + + const versions = listResult.LayerVersions ?? []; + + await Promise.all( + versions.map((version) => + lambdaClient.send( + new DeleteLayerVersionCommand({ + LayerName: layerName, + VersionNumber: version.Version, + }), + ), + ), + ); + + marker = listResult.NextMarker; + } while (isDefined(marker)); + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-tool-functions.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-tool-functions.service.ts new file mode 100644 index 00000000000..8d0cf4f032e --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-tool-functions.service.ts @@ -0,0 +1,364 @@ +import * as fs from 'fs/promises'; +import { join } from 'path'; + +import { + CreateFunctionCommand, + type CreateFunctionCommandInput, + GetFunctionCommand, + InvokeCommand, + LogType, + PublishLayerVersionCommand, + ResourceNotFoundException, +} from '@aws-sdk/client-lambda'; +import { isNonEmptyString } from '@sniptt/guards'; +import { isDefined } from 'twenty-shared/utils'; + +import { COMMON_LAYER_DEPENDENCIES_DIRNAME } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/common-layer-dependencies-dirname'; +import { + BUILDER_FUNCTION_NAME_PREFIX, + BUILDER_HANDLER_PATH, + BUILDER_LAMBDA_MEMORY_MB, + BUILDER_LAMBDA_TIMEOUT_SECONDS, + COMMON_LAYER_NAME_PREFIX, + LAMBDA_EPHEMERAL_STORAGE_MB, + YARN_INSTALL_FUNCTION_NAME_PREFIX, + YARN_INSTALL_HANDLER_PATH, + YARN_INSTALL_LAMBDA_MEMORY_MB, + YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/constants/lambda-driver.constant'; +import { + type BuilderLambdaPayload, + type BuilderLambdaResult, + type LambdaDriverOptions, + type YarnInstallLambdaPayload, + type YarnInstallLambdaResult, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type'; +import { computeHashedLambdaResourceName } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/compute-hashed-lambda-resource-name.util'; +import { type LambdaAwsClientService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/services/lambda-aws-client.service'; +import { copyBuilder } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-builder'; +import { copyCommonLayerDependencies } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-common-layer-dependencies'; +import { copyYarnInstall } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-yarn-install'; +import { createZipFile } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/create-zip-file'; +import { TemporaryDirManager } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/temporary-dir-manager'; +import { LogicFunctionRuntime } from 'src/engine/metadata-modules/logic-function/logic-function.entity'; +import { + LogicFunctionException, + LogicFunctionExceptionCode, +} from 'src/engine/metadata-modules/logic-function/logic-function.exception'; + +export class LambdaToolFunctionsService { + private commonLayerName: string | undefined; + private yarnInstallFunctionName: string | undefined; + private builderFunctionName: string | undefined; + + constructor( + private readonly options: Pick, + private readonly awsClient: LambdaAwsClientService, + ) {} + + async transpile( + params: Omit, + ): Promise { + await this.ensureBuilderLambdaExists(); + + const lambdaClient = await this.awsClient.getLambdaClient(); + const builderFunctionName = await this.getBuilderFunctionName(); + + const payload: BuilderLambdaPayload = { + action: 'transpile', + ...params, + }; + + const result = await lambdaClient.send( + new InvokeCommand({ + FunctionName: builderFunctionName, + Payload: JSON.stringify(payload), + LogType: LogType.Tail, + }), + { + abortSignal: AbortSignal.timeout(BUILDER_LAMBDA_TIMEOUT_SECONDS * 1000), + }, + ); + + if (result.FunctionError) { + const parsedResult = result.Payload + ? JSON.parse(result.Payload.transformToString()) + : {}; + + const userCompilationErrorRegex = /^Build failed with \d+ error/; + + const isUserCompilationError = + isNonEmptyString(parsedResult?.errorMessage) && + userCompilationErrorRegex.test(parsedResult.errorMessage); + + if (isUserCompilationError) { + throw new LogicFunctionException( + `Function code compilation failed: ${parsedResult.errorMessage}`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_COMPILATION_FAILED, + ); + } + + throw new LogicFunctionException( + `Builder Lambda failed: ${JSON.stringify(parsedResult)}`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_CREATE_FAILED, + ); + } + + const parsedResult: BuilderLambdaResult = result.Payload + ? JSON.parse(result.Payload.transformToString()) + : {}; + + if (!parsedResult.builtCode) { + throw new Error('Builder Lambda did not return builtCode'); + } + + return { builtCode: parsedResult.builtCode }; + } + + async runYarnInstallCreateLayer( + params: Omit, + ): Promise { + await this.ensureYarnInstallLambdaExists(); + + const lambdaClient = await this.awsClient.getLambdaClient(); + + const yarnInstallFunctionName = await this.getYarnInstallFunctionName(); + + const payload: YarnInstallLambdaPayload = { + action: 'createLayer', + ...params, + }; + + const result = await lambdaClient.send( + new InvokeCommand({ + FunctionName: yarnInstallFunctionName, + Payload: JSON.stringify(payload), + LogType: LogType.Tail, + }), + { + abortSignal: AbortSignal.timeout( + YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS * 1000, + ), + }, + ); + + if (result.FunctionError) { + const parsedResult = result.Payload + ? JSON.parse(result.Payload.transformToString()) + : {}; + + throw new LogicFunctionException( + `Yarn install Lambda failed: ${JSON.stringify(parsedResult)}`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_CREATE_FAILED, + ); + } + + const parsedResult: YarnInstallLambdaResult = result.Payload + ? JSON.parse(result.Payload.transformToString()) + : {}; + + if (!parsedResult.success) { + throw new Error('Yarn install Lambda did not report success'); + } + + return parsedResult; + } + + async ensureCommonLayerExists(): Promise { + const commonLayerName = await this.getCommonLayerName(); + const existingArn = + await this.awsClient.getExistingLayerArn(commonLayerName); + + if (isDefined(existingArn)) { + return existingArn; + } + + const temporaryDirManager = new TemporaryDirManager(); + const { sourceTemporaryDir, lambdaZipPath } = + await temporaryDirManager.init(); + + try { + await copyCommonLayerDependencies(sourceTemporaryDir); + + await createZipFile(sourceTemporaryDir, lambdaZipPath); + + const lambdaClient = await this.awsClient.getLambdaClient(); + + const result = await lambdaClient.send( + new PublishLayerVersionCommand({ + LayerName: commonLayerName, + Content: { ZipFile: await fs.readFile(lambdaZipPath) }, + CompatibleRuntimes: [ + LogicFunctionRuntime.NODE18, + LogicFunctionRuntime.NODE22, + ], + }), + ); + + if (!result.LayerVersionArn) { + throw new Error( + 'PublishLayerVersion did not return a LayerVersionArn for common layer', + ); + } + + return result.LayerVersionArn; + } finally { + await temporaryDirManager.clean(); + } + } + + private async ensureYarnInstallLambdaExists(): Promise { + const yarnInstallFunctionName = await this.getYarnInstallFunctionName(); + const lambdaClient = await this.awsClient.getLambdaClient(); + + try { + await lambdaClient.send( + new GetFunctionCommand({ FunctionName: yarnInstallFunctionName }), + ); + + return; + } catch (error) { + if (!(error instanceof ResourceNotFoundException)) { + throw error; + } + } + + const commonLayerArn = await this.ensureCommonLayerExists(); + + const temporaryDirManager = new TemporaryDirManager(); + const { sourceTemporaryDir, lambdaZipPath } = + await temporaryDirManager.init(); + + try { + await copyYarnInstall(sourceTemporaryDir); + + await createZipFile(sourceTemporaryDir, lambdaZipPath); + + const params: CreateFunctionCommandInput = { + Code: { + ZipFile: await fs.readFile(lambdaZipPath), + }, + FunctionName: yarnInstallFunctionName, + Layers: [commonLayerArn], + Handler: 'index.handler', + Role: this.options.lambdaRole, + Runtime: LogicFunctionRuntime.NODE22, + Timeout: YARN_INSTALL_LAMBDA_TIMEOUT_SECONDS, + MemorySize: YARN_INSTALL_LAMBDA_MEMORY_MB, + EphemeralStorage: { Size: LAMBDA_EPHEMERAL_STORAGE_MB }, + }; + + await lambdaClient.send(new CreateFunctionCommand(params)); + } finally { + await temporaryDirManager.clean(); + } + + await this.awsClient.waitFunctionActive(yarnInstallFunctionName); + } + + private async ensureBuilderLambdaExists(): Promise { + const builderFunctionName = await this.getBuilderFunctionName(); + const lambdaClient = await this.awsClient.getLambdaClient(); + + try { + await lambdaClient.send( + new GetFunctionCommand({ FunctionName: builderFunctionName }), + ); + + return; + } catch (error) { + if (!(error instanceof ResourceNotFoundException)) { + throw error; + } + } + + const commonLayerArn = await this.ensureCommonLayerExists(); + + const temporaryDirManager = new TemporaryDirManager(); + const { sourceTemporaryDir, lambdaZipPath } = + await temporaryDirManager.init(); + + try { + await copyBuilder(sourceTemporaryDir); + + await createZipFile(sourceTemporaryDir, lambdaZipPath); + + const params: CreateFunctionCommandInput = { + Code: { + ZipFile: await fs.readFile(lambdaZipPath), + }, + FunctionName: builderFunctionName, + Layers: [commonLayerArn], + Handler: 'index.handler', + Role: this.options.lambdaRole, + Runtime: LogicFunctionRuntime.NODE22, + Timeout: BUILDER_LAMBDA_TIMEOUT_SECONDS, + MemorySize: BUILDER_LAMBDA_MEMORY_MB, + EphemeralStorage: { Size: LAMBDA_EPHEMERAL_STORAGE_MB }, + }; + + await lambdaClient.send(new CreateFunctionCommand(params)); + } finally { + await temporaryDirManager.clean(); + } + + await this.awsClient.waitFunctionActive(builderFunctionName); + } + + private async getCommonLayerName(): Promise { + if (isDefined(this.commonLayerName)) { + return this.commonLayerName; + } + + const [packageJson, yarnLock] = await Promise.all([ + fs.readFile( + join(COMMON_LAYER_DEPENDENCIES_DIRNAME, 'package.json'), + 'utf-8', + ), + fs.readFile( + join(COMMON_LAYER_DEPENDENCIES_DIRNAME, 'yarn.lock'), + 'utf-8', + ), + ]); + + this.commonLayerName = computeHashedLambdaResourceName({ + prefix: COMMON_LAYER_NAME_PREFIX, + contents: [packageJson, yarnLock], + }); + + return this.commonLayerName; + } + + private async getYarnInstallFunctionName(): Promise { + if (isDefined(this.yarnInstallFunctionName)) { + return this.yarnInstallFunctionName; + } + + const handlerContent = await fs.readFile( + YARN_INSTALL_HANDLER_PATH, + 'utf-8', + ); + + this.yarnInstallFunctionName = computeHashedLambdaResourceName({ + prefix: YARN_INSTALL_FUNCTION_NAME_PREFIX, + contents: [handlerContent], + }); + + return this.yarnInstallFunctionName; + } + + private async getBuilderFunctionName(): Promise { + if (isDefined(this.builderFunctionName)) { + return this.builderFunctionName; + } + + const handlerContent = await fs.readFile(BUILDER_HANDLER_PATH, 'utf-8'); + + this.builderFunctionName = computeHashedLambdaResourceName({ + prefix: BUILDER_FUNCTION_NAME_PREFIX, + contents: [handlerContent], + }); + + return this.builderFunctionName; + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type.ts new file mode 100644 index 00000000000..6f23c8d5380 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/types/lambda-driver.type.ts @@ -0,0 +1,51 @@ +import { type LambdaClientConfig } from '@aws-sdk/client-lambda'; + +import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; +import { type SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service'; + +export type LambdaDriverExecutorPayload = { + code?: string; + params: object; + env: Record; + handlerName: string; +}; + +export type YarnInstallLambdaPayload = { + action: 'createLayer'; + packageJson: string; + yarnLock: string; + presignedUploadUrl: string; +}; + +export type YarnInstallLambdaResult = { + success: boolean; +}; + +export type BuilderLambdaPayload = { + action: 'transpile'; + sourceCode: string; + sourceFileName: string; + builtFileName: string; +}; + +export type BuilderLambdaResult = { + builtCode: string; +}; + +export interface LambdaDriverOptions extends LambdaClientConfig { + logicFunctionResourceService: LogicFunctionResourceService; + sdkClientArchiveService: SdkClientArchiveService; + cacheLockService: CacheLockService; + region: string; + lambdaRole: string; + subhostingRole?: string; + layerBucket: string; + layerBucketRegion: string; +} + +export enum LambdaExecutionPhase { + BUILD = 'build', + FETCH_CODE = 'fetch-code', + INVOKE = 'invoke', +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/compute-hashed-lambda-resource-name.util.spec.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/compute-hashed-lambda-resource-name.util.spec.ts new file mode 100644 index 00000000000..d05e725da8b --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/compute-hashed-lambda-resource-name.util.spec.ts @@ -0,0 +1,12 @@ +import { computeHashedLambdaResourceName } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/compute-hashed-lambda-resource-name.util'; + +describe('computeHashedLambdaResourceName', () => { + it('returns prefix concatenated with a 12-character sha256 hex checksum of the contents', () => { + const name = computeHashedLambdaResourceName({ + prefix: 'twenty-builder', + contents: ['handler-code'], + }); + + expect(name).toMatch(/^twenty-builder-[0-9a-f]{12}$/); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/get-lambda-deps-layer-name.util.spec.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/get-lambda-deps-layer-name.util.spec.ts new file mode 100644 index 00000000000..e74584f3706 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/get-lambda-deps-layer-name.util.spec.ts @@ -0,0 +1,34 @@ +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { getLambdaDepsLayerName } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-deps-layer-name.util'; + +const buildFlatApplication = ( + overrides: Partial = {}, +): FlatApplication => + ({ + yarnLockChecksum: 'abc123', + ...overrides, + }) as FlatApplication; + +describe('getLambdaDepsLayerName', () => { + it('returns deps- when yarnLockChecksum is set', () => { + expect(getLambdaDepsLayerName(buildFlatApplication())).toBe('deps-abc123'); + }); + + it('falls back to deps-default when yarnLockChecksum is undefined', () => { + expect( + getLambdaDepsLayerName( + buildFlatApplication({ yarnLockChecksum: undefined }), + ), + ).toBe('deps-default'); + }); + + it('falls back to deps-default when yarnLockChecksum is null', () => { + expect( + getLambdaDepsLayerName( + buildFlatApplication({ + yarnLockChecksum: null as unknown as string, + }), + ), + ).toBe('deps-default'); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/get-lambda-sdk-layer-name.util.spec.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/get-lambda-sdk-layer-name.util.spec.ts new file mode 100644 index 00000000000..259dc575e34 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/get-lambda-sdk-layer-name.util.spec.ts @@ -0,0 +1,25 @@ +import { getLambdaSdkLayerName } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-sdk-layer-name.util'; + +describe('getLambdaSdkLayerName', () => { + it('joins prefix with workspaceId and applicationUniversalIdentifier', () => { + expect( + getLambdaSdkLayerName({ + workspaceId: 'ws-1', + applicationUniversalIdentifier: 'app-2', + }), + ).toBe('sdk-ws-1-app-2'); + }); + + it('produces distinct names for distinct identifiers', () => { + const a = getLambdaSdkLayerName({ + workspaceId: 'a', + applicationUniversalIdentifier: 'x', + }); + const b = getLambdaSdkLayerName({ + workspaceId: 'b', + applicationUniversalIdentifier: 'x', + }); + + expect(a).not.toBe(b); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/parse-lambda-log-result.util.spec.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/parse-lambda-log-result.util.spec.ts new file mode 100644 index 00000000000..11928848749 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/__tests__/parse-lambda-log-result.util.spec.ts @@ -0,0 +1,84 @@ +import { parseLambdaLogResult } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/parse-lambda-log-result.util'; + +const toBase64 = (input: string): string => + Buffer.from(input, 'utf8').toString('base64'); + +describe('parseLambdaLogResult', () => { + it('returns empty defaults when logResult is undefined', () => { + expect(parseLambdaLogResult(undefined)).toEqual({ + logs: '', + initDurationMs: null, + billedDurationMs: null, + reportDurationMs: null, + coldStart: false, + }); + }); + + it('returns empty defaults when logResult is an empty string', () => { + expect(parseLambdaLogResult('')).toEqual({ + logs: '', + initDurationMs: null, + billedDurationMs: null, + reportDurationMs: null, + coldStart: false, + }); + }); + + it('extracts Init/Billed/Duration milliseconds from a REPORT line and flags coldStart=true', () => { + const raw = [ + 'START RequestId: 1 Version: $LATEST', + '2026-06-02T12:00:00.000Z 11111111-1111-1111-1111-111111111111 INFO hello', + 'END RequestId: 1', + 'REPORT RequestId: 1 Duration: 12.34 ms Billed Duration: 13 ms Memory Size: 512 MB Max Memory Used: 100 MB Init Duration: 250.50 ms', + ].join('\n'); + + const result = parseLambdaLogResult(toBase64(raw)); + + expect(result.initDurationMs).toBe('250.50'); + expect(result.billedDurationMs).toBe('13'); + expect(result.reportDurationMs).toBe('12.34'); + expect(result.coldStart).toBe(true); + }); + + it('flags coldStart=false when no Init Duration is present', () => { + const raw = + 'REPORT RequestId: 1 Duration: 5 ms Billed Duration: 5 ms Memory Size: 512 MB Max Memory Used: 100 MB'; + + const result = parseLambdaLogResult(toBase64(raw)); + + expect(result.initDurationMs).toBeNull(); + expect(result.coldStart).toBe(false); + expect(result.reportDurationMs).toBe('5'); + expect(result.billedDurationMs).toBe('5'); + }); + + it('strips START/END/REPORT lines, request-id from INFO lines, and normalizes tabs to spaces', () => { + const raw = [ + 'START RequestId: abc', + '2026-06-02T12:00:00.000Z\t11111111-1111-1111-1111-111111111111\tINFO\thello', + 'END RequestId: abc', + 'REPORT RequestId: abc Duration: 5 ms', + ].join('\n'); + + const result = parseLambdaLogResult(toBase64(raw)); + + expect(result.logs).toBe('2026-06-02T12:00:00.000Z INFO hello'); + }); + + it('trims leading and trailing whitespace from logs', () => { + const raw = [ + 'START RequestId: abc', + '', + '2026-06-02T12:00:00.000Z 11111111-1111-1111-1111-111111111111 INFO line', + '', + 'END RequestId: abc', + ].join('\n'); + + const result = parseLambdaLogResult(toBase64(raw)); + + expect(result.logs.startsWith('2026-06-02T12:00:00.000Z INFO line')).toBe( + true, + ); + expect(result.logs.endsWith('INFO line')).toBe(true); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/compute-hashed-lambda-resource-name.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/compute-hashed-lambda-resource-name.util.ts new file mode 100644 index 00000000000..0303f7f53cc --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/compute-hashed-lambda-resource-name.util.ts @@ -0,0 +1,21 @@ +import { createHash } from 'crypto'; + +const RESOURCE_NAME_CHECKSUM_LENGTH = 12; + +export const computeHashedLambdaResourceName = ({ + prefix, + contents, +}: { + prefix: string; + contents: ReadonlyArray; +}): string => { + const hash = createHash('sha256'); + + for (const content of contents) { + hash.update(content); + } + + const checksum = hash.digest('hex').slice(0, RESOURCE_NAME_CHECKSUM_LENGTH); + + return `${prefix}-${checksum}`; +}; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-deps-layer-name.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-deps-layer-name.util.ts new file mode 100644 index 00000000000..52455b80d51 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-deps-layer-name.util.ts @@ -0,0 +1,9 @@ +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; + +export const getLambdaDepsLayerName = ( + flatApplication: FlatApplication, +): string => { + const checksum = flatApplication.yarnLockChecksum ?? 'default'; + + return `deps-${checksum}`; +}; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-sdk-layer-name.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-sdk-layer-name.util.ts new file mode 100644 index 00000000000..14b0857eef9 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/get-lambda-sdk-layer-name.util.ts @@ -0,0 +1,7 @@ +export const getLambdaSdkLayerName = ({ + workspaceId, + applicationUniversalIdentifier, +}: { + workspaceId: string; + applicationUniversalIdentifier: string; +}): string => `sdk-${workspaceId}-${applicationUniversalIdentifier}`; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/parse-lambda-log-result.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/parse-lambda-log-result.util.ts new file mode 100644 index 00000000000..07350b331b2 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/parse-lambda-log-result.util.ts @@ -0,0 +1,50 @@ +export type ParsedLambdaLogResult = { + logs: string; + initDurationMs: string | null; + billedDurationMs: string | null; + reportDurationMs: string | null; + coldStart: boolean; +}; + +const EMPTY_PARSED_LAMBDA_LOG_RESULT: ParsedLambdaLogResult = { + logs: '', + initDurationMs: null, + billedDurationMs: null, + reportDurationMs: null, + coldStart: false, +}; + +export const parseLambdaLogResult = ( + logResult: string | undefined, +): ParsedLambdaLogResult => { + if (!logResult) { + return EMPTY_PARSED_LAMBDA_LOG_RESULT; + } + + const decoded = Buffer.from(logResult, 'base64').toString('utf8'); + + const initDurationMs = + decoded.match(/Init Duration:\s*([\d.]+)\s*ms/i)?.[1] ?? null; + const billedDurationMs = + decoded.match(/Billed Duration:\s*([\d.]+)\s*ms/i)?.[1] ?? null; + const reportDurationMs = + decoded.match(/\bDuration:\s*([\d.]+)\s*ms/i)?.[1] ?? null; + + const logs = decoded + .split('\t') + .join(' ') + .replace(/^(START|END|REPORT).*\n?/gm, '') + .replace( + /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z) [a-f0-9-]+ INFO /gm, + '$1 INFO ', + ) + .trim(); + + return { + logs, + initDurationMs, + billedDurationMs, + reportDurationMs, + coldStart: initDurationMs !== null, + }; +}; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/reprefix-lambda-zip-entries.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/reprefix-lambda-zip-entries.util.ts new file mode 100644 index 00000000000..8a250715429 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda/utils/reprefix-lambda-zip-entries.util.ts @@ -0,0 +1,36 @@ +// Re-wraps zip entries under a new prefix path without extracting to disk. +export const reprefixLambdaZipEntries = async ({ + sourceBuffer, + prefix, +}: { + sourceBuffer: Buffer; + prefix: string; +}): Promise => { + const { default: unzipper } = await import('unzipper'); + const archiver = (await import('archiver')).default; + + const directory = await unzipper.Open.buffer(sourceBuffer); + const archive = archiver('zip', { zlib: { level: 9 } }); + + const chunks: Buffer[] = []; + + archive.on('data', (chunk: Buffer) => chunks.push(chunk)); + + for (const entry of directory.files) { + if (entry.type === 'Directory') { + continue; + } + + archive.append(entry.stream(), { + name: `${prefix}/${entry.path}`, + }); + } + + await new Promise((resolve, reject) => { + archive.on('end', resolve); + archive.on('error', reject); + void archive.finalize(); + }); + + return Buffer.concat(chunks); +}; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local.driver.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local.driver.ts index 1f7b7c84705..16fc078c846 100644 --- a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local.driver.ts +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local.driver.ts @@ -1,26 +1,25 @@ import { promises as fs } from 'fs'; -import { spawn } from 'node:child_process'; import { dirname, join } from 'path'; import { build } from 'esbuild'; import { NODE_ESM_CJS_BANNER } from 'twenty-shared/application'; import { + type LogicFunctionDriver, type LogicFunctionExecuteParams, type LogicFunctionExecuteResult, - type LogicFunctionDriver, type LogicFunctionInstallPrebuiltBundleParams, type LogicFunctionTranspileParams, type LogicFunctionTranspileResult, } from 'src/engine/core-modules/logic-function/logic-function-drivers/interfaces/logic-function-driver.interface'; -import { isNonEmptyString } from '@sniptt/guards'; -import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; -import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; -import { LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/logic-function-executor-tmpdir-folder'; +import { LocalChildProcessRunnerService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service'; +import { LocalLayerManagerService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-layer-manager.service'; +import { LocalPrebuiltBundleService } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-prebuilt-bundle.service'; +import { type LocalDriverOptions } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/types/local-driver.type'; import { ConsoleListener } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/intercept-console'; import { TemporaryDirManager } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/temporary-dir-manager'; -import { HANDLER_NAME_REGEX } from 'src/engine/metadata-modules/logic-function/constants/handler.contant'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; import { LogicFunctionExecutionStatus } from 'src/engine/metadata-modules/logic-function/dtos/logic-function-execution-result.dto'; import { LogicFunctionExecutionMode } from 'src/engine/metadata-modules/logic-function/logic-function.entity'; import { @@ -29,179 +28,27 @@ import { } from 'src/engine/metadata-modules/logic-function/logic-function.exception'; import { type FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type'; import { isLogicFunctionReadyForPrebuiltInstall } from 'src/engine/metadata-modules/logic-function/utils/is-logic-function-ready-for-prebuilt-install.util'; -import { copyYarnEngineAndBuildDependencies } from 'src/engine/core-modules/application/application-package/utils/copy-yarn-engine-and-build-dependencies'; -import type { LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; -import type { SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service'; -import type { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; -const LAYER_BUILD_LOCK_TTL_MS = 120_000; -const LAYER_BUILD_LOCK_RETRY_MS = 500; -const LAYER_BUILD_LOCK_MAX_RETRIES = 240; -const LAYER_BUILD_READY_SENTINEL = '.twenty-layer-ready'; - -const PREBUILT_BUNDLE_FILE_NAME = 'prebuilt-logic-function.mjs'; -const PREBUILT_CHECKSUM_FILE_NAME = 'prebuilt-bundle.checksum'; - -const PREBUILT_INSTALL_LOCK_TTL_MS = 180_000; -const PREBUILT_INSTALL_LOCK_RETRY_MS = 1_000; -const PREBUILT_INSTALL_LOCK_MAX_RETRIES = 180; - -export interface LocalDriverOptions { - logicFunctionResourceService: LogicFunctionResourceService; - sdkClientArchiveService: SdkClientArchiveService; - cacheLockService: CacheLockService; - workspaceCacheService: WorkspaceCacheService; -} - -const pathExists = async (targetPath: string): Promise => { - try { - await fs.access(targetPath); - - return true; - } catch { - return false; - } -}; +export { type LocalDriverOptions } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/types/local-driver.type'; export class LocalDriver implements LogicFunctionDriver { private readonly logicFunctionResourceService: LogicFunctionResourceService; - private readonly sdkClientArchiveService: SdkClientArchiveService; - private readonly cacheLockService: CacheLockService; - private readonly workspaceCacheService: WorkspaceCacheService; + private readonly layerManager: LocalLayerManagerService; + private readonly childProcessRunner: LocalChildProcessRunnerService; + private readonly prebuiltBundle: LocalPrebuiltBundleService; constructor(options: LocalDriverOptions) { this.logicFunctionResourceService = options.logicFunctionResourceService; - this.sdkClientArchiveService = options.sdkClientArchiveService; - this.cacheLockService = options.cacheLockService; - this.workspaceCacheService = options.workspaceCacheService; - } - - private getDepsLayerPath(flatApplication: FlatApplication): string { - const checksum = flatApplication.yarnLockChecksum ?? 'default'; - - return join(LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER, 'deps', checksum); - } - - private getSdkLayerPath({ - workspaceId, - applicationUniversalIdentifier, - }: { - workspaceId: string; - applicationUniversalIdentifier: string; - }): string { - return join( - LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER, - 'sdk', - `${workspaceId}-${applicationUniversalIdentifier}`, + this.layerManager = new LocalLayerManagerService( + options.cacheLockService, + options.logicFunctionResourceService, + options.sdkClientArchiveService, + options.workspaceCacheService, ); - } - - private async createLayerIfNotExist({ - flatApplication, - applicationUniversalIdentifier, - }: { - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise { - const depsLayerPath = this.getDepsLayerPath(flatApplication); - const depsReadySentinelPath = join( - depsLayerPath, - LAYER_BUILD_READY_SENTINEL, - ); - - if (await pathExists(depsReadySentinelPath)) { - return; - } - - const lockKey = `local-driver-deps-layer:${flatApplication.yarnLockChecksum ?? 'default'}`; - - await this.cacheLockService.withLock( - async () => { - if (await pathExists(depsReadySentinelPath)) { - return; - } - - await fs.rm(depsLayerPath, { recursive: true, force: true }); - - await this.logicFunctionResourceService.copyDependenciesInMemory({ - applicationUniversalIdentifier, - workspaceId: flatApplication.workspaceId, - inMemoryFolderPath: depsLayerPath, - }); - await copyYarnEngineAndBuildDependencies(depsLayerPath); - await fs.writeFile(depsReadySentinelPath, ''); - }, - lockKey, - { - ttl: LAYER_BUILD_LOCK_TTL_MS, - ms: LAYER_BUILD_LOCK_RETRY_MS, - maxRetries: LAYER_BUILD_LOCK_MAX_RETRIES, - }, - ); - } - - private async ensureSdkLayer({ - flatApplication, - applicationUniversalIdentifier, - }: { - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise { - const sdkLayerPath = this.getSdkLayerPath({ - workspaceId: flatApplication.workspaceId, - applicationUniversalIdentifier, - }); - const sdkNodeModulesPath = join(sdkLayerPath, 'node_modules'); - const sdkReadySentinelPath = join(sdkLayerPath, LAYER_BUILD_READY_SENTINEL); - - if ( - (await pathExists(sdkReadySentinelPath)) && - !flatApplication.isSdkLayerStale - ) { - return; - } - - const lockKey = `local-driver-sdk-layer:${flatApplication.workspaceId}:${applicationUniversalIdentifier}`; - - await this.cacheLockService.withLock( - async () => { - const { flatApplicationMaps } = - await this.workspaceCacheService.getOrRecompute( - flatApplication.workspaceId, - ['flatApplicationMaps'], - ); - const freshFlatApplication = - flatApplicationMaps.byId[flatApplication.id]; - const isStale = freshFlatApplication?.isSdkLayerStale ?? true; - - if ((await pathExists(sdkReadySentinelPath)) && !isStale) { - return; - } - - await fs.rm(sdkLayerPath, { recursive: true, force: true }); - - const sdkPackagePath = join(sdkNodeModulesPath, 'twenty-client-sdk'); - - await this.sdkClientArchiveService.downloadAndExtractToPackage({ - workspaceId: flatApplication.workspaceId, - applicationId: flatApplication.id, - applicationUniversalIdentifier, - targetPackagePath: sdkPackagePath, - }); - - await this.sdkClientArchiveService.markSdkLayerFresh({ - applicationId: flatApplication.id, - workspaceId: flatApplication.workspaceId, - }); - - await fs.writeFile(sdkReadySentinelPath, ''); - }, - lockKey, - { - ttl: LAYER_BUILD_LOCK_TTL_MS, - ms: LAYER_BUILD_LOCK_RETRY_MS, - maxRetries: LAYER_BUILD_LOCK_MAX_RETRIES, - }, + this.childProcessRunner = new LocalChildProcessRunnerService(); + this.prebuiltBundle = new LocalPrebuiltBundleService( + options.cacheLockService, + options.logicFunctionResourceService, ); } @@ -241,55 +88,18 @@ export class LocalDriver implements LogicFunctionDriver { } } - async delete() {} + async delete(): Promise {} - // Symlinks everything from the deps layer except twenty-client-sdk, - // which comes from the SDK layer (workspace-specific generated client). - private async assembleNodeModules({ - sourceTemporaryDir, - flatApplication, - applicationUniversalIdentifier, - }: { - sourceTemporaryDir: string; - flatApplication: FlatApplication; - applicationUniversalIdentifier: string; - }): Promise { - const depsNodeModules = join( - this.getDepsLayerPath(flatApplication), - 'node_modules', - ); - const sdkNodeModules = join( - this.getSdkLayerPath({ - workspaceId: flatApplication.workspaceId, - applicationUniversalIdentifier, - }), - 'node_modules', - ); - const execNodeModules = join(sourceTemporaryDir, 'node_modules'); + async installPrebuiltBundle( + params: LogicFunctionInstallPrebuiltBundleParams, + ): Promise { + await this.prebuiltBundle.installPrebuiltBundle(params); + } - await fs.mkdir(execNodeModules, { recursive: true }); - - const entries = await fs.readdir(depsNodeModules, { - withFileTypes: true, - }); - - const symlinkPromises = entries - .filter((entry) => entry.name !== 'twenty-client-sdk') - .map((entry) => - fs.symlink( - join(depsNodeModules, entry.name), - join(execNodeModules, entry.name), - entry.isDirectory() ? 'dir' : 'file', - ), - ); - - await Promise.all(symlinkPromises); - - await fs.symlink( - join(sdkNodeModules, 'twenty-client-sdk'), - join(execNodeModules, 'twenty-client-sdk'), - 'dir', - ); + async getInstalledBundleChecksum( + flatLogicFunction: FlatLogicFunction, + ): Promise { + return this.prebuiltBundle.getInstalledBundleChecksum(flatLogicFunction); } async execute({ @@ -313,23 +123,22 @@ export class LocalDriver implements LogicFunctionDriver { ); } - await this.createLayerIfNotExist({ + await this.layerManager.ensureDepsLayer({ flatApplication, applicationUniversalIdentifier, }); - await this.ensureSdkLayer({ + await this.layerManager.ensureSdkLayer({ flatApplication, applicationUniversalIdentifier, }); const startTime = Date.now(); - const temporaryDirManager = new TemporaryDirManager(); try { const { sourceTemporaryDir } = await temporaryDirManager.init(); - await this.assembleNodeModules({ + await this.childProcessRunner.assembleNodeModules({ sourceTemporaryDir, flatApplication, applicationUniversalIdentifier, @@ -337,7 +146,7 @@ export class LocalDriver implements LogicFunctionDriver { const inMemoryBuiltHandlerPath = executionMode === LogicFunctionExecutionMode.PREBUILT - ? await this.copyPrebuiltBundleIntoExecutionDir({ + ? await this.prebuiltBundle.copyPrebuiltBundleIntoExecutionDir({ flatLogicFunction, sourceTemporaryDir, }) @@ -349,7 +158,6 @@ export class LocalDriver implements LogicFunctionDriver { }); let logs = ''; - const consoleListener = new ConsoleListener(); consoleListener.intercept((type, args) => { @@ -382,14 +190,14 @@ export class LocalDriver implements LogicFunctionDriver { }); try { - const runnerPath = await this.writeBootstrapRunner({ + const runnerPath = await this.childProcessRunner.writeBootstrapRunner({ dir: sourceTemporaryDir, builtFileAbsPath: inMemoryBuiltHandlerPath, handlerName: flatLogicFunction.handlerName, }); const { ok, result, error, stack, stdout, stderr } = - await this.runChildWithEnv({ + await this.childProcessRunner.runChildWithEnv({ runnerPath, env: env ?? {}, payload, @@ -440,269 +248,4 @@ export class LocalDriver implements LogicFunctionDriver { await temporaryDirManager.clean(); } } - - async writeBootstrapRunner({ - dir, - builtFileAbsPath, - handlerName, - }: { - dir: string; - builtFileAbsPath: string; - handlerName: string; - }) { - if (!HANDLER_NAME_REGEX.test(handlerName)) { - throw new Error( - `Invalid handlerName "${handlerName}": must be a valid JavaScript identifier or dotted path`, - ); - } - - const runnerPath = join(dir, '__runner.cjs'); - const code = ` - // Auto-generated. Do not edit. - const { pathToFileURL } = require('node:url'); - - (async () => { - try { - const builtUrl = pathToFileURL(${JSON.stringify(builtFileAbsPath)}); - const mod = await import(builtUrl.href); - if (typeof mod.${handlerName} !== 'function') { - throw new Error('Export "${handlerName}" not found in function bundle'); - } - - let payload = undefined; - if (process.send) { - process.on('message', async (msg) => { - if (!msg || msg.type !== 'run') return; - try { - const out = await mod.${handlerName}(msg.payload); - process.send && process.send({ ok: true, result: out }); - process.exit(0); - } catch (err) { - process.send && process.send({ ok: false, error: String(err), stack: err?.stack }); - process.exit(1); - } - }); - } else { - // Fallback: read payload from argv[2] (JSON) and print to stdout - const json = process.argv[2]; - payload = json ? JSON.parse(json) : undefined; - const out = await mod.${handlerName}(payload); - process.stdout.write(JSON.stringify({ ok: true, result: out })); - process.exit(0); - } - } catch (err) { - const msg = String(err); - if (process.send) { - process.send({ ok: false, error: msg, stack: err?.stack }); - } else { - process.stdout.write(msg); - } - process.exit(1); - } - })(); - `; - - await fs.writeFile(runnerPath, code, 'utf8'); - - return runnerPath; - } - - runChildWithEnv(options: { - runnerPath: string; - env: Record; - payload: unknown; - timeoutMs: number; - }) { - const { runnerPath, env, payload, timeoutMs } = options; - - return new Promise<{ - ok: boolean; - result?: unknown; - error?: string; - stack?: string; - stdout: string; - stderr: string; - }>((resolve) => { - // Strip NODE_OPTIONS to prevent tsx loader from being inherited - const { NODE_OPTIONS: _n1, ...cleanProcessEnv } = process.env; - const { NODE_OPTIONS: _n2, ...cleanUserEnv } = env; - - const child = spawn(process.execPath, [runnerPath], { - env: { ...cleanProcessEnv, ...cleanUserEnv }, - stdio: ['pipe', 'pipe', 'pipe', 'ipc'], - }); - - let stdout = ''; - let stderr = ''; - let settled = false; - - child.stdout?.on('data', (d) => (stdout += String(d))); - child.stderr?.on('data', (d) => (stderr += String(d))); - - child.on( - 'message', - ( - msg: - | { - ok: true; - result?: unknown; - stdout?: string; - stderr?: string; - } - | { - ok: false; - error: string; - stack?: string; - stdout?: string; - stderr?: string; - }, - ) => { - if (settled) return; - settled = true; - resolve({ ...msg, stdout, stderr }); - }, - ); - - child.on('exit', (code) => { - if (settled) return; - settled = true; - if (code === 0) { - resolve({ ok: true, stdout, stderr }); - } else { - resolve({ - ok: false, - error: `Exited with code ${code}`, - stdout, - stderr, - }); - } - }); - - const t = setTimeout(() => { - if (settled) return; - settled = true; - child.kill('SIGKILL'); - resolve({ - ok: false, - error: `Timed out after ${timeoutMs}ms`, - stdout, - stderr, - }); - }, timeoutMs); - - child.send?.({ type: 'run', payload }); - - child.on('close', () => clearTimeout(t)); - }); - } - - private getPrebuiltBundleDir(flatLogicFunction: FlatLogicFunction): string { - return join( - LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER, - 'prebuilt', - flatLogicFunction.id, - ); - } - - private getInstalledBundlePath(flatLogicFunction: FlatLogicFunction): string { - return join( - this.getPrebuiltBundleDir(flatLogicFunction), - PREBUILT_BUNDLE_FILE_NAME, - ); - } - - private async copyPrebuiltBundleIntoExecutionDir({ - flatLogicFunction, - sourceTemporaryDir, - }: { - flatLogicFunction: FlatLogicFunction; - sourceTemporaryDir: string; - }): Promise { - const installedBundlePath = this.getInstalledBundlePath(flatLogicFunction); - const localBundlePath = join(sourceTemporaryDir, PREBUILT_BUNDLE_FILE_NAME); - - await fs.copyFile(installedBundlePath, localBundlePath); - - return localBundlePath; - } - - private getInstalledChecksumPath( - flatLogicFunction: FlatLogicFunction, - ): string { - return join( - this.getPrebuiltBundleDir(flatLogicFunction), - PREBUILT_CHECKSUM_FILE_NAME, - ); - } - - private getPrebuiltInstallLockKey(flatLogicFunction: FlatLogicFunction) { - return `local-install:${flatLogicFunction.id}`; - } - - async installPrebuiltBundle({ - flatLogicFunction, - applicationUniversalIdentifier, - }: LogicFunctionInstallPrebuiltBundleParams): Promise { - if (!isNonEmptyString(flatLogicFunction.checksum)) { - throw new LogicFunctionException( - `Cannot install prebuilt bundle for function '${flatLogicFunction.id}' without a checksum`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED, - ); - } - - const checksum = flatLogicFunction.checksum; - - await this.cacheLockService.withLock( - async () => { - const prebuiltDir = this.getPrebuiltBundleDir(flatLogicFunction); - - await fs.mkdir(prebuiltDir, { recursive: true }); - - await this.logicFunctionResourceService.copyBuiltCodeInMemory({ - workspaceId: flatLogicFunction.workspaceId, - applicationUniversalIdentifier, - builtHandlerPath: flatLogicFunction.builtHandlerPath, - inMemoryDestinationPath: prebuiltDir, - }); - - const downloadedPath = join( - prebuiltDir, - flatLogicFunction.builtHandlerPath, - ); - const targetPath = this.getInstalledBundlePath(flatLogicFunction); - - if (downloadedPath !== targetPath) { - await fs.mkdir(dirname(targetPath), { recursive: true }); - await fs.rename(downloadedPath, targetPath); - } - - await fs.writeFile( - this.getInstalledChecksumPath(flatLogicFunction), - checksum, - 'utf8', - ); - }, - this.getPrebuiltInstallLockKey(flatLogicFunction), - { - ttl: PREBUILT_INSTALL_LOCK_TTL_MS, - ms: PREBUILT_INSTALL_LOCK_RETRY_MS, - maxRetries: PREBUILT_INSTALL_LOCK_MAX_RETRIES, - }, - ); - } - - async getInstalledBundleChecksum( - flatLogicFunction: FlatLogicFunction, - ): Promise { - try { - const checksum = await fs.readFile( - this.getInstalledChecksumPath(flatLogicFunction), - 'utf8', - ); - - return checksum.trim() || null; - } catch { - return null; - } - } } diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/constants/local-driver.constant.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/constants/local-driver.constant.ts new file mode 100644 index 00000000000..80e8e39409b --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/constants/local-driver.constant.ts @@ -0,0 +1,12 @@ +export const LAYER_BUILD_LOCK_TTL_MS = 120_000; +export const LAYER_BUILD_LOCK_RETRY_MS = 500; +export const LAYER_BUILD_LOCK_MAX_RETRIES = 240; + +export const LAYER_BUILD_READY_SENTINEL = '.twenty-layer-ready'; + +export const PREBUILT_BUNDLE_FILE_NAME = 'prebuilt-logic-function.mjs'; +export const PREBUILT_CHECKSUM_FILE_NAME = 'prebuilt-bundle.checksum'; + +export const PREBUILT_INSTALL_LOCK_TTL_MS = 180_000; +export const PREBUILT_INSTALL_LOCK_RETRY_MS = 1_000; +export const PREBUILT_INSTALL_LOCK_MAX_RETRIES = 180; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts new file mode 100644 index 00000000000..a15a693ea1e --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-child-process-runner.service.ts @@ -0,0 +1,214 @@ +import { promises as fs } from 'fs'; +import { spawn } from 'node:child_process'; +import { join } from 'path'; + +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { getLocalDepsLayerPath } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-deps-layer-path.util'; +import { getLocalSdkLayerPath } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-sdk-layer-path.util'; +import { HANDLER_NAME_REGEX } from 'src/engine/metadata-modules/logic-function/constants/handler.contant'; + +export class LocalChildProcessRunnerService { + // Symlinks everything from the deps layer except twenty-client-sdk, + // which comes from the SDK layer (workspace-specific generated client). + async assembleNodeModules({ + sourceTemporaryDir, + flatApplication, + applicationUniversalIdentifier, + }: { + sourceTemporaryDir: string; + flatApplication: FlatApplication; + applicationUniversalIdentifier: string; + }): Promise { + const depsNodeModules = join( + getLocalDepsLayerPath(flatApplication), + 'node_modules', + ); + const sdkNodeModules = join( + getLocalSdkLayerPath({ + workspaceId: flatApplication.workspaceId, + applicationUniversalIdentifier, + }), + 'node_modules', + ); + const execNodeModules = join(sourceTemporaryDir, 'node_modules'); + + await fs.mkdir(execNodeModules, { recursive: true }); + + const entries = await fs.readdir(depsNodeModules, { + withFileTypes: true, + }); + + const symlinkPromises = entries + .filter((entry) => entry.name !== 'twenty-client-sdk') + .map((entry) => + fs.symlink( + join(depsNodeModules, entry.name), + join(execNodeModules, entry.name), + entry.isDirectory() ? 'dir' : 'file', + ), + ); + + await Promise.all(symlinkPromises); + + await fs.symlink( + join(sdkNodeModules, 'twenty-client-sdk'), + join(execNodeModules, 'twenty-client-sdk'), + 'dir', + ); + } + + async writeBootstrapRunner({ + dir, + builtFileAbsPath, + handlerName, + }: { + dir: string; + builtFileAbsPath: string; + handlerName: string; + }) { + if (!HANDLER_NAME_REGEX.test(handlerName)) { + throw new Error( + `Invalid handlerName "${handlerName}": must be a valid JavaScript identifier or dotted path`, + ); + } + + const runnerPath = join(dir, '__runner.cjs'); + const code = ` + // Auto-generated. Do not edit. + const { pathToFileURL } = require('node:url'); + + (async () => { + try { + const builtUrl = pathToFileURL(${JSON.stringify(builtFileAbsPath)}); + const mod = await import(builtUrl.href); + if (typeof mod.${handlerName} !== 'function') { + throw new Error('Export "${handlerName}" not found in function bundle'); + } + + let payload = undefined; + if (process.send) { + process.on('message', async (msg) => { + if (!msg || msg.type !== 'run') return; + try { + const out = await mod.${handlerName}(msg.payload); + process.send && process.send({ ok: true, result: out }); + process.exit(0); + } catch (err) { + process.send && process.send({ ok: false, error: String(err), stack: err?.stack }); + process.exit(1); + } + }); + } else { + // Fallback: read payload from argv[2] (JSON) and print to stdout + const json = process.argv[2]; + payload = json ? JSON.parse(json) : undefined; + const out = await mod.${handlerName}(payload); + process.stdout.write(JSON.stringify({ ok: true, result: out })); + process.exit(0); + } + } catch (err) { + const msg = String(err); + if (process.send) { + process.send({ ok: false, error: msg, stack: err?.stack }); + } else { + process.stdout.write(msg); + } + process.exit(1); + } + })(); + `; + + await fs.writeFile(runnerPath, code, 'utf8'); + + return runnerPath; + } + + runChildWithEnv(options: { + runnerPath: string; + env: Record; + payload: unknown; + timeoutMs: number; + }) { + const { runnerPath, env, payload, timeoutMs } = options; + + return new Promise<{ + ok: boolean; + result?: unknown; + error?: string; + stack?: string; + stdout: string; + stderr: string; + }>((resolve) => { + // Strip NODE_OPTIONS to prevent tsx loader from being inherited + const { NODE_OPTIONS: _n1, ...cleanProcessEnv } = process.env; + const { NODE_OPTIONS: _n2, ...cleanUserEnv } = env; + + const child = spawn(process.execPath, [runnerPath], { + env: { ...cleanProcessEnv, ...cleanUserEnv }, + stdio: ['pipe', 'pipe', 'pipe', 'ipc'], + }); + + let stdout = ''; + let stderr = ''; + let settled = false; + + child.stdout?.on('data', (d) => (stdout += String(d))); + child.stderr?.on('data', (d) => (stderr += String(d))); + + child.on( + 'message', + ( + msg: + | { + ok: true; + result?: unknown; + stdout?: string; + stderr?: string; + } + | { + ok: false; + error: string; + stack?: string; + stdout?: string; + stderr?: string; + }, + ) => { + if (settled) return; + settled = true; + resolve({ ...msg, stdout, stderr }); + }, + ); + + child.on('exit', (code) => { + if (settled) return; + settled = true; + if (code === 0) { + resolve({ ok: true, stdout, stderr }); + } else { + resolve({ + ok: false, + error: `Exited with code ${code}`, + stdout, + stderr, + }); + } + }); + + const t = setTimeout(() => { + if (settled) return; + settled = true; + child.kill('SIGKILL'); + resolve({ + ok: false, + error: `Timed out after ${timeoutMs}ms`, + stdout, + stderr, + }); + }, timeoutMs); + + child.send?.({ type: 'run', payload }); + + child.on('close', () => clearTimeout(t)); + }); + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-layer-manager.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-layer-manager.service.ts new file mode 100644 index 00000000000..c8d62147723 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-layer-manager.service.ts @@ -0,0 +1,135 @@ +import { promises as fs } from 'fs'; +import { join } from 'path'; + +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { copyYarnEngineAndBuildDependencies } from 'src/engine/core-modules/application/application-package/utils/copy-yarn-engine-and-build-dependencies'; +import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; +import { + LAYER_BUILD_LOCK_MAX_RETRIES, + LAYER_BUILD_LOCK_RETRY_MS, + LAYER_BUILD_LOCK_TTL_MS, + LAYER_BUILD_READY_SENTINEL, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/constants/local-driver.constant'; +import { getLocalDepsLayerPath } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-deps-layer-path.util'; +import { getLocalSdkLayerPath } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-sdk-layer-path.util'; +import { pathExists } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/path-exists.util'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; +import { type SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service'; +import { type WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; + +type LayerAppContext = { + flatApplication: FlatApplication; + applicationUniversalIdentifier: string; +}; + +export class LocalLayerManagerService { + constructor( + private readonly cacheLockService: CacheLockService, + private readonly logicFunctionResourceService: LogicFunctionResourceService, + private readonly sdkClientArchiveService: SdkClientArchiveService, + private readonly workspaceCacheService: WorkspaceCacheService, + ) {} + + async ensureDepsLayer({ + flatApplication, + applicationUniversalIdentifier, + }: LayerAppContext): Promise { + const depsLayerPath = getLocalDepsLayerPath(flatApplication); + const depsReadySentinelPath = join( + depsLayerPath, + LAYER_BUILD_READY_SENTINEL, + ); + + if (await pathExists(depsReadySentinelPath)) { + return; + } + + const lockKey = `local-driver-deps-layer:${flatApplication.yarnLockChecksum ?? 'default'}`; + + await this.cacheLockService.withLock( + async () => { + if (await pathExists(depsReadySentinelPath)) { + return; + } + + await fs.rm(depsLayerPath, { recursive: true, force: true }); + + await this.logicFunctionResourceService.copyDependenciesInMemory({ + applicationUniversalIdentifier, + workspaceId: flatApplication.workspaceId, + inMemoryFolderPath: depsLayerPath, + }); + await copyYarnEngineAndBuildDependencies(depsLayerPath); + await fs.writeFile(depsReadySentinelPath, ''); + }, + lockKey, + { + ttl: LAYER_BUILD_LOCK_TTL_MS, + ms: LAYER_BUILD_LOCK_RETRY_MS, + maxRetries: LAYER_BUILD_LOCK_MAX_RETRIES, + }, + ); + } + + async ensureSdkLayer({ + flatApplication, + applicationUniversalIdentifier, + }: LayerAppContext): Promise { + const sdkLayerPath = getLocalSdkLayerPath({ + workspaceId: flatApplication.workspaceId, + applicationUniversalIdentifier, + }); + const sdkNodeModulesPath = join(sdkLayerPath, 'node_modules'); + const sdkReadySentinelPath = join(sdkLayerPath, LAYER_BUILD_READY_SENTINEL); + + if ( + (await pathExists(sdkReadySentinelPath)) && + !flatApplication.isSdkLayerStale + ) { + return; + } + + const lockKey = `local-driver-sdk-layer:${flatApplication.workspaceId}:${applicationUniversalIdentifier}`; + + await this.cacheLockService.withLock( + async () => { + const { flatApplicationMaps } = + await this.workspaceCacheService.getOrRecompute( + flatApplication.workspaceId, + ['flatApplicationMaps'], + ); + const freshFlatApplication = + flatApplicationMaps.byId[flatApplication.id]; + const isStale = freshFlatApplication?.isSdkLayerStale ?? true; + + if ((await pathExists(sdkReadySentinelPath)) && !isStale) { + return; + } + + await fs.rm(sdkLayerPath, { recursive: true, force: true }); + + const sdkPackagePath = join(sdkNodeModulesPath, 'twenty-client-sdk'); + + await this.sdkClientArchiveService.downloadAndExtractToPackage({ + workspaceId: flatApplication.workspaceId, + applicationId: flatApplication.id, + applicationUniversalIdentifier, + targetPackagePath: sdkPackagePath, + }); + + await this.sdkClientArchiveService.markSdkLayerFresh({ + applicationId: flatApplication.id, + workspaceId: flatApplication.workspaceId, + }); + + await fs.writeFile(sdkReadySentinelPath, ''); + }, + lockKey, + { + ttl: LAYER_BUILD_LOCK_TTL_MS, + ms: LAYER_BUILD_LOCK_RETRY_MS, + maxRetries: LAYER_BUILD_LOCK_MAX_RETRIES, + }, + ); + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-prebuilt-bundle.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-prebuilt-bundle.service.ts new file mode 100644 index 00000000000..e69cb2cbea0 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/services/local-prebuilt-bundle.service.ts @@ -0,0 +1,113 @@ +import { promises as fs } from 'fs'; +import { dirname, join } from 'path'; + +import { isNonEmptyString } from '@sniptt/guards'; + +import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; +import { + PREBUILT_BUNDLE_FILE_NAME, + PREBUILT_INSTALL_LOCK_MAX_RETRIES, + PREBUILT_INSTALL_LOCK_RETRY_MS, + PREBUILT_INSTALL_LOCK_TTL_MS, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/constants/local-driver.constant'; +import { + getLocalInstalledBundlePath, + getLocalInstalledChecksumPath, + getLocalPrebuiltBundleDir, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-prebuilt-bundle-paths.util'; +import { type LogicFunctionInstallPrebuiltBundleParams } from 'src/engine/core-modules/logic-function/logic-function-drivers/interfaces/logic-function-driver.interface'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; +import { + LogicFunctionException, + LogicFunctionExceptionCode, +} from 'src/engine/metadata-modules/logic-function/logic-function.exception'; +import { type FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type'; + +export class LocalPrebuiltBundleService { + constructor( + private readonly cacheLockService: CacheLockService, + private readonly logicFunctionResourceService: LogicFunctionResourceService, + ) {} + + async installPrebuiltBundle({ + flatLogicFunction, + applicationUniversalIdentifier, + }: LogicFunctionInstallPrebuiltBundleParams): Promise { + if (!isNonEmptyString(flatLogicFunction.checksum)) { + throw new LogicFunctionException( + `Cannot install prebuilt bundle for function '${flatLogicFunction.id}' without a checksum`, + LogicFunctionExceptionCode.LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED, + ); + } + + const checksum = flatLogicFunction.checksum; + + await this.cacheLockService.withLock( + async () => { + const prebuiltDir = getLocalPrebuiltBundleDir(flatLogicFunction); + + await fs.mkdir(prebuiltDir, { recursive: true }); + + await this.logicFunctionResourceService.copyBuiltCodeInMemory({ + workspaceId: flatLogicFunction.workspaceId, + applicationUniversalIdentifier, + builtHandlerPath: flatLogicFunction.builtHandlerPath, + inMemoryDestinationPath: prebuiltDir, + }); + + const downloadedPath = join( + prebuiltDir, + flatLogicFunction.builtHandlerPath, + ); + const targetPath = getLocalInstalledBundlePath(flatLogicFunction); + + if (downloadedPath !== targetPath) { + await fs.mkdir(dirname(targetPath), { recursive: true }); + await fs.rename(downloadedPath, targetPath); + } + + await fs.writeFile( + getLocalInstalledChecksumPath(flatLogicFunction), + checksum, + 'utf8', + ); + }, + `local-install:${flatLogicFunction.id}`, + { + ttl: PREBUILT_INSTALL_LOCK_TTL_MS, + ms: PREBUILT_INSTALL_LOCK_RETRY_MS, + maxRetries: PREBUILT_INSTALL_LOCK_MAX_RETRIES, + }, + ); + } + + async getInstalledBundleChecksum( + flatLogicFunction: FlatLogicFunction, + ): Promise { + try { + const checksum = await fs.readFile( + getLocalInstalledChecksumPath(flatLogicFunction), + 'utf8', + ); + + return checksum.trim() || null; + } catch { + return null; + } + } + + async copyPrebuiltBundleIntoExecutionDir({ + flatLogicFunction, + sourceTemporaryDir, + }: { + flatLogicFunction: FlatLogicFunction; + sourceTemporaryDir: string; + }): Promise { + const installedBundlePath = getLocalInstalledBundlePath(flatLogicFunction); + const localBundlePath = join(sourceTemporaryDir, PREBUILT_BUNDLE_FILE_NAME); + + await fs.copyFile(installedBundlePath, localBundlePath); + + return localBundlePath; + } +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/types/local-driver.type.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/types/local-driver.type.ts new file mode 100644 index 00000000000..9f64c72cd28 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/types/local-driver.type.ts @@ -0,0 +1,11 @@ +import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; +import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service'; +import { type SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service'; +import { type WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; + +export interface LocalDriverOptions { + logicFunctionResourceService: LogicFunctionResourceService; + sdkClientArchiveService: SdkClientArchiveService; + cacheLockService: CacheLockService; + workspaceCacheService: WorkspaceCacheService; +} diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/__tests__/get-local-deps-layer-path.util.spec.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/__tests__/get-local-deps-layer-path.util.spec.ts new file mode 100644 index 00000000000..4d55cd38baf --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/__tests__/get-local-deps-layer-path.util.spec.ts @@ -0,0 +1,19 @@ +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/logic-function-executor-tmpdir-folder'; +import { getLocalDepsLayerPath } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-deps-layer-path.util'; + +describe('getLocalDepsLayerPath', () => { + it('joins the tmpdir folder, the deps segment and the yarnLockChecksum', () => { + expect( + getLocalDepsLayerPath({ yarnLockChecksum: 'abc123' } as FlatApplication), + ).toBe(`${LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER}/deps/abc123`); + }); + + it('falls back to default when yarnLockChecksum is undefined', () => { + expect( + getLocalDepsLayerPath({ + yarnLockChecksum: undefined, + } as unknown as FlatApplication), + ).toBe(`${LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER}/deps/default`); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/__tests__/get-local-sdk-layer-path.util.spec.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/__tests__/get-local-sdk-layer-path.util.spec.ts new file mode 100644 index 00000000000..8421ad67346 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/__tests__/get-local-sdk-layer-path.util.spec.ts @@ -0,0 +1,13 @@ +import { LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/logic-function-executor-tmpdir-folder'; +import { getLocalSdkLayerPath } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-sdk-layer-path.util'; + +describe('getLocalSdkLayerPath', () => { + it('joins the tmpdir folder, the sdk segment and "-"', () => { + expect( + getLocalSdkLayerPath({ + workspaceId: 'ws-1', + applicationUniversalIdentifier: 'app-2', + }), + ).toBe(`${LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER}/sdk/ws-1-app-2`); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-deps-layer-path.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-deps-layer-path.util.ts new file mode 100644 index 00000000000..145613c943d --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-deps-layer-path.util.ts @@ -0,0 +1,12 @@ +import { join } from 'path'; + +import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type'; +import { LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/logic-function-executor-tmpdir-folder'; + +export const getLocalDepsLayerPath = ( + flatApplication: FlatApplication, +): string => { + const checksum = flatApplication.yarnLockChecksum ?? 'default'; + + return join(LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER, 'deps', checksum); +}; diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-prebuilt-bundle-paths.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-prebuilt-bundle-paths.util.ts new file mode 100644 index 00000000000..d1fa6d104b7 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-prebuilt-bundle-paths.util.ts @@ -0,0 +1,26 @@ +import { join } from 'path'; + +import { LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/logic-function-executor-tmpdir-folder'; +import { + PREBUILT_BUNDLE_FILE_NAME, + PREBUILT_CHECKSUM_FILE_NAME, +} from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/constants/local-driver.constant'; +import { type FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type'; + +export const getLocalPrebuiltBundleDir = ( + flatLogicFunction: FlatLogicFunction, +): string => + join(LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER, 'prebuilt', flatLogicFunction.id); + +export const getLocalInstalledBundlePath = ( + flatLogicFunction: FlatLogicFunction, +): string => + join(getLocalPrebuiltBundleDir(flatLogicFunction), PREBUILT_BUNDLE_FILE_NAME); + +export const getLocalInstalledChecksumPath = ( + flatLogicFunction: FlatLogicFunction, +): string => + join( + getLocalPrebuiltBundleDir(flatLogicFunction), + PREBUILT_CHECKSUM_FILE_NAME, + ); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-sdk-layer-path.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-sdk-layer-path.util.ts new file mode 100644 index 00000000000..e197d98d87c --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/get-local-sdk-layer-path.util.ts @@ -0,0 +1,16 @@ +import { join } from 'path'; + +import { LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/logic-function-executor-tmpdir-folder'; + +export const getLocalSdkLayerPath = ({ + workspaceId, + applicationUniversalIdentifier, +}: { + workspaceId: string; + applicationUniversalIdentifier: string; +}): string => + join( + LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER, + 'sdk', + `${workspaceId}-${applicationUniversalIdentifier}`, + ); diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/path-exists.util.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/path-exists.util.ts new file mode 100644 index 00000000000..9b9c3e4fca4 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/local/utils/path-exists.util.ts @@ -0,0 +1,11 @@ +import { promises as fs } from 'fs'; + +export const pathExists = async (targetPath: string): Promise => { + try { + await fs.access(targetPath); + + return true; + } catch { + return false; + } +};