diff --git a/packages/twenty-apps/public/last-contact/CHANGELOG.md b/packages/twenty-apps/public/last-contact/CHANGELOG.md index ab0ba30729..90c2c6a271 100644 --- a/packages/twenty-apps/public/last-contact/CHANGELOG.md +++ b/packages/twenty-apps/public/last-contact/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## 1.2.3 + +- Rework the last-contact backfill into a single fan-out instead of a logic function that called its own HTTP route in a loop with blocking sleeps. On install it counts people, opportunities and companies and enqueues one job per record batch via `enqueueJob`. Each job receives its batch id and processes the matching record window (offset pagination). Jobs are staggered with `delayMs` to stay under the hosted API rate limiting. + ## 1.2.0 - Compute last contact on Companies and Opportunities when the record or its relationships change, not only on new interactions: opportunities recompute from their point of contact on creation and when it changes, and companies recompute from their people on creation and when a person joins or leaves. diff --git a/packages/twenty-apps/public/last-contact/package.json b/packages/twenty-apps/public/last-contact/package.json index 6efa8c2d5e..b5fefa1d70 100644 --- a/packages/twenty-apps/public/last-contact/package.json +++ b/packages/twenty-apps/public/last-contact/package.json @@ -1,6 +1,6 @@ { "name": "@twentyhq/last-contact", - "version": "1.2.2", + "version": "1.2.3", "license": "MIT", "engines": { "node": "^24.5.0", @@ -28,8 +28,8 @@ "oxlint": "^0.16.0", "react": "^19.0.0", "react-dom": "^19.0.0", - "twenty-client-sdk": "^2.25.0", - "twenty-sdk": "^2.25.0", + "twenty-client-sdk": "^2.26.0", + "twenty-sdk": "^2.26.0", "typescript": "^5.9.3", "vite-tsconfig-paths": "^4.2.1", "vitest": "^4.0.0" diff --git a/packages/twenty-apps/public/last-contact/src/application-config.ts b/packages/twenty-apps/public/last-contact/src/application-config.ts index 69c31c6ba6..ff966bd582 100644 --- a/packages/twenty-apps/public/last-contact/src/application-config.ts +++ b/packages/twenty-apps/public/last-contact/src/application-config.ts @@ -22,12 +22,12 @@ export default defineApplication({ description: APP_DESCRIPTION, serverVariables: { [BACKFILL_BATCH_SIZE_ENV_VAR_NAME]: { - description: `How many records each last-contact backfill run processes before handing the next cursor back to the orchestrator. Defaults to ${DEFAULT_BACKFILL_BATCH_SIZE} when unset.`, + description: `How many records each last-contact backfill job processes. Also sets how many batch jobs are enqueued (record count divided by this). Defaults to ${DEFAULT_BACKFILL_BATCH_SIZE} when unset.`, isSecret: false, type: FieldType.NUMBER, }, [BACKFILL_SLEEP_MS_ENV_VAR_NAME]: { - description: `How many milliseconds the last-contact backfill orchestrator pauses between runs to stay under the API rate limiting. Defaults to ${DEFAULT_BACKFILL_SLEEP_MS} when unset.`, + description: `How many milliseconds to stagger consecutive backfill jobs by, so they do not all run at once and hit the API rate limiting. Defaults to ${DEFAULT_BACKFILL_SLEEP_MS} when unset.`, isSecret: false, type: FieldType.NUMBER, }, diff --git a/packages/twenty-apps/public/last-contact/src/constants/backfill.ts b/packages/twenty-apps/public/last-contact/src/constants/backfill.ts index e4bfd8bbcb..8ec5b9a7c1 100644 --- a/packages/twenty-apps/public/last-contact/src/constants/backfill.ts +++ b/packages/twenty-apps/public/last-contact/src/constants/backfill.ts @@ -1,39 +1,41 @@ -export const BACKFILL_ORCHESTRATOR_ROUTE_PATH = - '/last-contact/backfill-orchestrator'; -export const BACKFILL_PEOPLE_ROUTE_PATH = '/last-contact/backfill-people'; -export const BACKFILL_OPPORTUNITIES_ROUTE_PATH = - '/last-contact/backfill-opportunities'; -export const BACKFILL_COMPANIES_ROUTE_PATH = - '/last-contact/backfill-companies'; +import { + BACKFILL_COMPANIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, + BACKFILL_OPPORTUNITIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, + BACKFILL_PEOPLE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, +} from 'src/constants/universal-identifiers'; export type BackfillPhase = 'people' | 'opportunities' | 'companies'; -// People run first so companies and opportunities can read the freshly -// computed person last-contact, then opportunities, then companies. export const BACKFILL_PHASE_ORDER: BackfillPhase[] = [ 'people', 'opportunities', 'companies', ]; -export const BACKFILL_PHASE_ROUTE_PATHS: Record = { - people: BACKFILL_PEOPLE_ROUTE_PATH, - opportunities: BACKFILL_OPPORTUNITIES_ROUTE_PATH, - companies: BACKFILL_COMPANIES_ROUTE_PATH, +// GraphQL query field exposing the record connection for each phase. +export const BACKFILL_PHASE_QUERY_FIELD: Record = { + people: 'people', + opportunities: 'opportunities', + companies: 'companies', }; -export type BackfillState = { phase: BackfillPhase; cursor: string | null; iterations: number }; -export type BackfillBatchResult = { nextCursor: string | null; count: number }; +// Logic function each phase's batch jobs are enqueued against. +export const BACKFILL_PHASE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS: Record< + BackfillPhase, + string +> = { + people: BACKFILL_PEOPLE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, + opportunities: BACKFILL_OPPORTUNITIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, + companies: BACKFILL_COMPANIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, +}; -// Presence of this key acts as the backfill lock; it is deleted once every -// phase has completed. -export const BACKFILL_STATE_KV_KEY = 'last-contact:backfill-state'; +// A batch job resolves the records it owns from this index and the batch size. +export type BackfillBatchPayload = { batchId: number }; // Server variables, injected into process.env on every execution. export const BACKFILL_BATCH_SIZE_ENV_VAR_NAME = 'LAST_CONTACT_BACKFILL_BATCH_SIZE'; -export const BACKFILL_SLEEP_MS_ENV_VAR_NAME = - 'LAST_CONTACT_BACKFILL_SLEEP_MS'; +export const BACKFILL_SLEEP_MS_ENV_VAR_NAME = 'LAST_CONTACT_BACKFILL_SLEEP_MS'; export const DEFAULT_BACKFILL_BATCH_SIZE = 20; export const DEFAULT_BACKFILL_SLEEP_MS = 1_000; diff --git a/packages/twenty-apps/public/last-contact/src/constants/universal-identifiers.ts b/packages/twenty-apps/public/last-contact/src/constants/universal-identifiers.ts index 5cb9a7db4e..ed3dcc76ab 100644 --- a/packages/twenty-apps/public/last-contact/src/constants/universal-identifiers.ts +++ b/packages/twenty-apps/public/last-contact/src/constants/universal-identifiers.ts @@ -13,8 +13,6 @@ export const BACKFILL_POST_INSTALL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER = 'c94f671f-b3fa-47a2-8de6-dde94d13f8d1'; export const CALENDAR_EVENT_STARTED_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER = 'c56013d7-208b-46e2-a91f-27f481645591'; -export const BACKFILL_ORCHESTRATOR_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER = - '36cfa249-f101-4535-b420-193edccbaf52'; export const BACKFILL_PEOPLE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER = 'a1e15917-86d6-49a4-90d5-6a2555abc6f5'; export const BACKFILL_COMPANIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER = diff --git a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-companies-last-contact.ts b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-companies-last-contact.ts index 45709eeaf9..38dbf32735 100644 --- a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-companies-last-contact.ts +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-companies-last-contact.ts @@ -1,11 +1,9 @@ import { CoreApiClient } from 'twenty-client-sdk/core'; -import { defineLogicFunction, type RoutePayload } from 'twenty-sdk/define'; +import { defineLogicFunction } from 'twenty-sdk/define'; -import { - type BackfillBatchResult, - BACKFILL_COMPANIES_ROUTE_PATH, -} from 'src/constants/backfill'; +import { type BackfillBatchPayload } from 'src/constants/backfill'; import { BACKFILL_COMPANIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers'; +import { buildBackfillBatchArgs } from 'src/utils/backfill-batch-args'; import { getBackfillBatchSize } from 'src/utils/backfill-settings'; import { collectPeopleByCompany } from 'src/utils/collect-people-by-company'; import { executeWithRetry } from 'src/utils/execute-with-retry'; @@ -16,20 +14,14 @@ import { pickPersonLastContact, } from 'src/utils/person-last-contact-aggregation'; -type BackfillBody = { cursor?: string }; - -const handler = async ( - payload: RoutePayload, -): Promise => { +const handler = async ({ batchId }: BackfillBatchPayload): Promise => { const client = new CoreApiClient(); - const cursor = payload.body?.cursor; const { companies } = await executeWithRetry(() => client.query({ companies: { - __args: { first: getBackfillBatchSize(), after: cursor }, + __args: buildBackfillBatchArgs(batchId, getBackfillBatchSize()), edges: { node: { id: true } }, - pageInfo: { hasNextPage: true, endCursor: true }, }, }), ); @@ -39,7 +31,7 @@ const handler = async ( .filter(Boolean); if (companyIds.length === 0) { - return { nextCursor: null, count: 0 }; + return { batchId, count: 0 }; } const peopleByCompanyId = await collectPeopleByCompany(client, companyIds); @@ -67,24 +59,14 @@ const handler = async ( ); } - const nextCursor = - companies?.pageInfo.hasNextPage && companies.pageInfo.endCursor - ? companies.pageInfo.endCursor - : null; - - return { nextCursor, count: companyIds.length }; + return { batchId, count: companyIds.length }; }; export default defineLogicFunction({ universalIdentifier: BACKFILL_COMPANIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, name: 'backfill-companies-last-contact', description: - 'Backfills last-contact fields for one page of companies from the most recent contact of their people, returning the next cursor to the backfill orchestrator.', + 'Backfills last-contact fields for one batch of companies from the most recent contact of their people, resolved from the batch id in its payload.', timeoutSeconds: 120, handler, - httpRouteTriggerSettings: { - path: BACKFILL_COMPANIES_ROUTE_PATH, - httpMethod: 'POST', - isAuthRequired: true, - }, }); diff --git a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-get-state.ts b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-get-state.ts deleted file mode 100644 index af30b60869..0000000000 --- a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-get-state.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { defineLogicFunction } from 'twenty-sdk/define'; -import { kv } from 'twenty-sdk/logic-function'; - -import { - type BackfillState, - BACKFILL_STATE_KV_KEY, -} from 'src/constants/backfill'; - -const handler = async ({ clear = false }: { clear: boolean }) => { - const existingState = await kv.get(BACKFILL_STATE_KV_KEY); - - if (clear) { - await kv.delete(BACKFILL_STATE_KV_KEY); - } - - return { state: existingState, deleted: clear }; -}; - -export default defineLogicFunction({ - universalIdentifier: '6d8f9d22-61db-4865-8de2-9512c5d63b5d', - name: 'backfill-get-state', - description: 'Add a description for your logic function', - timeoutSeconds: 5, - handler, -}); diff --git a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-last-contact.ts b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-last-contact.ts index 69772c7371..1cc102c940 100644 --- a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-last-contact.ts +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-last-contact.ts @@ -1,42 +1,33 @@ +import { CoreApiClient } from 'twenty-client-sdk/core'; import { definePostInstallLogicFunction } from 'twenty-sdk/define'; -import { kv } from 'twenty-sdk/logic-function'; -import { - type BackfillState, - BACKFILL_ORCHESTRATOR_ROUTE_PATH, - BACKFILL_PHASE_ORDER, - BACKFILL_STATE_KV_KEY, -} from 'src/constants/backfill'; import { BACKFILL_POST_INSTALL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers'; -import { postToOwnRoute } from 'src/utils/post-to-own-route'; -import { getBackfillBatchSize, getBackfillSleepMs } from 'src/utils/backfill-settings'; +import { + getBackfillBatchSize, + getBackfillSleepMs, +} from 'src/utils/backfill-settings'; +import { enqueueBackfillJobs } from 'src/utils/enqueue-backfill-jobs'; const handler = async (): Promise => { - const existingState = await kv.get(BACKFILL_STATE_KV_KEY); + console.log( + 'Backfill params', + JSON.stringify({ + batchSize: getBackfillBatchSize(), + sleepMs: getBackfillSleepMs(), + }), + ); - console.log('Backfill params', JSON.stringify({ batchSize: getBackfillBatchSize(), sleepMs: getBackfillSleepMs() })); + const plans = await enqueueBackfillJobs(new CoreApiClient()); - if (existingState) { - return { outcome: 'already-running', state: existingState }; - } - - await kv.set(BACKFILL_STATE_KV_KEY, { - phase: BACKFILL_PHASE_ORDER[0], - cursor: null, - iterations: 0, - }); - - await postToOwnRoute({ path: BACKFILL_ORCHESTRATOR_ROUTE_PATH, body: {} }); - - return { outcome: 'started' }; + return { outcome: 'enqueued', plans }; }; export default definePostInstallLogicFunction({ universalIdentifier: BACKFILL_POST_INSTALL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, name: 'backfill-last-contact', description: - 'Starts the sequential last-contact backfill orchestrator after installation.', - timeoutSeconds: 60, + 'Counts people, opportunities and companies after installation and enqueues one backfill job per record batch.', + timeoutSeconds: 300, shouldRunOnVersionUpgrade: false, handler, }); diff --git a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-opportunities-last-contact.ts b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-opportunities-last-contact.ts index 721eb5fb69..27206ffd42 100644 --- a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-opportunities-last-contact.ts +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-opportunities-last-contact.ts @@ -1,11 +1,9 @@ import { CoreApiClient } from 'twenty-client-sdk/core'; -import { defineLogicFunction, type RoutePayload } from 'twenty-sdk/define'; +import { defineLogicFunction } from 'twenty-sdk/define'; -import { - type BackfillBatchResult, - BACKFILL_OPPORTUNITIES_ROUTE_PATH, -} from 'src/constants/backfill'; +import { type BackfillBatchPayload } from 'src/constants/backfill'; import { BACKFILL_OPPORTUNITIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers'; +import { buildBackfillBatchArgs } from 'src/utils/backfill-batch-args'; import { getBackfillBatchSize } from 'src/utils/backfill-settings'; import { executeWithRetry } from 'src/utils/execute-with-retry'; import { @@ -14,21 +12,16 @@ import { pickPersonLastContact, } from 'src/utils/person-last-contact-aggregation'; -type BackfillBody = { cursor?: string }; type OpportunityNode = { id: string; pointOfContactId: string | null }; -const handler = async ( - payload: RoutePayload, -): Promise => { +const handler = async ({ batchId }: BackfillBatchPayload): Promise => { const client = new CoreApiClient(); - const cursor = payload.body?.cursor; const { opportunities } = await executeWithRetry(() => client.query({ opportunities: { - __args: { first: getBackfillBatchSize(), after: cursor }, + __args: buildBackfillBatchArgs(batchId, getBackfillBatchSize()), edges: { node: { id: true, pointOfContactId: true } }, - pageInfo: { hasNextPage: true, endCursor: true }, }, }), ); @@ -38,7 +31,7 @@ const handler = async ( .filter((node: OpportunityNode) => Boolean(node.id)); if (nodes.length === 0) { - return { nextCursor: null, count: 0 }; + return { batchId, count: 0 }; } const personIds = [ @@ -69,12 +62,7 @@ const handler = async ( ); } - const nextCursor = - opportunities?.pageInfo.hasNextPage && opportunities.pageInfo.endCursor - ? opportunities.pageInfo.endCursor - : null; - - return { nextCursor, count: nodes.length }; + return { batchId, count: nodes.length }; }; export default defineLogicFunction({ @@ -82,12 +70,7 @@ export default defineLogicFunction({ BACKFILL_OPPORTUNITIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, name: 'backfill-opportunities-last-contact', description: - 'Backfills last-contact fields for one page of opportunities from their point of contact, returning the next cursor to the backfill orchestrator.', + 'Backfills last-contact fields for one batch of opportunities from their point of contact, resolved from the batch id in its payload.', timeoutSeconds: 120, handler, - httpRouteTriggerSettings: { - path: BACKFILL_OPPORTUNITIES_ROUTE_PATH, - httpMethod: 'POST', - isAuthRequired: true, - }, }); diff --git a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-orchestrator.ts b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-orchestrator.ts deleted file mode 100644 index e71d4dbfba..0000000000 --- a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-orchestrator.ts +++ /dev/null @@ -1,79 +0,0 @@ -import { defineLogicFunction } from 'twenty-sdk/define'; -import { kv } from 'twenty-sdk/logic-function'; - -import { - type BackfillBatchResult, - type BackfillState, - BACKFILL_ORCHESTRATOR_ROUTE_PATH, - BACKFILL_PHASE_ORDER, - BACKFILL_PHASE_ROUTE_PATHS, - BACKFILL_STATE_KV_KEY, -} from 'src/constants/backfill'; -import { BACKFILL_ORCHESTRATOR_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers'; -import { getBackfillSleepMs } from 'src/utils/backfill-settings'; -import { callOwnRoute, postToOwnRoute, sleep } from 'src/utils/post-to-own-route'; - -const MAX_ITERATIONS = 10_000; - -const handler = async (): Promise => { - const state = await kv.get(BACKFILL_STATE_KV_KEY); - - if (!state) { - return { outcome: 'no-active-backfill' }; - } - - if (state.iterations > MAX_ITERATIONS) { - return { outcome: 'max-iteration-reached' }; - } - - const { nextCursor } = await callOwnRoute({ - path: BACKFILL_PHASE_ROUTE_PATHS[state.phase], - body: { cursor: state.cursor ?? undefined }, - }); - - await sleep(getBackfillSleepMs()); - - if (nextCursor) { - await kv.set(BACKFILL_STATE_KV_KEY, { - phase: state.phase, - cursor: nextCursor, - iterations: (state.iterations ?? 0) + 1, - }); - await postToOwnRoute({ path: BACKFILL_ORCHESTRATOR_ROUTE_PATH, body: {} }); - - return { outcome: 'continued', phase: state.phase }; - } - - const nextPhase = - BACKFILL_PHASE_ORDER[BACKFILL_PHASE_ORDER.indexOf(state.phase) + 1]; - - if (nextPhase) { - await kv.set(BACKFILL_STATE_KV_KEY, { - phase: nextPhase, - cursor: null, - iterations: (state.iterations ?? 0) + 1, - }); - await postToOwnRoute({ path: BACKFILL_ORCHESTRATOR_ROUTE_PATH, body: {} }); - - return { outcome: 'phase-complete', phase: state.phase, nextPhase }; - } - - await kv.delete(BACKFILL_STATE_KV_KEY); - - return { outcome: 'done' }; -}; - -export default defineLogicFunction({ - universalIdentifier: - BACKFILL_ORCHESTRATOR_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, - name: 'backfill-orchestrator', - description: - 'Sequentially drives the people, opportunity and company last-contact backfills, tracking phase and cursor in the kv-store and re-triggering itself until every phase is complete.', - timeoutSeconds: 150, - handler, - httpRouteTriggerSettings: { - path: BACKFILL_ORCHESTRATOR_ROUTE_PATH, - httpMethod: 'POST', - isAuthRequired: true, - }, -}); diff --git a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-people-last-contact.ts b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-people-last-contact.ts index 8e80fbfc5f..ba501901ed 100644 --- a/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-people-last-contact.ts +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-people-last-contact.ts @@ -1,11 +1,9 @@ import { CoreApiClient } from 'twenty-client-sdk/core'; -import { defineLogicFunction, type RoutePayload } from 'twenty-sdk/define'; +import { defineLogicFunction } from 'twenty-sdk/define'; -import { - type BackfillBatchResult, - BACKFILL_PEOPLE_ROUTE_PATH, -} from 'src/constants/backfill'; +import { type BackfillBatchPayload } from 'src/constants/backfill'; import { BACKFILL_PEOPLE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers'; +import { buildBackfillBatchArgs } from 'src/utils/backfill-batch-args'; import { getBackfillBatchSize } from 'src/utils/backfill-settings'; import { executeWithRetry } from 'src/utils/execute-with-retry'; import { @@ -13,20 +11,14 @@ import { buildPersonUpdateData, } from 'src/utils/person-last-contact-aggregation'; -type BackfillBody = { cursor?: string }; - -const handler = async ( - payload: RoutePayload, -): Promise => { +const handler = async ({ batchId }: BackfillBatchPayload): Promise => { const client = new CoreApiClient(); - const cursor = payload.body?.cursor; const { people } = await executeWithRetry(() => client.query({ people: { - __args: { first: getBackfillBatchSize(), after: cursor }, + __args: buildBackfillBatchArgs(batchId, getBackfillBatchSize()), edges: { node: { id: true } }, - pageInfo: { hasNextPage: true, endCursor: true }, }, }), ); @@ -36,7 +28,7 @@ const handler = async ( .filter(Boolean); if (personIds.length === 0) { - return { nextCursor: null, count: 0 }; + return { batchId, count: 0 }; } const aggByPersonId = await buildPersonAggregates(client, personIds); @@ -56,24 +48,14 @@ const handler = async ( ); } - const nextCursor = - people?.pageInfo.hasNextPage && people.pageInfo.endCursor - ? people.pageInfo.endCursor - : null; - - return { nextCursor, count: personIds.length }; + return { batchId, count: personIds.length }; }; export default defineLogicFunction({ universalIdentifier: BACKFILL_PEOPLE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, name: 'backfill-people-last-contact', description: - 'Backfills last-contact fields for one page of people from their messages and calendar events, returning the next cursor to the backfill orchestrator.', + 'Backfills last-contact fields for one batch of people, resolved from the batch id in its payload.', timeoutSeconds: 120, handler, - httpRouteTriggerSettings: { - path: BACKFILL_PEOPLE_ROUTE_PATH, - httpMethod: 'POST', - isAuthRequired: true, - }, }); diff --git a/packages/twenty-apps/public/last-contact/src/utils/backfill-batch-args.ts b/packages/twenty-apps/public/last-contact/src/utils/backfill-batch-args.ts new file mode 100644 index 0000000000..c7f6a840e6 --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/utils/backfill-batch-args.ts @@ -0,0 +1,17 @@ +// A total order over createdAt then id keeps offset windows stable while the +// backfill runs: newly created records sort to the tail, so the batch each job +// owns does not shift under it. +const BACKFILL_ORDER_BY = [ + { createdAt: 'AscNullsFirst' }, + { id: 'AscNullsFirst' }, +]; + +// Query args selecting the record window a batch job owns. +export const buildBackfillBatchArgs = ( + batchId: number, + batchSize: number, +): { first: number; offset: number; orderBy: typeof BACKFILL_ORDER_BY } => ({ + first: batchSize, + offset: batchId * batchSize, + orderBy: BACKFILL_ORDER_BY, +}); diff --git a/packages/twenty-apps/public/last-contact/src/utils/enqueue-backfill-jobs.ts b/packages/twenty-apps/public/last-contact/src/utils/enqueue-backfill-jobs.ts new file mode 100644 index 0000000000..2e6edf8c5e --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/utils/enqueue-backfill-jobs.ts @@ -0,0 +1,70 @@ +import { type CoreApiClient } from 'twenty-client-sdk/core'; +import { enqueueJob } from 'twenty-sdk/logic-function'; + +import { + type BackfillPhase, + BACKFILL_PHASE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS, + BACKFILL_PHASE_ORDER, + BACKFILL_PHASE_QUERY_FIELD, +} from 'src/constants/backfill'; +import { + getBackfillBatchSize, + getBackfillSleepMs, +} from 'src/utils/backfill-settings'; +import { executeWithRetry } from 'src/utils/execute-with-retry'; + +// enqueueJob rejects delays beyond 7 days. +const MAX_ENQUEUE_DELAY_MS = 7 * 24 * 60 * 60 * 1_000; + +// Batch handlers are idempotent (they recompute from source and overwrite), so +// a batch that dies mid-run can safely be retried by the queue rather than +// leaving its records unbackfilled. +const BACKFILL_JOB_RETRY_LIMIT = 3; + +type BackfillPhasePlan = { phase: BackfillPhase; count: number; batches: number }; + +const countPhaseRecords = async ( + client: CoreApiClient, + phase: BackfillPhase, +): Promise => { + const field = BACKFILL_PHASE_QUERY_FIELD[phase]; + + const result = await executeWithRetry(() => + client.query({ [field]: { __args: { first: 1 }, totalCount: true } }), + ); + + return result?.[field]?.totalCount ?? 0; +}; + +// Counts every phase, then enqueues one job per record batch across all phases. +// Jobs are spaced by delayMs so thousands of batches do not all become eligible +// at once and overwhelm the API rate limiting. +export const enqueueBackfillJobs = async ( + client: CoreApiClient, +): Promise => { + const batchSize = getBackfillBatchSize(); + const sleepMs = getBackfillSleepMs(); + + const plans: BackfillPhasePlan[] = []; + let enqueuedCount = 0; + + for (const phase of BACKFILL_PHASE_ORDER) { + const count = await countPhaseRecords(client, phase); + const batches = Math.ceil(count / batchSize); + + for (let batchId = 0; batchId < batches; batchId++) { + await enqueueJob({ + logicFunctionUniversalIdentifier: + BACKFILL_PHASE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS[phase], + payload: { batchId }, + delayMs: Math.min(enqueuedCount * sleepMs, MAX_ENQUEUE_DELAY_MS), + retryLimit: BACKFILL_JOB_RETRY_LIMIT, + }); + enqueuedCount++; + } + + plans.push({ phase, count, batches }); + } + + return plans; +}; diff --git a/packages/twenty-apps/public/last-contact/src/utils/post-to-own-route.ts b/packages/twenty-apps/public/last-contact/src/utils/post-to-own-route.ts deleted file mode 100644 index 1ff98d58f1..0000000000 --- a/packages/twenty-apps/public/last-contact/src/utils/post-to-own-route.ts +++ /dev/null @@ -1,52 +0,0 @@ -import { RestApiClient } from 'twenty-client-sdk/rest'; - -const OWN_ROUTE_FLUSH_MS = 5_000; -const OWN_ROUTE_CALL_MS = 120_000; - -// Blocking POST to one of this app's own HTTP routes that waits for and returns -// the handler response. -export const callOwnRoute = async ({ - path, - body, -}: { - path: string; - body: object; -}): Promise => { - const client = new RestApiClient(); - - return client.post(`/s${path}`, body, { - signal: AbortSignal.timeout(OWN_ROUTE_CALL_MS), - }); -}; - -// Fire-and-forget POST to one of this app's own HTTP routes; a timeout only -// means the request was flushed, not that the target run failed. -export const postToOwnRoute = async ({ - path, - body, -}: { - path: string; - body: object; -}): Promise => { - try { - const client = new RestApiClient(); - - await client.post(`/s${path}`, body, { - signal: AbortSignal.timeout(OWN_ROUTE_FLUSH_MS), - }); - } catch (error) { - if ( - error instanceof Error && - (error.name === 'TimeoutError' || error.name === 'AbortError') - ) { - return; - } - - console.error( - `[last-contact] request to own route ${path} failed to fire: ${error instanceof Error ? error.message : String(error)}`, - ); - } -}; - -export const sleep = (durationMs: number): Promise => - new Promise((resolve) => setTimeout(resolve, durationMs)); diff --git a/packages/twenty-apps/public/last-contact/yarn.lock b/packages/twenty-apps/public/last-contact/yarn.lock index f50cec116d..ca2a84c770 100644 --- a/packages/twenty-apps/public/last-contact/yarn.lock +++ b/packages/twenty-apps/public/last-contact/yarn.lock @@ -953,8 +953,8 @@ __metadata: oxlint: "npm:^0.16.0" react: "npm:^19.0.0" react-dom: "npm:^19.0.0" - twenty-client-sdk: "npm:^2.25.0" - twenty-sdk: "npm:^2.25.0" + twenty-client-sdk: "npm:^2.26.0" + twenty-sdk: "npm:^2.26.0" typescript: "npm:^5.9.3" vite-tsconfig-paths: "npm:^4.2.1" vitest: "npm:^4.0.0" @@ -2833,22 +2833,22 @@ __metadata: languageName: node linkType: hard -"twenty-client-sdk@npm:2.25.0, twenty-client-sdk@npm:^2.25.0": - version: 2.25.0 - resolution: "twenty-client-sdk@npm:2.25.0" +"twenty-client-sdk@npm:2.26.0, twenty-client-sdk@npm:^2.26.0": + version: 2.26.0 + resolution: "twenty-client-sdk@npm:2.26.0" dependencies: "@genql/runtime": "npm:^2.10.0" esbuild: "npm:^0.28.1" graphql: "npm:^16.8.1" lodash: "npm:^4.17.21" prettier: "npm:^3.8.3" - checksum: 10c0/3584eb1138994213e2cedd3019bdbaec03e3a715d09e39080df15901284c1bc9e00ab48c8d6fbf482363a484c3e196afde7306dc48f5f870136c5f51156fdd93 + checksum: 10c0/f5d184567877d10175f887e5ee81510d43c8c5c4d98ad73fbc757a28547da8299ccbe3072d9de33f8c9e9a41eafe06f335b3706196f4a92fd33b0cdcc5d47acf languageName: node linkType: hard -"twenty-sdk@npm:^2.25.0": - version: 2.25.0 - resolution: "twenty-sdk@npm:2.25.0" +"twenty-sdk@npm:^2.26.0": + version: 2.26.0 + resolution: "twenty-sdk@npm:2.26.0" dependencies: "@sniptt/guards": "npm:^0.2.0" axios: "npm:^1.16.0" @@ -2867,12 +2867,12 @@ __metadata: semver: "npm:7.6.3" sharp: "npm:^0.34.5" tinyglobby: "npm:^0.2.15" - twenty-client-sdk: "npm:2.25.0" + twenty-client-sdk: "npm:2.26.0" typescript: "npm:^5.9.3" uuid: "npm:^13.0.2" bin: twenty: dist/cli.cjs - checksum: 10c0/c74ca145c5ec35e66817602bbc6319147aea97a5d1a372c30cc325407cb2c850c0a0363eee1bd9a999c951acb4cddba96ba9813324a6f34122bcd6c6a835de31 + checksum: 10c0/e3edaced59a93bfa3579dc863f799693c9361fc4993eec6942f711a394a678dff21a9c790a3ad6b1683f71ed3f80ec7e6bcc4172d123f538912f88ff7ca32cfa languageName: node linkType: hard