clean up corrupted versions after 1 minute

package.json

@@ -87,6 +87,7 @@
     "dotenv": "^16.4.5",
     "errorhandler": "^1.5.1",
     "express": "^4.19.2",
+    "fastq": "^1.17.1",
     "form-data": "^4.0.0",
     "fs-extra": "^11.2.0",
     "getport": "^0.1.0",
@@ -122,4 +123,4 @@
     "universalify": "^2.0.0",
     "yargs": "^17.7.2"
   }
-}
+}

src/registry/domain/components-cache/components-list.ts

@@ -1,13 +1,51 @@
+import * as fastq from 'fastq';
+import type { queueAsPromised } from 'fastq';
 import getUnixUTCTimestamp from 'oc-get-unix-utc-timestamp';
 import type { StorageAdapter } from 'oc-storage-adapters-utils';
 import semver from 'semver';
 import type { ComponentsList, Config } from '../../../types';
 import pLimit from '../../../utils/pLimit';
 import eventsHandler from '../events-handler';
 
+const delay = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms));
+const validateComponentVersion =
+  (conf: Config, cdn: StorageAdapter) =>
+  (componentName: string, componentVersion: string) => {
+    return cdn
+      .getJson(
+        // Check integrity of the package by checking existence of package.json
+        // OC will upload always the package.json last when publishing
+        `${conf.storage.options.componentsDir}/${componentName}/${componentVersion}/package.json`
+      )
+      .then(() => true)
+      .catch(() => false);
+  };
+
+interface QueueTask {
+  conf: Config;
+  cdn: StorageAdapter;
+  name: string;
+  version: string;
+}
+const cleanupQueue: queueAsPromised<QueueTask> = fastq.promise(
+  cleanupWorker,
+  1
+);
+
+async function cleanupWorker(arg: QueueTask): Promise<void> {
+  const validator = validateComponentVersion(arg.conf, arg.cdn);
+  const isValid = await validator(arg.name, arg.version);
+  if (!isValid) {
+    await arg.cdn.removeDir(
+      `${arg.conf.storage.options.componentsDir}/${arg.name}/${arg.version}`
+    );
+  }
+}
+
 export default function componentsList(conf: Config, cdn: StorageAdapter) {
   const filePath = (): string =>
     `${conf.storage.options.componentsDir}/components.json`;
+  const validator = validateComponentVersion(conf, cdn);
 
   const componentsList = {
     getFromJson: (): Promise<ComponentsList> => cdn.getJson(filePath(), true),
@@ -17,20 +55,6 @@ export default function componentsList(conf: Config, cdn: StorageAdapter) {
     ): Promise<ComponentsList> => {
       const componentsInfo: Record<string, string[]> = {};
 
-      const validateComponentVersion = (
-        componentName: string,
-        componentVersion: string
-      ) => {
-        return cdn
-          .getJson(
-            // Check integrity of the package by checking existence of package.json
-            // OC will upload always the package.json last when publishing
-            `${conf.storage.options.componentsDir}/${componentName}/${componentVersion}/package.json`
-          )
-          .then(() => true)
-          .catch(() => false);
-      };
-
       const getVersionsForComponent = async (
         componentName: string
       ): Promise<string[]> => {
@@ -45,7 +69,7 @@ export default function componentsList(conf: Config, cdn: StorageAdapter) {
           await Promise.all(
             unCheckedVersions.map((unCheckedVersion) =>
               limit(async () => {
-                const isValid = await validateComponentVersion(
+                const isValid = await validator(
                   componentName,
                   unCheckedVersion
                 );
@@ -63,6 +87,18 @@ export default function componentsList(conf: Config, cdn: StorageAdapter) {
               ', '
             )}.`
           });
+          delay(60_000).then(() => {
+            for (const invalidVersion of invalidVersions) {
+              cleanupQueue
+                .push({
+                  conf,
+                  cdn,
+                  name: componentName,
+                  version: invalidVersion
+                })
+                .catch(() => {});
+            }
+          });
         }
 
         const validVersions = allVersions.filter(