diff --git a/package-lock.json b/package-lock.json index ab88c849c5..2f58896b27 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,17 +1,17 @@ { "name": "internxt", - "version": "2.6.0", + "version": "2.6.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "internxt", - "version": "2.6.0", + "version": "2.6.1", "hasInstallScript": true, "license": "AGPL-3.0", "dependencies": { "@internxt/drive-desktop-core": "0.1.7", - "@internxt/inxt-js": "^3.2.2", + "@internxt/inxt-js": "^3.3.1", "@internxt/scan": "^1.0.7", "@internxt/sdk": "1.17.4", "async": "^3.2.4", @@ -3556,9 +3556,9 @@ } }, "node_modules/@internxt/inxt-js": { - "version": "3.2.2", - "resolved": "https://registry.npmjs.org/@internxt/inxt-js/-/inxt-js-3.2.2.tgz", - "integrity": "sha512-URoQYmpv+iDqGSwAL5UfH+hy0hpEhLsnidREU8KHbQKsD4Nmmmn0ZCZKFEQAzPeyhsTPI2/sh5NcLCO/bSEPOg==", + "version": "3.3.1", + "resolved": "https://registry.npmjs.org/@internxt/inxt-js/-/inxt-js-3.3.1.tgz", + "integrity": "sha512-aqfhm77lvobWNzkofSid9k8xDWpT/VGrhe4yyjkVZ6LAhadIELny4kMk2aNC8uwZkdUFH9Kjx/jyxgy5q1naag==", "license": "ISC", "dependencies": { "@internxt/sdk": "1.16.2", @@ -22284,9 +22284,9 @@ } }, "node_modules/undici": { - "version": "7.27.0", - "resolved": "https://registry.npmjs.org/undici/-/undici-7.27.0.tgz", - "integrity": "sha512-+t2Z/GwkZQDtu00813aP66ygViGtPHKhhoFZpQKpKrE+9jIgES+Zw+mFNaDWOVRKiuJjuqKHzD3B1sfGg8+ZOQ==", + "version": "7.28.0", + "resolved": "https://registry.npmjs.org/undici/-/undici-7.28.0.tgz", + "integrity": "sha512-cRZYrTDwWznlnRiPjggAGxZXanty6M8RV1ff8Wm4LWXBp7/IG8v5DnOm74DtUBp9OONpK75YlPnIjQqX0dBDtA==", "license": "MIT", "engines": { "node": ">=20.18.1" diff --git a/package.json b/package.json index 314008f6be..6ffc37f947 100644 --- a/package.json +++ b/package.json @@ -197,7 +197,7 @@ }, "dependencies": { "@internxt/drive-desktop-core": "0.1.7", - "@internxt/inxt-js": "^3.2.2", + "@internxt/inxt-js": "^3.3.1", "@internxt/scan": "^1.0.7", "@internxt/sdk": "1.17.4", "async": "^3.2.4", diff --git a/src/backend/features/backup/upload/upload-content-to-environment.test.ts b/src/backend/features/backup/upload/upload-content-to-environment.test.ts index 6fe9a99b90..7aca18427d 100644 --- a/src/backend/features/backup/upload/upload-content-to-environment.test.ts +++ b/src/backend/features/backup/upload/upload-content-to-environment.test.ts @@ -1,5 +1,4 @@ import { Environment } from '@internxt/inxt-js'; -import { UploadOptions } from '@internxt/inxt-js/build/lib/core'; import { Readable } from 'stream'; import { deepMocked } from '../../../../../tests/vitest/utils.helper'; import { uploadContentToEnvironment } from './upload-content-to-environment'; @@ -13,38 +12,39 @@ vi.mock('../../../../infra/local-file-system/safe-access', () => ({ })); describe('upload-content-to-environment', () => { + type UploadOptions = { + source: NodeJS.ReadableStream; + fileSize: number; + progressCallback: (progress: number) => void; + abortSignal?: AbortSignal; + }; + const createReadStreamMock = deepMocked(fs.createReadStream); const safeAccessMock = vi.mocked(safeAccessModule.safeAccess); const SMALL_SIZE = 1024; - const LARGE_SIZE = 200 * 1024 * 1024; // > 100MB threshold let environment: Environment; let abortController: AbortController; let fakeStream: Readable; - let capturedOpts: UploadOptions; - - function makeActionState() { - return { stop: vi.fn() }; - } - - function makeUploadFn(actionState = makeActionState()) { - return vi.fn((_bucket: string, opts: UploadOptions) => { - capturedOpts = opts; - return actionState; - }); - } + let capturedOpts: UploadOptions | null; + let uploadMock: ReturnType Promise>>; beforeEach(() => { abortController = new AbortController(); - capturedOpts = undefined as unknown as UploadOptions; + capturedOpts = null; fakeStream = Object.assign(new Readable({ read() {} }), { close: vi.fn(), destroy: vi.fn() }); createReadStreamMock.mockReturnValue(fakeStream as ReturnType); safeAccessMock.mockResolvedValue({ data: undefined }); + uploadMock = vi.fn(async (_bucket: string, opts: UploadOptions) => { + capturedOpts = opts; + return 'default-contents-id'; + }); + environment = { - upload: makeUploadFn(), - uploadMultipartFile: makeUploadFn(), + upload: uploadMock, + uploadMultipartFile: vi.fn(), } as unknown as Environment; }); @@ -61,102 +61,81 @@ describe('upload-content-to-environment', () => { async function startUpload(size = SMALL_SIZE) { const promise = callUpload(size); await Promise.resolve(); - await Promise.resolve(); - await Promise.resolve(); return { promise }; } - function triggerFinished(err: Error | null, contentsId: string | null) { - capturedOpts.finishedCallback(err, contentsId); - } - it('should resolve with contentsId on successful upload', async () => { const contentsId = 'abc123'; - const { promise } = await startUpload(); - expect(capturedOpts).toBeDefined(); - triggerFinished(null, contentsId); + uploadMock.mockResolvedValue(contentsId); - const result = await promise; + const result = await callUpload(); expect(result.data).toBe(contentsId); expect(result.error).toBeUndefined(); }); - it('should use upload for files below multipart threshold', async () => { - const { promise } = await startUpload(SMALL_SIZE); - triggerFinished(null, 'id'); - - await promise; - - expect(environment.upload).toHaveBeenCalled(); - expect(environment.uploadMultipartFile).not.toHaveBeenCalled(); - }); - - it('should use uploadMultipartFile for files above multipart threshold', async () => { - const { promise } = await startUpload(LARGE_SIZE); - triggerFinished(null, 'id'); + it('should call upload with file size and abortSignal', async () => { + uploadMock.mockImplementation(async (_bucket: string, opts: UploadOptions) => { + capturedOpts = opts; + return 'id'; + }); - await promise; + await callUpload(SMALL_SIZE); - expect(environment.uploadMultipartFile).toHaveBeenCalled(); - expect(environment.upload).not.toHaveBeenCalled(); + expect(capturedOpts?.fileSize).toBe(SMALL_SIZE); + expect(capturedOpts?.abortSignal).toBe(abortController.signal); + expect(uploadMock).toHaveBeenCalled(); }); it('should return NOT_ENOUGH_SPACE error when upload fails with "Max space used"', async () => { - const { promise } = await startUpload(); - triggerFinished(new Error('Max space used'), null); + uploadMock.mockRejectedValue(new Error('Max space used')); - const result = await promise; + const result = await callUpload(); expect(result.error?.cause).toBe('NOT_ENOUGH_SPACE'); }); it('should return RATE_LIMITED error on 429 with retry_after from message', async () => { - const { promise } = await startUpload(); const err = Object.assign(new Error(JSON.stringify({ retry_after: 10 })), { status: 429 }); - triggerFinished(err, null); + uploadMock.mockRejectedValue(err); - const result = await promise; + const result = await callUpload(); expect(result.error?.cause).toBe('RATE_LIMITED'); expect(result.error?.message).toBe('10000'); }); it('should return RATE_LIMITED with default delay when retry_after is missing', async () => { - const { promise } = await startUpload(); const err = Object.assign(new Error('{}'), { status: 429 }); - triggerFinished(err, null); + uploadMock.mockRejectedValue(err); - const result = await promise; + const result = await callUpload(); expect(result.error?.cause).toBe('RATE_LIMITED'); expect(result.error?.message).toBe('30000'); }); it('should return INTERNAL_SERVER_ERROR on 500+ errors', async () => { - const { promise } = await startUpload(); const err = Object.assign(new Error('Server error'), { status: 500 }); - triggerFinished(err, null); + uploadMock.mockRejectedValue(err); - const result = await promise; + const result = await callUpload(); expect(result.error?.cause).toBe('INTERNAL_SERVER_ERROR'); }); it('should return UNKNOWN error for generic errors', async () => { - const { promise } = await startUpload(); - triggerFinished(new Error('Something went wrong'), null); + uploadMock.mockRejectedValue(new Error('Something went wrong')); - const result = await promise; + const result = await callUpload(); expect(result.error?.cause).toBe('UNKNOWN'); }); - it('should return UNKNOWN error when contentsId is null on success', async () => { - const { promise } = await startUpload(); - triggerFinished(null, null); + it('should return UNKNOWN error when contentsId is empty on success', async () => { + uploadMock.mockResolvedValue(''); - const result = await promise; + const result = await callUpload(); expect(result.error?.cause).toBe('UNKNOWN'); }); @@ -179,8 +158,7 @@ describe('upload-content-to-environment', () => { expect(result.error).toBe(accessError); expect(createReadStreamMock).not.toHaveBeenCalled(); - expect(environment.upload).not.toHaveBeenCalled(); - expect(environment.uploadMultipartFile).not.toHaveBeenCalled(); + expect(uploadMock).not.toHaveBeenCalled(); }); it('should return ACTION_NOT_PERMITTED when createReadStream throws EACCES', async () => { @@ -193,35 +171,49 @@ describe('upload-content-to-environment', () => { expect(result.error?.cause).toBe('ACTION_NOT_PERMITTED'); expect(safeAccessMock).toHaveBeenCalledWith({ absolutePath: '/some/file.txt' }); expect(createReadStreamMock).toHaveBeenCalledWith('/some/file.txt'); - expect(environment.upload).not.toHaveBeenCalled(); - expect(environment.uploadMultipartFile).not.toHaveBeenCalled(); + expect(uploadMock).not.toHaveBeenCalled(); expect(result.data).toBeUndefined(); }); it('should return ACTION_NOT_PERMITTED when the read stream emits EACCES', async () => { + uploadMock.mockImplementation( + async (_bucket: string, opts: UploadOptions) => + await new Promise((_resolve, reject) => { + opts.source.once('error', reject); + }), + ); + const { promise } = await startUpload(); fakeStream.emit('error', Object.assign(new Error('permission denied'), { code: 'EACCES' })); - triggerFinished(null, 'contents-id-after-stream-error'); const result = await promise; expect(result.error?.cause).toBe('ACTION_NOT_PERMITTED'); expect(safeAccessMock).toHaveBeenCalledWith({ absolutePath: '/some/file.txt' }); expect(createReadStreamMock).toHaveBeenCalledWith('/some/file.txt'); - expect(environment.upload).toHaveBeenCalledTimes(1); - expect(environment.uploadMultipartFile).not.toHaveBeenCalled(); + expect(uploadMock).toHaveBeenCalledTimes(1); expect(result.data).toBeUndefined(); }); - it('should stop the upload and destroy the stream when signal is aborted', async () => { - const actionState = makeActionState(); - (environment.upload as unknown as ReturnType) = makeUploadFn(actionState); + it('should abort the upload and destroy the stream when signal is aborted', async () => { + uploadMock.mockImplementation( + async (_bucket: string, opts: UploadOptions) => + await new Promise((_resolve, reject) => { + opts.abortSignal?.addEventListener( + 'abort', + () => { + reject(new Error('aborted')); + }, + { once: true }, + ); + }), + ); - await startUpload(); + const { promise } = await startUpload(); abortController.abort(); + await promise; - expect(actionState.stop).toHaveBeenCalled(); expect(fakeStream.destroy).toHaveBeenCalled(); }); }); diff --git a/src/backend/features/backup/upload/upload-content-to-environment.ts b/src/backend/features/backup/upload/upload-content-to-environment.ts index f072b39c66..3038267862 100644 --- a/src/backend/features/backup/upload/upload-content-to-environment.ts +++ b/src/backend/features/backup/upload/upload-content-to-environment.ts @@ -1,8 +1,6 @@ import { Environment } from '@internxt/inxt-js'; import { DriveDesktopError } from '../../../../context/shared/domain/errors/DriveDesktopError'; -import { MULTIPART_UPLOAD_SIZE_THRESHOLD } from '../../../../context/shared/domain/UploadConstants'; import { createReadStream } from 'node:fs'; -import { UploadStrategyFunction } from '@internxt/inxt-js/build/lib/core'; import { Result } from '../../../../context/shared/domain/Result'; import { logger } from '@internxt/drive-desktop-core/build/backend'; import { isError } from '../../../../shared/errors/is-error'; @@ -25,63 +23,53 @@ export async function uploadContentToEnvironment({ signal, }: ContentUploadParams): Promise> { try { - const uploadFn: UploadStrategyFunction = - size > MULTIPART_UPLOAD_SIZE_THRESHOLD - ? environment.uploadMultipartFile.bind(environment) - : environment.upload.bind(environment); - const accessResult = await safeAccess({ absolutePath: path }); if (accessResult.error) { return { error: accessResult.error }; } const readable = createReadStream(path); + const abortHandler = () => { + readable.destroy(); + }; - return new Promise>((resolve) => { - let settled = false; - const resolveOnce = (result: Result) => { - if (settled) return; - settled = true; - resolve(result); - }; + readable.on('error', (err: Error & { code?: unknown; status?: unknown }) => { + logger.error({ tag: 'BACKUPS', msg: '[ENVLFU READ STREAM ERROR]', err, path }); + }); - readable.on('error', (err: Error & { code?: unknown; status?: unknown }) => { - logger.error({ tag: 'BACKUPS', msg: '[ENVLFU READ STREAM ERROR]', err, path }); - resolveOnce({ error: mapEnvironmentUploadError(err) }); - }); + signal.addEventListener('abort', abortHandler, { once: true }); - const state = uploadFn(bucket, { + try { + const contentsId = await environment.upload(bucket, { source: readable, fileSize: size, - finishedCallback: (err, contentsId) => { - readable.close(); - - if (err) { - logger.error({ tag: 'BACKUPS', msg: '[ENVLFU UPLOAD ERROR]', err }); - return resolveOnce({ error: mapEnvironmentUploadError(err) }); - } - - if (!contentsId) { - logger.error({ tag: 'BACKUPS', msg: '[ENVLFU UPLOAD ERROR] No contentsId returned' }); - return resolveOnce({ error: new DriveDesktopError('UNKNOWN') }); - } - - resolveOnce({ data: contentsId }); - }, + abortSignal: signal, progressCallback: (progress: number) => { logger.debug({ tag: 'SYNC-ENGINE', msg: '[UPLOAD PROGRESS]', progress }); }, }); - signal.addEventListener( - 'abort', - () => { - state.stop(); - readable.destroy(); - }, - { once: true }, - ); - }); + if (!contentsId) { + logger.error({ tag: 'BACKUPS', msg: '[ENVLFU UPLOAD ERROR] No contentsId returned' }); + return { error: new DriveDesktopError('UNKNOWN') }; + } + + return { data: contentsId }; + } catch (err) { + logger.error({ tag: 'BACKUPS', msg: '[ENVLFU UPLOAD ERROR]', err }); + + if (isError(err)) { + return { error: mapEnvironmentUploadError(err) }; + } + + return { error: new DriveDesktopError('UNKNOWN') }; + } finally { + signal.removeEventListener('abort', abortHandler); + + if (!readable.destroyed) { + readable.close(); + } + } } catch (err) { if (isError(err)) { return { error: mapEnvironmentUploadError(err) }; diff --git a/src/backend/features/thumbnails/upload-thumbnail-to-bucket.test.ts b/src/backend/features/thumbnails/upload-thumbnail-to-bucket.test.ts index 85915f488a..7dfe64db81 100644 --- a/src/backend/features/thumbnails/upload-thumbnail-to-bucket.test.ts +++ b/src/backend/features/thumbnails/upload-thumbnail-to-bucket.test.ts @@ -1,60 +1,71 @@ import { Environment } from '@internxt/inxt-js'; import { uploadThumbnailToBucket } from './upload-thumbnail-to-bucket'; import { UPLOAD_TIMEOUT_MS } from './thumbnail.constants'; -import { UploadOptions } from '@internxt/inxt-js/build/lib/core'; import { partialSpyOn } from '../../../../tests/vitest/utils.helper'; -function environmentMock(upload: (bucket: string, options: UploadOptions) => void): Environment { +type UploadMock = ( + bucket: string, + options: { source: NodeJS.ReadableStream; fileSize: number; abortSignal?: AbortSignal }, +) => Promise; + +function environmentMock(upload: UploadMock): Environment { return { upload } as unknown as Environment; } describe('upload-thumbnail-to-bucket', () => { const bucket = 'test-bucket'; const buffer = Buffer.from('image-data'); + const setTimeoutMock = partialSpyOn(global, 'setTimeout'); const clearTimeoutMock = partialSpyOn(global, 'clearTimeout'); it('should return data with contentsId on successful upload', async () => { const contentsId = 'contents-id-123'; - const environment = environmentMock((_bucket, { finishedCallback }) => { - finishedCallback(null, contentsId); - }); + const environment = environmentMock(async () => contentsId); const { data, error } = await uploadThumbnailToBucket(environment, bucket, buffer); expect(error).toBeUndefined(); expect(data).toBe(contentsId); + expect(setTimeoutMock).toBeCalled(); expect(clearTimeoutMock).toBeCalled(); }); - it('should return error when finishedCallback receives an error', async () => { + it('should return error when upload rejects', async () => { const uploadError = new Error('bucket error'); - const environment = environmentMock((_bucket, { finishedCallback }) => { - finishedCallback(uploadError, null); - }); + const environment = environmentMock(async () => Promise.reject(uploadError)); const { error } = await uploadThumbnailToBucket(environment, bucket, buffer); expect(error).toBe(uploadError); + expect(setTimeoutMock).toHaveBeenCalled(); expect(clearTimeoutMock).toHaveBeenCalled(); }); - it('should return error when finishedCallback has no contentsId', async () => { - const environment = environmentMock((_bucket, { finishedCallback }) => { - finishedCallback(null, null); - }); + it('should return error when upload resolves with empty contentsId', async () => { + const environment = environmentMock(async () => ''); const { error } = await uploadThumbnailToBucket(environment, bucket, buffer); expect(error).toBeInstanceOf(Error); + expect(setTimeoutMock).toHaveBeenCalled(); expect(clearTimeoutMock).toHaveBeenCalled(); }); it('should return error when upload times out', async () => { vi.useFakeTimers(); - const environment = environmentMock(() => { - // never calls finishedCallback - }); + const environment = environmentMock( + (_bucket, { abortSignal }) => + new Promise((_resolve, reject) => { + abortSignal?.addEventListener( + 'abort', + () => { + reject(new Error('aborted')); + }, + { once: true }, + ); + }), + ); const resultPromise = uploadThumbnailToBucket(environment, bucket, buffer); vi.advanceTimersByTime(UPLOAD_TIMEOUT_MS); @@ -62,6 +73,8 @@ describe('upload-thumbnail-to-bucket', () => { const { error } = await resultPromise; expect(error).toBeInstanceOf(Error); + expect(error?.message).toBe('Thumbnail bucket upload timed out'); + expect(clearTimeoutMock).toHaveBeenCalled(); vi.useRealTimers(); }); diff --git a/src/backend/features/thumbnails/upload-thumbnail-to-bucket.ts b/src/backend/features/thumbnails/upload-thumbnail-to-bucket.ts index 6a952e5eba..03d4712c2f 100644 --- a/src/backend/features/thumbnails/upload-thumbnail-to-bucket.ts +++ b/src/backend/features/thumbnails/upload-thumbnail-to-bucket.ts @@ -3,34 +3,37 @@ import { Readable } from 'node:stream'; import { Result } from '../../../context/shared/domain/Result'; import { UPLOAD_TIMEOUT_MS } from './thumbnail.constants'; -export function uploadThumbnailToBucket( +export async function uploadThumbnailToBucket( environment: Environment, bucket: string, buffer: Buffer, ): Promise> { - let timeoutId: ReturnType; + const abortController = new AbortController(); + const timeoutId = setTimeout(() => { + abortController.abort(); + }, UPLOAD_TIMEOUT_MS); - const upload = new Promise>((resolve) => { - environment.upload(bucket, { + try { + const contentsId = await environment.upload(bucket, { source: Readable.from(buffer), fileSize: buffer.length, - finishedCallback: (err: Error | null, contentsId: string | null) => { - clearTimeout(timeoutId); - if (err) { - return resolve({ error: err }); - } - if (!contentsId) { - return resolve({ error: new Error('Upload finished but no contentsId returned') }); - } - resolve({ data: contentsId }); - }, + abortSignal: abortController.signal, progressCallback: () => {}, }); - }); - const timeout = new Promise>((resolve) => { - timeoutId = setTimeout(() => resolve({ error: new Error('Thumbnail bucket upload timed out') }), UPLOAD_TIMEOUT_MS); - }); + if (!contentsId) { + return { error: new Error('Upload finished but no contentsId returned') }; + } - return Promise.race([upload, timeout]); + return { data: contentsId }; + } catch (error: unknown) { + if (abortController.signal.aborted) { + return { error: new Error('Thumbnail bucket upload timed out') }; + } + + const uploadError = error instanceof Error ? error : new Error('Thumbnail bucket upload failed'); + return { error: uploadError }; + } finally { + clearTimeout(timeoutId); + } } diff --git a/src/context/storage/StorageFiles/infrastructure/download/EnvironmentContentFileDownloader.ts b/src/context/storage/StorageFiles/infrastructure/download/EnvironmentContentFileDownloader.ts index abd2b5c47e..2b6e3ad611 100644 --- a/src/context/storage/StorageFiles/infrastructure/download/EnvironmentContentFileDownloader.ts +++ b/src/context/storage/StorageFiles/infrastructure/download/EnvironmentContentFileDownloader.ts @@ -1,6 +1,4 @@ -import { ActionState } from '@internxt/inxt-js/build/api'; -import { DownloadOneShardStrategy } from '@internxt/inxt-js/build/lib/core'; -import { DownloadStrategyFunction } from '@internxt/inxt-js/build/lib/core/download/strategy'; +import { Environment } from '@internxt/inxt-js'; import { EventEmitter, Readable } from 'stream'; import { Stopwatch } from '../../../../../apps/shared/types/Stopwatch'; import { StorageFile } from '../../domain/StorageFile'; @@ -10,20 +8,19 @@ import { logger } from '@internxt/drive-desktop-core/build/backend'; export class EnvironmentContentFileDownloader implements DownloaderHandler { private eventEmitter: EventEmitter; private stopwatch: Stopwatch; - - private state: ActionState | null; + private abortController: AbortController | null; constructor( - private readonly fn: DownloadStrategyFunction, + private readonly environment: Environment, private readonly bucket: string, ) { this.eventEmitter = new EventEmitter(); this.stopwatch = new Stopwatch(); - this.state = null; + this.abortController = null; } forceStop(): void { - this.state?.stop(); + this.abortController?.abort(); } download(file: StorageFile): Promise { @@ -34,44 +31,53 @@ export class EnvironmentContentFileDownloader implements DownloaderHandler { return this.executeDownload(fileId); } - private executeDownload(fileId: string): Promise { + private async executeDownload(fileId: string): Promise { this.stopwatch.start(); this.eventEmitter.emit('start'); - return new Promise((resolve, reject) => { - try { - this.state = this.fn( - this.bucket, - fileId, - { - progressCallback: (progress: number) => { - this.eventEmitter.emit('progress', progress, this.elapsedTime()); - }, - finishedCallback: async (err: Error, stream: Readable) => { - this.stopwatch.finish(); - if (err) { - this.eventEmitter.emit('error', err); - return reject(err); - } - this.eventEmitter.emit('finish', this.elapsedTime()); + this.abortController = new AbortController(); - resolve(stream); - }, + try { + const result = Reflect.apply(this.environment.download, this.environment, [ + this.bucket, + fileId, + { + progressCallback: (progress: number) => { + this.eventEmitter.emit('progress', progress, this.elapsedTime()); }, - { - label: 'Dynamic', - params: { - useProxy: false, - chunkSize: 4096 * 1024, - }, + abortSignal: this.abortController.signal, + }, + { + label: 'Dynamic', + params: { + useProxy: false, + chunkSize: 4096 * 1024, }, - ); - } catch (err) { - logger.error({ msg: 'Error in downloader:', err }); - this.eventEmitter.emit('error', err); - return reject(err); + }, + ]); + + if (!(result instanceof Promise)) { + throw new Error('Environment download strategy must return a Promise'); } - }); + + const stream = await result; + + if (!(stream instanceof Readable)) { + throw new Error('Download stream not available'); + } + + this.eventEmitter.emit('finish', this.elapsedTime()); + + return stream; + } catch (err: unknown) { + const downloadError = err instanceof Error ? err : new Error('Download failed'); + logger.error({ msg: 'Error in downloader', err: downloadError }); + this.eventEmitter.emit('error', downloadError); + throw downloadError; + } finally { + this.stopwatch.finish(); + this.abortController = null; + } } on(event: keyof DownloadEvents, handler: DownloadEvents[keyof DownloadEvents]): void { diff --git a/src/context/storage/StorageFiles/infrastructure/download/EnvironmentRemoteFileContentsManagersFactory.ts b/src/context/storage/StorageFiles/infrastructure/download/EnvironmentRemoteFileContentsManagersFactory.ts index 5e304adee7..d3afd26e83 100644 --- a/src/context/storage/StorageFiles/infrastructure/download/EnvironmentRemoteFileContentsManagersFactory.ts +++ b/src/context/storage/StorageFiles/infrastructure/download/EnvironmentRemoteFileContentsManagersFactory.ts @@ -12,6 +12,6 @@ export class EnvironmentFileDownloaderHandlerFactory implements DownloaderHandle ) {} downloader(): DownloaderHandler { - return new EnvironmentContentFileDownloader(this.environment.download, this.bucket); + return new EnvironmentContentFileDownloader(this.environment, this.bucket); } } diff --git a/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploader.ts b/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploader.ts index cba89a392f..b7f86e8b8f 100644 --- a/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploader.ts +++ b/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploader.ts @@ -26,36 +26,42 @@ export class EnvironmentTemporalFileUploader { this.eventEmitter.emit('start'); this.stopwatch.start(); - return new Promise((resolve, reject) => { - const state = this.fn(this.bucket, { + const abortHandler = () => { + contents.destroy(); + }; + + if (this.abortSignal) { + this.abortSignal.addEventListener('abort', abortHandler, { once: true }); + } + + try { + const contentsId = await this.fn(this.bucket, { source: contents, fileSize: size, - finishedCallback: (err: Error | null, contentsId: string) => { - this.stopwatch.finish(); - - if (err) { - this.eventEmitter.emit('error', err); - return reject(err); - } - this.eventEmitter.emit('finish', contentsId); - resolve(contentsId); - }, + abortSignal: this.abortSignal, progressCallback: (progress: number) => { this.eventEmitter.emit('progress', progress); }, }); + this.eventEmitter.emit('finish', contentsId); + return contentsId; + } catch (error) { + const uploadError = error instanceof Error ? error : new Error('Upload failed'); + + if (!contents.destroyed) { + contents.destroy(uploadError); + } + + this.eventEmitter.emit('error', uploadError); + throw uploadError; + } finally { + this.stopwatch.finish(); + if (this.abortSignal) { - this.abortSignal.addEventListener( - 'abort', - () => { - state.stop(); - contents.destroy(); - }, - { once: true }, - ); + this.abortSignal.removeEventListener('abort', abortHandler); } - }); + } } on(event: keyof UploadEvents, handler: UploadEvents[keyof UploadEvents]): void { diff --git a/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploaderFactory.ts b/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploaderFactory.ts index becd588edb..efc16e220f 100644 --- a/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploaderFactory.ts +++ b/src/context/storage/TemporalFiles/infrastructure/upload/EnvironmentTemporalFileUploaderFactory.ts @@ -7,7 +7,6 @@ import { EnvironmentTemporalFileUploader } from './EnvironmentTemporalFileUpload import { TemporalFileUploaderFactory } from '../../domain/upload/TemporalFileUploaderFactory'; import { Replaces } from '../../domain/upload/Replaces'; import { TemporalFile } from '../../domain/TemporalFile'; -import { MULTIPART_UPLOAD_SIZE_THRESHOLD } from '../../../../shared/domain/UploadConstants'; @Service() export class EnvironmentTemporalFileUploaderFactory implements TemporalFileUploaderFactory { @@ -92,11 +91,7 @@ export class EnvironmentTemporalFileUploaderFactory implements TemporalFileUploa throw new Error('Readable is needed to upload a file'); } - const fn = - document.size.value > MULTIPART_UPLOAD_SIZE_THRESHOLD - ? this.environment.uploadMultipartFile.bind(this.environment) - : this.environment.upload.bind(this.environment); - + const fn = this.environment.upload.bind(this.environment); const uploader = new EnvironmentTemporalFileUploader(fn, this.bucket, this._abortController?.signal); this.registerEvents(uploader);