From d4e0b78d20c3c54d83008f65ac3c00b48ed03ebf Mon Sep 17 00:00:00 2001 From: nitin <142569587+ehconitin@users.noreply.github.com> Date: Wed, 17 Jun 2026 21:12:03 +0530 Subject: [PATCH] Add stale Recall bot reconciliation (#21720) This PR adds a scheduled reconciliation pass for the Twenty Meeting Bot app so call recording state does not depend only on event-driven updates from calendar changes and Recall webhooks. Why we need this -- - A call recording row can be created, but the process can fail before the Recall bot id is written back. - Recall webhooks can be missed or delivered late, leaving Twenty stuck in an older local state. - A bot can disappear from Recall, leaving Twenty with a stale externalBotId. - A cancellation can fail locally, leaving an app-managed Recall bot that would still join the meeting. What this adds -- - A cron logic function that heals botless scheduled call recordings. - A convergence pass that pulls Recall bot state for stale local rows, including SCHEDULED rows. - Orphaned bot cleanup for app-managed Recall bots that are no longer claimed by an open call recording. - Guards so destructive bot cleanup does not affect bots claimed by another app registration. - Tests for the new stale-state, missed-webhook, and orphan cleanup behavior. Not included -- - Media ingestion. - Transcript pipeline. - Billing. - Marking call recordings as COMPLETED. Review in cubic --- ...ate-logic-function-universal-identifier.ts | 2 + .../constants/stale-bot-state-cron-pattern.ts | 1 + ...ind-open-scheduled-call-recordings.util.ts | 14 + .../converge-diverged-call-recordings.test.ts | 323 ++++++++++++++ .../heal-call-recordings-missing-bot.test.ts | 210 +++++++++ .../reap-orphaned-meeting-bots.test.ts | 410 ++++++++++++++++++ .../__tests__/reconcile-meeting-bot.test.ts | 16 +- .../cancel-call-recording-request.util.ts | 4 +- .../converge-diverged-call-recordings.util.ts | 323 ++++++++++++++ .../flows/ensure-meeting-bot.util.ts | 13 +- .../heal-call-recordings-missing-bot.util.ts | 82 ++++ .../flows/reap-orphaned-meeting-bots.util.ts | 182 ++++++++ .../flows/reconcile-meeting-bot.util.ts | 4 - .../reschedule-call-recording-bot.util.ts | 2 +- .../extract-recall-bot-convergence.test.ts | 129 ++++++ .../__tests__/recall-bot-api.test.ts | 284 ++++++++++-- .../recall-api/eject-recall-bot.util.ts | 28 ++ .../extract-recall-bot-convergence.util.ts | 137 ++++++ .../recall-api/get-recall-bot.util.ts | 42 ++ .../list-scheduled-recall-bots.util.ts | 106 +++++ .../reconcile-stale-bot-state.ts | 111 +++++ 21 files changed, 2359 insertions(+), 64 deletions(-) create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/constants/stale-bot-state-logic-function-universal-identifier.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/constants/stale-bot-state-cron-pattern.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/data/find-open-scheduled-call-recordings.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/converge-diverged-call-recordings.test.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/heal-call-recordings-missing-bot.test.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reap-orphaned-meeting-bots.test.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/converge-diverged-call-recordings.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/heal-call-recordings-missing-bot.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reap-orphaned-meeting-bots.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/extract-recall-bot-convergence.test.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/eject-recall-bot.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/extract-recall-bot-convergence.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/get-recall-bot.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/list-scheduled-recall-bots.util.ts create mode 100644 packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/reconcile-stale-bot-state.ts diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/constants/stale-bot-state-logic-function-universal-identifier.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/constants/stale-bot-state-logic-function-universal-identifier.ts new file mode 100644 index 0000000000..79ea9ad293 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/constants/stale-bot-state-logic-function-universal-identifier.ts @@ -0,0 +1,2 @@ +export const STALE_BOT_STATE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER = + 'e362aa9b-52c6-4b7e-bb20-927e0e8d7cbe'; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/constants/stale-bot-state-cron-pattern.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/constants/stale-bot-state-cron-pattern.ts new file mode 100644 index 0000000000..0133de3231 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/constants/stale-bot-state-cron-pattern.ts @@ -0,0 +1 @@ +export const STALE_BOT_STATE_CRON_PATTERN = '*/15 * * * *'; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/data/find-open-scheduled-call-recordings.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/data/find-open-scheduled-call-recordings.util.ts new file mode 100644 index 0000000000..ef8571624a --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/data/find-open-scheduled-call-recordings.util.ts @@ -0,0 +1,14 @@ +import { CoreApiClient } from 'twenty-client-sdk/core'; + +import { CallRecordingRequestStatus } from 'src/logic-functions/constants/call-recording-request-status'; +import { CallRecordingStatus } from 'src/logic-functions/constants/call-recording-status'; +import { type CallRecordingRecord } from 'src/logic-functions/types/call-recording-record.type'; +import { findCallRecordingsByFilter } from 'src/logic-functions/data/find-call-recordings-by-filter.util'; + +export const findOpenScheduledCallRecordings = async ( + client: CoreApiClient, +): Promise => + findCallRecordingsByFilter(client, { + recordingRequestStatus: { eq: CallRecordingRequestStatus.REQUESTED }, + status: { eq: CallRecordingStatus.SCHEDULED }, + }); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/converge-diverged-call-recordings.test.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/converge-diverged-call-recordings.test.ts new file mode 100644 index 0000000000..7cae8ae55a --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/converge-diverged-call-recordings.test.ts @@ -0,0 +1,323 @@ +import { type CoreApiClient } from 'twenty-client-sdk/core'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { convergeDivergedCallRecordings } from 'src/logic-functions/flows/converge-diverged-call-recordings.util'; + +const getRecallBotMock = vi.hoisted(() => vi.fn()); + +vi.mock('src/logic-functions/recall-api/get-recall-bot.util', () => ({ + getRecallBot: getRecallBotMock, +})); + +const NOW = new Date('2026-06-10T12:00:00.000Z'); + +type CallRecordingNode = Record; + +class FakeCoreApiClient { + mutations: Array<{ id: string; data: Record }> = []; + + constructor(private callRecordingNodes: CallRecordingNode[]) {} + + async query(query: any): Promise { + const callRecordingFilter = query.callRecordings.__args.filter + .or[0] as Record; + const requestedRecordingRequestStatus = + callRecordingFilter.recordingRequestStatus.eq; + const requestedCallRecordingStatuses = callRecordingFilter.status.in; + const requiresExternalBotId = + callRecordingFilter.externalBotId.is === 'NOT_NULL'; + const matchingCallRecordingNodes = this.callRecordingNodes.filter( + (callRecordingNode) => + callRecordingNode.recordingRequestStatus === + requestedRecordingRequestStatus && + requestedCallRecordingStatuses.includes(callRecordingNode.status) && + (!requiresExternalBotId || + (callRecordingNode.externalBotId !== null && + callRecordingNode.externalBotId !== undefined)), + ); + + return { + callRecordings: { + pageInfo: { hasNextPage: false, endCursor: undefined }, + edges: matchingCallRecordingNodes.map((node) => ({ node })), + }, + }; + } + + async mutation(mutation: any): Promise { + const { id, data } = mutation.updateCallRecording.__args; + + this.mutations.push({ id, data }); + + return { updateCallRecording: { id } }; + } +} + +const buildClient = (callRecordingNodes: CallRecordingNode[]) => + new FakeCoreApiClient(callRecordingNodes); + +const buildStuckRecordingNode = ( + overrides: CallRecordingNode = {}, +): CallRecordingNode => ({ + id: 'call-recording-1', + status: 'RECORDING', + recordingRequestStatus: 'REQUESTED', + startedAt: null, + endedAt: null, + externalBotId: 'recall-bot-1', + externalRecordingId: null, + createdAt: '2026-06-09T12:00:00.000Z', + calendarEvent: { endsAt: '2026-06-09T13:00:00.000Z' }, + ...overrides, +}); + +describe('convergeDivergedCallRecordings', () => { + beforeEach(() => { + vi.spyOn(console, 'warn').mockImplementation(() => {}); + getRecallBotMock.mockReset(); + }); + + it('fills timestamps and recording id for a stuck RECORDING record from the Recall bot state', async () => { + getRecallBotMock.mockResolvedValue({ + ok: true, + bot: { + status_changes: [ + { code: 'in_call_recording', created_at: '2026-06-09T13:02:30.000Z' }, + { code: 'call_ended', created_at: '2026-06-09T14:00:30.000Z' }, + { code: 'done', created_at: '2026-06-09T14:05:00.000Z' }, + ], + recordings: [ + { + id: 'recall-recording-1', + started_at: '2026-06-09T13:02:00.000Z', + completed_at: '2026-06-09T14:00:00.000Z', + }, + ], + }, + }); + const client = buildClient([buildStuckRecordingNode()]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(getRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'recall-bot-1', + }); + expect(client.mutations).toEqual([ + { + id: 'call-recording-1', + data: { + status: 'PROCESSING', + startedAt: '2026-06-09T13:02:00.000Z', + endedAt: '2026-06-09T14:00:00.000Z', + externalRecordingId: 'recall-recording-1', + }, + }, + ]); + expect(result).toEqual({ + candidateCount: 1, + updatedCallRecordingIds: ['call-recording-1'], + markedFailedCallRecordingIds: [], + unconvergeableCallRecordingIds: [], + }); + }); + + it('converges a SCHEDULED record when Recall moved forward but webhooks were missed', async () => { + getRecallBotMock.mockResolvedValue({ + ok: true, + bot: { + status_changes: [ + { code: 'joining_call', created_at: '2026-06-09T13:01:00.000Z' }, + { code: 'in_call_recording', created_at: '2026-06-09T13:02:00.000Z' }, + ], + }, + }); + const client = buildClient([ + buildStuckRecordingNode({ status: 'SCHEDULED' }), + ]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(getRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'recall-bot-1', + }); + expect(client.mutations).toEqual([ + { + id: 'call-recording-1', + data: { + status: 'RECORDING', + startedAt: '2026-06-09T13:02:00.000Z', + }, + }, + ]); + expect(result.updatedCallRecordingIds).toEqual(['call-recording-1']); + }); + + it('skips records whose meeting may still be live', async () => { + const client = buildClient([ + buildStuckRecordingNode({ + calendarEvent: { endsAt: '2026-06-10T11:50:00.000Z' }, + }), + ]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(getRecallBotMock).not.toHaveBeenCalled(); + expect(client.mutations).toEqual([]); + expect(result.candidateCount).toBe(1); + }); + + it('marks FAILED_UNKNOWN without clearing the bot id when Recall returns 404', async () => { + getRecallBotMock.mockResolvedValue({ + ok: false, + status: 404, + errorMessage: 'Recall API responded with HTTP 404', + }); + const client = buildClient([buildStuckRecordingNode()]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(client.mutations).toEqual([ + { + id: 'call-recording-1', + data: { status: 'FAILED_UNKNOWN' }, + }, + ]); + expect(result.markedFailedCallRecordingIds).toEqual(['call-recording-1']); + expect(console.warn).toHaveBeenCalled(); + }); + + it('does not select COMPLETED records as convergence candidates', async () => { + getRecallBotMock.mockResolvedValue({ + ok: false, + status: 404, + errorMessage: 'Recall API responded with HTTP 404', + }); + const client = buildClient([ + buildStuckRecordingNode({ + status: 'COMPLETED', + startedAt: '2026-06-09T13:02:00.000Z', + }), + ]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(client.mutations).toEqual([]); + expect(result.candidateCount).toBe(0); + expect(result.unconvergeableCallRecordingIds).toEqual([]); + expect(getRecallBotMock).not.toHaveBeenCalled(); + }); + + it('logs candidates whose meeting ended before the lookback bound instead of converging them', async () => { + const client = buildClient([ + buildStuckRecordingNode({ + calendarEvent: { endsAt: '2026-06-01T13:00:00.000Z' }, + }), + ]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(getRecallBotMock).not.toHaveBeenCalled(); + expect(client.mutations).toEqual([]); + expect(result.unconvergeableCallRecordingIds).toEqual(['call-recording-1']); + expect(console.warn).toHaveBeenCalled(); + }); + + it('converges candidates created long before a recently ended meeting', async () => { + getRecallBotMock.mockResolvedValue({ + ok: true, + bot: { + status_changes: [ + { code: 'in_call_recording', created_at: '2026-06-09T13:02:00.000Z' }, + ], + }, + }); + const client = buildClient([ + buildStuckRecordingNode({ + createdAt: '2026-06-01T12:00:00.000Z', + startedAt: '2026-06-09T13:02:00.000Z', + }), + ]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(getRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'recall-bot-1', + }); + expect(result.unconvergeableCallRecordingIds).toEqual([]); + }); + + it('applies the downgrade guard to pulled statuses while still filling timestamps', async () => { + getRecallBotMock.mockResolvedValue({ + ok: true, + bot: { + status_changes: [ + { code: 'in_call_recording', created_at: '2026-06-09T13:02:00.000Z' }, + ], + recordings: [ + { id: 'recall-recording-1', started_at: '2026-06-09T13:02:00.000Z' }, + ], + }, + }); + const client = buildClient([ + buildStuckRecordingNode({ status: 'PROCESSING' }), + ]); + + await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(client.mutations).toEqual([ + { + id: 'call-recording-1', + data: { + startedAt: '2026-06-09T13:02:00.000Z', + externalRecordingId: 'recall-recording-1', + }, + }, + ]); + }); + + it('does not mutate a record the bot state agrees with', async () => { + getRecallBotMock.mockResolvedValue({ + ok: true, + bot: { + status_changes: [ + { code: 'in_call_recording', created_at: '2026-06-09T13:02:00.000Z' }, + ], + }, + }); + const client = buildClient([ + buildStuckRecordingNode({ startedAt: '2026-06-09T13:02:00.000Z' }), + ]); + + const result = await convergeDivergedCallRecordings({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(client.mutations).toEqual([]); + expect(result.updatedCallRecordingIds).toEqual([]); + }); +}); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/heal-call-recordings-missing-bot.test.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/heal-call-recordings-missing-bot.test.ts new file mode 100644 index 0000000000..15f95362b1 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/heal-call-recordings-missing-bot.test.ts @@ -0,0 +1,210 @@ +import { type CoreApiClient } from 'twenty-client-sdk/core'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { healCallRecordingsMissingBot } from 'src/logic-functions/flows/heal-call-recordings-missing-bot.util'; + +const scheduleRecallBotMock = vi.hoisted(() => vi.fn()); + +vi.mock('src/logic-functions/recall-api/schedule-recall-bot.util', () => ({ + scheduleRecallBot: scheduleRecallBotMock, +})); + +const NOW = new Date('2026-01-01T12:00:00.000Z'); +const UPCOMING_STARTS_AT = '2026-01-01T13:00:00.000Z'; +const UPCOMING_ENDS_AT = '2026-01-01T14:00:00.000Z'; +const PAST_STARTS_AT = '2026-01-01T10:00:00.000Z'; +const PAST_ENDS_AT = '2026-01-01T11:00:00.000Z'; + +type CallRecordingNode = { + id: string; + status?: string; + recordingRequestStatus?: string | null; + calendarEventId?: string | null; + externalBotId?: string | null; +}; + +type CalendarEventNode = { + id: string; + startsAt?: string | null; + endsAt?: string | null; + iCalUid?: string | null; + conferenceLink?: { primaryLinkUrl?: string | null } | null; +}; + +class FakeCoreApiClient { + callRecordings: CallRecordingNode[]; + calendarEvents: CalendarEventNode[]; + + constructor({ + callRecordings = [], + calendarEvents = [], + }: { + callRecordings?: CallRecordingNode[]; + calendarEvents?: CalendarEventNode[]; + }) { + this.callRecordings = callRecordings; + this.calendarEvents = calendarEvents; + } + + async query(query: any): Promise { + if (query.callRecordings !== undefined) { + const filter = query.callRecordings.__args.filter; + const matches = + filter.id?.in !== undefined + ? this.callRecordings.filter((callRecording) => + filter.id.in.includes(callRecording.id), + ) + : this.callRecordings.filter( + (callRecording) => + callRecording.recordingRequestStatus === + filter.recordingRequestStatus.eq && + callRecording.status === filter.status.eq, + ); + + return { callRecordings: buildConnection(matches) }; + } + + if (query.calendarEvents !== undefined) { + const calendarEventIds = query.calendarEvents.__args.filter.id.in; + + return { + calendarEvents: buildConnection( + this.calendarEvents.filter((calendarEvent) => + calendarEventIds.includes(calendarEvent.id), + ), + ), + }; + } + + throw new Error(`Unhandled query: ${JSON.stringify(query)}`); + } + + async mutation(mutation: any): Promise { + if (mutation.updateCallRecording !== undefined) { + const { id, data } = mutation.updateCallRecording.__args; + const callRecording = this.callRecordings.find( + (candidate) => candidate.id === id, + ); + + if (callRecording !== undefined) { + Object.assign(callRecording, data); + } + + return { updateCallRecording: { id } }; + } + + throw new Error(`Unhandled mutation: ${JSON.stringify(mutation)}`); + } +} + +const buildConnection = (nodes: Node[]) => ({ + pageInfo: { hasNextPage: false, endCursor: undefined }, + edges: nodes.map((node) => ({ node })), +}); + +const buildBotlessCallRecording = ( + overrides: Partial = {}, +): CallRecordingNode => ({ + id: 'call-recording-1', + status: 'SCHEDULED', + recordingRequestStatus: 'REQUESTED', + calendarEventId: 'calendar-event-1', + externalBotId: null, + ...overrides, +}); + +const buildCalendarEvent = ( + overrides: Partial = {}, +): CalendarEventNode => ({ + id: 'calendar-event-1', + startsAt: UPCOMING_STARTS_AT, + endsAt: UPCOMING_ENDS_AT, + iCalUid: 'calendar-event-uid', + conferenceLink: { primaryLinkUrl: 'https://meet.example.com/customer-sync' }, + ...overrides, +}); + +describe('healCallRecordingsMissingBot', () => { + beforeEach(() => { + vi.spyOn(console, 'warn').mockImplementation(() => {}); + scheduleRecallBotMock.mockReset(); + scheduleRecallBotMock.mockResolvedValue({ + ok: true, + externalBotId: 'recall-bot-1', + }); + }); + + it('schedules a bot and writes the id for an upcoming botless recording', async () => { + const client = new FakeCoreApiClient({ + callRecordings: [buildBotlessCallRecording()], + calendarEvents: [buildCalendarEvent()], + }); + + const result = await healCallRecordingsMissingBot({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(result.scheduledCallRecordingIds).toEqual(['call-recording-1']); + expect(scheduleRecallBotMock).toHaveBeenCalledTimes(1); + expect(client.callRecordings[0].externalBotId).toBe('recall-bot-1'); + }); + + it('does not report a recording as scheduled when Recall scheduling fails', async () => { + scheduleRecallBotMock.mockResolvedValue({ + ok: false, + status: 500, + errorMessage: 'Recall API responded with HTTP 500', + }); + const client = new FakeCoreApiClient({ + callRecordings: [buildBotlessCallRecording()], + calendarEvents: [buildCalendarEvent()], + }); + + const result = await healCallRecordingsMissingBot({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(result.scheduledCallRecordingIds).toEqual([]); + expect(scheduleRecallBotMock).toHaveBeenCalledTimes(1); + expect(client.callRecordings[0].externalBotId).toBeNull(); + }); + + it('skips a recording whose meeting has already ended', async () => { + const client = new FakeCoreApiClient({ + callRecordings: [buildBotlessCallRecording()], + calendarEvents: [ + buildCalendarEvent({ + startsAt: PAST_STARTS_AT, + endsAt: PAST_ENDS_AT, + }), + ], + }); + + const result = await healCallRecordingsMissingBot({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(result.scheduledCallRecordingIds).toEqual([]); + expect(scheduleRecallBotMock).not.toHaveBeenCalled(); + }); + + it('does nothing when every scheduled recording already has a bot', async () => { + const client = new FakeCoreApiClient({ + callRecordings: [ + buildBotlessCallRecording({ externalBotId: 'recall-bot-existing' }), + ], + calendarEvents: [buildCalendarEvent()], + }); + + const result = await healCallRecordingsMissingBot({ + client: client as unknown as CoreApiClient, + now: NOW, + }); + + expect(result.scheduledCallRecordingIds).toEqual([]); + expect(scheduleRecallBotMock).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reap-orphaned-meeting-bots.test.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reap-orphaned-meeting-bots.test.ts new file mode 100644 index 0000000000..d654b29564 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reap-orphaned-meeting-bots.test.ts @@ -0,0 +1,410 @@ +import { type CoreApiClient } from 'twenty-client-sdk/core'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { APPLICATION_ID_ENV_VAR_NAME } from 'src/logic-functions/constants/application-id-env-var-name'; +import { reapOrphanedMeetingBots } from 'src/logic-functions/flows/reap-orphaned-meeting-bots.util'; + +const listScheduledRecallBotsMock = vi.hoisted(() => vi.fn()); +const cancelRecallBotMock = vi.hoisted(() => vi.fn()); +const ejectRecallBotMock = vi.hoisted(() => vi.fn()); + +vi.mock( + 'src/logic-functions/recall-api/list-scheduled-recall-bots.util', + () => ({ + listScheduledRecallBots: listScheduledRecallBotsMock, + }), +); + +vi.mock('src/logic-functions/recall-api/cancel-recall-bot.util', () => ({ + cancelRecallBot: cancelRecallBotMock, +})); + +vi.mock('src/logic-functions/recall-api/eject-recall-bot.util', () => ({ + ejectRecallBot: ejectRecallBotMock, +})); + +const JOIN_AT_AFTER = '2026-01-01T08:00:00.000Z'; +const JOIN_AT_BEFORE = '2026-01-02T12:00:00.000Z'; +const CURRENT_APPLICATION_ID = 'current-application-id'; +const ORIGINAL_APPLICATION_ID = process.env[APPLICATION_ID_ENV_VAR_NAME]; + +type CallRecordingNode = { + id: string; + recordingRequestStatus?: string | null; + externalBotId?: string | null; +}; + +class FakeCoreApiClient { + constructor(private callRecordings: CallRecordingNode[]) {} + + async query(query: any): Promise { + const callRecordingIds = query.callRecordings.__args.filter.id.in; + + return { + callRecordings: { + pageInfo: { hasNextPage: false, endCursor: undefined }, + edges: this.callRecordings + .filter((callRecording) => + callRecordingIds.includes(callRecording.id), + ) + .map((node) => ({ node })), + }, + }; + } +} + +const buildClient = (callRecordings: CallRecordingNode[]): CoreApiClient => + new FakeCoreApiClient(callRecordings) as unknown as CoreApiClient; + +const restoreOriginalApplicationId = () => { + if (ORIGINAL_APPLICATION_ID === undefined) { + delete process.env[APPLICATION_ID_ENV_VAR_NAME]; + + return; + } + + process.env[APPLICATION_ID_ENV_VAR_NAME] = ORIGINAL_APPLICATION_ID; +}; + +const buildBot = ({ + id, + twentyCallRecordingId, + twentyApplicationId, +}: { + id: string; + twentyCallRecordingId?: string; + twentyApplicationId?: string; +}) => ({ + id, + metadata: { + ...(twentyCallRecordingId === undefined ? {} : { twentyCallRecordingId }), + ...(twentyApplicationId === undefined ? {} : { twentyApplicationId }), + }, +}); + +const buildCurrentApplicationBot = ({ + id, + twentyCallRecordingId, +}: { + id: string; + twentyCallRecordingId: string; +}) => + buildBot({ + id, + twentyCallRecordingId, + twentyApplicationId: CURRENT_APPLICATION_ID, + }); + +describe('reapOrphanedMeetingBots', () => { + beforeEach(() => { + restoreOriginalApplicationId(); + process.env[APPLICATION_ID_ENV_VAR_NAME] = CURRENT_APPLICATION_ID; + vi.spyOn(console, 'warn').mockImplementation(() => {}); + listScheduledRecallBotsMock.mockReset(); + cancelRecallBotMock.mockReset(); + cancelRecallBotMock.mockResolvedValue({ ok: true }); + ejectRecallBotMock.mockReset(); + ejectRecallBotMock.mockResolvedValue({ ok: true }); + }); + + afterEach(() => { + restoreOriginalApplicationId(); + }); + + it('keeps bots that their call recording still references', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'claimed-bot', + twentyCallRecordingId: 'call-recording-1', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([ + { + id: 'call-recording-1', + recordingRequestStatus: 'REQUESTED', + externalBotId: 'claimed-bot', + }, + ]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: [], + }); + expect(cancelRecallBotMock).not.toHaveBeenCalled(); + }); + + it('cancels bots whose call recording request was canceled locally', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'stale-cancel-bot', + twentyCallRecordingId: 'call-recording-1', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([ + { + id: 'call-recording-1', + recordingRequestStatus: 'CANCELED', + externalBotId: 'stale-cancel-bot', + }, + ]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: ['stale-cancel-bot'], + }); + expect(cancelRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'stale-cancel-bot', + }); + }); + + it('cancels bots whose call recording references another bot', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'superseded-bot', + twentyCallRecordingId: 'call-recording-1', + }), + buildCurrentApplicationBot({ + id: 'claimed-bot', + twentyCallRecordingId: 'call-recording-1', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([ + { + id: 'call-recording-1', + recordingRequestStatus: 'REQUESTED', + externalBotId: 'claimed-bot', + }, + ]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 2, + canceledExternalBotIds: ['superseded-bot'], + }); + expect(cancelRecallBotMock).toHaveBeenCalledTimes(1); + expect(cancelRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'superseded-bot', + }); + }); + + it('cancels bots whose call recording no longer exists', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'orphan-bot', + twentyCallRecordingId: 'call-recording-gone', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: ['orphan-bot'], + }); + }); + + it('grants a grace round to requested recordings without a bot id yet', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'pending-bot', + twentyCallRecordingId: 'call-recording-1', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([ + { + id: 'call-recording-1', + recordingRequestStatus: 'REQUESTED', + externalBotId: null, + }, + ]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: [], + }); + expect(cancelRecallBotMock).not.toHaveBeenCalled(); + }); + + it('ignores bots that were not created by this app', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [buildBot({ id: 'unrelated-bot' })], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: [], + }); + expect(cancelRecallBotMock).not.toHaveBeenCalled(); + }); + + it('ignores untagged bots even when they carry call recording metadata', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildBot({ + id: 'untagged-bot', + twentyCallRecordingId: 'call-recording-gone', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: [], + }); + expect(cancelRecallBotMock).not.toHaveBeenCalled(); + }); + + it('ignores bots claimed by another application registration', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildBot({ + id: 'other-app-bot', + twentyCallRecordingId: 'call-recording-gone', + twentyApplicationId: 'other-application-id', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: [], + }); + expect(cancelRecallBotMock).not.toHaveBeenCalled(); + }); + + it('cancels orphaned bots claimed by this application registration', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'same-app-bot', + twentyCallRecordingId: 'call-recording-gone', + }), + ], + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: ['same-app-bot'], + }); + expect(cancelRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'same-app-bot', + }); + }); + + it('ejects an orphaned bot that already joined when deletion is rejected', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: true, + bots: [ + buildCurrentApplicationBot({ + id: 'in-call-orphan', + twentyCallRecordingId: 'call-recording-gone', + }), + ], + }); + cancelRecallBotMock.mockResolvedValue({ + ok: false, + status: 409, + errorMessage: 'Recall API responded with HTTP 409', + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 1, + canceledExternalBotIds: ['in-call-orphan'], + }); + expect(ejectRecallBotMock).toHaveBeenCalledWith({ + externalBotId: 'in-call-orphan', + }); + }); + + it('reports nothing reaped when listing Recall bots fails', async () => { + listScheduledRecallBotsMock.mockResolvedValue({ + ok: false, + status: 500, + errorMessage: 'Recall API responded with HTTP 500', + }); + + const result = await reapOrphanedMeetingBots({ + client: buildClient([]), + joinAtAfter: JOIN_AT_AFTER, + joinAtBefore: JOIN_AT_BEFORE, + }); + + expect(result).toEqual({ + scannedBotCount: 0, + canceledExternalBotIds: [], + }); + expect(cancelRecallBotMock).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reconcile-meeting-bot.test.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reconcile-meeting-bot.test.ts index 43270c423e..fd817d45da 100644 --- a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reconcile-meeting-bot.test.ts +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/__tests__/reconcile-meeting-bot.test.ts @@ -821,7 +821,7 @@ describe('reconcileMeetingBotForCalendarEventIds', () => { ]); }); - it('recreates the bot when the existing Recall bot no longer exists', async () => { + it('clears the stale bot id for the stale-state cron to re-create when the existing Recall bot no longer exists', async () => { rescheduleRecallBotMock.mockResolvedValue({ ok: false, status: 404, @@ -859,20 +859,12 @@ describe('reconcileMeetingBotForCalendarEventIds', () => { expect(rescheduleRecallBotMock).toHaveBeenCalledWith( expect.objectContaining({ externalBotId: 'recall-bot-stale' }), ); - expect(scheduleRecallBotMock).toHaveBeenCalledWith({ - meetingUrl: 'https://meet.example.com/customer-sync', - joinAt: FUTURE_STARTS_AT, - metadata: { - twentyCallRecordingId: buildCustomerSyncCallRecordingId(), - twentyCalendarEventId: 'calendar-event-1', - twentyRealMeetingKey: - 'link:meet.example.com/customer-sync:2026-01-01T13:00:00.000Z', - }, - }); + // The event path no longer re-creates the bot; the stale id is cleared and the cron heals the botless row. + expect(scheduleRecallBotMock).not.toHaveBeenCalled(); expect(client.callRecordings).toEqual([ expect.objectContaining({ id: buildCustomerSyncCallRecordingId(), - externalBotId: 'recall-bot-1', + externalBotId: null, }), ]); }); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/cancel-call-recording-request.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/cancel-call-recording-request.util.ts index 046fdb892e..700421d9f0 100644 --- a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/cancel-call-recording-request.util.ts +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/cancel-call-recording-request.util.ts @@ -6,7 +6,7 @@ import { type CallRecordingRecord } from 'src/logic-functions/types/call-recordi import { cancelRecallBot } from 'src/logic-functions/recall-api/cancel-recall-bot.util'; import { updateCallRecording } from 'src/logic-functions/data/update-call-recording.util'; -// TODO: Add the stale-state cron in the next split so it can finish the Recall half when this call fails. +// Intent-first: the stale-state cron finishes the Recall half when this call fails. export const cancelCallRecordingRequest = async ({ client, callRecording, @@ -31,7 +31,7 @@ export const cancelCallRecordingRequest = async ({ if (!cancelResult.ok) { console.warn( - `[twenty-meeting-bot] failed to cancel Recall bot for callRecording ${callRecording.id}, leaving it for the planned stale-state cron: ${cancelResult.errorMessage}`, + `[twenty-meeting-bot] failed to cancel Recall bot for callRecording ${callRecording.id}, leaving it for the stale-state cron: ${cancelResult.errorMessage}`, ); return; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/converge-diverged-call-recordings.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/converge-diverged-call-recordings.util.ts new file mode 100644 index 0000000000..f78edc4a0b --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/converge-diverged-call-recordings.util.ts @@ -0,0 +1,323 @@ +import { isUndefined } from '@sniptt/guards'; +import { CoreApiClient } from 'twenty-client-sdk/core'; + +import { CallRecordingRequestStatus } from 'src/logic-functions/constants/call-recording-request-status'; +import { CallRecordingStatus } from 'src/logic-functions/constants/call-recording-status'; +import { TWENTY_PAGE_SIZE } from 'src/logic-functions/constants/twenty-page-size'; +import { + extractRecallBotConvergence, + type RecallBotConvergence, +} from 'src/logic-functions/recall-api/extract-recall-bot-convergence.util'; +import { + fetchAllNodes, + type ConnectionPage, +} from 'src/logic-functions/data/fetch-all-nodes.util'; +import { getRecallBot } from 'src/logic-functions/recall-api/get-recall-bot.util'; +import { isCallRecordingStatusDowngrade } from 'src/logic-functions/domain/is-call-recording-status-downgrade.util'; +import { isNonEmptyString } from 'src/logic-functions/utils/is-non-empty-string.util'; +import { + updateCallRecording, + type CallRecordingUpdateFields, +} from 'src/logic-functions/data/update-call-recording.util'; + +const CONVERGENCE_LOOKBACK_DAYS = 7; +const LIVE_MEETING_GRACE_MINUTES = 30; + +const NON_TERMINAL_CALL_RECORDING_STATUSES = [ + CallRecordingStatus.SCHEDULED, + CallRecordingStatus.JOINING, + CallRecordingStatus.RECORDING, + CallRecordingStatus.PROCESSING, +]; + +type DivergedCallRecordingCandidate = { + id: string; + status: string | undefined; + startedAt: string | undefined; + endedAt: string | undefined; + externalBotId: string | undefined; + externalRecordingId: string | undefined; + createdAt: string | undefined; + calendarEventEndsAt: string | undefined; +}; + +type DivergedCallRecordingNode = { + id: string; + status?: string | null; + startedAt?: string | null; + endedAt?: string | null; + externalBotId?: string | null; + externalRecordingId?: string | null; + createdAt?: string | null; + calendarEvent?: { endsAt?: string | null } | null; +}; + +export type ConvergeDivergedCallRecordingsResult = { + candidateCount: number; + updatedCallRecordingIds: string[]; + markedFailedCallRecordingIds: string[]; + unconvergeableCallRecordingIds: string[]; +}; + +// Webhook deliveries get lost; this pull pass re-derives state from Recall. +export const convergeDivergedCallRecordings = async ({ + client, + now, +}: { + client: CoreApiClient; + now: Date; +}): Promise => { + const candidates = await fetchDivergedCallRecordingCandidates(client); + const convergenceLowerBound = new Date( + now.getTime() - CONVERGENCE_LOOKBACK_DAYS * 24 * 60 * 60 * 1000, + ); + const liveMeetingCutoff = new Date( + now.getTime() - LIVE_MEETING_GRACE_MINUTES * 60 * 1000, + ); + + const result: ConvergeDivergedCallRecordingsResult = { + candidateCount: candidates.length, + updatedCallRecordingIds: [], + markedFailedCallRecordingIds: [], + unconvergeableCallRecordingIds: [], + }; + + for (const candidate of candidates) { + if (isOutsideConvergenceBound(candidate, convergenceLowerBound)) { + console.warn( + `[twenty-meeting-bot] call recording ${candidate.id} diverged but its meeting ended more than ${CONVERGENCE_LOOKBACK_DAYS} days ago; it will not converge automatically`, + ); + result.unconvergeableCallRecordingIds.push(candidate.id); + continue; + } + + if (isUndefined(candidate.externalBotId)) { + console.warn( + `[twenty-meeting-bot] call recording ${candidate.id} diverged but has no Recall bot id; it will not converge automatically`, + ); + result.unconvergeableCallRecordingIds.push(candidate.id); + continue; + } + + if (isPossiblyStillLive(candidate, liveMeetingCutoff)) { + continue; + } + + await convergeCallRecording({ + client, + candidate, + externalBotId: candidate.externalBotId, + result, + }); + } + + return result; +}; + +const fetchDivergedCallRecordingCandidates = async ( + client: CoreApiClient, +): Promise => { + // No createdAt bound: older-than-lookback candidates must surface in logs. + const filter: Record = { + or: [ + { + recordingRequestStatus: { eq: CallRecordingRequestStatus.REQUESTED }, + status: { in: NON_TERMINAL_CALL_RECORDING_STATUSES }, + externalBotId: { is: 'NOT_NULL' }, + }, + ], + }; + const candidateNodes = await fetchAllNodes( + async (afterCursor) => { + const queryResult = await client.query({ + callRecordings: { + __args: { + filter, + first: TWENTY_PAGE_SIZE, + ...(isUndefined(afterCursor) ? {} : { after: afterCursor }), + }, + pageInfo: { + hasNextPage: true, + endCursor: true, + }, + edges: { + node: { + id: true, + status: true, + startedAt: true, + endedAt: true, + externalBotId: true, + externalRecordingId: true, + createdAt: true, + calendarEvent: { + endsAt: true, + }, + }, + }, + }, + }); + + return queryResult.callRecordings as + | ConnectionPage + | undefined; + }, + ); + + return candidateNodes.map((node) => ({ + id: node.id, + status: node.status ?? undefined, + startedAt: node.startedAt ?? undefined, + endedAt: node.endedAt ?? undefined, + externalBotId: isNonEmptyString(node.externalBotId) + ? node.externalBotId + : undefined, + externalRecordingId: isNonEmptyString(node.externalRecordingId) + ? node.externalRecordingId + : undefined, + createdAt: node.createdAt ?? undefined, + calendarEventEndsAt: node.calendarEvent?.endsAt ?? undefined, + })); +}; + +// Anchored to meeting end: createdAt is scheduling time and can predate the meeting by weeks. +const isOutsideConvergenceBound = ( + candidate: DivergedCallRecordingCandidate, + convergenceLowerBound: Date, +): boolean => { + const meetingEndReference = + candidate.calendarEventEndsAt ?? candidate.createdAt; + + return ( + !isUndefined(meetingEndReference) && + new Date(meetingEndReference).getTime() < convergenceLowerBound.getTime() + ); +}; + +// Inside the grace period the meeting may still be recording; webhooks own it. +const isPossiblyStillLive = ( + candidate: DivergedCallRecordingCandidate, + liveMeetingCutoff: Date, +): boolean => + !isUndefined(candidate.calendarEventEndsAt) && + new Date(candidate.calendarEventEndsAt).getTime() > + liveMeetingCutoff.getTime(); + +const convergeCallRecording = async ({ + client, + candidate, + externalBotId, + result, +}: { + client: CoreApiClient; + candidate: DivergedCallRecordingCandidate; + externalBotId: string; + result: ConvergeDivergedCallRecordingsResult; +}): Promise => { + const botResult = await getRecallBot({ externalBotId }); + + if (!botResult.ok) { + if (botResult.status === 404) { + await markCallRecordingFailedAfterBotLoss({ + client, + candidate, + externalBotId, + result, + }); + + return; + } + + console.warn( + `[twenty-meeting-bot] failed to fetch Recall bot ${externalBotId} for call recording ${candidate.id}: ${botResult.errorMessage}`, + ); + + return; + } + + const convergence = extractRecallBotConvergence(botResult.bot); + const updateData = buildConvergenceFieldUpdates({ candidate, convergence }); + + if (Object.keys(updateData).length === 0) { + return; + } + + await updateCallRecording(client, { + id: candidate.id, + data: updateData, + }); + + result.updatedCallRecordingIds.push(candidate.id); +}; + +// Pure merge: fill only unset candidate fields and never downgrade status. +const buildConvergenceFieldUpdates = ({ + candidate, + convergence, +}: { + candidate: DivergedCallRecordingCandidate; + convergence: RecallBotConvergence; +}): CallRecordingUpdateFields => { + const updateData: CallRecordingUpdateFields = {}; + + if ( + !isUndefined(convergence.status) && + convergence.status !== candidate.status && + !isCallRecordingStatusDowngrade({ + fromStatus: candidate.status, + toStatus: convergence.status, + }) + ) { + updateData.status = convergence.status; + } + + if (isUndefined(candidate.startedAt) && !isUndefined(convergence.startedAt)) { + updateData.startedAt = convergence.startedAt; + } + + if (isUndefined(candidate.endedAt) && !isUndefined(convergence.endedAt)) { + updateData.endedAt = convergence.endedAt; + } + + if ( + isUndefined(candidate.externalRecordingId) && + !isUndefined(convergence.externalRecordingId) + ) { + updateData.externalRecordingId = convergence.externalRecordingId; + } + + return updateData; +}; + +const markCallRecordingFailedAfterBotLoss = async ({ + client, + candidate, + externalBotId, + result, +}: { + client: CoreApiClient; + candidate: DivergedCallRecordingCandidate; + externalBotId: string; + result: ConvergeDivergedCallRecordingsResult; +}): Promise => { + // externalBotId is kept for audit even though the bot is gone at Recall. + console.warn( + `[twenty-meeting-bot] Recall bot ${externalBotId} for call recording ${candidate.id} no longer exists; it will not converge automatically`, + ); + + if ( + isCallRecordingStatusDowngrade({ + fromStatus: candidate.status, + toStatus: CallRecordingStatus.FAILED_UNKNOWN, + }) + ) { + result.unconvergeableCallRecordingIds.push(candidate.id); + + return; + } + + await updateCallRecording(client, { + id: candidate.id, + data: { status: CallRecordingStatus.FAILED_UNKNOWN }, + }); + result.markedFailedCallRecordingIds.push(candidate.id); +}; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/ensure-meeting-bot.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/ensure-meeting-bot.util.ts index a4160f871a..0c4215e38d 100644 --- a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/ensure-meeting-bot.util.ts +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/ensure-meeting-bot.util.ts @@ -8,17 +8,16 @@ import { findCallRecordingsByIds } from 'src/logic-functions/data/find-call-reco import { scheduleRecallBot } from 'src/logic-functions/recall-api/schedule-recall-bot.util'; import { updateCallRecording } from 'src/logic-functions/data/update-call-recording.util'; -// The sole place a Recall bot is created. The deterministic-create winner and active update path call it in this split. -// TODO: Add the convergence cron in the next split so botless REQUESTED rows are healed by the same writer. +// The sole place a Recall bot is created. Only the deterministic-create winner and the stale-state cron call it, so one writer per meeting POSTs exactly one bot. export const ensureMeetingBot = async ( client: CoreApiClient, { callRecording, calendarEvent }: MeetingRecording, -): Promise => { +): Promise => { const meetingUrl = calendarEvent.conferenceLinkUrl; const joinAt = calendarEvent.startsAt; if (isUndefined(meetingUrl) || isUndefined(joinAt)) { - return; + return false; } const freshCallRecording = ( @@ -31,7 +30,7 @@ export const ensureMeetingBot = async ( CallRecordingRequestStatus.REQUESTED || !isUndefined(freshCallRecording.externalBotId) ) { - return; + return false; } const scheduleResult = await scheduleRecallBot({ @@ -45,11 +44,13 @@ export const ensureMeetingBot = async ( `[twenty-meeting-bot] failed to schedule Recall bot for callRecording ${callRecording.id}: ${scheduleResult.errorMessage}`, ); - return; + return false; } await updateCallRecording(client, { id: callRecording.id, data: { externalBotId: scheduleResult.externalBotId }, }); + + return true; }; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/heal-call-recordings-missing-bot.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/heal-call-recordings-missing-bot.util.ts new file mode 100644 index 0000000000..6dbbe8af75 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/heal-call-recordings-missing-bot.util.ts @@ -0,0 +1,82 @@ +import { isUndefined } from '@sniptt/guards'; +import { CoreApiClient } from 'twenty-client-sdk/core'; + +import { type CalendarEventRecord } from 'src/logic-functions/types/calendar-event-record.type'; +import { ensureMeetingBot } from 'src/logic-functions/flows/ensure-meeting-bot.util'; +import { fetchCalendarEventsByIds } from 'src/logic-functions/data/fetch-calendar-events-by-ids.util'; +import { findOpenScheduledCallRecordings } from 'src/logic-functions/data/find-open-scheduled-call-recordings.util'; +import { getUniqueSortedIds } from 'src/logic-functions/utils/get-unique-sorted-ids.util'; + +export type HealCallRecordingsMissingBotResult = { + scheduledCallRecordingIds: string[]; +}; + +// Closes the create-winner crash gap: a run that inserted the row but died before POSTing leaves a botless recording, and the cron is the single writer that re-POSTs it. +export const healCallRecordingsMissingBot = async ({ + client, + now, +}: { + client: CoreApiClient; + now: Date; +}): Promise => { + const botlessCallRecordings = ( + await findOpenScheduledCallRecordings(client) + ).filter((callRecording) => isUndefined(callRecording.externalBotId)); + + if (botlessCallRecordings.length === 0) { + return { scheduledCallRecordingIds: [] }; + } + + const calendarEventsById = new Map( + ( + await fetchCalendarEventsByIds( + client, + getUniqueSortedIds( + botlessCallRecordings.map( + (callRecording) => callRecording.calendarEventId, + ), + ), + ) + ).map((calendarEvent) => [calendarEvent.id, calendarEvent]), + ); + const scheduledCallRecordingIds: string[] = []; + + for (const callRecording of botlessCallRecordings) { + const calendarEvent = isUndefined(callRecording.calendarEventId) + ? undefined + : calendarEventsById.get(callRecording.calendarEventId); + + if (isUndefined(calendarEvent) || hasMeetingEnded({ calendarEvent, now })) { + continue; + } + + const didScheduleMeetingBot = await ensureMeetingBot(client, { + callRecording, + calendarEvent, + }); + + if (didScheduleMeetingBot) { + scheduledCallRecordingIds.push(callRecording.id); + } + } + + return { scheduledCallRecordingIds }; +}; + +const hasMeetingEnded = ({ + calendarEvent, + now, +}: { + calendarEvent: CalendarEventRecord; + now: Date; +}): boolean => { + const reference = calendarEvent.endsAt ?? calendarEvent.startsAt; + + if (isUndefined(reference)) { + return false; + } + + const referenceTime = new Date(reference).getTime(); + + return !Number.isNaN(referenceTime) && referenceTime <= now.getTime(); +}; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reap-orphaned-meeting-bots.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reap-orphaned-meeting-bots.util.ts new file mode 100644 index 0000000000..a49a0ec150 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reap-orphaned-meeting-bots.util.ts @@ -0,0 +1,182 @@ +import { isNull, isUndefined } from '@sniptt/guards'; +import { CoreApiClient } from 'twenty-client-sdk/core'; + +import { APPLICATION_ID_ENV_VAR_NAME } from 'src/logic-functions/constants/application-id-env-var-name'; +import { CallRecordingRequestStatus } from 'src/logic-functions/constants/call-recording-request-status'; +import { type CallRecordingRecord } from 'src/logic-functions/types/call-recording-record.type'; +import { cancelRecallBot } from 'src/logic-functions/recall-api/cancel-recall-bot.util'; +import { ejectRecallBot } from 'src/logic-functions/recall-api/eject-recall-bot.util'; +import { findCallRecordingsByIds } from 'src/logic-functions/data/find-call-recordings-by-ids.util'; +import { getApplicationVariableValue } from 'src/logic-functions/utils/get-application-variable-value.util'; +import { getUniqueSortedIds } from 'src/logic-functions/utils/get-unique-sorted-ids.util'; +import { isNonEmptyString } from 'src/logic-functions/utils/is-non-empty-string.util'; +import { + listScheduledRecallBots, + type RecallScheduledBot, +} from 'src/logic-functions/recall-api/list-scheduled-recall-bots.util'; + +export type ReapOrphanedMeetingBotsResult = { + scannedBotCount: number; + canceledExternalBotIds: string[]; +}; + +// Bots no open CallRecording request claims would still join; cancel them on Recall. +export const reapOrphanedMeetingBots = async ({ + client, + joinAtAfter, + joinAtBefore, +}: { + client: CoreApiClient; + joinAtAfter: string; + joinAtBefore: string; +}): Promise => { + const listResult = await listScheduledRecallBots({ + joinAtAfter, + joinAtBefore, + }); + + if (!listResult.ok) { + console.warn( + `[twenty-meeting-bot] failed to list Recall bots for orphan reaping: ${listResult.errorMessage}`, + ); + + return { scannedBotCount: 0, canceledExternalBotIds: [] }; + } + + const currentApplicationId = getCurrentApplicationId(); + const appManagedBots = listResult.bots.filter((bot) => + isCurrentApplicationManagedBot({ bot, currentApplicationId }), + ); + + if (appManagedBots.length === 0) { + return { + scannedBotCount: listResult.bots.length, + canceledExternalBotIds: [], + }; + } + + const callRecordings = await findCallRecordingsByIds( + client, + getUniqueSortedIds( + appManagedBots.map((bot) => getClaimedCallRecordingId(bot)), + ), + ); + const callRecordingsById = new Map( + callRecordings.map((callRecording) => [callRecording.id, callRecording]), + ); + const canceledExternalBotIds: string[] = []; + + for (const bot of appManagedBots) { + const claimedCallRecordingId = getClaimedCallRecordingId(bot); + const callRecording = isUndefined(claimedCallRecordingId) + ? undefined + : callRecordingsById.get(claimedCallRecordingId); + + if (isBotClaimed({ bot, callRecording })) { + continue; + } + + console.warn( + `[twenty-meeting-bot] canceling orphaned Recall bot ${bot.id} (claimed callRecording: ${claimedCallRecordingId})`, + ); + + if (await cancelOrEjectRecallBot(bot.id)) { + canceledExternalBotIds.push(bot.id); + } + } + + return { + scannedBotCount: listResult.bots.length, + canceledExternalBotIds, + }; +}; + +const getClaimedCallRecordingId = ( + bot: RecallScheduledBot, +): string | undefined => { + const claimedCallRecordingId = bot.metadata.twentyCallRecordingId; + + return normalizeOptionalString(claimedCallRecordingId); +}; + +const getClaimedApplicationId = ( + bot: RecallScheduledBot, +): string | undefined => { + const claimedApplicationId = bot.metadata.twentyApplicationId; + + return normalizeOptionalString(claimedApplicationId); +}; + +const getCurrentApplicationId = (): string | undefined => + normalizeOptionalString( + getApplicationVariableValue(APPLICATION_ID_ENV_VAR_NAME), + ); + +const isCurrentApplicationManagedBot = ({ + bot, + currentApplicationId, +}: { + bot: RecallScheduledBot; + currentApplicationId: string | undefined; +}): boolean => { + if (isUndefined(getClaimedCallRecordingId(bot))) { + return false; + } + + const claimedApplicationId = getClaimedApplicationId(bot); + + return ( + !isUndefined(currentApplicationId) && + claimedApplicationId === currentApplicationId + ); +}; + +const isBotClaimed = ({ + bot, + callRecording, +}: { + bot: RecallScheduledBot; + callRecording: CallRecordingRecord | undefined; +}): boolean => { + if ( + callRecording?.recordingRequestStatus !== + CallRecordingRequestStatus.REQUESTED + ) { + return false; + } + + if (callRecording.externalBotId === bot.id) { + return true; + } + + // An id-less REQUESTED recording may have a bot-id write-back in flight; spare its bot. + return isUndefined(callRecording.externalBotId); +}; + +const cancelOrEjectRecallBot = async ( + externalBotId: string, +): Promise => { + const cancelResult = await cancelRecallBot({ externalBotId }); + + if (cancelResult.ok) { + return true; + } + + // Deleting only works for not-yet-joined bots; eject the ones already in a call. + if (!isNull(cancelResult.status)) { + const ejectResult = await ejectRecallBot({ externalBotId }); + + if (ejectResult.ok) { + return true; + } + } + + console.warn( + `[twenty-meeting-bot] failed to cancel orphaned Recall bot ${externalBotId}: ${cancelResult.errorMessage}`, + ); + + return false; +}; + +const normalizeOptionalString = (value: unknown): string | undefined => + isNonEmptyString(value) ? value.trim() : undefined; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reconcile-meeting-bot.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reconcile-meeting-bot.util.ts index 6f007d74b9..363cdd931a 100644 --- a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reconcile-meeting-bot.util.ts +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reconcile-meeting-bot.util.ts @@ -298,10 +298,6 @@ const updatePolicyManagedCallRecording = async ({ callRecording: existingCallRecording, calendarEvent: representativeCalendarEvent, }); - await ensureMeetingBot(client, { - callRecording: existingCallRecording, - calendarEvent: representativeCalendarEvent, - }); return { action: 'UPDATED', diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reschedule-call-recording-bot.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reschedule-call-recording-bot.util.ts index d43774fb2d..6ceef30afc 100644 --- a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reschedule-call-recording-bot.util.ts +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/flows/reschedule-call-recording-bot.util.ts @@ -36,7 +36,7 @@ export const rescheduleCallRecordingBot = async ( return; } - // The caller re-runs ensureMeetingBot so this botless REQUESTED row is re-created by the single writer. + // The bot vanished externally; drop the id so the stale-state cron re-creates it as the single writer. if (rescheduleResult.status === RECALL_BOT_NOT_FOUND_STATUS) { await updateCallRecording(client, { id: callRecording.id, diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/extract-recall-bot-convergence.test.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/extract-recall-bot-convergence.test.ts new file mode 100644 index 0000000000..1b46375a08 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/extract-recall-bot-convergence.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, it } from 'vitest'; + +import { extractRecallBotConvergence } from 'src/logic-functions/recall-api/extract-recall-bot-convergence.util'; + +describe('extractRecallBotConvergence', () => { + it('maps the latest status change code to a call recording status', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + { code: 'joining_call', created_at: '2026-01-01T12:58:00.000Z' }, + { code: 'in_call_recording', created_at: '2026-01-01T13:02:00.000Z' }, + { code: 'call_ended', created_at: '2026-01-01T14:00:00.000Z' }, + { code: 'done', created_at: '2026-01-01T14:05:00.000Z' }, + ], + }); + + // COMPLETED is reserved for full artifact ingestion, never bot state. + expect(convergence.status).toBe('PROCESSING'); + }); + + it('uses created_at to find the latest status when Recall returns status changes out of order', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + { code: 'done', created_at: '2026-01-01T14:05:00.000Z' }, + { code: 'joining_call', created_at: '2026-01-01T12:58:00.000Z' }, + { code: 'in_call_recording', created_at: '2026-01-01T13:02:00.000Z' }, + ], + }); + + expect(convergence.status).toBe('PROCESSING'); + }); + + it('prefers recording-object timestamps over status change entries', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + { code: 'in_call_recording', created_at: '2026-01-01T13:02:30.000Z' }, + { code: 'call_ended', created_at: '2026-01-01T14:00:30.000Z' }, + ], + recordings: [ + { + id: 'recall-recording-1', + started_at: '2026-01-01T13:02:00.000Z', + completed_at: '2026-01-01T14:00:00.000Z', + }, + ], + }); + + expect(convergence).toEqual({ + status: 'PROCESSING', + startedAt: '2026-01-01T13:02:00.000Z', + endedAt: '2026-01-01T14:00:00.000Z', + externalRecordingId: 'recall-recording-1', + }); + }); + + it('falls back to status change timestamps when recordings carry none', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + { code: 'in_call_recording', created_at: '2026-01-01T13:02:00.000Z' }, + { code: 'call_ended', created_at: '2026-01-01T14:00:00.000Z' }, + ], + recordings: [{ id: 'recall-recording-1' }], + }); + + expect(convergence).toEqual({ + status: 'PROCESSING', + startedAt: '2026-01-01T13:02:00.000Z', + endedAt: '2026-01-01T14:00:00.000Z', + externalRecordingId: 'recall-recording-1', + }); + }); + + it('normalizes microsecond-precision Recall timestamps to millisecond ISO', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + { code: 'done', created_at: '2026-06-10T12:20:00.123456+00:00' }, + ], + recordings: [ + { + id: 'recall-recording-1', + started_at: '2026-06-10T11:02:28.281597+00:00', + completed_at: '2026-06-10T12:17:28.281597+00:00', + }, + ], + }); + + expect(convergence.startedAt).toBe('2026-06-10T11:02:28.281Z'); + expect(convergence.endedAt).toBe('2026-06-10T12:17:28.281Z'); + }); + + it('returns nothing derivable from an empty bot response', () => { + expect(extractRecallBotConvergence({})).toEqual({ + status: undefined, + startedAt: undefined, + endedAt: undefined, + externalRecordingId: undefined, + }); + }); + + it('skips malformed status change entries', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + null, + 'not-an-object', + { created_at: '2026-01-01T13:00:00.000Z' }, + { code: 'in_call_recording', created_at: '2026-01-01T13:02:00.000Z' }, + ], + recordings: 'not-an-array', + }); + + expect(convergence).toEqual({ + status: 'RECORDING', + startedAt: '2026-01-01T13:02:00.000Z', + endedAt: undefined, + externalRecordingId: undefined, + }); + }); + + it('leaves the status undefined for unknown latest codes', () => { + const convergence = extractRecallBotConvergence({ + status_changes: [ + { code: 'in_call_recording', created_at: '2026-01-01T13:02:00.000Z' }, + { code: 'some_future_code', created_at: '2026-01-01T13:30:00.000Z' }, + ], + }); + + expect(convergence.status).toBeUndefined(); + expect(convergence.startedAt).toBe('2026-01-01T13:02:00.000Z'); + }); +}); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/recall-bot-api.test.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/recall-bot-api.test.ts index c525a0cbb1..b02abaa142 100644 --- a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/recall-bot-api.test.ts +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/__tests__/recall-bot-api.test.ts @@ -1,6 +1,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { cancelRecallBot } from 'src/logic-functions/recall-api/cancel-recall-bot.util'; +import { ejectRecallBot } from 'src/logic-functions/recall-api/eject-recall-bot.util'; +import { getRecallBot } from 'src/logic-functions/recall-api/get-recall-bot.util'; +import { listScheduledRecallBots } from 'src/logic-functions/recall-api/list-scheduled-recall-bots.util'; import { rescheduleRecallBot } from 'src/logic-functions/recall-api/reschedule-recall-bot.util'; import { scheduleRecallBot } from 'src/logic-functions/recall-api/schedule-recall-bot.util'; @@ -101,24 +104,6 @@ describe('recall bot api', () => { ); }); - it('cancels a scheduled Recall bot request', async () => { - fetchMock.mockResolvedValue({ - ok: true, - status: 204, - json: async () => ({}), - }); - - const result = await cancelRecallBot({ - externalBotId: 'recall-bot-id', - }); - - expect(result).toEqual({ ok: true }); - expect(fetchMock).toHaveBeenCalledWith( - 'https://ap-northeast-1.recall.ai/api/v1/bot/recall-bot-id/', - expect.objectContaining({ method: 'DELETE' }), - ); - }); - it('fails when the create response does not include a bot id', async () => { fetchMock.mockResolvedValue({ ok: true, @@ -200,6 +185,189 @@ describe('recall bot api', () => { ); }); + it('lists scheduled bots in a join-at window and follows pagination', async () => { + fetchMock + .mockResolvedValueOnce({ + ok: true, + status: 200, + json: async () => ({ + next: 'https://ap-northeast-1.recall.ai/api/v1/bot/?cursor=page-2', + results: [ + { id: 'bot-1', metadata: { twentyCallRecordingId: 'recording-1' } }, + ], + }), + }) + .mockResolvedValueOnce({ + ok: true, + status: 200, + json: async () => ({ + next: null, + results: [{ id: 'bot-2' }], + }), + }); + + const result = await listScheduledRecallBots({ + joinAtAfter: '2026-01-01T08:00:00.000Z', + joinAtBefore: '2026-01-02T12:00:00.000Z', + }); + + expect(result).toEqual({ + ok: true, + bots: [ + { id: 'bot-1', metadata: { twentyCallRecordingId: 'recording-1' } }, + { id: 'bot-2', metadata: {} }, + ], + }); + expect(fetchMock).toHaveBeenNthCalledWith( + 1, + 'https://ap-northeast-1.recall.ai/api/v1/bot/?join_at_after=2026-01-01T08%3A00%3A00.000Z&join_at_before=2026-01-02T12%3A00%3A00.000Z', + expect.objectContaining({ method: 'GET' }), + ); + expect(fetchMock).toHaveBeenNthCalledWith( + 2, + 'https://ap-northeast-1.recall.ai/api/v1/bot/?cursor=page-2', + expect.objectContaining({ method: 'GET' }), + ); + }); + + it('fails the scheduled bot list when the pagination cap would truncate results', async () => { + for (let pageIndex = 1; pageIndex <= 10; pageIndex++) { + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + json: async () => ({ + next: `https://ap-northeast-1.recall.ai/api/v1/bot/?cursor=page-${pageIndex + 1}`, + results: [{ id: `bot-${pageIndex}` }], + }), + }); + } + + const result = await listScheduledRecallBots({ + joinAtAfter: '2026-01-01T08:00:00.000Z', + joinAtBefore: '2026-01-02T12:00:00.000Z', + }); + + expect(result).toEqual({ + ok: false, + status: null, + errorMessage: 'Recall bot list exceeded 10 pages', + }); + expect(fetchMock).toHaveBeenCalledTimes(10); + }); + + it('stops paginating when the next link points outside the configured region', async () => { + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + json: async () => ({ + next: 'https://evil.example.com/api/v1/bot/?cursor=page-2', + results: [{ id: 'bot-1' }], + }), + }); + + const result = await listScheduledRecallBots({ + joinAtAfter: '2026-01-01T08:00:00.000Z', + joinAtBefore: '2026-01-02T12:00:00.000Z', + }); + + expect(result).toEqual({ + ok: true, + bots: [{ id: 'bot-1', metadata: {} }], + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('cancels a scheduled Recall bot request', async () => { + fetchMock.mockResolvedValue({ + ok: true, + status: 204, + json: async () => ({}), + }); + + const result = await cancelRecallBot({ + externalBotId: 'recall-bot-id', + }); + + expect(result).toEqual({ ok: true }); + expect(fetchMock).toHaveBeenCalledWith( + 'https://ap-northeast-1.recall.ai/api/v1/bot/recall-bot-id/', + expect.objectContaining({ method: 'DELETE' }), + ); + }); + + it('ejects a bot through the leave_call endpoint', async () => { + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + json: async () => ({ id: 'recall-bot-id' }), + }); + + const result = await ejectRecallBot({ + externalBotId: 'recall-bot-id', + }); + + expect(result).toEqual({ ok: true }); + expect(fetchMock).toHaveBeenCalledWith( + 'https://ap-northeast-1.recall.ai/api/v1/bot/recall-bot-id/leave_call/', + expect.objectContaining({ method: 'POST' }), + ); + }); + + it('fetches a single bot and returns the raw response', async () => { + const botResponse = { + id: 'recall-bot-id', + status_changes: [{ code: 'done' }], + recordings: [{ id: 'recall-recording-id' }], + }; + + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + json: async () => botResponse, + }); + + const result = await getRecallBot({ externalBotId: 'recall-bot-id' }); + + expect(result).toEqual({ ok: true, bot: botResponse }); + expect(fetchMock).toHaveBeenCalledWith( + 'https://ap-northeast-1.recall.ai/api/v1/bot/recall-bot-id/', + expect.objectContaining({ method: 'GET' }), + ); + }); + + it('fails when fetching a bot returns an empty response payload', async () => { + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + json: async () => null, + }); + + const result = await getRecallBot({ externalBotId: 'recall-bot-id' }); + + expect(result).toEqual({ + ok: false, + status: 200, + errorMessage: 'Recall API returned an empty bot response', + }); + }); + + it('reports the HTTP status when fetching a bot that no longer exists', async () => { + fetchMock.mockResolvedValue({ + ok: false, + status: 404, + json: async () => ({ detail: 'Not found.' }), + }); + + const result = await getRecallBot({ externalBotId: 'recall-bot-gone' }); + + expect(result).toEqual({ + ok: false, + status: 404, + errorMessage: + 'Recall API responded with HTTP 404: {"detail":"Not found."}', + }); + }); + describe('transient failure retries', () => { beforeEach(() => { vi.useFakeTimers(); @@ -217,22 +385,36 @@ describe('recall bot api', () => { json: async () => ({ id: 'recall-bot-id' }), }); - const resultPromise = rescheduleRecallBot({ - externalBotId: 'recall-bot-id', - meetingUrl: 'https://meet.google.com/abc-defg-hij', - joinAt: '2026-01-01T13:00:00.000Z', - metadata: { - twentyCallRecordingId: 'call-recording-id', - twentyCalendarEventId: 'calendar-event-id', - twentyRealMeetingKey: 'meeting-key', - }, - }); + const resultPromise = getRecallBot({ externalBotId: 'recall-bot-id' }); await vi.runAllTimersAsync(); expect(await resultPromise).toEqual({ ok: true, - externalBotId: 'recall-bot-id', + bot: { id: 'recall-bot-id' }, + }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('retries a 503 response and succeeds on the next attempt', async () => { + fetchMock.mockResolvedValueOnce({ + ok: false, + status: 503, + json: async () => ({ detail: 'service unavailable' }), + }); + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + json: async () => ({ id: 'recall-bot-id' }), + }); + + const resultPromise = getRecallBot({ externalBotId: 'recall-bot-id' }); + + await vi.runAllTimersAsync(); + + expect(await resultPromise).toEqual({ + ok: true, + bot: { id: 'recall-bot-id' }, }); expect(fetchMock).toHaveBeenCalledTimes(2); }); @@ -244,16 +426,7 @@ describe('recall bot api', () => { json: async () => ({ detail: 'server error' }), }); - const resultPromise = rescheduleRecallBot({ - externalBotId: 'recall-bot-id', - meetingUrl: 'https://meet.google.com/abc-defg-hij', - joinAt: '2026-01-01T13:00:00.000Z', - metadata: { - twentyCallRecordingId: 'call-recording-id', - twentyCalendarEventId: 'calendar-event-id', - twentyRealMeetingKey: 'meeting-key', - }, - }); + const resultPromise = getRecallBot({ externalBotId: 'recall-bot-id' }); await vi.runAllTimersAsync(); @@ -266,6 +439,24 @@ describe('recall bot api', () => { expect(fetchMock).toHaveBeenCalledTimes(3); }); + it('does not retry client errors', async () => { + fetchMock.mockResolvedValue({ + ok: false, + status: 400, + json: async () => ({ detail: 'bad request' }), + }); + + const result = await getRecallBot({ externalBotId: 'recall-bot-id' }); + + expect(result).toEqual({ + ok: false, + status: 400, + errorMessage: + 'Recall API responded with HTTP 400: {"detail":"bad request"}', + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + it('does not retry an allowed 404 on cancel', async () => { fetchMock.mockResolvedValue({ ok: false, @@ -280,5 +471,20 @@ describe('recall bot api', () => { expect(result).toEqual({ ok: true }); expect(fetchMock).toHaveBeenCalledTimes(1); }); + + it('does not retry an allowed 404 on eject', async () => { + fetchMock.mockResolvedValue({ + ok: false, + status: 404, + json: async () => ({ detail: 'not found' }), + }); + + const result = await ejectRecallBot({ + externalBotId: 'recall-bot-id', + }); + + expect(result).toEqual({ ok: true }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); }); }); diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/eject-recall-bot.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/eject-recall-bot.util.ts new file mode 100644 index 0000000000..c1bd3fc5dc --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/eject-recall-bot.util.ts @@ -0,0 +1,28 @@ +import { type RecallBotRemovalResult } from 'src/logic-functions/types/recall-bot-operation-result.type'; +import { getRecallApiConfig } from 'src/logic-functions/recall-api/get-recall-api-config.util'; +import { recallBotApiRequest } from 'src/logic-functions/recall-api/recall-bot-api-request.util'; + +export const ejectRecallBot = async ({ + externalBotId, +}: { + externalBotId: string; +}): Promise => { + const configResult = getRecallApiConfig(); + + if (!configResult.success) { + return { ok: false, status: null, errorMessage: configResult.error }; + } + + const result = await recallBotApiRequest({ + config: configResult.config, + path: `/bot/${externalBotId}/leave_call/`, + method: 'POST', + allowNotFound: true, + }); + + if (!result.ok) { + return result; + } + + return { ok: true }; +}; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/extract-recall-bot-convergence.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/extract-recall-bot-convergence.util.ts new file mode 100644 index 0000000000..d3ef27a030 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/extract-recall-bot-convergence.util.ts @@ -0,0 +1,137 @@ +import { isArray, isUndefined } from '@sniptt/guards'; + +import { type CallRecordingStatus } from 'src/logic-functions/constants/call-recording-status'; +import { asRecord } from 'src/logic-functions/utils/as-record.util'; +import { getString } from 'src/logic-functions/utils/get-string.util'; +import { mapRecallStatusCodeToCallRecordingStatus } from 'src/logic-functions/domain/map-recall-status-code-to-call-recording-status.util'; +import { normalizeRecallTimestamp } from 'src/logic-functions/recall-api/normalize-recall-timestamp.util'; + +export type RecallBotConvergence = { + status: CallRecordingStatus | undefined; + startedAt: string | undefined; + endedAt: string | undefined; + externalRecordingId: string | undefined; +}; + +type RecallBotStatusChange = { + code: string; + createdAt: string | undefined; +}; + +// Derives the state a full webhook history would have produced from GET /bot. +export const extractRecallBotConvergence = ( + bot: Record, +): RecallBotConvergence => { + const statusChanges = extractStatusChanges(bot); + const latestStatusChange = getLatestStatusChange(statusChanges); + const recording = extractFirstRecording(bot); + + return { + status: mapRecallStatusCodeToCallRecordingStatus(latestStatusChange?.code), + startedAt: normalizeRecallTimestamp( + recording?.startedAt ?? + findStatusChangeTimestamp(statusChanges, 'in_call_recording'), + ), + endedAt: normalizeRecallTimestamp( + recording?.completedAt ?? + findStatusChangeTimestamp(statusChanges, 'call_ended'), + ), + externalRecordingId: recording?.id, + }; +}; + +const extractStatusChanges = ( + bot: Record, +): RecallBotStatusChange[] => { + if (!isArray(bot.status_changes)) { + return []; + } + + return bot.status_changes.flatMap((statusChange: unknown) => { + const code = getString(asRecord(statusChange)?.code); + + if (isUndefined(code)) { + return []; + } + + return [{ code, createdAt: getString(asRecord(statusChange)?.created_at) }]; + }); +}; + +const getLatestStatusChange = ( + statusChanges: RecallBotStatusChange[], +): RecallBotStatusChange | undefined => + statusChanges.reduce( + (latestStatusChange, statusChange) => { + if (isUndefined(latestStatusChange)) { + return statusChange; + } + + const statusChangeTime = getStatusChangeTime(statusChange); + const latestStatusChangeTime = getStatusChangeTime(latestStatusChange); + + if ( + isUndefined(statusChangeTime) && + isUndefined(latestStatusChangeTime) + ) { + return statusChange; + } + + if (isUndefined(statusChangeTime)) { + return latestStatusChange; + } + + if (isUndefined(latestStatusChangeTime)) { + return statusChange; + } + + return statusChangeTime >= latestStatusChangeTime + ? statusChange + : latestStatusChange; + }, + undefined, + ); + +const getStatusChangeTime = ( + statusChange: RecallBotStatusChange, +): number | undefined => { + const normalizedTimestamp = normalizeRecallTimestamp(statusChange.createdAt); + + if (isUndefined(normalizedTimestamp)) { + return undefined; + } + + return new Date(normalizedTimestamp).getTime(); +}; + +const extractFirstRecording = ( + bot: Record, +): + | { + id: string | undefined; + startedAt: string | undefined; + completedAt: string | undefined; + } + | undefined => { + if (!isArray(bot.recordings)) { + return undefined; + } + + const recording = asRecord(bot.recordings[0]); + + if (isUndefined(recording)) { + return undefined; + } + + return { + id: getString(recording.id), + startedAt: getString(recording.started_at), + completedAt: getString(recording.completed_at), + }; +}; + +const findStatusChangeTimestamp = ( + statusChanges: RecallBotStatusChange[], + code: string, +): string | undefined => + statusChanges.find((statusChange) => statusChange.code === code)?.createdAt; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/get-recall-bot.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/get-recall-bot.util.ts new file mode 100644 index 0000000000..7772622d6d --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/get-recall-bot.util.ts @@ -0,0 +1,42 @@ +import { type RecallBotOperationFailure } from 'src/logic-functions/types/recall-bot-operation-result.type'; +import { asRecord } from 'src/logic-functions/utils/as-record.util'; +import { getRecallApiConfig } from 'src/logic-functions/recall-api/get-recall-api-config.util'; +import { recallBotApiRequest } from 'src/logic-functions/recall-api/recall-bot-api-request.util'; + +type GetRecallBotResult = + | { ok: true; bot: Record } + | RecallBotOperationFailure; + +export const getRecallBot = async ({ + externalBotId, +}: { + externalBotId: string; +}): Promise => { + const configResult = getRecallApiConfig(); + + if (!configResult.success) { + return { ok: false, status: null, errorMessage: configResult.error }; + } + + const result = await recallBotApiRequest>({ + config: configResult.config, + path: `/bot/${externalBotId}/`, + method: 'GET', + }); + + if (!result.ok) { + return result; + } + + const bot = asRecord(result.data); + + if (bot === undefined) { + return { + ok: false, + status: result.status, + errorMessage: 'Recall API returned an empty bot response', + }; + } + + return { ok: true, bot }; +}; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/list-scheduled-recall-bots.util.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/list-scheduled-recall-bots.util.ts new file mode 100644 index 0000000000..19fe22fae3 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/recall-api/list-scheduled-recall-bots.util.ts @@ -0,0 +1,106 @@ +import { isString, isUndefined } from '@sniptt/guards'; + +import { type RecallBotOperationFailure } from 'src/logic-functions/types/recall-bot-operation-result.type'; +import { asRecord } from 'src/logic-functions/utils/as-record.util'; +import { getRecallApiConfig } from 'src/logic-functions/recall-api/get-recall-api-config.util'; +import { recallBotApiRequest } from 'src/logic-functions/recall-api/recall-bot-api-request.util'; + +export type RecallScheduledBot = { + id: string; + metadata: Record; +}; + +type RecallBotListResponse = { + next?: unknown; + results?: unknown; +}; + +type ListScheduledRecallBotsResult = + | { ok: true; bots: RecallScheduledBot[] } + | RecallBotOperationFailure; + +const RECALL_BOT_LIST_MAX_PAGES = 10; + +export const listScheduledRecallBots = async ({ + joinAtAfter, + joinAtBefore, +}: { + joinAtAfter: string; + joinAtBefore: string; +}): Promise => { + const configResult = getRecallApiConfig(); + + if (!configResult.success) { + return { ok: false, status: null, errorMessage: configResult.error }; + } + + const bots: RecallScheduledBot[] = []; + let path: string | undefined = `/bot/?join_at_after=${encodeURIComponent( + joinAtAfter, + )}&join_at_before=${encodeURIComponent(joinAtBefore)}`; + + for ( + let pageIndex = 0; + !isUndefined(path) && pageIndex < RECALL_BOT_LIST_MAX_PAGES; + pageIndex++ + ) { + const result = await recallBotApiRequest({ + config: configResult.config, + path, + method: 'GET', + }); + + if (!result.ok) { + return result; + } + + bots.push(...extractRecallBots(result.data)); + path = extractNextPath(result.data, configResult.config.baseUrl); + } + + if (!isUndefined(path)) { + return { + ok: false, + status: null, + errorMessage: `Recall bot list exceeded ${RECALL_BOT_LIST_MAX_PAGES} pages`, + }; + } + + return { ok: true, bots }; +}; + +const extractRecallBots = ( + response: RecallBotListResponse | undefined, +): RecallScheduledBot[] => { + if (!Array.isArray(response?.results)) { + return []; + } + + return response.results.flatMap((candidate: unknown) => { + const bot = asRecord(candidate); + + if (isUndefined(bot) || !isString(bot.id)) { + return []; + } + + return [ + { + id: bot.id, + metadata: asRecord(bot.metadata) ?? {}, + }, + ]; + }); +}; + +const extractNextPath = ( + response: RecallBotListResponse | undefined, + baseUrl: string, +): string | undefined => { + const next = response?.next; + + if (!isString(next) || !next.startsWith(baseUrl)) { + return undefined; + } + + return next.slice(baseUrl.length); +}; diff --git a/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/reconcile-stale-bot-state.ts b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/reconcile-stale-bot-state.ts new file mode 100644 index 0000000000..f7a89bcae8 --- /dev/null +++ b/packages/twenty-apps/internal/twenty-meeting-bot/src/logic-functions/reconcile-stale-bot-state.ts @@ -0,0 +1,111 @@ +import { CoreApiClient } from 'twenty-client-sdk/core'; +import { defineLogicFunction } from 'twenty-sdk/define'; + +import { STALE_BOT_STATE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER } from 'src/constants/stale-bot-state-logic-function-universal-identifier'; +import { STALE_BOT_STATE_CRON_PATTERN } from 'src/logic-functions/constants/stale-bot-state-cron-pattern'; +import { + convergeDivergedCallRecordings, + type ConvergeDivergedCallRecordingsResult, +} from 'src/logic-functions/flows/converge-diverged-call-recordings.util'; +import { + healCallRecordingsMissingBot, + type HealCallRecordingsMissingBotResult, +} from 'src/logic-functions/flows/heal-call-recordings-missing-bot.util'; +import { + reapOrphanedMeetingBots, + type ReapOrphanedMeetingBotsResult, +} from 'src/logic-functions/flows/reap-orphaned-meeting-bots.util'; + +// Every unwanted bot passes through this join_at window before it can attend. +const REAPER_JOIN_AT_LOOKBACK_HOURS = 4; +const REAPER_JOIN_AT_LOOKAHEAD_HOURS = 24; + +type StepFailure = { error: string }; + +export const reconcileStaleBotStateHandler = async (): Promise => { + const now = new Date(); + const client = new CoreApiClient(); + + const botlessHealResult = await healCallRecordingsMissingBotSafely( + client, + now, + ); + const orphanedBotReapingResult = await reapOrphanedMeetingBotsInJoinAtWindow( + client, + now, + ); + const statusConvergenceResult = await convergeDivergedCallRecordingsSafely( + client, + now, + ); + + return { + botlessHealResult, + orphanedBotReapingResult, + statusConvergenceResult, + }; +}; + +const healCallRecordingsMissingBotSafely = async ( + client: CoreApiClient, + now: Date, +): Promise => { + try { + return await healCallRecordingsMissingBot({ client, now }); + } catch (error) { + return buildStepFailure('botless call recording healing', error); + } +}; + +const reapOrphanedMeetingBotsInJoinAtWindow = async ( + client: CoreApiClient, + now: Date, +): Promise => { + try { + return await reapOrphanedMeetingBots({ + client, + joinAtAfter: new Date( + now.getTime() - REAPER_JOIN_AT_LOOKBACK_HOURS * 60 * 60 * 1000, + ).toISOString(), + joinAtBefore: new Date( + now.getTime() + REAPER_JOIN_AT_LOOKAHEAD_HOURS * 60 * 60 * 1000, + ).toISOString(), + }); + } catch (error) { + return buildStepFailure('orphaned bot reaping', error); + } +}; + +const convergeDivergedCallRecordingsSafely = async ( + client: CoreApiClient, + now: Date, +): Promise => { + try { + return await convergeDivergedCallRecordings({ client, now }); + } catch (error) { + return buildStepFailure('call recording status convergence', error); + } +}; + +const buildStepFailure = (stepLabel: string, error: unknown): StepFailure => { + const errorMessage = error instanceof Error ? error.message : String(error); + + if (process.env.NODE_ENV !== 'test') { + console.error(`[twenty-meeting-bot] ${stepLabel} failed: ${errorMessage}`); + } + + return { error: `${stepLabel} failed` }; +}; + +export default defineLogicFunction({ + universalIdentifier: STALE_BOT_STATE_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIER, + name: 'reconcile-stale-bot-state', + description: + 'Converges call recordings with Recall on a schedule: pulls stale bot statuses, finishes failed cancellations, schedules bots for recordings still missing one, and reaps unclaimed bots. Reads calendar events only to heal already-decided recordings, never to discover meetings.', + // Pulling bot statuses for many recordings is the dominant cost. + timeoutSeconds: 300, + handler: reconcileStaleBotStateHandler, + cronTriggerSettings: { + pattern: STALE_BOT_STATE_CRON_PATTERN, + }, +});