✨Resilience strategy to read and write only to one storage

tdrive/backend/node/src/core/config/index.ts

@@ -1,3 +1,3 @@
 import config from "config";
 
-export default config;
+export default config;

tdrive/backend/node/src/core/platform/services/storage/default-storage-strategy.ts

@@ -0,0 +1,42 @@
+import { Readable, Stream } from "stream";
+import { ExecutionContext } from "../../framework/api/crud-service";
+import {
+  DeleteOptions,
+  ReadOptions,
+  StorageConnectorAPI,
+  WriteMetadata,
+  WriteOptions,
+} from "../storage/provider";
+
+export class DefaultStorageStrategy implements StorageConnectorAPI {
+  private connector: StorageConnectorAPI;
+
+  constructor(connector: StorageConnectorAPI) {
+    this.connector = connector;
+  }
+
+  write = (
+    path: string,
+    stream: Stream,
+    options?: WriteOptions,
+    context?: ExecutionContext,
+  ): Promise<WriteMetadata> => {
+    return this.connector.write(path, stream, options, context);
+  };
+
+  read = (path: string, options?: ReadOptions, context?: ExecutionContext): Promise<Readable> => {
+    return this.connector.read(path, options, context);
+  };
+
+  exists = (path: string, options?: ReadOptions, context?: ExecutionContext): Promise<boolean> => {
+    return this.connector.exists(path, options, context);
+  };
+
+  remove = (
+    path: string,
+    options?: DeleteOptions,
+    context?: ExecutionContext,
+  ): Promise<boolean> => {
+    return this.connector.remove(path, options, context);
+  };
+}

tdrive/backend/node/src/core/platform/services/storage/oneof-storage-strategy.ts

@@ -0,0 +1,115 @@
+import { Readable, Stream, PassThrough } from "stream";
+import {
+  DeleteOptions,
+  ReadOptions,
+  StorageConnectorAPI,
+  WriteMetadata,
+  WriteOptions,
+} from "../storage/provider";
+import { logger } from "../../../platform/framework";
+
+/**
+ * OneOfStorageStrategy is responsible for managing multiple storage backends.
+ * It writes to, reads from, removes, and checks for the existence of files
+ * across multiple storage systems.
+ */
+export class OneOfStorageStrategy implements StorageConnectorAPI {
+  constructor(private readonly storages: StorageConnectorAPI[]) {}
+
+  /**
+   * Writes a file to all configured storages.
+   * The write operation is considered successful if one of the storage succeed.
+   * @param path - The path where the file should be written.
+   * @param stream - The input stream to be written to storages.
+   * @param options - Write option: chunk number, encryption key, etc ...
+   * @throws Error if the write operation fails for one or more storages.
+   */
+  write = async (path: string, stream: Stream, options?: WriteOptions): Promise<WriteMetadata> => {
+    logger.debug("Creating streams for all storages ...");
+    const passThroughStreams = this.storages.map(() => new PassThrough());
+
+    // destroy all streams if there is an error in the input stream
+    stream.on("error", err => {
+      logger.error("Error in input stream, destroying all write streams:", err);
+      passThroughStreams.forEach(stream => stream.destroy(err));
+    });
+
+    // Pipe the input stream to each PassThrough stream
+    stream
+      .pipe(new PassThrough())
+      .on("data", chunk => {
+        passThroughStreams.forEach(stream => stream.write(chunk));
+      })
+      .on("end", () => {
+        passThroughStreams.forEach(stream => stream.end());
+      });
+
+    // Write to all storages with error handling
+    const writeResults = await Promise.allSettled(
+      this.storages.map(
+        (storage, index) => storage.write(path, passThroughStreams[index]),
+        options,
+      ),
+    );
+
+    // Log all errors and throw if all write operations fail
+    const errors = writeResults.filter(result => result.status === "rejected");
+    errors.forEach((error, index) => {
+      logger.error(`Error writing to storage ${index}:`, (error as PromiseRejectedResult).reason);
+    });
+    if (errors.length === this.storages.length) {
+      throw new Error("Write operation failed for all storages");
+    }
+
+    const successResult = writeResults.filter(
+      result => result.status === "fulfilled",
+    )[0] as PromiseFulfilledResult<WriteMetadata>;
+    return successResult.value;
+  };
+
+  /**
+   * Reads a file from the primary storage. If it fails, attempts to read from other storages.
+   * @param path - The path of the file to be read.
+   * @param options
+   * @returns A readable stream of the file.
+   * @throws Error if all storage read attempts fail.
+   */
+  read = async (path: string, options?: ReadOptions): Promise<Readable> => {
+    for (const storage of this.storages) {
+      try {
+        if (await storage.exists(path, options)) {
+          return await storage.read(path, options);
+        }
+      } catch (err) {
+        logger.error("Fallback storage read failed:", err);
+      }
+    }
+    throw new Error(`Error reading ${path} in all the storages.`);
+  };
+
+  /**
+   * Checks if a file exists in any of the configured storages.
+   * @param path - The path of the file to check.
+   * @param options
+   * @returns A boolean indicating whether the file exists in any storage.
+   */
+  exists = async (path: string, options?: ReadOptions): Promise<boolean> => {
+    for (const storage of this.storages) {
+      if (await storage.exists(path, options)) {
+        return true;
+      }
+    }
+    return false;
+  };
+
+  /**
+   * Removes a file from all configured storages.
+   * @param path - The path of the file to be removed.
+   * @param options
+   */
+  remove = async (path: string, options?: DeleteOptions): Promise<boolean> => {
+    return Promise.all(this.storages.map(storage => storage.remove(path, options))).then(array => {
+      return array.reduce((a, b) => a && b);
+    });
+  };
+}

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

@@ -9,7 +9,7 @@ import {
   TdriveServiceOptions,
 } from "../../framework";
 import LocalConnectorService, { LocalConfiguration } from "./connectors/local/service";
-import S3ConnectorService from "./connectors/S3/s3-service";
+import S3ConnectorService, { S3Configuration } from "./connectors/S3/s3-service";
 import StorageAPI, {
   DeleteOptions,
   ReadOptions,
@@ -18,6 +18,9 @@ import StorageAPI, {
   WriteOptions,
 } from "./provider";
 
+import { OneOfStorageStrategy } from "./oneof-storage-strategy";
+import { DefaultStorageStrategy } from "./default-storage-strategy";
+
 type EncryptionConfiguration = {
   secret: string | null;
   iv: string | null;
@@ -39,22 +42,10 @@ export default class StorageService extends TdriveService<StorageAPI> implements
 
   constructor(protected options?: TdriveServiceOptions<TdriveServiceConfiguration>) {
     super(options);
-    const type = this.getConnectorType();
-    if (type === "S3") {
-      logger.info("Using 'S3' connector for storage.");
-      try {
-        this.homeDir = this.configuration.get<string>("S3.homeDirectory");
-      } catch (e) {
-        this.logger.warn("Home directory is not set, using S3.bucket instead");
-      }
-      if (!this.homeDir) {
-        this.homeDir = this.configuration.get<string>("S3.bucket");
-      }
-      if (this.homeDir && this.homeDir.startsWith("/")) {
-        this.logger.error("For S3 connector home directory MUST NOT start with '/'");
-        throw new Error("For S3 connector home directory MUST NOT start with '/'");
-      }
-    }
+    //init home directory variable
+    this.initHomeDirectory();
+    //init connector to storage
+    this.getConnector();
   }
 
   api(): StorageAPI {
@@ -67,31 +58,7 @@ export default class StorageService extends TdriveService<StorageAPI> implements
 
   getConnector(): StorageConnectorAPI {
     if (!this.connector) {
-      const type = this.getConnectorType();
-      if (type === "S3") {
-        logger.info("Using 'S3' connector for storage.");
-        this.connector = new S3ConnectorService({
-          bucket: this.configuration.get<string>("S3.bucket"),
-          region: this.configuration.get<string>("S3.region"),
-          endPoint: this.configuration.get<string>("S3.endPoint"),
-          port: Number(this.configuration.get<number>("S3.port")),
-          useSSL: Boolean(this.configuration.get<boolean>("S3.useSSL")),
-          accessKey: this.configuration.get<string>("S3.accessKey"),
-          secretKey: this.configuration.get<string>("S3.secretKey"),
-          disableRemove: this.configuration.get<boolean>("S3.disableRemove"),
-        });
-      } else {
-        logger.info(
-          `Using 'local' connector for storage${
-            type === "local" ? "" : " (no other connector recognized from configuration type: '%s')"
-          }.`,
-          type,
-        );
-        logger.trace(`Home directory for the storage: ${this.homeDir}`);
-        this.connector = new LocalConnectorService(
-          this.configuration.get<LocalConfiguration>("local"),
-        );
-      }
+      this.connector = this.getStorageStrategy();
     }
     return this.connector;
   }
@@ -216,4 +183,106 @@ export default class StorageService extends TdriveService<StorageAPI> implements
 
     return this;
   }
+
+  /**
+   * Instantiates storage strategy with configuration.
+   * "storage": {
+   *     "secret": "",
+   *     "iv": "",
+   *     "strategy": "oneof",
+   *     "type": "local",
+   *     "S3": {
+   *       "endPoint": "play.min.io",
+   *       "port": 9000,
+   *       "useSSL": false,
+   *       "accessKey": "ABCD",
+   *       "secretKey": "x1yz",
+   *       "disableRemove": false
+   *     },
+   *     "local": {
+   *       "path": "/tdrive"
+   *     },
+   *     "oneof": [
+   *       {
+   *         "type": "S3",
+   *         "S3": {}
+   *       },
+   *       {
+   *         "type": "local",
+   *         "local": {}
+   *     }]
+   *   },
+   */
+  getStorageStrategy(): StorageConnectorAPI {
+    const strategy = this.configuration.get<string>("strategy");
+    if (strategy == "oneof") {
+      logger.info("Creating storage with 'oneof' strategy");
+      const connectors = this.configuration
+        .get<Array<StorageConfiguration>>("oneof")
+        .map(c => this.createConnector(c));
+      return new OneOfStorageStrategy(connectors);
+    } else {
+      logger.info("Creating storage with 'default' strategy");
+      return new DefaultStorageStrategy(this.createConnectorFromConfiguration(this.configuration));
+    }
+  }
+
+  createConnector(config: StorageConfiguration) {
+    const type = config.type;
+    if (type === "S3") {
+      return this.newS3Connector(config.S3);
+    } else {
+      return this.newLocalConnector(config.local, type);
+    }
+  }
+
+  createConnectorFromConfiguration(config: TdriveServiceConfiguration) {
+    const type = config.get<string>("type");
+    if (type === "S3") {
+      return this.newS3Connector(config.get("S3"));
+    } else {
+      return this.newLocalConnector(this.configuration.get<LocalConfiguration>("local"), type);
+    }
+  }
+
+  newS3Connector(config: S3Configuration) {
+    logger.info("Using 'S3' connector for storage.");
+    return new S3ConnectorService(config);
+  }
+
+  newLocalConnector(config: LocalConfiguration, type: string) {
+    logger.info(
+      `Using 'local' connector for storage${
+        type === "local" ? "" : " (no other connector recognized from configuration type: '%s')"
+      }.`,
+      type,
+    );
+    logger.trace(`Home directory for the storage: ${this.homeDir}`);
+    return new LocalConnectorService(config);
+  }
+
+  initHomeDirectory() {
+    const type = this.getConnectorType();
+    if (type === "S3") {
+      logger.info("Using 'S3' connector for storage.");
+      try {
+        this.homeDir = this.configuration.get<string>("S3.homeDirectory");
+      } catch (e) {
+        this.logger.warn("Home directory is not set, using S3.bucket instead");
+      }
+      if (!this.homeDir) {
+        this.homeDir = this.configuration.get<string>("S3.bucket");
+      }
+      if (this.homeDir && this.homeDir.startsWith("/")) {
+        this.logger.error("For S3 connector home directory MUST NOT start with '/'");
+        throw new Error("For S3 connector home directory MUST NOT start with '/'");
+      }
+    }
+  }
 }
+
+type StorageConfiguration = {
+  type: string;
+  S3?: S3Configuration;
+  local?: LocalConfiguration;
+};