mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-27 10:05:20 +00:00
Route delivery status through the store; delete the location index
ConversationStore maintains an exact messageID -> Set<ConversationID> map at every mutation point (append/upsert/remove/migrate/trim/clear), so delivery updates are ID-only lookups that fan out to mirrored ephemeral/stable copies. ChatDeliveryCoordinator shrinks 327 -> 119 lines: the positional location index, its growth-detection/rebuild machinery, and the duplicate no-downgrade check are deleted - the rule now lives in exactly one place. The middle-insertion regression tests are rewritten against the store since stale positional locations are structurally impossible now. delivery updates: 38k -> 262k/s (~6.9x); ingest pipelines unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
@@ -12,18 +12,19 @@ import Foundation
|
||||
/// coordinators off their `unowned let viewModel: ChatViewModel` back-refs.
|
||||
@MainActor
|
||||
protocol ChatDeliveryContext: AnyObject {
|
||||
/// Get-only derived views of the `ConversationStore`. `BitchatMessage`
|
||||
/// is a reference type, so the public-timeline status patch below writes
|
||||
/// through the shared message objects; step 4 replaces this with
|
||||
/// `setDeliveryStatus(for:in:)` store intents.
|
||||
var messages: [BitchatMessage] { get }
|
||||
var privateChats: [PeerID: [BitchatMessage]] { get }
|
||||
var isStartupPhase: Bool { get }
|
||||
/// Applies a delivery status to a private message by ID (single-writer
|
||||
/// store intent; full delivery migration is step 4). Returns `false`
|
||||
/// when the message is unknown or the update would downgrade the status.
|
||||
/// Applies a delivery status to every copy of the message across
|
||||
/// conversations (`ConversationStore` intent, ID-only: the store's
|
||||
/// message-ID → conversation map resolves which conversations hold the
|
||||
/// message, including mirrored ephemeral/stable private copies). The
|
||||
/// no-downgrade rule is enforced in the store. Returns `false` when the
|
||||
/// message is unknown or no copy changed.
|
||||
@discardableResult
|
||||
func setPrivateDeliveryStatus(_ status: DeliveryStatus, forMessageID messageID: String, peerID: PeerID) -> Bool
|
||||
func setDeliveryStatus(_ status: DeliveryStatus, forMessageID messageID: String) -> Bool
|
||||
/// Current delivery status of the message in whichever conversation holds it.
|
||||
func deliveryStatus(forMessageID messageID: String) -> DeliveryStatus?
|
||||
/// Message IDs across all direct conversations (read-receipt pruning).
|
||||
func privateMessageIDs() -> Set<String>
|
||||
/// Drops every recorded read receipt whose message ID is not in `validMessageIDs`.
|
||||
/// Returns the number of receipts removed. (Single mutation path for the
|
||||
/// owner's `sentReadReceipts`; this coordinator never reads the raw set.)
|
||||
@@ -35,6 +36,19 @@ protocol ChatDeliveryContext: AnyObject {
|
||||
}
|
||||
|
||||
extension ChatViewModel: ChatDeliveryContext {
|
||||
@discardableResult
|
||||
func setDeliveryStatus(_ status: DeliveryStatus, forMessageID messageID: String) -> Bool {
|
||||
conversations.setDeliveryStatus(status, forMessageID: messageID)
|
||||
}
|
||||
|
||||
func deliveryStatus(forMessageID messageID: String) -> DeliveryStatus? {
|
||||
conversations.deliveryStatus(forMessageID: messageID)
|
||||
}
|
||||
|
||||
func privateMessageIDs() -> Set<String> {
|
||||
conversations.directMessageIDs()
|
||||
}
|
||||
|
||||
func notifyUIChanged() {
|
||||
objectWillChange.send()
|
||||
}
|
||||
@@ -44,14 +58,12 @@ extension ChatViewModel: ChatDeliveryContext {
|
||||
}
|
||||
}
|
||||
|
||||
/// Thin mapper from delivery events (read receipts, transport delivery
|
||||
/// callbacks) onto `ConversationStore` delivery intents, plus read-receipt
|
||||
/// retention cleanup. The store's message-ID → conversation map replaces the
|
||||
/// positional `messageLocationIndex` this coordinator used to maintain.
|
||||
final class ChatDeliveryCoordinator {
|
||||
private unowned let context: any ChatDeliveryContext
|
||||
private var messageLocationIndex: [String: Set<MessageLocation>] = [:]
|
||||
private var indexedPublicMessageCount = 0
|
||||
private var indexedPublicTailMessageID: String?
|
||||
private var indexedPrivateMessageCounts: [PeerID: Int] = [:]
|
||||
private var indexedPrivateTailMessageIDs: [PeerID: String] = [:]
|
||||
private var hasBuiltMessageLocationIndex = false
|
||||
|
||||
init(context: any ChatDeliveryContext) {
|
||||
self.context = context
|
||||
@@ -59,15 +71,9 @@ final class ChatDeliveryCoordinator {
|
||||
|
||||
@MainActor
|
||||
func cleanupOldReadReceipts() {
|
||||
guard !context.isStartupPhase, !context.privateChats.isEmpty else {
|
||||
return
|
||||
}
|
||||
|
||||
let validMessageIDs = Set(
|
||||
context.privateChats.values.flatMap { messages in
|
||||
messages.map(\.id)
|
||||
}
|
||||
)
|
||||
guard !context.isStartupPhase else { return }
|
||||
let validMessageIDs = context.privateMessageIDs()
|
||||
guard !validMessageIDs.isEmpty else { return }
|
||||
|
||||
let removedCount = context.pruneSentReadReceipts(keeping: validMessageIDs)
|
||||
if removedCount > 0 {
|
||||
@@ -90,9 +96,7 @@ final class ChatDeliveryCoordinator {
|
||||
|
||||
@MainActor
|
||||
func deliveryStatus(for messageID: String) -> DeliveryStatus? {
|
||||
withValidLocations(for: messageID) { locations in
|
||||
locations.lazy.compactMap { self.deliveryStatus(at: $0) }.first
|
||||
}
|
||||
context.deliveryStatus(forMessageID: messageID)
|
||||
}
|
||||
|
||||
@MainActor
|
||||
@@ -106,222 +110,10 @@ final class ChatDeliveryCoordinator {
|
||||
break
|
||||
}
|
||||
|
||||
var didUpdateStatus = false
|
||||
let locations = withValidLocations(for: messageID) { $0 }
|
||||
guard !locations.isEmpty else { return false }
|
||||
|
||||
for location in locations {
|
||||
guard case .publicTimeline(let index) = location,
|
||||
index < context.messages.count,
|
||||
context.messages[index].id == messageID else {
|
||||
continue
|
||||
}
|
||||
|
||||
let currentStatus = context.messages[index].deliveryStatus
|
||||
if !shouldSkipUpdate(currentStatus: currentStatus, newStatus: status) {
|
||||
context.messages[index].deliveryStatus = status
|
||||
didUpdateStatus = true
|
||||
}
|
||||
}
|
||||
|
||||
for location in locations {
|
||||
guard case .privateChat(let peerID, let index) = location,
|
||||
let chatMessages = context.privateChats[peerID],
|
||||
index < chatMessages.count,
|
||||
chatMessages[index].id == messageID else {
|
||||
continue
|
||||
}
|
||||
|
||||
let currentStatus = chatMessages[index].deliveryStatus
|
||||
guard !shouldSkipUpdate(currentStatus: currentStatus, newStatus: status) else { continue }
|
||||
|
||||
if context.setPrivateDeliveryStatus(status, forMessageID: messageID, peerID: peerID) {
|
||||
didUpdateStatus = true
|
||||
}
|
||||
}
|
||||
|
||||
if didUpdateStatus {
|
||||
context.notifyUIChanged()
|
||||
}
|
||||
|
||||
return didUpdateStatus
|
||||
}
|
||||
}
|
||||
|
||||
private extension ChatDeliveryCoordinator {
|
||||
enum MessageLocation: Hashable {
|
||||
case publicTimeline(index: Int)
|
||||
case privateChat(peerID: PeerID, index: Int)
|
||||
}
|
||||
|
||||
func shouldSkipUpdate(currentStatus: DeliveryStatus?, newStatus: DeliveryStatus) -> Bool {
|
||||
guard let currentStatus else { return false }
|
||||
if currentStatus == newStatus { return true }
|
||||
|
||||
switch (currentStatus, newStatus) {
|
||||
case (.read, .delivered), (.read, .sent):
|
||||
return true
|
||||
default:
|
||||
guard context.setDeliveryStatus(status, forMessageID: messageID) else {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
func withValidLocations<T>(
|
||||
for messageID: String,
|
||||
_ body: (Set<MessageLocation>) -> T
|
||||
) -> T {
|
||||
let didRebuildIndex = refreshMessageLocationIndexForGrowth()
|
||||
|
||||
if let locations = messageLocationIndex[messageID],
|
||||
locations.allSatisfy({ isLocation($0, validFor: messageID) }) {
|
||||
return body(locations)
|
||||
}
|
||||
|
||||
guard !didRebuildIndex else {
|
||||
return body(messageLocationIndex[messageID] ?? [])
|
||||
}
|
||||
|
||||
if messageLocationIndex[messageID] == nil {
|
||||
return body([])
|
||||
}
|
||||
|
||||
rebuildMessageLocationIndex()
|
||||
return body(messageLocationIndex[messageID] ?? [])
|
||||
}
|
||||
|
||||
@MainActor
|
||||
func deliveryStatus(at location: MessageLocation) -> DeliveryStatus? {
|
||||
switch location {
|
||||
case .publicTimeline(let index):
|
||||
guard index < context.messages.count else { return nil }
|
||||
return context.messages[index].deliveryStatus
|
||||
case .privateChat(let peerID, let index):
|
||||
guard let messages = context.privateChats[peerID],
|
||||
index < messages.count else {
|
||||
return nil
|
||||
}
|
||||
return messages[index].deliveryStatus
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
func isLocation(_ location: MessageLocation, validFor messageID: String) -> Bool {
|
||||
switch location {
|
||||
case .publicTimeline(let index):
|
||||
return index < context.messages.count
|
||||
&& context.messages[index].id == messageID
|
||||
case .privateChat(let peerID, let index):
|
||||
guard let messages = context.privateChats[peerID],
|
||||
index < messages.count else {
|
||||
return false
|
||||
}
|
||||
return messages[index].id == messageID
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
@discardableResult
|
||||
func refreshMessageLocationIndexForGrowth() -> Bool {
|
||||
guard hasBuiltMessageLocationIndex else {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
|
||||
if context.messages.count < indexedPublicMessageCount {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
|
||||
if context.messages.count == indexedPublicMessageCount,
|
||||
context.messages.last?.id != indexedPublicTailMessageID {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
|
||||
if context.messages.count > indexedPublicMessageCount {
|
||||
// Growth is only a pure append if the previously indexed tail kept
|
||||
// its position; a middle insertion (out-of-order timestamp arrival)
|
||||
// shifts it and invalidates every indexed location after the
|
||||
// insertion point.
|
||||
if indexedPublicMessageCount > 0,
|
||||
context.messages[indexedPublicMessageCount - 1].id != indexedPublicTailMessageID {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
for index in indexedPublicMessageCount..<context.messages.count {
|
||||
add(.publicTimeline(index: index), for: context.messages[index].id)
|
||||
}
|
||||
indexedPublicMessageCount = context.messages.count
|
||||
indexedPublicTailMessageID = context.messages.last?.id
|
||||
}
|
||||
|
||||
let currentPeerIDs = Set(context.privateChats.keys)
|
||||
if !Set(indexedPrivateMessageCounts.keys).isSubset(of: currentPeerIDs) {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
|
||||
for (peerID, messages) in context.privateChats {
|
||||
let indexedCount = indexedPrivateMessageCounts[peerID] ?? 0
|
||||
if messages.count < indexedCount {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
|
||||
if messages.count == indexedCount,
|
||||
messages.last?.id != indexedPrivateTailMessageIDs[peerID] {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
|
||||
guard messages.count > indexedCount else { continue }
|
||||
// Same append-only check as the public timeline above.
|
||||
if indexedCount > 0,
|
||||
messages[indexedCount - 1].id != indexedPrivateTailMessageIDs[peerID] {
|
||||
rebuildMessageLocationIndex()
|
||||
return true
|
||||
}
|
||||
for index in indexedCount..<messages.count {
|
||||
add(.privateChat(peerID: peerID, index: index), for: messages[index].id)
|
||||
}
|
||||
indexedPrivateMessageCounts[peerID] = messages.count
|
||||
if let tailID = messages.last?.id {
|
||||
indexedPrivateTailMessageIDs[peerID] = tailID
|
||||
} else {
|
||||
indexedPrivateTailMessageIDs.removeValue(forKey: peerID)
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
@MainActor
|
||||
func rebuildMessageLocationIndex() {
|
||||
messageLocationIndex.removeAll(keepingCapacity: true)
|
||||
|
||||
for (index, message) in context.messages.enumerated() {
|
||||
add(.publicTimeline(index: index), for: message.id)
|
||||
}
|
||||
indexedPublicMessageCount = context.messages.count
|
||||
indexedPublicTailMessageID = context.messages.last?.id
|
||||
|
||||
indexedPrivateMessageCounts.removeAll(keepingCapacity: true)
|
||||
indexedPrivateTailMessageIDs.removeAll(keepingCapacity: true)
|
||||
for (peerID, messages) in context.privateChats {
|
||||
for (index, message) in messages.enumerated() {
|
||||
add(.privateChat(peerID: peerID, index: index), for: message.id)
|
||||
}
|
||||
indexedPrivateMessageCounts[peerID] = messages.count
|
||||
if let tailID = messages.last?.id {
|
||||
indexedPrivateTailMessageIDs[peerID] = tailID
|
||||
}
|
||||
}
|
||||
|
||||
hasBuiltMessageLocationIndex = true
|
||||
}
|
||||
|
||||
func add(_ location: MessageLocation, for messageID: String) {
|
||||
messageLocationIndex[messageID, default: []].insert(location)
|
||||
context.notifyUIChanged()
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user