diff --git a/packages/twenty-apps/public/twenty-last-contact/CHANGELOG.md b/packages/twenty-apps/public/twenty-last-contact/CHANGELOG.md index 8c4df79ac4..12796d27a8 100644 --- a/packages/twenty-apps/public/twenty-last-contact/CHANGELOG.md +++ b/packages/twenty-apps/public/twenty-last-contact/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## 1.1.1 + +- Throttle backfill updates and retry rate-limited or transient API failures with exponential backoff, so install/upgrade no longer fails behind Cloudflare rate limiting. + ## 1.1.0 - Add last contact on Companies and Opportunities. diff --git a/packages/twenty-apps/public/twenty-last-contact/package.json b/packages/twenty-apps/public/twenty-last-contact/package.json index 58b5c98d68..cde41ac6be 100644 --- a/packages/twenty-apps/public/twenty-last-contact/package.json +++ b/packages/twenty-apps/public/twenty-last-contact/package.json @@ -1,6 +1,6 @@ { "name": "@twentyhq/last-contact", - "version": "1.1.0", + "version": "1.1.1", "license": "MIT", "engines": { "node": "^24.5.0", diff --git a/packages/twenty-apps/public/twenty-last-contact/src/logic-functions/backfill-last-contact.ts b/packages/twenty-apps/public/twenty-last-contact/src/logic-functions/backfill-last-contact.ts index cdcad1278d..be0e46710e 100644 --- a/packages/twenty-apps/public/twenty-last-contact/src/logic-functions/backfill-last-contact.ts +++ b/packages/twenty-apps/public/twenty-last-contact/src/logic-functions/backfill-last-contact.ts @@ -2,9 +2,13 @@ import { definePostInstallLogicFunction } from 'twenty-sdk/define'; import { CoreApiClient } from 'twenty-client-sdk/core'; import { BACKFILL_POST_INSTALL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/universal-identifiers'; +import { executeWithRetry } from 'src/utils/execute-with-retry'; const PAGE_SIZE = 200; -const UPDATE_BATCH_SIZE = 20; +// 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; type EmailInteraction = { personId: string; @@ -53,23 +57,25 @@ const collectEmailInteractions = async ( let after: string | undefined; do { - const { messageParticipants } = await client.query({ - messageParticipants: { - __args: { - filter: { personId: { is: 'NOT_NULL' } }, - first: PAGE_SIZE, - after, - }, - edges: { - node: { - id: true, - personId: true, - message: { id: true, receivedAt: true }, + 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 }, }, - pageInfo: { hasNextPage: true, endCursor: true }, - }, - }); + }), + ); for (const edge of messageParticipants?.edges ?? []) { const { personId, message } = edge.node; @@ -98,23 +104,25 @@ const collectMeetingInteractions = async ( let after: string | undefined; do { - const { calendarEventParticipants } = await 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 }, + 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 }, }, - pageInfo: { hasNextPage: true, endCursor: true }, - }, - }); + }), + ); for (const edge of calendarEventParticipants?.edges ?? []) { const { personId, calendarEvent } = edge.node; @@ -151,22 +159,24 @@ const collectMessageMemberInfo = async ( let after: string | undefined; do { - const { messageParticipants } = await client.query({ - messageParticipants: { - __args: { - filter: { - messageId: { in: ids }, - workspaceMemberId: { is: 'NOT_NULL' }, + const { messageParticipants } = await executeWithRetry(() => + client.query({ + messageParticipants: { + __args: { + filter: { + messageId: { in: ids }, + workspaceMemberId: { is: 'NOT_NULL' }, + }, + first: PAGE_SIZE, + after, }, - first: PAGE_SIZE, - after, + edges: { + node: { messageId: true, role: true, workspaceMemberId: true }, + }, + pageInfo: { hasNextPage: true, endCursor: true }, }, - 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; @@ -203,26 +213,28 @@ const collectCalendarOwners = async ( let after: string | undefined; do { - const { calendarEventParticipants } = await client.query({ - calendarEventParticipants: { - __args: { - filter: { - calendarEventId: { in: ids }, - workspaceMemberId: { is: 'NOT_NULL' }, + const { calendarEventParticipants } = await executeWithRetry(() => + client.query({ + calendarEventParticipants: { + __args: { + filter: { + calendarEventId: { in: ids }, + workspaceMemberId: { is: 'NOT_NULL' }, + }, + first: PAGE_SIZE, + after, }, - first: PAGE_SIZE, - after, - }, - edges: { - node: { - calendarEventId: true, - isOrganizer: true, - workspaceMemberId: true, + edges: { + node: { + calendarEventId: true, + isOrganizer: true, + workspaceMemberId: true, + }, }, + pageInfo: { hasNextPage: true, endCursor: true }, }, - pageInfo: { hasNextPage: true, endCursor: true }, - }, - }); + }), + ); for (const edge of calendarEventParticipants?.edges ?? []) { const { calendarEventId, isOrganizer, workspaceMemberId } = edge.node; @@ -251,17 +263,19 @@ const collectPersonCompanies = async ( let after: string | undefined; do { - const { people } = await client.query({ - people: { - __args: { - filter: { companyId: { is: 'NOT_NULL' } }, - first: PAGE_SIZE, - after, + 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 }, }, - edges: { node: { id: true, companyId: true } }, - pageInfo: { hasNextPage: true, endCursor: true }, - }, - }); + }), + ); for (const edge of people?.edges ?? []) { const { id, companyId } = edge.node; @@ -285,15 +299,17 @@ const collectOpportunities = async ( let after: string | undefined; do { - const { opportunities: page } = await client.query({ - opportunities: { - __args: { first: PAGE_SIZE, after }, - edges: { - node: { id: true, pointOfContactId: true }, + const { opportunities: page } = await executeWithRetry(() => + client.query({ + opportunities: { + __args: { first: PAGE_SIZE, after }, + edges: { + node: { id: true, pointOfContactId: true }, + }, + pageInfo: { hasNextPage: true, endCursor: true }, }, - pageInfo: { hasNextPage: true, endCursor: true }, - }, - }); + }), + ); for (const edge of page?.edges ?? []) { const { id, pointOfContactId } = edge.node; @@ -401,12 +417,14 @@ const applyUpdates = async ( for (const batch of chunk(updates, UPDATE_BATCH_SIZE)) { await Promise.all( batch.map(({ id, data }) => - client.mutation({ - [mutationName]: { - __args: { id, data }, - id: true, - }, - }), + executeWithRetry(() => + client.mutation({ + [mutationName]: { + __args: { id, data }, + id: true, + }, + }), + ), ), ); } diff --git a/packages/twenty-apps/public/twenty-last-contact/src/utils/__tests__/execute-with-retry.test.ts b/packages/twenty-apps/public/twenty-last-contact/src/utils/__tests__/execute-with-retry.test.ts new file mode 100644 index 0000000000..b6f2671a63 --- /dev/null +++ b/packages/twenty-apps/public/twenty-last-contact/src/utils/__tests__/execute-with-retry.test.ts @@ -0,0 +1,74 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { executeWithRetry } from 'src/utils/execute-with-retry'; + +beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0); +}); + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +describe('executeWithRetry', () => { + it('returns the result on first success', async () => { + const execute = vi.fn().mockResolvedValue('ok'); + + await expect(executeWithRetry(execute)).resolves.toBe('ok'); + expect(execute).toHaveBeenCalledTimes(1); + }); + + it('retries rate-limited requests until they succeed', async () => { + const execute = vi + .fn() + .mockRejectedValueOnce(new Error('Too Many Requests: error code 1015')) + .mockResolvedValue('ok'); + + const promise = executeWithRetry(execute); + await vi.advanceTimersByTimeAsync(2_000); + + await expect(promise).resolves.toBe('ok'); + expect(execute).toHaveBeenCalledTimes(2); + }); + + it('does not retry non-retryable errors', async () => { + const execute = vi + .fn() + .mockRejectedValue(new Error('Bad Request: invalid data')); + + await expect(executeWithRetry(execute)).rejects.toThrow('Bad Request'); + expect(execute).toHaveBeenCalledTimes(1); + }); + + it('gives up after exhausting retries', async () => { + const execute = vi + .fn() + .mockRejectedValue(new Error('Gateway time-out: origin overloaded')); + + const promise = executeWithRetry(execute); + const assertion = expect(promise).rejects.toThrow('Gateway time-out'); + await vi.advanceTimersByTimeAsync(60_000); + + await assertion; + expect(execute).toHaveBeenCalledTimes(5); + }); + + it('waits for the server-provided retry_after when it exceeds the backoff', async () => { + const execute = vi + .fn() + .mockRejectedValueOnce( + new Error('Gateway time-out: {"retryable": true, "retry_after": 10}'), + ) + .mockResolvedValue('ok'); + + const promise = executeWithRetry(execute); + await vi.advanceTimersByTimeAsync(9_000); + expect(execute).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(1_000); + await expect(promise).resolves.toBe('ok'); + expect(execute).toHaveBeenCalledTimes(2); + }); +}); diff --git a/packages/twenty-apps/public/twenty-last-contact/src/utils/execute-with-retry.ts b/packages/twenty-apps/public/twenty-last-contact/src/utils/execute-with-retry.ts new file mode 100644 index 0000000000..6da606a656 --- /dev/null +++ b/packages/twenty-apps/public/twenty-last-contact/src/utils/execute-with-retry.ts @@ -0,0 +1,51 @@ +const MAX_ATTEMPTS = 5; +const INITIAL_RETRY_DELAY_MS = 2_000; +const MAX_RETRY_DELAY_MS = 30_000; +const MAX_JITTER_MS = 1_000; + +// The client SDK surfaces HTTP failures as plain Error messages built from the +// status text and raw response body, so retryability has to be detected from +// the message text. Covers rate limiting (429, Cloudflare 1015), transient +// gateway errors (502/503/504) and network-level failures. +const RETRYABLE_ERROR_PATTERN = + /\b(429|1015|too many requests|rate ?limit\w*|502|503|504|bad gateway|gateway time-?out|service unavailable|timed? ?out|fetch failed|econnreset|econnrefused|socket hang up)\b/i; + +const sleep = (durationMs: number): Promise => + new Promise((resolve) => setTimeout(resolve, durationMs)); + +const getErrorMessage = (error: unknown): string => + error instanceof Error ? error.message : String(error); + +const isRetryableError = (error: unknown): boolean => + RETRYABLE_ERROR_PATTERN.test(getErrorMessage(error)); + +const parseRetryAfterMs = (error: unknown): number | undefined => { + const match = getErrorMessage(error).match(/"retry_after"\s*:\s*(\d+)/); + return match ? Number(match[1]) * 1_000 : undefined; +}; + +export const executeWithRetry = async ( + execute: () => TResult, +): Promise> => { + for (let attempt = 1; ; attempt += 1) { + try { + return await execute(); + } catch (error) { + if (attempt >= MAX_ATTEMPTS || !isRetryableError(error)) { + throw error; + } + + const backoffMs = Math.min( + INITIAL_RETRY_DELAY_MS * 2 ** (attempt - 1), + MAX_RETRY_DELAY_MS, + ); + const retryAfterMs = Math.min( + parseRetryAfterMs(error) ?? 0, + MAX_RETRY_DELAY_MS, + ); + const jitterMs = Math.random() * MAX_JITTER_MS; + + await sleep(Math.max(backoffMs, retryAfterMs) + jitterMs); + } + } +};