import * as Cause from "effect/Cause";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 BackgroundPolicy from "../background/BackgroundPolicy.ts";import * as GitWorkflowService from "../git/GitWorkflowService.ts";import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.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);const MAX_FAILURE_DIAGNOSTIC_VALUES = 8;const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128; function boundedDiagnosticValue(value: string): string {  return value.slice(0, MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH);} function diagnosticValueTag(value: unknown): string {  try {    if (      typeof value === "object" &&      value !== null &&      "_tag" in value &&      typeof value._tag === "string"    ) {      return boundedDiagnosticValue(value._tag);    }    if (value instanceof Error) {      return boundedDiagnosticValue(value.name);    }    return typeof value;  } catch {    return "Uninspectable";  }} function diagnosticFailureOperation(value: unknown): string | undefined {  try {    if (      typeof value === "object" &&      value !== null &&      "operation" in value &&      typeof value.operation === "string"    ) {      return boundedDiagnosticValue(value.operation);    }  } catch {    return undefined;  }  return undefined;} function addUniqueDiagnosticValue(values: Array<string>, value: string | undefined): void {  if (    value !== undefined &&    values.length < MAX_FAILURE_DIAGNOSTIC_VALUES &&    !values.includes(value)  ) {    values.push(value);  }} export function remoteRefreshFailureDiagnostics(cause: Cause.Cause<unknown>) {  const failureTags: Array<string> = [];  const failureOperations: Array<string> = [];  const defectTags: Array<string> = [];  let failureCount = 0;  let defectCount = 0;  let interruptionCount = 0;   for (const reason of cause.reasons) {    if (Cause.isFailReason(reason)) {      failureCount += 1;      addUniqueDiagnosticValue(failureTags, diagnosticValueTag(reason.error));      addUniqueDiagnosticValue(failureOperations, diagnosticFailureOperation(reason.error));      continue;    }    if (Cause.isDieReason(reason)) {      defectCount += 1;      addUniqueDiagnosticValue(defectTags, diagnosticValueTag(reason.defect));      continue;    }    interruptionCount += 1;  }   return {    reasonCount: cause.reasons.length,    failureCount,    failureTags,    failureOperations,    defectCount,    defectTags,    interruptionCount,  };} 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;  readonly demandCwds: Ref.Ref<ReadonlyMap<string, number>>;} interface StreamStatusOptions {  readonly automaticRemoteRefreshInterval?: Effect.Effect<Duration.Duration, never>;} export class VcsAutoPullPolicy extends Context.Reference<{  readonly isEnabled: (cwd: string) => Effect.Effect<boolean, never>;}>("t3/vcs/VcsAutoPullPolicy", {  defaultValue: () => ({ isEnabled: () => Effect.succeed(false) }),}) {} export const autoPullPolicyLayer = Layer.effect(  VcsAutoPullPolicy,  Effect.gen(function* () {    const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;    return {      isEnabled: (cwd: string) =>        snapshots.getActiveProjectByWorkspaceRoot(cwd).pipe(          Effect.map((project) => project._tag === "Some" && project.value.autoPull === true),          Effect.orElseSucceed(() => false),        ),    };  }),); 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 autoPullPolicy = yield* VcsAutoPullPolicy;  const workflow = yield* GitWorkflowService.GitWorkflowService;  const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;  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 maybeAutoPull = Effect.fn("VcsStatusBroadcaster.maybeAutoPull")(function* (    cwd: string,    remote: VcsStatusRemoteResult | null,    policyCwds: ReadonlyArray<string>,  ) {    return yield* Effect.gen(function* () {      const autoPullEnabled = (yield* Effect.forEach(policyCwds, autoPullPolicy.isEnabled, {        concurrency: "unbounded",      })).some(Boolean);      if (        remote === null ||        !remote.hasUpstream ||        remote.aheadCount > 0 ||        remote.behindCount <= 0 ||        !autoPullEnabled      ) {        return null;      }       yield* workflow.invalidateLocalStatus(cwd);      const local = yield* workflow.localStatus({ cwd });      if (!local.isRepo || !local.isDefaultRef || local.hasWorkingTreeChanges) return null;       yield* workflow.pullCurrentBranch(cwd);      yield* workflow.invalidateStatus(cwd);      const [refreshedLocal, refreshedRemote] = yield* Effect.all(        [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd }, { refreshUpstream: false })],        { concurrency: "unbounded" },      );      yield* updateCachedStatus(cwd, refreshedLocal, refreshedRemote, { publish: true });      return { local: refreshedLocal, remote: refreshedRemote };    }).pipe(      Effect.catch(() =>        Effect.logWarning("Automatic project pull failed", { cwd }).pipe(Effect.as(null)),      ),    );  });   const refreshRemoteStatus = Effect.fn("VcsStatusBroadcaster.refreshRemoteStatus")(function* (    cwd: string,    options?: {      readonly refreshUpstream?: boolean;      readonly policyCwds?: ReadonlyArray<string>;    },  ) {    if (options?.refreshUpstream !== false) {      yield* workflow.invalidateRemoteStatus(cwd);    }    const remote = yield* workflow.remoteStatus({ cwd }, options);    const pulled = yield* maybeAutoPull(cwd, remote, options?.policyCwds ?? [cwd]);    if (pulled !== null) return pulled.remote;    return yield* updateCachedRemoteStatus(cwd, remote, { publish: true });  });   const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn(    "VcsStatusBroadcaster.refreshStatus",  )(function* (rawCwd) {    const cwd = yield* withFileSystem(normalizeCwd(rawCwd));    // invalidateStatus (not the two partial invalidations) so an explicit    // refresh also bypasses GitManager's slow PR-lookup cache.    yield* workflow.invalidateStatus(cwd);    const [local, remote] = yield* Effect.all(      [workflow.localStatus({ cwd }), workflow.remoteStatus({ cwd })],      { concurrency: "unbounded" },    );    const pulled = yield* maybeAutoPull(cwd, remote, [rawCwd]);    if (pulled !== null) return mergeGitStatusParts(pulled.local, pulled.remote);    return yield* updateCachedStatus(cwd, local, remote, { publish: true });  });   const makeRemoteRefreshLoop = (    cwd: string,    demandCwdsRef: Ref.Ref<ReadonlyMap<string, number>>,    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 demandCwds = yield* Ref.get(demandCwdsRef);        const shouldRun =          needsInitialRefresh ||          (yield* Effect.all(            [...demandCwds.keys()].map((demandCwd) =>              backgroundPolicy.shouldRunScopeWork({                type: "vcs-status",                cwd: demandCwd,              }),            ),            { concurrency: "unbounded" },          )).some(Boolean);        if (!shouldRun) {          return activeInterval;        }         const exit = yield* refreshRemoteStatus(cwd, {          refreshUpstream: !Duration.isZero(configuredInterval),          policyCwds: [...demandCwds.keys()],        }).pipe(Effect.exit);        if (Exit.isSuccess(exit)) {          yield* Ref.set(needsInitialRefreshRef, false);          yield* Ref.set(consecutiveFailuresRef, 0);          return activeInterval;        }         const interruptionReasons = exit.cause.reasons.filter(Cause.isInterruptReason);        if (interruptionReasons.length > 0) {          return yield* Effect.failCause(Cause.fromReasons<never>(interruptionReasons));        }         const consecutiveFailures = yield* Ref.updateAndGet(          consecutiveFailuresRef,

Lines 1–500 of 664. Load more