mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-26 13:25:20 +00:00
Cut private message path over to ConversationStore
All private-message mutations now flow through store intents: coordinators, PrivateChatManager (its @Published dicts deleted - now read-only views over the store), outbound sends, delivery status, and chat migration. The O(1) store dedup replaces the full-scan duplicate check; insertion order is maintained by the store so sanitizeChat's re-sort is a documented no-op. Both bootstrapper Combine bridges and the Task.yield store synchronization are deleted. ChatViewModel.privateChats/unreadPrivateMessages become get-only derived views (measured: naive rebuild equals a change-invalidated cache within noise, so the simpler form stays). Feature models still read the legacy store, fed by a coalescing LegacyConversationStoreBridge (one mirror per burst, marked for step-5 deletion). pipeline.privateIngest: 9.6k -> 14.7k msg/s (+53%). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
@@ -11,11 +11,14 @@ import BitFoundation
|
||||
import Foundation
|
||||
import SwiftUI
|
||||
|
||||
/// Manages all private chat functionality
|
||||
/// Manages private chat session policy (selection, read receipts,
|
||||
/// consolidation). Message storage lives in the single-writer
|
||||
/// `ConversationStore` (docs/CONVERSATION-STORE-DESIGN.md); the
|
||||
/// `privateChats` / `unreadMessages` properties below are read-only compat
|
||||
/// views derived from it (migration step 2 — the manager shrinks to
|
||||
/// read-receipt policy in step 5).
|
||||
final class PrivateChatManager: ObservableObject {
|
||||
@Published var privateChats: [PeerID: [BitchatMessage]] = [:]
|
||||
@Published var selectedPeer: PeerID? = nil
|
||||
@Published var unreadMessages: Set<PeerID> = []
|
||||
|
||||
private var selectedPeerFingerprint: String? = nil
|
||||
var sentReadReceipts: Set<String> = [] // Made accessible for ChatViewModel
|
||||
@@ -25,13 +28,34 @@ final class PrivateChatManager: ObservableObject {
|
||||
weak var messageRouter: MessageRouter?
|
||||
// Peer service for looking up peer info during consolidation
|
||||
weak var unifiedPeerService: UnifiedPeerService?
|
||||
/// Single source of truth for message state; injected by the
|
||||
/// bootstrapper (`wireServiceGraph`).
|
||||
var conversationStore: ConversationStore?
|
||||
|
||||
init(meshService: Transport? = nil) {
|
||||
init(meshService: Transport? = nil, conversationStore: ConversationStore? = nil) {
|
||||
self.meshService = meshService
|
||||
self.conversationStore = conversationStore
|
||||
}
|
||||
|
||||
// Cap for messages stored per private chat
|
||||
private let privateChatCap = TransportConfig.privateChatCap
|
||||
// MARK: - Derived message state (read-only compat views)
|
||||
|
||||
/// All private chats keyed by routing peer ID, derived from the store.
|
||||
/// Mutations go through the store's intent API only.
|
||||
@MainActor
|
||||
var privateChats: [PeerID: [BitchatMessage]] {
|
||||
conversationStore?.directMessagesByRoutingPeerID() ?? [:]
|
||||
}
|
||||
|
||||
/// Unread chats, derived from the store's unread state.
|
||||
@MainActor
|
||||
var unreadMessages: Set<PeerID> {
|
||||
conversationStore?.unreadDirectRoutingPeerIDs() ?? []
|
||||
}
|
||||
|
||||
@MainActor
|
||||
private func messages(for peerID: PeerID) -> [BitchatMessage] {
|
||||
conversationStore?.conversationsByID[.directPeer(peerID)]?.messages ?? []
|
||||
}
|
||||
|
||||
// MARK: - Message Consolidation
|
||||
|
||||
@@ -44,57 +68,51 @@ final class PrivateChatManager: ObservableObject {
|
||||
/// - Returns: True if any unread messages were found during consolidation
|
||||
@MainActor
|
||||
func consolidateMessages(for peerID: PeerID, peerNickname: String, persistedReadReceipts: Set<String>) -> Bool {
|
||||
guard let meshService = meshService else { return false }
|
||||
guard let meshService = meshService, let store = conversationStore else { return false }
|
||||
var hasUnreadMessages = false
|
||||
|
||||
// 1. Consolidate from stable Noise key (64-char hex)
|
||||
if let peer = unifiedPeerService?.getPeer(by: peerID) {
|
||||
let noiseKeyHex = PeerID(hexData: peer.noisePublicKey)
|
||||
let nostrMessages = messages(for: noiseKeyHex)
|
||||
|
||||
if noiseKeyHex != peerID, let nostrMessages = privateChats[noiseKeyHex], !nostrMessages.isEmpty {
|
||||
if privateChats[peerID] == nil {
|
||||
privateChats[peerID] = []
|
||||
}
|
||||
|
||||
let existingMessageIds = Set(privateChats[peerID]?.map { $0.id } ?? [])
|
||||
if noiseKeyHex != peerID, !nostrMessages.isEmpty {
|
||||
for message in nostrMessages {
|
||||
if !existingMessageIds.contains(message.id) {
|
||||
// Update senderPeerID for correct read receipts
|
||||
let updatedMessage = BitchatMessage(
|
||||
id: message.id,
|
||||
sender: message.sender,
|
||||
content: message.content,
|
||||
timestamp: message.timestamp,
|
||||
isRelay: message.isRelay,
|
||||
originalSender: message.originalSender,
|
||||
isPrivate: message.isPrivate,
|
||||
recipientNickname: message.recipientNickname,
|
||||
senderPeerID: message.senderPeerID == meshService.myPeerID ? meshService.myPeerID : peerID,
|
||||
mentions: message.mentions,
|
||||
deliveryStatus: message.deliveryStatus
|
||||
)
|
||||
privateChats[peerID]?.append(updatedMessage)
|
||||
// Update senderPeerID for correct read receipts
|
||||
let updatedMessage = BitchatMessage(
|
||||
id: message.id,
|
||||
sender: message.sender,
|
||||
content: message.content,
|
||||
timestamp: message.timestamp,
|
||||
isRelay: message.isRelay,
|
||||
originalSender: message.originalSender,
|
||||
isPrivate: message.isPrivate,
|
||||
recipientNickname: message.recipientNickname,
|
||||
senderPeerID: message.senderPeerID == meshService.myPeerID ? meshService.myPeerID : peerID,
|
||||
mentions: message.mentions,
|
||||
deliveryStatus: message.deliveryStatus
|
||||
)
|
||||
// Store append dedups by message ID (skips ones the
|
||||
// target chat already has).
|
||||
guard store.append(updatedMessage, to: .directPeer(peerID)) else { continue }
|
||||
|
||||
// Check for recent unread messages (< 60s, not sent by us, not already read)
|
||||
// Use persistedReadReceipts to correctly identify already-read messages after app restart
|
||||
if message.senderPeerID != meshService.myPeerID {
|
||||
let messageAge = Date().timeIntervalSince(message.timestamp)
|
||||
if messageAge < 60 && !persistedReadReceipts.contains(message.id) {
|
||||
hasUnreadMessages = true
|
||||
}
|
||||
// Check for recent unread messages (< 60s, not sent by us, not already read)
|
||||
// Use persistedReadReceipts to correctly identify already-read messages after app restart
|
||||
if message.senderPeerID != meshService.myPeerID {
|
||||
let messageAge = Date().timeIntervalSince(message.timestamp)
|
||||
if messageAge < 60 && !persistedReadReceipts.contains(message.id) {
|
||||
hasUnreadMessages = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
privateChats[peerID]?.sort { $0.timestamp < $1.timestamp }
|
||||
|
||||
if hasUnreadMessages {
|
||||
unreadMessages.insert(peerID)
|
||||
} else if unreadMessages.contains(noiseKeyHex) {
|
||||
unreadMessages.remove(noiseKeyHex)
|
||||
store.markUnread(.directPeer(peerID))
|
||||
} else {
|
||||
store.markRead(.directPeer(noiseKeyHex))
|
||||
}
|
||||
|
||||
privateChats.removeValue(forKey: noiseKeyHex)
|
||||
store.removeConversation(.directPeer(noiseKeyHex))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -112,52 +130,43 @@ final class PrivateChatManager: ObservableObject {
|
||||
}
|
||||
|
||||
if !tempPeerIDsToConsolidate.isEmpty {
|
||||
if privateChats[peerID] == nil {
|
||||
privateChats[peerID] = []
|
||||
}
|
||||
|
||||
let existingMessageIds = Set(privateChats[peerID]?.map { $0.id } ?? [])
|
||||
var consolidatedCount = 0
|
||||
var hadUnreadTemp = false
|
||||
let unreadPeerIDs = unreadMessages
|
||||
|
||||
for tempPeerID in tempPeerIDsToConsolidate {
|
||||
if unreadMessages.contains(tempPeerID) {
|
||||
if unreadPeerIDs.contains(tempPeerID) {
|
||||
hadUnreadTemp = true
|
||||
}
|
||||
|
||||
if let tempMessages = privateChats[tempPeerID] {
|
||||
for message in tempMessages {
|
||||
if !existingMessageIds.contains(message.id) {
|
||||
let updatedMessage = BitchatMessage(
|
||||
id: message.id,
|
||||
sender: message.sender,
|
||||
content: message.content,
|
||||
timestamp: message.timestamp,
|
||||
isRelay: message.isRelay,
|
||||
originalSender: message.originalSender,
|
||||
isPrivate: message.isPrivate,
|
||||
recipientNickname: message.recipientNickname,
|
||||
senderPeerID: peerID,
|
||||
mentions: message.mentions,
|
||||
deliveryStatus: message.deliveryStatus
|
||||
)
|
||||
privateChats[peerID]?.append(updatedMessage)
|
||||
consolidatedCount += 1
|
||||
}
|
||||
for message in messages(for: tempPeerID) {
|
||||
let updatedMessage = BitchatMessage(
|
||||
id: message.id,
|
||||
sender: message.sender,
|
||||
content: message.content,
|
||||
timestamp: message.timestamp,
|
||||
isRelay: message.isRelay,
|
||||
originalSender: message.originalSender,
|
||||
isPrivate: message.isPrivate,
|
||||
recipientNickname: message.recipientNickname,
|
||||
senderPeerID: peerID,
|
||||
mentions: message.mentions,
|
||||
deliveryStatus: message.deliveryStatus
|
||||
)
|
||||
if store.append(updatedMessage, to: .directPeer(peerID)) {
|
||||
consolidatedCount += 1
|
||||
}
|
||||
privateChats.removeValue(forKey: tempPeerID)
|
||||
unreadMessages.remove(tempPeerID)
|
||||
}
|
||||
store.removeConversation(.directPeer(tempPeerID))
|
||||
}
|
||||
|
||||
if hadUnreadTemp {
|
||||
unreadMessages.insert(peerID)
|
||||
store.markUnread(.directPeer(peerID))
|
||||
hasUnreadMessages = true
|
||||
SecureLogger.debug("📬 Transferred unread status from temp peer IDs to \(peerID)", category: .session)
|
||||
}
|
||||
|
||||
if consolidatedCount > 0 {
|
||||
privateChats[peerID]?.sort { $0.timestamp < $1.timestamp }
|
||||
SecureLogger.info("📥 Consolidated \(consolidatedCount) Nostr messages from temporary peer IDs to \(peerNickname)", category: .session)
|
||||
}
|
||||
}
|
||||
@@ -168,9 +177,7 @@ final class PrivateChatManager: ObservableObject {
|
||||
/// Syncs the read receipt tracking between manager and view model for sent messages
|
||||
@MainActor
|
||||
func syncReadReceiptsForSentMessages(peerID: PeerID, nickname: String, externalReceipts: inout Set<String>) {
|
||||
guard let messages = privateChats[peerID] else { return }
|
||||
|
||||
for message in messages {
|
||||
for message in messages(for: peerID) {
|
||||
if message.sender == nickname {
|
||||
if let status = message.deliveryStatus {
|
||||
switch status {
|
||||
@@ -184,86 +191,66 @@ final class PrivateChatManager: ObservableObject {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/// Start a private chat with a peer
|
||||
@MainActor
|
||||
func startChat(with peerID: PeerID) {
|
||||
selectedPeer = peerID
|
||||
|
||||
|
||||
// Store fingerprint for persistence across reconnections
|
||||
if let fingerprint = meshService?.getFingerprint(for: peerID) {
|
||||
selectedPeerFingerprint = fingerprint
|
||||
}
|
||||
|
||||
|
||||
// Mark messages as read
|
||||
markAsRead(from: peerID)
|
||||
|
||||
|
||||
// Initialize chat if needed
|
||||
if privateChats[peerID] == nil {
|
||||
privateChats[peerID] = []
|
||||
}
|
||||
conversationStore?.conversation(for: .directPeer(peerID))
|
||||
}
|
||||
|
||||
|
||||
/// End the current private chat
|
||||
func endChat() {
|
||||
selectedPeer = nil
|
||||
selectedPeerFingerprint = nil
|
||||
}
|
||||
|
||||
/// Remove duplicate messages by ID and keep chronological order
|
||||
func sanitizeChat(for peerID: PeerID) {
|
||||
guard let arr = privateChats[peerID] else { return }
|
||||
if arr.count <= 1 {
|
||||
return
|
||||
}
|
||||
/// No-op since the `ConversationStore` cutover: the store maintains
|
||||
/// chronological order and dedups by message ID on every insert, so the
|
||||
/// per-append re-sort/dedup sweep this performed is no longer needed.
|
||||
/// Kept only for API compatibility until step 5 removes the callers.
|
||||
func sanitizeChat(for peerID: PeerID) {}
|
||||
|
||||
var indexByID: [String: Int] = [:]
|
||||
indexByID.reserveCapacity(arr.count)
|
||||
var deduped: [BitchatMessage] = []
|
||||
deduped.reserveCapacity(arr.count)
|
||||
|
||||
for msg in arr.sorted(by: { $0.timestamp < $1.timestamp }) {
|
||||
if let existing = indexByID[msg.id] {
|
||||
deduped[existing] = msg
|
||||
} else {
|
||||
indexByID[msg.id] = deduped.count
|
||||
deduped.append(msg)
|
||||
}
|
||||
}
|
||||
|
||||
privateChats[peerID] = deduped
|
||||
}
|
||||
|
||||
/// Mark messages from a peer as read
|
||||
@MainActor
|
||||
func markAsRead(from peerID: PeerID) {
|
||||
unreadMessages.remove(peerID)
|
||||
|
||||
conversationStore?.markRead(.directPeer(peerID))
|
||||
|
||||
// Send read receipts for unread messages that haven't been sent yet
|
||||
if let messages = privateChats[peerID] {
|
||||
for message in messages {
|
||||
if message.senderPeerID == peerID && !message.isRelay && !sentReadReceipts.contains(message.id) {
|
||||
sendReadReceipt(for: message)
|
||||
}
|
||||
for message in messages(for: peerID) {
|
||||
if message.senderPeerID == peerID && !message.isRelay && !sentReadReceipts.contains(message.id) {
|
||||
sendReadReceipt(for: message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// MARK: - Private Methods
|
||||
|
||||
|
||||
private func sendReadReceipt(for message: BitchatMessage) {
|
||||
guard !sentReadReceipts.contains(message.id),
|
||||
let senderPeerID = message.senderPeerID else {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
sentReadReceipts.insert(message.id)
|
||||
|
||||
|
||||
// Create read receipt using the simplified method
|
||||
let receipt = ReadReceipt(
|
||||
originalMessageID: message.id,
|
||||
readerID: meshService?.myPeerID ?? PeerID(str: ""),
|
||||
readerNickname: meshService?.myNickname ?? ""
|
||||
)
|
||||
|
||||
|
||||
// Route via MessageRouter to avoid handshakeRequired spam when session isn't established
|
||||
if let router = messageRouter {
|
||||
SecureLogger.debug("PrivateChatManager: sending READ ack for \(message.id.prefix(8))… to \(senderPeerID.id.prefix(8))… via router", category: .session)
|
||||
|
||||
Reference in New Issue
Block a user