mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-25 04:45:20 +00:00
Cut public message path over to ConversationStore; delete PublicTimelineStore
Mesh and geohash timelines are now store conversations. All public mutation sites flow through store intents; PublicMessagePipeline keeps its 80ms UI batching but commits batches via store appends with each buffered entry carrying its destination conversation (a mid-batch channel switch now flushes instead of dropping the buffer). ChatViewModel.messages becomes a cached get-only view of the active conversation, invalidated through the change subject. The mesh late-insert threshold is consciously removed: it only ever ordered the non-rendered messages copy, so strict timestamp insertion makes the working set agree with rendered order. PublicTimelineStore and the per-message full-array legacy sync are deleted; the coalescing bridge mirrors public conversations for the remaining legacy readers. pipeline.publicIngest: 6.6k -> 9.5k msg/s (+45%); private steady; store.append 237k/s. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
@@ -52,23 +52,17 @@ private final class MockChatMediaTransferContext: ChatMediaTransferContext {
|
||||
return true
|
||||
}
|
||||
|
||||
var meshTimeline: [BitchatMessage] = []
|
||||
private(set) var refreshedChannels: [ChannelID?] = []
|
||||
private(set) var trimCount = 0
|
||||
private(set) var appendedPublicMessages: [(message: BitchatMessage, conversationID: ConversationID)] = []
|
||||
private(set) var removedMessages: [(messageID: String, cleanupFile: Bool)] = []
|
||||
private(set) var systemMessages: [String] = []
|
||||
private(set) var notifyUIChangedCount = 0
|
||||
|
||||
func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) {
|
||||
meshTimeline.append(message)
|
||||
@discardableResult
|
||||
func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool {
|
||||
appendedPublicMessages.append((message, conversationID))
|
||||
return true
|
||||
}
|
||||
|
||||
func refreshVisibleMessages(from channel: ChannelID?) {
|
||||
refreshedChannels.append(channel)
|
||||
}
|
||||
|
||||
func trimMessagesIfNeeded() { trimCount += 1 }
|
||||
|
||||
func removeMessage(withID messageID: String, cleanupFile: Bool) {
|
||||
removedMessages.append((messageID, cleanupFile))
|
||||
}
|
||||
@@ -129,22 +123,21 @@ struct ChatMediaTransferCoordinatorContextTests {
|
||||
#expect(message.senderPeerID == context.myPeerID)
|
||||
#expect(message.deliveryStatus == .sending)
|
||||
#expect(context.recordedContentKeys == ["[voice] note.m4a"])
|
||||
#expect(context.trimCount == 1)
|
||||
#expect(context.notifyUIChangedCount == 1)
|
||||
#expect(context.meshTimeline.isEmpty)
|
||||
#expect(context.appendedPublicMessages.isEmpty)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func enqueueMediaMessage_publicAppendsToTimelineAndRefreshes() async {
|
||||
func enqueueMediaMessage_publicAppendsToActiveConversation() async {
|
||||
let context = MockChatMediaTransferContext()
|
||||
let coordinator = ChatMediaTransferCoordinator(context: context)
|
||||
|
||||
let message = coordinator.enqueueMediaMessage(content: "[image] pic.jpg", targetPeer: nil)
|
||||
|
||||
#expect(context.meshTimeline.map(\.id) == [message.id])
|
||||
#expect(context.appendedPublicMessages.map(\.message.id) == [message.id])
|
||||
#expect(context.appendedPublicMessages.first?.conversationID == .mesh)
|
||||
#expect(!message.isPrivate)
|
||||
#expect(message.sender == "me")
|
||||
#expect(context.refreshedChannels.count == 1)
|
||||
#expect(context.privateChats.isEmpty)
|
||||
#expect(context.notifyUIChangedCount == 1)
|
||||
}
|
||||
@@ -210,7 +203,7 @@ struct ChatMediaTransferCoordinatorContextTests {
|
||||
#expect(!FileManager.default.fileExists(atPath: url.path))
|
||||
#expect(context.systemMessages == ["Voice notes are only available in mesh chats."])
|
||||
#expect(context.privateChats.isEmpty)
|
||||
#expect(context.meshTimeline.isEmpty)
|
||||
#expect(context.appendedPublicMessages.isEmpty)
|
||||
#expect(coordinator.transferIdToMessageIDs.isEmpty)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,16 +42,13 @@ private final class MockChatNostrContext: ChatNostrContext {
|
||||
|
||||
// Public timeline & pipeline
|
||||
var messages: [BitchatMessage] = []
|
||||
private(set) var pipelineResetCount = 0
|
||||
private(set) var pipelineChannelUpdates: [ChannelID] = []
|
||||
private(set) var pipelineFlushCount = 0
|
||||
private(set) var refreshedChannels: [ChannelID?] = []
|
||||
private(set) var publicSystemMessages: [String] = []
|
||||
var pendingGeohashSystemMessages: [String] = []
|
||||
private(set) var appendedGeohashMessages: [(message: BitchatMessage, geohash: String)] = []
|
||||
private(set) var synchronizedGeohashes: [String] = []
|
||||
|
||||
func resetPublicMessagePipeline() { pipelineResetCount += 1 }
|
||||
func updatePublicMessagePipelineChannel(_ channel: ChannelID) { pipelineChannelUpdates.append(channel) }
|
||||
func flushPublicMessagePipeline() { pipelineFlushCount += 1 }
|
||||
func refreshVisibleMessages(from channel: ChannelID?) { refreshedChannels.append(channel) }
|
||||
func addPublicSystemMessage(_ content: String) { publicSystemMessages.append(content) }
|
||||
|
||||
@@ -68,8 +65,6 @@ private final class MockChatNostrContext: ChatNostrContext {
|
||||
return true
|
||||
}
|
||||
|
||||
func synchronizePublicConversationStore(forGeohash geohash: String) { synchronizedGeohashes.append(geohash) }
|
||||
|
||||
// Inbound public messages
|
||||
private(set) var handledPublicMessages: [BitchatMessage] = []
|
||||
private(set) var mentionCheckedMessageIDs: [String] = []
|
||||
@@ -441,9 +436,8 @@ struct ChatNostrCoordinatorContextTests {
|
||||
|
||||
coordinator.subscriptions.switchLocationChannel(to: .mesh)
|
||||
|
||||
#expect(context.pipelineResetCount == 1)
|
||||
#expect(context.pipelineFlushCount == 1)
|
||||
#expect(context.activeChannel == .mesh)
|
||||
#expect(context.pipelineChannelUpdates == [.mesh])
|
||||
#expect(context.clearProcessedNostrEventsCount == 1)
|
||||
#expect(context.refreshedChannels == [.mesh])
|
||||
#expect(context.refreshTimerStopCount == 1)
|
||||
@@ -578,7 +572,6 @@ struct GeoPresenceTrackerTests {
|
||||
await drainMainQueue()
|
||||
#expect(context.appendedGeohashMessages.isEmpty)
|
||||
#expect(context.lastGeoNotificationAt["9q8yy"] == recent)
|
||||
#expect(context.synchronizedGeohashes.isEmpty)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
@@ -605,7 +598,7 @@ struct GeoPresenceTrackerTests {
|
||||
#expect(context.appendGeohashMessageIfAbsent(placeholder, toGeohash: "9q8yy"))
|
||||
|
||||
// Cooldown elapsed: the geohash is re-stamped and the append is
|
||||
// attempted (and rejected as a duplicate, so no store sync either).
|
||||
// attempted (and rejected as a duplicate, so no notification either).
|
||||
let stale = Date().addingTimeInterval(-TransportConfig.uiGeoNotifyCooldownSeconds - 1)
|
||||
context.lastGeoNotificationAt["9q8yy"] = stale
|
||||
tracker.cooldownPerGeohash("9q8yy", content: "sampled activity", event: event)
|
||||
@@ -614,7 +607,6 @@ struct GeoPresenceTrackerTests {
|
||||
let stamped = try #require(context.lastGeoNotificationAt["9q8yy"])
|
||||
#expect(stamped > stale)
|
||||
#expect(context.appendedGeohashMessages.count == 1)
|
||||
#expect(context.synchronizedGeohashes.isEmpty)
|
||||
}
|
||||
@Test @MainActor
|
||||
func handleFavoriteNotification_persistsFavoriteAndPostsLocalNotification() async throws {
|
||||
@@ -661,10 +653,9 @@ struct GeoPresenceTrackerTests {
|
||||
coordinator.presence.cooldownPerGeohash("u4pruyd", content: "hello geohash", event: first)
|
||||
await drainMainQueue()
|
||||
|
||||
// Sampled message recorded, store synced, and notification posted.
|
||||
// Sampled message recorded in the store and notification posted.
|
||||
#expect(context.appendedGeohashMessages.map(\.message.id) == [first.id])
|
||||
#expect(context.appendedGeohashMessages.first?.message.sender == "alice#" + String(first.pubkey.suffix(4)))
|
||||
#expect(context.synchronizedGeohashes == ["u4pruyd"])
|
||||
#expect(context.geohashActivityNotifications.count == 1)
|
||||
#expect(context.geohashActivityNotifications.first?.geohash == "u4pruyd")
|
||||
#expect(context.geohashActivityNotifications.first?.bodyPreview == "hello geohash")
|
||||
|
||||
@@ -49,21 +49,19 @@ private final class MockChatOutgoingContext: ChatOutgoingContext {
|
||||
}
|
||||
|
||||
// Public timeline (local echo)
|
||||
private(set) var appendedTimelineMessages: [(message: BitchatMessage, channel: ChannelID)] = []
|
||||
private(set) var refreshedChannels: [ChannelID?] = []
|
||||
private(set) var trimMessagesIfNeededCount = 0
|
||||
private(set) var appendedPublicMessages: [(message: BitchatMessage, conversationID: ConversationID)] = []
|
||||
private(set) var systemMessages: [String] = []
|
||||
|
||||
func parseMentions(from content: String) -> [String] {
|
||||
content.contains("@bob") ? ["bob"] : []
|
||||
}
|
||||
|
||||
func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) {
|
||||
appendedTimelineMessages.append((message, channel))
|
||||
@discardableResult
|
||||
func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool {
|
||||
appendedPublicMessages.append((message, conversationID))
|
||||
return true
|
||||
}
|
||||
|
||||
func refreshVisibleMessages(from channel: ChannelID?) { refreshedChannels.append(channel) }
|
||||
func trimMessagesIfNeeded() { trimMessagesIfNeededCount += 1 }
|
||||
func addSystemMessage(_ content: String) { systemMessages.append(content) }
|
||||
|
||||
// Content dedup
|
||||
@@ -129,7 +127,7 @@ struct ChatOutgoingCoordinatorContextTests {
|
||||
await drainMainActorTasks()
|
||||
|
||||
#expect(context.handledCommands == ["/who all"])
|
||||
#expect(context.appendedTimelineMessages.isEmpty)
|
||||
#expect(context.appendedPublicMessages.isEmpty)
|
||||
#expect(context.sentMeshMessages.isEmpty)
|
||||
}
|
||||
|
||||
@@ -153,7 +151,7 @@ struct ChatOutgoingCoordinatorContextTests {
|
||||
coordinator.sendMessage("dropped")
|
||||
await drainMainActorTasks()
|
||||
#expect(context.sentPrivateMessages.count == 1)
|
||||
#expect(context.appendedTimelineMessages.isEmpty)
|
||||
#expect(context.appendedPublicMessages.isEmpty)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
@@ -165,16 +163,14 @@ struct ChatOutgoingCoordinatorContextTests {
|
||||
await drainMainActorTasks()
|
||||
|
||||
// Local echo uses the trimmed content, own nickname/peer ID, mentions.
|
||||
#expect(context.appendedTimelineMessages.count == 1)
|
||||
let echo = context.appendedTimelineMessages[0]
|
||||
#expect(context.appendedPublicMessages.count == 1)
|
||||
let echo = context.appendedPublicMessages[0]
|
||||
#expect(echo.message.content == "hello @bob")
|
||||
#expect(echo.message.sender == "me")
|
||||
#expect(echo.message.senderPeerID == context.myPeerID)
|
||||
#expect(echo.message.mentions == ["bob"])
|
||||
#expect(echo.channel == .mesh)
|
||||
#expect(context.refreshedChannels == [.mesh])
|
||||
#expect(echo.conversationID == .mesh)
|
||||
#expect(context.recordedContentKeys.map(\.key) == ["key:hello @bob"])
|
||||
#expect(context.trimMessagesIfNeededCount == 1)
|
||||
|
||||
// The mesh send carries the original (untrimmed) content and reuses
|
||||
// the echo's message ID and timestamp; activity is stamped for "mesh".
|
||||
@@ -200,8 +196,9 @@ struct ChatOutgoingCoordinatorContextTests {
|
||||
|
||||
// Local echo carries the geohash sender suffix (#last-4-of-pubkey) and
|
||||
// the signed event's ID; the send context targets the same channel.
|
||||
#expect(context.appendedTimelineMessages.count == 1)
|
||||
let echo = context.appendedTimelineMessages[0].message
|
||||
#expect(context.appendedPublicMessages.count == 1)
|
||||
let echo = context.appendedPublicMessages[0].message
|
||||
#expect(context.appendedPublicMessages[0].conversationID == .geohash("u4pruydq"))
|
||||
#expect(echo.sender == "me#2222")
|
||||
#expect(context.recordedActivityKeys == ["geo:u4pruydq"])
|
||||
#expect(context.sentGeohashContexts.count == 1)
|
||||
@@ -215,7 +212,7 @@ struct ChatOutgoingCoordinatorContextTests {
|
||||
coordinator.sendMessage("doomed")
|
||||
await drainMainActorTasks()
|
||||
#expect(context.systemMessages.count == 1)
|
||||
#expect(context.appendedTimelineMessages.count == 1)
|
||||
#expect(context.appendedPublicMessages.count == 1)
|
||||
#expect(context.sentGeohashContexts.count == 1)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,15 +25,13 @@ import BitFoundation
|
||||
/// `ChatPublicConversationCoordinator` is testable without a `ChatViewModel`.
|
||||
@MainActor
|
||||
private final class MockChatPublicConversationContext: ChatPublicConversationContext {
|
||||
// Channel & visible timeline state
|
||||
var messages: [BitchatMessage] = []
|
||||
// Channel state
|
||||
var activeChannel: ChannelID = .mesh
|
||||
var currentGeohash: String?
|
||||
var nickname = "me"
|
||||
var myPeerID = PeerID(str: "0011223344556677")
|
||||
private(set) var isBatchingPublic = false
|
||||
private(set) var notifyUIChangedCount = 0
|
||||
private(set) var trimMessagesCount = 0
|
||||
|
||||
func setPublicBatching(_ isBatching: Bool) {
|
||||
isBatchingPublic = isBatching
|
||||
@@ -43,92 +41,59 @@ private final class MockChatPublicConversationContext: ChatPublicConversationCon
|
||||
notifyUIChangedCount += 1
|
||||
}
|
||||
|
||||
func trimMessagesIfNeeded() {
|
||||
trimMessagesCount += 1
|
||||
}
|
||||
|
||||
// Public timeline store
|
||||
var meshTimeline: [BitchatMessage] = []
|
||||
var geoTimelines: [String: [BitchatMessage]] = [:]
|
||||
// Public conversation store (single-writer intents)
|
||||
var conversations: [ConversationID: [BitchatMessage]] = [:]
|
||||
private(set) var queuedGeohashSystemMessages: [String] = []
|
||||
|
||||
func timelineMessages(for channel: ChannelID) -> [BitchatMessage] {
|
||||
switch channel {
|
||||
case .mesh: return meshTimeline
|
||||
case .location(let channel): return geoTimelines[channel.geohash] ?? []
|
||||
}
|
||||
func publicMessages(in conversationID: ConversationID) -> [BitchatMessage] {
|
||||
conversations[conversationID] ?? []
|
||||
}
|
||||
|
||||
func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) {
|
||||
switch channel {
|
||||
case .mesh: meshTimeline.append(message)
|
||||
case .location(let channel): geoTimelines[channel.geohash, default: []].append(message)
|
||||
}
|
||||
}
|
||||
|
||||
func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool {
|
||||
if geoTimelines[geohash]?.contains(where: { $0.id == message.id }) == true {
|
||||
@discardableResult
|
||||
func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool {
|
||||
guard conversations[conversationID]?.contains(where: { $0.id == message.id }) != true else {
|
||||
return false
|
||||
}
|
||||
geoTimelines[geohash, default: []].append(message)
|
||||
conversations[conversationID, default: []].append(message)
|
||||
return true
|
||||
}
|
||||
|
||||
func removeTimelineMessage(withID id: String) -> BitchatMessage? {
|
||||
if let index = meshTimeline.firstIndex(where: { $0.id == id }) {
|
||||
return meshTimeline.remove(at: index)
|
||||
}
|
||||
for (geohash, timeline) in geoTimelines {
|
||||
guard let index = timeline.firstIndex(where: { $0.id == id }) else { continue }
|
||||
@discardableResult
|
||||
func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool {
|
||||
appendPublicMessage(message, to: .geohash(geohash.lowercased()))
|
||||
}
|
||||
|
||||
func publicConversationContainsMessage(withID messageID: String, in conversationID: ConversationID) -> Bool {
|
||||
conversations[conversationID]?.contains(where: { $0.id == messageID }) == true
|
||||
}
|
||||
|
||||
@discardableResult
|
||||
func removePublicMessage(withID messageID: String) -> BitchatMessage? {
|
||||
for (conversationID, timeline) in conversations {
|
||||
guard let index = timeline.firstIndex(where: { $0.id == messageID }) else { continue }
|
||||
var updated = timeline
|
||||
let removed = updated.remove(at: index)
|
||||
geoTimelines[geohash] = updated
|
||||
conversations[conversationID] = updated
|
||||
return removed
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func removeGeohashTimelineMessages(in geohash: String, where predicate: (BitchatMessage) -> Bool) {
|
||||
geoTimelines[geohash]?.removeAll(where: predicate)
|
||||
func removePublicMessages(fromGeohash geohash: String, where predicate: (BitchatMessage) -> Bool) {
|
||||
conversations[.geohash(geohash.lowercased())]?.removeAll(where: predicate)
|
||||
}
|
||||
|
||||
func clearTimeline(for channel: ChannelID) {
|
||||
switch channel {
|
||||
case .mesh: meshTimeline.removeAll()
|
||||
case .location(let channel): geoTimelines[channel.geohash] = []
|
||||
}
|
||||
}
|
||||
private(set) var clearedConversations: [ConversationID] = []
|
||||
|
||||
func timelineGeohashKeys() -> [String] {
|
||||
Array(geoTimelines.keys)
|
||||
func clearPublicConversation(_ conversationID: ConversationID) {
|
||||
clearedConversations.append(conversationID)
|
||||
conversations[conversationID] = []
|
||||
}
|
||||
|
||||
func queueGeohashSystemMessage(_ content: String) {
|
||||
queuedGeohashSystemMessages.append(content)
|
||||
}
|
||||
|
||||
// Conversation stores
|
||||
private(set) var conversationActiveChannels: [ChannelID] = []
|
||||
private(set) var replacedChannelMessages: [(channel: ChannelID, messageIDs: [String])] = []
|
||||
private(set) var replacedConversationMessages: [(conversation: ConversationID, messageIDs: [String])] = []
|
||||
private(set) var selectionStoreSyncCount = 0
|
||||
|
||||
func setConversationActiveChannel(_ channel: ChannelID) {
|
||||
conversationActiveChannels.append(channel)
|
||||
}
|
||||
|
||||
func replaceConversationMessages(_ messages: [BitchatMessage], for channelID: ChannelID) {
|
||||
replacedChannelMessages.append((channelID, messages.map(\.id)))
|
||||
}
|
||||
|
||||
func replaceConversationMessages(_ messages: [BitchatMessage], for conversationID: ConversationID) {
|
||||
replacedConversationMessages.append((conversationID, messages.map(\.id)))
|
||||
}
|
||||
|
||||
func synchronizeConversationSelectionStore() {
|
||||
selectionStoreSyncCount += 1
|
||||
}
|
||||
|
||||
// Private chats
|
||||
var privateChats: [PeerID: [BitchatMessage]] = [:]
|
||||
var unreadPrivateMessages: Set<PeerID> = []
|
||||
@@ -222,7 +187,8 @@ private final class MockChatPublicConversationContext: ChatPublicConversationCon
|
||||
var blockedMessageIDs: Set<String> = []
|
||||
var rateLimitAllowed = true
|
||||
private(set) var rateLimitChecks: [(senderKey: String, contentKey: String)] = []
|
||||
private(set) var enqueuedMessageIDs: [String] = []
|
||||
private(set) var enqueuedMessages: [(messageID: String, conversationID: ConversationID)] = []
|
||||
var enqueuedMessageIDs: [String] { enqueuedMessages.map(\.messageID) }
|
||||
var stablePeerIDs: [PeerID: PeerID] = [:]
|
||||
|
||||
func processActionMessage(_ message: BitchatMessage) -> BitchatMessage {
|
||||
@@ -238,8 +204,8 @@ private final class MockChatPublicConversationContext: ChatPublicConversationCon
|
||||
return rateLimitAllowed
|
||||
}
|
||||
|
||||
func enqueuePublicMessage(_ message: BitchatMessage) {
|
||||
enqueuedMessageIDs.append(message.id)
|
||||
func enqueuePublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) {
|
||||
enqueuedMessages.append((message.id, conversationID))
|
||||
}
|
||||
|
||||
func cachedStablePeerID(for shortPeerID: PeerID) -> PeerID? {
|
||||
@@ -309,24 +275,24 @@ private func makePublicMessage(
|
||||
struct ChatPublicConversationCoordinatorContextTests {
|
||||
|
||||
@Test @MainActor
|
||||
func handlePublicMessage_meshMessage_appendsSyncsStoreAndEnqueues() async {
|
||||
func handlePublicMessage_meshMessage_enqueuesForBatchedStoreCommit() async {
|
||||
let context = MockChatPublicConversationContext()
|
||||
let coordinator = ChatPublicConversationCoordinator(context: context)
|
||||
let message = makePublicMessage(id: "mesh-msg-1", content: "Hello Mesh")
|
||||
|
||||
coordinator.handlePublicMessage(message)
|
||||
|
||||
#expect(context.meshTimeline.map(\.id) == ["mesh-msg-1"])
|
||||
#expect(context.replacedChannelMessages.count == 1)
|
||||
#expect(context.replacedChannelMessages.first?.channel == .mesh)
|
||||
#expect(context.replacedChannelMessages.first?.messageIDs == ["mesh-msg-1"])
|
||||
// Visible-channel arrival: buffered for the batched pipeline flush
|
||||
// (which commits to the store), not appended directly.
|
||||
#expect(context.rateLimitChecks.count == 1)
|
||||
#expect(context.rateLimitChecks.first?.senderKey == "mesh:aabbccddeeff0011")
|
||||
#expect(context.rateLimitChecks.first?.contentKey == "hello mesh")
|
||||
#expect(context.enqueuedMessageIDs == ["mesh-msg-1"])
|
||||
#expect(context.enqueuedMessages.map(\.messageID) == ["mesh-msg-1"])
|
||||
#expect(context.enqueuedMessages.first?.conversationID == .mesh)
|
||||
#expect(context.publicMessages(in: .mesh).isEmpty)
|
||||
|
||||
// Already visible in the timeline: stored again, but not re-enqueued.
|
||||
context.messages = [message]
|
||||
// Already committed to the store: not re-enqueued.
|
||||
context.appendPublicMessage(message, to: .mesh)
|
||||
coordinator.handlePublicMessage(message)
|
||||
#expect(context.enqueuedMessageIDs == ["mesh-msg-1"])
|
||||
}
|
||||
@@ -340,14 +306,14 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
context.blockedMessageIDs = ["blocked-msg"]
|
||||
coordinator.handlePublicMessage(makePublicMessage(id: "blocked-msg"))
|
||||
#expect(context.rateLimitChecks.isEmpty)
|
||||
#expect(context.meshTimeline.isEmpty)
|
||||
#expect(context.publicMessages(in: .mesh).isEmpty)
|
||||
#expect(context.enqueuedMessageIDs.isEmpty)
|
||||
|
||||
// Rate limited: consulted, then dropped before storage.
|
||||
context.rateLimitAllowed = false
|
||||
coordinator.handlePublicMessage(makePublicMessage(id: "limited-msg"))
|
||||
#expect(context.rateLimitChecks.count == 1)
|
||||
#expect(context.meshTimeline.isEmpty)
|
||||
#expect(context.publicMessages(in: .mesh).isEmpty)
|
||||
#expect(context.enqueuedMessageIDs.isEmpty)
|
||||
}
|
||||
|
||||
@@ -364,16 +330,15 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
senderPeerID: PeerID(nostr: senderHex)
|
||||
)
|
||||
|
||||
// On mesh channel: stored in the geohash timeline but not enqueued.
|
||||
// On mesh channel: a background-channel arrival lands in the geohash
|
||||
// conversation immediately, with no pipeline batching.
|
||||
context.activeChannel = .mesh
|
||||
coordinator.handlePublicMessage(geoMessage)
|
||||
#expect(context.geoTimelines[geohash]?.map(\.id) == ["geo-msg-1"])
|
||||
#expect(context.replacedConversationMessages.count == 1)
|
||||
#expect(context.replacedConversationMessages.first?.conversation == .geohash(geohash))
|
||||
#expect(context.meshTimeline.isEmpty)
|
||||
#expect(context.publicMessages(in: .geohash(geohash)).map(\.id) == ["geo-msg-1"])
|
||||
#expect(context.publicMessages(in: .mesh).isEmpty)
|
||||
#expect(context.enqueuedMessageIDs.isEmpty)
|
||||
|
||||
// On the matching location channel: enqueued for display.
|
||||
// On the matching location channel: enqueued for the batched flush.
|
||||
context.activeChannel = .location(GeohashChannel(level: .city, geohash: geohash))
|
||||
let second = makePublicMessage(
|
||||
id: "geo-msg-2",
|
||||
@@ -381,7 +346,8 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
senderPeerID: PeerID(nostr: senderHex)
|
||||
)
|
||||
coordinator.handlePublicMessage(second)
|
||||
#expect(context.enqueuedMessageIDs == ["geo-msg-2"])
|
||||
#expect(context.enqueuedMessages.map(\.messageID) == ["geo-msg-2"])
|
||||
#expect(context.enqueuedMessages.first?.conversationID == .geohash(geohash))
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
@@ -396,8 +362,7 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
|
||||
context.currentGeohash = geohash
|
||||
context.activeChannel = .location(GeohashChannel(level: .city, geohash: geohash))
|
||||
context.geoTimelines[geohash] = [geoMessage]
|
||||
context.messages = [geoMessage]
|
||||
context.conversations[.geohash(geohash)] = [geoMessage]
|
||||
context.nostrKeyMapping = [senderPeerID: hex, convKey: hex]
|
||||
context.privateChats[convKey] = [geoMessage]
|
||||
context.unreadPrivateMessages = [convKey]
|
||||
@@ -406,13 +371,14 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
|
||||
#expect(context.blockedNostrPubkeys.contains(hex))
|
||||
#expect(context.removedGeoParticipants == [hex])
|
||||
#expect(context.geoTimelines[geohash]?.isEmpty == true)
|
||||
#expect(context.privateChats[convKey] == nil)
|
||||
#expect(context.unreadPrivateMessages.isEmpty)
|
||||
#expect(context.nostrKeyMapping.isEmpty)
|
||||
// The blocked user's visible message is gone; a system notice was added.
|
||||
#expect(!context.messages.contains(where: { $0.id == "geo-bad-1" }))
|
||||
#expect(context.messages.last?.sender == "system")
|
||||
// The blocked user's message is purged from the geohash conversation
|
||||
// (the visible timeline is the same conversation now); a system
|
||||
// notice was appended to the active conversation.
|
||||
#expect(!context.publicMessages(in: .geohash(geohash)).contains(where: { $0.id == "geo-bad-1" }))
|
||||
#expect(context.publicMessages(in: .geohash(geohash)).last?.sender == "system")
|
||||
|
||||
coordinator.unblockGeohashUser(pubkeyHexLowercased: hex, displayName: "rude#abcd")
|
||||
#expect(!context.blockedNostrPubkeys.contains(hex))
|
||||
@@ -424,36 +390,27 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
let coordinator = ChatPublicConversationCoordinator(context: context)
|
||||
let peerID = PeerID(str: "0102030405060708")
|
||||
let message = makePublicMessage(id: "doomed-msg")
|
||||
context.messages = [message]
|
||||
context.meshTimeline = [message]
|
||||
context.conversations[.mesh] = [message]
|
||||
context.privateChats[peerID] = [message]
|
||||
|
||||
coordinator.removeMessage(withID: "doomed-msg", cleanupFile: true)
|
||||
|
||||
#expect(context.messages.isEmpty)
|
||||
#expect(context.meshTimeline.isEmpty)
|
||||
#expect(context.publicMessages(in: .mesh).isEmpty)
|
||||
#expect(context.privateChats[peerID] == nil)
|
||||
#expect(context.cleanedUpFileMessageIDs == ["doomed-msg"])
|
||||
#expect(context.notifyUIChangedCount == 1)
|
||||
// Timeline removal triggers a full conversation-store resync.
|
||||
#expect(context.replacedChannelMessages.contains(where: { $0.channel == .mesh && $0.messageIDs.isEmpty }))
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func addPublicSystemMessage_appendsRefreshesAndRecordsContentKey() async {
|
||||
func addPublicSystemMessage_appendsToActiveConversationAndRecordsContentKey() async {
|
||||
let context = MockChatPublicConversationContext()
|
||||
let coordinator = ChatPublicConversationCoordinator(context: context)
|
||||
|
||||
coordinator.addPublicSystemMessage("Tor Ready")
|
||||
|
||||
#expect(context.meshTimeline.count == 1)
|
||||
#expect(context.meshTimeline.first?.sender == "system")
|
||||
// refreshVisibleMessages mirrors the timeline into the visible list.
|
||||
#expect(context.messages.map(\.id) == context.meshTimeline.map(\.id))
|
||||
#expect(context.publicMessages(in: .mesh).count == 1)
|
||||
#expect(context.publicMessages(in: .mesh).first?.sender == "system")
|
||||
#expect(context.recordedContentKeys.map(\.key) == ["tor ready"])
|
||||
#expect(context.trimMessagesCount == 1)
|
||||
#expect(context.notifyUIChangedCount == 1)
|
||||
#expect(context.conversationActiveChannels == [.mesh])
|
||||
|
||||
// On mesh, geohash-only system messages are queued for the next geo visit.
|
||||
coordinator.addGeohashOnlySystemMessage("geo notice")
|
||||
@@ -477,22 +434,20 @@ struct ChatPublicConversationCoordinatorContextTests {
|
||||
let coordinator = ChatPublicConversationCoordinator(context: context)
|
||||
let pipeline = PublicMessagePipeline()
|
||||
let message = makePublicMessage(id: "pipeline-msg")
|
||||
context.messages = [message]
|
||||
context.contentTimestamps["key-1"] = Date(timeIntervalSince1970: 42)
|
||||
|
||||
#expect(coordinator.pipelineCurrentMessages(pipeline).map(\.id) == ["pipeline-msg"])
|
||||
#expect(coordinator.pipeline(pipeline, normalizeContent: "HeLLo") == "hello")
|
||||
#expect(coordinator.pipeline(pipeline, contentTimestampForKey: "key-1") == Date(timeIntervalSince1970: 42))
|
||||
|
||||
coordinator.pipeline(pipeline, setMessages: [])
|
||||
#expect(context.messages.isEmpty)
|
||||
// Commit lands in the store via the append intent; a duplicate ID
|
||||
// reports `false` (the store's dedup contract).
|
||||
#expect(coordinator.pipeline(pipeline, commit: message, to: .mesh))
|
||||
#expect(context.publicMessages(in: .mesh).map(\.id) == ["pipeline-msg"])
|
||||
#expect(!coordinator.pipeline(pipeline, commit: message, to: .mesh))
|
||||
|
||||
coordinator.pipeline(pipeline, recordContentKey: "key-2", timestamp: Date(timeIntervalSince1970: 7))
|
||||
#expect(context.recordedContentKeys.map(\.key) == ["key-2"])
|
||||
|
||||
coordinator.pipelineTrimMessages(pipeline)
|
||||
#expect(context.trimMessagesCount == 1)
|
||||
|
||||
coordinator.pipelinePrewarmMessage(pipeline, message: message)
|
||||
#expect(context.prewarmedMessageIDs == ["pipeline-msg"])
|
||||
|
||||
|
||||
@@ -233,7 +233,7 @@ struct ChatViewModelDeliveryStatusTests {
|
||||
isPrivate: false,
|
||||
deliveryStatus: .sending
|
||||
)
|
||||
viewModel.messages.append(message)
|
||||
viewModel.seedPublicMessages([message])
|
||||
|
||||
// Action: update to .sent
|
||||
viewModel.didUpdateMessageDeliveryStatus(messageID, status: .sent)
|
||||
@@ -253,7 +253,7 @@ struct ChatViewModelDeliveryStatusTests {
|
||||
let firstPeerID = PeerID(str: "0102030405060708")
|
||||
let secondPeerID = PeerID(str: "1112131415161718")
|
||||
|
||||
viewModel.messages = [
|
||||
viewModel.seedPublicMessages([
|
||||
BitchatMessage(
|
||||
id: messageID,
|
||||
sender: viewModel.nickname,
|
||||
@@ -264,7 +264,7 @@ struct ChatViewModelDeliveryStatusTests {
|
||||
senderPeerID: transport.myPeerID,
|
||||
deliveryStatus: .sent
|
||||
)
|
||||
]
|
||||
])
|
||||
viewModel.seedPrivateChat([
|
||||
BitchatMessage(
|
||||
id: messageID,
|
||||
|
||||
@@ -138,7 +138,7 @@ struct ChatViewModelRefactoringTests {
|
||||
// Wait for async processing with proper timeout
|
||||
let found = await TestHelpers.waitUntil(
|
||||
{
|
||||
viewModel.timelineStore.messages(for: .mesh).contains(where: { $0.content == "Public Hi" })
|
||||
viewModel.publicMessages(for: .mesh).contains(where: { $0.content == "Public Hi" })
|
||||
},
|
||||
timeout: TestConstants.defaultTimeout
|
||||
)
|
||||
|
||||
@@ -438,7 +438,7 @@ struct ChatViewModelReceivingTests {
|
||||
)
|
||||
|
||||
let found = await TestHelpers.waitUntil({
|
||||
viewModel.timelineStore.messages(for: .mesh).contains { $0.content == "Public hello from Bob" }
|
||||
viewModel.publicMessages(for: .mesh).contains { $0.content == "Public hello from Bob" }
|
||||
}, timeout: TestConstants.defaultTimeout)
|
||||
|
||||
#expect(found)
|
||||
@@ -709,10 +709,12 @@ struct ChatViewModelPublicConversationTests {
|
||||
let (viewModel, _) = makeTestableViewModel()
|
||||
|
||||
viewModel.addPublicSystemMessage("system refresh test")
|
||||
viewModel.messages.removeAll()
|
||||
viewModel.refreshVisibleMessages(from: .mesh)
|
||||
|
||||
// The system message lives in the mesh conversation itself, so the
|
||||
// derived `messages` view still surfaces it after a refresh.
|
||||
#expect(viewModel.messages.last?.content == "system refresh test")
|
||||
#expect(viewModel.publicMessages(for: .mesh).last?.content == "system refresh test")
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
@@ -726,6 +728,18 @@ struct ChatViewModelPublicConversationTests {
|
||||
viewModel.refreshVisibleMessages(from: .mesh)
|
||||
|
||||
#expect(viewModel.messages.isEmpty)
|
||||
#expect(viewModel.publicMessages(for: .mesh).isEmpty)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func queuedGeohashSystemMessages_drainOnce() async {
|
||||
let (viewModel, _) = makeTestableViewModel()
|
||||
|
||||
viewModel.queueGeohashSystemMessage("first")
|
||||
viewModel.queueGeohashSystemMessage("second")
|
||||
|
||||
#expect(viewModel.drainPendingGeohashSystemMessages() == ["first", "second"])
|
||||
#expect(viewModel.drainPendingGeohashSystemMessages().isEmpty)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -999,7 +1013,7 @@ struct ChatViewModelPanicTests {
|
||||
|
||||
// Set up some state
|
||||
transport.connectedPeers.insert(PeerID(str: "PEER1"))
|
||||
viewModel.messages = [
|
||||
viewModel.seedPublicMessages([
|
||||
BitchatMessage(
|
||||
id: "panic-1",
|
||||
sender: "Tester",
|
||||
@@ -1007,7 +1021,7 @@ struct ChatViewModelPanicTests {
|
||||
timestamp: Date(),
|
||||
isRelay: false
|
||||
)
|
||||
]
|
||||
])
|
||||
viewModel.seedPrivateChat([
|
||||
BitchatMessage(
|
||||
id: "pm-1",
|
||||
|
||||
@@ -468,4 +468,98 @@ struct ConversationStoreTests {
|
||||
store.append(makeMessage(id: "m3", timestamp: 3), to: b)
|
||||
#expect(bWillChangeCount > 0)
|
||||
}
|
||||
|
||||
// MARK: - Public timelines (mesh/geohash, ex-PublicTimelineStore behavior)
|
||||
|
||||
@Test("geohash conversations are separated by geohash and from mesh")
|
||||
@MainActor
|
||||
func geohashConversationSeparation() {
|
||||
let store = ConversationStore()
|
||||
store.append(makeMessage(id: "mesh-1", timestamp: 1), to: .mesh)
|
||||
store.append(makeMessage(id: "geo-a-1", timestamp: 2), to: .geohash("u4pruyd"))
|
||||
store.append(makeMessage(id: "geo-b-1", timestamp: 3), to: .geohash("9q8yy"))
|
||||
|
||||
#expect(store.conversation(for: .mesh).messages.map(\.id) == ["mesh-1"])
|
||||
#expect(store.conversation(for: .geohash("u4pruyd")).messages.map(\.id) == ["geo-a-1"])
|
||||
#expect(store.conversation(for: .geohash("9q8yy")).messages.map(\.id) == ["geo-b-1"])
|
||||
}
|
||||
|
||||
@Test("geohash append dedups by ID and reports duplicates")
|
||||
@MainActor
|
||||
func geohashAppendIfAbsentContract() {
|
||||
let store = ConversationStore()
|
||||
let message = makeMessage(id: "geo-1", timestamp: 1)
|
||||
|
||||
#expect(store.append(message, to: .geohash("u4pruyd")))
|
||||
#expect(!store.append(message, to: .geohash("u4pruyd")))
|
||||
// The same ID is still fresh in a different geohash.
|
||||
#expect(store.append(message, to: .geohash("9q8yy")))
|
||||
}
|
||||
|
||||
@Test("removePublicMessage searches mesh and geohash conversations only")
|
||||
@MainActor
|
||||
func removePublicMessageSearchesPublicConversations() {
|
||||
let store = ConversationStore()
|
||||
let direct = makeDirectConversationID("aa")
|
||||
store.append(makeMessage(id: "mesh-1", timestamp: 1), to: .mesh)
|
||||
store.append(makeMessage(id: "geo-1", timestamp: 2), to: .geohash("u4pruyd"))
|
||||
store.append(makeMessage(id: "dm-1", timestamp: 3, isPrivate: true), to: direct)
|
||||
|
||||
#expect(store.removePublicMessage(withID: "geo-1")?.id == "geo-1")
|
||||
#expect(store.conversation(for: .geohash("u4pruyd")).messages.isEmpty)
|
||||
|
||||
#expect(store.removePublicMessage(withID: "mesh-1")?.id == "mesh-1")
|
||||
#expect(store.conversation(for: .mesh).messages.isEmpty)
|
||||
|
||||
// Direct conversations are never touched.
|
||||
#expect(store.removePublicMessage(withID: "dm-1") == nil)
|
||||
#expect(store.conversation(for: direct).messages.map(\.id) == ["dm-1"])
|
||||
}
|
||||
|
||||
@Test("removeMessages(from:where:) purges matches and emits per removal")
|
||||
@MainActor
|
||||
func removeMessagesByPredicate() {
|
||||
let store = ConversationStore()
|
||||
let id = ConversationID.geohash("u4pruyd")
|
||||
store.append(makeMessage(id: "keep-1", timestamp: 1), to: id)
|
||||
store.append(makeMessage(id: "drop-1", timestamp: 2, content: "purge me"), to: id)
|
||||
store.append(makeMessage(id: "drop-2", timestamp: 3, content: "purge me"), to: id)
|
||||
store.append(makeMessage(id: "keep-2", timestamp: 4), to: id)
|
||||
|
||||
var removedIDs: [String] = []
|
||||
var cancellables = Set<AnyCancellable>()
|
||||
store.changes
|
||||
.sink { change in
|
||||
if case .messageRemoved(_, let messageID) = change {
|
||||
removedIDs.append(messageID)
|
||||
}
|
||||
}
|
||||
.store(in: &cancellables)
|
||||
|
||||
store.removeMessages(from: id, where: { $0.content == "purge me" })
|
||||
|
||||
#expect(store.conversation(for: id).messages.map(\.id) == ["keep-1", "keep-2"])
|
||||
#expect(removedIDs == ["drop-1", "drop-2"])
|
||||
// The ID index survives the purge: dedup and removal still work.
|
||||
#expect(!store.append(makeMessage(id: "keep-2", timestamp: 4), to: id))
|
||||
#expect(store.removeMessage(withID: "keep-1", from: id) != nil)
|
||||
#expect(store.conversation(for: id).messages.map(\.id) == ["keep-2"])
|
||||
}
|
||||
|
||||
@Test("trimmed public message IDs can return after falling off the cap")
|
||||
@MainActor
|
||||
func trimmedMessageIDsCanReturn() {
|
||||
let store = ConversationStore()
|
||||
let id = ConversationID.geohash("u4pruyd")
|
||||
let conversation = store.conversation(for: id)
|
||||
let first = makeMessage(id: "one", timestamp: 1)
|
||||
|
||||
store.append(first, to: id)
|
||||
for index in 0..<conversation.cap {
|
||||
store.append(makeMessage(id: "filler-\(index)", timestamp: 2 + TimeInterval(index)), to: id)
|
||||
}
|
||||
// "one" was trimmed by the cap, so its ID is free again.
|
||||
#expect(!conversation.containsMessage(withID: "one"))
|
||||
#expect(store.append(makeMessage(id: "one", timestamp: 2000), to: id))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -174,4 +174,73 @@ struct LegacyConversationStoreBridgeTests {
|
||||
#expect(legacy.directMessagesByPeerID()[oldPeerID] ?? [] == [])
|
||||
#expect(legacy.directMessagesByPeerID()[newPeerID]?.map(\.id) == ["coord-mig-1"])
|
||||
}
|
||||
|
||||
// MARK: - Public mirroring (migration step 3)
|
||||
|
||||
@Test("public messages mirror into Legacy via the coalesced flush")
|
||||
@MainActor
|
||||
func publicMessagesMirrorIntoLegacy() async {
|
||||
let (viewModel, store, legacy, _) = makeBridgedFixture()
|
||||
|
||||
viewModel.appendPublicMessage(
|
||||
BitchatMessage(
|
||||
id: "bridge-pub-1",
|
||||
sender: "alice",
|
||||
content: "hello mesh",
|
||||
timestamp: Date(),
|
||||
isRelay: false
|
||||
),
|
||||
to: .mesh
|
||||
)
|
||||
viewModel.appendGeohashMessageIfAbsent(
|
||||
BitchatMessage(
|
||||
id: "bridge-geo-1",
|
||||
sender: "bob#abcd",
|
||||
content: "hello geohash",
|
||||
timestamp: Date(),
|
||||
isRelay: false
|
||||
),
|
||||
toGeohash: "U4PRUYD"
|
||||
)
|
||||
|
||||
// The new store is synchronously authoritative (geohash keys are
|
||||
// normalized to lowercase).
|
||||
#expect(store.conversation(for: .mesh).messages.map(\.id) == ["bridge-pub-1"])
|
||||
#expect(store.conversation(for: .geohash("u4pruyd")).messages.map(\.id) == ["bridge-geo-1"])
|
||||
|
||||
// Legacy catches up within one coalesced flush.
|
||||
let mirrored = await TestHelpers.waitUntil(
|
||||
{
|
||||
legacy.messages(for: .mesh).map(\.id) == ["bridge-pub-1"]
|
||||
&& legacy.messages(for: .geohash("u4pruyd")).map(\.id) == ["bridge-geo-1"]
|
||||
},
|
||||
timeout: 1.0
|
||||
)
|
||||
#expect(mirrored)
|
||||
}
|
||||
|
||||
@Test("removing a public conversation empties its Legacy mirror immediately")
|
||||
@MainActor
|
||||
func publicConversationRemovalClearsLegacy() async {
|
||||
let (viewModel, store, legacy, _) = makeBridgedFixture()
|
||||
viewModel.appendPublicMessage(
|
||||
BitchatMessage(
|
||||
id: "bridge-pub-2",
|
||||
sender: "alice",
|
||||
content: "soon gone",
|
||||
timestamp: Date(),
|
||||
isRelay: false
|
||||
),
|
||||
to: .mesh
|
||||
)
|
||||
let mirrored = await TestHelpers.waitUntil(
|
||||
{ legacy.messages(for: .mesh).map(\.id) == ["bridge-pub-2"] },
|
||||
timeout: 1.0
|
||||
)
|
||||
#expect(mirrored)
|
||||
|
||||
// Panic-style removal: Legacy must never show stale public messages.
|
||||
store.removeConversation(.mesh)
|
||||
#expect(legacy.messages(for: .mesh).isEmpty)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -389,9 +389,9 @@ final class PerformanceBaselineTests: XCTestCase {
|
||||
/// Baseline for the full public-message ingest cycle through a real
|
||||
/// `ChatViewModel`: `didReceivePublicMessage` (transport delegate entry,
|
||||
/// main-actor Task hop per message) → `handlePublicMessage` (rate limit,
|
||||
/// `PublicTimelineStore` append, per-message full-array
|
||||
/// `LegacyConversationStore` sync) → `PublicMessagePipeline` timer-batched
|
||||
/// flush into `ChatViewModel.messages` → `PublicChatModel` mirror.
|
||||
/// pipeline enqueue) → `PublicMessagePipeline` timer-batched flush into
|
||||
/// the `ConversationStore` (derived `ChatViewModel.messages` view) →
|
||||
/// coalesced `LegacyConversationStoreBridge` mirror → `PublicChatModel`.
|
||||
/// Measures until `messages` and the feature model reflect every message,
|
||||
/// so the pipeline's flush latency is part of the cycle. Senders are
|
||||
/// spread 4-per-peer to stay under the 5-token sender rate bucket.
|
||||
@@ -567,13 +567,11 @@ private final class PerfNostrContext: ChatNostrContext {
|
||||
}
|
||||
|
||||
var messages: [BitchatMessage] = []
|
||||
func resetPublicMessagePipeline() {}
|
||||
func updatePublicMessagePipelineChannel(_ channel: ChannelID) {}
|
||||
func flushPublicMessagePipeline() {}
|
||||
func refreshVisibleMessages(from channel: ChannelID?) {}
|
||||
func addPublicSystemMessage(_ content: String) {}
|
||||
func drainPendingGeohashSystemMessages() -> [String] { [] }
|
||||
func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { true }
|
||||
func synchronizePublicConversationStore(forGeohash geohash: String) {}
|
||||
|
||||
private(set) var handledPublicMessageCount = 0
|
||||
func handlePublicMessage(_ message: BitchatMessage) { handledPublicMessageCount += 1 }
|
||||
|
||||
@@ -2,7 +2,10 @@
|
||||
// PublicMessagePipelineTests.swift
|
||||
// bitchatTests
|
||||
//
|
||||
// Tests for PublicMessagePipeline ordering and deduplication.
|
||||
// Tests for PublicMessagePipeline batching, content dedup, and per-message
|
||||
// conversation routing. Ordering and ID dedup live in the ConversationStore
|
||||
// the flush commits into (the old late-insert threshold is gone; see
|
||||
// ConversationStoreTests for ordered-insert coverage).
|
||||
//
|
||||
|
||||
import Testing
|
||||
@@ -13,14 +16,15 @@ import BitFoundation
|
||||
@MainActor
|
||||
private final class TestPipelineDelegate: PublicMessagePipelineDelegate {
|
||||
private let dedupService = MessageDeduplicationService()
|
||||
var messages: [BitchatMessage] = []
|
||||
/// Commits in arrival-at-commit order, per conversation.
|
||||
private(set) var committed: [(message: BitchatMessage, conversationID: ConversationID)] = []
|
||||
/// Message IDs the commit rejects (simulates the store's ID dedup).
|
||||
var rejectedMessageIDs: Set<String> = []
|
||||
private(set) var recordedContentKeys: [String] = []
|
||||
private(set) var batchingStates: [Bool] = []
|
||||
|
||||
func pipelineCurrentMessages(_ pipeline: PublicMessagePipeline) -> [BitchatMessage] {
|
||||
messages
|
||||
}
|
||||
|
||||
func pipeline(_ pipeline: PublicMessagePipeline, setMessages messages: [BitchatMessage]) {
|
||||
self.messages = messages
|
||||
func messages(in conversationID: ConversationID) -> [BitchatMessage] {
|
||||
committed.filter { $0.conversationID == conversationID }.map(\.message)
|
||||
}
|
||||
|
||||
func pipeline(_ pipeline: PublicMessagePipeline, normalizeContent content: String) -> String {
|
||||
@@ -33,19 +37,37 @@ private final class TestPipelineDelegate: PublicMessagePipelineDelegate {
|
||||
|
||||
func pipeline(_ pipeline: PublicMessagePipeline, recordContentKey key: String, timestamp: Date) {
|
||||
dedupService.recordContentKey(key, timestamp: timestamp)
|
||||
recordedContentKeys.append(key)
|
||||
}
|
||||
|
||||
func pipelineTrimMessages(_ pipeline: PublicMessagePipeline) {}
|
||||
func pipeline(_ pipeline: PublicMessagePipeline, commit message: BitchatMessage, to conversationID: ConversationID) -> Bool {
|
||||
guard !rejectedMessageIDs.contains(message.id) else { return false }
|
||||
committed.append((message, conversationID))
|
||||
return true
|
||||
}
|
||||
|
||||
func pipelinePrewarmMessage(_ pipeline: PublicMessagePipeline, message: BitchatMessage) {}
|
||||
|
||||
func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) {}
|
||||
func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) {
|
||||
batchingStates.append(isBatching)
|
||||
}
|
||||
}
|
||||
|
||||
@MainActor
|
||||
private func makeMessage(id: String, content: String, timestamp: Date) -> BitchatMessage {
|
||||
BitchatMessage(
|
||||
id: id,
|
||||
sender: "A",
|
||||
content: content,
|
||||
timestamp: timestamp,
|
||||
isRelay: false
|
||||
)
|
||||
}
|
||||
|
||||
struct PublicMessagePipelineTests {
|
||||
|
||||
@Test @MainActor
|
||||
func flush_sortsByTimestamp() async {
|
||||
func flush_commitsInTimestampOrder() async {
|
||||
let pipeline = PublicMessagePipeline()
|
||||
let delegate = TestPipelineDelegate()
|
||||
pipeline.delegate = delegate
|
||||
@@ -53,26 +75,13 @@ struct PublicMessagePipelineTests {
|
||||
let earlier = Date().addingTimeInterval(-10)
|
||||
let later = Date()
|
||||
|
||||
let messageA = BitchatMessage(
|
||||
id: "a",
|
||||
sender: "A",
|
||||
content: "Later",
|
||||
timestamp: later,
|
||||
isRelay: false
|
||||
)
|
||||
let messageB = BitchatMessage(
|
||||
id: "b",
|
||||
sender: "A",
|
||||
content: "Earlier",
|
||||
timestamp: earlier,
|
||||
isRelay: false
|
||||
)
|
||||
|
||||
pipeline.enqueue(messageA)
|
||||
pipeline.enqueue(messageB)
|
||||
pipeline.enqueue(makeMessage(id: "a", content: "Later", timestamp: later), to: .mesh)
|
||||
pipeline.enqueue(makeMessage(id: "b", content: "Earlier", timestamp: earlier), to: .mesh)
|
||||
pipeline.flushIfNeeded()
|
||||
|
||||
#expect(delegate.messages.map { $0.id } == ["b", "a"])
|
||||
#expect(delegate.messages(in: .mesh).map { $0.id } == ["b", "a"])
|
||||
// Batching state wrapped the flush.
|
||||
#expect(delegate.batchingStates == [true, false])
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
@@ -82,86 +91,41 @@ struct PublicMessagePipelineTests {
|
||||
pipeline.delegate = delegate
|
||||
|
||||
let now = Date()
|
||||
let messageA = BitchatMessage(
|
||||
id: "a",
|
||||
sender: "A",
|
||||
content: "Same",
|
||||
timestamp: now,
|
||||
isRelay: false
|
||||
)
|
||||
let messageB = BitchatMessage(
|
||||
id: "b",
|
||||
sender: "A",
|
||||
content: "Same",
|
||||
timestamp: now.addingTimeInterval(0.2),
|
||||
isRelay: false
|
||||
)
|
||||
|
||||
pipeline.enqueue(messageA)
|
||||
pipeline.enqueue(messageB)
|
||||
pipeline.enqueue(makeMessage(id: "a", content: "Same", timestamp: now), to: .mesh)
|
||||
pipeline.enqueue(makeMessage(id: "b", content: "Same", timestamp: now.addingTimeInterval(0.2)), to: .mesh)
|
||||
pipeline.flushIfNeeded()
|
||||
|
||||
#expect(delegate.messages.count == 1)
|
||||
#expect(delegate.messages.first?.content == "Same")
|
||||
#expect(delegate.messages(in: .mesh).count == 1)
|
||||
#expect(delegate.messages(in: .mesh).first?.content == "Same")
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func lateInsert_meshAppendsRecentOlderMessage() async {
|
||||
func flush_routesEachMessageToItsConversation() async {
|
||||
let pipeline = PublicMessagePipeline()
|
||||
let delegate = TestPipelineDelegate()
|
||||
pipeline.delegate = delegate
|
||||
pipeline.updateActiveChannel(.mesh)
|
||||
|
||||
let base = Date()
|
||||
let newer = BitchatMessage(
|
||||
id: "new",
|
||||
sender: "A",
|
||||
content: "New",
|
||||
timestamp: base,
|
||||
isRelay: false
|
||||
)
|
||||
let older = BitchatMessage(
|
||||
id: "old",
|
||||
sender: "A",
|
||||
content: "Old",
|
||||
timestamp: base.addingTimeInterval(-5),
|
||||
isRelay: false
|
||||
)
|
||||
|
||||
delegate.messages = [newer]
|
||||
pipeline.enqueue(older)
|
||||
pipeline.enqueue(makeMessage(id: "mesh-1", content: "mesh hello", timestamp: base), to: .mesh)
|
||||
// A channel switch mid-batch must not misroute already-buffered messages.
|
||||
pipeline.enqueue(makeMessage(id: "geo-1", content: "geo hello", timestamp: base.addingTimeInterval(1)), to: .geohash("u4pruydq"))
|
||||
pipeline.flushIfNeeded()
|
||||
|
||||
#expect(delegate.messages.map { $0.id } == ["new", "old"])
|
||||
#expect(delegate.messages(in: .mesh).map { $0.id } == ["mesh-1"])
|
||||
#expect(delegate.messages(in: .geohash("u4pruydq")).map { $0.id } == ["geo-1"])
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func lateInsert_locationInsertsByTimestamp() async {
|
||||
func flush_rejectedCommitDoesNotRecordContentKey() async {
|
||||
let pipeline = PublicMessagePipeline()
|
||||
let delegate = TestPipelineDelegate()
|
||||
pipeline.delegate = delegate
|
||||
pipeline.updateActiveChannel(.location(GeohashChannel(level: .city, geohash: "u4pruydq")))
|
||||
delegate.rejectedMessageIDs = ["dup"]
|
||||
|
||||
let base = Date()
|
||||
let newer = BitchatMessage(
|
||||
id: "new",
|
||||
sender: "A",
|
||||
content: "New",
|
||||
timestamp: base,
|
||||
isRelay: false
|
||||
)
|
||||
let older = BitchatMessage(
|
||||
id: "old",
|
||||
sender: "A",
|
||||
content: "Old",
|
||||
timestamp: base.addingTimeInterval(-5),
|
||||
isRelay: false
|
||||
)
|
||||
|
||||
delegate.messages = [newer]
|
||||
pipeline.enqueue(older)
|
||||
pipeline.enqueue(makeMessage(id: "dup", content: "already stored", timestamp: Date()), to: .mesh)
|
||||
pipeline.flushIfNeeded()
|
||||
|
||||
#expect(delegate.messages.map { $0.id } == ["old", "new"])
|
||||
#expect(delegate.messages(in: .mesh).isEmpty)
|
||||
#expect(delegate.recordedContentKeys.isEmpty)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,115 +0,0 @@
|
||||
import Foundation
|
||||
import BitFoundation
|
||||
import Testing
|
||||
@testable import bitchat
|
||||
|
||||
@Suite("PublicTimelineStore Tests")
|
||||
struct PublicTimelineStoreTests {
|
||||
|
||||
@Test("Mesh timeline deduplicates and trims to cap")
|
||||
func meshTimelineDeduplicatesAndTrims() {
|
||||
var store = PublicTimelineStore(meshCap: 2, geohashCap: 2)
|
||||
let first = TestHelpers.createTestMessage(content: "one")
|
||||
let second = TestHelpers.createTestMessage(content: "two")
|
||||
let third = TestHelpers.createTestMessage(content: "three")
|
||||
|
||||
store.append(first, to: .mesh)
|
||||
store.append(second, to: .mesh)
|
||||
store.append(first, to: .mesh)
|
||||
store.append(third, to: .mesh)
|
||||
|
||||
let messages = store.messages(for: .mesh)
|
||||
#expect(messages.map(\.content) == ["two", "three"])
|
||||
}
|
||||
|
||||
@Test("Timeline indexes allow trimmed message IDs to return")
|
||||
func timelineIndexesAllowTrimmedMessageIDsToReturn() {
|
||||
var store = PublicTimelineStore(meshCap: 2, geohashCap: 2)
|
||||
let first = timelineMessage(id: "one", content: "one", timestamp: 1)
|
||||
let second = timelineMessage(id: "two", content: "two", timestamp: 2)
|
||||
let third = timelineMessage(id: "three", content: "three", timestamp: 3)
|
||||
|
||||
store.append(first, to: .mesh)
|
||||
store.append(second, to: .mesh)
|
||||
store.append(third, to: .mesh)
|
||||
store.append(first, to: .mesh)
|
||||
|
||||
#expect(store.messages(for: .mesh).map(\.content) == ["three", "one"])
|
||||
|
||||
let geohash = "u4pruydq"
|
||||
let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash))
|
||||
let geoFirst = timelineMessage(id: "geo-one", content: "geo one", timestamp: 1)
|
||||
let geoSecond = timelineMessage(id: "geo-two", content: "geo two", timestamp: 2)
|
||||
let geoThird = timelineMessage(id: "geo-three", content: "geo three", timestamp: 3)
|
||||
|
||||
let didAppendGeoFirst = store.appendIfAbsent(geoFirst, toGeohash: geohash)
|
||||
let didAppendGeoSecond = store.appendIfAbsent(geoSecond, toGeohash: geohash)
|
||||
let didAppendGeoThird = store.appendIfAbsent(geoThird, toGeohash: geohash)
|
||||
let didReappendGeoFirst = store.appendIfAbsent(geoFirst, toGeohash: geohash)
|
||||
|
||||
#expect(didAppendGeoFirst)
|
||||
#expect(didAppendGeoSecond)
|
||||
#expect(didAppendGeoThird)
|
||||
#expect(didReappendGeoFirst)
|
||||
#expect(store.messages(for: channel).map(\.content) == ["geo one", "geo three"])
|
||||
}
|
||||
|
||||
@Test("Geohash appendIfAbsent remove and clear work together")
|
||||
func geohashStoreSupportsAppendRemoveAndClear() {
|
||||
var store = PublicTimelineStore(meshCap: 2, geohashCap: 3)
|
||||
let geohash = "u4pruydq"
|
||||
let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash))
|
||||
let first = TestHelpers.createTestMessage(content: "geo one")
|
||||
let second = TestHelpers.createTestMessage(content: "geo two")
|
||||
|
||||
let didAppendFirst = store.appendIfAbsent(first, toGeohash: geohash)
|
||||
let didAppendDuplicate = store.appendIfAbsent(first, toGeohash: geohash)
|
||||
|
||||
#expect(didAppendFirst)
|
||||
#expect(!didAppendDuplicate)
|
||||
store.append(second, toGeohash: geohash)
|
||||
let removed = store.removeMessage(withID: first.id)
|
||||
|
||||
#expect(removed?.id == first.id)
|
||||
#expect(store.messages(for: channel).map(\.content) == ["geo two"])
|
||||
|
||||
store.clear(channel: channel)
|
||||
#expect(store.messages(for: channel).isEmpty)
|
||||
}
|
||||
|
||||
@Test("Mutate geohash updates stored messages in place")
|
||||
func mutateGeohashAppliesTransformation() {
|
||||
var store = PublicTimelineStore(meshCap: 2, geohashCap: 3)
|
||||
let geohash = "u4pruydq"
|
||||
let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash))
|
||||
let first = TestHelpers.createTestMessage(content: "geo one")
|
||||
|
||||
store.append(first, toGeohash: geohash)
|
||||
store.mutateGeohash(geohash) { timeline in
|
||||
timeline.append(TestHelpers.createTestMessage(content: "geo two"))
|
||||
}
|
||||
|
||||
#expect(store.messages(for: channel).map(\.content) == ["geo one", "geo two"])
|
||||
}
|
||||
|
||||
@Test("Queued geohash system messages drain once")
|
||||
func pendingGeohashSystemMessagesDrainOnce() {
|
||||
var store = PublicTimelineStore(meshCap: 1, geohashCap: 1)
|
||||
|
||||
store.queueGeohashSystemMessage("first")
|
||||
store.queueGeohashSystemMessage("second")
|
||||
|
||||
#expect(store.drainPendingGeohashSystemMessages() == ["first", "second"])
|
||||
#expect(store.drainPendingGeohashSystemMessages().isEmpty)
|
||||
}
|
||||
|
||||
private func timelineMessage(id: String, content: String, timestamp: TimeInterval) -> BitchatMessage {
|
||||
BitchatMessage(
|
||||
id: id,
|
||||
sender: "alice",
|
||||
content: content,
|
||||
timestamp: Date(timeIntervalSince1970: timestamp),
|
||||
isRelay: false
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -153,6 +153,24 @@ extension ChatViewModel {
|
||||
}
|
||||
}
|
||||
|
||||
/// Test-only replacement for the deleted `messages` setter: seeds a
|
||||
/// public channel's conversation through the single-writer
|
||||
/// `ConversationStore` intents (upsert keeps re-seeding with updated
|
||||
/// copies working the way the old array assignment did).
|
||||
@MainActor
|
||||
func seedPublicMessages(_ messages: [BitchatMessage], for channel: ChannelID = .mesh) {
|
||||
for message in messages {
|
||||
conversations.upsertByID(message, in: ConversationID(channelID: channel))
|
||||
}
|
||||
}
|
||||
|
||||
/// Test-only replacement for `messages.removeAll()`: empties a public
|
||||
/// channel's conversation.
|
||||
@MainActor
|
||||
func clearPublicMessages(for channel: ChannelID = .mesh) {
|
||||
conversations.clear(ConversationID(channelID: channel))
|
||||
}
|
||||
|
||||
/// Test-only: drops every private chat and unread flag.
|
||||
@MainActor
|
||||
func clearAllPrivateChats() {
|
||||
|
||||
Reference in New Issue
Block a user