fix(mirror): reconcile stale solved issue links

packages/das/src/queue/fetch.processor.ts

@@ -1,7 +1,7 @@
 import { Processor, WorkerHost, InjectQueue } from "@nestjs/bullmq";
 import { Logger } from "@nestjs/common";
 import { InjectRepository } from "@nestjs/typeorm";
-import { IsNull, Repository } from "typeorm";
+import { In, IsNull, Repository } from "typeorm";
 import { Job, Queue } from "bullmq";
 import { Issue, PullRequest } from "../entities";
 import { GitHubFetcherService } from "../webhook/github-fetcher.service";
@@ -79,28 +79,33 @@ export class FetchProcessor extends WorkerHost {
   ): Promise<void> {
     this.logger.log(`Fetching PR metadata for ${repoFullName}#${prNumber}`);
 
+    const previousPr = await this.prRepo.findOneBy({ repoFullName, prNumber });
+    const previousClosingIssueNumbers = this.uniqueIssueNumbers(
+      previousPr?.closingIssueNumbers ?? [],
+    );
+
     const { closingIssueNumbers, body, lastEditedAt } =
       await this.fetcher.fetchPrMetadata(repoFullName, prNumber);
+    const currentClosingIssueNumbers =
+      this.uniqueIssueNumbers(closingIssueNumbers);
 
     await this.prRepo.update(
       { repoFullName, prNumber },
       {
-        closingIssueNumbers,
+        closingIssueNumbers: currentClosingIssueNumbers,
         body,
         lastEditedAt,
       },
     );
 
-    // If this PR is merged, mark each linked issue as solved_by_pr
     const pr = await this.prRepo.findOneBy({ repoFullName, prNumber });
-    if (pr?.state === "MERGED" && closingIssueNumbers.length > 0) {
-      for (const issueNumber of closingIssueNumbers) {
-        await this.issueRepo.update(
-          { repoFullName, issueNumber },
-          { solvedByPr: prNumber },
-        );
-      }
-    }
+    await this.reconcileSolvedIssueLinks(
+      repoFullName,
+      prNumber,
+      previousClosingIssueNumbers,
+      currentClosingIssueNumbers,
+      pr?.state === "MERGED",
+    );
   }
 
   private async handlePrFiles(data: PrFilesJobData): Promise<void> {
@@ -222,4 +227,48 @@ export class FetchProcessor extends WorkerHost {
       baseSha: generation.baseSha ?? IsNull(),
     };
   }
+
+  private async reconcileSolvedIssueLinks(
+    repoFullName: string,
+    prNumber: number,
+    previousIssueNumbers: number[],
+    currentIssueNumbers: number[],
+    isMerged: boolean,
+  ): Promise<void> {
+    const currentIssueNumberSet = new Set(currentIssueNumbers);
+    const staleIssueNumbers = previousIssueNumbers.filter(
+      (issueNumber) => !currentIssueNumberSet.has(issueNumber),
+    );
+
+    const clearIssueNumbers = isMerged
+      ? staleIssueNumbers
+      : this.uniqueIssueNumbers([
+          ...previousIssueNumbers,
+          ...currentIssueNumbers,
+        ]);
+
+    if (clearIssueNumbers.length > 0) {
+      await this.issueRepo.update(
+        {
+          repoFullName,
+          issueNumber: In(clearIssueNumbers),
+          solvedByPr: prNumber,
+        },
+        { solvedByPr: null },
+      );
+    }
+
+    if (!isMerged || currentIssueNumbers.length === 0) {
+      return;
+    }
+
+    await this.issueRepo.update(
+      { repoFullName, issueNumber: In(currentIssueNumbers) },
+      { solvedByPr: prNumber },
+    );
+  }
+
+  private uniqueIssueNumbers(issueNumbers: number[]): number[] {
+    return [...new Set(issueNumbers)];
+  }
 }