diff --git a/packages/opencode/src/effect/instances.ts b/packages/opencode/src/effect/instances.ts index 02d4bf4823..d60d793558 100644 --- a/packages/opencode/src/effect/instances.ts +++ b/packages/opencode/src/effect/instances.ts @@ -3,6 +3,7 @@ import { registerDisposer } from "./instance-registry" import { ProviderAuthService } from "@/provider/auth-service" import { QuestionService } from "@/question/service" import { PermissionService } from "@/permission/service" +import { FileWatcherService } from "@/file/watcher" import { Instance } from "@/project/instance" import type { Project } from "@/project/project" @@ -17,7 +18,7 @@ export class InstanceContext extends ServiceMap.Service { + try { + const binding = require( + `@parcel/watcher-${process.platform}-${process.arch}${process.platform === "linux" ? `-${OPENCODE_LIBC || "glibc"}` : ""}`, + ) + return createWrapper(binding) as typeof import("@parcel/watcher") + } catch (error) { + log.error("failed to load watcher binding", { error }) + return } +}) - const watcher = lazy((): typeof import("@parcel/watcher") | undefined => { - try { - const binding = require( - `@parcel/watcher-${process.platform}-${process.arch}${process.platform === "linux" ? `-${OPENCODE_LIBC || "glibc"}` : ""}`, - ) - return createWrapper(binding) as typeof import("@parcel/watcher") - } catch (error) { - log.error("failed to load watcher binding", { error }) - return - } - }) +function getBackend() { + if (process.platform === "win32") return "windows" + if (process.platform === "darwin") return "fs-events" + if (process.platform === "linux") return "inotify" +} - const state = Instance.state( - async () => { - log.info("init") - const cfg = await Config.get() - const backend = (() => { - if (process.platform === "win32") return "windows" - if (process.platform === "darwin") return "fs-events" - if (process.platform === "linux") return "inotify" - })() +export namespace FileWatcher { + export const Event = event +} + +const init = Effect.fn("FileWatcherService.init")(function* () {}) + +export namespace FileWatcherService { + export interface Service { + readonly init: () => Effect.Effect + } +} + +export class FileWatcherService extends ServiceMap.Service()( + "@opencode/FileWatcher", +) { + static readonly layer = Layer.effect( + FileWatcherService, + Effect.gen(function* () { + const instance = yield* InstanceContext + + if (Flag.OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER) return FileWatcherService.of({ init }) + + log.info("init", { directory: instance.directory }) + + const backend = getBackend() if (!backend) { - log.error("watcher backend not supported", { platform: process.platform }) - return {} + log.error("watcher backend not supported", { directory: instance.directory, platform: process.platform }) + return FileWatcherService.of({ init }) } - log.info("watcher backend", { platform: process.platform, backend }) const w = watcher() - if (!w) return {} + if (!w) return FileWatcherService.of({ init }) - const subscribe: ParcelWatcher.SubscribeCallback = (err, evts) => { + log.info("watcher backend", { directory: instance.directory, platform: process.platform, backend }) + + const subs: ParcelWatcher.AsyncSubscription[] = [] + yield* Effect.addFinalizer(() => Effect.promise(() => Promise.allSettled(subs.map((sub) => sub.unsubscribe())))) + + const cb: ParcelWatcher.SubscribeCallback = (err, evts) => { if (err) return for (const evt of evts) { - if (evt.type === "create") Bus.publish(Event.Updated, { file: evt.path, event: "add" }) - if (evt.type === "update") Bus.publish(Event.Updated, { file: evt.path, event: "change" }) - if (evt.type === "delete") Bus.publish(Event.Updated, { file: evt.path, event: "unlink" }) + if (evt.type === "create") Bus.publish(event.Updated, { file: evt.path, event: "add" }) + if (evt.type === "update") Bus.publish(event.Updated, { file: evt.path, event: "change" }) + if (evt.type === "delete") Bus.publish(event.Updated, { file: evt.path, event: "unlink" }) } } - const subs: ParcelWatcher.AsyncSubscription[] = [] + const subscribe = (dir: string, ignore: string[]) => + Effect.gen(function* () { + const sub = yield* Effect.promise(() => w.subscribe(dir, cb, { ignore, backend })) + subs.push(sub) + }).pipe( + Effect.timeout(SUBSCRIBE_TIMEOUT_MS), + Effect.catchCause((cause) => { + log.error("failed to subscribe", { dir, cause: Cause.pretty(cause) }) + return Effect.void + }), + ) + + const cfg = yield* Effect.promise(() => Config.get()) const cfgIgnores = cfg.watcher?.ignore ?? [] if (Flag.OPENCODE_EXPERIMENTAL_FILEWATCHER) { - const pending = w.subscribe(Instance.directory, subscribe, { - ignore: [...FileIgnore.PATTERNS, ...cfgIgnores, ...Protected.paths()], - backend, - }) - const sub = await withTimeout(pending, SUBSCRIBE_TIMEOUT_MS).catch((err) => { - log.error("failed to subscribe to Instance.directory", { error: err }) - pending.then((s) => s.unsubscribe()).catch(() => {}) - return undefined - }) - if (sub) subs.push(sub) + yield* subscribe(instance.directory, [...FileIgnore.PATTERNS, ...cfgIgnores, ...Protected.paths()]) } - if (Instance.project.vcs === "git") { - const result = await git(["rev-parse", "--git-dir"], { - cwd: Instance.worktree, - }) - const vcsDir = result.exitCode === 0 ? path.resolve(Instance.worktree, result.text().trim()) : undefined + if (instance.project.vcs === "git") { + const result = yield* Effect.promise(() => + git(["rev-parse", "--git-dir"], { + cwd: instance.project.worktree, + }), + ) + const vcsDir = result.exitCode === 0 ? path.resolve(instance.project.worktree, result.text().trim()) : undefined if (vcsDir && !cfgIgnores.includes(".git") && !cfgIgnores.includes(vcsDir)) { - const gitDirContents = await readdir(vcsDir).catch(() => []) - const ignoreList = gitDirContents.filter((entry) => entry !== "HEAD") - const pending = w.subscribe(vcsDir, subscribe, { - ignore: ignoreList, - backend, - }) - const sub = await withTimeout(pending, SUBSCRIBE_TIMEOUT_MS).catch((err) => { - log.error("failed to subscribe to vcsDir", { error: err }) - pending.then((s) => s.unsubscribe()).catch(() => {}) - return undefined - }) - if (sub) subs.push(sub) + const ignore = (yield* Effect.promise(() => readdir(vcsDir).catch(() => []))).filter( + (entry) => entry !== "HEAD", + ) + yield* subscribe(vcsDir, ignore) } } - return { subs } - }, - async (state) => { - if (!state.subs) return - await Promise.all(state.subs.map((sub) => sub?.unsubscribe())) - }, + return FileWatcherService.of({ init }) + }).pipe( + Effect.catchCause((cause) => { + log.error("failed to init watcher service", { cause: Cause.pretty(cause) }) + return Effect.succeed(FileWatcherService.of({ init: init })) + }), + ), ) - - export function init() { - if (Flag.OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER) { - return - } - state() - } } diff --git a/packages/opencode/src/project/bootstrap.ts b/packages/opencode/src/project/bootstrap.ts index a2be3733f8..bd819dc280 100644 --- a/packages/opencode/src/project/bootstrap.ts +++ b/packages/opencode/src/project/bootstrap.ts @@ -1,7 +1,7 @@ import { Plugin } from "../plugin" import { Format } from "../format" import { LSP } from "../lsp" -import { FileWatcher } from "../file/watcher" +import { FileWatcherService } from "../file/watcher" import { File } from "../file" import { Project } from "./project" import { Bus } from "../bus" @@ -12,6 +12,7 @@ import { Log } from "@/util/log" import { ShareNext } from "@/share/share-next" import { Snapshot } from "../snapshot" import { Truncate } from "../tool/truncation" +import { runPromiseInstance } from "@/effect/runtime" export async function InstanceBootstrap() { Log.Default.info("bootstrapping", { directory: Instance.directory }) @@ -19,7 +20,7 @@ export async function InstanceBootstrap() { ShareNext.init() Format.init() await LSP.init() - FileWatcher.init() + await runPromiseInstance(FileWatcherService.use((service) => service.init())) File.init() Vcs.init() Snapshot.init() diff --git a/packages/opencode/test/file/watcher.test.ts b/packages/opencode/test/file/watcher.test.ts new file mode 100644 index 0000000000..2ceba74fdd --- /dev/null +++ b/packages/opencode/test/file/watcher.test.ts @@ -0,0 +1,205 @@ +import { $ } from "bun" +import { afterEach, expect, mock, test } from "bun:test" +import fs from "fs/promises" +import path from "path" +import { tmpdir } from "../fixture/fixture" + +const flags = await import("../../src/flag/flag") + +mock.module("@/flag/flag", () => ({ + Flag: { + ...flags.Flag, + OPENCODE_EXPERIMENTAL_FILEWATCHER: true, + OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: false, + }, +})) + +async function load() { + const { runPromiseInstance } = await import("../../src/effect/runtime") + const watcher = await import("../../src/file/watcher") + const { GlobalBus } = await import("../../src/bus/global") + const { Instance } = await import("../../src/project/instance") + + return { + GlobalBus, + FileWatcher: watcher.FileWatcher, + FileWatcherService: watcher.FileWatcherService, + Instance, + runPromiseInstance, + } +} + +async function start(directory: string) { + const { FileWatcherService, Instance, runPromiseInstance } = await load() + await Instance.provide({ + directory, + fn: () => runPromiseInstance(FileWatcherService.use((service) => service.init())), + }) + await Bun.sleep(100) +} + +async function stop(directory: string) { + const { Instance } = await load() + await Instance.provide({ + directory, + fn: () => Instance.dispose(), + }) + await Bun.sleep(100) +} + +async function nextUpdate( + directory: string, + check: (evt: { file: string; event: "add" | "change" | "unlink" }) => boolean, + run: () => Promise, +) { + const { FileWatcher, GlobalBus } = await load() + + return await new Promise<{ file: string; event: "add" | "change" | "unlink" }>((resolve, reject) => { + const on = (evt: { + directory?: string + payload: { + type: string + properties: { + file: string + event: "add" | "change" | "unlink" + } + } + }) => { + if (evt.directory !== directory) return + if (evt.payload.type !== FileWatcher.Event.Updated.type) return + if (!check(evt.payload.properties)) return + clearTimeout(timeout) + GlobalBus.off("event", on) + resolve(evt.payload.properties) + } + + const timeout = setTimeout(() => { + GlobalBus.off("event", on) + reject(new Error("timed out waiting for file watcher event")) + }, 5000) + + GlobalBus.on("event", on) + + run().catch((err) => { + clearTimeout(timeout) + GlobalBus.off("event", on) + reject(err) + }) + }) +} + +afterEach(async () => { + const { Instance } = await load() + await Instance.disposeAll() +}) + +test("FileWatcherService publishes root create, update, and delete events", async () => { + await using tmp = await tmpdir({ git: true }) + const file = path.join(tmp.path, "watch.txt") + + await start(tmp.path) + + await expect( + nextUpdate( + tmp.path, + (evt) => evt.file === file && evt.event === "add", + () => fs.writeFile(file, "a"), + ), + ).resolves.toEqual({ + file, + event: "add", + }) + + await expect( + nextUpdate( + tmp.path, + (evt) => evt.file === file && evt.event === "change", + () => fs.writeFile(file, "b"), + ), + ).resolves.toEqual({ + file, + event: "change", + }) + + await expect( + nextUpdate( + tmp.path, + (evt) => evt.file === file && evt.event === "unlink", + () => fs.unlink(file), + ), + ).resolves.toEqual({ + file, + event: "unlink", + }) +}) + +test("FileWatcherService watches non-git roots", async () => { + await using tmp = await tmpdir() + const file = path.join(tmp.path, "plain.txt") + + await start(tmp.path) + + await expect( + nextUpdate( + tmp.path, + (evt) => evt.file === file && evt.event === "add", + () => fs.writeFile(file, "plain"), + ), + ).resolves.toEqual({ + file, + event: "add", + }) +}) + +test("FileWatcherService cleanup stops publishing events", async () => { + await using tmp = await tmpdir({ git: true }) + const file = path.join(tmp.path, "after-dispose.txt") + const { FileWatcher, GlobalBus } = await load() + let seen = false + + await start(tmp.path) + await stop(tmp.path) + + const on = (evt: { directory?: string; payload: { type: string; properties: { file: string } } }) => { + if (evt.directory !== tmp.path) return + if (evt.payload.type !== FileWatcher.Event.Updated.type) return + if (evt.payload.properties.file === file) seen = true + } + + GlobalBus.on("event", on) + + try { + await fs.writeFile(file, "gone") + await Bun.sleep(500) + expect(seen).toBe(false) + } finally { + GlobalBus.off("event", on) + } +}) + +test("FileWatcherService ignores non-HEAD git metadata changes", async () => { + await using tmp = await tmpdir({ git: true }) + const file = path.join(tmp.path, ".git", "index") + const edit = path.join(tmp.path, "tracked.txt") + const { FileWatcher, GlobalBus } = await load() + let seen = false + + await start(tmp.path) + + const on = (evt: { directory?: string; payload: { type: string; properties: { file: string } } }) => { + if (evt.directory !== tmp.path) return + if (evt.payload.type !== FileWatcher.Event.Updated.type) return + if (evt.payload.properties.file === file) seen = true + } + + GlobalBus.on("event", on) + + try { + await fs.writeFile(edit, "a") + await $`git add .`.cwd(tmp.path).quiet().nothrow() + await Bun.sleep(500) + expect(seen).toBe(false) + } finally { + GlobalBus.off("event", on) + } +})