From 9fe566c66fb36feb7bbcf971bfd164d8b73b8471 Mon Sep 17 00:00:00 2001 From: Kalven Schraut Date: Sat, 5 Sep 2026 00:30:48 -0500 Subject: [PATCH 1/5] feat(pull-requests): serve cached dependency context --- apps/server/src/auth/RpcAuthorization.test.ts | 6 + apps/server/src/auth/RpcAuthorization.ts | 1 + .../pullRequest/PullRequestService.test.ts | 417 ++++++++++++++++++ .../src/pullRequest/PullRequestService.ts | 299 ++++++++++++- apps/server/src/ws.ts | 6 + .../client-runtime/src/state/pullRequests.ts | 7 + packages/contracts/src/rpc.ts | 12 + 7 files changed, 744 insertions(+), 4 deletions(-) diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 25971b0c0aec..ab61281f6749 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -54,6 +54,12 @@ describe("RPC authorization scopes", () => { ); }); + it("reads dependency context under the same scope as pull request detail", () => { + expect(requiredScopeForRpcMethod(WS_METHODS.pullRequestsDependencyContext)).toBe( + requiredScopeForRpcMethod(WS_METHODS.pullRequestsDetail), + ); + }); + it("rejects unknown RPC method names", () => { for (const method of ["server.notRegistered", "toString", "constructor"]) { expect(() => requiredScopeForRpcMethod(method)).toThrow( diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index de6661f45886..12c87048593d 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -68,6 +68,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.pullRequestsListStats]: AuthOrchestrationReadScope, [WS_METHODS.pullRequestsSummary]: AuthOrchestrationReadScope, [WS_METHODS.pullRequestsDetail]: AuthOrchestrationReadScope, + [WS_METHODS.pullRequestsDependencyContext]: AuthOrchestrationReadScope, [WS_METHODS.pullRequestsActivity]: AuthOrchestrationReadScope, [WS_METHODS.pullRequestsThreadComments]: AuthOrchestrationReadScope, [WS_METHODS.pullRequestsDiffFileContents]: AuthOrchestrationReadScope, diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 0dd5a30fb396..56b16f103cfe 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -2761,6 +2761,23 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag }), ); +it.effect("publishes the first reference invalidation to an existing subscriber", () => + Effect.gen(function* () { + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; + const service = yield* makeService({ projects: [], providers: [] }); + const firstRefresh = yield* Stream.runHead(service.subscribeRefreshes).pipe( + Effect.forkChild({ startImmediately: true }), + ); + + yield* service.invalidate({ reference }); + // Ensure a lagging publication also completes the subscriber instead of leaving the test + // waiting: the reference invalidation itself must still be the first observed revision. + yield* service.refreshAfterTurn; + + assert.strictEqual(Option.getOrThrow(yield* Fiber.join(firstRefresh)), 2); + }), +); + it.effect("a mutation makes the next listing ask the host again, with no client asking", () => Effect.gen(function* () { let hostCalls = 0; @@ -2788,6 +2805,406 @@ it.effect("a mutation makes the next listing ask the host again, with no client }), ); +it.effect("coalesces bounded dependency reads and invalidates them by repository", () => + Effect.gen(function* () { + let relationshipReads = 0; + let nativeReads = 0; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const parent = { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + }; + const child = { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "migration", + }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: (input) => { + relationshipReads += 1; + assert.strictEqual(input.host, "github.com"); + assert.strictEqual(input.repository, "acme/web"); + assert.strictEqual(input.state, "open"); + assert.strictEqual(input.involvement, "all"); + assert.strictEqual(input.limit, 200); + assert.strictEqual(input.relationshipOnly, true); + return Effect.succeed({ items: [parent, child], truncated: false, continues: false }); + }, + getNativeDependencyMembership: () => { + nativeReads += 1; + return Effect.succeed({ status: "none" }); + }, + }), + ], + }); + + const [first, second] = yield* Effect.all( + [service.dependencyContext(reference), service.dependencyContext(reference)], + { concurrency: 2 }, + ); + assert.deepStrictEqual(first, second); + assert.deepStrictEqual(first.edges, [{ child: 2, parent: 1, certainty: "confirmed" }]); + assert.deepStrictEqual(first.native, { status: "none" }); + assert.strictEqual(relationshipReads, 1); + assert.strictEqual(nativeReads, 1); + + yield* service.dependencyContext({ ...reference, number: 1 }); + assert.strictEqual(relationshipReads, 1); + assert.strictEqual(nativeReads, 2); + + yield* service.invalidate({ reference }); + yield* service.dependencyContext(reference); + assert.strictEqual(relationshipReads, 2); + assert.strictEqual(nativeReads, 3); + + yield* service.update({ ...reference, title: "Renamed" }); + yield* service.dependencyContext(reference); + assert.strictEqual(relationshipReads, 3); + assert.strictEqual(nativeReads, 4); + assert.isAbove(Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)), 0); + }), +); + +it.effect("retries a failed relationship read immediately and caches the recovered context", () => + Effect.gen(function* () { + let relationshipReads = 0; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => { + relationshipReads += 1; + if (relationshipReads === 1) return Effect.fail(requestFailed); + return Effect.succeed({ + items: [ + { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + }, + { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "migration", + }, + ], + truncated: false, + continues: false, + }); + }, + }), + ], + }); + + assert.strictEqual((yield* service.dependencyContext(reference)).coverage, "unavailable"); + const recovered = yield* service.dependencyContext(reference); + assert.strictEqual(recovered.coverage, "complete"); + assert.deepStrictEqual(recovered.edges, [{ child: 2, parent: 1, certainty: "confirmed" }]); + assert.strictEqual(relationshipReads, 2); + assert.deepStrictEqual(yield* service.dependencyContext(reference), recovered); + yield* service.dependencyContext({ ...reference, number: 1 }); + assert.strictEqual(relationshipReads, 2); + }), +); + +it.effect("retries a failed focus summary and caches the recovered context", () => + Effect.gen(function* () { + let relationshipReads = 0; + let summaryReads = 0; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => { + relationshipReads += 1; + return Effect.succeed({ + items: [ + { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + }, + ], + truncated: false, + continues: false, + }); + }, + getChangeRequestSummary: () => { + summaryReads += 1; + return summaryReads === 1 + ? Effect.fail(requestFailed) + : Effect.succeed({ + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + baseBranch: "migration", + }); + }, + }), + ], + }); + + const degraded = yield* service.dependencyContext(reference); + assert.isTrue(degraded.issues.some((issue) => issue.reason === "host-unavailable")); + const recovered = yield* service.dependencyContext(reference); + assert.isFalse(recovered.issues.some((issue) => issue.reason === "host-unavailable")); + assert.isTrue(recovered.nodes.some((node) => node.ref.number === 2)); + assert.strictEqual(relationshipReads, 1); + assert.strictEqual(summaryReads, 2); + assert.deepStrictEqual(yield* service.dependencyContext(reference), recovered); + assert.strictEqual(summaryReads, 2); + }), +); + +it.effect("retries unavailable native membership and caches the recovered context", () => + Effect.gen(function* () { + let relationshipReads = 0; + let nativeReads = 0; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => { + relationshipReads += 1; + return Effect.succeed({ + items: [ + { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + }, + { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "migration", + }, + ], + truncated: false, + continues: false, + }); + }, + getNativeDependencyMembership: () => { + nativeReads += 1; + return nativeReads === 1 + ? Effect.fail(requestFailed) + : Effect.succeed({ status: "none" }); + }, + }), + ], + }); + + const degraded = yield* service.dependencyContext(reference); + assert.deepStrictEqual(degraded.edges, [{ child: 2, parent: 1, certainty: "confirmed" }]); + assert.strictEqual(degraded.coverage, "complete"); + assert.deepStrictEqual(degraded.native, { status: "unavailable" }); + const recovered = yield* service.dependencyContext(reference); + assert.deepStrictEqual(recovered.native, { status: "none" }); + assert.strictEqual(relationshipReads, 1); + assert.strictEqual(nativeReads, 2); + assert.deepStrictEqual(yield* service.dependencyContext(reference), recovered); + assert.strictEqual(nativeReads, 2); + }), +); + +it.effect("treats omitted raw relationship rows as partial coverage", () => + Effect.gen(function* () { + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const service = yield* makeService({ + projects: [ + project({ + id: "p1", + title: "web", + workspaceRoot: "/a", + repository: "acme/web", + provider: "gitlab", + }), + ], + providers: [ + fakeProvider("gitlab", { + listChangeRequests: () => + Effect.succeed({ + items: [ + { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + }, + { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "migration", + }, + ], + truncated: false, + cursorAdvance: 3, + continues: false, + }), + }), + ], + }); + + const result = yield* service.dependencyContext(reference); + assert.strictEqual(result.coverage, "partial"); + assert.deepStrictEqual(result.edges, [{ child: 2, parent: 1, certainty: "candidate" }]); + assert.isTrue(result.issues.some((issue) => issue.reason === "budget")); + }), +); + +it.effect("treats duplicate provider rows as partial relationship coverage", () => + Effect.gen(function* () { + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const parent = { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + }; + const child = { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "migration", + }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => + Effect.succeed({ + items: [parent, child, parent], + truncated: false, + continues: false, + }), + }), + ], + }); + + const result = yield* service.dependencyContext(reference); + assert.strictEqual(result.coverage, "partial"); + assert.deepStrictEqual(result.edges, [{ child: 2, parent: 1, certainty: "candidate" }]); + assert.isTrue(result.issues.some((issue) => issue.reason === "budget")); + }), +); + +it.effect("keeps a merged focus visible without linking it to a reused live branch", () => + Effect.gen(function* () { + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => + Effect.succeed({ + items: [ + { + ...changeRequest(1, "2026-07-04T00:00:00Z"), + headBranch: "reused-branch", + headRepositoryNameWithOwner: "acme/web", + }, + ], + truncated: false, + continues: false, + }), + getChangeRequestSummary: () => + Effect.succeed({ + ...changeRequest(2, "2026-07-03T00:00:00Z"), + state: "merged", + baseBranch: "reused-branch", + }), + }), + ], + }); + + const result = yield* service.dependencyContext(reference); + assert.deepStrictEqual( + result.nodes.map((node) => [node.ref.number, node.state]), + [[2, "merged"]], + ); + assert.deepStrictEqual(result.edges, []); + }), +); + +it.effect("appends ordered native members as nodes without fabricating branch edges", () => + Effect.gen(function* () { + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 2 }; + const focus = { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "migration", + }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => + Effect.succeed({ items: [focus], truncated: false, continues: false }), + getNativeDependencyMembership: () => + Effect.succeed({ + status: "present", + id: "STACK_1", + coverage: "complete", + members: [ + { + number: 1, + title: "Migration", + url: "https://github.com/acme/web/pull/1", + state: "merged", + isDraft: false, + headBranch: "migration", + headRepositoryNameWithOwner: "acme/web", + baseBranch: "main", + }, + focus, + ], + }), + }), + ], + }); + + const result = yield* service.dependencyContext(reference); + assert.deepStrictEqual(result.native, { + status: "present", + id: "STACK_1", + members: [1, 2], + coverage: "complete", + }); + assert.deepStrictEqual(result.nodes.map((node) => node.ref.number).sort(), [1, 2]); + assert.deepStrictEqual(result.edges, []); + }), +); + +it.effect("returns an unavailable context when the relationship listing fails", () => + Effect.gen(function* () { + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + listChangeRequests: () => Effect.fail(requestFailed), + }), + ], + }); + + const result = yield* service.dependencyContext({ + projectId: "p1" as ProjectId, + repository: "acme/web", + number: 2, + }); + assert.strictEqual(result.coverage, "unavailable"); + assert.deepStrictEqual(result.nodes, []); + assert.deepStrictEqual(result.issues, [{ reason: "host-unavailable" }]); + }), +); + it.effect("does not cache a failed listing", () => Effect.gen(function* () { let hostCalls = 0; diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index f6a0405ec9b2..445f267fc7a5 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -25,6 +25,7 @@ import { type PullRequestCommentInput, type PullRequestCommentUpdateInput, type PullRequestDetail, + type PullRequestDependencyContext, type PullRequestDiffFileContentsInput, type PullRequestDiffFileContentsResult, type PullRequestDiffStat, @@ -63,11 +64,17 @@ import * as SourceControlProviderRegistry from "../sourceControl/SourceControlPr import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit.ts"; import { type ProviderChangeRequest, + type ProviderChangeRequestPage, type ProviderListCursor, + type ProviderNativeDependencyMembership, type PullRequestProviderApi, PullRequestProviderError, } from "./PullRequestProvider.ts"; import { PullRequestProviderRegistry } from "./PullRequestProviderRegistry.ts"; +import { + buildPullRequestDependencyContext, + type ProviderDependencyNode, +} from "./pullRequestDependencyTopology.ts"; export interface PullRequestMergeEvent extends PullRequestRef { readonly mergedAt: string; @@ -117,6 +124,7 @@ const DIFF_CACHE_TTL = Duration.seconds(60); const COMMIT_DIFF_CACHE_TTL = Duration.minutes(10); /** Sized like the client's own stale time; a row's counts move only when somebody pushes. */ const LIST_STATS_CACHE_TTL = Duration.seconds(60); +const DEPENDENCY_CACHE_TTL = Duration.seconds(30); /** A diff can stay interactive while its next cached value is fetched off the critical path. */ const DIFF_STALE_WINDOW = Duration.minutes(10); /** How long one host's signed-in login is believed without asking its CLI again. */ @@ -129,6 +137,9 @@ const LIST_STATS_CACHE_CAPACITY = 32; const DETAIL_CACHE_CAPACITY = 128; const DIFF_CACHE_CAPACITY = 128; const VIEWER_CACHE_CAPACITY = 32; +const DEPENDENCY_CACHE_CAPACITY = 64; +const DEPENDENCY_RELATIONSHIP_LIMIT = 200; +const NATIVE_DEPENDENCY_MEMBER_LIMIT = 100; export type PullRequestError = PullRequestUnavailableError | PullRequestOperationError; @@ -153,6 +164,9 @@ export class PullRequestService extends Context.Service< readonly subscribeRefreshes: Stream.Stream; readonly refreshAfterTurn: Effect.Effect; readonly detail: (input: PullRequestRef) => Effect.Effect; + readonly dependencyContext: ( + input: PullRequestRef, + ) => Effect.Effect; readonly activity: ( input: PullRequestRef, ) => Effect.Effect; @@ -473,6 +487,14 @@ function withRateLimitBackoff( listChangeRequestStats: wrap("listChangeRequestStats", api.listChangeRequestStats), }), getChangeRequest: wrap("getChangeRequest", api.getChangeRequest), + ...(api.getNativeDependencyMembership === undefined + ? {} + : { + getNativeDependencyMembership: wrap( + "getNativeDependencyMembership", + api.getNativeDependencyMembership, + ), + }), ...(api.getChangeRequestSummary === undefined ? {} : { @@ -1296,7 +1318,13 @@ export const make = Effect.gen(function* () { ).pipe( Effect.map(([changeRequest, viewer, capabilities]): PullRequestDetail => ({ provider: project.api.kind, - capabilities, + capabilities: { + ...capabilities, + dependencies: { + branchRelationships: true, + nativeMembership: project.api.getNativeDependencyMembership !== undefined, + }, + }, projectId: project.project.id, projectTitle: project.project.title, workspaceRoot: project.project.workspaceRoot, @@ -1345,6 +1373,195 @@ export const make = Effect.gen(function* () { ), ); + type DependencyRelationshipRead = + | { readonly _tag: "Success"; readonly page: ProviderChangeRequestPage } + | { readonly _tag: "Failure"; readonly error: PullRequestProviderError }; + const dependencyRelationshipsUncached = (project: SupportedProject) => + project.api + .listChangeRequests({ + cwd: project.project.workspaceRoot, + host: project.host, + repository: project.repository, + state: "open", + involvement: "all", + // The viewer is ignored for an all-involvement read. Avoid a separate account request in + // the dependency budget just to fill a field no adapter consults in this mode. + viewer: "", + limit: DEPENDENCY_RELATIONSHIP_LIMIT, + relationshipOnly: true, + }) + .pipe( + Effect.map((page): DependencyRelationshipRead => ({ _tag: "Success", page })), + Effect.catch((error) => + Effect.succeed({ _tag: "Failure", error }), + ), + ); + // Replaced with the repository-index cache below once its epoch machinery has been created. + let dependencyRelationships: ( + project: SupportedProject, + ) => Effect.Effect = + dependencyRelationshipsUncached; + + const dependencyContextUncached: PullRequestService["Service"]["dependencyContext"] = (input) => + requireProject(input).pipe( + Effect.flatMap((project) => + Effect.gen(function* () { + const branchRead = yield* dependencyRelationships(project); + const nativeProviderRead = yield* project.api.getNativeDependencyMembership === undefined + ? Effect.succeed({ _tag: "Unsupported" as const }) + : project.api + .getNativeDependencyMembership({ + cwd: project.project.workspaceRoot, + host: project.host, + repository: project.repository, + number: input.number, + limit: NATIVE_DEPENDENCY_MEMBER_LIMIT, + }) + .pipe( + Effect.map((membership: ProviderNativeDependencyMembership) => ({ + _tag: "Success" as const, + membership, + })), + Effect.orElseSucceed(() => ({ _tag: "Failure" as const })), + ); + const nativeRead: PullRequestDependencyContext["native"] = + nativeProviderRead._tag === "Unsupported" + ? undefined + : nativeProviderRead._tag === "Failure" + ? { status: "unavailable" } + : nativeProviderRead.membership.status === "none" + ? { status: "none" } + : { + status: "present", + id: nativeProviderRead.membership.id, + members: nativeProviderRead.membership.members + .slice(0, NATIVE_DEPENDENCY_MEMBER_LIMIT) + .map((member) => member.number), + coverage: + nativeProviderRead.membership.coverage === "partial" || + nativeProviderRead.membership.members.length > + NATIVE_DEPENDENCY_MEMBER_LIMIT + ? "partial" + : "complete", + }; + + let rows: ProviderDependencyNode[] = []; + let branchComplete = false; + let branchUnavailable = false; + let budgetExhausted = false; + if (branchRead._tag === "Success") { + rows = branchRead.page.items.slice(0, DEPENDENCY_RELATIONSHIP_LIMIT); + budgetExhausted = + branchRead.page.truncated || + branchRead.page.items.length > DEPENDENCY_RELATIONSHIP_LIMIT || + new Set(branchRead.page.items.map((row) => row.number)).size !== + branchRead.page.items.length || + (branchRead.page.cursorAdvance !== undefined && + branchRead.page.cursorAdvance !== branchRead.page.items.length); + branchComplete = !budgetExhausted; + if (!rows.some((row) => row.number === input.number)) { + const nativeFocus = + nativeProviderRead._tag === "Success" && + nativeProviderRead.membership.status === "present" + ? (nativeProviderRead.membership.members.find( + (member) => member.number === input.number, + ) ?? null) + : null; + const summaryRead = project.api.getChangeRequestSummary; + const summary = + nativeFocus !== null || summaryRead === undefined + ? null + : yield* summaryRead({ + cwd: project.project.workspaceRoot, + host: project.host, + repository: project.repository, + number: input.number, + }).pipe(Effect.orElseSucceed(() => null)); + const focus: ProviderDependencyNode | null = + nativeFocus ?? + (summary === null + ? null + : { + number: summary.number, + title: summary.title, + url: summary.url, + state: summary.state, + isDraft: summary.isDraft === true, + headBranch: summary.headBranch, + headRepositoryNameWithOwner: null, + baseBranch: summary.baseBranch, + }); + if (focus === null) { + branchComplete = false; + branchUnavailable = true; + } else if (rows.length === DEPENDENCY_RELATIONSHIP_LIMIT) { + rows[rows.length - 1] = focus; + budgetExhausted = true; + branchComplete = false; + } else { + rows.push(focus); + } + } + } else { + branchUnavailable = true; + } + + const topology = buildPullRequestDependencyContext({ + projectId: project.project.id, + provider: project.api.kind, + host: project.host, + repository: project.repository, + focus: input.number, + rows, + complete: branchComplete, + }); + + const nativeMembership = + nativeProviderRead._tag === "Success" && + nativeProviderRead.membership.status === "present" + ? nativeProviderRead.membership.members.slice(0, NATIVE_DEPENDENCY_MEMBER_LIMIT) + : []; + const nodesByNumber = new Map(topology.nodes.map((node) => [node.ref.number, node])); + for (const member of nativeMembership) { + if (nodesByNumber.has(member.number)) continue; + nodesByNumber.set(member.number, { + ref: { + projectId: project.project.id, + repository: project.repository, + number: member.number, + }, + title: member.title, + url: member.url, + state: member.state, + isDraft: member.isDraft, + baseBranch: member.baseBranch, + head: + member.headRepositoryNameWithOwner == null + ? null + : { + repository: member.headRepositoryNameWithOwner, + branch: member.headBranch, + }, + }); + } + const issues = [...topology.issues]; + if (budgetExhausted && !issues.some((issue) => issue.reason === "budget")) { + issues.push({ reason: "budget" }); + } + if (branchUnavailable && !issues.some((issue) => issue.reason === "host-unavailable")) { + issues.push({ reason: "host-unavailable" }); + } + return { + ...topology, + nodes: [...nodesByNumber.values()], + coverage: branchRead._tag === "Failure" ? "unavailable" : topology.coverage, + issues, + ...(nativeRead === undefined ? {} : { native: nativeRead }), + }; + }), + ), + ); + const activityUncached: PullRequestService["Service"]["activity"] = (input) => requireProject(input).pipe( Effect.flatMap((project) => @@ -2162,7 +2379,9 @@ export const make = Effect.gen(function* () { let listingsEpoch = 0; let turnRefreshEpoch = 0; const refEpochs = new Map(); + const repositoryEpochs = new Map(); const REF_EPOCH_CAPACITY = 2_048; + const REPOSITORY_EPOCH_CAPACITY = 512; const refScope = (ref: PullRequestRef) => `${ref.projectId} ${ref.repository} ${ref.number}`; const refEpoch = (ref: PullRequestRef) => Math.max(turnRefreshEpoch, refEpochs.get(refScope(ref)) ?? 0); @@ -2176,6 +2395,18 @@ export const make = Effect.gen(function* () { } refEpochs.set(scope, ++epochCounter); }; + const repositoryScope = (ref: Pick) => + `${ref.projectId} ${ref.repository.trim().toLowerCase()}`; + const repositoryEpoch = (ref: Pick) => + Math.max(turnRefreshEpoch, listingsEpoch, repositoryEpochs.get(repositoryScope(ref)) ?? 0); + const bumpRepositoryEpoch = (ref: Pick) => { + const scope = repositoryScope(ref); + if (!repositoryEpochs.has(scope) && repositoryEpochs.size >= REPOSITORY_EPOCH_CAPACITY) { + const oldest = repositoryEpochs.keys().next().value; + if (oldest !== undefined) repositoryEpochs.delete(oldest); + } + repositoryEpochs.set(scope, ++epochCounter); + }; /** The positional filter slot of a cache key, back as the record `listUncached` takes. */ const filtersOfKey = ( @@ -2342,6 +2573,56 @@ export const make = Effect.gen(function* () { ); }; + const dependencyRelationshipCache = yield* Cache.makeWith( + (key: string) => { + const [, projectId, repository] = JSON.parse(key) as [number, string, string]; + return requireProject({ projectId, repository, number: 1 } as PullRequestRef).pipe( + Effect.flatMap(dependencyRelationshipsUncached), + ); + }, + { + capacity: DEPENDENCY_CACHE_CAPACITY, + timeToLive: (exit) => + Exit.isSuccess(exit) && exit.value._tag === "Success" + ? DEPENDENCY_CACHE_TTL + : Duration.zero, + }, + ); + dependencyRelationships = (project) => + Cache.get( + dependencyRelationshipCache, + JSON.stringify([ + repositoryEpoch({ + projectId: project.project.id, + repository: project.repository, + }), + project.project.id, + project.repository, + ]), + ); + + const dependencyCache = yield* Cache.makeWith( + (key: string) => { + const [, projectId, repository, number] = JSON.parse(key) as [number, string, string, number]; + return dependencyContextUncached({ projectId, repository, number } as PullRequestRef); + }, + { + capacity: DEPENDENCY_CACHE_CAPACITY, + timeToLive: (exit) => + Exit.isSuccess(exit) && + exit.value.coverage !== "unavailable" && + exit.value.native?.status !== "unavailable" && + !exit.value.issues.some((issue) => issue.reason === "host-unavailable") + ? DEPENDENCY_CACHE_TTL + : Duration.zero, + }, + ); + const dependencyContext: PullRequestService["Service"]["dependencyContext"] = (input) => + Cache.get( + dependencyCache, + JSON.stringify([repositoryEpoch(input), input.projectId, input.repository, input.number]), + ); + const activityCache = yield* Cache.makeWith( (key: string) => { const [, projectId, repository, number] = JSON.parse(key) as [number, string, string, number]; @@ -2428,12 +2709,18 @@ export const make = Effect.gen(function* () { const invalidate: PullRequestService["Service"]["invalidate"] = (input) => { const reference = input.reference; if (reference !== undefined) { - return Effect.sync(() => bumpRefEpoch(reference)); + return Effect.sync(() => { + bumpRefEpoch(reference); + bumpRepositoryEpoch(reference); + }).pipe(Effect.andThen(() => SubscriptionRef.set(pullRequestRefreshes, epochCounter))); } return Effect.sync(() => { listingsEpoch = ++epochCounter; viewersByHost.clear(); - }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights))); + }).pipe( + Effect.andThen(Cache.invalidateAll(viewerFlights)), + Effect.andThen(() => SubscriptionRef.set(pullRequestRefreshes, epochCounter)), + ); }; const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => { @@ -2453,8 +2740,9 @@ export const make = Effect.gen(function* () { Effect.tap(() => Effect.sync(() => { bumpRefEpoch(input); + bumpRepositoryEpoch(input); listingsEpoch = ++epochCounter; - }), + }).pipe(Effect.andThen(() => SubscriptionRef.set(pullRequestRefreshes, epochCounter))), ), ); const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn( @@ -2462,7 +2750,9 @@ export const make = Effect.gen(function* () { )(function* (input) { const repository = yield* runAction(input); bumpRefEpoch({ ...input, repository }); + bumpRepositoryEpoch({ ...input, repository }); listingsEpoch = ++epochCounter; + yield* SubscriptionRef.set(pullRequestRefreshes, epochCounter); if (input.action === "merge") { // A successful merge action can merely enqueue the PR or enable auto-merge. const confirmed = yield* summaryUncached({ ...input, repository }).pipe( @@ -2494,6 +2784,7 @@ export const make = Effect.gen(function* () { ), refreshAfterTurn, detail, + dependencyContext, activity, threadComments, diff, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index a25065c452db..ce8e6c3f20df 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2092,6 +2092,12 @@ const makeWsRpcLayer = ( observeRpcEffect(WS_METHODS.pullRequestsDetail, pullRequests.detail(input), { "rpc.aggregate": "pull-requests", }), + [WS_METHODS.pullRequestsDependencyContext]: (input) => + observeRpcEffect( + WS_METHODS.pullRequestsDependencyContext, + pullRequests.dependencyContext(input), + { "rpc.aggregate": "pull-requests" }, + ), [WS_METHODS.pullRequestsActivity]: (input) => observeRpcEffect(WS_METHODS.pullRequestsActivity, pullRequests.activity(input), { "rpc.aggregate": "pull-requests", diff --git a/packages/client-runtime/src/state/pullRequests.ts b/packages/client-runtime/src/state/pullRequests.ts index 6144d8507a9e..2cadc196bda5 100644 --- a/packages/client-runtime/src/state/pullRequests.ts +++ b/packages/client-runtime/src/state/pullRequests.ts @@ -121,6 +121,13 @@ export function createPullRequestEnvironmentAtoms( staleTimeMs: 15_000, refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), }), + /** One bounded repository relationship read for the open PR panel, never for list rows. */ + dependencyContext: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:pull-requests:dependency-context", + tag: WS_METHODS.pullRequestsDependencyContext, + staleTimeMs: 30_000, + refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), + }), activity, threadComments: createEnvironmentRpcCommand(runtime, { label: "environment-data:pull-requests:thread-comments", diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index f7f2c2b6faa7..5aeb20c5d782 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -96,6 +96,7 @@ import { PullRequestCommentInput, PullRequestCommentUpdateInput, PullRequestDetail, + PullRequestDependencyContext, PullRequestDiffFileContentsInput, PullRequestDiffFileContentsResult, PullRequestInvalidateInput, @@ -331,6 +332,7 @@ export const WS_METHODS = { pullRequestsListStats: "pullRequests.listStats", pullRequestsSummary: "pullRequests.summary", pullRequestsDetail: "pullRequests.detail", + pullRequestsDependencyContext: "pullRequests.dependencyContext", pullRequestsActivity: "pullRequests.activity", pullRequestsThreadComments: "pullRequests.threadComments", pullRequestsDiffFileContents: "pullRequests.diffFileContents", @@ -640,6 +642,15 @@ export const WsPullRequestsDetailRpc = Rpc.make(WS_METHODS.pullRequestsDetail, { error: PullRequestRpcError, }); +export const WsPullRequestsDependencyContextRpc = Rpc.make( + WS_METHODS.pullRequestsDependencyContext, + { + payload: PullRequestRef, + success: PullRequestDependencyContext, + error: PullRequestRpcError, + }, +); + export const WsPullRequestsActivityRpc = Rpc.make(WS_METHODS.pullRequestsActivity, { payload: PullRequestRef, success: PullRequestActivity, @@ -1214,6 +1225,7 @@ export const WsRpcGroup = RpcGroup.make( WsPullRequestsListStatsRpc, WsPullRequestsSummaryRpc, WsPullRequestsDetailRpc, + WsPullRequestsDependencyContextRpc, WsPullRequestsActivityRpc, WsPullRequestsThreadCommentsRpc, WsPullRequestsDiffFileContentsRpc, From e48f6afaa2cfcab1ea98e0623bb86f210c505884 Mon Sep 17 00:00:00 2001 From: Kalven Schraut Date: Mon, 7 Sep 2026 22:10:45 -0500 Subject: [PATCH 2/5] fix(pull-requests): scope refreshes to affected repositories --- .../pullRequest/PullRequestService.test.ts | 108 ++++++++- .../src/pullRequest/PullRequestService.ts | 128 +++++++--- apps/server/src/ws.ts | 7 +- .../pullRequest/PullRequestDetailPanel.tsx | 2 +- apps/web/src/routes/_chat.pull-requests.tsx | 8 +- apps/web/src/state/pullRequests.ts | 32 ++- .../src/state/pullRequests.test.ts | 222 +++++++++++++++--- .../client-runtime/src/state/pullRequests.ts | 103 +++++++- packages/contracts/src/pullRequest.ts | 10 + packages/contracts/src/rpc.ts | 5 +- 10 files changed, 533 insertions(+), 92 deletions(-) diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 56b16f103cfe..271d17302732 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -1995,6 +1995,105 @@ it.effect("invalidates the cached activity after reacting, like the other mutati }), ); +it.effect("a reaction preserves unrelated repository reads and every cached listing", () => + Effect.gen(function* () { + const reads: string[] = []; + const activity = { + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; + const unrelated = { projectId: "p2" as ProjectId, repository: "acme/api", number: 1 }; + const service = yield* makeService({ + projects: [ + project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }), + project({ id: "p2", title: "api", workspaceRoot: "/b", repository: "acme/api" }), + ], + providers: [ + fakeProvider("github", { + listChangeRequests: (input) => + Effect.sync(() => { + reads.push(`list:${input.repository}`); + return { items: [], truncated: false, continues: false }; + }), + getChangeRequestActivity: (input) => + Effect.sync(() => { + reads.push(`activity:${input.repository}`); + return activity; + }), + }), + ], + }); + yield* service.list({ state: "open" }); + yield* service.list({ state: "open", projectId: unrelated.projectId }); + yield* service.activity(reference); + yield* service.activity(unrelated); + const before = [...reads]; + + yield* service.setReaction({ ...reference, content: "heart", reacted: true }); + yield* service.list({ state: "open" }); + yield* service.list({ state: "open", projectId: unrelated.projectId }); + yield* service.activity(unrelated); + assert.deepStrictEqual(reads, before); + yield* service.activity(reference); + assert.deepStrictEqual(reads, [...before, "activity:acme/web"]); + const event = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)); + assert.deepStrictEqual(event.reference, reference); + assert.isFalse(event.listings); + }), +); + +it.effect( + "state changes invalidate sibling checkouts but preserve other repositories and hosts", + () => + Effect.gen(function* () { + const reads: string[] = []; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; + const service = yield* makeService({ + projects: [ + project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }), + project({ id: "sibling", title: "sibling", workspaceRoot: "/b", repository: "Acme/Web" }), + project({ id: "other", title: "other", workspaceRoot: "/c", repository: "acme/api" }), + project({ + id: "enterprise", + title: "enterprise", + workspaceRoot: "/d", + repository: "acme/web", + host: "github.example.com", + }), + ], + providers: [ + fakeProvider("github", { + listChangeRequests: (input) => + Effect.sync(() => { + reads.push(`${input.host}:${input.repository}`); + return { items: [], truncated: false, continues: false }; + }), + }), + ], + }); + const sibling = { state: "open", projectId: "sibling" as ProjectId } as const; + const other = { state: "open", projectIds: ["other" as ProjectId] } as const; + const enterprise = { state: "open", host: "github.example.com" } as const; + yield* service.list(sibling); + yield* service.list(other); + yield* service.list(enterprise); + const before = [...reads]; + yield* service.runAction({ ...reference, action: "close" }); + yield* service.list(other); + yield* service.list(enterprise); + assert.deepStrictEqual(reads, before); + yield* service.list(sibling); + assert.deepStrictEqual(reads, [...before, "github.com:Acme/Web"]); + const event = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)); + assert.deepStrictEqual(event.projectIds, ["p1" as ProjectId, "sibling" as ProjectId]); + assert.isTrue(event.listings); + }), +); + it.effect("refuses an empty reply before it reaches the host", () => Effect.gen(function* () { const service = yield* makeService({ @@ -2756,7 +2855,7 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag yield* service.refreshAfterTurn; const refresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)); yield* service.list({ state: "open" }); - assert.isAbove(refresh, 0); + assert.isAbove(refresh.revision, 0); assert.strictEqual(hostCalls, 3); }), ); @@ -2774,7 +2873,7 @@ it.effect("publishes the first reference invalidation to an existing subscriber" // waiting: the reference invalidation itself must still be the first observed revision. yield* service.refreshAfterTurn; - assert.strictEqual(Option.getOrThrow(yield* Fiber.join(firstRefresh)), 2); + assert.strictEqual(Option.getOrThrow(yield* Fiber.join(firstRefresh)).revision, 2); }), ); @@ -2866,7 +2965,10 @@ it.effect("coalesces bounded dependency reads and invalidates them by repository yield* service.dependencyContext(reference); assert.strictEqual(relationshipReads, 3); assert.strictEqual(nativeReads, 4); - assert.isAbove(Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)), 0); + assert.isAbove( + Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)).revision, + 0, + ); }), ); diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 445f267fc7a5..5e941ca60faa 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -42,6 +42,7 @@ import { type PullRequestProviderSummary, type PullRequestReactionInput, type PullRequestRef, + type PullRequestRefresh, type PullRequestReviewVerdict, type PullRequestReviewerCandidateList, type PullRequestReviewerRequestInput, @@ -55,7 +56,7 @@ import { type PullRequestThreadCommentsResult, type PullRequestUpdateInput, type SourceControlProviderInfo, - type SourceControlProviderKind, + SourceControlProviderKind, } from "@t3tools/contracts"; import { detectSourceControlProviderFromRemoteUrl } from "@t3tools/shared/sourceControl"; @@ -143,6 +144,8 @@ const NATIVE_DEPENDENCY_MEMBER_LIMIT = 100; export type PullRequestError = PullRequestUnavailableError | PullRequestOperationError; +const isSourceControlProviderKind = Schema.is(SourceControlProviderKind); + export class PullRequestService extends Context.Service< PullRequestService, { @@ -161,7 +164,7 @@ export class PullRequestService extends Context.Service< never, Scope.Scope >; - readonly subscribeRefreshes: Stream.Stream; + readonly subscribeRefreshes: Stream.Stream; readonly refreshAfterTurn: Effect.Effect; readonly detail: (input: PullRequestRef) => Effect.Effect; readonly dependencyContext: ( @@ -560,7 +563,10 @@ export function repositoryIdentityOf(project: OrchestrationProjectShell): string export const make = Effect.gen(function* () { const mergedPullRequests = yield* PubSub.sliding(64); - const pullRequestRefreshes = yield* SubscriptionRef.make(0); + const pullRequestRefreshes = yield* SubscriptionRef.make({ + revision: 0, + listings: true, + }); const registry = yield* PullRequestProviderRegistry; const projections = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const sourceControlProviders = yield* SourceControlProviderRegistry.SourceControlProviderRegistry; @@ -2378,11 +2384,13 @@ export const make = Effect.gen(function* () { let epochCounter = 0; let listingsEpoch = 0; let turnRefreshEpoch = 0; + const projectListingEpochs = new Map(); const refEpochs = new Map(); const repositoryEpochs = new Map(); const REF_EPOCH_CAPACITY = 2_048; const REPOSITORY_EPOCH_CAPACITY = 512; - const refScope = (ref: PullRequestRef) => `${ref.projectId} ${ref.repository} ${ref.number}`; + const refScope = (ref: PullRequestRef) => + `${ref.projectId} ${ref.repository.trim().toLowerCase()} ${ref.number}`; const refEpoch = (ref: PullRequestRef) => Math.max(turnRefreshEpoch, refEpochs.get(refScope(ref)) ?? 0); const refCacheKey = (ref: PullRequestRef) => @@ -2408,6 +2416,82 @@ export const make = Effect.gen(function* () { repositoryEpochs.set(scope, ++epochCounter); }; + const listingEpoch = (projectIds?: ReadonlyArray, host?: string) => { + let epoch = listingsEpoch; + for (const [projectId, entry] of projectListingEpochs) { + if ( + (projectIds === undefined || projectIds.includes(projectId)) && + (host === undefined || entry.host === undefined || entry.host === host.toLowerCase()) + ) + epoch = Math.max(epoch, entry.revision); + } + return epoch; + }; + const bumpListingEpoch = (ref: PullRequestRef, host: string | undefined) => { + if ( + !projectListingEpochs.has(ref.projectId) && + projectListingEpochs.size >= REPOSITORY_EPOCH_CAPACITY + ) { + listingsEpoch = ++epochCounter; + projectListingEpochs.clear(); + } + projectListingEpochs.set(ref.projectId, { revision: ++epochCounter, host }); + }; + const publishRefresh = ( + reference?: PullRequestRef, + listings = true, + projectIds?: ReadonlyArray, + host?: string, + ) => + SubscriptionRef.set(pullRequestRefreshes, { + revision: epochCounter, + ...(reference === undefined + ? {} + : { + reference: { + projectId: reference.projectId, + repository: reference.repository, + number: reference.number, + }, + }), + ...(projectIds === undefined ? {} : { projectIds }), + ...(host === undefined ? {} : { host }), + listings, + }); + + // Sibling checkouts share host data even when a client addresses them through another project. + const invalidateReference = Effect.fn("PullRequestService.invalidateReference")(function* ( + reference: PullRequestRef, + listings: boolean, + ) { + const snapshot = yield* projections.getShellSnapshot().pipe(Effect.orElseSucceed(() => null)); + const source = snapshot?.projects.find((project) => project.id === reference.projectId); + const identity = source?.repositoryIdentity; + const provider = identity?.provider; + const kind = isSourceControlProviderKind(provider) ? provider : "unknown"; + const host = identity == null ? undefined : pullRequestHostOf(identity, kind); + const projectIds = new Set([reference.projectId]); + if (identity !== undefined && identity !== null) { + for (const project of snapshot?.projects ?? []) { + if ( + project.repositoryIdentity?.provider === identity.provider && + pullRequestHostOf(project.repositoryIdentity, kind) === host && + repositoryIdentityOf(project)?.trim().toLowerCase() === + reference.repository.trim().toLowerCase() + ) { + projectIds.add(project.id); + } + } + } + for (const projectId of projectIds) { + const ref = { ...reference, projectId }; + bumpRefEpoch(ref); + bumpRepositoryEpoch(ref); + if (listings) bumpListingEpoch(ref, host); + } + yield* publishRefresh(reference, listings, [...projectIds], host); + }); + /** The positional filter slot of a cache key, back as the record `listUncached` takes. */ const filtersOfKey = ( slots: ReadonlyArray< @@ -2497,7 +2581,10 @@ export const make = Effect.gen(function* () { ); const list: PullRequestService["Service"]["list"] = (input) => { const key = JSON.stringify([ - listingsEpoch, + listingEpoch( + input.projectId === undefined ? input.projectIds : [input.projectId], + input.host, + ), input.state, input.involvement ?? null, // Positional so two identical filter sets key alike however their record was assembled. @@ -2696,7 +2783,7 @@ export const make = Effect.gen(function* () { const listStats: PullRequestService["Service"]["listStats"] = (input) => { if (input.refs.length === 0) return Effect.succeed({ stats: [] }); const key = JSON.stringify([ - listingsEpoch, + listingEpoch(input.refs.map((ref) => ref.projectId)), input.refs .map((ref) => [ref.projectId, ref.repository, ref.number] as const) .toSorted((left, right) => @@ -2709,23 +2796,20 @@ export const make = Effect.gen(function* () { const invalidate: PullRequestService["Service"]["invalidate"] = (input) => { const reference = input.reference; if (reference !== undefined) { - return Effect.sync(() => { - bumpRefEpoch(reference); - bumpRepositoryEpoch(reference); - }).pipe(Effect.andThen(() => SubscriptionRef.set(pullRequestRefreshes, epochCounter))); + return invalidateReference(reference, false); } return Effect.sync(() => { listingsEpoch = ++epochCounter; viewersByHost.clear(); }).pipe( Effect.andThen(Cache.invalidateAll(viewerFlights)), - Effect.andThen(() => SubscriptionRef.set(pullRequestRefreshes, epochCounter)), + Effect.andThen(() => publishRefresh()), ); }; const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => { turnRefreshEpoch = listingsEpoch = ++epochCounter; - return SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch); + return publishRefresh(); }); // A mutation's own client re-reads right after it, and every other client's next read must @@ -2734,25 +2818,15 @@ export const make = Effect.gen(function* () { const invalidatedByMutation = ( method: (input: I) => Effect.Effect, + listings = true, ): ((input: I) => Effect.Effect) => (input) => - method(input).pipe( - Effect.tap(() => - Effect.sync(() => { - bumpRefEpoch(input); - bumpRepositoryEpoch(input); - listingsEpoch = ++epochCounter; - }).pipe(Effect.andThen(() => SubscriptionRef.set(pullRequestRefreshes, epochCounter))), - ), - ); + method(input).pipe(Effect.tap(() => invalidateReference(input, listings))); const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn( "PullRequestService.runActionAndInvalidate", )(function* (input) { const repository = yield* runAction(input); - bumpRefEpoch({ ...input, repository }); - bumpRepositoryEpoch({ ...input, repository }); - listingsEpoch = ++epochCounter; - yield* SubscriptionRef.set(pullRequestRefreshes, epochCounter); + yield* invalidateReference({ ...input, repository }, true); if (input.action === "merge") { // A successful merge action can merely enqueue the PR or enable auto-merge. const confirmed = yield* summaryUncached({ ...input, repository }).pipe( @@ -2780,7 +2854,7 @@ export const make = Effect.gen(function* () { Effect.map((subscription) => Stream.fromSubscription(subscription)), ), subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe( - Stream.filter((revision) => revision > 0), + Stream.filter((event) => event.revision > 0), ), refreshAfterTurn, detail, @@ -2796,7 +2870,7 @@ export const make = Effect.gen(function* () { submitReview: invalidatedByMutation(submitReview), replyToThread: invalidatedByMutation(replyToThread), setThreadResolution: invalidatedByMutation(setThreadResolution), - setReaction: invalidatedByMutation(setReaction), + setReaction: invalidatedByMutation(setReaction, false), // The candidate list is deliberately read fresh per menu-open, so it stays uncached. reviewerCandidates, requestReviewers: invalidatedByMutation(requestReviewers), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index ce8e6c3f20df..5517f10d07d9 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2160,10 +2160,13 @@ const makeWsRpcLayer = ( observeRpcEffect(WS_METHODS.pullRequestsInvalidate, pullRequests.invalidate(input), { "rpc.aggregate": "pull-requests", }), - [WS_METHODS.pullRequestsSubscribeRefreshes]: () => + [WS_METHODS.pullRequestsSubscribeRefreshes]: (input) => observeRpcStream( WS_METHODS.pullRequestsSubscribeRefreshes, - pullRequests.subscribeRefreshes, + pullRequests.subscribeRefreshes.pipe( + Stream.filter((event) => input.scoped || event.reference === undefined), + Stream.map((event) => (input.scoped ? event : event.revision)), + ), { "rpc.aggregate": "pull-requests" }, ), [WS_METHODS.pullRequestsReviewerCandidates]: (input) => diff --git a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx index 3ed0c61139d3..5c320c22a63f 100644 --- a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx +++ b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx @@ -584,7 +584,7 @@ export function PullRequestDetailPanel({ const activityQuery = useEnvironmentQuery( pullRequestEnvironment.activity({ environmentId, input: reference }), ); - const turnRefresh = usePullRequestTurnRefresh(environmentId); + const turnRefresh = usePullRequestTurnRefresh(environmentId, reference); const [cachedDetail, setCachedDetail] = useState(() => readPullRequestDetailSnapshot( typeof window === "undefined" ? undefined : window.localStorage, diff --git a/apps/web/src/routes/_chat.pull-requests.tsx b/apps/web/src/routes/_chat.pull-requests.tsx index bcc22daa4a9d..913c59342f7d 100644 --- a/apps/web/src/routes/_chat.pull-requests.tsx +++ b/apps/web/src/routes/_chat.pull-requests.tsx @@ -634,7 +634,13 @@ function PullRequestsRouteView() { [environmentQueries], ); const turnRefreshes = usePullRequestTurnRefreshes( - environmentQueries.map(({ environmentId }) => environmentId), + environmentQueries.map(({ environmentId, projectIds }) => ({ + environmentId, + input: { + ...(projectIds === undefined ? {} : { projectIds }), + ...(search.host === undefined ? {} : { host: search.host }), + }, + })), ); const turnRefreshToken = turnRefreshes .map(([environmentId, revision]) => `${environmentId}:${revision}`) diff --git a/apps/web/src/state/pullRequests.ts b/apps/web/src/state/pullRequests.ts index 99b2ef37c99a..6894efd46c93 100644 --- a/apps/web/src/state/pullRequests.ts +++ b/apps/web/src/state/pullRequests.ts @@ -150,22 +150,38 @@ const usePullRequestStatsQuery = createMergedEnvironmentQuery( pullRequestEnvironment.listStats, ); +type PullRequestListRefreshScope = Pick; + const usePullRequestTurnRefreshQuery = createMergedEnvironmentQuery( "web-pull-requests:turn-refreshes", - ({ environmentId }: EnvironmentQueryTarget>>) => - pullRequestEnvironment.refreshes({ environmentId, input: {} }), + ({ environmentId, input }: EnvironmentQueryTarget) => + pullRequestEnvironment.refreshes({ + environmentId, + input: { + kind: "list", + ...(input.host === undefined ? {} : { host: input.host }), + ...(input.projectId !== undefined + ? { projectIds: [input.projectId] } + : input.projectIds === undefined + ? {} + : { projectIds: input.projectIds }), + }, + }), ); export function usePullRequestTurnRefreshes( - environmentIds: ReadonlyArray, + targets: ReadonlyArray>, ): ReadonlyArray { - return usePullRequestTurnRefreshQuery( - environmentIds.map((environmentId) => ({ environmentId, input: {} })), - ).values; + return usePullRequestTurnRefreshQuery(targets).values; } -export function usePullRequestTurnRefresh(environmentId: EnvironmentId): number | null { - const result = useAtomValue(pullRequestEnvironment.refreshes({ environmentId, input: {} })); +export function usePullRequestTurnRefresh( + environmentId: EnvironmentId, + reference: PullRequestRef, +): number | null { + const result = useAtomValue( + pullRequestEnvironment.refreshes({ environmentId, input: { kind: "reference", reference } }), + ); return Option.getOrNull(AsyncResult.value(result)); } diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 6a22b22f1bda..550ff891897e 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -1,4 +1,10 @@ -import { EnvironmentId, ProjectId, WS_METHODS } from "@t3tools/contracts"; +import { + EnvironmentId, + ProjectId, + WS_METHODS, + type PullRequestRefresh, + type PullRequestRef, +} from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Latch from "effect/Latch"; @@ -41,6 +47,48 @@ function session(client: WsRpcProtocolClient): RpcSession { }; } +const makeAtoms = Effect.fn("makeAtoms")(function* (client: WsRpcProtocolClient) { + const connectionState: SupervisorConnectionState = { + ...AVAILABLE_CONNECTION_STATE, + desired: true, + network: "online", + phase: "connected", + attempt: 1, + generation: 1, + }; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(connectionState), + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ + run: (_environmentId, effect) => + Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + runStream: (_environmentId, stream) => + Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + followStream: (_environmentId, stream) => + Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + } as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const runtime = Atom.runtime( + Layer.merge( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), + Layer.succeed( + PullRequestDiffLoader, + PullRequestDiffLoader.of({ load: () => Effect.die("unused") }), + ), + ), + ); + const atoms = createPullRequestEnvironmentAtoms(runtime); + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => + Effect.sync(() => registry.dispose()), + ); + return { atoms, registry }; +}); + it.effect("refreshes pull request activity after a comment is updated", () => Effect.scoped( Effect.gen(function* () { @@ -76,44 +124,7 @@ it.effect("refreshes pull request activity after a comment is updated", () => commentBody = input.body; }), } as unknown as WsRpcProtocolClient; - const connectionState: SupervisorConnectionState = { - ...AVAILABLE_CONNECTION_STATE, - desired: true, - network: "online", - phase: "connected", - attempt: 1, - generation: 1, - }; - const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ - target: TARGET, - state: yield* SubscriptionRef.make(connectionState), - session: yield* SubscriptionRef.make(Option.some(session(client))), - prepared: yield* SubscriptionRef.make(Option.none()), - connect: Effect.void, - disconnect: Effect.void, - retryNow: Effect.void, - } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); - const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ - run: (_environmentId, effect) => - Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - runStream: (_environmentId, stream) => - Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - followStream: (_environmentId, stream) => - Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - } as EnvironmentRegistry.EnvironmentRegistry["Service"]); - const runtime = Atom.runtime( - Layer.merge( - Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), - Layer.succeed( - PullRequestDiffLoader, - PullRequestDiffLoader.of({ load: () => Effect.die("unused") }), - ), - ), - ); - const atoms = createPullRequestEnvironmentAtoms(runtime); - const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => - Effect.sync(() => registry.dispose()), - ); + const { atoms, registry } = yield* makeAtoms(client); const reference = { projectId: ProjectId.make("project-1"), repository: "acme/web", @@ -161,3 +172,136 @@ it.effect("refreshes pull request activity after a comment is updated", () => }), ), ); + +it.effect( + "scoped refreshes update matching clients without re-fetching unrelated active repositories", + () => + Effect.scoped( + Effect.gen(function* () { + const refreshEvents = yield* PubSub.unbounded(); + const subscribed = Latch.makeUnsafe(); + let revision = 0; + const activityReads = new Map(); + const listReads = new Map(); + const reference = { + projectId: ProjectId.make("project-1"), + repository: "acme/web", + number: 1, + }; + const sibling = { + ...reference, + projectId: ProjectId.make("sibling"), + repository: "Acme/Web", + }; + const unrelated = { + ...reference, + projectId: ProjectId.make("project-2"), + repository: "acme/api", + }; + const client = { + [WS_METHODS.pullRequestsSubscribeRefreshes]: () => + Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(refreshEvents); + subscribed.openUnsafe(); + return Stream.fromSubscription(subscription); + }), + ), + [WS_METHODS.pullRequestsActivity]: (input: PullRequestRef) => + Effect.sync(() => { + activityReads.set(input.projectId, (activityReads.get(input.projectId) ?? 0) + 1); + return { + author: null, + reviewers: [], + comments: [], + commentCount: revision, + commentsTruncated: false, + reviewThreads: [], + commits: [], + reactions: [], + }; + }), + [WS_METHODS.pullRequestsList]: (input: { projectId: string }) => + Effect.sync(() => { + listReads.set(input.projectId, (listReads.get(input.projectId) ?? 0) + 1); + return { + entries: [], + providers: [], + viewers: {}, + errors: [], + truncated: false, + nextCursors: {}, + }; + }), + } as unknown as WsRpcProtocolClient; + const { atoms, registry } = yield* makeAtoms(client); + const activityAtoms = [reference, sibling, unrelated].map((input) => + atoms.activity({ environmentId: TARGET.environmentId, input }), + ); + const listAtoms = [reference, unrelated].map((input) => + atoms.list({ + environmentId: TARGET.environmentId, + input: { state: "open", projectId: input.projectId }, + }), + ); + for (const atom of [...activityAtoms, ...listAtoms]) { + const unmount = registry.mount(atom); + yield* Effect.addFinalizer(() => Effect.sync(unmount)); + } + for (const atom of activityAtoms) + yield* AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }); + for (const atom of listAtoms) + yield* AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }); + yield* subscribed.await; + const beforeActivity = new Map(activityReads); + const beforeLists = new Map(listReads); + const updated = [Latch.makeUnsafe(), Latch.makeUnsafe()]; + for (const [index, atom] of activityAtoms.slice(0, 2).entries()) { + const stop = registry.subscribe(atom, (result) => { + if (AsyncResult.isSuccess(result) && result.value.commentCount === 1) + updated[index]!.openUnsafe(); + }); + yield* Effect.addFinalizer(() => Effect.sync(stop)); + } + revision = 1; + yield* PubSub.publish(refreshEvents, { + revision, + reference, + projectIds: [reference.projectId, sibling.projectId], + listings: false, + }); + for (const latch of updated) yield* latch.await; + expect(activityReads.get(reference.projectId)).toBe( + beforeActivity.get(reference.projectId)! + 1, + ); + expect(activityReads.get(sibling.projectId)).toBe( + beforeActivity.get(sibling.projectId)! + 1, + ); + expect(activityReads.get(unrelated.projectId)).toBe( + beforeActivity.get(unrelated.projectId), + ); + expect(listReads).toEqual(beforeLists); + + const listed = Latch.makeUnsafe(); + const stop = registry.subscribe(listAtoms[0]!, (result) => { + if ( + AsyncResult.isSuccess(result) && + listReads.get(reference.projectId) === beforeLists.get(reference.projectId)! + 1 + ) + listed.openUnsafe(); + }); + yield* Effect.addFinalizer(() => Effect.sync(stop)); + yield* PubSub.publish(refreshEvents, { + revision: 2, + reference, + projectIds: [reference.projectId, sibling.projectId], + listings: true, + }); + yield* listed.await; + expect(listReads.get(unrelated.projectId)).toBe(beforeLists.get(unrelated.projectId)); + expect(activityReads.get(unrelated.projectId)).toBe( + beforeActivity.get(unrelated.projectId), + ); + }), + ), +); diff --git a/packages/client-runtime/src/state/pullRequests.ts b/packages/client-runtime/src/state/pullRequests.ts index 2cadc196bda5..c574f03c4a74 100644 --- a/packages/client-runtime/src/state/pullRequests.ts +++ b/packages/client-runtime/src/state/pullRequests.ts @@ -3,13 +3,17 @@ import { type PullRequestDetail, type PullRequestDiffInput, type PullRequestSummary, + type PullRequestRef, + type PullRequestRefresh, + type EnvironmentId, + type ProjectId, type VcsStatusResult, } from "@t3tools/contracts"; import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as SubscriptionRef from "effect/SubscriptionRef"; -import { Atom } from "effect/unstable/reactivity"; +import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { createAtomCommandScheduler, @@ -35,13 +39,73 @@ export class EnvironmentHttpConnectionNotReadyError extends Data.TaggedError( export const LINKED_PULL_REQUEST_IDLE_TTL_MS = 5_000; +type PullRequestRefreshScope = + | { readonly kind?: undefined } + | { readonly kind: "reference" | "repository"; readonly reference: PullRequestRef } + | { + readonly kind: "list"; + readonly projectIds?: ReadonlyArray; + readonly host?: string; + }; + +function matchesRefresh(scope: PullRequestRefreshScope, event: PullRequestRefresh): boolean { + if (event.reference === undefined) return true; + if (scope.kind === undefined) return false; + const projectIds = event.projectIds ?? [event.reference.projectId]; + if (scope.kind === "list") { + return ( + event.listings && + (scope.host === undefined || + event.host === undefined || + scope.host.toLowerCase() === event.host) && + (scope.projectIds === undefined || + scope.projectIds.some((projectId) => projectIds.includes(projectId))) + ); + } + return ( + projectIds.includes(scope.reference.projectId) && + scope.reference.repository.trim().toLowerCase() === + event.reference.repository.trim().toLowerCase() && + (scope.kind === "repository" || scope.reference.number === event.reference.number) + ); +} + function createPullRequestRefreshAtomFamily( runtime: Atom.AtomRuntime, ) { - return createEnvironmentRpcSubscriptionAtomFamily(runtime, { - label: "environment-data:pull-requests:turn-refreshes", + const events = createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:pull-requests:refreshes", tag: WS_METHODS.pullRequestsSubscribeRefreshes, }); + const scoped = Atom.family((key: string) => { + const { environmentId, input } = JSON.parse(key) as { + environmentId: EnvironmentId; + input: PullRequestRefreshScope; + }; + return Atom.make((get): AsyncResult.AsyncResult => { + const result = get(events({ environmentId, input: { scoped: true } })); + // Numeric revisions are from servers predating scoped refreshes. + if (AsyncResult.isSuccess(result)) { + const event = + typeof result.value === "number" + ? { revision: result.value, listings: true } + : result.value; + if (matchesRefresh(input, event)) { + const previous = Option.getOrUndefined(get.self()); + return previous !== undefined && + AsyncResult.isSuccess(previous) && + previous.value === event.revision + ? previous + : AsyncResult.success(event.revision); + } + } + return Option.getOrElse(get.self(), () => AsyncResult.initial()); + }); + }); + return (target: { + readonly environmentId: EnvironmentId; + readonly input: PullRequestRefreshScope; + }) => scoped(JSON.stringify(target)); } /** Refresh only the live fields a linked thread renders. */ @@ -55,7 +119,8 @@ export function createLinkedPullRequestSummaryAtomFamily( staleTimeMs: 60_000, refreshIntervalMs: 60_000, idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS, - refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), + refreshTrigger: ({ environmentId, input }) => + refreshes({ environmentId, input: { kind: "reference", reference: input } }), }); } @@ -92,7 +157,8 @@ export function createPullRequestEnvironmentAtoms( label: "environment-data:pull-requests:activity", tag: WS_METHODS.pullRequestsActivity, staleTimeMs: 15_000, - refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), + refreshTrigger: ({ environmentId, input }) => + refreshes({ environmentId, input: { kind: "reference", reference: input } }), }); return { refreshes, @@ -101,7 +167,20 @@ export function createPullRequestEnvironmentAtoms( tag: WS_METHODS.pullRequestsList, staleTimeMs: 30_000, refreshTrigger: ({ environmentId, input }) => - input.cursors === undefined ? refreshes({ environmentId, input: {} }) : undefined, + input.cursors === undefined + ? refreshes({ + environmentId, + input: { + kind: "list", + ...(input.host === undefined ? {} : { host: input.host }), + ...(input.projectId !== undefined + ? { projectIds: [input.projectId] } + : input.projectIds !== undefined + ? { projectIds: input.projectIds } + : {}), + }, + }) + : undefined, }), /** * The line counts for rows the listing has already handed over. Its own query because the @@ -113,20 +192,26 @@ export function createPullRequestEnvironmentAtoms( label: "environment-data:pull-requests:list-stats", tag: WS_METHODS.pullRequestsListStats, staleTimeMs: 60_000, - refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), + refreshTrigger: ({ environmentId, input }) => + refreshes({ + environmentId, + input: { kind: "list", projectIds: input.refs.map((ref) => ref.projectId) }, + }), }), detail: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:pull-requests:detail", tag: WS_METHODS.pullRequestsDetail, staleTimeMs: 15_000, - refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), + refreshTrigger: ({ environmentId, input }) => + refreshes({ environmentId, input: { kind: "reference", reference: input } }), }), /** One bounded repository relationship read for the open PR panel, never for list rows. */ dependencyContext: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:pull-requests:dependency-context", tag: WS_METHODS.pullRequestsDependencyContext, staleTimeMs: 30_000, - refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), + refreshTrigger: ({ environmentId, input }) => + refreshes({ environmentId, input: { kind: "repository", reference: input } }), }), activity, threadComments: createEnvironmentRpcCommand(runtime, { diff --git a/packages/contracts/src/pullRequest.ts b/packages/contracts/src/pullRequest.ts index d17258352eda..ed177fbb3020 100644 --- a/packages/contracts/src/pullRequest.ts +++ b/packages/contracts/src/pullRequest.ts @@ -825,6 +825,16 @@ export const PullRequestInvalidateInput = Schema.Struct({ }); export type PullRequestInvalidateInput = typeof PullRequestInvalidateInput.Type; +/** A reference change stays local; account and turn refreshes cover the environment. */ +export const PullRequestRefresh = Schema.Struct({ + revision: NonNegativeInt, + reference: Schema.optional(PullRequestRef), + projectIds: Schema.optional(Schema.Array(ProjectId)), + host: Schema.optional(TrimmedNonEmptyString), + listings: Schema.Boolean, +}); +export type PullRequestRefresh = typeof PullRequestRefresh.Type; + export const PullRequestDetail = Schema.Struct({ provider: SourceControlProviderKind, capabilities: PullRequestCapabilities, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 5aeb20c5d782..11e5053d3e1e 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -100,6 +100,7 @@ import { PullRequestDiffFileContentsInput, PullRequestDiffFileContentsResult, PullRequestInvalidateInput, + PullRequestRefresh, PullRequestListInput, PullRequestListResult, PullRequestListStatsInput, @@ -729,8 +730,8 @@ export const WsPullRequestsInvalidateRpc = Rpc.make(WS_METHODS.pullRequestsInval export const WsPullRequestsSubscribeRefreshesRpc = Rpc.make( WS_METHODS.pullRequestsSubscribeRefreshes, { - payload: Schema.Struct({}), - success: NonNegativeInt, + payload: Schema.Struct({ scoped: Schema.optional(Schema.Boolean) }), + success: Schema.Union([NonNegativeInt, PullRequestRefresh]), error: EnvironmentAuthorizationError, stream: true, }, From c386ee716fc142e9062841dd497f24295794cbae Mon Sep 17 00:00:00 2001 From: Kalven Schraut Date: Mon, 7 Sep 2026 22:20:20 -0500 Subject: [PATCH 3/5] fix(pull-requests): preserve scoped refreshes in stream batches --- apps/server/src/ws.ts | 1 - .../src/state/pullRequests.test.ts | 39 +++++-- .../client-runtime/src/state/pullRequests.ts | 109 +++++++++++++----- 3 files changed, 113 insertions(+), 36 deletions(-) diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5517f10d07d9..3edb6d0eeb2e 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2164,7 +2164,6 @@ const makeWsRpcLayer = ( observeRpcStream( WS_METHODS.pullRequestsSubscribeRefreshes, pullRequests.subscribeRefreshes.pipe( - Stream.filter((event) => input.scoped || event.reference === undefined), Stream.map((event) => (input.scoped ? event : event.revision)), ), { "rpc.aggregate": "pull-requests" }, diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 550ff891897e..9cc279030690 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -291,16 +291,41 @@ it.effect( listed.openUnsafe(); }); yield* Effect.addFinalizer(() => Effect.sync(stop)); - yield* PubSub.publish(refreshEvents, { - revision: 2, - reference, - projectIds: [reference.projectId, sibling.projectId], - listings: true, - }); + const burstUpdated = [Latch.makeUnsafe(), Latch.makeUnsafe()]; + for (const [index, atom] of [activityAtoms[0]!, activityAtoms[2]!].entries()) { + const stopActivity = registry.subscribe(atom, (result) => { + if (AsyncResult.isSuccess(result) && result.value.commentCount === 2) { + burstUpdated[index]!.openUnsafe(); + } + }); + yield* Effect.addFinalizer(() => Effect.sync(stopActivity)); + } + revision = 2; + // One chunk must retain both repositories and the earlier listing invalidation even + // when a later reaction on that reference does not affect listings. + yield* PubSub.publishAll(refreshEvents, [ + { + revision: 2, + reference, + projectIds: [reference.projectId, sibling.projectId], + listings: true, + }, + { + revision: 3, + reference, + projectIds: [reference.projectId, sibling.projectId], + listings: false, + }, + { revision: 4, reference: unrelated, listings: false }, + ]); yield* listed.await; + for (const latch of burstUpdated) yield* latch.await; expect(listReads.get(unrelated.projectId)).toBe(beforeLists.get(unrelated.projectId)); + expect(activityReads.get(reference.projectId)).toBe( + beforeActivity.get(reference.projectId)! + 2, + ); expect(activityReads.get(unrelated.projectId)).toBe( - beforeActivity.get(unrelated.projectId), + beforeActivity.get(unrelated.projectId)! + 1, ); }), ), diff --git a/packages/client-runtime/src/state/pullRequests.ts b/packages/client-runtime/src/state/pullRequests.ts index c574f03c4a74..87edfeed0bce 100644 --- a/packages/client-runtime/src/state/pullRequests.ts +++ b/packages/client-runtime/src/state/pullRequests.ts @@ -11,6 +11,8 @@ import { } from "@t3tools/contracts"; import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; +import * as HashMap from "effect/HashMap"; +import * as Stream from "effect/Stream"; import * as Option from "effect/Option"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -48,25 +50,79 @@ type PullRequestRefreshScope = readonly host?: string; }; -function matchesRefresh(scope: PullRequestRefreshScope, event: PullRequestRefresh): boolean { - if (event.reference === undefined) return true; - if (scope.kind === undefined) return false; - const projectIds = event.projectIds ?? [event.reference.projectId]; - if (scope.kind === "list") { - return ( - event.listings && - (scope.host === undefined || - event.host === undefined || - scope.host.toLowerCase() === event.host) && - (scope.projectIds === undefined || - scope.projectIds.some((projectId) => projectIds.includes(projectId))) +const emptyRefreshRevisions = (global = 0) => ({ + global, + references: HashMap.empty(), + repositories: HashMap.empty(), + listings: HashMap.empty(), +}); +type RefreshRevisions = ReturnType; + +const repositoryRefreshKey = (reference: PullRequestRef, projectId = reference.projectId) => + JSON.stringify([projectId, reference.repository.trim().toLowerCase()]); +const referenceRefreshKey = (reference: PullRequestRef, projectId = reference.projectId) => + JSON.stringify([projectId, reference.repository.trim().toLowerCase(), reference.number]); + +/** Accumulate before the atom keeps only the last value of a stream chunk. */ +function accumulateRefresh( + previous: RefreshRevisions, + value: number | PullRequestRefresh, +): RefreshRevisions { + // Numeric revisions are from servers predating scoped refreshes. + const event: PullRequestRefresh = + typeof value === "number" ? { revision: value, listings: true } : value; + if (event.reference === undefined) return emptyRefreshRevisions(event.revision); + let { references, repositories, listings } = previous; + for (const projectId of event.projectIds ?? [event.reference.projectId]) { + references = HashMap.set( + references, + referenceRefreshKey(event.reference, projectId), + event.revision, ); + repositories = HashMap.set( + repositories, + repositoryRefreshKey(event.reference, projectId), + event.revision, + ); + if (event.listings) { + listings = HashMap.set(listings, projectId, { revision: event.revision, host: event.host }); + } + } + // A long-lived connection must not retain every PR ever visited. Eviction refreshes all + // scopes once so a forgotten revision cannot leave an already mounted query stale. + if ( + HashMap.size(references) > 2_048 || + HashMap.size(repositories) > 2_048 || + HashMap.size(listings) > 2_048 + ) { + return emptyRefreshRevisions(event.revision); + } + return { global: previous.global, references, repositories, listings }; +} + +function refreshRevision(scope: PullRequestRefreshScope, revisions: RefreshRevisions): number { + if (scope.kind === undefined) return revisions.global; + if (scope.kind === "list") { + let revision = revisions.global; + for (const [projectId, listing] of revisions.listings) { + if ( + (scope.projectIds === undefined || scope.projectIds.includes(projectId)) && + (scope.host === undefined || + listing.host === undefined || + scope.host.toLowerCase() === listing.host) + ) { + revision = Math.max(revision, listing.revision); + } + } + return revision; } - return ( - projectIds.includes(scope.reference.projectId) && - scope.reference.repository.trim().toLowerCase() === - event.reference.repository.trim().toLowerCase() && - (scope.kind === "repository" || scope.reference.number === event.reference.number) + const revision = + scope.kind === "reference" + ? HashMap.get(revisions.references, referenceRefreshKey(scope.reference)) + : HashMap.get(revisions.repositories, repositoryRefreshKey(scope.reference)); + return Math.max( + revisions.global, + Option.getOrElse(revision, () => 0), ); } @@ -76,30 +132,27 @@ function createPullRequestRefreshAtomFamily( const events = createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:pull-requests:refreshes", tag: WS_METHODS.pullRequestsSubscribeRefreshes, + transform: (stream) => stream.pipe(Stream.scan(emptyRefreshRevisions(), accumulateRefresh)), }); const scoped = Atom.family((key: string) => { const { environmentId, input } = JSON.parse(key) as { environmentId: EnvironmentId; input: PullRequestRefreshScope; }; - return Atom.make((get): AsyncResult.AsyncResult => { + return Atom.make>((get) => { const result = get(events({ environmentId, input: { scoped: true } })); - // Numeric revisions are from servers predating scoped refreshes. + const previous = Option.getOrUndefined(get.self()); if (AsyncResult.isSuccess(result)) { - const event = - typeof result.value === "number" - ? { revision: result.value, listings: true } - : result.value; - if (matchesRefresh(input, event)) { - const previous = Option.getOrUndefined(get.self()); + const revision = refreshRevision(input, result.value); + if (revision > 0) { return previous !== undefined && AsyncResult.isSuccess(previous) && - previous.value === event.revision + previous.value === revision ? previous - : AsyncResult.success(event.revision); + : AsyncResult.success(revision); } } - return Option.getOrElse(get.self(), () => AsyncResult.initial()); + return previous ?? AsyncResult.initial(); }); }); return (target: { From c0873a012eee87a2f30e3b17ce4e1b446bd43792 Mon Sep 17 00:00:00 2001 From: Kalven Schraut Date: Mon, 7 Sep 2026 22:24:38 -0500 Subject: [PATCH 4/5] fix(client): preserve scoped refresh atom type inference --- packages/client-runtime/src/state/pullRequests.test.ts | 2 +- packages/client-runtime/src/state/pullRequests.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 9cc279030690..8cab6bf00911 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -245,7 +245,7 @@ it.effect( }), ); for (const atom of [...activityAtoms, ...listAtoms]) { - const unmount = registry.mount(atom); + const unmount = registry.mount(atom); yield* Effect.addFinalizer(() => Effect.sync(unmount)); } for (const atom of activityAtoms) diff --git a/packages/client-runtime/src/state/pullRequests.ts b/packages/client-runtime/src/state/pullRequests.ts index 87edfeed0bce..edb014cfdd1b 100644 --- a/packages/client-runtime/src/state/pullRequests.ts +++ b/packages/client-runtime/src/state/pullRequests.ts @@ -141,7 +141,7 @@ function createPullRequestRefreshAtomFamily( }; return Atom.make>((get) => { const result = get(events({ environmentId, input: { scoped: true } })); - const previous = Option.getOrUndefined(get.self()); + const previous = Option.getOrUndefined(get.self>()); if (AsyncResult.isSuccess(result)) { const revision = refreshRevision(input, result.value); if (revision > 0) { From 462f026f21e6593748743f4a10f1b1677cc1f940 Mon Sep 17 00:00:00 2001 From: Kalven Schraut Date: Mon, 7 Sep 2026 23:08:30 -0500 Subject: [PATCH 5/5] fix(server): pass Azure project scope to dependency topology --- .../pullRequest/PullRequestService.test.ts | 60 +++++++++++++++++++ .../src/pullRequest/PullRequestService.ts | 1 + 2 files changed, 61 insertions(+) diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 271d17302732..f16dc8f84053 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -2904,6 +2904,66 @@ it.effect("a mutation makes the next listing ask the host again, with no client }), ); +it.effect( + "matches Azure dependencies within the stored project while keeping CLI selectors short", + () => + Effect.gen(function* () { + const parent = { + ...changeRequest(1, "2026-07-02T00:00:00Z"), + headBranch: "migration", + headRepositoryNameWithOwner: "Project/web", + }; + const child = { + ...changeRequest(2, "2026-07-03T00:00:00Z"), + headBranch: "api", + headRepositoryNameWithOwner: "Project/web", + baseBranch: "migration", + }; + const service = yield* makeService({ + projects: [ + project({ + id: "azure-project", + title: "web", + workspaceRoot: "/azure", + repository: "contoso/Project/_git/web", + provider: "azure-devops", + host: "dev.azure.com", + }), + ], + providers: [ + fakeProvider("azure-devops", { + listChangeRequests: (input) => { + assert.strictEqual(input.repository, "web"); + assert.strictEqual(input.cwd, "/azure"); + return Effect.succeed({ + items: [ + parent, + child, + { ...parent, number: 3, headRepositoryNameWithOwner: "Other/web" }, + ], + truncated: false, + continues: false, + }); + }, + }), + ], + }); + + const result = yield* service.dependencyContext({ + projectId: "azure-project" as ProjectId, + repository: "web", + number: 2, + }); + assert.deepStrictEqual(result.edges, [{ child: 2, parent: 1, certainty: "confirmed" }]); + assert.deepStrictEqual( + result.nodes.map((node) => node.ref.number), + [1, 2], + ); + assert.strictEqual(result.focus.repository, "web"); + assert.strictEqual(result.coverage, "complete"); + }), +); + it.effect("coalesces bounded dependency reads and invalidates them by repository", () => Effect.gen(function* () { let relationshipReads = 0; diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 5e941ca60faa..5b9cb05f70c8 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -1517,6 +1517,7 @@ export const make = Effect.gen(function* () { provider: project.api.kind, host: project.host, repository: project.repository, + repositoryPath: project.project.repositoryIdentity?.displayName ?? null, focus: input.number, rows, complete: branchComplete,