From aeb28400e9e814ead4e7be3da482f80e9f71c72a Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Mon, 29 Apr 2024 19:32:53 +0400 Subject: [PATCH 1/2] core: take entity lock before processing file (#4105) --- src/Simplex/Chat.hs | 18 +++++++++--- src/Simplex/Chat/Store/Files.hs | 44 ++++++++++++++++++++++-------- src/Simplex/Chat/Store/Messages.hs | 1 - 3 files changed, 47 insertions(+), 16 deletions(-) diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index 7b82d997cd..f8285770e0 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -3577,14 +3577,19 @@ processAgentMessageNoConn = \case processAgentMsgSndFile :: ACorrId -> SndFileId -> ACommand 'Agent 'AESndFile -> CM () processAgentMsgSndFile _corrId aFileId msg = do - fileId <- withStore (`getXFTPSndFileDBId` AgentSndFileId aFileId) - withFileLock "processAgentMsgSndFile" fileId $ + (cRef_, fileId) <- withStore (`getXFTPSndFileDBIds` AgentSndFileId aFileId) + withEntityLock_ cRef_ $ withFileLock "processAgentMsgSndFile" fileId $ withStore' (`getUserByASndFileId` AgentSndFileId aFileId) >>= \case Just user -> process user fileId `catchChatError` (toView . CRChatError (Just user)) _ -> do lift $ withAgent' (`xftpDeleteSndFileInternal` aFileId) throwChatError $ CENoSndFileUser $ AgentSndFileId aFileId where + withEntityLock_ :: Maybe ChatRef -> CM a -> CM a + withEntityLock_ cRef_ = case cRef_ of + Just (ChatRef CTDirect contactId) -> withContactLock "processAgentMsgSndFile" contactId + Just (ChatRef CTGroup groupId) -> withGroupLock "processAgentMsgSndFile" groupId + _ -> id process :: User -> FileTransferId -> CM () process user fileId = do (ft@FileTransferMeta {xftpRedirectFor, cancelled}, sfts) <- withStore $ \db -> getSndFileTransfer db user fileId @@ -3699,14 +3704,19 @@ splitFileDescr rfdText = do processAgentMsgRcvFile :: ACorrId -> RcvFileId -> ACommand 'Agent 'AERcvFile -> CM () processAgentMsgRcvFile _corrId aFileId msg = do - fileId <- withStore (`getXFTPRcvFileDBId` AgentRcvFileId aFileId) - withFileLock "processAgentMsgRcvFile" fileId $ + (cRef_, fileId) <- withStore (`getXFTPRcvFileDBIds` AgentRcvFileId aFileId) + withEntityLock_ cRef_ $ withFileLock "processAgentMsgRcvFile" fileId $ withStore' (`getUserByARcvFileId` AgentRcvFileId aFileId) >>= \case Just user -> process user fileId `catchChatError` (toView . CRChatError (Just user)) _ -> do lift $ withAgent' (`xftpDeleteRcvFile` aFileId) throwChatError $ CENoRcvFileUser $ AgentRcvFileId aFileId where + withEntityLock_ :: Maybe ChatRef -> CM a -> CM a + withEntityLock_ cRef_ = case cRef_ of + Just (ChatRef CTDirect contactId) -> withContactLock "processAgentMsgRcvFile" contactId + Just (ChatRef CTGroup groupId) -> withGroupLock "processAgentMsgRcvFile" groupId + _ -> id process :: User -> FileTransferId -> CM () process user fileId = do ft <- withStore $ \db -> getRcvFileTransfer db user fileId diff --git a/src/Simplex/Chat/Store/Files.hs b/src/Simplex/Chat/Store/Files.hs index 81a5897c89..528290aa59 100644 --- a/src/Simplex/Chat/Store/Files.hs +++ b/src/Simplex/Chat/Store/Files.hs @@ -29,8 +29,8 @@ module Simplex.Chat.Store.Files createExtraSndFTDescrs, updateSndFTDeliveryXFTP, setSndFTAgentDeleted, - getXFTPSndFileDBId, - getXFTPRcvFileDBId, + getXFTPSndFileDBIds, + getXFTPRcvFileDBIds, updateFileCancelled, updateCIFileStatus, getSharedMsgIdByFileId, @@ -109,7 +109,7 @@ import Simplex.Chat.Store.Shared import Simplex.Chat.Types import Simplex.Chat.Util (week) import Simplex.Messaging.Agent.Protocol (AgentMsgId, ConnId, UserId) -import Simplex.Messaging.Agent.Store.SQLite (firstRow, maybeFirstRow) +import Simplex.Messaging.Agent.Store.SQLite (firstRow, firstRow', maybeFirstRow) import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs (..)) @@ -336,15 +336,37 @@ setSndFTAgentDeleted db User {userId} fileId = do "UPDATE files SET agent_snd_file_deleted = 1, updated_at = ? WHERE user_id = ? AND file_id = ?" (currentTs, userId, fileId) -getXFTPSndFileDBId :: DB.Connection -> AgentSndFileId -> ExceptT StoreError IO FileTransferId -getXFTPSndFileDBId db aSndFileId = - ExceptT . firstRow fromOnly (SESndFileNotFoundXFTP aSndFileId) $ - DB.query db "SELECT file_id FROM files WHERE agent_snd_file_id = ?" (Only aSndFileId) +getXFTPSndFileDBIds :: DB.Connection -> AgentSndFileId -> ExceptT StoreError IO (Maybe ChatRef, FileTransferId) +getXFTPSndFileDBIds db aSndFileId = + ExceptT . firstRow' toFileRef (SESndFileNotFoundXFTP aSndFileId) $ + DB.query + db + [sql| + SELECT file_id, contact_id, group_id, note_folder_id + FROM files + WHERE agent_snd_file_id = ? + |] + (Only aSndFileId) -getXFTPRcvFileDBId :: DB.Connection -> AgentRcvFileId -> ExceptT StoreError IO FileTransferId -getXFTPRcvFileDBId db aRcvFileId = - ExceptT . firstRow fromOnly (SERcvFileNotFoundXFTP aRcvFileId) $ - DB.query db "SELECT file_id FROM rcv_files WHERE agent_rcv_file_id = ?" (Only aRcvFileId) +getXFTPRcvFileDBIds :: DB.Connection -> AgentRcvFileId -> ExceptT StoreError IO (Maybe ChatRef, FileTransferId) +getXFTPRcvFileDBIds db aRcvFileId = + ExceptT . firstRow' toFileRef (SERcvFileNotFoundXFTP aRcvFileId) $ + DB.query + db + [sql| + SELECT rf.file_id, f.contact_id, f.group_id, f.note_folder_id + FROM rcv_files rf + JOIN files f ON f.file_id = rf.file_id + WHERE rf.agent_rcv_file_id = ? + |] + (Only aRcvFileId) + +toFileRef :: (FileTransferId, Maybe Int64, Maybe Int64, Maybe Int64) -> Either StoreError (Maybe ChatRef, FileTransferId) +toFileRef = \case + (fileId, Just contactId, Nothing, Nothing) -> Right (Just $ ChatRef CTDirect contactId, fileId) + (fileId, Nothing, Just groupId, Nothing) -> Right (Just $ ChatRef CTGroup groupId, fileId) + (fileId, Nothing, Nothing, Just folderId) -> Right (Just $ ChatRef CTLocal folderId, fileId) + (fileId, _, _, _) -> Right (Nothing, fileId) updateFileCancelled :: MsgDirectionI d => DB.Connection -> User -> Int64 -> CIFileStatus d -> IO () updateFileCancelled db User {userId} fileId ciFileStatus = do diff --git a/src/Simplex/Chat/Store/Messages.hs b/src/Simplex/Chat/Store/Messages.hs index 0d6592a311..8163bd336c 100644 --- a/src/Simplex/Chat/Store/Messages.hs +++ b/src/Simplex/Chat/Store/Messages.hs @@ -146,7 +146,6 @@ import Simplex.Messaging.Agent.Store.SQLite (firstRow, firstRow', maybeFirstRow) import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs (..)) -import Simplex.Messaging.Crypto.Ratchet (PQSupport) import Simplex.Messaging.Util (eitherToMaybe) import UnliftIO.STM From 37e03a838cbdeb47d7f361cf951886e5865a51bf Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Mon, 29 Apr 2024 19:49:04 +0400 Subject: [PATCH 2/2] core: forward group message before ack (fixes forwarding message that deleted connection causing error in ackMsg) (#4108) --- src/Simplex/Chat.hs | 45 ++++++++++++++++++++------------------- tests/ChatTests/Groups.hs | 13 +++++++++++ 2 files changed, 36 insertions(+), 22 deletions(-) diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index f8285770e0..e13dbfcbff 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -2932,7 +2932,7 @@ callTimed ct aciContent = case aciContentCallStatus aciContent of Just callStatus | callComplete callStatus -> do - contactCITimed ct + contactCITimed ct _ -> pure Nothing where aciContentCallStatus :: ACIContent -> Maybe CICallStatus @@ -4269,12 +4269,8 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = Right (ACMsg _ chatMsg) -> processEvent chatMsg `catchChatError` \e -> toView $ CRChatError (Just user) e Left e -> toView $ CRChatError (Just user) (ChatError . CEException $ "error parsing chat message: " <> e) + forwardMsg_ `catchChatError` \_ -> pure () checkSendRcpt $ rights aChatMsgs - -- currently only a single message is forwarded - let GroupMember {memberRole = membershipMemRole} = membership - when (membershipMemRole >= GRAdmin && not (blockedByAdmin m)) $ case aChatMsgs of - [Right (ACMsg _ chatMsg)] -> forwardMsg_ chatMsg - _ -> pure () where aChatMsgs = parseChatMessages msgBody brokerTs = metaBrokerTs msgMeta @@ -4322,22 +4318,27 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage = where aChatMsgHasReceipt (ACMsg _ ChatMessage {chatMsgEvent}) = hasDeliveryReceipt (toCMEventTag chatMsgEvent) - forwardMsg_ :: MsgEncodingI e => ChatMessage e -> CM () - forwardMsg_ chatMsg = - forM_ (forwardedGroupMsg chatMsg) $ \chatMsg' -> do - ChatConfig {highlyAvailable} <- asks config - -- members introduced to this invited member - introducedMembers <- - if memberCategory m == GCInviteeMember - then withStore' $ \db -> getForwardIntroducedMembers db vr user m highlyAvailable - else pure [] - -- invited members to which this member was introduced - invitedMembers <- withStore' $ \db -> getForwardInvitedMembers db vr user m highlyAvailable - let GroupMember {memberId} = m - ms = forwardedToGroupMembers (introducedMembers <> invitedMembers) chatMsg' - msg = XGrpMsgForward memberId chatMsg' brokerTs - unless (null ms) . void $ - sendGroupMessage' user gInfo ms msg + forwardMsg_ :: CM () + forwardMsg_ = do + let GroupMember {memberRole = membershipMemRole} = membership + when (membershipMemRole >= GRAdmin && not (blockedByAdmin m)) $ case aChatMsgs of + -- currently only a single message is forwarded + [Right (ACMsg _ chatMsg)] -> + forM_ (forwardedGroupMsg chatMsg) $ \chatMsg' -> do + ChatConfig {highlyAvailable} <- asks config + -- members introduced to this invited member + introducedMembers <- + if memberCategory m == GCInviteeMember + then withStore' $ \db -> getForwardIntroducedMembers db vr user m highlyAvailable + else pure [] + -- invited members to which this member was introduced + invitedMembers <- withStore' $ \db -> getForwardInvitedMembers db vr user m highlyAvailable + let GroupMember {memberId} = m + ms = forwardedToGroupMembers (introducedMembers <> invitedMembers) chatMsg' + msg = XGrpMsgForward memberId chatMsg' brokerTs + unless (null ms) . void $ + sendGroupMessage' user gInfo ms msg + _ -> pure () RCVD msgMeta msgRcpt -> withAckMessage' agentConnId msgMeta $ groupMsgReceived gInfo m conn msgMeta msgRcpt diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index ff577dc737..4d9d94e24d 100644 --- a/tests/ChatTests/Groups.hs +++ b/tests/ChatTests/Groups.hs @@ -120,6 +120,7 @@ chatGroupTests = do it "forward file (x.msg.file.descr)" testGroupMsgForwardFile it "forward role change (x.grp.mem.role)" testGroupMsgForwardChangeRole it "forward new member announcement (x.grp.mem.new)" testGroupMsgForwardNewMember + it "forward member leaving (x.grp.leave)" testGroupMsgForwardLeave describe "group history" $ do it "text messages" testGroupHistory it "history is sent when joining via group link" testGroupHistoryGroupLink @@ -4437,6 +4438,18 @@ testGroupMsgForwardNewMember = "dan (Daniel): member" ] +testGroupMsgForwardLeave :: HasCallStack => FilePath -> IO () +testGroupMsgForwardLeave = + testChat3 aliceProfile bobProfile cathProfile $ + \alice bob cath -> do + setupGroupForwarding3 "team" alice bob cath + + bob ##> "/leave #team" + bob <## "#team: you left the group" + bob <## "use /d #team to delete the group" + alice <## "#team: bob left the group" + cath <## "#team: bob left the group" + testGroupHistory :: HasCallStack => FilePath -> IO () testGroupHistory = testChat3 aliceProfile bobProfile cathProfile $