commits reading cache - save intermediate results into a local file (handy when reading a repo with a multi-year history)

Author: anton-107

.gitignore

@@ -1,3 +1,4 @@
 .idea
 node_modules
 coverage
+.cache

.prettierignore

@@ -1,2 +1,3 @@
 dist/**/*
-coverage/**/*
+coverage/**/*
+.cache/**/*

src/cache/resumable-processor.ts

@@ -0,0 +1,90 @@
+import fs from "fs/promises";
+import path from "path";
+
+interface ProgressCache<R> {
+  completedItems: { [key: number]: R };
+  lastProcessedIndex: number;
+}
+
+export class ResumableProcessor {
+  private cachePath: string;
+  private snapshotInterval: number;
+
+  constructor(cachePath = ".cache", snapshotInterval = 100) {
+    this.cachePath = cachePath;
+    this.snapshotInterval = snapshotInterval;
+  }
+
+  private async ensureCacheDirectory(): Promise<void> {
+    await fs.mkdir(this.cachePath, { recursive: true });
+  }
+
+  private getCachePath(jobId: string): string {
+    const sanitizedJobId = jobId.replace(/\//g, "-");
+    return path.join(this.cachePath, `${sanitizedJobId}.json`);
+  }
+
+  private async loadCache<R>(jobId: string): Promise<ProgressCache<R>> {
+    try {
+      const data = await fs.readFile(this.getCachePath(jobId), "utf-8");
+      return JSON.parse(data);
+    } catch {
+      return { completedItems: {}, lastProcessedIndex: -1 };
+    }
+  }
+
+  private async saveCache<R>(
+    jobId: string,
+    cache: ProgressCache<R>
+  ): Promise<void> {
+    await fs.writeFile(
+      this.getCachePath(jobId),
+      JSON.stringify(cache, null, 2)
+    );
+  }
+
+  // eslint-disable-next-line max-params
+  async process<T, R>(
+    jobId: string,
+    items: T[],
+    processFn: (item: T) => Promise<R>,
+    notifyFn: (result: R) => void
+  ): Promise<R[]> {
+    await this.ensureCacheDirectory();
+    const cache = await this.loadCache<R>(jobId);
+    const results: R[] = new Array(items.length);
+
+    // Restore completed items from cache
+    for (const [index, value] of Object.entries(cache.completedItems)) {
+      const idx = parseInt(index);
+      if (idx < items.length) {
+        results[idx] = value;
+        notifyFn(value);
+      }
+    }
+
+    // Process remaining items
+    for (let i = cache.lastProcessedIndex + 1; i < items.length; i++) {
+      results[i] = await processFn(items[i]);
+      cache.completedItems[i] = results[i];
+      cache.lastProcessedIndex = i;
+
+      // Save snapshot at intervals
+      if ((i + 1) % this.snapshotInterval === 0) {
+        await this.saveCache(jobId, cache);
+      }
+    }
+
+    // Final save
+    await this.saveCache(jobId, cache);
+    return results;
+  }
+
+  async clearCache(jobId: string): Promise<void> {
+    try {
+      await fs.unlink(this.getCachePath(jobId));
+    } catch {
+      // Ignore if file doesn't exist
+    }
+  }
+}

src/git-reader/git-repository.ts

@@ -2,6 +2,7 @@ import fs from "fs";
 import git from "isomorphic-git";
 import { Readable } from "stream";
 
+import { ResumableProcessor } from "../cache/resumable-processor.js";
 import { time, timeLog } from "../index.js";
 import { ChangedFile, Commit, ExpandedCommit } from "../interfaces.js";
 
@@ -10,6 +11,8 @@ interface GitReadOptions {
 }
 
 export class GitRepository {
+  private previousCommit: Commit = null;
+
   constructor(private repoPath: string) {}
 
   public async getListOfCommits(): Promise<Commit[]> {
@@ -18,39 +21,58 @@ export class GitRepository {
   public async getListOfCommitsWithChangedFiles(
     options: GitReadOptions = {}
   ): Promise<ExpandedCommit[]> {
-    const results: ExpandedCommit[] = [];
     const commits = await this.getListOfCommits();
-    let previousCommit;
     time("getListOfCommitsWithChangedFiles");
-    for (const c of commits) {
-      if (!previousCommit) {
-        previousCommit = c;
-        continue;
-      }
-      time(`getFilesDiff-${previousCommit.oid}-${c.oid}`);
-      const changedFiles: ChangedFile[] = await this.getFilesDiff(
-        previousCommit.oid,
-        c.oid
-      );
-      timeLog(`getFilesDiff-${previousCommit.oid}-${c.oid}`);
-
-      const result = {
-        oid: c.oid,
-        commit: previousCommit,
-        changedFiles,
-      };
-
-      results.push(result);
-
-      if (options.stream) {
-        options.stream.push(result);
-      }
 
-      previousCommit = c;
-    }
+    const taskProcessor = new ResumableProcessor(".cache", 100);
+    const results: ExpandedCommit[] = await taskProcessor.process(
+      this.repoPath,
+      commits,
+      this.getFilesDiffWithCursor.bind(this, options),
+      this.updateGitReadStream.bind(this, options)
+    );
+
     timeLog("getListOfCommitsWithChangedFiles");
     return results;
   }
+  private async getFilesDiffWithCursor(
+    options: GitReadOptions = {},
+    cursor: Commit
+  ): Promise<ExpandedCommit> {
+    if (!this.previousCommit) {
+      this.previousCommit = cursor;
+      return null;
+    }
+    const taskID = `getFilesDiff-${this.previousCommit.oid}-${cursor.oid}`;
+    time(taskID);
+    const changedFiles: ChangedFile[] = await this.getFilesDiff(
+      this.previousCommit.oid,
+      cursor.oid
+    );
+    timeLog(taskID);
+
+    const result = {
+      oid: cursor.oid,
+      commit: this.previousCommit,
+      changedFiles,
+    };
+
+    this.updateGitReadStream(options, result);
+
+    this.previousCommit = cursor;
+
+    return result;
+  }
+  private updateGitReadStream(
+    options: GitReadOptions = {},
+    result: ExpandedCommit
+  ) {
+    if (options.stream) {
+      if (result !== null) {
+        options.stream.push(result);
+      }
+    }
+  }
   private async getFilesDiff(
     prevOID: string,
     nextOID: string

src/index.ts

@@ -107,7 +107,7 @@ async function main() {
 
   let commitsCounter = 0;
   commitsStream.on("data", (commit: ExpandedCommit) => {
-    debug("Commit", commit);
+    log("Commit", commit);
 
     commitsCounter += 1;
     intermediateAggregateMonthly.addCommit(commit);
@@ -116,9 +116,15 @@ async function main() {
     quarterlyDashboard.updateData(intermediateAggregateQuarterly.getData());
     summaryDashboard.setCurrentProgress(commitsCounter, commit);
   });
+  commitsStream.on("error", (err) => {
+    log("error reading commits", { err });
+  });
   commitsStream.on("end", () => {
     log("done reading commits", {});
   });
+  commitsStream.on("close", () => {
+    log("stream closed", {});
+  });
 
   // number of commits by author:
   const commits = await repo.getListOfCommits();