diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 124c4e84f261..34c5a260a494 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); }), ); @@ -2805,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; @@ -2866,7 +3025,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 77843e75427e..4a5716d6d208 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; @@ -1515,6 +1521,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, @@ -2382,11 +2389,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) => @@ -2412,6 +2421,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< @@ -2501,7 +2586,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. @@ -2700,7 +2788,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) => @@ -2713,23 +2801,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 @@ -2738,25 +2823,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( @@ -2784,7 +2859,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, @@ -2800,7 +2875,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..3edb6d0eeb2e 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2160,10 +2160,12 @@ 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.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 65d7b4aefca1..44f036d22428 100644 --- a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx +++ b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx @@ -591,7 +591,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 89059fb10013..411f0d58bc94 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..8cab6bf00911 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,161 @@ 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)); + 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)! + 1, + ); + }), + ), +); diff --git a/packages/client-runtime/src/state/pullRequests.ts b/packages/client-runtime/src/state/pullRequests.ts index 2cadc196bda5..c33edeb1ce7a 100644 --- a/packages/client-runtime/src/state/pullRequests.ts +++ b/packages/client-runtime/src/state/pullRequests.ts @@ -3,13 +3,19 @@ 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 HashMap from "effect/HashMap"; +import * as Stream from "effect/Stream"; 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 +41,124 @@ 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; + }; + +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; + } + 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), + ); +} + 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, + 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) => { + const result = get(events({ environmentId, input: { scoped: true } })); + const previous = Option.getOrUndefined(get.self>()); + if (AsyncResult.isSuccess(result)) { + const revision = refreshRevision(input, result.value); + if (revision > 0) { + return previous !== undefined && + AsyncResult.isSuccess(previous) && + previous.value === revision + ? previous + : AsyncResult.success(revision); + } + } + return previous ?? 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 +172,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 +210,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 +220,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,13 +245,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, input }) => + refreshes({ environmentId, input: { kind: "repository", reference: input } }), }), /** One bounded repository relationship read for the open PR panel, never for list rows. */ dependencyContext: createEnvironmentRpcQueryAtomFamily(runtime, { diff --git a/packages/contracts/src/pullRequest.ts b/packages/contracts/src/pullRequest.ts index e02aa490bdd0..cdbf55734c4d 100644 --- a/packages/contracts/src/pullRequest.ts +++ b/packages/contracts/src/pullRequest.ts @@ -826,6 +826,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, },