Postgres POC (duplicated SQLite code)

CHANGELOG.md

@@ -1,15 +1,3 @@
-# 1.0.3
-
-SMP server:
-- Reduce server message queue quota to 128 messages.
-
-SMP agent:
-- Add "yes to migrations" option.
-- Make new SMP client attempt to reconnect on network error.
-- Reduce connection handshake expiration to 2 days.
-
-JSON encoding of types used in simplex-chat, some other minor adjustments.
-
 # 1.0.2
 
 General:

apps/smp-server/Main.hs

@@ -277,8 +277,8 @@ runServer IniOptions {enableStoreLog, port, enableWebsockets} = do
       ServerConfig
         { transports = (port, transport @TLS) : [("80", transport @WS) | enableWebsockets],
           tbqSize = 16,
-          serverTbqSize = 64,
-          msgQueueQuota = 128,
+          serverTbqSize = 128,
+          msgQueueQuota = 256,
           queueIdBytes = 24,
           msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
           caCertificateFile = caCrtFile,

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 1.0.3
+version: 1.0.2
 synopsis: SimpleXMQ message broker
 description:  |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        1.0.3
+version:        1.0.2
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -31,12 +31,16 @@ library
   exposed-modules:
       Simplex.Messaging.Agent
       Simplex.Messaging.Agent.Client
+      Simplex.Messaging.Agent.Env.Postgres
       Simplex.Messaging.Agent.Env.SQLite
       Simplex.Messaging.Agent.Protocol
       Simplex.Messaging.Agent.QueryString
       Simplex.Messaging.Agent.RetryInterval
       Simplex.Messaging.Agent.Server
       Simplex.Messaging.Agent.Store
+      Simplex.Messaging.Agent.Store.Postgres
+      Simplex.Messaging.Agent.Store.Postgres.Migrations
+      Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial
       Simplex.Messaging.Agent.Store.SQLite
       Simplex.Messaging.Agent.Store.SQLite.Migrations
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial

src/Simplex/Messaging/Agent.hs

@@ -305,6 +305,16 @@ subscribeConnection' c connId =
     SomeConn _ (DuplexConnection _ rq sq) -> do
       resumeMsgDelivery c connId sq
       subscribeQueue c rq connId
+      case status (sq :: SndQueue) of
+        Confirmed -> do
+          -- TODO if there is no confirmation saved, just update the status without securing the queue
+          AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-
+            withStore (`getAcceptedConfirmation` connId)
+          secureQueue c rq senderKey
+          withStore $ \st -> setRcvQueueStatus st rq Secured
+        Secured -> pure ()
+        Active -> pure ()
+        _ -> throwError $ INTERNAL "unexpected queue status"
     SomeConn _ (SndConnection _ sq) -> do
       resumeMsgDelivery c connId sq
       case status (sq :: SndQueue) of

src/Simplex/Messaging/Agent/Client.hs

@@ -37,7 +37,7 @@ module Simplex.Messaging.Agent.Client
 where
 
 import Control.Concurrent (forkIO)
-import Control.Concurrent.Async (Async, uninterruptibleCancel)
+import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
 import Control.Monad.Except
@@ -62,10 +62,8 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (bshow, liftEitherError, liftError, tryError)
+import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)
 import Simplex.Messaging.Version
-import System.Timeout (timeout)
-import UnliftIO (async)
 import UnliftIO.Exception (Exception, IOException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -78,13 +76,11 @@ data AgentClient = AgentClient
     msgQ :: TBQueue SMPServerTransmission,
     smpClients :: TVar (Map SMPServer SMPClientVar),
     subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
-    pendingSubscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
     subscrConns :: TVar (Map ConnId SMPServer),
     connMsgsQueued :: TVar (Map ConnId Bool),
     smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),
     smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),
     reconnections :: TVar [Async ()],
-    asyncClients :: TVar [Async ()],
     clientId :: Int,
     agentEnv :: Env,
     smpSubscriber :: Async (),
@@ -99,16 +95,14 @@ newAgentClient agentEnv = do
   msgQ <- newTBQueue qSize
   smpClients <- newTVar M.empty
   subscrSrvrs <- newTVar M.empty
-  pendingSubscrSrvrs <- newTVar M.empty
   subscrConns <- newTVar M.empty
   connMsgsQueued <- newTVar M.empty
   smpQueueMsgQueues <- newTVar M.empty
   smpQueueMsgDeliveries <- newTVar M.empty
   reconnections <- newTVar []
-  asyncClients <- newTVar []
   clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)
   lock <- newTMVar ()
-  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock}
+  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
 
 -- | Agent monad with MonadReader Env and MonadError AgentErrorType
 type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
@@ -139,39 +133,20 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
       pure smpVar
 
     waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    waitForSMPClient smpVar = do
-      SMPClientConfig {tcpTimeout} <- asks $ smpCfg . config
-      smpClient_ <- liftIO $ tcpTimeout `timeout` atomically (readTMVar smpVar)
-      liftEither $ case smpClient_ of
-        Just (Right smpClient) -> Right smpClient
-        Just (Left e) -> Left e
-        Nothing -> Left $ BROKER TIMEOUT
+    waitForSMPClient = liftIOEither . atomically . readTMVar
 
     newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    newSMPClient smpVar = tryConnectClient pure tryConnectAsync
-      where
-        tryConnectClient :: (SMPClient -> m a) -> m () -> m a
-        tryConnectClient successAction retryAction =
-          tryError connectClient >>= \r -> case r of
-            Right smp -> do
-              logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-              atomically $ putTMVar smpVar r
-              successAction smp
-            Left e -> do
-              if e == BROKER NETWORK || e == BROKER TIMEOUT
-                then retryAction
-                else atomically $ do
-                  putTMVar smpVar (Left e)
-                  modifyTVar smpClients $ M.delete srv
-              throwError e
-        tryConnectAsync :: m ()
-        tryConnectAsync = do
-          a <- async connectAsync
-          atomically $ modifyTVar (asyncClients c) (a :)
-        connectAsync :: m ()
-        connectAsync = do
-          ri <- asks $ reconnectInterval . config
-          withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop
+    newSMPClient smpVar =
+      tryError connectClient >>= \r -> case r of
+        Right smp -> do
+          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
+          atomically $ putTMVar smpVar r
+          pure smp
+        Left e -> do
+          atomically $ do
+            putTMVar smpVar r
+            modifyTVar smpClients $ M.delete srv
+          throwError e
 
     connectClient :: m SMPClient
     connectClient = do
@@ -194,43 +169,33 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
       cs <- M.lookup srv <$> readTVar (subscrSrvrs c)
       modifyTVar (subscrSrvrs c) $ M.delete srv
       modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs
-      mapM_ (modifyTVar (pendingSubscrSrvrs c) . addPendingSubs) cs
       return cs
       where
-        addPendingSubs :: Map ConnId RcvQueue -> Map SMPServer (Map ConnId RcvQueue) -> Map SMPServer (Map ConnId RcvQueue)
-        addPendingSubs cs = M.alter (Just . addSubs cs) srv
-        addSubs cs = maybe cs (M.union cs)
         deleteKeys :: Ord k => Set k -> Map k a -> Map k a
         deleteKeys ks m = S.foldr' M.delete m ks
 
     serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()
     serverDown u cs = unless (M.null cs) $ do
       mapM_ (notifySub DOWN) $ M.keysSet cs
-      unliftIO u reconnectServer
-
-    reconnectServer :: m ()
-    reconnectServer = do
-      a <- async tryReconnectClient
+      a <- async . unliftIO u $ tryReconnectClient cs
       atomically $ modifyTVar (reconnections c) (a :)
 
-    tryReconnectClient :: m ()
-    tryReconnectClient = do
+    tryReconnectClient :: Map ConnId RcvQueue -> m ()
+    tryReconnectClient cs = do
       ri <- asks $ reconnectInterval . config
       withRetryInterval ri $ \loop ->
-        reconnectClient `catchError` const loop
+        reconnectClient cs `catchError` const loop
 
-    reconnectClient :: m ()
-    reconnectClient = do
+    reconnectClient :: Map ConnId RcvQueue -> m ()
+    reconnectClient cs = do
       withAgentLock c . withSMP c srv $ \smp -> do
         subs <- readTVarIO $ subscrConns c
-        cs <- M.lookup srv <$> readTVarIO (pendingSubscrSrvrs c)
-        forM_ (maybe [] M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
+        forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
           when (isNothing $ M.lookup connId subs) $ do
             subscribeSMPQueue smp rcvPrivateKey rcvId
               `catchError` \case
-                e@SMPResponseTimeout -> throwError e
-                e@SMPNetworkError -> throwError e
-                e -> liftIO $ notifySub (ERR $ smpClientError e) connId
+                SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId
+                e -> throwError e
             addSubscription c rq connId
             liftIO $ notifySub UP connId
 
@@ -241,7 +206,6 @@ closeAgentClient :: MonadUnliftIO m => AgentClient -> m ()
 closeAgentClient c = liftIO $ do
   closeSMPServerClients c
   cancelActions $ reconnections c
-  cancelActions $ asyncClients c
   cancelActions $ smpQueueMsgDeliveries c
 
 closeSMPServerClients :: AgentClient -> IO ()
@@ -330,40 +294,26 @@ newRcvQueue_ a c srv = do
 
 subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
 subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
-  addPendingSubscription c rq connId
-  withLogSMP c server rcvId "SUB" $ \smp -> do
-    liftIO (runExceptT $ subscribeSMPQueue smp rcvPrivateKey rcvId) >>= \case
-      Left e -> do
-        atomically . when (e /= SMPNetworkError && e /= SMPResponseTimeout) $
-          removePendingSubscription c server connId
-        throwError e
-      Right _ -> addSubscription c rq connId
+  withLogSMP c server rcvId "SUB" $ \smp ->
+    subscribeSMPQueue smp rcvPrivateKey rcvId
+  addSubscription c rq connId
 
 addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
 addSubscription c rq@RcvQueue {server} connId = atomically $ do
   modifyTVar (subscrConns c) $ M.insert connId server
-  addSubs_ (subscrSrvrs c) rq connId
-  removePendingSubscription c server connId
-
-addPendingSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
-addPendingSubscription c rq connId =
-  atomically $ addSubs_ (pendingSubscrSrvrs c) rq connId
-
-addSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> RcvQueue -> ConnId -> STM ()
-addSubs_ ss rq@RcvQueue {server} connId = modifyTVar ss $ M.alter (Just . addSub) server
+  modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
   where
-    addSub = maybe (M.singleton connId rq) (M.insert connId rq)
+    addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue
+    addSub (Just cs) = M.insert connId rq cs
+    addSub _ = M.singleton connId rq
 
-removeSubscription :: MonadUnliftIO m => AgentClient -> ConnId -> m ()
-removeSubscription c@AgentClient {subscrConns} connId = atomically $ do
-  server_ <- stateTVar subscrConns $ \cs -> (M.lookup connId cs, M.delete connId cs)
-  mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
-
-removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()
-removePendingSubscription c = removeSubs_ (pendingSubscrSrvrs c)
-
-removeSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> SMPServer -> ConnId -> STM ()
-removeSubs_ ss server connId = modifyTVar ss $ M.update delSub server
+removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()
+removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do
+  cs <- readTVar subscrConns
+  writeTVar subscrConns $ M.delete connId cs
+  mapM_
+    (modifyTVar subscrSrvrs . M.alter (>>= delSub))
+    (M.lookup connId cs)
   where
     delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)
     delSub cs =
@@ -420,6 +370,7 @@ deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
   withLogSMP c server rcvId "DEL" $ \smp ->
     deleteSMPQueue smp rcvPrivateKey rcvId
 
+-- TODO this is just wrong
 sendAgentMessage :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
 sendAgentMessage c sq@SndQueue {server, sndId, sndPrivateKey} agentMsg =
   withLogSMP_ c server sndId "SEND <MSG>" $ \smp -> do

src/Simplex/Messaging/Agent/Env/Postgres.hs

@@ -0,0 +1,87 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
+{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
+
+module Simplex.Messaging.Agent.Env.Postgres
+  ( AgentConfig (..),
+    defaultAgentConfig,
+    Env (..),
+    newSMPAgentEnv,
+  )
+where
+
+import Control.Monad.IO.Unlift
+import Crypto.Random
+import Data.List.NonEmpty (NonEmpty)
+import Data.Time.Clock (NominalDiffTime, nominalDay)
+import Network.Socket
+import Numeric.Natural
+import Simplex.Messaging.Agent.Protocol (SMPServer)
+import Simplex.Messaging.Agent.RetryInterval
+import Simplex.Messaging.Agent.Store.SQLite
+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import Simplex.Messaging.Client
+import qualified Simplex.Messaging.Crypto as C
+import System.Random (StdGen, newStdGen)
+import UnliftIO.STM
+
+data AgentConfig = AgentConfig
+  { tcpPort :: ServiceName,
+    smpServers :: NonEmpty SMPServer,
+    cmdSignAlg :: C.SignAlg,
+    connIdBytes :: Int,
+    tbqSize :: Natural,
+    dbFile :: FilePath,
+    dbPoolSize :: Int,
+    smpCfg :: SMPClientConfig,
+    reconnectInterval :: RetryInterval,
+    helloTimeout :: NominalDiffTime,
+    caCertificateFile :: FilePath,
+    privateKeyFile :: FilePath,
+    certificateFile :: FilePath
+  }
+
+defaultAgentConfig :: AgentConfig
+defaultAgentConfig =
+  AgentConfig
+    { tcpPort = "5224",
+      smpServers = undefined, -- TODO move it elsewhere?
+      cmdSignAlg = C.SignAlg C.SEd448,
+      connIdBytes = 12,
+      tbqSize = 16,
+      dbFile = "smp-agent.db",
+      dbPoolSize = 4,
+      smpCfg = smpDefaultConfig,
+      reconnectInterval =
+        RetryInterval
+          { initialInterval = second,
+            increaseAfter = 10 * second,
+            maxInterval = 10 * second
+          },
+      helloTimeout = 7 * nominalDay,
+      -- CA certificate private key is not needed for initialization
+      -- ! we do not generate these
+      caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
+      privateKeyFile = "/etc/opt/simplex-agent/agent.key",
+      certificateFile = "/etc/opt/simplex-agent/agent.crt"
+    }
+  where
+    second = 1_000_000
+
+data Env = Env
+  { config :: AgentConfig,
+    store :: SQLiteStore,
+    idsDrg :: TVar ChaChaDRG,
+    clientCounter :: TVar Int,
+    randomServer :: TVar StdGen
+  }
+
+newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
+newSMPAgentEnv cfg = do
+  idsDrg <- newTVarIO =<< drgNew
+  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
+  clientCounter <- newTVarIO 0
+  randomServer <- newTVarIO =<< liftIO newStdGen
+  return Env {config = cfg, store, idsDrg, clientCounter, randomServer}

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

@@ -35,7 +35,6 @@ data AgentConfig = AgentConfig
     tbqSize :: Natural,
     dbFile :: FilePath,
     dbPoolSize :: Int,
-    yesToMigrations :: Bool,
     smpCfg :: SMPClientConfig,
     reconnectInterval :: RetryInterval,
     helloTimeout :: NominalDiffTime,
@@ -51,10 +50,9 @@ defaultAgentConfig =
       smpServers = undefined, -- TODO move it elsewhere?
       cmdSignAlg = C.SignAlg C.SEd448,
       connIdBytes = 12,
-      tbqSize = 64,
+      tbqSize = 16,
       dbFile = "smp-agent.db",
       dbPoolSize = 4,
-      yesToMigrations = False,
       smpCfg = smpDefaultConfig,
       reconnectInterval =
         RetryInterval
@@ -62,7 +60,7 @@ defaultAgentConfig =
             increaseAfter = 10 * second,
             maxInterval = 10 * second
           },
-      helloTimeout = 2 * nominalDay,
+      helloTimeout = 7 * nominalDay,
       -- CA certificate private key is not needed for initialization
       -- ! we do not generate these
       caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
@@ -81,9 +79,9 @@ data Env = Env
   }
 
 newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
-newSMPAgentEnv config@AgentConfig {dbFile, dbPoolSize, yesToMigrations} = do
+newSMPAgentEnv cfg = do
   idsDrg <- newTVarIO =<< drgNew
-  store <- liftIO $ createSQLiteStore dbFile dbPoolSize Migrations.app yesToMigrations
+  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
   clientCounter <- newTVarIO 0
   randomServer <- newTVarIO =<< liftIO newStdGen
-  return Env {config, store, idsDrg, clientCounter, randomServer}
+  return Env {config = cfg, store, idsDrg, clientCounter, randomServer}

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

@@ -0,0 +1,953 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
+{-# LANGUAGE UndecidableInstances #-}
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+
+module Simplex.Messaging.Agent.Store.Postgres
+  ( SQLiteStore (..),
+    createSQLiteStore,
+    connectSQLiteStore,
+    withConnection,
+    withTransaction,
+    fromTextField_,
+    firstRow,
+  )
+where
+
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.STM
+import Control.Exception (bracket)
+import Control.Monad.Except
+import Control.Monad.IO.Unlift (MonadUnliftIO)
+import Crypto.Random (ChaChaDRG, randomBytesGenerate)
+import Data.Bifunctor (second)
+import Data.ByteString (ByteString)
+import qualified Data.ByteString.Base64.URL as U
+import Data.Char (toLower)
+import Data.Functor (($>))
+import Data.List (find, foldl')
+import qualified Data.Map.Strict as M
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Text.Encoding (decodeLatin1)
+import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLData (..), SQLError, ToRow, field)
+import qualified Database.SQLite.Simple as DB
+import Database.SQLite.Simple.FromField
+import Database.SQLite.Simple.Internal (Field (..))
+import Database.SQLite.Simple.Ok (Ok (Ok))
+import Database.SQLite.Simple.QQ (sql)
+import Database.SQLite.Simple.ToField (ToField (..))
+import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Agent.Store
+import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration)
+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Crypto.Ratchet (RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys)
+import Simplex.Messaging.Encoding
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Parsers (blobFieldParser)
+import Simplex.Messaging.Protocol (MsgBody)
+import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Util (bshow, liftIOEither)
+import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
+import System.Exit (exitFailure)
+import System.FilePath (takeDirectory)
+import System.IO (hFlush, stdout)
+import qualified UnliftIO.Exception as E
+
+-- * SQLite Store implementation
+
+data SQLiteStore = SQLiteStore
+  { dbFilePath :: FilePath,
+    dbConnPool :: TBQueue DB.Connection,
+    dbNew :: Bool
+  }
+
+createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
+createSQLiteStore dbFilePath poolSize migrations = do
+  let dbDir = takeDirectory dbFilePath
+  createDirectoryIfMissing False dbDir
+  st <- connectSQLiteStore dbFilePath poolSize
+  checkThreadsafe st
+  migrateSchema st migrations
+  pure st
+
+checkThreadsafe :: SQLiteStore -> IO ()
+checkThreadsafe st = withConnection st $ \db -> do
+  compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[Text]]
+  let threadsafeOption = find (T.isPrefixOf "THREADSAFE=") (concat compileOptions)
+  case threadsafeOption of
+    Just "THREADSAFE=0" -> confirmOrExit "SQLite compiled with non-threadsafe code."
+    Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
+    _ -> return ()
+
+migrateSchema :: SQLiteStore -> [Migration] -> IO ()
+migrateSchema st migrations = withConnection st $ \db -> do
+  Migrations.initialize db
+  Migrations.get db migrations >>= \case
+    Left e -> confirmOrExit $ "Database error: " <> e
+    Right [] -> pure ()
+    Right ms -> do
+      unless (dbNew st) $ do
+        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
+        let f = dbFilePath st
+        copyFile f (f <> ".bak")
+      Migrations.run db ms
+
+confirmOrExit :: String -> IO ()
+confirmOrExit s = do
+  putStrLn s
+  putStr "Continue (y/N): "
+  hFlush stdout
+  ok <- getLine
+  when (map toLower ok /= "y") exitFailure
+
+connectSQLiteStore :: FilePath -> Int -> IO SQLiteStore
+connectSQLiteStore dbFilePath poolSize = do
+  dbNew <- not <$> doesFileExist dbFilePath
+  dbConnPool <- newTBQueueIO $ toEnum poolSize
+  replicateM_ poolSize $
+    connectDB dbFilePath >>= atomically . writeTBQueue dbConnPool
+  pure SQLiteStore {dbFilePath, dbConnPool, dbNew}
+
+connectDB :: FilePath -> IO DB.Connection
+connectDB path = do
+  dbConn <- DB.open path
+  DB.execute_ dbConn "PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL;"
+  pure dbConn
+
+checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
+checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)
+
+handleSQLError :: StoreError -> SQLError -> StoreError
+handleSQLError err e
+  | DB.sqlError e == DB.ErrorConstraint = err
+  | otherwise = SEInternal $ bshow e
+
+withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withConnection SQLiteStore {dbConnPool} =
+  bracket
+    (atomically $ readTBQueue dbConnPool)
+    (atomically . writeTBQueue dbConnPool)
+
+withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransaction st action = withConnection st $ loop 100 100_000
+  where
+    loop :: Int -> Int -> DB.Connection -> IO a
+    loop t tLim db =
+      DB.withImmediateTransaction db (action db) `E.catch` \(e :: SQLError) ->
+        if tLim > t && DB.sqlError e == DB.ErrorBusy
+          then do
+            threadDelay t
+            loop (t * 9 `div` 8) (tLim - t) db
+          else E.throwIO e
+
+createConn_ ::
+  (MonadUnliftIO m, MonadError StoreError m) =>
+  SQLiteStore ->
+  TVar ChaChaDRG ->
+  ConnData ->
+  (DB.Connection -> ByteString -> IO ()) ->
+  m ByteString
+createConn_ st gVar cData create =
+  liftIOEither . checkConstraint SEConnDuplicate . withTransaction st $ \db ->
+    case cData of
+      ConnData {connId = ""} -> createWithRandomId gVar $ create db
+      ConnData {connId} -> create db connId $> Right connId
+
+instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where
+  createRcvConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> m ConnId
+  createRcvConn st gVar cData q@RcvQueue {server} cMode =
+    createConn_ st gVar cData $ \db connId -> do
+      upsertServer_ db server
+      DB.execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, cMode)
+      insertRcvQueue_ db connId q
+
+  createSndConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId
+  createSndConn st gVar cData q@SndQueue {server} =
+    createConn_ st gVar cData $ \db connId -> do
+      upsertServer_ db server
+      DB.execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, SCMInvitation)
+      insertSndQueue_ db connId q
+
+  getConn :: SQLiteStore -> ConnId -> m SomeConn
+  getConn st connId =
+    liftIOEither . withTransaction st $ \db ->
+      getConn_ db connId
+
+  getRcvConn :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m SomeConn
+  getRcvConn st SMPServer {host, port} rcvId =
+    liftIOEither . withTransaction st $ \db ->
+      DB.queryNamed
+        db
+        [sql|
+          SELECT q.conn_id
+          FROM rcv_queues q
+          WHERE q.host = :host AND q.port = :port AND q.rcv_id = :rcv_id;
+        |]
+        [":host" := host, ":port" := port, ":rcv_id" := rcvId]
+        >>= \case
+          [Only connId] -> getConn_ db connId
+          _ -> pure $ Left SEConnNotFound
+
+  deleteConn :: SQLiteStore -> ConnId -> m ()
+  deleteConn st connId =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        "DELETE FROM connections WHERE conn_id = :conn_id;"
+        [":conn_id" := connId]
+
+  upgradeRcvConnToDuplex :: SQLiteStore -> ConnId -> SndQueue -> m ()
+  upgradeRcvConnToDuplex st connId sq@SndQueue {server} =
+    liftIOEither . withTransaction st $ \db ->
+      getConn_ db connId >>= \case
+        Right (SomeConn _ RcvConnection {}) -> do
+          upsertServer_ db server
+          insertSndQueue_ db connId sq
+          pure $ Right ()
+        Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+        _ -> pure $ Left SEConnNotFound
+
+  upgradeSndConnToDuplex :: SQLiteStore -> ConnId -> RcvQueue -> m ()
+  upgradeSndConnToDuplex st connId rq@RcvQueue {server} =
+    liftIOEither . withTransaction st $ \db ->
+      getConn_ db connId >>= \case
+        Right (SomeConn _ SndConnection {}) -> do
+          upsertServer_ db server
+          insertRcvQueue_ db connId rq
+          pure $ Right ()
+        Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+        _ -> pure $ Left SEConnNotFound
+
+  setRcvQueueStatus :: SQLiteStore -> RcvQueue -> QueueStatus -> m ()
+  setRcvQueueStatus st RcvQueue {rcvId, server = SMPServer {host, port}} status =
+    -- ? throw error if queue does not exist?
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE rcv_queues
+          SET status = :status
+          WHERE host = :host AND port = :port AND rcv_id = :rcv_id;
+        |]
+        [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId]
+
+  setRcvQueueConfirmedE2E :: SQLiteStore -> RcvQueue -> C.DhSecretX25519 -> m ()
+  setRcvQueueConfirmedE2E st RcvQueue {rcvId, server = SMPServer {host, port}} e2eDhSecret =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE rcv_queues
+          SET e2e_dh_secret = :e2e_dh_secret,
+              status = :status
+          WHERE host = :host AND port = :port AND rcv_id = :rcv_id
+        |]
+        [ ":status" := Confirmed,
+          ":e2e_dh_secret" := e2eDhSecret,
+          ":host" := host,
+          ":port" := port,
+          ":rcv_id" := rcvId
+        ]
+
+  setSndQueueStatus :: SQLiteStore -> SndQueue -> QueueStatus -> m ()
+  setSndQueueStatus st SndQueue {sndId, server = SMPServer {host, port}} status =
+    -- ? throw error if queue does not exist?
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE snd_queues
+          SET status = :status
+          WHERE host = :host AND port = :port AND snd_id = :snd_id;
+        |]
+        [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId]
+
+  createConfirmation :: SQLiteStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId
+  createConfirmation st gVar NewConfirmation {connId, senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo}, ratchetState} =
+    liftIOEither . withTransaction st $ \db ->
+      createWithRandomId gVar $ \confirmationId ->
+        DB.execute
+          db
+          [sql|
+            INSERT INTO conn_confirmations
+            (confirmation_id, conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, accepted) VALUES (?, ?, ?, ?, ?, ?, 0);
+          |]
+          (confirmationId, connId, senderKey, e2ePubKey, ratchetState, connInfo)
+
+  acceptConfirmation :: SQLiteStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation
+  acceptConfirmation st confirmationId ownConnInfo =
+    liftIOEither . withTransaction st $ \db -> do
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE conn_confirmations
+          SET accepted = 1,
+              own_conn_info = :own_conn_info
+          WHERE confirmation_id = :confirmation_id;
+        |]
+        [ ":own_conn_info" := ownConnInfo,
+          ":confirmation_id" := confirmationId
+        ]
+      firstRow confirmation SEConfirmationNotFound $
+        DB.query
+          db
+          [sql|
+            SELECT conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info
+            FROM conn_confirmations
+            WHERE confirmation_id = ?;
+          |]
+          (Only confirmationId)
+    where
+      confirmation (connId, senderKey, e2ePubKey, ratchetState, connInfo) =
+        AcceptedConfirmation
+          { confirmationId,
+            connId,
+            senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo},
+            ratchetState,
+            ownConnInfo
+          }
+
+  getAcceptedConfirmation :: SQLiteStore -> ConnId -> m AcceptedConfirmation
+  getAcceptedConfirmation st connId =
+    liftIOEither . withTransaction st $ \db ->
+      firstRow confirmation SEConfirmationNotFound $
+        DB.query
+          db
+          [sql|
+            SELECT confirmation_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, own_conn_info
+            FROM conn_confirmations
+            WHERE conn_id = ? AND accepted = 1;
+          |]
+          (Only connId)
+    where
+      confirmation (confirmationId, senderKey, e2ePubKey, ratchetState, connInfo, ownConnInfo) =
+        AcceptedConfirmation
+          { confirmationId,
+            connId,
+            senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo},
+            ratchetState,
+            ownConnInfo
+          }
+
+  removeConfirmations :: SQLiteStore -> ConnId -> m ()
+  removeConfirmations st connId =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          DELETE FROM conn_confirmations
+          WHERE conn_id = :conn_id;
+        |]
+        [":conn_id" := connId]
+
+  createInvitation :: SQLiteStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId
+  createInvitation st gVar NewInvitation {contactConnId, connReq, recipientConnInfo} =
+    liftIOEither . withTransaction st $ \db ->
+      createWithRandomId gVar $ \invitationId ->
+        DB.execute
+          db
+          [sql|
+            INSERT INTO conn_invitations
+            (invitation_id,  contact_conn_id, cr_invitation, recipient_conn_info, accepted) VALUES (?, ?, ?, ?, 0);
+          |]
+          (invitationId, contactConnId, connReq, recipientConnInfo)
+
+  getInvitation :: SQLiteStore -> InvitationId -> m Invitation
+  getInvitation st invitationId =
+    liftIOEither . withTransaction st $ \db ->
+      firstRow invitation SEInvitationNotFound $
+        DB.query
+          db
+          [sql|
+            SELECT contact_conn_id, cr_invitation, recipient_conn_info, own_conn_info, accepted
+            FROM conn_invitations
+            WHERE invitation_id = ?
+              AND accepted = 0
+          |]
+          (Only invitationId)
+    where
+      invitation (contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted) =
+        Invitation {invitationId, contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted}
+
+  acceptInvitation :: SQLiteStore -> InvitationId -> ConnInfo -> m ()
+  acceptInvitation st invitationId ownConnInfo =
+    liftIO . withTransaction st $ \db -> do
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE conn_invitations
+          SET accepted = 1,
+              own_conn_info = :own_conn_info
+          WHERE invitation_id = :invitation_id
+        |]
+        [ ":own_conn_info" := ownConnInfo,
+          ":invitation_id" := invitationId
+        ]
+
+  deleteInvitation :: SQLiteStore -> ConnId -> InvitationId -> m ()
+  deleteInvitation st contactConnId invId =
+    liftIOEither . withTransaction st $ \db ->
+      runExceptT $
+        ExceptT (getConn_ db contactConnId) >>= \case
+          SomeConn SCContact _ ->
+            liftIO $ DB.execute db "DELETE FROM conn_invitations WHERE contact_conn_id = ? AND invitation_id = ?" (contactConnId, invId)
+          _ -> throwError SEConnNotFound
+
+  updateRcvIds :: SQLiteStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
+  updateRcvIds st connId =
+    liftIO . withTransaction st $ \db -> do
+      (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId
+      let internalId = InternalId $ unId lastInternalId + 1
+          internalRcvId = InternalRcvId $ unRcvId lastInternalRcvId + 1
+      updateLastIdsRcv_ db connId internalId internalRcvId
+      pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash)
+
+  createRcvMsg :: SQLiteStore -> ConnId -> RcvMsgData -> m ()
+  createRcvMsg st connId rcvMsgData =
+    liftIO . withTransaction st $ \db -> do
+      insertRcvMsgBase_ db connId rcvMsgData
+      insertRcvMsgDetails_ db connId rcvMsgData
+      updateHashRcv_ db connId rcvMsgData
+
+  updateSndIds :: SQLiteStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
+  updateSndIds st connId =
+    liftIO . withTransaction st $ \db -> do
+      (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId
+      let internalId = InternalId $ unId lastInternalId + 1
+          internalSndId = InternalSndId $ unSndId lastInternalSndId + 1
+      updateLastIdsSnd_ db connId internalId internalSndId
+      pure (internalId, internalSndId, prevSndHash)
+
+  createSndMsg :: SQLiteStore -> ConnId -> SndMsgData -> m ()
+  createSndMsg st connId sndMsgData =
+    liftIO . withTransaction st $ \db -> do
+      insertSndMsgBase_ db connId sndMsgData
+      insertSndMsgDetails_ db connId sndMsgData
+      updateHashSnd_ db connId sndMsgData
+
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
+  getPendingMsgData st connId msgId =
+    liftIOEither . withTransaction st $ \db -> runExceptT $ do
+      rq_ <- liftIO $ getRcvQueueByConnId_ db connId
+      msgData <-
+        ExceptT . firstRow id SEMsgNotFound $
+          DB.query
+            db
+            [sql|
+                SELECT m.msg_type, m.msg_body, m.internal_ts
+                FROM messages m
+                JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
+                WHERE m.conn_id = ? AND m.internal_id = ?
+              |]
+            (connId, msgId)
+      pure (rq_, msgData)
+
+  getPendingMsgs :: SQLiteStore -> ConnId -> m [InternalId]
+  getPendingMsgs st connId =
+    liftIO . withTransaction st $ \db ->
+      map fromOnly
+        <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId)
+
+  checkRcvMsg :: SQLiteStore -> ConnId -> InternalId -> m ()
+  checkRcvMsg st connId msgId =
+    liftIOEither . withTransaction st $ \db ->
+      hasMsg
+        <$> DB.query
+          db
+          [sql|
+            SELECT conn_id, internal_id
+            FROM rcv_messages
+            WHERE conn_id = ? AND internal_id = ?
+          |]
+          (connId, msgId)
+    where
+      hasMsg :: [(ConnId, InternalId)] -> Either StoreError ()
+      hasMsg r = if null r then Left SEMsgNotFound else Right ()
+
+  deleteMsg :: SQLiteStore -> ConnId -> InternalId -> m ()
+  deleteMsg st connId msgId =
+    liftIO . withTransaction st $ \db ->
+      DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
+
+  createRatchetX3dhKeys :: SQLiteStore -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m ()
+  createRatchetX3dhKeys st connId x3dhPrivKey1 x3dhPrivKey2 =
+    liftIO . withTransaction st $ \db ->
+      DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
+
+  getRatchetX3dhKeys :: SQLiteStore -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448)
+  getRatchetX3dhKeys st connId =
+    liftIOEither . withTransaction st $ \db ->
+      fmap hasKeys $
+        firstRow id SEX3dhKeysNotFound $
+          DB.query db "SELECT x3dh_priv_key_1, x3dh_priv_key_2 FROM ratchets WHERE conn_id = ?" (Only connId)
+    where
+      hasKeys = \case
+        Right (Just k1, Just k2) -> Right (k1, k2)
+        _ -> Left SEX3dhKeysNotFound
+
+  createRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> m ()
+  createRatchet st connId rc =
+    liftIO . withTransaction st $ \db -> do
+      DB.executeNamed
+        db
+        [sql|
+          INSERT INTO ratchets (conn_id, ratchet_state)
+          VALUES (:conn_id, :ratchet_state)
+          ON CONFLICT (conn_id) DO UPDATE SET
+            ratchet_state = :ratchet_state,
+            x3dh_priv_key_1 = NULL,
+            x3dh_priv_key_2 = NULL
+        |]
+        [":conn_id" := connId, ":ratchet_state" := rc]
+
+  getRatchet :: SQLiteStore -> ConnId -> m RatchetX448
+  getRatchet st connId =
+    liftIOEither . withTransaction st $ \db ->
+      ratchet
+        <$> DB.query db "SELECT ratchet_state FROM ratchets WHERE conn_id = ?" (Only connId)
+    where
+      ratchet (Only (Just rc) : _) = Right rc
+      ratchet _ = Left SERatchetNotFound
+
+  getSkippedMsgKeys :: SQLiteStore -> ConnId -> m SkippedMsgKeys
+  getSkippedMsgKeys st connId =
+    liftIO . withTransaction st $ \db ->
+      skipped <$> DB.query db "SELECT header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ?" (Only connId)
+    where
+      skipped ms = foldl' addSkippedKey M.empty ms
+      addSkippedKey smks (hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks
+        where
+          addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk)
+
+  updateRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m ()
+  updateRatchet st connId rc skipped =
+    liftIO . withTransaction st $ \db -> do
+      DB.execute db "UPDATE ratchets SET ratchet_state = ? WHERE conn_id = ?" (rc, connId)
+      case skipped of
+        SMDNoChange -> pure ()
+        SMDRemove hk msgN ->
+          DB.execute db "DELETE FROM skipped_messages WHERE conn_id = ? AND header_key = ? AND msg_n = ?" (connId, hk, msgN)
+        SMDAdd smks ->
+          forM_ (M.assocs smks) $ \(hk, mks) ->
+            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)
+
+-- * Auxiliary helpers
+
+instance ToField QueueStatus where toField = toField . serializeQueueStatus
+
+instance FromField QueueStatus where fromField = fromTextField_ queueStatusT
+
+instance ToField InternalRcvId where toField (InternalRcvId x) = toField x
+
+instance FromField InternalRcvId where fromField x = InternalRcvId <$> fromField x
+
+instance ToField InternalSndId where toField (InternalSndId x) = toField x
+
+instance FromField InternalSndId where fromField x = InternalSndId <$> fromField x
+
+instance ToField InternalId where toField (InternalId x) = toField x
+
+instance FromField InternalId where fromField x = InternalId <$> fromField x
+
+instance ToField AMsgType where toField = toField . smpEncode
+
+instance FromField AMsgType where fromField = blobFieldParser smpP
+
+instance ToField MsgIntegrity where toField = toField . strEncode
+
+instance FromField MsgIntegrity where fromField = blobFieldParser strP
+
+instance ToField SMPQueueUri where toField = toField . strEncode
+
+instance FromField SMPQueueUri where fromField = blobFieldParser strP
+
+instance ToField AConnectionRequestUri where toField = toField . strEncode
+
+instance FromField AConnectionRequestUri where fromField = blobFieldParser strP
+
+instance ConnectionModeI c => ToField (ConnectionRequestUri c) where toField = toField . strEncode
+
+instance (E.Typeable c, ConnectionModeI c) => FromField (ConnectionRequestUri c) where fromField = blobFieldParser strP
+
+instance ToField ConnectionMode where toField = toField . decodeLatin1 . strEncode
+
+instance FromField ConnectionMode where fromField = fromTextField_ connModeT
+
+instance ToField (SConnectionMode c) where toField = toField . connMode
+
+instance FromField AConnectionMode where fromField = fromTextField_ $ fmap connMode' . connModeT
+
+fromTextField_ :: (E.Typeable a) => (Text -> Maybe a) -> Field -> Ok a
+fromTextField_ fromText = \case
+  f@(Field (SQLText t) _) ->
+    case fromText t of
+      Just x -> Ok x
+      _ -> returnError ConversionFailed f ("invalid text: " <> T.unpack t)
+  f -> returnError ConversionFailed f "expecting SQLText column type"
+
+listToEither :: e -> [a] -> Either e a
+listToEither _ (x : _) = Right x
+listToEither e _ = Left e
+
+firstRow :: (a -> b) -> e -> IO [a] -> IO (Either e b)
+firstRow f e a = second f . listToEither e <$> a
+
+{- ORMOLU_DISABLE -}
+-- SQLite.Simple only has these up to 10 fields, which is insufficient for some of our queries
+instance (FromField a, FromField b, FromField c, FromField d, FromField e,
+          FromField f, FromField g, FromField h, FromField i, FromField j,
+          FromField k) =>
+  FromRow (a,b,c,d,e,f,g,h,i,j,k) where
+  fromRow = (,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field
+                         <*> field <*> field <*> field <*> field <*> field
+                         <*> field
+
+instance (FromField a, FromField b, FromField c, FromField d, FromField e,
+          FromField f, FromField g, FromField h, FromField i, FromField j,
+          FromField k, FromField l) =>
+  FromRow (a,b,c,d,e,f,g,h,i,j,k,l) where
+  fromRow = (,,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field
+                          <*> field <*> field <*> field <*> field <*> field
+                          <*> field <*> field
+
+instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f,
+          ToField g, ToField h, ToField i, ToField j, ToField k, ToField l) =>
+  ToRow (a,b,c,d,e,f,g,h,i,j,k,l) where
+  toRow (a,b,c,d,e,f,g,h,i,j,k,l) =
+    [ toField a, toField b, toField c, toField d, toField e, toField f,
+      toField g, toField h, toField i, toField j, toField k, toField l
+    ]
+
+{- ORMOLU_ENABLE -}
+
+-- * Server upsert helper
+
+upsertServer_ :: DB.Connection -> SMPServer -> IO ()
+upsertServer_ dbConn SMPServer {host, port, keyHash} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO servers (host, port, key_hash) VALUES (:host,:port,:key_hash)
+      ON CONFLICT (host, port) DO UPDATE SET
+        host=excluded.host,
+        port=excluded.port,
+        key_hash=excluded.key_hash;
+    |]
+    [":host" := host, ":port" := port, ":key_hash" := keyHash]
+
+-- * createRcvConn helpers
+
+insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
+insertRcvQueue_ dbConn connId RcvQueue {..} = do
+  DB.executeNamed
+    dbConn
+    [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)
+      VALUES
+        (:host,:port,:rcv_id,:conn_id,:rcv_private_key,:rcv_dh_secret,:e2e_priv_key,:e2e_dh_secret,:snd_id,:status);
+    |]
+    [ ":host" := host server,
+      ":port" := port server,
+      ":rcv_id" := rcvId,
+      ":conn_id" := connId,
+      ":rcv_private_key" := rcvPrivateKey,
+      ":rcv_dh_secret" := rcvDhSecret,
+      ":e2e_priv_key" := e2ePrivKey,
+      ":e2e_dh_secret" := e2eDhSecret,
+      ":snd_id" := sndId,
+      ":status" := status
+    ]
+
+-- * createSndConn helpers
+
+insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
+insertSndQueue_ dbConn connId SndQueue {..} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO snd_queues
+        ( host, port, snd_id, conn_id, snd_private_key, e2e_dh_secret, status)
+      VALUES
+        (:host,:port,:snd_id,:conn_id,:snd_private_key,:e2e_dh_secret,:status);
+    |]
+    [ ":host" := host server,
+      ":port" := port server,
+      ":snd_id" := sndId,
+      ":conn_id" := connId,
+      ":snd_private_key" := sndPrivateKey,
+      ":e2e_dh_secret" := e2eDhSecret,
+      ":status" := status
+    ]
+
+-- * getConn helpers
+
+getConn_ :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
+getConn_ dbConn connId =
+  getConnData_ dbConn connId >>= \case
+    Nothing -> pure $ Left SEConnNotFound
+    Just (connData, cMode) -> do
+      rQ <- getRcvQueueByConnId_ dbConn connId
+      sQ <- getSndQueueByConnId_ dbConn connId
+      pure $ case (rQ, sQ, cMode) of
+        (Just rcvQ, Just sndQ, CMInvitation) -> Right $ SomeConn SCDuplex (DuplexConnection connData rcvQ sndQ)
+        (Just rcvQ, Nothing, CMInvitation) -> Right $ SomeConn SCRcv (RcvConnection connData rcvQ)
+        (Nothing, Just sndQ, CMInvitation) -> Right $ SomeConn SCSnd (SndConnection connData sndQ)
+        (Just rcvQ, Nothing, CMContact) -> Right $ SomeConn SCContact (ContactConnection connData rcvQ)
+        _ -> Left SEConnNotFound
+
+getConnData_ :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
+getConnData_ dbConn connId' =
+  connData
+    <$> DB.query dbConn "SELECT conn_id, conn_mode FROM connections WHERE conn_id = ?;" (Only connId')
+  where
+    connData [(connId, cMode)] = Just (ConnData {connId}, cMode)
+    connData _ = Nothing
+
+getRcvQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)
+getRcvQueueByConnId_ dbConn connId =
+  rcvQueue
+    <$> DB.query
+      dbConn
+      [sql|
+        SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
+          q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status
+        FROM rcv_queues q
+        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+        WHERE q.conn_id = ?;
+      |]
+      (Only connId)
+  where
+    rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)] =
+      let server = SMPServer host port keyHash
+       in Just RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status}
+    rcvQueue _ = Nothing
+
+getSndQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe SndQueue)
+getSndQueueByConnId_ dbConn connId =
+  sndQueue
+    <$> DB.query
+      dbConn
+      [sql|
+        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status
+        FROM snd_queues q
+        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+        WHERE q.conn_id = ?;
+      |]
+      (Only connId)
+  where
+    sndQueue [(keyHash, host, port, sndId, sndPrivateKey, e2eDhSecret, status)] =
+      let server = SMPServer host port keyHash
+       in Just SndQueue {server, sndId, sndPrivateKey, e2eDhSecret, status}
+    sndQueue _ = Nothing
+
+-- * updateRcvIds helpers
+
+retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
+retrieveLastIdsAndHashRcv_ dbConn connId = do
+  [(lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)] <-
+    DB.queryNamed
+      dbConn
+      [sql|
+        SELECT last_internal_msg_id, last_internal_rcv_msg_id, last_external_snd_msg_id, last_rcv_msg_hash
+        FROM connections
+        WHERE conn_id = :conn_id;
+      |]
+      [":conn_id" := connId]
+  return (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)
+
+updateLastIdsRcv_ :: DB.Connection -> ConnId -> InternalId -> InternalRcvId -> IO ()
+updateLastIdsRcv_ dbConn connId 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_id = :conn_id;
+    |]
+    [ ":last_internal_msg_id" := newInternalId,
+      ":last_internal_rcv_msg_id" := newInternalRcvId,
+      ":conn_id" := connId
+    ]
+
+-- * createRcvMsg helpers
+
+insertRcvMsgBase_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
+insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgType, msgBody, internalRcvId} = do
+  let MsgMeta {recipient = (internalId, internalTs)} = msgMeta
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO messages
+        ( conn_id, internal_id, internal_ts, internal_rcv_id, internal_snd_id, msg_type, msg_body)
+      VALUES
+        (:conn_id,:internal_id,:internal_ts,:internal_rcv_id,            NULL,:msg_type, :msg_body);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_id" := internalId,
+      ":internal_ts" := internalTs,
+      ":internal_rcv_id" := internalRcvId,
+      ":msg_type" := msgType,
+      ":msg_body" := msgBody
+    ]
+
+insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
+insertRcvMsgDetails_ dbConn connId RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash} = do
+  let MsgMeta {integrity, recipient, broker, sndMsgId} = msgMeta
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO rcv_messages
+        ( conn_id, internal_rcv_id, internal_id, external_snd_id,
+          broker_id, broker_ts,
+          internal_hash, external_prev_snd_hash, integrity)
+      VALUES
+        (:conn_id,:internal_rcv_id,:internal_id,:external_snd_id,
+         :broker_id,:broker_ts,
+         :internal_hash,:external_prev_snd_hash,:integrity);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_rcv_id" := internalRcvId,
+      ":internal_id" := fst recipient,
+      ":external_snd_id" := sndMsgId,
+      ":broker_id" := fst broker,
+      ":broker_ts" := snd broker,
+      ":internal_hash" := internalHash,
+      ":external_prev_snd_hash" := externalPrevSndHash,
+      ":integrity" := integrity
+    ]
+
+updateHashRcv_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
+updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =
+  DB.executeNamed
+    dbConn
+    -- last_internal_rcv_msg_id equality check prevents race condition in case next id was reserved
+    [sql|
+      UPDATE connections
+      SET last_external_snd_msg_id = :last_external_snd_msg_id,
+          last_rcv_msg_hash = :last_rcv_msg_hash
+      WHERE conn_id = :conn_id
+        AND last_internal_rcv_msg_id = :last_internal_rcv_msg_id;
+    |]
+    [ ":last_external_snd_msg_id" := sndMsgId (msgMeta :: MsgMeta),
+      ":last_rcv_msg_hash" := internalHash,
+      ":conn_id" := connId,
+      ":last_internal_rcv_msg_id" := internalRcvId
+    ]
+
+-- * updateSndIds helpers
+
+retrieveLastIdsAndHashSnd_ :: DB.Connection -> ConnId -> IO (InternalId, InternalSndId, PrevSndMsgHash)
+retrieveLastIdsAndHashSnd_ dbConn connId = do
+  [(lastInternalId, lastInternalSndId, lastSndHash)] <-
+    DB.queryNamed
+      dbConn
+      [sql|
+        SELECT last_internal_msg_id, last_internal_snd_msg_id, last_snd_msg_hash
+        FROM connections
+        WHERE conn_id = :conn_id;
+      |]
+      [":conn_id" := connId]
+  return (lastInternalId, lastInternalSndId, lastSndHash)
+
+updateLastIdsSnd_ :: DB.Connection -> ConnId -> InternalId -> InternalSndId -> IO ()
+updateLastIdsSnd_ dbConn connId 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_id = :conn_id;
+    |]
+    [ ":last_internal_msg_id" := newInternalId,
+      ":last_internal_snd_msg_id" := newInternalSndId,
+      ":conn_id" := connId
+    ]
+
+-- * createSndMsg helpers
+
+insertSndMsgBase_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
+insertSndMsgBase_ dbConn connId SndMsgData {..} = do
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO messages
+        ( conn_id, internal_id, internal_ts, internal_rcv_id, internal_snd_id, msg_type, msg_body)
+      VALUES
+        (:conn_id,:internal_id,:internal_ts,            NULL,:internal_snd_id,:msg_type, :msg_body);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_id" := internalId,
+      ":internal_ts" := internalTs,
+      ":internal_snd_id" := internalSndId,
+      ":msg_type" := msgType,
+      ":msg_body" := msgBody
+    ]
+
+insertSndMsgDetails_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
+insertSndMsgDetails_ dbConn connId SndMsgData {..} =
+  DB.executeNamed
+    dbConn
+    [sql|
+      INSERT INTO snd_messages
+        ( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash)
+      VALUES
+        (:conn_id,:internal_snd_id,:internal_id,:internal_hash,:previous_msg_hash);
+    |]
+    [ ":conn_id" := connId,
+      ":internal_snd_id" := internalSndId,
+      ":internal_id" := internalId,
+      ":internal_hash" := internalHash,
+      ":previous_msg_hash" := prevMsgHash
+    ]
+
+updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
+updateHashSnd_ dbConn connId SndMsgData {..} =
+  DB.executeNamed
+    dbConn
+    -- last_internal_snd_msg_id equality check prevents race condition in case next id was reserved
+    [sql|
+      UPDATE connections
+      SET last_snd_msg_hash = :last_snd_msg_hash
+      WHERE conn_id = :conn_id
+        AND last_internal_snd_msg_id = :last_internal_snd_msg_id;
+    |]
+    [ ":last_snd_msg_hash" := internalHash,
+      ":conn_id" := connId,
+      ":last_internal_snd_msg_id" := internalSndId
+    ]
+
+-- create record with a random ID
+createWithRandomId :: TVar ChaChaDRG -> (ByteString -> IO ()) -> IO (Either StoreError ByteString)
+createWithRandomId gVar create = tryCreate 3
+  where
+    tryCreate :: Int -> IO (Either StoreError ByteString)
+    tryCreate 0 = pure $ Left SEUniqueID
+    tryCreate n = do
+      id' <- randomId gVar 12
+      E.try (create id') >>= \case
+        Right _ -> pure $ Right id'
+        Left e
+          | DB.sqlError e == DB.ErrorConstraint -> tryCreate (n - 1)
+          | otherwise -> pure . Left . SEInternal $ bshow e
+
+randomId :: TVar ChaChaDRG -> Int -> IO ByteString
+randomId gVar n = U.encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)

src/Simplex/Messaging/Agent/Store/Postgres/Migrations.hs

@@ -0,0 +1,72 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE TupleSections #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations
+  ( Migration (..),
+    app,
+    initialize,
+    get,
+    run,
+  )
+where
+
+import Control.Monad (forM_)
+import Data.Function (on)
+import Data.List (intercalate, sortBy)
+import Data.Text (Text)
+import Data.Time.Clock (getCurrentTime)
+import Database.SQLite.Simple (Connection, Only (..), Query (..))
+import qualified Database.SQLite.Simple as DB
+import Database.SQLite.Simple.QQ (sql)
+import qualified Database.SQLite3 as SQLite3
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial
+
+data Migration = Migration {name :: String, up :: Text}
+  deriving (Show)
+
+schemaMigrations :: [(String, Query)]
+schemaMigrations =
+  [ ("20220101_initial", m20220202_initial)
+  ]
+
+-- | The list of migrations in ascending order by date
+app :: [Migration]
+app = sortBy (compare `on` name) $ map migration schemaMigrations
+  where
+    migration (name, query) = Migration {name = name, up = fromQuery query}
+
+get :: Connection -> [Migration] -> IO (Either String [Migration])
+get conn migrations =
+  migrationsToRun migrations . map fromOnly
+    <$> DB.query_ conn "SELECT name FROM migrations ORDER BY name ASC;"
+
+run :: Connection -> [Migration] -> IO ()
+run conn ms = DB.withImmediateTransaction conn . forM_ ms $
+  \Migration {name, up} -> insert name >> execSQL up
+  where
+    insert name = DB.execute conn "INSERT INTO migrations (name, ts) VALUES (?, ?);" . (name,) =<< getCurrentTime
+    execSQL = SQLite3.exec $ DB.connectionHandle conn
+
+initialize :: Connection -> IO ()
+initialize conn =
+  DB.execute_
+    conn
+    [sql|
+      CREATE TABLE IF NOT EXISTS migrations (
+        name TEXT NOT NULL,
+        ts TEXT NOT NULL,
+        PRIMARY KEY (name)
+      );
+    |]
+
+migrationsToRun :: [Migration] -> [String] -> Either String [Migration]
+migrationsToRun appMs [] = Right appMs
+migrationsToRun [] dbMs = Left $ "database version is newer than the app: " <> intercalate ", " dbMs
+migrationsToRun (a : as) (d : ds)
+  | name a == d = migrationsToRun as ds
+  | otherwise = Left $ "different migration in the app/database: " <> name a <> " / " <> d

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs

@@ -0,0 +1,148 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220202_initial :: Query
+m20220202_initial =
+  [sql|
+CREATE TABLE servers (
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  key_hash BLOB NOT NULL,
+  PRIMARY KEY (host, port)
+) WITHOUT ROWID;
+
+CREATE TABLE connections (
+  conn_id BLOB NOT NULL PRIMARY KEY,
+  conn_mode TEXT NOT NULL,
+  last_internal_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_internal_rcv_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_internal_snd_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_external_snd_msg_id INTEGER NOT NULL DEFAULT 0,
+  last_rcv_msg_hash BLOB NOT NULL DEFAULT x'',
+  last_snd_msg_hash BLOB NOT NULL DEFAULT x'',
+  smp_agent_version INTEGER NOT NULL DEFAULT 1
+) WITHOUT ROWID;
+
+CREATE TABLE rcv_queues (
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  rcv_id BLOB NOT NULL,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  rcv_private_key BLOB NOT NULL,
+  rcv_dh_secret BLOB NOT NULL,
+  e2e_priv_key BLOB NOT NULL,
+  e2e_dh_secret BLOB,
+  snd_id BLOB NOT NULL,
+  snd_key BLOB,
+  status TEXT NOT NULL,
+  smp_server_version INTEGER NOT NULL DEFAULT 1,
+  smp_client_version INTEGER,
+  PRIMARY KEY (host, port, rcv_id),
+  FOREIGN KEY (host, port) REFERENCES servers
+    ON DELETE RESTRICT ON UPDATE CASCADE,
+  UNIQUE (host, port, snd_id)
+) WITHOUT ROWID;
+
+CREATE TABLE snd_queues (
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  snd_id BLOB NOT NULL,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  snd_private_key BLOB NOT NULL,
+  e2e_dh_secret BLOB NOT NULL,
+  status TEXT NOT NULL,
+  smp_server_version INTEGER NOT NULL DEFAULT 1,
+  smp_client_version INTEGER NOT NULL DEFAULT 1,
+  PRIMARY KEY (host, port, snd_id),
+  FOREIGN KEY (host, port) REFERENCES servers
+    ON DELETE RESTRICT ON UPDATE CASCADE
+) WITHOUT ROWID;
+
+CREATE TABLE messages (
+  conn_id BLOB NOT NULL REFERENCES connections (conn_id)
+    ON DELETE CASCADE,
+  internal_id INTEGER NOT NULL,
+  internal_ts TEXT NOT NULL,
+  internal_rcv_id INTEGER,
+  internal_snd_id INTEGER,
+  msg_type BLOB NOT NULL, -- (H)ELLO, (R)EPLY, (D)ELETE. Should SMP confirmation be saved too?
+  msg_body BLOB NOT NULL DEFAULT x'',
+  PRIMARY KEY (conn_id, internal_id),
+  FOREIGN KEY (conn_id, internal_rcv_id) REFERENCES rcv_messages
+    ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  FOREIGN KEY (conn_id, internal_snd_id) REFERENCES snd_messages
+    ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED
+) WITHOUT ROWID;
+
+CREATE TABLE rcv_messages (
+  conn_id BLOB NOT NULL,
+  internal_rcv_id INTEGER NOT NULL,
+  internal_id INTEGER NOT NULL,
+  external_snd_id INTEGER NOT NULL,
+  broker_id BLOB NOT NULL,
+  broker_ts TEXT NOT NULL,
+  internal_hash BLOB NOT NULL,
+  external_prev_snd_hash BLOB NOT NULL,
+  integrity BLOB NOT NULL,
+  PRIMARY KEY (conn_id, internal_rcv_id),
+  FOREIGN KEY (conn_id, internal_id) REFERENCES messages
+    ON DELETE CASCADE
+) WITHOUT ROWID;
+
+CREATE TABLE snd_messages (
+  conn_id BLOB NOT NULL,
+  internal_snd_id INTEGER NOT NULL,
+  internal_id INTEGER NOT NULL,
+  internal_hash BLOB NOT NULL,
+  previous_msg_hash BLOB NOT NULL DEFAULT x'',
+  PRIMARY KEY (conn_id, internal_snd_id),
+  FOREIGN KEY (conn_id, internal_id) REFERENCES messages
+    ON DELETE CASCADE
+) WITHOUT ROWID;
+
+CREATE TABLE conn_confirmations (
+  confirmation_id BLOB NOT NULL PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  e2e_snd_pub_key BLOB NOT NULL, -- TODO per-queue key. Split?
+  sender_key BLOB NOT NULL, -- TODO per-queue key. Split?
+  ratchet_state BLOB NOT NULL,
+  sender_conn_info BLOB NOT NULL,
+  accepted INTEGER NOT NULL,
+  own_conn_info BLOB,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+) WITHOUT ROWID;
+
+CREATE TABLE conn_invitations (
+  invitation_id BLOB NOT NULL PRIMARY KEY,
+  contact_conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  cr_invitation BLOB NOT NULL,
+  recipient_conn_info BLOB NOT NULL,
+  accepted INTEGER NOT NULL DEFAULT 0,
+  own_conn_info BLOB,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+) WITHOUT ROWID;
+
+CREATE TABLE ratchets (
+  conn_id BLOB NOT NULL PRIMARY KEY REFERENCES connections
+    ON DELETE CASCADE,
+  -- x3dh keys are not saved on the sending side (the side accepting the connection)
+  x3dh_priv_key_1 BLOB,
+  x3dh_priv_key_2 BLOB,
+  -- ratchet is initially empty on the receiving side (the side offering the connection)
+  ratchet_state BLOB,
+  e2e_version INTEGER NOT NULL DEFAULT 1
+) WITHOUT ROWID;
+
+CREATE TABLE skipped_messages (
+  skipped_message_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES ratchets
+    ON DELETE CASCADE,
+  header_key BLOB NOT NULL,
+  msg_n INTEGER NOT NULL,
+  msg_key BLOB NOT NULL
+);
+|]

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

@@ -76,13 +76,13 @@ data SQLiteStore = SQLiteStore
     dbNew :: Bool
   }
 
-createSQLiteStore :: FilePath -> Int -> [Migration] -> Bool -> IO SQLiteStore
-createSQLiteStore dbFilePath poolSize migrations yesToMigrations = do
+createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
+createSQLiteStore dbFilePath poolSize migrations = do
   let dbDir = takeDirectory dbFilePath
   createDirectoryIfMissing False dbDir
   st <- connectSQLiteStore dbFilePath poolSize
   checkThreadsafe st
-  migrateSchema st migrations yesToMigrations
+  migrateSchema st migrations
   pure st
 
 checkThreadsafe :: SQLiteStore -> IO ()
@@ -94,16 +94,15 @@ checkThreadsafe st = withConnection st $ \db -> do
     Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
     _ -> return ()
 
-migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()
-migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do
+migrateSchema :: SQLiteStore -> [Migration] -> IO ()
+migrateSchema st migrations = withConnection st $ \db -> do
   Migrations.initialize db
   Migrations.get db migrations >>= \case
     Left e -> confirmOrExit $ "Database error: " <> e
     Right [] -> pure ()
     Right ms -> do
       unless (dbNew st) $ do
-        unless yesToMigrations $
-          confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
+        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
         let f = dbFilePath st
         copyFile f (f <> ".bak")
       Migrations.run db ms

src/Simplex/Messaging/Client.hs

@@ -107,7 +107,7 @@ data SMPClientConfig = SMPClientConfig
 smpDefaultConfig :: SMPClientConfig
 smpDefaultConfig =
   SMPClientConfig
-    { qSize = 64,
+    { qSize = 16,
       defaultTransport = ("5223", transport @TLS),
       tcpTimeout = 4_000_000,
       smpPing = 30_000_000

src/Simplex/Messaging/Parsers.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE CPP #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 
@@ -98,18 +97,13 @@ enumJSON tagModifier =
     }
 
 sumTypeJSON :: (String -> String) -> J.Options
-#if defined(darwin_HOST_OS)
 sumTypeJSON = singleFieldJSON
-#else
-sumTypeJSON = taggedObjectJSON
-#endif
 
 taggedObjectJSON :: (String -> String) -> J.Options
 taggedObjectJSON tagModifier =
   J.defaultOptions
     { J.sumEncoding = J.TaggedObject "type" "data",
       J.constructorTagModifier = tagModifier,
-      J.allNullaryToStringTag = False,
       J.nullaryToObject = True,
       J.omitNothingFields = True
     }
@@ -119,7 +113,6 @@ singleFieldJSON tagModifier =
   J.defaultOptions
     { J.sumEncoding = J.ObjectWithSingleField,
       J.constructorTagModifier = tagModifier,
-      J.allNullaryToStringTag = False,
       J.nullaryToObject = True,
       J.omitNothingFields = True
     }

src/Simplex/Messaging/Transport.hs

@@ -96,7 +96,7 @@ supportedSMPVersions :: VersionRange
 supportedSMPVersions = mkVersionRange 1 1
 
 simplexMQVersion :: String
-simplexMQVersion = "1.0.3"
+simplexMQVersion = "1.0.2"
 
 -- * Transport connection class
 

tests/AgentTests.hs

@@ -62,11 +62,9 @@ agentTests (ATransport t) = do
       smpAgentTest3_1_1 $ testSubscription t
     it "should send notifications to client when server disconnects" $
       smpAgentServerTest $ testSubscrNotification t
-  describe "Message delivery and server reconnection" $ do
+  describe "Message delivery" $ do
     it "should deliver messages after losing server connection and re-connecting" $
       smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
-    it "should connect to the server when server goes up if it initially was down" $
-      smpAgentTestN [] $ testServerConnectionAfterError t
     it "should deliver pending messages after agent restarting" $
       smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
     it "should concurrently deliver messages to connections without blocking" $
@@ -302,43 +300,6 @@ testMsgDeliveryServerRestart t alice bob = do
   where
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
 
-testServerConnectionAfterError :: forall c. Transport c => TProxy c -> [c] -> IO ()
-testServerConnectionAfterError t _ = do
-  withAgent1 $ \bob -> do
-    withAgent2 $ \alice -> do
-      withServer $ do
-        connect (bob, "bob") (alice, "alice")
-
-      bob <# ("", "alice", DOWN)
-      alice <# ("", "bob", DOWN)
-      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 4)
-      alice #:# "nothing else delivered before the server is restarted"
-      bob #:# "nothing else delivered before the server is restarted"
-
-  withAgent1 $ \bob -> do
-    withAgent2 $ \alice -> do
-      bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
-      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
-      withServer $ do
-        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 4; _ -> False
-        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 4; _ -> False
-        bob <# ("", "alice", UP)
-        bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
-        bob #: ("2", "alice", "ACK 4") #> ("2", "alice", OK)
-        alice #: ("1", "bob", "SEND 11\nhello again") #> ("1", "bob", MID 5)
-        alice <# ("", "bob", SENT 5)
-        bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
-
-  removeFile testStoreLogFile
-  removeFile testDB
-  removeFile testDB2
-  where
-    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
-    withAgent1 = withAgent agentTestPort testDB
-    withAgent2 = withAgent agentTestPort2 testDB2
-    withAgent :: String -> String -> (c -> IO a) -> IO a
-    withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
-
 testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
 testMsgDeliveryAgentRestart t bob = do
   withAgent $ \alice -> do

tests/AgentTests/SQLiteTests.hs

@@ -51,7 +51,7 @@ createStore = do
   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
-  createSQLiteStore (testDB <> show r) 4 Migrations.app True
+  createSQLiteStore (testDB <> show r) 4 Migrations.app
 
 removeStore :: SQLiteStore -> IO ()
 removeStore store = do