Gate CallService by handler flags on the Bus

src/HomeAssistant/Runtime.hs

@@ -21,6 +21,7 @@ import Data.Time (getCurrentTime, getCurrentTimeZone)
 import Data.Void (Void)
 import HomeAssistant.Controller (HASS, HASSEff (..))
 import HomeAssistant.Runtime.Bus
+import HomeAssistant.Runtime.Flags (loadFlags)
 import HomeAssistant.Runtime.Connection (writerAction, readerAction)
 import Network.Socket (withSocketsDo)
 import System.Environment (getEnv, lookupEnv)
@@ -102,10 +103,12 @@ defaultMain = withSocketsDo $ do
   System.Metrics.registerGcMetrics store
   appMetrics <- HomeAssistant.Runtime.Metrics.registerAppMetrics store
   rateLimitMetrics <- registerRateLimitMetrics store
-  withBus severity appMetrics $ \bus -> do
+  rootPath <- fromMaybe "/tmp/" <$> lookupEnv "HA_LIB_DIR"
+  let allNames = [n | Controller _ n _ <- controllers]
+  flags <- loadFlags rootPath allNames
+  withBus severity appMetrics flags $ \bus -> do
     token <- getEnv "HA_TOKEN"
     host <- getEnv "HA_HOST"
-    rootPath <- fromMaybe "/tmp/" <$> lookupEnv "HA_LIB_DIR"
     rrdPath <- fromMaybe "hass-controller.rrd" <$> lookupEnv "HA_RRD_PATH"
     rrdtool <- fromMaybe "rrdtool" <$> lookupEnv "HA_RRDTOOL"
     metricsPort <- lookupPort

src/HomeAssistant/Runtime/Bus.hs

@@ -31,7 +31,9 @@ import System.IO (stdout)
 import System.Metrics.Counter (inc)
 import Data.UUID (toText)
 import AFRP (Request(..), Event(..))
+import Control.Monad (when)
 import Control.Monad.IO.Class (MonadIO, liftIO)
+import HomeAssistant.Runtime.Flags (Flags, isEnabled)
 
 -- | Shared runtime state: inbound is a broadcast channel (controllers
 -- read from 'dupTChan' copies), outbound queues service calls for the
@@ -43,10 +45,11 @@ data Bus = Bus
   , busGen      :: CallIdGen
   , busLogEnv   :: LogEnv
   , busMetrics  :: AppMetrics
+  , busFlags    :: Flags
   }
 
-withBus :: Severity -> AppMetrics -> (Bus -> IO a) -> IO a
+withBus :: Severity -> AppMetrics -> Flags -> (Bus -> IO a) -> IO a
-withBus severity metrics callback = do
+withBus severity metrics flags callback = do
   handleScribe <- mkHandleScribe ColorIfTerminal stdout (permitItem severity) V2
   let makeLogEnv = registerScribe "stdout" handleScribe defaultScribeSettings =<< initLogEnv "hass-controller" "production"
   -- closeScribes will stop accepting new logs, flush existing ones and clean up resources
@@ -58,6 +61,7 @@ withBus severity metrics callback = do
           <*> mkCallIdGen 0
           <*> pure le
           <*> pure metrics
+          <*> pure flags
     callback bus
 
 recordInbound :: Bus -> IO ()
@@ -70,7 +74,8 @@ channelHassEval :: (MonadIO m, KatipContext m) => Bus -> HASSEff a -> m a
 channelHassEval bus = \case
   CallService req svc -> katipAddContext (sl "traceId" (toText (requestTraceId req))) $ do
     logFM DebugS (ls $ show svc)
-    liftIO $ atomically $ writeTChan (busOutbound bus) (req, svc)
+    enabled <- isEnabled (busFlags bus) (requestHandler req)
+    when enabled $ liftIO $ atomically $ writeTChan (busOutbound bus) (req, svc)
   Debug x -> logFM DebugS (ls $ show x)
   Trace req x -> katipAddContext (sl "traceId" (toText (requestTraceId req))) $
     logFM InfoS (ls $ show x)

test/BusSpec.hs

@@ -6,6 +6,7 @@ import AFRP (Request (..), Event (..))
 import Control.Concurrent.STM
   ( atomically
   , dupTChan
+  , isEmptyTChan
   , readTChan
   , writeTChan
   )
@@ -15,9 +16,11 @@ import Data.Time (UTCTime (..), utc)
 import Data.UUID (nil)
 import HomeAssistant.Controller (HASSEff (..), Service (..), Target(..))
 import HomeAssistant.Runtime.Bus
+import HomeAssistant.Runtime.Flags (loadFlags, setEnabled)
 import HomeAssistant.Runtime.Metrics (registerAppMetrics)
 import Katip (Namespace (Namespace), runKatipContextT, Severity (..))
 import qualified System.Metrics as Metrics
+import System.IO.Temp (withTempDirectory)
 import Test.Hspec
 
 spec :: Spec
@@ -40,6 +43,15 @@ spec = describe "Bus" $ do
     (_, svc') <- atomically $ readTChan (busOutbound bus)
     svc' `shouldBe` svc
 
+  it "channelHassEval drops CallService when the handler is disabled" $
+    withTestFlagsBus $ \bus -> do
+      setEnabled (busFlags bus) "test-handler" False
+      let req = Request (UTCTime (toEnum 0) 0) utc nil "test-handler"
+          svc = Service "light" "turn_on" Nothing [EntityId "light.test"]
+      runKatipContextT (busLogEnv bus) () (Namespace ["test"]) $
+        channelHassEval bus (CallService req svc)
+      atomically (isEmptyTChan (busOutbound bus)) `shouldReturn` True
+
 
   it "generates unique sequential call ids" $ do
     gen <- mkCallIdGen 0
@@ -51,16 +63,28 @@ spec = describe "Bus" $ do
     it "increments the trigger and service counters" $ do
       store <- Metrics.newStore
       m <- registerAppMetrics store
-      withBus InfoS m $ \bus -> do
+      withTempDirectory "/tmp" "bus-spec" $ \dir -> do
-        recordInbound bus
+        flags <- loadFlags dir []
-        recordInbound bus
+        withBus InfoS m flags $ \bus -> do
-        recordOutbound bus
+          recordInbound bus
-        sample <- Metrics.sampleAll store
+          recordInbound bus
-        HM.lookup "hass.trigger.in" sample `shouldBe` Just (Metrics.Counter 2)
+          recordOutbound bus
-        HM.lookup "hass.service.out" sample `shouldBe` Just (Metrics.Counter 1)
+          sample <- Metrics.sampleAll store
+          HM.lookup "hass.trigger.in" sample `shouldBe` Just (Metrics.Counter 2)
+          HM.lookup "hass.service.out" sample `shouldBe` Just (Metrics.Counter 1)
 
 withTestBus :: (Bus -> IO a) -> IO a
 withTestBus action = do
   store <- Metrics.newStore
   m <- registerAppMetrics store
-  withBus InfoS m action
+  withTempDirectory "/tmp" "bus-spec" $ \dir -> do
+    flags <- loadFlags dir []
+    withBus InfoS m flags action
+
+withTestFlagsBus :: (Bus -> IO a) -> IO a
+withTestFlagsBus action = do
+  store <- Metrics.newStore
+  m <- registerAppMetrics store
+  withTempDirectory "/tmp" "bus-spec" $ \dir -> do
+    flags <- loadFlags dir ["test-handler"]
+    withBus InfoS m flags action