import { realpathSync } from "node:fs"; import { Context, Duration, Effect, Exit, Fiber, Layer, PubSub, Ref, Scope, Stream, SynchronizedRef,} from "effect";import type { GitManagerServiceError, VcsStatusInput, VcsStatusLocalResult, VcsStatusRemoteResult, VcsStatusResult, VcsStatusStreamEvent,} from "@t3tools/contracts";import { mergeGitStatusParts } from "@t3tools/shared/git"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; const VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30); interface VcsStatusChange { readonly cwd: string; readonly event: VcsStatusStreamEvent;} interface CachedValue<T> { readonly fingerprint: string; readonly value: T;} interface CachedVcsStatus { readonly local: CachedValue<VcsStatusLocalResult> | null; readonly remote: CachedValue<VcsStatusRemoteResult | null> | null;} interface ActiveRemotePoller { readonly fiber: Fiber.Fiber<void, never>; readonly subscriberCount: number;} export interface VcsStatusBroadcasterShape { readonly getStatus: ( input: VcsStatusInput, ) => Effect.Effect<VcsStatusResult, GitManagerServiceError>; readonly refreshLocalStatus: ( cwd: string, ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>; readonly refreshStatus: (cwd: string) => Effect.Effect<VcsStatusResult, GitManagerServiceError>; readonly streamStatus: ( input: VcsStatusInput, ) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>;} export class VcsStatusBroadcaster extends Context.Service< VcsStatusBroadcaster, VcsStatusBroadcasterShape>()("t3/vcs/VcsStatusBroadcaster") {} function fingerprintStatusPart(status: unknown): string { return JSON.stringify(status);} function normalizeCwd(cwd: string): string { try { return realpathSync.native(cwd); } catch { return cwd; }} export const layer = Layer.effect( VcsStatusBroadcaster, Effect.gen(function* () { const workflow = yield* GitWorkflowService.GitWorkflowService; const changesPubSub = yield* Effect.acquireRelease( PubSub.unbounded<VcsStatusChange>(), (pubsub) => PubSub.shutdown(pubsub), ); const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) => Scope.close(scope, Exit.void), ); const cacheRef = yield* Ref.make(new Map<string, CachedVcsStatus>()); const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>()); const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* ( cwd: string, ) { return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null)); }); const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")( function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) { const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local, } satisfies CachedValue<VcsStatusLocalResult>; const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { const previous = cache.get(cwd) ?? { local: null, remote: null }; const nextCache = new Map(cache); nextCache.set(cwd, { ...previous, local: nextLocal, }); return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const; }); if (options?.publish && shouldPublish) { yield* PubSub.publish(changesPubSub, { cwd, event: { _tag: "localUpdated", local, }, }); } return local; }, ); const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")( function* ( cwd: string, remote: VcsStatusRemoteResult | null, options?: { publish?: boolean }, ) { const nextRemote = { fingerprint: fingerprintStatusPart(remote), value: remote, } satisfies CachedValue<VcsStatusRemoteResult | null>; const shouldPublish = yield* Ref.modify(cacheRef, (cache) => { const previous = cache.get(cwd) ?? { local: null, remote: null }; const nextCache = new Map(cache); nextCache.set(cwd, { ...previous, remote: nextRemote, }); return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const; }); if (options?.publish && shouldPublish) { yield* PubSub.publish(changesPubSub, { cwd, event: { _tag: "remoteUpdated", remote, }, }); } return remote; }, ); const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* ( cwd: string, ) { const local = yield* workflow.localStatus({ cwd }); return yield* updateCachedLocalStatus(cwd, local); }); const loadRemoteStatus = Effect.fn("VcsStatusBroadcaster.loadRemoteStatus")(function* ( cwd: string, ) { const remote = yield* workflow.remoteStatus({ cwd }); return yield* updateCachedRemoteStatus(cwd, remote); }); const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* ( cwd: string, ) { const cached = yield* getCachedStatus(cwd); if (cached?.local) { return cached.local.value; } return yield* loadLocalStatus(cwd); }); const getOrLoadRemoteStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadRemoteStatus")( function* (cwd: string) { const cached = yield* getCachedStatus(cwd); if (cached?.remote) { return cached.remote.value; } return yield* loadRemoteStatus(cwd); }, ); const getStatus: VcsStatusBroadcasterShape["getStatus"] = Effect.fn( "VcsStatusBroadcaster.getStatus", )(function* (input) { const cwd = normalizeCwd(input.cwd); const [local, remote] = yield* Effect.all([ getOrLoadLocalStatus(cwd), getOrLoadRemoteStatus(cwd), ]); return mergeGitStatusParts(local, remote); }); const refreshLocalStatus: VcsStatusBroadcasterShape["refreshLocalStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshLocalStatus", )(function* (rawCwd) { const cwd = normalizeCwd(rawCwd); yield* workflow.invalidateLocalStatus(cwd); const local = yield* workflow.localStatus({ cwd }); return yield* updateCachedLocalStatus(cwd, local, { publish: true }); }); const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* ( cwd: string, ) { yield* workflow.invalidateRemoteStatus(cwd); const remote = yield* workflow.remoteStatus({ cwd }); return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); }); const refreshStatus: VcsStatusBroadcasterShape["refreshStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd) { const cwd = normalizeCwd(rawCwd); const [local, remote] = yield* Effect.all([ refreshLocalStatus(cwd), refreshRemoteStatus(cwd), ]); return mergeGitStatusParts(local, remote); }); const makeRemoteRefreshLoop = (cwd: string) => { const logRefreshFailure = (error: Error) => Effect.logWarning("VCS remote status refresh failed", { cwd, detail: error.message, }); return refreshRemoteStatus(cwd).pipe( Effect.catch(logRefreshFailure), Effect.andThen( Effect.forever( Effect.sleep(VCS_STATUS_REFRESH_INTERVAL).pipe( Effect.andThen(refreshRemoteStatus(cwd).pipe(Effect.catch(logRefreshFailure))), ), ), ), ); }; const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* ( cwd: string, ) { yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => { const existing = activePollers.get(cwd); if (existing) { const nextPollers = new Map(activePollers); nextPollers.set(cwd, { ...existing, subscriberCount: existing.subscriberCount + 1, }); return Effect.succeed([undefined, nextPollers] as const); } return makeRemoteRefreshLoop(cwd).pipe( Effect.forkIn(broadcasterScope), Effect.map((fiber) => { const nextPollers = new Map(activePollers); nextPollers.set(cwd, { fiber, subscriberCount: 1, }); return [undefined, nextPollers] as const; }), ); }); }); const releaseRemotePoller = Effect.fn("VcsStatusBroadcaster.releaseRemotePoller")(function* ( cwd: string, ) { const pollerToInterrupt = yield* SynchronizedRef.modify(pollersRef, (activePollers) => { const existing = activePollers.get(cwd); if (!existing) { return [null, activePollers] as const; } if (existing.subscriberCount > 1) { const nextPollers = new Map(activePollers); nextPollers.set(cwd, { ...existing, subscriberCount: existing.subscriberCount - 1, }); return [null, nextPollers] as const; } const nextPollers = new Map(activePollers); nextPollers.delete(cwd); return [existing.fiber, nextPollers] as const; }); if (pollerToInterrupt) { yield* Fiber.interrupt(pollerToInterrupt).pipe(Effect.ignore); } }); const streamStatus: VcsStatusBroadcasterShape["streamStatus"] = (input) => Stream.unwrap( Effect.gen(function* () { const cwd = normalizeCwd(input.cwd); const subscription = yield* PubSub.subscribe(changesPubSub); const initialLocal = yield* getOrLoadLocalStatus(cwd); const initialRemote = (yield* getCachedStatus(cwd))?.remote?.value ?? null; yield* retainRemotePoller(cwd); const release = releaseRemotePoller(cwd).pipe(Effect.ignore, Effect.asVoid); return Stream.concat( Stream.make({ _tag: "snapshot" as const, local: initialLocal, remote: initialRemote, }), Stream.fromSubscription(subscription).pipe( Stream.filter((event) => event.cwd === cwd), Stream.map((event) => event.event), ), ).pipe(Stream.ensuring(release)); }), ); return VcsStatusBroadcaster.of({ getStatus, refreshLocalStatus, refreshStatus, streamStatus, }); }),);
import { realpathSync } from "node:fs";
import {
Context,
Duration,
Effect,
Exit,
Fiber,
Layer,
PubSub,
Ref,
Scope,
Stream,
SynchronizedRef,
} from "effect";
import type {
GitManagerServiceError,
VcsStatusInput,
VcsStatusLocalResult,
VcsStatusRemoteResult,
VcsStatusResult,
VcsStatusStreamEvent,
} from "@t3tools/contracts";
import { mergeGitStatusParts } from "@t3tools/shared/git";
import * as GitWorkflowService from "../git/GitWorkflowService.ts";
const VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30);
interface VcsStatusChange {
readonly cwd: string;
readonly event: VcsStatusStreamEvent;
}
interface CachedValue<T> {
readonly fingerprint: string;
readonly value: T;
}
interface CachedVcsStatus {
readonly local: CachedValue<VcsStatusLocalResult> | null;
readonly remote: CachedValue<VcsStatusRemoteResult | null> | null;
}
interface ActiveRemotePoller {
readonly fiber: Fiber.Fiber<void, never>;
readonly subscriberCount: number;
}
export interface VcsStatusBroadcasterShape {
readonly getStatus: (
input: VcsStatusInput,
) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;
readonly refreshLocalStatus: (
cwd: string,
) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>;
readonly refreshStatus: (cwd: string) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;
readonly streamStatus: (
input: VcsStatusInput,
) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>;
}
export class VcsStatusBroadcaster extends Context.Service<
VcsStatusBroadcaster,
VcsStatusBroadcasterShape
>()("t3/vcs/VcsStatusBroadcaster") {}
function fingerprintStatusPart(status: unknown): string {
return JSON.stringify(status);
}
function normalizeCwd(cwd: string): string {
try {
return realpathSync.native(cwd);
} catch {
return cwd;
}
}
export const layer = Layer.effect(
VcsStatusBroadcaster,
Effect.gen(function* () {
const workflow = yield* GitWorkflowService.GitWorkflowService;
const changesPubSub = yield* Effect.acquireRelease(
PubSub.unbounded<VcsStatusChange>(),
(pubsub) => PubSub.shutdown(pubsub),
);
const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
Scope.close(scope, Exit.void),
);
const cacheRef = yield* Ref.make(new Map<string, CachedVcsStatus>());
const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>());
const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* (
cwd: string,
) {
return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null));
});
const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")(
function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) {
const nextLocal = {
fingerprint: fingerprintStatusPart(local),
value: local,
} satisfies CachedValue<VcsStatusLocalResult>;
const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {
const previous = cache.get(cwd) ?? { local: null, remote: null };
const nextCache = new Map(cache);
nextCache.set(cwd, {
...previous,
local: nextLocal,
});
return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const;
});
if (options?.publish && shouldPublish) {
yield* PubSub.publish(changesPubSub, {
cwd,
event: {
_tag: "localUpdated",
local,
},
});
}
return local;
},
);
const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")(
function* (
cwd: string,
remote: VcsStatusRemoteResult | null,
options?: { publish?: boolean },
) {
const nextRemote = {
fingerprint: fingerprintStatusPart(remote),
value: remote,
} satisfies CachedValue<VcsStatusRemoteResult | null>;
const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {
const previous = cache.get(cwd) ?? { local: null, remote: null };
const nextCache = new Map(cache);
nextCache.set(cwd, {
...previous,
remote: nextRemote,
});
return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const;
});
if (options?.publish && shouldPublish) {
yield* PubSub.publish(changesPubSub, {
cwd,
event: {
_tag: "remoteUpdated",
remote,
},
});
}
return remote;
},
);
const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (
cwd: string,
) {
const local = yield* workflow.localStatus({ cwd });
return yield* updateCachedLocalStatus(cwd, local);
});
const loadRemoteStatus = Effect.fn("VcsStatusBroadcaster.loadRemoteStatus")(function* (
cwd: string,
) {
const remote = yield* workflow.remoteStatus({ cwd });
return yield* updateCachedRemoteStatus(cwd, remote);
});
const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* (
cwd: string,
) {
const cached = yield* getCachedStatus(cwd);
if (cached?.local) {
return cached.local.value;
}
return yield* loadLocalStatus(cwd);
});
const getOrLoadRemoteStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadRemoteStatus")(
function* (cwd: string) {
const cached = yield* getCachedStatus(cwd);
if (cached?.remote) {
return cached.remote.value;
}
return yield* loadRemoteStatus(cwd);
},
);
const getStatus: VcsStatusBroadcasterShape["getStatus"] = Effect.fn(
"VcsStatusBroadcaster.getStatus",
)(function* (input) {
const cwd = normalizeCwd(input.cwd);
const [local, remote] = yield* Effect.all([
getOrLoadLocalStatus(cwd),
getOrLoadRemoteStatus(cwd),
]);
return mergeGitStatusParts(local, remote);
});
const refreshLocalStatus: VcsStatusBroadcasterShape["refreshLocalStatus"] = Effect.fn(
"VcsStatusBroadcaster.refreshLocalStatus",
)(function* (rawCwd) {
const cwd = normalizeCwd(rawCwd);
yield* workflow.invalidateLocalStatus(cwd);
const local = yield* workflow.localStatus({ cwd });
return yield* updateCachedLocalStatus(cwd, local, { publish: true });
});
const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (
cwd: string,
) {
yield* workflow.invalidateRemoteStatus(cwd);
const remote = yield* workflow.remoteStatus({ cwd });
return yield* updateCachedRemoteStatus(cwd, remote, { publish: true });
});
const refreshStatus: VcsStatusBroadcasterShape["refreshStatus"] = Effect.fn(
"VcsStatusBroadcaster.refreshStatus",
)(function* (rawCwd) {
const cwd = normalizeCwd(rawCwd);
const [local, remote] = yield* Effect.all([
refreshLocalStatus(cwd),
refreshRemoteStatus(cwd),
]);
return mergeGitStatusParts(local, remote);
});
const makeRemoteRefreshLoop = (cwd: string) => {
const logRefreshFailure = (error: Error) =>
Effect.logWarning("VCS remote status refresh failed", {
cwd,
detail: error.message,
});
return refreshRemoteStatus(cwd).pipe(
Effect.catch(logRefreshFailure),
Effect.andThen(
Effect.forever(
Effect.sleep(VCS_STATUS_REFRESH_INTERVAL).pipe(
Effect.andThen(refreshRemoteStatus(cwd).pipe(Effect.catch(logRefreshFailure))),
),
),
),
);
};
const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* (
cwd: string,
) {
yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => {
const existing = activePollers.get(cwd);
if (existing) {
const nextPollers = new Map(activePollers);
nextPollers.set(cwd, {
...existing,
subscriberCount: existing.subscriberCount + 1,
});
return Effect.succeed([undefined, nextPollers] as const);
}
return makeRemoteRefreshLoop(cwd).pipe(
Effect.forkIn(broadcasterScope),
Effect.map((fiber) => {
const nextPollers = new Map(activePollers);
nextPollers.set(cwd, {
fiber,
subscriberCount: 1,
});
return [undefined, nextPollers] as const;
}),
);
});
});
const releaseRemotePoller = Effect.fn("VcsStatusBroadcaster.releaseRemotePoller")(function* (
cwd: string,
) {
const pollerToInterrupt = yield* SynchronizedRef.modify(pollersRef, (activePollers) => {
const existing = activePollers.get(cwd);
if (!existing) {
return [null, activePollers] as const;
}
if (existing.subscriberCount > 1) {
const nextPollers = new Map(activePollers);
nextPollers.set(cwd, {
...existing,
subscriberCount: existing.subscriberCount - 1,
});
return [null, nextPollers] as const;
}
const nextPollers = new Map(activePollers);
nextPollers.delete(cwd);
return [existing.fiber, nextPollers] as const;
});
if (pollerToInterrupt) {
yield* Fiber.interrupt(pollerToInterrupt).pipe(Effect.ignore);
}
});
const streamStatus: VcsStatusBroadcasterShape["streamStatus"] = (input) =>
Stream.unwrap(
Effect.gen(function* () {
const cwd = normalizeCwd(input.cwd);
const subscription = yield* PubSub.subscribe(changesPubSub);
const initialLocal = yield* getOrLoadLocalStatus(cwd);
const initialRemote = (yield* getCachedStatus(cwd))?.remote?.value ?? null;
yield* retainRemotePoller(cwd);
const release = releaseRemotePoller(cwd).pipe(Effect.ignore, Effect.asVoid);
return Stream.concat(
Stream.make({
_tag: "snapshot" as const,
local: initialLocal,
remote: initialRemote,
}),
Stream.fromSubscription(subscription).pipe(
Stream.filter((event) => event.cwd === cwd),
Stream.map((event) => event.event),
),
).pipe(Stream.ensuring(release));
}),
);
return VcsStatusBroadcaster.of({
getStatus,
refreshLocalStatus,
refreshStatus,
streamStatus,
});
}),
);