uses redis streams?

mshanemc · Jul 27, 2020 · 1134a01 · 1134a01
1 parent be2be4f
commit 1134a01
Showing 1 changed file with 25 additions and 14 deletions.
diff --git a/src/server/lib/skimmerSupport.ts b/src/server/lib/skimmerSupport.ts
@@ -8,13 +8,14 @@ import {
     orgDeleteExchange,
     getHerokuCDSs,
     getAppNamesFromHerokuCDSs,
-    getKeysForCDSs,
+    // getKeysForCDSs,
     cdsRetrieve,
     cdsDelete,
     getDeleteRequest,
     getAllPooledOrgs
 } from './redisNormal';
 import { PoolConfig } from './types';
+import { CDS } from './CDS';
 import { herokuDelete } from './herokuDelete';
 import { exec2JSON } from './execProm';
 import { getPoolName, getPoolConfig } from './namedUtilities';
@@ -100,19 +101,29 @@ const herokuExpirationCheck = async (): Promise<void> => {
 };
 
 const removeOldDeployIds = async (): Promise<void> => {
-    const deployIds = await getKeysForCDSs();
-    const CDSs = (await Promise.all(deployIds.map((deployId) => cdsRetrieve(deployId)))).filter((cds) => cds.mainUser && cds.mainUser.username);
-    await Promise.all(
-        CDSs.map((cds) => {
-            if (!cds.expirationDate && moment().diff(moment(cds.browserStartTime), 'hours') > hoursToKeepBYOO) {
-                return cdsDelete(cds.deployId);
-            }
-            if (cds.expirationDate && moment().isAfter(moment(cds.expirationDate).endOf('day'))) {
-                return cdsDelete(cds.deployId);
-            }
-            return undefined;
-        }).filter((item) => item)
-    );
+    // const deployIds = await getKeysForCDSs();
+    const stream = redis.scanStream({
+        match: '*-*-*'
+    });
+    stream.on('data', async (keyResults) => {
+        // Pause the stream from scanning more keys until we've migrated the current keys.
+        stream.pause();
+        const CDSs = ((await Promise.all(keyResults.map((deployId) => cdsRetrieve(deployId)))) as CDS[]).filter(
+            (cds) => cds.mainUser && cds.mainUser.username
+        );
+        await Promise.all(
+            CDSs.map((cds) => {
+                if (!cds.expirationDate && moment().diff(moment(cds.browserStartTime), 'hours') > hoursToKeepBYOO) {
+                    return cdsDelete(cds.deployId);
+                }
+                if (cds.expirationDate && moment().isAfter(moment(cds.expirationDate).endOf('day'))) {
+                    return cdsDelete(cds.deployId);
+                }
+                return undefined;
+            }).filter((item) => item)
+        );
+        stream.resume();
+    });
 };
 
 const processDeleteQueue = async (): Promise<void> => {