feat: implement download progress tracking in StorageFileDownloader and related components

src/apps/renderer/pages/Widget/Item.tsx

@@ -7,16 +7,17 @@ import { fileIcon } from '../../assets/icons/getIcon';
 
 export function Item({ name, action, progress }: DriveOperationInfo) {
   const { translate } = useTranslationContext();
-  const progressDisplay = progress ? `${Math.ceil(progress * 100)}%` : '';
+  const hasProgress = progress !== undefined && progress !== null;
+  const progressDisplay = hasProgress ? `${Math.ceil(progress * 100)}%` : '';
 
   let description = '';
 
   if (action === 'DOWNLOADING') {
-    description = progress
+    description = hasProgress
       ? translate('widget.body.activity.operation.downloading')
       : translate('widget.body.activity.operation.decrypting');
   } else if (action === 'UPLOADING') {
-    description = progress
+    description = hasProgress
       ? translate('widget.body.activity.operation.uploading')
       : translate('widget.body.activity.operation.encrypting');
   } else if (action === 'DOWNLOADED') {

src/apps/shared/fs/read-stream-to-buffer.test.ts

@@ -0,0 +1,48 @@
+import { Readable } from 'stream';
+import { readStreamToBuffer } from './read-stream-to-buffer';
+import { calls } from 'tests/vitest/utils.helper';
+
+describe('readStreamToBuffer', () => {
+  it('returns the full buffer and reports progress', async () => {
+    const onProgress = vi.fn();
+    const stream = Readable.from([Buffer.from('he'), Buffer.from('llo')]);
+
+    const result = await readStreamToBuffer({ stream, onProgress });
+
+    expect(result).toEqual(Buffer.from('hello'));
+    calls(onProgress).toHaveLength(2);
+    calls(onProgress).toMatchObject([2, 5]);
+  });
+
+  it('returns an empty buffer and does not call onProgress when stream is empty', async () => {
+    const onProgress = vi.fn();
+    const stream = Readable.from([]);
+
+    const result = await readStreamToBuffer({ stream, onProgress });
+
+    expect(result).toEqual(Buffer.alloc(0));
+    calls(onProgress).toHaveLength(0);
+  });
+
+  it('handles a single chunk correctly', async () => {
+    const onProgress = vi.fn();
+    const stream = Readable.from([Buffer.from('world')]);
+
+    const result = await readStreamToBuffer({ stream, onProgress });
+
+    expect(result).toEqual(Buffer.from('world'));
+    calls(onProgress).toHaveLength(1);
+    calls(onProgress).toMatchObject([5]);
+  });
+
+  it('rejects when the stream emits an error', async () => {
+    const onProgress = vi.fn();
+    const stream = new Readable({
+      read() {
+        this.destroy(new Error('stream failure'));
+      },
+    });
+
+    await expect(readStreamToBuffer({ stream, onProgress })).rejects.toThrow('stream failure');
+  });
+});

src/apps/shared/fs/read-stream-to-buffer.ts

@@ -0,0 +1,28 @@
+import { Readable, Writable, pipeline } from 'stream';
+import { promisify } from 'util';
+
+const promisifiedPipeline = promisify(pipeline);
+
+type Props = {
+  stream: Readable;
+  onProgress: (bytesWritten: number) => void;
+};
+
+export async function readStreamToBuffer({ stream, onProgress }: Props) {
+  const bufferArray: any[] = [];
+  let bytesWritten = 0;
+
+  const bufferWriter = new Writable({
+    write: (chunk, _, callback) => {
+      bufferArray.push(chunk);
+      bytesWritten += chunk.length;
+      onProgress(bytesWritten);
+
+      callback();
+    },
+  });
+
+  await promisifiedPipeline(stream, bufferWriter);
+
+  return Buffer.concat(bufferArray);
+}

src/apps/shared/fs/write-readable-to-file.test.ts

@@ -0,0 +1,52 @@
+import fs from 'fs';
+import { Readable, PassThrough } from 'stream';
+import { writeReadableToFile } from './write-readable-to-file';
+import { calls } from 'tests/vitest/utils.helper';
+
+vi.mock('fs');
+
+const mockedFS = vi.mocked(fs, true);
+
+describe('writeReadableToFile', () => {
+  it('writes the readable to the given path and reports progress', async () => {
+    const onProgress = vi.fn();
+    const writable = new PassThrough();
+    mockedFS.createWriteStream.mockReturnValue(writable as unknown as fs.WriteStream);
+
+    const readable = Readable.from([Buffer.from('he'), Buffer.from('llo')]);
+
+    const promise = writeReadableToFile({
+      readable,
+      path: '/tmp/test-file.txt',
+      onProgress,
+    });
+
+    await promise;
+
+    expect(mockedFS.createWriteStream).toHaveBeenCalledWith('/tmp/test-file.txt');
+    calls(onProgress).toHaveLength(2);
+    calls(onProgress).toMatchObject([2, 5]);
+  });
+
+  it('rejects when the writable stream emits an error', async () => {
+    const onProgress = vi.fn();
+    const writable = new PassThrough();
+    mockedFS.createWriteStream.mockReturnValue(writable as unknown as fs.WriteStream);
+
+    const readable = new Readable({
+      read() {},
+    });
+
+    const promise = writeReadableToFile({
+      readable,
+      path: '/tmp/fail.txt',
+      onProgress,
+    });
+
+    const error = new Error('disk full');
+    writable.destroy(error);
+
+    await expect(promise).rejects.toThrow('disk full');
+    calls(onProgress).toHaveLength(0);
+  });
+});

src/apps/shared/fs/write-readable-to-file.ts

@@ -1,15 +1,26 @@
 import fs, { PathLike } from 'fs';
 import { Readable } from 'stream';
 
-export class WriteReadableToFile {
-  static write(readable: Readable, path: PathLike): Promise<void> {
-    const writableStream = fs.createWriteStream(path);
+type Props = {
+  readable: Readable;
+  path: PathLike;
+  onProgress: (bytesWritten: number) => void;
+};
 
-    readable.pipe(writableStream);
+export function writeReadableToFile({ readable, path, onProgress }: Props) {
+  const writableStream = fs.createWriteStream(path);
 
-    return new Promise<void>((resolve, reject) => {
-      writableStream.on('finish', resolve);
-      writableStream.on('error', reject);
-    });
-  }
+  let bytesWritten = 0;
+
+  readable.on('data', (chunk: Buffer) => {
+    bytesWritten += chunk.length;
+    onProgress(bytesWritten);
+  });
+
+  readable.pipe(writableStream);
+
+  return new Promise<void>((resolve, reject) => {
+    writableStream.on('finish', resolve);
+    writableStream.on('error', reject);
+  });
 }

src/context/shared/domain/DownloadProgressTracker.ts

@@ -1,5 +1,5 @@
 export abstract class DownloadProgressTracker {
-  abstract downloadStarted(name: string, extension: string, size: number): Promise<void>;
+  abstract downloadStarted(name: string, extension: string): Promise<void>;
 
   abstract downloadUpdate(
     name: string,
@@ -9,13 +9,6 @@ export abstract class DownloadProgressTracker {
       percentage: number;
     },
   ): Promise<void>;
-  abstract downloadFinished(
-    name: string,
-    extension: string,
-    size: number,
-    progress: {
-      elapsedTime: number;
-    },
-  ): Promise<void>;
+  abstract downloadFinished(name: string, extension: string): Promise<void>;
   abstract error(name: string, extension: string): Promise<void>;
 }

src/context/shared/infrastructure/MainProcess/MainProcessDownloadProgressTracker.ts

@@ -6,13 +6,12 @@ import { Service } from 'diod';
 
 @Service()
 export class MainProcessDownloadProgressTracker extends SyncMessenger implements DownloadProgressTracker {
-  async downloadStarted(name: string, extension: string, size: number): Promise<void> {
+  async downloadStarted(name: string, extension: string): Promise<void> {
     setTrayStatus('SYNCING');
 
     broadcastToWindows('sync-info-update', {
       action: 'DOWNLOADING',
       name: this.nameWithExtension(name, extension),
-      progress: 0,
     });
   }
 
@@ -28,14 +27,7 @@ export class MainProcessDownloadProgressTracker extends SyncMessenger implements
     });
   }
 
-  async downloadFinished(
-    name: string,
-    extension: string,
-    size: number,
-    progress: {
-      elapsedTime: number;
-    },
-  ): Promise<void> {
+  async downloadFinished(name: string, extension: string) {
     const nameWithExtension = this.nameWithExtension(name, extension);
 
     setTrayStatus('IDLE');

src/context/storage/StorageFiles/__mocks__/DownloadProgressTrackerMock.ts

@@ -1,28 +1,8 @@
 import { DownloadProgressTracker } from '../../../shared/domain/DownloadProgressTracker';
 
 export class DownloadProgressTrackerMock implements DownloadProgressTracker {
-  private downloadStartedMock = vi.fn();
-  private downloadUpdateMock = vi.fn();
-  private downloadFinishedMock = vi.fn();
-  private errorMock = vi.fn();
-
-  downloadStarted(name: string, extension: string, size: number): Promise<void> {
-    return this.downloadStartedMock(name, extension, size);
-  }
-
-  downloadUpdate(
-    name: string,
-    extension: string,
-    progress: { elapsedTime: number; percentage: number },
-  ): Promise<void> {
-    return this.downloadUpdateMock(name, extension, progress);
-  }
-
-  downloadFinished(name: string, extension: string, size: number, progress: { elapsedTime: number }): Promise<void> {
-    return this.downloadFinishedMock(name, extension, size, progress);
-  }
-
-  error(name: string, extension: string): Promise<void> {
-    return this.errorMock(name, extension);
-  }
+  downloadStarted = vi.fn();
+  downloadUpdate = vi.fn();
+  downloadFinished = vi.fn();
+  error = vi.fn();
 }

src/context/storage/StorageFiles/application/download/StorageFileDownloader/StorageFileDownloader.test.ts

@@ -34,31 +34,14 @@ describe('StorageFileDownloader', () => {
   });
 
   describe('registerEvents', () => {
-    it('should handle start download', async () => {
-      await sut.run(file, metadata);
-      expect(downloaderHandler.on).toHaveBeenCalledWith('start', expect.any(Function));
-    });
-
-    it('should handle download progress', async () => {
-      await sut.run(file, metadata);
-
-      expect(downloaderHandler.on).toHaveBeenCalledWith('progress', expect.any(Function));
-    });
-
     it('should handle download errors', async () => {
       await sut.run(file, metadata);
 
       expect(downloaderHandler.on).toHaveBeenCalledWith('error', expect.any(Function));
     });
-
-    it('should handle download finish', async () => {
-      await sut.run(file, metadata);
-
-      expect(downloaderHandler.on).toHaveBeenCalledWith('finish', expect.any(Function));
-    });
   });
 
-  it('should successfully download a file', async () => {
+  it('should successfully download a file and return download result', async () => {
     const mockStream = new Readable({
       read() {
         this.push('mock data');
@@ -68,9 +51,11 @@ describe('StorageFileDownloader', () => {
 
     downloaderHandler.download.mockResolvedValue(mockStream);
 
-    const stream = await sut.run(file, metadata);
+    const result = await sut.run(file, metadata);
 
-    expect(stream).toBeInstanceOf(Readable);
+    expect(result.stream).toBeInstanceOf(Readable);
+    expect(result.metadata).toEqual(metadata);
+    expect(result.handler).toBe(downloaderHandler);
     expect(downloaderHandler.download).toHaveBeenCalledWith(file);
   });
 });

src/context/storage/StorageFiles/application/download/StorageFileDownloader/StorageFileDownloader.ts

@@ -13,50 +13,24 @@ export class StorageFileDownloader {
     private readonly tracker: DownloadProgressTracker,
   ) {}
 
-  private async registerEvents(
-    handler: DownloaderHandler,
-    { name, type, size }: { name: string; type: string; size: number },
-  ) {
-    handler.on('start', () => {
-      this.tracker.downloadStarted(name, type, size);
-    });
-
-    handler.on('progress', (progress: number, elapsedTime: number) => {
-      this.tracker.downloadUpdate(name, type, {
-        elapsedTime,
-        percentage: progress,
-      });
-    });
-
-    handler.on('error', () => {
-      this.tracker.error(name, type);
-    });
-
-    handler.on('finish', () => {
-      this.tracker.downloadFinished(name, type, size, {
-        elapsedTime: handler.elapsedTime(),
-      });
-    });
-  }
-
   async run(
     file: StorageFile,
     metadata: {
       name: string;
       type: string;
       size: number;
     },
-  ): Promise<Readable> {
+  ): Promise<{ stream: Readable; metadata: typeof metadata; handler: DownloaderHandler }> {
     const downloader = this.managerFactory.downloader();
 
-    await this.registerEvents(downloader, metadata);
+    downloader.on('error', () => this.tracker.error(metadata.name, metadata.type));
 
     const stream = await downloader.download(file);
 
     logger.debug({
       msg: `stream created "${metadata.name}.${metadata.type}" with ${file.id.value}`,
     });
 
-    return stream;
+    return { stream, metadata, handler: downloader };
   }
 }