refactor(server): split LambdaDriver and LocalDriver into focused sub-services (#21153)
## Why `LambdaDriver` (1.4k lines) and `LocalDriver` (709 lines) each mixed many unrelated concerns in a single class. ## What changed **This PR only moves code** — every line is byte-for-byte the same as `main` (same logic, same comments, same constants, same lock keys/TTLs, same error mapping). The drivers are now thin orchestrators wiring co-located sub-services + pure utils (each with unit tests). | File | Before | After | |---|---|---| | `lambda.driver.ts` | 1425 | 252 | | `local.driver.ts` | 709 | 231 | | Largest sub-service | — | 371 (`lambda-executor-manager`) | | Unit tests on these paths | 0 | 15 | Lambda split: `lambda-aws-client` / `lambda-tool-functions` / `lambda-layer-manager` / `lambda-executor-manager` + `constants` + `types` + 5 pure utils. Local split: `local-layer-manager` / `local-child-process-runner` / `local-prebuilt-bundle` + `constants` + `types` + 3 pure utils. No behavior change, no signature change, no public-API change.
This commit is contained in:
Charles Bochet
+57
-1229
File diff suppressed because it is too large
Load Diff
+43
@@ -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',
|
||||
),
|
||||
);
|
||||
+145
@@ -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<string> {
|
||||
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<void> {
|
||||
await waitUntilFunctionActiveV2(
|
||||
{ client: await this.getLambdaClient(), maxWaitTime },
|
||||
{ FunctionName: functionName },
|
||||
);
|
||||
}
|
||||
|
||||
async waitFunctionUpdated(
|
||||
functionName: string,
|
||||
maxWaitTime: number = UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS,
|
||||
): Promise<void> {
|
||||
await waitUntilFunctionUpdatedV2(
|
||||
{ client: await this.getLambdaClient(), maxWaitTime },
|
||||
{ FunctionName: functionName },
|
||||
);
|
||||
}
|
||||
|
||||
async getExistingLayerArn(layerName: string): Promise<string | undefined> {
|
||||
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!;
|
||||
}
|
||||
}
|
||||
+369
@@ -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<LambdaDriverOptions, 'lambdaRole'>,
|
||||
private readonly awsClient: LambdaAwsClientService,
|
||||
private readonly layerManager: LambdaLayerManagerService,
|
||||
private readonly cacheLockService: CacheLockService,
|
||||
private readonly logicFunctionResourceService: LogicFunctionResourceService,
|
||||
) {}
|
||||
|
||||
async getLambdaExecutor(
|
||||
flatLogicFunction: FlatLogicFunction,
|
||||
): Promise<GetFunctionCommandOutput | undefined> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string | null> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
+253
@@ -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<LambdaDriverOptions, 'layerBucket'>,
|
||||
private readonly awsClient: LambdaAwsClientService,
|
||||
private readonly toolFunctions: LambdaToolFunctionsService,
|
||||
private readonly logicFunctionResourceService: LogicFunctionResourceService,
|
||||
private readonly sdkClientArchiveService: SdkClientArchiveService,
|
||||
) {}
|
||||
|
||||
async ensureDepsLayer(context: LayerAppContext): Promise<string> {
|
||||
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<string> {
|
||||
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<void> {
|
||||
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<string> {
|
||||
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<void> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
+364
@@ -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<LambdaDriverOptions, 'lambdaRole'>,
|
||||
private readonly awsClient: LambdaAwsClientService,
|
||||
) {}
|
||||
|
||||
async transpile(
|
||||
params: Omit<BuilderLambdaPayload, 'action'>,
|
||||
): Promise<BuilderLambdaResult> {
|
||||
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<YarnInstallLambdaPayload, 'action'>,
|
||||
): Promise<YarnInstallLambdaResult> {
|
||||
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<string> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string> {
|
||||
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<string> {
|
||||
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<string> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
+51
@@ -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<string, string>;
|
||||
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',
|
||||
}
|
||||
+12
@@ -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}$/);
|
||||
});
|
||||
});
|
||||
+34
@@ -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> = {},
|
||||
): FlatApplication =>
|
||||
({
|
||||
yarnLockChecksum: 'abc123',
|
||||
...overrides,
|
||||
}) as FlatApplication;
|
||||
|
||||
describe('getLambdaDepsLayerName', () => {
|
||||
it('returns deps-<checksum> 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');
|
||||
});
|
||||
});
|
||||
+25
@@ -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);
|
||||
});
|
||||
});
|
||||
+84
@@ -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);
|
||||
});
|
||||
});
|
||||
+21
@@ -0,0 +1,21 @@
|
||||
import { createHash } from 'crypto';
|
||||
|
||||
const RESOURCE_NAME_CHECKSUM_LENGTH = 12;
|
||||
|
||||
export const computeHashedLambdaResourceName = ({
|
||||
prefix,
|
||||
contents,
|
||||
}: {
|
||||
prefix: string;
|
||||
contents: ReadonlyArray<string>;
|
||||
}): 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}`;
|
||||
};
|
||||
+9
@@ -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}`;
|
||||
};
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
export const getLambdaSdkLayerName = ({
|
||||
workspaceId,
|
||||
applicationUniversalIdentifier,
|
||||
}: {
|
||||
workspaceId: string;
|
||||
applicationUniversalIdentifier: string;
|
||||
}): string => `sdk-${workspaceId}-${applicationUniversalIdentifier}`;
|
||||
+50
@@ -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,
|
||||
};
|
||||
};
|
||||
+36
@@ -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<Buffer> => {
|
||||
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<void>((resolve, reject) => {
|
||||
archive.on('end', resolve);
|
||||
archive.on('error', reject);
|
||||
void archive.finalize();
|
||||
});
|
||||
|
||||
return Buffer.concat(chunks);
|
||||
};
|
||||
+35
-492
@@ -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<boolean> => {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {}
|
||||
|
||||
// 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<void> {
|
||||
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<void> {
|
||||
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<string | null> {
|
||||
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<string, string>;
|
||||
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<string> {
|
||||
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<void> {
|
||||
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<string | null> {
|
||||
try {
|
||||
const checksum = await fs.readFile(
|
||||
this.getInstalledChecksumPath(flatLogicFunction),
|
||||
'utf8',
|
||||
);
|
||||
|
||||
return checksum.trim() || null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+12
@@ -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;
|
||||
+214
@@ -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<void> {
|
||||
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<string, string>;
|
||||
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));
|
||||
});
|
||||
}
|
||||
}
|
||||
+135
@@ -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<void> {
|
||||
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<void> {
|
||||
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,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
+113
@@ -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<void> {
|
||||
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<string | null> {
|
||||
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<string> {
|
||||
const installedBundlePath = getLocalInstalledBundlePath(flatLogicFunction);
|
||||
const localBundlePath = join(sourceTemporaryDir, PREBUILT_BUNDLE_FILE_NAME);
|
||||
|
||||
await fs.copyFile(installedBundlePath, localBundlePath);
|
||||
|
||||
return localBundlePath;
|
||||
}
|
||||
}
|
||||
+11
@@ -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;
|
||||
}
|
||||
+19
@@ -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`);
|
||||
});
|
||||
});
|
||||
+13
@@ -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 "<workspaceId>-<applicationUniversalIdentifier>"', () => {
|
||||
expect(
|
||||
getLocalSdkLayerPath({
|
||||
workspaceId: 'ws-1',
|
||||
applicationUniversalIdentifier: 'app-2',
|
||||
}),
|
||||
).toBe(`${LOGIC_FUNCTION_EXECUTOR_TMPDIR_FOLDER}/sdk/ws-1-app-2`);
|
||||
});
|
||||
});
|
||||
+12
@@ -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);
|
||||
};
|
||||
+26
@@ -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,
|
||||
);
|
||||
+16
@@ -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}`,
|
||||
);
|
||||
+11
@@ -0,0 +1,11 @@
|
||||
import { promises as fs } from 'fs';
|
||||
|
||||
export const pathExists = async (targetPath: string): Promise<boolean> => {
|
||||
try {
|
||||
await fs.access(targetPath);
|
||||
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
};
|
||||
Reference in New Issue
Block a user