import { assert, it } from "@effect/vitest";import { Deferred, Effect, Exit, Layer, Option, Scope, Stream } from "effect";import type { GitStatusLocalResult, GitStatusRemoteResult, GitStatusResult, GitStatusStreamEvent,} from "@t3tools/contracts";import { describe } from "vitest"; import { GitStatusBroadcaster } from "../Services/GitStatusBroadcaster.ts";import { GitStatusBroadcasterLive } from "./GitStatusBroadcaster.ts";import { type GitManagerShape, GitManager } from "../Services/GitManager.ts"; const baseLocalStatus: GitStatusLocalResult = { isRepo: true, hostingProvider: { kind: "github", name: "GitHub", baseUrl: "https://github.com", }, hasOriginRemote: true, isDefaultBranch: false, branch: "feature/status-broadcast", hasWorkingTreeChanges: false, workingTree: { files: [], insertions: 0, deletions: 0 },}; const baseRemoteStatus: GitStatusRemoteResult = { hasUpstream: true, aheadCount: 0, behindCount: 0, pr: null,}; const baseStatus: GitStatusResult = { ...baseLocalStatus, ...baseRemoteStatus,}; function makeTestLayer(state: { currentLocalStatus: GitStatusLocalResult; currentRemoteStatus: GitStatusRemoteResult | null; localStatusCalls: number; remoteStatusCalls: number; localInvalidationCalls: number; remoteInvalidationCalls: number;}) { const gitManager: GitManagerShape = { localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; return state.currentLocalStatus; }), remoteStatus: () => Effect.sync(() => { state.remoteStatusCalls += 1; return state.currentRemoteStatus; }), status: () => Effect.die("status should not be called in this test"), invalidateLocalStatus: () => Effect.sync(() => { state.localInvalidationCalls += 1; }), invalidateRemoteStatus: () => Effect.sync(() => { state.remoteInvalidationCalls += 1; }), invalidateStatus: () => Effect.die("invalidateStatus should not be called in this test"), resolvePullRequest: () => Effect.die("resolvePullRequest should not be called in this test"), preparePullRequestThread: () => Effect.die("preparePullRequestThread should not be called in this test"), runStackedAction: () => Effect.die("runStackedAction should not be called in this test"), }; return GitStatusBroadcasterLive.pipe(Layer.provide(Layer.succeed(GitManager, gitManager)));} describe("GitStatusBroadcasterLive", () => { it.effect("reuses the cached git 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* GitStatusBroadcaster; 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* GitStatusBroadcaster; const initial = yield* broadcaster.getStatus({ cwd: "/repo" }); state.currentLocalStatus = { ...baseLocalStatus, branch: "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("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* GitStatusBroadcaster; const initial = yield* broadcaster.getStatus({ cwd: "/repo" }); state.currentLocalStatus = { ...baseLocalStatus, branch: "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("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* GitStatusBroadcaster; const snapshotDeferred = yield* Deferred.make<GitStatusStreamEvent>(); const remoteUpdatedDeferred = yield* Deferred.make<GitStatusStreamEvent>(); 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 GitStatusStreamEvent); assert.deepStrictEqual(remoteUpdated, { _tag: "remoteUpdated", remote: baseRemoteStatus, } satisfies GitStatusStreamEvent); }).pipe(Effect.provide(makeTestLayer(state))); }); it.effect("stops the remote poller after the last stream subscriber disconnects", () => { const state = { currentLocalStatus: baseLocalStatus, currentRemoteStatus: baseRemoteStatus, localStatusCalls: 0, remoteStatusCalls: 0, localInvalidationCalls: 0, remoteInvalidationCalls: 0, }; let remoteInterruptedDeferred: Deferred.Deferred<void, never> | null = null; let remoteStartedDeferred: Deferred.Deferred<void, never> | null = null; const testLayer = GitStatusBroadcasterLive.pipe( Layer.provide( Layer.succeed(GitManager, { localStatus: () => Effect.sync(() => { state.localStatusCalls += 1; return state.currentLocalStatus; }), remoteStatus: () => Effect.sync(() => { state.remoteStatusCalls += 1; }).pipe( Effect.andThen( remoteStartedDeferred ? Deferred.succeed(remoteStartedDeferred, undefined).pipe(Effect.ignore) : Effect.void, ), Effect.andThen(Effect.never as Effect.Effect<GitStatusRemoteResult | null, never>), Effect.onInterrupt(() => remoteInterruptedDeferred ? Deferred.succeed(remoteInterruptedDeferred, undefined).pipe(Effect.ignore) : Effect.void, ), ), status: () => Effect.die("status should not be called in this test"), invalidateLocalStatus: () => Effect.sync(() => { state.localInvalidationCalls += 1; }), invalidateRemoteStatus: () => Effect.sync(() => { state.remoteInvalidationCalls += 1; }), invalidateStatus: () => Effect.die("invalidateStatus should not be called in this test"), resolvePullRequest: () => Effect.die("resolvePullRequest should not be called in this test"), preparePullRequestThread: () => Effect.die("preparePullRequestThread should not be called in this test"), runStackedAction: () => Effect.die("runStackedAction should not be called in this test"), } satisfies GitManagerShape), ), ); return Effect.gen(function* () { const remoteInterrupted = yield* Deferred.make<void>(); const remoteStarted = yield* Deferred.make<void>(); remoteInterruptedDeferred = remoteInterrupted; remoteStartedDeferred = remoteStarted; const broadcaster = yield* GitStatusBroadcaster; const firstSnapshot = yield* Deferred.make<GitStatusStreamEvent>(); const secondSnapshot = yield* Deferred.make<GitStatusStreamEvent>(); const firstScope = yield* Scope.make(); const secondScope = yield* Scope.make(); yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => event._tag === "snapshot" ? Deferred.succeed(firstSnapshot, event).pipe(Effect.ignore) : Effect.void, ).pipe(Effect.forkIn(firstScope)); yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) => event._tag === "snapshot" ? Deferred.succeed(secondSnapshot, event).pipe(Effect.ignore) : Effect.void, ).pipe(Effect.forkIn(secondScope)); yield* Deferred.await(firstSnapshot); yield* Deferred.await(secondSnapshot); yield* Deferred.await(remoteStarted); assert.equal(state.remoteStatusCalls, 1); yield* Scope.close(firstScope, Exit.void); assert.equal(Option.isNone(yield* Deferred.poll(remoteInterrupted)), true); yield* Scope.close(secondScope, Exit.void).pipe(Effect.forkScoped); yield* Deferred.await(remoteInterrupted); assert.equal(Option.isSome(yield* Deferred.poll(remoteInterrupted)), true); }).pipe(Effect.provide(testLayer)); });});
import { assert, it } from "@effect/vitest";
import { Deferred, Effect, Exit, Layer, Option, Scope, Stream } from "effect";
import type {
GitStatusLocalResult,
GitStatusRemoteResult,
GitStatusResult,
GitStatusStreamEvent,
} from "@t3tools/contracts";
import { describe } from "vitest";
import { GitStatusBroadcaster } from "../Services/GitStatusBroadcaster.ts";
import { GitStatusBroadcasterLive } from "./GitStatusBroadcaster.ts";
import { type GitManagerShape, GitManager } from "../Services/GitManager.ts";
const baseLocalStatus: GitStatusLocalResult = {
isRepo: true,
hostingProvider: {
kind: "github",
name: "GitHub",
baseUrl: "https://github.com",
},
hasOriginRemote: true,
isDefaultBranch: false,
branch: "feature/status-broadcast",
hasWorkingTreeChanges: false,
workingTree: { files: [], insertions: 0, deletions: 0 },
};
const baseRemoteStatus: GitStatusRemoteResult = {
hasUpstream: true,
aheadCount: 0,
behindCount: 0,
pr: null,
};
const baseStatus: GitStatusResult = {
...baseLocalStatus,
...baseRemoteStatus,
};
function makeTestLayer(state: {
currentLocalStatus: GitStatusLocalResult;
currentRemoteStatus: GitStatusRemoteResult | null;
localStatusCalls: number;
remoteStatusCalls: number;
localInvalidationCalls: number;
remoteInvalidationCalls: number;
}) {
const gitManager: GitManagerShape = {
localStatus: () =>
Effect.sync(() => {
state.localStatusCalls += 1;
return state.currentLocalStatus;
}),
remoteStatus: () =>
Effect.sync(() => {
state.remoteStatusCalls += 1;
return state.currentRemoteStatus;
}),
status: () => Effect.die("status should not be called in this test"),
invalidateLocalStatus: () =>
Effect.sync(() => {
state.localInvalidationCalls += 1;
}),
invalidateRemoteStatus: () =>
Effect.sync(() => {
state.remoteInvalidationCalls += 1;
}),
invalidateStatus: () => Effect.die("invalidateStatus should not be called in this test"),
resolvePullRequest: () => Effect.die("resolvePullRequest should not be called in this test"),
preparePullRequestThread: () =>
Effect.die("preparePullRequestThread should not be called in this test"),
runStackedAction: () => Effect.die("runStackedAction should not be called in this test"),
};
return GitStatusBroadcasterLive.pipe(Layer.provide(Layer.succeed(GitManager, gitManager)));
}
describe("GitStatusBroadcasterLive", () => {
it.effect("reuses the cached git 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* GitStatusBroadcaster;
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* GitStatusBroadcaster;
const initial = yield* broadcaster.getStatus({ cwd: "/repo" });
state.currentLocalStatus = {
...baseLocalStatus,
branch: "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("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* GitStatusBroadcaster;
const initial = yield* broadcaster.getStatus({ cwd: "/repo" });
state.currentLocalStatus = {
...baseLocalStatus,
branch: "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("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* GitStatusBroadcaster;
const snapshotDeferred = yield* Deferred.make<GitStatusStreamEvent>();
const remoteUpdatedDeferred = yield* Deferred.make<GitStatusStreamEvent>();
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 GitStatusStreamEvent);
assert.deepStrictEqual(remoteUpdated, {
_tag: "remoteUpdated",
remote: baseRemoteStatus,
} satisfies GitStatusStreamEvent);
}).pipe(Effect.provide(makeTestLayer(state)));
});
it.effect("stops the remote poller after the last stream subscriber disconnects", () => {
const state = {
currentLocalStatus: baseLocalStatus,
currentRemoteStatus: baseRemoteStatus,
localStatusCalls: 0,
remoteStatusCalls: 0,
localInvalidationCalls: 0,
remoteInvalidationCalls: 0,
};
let remoteInterruptedDeferred: Deferred.Deferred<void, never> | null = null;
let remoteStartedDeferred: Deferred.Deferred<void, never> | null = null;
const testLayer = GitStatusBroadcasterLive.pipe(
Layer.provide(
Layer.succeed(GitManager, {
localStatus: () =>
Effect.sync(() => {
state.localStatusCalls += 1;
return state.currentLocalStatus;
}),
remoteStatus: () =>
Effect.sync(() => {
state.remoteStatusCalls += 1;
}).pipe(
Effect.andThen(
remoteStartedDeferred
? Deferred.succeed(remoteStartedDeferred, undefined).pipe(Effect.ignore)
: Effect.void,
),
Effect.andThen(Effect.never as Effect.Effect<GitStatusRemoteResult | null, never>),
Effect.onInterrupt(() =>
remoteInterruptedDeferred
? Deferred.succeed(remoteInterruptedDeferred, undefined).pipe(Effect.ignore)
: Effect.void,
),
),
status: () => Effect.die("status should not be called in this test"),
invalidateLocalStatus: () =>
Effect.sync(() => {
state.localInvalidationCalls += 1;
}),
invalidateRemoteStatus: () =>
Effect.sync(() => {
state.remoteInvalidationCalls += 1;
}),
invalidateStatus: () => Effect.die("invalidateStatus should not be called in this test"),
resolvePullRequest: () =>
Effect.die("resolvePullRequest should not be called in this test"),
preparePullRequestThread: () =>
Effect.die("preparePullRequestThread should not be called in this test"),
runStackedAction: () => Effect.die("runStackedAction should not be called in this test"),
} satisfies GitManagerShape),
),
);
return Effect.gen(function* () {
const remoteInterrupted = yield* Deferred.make<void>();
const remoteStarted = yield* Deferred.make<void>();
remoteInterruptedDeferred = remoteInterrupted;
remoteStartedDeferred = remoteStarted;
const broadcaster = yield* GitStatusBroadcaster;
const firstSnapshot = yield* Deferred.make<GitStatusStreamEvent>();
const secondSnapshot = yield* Deferred.make<GitStatusStreamEvent>();
const firstScope = yield* Scope.make();
const secondScope = yield* Scope.make();
yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) =>
event._tag === "snapshot"
? Deferred.succeed(firstSnapshot, event).pipe(Effect.ignore)
: Effect.void,
).pipe(Effect.forkIn(firstScope));
yield* Stream.runForEach(broadcaster.streamStatus({ cwd: "/repo" }), (event) =>
event._tag === "snapshot"
? Deferred.succeed(secondSnapshot, event).pipe(Effect.ignore)
: Effect.void,
).pipe(Effect.forkIn(secondScope));
yield* Deferred.await(firstSnapshot);
yield* Deferred.await(secondSnapshot);
yield* Deferred.await(remoteStarted);
assert.equal(state.remoteStatusCalls, 1);
yield* Scope.close(firstScope, Exit.void);
assert.equal(Option.isNone(yield* Deferred.poll(remoteInterrupted)), true);
yield* Scope.close(secondScope, Exit.void).pipe(Effect.forkScoped);
yield* Deferred.await(remoteInterrupted);
assert.equal(Option.isSome(yield* Deferred.poll(remoteInterrupted)), true);
}).pipe(Effect.provide(testLayer));
});
});