import * as Context from "effect/Context";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as PubSub from "effect/PubSub";import * as Ref from "effect/Ref";import * as Schedule from "effect/Schedule";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as SynchronizedRef from "effect/SynchronizedRef";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 DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30);const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30);const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15); 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;} interface StreamStatusOptions { readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>;} export function remoteRefreshFailureDelay( consecutiveFailures: number, configuredInterval: Duration.Duration,) { const exponent = Math.max(0, consecutiveFailures - 1); const backoffMs = Duration.toMillis(VCS_STATUS_REFRESH_FAILURE_BASE_DELAY) * Math.pow(2, exponent); const cappedBackoff = Duration.min( Duration.millis(backoffMs), VCS_STATUS_REFRESH_FAILURE_MAX_DELAY, ); return Duration.max(configuredInterval, cappedBackoff);} export class VcsStatusBroadcaster extends Context.Service< VcsStatusBroadcaster, { 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, options?: StreamStatusOptions, ) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>; }>()("t3/vcs/VcsStatusBroadcaster") {} function fingerprintStatusPart(status: unknown): string { return JSON.stringify(status);} const normalizeCwd = (cwd: string) => Effect.service(FileSystem.FileSystem).pipe( Effect.flatMap((fs) => fs.realPath(cwd)), Effect.orElseSucceed(() => cwd), ); export const make = Effect.gen(function* () { const workflow = yield* GitWorkflowService.GitWorkflowService; const fs = yield* FileSystem.FileSystem; 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 updateCachedStatus = Effect.fn("VcsStatusBroadcaster.updateCachedStatus")(function* ( cwd: string, local: VcsStatusLocalResult, remote: VcsStatusRemoteResult | null, options?: { publish?: boolean }, ) { const nextLocal = { fingerprint: fingerprintStatusPart(local), value: local, } satisfies CachedValue<VcsStatusLocalResult>; 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, { local: nextLocal, remote: nextRemote, }); return [ previous.local?.fingerprint !== nextLocal.fingerprint || previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache, ] as const; }); if (options?.publish && shouldPublish) { yield* PubSub.publish(changesPubSub, { cwd, event: { _tag: "snapshot", local, remote, }, }); } return mergeGitStatusParts(local, remote); }); const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* ( cwd: string, ) { const local = yield* workflow.localStatus({ cwd }); return yield* updateCachedLocalStatus(cwd, local); }); 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 withFileSystem = Effect.provideService(FileSystem.FileSystem, fs); const getStatus: VcsStatusBroadcaster["Service"]["getStatus"] = Effect.fn( "VcsStatusBroadcaster.getStatus", )(function* (input) { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const cached = yield* getCachedStatus(cwd); if (cached?.local && cached.remote) { return mergeGitStatusParts(cached.local.value, cached.remote.value); } const [local, remote] = yield* Effect.all( [ cached?.local ? Effect.succeed(cached.local.value) : workflow.localStatus({ cwd }), cached?.remote ? Effect.succeed(cached.remote.value) : workflow.remoteStatus({ cwd }), ], { concurrency: "unbounded" }, ); return yield* updateCachedStatus(cwd, local, remote); }); const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")( function* (cwd: string) { yield* workflow.invalidateLocalStatus(cwd); const local = yield* workflow.localStatus({ cwd }); return yield* updateCachedLocalStatus(cwd, local, { publish: true }); }, ); const refreshLocalStatus: VcsStatusBroadcaster["Service"]["refreshLocalStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshLocalStatus", )(function* (rawCwd) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); return yield* refreshLocalStatusCore(cwd); }); const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* ( cwd: string, options?: { readonly refreshUpstream?: boolean }, ) { if (options?.refreshUpstream !== false) { yield* workflow.invalidateRemoteStatus(cwd); } const remote = yield* workflow.remoteStatus({ cwd }, options); return yield* updateCachedRemoteStatus(cwd, remote, { publish: true }); }); const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshStatus", )(function* (rawCwd) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); yield* Effect.all([workflow.invalidateLocalStatus(cwd), workflow.invalidateRemoteStatus(cwd)], { concurrency: "unbounded", discard: true, }); const [local, remote] = yield* Effect.all( [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })], { concurrency: "unbounded" }, ); return yield* updateCachedStatus(cwd, local, remote, { publish: true }); }); const makeRemoteRefreshLoop = ( cwd: string, automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>, refreshImmediately: boolean, ) => { return Effect.gen(function* () { const consecutiveFailuresRef = yield* Ref.make(0); const needsInitialRefreshRef = yield* Ref.make(refreshImmediately); const refreshRemoteStatusIfEnabled = Effect.gen(function* () { const configuredInterval = yield* automaticRemoteRefreshInterval; const activeInterval = Duration.isZero(configuredInterval) ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL : configuredInterval; const needsInitialRefresh = yield* Ref.get(needsInitialRefreshRef); if (Duration.isZero(configuredInterval) && !needsInitialRefresh) { return activeInterval; } const exit = yield* refreshRemoteStatus(cwd, { refreshUpstream: !Duration.isZero(configuredInterval), }).pipe(Effect.exit); if (Exit.isSuccess(exit)) { yield* Ref.set(needsInitialRefreshRef, false); yield* Ref.set(consecutiveFailuresRef, 0); return activeInterval; } const consecutiveFailures = yield* Ref.updateAndGet( consecutiveFailuresRef, (count) => count + 1, ); const nextDelay = remoteRefreshFailureDelay(consecutiveFailures, activeInterval); yield* Effect.logWarning("VCS remote status refresh failed", { cwd, detail: exit.cause.toString(), consecutiveFailures, nextDelayMs: Duration.toMillis(nextDelay), }); return nextDelay; }); if (!refreshImmediately) { const configuredInterval = yield* automaticRemoteRefreshInterval; yield* Effect.sleep( Duration.isZero(configuredInterval) ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL : configuredInterval, ); } return yield* refreshRemoteStatusIfEnabled.pipe( Effect.repeat( Schedule.identity<Duration.Duration>().pipe( Schedule.addDelay((delay) => Effect.succeed(delay)), ), ), Effect.asVoid, ); }); }; const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* ( cwd: string, automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>, refreshImmediately: boolean, ) { 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, automaticRemoteRefreshInterval, refreshImmediately).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: VcsStatusBroadcaster["Service"]["streamStatus"] = (input, options) => Stream.unwrap( Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); const subscription = yield* PubSub.subscribe(changesPubSub); const initialLocal = yield* getOrLoadLocalStatus(cwd); const cachedStatus = yield* getCachedStatus(cwd); const initialRemote = cachedStatus?.remote?.value ?? null; yield* retainRemotePoller( cwd, options?.automaticRemoteRefreshInterval ?? Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL), cachedStatus?.remote === null || cachedStatus?.remote === undefined, ); 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, });}); export const layer = Layer.effect(VcsStatusBroadcaster, make);
import * as Context from "effect/Context";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as PubSub from "effect/PubSub";
import * as Ref from "effect/Ref";
import * as Schedule from "effect/Schedule";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import * as SynchronizedRef from "effect/SynchronizedRef";
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 DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30);
const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30);
const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15);
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;
}
interface StreamStatusOptions {
readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>;
}
export function remoteRefreshFailureDelay(
consecutiveFailures: number,
configuredInterval: Duration.Duration,
) {
const exponent = Math.max(0, consecutiveFailures - 1);
const backoffMs =
Duration.toMillis(VCS_STATUS_REFRESH_FAILURE_BASE_DELAY) * Math.pow(2, exponent);
const cappedBackoff = Duration.min(
Duration.millis(backoffMs),
VCS_STATUS_REFRESH_FAILURE_MAX_DELAY,
);
return Duration.max(configuredInterval, cappedBackoff);
}
export class VcsStatusBroadcaster extends Context.Service<
VcsStatusBroadcaster,
{
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,
options?: StreamStatusOptions,
) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>;
}
>()("t3/vcs/VcsStatusBroadcaster") {}
function fingerprintStatusPart(status: unknown): string {
return JSON.stringify(status);
}
const normalizeCwd = (cwd: string) =>
Effect.service(FileSystem.FileSystem).pipe(
Effect.flatMap((fs) => fs.realPath(cwd)),
Effect.orElseSucceed(() => cwd),
);
export const make = Effect.gen(function* () {
const workflow = yield* GitWorkflowService.GitWorkflowService;
const fs = yield* FileSystem.FileSystem;
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 updateCachedStatus = Effect.fn("VcsStatusBroadcaster.updateCachedStatus")(function* (
cwd: string,
local: VcsStatusLocalResult,
remote: VcsStatusRemoteResult | null,
options?: { publish?: boolean },
) {
const nextLocal = {
fingerprint: fingerprintStatusPart(local),
value: local,
} satisfies CachedValue<VcsStatusLocalResult>;
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, {
local: nextLocal,
remote: nextRemote,
});
return [
previous.local?.fingerprint !== nextLocal.fingerprint ||
previous.remote?.fingerprint !== nextRemote.fingerprint,
nextCache,
] as const;
});
if (options?.publish && shouldPublish) {
yield* PubSub.publish(changesPubSub, {
cwd,
event: {
_tag: "snapshot",
local,
remote,
},
});
}
return mergeGitStatusParts(local, remote);
});
const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (
cwd: string,
) {
const local = yield* workflow.localStatus({ cwd });
return yield* updateCachedLocalStatus(cwd, local);
});
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 withFileSystem = Effect.provideService(FileSystem.FileSystem, fs);
const getStatus: VcsStatusBroadcaster["Service"]["getStatus"] = Effect.fn(
"VcsStatusBroadcaster.getStatus",
)(function* (input) {
const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
const cached = yield* getCachedStatus(cwd);
if (cached?.local && cached.remote) {
return mergeGitStatusParts(cached.local.value, cached.remote.value);
}
const [local, remote] = yield* Effect.all(
[
cached?.local ? Effect.succeed(cached.local.value) : workflow.localStatus({ cwd }),
cached?.remote ? Effect.succeed(cached.remote.value) : workflow.remoteStatus({ cwd }),
],
{ concurrency: "unbounded" },
);
return yield* updateCachedStatus(cwd, local, remote);
});
const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")(
function* (cwd: string) {
yield* workflow.invalidateLocalStatus(cwd);
const local = yield* workflow.localStatus({ cwd });
return yield* updateCachedLocalStatus(cwd, local, { publish: true });
},
);
const refreshLocalStatus: VcsStatusBroadcaster["Service"]["refreshLocalStatus"] = Effect.fn(
"VcsStatusBroadcaster.refreshLocalStatus",
)(function* (rawCwd) {
const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
return yield* refreshLocalStatusCore(cwd);
});
const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (
cwd: string,
options?: { readonly refreshUpstream?: boolean },
) {
if (options?.refreshUpstream !== false) {
yield* workflow.invalidateRemoteStatus(cwd);
}
const remote = yield* workflow.remoteStatus({ cwd }, options);
return yield* updateCachedRemoteStatus(cwd, remote, { publish: true });
});
const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn(
"VcsStatusBroadcaster.refreshStatus",
)(function* (rawCwd) {
const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
yield* Effect.all([workflow.invalidateLocalStatus(cwd), workflow.invalidateRemoteStatus(cwd)], {
concurrency: "unbounded",
discard: true,
});
const [local, remote] = yield* Effect.all(
[workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })],
{ concurrency: "unbounded" },
);
return yield* updateCachedStatus(cwd, local, remote, { publish: true });
});
const makeRemoteRefreshLoop = (
cwd: string,
automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>,
refreshImmediately: boolean,
) => {
return Effect.gen(function* () {
const consecutiveFailuresRef = yield* Ref.make(0);
const needsInitialRefreshRef = yield* Ref.make(refreshImmediately);
const refreshRemoteStatusIfEnabled = Effect.gen(function* () {
const configuredInterval = yield* automaticRemoteRefreshInterval;
const activeInterval = Duration.isZero(configuredInterval)
? DEFAULT_VCS_STATUS_REFRESH_INTERVAL
: configuredInterval;
const needsInitialRefresh = yield* Ref.get(needsInitialRefreshRef);
if (Duration.isZero(configuredInterval) && !needsInitialRefresh) {
return activeInterval;
}
const exit = yield* refreshRemoteStatus(cwd, {
refreshUpstream: !Duration.isZero(configuredInterval),
}).pipe(Effect.exit);
if (Exit.isSuccess(exit)) {
yield* Ref.set(needsInitialRefreshRef, false);
yield* Ref.set(consecutiveFailuresRef, 0);
return activeInterval;
}
const consecutiveFailures = yield* Ref.updateAndGet(
consecutiveFailuresRef,
(count) => count + 1,
);
const nextDelay = remoteRefreshFailureDelay(consecutiveFailures, activeInterval);
yield* Effect.logWarning("VCS remote status refresh failed", {
cwd,
detail: exit.cause.toString(),
consecutiveFailures,
nextDelayMs: Duration.toMillis(nextDelay),
});
return nextDelay;
});
if (!refreshImmediately) {
const configuredInterval = yield* automaticRemoteRefreshInterval;
yield* Effect.sleep(
Duration.isZero(configuredInterval)
? DEFAULT_VCS_STATUS_REFRESH_INTERVAL
: configuredInterval,
);
}
return yield* refreshRemoteStatusIfEnabled.pipe(
Effect.repeat(
Schedule.identity<Duration.Duration>().pipe(
Schedule.addDelay((delay) => Effect.succeed(delay)),
),
),
Effect.asVoid,
);
});
};
const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* (
cwd: string,
automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>,
refreshImmediately: boolean,
) {
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, automaticRemoteRefreshInterval, refreshImmediately).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: VcsStatusBroadcaster["Service"]["streamStatus"] = (input, options) =>
Stream.unwrap(
Effect.gen(function* () {
const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
const subscription = yield* PubSub.subscribe(changesPubSub);
const initialLocal = yield* getOrLoadLocalStatus(cwd);
const cachedStatus = yield* getCachedStatus(cwd);
const initialRemote = cachedStatus?.remote?.value ?? null;
yield* retainRemotePoller(
cwd,
options?.automaticRemoteRefreshInterval ??
Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL),
cachedStatus?.remote === null || cachedStatus?.remote === undefined,
);
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,
});
});
export const layer = Layer.effect(VcsStatusBroadcaster, make);