Merge branch 'master' into master-ghc9

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.4.0.2
+version: 5.4.0.3
 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:        5.4.0.2
+version:        5.4.0.3
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Notifications/Server.hs

@@ -275,12 +275,11 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     updateSubStatus smpQueue status = do
       st <- asks store
-      atomically (findNtfSubscription st smpQueue)
-        >>= mapM_
-          ( \NtfSubData {ntfSubId, subStatus} -> do
-              atomically $ writeTVar subStatus status
-              withNtfLog $ \sl -> logSubscriptionStatus sl ntfSubId status
-          )
+      atomically (findNtfSubscription st smpQueue) >>= mapM_ update
+      where
+        update NtfSubData {ntfSubId, subStatus} = do
+          old <- atomically $ stateTVar subStatus (,status)
+          when (old /= status) $ withNtfLog $ \sl -> logSubscriptionStatus sl ntfSubId status
 
 ntfPush :: NtfPushServer -> M ()
 ntfPush s@NtfPushServer {pushQ} = forever $ do
@@ -292,7 +291,10 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
       | status /= NTInvalid && status /= NTExpired ->
         deliverNotification pp tkn ntf >>= \case
           Right _ -> do
-            status_ <- atomically $ stateTVar tknStatus $ \status' -> if status' == NTActive then (Nothing, NTActive) else (Just NTConfirmed, NTConfirmed)
+            status_ <- atomically $ stateTVar tknStatus $ \case
+              NTActive -> (Nothing, NTActive)
+              NTConfirmed -> (Nothing, NTConfirmed)
+              _ -> (Just NTConfirmed, NTConfirmed)
             forM_ status_ $ \status' -> withNtfLog $ \sl -> logTokenStatus sl ntfTknId status'
           _ -> pure ()
       | otherwise -> logError "bad notification token status"
@@ -309,7 +311,7 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
       | status == NTActive = action
       | otherwise = liftIO $ logError "bad notification token status"
     deliverNotification :: PushProvider -> NtfTknData -> PushNotification -> M (Either PushProviderError ())
-    deliverNotification pp tkn@NtfTknData {ntfTknId, tknStatus} ntf = do
+    deliverNotification pp tkn ntf = do
       deliver <- liftIO $ getPushClient s pp
       liftIO (runExceptT $ deliver tkn ntf) >>= \case
         Right _ -> pure $ Right ()
@@ -318,19 +320,20 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
           PPRetryLater -> retryDeliver
           PPCryptoError _ -> err e
           PPResponseError _ _ -> err e
-          PPTokenInvalid -> updateTknStatus NTInvalid >> err e
+          PPTokenInvalid -> updateTknStatus tkn NTInvalid >> err e
           PPPermanentError -> err e
       where
         retryDeliver :: M (Either PushProviderError ())
         retryDeliver = do
           deliver <- liftIO $ newPushClient s pp
           liftIO (runExceptT $ deliver tkn ntf) >>= either err (pure . Right)
-        updateTknStatus :: NtfTknStatus -> M ()
-        updateTknStatus status = do
-          atomically $ writeTVar tknStatus status
-          withNtfLog $ \sl -> logTokenStatus sl ntfTknId status
         err e = logError (T.pack $ "Push provider error (" <> show pp <> "): " <> show e) $> Left e
 
+updateTknStatus :: NtfTknData -> NtfTknStatus -> M ()
+updateTknStatus NtfTknData {ntfTknId, tknStatus} status = do
+  old <- atomically $ stateTVar tknStatus (,status)
+  when (old /= status) $ withNtfLog $ \sl -> logTokenStatus sl ntfTknId status
+
 runNtfClientTransport :: Transport c => THandle c -> M ()
 runNtfClientTransport th@THandle {sessionId} = do
   qSize <- asks $ clientQSize . config
@@ -462,10 +465,9 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
             | (status == NTRegistered || status == NTConfirmed || status == NTActive) && tknRegCode == code -> do
               logDebug "TVFY - token verified"
               st <- asks store
-              atomically $ writeTVar tknStatus NTActive
+              updateTknStatus tkn NTActive
               tIds <- atomically $ removeInactiveTokenRegistrations st tkn
               forM_ tIds cancelInvervalNotifications
-              withNtfLog $ \s -> logTokenStatus s tknId NTActive
               incNtfStat tknVerified
               pure NROk
             | otherwise -> do

src/Simplex/Messaging/Notifications/Server/Env.hs

@@ -3,11 +3,13 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 
 module Simplex.Messaging.Notifications.Server.Env where
 
 import Control.Concurrent (ThreadId)
 import Control.Concurrent.Async (Async)
+import Control.Logger.Simple
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
@@ -84,7 +86,9 @@ newNtfServerEnv :: (MonadUnliftIO m, MonadRandom m) => NtfServerConfig -> m NtfE
 newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile} = do
   idsDrg <- newTVarIO =<< drgNew
   store <- atomically newNtfStore
+  logInfo "restoring subscriptions..."
   storeLog <- liftIO $ mapM (`readWriteNtfStore` store) storeLogFile
+  logInfo "restored subscriptions"
   subscriber <- atomically $ newNtfSubscriber subQSize smpAgentCfg
   pushServer <- atomically $ newNtfPushServer pushQSize apnsConfig
   tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -30,7 +30,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 ntfServerVersion :: String
-ntfServerVersion = "1.6.1"
+ntfServerVersion = "1.6.2"
 
 defaultSMPBatchDelay :: Int
 defaultSMPBatchDelay = 10000

src/Simplex/Messaging/Notifications/Server/StoreLog.hs

@@ -4,6 +4,7 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE StrictData #-}
 
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
@@ -24,7 +25,7 @@ module Simplex.Messaging.Notifications.Server.StoreLog
 where
 
 import Control.Concurrent.STM
-import Control.Monad (void)
+import Control.Monad
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import qualified Data.ByteString.Char8 as B
 import Data.Word (Word16)
@@ -153,6 +154,7 @@ instance StrEncoding NtfSubRec where
 
 logNtfStoreRecord :: StoreLog 'WriteMode -> NtfStoreLogRecord -> IO ()
 logNtfStoreRecord = writeStoreLogRecord
+{-# INLINE logNtfStoreRecord #-}
 
 logCreateToken :: StoreLog 'WriteMode -> NtfTknData -> IO ()
 logCreateToken s tkn = logNtfStoreRecord s . CreateToken =<< atomically (mkTknRec tkn)
@@ -196,9 +198,11 @@ readNtfStore f st = mapM_ addNtfLogRecord . B.lines =<< B.readFile f
         CreateToken r@NtfTknRec {ntfTknId} -> do
           tkn <- mkTknData r
           addNtfToken st ntfTknId tkn
-        TokenStatus tknId status ->
-          getNtfToken st tknId
-            >>= mapM_ (\NtfTknData {tknStatus} -> writeTVar tknStatus status)
+        TokenStatus tknId status -> do
+          tkn_ <- getNtfToken st tknId
+          forM_ tkn_ $ \tkn@NtfTknData {tknStatus} -> do
+            writeTVar tknStatus status
+            when (status == NTActive) $ void $ removeInactiveTokenRegistrations st tkn
         UpdateToken tknId token' tknRegCode ->
           getNtfToken st tknId
             >>= mapM_
@@ -223,7 +227,5 @@ readNtfStore f st = mapM_ addNtfLogRecord . B.lines =<< B.readFile f
 
 writeNtfStore :: StoreLog 'WriteMode -> NtfStore -> IO ()
 writeNtfStore s NtfStore {tokens, subscriptions} = do
-  atomically (readTVar tokens >>= mapM mkTknRec)
-    >>= mapM_ (writeStoreLogRecord s . CreateToken)
-  atomically (readTVar subscriptions >>= mapM mkSubRec)
-    >>= mapM_ (writeStoreLogRecord s . CreateSubscription)
+  mapM_ (logCreateToken s) =<< readTVarIO tokens
+  mapM_ (logCreateSubscription s) =<< readTVarIO subscriptions