From f5c693a2b5b8678d700af5818c9001716b2bfe2f Mon Sep 17 00:00:00 2001 From: James Brechtel Date: Wed, 20 May 2026 10:15:50 -0400 Subject: [PATCH] feat: add SSE endpoint and ChangeLog for live recipe reload --- src/Roux/Server.hs | 102 ++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 92 insertions(+), 10 deletions(-) diff --git a/src/Roux/Server.hs b/src/Roux/Server.hs index 7cfc764..f344c30 100644 --- a/src/Roux/Server.hs +++ b/src/Roux/Server.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -11,15 +12,20 @@ module Roux.Server ( ) where import Control.Concurrent (forkIO, threadDelay) -import Control.Exception (IOException, SomeException, catch, try) -import Control.Monad (forever, when) +import Control.Exception (IOException, SomeException, catch, finally, try) +import Control.Monad (forM_, forever, when) +import Data.ByteString.Builder (lazyByteString) import Data.ByteString.Lazy (ByteString) +import qualified Data.ByteString.Lazy as LB import Data.Char (toLower) -import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) +import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef, writeIORef) import Data.List (isSuffixOf) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map import Data.Text (Text) +import qualified Data.Text as T +import Data.Text.Encoding (encodeUtf8) +import Data.Time.Clock (diffUTCTime, getCurrentTime) import qualified Network.HTTP.Types as HTTP import qualified Network.Wai as Wai @@ -58,21 +64,95 @@ app recipeDir = do state <- newIORef recipeMap -- Resolve to absolute path so makeRelative works with fsnotify events absDir <- makeAbsolute recipeDir + -- Change log for SSE notifications + changeLog <- newIORef (0, []) -- Start the filesystem watcher _ <- forkIO $ - watchRecipes absDir state `catch` \e -> + watchRecipes absDir state changeLog `catch` \e -> putStrLn $ "[roux] watcher thread crashed: " <> show (e :: SomeException) - pure (handleRequest state) + pure (router state changeLog) where isRight (Right _) = True isRight (Left _) = False +{- | A change log with a monotonic version counter and +the last N changed filenames (newest first). +-} +type ChangeLog = IORef (Int, [(Int, FilePath)]) + +-- | Record a recipe file change in the change log. +recordChange :: ChangeLog -> FilePath -> IO () +recordChange changeLog path = atomicModifyIORef' changeLog $ \(version, entries) -> + let newVersion = version + 1 + newEntries = (newVersion, path) : take 99 entries + in ((newVersion, newEntries), ()) + +{- | SSE endpoint — streams recipe-change events to connected clients. +Polls 'ChangeLog' every 500ms and pushes any new events since the +client's last seen version. Sends a @ping@ event every 30s as a +keepalive. Client disconnect is detected by Warp when the write +action throws. +-} +sseHandler :: ChangeLog -> Wai.Application +sseHandler changeLog _request respond = do + respond + $ Wai.responseStream + HTTP.status200 + [ ("Content-Type", "text/event-stream") + , ("Cache-Control", "no-cache") + , ("Connection", "keep-alive") + ] + $ \write flush -> do + lastSent <- newIORef 0 + lastPing <- newIORef =<< getCurrentTime + let sendPing = do + write "event: ping\ndata: keepalive\n\n" + flush + let go = do + now <- getCurrentTime + (version, entries) <- readIORef changeLog + sent <- readIORef lastSent + when (version > sent) $ do + let newEntries = takeWhile (\(v, _) -> v > sent) entries + forM_ newEntries $ \(_v, path) -> do + write + ( lazyByteString $ + LB.fromStrict $ + encodeUtf8 $ + T.unlines + [ "event: recipe-changed" + , "data: " <> T.pack path + , "" + , "" + ] + ) + writeIORef lastSent version + flush + pingTime <- readIORef lastPing + when (diffUTCTime now pingTime > 30) $ do + writeIORef lastPing now + sendPing + threadDelay 500_000 -- 500ms poll interval + go + -- Send an initial ping so the client detects connection immediately + pingTime <- getCurrentTime + writeIORef lastPing pingTime + sendPing + go `finally` pure () + +-- | Route requests — SSE endpoint or recipe pages. +router :: IORef (Map FilePath Idx.RecipeInfo) -> ChangeLog -> Wai.Application +router state changeLog request respond = + case Wai.pathInfo request of + ["events"] -> sseHandler changeLog request respond + _ -> handleRequest state request respond + {- | Watch a directory for .cook file changes and update the shared state. Uses watchDir callback for simplicity. -} -watchRecipes :: FilePath -> IORef (Map FilePath Idx.RecipeInfo) -> IO () -watchRecipes dir state = do +watchRecipes :: FilePath -> IORef (Map FilePath Idx.RecipeInfo) -> ChangeLog -> IO () +watchRecipes dir state changeLog = do putStrLn $ "[roux] watching " <> dir <> " for changes" let cfg = defaultConfig{confWatchMode = WatchModeOS} withManagerConf cfg $ \mgr -> do @@ -81,7 +161,7 @@ watchRecipes dir state = do ( do when (eventIsDirectory ev == IsFile && ".cook" `isSuffixOf` eventPath ev) $ do let path = makeRelative dir (eventPath ev) - reReadRecipe dir path state + reReadRecipe dir path state changeLog ) ( \e -> putStrLn $ @@ -93,8 +173,8 @@ watchRecipes dir state = do {- | Re-read a single recipe file and update the shared state. On read/parse errors, logs the error and keeps the previous version. -} -reReadRecipe :: FilePath -> FilePath -> IORef (Map FilePath Idx.RecipeInfo) -> IO () -reReadRecipe dir path state = do +reReadRecipe :: FilePath -> FilePath -> IORef (Map FilePath Idx.RecipeInfo) -> ChangeLog -> IO () +reReadRecipe dir path state changeLog = do let key = map toLower (stripCookExt path) let fullPath = dir path result <- try (doesFileExist fullPath) @@ -104,6 +184,7 @@ reReadRecipe dir path state = do Right False -> do atomicModifyIORef' state $ \m -> (Map.delete key m, ()) + recordChange changeLog path putStrLn $ "[roux] removed " <> path Right True -> do content <- try (Idx.loadRecipe dir path) @@ -117,6 +198,7 @@ reReadRecipe dir path state = do Right _ -> do atomicModifyIORef' state $ \m -> (Map.insert key info m, ()) + recordChange changeLog path putStrLn $ "[roux] updated " <> path -- | Route an incoming request to the appropriate handler.