store functions

rfcs/2022-08-14-queue-rotation.md

@@ -40,7 +40,7 @@ A ->> S ->> B: QNEW (R'): address of the new queue
 B ->> R ->> A: QKEYS (R'): sender's key for the new queue (to avoid the race of SMP confirmation for the initial exchange)
 B ->> R ->> A: continue sending new messages to the old queue
 A ->> R': secure queue
-A ->> S ->> B: QREADY (R'): instruction to use new queue
+A ->> S ->> B: QREADY (R'): notify sender that the queue is secured
 B ->> R' ->> A: QHELLO: to validate that the sender can send messages to the new queue before switching to it
 A ->> S ->> B: QSWITCH (R'): instruction to start using the new queue
 B ->> R' ->> A: the first message received to the new queue before the old one is drained and deleted should not be processed, it should be stored in the agent memory (and not acknowledged) and only processed once the old queue is drained.

src/Simplex/Messaging/Agent.hs

@@ -1173,7 +1173,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                 | otherwise -> ignored
               _ -> ignored
             ignored = pure "END from disconnected client - ignored"
-        LEN 0 -> do
+        SMP.LEN 0 -> do
           -- load nextRq
           -- currRcvQueueDrained c rq nextRq
           pure ()

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

@@ -378,8 +378,31 @@ setRcvQueueNtfCreds db connId clientNtfCreds =
       Nothing -> (Nothing, Nothing, Nothing, Nothing)
 
 getNextRcvQueue :: DB.Connection -> Maybe Int64 -> IO (Maybe RcvQueue)
-getNextRcvQueue _db = \case
-  Just _rqId -> pure Nothing
+getNextRcvQueue db = \case
+  Just rqId ->
+    listToMaybe . map rcvQueue
+      <$> DB.query
+        db
+        [sql|
+          SELECT q.host, q.port, s.key_hash,
+            q.rcv_id, q.rcv_private_key, q.rcv_dh_secret, q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status,
+            q.rcv_queue_action, q.rcv_queue_action_ts, q.next_rcv_queue_id,
+            q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
+            q.smp_client_version, q.created_at, q.updated_at
+          FROM rcv_queues q
+          INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+          WHERE q.rcv_queue_id = ? AND q.next_rcv_queue = ?
+        |]
+        (rqId, True)
+    where
+      rcvQueue (srvRow :. (rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, rqAction_, rqActionTs_, dbNextRcvQueueId) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_) :. (smpClientVersion_, createdAt, updatedAt)) =
+        let server = toSMPServer srvRow
+            smpClientVersion = fromMaybe 1 smpClientVersion_
+            rcvQueueAction = (,) <$> rqAction_ <*> rqActionTs_
+            clientNtfCreds = case (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_) of
+              (Just ntfPublicKey, Just ntfPrivateKey, Just notifierId, Just rcvNtfDhSecret) -> Just $ ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
+              _ -> Nothing
+         in RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, rcvQueueAction, dbNextRcvQueueId, smpClientVersion, clientNtfCreds, createdAt, updatedAt}
   _ -> pure Nothing
 
 type SMPConfirmationRow = (SndPublicVerifyKey, C.PublicKeyX25519, ConnInfo, Maybe [SMPQueueInfo], Maybe Version)
@@ -1100,26 +1123,32 @@ upsertNtfServer_ db ProtocolServer {host, port, keyHash} = do
 -- * createRcvConn helpers
 
 insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
-insertRcvQueue_ dbConn connId RcvQueue {..} = do
+insertRcvQueue_ db connId RcvQueue {..} = do
+  qId <- newQueueId_ <$> DB.query_ db "SELECT rcv_queue_id FROM rcv_queues ORDER BY rcv_queue_id DESC LIMIT 1"
   DB.execute
-    dbConn
+    db
     [sql|
       INSERT INTO rcv_queues
-        (host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, smp_client_version, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?);
+        (rcv_queue_id, host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, smp_client_version, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    ((host server, port server, rcvId, connId) :. (rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (smpClientVersion, createdAt, updatedAt))
+    ((qId, host server, port server, rcvId, connId) :. (rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (smpClientVersion, createdAt, updatedAt))
 
 -- * createSndConn helpers
 
 insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
-insertSndQueue_ dbConn connId SndQueue {..} = do
+insertSndQueue_ db connId SndQueue {..} = do
+  qId <- newQueueId_ <$> DB.query_ db "SELECT snd_queue_id FROM snd_queues ORDER BY snd_queue_id DESC LIMIT 1"
   DB.execute
-    dbConn
+    db
     [sql|
       INSERT INTO snd_queues
-        (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, smp_client_version, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?);
+        (snd_queue_id, host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, smp_client_version, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    ((host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) :. (smpClientVersion, createdAt, updatedAt))
+    ((qId, host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) :. (smpClientVersion, createdAt, updatedAt))
+
+newQueueId_ :: [Only (Maybe Int64)] -> Int64
+newQueueId_ [] = 1
+newQueueId_ (Only maxId_ : _) = maybe 1 (+ 1) maxId_
 
 -- * getConn helpers