Fix timing issues waiting for scala-cli linking

Before there were two concurrent proesses
- one was publishing a browser refresh signal when linking was done
- one was watching the file-system for changes and checking updating the map of hashes
The problem is that sometimes the event for changed file comes after linking has finished, creating an issue

Now linking publishes an event in `linkingTopic`.
`fileWatcher` waits reacts to events in `linkingTopic` by listing all files in the directory, calculating the hash and then publishing an event in `refreshTopic` which refreshes the browser

project/src/build.runner.scala

@@ -16,31 +16,33 @@ import cats.effect.IO
 import cats.effect.OutcomeIO
 import cats.effect.ResourceIO
 
+import scala.concurrent.duration.*
+
 sealed trait BuildTool
 class ScalaCli extends BuildTool
 class Mill extends BuildTool
 
 def buildRunner(
     tool: BuildTool,
-    refreshTopic: Topic[IO, String],
+    linkingTopic: Topic[IO, Unit],
     workDir: fs2.io.file.Path,
     outDir: fs2.io.file.Path,
     extraBuildArgs: List[String],
     millModuleName: Option[String]
 )(
     logger: Scribe[IO]
 ): ResourceIO[IO[OutcomeIO[Unit]]] = tool match
-  case scli: ScalaCli => buildRunnerScli(refreshTopic, workDir, outDir, extraBuildArgs)(logger)
+  case scli: ScalaCli => buildRunnerScli(linkingTopic, workDir, outDir, extraBuildArgs)(logger)
   case m: Mill =>
     buildRunnerMill(
-      refreshTopic,
+      linkingTopic,
       workDir,
       millModuleName.getOrElse(throw new Exception("must have a module name when running with mill")),
       extraBuildArgs
     )(logger)
 
 def buildRunnerScli(
-    refreshTopic: Topic[IO, String],
+    linkingTopic: Topic[IO, Unit],
     workDir: fs2.io.file.Path,
     outDir: fs2.io.file.Path,
     extraBuildArgs: List[String]
@@ -79,7 +81,7 @@ def buildRunnerScli(
                   aChunk =>
                     if aChunk.toString.contains("main.js, run it with") then
                       logger.trace("Detected that linking was successful, emitting refresh event") >>
-                        refreshTopic.publish1("refresh")
+                        linkingTopic.publish1(())
                     else
                       logger.trace(s"$aChunk :: Linking unfinished") >>
                         IO.unit
@@ -92,7 +94,7 @@ def buildRunnerScli(
 end buildRunnerScli
 
 def buildRunnerMill(
-    refreshTopic: Topic[IO, String],
+    linkingTopic: Topic[IO, Unit],
     workDir: fs2.io.file.Path,
     moduleName: String,
     extraBuildArgs: List[String]
@@ -105,15 +107,15 @@ def buildRunnerMill(
     .Stream
     .resource(Watcher.default[IO].evalTap(_.watch(watchLinkComplePath.toNioPath)))
     .flatMap {
-      _.events()
+      _.events(100.millis)
         .evalTap {
           (e: Event) =>
             e match
               case Created(path, count) => logger.info("fastLinkJs.json was created")
               case Deleted(path, count) => logger.info("fastLinkJs.json was deleted")
               case Modified(path, count) =>
                 logger.info("fastLinkJs.json was modified - link successful => trigger a refresh") >>
-                  refreshTopic.publish1("refresh")
+                  linkingTopic.publish1(())
               case Overflow(count)                         => logger.info("overflow")
               case NonStandard(event, registeredDirectory) => logger.info("non-standard")
 

project/src/hasher.scala

@@ -1,15 +1,16 @@
-import java.security.MessageDigest
-
 import cats.effect.IO
+import fs2.io.file.*
 
-val md = MessageDigest.getInstance("MD5")
+// TODO: Use last modified time once scala-cli stops
+//       copy pasting the files from a temporary directory
+//       and performs linking in place
+// def fileHash(filePath: Path): IO[String] =
+//   Files[IO].getLastModifiedTime(filePath).map(_.toNanos.toString)
 
-def fielHash(filePath: fs2.io.file.Path): IO[String] =
-  fs2
-    .io
-    .file
-    .Files[IO]
-    .readUtf8Lines(filePath)
+def fileHash(filePath: fs2.io.file.Path): IO[String] =
+  Files[IO]
+    .readAll(filePath)
+    .through(fs2.hash.md5)
+    .through(fs2.text.hex.encode)
     .compile
-    .toList
-    .map(lines => md.digest(lines.mkString("\n").getBytes).map("%02x".format(_)).mkString)
+    .string

project/src/live.server.scala

@@ -267,14 +267,15 @@ object LiveServer
 
           fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
           fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
-          refreshPub <- Topic[IO, String].toResource
+          refreshTopic <- Topic[IO, Unit].toResource
+          linkingTopic <- Topic[IO, Unit].toResource
           client <- EmberClientBuilder.default[IO].build
 
           proxyRoutes: HttpRoutes[IO] <- makeProxyRoutes(client, pathPrefix, proxyConfig)
 
           _ <- buildRunner(
             buildTool,
-            refreshPub,
+            linkingTopic,
             fs2.io.file.Path(baseDir),
             fs2.io.file.Path(outDir),
             extraBuildArgs,
@@ -283,11 +284,13 @@ object LiveServer
 
           indexHtmlTemplate = externalIndexHtmlTemplate.getOrElse(vanillaTemplate(stylesDir.isDefined).render)
 
-          app <- routes(outDir.toString(), refreshPub, stylesDir, proxyRoutes, indexHtmlTemplate, fileToHashRef)(logger)
+          app <- routes(outDir.toString(), refreshTopic, stylesDir, proxyRoutes, indexHtmlTemplate, fileToHashRef)(
+            logger
+          )
 
           _ <- seedMapOnStart(outDir, fileToHashMapRef)(logger)
           // _ <- stylesDir.fold(Resource.unit)(sd => seedMapOnStart(sd, mr))
-          _ <- fileWatcher(fs2.io.file.Path(outDir), fileToHashMapRef)(logger)
+          _ <- fileWatcher(fs2.io.file.Path(outDir), fileToHashMapRef, linkingTopic, refreshTopic)(logger)
           // _ <- stylesDir.fold(Resource.unit[IO])(sd => fileWatcher(fs2.io.file.Path(sd), mr))
           _ <- logger.info(s"Start dev server on http://localhost:$port").toResource
           server <- buildServer(app, port)

project/src/routes.scala

@@ -16,8 +16,6 @@ import org.typelevel.ci.CIStringSyntax
 
 import fs2.*
 import fs2.concurrent.Topic
-import fs2.io.Watcher
-import fs2.io.Watcher.Event
 import fs2.io.file.Files
 
 import scribe.Scribe
@@ -29,7 +27,6 @@ import cats.effect.IO
 import cats.effect.kernel.Ref
 import cats.effect.kernel.Resource
 import cats.effect.std.*
-import cats.effect.std.MapRef
 import cats.syntax.all.*
 
 import _root_.io.circe.syntax.EncoderOps
@@ -89,7 +86,7 @@ end ETagMiddleware
 
 def routes(
     stringPath: String,
-    refreshTopic: Topic[IO, String],
+    refreshTopic: Topic[IO, Unit],
     stylesPath: Option[String],
     proxyRoutes: HttpRoutes[IO],
     indexHtmlTemplate: String,
@@ -162,7 +159,7 @@ def seedMapOnStart(stringPath: String, mr: MapRef[IO, String, Option[String]])(l
           .isRegularFile(f)
           .ifM(
             // logger.trace(s"hashing $f") >>
-            fielHash(f).flatMap(
+            fileHash(f).flatMap(
               h =>
                 val key = asFs2.relativize(f)
                 logger.trace(s"hashing $f to put at $key with hash : $h") >>
@@ -179,46 +176,33 @@ end seedMapOnStart
 
 private def fileWatcher(
     stringPath: fs2.io.file.Path,
-    mr: MapRef[IO, String, Option[String]]
-)(logger: Scribe[IO]): ResourceIO[IO[OutcomeIO[Unit]]] =
-  fs2
-    .Stream
-    .resource(Watcher.default[IO].evalTap(_.watch(stringPath.toNioPath)))
-    .flatMap {
-      w =>
-        w.events()
-          .evalTap(
-            (e: Event) =>
-              e match
-                case Event.Created(path, i) =>
-                  // if path.endsWith(".js") then
-                  logger.trace(s"created $path, calculating hash") >>
-                    fielHash(fs2.io.file.Path(path.toString())).flatMap(
-                      h =>
-                        val serveAt = stringPath.relativize(fs2.io.file.Path(path.toString()))
-                        logger.trace(s"$serveAt :: hash -> $h") >>
-                          mr.setKeyValue(serveAt.toString(), h)
-                    )
-                // else IO.unit
-                case Event.Modified(path, i) =>
-                  // if path.endsWith(".js") then
-                  logger.trace(s"modified $path, calculating hash") >>
-                    fielHash(fs2.io.file.Path(path.toString())).flatMap(
-                      h =>
-                        val serveAt = stringPath.relativize(fs2.io.file.Path(path.toString()))
-                        logger.trace(s"$serveAt :: hash -> $h") >>
-                          mr.setKeyValue(serveAt.toString(), h)
-                    )
-                // else IO.unit
-                case Event.Deleted(path, i) =>
+    mr: MapRef[IO, String, Option[String]],
+    linkingTopic: Topic[IO, Unit],
+    refreshTopic: Topic[IO, Unit]
+)(logger: Scribe[IO]): ResourceIO[Unit] =
+  linkingTopic
+    .subscribe(10)
+    .evalTap {
+      _ =>
+        fs2
+          .io
+          .file
+          .Files[IO]
+          .list(stringPath)
+          .evalTap {
+            path =>
+              fileHash(path).flatMap(
+                h =>
                   val serveAt = stringPath.relativize(fs2.io.file.Path(path.toString()))
-                  logger.trace(s"deleted $path, removing key $serveAt") >>
-                    mr.unsetKey(serveAt.toString())
-                case e: Event.Overflow    => logger.info("overflow")
-                case e: Event.NonStandard => logger.info("non-standard")
-          )
+                  logger.trace(s"$serveAt :: hash -> $h") >>
+                    mr.setKeyValue(serveAt.toString(), h)
+              )
+          }
+          .compile
+          .drain >> refreshTopic.publish1(())
     }
     .compile
     .drain
     .background
+    .void
 end fileWatcher

project/test/src/RoutesSpec.scala

@@ -1,7 +1,5 @@
 import java.security.MessageDigest
 
-import scala.concurrent.duration.DurationInt
-
 import org.http4s.HttpRoutes
 import org.typelevel.ci.CIStringSyntax
 
@@ -13,7 +11,8 @@ import cats.effect.kernel.Ref
 import cats.effect.std.MapRef
 
 import munit.CatsEffectSuite
-// import cats.effect.unsafe.implicits.global
+
+import scala.concurrent.duration.*
 
 class ExampleSuite extends CatsEffectSuite:
 
@@ -56,22 +55,30 @@ class ExampleSuite extends CatsEffectSuite:
 
   files.test("watched map is updated") {
     tempDir =>
-      val newStr = "const hi = 'bye, world'"
-      val newHash = md.digest(testStr.getBytes()).map("%02x".format(_)).mkString
       val toCheck = for
         logger <- IO(scribe.cats[IO]).toResource
         fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
+        linkingTopic <- Topic[IO, Unit].toResource
+        refreshTopic <- Topic[IO, Unit].toResource
         fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
+        _ <- fileWatcher(fs2.io.file.Path(tempDir.toString), fileToHashMapRef, linkingTopic, refreshTopic)(logger)
+        _ <- IO.sleep(100.millis).toResource // wait for watcher to start
         _ <- seedMapOnStart(tempDir.toString, fileToHashMapRef)(logger)
-        _ <- fileWatcher(fs2.io.file.Path(tempDir.toString), fileToHashMapRef)(logger)
-        _ <- IO(os.write.over(tempDir / "test.js", newStr)).toResource
-        _ <- IO.sleep(1.second).toResource
-        updatedMap <- fileToHashRef.get.toResource
-      yield updatedMap
+        _ <- IO.blocking(os.write.over(tempDir / "test.js", "const hi = 'bye, world'")).toResource
+        _ <- linkingTopic.publish1(()).toResource
+        _ <- refreshTopic.subscribe(1).head.compile.resource.drain
+        oldHash <- fileToHashRef.get.map(_("test.js")).toResource
+        _ <- IO.blocking(os.write.over(tempDir / "test.js", "const hi = 'modified, workd'")).toResource
+        _ <- linkingTopic.publish1(()).toResource
+        _ <- refreshTopic.subscribe(1).head.compile.resource.drain
+        newHash <- fileToHashRef.get.map(_("test.js")).toResource
+      yield oldHash -> newHash
 
       toCheck.use {
-        updatedMap =>
-          assertIO(IO(updatedMap.get("test.js")), Some(newHash))
+        case (oldHash, newHash) =>
+          IO(assertNotEquals(oldHash, newHash)) >>
+            IO(assertEquals(oldHash, "27b2d040a66fb938f134c4b66fb7e9ce")) >>
+            IO(assertEquals(newHash, "3ebb82d4d6236c6bfbb90d65943b3e3d"))
       }
 
   }
@@ -84,7 +91,7 @@ class ExampleSuite extends CatsEffectSuite:
           fileToHashRef <- Ref[IO].of(Map.empty[String, String]).toResource
           fileToHashMapRef = MapRef.fromSingleImmutableMapRef[IO, String, String](fileToHashRef)
           _ <- seedMapOnStart(tempDir.toString, fileToHashMapRef)(logger)
-          refreshPub <- Topic[IO, String].toResource
+          refreshPub <- Topic[IO, Unit].toResource
           theseRoutes <- routes(
             tempDir.toString,
             refreshPub,