home-assistant-controller/src/HomeAssistant/Runtime.hs

{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}

module HomeAssistant.Runtime
  ( defaultMain,
    step,
    CallIdGen,
    mkCallIdGen,
    dryRunHassEval,
    Controller (..),
    runController,
  )
where

import AFRP (Auto, DecodedAuto (..), Event (..), Mealy (..), Request (..), load, stepAutoSerializing)
import Control.Concurrent.STM (atomically, dupTChan, readTChan)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Aeson (Value)
import Data.Foldable (forM_)
import Data.Maybe (fromMaybe)
import Data.Text qualified as T
import Data.Time (getCurrentTime, getCurrentTimeZone)
import Data.UUID (UUID, toText)
import Data.UUID.V4 qualified as UUID.V4
import Data.Void (Void)
import GHC.Stack (HasCallStack)
import HomeAssistant.Controller (HASS, HASSEff (..))
import HomeAssistant.Controller.Bedroom (bedroomButtonController, bedroomDrawerController, bedroomPresenceController, humidifierController)
import HomeAssistant.Controller.Children (childrenBedroomButtonController, schoolLightController, childrenAirFilterController)
import HomeAssistant.Controller.Hallway (hallwayLightsController)
import HomeAssistant.Controller.Kitchen (kitchenMotionController)
import HomeAssistant.Controller.Livingroom (livingroomPlantLights, livingroomPresenceController)
import HomeAssistant.Runtime.Bus
import HomeAssistant.Runtime.Connection (readerAction, writerAction)
import HomeAssistant.Runtime.Flags (loadFlags)
import HomeAssistant.Runtime.Metrics qualified
import HomeAssistant.Runtime.RateLimit (registerRateLimitMetrics, slidingWindowLimiter)
import HomeAssistant.Runtime.Supervisor (supervised)
import HttpServer qualified
import Katip (KatipContext, Namespace (Namespace), Severity (..), logF, logLocM, ls, runKatipContextT, runKatipT, sl, katipAddNamespace)
import Network.Socket (withSocketsDo)
import System.Environment (getEnv, lookupEnv)
import System.FilePath ((</>))
import System.Metrics qualified
import Text.Read (readMaybe)
import UnliftIO.Async
import Control.Monad.Catch (MonadCatch)
import Control.Exception.Annotated (checkpoint, Annotation (..))
import AFRP.Diagram (renderF)

step :: (MonadIO m) => FilePath -> T.Text -> UUID -> Auto m a b -> a -> m (b, Auto m a b)
step path name trace st a = do
  now <- liftIO getCurrentTime
  tz <- liftIO getCurrentTimeZone
  let req = Request now tz trace name
  stepAutoSerializing path st req a

data Controller = forall b. Controller T.Text (HASS (Event Value) b)

controllers :: [Controller]
controllers =
  [ Controller "bedroom-presence" bedroomPresenceController,
    Controller "bedroom-button" bedroomButtonController,
    Controller "bedroom-drawer" bedroomDrawerController,
    Controller "bedroom-humidifier" humidifierController,
    Controller "school-light-controller" schoolLightController,
    Controller "kitchen-motion-controller" kitchenMotionController,
    Controller "livingroom-presence" livingroomPresenceController,
    Controller "hallway-motion-controller" hallwayLightsController,
    Controller "children-button" childrenBedroomButtonController,
    Controller "livingroom-plants" livingroomPlantLights,
    Controller "children-filter" childrenAirFilterController
  ]

-- | Steps the machine for every inbound message; service calls go to the
-- bus. A restart re-dups the inbound channel and starts from the machine's
-- initial state; messages broadcast during the restart window are lost.
runController :: forall m. (HasCallStack, MonadCatch m, KatipContext m, MonadIO m) => FilePath -> Bus -> Controller -> m Void
runController rootDir bus (Controller name machine) = checkpoint (Annotation name) $ katipAddNamespace ns $ do
  inbound <- liftIO $ atomically (dupTChan (busInbound bus))
  let workerDefinition = runMealy machine (channelHassEval @m bus)
  let path = rootDir </> T.unpack name
  worker <-
    load path workerDefinition >>= \case
      Decoded a -> pure a
      FailDecode err a -> a <$ logLocM WarningS (ls ("Failed to load (" <> T.unpack name <> "): " <> err))
  go path inbound worker
  where
    ns = Namespace [name]
    go path inbound f = do
      msg <- liftIO $ atomically (readTChan inbound)
      uuid <-  liftIO UUID.V4.nextRandom
      (_, next) <- step path name uuid f msg
      go path inbound next

lookupPort :: (MonadFail m, MonadIO m) => m Int
lookupPort = do
  m <- liftIO (lookupEnv "HA_METRICS_PORT")
  case m of
    Nothing -> pure 8124
    Just s -> case readMaybe s of
      Just p | p >= 1 && p <= 65535 -> pure p
      _ -> fail ("HA_METRICS_PORT must be a port in 1..65535: " <> s)

defaultMain :: IO ()
defaultMain =
  withSocketsDo $
    getSeverity >>= \s -> withLogEnv s $ \le -> runKatipContextT le () mempty $ do
      logLocM InfoS "Hello world"
      store <- liftIO System.Metrics.newStore
      liftIO $ System.Metrics.registerGcMetrics store
      appMetrics <- liftIO $ HomeAssistant.Runtime.Metrics.registerAppMetrics store
      rateLimitMetrics <- registerRateLimitMetrics store
      rootPath <- fromMaybe "/tmp/" <$> liftIO (lookupEnv "HA_LIB_DIR")
      let allNames = map (\(Controller n _) -> n) controllers
      flags <- loadFlags rootPath allNames
      withBus le appMetrics flags $ \bus -> do
        token <- liftIO $ getEnv "HA_TOKEN"
        host <- liftIO $ getEnv "HA_HOST"
        let rrdPath = rootPath </> "hass-controller.rrd"
        rrdtool <- fromMaybe "rrdtool" <$> liftIO (lookupEnv "HA_RRDTOOL")
        metricsPort <- lookupPort
        writerLimiter <- slidingWindowLimiter rateLimitMetrics 10 20
        forM_ controllers $ \(Controller n m) ->
          logLocM InfoS $ ls $ n <> ": " <> fromMaybe "" (runDiagramC m renderF)
        let ents = foldMap (\(Controller _ m) -> entities m) controllers
            workers =
              [ ("reader", readerAction host 8123 token ents bus),
                ("writer", writerAction writerLimiter bus),
                ("metrics", HomeAssistant.Runtime.Metrics.metricsAction store rrdPath rrdtool),
                ("metrics-http", HttpServer.runHttpServer (busLogEnv bus) rrdPath rrdtool (busFlags bus) metricsPort)
              ]
                ++ [(name, runController rootPath bus c) | c@(Controller name _) <- controllers]
        runKatipContextT (busLogEnv bus) () mempty $
          mapConcurrently_ (uncurry supervised) workers
  where
    getSeverity = maybe InfoS (const DebugS) <$> lookupEnv "HA_DEBUG"

dryRunHassEval :: Namespace -> Bus -> HASSEff a -> IO a
dryRunHassEval ns bus = \case
  CallService req x -> runKatipT (busLogEnv bus) $ do
    callId <- liftIO $ generateCallId (busGen bus)
    logF (sl "traceId" (toText (requestTraceId req))) ns DebugS (ls $ show (callId, x))
  CallServices req xs -> runKatipT (busLogEnv bus) $ forM_ xs $ \x -> do
    callId <- liftIO $ generateCallId (busGen bus)
    logF (sl "traceId" (toText (requestTraceId req))) ns DebugS (ls $ show (callId, x))
  Debug x -> runKatipT (busLogEnv bus) $ do
    logF () ns DebugS (ls $ show x)
  Trace req x -> runKatipT (busLogEnv bus) $ do
    logF (sl "traceId" (toText (requestTraceId req))) ns InfoS (ls $ show x)