🚑 Check files that are missing in storage (#416)

* 🚑Check files that are missing in storage

tdrive/backend/node/src/core/platform/services/storage/connectors/S3/s3-service.ts

@@ -2,6 +2,7 @@ import * as Minio from "minio";
 import { logger } from "../../../../../../core/platform/framework";
 import { Readable } from "stream";
 import { StorageConnectorAPI, WriteMetadata } from "../../provider";
+import fs from "fs";
 
 export type S3Configuration = {
   bucket: string;
@@ -77,4 +78,27 @@ export default class S3ConnectorService implements StorageConnectorAPI {
     } catch (err) {}
     return false;
   }
+
+  async exists(path: string): Promise<boolean> {
+    logger.trace(`Reading file ... ${path}`);
+    const tries = 2;
+    for (let i = 0; i <= tries; i++) {
+      try {
+        const stat = await this.client.statObject(this.minioConfiguration.bucket, path);
+        if (stat?.size > 0) {
+          break;
+        }
+      } catch (e) {
+        logger.error(`Error getting information from S3`, e);
+      }
+
+      if (i === tries) {
+        logger.info(`Unable to get file after ${tries} tries:`);
+        return false;
+      }
+      await new Promise(r => setTimeout(r, 500));
+      logger.info(`File ${path} not found in S3 bucket, retrying...`);
+    }
+    return true;
+  }
 }

tdrive/backend/node/src/core/platform/services/storage/connectors/local/service.ts

@@ -72,4 +72,10 @@ export default class LocalConnectorService implements StorageConnectorAPI {
   delete(path: string): Promise<void> {
     return rm(this.getFullPath(path), { recursive: false, force: true });
   }
+
+  async exists(path: string): Promise<boolean> {
+    logger.trace(`Reading file ... ${path}`);
+    const fullPath = this.getFullPath(path);
+    return fs.existsSync(fullPath);
+  }
 }

tdrive/backend/node/src/core/platform/services/storage/index.ts

@@ -75,6 +75,10 @@ export default class StorageService extends TdriveService<StorageAPI> implements
     return this.homeDir;
   }
 
+  exists(path: string, options?: ReadOptions): Promise<boolean> {
+    return this.getConnector().exists(path, options);
+  }
+
   async write(path: string, stream: Stream, options?: WriteOptions): Promise<WriteMetadata> {
     try {
       if (options?.encryptionKey) {

tdrive/backend/node/src/core/platform/services/storage/provider.ts

@@ -43,6 +43,13 @@ export interface StorageConnectorAPI {
    */
   read(path: string, options?: ReadOptions, context?: ExecutionContext): Promise<Readable>;
 
+  /**
+   * Check that the file is exists
+   *
+   * @param path
+   */
+  exists(path: string, options?: ReadOptions, context?: ExecutionContext): Promise<boolean>;
+
   /**
    * Remove a path
    *

tdrive/backend/node/src/services/files/services/index.ts

@@ -11,6 +11,8 @@ import {
   CrudException,
   DeleteResult,
   ExecutionContext,
+  ListResult,
+  Pagination,
 } from "../../../core/platform/framework/api/crud-service";
 import gr from "../../global-resolver";
 import {
@@ -19,16 +21,38 @@ import {
 } from "../../../services/previews/types";
 import { PreviewFinishedProcessor } from "./preview";
 import _ from "lodash";
+import User from "../../user/entities/user";
+import { DriveFile } from "../../documents/entities/drive-file";
+import { FileVersion } from "../../documents/entities/file-version";
 
 export class FileServiceImpl {
   version: "1";
   repository: Repository<File>;
+  userRepository: Repository<User>;
+  documentRepository: Repository<DriveFile>;
+  versionRepository: Repository<FileVersion>;
   private algorithm = "aes-256-cbc";
   private max_preview_file_size = 50000000;
+  private checkConsistencyInProgress = false;
 
   async init(): Promise<this> {
     try {
       await Promise.all([(this.repository = await gr.database.getRepository<File>("files", File))]);
+      await Promise.all([
+        (this.userRepository = await gr.database.getRepository<User>("user", User)),
+      ]);
+      await Promise.all([
+        (this.documentRepository = await gr.database.getRepository<DriveFile>(
+          "drive_files",
+          DriveFile,
+        )),
+      ]);
+      await Promise.all([
+        (this.versionRepository = await gr.database.getRepository<FileVersion>(
+          "drive_file_versions",
+          FileVersion,
+        )),
+      ]);
       gr.platformServices.messageQueue.processor.addHandler(
         new PreviewFinishedProcessor(this, this.repository),
       );
@@ -308,6 +332,64 @@ export class FileServiceImpl {
 
     return new DeleteResult("files", fileToDelete, true);
   }
+
+  async checkConsistency(): Promise<any> {
+    const data = [];
+    if (!this.checkConsistencyInProgress) {
+      this.checkConsistencyInProgress = true;
+      let result: ListResult<FileVersion>;
+      let page: Pagination = { limitStr: "20" };
+      try {
+        do {
+          result = await this.versionRepository.find({}, { pagination: page });
+          //check that the file exists
+          const jobs: Promise<void>[] = [];
+          for (const version of result.getEntities()) {
+            if (version.file_metadata.external_id) {
+              const checkFile = async () => {
+                try {
+                  const file = await this.getFile({
+                    id: version.file_metadata.external_id,
+                    company_id: "00000000-0000-4000-0000-000000000000",
+                  });
+                  const exist = await gr.platformServices.storage.exists(getFilePath(file));
+                  if (exist) {
+                    logger.info(`File ${version.file_metadata.external_id} exists in S3`);
+                  } else {
+                    logger.info(`File ${version.file_metadata.external_id} DOES NOT exists in S3`);
+                    const doc = await this.documentRepository.findOne({
+                      id: version.drive_item_id,
+                      company_id: "00000000-0000-4000-0000-000000000000",
+                    });
+                    const user = await this.userRepository.findOne({ id: doc.creator });
+                    data.push({
+                      file_id: version.file_metadata.external_id,
+                      user: user?.email_canonical,
+                      doc: {
+                        id: version.drive_item_id,
+                        name: doc.name,
+                      },
+                    });
+                    console.log(`Missing files:: ${JSON.stringify(data)}`);
+                  }
+                } catch (e) {
+                  logger.warn(`Can't find ${version.file_metadata.external_id} in DB`);
+                }
+              };
+              jobs.push(checkFile.bind(this)());
+            }
+          }
+          await Promise.all(jobs);
+          console.log(`Missing files:: ${JSON.stringify(data)}`);
+          //go to next page
+          page = Pagination.fromPaginable(result.nextPage);
+        } while (page.page_token);
+      } finally {
+        this.checkConsistencyInProgress = false;
+      }
+    }
+    return data;
+  }
 }
 
 function getFilePath(entity: File): string {

tdrive/backend/node/src/services/files/web/controllers/files.ts

@@ -59,6 +59,11 @@ export class FileController {
     }
   }
 
+  async checkConsistency(request: FastifyRequest, response: FastifyReply): Promise<void> {
+    const data = await gr.services.files.checkConsistency();
+    response.send(data);
+  }
+
   async thumbnail(
     request: FastifyRequest<{ Params: { company_id: string; id: string; index: string } }>,
     response: FastifyReply,

tdrive/backend/node/src/services/files/web/routes.ts

@@ -46,6 +46,12 @@ const routes: FastifyPluginCallback = (fastify: FastifyInstance, options, next)
     preValidation: [fastify.authenticate],
     handler: fileController.delete.bind(fileController),
   });
+  fastify.route({
+    method: "GET",
+    url: "/check-files",
+    preValidation: [fastify.authenticate],
+    handler: fileController.checkConsistency.bind(fileController),
+  });
 
   next();
 };