{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeApplications #-}

-- | Out-of-process bus helpers: file cursors and an fsnotify-based tail
-- loop. Used by the long-running agent executables that watch the JSONL log
-- directly instead of sharing STM state with the scribe.
module Free.Agent.Bus.File
  ( -- * Cursor files
    cursorPath,
    readCursor,
    writeCursor,

    -- * Event-tail loop
    QuiesceConfig (..),
    Flow (..),
    tailLog,
  )
where

import Circuit.Agent (Name, Post (..), PostId, deliversTo)
import Circuit.Agent.Framing (Stamped, stamp, stamped, unframeStored)
import Control.Concurrent (threadDelay)
import Control.Concurrent.STM
  ( TMVar,
    atomically,
    check,
    newTMVarIO,
    newTVarIO,
    orElse,
    readTVar,
    readTVarIO,
    takeTMVar,
    tryPutTMVar,
    writeTVar,
  )
import Control.Monad (guard, unless, when)
import Data.ByteString qualified as BS
import Data.IORef (newIORef, readIORef, writeIORef)
import Data.Maybe (mapMaybe)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.Encoding qualified as TE
import Data.Text.IO qualified as TIO
import System.Directory (doesFileExist)
import System.FSNotify (Event (..), watchDir, withManager)
import System.FilePath (takeDirectory, takeFileName, (</>))
import System.IO
  ( IOMode (AppendMode, ReadMode),
    SeekMode (AbsoluteSeek),
    hSeek,
    withFile,
  )
import System.Timeout (timeout)
import Text.Read (readMaybe)

-- | Quiescence configuration for long-running agents.
data QuiesceConfig = QuiesceConfig
  { -- | Number of empty cycles before taking action.
    QuiesceConfig -> Int
qcCycles :: Int,
    -- | Recipient name for the quiescence marker.
    QuiesceConfig -> Text
qcPitboss :: Name,
    -- | Length of one cycle in microseconds.
    QuiesceConfig -> Int
qcCycleMicros :: Int
  }
  deriving (Int -> QuiesceConfig -> ShowS
[QuiesceConfig] -> ShowS
QuiesceConfig -> FilePath
(Int -> QuiesceConfig -> ShowS)
-> (QuiesceConfig -> FilePath)
-> ([QuiesceConfig] -> ShowS)
-> Show QuiesceConfig
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> QuiesceConfig -> ShowS
showsPrec :: Int -> QuiesceConfig -> ShowS
$cshow :: QuiesceConfig -> FilePath
show :: QuiesceConfig -> FilePath
$cshowList :: [QuiesceConfig] -> ShowS
showList :: [QuiesceConfig] -> ShowS
Show)

-- | Flow control returned by a 'tailLog' callback: keep listening, or halt
-- the loop after this post. This is decided quiet at the seat level: a
-- callback that returns 'Halt' ends the exchange by content, not by
-- timeout.
data Flow = Continue | Halt
  deriving (Flow -> Flow -> Bool
(Flow -> Flow -> Bool) -> (Flow -> Flow -> Bool) -> Eq Flow
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Flow -> Flow -> Bool
== :: Flow -> Flow -> Bool
$c/= :: Flow -> Flow -> Bool
/= :: Flow -> Flow -> Bool
Eq, Int -> Flow -> ShowS
[Flow] -> ShowS
Flow -> FilePath
(Int -> Flow -> ShowS)
-> (Flow -> FilePath) -> ([Flow] -> ShowS) -> Show Flow
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Flow -> ShowS
showsPrec :: Int -> Flow -> ShowS
$cshow :: Flow -> FilePath
show :: Flow -> FilePath
$cshowList :: [Flow] -> ShowS
showList :: [Flow] -> ShowS
Show)

-- | Path to the cursor file for an agent.
--
-- The cursor stores the next 'postId' the agent should process, so restarts
-- catch up without re-reading the whole log. A missing cursor defaults to 0,
-- meaning "start from the first post".
cursorPath :: FilePath -> Name -> FilePath
cursorPath :: FilePath -> Text -> FilePath
cursorPath FilePath
root Text
name = FilePath
root FilePath -> ShowS
</> (FilePath
".cursor-" FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> Text -> FilePath
T.unpack Text
name)

-- | Read the cursor for an agent. If no cursor file exists, defaults to the
-- latest post id + 1 in the bus log so the agent skips history on cold boot.
-- Returns 0 only when the log itself is absent or empty.
readCursor :: FilePath -> Name -> IO PostId
readCursor :: FilePath -> Text -> IO PostId
readCursor FilePath
root Text
name = do
  let path :: FilePath
path = FilePath -> Text -> FilePath
cursorPath FilePath
root Text
name
  exists <- FilePath -> IO Bool
doesFileExist FilePath
path
  if not exists
    then latestPostId root
    else do
      txt <- TIO.readFile path
      pure $ maybe 0 fromIntegral (readMaybe @Integer (T.unpack (T.strip txt)))

-- | Return the id that would follow the last post in the log (i.e. the
-- post count), or 0 if the log doesn't exist or is empty. Used as the
-- default cursor on cold boot so the agent only processes new posts.
latestPostId :: FilePath -> IO PostId
latestPostId :: FilePath -> IO PostId
latestPostId FilePath
root = do
  let logPath :: FilePath
logPath = FilePath
root FilePath -> ShowS
</> FilePath
"log.jsonl"
  logExists <- FilePath -> IO Bool
doesFileExist FilePath
logPath
  if not logExists
    then pure 0
    else do
      txt <- TIO.readFile logPath
      let ls = (Text -> Bool) -> [Text] -> [Text]
forall a. (a -> Bool) -> [a] -> [a]
filter (Bool -> Bool
not (Bool -> Bool) -> (Text -> Bool) -> Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Bool
T.null) (Text -> [Text]
T.lines Text
txt)
      case ls of
        [] -> PostId -> IO PostId
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure PostId
0
        [Text]
_ -> case forall a. PostBody a => Text -> Maybe (Stamped a)
unframeStored @Text ([Text] -> Text
forall a. HasCallStack => [a] -> a
last [Text]
ls) of
          Just Stamped Text
stored -> PostId -> IO PostId
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ((UTCTime, PostId) -> PostId
forall a b. (a, b) -> b
snd (Stamped Text -> (UTCTime, PostId)
forall r a. Stamped r a -> r
stamp Stamped Text
stored) PostId -> PostId -> PostId
forall a. Num a => a -> a -> a
+ PostId
1)
          Maybe (Stamped Text)
Nothing -> PostId -> IO PostId
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int -> PostId
forall a b. (Integral a, Num b) => a -> b
fromIntegral ([Text] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [Text]
ls))

-- | Persist the cursor for an agent. Writes @stamp + 1@ so the next wake
-- starts after the post just processed.
writeCursor :: FilePath -> Name -> PostId -> IO ()
writeCursor :: FilePath -> Text -> PostId -> IO ()
writeCursor FilePath
root Text
name PostId
pid =
  FilePath -> Text -> IO ()
TIO.writeFile (FilePath -> Text -> FilePath
cursorPath FilePath
root Text
name) (FilePath -> Text
T.pack (PostId -> FilePath
forall a. Show a => a -> FilePath
show PostId
pid))

-- | Event-tail a log file and invoke the callback for every new stored post
-- addressed to any of the subscribed names.
--
-- On startup the file is scanned from the beginning; posts with 'stamp'
-- greater than or equal to the supplied cursor and addressed to any subscribed
-- name are delivered. After catch-up, fsnotify wakes a drain for new lines.
--
-- Reading is offset-based: each drain opens the file, reads the complete
-- lines appended since the last offset, and closes the handle /before/
-- invoking callbacks. A partial trailing line (writer mid-append) is left
-- for the next drain. The handle must be closed before callbacks run
-- because callbacks may append to the log in-process, and GHC locks files
-- per process — a held read handle makes the append fail with
-- "resource busy (file is locked)".
--
-- When a quiescence config is supplied, the main loop waits on a signal with a
-- timeout instead of blocking forever. Each timeout without a signal increments
-- an empty-cycle counter; a signal resets it. After the configured number of
-- empty cycles, the provided action is run and the loop exits.
--
-- The loop also exits when a callback returns 'Halt'.
tailLog ::
  -- | Path to @log.jsonl@.
  FilePath ->
  -- | Subscribed names.
  [Name] ->
  -- | Starting cursor.
  PostId ->
  -- | Optional quiescence config and action.
  Maybe (QuiesceConfig, IO ()) ->
  -- | Callback for each delivered stored post; return 'Halt' to stop.
  (Stamped Text -> IO Flow) ->
  IO ()
tailLog :: FilePath
-> [Text]
-> PostId
-> Maybe (QuiesceConfig, IO ())
-> (Stamped Text -> IO Flow)
-> IO ()
tailLog FilePath
path [Text]
names PostId
startCursor Maybe (QuiesceConfig, IO ())
mQuiesce Stamped Text -> IO Flow
cb = do
  exists <- FilePath -> IO Bool
doesFileExist FilePath
path
  unless exists $ do
    -- Touch an empty log so the scribe has a file to append to.
    withFile path AppendMode (\Handle
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
  (off0, halted0) <- drainFrom 0 (filterStoredSince startCursor)
  offRef <- newIORef off0
  halted <- newTVarIO halted0
  let logName = ShowS
takeFileName FilePath
path
      dir = ShowS
takeDirectory FilePath
path
  withManager $ \WatchManager
mgr -> do
    signal <- () -> IO (TMVar ())
forall a. a -> IO (TMVar a)
newTMVarIO ()
    busy <- newTVarIO True -- quiescence gated: don't count empty cycles until first drain completes
    _ <- watchDir mgr dir (\Event
ev -> ShowS
takeFileName (Event -> FilePath
eventPath Event
ev) FilePath -> FilePath -> Bool
forall a. Eq a => a -> a -> Bool
== FilePath
logName) $ \Event
_ev -> do
      already <- TVar Bool -> IO Bool
forall a. TVar a -> IO a
readTVarIO TVar Bool
halted
      unless already $ do
        atomically $ writeTVar busy True
        off <- readIORef offRef
        (off', h) <- drainFrom off filterStored
        writeIORef offRef off'
        atomically $ do
          when h (writeTVar halted True)
          writeTVar busy False
          _ <- tryPutTMVar signal ()
          pure ()
    case mQuiesce of
      Maybe (QuiesceConfig, IO ())
Nothing -> IORef Integer -> TMVar () -> TVar Bool -> IO ()
pollLoop IORef Integer
offRef TMVar ()
signal TVar Bool
halted
      Just (QuiesceConfig
qc, IO ()
onQuiesce) -> QuiesceConfig
-> TMVar () -> TVar Bool -> TVar Bool -> Int -> IO () -> IO ()
forall {a}.
QuiesceConfig
-> TMVar a -> TVar Bool -> TVar Bool -> Int -> IO () -> IO ()
quiesceLoop QuiesceConfig
qc TMVar ()
signal TVar Bool
busy TVar Bool
halted Int
0 IO ()
onQuiesce
  where
    -- \| No-quiesce poll loop: wait on fsnotify with a 1 s timeout, then
    -- manually re-drain.  FSEvents on macOS does not reliably fire on
    -- append, so the timeout fallback ensures delivery within ~1 s.
    pollLoop :: IORef Integer -> TMVar () -> TVar Bool -> IO ()
pollLoop IORef Integer
offRef TMVar ()
signal TVar Bool
halted = do
      h <- TVar Bool -> IO Bool
forall a. TVar a -> IO a
readTVarIO TVar Bool
halted
      if h
        then pure ()
        else do
          m <- timeout 1_000_000 (atomically (takeTMVar signal))
          case m of
            Just () -> do
              off <- IORef Integer -> IO Integer
forall a. IORef a -> IO a
readIORef IORef Integer
offRef
              (off', h') <- drainFrom off filterStored
              writeIORef offRef off'
              when h' (atomically (writeTVar halted True))
            Maybe ()
Nothing -> do
              -- timeout: poll the file manually
              off <- IORef Integer -> IO Integer
forall a. IORef a -> IO a
readIORef IORef Integer
offRef
              (off', h') <- drainFrom off filterStored
              writeIORef offRef off'
              when h' (atomically (writeTVar halted True))
          pollLoop offRef signal halted
    drainFrom :: Integer -> (Text -> Maybe (Stamped Text)) -> IO (Integer, Bool)
drainFrom Integer
off Text -> Maybe (Stamped Text)
filt = do
      (ls, off') <- Integer -> IO ([Text], Integer)
readCompleteLines Integer
off
      halted <- deliver (mapMaybe filt ls)
      pure (off', halted)

    -- Deliver posts oldest first; stop early on 'Halt'.
    deliver :: [Stamped Text] -> IO Bool
deliver [] = Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
    deliver (Stamped Text
p : [Stamped Text]
ps) = do
      flow <- Stamped Text -> IO Flow
cb Stamped Text
p
      case flow of
        Flow
Halt -> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
        Flow
Continue -> [Stamped Text] -> IO Bool
deliver [Stamped Text]
ps

    readCompleteLines :: Integer -> IO ([Text], Integer)
readCompleteLines Integer
off = FilePath
-> IOMode
-> (Handle -> IO ([Text], Integer))
-> IO ([Text], Integer)
forall r. FilePath -> IOMode -> (Handle -> IO r) -> IO r
withFile FilePath
path IOMode
ReadMode ((Handle -> IO ([Text], Integer)) -> IO ([Text], Integer))
-> (Handle -> IO ([Text], Integer)) -> IO ([Text], Integer)
forall a b. (a -> b) -> a -> b
$ \Handle
h -> do
      Handle -> SeekMode -> Integer -> IO ()
hSeek Handle
h SeekMode
AbsoluteSeek Integer
off
      bs <- Handle -> IO ByteString
BS.hGetContents Handle
h
      pure (completeLines off bs)

    -- Lines up to the last newline are complete; the byte offset advances
    -- past it. Anything after the last newline is a writer mid-append and
    -- waits for the next drain.
    completeLines :: b -> ByteString -> ([Text], b)
completeLines b
off ByteString
bs =
      case Word8 -> ByteString -> Maybe Int
BS.elemIndexEnd Word8
0x0A ByteString
bs of
        Maybe Int
Nothing -> ([], b
off)
        Just Int
i ->
          ( Text -> [Text]
T.lines (ByteString -> Text
TE.decodeUtf8 (Int -> ByteString -> ByteString
BS.take (Int
i Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1) ByteString
bs)),
            b
off b -> b -> b
forall a. Num a => a -> a -> a
+ Int -> b
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
i b -> b -> b
forall a. Num a => a -> a -> a
+ b
1
          )

    filterStoredSince :: PostId -> Text -> Maybe (Stamped Text)
filterStoredSince PostId
cursor Text
line = do
      stored <- forall a. PostBody a => Text -> Maybe (Stamped a)
unframeStored @Text Text
line
      guard (snd (stamp stored) >= cursor)
      guard (deliversTo (stamped stored) names)
      pure stored

    filterStored :: Text -> Maybe (Stamped Text)
filterStored Text
line = do
      stored <- forall a. PostBody a => Text -> Maybe (Stamped a)
unframeStored @Text Text
line
      if deliversTo (stamped stored) names then Just stored else Nothing

    -- Wait for a halt or a new-posts signal, whichever lands first.
    awaitEvent :: TMVar a -> TVar Bool -> STM Bool
awaitEvent TMVar a
signal TVar Bool
halted =
      (TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar TVar Bool
halted STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
check STM () -> STM Bool -> STM Bool
forall a b. STM a -> STM b -> STM b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True)
        STM Bool -> STM Bool -> STM Bool
forall a. STM a -> STM a -> STM a
`orElse` (TMVar a -> STM a
forall a. TMVar a -> STM a
takeTMVar TMVar a
signal STM a -> STM Bool -> STM Bool
forall a b. STM a -> STM b -> STM b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False)

    quiesceLoop :: QuiesceConfig
-> TMVar a -> TVar Bool -> TVar Bool -> Int -> IO () -> IO ()
quiesceLoop QuiesceConfig
qc TMVar a
signal TVar Bool
busy TVar Bool
halted Int
count IO ()
onQuiesce = do
      m <- Int -> IO Bool -> IO (Maybe Bool)
forall a. Int -> IO a -> IO (Maybe a)
timeout (QuiesceConfig -> Int
qcCycleMicros QuiesceConfig
qc) (STM Bool -> IO Bool
forall a. STM a -> IO a
atomically (TMVar a -> TVar Bool -> STM Bool
forall {a}. TMVar a -> TVar Bool -> STM Bool
awaitEvent TMVar a
signal TVar Bool
halted))
      case m of
        Just Bool
True -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Just Bool
False -> QuiesceConfig
-> TMVar a -> TVar Bool -> TVar Bool -> Int -> IO () -> IO ()
quiesceLoop QuiesceConfig
qc TMVar a
signal TVar Bool
busy TVar Bool
halted Int
0 IO ()
onQuiesce
        Maybe Bool
Nothing -> do
          inProgress <- TVar Bool -> IO Bool
forall a. TVar a -> IO a
readTVarIO TVar Bool
busy
          if inProgress
            then quiesceLoop qc signal busy halted 0 onQuiesce
            else do
              let count' = Int
count Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1
              if count' >= qcCycles qc
                then onQuiesce
                else quiesceLoop qc signal busy halted count' onQuiesce