converge: phase 2 — SSE relay client for realtime pull on push

- 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
2026-05-14 14:13:43 -04:00
parent 25881f3ff8
commit 7feff5abfa
5 changed files with 193 additions and 6 deletions
@@ -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
@@ -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
        }
@@ -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

@@ -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
@@ -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