feat(apps): add enqueueJob to run a logic function on the workers (#23527)
Closes twentyhq/core-team-issues#2742 A logic function run is capped by its own `timeoutSeconds` (900s max), so anything that can't finish in one run — a full re-sync, a per-record fan-out, a rate-limited third-party API — had no way to continue. This adds a way to hand that work to the workers. ## What it looks like for an app author ```ts import { enqueueJob } from 'twenty-sdk/logic-function'; await enqueueJob({ logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33', payload: { cursor: nextCursor }, retryLimit: 3, priority: 2, delayMs: 60_000, }); ``` The target runs in its own process with its own timeout budget. The classic shape is a function that enqueues *itself* with the next cursor until there is nothing left. ## Changes **twenty-shared** — `EnqueueJobInput` / `EnqueueJobOptions` / `EnqueueJobResult` in `application`. **twenty-server** — new `application-job` module under `core-modules/application`, following the `application-key-value` pattern: - `enqueueJob` mutation on the metadata API, `@AuthApplication`-scoped - the lookup is scoped to `applicationId` + `workspaceId` — that's the authorization boundary, an app can only enqueue its own logic functions, anything else is `LOGIC_FUNCTION_NOT_FOUND` - pushes a `LogicFunctionTriggerJob` onto the existing `logicFunctionQueue`, so the enqueued run goes through the same executor (and the same execution throttling) as every other trigger - the queued run inherits the caller's `userId`/`userWorkspaceId`, so its app access token carries the same permissions as the function that queued it **Job options** are range-checked via `ResolverValidationPipe`, since the values come from application code and an unbounded delay or retry count would let an app pin work in the shared queue: | Option | Default | Range | |--------|---------|-------| | `retryLimit` | `0` | `0`–`10` | | `priority` | queue default | `1`–`10` (lower first) | | `delayMs` | `0` | `0`–7 days | `retryLimit` defaults to `0` rather than inheriting the server-route path's `3`: retries re-run the whole handler, so opting in should be the author's explicit choice. **twenty-sdk** — `enqueueJob` in `twenty-sdk/logic-function`, same shape as `runAgent`/`kv`. **Docs** — new "Background Jobs" page under Extend → Apps → Logic, plus nav and overview entries. **Generated** — regenerated `twenty-front/src/generated-metadata` and `twenty-client-sdk/src/metadata/generated` for the new mutation. ## Tests - `application-job.service.spec.ts` — 5 unit tests: job options mapping, defaults, acting-user propagation, application-scoped lookup, not-found - `enqueue-job.integration-spec.ts` — 5 integration tests: rejects a non-`APPLICATION_ACCESS` token, enqueues a function the app owns, rejects a function owned by another application, rejects an unknown identifier, rejects out-of-range options All green locally, along with `typecheck` for `twenty-server`/`twenty-sdk` and oxlint/oxfmt on the touched files. ## Notes for review - The target is addressed by `universalIdentifier`, matching `runAgent({ agentUniversalIdentifier })` and `ServerRouteDispatchResult.targetLogicFunctionUniversalIdentifier`. Addressing by `name` would be friendlier, but logic function names aren't validated for uniqueness within an app — happy to add it as a convenience if you'd rather. - `enqueueJob` returns as soon as the job is accepted; it can't return the target's result, since the queue driver's `add` returns void. Documented, with a pointer to the KV store for handing results back. --- _Generated by [Claude Code](https://claude.ai/code/session_01QrYvGonS3HMdeuMAVjs5hR)_ <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/twentyhq/twenty/pull/23527?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a> <!-- End of auto-generated description by cubic. --> --------- Co-authored-by: github-actions <github-actions@twenty.com>
This commit is contained in:
@@ -2989,6 +2989,11 @@ type WorkspaceAiStats {
|
||||
toolsCount: Int!
|
||||
}
|
||||
|
||||
type EnqueueJobResult {
|
||||
enqueued: Boolean!
|
||||
logicFunctionUniversalIdentifier: String!
|
||||
}
|
||||
|
||||
type AppKeyValue {
|
||||
key: String!
|
||||
value: JSON
|
||||
@@ -3543,6 +3548,7 @@ type Mutation {
|
||||
updateCalendarChannel(input: UpdateCalendarChannelInput!): CalendarChannel!
|
||||
setAppKeyValue(input: SetAppKeyValueInput!): AppKeyValue!
|
||||
deleteAppKeyValue(key: String!, scope: AppKeyValueScope = WORKSPACE): Boolean!
|
||||
enqueueJob(input: EnqueueJobInput!): EnqueueJobResult!
|
||||
createChatThread: AgentChatThread!
|
||||
sendChatMessage(threadId: UUID!, text: String!, messageId: UUID!, browsingContext: JSON, modelId: String, fileAttachments: [FileAttachmentInput!]): SendChatMessageResult!
|
||||
retryChatMessage(threadId: UUID!, modelId: String): SendChatMessageResult!
|
||||
@@ -4691,6 +4697,13 @@ input SetAppKeyValueInput {
|
||||
scope: AppKeyValueScope = WORKSPACE
|
||||
}
|
||||
|
||||
input EnqueueJobInput {
|
||||
logicFunctionUniversalIdentifier: String!
|
||||
payload: JSON
|
||||
retryLimit: Int
|
||||
delayMs: Int
|
||||
}
|
||||
|
||||
input FileAttachmentInput {
|
||||
id: UUID!
|
||||
filename: String!
|
||||
|
||||
@@ -2682,6 +2682,12 @@ export interface WorkspaceAiStats {
|
||||
__typename: 'WorkspaceAiStats'
|
||||
}
|
||||
|
||||
export interface EnqueueJobResult {
|
||||
enqueued: Scalars['Boolean']
|
||||
logicFunctionUniversalIdentifier: Scalars['String']
|
||||
__typename: 'EnqueueJobResult'
|
||||
}
|
||||
|
||||
export interface AppKeyValue {
|
||||
key: Scalars['String']
|
||||
value?: Scalars['JSON']
|
||||
@@ -3062,6 +3068,7 @@ export interface Mutation {
|
||||
updateCalendarChannel: CalendarChannel
|
||||
setAppKeyValue: AppKeyValue
|
||||
deleteAppKeyValue: Scalars['Boolean']
|
||||
enqueueJob: EnqueueJobResult
|
||||
createChatThread: AgentChatThread
|
||||
sendChatMessage: SendChatMessageResult
|
||||
retryChatMessage: SendChatMessageResult
|
||||
@@ -5964,6 +5971,13 @@ export interface WorkspaceAiStatsGenqlSelection{
|
||||
__scalar?: boolean | number
|
||||
}
|
||||
|
||||
export interface EnqueueJobResultGenqlSelection{
|
||||
enqueued?: boolean | number
|
||||
logicFunctionUniversalIdentifier?: boolean | number
|
||||
__typename?: boolean | number
|
||||
__scalar?: boolean | number
|
||||
}
|
||||
|
||||
export interface AppKeyValueGenqlSelection{
|
||||
key?: boolean | number
|
||||
value?: boolean | number
|
||||
@@ -6376,6 +6390,7 @@ export interface MutationGenqlSelection{
|
||||
updateCalendarChannel?: (CalendarChannelGenqlSelection & { __args: {input: UpdateCalendarChannelInput} })
|
||||
setAppKeyValue?: (AppKeyValueGenqlSelection & { __args: {input: SetAppKeyValueInput} })
|
||||
deleteAppKeyValue?: { __args: {key: Scalars['String'], scope?: (AppKeyValueScope | null)} }
|
||||
enqueueJob?: (EnqueueJobResultGenqlSelection & { __args: {input: EnqueueJobInput} })
|
||||
createChatThread?: AgentChatThreadGenqlSelection
|
||||
sendChatMessage?: (SendChatMessageResultGenqlSelection & { __args: {threadId: Scalars['UUID'], text: Scalars['String'], messageId: Scalars['UUID'], browsingContext?: (Scalars['JSON'] | null), modelId?: (Scalars['String'] | null), fileAttachments?: (FileAttachmentInput[] | null)} })
|
||||
retryChatMessage?: (SendChatMessageResultGenqlSelection & { __args: {threadId: Scalars['UUID'], modelId?: (Scalars['String'] | null)} })
|
||||
@@ -6805,6 +6820,8 @@ export interface UpdateCalendarChannelInputUpdates {visibility?: (CalendarChanne
|
||||
|
||||
export interface SetAppKeyValueInput {key: Scalars['String'],value?: (Scalars['JSON'] | null),scope?: (AppKeyValueScope | null)}
|
||||
|
||||
export interface EnqueueJobInput {logicFunctionUniversalIdentifier: Scalars['String'],payload?: (Scalars['JSON'] | null),retryLimit?: (Scalars['Int'] | null),delayMs?: (Scalars['Int'] | null)}
|
||||
|
||||
export interface FileAttachmentInput {id: Scalars['UUID'],filename: Scalars['String']}
|
||||
|
||||
export interface AgentChatQuestionAnswerInput {questionIndex: Scalars['Int'],selectedOptionIndices: Scalars['Int'][],freeText?: (Scalars['String'] | null)}
|
||||
@@ -8891,6 +8908,14 @@ export interface LogicFunctionLogsInput {applicationId?: (Scalars['UUID'] | null
|
||||
|
||||
|
||||
|
||||
const EnqueueJobResult_possibleTypes: string[] = ['EnqueueJobResult']
|
||||
export const isEnqueueJobResult = (obj?: { __typename?: any } | null): obj is EnqueueJobResult => {
|
||||
if (!obj?.__typename) throw new Error('__typename is missing in "isEnqueueJobResult"')
|
||||
return EnqueueJobResult_possibleTypes.includes(obj.__typename)
|
||||
}
|
||||
|
||||
|
||||
|
||||
const AppKeyValue_possibleTypes: string[] = ['AppKeyValue']
|
||||
export const isAppKeyValue = (obj?: { __typename?: any } | null): obj is AppKeyValue => {
|
||||
if (!obj?.__typename) throw new Error('__typename is missing in "isAppKeyValue"')
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,165 @@
|
||||
---
|
||||
title: Background Jobs
|
||||
description: Hand long or rate-limited work to the Twenty workers by enqueuing another logic function run instead of doing everything inline.
|
||||
icon: "layer-group"
|
||||
---
|
||||
|
||||
A logic function run is capped by its `timeoutSeconds` (900 seconds maximum). Anything that can't finish in that window — a full re-sync, a per-record fan-out, a third-party API that rate-limits you — has to be split into smaller runs.
|
||||
|
||||
`enqueueJob` does exactly that: it asks the Twenty workers to run one of your app's logic functions later, in its own process, with its own timeout budget. The caller returns immediately.
|
||||
|
||||
```text
|
||||
┌─────────────────┐ enqueueJob(...) ┌──────────────┐ ┌────────────────────┐
|
||||
│ Logic function │ ─────────────────▶ │ Job queue │──▶│ Logic function │
|
||||
│ (returns now) │ │ (workers) │ │ (fresh run/timeout)│
|
||||
└─────────────────┘ └──────────────┘ └────────────────────┘
|
||||
```
|
||||
|
||||
## Enqueue a run
|
||||
|
||||
Import `enqueueJob` from `twenty-sdk/logic-function` and point it at the `universalIdentifier` of the logic function you want to run.
|
||||
|
||||
```ts src/logic-functions/sync-all-contacts.ts
|
||||
import { enqueueJob } from 'twenty-sdk/logic-function';
|
||||
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33',
|
||||
payload: { page: 1 },
|
||||
});
|
||||
```
|
||||
|
||||
The target function receives `payload` as its handler argument, exactly like any other trigger. It must belong to the **same application** as the caller — enqueuing another app's function is rejected with `Logic function not found`.
|
||||
|
||||
<Note>
|
||||
`enqueueJob` returns as soon as the job is accepted, not when it has run. It does not return the target's result — have the target write what it produces to the [key-value store](/developers/extend/apps/logic/key-value-store) or to a workspace record if you need to read it back.
|
||||
</Note>
|
||||
|
||||
## Job options
|
||||
|
||||
| Option | Default | Range | What it does |
|
||||
|--------|---------|-------|--------------|
|
||||
| `retryLimit` | `0` | `0`–`10` | Extra attempts if the run throws. Only raise this for handlers that are safe to run twice. |
|
||||
| `delayMs` | `0` | `0`–`604800000` (7 days) | Wait this long before the run becomes eligible. |
|
||||
|
||||
```ts
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33',
|
||||
payload: { page: 1 },
|
||||
retryLimit: 3,
|
||||
delayMs: 60_000,
|
||||
});
|
||||
```
|
||||
|
||||
<Note>
|
||||
**Priority is not configurable yet.** Enqueued jobs always run at the lowest priority, so platform work is never delayed behind application jobs. Control over priority is coming soon.
|
||||
</Note>
|
||||
|
||||
The queued run inherits the acting user of the function that enqueued it, so it acts with the same permissions.
|
||||
|
||||
## Use it: page through a long sync
|
||||
|
||||
The classic shape is a function that enqueues *itself* with the next cursor. Each run does one page of work well inside its own timeout, and the chain stops when there is nothing left.
|
||||
|
||||
```ts src/logic-functions/sync-contacts-page.ts
|
||||
import { defineLogicFunction } from 'twenty-sdk/define';
|
||||
import { enqueueJob } from 'twenty-sdk/logic-function';
|
||||
|
||||
const SYNC_CONTACTS_PAGE = '9f1c3d7e-51b8-4a29-8f0d-7c4e2a6b1d33';
|
||||
|
||||
const handler = async (params: { cursor?: string }) => {
|
||||
const { contacts, nextCursor } = await fetchContactsPage(params.cursor);
|
||||
|
||||
await importContacts(contacts);
|
||||
|
||||
if (nextCursor) {
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: SYNC_CONTACTS_PAGE,
|
||||
payload: { cursor: nextCursor },
|
||||
delayMs: 2_000,
|
||||
});
|
||||
}
|
||||
|
||||
return { imported: contacts.length, done: !nextCursor };
|
||||
};
|
||||
|
||||
export default defineLogicFunction({
|
||||
universalIdentifier: SYNC_CONTACTS_PAGE,
|
||||
name: 'sync-contacts-page',
|
||||
timeoutSeconds: 120,
|
||||
handler,
|
||||
});
|
||||
```
|
||||
|
||||
## Fan out per record
|
||||
|
||||
When the work is naturally per-item, enqueue one job per item and let the workers process them in parallel instead of looping inline.
|
||||
|
||||
```ts
|
||||
const companies = await listCompaniesToEnrich();
|
||||
|
||||
await Promise.all(
|
||||
companies.map((company) =>
|
||||
enqueueJob({
|
||||
logicFunctionUniversalIdentifier: ENRICH_COMPANY,
|
||||
payload: { companyId: company.id },
|
||||
retryLimit: 2,
|
||||
}),
|
||||
),
|
||||
);
|
||||
```
|
||||
|
||||
## Good practice for long-running work
|
||||
|
||||
Two rules cover almost every long job: **recurse instead of looping**, and **process a bounded chunk per run**.
|
||||
|
||||
A run that tries to do everything is the failure mode — it hits the timeout, and with a retry it starts the whole thing again from zero. Instead, size one chunk so it comfortably finishes inside `timeoutSeconds`, persist your position, and enqueue the next run.
|
||||
|
||||
```ts src/logic-functions/enrich-companies-batch.ts
|
||||
import { defineLogicFunction } from 'twenty-sdk/define';
|
||||
import { enqueueJob, kv } from 'twenty-sdk/logic-function';
|
||||
|
||||
const ENRICH_COMPANIES_BATCH = '3f9d1c02-8a44-4f0e-b1d7-9c2e5a7b4f10';
|
||||
const CHUNK_SIZE = 50;
|
||||
|
||||
const handler = async (params: { offset?: number }) => {
|
||||
const offset = params.offset ?? 0;
|
||||
const companies = await listCompaniesToEnrich({
|
||||
offset,
|
||||
limit: CHUNK_SIZE,
|
||||
});
|
||||
|
||||
for (const company of companies) {
|
||||
await enrichCompany(company);
|
||||
}
|
||||
|
||||
await kv.set('enrich:progress', { offset: offset + companies.length });
|
||||
|
||||
if (companies.length === CHUNK_SIZE) {
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: ENRICH_COMPANIES_BATCH,
|
||||
payload: { offset: offset + CHUNK_SIZE },
|
||||
});
|
||||
}
|
||||
|
||||
return { processed: companies.length, done: companies.length < CHUNK_SIZE };
|
||||
};
|
||||
|
||||
export default defineLogicFunction({
|
||||
universalIdentifier: ENRICH_COMPANIES_BATCH,
|
||||
name: 'enrich-companies-batch',
|
||||
timeoutSeconds: 300,
|
||||
handler,
|
||||
});
|
||||
```
|
||||
|
||||
What makes this hold up:
|
||||
|
||||
- **Size the chunk from the slowest item, not the average.** `CHUNK_SIZE × worst-case item time` has to fit in `timeoutSeconds` with room to spare, or the tail of a chunk is lost when the run is cut off.
|
||||
- **Make the terminating condition explicit.** Recurse only while a full chunk came back. A chain that stops on "no results" alone will keep going forever if the source ever returns a short page mid-way.
|
||||
- **Persist progress before enqueuing the next run,** so a failed link restarts from the last completed chunk instead of the beginning.
|
||||
- **Keep each chunk idempotent.** Reprocessing one chunk after a retry must not double-write — key writes on the record or external id you are processing.
|
||||
- **Prefer a chunked chain over one giant fan-out** when the work hits a rate-limited third party: a chain with `delayMs` paces itself, whereas thousands of jobs enqueued at once all become eligible immediately.
|
||||
|
||||
<Warning>
|
||||
Retries re-run the whole handler. Keep enqueued handlers idempotent before setting `retryLimit` above `0`.
|
||||
</Warning>
|
||||
@@ -37,6 +37,9 @@ A Twenty app's **logic layer** is the code that *runs* — server-side TypeScrip
|
||||
<Card title="Key-Value Store" icon="database" href="/developers/extend/apps/logic/key-value-store">
|
||||
Persist state between logic function runs — caches, cursors, and cross-workspace claims.
|
||||
</Card>
|
||||
<Card title="Background Jobs" icon="layer-group" href="/developers/extend/apps/logic/background-jobs">
|
||||
Enqueue a logic function run on the workers to get past the per-run timeout.
|
||||
</Card>
|
||||
</CardGroup>
|
||||
|
||||
## Trigger types at a glance
|
||||
|
||||
@@ -436,6 +436,7 @@
|
||||
"developers/extend/apps/logic/overview",
|
||||
"developers/extend/apps/logic/logic-functions",
|
||||
"developers/extend/apps/logic/key-value-store",
|
||||
"developers/extend/apps/logic/background-jobs",
|
||||
"developers/extend/apps/logic/skills-and-agents",
|
||||
"developers/extend/apps/logic/connections"
|
||||
]
|
||||
|
||||
@@ -437,6 +437,7 @@
|
||||
"developers/extend/apps/logic/overview",
|
||||
"developers/extend/apps/logic/logic-functions",
|
||||
"developers/extend/apps/logic/key-value-store",
|
||||
"developers/extend/apps/logic/background-jobs",
|
||||
"developers/extend/apps/logic/skills-and-agents",
|
||||
"developers/extend/apps/logic/connections"
|
||||
]
|
||||
|
||||
@@ -1680,6 +1680,19 @@ export enum EngineComponentKey {
|
||||
VIEW_PREVIOUS_AI_CHATS = 'VIEW_PREVIOUS_AI_CHATS'
|
||||
}
|
||||
|
||||
export type EnqueueJobInput = {
|
||||
delayMs?: InputMaybe<Scalars['Int']['input']>;
|
||||
logicFunctionUniversalIdentifier: Scalars['String']['input'];
|
||||
payload?: InputMaybe<Scalars['JSON']['input']>;
|
||||
retryLimit?: InputMaybe<Scalars['Int']['input']>;
|
||||
};
|
||||
|
||||
export type EnqueueJobResult = {
|
||||
__typename?: 'EnqueueJobResult';
|
||||
enqueued: Scalars['Boolean']['output'];
|
||||
logicFunctionUniversalIdentifier: Scalars['String']['output'];
|
||||
};
|
||||
|
||||
export type EnterpriseLicenseInfoDto = {
|
||||
__typename?: 'EnterpriseLicenseInfoDTO';
|
||||
expiresAt?: Maybe<Scalars['DateTime']['output']>;
|
||||
@@ -2673,6 +2686,7 @@ export type Mutation = {
|
||||
editSSOIdentityProvider: EditSso;
|
||||
emailPasswordResetLink: EmailPasswordResetLink;
|
||||
endSubscriptionTrialPeriod: BillingEndTrialPeriod;
|
||||
enqueueJob: EnqueueJobResult;
|
||||
enrichWorkspaceCompany: WorkspaceCompanyEnrichmentResult;
|
||||
evaluateAgentTurn: AgentTurnEvaluation;
|
||||
executeOneLogicFunction: LogicFunctionExecutionResult;
|
||||
@@ -3328,6 +3342,11 @@ export type MutationEmailPasswordResetLinkArgs = {
|
||||
};
|
||||
|
||||
|
||||
export type MutationEnqueueJobArgs = {
|
||||
input: EnqueueJobInput;
|
||||
};
|
||||
|
||||
|
||||
export type MutationEvaluateAgentTurnArgs = {
|
||||
turnId: Scalars['UUID']['input'];
|
||||
};
|
||||
|
||||
@@ -58,6 +58,13 @@ export type { AppConnection } from '@/sdk/logic-function/connections/types/app-c
|
||||
export { runAgent } from '@/sdk/logic-function/agents/run-agent';
|
||||
export type { RunAgentInput, RunAgentResult } from 'twenty-shared/application';
|
||||
|
||||
export { enqueueJob } from '@/sdk/logic-function/jobs/enqueue-job';
|
||||
export type {
|
||||
EnqueueJobInput,
|
||||
EnqueueJobOptions,
|
||||
EnqueueJobResult,
|
||||
} from 'twenty-shared/application';
|
||||
|
||||
export { kv } from '@/sdk/logic-function/key-value/kv';
|
||||
export type { AppKeyValue, AppKeyValueScope } from 'twenty-shared/application';
|
||||
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
import {
|
||||
afterEach,
|
||||
beforeEach,
|
||||
describe,
|
||||
expect,
|
||||
it,
|
||||
vi,
|
||||
type MockInstance,
|
||||
} from 'vitest';
|
||||
|
||||
const TARGET_UNIVERSAL_IDENTIFIER = '5a2f4d2a-1a1e-4c66-8a54-1f0a2b3c4d5e';
|
||||
|
||||
const importEnqueueJob = async () => {
|
||||
const module = await import('@/sdk/logic-function/jobs/enqueue-job');
|
||||
|
||||
return module.enqueueJob;
|
||||
};
|
||||
|
||||
const graphqlResponse = (data: unknown) =>
|
||||
new Response(JSON.stringify({ data }), {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
});
|
||||
|
||||
const successResponse = () =>
|
||||
graphqlResponse({
|
||||
enqueueJob: {
|
||||
enqueued: true,
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
},
|
||||
});
|
||||
|
||||
describe('enqueueJob', () => {
|
||||
let fetchSpy: MockInstance<typeof fetch>;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules();
|
||||
process.env.TWENTY_API_URL = 'https://api.test';
|
||||
process.env.TWENTY_APP_ACCESS_TOKEN = 'app-token';
|
||||
fetchSpy = vi.spyOn(globalThis, 'fetch');
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
delete process.env.TWENTY_API_URL;
|
||||
delete process.env.TWENTY_APP_ACCESS_TOKEN;
|
||||
fetchSpy.mockRestore();
|
||||
});
|
||||
|
||||
it('calls the enqueueJob mutation on the metadata API and returns its result', async () => {
|
||||
fetchSpy.mockResolvedValue(successResponse());
|
||||
|
||||
const enqueueJob = await importEnqueueJob();
|
||||
|
||||
const result = await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
payload: { batchIndex: 2 },
|
||||
retryLimit: 3,
|
||||
delayMs: 1000,
|
||||
});
|
||||
|
||||
expect(result).toEqual({
|
||||
enqueued: true,
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
});
|
||||
|
||||
const [url, requestInit] = fetchSpy.mock.calls[0];
|
||||
|
||||
expect(url).toBe('https://api.test/metadata');
|
||||
|
||||
const sentBody = JSON.parse(requestInit?.body as string);
|
||||
|
||||
expect(sentBody.query).toContain(
|
||||
'enqueueJob(input:$v1){enqueued,logicFunctionUniversalIdentifier}',
|
||||
);
|
||||
expect(Object.values(sentBody.variables)).toEqual([
|
||||
{
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
payload: { batchIndex: 2 },
|
||||
retryLimit: 3,
|
||||
delayMs: 1000,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('sends the app access token as a bearer credential', async () => {
|
||||
fetchSpy.mockResolvedValue(successResponse());
|
||||
|
||||
const enqueueJob = await importEnqueueJob();
|
||||
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
});
|
||||
|
||||
const [, requestInit] = fetchSpy.mock.calls[0];
|
||||
const headers = new Headers(requestInit?.headers);
|
||||
|
||||
expect(headers.get('authorization')).toBe('Bearer app-token');
|
||||
});
|
||||
|
||||
it('surfaces GraphQL errors as a rejection', async () => {
|
||||
fetchSpy.mockResolvedValue(
|
||||
new Response(
|
||||
JSON.stringify({ errors: [{ message: 'Logic function not found' }] }),
|
||||
{ status: 200, headers: { 'Content-Type': 'application/json' } },
|
||||
),
|
||||
);
|
||||
|
||||
const enqueueJob = await importEnqueueJob();
|
||||
|
||||
await expect(
|
||||
enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
}),
|
||||
).rejects.toThrow(/Logic function not found/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,21 @@
|
||||
import { MetadataApiClient } from 'twenty-client-sdk/metadata';
|
||||
import {
|
||||
type EnqueueJobInput,
|
||||
type EnqueueJobResult,
|
||||
} from 'twenty-shared/application';
|
||||
|
||||
export const enqueueJob = async (
|
||||
input: EnqueueJobInput,
|
||||
): Promise<EnqueueJobResult> => {
|
||||
const client = new MetadataApiClient();
|
||||
|
||||
const { enqueueJob: result } = await client.mutation({
|
||||
enqueueJob: {
|
||||
__args: { input },
|
||||
enqueued: true,
|
||||
logicFunctionUniversalIdentifier: true,
|
||||
},
|
||||
});
|
||||
|
||||
return result;
|
||||
};
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
|
||||
import { ApplicationJobResolver } from 'src/engine/core-modules/application/application-job/application-job.resolver';
|
||||
import { ApplicationJobService } from 'src/engine/core-modules/application/application-job/services/application-job.service';
|
||||
import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
|
||||
|
||||
@Module({
|
||||
imports: [WorkspaceCacheModule],
|
||||
providers: [ApplicationJobService, ApplicationJobResolver],
|
||||
exports: [ApplicationJobService],
|
||||
})
|
||||
export class ApplicationJobModule {}
|
||||
+46
@@ -0,0 +1,46 @@
|
||||
import { UseFilters, UseGuards, UsePipes } from '@nestjs/common';
|
||||
import { Args, Mutation } from '@nestjs/graphql';
|
||||
|
||||
import { type EnqueueJobResult } from 'twenty-shared/application';
|
||||
|
||||
import { MetadataResolver } from 'src/engine/api/graphql/graphql-config/decorators/metadata-resolver.decorator';
|
||||
import { ApplicationExceptionFilter } from 'src/engine/core-modules/application/application-exception-filter';
|
||||
import { EnqueueJobResultDTO } from 'src/engine/core-modules/application/application-job/dtos/enqueue-job-result.dto';
|
||||
import { EnqueueJobInputDTO } from 'src/engine/core-modules/application/application-job/dtos/enqueue-job.input';
|
||||
import { ApplicationJobService } from 'src/engine/core-modules/application/application-job/services/application-job.service';
|
||||
import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type';
|
||||
import { type AuthContextUser } from 'src/engine/core-modules/auth/types/auth-context.type';
|
||||
import { ResolverValidationPipe } from 'src/engine/core-modules/graphql/pipes/resolver-validation.pipe';
|
||||
import { type FlatWorkspace } from 'src/engine/core-modules/workspace/types/flat-workspace.type';
|
||||
import { AuthApplication } from 'src/engine/decorators/auth/auth-application.decorator';
|
||||
import { AuthUserWorkspaceId } from 'src/engine/decorators/auth/auth-user-workspace-id.decorator';
|
||||
import { AuthUser } from 'src/engine/decorators/auth/auth-user.decorator';
|
||||
import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator';
|
||||
import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
|
||||
import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
|
||||
|
||||
@UseGuards(WorkspaceAuthGuard, NoPermissionGuard)
|
||||
@UseFilters(ApplicationExceptionFilter)
|
||||
@UsePipes(ResolverValidationPipe)
|
||||
@MetadataResolver()
|
||||
export class ApplicationJobResolver {
|
||||
constructor(private readonly applicationJobService: ApplicationJobService) {}
|
||||
|
||||
@Mutation(() => EnqueueJobResultDTO)
|
||||
async enqueueJob(
|
||||
@AuthApplication() application: FlatApplication,
|
||||
@AuthWorkspace() workspace: FlatWorkspace,
|
||||
@AuthUser({ allowUndefined: true }) user: AuthContextUser | undefined,
|
||||
@AuthUserWorkspaceId({ allowUndefined: true })
|
||||
userWorkspaceId: string | undefined,
|
||||
@Args('input') input: EnqueueJobInputDTO,
|
||||
): Promise<EnqueueJobResult> {
|
||||
return this.applicationJobService.enqueueJob({
|
||||
applicationId: application.id,
|
||||
workspaceId: workspace.id,
|
||||
userId: user?.id ?? null,
|
||||
userWorkspaceId: userWorkspaceId ?? null,
|
||||
input,
|
||||
});
|
||||
}
|
||||
}
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
export const ENQUEUE_JOB_MIN_RETRY_LIMIT = 0;
|
||||
export const ENQUEUE_JOB_MAX_RETRY_LIMIT = 10;
|
||||
export const ENQUEUE_JOB_DEFAULT_RETRY_LIMIT = 0;
|
||||
|
||||
export const ENQUEUE_JOB_PRIORITY = 10;
|
||||
|
||||
export const ENQUEUE_JOB_MIN_DELAY_MS = 0;
|
||||
export const ENQUEUE_JOB_MAX_DELAY_MS = 7 * 24 * 60 * 60 * 1000;
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
import { Field, ObjectType } from '@nestjs/graphql';
|
||||
|
||||
import { type EnqueueJobResult } from 'twenty-shared/application';
|
||||
|
||||
@ObjectType('EnqueueJobResult')
|
||||
export class EnqueueJobResultDTO implements EnqueueJobResult {
|
||||
@Field()
|
||||
enqueued: boolean;
|
||||
|
||||
@Field()
|
||||
logicFunctionUniversalIdentifier: string;
|
||||
}
|
||||
+47
@@ -0,0 +1,47 @@
|
||||
import { Field, InputType, Int } from '@nestjs/graphql';
|
||||
|
||||
import {
|
||||
IsInt,
|
||||
IsNotEmpty,
|
||||
IsObject,
|
||||
IsOptional,
|
||||
IsUUID,
|
||||
Max,
|
||||
Min,
|
||||
} from 'class-validator';
|
||||
import GraphQLJSON from 'graphql-type-json';
|
||||
import { type EnqueueJobInput } from 'twenty-shared/application';
|
||||
|
||||
import {
|
||||
ENQUEUE_JOB_MAX_DELAY_MS,
|
||||
ENQUEUE_JOB_MAX_RETRY_LIMIT,
|
||||
ENQUEUE_JOB_MIN_DELAY_MS,
|
||||
ENQUEUE_JOB_MIN_RETRY_LIMIT,
|
||||
} from 'src/engine/core-modules/application/application-job/constants/enqueue-job.constant';
|
||||
|
||||
@InputType('EnqueueJobInput')
|
||||
export class EnqueueJobInputDTO implements EnqueueJobInput {
|
||||
@IsUUID()
|
||||
@IsNotEmpty()
|
||||
@Field()
|
||||
logicFunctionUniversalIdentifier: string;
|
||||
|
||||
@IsObject()
|
||||
@IsOptional()
|
||||
@Field(() => GraphQLJSON, { nullable: true })
|
||||
payload?: Record<string, unknown>;
|
||||
|
||||
@IsInt()
|
||||
@Min(ENQUEUE_JOB_MIN_RETRY_LIMIT)
|
||||
@Max(ENQUEUE_JOB_MAX_RETRY_LIMIT)
|
||||
@IsOptional()
|
||||
@Field(() => Int, { nullable: true })
|
||||
retryLimit?: number;
|
||||
|
||||
@IsInt()
|
||||
@Min(ENQUEUE_JOB_MIN_DELAY_MS)
|
||||
@Max(ENQUEUE_JOB_MAX_DELAY_MS)
|
||||
@IsOptional()
|
||||
@Field(() => Int, { nullable: true })
|
||||
delayMs?: number;
|
||||
}
|
||||
+200
@@ -0,0 +1,200 @@
|
||||
import { ENQUEUE_JOB_PRIORITY } from 'src/engine/core-modules/application/application-job/constants/enqueue-job.constant';
|
||||
import { type EnqueueJobInputDTO } from 'src/engine/core-modules/application/application-job/dtos/enqueue-job.input';
|
||||
import { ApplicationJobService } from 'src/engine/core-modules/application/application-job/services/application-job.service';
|
||||
import {
|
||||
ApplicationException,
|
||||
ApplicationExceptionCode,
|
||||
} from 'src/engine/core-modules/application/application.exception';
|
||||
import { LogicFunctionTriggerJob } from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job';
|
||||
import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
|
||||
import { type WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
|
||||
|
||||
const APPLICATION_ID = 'c2a9e9d0-1f42-4f0e-9a0e-6d2e4b2a1f01';
|
||||
const WORKSPACE_ID = '20202020-1c25-4d02-bf25-6aeccf7ea419';
|
||||
const TARGET_UNIVERSAL_IDENTIFIER = '5a2f4d2a-1a1e-4c66-8a54-1f0a2b3c4d5e';
|
||||
const TARGET_LOGIC_FUNCTION_ID = 'ab6a2e5c-8c1f-4d0a-9bd1-52c1f5a9e100';
|
||||
|
||||
const buildFlatLogicFunction = (
|
||||
overrides: Partial<{
|
||||
applicationId: string;
|
||||
deletedAt: Date | null;
|
||||
}> = {},
|
||||
) => ({
|
||||
id: TARGET_LOGIC_FUNCTION_ID,
|
||||
universalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
applicationId: APPLICATION_ID,
|
||||
deletedAt: null,
|
||||
...overrides,
|
||||
});
|
||||
|
||||
describe('ApplicationJobService', () => {
|
||||
let service: ApplicationJobService;
|
||||
let workspaceCacheService: jest.Mocked<
|
||||
Pick<WorkspaceCacheService, 'getOrRecompute'>
|
||||
>;
|
||||
let messageQueueService: jest.Mocked<Pick<MessageQueueService, 'add'>>;
|
||||
|
||||
const setCachedLogicFunctions = (
|
||||
flatLogicFunctions: ReturnType<typeof buildFlatLogicFunction>[],
|
||||
) => {
|
||||
workspaceCacheService.getOrRecompute.mockResolvedValue({
|
||||
flatLogicFunctionMaps: {
|
||||
byUniversalIdentifier: Object.fromEntries(
|
||||
flatLogicFunctions.map((flatLogicFunction) => [
|
||||
flatLogicFunction.universalIdentifier,
|
||||
flatLogicFunction,
|
||||
]),
|
||||
),
|
||||
},
|
||||
// oxlint-disable-next-line typescript/no-explicit-any
|
||||
} as any);
|
||||
};
|
||||
|
||||
const enqueueJob = (
|
||||
input: EnqueueJobInputDTO,
|
||||
overrides: { userId?: string | null; userWorkspaceId?: string | null } = {},
|
||||
) =>
|
||||
service.enqueueJob({
|
||||
applicationId: APPLICATION_ID,
|
||||
workspaceId: WORKSPACE_ID,
|
||||
userId: overrides.userId ?? null,
|
||||
userWorkspaceId: overrides.userWorkspaceId ?? null,
|
||||
input,
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
|
||||
workspaceCacheService = { getOrRecompute: jest.fn() };
|
||||
setCachedLogicFunctions([buildFlatLogicFunction()]);
|
||||
messageQueueService = { add: jest.fn().mockResolvedValue(undefined) };
|
||||
|
||||
service = new ApplicationJobService(
|
||||
workspaceCacheService as unknown as WorkspaceCacheService,
|
||||
messageQueueService as unknown as MessageQueueService,
|
||||
);
|
||||
});
|
||||
|
||||
it('should enqueue a logic function trigger job with the provided job options', async () => {
|
||||
const result = await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
payload: { batchIndex: 2 },
|
||||
retryLimit: 3,
|
||||
delayMs: 1000,
|
||||
});
|
||||
|
||||
expect(result).toEqual({
|
||||
enqueued: true,
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
});
|
||||
|
||||
expect(messageQueueService.add).toHaveBeenCalledWith(
|
||||
LogicFunctionTriggerJob.name,
|
||||
{
|
||||
logicFunctionId: TARGET_LOGIC_FUNCTION_ID,
|
||||
workspaceId: WORKSPACE_ID,
|
||||
payload: { batchIndex: 2 },
|
||||
},
|
||||
{ retryLimit: 3, priority: ENQUEUE_JOB_PRIORITY, delay: 1000 },
|
||||
);
|
||||
});
|
||||
|
||||
it('should always enqueue at the lowest priority so platform jobs go first', async () => {
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
});
|
||||
|
||||
expect(messageQueueService.add).toHaveBeenCalledWith(
|
||||
LogicFunctionTriggerJob.name,
|
||||
expect.anything(),
|
||||
expect.objectContaining({ priority: ENQUEUE_JOB_PRIORITY }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should default to no retry and omit unset queue options', async () => {
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
});
|
||||
|
||||
expect(messageQueueService.add).toHaveBeenCalledWith(
|
||||
LogicFunctionTriggerJob.name,
|
||||
{
|
||||
logicFunctionId: TARGET_LOGIC_FUNCTION_ID,
|
||||
workspaceId: WORKSPACE_ID,
|
||||
payload: {},
|
||||
},
|
||||
{ retryLimit: 0, priority: ENQUEUE_JOB_PRIORITY },
|
||||
);
|
||||
});
|
||||
|
||||
it('should forward the acting user so the queued run keeps the caller permissions', async () => {
|
||||
await enqueueJob(
|
||||
{ logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER },
|
||||
{ userId: 'user-1', userWorkspaceId: 'user-workspace-1' },
|
||||
);
|
||||
|
||||
expect(messageQueueService.add).toHaveBeenCalledWith(
|
||||
LogicFunctionTriggerJob.name,
|
||||
expect.objectContaining({
|
||||
userId: 'user-1',
|
||||
userWorkspaceId: 'user-workspace-1',
|
||||
}),
|
||||
expect.anything(),
|
||||
);
|
||||
});
|
||||
|
||||
it('should resolve the target from the workspace cache instead of the database', async () => {
|
||||
await enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
});
|
||||
|
||||
expect(workspaceCacheService.getOrRecompute).toHaveBeenCalledWith(
|
||||
WORKSPACE_ID,
|
||||
['flatLogicFunctionMaps'],
|
||||
);
|
||||
});
|
||||
|
||||
it('should throw LOGIC_FUNCTION_NOT_FOUND when the function belongs to another application', async () => {
|
||||
setCachedLogicFunctions([
|
||||
buildFlatLogicFunction({ applicationId: 'another-application-id' }),
|
||||
]);
|
||||
|
||||
await expect(
|
||||
enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
}),
|
||||
).rejects.toMatchObject({
|
||||
code: ApplicationExceptionCode.LOGIC_FUNCTION_NOT_FOUND,
|
||||
});
|
||||
|
||||
expect(messageQueueService.add).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should throw LOGIC_FUNCTION_NOT_FOUND when the function is soft deleted', async () => {
|
||||
setCachedLogicFunctions([
|
||||
buildFlatLogicFunction({ deletedAt: new Date('2026-01-01') }),
|
||||
]);
|
||||
|
||||
await expect(
|
||||
enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
}),
|
||||
).rejects.toMatchObject({
|
||||
code: ApplicationExceptionCode.LOGIC_FUNCTION_NOT_FOUND,
|
||||
});
|
||||
|
||||
expect(messageQueueService.add).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should throw LOGIC_FUNCTION_NOT_FOUND when the function is unknown', async () => {
|
||||
setCachedLogicFunctions([]);
|
||||
|
||||
await expect(
|
||||
enqueueJob({
|
||||
logicFunctionUniversalIdentifier: TARGET_UNIVERSAL_IDENTIFIER,
|
||||
}),
|
||||
).rejects.toThrow(ApplicationException);
|
||||
|
||||
expect(messageQueueService.add).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
+87
@@ -0,0 +1,87 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { type EnqueueJobResult } from 'twenty-shared/application';
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import {
|
||||
ENQUEUE_JOB_DEFAULT_RETRY_LIMIT,
|
||||
ENQUEUE_JOB_PRIORITY,
|
||||
} from 'src/engine/core-modules/application/application-job/constants/enqueue-job.constant';
|
||||
import { type EnqueueJobInputDTO } from 'src/engine/core-modules/application/application-job/dtos/enqueue-job.input';
|
||||
import {
|
||||
ApplicationException,
|
||||
ApplicationExceptionCode,
|
||||
} from 'src/engine/core-modules/application/application.exception';
|
||||
import {
|
||||
LogicFunctionTriggerJob,
|
||||
type LogicFunctionTriggerJobData,
|
||||
} from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job';
|
||||
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
|
||||
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
|
||||
import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
|
||||
import { findFlatEntityByUniversalIdentifier } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-universal-identifier.util';
|
||||
import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
|
||||
|
||||
@Injectable()
|
||||
export class ApplicationJobService {
|
||||
constructor(
|
||||
private readonly workspaceCacheService: WorkspaceCacheService,
|
||||
@InjectMessageQueue(MessageQueue.logicFunctionQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
) {}
|
||||
|
||||
async enqueueJob({
|
||||
applicationId,
|
||||
workspaceId,
|
||||
userId,
|
||||
userWorkspaceId,
|
||||
input,
|
||||
}: {
|
||||
applicationId: string;
|
||||
workspaceId: string;
|
||||
userId: string | null;
|
||||
userWorkspaceId: string | null;
|
||||
input: EnqueueJobInputDTO;
|
||||
}): Promise<EnqueueJobResult> {
|
||||
const { logicFunctionUniversalIdentifier } = input;
|
||||
|
||||
const { flatLogicFunctionMaps } =
|
||||
await this.workspaceCacheService.getOrRecompute(workspaceId, [
|
||||
'flatLogicFunctionMaps',
|
||||
]);
|
||||
|
||||
const flatLogicFunction = findFlatEntityByUniversalIdentifier({
|
||||
flatEntityMaps: flatLogicFunctionMaps,
|
||||
universalIdentifier: logicFunctionUniversalIdentifier,
|
||||
});
|
||||
|
||||
if (
|
||||
!isDefined(flatLogicFunction) ||
|
||||
isDefined(flatLogicFunction.deletedAt) ||
|
||||
flatLogicFunction.applicationId !== applicationId
|
||||
) {
|
||||
throw new ApplicationException(
|
||||
`Logic function ${logicFunctionUniversalIdentifier} not found in this application`,
|
||||
ApplicationExceptionCode.LOGIC_FUNCTION_NOT_FOUND,
|
||||
);
|
||||
}
|
||||
|
||||
await this.messageQueueService.add<LogicFunctionTriggerJobData>(
|
||||
LogicFunctionTriggerJob.name,
|
||||
{
|
||||
logicFunctionId: flatLogicFunction.id,
|
||||
workspaceId,
|
||||
payload: input.payload ?? {},
|
||||
...(isDefined(userId) ? { userId } : {}),
|
||||
...(isDefined(userWorkspaceId) ? { userWorkspaceId } : {}),
|
||||
},
|
||||
{
|
||||
retryLimit: input.retryLimit ?? ENQUEUE_JOB_DEFAULT_RETRY_LIMIT,
|
||||
priority: ENQUEUE_JOB_PRIORITY,
|
||||
...(isDefined(input.delayMs) ? { delay: input.delayMs } : {}),
|
||||
},
|
||||
);
|
||||
|
||||
return { enqueued: true, logicFunctionUniversalIdentifier };
|
||||
}
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import { AiChatModule } from 'src/engine/metadata-modules/ai/ai-chat/ai-chat.mod
|
||||
import { AiGenerateTextModule } from 'src/engine/metadata-modules/ai/ai-generate-text/ai-generate-text.module';
|
||||
import { AiWorkspaceStatsModule } from 'src/engine/metadata-modules/ai/ai-workspace-stats/ai-workspace-stats.module';
|
||||
import { ApplicationConnectionsModule } from 'src/engine/core-modules/application/connection-provider/connections/application-connections.module';
|
||||
import { ApplicationJobModule } from 'src/engine/core-modules/application/application-job/application-job.module';
|
||||
import { ApplicationKeyValueModule } from 'src/engine/core-modules/application/application-key-value/application-key-value.module';
|
||||
import { CalendarChannelMetadataModule } from 'src/engine/metadata-modules/calendar-channel/calendar-channel-metadata.module';
|
||||
import { ConnectedAccountMetadataModule } from 'src/engine/metadata-modules/connected-account/connected-account-metadata.module';
|
||||
@@ -50,6 +51,7 @@ import { WorkspaceMetadataVersionModule } from 'src/engine/metadata-modules/work
|
||||
AiGenerateTextModule,
|
||||
AiWorkspaceStatsModule,
|
||||
ApplicationConnectionsModule,
|
||||
ApplicationJobModule,
|
||||
ApplicationKeyValueModule,
|
||||
MinimalMetadataModule,
|
||||
ViewModule,
|
||||
|
||||
+199
@@ -0,0 +1,199 @@
|
||||
import { existsSync, mkdirSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
|
||||
import gql from 'graphql-tag';
|
||||
import { findManyApplications } from 'test/integration/graphql/utils/find-many-applications.util';
|
||||
import { generateApplicationToken } from 'test/integration/metadata/suites/application/utils/generate-application-token.util';
|
||||
import { createOneLogicFunction } from 'test/integration/metadata/suites/logic-function/utils/create-logic-function.util';
|
||||
import { deleteLogicFunction } from 'test/integration/metadata/suites/logic-function/utils/delete-logic-function.util';
|
||||
import { executeLogicFunction } from 'test/integration/metadata/suites/logic-function/utils/execute-logic-function.util';
|
||||
import { updateLogicFunctionSource } from 'test/integration/metadata/suites/logic-function/utils/update-logic-function-source.util';
|
||||
import { makeMetadataAPIRequest } from 'test/integration/metadata/suites/utils/make-metadata-api-request.util';
|
||||
import { expectEventually } from 'test/integration/utils/expect-eventually.util';
|
||||
import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
|
||||
import { WORKSPACE_CUSTOM_APPLICATION_NAME } from 'src/engine/core-modules/application/constants/workspace-custom-application.constant';
|
||||
import { TWENTY_STANDARD_APPLICATION } from 'src/engine/workspace-manager/twenty-standard-application/constants/twenty-standard-applications';
|
||||
|
||||
const MARKER_DIRECTORY = join(tmpdir(), `enqueue-job-${uuidv4()}`);
|
||||
|
||||
const TARGET_SOURCE_CODE = `import { writeFileSync } from 'node:fs';
|
||||
|
||||
export const main = async (params: { markerPath: string }): Promise<object> => {
|
||||
writeFileSync(params.markerPath, 'ran', 'utf-8');
|
||||
|
||||
return { ok: true };
|
||||
};`;
|
||||
|
||||
const ENQUEUE_JOB = gql`
|
||||
mutation EnqueueJob($input: EnqueueJobInput!) {
|
||||
enqueueJob(input: $input) {
|
||||
enqueued
|
||||
logicFunctionUniversalIdentifier
|
||||
}
|
||||
}
|
||||
`;
|
||||
|
||||
describe('enqueueJob (e2e)', () => {
|
||||
let customApplicationToken: string;
|
||||
let standardApplicationToken: string;
|
||||
let logicFunctionId: string;
|
||||
let logicFunctionUniversalIdentifier: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
mkdirSync(MARKER_DIRECTORY, { recursive: true });
|
||||
|
||||
const { data } = await findManyApplications({ expectToFail: false });
|
||||
|
||||
const customApplication = data.findManyApplications.find(
|
||||
(application) => application.name === WORKSPACE_CUSTOM_APPLICATION_NAME,
|
||||
);
|
||||
const standardApplication = data.findManyApplications.find(
|
||||
(application) =>
|
||||
application.universalIdentifier ===
|
||||
TWENTY_STANDARD_APPLICATION.universalIdentifier,
|
||||
);
|
||||
|
||||
expect(customApplication).toBeDefined();
|
||||
expect(standardApplication).toBeDefined();
|
||||
|
||||
const [{ data: customTokenData }, { data: standardTokenData }] =
|
||||
await Promise.all([
|
||||
generateApplicationToken({
|
||||
applicationId: customApplication!.id,
|
||||
expectToFail: false,
|
||||
}),
|
||||
generateApplicationToken({
|
||||
applicationId: standardApplication!.id,
|
||||
expectToFail: false,
|
||||
}),
|
||||
]);
|
||||
|
||||
customApplicationToken =
|
||||
customTokenData.generateApplicationToken.applicationAccessToken.token;
|
||||
standardApplicationToken =
|
||||
standardTokenData.generateApplicationToken.applicationAccessToken.token;
|
||||
|
||||
const { data: createData } = await createOneLogicFunction({
|
||||
input: { name: `enqueue-job-target-${uuidv4()}` },
|
||||
gqlFields: 'id universalIdentifier',
|
||||
expectToFail: false,
|
||||
});
|
||||
|
||||
expect(createData.createOneLogicFunction.universalIdentifier).toBeDefined();
|
||||
|
||||
logicFunctionId = createData.createOneLogicFunction.id;
|
||||
logicFunctionUniversalIdentifier =
|
||||
createData.createOneLogicFunction.universalIdentifier!;
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await deleteLogicFunction({
|
||||
input: { id: logicFunctionId },
|
||||
expectToFail: false,
|
||||
});
|
||||
|
||||
rmSync(MARKER_DIRECTORY, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('rejects requests that do not carry an APPLICATION_ACCESS token', async () => {
|
||||
const response = await makeMetadataAPIRequest({
|
||||
query: ENQUEUE_JOB,
|
||||
variables: { input: { logicFunctionUniversalIdentifier } },
|
||||
});
|
||||
|
||||
expect(response.body.errors).toBeDefined();
|
||||
expect(response.body.errors[0].message).toContain('APPLICATION_ACCESS');
|
||||
});
|
||||
|
||||
it('enqueues a logic function owned by the calling application and the worker runs it', async () => {
|
||||
const markerPath = join(MARKER_DIRECTORY, 'enqueued.txt');
|
||||
|
||||
await updateLogicFunctionSource({
|
||||
input: {
|
||||
id: logicFunctionId,
|
||||
update: { sourceHandlerCode: TARGET_SOURCE_CODE },
|
||||
},
|
||||
expectToFail: false,
|
||||
});
|
||||
|
||||
const { data: buildData } = await executeLogicFunction({
|
||||
input: {
|
||||
id: logicFunctionId,
|
||||
payload: { markerPath: join(MARKER_DIRECTORY, 'build.txt') },
|
||||
},
|
||||
expectToFail: false,
|
||||
});
|
||||
|
||||
expect(buildData.executeOneLogicFunction.error).toBeNull();
|
||||
|
||||
const response = await makeMetadataAPIRequest(
|
||||
{
|
||||
query: ENQUEUE_JOB,
|
||||
variables: {
|
||||
input: { logicFunctionUniversalIdentifier, payload: { markerPath } },
|
||||
},
|
||||
},
|
||||
customApplicationToken,
|
||||
);
|
||||
|
||||
expect(response.body.errors).toBeUndefined();
|
||||
expect(response.body.data.enqueueJob).toEqual({
|
||||
enqueued: true,
|
||||
logicFunctionUniversalIdentifier,
|
||||
});
|
||||
|
||||
await waitForAllJobsToFinish();
|
||||
|
||||
await expectEventually(() => {
|
||||
expect(existsSync(markerPath)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects a logic function that belongs to another application', async () => {
|
||||
const response = await makeMetadataAPIRequest(
|
||||
{
|
||||
query: ENQUEUE_JOB,
|
||||
variables: {
|
||||
input: { logicFunctionUniversalIdentifier },
|
||||
},
|
||||
},
|
||||
standardApplicationToken,
|
||||
);
|
||||
|
||||
expect(response.body.errors).toBeDefined();
|
||||
expect(response.body.errors[0].message).toContain('not found');
|
||||
});
|
||||
|
||||
it('rejects an unknown logic function', async () => {
|
||||
const response = await makeMetadataAPIRequest(
|
||||
{
|
||||
query: ENQUEUE_JOB,
|
||||
variables: {
|
||||
input: { logicFunctionUniversalIdentifier: uuidv4() },
|
||||
},
|
||||
},
|
||||
customApplicationToken,
|
||||
);
|
||||
|
||||
expect(response.body.errors).toBeDefined();
|
||||
expect(response.body.errors[0].message).toContain('not found');
|
||||
});
|
||||
|
||||
it('rejects job options outside of their allowed range', async () => {
|
||||
const response = await makeMetadataAPIRequest(
|
||||
{
|
||||
query: ENQUEUE_JOB,
|
||||
variables: {
|
||||
input: { logicFunctionUniversalIdentifier, retryLimit: 99 },
|
||||
},
|
||||
},
|
||||
customApplicationToken,
|
||||
);
|
||||
|
||||
expect(response.body.errors).toBeDefined();
|
||||
expect(response.body.errors[0].message).toContain('retryLimit');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,14 @@
|
||||
export type EnqueueJobOptions = {
|
||||
retryLimit?: number;
|
||||
delayMs?: number;
|
||||
};
|
||||
|
||||
export type EnqueueJobInput = EnqueueJobOptions & {
|
||||
logicFunctionUniversalIdentifier: string;
|
||||
payload?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
export type EnqueueJobResult = {
|
||||
enqueued: boolean;
|
||||
logicFunctionUniversalIdentifier: string;
|
||||
};
|
||||
@@ -88,6 +88,11 @@ export { getViewFilterUniversalIdentifier } from './deterministic-identifier/get
|
||||
export { getViewGroupUniversalIdentifier } from './deterministic-identifier/get-view-group-universal-identifier.util';
|
||||
export { getViewSortUniversalIdentifier } from './deterministic-identifier/get-view-sort-universal-identifier.util';
|
||||
export { getViewUniversalIdentifier } from './deterministic-identifier/get-view-universal-identifier.util';
|
||||
export type {
|
||||
EnqueueJobOptions,
|
||||
EnqueueJobInput,
|
||||
EnqueueJobResult,
|
||||
} from './enqueueJobType';
|
||||
export { SyncableEntity } from './enums/syncable-entities.enum';
|
||||
export type {
|
||||
RegularFieldManifest,
|
||||
|
||||
@@ -57,6 +57,8 @@ export const DOCUMENTATION_PATHS = {
|
||||
DEVELOPERS_EXTEND_APPS_LAYOUT_PAGE_LAYOUTS:
|
||||
'/developers/extend/apps/layout/page-layouts',
|
||||
DEVELOPERS_EXTEND_APPS_LAYOUT_VIEWS: '/developers/extend/apps/layout/views',
|
||||
DEVELOPERS_EXTEND_APPS_LOGIC_BACKGROUND_JOBS:
|
||||
'/developers/extend/apps/logic/background-jobs',
|
||||
DEVELOPERS_EXTEND_APPS_LOGIC_CONNECTIONS:
|
||||
'/developers/extend/apps/logic/connections',
|
||||
DEVELOPERS_EXTEND_APPS_LOGIC_KEY_VALUE_STORE:
|
||||
|
||||
Reference in New Issue
Block a user