fix(call-recorder): stream Recall media to storage to fix OOM (#22652)
## Summary Fixes Call Recorder media ingestion OOMs by streaming Recall media into Twenty direct uploads instead of buffering the full file in memory. ## Changes - Opens the Recall media download stream and uses its `Content-Length` as the direct upload size. - Creates a Twenty direct upload target, streams the media body to it with Node `http`/`https` backpressure, then completes the upload. - Cleans up download/upload streams on target creation, upload, and storage response failures. - Keeps the media size cap for now while making it no longer required for memory safety. - Bumps `twenty-client-sdk` and `twenty-sdk` to `2.19.0`. ## Tests - `yarn test:unit src/logic-functions/flows/__tests__/ingest-call-recording-media.test.ts src/logic-functions/flows/__tests__/put-media-download-body-to-upload-target.test.ts` - `yarn typecheck`
This commit is contained in:
nitin
@@ -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",
|
||||
|
||||
+244
-118
@@ -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<Uint8Array> =>
|
||||
new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
@@ -37,16 +49,14 @@ const buildBodyStream = (chunks: Uint8Array[]): ReadableStream<Uint8Array> =>
|
||||
},
|
||||
});
|
||||
|
||||
const buildFetchResponse = ({
|
||||
contentType = 'video/mp4',
|
||||
const buildDownloadResponse = ({
|
||||
contentLengthBytes,
|
||||
body,
|
||||
}: {
|
||||
contentType?: string;
|
||||
contentLengthBytes?: number;
|
||||
body?: unknown;
|
||||
} = {}) => {
|
||||
const headers = new Map<string, string>([['content-type', contentType]]);
|
||||
const headers = new Map<string, string>();
|
||||
|
||||
if (contentLengthBytes !== undefined) {
|
||||
headers.set('content-length', String(contentLengthBytes));
|
||||
@@ -62,38 +72,105 @@ const buildFetchResponse = ({
|
||||
};
|
||||
};
|
||||
|
||||
const stubFetchByUrl = (responsesByUrl: Record<string, unknown>) => {
|
||||
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<string, unknown>;
|
||||
}) => {
|
||||
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<string, string>;
|
||||
createFileUploadErrorByFilename?: Record<string, Error>;
|
||||
}) => {
|
||||
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),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
+210
@@ -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<typeof import('node:http')>(
|
||||
'node:http',
|
||||
);
|
||||
|
||||
return { ...actualHttp, request: requestOverHttpMock };
|
||||
});
|
||||
|
||||
vi.mock('node:https', async () => {
|
||||
const actualHttps = await vi.importActual<typeof import('node:https')>(
|
||||
'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<void>;
|
||||
close?: boolean;
|
||||
} = {}): ReadableStream<Uint8Array> =>
|
||||
new ReadableStream<Uint8Array>({
|
||||
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<Uint8Array>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
+160
-64
@@ -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<Uint8Array>; 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<IngestMediaArtifactResult> => {
|
||||
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<DownloadMediaFileResult> => {
|
||||
}): Promise<OpenMediaDownloadResult> => {
|
||||
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<typeof MetadataApiClient>;
|
||||
fileName: string;
|
||||
fieldMetadataUniversalIdentifier: string;
|
||||
body: ReadableStream<Uint8Array>;
|
||||
sizeBytes: number;
|
||||
}): Promise<string> => {
|
||||
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<Uint8Array> | 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<Uint8Array>;
|
||||
contentType: string;
|
||||
maxMediaFileSizeBytes: number;
|
||||
}): Promise<DownloadMediaFileResult> => {
|
||||
const reader = body.getReader();
|
||||
const chunks: Uint8Array[] = [];
|
||||
let downloadedBytes = 0;
|
||||
metadataClient: InstanceType<typeof MetadataApiClient>;
|
||||
fileName: string;
|
||||
sizeBytes: number;
|
||||
fieldMetadataUniversalIdentifier: string;
|
||||
}): Promise<MediaUploadTarget> => {
|
||||
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<typeof MetadataApiClient>;
|
||||
fileId: string;
|
||||
}): Promise<string> => {
|
||||
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 = (
|
||||
|
||||
+144
@@ -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<Uint8Array>;
|
||||
fileName: string;
|
||||
sizeBytes: number;
|
||||
uploadTarget: MediaUploadTarget;
|
||||
}): Promise<void> => {
|
||||
// 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<Uint8Array>,
|
||||
);
|
||||
|
||||
await streamMediaDownloadReadableToUploadTarget({
|
||||
fileName,
|
||||
mediaDownloadReadable,
|
||||
sizeBytes,
|
||||
uploadTarget,
|
||||
});
|
||||
};
|
||||
|
||||
const streamMediaDownloadReadableToUploadTarget = async ({
|
||||
fileName,
|
||||
mediaDownloadReadable,
|
||||
sizeBytes,
|
||||
uploadTarget,
|
||||
}: {
|
||||
fileName: string;
|
||||
mediaDownloadReadable: Readable;
|
||||
sizeBytes: number;
|
||||
uploadTarget: MediaUploadTarget;
|
||||
}): Promise<void> => {
|
||||
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<IncomingMessage> =>
|
||||
new Promise<IncomingMessage>((resolve, reject) => {
|
||||
uploadRequest.once('response', resolve);
|
||||
uploadRequest.once('error', reject);
|
||||
});
|
||||
|
||||
const waitForUploadResponseOrPipelineFailure = async ({
|
||||
uploadPipelinePromise,
|
||||
uploadResponsePromise,
|
||||
}: {
|
||||
uploadPipelinePromise: Promise<void>;
|
||||
uploadResponsePromise: Promise<IncomingMessage>;
|
||||
}): Promise<IncomingMessage> =>
|
||||
Promise.race([
|
||||
uploadResponsePromise,
|
||||
uploadPipelinePromise.then(async () => await uploadResponsePromise),
|
||||
]);
|
||||
|
||||
const drainUploadResponseBody = async ({
|
||||
uploadResponse,
|
||||
}: {
|
||||
uploadResponse: IncomingMessage;
|
||||
}): Promise<void> => {
|
||||
uploadResponse.resume();
|
||||
|
||||
await finished(uploadResponse);
|
||||
};
|
||||
|
||||
const isSuccessfulUploadStatusCode = (uploadStatusCode: number): boolean =>
|
||||
uploadStatusCode >= HTTP_STATUS_OK_LOWER_BOUND &&
|
||||
uploadStatusCode < HTTP_STATUS_OK_UPPER_BOUND;
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user