agent: contact request rejection and service requests

simplexmq.cabal

@@ -190,7 +190,7 @@ library
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251230_strict_tables
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260410_receive_attempts
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260411_service_certs
-          Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260712_address_dr
+          Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260712_address_dr_rpc
   else
       exposed-modules:
           Simplex.Messaging.Agent.Store.SQLite
@@ -243,7 +243,7 @@ library
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20251230_strict_tables
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260410_receive_attempts
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260411_service_certs
-          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260712_address_dr
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260712_address_dr_rpc
           Simplex.Messaging.Agent.Store.SQLite.Util
   if flag(client_postgres) || flag(server_postgres)
       exposed-modules:

src/Simplex/Messaging/Agent/Client.hs

@@ -384,6 +384,7 @@ data AgentClient = AgentClient
     clientId :: Int,
     agentEnv :: Env,
     proxySessTs :: TVar UTCTime,
+    serviceRequests :: TMap ConnId (TMVar (Either AgentErrorType SMP.MsgBody)),
     smpServersStats :: TMap (UserId, SMPServer) AgentSMPServerStats,
     xftpServersStats :: TMap (UserId, XFTPServer) AgentXFTPServerStats,
     ntfServersStats :: TMap (UserId, NtfServer) AgentNtfServerStats,
@@ -547,6 +548,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices
   invLocks <- TM.emptyIO
   deleteLock <- createLockIO
   smpSubWorkers <- TM.emptyIO
+  serviceRequests <- TM.emptyIO
   smpServersStats <- TM.emptyIO
   xftpServersStats <- TM.emptyIO
   ntfServersStats <- TM.emptyIO
@@ -592,6 +594,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices
         clientId,
         agentEnv,
         proxySessTs,
+        serviceRequests,
         smpServersStats,
         xftpServersStats,
         ntfServersStats,

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

@@ -153,6 +153,9 @@ data AgentConfig = AgentConfig
     userNetworkInterval :: Int,
     userOfflineDelay :: NominalDiffTime,
     messageTimeout :: NominalDiffTime,
+    serviceRequestTimeout :: NominalDiffTime, -- client side: default time the client waits for a service response (overridable per request)
+    serviceResponseTimeout :: NominalDiffTime, -- service side: time a received service request is valid to respond to
+
     connDeleteDeliveryTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
     quotaExceededTimeout :: NominalDiffTime,
@@ -229,6 +232,8 @@ defaultAgentConfig =
       userNetworkInterval = 1800_000000, -- 30 minutes, should be less than Int32 max value
       userOfflineDelay = 2, -- if network offline event happens in less than 2 seconds after it was set online, it is ignored
       messageTimeout = 2 * nominalDay,
+      serviceRequestTimeout = 30,
+      serviceResponseTimeout = 180,
       connDeleteDeliveryTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
       quotaExceededTimeout = 7 * nominalDay,

src/Simplex/Messaging/Agent/Protocol.hs

@@ -161,6 +161,7 @@ module Simplex.Messaging.Agent.Protocol
     ConnectionErrorType (..),
     BrokerErrorType (..),
     SMPAgentError (..),
+    AgentServiceError (..),
     DroppedMsg (..),
     AgentCryptoError (..),
     cryptoErrToSyncState,
@@ -414,7 +415,9 @@ data AEvent (e :: AEntity) where
   LINK :: ConnShortLink 'CMContact -> UserConnLinkData 'CMContact -> AEvent AEConn
   LDATA :: FixedLinkData 'CMContact -> ConnLinkData 'CMContact -> ConnectionRequestUri 'CMContact -> AEvent AEConn
   CONF :: ConfirmationId -> PQSupport -> [SMPServer] -> ConnInfo -> AEvent AEConn -- ConnInfo is from sender, [SMPServer] will be empty only in v1 handshake
-  REQ :: InvitationId -> PQSupport -> NonEmpty SMPServer -> ConnInfo -> AEvent AEConn -- ConnInfo is from sender
+  REQ :: InvitationId -> PQSupport -> NonEmpty SMPServer -> ConnInfo -> Bool -> AEvent AEConn -- ConnInfo is from sender; Bool - rejection reason can be sent
+  SREQ :: InvitationId -> MsgBody -> AEvent AEConn
+  RJCT :: ConnInfo -> AEvent AEConn
   INFO :: PQSupport -> ConnInfo -> AEvent AEConn
   CON :: PQEncryption -> AEvent AEConn -- notification that connection is established
   END :: AEvent AEConn
@@ -496,6 +499,8 @@ data AEventTag (e :: AEntity) where
   LDATA_ :: AEventTag AEConn
   CONF_ :: AEventTag AEConn
   REQ_ :: AEventTag AEConn
+  SREQ_ :: AEventTag AEConn
+  RJCT_ :: AEventTag AEConn
   INFO_ :: AEventTag AEConn
   CON_ :: AEventTag AEConn
   END_ :: AEventTag AEConn
@@ -559,6 +564,8 @@ aEventTag = \case
   LDATA {} -> LDATA_
   CONF {} -> CONF_
   REQ {} -> REQ_
+  SREQ {} -> SREQ_
+  RJCT {} -> RJCT_
   INFO {} -> INFO_
   CON _ -> CON_
   END -> END_
@@ -858,7 +865,7 @@ data AgentMsgEnvelope
         connReq :: ConnectionRequestUri 'CMInvitation,
         connInfo :: ByteString -- this message is only encrypted with per-queue E2E, not with double ratchet,
       }
-  | AgentInvitationDR -- invitation establishing double ratchet e2e with the contact address that included DR keys
+  | AgentContactRequest -- DR request to a contact address that published DR keys: a contact invitation or a service (RPC) request
       { agentVersion :: VersionSMPA,
         e2eSndParams :: SndE2ERatchetParams 'C.X448,
         ratchetKeyId :: RatchetKeyId,
@@ -879,8 +886,8 @@ instance Encoding AgentMsgEnvelope where
       smpEncode (agentVersion, 'M', Tail encAgentMessage)
     AgentInvitation {agentVersion, connReq, connInfo} ->
       smpEncode (agentVersion, 'I', Large $ strEncode connReq, Tail connInfo)
-    AgentInvitationDR {agentVersion, e2eSndParams, ratchetKeyId, encConnInfo} ->
-      smpEncode (agentVersion, 'J', e2eSndParams, ratchetKeyId, Tail encConnInfo)
+    AgentContactRequest {agentVersion, e2eSndParams, ratchetKeyId, encConnInfo} ->
+      smpEncode (agentVersion, 'A', e2eSndParams, ratchetKeyId, Tail encConnInfo)
     AgentRatchetKey {agentVersion, e2eEncryption, info} ->
       smpEncode (agentVersion, 'R', e2eEncryption, Tail info)
   smpP = do
@@ -896,9 +903,9 @@ instance Encoding AgentMsgEnvelope where
         connReq <- strDecode . unLarge <$?> smpP
         Tail connInfo <- smpP
         pure AgentInvitation {agentVersion, connReq, connInfo}
-      'J' -> do
+      'A' -> do
         (e2eSndParams, ratchetKeyId, Tail encConnInfo) <- smpP
-        pure AgentInvitationDR {agentVersion, e2eSndParams, ratchetKeyId, encConnInfo}
+        pure AgentContactRequest {agentVersion, e2eSndParams, ratchetKeyId, encConnInfo}
       'R' -> do
         e2eEncryption <- smpP
         Tail info <- smpP
@@ -916,6 +923,9 @@ data AgentMessage
     AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo
   | AgentRatchetInfo ByteString
   | AgentMessage APrivHeader AMessage
+  | AgentServiceRequest (NonEmpty SMPQueueInfo) MsgBody
+  | AgentServiceResponse MsgBody
+  | AgentRejection ByteString
   deriving (Show)
 
 instance Encoding AgentMessage where
@@ -924,12 +934,18 @@ instance Encoding AgentMessage where
     AgentConnInfoReply smpQueues cInfo -> smpEncode ('D', smpQueues, Tail cInfo) -- 'D' stands for "duplex"
     AgentRatchetInfo info -> smpEncode ('R', Tail info)
     AgentMessage hdr aMsg -> smpEncode ('M', hdr, aMsg)
+    AgentServiceRequest qs body -> smpEncode ('A', qs, Tail body)
+    AgentServiceResponse body -> smpEncode ('P', Tail body)
+    AgentRejection reason -> smpEncode ('J', Tail reason)
   smpP =
     smpP >>= \case
       'I' -> AgentConnInfo . unTail <$> smpP
       'D' -> AgentConnInfoReply <$> smpP <*> (unTail <$> smpP)
       'R' -> AgentRatchetInfo . unTail <$> smpP
       'M' -> AgentMessage <$> smpP <*> smpP
+      'A' -> AgentServiceRequest <$> smpP <*> (unTail <$> smpP)
+      'P' -> AgentServiceResponse . unTail <$> smpP
+      'J' -> AgentRejection . unTail <$> smpP
       _ -> fail "bad AgentMessage"
 
 -- internal type for storing message type in the database
@@ -946,6 +962,9 @@ data AgentMessageType
   | AM_QUSE_
   | AM_QTEST_
   | AM_EREADY_
+  | AM_SRV_REQ
+  | AM_SRV_RESP
+  | AM_RJCT
   deriving (Eq, Show)
 
 instance Encoding AgentMessageType where
@@ -962,6 +981,9 @@ instance Encoding AgentMessageType where
     AM_QUSE_ -> "QU"
     AM_QTEST_ -> "QT"
     AM_EREADY_ -> "E"
+    AM_SRV_REQ -> "A"
+    AM_SRV_RESP -> "P"
+    AM_RJCT -> "J"
   smpP =
     A.anyChar >>= \case
       'C' -> pure AM_CONN_INFO
@@ -979,6 +1001,9 @@ instance Encoding AgentMessageType where
           'T' -> pure AM_QTEST_
           _ -> fail "bad AgentMessageType"
       'E' -> pure AM_EREADY_
+      'A' -> pure AM_SRV_REQ
+      'P' -> pure AM_SRV_RESP
+      'J' -> pure AM_RJCT
       _ -> fail "bad AgentMessageType"
 
 agentMessageType :: AgentMessage -> AgentMessageType
@@ -987,6 +1012,9 @@ agentMessageType = \case
   AgentConnInfoReply {} -> AM_CONN_INFO_REPLY
   AgentRatchetInfo _ -> AM_RATCHET_INFO
   AgentMessage _ aMsg -> aMessageType aMsg
+  AgentServiceRequest {} -> AM_SRV_REQ
+  AgentServiceResponse {} -> AM_SRV_RESP
+  AgentRejection {} -> AM_RJCT
 
 data APrivHeader = APrivHeader
   { -- | sequential ID assigned by the sending agent
@@ -1860,6 +1888,7 @@ data DRInvitation = DRInvitation
 
 data JoinRequest
   = JRConnReq {enableNtfs :: Bool, joinConnReq :: AConnectionRequestUri, joinPQSupport :: PQSupport}
+  | JRServiceReq {contactReq :: ConnectionRequestUri 'CMContact, joinPQSupport :: PQSupport}
   | JRInvitationDR DRInvitation
   deriving (Show)
 
@@ -2170,8 +2199,16 @@ data SMPAgentError
     A_DUPLICATE {droppedMsg_ :: Maybe DroppedMsg}
   | -- | error in the message to add/delete/etc queue in connection
     A_QUEUE {queueErr :: String}
+  | A_SERVICE {serviceError :: AgentServiceError}
   deriving (Eq, Show, Exception)
 
+data AgentServiceError
+  = ASERejected {rejectReason :: String}
+  | ASETimeout
+  | ASENoPendingRequest
+  | ASENotDRAddress
+  deriving (Eq, Show)
+
 data AgentCryptoError
   = -- | AES decryption error
     DECRYPT_AES
@@ -2202,9 +2239,11 @@ $(J.deriveJSON defaultJSON ''DRInvitation)
 instance StrEncoding JoinRequest where
   strEncode = \case
     JRConnReq ntfs cReq pqSup -> strEncode (ntfs, cReq, pqSup)
+    JRServiceReq cReq pqSup -> strEncode ('S', cReq, pqSup)
     JRInvitationDR dr -> serializeBinary $ LB.toStrict (J'.encode dr)
   strP =
-    (JRConnReq <$> strP <*> _strP <*> (_strP <|> pure PQSupportOff))
+    (A.char 'S' *> (JRServiceReq <$> _strP <*> _strP))
+      <|> (JRConnReq <$> strP <*> _strP <*> (_strP <|> pure PQSupportOff))
       <|> (JRInvitationDR <$> (J'.eitherDecodeStrict' <$?> (A.take =<< (A.decimal <* "\n"))))
 
 -- | SMP agent command and response parser for commands stored in db (fully parses binary bodies)
@@ -2290,6 +2329,8 @@ $(J.deriveJSON (sumTypeJSON id) ''ConnectionErrorType)
 
 $(J.deriveJSON (sumTypeJSON id) ''AgentCryptoError)
 
+$(J.deriveJSON (sumTypeJSON id) ''AgentServiceError)
+
 $(J.deriveJSON defaultJSON ''DroppedMsg)
 
 $(J.deriveJSON (sumTypeJSON id) ''SMPAgentError)

src/Simplex/Messaging/Agent/Store.hs

@@ -465,7 +465,9 @@ data ConnData = ConnData
     lastExternalSndId :: PrevExternalSndId,
     deleted :: Bool,
     ratchetSyncState :: RatchetSyncState,
-    pqSupport :: PQSupport
+    pqSupport :: PQSupport,
+    -- client side: set on the requester's connection for a service request; Nothing otherwise. The time the client stops waiting for the response (created + serviceRequestTimeout or per-call override).
+    serviceRequestExpiresAt :: Maybe UTCTime
   }
   deriving (Eq, Show)
 
@@ -537,6 +539,7 @@ data InternalCommand
   | ICDeleteRcvQueue SMP.RecipientId
   | ICQSecure SMP.RecipientId SMP.SndPublicAuthKey
   | ICQDelete SMP.RecipientId
+  | ICReplyDel
 
 data InternalCommandTag
   = ICAck_
@@ -547,6 +550,7 @@ data InternalCommandTag
   | ICDeleteRcvQueue_
   | ICQSecure_
   | ICQDelete_
+  | ICReplyDel_
   deriving (Show)
 
 instance StrEncoding InternalCommand where
@@ -559,6 +563,7 @@ instance StrEncoding InternalCommand where
     ICDeleteRcvQueue rId -> strEncode (ICDeleteRcvQueue_, rId)
     ICQSecure rId senderKey -> strEncode (ICQSecure_, rId, senderKey)
     ICQDelete rId -> strEncode (ICQDelete_, rId)
+    ICReplyDel -> strEncode ICReplyDel_
   strP =
     strP >>= \case
       ICAck_ -> ICAck <$> _strP <*> _strP
@@ -569,6 +574,7 @@ instance StrEncoding InternalCommand where
       ICDeleteRcvQueue_ -> ICDeleteRcvQueue <$> _strP
       ICQSecure_ -> ICQSecure <$> _strP <*> _strP
       ICQDelete_ -> ICQDelete <$> _strP
+      ICReplyDel_ -> pure ICReplyDel
 
 instance StrEncoding InternalCommandTag where
   strEncode = \case
@@ -580,6 +586,7 @@ instance StrEncoding InternalCommandTag where
     ICDeleteRcvQueue_ -> "DELETE_RCV_QUEUE"
     ICQSecure_ -> "QSECURE"
     ICQDelete_ -> "QDELETE"
+    ICReplyDel_ -> "REPLY_DEL"
   strP =
     A.takeTill (== ' ') >>= \case
       "ACK" -> pure ICAck_
@@ -590,6 +597,7 @@ instance StrEncoding InternalCommandTag where
       "DELETE_RCV_QUEUE" -> pure ICDeleteRcvQueue_
       "QSECURE" -> pure ICQSecure_
       "QDELETE" -> pure ICQDelete_
+      "REPLY_DEL" -> pure ICReplyDel_
       _ -> fail "bad InternalCommandTag"
 
 agentCommandTag :: AgentCommand -> AgentCommandTag
@@ -607,6 +615,7 @@ internalCmdTag = \case
   ICDeleteRcvQueue {} -> ICDeleteRcvQueue_
   ICQSecure {} -> ICQSecure_
   ICQDelete _ -> ICQDelete_
+  ICReplyDel -> ICReplyDel_
 
 -- * Confirmation types
 
@@ -629,7 +638,9 @@ data AcceptedConfirmation = AcceptedConfirmation
 data NewInvitation = NewInvitation
   { contactConnId :: ConnId,
     connReq :: ContactRequest,
-    recipientConnInfo :: ConnInfo
+    recipientConnInfo :: ConnInfo,
+    -- service side: the received request is a service request (SREQ) not a contact request (REQ)
+    serviceRequest :: Bool
   }
 
 data Invitation = Invitation
@@ -638,7 +649,10 @@ data Invitation = Invitation
     connReq :: ContactRequest,
     recipientConnInfo :: ConnInfo,
     ownConnInfo :: Maybe ConnInfo,
-    accepted :: Bool
+    accepted :: Bool,
+    -- service side: the received request is a service request (SREQ) not a contact request (REQ)
+    serviceRequest :: Bool,
+    createdAt :: UTCTime
   }
 
 data ContactRequest