diff --git a/CHANGELOG.md b/CHANGELOG.md index 203bd71da..aa0ac70c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,20 @@ The dashboard's **What's New** rail reads this file at request time entries. Keep entries terse — one line per item under each version date, grouped by `Added` / `Changed` / `Fixed` / `Removed`. +## [2026-07-15] — v0.18.0 · Reliable GitHub state detection + +### Added + +- **Pull-request review state is aggregated from GitHub.** Forge records the latest decisive review per reviewer together with outstanding reviewer and team requests, without copying PR discussion into issue comments. +- **Legacy GitHub link attachments recover into native relations.** A bounded worker repair promotes old issue/PR links when their native resource already exists, restoring normal state sync without provider calls. + +### Fixed + +- **Webhook delivery is retryable and ordered.** Failed or abandoned deliveries use atomic processing leases, concurrent and out-of-order updates cannot regress newer PR state, and repository mappings must match both the requested repo and installation. +- **Checks and commit statuses invalidate immediately.** Queued/rerun checks, SHA-only check events, and legacy `status` webhooks mark aggregate evidence dirty; provider refresh remains the only source trusted for safe automatic completion. +- **Linked GitHub changes reach live issue views.** Meaningful checks, review, draft, mergeability, head-SHA, title, and lifecycle changes publish issue-scoped activity while duplicate links and mirrored GitHub issue comments remain idempotent. +- **Native GitHub installation cannot borrow a runtime-auth app.** Issue/PR sync now requires the configured instance GitHub App with the documented read permissions and active webhook events. + ## [2026-07-15] — v0.17.0 · Clear configuration ownership ### Added diff --git a/DEVLOG.md b/DEVLOG.md index 2f3dbcad..3a9bf550 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -118,6 +118,7 @@ fresh production build passed; and the full serial Playwright suite passed hard-coded two rail columns despite the existing work-count threshold; the assertion now follows the rendered layout mode and passed both alone and in the full suite. + ## 2026-07-14 — Settings scope and information-architecture review Mapped Forge's personal/global, workspace, Mission Control/Activity, and @@ -12966,3 +12967,83 @@ repository warnings; the focused comment router suite passed 15 tests; the isolated stale-work suite passed 9 tests after a known full-suite concurrency failure; the fresh E2E production build passed; and the existing issue-flow Playwright journey passed. + +--- + +## 2026-07-14 — Native GitHub state hardening + +Audited native link creation, webhook ingestion, provider refresh, stale +reconciliation, checks/status aggregation, completion/recovery policy, and the +GitHub-discussion boundary. Closed repo/installation mismatches, casing-based +resource duplication, permanent failed-delivery dedupe, concurrent and +out-of-order regressions, same-SHA rerun gaps, missing legacy status-webhook +invalidation, comment duplication, and missing issue-scoped realtime events. + +Pull refreshes now aggregate the latest decisive review per reviewer and +outstanding review requests while leaving PR comments and review bodies on +GitHub. The worker promotes recoverable legacy generic GitHub attachments into +native relations without provider calls. Native sync no longer substitutes a +separate runtime-auth GitHub App, and the documented instance-app permissions +now include checks, commit statuses, and the `status` webhook event. + +PR review caught twenty-six identity/bounded-prefix/race edge cases before merge. Lifecycle +rules now compare the provider's lifecycle version rather than volatile check +metadata, so a concurrent check hint cannot suppress a merged/closed action. +Legacy-link recovery selects only attachments with an existing matching native +resource and a live target issue, so permanently unmatched or soft-deleted +links cannot starve or abort recoverable rows. +Canonical repo casing now rekeys the existing resource and preserves or merges +its issue relations instead of creating a new unlinked canonical row, while +case-insensitive lifecycle and review-webhook lookups still reject delayed +provider state without regressing a newer terminal resource. Equal-second +terminal tie-breaks now admit only an explicit `reopened` action, so a genuine +reopen is not discarded while ambiguous open/synchronize deliveries remain +conservative. Individual review events never replace the provider aggregate; +they preserve the latest decisive review state and mark it dirty for provider +refresh. Reviewer-requested/removed events use a separate dirty hint so they +cannot clear that aggregate, and review dirty/hint changes participate in the +issue activity fingerprint so open views refresh immediately. Review-event +ordering compares both provider-aggregate and prior webhook timestamps so a +late changes-requested event cannot fire after a newer approval hint. SHA-only +check/status events now filter by the JSON head SHA in the database instead of +loading every PR for a repository into the worker. Provider review aggregation +prioritizes changes requested, then outstanding reviewer/team requests, then +approvals. Mirrored issue comments no longer advance the native resource's +lifecycle freshness clock, so a newer comment delivery cannot suppress later +issue opened/closed/reopened side effects. +Rerequested/requested check webhooks now ignore the prior run's stale +conclusion and only dirty the aggregate for reconciliation, preventing a rerun +request from firing the checks-failed status rule. GitHub App setup now calls +out the write-level Checks permission required for rerun webhook actions. The +same-second reopen exception is limited to closed/unmerged PRs, so a stale +reopen can never regress an already merged resource. +Partial/failed review reads no longer contribute an ordering watermark, so a +delayed decisive review webhook can recover the only available review signal +and invalidate the aggregate normally. Review ordering now also rejects stale +non-decisive events so a late comment cannot rewind the decisive-event +watermark. Webhook completion/failure updates are conditional on the exact +processing lease, preventing an expired handler from overwriting a reclaimed +attempt's result. A redelivery that encounters a live `RECEIVED` lease now +returns a retryable failure rather than a completed-duplicate acknowledgement; +only terminal delivery rows are acknowledged as durable duplicates. +Mixed-case duplicate canonicalization now ranks rows with a provider +`externalUpdatedAt` above internally touched/unversioned rows before comparing +recency, preserving the authoritative lifecycle snapshot and all moved links. +Review-first PR deliveries now seed identity/review hints without advancing the +PR lifecycle freshness clock, so delayed opened/synchronize webhooks can still +link issue keys and apply configured lifecycle rules. +Legacy attachment recovery compares captured GitHub numbers as normalized text +rather than casting untrusted URL digits to `int4`, so an oversized malformed +link cannot abort a reconciliation sweep or starve valid recovery candidates. +Review-dismissed webhooks bypass submission-time staleness because GitHub +retains the original review timestamp; they dirty the provider aggregate while +preserving any newer webhook watermark instead of rewinding it. + +Verification: the focused GitHub/client/reconciliation/completion suites passed +56 tests; lint passed with existing repository warnings; typecheck passed; and +the CI-style serial Vitest gate passed 1,274 tests with one intentional live +connector skip. The canonical parallel gate also passed 1,274 tests and the +fresh production E2E build completed. The browser run passed 35 of 38 journeys +before Chromium crashed on a memory-saturated host; each of the three reported +journeys passed immediately in a fresh isolated browser process. Release, +deployment, and production smoke results follow after the pull request lands. diff --git a/docs/reference/env.md b/docs/reference/env.md index 99d4f1a3..7a615e94 100644 --- a/docs/reference/env.md +++ b/docs/reference/env.md @@ -88,9 +88,13 @@ Configure the GitHub App with: - Setup URL: `https://forge.example/api/connections/github/setup` - Webhook URL: `https://forge.example/api/ingest/github` - Webhook events: `issues`, `issue_comment`, `pull_request`, - `pull_request_review`, `check_suite`, `check_run` -- Read permissions for issues and pull requests. Repository metadata access is - needed for repository selection/listing. + `pull_request_review`, `check_suite`, `check_run`, `status` +- Read permissions for issues, pull requests, and commit statuses. +- **Read and write** permission for checks. GitHub requires write-level Checks + permission to deliver the `requested`/`rerequested` check-suite and check-run + actions Forge uses to invalidate a cached CI aggregate when a rerun starts. + Forge otherwise reads check data and does not create check runs. +- Repository metadata access for repository selection/listing. ## Storage (MinIO / S3) diff --git a/package.json b/package.json index 0ac921bc..69c61733 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "forge", - "version": "0.17.0", + "version": "0.18.0", "private": true, "description": "Forge — a fast, minimalist, keyboard-driven project management platform with pluggable agents.", "license": "MIT", diff --git a/prisma/migrations/0104_github_webhook_delivery_leases/migration.sql b/prisma/migrations/0104_github_webhook_delivery_leases/migration.sql new file mode 100644 index 00000000..3b5ea673 --- /dev/null +++ b/prisma/migrations/0104_github_webhook_delivery_leases/migration.sql @@ -0,0 +1,12 @@ +-- Retry-safe processing leases for native GitHub webhook deliveries. +-- Existing terminal rows keep their historical attempt count; RECEIVED rows +-- use receivedAt as their initial lease start so a crashed request can be +-- reclaimed after the application-level stale threshold. + +ALTER TABLE "ExternalWebhookEvent" + ADD COLUMN "processingStartedAt" TIMESTAMP(3), + ADD COLUMN "attemptCount" INTEGER NOT NULL DEFAULT 1; + +UPDATE "ExternalWebhookEvent" +SET "processingStartedAt" = "receivedAt" +WHERE "processingStartedAt" IS NULL; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 9ca4077c..14b3e7d5 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -2267,6 +2267,11 @@ model ExternalWebhookEvent { status String error String? @db.Text receivedAt DateTime @default(now()) + /// Start of the current processing lease. Failed or abandoned deliveries + /// can be atomically reclaimed without racing an in-flight request. + processingStartedAt DateTime? + /// Number of processing claims, including the initial delivery. + attemptCount Int @default(1) processedAt DateTime? workspace Workspace? @relation(fields: [workspaceId], references: [id], onDelete: Cascade) diff --git a/src/app/api/connections/github/install/route.ts b/src/app/api/connections/github/install/route.ts index 18408e72..1c1c42ff 100644 --- a/src/app/api/connections/github/install/route.ts +++ b/src/app/api/connections/github/install/route.ts @@ -2,7 +2,6 @@ import "server-only"; import { createHmac, randomBytes } from "node:crypto"; import { NextResponse, type NextRequest } from "next/server"; import { auth } from "@/server/auth"; -import { db } from "@/server/db"; import { publicOrigin } from "@/server/integrations/public-origin"; const COOKIE = "forge_github_app_install"; @@ -31,32 +30,6 @@ function signState(state: string, returnTo: string): string { return `${state}.${encodedReturnTo}.${sig}`; } -/** - * Resolve the GitHub App slug to deep-link the install flow at. Prefer the - * `GITHUB_APP_SLUG` env, but fall back to a GitHub App the user already - * configured (e.g. via Settings → GitHub Apps "Create with GitHub") so a - * single-tenant deploy that never set the env var still works. Scopes to the - * caller's workspaces; prefers the workspace in `returnTo` when present. - */ -async function resolveAppSlug(userId: string, returnTo: string): Promise { - const fromEnv = process.env.GITHUB_APP_SLUG?.trim(); - if (fromEnv) return fromEnv; - - const wsSlug = returnTo.match(/^\/w\/([^/]+)(?:\/|$)/)?.[1]; - const memberOf = { memberships: { some: { userId } } }; - const pick = async (workspaceWhere: Record) => - db.githubApp.findFirst({ - where: { slug: { not: null }, workspace: workspaceWhere }, - orderBy: [{ lastMintedAt: "desc" }, { createdAt: "desc" }], - select: { slug: true }, - }); - - // Prefer the workspace the operator came from, then any workspace they belong to. - const scoped = wsSlug ? await pick({ ...memberOf, slug: wsSlug }) : null; - const app = scoped ?? (await pick(memberOf)); - return app?.slug ?? null; -} - export async function GET(req: NextRequest) { const session = await auth(); if (!session?.user?.id) { @@ -64,11 +37,14 @@ export async function GET(req: NextRequest) { } const returnTo = safeReturnTo(req); - const appSlug = await resolveAppSlug(session.user.id, returnTo); + // Native issue/PR sync uses the instance GitHub App credentials. Workspace + // runtime-auth apps deliberately have different permissions, keys, and an + // inactive webhook, so their slug must never be substituted here. + const appSlug = process.env.GITHUB_APP_SLUG?.trim() || null; if (!appSlug) { return settingsRedirect( req, - "No GitHub App is configured. Create one in Settings → GitHub Apps (Create with GitHub), or set GITHUB_APP_SLUG.", + "Native GitHub sync is not configured. Set GITHUB_APP_SLUG and the matching instance GitHub App credentials.", ); } diff --git a/src/server/services/__tests__/github-webhook-hardening.test.ts b/src/server/services/__tests__/github-webhook-hardening.test.ts new file mode 100644 index 00000000..6287a04e --- /dev/null +++ b/src/server/services/__tests__/github-webhook-hardening.test.ts @@ -0,0 +1,1544 @@ +import { afterAll, afterEach, describe, expect, it } from "vitest"; +import { gitHubReviewDecision } from "@/server/services/github/client"; +import { + claimGitHubWebhookDelivery, + finishGitHubWebhookDelivery, + processGitHubWebhook, +} from "@/server/services/github/webhook"; +import { + gitHubResourceVersionMatches, + recoverGenericGitHubAttachments, + resolveGitHubRepoMapping, + upsertExternalResource, + upsertExternalResourceFromWebhook, +} from "@/server/services/github/resource-sync"; +import { + createIssue, + createWorkspaceFixture, + disconnectPrisma, + getPrisma, + type TestFixture, +} from "@/server/routers/__tests__/helpers"; + +const fixtures: TestFixture[] = []; +const deliveryIds: string[] = []; + +function delivery(label: string): string { + const id = `gh-hardening-${label}-${Date.now()}-${Math.random().toString(36).slice(2)}`; + deliveryIds.push(id); + return id; +} + +afterEach(async () => { + const prisma = getPrisma(); + if (deliveryIds.length) { + await prisma.externalWebhookEvent.deleteMany({ + where: { deliveryId: { in: deliveryIds.splice(0) } }, + }); + } + while (fixtures.length) await fixtures.pop()!.cleanup(); +}); + +afterAll(async () => { + await disconnectPrisma(); +}); + +async function setup(options: { syncComments?: boolean; autoCreateIssues?: boolean } = {}) { + const fixture = await createWorkspaceFixture({ keyPrefix: "GH" }); + fixtures.push(fixture); + const prisma = getPrisma(); + const issue = await createIssue(fixture, { statusCategory: "IN_PROGRESS" }); + const connection = await prisma.connection.create({ + data: { + ownerId: fixture.user.id, + provider: "GITHUB", + label: "GitHub webhook test", + status: "CONNECTED", + config: { installationId: "101" }, + }, + }); + const mapping = await prisma.connectionMapping.create({ + data: { + workspaceId: fixture.workspace.id, + connectionId: connection.id, + kind: "repo", + target: "acme/forge", + config: + options.syncComments || options.autoCreateIssues + ? { + github: { + syncComments: options.syncComments ?? false, + autoCreateIssues: options.autoCreateIssues ?? false, + }, + } + : undefined, + }, + }); + return { fixture, prisma, issue, mapping }; +} + +function pullRequest(overrides: Record = {}) { + return { + id: 4200, + node_id: "PR_node_42", + number: 42, + title: "Ship GitHub hardening", + body: null, + state: "open" as const, + html_url: "https://github.com/acme/forge/pull/42", + url: "https://api.github.com/repos/acme/forge/pulls/42", + draft: false, + merged: false, + mergeable_state: "clean", + head: { ref: "hardening", sha: "head-new", repo: { full_name: "acme/forge" } }, + base: { ref: "main", sha: "base", repo: { full_name: "acme/forge" } }, + created_at: "2026-07-14T09:00:00Z", + updated_at: "2026-07-14T12:00:00Z", + ...overrides, + }; +} + +describe("GitHub webhook hardening", () => { + it("keeps outstanding review requests ahead of approvals", () => { + expect( + gitHubReviewDecision({ + approvedCount: 1, + changesRequestedCount: 0, + requestedCount: 2, + }), + ).toBe("REVIEW_REQUESTED"); + expect( + gitHubReviewDecision({ + approvedCount: 1, + changesRequestedCount: 1, + requestedCount: 2, + }), + ).toBe("CHANGES_REQUESTED"); + }); + + it("rejects a mapping id paired with a different repository", async () => { + const { fixture, prisma, mapping } = await setup(); + await expect( + resolveGitHubRepoMapping({ + db: prisma, + workspaceId: fixture.workspace.id, + mappingId: mapping.id, + repoFullName: "other/repository", + }), + ).rejects.toThrow(/does not match/i); + }); + + it("requires the payload installation to match the mapped connection", async () => { + const { prisma } = await setup(); + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("wrong-installation"), + event: "pull_request", + payload: { + action: "opened", + installation: { id: 202 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + }, + }); + expect(result).toMatchObject({ processed: 0, skipped: "no-mapping" }); + }); + + it("canonicalizes repository casing without orphaning existing issue links", async () => { + const { fixture, prisma, issue, mapping } = await setup(); + const legacy = await prisma.externalResource.create({ + data: { + workspaceId: fixture.workspace.id, + provider: "GITHUB", + connectionMappingId: mapping.id, + resourceType: "PULL_REQUEST", + repoFullName: "Acme/Forge", + number: 42, + url: "https://github.com/Acme/Forge/pull/42", + title: "Legacy casing", + state: "open", + }, + }); + const link = await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: legacy.id, + kind: "IMPLEMENTS", + }, + }); + + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Canonical casing", + state: "open", + }, + }); + + expect(resource).toMatchObject({ id: legacy.id, repoFullName: "acme/forge" }); + await expect( + prisma.externalResourceLink.findUniqueOrThrow({ where: { id: link.id } }), + ).resolves.toMatchObject({ externalResourceId: resource.id, issueId: issue.id }); + await expect( + prisma.externalResource.count({ + where: { + workspaceId: fixture.workspace.id, + provider: "GITHUB", + resourceType: "PULL_REQUEST", + number: 42, + }, + }), + ).resolves.toBe(1); + }); + + it("applies webhook freshness guards across repository casing", async () => { + const { fixture, prisma, issue, mapping } = await setup(); + const merged = await prisma.externalResource.create({ + data: { + workspaceId: fixture.workspace.id, + provider: "GITHUB", + connectionMappingId: mapping.id, + resourceType: "PULL_REQUEST", + repoFullName: "Acme/Forge", + number: 42, + url: "https://github.com/Acme/Forge/pull/42", + title: "Newest merged state", + state: "merged", + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }, + }); + const internallyNewerButUnversioned = await prisma.externalResource.create({ + data: { + workspaceId: fixture.workspace.id, + provider: "GITHUB", + connectionMappingId: mapping.id, + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Internally touched stale state", + state: "open", + }, + }); + const staleLink = await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: internallyNewerButUnversioned.id, + kind: "IMPLEMENTS", + }, + }); + + const result = await upsertExternalResourceFromWebhook(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Delayed open state", + state: "open", + externalUpdatedAt: new Date("2026-07-14T11:00:00Z"), + }, + }); + + expect(result).toMatchObject({ applied: false }); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: merged.id } }), + ).resolves.toMatchObject({ + repoFullName: "acme/forge", + title: "Newest merged state", + state: "merged", + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }); + await expect( + prisma.externalResource.findUnique({ where: { id: internallyNewerButUnversioned.id } }), + ).resolves.toBeNull(); + await expect( + prisma.externalResourceLink.findUniqueOrThrow({ where: { id: staleLink.id } }), + ).resolves.toMatchObject({ externalResourceId: merged.id }); + }); + + it("does not let an older PR webhook regress a newer merged snapshot", async () => { + const { fixture, prisma, issue, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Merged title", + state: "merged", + metadata: { merged: true, head: { sha: "head-new" } }, + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }, + }); + await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: resource.id, + kind: "IMPLEMENTS", + }, + }); + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("out-of-order"), + event: "pull_request", + payload: { + action: "reopened", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ updated_at: "2026-07-14T11:00:00Z" }), + }, + }); + expect(result.processed).toBe(0); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + state: "merged", + title: "Merged title", + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }); + }); + + it("does not regress merged state for same-timestamp events", async () => { + const { fixture, prisma, mapping } = await setup(); + await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Merged title", + state: "merged", + metadata: { merged: true }, + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("equal-timestamp"), + event: "pull_request", + payload: { + action: "synchronize", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ updated_at: "2026-07-14T13:00:00Z" }), + }, + }); + + expect(result.processed).toBe(0); + const staleReopen = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("equal-timestamp-reopen-after-merge"), + event: "pull_request", + payload: { + action: "reopened", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ updated_at: "2026-07-14T13:00:00Z" }), + }, + }); + expect(staleReopen.processed).toBe(0); + await expect( + prisma.externalResource.findFirstOrThrow({ + where: { workspaceId: fixture.workspace.id, number: 42 }, + }), + ).resolves.toMatchObject({ state: "merged", title: "Merged title" }); + }); + + it("accepts an explicit reopen when GitHub timestamps share a second", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Closed title", + state: "closed", + metadata: { merged: false }, + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("same-second-reopen"), + event: "pull_request", + payload: { + action: "reopened", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ updated_at: "2026-07-14T13:00:00Z" }), + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ state: "open", title: "Ship GitHub hardening" }); + }); + + it("serializes concurrent snapshots so the newest provider state wins", async () => { + const { fixture, prisma, mapping } = await setup(); + const base = { + provider: "GITHUB" as const, + resourceType: "PULL_REQUEST" as const, + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + }; + + await Promise.all([ + upsertExternalResourceFromWebhook(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + ...base, + title: "Stale open title", + state: "open", + metadata: { merged: false }, + externalUpdatedAt: new Date("2026-07-14T11:00:00Z"), + }, + }), + upsertExternalResourceFromWebhook(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + ...base, + title: "Newest merged title", + state: "merged", + metadata: { merged: true }, + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }, + }), + ]); + + await expect( + prisma.externalResource.findUniqueOrThrow({ + where: { + workspaceId_provider_repoFullName_resourceType_number: { + workspaceId: fixture.workspace.id, + provider: "GITHUB", + repoFullName: "acme/forge", + resourceType: "PULL_REQUEST", + number: 42, + }, + }, + }), + ).resolves.toMatchObject({ + title: "Newest merged title", + state: "merged", + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }); + }); + + it("keeps lifecycle side effects eligible across metadata-only check races", async () => { + const expected = { + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + title: "Merged PR", + state: "merged", + metadata: { merged: true, head: { sha: "head-new" }, checks: { status: "completed" } }, + }; + expect( + gitHubResourceVersionMatches( + { + ...expected, + metadata: { + merged: true, + head: { sha: "head-new" }, + checks: { status: "dirty", source: "webhook-hint" }, + }, + }, + expected, + ), + ).toBe(true); + expect( + gitHubResourceVersionMatches( + { + ...expected, + metadata: { merged: true, head: { sha: "different-head" } }, + }, + expected, + ), + ).toBe(false); + }); + + it("invalidates trusted checks when a same-SHA rerun starts and emits issue activity", async () => { + const { fixture, prisma, issue, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Checks rerun", + state: "open", + metadata: { + head: { sha: "head-new" }, + checks: { + status: "completed", + conclusion: "success", + source: "api-aggregate", + headSha: "head-new", + }, + }, + }, + }); + await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: resource.id, + kind: "IMPLEMENTS", + }, + }); + const unrelated = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 43, + url: "https://github.com/acme/forge/pull/43", + title: "Other head", + state: "open", + metadata: { + head: { sha: "head-other" }, + checks: { status: "completed", conclusion: "success", source: "api-aggregate" }, + }, + }, + }); + + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("check-rerun"), + event: "check_suite", + payload: { + action: "rerequested", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + check_suite: { + head_sha: "head-new", + status: "queued", + conclusion: null, + pull_requests: [], + }, + }, + }); + + const refreshed = await prisma.externalResource.findUniqueOrThrow({ + where: { id: resource.id }, + }); + expect(refreshed.lastSyncedAt).toBeNull(); + expect(refreshed.metadata).toMatchObject({ + checks: { status: "dirty", source: "webhook-hint", headSha: "head-new" }, + }); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: unrelated.id } }), + ).resolves.toMatchObject({ + metadata: { checks: { status: "completed", source: "api-aggregate" } }, + }); + const events = await prisma.activityEvent.findMany({ + where: { workspaceId: fixture.workspace.id, subjectType: "issue", subjectId: issue.id }, + }); + expect( + events.some( + (event) => (event.payload as { change?: string }).change === "external-resource-state", + ), + ).toBe(true); + + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("commit-status"), + event: "status", + payload: { + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + sha: "head-new", + state: "success", + }, + }); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + metadata: { + checks: { + status: "dirty", + source: "webhook-hint", + event: "status", + lastStatusState: "success", + }, + }, + }); + }); + + it("does not apply a stale failed conclusion from a check rerun request", async () => { + const { fixture, prisma, issue, mapping } = await setup(); + const failedStatus = await prisma.status.findFirstOrThrow({ + where: { workspaceId: fixture.workspace.id, category: "CANCELED" }, + }); + await prisma.connectionMapping.update({ + where: { id: mapping.id }, + data: { + config: { github: { statusRules: { checksFailedStatusId: failedStatus.id } } }, + }, + }); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Rerequest failed check", + state: "open", + metadata: { + head: { sha: "head-new" }, + checks: { status: "completed", conclusion: "failure", source: "api-aggregate" }, + }, + }, + }); + await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: resource.id, + kind: "IMPLEMENTS", + }, + }); + + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("check-run-rerequested-stale-failure"), + event: "check_run", + payload: { + action: "rerequested", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + check_run: { + head_sha: "head-new", + status: "completed", + conclusion: "failure", + pull_requests: [{ number: 42 }], + }, + }, + }); + + const issueAfter = await prisma.issue.findUniqueOrThrow({ + where: { id: issue.id }, + include: { status: true }, + }); + expect(issueAfter.status.category).toBe("IN_PROGRESS"); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + metadata: { + checks: { + status: "dirty", + event: "check_run", + observedConclusion: null, + lastRunConclusion: null, + }, + }, + }); + }); + + it("ignores an older review hint instead of replacing a newer decision", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Reviewed PR", + state: "open", + metadata: { + reviewDecision: "APPROVED", + review: { decision: "APPROVED", updatedAt: "2026-07-14T13:00:00Z" }, + }, + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("stale-review"), + event: "pull_request_review", + payload: { + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + review: { + id: 800, + state: "changes_requested", + submitted_at: "2026-07-14T12:00:00Z", + user: { login: "reviewer" }, + }, + }, + }); + + expect(result.processed).toBe(0); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + metadata: { + reviewDecision: "APPROVED", + review: { decision: "APPROVED", updatedAt: "2026-07-14T13:00:00Z" }, + }, + }); + }); + + it("preserves aggregate review state when another reviewer is requested", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Approved before request", + state: "open", + metadata: { + reviewDecision: "APPROVED", + review: { decision: "APPROVED", updatedAt: "2026-07-14T13:00:00Z" }, + }, + }, + }); + const linkedIssue = await createIssue(fixture, { statusCategory: "IN_PROGRESS" }); + await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: linkedIssue.id, + externalResourceId: resource.id, + kind: "IMPLEMENTS", + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("review-requested"), + event: "pull_request", + payload: { + action: "review_requested", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ updated_at: "2026-07-14T14:00:00Z" }), + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + lastSyncedAt: null, + metadata: { + reviewDecision: "APPROVED", + review: { decision: "APPROVED", updatedAt: "2026-07-14T13:00:00Z" }, + reviewHint: { dirty: true, event: "review_requested", source: "webhook-hint" }, + }, + }); + await expect( + prisma.activityEvent.count({ + where: { + workspaceId: fixture.workspace.id, + subjectType: "issue", + subjectId: linkedIssue.id, + }, + }), + ).resolves.toBeGreaterThan(0); + }); + + it("keeps the decisive review state when a comment-only review arrives", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Approved PR", + state: "open", + metadata: { + reviewDecision: "APPROVED", + review: { decision: "APPROVED", updatedAt: "2026-07-14T13:00:00Z" }, + }, + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("comment-only-review"), + event: "pull_request_review", + payload: { + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + review: { + id: 802, + state: "commented", + submitted_at: "2026-07-14T14:00:00Z", + user: { login: "reviewer" }, + }, + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + lastSyncedAt: null, + metadata: { + reviewDecision: "APPROVED", + review: { + decision: "APPROVED", + updatedAt: "2026-07-14T13:00:00Z", + dirty: true, + lastEventState: "COMMENTED", + lastEventAt: "2026-07-14T14:00:00Z", + }, + }, + }); + }); + + it("does not treat one reviewer approval as the aggregate decision", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Multiple reviewers", + state: "open", + metadata: { + reviewDecision: "CHANGES_REQUESTED", + review: { decision: "CHANGES_REQUESTED", updatedAt: "2026-07-14T13:00:00Z" }, + }, + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("single-approval-review"), + event: "pull_request_review", + payload: { + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + review: { + id: 803, + state: "approved", + submitted_at: "2026-07-14T14:00:00Z", + user: { login: "second-reviewer" }, + }, + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + lastSyncedAt: null, + metadata: { + reviewDecision: "CHANGES_REQUESTED", + review: { + decision: "CHANGES_REQUESTED", + dirty: true, + lastEventDecision: "APPROVED", + lastReviewer: "second-reviewer", + }, + }, + }); + }); + + it("rejects review events older than the latest webhook hint", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Review ordering", + state: "open", + }, + }); + const reviewPayload = (state: string, submittedAt: string) => ({ + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + review: { + id: state === "approved" ? 804 : 805, + state, + submitted_at: submittedAt, + user: { login: "reviewer" }, + }, + }); + + await expect( + processGitHubWebhook({ + db: prisma, + deliveryId: delivery("newer-review-event"), + event: "pull_request_review", + payload: reviewPayload("approved", "2026-07-14T14:00:00Z"), + }), + ).resolves.toMatchObject({ processed: 1 }); + await expect( + processGitHubWebhook({ + db: prisma, + deliveryId: delivery("older-comment-review-event"), + event: "pull_request_review", + payload: reviewPayload("commented", "2026-07-14T13:30:00Z"), + }), + ).resolves.toMatchObject({ processed: 0 }); + await expect( + processGitHubWebhook({ + db: prisma, + deliveryId: delivery("older-review-event"), + event: "pull_request_review", + payload: reviewPayload("changes_requested", "2026-07-14T13:00:00Z"), + }), + ).resolves.toMatchObject({ processed: 0 }); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + metadata: { + review: { + lastEventDecision: "APPROVED", + lastEventAt: "2026-07-14T14:00:00Z", + }, + }, + }); + }); + + it("prevents an expired webhook attempt from overwriting its replacement", async () => { + const prisma = getPrisma(); + const deliveryId = delivery("lease-owner"); + const firstLease = new Date("2026-07-14T10:00:00Z"); + const secondLease = new Date("2026-07-14T10:06:00Z"); + await expect( + claimGitHubWebhookDelivery({ + db: prisma, + deliveryId, + event: "pull_request", + action: "opened", + repoFullName: "acme/forge", + now: firstLease, + }), + ).resolves.toBe("CLAIMED"); + await expect( + claimGitHubWebhookDelivery({ + db: prisma, + deliveryId, + event: "pull_request", + action: "opened", + repoFullName: "acme/forge", + now: secondLease, + }), + ).resolves.toBe("CLAIMED"); + + await expect( + finishGitHubWebhookDelivery({ + db: prisma, + deliveryId, + leaseStartedAt: firstLease, + status: "FAILED", + error: "late failure", + }), + ).resolves.toBe(false); + await expect( + prisma.externalWebhookEvent.findUniqueOrThrow({ + where: { provider_deliveryId: { provider: "GITHUB", deliveryId } }, + }), + ).resolves.toMatchObject({ + status: "RECEIVED", + processingStartedAt: secondLease, + attemptCount: 2, + error: null, + }); + await expect( + finishGitHubWebhookDelivery({ + db: prisma, + deliveryId, + leaseStartedAt: secondLease, + status: "PROCESSED", + }), + ).resolves.toBe(true); + await expect( + claimGitHubWebhookDelivery({ + db: prisma, + deliveryId, + event: "pull_request", + action: "opened", + repoFullName: "acme/forge", + now: new Date("2026-07-14T10:07:00Z"), + }), + ).resolves.toBe("DUPLICATE"); + }); + + it("does not use a partial review read as an ordering watermark", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Partial review read", + state: "open", + metadata: { + reviewDecision: null, + review: { + source: "api-aggregate", + partial: true, + diagnostic: "GitHub reviews unavailable", + updatedAt: "2026-07-14T15:00:00Z", + }, + }, + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("review-after-partial-read"), + event: "pull_request_review", + payload: { + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + review: { + id: 804, + state: "changes_requested", + submitted_at: "2026-07-14T14:00:00Z", + user: { login: "recovering-reviewer" }, + }, + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + lastSyncedAt: null, + metadata: { + reviewDecision: null, + review: { + partial: true, + dirty: true, + lastEventDecision: "CHANGES_REQUESTED", + lastEventAt: "2026-07-14T14:00:00Z", + }, + }, + }); + }); + + it("does not let a review-first delivery suppress PR lifecycle side effects", async () => { + const { fixture, prisma, issue } = await setup(); + const keyedTitle = `Implement ${fixture.workspace.key}-${issue.number}`; + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("review-before-pr-opened"), + event: "pull_request_review", + payload: { + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ + title: keyedTitle, + updated_at: "2026-07-14T13:00:00Z", + }), + review: { + id: 806, + state: "approved", + submitted_at: "2026-07-14T13:00:00Z", + user: { login: "reviewer" }, + }, + }, + }); + + const opened = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("pr-opened-after-review"), + event: "pull_request", + payload: { + action: "opened", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ + title: keyedTitle, + updated_at: "2026-07-14T12:00:00Z", + }), + }, + }); + + expect(opened.processed).toBe(1); + const resource = await prisma.externalResource.findFirstOrThrow({ + where: { + workspaceId: fixture.workspace.id, + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + }, + }); + expect(resource.externalUpdatedAt).toEqual(new Date("2026-07-14T12:00:00Z")); + await expect( + prisma.externalResourceLink.count({ + where: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: resource.id, + kind: "IMPLEMENTS", + }, + }), + ).resolves.toBe(1); + }); + + it("accepts a review dismissal without rewinding its submission watermark", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Dismissed review", + state: "open", + metadata: { + reviewDecision: "APPROVED", + review: { + decision: "APPROVED", + updatedAt: "2026-07-14T15:00:00Z", + lastEventAt: "2026-07-14T14:00:00Z", + }, + }, + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("dismiss-old-review"), + event: "pull_request_review", + payload: { + action: "dismissed", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest(), + review: { + id: 807, + state: "dismissed", + submitted_at: "2026-07-14T13:00:00Z", + user: { login: "reviewer" }, + }, + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + lastSyncedAt: null, + metadata: { + reviewDecision: "APPROVED", + review: { + dirty: true, + lastEventState: "DISMISSED", + lastEventAt: "2026-07-14T14:00:00Z", + }, + }, + }); + }); + + it("canonicalizes legacy identity before applying review hints", async () => { + const { fixture, prisma, mapping } = await setup(); + const resource = await prisma.externalResource.create({ + data: { + workspaceId: fixture.workspace.id, + provider: "GITHUB", + connectionMappingId: mapping.id, + resourceType: "PULL_REQUEST", + repoFullName: "Acme/Forge", + number: 42, + url: "https://github.com/Acme/Forge/pull/42", + title: "Merged canonicalization target", + state: "merged", + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + }, + }); + + const result = await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("mixed-case-review"), + event: "pull_request_review", + payload: { + action: "submitted", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + pull_request: pullRequest({ updated_at: "2026-07-14T11:00:00Z" }), + review: { + id: 801, + state: "approved", + submitted_at: "2026-07-14T12:00:00Z", + user: { login: "reviewer" }, + }, + }, + }); + + expect(result.processed).toBe(1); + await expect( + prisma.externalResource.findUniqueOrThrow({ where: { id: resource.id } }), + ).resolves.toMatchObject({ + repoFullName: "acme/forge", + title: "Merged canonicalization target", + state: "merged", + externalUpdatedAt: new Date("2026-07-14T13:00:00Z"), + metadata: { + reviewDecision: null, + review: { dirty: true, lastEventDecision: "APPROVED" }, + }, + }); + }); + + it("keeps PR discussion on GitHub and deduplicates mirrored issue comments", async () => { + const { fixture, prisma, issue, mapping } = await setup({ syncComments: true }); + const issueResource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "ISSUE", + repoFullName: "acme/forge", + number: 7, + url: "https://github.com/acme/forge/issues/7", + title: "Issue discussion", + state: "open", + }, + }); + await prisma.externalResourceLink.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + externalResourceId: issueResource.id, + kind: "SOURCE", + }, + }); + const issuePayload = { + action: "created", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + issue: { + id: 700, + number: 7, + title: "Issue discussion", + state: "open" as const, + html_url: "https://github.com/acme/forge/issues/7", + url: "https://api.github.com/repos/acme/forge/issues/7", + }, + comment: { + id: 9001, + html_url: "https://github.com/acme/forge/issues/7#issuecomment-9001", + body: "One durable comment", + user: { login: "octocat" }, + }, + }; + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("comment-a"), + event: "issue_comment", + payload: issuePayload, + }); + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("comment-b"), + event: "issue_comment", + payload: issuePayload, + }); + expect( + await prisma.comment.count({ + where: { workspaceId: fixture.workspace.id, issueId: issue.id }, + }), + ).toBe(1); + + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("pr-comment"), + event: "issue_comment", + payload: { + ...issuePayload, + issue: { + ...issuePayload.issue, + number: 42, + pull_request: { url: "https://api.github.com/repos/acme/forge/pulls/42" }, + }, + comment: { ...issuePayload.comment, id: 9002 }, + }, + }); + expect( + await prisma.externalResource.count({ + where: { workspaceId: fixture.workspace.id, resourceType: "ISSUE", number: 42 }, + }), + ).toBe(0); + }); + + it("does not let a newer issue comment suppress issue lifecycle side effects", async () => { + const { fixture, prisma } = await setup({ syncComments: true, autoCreateIssues: true }); + const issue = { + id: 700, + number: 7, + title: "Lifecycle after comment", + state: "open" as const, + html_url: "https://github.com/acme/forge/issues/7", + url: "https://api.github.com/repos/acme/forge/issues/7", + created_at: "2026-07-14T09:00:00Z", + updated_at: "2026-07-14T13:00:00Z", + }; + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("comment-before-opened"), + event: "issue_comment", + payload: { + action: "created", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + issue, + comment: { + id: 9003, + html_url: "https://github.com/acme/forge/issues/7#issuecomment-9003", + body: "Comment arrived first", + user: { login: "octocat" }, + }, + }, + }); + + await processGitHubWebhook({ + db: prisma, + deliveryId: delivery("opened-after-comment"), + event: "issues", + payload: { + action: "opened", + installation: { id: 101 }, + repository: { full_name: "acme/forge" }, + issue: { ...issue, updated_at: "2026-07-14T12:00:00Z" }, + }, + }); + + const resource = await prisma.externalResource.findFirstOrThrow({ + where: { + workspaceId: fixture.workspace.id, + resourceType: "ISSUE", + repoFullName: "acme/forge", + number: 7, + }, + }); + expect(resource.externalUpdatedAt).toEqual(new Date("2026-07-14T12:00:00Z")); + await expect( + prisma.externalResourceLink.count({ + where: { + workspaceId: fixture.workspace.id, + externalResourceId: resource.id, + kind: "SOURCE", + }, + }), + ).resolves.toBe(1); + }); + + it("reclaims failed deliveries once and promotes legacy GitHub link attachments", async () => { + const { fixture, prisma, issue, mapping } = await setup(); + const failedId = delivery("failed"); + await prisma.externalWebhookEvent.create({ + data: { + provider: "GITHUB", + deliveryId: failedId, + event: "pull_request", + status: "FAILED", + processingStartedAt: new Date("2026-07-14T10:00:00Z"), + processedAt: new Date("2026-07-14T10:01:00Z"), + error: "transient", + }, + }); + await expect( + claimGitHubWebhookDelivery({ + db: prisma, + deliveryId: failedId, + event: "pull_request", + action: "opened", + repoFullName: "acme/forge", + now: new Date("2026-07-14T10:02:00Z"), + }), + ).resolves.toBe("CLAIMED"); + await expect( + claimGitHubWebhookDelivery({ + db: prisma, + deliveryId: failedId, + event: "pull_request", + action: "opened", + repoFullName: "acme/forge", + now: new Date("2026-07-14T10:02:01Z"), + }), + ).resolves.toBe("IN_FLIGHT"); + await expect( + prisma.externalWebhookEvent.findUniqueOrThrow({ + where: { provider_deliveryId: { provider: "GITHUB", deliveryId: failedId } }, + }), + ).resolves.toMatchObject({ status: "RECEIVED", attemptCount: 2, error: null }); + + const unmatched = await prisma.attachment.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + targetType: "issue", + targetId: issue.id, + kind: "LINK", + filename: "Unmatched GitHub PR", + mimeType: "text/url", + size: 0, + url: "https://github.com/acme/forge/pull/999", + externalUrl: "https://github.com/acme/forge/pull/999", + createdAt: new Date("2026-07-14T09:00:00Z"), + }, + }); + const oversized = await prisma.attachment.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + targetType: "issue", + targetId: issue.id, + kind: "LINK", + filename: "Oversized GitHub PR number", + mimeType: "text/url", + size: 0, + url: "https://github.com/acme/forge/pull/999999999999999999999999999999999", + externalUrl: "https://github.com/acme/forge/pull/999999999999999999999999999999999", + createdAt: new Date("2026-07-14T07:00:00Z"), + }, + }); + const resource = await upsertExternalResource(prisma, { + workspaceId: fixture.workspace.id, + connectionMappingId: mapping.id, + snapshot: { + provider: "GITHUB", + resourceType: "PULL_REQUEST", + repoFullName: "acme/forge", + number: 42, + url: "https://github.com/acme/forge/pull/42", + title: "Legacy generic attachment", + state: "open", + }, + }); + const deletedIssue = await createIssue(fixture, { statusCategory: "IN_PROGRESS" }); + await prisma.issue.update({ + where: { id: deletedIssue.id }, + data: { deletedAt: new Date("2026-07-14T08:30:00Z") }, + }); + const deletedIssueAttachment = await prisma.attachment.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: deletedIssue.id, + targetType: "issue", + targetId: deletedIssue.id, + kind: "LINK", + filename: "Deleted issue GitHub PR", + mimeType: "text/url", + size: 0, + url: "https://github.com/acme/forge/pull/42", + externalUrl: "https://github.com/acme/forge/pull/42", + createdAt: new Date("2026-07-14T08:00:00Z"), + }, + }); + const attachment = await prisma.attachment.create({ + data: { + workspaceId: fixture.workspace.id, + issueId: issue.id, + targetType: "issue", + targetId: issue.id, + kind: "LINK", + filename: "GitHub PR", + mimeType: "text/url", + size: 0, + url: "https://github.com/ACME/Forge/pull/42?diff=split", + externalUrl: "https://github.com/ACME/Forge/pull/42?diff=split", + }, + }); + await expect( + recoverGenericGitHubAttachments(prisma, { workspaceId: fixture.workspace.id, limit: 1 }), + ).resolves.toMatchObject({ + inspected: 1, + recovered: 1, + unmatched: 0, + }); + expect(await prisma.attachment.findUnique({ where: { id: attachment.id } })).toBeNull(); + await expect( + prisma.attachment.findUnique({ where: { id: unmatched.id } }), + ).resolves.toBeTruthy(); + await expect( + prisma.attachment.findUnique({ where: { id: oversized.id } }), + ).resolves.toBeTruthy(); + await expect( + prisma.attachment.findUnique({ where: { id: deletedIssueAttachment.id } }), + ).resolves.toBeTruthy(); + await expect( + prisma.externalResourceLink.findUnique({ + where: { + issueId_externalResourceId_kind: { + issueId: issue.id, + externalResourceId: resource.id, + kind: "RELATES_TO", + }, + }, + }), + ).resolves.toBeTruthy(); + }); +}); diff --git a/src/server/services/github/client.ts b/src/server/services/github/client.ts index 5759b898..6be03d8e 100644 --- a/src/server/services/github/client.ts +++ b/src/server/services/github/client.ts @@ -6,6 +6,7 @@ import type { GitHubResourceSnapshot } from "@/server/services/github/types"; const GITHUB_API_BASE = "https://api.github.com"; const DEFAULT_GITHUB_REQUEST_TIMEOUT_MS = 10_000; const MAX_CHECK_SUITE_PAGES = 10; +const MAX_REVIEW_PAGES = 10; type GitHubUser = { login?: string | null }; type GitHubLabel = { name?: string | null; color?: string | null }; @@ -46,12 +47,44 @@ export type GitHubPullResponse = { mergeable_state?: string | null; head?: { ref?: string | null; sha?: string | null; repo?: { full_name?: string | null } | null }; base?: { ref?: string | null; sha?: string | null; repo?: { full_name?: string | null } | null }; + requested_reviewers?: GitHubUser[]; + requested_teams?: Array<{ slug?: string | null }>; created_at?: string | null; updated_at?: string | null; closed_at?: string | null; merged_at?: string | null; }; +type GitHubPullReviewResponse = { + id: number; + state?: string | null; + submitted_at?: string | null; + user?: GitHubUser | null; +}; + +export type GitHubReviewSummary = { + decision: "CHANGES_REQUESTED" | "APPROVED" | "REVIEW_REQUESTED" | null; + approvedCount: number; + changesRequestedCount: number; + requestedCount: number; + reviewCount: number; + updatedAt: string; + source: "api-aggregate"; + partial: boolean; + diagnostic: string | null; +}; + +export function gitHubReviewDecision(args: { + approvedCount: number; + changesRequestedCount: number; + requestedCount: number; +}): GitHubReviewSummary["decision"] { + if (args.changesRequestedCount > 0) return "CHANGES_REQUESTED"; + if (args.requestedCount > 0) return "REVIEW_REQUESTED"; + if (args.approvedCount > 0) return "APPROVED"; + return null; +} + export type GitHubRepoResponse = { id: number; node_id?: string | null; @@ -278,6 +311,72 @@ export async function getGitHubPullRequest(args: { ); } +/** + * Aggregate the latest decisive review from each reviewer. GitHub's pull REST + * payload does not expose a review decision, and a single review webhook is + * not an aggregate: a later approval from the same reviewer supersedes their + * earlier changes request, while COMMENTED reviews do not erase approvals. + */ +export async function getGitHubPullRequestReviewSummary(args: { + installationId: string | number; + owner: string; + repo: string; + number: number; + requestedReviewers?: number; + requestedTeams?: number; + requestTimeoutMs?: number; + signal?: AbortSignal; +}): Promise { + const reviews: GitHubPullReviewResponse[] = []; + let truncated = false; + for (let page = 1; page <= MAX_REVIEW_PAGES; page += 1) { + const rows = await githubRequest( + args.installationId, + `/repos/${encodeURIComponent(args.owner)}/${encodeURIComponent(args.repo)}/pulls/${args.number}/reviews?per_page=100&page=${page}`, + { signal: args.signal }, + args.requestTimeoutMs, + ); + reviews.push(...rows); + if (rows.length < 100) break; + if (page === MAX_REVIEW_PAGES) truncated = true; + } + + const ordered = [...reviews].sort((a, b) => { + const at = a.submitted_at ? Date.parse(a.submitted_at) : 0; + const bt = b.submitted_at ? Date.parse(b.submitted_at) : 0; + return at === bt ? a.id - b.id : at - bt; + }); + const latest = new Map(); + for (const review of ordered) { + const login = review.user?.login?.toLowerCase(); + if (!login) continue; + const state = review.state?.toUpperCase(); + if (state === "APPROVED" || state === "CHANGES_REQUESTED") { + latest.set(login, state); + } else if (state === "DISMISSED") { + latest.delete(login); + } + } + const approvedCount = [...latest.values()].filter((state) => state === "APPROVED").length; + const changesRequestedCount = [...latest.values()].filter( + (state) => state === "CHANGES_REQUESTED", + ).length; + const requestedCount = (args.requestedReviewers ?? 0) + (args.requestedTeams ?? 0); + return { + decision: gitHubReviewDecision({ approvedCount, changesRequestedCount, requestedCount }), + approvedCount, + changesRequestedCount, + requestedCount, + reviewCount: reviews.length, + updatedAt: new Date().toISOString(), + source: "api-aggregate", + partial: truncated, + diagnostic: truncated + ? `More than ${MAX_REVIEW_PAGES * 100} pull-request reviews require another page.` + : null, + }; +} + /** * Read both GitHub Checks and legacy commit statuses for a PR head. A passing * result is only returned when every discovered signal is complete and green; @@ -520,6 +619,10 @@ export function pullRequestSnapshot( mergedAt: pr.merged_at ?? null, closedAt: pr.closed_at ?? null, mergeableState: pr.mergeable_state ?? null, + requestedReviewers: (pr.requested_reviewers ?? []) + .map((reviewer) => reviewer.login ?? "") + .filter(Boolean), + requestedTeams: (pr.requested_teams ?? []).map((team) => team.slug ?? "").filter(Boolean), head: { ref: pr.head?.ref ?? null, sha: pr.head?.sha ?? null, diff --git a/src/server/services/github/resource-sync.ts b/src/server/services/github/resource-sync.ts index ce6257a6..4f71c959 100644 --- a/src/server/services/github/resource-sync.ts +++ b/src/server/services/github/resource-sync.ts @@ -17,9 +17,11 @@ import { getGitHubIssue, getGitHubPullRequest, getGitHubPullRequestChecks, + getGitHubPullRequestReviewSummary, GitHubRequestError, issueSnapshot, pullRequestSnapshot, + type GitHubPullResponse, } from "@/server/services/github/client"; import { githubInstallationId, @@ -66,6 +68,215 @@ function publicResource(row: ExternalResource) { return row; } +/** + * Webhook snapshots are not guaranteed to arrive in order. Ignore a payload + * whose provider updated_at predates the row already stored, so an old opened, + * synchronize, or reopened delivery cannot regress a closed/merged resource. + */ +export async function upsertExternalResourceFromWebhook( + db: PrismaClient, + args: { + workspaceId: string; + connectionMappingId?: string | null; + snapshot: GitHubResourceSnapshot; + invalidateSync?: boolean; + allowEqualTimestampReopen?: boolean; + }, +): Promise<{ resource: ExternalResource; previous: ExternalResource | null; applied: boolean }> { + return db.$transaction(async (tx) => { + await acquireGitHubResourceOrderLock(tx, { + workspaceId: args.workspaceId, + repoFullName: args.snapshot.repoFullName, + resourceType: args.snapshot.resourceType, + number: args.snapshot.number, + }); + await canonicalizeGitHubResourceIdentity(tx, args); + const key = { + workspaceId: args.workspaceId, + provider: args.snapshot.provider, + repoFullName: args.snapshot.repoFullName, + resourceType: args.snapshot.resourceType, + number: args.snapshot.number, + }; + const previous = await tx.externalResource.findUnique({ + where: { workspaceId_provider_repoFullName_resourceType_number: key }, + }); + if (previous?.externalUpdatedAt && !args.snapshot.externalUpdatedAt) { + return { resource: previous, previous, applied: false }; + } + if (previous?.externalUpdatedAt && args.snapshot.externalUpdatedAt) { + const previousTime = previous.externalUpdatedAt.getTime(); + const incomingTime = args.snapshot.externalUpdatedAt.getTime(); + const equalTimestampRegression = + previousTime === incomingTime && + ((previous.state === "merged" && args.snapshot.state !== "merged") || + (previous.state === "closed" && + !args.allowEqualTimestampReopen && + args.snapshot.state !== "closed" && + args.snapshot.state !== "merged")); + if (previousTime > incomingTime || equalTimestampRegression) { + return { resource: previous, previous, applied: false }; + } + } + let resource = await upsertExternalResource(tx, args); + if (args.invalidateSync && resource.syncTerminalAt === null) { + resource = await tx.externalResource.update({ + where: { id: resource.id }, + data: { lastSyncedAt: null }, + }); + } + return { resource, previous, applied: true }; + }); +} + +export async function acquireGitHubResourceOrderLock( + db: DbClient, + resource: { + workspaceId: string; + repoFullName: string; + resourceType: string; + number: number; + }, +): Promise { + const key = [ + resource.workspaceId, + GITHUB_PROVIDER, + resource.repoFullName.toLowerCase(), + resource.resourceType, + resource.number, + ].join(":"); + await db.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))`; +} + +export function gitHubResourceVersionMatches( + current: Pick, + expected: Pick, +): boolean { + const currentHead = metadataRecord(metadataRecord(current.metadata).head).sha; + const expectedHead = metadataRecord(metadataRecord(expected.metadata).head).sha; + return ( + current.externalUpdatedAt?.getTime() === expected.externalUpdatedAt?.getTime() && + current.title === expected.title && + current.state === expected.state && + currentHead === expectedHead + ); +} + +function metadataRecord(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + +/** State that should wake issue subscribers; excludes polling timestamps. */ +function resourceStateFingerprint( + resource: Pick, +) { + const metadata = metadataRecord(resource.metadata); + const checks = metadataRecord(metadata.checks); + const review = metadataRecord(metadata.review); + const reviewHint = metadataRecord(metadata.reviewHint); + const head = metadataRecord(metadata.head); + return JSON.stringify({ + title: resource.title, + state: resource.state, + draft: metadata.draft ?? null, + merged: metadata.merged ?? null, + mergeableState: metadata.mergeableState ?? null, + headSha: head.sha ?? null, + reviewDecision: metadata.reviewDecision ?? review.decision ?? null, + review: { + approvedCount: review.approvedCount ?? null, + changesRequestedCount: review.changesRequestedCount ?? null, + requestedCount: review.requestedCount ?? null, + partial: review.partial ?? null, + dirty: review.dirty ?? null, + lastEventDecision: review.lastEventDecision ?? null, + lastEventState: review.lastEventState ?? null, + lastEventAt: review.lastEventAt ?? null, + }, + reviewHint: { + dirty: reviewHint.dirty ?? null, + event: reviewHint.event ?? null, + updatedAt: reviewHint.updatedAt ?? null, + }, + checks: { + status: checks.status ?? null, + conclusion: checks.conclusion ?? null, + source: checks.source ?? null, + partial: checks.partial ?? null, + headSha: checks.headSha ?? null, + observedConclusion: checks.observedConclusion ?? null, + }, + }); +} + +function resourceAuditState(resource: Pick) { + return JSON.parse(resourceStateFingerprint(resource)) as Prisma.InputJsonObject; +} + +export function gitHubResourceStateChanged( + before: Pick | null, + after: Pick, +): boolean { + return before !== null && resourceStateFingerprint(before) !== resourceStateFingerprint(after); +} + +/** + * Publish one issue-scoped activity event when GitHub state changed without a + * Forge field patch. This is the SSE/cache-invalidation bridge for checks, + * draft/review state, mergeability, and head-SHA changes. + */ +export async function recordGitHubResourceChangeToLinkedIssues(args: { + db: DbClient; + workspaceId: string; + before: Pick | null; + after: Pick< + ExternalResource, + "id" | "title" | "state" | "metadata" | "resourceType" | "repoFullName" | "number" | "url" + >; + actor: ActorMeta; + source: "github-webhook" | "github-manual" | "github-reconciliation"; + skipIssueIds?: Set; +}): Promise { + if (!gitHubResourceStateChanged(args.before, args.after)) return 0; + const links = await args.db.externalResourceLink.findMany({ + where: { workspaceId: args.workspaceId, externalResourceId: args.after.id }, + select: { issueId: true }, + }); + let recorded = 0; + for (const link of links) { + if (args.skipIssueIds?.has(link.issueId)) continue; + await recordChange(args.db, { + workspaceId: args.workspaceId, + actorId: args.actor.actorId, + actorAgentId: args.actor.actorAgentId ?? null, + entity: "ExternalResource", + entityId: args.after.id, + action: "github-state-sync", + before: args.before ? resourceAuditState(args.before) : undefined, + after: resourceAuditState(args.after), + eventKind: EventKind.ISSUE_UPDATED, + subjectType: "issue", + subjectId: link.issueId, + payload: { + source: args.source, + change: "external-resource-state", + externalResourceId: args.after.id, + resourceType: args.after.resourceType, + repo: args.after.repoFullName, + number: args.after.number, + state: args.after.state, + url: args.after.url, + }, + ip: args.actor.ip ?? null, + userAgent: args.actor.userAgent ?? null, + }); + recorded += 1; + } + return recorded; +} + export function gitHubPartialChecksError(metadata: unknown): GitHubRequestError | null { if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) return null; const checks = (metadata as Record).checks; @@ -114,6 +325,12 @@ export async function resolveGitHubRepoMapping(args: { if (!mapping) { throw new TRPCError({ code: "NOT_FOUND", message: "GitHub repo mapping not found." }); } + if (repoFullName && !sameRepo(mapping.target, repoFullName)) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "GitHub mapping does not match the requested repository.", + }); + } return mapping; } @@ -143,6 +360,59 @@ export async function resolveGitHubRepoMapping(args: { return mapping; } +async function enrichPullRequestSnapshot(args: { + snapshot: GitHubResourceSnapshot; + pullRequest: GitHubPullResponse; + installationId: string; + owner: string; + repo: string; + requestTimeoutMs?: number; + signal?: AbortSignal; +}): Promise { + const metadata = metadataRecord(args.snapshot.metadata); + try { + const review = await getGitHubPullRequestReviewSummary({ + installationId: args.installationId, + owner: args.owner, + repo: args.repo, + number: args.pullRequest.number, + requestedReviewers: args.pullRequest.requested_reviewers?.length ?? 0, + requestedTeams: args.pullRequest.requested_teams?.length ?? 0, + requestTimeoutMs: args.requestTimeoutMs, + signal: args.signal, + }); + args.snapshot.metadata = { ...metadata, review, reviewDecision: review.decision }; + } catch (error) { + // Review visibility should not prevent PR/check reconciliation. Preserve a + // previously trusted decision during the metadata merge and expose the + // partial read diagnostically instead. + args.snapshot.metadata = { + ...metadata, + review: { + source: "api-aggregate", + partial: true, + diagnostic: + error instanceof Error ? error.message.slice(0, 2_000) : "GitHub reviews unavailable", + updatedAt: new Date().toISOString(), + }, + }; + } + if (args.pullRequest.head?.sha) { + args.snapshot.metadata = { + ...metadataRecord(args.snapshot.metadata), + checks: await getGitHubPullRequestChecks({ + installationId: args.installationId, + owner: args.owner, + repo: args.repo, + headSha: args.pullRequest.head.sha, + requestTimeoutMs: args.requestTimeoutMs, + signal: args.signal, + }), + }; + } + return args.snapshot; +} + export async function fetchGitHubSnapshotForParsed( parsed: ParsedGitHubUrl, mapping: GitHubMappingWithConnection, @@ -158,21 +428,18 @@ export async function fetchGitHubSnapshotForParsed( requestTimeoutMs: options.requestTimeoutMs, signal: options.signal, }); - const snapshot = pullRequestSnapshot(parsed.repoFullName, pr); - if (pr.head?.sha) { - snapshot.metadata = { - ...(snapshot.metadata as Record), - checks: await getGitHubPullRequestChecks({ - installationId, - owner: parsed.owner, - repo: parsed.repo, - headSha: pr.head.sha, - requestTimeoutMs: options.requestTimeoutMs, - signal: options.signal, - }), - }; - } - return snapshot; + // Persist the mapping's canonical repo spelling, not the caller's URL + // casing. GitHub repo identity is case-insensitive while the DB key is not. + const snapshot = pullRequestSnapshot(mapping.target, pr); + return enrichPullRequestSnapshot({ + snapshot, + pullRequest: pr, + installationId, + owner: parsed.owner, + repo: parsed.repo, + requestTimeoutMs: options.requestTimeoutMs, + signal: options.signal, + }); } const issue = await getGitHubIssue({ installationId, @@ -191,23 +458,18 @@ export async function fetchGitHubSnapshotForParsed( requestTimeoutMs: options.requestTimeoutMs, signal: options.signal, }); - const snapshot = pullRequestSnapshot(parsed.repoFullName, pr); - if (pr.head?.sha) { - snapshot.metadata = { - ...(snapshot.metadata as Record), - checks: await getGitHubPullRequestChecks({ - installationId, - owner: parsed.owner, - repo: parsed.repo, - headSha: pr.head.sha, - requestTimeoutMs: options.requestTimeoutMs, - signal: options.signal, - }), - }; - } - return snapshot; + const snapshot = pullRequestSnapshot(mapping.target, pr); + return enrichPullRequestSnapshot({ + snapshot, + pullRequest: pr, + installationId, + owner: parsed.owner, + repo: parsed.repo, + requestTimeoutMs: options.requestTimeoutMs, + signal: options.signal, + }); } - return issueSnapshot(parsed.repoFullName, issue); + return issueSnapshot(mapping.target, issue); } export async function fetchGitHubSnapshot(args: { @@ -234,7 +496,77 @@ export async function fetchGitHubSnapshot(args: { ); } -export async function upsertExternalResource( +export async function canonicalizeGitHubResourceIdentity( + db: DbClient, + args: { + workspaceId: string; + snapshot: GitHubResourceSnapshot; + }, +): Promise { + const candidates = await db.externalResource.findMany({ + where: { + workspaceId: args.workspaceId, + provider: args.snapshot.provider, + resourceType: args.snapshot.resourceType, + number: args.snapshot.number, + }, + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + include: { links: true }, + }); + const equivalent = candidates.filter((row) => + sameRepo(row.repoFullName, args.snapshot.repoFullName), + ); + if (equivalent.length === 0) return; + + const stateRank = (state: string) => (state === "merged" ? 2 : state === "closed" ? 1 : 0); + const canonical = equivalent.reduce((newest, row) => { + const newestHasProviderVersion = newest.externalUpdatedAt !== null; + const rowHasProviderVersion = row.externalUpdatedAt !== null; + if (rowHasProviderVersion !== newestHasProviderVersion) { + return rowHasProviderVersion ? row : newest; + } + const newestAt = (newest.externalUpdatedAt ?? newest.updatedAt).getTime(); + const rowAt = (row.externalUpdatedAt ?? row.updatedAt).getTime(); + if (rowAt !== newestAt) return rowAt > newestAt ? row : newest; + return stateRank(row.state) > stateRank(newest.state) ? row : newest; + }); + + // Older builds could create one row per repository spelling. Preserve every + // issue relation on the canonical row, keeping the original link id unless + // the same issue/kind relation already exists there. + for (const duplicate of equivalent) { + if (duplicate.id === canonical.id) continue; + for (const link of duplicate.links) { + const collision = await db.externalResourceLink.findUnique({ + where: { + issueId_externalResourceId_kind: { + issueId: link.issueId, + externalResourceId: canonical.id, + kind: link.kind, + }, + }, + select: { id: true }, + }); + if (collision) { + await db.externalResourceLink.delete({ where: { id: link.id } }); + } else { + await db.externalResourceLink.update({ + where: { id: link.id }, + data: { externalResourceId: canonical.id }, + }); + } + } + await db.externalResource.delete({ where: { id: duplicate.id } }); + } + if (canonical.repoFullName !== args.snapshot.repoFullName) { + await db.externalResource.update({ + where: { id: canonical.id }, + data: { repoFullName: args.snapshot.repoFullName }, + }); + } +} + +async function upsertExternalResourceInTransaction( db: DbClient, args: { workspaceId: string; @@ -243,6 +575,13 @@ export async function upsertExternalResource( }, ): Promise { const now = new Date(); + await acquireGitHubResourceOrderLock(db, { + workspaceId: args.workspaceId, + repoFullName: args.snapshot.repoFullName, + resourceType: args.snapshot.resourceType, + number: args.snapshot.number, + }); + await canonicalizeGitHubResourceIdentity(db, args); const existing = await db.externalResource.findUnique({ where: { workspaceId_provider_repoFullName_resourceType_number: { @@ -370,6 +709,20 @@ export async function upsertExternalResource( return row; } +export async function upsertExternalResource( + db: DbClient, + args: { + workspaceId: string; + connectionMappingId?: string | null; + snapshot: GitHubResourceSnapshot; + }, +): Promise { + if ("$transaction" in db) { + return db.$transaction((tx) => upsertExternalResourceInTransaction(tx, args)); + } + return upsertExternalResourceInTransaction(db, args); +} + export async function linkExternalResourceToIssue( db: DbClient, args: { @@ -403,6 +756,17 @@ export async function linkExternalResourceToIssue( throw new TRPCError({ code: "NOT_FOUND", message: "External resource not found." }); } + const existingLink = await db.externalResourceLink.findUnique({ + where: { + issueId_externalResourceId_kind: { + issueId: args.issueId, + externalResourceId: args.externalResourceId, + kind: args.kind, + }, + }, + }); + if (existingLink) return existingLink; + const link = await db.externalResourceLink.upsert({ where: { issueId_externalResourceId_kind: { @@ -518,6 +882,84 @@ export async function listLinkedGitHubResources(args: { }); } +/** + * Promote legacy generic GitHub link attachments when the corresponding + * native resource already exists. New attachment writes route natively; this + * bounded repair pass handles rows created before that routing fix without + * making provider requests or inventing state from an opaque URL. + */ +export async function recoverGenericGitHubAttachments( + db: PrismaClient, + options: { workspaceId?: string; limit?: number } = {}, +): Promise<{ inspected: number; recovered: number; unmatched: number }> { + const limit = Math.max(1, Math.min(options.limit ?? 500, 2_000)); + // Filter to rows that already have a matching native resource in SQL. If + // unmatched legacy links occupied the oldest bounded prefix, an ordinary + // attachment query would inspect that same prefix forever and starve every + // recoverable row behind it. + const urlPattern = + "^https?://(?:www\\.)?github\\.com/([^/]+)/([^/]+)/(issues|pull)/([0-9]+)(?:[/?#].*)?$"; + const workspaceFilter = options.workspaceId + ? Prisma.sql`AND a."workspaceId" = ${options.workspaceId}` + : Prisma.empty; + const candidates = await db.$queryRaw< + Array<{ + id: string; + workspaceId: string; + targetId: string; + resourceId: string; + }> + >(Prisma.sql` + SELECT + a."id", + a."workspaceId", + a."targetId", + r."id" AS "resourceId" + FROM "Attachment" a + CROSS JOIN LATERAL regexp_match( + COALESCE(NULLIF(a."externalUrl", ''), a."url"), + ${urlPattern}, + 'i' + ) AS matched(parts) + JOIN "Issue" i + ON i."id" = a."targetId" + AND i."workspaceId" = a."workspaceId" + AND i."deletedAt" IS NULL + JOIN "ExternalResource" r + ON r."workspaceId" = a."workspaceId" + AND r."provider" = ${GITHUB_PROVIDER} + AND lower(r."repoFullName") = lower((matched.parts)[1] || '/' || (matched.parts)[2]) + AND r."resourceType" = CASE + WHEN lower((matched.parts)[3]) = 'pull' THEN 'PULL_REQUEST' + ELSE 'ISSUE' + END + AND r."number"::text = COALESCE(NULLIF(ltrim((matched.parts)[4], '0'), ''), '0') + WHERE a."kind" = 'LINK' + AND a."targetType" = 'issue' + AND a."targetId" IS NOT NULL + ${workspaceFilter} + ORDER BY a."createdAt" ASC, a."id" ASC + LIMIT ${limit} + `); + let recovered = 0; + for (const attachment of candidates) { + await db.$transaction(async (tx) => { + await linkExternalResourceToIssue(tx, { + workspaceId: attachment.workspaceId, + issueId: attachment.targetId, + externalResourceId: attachment.resourceId, + kind: "RELATES_TO", + actor: { actorId: null }, + }); + await tx.attachment.deleteMany({ + where: { id: attachment.id, workspaceId: attachment.workspaceId, kind: "LINK" }, + }); + }); + recovered += 1; + } + return { inspected: candidates.length, recovered, unmatched: 0 }; +} + export async function importGitHubIssue(args: { db: PrismaClient; workspaceId: string; @@ -625,12 +1067,12 @@ async function applyIssuePatchFromGitHub(args: { statusId?: string | null; actor: ActorMeta; payload: Prisma.InputJsonObject; -}): Promise { +}): Promise { const before = await args.tx.issue.findFirst({ where: { id: args.issueId, workspaceId: args.workspaceId, deletedAt: null }, include: { status: true }, }); - if (!before) return; + if (!before) return false; const patch: Prisma.IssueUpdateInput = {}; if (args.title && args.title !== before.title) patch.title = args.title; @@ -641,7 +1083,7 @@ async function applyIssuePatchFromGitHub(args: { where: { id: args.statusId, workspaceId: args.workspaceId }, select: { id: true, category: true }, }); - if (!status) return; + if (!status) return false; patch.status = { connect: { id: status.id } }; nextCategory = status.category; if (status.category === "IN_PROGRESS" && !before.startedAt) patch.startedAt = new Date(); @@ -651,7 +1093,7 @@ async function applyIssuePatchFromGitHub(args: { if (status.category !== "CANCELED") patch.canceledAt = null; } - if (Object.keys(patch).length === 0) return; + if (Object.keys(patch).length === 0) return false; const after = await args.tx.issue.update({ where: { id: before.id }, @@ -674,6 +1116,7 @@ async function applyIssuePatchFromGitHub(args: { ip: args.actor.ip ?? null, userAgent: args.actor.userAgent ?? null, }); + return true; } export async function applyGitHubSnapshotToLinkedIssues(args: { @@ -684,14 +1127,15 @@ export async function applyGitHubSnapshotToLinkedIssues(args: { snapshot: GitHubResourceSnapshot; actor: ActorMeta; statusRuleId?: string | null; -}): Promise { +}): Promise> { const config = readGitHubMappingConfig(args.mapping.config); const links = await args.tx.externalResourceLink.findMany({ where: { workspaceId: args.workspaceId, externalResourceId: args.resourceId }, select: { issueId: true, kind: true }, }); + const changedIssueIds = new Set(); for (const link of links) { - await applyIssuePatchFromGitHub({ + const changed = await applyIssuePatchFromGitHub({ tx: args.tx, workspaceId: args.workspaceId, issueId: link.issueId, @@ -708,7 +1152,9 @@ export async function applyGitHubSnapshotToLinkedIssues(args: { state: args.snapshot.state, }, }); + if (changed) changedIssueIds.add(link.issueId); } + return changedIssueIds; } export async function syncGitHubExternalResource(args: { @@ -815,7 +1261,7 @@ export async function syncGitHubExternalResource(args: { connectionMappingId: mapping.id, snapshot, }); - await applyGitHubSnapshotToLinkedIssues({ + const changedIssueIds = await applyGitHubSnapshotToLinkedIssues({ tx, workspaceId: args.workspaceId, resourceId: resource.id, @@ -823,6 +1269,15 @@ export async function syncGitHubExternalResource(args: { snapshot, actor: args.actor, }); + await recordGitHubResourceChangeToLinkedIssues({ + db: tx, + workspaceId: args.workspaceId, + before: existing, + after: resource, + actor: args.actor, + source: args.skipCollisionGuard ? "github-reconciliation" : "github-manual", + skipIssueIds: changedIssueIds, + }); return publicResource(resource); }); const partialChecksError = gitHubPartialChecksError(snapshot.metadata); diff --git a/src/server/services/github/webhook.ts b/src/server/services/github/webhook.ts index 35318619..616cbcb1 100644 --- a/src/server/services/github/webhook.ts +++ b/src/server/services/github/webhook.ts @@ -1,6 +1,6 @@ import "server-only"; import { createHmac, timingSafeEqual } from "node:crypto"; -import { CommentKind, Prisma, type PrismaClient } from "@prisma/client"; +import { CommentKind, type ExternalResource, type Prisma, type PrismaClient } from "@prisma/client"; import { recordChange } from "@/server/audit"; import { createIssueWithSideEffects } from "@/server/services/issue-create"; import { reconcileGitHubPullRequestCompletion } from "@/server/services/completion-candidate"; @@ -17,8 +17,13 @@ import { } from "@/server/services/github/mapping-policy"; import { applyGitHubSnapshotToLinkedIssues, + acquireGitHubResourceOrderLock, + canonicalizeGitHubResourceIdentity, + gitHubResourceVersionMatches, linkExternalResourceToIssue, + recordGitHubResourceChangeToLinkedIssues, upsertExternalResource, + upsertExternalResourceFromWebhook, type ActorMeta, } from "@/server/services/github/resource-sync"; import { GITHUB_PROVIDER, type GitHubResourceSnapshot } from "@/server/services/github/types"; @@ -33,11 +38,18 @@ type GitHubWebhookInstallation = { type GitHubWebhookPayload = { action?: string; + sha?: string | null; + state?: string | null; installation?: GitHubWebhookInstallation; repository?: GitHubWebhookRepository; issue?: GitHubIssueResponse; pull_request?: GitHubPullResponse; - review?: { state?: string | null }; + review?: { + id?: number | null; + state?: string | null; + submitted_at?: string | null; + user?: { login?: string | null } | null; + }; check_suite?: { head_sha?: string | null; conclusion?: string | null; @@ -51,12 +63,15 @@ type GitHubWebhookPayload = { pull_requests?: Array<{ number?: number | null; url?: string | null }>; }; comment?: { + id?: number | null; html_url?: string | null; body?: string | null; user?: { login?: string | null } | null; }; }; +const WEBHOOK_PROCESSING_LEASE_MS = 5 * 60_000; + export type GitHubWebhookResult = { ok: true; duplicate?: boolean; @@ -108,13 +123,101 @@ async function activeMappingsForPayload(args: { if (row.target.toLowerCase() !== args.repoFullName.toLowerCase()) return false; if (!args.installationId) return true; const cfg = row.connection.config; - if (!cfg || typeof cfg !== "object" || !("installationId" in cfg)) return true; + if (!cfg || typeof cfg !== "object" || !("installationId" in cfg)) return false; return ( String((cfg as { installationId?: unknown }).installationId) === String(args.installationId) ); }); } +/** Claim a new, failed, or abandoned delivery without racing an in-flight request. */ +export async function claimGitHubWebhookDelivery(args: { + db: PrismaClient; + deliveryId: string; + event: string; + action: string | null; + repoFullName: string | null; + now?: Date; +}): Promise<"CLAIMED" | "IN_FLIGHT" | "DUPLICATE"> { + const now = args.now ?? new Date(); + const inserted = await args.db.externalWebhookEvent.createMany({ + data: [ + { + provider: GITHUB_PROVIDER, + deliveryId: args.deliveryId, + event: args.event, + action: args.action, + repoFullName: args.repoFullName, + status: "RECEIVED", + processingStartedAt: now, + attemptCount: 1, + }, + ], + skipDuplicates: true, + }); + if (inserted.count === 1) return "CLAIMED"; + + const staleBefore = new Date(now.getTime() - WEBHOOK_PROCESSING_LEASE_MS); + const reclaimed = await args.db.externalWebhookEvent.updateMany({ + where: { + provider: GITHUB_PROVIDER, + deliveryId: args.deliveryId, + OR: [ + { status: "FAILED" }, + { + status: "RECEIVED", + OR: [{ processingStartedAt: null }, { processingStartedAt: { lte: staleBefore } }], + }, + ], + }, + data: { + event: args.event, + action: args.action, + repoFullName: args.repoFullName, + status: "RECEIVED", + error: null, + processedAt: null, + processingStartedAt: now, + attemptCount: { increment: 1 }, + }, + }); + if (reclaimed.count === 1) return "CLAIMED"; + const existing = await args.db.externalWebhookEvent.findUnique({ + where: { + provider_deliveryId: { provider: GITHUB_PROVIDER, deliveryId: args.deliveryId }, + }, + select: { status: true }, + }); + return existing?.status === "RECEIVED" ? "IN_FLIGHT" : "DUPLICATE"; +} + +/** Finish only the attempt that still owns the processing lease. */ +export async function finishGitHubWebhookDelivery(args: { + db: PrismaClient; + deliveryId: string; + leaseStartedAt: Date; + status: "PROCESSED" | "SKIPPED" | "FAILED"; + workspaceId?: string | null; + error?: string | null; + processedAt?: Date; +}): Promise { + const updated = await args.db.externalWebhookEvent.updateMany({ + where: { + provider: GITHUB_PROVIDER, + deliveryId: args.deliveryId, + status: "RECEIVED", + processingStartedAt: args.leaseStartedAt, + }, + data: { + workspaceId: args.workspaceId, + status: args.status, + error: args.error ?? null, + processedAt: args.processedAt ?? new Date(), + }, + }); + return updated.count === 1; +} + function forgeIssueKeys(text: string | null | undefined, workspaceKey: string): number[] { if (!text) return []; const escaped = workspaceKey.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); @@ -133,13 +236,10 @@ async function ensureSourceIssueForSnapshot(args: { workspaceId: string; mapping: Parameters[0]["mapping"] & { id: string }; snapshot: GitHubResourceSnapshot; + resource: ExternalResource; actor: ActorMeta; }) { - const resource = await upsertExternalResource(args.db, { - workspaceId: args.workspaceId, - connectionMappingId: args.mapping.id, - snapshot: args.snapshot, - }); + const resource = args.resource; const existing = await args.db.externalResourceLink.findFirst({ where: { workspaceId: args.workspaceId, @@ -185,11 +285,14 @@ async function processIssueEvent(args: { const workspaceId = args.mapping.workspace.id; const config = readGitHubMappingConfig(args.mapping.config); const snapshot = issueSnapshot(args.mapping.target, args.payload.issue); - let resource = await upsertExternalResource(args.db, { + const persisted = await upsertExternalResourceFromWebhook(args.db, { workspaceId, connectionMappingId: args.mapping.id, snapshot, + allowEqualTimestampReopen: args.payload.action === "reopened", }); + if (!persisted.applied) return 0; + let resource = persisted.resource; if (args.payload.action === "opened" && config.autoCreateIssues) { const created = await ensureSourceIssueForSnapshot({ @@ -197,6 +300,7 @@ async function processIssueEvent(args: { workspaceId, mapping: args.mapping, snapshot, + resource, actor: args.actor, }); resource = created.resource; @@ -209,17 +313,40 @@ async function processIssueEvent(args: { ? statusRuleForGitHubEvent(config, "issueReopenedStatusId") : null; - await args.db.$transaction((tx) => - applyGitHubSnapshotToLinkedIssues({ + await args.db.$transaction(async (tx) => { + await acquireGitHubResourceOrderLock(tx, { + workspaceId, + repoFullName: resource.repoFullName, + resourceType: resource.resourceType, + number: resource.number, + }); + const current = await tx.externalResource.findUnique({ where: { id: resource.id } }); + if (!current || !gitHubResourceVersionMatches(current, resource)) return; + const appliedSnapshot = { + ...snapshot, + title: current.title, + state: current.state, + metadata: current.metadata, + }; + const changedIssueIds = await applyGitHubSnapshotToLinkedIssues({ tx, workspaceId, resourceId: resource.id, mapping: args.mapping, - snapshot, + snapshot: appliedSnapshot, actor: args.actor, statusRuleId: rule, - }), - ); + }); + await recordGitHubResourceChangeToLinkedIssues({ + db: tx, + workspaceId, + before: persisted.previous, + after: current, + actor: args.actor, + source: "github-webhook", + skipIssueIds: changedIssueIds, + }); + }); return 1; } @@ -233,33 +360,51 @@ async function processPullRequestEvent(args: { const workspaceId = args.mapping.workspace.id; const config = readGitHubMappingConfig(args.mapping.config); const snapshot = pullRequestSnapshot(args.mapping.target, args.payload.pull_request); - const resource = await upsertExternalResource(args.db, { + const actionNeedsAggregateRefresh = [ + "opened", + "reopened", + "synchronize", + "ready_for_review", + "converted_to_draft", + "review_requested", + "review_request_removed", + "closed", + ].includes(args.payload.action ?? ""); + if ( + args.payload.action === "review_requested" || + args.payload.action === "review_request_removed" + ) { + snapshot.metadata = { + ...(snapshot.metadata && typeof snapshot.metadata === "object" ? snapshot.metadata : {}), + reviewHint: { + dirty: true, + event: args.payload.action, + source: "webhook-hint", + updatedAt: args.payload.pull_request.updated_at ?? new Date().toISOString(), + }, + }; + } + const persisted = await upsertExternalResourceFromWebhook(args.db, { workspaceId, connectionMappingId: args.mapping.id, snapshot, + invalidateSync: actionNeedsAggregateRefresh, + allowEqualTimestampReopen: args.payload.action === "reopened", }); + if (!persisted.applied) return 0; + const resource = persisted.resource; const keyNumbers = [ ...forgeIssueKeys(args.payload.pull_request.title, args.mapping.workspace.key), ...forgeIssueKeys(args.payload.pull_request.body, args.mapping.workspace.key), ]; - if (keyNumbers.length > 0) { - const issues = await args.db.issue.findMany({ - where: { workspaceId, number: { in: keyNumbers }, deletedAt: null }, - select: { id: true }, - }); - await args.db.$transaction(async (tx) => { - for (const issue of issues) { - await linkExternalResourceToIssue(tx, { - workspaceId, - issueId: issue.id, - externalResourceId: resource.id, - kind: "IMPLEMENTS", - actor: args.actor, - }); - } - }); - } + const issues = + keyNumbers.length > 0 + ? await args.db.issue.findMany({ + where: { workspaceId, number: { in: keyNumbers }, deletedAt: null }, + select: { id: true }, + }) + : []; const rule = args.payload.action === "opened" @@ -270,17 +415,48 @@ async function processPullRequestEvent(args: { ? statusRuleForGitHubEvent(config, "prMergedStatusId") : null; - await args.db.$transaction((tx) => - applyGitHubSnapshotToLinkedIssues({ + await args.db.$transaction(async (tx) => { + await acquireGitHubResourceOrderLock(tx, { + workspaceId, + repoFullName: resource.repoFullName, + resourceType: resource.resourceType, + number: resource.number, + }); + const current = await tx.externalResource.findUnique({ where: { id: resource.id } }); + if (!current || !gitHubResourceVersionMatches(current, resource)) return; + for (const issue of issues) { + await linkExternalResourceToIssue(tx, { + workspaceId, + issueId: issue.id, + externalResourceId: resource.id, + kind: "IMPLEMENTS", + actor: args.actor, + }); + } + const changedIssueIds = await applyGitHubSnapshotToLinkedIssues({ tx, workspaceId, resourceId: resource.id, mapping: args.mapping, - snapshot, + snapshot: { + ...snapshot, + title: current.title, + state: current.state, + metadata: current.metadata, + }, actor: args.actor, statusRuleId: rule, - }), - ); + }); + await recordGitHubResourceChangeToLinkedIssues({ + db: tx, + workspaceId, + before: persisted.previous, + after: current, + actor: args.actor, + source: "github-webhook", + skipIssueIds: changedIssueIds, + }); + }); await reconcileGitHubPullRequestCompletion(args.db, { workspaceId, externalResourceId: resource.id, @@ -297,38 +473,158 @@ async function processReviewEvent(args: { actor: ActorMeta; }): Promise { if (!args.payload.pull_request || !args.mapping.workspace?.id) return 0; + const workspaceId = args.mapping.workspace.id; const config = readGitHubMappingConfig(args.mapping.config); const snapshot = pullRequestSnapshot(args.mapping.target, args.payload.pull_request); - snapshot.metadata = { - ...(snapshot.metadata && typeof snapshot.metadata === "object" ? snapshot.metadata : {}), - reviewDecision: args.payload.review?.state?.toUpperCase() ?? null, - }; - const resource = await upsertExternalResource(args.db, { - workspaceId: args.mapping.workspace.id, - connectionMappingId: args.mapping.id, - snapshot, - }); + // pull_request.updated_at in a review payload reflects review activity, not + // a PR lifecycle transition. A review-first delivery may seed identity and + // review hints, but must not make delayed opened/synchronize events stale. + snapshot.externalUpdatedAt = null; + const reviewAction = args.payload.action ?? null; + const submittedAt = args.payload.review?.submitted_at ?? null; + const reviewState = args.payload.review?.state?.toUpperCase() ?? null; + const decision = + reviewState === "APPROVED" || reviewState === "CHANGES_REQUESTED" ? reviewState : null; const rule = - args.payload.review?.state === "changes_requested" + decision === "CHANGES_REQUESTED" ? statusRuleForGitHubEvent(config, "prChangesRequestedStatusId") : null; - await args.db.$transaction((tx) => - applyGitHubSnapshotToLinkedIssues({ + const persisted = await args.db.$transaction(async (tx) => { + await acquireGitHubResourceOrderLock(tx, { + workspaceId, + repoFullName: snapshot.repoFullName, + resourceType: snapshot.resourceType, + number: snapshot.number, + }); + await canonicalizeGitHubResourceIdentity(tx, { workspaceId, snapshot }); + const previous = await tx.externalResource.findUnique({ + where: { + workspaceId_provider_repoFullName_resourceType_number: { + workspaceId, + provider: GITHUB_PROVIDER, + repoFullName: snapshot.repoFullName, + resourceType: "PULL_REQUEST", + number: snapshot.number, + }, + }, + }); + const existingMetadata = + previous?.metadata && + typeof previous.metadata === "object" && + !Array.isArray(previous.metadata) + ? (previous.metadata as Record) + : {}; + const existingReview = + existingMetadata.review && + typeof existingMetadata.review === "object" && + !Array.isArray(existingMetadata.review) + ? (existingMetadata.review as Record) + : {}; + // A partial provider aggregate timestamps the failed/truncated read, not a + // trusted review decision. It must not suppress a delayed decisive webhook. + const trustedAggregateUpdatedAt = + existingReview.partial === true ? null : existingReview.updatedAt; + const existingTimes = [trustedAggregateUpdatedAt, existingReview.lastEventAt] + .filter((value): value is string => typeof value === "string") + .map((value) => Date.parse(value)) + .filter(Number.isFinite); + const existingUpdatedAt = existingTimes.length > 0 ? Math.max(...existingTimes) : NaN; + const incomingUpdatedAt = submittedAt ? Date.parse(submittedAt) : NaN; + if ( + reviewAction !== "dismissed" && + Number.isFinite(existingUpdatedAt) && + Number.isFinite(incomingUpdatedAt) && + existingUpdatedAt > incomingUpdatedAt + ) { + return null; + } + // One review webhook is never the repository-wide decision: another + // reviewer may still have changes requested. Preserve the last provider + // aggregate and make every individual event a refresh hint. + const persistedDecision = + typeof existingMetadata.reviewDecision === "string" ? existingMetadata.reviewDecision : null; + // GitHub's dismissed payload retains the original review submitted_at; + // it does not timestamp the dismissal. Accept the invalidation without + // rewinding a newer webhook watermark. + const lastEventAt = + reviewAction === "dismissed" && typeof existingReview.lastEventAt === "string" + ? existingReview.lastEventAt + : (submittedAt ?? new Date().toISOString()); + const review = { + ...existingReview, + source: "webhook-hint", + dirty: true, + lastEventDecision: decision, + lastEventState: reviewState, + lastEventAt, + lastReviewId: args.payload.review?.id ?? null, + lastReviewer: args.payload.review?.user?.login ?? null, + }; + let resource; + if (previous) { + resource = await tx.externalResource.update({ + where: { id: previous.id }, + data: { + connectionMappingId: args.mapping.id, + lastSyncedAt: null, + metadata: { + ...existingMetadata, + reviewDecision: persistedDecision, + review, + } as Prisma.InputJsonValue, + }, + }); + } else { + snapshot.metadata = { + ...(snapshot.metadata && typeof snapshot.metadata === "object" ? snapshot.metadata : {}), + reviewDecision: persistedDecision, + review, + }; + resource = await upsertExternalResource(tx, { + workspaceId, + connectionMappingId: args.mapping.id, + snapshot, + }); + if (resource.syncTerminalAt === null) { + resource = await tx.externalResource.update({ + where: { id: resource.id }, + data: { lastSyncedAt: null }, + }); + } + } + const appliedSnapshot = { + ...snapshot, + title: resource.title, + state: resource.state, + metadata: resource.metadata, + }; + const changedIssueIds = await applyGitHubSnapshotToLinkedIssues({ tx, - workspaceId: args.mapping.workspace.id, + workspaceId, resourceId: resource.id, mapping: args.mapping, - snapshot, + snapshot: appliedSnapshot, actor: args.actor, statusRuleId: rule, - }), - ); + }); + await recordGitHubResourceChangeToLinkedIssues({ + db: tx, + workspaceId, + before: previous, + after: resource, + actor: args.actor, + source: "github-webhook", + skipIssueIds: changedIssueIds, + }); + return resource; + }); + if (!persisted) return 0; return 1; } -function completedCheckPullRequestNumbers(payload: GitHubWebhookPayload): number[] { +function checkPullRequestNumbers(payload: GitHubWebhookPayload): number[] { const check = payload.check_suite ?? payload.check_run; - if (!check || check.status !== "completed") return []; + if (!check) return []; const out = new Set(); for (const pr of check.pull_requests ?? []) { @@ -341,7 +637,7 @@ function completedCheckPullRequestNumbers(payload: GitHubWebhookPayload): number } export function githubCheckWebhookHint(args: { - event: "check_suite" | "check_run"; + event: "check_suite" | "check_run" | "status"; conclusion: string | null; headSha?: string | null; }) { @@ -363,16 +659,32 @@ async function processCheckEvent(args: { actor: ActorMeta; }): Promise { if (!args.mapping.workspace?.id) return 0; - const numbers = completedCheckPullRequestNumbers(args.payload); - if (numbers.length === 0) return 0; + const numbers = checkPullRequestNumbers(args.payload); + const eventHeadSha = + args.payload.check_suite?.head_sha ?? args.payload.check_run?.head_sha ?? args.payload.sha; + if (numbers.length === 0 && !eventHeadSha) return 0; const workspaceId = args.mapping.workspace.id; const config = readGitHubMappingConfig(args.mapping.config); - const conclusion = - args.payload.check_suite?.conclusion ?? args.payload.check_run?.conclusion ?? null; + // GitHub sends rerequested (and check-suite requested) before the check run + // itself is reset, so those payloads can still carry the previous failure. + // They are cache-invalidation signals only; reconciliation will fetch the + // new aggregate once GitHub has advanced it. + const resetsChecks = ["requested", "rerequested"].includes(args.payload.action ?? ""); + const conclusion = resetsChecks + ? null + : (args.payload.check_suite?.conclusion ?? + args.payload.check_run?.conclusion ?? + args.payload.state ?? + null); const successfulConclusion = conclusion !== null && ["success", "neutral", "skipped"].includes(conclusion); - const failedConclusion = conclusion !== null && !successfulConclusion; + const checkStatus = + args.payload.check_suite?.status ?? + args.payload.check_run?.status ?? + (args.payload.state ? (args.payload.state === "pending" ? "pending" : "completed") : null); + const failedConclusion = + checkStatus === "completed" && conclusion !== null && !successfulConclusion; const statusRuleId = failedConclusion ? statusRuleForGitHubEvent(config, "checksFailedStatusId") : null; @@ -383,21 +695,32 @@ async function processCheckEvent(args: { }) : null; - const resources = await args.db.externalResource.findMany({ + const candidates = await args.db.externalResource.findMany({ where: { workspaceId, provider: GITHUB_PROVIDER, - repoFullName: args.mapping.target, + repoFullName: { equals: args.mapping.target, mode: "insensitive" }, resourceType: "PULL_REQUEST", - number: { in: numbers }, + ...(numbers.length > 0 + ? { number: { in: numbers } } + : { metadata: { path: ["head", "sha"], equals: eventHeadSha! } }), }, - select: { id: true, number: true, state: true, metadata: true }, }); + const resources = candidates; if (resources.length === 0) return 0; let processed = 0; + const processedResourceIds: string[] = []; await args.db.$transaction(async (tx) => { - for (const resource of resources) { + for (const candidate of resources) { + await acquireGitHubResourceOrderLock(tx, { + workspaceId, + repoFullName: candidate.repoFullName, + resourceType: candidate.resourceType, + number: candidate.number, + }); + const resource = await tx.externalResource.findUnique({ where: { id: candidate.id } }); + if (!resource) continue; const metadata = resource.metadata && typeof resource.metadata === "object" && @@ -412,7 +735,6 @@ async function processCheckEvent(args: { metadata.head && typeof metadata.head === "object" && !Array.isArray(metadata.head) ? (metadata.head as Record) : {}; - const eventHeadSha = args.payload.check_suite?.head_sha ?? args.payload.check_run?.head_sha; if (eventHeadSha && typeof head.sha === "string" && head.sha !== eventHeadSha) { continue; } @@ -420,11 +742,15 @@ async function processCheckEvent(args: { // Persist it as a dirty hint; the worker must fetch every suite plus the // combined commit status before completion can trust a conclusion. const webhookHint = githubCheckWebhookHint({ - event: args.payload.check_suite ? "check_suite" : "check_run", + event: args.payload.check_suite + ? "check_suite" + : args.payload.check_run + ? "check_run" + : "status", conclusion, headSha: eventHeadSha, }); - await tx.externalResource.update({ + const after = await tx.externalResource.update({ where: { id: resource.id }, data: { metadata: { @@ -433,6 +759,7 @@ async function processCheckEvent(args: { ...existingChecks, ...webhookHint, ...(args.payload.check_run ? { lastRunConclusion: conclusion } : {}), + ...(args.payload.state ? { lastStatusState: args.payload.state } : {}), }, } as Prisma.InputJsonValue, lastSyncedAt: null, @@ -440,61 +767,78 @@ async function processCheckEvent(args: { }, }); processed += 1; - if (!status) continue; - const links = await tx.externalResourceLink.findMany({ - where: { workspaceId, externalResourceId: resource.id }, - select: { issueId: true }, - }); - for (const link of links) { - const before = await tx.issue.findFirst({ - where: { id: link.issueId, workspaceId, deletedAt: null }, - include: { status: true }, - }); - if (!before || before.statusId === status.id) continue; - - const after = await tx.issue.update({ - where: { id: before.id }, - data: { - status: { connect: { id: status.id } }, - ...(status.category === "IN_PROGRESS" && !before.startedAt - ? { startedAt: new Date() } - : {}), - completedAt: status.category === "DONE" ? new Date() : null, - canceledAt: status.category === "CANCELED" ? new Date() : null, - }, - include: { status: true }, - }); - await recordChange(tx, { - workspaceId, - actorId: args.actor.actorId, - actorAgentId: args.actor.actorAgentId ?? null, - entity: "Issue", - entityId: before.id, - action: "github-checks-failed", - before, - after, - eventKind: "ISSUE_STATUS_CHANGED", - subjectType: "issue", - subjectId: before.id, - payload: { - source: "github", - change: "checks-failed", - externalResourceId: resource.id, - repo: args.mapping.target, - number: resource.number, - event: args.payload.check_suite ? "check_suite" : "check_run", - conclusion, - }, - ip: args.actor.ip ?? null, - userAgent: args.actor.userAgent ?? null, + processedResourceIds.push(resource.id); + const changedIssueIds = new Set(); + if (status) { + const links = await tx.externalResourceLink.findMany({ + where: { workspaceId, externalResourceId: resource.id }, + select: { issueId: true }, }); + for (const link of links) { + const before = await tx.issue.findFirst({ + where: { id: link.issueId, workspaceId, deletedAt: null }, + include: { status: true }, + }); + if (!before || before.statusId === status.id) continue; + + const issueAfter = await tx.issue.update({ + where: { id: before.id }, + data: { + status: { connect: { id: status.id } }, + ...(status.category === "IN_PROGRESS" && !before.startedAt + ? { startedAt: new Date() } + : {}), + completedAt: status.category === "DONE" ? new Date() : null, + canceledAt: status.category === "CANCELED" ? new Date() : null, + }, + include: { status: true }, + }); + await recordChange(tx, { + workspaceId, + actorId: args.actor.actorId, + actorAgentId: args.actor.actorAgentId ?? null, + entity: "Issue", + entityId: before.id, + action: "github-checks-failed", + before, + after: issueAfter, + eventKind: "ISSUE_STATUS_CHANGED", + subjectType: "issue", + subjectId: before.id, + payload: { + source: "github", + change: "checks-failed", + externalResourceId: resource.id, + repo: args.mapping.target, + number: resource.number, + event: args.payload.check_suite + ? "check_suite" + : args.payload.check_run + ? "check_run" + : "status", + conclusion, + }, + ip: args.actor.ip ?? null, + userAgent: args.actor.userAgent ?? null, + }); + changedIssueIds.add(before.id); + } } + await recordGitHubResourceChangeToLinkedIssues({ + db: tx, + workspaceId, + before: resource, + after, + actor: args.actor, + source: "github-webhook", + skipIssueIds: changedIssueIds, + }); } }); - for (const resource of resources) { + for (const externalResourceId of processedResourceIds) { await reconcileGitHubPullRequestCompletion(args.db, { workspaceId, - externalResourceId: resource.id, + externalResourceId, actorId: args.actor.actorId, actorAgentId: args.actor.actorAgentId, }); @@ -507,24 +851,55 @@ async function processIssueComment(args: { mapping: Awaited>[number]; payload: GitHubWebhookPayload; actor: ActorMeta; + deliveryId: string; }): Promise { if (args.payload.action !== "created" || !args.payload.issue || !args.payload.comment) return 0; + // PR conversation comments and review bodies remain on GitHub. Forge only + // mirrors issue discussion when syncComments is explicitly enabled; review + // webhooks update bounded state metadata, never Forge BODY comments. + if (args.payload.issue.pull_request) return 0; if (!args.mapping.workspace?.id) return 0; const workspaceId = args.mapping.workspace.id; const config = readGitHubMappingConfig(args.mapping.config); if (!config.syncComments) return 0; const snapshot = issueSnapshot(args.mapping.target, args.payload.issue); - const resource = await upsertExternalResource(args.db, { + // issue.updated_at on an issue_comment payload reflects the comment, not an + // issue lifecycle transition. Persisting it as the resource version could + // make a later opened/closed/reopened delivery look stale and suppress its + // Forge side effects. A comment may seed the native resource identity, but + // lifecycle webhooks (or provider reconciliation) own its freshness clock. + snapshot.externalUpdatedAt = null; + const persisted = await upsertExternalResourceFromWebhook(args.db, { workspaceId, connectionMappingId: args.mapping.id, snapshot, }); + const resource = persisted.resource; const links = await args.db.externalResourceLink.findMany({ where: { workspaceId, externalResourceId: resource.id, kind: "SOURCE" }, select: { issueId: true }, }); for (const link of links) { await args.db.$transaction(async (tx) => { + await acquireGitHubResourceOrderLock(tx, { + workspaceId, + repoFullName: resource.repoFullName, + resourceType: resource.resourceType, + number: resource.number, + }); + const sourceCommentId = args.payload.comment?.id + ? String(args.payload.comment.id) + : args.deliveryId; + const dedupeKey = `${sourceCommentId}:${link.issueId}`; + const duplicate = await tx.activityEvent.findFirst({ + where: { + workspaceId, + kind: "COMMENT_CREATED", + payload: { path: ["githubCommentDedupeKey"], equals: dedupeKey }, + }, + select: { id: true }, + }); + if (duplicate) return; const comment = await tx.comment.create({ data: { workspaceId, @@ -556,6 +931,9 @@ async function processIssueComment(args: { source: "github", externalResourceId: resource.id, commentUrl: args.payload.comment?.html_url ?? null, + githubCommentId: args.payload.comment?.id ?? null, + githubDeliveryId: args.deliveryId, + githubCommentDedupeKey: dedupeKey, }, ip: args.actor.ip ?? null, userAgent: args.actor.userAgent ?? null, @@ -575,29 +953,33 @@ export async function processGitHubWebhook(args: { const repoFullName = repoFromPayload(args.payload); const action = args.payload.action ?? null; const actor = args.actor ?? { actorId: null }; + const leaseStartedAt = new Date(); - try { - await args.db.externalWebhookEvent.create({ - data: { - provider: GITHUB_PROVIDER, - deliveryId: args.deliveryId, - event: args.event, - action, - repoFullName, - status: "RECEIVED", - }, - }); - } catch (err) { - if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === "P2002") { - return { ok: true, duplicate: true, processed: 0 }; - } - throw err; + const claim = await claimGitHubWebhookDelivery({ + db: args.db, + deliveryId: args.deliveryId, + event: args.event, + action, + repoFullName, + now: leaseStartedAt, + }); + if (claim === "DUPLICATE") { + return { ok: true, duplicate: true, processed: 0 }; + } + if (claim === "IN_FLIGHT") { + // There is no independent abandoned-delivery sweeper. Keep GitHub's + // redelivery alive until the current attempt finishes or its lease can be + // reclaimed instead of acknowledging potentially unprocessed work. + throw new Error("GitHub webhook delivery is already in progress; retry later."); } if (!repoFullName) { - await args.db.externalWebhookEvent.update({ - where: { provider_deliveryId: { provider: GITHUB_PROVIDER, deliveryId: args.deliveryId } }, - data: { status: "SKIPPED", error: "Missing repository.full_name.", processedAt: new Date() }, + await finishGitHubWebhookDelivery({ + db: args.db, + deliveryId: args.deliveryId, + leaseStartedAt, + status: "SKIPPED", + error: "Missing repository.full_name.", }); return { ok: true, processed: 0, skipped: "missing-repository" }; } @@ -608,16 +990,20 @@ export async function processGitHubWebhook(args: { installationId: args.payload.installation?.id ?? null, }); if (mappings.length === 0) { - await args.db.externalWebhookEvent.update({ - where: { provider_deliveryId: { provider: GITHUB_PROVIDER, deliveryId: args.deliveryId } }, - data: { - status: "SKIPPED", - error: "No active Forge mapping matched.", - processedAt: new Date(), - }, + await finishGitHubWebhookDelivery({ + db: args.db, + deliveryId: args.deliveryId, + leaseStartedAt, + status: "SKIPPED", + error: "No active Forge mapping matched.", }); return { ok: true, processed: 0, skipped: "no-mapping" }; } + const mappedWorkspaceIds = [...new Set(mappings.map((mapping) => mapping.workspaceId))]; + // A delivery can fan into several tenants. Keep the provider-global dedupe + // row unowned in that case so deleting one workspace cannot erase replay + // protection for the others. + const eventWorkspaceId = mappedWorkspaceIds.length === 1 ? mappedWorkspaceIds[0] : null; let processed = 0; try { @@ -635,6 +1021,7 @@ export async function processGitHubWebhook(args: { mapping, payload: args.payload, actor, + deliveryId: args.deliveryId, }); } else if (args.event === "pull_request") { processed += await processPullRequestEvent({ @@ -650,7 +1037,11 @@ export async function processGitHubWebhook(args: { payload: args.payload, actor, }); - } else if (args.event === "check_suite" || args.event === "check_run") { + } else if ( + args.event === "check_suite" || + args.event === "check_run" || + args.event === "status" + ) { processed += await processCheckEvent({ db: args.db, mapping, @@ -659,24 +1050,22 @@ export async function processGitHubWebhook(args: { }); } } - await args.db.externalWebhookEvent.update({ - where: { provider_deliveryId: { provider: GITHUB_PROVIDER, deliveryId: args.deliveryId } }, - data: { - workspaceId: mappings[0]?.workspaceId ?? null, - status: processed > 0 ? "PROCESSED" : "SKIPPED", - processedAt: new Date(), - }, + await finishGitHubWebhookDelivery({ + db: args.db, + deliveryId: args.deliveryId, + leaseStartedAt, + workspaceId: eventWorkspaceId, + status: processed > 0 ? "PROCESSED" : "SKIPPED", }); return { ok: true, processed }; } catch (err) { - await args.db.externalWebhookEvent.update({ - where: { provider_deliveryId: { provider: GITHUB_PROVIDER, deliveryId: args.deliveryId } }, - data: { - workspaceId: mappings[0]?.workspaceId ?? null, - status: "FAILED", - error: err instanceof Error ? err.message.slice(0, 2000) : "Unknown error.", - processedAt: new Date(), - }, + await finishGitHubWebhookDelivery({ + db: args.db, + deliveryId: args.deliveryId, + leaseStartedAt, + workspaceId: eventWorkspaceId, + status: "FAILED", + error: err instanceof Error ? err.message.slice(0, 2000) : "Unknown error.", }); throw err; } diff --git a/src/server/worker.ts b/src/server/worker.ts index 379e6c95..5aaf52c1 100644 --- a/src/server/worker.ts +++ b/src/server/worker.ts @@ -30,6 +30,7 @@ import { purgeExpiredSessionKeys } from "@/server/services/api-key-purge"; import { sweepIdleEphemeralAgents } from "@/server/services/ephemeral-idle"; import { sweepCompletionCandidates } from "@/server/services/completion-candidate"; import { sweepGitHubStatusReconciliation } from "@/server/services/github/reconciliation"; +import { recoverGenericGitHubAttachments } from "@/server/services/github/resource-sync"; import { logger } from "@/server/logger"; import { webhookQueue, maintenanceQueue } from "@/server/queues"; @@ -325,7 +326,9 @@ export const maintenanceWorker = new Worker( return sweepCompletionCandidates(db); } case "github-reconciliation-sweep": { - return sweepGitHubStatusReconciliation(db); + const recoveredAttachments = await recoverGenericGitHubAttachments(db); + const reconciliation = await sweepGitHubStatusReconciliation(db); + return { recoveredAttachments, reconciliation }; } case "required-ack-check": { const eventId = job.data?.agentAssignedEventId as string | undefined; diff --git a/tests/unit/github-client.test.ts b/tests/unit/github-client.test.ts index cc1eb5e6..38e67ec7 100644 --- a/tests/unit/github-client.test.ts +++ b/tests/unit/github-client.test.ts @@ -7,6 +7,7 @@ vi.mock("@/server/services/github/installation-token", () => ({ import { getGitHubPullRequest, getGitHubPullRequestChecks, + getGitHubPullRequestReviewSummary, GitHubRequestError, } from "@/server/services/github/client"; @@ -41,6 +42,84 @@ describe("GitHub REST client resilience", () => { expect(checks.diagnostic).toContain("Checks permission required"); }); + it("aggregates the latest decisive review per reviewer", async () => { + vi.spyOn(globalThis, "fetch").mockResolvedValueOnce( + new Response( + JSON.stringify([ + { + id: 1, + state: "CHANGES_REQUESTED", + submitted_at: "2026-07-14T10:00:00Z", + user: { login: "alice" }, + }, + { + id: 2, + state: "APPROVED", + submitted_at: "2026-07-14T11:00:00Z", + user: { login: "alice" }, + }, + { + id: 3, + state: "APPROVED", + submitted_at: "2026-07-14T10:30:00Z", + user: { login: "bob" }, + }, + { + id: 4, + state: "COMMENTED", + submitted_at: "2026-07-14T11:30:00Z", + user: { login: "bob" }, + }, + { + id: 5, + state: "CHANGES_REQUESTED", + submitted_at: "2026-07-14T12:00:00Z", + user: { login: "carol" }, + }, + ]), + { status: 200 }, + ), + ); + + await expect( + getGitHubPullRequestReviewSummary({ + installationId: 1, + owner: "acme", + repo: "forge", + number: 42, + requestedReviewers: 1, + }), + ).resolves.toMatchObject({ + decision: "CHANGES_REQUESTED", + approvedCount: 2, + changesRequestedCount: 1, + requestedCount: 1, + reviewCount: 5, + source: "api-aggregate", + partial: false, + }); + }); + + it("reports review requested when no decisive review exists", async () => { + vi.spyOn(globalThis, "fetch").mockResolvedValueOnce( + new Response(JSON.stringify([]), { status: 200 }), + ); + await expect( + getGitHubPullRequestReviewSummary({ + installationId: 1, + owner: "acme", + repo: "forge", + number: 42, + requestedTeams: 1, + }), + ).resolves.toMatchObject({ + decision: "REVIEW_REQUESTED", + approvedCount: 0, + changesRequestedCount: 0, + requestedCount: 1, + }); + }); + it("aggregates every check-suite page and ignores empty legacy status pending", async () => { const firstPage = Array.from({ length: 100 }, () => ({ status: "completed",