admin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Poc lambda deployment duration (#10340)

Browse Source 
closes https://github.com/twentyhq/core-team-issues/issues/436

## Acheivements
Improve aws lambda deployment time from ~10/15 secs to less that 1 sec

## Done
- migrate with the new code executor architecture for local and lambda
drivers
- support old and new executor architecture to avoid breaking changes
- first run is long, next runs are quick even if code step is updated

## Demo using `lambda` driver
### Before


https://github.com/user-attachments/assets/7f7664b4-658f-4689-8949-ea2c31131252


### After



https://github.com/user-attachments/assets/d486c8e2-f8f8-4dbd-a801-c9901e440b29
This commit is contained in:
martmull
2025-02-20 10:49:57 +01:00
committed by GitHub
parent 3f93aba5fc
commit 927b8c717e
20 changed files with 250 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
packages/twenty-front/src/generated-metadata/gql.ts
View File

@ -34,7 +34,6 @@ const documents = {
"\n mutation DeleteOneRelationMetadataItem($idToDelete: UUID!) {\n deleteOneRelation(input: { id: $idToDelete }) {\n id\n }\n }\n": types.DeleteOneRelationMetadataItemDocument,
"\n query ObjectMetadataItems {\n objects(paging: { first: 1000 }) {\n edges {\n node {\n id\n dataSourceId\n nameSingular\n namePlural\n labelSingular\n labelPlural\n description\n icon\n isCustom\n isRemote\n isActive\n isSystem\n createdAt\n updatedAt\n labelIdentifierFieldMetadataId\n imageIdentifierFieldMetadataId\n shortcut\n isLabelSyncedWithName\n duplicateCriteria\n indexMetadatas(paging: { first: 100 }) {\n edges {\n node {\n id\n createdAt\n updatedAt\n name\n indexWhereClause\n indexType\n isUnique\n indexFieldMetadatas(paging: { first: 100 }) {\n edges {\n node {\n id\n createdAt\n updatedAt\n order\n fieldMetadataId\n }\n }\n }\n }\n }\n }\n fieldsList {\n id\n type\n name\n label\n description\n icon\n isCustom\n isActive\n isSystem\n isNullable\n isUnique\n createdAt\n updatedAt\n defaultValue\n options\n settings\n isLabelSyncedWithName\n relationDefinition {\n relationId\n direction\n sourceObjectMetadata {\n id\n nameSingular\n namePlural\n }\n sourceFieldMetadata {\n id\n name\n }\n targetObjectMetadata {\n id\n nameSingular\n namePlural\n }\n targetFieldMetadata {\n id\n name\n }\n }\n }\n }\n }\n pageInfo {\n hasNextPage\n hasPreviousPage\n startCursor\n endCursor\n }\n }\n }\n": types.ObjectMetadataItemsDocument,
"\n fragment ServerlessFunctionFields on ServerlessFunction {\n id\n name\n description\n runtime\n timeoutSeconds\n syncStatus\n latestVersion\n latestVersionInputSchema\n publishedVersions\n createdAt\n updatedAt\n }\n": types.ServerlessFunctionFieldsFragmentDoc,
"\n \n mutation BuildDraftServerlessFunction(\n $input: BuildDraftServerlessFunctionInput!\n ) {\n buildDraftServerlessFunction(input: $input) {\n ...ServerlessFunctionFields\n }\n }\n": types.BuildDraftServerlessFunctionDocument,
"\n \n mutation CreateOneServerlessFunctionItem(\n $input: CreateServerlessFunctionInput!\n ) {\n createOneServerlessFunction(input: $input) {\n ...ServerlessFunctionFields\n }\n }\n": types.CreateOneServerlessFunctionItemDocument,
"\n \n mutation DeleteOneServerlessFunction($input: ServerlessFunctionIdInput!) {\n deleteOneServerlessFunction(input: $input) {\n ...ServerlessFunctionFields\n }\n }\n": types.DeleteOneServerlessFunctionDocument,
"\n mutation ExecuteOneServerlessFunction(\n $input: ExecuteServerlessFunctionInput!\n ) {\n executeOneServerlessFunction(input: $input) {\n data\n duration\n status\n error\n }\n }\n": types.ExecuteOneServerlessFunctionDocument,
@ -144,10 +143,6 @@ export function graphql(source: "\n query ObjectMetadataItems {\n objects(pa
* The graphql function is used to parse GraphQL queries into a document that can be used by GraphQL clients.
*/
export function graphql(source: "\n fragment ServerlessFunctionFields on ServerlessFunction {\n id\n name\n description\n runtime\n timeoutSeconds\n syncStatus\n latestVersion\n latestVersionInputSchema\n publishedVersions\n createdAt\n updatedAt\n }\n"): (typeof documents)["\n fragment ServerlessFunctionFields on ServerlessFunction {\n id\n name\n description\n runtime\n timeoutSeconds\n syncStatus\n latestVersion\n latestVersionInputSchema\n publishedVersions\n createdAt\n updatedAt\n }\n"];
/**
* The graphql function is used to parse GraphQL queries into a document that can be used by GraphQL clients.
*/
export function graphql(source: "\n \n mutation BuildDraftServerlessFunction(\n $input: BuildDraftServerlessFunctionInput!\n ) {\n buildDraftServerlessFunction(input: $input) {\n ...ServerlessFunctionFields\n }\n }\n"): (typeof documents)["\n \n mutation BuildDraftServerlessFunction(\n $input: BuildDraftServerlessFunctionInput!\n ) {\n buildDraftServerlessFunction(input: $input) {\n ...ServerlessFunctionFields\n }\n }\n"];
/**
* The graphql function is used to parse GraphQL queries into a document that can be used by GraphQL clients.
*/

19
packages/twenty-front/src/generated-metadata/graphql.ts
View File

File diff suppressed because one or more lines are too long

11
packages/twenty-front/src/generated/graphql.tsx
View File

@ -243,11 +243,6 @@ export type BooleanFieldComparison = {
isNot?: InputMaybe<Scalars['Boolean']>;
};
export type BuildDraftServerlessFunctionInput = {
/** The id of the function. */
id: Scalars['ID'];
};
export enum CalendarChannelVisibility {
METADATA = 'METADATA',
SHARE_EVERYTHING = 'SHARE_EVERYTHING'
@ -760,7 +755,6 @@ export type Mutation = {
activateWorkflowVersion: Scalars['Boolean'];
activateWorkspace: Workspace;
authorizeApp: AuthorizeApp;
buildDraftServerlessFunction: ServerlessFunction;
checkCustomDomainValidRecords?: Maybe<CustomDomainValidRecords>;
checkoutSession: BillingSessionOutput;
computeStepOutputSchema: Scalars['JSON'];
@ -837,11 +831,6 @@ export type MutationAuthorizeAppArgs = {
};
export type MutationBuildDraftServerlessFunctionArgs = {
input: BuildDraftServerlessFunctionInput;
};
export type MutationCheckoutSessionArgs = {
plan?: BillingPlanKey;
recurringInterval: SubscriptionInterval;

19
packages/twenty-front/src/modules/serverless-functions/components/ServerlessFunctionExecutionResult.tsx
View File

@ -9,7 +9,6 @@ import {
IconSquareRoundedCheck,
IconSquareRoundedX,
IconLoader,
IconSettings,
AnimatedCircleLoading,
} from 'twenty-ui';
import { ServerlessFunctionExecutionStatus } from '~/generated-metadata/graphql';
@ -41,11 +40,9 @@ const StyledOutput = styled.div<{ accent?: OutputAccent }>`
export const ServerlessFunctionExecutionResult = ({
serverlessFunctionTestData,
isTesting = false,
isBuilding = false,
}: {
serverlessFunctionTestData: ServerlessFunctionTestData;
isTesting?: boolean;
isBuilding?: boolean;
}) => {
const theme = useTheme();
@ -70,23 +67,17 @@ export const ServerlessFunctionExecutionResult = ({
const IdleLeftNode = 'Output';
const PendingLeftNode = (isTesting || isBuilding) && (
const PendingLeftNode = isTesting && (
<StyledOutput>
<AnimatedCircleLoading>
{isTesting ? (
<IconLoader size={theme.icon.size.md} />
) : (
<IconSettings size={theme.icon.size.md} />
)}
<IconLoader size={theme.icon.size.md} />
</AnimatedCircleLoading>
<StyledInfoContainer>
{isTesting ? 'Running function' : 'Building function'}
</StyledInfoContainer>
<StyledInfoContainer>Running function</StyledInfoContainer>
</StyledOutput>
);
const computeLeftNode = () => {
if (isTesting || isBuilding) {
if (isTesting) {
return PendingLeftNode;
}
if (
@ -115,7 +106,7 @@ export const ServerlessFunctionExecutionResult = ({
language={serverlessFunctionTestData.language}
height={serverlessFunctionTestData.height}
options={{ readOnly: true, domReadOnly: true }}
isLoading={isTesting || isBuilding}
isLoading={isTesting}
withHeader
/>
</StyledContainer>

15
packages/twenty-front/src/modules/serverless-functions/hooks/useTestServerlessFunction.ts
View File

@ -1,4 +1,3 @@
import { useBuildDraftServerlessFunction } from '@/settings/serverless-functions/hooks/useBuildDraftServerlessFunction';
import { useExecuteOneServerlessFunction } from '@/settings/serverless-functions/hooks/useExecuteOneServerlessFunction';
import { serverlessFunctionTestDataFamilyState } from '@/workflow/states/serverlessFunctionTestDataFamilyState';
import { useState } from 'react';
@ -14,21 +13,12 @@ export const useTestServerlessFunction = ({
callback?: (testResult: object) => void;
}) => {
const [isTesting, setIsTesting] = useState(false);
const [isBuilding, setIsBuilding] = useState(false);
const { executeOneServerlessFunction } = useExecuteOneServerlessFunction();
const { buildDraftServerlessFunction } = useBuildDraftServerlessFunction();
const [serverlessFunctionTestData, setServerlessFunctionTestData] =
useRecoilState(serverlessFunctionTestDataFamilyState(serverlessFunctionId));
const testServerlessFunction = async (shouldBuild = true) => {
const testServerlessFunction = async () => {
try {
if (shouldBuild) {
setIsBuilding(true);
await buildDraftServerlessFunction({
id: serverlessFunctionId,
});
setIsBuilding(false);
}
setIsTesting(true);
await sleep(200); // Delay artificially to avoid flashing the UI
const result = await executeOneServerlessFunction({
@ -67,11 +57,10 @@ export const useTestServerlessFunction = ({
},
}));
} catch (error) {
setIsBuilding(false);
setIsTesting(false);
throw error;
}
};
return { testServerlessFunction, isTesting, isBuilding };
return { testServerlessFunction, isTesting };
};

13
packages/twenty-front/src/modules/settings/serverless-functions/graphql/mutations/buildDraftServerlessFunction.ts
View File

@ -1,13 +0,0 @@
import { gql } from '@apollo/client';
import { SERVERLESS_FUNCTION_FRAGMENT } from '@/settings/serverless-functions/graphql/fragments/serverlessFunctionFragment';
export const BUILD_DRAFT_SERVERLESS_FUNCTION = gql`
${SERVERLESS_FUNCTION_FRAGMENT}
mutation BuildDraftServerlessFunction(
$input: BuildDraftServerlessFunctionInput!
) {
buildDraftServerlessFunction(input: $input) {
...ServerlessFunctionFields
}
}
`;

29
packages/twenty-front/src/modules/settings/serverless-functions/hooks/useBuildDraftServerlessFunction.ts
View File

@ -1,29 +0,0 @@
import { useApolloMetadataClient } from '@/object-metadata/hooks/useApolloMetadataClient';
import { useMutation } from '@apollo/client';
import { BUILD_DRAFT_SERVERLESS_FUNCTION } from '@/settings/serverless-functions/graphql/mutations/buildDraftServerlessFunction';
import {
BuildDraftServerlessFunctionMutation,
BuildDraftServerlessFunctionMutationVariables,
BuildDraftServerlessFunctionInput,
} from '~/generated-metadata/graphql';
export const useBuildDraftServerlessFunction = () => {
const apolloMetadataClient = useApolloMetadataClient();
const [mutate] = useMutation<
BuildDraftServerlessFunctionMutation,
BuildDraftServerlessFunctionMutationVariables
>(BUILD_DRAFT_SERVERLESS_FUNCTION, {
client: apolloMetadataClient,
});
const buildDraftServerlessFunction = async (
input: BuildDraftServerlessFunctionInput,
) => {
return await mutate({
variables: {
input,
},
});
};
return { buildDraftServerlessFunction };
};

18
packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionFormServerlessFunction.tsx
View File

@ -81,8 +81,6 @@ export const WorkflowEditActionFormServerlessFunction = ({
}: WorkflowEditActionFormServerlessFunctionProps) => {
const theme = useTheme();
const { getIcon } = useIcons();
const [shouldBuildServerlessFunction, setShouldBuildServerlessFunction] =
useState(false);
const serverlessFunctionId = action.settings.input.serverlessFunctionId;
const serverlessFunctionVersion =
action.settings.input.serverlessFunctionVersion;
@ -123,14 +121,12 @@ export const WorkflowEditActionFormServerlessFunction = ({
});
};
const { testServerlessFunction, isTesting, isBuilding } =
useTestServerlessFunction({
serverlessFunctionId,
callback: updateOutputSchemaFromTestResult,
});
const { testServerlessFunction, isTesting } = useTestServerlessFunction({
serverlessFunctionId,
callback: updateOutputSchemaFromTestResult,
});
const handleSave = useDebouncedCallback(async () => {
setShouldBuildServerlessFunction(true);
await updateOneServerlessFunction({
name: formValues.name,
description: formValues.description,
@ -238,8 +234,7 @@ export const WorkflowEditActionFormServerlessFunction = ({
}
if (!isTesting) {
await testServerlessFunction(shouldBuildServerlessFunction);
setShouldBuildServerlessFunction(false);
await testServerlessFunction();
}
};
@ -348,7 +343,6 @@ export const WorkflowEditActionFormServerlessFunction = ({
<InputLabel>Result</InputLabel>
<ServerlessFunctionExecutionResult
serverlessFunctionTestData={serverlessFunctionTestData}
isBuilding={isBuilding}
isTesting={isTesting}
/>
</StyledCodeEditorContainer>
@ -361,7 +355,7 @@ export const WorkflowEditActionFormServerlessFunction = ({
<CmdEnterActionButton
title="Test"
onClick={handleRunFunction}
disabled={isTesting || isBuilding || actionOptions.readonly}
disabled={isTesting || actionOptions.readonly}
/>,
]}
/>

4
packages/twenty-server/nest-cli.json
View File

@ -21,6 +21,10 @@
{
"include": "**/serverless/drivers/layers/engine/**",
"outDir": "dist/assets"
},
{
"include": "**/serverless/drivers/constants/executor/index.mjs",
"outDir": "dist/assets"
}
],
"watchAssets": true

21
packages/twenty-server/src/engine/core-modules/serverless/drivers/constants/executor/index.mjs Normal file
View File

@ -0,0 +1,21 @@
import { promises as fs } from 'fs';
import { v4 } from 'uuid';
export const handler = async (event) => {
const mainPath = `/tmp/${v4()}.mjs`;
try {
const { code, params } = event;
await fs.writeFile(mainPath, code, 'utf8');
process.env = {}
const mainFile = await import(mainPath);
return await mainFile.main(params);
} finally {
await fs.rm(mainPath, { force: true });
}
};

6
packages/twenty-server/src/engine/core-modules/serverless/drivers/interfaces/serverless-driver.interface.ts
View File

@ -16,11 +16,7 @@ export type ServerlessExecuteResult = {
export interface ServerlessDriver {
delete(serverlessFunction: ServerlessFunctionEntity): Promise<void>;
build(
serverlessFunction: ServerlessFunctionEntity,
version: string,
): Promise<void>;
publish(serverlessFunction: ServerlessFunctionEntity): Promise<string>;
build(serverlessFunction: ServerlessFunctionEntity): Promise<void>;
execute(
serverlessFunction: ServerlessFunctionEntity,
payload: object,

239
packages/twenty-server/src/engine/core-modules/serverless/drivers/lambda.driver.ts
View File

@ -1,7 +1,9 @@
import * as fs from 'fs/promises';
import { join } from 'path';
import ts, { transpileModule } from 'typescript';
import {
CreateFunctionCommandInput,
CreateFunctionCommand,
DeleteFunctionCommand,
GetFunctionCommand,
@ -13,18 +15,10 @@ import {
ListLayerVersionsCommandInput,
PublishLayerVersionCommand,
PublishLayerVersionCommandInput,
PublishVersionCommand,
PublishVersionCommandInput,
ResourceNotFoundException,
UpdateFunctionCodeCommand,
UpdateFunctionConfigurationCommand,
UpdateFunctionConfigurationCommandInput,
waitUntilFunctionUpdatedV2,
} from '@aws-sdk/client-lambda';
import { AssumeRoleCommand, STSClient } from '@aws-sdk/client-sts';
import { CreateFunctionCommandInput } from '@aws-sdk/client-lambda/dist-types/commands/CreateFunctionCommand';
import { UpdateFunctionCodeCommandInput } from '@aws-sdk/client-lambda/dist-types/commands/UpdateFunctionCodeCommand';
import dotenv from 'dotenv';
import { isDefined } from 'twenty-shared';
import {
@ -34,10 +28,6 @@ import {
import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service';
import { COMMON_LAYER_NAME } from 'src/engine/core-modules/serverless/drivers/constants/common-layer-name';
import { ENV_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/env-file-name';
import { OUTDIR_FOLDER } from 'src/engine/core-modules/serverless/drivers/constants/outdir-folder';
import { SERVERLESS_TMPDIR_FOLDER } from 'src/engine/core-modules/serverless/drivers/constants/serverless-tmpdir-folder';
import { compileTypescript } from 'src/engine/core-modules/serverless/drivers/utils/compile-typescript';
import { copyAndBuildDependencies } from 'src/engine/core-modules/serverless/drivers/utils/copy-and-build-dependencies';
import { createZipFile } from 'src/engine/core-modules/serverless/drivers/utils/create-zip-file';
import {
@ -54,9 +44,13 @@ import {
ServerlessFunctionException,
ServerlessFunctionExceptionCode,
} from 'src/engine/metadata-modules/serverless-function/serverless-function.exception';
import { copyExecutor } from 'src/engine/core-modules/serverless/drivers/utils/copy-executor';
import { INDEX_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/index-file-name';
import { readFileContent } from 'src/engine/core-modules/file-storage/utils/read-file-content';
const UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS = 60;
const CREDENTIALS_DURATION_IN_SECONDS = 10 * 60 * 60; // 10h
const LAMBDA_EXECUTOR_DESCRIPTION = 'User script executor';
export interface LambdaDriverOptions extends LambdaClientConfig {
fileStorageService: FileStorageService;
@ -127,10 +121,12 @@ export class LambdaDriver implements ServerlessDriver {
}
private async waitFunctionUpdates(
serverlessFunctionId: string,
serverlessFunction: ServerlessFunctionEntity,
maxWaitTime: number = UPDATE_FUNCTION_DURATION_TIMEOUT_IN_SECONDS,
) {
const waitParams = { FunctionName: serverlessFunctionId };
const waitParams = {
FunctionName: serverlessFunction.id,
};
await waitUntilFunctionUpdatedV2(
{ client: await this.getLambdaClient(), maxWaitTime },
@ -192,29 +188,26 @@ export class LambdaDriver implements ServerlessDriver {
return result.LayerVersionArn;
}
private async checkFunctionExists(functionName: string): Promise<boolean> {
private async getLambdaExecutor(
serverlessFunction: ServerlessFunctionEntity,
) {
try {
const getFunctionCommand = new GetFunctionCommand({
FunctionName: functionName,
const getFunctionCommand: GetFunctionCommand = new GetFunctionCommand({
FunctionName: serverlessFunction.id,
});
await (await this.getLambdaClient()).send(getFunctionCommand);
return true;
return await (await this.getLambdaClient()).send(getFunctionCommand);
} catch (error) {
if (error instanceof ResourceNotFoundException) {
return false;
if (!(error instanceof ResourceNotFoundException)) {
throw error;
}
throw error;
}
}
async delete(serverlessFunction: ServerlessFunctionEntity) {
const functionExists = await this.checkFunctionExists(
serverlessFunction.id,
);
const lambdaExecutor = await this.getLambdaExecutor(serverlessFunction);
if (functionExists) {
if (isDefined(lambdaExecutor)) {
const deleteFunctionCommand = new DeleteFunctionCommand({
FunctionName: serverlessFunction.id,
});
@ -223,162 +216,92 @@ export class LambdaDriver implements ServerlessDriver {
}
}
private getInMemoryServerlessFunctionFolderPath = (
serverlessFunction: ServerlessFunctionEntity,
version: string,
) => {
return join(SERVERLESS_TMPDIR_FOLDER, serverlessFunction.id, version);
};
async build(serverlessFunction: ServerlessFunctionEntity) {
const lambdaExecutor = await this.getLambdaExecutor(serverlessFunction);
async build(serverlessFunction: ServerlessFunctionEntity, version: 'draft') {
if (version !== 'draft') {
throw new Error("We can only build 'draft' version with lambda driver");
if (isDefined(lambdaExecutor)) {
if (
lambdaExecutor.Configuration?.Description ===
LAMBDA_EXECUTOR_DESCRIPTION
) {
return;
}
await this.delete(serverlessFunction);
}
const inMemoryServerlessFunctionFolderPath =
this.getInMemoryServerlessFunctionFolderPath(serverlessFunction, version);
const folderPath = getServerlessFolder({
serverlessFunction,
version,
});
await this.fileStorageService.download({
from: { folderPath },
to: { folderPath: inMemoryServerlessFunctionFolderPath },
});
compileTypescript(inMemoryServerlessFunctionFolderPath);
const lambdaZipPath = join(
inMemoryServerlessFunctionFolderPath,
'lambda.zip',
const layerArn = await this.createLayerIfNotExists(
serverlessFunction.layerVersion,
);
await createZipFile(
join(inMemoryServerlessFunctionFolderPath, OUTDIR_FOLDER),
lambdaZipPath,
);
const lambdaBuildDirectoryManager = new LambdaBuildDirectoryManager();
const envFileContent = await fs.readFile(
join(inMemoryServerlessFunctionFolderPath, ENV_FILE_NAME),
);
const { sourceTemporaryDir, lambdaZipPath } =
await lambdaBuildDirectoryManager.init();
const envVariables = dotenv.parse(envFileContent);
await copyExecutor(sourceTemporaryDir);
const functionExists = await this.checkFunctionExists(
serverlessFunction.id,
);
await createZipFile(sourceTemporaryDir, lambdaZipPath);
if (!functionExists) {
const layerArn = await this.createLayerIfNotExists(
serverlessFunction.layerVersion,
);
const params: CreateFunctionCommandInput = {
Code: {
ZipFile: await fs.readFile(lambdaZipPath),
},
FunctionName: serverlessFunction.id,
Handler: 'src/index.main',
Layers: [layerArn],
Environment: {
Variables: envVariables,
},
Role: this.options.lambdaRole,
Runtime: serverlessFunction.runtime,
Description: 'Lambda function to run user script',
Timeout: serverlessFunction.timeoutSeconds,
};
const command = new CreateFunctionCommand(params);
await (await this.getLambdaClient()).send(command);
} else {
const updateCodeParams: UpdateFunctionCodeCommandInput = {
const params: CreateFunctionCommandInput = {
Code: {
ZipFile: await fs.readFile(lambdaZipPath),
FunctionName: serverlessFunction.id,
};
const updateCodeCommand = new UpdateFunctionCodeCommand(updateCodeParams);
await (await this.getLambdaClient()).send(updateCodeCommand);
const updateConfigurationParams: UpdateFunctionConfigurationCommandInput =
{
Environment: {
Variables: envVariables,
},
FunctionName: serverlessFunction.id,
Timeout: serverlessFunction.timeoutSeconds,
};
const updateConfigurationCommand = new UpdateFunctionConfigurationCommand(
updateConfigurationParams,
);
await this.waitFunctionUpdates(serverlessFunction.id);
await (await this.getLambdaClient()).send(updateConfigurationCommand);
}
await this.waitFunctionUpdates(serverlessFunction.id);
}
async publish(serverlessFunction: ServerlessFunctionEntity) {
await this.build(serverlessFunction, 'draft');
const params: PublishVersionCommandInput = {
},
FunctionName: serverlessFunction.id,
Layers: [layerArn],
Handler: 'index.handler',
Role: this.options.lambdaRole,
Runtime: serverlessFunction.runtime,
Description: LAMBDA_EXECUTOR_DESCRIPTION,
Timeout: serverlessFunction.timeoutSeconds,
};
const command = new PublishVersionCommand(params);
const command = new CreateFunctionCommand(params);
const result = await (await this.getLambdaClient()).send(command);
const newVersion = result.Version;
await (await this.getLambdaClient()).send(command);
if (!newVersion) {
throw new Error('New published version is undefined');
}
const draftFolderPath = getServerlessFolder({
serverlessFunction: serverlessFunction,
version: 'draft',
});
const newFolderPath = getServerlessFolder({
serverlessFunction: serverlessFunction,
version: newVersion,
});
await this.fileStorageService.copy({
from: { folderPath: draftFolderPath },
to: { folderPath: newFolderPath },
});
return newVersion;
await lambdaBuildDirectoryManager.clean();
}
async execute(
functionToExecute: ServerlessFunctionEntity,
serverlessFunction: ServerlessFunctionEntity,
payload: object,
version: string,
): Promise<ServerlessExecuteResult> {
const computedVersion =
version === 'latest' ? functionToExecute.latestVersion : version;
const functionName =
computedVersion === 'draft'
? functionToExecute.id
: `${functionToExecute.id}:${computedVersion}`;
if (version === 'draft') {
await this.waitFunctionUpdates(functionToExecute.id);
}
await this.build(serverlessFunction);
await this.waitFunctionUpdates(serverlessFunction);
const startTime = Date.now();
const computedVersion =
version === 'latest' ? serverlessFunction.latestVersion : version;
const folderPath = getServerlessFolder({
serverlessFunction,
version: computedVersion,
});
const tsCodeStream = await this.fileStorageService.read({
folderPath: join(folderPath, 'src'),
filename: INDEX_FILE_NAME,
});
const tsCode = await readFileContent(tsCodeStream);
const compiledCode = transpileModule(tsCode, {
compilerOptions: {
module: ts.ModuleKind.ESNext,
target: ts.ScriptTarget.ES2017,
},
}).outputText;
const executorPayload = {
params: payload,
code: compiledCode,
};
const params: InvokeCommandInput = {
FunctionName: functionName,
Payload: JSON.stringify(payload),
FunctionName: serverlessFunction.id,
Payload: JSON.stringify(executorPayload),
};
const command = new InvokeCommand(params);

276
packages/twenty-server/src/engine/core-modules/serverless/drivers/local.driver.ts
View File

@ -1,27 +1,23 @@
import { fork } from 'child_process';
import { promises as fs } from 'fs';
import { join } from 'path';
import dotenv from 'dotenv';
import ts, { transpileModule } from 'typescript';
import { v4 } from 'uuid';
import {
ServerlessDriver,
ServerlessExecuteError,
ServerlessExecuteResult,
} from 'src/engine/core-modules/serverless/drivers/interfaces/serverless-driver.interface';
import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service';
import { getServerlessFolder } from 'src/engine/core-modules/serverless/utils/serverless-get-folder.utils';
import { ServerlessFunctionExecutionStatus } from 'src/engine/metadata-modules/serverless-function/dtos/serverless-function-execution-result.dto';
import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity';
import { COMMON_LAYER_NAME } from 'src/engine/core-modules/serverless/drivers/constants/common-layer-name';
import { copyAndBuildDependencies } from 'src/engine/core-modules/serverless/drivers/utils/copy-and-build-dependencies';
import { SERVERLESS_TMPDIR_FOLDER } from 'src/engine/core-modules/serverless/drivers/constants/serverless-tmpdir-folder';
import { compileTypescript } from 'src/engine/core-modules/serverless/drivers/utils/compile-typescript';
import { OUTDIR_FOLDER } from 'src/engine/core-modules/serverless/drivers/constants/outdir-folder';
import { ENV_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/env-file-name';
const LISTENER_FILE_NAME = 'listener.js';
import { INDEX_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/index-file-name';
import { readFileContent } from 'src/engine/core-modules/file-storage/utils/read-file-content';
import { ServerlessFunctionExecutionStatus } from 'src/engine/metadata-modules/serverless-function/dtos/serverless-function-execution-result.dto';
export interface LocalDriverOptions {
fileStorageService: FileStorageService;
@ -34,13 +30,6 @@ export class LocalDriver implements ServerlessDriver {
this.fileStorageService = options.fileStorageService;
}
private getInMemoryServerlessFunctionFolderPath = (
serverlessFunction: ServerlessFunctionEntity,
version: string,
) => {
return join(SERVERLESS_TMPDIR_FOLDER, serverlessFunction.id, version);
};
private getInMemoryLayerFolderPath = (version: number) => {
return join(SERVERLESS_TMPDIR_FOLDER, COMMON_LAYER_NAME, `${version}`);
};
@ -58,107 +47,29 @@ export class LocalDriver implements ServerlessDriver {
async delete() {}
async build(serverlessFunction: ServerlessFunctionEntity, version: string) {
const computedVersion =
version === 'latest' ? serverlessFunction.latestVersion : version;
async build(serverlessFunction: ServerlessFunctionEntity) {
await this.createLayerIfNotExists(serverlessFunction.layerVersion);
const inMemoryServerlessFunctionFolderPath =
this.getInMemoryServerlessFunctionFolderPath(
serverlessFunction,
computedVersion,
);
const folderPath = getServerlessFolder({
serverlessFunction,
version,
});
await this.fileStorageService.download({
from: { folderPath },
to: { folderPath: inMemoryServerlessFunctionFolderPath },
});
compileTypescript(inMemoryServerlessFunctionFolderPath);
const envFileContent = await fs.readFile(
join(inMemoryServerlessFunctionFolderPath, ENV_FILE_NAME),
);
const envVariables = dotenv.parse(envFileContent);
const listener = `
const index_1 = require("./src/index");
process.env = ${JSON.stringify(envVariables)}
process.on('message', async (message) => {
const { params } = message;
try {
const result = await index_1.main(params);
process.send(result);
} catch (error) {
process.send({
errorType: error.name,
errorMessage: error.message,
stackTrace: error.stack.split('\\n').filter((line) => line.trim() !== ''),
});
}
});
`;
await fs.writeFile(
join(
inMemoryServerlessFunctionFolderPath,
OUTDIR_FOLDER,
LISTENER_FILE_NAME,
),
listener,
);
try {
await fs.symlink(
join(
this.getInMemoryLayerFolderPath(serverlessFunction.layerVersion),
'node_modules',
),
join(
inMemoryServerlessFunctionFolderPath,
OUTDIR_FOLDER,
'node_modules',
),
'dir',
);
} catch (err) {
if (err.code !== 'EEXIST') {
throw err;
}
}
}
async publish(serverlessFunction: ServerlessFunctionEntity) {
const newVersion = serverlessFunction.latestVersion
? `${parseInt(serverlessFunction.latestVersion, 10) + 1}`
: '1';
private async executeWithTimeout<T>(
fn: () => Promise<T>,
timeoutMs: number,
): Promise<T> {
return new Promise<T>((resolve, reject) => {
const timer = setTimeout(() => {
reject(new Error(`Task timed out after ${timeoutMs / 1_000} seconds`));
}, timeoutMs);
const draftFolderPath = getServerlessFolder({
serverlessFunction: serverlessFunction,
version: 'draft',
fn()
.then((result) => {
clearTimeout(timer);
resolve(result);
})
.catch((err) => {
clearTimeout(timer);
reject(err);
});
});
const newFolderPath = getServerlessFolder({
serverlessFunction: serverlessFunction,
version: newVersion,
});
await this.fileStorageService.copy({
from: { folderPath: draftFolderPath },
to: { folderPath: newFolderPath },
});
await this.build(serverlessFunction, newVersion);
return newVersion;
}
async execute(
@ -166,100 +77,73 @@ export class LocalDriver implements ServerlessDriver {
payload: object,
version: string,
): Promise<ServerlessExecuteResult> {
await this.build(serverlessFunction);
const startTime = Date.now();
const computedVersion =
version === 'latest' ? serverlessFunction.latestVersion : version;
const listenerFile = join(
this.getInMemoryServerlessFunctionFolderPath(
serverlessFunction,
computedVersion,
),
OUTDIR_FOLDER,
LISTENER_FILE_NAME,
const folderPath = getServerlessFolder({
serverlessFunction,
version: computedVersion,
});
const tsCodeStream = await this.fileStorageService.read({
folderPath: join(folderPath, 'src'),
filename: INDEX_FILE_NAME,
});
const tsCode = await readFileContent(tsCodeStream);
const compiledCode = transpileModule(tsCode, {
compilerOptions: {
module: ts.ModuleKind.CommonJS,
target: ts.ScriptTarget.ES2017,
},
}).outputText;
const compiledCodeFolderPath = join(
SERVERLESS_TMPDIR_FOLDER,
`compiled-code-${v4()}`,
);
const compiledCodeFilePath = join(compiledCodeFolderPath, 'main.js');
await fs.mkdir(compiledCodeFolderPath, { recursive: true });
await fs.writeFile(compiledCodeFilePath, compiledCode, 'utf8');
try {
return await new Promise((resolve, reject) => {
const child = fork(listenerFile, { silent: true });
await fs.symlink(
join(
this.getInMemoryLayerFolderPath(serverlessFunction.layerVersion),
'node_modules',
),
join(compiledCodeFolderPath, 'node_modules'),
'dir',
);
} catch (err) {
if (err.code !== 'EEXIST') {
throw err;
}
}
const timeoutMs = serverlessFunction.timeoutSeconds * 1_000;
try {
const mainFile = await import(compiledCodeFilePath);
const timeoutHandler = setTimeout(() => {
child.kill();
const duration = Date.now() - startTime;
const result = await this.executeWithTimeout<object | null>(
() => mainFile.main(payload),
serverlessFunction.timeoutSeconds * 1_000,
);
reject(new Error(`Task timed out after ${duration / 1_000} seconds`));
}, timeoutMs);
const duration = Date.now() - startTime;
child.on('message', (message: object | ServerlessExecuteError) => {
clearTimeout(timeoutHandler);
const duration = Date.now() - startTime;
if ('errorType' in message) {
resolve({
data: null,
duration,
error: message,
status: ServerlessFunctionExecutionStatus.ERROR,
});
} else {
resolve({
data: message,
duration,
status: ServerlessFunctionExecutionStatus.SUCCESS,
});
}
child.kill();
});
child.stderr?.on('data', (data) => {
clearTimeout(timeoutHandler);
const stackTrace = data
.toString()
.split('\n')
.filter((line: string) => line.trim() !== '');
const errorTrace = stackTrace.filter((line: string) =>
line.includes('Error: '),
)?.[0];
let errorType = 'Unknown';
let errorMessage = '';
if (errorTrace) {
errorType = errorTrace.split(':')[0];
errorMessage = errorTrace.split(': ')[1];
}
const duration = Date.now() - startTime;
resolve({
data: null,
duration,
status: ServerlessFunctionExecutionStatus.ERROR,
error: {
errorType,
errorMessage,
stackTrace: stackTrace,
},
});
child.kill();
});
child.on('error', (error) => {
clearTimeout(timeoutHandler);
reject(error);
child.kill();
});
child.on('exit', (code) => {
clearTimeout(timeoutHandler);
if (code && code !== 0) {
reject(new Error(`Child process exited with code ${code}`));
}
});
child.send({ params: payload });
});
return {
data: result,
duration,
status: ServerlessFunctionExecutionStatus.SUCCESS,
};
} catch (error) {
return {
data: null,
@ -271,6 +155,8 @@ export class LocalDriver implements ServerlessDriver {
},
status: ServerlessFunctionExecutionStatus.ERROR,
};
} finally {
await fs.rm(compiledCodeFolderPath, { recursive: true, force: true });
}
}
}

21
packages/twenty-server/src/engine/core-modules/serverless/drivers/utils/compile-typescript.ts
View File

@ -1,21 +0,0 @@
import { join } from 'path';
import ts, { createProgram } from 'typescript';
import { OUTDIR_FOLDER } from 'src/engine/core-modules/serverless/drivers/constants/outdir-folder';
import { INDEX_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/constants/index-file-name';
export const compileTypescript = (folderPath: string) => {
const options: ts.CompilerOptions = {
module: ts.ModuleKind.CommonJS,
target: ts.ScriptTarget.ES2017,
moduleResolution: ts.ModuleResolutionKind.Node10,
esModuleInterop: true,
resolveJsonModule: true,
allowSyntheticDefaultImports: true,
outDir: join(folderPath, OUTDIR_FOLDER, 'src'),
types: ['node'],
};
createProgram([join(folderPath, 'src', INDEX_FILE_NAME)], options).emit();
};

12
packages/twenty-server/src/engine/core-modules/serverless/drivers/utils/copy-executor.ts Normal file
View File

@ -0,0 +1,12 @@
import { promises as fs } from 'fs';
import { getExecutorFilePath } from 'src/engine/core-modules/serverless/drivers/utils/get-executor-file-path';
export const copyExecutor = async (buildDirectory: string) => {
await fs.mkdir(buildDirectory, {
recursive: true,
});
await fs.cp(getExecutorFilePath(), buildDirectory, {
recursive: true,
});
};

12
packages/twenty-server/src/engine/core-modules/serverless/drivers/utils/get-executor-file-path.ts Normal file
View File

@ -0,0 +1,12 @@
import path from 'path';
import { ASSET_PATH } from 'src/constants/assets-path';
export const getExecutorFilePath = (): string => {
const baseTypescriptProjectPath = path.join(
ASSET_PATH,
`engine/core-modules/serverless/drivers/constants/executor`,
);
return path.resolve(__dirname, baseTypescriptProjectPath);
};

6
packages/twenty-server/src/engine/core-modules/serverless/drivers/utils/get-layer-dependencies-dir-name.ts
View File

@ -1,4 +1,4 @@
import path, { join } from 'path';
import path from 'path';
import { LAST_LAYER_VERSION } from 'src/engine/core-modules/serverless/drivers/layers/last-layer-version';
import { ASSET_PATH } from 'src/constants/assets-path';
@ -8,10 +8,10 @@ export const getLayerDependenciesDirName = (
): string => {
const formattedVersion = version === 'latest' ? LAST_LAYER_VERSION : version;
const baseTypescriptProjectPath = join(
const baseTypescriptProjectPath = path.join(
ASSET_PATH,
`engine/core-modules/serverless/drivers/layers/${formattedVersion}`,
);
return path.resolve(baseTypescriptProjectPath);
return path.resolve(__dirname, baseTypescriptProjectPath);
};

11
packages/twenty-server/src/engine/core-modules/serverless/serverless.service.ts
View File

@ -16,15 +16,8 @@ export class ServerlessService implements ServerlessDriver {
return this.driver.delete(serverlessFunction);
}
async build(
serverlessFunction: ServerlessFunctionEntity,
version: string,
): Promise<void> {
return this.driver.build(serverlessFunction, version);
}
async publish(serverlessFunction: ServerlessFunctionEntity): Promise<string> {
return this.driver.publish(serverlessFunction);
async build(serverlessFunction: ServerlessFunctionEntity): Promise<void> {
return this.driver.build(serverlessFunction);
}
async execute(

19
packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.resolver.ts
View File

@ -24,7 +24,6 @@ import {
} from 'src/engine/metadata-modules/serverless-function/serverless-function.exception';
import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service';
import { serverlessFunctionGraphQLApiExceptionHandler } from 'src/engine/metadata-modules/serverless-function/utils/serverless-function-graphql-api-exception-handler.utils';
import { BuildDraftServerlessFunctionInput } from 'src/engine/metadata-modules/serverless-function/dtos/build-draft-serverless-function.input';
@UseGuards(WorkspaceAuthGuard)
@Resolver()
@ -205,22 +204,4 @@ export class ServerlessFunctionResolver {
serverlessFunctionGraphQLApiExceptionHandler(error);
}
}
@Mutation(() => ServerlessFunctionDTO)
async buildDraftServerlessFunction(
@Args('input') input: BuildDraftServerlessFunctionInput,
@AuthWorkspace() { id: workspaceId }: Workspace,
) {
try {
await this.checkFeatureFlag(workspaceId);
const { id } = input;
return await this.serverlessFunctionService.buildDraftServerlessFunction(
id,
workspaceId,
);
} catch (error) {
serverlessFunctionGraphQLApiExceptionHandler(error);
}
}
}

66
packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts
View File

@ -137,21 +137,12 @@ export class ServerlessFunctionService {
workspaceId,
});
if (
version === 'draft' &&
functionToExecute.syncStatus !== ServerlessFunctionSyncStatus.READY
) {
await this.buildDraftServerlessFunction(
functionToExecute.id,
workspaceId,
);
}
const resultServerlessFunction = await this.serverlessService.execute(
functionToExecute,
payload,
version,
);
const eventInput = {
action: 'serverlessFunction.executed',
payload: {
@ -200,9 +191,23 @@ export class ServerlessFunctionService {
}
}
const newVersion = await this.serverlessService.publish(
existingServerlessFunction,
);
const newVersion = existingServerlessFunction.latestVersion
? `${parseInt(existingServerlessFunction.latestVersion, 10) + 1}`
: '1';
const draftFolderPath = getServerlessFolder({
serverlessFunction: existingServerlessFunction,
version: 'draft',
});
const newFolderPath = getServerlessFolder({
serverlessFunction: existingServerlessFunction,
version: newVersion,
});
await this.fileStorageService.copy({
from: { folderPath: draftFolderPath },
to: { folderPath: newFolderPath },
});
const newPublishedVersions = [
...existingServerlessFunction.publishedVersions,
@ -264,7 +269,6 @@ export class ServerlessFunctionService {
{
name: serverlessFunctionInput.name,
description: serverlessFunctionInput.description,
syncStatus: ServerlessFunctionSyncStatus.NOT_READY,
timeoutSeconds: serverlessFunctionInput.timeoutSeconds,
},
);
@ -343,6 +347,8 @@ export class ServerlessFunctionService {
});
}
await this.serverlessService.build(createdServerlessFunction);
return this.serverlessFunctionRepository.findOneBy({
id: createdServerlessFunction.id,
});
@ -380,10 +386,6 @@ export class ServerlessFunctionService {
}),
},
});
await this.serverlessFunctionRepository.update(serverlessFunction.id, {
syncStatus: ServerlessFunctionSyncStatus.NOT_READY,
});
}
private async throttleExecution(workspaceId: string) {
@ -400,32 +402,4 @@ export class ServerlessFunctionService {
);
}
}
async buildDraftServerlessFunction(id: string, workspaceId: string) {
const functionToBuild = await this.findOneOrFail({
id,
workspaceId,
});
if (functionToBuild.syncStatus === ServerlessFunctionSyncStatus.READY) {
return functionToBuild;
}
if (functionToBuild.syncStatus === ServerlessFunctionSyncStatus.BUILDING) {
throw new ServerlessFunctionException(
'This function is currently building. Please try later',
ServerlessFunctionExceptionCode.SERVERLESS_FUNCTION_BUILDING,
);
}
await this.serverlessFunctionRepository.update(functionToBuild.id, {
syncStatus: ServerlessFunctionSyncStatus.BUILDING,
});
await this.serverlessService.build(functionToBuild, 'draft');
await this.serverlessFunctionRepository.update(functionToBuild.id, {
syncStatus: ServerlessFunctionSyncStatus.READY,
});
return functionToBuild;
}
}