230 lines
8.5 KiB
JavaScript
230 lines
8.5 KiB
JavaScript
'use strict';
|
|
|
|
const fs = require('node:fs');
|
|
|
|
// A watched repository is only re-read when the filesystem reports activity. The
|
|
// interval below stays as a safety net for watchers that silently stop
|
|
// delivering, which happens on network shares and removed folders.
|
|
const SAFETY_CHECK_INTERVAL_MS = 30_000;
|
|
const WATCH_DEBOUNCE_MS = 250;
|
|
// Busy trees (a build, an install, a fetch) produce a continuous event stream.
|
|
// This bounds how often that can turn into a Git read.
|
|
const MIN_WATCH_CHECK_INTERVAL_MS = 1_000;
|
|
|
|
class RepositoryMonitor {
|
|
constructor({ store, git, onChange, diagnostics = null }) {
|
|
this.store = store;
|
|
this.git = git;
|
|
this.onChange = onChange;
|
|
this.diagnostics = diagnostics;
|
|
this.paths = [];
|
|
this.fingerprints = new Map();
|
|
this.timer = null;
|
|
this.running = false;
|
|
this.paused = new Set();
|
|
this.active = false;
|
|
this.watchers = new Map();
|
|
this.changed = new Set();
|
|
this.lastCheckedAt = new Map();
|
|
this.lastFetchedAt = new Map();
|
|
this.watchTimer = null;
|
|
this.fetchRunning = false;
|
|
}
|
|
|
|
setPaths(paths) {
|
|
this.paths = [...new Set((paths || []).filter(Boolean))];
|
|
const watched = new Set(this.paths);
|
|
for (const existing of [...this.fingerprints.keys()]) {
|
|
if (!watched.has(existing)) this.fingerprints.delete(existing);
|
|
}
|
|
// A repository that is unlinked while a mutation holds it paused would keep
|
|
// that pause forever, silently freezing its status once it is watched again.
|
|
for (const existing of [...this.paused]) {
|
|
if (!watched.has(existing)) this.paused.delete(existing);
|
|
}
|
|
for (const existing of [...this.changed]) {
|
|
if (!watched.has(existing)) this.changed.delete(existing);
|
|
}
|
|
for (const existing of [...this.lastCheckedAt.keys()]) {
|
|
if (!watched.has(existing)) this.lastCheckedAt.delete(existing);
|
|
}
|
|
for (const existing of [...this.lastFetchedAt.keys()]) {
|
|
if (!watched.has(existing)) this.lastFetchedAt.delete(existing);
|
|
}
|
|
const now = Date.now();
|
|
for (const localPath of this.paths) {
|
|
if (!this.lastFetchedAt.has(localPath)) this.lastFetchedAt.set(localPath, now);
|
|
}
|
|
this.syncWatchers();
|
|
}
|
|
|
|
syncWatchers() {
|
|
for (const [localPath, watcher] of [...this.watchers]) {
|
|
if (this.active && this.paths.includes(localPath)) continue;
|
|
this.closeWatcher(localPath, watcher);
|
|
}
|
|
if (!this.active) return;
|
|
for (const localPath of this.paths) {
|
|
if (this.watchers.has(localPath)) continue;
|
|
try {
|
|
const watcher = fs.watch(
|
|
localPath,
|
|
{ recursive: true, persistent: false },
|
|
() => this.noteFilesystemChange(localPath)
|
|
);
|
|
watcher.on('error', () => this.dropWatcher(localPath));
|
|
this.watchers.set(localPath, watcher);
|
|
} catch {
|
|
// Watching is unavailable for this folder. Leaving it unwatched makes
|
|
// shouldCheck() fall back to the interval for that repository only.
|
|
}
|
|
}
|
|
}
|
|
|
|
closeWatcher(localPath, watcher = this.watchers.get(localPath)) {
|
|
if (!watcher) return;
|
|
try { watcher.close(); } catch { /* already closed */ }
|
|
this.watchers.delete(localPath);
|
|
}
|
|
|
|
dropWatcher(localPath) {
|
|
this.closeWatcher(localPath);
|
|
this.changed.add(localPath);
|
|
}
|
|
|
|
noteFilesystemChange(localPath) {
|
|
this.changed.add(localPath);
|
|
this.scheduleWatchTick();
|
|
}
|
|
|
|
scheduleWatchTick() {
|
|
if (this.watchTimer) return;
|
|
this.watchTimer = setTimeout(() => {
|
|
this.watchTimer = null;
|
|
this.tick().catch((error) => this.diagnostics?.warning('repository-monitor.tick.failed', error));
|
|
}, WATCH_DEBOUNCE_MS);
|
|
this.watchTimer.unref?.();
|
|
}
|
|
|
|
shouldCheck(localPath, now) {
|
|
if (this.paused.has(localPath)) return false;
|
|
if (!this.watchers.has(localPath)) return true;
|
|
const sinceLastCheck = now - (this.lastCheckedAt.get(localPath) || 0);
|
|
if (this.changed.has(localPath)) return sinceLastCheck >= MIN_WATCH_CHECK_INTERVAL_MS;
|
|
return sinceLastCheck >= SAFETY_CHECK_INTERVAL_MS;
|
|
}
|
|
|
|
fetchIntervalMs() {
|
|
const minutes = Number(this.store.data.preferences.fetchIntervalMinutes);
|
|
return Number.isFinite(minutes) && minutes > 0 ? Math.min(minutes, 240) * 60_000 : 0;
|
|
}
|
|
|
|
shouldFetch(localPath, now) {
|
|
const interval = this.fetchIntervalMs();
|
|
return interval > 0
|
|
&& !this.paused.has(localPath)
|
|
&& now - (this.lastFetchedAt.get(localPath) || now) >= interval;
|
|
}
|
|
|
|
async recordStatus(localPath, status, reason) {
|
|
const next = this.git.statusFingerprint(status);
|
|
const previous = this.fingerprints.get(localPath);
|
|
this.fingerprints.set(localPath, next);
|
|
if (previous && previous !== next) {
|
|
await this.diagnostics?.debug('repository-monitor.changed', { localPath, head: status.head, branch: status.branch?.head, counts: status.counts, reason });
|
|
this.onChange?.({ localPath, status, reason });
|
|
}
|
|
}
|
|
|
|
async fetchRemoteUpdates(now = Date.now()) {
|
|
if (this.fetchRunning) return;
|
|
const queue = this.paths.filter((localPath) => this.shouldFetch(localPath, now));
|
|
if (!queue.length) return;
|
|
this.fetchRunning = true;
|
|
try {
|
|
const workers = Array.from({ length: Math.min(2, queue.length) }, async () => {
|
|
while (queue.length) {
|
|
const localPath = queue.shift();
|
|
// Mark the attempt before awaiting the network. A failing remote should
|
|
// not be retried every local poll interval.
|
|
this.lastFetchedAt.set(localPath, Date.now());
|
|
try {
|
|
const result = await this.git.fetch(localPath);
|
|
await this.recordStatus(localPath, result.status, 'remote-state-changed');
|
|
await this.diagnostics?.debug('repository-monitor.fetch.completed', {
|
|
localPath,
|
|
branch: result.status?.branch?.head,
|
|
ahead: result.status?.branch?.ahead,
|
|
behind: result.status?.branch?.behind,
|
|
});
|
|
} catch (error) {
|
|
await this.diagnostics?.warning('repository-monitor.fetch.failed', { localPath, message: error.message });
|
|
}
|
|
}
|
|
});
|
|
await Promise.all(workers);
|
|
} finally {
|
|
this.fetchRunning = false;
|
|
}
|
|
}
|
|
|
|
pause(localPath) { if (localPath) this.paused.add(localPath); }
|
|
resume(localPath) { if (localPath) this.paused.delete(localPath); }
|
|
|
|
restart() {
|
|
this.stop();
|
|
if (!this.store.data.preferences.autoRefresh) return;
|
|
this.active = true;
|
|
this.syncWatchers();
|
|
const seconds = Math.min(Math.max(Number(this.store.data.preferences.repositoryPollSeconds) || 4, 2), 60);
|
|
this.timer = setInterval(() => this.tick().catch((error) => this.diagnostics?.warning('repository-monitor.tick.failed', error)), seconds * 1000);
|
|
this.timer.unref?.();
|
|
}
|
|
|
|
stop() {
|
|
if (this.timer) clearInterval(this.timer);
|
|
this.timer = null;
|
|
if (this.watchTimer) clearTimeout(this.watchTimer);
|
|
this.watchTimer = null;
|
|
this.active = false;
|
|
this.syncWatchers();
|
|
}
|
|
|
|
async tick() {
|
|
void this.fetchRemoteUpdates().catch((error) => this.diagnostics?.warning('repository-monitor.fetch-cycle.failed', error));
|
|
if (this.running || !this.paths.length) return;
|
|
this.running = true;
|
|
try {
|
|
const now = Date.now();
|
|
const queue = this.paths.filter((localPath) => this.shouldCheck(localPath, now));
|
|
const workers = Array.from({ length: Math.min(4, queue.length) }, async () => {
|
|
while (queue.length) {
|
|
const localPath = queue.shift();
|
|
this.changed.delete(localPath);
|
|
this.lastCheckedAt.set(localPath, Date.now());
|
|
try {
|
|
const status = await this.git.status(localPath);
|
|
await this.recordStatus(localPath, status, 'working-tree-changed');
|
|
} catch (error) {
|
|
const next = `error:${error.message}`;
|
|
const previous = this.fingerprints.get(localPath);
|
|
this.fingerprints.set(localPath, next);
|
|
if (previous && previous !== next) {
|
|
await this.diagnostics?.warning('repository-monitor.unavailable', { localPath, message: error.message });
|
|
this.onChange?.({ localPath, error: error.message, reason: 'repository-unavailable' });
|
|
}
|
|
}
|
|
}
|
|
});
|
|
await Promise.all(workers);
|
|
} finally {
|
|
this.running = false;
|
|
// Activity that arrived while the check was running keeps its flag set, so
|
|
// it must not wait for the safety interval.
|
|
if (this.active && this.changed.size) this.scheduleWatchTick();
|
|
}
|
|
}
|
|
}
|
|
|
|
module.exports = { RepositoryMonitor, SAFETY_CHECK_INTERVAL_MS, WATCH_DEBOUNCE_MS, MIN_WATCH_CHECK_INTERVAL_MS };
|