smp: support client notices (#1659)

* agent: support client notices

* improve

* fix, test

* rename

* cleanup

* send and process notices in more cases

* dont delete

* dont remove notice on other permanent errors

* dont remove notice if there is no notice ID in queue

* add server to error

* allow deleting

* only use notice if key hash matches

Author: epoberezkin

simplexmq.cabal

@@ -103,6 +103,7 @@ library
       Simplex.Messaging.Agent.Store.AgentStore
       Simplex.Messaging.Agent.Store.Common
       Simplex.Messaging.Agent.Store.DB
+      Simplex.Messaging.Agent.Store.Entity
       Simplex.Messaging.Agent.Store.Interface
       Simplex.Messaging.Agent.Store.Migrations
       Simplex.Messaging.Agent.Store.Migrations.App
@@ -130,12 +131,13 @@ library
       Simplex.Messaging.Notifications.Types
       Simplex.Messaging.Parsers
       Simplex.Messaging.Protocol
+      Simplex.Messaging.Protocol.Types
       Simplex.Messaging.Server.Expiration
       Simplex.Messaging.Server.QueueStore.Postgres.Config
       Simplex.Messaging.Server.QueueStore.QueueInfo
       Simplex.Messaging.ServiceScheme
       Simplex.Messaging.Session
-      Simplex.Messaging.Agent.Store.Entity
+      Simplex.Messaging.SystemTime
       Simplex.Messaging.TMap
       Simplex.Messaging.Transport
       Simplex.Messaging.Transport.Buffer
@@ -164,6 +166,7 @@ library
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250322_short_links
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250702_conn_invitations_remove_cascade_delete
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251009_queue_to_subscribe
+          Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251010_client_notices
   else
       exposed-modules:
           Simplex.Messaging.Agent.Store.SQLite
@@ -212,6 +215,7 @@ library
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20250322_short_links
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20250702_conn_invitations_remove_cascade_delete
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20251009_queue_to_subscribe
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20251010_client_notices
   if flag(client_postgres) || flag(server_postgres)
       exposed-modules:
           Simplex.Messaging.Agent.Store.Postgres

src/Simplex/FileTransfer/Server.hs

@@ -58,8 +58,9 @@ import Simplex.Messaging.Protocol (BlockingInfo, EntityId (..), RcvPublicAuthKey
 import Simplex.Messaging.Server (controlPortAuth, dummyVerifyCmd, verifyCmdAuthorization)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, ServerEntityStatus (..), getRoundedSystemTime)
+import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
 import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (CertChainPubKey (..), SessionId, THandleAuth (..), THandleParams (..), TransportPeer (..), defaultSupportedParams)
@@ -451,7 +452,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
         let rIds = L.map (\(FileRecipient rId _) -> rId) rcps
         pure $ FRSndIds sId rIds
       pure $ either FRErr id r
-    addFileRetry :: FileStore -> FileInfo -> Int -> RoundedSystemTime -> M (Either XFTPErrorType XFTPFileId)
+    addFileRetry :: FileStore -> FileInfo -> Int -> RoundedFileTime -> M (Either XFTPErrorType XFTPFileId)
     addFileRetry st file n ts =
       retryAdd n $ \sId -> runExceptT $ do
         ExceptT $ addFile st sId file ts EntityActive
@@ -579,8 +580,8 @@ deleteOrBlockServerFile_ FileRec {filePath, fileInfo} stat storeAction = runExce
       liftIO $ atomicModifyIORef'_ (filesCount stats) (subtract 1)
       liftIO $ atomicModifyIORef'_ (filesSize stats) (subtract $ fromIntegral $ size fileInfo)
 
-getFileTime :: IO RoundedSystemTime
-getFileTime = getRoundedSystemTime fileTimePrecision
+getFileTime :: IO RoundedFileTime
+getFileTime = getRoundedSystemTime
 
 expireServerFiles :: Maybe Int -> ExpirationConfig -> M ()
 expireServerFiles itemDelay expCfg = do

src/Simplex/FileTransfer/Server/Store.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -8,6 +9,7 @@ module Simplex.FileTransfer.Server.Store
   ( FileStore (..),
     FileRec (..),
     FileRecipient (..),
+    RoundedFileTime,
     newFileStore,
     addFile,
     setFilePath,
@@ -33,7 +35,8 @@ import Simplex.FileTransfer.Transport (XFTPErrorType (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BlockingInfo, RcvPublicAuthKey, RecipientId, SenderId)
-import Simplex.Messaging.Server.QueueStore (RoundedSystemTime (..), ServerEntityStatus (..))
+import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
+import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (ifM, ($>>=))
@@ -49,10 +52,12 @@ data FileRec = FileRec
     fileInfo :: FileInfo,
     filePath :: TVar (Maybe FilePath),
     recipientIds :: TVar (Set RecipientId),
-    createdAt :: RoundedSystemTime,
+    createdAt :: RoundedFileTime,
     fileStatus :: TVar ServerEntityStatus
   }
 
+type RoundedFileTime = RoundedSystemTime 3600
+
 fileTimePrecision :: Int64
 fileTimePrecision = 3600 -- truncate creation time to 1 hour
 
@@ -70,14 +75,14 @@ newFileStore = do
   usedStorage <- newTVarIO 0
   pure FileStore {files, recipients, usedStorage}
 
-addFile :: FileStore -> SenderId -> FileInfo -> RoundedSystemTime -> ServerEntityStatus -> STM (Either XFTPErrorType ())
+addFile :: FileStore -> SenderId -> FileInfo -> RoundedFileTime -> ServerEntityStatus -> STM (Either XFTPErrorType ())
 addFile FileStore {files} sId fileInfo createdAt status =
   ifM (TM.member sId files) (pure $ Left DUPLICATE_) $ do
     f <- newFileRec sId fileInfo createdAt status
     TM.insert sId f files
     pure $ Right ()
 
-newFileRec :: SenderId -> FileInfo -> RoundedSystemTime -> ServerEntityStatus -> STM FileRec
+newFileRec :: SenderId -> FileInfo -> RoundedFileTime -> ServerEntityStatus -> STM FileRec
 newFileRec senderId fileInfo createdAt status = do
   recipientIds <- newTVar S.empty
   filePath <- newTVar Nothing

src/Simplex/FileTransfer/Server/StoreLog.hs

@@ -34,13 +34,13 @@ import Simplex.FileTransfer.Protocol (FileInfo (..))
 import Simplex.FileTransfer.Server.Store
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BlockingInfo, RcvPublicAuthKey, RecipientId, SenderId)
-import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, ServerEntityStatus (..))
+import Simplex.Messaging.Server.QueueStore (ServerEntityStatus (..))
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Util (bshow)
 import System.IO
 
 data FileStoreLogRecord
-  = AddFile SenderId FileInfo RoundedSystemTime ServerEntityStatus
+  = AddFile SenderId FileInfo RoundedFileTime ServerEntityStatus
   | PutFile SenderId FilePath
   | AddRecipients SenderId (NonEmpty FileRecipient)
   | DeleteFile SenderId
@@ -69,7 +69,7 @@ instance StrEncoding FileStoreLogRecord where
 logFileStoreRecord :: StoreLog 'WriteMode -> FileStoreLogRecord -> IO ()
 logFileStoreRecord = writeStoreLogRecord
 
-logAddFile :: StoreLog 'WriteMode -> SenderId -> FileInfo -> RoundedSystemTime -> ServerEntityStatus -> IO ()
+logAddFile :: StoreLog 'WriteMode -> SenderId -> FileInfo -> RoundedFileTime -> ServerEntityStatus -> IO ()
 logAddFile s = logFileStoreRecord s .:: AddFile
 
 logPutFile :: StoreLog 'WriteMode -> SenderId -> FilePath -> IO ()

src/Simplex/FileTransfer/Types.hs

@@ -15,14 +15,14 @@ import Data.Text.Encoding (encodeUtf8)
 import Data.Word (Word32)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
 import Simplex.FileTransfer.Description
+import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..), fromTextField_)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile (..))
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol (XFTPServer)
 import System.FilePath ((</>))
-import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..), fromTextField_)
 
 type RcvFileId = ByteString -- Agent entity ID
 

src/Simplex/Messaging/Agent.hs

@@ -151,7 +151,7 @@ import qualified Data.Aeson.TH as JQ
 import Data.Bifunctor (bimap, first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Composition ((.:), (.:.), (.::), (.::.))
+import Data.Composition
 import Data.Either (isRight, partitionEithers, rights)
 import Data.Foldable (foldl', toList)
 import Data.Functor (($>))
@@ -189,10 +189,11 @@ import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.AgentStore
 import Simplex.Messaging.Agent.Store.Common (DBStore)
 import qualified Simplex.Messaging.Agent.Store.DB as DB
+import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Agent.Store.Interface (closeDBStore, execSQL, getCurrentMigrations)
 import Simplex.Messaging.Agent.Store.Shared (UpMigration (..), upMigration)
 import qualified Simplex.Messaging.Agent.TSessionSubs as SS
-import Simplex.Messaging.Client (NetworkRequestMode (..), SMPClientError, ServerTransmission (..), ServerTransmissionBatch, nonBlockingWriteTBQueue, temporaryClientError, unexpectedResponse)
+import Simplex.Messaging.Client (NetworkRequestMode (..), SMPClientError, ServerTransmission (..), ServerTransmissionBatch, nonBlockingWriteTBQueue, smpErrorClientNotice, temporaryClientError, unexpectedResponse)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile, CryptoFileArgs)
 import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
@@ -227,7 +228,7 @@ import Simplex.Messaging.Protocol
   )
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
-import Simplex.Messaging.Agent.Store.Entity
+import Simplex.Messaging.SystemTime
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion)
 import Simplex.Messaging.Util
@@ -251,13 +252,14 @@ getSMPAgentClient = getSMPAgentClient_ 1
 {-# INLINE getSMPAgentClient #-}
 
 getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> Bool -> IO AgentClient
-getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp} store backgroundMode =
+getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, presetServers} store backgroundMode =
   newSMPAgentEnv cfg store >>= runReaderT runAgent
   where
     runAgent = do
       liftIO $ checkServers "SMP" smp >> checkServers "XFTP" xftp
       currentTs <- liftIO getCurrentTime
-      c@AgentClient {acThread} <- liftIO . newAgentClient clientId initServers currentTs =<< ask
+      notices <- liftIO $ withTransaction store (`getClientNotices` presetServers) `catchAll_` pure []
+      c@AgentClient {acThread} <- liftIO . newAgentClient clientId initServers currentTs notices =<< ask
       t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
       atomically . writeTVar acThread . Just =<< mkWeakThreadId t
       pure c
@@ -379,8 +381,8 @@ deleteConnectionsAsync c waitDelivery = withAgentEnv c . deleteConnectionsAsync'
 {-# INLINE deleteConnectionsAsync #-}
 
 -- | Create SMP agent connection (NEW command)
-createConnection :: ConnectionModeI c => AgentClient -> NetworkRequestMode -> UserId -> Bool -> SConnectionMode c -> Maybe UserLinkData -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> AE (ConnId, (CreatedConnLink c, Maybe ClientServiceId))
-createConnection c nm userId enableNtfs = withAgentEnv c .::. newConn c nm userId enableNtfs
+createConnection :: ConnectionModeI c => AgentClient -> NetworkRequestMode -> UserId -> Bool -> Bool -> SConnectionMode c -> Maybe UserLinkData -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> AE (ConnId, (CreatedConnLink c, Maybe ClientServiceId))
+createConnection c nm userId enableNtfs checkNotices = withAgentEnv c .::. newConn c nm userId enableNtfs checkNotices
 {-# INLINE createConnection #-}
 
 -- | Create or update user's contact connection short link
@@ -863,13 +865,27 @@ switchConnectionAsync' c corrId connId =
             connectionStats c $ DuplexConnection cData rqs' sqs
       _ -> throwE $ CMD PROHIBITED "switchConnectionAsync: not duplex"
 
-newConn :: ConnectionModeI c => AgentClient -> NetworkRequestMode -> UserId -> Bool -> SConnectionMode c -> Maybe UserLinkData -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> AM (ConnId, (CreatedConnLink c, Maybe ClientServiceId))
-newConn c nm userId enableNtfs cMode userData_ clientData pqInitKeys subMode = do
+newConn :: ConnectionModeI c => AgentClient -> NetworkRequestMode -> UserId -> Bool -> Bool -> SConnectionMode c -> Maybe UserLinkData -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> AM (ConnId, (CreatedConnLink c, Maybe ClientServiceId))
+newConn c nm userId enableNtfs checkNotices cMode userData_ clientData pqInitKeys subMode = do
   srv <- getSMPServer c userId
+  when (checkNotices && connMode cMode == CMContact) $ checkClientNotices c srv
   connId <- newConnNoQueues c userId enableNtfs cMode (CR.connPQEncryption pqInitKeys)
   (connId,) <$> newRcvConnSrv c nm userId connId enableNtfs cMode userData_ clientData pqInitKeys subMode srv
     `catchE` \e -> withStore' c (`deleteConnRecord` connId) >> throwE e
 
+checkClientNotices :: AgentClient -> SMPServerWithAuth -> AM ()
+checkClientNotices AgentClient {clientNotices, presetServers} (ProtoServerWithAuth srv@(ProtocolServer {host}) _) = do
+  notices <- readTVarIO clientNotices
+  unless (M.null notices) $ checkNotices notices =<< liftIO getSystemSeconds
+  where
+    srvKey
+      | isPresetServer srv presetServers = Nothing -- Nothing is used as key for preset servers
+      | otherwise = Just srv
+    checkNotices notices ts =
+      forM_ (M.lookup srvKey notices) $ \expires_ ->
+        when (maybe True (ts <) expires_) $
+          throwError NOTICE {server = safeDecodeUtf8 $ strEncode $ L.head host, preset = isNothing srvKey, expiresAt = roundedToUTCTime <$> expires_}
+
 setConnShortLink' :: AgentClient -> NetworkRequestMode -> ConnId -> SConnectionMode c -> UserLinkData -> Maybe CRClientData -> AM (ConnShortLink c)
 setConnShortLink' c nm connId cMode userData clientData =
   withConnLock c connId "setConnShortLink" $ do
@@ -2794,18 +2810,20 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
     STEvent msgOrErr ->
       withRcvConn entId $ \rq@RcvQueue {connId} conn -> case msgOrErr of
         Right msg -> runProcessSMP rq conn (toConnData conn) msg
-        Left e -> lift $ notifyErr connId e
+        Left e -> lift $ do
+          processClientNotice rq e
+          notifyErr connId e
     STResponse (Cmd SRecipient cmd) respOrErr ->
       withRcvConn entId $ \rq conn -> case cmd of
         SMP.SUB -> case respOrErr of
-          Right SMP.OK -> processSubOk rq upConnIds
+          Right SMP.OK -> liftIO $ processSubOk rq upConnIds
           -- TODO [certs rcv] associate queue with the service
-          Right (SMP.SOK serviceId_) -> processSubOk rq upConnIds
+          Right (SMP.SOK serviceId_) -> liftIO $ processSubOk rq upConnIds
           Right msg@SMP.MSG {} -> do
-            processSubOk rq upConnIds -- the connection is UP even when processing this particular message fails
+            liftIO $ processSubOk rq upConnIds -- the connection is UP even when processing this particular message fails
             runProcessSMP rq conn (toConnData conn) msg
-          Right r -> processSubErr rq $ unexpectedResponse r
-          Left e -> unless (temporaryClientError e) $ processSubErr rq e -- timeout/network was already reported
+          Right r -> lift $ processSubErr rq $ unexpectedResponse r
+          Left e -> lift $ unless (temporaryClientError e) $ processSubErr rq e -- timeout/network was already reported
         SMP.ACK _ -> case respOrErr of
           Right msg@SMP.MSG {} -> runProcessSMP rq conn (toConnData conn) msg
           _ -> pure () -- TODO process OK response to ACK
@@ -2827,21 +2845,28 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
           tryAllErrors' (a rq conn) >>= \case
             Left e -> notify' connId (ERR e)
             Right () -> pure ()
-    processSubOk :: RcvQueue -> TVar [ConnId] -> AM ()
+    processSubOk :: RcvQueue -> TVar [ConnId] -> IO ()
     processSubOk rq@RcvQueue {connId} upConnIds =
       atomically . whenM (isPendingSub rq) $ do
         SS.addActiveSub tSess sessId (rcvQueueSub rq) $ currentSubs c
         modifyTVar' upConnIds (connId :)
-    processSubErr :: RcvQueue -> SMPClientError -> AM ()
+    processSubErr :: RcvQueue -> SMPClientError -> AM' ()
     processSubErr rq@RcvQueue {connId} e = do
       atomically . whenM (isPendingSub rq) $
         failSubscription c tSess rq e >> incSMPServerStat c userId srv connSubErrs
-      lift $ notifyErr connId e
+      processClientNotice rq e
+      notifyErr connId e
     isPendingSub :: RcvQueue -> STM Bool
     isPendingSub rq = do
       pending <- (&&) <$> SS.hasPendingSub tSess (queueId rq) (currentSubs c) <*> activeClientSession c tSess sessId
       unless pending $ incSMPServerStat c userId srv connSubIgnored
       pure pending
+    processClientNotice rq e =
+      forM_ (smpErrorClientNotice e) $ \notice_ ->
+        E.bracket_
+          (atomically $ takeTMVar $ clientNoticesLock c)
+          (atomically $ putTMVar (clientNoticesLock c) ())
+          (processClientNotices c tSess [(rcvQueueSub rq, notice_)])
     notify' :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m ()
     notify' connId msg = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) msg)
     notifyErr :: ConnId -> SMPClientError -> AM' ()