feat: parallelize saving to storages (#357)

* parallelize saving to storages

* Updated snapshots

* [autofix.ci] apply automated fixes

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

src/__snapshots__/messages.test.ts.snap

@@ -5,6 +5,7 @@ exports[`messages getSummaryMessages should not add empty groups 1`] = `
   "📝 Storage (TheTable)
 	0 added
 	0 skipped (0 existing, 0 pending)
+	took 0.00s
 	-----
 	Group2:
 		description1:	+10.00",
@@ -15,7 +16,8 @@ exports[`messages getSummaryMessages should not add empty groups 2`] = `
 [
   "📝 Storage (TheTable)
 	0 added
-	0 skipped (0 existing, 0 pending)",
+	0 skipped (0 existing, 0 pending)
+	took 0.00s",
 ]
 `;
 
@@ -38,10 +40,12 @@ exports[`messages getSummaryMessages should return a summary message 2`] = `
 [
   "📝 Storage 1 (TheTable)
 	1 added
-	0 skipped (0 existing, 0 pending)",
+	0 skipped (0 existing, 0 pending)
+	took 350.00s",
   "📝 Storage 2 (TheTable)
 	7 added
 	0 skipped (0 existing, 3 pending)
+	took 350.00s
 	-----
 	Group1:
 		description1:	+10.00",
@@ -80,6 +84,7 @@ exports[`messages getSummaryMessages should return a summary message with instal
   "📝 Storage 1 (TheTable)
 	2 added
 	0 skipped (0 existing, 0 pending)
+	took 0.00s
 	-----
 	SomeGroup:
 		should be +20:	+20.00

src/messages.test.ts

@@ -103,7 +103,7 @@ describe("messages", () => {
       const summary = getSummaryMessages(results);
       expect(summary).toMatchSnapshot();
 
-      const saveSummaries = stats.map(statsString);
+      const saveSummaries = stats.map((stats) => statsString(stats, 350000));
       expect(saveSummaries).toMatchSnapshot();
     });
 
@@ -115,7 +115,7 @@ describe("messages", () => {
       const summary = getSummaryMessages(results);
       expect(summary).toMatchSnapshot();
 
-      const saveSummaries = stats.map(statsString);
+      const saveSummaries = stats.map((stats) => statsString(stats, 0));
       expect(saveSummaries).toMatchSnapshot();
     });
 
@@ -155,7 +155,7 @@ describe("messages", () => {
       const summary = getSummaryMessages(results);
       expect(summary).toMatchSnapshot();
 
-      const saveSummaries = stats.map(statsString);
+      const saveSummaries = stats.map((stats) => statsString(stats, 0));
       expect(saveSummaries).toMatchSnapshot();
     });
 
@@ -208,7 +208,7 @@ describe("messages", () => {
       const summary = getSummaryMessages(results);
       expect(summary).toMatchSnapshot();
 
-      const saveSummaries = stats.map(statsString);
+      const saveSummaries = stats.map((stats) => statsString(stats, 0));
       expect(saveSummaries).toMatchSnapshot();
     });
 
@@ -237,7 +237,7 @@ describe("messages", () => {
         }),
       ];
 
-      const saveSummaries = stats.map(statsString);
+      const saveSummaries = stats.map((stats) => statsString(stats, 0));
       expect(saveSummaries).toMatchSnapshot();
     });
 
@@ -250,7 +250,7 @@ describe("messages", () => {
         }),
       ];
 
-      const saveSummaries = stats.map(statsString);
+      const saveSummaries = stats.map((stats) => statsString(stats, 0));
       expect(saveSummaries).toMatchSnapshot();
     });
   });

src/saveStats.ts

@@ -67,13 +67,14 @@ export function createSaveStats<TInit extends Partial<SaveStats>>(
   };
 }
 
-export function statsString(stats: SaveStats): string {
+export function statsString(stats: SaveStats, saveDurationMs: number): string {
   return `
 📝 ${stats.name}${stats.table ? ` (${stats.table})` : ""}
 \t${stats.added} added
 \t${stats.skipped} skipped (${stats.existing} existing, ${
     stats.pending
   } pending)
+\ttook ${(saveDurationMs / 1000).toFixed(2)}s
 ${highlightedTransactionsString(stats.highlightedTransactions, 1)}`.trim();
 }
 

src/storage/index.ts

@@ -10,8 +10,9 @@ import { WebPostStorage } from "./web-post.js";
 import { saving } from "../messages.js";
 import { createLogger } from "../utils/logger.js";
 import { statsString } from "../saveStats.js";
+import { parallel } from "async";
 
-const logger = createLogger("storage");
+const baseLogger = createLogger("storage");
 
 export const storages = [
   new LocalJsonStorage(),
@@ -34,26 +35,32 @@ export async function saveResults(results: Array<AccountScrapeResult>) {
     return;
   }
 
-  for (let storage of storages) {
-    const { name } = storage.constructor;
-    try {
-      logger(`Initializing ${name}`);
-      await storage.init();
-    } catch (e) {
-      logger(`Error initializing ${name}`, e);
-      sendError(e, `init::${name}`);
-    }
+  await parallel(
+    storages.map((storage) => async () => {
+      const { name } = storage.constructor;
+      const logger = baseLogger.extend(name);
+      try {
+        logger(`initializing`);
+        await storage.init();
+      } catch (e) {
+        logger(`error initializing`, e);
+        sendError(e, `init::${name}`);
+      }
 
-    try {
-      logger(`Saving ${txns.length} transactions to ${name}`);
-      const message = await send(saving(name));
-      const stats = await storage.saveTransactions(txns);
-      await editMessage(message?.message_id, statsString(stats));
-    } catch (e) {
-      logger(`Error saving transactions to ${name}`, e);
-      sendError(e, `saveTransactions::${name}`);
-    }
-  }
+      try {
+        logger(`saving ${txns.length} transactions`);
+        const message = await send(saving(name));
+        const start = performance.now();
+        const stats = await storage.saveTransactions(txns);
+        const duration = performance.now() - start;
+        logger(`saved`);
+        await editMessage(message?.message_id, statsString(stats, duration));
+      } catch (e) {
+        logger(`error saving transactions`, e);
+        sendError(e, `saveTransactions::${name}`);
+      }
+    }),
+  );
 }
 
 function resultsToTransactions(