client/agent transmission types, parse NEW/INV commands

design/agent2.gv

@@ -71,12 +71,6 @@ digraph SMPAgent {
       sOutq [shape=rarrow label="srv send\nTBQueue"]
       sSock [label="SMP client connection TCP socket" color=blue]
 
-      subgraph clusterQueue {
-        graph [fontsize=11]
-        label="1 per SMP queue\n(+ 1 TBQueue\nw/out SMP qID)"
-        sCmdq [shape=rarrow label="commands\nTBQueue"]
-      }
-
       subgraph clusterServerThreads {
         graph [fontsize=11]
         label="SMP client threads"
@@ -89,7 +83,7 @@ digraph SMPAgent {
       {uAgent uProcess} -> sOutq -> sSnd -> sSock -> sAgent [color=blue]
     }
 
-    userState [label="connected\nservers,\nsubscribed\nqueues\n(STM)" shape="folder"]
+    userState [label="connected\nservers,\nsubscribed\nqueues,\nsent\ncommands\n(STM)" shape="folder"]
 
     {uAgent uProcess} -> userState [dir=both color="#FF8888"]
     {uAgent uProcess} -> connectionsStore [dir=both color="#880000"]
@@ -98,8 +92,6 @@ digraph SMPAgent {
     connectClnt -> {uRcv uSnd} [style=dashed label="race" color=orange fontcolor=orange]
     runClnt -> {uAgent uProcess} [style=dashed label="race" color=orange fontcolor=orange]
 
-    sSnd -> sCmdq
-    sCmdq -> sAgent [style=dashed]
     sAgent -> uRespq -> uProcess
 
     uAgent -> runClient [style=dashed label="fork" color=orange fontcolor=orange]

package.yaml

@@ -20,6 +20,7 @@ dependencies:
   - iso8601-time == 0.1.*
   - mtl
   - network == 3.1.*
+  - split == 0.2.*
   - sqlite-simple == 0.4.*
   - stm
   - template-haskell == 2.15.*

src/Simplex/Messaging/Agent.hs

@@ -9,8 +9,10 @@ module Simplex.Messaging.Agent (runSMPAgent) where
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
 import Crypto.Random
+import qualified Data.ByteString.Char8 as B
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Transmission
+import Simplex.Messaging.Server.Transmission (CorrId (..))
 import Simplex.Messaging.Transport
 import UnliftIO.Async
 import UnliftIO.IO
@@ -35,23 +37,22 @@ runClient :: MonadUnliftIO m => AgentClient -> m ()
 runClient c = race_ (respond c) (process c)
 
 receive :: MonadUnliftIO m => Handle -> AgentClient -> m ()
-receive h AgentClient {rcvQ, sndQ} = forever $ do
-  aCmdGet SClient h >>= \case
-    Right cmd -> atomically $ writeTBQueue rcvQ cmd
-    Left e -> atomically $ writeTBQueue sndQ $ ERR e
+receive h AgentClient {rcvQ, sndQ} =
+  forever $
+    tGet SClient h >>= \(corrId, cAlias, command) -> atomically $ case command of
+      Right cmd -> writeTBQueue rcvQ (corrId, cAlias, cmd)
+      Left e -> writeTBQueue sndQ (corrId, cAlias, ERR e)
 
 send :: MonadUnliftIO m => Handle -> AgentClient -> m ()
-send h AgentClient {sndQ} = forever $ do
-  cmd <- atomically $ readTBQueue sndQ
-  putLn h $ serializeCommand cmd
+send h AgentClient {sndQ} = forever $ atomically (readTBQueue sndQ) >>= tPut h
 
 process :: MonadUnliftIO m => AgentClient -> m ()
 process AgentClient {rcvQ, respQ} = forever $ do
-  cmd <- atomically (readTBQueue rcvQ)
+  (corrId, cAlias, cmd) <- atomically (readTBQueue rcvQ)
   liftIO $ print cmd
   atomically $ writeTBQueue respQ ()
 
 respond :: MonadUnliftIO m => AgentClient -> m ()
 respond AgentClient {respQ, sndQ} = forever . atomically $ do
   readTBQueue respQ
-  writeTBQueue sndQ $ ERR UNKNOWN
+  writeTBQueue sndQ (CorrId B.empty, B.empty, ERR UNKNOWN)

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -31,8 +31,8 @@ data Env = Env
   }
 
 data AgentClient = AgentClient
-  { rcvQ :: TBQueue (ACommand Client),
-    sndQ :: TBQueue (ACommand Agent),
+  { rcvQ :: TBQueue (ATransmission Client),
+    sndQ :: TBQueue (ATransmission Agent),
     respQ :: TBQueue (),
     servers :: Map (HostName, ServiceName) ServerClient
   }

src/Simplex/Messaging/Agent/Transmission.hs

@@ -1,6 +1,8 @@
 {-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
@@ -9,14 +11,19 @@
 
 module Simplex.Messaging.Agent.Transmission where
 
+import Control.Monad
 import Control.Monad.IO.Class
+-- import Numeric.Natural
+import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Kind
+import Data.List.Split (splitOn)
 import Data.Time.Clock (UTCTime)
+import Data.Type.Equality
+import Data.Typeable ()
 import Network.Socket
--- import Numeric.Natural
-import Simplex.Messaging.Server.Transmission (CorrId (..), Encoded, MsgBody, PublicKey, QueueId, errMessageBody)
+import Simplex.Messaging.Server.Transmission (CorrId (..), Encoded, MsgBody, PublicKey, QueueId, errBadParameters, errMessageBody)
 import Simplex.Messaging.Transport
 import System.IO
 import Text.Read
@@ -38,6 +45,11 @@ deriving instance Show (SAParty p)
 
 deriving instance Eq (SAParty p)
 
+instance TestEquality SAParty where
+  testEquality SAgent SAgent = Just Refl
+  testEquality SClient SClient = Just Refl
+  testEquality _ _ = Nothing
+
 data ACmd where
   ACmd :: SAParty p -> ACommand p -> ACmd
 
@@ -72,9 +84,14 @@ data AMessage where
   A_ACK :: AgentMsgId -> AckStatus -> AMessage
   A_DEL :: AMessage
 
-data SMPServer = SMPServer HostName ServiceName KeyFingerprint deriving (Show)
+data SMPServer = SMPServer
+  { host :: HostName,
+    port :: Maybe ServiceName,
+    keyHash :: Maybe KeyHash
+  }
+  deriving (Show)
 
-type KeyFingerprint = Encoded
+type KeyHash = Encoded
 
 type ConnAlias = ByteString
 
@@ -115,8 +132,57 @@ data AckStatus = AckOk | AckError AckErrorType
 data AckErrorType = AckUnknown | AckProhibited | AckSyntax Int -- etc.
   deriving (Show)
 
+errBadInvitation :: Int
+errBadInvitation = 10
+
+errNoConnAlias :: Int
+errNoConnAlias = 11
+
 parseCommand :: ByteString -> Either ErrorType ACmd
-parseCommand _ = Left UNKNOWN
+parseCommand command = case B.words command of
+  ["NEW", srv] -> newConn srv . Right $ AckMode On
+  ["NEW", srv, am] -> newConn srv $ ackMode am
+  ["INV", qInfo] -> ACmd SAgent . INV <$> smpQueueInfo qInfo
+  "NEW" : _ -> errParams
+  "INV" : _ -> errParams
+  _ -> Left UNKNOWN
+  where
+    newConn :: ByteString -> Either ErrorType AckMode -> Either ErrorType ACmd
+    newConn srv am = ACmd SClient <$> liftM2 NEW (smpServer srv) am
+
+    smpServer :: ByteString -> Either ErrorType SMPServer
+    smpServer srv =
+      let (s, kf) = span (/= '#') $ B.unpack srv
+          (h, p) = span (/= ':') s
+       in SMPServer h (srvPart p) <$> traverse dec64 (srvPart kf)
+
+    smpQueueInfo :: ByteString -> Either ErrorType SMPQueueInfo
+    smpQueueInfo qInfo = case splitOn "::" $ B.unpack qInfo of
+      ["smp", srv, qId, ek] -> liftM3 SMPQueueInfo (smpServer $ B.pack srv) (dec64 qId) (dec64 ek)
+      _ -> errInv
+
+    dec64 :: String -> Either ErrorType ByteString
+    dec64 s = either (const errInv) Right . decode $ B.pack s
+
+    srvPart :: String -> Maybe String
+    srvPart s = if length s > 1 then Just $ tail s else Nothing
+
+    ackMode :: ByteString -> Either ErrorType AckMode
+    ackMode am = case B.split '=' am of
+      ["ACK", mode] -> AckMode <$> getMode mode
+      _ -> errParams
+
+    getMode :: ByteString -> Either ErrorType Mode
+    getMode mode = case mode of
+      "ON" -> Right On
+      "OFF" -> Right Off
+      _ -> errParams
+
+    errParams :: Either ErrorType a
+    errParams = Left $ SYNTAX errBadParameters
+
+    errInv :: Either ErrorType a
+    errInv = Left $ SYNTAX errBadInvitation
 
 serializeCommand :: ACommand p -> ByteString
 serializeCommand = B.pack . show
@@ -137,70 +203,45 @@ tGetRaw h = do
 tPut :: MonadIO m => Handle -> ATransmission p -> m ()
 tPut h (corrId, connAlias, command) = tPutRaw h (bs corrId, connAlias, serializeCommand command)
 
-fromClient :: ACmd -> Either ErrorType ACmd
-fromClient = \case
-  ACmd SAgent _ -> Left PROHIBITED
-  cmd -> Right cmd
-
-fromAgent :: ACmd -> Either ErrorType ACmd
-fromAgent = \case
-  ACmd SClient _ -> Left PROHIBITED
-  cmd -> Right cmd
-
 -- | get client and agent transmissions
--- tGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (ATransmissionOrError p)
--- tGet fromParty h = tGetRaw h >>= tParseLoadBody
---   where
---     tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p)
---     tParseLoadBody t@(corrId, connAlias, command) = do
---       let cmd = parseCommand command >>= fromParty -- >>= tCredentials t
---       fullCmd <- either (return . Left) cmdWithMsgBody cmd
---       return (CorrId corrId, connAlias, fullCmd)
+tGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (ATransmissionOrError p)
+tGet party h = tGetRaw h >>= tParseLoadBody
+  where
+    tParseLoadBody :: ARawTransmission -> m (ATransmissionOrError p)
+    tParseLoadBody t@(corrId, connAlias, command) = do
+      let cmd = parseCommand command >>= fromParty >>= tConnAlias t
+      fullCmd <- either (return . Left) cmdWithMsgBody cmd
+      return (CorrId corrId, connAlias, fullCmd)
 
---     fromParty :: ACmd -> Either ErrorType (ACommand p)
---     fromParty (ACmd p cmd)
---       | p == party = Right cmd
---       | otherwise = Left PROHIBITED
+    fromParty :: ACmd -> Either ErrorType (ACommand p)
+    fromParty (ACmd (p :: p1) cmd) = case testEquality party p of
+      Just Refl -> Right cmd
+      _ -> Left PROHIBITED
 
---     tCredentials :: RawTransmission -> Cmd -> Either ErrorType Cmd
---     tCredentials (signature, _, queueId, _) cmd = case cmd of
---       -- IDS response should not have queue ID
---       Cmd SBroker (IDS _ _) -> Right cmd
---       -- ERROR response does not always have queue ID
---       Cmd SBroker (ERR _) -> Right cmd
---       -- other responses must have queue ID
---       Cmd SBroker _
---         | B.null queueId -> Left $ SYNTAX errNoConnectionId
---         | otherwise -> Right cmd
---       -- CREATE must NOT have signature or queue ID
---       Cmd SRecipient (NEW _)
---         | B.null signature && B.null queueId -> Right cmd
---         | otherwise -> Left $ SYNTAX errHasCredentials
---       -- SEND must have queue ID, signature is not always required
---       Cmd SSender (SEND _)
---         | B.null queueId -> Left $ SYNTAX errNoConnectionId
---         | otherwise -> Right cmd
---       -- other client commands must have both signature and queue ID
---       Cmd SRecipient _
---         | B.null signature || B.null queueId -> Left $ SYNTAX errNoCredentials
---         | otherwise -> Right cmd
+    tConnAlias :: ARawTransmission -> ACommand p -> Either ErrorType (ACommand p)
+    tConnAlias (_, connAlias, _) cmd = case cmd of
+      -- NEW has optional connAlias
+      NEW _ _ -> Right cmd
+      -- ERROR response does not always have connAlias
+      ERR _ -> Right cmd
+      -- other responses must have connAlias
+      _
+        | B.null connAlias -> Left $ SYNTAX errNoConnAlias
+        | otherwise -> Right cmd
 
--- cmdWithMsgBody :: ACommand p -> m (Either ErrorType (ACommand p))
--- cmdWithMsgBody = \case
---   SEND body -> SEND <$$> getMsgBody body
---   MSG agentMsgId srvTS agentTS status body -> MSG agentMsgId srvTS agentTS status <$$> getMsgBody body
---   cmd -> return $ Right cmd
+    cmdWithMsgBody :: ACommand p -> m (Either ErrorType (ACommand p))
+    cmdWithMsgBody = \case
+      SEND body -> SEND <$$> getMsgBody body
+      MSG agentMsgId srvTS agentTS status body -> MSG agentMsgId srvTS agentTS status <$$> getMsgBody body
+      cmd -> return $ Right cmd
 
--- getMsgBody :: MsgBody -> m (Either ErrorType MsgBody)
--- getMsgBody msgBody =
---   case B.unpack msgBody of
---     ':' : body -> return . Right $ B.pack body
---     str -> case readMaybe str :: Maybe Int of
---       Just size -> do
---         body <- getBytes h size
---         s <- getLn h
---         return $ if B.null s then Right body else Left SIZE
---       Nothing -> return . Left $ SYNTAX errMessageBody
-
-aCmdGet :: forall m p. MonadIO m => SAParty p -> Handle -> m (Either ErrorType (ACommand p))
-aCmdGet _ h = getLn h >>= (\_ -> return $ Left UNKNOWN)
+    getMsgBody :: MsgBody -> m (Either ErrorType MsgBody)
+    getMsgBody msgBody =
+      case B.unpack msgBody of
+        ':' : body -> return . Right $ B.pack body
+        str -> case readMaybe str :: Maybe Int of
+          Just size -> do
+            body <- getBytes h size
+            s <- getLn h
+            return $ if B.null s then Right body else Left SIZE
+          Nothing -> return . Left $ SYNTAX errMessageBody

src/Simplex/Messaging/Server/Transmission.hs

@@ -145,7 +145,7 @@ serializeCommand = \case
 type Encoded = ByteString
 
 -- newtype to avoid accidentally changing order of transmission parts
-newtype CorrId = CorrId {bs :: ByteString} deriving (Eq)
+newtype CorrId = CorrId {bs :: ByteString} deriving (Eq, Show)
 
 instance IsString CorrId where
   fromString = CorrId . fromString
@@ -182,8 +182,8 @@ errNoCredentials = 3
 errHasCredentials :: Int
 errHasCredentials = 4
 
-errNoConnectionId :: Int
-errNoConnectionId = 5
+errNoQueueId :: Int
+errNoQueueId = 5
 
 errMessageBody :: Int
 errMessageBody = 6
@@ -237,19 +237,19 @@ tGet fromParty h = do
     tCredentials (signature, _, queueId, _) cmd = case cmd of
       -- IDS response should not have queue ID
       Cmd SBroker (IDS _ _) -> Right cmd
-      -- ERROR response does not always have queue ID
+      -- ERR response does not always have queue ID
       Cmd SBroker (ERR _) -> Right cmd
       -- other responses must have queue ID
       Cmd SBroker _
-        | B.null queueId -> Left $ SYNTAX errNoConnectionId
+        | B.null queueId -> Left $ SYNTAX errNoQueueId
         | otherwise -> Right cmd
-      -- CREATE must NOT have signature or queue ID
+      -- NEW must NOT have signature or queue ID
       Cmd SRecipient (NEW _)
         | B.null signature && B.null queueId -> Right cmd
         | otherwise -> Left $ SYNTAX errHasCredentials
       -- SEND must have queue ID, signature is not always required
       Cmd SSender (SEND _)
-        | B.null queueId -> Left $ SYNTAX errNoConnectionId
+        | B.null queueId -> Left $ SYNTAX errNoQueueId
         | otherwise -> Right cmd
       -- other client commands must have both signature and queue ID
       Cmd SRecipient _