mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-26 02:45:19 +00:00
fix: ensure messages maintain correct order during sync
- Sort retry queue by original timestamp when adding and processing messages - Add 50ms delay between retry messages to prevent out-of-order delivery - Messages now sync in the exact order they were sent (ping 1, ping 2, ping 3)
This commit is contained in:
Nelson Campos
@@ -87,6 +87,13 @@ class MessageRetryService {
|
|||||||
)
|
)
|
||||||
|
|
||||||
retryQueue.append(retryMessage)
|
retryQueue.append(retryMessage)
|
||||||
|
|
||||||
|
// Sort the queue by original timestamp to maintain message order
|
||||||
|
retryQueue.sort { (msg1, msg2) in
|
||||||
|
let time1 = msg1.originalTimestamp ?? Date.distantPast
|
||||||
|
let time2 = msg2.originalTimestamp ?? Date.distantPast
|
||||||
|
return time1 < time2
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private func processRetryQueue() {
|
private func processRetryQueue() {
|
||||||
@@ -106,107 +113,123 @@ class MessageRetryService {
|
|||||||
|
|
||||||
retryQueue = updatedQueue
|
retryQueue = updatedQueue
|
||||||
|
|
||||||
for message in messagesToRetry {
|
// Sort messages by original timestamp to maintain order
|
||||||
|
messagesToRetry.sort { (msg1, msg2) in
|
||||||
|
let time1 = msg1.originalTimestamp ?? Date.distantPast
|
||||||
|
let time2 = msg2.originalTimestamp ?? Date.distantPast
|
||||||
|
return time1 < time2
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send messages with delay to maintain order
|
||||||
|
for (index, message) in messagesToRetry.enumerated() {
|
||||||
// Check if we should still retry
|
// Check if we should still retry
|
||||||
if message.retryCount >= message.maxRetries {
|
if message.retryCount >= message.maxRetries {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check connectivity before retrying
|
// Add delay between messages to ensure proper ordering
|
||||||
let viewModel = meshService.delegate as? ChatViewModel
|
let delay = Double(index) * 0.05 // 50ms between messages
|
||||||
let connectedPeers = viewModel?.connectedPeers ?? []
|
|
||||||
|
|
||||||
if message.isPrivate {
|
DispatchQueue.main.asyncAfter(deadline: .now() + delay) { [weak self] in
|
||||||
// For private messages, check if recipient is connected
|
guard let self = self,
|
||||||
if let recipientID = message.recipientPeerID,
|
let meshService = self.meshService else { return }
|
||||||
connectedPeers.contains(recipientID) {
|
|
||||||
// Retry private message
|
// Check connectivity before retrying
|
||||||
meshService.sendPrivateMessage(
|
let viewModel = meshService.delegate as? ChatViewModel
|
||||||
message.content,
|
let connectedPeers = viewModel?.connectedPeers ?? []
|
||||||
to: recipientID,
|
|
||||||
recipientNickname: message.recipientNickname ?? "unknown",
|
if message.isPrivate {
|
||||||
messageID: message.originalMessageID
|
// For private messages, check if recipient is connected
|
||||||
)
|
if let recipientID = message.recipientPeerID,
|
||||||
|
connectedPeers.contains(recipientID) {
|
||||||
|
// Retry private message
|
||||||
|
meshService.sendPrivateMessage(
|
||||||
|
message.content,
|
||||||
|
to: recipientID,
|
||||||
|
recipientNickname: message.recipientNickname ?? "unknown",
|
||||||
|
messageID: message.originalMessageID
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
// Recipient not connected, keep in queue with updated retry time
|
||||||
|
var updatedMessage = message
|
||||||
|
updatedMessage = RetryableMessage(
|
||||||
|
id: message.id,
|
||||||
|
originalMessageID: message.originalMessageID,
|
||||||
|
originalTimestamp: message.originalTimestamp,
|
||||||
|
content: message.content,
|
||||||
|
mentions: message.mentions,
|
||||||
|
channel: message.channel,
|
||||||
|
isPrivate: message.isPrivate,
|
||||||
|
recipientPeerID: message.recipientPeerID,
|
||||||
|
recipientNickname: message.recipientNickname,
|
||||||
|
channelKey: message.channelKey,
|
||||||
|
retryCount: message.retryCount + 1,
|
||||||
|
nextRetryTime: Date().addingTimeInterval(self.retryInterval * Double(message.retryCount + 2))
|
||||||
|
)
|
||||||
|
self.retryQueue.append(updatedMessage)
|
||||||
|
}
|
||||||
|
} else if let channel = message.channel, let channelKeyData = message.channelKey {
|
||||||
|
// For channel messages, check if we have peers in the channel
|
||||||
|
if !connectedPeers.isEmpty {
|
||||||
|
// Recreate SymmetricKey from data
|
||||||
|
let channelKey = SymmetricKey(data: channelKeyData)
|
||||||
|
meshService.sendEncryptedChannelMessage(
|
||||||
|
message.content,
|
||||||
|
mentions: message.mentions ?? [],
|
||||||
|
channel: channel,
|
||||||
|
channelKey: channelKey,
|
||||||
|
messageID: message.originalMessageID,
|
||||||
|
timestamp: message.originalTimestamp
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
// No peers connected, keep in queue
|
||||||
|
var updatedMessage = message
|
||||||
|
updatedMessage = RetryableMessage(
|
||||||
|
id: message.id,
|
||||||
|
originalMessageID: message.originalMessageID,
|
||||||
|
originalTimestamp: message.originalTimestamp,
|
||||||
|
content: message.content,
|
||||||
|
mentions: message.mentions,
|
||||||
|
channel: message.channel,
|
||||||
|
isPrivate: message.isPrivate,
|
||||||
|
recipientPeerID: message.recipientPeerID,
|
||||||
|
recipientNickname: message.recipientNickname,
|
||||||
|
channelKey: message.channelKey,
|
||||||
|
retryCount: message.retryCount + 1,
|
||||||
|
nextRetryTime: Date().addingTimeInterval(self.retryInterval * Double(message.retryCount + 2))
|
||||||
|
)
|
||||||
|
self.retryQueue.append(updatedMessage)
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
// Recipient not connected, keep in queue with updated retry time
|
// Regular message
|
||||||
var updatedMessage = message
|
if !connectedPeers.isEmpty {
|
||||||
updatedMessage = RetryableMessage(
|
meshService.sendMessage(
|
||||||
id: message.id,
|
message.content,
|
||||||
originalMessageID: message.originalMessageID,
|
mentions: message.mentions ?? [],
|
||||||
originalTimestamp: message.originalTimestamp,
|
channel: message.channel,
|
||||||
content: message.content,
|
to: nil,
|
||||||
mentions: message.mentions,
|
messageID: message.originalMessageID,
|
||||||
channel: message.channel,
|
timestamp: message.originalTimestamp
|
||||||
isPrivate: message.isPrivate,
|
)
|
||||||
recipientPeerID: message.recipientPeerID,
|
} else {
|
||||||
recipientNickname: message.recipientNickname,
|
// No peers connected, keep in queue
|
||||||
channelKey: message.channelKey,
|
var updatedMessage = message
|
||||||
retryCount: message.retryCount + 1,
|
updatedMessage = RetryableMessage(
|
||||||
nextRetryTime: Date().addingTimeInterval(retryInterval * Double(message.retryCount + 2))
|
id: message.id,
|
||||||
)
|
originalMessageID: message.originalMessageID,
|
||||||
retryQueue.append(updatedMessage)
|
originalTimestamp: message.originalTimestamp,
|
||||||
}
|
content: message.content,
|
||||||
} else if let channel = message.channel, let channelKeyData = message.channelKey {
|
mentions: message.mentions,
|
||||||
// For channel messages, check if we have peers in the channel
|
channel: message.channel,
|
||||||
if !connectedPeers.isEmpty {
|
isPrivate: message.isPrivate,
|
||||||
// Recreate SymmetricKey from data
|
recipientPeerID: message.recipientPeerID,
|
||||||
let channelKey = SymmetricKey(data: channelKeyData)
|
recipientNickname: message.recipientNickname,
|
||||||
meshService.sendEncryptedChannelMessage(
|
channelKey: message.channelKey,
|
||||||
message.content,
|
retryCount: message.retryCount + 1,
|
||||||
mentions: message.mentions ?? [],
|
nextRetryTime: Date().addingTimeInterval(self.retryInterval * Double(message.retryCount + 2))
|
||||||
channel: channel,
|
)
|
||||||
channelKey: channelKey,
|
self.retryQueue.append(updatedMessage)
|
||||||
messageID: message.originalMessageID,
|
}
|
||||||
timestamp: message.originalTimestamp
|
|
||||||
)
|
|
||||||
} else {
|
|
||||||
// No peers connected, keep in queue
|
|
||||||
var updatedMessage = message
|
|
||||||
updatedMessage = RetryableMessage(
|
|
||||||
id: message.id,
|
|
||||||
originalMessageID: message.originalMessageID,
|
|
||||||
originalTimestamp: message.originalTimestamp,
|
|
||||||
content: message.content,
|
|
||||||
mentions: message.mentions,
|
|
||||||
channel: message.channel,
|
|
||||||
isPrivate: message.isPrivate,
|
|
||||||
recipientPeerID: message.recipientPeerID,
|
|
||||||
recipientNickname: message.recipientNickname,
|
|
||||||
channelKey: message.channelKey,
|
|
||||||
retryCount: message.retryCount + 1,
|
|
||||||
nextRetryTime: Date().addingTimeInterval(retryInterval * Double(message.retryCount + 2))
|
|
||||||
)
|
|
||||||
retryQueue.append(updatedMessage)
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// Regular message
|
|
||||||
if !connectedPeers.isEmpty {
|
|
||||||
meshService.sendMessage(
|
|
||||||
message.content,
|
|
||||||
mentions: message.mentions ?? [],
|
|
||||||
channel: message.channel,
|
|
||||||
to: nil,
|
|
||||||
messageID: message.originalMessageID,
|
|
||||||
timestamp: message.originalTimestamp
|
|
||||||
)
|
|
||||||
} else {
|
|
||||||
// No peers connected, keep in queue
|
|
||||||
var updatedMessage = message
|
|
||||||
updatedMessage = RetryableMessage(
|
|
||||||
id: message.id,
|
|
||||||
originalMessageID: message.originalMessageID,
|
|
||||||
originalTimestamp: message.originalTimestamp,
|
|
||||||
content: message.content,
|
|
||||||
mentions: message.mentions,
|
|
||||||
channel: message.channel,
|
|
||||||
isPrivate: message.isPrivate,
|
|
||||||
recipientPeerID: message.recipientPeerID,
|
|
||||||
recipientNickname: message.recipientNickname,
|
|
||||||
channelKey: message.channelKey,
|
|
||||||
retryCount: message.retryCount + 1,
|
|
||||||
nextRetryTime: Date().addingTimeInterval(retryInterval * Double(message.retryCount + 2))
|
|
||||||
)
|
|
||||||
retryQueue.append(updatedMessage)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user