xftp: agent command entities (#676)

* xftp: agent command entities

* progress event

* parameterize agent command by entity

* refactor

* Eq instance for APartyCmdTag

tests/AgentTests.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE PostfixOperators #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 
 module AgentTests (agentTests) where
@@ -19,6 +20,7 @@ import Control.Concurrent
 import Control.Monad (forM_)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.Type.Equality
 import Network.HTTP.Types (urlEncode)
 import SMPAgentClient
 import SMPClient (testKeyHash, testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn)
@@ -81,43 +83,53 @@ agentTests (ATransport t) = do
     it "should resume delivering messages after exceeding quota once all messages are received" $
       smpAgentTest2_2_1 $ testResumeDeliveryQuotaExceeded t
 
-tGetAgent :: Transport c => c -> IO (ATransmissionOrError 'Agent)
-tGetAgent h = do
-  t@(_, _, cmd) <- tGet SAgent h
-  case cmd of
-    Right CONNECT {} -> tGetAgent h
-    Right DISCONNECT {} -> tGetAgent h
-    _ -> pure t
+type AEntityTransmission p e = (ACorrId, ConnId, ACommand p e)
+
+type AEntityTransmissionOrError p e = (ACorrId, ConnId, Either AgentErrorType (ACommand p e))
+
+tGetAgent :: Transport c => c -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
+tGetAgent = tGetAgent'
+
+tGetAgent' :: forall c e. (Transport c, AEntityI e) => c -> IO (AEntityTransmissionOrError 'Agent e)
+tGetAgent' h = do
+  (corrId, connId, cmdOrErr) <- tGet SAgent h
+  case cmdOrErr of
+    Right (APC _ CONNECT {}) -> tGetAgent' h
+    Right (APC _ DISCONNECT {}) -> tGetAgent' h
+    Right (APC e cmd) -> case testEquality e (sAEntity @e) of
+      Just Refl -> pure (corrId, connId, Right cmd)
+      _ -> error $ "unexpected command " <> show cmd
+    Left err -> pure (corrId, connId, Left err)
 
 -- | receive message to handle `h`
-(<#:) :: Transport c => c -> IO (ATransmissionOrError 'Agent)
+(<#:) :: Transport c => c -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
 (<#:) = tGetAgent
 
 -- | send transmission `t` to handle `h` and get response
-(#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (ATransmissionOrError 'Agent)
+(#:) :: Transport c => c -> (ByteString, ByteString, ByteString) -> IO (AEntityTransmissionOrError 'Agent 'AEConn)
 h #: t = tPutRaw h t >> (<#:) h
 
 -- | action and expected response
 -- `h #:t #> r` is the test that sends `t` to `h` and validates that the response is `r`
-(#>) :: IO (ATransmissionOrError 'Agent) -> ATransmission 'Agent -> Expectation
+(#>) :: IO (AEntityTransmissionOrError 'Agent 'AEConn) -> AEntityTransmission 'Agent 'AEConn -> Expectation
 action #> (corrId, connId, cmd) = action `shouldReturn` (corrId, connId, Right cmd)
 
 -- | action and predicate for the response
 -- `h #:t =#> p` is the test that sends `t` to `h` and validates the response using `p`
-(=#>) :: IO (ATransmissionOrError 'Agent) -> (ATransmission 'Agent -> Bool) -> Expectation
+(=#>) :: IO (AEntityTransmissionOrError 'Agent 'AEConn) -> (AEntityTransmission 'Agent 'AEConn -> Bool) -> Expectation
 action =#> p = action >>= (`shouldSatisfy` p . correctTransmission)
 
-correctTransmission :: ATransmissionOrError a -> ATransmission a
+correctTransmission :: AEntityTransmissionOrError p e -> AEntityTransmission p e
 correctTransmission (corrId, connId, cmdOrErr) = case cmdOrErr of
   Right cmd -> (corrId, connId, cmd)
   Left e -> error $ show e
 
 -- | receive message to handle `h` and validate that it is the expected one
-(<#) :: Transport c => c -> ATransmission 'Agent -> Expectation
+(<#) :: Transport c => c -> AEntityTransmission 'Agent 'AEConn -> Expectation
 h <# (corrId, connId, cmd) = (h <#:) `shouldReturn` (corrId, connId, Right cmd)
 
 -- | receive message to handle `h` and validate it using predicate `p`
-(<#=) :: Transport c => c -> (ATransmission 'Agent -> Bool) -> Expectation
+(<#=) :: Transport c => c -> (AEntityTransmission 'Agent 'AEConn -> Bool) -> Expectation
 h <#= p = (h <#:) >>= (`shouldSatisfy` p . correctTransmission)
 
 -- | test that nothing is delivered to handle `h` during 10ms
@@ -129,7 +141,7 @@ h #:# err = tryGet `shouldReturn` ()
         Just _ -> error err
         _ -> return ()
 
-pattern Msg :: MsgBody -> ACommand 'Agent
+pattern Msg :: MsgBody -> ACommand 'Agent 'AEConn
 pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody
 
 testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()

tests/AgentTests/FunctionalAPITests.hs

@@ -8,6 +8,7 @@
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
 
 module AgentTests.FunctionalAPITests
@@ -19,6 +20,8 @@ module AgentTests.FunctionalAPITests
     runRight,
     runRight_,
     get,
+    get',
+    rfGet,
     (##>),
     (=##>),
     pattern Msg,
@@ -38,6 +41,7 @@ import qualified Data.Map as M
 import Data.Maybe (isNothing)
 import qualified Data.Set as S
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
+import Data.Type.Equality
 import SMPAgentClient
 import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent
@@ -58,21 +62,31 @@ import Simplex.Messaging.Version
 import Test.Hspec
 import UnliftIO
 
-(##>) :: (HasCallStack, MonadIO m) => m (ATransmission 'Agent) -> ATransmission 'Agent -> m ()
+type AEntityTransmission e = (ACorrId, ConnId, ACommand 'Agent e)
+
+(##>) :: (HasCallStack, MonadIO m) => m (AEntityTransmission e) -> AEntityTransmission e -> m ()
 a ##> t = a >>= \t' -> liftIO (t' `shouldBe` t)
 
-(=##>) :: (HasCallStack, MonadIO m) => m (ATransmission 'Agent) -> (ATransmission 'Agent -> Bool) -> m ()
+(=##>) :: (HasCallStack, MonadIO m) => m (AEntityTransmission e) -> (AEntityTransmission e -> Bool) -> m ()
 a =##> p = a >>= \t -> liftIO (t `shouldSatisfy` p)
 
-get :: MonadIO m => AgentClient -> m (ATransmission 'Agent)
-get c = do
-  t@(_, _, cmd) <- atomically (readTBQueue $ subQ c)
-  case cmd of
-    CONNECT {} -> get c
-    DISCONNECT {} -> get c
-    _ -> pure t
+get :: MonadIO m => AgentClient -> m (AEntityTransmission 'AEConn)
+get = get' @'AEConn
 
-pattern Msg :: MsgBody -> ACommand 'Agent
+rfGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AERcvFile)
+rfGet = get' @'AERcvFile
+
+get' :: forall e m. (MonadIO m, AEntityI e) => AgentClient -> m (AEntityTransmission e)
+get' c = do
+  (corrId, connId, APC e cmd) <- atomically (readTBQueue $ subQ c)
+  case cmd of
+    CONNECT {} -> get' c
+    DISCONNECT {} -> get' c
+    _ -> case testEquality e (sAEntity @e) of
+      Just Refl -> pure (corrId, connId, cmd)
+      _ -> error $ "unexpected command " <> show cmd
+
+pattern Msg :: MsgBody -> ACommand 'Agent 'AEConn
 pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody
 
 smpCfgV1 :: ProtocolClientConfig
@@ -524,7 +538,7 @@ testSuspendingAgent = do
     get b =##> \case ("", c, Msg "hello") -> c == aId; _ -> False
     ackMessage b aId 4
     suspendAgent b 1000000
-    get b ##> ("", "", SUSPENDED)
+    get' b ##> ("", "", SUSPENDED)
     5 <- sendMessage a bId SMP.noMsgFlags "hello 2"
     get a ##> ("", bId, SENT 5)
     Nothing <- 100000 `timeout` get b
@@ -551,11 +565,11 @@ testSuspendingAgentCompleteSending t = do
     liftIO $ threadDelay 100000
     suspendAgent b 5000000
 
-  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ @AgentErrorType $ do
     get b =##> \case ("", c, SENT 5) -> c == aId; ("", "", UP {}) -> True; _ -> False
     get b =##> \case ("", c, SENT 5) -> c == aId; ("", "", UP {}) -> True; _ -> False
     get b =##> \case ("", c, SENT 6) -> c == aId; ("", "", UP {}) -> True; _ -> False
-    ("", "", SUSPENDED) <- get b
+    ("", "", SUSPENDED) <- get' @'AENone b
 
     get a =##> \case ("", c, Msg "hello too") -> c == bId; ("", "", UP {}) -> True; _ -> False
     get a =##> \case ("", c, Msg "hello too") -> c == bId; ("", "", UP {}) -> True; _ -> False
@@ -581,7 +595,7 @@ testSuspendingAgentTimeout t = do
     5 <- sendMessage b aId SMP.noMsgFlags "hello too"
     6 <- sendMessage b aId SMP.noMsgFlags "how are you?"
     suspendAgent b 100000
-    ("", "", SUSPENDED) <- get b
+    ("", "", SUSPENDED) <- get' @'AENone b
     pure ()
 
 testBatchedSubscriptions :: ATransport -> IO ()
@@ -777,7 +791,7 @@ testUsers = do
     deleteUser a auId True
     get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId'; _ -> False
     get a =##> \case ("", c, DEL_CONN) -> c == bId'; _ -> False
-    get a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
+    get' @'AENone a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
     exchangeGreetingsMsgId 6 a bId b aId
     liftIO $ noMessages a "nothing else should be delivered to alice"
 
@@ -813,7 +827,7 @@ testUsersNoServer t = do
     deleteUser a auId True
     get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c == bId' && (e == TIMEOUT || e == NETWORK); _ -> False
     get a =##> \case ("", c, DEL_CONN) -> c == bId'; _ -> False
-    get a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
+    get' @'AENone a =##> \case ("", "", DEL_USER u) -> u == auId; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
   withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
     get a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False

tests/SMPAgentClient.hs

@@ -66,8 +66,8 @@ smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> get h
     get h = do
       t@(_, _, cmdStr) <- tGetRaw h
       case parseAll networkCommandP cmdStr of
-        Right (ACmd SAgent CONNECT {}) -> get h
-        Right (ACmd SAgent DISCONNECT {}) -> get h
+        Right (ACmd SAgent _ CONNECT {}) -> get h
+        Right (ACmd SAgent _ DISCONNECT {}) -> get h
         _ -> pure t
 
 runSmpAgentTest :: forall c a. Transport c => (c -> IO a) -> IO a

tests/XFTPAgent.hs

@@ -5,7 +5,7 @@
 
 module XFTPAgent where
 
-import AgentTests.FunctionalAPITests (get, runRight_)
+import AgentTests.FunctionalAPITests (get, rfGet, runRight_)
 import Control.Monad.Except
 import Data.Bifunctor (first)
 import qualified Data.ByteString as LB
@@ -13,7 +13,7 @@ import SMPAgentClient (agentCfg, initAgentServers)
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), checkParty)
 import Simplex.Messaging.Agent (disconnectAgentClient, getSMPAgentClient, xftpReceiveFile)
-import Simplex.Messaging.Agent.Protocol (ACommand (FRCVD), AgentErrorType (..))
+import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..))
 import Simplex.Messaging.Encoding.String (StrEncoding (..))
 import System.Directory (getFileSize)
 import System.FilePath ((</>))
@@ -48,7 +48,7 @@ testXFTPAgentReceive = withXFTPServer $ do
   runRight_ $ do
     fd :: ValidFileDescription 'FPRecipient <- getFileDescription fdRcv
     fId <- xftpReceiveFile rcp 1 fd recipientFiles
-    ("", "", FRCVD fId' path) <- get rcp
+    ("", "", RFDONE fId' path) <- rfGet rcp
     liftIO $ do
       fId' `shouldBe` fId
       LB.readFile path `shouldReturn` file
@@ -89,7 +89,7 @@ testXFTPAgentReceiveRestore = do
   rcp' <- getSMPAgentClient agentCfg initAgentServers
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file using agent - should succeed with server up
-    ("", "", FRCVD fId' path) <- get rcp'
+    ("", "", RFDONE fId' path) <- rfGet rcp'
     liftIO $ do
       fId' `shouldBe` 1
       file <- LB.readFile filePath