v2.2.0

* fix sockets/threads/memory leak

* refactor

CHANGELOG.md

@@ -1,3 +1,20 @@
+# 2.2.0
+
+SMP server:
+
+- Fix sockets/threads/memory leak
+
+SMP agent:
+
+- Support stopping and resuming agent with `disconnectAgentClient` / `resumeAgentClient`
+
+# 2.1.1
+
+SMP server:
+
+- gracefully close sockets on client disconnection
+- CLI warning when deleting server configuration
+
 # 2.1.0
 
 SMP server:

apps/ntf-server/Main.hs

@@ -5,7 +5,7 @@ module Main where
 import Control.Logger.Simple
 import Simplex.Messaging.Client.Agent (defaultSMPClientAgentConfig)
 import Simplex.Messaging.Notifications.Server (runNtfServer)
-import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..), defaultInactiveClientExpiration)
+import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
 import Simplex.Messaging.Notifications.Server.Push.APNS (defaultAPNSPushClientConfig)
 import Simplex.Messaging.Server.CLI (ServerCLIConfig (..), protocolServerCLI)
 import System.FilePath (combine)
@@ -66,7 +66,7 @@ ntfServerCLIConfig =
                 pushQSize = 128,
                 smpAgentCfg = defaultSMPClientAgentConfig,
                 apnsConfig = defaultAPNSPushClientConfig,
-                inactiveClientExpiration = Just defaultInactiveClientExpiration,
+                inactiveClientExpiration = Nothing,
                 caCertificateFile = caCrtFile,
                 privateKeyFile = serverKeyFile,
                 certificateFile = serverCrtFile

apps/smp-server/Main.hs

@@ -7,8 +7,9 @@
 module Main where
 
 import Control.Logger.Simple
+import Data.Ini (lookupValue)
 import Simplex.Messaging.Server (runSMPServer)
-import Simplex.Messaging.Server.CLI (ServerCLIConfig (..), protocolServerCLI, readStrictIni, strictIni)
+import Simplex.Messaging.Server.CLI (ServerCLIConfig (..), protocolServerCLI, readStrictIni)
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defaultInactiveClientExpiration, defaultMessageExpiration)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (simplexMQVersion)
@@ -64,9 +65,9 @@ smpServerCLIConfig =
               <> "websockets: off\n\n"
               <> "[INACTIVE_CLIENTS]\n\
                  \# TTL and interval to check inactive clients\n\
-                 \disconnect: on\n"
-              <> ("ttl: " <> show (ttl defaultInactiveClientExpiration) <> "\n")
-              <> ("check_interval: " <> show (checkInterval defaultInactiveClientExpiration) <> "\n"),
+                 \disconnect: off\n"
+              <> ("# ttl: " <> show (ttl defaultInactiveClientExpiration) <> "\n")
+              <> ("# check_interval: " <> show (checkInterval defaultInactiveClientExpiration) <> "\n"),
           mkServerConfig = \storeLogFile transports ini ->
             ServerConfig
               { transports,
@@ -82,7 +83,7 @@ smpServerCLIConfig =
                 allowNewQueues = True,
                 messageExpiration = Just defaultMessageExpiration,
                 inactiveClientExpiration =
-                  if strictIni "INACTIVE_CLIENTS" "disconnect" ini == "on"
+                  if lookupValue "INACTIVE_CLIENTS" "disconnect" ini == Right "on"
                     then
                       Just
                         ExpirationConfig

package.yaml

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

protocol/diagrams/notifications/subscription.mmd

@@ -0,0 +1,40 @@
+sequenceDiagram
+  participant M as mobile app
+  participant C as chat core
+  participant A as agent
+  participant S as SMP server
+  participant N as NTF server
+  participant APN as APN
+
+  note over M, APN: register subscription
+
+  alt register existing
+    M -->> A: on /_ntf register, for subscribed queues
+  else create new connection
+    A -->> S: NEW / JOIN
+    note over A, S: ...<br>Connection handshake<br>...
+    S -->> A: CON
+  end
+  A ->> S: NKEY nKey
+  S ->> A: NID nId
+  A ->> N: SNEW tknId dhKey (smpServer, nId, nKey)
+  N ->> A: ID subId dhKey
+  N ->> S: NSUB nId
+  S ->> N: OK [/ NMSG]
+
+  note over M, APN: notify about message
+
+  S ->> N: NMSG
+  N ->> APN: APNSMutableContent<br>ntfQueue, nonce
+  APN ->> M: UNMutableNotificationContent
+  note over M, S: ...<br>Client awaken, message is received<br>...
+  S ->> M: message
+  note over M: mutate notification
+
+  note over M, APN: change APN token
+
+  APN ->> M: new device token
+  M -->> C: /_ntf_sub update tkn
+  C -->> A: updateNtfToken()
+  A -->> N: TUPD tknId newDeviceToken
+  note over M, N: ...<br>Verify token<br>...

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        2.1.0
+version:        2.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -329,6 +329,7 @@ test-suite smp-server-test
       AgentTests.DoubleRatchetTests
       AgentTests.FunctionalAPITests
       AgentTests.NotificationTests
+      AgentTests.SchemaDump
       AgentTests.SQLiteTests
       CoreTests.EncodingTests
       CoreTests.ProtocolErrorTests

src/Simplex/Messaging/Agent.hs

@@ -35,7 +35,8 @@ module Simplex.Messaging.Agent
     AgentMonad,
     AgentErrorMonad,
     getSMPAgentClient,
-    disconnectAgentClient, -- used in tests
+    disconnectAgentClient,
+    resumeAgentClient,
     withAgentLock,
     createConnection,
     joinConnection,
@@ -113,6 +114,9 @@ getSMPAgentClient cfg initServers = newSMPAgentEnv cfg >>= runReaderT runAgent
 disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m ()
 disconnectAgentClient c = closeAgentClient c >> logConnection c False
 
+resumeAgentClient :: MonadIO m => AgentClient -> m ()
+resumeAgentClient c = atomically $ writeTVar (active c) True
+
 -- |
 type AgentErrorMonad m = (MonadUnliftIO m, MonadError AgentErrorType m)
 

src/Simplex/Messaging/Agent/Client.hs

@@ -39,6 +39,7 @@ module Simplex.Messaging.Agent.Client
     logServer,
     removeSubscription,
     hasActiveSubscription,
+    agentDbPath,
   )
 where
 
@@ -63,6 +64,7 @@ import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
+import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..))
 import Simplex.Messaging.Client
 import Simplex.Messaging.Client.Agent ()
 import qualified Simplex.Messaging.Crypto as C
@@ -132,6 +134,9 @@ newAgentClient InitialAgentServers {smp, ntf} agentEnv = do
   lock <- newTMVar ()
   return AgentClient {active, rcvQ, subQ, msgQ, smpServers, ntfServers, smpClients, ntfClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock}
 
+agentDbPath :: AgentClient -> FilePath
+agentDbPath AgentClient {agentEnv = Env {store = SQLiteStore {dbFilePath}}} = dbFilePath
+
 -- | Agent monad with MonadReader Env and MonadError AgentErrorType
 type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
 
@@ -184,10 +189,11 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
             _ -> TM.insert srv cVar ps
 
     serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()
-    serverDown u cs = unless (M.null cs) $ do
-      let conns = M.keys cs
-      unless (null conns) . notifySub "" $ DOWN srv conns
-      whenM (readTVarIO active) $ unliftIO u reconnectServer
+    serverDown u cs = unless (M.null cs) $
+      whenM (readTVarIO active) $ do
+        let conns = M.keys cs
+        unless (null conns) . notifySub "" $ DOWN srv conns
+        unliftIO u reconnectServer
 
     reconnectServer :: m ()
     reconnectServer = do
@@ -304,21 +310,30 @@ newProtocolClient c srv clients connectClient reconnectClient clientVar = tryCon
 closeAgentClient :: MonadIO m => AgentClient -> m ()
 closeAgentClient c = liftIO $ do
   atomically $ writeTVar (active c) False
-  closeSMPServerClients c
+  closeProtocolServerClients (clientTimeout smpCfg) $ smpClients c
+  closeProtocolServerClients (clientTimeout ntfCfg) $ ntfClients c
   cancelActions $ reconnections c
   cancelActions $ asyncClients c
   cancelActions $ smpQueueMsgDeliveries c
-
-closeSMPServerClients :: AgentClient -> IO ()
-closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ (forkIO . closeClient)
+  clear subscrSrvrs
+  clear pendingSubscrSrvrs
+  clear subscrConns
+  clear connMsgsQueued
+  clear smpQueueMsgQueues
   where
-    closeClient smpVar =
-      atomically (readTMVar smpVar) >>= \case
-        Right smp -> closeProtocolClient smp `catchAll_` pure ()
+    clientTimeout sel = tcpTimeout . sel . config $ agentEnv c
+    clear sel = atomically $ writeTVar (sel c) M.empty
+
+closeProtocolServerClients :: Int -> TMap ProtocolServer (ClientVar msg) -> IO ()
+closeProtocolServerClients tcpTimeout cs = readTVarIO cs >>= mapM_ (forkIO . closeClient) >> atomically (writeTVar cs M.empty)
+  where
+    closeClient cVar =
+      tcpTimeout `timeout` atomically (readTMVar cVar) >>= \case
+        Just (Right client) -> closeProtocolClient client `catchAll_` pure ()
         _ -> pure ()
 
-cancelActions :: Foldable f => TVar (f (Async ())) -> IO ()
-cancelActions as = readTVarIO as >>= mapM_ uninterruptibleCancel
+cancelActions :: (Foldable f, Monoid (f (Async ()))) => TVar (f (Async ())) -> IO ()
+cancelActions as = readTVarIO as >>= mapM_ uninterruptibleCancel >> atomically (writeTVar as mempty)
 
 withAgentLock :: MonadUnliftIO m => AgentClient -> m a -> m a
 withAgentLock AgentClient {lock} =

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

@@ -45,7 +45,7 @@ import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (getCurrentTime)
-import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLError, ToRow, field)
+import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLError, ToRow, field, (:.) (..))
 import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.FromField
 import Database.SQLite.Simple.QQ (sql)
@@ -587,16 +587,16 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
             db
             [sql|
               SELECT s.ntf_host, s.ntf_port, s.ntf_key_hash,
-                t.tkn_id, t.tkn_pub_key, t.tkn_priv_key, t.tkn_pub_dh_key, t.tkn_priv_dh_key, t.tkn_dh_secret, t.tkn_status, t.tkn_action
+                t.provider, t.device_token, t.tkn_id, t.tkn_pub_key, t.tkn_priv_key, t.tkn_pub_dh_key, t.tkn_priv_dh_key, t.tkn_dh_secret, t.tkn_status, t.tkn_action
               FROM ntf_tokens t
               JOIN ntf_servers s USING (ntf_host, ntf_port)
             |]
       pure . first listToMaybe $ partition ((t ==) . deviceToken) tokens
     where
-      ntfToken (host, port, keyHash, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction) =
+      ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction)) =
         let ntfServer = ProtocolServer {host, port, keyHash}
             ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey)
-         in NtfToken {deviceToken = t, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction}
+         in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction}
 
   updateNtfTokenRegistration :: SQLiteStore -> NtfToken -> NtfTokenId -> C.DhSecretX25519 -> m ()
   updateNtfTokenRegistration st NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} tknId ntfDhSecret =

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

@@ -19,7 +19,7 @@ CREATE TABLE ntf_servers (
 
 CREATE TABLE ntf_tokens (
   provider TEXT NOT NULL, -- apn
-  device_token TEXT NOT NULL,
+  device_token TEXT NOT NULL, -- ! this field is mislabeled and is actually saved as binary
   ntf_host TEXT NOT NULL,
   ntf_port TEXT NOT NULL,
   tkn_id BLOB, -- token ID assigned by notifications server

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -0,0 +1,161 @@
+CREATE TABLE migrations(
+  name TEXT NOT NULL,
+  ts TEXT NOT NULL,
+  PRIMARY KEY(name)
+);
+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,
+  snd_public_key BLOB,
+  e2e_pub_key BLOB,
+  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
+);
+CREATE TABLE ntf_servers(
+  ntf_host TEXT NOT NULL,
+  ntf_port TEXT NOT NULL,
+  ntf_key_hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now')),
+  PRIMARY KEY(ntf_host, ntf_port)
+) WITHOUT ROWID;
+CREATE TABLE ntf_tokens(
+  provider TEXT NOT NULL, -- apn
+  device_token TEXT NOT NULL, -- ! this field is mislabeled and is actually saved as binary
+  ntf_host TEXT NOT NULL,
+  ntf_port TEXT NOT NULL,
+  tkn_id BLOB, -- token ID assigned by notifications server
+  tkn_pub_key BLOB NOT NULL, -- client's public key to verify token commands(used by server, for repeat registraions)
+tkn_priv_key BLOB NOT NULL, -- client's private key to sign token commands
+tkn_pub_dh_key BLOB NOT NULL, -- client's public DH key(for repeat registraions)
+tkn_priv_dh_key BLOB NOT NULL, -- client's private DH key(for repeat registraions)
+tkn_dh_secret BLOB, -- DH secret for e2e encryption of notifications
+  tkn_status TEXT NOT NULL,
+  tkn_action BLOB,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now')), -- this is to check token status periodically to know when it was last checked
+  PRIMARY KEY(provider, device_token, ntf_host, ntf_port),
+  FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
+  ON DELETE RESTRICT ON UPDATE CASCADE
+) WITHOUT ROWID;

src/Simplex/Messaging/Notifications/Client.hs

@@ -48,7 +48,7 @@ ntfEnableCron c pKey tknId int = okNtfCommand (TCRN int) c pKey tknId
 ntfCreateSubsciption :: NtfClient -> C.APrivateSignKey -> NewNtfEntity 'Subscription -> ExceptT ProtocolClientError IO (NtfSubscriptionId, C.PublicKeyX25519)
 ntfCreateSubsciption c pKey newSub =
   sendNtfCommand c (Just pKey) "" (SNEW newSub) >>= \case
-    NRId tknId dhKey -> pure (tknId, dhKey)
+    NRId subId dhKey -> pure (subId, dhKey)
     _ -> throwE PCEUnexpectedResponse
 
 ntfCheckSubscription :: NtfClient -> C.APrivateSignKey -> NtfSubscriptionId -> ExceptT ProtocolClientError IO NtfSubStatus

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -120,7 +120,7 @@ instance ToJSON NtfRegCode where
 
 data NewNtfEntity (e :: NtfEntity) where
   NewNtfTkn :: DeviceToken -> C.APublicVerifyKey -> C.PublicKeyX25519 -> NewNtfEntity 'Token
-  NewNtfSub :: NtfTokenId -> SMPQueueNtf -> NewNtfEntity 'Subscription
+  NewNtfSub :: NtfTokenId -> SMPQueueNtf -> NewNtfEntity 'Subscription -- NtfTokenId -> C.APublicVerifyKey -> SMPQueueNtf
 
 deriving instance Show (NewNtfEntity e)
 

src/Simplex/Messaging/Notifications/Server/Push/testpush.sh

@@ -15,6 +15,12 @@ export JWT_HEADER_CLAIMS="${JWT_HEADER}.${JWT_CLAIMS}"
 export JWT_SIGNED_HEADER_CLAIMS=$(printf "${JWT_HEADER_CLAIMS}" | openssl dgst -binary -sha256 -sign "${APNS_KEY_FILE}" | openssl base64 -e -A | tr -- '+/' '-_' | tr -d =)
 export AUTHENTICATION_TOKEN="${JWT_HEADER}.${JWT_CLAIMS}.${JWT_SIGNED_HEADER_CLAIMS}"
 
-curl -v --header "apns-topic: $TOPIC" --header "apns-push-type: alert" --header "authorization: bearer $AUTHENTICATION_TOKEN" --data '{"aps":{"alert":"you have a new message"},"data":{"test":"123"}}' --http2 https://${APNS_HOST_NAME}/3/device/${DEVICE_TOKEN}
+# simple alert
+# curl -v --header "apns-topic: $TOPIC" --header "apns-push-type: alert" --header "authorization: bearer $AUTHENTICATION_TOKEN" --data '{"aps":{"alert":"you have a new message"},"data":{"test":"123"}}' --http2 https://${APNS_HOST_NAME}/3/device/${DEVICE_TOKEN}
 
+# background notification
 # curl -v --header "apns-topic: $TOPIC" --header "apns-push-type: background" --header "apns-priority: 5" --header "authorization: bearer $AUTHENTICATION_TOKEN" --data '{"aps":{"content-available":1}}' --http2 https://${APNS_HOST_NAME}/3/device/${DEVICE_TOKEN}
+
+# mutable-content notification
+# NTF_CAT_CHECK_MESSAGE category will not show alert if the app is in foreground
+curl -v --header "apns-topic: $TOPIC" --header "apns-push-type: alert" --header "authorization: bearer $AUTHENTICATION_TOKEN" --data '{"aps":{"category": "NTF_CAT_CHECK_MESSAGE__SECRET", "mutable-content": 1, "alert":"received encrypted message"}, "data": {"test":"123"}}' --http2 https://${APNS_HOST_NAME}/3/device/${DEVICE_TOKEN}

src/Simplex/Messaging/Server/CLI.hs

@@ -26,7 +26,7 @@ import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, removeDirectoryRecursive)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
-import System.IO (BufferMode (..), IOMode (..), hGetLine, hSetBuffering, stderr, stdout, withFile)
+import System.IO (BufferMode (..), IOMode (..), hFlush, hGetLine, hSetBuffering, stderr, stdout, withFile)
 import System.Process (readCreateProcess, shell)
 import Text.Read (readMaybe)
 
@@ -58,11 +58,22 @@ protocolServerCLI cliCfg@ServerCLIConfig {iniFile, executableName} server =
       doesFileExist iniFile >>= \case
         True -> readIniFile iniFile >>= either exitError (runServer cliCfg server)
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
-    Delete -> cleanup cliCfg >> putStrLn "Deleted configuration and log files"
+    Delete -> do
+      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      cleanup cliCfg
+      putStrLn "Deleted configuration and log files"
 
 exitError :: String -> IO ()
 exitError msg = putStrLn msg >> exitFailure
 
+confirmOrExit :: String -> IO ()
+confirmOrExit s = do
+  putStrLn s
+  putStr "Continue (Y/n): "
+  hFlush stdout
+  ok <- getLine
+  when (ok /= "Y") exitFailure
+
 data CliCommand
   = Init InitOptions
   | Start

src/Simplex/Messaging/Transport.hs

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

src/Simplex/Messaging/Transport/Server.hs

@@ -12,20 +12,22 @@ module Simplex.Messaging.Transport.Server
   )
 where
 
+import Control.Concurrent.STM (stateTVar)
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import qualified Crypto.Store.X509 as SX
 import Data.Default (def)
-import Data.Set (Set)
-import qualified Data.Set as S
 import qualified Data.X509 as X
 import Data.X509.Validation (Fingerprint (..))
 import qualified Data.X509.Validation as XV
 import Network.Socket
 import qualified Network.TLS as T
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (catchAll_)
 import System.Exit (exitFailure)
+import System.Mem.Weak (Weak, deRefWeak)
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -36,37 +38,29 @@ import UnliftIO.STM
 runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()
 runTransportServer started port serverParams server = do
   u <- askUnliftIO
-  liftIO $ do
-    clients <- newTVarIO S.empty
+  liftIO . runTCPServer started port $ \conn ->
     E.bracket
-      (startTCPServer started port)
-      (closeServer started clients)
-      $ \sock -> forever $ do
-        (connSock, _) <- accept sock
-        tid <- forkIO $ connectClient u connSock `catchAll_` close connSock `catchAll_` pure ()
-        atomically . modifyTVar' clients $ S.insert tid
-  where
-    connectClient :: UnliftIO m -> Socket -> IO ()
-    connectClient u connSock =
-      E.bracket
-        (connectTLS serverParams connSock >>= getServerConnection)
-        closeConnection
-        (unliftIO u . server)
+      (connectTLS serverParams conn >>= getServerConnection)
+      closeConnection
+      (unliftIO u . server)
 
+-- | Run TCP server without TLS
 runTCPServer :: TMVar Bool -> ServiceName -> (Socket -> IO ()) -> IO ()
 runTCPServer started port server = do
-  clients <- newTVarIO S.empty
+  clients <- atomically TM.empty
+  clientId <- newTVarIO 0
   E.bracket
     (startTCPServer started port)
     (closeServer started clients)
-    $ \sock -> forever $ do
-      (connSock, _) <- accept sock
-      tid <- forkIO $ server connSock `catchAll_` pure ()
-      atomically . modifyTVar' clients $ S.insert tid
+    $ \sock -> forever . E.bracketOnError (accept sock) (close . fst) $ \(conn, _peer) -> do
+      -- catchAll_ is needed here in case the connection was closed earlier
+      cId <- atomically $ stateTVar clientId $ \cId -> (cId + 1, cId + 1)
+      tId <- mkWeakThreadId =<< forkFinally (server conn) (const $ gracefulClose conn 5000 `catchAll_` atomically (TM.delete cId clients))
+      atomically $ TM.insert cId tId clients
 
-closeServer :: TMVar Bool -> TVar (Set ThreadId) -> Socket -> IO ()
+closeServer :: TMVar Bool -> TMap Int (Weak ThreadId) -> Socket -> IO ()
 closeServer started clients sock = do
-  readTVarIO clients >>= mapM_ killThread
+  readTVarIO clients >>= mapM_ (deRefWeak >=> mapM_ killThread)
   close sock
   void . atomically $ tryPutTMVar started False
 

tests/AgentTests.hs

@@ -13,6 +13,7 @@ import AgentTests.ConnectionRequestTests
 import AgentTests.DoubleRatchetTests (doubleRatchetTests)
 import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.SQLiteTests (storeTests)
+import AgentTests.SchemaDump (schemaDumpTest)
 import Control.Concurrent
 import Control.Monad (forM_)
 import Data.ByteString.Char8 (ByteString)
@@ -36,6 +37,7 @@ agentTests (ATransport t) = do
   describe "Double ratchet tests" doubleRatchetTests
   describe "Functional API" $ functionalAPITests (ATransport t)
   describe "SQLite store" storeTests
+  describe "SQLite schema dump" schemaDumpTest
   describe "SMP agent protocol syntax" $ syntaxTests t
   describe "Establishing duplex connection" $ do
     it "should connect via one server and one agent" $

tests/AgentTests/SchemaDump.hs

@@ -0,0 +1,29 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+module AgentTests.SchemaDump where
+
+import Control.Monad (void)
+import Simplex.Messaging.Agent.Store.SQLite
+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import System.Process (readCreateProcess, shell)
+import Test.Hspec
+
+testDB :: FilePath
+testDB = "tests/tmp/test_agent_schema.db"
+
+schema :: FilePath
+schema = "src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql"
+
+schemaDumpTest :: Spec
+schemaDumpTest =
+  it "verify and overwrite schema dump" testVerifySchemaDump
+
+testVerifySchemaDump :: IO ()
+testVerifySchemaDump = do
+  void $ createSQLiteStore testDB 1 Migrations.app False
+  void $ readCreateProcess (shell $ "touch " <> schema) ""
+  savedSchema <- readFile schema
+  savedSchema `seq` pure ()
+  void $ readCreateProcess (shell $ "sqlite3 " <> testDB <> " '.schema --indent' > " <> schema) ""
+  currentSchema <- readFile schema
+  savedSchema `shouldBe` currentSchema