feat: add SSE endpoint and ChangeLog for live recipe reload
This commit is contained in:
+92
-10
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user