wrap agent command for async command processing, to allow extension with internal commands (#528)

* wrap agent command for async command processing, to allow extension with internal commands

* change encoding for constructor

.gitignore

@@ -5,4 +5,5 @@
 tests/tmp
 dist-newstyle/
 
-src/tor
+cabal.project.local
+cabal.project.local~

src/Simplex/Messaging/Agent.hs

@@ -641,7 +641,7 @@ sendMessage' c connId msgFlags msg =
 enqueueCommand :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> m ()
 enqueueCommand c corrId connId server aCommand = do
   resumeSrvCmds c server
-  commandId <- withStore' c $ \db -> createCommand db corrId connId server aCommand
+  commandId <- withStore' c $ \db -> createCommand db corrId connId server $ AClientCommand aCommand
   queuePendingCommands c server [commandId]
 
 resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
@@ -689,9 +689,9 @@ runCommandProcessing c@AgentClient {subQ} server = do
     atomically $ beginAgentOperation c AOSndNetwork
     E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
       Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", ERR . INTERNAL $ show e)
-      Right (corrId, connId, ACmd _ cmd) -> processCmd ri corrId connId cmdId cmd
+      Right (corrId, connId, AClientCommand cmd) -> processCmd ri corrId connId cmdId cmd
   where
-    processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> ACommand p -> m ()
+    processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> ACommand 'Client -> m ()
     processCmd ri corrId connId cmdId = \case
       NEW enableNtfs (ACM cMode) -> do
         usedSrvs <- newTVarIO ([] :: [SMPServer])

src/Simplex/Messaging/Agent/Protocol.hs

@@ -105,6 +105,7 @@ module Simplex.Messaging.Agent.Protocol
     agentMessageType,
     extraSMPServerHosts,
     updateSMPServerHosts,
+    checkParty,
 
     -- * TCP transport functions
     tPut,
@@ -222,7 +223,7 @@ instance APartyI Agent where sAParty = SAgent
 
 instance APartyI Client where sAParty = SClient
 
-data ACmd = forall p. ACmd (SAParty p) (ACommand p)
+data ACmd = forall p. APartyI p => ACmd (SAParty p) (ACommand p)
 
 deriving instance Show ACmd
 
@@ -1008,9 +1009,9 @@ networkCommandP = commandP A.takeByteString
 dbCommandP :: Parser ACmd
 dbCommandP = commandP $ A.take =<< (A.decimal <* "\n")
 
-instance Encoding ACmdTag where
-  smpEncode (ACmdTag _ cmd) = smpEncode cmd
-  smpP =
+instance StrEncoding ACmdTag where
+  strEncode (ACmdTag _ cmd) = strEncode cmd
+  strP =
     A.takeTill (== ' ') >>= \case
       "NEW" -> pure $ ACmdTag SClient NEW_
       "INV" -> pure $ ACmdTag SAgent INV_
@@ -1043,8 +1044,8 @@ instance Encoding ACmdTag where
       "SUSPENDED" -> pure $ ACmdTag SAgent SUSPENDED_
       _ -> fail "bad ACmdTag"
 
-instance APartyI p => Encoding (ACommandTag p) where
-  smpEncode = \case
+instance APartyI p => StrEncoding (ACommandTag p) where
+  strEncode = \case
     NEW_ -> "NEW"
     INV_ -> "INV"
     JOIN_ -> "JOIN"
@@ -1074,7 +1075,7 @@ instance APartyI p => Encoding (ACommandTag p) where
     OK_ -> "OK"
     ERR_ -> "ERR"
     SUSPENDED_ -> "SUSPENDED"
-  smpP = (\(ACmdTag _ t) -> checkParty t) <$?> smpP
+  strP = (\(ACmdTag _ t) -> checkParty t) <$?> strP
 
 checkParty :: forall t p p'. (APartyI p, APartyI p') => t p' -> Either String (t p)
 checkParty x = case testEquality (sAParty @p) (sAParty @p') of
@@ -1084,7 +1085,7 @@ checkParty x = case testEquality (sAParty @p) (sAParty @p') of
 -- | SMP agent command and response parser
 commandP :: Parser ByteString -> Parser ACmd
 commandP binaryP =
-  smpP
+  strP
     >>= \case
       ACmdTag SClient cmd ->
         ACmd SClient <$> case cmd of

src/Simplex/Messaging/Agent/Store.hs

@@ -3,13 +3,16 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
 module Simplex.Messaging.Agent.Store where
 
 import Control.Exception (Exception)
+import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
 import Data.Kind (Type)
@@ -18,6 +21,7 @@ import Data.Type.Equality
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (RatchetX448)
+import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
   ( MsgBody,
     MsgFlags,
@@ -31,6 +35,7 @@ import Simplex.Messaging.Protocol
     SndPrivateSignKey,
   )
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Util ((<$?>))
 import Simplex.Messaging.Version
 
 -- * Queue types
@@ -159,6 +164,30 @@ data ConnData = ConnData
   }
   deriving (Eq, Show)
 
+data AgentCommand = AClientCommand (ACommand 'Client)
+
+instance StrEncoding AgentCommand where
+  strEncode = \case
+    AClientCommand cmd -> "CLIENT " <> serializeCommand cmd
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "CLIENT" -> AClientCommand <$> (A.space *> ((\(ACmd _ cmd) -> checkParty cmd) <$?> dbCommandP))
+      _ -> fail "bad AgentCommand"
+
+data AgentCommandTag = AClientCommandTag (ACommandTag 'Client)
+
+instance StrEncoding AgentCommandTag where
+  strEncode = \case
+    AClientCommandTag t -> "CLIENT " <> strEncode t
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "CLIENT" -> AClientCommandTag <$> (A.space *> strP)
+      _ -> fail "bad AgentCommandTag"
+
+agentCommandTag :: AgentCommand -> AgentCommandTag
+agentCommandTag = \case
+  AClientCommand cmd -> AClientCommandTag $ aCommandTag cmd
+
 -- * Confirmation types
 
 data NewConfirmation = NewConfirmation

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

@@ -118,7 +118,7 @@ import Data.Char (toLower)
 import Data.Function (on)
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.List (find, foldl', groupBy)
+import Data.List (foldl', groupBy)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe, listToMaybe)
@@ -702,12 +702,12 @@ updateRatchet db connId rc skipped = do
         forM_ (M.assocs mks) $ \(msgN, mk) ->
           DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
 
-createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId
+createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO AsyncCmdId
 createCommand db corrId connId srv cmd = do
   DB.execute
     db
     "INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command) VALUES (?,?,?,?,?,?)"
-    (host_, port_, corrId, connId, aCommandTag cmd, cmd)
+    (host_, port_, corrId, connId, agentCommandTag cmd, cmd)
   insertedRowId db
   where
     (host_, port_) =
@@ -734,7 +734,7 @@ getPendingCommands db connId = do
   where
     srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId)
 
-getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, ACmd))
+getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, AgentCommand))
 getPendingCommand db msgId = do
   firstRow id SECmdNotFound $
     DB.query
@@ -1108,13 +1108,13 @@ instance ToField (NonEmpty TransportHost) where toField = toField . decodeLatin1
 
 instance FromField (NonEmpty TransportHost) where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
 
-instance ToField (ACommand p) where toField = toField . serializeCommand
+instance ToField AgentCommand where toField = toField . strEncode
 
-instance FromField ACmd where fromField = blobFieldParser dbCommandP
+instance FromField AgentCommand where fromField = blobFieldParser strP
 
-instance APartyI p => ToField (ACommandTag p) where toField = toField . smpEncode
+instance ToField AgentCommandTag where toField = toField . strEncode
 
-instance FromField ACmdTag where fromField = blobFieldParser smpP
+instance FromField AgentCommandTag where fromField = blobFieldParser strP
 
 listToEither :: e -> [a] -> Either e a
 listToEither _ (x : _) = Right x