agent: save messages (#45)

src/Simplex/Messaging/Agent.hs

@@ -29,7 +29,6 @@ import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
-import Simplex.Messaging.Agent.Store.Types
 import Simplex.Messaging.Agent.Transmission
 import Simplex.Messaging.Client (SMPServerTransmission)
 import qualified Simplex.Messaging.Crypto as C
@@ -108,7 +107,7 @@ withStore action = do
   store <- asks db
   runExceptT (action store `E.catch` handleInternal) >>= \case
     Right c -> return c
-    Left e -> throwError $ STORE e
+    Left _ -> throwError STORE
   where
     handleInternal :: (MonadError StoreError m') => SomeException -> m' a
     handleInternal _ = throwError SEInternal
@@ -163,9 +162,9 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
         _ -> throwError PROHIBITED -- NOT_READY ?
       where
         sendMsg sq = do
+          withStore $ \st -> createSndMsg st connAlias msgBody
           sendAgentMessage c sq $ A_MSG msgBody
           -- TODO respond $ SENT aMsgId
-          -- TODO send message to DB
           respond $ SENT 0
 
     suspendConnection :: m ()
@@ -249,6 +248,12 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
             A_MSG body -> do
               logServer "<--" c srv rId "MSG <MSG>"
               -- TODO check message status
+              withStore $ \st -> createRcvMsg st connAlias body senderMsgId senderTimestamp srvMsgId srvTs
+              -- TODO either:
+              -- TODO   - pass MSG{.., msg = generatedRcvMsg} to notify,
+              -- TODO   - retrieve internalTs from generatedRcvMsg,
+              -- TODO   - pass recipientTs to createRcvMsg (second cleanest),
+              -- TODO   - remove m_recipient from MSG (might be the cleanest?).
               recipientTs <- liftIO getCurrentTime
               notify connAlias $
                 MSG

src/Simplex/Messaging/Agent/Store.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
@@ -6,26 +7,46 @@
 {-# LANGUAGE StandaloneDeriving #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
-module Simplex.Messaging.Agent.Store
-  ( ReceiveQueue (..),
-    SendQueue (..),
-    Connection (..),
-    SConnType (..),
-    SomeConn (..),
-    MessageDelivery (..),
-    DeliveryStatus (..),
-    MonadAgentStore (..),
-  )
-where
+module Simplex.Messaging.Agent.Store where
 
+import Control.Exception (Exception)
+import Data.Int (Int64)
 import Data.Kind (Type)
-import Data.Time.Clock (UTCTime)
+import Data.Time (UTCTime)
 import Data.Type.Equality
-import Simplex.Messaging.Agent.Store.Types (ConnType (..))
 import Simplex.Messaging.Agent.Transmission
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Types (RecipientPrivateKey, SenderPrivateKey, SenderPublicKey)
+import Simplex.Messaging.Types
+  ( MsgBody,
+    MsgId,
+    RecipientPrivateKey,
+    SenderPrivateKey,
+    SenderPublicKey,
+  )
 
+-- * Store management
+
+-- | Store class type. Defines store access methods for implementations.
+class Monad m => MonadAgentStore s m where
+  -- Queue and Connection management
+  createRcvConn :: s -> ReceiveQueue -> m ()
+  createSndConn :: s -> SendQueue -> m ()
+  getConn :: s -> ConnAlias -> m SomeConn
+  getRcvQueue :: s -> SMPServer -> SMP.RecipientId -> m ReceiveQueue
+  deleteConn :: s -> ConnAlias -> m ()
+  upgradeRcvConnToDuplex :: s -> ConnAlias -> SendQueue -> m ()
+  upgradeSndConnToDuplex :: s -> ConnAlias -> ReceiveQueue -> m ()
+  setRcvQueueStatus :: s -> ReceiveQueue -> QueueStatus -> m ()
+  setSndQueueStatus :: s -> SendQueue -> QueueStatus -> m ()
+
+  -- Msg management
+  createRcvMsg :: s -> ConnAlias -> MsgBody -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m ()
+  createSndMsg :: s -> ConnAlias -> MsgBody -> m ()
+  getMsg :: s -> ConnAlias -> InternalId -> m Msg
+
+-- * Queue types
+
+-- | A receive queue. SMP queue through which the agent receives messages from a sender.
 data ReceiveQueue = ReceiveQueue
   { server :: SMPServer,
     rcvId :: SMP.RecipientId,
@@ -39,6 +60,7 @@ data ReceiveQueue = ReceiveQueue
   }
   deriving (Eq, Show)
 
+-- | A send queue. SMP queue through which the agent sends messages to a recipient.
 data SendQueue = SendQueue
   { server :: SMPServer,
     sndId :: SMP.SenderId,
@@ -50,15 +72,30 @@ data SendQueue = SendQueue
   }
   deriving (Eq, Show)
 
+-- * Connection types
+
+-- | Type of a connection.
+data ConnType = CReceive | CSend | CDuplex deriving (Eq, Show)
+
+-- | Connection of a specific type.
+--
+-- - ReceiveConnection is a connection that only has a receive queue set up,
+--   typically created by a recipient initiating a duplex connection.
+--
+-- - SendConnection is a connection that only has a send queue set up, typically
+--   created by a sender joining a duplex connection through a recipient's invitation.
+--
+-- - DuplexConnection is a connection that has both receive and send queues set up,
+--   typically created by upgrading a receive or a send connection with a missing queue.
 data Connection (d :: ConnType) where
   ReceiveConnection :: ConnAlias -> ReceiveQueue -> Connection CReceive
   SendConnection :: ConnAlias -> SendQueue -> Connection CSend
   DuplexConnection :: ConnAlias -> ReceiveQueue -> SendQueue -> Connection CDuplex
 
-deriving instance Show (Connection d)
-
 deriving instance Eq (Connection d)
 
+deriving instance Show (Connection d)
+
 data SConnType :: ConnType -> Type where
   SCReceive :: SConnType CReceive
   SCSend :: SConnType CSend
@@ -74,8 +111,9 @@ instance TestEquality SConnType where
   testEquality SCDuplex SCDuplex = Just Refl
   testEquality _ _ = Nothing
 
-data SomeConn where
-  SomeConn :: SConnType d -> Connection d -> SomeConn
+-- | Connection of an unknown type.
+-- Used to refer to an arbitrary connection when retrieving from store.
+data SomeConn = forall d. SomeConn (SConnType d) (Connection d)
 
 instance Eq SomeConn where
   SomeConn d c == SomeConn d' c' = case testEquality d d' of
@@ -84,33 +122,100 @@ instance Eq SomeConn where
 
 deriving instance Show SomeConn
 
-data MessageDelivery = MessageDelivery
-  { connAlias :: ConnAlias,
-    agentMsgId :: Int,
-    timestamp :: UTCTime,
-    message :: AMessage,
-    direction :: QueueDirection,
-    msgStatus :: DeliveryStatus
+-- * Message types
+
+-- | A message in either direction that is stored by the agent.
+data Msg = MRcv RcvMsg | MSnd SndMsg
+  deriving (Eq, Show)
+
+-- | A message received by the agent from a sender.
+data RcvMsg = RcvMsg
+  { msgBase :: MsgBase,
+    internalRcvId :: InternalRcvId,
+    -- | Id of the message at sender, corresponds to `internalSndId` from the sender's side.
+    externalSndId :: ExternalSndId,
+    externalSndTs :: ExternalSndTs,
+    brokerId :: BrokerId,
+    brokerTs :: BrokerTs,
+    rcvStatus :: RcvStatus,
+    -- | Timestamp of acknowledgement to broker, corresponds to `AcknowledgedToBroker` status.
+    -- Do not mix up with `brokerTs` - timestamp created at broker after it receives the message from sender.
+    ackBrokerTs :: AckBrokerTs,
+    -- | Timestamp of acknowledgement to sender, corresponds to `AcknowledgedToSender` status.
+    -- Do not mix up with `externalSndTs` - timestamp created at sender before sending,
+    -- which in its turn corresponds to `internalTs` in sending agent.
+    ackSenderTs :: AckSenderTs
   }
+  deriving (Eq, Show)
 
-data DeliveryStatus
-  = MDTransmitted -- SMP: SEND sent / MSG received
-  | MDConfirmed -- SMP: OK received / ACK sent
-  | MDAcknowledged AckStatus -- SAMP: RCVD sent to agent client / ACK received from agent client and sent to the server
+type InternalRcvId = Int64
 
-class Monad m => MonadAgentStore s m where
-  createRcvConn :: s -> ReceiveQueue -> m ()
-  createSndConn :: s -> SendQueue -> m ()
-  getConn :: s -> ConnAlias -> m SomeConn
-  getRcvQueue :: s -> SMPServer -> SMP.RecipientId -> m ReceiveQueue
-  deleteConn :: s -> ConnAlias -> m ()
-  upgradeRcvConnToDuplex :: s -> ConnAlias -> SendQueue -> m ()
-  upgradeSndConnToDuplex :: s -> ConnAlias -> ReceiveQueue -> m ()
-  removeSndAuth :: s -> ConnAlias -> m ()
-  setRcvQueueStatus :: s -> ReceiveQueue -> QueueStatus -> m ()
-  setSndQueueStatus :: s -> SendQueue -> QueueStatus -> m ()
-  createMsg :: s -> ConnAlias -> QueueDirection -> AgentMsgId -> AMessage -> m ()
-  getLastMsg :: s -> ConnAlias -> QueueDirection -> m MessageDelivery
-  getMsg :: s -> ConnAlias -> QueueDirection -> AgentMsgId -> m MessageDelivery
-  updateMsgStatus :: s -> ConnAlias -> QueueDirection -> AgentMsgId -> m ()
-  deleteMsg :: s -> ConnAlias -> QueueDirection -> AgentMsgId -> m ()
+type ExternalSndId = Integer
+
+type ExternalSndTs = UTCTime
+
+type BrokerId = MsgId
+
+type BrokerTs = UTCTime
+
+data RcvStatus
+  = Received
+  | AcknowledgedToBroker
+  | AcknowledgedToSender
+  deriving (Eq, Show)
+
+type AckBrokerTs = UTCTime
+
+type AckSenderTs = UTCTime
+
+-- | A message sent by the agent to a recipient.
+data SndMsg = SndMsg
+  { msgBase :: MsgBase,
+    -- | Id of the message sent / to be sent, as in its number in order of sending.
+    internalSndId :: InternalSndId,
+    sndStatus :: SndStatus,
+    -- | Timestamp of the message received by broker, corresponds to `Sent` status.
+    sentTs :: SentTs,
+    -- | Timestamp of the message received by recipient, corresponds to `Delivered` status.
+    deliveredTs :: DeliveredTs
+  }
+  deriving (Eq, Show)
+
+type InternalSndId = Int64
+
+data SndStatus
+  = Created
+  | Sent
+  | Delivered
+  deriving (Eq, Show)
+
+type SentTs = UTCTime
+
+type DeliveredTs = UTCTime
+
+-- | Base message data independent of direction.
+data MsgBase = MsgBase
+  { connAlias :: ConnAlias,
+    -- | Monotonically increasing id of a message per connection, internal to the agent.
+    -- Preserves ordering between both received and sent messages.
+    internalId :: InternalId,
+    internalTs :: InternalTs,
+    msgBody :: MsgBody
+  }
+  deriving (Eq, Show)
+
+type InternalId = Int64
+
+type InternalTs = UTCTime
+
+-- * Store errors
+
+data StoreError
+  = SEInternal
+  | SENotFound
+  | SEBadConn
+  | SEBadConnType ConnType
+  | SEBadQueueStatus
+  | SEBadQueueDirection
+  | SENotImplemented -- TODO remove
+  deriving (Eq, Show, Exception)

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

@@ -16,9 +16,9 @@ import qualified Database.SQLite.Simple as DB
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite.Schema (createSchema)
 import Simplex.Messaging.Agent.Store.SQLite.Util
-import Simplex.Messaging.Agent.Store.Types
 import Simplex.Messaging.Agent.Transmission
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Types (MsgBody)
 import Simplex.Messaging.Util (liftIOEither)
 
 data SQLiteStore = SQLiteStore
@@ -78,9 +78,6 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
     liftIOEither $
       updateSndConnWithRcvQueue dbConn connAlias rcvQueue
 
-  removeSndAuth :: SQLiteStore -> ConnAlias -> m ()
-  removeSndAuth _st _connAlias = throwError SENotImplemented
-
   setRcvQueueStatus :: SQLiteStore -> ReceiveQueue -> QueueStatus -> m ()
   setRcvQueueStatus SQLiteStore {dbConn} rcvQueue status =
     liftIO $
@@ -91,24 +88,15 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
     liftIO $
       updateSndQueueStatus dbConn sndQueue status
 
-  createMsg :: SQLiteStore -> ConnAlias -> QueueDirection -> AgentMsgId -> AMessage -> m ()
-  createMsg SQLiteStore {dbConn} connAlias qDirection agentMsgId aMsg =
+  createRcvMsg :: SQLiteStore -> ConnAlias -> MsgBody -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m ()
+  createRcvMsg SQLiteStore {dbConn} connAlias msgBody externalSndId externalSndTs brokerId brokerTs =
     liftIOEither $
-      insertMsg dbConn connAlias agentMsgId aMsg
-    where
-      insertMsg = case qDirection of
-        RCV -> insertRcvMsg
-        SND -> insertSndMsg
+      insertRcvMsg dbConn connAlias msgBody externalSndId externalSndTs brokerId brokerTs
 
-  getLastMsg :: SQLiteStore -> ConnAlias -> QueueDirection -> m MessageDelivery
-  getLastMsg _st _connAlias _dir = throwError SENotImplemented
+  createSndMsg :: SQLiteStore -> ConnAlias -> MsgBody -> m ()
+  createSndMsg SQLiteStore {dbConn} connAlias msgBody =
+    liftIOEither $
+      insertSndMsg dbConn connAlias msgBody
 
-  getMsg :: SQLiteStore -> ConnAlias -> QueueDirection -> AgentMsgId -> m MessageDelivery
-  getMsg _st _connAlias _dir _msgId = throwError SENotImplemented
-
-  -- ? missing status parameter?
-  updateMsgStatus :: SQLiteStore -> ConnAlias -> QueueDirection -> AgentMsgId -> m ()
-  updateMsgStatus _st _connAlias _dir _msgId = throwError SENotImplemented
-
-  deleteMsg :: SQLiteStore -> ConnAlias -> QueueDirection -> AgentMsgId -> m ()
-  deleteMsg _st _connAlias _dir _msgId = throwError SENotImplemented
+  getMsg :: SQLiteStore -> ConnAlias -> InternalId -> m Msg
+  getMsg _st _connAlias _id = throwError SENotImplemented

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

@@ -6,6 +6,20 @@ module Simplex.Messaging.Agent.Store.SQLite.Schema (createSchema) where
 import Database.SQLite.Simple (Connection, Query, execute_)
 import Database.SQLite.Simple.QQ (sql)
 
+createSchema :: Connection -> IO ()
+createSchema conn =
+  mapM_
+    (execute_ conn)
+    [ enableFKs,
+      servers,
+      rcvQueues,
+      sndQueues,
+      connections,
+      messages,
+      rcvMessages,
+      sndMessages
+    ]
+
 enableFKs :: Query
 enableFKs = "PRAGMA foreign_keys = ON;"
 
@@ -28,17 +42,17 @@ rcvQueues =
       host TEXT NOT NULL,
       port TEXT NOT NULL,
       rcv_id BLOB NOT NULL,
-      conn_alias TEXT NOT NULL,
+      conn_alias BLOB NOT NULL,
       rcv_private_key BLOB NOT NULL,
       snd_id BLOB,
       snd_key BLOB,
       decrypt_key BLOB NOT NULL,
       verify_key BLOB,
       status TEXT NOT NULL,
-      PRIMARY KEY(host, port, rcv_id),
-      FOREIGN KEY(host, port) REFERENCES servers(host, port),
-      FOREIGN KEY(conn_alias)
-        REFERENCES connections(conn_alias)
+      PRIMARY KEY (host, port, rcv_id),
+      FOREIGN KEY (host, port) REFERENCES servers (host, port),
+      FOREIGN KEY (conn_alias)
+        REFERENCES connections (conn_alias)
         ON DELETE CASCADE
         DEFERRABLE INITIALLY DEFERRED,
       UNIQUE (host, port, snd_id)
@@ -52,15 +66,15 @@ sndQueues =
       host TEXT NOT NULL,
       port TEXT NOT NULL,
       snd_id BLOB NOT NULL,
-      conn_alias TEXT NOT NULL,
+      conn_alias BLOB NOT NULL,
       snd_private_key BLOB NOT NULL,
       encrypt_key BLOB NOT NULL,
       sign_key BLOB NOT NULL,
       status TEXT NOT NULL,
-      PRIMARY KEY(host, port, snd_id),
-      FOREIGN KEY(host, port) REFERENCES servers(host, port),
-      FOREIGN KEY(conn_alias)
-        REFERENCES connections(conn_alias)
+      PRIMARY KEY (host, port, snd_id),
+      FOREIGN KEY (host, port) REFERENCES servers (host, port),
+      FOREIGN KEY (conn_alias)
+        REFERENCES connections (conn_alias)
         ON DELETE CASCADE
         DEFERRABLE INITIALLY DEFERRED
     ) WITHOUT ROWID;
@@ -70,16 +84,19 @@ connections :: Query
 connections =
   [sql|
     CREATE TABLE IF NOT EXISTS connections(
-      conn_alias TEXT NOT NULL,
+      conn_alias BLOB NOT NULL,
       rcv_host TEXT,
       rcv_port TEXT,
       rcv_id BLOB,
       snd_host TEXT,
       snd_port TEXT,
       snd_id BLOB,
-      PRIMARY KEY(conn_alias),
-      FOREIGN KEY(rcv_host, rcv_port, rcv_id) REFERENCES rcv_queues(host, port, rcv_id),
-      FOREIGN KEY(snd_host, snd_port, snd_id) REFERENCES snd_queues(host, port, snd_id)
+      last_internal_msg_id INTEGER NOT NULL,
+      last_internal_rcv_msg_id INTEGER NOT NULL,
+      last_internal_snd_msg_id INTEGER NOT NULL,
+      PRIMARY KEY (conn_alias),
+      FOREIGN KEY (rcv_host, rcv_port, rcv_id) REFERENCES rcv_queues (host, port, rcv_id),
+      FOREIGN KEY (snd_host, snd_port, snd_id) REFERENCES snd_queues (host, port, snd_id)
     ) WITHOUT ROWID;
   |]
 
@@ -87,17 +104,61 @@ messages :: Query
 messages =
   [sql|
     CREATE TABLE IF NOT EXISTS messages(
-      agent_msg_id INTEGER NOT NULL,
-      conn_alias TEXT NOT NULL,
-      timestamp TEXT NOT NULL,
-      message BLOB NOT NULL,
-      direction TEXT NOT NULL,
-      msg_status TEXT NOT NULL,
-      PRIMARY KEY(agent_msg_id, conn_alias),
-      FOREIGN KEY(conn_alias) REFERENCES connections(conn_alias)
+      conn_alias BLOB NOT NULL,
+      internal_id INTEGER NOT NULL,
+      internal_ts TEXT NOT NULL,
+      internal_rcv_id INTEGER,
+      internal_snd_id INTEGER,
+      body TEXT NOT NULL,
+      PRIMARY KEY (conn_alias, internal_id),
+      FOREIGN KEY (conn_alias)
+        REFERENCES connections (conn_alias)
+        ON DELETE CASCADE,
+      FOREIGN KEY (conn_alias, internal_rcv_id)
+        REFERENCES rcv_messages (conn_alias, internal_rcv_id)
+        ON DELETE CASCADE
+        DEFERRABLE INITIALLY DEFERRED,
+      FOREIGN KEY (conn_alias, internal_snd_id)
+        REFERENCES snd_messages (conn_alias, internal_snd_id)
+        ON DELETE CASCADE
+        DEFERRABLE INITIALLY DEFERRED
     ) WITHOUT ROWID;
   |]
 
-createSchema :: Connection -> IO ()
-createSchema conn =
-  mapM_ (execute_ conn) [enableFKs, servers, rcvQueues, sndQueues, connections, messages]
+rcvMessages :: Query
+rcvMessages =
+  [sql|
+    CREATE TABLE IF NOT EXISTS rcv_messages(
+      conn_alias BLOB NOT NULL,
+      internal_rcv_id INTEGER NOT NULL,
+      internal_id INTEGER NOT NULL,
+      external_snd_id INTEGER NOT NULL,
+      external_snd_ts TEXT NOT NULL,
+      broker_id BLOB NOT NULL,
+      broker_ts TEXT NOT NULL,
+      rcv_status TEXT NOT NULL,
+      ack_brocker_ts TEXT,
+      ack_sender_ts TEXT,
+      PRIMARY KEY (conn_alias, internal_rcv_id),
+      FOREIGN KEY (conn_alias, internal_id)
+        REFERENCES messages (conn_alias, internal_id)
+        ON DELETE CASCADE
+    ) WITHOUT ROWID;
+  |]
+
+sndMessages :: Query
+sndMessages =
+  [sql|
+    CREATE TABLE IF NOT EXISTS snd_messages(
+      conn_alias BLOB NOT NULL,
+      internal_snd_id INTEGER NOT NULL,
+      internal_id INTEGER NOT NULL,
+      snd_status TEXT NOT NULL,
+      sent_ts TEXT,
+      delivered_ts TEXT,
+      PRIMARY KEY (conn_alias, internal_snd_id),
+      FOREIGN KEY (conn_alias, internal_id)
+        REFERENCES messages (conn_alias, internal_id)
+        ON DELETE CASCADE
+    ) WITHOUT ROWID;
+  |]

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

@@ -25,6 +25,7 @@ where
 import Control.Monad.Except (MonadIO (liftIO))
 import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
+import Data.Text.Encoding (decodeUtf8)
 import Data.Time (getCurrentTime)
 import Database.SQLite.Simple as DB
 import Database.SQLite.Simple.FromField
@@ -34,12 +35,14 @@ import Database.SQLite.Simple.QQ (sql)
 import Database.SQLite.Simple.ToField (ToField (..))
 import Network.Socket (HostName, ServiceName)
 import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.Types
 import Simplex.Messaging.Agent.Transmission
 import Simplex.Messaging.Protocol as SMP (RecipientId)
+import Simplex.Messaging.Types (MsgBody)
 import Text.Read (readMaybe)
 import qualified UnliftIO.Exception as E
 
+-- * Auxiliary helpers
+
 -- ? replace with ToField? - it's easy to forget to use this
 serializePort_ :: Maybe ServiceName -> ServiceName
 serializePort_ = fromMaybe "_"
@@ -52,7 +55,9 @@ instance ToField QueueStatus where toField = toField . show
 
 instance FromField QueueStatus where fromField = fromFieldToReadable_
 
-instance ToField QueueDirection where toField = toField . show
+instance ToField RcvStatus where toField = toField . show
+
+instance ToField SndStatus where toField = toField . show
 
 fromFieldToReadable_ :: forall a. (Read a, E.Typeable a) => Field -> Ok a
 fromFieldToReadable_ = \case
@@ -74,6 +79,8 @@ instance (FromField a, FromField b, FromField c, FromField d, FromField e,
                          <*> field
 {- ORMOLU_ENABLE -}
 
+-- * createRcvConn and createSndConn helpers
+
 createRcvQueueAndConn :: DB.Connection -> ReceiveQueue -> IO ()
 createRcvQueueAndConn dbConn rcvQueue =
   DB.withTransaction dbConn $ do
@@ -145,9 +152,11 @@ insertRcvConnectionQuery_ :: Query
 insertRcvConnectionQuery_ =
   [sql|
     INSERT INTO connections
-      ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id)
+      ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id,
+        last_internal_msg_id, last_internal_rcv_msg_id, last_internal_snd_msg_id)
     VALUES
-      (:conn_alias,:rcv_host,:rcv_port,:rcv_id,     NULL,     NULL,   NULL);
+      (:conn_alias,:rcv_host,:rcv_port,:rcv_id,     NULL,     NULL,   NULL,
+                           0,                        0,                        0);
   |]
 
 insertSndQueue_ :: DB.Connection -> SendQueue -> IO ()
@@ -187,11 +196,15 @@ insertSndConnectionQuery_ :: Query
 insertSndConnectionQuery_ =
   [sql|
     INSERT INTO connections
-      ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id)
+      ( conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id,
+        last_internal_msg_id, last_internal_rcv_msg_id, last_internal_snd_msg_id)
     VALUES
-      (:conn_alias,     NULL,     NULL,   NULL,:snd_host,:snd_port,:snd_id);
+      (:conn_alias,     NULL,     NULL,   NULL,:snd_host,:snd_port,:snd_id,
+                           0,                        0,                        0);
   |]
 
+-- * getConn helpers
+
 retrieveConnQueues :: DB.Connection -> ConnAlias -> IO (Maybe ReceiveQueue, Maybe SendQueue)
 retrieveConnQueues dbConn connAlias =
   DB.withTransaction -- Avoid inconsistent state between queue reads
@@ -254,9 +267,8 @@ retrieveSndQueueByConnAliasQuery_ =
     WHERE q.conn_alias = :conn_alias;
   |]
 
--- ? make server an argument and pass it for the queue instead of joining with 'servers'?
--- ? the downside would be the queue having an outdated 'key_hash' if it has changed,
--- ? but maybe it's unwanted behavior?
+-- * getRcvQueue helpers
+
 retrieveRcvQueue :: DB.Connection -> HostName -> Maybe ServiceName -> SMP.RecipientId -> IO (Maybe ReceiveQueue)
 retrieveRcvQueue dbConn host port rcvId = do
   r <-
@@ -281,6 +293,8 @@ retrieveRcvQueueQuery_ =
     WHERE q.host = :host AND q.port = :port AND q.rcv_id = :rcv_id;
   |]
 
+-- * deleteConn helper
+
 deleteConnCascade :: DB.Connection -> ConnAlias -> IO ()
 deleteConnCascade dbConn connAlias =
   DB.executeNamed
@@ -288,7 +302,8 @@ deleteConnCascade dbConn connAlias =
     "DELETE FROM connections WHERE conn_alias = :conn_alias;"
     [":conn_alias" := connAlias]
 
--- ? rewrite with ExceptT?
+-- * upgradeRcvConnToDuplex helpers
+
 updateRcvConnWithSndQueue :: DB.Connection -> ConnAlias -> SendQueue -> IO (Either StoreError ())
 updateRcvConnWithSndQueue dbConn connAlias sndQueue =
   DB.withTransaction dbConn $ do
@@ -319,7 +334,8 @@ updateConnWithSndQueueQuery_ =
     WHERE conn_alias = :conn_alias;
   |]
 
--- ? rewrite with ExceptT?
+-- * upgradeSndConnToDuplex helpers
+
 updateSndConnWithRcvQueue :: DB.Connection -> ConnAlias -> ReceiveQueue -> IO (Either StoreError ())
 updateSndConnWithRcvQueue dbConn connAlias rcvQueue =
   DB.withTransaction dbConn $ do
@@ -350,6 +366,8 @@ updateConnWithRcvQueueQuery_ =
     WHERE conn_alias = :conn_alias;
   |]
 
+-- * setRcvQueueStatus helpers
+
 -- ? throw error if queue doesn't exist?
 updateRcvQueueStatus :: DB.Connection -> ReceiveQueue -> QueueStatus -> IO ()
 updateRcvQueueStatus dbConn ReceiveQueue {rcvId, server = SMPServer {host, port}} status =
@@ -366,6 +384,8 @@ updateRcvQueueStatusQuery_ =
     WHERE host = :host AND port = :port AND rcv_id = :rcv_id;
   |]
 
+-- * setSndQueueStatus helpers
+
 -- ? throw error if queue doesn't exist?
 updateSndQueueStatus :: DB.Connection -> SendQueue -> QueueStatus -> IO ()
 updateSndQueueStatus dbConn SendQueue {sndId, server = SMPServer {host, port}} status =
@@ -382,50 +402,183 @@ updateSndQueueStatusQuery_ =
     WHERE host = :host AND port = :port AND snd_id = :snd_id;
   |]
 
--- ? rewrite with ExceptT?
-insertRcvMsg :: DB.Connection -> ConnAlias -> AgentMsgId -> AMessage -> IO (Either StoreError ())
-insertRcvMsg dbConn connAlias agentMsgId aMsg =
+-- * createRcvMsg helpers
+
+insertRcvMsg ::
+  DB.Connection ->
+  ConnAlias ->
+  MsgBody ->
+  ExternalSndId ->
+  ExternalSndTs ->
+  BrokerId ->
+  BrokerTs ->
+  IO (Either StoreError ())
+insertRcvMsg dbConn connAlias msgBody externalSndId externalSndTs brokerId brokerTs =
   DB.withTransaction dbConn $ do
     queues <- retrieveConnQueues_ dbConn connAlias
     case queues of
       (Just _rcvQ, _) -> do
-        insertMsg_ dbConn connAlias RCV agentMsgId aMsg
+        (lastInternalId, lastInternalRcvId) <- retrieveLastInternalIdsRcv_ dbConn connAlias
+        let internalId = lastInternalId + 1
+        let internalRcvId = lastInternalRcvId + 1
+        insertRcvMsgBase_ dbConn connAlias internalId internalRcvId msgBody
+        insertRcvMsgDetails_ dbConn connAlias internalRcvId internalId externalSndId externalSndTs brokerId brokerTs
+        updateLastInternalIdsRcv_ dbConn connAlias internalId internalRcvId
         return $ Right ()
-      (Nothing, Just _sndQ) -> return $ Left SEBadQueueDirection
+      (Nothing, Just _sndQ) -> return $ Left (SEBadConnType CSend)
       _ -> return $ Left SEBadConn
 
--- ? rewrite with ExceptT?
-insertSndMsg :: DB.Connection -> ConnAlias -> AgentMsgId -> AMessage -> IO (Either StoreError ())
-insertSndMsg dbConn connAlias agentMsgId aMsg =
+retrieveLastInternalIdsRcv_ :: DB.Connection -> ConnAlias -> IO (InternalId, InternalRcvId)
+retrieveLastInternalIdsRcv_ dbConn connAlias = do
+  [(lastInternalId, lastInternalRcvId)] <-
+    DB.queryNamed
+      dbConn
+      [sql|
+        SELECT last_internal_msg_id, last_internal_rcv_msg_id
+        FROM connections
+        WHERE conn_alias = :conn_alias;
+      |]
+      [":conn_alias" := connAlias]
+  return (lastInternalId, lastInternalRcvId)
+
+insertRcvMsgBase_ :: DB.Connection -> ConnAlias -> InternalId -> InternalRcvId -> MsgBody -> IO ()
+insertRcvMsgBase_ dbConn connAlias internalId internalRcvId msgBody = do
+  internalTs <- liftIO getCurrentTime
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO messages
+        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body)
+      VALUES
+        (:conn_alias,:internal_id,:internal_ts,:internal_rcv_id,            NULL,:body);
+    |]
+    [ ":conn_alias" := connAlias,
+      ":internal_id" := internalId,
+      ":internal_ts" := internalTs,
+      ":internal_rcv_id" := internalRcvId,
+      ":body" := decodeUtf8 msgBody
+    ]
+
+insertRcvMsgDetails_ ::
+  DB.Connection ->
+  ConnAlias ->
+  InternalRcvId ->
+  InternalId ->
+  ExternalSndId ->
+  ExternalSndTs ->
+  BrokerId ->
+  BrokerTs ->
+  IO ()
+insertRcvMsgDetails_ dbConn connAlias internalRcvId internalId externalSndId externalSndTs brokerId brokerTs =
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO rcv_messages
+        ( conn_alias, internal_rcv_id, internal_id, external_snd_id, external_snd_ts,
+          broker_id, broker_ts, rcv_status, ack_brocker_ts, ack_sender_ts)
+      VALUES
+        (:conn_alias,:internal_rcv_id,:internal_id,:external_snd_id,:external_snd_ts,
+         :broker_id,:broker_ts,:rcv_status,           NULL,          NULL);
+    |]
+    [ ":conn_alias" := connAlias,
+      ":internal_rcv_id" := internalRcvId,
+      ":internal_id" := internalId,
+      ":external_snd_id" := externalSndId,
+      ":external_snd_ts" := externalSndTs,
+      ":broker_id" := brokerId,
+      ":broker_ts" := brokerTs,
+      ":rcv_status" := Received
+    ]
+
+updateLastInternalIdsRcv_ :: DB.Connection -> ConnAlias -> InternalId -> InternalRcvId -> IO ()
+updateLastInternalIdsRcv_ dbConn connAlias newInternalId newInternalRcvId =
+  DB.executeNamed
+    dbConn
+    [sql|
+      UPDATE connections
+      SET last_internal_msg_id = :last_internal_msg_id, last_internal_rcv_msg_id = :last_internal_rcv_msg_id
+      WHERE conn_alias = :conn_alias;
+    |]
+    [ ":last_internal_msg_id" := newInternalId,
+      ":last_internal_rcv_msg_id" := newInternalRcvId,
+      ":conn_alias" := connAlias
+    ]
+
+-- * createSndMsg helpers
+
+insertSndMsg :: DB.Connection -> ConnAlias -> MsgBody -> IO (Either StoreError ())
+insertSndMsg dbConn connAlias msgBody =
   DB.withTransaction dbConn $ do
     queues <- retrieveConnQueues_ dbConn connAlias
     case queues of
       (_, Just _sndQ) -> do
-        insertMsg_ dbConn connAlias SND agentMsgId aMsg
+        (lastInternalId, lastInternalSndId) <- retrieveLastInternalIdsSnd_ dbConn connAlias
+        let internalId = lastInternalId + 1
+        let internalSndId = lastInternalSndId + 1
+        insertSndMsgBase_ dbConn connAlias internalId internalSndId msgBody
+        insertSndMsgDetails_ dbConn connAlias internalSndId internalId
+        updateLastInternalIdsSnd_ dbConn connAlias internalId internalSndId
         return $ Right ()
-      (Just _rcvQ, Nothing) -> return $ Left SEBadQueueDirection
+      (Just _rcvQ, Nothing) -> return $ Left (SEBadConnType CReceive)
       _ -> return $ Left SEBadConn
 
--- TODO add parser and serializer for DeliveryStatus? Pass DeliveryStatus?
-insertMsg_ :: DB.Connection -> ConnAlias -> QueueDirection -> AgentMsgId -> AMessage -> IO ()
-insertMsg_ dbConn connAlias qDirection agentMsgId aMsg = do
-  let msg = serializeAgentMessage aMsg
-  ts <- liftIO getCurrentTime
+retrieveLastInternalIdsSnd_ :: DB.Connection -> ConnAlias -> IO (InternalId, InternalSndId)
+retrieveLastInternalIdsSnd_ dbConn connAlias = do
+  [(lastInternalId, lastInternalSndId)] <-
+    DB.queryNamed
+      dbConn
+      [sql|
+        SELECT last_internal_msg_id, last_internal_snd_msg_id
+        FROM connections
+        WHERE conn_alias = :conn_alias;
+      |]
+      [":conn_alias" := connAlias]
+  return (lastInternalId, lastInternalSndId)
+
+insertSndMsgBase_ :: DB.Connection -> ConnAlias -> InternalId -> InternalSndId -> MsgBody -> IO ()
+insertSndMsgBase_ dbConn connAlias internalId internalSndId msgBody = do
+  internalTs <- liftIO getCurrentTime
   DB.executeNamed
     dbConn
-    insertMsgQuery_
-    [ ":agent_msg_id" := agentMsgId,
-      ":conn_alias" := connAlias,
-      ":timestamp" := ts,
-      ":message" := msg,
-      ":direction" := qDirection
+    [sql|
+      INSERT INTO messages
+        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body)
+      VALUES
+        (:conn_alias,:internal_id,:internal_ts,            NULL,:internal_snd_id,:body);
+    |]
+    [ ":conn_alias" := connAlias,
+      ":internal_id" := internalId,
+      ":internal_ts" := internalTs,
+      ":internal_snd_id" := internalSndId,
+      ":body" := decodeUtf8 msgBody
     ]
 
-insertMsgQuery_ :: Query
-insertMsgQuery_ =
-  [sql|
-    INSERT INTO messages
-      ( agent_msg_id, conn_alias, timestamp, message, direction, msg_status)
-    VALUES
-      (:agent_msg_id,:conn_alias,:timestamp,:message,:direction,"MDTransmitted");
-  |]
+insertSndMsgDetails_ :: DB.Connection -> ConnAlias -> InternalSndId -> InternalId -> IO ()
+insertSndMsgDetails_ dbConn connAlias internalSndId internalId =
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO snd_messages
+        ( conn_alias, internal_snd_id, internal_id, snd_status, sent_ts, delivered_ts)
+      VALUES
+        (:conn_alias,:internal_snd_id,:internal_id,:snd_status,    NULL,         NULL);
+    |]
+    [ ":conn_alias" := connAlias,
+      ":internal_snd_id" := internalSndId,
+      ":internal_id" := internalId,
+      ":snd_status" := Created
+    ]
+
+updateLastInternalIdsSnd_ :: DB.Connection -> ConnAlias -> InternalId -> InternalSndId -> IO ()
+updateLastInternalIdsSnd_ dbConn connAlias newInternalId newInternalSndId =
+  DB.executeNamed
+    dbConn
+    [sql|
+      UPDATE connections
+      SET last_internal_msg_id = :last_internal_msg_id, last_internal_snd_msg_id = :last_internal_snd_msg_id
+      WHERE conn_alias = :conn_alias;
+    |]
+    [ ":last_internal_msg_id" := newInternalId,
+      ":last_internal_snd_msg_id" := newInternalSndId,
+      ":conn_alias" := connAlias
+    ]

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

@@ -1,21 +0,0 @@
-{-# LANGUAGE DeriveAnyClass #-}
-
-module Simplex.Messaging.Agent.Store.Types
-  ( ConnType (..),
-    StoreError (..),
-  )
-where
-
-import Control.Exception (Exception)
-
-data ConnType = CSend | CReceive | CDuplex deriving (Eq, Show)
-
-data StoreError
-  = SEInternal
-  | SENotFound
-  | SEBadConn
-  | SEBadConnType ConnType
-  | SEBadQueueStatus
-  | SEBadQueueDirection
-  | SENotImplemented -- TODO remove
-  deriving (Eq, Show, Exception)

src/Simplex/Messaging/Agent/Transmission.hs

@@ -27,7 +27,6 @@ import Data.Type.Equality
 import Data.Typeable ()
 import Network.Socket
 import Numeric.Natural
-import Simplex.Messaging.Agent.Store.Types
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Parsers
 import qualified Simplex.Messaging.Protocol as SMP
@@ -246,7 +245,7 @@ data AgentErrorType
   | SMP ErrorType
   | CRYPTO C.CryptoError
   | SIZE
-  | STORE StoreError
+  | STORE
   | INTERNAL -- etc. TODO SYNTAX Natural
   deriving (Eq, Show, Exception)
 

tests/AgentTests/SQLiteTests.hs

@@ -7,12 +7,13 @@ module AgentTests.SQLiteTests (storeTests) where
 
 import Control.Monad.Except (ExceptT, runExceptT)
 import qualified Crypto.PubKey.RSA as R
+import Data.Text.Encoding (encodeUtf8)
+import Data.Time
 import Data.Word (Word32)
 import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.QQ (sql)
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
-import Simplex.Messaging.Agent.Store.Types
 import Simplex.Messaging.Agent.Transmission
 import qualified Simplex.Messaging.Crypto as C
 import System.Random (Random (randomIO))
@@ -28,7 +29,7 @@ withStore = before createStore . after removeStore
     createStore :: IO SQLiteStore
     createStore = do
       -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
-      -- IO operations on multiple similarly named files; error specific to some environments
+      -- IO operations on multiple similarly named files; error seems to be environment specific
       r <- randomIO :: IO Word32
       newSQLiteStore $ testDB <> show r
 
@@ -52,36 +53,35 @@ storeTests = withStore do
     describe "createSndConn" testCreateSndConn
     describe "getRcvQueue" testGetRcvQueue
     describe "deleteConn" do
-      describe "Receive connection" testDeleteConnReceive
-      describe "Send connection" testDeleteConnSend
-      describe "Duplex connection" testDeleteConnDuplex
+      describe "rcv" testDeleteRcvConn
+      describe "snd" testDeleteSndConn
+      describe "duplex" testDeleteDuplexConn
     describe "upgradeRcvConnToDuplex" testUpgradeRcvConnToDuplex
     describe "upgradeSndConnToDuplex" testUpgradeSndConnToDuplex
-    describe "Set queue status" do
+    describe "set queue status" do
       describe "setRcvQueueStatus" testSetRcvQueueStatus
       describe "setSndQueueStatus" testSetSndQueueStatus
-      describe "Duplex connection" testSetQueueStatusConnDuplex
-      xdescribe "Nonexistent send queue" testSetNonexistentSendQueueStatus
-      xdescribe "Nonexistent receive queue" testSetNonexistentReceiveQueueStatus
-    describe "createMsg" do
-      describe "A_MSG in RCV direction" testCreateMsgRcv
-      describe "A_MSG in SND direction" testCreateMsgSnd
-      describe "HELLO message" testCreateMsgHello
-      describe "REPLY message" testCreateMsgReply
-      describe "Bad queue direction - SND" testCreateMsgBadDirectionSnd
-      describe "Bad queue direction - RCV" testCreateMsgBadDirectionRcv
+      describe "duplex connection" testSetQueueStatusDuplex
+      xdescribe "rcv queue doesn't exist" testSetRcvQueueStatusNoQueue
+      xdescribe "snd queue doesn't exist" testSetSndQueueStatusNoQueue
+    describe "createRcvMsg" do
+      describe "rcv queue exists" testCreateRcvMsg
+      describe "rcv queue doesn't exist" testCreateRcvMsgNoQueue
+    describe "createSndMsg" do
+      describe "snd queue exists" testCreateSndMsg
+      describe "snd queue doesn't exist" testCreateSndMsgNoQueue
 
 testForeignKeysEnabled :: SpecWith SQLiteStore
 testForeignKeysEnabled = do
   it "should throw error if foreign keys are enabled" $ \store -> do
-    let inconsistent_query =
+    let inconsistentQuery =
           [sql|
             INSERT INTO connections
               (conn_alias, rcv_host, rcv_port, rcv_id, snd_host, snd_port, snd_id)
             VALUES
               ("conn1", "smp.simplex.im", "5223", "1234", "smp.simplex.im", "5223", "2345");
           |]
-    DB.execute_ (dbConn store) inconsistent_query
+    DB.execute_ (dbConn store) inconsistentQuery
       `shouldThrow` (\e -> DB.sqlError e == DB.ErrorConstraint)
 
 rcvQueue1 :: ReceiveQueue
@@ -110,18 +110,6 @@ sndQueue1 =
       status = New
     }
 
--- sndQueue2 :: SendQueue
--- sndQueue2 =
---           SendQueue
---             { server = SMPServer "smp.simplex.im" (Just "5223") (Just "1234"),
---               sndId = "1234",
---               connAlias = "conn1",
---               sndPrivateKey = "abcd",
---               encryptKey = "dcba",
---               signKey = "edcb",
---               status = New
---             }
-
 testCreateRcvConn :: SpecWith SQLiteStore
 testCreateRcvConn = do
   it "should create receive connection and add send queue" $ \store -> do
@@ -148,7 +136,7 @@ testCreateSndConn = do
 
 testGetRcvQueue :: SpecWith SQLiteStore
 testGetRcvQueue = do
-  it "should get receive queue and conn alias" $ \store -> do
+  it "should get receive queue" $ \store -> do
     let smpServer = SMPServer "smp.simplex.im" (Just "5223") (Just "1234")
     let recipientId = "1234"
     createRcvConn store rcvQueue1
@@ -156,8 +144,8 @@ testGetRcvQueue = do
     getRcvQueue store smpServer recipientId
       `returnsResult` rcvQueue1
 
-testDeleteConnReceive :: SpecWith SQLiteStore
-testDeleteConnReceive = do
+testDeleteRcvConn :: SpecWith SQLiteStore
+testDeleteRcvConn = do
   it "should create receive connection and delete it" $ \store -> do
     createRcvConn store rcvQueue1
       `returnsResult` ()
@@ -169,8 +157,8 @@ testDeleteConnReceive = do
     getConn store "conn1"
       `throwsError` SEBadConn
 
-testDeleteConnSend :: SpecWith SQLiteStore
-testDeleteConnSend = do
+testDeleteSndConn :: SpecWith SQLiteStore
+testDeleteSndConn = do
   it "should create send connection and delete it" $ \store -> do
     createSndConn store sndQueue1
       `returnsResult` ()
@@ -182,8 +170,8 @@ testDeleteConnSend = do
     getConn store "conn1"
       `throwsError` SEBadConn
 
-testDeleteConnDuplex :: SpecWith SQLiteStore
-testDeleteConnDuplex = do
+testDeleteDuplexConn :: SpecWith SQLiteStore
+testDeleteDuplexConn = do
   it "should create duplex connection and delete it" $ \store -> do
     createRcvConn store rcvQueue1
       `returnsResult` ()
@@ -267,8 +255,8 @@ testSetSndQueueStatus = do
     getConn store "conn1"
       `returnsResult` SomeConn SCSend (SendConnection "conn1" sndQueue1 {status = Confirmed})
 
-testSetQueueStatusConnDuplex :: SpecWith SQLiteStore
-testSetQueueStatusConnDuplex = do
+testSetQueueStatusDuplex :: SpecWith SQLiteStore
+testSetQueueStatusDuplex = do
   it "should update statuses of receive and send queues in duplex connection" $ \store -> do
     createRcvConn store rcvQueue1
       `returnsResult` ()
@@ -288,83 +276,54 @@ testSetQueueStatusConnDuplex = do
         ( DuplexConnection "conn1" rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed}
         )
 
-testSetNonexistentSendQueueStatus :: SpecWith SQLiteStore
-testSetNonexistentSendQueueStatus = do
-  it "should throw error on attempt to update status of nonexistent send queue" $ \store -> do
-    setSndQueueStatus store sndQueue1 Confirmed
-      `throwsError` SEInternal
-
-testSetNonexistentReceiveQueueStatus :: SpecWith SQLiteStore
-testSetNonexistentReceiveQueueStatus = do
+testSetRcvQueueStatusNoQueue :: SpecWith SQLiteStore
+testSetRcvQueueStatusNoQueue = do
   it "should throw error on attempt to update status of nonexistent receive queue" $ \store -> do
     setRcvQueueStatus store rcvQueue1 Confirmed
       `throwsError` SEInternal
 
-testCreateMsgRcv :: SpecWith SQLiteStore
-testCreateMsgRcv = do
-  it "should create a message in RCV direction" $ \store -> do
+testSetSndQueueStatusNoQueue :: SpecWith SQLiteStore
+testSetSndQueueStatusNoQueue = do
+  it "should throw error on attempt to update status of nonexistent send queue" $ \store -> do
+    setSndQueueStatus store sndQueue1 Confirmed
+      `throwsError` SEInternal
+
+testCreateRcvMsg :: SpecWith SQLiteStore
+testCreateRcvMsg = do
+  it "should create a rcv message" $ \store -> do
     createRcvConn store rcvQueue1
       `returnsResult` ()
-    let msg = A_MSG "hello"
-    let msgId = 1
     -- TODO getMsg to check message
-    createMsg store "conn1" RCV msgId msg
+    let ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
+    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") 1 ts "1" ts
       `returnsResult` ()
 
-testCreateMsgSnd :: SpecWith SQLiteStore
-testCreateMsgSnd = do
-  it "should create a message in SND direction" $ \store -> do
+testCreateRcvMsgNoQueue :: SpecWith SQLiteStore
+testCreateRcvMsgNoQueue = do
+  it "should throw error on attempt to create a rcv message w/t a rcv queue" $ \store -> do
+    let ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
+    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") 1 ts "1" ts
+      `throwsError` SEBadConn
     createSndConn store sndQueue1
       `returnsResult` ()
-    let msg = A_MSG "hi"
-    let msgId = 1
-    -- TODO getMsg to check message
-    createMsg store "conn1" SND msgId msg
-      `returnsResult` ()
+    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") 1 ts "1" ts
+      `throwsError` SEBadConnType CSend
 
-testCreateMsgHello :: SpecWith SQLiteStore
-testCreateMsgHello = do
-  it "should create a HELLO message" $ \store -> do
-    createRcvConn store rcvQueue1
-      `returnsResult` ()
-    let verificationKey = C.PublicKey $ R.PublicKey 1 2 3
-    let am = AckMode On
-    let msg = HELLO verificationKey am
-    let msgId = 1
-    -- TODO getMsg to check message
-    createMsg store "conn1" RCV msgId msg
-      `returnsResult` ()
-
-testCreateMsgReply :: SpecWith SQLiteStore
-testCreateMsgReply = do
-  it "should create a REPLY message" $ \store -> do
-    createRcvConn store rcvQueue1
-      `returnsResult` ()
-    let smpServer = SMPServer "smp.simplex.im" (Just "5223") (Just "1234")
-    let senderId = "sender1"
-    let encryptionKey = C.PublicKey $ R.PublicKey 1 2 3
-    let msg = REPLY $ SMPQueueInfo smpServer senderId encryptionKey
-    let msgId = 1
-    -- TODO getMsg to check message
-    createMsg store "conn1" RCV msgId msg
-      `returnsResult` ()
-
-testCreateMsgBadDirectionSnd :: SpecWith SQLiteStore
-testCreateMsgBadDirectionSnd = do
-  it "should throw error on attempt to create a message in ineligible SND direction" $ \store -> do
-    createRcvConn store rcvQueue1
-      `returnsResult` ()
-    let msg = A_MSG "hello"
-    let msgId = 1
-    createMsg store "conn1" SND msgId msg
-      `throwsError` SEBadQueueDirection
-
-testCreateMsgBadDirectionRcv :: SpecWith SQLiteStore
-testCreateMsgBadDirectionRcv = do
-  it "should throw error on attempt to create a message in ineligible RCV direction" $ \store -> do
+testCreateSndMsg :: SpecWith SQLiteStore
+testCreateSndMsg = do
+  it "should create a snd message" $ \store -> do
     createSndConn store sndQueue1
       `returnsResult` ()
-    let msg = A_MSG "hello"
-    let msgId = 1
-    createMsg store "conn1" RCV msgId msg
-      `throwsError` SEBadQueueDirection
+    -- TODO getMsg to check message
+    createSndMsg store "conn1" (encodeUtf8 "Hello world!")
+      `returnsResult` ()
+
+testCreateSndMsgNoQueue :: SpecWith SQLiteStore
+testCreateSndMsgNoQueue = do
+  it "should throw error on attempt to create a snd message w/t a snd queue" $ \store -> do
+    createSndMsg store "conn1" (encodeUtf8 "Hello world!")
+      `throwsError` SEBadConn
+    createRcvConn store rcvQueue1
+      `returnsResult` ()
+    createSndMsg store "conn1" (encodeUtf8 "Hello world!")
+      `throwsError` SEBadConnType CReceive