blade/simplex-chat
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplex-chat.git synced 2024-12-17 17:20:21 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into ab/async-subs

Browse Source
This commit is contained in:
Alexander Bondarenko
2024-05-30 19:08:51 +03:00
parent ddd9c6f16f 54088c6d5a
commit ed12f34330
166 changed files with 9704 additions and 1913 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Chat.hs
+109 -50
View File
@@ -94,7 +94,7 @@ import qualified Simplex.FileTransfer.Description as FD
import Simplex.FileTransfer.Protocol (FileParty (..), FilePartyI)
import qualified Simplex.FileTransfer.Transport as XFTP
import Simplex.Messaging.Agent as Agent
import Simplex.Messaging.Agent.Client (AgentStatsKey (..), SubInfo (..), agentClientStore, getAgentWorkersDetails, getAgentWorkersSummary, ipAddressProtected, temporaryAgentError, withLockMap)
import Simplex.Messaging.Agent.Client (AgentStatsKey (..), SubInfo (..), agentClientStore, getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, ipAddressProtected, temporaryAgentError, withLockMap)
import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore, defaultAgentConfig)
import Simplex.Messaging.Agent.Lock (withLock)
import Simplex.Messaging.Agent.Protocol
@@ -1472,14 +1472,14 @@ processChatCommand' vr = \case
ct@Contact {activeConn} <- withStore $ \db -> getContact db vr user contactId
case activeConn of
Just conn -> do
withStore' $ \db -> setConnectionAuthErrCounter db user conn 0
withStore' $ \db -> setAuthErrCounter db user conn 0
ok user
Nothing -> throwChatError $ CEContactNotActive ct
APIEnableGroupMember gId gMemberId -> withUser $ \user -> do
GroupMember {activeConn} <- withStore $ \db -> getGroupMember db vr user gId gMemberId
case activeConn of
Just conn -> do
withStore' $ \db -> setConnectionAuthErrCounter db user conn 0
withStore' $ \db -> setAuthErrCounter db user conn 0
ok user
_ -> throwChatError CEGroupMemberNotActive
SetShowMessages cName ntfOn -> updateChatSettings cName (\cs -> cs {enableNtfs = ntfOn})
@@ -1550,8 +1550,8 @@ processChatCommand' vr = \case
let chatV = agentToChatVersion agentV
dm <- encodeConnInfoPQ pqSup' chatV $ XInfo profileToSend
connId <- withAgent $ \a -> prepareConnectionToJoin a (aUserId user) True cReq pqSup'
conn <- withStore' $ \db -> createDirectConnection db user connId cReq ConnJoined (incognitoProfile $> profileToSend) subMode chatV pqSup'
void . withAgent $ \a -> joinConnection a (aUserId user) (Just connId) True cReq dm pqSup' subMode
conn@PendingContactConnection {pccConnId} <- withStore' $ \db -> createDirectConnection db user connId cReq ConnJoined (incognitoProfile $> profileToSend) subMode chatV pqSup'
joinPreparedAgentConnection user pccConnId connId cReq dm pqSup' subMode
pure $ CRSentConfirmation user conn
APIConnect userId incognito (Just (ACR SCMContact cReq)) -> withUserId userId $ \user -> connectViaContact user incognito cReq
APIConnect _ _ Nothing -> throwChatError CEInvalidConnReq
@@ -1806,12 +1806,20 @@ processChatCommand' vr = \case
dm <- encodeConnInfo $ XGrpAcpt membershipMemId
agentConnId <- withAgent $ \a -> prepareConnectionToJoin a (aUserId user) True connRequest PQSupportOff
let chatV = vr `peerConnChatVersion` peerChatVRange
withStore' $ \db -> do
createMemberConnection db userId fromMember agentConnId chatV peerChatVRange subMode
cId <- withStore' $ \db -> do
Connection {connId = cId} <- createMemberConnection db userId fromMember agentConnId chatV peerChatVRange subMode
updateGroupMemberStatus db userId fromMember GSMemAccepted
updateGroupMemberStatus db userId membership GSMemAccepted
void . withAgent $ \a -> joinConnection a (aUserId user) (Just agentConnId) True connRequest dm PQSupportOff subMode
updateCIGroupInvitationStatus user g CIGISAccepted `catchChatError` \_ -> pure ()
pure cId
void (withAgent $ \a -> joinConnection a (aUserId user) (Just agentConnId) True connRequest dm PQSupportOff subMode)
`catchChatError` \e -> do
withStore' $ \db -> do
deleteConnectionRecord db user cId
updateGroupMemberStatus db userId fromMember GSMemInvited
updateGroupMemberStatus db userId membership GSMemInvited
withAgent $ \a -> deleteConnectionAsync a False agentConnId
throwError e
updateCIGroupInvitationStatus user g CIGISAccepted `catchChatError` (toView . CRChatError (Just user))
pure $ CRUserAcceptedGroupSent user g {membership = membership {memberStatus = GSMemAccepted}} Nothing
Nothing -> throwChatError $ CEContactNotActive ct
APIMemberRole groupId memberId memRole -> withUser $ \user -> do
@@ -2239,6 +2247,7 @@ processChatCommand' vr = \case
SubInfo {server, subError = Just e} -> M.alter (Just . maybe [e] (e :)) server m
_ -> m
GetAgentSubsDetails -> lift $ CRAgentSubsDetails <$> withAgent' getAgentSubscriptions
GetAgentQueuesInfo -> lift $ CRAgentQueuesInfo <$> withAgent' getAgentQueuesInfo
-- CustomChatCommand is unsupported, it can be processed in preCmdHook
-- in a modified CLI app or core - the hook should return Either ChatResponse ChatCommand
CustomChatCommand _cmd -> withUser $ \user -> pure $ chatCmdError (Just user) "not supported"
@@ -2338,8 +2347,8 @@ processChatCommand' vr = \case
-- [incognito] generate profile to send
incognitoProfile <- if incognito then Just <$> liftIO generateRandomProfile else pure Nothing
subMode <- chatReadVar subscriptionMode
conn <- withStore' $ \db -> createConnReqConnection db userId connId cReqHash xContactId incognitoProfile groupLinkId subMode chatV pqSup
joinContact user connId cReq incognitoProfile xContactId inGroup pqSup chatV
conn@PendingContactConnection {pccConnId} <- withStore' $ \db -> createConnReqConnection db userId connId cReqHash xContactId incognitoProfile groupLinkId subMode chatV pqSup
joinContact user pccConnId connId cReq incognitoProfile xContactId inGroup pqSup chatV
pure $ CRSentInvitation user conn incognitoProfile
connectContactViaAddress :: User -> IncognitoEnabled -> Contact -> ConnectionRequestUri 'CMContact -> CM ChatResponse
connectContactViaAddress user incognito ct cReq =
@@ -2351,8 +2360,8 @@ processChatCommand' vr = \case
-- [incognito] generate profile to send
incognitoProfile <- if incognito then Just <$> liftIO generateRandomProfile else pure Nothing
subMode <- chatReadVar subscriptionMode
ct' <- withStore $ \db -> createAddressContactConnection db vr user ct connId cReqHash newXContactId incognitoProfile subMode chatV pqSup
joinContact user connId cReq incognitoProfile newXContactId False pqSup chatV
(pccConnId, ct') <- withStore $ \db -> createAddressContactConnection db vr user ct connId cReqHash newXContactId incognitoProfile subMode chatV pqSup
joinContact user pccConnId connId cReq incognitoProfile newXContactId False pqSup chatV
pure $ CRSentInvitationToContact user ct' incognitoProfile
prepareContact :: User -> ConnectionRequestUri 'CMContact -> PQSupport -> CM (ConnId, VersionChat)
prepareContact user cReq pqSup = do
@@ -2365,12 +2374,19 @@ processChatCommand' vr = \case
let chatV = agentToChatVersion agentV
connId <- withAgent $ \a -> prepareConnectionToJoin a (aUserId user) True cReq pqSup
pure (connId, chatV)
joinContact :: User -> ConnId -> ConnectionRequestUri 'CMContact -> Maybe Profile -> XContactId -> Bool -> PQSupport -> VersionChat -> CM ()
joinContact user connId cReq incognitoProfile xContactId inGroup pqSup chatV = do
joinContact :: User -> Int64 -> ConnId -> ConnectionRequestUri 'CMContact -> Maybe Profile -> XContactId -> Bool -> PQSupport -> VersionChat -> CM ()
joinContact user pccConnId connId cReq incognitoProfile xContactId inGroup pqSup chatV = do
let profileToSend = userProfileToSend user incognitoProfile Nothing inGroup
dm <- encodeConnInfoPQ pqSup chatV (XContact profileToSend $ Just xContactId)
subMode <- chatReadVar subscriptionMode
void . withAgent $ \a -> joinConnection a (aUserId user) (Just connId) True cReq dm pqSup subMode
joinPreparedAgentConnection user pccConnId connId cReq dm pqSup subMode
joinPreparedAgentConnection :: User -> Int64 -> ConnId -> ConnectionRequestUri m -> ByteString -> PQSupport -> SubscriptionMode -> CM ()
joinPreparedAgentConnection user pccConnId connId cReq connInfo pqSup subMode = do
void (withAgent $ \a -> joinConnection a (aUserId user) (Just connId) True cReq connInfo pqSup subMode)
`catchChatError` \e -> do
withStore' $ \db -> deleteConnectionRecord db user pccConnId
withAgent $ \a -> deleteConnectionAsync a False connId
throwError e
contactMember :: Contact -> [GroupMember] -> Maybe GroupMember
contactMember Contact {contactId} =
find $ \GroupMember {memberContactId = cId, memberStatus = s} ->
@@ -3934,14 +3950,19 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
withAckMessage' "new contact msg" agentConnId meta $
void $
saveDirectRcvMSG conn meta msgBody
SENT msgId _proxy ->
SENT msgId _proxy -> do
void $ continueSending connEntity conn
sentMsgDeliveryEvent conn msgId
OK ->
-- [async agent commands] continuation on receiving OK
when (corrId /= "") $ withCompletedCommand conn agentMsg $ \_cmdData -> pure ()
QCONT ->
void $ continueSending connEntity conn
MWARN _ err ->
processConnMWARN connEntity conn err
MERR _ err -> do
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
incAuthErrCounter connEntity conn err
processConnMERR connEntity conn err
MERRS _ err -> do
-- error cannot be AUTH error here
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
@@ -4072,6 +4093,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
let connectedIncognito = contactConnIncognito ct || incognitoMembership gInfo
when (memberCategory m == GCPreMember) $ probeMatchingContactsAndMembers ct connectedIncognito True
SENT msgId proxy -> do
void $ continueSending connEntity conn
sentMsgDeliveryEvent conn msgId
checkSndInlineFTComplete conn msgId
ci_ <- withStore $ \db -> do
@@ -4112,12 +4134,15 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
OK ->
-- [async agent commands] continuation on receiving OK
when (corrId /= "") $ withCompletedCommand conn agentMsg $ \_cmdData -> pure ()
MWARN msgId err ->
QCONT ->
void $ continueSending connEntity conn
MWARN msgId err -> do
updateDirectItemStatus ct conn msgId (CISSndWarning $ agentSndError err)
processConnMWARN connEntity conn err
MERR msgId err -> do
updateDirectItemStatus ct conn msgId (CISSndError $ agentSndError err)
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
incAuthErrCounter connEntity conn err
processConnMERR connEntity conn err
MERRS msgIds err -> do
-- error cannot be AUTH error here
updateDirectItemsStatus ct conn (L.toList msgIds) (CISSndError $ agentSndError err)
@@ -4466,9 +4491,11 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
withAckMessage' "group rcvd" agentConnId msgMeta $
groupMsgReceived gInfo m conn msgMeta msgRcpt
SENT msgId proxy -> do
continued <- continueSending connEntity conn
sentMsgDeliveryEvent conn msgId
checkSndInlineFTComplete conn msgId
updateGroupItemStatus gInfo m conn msgId (CISSndSent SSPComplete) (Just $ isJust proxy)
when continued $ sendPendingGroupMessages user m conn
SWITCH qd phase cStats -> do
toView $ CRGroupMemberSwitch user gInfo m (SwitchProgress qd phase cStats)
when (phase `elem` [SPStarted, SPCompleted]) $ case qd of
@@ -4504,13 +4531,17 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
OK ->
-- [async agent commands] continuation on receiving OK
when (corrId /= "") $ withCompletedCommand conn agentMsg $ \_cmdData -> pure ()
MWARN msgId err ->
QCONT -> do
continued <- continueSending connEntity conn
when continued $ sendPendingGroupMessages user m conn
MWARN msgId err -> do
withStore' $ \db -> updateGroupItemErrorStatus db msgId (groupMemberId' m) (CISSndWarning $ agentSndError err)
processConnMWARN connEntity conn err
MERR msgId err -> do
withStore' $ \db -> updateGroupItemErrorStatus db msgId (groupMemberId' m) (CISSndError $ agentSndError err)
-- group errors are silenced to reduce load on UI event log
-- toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
incAuthErrCounter connEntity conn err
processConnMERR connEntity conn err
MERRS msgIds err -> do
let newStatus = CISSndError $ agentSndError err
-- error cannot be AUTH error here
@@ -4640,7 +4671,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
when (corrId /= "") $ withCompletedCommand conn agentMsg $ \_cmdData -> pure ()
MERR _ err -> do
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
incAuthErrCounter connEntity conn err
processConnMERR connEntity conn err
ERR err -> do
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
when (corrId /= "") $ withCompletedCommand conn agentMsg $ \_cmdData -> pure ()
@@ -4692,7 +4723,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
_ -> pure ()
MERR _ err -> do
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
incAuthErrCounter connEntity conn err
processConnMERR connEntity conn err
ERR err -> do
toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
when (corrId /= "") $ withCompletedCommand conn agentMsg $ \_cmdData -> pure ()
@@ -4732,17 +4763,40 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
| memberRole <= GRObserver = messageError "member is not allowed to send messages"
| otherwise = a
incAuthErrCounter :: ConnectionEntity -> Connection -> AgentErrorType -> CM ()
incAuthErrCounter connEntity conn err = do
processConnMERR :: ConnectionEntity -> Connection -> AgentErrorType -> CM ()
processConnMERR connEntity conn err = do
case err of
SMP _ SMP.AUTH -> do
authErrCounter' <- withStore' $ \db -> incConnectionAuthErrCounter db user conn
authErrCounter' <- withStore' $ \db -> incAuthErrCounter db user conn
when (authErrCounter' >= authErrDisableCount) $ case connEntity of
RcvDirectMsgConnection ctConn (Just ct) -> do
toView $ CRContactDisabled user ct {activeConn = Just ctConn {authErrCounter = authErrCounter'}}
_ -> toView $ CRConnectionDisabled connEntity
SMP _ SMP.QUOTA ->
unless (connInactive conn) $ do
withStore' $ \db -> setQuotaErrCounter db user conn quotaErrSetOnMERR
toView $ CRConnectionInactive connEntity True
_ -> pure ()
processConnMWARN :: ConnectionEntity -> Connection -> AgentErrorType -> CM ()
processConnMWARN connEntity conn err = do
case err of
SMP _ SMP.QUOTA ->
unless (connInactive conn) $ do
quotaErrCounter' <- withStore' $ \db -> incQuotaErrCounter db user conn
when (quotaErrCounter' >= quotaErrInactiveCount) $
toView $ CRConnectionInactive connEntity True
_ -> pure ()
continueSending :: ConnectionEntity -> Connection -> CM Bool
continueSending connEntity conn =
if connInactive conn
then do
withStore' $ \db -> setQuotaErrCounter db user conn 0
toView $ CRConnectionInactive connEntity False
pure True
else pure False
-- TODO v5.7 / v6.0 - together with deprecating old group protocol establishing direct connections?
-- we could save command records only for agent APIs we process continuations for (INV)
withCompletedCommand :: forall e. AEntityI e => Connection -> ACommand 'Agent e -> (CommandData -> CM ()) -> CM ()
@@ -6533,16 +6587,20 @@ sendGroupMemberMessages user conn events groupId = do
let idsEvts = L.map (GroupId groupId,) events
(errs, msgs) <- lift $ partitionEithers . L.toList <$> createSndMessages idsEvts
unless (null errs) $ toView $ CRChatErrors (Just user) errs
forM_ (L.nonEmpty msgs) $ \msgs' -> do
-- TODO v5.7 based on version (?)
-- let shouldCompress = False
-- let batched = if shouldCompress then batchSndMessagesBinary msgs' else batchSndMessagesJSON msgs'
let batched = batchSndMessagesJSON msgs'
let (errs', msgBatches) = partitionEithers batched
-- shouldn't happen, as large messages would have caused createNewSndMessage to throw SELargeMsg
unless (null errs') $ toView $ CRChatErrors (Just user) errs'
forM_ msgBatches $ \batch ->
processSndMessageBatch conn batch `catchChatError` (toView . CRChatError (Just user))
forM_ (L.nonEmpty msgs) $ \msgs' ->
batchSendGroupMemberMessages user conn msgs'
batchSendGroupMemberMessages :: User -> Connection -> NonEmpty SndMessage -> CM ()
batchSendGroupMemberMessages user conn msgs = do
-- TODO v5.7 based on version (?)
-- let shouldCompress = False
-- let batched = if shouldCompress then batchSndMessagesBinary msgs' else batchSndMessagesJSON msgs'
let batched = batchSndMessagesJSON msgs
let (errs', msgBatches) = partitionEithers batched
-- shouldn't happen, as large messages would have caused createNewSndMessage to throw SELargeMsg
unless (null errs') $ toView $ CRChatErrors (Just user) errs'
forM_ msgBatches $ \batch ->
processSndMessageBatch conn batch `catchChatError` (toView . CRChatError (Just user))
processSndMessageBatch :: Connection -> MsgBatch -> CM ()
processSndMessageBatch conn@Connection {connId} (MsgBatch batchBody sndMsgs) = do
@@ -6700,6 +6758,7 @@ memberSendAction chatMsgEvent members m@GroupMember {invitedByGroupMemberId} = c
Nothing -> pendingOrForwarded
Just conn@Connection {connStatus}
| connDisabled conn || connStatus == ConnDeleted -> Nothing
| connInactive conn -> Just MSAPending
| connStatus == ConnSndReady || connStatus == ConnReady -> Just (MSASend conn)
| otherwise -> pendingOrForwarded
where
@@ -6730,21 +6789,20 @@ sendGroupMemberMessage user m@GroupMember {groupMemberId} chatMsgEvent groupId i
MSASend conn -> deliverMessage conn (toCMEventTag chatMsgEvent) msgBody msgId >> postDeliver
MSAPending -> withStore' $ \db -> createPendingGroupMessage db groupMemberId msgId introId_
-- TODO ensure order - pending messages interleave with user input messages
sendPendingGroupMessages :: User -> GroupMember -> Connection -> CM ()
sendPendingGroupMessages user GroupMember {groupMemberId, localDisplayName} conn = do
pendingMessages <- withStore' $ \db -> getPendingGroupMessages db groupMemberId
-- TODO ensure order - pending messages interleave with user input messages
forM_ pendingMessages $ \pgm ->
processPendingMessage pgm `catchChatError` (toView . CRChatError (Just user))
sendPendingGroupMessages user GroupMember {groupMemberId} conn = do
pgms <- withStore' $ \db -> getPendingGroupMessages db groupMemberId
forM_ (L.nonEmpty pgms) $ \pgms' -> do
let msgs = L.map (\(sndMsg, _, _) -> sndMsg) pgms'
batchSendGroupMemberMessages user conn msgs
lift . void . withStoreBatch' $ \db -> L.map (\SndMessage {msgId} -> deletePendingGroupMessage db groupMemberId msgId) msgs
lift . void . withStoreBatch' $ \db -> L.map (\(_, tag, introId_) -> updateIntro_ db tag introId_) pgms'
where
processPendingMessage PendingGroupMessage {msgId, cmEventTag = ACMEventTag _ tag, msgBody, introId_} = do
void $ deliverMessage conn tag msgBody msgId
withStore' $ \db -> deletePendingGroupMessage db groupMemberId msgId
case tag of
XGrpMemFwd_ -> case introId_ of
Just introId -> withStore' $ \db -> updateIntroStatus db introId GMIntroInvForwarded
_ -> throwChatError $ CEGroupMemberIntroNotFound localDisplayName
_ -> pure ()
updateIntro_ :: DB.Connection -> ACMEventTag -> Maybe Int64 -> IO ()
updateIntro_ db tag introId_ = case (tag, introId_) of
(ACMEventTag _ XGrpMemFwd_, Just introId) -> updateIntroStatus db introId GMIntroInvForwarded
_ -> pure ()
-- TODO [batch send] refactor direct message processing same as groups (e.g. checkIntegrity before processing)
saveDirectRcvMSG :: Connection -> MsgMeta -> MsgBody -> CM (Connection, RcvMessage)
@@ -7468,6 +7526,7 @@ chatCommandP =
"/get workers" $> GetAgentWorkers,
"/get workers details" $> GetAgentWorkersDetails,
"/get msgs" $> GetAgentMsgCounts,
"/get queues" $> GetAgentQueuesInfo,
"//" *> (CustomChatCommand <$> A.takeByteString)
]
where