diff --git a/src/Simplex/Chat/Library/Internal.hs b/src/Simplex/Chat/Library/Internal.hs index 172caafeed..d996383d90 100644 --- a/src/Simplex/Chat/Library/Internal.hs +++ b/src/Simplex/Chat/Library/Internal.hs @@ -1884,14 +1884,30 @@ closeFileHandle fileId files = do -- The roster file has no chat item, so chat-item file enumeration misses it; clean it up by group. cleanupGroupRosterFile :: User -> GroupInfo -> CM () -cleanupGroupRosterFile User {userId} gInfo@GroupInfo {groupId} = do +cleanupGroupRosterFile User {userId} GroupInfo {groupId} = do infos <- withStore' $ \db -> getGroupRosterFileInfo db userId groupId forM_ infos $ \(fileId, filePath_) -> do lift $ closeFileHandle fileId rcvFiles forM_ filePath_ removeFsFile withStore' $ \db -> do deleteGroupRosterFile db userId groupId - clearRosterPending db gInfo + deleteGroupRosterTransfers db groupId + +-- Supersede/cancel one source relay's in-flight roster transfer: remove its on-disk file + cached +-- handle first (the cascade only does rows), then the files + transfer rows. +cleanupRosterTransfer :: GroupInfo -> GroupMemberId -> CM () +cleanupRosterTransfer gInfo fromMemberId = + withStore' (\db -> getRosterTransferId db gInfo fromMemberId) >>= mapM_ cleanupRosterTransferById + +cleanupRosterTransferById :: Int64 -> CM () +cleanupRosterTransferById transferId = do + file_ <- withStore' $ \db -> getRosterTransferFile db transferId + forM_ file_ $ \(fileId, filePath_) -> do + lift $ closeFileHandle fileId rcvFiles + forM_ filePath_ removeFsFile + withStore' $ \db -> do + deleteRosterTransferFile db transferId + deleteRosterTransfer db transferId -- MUST evict the cached AppendMode handle before deleting chunks, else re-driven bytes append -- after the stale prefix and corrupt the blob. diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index 23463c16ec..d2515ebadc 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -1084,7 +1084,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = XGrpDel -> Just (DeliveryTaskContext (DJSGroup {jobSpec = DJRelayRemoved}) False) <$ xGrpDel gInfo' m'' msg brokerTs XGrpInfo p' -> fmap ctx <$> xGrpInfo gInfo' m'' p' msg brokerTs XGrpPrefs ps' -> fmap ctx <$> xGrpPrefs gInfo' m'' ps' msg - XGrpRoster gr -> fmap ctx <$> xGrpRoster gInfo' m'' gr verifiedMsg sharedMsgId_ brokerTs + XGrpRoster gr -> fmap ctx <$> xGrpRoster gInfo' m'' m'' gr verifiedMsg sharedMsgId_ brokerTs XGrpRosterAck ackVer ackErr -> Nothing <$ xGrpRosterAck gInfo' m'' ackVer ackErr -- TODO [knocking] why don't we forward these messages? XGrpDirectInv connReq mContent_ msgScope -> memberCanSend (Just m'') msgScope $ Nothing <$ xGrpDirectInv gInfo' m'' conn' connReq mContent_ msg brokerTs @@ -1092,7 +1092,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = XInfoProbe probe -> Nothing <$ xInfoProbe (COMGroupMember m'') probe XInfoProbeCheck probeHash -> Nothing <$ xInfoProbeCheck (COMGroupMember m'') probeHash XInfoProbeOk probe -> Nothing <$ xInfoProbeOk (COMGroupMember m'') probe - BFileChunk sharedMsgId chunk -> Nothing <$ bFileChunkGroup gInfo' sharedMsgId chunk msgMeta + BFileChunk sharedMsgId chunk -> Nothing <$ bFileChunkGroup gInfo' m'' sharedMsgId chunk msgMeta _ -> Nothing <$ messageError ("unsupported message: " <> tshow event) forM deliveryTaskContext_ $ \taskContext -> pure $ NewMessageDeliveryTask {messageId = msgId, taskContext} @@ -1318,7 +1318,10 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = receiveFileChunk :: Maybe GroupInfo -> RcvFileTransfer -> Maybe Connection -> MsgMeta -> FileChunk -> CM () receiveFileChunk gInfo_ ft@RcvFileTransfer {fileId, fileType, chunkSize} conn_ MsgMeta {recipient = (msgId, _), integrity} = \case FileChunkCancel -> case fileType of - FTRoster -> forM_ gInfo_ $ cleanupGroupRosterFile user + -- cancel only this source's transfer; other relays' in-flight transfers are independent + FTRoster -> do + t_ <- withStore' $ \db -> getRosterTransfer db fileId + forM_ t_ $ \RcvRosterTransfer {rosterTransferId} -> cleanupRosterTransferById rosterTransferId FTNormal -> unless (rcvFileCompleteOrCancelled ft) $ do cancelRcvFileTransfer user ft @@ -2478,9 +2481,9 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = -- A group BFileChunk is a normal inline file chunk or a roster blob chunk, both located by -- (group_id, shared_msg_id). A chunk matching no in-flight transfer (an orphaned re-served roster -- chunk, or a missing normal file) is ignored; the outer withAckMessage acks it. - bFileChunkGroup :: GroupInfo -> SharedMsgId -> FileChunk -> MsgMeta -> CM () - bFileChunkGroup gInfo@GroupInfo {groupId} sharedMsgId chunk meta = do - fileId_ <- withStore' $ \db -> getGroupRcvFileIdBySharedMsgId db userId groupId sharedMsgId + bFileChunkGroup :: GroupInfo -> GroupMember -> SharedMsgId -> FileChunk -> MsgMeta -> CM () + bFileChunkGroup gInfo@GroupInfo {groupId} fromMember sharedMsgId chunk meta = do + fileId_ <- withStore' $ \db -> getGroupRcvFileId db userId groupId (groupMemberId' fromMember) sharedMsgId forM_ fileId_ $ \fileId -> do ft <- withStore $ \db -> getRcvFileTransfer db user fileId case fileType ft of @@ -3265,8 +3268,10 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = -- The header only starts the transfer; the roster is applied and the version bumped only at -- blob completion, so a withheld or corrupted blob leaves the last good roster intact. - xGrpRoster :: GroupInfo -> GroupMember -> GroupRoster -> VerifiedMsg e -> Maybe SharedMsgId -> UTCTime -> CM (Maybe DeliveryJobScope) - xGrpRoster gInfo author GroupRoster {version = newVer, fileInv = InlineFileInvitation {fileSize, fileDigest}} verifiedMsg sharedMsgId_ brokerTs + -- fromMember is the relay that delivered THIS roster copy (the owner on a relay receiving directly, + -- a relay on a member receiving a forward); author is the owner who signed it. + xGrpRoster :: GroupInfo -> GroupMember -> GroupMember -> GroupRoster -> VerifiedMsg e -> Maybe SharedMsgId -> UTCTime -> CM (Maybe DeliveryJobScope) + xGrpRoster gInfo fromMember author GroupRoster {version = newVer, fileInv = InlineFileInvitation {fileSize, fileDigest}} verifiedMsg sharedMsgId_ brokerTs -- only an owner may sign a roster; otherwise a relay could route it as a member whose key it controls | memberRole' author /= GROwner = messageError "x.grp.roster: not signed by an owner" $> Nothing | fileSize > maxGroupRosterBytes = messageError "x.grp.roster: roster blob size exceeds limit" $> Nothing @@ -3276,68 +3281,73 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = VMSigned _ sm _ -> case sharedMsgId_ of Nothing -> Nothing <$ messageWarning "x.grp.roster: missing shared message id" Just sharedMsgId -> do - pendingVer_ <- fmap (\(v, _, _, _, _) -> v) <$> withStore' (\db -> getRosterPending db gInfo) - -- accept a version not below BOTH applied and pending (>=, Nothing below 0): a preceding signed - -- event may have already advanced rosterVersion to this blob's version; a lower one is a downgrade. - if newVer `aboveRoster` rosterVersion gInfo && newVer `aboveRoster` pendingVer_ + -- per-source pending version (THIS relay's own in-flight transfer), not a single group slot + pendingVer_ <- withStore' $ \db -> getRosterTransferVersion db gInfo (groupMemberId' fromMember) + -- accept a version not below BOTH applied and this source's pending (>=, Nothing below 0): a preceding + -- signed event may have already advanced rosterVersion to this blob's version; a lower one is a downgrade. + if newVer `notBelowRoster` rosterVersion gInfo && newVer `notBelowRoster` pendingVer_ then startRosterTransfer sm sharedMsgId else pure Nothing where startRosterTransfer sm sharedMsgId = do - -- supersede any in-flight roster file (older version or a restart) before the new transfer - cleanupGroupRosterFile user gInfo + -- supersede THIS source's own in-flight transfer (older version or a restart); other relays' transfers are independent + cleanupRosterTransfer gInfo (groupMemberId' fromMember) 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 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) + -- transfer record + its scratch file in one transaction (file owned by the transfer, keyed per source) + rft@RcvFileTransfer {fileId} <- withStore $ \db -> do + transferId <- liftIO $ createRosterTransfer db gInfo (groupMemberId' fromMember) newVer fileDigest (groupMemberId' author) brokerTs relayHdr + createRosterRcvFile db userId gInfo fromMember transferId 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 -- transient scratch file (consumed into roster_blob, then deleted): temp folder, not the user's files folder / Downloads tmpDir <- lift getChatTempDirectory rosterTs <- liftIO getCurrentTime let GroupInfo {groupId = gId} = gInfo - rosterFile = "roster_" <> show gId <> "_" <> formatTime defaultTimeLocale "%Y%m%d_%H%M%S" rosterTs + rosterFile = "roster_" <> show gId <> "_" <> show (groupMemberId' fromMember) <> "_" <> formatTime defaultTimeLocale "%Y%m%d_%H%M%S" rosterTs filePath <- getRcvFilePath fileId (Just tmpDir) rosterFile False withStore' $ \db -> startRcvInlineFT db user rft filePath (Just IFMSent) pure Nothing -- Roster version comparison treating Nothing (un-materialized) as below 0. Non-strict (>=) so a relay -- accepts the owner's blob at the version a preceding signed event already advanced rosterVersion to. - aboveRoster :: VersionRoster -> Maybe VersionRoster -> Bool - aboveRoster v = maybe True (v >=) + notBelowRoster :: VersionRoster -> Maybe VersionRoster -> Bool + notBelowRoster v = maybe True (v >=) -- Blob arrived: verify the owner-attested digest over the plaintext and guard against -- downgrade before applying; on a relay, ack the owner and re-serve to members. rosterCompletion :: GroupInfo -> RcvFileTransfer -> CM () - rosterCompletion gInfo RcvFileTransfer {fileStatus} = - withStore' (\db -> getRosterPending db gInfo) >>= \case - Nothing -> cleanupGroupRosterFile user gInfo - Just (pendingVer, pendingDigest, ownerGMId, rosterBrokerTs, _) -> do + rosterCompletion gInfo RcvFileTransfer {fileId, fileStatus} = + withStore' (\db -> getRosterTransfer db fileId) >>= \case + -- defensive: the file always has its transfer (created together, deleted together) + Nothing -> lift (closeFileHandle fileId rcvFiles) >> forM_ (rosterFilePath fileStatus) removeFsFile + Just RcvRosterTransfer {rosterTransferId = transferId, rosterTransferVersion = pendingVer, rosterTransferDigest = pendingDigest, rosterTransferOwnerGMId = ownerGMId, rosterTransferBrokerTs = rosterBrokerTs, rosterTransferHeader = header_} -> do owner_ <- withStore' $ \db -> eitherToMaybe <$> runExceptT (getGroupMemberById db cxt user ownerGMId) blob <- readAssembledRoster let isRelay = isUserGrpFwdRelay gInfo ackErr err = do - cleanupGroupRosterFile user gInfo + cleanupRosterTransferById transferId when isRelay $ forM_ owner_ $ \owner -> sendRosterAck gInfo owner pendingVer (Just err) if FD.FileDigest (LC.sha512Hash (LB.fromStrict blob)) /= pendingDigest then ackErr "relay could not verify the roster blob" else case parseAll rosterBlobP blob of Left _ -> ackErr "relay could not parse the roster blob" Right entries -> case owner_ of - Nothing -> cleanupGroupRosterFile user gInfo + Nothing -> cleanupRosterTransferById transferId Just author -> do defaultRole <- unknownMemberRole gInfo - -- gate against the persisted roster_version inside the apply transaction, so a reordered - -- same-batch event cannot make this completion downgrade it; stale completion is rejected. + -- gate against the persisted roster_version inside the apply transaction: a roster from another + -- relay (or a preceding signed event) may already have advanced it past this one; a stale + -- completion (e.g. relay1 sent v5 then v6, relay2's v5 completes after v6) is rejected. results_ <- withStore $ \db -> do cur <- liftIO $ getGroupRosterVersion db gInfo if maybe False (pendingVer <) cur then pure Nothing else do res <- processRosterEntries db gInfo defaultRole (validateGroupRoster entries) - liftIO $ promoteRosterPending db gInfo blob + liftIO $ setGroupLiveRoster db gInfo pendingVer ownerGMId rosterBrokerTs header_ blob pure (Just res) - cleanupGroupRosterFile user gInfo + cleanupRosterTransferById transferId forM_ results_ $ \results -> do emitRosterResults gInfo author rosterBrokerTs results -- ack while setting up (own status accepted/acknowledged); a serving (active) relay must not ack broadcasts. @@ -3345,11 +3355,14 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = sendRosterAck gInfo author pendingVer Nothing withStore' $ \db -> void $ updateRelayOwnStatusFromTo db gInfo RSAccepted RSAcknowledgedRoster where - readAssembledRoster = case fileStatus of - RFSAccepted fp -> readAt fp - RFSConnected fp -> readAt fp - RFSComplete fp -> readAt fp - _ -> throwChatError $ CEInternalError "roster file not in progress" + rosterFilePath = \case + RFSAccepted p -> Just p + RFSConnected p -> Just p + RFSComplete p -> Just p + _ -> Nothing + readAssembledRoster = case rosterFilePath fileStatus of + Just fp -> readAt fp + Nothing -> throwChatError $ CEInternalError "roster file not in progress" readAt fp = lift (toFSFilePath fp) >>= liftIO . B.readFile -- TOFU-apply an owner-signed (key, role) to a resolved member: pin the key if absent; for a keyed @@ -3755,7 +3768,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = XGrpDel -> withAuthor XGrpDel_ $ \author -> void $ xGrpDel gInfo author rcvMsg msgTs XGrpInfo p' -> withAuthor XGrpInfo_ $ \author -> void $ xGrpInfo gInfo author p' rcvMsg msgTs XGrpPrefs ps' -> withAuthor XGrpPrefs_ $ \author -> void $ xGrpPrefs gInfo author ps' rcvMsg - XGrpRoster gr -> withAuthor XGrpRoster_ $ \author -> void $ xGrpRoster gInfo author gr verifiedMsg sharedMsgId_ msgTs + XGrpRoster gr -> withAuthor XGrpRoster_ $ \author -> void $ xGrpRoster gInfo m author gr verifiedMsg sharedMsgId_ msgTs _ -> messageError $ "x.grp.msg.forward: unsupported forwarded event " <> T.pack (show $ toCMEventTag event) where withAuthor :: CMEventTag e -> (GroupMember -> CM ()) -> CM () diff --git a/src/Simplex/Chat/Store/Files.hs b/src/Simplex/Chat/Store/Files.hs index 2c2475a9ba..dee72731a8 100644 --- a/src/Simplex/Chat/Store/Files.hs +++ b/src/Simplex/Chat/Store/Files.hs @@ -31,9 +31,11 @@ module Simplex.Chat.Store.Files getSharedMsgIdByFileId, getFileIdBySharedMsgId, getGroupFileIdBySharedMsgId, - getGroupRcvFileIdBySharedMsgId, + getGroupRcvFileId, getGroupRosterFileInfo, deleteGroupRosterFile, + getRosterTransferFile, + deleteRosterTransferFile, getRcvFileLastChunkNo, getDirectFileIdBySharedMsgId, getChatRefByFileId, @@ -41,6 +43,7 @@ module Simplex.Chat.Store.Files updateSndFileStatus, createRcvFileTransfer, createRcvGroupFileTransfer, + createRosterRcvFile, createRcvStandaloneFileTransfer, appendRcvFD, getRcvFileDescrByRcvFileId, @@ -325,20 +328,42 @@ getGroupFileIdBySharedMsgId db userId groupId sharedMsgId = |] (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 +-- Resolve the in-flight received group inline file for a chunk: read its file_type by shared_msg_id +-- (LIMIT 1 is safe -- all files sharing a shared_msg_id share a type), then look up by type: a roster +-- file is scoped to its source relay (every relay re-serves the owner's same shared_msg_id, so the source +-- disambiguates), a normal file is by shared_msg_id. Nothing => no in-flight transfer (orphaned chunk). +getGroupRcvFileId :: DB.Connection -> UserId -> Int64 -> GroupMemberId -> SharedMsgId -> IO (Maybe Int64) +getGroupRcvFileId db userId groupId fromMemberId sharedMsgId = do + fileType_ <- getFileType + case fileType_ of + Just FTRoster -> + maybeFirstRow fromOnly $ + DB.query db (rcvFileIdQ <> " AND r.group_member_id = ?") (userId, groupId, sharedMsgId, FTRoster, fromMemberId) + Just FTNormal -> + maybeFirstRow fromOnly $ + DB.query db rcvFileIdQ (userId, groupId, sharedMsgId, FTNormal) + Nothing -> pure Nothing + where + getFileType = + maybeFirstRow fromOnly $ + DB.query db "SELECT file_type FROM files WHERE user_id = ? AND group_id = ? AND shared_msg_id = ? LIMIT 1" (userId, groupId, sharedMsgId) + rcvFileIdQ = [sql| 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 = ? + WHERE f.user_id = ? AND f.group_id = ? AND f.shared_msg_id = ? AND f.file_type = ? |] - (userId, groupId, sharedMsgId) + +-- The roster scratch file for a transfer (for fs/handle cleanup before deleting the transfer). +-- A transfer owns exactly one file (created together in one transaction), so this is single-valued. +getRosterTransferFile :: DB.Connection -> Int64 -> IO (Maybe (Int64, Maybe FilePath)) +getRosterTransferFile db transferId = + maybeFirstRow id $ DB.query db "SELECT file_id, file_path FROM files WHERE roster_transfer_id = ?" (Only transferId) + +-- Deletes a transfer's file row; rcv_files and rcv_file_chunks cascade on the FK. +deleteRosterTransferFile :: DB.Connection -> Int64 -> IO () +deleteRosterTransferFile db transferId = + DB.execute db "DELETE FROM files WHERE roster_transfer_id = ?" (Only transferId) -- For roster-file cleanup keyed on the group (not a chat item): every matching file_id and its on-disk -- path, so the caller evicts the handle and removes the file for each — delete-all like deleteGroupRosterFile. @@ -444,6 +469,25 @@ createRcvGroupFileTransfer db userId GroupInfo {groupId, localDisplayName = gNam (fileId, FSNew, fileConnReq, fileInline, rcvFileInline, grpMemberId_, rfdId, currentTs, currentTs) pure RcvFileTransfer {fileId, xftpRcvFile, fileInvitation = f, fileStatus = RFSNew, fileType, rcvFileInline, senderDisplayName = senderName, chunkSize, cancelled = False, grpMemberId = grpMemberId_, cryptoArgs = Nothing} +-- Roster scratch file owned by a per-source transfer: group_member_id is the delivering relay (so chunk +-- streams from different relays are distinct files), roster_transfer_id links to the metadata record. +createRosterRcvFile :: DB.Connection -> UserId -> GroupInfo -> GroupMember -> Int64 -> SharedMsgId -> FileInvitation -> Maybe InlineFileMode -> Integer -> ExceptT StoreError IO RcvFileTransfer +createRosterRcvFile db userId GroupInfo {groupId} src@GroupMember {localDisplayName = senderName} transferId sharedMsgId f@FileInvitation {fileName, fileSize, fileConnReq, fileInline} rcvFileInline chunkSize = do + currentTs <- liftIO getCurrentTime + let grpMemberId_ = groupMemberId' src + 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, file_type, shared_msg_id, roster_transfer_id, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)" + ((userId, groupId, fileName, fileSize, chunkSize, fileInline, CIFSRcvInvitation, FPSMP, FTRoster) :. (sharedMsgId, transferId, 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, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?)" + (fileId, FSNew, fileConnReq, fileInline, rcvFileInline, grpMemberId_, currentTs, currentTs) + pure RcvFileTransfer {fileId, xftpRcvFile = Nothing, fileInvitation = f, fileStatus = RFSNew, fileType = FTRoster, rcvFileInline, senderDisplayName = senderName, chunkSize, cancelled = False, grpMemberId = Just grpMemberId_, cryptoArgs = Nothing} + createRcvStandaloneFileTransfer :: DB.Connection -> UserId -> CryptoFile -> Int64 -> Word32 -> ExceptT StoreError IO Int64 createRcvStandaloneFileTransfer db userId (CryptoFile filePath cfArgs_) fileSize chunkSize = do currentTs <- liftIO getCurrentTime diff --git a/src/Simplex/Chat/Store/Groups.hs b/src/Simplex/Chat/Store/Groups.hs index 3ca6223a06..9c0fffa086 100644 --- a/src/Simplex/Chat/Store/Groups.hs +++ b/src/Simplex/Chat/Store/Groups.hs @@ -91,10 +91,14 @@ module Simplex.Chat.Store.Groups setGroupRosterVersion, getGroupRosterVersion, getGroupRoster, - setRosterPending, - getRosterPending, - promoteRosterPending, - clearRosterPending, + RcvRosterTransfer (..), + createRosterTransfer, + getRosterTransferVersion, + getRosterTransferId, + getRosterTransfer, + setGroupLiveRoster, + deleteRosterTransfer, + deleteGroupRosterTransfers, setGroupMemberKeyRole, createRelayForOwner, getCreateRelayForMember, @@ -1479,90 +1483,115 @@ getGroupRoster db GroupInfo {groupId} = (\sigs -> (ownerGMId, brokerTs, SignedMsg cb sigs body, (\(Binary b) -> b) <$> blob_)) <$> eitherToMaybe (smpDecode sigsBs) toRoster _ = Nothing --- The signed-header columns are relay-only (Nothing on a member); promoted to the live header at --- completion so the relay can re-forward the roster. -setRosterPending :: DB.Connection -> GroupInfo -> VersionRoster -> FD.FileDigest -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> IO () -setRosterPending db GroupInfo {groupId} v digest ownerGMId brokerTs sm_ = do - currentTs <- getCurrentTime +-- A per-source in-flight roster transfer, keyed (group_id, from_member_id): replaces the single +-- roster_pending_* slot, so two relays serving one member can't share a chunk stream. The signed-header +-- columns are relay-only (NULL on members), promoted to the live roster_msg_* on groups at completion. +createRosterTransfer :: DB.Connection -> GroupInfo -> GroupMemberId -> VersionRoster -> FD.FileDigest -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> IO Int64 +createRosterTransfer db GroupInfo {groupId} fromMemberId v digest ownerGMId brokerTs sm_ = do + -- one in-flight transfer per (group, source): drop any prior row from this source so the INSERT can't hit + -- the UNIQUE constraint even if the caller's fs/handle cleanup was skipped (the scratch file would then leak + -- until group delete, but the transfer never gets stuck). Normally cleanupRosterTransfer ran first. + DB.execute db "DELETE FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ?" (groupId, fromMemberId) DB.execute db [sql| - UPDATE groups - SET roster_pending_version = ?, roster_pending_digest = ?, - roster_pending_sending_owner_gm_id = ?, roster_pending_broker_ts = ?, - roster_pending_msg_chat_binding = ?, roster_pending_msg_signatures = ?, roster_pending_msg_body = ?, - updated_at = ? - WHERE group_id = ? + INSERT INTO rcv_roster_transfers + (group_id, from_member_id, roster_version, roster_digest, sending_owner_gm_id, broker_ts, + roster_msg_chat_binding, roster_msg_signatures, roster_msg_body) + VALUES (?,?,?,?,?,?,?,?,?) |] - ((v, Binary (FD.unFileDigest digest), ownerGMId, brokerTs) - :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_, currentTs, groupId)) + ( (groupId, fromMemberId, v, Binary (FD.unFileDigest digest), ownerGMId, brokerTs) + :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_) + ) + insertedRowId db -getRosterPending :: DB.Connection -> GroupInfo -> IO (Maybe (VersionRoster, FD.FileDigest, GroupMemberId, UTCTime, Maybe SignedMsg)) -getRosterPending db GroupInfo {groupId} = - (>>= toPending) +getRosterTransferVersion :: DB.Connection -> GroupInfo -> GroupMemberId -> IO (Maybe VersionRoster) +getRosterTransferVersion db GroupInfo {groupId} fromMemberId = + maybeFirstRow fromOnly $ + DB.query db "SELECT roster_version FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ?" (groupId, fromMemberId) + +getRosterTransferId :: DB.Connection -> GroupInfo -> GroupMemberId -> IO (Maybe Int64) +getRosterTransferId db GroupInfo {groupId} fromMemberId = + maybeFirstRow fromOnly $ + DB.query db "SELECT roster_transfer_id FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ?" (groupId, fromMemberId) + +-- An in-flight received roster transfer (a rcv_roster_transfers row joined to its scratch file), read at +-- completion. The header is the relay's re-serve SignedMsg -- present only on a serving relay (NULL on a +-- member, whose live roster_msg_* stay NULL so it never re-serves). +data RcvRosterTransfer = RcvRosterTransfer + { rosterTransferId :: Int64, + rosterTransferVersion :: VersionRoster, + rosterTransferDigest :: FD.FileDigest, + rosterTransferOwnerGMId :: GroupMemberId, + rosterTransferBrokerTs :: UTCTime, + rosterTransferHeader :: Maybe SignedMsg + } + deriving (Show) + +-- The in-flight transfer for a received roster file (joined via files.roster_transfer_id), with its +-- relay-only signed header. Read at completion to apply, promote into the live roster, and ack. +getRosterTransfer :: DB.Connection -> Int64 -> IO (Maybe RcvRosterTransfer) +getRosterTransfer db fileId = + (>>= toTransfer) <$> maybeFirstRow id ( DB.query db [sql| - SELECT roster_pending_version, roster_pending_digest, - roster_pending_sending_owner_gm_id, roster_pending_broker_ts, - roster_pending_msg_chat_binding, roster_pending_msg_signatures, roster_pending_msg_body - FROM groups WHERE group_id = ? + SELECT t.roster_transfer_id, t.roster_version, t.roster_digest, t.sending_owner_gm_id, t.broker_ts, + t.roster_msg_chat_binding, t.roster_msg_signatures, t.roster_msg_body + FROM rcv_roster_transfers t + JOIN files f ON f.roster_transfer_id = t.roster_transfer_id + WHERE f.file_id = ? |] - (Only groupId) + (Only fileId) ) where - toPending (Just v, Just (Binary d), Just ownerGMId, Just brokerTs, cb_, sigs_, body_) = - Just (v, FD.FileDigest d, ownerGMId, brokerTs, sm_) + toTransfer (tId, v, Binary d, ownerGMId, brokerTs, cb_, sigs_, body_) = + Just + RcvRosterTransfer + { rosterTransferId = tId, + rosterTransferVersion = v, + rosterTransferDigest = FD.FileDigest d, + rosterTransferOwnerGMId = ownerGMId, + rosterTransferBrokerTs = brokerTs, + rosterTransferHeader = sm_ + } where sm_ = case (cb_, sigs_, body_) of (Just cb, Just (Binary sigsBs), Just (Binary body)) -> (\sigs -> SignedMsg cb sigs body) <$> eitherToMaybe (smpDecode sigsBs) _ -> Nothing - toPending _ = Nothing --- Promote pending -> live (header + blob + version) and clear pending, in one statement. -promoteRosterPending :: DB.Connection -> GroupInfo -> ByteString -> IO () -promoteRosterPending db GroupInfo {groupId} blob = do +-- Write the single live roster on groups from a completed transfer's values (header NULL on a member, +-- so its live roster_msg_* stay NULL and it never re-serves; only relays re-serve). +setGroupLiveRoster :: DB.Connection -> GroupInfo -> VersionRoster -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> ByteString -> IO () +setGroupLiveRoster db GroupInfo {groupId} v ownerGMId brokerTs sm_ blob = do currentTs <- getCurrentTime DB.execute db [sql| UPDATE groups SET - roster_version = roster_pending_version, - roster_blob = ?, - roster_sending_owner_gm_id = roster_pending_sending_owner_gm_id, - roster_broker_ts = roster_pending_broker_ts, - roster_msg_chat_binding = roster_pending_msg_chat_binding, - roster_msg_signatures = roster_pending_msg_signatures, - roster_msg_body = roster_pending_msg_body, - roster_pending_version = NULL, - roster_pending_digest = NULL, - roster_pending_sending_owner_gm_id = NULL, - roster_pending_broker_ts = NULL, - roster_pending_msg_chat_binding = NULL, - roster_pending_msg_signatures = NULL, - roster_pending_msg_body = NULL, + roster_version = ?, roster_blob = ?, + roster_sending_owner_gm_id = ?, roster_broker_ts = ?, + roster_msg_chat_binding = ?, roster_msg_signatures = ?, roster_msg_body = ?, updated_at = ? WHERE group_id = ? |] - (Binary blob, currentTs, groupId) + ( (v, Binary blob, ownerGMId, brokerTs) + :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_, currentTs, groupId) + ) -clearRosterPending :: DB.Connection -> GroupInfo -> IO () -clearRosterPending db GroupInfo {groupId} = do - currentTs <- getCurrentTime - DB.execute - db - [sql| - UPDATE groups SET - roster_pending_version = NULL, roster_pending_digest = NULL, - roster_pending_sending_owner_gm_id = NULL, roster_pending_broker_ts = NULL, - roster_pending_msg_chat_binding = NULL, roster_pending_msg_signatures = NULL, roster_pending_msg_body = NULL, - updated_at = ? - WHERE group_id = ? - |] - (currentTs, groupId) +-- Delete one in-flight transfer row (its files/rcv_files/rcv_file_chunks are removed separately, with +-- the on-disk file). Caller removes the fs file + cached handle first. +deleteRosterTransfer :: DB.Connection -> Int64 -> IO () +deleteRosterTransfer db transferId = + DB.execute db "DELETE FROM rcv_roster_transfers WHERE roster_transfer_id = ?" (Only transferId) + +-- All in-flight transfers for a group (group delete). +deleteGroupRosterTransfers :: DB.Connection -> Int64 -> IO () +deleteGroupRosterTransfers db groupId = + DB.execute db "DELETE FROM rcv_roster_transfers WHERE group_id = ?" (Only groupId) setGroupMemberKeyRole :: DB.Connection -> GroupMember -> C.PublicKeyEd25519 -> GroupMemberRole -> IO () setGroupMemberKeyRole db GroupMember {groupMemberId} pubKey role = do diff --git a/src/Simplex/Chat/Store/Postgres/Migrations/M20260601_group_roster.hs b/src/Simplex/Chat/Store/Postgres/Migrations/M20260601_group_roster.hs index 618b354fda..6497e525a9 100644 --- a/src/Simplex/Chat/Store/Postgres/Migrations/M20260601_group_roster.hs +++ b/src/Simplex/Chat/Store/Postgres/Migrations/M20260601_group_roster.hs @@ -16,31 +16,44 @@ ALTER TABLE groups ADD COLUMN roster_msg_signatures BYTEA; ALTER TABLE groups ADD COLUMN roster_sending_owner_gm_id BIGINT; ALTER TABLE groups ADD COLUMN roster_broker_ts TIMESTAMPTZ; ALTER TABLE groups ADD COLUMN roster_blob BYTEA; -ALTER TABLE groups ADD COLUMN roster_pending_version BIGINT; -ALTER TABLE groups ADD COLUMN roster_pending_digest BYTEA; -ALTER TABLE groups ADD COLUMN roster_pending_msg_body BYTEA; -ALTER TABLE groups ADD COLUMN roster_pending_msg_chat_binding TEXT; -ALTER TABLE groups ADD COLUMN roster_pending_msg_signatures BYTEA; -ALTER TABLE groups ADD COLUMN roster_pending_sending_owner_gm_id BIGINT; -ALTER TABLE groups ADD COLUMN roster_pending_broker_ts TIMESTAMPTZ; + +CREATE TABLE rcv_roster_transfers( + roster_transfer_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, + group_id BIGINT NOT NULL REFERENCES groups ON DELETE CASCADE, + from_member_id BIGINT NOT NULL REFERENCES group_members ON DELETE CASCADE, + roster_version BIGINT NOT NULL, + roster_digest BYTEA NOT NULL, + sending_owner_gm_id BIGINT NOT NULL, + broker_ts TIMESTAMPTZ NOT NULL, + roster_msg_body BYTEA, + roster_msg_chat_binding TEXT, + roster_msg_signatures BYTEA, + created_at TEXT NOT NULL DEFAULT (now()), + updated_at TEXT NOT NULL DEFAULT (now()) +); +CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON rcv_roster_transfers(group_id, from_member_id); +CREATE INDEX idx_rcv_roster_transfers_from_member_id ON rcv_roster_transfers(from_member_id); + ALTER TABLE files ADD COLUMN shared_msg_id BYTEA; ALTER TABLE files ADD COLUMN file_type TEXT NOT NULL DEFAULT 'normal'; +ALTER TABLE files ADD COLUMN roster_transfer_id BIGINT; CREATE INDEX idx_files_group_id_shared_msg_id ON files(group_id, shared_msg_id); +CREATE INDEX idx_files_roster_transfer_id ON files(roster_transfer_id); |] down_m20260601_group_roster :: Text down_m20260601_group_roster = [r| -DROP INDEX IF EXISTS idx_files_group_id_shared_msg_id; +DROP INDEX idx_files_roster_transfer_id; +DROP INDEX idx_files_group_id_shared_msg_id; +ALTER TABLE files DROP COLUMN roster_transfer_id; ALTER TABLE files DROP COLUMN file_type; ALTER TABLE files DROP COLUMN shared_msg_id; -ALTER TABLE groups DROP COLUMN roster_pending_broker_ts; -ALTER TABLE groups DROP COLUMN roster_pending_sending_owner_gm_id; -ALTER TABLE groups DROP COLUMN roster_pending_msg_signatures; -ALTER TABLE groups DROP COLUMN roster_pending_msg_chat_binding; -ALTER TABLE groups DROP COLUMN roster_pending_msg_body; -ALTER TABLE groups DROP COLUMN roster_pending_digest; -ALTER TABLE groups DROP COLUMN roster_pending_version; + +DROP INDEX idx_rcv_roster_transfers_from_member_id; +DROP INDEX idx_rcv_roster_transfers_group_id_from_member_id; +DROP TABLE rcv_roster_transfers; + ALTER TABLE groups DROP COLUMN roster_blob; ALTER TABLE groups DROP COLUMN roster_broker_ts; ALTER TABLE groups DROP COLUMN roster_sending_owner_gm_id; diff --git a/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql b/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql index 9711508cf1..bbd4aeb593 100644 --- a/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql +++ b/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql @@ -754,7 +754,8 @@ CREATE TABLE test_chat_schema.files ( note_folder_id bigint, redirect_file_id bigint, shared_msg_id bytea, - file_type text DEFAULT 'normal'::text NOT NULL + file_type text DEFAULT 'normal'::text NOT NULL, + roster_transfer_id bigint ); @@ -987,14 +988,7 @@ CREATE TABLE test_chat_schema.groups ( roster_msg_signatures bytea, roster_sending_owner_gm_id bigint, roster_broker_ts timestamp with time zone, - roster_blob bytea, - roster_pending_version bigint, - roster_pending_digest bytea, - roster_pending_msg_body bytea, - roster_pending_msg_chat_binding text, - roster_pending_msg_signatures bytea, - roster_pending_sending_owner_gm_id bigint, - roster_pending_broker_ts timestamp with time zone + roster_blob bytea ); @@ -1221,6 +1215,34 @@ CREATE TABLE test_chat_schema.rcv_files ( +CREATE TABLE test_chat_schema.rcv_roster_transfers ( + roster_transfer_id bigint NOT NULL, + group_id bigint NOT NULL, + from_member_id bigint NOT NULL, + roster_version bigint NOT NULL, + roster_digest bytea NOT NULL, + sending_owner_gm_id bigint NOT NULL, + broker_ts timestamp with time zone NOT NULL, + roster_msg_body bytea, + roster_msg_chat_binding text, + roster_msg_signatures bytea, + created_at text DEFAULT now() NOT NULL, + updated_at text DEFAULT now() NOT NULL +); + + + +ALTER TABLE test_chat_schema.rcv_roster_transfers ALTER COLUMN roster_transfer_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME test_chat_schema.rcv_roster_transfers_roster_transfer_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + CREATE TABLE test_chat_schema.received_probes ( received_probe_id bigint NOT NULL, contact_id bigint, @@ -1754,6 +1776,11 @@ ALTER TABLE ONLY test_chat_schema.rcv_files +ALTER TABLE ONLY test_chat_schema.rcv_roster_transfers + ADD CONSTRAINT rcv_roster_transfers_pkey PRIMARY KEY (roster_transfer_id); + + + ALTER TABLE ONLY test_chat_schema.received_probes ADD CONSTRAINT received_probes_pkey PRIMARY KEY (received_probe_id); @@ -2295,6 +2322,10 @@ CREATE INDEX idx_files_redirect_file_id ON test_chat_schema.files USING btree (r +CREATE INDEX idx_files_roster_transfer_id ON test_chat_schema.files USING btree (roster_transfer_id); + + + CREATE INDEX idx_files_user_id ON test_chat_schema.files USING btree (user_id); @@ -2467,6 +2498,14 @@ CREATE INDEX idx_rcv_files_group_member_id ON test_chat_schema.rcv_files USING b +CREATE INDEX idx_rcv_roster_transfers_from_member_id ON test_chat_schema.rcv_roster_transfers USING btree (from_member_id); + + + +CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON test_chat_schema.rcv_roster_transfers USING btree (group_id, from_member_id); + + + CREATE INDEX idx_received_probes_contact_id ON test_chat_schema.received_probes USING btree (contact_id); @@ -3152,6 +3191,16 @@ ALTER TABLE ONLY test_chat_schema.rcv_files +ALTER TABLE ONLY test_chat_schema.rcv_roster_transfers + ADD CONSTRAINT rcv_roster_transfers_from_member_id_fkey FOREIGN KEY (from_member_id) REFERENCES test_chat_schema.group_members(group_member_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY test_chat_schema.rcv_roster_transfers + ADD CONSTRAINT rcv_roster_transfers_group_id_fkey FOREIGN KEY (group_id) REFERENCES test_chat_schema.groups(group_id) ON DELETE CASCADE; + + + ALTER TABLE ONLY test_chat_schema.received_probes ADD CONSTRAINT received_probes_contact_id_fkey FOREIGN KEY (contact_id) REFERENCES test_chat_schema.contacts(contact_id) ON DELETE CASCADE; diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/M20260601_group_roster.hs b/src/Simplex/Chat/Store/SQLite/Migrations/M20260601_group_roster.hs index f5179e3a07..2785857fd6 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations/M20260601_group_roster.hs +++ b/src/Simplex/Chat/Store/SQLite/Migrations/M20260601_group_roster.hs @@ -15,31 +15,44 @@ ALTER TABLE groups ADD COLUMN roster_msg_signatures BLOB; ALTER TABLE groups ADD COLUMN roster_sending_owner_gm_id INTEGER; ALTER TABLE groups ADD COLUMN roster_broker_ts TEXT; ALTER TABLE groups ADD COLUMN roster_blob BLOB; -ALTER TABLE groups ADD COLUMN roster_pending_version INTEGER; -ALTER TABLE groups ADD COLUMN roster_pending_digest BLOB; -ALTER TABLE groups ADD COLUMN roster_pending_msg_body BLOB; -ALTER TABLE groups ADD COLUMN roster_pending_msg_chat_binding TEXT; -ALTER TABLE groups ADD COLUMN roster_pending_msg_signatures BLOB; -ALTER TABLE groups ADD COLUMN roster_pending_sending_owner_gm_id INTEGER; -ALTER TABLE groups ADD COLUMN roster_pending_broker_ts TEXT; + +CREATE TABLE rcv_roster_transfers( + roster_transfer_id INTEGER PRIMARY KEY, + group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE, + from_member_id INTEGER NOT NULL REFERENCES group_members ON DELETE CASCADE, + roster_version INTEGER NOT NULL, + roster_digest BLOB NOT NULL, + sending_owner_gm_id INTEGER NOT NULL, + broker_ts TEXT NOT NULL, + roster_msg_body BLOB, + roster_msg_chat_binding TEXT, + roster_msg_signatures BLOB, + created_at TEXT NOT NULL DEFAULT(datetime('now')), + updated_at TEXT NOT NULL DEFAULT(datetime('now')) +) STRICT; +CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON rcv_roster_transfers(group_id, from_member_id); +CREATE INDEX idx_rcv_roster_transfers_from_member_id ON rcv_roster_transfers(from_member_id); + ALTER TABLE files ADD COLUMN shared_msg_id BLOB; ALTER TABLE files ADD COLUMN file_type TEXT NOT NULL DEFAULT 'normal'; +ALTER TABLE files ADD COLUMN roster_transfer_id INTEGER; CREATE INDEX idx_files_group_id_shared_msg_id ON files(group_id, shared_msg_id); +CREATE INDEX idx_files_roster_transfer_id ON files(roster_transfer_id); |] down_m20260601_group_roster :: Query down_m20260601_group_roster = [sql| -DROP INDEX IF EXISTS idx_files_group_id_shared_msg_id; +DROP INDEX idx_files_roster_transfer_id; +DROP INDEX idx_files_group_id_shared_msg_id; +ALTER TABLE files DROP COLUMN roster_transfer_id; ALTER TABLE files DROP COLUMN file_type; ALTER TABLE files DROP COLUMN shared_msg_id; -ALTER TABLE groups DROP COLUMN roster_pending_broker_ts; -ALTER TABLE groups DROP COLUMN roster_pending_sending_owner_gm_id; -ALTER TABLE groups DROP COLUMN roster_pending_msg_signatures; -ALTER TABLE groups DROP COLUMN roster_pending_msg_chat_binding; -ALTER TABLE groups DROP COLUMN roster_pending_msg_body; -ALTER TABLE groups DROP COLUMN roster_pending_digest; -ALTER TABLE groups DROP COLUMN roster_pending_version; + +DROP INDEX idx_rcv_roster_transfers_from_member_id; +DROP INDEX idx_rcv_roster_transfers_group_id_from_member_id; +DROP TABLE rcv_roster_transfers; + ALTER TABLE groups DROP COLUMN roster_blob; ALTER TABLE groups DROP COLUMN roster_broker_ts; ALTER TABLE groups DROP COLUMN roster_sending_owner_gm_id; diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql b/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql index 36f84cf89d..bc905b8ea9 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql +++ b/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql @@ -198,14 +198,7 @@ CREATE TABLE groups( roster_msg_signatures BLOB, roster_sending_owner_gm_id INTEGER, roster_broker_ts TEXT, - roster_blob BLOB, - roster_pending_version INTEGER, - roster_pending_digest BLOB, - roster_pending_msg_body BLOB, - roster_pending_msg_chat_binding TEXT, - roster_pending_msg_signatures BLOB, - roster_pending_sending_owner_gm_id INTEGER, - roster_pending_broker_ts TEXT, -- received + roster_blob BLOB, -- received FOREIGN KEY(user_id, local_display_name) REFERENCES display_names(user_id, local_display_name) ON DELETE CASCADE @@ -293,7 +286,8 @@ CREATE TABLE files( note_folder_id INTEGER DEFAULT NULL REFERENCES note_folders ON DELETE CASCADE, redirect_file_id INTEGER REFERENCES files ON DELETE CASCADE, shared_msg_id BLOB, - file_type TEXT NOT NULL DEFAULT 'normal' + file_type TEXT NOT NULL DEFAULT 'normal', + roster_transfer_id INTEGER ) STRICT; CREATE TABLE snd_files( file_id INTEGER NOT NULL REFERENCES files ON DELETE CASCADE, @@ -813,6 +807,20 @@ CREATE TABLE group_relays( , base_web_url TEXT ) STRICT; +CREATE TABLE rcv_roster_transfers( + roster_transfer_id INTEGER PRIMARY KEY, + group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE, + from_member_id INTEGER NOT NULL REFERENCES group_members ON DELETE CASCADE, + roster_version INTEGER NOT NULL, + roster_digest BLOB NOT NULL, + sending_owner_gm_id INTEGER NOT NULL, + broker_ts TEXT NOT NULL, + roster_msg_body BLOB, + roster_msg_chat_binding TEXT, + roster_msg_signatures BLOB, + created_at TEXT NOT NULL DEFAULT(datetime('now')), + updated_at TEXT NOT NULL DEFAULT(datetime('now')) +) STRICT; CREATE INDEX contact_profiles_index ON contact_profiles( display_name, full_name @@ -1332,10 +1340,18 @@ ON groups( relay_request_group_link ) WHERE relay_request_group_link IS NOT NULL; +CREATE UNIQUE INDEX idx_rcv_roster_transfers_group_id_from_member_id ON rcv_roster_transfers( + group_id, + from_member_id +); +CREATE INDEX idx_rcv_roster_transfers_from_member_id ON rcv_roster_transfers( + from_member_id +); CREATE INDEX idx_files_group_id_shared_msg_id ON files( group_id, shared_msg_id ); +CREATE INDEX idx_files_roster_transfer_id ON files(roster_transfer_id); CREATE TRIGGER on_group_members_insert_update_summary AFTER INSERT ON group_members FOR EACH ROW diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index d2f23e2b89..26c449fed9 100644 --- a/tests/ChatTests/Groups.hs +++ b/tests/ChatTests/Groups.hs @@ -298,6 +298,7 @@ chatGroupTests = do it "promoted member enters the roster and can post" testChannelPromotedMemberCanPost it "observer cannot post until promoted" testChannelObserverCannotPost it "promoted member re-connecting via a new relay is accepted via the roster-pinned key" testChannelPromotedMemberRejoinViaRelay + it "2 relays: multi-chunk roster reassembles per source (no stream interleaving)" testChannelRosterMultiRelayMultipart describe "channel message operations" $ do it "should update channel message" testChannelMessageUpdate it "should delete channel message" testChannelMessageDelete @@ -8746,7 +8747,7 @@ prepareChannel2Relays gName owner relay1 relay2 = do owner <## ("#" <> gName <> ": group link relays updated, current relays:") owner <### [ EndsWith ": active", - Predicate (\l -> ": invited" `isSuffixOf` l || ": accepted" `isSuffixOf` l) + Predicate (\l -> ": invited" `isSuffixOf` l || ": accepted" `isSuffixOf` l || ": acknowledged_roster" `isSuffixOf` l) ] owner <## "group link:" void $ getTermLine owner -- consume group link line @@ -10539,7 +10540,7 @@ testChannelRosterDigestMismatchRejected ps = -- frank joins; bob re-serves the valid header with the corrupted blob, frank rejects it threadDelay 100000 memberJoinChannel "team" [bob] [alice, cath] shortLink fullLink frank - threadDelay 100000 + threadDelay 1000000 -- the rejected roster never elevates cath: the intro caps her to the channel default, so she -- stays observer (not moderator), and the version must not advance to the corrupted roster's version 1 checkMemberRow frank "cath" (Just "observer") @@ -10640,6 +10641,42 @@ testChannelPromotedMemberRejoinViaRelay ps = threadDelay 100000 checkMemberRow dan "cath" (Just "member") +testChannelRosterMultiRelayMultipart :: HasCallStack => TestParams -> IO () +testChannelRosterMultiRelayMultipart ps = + withNewTestChatCfgOpts ps cfg testOpts "alice" aliceProfile $ \alice -> + withNewTestChatCfgOpts ps cfg relayTestOpts "bob" bobProfile $ \bob -> + withNewTestChatCfgOpts ps cfg relayTestOpts "cath" cathProfile $ \cath -> + withNewTestChatCfgOpts ps cfg testOpts "dan" danProfile $ \dan -> + withNewTestChatCfgOpts ps cfg testOpts "eve" eveProfile $ \eve -> + withNewTestChatCfgOpts ps cfg testOpts "frank" frankProfile $ \frank -> do + createChannel2Relays "team" alice bob cath dan eve frank + + -- promote eve to moderator: the owner-signed roster broadcasts through BOTH relays to dan and + -- frank (each connected to both). At fileChunkSize=30 the blob spans multiple chunks, so each + -- member receives two interleaved multi-chunk streams (one per relay) for the same roster. + threadDelay 1000000 + alice ##> "/mr #team eve moderator" + alice <## "#team: you changed the role of eve to moderator (signed)" + concurrentlyN_ + [ bob <## "#team: alice changed the role of eve from observer to moderator (signed)", + cath <## "#team: alice changed the role of eve from observer to moderator (signed)", + eve <## "#team: alice changed your role from observer to moderator (signed)", + dan <### [EndsWith "to moderator (signed)"], + frank <### [EndsWith "to moderator (signed)"] + ] + threadDelay 1000000 -- let both relays' interleaved multipart streams settle + + -- per-source transfers keep the streams independent, so each member reassembles the blob and pins + -- eve as the single moderator WITH her owner-attested key (role + key both come from the blob) + checkOneModeratorWithKey dan + checkOneModeratorWithKey frank + where + cfg = testCfg {fileChunkSize = 30} + checkOneModeratorWithKey cc = do + rows <- withCCTransaction cc $ \db -> + DB.query_ db "SELECT member_pub_key FROM group_members WHERE member_role = 'moderator'" :: IO [Only (Maybe ByteString)] + map (\(Only k) -> isJust k) rows `shouldBe` [True] + testChannelRemoveRelay :: HasCallStack => TestParams -> IO () testChannelRemoveRelay ps = withNewTestChat ps "alice" aliceProfile $ \alice ->