core: use concurrency-safe agent async api

2026-07-18 05:36:12 +00:00 · 2026-06-01 11:55:00 +00:00
parent 60e75aa398
commit 10764a4ccc
3 changed files with 79 additions and 45 deletions
@@ -919,13 +919,15 @@ acceptContactRequestAsync
    let profileToSend = userProfileDirect user (fromIncognitoProfile <$> incognitoProfile) (Just ct) True
    vr <- chatVersionRange
    let chatV = vr `peerConnChatVersion` cReqChatVRange
-    (cmdId, acId) <- agentAcceptContactAsync user True cReqInvId (XInfo profileToSend) subMode cReqPQSup chatV
+    (cmdId, acId) <- prepareAgentAccept user True cReqInvId cReqPQSup
    currentTs <- liftIO getCurrentTime
-    withStore $ \db -> do
+    ct' <- withStore $ \db -> do
      forM_ xContactId $ \xcId -> liftIO $ setContactAcceptedXContactId db ct xcId
      Connection {connId} <- liftIO $ createAcceptedContactConn db user (Just uclId) contactId acId chatV cReqChatVRange cReqPQSup incognitoProfile subMode currentTs
      liftIO $ setCommandConnId db user cmdId connId
      getContact db vr user contactId
+    agentAcceptContactAsync user cmdId acId True cReqInvId (XInfo profileToSend) cReqPQSup chatV subMode
+    pure ct'

 acceptGroupJoinRequestAsync :: User -> Int64 -> GroupInfo -> InvitationId -> VersionRangeChat -> Profile -> Maybe XContactId -> Maybe MemberId -> Maybe SharedMsgId -> GroupAcceptance -> GroupMemberRole -> Maybe IncognitoProfile -> Maybe MemberKey -> CM GroupMember
 acceptGroupJoinRequestAsync
@@ -963,10 +965,11 @@ acceptGroupJoinRequestAsync
    subMode <- chatReadVar subscriptionMode
    vr <- chatVersionRange
    let chatV = vr `peerConnChatVersion` cReqChatVRange
-    connIds <- agentAcceptContactAsync user True cReqInvId msg subMode PQSupportOff chatV
-    withStore $ \db -> do
-      liftIO $ createJoiningMemberConnection db user uclId connIds chatV cReqChatVRange groupMemberId subMode
-      getGroupMemberById db vr user groupMemberId
+    (cmdId, acId) <- prepareAgentAccept user True cReqInvId PQSupportOff
+    withStore $ \db ->
+      liftIO $ createJoiningMemberConnection db user uclId (cmdId, acId) chatV cReqChatVRange groupMemberId subMode
+    agentAcceptContactAsync user cmdId acId True cReqInvId msg PQSupportOff chatV subMode
+    withStore $ \db -> getGroupMemberById db vr user groupMemberId

 acceptGroupJoinSendRejectAsync :: User -> Int64 -> GroupInfo -> InvitationId -> VersionRangeChat -> Profile -> Maybe XContactId -> GroupRejectionReason -> CM GroupMember
 acceptGroupJoinSendRejectAsync
@@ -993,10 +996,11 @@ acceptGroupJoinSendRejectAsync
    subMode <- chatReadVar subscriptionMode
    vr <- chatVersionRange
    let chatV = vr `peerConnChatVersion` cReqChatVRange
-    connIds <- agentAcceptContactAsync user False cReqInvId msg subMode PQSupportOff chatV
-    withStore $ \db -> do
-      liftIO $ createJoiningMemberConnection db user uclId connIds chatV cReqChatVRange groupMemberId subMode
-      getGroupMemberById db vr user groupMemberId
+    (cmdId, acId) <- prepareAgentAccept user False cReqInvId PQSupportOff
+    withStore $ \db ->
+      liftIO $ createJoiningMemberConnection db user uclId (cmdId, acId) chatV cReqChatVRange groupMemberId subMode
+    agentAcceptContactAsync user cmdId acId False cReqInvId msg PQSupportOff chatV subMode
+    withStore $ \db -> getGroupMemberById db vr user groupMemberId

 acceptBusinessJoinRequestAsync :: User -> Int64 -> GroupInfo -> GroupMember -> UserContactRequest -> CM (GroupInfo, GroupMember)
 acceptBusinessJoinRequestAsync
@@ -1025,10 +1029,11 @@ acceptBusinessJoinRequestAsync
              }
    subMode <- chatReadVar subscriptionMode
    let chatV = vr `peerConnChatVersion` cReqChatVRange
-    connIds <- agentAcceptContactAsync user True cReqInvId msg subMode PQSupportOff chatV
+    (cmdId, acId) <- prepareAgentAccept user True cReqInvId PQSupportOff
    withStore' $ \db -> do
      forM_ xContactId $ \xcId -> setBusinessChatAcceptedXContactId db gInfo xcId
-      createJoiningMemberConnection db user uclId connIds chatV cReqChatVRange groupMemberId subMode
+      createJoiningMemberConnection db user uclId (cmdId, acId) chatV cReqChatVRange groupMemberId subMode
+    agentAcceptContactAsync user cmdId acId True cReqInvId msg PQSupportOff chatV subMode
    let cd = CDGroupSnd gInfo Nothing
    -- TODO [short links] move to profileContactRequest?
    createInternalChatItem user cd (CISndGroupE2EEInfo $ e2eInfoGroup gInfo) Nothing
@@ -1050,12 +1055,14 @@ acceptRelayJoinRequestAsync
    subMode <- chatReadVar subscriptionMode
    vr <- chatVersionRange
    let chatV = vr `peerConnChatVersion` cReqChatVRange
-    connIds <- agentAcceptContactAsync user True cReqInvId msg subMode PQSupportOff chatV
-    withStore $ \db -> do
-      liftIO $ createJoiningMemberConnection db user uclId connIds chatV cReqChatVRange groupMemberId subMode
+    (cmdId, acId) <- prepareAgentAccept user True cReqInvId PQSupportOff
+    r <- withStore $ \db -> do
+      liftIO $ createJoiningMemberConnection db user uclId (cmdId, acId) chatV cReqChatVRange groupMemberId subMode
      gInfo' <- liftIO $ updateRelayOwnStatusFromTo db gInfo RSInvited RSAccepted
      ownerMember' <- getGroupMemberById db vr user groupMemberId
      pure (gInfo', ownerMember')
+    agentAcceptContactAsync user cmdId acId True cReqInvId msg PQSupportOff chatV subMode
+    pure r

 rejectRelayInvitationAsync
  :: User
@@ -1075,9 +1082,10 @@ rejectRelayInvitationAsync user uclId vr groupRelayInv invId reqChatVRange initi
  subMode <- chatReadVar subscriptionMode
  chatVR <- chatVersionRange
  let chatV = chatVR `peerConnChatVersion` reqChatVRange
-  connIds <- agentAcceptContactAsync user False invId msg subMode PQSupportOff chatV
+  (cmdId, acId) <- prepareAgentAccept user False invId PQSupportOff
  withStore' $ \db ->
-    createJoiningMemberConnection db user uclId connIds chatV reqChatVRange groupMemberId subMode
+    createJoiningMemberConnection db user uclId (cmdId, acId) chatV reqChatVRange groupMemberId subMode
+  agentAcceptContactAsync user cmdId acId False invId msg PQSupportOff chatV subMode

 businessGroupProfile :: Profile -> GroupPreferences -> GroupProfile
 businessGroupProfile Profile {displayName, fullName, shortDescr, image} groupPreferences =
@@ -2479,18 +2487,28 @@ msgContentHasLink mc ft_ = case msgContentTag mc of
  MCLink_ -> True
  _ -> maybe False hasLinks ft_

-createAgentConnectionAsync :: ConnectionModeI c => User -> CommandFunction -> Bool -> SConnectionMode c -> SubscriptionMode -> CM (CommandId, ConnId)
-createAgentConnectionAsync user cmdFunction enableNtfs cMode subMode = do
+prepareAgentCreation :: ConnectionModeI c => User -> CommandFunction -> Bool -> SConnectionMode c -> CM (CommandId, ConnId)
+prepareAgentCreation user cmdFunction enableNtfs cMode = do
  cmdId <- withStore' $ \db -> createCommand db user Nothing cmdFunction
-  connId <- withAgent $ \a -> createConnectionAsync a (aUserId user) (aCorrId cmdId) enableNtfs cMode IKPQOff subMode
+  connId <- withAgent $ \a -> prepareConnectionToCreate a (aUserId user) enableNtfs cMode PQSupportOff
  pure (cmdId, connId)

-joinAgentConnectionAsync :: User -> Maybe Connection -> Bool -> ConnectionRequestUri c -> ConnInfo -> SubscriptionMode -> CM (CommandId, ConnId)
-joinAgentConnectionAsync user conn_ enableNtfs cReqUri cInfo subMode = do
+createAgentConnectionAsync :: ConnectionModeI c => User -> CommandId -> ConnId -> Bool -> SConnectionMode c -> SubscriptionMode -> CM ()
+createAgentConnectionAsync user cmdId connId enableNtfs cMode subMode =
+  withAgent $ \a -> createConnectionAsync a (aCorrId cmdId) connId enableNtfs cMode IKPQOff subMode
+
+prepareAgentJoin :: User -> Maybe Connection -> Bool -> ConnectionRequestUri c -> CM (CommandId, ConnId)
+prepareAgentJoin user conn_ enableNtfs cReqUri = do
  cmdId <- withStore' $ \db -> createCommand db user (dbConnId <$> conn_) CFJoinConn
-  connId <- withAgent $ \a -> joinConnectionAsync a (aUserId user) (aCorrId cmdId) (aConnId <$> conn_) enableNtfs cReqUri cInfo PQSupportOff subMode
+  connId <- case conn_ of
+    Just conn -> pure $ aConnId conn
+    Nothing -> withAgent $ \a -> prepareConnectionToJoin a (aUserId user) enableNtfs cReqUri PQSupportOff
  pure (cmdId, connId)

+joinAgentConnectionAsync :: ConnectionModeI c => User -> CommandId -> Bool -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SubscriptionMode -> CM ()
+joinAgentConnectionAsync user cmdId updateConn connId enableNtfs cReqUri cInfo subMode =
+  withAgent $ \a -> joinConnectionAsync a (aCorrId cmdId) updateConn connId enableNtfs cReqUri cInfo PQSupportOff subMode
+
 allowAgentConnectionAsync :: MsgEncodingI e => User -> Connection -> ConfirmationId -> ChatMsgEvent e -> CM ()
 allowAgentConnectionAsync user conn@Connection {connId, pqSupport, connChatVersion} confId msg = do
  cmdId <- withStore' $ \db -> createCommand db user (Just connId) CFAllowConn
@@ -2498,13 +2516,17 @@ allowAgentConnectionAsync user conn@Connection {connId, pqSupport, connChatVersi
  withAgent $ \a -> allowConnectionAsync a (aCorrId cmdId) (aConnId conn) confId dm
  withStore' $ \db -> updateConnectionStatus db conn ConnAccepted

-agentAcceptContactAsync :: MsgEncodingI e => User -> Bool -> InvitationId -> ChatMsgEvent e -> SubscriptionMode -> PQSupport -> VersionChat -> CM (CommandId, ConnId)
-agentAcceptContactAsync user enableNtfs invId msg subMode pqSup chatV = do
+prepareAgentAccept :: User -> Bool -> InvitationId -> PQSupport -> CM (CommandId, ConnId)
+prepareAgentAccept user enableNtfs invId pqSup = do
  cmdId <- withStore' $ \db -> createCommand db user Nothing CFAcceptContact
-  dm <- encodeConnInfoPQ pqSup chatV msg
-  connId <- withAgent $ \a -> acceptContactAsync a (aUserId user) (aCorrId cmdId) enableNtfs invId dm pqSup subMode
+  connId <- withAgent $ \a -> prepareConnectionToAccept a (aUserId user) enableNtfs invId pqSup
  pure (cmdId, connId)

+agentAcceptContactAsync :: MsgEncodingI e => User -> CommandId -> ConnId -> Bool -> InvitationId -> ChatMsgEvent e -> PQSupport -> VersionChat -> SubscriptionMode -> CM ()
+agentAcceptContactAsync user cmdId connId enableNtfs invId msg pqSup chatV subMode = do
+  dm <- encodeConnInfoPQ pqSup chatV msg
+  withAgent $ \a -> acceptContactAsync a (aCorrId cmdId) connId enableNtfs invId dm pqSup subMode
+
 deleteAgentConnectionAsync :: ConnId -> CM ()
 deleteAgentConnectionAsync acId = deleteAgentConnectionAsync' acId False
 {-# INLINE deleteAgentConnectionAsync #-}
@@ -627,9 +627,10 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
            forM_ gli_ $ \GroupLinkInfo {groupId, memberRole = gLinkMemRole} -> do
              groupInfo <- withStore $ \db -> getGroupInfo db vr user groupId
              subMode <- chatReadVar subscriptionMode
-              groupConnIds <- createAgentConnectionAsync user CFCreateConnGrpInv True SCMInvitation subMode
+              groupConnIds <- prepareAgentCreation user CFCreateConnGrpInv True SCMInvitation
              gVar <- asks random
              withStore $ \db -> createNewContactMemberAsync db gVar user groupInfo ct' gLinkMemRole groupConnIds connChatVersion peerChatVRange subMode
+              uncurry (createAgentConnectionAsync user) groupConnIds True SCMInvitation subMode
        -- TODO REMOVE LEGACY ^^^
        SENT msgId proxy -> do
          void $ continueSending connEntity conn
@@ -1199,7 +1200,8 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
                        Nothing -> throwChatError $ CEInternalError "no group keys for channel membership"
                      dm <- encodeConnInfo $ XMember profileToSend membershipMemId (MemberKey memberPubKey)
                      subMode <- chatReadVar subscriptionMode
-                      void $ joinAgentConnectionAsync user (Just conn) True cReq dm subMode
+                      (cmdId, connId) <- prepareAgentJoin user (Just conn) True cReq
+                      joinAgentConnectionAsync user cmdId True connId True cReq dm subMode
            CFGetRelayDataAccept -> do
              let GroupMember {memberId = MemberId expectedMemberId} = m
              if linkEntityId == Just expectedMemberId
@@ -1583,10 +1585,11 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
              subMode <- chatReadVar subscriptionMode
              chatVR <- chatVersionRange
              let chatV = chatVR `peerConnChatVersion` chatVRange
-              (cmdId, acId) <- agentAcceptContactAsync user True invId msg subMode PQSupportOff chatV
+              (cmdId, acId) <- prepareAgentAccept user True invId PQSupportOff
              withStore $ \db -> do
                Connection {connId = testCId} <- createRelayTestConnection db vr user acId ConnAccepted chatV subMode
                liftIO $ setCommandConnId db user cmdId testCId
+              agentAcceptContactAsync user cmdId acId True invId msg PQSupportOff chatV subMode
        -- TODO [relays] owner, relays: TBC how to communicate member rejection rules from owner to relays
        -- TODO [relays] relay: TBC communicate rejection when memberId already exists (currently checked in createJoiningMember)
        memberJoinRequestViaRelay :: InvitationId -> VersionRangeChat -> Profile -> MemberId -> MemberKey -> CM ()
@@ -2507,12 +2510,13 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
              then do
                subMode <- chatReadVar subscriptionMode
                dm <- encodeConnInfo $ XGrpAcpt membershipMemId
-                connIds <- joinAgentConnectionAsync user Nothing True connRequest dm subMode
+                connIds@(cmdId, acId) <- prepareAgentJoin user Nothing True connRequest
                withStore' $ \db -> do
                  setViaGroupLinkUri db groupId connId
                  createMemberConnectionAsync db user hostId connIds connChatVersion peerChatVRange subMode
                  updateGroupMemberStatusById db userId hostId GSMemAccepted
                  updateGroupMemberStatus db userId membership GSMemAccepted
+                joinAgentConnectionAsync user cmdId False acId True connRequest dm subMode
                toView $ CEvtUserAcceptedGroupSent user gInfo {membership = membership {memberStatus = GSMemAccepted}} (Just ct)
              else do
                let content = CIRcvGroupInvitation (CIGroupInvitation {groupId, groupMemberId, localDisplayName, groupProfile, status = CIGISPending}) memRole
@@ -3047,15 +3051,16 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
                      | maxVersion mcvr >= groupDirectInvVersion -> do
                          subMode <- chatReadVar subscriptionMode
                          -- [async agent commands] commands should be asynchronous, continuation is to send XGrpMemInv - have to remember one has completed and process on second
-                          groupConnIds <- createConn subMode
+                          groupConnIds <- prepareConn
                          let chatV = maybe (minVersion vr) (\peerVR -> vr `peerConnChatVersion` fromChatVRange peerVR) memChatVRange
                          void $ withStore $ \db -> do
                            reMember <- createIntroReMember db user gInfo memInfo memRestrictions
                            createIntroReMemberConn db user m reMember chatV memInfo groupConnIds subMode
+                          uncurry (createAgentConnectionAsync user) groupConnIds (chatHasNtfs chatSettings) SCMInvitation subMode
                      | otherwise -> messageError "x.grp.mem.intro: member chat version range incompatible"
        _ -> messageError "x.grp.mem.intro can be only sent by host member"
      where
-        createConn subMode = createAgentConnectionAsync user CFCreateConnGrpMemInv (chatHasNtfs chatSettings) SCMInvitation subMode
+        prepareConn = prepareAgentCreation user CFCreateConnGrpMemInv (chatHasNtfs chatSettings) SCMInvitation

    sendXGrpMemInv :: Int64 -> Maybe ConnReqInvitation -> XGrpMemIntroCont -> CM ()
    sendXGrpMemInv hostConnId directConnReq XGrpMemIntroCont {groupId, groupMemberId, memberId, groupConnReq} = do
@@ -3099,12 +3104,16 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
          allowSimplexLinks = groupFeatureUserAllowed SGFSimplexLinks gInfo
      dm <- encodeConnInfo $ XGrpMemInfo membershipMemId membershipProfile
      -- [async agent commands] no continuation needed, but commands should be asynchronous for stability
-      groupConnIds <- joinAgentConnectionAsync user Nothing (chatHasNtfs chatSettings) groupConnReq dm subMode
-      directConnIds <- forM directConnReq $ \dcr -> joinAgentConnectionAsync user Nothing True dcr dm subMode
+      let enableNtfsGrp = chatHasNtfs chatSettings
+      groupConnIds@(gCmdId, gAcId) <- prepareAgentJoin user Nothing enableNtfsGrp groupConnReq
+      directConnIds <- forM directConnReq $ \dcr -> prepareAgentJoin user Nothing True dcr
      let customUserProfileId = localProfileId <$> incognitoMembershipProfile gInfo
          mcvr = maybe chatInitialVRange fromChatVRange memChatVRange
          chatV = vr `peerConnChatVersion` mcvr
      withStore' $ \db -> createIntroToMemberContact db user m toMember chatV mcvr groupConnIds directConnIds customUserProfileId subMode
+      joinAgentConnectionAsync user gCmdId False gAcId enableNtfsGrp groupConnReq dm subMode
+      forM_ ((,) <$> directConnIds <*> directConnReq) $ \((dCmdId, dAcId), dcr) ->
+        joinAgentConnectionAsync user dCmdId False dAcId True dcr dm subMode

    xGrpMemRole :: GroupInfo -> GroupMember -> MemberId -> GroupMemberRole -> RcvMessage -> UTCTime -> CM (Maybe DeliveryJobScope)
    xGrpMemRole gInfo@GroupInfo {membership} m@GroupMember {memberRole = senderRole} memId memRole msg@RcvMessage {msgSigned} brokerTs
@@ -3359,11 +3368,12 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
            }
        joinExistingContact subMode mCt@Contact {contactId = mContactId}
          | autoAcceptMemberContacts user = do
-              (cmdId, acId) <- joinConn subMode
+              (cmdId, acId) <- prepareJoinMemberContact
              mCt' <- withStore $ \db -> do
                updateMemberContactInvited db user mCt groupDirectInv
                void $ liftIO $ createMemberContactConn db user acId (Just cmdId) g mConn ConnJoined mContactId subMode
                getContact db vr user mContactId
+              joinMemberContactAsync cmdId acId subMode
              securityCodeChanged mCt'
              createItems mCt' m
          | otherwise = do
@@ -3377,13 +3387,14 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
              createItems mCt' m
        createNewContact subMode
          | autoAcceptMemberContacts user = do
-              (cmdId, acId) <- joinConn subMode
+              (cmdId, acId) <- prepareJoinMemberContact
              -- [incognito] reuse membership incognito profile
              (mCt, m') <- withStore $ \db -> do
                (mContactId, m') <- liftIO $ createMemberContactInvited db user g m groupDirectInv
                void $ liftIO $ createMemberContactConn db user acId (Just cmdId) g mConn ConnJoined mContactId subMode
                mCt <- getContact db vr user mContactId
                pure (mCt, m')
+              joinMemberContactAsync cmdId acId subMode
              createInternalChatItem user (CDDirectSnd mCt) CIChatBanner (Just epochStart)
              createItems mCt m'
          | otherwise = do
@@ -3396,12 +3407,13 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
              createInternalChatItem user (CDDirectSnd mCt) CIChatBanner (Just epochStart)
              createInternalChatItem user (CDDirectRcv mCt) (CIRcvDirectEvent $ RDEGroupInvLinkReceived gp) Nothing
              createItems mCt m'
-        joinConn subMode = do
+        prepareJoinMemberContact = prepareAgentJoin user Nothing True connReq
+        joinMemberContactAsync cmdId acId subMode = do
          -- [incognito] send membership incognito profile
          let p = userProfileDirect user (fromLocalProfile <$> incognitoMembershipProfile g) Nothing True
          -- TODO PQ should negotitate contact connection with PQSupportOn? (use encodeConnInfoPQ)
          dm <- encodeConnInfo $ XInfo p
-          joinAgentConnectionAsync user Nothing True connReq dm subMode
+          joinAgentConnectionAsync user cmdId False acId True connReq dm subMode
        createItems mCt' m' = do
          (g', m'', scopeInfo) <- mkGroupChatScope g m'
          createInternalChatItem user (CDGroupRcv g' scopeInfo m'') (CIRcvGroupEvent RGEMemberCreatedContact) Nothing