From c4102db3d8dffb6f78d12b52966743164b9d4a1d Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Tue, 9 Jun 2026 12:04:42 +0400 Subject: [PATCH] wip --- src/Simplex/Chat/Library/Subscriber.hs | 157 ++++++++++--------------- src/Simplex/Chat/Store/Files.hs | 67 +++-------- 2 files changed, 78 insertions(+), 146 deletions(-) diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index e498bf266b..090f2423e0 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -1311,39 +1311,56 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = r n'' = Just (ci, CIRcvDecryptionError mde n'') mdeUpdatedCI _ _ = Nothing - receiveFileChunk :: RcvFileTransfer -> Maybe Connection -> MsgMeta -> FileChunk -> CM () - receiveFileChunk ft@RcvFileTransfer {fileId, chunkSize} conn_ meta@MsgMeta {recipient = (msgId, _), integrity} = \case - FileChunkCancel -> - unless (rcvFileCompleteOrCancelled ft) $ do - cancelRcvFileTransfer user ft - ci <- withStore $ \db -> getChatItemByFileId db cxt user fileId - toView $ CEvtRcvFileSndCancelled user ci ft - FileChunk {chunkNo, chunkBytes = chunk} -> do - case integrity of - MsgOk -> pure () - MsgError MsgDuplicate -> pure () -- TODO remove once agent removes duplicates - MsgError e -> - badRcvFileChunk ft $ "invalid file chunk number " <> show chunkNo <> ": " <> show e - withStore' (\db -> createRcvFileChunk db ft chunkNo msgId) >>= \case - RcvChunkOk -> - if B.length chunk /= fromInteger chunkSize - then badRcvFileChunk ft "incorrect chunk size" - else withAckMessage' "file msg" agentConnId meta $ appendFileChunk ft chunkNo chunk False - RcvChunkFinal -> - if B.length chunk > fromInteger chunkSize - then badRcvFileChunk ft "incorrect chunk size" - else do - appendFileChunk ft chunkNo chunk True - ci <- withStore $ \db -> do - liftIO $ do - updateRcvFileStatus db fileId FSComplete - updateCIFileStatus db user fileId CIFSRcvComplete - deleteRcvFileChunks db ft - getChatItemByFileId db cxt user fileId - toView $ CEvtRcvFileComplete user ci - mapM_ (deleteAgentConnectionAsync . aConnId) conn_ - RcvChunkDuplicate -> withAckMessage' "file msg" agentConnId meta $ pure () - RcvChunkError -> badRcvFileChunk ft $ "incorrect chunk number " <> show chunkNo + -- Receives an inline file chunk for a normal file (chat-item-linked) or the group roster blob + -- (file_type = roster, no chat item); fileType selects start, completion, and cancel. The agent + -- message is acked once by the caller's outer per-MSG withAckMessage, not here. + receiveFileChunk :: Maybe GroupInfo -> RcvFileTransfer -> Maybe Connection -> MsgMeta -> FileChunk -> CM () + receiveFileChunk gInfo_ ft@RcvFileTransfer {fileId, fileType, fileStatus, chunkSize} conn_ MsgMeta {recipient = (msgId, _), integrity} = \case + FileChunkCancel -> case fileType of + FTRoster -> forM_ gInfo_ $ cleanupGroupRosterFile user + FTNormal -> + unless (rcvFileCompleteOrCancelled ft) $ do + cancelRcvFileTransfer user ft + ci <- withStore $ \db -> getChatItemByFileId db cxt user fileId + toView $ CEvtRcvFileSndCancelled user ci ft + FileChunk {chunkNo, chunkBytes = chunk} + -- a normal inline file not accepted for inline receipt: chunk 1 prohibited, rest ignored + | fileType == FTNormal, RFSNew <- fileStatus -> + when (chunkNo == 1) $ throwChatError $ CEInlineFileProhibited fileId + | otherwise -> do + -- normal: transition to receiving on chunk 1 before the integrity check (prior order) + when (fileType == FTNormal && chunkNo == 1) $ startReceivingFile user fileId + case integrity of + MsgOk -> pure () + MsgError MsgDuplicate -> pure () -- TODO remove once agent removes duplicates + MsgError e -> + badRcvFileChunk ft $ "invalid file chunk number " <> show chunkNo <> ": " <> show e + -- roster: a re-driven transfer (relay restart / re-subscribe / QCONT) restarts from + -- chunk 1; discard partials so stale bytes can't corrupt the reassembled blob + when (fileType == FTRoster && chunkNo == 1) $ do + last_ <- withStore' $ \db -> getRcvFileLastChunkNo db ft + when (isJust last_) $ resetRosterPartialChunks ft + withStore' (\db -> createRcvFileChunk db ft chunkNo msgId) >>= \case + RcvChunkOk + | B.length chunk /= fromInteger chunkSize -> badRcvFileChunk ft "incorrect chunk size" + | otherwise -> appendFileChunk ft chunkNo chunk False + RcvChunkFinal + | B.length chunk > fromInteger chunkSize -> badRcvFileChunk ft "incorrect chunk size" + | otherwise -> do + appendFileChunk ft chunkNo chunk True + case fileType of + FTRoster -> forM_ gInfo_ $ \gInfo -> rosterCompletion gInfo ft + FTNormal -> do + ci <- withStore $ \db -> do + liftIO $ do + updateRcvFileStatus db fileId FSComplete + updateCIFileStatus db user fileId CIFSRcvComplete + deleteRcvFileChunks db ft + getChatItemByFileId db cxt user fileId + toView $ CEvtRcvFileComplete user ci + mapM_ (deleteAgentConnectionAsync . aConnId) conn_ + RcvChunkDuplicate -> pure () + RcvChunkError -> badRcvFileChunk ft $ "incorrect chunk number " <> show chunkNo processContactConnMessage :: AEvent e -> ConnectionEntity -> Connection -> UserContact -> CM () processContactConnMessage agentMsg connEntity conn UserContact {userContactLinkId = uclId, groupId = ucGroupId_} = case agentMsg of @@ -2166,7 +2183,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = unless (maybe False memberBlocked m') $ autoAcceptFile file_ processFileInv gInfo' m' = let fileMember_ = if sentAsGroup then Nothing else m' - in processFileInvitation fInv_ content $ \db -> createRcvGroupFileTransfer db userId gInfo' fileMember_ + in processFileInvitation fInv_ content $ \db -> createRcvGroupFileTransfer db userId gInfo' fileMember_ FTNormal sharedMsgId_ newChatItem gInfo' m' scopeInfo ciContent ciFile_ timed live = do let mentions' = if maybe False memberBlocked m' then M.empty else mentions (ci, cInfo) <- saveRcvCI gInfo' m' scopeInfo ciContent ciFile_ timed live mentions' @@ -2373,7 +2390,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = processGroupFileInvitation' gInfo m fInv@FileInvitation {fileName, fileSize} msg@RcvMessage {sharedMsgId_} brokerTs = do ChatConfig {fileChunkSize} <- asks config inline <- receiveInlineMode fInv Nothing fileChunkSize - RcvFileTransfer {fileId, xftpRcvFile} <- withStore $ \db -> createRcvGroupFileTransfer db userId gInfo (Just m) fInv inline fileChunkSize + RcvFileTransfer {fileId, xftpRcvFile} <- withStore $ \db -> createRcvGroupFileTransfer db userId gInfo (Just m) FTNormal sharedMsgId_ fInv inline fileChunkSize let fileProtocol = if isJust xftpRcvFile then FPXFTP else FPSMP ciFile = Just $ CIFile {fileId, fileName, fileSize, fileSource = Nothing, fileStatus = CIFSRcvInvitation, fileProtocol} content = ciContentNoParse $ CIRcvMsgContent $ MCFile "" @@ -2467,67 +2484,18 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = bFileChunk :: Contact -> SharedMsgId -> FileChunk -> MsgMeta -> CM () bFileChunk ct sharedMsgId chunk meta = do ft <- withStore $ \db -> getDirectFileIdBySharedMsgId db user ct sharedMsgId >>= getRcvFileTransfer db user - receiveInlineChunk ft chunk meta + receiveFileChunk Nothing ft Nothing meta chunk - -- A group BFileChunk is either a roster blob chunk (located by (group_id, shared_msg_id, - -- file_type = roster)) or a normal inline file chunk (located via its chat item). An - -- orphaned roster chunk that matches no in-flight transfer is ACKed and ignored. + -- A group BFileChunk is a normal inline file chunk or a roster blob chunk; both are located by + -- (group_id, shared_msg_id) and dispatched by file_type inside receiveFileChunk. A chunk that + -- matches no in-flight transfer (an orphaned re-served roster chunk to an up-to-date member, or + -- a missing normal file) is ignored; the outer withAckMessage acks the agent message. bFileChunkGroup :: GroupInfo -> SharedMsgId -> FileChunk -> MsgMeta -> CM () bFileChunkGroup gInfo@GroupInfo {groupId} sharedMsgId chunk meta = do - rosterFileId_ <- withStore' $ \db -> getGroupRosterFileId db userId groupId sharedMsgId - case rosterFileId_ of - Just fileId -> do - ft <- withStore $ \db -> getRcvFileTransfer db user fileId - receiveRosterChunk gInfo ft meta chunk - Nothing -> do - normalFileId_ <- withStore' $ \db -> eitherToMaybe <$> runExceptT (getGroupFileIdBySharedMsgId db userId groupId sharedMsgId) - case normalFileId_ of - Just fileId -> do - ft <- withStore $ \db -> getRcvFileTransfer db user fileId - receiveInlineChunk ft chunk meta - -- orphaned roster chunk (header short-circuited; version already applied/superseded): - -- ignore so an up-to-date member tolerates the unconditional re-serve (the outer - -- withAckMessage acks the agent message) - Nothing -> pure () - - -- Roster blob receive: reset-on-chunk-1 for a re-driven transfer, then append; the final - -- chunk drives completion (verify, apply, promote). No chat item, so no file-start/complete UI. - receiveRosterChunk :: GroupInfo -> RcvFileTransfer -> MsgMeta -> FileChunk -> CM () - receiveRosterChunk gInfo ft@RcvFileTransfer {chunkSize} MsgMeta {recipient = (msgId, _), integrity} = \case - FileChunkCancel -> cleanupGroupRosterFile user gInfo - FileChunk {chunkNo, chunkBytes = chunk} -> do - case integrity of - MsgOk -> pure () - MsgError MsgDuplicate -> pure () - MsgError e -> badRcvFileChunk ft $ "invalid file chunk number " <> show chunkNo <> ": " <> show e - -- a re-driven transfer (relay restart / re-subscribe / QCONT) restarts from the start; - -- discard partials so stale bytes can't corrupt the reassembled blob - when (chunkNo == 1) $ do - last_ <- withStore' $ \db -> getRcvFileLastChunkNo db ft - when (isJust last_) $ resetRosterPartialChunks ft - -- the outer withAckMessage ("group msg") acks the agent message; roster chunks must NOT - -- ack again (the agent message is already acked -> SEMsgNotFound) - withStore' (\db -> createRcvFileChunk db ft chunkNo msgId) >>= \case - RcvChunkOk - | B.length chunk /= fromInteger chunkSize -> badRcvFileChunk ft "incorrect chunk size" - | otherwise -> appendFileChunk ft chunkNo chunk False - RcvChunkFinal - | B.length chunk > fromInteger chunkSize -> badRcvFileChunk ft "incorrect chunk size" - | otherwise -> do - appendFileChunk ft chunkNo chunk True - rosterCompletion gInfo ft - RcvChunkDuplicate -> pure () - RcvChunkError -> badRcvFileChunk ft ("incorrect chunk number " <> show chunkNo) - - receiveInlineChunk :: RcvFileTransfer -> FileChunk -> MsgMeta -> CM () - receiveInlineChunk RcvFileTransfer {fileId, fileStatus = RFSNew} FileChunk {chunkNo} _ - | chunkNo == 1 = throwChatError $ CEInlineFileProhibited fileId - | otherwise = pure () - receiveInlineChunk ft@RcvFileTransfer {fileId} chunk meta = do - case chunk of - FileChunk {chunkNo} -> when (chunkNo == 1) $ startReceivingFile user fileId - _ -> pure () - receiveFileChunk ft Nothing meta chunk + fileId_ <- withStore' $ \db -> getGroupRcvFileIdBySharedMsgId db userId groupId sharedMsgId + forM_ fileId_ $ \fileId -> do + ft <- withStore $ \db -> getRcvFileTransfer db user fileId + receiveFileChunk (Just gInfo) ft Nothing meta chunk xFileCancelGroup :: GroupInfo -> Maybe GroupMember -> SharedMsgId -> CM (Maybe DeliveryTaskContext) xFileCancelGroup g@GroupInfo {groupId} m_ sharedMsgId = do @@ -3266,7 +3234,8 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = let relayHdr = if isUserGrpFwdRelay gInfo then Just sm else Nothing withStore' $ \db -> setRosterPending db gInfo newVer fileDigest (groupMemberId' author) brokerTs relayHdr chSize <- asks $ fileChunkSize . config - rft@RcvFileTransfer {fileId} <- withStore' $ \db -> createGroupRosterRcvFile db userId gInfo sharedMsgId fileSize (fromIntegral chSize) + let rosterFInv = FileInvitation {fileName = "roster", fileSize, fileDigest = Nothing, fileConnReq = Nothing, fileInline = Just IFMSent, fileDescr = Nothing} + rft@RcvFileTransfer {fileId} <- withStore $ \db -> createRcvGroupFileTransfer db userId gInfo Nothing FTRoster (Just sharedMsgId) rosterFInv (Just IFMSent) (fromIntegral chSize) -- accept the chat-item-free file before chunk 1 (FIFO before it) so chunk 1 isn't rejected on RFSNew filePath <- getRcvFilePath fileId Nothing "roster" False withStore' $ \db -> startRcvInlineFT db user rft filePath (Just IFMSent) diff --git a/src/Simplex/Chat/Store/Files.hs b/src/Simplex/Chat/Store/Files.hs index dfb63a9204..094ce1a967 100644 --- a/src/Simplex/Chat/Store/Files.hs +++ b/src/Simplex/Chat/Store/Files.hs @@ -31,8 +31,7 @@ module Simplex.Chat.Store.Files getSharedMsgIdByFileId, getFileIdBySharedMsgId, getGroupFileIdBySharedMsgId, - getGroupRosterFileId, - createGroupRosterRcvFile, + getGroupRcvFileIdBySharedMsgId, getGroupRosterFileInfo, deleteGroupRosterFile, getRcvFileLastChunkNo, @@ -325,56 +324,20 @@ getGroupFileIdBySharedMsgId db userId groupId sharedMsgId = |] (userId, groupId, sharedMsgId) --- The roster blob file is located by (group_id, shared_msg_id, file_type = roster), --- not by a chat item (it has none). Nothing => no in-flight roster transfer (orphaned --- chunk to an up-to-date member), which the caller ACKs and ignores. -getGroupRosterFileId :: DB.Connection -> UserId -> Int64 -> SharedMsgId -> IO (Maybe Int64) -getGroupRosterFileId db userId groupId sharedMsgId = +-- Locates a received group file (normal or roster) by the header's shared_msg_id without the +-- chat_items join, so it finds the chat-item-free roster blob and normal inline files alike. +-- Nothing => no in-flight transfer (an orphaned re-served chunk), which the caller ACKs and ignores. +getGroupRcvFileIdBySharedMsgId :: DB.Connection -> UserId -> Int64 -> SharedMsgId -> IO (Maybe Int64) +getGroupRcvFileIdBySharedMsgId db userId groupId sharedMsgId = maybeFirstRow fromOnly $ DB.query db [sql| - SELECT file_id FROM files - WHERE user_id = ? AND group_id = ? AND shared_msg_id = ? AND file_type = ? + SELECT f.file_id FROM files f + JOIN rcv_files r ON r.file_id = f.file_id + WHERE f.user_id = ? AND f.group_id = ? AND f.shared_msg_id = ? |] - (userId, groupId, sharedMsgId, FTRoster) - --- Chat-item-free received file for the roster blob: cryptoArgs are never set (the --- blob is verified as plaintext against the owner-signed digest), file_type = roster, --- located by the header's shared_msg_id. -createGroupRosterRcvFile :: DB.Connection -> UserId -> GroupInfo -> SharedMsgId -> Integer -> Integer -> IO RcvFileTransfer -createGroupRosterRcvFile db userId GroupInfo {groupId, localDisplayName = gName} sharedMsgId fileSize chunkSize = do - currentTs <- getCurrentTime - fileId <- do - DB.execute - db - [sql| - INSERT INTO files - (user_id, group_id, file_name, file_size, chunk_size, file_inline, ci_file_status, protocol, file_type, shared_msg_id, created_at, updated_at) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?) - |] - (userId, groupId, rosterFileName, fileSize, chunkSize, Just IFMSent, CIFSRcvInvitation, FPSMP, FTRoster, sharedMsgId, currentTs, currentTs) - insertedRowId db - DB.execute - db - "INSERT INTO rcv_files (file_id, file_status, file_inline, rcv_file_inline, created_at, updated_at) VALUES (?,?,?,?,?,?)" - (fileId, FSNew, Just IFMSent, Just IFMSent, currentTs, currentTs) - pure - RcvFileTransfer - { fileId, - xftpRcvFile = Nothing, - fileInvitation = FileInvitation {fileName = rosterFileName, fileSize, fileDigest = Nothing, fileConnReq = Nothing, fileInline = Just IFMSent, fileDescr = Nothing}, - fileStatus = RFSNew, - fileType = FTRoster, - rcvFileInline = Just IFMSent, - senderDisplayName = gName, - chunkSize, - cancelled = False, - grpMemberId = Nothing, - cryptoArgs = Nothing - } - where - rosterFileName = "roster" + (userId, groupId, sharedMsgId) -- For roster-file cleanup keyed on the group (not a chat item): the file_id and its -- on-disk path, so the caller can evict the cached handle and remove the file. @@ -458,8 +421,8 @@ createRcvFileTransfer db userId Contact {contactId, localDisplayName = c} f@File (fileId, FSNew, fileConnReq, fileInline, rcvFileInline, rfdId, currentTs, currentTs) pure RcvFileTransfer {fileId, xftpRcvFile, fileInvitation = f, fileStatus = RFSNew, fileType = FTNormal, rcvFileInline, senderDisplayName = c, chunkSize, cancelled = False, grpMemberId = Nothing, cryptoArgs = Nothing} -createRcvGroupFileTransfer :: DB.Connection -> UserId -> GroupInfo -> Maybe GroupMember -> FileInvitation -> Maybe InlineFileMode -> Integer -> ExceptT StoreError IO RcvFileTransfer -createRcvGroupFileTransfer db userId GroupInfo {groupId, localDisplayName = gName} m_ f@FileInvitation {fileName, fileSize, fileConnReq, fileInline, fileDescr} rcvFileInline chunkSize = do +createRcvGroupFileTransfer :: DB.Connection -> UserId -> GroupInfo -> Maybe GroupMember -> FileType -> Maybe SharedMsgId -> FileInvitation -> Maybe InlineFileMode -> Integer -> ExceptT StoreError IO RcvFileTransfer +createRcvGroupFileTransfer db userId GroupInfo {groupId, localDisplayName = gName} m_ fileType sharedMsgId_ f@FileInvitation {fileName, fileSize, fileConnReq, fileInline, fileDescr} rcvFileInline chunkSize = do currentTs <- liftIO getCurrentTime rfd_ <- mapM (createRcvFD_ db userId currentTs) fileDescr let rfdId = (\RcvFileDescr {fileDescrId} -> fileDescrId) <$> rfd_ @@ -471,15 +434,15 @@ createRcvGroupFileTransfer db userId GroupInfo {groupId, localDisplayName = gNam fileId <- liftIO $ do DB.execute db - "INSERT INTO files (user_id, group_id, file_name, file_size, chunk_size, file_inline, ci_file_status, protocol, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?)" - (userId, groupId, fileName, fileSize, chunkSize, fileInline, CIFSRcvInvitation, fileProtocol, currentTs, currentTs) + "INSERT INTO files (user_id, group_id, file_name, file_size, chunk_size, file_inline, ci_file_status, protocol, file_type, shared_msg_id, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)" + (userId, groupId, fileName, fileSize, chunkSize, fileInline, CIFSRcvInvitation, fileProtocol, fileType, sharedMsgId_, currentTs, currentTs) insertedRowId db liftIO $ DB.execute db "INSERT INTO rcv_files (file_id, file_status, file_queue_info, file_inline, rcv_file_inline, group_member_id, file_descr_id, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?)" (fileId, FSNew, fileConnReq, fileInline, rcvFileInline, grpMemberId_, rfdId, currentTs, currentTs) - pure RcvFileTransfer {fileId, xftpRcvFile, fileInvitation = f, fileStatus = RFSNew, fileType = FTNormal, rcvFileInline, senderDisplayName = senderName, chunkSize, cancelled = False, grpMemberId = grpMemberId_, cryptoArgs = Nothing} + pure RcvFileTransfer {fileId, xftpRcvFile, fileInvitation = f, fileStatus = RFSNew, fileType, rcvFileInline, senderDisplayName = senderName, chunkSize, cancelled = False, grpMemberId = grpMemberId_, cryptoArgs = Nothing} createRcvStandaloneFileTransfer :: DB.Connection -> UserId -> CryptoFile -> Int64 -> Word32 -> ExceptT StoreError IO Int64 createRcvStandaloneFileTransfer db userId (CryptoFile filePath cfArgs_) fileSize chunkSize = do