path: root/stdlib/source/library/lux/world/file/watch.lux

(.require
 [library
  [lux (.except all and)
   ["[0]" ffi (.only import)]
   [abstract
    ["[0]" monad (.only do)]]
   [control
    ["[0]" io (.only IO)]
    ["[0]" maybe]
    ["[0]" try (.only Try)]
    ["[0]" exception (.only Exception)]
    [concurrency
     ["[0]" async (.only Async)]
     ["[0]" stm (.only STM Var)]]
    [function
     [predicate (.only Predicate)]]]
   [data
    ["[0]" product]
    ["[0]" text (.only)
     ["%" \\format (.only format)]]
    [collection
     ["[0]" dictionary (.only Dictionary)]
     ["[0]" list (.use "[1]#[0]" functor monoid mix)]
     ["[0]" set]
     ["[0]" array]]]
   [math
    [number
     ["n" nat]]]
   [meta
    ["@" target]
    [type
     ["[0]" nominal (.only representation abstraction)]]]
   [world
    [time
     ["[0]" instant (.only Instant) (.use "[1]#[0]" equivalence)]]]]]
 ["[0]" //])

(nominal.def .public Concern
  (Record
   [#creation Bit
    #modification Bit
    #deletion Bit])

  (def none
    Concern
    (abstraction
     [#creation false
      #modification false
      #deletion false]))

  (with_template [<concern> <predicate> <event> <create> <modify> <delete>]
    [(def .public <concern>
       Concern
       (abstraction
        [#creation <create>
         #modification <modify>
         #deletion <delete>]))

     (def .public <predicate>
       (Predicate Concern)
       (|>> representation (the <event>)))]

    [creation creation? #creation
     true false false]
    [modification modification? #modification
     false true false]
    [deletion deletion? #deletion
     false false true]
    )

  (def .public (and left right)
    (-> Concern Concern Concern)
    (abstraction
     [#creation (or (..creation? left) (..creation? right))
      #modification (or (..modification? left) (..modification? right))
      #deletion (or (..deletion? left) (..deletion? right))]))

  (def .public all
    Concern
    (.all ..and
          ..creation
          ..modification
          ..deletion
          ))
  )

(type .public (Watcher !)
  (Interface
   (is (-> Concern //.Path (! (Try Any)))
       start)
   (is (-> //.Path (! (Try Concern)))
       concern)
   (is (-> //.Path (! (Try Concern)))
       stop)
   (is (-> [] (! (Try (List [Concern //.Path]))))
       poll)))

(with_template [<name>]
  [(exception.def .public (<name> path)
     (Exception //.Path)
     (exception.report
      (list ["Path" (%.text path)])))]

  [not_being_watched]
  [cannot_poll_a_non_existent_directory]
  )

(type File_Tracker
  (Dictionary //.Path Instant))

(type Directory_Tracker
  (Dictionary //.Path [Concern File_Tracker]))

(def (update_watch! new_concern path tracker)
  (-> Concern //.Path (Var Directory_Tracker) (STM Bit))
  (do [! stm.monad]
    [@tracker (stm.read tracker)]
    (when (dictionary.value path @tracker)
      {.#Some [old_concern last_modified]}
      (do !
        [_ (stm.update (dictionary.has path [new_concern last_modified]) tracker)]
        (in true))
      
      {.#None}
      (in false))))

(def (file_tracker fs directory)
  (-> (//.System Async) //.Path (Async (Try File_Tracker)))
  (do [! (try.with async.monad)]
    [files (of fs directory_files directory)]
    (monad.mix !
               (function (_ file tracker)
                 (do !
                   [last_modified (of fs last_modified file)]
                   (in (dictionary.has file last_modified tracker))))
               (is File_Tracker
                   (dictionary.empty text.hash))
               files)))

(def (available_files fs directory)
  (-> (//.System Async) //.Path (Async (Try (List [//.Path Instant]))))
  (do [! (try.with async.monad)]
    [files (of fs directory_files directory)]
    (monad.each ! (function (_ file)
                    (|> file
                        (of fs last_modified)
                        (of ! each (|>> [file]))))
                files)))

(def (available_directory_changes fs [directory [the_concern file_tracker]])
  (-> (//.System Async) [//.Path [Concern File_Tracker]]
      (Async (Try [[//.Path [Concern File_Tracker]]
                   [(List [//.Path Instant])
                    (List [//.Path Instant Instant])
                    (List //.Path)]])))
  (do [! (try.with async.monad)]
    [current_files (..available_files fs directory)
     .let [creations (if (..creation? the_concern)
                       (list.only (|>> product.left (dictionary.key? file_tracker) not)
                                  current_files)
                       (list))
           available (|> current_files
                         (list#each product.left)
                         (set.of_list text.hash))
           deletions (if (..deletion? the_concern)
                       (|> (dictionary.entries file_tracker)
                           (list#each product.left)
                           (list.only (|>> (set.member? available) not)))
                       (list))
           modifications (list.all (function (_ [path current_modification])
                                     (do maybe.monad
                                       [previous_modification (dictionary.value path file_tracker)]
                                       (in [path previous_modification current_modification])))
                                   current_files)]]
    (in [[directory
          [the_concern
           (let [with_deletions (list#mix dictionary.lacks file_tracker deletions)
                 with_creations (list#mix (function (_ [path last_modified] tracker)
                                            (dictionary.has path last_modified tracker))
                                          with_deletions
                                          creations)
                 with_modifications (list#mix (function (_ [path previous_modification current_modification] tracker)
                                                (dictionary.has path current_modification tracker))
                                              with_creations
                                              modifications)]
             with_modifications)]]
         [creations
          modifications
          deletions]])))

(def .public (polling fs)
  (-> (//.System Async) (Watcher Async))
  (let [tracker (is (Var Directory_Tracker)
                    (stm.var (dictionary.empty text.hash)))]
    (implementation
     (def (start new_concern path)
       (do [! async.monad]
         [exists? (of fs directory? path)]
         (if exists?
           (do !
             [updated? (stm.commit! (..update_watch! new_concern path tracker))]
             (if updated?
               (in {try.#Success []})
               (do (try.with !)
                 [file_tracker (..file_tracker fs path)]
                 (do !
                   [_ (stm.commit! (stm.update (dictionary.has path [new_concern file_tracker]) tracker))]
                   (in {try.#Success []})))))
           (in (exception.except ..cannot_poll_a_non_existent_directory [path])))))
     (def (concern path)
       (stm.commit!
        (do stm.monad
          [@tracker (stm.read tracker)]
          (in (when (dictionary.value path @tracker)
                {.#Some [it file_tracker]}
                {try.#Success it}

                {.#None}
                (exception.except ..not_being_watched [path]))))))
     (def (stop path)
       (stm.commit!
        (do [! stm.monad]
          [@tracker (stm.read tracker)]
          (when (dictionary.value path @tracker)
            {.#Some [the_concern file_tracker]}
            (do !
              [_ (stm.update (dictionary.lacks path) tracker)]
              (in {try.#Success the_concern}))

            {.#None}
            (in (exception.except ..not_being_watched [path]))))))
     (def (poll _)
       (do async.monad
         [@tracker (stm.commit! (stm.read tracker))]
         (do [! (try.with async.monad)]
           [changes (|> @tracker
                        dictionary.entries
                        (monad.each ! (..available_directory_changes fs)))
            _ (do async.monad
                [_ (stm.commit! (stm.write (|> changes
                                               (list#each product.left)
                                               (dictionary.of_list text.hash))
                                           tracker))]
                (in {try.#Success []}))
            .let [[creations modifications deletions]
                  (list#mix (function (_ [_ [creations modifications deletions]]
                                         [all_creations all_modifications all_deletions])
                              [(list#composite creations all_creations)
                               (list#composite modifications all_modifications)
                               (list#composite deletions all_deletions)])
                            [(list) (list) (list)]
                            changes)]]
           (in (.all list#composite
                     (list#each (|>> product.left [..creation]) creations)
                     (|> modifications
                         (list.only (function (_ [path previous_modification current_modification])
                                      (not (instant#= previous_modification current_modification))))
                         (list#each (|>> product.left [..modification])))
                     (list#each (|>> [..deletion]) deletions)
                     )))))
     )))

(def .public (mock separator)
  (-> Text [(//.System Async) (Watcher Async)])
  (let [fs (//.mock separator)]
    [fs
     (..polling fs)]))

(with_expansions [<jvm> (these (import java/lang/Object
                                 "[1]::[0]")

                               (import java/lang/String
                                 "[1]::[0]")

                               (import (java/util/List a)
                                 "[1]::[0]"
                                 (size [] int)
                                 (get [int] a))

                               (def (default_list list)
                                 (All (_ a) (-> (java/util/List a) (List a)))
                                 (let [size (.nat (ffi.of_int (java/util/List::size list)))]
                                   (loop (again [idx 0
                                                 output {.#End}])
                                     (if (n.< size idx)
                                       (again (++ idx)
                                              {.#Item (java/util/List::get (ffi.as_int (.int idx)) list)
                                                      output})
                                       output))))
                               
                               (import (java/nio/file/WatchEvent$Kind a)
                                 "[1]::[0]")

                               (import (java/nio/file/WatchEvent a)
                                 "[1]::[0]"
                                 (kind [] (java/nio/file/WatchEvent$Kind a)))

                               (import java/nio/file/Watchable
                                 "[1]::[0]")

                               (import java/nio/file/Path
                                 "[1]::[0]"
                                 (register [java/nio/file/WatchService [(java/nio/file/WatchEvent$Kind [? < java/lang/Object])]] "io" "try" java/nio/file/WatchKey)
                                 (toString [] java/lang/String))

                               (import java/nio/file/StandardWatchEventKinds
                                 "[1]::[0]"
                                 ("read_only" "static" ENTRY_CREATE (java/nio/file/WatchEvent$Kind java/nio/file/Path))
                                 ("read_only" "static" ENTRY_MODIFY (java/nio/file/WatchEvent$Kind java/nio/file/Path))
                                 ("read_only" "static" ENTRY_DELETE (java/nio/file/WatchEvent$Kind java/nio/file/Path)))

                               (def (default_event_concern event)
                                 (All (_ a)
                                   (-> (java/nio/file/WatchEvent a) Concern))
                                 (let [kind (as (java/nio/file/WatchEvent$Kind java/nio/file/Path)
                                                (java/nio/file/WatchEvent::kind event))]
                                   (cond (same? (java/nio/file/StandardWatchEventKinds::ENTRY_CREATE)
                                                kind)
                                         ..creation
                                         
                                         (same? (java/nio/file/StandardWatchEventKinds::ENTRY_MODIFY)
                                                kind)
                                         ..modification
                                         
                                         (same? (java/nio/file/StandardWatchEventKinds::ENTRY_DELETE)
                                                kind)
                                         ..deletion
                                         
                                         ... else
                                         ..none
                                         )))

                               (import java/nio/file/WatchKey
                                 "[1]::[0]"
                                 (reset [] "io" boolean)
                                 (cancel [] "io" void)
                                 (watchable [] java/nio/file/Watchable)
                                 (pollEvents [] "io" (java/util/List (java/nio/file/WatchEvent ?))))

                               (def default_key_concern
                                 (-> java/nio/file/WatchKey (IO Concern))
                                 (|>> java/nio/file/WatchKey::pollEvents
                                      (of io.monad each (|>> ..default_list
                                                             (list#each default_event_concern)
                                                             (list#mix ..and ..none)))))

                               (import java/nio/file/WatchService
                                 "[1]::[0]"
                                 (poll [] "io" "try" "?" java/nio/file/WatchKey))

                               (import java/nio/file/FileSystem
                                 "[1]::[0]"
                                 (newWatchService [] "io" "try" java/nio/file/WatchService))

                               (import java/nio/file/FileSystems
                                 "[1]::[0]"
                                 ("static" getDefault [] java/nio/file/FileSystem))

                               (import java/io/File
                                 "[1]::[0]"
                                 (new [java/lang/String])
                                 (toPath [] java/nio/file/Path))

                               (type Watch_Event
                                 (java/nio/file/WatchEvent$Kind java/lang/Object))

                               (def (default_start watch_events watcher path)
                                 (-> (List Watch_Event) java/nio/file/WatchService //.Path (Async (Try java/nio/file/WatchKey)))
                                 (let [watch_events' (list#mix (function (_ [index watch_event] watch_events')
                                                                 (ffi.write! index watch_event watch_events'))
                                                               (ffi.array (java/nio/file/WatchEvent$Kind java/lang/Object)
                                                                          (list.size watch_events))
                                                               (list.enumeration watch_events))]
                                   (async.future
                                    (java/nio/file/Path::register watcher
                                                                  watch_events'
                                                                  (|> path ffi.as_string java/io/File::new java/io/File::toPath)))))

                               (def (default_poll watcher)
                                 (-> java/nio/file/WatchService (IO (Try (List [Concern //.Path]))))
                                 (loop (again [output (is (List [Concern //.Path])
                                                          (list))])
                                   (do (try.with io.monad)
                                     [?key (java/nio/file/WatchService::poll watcher)]
                                     (when ?key
                                       {.#Some key}
                                       (do [! io.monad]
                                         [valid? (java/nio/file/WatchKey::reset key)]
                                         (if (ffi.of_boolean valid?)
                                           (do !
                                             [.let [path (|> key
                                                             java/nio/file/WatchKey::watchable
                                                             (as java/nio/file/Path)
                                                             java/nio/file/Path::toString
                                                             ffi.of_string
                                                             (as //.Path))]
                                              the_concern (..default_key_concern key)]
                                             (again {.#Item [the_concern path]
                                                            output}))
                                           (again output)))
                                       
                                       {.#None}
                                       (in output)))))

                               (def (watch_events concern)
                                 (-> Concern (List Watch_Event))
                                 (.all list#composite
                                       (if (..creation? concern)
                                         (list (as Watch_Event (java/nio/file/StandardWatchEventKinds::ENTRY_CREATE)))
                                         (list))
                                       (if (..modification? concern)
                                         (list (as Watch_Event (java/nio/file/StandardWatchEventKinds::ENTRY_MODIFY)))
                                         (list))
                                       (if (..deletion? concern)
                                         (list (as Watch_Event (java/nio/file/StandardWatchEventKinds::ENTRY_DELETE)))
                                         (list))
                                       ))

                               (def .public default
                                 (IO (Try (Watcher Async)))
                                 (do (try.with io.monad)
                                   [watcher (java/nio/file/FileSystem::newWatchService
                                             (java/nio/file/FileSystems::getDefault))
                                    .let [tracker (stm.var (is (Dictionary //.Path [Concern java/nio/file/WatchKey])
                                                               (dictionary.empty text.hash)))

                                          stop (is (-> //.Path (Async (Try Concern)))
                                                   (function (_ path)
                                                     (do [! async.monad]
                                                       [@tracker (stm.commit! (stm.read tracker))]
                                                       (when (dictionary.value path @tracker)
                                                         {.#Some [the_concern key]}
                                                         (do !
                                                           [_ (async.future
                                                               (java/nio/file/WatchKey::cancel key))
                                                            _ (stm.commit! (stm.update (dictionary.lacks path) tracker))]
                                                           (in {try.#Success the_concern}))

                                                         {.#None}
                                                         (in (exception.except ..not_being_watched [path]))))))]]
                                   (in (is (Watcher Async)
                                           (implementation
                                            (def (start the_concern path)
                                              (do async.monad
                                                [?concern (stop path)]
                                                (do (try.with async.monad)
                                                  [key (..default_start (..watch_events (..and (try.else ..none ?concern)
                                                                                               the_concern))
                                                                        watcher
                                                                        path)]
                                                  (do async.monad
                                                    [_ (stm.commit! (stm.update (dictionary.has path [the_concern key]) tracker))]
                                                    (in {try.#Success []})))))
                                            (def (concern path)
                                              (do async.monad
                                                [@tracker (stm.commit! (stm.read tracker))]
                                                (when (dictionary.value path @tracker)
                                                  {.#Some [it key]}
                                                  (in {try.#Success it})

                                                  {.#None}
                                                  (in (exception.except ..not_being_watched [path])))))
                                            (def stop stop)
                                            (def (poll _)
                                              (async.future (..default_poll watcher)))
                                            )))))
                               )]
  (for @.old (these <jvm>)
       @.jvm (these <jvm>)
       (these)))