diff --git a/packages/client-runtime/src/state/review.test.ts b/packages/client-runtime/src/state/review.test.ts new file mode 100644 index 00000000000..8d7a2b47df0 --- /dev/null +++ b/packages/client-runtime/src/state/review.test.ts @@ -0,0 +1,40 @@ +import { EnvironmentId } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import { Atom, AtomRegistry } from "effect/unstable/reactivity"; + +import { invalidateReviewDiffPreviews, reviewDiffPreviewRevisionAtom } from "./review.ts"; + +const TARGET = { + environmentId: EnvironmentId.make("environment-1"), +}; + +describe("review diff preview invalidation", () => { + it("scopes revision changes to the invalidated environment", () => { + const registry = AtomRegistry.make(); + const otherEnvironment = { + environmentId: EnvironmentId.make("environment-2"), + }; + + expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(0); + expect(registry.get(reviewDiffPreviewRevisionAtom(otherEnvironment))).toBe(0); + + invalidateReviewDiffPreviews(registry, TARGET); + + expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(1); + expect(registry.get(reviewDiffPreviewRevisionAtom(otherEnvironment))).toBe(0); + registry.dispose(); + }); + + it("notifies reactive dependents without replacing their atom identity", () => { + const registry = AtomRegistry.make(); + const dependent = Atom.make((get) => get(reviewDiffPreviewRevisionAtom(TARGET))); + const unmount = registry.mount(dependent); + + expect(registry.get(dependent)).toBe(0); + invalidateReviewDiffPreviews(registry, TARGET); + expect(registry.get(dependent)).toBe(1); + + unmount(); + registry.dispose(); + }); +}); diff --git a/packages/client-runtime/src/state/review.ts b/packages/client-runtime/src/state/review.ts index 0d78d6edd9f..862482861d5 100644 --- a/packages/client-runtime/src/state/review.ts +++ b/packages/client-runtime/src/state/review.ts @@ -1,9 +1,31 @@ -import { WS_METHODS } from "@t3tools/contracts"; -import { Atom } from "effect/unstable/reactivity"; +import { type EnvironmentId, WS_METHODS } from "@t3tools/contracts"; +import { Atom, type AtomRegistry } from "effect/unstable/reactivity"; import { createEnvironmentRpcQueryAtomFamily } from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; +export interface ReviewDiffPreviewInvalidationTarget { + readonly environmentId: EnvironmentId; +} + +const diffPreviewRevisionByEnvironment = Atom.family((environmentId: EnvironmentId) => + Atom.make(0).pipe( + Atom.keepAlive, + Atom.withLabel(`environment-data:review:diff-preview-revision:${environmentId}`), + ), +); + +export function reviewDiffPreviewRevisionAtom(target: ReviewDiffPreviewInvalidationTarget) { + return diffPreviewRevisionByEnvironment(target.environmentId); +} + +export function invalidateReviewDiffPreviews( + registry: AtomRegistry.AtomRegistry, + target: ReviewDiffPreviewInvalidationTarget, +): void { + registry.update(reviewDiffPreviewRevisionAtom(target), (revision) => revision + 1); +} + export function createReviewEnvironmentAtoms( runtime: Atom.AtomRuntime, ) { @@ -12,6 +34,7 @@ export function createReviewEnvironmentAtoms( label: "environment-data:review:diff-preview", tag: WS_METHODS.reviewGetDiffPreview, staleTimeMs: 5_000, + refreshSignal: reviewDiffPreviewRevisionAtom, }), }; } diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index fb5e9a0ab55..ad1f04dcf79 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -52,6 +52,10 @@ interface EnvironmentQueryAtomOptions extends EnvironmentAtomOpt readonly staleTimeMs?: number; readonly idleTtlMs?: number; readonly refreshIntervalMs?: number; + readonly refreshSignal?: (target: { + readonly environmentId: EnvironmentIdType; + readonly input: Input; + }) => Atom.Atom; } interface EnvironmentSubscriptionAtomOptions { @@ -488,6 +492,9 @@ function createEnvironmentQueryAtomFamily( const idleTtlMs = options.idleTtlMs ?? 5 * 60_000; const queryAtom = runtime .atom((get) => { + if (options.refreshSignal !== undefined) { + get(options.refreshSignal(target)); + } const generation = Option.getOrNull( AsyncResult.value(get(rpcGenerationAtom(target.environmentId))), ); @@ -576,6 +583,10 @@ export function createEnvironmentRpcQueryAtomFamily; + }) => Atom.Atom; }, ) { return createEnvironmentQueryAtomFamily(runtime, { @@ -585,6 +596,7 @@ export function createEnvironmentRpcQueryAtomFamily) => request(options.tag, input), }); } diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index 0a6264c6207..d2db9f674a8 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -3,6 +3,8 @@ import { WS_METHODS, type VcsListRefsInput, type VcsListRefsResult, + type VcsStatusLocalResult, + type VcsStatusRemoteResult, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; @@ -32,7 +34,9 @@ import { commitVcsRefsRefresh, createVcsEnvironmentAtoms, makeCachedVcsRefsChanges, + projectVcsStatusChanges, } from "./vcs.ts"; +import { reviewDiffPreviewRevisionAtom } from "./review.ts"; import { invalidateCachedVcsRefs, invalidateVcsRefs, @@ -82,6 +86,36 @@ const LIVE_REFS: VcsListRefsResult = { ], }; +const LOCAL_STATUS_WITH_FILE: VcsStatusLocalResult = { + isRepo: true, + hasPrimaryRemote: true, + isDefaultRef: false, + refName: "feature", + hasWorkingTreeChanges: true, + workingTree: { + files: [{ path: "temporary.txt", insertions: 1, deletions: 0 }], + insertions: 1, + deletions: 0, + }, +}; + +const LOCAL_STATUS_AFTER_DELETION: VcsStatusLocalResult = { + ...LOCAL_STATUS_WITH_FILE, + hasWorkingTreeChanges: false, + workingTree: { + files: [], + insertions: 0, + deletions: 0, + }, +}; + +const REMOTE_STATUS: VcsStatusRemoteResult = { + hasUpstream: true, + aheadCount: 0, + behindCount: 0, + pr: null, +}; + function session(client: WsRpcProtocolClient): RpcSession { return { client, @@ -114,6 +148,45 @@ function cacheWithRefs( } describe("cached VCS refs", () => { + it.effect( + "invalidates diff previews after a changed status event, not the initial snapshot", + () => + Effect.gen(function* () { + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => + Effect.sync(() => registry.dispose()), + ); + const revisionAtom = reviewDiffPreviewRevisionAtom(TARGET); + const revisions: number[] = []; + + const statuses = yield* projectVcsStatusChanges( + TARGET.environmentId, + Stream.make( + { + _tag: "snapshot" as const, + local: LOCAL_STATUS_WITH_FILE, + remote: REMOTE_STATUS, + }, + { + _tag: "localUpdated" as const, + local: LOCAL_STATUS_AFTER_DELETION, + }, + ), + ).pipe( + Stream.tap(() => + Effect.sync(() => { + revisions.push(registry.get(revisionAtom)); + }), + ), + Stream.provideService(AtomRegistry.AtomRegistry, registry), + Stream.runCollect, + ); + + expect(statuses).toHaveLength(2); + expect(statuses[1]?.workingTree.files).toEqual([]); + expect(revisions).toEqual([0, 1]); + }), + ); + it("invalidates all ref streams in the mutated environment", () => { const registry = AtomRegistry.make(); const environment = { @@ -260,6 +333,7 @@ describe("cached VCS refs", () => { revision: 1, persistedCacheReadable: true, }); + expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(1); expect( yield* commitVcsRefsRefresh(registry, cache, { @@ -329,6 +403,7 @@ describe("cached VCS refs", () => { expect(AsyncResult.isFailure(result)).toBe(true); expect(yield* Ref.get(clears)).toBe(1); expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(1); + expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(1); const refreshResult = yield* Effect.promise(() => atoms.refreshStatus.run(registry, { @@ -340,6 +415,7 @@ describe("cached VCS refs", () => { expect(AsyncResult.isSuccess(refreshResult)).toBe(true); expect(yield* Ref.get(clears)).toBe(2); expect(registry.get(vcsRefsCacheStateAtom(TARGET)).revision).toBe(2); + expect(registry.get(reviewDiffPreviewRevisionAtom(TARGET))).toBe(2); }), ), ); diff --git a/packages/client-runtime/src/state/vcs.ts b/packages/client-runtime/src/state/vcs.ts index a0d4510be7f..981c462b259 100644 --- a/packages/client-runtime/src/state/vcs.ts +++ b/packages/client-runtime/src/state/vcs.ts @@ -3,6 +3,7 @@ import { type VcsListRefsInput, type VcsListRefsResult, type VcsStatusResult, + type VcsStatusStreamEvent, WS_METHODS, } from "@t3tools/contracts"; import { applyGitStatusStreamEvent } from "@t3tools/shared/git"; @@ -15,7 +16,7 @@ import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/unstable/reactivity"; -import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts"; +import { createEnvironmentRpcCommand } from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; @@ -23,6 +24,7 @@ import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts"; +import { invalidateReviewDiffPreviews } from "./review.ts"; import { invalidateCachedVcsRefs, vcsRefsCacheStateAtom, @@ -233,6 +235,34 @@ export function cachedVcsRefsChanges( ); } +export function projectVcsStatusChanges( + environmentId: EnvironmentId, + events: Stream.Stream, +) { + return events.pipe( + Stream.mapAccumEffect( + () => null as VcsStatusResult | null, + (current, event) => { + const next = applyGitStatusStreamEvent(current, event); + return Effect.gen(function* () { + if (current !== null) { + const registry = yield* AtomRegistry.AtomRegistry; + invalidateReviewDiffPreviews(registry, { environmentId }); + } + return [next, [next]] as const; + }); + }, + ), + ); +} + +export function vcsStatusChanges( + environmentId: EnvironmentId, + input: EnvironmentRpcInput, +) { + return projectVcsStatusChanges(environmentId, subscribe(WS_METHODS.subscribeVcsStatus, input)); +} + export function createVcsEnvironmentAtoms( runtime: Atom.AtomRuntime, ) { @@ -259,6 +289,23 @@ export function createVcsEnvironmentAtoms( readonly environmentId: EnvironmentId; readonly input: VcsListRefsInput; }) => listRefsByEnvironment(target.environmentId)(JSON.stringify(target.input)); + const statusByEnvironment = Atom.family((environmentId: EnvironmentId) => + Atom.family((inputKey: string) => { + const input = JSON.parse(inputKey) as EnvironmentRpcInput< + typeof WS_METHODS.subscribeVcsStatus + >; + return runtime + .atom(followStreamInEnvironment(environmentId, vcsStatusChanges(environmentId, input))) + .pipe( + Atom.setIdleTTL(5 * 60_000), + Atom.withLabel(`environment-data:vcs:status:${environmentId}:${inputKey}`), + ); + }), + ); + const status = (target: { + readonly environmentId: EnvironmentId; + readonly input: EnvironmentRpcInput; + }) => statusByEnvironment(target.environmentId)(JSON.stringify(target.input)); const invalidateRefs = ( target: { readonly environmentId: EnvironmentId; readonly input: { readonly cwd: string } }, registry: AtomRegistry.AtomRegistry, @@ -270,19 +317,7 @@ export function createVcsEnvironmentAtoms( return { listRefs, - status: createEnvironmentSubscriptionAtomFamily(runtime, { - label: "environment-data:vcs:status", - subscribe: (input: EnvironmentRpcInput) => - subscribe(WS_METHODS.subscribeVcsStatus, input).pipe( - Stream.mapAccum( - () => null as VcsStatusResult | null, - (current, event) => { - const next = applyGitStatusStreamEvent(current, event); - return [next, [next]] as const; - }, - ), - ), - }), + status, pull: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:pull", tag: WS_METHODS.vcsPull, diff --git a/packages/client-runtime/src/state/vcsRefInvalidation.ts b/packages/client-runtime/src/state/vcsRefInvalidation.ts index ff9de7bd1cc..4cf13f94911 100644 --- a/packages/client-runtime/src/state/vcsRefInvalidation.ts +++ b/packages/client-runtime/src/state/vcsRefInvalidation.ts @@ -5,6 +5,7 @@ import { Atom, type AtomRegistry } from "effect/unstable/reactivity"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; +import { invalidateReviewDiffPreviews } from "./review.ts"; export interface VcsRefsInvalidationTarget { readonly environmentId: EnvironmentId; @@ -78,6 +79,7 @@ export const invalidateCachedVcsRefs = Effect.fn("VcsRefsState.invalidateCached" ), ); invalidateVcsRefs(registry, target, persistedCacheReadable); + invalidateReviewDiffPreviews(registry, target); }), ); });