refactor(core): consolidate filesystem services (#30447)

packages/core/src/filesystem/watcher.ts

@@ -0,0 +1,140 @@
+export * as Watcher from "./watcher"
+
+// @ts-ignore
+import { createWrapper } from "@parcel/watcher/wrapper"
+import type ParcelWatcher from "@parcel/watcher"
+import { Cause, Context, Effect, Layer, Schema } from "effect"
+import path from "path"
+import { Config } from "../config"
+import { EventV2 } from "../event"
+import { Flag } from "../flag/flag"
+import { FSUtil } from "../fs-util"
+import { Git } from "../git"
+import { Location } from "../location"
+import { lazy } from "../util/lazy"
+import * as Log from "../util/log"
+import { Ignore } from "./ignore"
+import { Protected } from "./protected"
+
+declare const OPENCODE_LIBC: string | undefined
+
+const log = Log.create({ service: "file.watcher" })
+const SUBSCRIBE_TIMEOUT_MS = 10_000
+
+export const Event = {
+  Updated: EventV2.define({
+    type: "file.watcher.updated",
+    schema: {
+      file: Schema.String,
+      event: Schema.Literals(["add", "change", "unlink"]),
+    },
+  }),
+}
+
+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 (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"
+}
+
+function protecteds(dir: string) {
+  return Protected.paths().filter((item) => {
+    const relative = path.relative(dir, item)
+    return relative !== "" && !relative.startsWith("..") && !path.isAbsolute(relative)
+  })
+}
+
+export const hasNativeBinding = () => !!watcher()
+
+export interface Interface {}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/v2/FileWatcher") {}
+
+export const layer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    if (yield* Flag.OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER) return Service.of({})
+
+    const backend = getBackend()
+    const location = yield* Location.Service
+    if (!backend) {
+      log.error("watcher backend not supported", { directory: location.directory, platform: process.platform })
+      return Service.of({})
+    }
+
+    const w = watcher()
+    if (!w) return Service.of({})
+
+    log.info("watcher backend", { directory: location.directory, platform: process.platform, backend })
+    const events = yield* EventV2.Service
+    const fs = yield* FSUtil.Service
+    const git = yield* Git.Service
+    const context = yield* Effect.context()
+    const runFork = Effect.runForkWith(context)
+    const subscriptions: ParcelWatcher.AsyncSubscription[] = []
+    yield* Effect.addFinalizer(() =>
+      Effect.promise(() => Promise.allSettled(subscriptions.map((subscription) => subscription.unsubscribe()))),
+    )
+
+    const callback: ParcelWatcher.SubscribeCallback = (_error, updates) => {
+      for (const update of updates) {
+        if (update.type === "create") runFork(events.publish(Event.Updated, { file: update.path, event: "add" }))
+        if (update.type === "update") runFork(events.publish(Event.Updated, { file: update.path, event: "change" }))
+        if (update.type === "delete") runFork(events.publish(Event.Updated, { file: update.path, event: "unlink" }))
+      }
+    }
+
+    const subscribe = (directory: string, ignore: string[]) => {
+      const pending = w.subscribe(directory, callback, { ignore, backend })
+      return Effect.promise(() => pending).pipe(
+        Effect.tap((subscription) => Effect.sync(() => subscriptions.push(subscription))),
+        Effect.timeout(SUBSCRIBE_TIMEOUT_MS),
+        Effect.catchCause((cause) => {
+          log.error("failed to subscribe", { directory, cause: Cause.pretty(cause) })
+          pending.then((subscription) => subscription.unsubscribe()).catch(() => {})
+          return Effect.void
+        }),
+      )
+    }
+
+    const config = (yield* (yield* Config.Service).get()).flatMap((item) => item.info.watcher?.ignore ?? [])
+    if (yield* Flag.OPENCODE_EXPERIMENTAL_FILEWATCHER) {
+      yield* Effect.forkScoped(
+        subscribe(location.directory, [...Ignore.PATTERNS, ...config, ...protecteds(location.directory)]),
+      )
+    }
+
+    if (location.vcs?.type === "git") {
+      const resolved = yield* git.dir(location.directory)
+      const vcs = resolved ? yield* fs.realPath(resolved).pipe(Effect.catch(() => Effect.succeed(resolved))) : undefined
+      if (vcs && !config.includes(".git") && !config.includes(vcs) && (!resolved || !config.includes(resolved))) {
+        const ignore = (yield* fs.readDirectoryEntries(vcs).pipe(Effect.catch(() => Effect.succeed([])))).flatMap(
+          (entry) => (entry.name === "HEAD" ? [] : [entry.name]),
+        )
+        yield* Effect.forkScoped(subscribe(vcs, ignore))
+      }
+    }
+
+    return Service.of({})
+  }).pipe(
+    Effect.catchCause((cause) => {
+      log.error("failed to init watcher service", { cause: Cause.pretty(cause) })
+      return Effect.succeed(Service.of({}))
+    }),
+  ),
+)
+
+export const locationLayer = layer.pipe(Layer.provide(Config.locationLayer), Layer.provide(Git.defaultLayer))