1// The indexer service: owns the singleton progress root that all scanning
2// and processing operations report into, watches the file store, runs
3// periodic full sweeps, and pings web nodes when the database changes.
4//
5// One progress root for the lifetime of the process; it is never ended.
6// `GET /progress` on the source of truth attaches encoders to it, which
7// snapshot current state on connect and then stream deltas.
8const console = log.scoped("indexer");
9
10/** the singleton progress root. */
11export const root = new progress.Root();
12
13/** bumped on every database mutation; `GET /db` uses this to invalidate */
14export let generation = 0;
15
16// -- database change events --
17// consumers (the /db/events sse route) subscribe here; the indexer emits a
18// debounced beacon after changes. anything else that swaps the database
19// (e.g. the /reload route) can emit through `emitDbChange`.
20const dbSubscribers = new Set<(generation: number) => void>();
21
22/** subscribe to database-changed beacons. dispose to unsubscribe. */
23export function subscribeDbChange(fn: (generation: number) => void): ts.Dispose {
24 dbSubscribers.add(fn);
25 return ts.defer(() => void dbSubscribers.delete(fn));
26}
27
28/** notify all subscribers that the database changed */
29export function emitDbChange() {
30 generation += 1;
31 for (const fn of dbSubscribers) fn(generation);
32}
33
34export interface ServiceOptions {
35 /** raw file store root. default: paths.rawFileRoot */
36 root?: string;
37 /** full sweep interval. default: CLOVER_SCAN_INTERVAL or 24h */
38 sweepIntervalMs?: number;
39 /** stability window before indexing a changed file */
40 settleMs?: number;
41 /** disable the file watcher (CLOVER_WATCH=0) */
42 watch?: boolean;
43}
44
45let started = false;
46
47export function start(options: ServiceOptions = {}) {
48 ASSERT(!started, "indexer service started twice");
49 started = true;
50
51 const cores = Number(process.env.CLOVER_INDEX_CORES ?? 0);
52 if (cores > 0) queue.setConcurrency(cores);
53
54 const service = new Service(options);
55 service.init();
56 return service;
57}
58
59export class Service {
60 scanner: Scanner;
61 sweepIntervalMs: number;
62 watchEnabled: boolean;
63 watcher: fs.FSWatcher | null = null;
64 lastSweep: { startedAt: number; endedAt: number | null; error: string | null } | null = null;
65
66 // paths that changed according to the watcher, waiting to be scanned
67 #dirty = new Map<string, number>();
68 #dirtyTimer: NodeJS.Timeout | null = null;
69 #sweeping = false;
70
71 constructor(options: ServiceOptions) {
72 this.scanner = new Scanner({
73 root: Path.resolve(options.root ?? paths.rawFileRoot),
74 progress: root,
75 settleMs: options.settleMs
76 ?? Number(process.env.CLOVER_SETTLE_MS ?? 10_000),
77 onChange: (kind) => this.#onDbChange(kind),
78 onIdle: () => this.#onIdle(),
79 });
80 this.sweepIntervalMs = options.sweepIntervalMs
81 ?? parseDuration(process.env.CLOVER_SCAN_INTERVAL ?? "24h");
82 this.watchEnabled = options.watch ?? process.env.CLOVER_WATCH !== "0";
83 }
84
85 init() {
86 void this.sweep("boot");
87 setInterval(() => void this.sweep("periodic"), this.sweepIntervalMs)
88 .unref();
89 if (this.watchEnabled) this.#startWatcher();
90 console.info(
91 `indexer service started (root: ${this.scanner.root}, `
92 + `sweep every ${string.formatDurationLetters(this.sweepIntervalMs / 1000)}, `
93 + `watch: ${this.watchEnabled})`,
94 );
95 }
96
97 /** run a full sweep (or a subtree scan when `subPath` is given) */
98 async sweep(reason: string, subPath?: string) {
99 if (this.#sweeping && !subPath) {
100 console.warn(`skipping ${reason} sweep; one is already running`);
101 return;
102 }
103 try {
104 if (!subPath) {
105 this.#sweeping = true;
106 this.lastSweep = { startedAt: Date.now() / 1000, endedAt: null, error: null };
107 }
108 const target = subPath
109 ? this.scanner.root.join("." + path.posix.normalize("/" + subPath))
110 : this.scanner.root;
111 await this.scanner.scanPath(target);
112 this.#runDirMeta();
113 if (!subPath) await this.#maintenance();
114 } catch (err) {
115 if (!subPath && this.lastSweep) this.lastSweep.error = String(err);
116 console.error(`${reason} scan failed:`, err);
117 } finally {
118 if (!subPath) {
119 this.#sweeping = false;
120 if (this.lastSweep) this.lastSweep.endedAt = Date.now() / 1000;
121 }
122 }
123 }
124
125 // delete derived assets whose last referencing file is gone, plus tmp
126 // directories left behind by crashed producers. folded in from the old
127 // `file-trim` binary.
128 async #maintenance() {
129 using node = this.scanner.indexNode.start("maintenance");
130 const orphaned = derived.findOrphanedRoots();
131 for (const orphan of orphaned) {
132 node.text = `delete orphaned ${orphan.key}`;
133 await derived.deleteRootFiles(orphan);
134 derived.deleteRoot(orphan);
135 }
136 const cleaned = await derived.cleanAbandonedTmp();
137 if (orphaned.length + cleaned > 0) {
138 console.info(
139 `maintenance: ${orphaned.length} orphaned roots, ${cleaned} stale tmp dirs`,
140 );
141 this.#onDbChange("processed");
142 }
143 }
144
145 // -- file watcher --
146
147 #startWatcher() {
148 try {
149 this.watcher = fs.watch(
150 this.scanner.root.toString(),
151 { recursive: true },
152 (_event, subPath) => subPath && this.#markDirty(subPath),
153 );
154 this.watcher.on("error", (err) => {
155 console.error("file watcher died, relying on periodic sweeps:", err);
156 this.watcher = null;
157 });
158 } catch (err) {
159 console.error("file watcher unavailable, relying on sweeps:", err);
160 this.watcher = null;
161 }
162 }
163
164 #markDirty(subPath: string) {
165 subPath = subPath.replaceAll("\\", "/");
166 // ignore events for paths the scanner would skip anyway (.DS_Store and
167 // friends), but also for anything inside a skipped directory.
168 if (subPath.split("/").some((part) => skipBasename(part))) return;
169 this.#dirty.set(subPath, Date.now());
170 this.#dirtyTimer ??= setTimeout(() => {
171 this.#dirtyTimer = null;
172 void this.#flushDirty();
173 }, watchDebounceMs);
174 }
175
176 // targets currently being scanned. a path that keeps emitting events (a
177 // large upload sitting in the settle gate) must not pile up concurrent
178 // scans of itself; it is re-checked once the in-flight scan completes.
179 #scanning = new Map<string, Promise<void>>();
180
181 async #flushDirty() {
182 if (this.#dirty.size === 0) return;
183 const all = [...this.#dirty.keys()];
184 this.#dirty.clear();
185
186 // coalesce: if a parent path is queued, skip its children. scanning is
187 // recursive for directories, so the parent covers them. comparisons are
188 // case-insensitive to match the file store's semantics; rename events
189 // can spell the same physical path two ways.
190 const sorted = all.sort();
191 const targets: string[] = [];
192 for (const p of sorted) {
193 const lp = p.toLowerCase();
194 const covered = targets.some((t) => {
195 const lt = t.toLowerCase();
196 return lp === lt || lp.startsWith(lt + "/");
197 });
198 if (!covered) targets.push(p);
199 }
200
201 const fresh: string[] = [];
202 for (const target of targets) {
203 // defer paths already being scanned; the tail of the flush that owns
204 // them re-arms the timer, which picks these back up.
205 if (this.#scanning.has(target.toLowerCase())) {
206 this.#dirty.set(target, Date.now());
207 } else fresh.push(target);
208 }
209 if (fresh.length === 0) return;
210
211 // parallel: one file mid-upload (settling) must not block the others
212 await Promise.all(fresh.map((target) => {
213 const job = this.scanner
214 .scanPath(this.scanner.root.join(target))
215 .catch((err) => console.error(`watch scan of ${target} failed:`, err))
216 .finally(() => {
217 this.#scanning.delete(target.toLowerCase());
218 });
219 this.#scanning.set(target.toLowerCase(), job);
220 return job;
221 }));
222 this.#runDirMeta();
223
224 // events that arrived during the scans (including deferred re-checks of
225 // the paths scanned just now) get their own flush.
226 if (this.#dirty.size > 0) {
227 this.#dirtyTimer ??= setTimeout(() => {
228 this.#dirtyTimer = null;
229 void this.#flushDirty();
230 }, watchDebounceMs);
231 }
232 }
233
234 // -- directory metadata --
235
236 // the dir meta pass also runs after processors finish, since readme.txt
237 // contents arrive via the text-contents processor.
238 #onIdle() {
239 if (this.#dirMetaTimer) return;
240 this.#dirMetaTimer = setTimeout(() => {
241 this.#dirMetaTimer = null;
242 this.#runDirMeta();
243 }, 1000);
244 this.#dirMetaTimer.unref();
245 }
246 #dirMetaTimer: NodeJS.Timeout | null = null;
247
248 #runDirMeta() {
249 try {
250 if (dirmeta.run(this.scanner.indexNode)) this.#onDbChange("metadata");
251 } catch (err) {
252 console.error("directory metadata pass failed:", err);
253 }
254 }
255
256 // -- database change beacons --
257 // consumers subscribe via /db/events and pull /db when beaconed. metadata
258 // changes (new files) flush fast; processor completions are batched
259 // coarsely since they arrive in bursts during encodes.
260
261 #beaconTimer: NodeJS.Timeout | null = null;
262 #beaconDeadline = Infinity;
263
264 #onDbChange(kind: ChangeKind) {
265 const delay = kind === "metadata" ? metadataBeaconMs : processedBeaconMs;
266 const deadline = Date.now() + delay;
267 if (deadline < this.#beaconDeadline) {
268 this.#beaconDeadline = deadline;
269 if (this.#beaconTimer) clearTimeout(this.#beaconTimer);
270 this.#beaconTimer = setTimeout(() => {
271 this.#beaconTimer = null;
272 this.#beaconDeadline = Infinity;
273 emitDbChange();
274 }, delay);
275 this.#beaconTimer.unref();
276 }
277 }
278}
279
280export function parseDuration(text: string): number {
281 const match = text.match(/^(\d+(?:\.\d+)?)\s*(ms|s|m|h|d)?$/);
282 if (!match) throw new Error(`cannot parse duration: ${JSON.stringify(text)}`);
283 const scale = { ms: 1, s: 1000, m: 60_000, h: 3_600_000, d: 86_400_000 };
284 return Number(match[1]) * scale[(match[2] ?? "ms") as keyof typeof scale];
285}
286
287const watchDebounceMs = 1000;
288const metadataBeaconMs = 2_000;
289const processedBeaconMs = 30_000;
290
291import * as fs from "node:fs";
292import * as path from "node:path";
293
294import { Path } from "#sitegen/path";
295import { ASSERT } from "@clo/lib/assert";
296import * as log from "@clo/lib/log";
297import * as progress from "@clo/lib/progress";
298import * as queue from "@clo/lib/queue";
299import * as string from "@clo/lib/string";
300import * as ts from "@clo/lib/ts";
301
302import * as derived from "#src/file-viewer/models/derived.ts";
303import * as paths from "#src/file-viewer/paths.ts";
304import * as dirmeta from "./dirmeta.ts";
305import { type ChangeKind, Scanner, skipBasename } from "./scan.ts";