Replace last-contact backfill with cursor-paginated per-record backfills (#23582)
## What Replaces the single-pass last-contact backfill in the `last-contact` app with three independent cursor-paginated backfills, one per object: - `backfill-people-last-contact` - `backfill-companies-last-contact` - `backfill-opportunities-last-contact` The post-install `backfill-last-contact` function now just dispatches the three by posting to their own HTTP routes. ## How it works Each backfill function: 1. Selects the first 20 records after the given cursor. 2. Computes the last-contact columns for each record, one by one, from raw message and calendar interactions (people from their own interactions, companies from the most recent contact of their people, opportunities from their point of contact). 3. Updates each record individually. 4. Sleeps briefly, then re-triggers itself with the next cursor until there are no more records. Because every batch computes from raw interaction data, the three backfills are order-independent and can run concurrently. ## Why The previous backfill loaded everything and fired updates in bursts, which hit hosted API rate limiting on large workspaces. Spreading updates 20 records at a time with a pause between pages keeps the load under the limit. This is a temporary fix until the `enqueueJob` utility handles throttling natively. ## Notes - App version bumped to 1.1.4 so the upgrade hook re-runs on existing installs. - No tests added, per the temporary nature of the change. - Typecheck and lint pass. --- _Generated by [Claude Code](https://claude.ai/code/session_01GxPKyuxZnx5oyUap3wcBTb)_ <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/twentyhq/twenty/pull/23582?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. -->
This commit is contained in:
martmull
@@ -3,6 +3,7 @@
|
||||
## 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.
|
||||
- Rework the last-contact backfill into a sequential, cursor-paginated process orchestrated through the kv-store (people, then opportunities, then companies). Each run handles one batch and hands the next cursor back to the orchestrator, which pauses between runs to stay under the hosted API rate limiting. Batch size and pause are server variables.
|
||||
|
||||
## 1.1.3
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@twentyhq/last-contact",
|
||||
"version": "1.2.0",
|
||||
"version": "1.2.1",
|
||||
"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.23.0-alpha.2",
|
||||
"twenty-sdk": "2.23.0-alpha.2",
|
||||
"twenty-client-sdk": "^2.25.0",
|
||||
"twenty-sdk": "^2.25.0",
|
||||
"typescript": "^5.9.3",
|
||||
"vite-tsconfig-paths": "^4.2.1",
|
||||
"vitest": "^4.0.0"
|
||||
|
||||
@@ -1,5 +1,11 @@
|
||||
import { defineApplication } from 'twenty-sdk/define';
|
||||
import { defineApplication, FieldType } from 'twenty-sdk/define';
|
||||
|
||||
import {
|
||||
BACKFILL_BATCH_SIZE_ENV_VAR_NAME,
|
||||
BACKFILL_SLEEP_MS_ENV_VAR_NAME,
|
||||
DEFAULT_BACKFILL_BATCH_SIZE,
|
||||
DEFAULT_BACKFILL_SLEEP_MS,
|
||||
} from 'src/constants/backfill';
|
||||
import {
|
||||
APP_DESCRIPTION,
|
||||
APP_DISPLAY_NAME,
|
||||
@@ -14,4 +20,16 @@ export default defineApplication({
|
||||
screenshots: ['public/gallery/cover.png'],
|
||||
displayName: APP_DISPLAY_NAME,
|
||||
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.`,
|
||||
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.`,
|
||||
isSecret: false,
|
||||
type: FieldType.NUMBER,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
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';
|
||||
|
||||
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,
|
||||
};
|
||||
|
||||
export type BackfillState = { phase: BackfillPhase; cursor: string | null; iterations: number };
|
||||
export type BackfillBatchResult = { nextCursor: string | null; count: number };
|
||||
|
||||
// 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';
|
||||
|
||||
// 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 DEFAULT_BACKFILL_BATCH_SIZE = 20;
|
||||
export const DEFAULT_BACKFILL_SLEEP_MS = 1_000;
|
||||
@@ -13,6 +13,14 @@ 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 =
|
||||
'086d247f-28da-41be-8edd-e6c58592d666';
|
||||
export const BACKFILL_OPPORTUNITIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER =
|
||||
'af176746-4334-47f8-a017-88a474a03fa3';
|
||||
export const OPPORTUNITY_CREATED_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER =
|
||||
'6659b54f-2f46-412b-aa7d-03aa9f1c5133';
|
||||
export const OPPORTUNITY_UPDATED_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER =
|
||||
|
||||
+90
@@ -0,0 +1,90 @@
|
||||
import { CoreApiClient } from 'twenty-client-sdk/core';
|
||||
import { defineLogicFunction, type RoutePayload } from 'twenty-sdk/define';
|
||||
|
||||
import {
|
||||
type BackfillBatchResult,
|
||||
BACKFILL_COMPANIES_ROUTE_PATH,
|
||||
} from 'src/constants/backfill';
|
||||
import { BACKFILL_COMPANIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers';
|
||||
import { getBackfillBatchSize } from 'src/utils/backfill-settings';
|
||||
import { collectPeopleByCompany } from 'src/utils/collect-people-by-company';
|
||||
import { executeWithRetry } from 'src/utils/execute-with-retry';
|
||||
import {
|
||||
buildRelatedUpdateData,
|
||||
buildPersonAggregates,
|
||||
pickLatestLastContact,
|
||||
pickPersonLastContact,
|
||||
} from 'src/utils/person-last-contact-aggregation';
|
||||
|
||||
type BackfillBody = { cursor?: string };
|
||||
|
||||
const handler = async (
|
||||
payload: RoutePayload<BackfillBody>,
|
||||
): Promise<BackfillBatchResult> => {
|
||||
const client = new CoreApiClient();
|
||||
const cursor = payload.body?.cursor;
|
||||
|
||||
const { companies } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
companies: {
|
||||
__args: { first: getBackfillBatchSize(), after: cursor },
|
||||
edges: { node: { id: true } },
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const companyIds = (companies?.edges ?? [])
|
||||
.map((edge: { node: { id: string } }) => edge.node.id)
|
||||
.filter(Boolean);
|
||||
|
||||
if (companyIds.length === 0) {
|
||||
return { nextCursor: null, count: 0 };
|
||||
}
|
||||
|
||||
const peopleByCompanyId = await collectPeopleByCompany(client, companyIds);
|
||||
const personIds = [...new Set([...peopleByCompanyId.values()].flat())];
|
||||
const aggByPersonId = await buildPersonAggregates(client, personIds);
|
||||
|
||||
for (const companyId of companyIds) {
|
||||
const lastContact = pickLatestLastContact(
|
||||
(peopleByCompanyId.get(companyId) ?? [])
|
||||
.map((personId) => pickPersonLastContact(aggByPersonId.get(personId)))
|
||||
.filter((contact) => contact !== undefined),
|
||||
);
|
||||
|
||||
if (!lastContact) {
|
||||
continue;
|
||||
}
|
||||
|
||||
await executeWithRetry(() =>
|
||||
client.mutation({
|
||||
updateCompany: {
|
||||
__args: { id: companyId, data: buildRelatedUpdateData(lastContact) },
|
||||
id: true,
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const nextCursor =
|
||||
companies?.pageInfo.hasNextPage && companies.pageInfo.endCursor
|
||||
? companies.pageInfo.endCursor
|
||||
: null;
|
||||
|
||||
return { nextCursor, 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.',
|
||||
timeoutSeconds: 120,
|
||||
handler,
|
||||
httpRouteTriggerSettings: {
|
||||
path: BACKFILL_COMPANIES_ROUTE_PATH,
|
||||
httpMethod: 'POST',
|
||||
isAuthRequired: true,
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,25 @@
|
||||
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,
|
||||
});
|
||||
+21
-516
@@ -1,534 +1,39 @@
|
||||
import { definePostInstallLogicFunction } from 'twenty-sdk/define';
|
||||
import { CoreApiClient } from 'twenty-client-sdk/core';
|
||||
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 { executeWithRetry } from 'src/utils/execute-with-retry';
|
||||
import { postToOwnRoute } from 'src/utils/post-to-own-route';
|
||||
|
||||
const PAGE_SIZE = 200;
|
||||
// Kept low so update bursts stay under Cloudflare rate limiting on hosted
|
||||
// workspaces; executeWithRetry absorbs the occasional 429 that still slips
|
||||
// through.
|
||||
const UPDATE_BATCH_SIZE = 10;
|
||||
const handler = async (): Promise<object> => {
|
||||
const existingState = await kv.get<BackfillState>(BACKFILL_STATE_KV_KEY);
|
||||
|
||||
type EmailInteraction = {
|
||||
personId: string;
|
||||
messageId: string;
|
||||
receivedAt: string;
|
||||
};
|
||||
type MeetingInteraction = {
|
||||
personId: string;
|
||||
calendarEventId: string;
|
||||
startsAt: string;
|
||||
};
|
||||
type MessageMemberInfo = { ownerId: string; fromIsMember: boolean };
|
||||
type ContactItem = { kind: 'email' | 'meeting'; id: string };
|
||||
type LastContact = { at: string; item: ContactItem };
|
||||
type OpportunityRow = {
|
||||
id: string;
|
||||
pointOfContactId: string | null;
|
||||
};
|
||||
|
||||
type PersonAgg = {
|
||||
lastContactAt?: string;
|
||||
lastContactById?: string | null;
|
||||
item?: { kind: 'email' | 'meeting'; id: string };
|
||||
lastOutboundAt?: string;
|
||||
lastInboundAt?: string;
|
||||
lastEmail?: { at: string; id: string };
|
||||
lastMeeting?: { at: string; id: string };
|
||||
};
|
||||
type AggByPersonId = Map<string, PersonAgg>;
|
||||
|
||||
type PersonUpdateData = Record<string, string | null>;
|
||||
type RecordUpdate = { id: string; data: PersonUpdateData };
|
||||
|
||||
const chunk = <T>(items: T[], size: number): T[][] => {
|
||||
const chunks: T[][] = [];
|
||||
for (let i = 0; i < items.length; i += size) {
|
||||
chunks.push(items.slice(i, i + size));
|
||||
}
|
||||
return chunks;
|
||||
};
|
||||
|
||||
const collectEmailInteractions = async (
|
||||
client: CoreApiClient,
|
||||
): Promise<EmailInteraction[]> => {
|
||||
const interactions: EmailInteraction[] = [];
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { messageParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
messageParticipants: {
|
||||
__args: {
|
||||
filter: { personId: { is: 'NOT_NULL' } },
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: {
|
||||
id: true,
|
||||
personId: true,
|
||||
message: { id: true, receivedAt: true },
|
||||
},
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of messageParticipants?.edges ?? []) {
|
||||
const { personId, message } = edge.node;
|
||||
if (personId && message?.id && message?.receivedAt) {
|
||||
interactions.push({
|
||||
personId,
|
||||
messageId: message.id,
|
||||
receivedAt: message.receivedAt,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
after = messageParticipants?.pageInfo.hasNextPage
|
||||
? (messageParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
|
||||
return interactions;
|
||||
};
|
||||
|
||||
const collectMeetingInteractions = async (
|
||||
client: CoreApiClient,
|
||||
): Promise<MeetingInteraction[]> => {
|
||||
const now = new Date().toISOString();
|
||||
const interactions: MeetingInteraction[] = [];
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { calendarEventParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
calendarEventParticipants: {
|
||||
__args: {
|
||||
filter: { personId: { is: 'NOT_NULL' } },
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: {
|
||||
id: true,
|
||||
personId: true,
|
||||
calendarEvent: { id: true, startsAt: true, isCanceled: true },
|
||||
},
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of calendarEventParticipants?.edges ?? []) {
|
||||
const { personId, calendarEvent } = edge.node;
|
||||
if (
|
||||
personId &&
|
||||
calendarEvent?.id &&
|
||||
calendarEvent?.startsAt &&
|
||||
!calendarEvent.isCanceled &&
|
||||
calendarEvent.startsAt <= now
|
||||
) {
|
||||
interactions.push({
|
||||
personId,
|
||||
calendarEventId: calendarEvent.id,
|
||||
startsAt: calendarEvent.startsAt,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
after = calendarEventParticipants?.pageInfo.hasNextPage
|
||||
? (calendarEventParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
|
||||
return interactions;
|
||||
};
|
||||
|
||||
const collectMessageMemberInfo = async (
|
||||
client: CoreApiClient,
|
||||
messageIds: string[],
|
||||
): Promise<Map<string, MessageMemberInfo>> => {
|
||||
const infoByMessageId = new Map<string, MessageMemberInfo>();
|
||||
|
||||
for (const ids of chunk(messageIds, PAGE_SIZE)) {
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { messageParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
messageParticipants: {
|
||||
__args: {
|
||||
filter: {
|
||||
messageId: { in: ids },
|
||||
workspaceMemberId: { is: 'NOT_NULL' },
|
||||
},
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: { messageId: true, role: true, workspaceMemberId: true },
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of messageParticipants?.edges ?? []) {
|
||||
const { messageId, role, workspaceMemberId } = edge.node;
|
||||
if (!messageId || !workspaceMemberId) {
|
||||
continue;
|
||||
}
|
||||
const info = infoByMessageId.get(messageId) ?? {
|
||||
ownerId: workspaceMemberId,
|
||||
fromIsMember: false,
|
||||
};
|
||||
if (role === 'FROM') {
|
||||
info.ownerId = workspaceMemberId;
|
||||
info.fromIsMember = true;
|
||||
}
|
||||
infoByMessageId.set(messageId, info);
|
||||
}
|
||||
|
||||
after = messageParticipants?.pageInfo.hasNextPage
|
||||
? (messageParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
if (existingState) {
|
||||
return { outcome: 'already-running', state: existingState };
|
||||
}
|
||||
|
||||
return infoByMessageId;
|
||||
};
|
||||
await kv.set<BackfillState>(BACKFILL_STATE_KV_KEY, {
|
||||
phase: BACKFILL_PHASE_ORDER[0],
|
||||
cursor: null,
|
||||
iterations: 0,
|
||||
});
|
||||
|
||||
const collectCalendarOwners = async (
|
||||
client: CoreApiClient,
|
||||
calendarEventIds: string[],
|
||||
): Promise<Map<string, string>> => {
|
||||
const ownerByCalendarEventId = new Map<string, string>();
|
||||
await postToOwnRoute({ path: BACKFILL_ORCHESTRATOR_ROUTE_PATH, body: {} });
|
||||
|
||||
for (const ids of chunk(calendarEventIds, PAGE_SIZE)) {
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { calendarEventParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
calendarEventParticipants: {
|
||||
__args: {
|
||||
filter: {
|
||||
calendarEventId: { in: ids },
|
||||
workspaceMemberId: { is: 'NOT_NULL' },
|
||||
},
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: {
|
||||
calendarEventId: true,
|
||||
isOrganizer: true,
|
||||
workspaceMemberId: true,
|
||||
},
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of calendarEventParticipants?.edges ?? []) {
|
||||
const { calendarEventId, isOrganizer, workspaceMemberId } = edge.node;
|
||||
if (
|
||||
calendarEventId &&
|
||||
workspaceMemberId &&
|
||||
(!ownerByCalendarEventId.has(calendarEventId) || isOrganizer === true)
|
||||
) {
|
||||
ownerByCalendarEventId.set(calendarEventId, workspaceMemberId);
|
||||
}
|
||||
}
|
||||
|
||||
after = calendarEventParticipants?.pageInfo.hasNextPage
|
||||
? (calendarEventParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
}
|
||||
|
||||
return ownerByCalendarEventId;
|
||||
};
|
||||
|
||||
const collectPersonCompanies = async (
|
||||
client: CoreApiClient,
|
||||
): Promise<Map<string, string>> => {
|
||||
const companyByPersonId = new Map<string, string>();
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { people } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
people: {
|
||||
__args: {
|
||||
filter: { companyId: { is: 'NOT_NULL' } },
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: { node: { id: true, companyId: true } },
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of people?.edges ?? []) {
|
||||
const { id, companyId } = edge.node;
|
||||
if (id && companyId) {
|
||||
companyByPersonId.set(id, companyId);
|
||||
}
|
||||
}
|
||||
|
||||
after = people?.pageInfo.hasNextPage
|
||||
? (people.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
|
||||
return companyByPersonId;
|
||||
};
|
||||
|
||||
const collectOpportunities = async (
|
||||
client: CoreApiClient,
|
||||
): Promise<OpportunityRow[]> => {
|
||||
const opportunities: OpportunityRow[] = [];
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { opportunities: page } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
opportunities: {
|
||||
__args: { first: PAGE_SIZE, after },
|
||||
edges: {
|
||||
node: { id: true, pointOfContactId: true },
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of page?.edges ?? []) {
|
||||
const { id, pointOfContactId } = edge.node;
|
||||
if (id) {
|
||||
opportunities.push({
|
||||
id,
|
||||
pointOfContactId: pointOfContactId ?? null,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
after = page?.pageInfo.hasNextPage
|
||||
? (page.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
|
||||
return opportunities;
|
||||
};
|
||||
|
||||
const buildRelatedData = ({ at, item }: LastContact): PersonUpdateData => ({
|
||||
lastContactAt: at,
|
||||
lastContactItemMessageId: item.kind === 'email' ? item.id : null,
|
||||
lastContactItemCalendarEventId: item.kind === 'meeting' ? item.id : null,
|
||||
});
|
||||
|
||||
const personLastContact = (agg: PersonAgg): LastContact | undefined =>
|
||||
agg.lastContactAt && agg.item
|
||||
? { at: agg.lastContactAt, item: agg.item }
|
||||
: undefined;
|
||||
|
||||
const foldEmail = (
|
||||
agg: PersonAgg,
|
||||
receivedAt: string,
|
||||
messageId: string,
|
||||
info: MessageMemberInfo | undefined,
|
||||
): void => {
|
||||
if (!agg.lastEmail || receivedAt > agg.lastEmail.at) {
|
||||
agg.lastEmail = { at: receivedAt, id: messageId };
|
||||
}
|
||||
if (info?.fromIsMember) {
|
||||
if (!agg.lastOutboundAt || receivedAt > agg.lastOutboundAt) {
|
||||
agg.lastOutboundAt = receivedAt;
|
||||
}
|
||||
} else if (!agg.lastInboundAt || receivedAt > agg.lastInboundAt) {
|
||||
agg.lastInboundAt = receivedAt;
|
||||
}
|
||||
if (!agg.lastContactAt || receivedAt > agg.lastContactAt) {
|
||||
agg.lastContactAt = receivedAt;
|
||||
agg.lastContactById = info?.ownerId ?? null;
|
||||
agg.item = { kind: 'email', id: messageId };
|
||||
}
|
||||
};
|
||||
|
||||
const foldMeeting = (
|
||||
agg: PersonAgg,
|
||||
startsAt: string,
|
||||
calendarEventId: string,
|
||||
ownerId: string | null,
|
||||
): void => {
|
||||
if (!agg.lastMeeting || startsAt > agg.lastMeeting.at) {
|
||||
agg.lastMeeting = { at: startsAt, id: calendarEventId };
|
||||
}
|
||||
if (!agg.lastOutboundAt || startsAt > agg.lastOutboundAt) {
|
||||
agg.lastOutboundAt = startsAt;
|
||||
}
|
||||
if (!agg.lastInboundAt || startsAt > agg.lastInboundAt) {
|
||||
agg.lastInboundAt = startsAt;
|
||||
}
|
||||
if (!agg.lastContactAt || startsAt > agg.lastContactAt) {
|
||||
agg.lastContactAt = startsAt;
|
||||
agg.lastContactById = ownerId;
|
||||
agg.item = { kind: 'meeting', id: calendarEventId };
|
||||
}
|
||||
};
|
||||
|
||||
const buildData = (agg: PersonAgg): PersonUpdateData => ({
|
||||
...(agg.lastContactAt
|
||||
? {
|
||||
lastContactAt: agg.lastContactAt,
|
||||
lastContactById: agg.lastContactById ?? null,
|
||||
}
|
||||
: {}),
|
||||
...(agg.lastOutboundAt ? { lastOutboundAt: agg.lastOutboundAt } : {}),
|
||||
...(agg.lastInboundAt ? { lastInboundAt: agg.lastInboundAt } : {}),
|
||||
...(agg.lastEmail ? { lastEmailId: agg.lastEmail.id } : {}),
|
||||
...(agg.lastMeeting ? { lastMeetingId: agg.lastMeeting.id } : {}),
|
||||
...(agg.item?.kind === 'email'
|
||||
? {
|
||||
lastContactItemMessageId: agg.item.id,
|
||||
lastContactItemCalendarEventId: null,
|
||||
}
|
||||
: agg.item?.kind === 'meeting'
|
||||
? {
|
||||
lastContactItemCalendarEventId: agg.item.id,
|
||||
lastContactItemMessageId: null,
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
|
||||
const applyUpdates = async (
|
||||
client: CoreApiClient,
|
||||
mutationName: string,
|
||||
updates: RecordUpdate[],
|
||||
): Promise<void> => {
|
||||
for (const batch of chunk(updates, UPDATE_BATCH_SIZE)) {
|
||||
await Promise.all(
|
||||
batch.map(({ id, data }) =>
|
||||
executeWithRetry(() =>
|
||||
client.mutation({
|
||||
[mutationName]: {
|
||||
__args: { id, data },
|
||||
id: true,
|
||||
},
|
||||
}),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
const handler = async (): Promise<void> => {
|
||||
const client = new CoreApiClient();
|
||||
|
||||
const [emails, meetings, personCompanies, opportunities] = await Promise.all([
|
||||
collectEmailInteractions(client),
|
||||
collectMeetingInteractions(client),
|
||||
collectPersonCompanies(client),
|
||||
collectOpportunities(client),
|
||||
]);
|
||||
|
||||
const messageIds = [...new Set(emails.map((email) => email.messageId))];
|
||||
const calendarEventIds = [
|
||||
...new Set(meetings.map((meeting) => meeting.calendarEventId)),
|
||||
];
|
||||
|
||||
const [messageMemberInfo, calendarOwners] = await Promise.all([
|
||||
collectMessageMemberInfo(client, messageIds),
|
||||
collectCalendarOwners(client, calendarEventIds),
|
||||
]);
|
||||
|
||||
const aggByPersonId: AggByPersonId = new Map();
|
||||
const aggFor = (personId: string): PersonAgg => {
|
||||
const existing = aggByPersonId.get(personId);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const created: PersonAgg = {};
|
||||
aggByPersonId.set(personId, created);
|
||||
return created;
|
||||
};
|
||||
|
||||
for (const email of emails) {
|
||||
foldEmail(
|
||||
aggFor(email.personId),
|
||||
email.receivedAt,
|
||||
email.messageId,
|
||||
messageMemberInfo.get(email.messageId),
|
||||
);
|
||||
}
|
||||
for (const meeting of meetings) {
|
||||
foldMeeting(
|
||||
aggFor(meeting.personId),
|
||||
meeting.startsAt,
|
||||
meeting.calendarEventId,
|
||||
calendarOwners.get(meeting.calendarEventId) ?? null,
|
||||
);
|
||||
}
|
||||
|
||||
const personUpdates = [...aggByPersonId.entries()].map(([personId, agg]) => ({
|
||||
id: personId,
|
||||
data: buildData(agg),
|
||||
}));
|
||||
|
||||
const companyLastContact = new Map<string, LastContact>();
|
||||
for (const [personId, agg] of aggByPersonId) {
|
||||
const contact = personLastContact(agg);
|
||||
if (!contact) {
|
||||
continue;
|
||||
}
|
||||
const companyId = personCompanies.get(personId);
|
||||
if (!companyId) {
|
||||
continue;
|
||||
}
|
||||
const existing = companyLastContact.get(companyId);
|
||||
if (!existing || contact.at > existing.at) {
|
||||
companyLastContact.set(companyId, contact);
|
||||
}
|
||||
}
|
||||
|
||||
const opportunityUpdates = opportunities
|
||||
.map((opportunity): RecordUpdate | undefined => {
|
||||
const pointOfContactAgg = opportunity.pointOfContactId
|
||||
? aggByPersonId.get(opportunity.pointOfContactId)
|
||||
: undefined;
|
||||
const lastContact = pointOfContactAgg ? personLastContact(pointOfContactAgg) : undefined;
|
||||
return lastContact
|
||||
? { id: opportunity.id, data: buildRelatedData(lastContact) }
|
||||
: undefined;
|
||||
})
|
||||
.filter((update): update is RecordUpdate => Boolean(update));
|
||||
|
||||
const companyUpdates: RecordUpdate[] = [...companyLastContact.entries()].map(
|
||||
([companyId, contact]) => ({
|
||||
id: companyId,
|
||||
data: buildRelatedData(contact),
|
||||
}),
|
||||
);
|
||||
|
||||
await applyUpdates(client, 'updatePerson', personUpdates);
|
||||
await applyUpdates(client, 'updateCompany', companyUpdates);
|
||||
await applyUpdates(client, 'updateOpportunity', opportunityUpdates);
|
||||
return { outcome: 'started' };
|
||||
};
|
||||
|
||||
export default definePostInstallLogicFunction({
|
||||
universalIdentifier: BACKFILL_POST_INSTALL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER,
|
||||
name: 'backfill-last-contact',
|
||||
description:
|
||||
'Fills person, company and opportunity last-contact fields from existing messages and calendar events after installation.',
|
||||
timeoutSeconds: 300,
|
||||
'Starts the sequential last-contact backfill orchestrator after installation.',
|
||||
timeoutSeconds: 60,
|
||||
shouldRunOnVersionUpgrade: true,
|
||||
handler,
|
||||
});
|
||||
|
||||
+93
@@ -0,0 +1,93 @@
|
||||
import { CoreApiClient } from 'twenty-client-sdk/core';
|
||||
import { defineLogicFunction, type RoutePayload } from 'twenty-sdk/define';
|
||||
|
||||
import {
|
||||
type BackfillBatchResult,
|
||||
BACKFILL_OPPORTUNITIES_ROUTE_PATH,
|
||||
} from 'src/constants/backfill';
|
||||
import { BACKFILL_OPPORTUNITIES_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers';
|
||||
import { getBackfillBatchSize } from 'src/utils/backfill-settings';
|
||||
import { executeWithRetry } from 'src/utils/execute-with-retry';
|
||||
import {
|
||||
buildRelatedUpdateData,
|
||||
buildPersonAggregates,
|
||||
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 client = new CoreApiClient();
|
||||
const cursor = payload.body?.cursor;
|
||||
|
||||
const { opportunities } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
opportunities: {
|
||||
__args: { first: getBackfillBatchSize(), after: cursor },
|
||||
edges: { node: { id: true, pointOfContactId: true } },
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const nodes: OpportunityNode[] = (opportunities?.edges ?? [])
|
||||
.map((edge: { node: OpportunityNode }) => edge.node)
|
||||
.filter((node: OpportunityNode) => Boolean(node.id));
|
||||
|
||||
if (nodes.length === 0) {
|
||||
return { nextCursor: null, count: 0 };
|
||||
}
|
||||
|
||||
const personIds = [
|
||||
...new Set(
|
||||
nodes
|
||||
.map((node) => node.pointOfContactId)
|
||||
.filter((id): id is string => Boolean(id)),
|
||||
),
|
||||
];
|
||||
const aggByPersonId = await buildPersonAggregates(client, personIds);
|
||||
|
||||
for (const node of nodes) {
|
||||
const lastContact = node.pointOfContactId
|
||||
? pickPersonLastContact(aggByPersonId.get(node.pointOfContactId))
|
||||
: undefined;
|
||||
|
||||
if (!lastContact) {
|
||||
continue;
|
||||
}
|
||||
|
||||
await executeWithRetry(() =>
|
||||
client.mutation({
|
||||
updateOpportunity: {
|
||||
__args: { id: node.id, data: buildRelatedUpdateData(lastContact) },
|
||||
id: true,
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const nextCursor =
|
||||
opportunities?.pageInfo.hasNextPage && opportunities.pageInfo.endCursor
|
||||
? opportunities.pageInfo.endCursor
|
||||
: null;
|
||||
|
||||
return { nextCursor, count: nodes.length };
|
||||
};
|
||||
|
||||
export default defineLogicFunction({
|
||||
universalIdentifier:
|
||||
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.',
|
||||
timeoutSeconds: 120,
|
||||
handler,
|
||||
httpRouteTriggerSettings: {
|
||||
path: BACKFILL_OPPORTUNITIES_ROUTE_PATH,
|
||||
httpMethod: 'POST',
|
||||
isAuthRequired: true,
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,79 @@
|
||||
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,
|
||||
},
|
||||
});
|
||||
+79
@@ -0,0 +1,79 @@
|
||||
import { CoreApiClient } from 'twenty-client-sdk/core';
|
||||
import { defineLogicFunction, type RoutePayload } from 'twenty-sdk/define';
|
||||
|
||||
import {
|
||||
type BackfillBatchResult,
|
||||
BACKFILL_PEOPLE_ROUTE_PATH,
|
||||
} from 'src/constants/backfill';
|
||||
import { BACKFILL_PEOPLE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers';
|
||||
import { getBackfillBatchSize } from 'src/utils/backfill-settings';
|
||||
import { executeWithRetry } from 'src/utils/execute-with-retry';
|
||||
import {
|
||||
buildPersonAggregates,
|
||||
buildPersonUpdateData,
|
||||
} from 'src/utils/person-last-contact-aggregation';
|
||||
|
||||
type BackfillBody = { cursor?: string };
|
||||
|
||||
const handler = async (
|
||||
payload: RoutePayload<BackfillBody>,
|
||||
): Promise<BackfillBatchResult> => {
|
||||
const client = new CoreApiClient();
|
||||
const cursor = payload.body?.cursor;
|
||||
|
||||
const { people } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
people: {
|
||||
__args: { first: getBackfillBatchSize(), after: cursor },
|
||||
edges: { node: { id: true } },
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const personIds = (people?.edges ?? [])
|
||||
.map((edge: { node: { id: string } }) => edge.node.id)
|
||||
.filter(Boolean);
|
||||
|
||||
if (personIds.length === 0) {
|
||||
return { nextCursor: null, count: 0 };
|
||||
}
|
||||
|
||||
const aggByPersonId = await buildPersonAggregates(client, personIds);
|
||||
|
||||
for (const personId of personIds) {
|
||||
const agg = aggByPersonId.get(personId);
|
||||
const data = agg ? buildPersonUpdateData(agg) : {};
|
||||
|
||||
if (Object.keys(data).length === 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
await executeWithRetry(() =>
|
||||
client.mutation({
|
||||
updatePerson: { __args: { id: personId, data }, id: true },
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const nextCursor =
|
||||
people?.pageInfo.hasNextPage && people.pageInfo.endCursor
|
||||
? people.pageInfo.endCursor
|
||||
: null;
|
||||
|
||||
return { nextCursor, 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.',
|
||||
timeoutSeconds: 120,
|
||||
handler,
|
||||
httpRouteTriggerSettings: {
|
||||
path: BACKFILL_PEOPLE_ROUTE_PATH,
|
||||
httpMethod: 'POST',
|
||||
isAuthRequired: true,
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,31 @@
|
||||
import {
|
||||
BACKFILL_BATCH_SIZE_ENV_VAR_NAME,
|
||||
BACKFILL_SLEEP_MS_ENV_VAR_NAME,
|
||||
DEFAULT_BACKFILL_BATCH_SIZE,
|
||||
DEFAULT_BACKFILL_SLEEP_MS,
|
||||
} from 'src/constants/backfill';
|
||||
|
||||
// Application and server variables are injected into process.env on every
|
||||
// execution.
|
||||
const readPositiveInteger = (
|
||||
envVarName: string,
|
||||
fallback: number,
|
||||
): number => {
|
||||
const rawValue = process.env[envVarName];
|
||||
|
||||
if (rawValue === undefined || rawValue.trim().length === 0) {
|
||||
return fallback;
|
||||
}
|
||||
|
||||
const parsedValue = Number(rawValue);
|
||||
|
||||
return Number.isInteger(parsedValue) && parsedValue > 0
|
||||
? parsedValue
|
||||
: fallback;
|
||||
};
|
||||
|
||||
export const getBackfillBatchSize = (): number =>
|
||||
readPositiveInteger(BACKFILL_BATCH_SIZE_ENV_VAR_NAME, DEFAULT_BACKFILL_BATCH_SIZE);
|
||||
|
||||
export const getBackfillSleepMs = (): number =>
|
||||
readPositiveInteger(BACKFILL_SLEEP_MS_ENV_VAR_NAME, DEFAULT_BACKFILL_SLEEP_MS);
|
||||
@@ -0,0 +1,47 @@
|
||||
import { type CoreApiClient } from 'twenty-client-sdk/core';
|
||||
|
||||
import { executeWithRetry } from 'src/utils/execute-with-retry';
|
||||
|
||||
const PAGE_SIZE = 200;
|
||||
|
||||
export const collectPeopleByCompany = async (
|
||||
client: CoreApiClient,
|
||||
companyIds: string[],
|
||||
): Promise<Map<string, string[]>> => {
|
||||
const peopleByCompanyId = new Map<string, string[]>();
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { people } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
people: {
|
||||
__args: {
|
||||
filter: { companyId: { in: companyIds } },
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: { node: { id: true, companyId: true } },
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of people?.edges ?? []) {
|
||||
const { id, companyId } = edge.node;
|
||||
if (id && companyId) {
|
||||
const existing = peopleByCompanyId.get(companyId);
|
||||
if (existing) {
|
||||
existing.push(id);
|
||||
} else {
|
||||
peopleByCompanyId.set(companyId, [id]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
after = people?.pageInfo.hasNextPage
|
||||
? (people.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
|
||||
return peopleByCompanyId;
|
||||
};
|
||||
+405
@@ -0,0 +1,405 @@
|
||||
import { type CoreApiClient } from 'twenty-client-sdk/core';
|
||||
|
||||
import { executeWithRetry } from 'src/utils/execute-with-retry';
|
||||
|
||||
const PAGE_SIZE = 200;
|
||||
|
||||
type EmailInteraction = {
|
||||
personId: string;
|
||||
messageId: string;
|
||||
receivedAt: string;
|
||||
};
|
||||
type MeetingInteraction = {
|
||||
personId: string;
|
||||
calendarEventId: string;
|
||||
startsAt: string;
|
||||
};
|
||||
type MessageMemberInfo = { ownerId: string; fromIsMember: boolean };
|
||||
type ContactItem = { kind: 'email' | 'meeting'; id: string };
|
||||
|
||||
export type LastContact = { at: string; item: ContactItem };
|
||||
export type PersonUpdateData = Record<string, string | null>;
|
||||
|
||||
export type PersonAgg = {
|
||||
lastContactAt?: string;
|
||||
lastContactById?: string | null;
|
||||
item?: ContactItem;
|
||||
lastOutboundAt?: string;
|
||||
lastInboundAt?: string;
|
||||
lastEmail?: { at: string; id: string };
|
||||
lastMeeting?: { at: string; id: string };
|
||||
};
|
||||
|
||||
const chunk = <T>(items: T[], size: number): T[][] => {
|
||||
const chunks: T[][] = [];
|
||||
for (let i = 0; i < items.length; i += size) {
|
||||
chunks.push(items.slice(i, i + size));
|
||||
}
|
||||
return chunks;
|
||||
};
|
||||
|
||||
const collectEmailInteractions = async (
|
||||
client: CoreApiClient,
|
||||
personIds: string[],
|
||||
): Promise<EmailInteraction[]> => {
|
||||
const interactions: EmailInteraction[] = [];
|
||||
|
||||
for (const ids of chunk(personIds, PAGE_SIZE)) {
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { messageParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
messageParticipants: {
|
||||
__args: {
|
||||
filter: { personId: { in: ids } },
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: {
|
||||
id: true,
|
||||
personId: true,
|
||||
message: { id: true, receivedAt: true },
|
||||
},
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of messageParticipants?.edges ?? []) {
|
||||
const { personId, message } = edge.node;
|
||||
if (personId && message?.id && message?.receivedAt) {
|
||||
interactions.push({
|
||||
personId,
|
||||
messageId: message.id,
|
||||
receivedAt: message.receivedAt,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
after = messageParticipants?.pageInfo.hasNextPage
|
||||
? (messageParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
}
|
||||
|
||||
return interactions;
|
||||
};
|
||||
|
||||
const collectMeetingInteractions = async (
|
||||
client: CoreApiClient,
|
||||
personIds: string[],
|
||||
): Promise<MeetingInteraction[]> => {
|
||||
const now = new Date().toISOString();
|
||||
const interactions: MeetingInteraction[] = [];
|
||||
|
||||
for (const ids of chunk(personIds, PAGE_SIZE)) {
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { calendarEventParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
calendarEventParticipants: {
|
||||
__args: {
|
||||
filter: { personId: { in: ids } },
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: {
|
||||
id: true,
|
||||
personId: true,
|
||||
calendarEvent: { id: true, startsAt: true, isCanceled: true },
|
||||
},
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of calendarEventParticipants?.edges ?? []) {
|
||||
const { personId, calendarEvent } = edge.node;
|
||||
if (
|
||||
personId &&
|
||||
calendarEvent?.id &&
|
||||
calendarEvent?.startsAt &&
|
||||
!calendarEvent.isCanceled &&
|
||||
calendarEvent.startsAt <= now
|
||||
) {
|
||||
interactions.push({
|
||||
personId,
|
||||
calendarEventId: calendarEvent.id,
|
||||
startsAt: calendarEvent.startsAt,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
after = calendarEventParticipants?.pageInfo.hasNextPage
|
||||
? (calendarEventParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
}
|
||||
|
||||
return interactions;
|
||||
};
|
||||
|
||||
const collectMessageMemberInfo = async (
|
||||
client: CoreApiClient,
|
||||
messageIds: string[],
|
||||
): Promise<Map<string, MessageMemberInfo>> => {
|
||||
const infoByMessageId = new Map<string, MessageMemberInfo>();
|
||||
|
||||
for (const ids of chunk(messageIds, PAGE_SIZE)) {
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { messageParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
messageParticipants: {
|
||||
__args: {
|
||||
filter: {
|
||||
messageId: { in: ids },
|
||||
workspaceMemberId: { is: 'NOT_NULL' },
|
||||
},
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: { messageId: true, role: true, workspaceMemberId: true },
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of messageParticipants?.edges ?? []) {
|
||||
const { messageId, role, workspaceMemberId } = edge.node;
|
||||
if (!messageId || !workspaceMemberId) {
|
||||
continue;
|
||||
}
|
||||
const info = infoByMessageId.get(messageId) ?? {
|
||||
ownerId: workspaceMemberId,
|
||||
fromIsMember: false,
|
||||
};
|
||||
if (role === 'FROM') {
|
||||
info.ownerId = workspaceMemberId;
|
||||
info.fromIsMember = true;
|
||||
}
|
||||
infoByMessageId.set(messageId, info);
|
||||
}
|
||||
|
||||
after = messageParticipants?.pageInfo.hasNextPage
|
||||
? (messageParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
}
|
||||
|
||||
return infoByMessageId;
|
||||
};
|
||||
|
||||
const collectCalendarOwners = async (
|
||||
client: CoreApiClient,
|
||||
calendarEventIds: string[],
|
||||
): Promise<Map<string, string>> => {
|
||||
const ownerByCalendarEventId = new Map<string, string>();
|
||||
|
||||
for (const ids of chunk(calendarEventIds, PAGE_SIZE)) {
|
||||
let after: string | undefined;
|
||||
|
||||
do {
|
||||
const { calendarEventParticipants } = await executeWithRetry(() =>
|
||||
client.query({
|
||||
calendarEventParticipants: {
|
||||
__args: {
|
||||
filter: {
|
||||
calendarEventId: { in: ids },
|
||||
workspaceMemberId: { is: 'NOT_NULL' },
|
||||
},
|
||||
first: PAGE_SIZE,
|
||||
after,
|
||||
},
|
||||
edges: {
|
||||
node: {
|
||||
calendarEventId: true,
|
||||
isOrganizer: true,
|
||||
workspaceMemberId: true,
|
||||
},
|
||||
},
|
||||
pageInfo: { hasNextPage: true, endCursor: true },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
for (const edge of calendarEventParticipants?.edges ?? []) {
|
||||
const { calendarEventId, isOrganizer, workspaceMemberId } = edge.node;
|
||||
if (
|
||||
calendarEventId &&
|
||||
workspaceMemberId &&
|
||||
(!ownerByCalendarEventId.has(calendarEventId) || isOrganizer === true)
|
||||
) {
|
||||
ownerByCalendarEventId.set(calendarEventId, workspaceMemberId);
|
||||
}
|
||||
}
|
||||
|
||||
after = calendarEventParticipants?.pageInfo.hasNextPage
|
||||
? (calendarEventParticipants.pageInfo.endCursor ?? undefined)
|
||||
: undefined;
|
||||
} while (after);
|
||||
}
|
||||
|
||||
return ownerByCalendarEventId;
|
||||
};
|
||||
|
||||
const foldEmail = (
|
||||
agg: PersonAgg,
|
||||
receivedAt: string,
|
||||
messageId: string,
|
||||
info: MessageMemberInfo | undefined,
|
||||
): void => {
|
||||
if (!agg.lastEmail || receivedAt > agg.lastEmail.at) {
|
||||
agg.lastEmail = { at: receivedAt, id: messageId };
|
||||
}
|
||||
if (info?.fromIsMember) {
|
||||
if (!agg.lastOutboundAt || receivedAt > agg.lastOutboundAt) {
|
||||
agg.lastOutboundAt = receivedAt;
|
||||
}
|
||||
} else if (!agg.lastInboundAt || receivedAt > agg.lastInboundAt) {
|
||||
agg.lastInboundAt = receivedAt;
|
||||
}
|
||||
if (!agg.lastContactAt || receivedAt > agg.lastContactAt) {
|
||||
agg.lastContactAt = receivedAt;
|
||||
agg.lastContactById = info?.ownerId ?? null;
|
||||
agg.item = { kind: 'email', id: messageId };
|
||||
}
|
||||
};
|
||||
|
||||
const foldMeeting = (
|
||||
agg: PersonAgg,
|
||||
startsAt: string,
|
||||
calendarEventId: string,
|
||||
ownerId: string | null,
|
||||
): void => {
|
||||
if (!agg.lastMeeting || startsAt > agg.lastMeeting.at) {
|
||||
agg.lastMeeting = { at: startsAt, id: calendarEventId };
|
||||
}
|
||||
if (!agg.lastOutboundAt || startsAt > agg.lastOutboundAt) {
|
||||
agg.lastOutboundAt = startsAt;
|
||||
}
|
||||
if (!agg.lastInboundAt || startsAt > agg.lastInboundAt) {
|
||||
agg.lastInboundAt = startsAt;
|
||||
}
|
||||
if (!agg.lastContactAt || startsAt > agg.lastContactAt) {
|
||||
agg.lastContactAt = startsAt;
|
||||
agg.lastContactById = ownerId;
|
||||
agg.item = { kind: 'meeting', id: calendarEventId };
|
||||
}
|
||||
};
|
||||
|
||||
// Aggregates every email and meeting interaction of the given people into one
|
||||
// last-contact snapshot per person, resolving the owning team member and the
|
||||
// inbound/outbound direction from the message and calendar participants.
|
||||
export const buildPersonAggregates = async (
|
||||
client: CoreApiClient,
|
||||
personIds: string[],
|
||||
): Promise<Map<string, PersonAgg>> => {
|
||||
const aggByPersonId = new Map<string, PersonAgg>();
|
||||
|
||||
if (personIds.length === 0) {
|
||||
return aggByPersonId;
|
||||
}
|
||||
|
||||
const [emails, meetings] = await Promise.all([
|
||||
collectEmailInteractions(client, personIds),
|
||||
collectMeetingInteractions(client, personIds),
|
||||
]);
|
||||
|
||||
const messageIds = [...new Set(emails.map((email) => email.messageId))];
|
||||
const calendarEventIds = [
|
||||
...new Set(meetings.map((meeting) => meeting.calendarEventId)),
|
||||
];
|
||||
|
||||
const [messageMemberInfo, calendarOwners] = await Promise.all([
|
||||
collectMessageMemberInfo(client, messageIds),
|
||||
collectCalendarOwners(client, calendarEventIds),
|
||||
]);
|
||||
|
||||
const aggFor = (personId: string): PersonAgg => {
|
||||
const existing = aggByPersonId.get(personId);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const created: PersonAgg = {};
|
||||
aggByPersonId.set(personId, created);
|
||||
return created;
|
||||
};
|
||||
|
||||
for (const email of emails) {
|
||||
foldEmail(
|
||||
aggFor(email.personId),
|
||||
email.receivedAt,
|
||||
email.messageId,
|
||||
messageMemberInfo.get(email.messageId),
|
||||
);
|
||||
}
|
||||
for (const meeting of meetings) {
|
||||
foldMeeting(
|
||||
aggFor(meeting.personId),
|
||||
meeting.startsAt,
|
||||
meeting.calendarEventId,
|
||||
calendarOwners.get(meeting.calendarEventId) ?? null,
|
||||
);
|
||||
}
|
||||
|
||||
return aggByPersonId;
|
||||
};
|
||||
|
||||
export const pickPersonLastContact = (
|
||||
agg: PersonAgg | undefined,
|
||||
): LastContact | undefined =>
|
||||
agg?.lastContactAt && agg.item
|
||||
? { at: agg.lastContactAt, item: agg.item }
|
||||
: undefined;
|
||||
|
||||
export const pickLatestLastContact = (
|
||||
contacts: LastContact[],
|
||||
): LastContact | undefined =>
|
||||
contacts.reduce<LastContact | undefined>(
|
||||
(latest, contact) =>
|
||||
!latest || contact.at > latest.at ? contact : latest,
|
||||
undefined,
|
||||
);
|
||||
|
||||
export const buildPersonUpdateData = (agg: PersonAgg): PersonUpdateData => ({
|
||||
...(agg.lastContactAt
|
||||
? {
|
||||
lastContactAt: agg.lastContactAt,
|
||||
lastContactById: agg.lastContactById ?? null,
|
||||
}
|
||||
: {}),
|
||||
...(agg.lastOutboundAt ? { lastOutboundAt: agg.lastOutboundAt } : {}),
|
||||
...(agg.lastInboundAt ? { lastInboundAt: agg.lastInboundAt } : {}),
|
||||
...(agg.lastEmail ? { lastEmailId: agg.lastEmail.id } : {}),
|
||||
...(agg.lastMeeting ? { lastMeetingId: agg.lastMeeting.id } : {}),
|
||||
...(agg.item?.kind === 'email'
|
||||
? {
|
||||
lastContactItemMessageId: agg.item.id,
|
||||
lastContactItemCalendarEventId: null,
|
||||
}
|
||||
: agg.item?.kind === 'meeting'
|
||||
? {
|
||||
lastContactItemCalendarEventId: agg.item.id,
|
||||
lastContactItemMessageId: null,
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
|
||||
export const buildRelatedUpdateData = ({
|
||||
at,
|
||||
item,
|
||||
}: LastContact): PersonUpdateData => ({
|
||||
lastContactAt: at,
|
||||
lastContactItemMessageId: item.kind === 'email' ? item.id : null,
|
||||
lastContactItemCalendarEventId: item.kind === 'meeting' ? item.id : null,
|
||||
});
|
||||
@@ -0,0 +1,52 @@
|
||||
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.23.0-alpha.2"
|
||||
twenty-sdk: "npm:2.23.0-alpha.2"
|
||||
twenty-client-sdk: "npm:^2.25.0"
|
||||
twenty-sdk: "npm:^2.25.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.23.0-alpha.2":
|
||||
version: 2.23.0-alpha.2
|
||||
resolution: "twenty-client-sdk@npm:2.23.0-alpha.2"
|
||||
"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"
|
||||
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/9bdee21c07c3a3369199289e16860213c9dde995e62bbdbce20ad0aa4490d8a664cd3c9a1c654bc6291189c8941616dda38df359828472330b15efdf9f4c089f
|
||||
checksum: 10c0/3584eb1138994213e2cedd3019bdbaec03e3a715d09e39080df15901284c1bc9e00ab48c8d6fbf482363a484c3e196afde7306dc48f5f870136c5f51156fdd93
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
"twenty-sdk@npm:2.23.0-alpha.2":
|
||||
version: 2.23.0-alpha.2
|
||||
resolution: "twenty-sdk@npm:2.23.0-alpha.2"
|
||||
"twenty-sdk@npm:^2.25.0":
|
||||
version: 2.25.0
|
||||
resolution: "twenty-sdk@npm:2.25.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.23.0-alpha.2"
|
||||
twenty-client-sdk: "npm:2.25.0"
|
||||
typescript: "npm:^5.9.3"
|
||||
uuid: "npm:^13.0.2"
|
||||
bin:
|
||||
twenty: dist/cli.cjs
|
||||
checksum: 10c0/82f820a23bc84fb45b6ca06e237b02d71cb9c5fddc98bcdaa4838e1268a7c687242ff89f429ae7c5b30c211325de8f8d349d57f18acdb8130c1d2f677c1db913
|
||||
checksum: 10c0/c74ca145c5ec35e66817602bbc6319147aea97a5d1a372c30cc325407cb2c850c0a0363eee1bd9a999c951acb4cddba96ba9813324a6f34122bcd6c6a835de31
|
||||
languageName: node
|
||||
linkType: hard
|
||||
|
||||
|
||||
Reference in New Issue
Block a user