diff --git a/packages/twenty-apps/public/last-contact/CHANGELOG.md b/packages/twenty-apps/public/last-contact/CHANGELOG.md index f02b8d778b..ab0ba30729 100644 --- a/packages/twenty-apps/public/last-contact/CHANGELOG.md +++ b/packages/twenty-apps/public/last-contact/CHANGELOG.md @@ -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 diff --git a/packages/twenty-apps/public/last-contact/package.json b/packages/twenty-apps/public/last-contact/package.json index 5cf28afce1..2bae33bf09 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.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" 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 891d82a3df..69c31c6ba6 100644 --- a/packages/twenty-apps/public/last-contact/src/application-config.ts +++ b/packages/twenty-apps/public/last-contact/src/application-config.ts @@ -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, + }, + }, }); diff --git a/packages/twenty-apps/public/last-contact/src/constants/backfill.ts b/packages/twenty-apps/public/last-contact/src/constants/backfill.ts new file mode 100644 index 0000000000..e4bfd8bbcb --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/constants/backfill.ts @@ -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 = { + 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; 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 f024a43b9f..5cb9a7db4e 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,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 = 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 new file mode 100644 index 0000000000..45709eeaf9 --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-companies-last-contact.ts @@ -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, +): Promise => { + 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, + }, +}); 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 new file mode 100644 index 0000000000..af30b60869 --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-get-state.ts @@ -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(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 be0e46710e..fc48d71029 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,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 => { + const existingState = await kv.get(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; - -type PersonUpdateData = Record; -type RecordUpdate = { id: string; data: PersonUpdateData }; - -const chunk = (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 => { - 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 => { - 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> => { - const infoByMessageId = new Map(); - - 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(BACKFILL_STATE_KV_KEY, { + phase: BACKFILL_PHASE_ORDER[0], + cursor: null, + iterations: 0, + }); -const collectCalendarOwners = async ( - client: CoreApiClient, - calendarEventIds: string[], -): Promise> => { - const ownerByCalendarEventId = new Map(); + 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> => { - const companyByPersonId = new Map(); - 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 => { - 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 => { - 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 => { - 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(); - 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, }); 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 new file mode 100644 index 0000000000..721eb5fb69 --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-opportunities-last-contact.ts @@ -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, +): Promise => { + 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, + }, +}); 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 new file mode 100644 index 0000000000..e71d4dbfba --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-orchestrator.ts @@ -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 => { + 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 new file mode 100644 index 0000000000..8e80fbfc5f --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/logic-functions/backfill-people-last-contact.ts @@ -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, +): Promise => { + 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, + }, +}); diff --git a/packages/twenty-apps/public/last-contact/src/utils/backfill-settings.ts b/packages/twenty-apps/public/last-contact/src/utils/backfill-settings.ts new file mode 100644 index 0000000000..279854857f --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/utils/backfill-settings.ts @@ -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); diff --git a/packages/twenty-apps/public/last-contact/src/utils/collect-people-by-company.ts b/packages/twenty-apps/public/last-contact/src/utils/collect-people-by-company.ts new file mode 100644 index 0000000000..735a01bc9e --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/utils/collect-people-by-company.ts @@ -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> => { + const peopleByCompanyId = new Map(); + 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; +}; diff --git a/packages/twenty-apps/public/last-contact/src/utils/person-last-contact-aggregation.ts b/packages/twenty-apps/public/last-contact/src/utils/person-last-contact-aggregation.ts new file mode 100644 index 0000000000..5b4b3faeb6 --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/utils/person-last-contact-aggregation.ts @@ -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; + +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 = (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 => { + 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 => { + 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> => { + const infoByMessageId = new Map(); + + 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> => { + const ownerByCalendarEventId = new Map(); + + 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> => { + const aggByPersonId = new Map(); + + 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( + (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, +}); 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 new file mode 100644 index 0000000000..1ff98d58f1 --- /dev/null +++ b/packages/twenty-apps/public/last-contact/src/utils/post-to-own-route.ts @@ -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 ({ + 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 3e2d29b5ef..f50cec116d 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.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