import { EventEmitter } from 'events'
import { readFileSync } from 'fs'
import { relative } from 'path'
import { FsWatcher } from './fs-watcher.js'
import { getLang, isSourceFile } from './lang.js'
/**
* LiveStore — holds the current tree + broadcasts deltas.
*
* Built on the push model: file changes come in as fs.watch events, get
* re-parsed, and emit new file-parsed events into a live event stream. All
* connected clients (SSE consumers) receive the delta.
*
* Initial state replay: a newly-connected client receives the events that
* would rebuild the current tree — no full JSON dump needed.
*
* The store does NOT own the tree directly. It holds the graph produced by
* the initial scan and replays events to new subscribers. Incremental updates
* from the watcher amend the graph in place and emit the corresponding events.
*/
export class LiveStore extends EventEmitter {
constructor (graph, opts = {}) {
super()
this.graph = graph
this.rootDir = graph.root
this.resolutionContext = graph.resolutionContext || { ignore: () => false, projectType: graph.projectType }
this.ignore = this.resolutionContext.ignore
this._watcher = null
this._history = [] // append-only log of live events for replay
this._enabledWatcher = opts.watch !== false
}
start () {
if (!this._enabledWatcher) return
this._watcher = new FsWatcher(this.rootDir, this.ignore)
this._watcher.on('change', e => this._onChange(e))
this._watcher.on('remove', e => this._onRemove(e))
// Watcher failures (ENOSPC, EMFILE, EACCES) are recoverable — the snapshot
// is still served, just no live updates. Log and degrade rather than crash.
this._watcher.on('error', err => {
this.watcherError = err
console.warn('[live-store] fs.watch disabled:', err.code || err.message)
})
}
stop () {
if (this._watcher) {
this._watcher.close()
this._watcher = null
}
}
// --- Subscription ---
/**
* Attach a subscriber. Called with every historic event (to rebuild state),
* then with every future live event as it fires.
*
* Returns an unsubscribe function.
*/
subscribe (handler) {
// Replay history (snapshot of the current state as events)
for (const ev of this._history) handler(ev)
this.on('event', handler)
return () => this.off('event', handler)
}
// --- Initial state replay ---
/**
* Seed the history from the initial graph. Converts the already-built tree
* back into the event sequence that would reconstruct it. A new subscriber
* then receives these as their "bootstrap" before live events start.
*/
seedFromGraph () {
const g = this.graph
this._log({
type: 'scan-start',
root: g.root,
projectType: g.projectType,
languages: g.languages,
shallow: g.shallow || false
})
for (const [name, mod] of g.modules) {
this._log({
type: 'module-start',
name,
version: mod.version,
path: mod.path,
absPath: mod.absPath,
entry: mod.entry,
projectType: mod.projectType,
dependencies: mod.dependencies,
devDependencies: mod.devDependencies,
importMaps: mod.importMaps,
testCommand: mod.testCommand,
// mod.type is JS module type ('commonjs'/'module') — avoid the name
// collision with the SSE event type.
moduleType: mod.type
})
for (const [relInPkg, fileInfo] of mod.files) {
this._log({
type: 'file-discovered',
modName: name,
absPath: fileInfo.absPath,
relPath: fileInfo.relPath,
relInPkg,
size: fileInfo.size || 0
})
if (fileInfo.parsed) {
this._log({
type: 'file-parsed',
modName: name,
relInPkg,
relPath: fileInfo.relPath,
parsed: fileInfo.parsed
})
}
for (const imp of fileInfo.localImports || []) {
this._log({
type: 'import-edge',
modName: name,
fromRelInPkg: relInPkg,
specifier: imp.specifier,
resolvedRelPath: imp.resolved,
implicit: !!imp.implicit
})
}
}
this._log({ type: 'module-end', name })
}
this._log({ type: 'scan-end', root: g.root })
}
_log (ev) {
this._history.push(ev)
this.emit('event', ev)
}
// --- Change handlers ---
_onChange (e) {
// Which module does this file belong to? Find by absPath prefix.
const modEntry = this._findModuleForPath(e.absPath)
if (!modEntry) return // file in a dir we don't track (test output, new repo, etc.)
const [modName, mod] = modEntry
const relInPkg = relative(mod.absPath, e.absPath)
// Non-source files get a lightweight change event but no re-parse
if (!isSourceFile(e.absPath)) {
this._log({ type: 'file-changed', modName, absPath: e.absPath, relPath: e.relPath, relInPkg })
return
}
// Read + parse the file; emit file-parsed so subscribers can update their tree
let source
try { source = readFileSync(e.absPath, 'utf-8') } catch { return }
const lang = getLang(e.absPath)
if (!lang) return
let parsed
try { parsed = lang.parse(source, relInPkg) }
catch (err) { parsed = { error: err.message, filePath: relInPkg } }
// Update the in-memory graph for future subscribers / replay
let fileInfo = mod.files.get(relInPkg)
const isNew = !fileInfo
if (isNew) {
fileInfo = {
absPath: e.absPath,
relPath: e.relPath,
relInPkg,
size: source.length,
localImports: [],
parsed: null
}
mod.files.set(relInPkg, fileInfo)
this.graph.files.set(e.relPath, fileInfo)
this._log({
type: 'file-discovered',
modName, absPath: e.absPath, relPath: e.relPath, relInPkg, size: source.length
})
}
fileInfo.parsed = parsed
fileInfo.size = source.length
this._log({ type: 'file-parsed', modName, relInPkg, relPath: e.relPath, parsed })
}
_onRemove (e) {
const modEntry = this._findModuleForPath(e.absPath)
if (!modEntry) return
const [modName, mod] = modEntry
const relInPkg = relative(mod.absPath, e.absPath)
if (mod.files.has(relInPkg)) {
mod.files.delete(relInPkg)
this.graph.files.delete(e.relPath)
this._log({ type: 'file-removed', modName, relPath: e.relPath, relInPkg })
}
}
_findModuleForPath (absPath) {
// Linear scan — fine for now. Modules are typically <100. If this grows
// we can index by prefix.
let bestEntry = null
let bestLen = -1
for (const [name, mod] of this.graph.modules) {
if (absPath === mod.absPath || absPath.startsWith(mod.absPath + '/')) {
if (mod.absPath.length > bestLen) {
bestEntry = [name, mod]
bestLen = mod.absPath.length
}
}
}
return bestEntry
}
}