Files
smartwatch/ts/watchers/watcher.deno.ts

291 lines
7.8 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;
// Debounce: pending emits per file path
private pendingEmits: Map<string, ReturnType<typeof setTimeout>> = new Map();
public readonly events$ = new smartrx.rxjs.Subject<IWatchEvent>();
constructor(private options: IWatcherOptions) {}
/**
* Check if a file is a temporary file created by editors
*/
private isTemporaryFile(filePath: string): boolean {
const basename = filePath.split('/').pop() || '';
// Editor temp files: *.tmp.*, *.swp, *.swx, *~, .#*
if (basename.includes('.tmp.')) return true;
if (basename.endsWith('.swp') || basename.endsWith('.swx')) return true;
if (basename.endsWith('~')) return true;
if (basename.startsWith('.#')) return true;
return false;
}
get isWatching(): boolean {
return this._isWatching;
}
async start(): Promise<void> {
if (this._isWatching) {
return;
}
try {
// 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 debounced emits
for (const timeout of this.pendingEmits.values()) {
clearTimeout(timeout);
}
this.pendingEmits.clear();
// Close the watcher
if (this.watcher) {
(this.watcher as any).close();
this.watcher = null;
}
this.watchedFiles.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) {
this.handleDenoEvent(event.kind, filePath);
}
}
} catch (error: any) {
if (this._isWatching) {
this.events$.next({ type: 'error', path: '', error });
}
}
}
/**
* Handle a Deno file system event - debounce and normalize
*/
private handleDenoEvent(
kind: 'create' | 'modify' | 'remove' | 'access' | 'any' | 'other',
filePath: string
): void {
// Ignore 'access' events (just reading the file)
if (kind === 'access') {
return;
}
// Skip temporary files created by editors (atomic saves)
if (this.isTemporaryFile(filePath)) {
return;
}
// Debounce: cancel any pending emit for this file
const existing = this.pendingEmits.get(filePath);
if (existing) {
clearTimeout(existing);
}
// Schedule debounced emit
const timeout = setTimeout(() => {
this.pendingEmits.delete(filePath);
this.emitFileEvent(filePath, kind);
}, this.options.debounceMs);
this.pendingEmits.set(filePath, timeout);
}
/**
* Emit the actual file event after debounce
*/
private async emitFileEvent(
filePath: string,
kind: 'create' | 'modify' | 'remove' | 'access' | 'any' | 'other'
): Promise<void> {
try {
if (kind === 'create') {
const stats = await this.statSafe(filePath);
if (stats) {
this.watchedFiles.add(filePath);
const eventType: TWatchEventType = stats.isDirectory() ? 'addDir' : 'add';
this.events$.next({ type: eventType, path: filePath, stats });
}
} else if (kind === 'modify') {
const stats = await this.statSafe(filePath);
if (stats && !stats.isDirectory()) {
this.events$.next({ type: 'change', path: filePath, stats });
}
} 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 });
}
}
/**
* 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}`;
// Skip temp files during initial scan too
if (this.isTemporaryFile(fullPath)) {
continue;
}
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;
}
}