330 lines
8.9 KiB
TypeScript
330 lines
8.9 KiB
TypeScript
import * as smartrx from '@push.rocks/smartrx';
|
|
import type { IWatcher, IWatcherOptions, IWatchEvent, TWatchEventType } from './interfaces.js';
|
|
|
|
// Type definitions for Deno APIs (these exist at runtime in Deno)
|
|
declare const Deno: {
|
|
watchFs(paths: string | string[], options?: { recursive?: boolean }): AsyncIterable<{
|
|
kind: 'create' | 'modify' | 'remove' | 'access' | 'any' | 'other';
|
|
paths: string[];
|
|
flag?: { rescan: boolean };
|
|
}> & { close(): void };
|
|
stat(path: string): Promise<{
|
|
isFile: boolean;
|
|
isDirectory: boolean;
|
|
isSymlink: boolean;
|
|
size: number;
|
|
mtime: Date | null;
|
|
atime: Date | null;
|
|
birthtime: Date | null;
|
|
mode: number | null;
|
|
uid: number | null;
|
|
gid: number | null;
|
|
}>;
|
|
lstat(path: string): Promise<{
|
|
isFile: boolean;
|
|
isDirectory: boolean;
|
|
isSymlink: boolean;
|
|
size: number;
|
|
mtime: Date | null;
|
|
atime: Date | null;
|
|
birthtime: Date | null;
|
|
mode: number | null;
|
|
uid: number | null;
|
|
gid: number | null;
|
|
}>;
|
|
readDir(path: string): AsyncIterable<{
|
|
name: string;
|
|
isFile: boolean;
|
|
isDirectory: boolean;
|
|
isSymlink: boolean;
|
|
}>;
|
|
};
|
|
|
|
/**
|
|
* Convert Deno stat to Node.js-like Stats object
|
|
*/
|
|
function denoStatToNodeStats(denoStat: Awaited<ReturnType<typeof Deno.stat>>): any {
|
|
return {
|
|
isFile: () => denoStat.isFile,
|
|
isDirectory: () => denoStat.isDirectory,
|
|
isSymbolicLink: () => denoStat.isSymlink,
|
|
size: denoStat.size,
|
|
mtime: denoStat.mtime,
|
|
atime: denoStat.atime,
|
|
birthtime: denoStat.birthtime,
|
|
mode: denoStat.mode,
|
|
uid: denoStat.uid,
|
|
gid: denoStat.gid
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Deno file watcher using native Deno.watchFs API
|
|
*/
|
|
export class DenoWatcher implements IWatcher {
|
|
private watcher: ReturnType<typeof Deno.watchFs> | null = null;
|
|
private watchedFiles: Set<string> = new Set();
|
|
private _isWatching = false;
|
|
private abortController: AbortController | null = null;
|
|
private recentEvents: Map<string, number> = new Map();
|
|
private throttleMs = 50;
|
|
private pendingWrites: Map<string, ReturnType<typeof setTimeout>> = new Map();
|
|
|
|
public readonly events$ = new smartrx.rxjs.Subject<IWatchEvent>();
|
|
|
|
constructor(private options: IWatcherOptions) {}
|
|
|
|
get isWatching(): boolean {
|
|
return this._isWatching;
|
|
}
|
|
|
|
async start(): Promise<void> {
|
|
if (this._isWatching) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
this.abortController = new AbortController();
|
|
|
|
// Start watching all base paths
|
|
this.watcher = Deno.watchFs(this.options.basePaths, { recursive: true });
|
|
this._isWatching = true;
|
|
|
|
// Perform initial scan
|
|
for (const basePath of this.options.basePaths) {
|
|
await this.scanDirectory(basePath, 0);
|
|
}
|
|
|
|
// Emit ready event
|
|
this.events$.next({ type: 'ready', path: '' });
|
|
|
|
// Start processing events
|
|
this.processEvents();
|
|
} catch (error: any) {
|
|
this.events$.next({ type: 'error', path: '', error });
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async stop(): Promise<void> {
|
|
this._isWatching = false;
|
|
|
|
// Cancel all pending write stabilizations
|
|
for (const timeout of this.pendingWrites.values()) {
|
|
clearTimeout(timeout);
|
|
}
|
|
this.pendingWrites.clear();
|
|
|
|
// Close the watcher
|
|
if (this.watcher) {
|
|
(this.watcher as any).close();
|
|
this.watcher = null;
|
|
}
|
|
|
|
this.watchedFiles.clear();
|
|
this.recentEvents.clear();
|
|
}
|
|
|
|
/**
|
|
* Process events from the Deno watcher
|
|
*/
|
|
private async processEvents(): Promise<void> {
|
|
if (!this.watcher) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
for await (const event of this.watcher) {
|
|
if (!this._isWatching) {
|
|
break;
|
|
}
|
|
|
|
for (const filePath of event.paths) {
|
|
await this.handleDenoEvent(event.kind, filePath);
|
|
}
|
|
}
|
|
} catch (error: any) {
|
|
if (this._isWatching) {
|
|
this.events$.next({ type: 'error', path: '', error });
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle a Deno file system event
|
|
*/
|
|
private async handleDenoEvent(
|
|
kind: 'create' | 'modify' | 'remove' | 'access' | 'any' | 'other',
|
|
filePath: string
|
|
): Promise<void> {
|
|
// Ignore 'access' events (just reading the file)
|
|
if (kind === 'access') {
|
|
return;
|
|
}
|
|
|
|
// Throttle duplicate events
|
|
if (!this.shouldEmit(filePath, kind)) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
if (kind === 'create') {
|
|
const stats = await this.statSafe(filePath);
|
|
if (stats) {
|
|
// Wait for write to stabilize
|
|
await this.waitForWriteFinish(filePath);
|
|
const finalStats = await this.statSafe(filePath);
|
|
|
|
if (finalStats) {
|
|
this.watchedFiles.add(filePath);
|
|
const eventType: TWatchEventType = finalStats.isDirectory() ? 'addDir' : 'add';
|
|
this.events$.next({ type: eventType, path: filePath, stats: finalStats });
|
|
}
|
|
}
|
|
} else if (kind === 'modify') {
|
|
const stats = await this.statSafe(filePath);
|
|
if (stats && !stats.isDirectory()) {
|
|
// Wait for write to stabilize
|
|
await this.waitForWriteFinish(filePath);
|
|
const finalStats = await this.statSafe(filePath);
|
|
|
|
if (finalStats) {
|
|
this.events$.next({ type: 'change', path: filePath, stats: finalStats });
|
|
}
|
|
}
|
|
} else if (kind === 'remove') {
|
|
const wasDirectory = this.isKnownDirectory(filePath);
|
|
this.watchedFiles.delete(filePath);
|
|
this.events$.next({
|
|
type: wasDirectory ? 'unlinkDir' : 'unlink',
|
|
path: filePath
|
|
});
|
|
}
|
|
} catch (error: any) {
|
|
this.events$.next({ type: 'error', path: filePath, error });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Wait for file write to complete (polling-based)
|
|
*/
|
|
private async waitForWriteFinish(filePath: string): Promise<void> {
|
|
return new Promise((resolve) => {
|
|
let lastSize = -1;
|
|
let lastChange = Date.now();
|
|
|
|
const poll = async () => {
|
|
try {
|
|
const stats = await this.statSafe(filePath);
|
|
if (!stats) {
|
|
resolve();
|
|
return;
|
|
}
|
|
|
|
const now = Date.now();
|
|
if (stats.size !== lastSize) {
|
|
lastSize = stats.size;
|
|
lastChange = now;
|
|
this.pendingWrites.set(filePath, setTimeout(poll, this.options.pollInterval));
|
|
} else if (now - lastChange >= this.options.stabilityThreshold) {
|
|
this.pendingWrites.delete(filePath);
|
|
resolve();
|
|
} else {
|
|
this.pendingWrites.set(filePath, setTimeout(poll, this.options.pollInterval));
|
|
}
|
|
} catch {
|
|
this.pendingWrites.delete(filePath);
|
|
resolve();
|
|
}
|
|
};
|
|
|
|
this.pendingWrites.set(filePath, setTimeout(poll, this.options.pollInterval));
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Scan directory and emit 'add' events for existing files
|
|
*/
|
|
private async scanDirectory(dirPath: string, depth: number): Promise<void> {
|
|
if (depth > this.options.depth) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
for await (const entry of Deno.readDir(dirPath)) {
|
|
const fullPath = `${dirPath}/${entry.name}`;
|
|
const stats = await this.statSafe(fullPath);
|
|
|
|
if (!stats) {
|
|
continue;
|
|
}
|
|
|
|
if (entry.isDirectory) {
|
|
this.watchedFiles.add(fullPath);
|
|
this.events$.next({ type: 'addDir', path: fullPath, stats });
|
|
await this.scanDirectory(fullPath, depth + 1);
|
|
} else if (entry.isFile) {
|
|
this.watchedFiles.add(fullPath);
|
|
this.events$.next({ type: 'add', path: fullPath, stats });
|
|
}
|
|
}
|
|
} catch (error: any) {
|
|
if (error.code !== 'ENOENT' && error.code !== 'EACCES') {
|
|
this.events$.next({ type: 'error', path: dirPath, error });
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Safely stat a path, returning null if it doesn't exist
|
|
*/
|
|
private async statSafe(filePath: string): Promise<any | null> {
|
|
try {
|
|
const statFn = this.options.followSymlinks ? Deno.stat : Deno.lstat;
|
|
const denoStats = await statFn(filePath);
|
|
return denoStatToNodeStats(denoStats);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Check if a path was known to be a directory
|
|
*/
|
|
private isKnownDirectory(filePath: string): boolean {
|
|
for (const watched of this.watchedFiles) {
|
|
if (watched.startsWith(filePath + '/')) {
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Throttle duplicate events
|
|
*/
|
|
private shouldEmit(filePath: string, eventType: string): boolean {
|
|
const key = `${filePath}:${eventType}`;
|
|
const now = Date.now();
|
|
const lastEmit = this.recentEvents.get(key);
|
|
|
|
if (lastEmit && now - lastEmit < this.throttleMs) {
|
|
return false;
|
|
}
|
|
|
|
this.recentEvents.set(key, now);
|
|
|
|
// Clean up old entries periodically
|
|
if (this.recentEvents.size > 1000) {
|
|
const cutoff = now - this.throttleMs * 2;
|
|
for (const [k, time] of this.recentEvents) {
|
|
if (time < cutoff) {
|
|
this.recentEvents.delete(k);
|
|
}
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
}
|