.

src/api/client/message-api/file-uploader.ts

@@ -9,7 +9,7 @@ import path from 'path';
 import { ITDLibClient, TDLibApi } from 'src/api/interface';
 import { SendMessageResp, UploadedFile } from 'src/api/types';
 import { Queue, generateFileId, getAppropriatedPartSize } from 'src/api/utils';
-import { AggregatedError, TechnicalError } from 'src/errors/base';
+import { AggregatedError } from 'src/errors/base';
 import { FileTooBig } from 'src/errors/telegram';
 import { manager } from 'src/server/manager';
 import { Logger } from 'src/utils/logger';
@@ -35,13 +35,25 @@ export abstract class FileUploader<T extends GeneralFileMessage> {
 
   private _errors: { [key: number]: Error } = {};
 
+  private _uploadingChunks: Array<{ chunk: Buffer; filePart: number }>;
+
   constructor(
     protected readonly client: ITDLibClient,
     protected readonly fileSize: bigInt.BigInteger,
     private readonly onComplete: () => Promise<void>,
+    private readonly workers: {
+      small?: number;
+      big?: number;
+    } = {
+      small: 3,
+      big: 15,
+    },
   ) {
     this.fileId = generateFileId();
     this.isBig = isBig(fileSize);
+    this._uploadingChunks = new Array(
+      this.isBig ? this.workers.big : this.workers.small,
+    );
   }
 
   protected abstract get defaultFileName(): string;
@@ -60,24 +72,11 @@ export abstract class FileUploader<T extends GeneralFileMessage> {
       : quotient.toJSNumber() + 1;
   }
 
-  private async uploadNextPart(workerId: number): Promise<bigInt.BigInteger> {
-    if (this.done()) {
-      return bigInt.zero;
-    }
-
-    const chunkLength =
-      this.uploaded.add(this.chunkSize) > this.fileSize
-        ? this.fileSize.minus(this.uploaded)
-        : this.chunkSize;
-    this.uploaded = this.uploaded.add(chunkLength);
-    const filePart = this.partCnt++; // 0-indexed
-
-    if (chunkLength.eq(0)) {
-      return bigInt.zero;
-    }
-
-    const chunk = await this.read(chunkLength.toJSNumber());
-
+  private async saveChunk(
+    workerId: number,
+    chunk: Buffer,
+    filePart: number,
+  ): Promise<void> {
     while (true) {
       try {
         const rsp = this.isBig
@@ -92,39 +91,68 @@ export abstract class FileUploader<T extends GeneralFileMessage> {
               filePart,
               bytes: chunk,
             });
-        Logger.debug(
-          `[worker ${workerId}] uploaded chunk ${filePart} success=${rsp.success}`,
-        );
-        if (!rsp.success) {
-          throw new TechnicalError(
-            `File chunk ${filePart} of ${this.fileName} failed to upload`,
-          );
+        if (rsp.success) {
+          this._uploadingChunks[workerId] = null;
+          return;
         }
-        return chunkLength;
       } catch (err) {
         if (err instanceof RPCError) {
           if (err.errorMessage === 'FILE_PARTS_INVALID') {
             throw new FileTooBig(this.fileSize);
           }
         }
+        console.error(err);
+      }
+    }
+  }
+
+  private async saveUncompletedChunks() {
+    while (true) {
+      const unfinishedChunks = this._uploadingChunks.filter((x) => x !== null);
+
+      if (unfinishedChunks.length === 0) {
+        break;
+      }
+
+      for (const { chunk, filePart } of unfinishedChunks) {
+        if (chunk) {
+          await this.saveChunk(0, chunk, filePart);
+        }
       }
     }
   }
 
+  private async uploadNextPart(workerId: number): Promise<bigInt.BigInteger> {
+    if (this.done()) {
+      return bigInt.zero;
+    }
+
+    const chunkLength =
+      this.uploaded.add(this.chunkSize) > this.fileSize
+        ? this.fileSize.minus(this.uploaded)
+        : this.chunkSize;
+    this.uploaded = this.uploaded.add(chunkLength);
+    const filePart = this.partCnt++; // 0-indexed
+
+    if (chunkLength.eq(0)) {
+      return bigInt.zero;
+    }
+
+    const chunk = await this.read(chunkLength.toJSNumber());
+    this._uploadingChunks[workerId] = { chunk, filePart };
+
+    await this.saveChunk(workerId, chunk, filePart);
+
+    return chunkLength;
+  }
+
   public async upload(
     file: T,
     callback?: (
       uploaded: bigInt.BigInteger,
       totalSize: bigInt.BigInteger,
     ) => void,
     fileName?: string,
-    workers: {
-      small?: number;
-      big?: number;
-    } = {
-      small: 3,
-      big: 15,
-    },
   ): Promise<void> {
     const task = manager.createUploadTask(this.fileName, this.fileSize);
     this.prepare(file);
@@ -149,6 +177,7 @@ export abstract class FileUploader<T extends GeneralFileMessage> {
           }
           if (!onCompleteEmitted) {
             onCompleteEmitted = true;
+            await this.saveUncompletedChunks();
             await this.onComplete();
           }
           return true;
@@ -160,7 +189,7 @@ export abstract class FileUploader<T extends GeneralFileMessage> {
 
       const promises: Array<Promise<boolean>> = [];
 
-      const numWorkers = this.isBig ? workers.big : workers.small;
+      const numWorkers = this.isBig ? this.workers.big : this.workers.small;
       for (let i = 0; i < numWorkers; i++) {
         promises.push(createWorker(i));
       }

src/api/ops/create-dir.ts

@@ -1,11 +1,13 @@
 import { Client } from 'src/api';
 import { RelativePathError } from 'src/errors/path';
+import { Logger } from 'src/utils/logger';
 
 import { navigateToDir } from './navigate-to-dir';
 import { splitPath } from './utils';
 
 export const createDir =
   (client: Client) => async (path: string, parents: boolean) => {
+    Logger.info(`Creating directory ${path}`);
     if (!parents) {
       const [basePath, name] = splitPath(path);
       const dir = await navigateToDir(client)(basePath);

src/server/webdav/tgfs-filesystem.ts

@@ -277,7 +277,7 @@ export class TGFSFileSystem extends FileSystem {
           fileRef,
           fileRef.name,
         );
-        
+
         callback(null, Readable.from(chunks));
       } catch (err) {
         handleError(callback)(err);