diff --git a/apps/mobile/src/features/files/preload-workspace-file.ts b/apps/mobile/src/features/files/preload-workspace-file.ts index b9e21cfd98f..fbc3fc41426 100644 --- a/apps/mobile/src/features/files/preload-workspace-file.ts +++ b/apps/mobile/src/features/files/preload-workspace-file.ts @@ -1,5 +1,6 @@ import { executeAtomQuery } from "@t3tools/client-runtime/state/runtime"; import type { EnvironmentId } from "@t3tools/contracts"; +import { Atom } from "effect/unstable/reactivity"; import { appAtomRegistry } from "../../state/atom-registry"; import { projectEnvironment } from "../../state/projects"; @@ -10,6 +11,7 @@ import type { ReviewDiffTheme } from "../review/shikiReviewHighlighter"; const inFlightPreloads = new Map>(); const MAX_HIGHLIGHT_PRELOAD_CHARACTERS = 256 * 1024; +const WORKSPACE_FILE_PRELOAD_RETAIN_MS = 1_000; function preloadKey(input: { readonly cwd: string; @@ -36,10 +38,12 @@ export function preloadWorkspaceFileContents(input: { const preload = executeAtomQuery( appAtomRegistry, - projectEnvironment.readFile({ - environmentId: input.environmentId, - input: { cwd: input.cwd, relativePath: input.relativePath }, - }), + projectEnvironment + .readFile({ + environmentId: input.environmentId, + input: { cwd: input.cwd, relativePath: input.relativePath }, + }) + .pipe(Atom.setIdleTTL(WORKSPACE_FILE_PRELOAD_RETAIN_MS)), { label: "workspace file preload", reportDefect: false, diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index fb753b9aa4b..cdf12e5c627 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -56,6 +56,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.projectsListEntries]: AuthOrchestrationReadScope, [WS_METHODS.projectsReadFile]: AuthOrchestrationReadScope, [WS_METHODS.projectsSearchContents]: AuthOrchestrationReadScope, + [WS_METHODS.projectsWatchFile]: AuthOrchestrationReadScope, [WS_METHODS.projectsSearchEntries]: AuthOrchestrationReadScope, [WS_METHODS.projectsWriteFile]: AuthOrchestrationOperateScope, [WS_METHODS.shellOpenInEditor]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/workspace/WorkspaceFileSystem.test.ts b/apps/server/src/workspace/WorkspaceFileSystem.test.ts index cecffbc1993..f9b51f12a77 100644 --- a/apps/server/src/workspace/WorkspaceFileSystem.test.ts +++ b/apps/server/src/workspace/WorkspaceFileSystem.test.ts @@ -2,8 +2,11 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { it, describe, expect } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Stream from "effect/Stream"; import * as ServerConfig from "../config.ts"; import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; @@ -265,4 +268,42 @@ it.layer(TestLayer, { excludeTestServices: true })("WorkspaceFileSystemLive", (i }), ); }); + + describe("watchFile", () => { + it.effect("emits when a missing workspace file is created", () => + Effect.gen(function* () { + const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; + const cwd = yield* makeTempDir; + const eventFiber = yield* workspaceFileSystem + .watchFile({ cwd, relativePath: "t3.json" }) + .pipe(Stream.runHead, Effect.timeout("5 seconds"), Effect.forkChild); + + yield* Effect.sleep("100 millis"); + yield* writeTextFile(cwd, "t3.json", '{"scripts":[]}'); + + const event = yield* Fiber.join(eventFiber); + expect(Option.getOrUndefined(event)).toEqual({ relativePath: "t3.json" }); + }), + ); + + it.effect("emits when an in-workspace symlink target changes", () => + Effect.gen(function* () { + const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* makeTempDir; + yield* writeTextFile(cwd, "src/target.txt", "initial"); + yield* fileSystem.symlink(path.join(cwd, "src/target.txt"), path.join(cwd, "linked.txt")); + const eventFiber = yield* workspaceFileSystem + .watchFile({ cwd, relativePath: "linked.txt" }) + .pipe(Stream.runHead, Effect.timeout("5 seconds"), Effect.forkChild); + + yield* Effect.sleep("100 millis"); + yield* writeTextFile(cwd, "src/target.txt", "updated"); + + const event = yield* Fiber.join(eventFiber); + expect(Option.getOrUndefined(event)).toEqual({ relativePath: "linked.txt" }); + }), + ); + }); }); diff --git a/apps/server/src/workspace/WorkspaceFileSystem.ts b/apps/server/src/workspace/WorkspaceFileSystem.ts index e2dc9cbbb39..220e96bd200 100644 --- a/apps/server/src/workspace/WorkspaceFileSystem.ts +++ b/apps/server/src/workspace/WorkspaceFileSystem.ts @@ -10,17 +10,20 @@ import * as NodeFSP from "node:fs/promises"; import type { + ProjectFileChangeEvent, ProjectReadFileInput, ProjectReadFileResult, ProjectWriteFileInput, ProjectWriteFileResult, } from "@t3tools/contracts"; import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as WorkspaceEntries from "./WorkspaceEntries.ts"; import * as WorkspacePaths from "./WorkspacePaths.ts"; @@ -37,10 +40,12 @@ export class WorkspaceFileSystemOperationError extends Schema.TaggedErrorClass; + /** Watch a workspace-relative file and emit after its directory entry changes. */ + readonly watchFile: ( + input: ProjectReadFileInput, + ) => Stream.Stream< + ProjectFileChangeEvent, + WorkspaceFileSystemError | WorkspacePaths.WorkspacePathOutsideRootError + >; /** * Write a file relative to the workspace root. * @@ -259,6 +271,126 @@ export const make = Effect.gen(function* () { ); }); + const watchFile: WorkspaceFileSystem["Service"]["watchFile"] = (input) => + Stream.unwrap( + Effect.gen(function* () { + const target = yield* workspacePaths.resolveRelativePathWithinRoot({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + }); + const watchDirectory = path.dirname(target.absolutePath); + const realWorkspaceRoot = yield* Effect.tryPromise({ + try: () => NodeFSP.realpath(input.cwd), + catch: (cause) => + new WorkspaceFileSystemOperationError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedPath: target.absolutePath, + operationPath: input.cwd, + operation: "realpath-workspace-root", + cause, + }), + }); + const realWatchDirectory = yield* Effect.tryPromise({ + try: () => NodeFSP.realpath(watchDirectory), + catch: (cause) => + new WorkspaceFileSystemOperationError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedPath: target.absolutePath, + operationPath: watchDirectory, + operation: "realpath-watch-directory", + cause, + }), + }); + const relativeRealDirectory = path.relative(realWorkspaceRoot, realWatchDirectory); + if ( + relativeRealDirectory.startsWith(`..${path.sep}`) || + relativeRealDirectory === ".." || + path.isAbsolute(relativeRealDirectory) + ) { + return yield* new WorkspaceFilePathEscapeError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedWorkspaceRoot: realWorkspaceRoot, + resolvedPath: realWatchDirectory, + }); + } + + const realTargetPath = yield* Effect.tryPromise({ + try: async () => { + try { + return await NodeFSP.realpath(target.absolutePath); + } catch (cause) { + if ((cause as NodeJS.ErrnoException).code === "ENOENT") { + return null; + } + throw cause; + } + }, + catch: (cause) => + new WorkspaceFileSystemOperationError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedPath: target.absolutePath, + operationPath: target.absolutePath, + operation: "realpath-target", + cause, + }), + }); + if (realTargetPath !== null) { + const relativeRealTarget = path.relative(realWorkspaceRoot, realTargetPath); + if ( + relativeRealTarget.startsWith(`..${path.sep}`) || + relativeRealTarget === ".." || + path.isAbsolute(relativeRealTarget) + ) { + return yield* new WorkspaceFilePathEscapeError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedWorkspaceRoot: realWorkspaceRoot, + resolvedPath: realTargetPath, + }); + } + } + + const watchEntry = (watchPath: string, fileName: string) => { + const watchedAbsolutePath = path.join(watchPath, fileName); + return fileSystem.watch(watchPath).pipe( + Stream.filter((event) => { + return ( + event.path === fileName || + event.path === watchedAbsolutePath || + path.resolve(watchPath, event.path) === watchedAbsolutePath + ); + }), + Stream.mapError( + (cause) => + new WorkspaceFileSystemOperationError({ + workspaceRoot: input.cwd, + relativePath: input.relativePath, + resolvedPath: target.absolutePath, + operationPath: watchPath, + operation: "watch", + cause, + }), + ), + ); + }; + + const lexicalTargetPath = path.join(realWatchDirectory, path.basename(target.absolutePath)); + const lexicalEvents = watchEntry(realWatchDirectory, path.basename(target.absolutePath)); + const resolvedTargetEvents = + realTargetPath !== null && realTargetPath !== lexicalTargetPath + ? watchEntry(path.dirname(realTargetPath), path.basename(realTargetPath)) + : Stream.empty; + return Stream.merge(lexicalEvents, resolvedTargetEvents).pipe( + Stream.debounce(Duration.millis(100)), + Stream.map(() => ({ relativePath: target.relativePath })), + ); + }), + ); + const writeFile: WorkspaceFileSystem["Service"]["writeFile"] = Effect.fn( "WorkspaceFileSystem.writeFile", )(function* (input) { @@ -297,7 +429,7 @@ export const make = Effect.gen(function* () { return { relativePath: target.relativePath }; }); - return WorkspaceFileSystem.of({ readFile, writeFile }); + return WorkspaceFileSystem.of({ readFile, watchFile, writeFile }); }); export const layer = Layer.effect(WorkspaceFileSystem, make); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 909a51a4cf5..f8f99b5640a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1664,6 +1664,21 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "workspace" }, ), + [WS_METHODS.projectsWatchFile]: (input) => + observeRpcStream( + WS_METHODS.projectsWatchFile, + workspaceFileSystem.watchFile(input).pipe( + Stream.mapError( + (cause) => + new ProjectReadFileError({ + ...input, + ...projectFileFailureContext(cause), + cause, + }), + ), + ), + { "rpc.aggregate": "workspace" }, + ), [WS_METHODS.projectsWriteFile]: (input) => observeRpcEffect( WS_METHODS.projectsWriteFile, diff --git a/apps/web/src/components/files/projectFilesQueryState.ts b/apps/web/src/components/files/projectFilesQueryState.ts index d165c1d1a7a..febffe3d5ee 100644 --- a/apps/web/src/components/files/projectFilesQueryState.ts +++ b/apps/web/src/components/files/projectFilesQueryState.ts @@ -169,9 +169,10 @@ export function useProjectFileQuery( relativePath: string | null, enabled = true, ): ProjectQueryState { - const atom = enabled - ? getProjectFileQueryAtom(environmentId, cwd, relativePath) - : EMPTY_PROJECT_FILE_QUERY_ATOM; + const atom = + enabled && relativePath !== null + ? getProjectFileQueryAtom(environmentId, cwd, relativePath) + : EMPTY_PROJECT_FILE_QUERY_ATOM; const result = useAtomValue(atom); const refreshAtom = useAtomRefresh(atom); const refresh = useCallback(() => refreshAtom(), [refreshAtom]); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index bfe57a6c0dd..cb846e6069c 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -50,6 +50,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribePreviewEvents | typeof WS_METHODS.subscribeDiscoveredLocalServers | typeof WS_METHODS.subscribeResourceTelemetry + | typeof WS_METHODS.projectsWatchFile | typeof WS_METHODS.previewAutomationConnect | typeof WS_METHODS.subscribeVcsStatus | typeof WS_METHODS.terminalAttach; diff --git a/packages/client-runtime/src/state/projectCommands.ts b/packages/client-runtime/src/state/projectCommands.ts index 3defcc32154..75fb65f6d5b 100644 --- a/packages/client-runtime/src/state/projectCommands.ts +++ b/packages/client-runtime/src/state/projectCommands.ts @@ -7,6 +7,8 @@ import { createEnvironmentCommand, createEnvironmentRpcCommand, createEnvironmentRpcQueryAtomFamily, + createEnvironmentRpcSubscriptionAtomFamily, + refreshQueryOnSuccess, } from "./runtime.ts"; import { type CreateProjectInput, @@ -54,6 +56,32 @@ export function createProjectEnvironmentAtoms( key: ({ environmentId, input }: { environmentId: string; input: { projectId: string } }) => JSON.stringify([environmentId, input.projectId]), }; + const readFileQuery = createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:projects:read-file-query", + tag: WS_METHODS.projectsReadFile, + // Workspace files can change outside T3 Code, so always revalidate cached reads on mount. + staleTimeMs: 0, + idleTtlMs: 0, + }); + const fileChanges = createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:projects:file-changes", + tag: WS_METHODS.projectsWatchFile, + idleTtlMs: 0, + }); + type ReadFileAtom = ReturnType; + const liveReadFileAtoms = new WeakMap(); + const readFile = (target: Parameters[0]): ReadFileAtom => { + const queryAtom = readFileQuery(target); + const cached = liveReadFileAtoms.get(queryAtom); + if (cached) return cached; + + const changesAtom = fileChanges(target); + const liveAtom = refreshQueryOnSuccess(queryAtom, changesAtom).pipe( + Atom.withLabel(`environment-data:projects:read-file:${target.input.relativePath}`), + ); + liveReadFileAtoms.set(queryAtom, liveAtom); + return liveAtom; + }; return { searchEntries: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:projects:search-entries", @@ -66,12 +94,7 @@ export function createProjectEnvironmentAtoms( staleTimeMs: 30_000, idleTtlMs: 5 * 60_000, }), - readFile: createEnvironmentRpcQueryAtomFamily(runtime, { - label: "environment-data:projects:read-file", - tag: WS_METHODS.projectsReadFile, - staleTimeMs: 30_000, - idleTtlMs: 5 * 60_000, - }), + readFile, optimisticFile: (target: OptimisticProjectFileTarget) => optimisticFileFamily(optimisticProjectFileKey(target)), create: createEnvironmentCommand(runtime, { diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index f36087ebf66..6bb912339c7 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -18,6 +18,7 @@ import { executeAtomQuery, isAtomCommandInterrupted, mapAtomCommandResult, + refreshQueryOnSuccess, runAtomCommand, settleAsyncResult, settlePromise, @@ -62,6 +63,135 @@ describe("settleAsyncResult", () => { }); }); +describe("refreshQueryOnSuccess", () => { + it.effect("refreshes the query for new success signals without replaying a cached signal", () => + Effect.gen(function* () { + let reads = 0; + const query = Atom.make( + Effect.sync(() => { + reads += 1; + return reads; + }), + ); + const signal = Atom.make>( + AsyncResult.initial(false), + ); + const liveQuery = refreshQueryOnSuccess(query, signal); + const registry = AtomRegistry.make(); + const unmount = registry.mount(liveQuery); + + expect( + yield* AtomRegistry.getResult(registry, liveQuery, { + suspendOnWaiting: true, + }), + ).toBe(1); + expect(reads).toBe(1); + + registry.set(signal, AsyncResult.success("changed")); + expect( + yield* AtomRegistry.getResult(registry, liveQuery, { + suspendOnWaiting: true, + }), + ).toBe(2); + + unmount(); + registry.dispose(); + }), + ); + + it.effect("keeps a preloaded zero-TTL query warm through the consumer handoff", () => + Effect.gen(function* () { + let reads = 0; + const query = Atom.make( + Effect.sync(() => { + reads += 1; + return reads; + }), + ).pipe(Atom.setIdleTTL(0)); + const signal = Atom.make>( + AsyncResult.initial(false), + ).pipe(Atom.setIdleTTL(0)); + const liveQuery = refreshQueryOnSuccess(query, signal); + const preloadedQuery = liveQuery.pipe(Atom.setIdleTTL(1_000)); + const registry = AtomRegistry.make(); + + const preload = yield* Effect.promise(() => + executeAtomQuery(registry, preloadedQuery, { + label: "preload", + }), + ); + expect(AsyncResult.isSuccess(preload)).toBe(true); + expect(reads).toBe(1); + + const unmountConsumer = registry.mount(liveQuery); + expect( + yield* AtomRegistry.getResult(registry, liveQuery, { + suspendOnWaiting: true, + }), + ).toBe(1); + expect(reads).toBe(1); + + unmountConsumer(); + registry.dispose(); + }), + ); + + it.effect("revalidates a stale cached query after the live query remounts", () => + Effect.gen(function* () { + let contents = "initial"; + let reads = 0; + const query = Atom.make( + Effect.sync(() => { + reads += 1; + return contents; + }), + ).pipe(Atom.swr({ staleTime: 0, revalidateOnMount: true }), Atom.setIdleTTL(0)); + const signal = Atom.make>( + AsyncResult.initial(false), + ); + const liveQuery = refreshQueryOnSuccess(query, signal); + const scheduledTasks: Array<() => void> = []; + const registry = AtomRegistry.make({ + defaultIdleTTL: 5 * 60_000, + scheduleTask: (task) => { + let active = true; + scheduledTasks.push(() => { + if (active) task(); + }); + return () => { + active = false; + }; + }, + }); + const unmountInitial = registry.mount(liveQuery); + + expect( + yield* AtomRegistry.getResult(registry, liveQuery, { + suspendOnWaiting: true, + }), + ).toBe("initial"); + const readsAfterInitialMount = reads; + + unmountInitial(); + contents = "updated"; + while (scheduledTasks.length > 0) { + scheduledTasks.shift()?.(); + } + + const unmountUpdated = registry.mount(liveQuery); + expect( + yield* AtomRegistry.getResult(registry, liveQuery, { + suspendOnWaiting: true, + }), + ).toBe("updated"); + expect(reads).toBe(readsAfterInitialMount + 1); + + unmountUpdated(); + registry.dispose(); + }), + ); +}); + describe("atom command result helpers", () => { it("maps successful command values", () => { const result = mapAtomCommandResult(AsyncResult.success(2), (value) => value * 3); diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index 0a00e919c4b..168f7f30d0c 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -79,6 +79,29 @@ export interface AtomCommandReporter { readonly error: (message: string, cause: Cause.Cause) => void; } +export function refreshQueryOnSuccess( + queryAtom: Atom.Atom>, + signalAtom: Atom.Atom>, +): Atom.Atom> { + return Atom.transform( + queryAtom, + (get) => { + const current = get.once(queryAtom); + get.once(signalAtom); + get.subscribe(queryAtom, (result) => { + get.setSelf(result); + }); + get.subscribe(signalAtom, (result) => { + if (AsyncResult.isSuccess(result)) { + get.refresh(queryAtom); + } + }); + return current; + }, + { initialValueTarget: queryAtom }, + ).pipe(Atom.setIdleTTL(0)); +} + export interface AtomCommand { readonly label: string; readonly run: (registry: AtomRegistry.AtomRegistry, input: W) => Promise>; diff --git a/packages/contracts/src/project.ts b/packages/contracts/src/project.ts index a1b11df73b2..252cb34b1ed 100644 --- a/packages/contracts/src/project.ts +++ b/packages/contracts/src/project.ts @@ -204,6 +204,11 @@ export const ProjectReadFileResult = Schema.Struct({ }); export type ProjectReadFileResult = typeof ProjectReadFileResult.Type; +export const ProjectFileChangeEvent = Schema.Struct({ + relativePath: TrimmedNonEmptyString, +}); +export type ProjectFileChangeEvent = typeof ProjectFileChangeEvent.Type; + export const ProjectFileFailure = Schema.Literals([ "workspace_path_outside_root", "resolved_path_outside_root", @@ -216,10 +221,12 @@ export type ProjectFileFailure = typeof ProjectFileFailure.Type; export const ProjectFileOperation = Schema.Literals([ "realpath-workspace-root", "realpath-target", + "realpath-watch-directory", "open", "stat", "read", "close", + "watch", "make-directory", "write-file", ]); diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 400011f8843..7c9f4a290bb 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -73,6 +73,7 @@ import { ProjectListEntriesError, ProjectListEntriesInput, ProjectListEntriesResult, + ProjectFileChangeEvent, ProjectReadFileError, ProjectReadFileInput, ProjectReadFileResult, @@ -170,6 +171,7 @@ export const WS_METHODS = { projectsListEntries: "projects.listEntries", projectsReadFile: "projects.readFile", projectsSearchContents: "projects.searchContents", + projectsWatchFile: "projects.watchFile", projectsSearchEntries: "projects.searchEntries", projectsWriteFile: "projects.writeFile", @@ -460,6 +462,13 @@ export const WsProjectsReadFileRpc = Rpc.make(WS_METHODS.projectsReadFile, { error: Schema.Union([ProjectReadFileError, EnvironmentAuthorizationError]), }); +export const WsProjectsWatchFileRpc = Rpc.make(WS_METHODS.projectsWatchFile, { + payload: ProjectReadFileInput, + success: ProjectFileChangeEvent, + error: Schema.Union([ProjectReadFileError, EnvironmentAuthorizationError]), + stream: true, +}); + export const WsProjectsWriteFileRpc = Rpc.make(WS_METHODS.projectsWriteFile, { payload: ProjectWriteFileInput, success: ProjectWriteFileResult, @@ -812,6 +821,7 @@ export const WsRpcGroup = RpcGroup.make( WsProjectsListEntriesRpc, WsProjectsReadFileRpc, WsProjectsSearchContentsRpc, + WsProjectsWatchFileRpc, WsProjectsSearchEntriesRpc, WsProjectsWriteFileRpc, WsShellOpenInEditorRpc,