Add Runtime.Connection reader and writer actions

src/HomeAssistant/Runtime/Connection.hs

@@ -0,0 +1,97 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
+
+module HomeAssistant.Runtime.Connection
+  ( readerAction
+  , writerAction
+  , encodeService
+  ) where
+
+import Control.Concurrent.STM
+  ( atomically
+  , readTChan
+  , readTVar
+  , retry
+  , writeTChan
+  , writeTVar
+  )
+import Control.Lens ((^?))
+import Control.Monad (forever)
+import Data.Aeson (Value, eitherDecode, encode, object, (.=))
+import Data.Aeson.Lens (key, _String)
+import qualified Data.ByteString.Lazy as BL
+import qualified Data.Text as T
+import Data.Void (Void)
+import HomeAssistant.Controller (Service (..))
+import HomeAssistant.Runtime.Bus
+import qualified Network.WebSockets as WS
+
+-- | Connect, authenticate, subscribe, then receive and broadcast forever.
+-- Restarting this action reconnects. All setup sends happen before the
+-- connection is published in the bus, so only the writer sends afterwards.
+readerAction :: String -> Int -> String -> Bus -> IO Void
+readerAction host port token bus =
+  WS.runClient host port "/api/websocket" $ \conn -> do
+    handshake conn token
+    subscribe bus conn
+    atomically $ writeTVar (busConn bus) (Just conn)
+    putStrLn "[reader] connected"
+    receiveLoop bus conn
+
+handshake :: WS.Connection -> String -> IO ()
+handshake conn token = do
+  required <- receiveJSON conn
+  expectType "auth_required" required
+  WS.sendTextData conn $ encode $ object
+    [ "type"         .= ("auth" :: T.Text)
+    , "access_token" .= token
+    ]
+  ok <- receiveJSON conn
+  expectType "auth_ok" ok
+
+expectType :: T.Text -> Value -> IO ()
+expectType expected msg =
+  case msg ^? key "type" . _String of
+    Just t | t == expected -> pure ()
+    _ -> fail $ "expected " <> T.unpack expected <> ", got: " <> show msg
+
+subscribe :: Bus -> WS.Connection -> IO ()
+subscribe bus conn = do
+  sid <- generateCallId (busGen bus)
+  WS.sendTextData conn $ encode $ object
+    [ "id"         .= sid
+    , "type"       .= ("subscribe_events" :: T.Text)
+    , "event_type" .= ("state_changed" :: T.Text)
+    ]
+
+-- | Undecodable messages are skipped: reconnecting cannot fix a decode
+-- problem, so crashing here would only produce a hot restart loop.
+receiveLoop :: Bus -> WS.Connection -> IO Void
+receiveLoop bus conn = forever $ do
+  msg <- WS.receiveData conn :: IO BL.ByteString
+  case eitherDecode msg of
+    Left err -> putStrLn $ "[reader] skipping undecodable message: " <> err
+    Right v  -> atomically $ writeTChan (busInbound bus) v
+
+receiveJSON :: WS.Connection -> IO Value
+receiveJSON conn = do
+  msg <- WS.receiveData conn
+  case eitherDecode msg of
+    Left err -> fail $ "Invalid JSON from Home Assistant: " ++ err
+    Right x  -> pure x
+
+writerAction :: Bus -> IO Void
+writerAction bus = forever $ do
+  svc <- atomically $ readTChan (busOutbound bus)
+  conn <- atomically $ readTVar (busConn bus) >>= maybe retry pure
+  callId <- generateCallId (busGen bus)
+  WS.sendTextData conn $ encode $ encodeService callId svc
+
+encodeService :: Int -> Service -> Value
+encodeService callId Service{..} = object $
+  [ "id"      .= callId
+  , "type"    .= ("call_service" :: T.Text)
+  , "domain"  .= serviceDomain
+  , "service" .= serviceName
+  , "target"  .= object ["entity_id" .= serviceTarget]
+  ] <> maybe [] (\d -> ["service_data" .= d]) serviceData