BREAKING CHANGE(smartwatch): Introduce Smartwatch: cross-runtime native file watching for Node.js, Deno and Bun; rename smartchok to smartwatch and bump major version to 2.0.0

This commit is contained in:
2025-11-30 03:04:49 +00:00
parent aab3ce213b
commit 0f17be179c
16 changed files with 1011 additions and 162 deletions

33
ts/watchers/index.ts Normal file
View File

@@ -0,0 +1,33 @@
import { Smartenv } from '@push.rocks/smartenv';
import type { IWatcher, IWatcherOptions, IWatchEvent, TWatchEventType } from './interfaces.js';
export type { IWatcher, IWatcherOptions, IWatchEvent, TWatchEventType };
/**
* Creates a platform-appropriate file watcher based on the current runtime
* Uses @push.rocks/smartenv for runtime detection
*/
export async function createWatcher(options: IWatcherOptions): Promise<IWatcher> {
const env = new Smartenv();
if (env.isDeno) {
// Deno runtime - use Deno.watchFs
const { DenoWatcher } = await import('./watcher.deno.js');
return new DenoWatcher(options);
} else {
// Node.js or Bun - both use fs.watch (Bun has Node.js compatibility)
const { NodeWatcher } = await import('./watcher.node.js');
return new NodeWatcher(options);
}
}
/**
* Default watcher options
*/
export const defaultWatcherOptions: IWatcherOptions = {
basePaths: [],
depth: 4,
followSymlinks: false,
stabilityThreshold: 300,
pollInterval: 100
};

47
ts/watchers/interfaces.ts Normal file
View File

@@ -0,0 +1,47 @@
import type * as fs from 'fs';
import type * as smartrx from '@push.rocks/smartrx';
/**
* File system event types that the watcher emits
*/
export type TWatchEventType = 'add' | 'addDir' | 'change' | 'unlink' | 'unlinkDir' | 'ready' | 'error';
/**
* Data structure for watch events
*/
export interface IWatchEvent {
type: TWatchEventType;
path: string;
stats?: fs.Stats;
error?: Error;
}
/**
* Options for creating a watcher
*/
export interface IWatcherOptions {
/** Base paths to watch (extracted from glob patterns) */
basePaths: string[];
/** Maximum directory depth to watch */
depth: number;
/** Whether to follow symbolic links */
followSymlinks: boolean;
/** Stability threshold for write detection (ms) */
stabilityThreshold: number;
/** Poll interval for write detection (ms) */
pollInterval: number;
}
/**
* Common interface for file watchers across runtimes
*/
export interface IWatcher {
/** Start watching files */
start(): Promise<void>;
/** Stop watching and clean up */
stop(): Promise<void>;
/** Whether the watcher is currently active */
readonly isWatching: boolean;
/** Subject that emits watch events */
readonly events$: smartrx.rxjs.Subject<IWatchEvent>;
}

329
ts/watchers/watcher.deno.ts Normal file
View File

@@ -0,0 +1,329 @@
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;
}
}

281
ts/watchers/watcher.node.ts Normal file
View File

@@ -0,0 +1,281 @@
import * as fs from 'fs';
import * as path from 'path';
import * as smartrx from '@push.rocks/smartrx';
import type { IWatcher, IWatcherOptions, IWatchEvent, TWatchEventType } from './interfaces.js';
import { WriteStabilizer } from '../utils/write-stabilizer.js';
/**
* Node.js/Bun file watcher using native fs.watch API
*/
export class NodeWatcher implements IWatcher {
private watchers: Map<string, fs.FSWatcher> = new Map();
private watchedFiles: Set<string> = new Set();
private _isWatching = false;
private writeStabilizer: WriteStabilizer;
private recentEvents: Map<string, number> = new Map();
private throttleMs = 50;
public readonly events$ = new smartrx.rxjs.Subject<IWatchEvent>();
constructor(private options: IWatcherOptions) {
this.writeStabilizer = new WriteStabilizer(
options.stabilityThreshold,
options.pollInterval
);
}
get isWatching(): boolean {
return this._isWatching;
}
async start(): Promise<void> {
if (this._isWatching) {
return;
}
try {
// Start watching each base path
for (const basePath of this.options.basePaths) {
await this.watchPath(basePath, 0);
}
this._isWatching = true;
// Perform initial scan to emit 'add' events for existing files
for (const basePath of this.options.basePaths) {
await this.scanDirectory(basePath, 0);
}
// Emit ready event
this.events$.next({ type: 'ready', path: '' });
} catch (error: any) {
this.events$.next({ type: 'error', path: '', error });
throw error;
}
}
async stop(): Promise<void> {
this._isWatching = false;
this.writeStabilizer.cancelAll();
// Close all watchers
for (const [watchPath, watcher] of this.watchers) {
watcher.close();
}
this.watchers.clear();
this.watchedFiles.clear();
this.recentEvents.clear();
}
/**
* Start watching a path (file or directory)
*/
private async watchPath(watchPath: string, depth: number): Promise<void> {
if (depth > this.options.depth) {
return;
}
try {
const stats = await this.statSafe(watchPath);
if (!stats) {
return;
}
if (stats.isDirectory()) {
// Watch the directory with recursive option (Node.js 20+ supports this on all platforms)
const watcher = fs.watch(
watchPath,
{ recursive: true, persistent: true },
(eventType, filename) => {
if (filename) {
this.handleFsEvent(watchPath, filename, eventType);
}
}
);
watcher.on('error', (error) => {
this.events$.next({ type: 'error', path: watchPath, error });
});
this.watchers.set(watchPath, watcher);
}
} catch (error: any) {
this.events$.next({ type: 'error', path: watchPath, error });
}
}
/**
* Handle raw fs.watch events and normalize them
*/
private async handleFsEvent(
basePath: string,
filename: string,
eventType: 'rename' | 'change' | string
): Promise<void> {
const fullPath = path.join(basePath, filename);
// Throttle duplicate events
if (!this.shouldEmit(fullPath, eventType)) {
return;
}
try {
const stats = await this.statSafe(fullPath);
if (eventType === 'rename') {
// 'rename' can mean add or unlink - check if file exists
if (stats) {
// File exists - it's either a new file or was renamed to this location
if (stats.isDirectory()) {
if (!this.watchedFiles.has(fullPath)) {
this.watchedFiles.add(fullPath);
this.events$.next({ type: 'addDir', path: fullPath, stats });
}
} else {
// Wait for write to stabilize before emitting
try {
const stableStats = await this.writeStabilizer.waitForWriteFinish(fullPath);
const wasWatched = this.watchedFiles.has(fullPath);
this.watchedFiles.add(fullPath);
this.events$.next({
type: wasWatched ? 'change' : 'add',
path: fullPath,
stats: stableStats
});
} catch {
// File was deleted during stabilization
if (this.watchedFiles.has(fullPath)) {
this.watchedFiles.delete(fullPath);
this.events$.next({ type: 'unlink', path: fullPath });
}
}
}
} else {
// File doesn't exist - it was deleted
if (this.watchedFiles.has(fullPath)) {
const wasDir = this.isKnownDirectory(fullPath);
this.watchedFiles.delete(fullPath);
this.events$.next({
type: wasDir ? 'unlinkDir' : 'unlink',
path: fullPath
});
}
}
} else if (eventType === 'change') {
// File was modified
if (stats && !stats.isDirectory()) {
try {
const stableStats = await this.writeStabilizer.waitForWriteFinish(fullPath);
// Check if this is a new file (not yet in watchedFiles)
const wasWatched = this.watchedFiles.has(fullPath);
if (!wasWatched) {
// This is actually an 'add' - file wasn't being watched before
this.watchedFiles.add(fullPath);
this.events$.next({ type: 'add', path: fullPath, stats: stableStats });
} else {
this.events$.next({ type: 'change', path: fullPath, stats: stableStats });
}
} catch {
// File was deleted during write
if (this.watchedFiles.has(fullPath)) {
this.watchedFiles.delete(fullPath);
this.events$.next({ type: 'unlink', path: fullPath });
}
}
}
}
} catch (error: any) {
this.events$.next({ type: 'error', path: fullPath, 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 {
const entries = await fs.promises.readdir(dirPath, { withFileTypes: true });
for (const entry of entries) {
const fullPath = path.join(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<fs.Stats | null> {
try {
if (this.options.followSymlinks) {
return await fs.promises.stat(filePath);
} else {
return await fs.promises.lstat(filePath);
}
} catch {
return null;
}
}
/**
* Check if a path was known to be a directory (for proper unlink event type)
*/
private isKnownDirectory(filePath: string): boolean {
// Check if any watched files are children of this path
for (const watched of this.watchedFiles) {
if (watched.startsWith(filePath + path.sep)) {
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;
}
}