1import * as Cause from "effect/Cause";
2import * as Context from "effect/Context";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as FileSystem from "effect/FileSystem";
8import * as Layer from "effect/Layer";
9import * as PubSub from "effect/PubSub";
10import * as Ref from "effect/Ref";
11import * as Schedule from "effect/Schedule";
12import * as Scope from "effect/Scope";
13import * as Stream from "effect/Stream";
14import * as SynchronizedRef from "effect/SynchronizedRef";
15import type {
16 GitManagerServiceError,
17 VcsStatusInput,
18 VcsStatusLocalResult,
19 VcsStatusRemoteResult,
20 VcsStatusResult,
21 VcsStatusStreamEvent,
22 VcsStatusSubscriptionInput,
23} from "@t3tools/contracts";
24import { mergeGitStatusParts } from "@t3tools/shared/git";
25import * as KeyedLock from "@t3tools/shared/KeyedLock";
26import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
27
28import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
29import * as GitWorkflowService from "../git/GitWorkflowService.ts";
30import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
31import * as ServerSettings from "../serverSettings.ts";
32
33const DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30);
34const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30);
35const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15);
36const MAX_FAILURE_DIAGNOSTIC_VALUES = 8;
37const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128;
38
39function boundedDiagnosticValue(value: string): string {
40 return value.slice(0, MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH);
41}
42
43function diagnosticValueTag(value: unknown): string {
44 try {
45 if (
46 typeof value === "object" &&
47 value !== null &&
48 "_tag" in value &&
49 typeof value._tag === "string"
50 ) {
51 return boundedDiagnosticValue(value._tag);
52 }
53 if (value instanceof Error) {
54 return boundedDiagnosticValue(value.name);
55 }
56 return typeof value;
57 } catch {
58 return "Uninspectable";
59 }
60}
61
62function diagnosticFailureOperation(value: unknown): string | undefined {
63 try {
64 if (
65 typeof value === "object" &&
66 value !== null &&
67 "operation" in value &&
68 typeof value.operation === "string"
69 ) {
70 return boundedDiagnosticValue(value.operation);
71 }
72 } catch {
73 return undefined;
74 }
75 return undefined;
76}
77
78function addUniqueDiagnosticValue(values: Array<string>, value: string | undefined): void {
79 if (
80 value !== undefined &&
81 values.length < MAX_FAILURE_DIAGNOSTIC_VALUES &&
82 !values.includes(value)
83 ) {
84 values.push(value);
85 }
86}
87
88export function remoteRefreshFailureDiagnostics(cause: Cause.Cause<unknown>) {
89 const failureTags: Array<string> = [];
90 const failureOperations: Array<string> = [];
91 const defectTags: Array<string> = [];
92 let failureCount = 0;
93 let defectCount = 0;
94 let interruptionCount = 0;
95
96 for (const reason of cause.reasons) {
97 if (Cause.isFailReason(reason)) {
98 failureCount += 1;
99 addUniqueDiagnosticValue(failureTags, diagnosticValueTag(reason.error));
100 addUniqueDiagnosticValue(failureOperations, diagnosticFailureOperation(reason.error));
101 continue;
102 }
103 if (Cause.isDieReason(reason)) {
104 defectCount += 1;
105 addUniqueDiagnosticValue(defectTags, diagnosticValueTag(reason.defect));
106 continue;
107 }
108 interruptionCount += 1;
109 }
110
111 return {
112 reasonCount: cause.reasons.length,
113 failureCount,
114 failureTags,
115 failureOperations,
116 defectCount,
117 defectTags,
118 interruptionCount,
119 };
120}
121
122interface VcsStatusChange {
123 readonly cwd: string;
124 readonly event: VcsStatusStreamEvent;
125}
126
127interface CachedValue<T> {
128 readonly fingerprint: string;
129 readonly value: T;
130}
131
132interface CachedVcsStatus {
133 readonly local: CachedValue<VcsStatusLocalResult> | null;
134 readonly remote: CachedValue<VcsStatusRemoteResult | null> | null;
135}
136
137interface ActiveRemotePoller {
138 readonly fiber: Fiber.Fiber<void, never>;
139 readonly subscriberCount: number;
140 readonly demandCwds: Ref.Ref<ReadonlyMap<string, number>>;
141}
142
143interface StreamStatusOptions {
144 readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>;
145}
146
147export class VcsAutoPullPolicy extends Context.Reference<{
148 readonly isEnabled: (cwd: string) => Effect.Effect<boolean, never>;
149}>("t3/vcs/VcsAutoPullPolicy", {
150 defaultValue: () => ({ isEnabled: () => Effect.succeed(false) }),
151}) {}
152
153export const layerAutoPullPolicy = Layer.effect(
154 VcsAutoPullPolicy,
155 Effect.gen(function* () {
156 const projects = yield* ProjectStore.ProjectStoreV2;
157 const serverSettings = yield* ServerSettings.ServerSettingsService;
158 return {
159 isEnabled: Effect.fn("VcsAutoPullPolicy.isEnabled")(
160 function* (cwd: string) {
161 const project = yield* projects.findActiveByWorkspaceRoot(cwd);
162 if (project._tag === "None") return false;
163 const settings = yield* serverSettings.getSettings;
164 return resolveProjectSettings(settings, project.value.projectId).settings.defaultAutoPull;
165 },
166 Effect.orElseSucceed(() => false),
167 ),
168 };
169 }),
170);
171
172export function remoteRefreshFailureDelay(
173 consecutiveFailures: number,
174 configuredInterval: Duration.Duration,
175) {
176 const exponent = Math.max(0, consecutiveFailures - 1);
177 const backoffMs =
178 Duration.toMillis(VCS_STATUS_REFRESH_FAILURE_BASE_DELAY) * Math.pow(2, exponent);
179 const cappedBackoff = Duration.min(
180 Duration.millis(backoffMs),
181 VCS_STATUS_REFRESH_FAILURE_MAX_DELAY,
182 );
183 return Duration.max(configuredInterval, cappedBackoff);
184}
185
186export class VcsStatusBroadcaster extends Context.Service<
187 VcsStatusBroadcaster,
188 {
189 readonly getStatus: (
190 input: VcsStatusInput,
191 ) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;
192 readonly refreshLocalStatus: (
193 cwd: string,
194 ) => Effect.Effect<VcsStatusLocalResult, GitManagerServiceError>;
195 readonly refreshStatus: (cwd: string) => Effect.Effect<VcsStatusResult, GitManagerServiceError>;
196 /**
197 * Refresh a loaded cwd after a turn if background policy allows it.
198 * GitManager retries missing PRs for the current branch and keeps known
199 * PRs and failed lookup backoff cached. This does not fetch Git remotes.
200 */
201 readonly refreshPullRequestStatus: (
202 cwd: string,
203 ) => Effect.Effect<VcsStatusRemoteResult | null, GitManagerServiceError>;
204 readonly streamStatus: (
205 input: VcsStatusSubscriptionInput,
206 options?: StreamStatusOptions,
207 ) => Stream.Stream<VcsStatusStreamEvent, GitManagerServiceError>;
208 }
209>()("t3/vcs/VcsStatusBroadcaster") {}
210
211function fingerprintStatusPart(status: unknown): string {
212 return JSON.stringify(status);
213}
214
215const normalizeCwd = (cwd: string) =>
216 Effect.service(FileSystem.FileSystem).pipe(
217 Effect.flatMap((fs) => fs.realPath(cwd)),
218 Effect.orElseSucceed(() => cwd),
219 );
220
221/** @public Service construction is part of the canonical Effect module API. */
222export const make = Effect.gen(function* () {
223 const autoPullPolicy = yield* VcsAutoPullPolicy;
224 const workflow = yield* GitWorkflowService.GitWorkflowService;
225 const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
226 const fs = yield* FileSystem.FileSystem;
227 const changesPubSub = yield* Effect.acquireRelease(
228 PubSub.unbounded<VcsStatusChange>(),
229 (pubsub) => PubSub.shutdown(pubsub),
230 );
231 const broadcasterScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
232 Scope.close(scope, Exit.void),
233 );
234 const cacheRef = yield* Ref.make(new Map<string, CachedVcsStatus>());
235 // One permit per cwd for remote reads that write the cache. Without it a
236 // periodic poll that started before `gh pr create` can finish after the
237 // turn-end refresh and overwrite the fresh PR with its stale `pr: null`.
238 const remoteWriteLocks = yield* KeyedLock.make<string>();
239 const withRemoteWriteLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) =>
240 remoteWriteLocks.withLock(cwd, effect);
241 const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>());
242
243 const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* (
244 cwd: string,
245 ) {
246 return yield* Ref.get(cacheRef).pipe(Effect.map((cache) => cache.get(cwd) ?? null));
247 });
248
249 const updateCachedLocalStatus = Effect.fn("VcsStatusBroadcaster.updateCachedLocalStatus")(
250 function* (cwd: string, local: VcsStatusLocalResult, options?: { publish?: boolean }) {
251 const nextLocal = {
252 fingerprint: fingerprintStatusPart(local),
253 value: local,
254 } satisfies CachedValue<VcsStatusLocalResult>;
255 const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {
256 const previous = cache.get(cwd) ?? { local: null, remote: null };
257 const nextCache = new Map(cache);
258 nextCache.set(cwd, {
259 ...previous,
260 local: nextLocal,
261 });
262 return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const;
263 });
264
265 if (options?.publish && shouldPublish) {
266 yield* PubSub.publish(changesPubSub, {
267 cwd,
268 event: {
269 _tag: "localUpdated",
270 local,
271 },
272 });
273 }
274
275 return local;
276 },
277 );
278
279 const updateCachedRemoteStatus = Effect.fn("VcsStatusBroadcaster.updateCachedRemoteStatus")(
280 function* (cwd: string, remote: VcsStatusRemoteResult | null, options?: { publish?: boolean }) {
281 const nextRemote = {
282 fingerprint: fingerprintStatusPart(remote),
283 value: remote,
284 } satisfies CachedValue<VcsStatusRemoteResult | null>;
285 const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {
286 const previous = cache.get(cwd) ?? { local: null, remote: null };
287 const nextCache = new Map(cache);
288 nextCache.set(cwd, {
289 ...previous,
290 remote: nextRemote,
291 });
292 return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const;
293 });
294
295 if (options?.publish && shouldPublish) {
296 yield* PubSub.publish(changesPubSub, {
297 cwd,
298 event: {
299 _tag: "remoteUpdated",
300 remote,
301 },
302 });
303 }
304
305 return remote;
306 },
307 );
308
309 const updateCachedStatus = Effect.fn("VcsStatusBroadcaster.updateCachedStatus")(function* (
310 cwd: string,
311 local: VcsStatusLocalResult,
312 remote: VcsStatusRemoteResult | null,
313 options?: { publish?: boolean },
314 ) {
315 const nextLocal = {
316 fingerprint: fingerprintStatusPart(local),
317 value: local,
318 } satisfies CachedValue<VcsStatusLocalResult>;
319 const nextRemote = {
320 fingerprint: fingerprintStatusPart(remote),
321 value: remote,
322 } satisfies CachedValue<VcsStatusRemoteResult | null>;
323 const shouldPublish = yield* Ref.modify(cacheRef, (cache) => {
324 const previous = cache.get(cwd) ?? { local: null, remote: null };
325 const nextCache = new Map(cache);
326 nextCache.set(cwd, {
327 local: nextLocal,
328 remote: nextRemote,
329 });
330 return [
331 previous.local?.fingerprint !== nextLocal.fingerprint ||
332 previous.remote?.fingerprint !== nextRemote.fingerprint,
333 nextCache,
334 ] as const;
335 });
336
337 if (options?.publish && shouldPublish) {
338 yield* PubSub.publish(changesPubSub, {
339 cwd,
340 event: {
341 _tag: "snapshot",
342 local,
343 remote,
344 },
345 });
346 }
347
348 return mergeGitStatusParts(local, remote);
349 });
350
351 const loadLocalStatus = Effect.fn("VcsStatusBroadcaster.loadLocalStatus")(function* (
352 cwd: string,
353 ) {
354 const local = yield* workflow.localStatus({ cwd });
355 return yield* updateCachedLocalStatus(cwd, local);
356 });
357
358 const getOrLoadLocalStatus = Effect.fn("VcsStatusBroadcaster.getOrLoadLocalStatus")(function* (
359 cwd: string,
360 ) {
361 const cached = yield* getCachedStatus(cwd);
362 if (cached?.local) {
363 return cached.local.value;
364 }
365 return yield* loadLocalStatus(cwd);
366 });
367
368 const withFileSystem = Effect.provideService(FileSystem.FileSystem, fs);
369
370 const getStatus: VcsStatusBroadcaster["Service"]["getStatus"] = Effect.fn(
371 "VcsStatusBroadcaster.getStatus",
372 )(function* (input) {
373 const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
374 const cached = yield* getCachedStatus(cwd);
375 if (cached?.local && cached.remote) {
376 return mergeGitStatusParts(cached.local.value, cached.remote.value);
377 }
378 return yield* withRemoteWriteLock(
379 cwd,
380 Effect.gen(function* () {
381 const latest = yield* getCachedStatus(cwd);
382 const [local, remote] = yield* Effect.all(
383 [
384 latest?.local ? Effect.succeed(latest.local.value) : workflow.localStatus({ cwd }),
385 latest?.remote ? Effect.succeed(latest.remote.value) : workflow.remoteStatus({ cwd }),
386 ],
387 { concurrency: "unbounded" },
388 );
389 return yield* updateCachedStatus(cwd, local, remote);
390 }),
391 );
392 });
393
394 const refreshLocalStatusCore = Effect.fn("VcsStatusBroadcaster.refreshLocalStatusCore")(
395 function* (cwd: string) {
396 yield* workflow.invalidateLocalStatus(cwd);
397 const local = yield* workflow.localStatus({ cwd });
398 return yield* updateCachedLocalStatus(cwd, local, { publish: true });
399 },
400 );
401
402 const refreshLocalStatus: VcsStatusBroadcaster["Service"]["refreshLocalStatus"] = Effect.fn(
403 "VcsStatusBroadcaster.refreshLocalStatus",
404 )(function* (rawCwd) {
405 const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
406 return yield* refreshLocalStatusCore(cwd);
407 });
408
409 const maybeAutoPull = Effect.fn("VcsStatusBroadcaster.maybeAutoPull")(function* (
410 cwd: string,
411 remote: VcsStatusRemoteResult | null,
412 policyCwds: ReadonlyArray<string>,
413 ) {
414 return yield* Effect.gen(function* () {
415 const autoPullEnabled = (yield* Effect.forEach(policyCwds, autoPullPolicy.isEnabled, {
416 concurrency: "unbounded",
417 })).some(Boolean);
418 if (
419 remote === null ||
420 !remote.hasUpstream ||
421 remote.aheadCount > 0 ||
422 remote.behindCount <= 0 ||
423 !autoPullEnabled
424 ) {
425 return null;
426 }
427
428 yield* workflow.invalidateLocalStatus(cwd);
429 const local = yield* workflow.localStatus({ cwd });
430 if (!local.isRepo || !local.isDefaultRef || local.hasWorkingTreeChanges) return null;
431
432 yield* workflow.pullCurrentBranch(cwd);
433 yield* workflow.invalidateStatus(cwd);
434 const [refreshedLocal, refreshedRemote] = yield* Effect.all(
435 [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd }, { refreshUpstream: false })],
436 { concurrency: "unbounded" },
437 );
438 yield* updateCachedStatus(cwd, refreshedLocal, refreshedRemote, { publish: true });
439 return { local: refreshedLocal, remote: refreshedRemote };
440 }).pipe(
441 Effect.catch(() =>
442 Effect.logWarning("Automatic project pull failed", { cwd }).pipe(Effect.as(null)),
443 ),
444 );
445 });
446
447 const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (
448 cwd: string,
449 options?: {
450 readonly refreshUpstream?: boolean;
451 readonly policyCwds?: ReadonlyArray<string>;
452 },
453 ) {
454 return yield* withRemoteWriteLock(
455 cwd,
456 Effect.gen(function* () {
457 if (options?.refreshUpstream !== false) {
458 yield* workflow.invalidateRemoteStatus(cwd);
459 }
460 const previousRemote = (yield* getCachedStatus(cwd))?.remote?.value;
461 const remote = yield* workflow.remoteStatus({ cwd }, options);
462 const pulled = yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]);
463 if (pulled !== null) return pulled.remote;
464 // Local status holds the Changes totals, which compare against remote refs. A fetch can
465 // move them with no local trigger (a push from a terminal, a PR merged on the host), so
466 // re-read local status on the first fetch and whenever divergence moves.
467 if (
468 remote &&
469 (!previousRemote ||
470 previousRemote.aheadCount !== remote.aheadCount ||
471 previousRemote.behindCount !== remote.behindCount ||
472 previousRemote.aheadOfDefaultCount !== remote.aheadOfDefaultCount)
473 ) {
474 yield* refreshLocalStatusCore(cwd);
475 }
476 return yield* updateCachedRemoteStatus(cwd, remote, { publish: true });
477 }),
478 );
479 });
480
481 const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn(
482 "VcsStatusBroadcaster.refreshStatus",
483 )(function* (rawCwd) {
484 const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
485 // invalidateStatus (not the two partial invalidations) so an explicit
486 // refresh also bypasses GitManager's slow PR-lookup cache.
487 return yield* withRemoteWriteLock(
488 cwd,
489 Effect.gen(function* () {
490 yield* workflow.invalidateStatus(cwd);
491 // Local after remote: the fetch can move the base that the Changes totals compare with.
492 const remote = yield* workflow.remoteStatus({ cwd });
493 const local = yield* workflow.localStatus({ cwd });
494 const pulled = yield* maybeAutoPull(cwd, remote, [rawCwd]);
495 if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote);
496 return yield* updateCachedStatus(cwd, local, remote, { publish: true });
497 }),
498 );
499 });
500
501 const refreshPullRequestStatus: VcsStatusBroadcaster["Service"]["refreshPullRequestStatus"] =
502 Effect.fn("VcsStatusBroadcaster.refreshPullRequestStatus")(function* (rawCwd) {
503 const cwd = yield* withFileSystem(normalizeCwd(rawCwd));
504 return yield* withRemoteWriteLock(
505 cwd,
506 Effect.gen(function* () {
507 const cached = yield* getCachedStatus(cwd);
508 if (cached?.remote?.value == null) return null;
509 const poller = (yield* SynchronizedRef.get(pollersRef)).get(cwd);
510 const demandCwds = poller ? [...(yield* Ref.get(poller.demandCwds)).keys()] : [rawCwd];
511 const shouldRefresh = (yield* Effect.forEach(
512 demandCwds,
513 (demandCwd) =>
514 backgroundPolicy.shouldRunScopeWork({ type: "vcs-status", cwd: demandCwd }),
515 { concurrency: "unbounded" },
516 )).some(Boolean);
517 if (!shouldRefresh) return null;
518 // Resolve the checked-out branch again. A cached PR can belong to
519 // the previous branch after an agent checks out another branch.
520 const remote = yield* workflow.remoteStatus(
521 { cwd },
522 { refreshUpstream: false, refreshMissingPullRequest: true },
523 );
524 return yield* updateCachedRemoteStatus(cwd, remote, { publish: true });
525 }),
526 );
527 });
528
529 const makeRemoteRefreshLoop = (
530 cwd: string,
531 demandCwdsRef: Ref.Ref<ReadonlyMap<string, number>>,
532 automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>,
533 refreshImmediately: boolean,
534 ) => {
535 return Effect.gen(function* () {
536 const consecutiveFailuresRef = yield* Ref.make(0);
537 const needsInitialRefreshRef = yield* Ref.make(refreshImmediately);
538 const refreshRemoteStatusIfEnabled = Effect.gen(function* () {
539 const configuredInterval = yield* automaticRemoteRefreshInterval;
540 const activeInterval = Duration.isZero(configuredInterval)
541 ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL
542 : configuredInterval;
543 const needsInitialRefresh = yield* Ref.get(needsInitialRefreshRef);
544 if (Duration.isZero(configuredInterval) && !needsInitialRefresh) {
545 return activeInterval;
546 }
547
548 const demandCwds = yield* Ref.get(demandCwdsRef);
549 const shouldRun =
550 needsInitialRefresh ||
551 (yield* Effect.forEach(
552 [...demandCwds.keys()],
553 (demandCwd) =>
554 backgroundPolicy.shouldRunScopeWork({
555 type: "vcs-status",
556 cwd: demandCwd,
557 }),
558 { concurrency: "unbounded" },
559 )).some(Boolean);
560 if (!shouldRun) {
561 return activeInterval;
562 }
563
564 const exit = yield* refreshRemoteStatus(cwd, {
565 refreshUpstream: !Duration.isZero(configuredInterval),
566 policyCwds: [...demandCwds.keys()],
567 }).pipe(Effect.exit);
568 if (Exit.isSuccess(exit)) {
569 yield* Ref.set(needsInitialRefreshRef, false);
570 yield* Ref.set(consecutiveFailuresRef, 0);
571 return activeInterval;
572 }
573
574 const interruptionReasons = exit.cause.reasons.filter(Cause.isInterruptReason);
575 if (interruptionReasons.length > 0) {
576 return yield* Effect.failCause(Cause.fromReasons<never>(interruptionReasons));
577 }
578
579 const consecutiveFailures = yield* Ref.updateAndGet(
580 consecutiveFailuresRef,
581 (count) => count + 1,
582 );
583 const nextDelay = remoteRefreshFailureDelay(consecutiveFailures, activeInterval);
584 yield* Effect.logWarning("VCS remote status refresh failed", {
585 cwdLength: cwd.length,
586 ...remoteRefreshFailureDiagnostics(exit.cause),
587 consecutiveFailures,
588 nextDelayMs: Duration.toMillis(nextDelay),
589 });
590 return nextDelay;
591 });
592
593 if (!refreshImmediately) {
594 const configuredInterval = yield* automaticRemoteRefreshInterval;
595 yield* Effect.sleep(
596 Duration.isZero(configuredInterval)
597 ? DEFAULT_VCS_STATUS_REFRESH_INTERVAL
598 : configuredInterval,
599 );
600 }
601
602 return yield* refreshRemoteStatusIfEnabled.pipe(
603 Effect.repeat(
604 Schedule.identity<Duration.Duration>().pipe(
605 Schedule.addDelay(({ output: delay }) => Effect.succeed(delay)),
606 ),
607 ),
608 Effect.asVoid,
609 );
610 });
611 };
612
613 const retainRemotePoller = Effect.fn("VcsStatusBroadcaster.retainRemotePoller")(function* (
614 cwd: string,
615 demandCwd: string,
616 automaticRemoteRefreshInterval: Effect.Effect<Duration.Duration, never>,
617 refreshImmediately: boolean,
618 ) {
619 yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => {
620 const existing = activePollers.get(cwd);
621 if (existing) {
622 return Ref.update(existing.demandCwds, (demandCwds) => {
623 const next = new Map(demandCwds);
624 next.set(demandCwd, (next.get(demandCwd) ?? 0) + 1);
625 return next;
626 }).pipe(
627 Effect.map(() => {
628 const nextPollers = new Map(activePollers);
629 nextPollers.set(cwd, {
630 ...existing,
631 subscriberCount: existing.subscriberCount + 1,
632 });
633 return [undefined, nextPollers] as const;
634 }),
635 );
636 }
637
638 return Ref.make<ReadonlyMap<string, number>>(new Map([[demandCwd, 1]])).pipe(
639 Effect.flatMap((demandCwds) =>
640 makeRemoteRefreshLoop(
641 cwd,
642 demandCwds,
643 automaticRemoteRefreshInterval,
644 refreshImmediately,
645 ).pipe(
646 Effect.forkIn(broadcasterScope),
647 Effect.map((fiber) => {
648 const nextPollers = new Map(activePollers);
649 nextPollers.set(cwd, {
650 fiber,
651 subscriberCount: 1,
652 demandCwds,
653 });
654 return [undefined, nextPollers] as const;
655 }),
656 ),
657 ),
658 );
659 });
660 });
661
662 const releaseRemotePoller = Effect.fn("VcsStatusBroadcaster.releaseRemotePoller")(function* (
663 cwd: string,
664 demandCwd: string,
665 ) {
666 const pollerToInterrupt = yield* SynchronizedRef.modifyEffect(pollersRef, (activePollers) => {
667 const existing = activePollers.get(cwd);
668 if (!existing) {
669 return Effect.succeed([null, activePollers] as const);
670 }
671
672 if (existing.subscriberCount > 1) {
673 return Ref.update(existing.demandCwds, (demandCwds) => {
674 const nextDemandCwds = new Map(demandCwds);
675 const count = nextDemandCwds.get(demandCwd) ?? 0;
676 if (count <= 1) {
677 nextDemandCwds.delete(demandCwd);
678 } else {
679 nextDemandCwds.set(demandCwd, count - 1);
680 }
681 return nextDemandCwds;
682 }).pipe(
683 Effect.as([
684 null,
685 new Map(activePollers).set(cwd, {
686 ...existing,
687 subscriberCount: existing.subscriberCount - 1,
688 }),
689 ] as const),
690 );
691 }
692
693 return Effect.succeed([
694 existing.fiber,
695 new Map([...activePollers].filter(([activeCwd]) => activeCwd !== cwd)),
696 ] as const);
697 });
698
699 if (pollerToInterrupt) {
700 yield* Fiber.interrupt(pollerToInterrupt).pipe(Effect.ignore);
701 }
702 });
703
704 const streamStatus: VcsStatusBroadcaster["Service"]["streamStatus"] = (input, options) =>
705 Stream.unwrap(
706 Effect.gen(function* () {
707 const cwd = yield* withFileSystem(normalizeCwd(input.cwd));
708 const subscription = yield* PubSub.subscribe(changesPubSub);
709 const initialLocal = yield* getOrLoadLocalStatus(cwd);
710 const cachedStatus = yield* getCachedStatus(cwd);
711 const initialRemote = cachedStatus?.remote?.value ?? null;
712 if (input.includeRemote !== false) {
713 yield* retainRemotePoller(
714 cwd,
715 input.cwd,
716 options?.automaticRemoteRefreshInterval ??
717 Effect.succeed(DEFAULT_VCS_STATUS_REFRESH_INTERVAL),
718 cachedStatus?.remote === null || cachedStatus?.remote === undefined,
719 );
720 }
721 const release =
722 input.includeRemote === false
723 ? Effect.void
724 : releaseRemotePoller(cwd, input.cwd).pipe(Effect.ignore, Effect.asVoid);
725
726 return Stream.concat(
727 Stream.make({
728 _tag: "snapshot" as const,
729 local: initialLocal,
730 remote: initialRemote,
731 }),
732 Stream.fromSubscription(subscription).pipe(
733 Stream.filter((event) => event.cwd === cwd),
734 Stream.map((event) => event.event),
735 ),
736 ).pipe(Stream.ensuring(release));
737 }),
738 );
739
740 return VcsStatusBroadcaster.of({
741 getStatus,
742 refreshLocalStatus,
743 refreshStatus,
744 refreshPullRequestStatus,
745 streamStatus,
746 });
747});
748
749export const layer = Layer.effect(VcsStatusBroadcaster, make);