From 7feff5abfa8595293aa0e9ca444c95c6bf93e5dc Mon Sep 17 00:00:00 2001 From: James Brechtel Date: Thu, 14 May 2026 14:13:43 -0400 Subject: [PATCH] =?UTF-8?q?converge:=20phase=202=20=E2=80=94=20SSE=20relay?= =?UTF-8?q?=20client=20for=20realtime=20pull=20on=20push?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RepoConfig gains rcGiteaRepo :: Maybe Text for relay matching - Options gains optRelayUrl :: Maybe Text (--relay-url CLI, relay_url config) - connectRelay connects to converge-relay SSE endpoint, parses push events, triggers gitPull on matching repos (by gitea_repo + branch) - Exponential backoff reconnection (1s .. 60s) on disconnect/error - Skips pull if repo is mid-operation (merge/rebase/etc.) - Dependencies: http-client, http-client-tls, http-types, aeson, bytestring - SSE parser handles event:/data: fields, blank-line delimiters, CRLF endings - Updated SPEC.md, example config, and help text --- SPEC.md | 5 ++ app/Main.hs | 23 ++++-- converge.example.yaml | 4 ++ package.yaml | 5 ++ src/Converge.hs | 162 +++++++++++++++++++++++++++++++++++++++++- 5 files changed, 193 insertions(+), 6 deletions(-) diff --git a/SPEC.md b/SPEC.md index d8058e2..dde05c6 100644 --- a/SPEC.md +++ b/SPEC.md @@ -18,3 +18,8 @@ ## Can notify on pull: per-repo notify_on_pull setting sends a desktop notification when new commits are pulled from the remote ## --sync / -s flag runs a single sync cycle (commit + pull + push) and exits, useful for wake-from-sleep hooks + +## Can connect to a converge-relay SSE endpoint and automatically pull when push events arrive +### Configured via --relay-url CLI flag or relay_url config file entry +### Matches repos by gitea_repo field (Gitea repository full_name) and branch +### Reconnects automatically with exponential backoff on disconnect or error diff --git a/app/Main.hs b/app/Main.hs index 975aa80..cbaab2b 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -24,6 +24,8 @@ data CliArgs = CliArgs -- ^ Log file path. , cliSync :: !Bool -- ^ Run a single sync cycle and exit (useful for wake-from-sleep hooks). + , cliRelayUrl :: !(Maybe Text) + -- ^ URL of a converge-relay SSE endpoint (overrides config). } cliParser :: Parser CliArgs @@ -97,6 +99,13 @@ cliParser = <> short 's' <> help "Run a single sync cycle (commit + pull + push) and exit" ) + <*> optional + ( strOption + ( long "relay-url" + <> metavar "URL" + <> help "URL of a converge-relay SSE endpoint for realtime push notifications" + ) + ) parserInfo :: ParserInfo CliArgs parserInfo = @@ -114,7 +123,7 @@ main = do (options, configSource) <- case cliConfig cli of Just cfgPath -> do cfg <- loadConfig cfgPath - let opts = applyOverrides mDebounceOverride (cliLogLevel cli) (cliLogFile cli) (configToOptions cfg) + let opts = applyOverrides mDebounceOverride (cliLogLevel cli) (cliLogFile cli) (cliRelayUrl cli) (configToOptions cfg) pure (opts, Just cfgPath) Nothing -> do exists <- configFileExists @@ -122,7 +131,7 @@ main = do then do cfgPath <- defaultConfigPath cfg <- loadConfig cfgPath - let opts = applyOverrides mDebounceOverride (cliLogLevel cli) (cliLogFile cli) (configToOptions cfg) + let opts = applyOverrides mDebounceOverride (cliLogLevel cli) (cliLogFile cli) (cliRelayUrl cli) (configToOptions cfg) pure (opts, Just cfgPath) else pure @@ -130,6 +139,7 @@ main = do mDebounceOverride (cliLogLevel cli) (cliLogFile cli) + (cliRelayUrl cli) defaultOptions { optRepos = [ defaultRepoConfig @@ -168,13 +178,16 @@ main = do mapM_ (\r -> hPutStrLn stderr (" - " <> rcPath r)) repos runConverge options --- | Apply CLI overrides (debounce, log level, log file) to Options. -applyOverrides :: Maybe Int -> Maybe LogLevel -> Maybe FilePath -> Options -> Options -applyOverrides mDebounce mLogLevel mLogFile opts = +-- | Apply CLI overrides (debounce, log level, log file, relay url) to Options. +applyOverrides :: Maybe Int -> Maybe LogLevel -> Maybe FilePath -> Maybe Text -> Options -> Options +applyOverrides mDebounce mLogLevel mLogFile mRelayUrl opts = opts { optDebounceMs = maybe (optDebounceMs opts) id mDebounce , optLogLevel = fromMaybe (optLogLevel opts) mLogLevel , optLogFile = case mLogFile of Just _ -> mLogFile -- CLI --log-file overrides config Nothing -> optLogFile opts + , optRelayUrl = case mRelayUrl of + Just _ -> mRelayUrl -- CLI --relay-url overrides config + Nothing -> optRelayUrl opts } diff --git a/converge.example.yaml b/converge.example.yaml index 9e46efb..0725ccd 100644 --- a/converge.example.yaml +++ b/converge.example.yaml @@ -2,14 +2,18 @@ # Place at ~/.config/converge/config.yaml to use by default, # or pass explicitly with: converge --config converge.yaml +# relay_url: "http://localhost:8080/events" # converge-relay SSE endpoint for realtime push notifications + repos: - path: /home/jbrechtel/org remote: origin branch: main + # gitea_repo: "myorg/org-repo" # Gitea repository full_name for relay matching # notify_on_pull: true # send a notification when new commits are pulled - path: /work/personal/hawat # remote defaults to "origin" # branch defaults to "main" + # gitea_repo: "myorg/hawat" # debounce_ms: 5000 # optional, defaults to 5000 diff --git a/package.yaml b/package.yaml index f4c5fde..1347a8d 100644 --- a/package.yaml +++ b/package.yaml @@ -29,10 +29,15 @@ ghc-options: dependencies: - base >= 4.7 && < 5 + - aeson - async + - bytestring - directory - filepath - fsnotify + - http-client + - http-client-tls + - http-types - optparse-applicative - process - text diff --git a/src/Converge.hs b/src/Converge.hs index d19ac6b..39680d0 100644 --- a/src/Converge.hs +++ b/src/Converge.hs @@ -35,20 +35,31 @@ module Converge ( -- * File filtering isGitIgnored, + + -- * Relay listener (SSE client) + connectRelay, ) where +import Control.Applicative ((<|>)) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.Async (mapConcurrently_) import Control.Concurrent.MVar (MVar, newEmptyMVar, newMVar, takeMVar, tryPutMVar, tryTakeMVar, withMVar) -import Control.Exception (bracket) +import Control.Exception (SomeException, bracket, try) import Control.Monad (forever, unless, void, when) +import Data.Aeson (decode) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy.Char8 as BL import Data.Maybe (fromMaybe) import Data.Text (Text) import qualified Data.Text as T +import Data.Text.Encoding (decodeUtf8) import Data.Time.Clock (getCurrentTime) import Data.Time.Format (defaultTimeLocale, formatTime) import Data.Yaml (FromJSON (..), decodeFileThrow, withObject, withText, (.:), (.:?)) +import Network.HTTP.Client +import Network.HTTP.Client.TLS (tlsManagerSettings) +import Network.HTTP.Types.Status (status200) import System.Directory (XdgDirectory (XdgConfig), doesDirectoryExist, doesFileExist, getCurrentDirectory, getXdgDirectory) import System.Exit (ExitCode (..)) import System.FSNotify @@ -66,6 +77,8 @@ data RepoConfig = RepoConfig -- ^ Branch to pull. , rcNotifyOnPull :: !Bool -- ^ Send a notification when new commits are pulled from the remote. + , rcGiteaRepo :: !(Maybe Text) + -- ^ Gitea repository full_name (e.g. "myorg/myrepo") for relay matching. } deriving (Show, Eq) @@ -97,6 +110,8 @@ data Options = Options -- ^ Minimum log level to emit. , optLogFile :: !(Maybe FilePath) -- ^ Optional log file path (logs to stderr when Nothing). + , optRelayUrl :: !(Maybe Text) + -- ^ URL of a converge-relay SSE endpoint for realtime push notifications. } deriving (Show, Eq) @@ -108,6 +123,7 @@ defaultRepoConfig = , rcRemote = "origin" , rcBranch = "main" , rcNotifyOnPull = False + , rcGiteaRepo = Nothing } -- | Sensible defaults for options. @@ -119,6 +135,7 @@ defaultOptions = , optNotifyCmd = "notify-send" , optLogLevel = Info , optLogFile = Nothing + , optRelayUrl = Nothing } {- | Logger that serializes output so concurrent repo watchers @@ -153,6 +170,7 @@ data Config = Config , cfgDebounceMs :: !(Maybe Int) , cfgLogLevel :: !(Maybe LogLevel) , cfgLogFile :: !(Maybe FilePath) + , cfgRelayUrl :: !(Maybe Text) } deriving (Show, Eq) @@ -163,6 +181,7 @@ instance FromJSON Config where <*> o .:? "debounce_ms" <*> o .:? "log_level" <*> o .:? "log_file" + <*> o .:? "relay_url" instance FromJSON RepoConfig where parseJSON = withObject "RepoConfig" $ \o -> @@ -171,6 +190,7 @@ instance FromJSON RepoConfig where <*> (fromMaybe "origin" <$> o .:? "remote") <*> (fromMaybe "main" <$> o .:? "branch") <*> (fromMaybe False <$> o .:? "notify_on_pull") + <*> o .:? "gitea_repo" -- | Load a YAML config file. loadConfig :: FilePath -> IO Config @@ -184,6 +204,7 @@ configToOptions cfg = , optDebounceMs = fromMaybe 5000 (cfgDebounceMs cfg) , optLogLevel = fromMaybe Info (cfgLogLevel cfg) , optLogFile = cfgLogFile cfg + , optRelayUrl = cfgRelayUrl cfg } -- | The default XDG config file path: @$XDG_CONFIG_HOME/converge/config.yaml@. @@ -239,6 +260,13 @@ logRepo logger repo msg = logRepoAt logger Info repo msg runConverge :: Options -> IO () runConverge opts = do logger <- newLogger (optLogLevel opts) (optLogFile opts) + -- Start relay listener if configured (runs in background forever) + case optRelayUrl opts of + Just url + | url /= "" -> do + _ <- forkIO $ connectRelay url logger (optRepos opts) + pure () + _ -> pure () mapConcurrently_ (watchRepo opts logger) (optRepos opts) {- | Watch a single repository for changes. @@ -518,3 +546,135 @@ runGitIn :: FilePath -> [String] -> IO (ExitCode, Text, Text) runGitIn repoPath args = do (code, out, err) <- readProcessWithExitCode "git" ("-C" : repoPath : args) "" pure (code, T.pack out, T.pack err) + +---------------------------------------------------------------------- +-- Relay listener (SSE client) +---------------------------------------------------------------------- + +-- | Push event received from a converge-relay SSE endpoint. +data PushEvent = PushEvent + { peFullName :: !Text + -- ^ Gitea repository full_name (e.g. "myorg/myrepo"). + , peBranch :: !Text + -- ^ Branch name. + , peRef :: !Text + -- ^ Full ref (e.g. "refs/heads/main"). + } + deriving (Show, Eq) + +instance FromJSON PushEvent where + parseJSON = withObject "PushEvent" $ \o -> + PushEvent + <$> o .: "full_name" + <*> o .: "branch" + <*> o .: "ref" + +{- | Connect to a converge-relay SSE endpoint and trigger pulls on matching + repositories when push events arrive. Runs forever, reconnecting with + exponential backoff (1s .. 60s) on disconnection or error. + + This is spawned as a background thread by 'runConverge' when an + @optRelayUrl@ is configured. +-} +connectRelay :: Text -> Logger -> [RepoConfig] -> IO () +connectRelay url logger repos = do + logMsg logger Info ("relay: connecting to " <> url) + req <- parseRequest (T.unpack url) + manager <- newManager tlsManagerSettings + let loop delay = do + let delaySec = min delay (60 :: Int) + result <- try $ + withResponse req manager $ \resp -> + if responseStatus resp /= status200 + then logMsg logger Error ("relay returned " <> T.pack (show (responseStatus resp))) + else do + logMsg logger Info "relay: connected, waiting for push events" + processSSE logger repos (responseBody resp) + case result of + Left (e :: SomeException) -> + logMsg logger Warn ("relay: connection error: " <> T.pack (show e) <> ", reconnecting in " <> T.pack (show delaySec) <> "s") + Right () -> + logMsg logger Info ("relay: connection closed, reconnecting in " <> T.pack (show delaySec) <> "s") + threadDelay (delaySec * 1000000) + loop (delaySec * 2) + loop 1 + +-- | Process an SSE stream from a 'BodyReader', dispatching push events to matching repos. +processSSE :: Logger -> [RepoConfig] -> BodyReader -> IO () +processSSE logger repos reader = go BS.empty Nothing [] + where + go buf mEvtType dataLines = do + chunk <- brRead reader + if BS.null chunk + then pure () + else do + let (completeLines, leftover) = extractLines (buf <> chunk) + let (mEvtType', dataLines', events) = foldLines mEvtType dataLines completeLines + mapM_ (handleSSEEvent logger repos) events + go leftover mEvtType' dataLines' + +-- | Split a 'BS.ByteString' into complete lines (stripping trailing \r) and leftover partial data. +extractLines :: BS.ByteString -> ([BS.ByteString], BS.ByteString) +extractLines bs = case BS.breakSubstring "\n" bs of + (line, rest) + | not (BS.null rest) -> + let line' = stripTrailingCr line + (lines', leftover) = extractLines (BS.drop 1 rest) + in (line' : lines', leftover) + _ -> ([], bs) + +-- | Strip trailing \r from a line (for CRLF line endings). +stripTrailingCr :: BS.ByteString -> BS.ByteString +stripTrailingCr bs + | not (BS.null bs) && BS.last bs == cr = BS.init bs + | otherwise = bs + where + cr = 13 -- '\r' + +-- | Fold a list of SSE lines into accumulated state and parsed events. +foldLines :: Maybe Text -> [BS.ByteString] -> [BS.ByteString] -> (Maybe Text, [BS.ByteString], [PushEvent]) +foldLines evtType0 dLines0 sseLines = foldl' step (evtType0, dLines0, []) sseLines + where + step (evt, dl, events) line + | BS.null line = + case dl of + [] -> (Nothing, [], events) + _ -> case parseEvent evt dl of + Just evt' -> (Nothing, [], evt' : events) + Nothing -> (Nothing, [], events) + | Just rest <- pick "event: " line <|> pick "event:" line = + (Just (decodeUtf8 (BS.dropWhile (== space) rest)), dl, events) + | Just rest <- pick "data: " line <|> pick "data:" line = + (evt, BS.dropWhile (== space) rest : dl, events) + | otherwise = (evt, dl, events) + pick = BS.stripPrefix + space = 32 -- ' ' + +-- | Parse accumulated SSE data lines into a 'PushEvent' (push events only). +parseEvent :: Maybe Text -> [BS.ByteString] -> Maybe PushEvent +parseEvent (Just "push") dataLines = + decode (BL.fromStrict (BS.intercalate "\n" (reverse dataLines))) +parseEvent _ _ = Nothing + +-- | Dispatch a parsed push event to matching repos (by 'rcGiteaRepo' and 'rcBranch'). +handleSSEEvent :: Logger -> [RepoConfig] -> PushEvent -> IO () +handleSSEEvent logger repos evt = do + let matching = + filter + ( \r -> + rcGiteaRepo r == Just (peFullName evt) + && rcBranch r == peBranch evt + ) + repos + if null matching + then logMsg logger Debug ("relay: ignoring event for unconfigured repo: " <> peFullName evt <> " (branch: " <> peBranch evt <> ")") + else + mapM_ + ( \repo -> do + logRepo logger repo "relay: push event received, pulling" + inProgress <- isInMiddleOfOperation (rcPath repo) + if inProgress + then logRepoAt logger Warn repo "relay: skipping pull -- operation in progress" + else void $ gitPull logger repo + ) + matching