path-watcher.js

const fs = require('fs')
const path = require('path')

const {Emitter, Disposable, CompositeDisposable} = require('event-kit')
const nsfw = require('@atom/nsfw')
const watcher = require('@atom/watcher')
const {NativeWatcherRegistry} = require('./native-watcher-registry')

// Private: Associate native watcher action flags with descriptive String equivalents.
const ACTION_MAP = new Map([
  [nsfw.actions.MODIFIED, 'modified'],
  [nsfw.actions.CREATED, 'created'],
  [nsfw.actions.DELETED, 'deleted'],
  [nsfw.actions.RENAMED, 'renamed']
])

// Private: Possible states of a {NativeWatcher}.
const WATCHER_STATE = {
  STOPPED: Symbol('stopped'),
  STARTING: Symbol('starting'),
  RUNNING: Symbol('running'),
  STOPPING: Symbol('stopping')
}

// Private: Interface with and normalize events from a filesystem watcher implementation.
class NativeWatcher {

  // Private: Initialize a native watcher on a path.
  //
  // Events will not be produced until {start()} is called.
  constructor (normalizedPath) {
    this.normalizedPath = normalizedPath
    this.emitter = new Emitter()
    this.subs = new CompositeDisposable()

    this.state = WATCHER_STATE.STOPPED

    this.onEvents = this.onEvents.bind(this)
    this.onError = this.onError.bind(this)
  }

  // Private: Begin watching for filesystem events.
  //
  // Has no effect if the watcher has already been started.
  async start () {
    if (this.state !== WATCHER_STATE.STOPPED) {
      return
    }
    this.state = WATCHER_STATE.STARTING

    await this.doStart()

    this.state = WATCHER_STATE.RUNNING
    this.emitter.emit('did-start')
  }

  doStart () {
    return Promise.reject('doStart() not overridden')
  }

  // Private: Return true if the underlying watcher is actively listening for filesystem events.
  isRunning () {
    return this.state === WATCHER_STATE.RUNNING
  }

  // Private: Register a callback to be invoked when the filesystem watcher has been initialized.
  //
  // Returns: A {Disposable} to revoke the subscription.
  onDidStart (callback) {
    return this.emitter.on('did-start', callback)
  }

  // Private: Register a callback to be invoked with normalized filesystem events as they arrive. Starts the watcher
  // automatically if it is not already running. The watcher will be stopped automatically when all subscribers
  // dispose their subscriptions.
  //
  // Returns: A {Disposable} to revoke the subscription.
  onDidChange (callback) {
    this.start()

    const sub = this.emitter.on('did-change', callback)
    return new Disposable(() => {
      sub.dispose()
      if (this.emitter.listenerCountForEventName('did-change') === 0) {
        this.stop()
      }
    })
  }

  // Private: Register a callback to be invoked when a {Watcher} should attach to a different {NativeWatcher}.
  //
  // Returns: A {Disposable} to revoke the subscription.
  onShouldDetach (callback) {
    return this.emitter.on('should-detach', callback)
  }

  // Private: Register a callback to be invoked when a {NativeWatcher} is about to be stopped.
  //
  // Returns: A {Disposable} to revoke the subscription.
  onWillStop (callback) {
    return this.emitter.on('will-stop', callback)
  }

  // Private: Register a callback to be invoked when the filesystem watcher has been stopped.
  //
  // Returns: A {Disposable} to revoke the subscription.
  onDidStop (callback) {
    return this.emitter.on('did-stop', callback)
  }

  // Private: Register a callback to be invoked with any errors reported from the watcher.
  //
  // Returns: A {Disposable} to revoke the subscription.
  onDidError (callback) {
    return this.emitter.on('did-error', callback)
  }

  // Private: Broadcast an `onShouldDetach` event to prompt any {Watcher} instances bound here to attach to a new
  // {NativeWatcher} instead.
  //
  // * `replacement` the new {NativeWatcher} instance that a live {Watcher} instance should reattach to instead.
  // * `watchedPath` absolute path watched by the new {NativeWatcher}.
  reattachTo (replacement, watchedPath, options) {
    this.emitter.emit('should-detach', {replacement, watchedPath, options})
  }

  // Private: Stop the native watcher and release any operating system resources associated with it.
  //
  // Has no effect if the watcher is not running.
  async stop () {
    if (this.state !== WATCHER_STATE.RUNNING) {
      return
    }
    this.state = WATCHER_STATE.STOPPING
    this.emitter.emit('will-stop')

    await this.doStop()

    this.state = WATCHER_STATE.STOPPED

    this.emitter.emit('did-stop')
  }

  doStop () {
    return Promise.resolve()
  }

  // Private: Detach any event subscribers.
  dispose () {
    this.emitter.dispose()
  }

  // Private: Callback function invoked by the native watcher when a debounced group of filesystem events arrive.
  // Normalize and re-broadcast them to any subscribers.
  //
  // * `events` An Array of filesystem events.
  onEvents (events) {
    this.emitter.emit('did-change', events)
  }

  // Private: Callback function invoked by the native watcher when an error occurs.
  //
  // * `err` The native filesystem error.
  onError (err) {
    this.emitter.emit('did-error', err)
  }
}

// Private: Emulate a "filesystem watcher" by subscribing to Atom events like buffers being saved. This will miss
// any changes made to files outside of Atom, but it also has no overhead.
class AtomNativeWatcher extends NativeWatcher {
  async doStart () {
    const getRealPath = givenPath => {
      if (!givenPath) {
        return Promise.resolve(null)
      }

      return new Promise(resolve => {
        fs.realpath(givenPath, (err, resolvedPath) => {
          err ? resolve(null) : resolve(resolvedPath)
        })
      })
    }

    this.subs.add(atom.workspace.observeTextEditors(async editor => {
      let realPath = await getRealPath(editor.getPath())
      if (!realPath || !realPath.startsWith(this.normalizedPath)) {
        return
      }

      const announce = (action, oldPath) => {
        const payload = {action, path: realPath}
        if (oldPath) payload.oldPath = oldPath
        this.onEvents([payload])
      }

      const buffer = editor.getBuffer()

      this.subs.add(buffer.onDidConflict(() => announce('modified')))
      this.subs.add(buffer.onDidReload(() => announce('modified')))
      this.subs.add(buffer.onDidSave(event => {
        if (event.path === realPath) {
          announce('modified')
        } else {
          const oldPath = realPath
          realPath = event.path
          announce('renamed', oldPath)
        }
      }))

      this.subs.add(buffer.onDidDelete(() => announce('deleted')))

      this.subs.add(buffer.onDidChangePath(newPath => {
        if (newPath !== this.normalizedPath) {
          const oldPath = this.normalizedPath
          this.normalizedPath = newPath
          announce('renamed', oldPath)
        }
      }))
    }))

    // Giant-ass brittle hack to hook files (and eventually directories) created from the TreeView.
    const treeViewPackage = await atom.packages.getLoadedPackage('tree-view')
    if (!treeViewPackage) return
    await treeViewPackage.activationPromise
    const treeViewModule = treeViewPackage.mainModule
    if (!treeViewModule) return
    const treeView = treeViewModule.getTreeViewInstance()

    const isOpenInEditor = async eventPath => {
      const openPaths = await Promise.all(
        atom.workspace.getTextEditors().map(editor => getRealPath(editor.getPath()))
      )
      return openPaths.includes(eventPath)
    }

    this.subs.add(treeView.onFileCreated(async event => {
      const realPath = await getRealPath(event.path)
      if (!realPath) return

      this.onEvents([{action: 'added', path: realPath}])
    }))

    this.subs.add(treeView.onEntryDeleted(async event => {
      const realPath = await getRealPath(event.path)
      if (!realPath || await isOpenInEditor(realPath)) return

      this.onEvents([{action: 'deleted', path: realPath}])
    }))

    this.subs.add(treeView.onEntryMoved(async event => {
      const [realNewPath, realOldPath] = await Promise.all([
        getRealPath(event.newPath),
        getRealPath(event.initialPath)
      ])
      if (!realNewPath || !realOldPath || await isOpenInEditor(realNewPath) || await isOpenInEditor(realOldPath)) return

      this.onEvents([{action: 'renamed', path: realNewPath, oldPath: realOldPath}])
    }))
  }
}

// Private: Implement a native watcher by translating events from an NSFW watcher.
class NSFWNativeWatcher extends NativeWatcher {
  async doStart (rootPath, eventCallback, errorCallback) {
    const handler = events => {
      this.onEvents(events.map(event => {
        const action = ACTION_MAP.get(event.action) || `unexpected (${event.action})`
        const payload = {action}

        if (event.file) {
          payload.path = path.join(event.directory, event.file)
        } else {
          payload.oldPath = path.join(event.directory, event.oldFile)
          payload.path = path.join(event.directory, event.newFile)
        }

        return payload
      }))
    }

    this.watcher = await nsfw(
      this.normalizedPath,
      handler,
      {debounceMS: 100, errorCallback: this.onError}
    )

    await this.watcher.start()
  }

  doStop () {
    return this.watcher.stop()
  }
}

// Extended: Manage a subscription to filesystem events that occur beneath a root directory. Construct these by
// calling `watchPath`. To watch for events within active project directories, use {Project::onDidChangeFiles}
// instead.
//
// Multiple PathWatchers may be backed by a single native watcher to conserve operation system resources.
//
// Call {::dispose} to stop receiving events and, if possible, release underlying resources. A PathWatcher may be
// added to a {CompositeDisposable} to manage its lifetime along with other {Disposable} resources like event
// subscriptions.
//
// ```js
// const {watchPath} = require('atom')
//
// const disposable = await watchPath('/var/log', {}, events => {
//   console.log(`Received batch of ${events.length} events.`)
//   for (const event of events) {
//     // "created", "modified", "deleted", "renamed"
//     console.log(`Event action: ${event.action}`)
//
//     // absolute path to the filesystem entry that was touched
//     console.log(`Event path: ${event.path}`)
//
//     if (event.action === 'renamed') {
//       console.log(`.. renamed from: ${event.oldPath}`)
//     }
//   }
// })
//
//  // Immediately stop receiving filesystem events. If this is the last
//  // watcher, asynchronously release any OS resources required to
//  // subscribe to these events.
//  disposable.dispose()
// ```
//
// `watchPath` accepts the following arguments:
//
// `rootPath` {String} specifies the absolute path to the root of the filesystem content to watch.
//
// `options` Control the watcher's behavior. Currently a placeholder.
//
// `eventCallback` {Function} to be called each time a batch of filesystem events is observed. Each event object has
// the keys: `action`, a {String} describing the filesystem action that occurred, one of `"created"`, `"modified"`,
// `"deleted"`, or `"renamed"`; `path`, a {String} containing the absolute path to the filesystem entry that was acted
// upon; for rename events only, `oldPath`, a {String} containing the filesystem entry's former absolute path.
class PathWatcher {

  // Private: Instantiate a new PathWatcher. Call {watchPath} instead.
  //
  // * `nativeWatcherRegistry` {NativeWatcherRegistry} used to find and consolidate redundant watchers.
  // * `watchedPath` {String} containing the absolute path to the root of the watched filesystem tree.
  // * `options` See {watchPath} for options.
  //
  constructor (nativeWatcherRegistry, watchedPath, options) {
    this.watchedPath = watchedPath
    this.nativeWatcherRegistry = nativeWatcherRegistry

    this.normalizedPath = null
    this.native = null
    this.changeCallbacks = new Map()

    this.attachedPromise = new Promise(resolve => {
      this.resolveAttachedPromise = resolve
    })

    this.startPromise = new Promise((resolve, reject) => {
      this.resolveStartPromise = resolve
      this.rejectStartPromise = reject
    })

    this.normalizedPathPromise = new Promise((resolve, reject) => {
      fs.realpath(watchedPath, (err, real) => {
        if (err) {
          reject(err)
          return
        }

        this.normalizedPath = real
        resolve(real)
      })
    })
    this.normalizedPathPromise.catch(err => this.rejectStartPromise(err))

    this.emitter = new Emitter()
    this.subs = new CompositeDisposable()
  }

  // Private: Return a {Promise} that will resolve with the normalized root path.
  getNormalizedPathPromise () {
    return this.normalizedPathPromise
  }

  // Private: Return a {Promise} that will resolve the first time that this watcher is attached to a native watcher.
  getAttachedPromise () {
    return this.attachedPromise
  }

  // Extended: Return a {Promise} that will resolve when the underlying native watcher is ready to begin sending events.
  // When testing filesystem watchers, it's important to await this promise before making filesystem changes that you
  // intend to assert about because there will be a delay between the instantiation of the watcher and the activation
  // of the underlying OS resources that feed its events.
  //
  // PathWatchers acquired through `watchPath` are already started.
  //
  // ```js
  // const {watchPath} = require('atom')
  // const ROOT = path.join(__dirname, 'fixtures')
  // const FILE = path.join(ROOT, 'filename.txt')
  //
  // describe('something', function () {
  //   it("doesn't miss events", async function () {
  //     const watcher = watchPath(ROOT, {}, events => {})
  //     await watcher.getStartPromise()
  //     fs.writeFile(FILE, 'contents\n', err => {
  //       // The watcher is listening and the event should be
  //       // received asynchronously
  //     }
  //   })
  // })
  // ```
  getStartPromise () {
    return this.startPromise
  }

  // Private: Attach another {Function} to be called with each batch of filesystem events. See {watchPath} for the
  // spec of the callback's argument.
  //
  // * `callback` {Function} to be called with each batch of filesystem events.
  //
  // Returns a {Disposable} that will stop the underlying watcher when all callbacks mapped to it have been disposed.
  onDidChange (callback) {
    if (this.native) {
      const sub = this.native.onDidChange(events => this.onNativeEvents(events, callback))
      this.changeCallbacks.set(callback, sub)

      this.native.start()
    } else {
      // Attach to a new native listener and retry
      this.nativeWatcherRegistry.attach(this).then(() => {
        this.onDidChange(callback)
      })
    }

    return new Disposable(() => {
      const sub = this.changeCallbacks.get(callback)
      this.changeCallbacks.delete(callback)
      sub.dispose()
    })
  }

  // Extended: Invoke a {Function} when any errors related to this watcher are reported.
  //
  // * `callback` {Function} to be called when an error occurs.
  //   * `err` An {Error} describing the failure condition.
  //
  // Returns a {Disposable}.
  onDidError (callback) {
    return this.emitter.on('did-error', callback)
  }

  // Private: Wire this watcher to an operating system-level native watcher implementation.
  attachToNative (native) {
    this.subs.dispose()
    this.native = native

    if (native.isRunning()) {
      this.resolveStartPromise()
    } else {
      this.subs.add(native.onDidStart(() => {
        this.resolveStartPromise()
      }))
    }

    // Transfer any native event subscriptions to the new NativeWatcher.
    for (const [callback, formerSub] of this.changeCallbacks) {
      const newSub = native.onDidChange(events => this.onNativeEvents(events, callback))
      this.changeCallbacks.set(callback, newSub)
      formerSub.dispose()
    }

    this.subs.add(native.onDidError(err => {
      this.emitter.emit('did-error', err)
    }))

    this.subs.add(native.onShouldDetach(({replacement, watchedPath}) => {
      if (this.native === native && replacement !== native && this.normalizedPath.startsWith(watchedPath)) {
        this.attachToNative(replacement)
      }
    }))

    this.subs.add(native.onWillStop(() => {
      if (this.native === native) {
        this.subs.dispose()
        this.native = null
      }
    }))

    this.resolveAttachedPromise()
  }

  // Private: Invoked when the attached native watcher creates a batch of native filesystem events. The native watcher's
  // events may include events for paths above this watcher's root path, so filter them to only include the relevant
  // ones, then re-broadcast them to our subscribers.
  onNativeEvents (events, callback) {
    const isWatchedPath = eventPath => eventPath.startsWith(this.normalizedPath)

    const filtered = []
    for (let i = 0; i < events.length; i++) {
      const event = events[i]

      if (event.action === 'renamed') {
        const srcWatched = isWatchedPath(event.oldPath)
        const destWatched = isWatchedPath(event.path)

        if (srcWatched && destWatched) {
          filtered.push(event)
        } else if (srcWatched && !destWatched) {
          filtered.push({action: 'deleted', kind: event.kind, path: event.oldPath})
        } else if (!srcWatched && destWatched) {
          filtered.push({action: 'created', kind: event.kind, path: event.path})
        }
      } else {
        if (isWatchedPath(event.path)) {
          filtered.push(event)
        }
      }
    }

    if (filtered.length > 0) {
      callback(filtered)
    }
  }

  // Extended: Unsubscribe all subscribers from filesystem events. Native resources will be released asynchronously,
  // but this watcher will stop broadcasting events immediately.
  dispose () {
    for (const sub of this.changeCallbacks.values()) {
      sub.dispose()
    }

    this.emitter.dispose()
    this.subs.dispose()
  }
}

// Private: Globally tracked state used to de-duplicate related [PathWatchers]{PathWatcher} backed by emulated Atom
// events or NSFW.
class PathWatcherManager {

  // Private: Access the currently active manager instance, creating one if necessary.
  static active () {
    if (!this.activeManager) {
      this.activeManager = new PathWatcherManager(atom.config.get('core.fileSystemWatcher'))
      this.sub = atom.config.onDidChange('core.fileSystemWatcher', ({newValue}) => { this.transitionTo(newValue) })
    }
    return this.activeManager
  }

  // Private: Replace the active {PathWatcherManager} with a new one that creates [NativeWatchers]{NativeWatcher}
  // based on the value of `setting`.
  static async transitionTo (setting) {
    const current = this.active()

    if (this.transitionPromise) {
      await this.transitionPromise
    }

    if (current.setting === setting) {
      return
    }
    current.isShuttingDown = true

    let resolveTransitionPromise = () => {}
    this.transitionPromise = new Promise(resolve => {
      resolveTransitionPromise = resolve
    })

    const replacement = new PathWatcherManager(setting)
    this.activeManager = replacement

    await Promise.all(
      Array.from(current.live, async ([root, native]) => {
        const w = await replacement.createWatcher(root, {}, () => {})
        native.reattachTo(w.native, root, w.native.options || {})
      })
    )

    current.stopAllWatchers()

    resolveTransitionPromise()
    this.transitionPromise = null
  }

  // Private: Initialize global {PathWatcher} state.
  constructor (setting) {
    this.setting = setting
    this.live = new Map()

    const initLocal = NativeConstructor => {
      this.nativeRegistry = new NativeWatcherRegistry(
        normalizedPath => {
          const nativeWatcher = new NativeConstructor(normalizedPath)

          this.live.set(normalizedPath, nativeWatcher)
          const sub = nativeWatcher.onWillStop(() => {
            this.live.delete(normalizedPath)
            sub.dispose()
          })

          return nativeWatcher
        }
      )
    }

    if (setting === 'atom') {
      initLocal(AtomNativeWatcher)
    } else if (setting === 'experimental') {
      //
    } else if (setting === 'poll') {
      //
    } else {
      initLocal(NSFWNativeWatcher)
    }

    this.isShuttingDown = false
  }

  useExperimentalWatcher () {
    return this.setting === 'experimental' || this.setting === 'poll'
  }

  // Private: Create a {PathWatcher} tied to this global state. See {watchPath} for detailed arguments.
  async createWatcher (rootPath, options, eventCallback) {
    if (this.isShuttingDown) {
      await this.constructor.transitionPromise
      return PathWatcherManager.active().createWatcher(rootPath, options, eventCallback)
    }

    if (this.useExperimentalWatcher()) {
      if (this.setting === 'poll') {
        options.poll = true
      }

      const w = await watcher.watchPath(rootPath, options, eventCallback)
      this.live.set(rootPath, w.native)
      return w
    }

    const w = new PathWatcher(this.nativeRegistry, rootPath, options)
    w.onDidChange(eventCallback)
    await w.getStartPromise()
    return w
  }

  // Private: Directly access the {NativeWatcherRegistry}.
  getRegistry () {
    if (this.useExperimentalWatcher()) {
      return watcher.getRegistry()
    }

    return this.nativeRegistry
  }

  // Private: Sample watcher usage statistics. Only available for experimental watchers.
  status () {
    if (this.useExperimentalWatcher()) {
      return watcher.status()
    }

    return {}
  }

  // Private: Return a {String} depicting the currently active native watchers.
  print () {
    if (this.useExperimentalWatcher()) {
      return watcher.printWatchers()
    }

    return this.nativeRegistry.print()
  }

  // Private: Stop all living watchers.
  //
  // Returns a {Promise} that resolves when all native watcher resources are disposed.
  stopAllWatchers () {
    if (this.useExperimentalWatcher()) {
      return watcher.stopAllWatchers()
    }

    return Promise.all(
      Array.from(this.live, ([, w]) => w.stop())
    )
  }
}

// Extended: Invoke a callback with each filesystem event that occurs beneath a specified path. If you only need to
// watch events within the project's root paths, use {Project::onDidChangeFiles} instead.
//
// watchPath handles the efficient re-use of operating system resources across living watchers. Watching the same path
// more than once, or the child of a watched path, will re-use the existing native watcher.
//
// * `rootPath` {String} specifies the absolute path to the root of the filesystem content to watch.
// * `options` Control the watcher's behavior.
// * `eventCallback` {Function} or other callable to be called each time a batch of filesystem events is observed.
//    * `events` {Array} of objects that describe the events that have occurred.
//      * `action` {String} describing the filesystem action that occurred. One of `"created"`, `"modified"`,
//        `"deleted"`, or `"renamed"`.
//      * `path` {String} containing the absolute path to the filesystem entry that was acted upon.
//      * `oldPath` For rename events, {String} containing the filesystem entry's former absolute path.
//
// Returns a {Promise} that will resolve to a {PathWatcher} once it has started. Note that every {PathWatcher}
// is a {Disposable}, so they can be managed by a {CompositeDisposable} if desired.
//
// ```js
// const {watchPath} = require('atom')
//
// const disposable = await watchPath('/var/log', {}, events => {
//   console.log(`Received batch of ${events.length} events.`)
//   for (const event of events) {
//     // "created", "modified", "deleted", "renamed"
//     console.log(`Event action: ${event.action}`)
//     // absolute path to the filesystem entry that was touched
//     console.log(`Event path: ${event.path}`)
//     if (event.action === 'renamed') {
//       console.log(`.. renamed from: ${event.oldPath}`)
//     }
//   }
// })
//
//  // Immediately stop receiving filesystem events. If this is the last watcher, asynchronously release any OS
//  // resources required to subscribe to these events.
//  disposable.dispose()
// ```
//
function watchPath (rootPath, options, eventCallback) {
  return PathWatcherManager.active().createWatcher(rootPath, options, eventCallback)
}

// Private: Return a Promise that resolves when all {NativeWatcher} instances associated with a FileSystemManager
// have stopped listening. This is useful for `afterEach()` blocks in unit tests.
function stopAllWatchers () {
  return PathWatcherManager.active().stopAllWatchers()
}

// Private: Show the currently active native watchers in a formatted {String}.
watchPath.printWatchers = function () {
  return PathWatcherManager.active().print()
}

// Private: Access the active {NativeWatcherRegistry}.
watchPath.getRegistry = function () {
  return PathWatcherManager.active().getRegistry()
}

// Private: Sample usage statistics for the active watcher.
watchPath.status = function () {
  return PathWatcherManager.active().status()
}

// Private: Configure @atom/watcher ("experimental") directly.
watchPath.configure = function (...args) {
  return watcher.configure(...args)
}

module.exports = {watchPath, stopAllWatchers}