refactor(last-contact): drive backfill with enqueued jobs (#23646)
## What
Reworks the last-contact backfill, which previously ran an orchestrator
that called its own HTTP route in a recursive loop with blocking
`sleep`s, into a single upfront fan-out of enqueued jobs.
On install, `backfill-last-contact` counts people, opportunities and
companies, then enqueues one job per record batch (`ceil(count /
batchSize)`) for each. Each batch job receives its `batchId` in its
payload and processes the matching record window via offset pagination
(`first`/`offset` with a stable `createdAt, id` ordering). No
self-calling loop, no blocking sleeps, no per-batch chaining, no kv
progress state.
Phases don't need to run in sequence: each phase recomputes last-contact
from source interactions (message and calendar participants), not from
the person's stored field, so the jobs are independent and safe to run
concurrently.
## Changes
- `backfill-last-contact` (post-install): counts each phase and fans out
all batch jobs via `enqueueJob`; timeout raised to 300s. Enqueues are
concurrency-limited and staggered with `delayMs` (from
`LAST_CONTACT_BACKFILL_SLEEP_MS`) so thousands of jobs don't all become
eligible at once.
- `src/utils/enqueue-backfill-jobs.ts` (new): counts a phase via the
connection `totalCount`, builds the batch plan, and enqueues the jobs.
- `src/utils/backfill-batch-args.ts` (new): `first`/`offset`/`orderBy`
window for a given `batchId`, ordered by `createdAt, id` so offsets stay
stable while the backfill runs.
- `backfill-{people,opportunities,companies}-last-contact`: handlers now
take `{ batchId }`, query their offset window, process it, and return `{
batchId, count }`. No HTTP route triggers, no cursor/kv state.
- Removed the chaining util (`advance-backfill.ts`) and the
now-purposeless kv-state debug helper (`backfill-get-state.ts`).
- Bumped `twenty-sdk`/`twenty-client-sdk` to `^2.26.0` (first published
version exporting `enqueueJob`), app version to `1.2.3`, plus changelog.
Updated the server-variable copy.
## Testing
- `yarn typecheck` clean
- `yarn lint` clean (0 warnings, 0 errors)
- `yarn test:unit` green (38 passed)
This commit is contained in:
martmull
@@ -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.
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -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<BackfillPhase, string> = {
|
||||
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<BackfillPhase, string> = {
|
||||
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;
|
||||
|
||||
@@ -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 =
|
||||
|
||||
+8
-26
@@ -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<BackfillBody>,
|
||||
): Promise<BackfillBatchResult> => {
|
||||
const handler = async ({ batchId }: BackfillBatchPayload): Promise<object> => {
|
||||
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,
|
||||
},
|
||||
});
|
||||
|
||||
@@ -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<BackfillState>(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,
|
||||
});
|
||||
+17
-26
@@ -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<object> => {
|
||||
const existingState = await kv.get<BackfillState>(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<BackfillState>(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,
|
||||
});
|
||||
|
||||
+8
-25
@@ -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<BackfillBody>,
|
||||
): Promise<BackfillBatchResult> => {
|
||||
const handler = async ({ batchId }: BackfillBatchPayload): Promise<object> => {
|
||||
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,
|
||||
},
|
||||
});
|
||||
|
||||
@@ -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<object> => {
|
||||
const state = await kv.get<BackfillState>(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<BackfillBatchResult>({
|
||||
path: BACKFILL_PHASE_ROUTE_PATHS[state.phase],
|
||||
body: { cursor: state.cursor ?? undefined },
|
||||
});
|
||||
|
||||
await sleep(getBackfillSleepMs());
|
||||
|
||||
if (nextCursor) {
|
||||
await kv.set<BackfillState>(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<BackfillState>(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,
|
||||
},
|
||||
});
|
||||
+8
-26
@@ -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<BackfillBody>,
|
||||
): Promise<BackfillBatchResult> => {
|
||||
const handler = async ({ batchId }: BackfillBatchPayload): Promise<object> => {
|
||||
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,
|
||||
},
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
@@ -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<number> => {
|
||||
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<BackfillPhasePlan[]> => {
|
||||
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;
|
||||
};
|
||||
@@ -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 <TResponse>({
|
||||
path,
|
||||
body,
|
||||
}: {
|
||||
path: string;
|
||||
body: object;
|
||||
}): Promise<TResponse> => {
|
||||
const client = new RestApiClient();
|
||||
|
||||
return client.post<TResponse>(`/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<void> => {
|
||||
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<void> =>
|
||||
new Promise((resolve) => setTimeout(resolve, durationMs));
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user