import { type EnvironmentId, type VcsListRefsInput, type VcsListRefsResult, type VcsStatusResult, WS_METHODS,} from "@t3tools/contracts";import { applyGitStatusStreamEvent } from "@t3tools/shared/git";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Result from "effect/Result";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { Atom } from "effect/unstable/reactivity"; import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts";import type { EnvironmentRegistry } from "../connection/registry.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import { safeErrorLogAttributes } from "../errors/safeLog.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";import { followStreamInEnvironment } from "./runtime.ts";import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts"; const OFFLINE_BRANCH_LIST_LIMIT = 100;const VCS_REFS_REVALIDATE_INTERVAL = "5 seconds"; function canUseVcsRefsCache(input: VcsListRefsInput): boolean { return ( input.query === undefined && input.cursor === undefined && input.includeMatchingRemoteRefs === undefined && input.refKind === undefined && input.limit === OFFLINE_BRANCH_LIST_LIMIT );} /** * Retains the last unfiltered branch-list response for the new-task picker. * Filtered or paginated lists intentionally stay live-only: treating a * partial result as a complete offline list would make branch selection * misleading. */export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChanges")(function* ( input: VcsListRefsInput,) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const environmentId = supervisor.target.environmentId; const useCache = canUseVcsRefsCache(input); const cached = useCache ? yield* cache.loadVcsRefs(environmentId, input.cwd).pipe( Effect.catch((error) => Effect.logWarning("Could not load cached Git refs.").pipe( Effect.annotateLogs({ environmentId, cwd: input.cwd, ...safeErrorLogAttributes(error), }), Effect.as(Option.none<VcsListRefsResult>()), ), ), ) : Option.none<VcsListRefsResult>(); const refresh = Effect.fn("CachedVcsRefsState.refresh")(function* () { const refs = yield* request(WS_METHODS.vcsListRefs, input).pipe( Effect.provideService(EnvironmentSupervisor, supervisor), ); if (useCache) { yield* cache.saveVcsRefs(environmentId, input.cwd, refs).pipe( Effect.catch((error) => Effect.logWarning("Could not persist cached Git refs.").pipe( Effect.annotateLogs({ environmentId, cwd: input.cwd, ...safeErrorLogAttributes(error), }), ), ), ); } return refs; }); const cachedRefs = Stream.fromEffect( SubscriptionRef.get(supervisor.state).pipe( Effect.flatMap((connection) => connection.phase === "connected" ? Effect.succeed(Option.none<VcsListRefsResult>()) : Effect.succeed(cached), ), ), ).pipe( Stream.filterMap((refs) => Option.match(refs, { onNone: () => Result.failVoid, onSome: Result.succeed, }), ), ); const refreshedRefs = Stream.concat( Stream.fromEffect(SubscriptionRef.get(supervisor.state)), SubscriptionRef.changes(supervisor.state), ).pipe( Stream.map((connection) => (connection.phase === "connected" ? connection.generation : null)), Stream.changes, Stream.switchMap((generation) => generation === null ? Stream.empty : Stream.tick(VCS_REFS_REVALIDATE_INTERVAL).pipe( Stream.mapEffect( () => refresh().pipe( Effect.map(Option.some), Effect.catch((error) => Effect.logWarning("Could not refresh Git refs.").pipe( Effect.annotateLogs({ environmentId, cwd: input.cwd, ...safeErrorLogAttributes(error), }), Effect.as(Option.none<VcsListRefsResult>()), ), ), ), { concurrency: 1 }, ), Stream.filterMap((refs) => Option.match(refs, { onNone: () => Result.failVoid, onSome: Result.succeed, }), ), ), ), ); return Stream.concat(cachedRefs, refreshedRefs);}); export function cachedVcsRefsChanges(environmentId: EnvironmentId, input: VcsListRefsInput) { return followStreamInEnvironment(environmentId, Stream.unwrap(makeCachedVcsRefsChanges(input)));} export function createVcsEnvironmentAtoms<R, E>( runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,) { const listRefsByEnvironment = Atom.family((environmentId: EnvironmentId) => Atom.family((inputKey: string) => { const input = JSON.parse(inputKey) as VcsListRefsInput; return runtime .atom(cachedVcsRefsChanges(environmentId, input)) .pipe( Atom.setIdleTTL(5 * 60_000), Atom.withLabel(`environment-data:vcs:list-refs:${environmentId}:${inputKey}`), ); }), ); const listRefs = (target: { readonly environmentId: EnvironmentId; readonly input: VcsListRefsInput; }) => listRefsByEnvironment(target.environmentId)(JSON.stringify(target.input)); return { listRefs, status: createEnvironmentSubscriptionAtomFamily(runtime, { label: "environment-data:vcs:status", subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>) => subscribe(WS_METHODS.subscribeVcsStatus, input).pipe( Stream.mapAccum( () => null as VcsStatusResult | null, (current, event) => { const next = applyGitStatusStreamEvent(current, event); return [next, [next]] as const; }, ), ), }), pull: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:pull", tag: WS_METHODS.vcsPull, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), refreshStatus: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:refresh-status", tag: WS_METHODS.vcsRefreshStatus, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), createWorktree: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:create-worktree", tag: WS_METHODS.vcsCreateWorktree, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), removeWorktree: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:remove-worktree", tag: WS_METHODS.vcsRemoveWorktree, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), createRef: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:create-ref", tag: WS_METHODS.vcsCreateRef, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), switchRef: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:switch-ref", tag: WS_METHODS.vcsSwitchRef, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), init: createEnvironmentRpcCommand(runtime, { label: "environment-data:vcs:init", tag: WS_METHODS.vcsInit, scheduler: vcsCommandScheduler, concurrency: vcsCommandConcurrency, }), };} export * from "./gitActions.ts";export * from "./vcsAction.ts";export * from "./vcsRef.ts";export * from "./vcsStatus.ts";
import {
type EnvironmentId,
type VcsListRefsInput,
type VcsListRefsResult,
type VcsStatusResult,
WS_METHODS,
} from "@t3tools/contracts";
import { applyGitStatusStreamEvent } from "@t3tools/shared/git";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";
import * as Result from "effect/Result";
import * as Stream from "effect/Stream";
import * as SubscriptionRef from "effect/SubscriptionRef";
import { Atom } from "effect/unstable/reactivity";
import { createEnvironmentRpcCommand, createEnvironmentSubscriptionAtomFamily } from "./runtime.ts";
import type { EnvironmentRegistry } from "../connection/registry.ts";
import { EnvironmentSupervisor } from "../connection/supervisor.ts";
import { safeErrorLogAttributes } from "../errors/safeLog.ts";
import { EnvironmentCacheStore } from "../platform/persistence.ts";
import { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
import { followStreamInEnvironment } from "./runtime.ts";
import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
const OFFLINE_BRANCH_LIST_LIMIT = 100;
const VCS_REFS_REVALIDATE_INTERVAL = "5 seconds";
function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
return (
input.query === undefined &&
input.cursor === undefined &&
input.includeMatchingRemoteRefs === undefined &&
input.refKind === undefined &&
input.limit === OFFLINE_BRANCH_LIST_LIMIT
);
}
/**
* Retains the last unfiltered branch-list response for the new-task picker.
* Filtered or paginated lists intentionally stay live-only: treating a
* partial result as a complete offline list would make branch selection
* misleading.
*/
export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChanges")(function* (
input: VcsListRefsInput,
) {
const supervisor = yield* EnvironmentSupervisor;
const cache = yield* EnvironmentCacheStore;
const environmentId = supervisor.target.environmentId;
const useCache = canUseVcsRefsCache(input);
const cached = useCache
? yield* cache.loadVcsRefs(environmentId, input.cwd).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not load cached Git refs.").pipe(
Effect.annotateLogs({
environmentId,
cwd: input.cwd,
...safeErrorLogAttributes(error),
}),
Effect.as(Option.none<VcsListRefsResult>()),
),
),
)
: Option.none<VcsListRefsResult>();
const refresh = Effect.fn("CachedVcsRefsState.refresh")(function* () {
const refs = yield* request(WS_METHODS.vcsListRefs, input).pipe(
Effect.provideService(EnvironmentSupervisor, supervisor),
);
if (useCache) {
yield* cache.saveVcsRefs(environmentId, input.cwd, refs).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not persist cached Git refs.").pipe(
Effect.annotateLogs({
environmentId,
cwd: input.cwd,
...safeErrorLogAttributes(error),
}),
),
),
);
}
return refs;
});
const cachedRefs = Stream.fromEffect(
SubscriptionRef.get(supervisor.state).pipe(
Effect.flatMap((connection) =>
connection.phase === "connected"
? Effect.succeed(Option.none<VcsListRefsResult>())
: Effect.succeed(cached),
),
),
).pipe(
Stream.filterMap((refs) =>
Option.match(refs, {
onNone: () => Result.failVoid,
onSome: Result.succeed,
}),
),
);
const refreshedRefs = Stream.concat(
Stream.fromEffect(SubscriptionRef.get(supervisor.state)),
SubscriptionRef.changes(supervisor.state),
).pipe(
Stream.map((connection) => (connection.phase === "connected" ? connection.generation : null)),
Stream.changes,
Stream.switchMap((generation) =>
generation === null
? Stream.empty
: Stream.tick(VCS_REFS_REVALIDATE_INTERVAL).pipe(
Stream.mapEffect(
() =>
refresh().pipe(
Effect.map(Option.some),
Effect.catch((error) =>
Effect.logWarning("Could not refresh Git refs.").pipe(
Effect.annotateLogs({
environmentId,
cwd: input.cwd,
...safeErrorLogAttributes(error),
}),
Effect.as(Option.none<VcsListRefsResult>()),
),
),
),
{ concurrency: 1 },
),
Stream.filterMap((refs) =>
Option.match(refs, {
onNone: () => Result.failVoid,
onSome: Result.succeed,
}),
),
),
),
);
return Stream.concat(cachedRefs, refreshedRefs);
});
export function cachedVcsRefsChanges(environmentId: EnvironmentId, input: VcsListRefsInput) {
return followStreamInEnvironment(environmentId, Stream.unwrap(makeCachedVcsRefsChanges(input)));
}
export function createVcsEnvironmentAtoms<R, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
) {
const listRefsByEnvironment = Atom.family((environmentId: EnvironmentId) =>
Atom.family((inputKey: string) => {
const input = JSON.parse(inputKey) as VcsListRefsInput;
return runtime
.atom(cachedVcsRefsChanges(environmentId, input))
.pipe(
Atom.setIdleTTL(5 * 60_000),
Atom.withLabel(`environment-data:vcs:list-refs:${environmentId}:${inputKey}`),
);
}),
);
const listRefs = (target: {
readonly environmentId: EnvironmentId;
readonly input: VcsListRefsInput;
}) => listRefsByEnvironment(target.environmentId)(JSON.stringify(target.input));
return {
listRefs,
status: createEnvironmentSubscriptionAtomFamily(runtime, {
label: "environment-data:vcs:status",
subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>) =>
subscribe(WS_METHODS.subscribeVcsStatus, input).pipe(
Stream.mapAccum(
() => null as VcsStatusResult | null,
(current, event) => {
const next = applyGitStatusStreamEvent(current, event);
return [next, [next]] as const;
},
),
),
}),
pull: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:pull",
tag: WS_METHODS.vcsPull,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
refreshStatus: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:refresh-status",
tag: WS_METHODS.vcsRefreshStatus,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
createWorktree: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:create-worktree",
tag: WS_METHODS.vcsCreateWorktree,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
removeWorktree: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:remove-worktree",
tag: WS_METHODS.vcsRemoveWorktree,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
createRef: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:create-ref",
tag: WS_METHODS.vcsCreateRef,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
switchRef: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:switch-ref",
tag: WS_METHODS.vcsSwitchRef,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
init: createEnvironmentRpcCommand(runtime, {
label: "environment-data:vcs:init",
tag: WS_METHODS.vcsInit,
scheduler: vcsCommandScheduler,
concurrency: vcsCommandConcurrency,
}),
};
}
export * from "./gitActions.ts";
export * from "./vcsAction.ts";
export * from "./vcsRef.ts";
export * from "./vcsStatus.ts";