export * as Watcher from "./watcher" // @ts-ignore import { createWrapper } from "@parcel/watcher/wrapper" import type ParcelWatcher from "@parcel/watcher" import { FileSystem } from "@opencode-ai/schema/filesystem" import { makeGlobalNode } from "../effect/app-node" import { Cause, Context, Effect, Layer, PubSub, Scope, Stream } from "effect" import { KeyedMutex } from "../effect/keyed-mutex" import { Flag } from "../flag/flag" import { lazy } from "../util/lazy" import { watch as watchFileSystem } from "node:fs" import path from "path" declare const OPENCODE_LIBC: string | undefined const SUBSCRIBE_TIMEOUT_MS = 10_000 export const Event = { Updated: FileSystem.Event.Changed } const watcher = lazy((): typeof import("@parcel/watcher") | undefined => { try { const libc = typeof OPENCODE_LIBC === "undefined" ? undefined : OPENCODE_LIBC const binding = require( `@parcel/watcher-${process.platform}-${process.arch}${process.platform === "linux" ? `-${libc || "glibc"}` : ""}`, ) return createWrapper(binding) as typeof import("@parcel/watcher") } catch { return } }) function getBackend() { if (process.platform === "win32") return "windows" if (process.platform === "darwin") return "fs-events" if (process.platform === "linux") return "inotify" } export const hasNativeBinding = () => !!watcher() export type Update = ParcelWatcher.Event export type WatchInput = | { readonly path: string; readonly type: "file" } | { readonly path: string; readonly type: "directory"; readonly ignore?: readonly string[] } export interface Interface { readonly subscribe: (input: WatchInput) => Stream.Stream } export class Service extends Context.Service()("@opencode/Watcher") {} const layer = Layer.effect( Service, Effect.gen(function* () { const backend = getBackend() const native = watcher() if (Flag.OPENCODE_DISABLE_FILEWATCHER) { return Service.of({ subscribe: () => Stream.empty }) } type Entry = { readonly pubsub: PubSub.PubSub readonly subscription: { readonly unsubscribe: () => Promise } refs: number } const entries = new Map() const locks = KeyedMutex.makeUnsafe() const acquire = Effect.fn("Watcher.acquire")(function* (input: WatchInput) { const scope = yield* Scope.Scope const target = path.resolve(input.path) const directory = input.type === "file" ? path.dirname(target) : target const ignore = [...new Set(input.type === "directory" ? (input.ignore ?? []) : [])].toSorted() const id = JSON.stringify([input.type, target, ignore]) const pubsub = yield* locks.withLock(id)( Effect.gen(function* () { const existing = entries.get(id) if (existing) { existing.refs++ return existing.pubsub } const pubsub = yield* PubSub.unbounded() const subscription = yield* input.type === "file" ? Effect.sync(() => { const subscription = watchFileSystem(directory, { recursive: false }, (_event, file) => { if (file && path.resolve(directory, file.toString()) !== target) return PubSub.publishUnsafe(pubsub, { path: target, type: "update", } satisfies Update) }) if ("on" in subscription && typeof subscription.on === "function") { subscription.on("error", (error: unknown) => Effect.runFork(Effect.logError("watcher callback failed", { path: target, error })), ) } return { unsubscribe: () => Promise.resolve(subscription.close()) } }) : subscribeDirectory(native, backend, directory, ignore, pubsub) if (subscription) { entries.set(id, { pubsub, subscription, refs: 1 }) yield* Effect.logInfo("watcher started", { path: target, type: input.type, backend: input.type === "file" ? "node" : backend, ignores: ignore.length, }) return pubsub } yield* PubSub.shutdown(pubsub) return pubsub }), ) yield* Scope.addFinalizer( scope, locks.withLock(id)( Effect.gen(function* () { const entry = entries.get(id) if (!entry) return entry.refs-- if (entry.refs > 0) return entries.delete(id) yield* Effect.promise(() => entry.subscription.unsubscribe()).pipe(Effect.ignore) yield* PubSub.shutdown(entry.pubsub) yield* Effect.logInfo("watcher stopped", { path: target, type: input.type }) }), ), ) return pubsub }) const subscribe = (input: WatchInput) => Stream.unwrap(acquire(input).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))) return Service.of({ subscribe }) }), ) export const node = makeGlobalNode({ service: Service, layer, deps: [] }) function subscribeDirectory( native: typeof import("@parcel/watcher") | undefined, backend: ParcelWatcher.BackendType | undefined, directory: string, ignore: string[], pubsub: PubSub.PubSub, ) { if (!native || !backend) { return Effect.logError("watcher backend not supported", { directory, platform: process.platform }).pipe( Effect.as(undefined), ) } const callback: ParcelWatcher.SubscribeCallback = (error, updates) => { if (error) Effect.runFork(Effect.logError("watcher callback failed", { error })) for (const update of updates) PubSub.publishUnsafe(pubsub, update) } const pending = native.subscribe(directory, callback, { ignore, backend }) return Effect.promise(() => pending).pipe( Effect.timeout(SUBSCRIBE_TIMEOUT_MS), Effect.catchCause((cause) => { pending.then((subscription) => subscription.unsubscribe()).catch(() => {}) return Effect.logError("failed to subscribe", { directory, cause: Cause.pretty(cause), }).pipe(Effect.as(undefined)) }), ) }