// Copyright 2025 International Digital Economy Academy
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

///|
using @event_loop {type FileIdentity}

///|
/// A file system event generated by `@fs.Watcher`.
///
/// - `Modify(path)`: the regular file at `path` has been modified
/// - `Create(path)`: a new file or directory is created at `path`
/// - `Remove(path)`: the file or directory at `path` is removed
/// - `Rename(old~, new~)`, the file previously at `old` has been moved to `new`
///
/// All paths are relative to the base directory of the watcher, using `/` as path separator.
pub(all) enum FsEvent {
  Modify(String)
  Create(String)
  Remove(String)
  Rename(old~ : String, new~ : String)
} derive(Eq, Debug)

///|
/// Info about a watched file in the watched tree.
/// There are three stages in each `wait` iteration of the watcher:
///
/// - The collect pass: when the watcher is idle, the backend silently
///   collect file system changes in the background.
///   It may, from time to time, mark modified files and directories as dirty.
///   This marking is implemented through the `modified` and `children_to_scan` fields:
///
///   * `modified`: the file/directory itself is modified.
///     For directory this mean the list of files in the directory has changed
///   * `children_to_scan`: for directory only. Some children in the sub tree is modified
///
/// - The scan pass, implemented through `Watcher::synchronize_tree`.
///   In this pass, we scan those modified directories to collect up-to-date knowledge
///   on the structure of the watched tree.
///   Since many operations in the scan pass are path-based,
///   the scan must be performed from parent to child.
///   The dirty mark in the collect pass is used to avoid unnecessary scanning.
///
///   During the scan pass, new changes may happen concurrently on the file system.
///   So the scan pass is run in parallel with the collect pass.
///   When new events arrive during scanning, it is possible that
///   a previously scanned directory will be marked as dirty again.
///   The scan pass must be capable of handling this case.
///   In addition, the scan pass must be also capable of handling scan failure,
///   in order to handle file system changes that happen concurrently with scanning.
///
///   The scan pass consumes and clears `modified` and `children_to_scan`,
///   and record discovered changes in `changed_entries` and `children_with_events`,
///   which are used by the next report pass:
///
///   * `changed_entries`: for directory only. Record the list of locations in the directory
///     that have changed since the last query. The value store for each location is the
///     ID of the old file on that location, or `None` if the file previously does not exist.
///   * `children_with_events`: for directory only.
///     Record the list of children that contain any event to report.
///
///   The watcher maintain a tree snapshot of the file system through the `children` field,
///   The state of the this snapshot, and the meaning of other related fields
///   such as `children_with_events`, should always be in-sync with
///   the latest progress of the scan pass.
///
/// - The report pass. This pass consumes `changed_entries` and `children_with_events`,
///   and report the list of events for the user, through the return value of `wait()`.
///   Event reporting is not performed during the scan pass,
///   because the scan pass may need to rescan already-scanned directory from time to time,
///   while we want to make the returned event list minimal.
///   The report pass is implemented though `emit_events_for_file`.
///   Since we report events in a path-based manner, the report pass also need to operate
///   from parent to children, that's why we need to maintain `children_with_events`
///   instead of a global list.
priv struct WatchedFile {
  /// A platform dependent "watch-descriptor":
  /// - for `inotify`, this is the inotify watch descriptor of the file
  /// - for `kqueue`, this is just a open handle to the file
  /// - for Windows, this field is meaningless
  wd : @fd_util.Fd
  /// The identity of this file.
  ///
  /// Mount point note: for mount points/symbolic links/reparse points,
  /// `identity` is the file identity of the directory entry in the parent,
  /// not the actual physical identity of the mounted file system/redirected file.
  identity : FileIdentity
  paths : Set[(FileIdentity, String)]
  is_dir : Bool
  /// Whether the file itself has been modified since the last query.
  /// For directory this means the list of files have changed.
  /// The collect pass set this field while the scan pass reset it after scanning.
  /// The status of this list should be based on the latest progress of the scan pass.
  mut modified : Bool

  // the following fields are for directory only

  /// The children of a directory, indexed by their file name.
  /// The status of this map should be based on the latest progress of the scan pass.
  children : Map[String, FileIdentity]
  /// The list of children that still need scanning in the scan pass.
  /// The collect pass add members to this list and the scan pass remove members from it.
  /// The status of this list should be based on the latest progress of the scan pass.
  children_to_scan : Map[String, FileIdentity]
  /// The list of entries whose content have changed since the last query,
  /// as discovered by the scan pass.
  /// For entry name the identity of the old file at this location
  /// (or `None` if the entry did not previously exist) is stored.
  /// The scan pass set this list, while the report pass consume and clear this list
  changed_entries : Map[String, FileIdentity?]
  /// The list of children (recursively) that contain event for the report pass.
  /// The scan pass set this list, while the report pass consume and clear this list
  children_with_events : Map[String, FileIdentity]
}

///|
priv trait WatcherBackend {
  fn close(Self) -> Unit
  async fn wait(Self) -> Unit
  async fn add_file(
    Self,
    StringView,
    is_dir~ : Bool,
    identity~ : FileIdentity,
    context~ : String,
  ) -> @fd_util.Fd
  fn remove_file(Self, @fd_util.Fd, context~ : String) -> Unit raise
}

///|
/// A file system watcher for watching changes in a directory.
struct Watcher {
  mut backend : &WatcherBackend?
  root_id : FileIdentity
  base_path : String
  watched : Map[FileIdentity, WatchedFile]
  pending_remove : Set[FileIdentity]
  ignored_paths : (String) -> Bool
  report_child_event : Bool
  mut synchronize_task : @coroutine.Coroutine?
}

///|
pub fn Watcher::close(self : Watcher) -> Unit {
  if self.backend is Some(backend) {
    self.backend = None
    backend.close()
  }
  if self.synchronize_task is Some(task) {
    task.cancel()
  }
}

///|
#cfg(platform="windows")
fn Watcher::new_backend(
  self : Watcher,
  root : @event_loop.IoHandle,
  root_id~ : FileIdentity,
  debounce_timeout~ : Int,
  max_debounce_delay~ : Int,
  context~ : String,
) -> &WatcherBackend {
  ignore(context)
  WindowsWatcher(self, root, root_id~, debounce_timeout~, max_debounce_delay~)
}

///|
#cfg(not(platform="windows"))
fn Watcher::new_backend(
  self : Watcher,
  root : @event_loop.IoHandle,
  root_id~ : FileIdentity,
  debounce_timeout~ : Int,
  max_debounce_delay~ : Int,
  context~ : String,
) -> &WatcherBackend raise {
  ignore(root_id)
  root.close()
  match @event_loop.platform {
    Linux =>
      InotifyWatcher(self, debounce_timeout~, max_debounce_delay~, context~)
      as &WatcherBackend
    MacOS =>
      KqueueWatcher(self, debounce_timeout~, max_debounce_delay~, context~)
      as &WatcherBackend
    Windows => panic()
  }
}

///|
/// Create a new file system watcher that watches the directory `path` recursively.
/// Currently only recursive directory watching is supported.
///
/// When files are created, removed, modified or renamed inside `path`,
/// the watcher will get notified and report events for the change.
/// New files will be watched automatically, and removed files will be unwatched automatically.
///
/// File system events often come in batch.
/// For example, removing a directory recursively will result in several removal event in a row.
/// By default `Watcher` will perform debouncing internally:
/// when many events are happening in sequence, instead of immediately return on the first event,
/// `Watcher::wait_any` will wait until no event is happening
/// within the last `debounce_timeout` ms (defaults to 20ms),
/// or if `max_debounce_delay` ms (defaults to 200ms) has elapsed since the first event.
///
/// `ignored_paths`, if present, can be used to filter out paths that the user don't want to watch.
/// When a file or directory is going to be watched, its path
/// (relative to root of watched tree, using `/` as path separator) will be supplied to `ignored_paths`.
/// If `ignored_paths` return `true`, the file or directory will be ignored.
/// When a directory is ignored, all files/directories within it are also ignored.
/// So to ignore a single directory,
/// `ignored_paths` only need to handle paths of the files inside that ignored directory.
///
/// When a create/remove event is reported for a directory:
/// - if `report_child_event=true`, respective create/remove events will be emitted
///   for everything inside that directory.
///   This mode is useful if tracking the exact list of files in desirable.
/// - If `report_child_event=false` (the default),
///   only a single event for the directory itself will be emitted,
///   making the watcher less noisy
///
/// If `report_event_on_init=true` (`false` by default),
/// the first `wait` call will return immediately after watcher creation,
/// reporting events describing the initial structure of the watched directory.
/// This is useful for keeping the knowledge of the caller in sync with the watcher.
/// Note that you probably want to set `report_child_event=true` as well in this case.
///
/// Currently, the behavior when `path` itself is renamed or removed is undefined.
pub async fn Watcher::Watcher(
  path : String,
  debounce_timeout? : Int = 20,
  max_debounce_delay? : Int = 200,
  report_child_event? : Bool = false,
  report_event_on_init? : Bool = false,
  ignored_paths? : (String) -> Bool = _ => false,
) -> Watcher {
  let context = "@fs.Watcher()"
  let path = path.trim_end(chars="/")
  let path = if path is "" { "/" } else { path.to_owned() }
  let (root_file, root_id) = @event_loop.open(
    path,
    if @event_loop.platform is Windows {
      3
    } else {
      0
    },
    create=0,
    append=false,
    sync=0,
    mode=0,
    context~,
  )
  guard root_file.kind() is Directory else {
    root_file.close()
    raise @os_error.OSError(@os_error.errno_ENOTDIR, context~)
  }
  let self = {
    backend: None,
    root_id,
    base_path: path,
    watched: Map([]),
    pending_remove: Set([]),
    ignored_paths,
    report_child_event,
    synchronize_task: None,
  }
  try {
    let backend = self.new_backend(
      root_file,
      root_id~,
      debounce_timeout~,
      max_debounce_delay~,
      context~,
    )
    self.backend = Some(backend)
    let root_wd = backend.add_file(
      path,
      is_dir=true,
      identity=root_id,
      context~,
    )
    let root = {
      wd: root_wd,
      identity: root_id,
      paths: Set([]),
      is_dir: true,
      children: Map([]),
      modified: true,
      children_to_scan: Map([]),
      changed_entries: Map([]),
      children_with_events: Map([]),
    }
    self.watched[root_id] = root
    self.synchronize_tree(context~)
    if !report_event_on_init {
      self.emit_events_for_file(root, path="", on_event=(_, _) => (), context~)
    }
    self
  } catch {
    err => {
      self.close()
      raise err
    }
  }
}

///|
fn rel_path(parent : String, name : String) -> String {
  if parent is "" {
    name
  } else {
    "\{parent}/\{name}"
  }
}

///|
/// Callback for handling event.
/// The first parameter is a rename hint,
/// if it is `Some(id)`, the second event parameter must be `Create` or `Remove`,
/// and `id` is the identity of the created/removed file.
/// Note that some `Create`/`Remove` event may not be a potential rename candidate,
/// in this case the first parameter would be `None`.
priv struct EventCallback((FileIdentity?, FsEvent) -> Unit)

///|
fn Watcher::emit_events_for_file(
  self : Watcher,
  file : WatchedFile,
  on_event~ : EventCallback,
  path~ : String,
  context~ : String,
) -> Unit raise {
  guard file.is_dir else {
    // The only possible event for a regular file is `Modify`,
    // so no extra information about event detail is recorded for regular files.
    on_event(None, Modify(path))
  }
  let events : Map[FileIdentity, FsEvent] = Map([])
  for name, prev_id in file.changed_entries {
    let curr_id = file.children.get(name)
    let path = rel_path(path, name)
    if prev_id is Some(prev_id) {
      match events.get(prev_id) {
        _ if curr_id is Some(_) => on_event(None, Remove(path))
        Some(Create(new)) => {
          events.remove(prev_id)
          on_event(None, Rename(old=path, new~))
        }
        Some(_) => {
          self.emit_remove_event_for_children(prev_id, on_event~, path~)
          on_event(Some(prev_id), Remove(path))
        }
        None => events[prev_id] = Remove(path)
      }
    }
    if curr_id is Some(curr_id) {
      match events.get(curr_id) {
        Some(Remove(old)) => {
          events.remove(curr_id)
          on_event(None, Rename(old~, new=path))
        }
        Some(_) => {
          on_event(Some(curr_id), Create(path))
          self.emit_create_event_for_children(curr_id, on_event~, path~)
        }
        None => events[curr_id] = Create(path)
      }
    }
  }
  file.changed_entries.clear()
  for id, event in events {
    if event is Remove(path) {
      self.emit_remove_event_for_children(id, on_event~, path~)
    }
    on_event(Some(id), event)
    if event is Create(path) {
      self.emit_create_event_for_children(id, on_event~, path~)
    }
  }
  for name, id in file.children_with_events {
    self.emit_events_for_file(
      self.watched[id],
      on_event~,
      path=rel_path(path, name),
      context~,
    )
  }
  file.children_with_events.clear()
}

///|
/// Emit create event for the whole tree under `identity`.
/// If the tree is not new, but considered "created" due to failed rename serialization,
/// the event state of the whole tree will be reset by `emit_create_event_for_children`.
fn Watcher::emit_create_event_for_children(
  self : Watcher,
  identity : FileIdentity,
  on_event~ : EventCallback,
  path~ : String,
) -> Unit {
  let file = self.watched[identity]
  guard file.is_dir else { return }
  file.changed_entries.clear()
  file.children_with_events.clear()
  for name, child_id in file.children {
    let path = rel_path(path, name)
    if self.report_child_event {
      on_event(None, Create(path))
    }
    self.emit_create_event_for_children(child_id, on_event~, path~)
  }
}

///|
/// Emit remove event for the whole tree under `identity`.
/// If the sub tree is still present in the watched tree,
/// but considered "removed" due to failed rename serialization,
/// the event state of the whole tree will be reset by `emit_remove_event_for_children`.
fn Watcher::emit_remove_event_for_children(
  self : Watcher,
  identity : FileIdentity,
  on_event~ : EventCallback,
  path~ : String,
) -> Unit {
  let file = self.watched[identity]
  guard file.is_dir else { return }
  file.changed_entries.clear()
  file.children_with_events.clear()
  for name, child_id in file.children {
    let path = rel_path(path, name)
    self.emit_remove_event_for_children(child_id, on_event~, path~)
    if self.report_child_event {
      on_event(None, Remove(path))
    }
  }
}

///|
async fn Watcher::wait_aux(
  self : Watcher,
  on_event~ : (FileIdentity?, FsEvent) -> Unit,
) -> Unit {
  let context = "@fs.Watcher::wait()"
  guard self.backend is Some(backend) else {
    raise Failure::Failure("watcher already closed")
  }
  if self.synchronize_task is Some(task) {
    task.wait()
  }
  let root = self.watched[self.root_id]
  while !root.has_events() {
    backend.wait()
    self.synchronize_tree(context~)
  }
  self.emit_events_for_file(root, path="", on_event~, context~)
  while self.pending_remove.iter().next() is Some(identity) {
    self.pending_remove.remove(identity)
    let file = self.watched[identity]
    guard file.paths.is_empty() else { continue }
    if file.is_dir {
      for child in file.children.keys() {
        self.remove_file_at(file, child)
      }
    }
    self.watched.remove(identity)
    backend.remove_file(file.wd, context~)
  }
}

///|
/// Wait for any change in the watched tree,
/// and return a list of events describing what changed 
/// since the last `wait`/`wait_any` call or watcher creation.
///
/// The returned list describe the net change of the watched tree since the last query,
/// instead of recording a precise list of transactions that happen since the last wait.
/// For example, a create event followed by a removed event on the same path
/// will get fused and result in nothing in the returned event list.
///
/// `wait` performs rename detection based on the physical ID of files. The rule is:
/// - renaming for regular files are detected globally for the whole watched tree
/// - renaming for directory are only detected locally within the same parent directory.
///   Directory renaming across directory are treated as separated remove + create events.
/// - if the destination of a rename already contains something,
///   a remove event will be emitted for the old file on the destination.
/// When the watcher cannot serialize the file system changes into simple rename events,
/// for example two files are atomically swapped,
/// a remove event on the old path + a create event on the new path will be created.
///
/// This method is cancellation safe.
/// When it get cancelled and raises the canellation error,
/// it is guaranteed that no event will be consumed.
pub async fn Watcher::wait(self : Watcher) -> Array[FsEvent] {
  let events = []
  let pending_rename : Map[FileIdentity, FsEvent] = Map([])
  fn on_event(rename_id, event : FsEvent) {
    guard rename_id is Some(id) &&
      !(self.watched.get(id) is Some({ is_dir: true, .. })) else {
      events.push(event)
    }
    match (pending_rename.get(id), event) {
      (None, Create(_) | Remove(_)) => pending_rename[id] = event
      (Some(Remove(old)), Create(new)) | (Some(Create(new)), Remove(old)) => {
        pending_rename.remove(id)
        events.push(Rename(old~, new~))
      }
      _ => events.push(event)
    }
  }
  self.wait_aux(on_event~)
  for _, event in pending_rename {
    events.push(event)
  }
  events
}

///|
/// Wait for any change in the watched tree
/// since the last `wait`/`wait_any` call or watcher creation.
///
/// This method is cancellation safe.
/// When it get cancelled and raises the canellation error,
/// it is guaranteed that no event will be consumed.
pub async fn Watcher::wait_any(self : Watcher) -> Unit {
  self.wait_aux(on_event=(_, _) => ())
}

///|
fn WatchedFile::is_clean(file : WatchedFile) -> Bool {
  !file.modified && file.children_to_scan.is_empty()
}

///|
fn WatchedFile::has_events(file : WatchedFile) -> Bool {
  !(file.is_dir &&
  file.changed_entries.is_empty() &&
  file.children_with_events.is_empty())
}

///|
async fn Watcher::get_file_at(
  self : Watcher,
  identity~ : FileIdentity,
  is_dir~ : Bool,
  path~ : String,
  context~ : String,
) -> WatchedFile {
  if self.watched.get(identity) is Some(existing) {
    return existing
  }

  let full_path = "\{self.base_path}/\{path}"

  guard self.backend is Some(backend) else {
    raise Failure::Failure("watcher already closed")
  }
  let wd = backend.add_file(full_path, is_dir~, identity~, context~)
  let new_file = {
    wd,
    identity,
    paths: Set([]),
    is_dir,
    children: Map([]),
    modified: is_dir, // new directory need rescan
    children_to_scan: Map([]),
    changed_entries: Map([]),
    children_with_events: Map([]),
  }
  self.watched[identity] = new_file
  new_file
}

///|
async fn Watcher::add_file_at(
  self : Watcher,
  dir : WatchedFile,
  name : String,
  is_dir~ : Bool,
  identity~ : FileIdentity,
  path~ : String,
  context~ : String,
) -> Unit {
  if (self.ignored_paths)(path) {
    return
  }
  if dir.children.get(name) is Some(curr_id) {
    if curr_id == identity {
      return
    } else {
      self.remove_file_at(dir, name)
    }
  }
  let location = (dir.identity, name)
  let file = self.get_file_at(path~, is_dir~, identity~, context~)
  dir.children[name] = file.identity
  file.paths.add(location)
  if !file.is_clean() {
    dir.children_to_scan[name] = file.identity
  }
  match dir.changed_entries.get(name) {
    None => dir.changed_entries[name] = None
    Some(Some(prev_id)) if prev_id == identity =>
      dir.changed_entries.remove(name)
    Some(_) => ()
  }
}

///|
fn Watcher::remove_file_at(
  self : Watcher,
  dir : WatchedFile,
  name : String,
) -> Unit {
  let curr_id = dir.children.get(name)
  guard curr_id is Some(curr_id) else { return }
  let path = (dir.identity, name)
  dir.children.remove(name)
  dir.children_to_scan.remove(name)
  dir.children_with_events.remove(name)
  let curr = self.watched[curr_id]
  curr.paths.remove(path)
  if curr.paths.is_empty() {
    self.pending_remove.add(curr_id)
  }
  match dir.changed_entries.get(name) {
    None => dir.changed_entries[name] = Some(curr_id)
    Some(None) => dir.changed_entries.remove(name)
    Some(Some(_)) => ()
  }
}

///|
async fn Watcher::scan_dir(
  self : Watcher,
  dir : WatchedFile,
  path~ : String,
  context~ : String,
) -> Unit {
  let removed_children = dir.children.copy()
  let (dir_fd, dir_id) = @event_loop.open(
    "\{self.base_path}/\{path}",
    0,
    create=0,
    append=false,
    sync=0,
    mode=0,
    context~,
  )
  guard dir_fd.kind() is Directory else {
    dir_fd.close()
    raise @os_error.OSError(@os_error.errno_ENOTDIR, context~)
  }
  let dir_obj = Directory::from_io_handle(dir_fd)
  defer dir_obj.close()
  while dir_obj.next_aux(include_special=false, include_hidden=true, context~)
        is Some(ent) {
    let new_file_id = { ..dir_id, file_id: ent.file_id }
    self.add_file_at(
      dir,
      ent.name,
      is_dir=ent.is_dir,
      identity=new_file_id,
      path=rel_path(path, ent.name),
      context~,
    ) catch {
      @os_error.OSError(_) as err if err.is_ENOENT() => continue
      err => raise err
    }
    removed_children.remove(ent.name)
  }
  for name in removed_children.keys() {
    self.remove_file_at(dir, name)
  }
}

///|
async fn Watcher::synchronize_tree(self : Watcher, context~ : String) -> Unit {
  if self.synchronize_task is Some(task) {
    task.wait()
    return
  }
  let task = @coroutine.spawn <| () => {
    defer {
      self.synchronize_task = None
    }
    let root = self.watched[self.root_id]
    while !root.is_clean() {
      self.synchronize_file(root, path="", context~)
    }
  }
  self.synchronize_task = Some(task)
  task.wait()
}

///|
async fn Watcher::synchronize_file(
  self : Watcher,
  file : WatchedFile,
  path~ : String,
  context~ : String,
) -> Unit {
  if file.modified {
    file.modified = false
    if file.is_dir {
      self.scan_dir(file, path~, context~)
    }
  } else if file.children_to_scan.iter().next() is Some((name, child_id)) {
    let child = self.watched[child_id]
    try
      self.synchronize_file(child, path=rel_path(path, name), context~)
    catch {
      @os_error.OSError(_) as err if err.is_ENOENT() || err.is_ENOTDIR() =>
        self.mark_as_modified(file.identity)
      err => raise err
    } noraise {
      _ => {
        if child.is_clean() {
          file.children_to_scan.remove(name)
        }
        if child.has_events() {
          file.children_with_events[name] = child.identity
        } else {
          file.children_with_events.remove(name)
        }
      }
    }
  }
}

///|
fn Watcher::mark_as_modified(self : Watcher, id : FileIdentity) -> Unit {
  let file = self.watched[id]
  if file.modified {
    return
  }
  file.modified = true
  if !file.children_to_scan.is_empty() || id == self.root_id {
    return
  }
  for path in file.paths {
    let (dir, name) = path
    self.add_dirty_child(dir, name, id)
  }
}

///|
fn Watcher::add_dirty_child(
  self : Watcher,
  dir_id : FileIdentity,
  name : String,
  child_id : FileIdentity,
) -> Unit {
  let dir = self.watched[dir_id]
  let was_clean = dir.is_clean()
  dir.children_to_scan[name] = child_id
  if was_clean && dir_id != self.root_id {
    for path in dir.paths {
      let (parent, name) = path
      self.add_dirty_child(parent, name, dir_id)
    }
  }
}

///|
fn Watcher::overflow(self : Watcher) -> Unit {
  for file in self.watched.values() {
    file.modified = true
    if file.is_dir {
      for name, id in file.children {
        file.children_to_scan[name] = id
      }
    }
  }
}