agent store: add update queue methods (#19)

src/Simplex/Messaging/Agent.hs

@@ -212,12 +212,12 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
       print cmd
   where
     secureQueue :: ReceiveQueue -> SMP.SenderKey -> m ()
-    secureQueue ReceiveQueue {rcvId, rcvPrivateKey} senderKey = do
-      withStore $ \st -> updateReceiveQueueStatus st rcvId Confirmed
+    secureQueue rq@ReceiveQueue {rcvPrivateKey} senderKey = do
+      withStore $ \st -> updateRcvQueueStatus st rq Confirmed
       -- TODO update sender key in the store
       smp <- getSMPServerClient c srv
       liftSMP $ secureSMPQueue smp rcvPrivateKey rId senderKey
-      withStore $ \st -> updateReceiveQueueStatus st rcvId Secured
+      withStore $ \st -> updateRcvQueueStatus st rq Secured
 
 decryptMessage :: MonadUnliftIO m => PrivateKey -> ByteString -> m ByteString
 decryptMessage _decryptKey = return
@@ -272,12 +272,12 @@ sendConfirmation ::
   SendQueue ->
   SMP.SenderKey ->
   m ()
-sendConfirmation c SendQueue {server, sndId} senderKey = do
+sendConfirmation c sq@SendQueue {server, sndId} senderKey = do
   -- TODO send initial confirmation with signature - change in SMP server
   smp <- getSMPServerClient c server
   msg <- mkConfirmation
   liftSMP $ sendSMPMessage smp "" sndId msg
-  withStore $ \st -> updateSendQueueStatus st sndId Confirmed
+  withStore $ \st -> updateSndQueueStatus st sq Confirmed
   where
     mkConfirmation :: m SMP.MsgBody
     mkConfirmation = do
@@ -291,11 +291,11 @@ sendHello ::
   AgentClient ->
   SendQueue ->
   m ()
-sendHello c SendQueue {server, sndId, sndPrivateKey, encryptKey} = do
+sendHello c sq@SendQueue {server, sndId, sndPrivateKey, encryptKey} = do
   smp <- getSMPServerClient c server
   msg <- mkHello "" $ AckMode On -- TODO verifyKey
   _send smp 20 msg
-  withStore $ \st -> updateSendQueueStatus st sndId Active
+  withStore $ \st -> updateSndQueueStatus st sq Active
   where
     mkHello :: PublicKey -> AckMode -> m ByteString
     mkHello verifyKey ackMode =

src/Simplex/Messaging/Agent/Store.hs

@@ -103,9 +103,8 @@ class Monad m => MonadAgentStore s m where
   addSndQueue :: s -> ConnAlias -> SendQueue -> m ()
   addRcvQueue :: s -> ConnAlias -> ReceiveQueue -> m ()
   removeSndAuth :: s -> ConnAlias -> m ()
-  updateQueueStatus :: s -> ConnAlias -> QueueDirection -> QueueStatus -> m ()
-  updateReceiveQueueStatus :: s -> RecipientId -> QueueStatus -> m ()
-  updateSendQueueStatus :: s -> SenderId -> QueueStatus -> m ()
+  updateRcvQueueStatus :: s -> ReceiveQueue -> QueueStatus -> m ()
+  updateSndQueueStatus :: s -> SendQueue -> QueueStatus -> m ()
   createMsg :: s -> ConnAlias -> QueueDirection -> AgentMsgId -> AMessage -> m ()
   getLastMsg :: s -> ConnAlias -> QueueDirection -> m MessageDelivery
   getMsg :: s -> ConnAlias -> QueueDirection -> AgentMsgId -> m MessageDelivery

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

@@ -39,6 +39,7 @@ import Simplex.Messaging.Util
 import Text.Read
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+import Network.Socket
 
 addRcvQueueQuery :: Query
 addRcvQueueQuery =
@@ -310,29 +311,39 @@ deleteConnection store connAlias = do
     "DELETE FROM connections WHERE conn_alias = ?"
     (Only connAlias)
 
-updateRcvQueueStatus :: MonadUnliftIO m => SQLiteStore -> QueueRowId -> QueueStatus -> m ()
-updateRcvQueueStatus store rcvQueueId status =
+updateReceiveQueueStatus :: MonadUnliftIO m => SQLiteStore -> RecipientId -> HostName -> Maybe ServiceName -> QueueStatus -> m ()
+updateReceiveQueueStatus store rcvQueueId host port status =
   executeWithLock
     store
     rcvQueuesLock
     [s|
       UPDATE receive_queues
       SET status = ?
-      WHERE receive_queue_id = ?;
+      WHERE rcv_id = ?
+      AND server_id IN (
+        SELECT server_id
+        FROM servers
+        WHERE host = ? AND port = ?
+      );
     |]
-    (Only status :. Only rcvQueueId)
+    (Only status :. Only rcvQueueId :. Only host :. Only port)
 
-updateSndQueueStatus :: MonadUnliftIO m => SQLiteStore -> QueueRowId -> QueueStatus -> m ()
-updateSndQueueStatus store sndQueueId status =
+updateSendQueueStatus :: MonadUnliftIO m => SQLiteStore -> SenderId -> HostName -> Maybe ServiceName -> QueueStatus -> m ()
+updateSendQueueStatus store sndQueueId host port status =
   executeWithLock
     store
     sndQueuesLock
     [s|
       UPDATE send_queues
       SET status = ?
-      WHERE send_queue_id = ?;
+      WHERE snd_id = ?
+      AND server_id IN (
+        SELECT server_id
+        FROM servers
+        WHERE host = ? AND port = ?
+      );
     |]
-    (Only status :. Only sndQueueId)
+    (Only status :. Only sndQueueId :. Only host :. Only port)
 
 instance ToField QueueDirection where toField = toField . show
 
@@ -426,27 +437,17 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
     when (isNothing rcvQId && isNothing sndQId) $ throwError SEBadConn
 
   removeSndAuth :: SQLiteStore -> ConnAlias -> m ()
-  removeSndAuth _st _connAlias = throwError SEInternal
+  removeSndAuth _st _connAlias = throwError SENotImplemented
 
-  updateQueueStatus :: SQLiteStore -> ConnAlias -> QueueDirection -> QueueStatus -> m ()
-  updateQueueStatus st connAlias qDirection status = do
-    case qDirection of
-      RCV -> do
-        (rcvQId, _) <- getConnection st connAlias
-        case rcvQId of
-          Just qId -> updateRcvQueueStatus st qId status
-          Nothing -> throwError SEBadQueueDirection
-      SND -> do
-        (_, sndQId) <- getConnection st connAlias
-        case sndQId of
-          Just qId -> updateSndQueueStatus st qId status
-          Nothing -> throwError SEBadQueueDirection
+  -- TODO throw error if queue doesn't exist
+  updateRcvQueueStatus :: SQLiteStore -> ReceiveQueue -> QueueStatus -> m ()
+  updateRcvQueueStatus st ReceiveQueue {rcvId, server = SMPServer {host, port}} status =
+    updateReceiveQueueStatus st rcvId host port status
 
-  updateReceiveQueueStatus :: SQLiteStore -> RecipientId -> QueueStatus -> m ()
-  updateReceiveQueueStatus _st _rId _status = throwError SENotImplemented
-
-  updateSendQueueStatus :: SQLiteStore -> SenderId -> QueueStatus -> m ()
-  updateSendQueueStatus _st _sId _status = throwError SENotImplemented
+  -- TODO throw error if queue doesn't exist
+  updateSndQueueStatus :: SQLiteStore -> SendQueue -> QueueStatus -> m ()
+  updateSndQueueStatus st SendQueue {sndId, server = SMPServer {host, port}} status =
+    updateSendQueueStatus st sndId host port status
 
   -- TODO decrease duplication of queue direction checks?
   createMsg :: SQLiteStore -> ConnAlias -> QueueDirection -> AgentMsgId -> AMessage -> m ()