{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} module HomeAssistant.Runtime.Bus ( Bus(..) , CallIdGen(..) , mkCallIdGen , withBus , recordInbound , recordOutbound , channelHassEval , withLogEnv ) where import Control.Concurrent.STM ( TChan , TVar , atomically , newBroadcastTChanIO , newTChanIO , newTVarIO , writeTChan ) import Data.Aeson (Value) import Data.IORef (atomicModifyIORef', newIORef) import HomeAssistant.Controller (HASSEff (..), Service) import HomeAssistant.Runtime.Metrics (AppMetrics (..)) import Network.WebSockets (Connection) import Katip (LogEnv, closeScribes, mkHandleScribe, ColorStrategy (..), permitItem, Severity (..), Verbosity (V2), registerScribe, defaultScribeSettings, initLogEnv, ls, sl, logFM, katipAddContext, KatipContext) import Control.Exception (bracket) import System.IO (stdout) import System.Metrics.Counter (inc) import Data.UUID (toText) import AFRP (Request(..), Event(..)) import Control.Monad.IO.Class (MonadIO, liftIO) import HomeAssistant.Runtime.Flags (Flags) import Katip.Scribes.Journal (mkJournalScribe) import System.Environment (lookupEnv) import Data.Maybe (isJust) -- | Shared runtime state: inbound is a broadcast channel (controllers -- read from 'dupTChan' copies), outbound queues service calls for the -- writer, conn holds the current websocket (Nothing before first connect). data Bus = Bus { busInbound :: TChan (Event Value) , busOutbound :: TChan (Request, Service) , busConn :: TVar (Maybe Connection) , busGen :: CallIdGen , busLogEnv :: LogEnv , busMetrics :: AppMetrics , busFlags :: Flags } withLogEnv :: Severity -> (LogEnv -> IO a) -> IO a withLogEnv severity f = do let makeLogEnv = registerJournalScribe =<< registerHandleScribe =<< initLogEnv "hass-controller" "production" -- closeScribes will stop accepting new logs, flush existing ones and clean up resources bracket makeLogEnv closeScribes $ \le -> f le where registerJournalScribe :: LogEnv -> IO LogEnv registerJournalScribe le = do journalScribe <- mkJournalScribe (permitItem severity) V2 registerScribe "journalctl" journalScribe defaultScribeSettings le registerHandleScribe le = do systemdUnit <- isJust <$> lookupEnv "HA_CONTROLLER_SYSTEMD" if systemdUnit then pure le -- ignore stdout when running in systemd else do handleScribe <- mkHandleScribe ColorIfTerminal stdout (permitItem severity) V2 registerScribe "stdout" handleScribe defaultScribeSettings le withBus :: MonadIO m => LogEnv -> AppMetrics -> Flags -> (Bus -> m a) -> m a withBus le metrics flags callback = do bus <- Bus <$> liftIO newBroadcastTChanIO <*> liftIO newTChanIO <*> liftIO (newTVarIO Nothing) <*> liftIO (mkCallIdGen 0) <*> pure le <*> pure metrics <*> pure flags callback bus recordInbound :: Bus -> IO () recordInbound bus = inc (amTriggersIn (busMetrics bus)) recordOutbound :: Bus -> IO () recordOutbound bus = inc (amServicesOut (busMetrics bus)) channelHassEval :: forall m a. (MonadIO m, KatipContext m) => Bus -> HASSEff a -> m a channelHassEval bus = \case CallService req svc -> emit req svc CallServices req svcs -> mapM_ (emit req) svcs Debug x -> logFM DebugS (ls $ show x) Trace req x -> katipAddContext (sl "traceId" (toText (requestTraceId req))) $ logFM InfoS (ls $ show x) where emit :: Request -> Service -> m () emit req svc = liftIO $ atomically $ writeTChan (busOutbound bus) (req, svc) newtype CallIdGen = CallIdGen { generateCallId :: IO Int } mkCallIdGen :: Int -> IO CallIdGen mkCallIdGen start = do gen <- newIORef start pure $ CallIdGen $ atomicModifyIORef' gen (\old -> let new = old + 1 in new `seq` (new, new))