Release 6/13

.env.example

@@ -23,3 +23,13 @@ REDIS_PORT=6379
 
 # Validator API Keys (comma-separated)
 API_KEYS=
+
+# PR state reconciliation (self-heals missed pull_request.closed webhooks).
+# Hourly sweep re-checks every still-open PR within the window against GitHub.
+PR_RECONCILE_INTERVAL_MS=3600000
+PR_RECONCILE_WINDOW_DAYS=45
+
+# Nightly full repo backfill (coarse safety net; heavier — set false to disable).
+NIGHTLY_BACKFILL_ENABLED=true
+NIGHTLY_BACKFILL_INTERVAL_MS=86400000
+NIGHTLY_BACKFILL_DAYS=40

packages/das/src/api/admin.controller.ts

@@ -14,6 +14,7 @@ import { ApiTags, ApiOperation, ApiSecurity, ApiBody } from "@nestjs/swagger";
 import { RequireApiKeyGuard } from "./require-api-key.guard";
 import { Repo } from "../entities";
 import { FETCH_QUEUE, FETCH_JOBS } from "../queue/constants";
+import { validateRepoFullName } from "../utils/repo-full-name";
 
 interface BackfillBody {
   repoFullName: string;
@@ -24,18 +25,6 @@ interface RegisterBody {
   repoFullName: string;
 }
 
-// GitHub owner/repo pattern: alphanum + `.`, `_`, `-`, length reasonable.
-const REPO_FULL_NAME_PATTERN = /^[\w.-]{1,100}\/[\w.-]{1,100}$/;
-
-function validateRepoFullName(value: unknown): string {
-  if (typeof value !== "string" || !REPO_FULL_NAME_PATTERN.test(value)) {
-    throw new BadRequestException(
-      'repoFullName must match "owner/repo" (alphanumerics, dot, dash, underscore)',
-    );
-  }
-  return value;
-}
-
 function validateDays(value: unknown): number | undefined {
   if (value === undefined || value === null) return undefined;
   if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {

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

@@ -2,6 +2,7 @@ import { Processor, WorkerHost, InjectQueue } from "@nestjs/bullmq";
 import { Logger } from "@nestjs/common";
 import { InjectRepository } from "@nestjs/typeorm";
 import { IsNull, Repository } from "typeorm";
+import { QueryDeepPartialEntity } from "typeorm/query-builder/QueryPartialEntity";
 import { Job, Queue } from "bullmq";
 import { Issue, PullRequest } from "../entities";
 import { GitHubFetcherService } from "../webhook/github-fetcher.service";
@@ -123,19 +124,47 @@ export class FetchProcessor extends WorkerHost {
   ): Promise<void> {
     this.logger.log(`Fetching PR metadata for ${repoFullName}#${prNumber}`);
 
-    const { closingIssueNumbers, body, lastEditedAt } =
-      await this.fetcher.fetchPrMetadata(repoFullName, prNumber);
+    const {
+      closingIssueNumbers,
+      body,
+      lastEditedAt,
+      state,
+      mergedAt,
+      closedAt,
+      mergedByLogin,
+    } = await this.fetcher.fetchPrMetadata(repoFullName, prNumber);
     const currentClosingIssueNumbers =
       this.uniqueIssueNumbers(closingIssueNumbers);
 
-    await this.prRepo.update(
-      { repoFullName, prNumber },
-      {
-        closingIssueNumbers: currentClosingIssueNumbers,
-        body,
-        lastEditedAt,
-      },
-    );
+    // Re-assert authoritative state from GraphQL so a missed
+    // `pull_request.closed` webhook self-heals (see PrReconcileService, which
+    // enqueues this job for every still-open PR on a schedule). MERGED is
+    // terminal: never let an in-flight stale fetch revert a merged PR back to
+    // OPEN/CLOSED — only forward transitions are applied.
+    const existing = await this.prRepo.findOne({
+      where: { repoFullName, prNumber },
+      select: { state: true },
+    });
+    const applyState = !(existing?.state === "MERGED" && state !== "MERGED");
+
+    if (applyState && existing && existing.state !== state) {
+      this.logger.warn(
+        `State drift corrected for ${repoFullName}#${prNumber}: ` +
+          `${existing.state} → ${state} (missed webhook)`,
+      );
+    }
+
+    // Cast past the entity's non-null column types: merged_at/closed_at/
+    // merged_by_login are nullable in the DB, and writing null is correct
+    // (clears them on a reopened PR).
+    const update = {
+      closingIssueNumbers: currentClosingIssueNumbers,
+      body,
+      lastEditedAt,
+      ...(applyState ? { state, mergedAt, closedAt, mergedByLogin } : {}),
+    } as QueryDeepPartialEntity<PullRequest>;
+
+    await this.prRepo.update({ repoFullName, prNumber }, update);
 
     // Issue solver attribution is closure-driven (ISSUE_CLOSURE jobs read
     // ClosedEvent.closer). PR metadata only refreshes the PR-side text view

packages/das/src/utils/repo-full-name.ts

@@ -0,0 +1,13 @@
+import { BadRequestException } from "@nestjs/common";
+
+/** GitHub owner/repo pattern: alphanum + `.`, `_`, `-`, length reasonable. */
+export const REPO_FULL_NAME_PATTERN = /^[\w.-]{1,100}\/[\w.-]{1,100}$/;
+
+export function validateRepoFullName(value: unknown): string {
+  if (typeof value !== "string" || !REPO_FULL_NAME_PATTERN.test(value)) {
+    throw new BadRequestException(
+      'repoFullName must match "owner/repo" (alphanumerics, dot, dash, underscore)',
+    );
+  }
+  return value;
+}

packages/das/src/webhook/github-fetcher.service.ts

@@ -287,16 +287,30 @@ export class GitHubFetcherService implements OnModuleInit {
     closingIssueNumbers: number[];
     body: string | null;
     lastEditedAt: string | null;
+    state: string;
+    mergedAt: string | null;
+    closedAt: string | null;
+    mergedByLogin: string | null;
   }> {
     const [owner, repo] = repoFullName.split("/");
     const token = await this.getTokenForRepo(repoFullName);
 
+    // `state`/`mergedAt`/`closedAt`/`mergedBy` are returned alongside the body
+    // so the metadata-fetch path can re-assert authoritative PR state — this is
+    // what lets a missed `pull_request.closed` webhook self-heal (the webhook
+    // handler is otherwise the only writer of state). GraphQL `state` is the
+    // source of truth (OPEN / CLOSED / MERGED), unlike REST which reports a
+    // merged PR as `closed` + `merged: true`.
     const query = `
       query($owner: String!, $repo: String!, $pr: Int!) {
         repository(owner: $owner, name: $repo) {
           pullRequest(number: $pr) {
             bodyText
             lastEditedAt
+            state
+            mergedAt
+            closedAt
+            mergedBy { login }
             closingIssuesReferences(first: 10) {
               nodes {
                 number
@@ -343,6 +357,10 @@ export class GitHubFetcherService implements OnModuleInit {
       ),
       body: pr.bodyText ?? null,
       lastEditedAt: pr.lastEditedAt ?? null,
+      state: pr.state,
+      mergedAt: pr.mergedAt ?? null,
+      closedAt: pr.closedAt ?? null,
+      mergedByLogin: pr.mergedBy?.login ?? null,
     };
   }
 

packages/das/src/webhook/handlers/installation.handler.ts

@@ -38,30 +38,45 @@ export class InstallationHandler {
       payload.repositories ?? payload.repositories_added ?? [];
 
     for (const repo of repos) {
+      const repoFullName = String(repo.full_name);
       // Atomic upsert: insert with addedAt on first encounter; on conflict only
       // update installationId so addedAt is never overwritten on re-fires.
       await this.repoRepo
         .createQueryBuilder()
         .insert()
         .into(Repo)
         .values({
-          repoFullName: repo.full_name,
+          repoFullName,
           installationId: String(installationId),
           addedAt: new Date().toISOString(),
         })
         .orUpdate(["installation_id"], ["repo_full_name"])
         .execute();
-      this.logger.log(`Tracking repo: ${repo.full_name}`);
+      this.logger.log(`Tracking repo: ${repoFullName}`);
     }
 
     // installation_repositories.removed — soft clear, preserve historical data.
+    // Match case-insensitively so removal events clear rows even if earlier
+    // registration paths preserved a different Owner/Repo casing (#120).
     const removed: any[] = payload.repositories_removed ?? [];
     for (const repo of removed) {
-      await this.repoRepo.update(repo.full_name, {
-        installationId: null,
-        registered: false,
-      });
-      this.logger.log(`Stopped tracking repo: ${repo.full_name}`);
+      const repoFullName = String(repo.full_name);
+      const result = await this.repoRepo
+        .createQueryBuilder()
+        .update()
+        .set({ installationId: null, registered: false })
+        .where("LOWER(repo_full_name) = LOWER(:repoFullName)", {
+          repoFullName,
+        })
+        .execute();
+
+      if (!result.affected) {
+        this.logger.warn(
+          `No repo row matched removal for ${repo.full_name} (canonical: ${repoFullName})`,
+        );
+      } else {
+        this.logger.log(`Stopped tracking repo: ${repoFullName}`);
+      }
     }
   }
 }

packages/das/src/webhook/handlers/pull-request.handler.ts

@@ -26,17 +26,24 @@ export class PullRequestHandler {
     const prNumber: number = pr.number;
     const action: string = payload.action;
 
+    // A merged PR is reported by REST as state=closed + merged=true. Key off
+    // `merged` alone — requiring a non-null `merged_at` too risks pinning the
+    // PR to OPEN if GitHub sends the closed event before merged_at is populated.
+    // Synthesize merged_at from closed_at when absent so the merge gate (which
+    // requires merged_at) downstream still passes.
+    const isMerged = Boolean(pr.merged);
+
     const data: Partial<PullRequest> = {
       repoFullName,
       prNumber,
       authorGithubId: String(pr.user.id),
       authorLogin: pr.user.login,
       authorAssociation: pr.author_association,
       title: pr.title,
-      state: pr.merged && pr.merged_at ? "MERGED" : pr.state.toUpperCase(),
+      state: isMerged ? "MERGED" : pr.state.toUpperCase(),
       createdAt: pr.created_at,
       closedAt: pr.closed_at ?? null,
-      mergedAt: pr.merged_at ?? null,
+      mergedAt: isMerged ? (pr.merged_at ?? pr.closed_at ?? null) : null,
       // last_edited_at is populated by the fetch-pr-metadata job via GraphQL —
       // REST's updated_at changes on any interaction, not just body edits.
       mergedByLogin: pr.merged_by?.login ?? null,

packages/das/src/webhook/pr-reconcile.service.ts

@@ -0,0 +1,90 @@
+import {
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+  OnModuleInit,
+} from "@nestjs/common";
+import { InjectQueue } from "@nestjs/bullmq";
+import { InjectRepository } from "@nestjs/typeorm";
+import { Queue } from "bullmq";
+import { Repository } from "typeorm";
+import { PullRequest } from "../entities";
+import { FETCH_QUEUE, FETCH_JOBS } from "../queue/constants";
+
+// PR state (OPEN → MERGED/CLOSED) is written only by the pull_request webhook
+// handler. A single missed/dropped `pull_request.closed` delivery therefore
+// leaves a merged PR stuck OPEN forever, with no path to recover. This sweep
+// closes that gap: on a schedule it re-enqueues a metadata fetch for every
+// still-open PR in registered repos, and the metadata handler re-asserts
+// authoritative GraphQL state — so missed merge events self-heal.
+const RECONCILE_INTERVAL_MS = Number(
+  process.env.PR_RECONCILE_INTERVAL_MS ?? 60 * 60 * 1000, // hourly
+);
+// Bound the sweep to the validator's scoring window — older PRs are no longer
+// scored, so refreshing them buys nothing and only spends GitHub API budget.
+const RECONCILE_WINDOW_DAYS = Number(
+  process.env.PR_RECONCILE_WINDOW_DAYS ?? 45,
+);
+
+@Injectable()
+export class PrReconcileService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(PrReconcileService.name);
+  private timer: NodeJS.Timeout | null = null;
+
+  constructor(
+    @InjectRepository(PullRequest)
+    private readonly prRepo: Repository<PullRequest>,
+    @InjectQueue(FETCH_QUEUE)
+    private readonly fetchQueue: Queue,
+  ) {}
+
+  onModuleInit(): void {
+    // Run once at startup, then on the interval.
+    void this.reconcile();
+    this.timer = setInterval(
+      () => void this.reconcile(),
+      RECONCILE_INTERVAL_MS,
+    );
+  }
+
+  onModuleDestroy(): void {
+    if (this.timer) clearInterval(this.timer);
+  }
+
+  private async reconcile(): Promise<void> {
+    try {
+      const rows: { repo_full_name: string; pr_number: number }[] =
+        await this.prRepo.query(
+          `SELECT p.repo_full_name, p.pr_number
+           FROM pull_requests p
+           JOIN repos r ON r.repo_full_name = p.repo_full_name
+           WHERE p.state = 'OPEN'
+             AND r.registered = true
+             AND p.created_at > NOW() - INTERVAL '${RECONCILE_WINDOW_DAYS} days'`,
+        );
+
+      this.logger.log(
+        `Reconciling ${rows.length} open PRs against GitHub ` +
+          `(window ${RECONCILE_WINDOW_DAYS}d)`,
+      );
+
+      for (const row of rows) {
+        await this.fetchQueue.add(
+          FETCH_JOBS.PR_METADATA,
+          { repoFullName: row.repo_full_name, prNumber: row.pr_number },
+          {
+            // Same stable per-PR jobId as the webhook path — a reconcile job
+            // dedupes against an already-pending webhook-triggered fetch.
+            jobId: `meta-${row.repo_full_name}-${row.pr_number}`,
+            removeOnComplete: true,
+            removeOnFail: true,
+            attempts: 3,
+            backoff: { type: "exponential", delay: 5000 },
+          },
+        );
+      }
+    } catch (err) {
+      this.logger.error(`Reconcile failed: ${String(err)}`);
+    }
+  }
+}