import { assert, it, describe } from "@effect/vitest";import * as NodeServices from "@effect/platform-node/NodeServices";import * as Deferred from "effect/Deferred";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock";import type {  VcsStatusLocalResult,  VcsStatusRemoteResult,  VcsStatusResult,  VcsStatusStreamEvent,} from "@t3tools/contracts";import { GitManagerError } from "@t3tools/contracts"; import * as VcsStatusBroadcaster from "./VcsStatusBroadcaster.ts";import * as GitWorkflowService from "../git/GitWorkflowService.ts"; const baseLocalStatus: VcsStatusLocalResult = {  isRepo: true,  sourceControlProvider: {    kind: "github",    name: "GitHub",    baseUrl: "https://github.com",  },  hasPrimaryRemote: true,  isDefaultRef: false,  refName: "feature/status-broadcast",  hasWorkingTreeChanges: false,  workingTree: { files: [], insertions: 0, deletions: 0 },}; const baseRemoteStatus: VcsStatusRemoteResult = {  hasUpstream: true,  aheadCount: 0,  behindCount: 0,  pr: null,}; const remoteStatusWithPr: VcsStatusRemoteResult = {  ...baseRemoteStatus,  pr: {    number: 2978,    title: "[codex] Rewrite client connection architecture",    url: "https://github.com/pingdotgg/t3code/pull/2978",    baseRef: "main",    headRef: "codex/connection-state-audit",    state: "open",  },}; const baseStatus: VcsStatusResult = {  ...baseLocalStatus,  ...baseRemoteStatus,}; function makeTestLayer(state: {  currentLocalStatus: VcsStatusLocalResult;  currentRemoteStatus: VcsStatusRemoteResult | null;  localStatusCalls: number;  remoteStatusCalls: number;  localInvalidationCalls: number;  remoteInvalidationCalls: number;  remoteStatusRefreshUpstreamValues?: Array<boolean | undefined>;}) {  return VcsStatusBroadcaster.layer.pipe(    Layer.provideMerge(NodeServices.layer),    Layer.provide(      Layer.mock(GitWorkflowService.GitWorkflowService)({        localStatus: () =>          Effect.sync(() => {            state.localStatusCalls += 1;            return state.currentLocalStatus;          }),        remoteStatus: (_input, options) =>          Effect.sync(() => {            state.remoteStatusCalls += 1;            state.remoteStatusRefreshUpstreamValues?.push(options?.refreshUpstream);            return state.currentRemoteStatus;          }),        invalidateLocalStatus: () =>          Effect.sync(() => {            state.localInvalidationCalls += 1;          }),        invalidateRemoteStatus: () =>          Effect.sync(() => {            state.remoteInvalidationCalls += 1;          }),      }),    ),  );} describe("VcsStatusBroadcaster", () => {  it.effect("reuses the cached VCS status across repeated reads", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;       const first = yield* broadcaster.getStatus({ cwd: "/repo" });      const second = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.deepStrictEqual(first, baseStatus);      assert.deepStrictEqual(second, baseStatus);      assert.equal(state.localStatusCalls, 1);      assert.equal(state.remoteStatusCalls, 1);      assert.equal(state.localInvalidationCalls, 0);      assert.equal(state.remoteInvalidationCalls, 0);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("refreshes the cached snapshot after explicit invalidation", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const initial = yield* broadcaster.getStatus({ cwd: "/repo" });       state.currentLocalStatus = {        ...baseLocalStatus,        refName: "feature/updated-status",      };      state.currentRemoteStatus = {        ...baseRemoteStatus,        aheadCount: 2,      };      const refreshed = yield* broadcaster.refreshStatus("/repo");      const cached = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.deepStrictEqual(initial, baseStatus);      assert.deepStrictEqual(refreshed, {        ...state.currentLocalStatus,        ...state.currentRemoteStatus,      });      assert.deepStrictEqual(cached, {        ...state.currentLocalStatus,        ...state.currentRemoteStatus,      });      assert.equal(state.localStatusCalls, 2);      assert.equal(state.remoteStatusCalls, 2);      assert.equal(state.localInvalidationCalls, 1);      assert.equal(state.remoteInvalidationCalls, 1);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("keeps the cached snapshot unchanged when a refresh branch fails", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,      failRemoteStatus: false,    };    const testLayer = VcsStatusBroadcaster.layer.pipe(      Layer.provideMerge(NodeServices.layer),      Layer.provide(        Layer.mock(GitWorkflowService.GitWorkflowService)({          localStatus: () =>            Effect.sync(() => {              state.localStatusCalls += 1;              return state.currentLocalStatus;            }),          remoteStatus: () =>            Effect.suspend(() => {              state.remoteStatusCalls += 1;              return state.failRemoteStatus                ? Effect.fail(                    new GitManagerError({                      operation: "VcsStatusBroadcaster.test",                      detail: "remote status failed",                    }),                  )                : Effect.succeed(state.currentRemoteStatus);            }),          invalidateLocalStatus: () =>            Effect.sync(() => {              state.localInvalidationCalls += 1;            }),          invalidateRemoteStatus: () =>            Effect.sync(() => {              state.remoteInvalidationCalls += 1;            }),        }),      ),    );     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      yield* broadcaster.getStatus({ cwd: "/repo" });       state.currentLocalStatus = {        ...baseLocalStatus,        refName: "feature/partial-refresh",      };      state.currentRemoteStatus = {        ...baseRemoteStatus,        aheadCount: 3,      };      state.failRemoteStatus = true;       const refreshExit = yield* broadcaster.refreshStatus("/repo").pipe(Effect.exit);      const cached = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.isTrue(Exit.isFailure(refreshExit));      assert.deepStrictEqual(cached, baseStatus);    }).pipe(Effect.provide(testLayer));  });   it.effect("refreshes only the cached local snapshot when requested", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const initial = yield* broadcaster.getStatus({ cwd: "/repo" });       state.currentLocalStatus = {        ...baseLocalStatus,        refName: "feature/local-only-refresh",        hasWorkingTreeChanges: true,      };       const refreshedLocal = yield* broadcaster.refreshLocalStatus("/repo");      const cached = yield* broadcaster.getStatus({ cwd: "/repo" });       assert.deepStrictEqual(initial, baseStatus);      assert.deepStrictEqual(refreshedLocal, state.currentLocalStatus);      assert.deepStrictEqual(cached, {        ...state.currentLocalStatus,        ...baseRemoteStatus,      });      assert.equal(state.localStatusCalls, 2);      assert.equal(state.remoteStatusCalls, 1);      assert.equal(state.localInvalidationCalls, 1);      assert.equal(state.remoteInvalidationCalls, 0);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("normalizes symlinked CWDs before cache lookup and workflow calls", () => {    const seenCwds: string[] = [];    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };    const testLayer = VcsStatusBroadcaster.layer.pipe(      Layer.provideMerge(NodeServices.layer),      Layer.provide(        Layer.mock(GitWorkflowService.GitWorkflowService)({          localStatus: (input) =>            Effect.sync(() => {              seenCwds.push(input.cwd);              state.localStatusCalls += 1;              return state.currentLocalStatus;            }),          remoteStatus: (input) =>            Effect.sync(() => {              seenCwds.push(input.cwd);              state.remoteStatusCalls += 1;              return state.currentRemoteStatus;            }),          invalidateLocalStatus: () =>            Effect.sync(() => {              state.localInvalidationCalls += 1;            }),          invalidateRemoteStatus: () =>            Effect.sync(() => {              state.remoteInvalidationCalls += 1;            }),        } satisfies Partial<GitWorkflowService.GitWorkflowService["Service"]>),      ),    );     return Effect.gen(function* () {      const fileSystem = yield* FileSystem.FileSystem;      const path = yield* Path.Path;      const realDir = yield* fileSystem.makeTempDirectoryScoped({        prefix: "t3-vcs-status-real-",      });      const linkParent = yield* fileSystem.makeTempDirectoryScoped({        prefix: "t3-vcs-status-link-",      });      const linkDir = path.join(linkParent, "repo-link");      yield* fileSystem.symlink(realDir, linkDir);      const realPath = yield* fileSystem.realPath(realDir);       const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      yield* broadcaster.getStatus({ cwd: linkDir });      yield* broadcaster.getStatus({ cwd: realDir });       assert.deepStrictEqual(seenCwds, [realPath, realPath]);      assert.equal(state.localStatusCalls, 1);      assert.equal(state.remoteStatusCalls, 1);    }).pipe(Effect.provide(testLayer));  });   it.effect("streams a local snapshot first and remote updates later", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: baseRemoteStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const snapshotDeferred = yield* Deferred.make<VcsStatusStreamEvent>();      const remoteUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();      yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => {        if (event._tag === "snapshot") {          return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore);        }        if (event._tag === "remoteUpdated") {          return Deferred.succeed(remoteUpdatedDeferred, event).pipe(Effect.ignore);        }        return Effect.void;      }).pipe(Effect.forkScoped);       const snapshot = yield* Deferred.await(snapshotDeferred);      yield* broadcaster.refreshStatus("/repo");      const remoteUpdated = yield* Deferred.await(remoteUpdatedDeferred);       assert.deepStrictEqual(snapshot, {        _tag: "snapshot",        local: baseLocalStatus,        remote: null,      } satisfies VcsStatusStreamEvent);      assert.deepStrictEqual(remoteUpdated, {        _tag: "remoteUpdated",        remote: baseRemoteStatus,      } satisfies VcsStatusStreamEvent);    }).pipe(Effect.provide(makeTestLayer(state)));  });   it.effect("loads remote status once when periodic refreshes are disabled", () => {    const state = {      currentLocalStatus: baseLocalStatus,      currentRemoteStatus: remoteStatusWithPr,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,      remoteStatusRefreshUpstreamValues: [] as Array<boolean | undefined>,    };     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const scope = yield* Scope.make();      const snapshotDeferred = yield* Deferred.make<VcsStatusStreamEvent>();      const remoteUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();      yield* Stream.runForEach(        broadcaster.streamStatus(          { cwd: "/repo" },          { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) },        ),        (event) => {          if (event._tag === "snapshot") {            return Deferred.succeed(snapshotDeferred, event).pipe(Effect.ignore);          }          if (event._tag === "remoteUpdated") {            return Deferred.succeed(remoteUpdatedDeferred, event).pipe(Effect.ignore);          }          return Effect.void;        },      ).pipe(Effect.forkIn(scope));       const snapshot = yield* Deferred.await(snapshotDeferred);      const remoteUpdated = yield* Deferred.await(remoteUpdatedDeferred);       assert.deepStrictEqual(snapshot, {        _tag: "snapshot",        local: baseLocalStatus,        remote: null,      } satisfies VcsStatusStreamEvent);      assert.deepStrictEqual(remoteUpdated, {        _tag: "remoteUpdated",        remote: remoteStatusWithPr,      } satisfies VcsStatusStreamEvent);      assert.equal(state.remoteStatusCalls, 1);      assert.equal(state.remoteInvalidationCalls, 0);      assert.deepStrictEqual(state.remoteStatusRefreshUpstreamValues, [false]);       yield* TestClock.adjust(Duration.minutes(2));      assert.equal(state.remoteStatusCalls, 1);      assert.equal(state.remoteInvalidationCalls, 0);       yield* Scope.close(scope, Exit.void);    }).pipe(Effect.provide(Layer.merge(makeTestLayer(state), TestClock.layer())));  });   it.effect("retries the initial remote load when periodic refreshes are disabled", () => {    const state = {      currentLocalStatus: baseLocalStatus,      localStatusCalls: 0,      remoteStatusCalls: 0,      localInvalidationCalls: 0,      remoteInvalidationCalls: 0,      remoteStatusRefreshUpstreamValues: [] as Array<boolean | undefined>,    };    let firstRemoteAttemptDeferred: Deferred.Deferred<void> | null = null;    const testLayer = VcsStatusBroadcaster.layer.pipe(      Layer.provideMerge(NodeServices.layer),      Layer.provide(        Layer.mock(GitWorkflowService.GitWorkflowService)({          localStatus: () =>            Effect.sync(() => {              state.localStatusCalls += 1;              return state.currentLocalStatus;            }),          remoteStatus: (_input, options) =>            Effect.suspend(() => {              state.remoteStatusCalls += 1;              state.remoteStatusRefreshUpstreamValues.push(options?.refreshUpstream);              if (state.remoteStatusCalls === 1) {                return Effect.fail(                  new GitManagerError({                    operation: "VcsStatusBroadcaster.test",                    detail: "initial remote status failed",                  }),                ).pipe(                  Effect.ensuring(                    firstRemoteAttemptDeferred                      ? Deferred.succeed(firstRemoteAttemptDeferred, undefined).pipe(Effect.ignore)                      : Effect.void,                  ),                );              }              return Effect.succeed(remoteStatusWithPr);            }),          invalidateLocalStatus: () =>            Effect.sync(() => {              state.localInvalidationCalls += 1;            }),          invalidateRemoteStatus: () =>            Effect.sync(() => {              state.remoteInvalidationCalls += 1;            }),        }),      ),    );     return Effect.gen(function* () {      const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;      const scope = yield* Scope.make();      firstRemoteAttemptDeferred = yield* Deferred.make<void>();      const remoteUpdatedDeferred = yield* Deferred.make<VcsStatusStreamEvent>();      yield* Stream.runForEach(        broadcaster.streamStatus(          { cwd: "/repo" },          { automaticRemoteRefreshInterval: Effect.succeed(Duration.zero) },        ),        (event) =>          event._tag === "remoteUpdated"            ? Deferred.succeed(remoteUpdatedDeferred, event).pipe(Effect.ignore)            : Effect.void,      ).pipe(Effect.forkIn(scope));       yield* Deferred.await(firstRemoteAttemptDeferred);      yield* Effect.yieldNow;      assert.equal(state.remoteStatusCalls, 1);       yield* TestClock.adjust(Duration.seconds(30));      const remoteUpdated = yield* Deferred.await(remoteUpdatedDeferred);       assert.deepStrictEqual(remoteUpdated, {        _tag: "remoteUpdated",

Lines 1–500 of 660. Load more