|
| 1 | +export * as FileWatch from "./file-watch.js" |
| 2 | + |
| 3 | +import { makeGlobalNode } from "@opencode/util/effect/app-node" |
| 4 | +import { FSUtil } from "@opencode/util/fs-util" |
| 5 | +import { FileSystem } from "@opencode/schema/filesystem" |
| 6 | +import type { Location } from "@opencode/schema/location" |
| 7 | +import { |
| 8 | + Cause, |
| 9 | + Clock, |
| 10 | + Context, |
| 11 | + Duration, |
| 12 | + Effect, |
| 13 | + Exit, |
| 14 | + Layer, |
| 15 | + Schedule, |
| 16 | + Schema, |
| 17 | + Scope, |
| 18 | + Semaphore, |
| 19 | + Stream, |
| 20 | +} from "effect" |
| 21 | +import { Bus } from "../bus.js" |
| 22 | +import { Watcher } from "./watcher.js" |
| 23 | + |
| 24 | +export const TTL = Duration.hours(24) |
| 25 | +export const SWEEP_INTERVAL = Duration.minutes(5) |
| 26 | +export const MAX_PER_LOCATION = 256 |
| 27 | + |
| 28 | +export class DisabledError extends Schema.TaggedError<DisabledError>()("FileWatch.DisabledError", {}) { |
| 29 | + override get message() { |
| 30 | + return "File watching is disabled" |
| 31 | + } |
| 32 | +} |
| 33 | + |
| 34 | +export type Input = { |
| 35 | + readonly location: Location.Ref |
| 36 | + /** Absolute path of the file to watch. */ |
| 37 | + readonly file: string |
| 38 | +} |
| 39 | + |
| 40 | +export interface Interface { |
| 41 | + /** Starts watching a file, or resets its expiry when it is already watched. */ |
| 42 | + readonly watch: (input: Input) => Effect.Effect<void, DisabledError> |
| 43 | + readonly unwatch: (input: Input) => Effect.Effect<void> |
| 44 | +} |
| 45 | + |
| 46 | +export class Service extends Context.Service<Service, Interface>()("@opencode/FileWatch") {} |
| 47 | + |
| 48 | +export const Options = Schema.Struct({ |
| 49 | + enabled: Schema.optional(Schema.Boolean), |
| 50 | +}) |
| 51 | +export type Options = typeof Options.Type |
| 52 | + |
| 53 | +type Entry = { expires: number; scope: Scope.Closeable } |
| 54 | + |
| 55 | +// Global rather than location-scoped so watches outlive idle location eviction. |
| 56 | +export const layer = (options?: Options) => |
| 57 | + Layer.effect( |
| 58 | + Service, |
| 59 | + Effect.gen(function* () { |
| 60 | + if (options?.enabled === false) { |
| 61 | + return Service.of({ watch: () => Effect.fail(new DisabledError()), unwatch: () => Effect.void }) |
| 62 | + } |
| 63 | + const watcher = yield* Watcher.Service |
| 64 | + const bus = yield* Bus.Service |
| 65 | + const fs = yield* FSUtil.Service |
| 66 | + const scope = yield* Effect.scope |
| 67 | + const lock = yield* Semaphore.make(1) |
| 68 | + const locations = new Map<string, Map<string, Entry>>() |
| 69 | + const locationKey = (location: Location.Ref) => JSON.stringify([location.directory, location.workspaceID]) |
| 70 | + |
| 71 | + const remove = (key: string, file: string) => { |
| 72 | + const files = locations.get(key) |
| 73 | + const entry = files?.get(file) |
| 74 | + if (!files || !entry) return |
| 75 | + files.delete(file) |
| 76 | + if (files.size === 0) locations.delete(key) |
| 77 | + return entry |
| 78 | + } |
| 79 | + |
| 80 | + const stop = (key: string, file: string) => |
| 81 | + Effect.suspend(() => { |
| 82 | + const entry = remove(key, file) |
| 83 | + if (!entry) return Effect.void |
| 84 | + return Scope.close(entry.scope, Exit.void) |
| 85 | + }) |
| 86 | + |
| 87 | + const start = Effect.fnUntraced(function* (key: string, input: Input) { |
| 88 | + const child = yield* Scope.fork(scope) |
| 89 | + yield* Effect.gen(function* () { |
| 90 | + const updates = yield* watcher.subscribe({ path: input.file, type: "file" }) |
| 91 | + yield* Stream.runForEach(updates, () => |
| 92 | + Effect.gen(function* () { |
| 93 | + const exists = yield* fs.existsSafe(input.file) |
| 94 | + yield* bus.publish( |
| 95 | + FileSystem.Event.Changed, |
| 96 | + { file: input.file, event: exists ? "change" : "unlink" }, |
| 97 | + { location: input.location }, |
| 98 | + ) |
| 99 | + }), |
| 100 | + ) |
| 101 | + }).pipe( |
| 102 | + Effect.catchCauseIf( |
| 103 | + (cause) => !Cause.hasInterrupts(cause), |
| 104 | + (cause) => Effect.logError("file watch failed", { file: input.file, cause }), |
| 105 | + ), |
| 106 | + // The watch ended or failed (e.g. its directory vanished): forget it so the next watch starts a new one. |
| 107 | + // Interrupts skip this. The scope closes from another fiber because closing it interrupts this one. |
| 108 | + Effect.andThen( |
| 109 | + lock.withPermit( |
| 110 | + Effect.suspend(() => { |
| 111 | + if (locations.get(key)?.get(input.file)?.scope !== child) return Effect.void |
| 112 | + remove(key, input.file) |
| 113 | + return Scope.close(child, Exit.void).pipe(Effect.forkIn(scope), Effect.asVoid) |
| 114 | + }), |
| 115 | + ), |
| 116 | + ), |
| 117 | + Effect.forkIn(child, { startImmediately: true }), |
| 118 | + ) |
| 119 | + return child |
| 120 | + }) |
| 121 | + |
| 122 | + const sweep = lock.withPermit( |
| 123 | + Effect.gen(function* () { |
| 124 | + const now = yield* Clock.currentTimeMillis |
| 125 | + const expired = [...locations].flatMap(([key, files]) => |
| 126 | + [...files].filter(([, entry]) => entry.expires <= now).map(([file]) => [key, file] as const), |
| 127 | + ) |
| 128 | + yield* Effect.forEach(expired, ([key, file]) => stop(key, file), { discard: true }) |
| 129 | + }), |
| 130 | + ) |
| 131 | + yield* sweep.pipe(Effect.repeat(Schedule.spaced(SWEEP_INTERVAL)), Effect.forkScoped) |
| 132 | + |
| 133 | + return Service.of({ |
| 134 | + watch: (input) => |
| 135 | + lock.withPermit( |
| 136 | + Effect.gen(function* () { |
| 137 | + const key = locationKey(input.location) |
| 138 | + const expires = (yield* Clock.currentTimeMillis) + Duration.toMillis(TTL) |
| 139 | + const existing = locations.get(key)?.get(input.file) |
| 140 | + if (existing) { |
| 141 | + existing.expires = expires |
| 142 | + return |
| 143 | + } |
| 144 | + const full = locations.get(key) |
| 145 | + if (full && full.size >= MAX_PER_LOCATION) { |
| 146 | + const soonest = [...full].reduce((a, b) => (b[1].expires < a[1].expires ? b : a)) |
| 147 | + yield* stop(key, soonest[0]) |
| 148 | + } |
| 149 | + const entry = { expires, scope: yield* start(key, input) } |
| 150 | + const files = locations.get(key) ?? new Map<string, Entry>() |
| 151 | + files.set(input.file, entry) |
| 152 | + locations.set(key, files) |
| 153 | + }), |
| 154 | + ), |
| 155 | + unwatch: (input) => lock.withPermit(stop(locationKey(input.location), input.file)), |
| 156 | + }) |
| 157 | + }), |
| 158 | + ) |
| 159 | + |
| 160 | +export function configured(options?: Options) { |
| 161 | + return makeGlobalNode({ service: Service, layer: layer(options), deps: [Watcher.node, Bus.node, FSUtil.node] }) |
| 162 | +} |
| 163 | + |
| 164 | +export const node = configured() |
0 commit comments