diff --git a/packages/twenty-apps/public/call-recorder/package.json b/packages/twenty-apps/public/call-recorder/package.json index 0044d27611..d87f9ba467 100644 --- a/packages/twenty-apps/public/call-recorder/package.json +++ b/packages/twenty-apps/public/call-recorder/package.json @@ -1,6 +1,6 @@ { "name": "@twentyhq/call-recorder", - "version": "1.0.8", + "version": "1.0.9", "license": "MIT", "engines": { "node": "^24.5.0", @@ -35,8 +35,8 @@ "oxlint": "^0.16.0", "react": "^19.0.0", "react-dom": "^19.0.0", - "twenty-client-sdk": "2.19.0-alpha.1", - "twenty-sdk": "2.19.0-alpha.1", + "twenty-client-sdk": "2.19.0", + "twenty-sdk": "2.19.0", "twenty-ui": "^1.0.0-alpha.1", "typescript": "^5.9.3", "vite-tsconfig-paths": "^4.2.1", diff --git a/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/ingest-call-recording-media.test.ts b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/ingest-call-recording-media.test.ts index 93ca765dc1..08a8372fe3 100644 --- a/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/ingest-call-recording-media.test.ts +++ b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/ingest-call-recording-media.test.ts @@ -1,14 +1,16 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { CALL_RECORDING_VIDEO_FIELD_UNIVERSAL_IDENTIFIER } from 'src/constants/call-recording-video-field-universal-identifier'; import { CALL_RECORDER_MAX_MEDIA_FILE_SIZE_MB_ENV_VAR_NAME } from 'src/logic-functions/constants/call-recorder-max-media-file-size-mb-env-var-name'; import { ingestCallRecordingMedia } from 'src/logic-functions/flows/ingest-call-recording-media.util'; -const uploadFileMock = vi.hoisted(() => vi.fn()); +const mutationMock = vi.hoisted(() => vi.fn()); const getRecallRecordingMock = vi.hoisted(() => vi.fn()); +const putMediaDownloadBodyToUploadTargetMock = vi.hoisted(() => vi.fn()); vi.mock('twenty-client-sdk/metadata', () => ({ MetadataApiClient: class { - uploadFile = uploadFileMock; + mutation = mutationMock; }, })); @@ -16,6 +18,13 @@ vi.mock('src/logic-functions/recall-api/get-recall-recording.util', () => ({ getRecallRecording: getRecallRecordingMock, })); +vi.mock( + 'src/logic-functions/flows/put-media-download-body-to-upload-target.util', + () => ({ + putMediaDownloadBodyToUploadTarget: putMediaDownloadBodyToUploadTargetMock, + }), +); + const VIDEO_URL = 'https://media.example.com/video.mp4'; const AUDIO_URL = 'https://media.example.com/audio.mp3'; @@ -27,6 +36,9 @@ const RECORDING_WITH_MEDIA = { }, }; +const uploadUrlForFilename = (filename: string) => + `https://storage.example.com/${filename}`; + const buildBodyStream = (chunks: Uint8Array[]): ReadableStream => new ReadableStream({ start(controller) { @@ -37,16 +49,14 @@ const buildBodyStream = (chunks: Uint8Array[]): ReadableStream => }, }); -const buildFetchResponse = ({ - contentType = 'video/mp4', +const buildDownloadResponse = ({ contentLengthBytes, body, }: { - contentType?: string; contentLengthBytes?: number; body?: unknown; } = {}) => { - const headers = new Map([['content-type', contentType]]); + const headers = new Map(); if (contentLengthBytes !== undefined) { headers.set('content-length', String(contentLengthBytes)); @@ -62,38 +72,105 @@ const buildFetchResponse = ({ }; }; -const stubFetchByUrl = (responsesByUrl: Record) => { - vi.stubGlobal( - 'fetch', - vi.fn().mockImplementation((url: string) => { - const response = responsesByUrl[url]; +const fetchMock = vi.fn(); - if (response === undefined) { +type DirectUploadMutationRequest = + | { createFileUpload: { __args: { filename: string } } } + | { completeFileUpload: { __args: { fileId: string } } }; + +const stubFetch = ({ + downloadsByUrl, +}: { + downloadsByUrl: Record; +}) => { + fetchMock.mockReset(); + fetchMock.mockImplementation( + (url: string, init?: { method?: string }) => { + if (init?.method === 'PUT') { + throw new Error('Upload requests should go through the upload bridge'); + } + + const downloadResponse = downloadsByUrl[url]; + + if (downloadResponse === undefined) { throw new Error(`Unhandled fetch url in test: ${url}`); } - return Promise.resolve(response); - }), + return Promise.resolve(downloadResponse); + }, ); + + vi.stubGlobal('fetch', fetchMock); }; +// createFileUpload returns a presigned target whose fileId echoes the filename, +// and completeFileUpload resolves that target to the final stored file id. +const stubDirectUpload = ({ + finalFileIdByFilename, + createFileUploadErrorByFilename = {}, +}: { + finalFileIdByFilename: Record; + createFileUploadErrorByFilename?: Record; +}) => { + mutationMock.mockImplementation((request: DirectUploadMutationRequest) => { + if ('createFileUpload' in request) { + const { filename } = request.createFileUpload.__args; + const createFileUploadError = createFileUploadErrorByFilename[filename]; + + if (createFileUploadError) { + return Promise.reject(createFileUploadError); + } + + return Promise.resolve({ + createFileUpload: { + fileId: filename, + uploadUrl: uploadUrlForFilename(filename), + contentType: 'application/octet-stream', + }, + }); + } + + if ('completeFileUpload' in request) { + const { fileId } = request.completeFileUpload.__args; + + return Promise.resolve({ + completeFileUpload: { id: finalFileIdByFilename[fileId] }, + }); + } + + throw new Error('Unhandled mutation in test'); + }); +}; + +const getUploadBridgeCall = (fileName: string) => + putMediaDownloadBodyToUploadTargetMock.mock.calls.find( + ([uploadInput]) => uploadInput.fileName === fileName, + ); + describe('ingestCallRecordingMedia', () => { beforeEach(() => { vi.spyOn(console, 'warn').mockImplementation(() => {}); - uploadFileMock.mockReset(); + vi.spyOn(console, 'log').mockImplementation(() => {}); + mutationMock.mockReset(); getRecallRecordingMock.mockReset(); + putMediaDownloadBodyToUploadTargetMock.mockReset(); + putMediaDownloadBodyToUploadTargetMock.mockResolvedValue(undefined); getRecallRecordingMock.mockResolvedValue({ ok: true, recording: RECORDING_WITH_MEDIA, }); - vi.stubGlobal( - 'fetch', - vi - .fn() - .mockImplementation(() => - Promise.resolve(buildFetchResponse({ contentLengthBytes: 8 })), - ), - ); + stubDirectUpload({ + finalFileIdByFilename: { + 'video.mp4': 'file-video-1', + 'audio.mp3': 'file-audio-1', + }, + }); + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ contentLengthBytes: 8 }), + [AUDIO_URL]: buildDownloadResponse({ contentLengthBytes: 8 }), + }, + }); }); afterEach(() => { @@ -101,11 +178,7 @@ describe('ingestCallRecordingMedia', () => { vi.unstubAllEnvs(); }); - it('downloads and uploads every missing artifact', async () => { - uploadFileMock - .mockResolvedValueOnce({ id: 'file-video-1' }) - .mockResolvedValueOnce({ id: 'file-audio-1' }); - + it('streams and uploads every missing artifact', async () => { const updateFields = await ingestCallRecordingMedia({ callRecordingId: 'call-recording-1', externalRecordingId: 'recall-recording-1', @@ -117,12 +190,57 @@ describe('ingestCallRecordingMedia', () => { video: [{ fileId: 'file-video-1', label: 'video.mp4' }], audio: [{ fileId: 'file-audio-1', label: 'audio.mp3' }], }); - expect(uploadFileMock).toHaveBeenCalledTimes(2); + expect(getUploadBridgeCall('video.mp4')).toMatchObject([ + expect.objectContaining({ + fileName: 'video.mp4', + sizeBytes: 8, + uploadTarget: expect.objectContaining({ + uploadUrl: uploadUrlForFilename('video.mp4'), + }), + }), + ]); + expect(getUploadBridgeCall('audio.mp3')).toMatchObject([ + expect.objectContaining({ + fileName: 'audio.mp3', + sizeBytes: 8, + uploadTarget: expect.objectContaining({ + uploadUrl: uploadUrlForFilename('audio.mp3'), + }), + }), + ]); + }); + + it('declares the presigned upload with the download size, folder and field identifier', async () => { + await ingestCallRecordingMedia({ + callRecordingId: 'call-recording-1', + externalRecordingId: 'recall-recording-1', + hasAudio: true, + hasVideo: false, + }); + + expect(mutationMock).toHaveBeenCalledWith( + expect.objectContaining({ + createFileUpload: expect.objectContaining({ + __args: { + filename: 'video.mp4', + size: 8, + fileFolder: 'FilesField', + fieldMetadataUniversalIdentifier: + CALL_RECORDING_VIDEO_FIELD_UNIVERSAL_IDENTIFIER, + }, + }), + }), + ); + expect(mutationMock).toHaveBeenCalledWith( + expect.objectContaining({ + completeFileUpload: expect.objectContaining({ + __args: { fileId: 'video.mp4' }, + }), + }), + ); }); it('skips artifacts already on the record', async () => { - uploadFileMock.mockResolvedValue({ id: 'file-audio-1' }); - const updateFields = await ingestCallRecordingMedia({ callRecordingId: 'call-recording-1', externalRecordingId: 'recall-recording-1', @@ -133,7 +251,7 @@ describe('ingestCallRecordingMedia', () => { expect(updateFields).toEqual({ audio: [{ fileId: 'file-audio-1', label: 'audio.mp3' }], }); - expect(uploadFileMock).toHaveBeenCalledTimes(1); + expect(getUploadBridgeCall('video.mp4')).toBeUndefined(); }); it('does not fetch the recording when both artifacts are present', async () => { @@ -146,12 +264,30 @@ describe('ingestCallRecordingMedia', () => { expect(updateFields).toEqual({}); expect(getRecallRecordingMock).not.toHaveBeenCalled(); - expect(uploadFileMock).not.toHaveBeenCalled(); + expect(mutationMock).not.toHaveBeenCalled(); }); - it('omits an artifact and warns when its transfer fails', async () => { - uploadFileMock.mockRejectedValueOnce(new Error('upload exploded')); - uploadFileMock.mockResolvedValueOnce({ id: 'file-audio-1' }); + it('cancels the opened download body when creating its upload target fails', async () => { + const cancelMock = vi.fn().mockResolvedValue(undefined); + + stubDirectUpload({ + finalFileIdByFilename: { + 'video.mp4': 'file-video-1', + 'audio.mp3': 'file-audio-1', + }, + createFileUploadErrorByFilename: { + 'video.mp4': new Error('upload target exploded'), + }, + }); + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ + contentLengthBytes: 8, + body: { cancel: cancelMock }, + }), + [AUDIO_URL]: buildDownloadResponse({ contentLengthBytes: 8 }), + }, + }); const updateFields = await ingestCallRecordingMedia({ callRecordingId: 'call-recording-1', @@ -163,9 +299,41 @@ describe('ingestCallRecordingMedia', () => { expect(updateFields).toEqual({ audio: [{ fileId: 'file-audio-1', label: 'audio.mp3' }], }); - expect(Object.keys(updateFields)).toEqual(['audio']); + expect(cancelMock).toHaveBeenCalledTimes(1); + expect(getUploadBridgeCall('video.mp4')).toBeUndefined(); expect(console.warn).toHaveBeenCalledWith( - expect.stringContaining('upload exploded'), + expect.stringContaining('upload target exploded'), + ); + }); + + it('omits an artifact and warns when the download has no content length', async () => { + const cancelMock = vi.fn().mockRejectedValue(new Error('cancel exploded')); + + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ + body: { cancel: cancelMock }, + }), + [AUDIO_URL]: buildDownloadResponse({ contentLengthBytes: 8 }), + }, + }); + + const updateFields = await ingestCallRecordingMedia({ + callRecordingId: 'call-recording-1', + externalRecordingId: 'recall-recording-1', + hasAudio: false, + hasVideo: false, + }); + + expect(updateFields).toEqual({ + audio: [{ fileId: 'file-audio-1', label: 'audio.mp3' }], + }); + expect(getUploadBridgeCall('video.mp4')).toBeUndefined(); + expect(console.warn).toHaveBeenCalledWith( + expect.stringContaining('content-length'), + ); + expect(console.warn).toHaveBeenCalledWith( + expect.stringContaining('download-body-cancel-failed'), ); }); @@ -183,7 +351,7 @@ describe('ingestCallRecordingMedia', () => { }); expect(updateFields).toEqual({}); - expect(uploadFileMock).not.toHaveBeenCalled(); + expect(mutationMock).not.toHaveBeenCalled(); }); it('warns and returns nothing when the recording fetch fails', async () => { @@ -201,23 +369,24 @@ describe('ingestCallRecordingMedia', () => { }); expect(updateFields).toEqual({}); - expect(uploadFileMock).not.toHaveBeenCalled(); + expect(mutationMock).not.toHaveBeenCalled(); expect(console.warn).toHaveBeenCalledWith( expect.stringContaining('recording boom'), ); }); it('skips an oversized file without reading its body and records the reason', async () => { - const cancelMock = vi.fn().mockResolvedValue(undefined); + const cancelMock = vi.fn().mockRejectedValue(new Error('cancel exploded')); - stubFetchByUrl({ - [VIDEO_URL]: buildFetchResponse({ - contentLengthBytes: 200 * 1024 * 1024, - body: { cancel: cancelMock }, - }), - [AUDIO_URL]: buildFetchResponse({ contentLengthBytes: 8 }), + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ + contentLengthBytes: 200 * 1024 * 1024, + body: { cancel: cancelMock }, + }), + [AUDIO_URL]: buildDownloadResponse({ contentLengthBytes: 8 }), + }, }); - uploadFileMock.mockResolvedValue({ id: 'file-audio-1' }); const updateFields = await ingestCallRecordingMedia({ callRecordingId: 'call-recording-1', @@ -230,21 +399,26 @@ describe('ingestCallRecordingMedia', () => { audio: [{ fileId: 'file-audio-1', label: 'audio.mp3' }], callRecorderFailureReason: 'video_file_too_large', }); - expect(uploadFileMock).toHaveBeenCalledTimes(1); + expect(getUploadBridgeCall('video.mp4')).toBeUndefined(); expect(cancelMock).toHaveBeenCalled(); expect(console.warn).toHaveBeenCalledWith( expect.stringContaining('artifact-too-large'), ); + expect(console.warn).toHaveBeenCalledWith( + expect.stringContaining('download-body-cancel-failed'), + ); }); it('records both markers when video and audio exceed the cap', async () => { - stubFetchByUrl({ - [VIDEO_URL]: buildFetchResponse({ - contentLengthBytes: 200 * 1024 * 1024, - }), - [AUDIO_URL]: buildFetchResponse({ - contentLengthBytes: 120 * 1024 * 1024, - }), + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ + contentLengthBytes: 200 * 1024 * 1024, + }), + [AUDIO_URL]: buildDownloadResponse({ + contentLengthBytes: 120 * 1024 * 1024, + }), + }, }); const updateFields = await ingestCallRecordingMedia({ @@ -257,18 +431,19 @@ describe('ingestCallRecordingMedia', () => { expect(updateFields).toEqual({ callRecorderFailureReason: 'video_file_too_large,audio_file_too_large', }); - expect(uploadFileMock).not.toHaveBeenCalled(); + expect(mutationMock).not.toHaveBeenCalled(); }); it('honors the cap configured through the environment', async () => { vi.stubEnv(CALL_RECORDER_MAX_MEDIA_FILE_SIZE_MB_ENV_VAR_NAME, '1'); - stubFetchByUrl({ - [VIDEO_URL]: buildFetchResponse({ - contentLengthBytes: 2 * 1024 * 1024, - }), - [AUDIO_URL]: buildFetchResponse({ contentLengthBytes: 8 }), + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ + contentLengthBytes: 2 * 1024 * 1024, + }), + [AUDIO_URL]: buildDownloadResponse({ contentLengthBytes: 8 }), + }, }); - uploadFileMock.mockResolvedValue({ id: 'file-audio-1' }); const updateFields = await ingestCallRecordingMedia({ callRecordingId: 'call-recording-1', @@ -288,11 +463,12 @@ describe('ingestCallRecordingMedia', () => { CALL_RECORDER_MAX_MEDIA_FILE_SIZE_MB_ENV_VAR_NAME, 'not-a-number', ); - uploadFileMock.mockResolvedValue({ id: 'file-video-1' }); - stubFetchByUrl({ - [VIDEO_URL]: buildFetchResponse({ - contentLengthBytes: 2 * 1024 * 1024, - }), + stubFetch({ + downloadsByUrl: { + [VIDEO_URL]: buildDownloadResponse({ + contentLengthBytes: 2 * 1024 * 1024, + }), + }, }); const updateFields = await ingestCallRecordingMedia({ @@ -306,54 +482,4 @@ describe('ingestCallRecordingMedia', () => { video: [{ fileId: 'file-video-1', label: 'video.mp4' }], }); }); - - it('enforces the cap while reading a response without content-length', async () => { - vi.stubEnv(CALL_RECORDER_MAX_MEDIA_FILE_SIZE_MB_ENV_VAR_NAME, '1'); - stubFetchByUrl({ - [VIDEO_URL]: buildFetchResponse({ - body: buildBodyStream([ - new Uint8Array(700_000), - new Uint8Array(700_000), - ]), - }), - }); - - const updateFields = await ingestCallRecordingMedia({ - callRecordingId: 'call-recording-1', - externalRecordingId: 'recall-recording-1', - hasAudio: true, - hasVideo: false, - }); - - expect(updateFields).toEqual({ - callRecorderFailureReason: 'video_file_too_large', - }); - expect(uploadFileMock).not.toHaveBeenCalled(); - }); - - it('ingests a response without content-length when it stays within the cap', async () => { - uploadFileMock.mockResolvedValue({ id: 'file-video-1' }); - stubFetchByUrl({ - [VIDEO_URL]: buildFetchResponse({ - body: buildBodyStream([new Uint8Array([1, 2, 3]), new Uint8Array([4])]), - }), - }); - - const updateFields = await ingestCallRecordingMedia({ - callRecordingId: 'call-recording-1', - externalRecordingId: 'recall-recording-1', - hasAudio: true, - hasVideo: false, - }); - - expect(updateFields).toEqual({ - video: [{ fileId: 'file-video-1', label: 'video.mp4' }], - }); - expect(uploadFileMock).toHaveBeenCalledWith( - Buffer.from([1, 2, 3, 4]), - 'video.mp4', - 'video/mp4', - expect.any(String), - ); - }); }); diff --git a/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/put-media-download-body-to-upload-target.test.ts b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/put-media-download-body-to-upload-target.test.ts new file mode 100644 index 0000000000..1f2115c6a2 --- /dev/null +++ b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/__tests__/put-media-download-body-to-upload-target.test.ts @@ -0,0 +1,210 @@ +import { type ClientRequest, type IncomingMessage } from 'node:http'; +import { PassThrough, Readable } from 'node:stream'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const requestOverHttpMock = vi.hoisted(() => vi.fn()); +const requestOverHttpsMock = vi.hoisted(() => vi.fn()); + +vi.mock('node:http', async () => { + const actualHttp = await vi.importActual( + 'node:http', + ); + + return { ...actualHttp, request: requestOverHttpMock }; +}); + +vi.mock('node:https', async () => { + const actualHttps = await vi.importActual( + 'node:https', + ); + + return { ...actualHttps, request: requestOverHttpsMock }; +}); + +import { putMediaDownloadBodyToUploadTarget } from 'src/logic-functions/flows/put-media-download-body-to-upload-target.util'; + +const HTTPS_UPLOAD_URL = 'https://storage.example.com/video.mp4'; +const HTTP_UPLOAD_URL = 'http://storage.example.com/video.mp4'; + +const buildMediaDownloadBody = ({ + chunks = [new Uint8Array([1, 2, 3]), new Uint8Array([4])], + cancel, + close = true, +}: { + chunks?: Uint8Array[]; + cancel?: () => void | Promise; + close?: boolean; +} = {}): ReadableStream => + new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(chunk); + } + + if (close) { + controller.close(); + } + }, + cancel, + }); + +const buildUploadResponse = ({ + statusCode = 200, + chunks = [], +}: { + statusCode?: number; + chunks?: Buffer[]; +} = {}): IncomingMessage => { + const uploadResponse = Readable.from(chunks) as IncomingMessage; + + uploadResponse.statusCode = statusCode; + + return uploadResponse; +}; + +const buildUploadRequest = ({ + response, + emitResponseOnFinish = true, +}: { + response?: IncomingMessage; + emitResponseOnFinish?: boolean; +} = {}) => { + const uploadRequest = new PassThrough(); + const uploadedBytes: number[] = []; + + uploadRequest.on('data', (chunk: Buffer) => { + uploadedBytes.push(...chunk); + }); + + if (response !== undefined) { + if (emitResponseOnFinish) { + uploadRequest.on('finish', () => { + uploadRequest.emit('response', response); + }); + } else { + queueMicrotask(() => { + uploadRequest.emit('response', response); + }); + } + } + + return { + uploadedBytes, + uploadRequest: uploadRequest as unknown as ClientRequest, + }; +}; + +const putDefaultMediaDownloadBodyToUploadTarget = ({ + mediaDownloadBody = buildMediaDownloadBody(), + uploadUrl = HTTPS_UPLOAD_URL, +}: { + mediaDownloadBody?: ReadableStream; + uploadUrl?: string; +} = {}) => + putMediaDownloadBodyToUploadTarget({ + fileName: 'video.mp4', + mediaDownloadBody, + sizeBytes: 4, + uploadTarget: { + uploadUrl, + contentType: 'application/octet-stream', + }, + }); + +describe('putMediaDownloadBodyToUploadTarget', () => { + beforeEach(() => { + requestOverHttpMock.mockReset(); + requestOverHttpsMock.mockReset(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('streams the media download body to the upload target with declared headers', async () => { + const { uploadRequest, uploadedBytes } = buildUploadRequest({ + response: buildUploadResponse(), + }); + + requestOverHttpsMock.mockReturnValue(uploadRequest); + + await putDefaultMediaDownloadBodyToUploadTarget(); + + const [uploadUrl, uploadRequestOptions] = requestOverHttpsMock.mock.calls[0]; + + expect(uploadUrl.href).toBe(HTTPS_UPLOAD_URL); + expect(uploadRequestOptions).toMatchObject({ + method: 'PUT', + headers: { + 'Content-Length': 4, + 'Content-Type': 'application/octet-stream', + }, + }); + expect(uploadRequestOptions.signal).toBeInstanceOf(AbortSignal); + expect(uploadedBytes).toEqual([1, 2, 3, 4]); + }); + + it('uses the http client for http upload targets', async () => { + const { uploadRequest } = buildUploadRequest({ + response: buildUploadResponse(), + }); + + requestOverHttpMock.mockReturnValue(uploadRequest); + + await putDefaultMediaDownloadBodyToUploadTarget({ + uploadUrl: HTTP_UPLOAD_URL, + }); + + expect(requestOverHttpMock).toHaveBeenCalledTimes(1); + expect(requestOverHttpsMock).not.toHaveBeenCalled(); + }); + + it('destroys the media download readable when storage returns a failed status', async () => { + const mediaDownloadBodyCancelMock = vi.fn().mockResolvedValue(undefined); + const uploadResponse = buildUploadResponse({ + statusCode: 500, + chunks: [Buffer.from('storage failed')], + }); + const { uploadRequest } = buildUploadRequest({ + emitResponseOnFinish: false, + response: uploadResponse, + }); + + requestOverHttpsMock.mockReturnValue(uploadRequest); + + await expect( + putDefaultMediaDownloadBodyToUploadTarget({ + mediaDownloadBody: buildMediaDownloadBody({ + cancel: mediaDownloadBodyCancelMock, + close: false, + }), + }), + ).rejects.toThrow('upload of video.mp4 failed with status 500'); + + expect(mediaDownloadBodyCancelMock).toHaveBeenCalledTimes(1); + expect(uploadResponse.readableEnded).toBe(true); + }); + + it('destroys the media download readable when the upload request fails', async () => { + const mediaDownloadBodyCancelMock = vi.fn().mockResolvedValue(undefined); + const { uploadRequest } = buildUploadRequest(); + + requestOverHttpsMock.mockReturnValue(uploadRequest); + + queueMicrotask(() => { + uploadRequest.emit('error', new Error('upload socket closed')); + }); + + await expect( + putDefaultMediaDownloadBodyToUploadTarget({ + mediaDownloadBody: buildMediaDownloadBody({ + cancel: mediaDownloadBodyCancelMock, + close: false, + }), + }), + ).rejects.toThrow('upload socket closed'); + + expect(mediaDownloadBodyCancelMock).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/ingest-call-recording-media.util.ts b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/ingest-call-recording-media.util.ts index 308fb6aade..144ad7ee08 100644 --- a/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/ingest-call-recording-media.util.ts +++ b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/ingest-call-recording-media.util.ts @@ -8,6 +8,7 @@ import { AUDIO_FILE_TOO_LARGE_FAILURE_REASON, VIDEO_FILE_TOO_LARGE_FAILURE_REASON, } from 'src/logic-functions/constants/media-file-too-large-failure-reasons'; +import { putMediaDownloadBodyToUploadTarget } from 'src/logic-functions/flows/put-media-download-body-to-upload-target.util'; import { extractRecallMediaUrls } from 'src/logic-functions/recall-api/extract-recall-media-urls.util'; import { getRecallRecording } from 'src/logic-functions/recall-api/get-recall-recording.util'; import { type CallRecordingMediaFile } from 'src/logic-functions/types/call-recording-media-file.type'; @@ -24,11 +25,18 @@ type IngestMediaArtifactResult = | { outcome: 'too-large' } | { outcome: 'failed' }; -type DownloadMediaFileResult = - | { outcome: 'downloaded'; buffer: Buffer; contentType: string } - | { outcome: 'too-large'; sizeBytes: number | undefined }; +type OpenMediaDownloadResult = + | { outcome: 'opened'; body: ReadableStream; sizeBytes: number } + | { outcome: 'too-large'; sizeBytes: number }; + +type MediaUploadTarget = { + fileId: string; + uploadUrl: string; + contentType: string; +}; const MEDIA_DOWNLOAD_TIMEOUT_MS = 120_000; +const MEDIA_FILE_FOLDER = 'FilesField'; export const ingestCallRecordingMedia = async ({ callRecordingId, @@ -57,7 +65,7 @@ export const ingestCallRecordingMedia = async ({ const mediaUrls = extractRecallMediaUrls(recordingResult.recording); const metadataClient = new MetadataApiClient(); - // TODO: drop the size cap once core streams uploads without buffering whole files in memory. + // TODO: raise this cap via config, monitor streamed uploads in prod, then remove the cap once verified. const maxMediaFileSizeBytes = getMaxMediaFileSizeBytes(); const updateFields: CallRecordingMediaUpdateFields = {}; const tooLargeFailureReasons: string[] = []; @@ -125,37 +133,33 @@ const ingestMediaArtifact = async ({ maxMediaFileSizeBytes: number; }): Promise => { try { - const downloadResult = await downloadMediaFile({ + const download = await openMediaDownload({ callRecordingId, fileName, url, maxMediaFileSizeBytes, }); - if (downloadResult.outcome === 'too-large') { + if (download.outcome === 'too-large') { console.warn( - `[call-recorder] media-ingestion phase=artifact-too-large callRecordingId=${callRecordingId} fileName=${fileName} sizeBytes=${downloadResult.sizeBytes ?? 'unknown'} maxMediaFileSizeBytes=${maxMediaFileSizeBytes}`, + `[call-recorder] media-ingestion phase=artifact-too-large callRecordingId=${callRecordingId} fileName=${fileName} sizeBytes=${download.sizeBytes} maxMediaFileSizeBytes=${maxMediaFileSizeBytes}`, ); return { outcome: 'too-large' }; } - const { buffer, contentType } = downloadResult; - - console.log( - `[call-recorder] media-ingestion phase=artifact-upload-start callRecordingId=${callRecordingId} fileName=${fileName} downloadedBytes=${buffer.byteLength} contentType=${contentType} ${formatMemoryUsageForLog()}`, - ); - - const uploadedFile = await metadataClient.uploadFile( - buffer, + const fileId = await uploadMediaStreamToStorage({ + callRecordingId, + metadataClient, fileName, - contentType, fieldMetadataUniversalIdentifier, - ); + body: download.body, + sizeBytes: download.sizeBytes, + }); return { outcome: 'ingested', - files: [{ fileId: uploadedFile.id, label: fileName }], + files: [{ fileId, label: fileName }], }; } catch (error) { console.warn( @@ -166,7 +170,7 @@ const ingestMediaArtifact = async ({ } }; -const downloadMediaFile = async ({ +const openMediaDownload = async ({ callRecordingId, fileName, url, @@ -176,29 +180,44 @@ const downloadMediaFile = async ({ fileName: string; url: string; maxMediaFileSizeBytes: number; -}): Promise => { +}): Promise => { const response = await fetch(url, { signal: AbortSignal.timeout(MEDIA_DOWNLOAD_TIMEOUT_MS), }); - const contentType = - response.headers.get('content-type') ?? 'application/octet-stream'; const contentLengthBytes = parseContentLengthBytes( response.headers.get('content-length'), ); console.log( - `[call-recorder] media-ingestion phase=artifact-download-response callRecordingId=${callRecordingId} fileName=${fileName} responseStatus=${response.status} contentLengthBytes=${contentLengthBytes ?? 'unknown'} contentType=${contentType} ${formatMemoryUsageForLog()}`, + `[call-recorder] media-ingestion phase=artifact-download-response callRecordingId=${callRecordingId} fileName=${fileName} responseStatus=${response.status} contentLengthBytes=${contentLengthBytes ?? 'unknown'} ${formatMemoryUsageForLog()}`, ); if (!response.ok) { + await cancelMediaDownloadBody({ + callRecordingId, + fileName, + body: response.body, + }); + throw new Error(`download failed with status ${response.status}`); } - if ( - !isUndefined(contentLengthBytes) && - contentLengthBytes > maxMediaFileSizeBytes - ) { - await response.body?.cancel(); + if (isUndefined(contentLengthBytes)) { + await cancelMediaDownloadBody({ + callRecordingId, + fileName, + body: response.body, + }); + + throw new Error('download response is missing content-length'); + } + + if (contentLengthBytes > maxMediaFileSizeBytes) { + await cancelMediaDownloadBody({ + callRecordingId, + fileName, + body: response.body, + }); return { outcome: 'too-large', sizeBytes: contentLengthBytes }; } @@ -207,49 +226,126 @@ const downloadMediaFile = async ({ throw new Error('download returned no body'); } - return readBodyWithinSizeCap({ - body: response.body, - contentType, - maxMediaFileSizeBytes, + return { outcome: 'opened', body: response.body, sizeBytes: contentLengthBytes }; +}; + +const uploadMediaStreamToStorage = async ({ + callRecordingId, + metadataClient, + fileName, + fieldMetadataUniversalIdentifier, + body, + sizeBytes, +}: { + callRecordingId: string; + metadataClient: InstanceType; + fileName: string; + fieldMetadataUniversalIdentifier: string; + body: ReadableStream; + sizeBytes: number; +}): Promise => { + const uploadTarget = await createFileUploadTarget({ + metadataClient, + fileName, + sizeBytes, + fieldMetadataUniversalIdentifier, + }).catch(async (error) => { + await cancelMediaDownloadBody({ + callRecordingId, + fileName, + body, + }); + + throw error; + }); + + console.log( + `[call-recorder] media-ingestion phase=artifact-upload-start callRecordingId=${callRecordingId} fileName=${fileName} declaredBytes=${sizeBytes} ${formatMemoryUsageForLog()}`, + ); + + await putMediaDownloadBodyToUploadTarget({ + fileName, + mediaDownloadBody: body, + sizeBytes, + uploadTarget, + }); + + return completeFileUpload({ metadataClient, fileId: uploadTarget.fileId }); +}; + +const cancelMediaDownloadBody = async ({ + callRecordingId, + fileName, + body, +}: { + callRecordingId: string; + fileName: string; + body: ReadableStream | null; +}) => { + if (isNull(body)) { + return; + } + + await body.cancel().catch((error) => { + console.warn( + `[call-recorder] media-ingestion phase=download-body-cancel-failed callRecordingId=${callRecordingId} fileName=${fileName}: ${error instanceof Error ? error.message : String(error)}`, + ); }); }; -const readBodyWithinSizeCap = async ({ - body, - contentType, - maxMediaFileSizeBytes, +const createFileUploadTarget = async ({ + metadataClient, + fileName, + sizeBytes, + fieldMetadataUniversalIdentifier, }: { - body: ReadableStream; - contentType: string; - maxMediaFileSizeBytes: number; -}): Promise => { - const reader = body.getReader(); - const chunks: Uint8Array[] = []; - let downloadedBytes = 0; + metadataClient: InstanceType; + fileName: string; + sizeBytes: number; + fieldMetadataUniversalIdentifier: string; +}): Promise => { + const mutationResult = await metadataClient.mutation({ + createFileUpload: { + __args: { + filename: fileName, + size: sizeBytes, + fileFolder: MEDIA_FILE_FOLDER, + fieldMetadataUniversalIdentifier, + }, + fileId: true, + uploadUrl: true, + contentType: true, + }, + }); + const uploadTarget = mutationResult.createFileUpload; - for (;;) { - const { done, value } = await reader.read(); - - if (done) { - break; - } - - downloadedBytes += value.byteLength; - - if (downloadedBytes > maxMediaFileSizeBytes) { - await reader.cancel(); - - return { outcome: 'too-large', sizeBytes: undefined }; - } - - chunks.push(value); + if (isUndefined(uploadTarget)) { + throw new Error('createFileUpload mutation did not return an upload target'); } - return { - outcome: 'downloaded', - buffer: Buffer.concat(chunks), - contentType, - }; + return uploadTarget; +}; + +const completeFileUpload = async ({ + metadataClient, + fileId, +}: { + metadataClient: InstanceType; + fileId: string; +}): Promise => { + const mutationResult = await metadataClient.mutation({ + completeFileUpload: { + __args: { fileId }, + id: true, + }, + }); + const uploadedFileId = mutationResult.completeFileUpload?.id; + + if (isUndefined(uploadedFileId)) { + throw new Error('completeFileUpload mutation did not return a file id'); + } + + return uploadedFileId; }; const parseContentLengthBytes = ( diff --git a/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/put-media-download-body-to-upload-target.util.ts b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/put-media-download-body-to-upload-target.util.ts new file mode 100644 index 0000000000..7441242e43 --- /dev/null +++ b/packages/twenty-apps/public/call-recorder/src/logic-functions/flows/put-media-download-body-to-upload-target.util.ts @@ -0,0 +1,144 @@ +import { + request as requestOverHttp, + type ClientRequest, + type IncomingMessage, +} from 'node:http'; +import { request as requestOverHttps } from 'node:https'; +import { Readable } from 'node:stream'; +import { finished, pipeline } from 'node:stream/promises'; +import { type ReadableStream as NodeWebReadableStream } from 'node:stream/web'; + +type MediaUploadTarget = { + uploadUrl: string; + contentType: string; +}; + +const MEDIA_UPLOAD_TIMEOUT_MS = 14 * 60 * 1000; +const HTTP_STATUS_OK_LOWER_BOUND = 200; +const HTTP_STATUS_OK_UPPER_BOUND = 300; + +export const putMediaDownloadBodyToUploadTarget = async ({ + mediaDownloadBody, + fileName, + sizeBytes, + uploadTarget, +}: { + mediaDownloadBody: ReadableStream; + fileName: string; + sizeBytes: number; + uploadTarget: MediaUploadTarget; +}): Promise => { + // Use node:http instead of fetch here: fetch can buffer ReadableStream + // request bodies in memory, which OOMs Lambda for large recordings. + const mediaDownloadReadable = Readable.fromWeb( + mediaDownloadBody as NodeWebReadableStream, + ); + + await streamMediaDownloadReadableToUploadTarget({ + fileName, + mediaDownloadReadable, + sizeBytes, + uploadTarget, + }); +}; + +const streamMediaDownloadReadableToUploadTarget = async ({ + fileName, + mediaDownloadReadable, + sizeBytes, + uploadTarget, +}: { + fileName: string; + mediaDownloadReadable: Readable; + sizeBytes: number; + uploadTarget: MediaUploadTarget; +}): Promise => { + const uploadUrl = new URL(uploadTarget.uploadUrl); + const requestUpload = + uploadUrl.protocol === 'http:' ? requestOverHttp : requestOverHttps; + const uploadRequest = requestUpload(uploadUrl, { + method: 'PUT', + headers: { + 'Content-Type': uploadTarget.contentType, + 'Content-Length': sizeBytes, + }, + signal: AbortSignal.timeout(MEDIA_UPLOAD_TIMEOUT_MS), + }); + + const uploadResponsePromise = waitForUploadResponse({ uploadRequest }); + const uploadPipelinePromise = pipeline(mediaDownloadReadable, uploadRequest); + + let uploadResponse: IncomingMessage; + + try { + uploadResponse = await waitForUploadResponseOrPipelineFailure({ + uploadPipelinePromise, + uploadResponsePromise, + }); + } catch (error) { + mediaDownloadReadable.destroy(); + uploadRequest.destroy(); + await uploadPipelinePromise.catch(() => undefined); + + throw error; + } + + const uploadResponseBodyDrainPromise = drainUploadResponseBody({ + uploadResponse, + }); + const uploadStatusCode = uploadResponse.statusCode ?? 0; + + if (!isSuccessfulUploadStatusCode(uploadStatusCode)) { + const uploadError = new Error( + `upload of ${fileName} failed with status ${uploadStatusCode}`, + ); + + mediaDownloadReadable.destroy(uploadError); + uploadRequest.destroy(uploadError); + + await Promise.allSettled([ + uploadPipelinePromise, + uploadResponseBodyDrainPromise, + ]); + + throw uploadError; + } + + await Promise.all([uploadPipelinePromise, uploadResponseBodyDrainPromise]); +}; + +const waitForUploadResponse = ({ + uploadRequest, +}: { + uploadRequest: ClientRequest; +}): Promise => + new Promise((resolve, reject) => { + uploadRequest.once('response', resolve); + uploadRequest.once('error', reject); + }); + +const waitForUploadResponseOrPipelineFailure = async ({ + uploadPipelinePromise, + uploadResponsePromise, +}: { + uploadPipelinePromise: Promise; + uploadResponsePromise: Promise; +}): Promise => + Promise.race([ + uploadResponsePromise, + uploadPipelinePromise.then(async () => await uploadResponsePromise), + ]); + +const drainUploadResponseBody = async ({ + uploadResponse, +}: { + uploadResponse: IncomingMessage; +}): Promise => { + uploadResponse.resume(); + + await finished(uploadResponse); +}; + +const isSuccessfulUploadStatusCode = (uploadStatusCode: number): boolean => + uploadStatusCode >= HTTP_STATUS_OK_LOWER_BOUND && + uploadStatusCode < HTTP_STATUS_OK_UPPER_BOUND; diff --git a/packages/twenty-apps/public/call-recorder/yarn.lock b/packages/twenty-apps/public/call-recorder/yarn.lock index d763ec50b8..1e6bc7c2ba 100644 --- a/packages/twenty-apps/public/call-recorder/yarn.lock +++ b/packages/twenty-apps/public/call-recorder/yarn.lock @@ -1376,8 +1376,8 @@ __metadata: react: "npm:^19.0.0" react-dom: "npm:^19.0.0" sharp: "npm:^0.34.5" - twenty-client-sdk: "npm:2.19.0-alpha.1" - twenty-sdk: "npm:2.19.0-alpha.1" + twenty-client-sdk: "npm:2.19.0" + twenty-sdk: "npm:2.19.0" twenty-ui: "npm:^1.0.0-alpha.1" typescript: "npm:^5.9.3" vite-tsconfig-paths: "npm:^4.2.1" @@ -3588,22 +3588,22 @@ __metadata: languageName: node linkType: hard -"twenty-client-sdk@npm:2.19.0-alpha.1": - version: 2.19.0-alpha.1 - resolution: "twenty-client-sdk@npm:2.19.0-alpha.1" +"twenty-client-sdk@npm:2.19.0": + version: 2.19.0 + resolution: "twenty-client-sdk@npm:2.19.0" dependencies: "@genql/runtime": "npm:^2.10.0" esbuild: "npm:^0.28.1" graphql: "npm:^16.8.1" lodash: "npm:^4.17.21" prettier: "npm:^3.8.3" - checksum: 10c0/daeac1c1b439f2a5c6ecd5ac5a97b04e56832bdd2e5530f5269bf469fd776849c750ea103b6837910dab3378adc146d4703008f7374c44eb4263fae1e120eb7c + checksum: 10c0/647922bda96a98fa163bed50b1e9065e9fc6f624e7ee40abcbe99a4c4bbfdd9a54b19fd2b774e06e6f758dd63a63724daf5ee34d716d7fe1740c04f120d266c0 languageName: node linkType: hard -"twenty-sdk@npm:2.19.0-alpha.1": - version: 2.19.0-alpha.1 - resolution: "twenty-sdk@npm:2.19.0-alpha.1" +"twenty-sdk@npm:2.19.0": + version: 2.19.0 + resolution: "twenty-sdk@npm:2.19.0" dependencies: "@sniptt/guards": "npm:^0.2.0" axios: "npm:^1.16.0" @@ -3622,12 +3622,12 @@ __metadata: semver: "npm:7.6.3" sharp: "npm:^0.34.5" tinyglobby: "npm:^0.2.15" - twenty-client-sdk: "npm:2.19.0-alpha.1" + twenty-client-sdk: "npm:2.19.0" typescript: "npm:^5.9.3" uuid: "npm:^13.0.2" bin: twenty: dist/cli.cjs - checksum: 10c0/c36772f6e6193c24ff5c6ba7636f4ae3b6110fed5697bc94cfbd282ae290b8de10b0f1df2b86a2c38eaed8f5a57791222b1f49b0d46baef6bb4b12fd0622cc2f + checksum: 10c0/d5d4126997510112008116ae807f0cbbe46f7a46ca7c061ae91363edc0f009eb159a92fdeaaf16b6f636f2faffc43dd9355ddfeb6ec3506c440fb8f1f39fe9e2 languageName: node linkType: hard