{-# LANGUAGE AllowAmbiguousTypes #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications #-}

-- | The stamping office: a single-writer daemon for the bus.
--
-- The bus has two stamping regimes:
--
--   * Office off: writers self-stamp with 'postLocal' — the entangled case,
--     where concurrent writers may race for ids.
--   * Office on: writers send bare 'Post' lines to the daemon over a named
--     pipe (@root/bus.fifo@); the daemon stamps them in arrival order and
--     appends to @log.jsonl@. This serialisation is the bus's ⅋ discipline
--     made into a process.
--
-- The daemon is intentionally stateless: it assigns ids by counting lines in
-- the log under the same exclusive lock that 'postLocal' uses, so the two
-- regimes can coexist during a transition without colliding.
module Free.Agent.Bus.Daemon
  ( -- * Daemon
    runDaemon,

    -- * Client
    postViaDaemon,

    -- * Paths
    fifoPath,
    receiptPath,
  )
where

import Circuit.Agent (Name, Post (..), PostId, deliversTo, mkPost, sortNub)
import Circuit.Agent.Framing
  ( PostBody,
    Stamped,
    framePost,
    frameStored,
    parsePost,
    stamp,
    stamped,
    unframeStored,
    pattern Stamped,
  )
import Control.Concurrent (threadDelay)
import Control.Monad (forever, unless, when)
import Data.ByteString qualified as BS
import Data.Foldable (traverse_)
import Data.Maybe (mapMaybe)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.IO qualified as TIO
import Data.Time (UTCTime, getCurrentTime)
import Free.Agent.Bus (appendStoredPostsUnlocked)
import System.Directory (createDirectoryIfMissing, doesFileExist)
import System.FileLock (SharedExclusive (Exclusive), withFileLock)
import System.FilePath (takeDirectory, (<.>), (</>))
import System.IO
  ( BufferMode (LineBuffering),
    Handle,
    IOMode (AppendMode, ReadWriteMode),
    hFlush,
    hGetLine,
    hSetBuffering,
    stderr,
    withFile,
  )
import System.Posix.Files (createNamedPipe)
import Text.Printf (printf)

-- | Path to the named pipe used to submit bare posts to the daemon.
fifoPath :: FilePath -> FilePath
fifoPath :: String -> String
fifoPath String
root = String
root String -> String -> String
</> String
"bus.fifo"

-- | Path to the per-sender receipt file. The daemon overwrites this with the
-- latest stamped post from that sender; the client waits for a receipt that
-- matches its submission.
receiptPath :: FilePath -> Name -> FilePath
receiptPath :: String -> Name -> String
receiptPath String
root Name
name = String
root String -> String -> String
</> String -> String -> String
forall r. PrintfType r => String -> r
printf String
".receipt-%s" (Name -> String
T.unpack Name
name)

-- | Run the stamping office forever.
--
-- Creates the bus directory, touches @log.jsonl@, creates the FIFO if absent,
-- then reads one bare 'Post' line at a time. Malformed lines are reported on
-- stderr and skipped; valid lines are stamped and appended.
runDaemon :: forall a. (PostBody a) => FilePath -> IO ()
runDaemon :: forall a. PostBody a => String -> IO ()
runDaemon String
root = do
  Bool -> String -> IO ()
createDirectoryIfMissing Bool
True String
root
  let path :: String
path = String
root String -> String -> String
</> String
"log.jsonl"
  exists <- String -> IO Bool
doesFileExist String
path
  unless exists $ withFile path AppendMode (\Handle
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
  let fifo = String -> String
fifoPath String
root
  fifoExists <- doesFileExist fifo
  unless fifoExists $ createNamedPipe fifo 0o600
  withFile fifo ReadWriteMode $ \Handle
h -> do
    Handle -> BufferMode -> IO ()
hSetBuffering Handle
h BufferMode
LineBuffering
    IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      line <- String -> Name
T.pack (String -> Name) -> IO String -> IO Name
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Handle -> IO String
hGetLine Handle
h
      case parsePost @a line of
        Maybe (Post a)
Nothing -> Handle -> Name -> IO ()
TIO.hPutStrLn Handle
stderr (Name
"🔴 daemon: invalid post JSON: " Name -> Name -> Name
forall a. Semigroup a => a -> a -> a
<> Name
line)
        Just Post a
p -> do
          stored <- String -> Post a -> IO (Stamped a)
forall a. PostBody a => String -> Post a -> IO (Stamped a)
stampOne String
path Post a
p
          TIO.writeFile (receiptPath root (from p)) (frameStored stored)
          writePings path [stored]

-- | Stamp a single post and append it to the log.
--
-- This is the same protocol as 'postLocal' in "Free.Agent.Bus": ids are the
-- current line count read under the exclusive file lock.
stampOne :: (PostBody a) => FilePath -> Post a -> IO (Stamped a)
stampOne :: forall a. PostBody a => String -> Post a -> IO (Stamped a)
stampOne String
path Post a
p = do
  ts <- IO UTCTime
getCurrentTime
  withFileLock (path <.> "lock") Exclusive $ \FileLock
_lock -> do
    n <- Word8 -> ByteString -> Int
BS.count Word8
0x0A (ByteString -> Int) -> IO ByteString -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> String -> IO ByteString
BS.readFile String
path
    let stored = (UTCTime, PostId) -> Post a -> Stamped a
forall a. (UTCTime, PostId) -> Post a -> Stamped a
Stamped (UTCTime
ts, Int -> PostId
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n) Post a
p
    appendStoredPostsUnlocked path [stored]
    pure stored

-- | Submit a bare post to the daemon and wait for the receipt.
postViaDaemon :: FilePath -> Post Text -> IO (Stamped Text)
postViaDaemon :: String -> Post Name -> IO (Stamped Name)
postViaDaemon String
root Post Name
p = do
  let fifo :: String
fifo = String -> String
fifoPath String
root
  fifoExists <- String -> IO Bool
doesFileExist String
fifo
  unless fifoExists $
    fail ("🔴 no bus.fifo at " <> fifo <> "; is the daemon running?")
  withFile fifo AppendMode $ \Handle
h -> do
    Handle -> BufferMode -> IO ()
hSetBuffering Handle
h BufferMode
LineBuffering
    Handle -> Name -> IO ()
TIO.hPutStrLn Handle
h (Post Name -> Name
forall a. PostBody a => Post a -> Name
framePost Post Name
p)
    Handle -> IO ()
hFlush Handle
h
  waitReceipt root p

-- | Poll the receipt file until it contains a stamped post matching the
-- submitted post, or a timeout expires.
waitReceipt :: FilePath -> Post Text -> IO (Stamped Text)
waitReceipt :: String -> Post Name -> IO (Stamped Name)
waitReceipt String
root Post Name
p = Int -> IO (Stamped Name)
go (Int
200 :: Int)
  where
    go :: Int -> IO (Stamped Name)
go Int
0 = String -> IO (Stamped Name)
forall a. HasCallStack => String -> IO a
forall (m :: * -> *) a.
(MonadFail m, HasCallStack) =>
String -> m a
fail String
"🔴 daemon receipt timeout"
    go Int
n = do
      let path :: String
path = String -> Name -> String
receiptPath String
root (Post Name -> Name
forall a. Post a -> Name
from Post Name
p)
      exists <- String -> IO Bool
doesFileExist String
path
      if not exists
        then delay >> go (n - 1)
        else do
          line <- TIO.readFile path
          case unframeStored @Text line of
            Just Stamped Name
stored
              | Stamped Name -> Bool
matches Stamped Name
stored -> Stamped Name -> IO (Stamped Name)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Stamped Name
stored
              | Bool
otherwise -> IO ()
delay IO () -> IO (Stamped Name) -> IO (Stamped Name)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Int -> IO (Stamped Name)
go (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1)
            Maybe (Stamped Name)
Nothing -> IO ()
delay IO () -> IO (Stamped Name) -> IO (Stamped Name)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Int -> IO (Stamped Name)
go (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1)
    delay :: IO ()
delay = Int -> IO ()
threadDelay Int
50_000
    matches :: Stamped Name -> Bool
matches Stamped Name
stored =
      let q :: Post Name
q = Stamped Name -> Post Name
forall r a. Stamped r a -> a
stamped Stamped Name
stored
       in Post Name -> Name
forall a. Post a -> Name
from Post Name
q Name -> Name -> Bool
forall a. Eq a => a -> a -> Bool
== Post Name -> Name
forall a. Post a -> Name
from Post Name
p
            Bool -> Bool -> Bool
&& Post Name -> [Name]
forall a. Post a -> [Name]
to Post Name
q [Name] -> [Name] -> Bool
forall a. Eq a => a -> a -> Bool
== Post Name -> [Name]
forall a. Post a -> [Name]
to Post Name
p
            Bool -> Bool -> Bool
&& Post Name -> [PostId]
forall a. Post a -> [PostId]
thread Post Name
q [PostId] -> [PostId] -> Bool
forall a. Eq a => a -> a -> Bool
== Post Name -> [PostId]
forall a. Post a -> [PostId]
thread Post Name
p
            Bool -> Bool -> Bool
&& Post Name -> Name
forall a. Post a -> a
body Post Name
q Name -> Name -> Bool
forall a. Eq a => a -> a -> Bool
== Post Name -> Name
forall a. Post a -> a
body Post Name
p

-- | Write a @.ping-NAME@ file for each named recipient of each post.
--
-- Copied from "Free.Agent.Bus" because the original helper is internal to
-- that module.
writePings :: FilePath -> [Stamped a] -> IO ()
writePings :: forall a. String -> [Stamped a] -> IO ()
writePings String
path [Stamped a]
posts = ((Name, PostId) -> IO ()) -> [(Name, PostId)] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Name, PostId) -> IO ()
writeOne [(Name, PostId)]
recipients
  where
    root :: String
root = String -> String
takeDirectory String
path
    pingFile :: Name -> String
pingFile Name
name = String
root String -> String -> String
</> String -> String -> String
forall r. PrintfType r => String -> r
printf String
".ping-%s" (Name -> String
T.unpack Name
name)
    recipients :: [(Name, PostId)]
recipients = (Stamped a -> [(Name, PostId)]) -> [Stamped a] -> [(Name, PostId)]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap Stamped a -> [(Name, PostId)]
forall {a} {b} {a}. Stamped (a, b) (Post a) -> [(Name, b)]
postRecipients [Stamped a]
posts
    postRecipients :: Stamped (a, b) (Post a) -> [(Name, b)]
postRecipients Stamped (a, b) (Post a)
stored =
      case Post a -> [Name]
forall a. Post a -> [Name]
to (Stamped (a, b) (Post a) -> Post a
forall r a. Stamped r a -> a
stamped Stamped (a, b) (Post a)
stored) of
        [] -> []
        [Name
""] -> []
        [Name]
ts -> [(Name
name, (a, b) -> b
forall a b. (a, b) -> b
snd (Stamped (a, b) (Post a) -> (a, b)
forall r a. Stamped r a -> r
stamp Stamped (a, b) (Post a)
stored)) | Name
name <- [Name]
ts]
    writeOne :: (Name, PostId) -> IO ()
writeOne (Name
name, PostId
pid) =
      String -> Name -> IO ()
TIO.writeFile (Name -> String
pingFile Name
name) (String -> Name
T.pack (PostId -> String
forall a. Show a => a -> String
show PostId
pid))