diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt index c584455e..faf67378 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt @@ -329,6 +329,16 @@ class BluetoothConnectionManager( ) } + suspend fun broadcastControlPacketAndAwaitAcceptance(routed: RoutedPacket): Boolean { + if (!isActive || !isBleTransportEnabled()) return false + + return packetBroadcaster.broadcastControlPacketAndAwaitAcceptance( + routed, + serverManager.getGattServer(), + serverManager.getCharacteristic() + ) + } + fun sendToPeer(peerID: String, routed: RoutedPacket): Boolean { if (!isActive || !isBleTransportEnabled()) return false return packetBroadcaster.sendToPeer( diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt index 37bcfffd..b17294f1 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt @@ -54,6 +54,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic private val peerManager = PeerManager() private val fragmentManager = FragmentManager() private val serviceScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val readReceiptRetrySender = RetryingControlPacketSender(serviceScope) private val authenticatedPeerStateStore = SecureAuthenticatedPeerStateStore(context) private val authenticatedPeerState by lazy { AuthenticatedPeerStateCoordinator( @@ -174,6 +175,11 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic connectionManager.broadcastPacket(packet) } + override suspend fun sendAndReport(packet: RoutedPacket): Boolean { + if (!isBleTransportEnabled()) return false + return connectionManager.broadcastControlPacketAndAwaitAcceptance(packet) + } + override fun sendToPeer(peerID: String, packet: BitchatPacket) { if (!isBleTransportEnabled()) return connectionManager.sendPacketToPeer(peerID, packet) @@ -187,6 +193,15 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic return true } + private suspend fun broadcastRoutedPacketAndReport(routed: RoutedPacket): Boolean { + if (!isBleTransportEnabled()) return false + val acceptedByBle = + connectionManager.broadcastControlPacketAndAwaitAcceptance(routed) + val acceptedByBridgedTransport = + TransportBridgeService.broadcastAndReport("BLE", routed) + return acceptedByBle || acceptedByBridgedTransport + } + private fun isBleTransportEnabled(): Boolean { return try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().bleEnabled.value @@ -509,10 +524,24 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic } override fun onDeliveryAckReceived(messageID: String, peerID: String) { + // Status events can arrive while MainActivity has detached the UI delegate. + // Persist first so the next UI collector observes the advancement. + try { + com.bitchat.android.services.AppStateStore.updatePrivateMessageStatus( + messageID, + com.bitchat.android.model.DeliveryStatus.Delivered(peerID, Date()) + ) + } catch (_: Exception) { } delegate?.didReceiveDeliveryAck(messageID, peerID) } override fun onReadReceiptReceived(messageID: String, peerID: String) { + try { + com.bitchat.android.services.AppStateStore.updatePrivateMessageStatus( + messageID, + com.bitchat.android.model.DeliveryStatus.Read(peerID, Date()) + ) + } catch (_: Exception) { } delegate?.didReceiveReadReceipt(messageID, peerID) } @@ -1077,12 +1106,6 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic } try { - // Avoid duplicate read receipts: check persistent store first - val seenStore = try { com.bitchat.android.services.SeenMessageStore.getInstance(context.applicationContext) } catch (_: Exception) { null } - if (seenStore?.hasRead(messageID) == true) { - return@launch - } - // Create read receipt payload using NoisePayloadType exactly like iOS val readReceiptPayload = com.bitchat.android.model.NoisePayload( type = com.bitchat.android.model.NoisePayloadType.READ_RECEIPT, @@ -1106,10 +1129,31 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic // Sign the packet before broadcasting val signedPacket = signPacketBeforeBroadcast(packet) - broadcastRoutedPacket(RoutedPacket(signedPacket)) - - // Persist as read after successful send - try { seenStore?.markRead(messageID) } catch (_: Exception) { } + val retryKey = "$recipientPeerID:$messageID" + readReceiptRetrySender.enqueue( + key = retryKey, + sendAttempt = { attempt -> + // Keep the addressed packet on the normal broadcaster actor so receipt + // attempts are ordered with other BLE traffic and can use mesh routing. + val accepted = + broadcastRoutedPacketAndReport(RoutedPacket(signedPacket)) + Log.d( + TAG, + "Read receipt attempt $attempt accepted=$accepted " + + "peer=${recipientPeerID.take(8)} message=${messageID.take(8)}" + ) + accepted + }, + onComplete = { accepted -> + if (accepted) { + try { + com.bitchat.android.services.SeenMessageStore + .getInstance(context.applicationContext) + .markReadReceiptSent(messageID) + } catch (_: Exception) { } + } + } + ) } catch (e: Exception) { Log.e(TAG, "Failed to send read receipt to $recipientPeerID: ${e.message}") diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt index 0e9b0c76..1ec1baea 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt @@ -10,6 +10,8 @@ import com.bitchat.android.protocol.SpecialRecipients import com.bitchat.android.model.RoutedPacket import com.bitchat.android.protocol.MessageType import com.bitchat.android.util.toHexString +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob @@ -110,7 +112,8 @@ class BluetoothPacketBroadcaster( private data class BroadcastRequest( val routed: RoutedPacket, val gattServer: BluetoothGattServer?, - val characteristic: BluetoothGattCharacteristic? + val characteristic: BluetoothGattCharacteristic?, + val accepted: CompletableDeferred? = null ) // Actor scope for the broadcaster @@ -123,7 +126,17 @@ class BluetoothPacketBroadcaster( capacity = Channel.UNLIMITED ) { for (request in channel) { - broadcastSinglePacketInternal(request.routed, request.gattServer, request.characteristic) + val accepted = try { + broadcastSinglePacketInternal( + request.routed, + request.gattServer, + request.characteristic + ) + } catch (e: Exception) { + Log.w(TAG, "Broadcast request failed: ${e.message}") + false + } + request.accepted?.complete(accepted) } } @@ -242,6 +255,29 @@ class BluetoothPacketBroadcaster( } } + /** + * Serializes a small control packet with normal BLE traffic and waits for the platform write + * API to accept at least one notification/write. + */ + suspend fun broadcastControlPacketAndAwaitAcceptance( + routed: RoutedPacket, + gattServer: BluetoothGattServer?, + characteristic: BluetoothGattCharacteristic? + ): Boolean { + val accepted = CompletableDeferred() + return try { + broadcasterActor.send( + BroadcastRequest(routed, gattServer, characteristic, accepted) + ) + accepted.await() + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + Log.w(TAG, "Failed to queue control packet: ${e.message}") + broadcastSinglePacketInternal(routed, gattServer, characteristic) + } + } + /** * Targeted send to a specific peer (by peerID) if directly connected. * Returns true if sent to at least one matching connection. @@ -274,11 +310,11 @@ class BluetoothPacketBroadcaster( routed: RoutedPacket, gattServer: BluetoothGattServer?, characteristic: BluetoothGattCharacteristic? - ) { + ): Boolean { val packet = routed.packet // iOS-compatible: Use selective padding policy for BLE val padForBLE = BLEPacketPaddingPolicy.shouldPadForBLE(packet.type) - val data = packet.toBinaryData(padding = padForBLE) ?: return + val data = packet.toBinaryData(padding = padForBLE) ?: return false val typeName = MessageType.fromValue(packet.type)?.name ?: packet.type.toString() val senderPeerID = routed.peerID ?: packet.senderID.toHexString() val incomingAddr = routed.relayAddress @@ -320,7 +356,7 @@ class BluetoothPacketBroadcaster( } } - if (sent) return + if (sent) return true Log.d(TAG, "Source Routing: First hop $firstHop not connected. Falling back to standard broadcast logic.") } @@ -337,7 +373,7 @@ class BluetoothPacketBroadcaster( if (notifyDevice(targetDevice, data, gattServer, characteristic)) { val toPeer = connectionTracker.addressPeerMap[targetDevice.address] logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, toPeer, targetDevice.address, packet.ttl, packet.version, routeInfo) - return // Sent, no need to continue + return true } } @@ -350,7 +386,7 @@ class BluetoothPacketBroadcaster( if (writeToDeviceConn(targetDeviceConn, data)) { val toPeer = connectionTracker.addressPeerMap[targetDeviceConn.device.address] logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, toPeer, targetDeviceConn.device.address, packet.ttl, packet.version, routeInfo) - return // Sent, no need to continue + return true } } } @@ -360,6 +396,7 @@ class BluetoothPacketBroadcaster( val connectedDevices = connectionTracker.getConnectedDevices() val senderID = packet.senderID.toHexString() + var accepted = false // Send to server connections (devices connected to our GATT server) subscribedDevices.forEach { device -> @@ -371,6 +408,7 @@ class BluetoothPacketBroadcaster( } val sent = notifyDevice(device, data, gattServer, characteristic) if (sent) { + accepted = true val toPeer = connectionTracker.addressPeerMap[device.address] logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, toPeer, device.address, packet.ttl, packet.version, routeInfo) } @@ -387,11 +425,13 @@ class BluetoothPacketBroadcaster( } val sent = writeToDeviceConn(deviceConn, data) if (sent) { + accepted = true val toPeer = connectionTracker.addressPeerMap[deviceConn.device.address] logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, toPeer, deviceConn.device.address, packet.ttl, packet.version, routeInfo) } } } + return accepted } /** diff --git a/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt b/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt index 6fbddacc..bd7e08aa 100644 --- a/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt +++ b/app/src/main/java/com/bitchat/android/mesh/MeshCore.kt @@ -52,6 +52,7 @@ class MeshCore( private val peerManager = PeerManager() val fragmentManager = FragmentManager() + private val readReceiptRetrySender = RetryingControlPacketSender(scope) private val authenticatedPeerStateStore = SecureAuthenticatedPeerStateStore(context) private val authenticatedPeerState by lazy { AuthenticatedPeerStateCoordinator( @@ -191,11 +192,22 @@ class MeshCore( transport.broadcastPacket(packet) } + fun sendFromBridgeAndReport(packet: RoutedPacket): Boolean { + return transport.broadcastPacket(packet) + } + private fun dispatchGlobal(routed: RoutedPacket) { transport.broadcastPacket(routed) TransportBridgeService.broadcast(transport.id, routed) } + private suspend fun dispatchGlobalAndReport(routed: RoutedPacket): Boolean { + val acceptedByLocalTransport = transport.broadcastPacket(routed) + val acceptedByBridgedTransport = + TransportBridgeService.broadcastAndReport(transport.id, routed) + return acceptedByLocalTransport || acceptedByBridgedTransport + } + private fun startPeriodicBroadcastAnnounce() { announceJob?.cancel() announceJob = scope.launch { @@ -406,10 +418,22 @@ class MeshCore( } override fun onDeliveryAckReceived(messageID: String, peerID: String) { + try { + com.bitchat.android.services.AppStateStore.updatePrivateMessageStatus( + messageID, + com.bitchat.android.model.DeliveryStatus.Delivered(peerID, java.util.Date()) + ) + } catch (_: Exception) { } delegate?.didReceiveDeliveryAck(messageID, peerID) } override fun onReadReceiptReceived(messageID: String, peerID: String) { + try { + com.bitchat.android.services.AppStateStore.updatePrivateMessageStatus( + messageID, + com.bitchat.android.model.DeliveryStatus.Read(peerID, java.util.Date()) + ) + } catch (_: Exception) { } delegate?.didReceiveReadReceipt(messageID, peerID) } @@ -699,8 +723,24 @@ class MeshCore( signature = null, ttl = maxTtl ) - dispatchGlobal(RoutedPacket(signPacketBeforeBroadcast(packet))) - hooks.onReadReceiptSent?.invoke(messageID) + val signedPacket = signPacketBeforeBroadcast(packet) + val retryKey = "$recipientPeerID:$messageID" + readReceiptRetrySender.enqueue( + key = retryKey, + sendAttempt = { + dispatchGlobalAndReport(RoutedPacket(signedPacket)) + }, + onComplete = { accepted -> + if (accepted) { + try { + com.bitchat.android.services.SeenMessageStore + .getInstance(context.applicationContext) + .markReadReceiptSent(messageID) + } catch (_: Exception) { } + hooks.onReadReceiptSent?.invoke(messageID) + } + } + ) } catch (e: Exception) { Log.e("MeshCore", "Failed to send read receipt: ${e.message}") } diff --git a/app/src/main/java/com/bitchat/android/mesh/MeshTransport.kt b/app/src/main/java/com/bitchat/android/mesh/MeshTransport.kt index b3f2fcc4..354fcd7e 100644 --- a/app/src/main/java/com/bitchat/android/mesh/MeshTransport.kt +++ b/app/src/main/java/com/bitchat/android/mesh/MeshTransport.kt @@ -9,7 +9,10 @@ import com.bitchat.android.protocol.BitchatPacket interface MeshTransport { val id: String - fun broadcastPacket(routed: RoutedPacket) + /** + * Broadcasts a packet and reports whether at least one concrete transport write was accepted. + */ + fun broadcastPacket(routed: RoutedPacket): Boolean fun sendPacketToPeer(peerID: String, packet: BitchatPacket): Boolean diff --git a/app/src/main/java/com/bitchat/android/mesh/RetryingControlPacketSender.kt b/app/src/main/java/com/bitchat/android/mesh/RetryingControlPacketSender.kt new file mode 100644 index 00000000..ba81800c --- /dev/null +++ b/app/src/main/java/com/bitchat/android/mesh/RetryingControlPacketSender.kt @@ -0,0 +1,82 @@ +package com.bitchat.android.mesh + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * Sends small idempotent control packets redundantly while serializing the attempts submitted + * through this sender. Android BLE only reports that a GATT write/notification was accepted; + * it does not prove that the remote application processed the packet. Reusing the exact encoded + * packet makes retries safe: the receiver's packet/replay protection drops copies it already saw. + */ +internal class RetryingControlPacketSender( + private val scope: CoroutineScope, + private val maxAttempts: Int = 3, + private val retryDelayMs: Long = 750L, + private val interSendDelayMs: Long = 75L +) { + private val sendMutex = Mutex() + private val jobsLock = Any() + private val activeJobs = mutableMapOf() + + init { + require(maxAttempts > 0) + require(retryDelayMs >= 0) + require(interSendDelayMs >= 0) + } + + /** + * Coalesces concurrent requests for the same logical packet. Once the retry window finishes, + * a later user action may enqueue the packet again; duplicate receipt processing is idempotent. + */ + fun enqueue( + key: String, + sendAttempt: suspend (attempt: Int) -> Boolean, + onComplete: (acceptedAtLeastOnce: Boolean) -> Unit = {} + ) { + val job = synchronized(jobsLock) { + if (activeJobs[key]?.isActive == true) return + + scope.launch(start = CoroutineStart.LAZY) { + var acceptedAtLeastOnce = false + var completed = false + try { + repeat(maxAttempts) { index -> + if (!isActive) return@launch + val attempt = index + 1 + sendMutex.withLock { + try { + val accepted = try { + sendAttempt(attempt) + } catch (_: Exception) { + false + } + acceptedAtLeastOnce = accepted || acceptedAtLeastOnce + } finally { + if (interSendDelayMs > 0) delay(interSendDelayMs) + } + } + if (attempt < maxAttempts && retryDelayMs > 0) { + delay(retryDelayMs) + } + } + completed = true + } finally { + synchronized(jobsLock) { + activeJobs.remove(key) + } + if (completed) { + try { onComplete(acceptedAtLeastOnce) } catch (_: Exception) { } + } + } + }.also { activeJobs[key] = it } + } + job.start() + } +} diff --git a/app/src/main/java/com/bitchat/android/nostr/NostrDirectMessageHandler.kt b/app/src/main/java/com/bitchat/android/nostr/NostrDirectMessageHandler.kt index 469972c8..fce95c2d 100644 --- a/app/src/main/java/com/bitchat/android/nostr/NostrDirectMessageHandler.kt +++ b/app/src/main/java/com/bitchat/android/nostr/NostrDirectMessageHandler.kt @@ -156,7 +156,7 @@ class NostrDirectMessageHandler( ) val isViewing = state.getSelectedPrivateChatPeerValue() == conversationID - val suppressUnread = seenStore.hasRead(pm.messageID) + val suppressUnread = seenStore.hasBeenReadLocally(pm.messageID) withContext(Dispatchers.Main) { privateChatManager.handleIncomingPrivateMessage( @@ -175,7 +175,8 @@ class NostrDirectMessageHandler( if (isViewing && !suppressUnread) { val nostrTransport = NostrTransport.getInstance(application) nostrTransport.sendReadReceiptGeohash(pm.messageID, senderPubkey, recipientIdentity) - seenStore.markRead(pm.messageID) + seenStore.markReadLocally(pm.messageID) + seenStore.markReadReceiptSent(pm.messageID) } } NoisePayloadType.DELIVERED -> { diff --git a/app/src/main/java/com/bitchat/android/service/TransportBridgeService.kt b/app/src/main/java/com/bitchat/android/service/TransportBridgeService.kt index f8b10766..67a7a1d8 100644 --- a/app/src/main/java/com/bitchat/android/service/TransportBridgeService.kt +++ b/app/src/main/java/com/bitchat/android/service/TransportBridgeService.kt @@ -4,6 +4,7 @@ import android.util.Log import com.bitchat.android.model.RoutedPacket import com.bitchat.android.protocol.BitchatPacket import com.bitchat.android.util.toHexString +import kotlinx.coroutines.CancellationException import java.security.MessageDigest import java.util.Collections import java.util.LinkedHashMap @@ -31,6 +32,14 @@ object TransportBridgeService { */ fun send(packet: RoutedPacket) + /** + * Send a packet and report whether at least one concrete transport write was accepted. + * + * Receipt retries use this path so a registered-but-disconnected transport cannot be + * mistaken for a successful send. + */ + suspend fun sendAndReport(packet: RoutedPacket): Boolean = false + /** * Send a packet to a specific peer via this transport (optional). */ @@ -45,6 +54,11 @@ object TransportBridgeService { } } ) + private data class PreparedForward( + val packet: BitchatPacket, + val seenKey: String, + val reservedAtMs: Long + ) /** * Register a transport layer to receive bridged packets. @@ -74,7 +88,8 @@ object TransportBridgeService { fun broadcast(sourceId: String, packet: RoutedPacket) { val targets = transports.filterKeys { it != sourceId } if (targets.isEmpty()) return - val forwardedPacket = prepareForwardedPacket("broadcast", packet.packet) ?: return + val prepared = prepareForwardedPacket("broadcast", packet.packet) ?: return + val forwardedPacket = prepared.packet // Prepared private-media fragments must remain the admitted plan when // crossing transports, but relay TTL still has to advance on every // hop. TTL is excluded from the signature and does not affect size. @@ -96,13 +111,51 @@ object TransportBridgeService { } } + /** + * Broadcasts through every other active transport and reports whether any concrete write was + * accepted. Failed attempts release their duplicate-suppression reservation so a later retry + * can use a transport that reconnects during the retry window. + */ + suspend fun broadcastAndReport(sourceId: String, packet: RoutedPacket): Boolean { + val targets = transports.filterKeys { it != sourceId } + if (targets.isEmpty()) return false + val kind = "broadcast" + val prepared = prepareForwardedPacket(kind, packet.packet) ?: return false + val forwardedPacket = prepared.packet + val forwarded = packet.copy( + packet = forwardedPacket, + preparedPackets = packet.preparedPackets?.map { prepared -> + prepared.copy(ttl = forwardedPacket.ttl) + } + ) + + var accepted = false + targets.forEach { (id, layer) -> + val targetAccepted = try { + layer.sendAndReport(forwarded) + } catch (e: CancellationException) { + releaseSeenPacket(prepared) + throw e + } catch (e: Exception) { + Log.e(TAG, "Failed to bridge packet to $id: ${e.message}") + false + } + accepted = targetAccepted || accepted + } + if (!accepted) { + releaseSeenPacket(prepared) + } + return accepted + } + /** * Send a packet to a specific peer across all other transports. */ fun sendToPeer(sourceId: String, peerID: String, packet: BitchatPacket) { val targets = transports.filterKeys { it != sourceId } if (targets.isEmpty()) return - val forwardedPacket = prepareForwardedPacket("peer:$peerID", packet) ?: return + val forwardedPacket = + prepareForwardedPacket("peer:$peerID", packet)?.packet ?: return targets.forEach { (id, layer) -> try { @@ -147,7 +200,7 @@ object TransportBridgeService { } } - private fun prepareForwardedPacket(kind: String, packet: BitchatPacket): BitchatPacket? { + private fun prepareForwardedPacket(kind: String, packet: BitchatPacket): PreparedForward? { if (packet.ttl == 0u.toUByte()) { Log.d(TAG, "Dropping bridged packet type ${packet.type}: TTL expired") return null @@ -165,7 +218,19 @@ object TransportBridgeService { seenPackets[key] = now } - return packet.copy(ttl = (packet.ttl - 1u).toUByte()) + return PreparedForward( + packet = packet.copy(ttl = (packet.ttl - 1u).toUByte()), + seenKey = key, + reservedAtMs = now + ) + } + + private fun releaseSeenPacket(prepared: PreparedForward) { + synchronized(seenPackets) { + if (seenPackets[prepared.seenKey] == prepared.reservedAtMs) { + seenPackets.remove(prepared.seenKey) + } + } } private fun pruneSeen(now: Long) { diff --git a/app/src/main/java/com/bitchat/android/services/SeenMessageStore.kt b/app/src/main/java/com/bitchat/android/services/SeenMessageStore.kt index 15e1c7ca..3b531846 100644 --- a/app/src/main/java/com/bitchat/android/services/SeenMessageStore.kt +++ b/app/src/main/java/com/bitchat/android/services/SeenMessageStore.kt @@ -4,9 +4,14 @@ import android.content.Context import android.util.Log import com.bitchat.android.identity.SecureIdentityStateManager import com.google.gson.Gson +import com.google.gson.annotations.SerializedName /** - * Persistent store for message IDs we've already acknowledged (DELIVERED) or READ. + * Persistent store for message IDs we've already acknowledged as delivered, read locally, or + * admitted to a completed read-receipt send window. + * + * Local read state must not be used as proof that a read-receipt packet reached the sender. + * Transport delivery is best-effort and retryable, while local read state drives unread UI. * Limits to last MAX_IDS entries per set to avoid memory bloat. */ class SeenMessageStore private constructor(private val context: Context) { @@ -27,12 +32,14 @@ class SeenMessageStore private constructor(private val context: Context) { private val secure = SecureIdentityStateManager(context) private val delivered = LinkedHashSet(MAX_IDS) - private val read = LinkedHashSet(MAX_IDS) + private val locallyRead = LinkedHashSet(MAX_IDS) + private val readReceiptsSent = LinkedHashSet(MAX_IDS) init { load() } @Synchronized fun hasDelivered(id: String) = delivered.contains(id) - @Synchronized fun hasRead(id: String) = read.contains(id) + @Synchronized fun hasBeenReadLocally(id: String) = locallyRead.contains(id) + @Synchronized fun hasReadReceiptBeenSent(id: String) = readReceiptsSent.contains(id) @Synchronized fun markDelivered(id: String) { if (delivered.remove(id)) delivered.add(id) else { @@ -42,17 +49,26 @@ class SeenMessageStore private constructor(private val context: Context) { persist() } - @Synchronized fun markRead(id: String) { - if (read.remove(id)) read.add(id) else { - read.add(id) - trim(read) + @Synchronized fun markReadLocally(id: String) { + if (locallyRead.remove(id)) locallyRead.add(id) else { + locallyRead.add(id) + trim(locallyRead) + } + persist() + } + + @Synchronized fun markReadReceiptSent(id: String) { + if (readReceiptsSent.remove(id)) readReceiptsSent.add(id) else { + readReceiptsSent.add(id) + trim(readReceiptsSent) } persist() } @Synchronized fun clear() { delivered.clear() - read.clear() + locallyRead.clear() + readReceiptsSent.clear() persist() } @@ -68,10 +84,19 @@ class SeenMessageStore private constructor(private val context: Context) { try { val json = secure.getSecureValue(STORAGE_KEY) ?: return val data = gson.fromJson(json, StorePayload::class.java) ?: return - delivered.clear(); read.clear() + delivered.clear(); locallyRead.clear(); readReceiptsSent.clear() data.delivered.takeLast(MAX_IDS).forEach { delivered.add(it) } - data.read.takeLast(MAX_IDS).forEach { read.add(it) } - Log.d(TAG, "Loaded delivered=${delivered.size}, read=${read.size}") + data.locallyRead.takeLast(MAX_IDS).forEach { locallyRead.add(it) } + // Older payloads used the local-read set to suppress receipt sends. Seed the new + // explicit set once during migration to avoid replaying an entire chat history. + (data.readReceiptsSent ?: data.locallyRead) + .takeLast(MAX_IDS) + .forEach { readReceiptsSent.add(it) } + Log.d( + TAG, + "Loaded delivered=${delivered.size}, locallyRead=${locallyRead.size}, " + + "readReceiptsSent=${readReceiptsSent.size}" + ) } catch (e: Exception) { Log.e(TAG, "Failed to load SeenMessageStore: ${e.message}") } @@ -79,7 +104,11 @@ class SeenMessageStore private constructor(private val context: Context) { @Synchronized private fun persist() { try { - val payload = StorePayload(delivered.toList(), read.toList()) + val payload = StorePayload( + delivered = delivered.toList(), + locallyRead = locallyRead.toList(), + readReceiptsSent = readReceiptsSent.toList() + ) val json = gson.toJson(payload) secure.storeSecureValue(STORAGE_KEY, json) } catch (e: Exception) { @@ -89,6 +118,10 @@ class SeenMessageStore private constructor(private val context: Context) { private data class StorePayload( val delivered: List = emptyList(), - val read: List = emptyList() + // Keep the existing JSON field name for backward-compatible secure-store migration. + @SerializedName("read") + val locallyRead: List = emptyList(), + @SerializedName("read_receipts_sent") + val readReceiptsSent: List? = null ) } diff --git a/app/src/main/java/com/bitchat/android/ui/ChatViewModel.kt b/app/src/main/java/com/bitchat/android/ui/ChatViewModel.kt index 99a83f6e..a684e889 100644 --- a/app/src/main/java/com/bitchat/android/ui/ChatViewModel.kt +++ b/app/src/main/java/com/bitchat/android/ui/ChatViewModel.kt @@ -106,6 +106,9 @@ class ChatViewModel( // Specialized managers private val dataManager = DataManager(application.applicationContext) private val identityManager by lazy { SecureIdentityStateManager(getApplication()) } + private val seenMessageStore by lazy { + com.bitchat.android.services.SeenMessageStore.getInstance(getApplication()) + } private val messageManager = MessageManager(state) private val channelManager = ChannelManager(state, messageManager, dataManager, viewModelScope) @@ -116,7 +119,18 @@ class ChatViewModel( override fun getMyPeerID(): String = mesh.myPeerID } - val privateChatManager = PrivateChatManager(state, messageManager, dataManager, noiseSessionDelegate) + val privateChatManager = PrivateChatManager( + state, + messageManager, + dataManager, + noiseSessionDelegate, + hasReadReceiptBeenSent = { messageID -> + seenMessageStore.hasReadReceiptBeenSent(messageID) + }, + markMessageReadLocally = { messageID -> + seenMessageStore.markReadLocally(messageID) + } + ) private val commandProcessor = CommandProcessor(state, messageManager, channelManager, privateChatManager) private val notificationManager = NotificationManager( application.applicationContext, @@ -153,7 +167,10 @@ class ChatViewModel( coroutineScope = viewModelScope, onHapticFeedback = { ChatViewModelUtils.triggerHapticFeedback(application.applicationContext) }, getMyPeerID = { mesh.myPeerID }, - getMeshService = { mesh } + getMeshService = { mesh }, + markMessageReadLocally = { messageID -> + seenMessageStore.markReadLocally(messageID) + } ) // New Geohash architecture ViewModel (replaces God object service usage in UI path) @@ -184,14 +201,14 @@ class ChatViewModel( state.nickname, state.connectedPeers ) { unreadConversationIDs, chats, currentNickname, connectedPeerIDs -> - val seenStore = com.bitchat.android.services.SeenMessageStore.getInstance(getApplication()) + val seenStore = seenMessageStore val connectedPeerIDSet = connectedPeerIDs.mapTo(mutableSetOf()) { it.lowercase() } buildUnreadConversationSummaries( unreadConversationIDs = unreadConversationIDs, privateChats = chats, currentUserIdentifiers = setOf(currentNickname, mesh.myPeerID), canonicalize = ContactDirectory::canonicalConversationId, - isMessageRead = { message -> seenStore.hasRead(message.id) } + isMessageRead = { message -> seenStore.hasBeenReadLocally(message.id) } ).map { summary -> val resolution = ContactDirectory.resolve(summary.conversationID) val resolvedNostrPubkey = summary.nostrPubkey @@ -297,15 +314,13 @@ class ChatViewModel( val (canonicalChats, unreadConversationIDs) = withContext(Dispatchers.IO) { val canonical = ContactDirectory.canonicalizePrivateChats(byPeer) val unread = try { - val seen = com.bitchat.android.services.SeenMessageStore - .getInstance(getApplication()) val myNick = state.getNicknameValue().ifBlank { mesh.myPeerID } canonical .filterValues { messages -> messages.any { message -> message.sender != myNick && message.sender != "system" && - !seen.hasRead(message.id) + !seenMessageStore.hasBeenReadLocally(message.id) } } .keys @@ -463,42 +478,24 @@ class ChatViewModel( ensureGeohashDMSubscriptionIfNeeded(peerID) } - val (conversationID, unreadAliases) = withContext(Dispatchers.IO) { + val (conversationID, success) = withContext(Dispatchers.IO) { val canonicalID = ContactDirectory.canonicalConversationId(peerID) - canonicalID to matchingUnreadAliases( + val unreadAliases = matchingUnreadAliases( unreadConversationIDs = state.getUnreadPrivateMessagesValue(), canonicalConversationID = canonicalID, canonicalize = ContactDirectory::canonicalConversationId ) + canonicalID to privateChatManager.startPrivateChat( + peerID = canonicalID, + meshService = mesh, + unreadAliases = unreadAliases + ) } - val success = privateChatManager.startPrivateChat( - peerID = conversationID, - meshService = mesh, - unreadAliases = unreadAliases - ) if (success) { // Notify notification manager about current private chat setCurrentPrivateChatPeer(conversationID) // Clear notifications for this sender since user is now viewing the chat clearNotificationsForSender(conversationID) - - // Persistently mark all messages in this conversation as read so Nostr fetches - // after app restarts won't re-mark them as unread. - withContext(Dispatchers.IO) { - try { - val seen = - com.bitchat.android.services.SeenMessageStore.getInstance(getApplication()) - val chats = state.getPrivateChatsValue() - val messages = chats[conversationID] ?: emptyList() - messages.forEach { msg -> - try { - seen.markRead(msg.id) - } catch (_: Exception) { - } - } - } catch (_: Exception) { - } - } } } diff --git a/app/src/main/java/com/bitchat/android/ui/MeshDelegateHandler.kt b/app/src/main/java/com/bitchat/android/ui/MeshDelegateHandler.kt index f98c5c87..de2f12c5 100644 --- a/app/src/main/java/com/bitchat/android/ui/MeshDelegateHandler.kt +++ b/app/src/main/java/com/bitchat/android/ui/MeshDelegateHandler.kt @@ -22,7 +22,8 @@ class MeshDelegateHandler( private val coroutineScope: CoroutineScope, private val onHapticFeedback: () -> Unit, private val getMyPeerID: () -> String, - private val getMeshService: () -> MeshService + private val getMeshService: () -> MeshService, + private val markMessageReadLocally: (messageID: String) -> Unit = {} ) : BluetoothMeshDelegate { override fun didReceiveMessage(message: BitchatMessage) { @@ -247,46 +248,44 @@ class MeshDelegateHandler( val shouldSendReadReceipt = !isAppInBackground && senderConversationID != null && focusedConversationID == senderConversationID - - if (shouldSendReadReceipt) { - android.util.Log.d( - "MeshDelegateHandler", - "Sending reactive read receipt for focused chat with $senderConversationID (message=${message.id})" - ) - val nickname = state.getNicknameValue() ?: "unknown" - val mesh = getMeshService() - val sent = try { - val meshPeerID = senderConversationID - ?.let { ContactDirectory.resolve(it).meshPeerID } - ?: senderPeerID?.takeIf { - com.bitchat.android.services.ContactIdentityResolver.isMeshPeerId(it) - } - if (meshPeerID != null && - mesh.getPeerInfo(meshPeerID)?.isConnected == true && - mesh.hasEstablishedSession(meshPeerID) - ) { - mesh.sendReadReceipt(message.id, meshPeerID, nickname) - true - } else { - false + + if (shouldSendReadReceipt) { + android.util.Log.d( + "MeshDelegateHandler", + "Sending reactive read receipt for focused chat with $senderConversationID (message=${message.id})" + ) + // UI focus is the source of truth for local read state. Transport acceptance is a + // separate fact and may remain retryable when the peer disconnects. + try { markMessageReadLocally(message.id) } catch (_: Exception) { } + + val nickname = state.getNicknameValue().ifBlank { "unknown" } + val mesh = getMeshService() + try { + val meshPeerID = ContactDirectory.resolve(senderConversationID).meshPeerID + ?: senderPeerID.takeIf { + com.bitchat.android.services.ContactIdentityResolver.isMeshPeerId(it) } - } catch (_: Exception) { - false + if (meshPeerID != null && + mesh.getPeerInfo(meshPeerID)?.isConnected == true && + mesh.hasEstablishedSession(meshPeerID) + ) { + mesh.sendReadReceipt(message.id, meshPeerID, nickname) } - if (sent) { - // Ensure unread badge is cleared for this peer immediately - try { - val current = state.getUnreadPrivateMessagesValue().toMutableSet() - val changed = current.remove(senderPeerID) or current.remove(senderConversationID) - if (changed) { - state.setUnreadPrivateMessages(current) - } - } catch (_: Exception) { } + } catch (_: Exception) { } + + // Ensure unread badge is cleared for this peer immediately. + try { + val current = state.getUnreadPrivateMessagesValue().toMutableSet() + val changed = current.remove(senderPeerID) or current.remove(senderConversationID) + if (changed) { + state.setUnreadPrivateMessages(current) } - } else { - android.util.Log.d("MeshDelegateHandler", "Skipping read receipt - chat not focused (background: $isAppInBackground, current peer: $currentPrivateChatPeer, sender: $senderPeerID)") - } + } catch (_: Exception) { } + } else { + android.util.Log.d("MeshDelegateHandler", "Skipping read receipt - chat not focused (background: $isAppInBackground, current peer: $currentPrivateChatPeer, sender: $senderPeerID)") } + } + /** * Expose mesh peer info for components that need to resolve identities (e.g., Nostr mapping) */ diff --git a/app/src/main/java/com/bitchat/android/ui/PrivateChatManager.kt b/app/src/main/java/com/bitchat/android/ui/PrivateChatManager.kt index 8d7c5e1d..bd1b51bf 100644 --- a/app/src/main/java/com/bitchat/android/ui/PrivateChatManager.kt +++ b/app/src/main/java/com/bitchat/android/ui/PrivateChatManager.kt @@ -33,7 +33,9 @@ class PrivateChatManager( private val state: ChatState, private val messageManager: MessageManager, private val dataManager: DataManager, - private val noiseSessionDelegate: NoiseSessionDelegate + private val noiseSessionDelegate: NoiseSessionDelegate, + private val hasReadReceiptBeenSent: (messageID: String) -> Boolean = { false }, + private val markMessageReadLocally: (messageID: String) -> Unit = {} ) { companion object { @@ -402,7 +404,14 @@ class PrivateChatManager( senderPeerID == meshPeerID || ContactDirectory.canonicalConversationId(senderPeerID) == canonicalConversationID ) - if (isFromTarget && meshPeerID != null) { + if (isFromTarget) { + try { + markMessageReadLocally(msg.id) + } catch (e: Exception) { + Log.w(TAG, "Failed to persist local read for message ${msg.id}: ${e.message}") + } + } + if (isFromTarget && meshPeerID != null && !hasReadReceiptBeenSent(msg.id)) { try { if (hasMesh) { meshService.sendReadReceipt(msg.id, meshPeerID, myNickname) diff --git a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt index c5014456..d2be00e2 100644 --- a/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt +++ b/app/src/main/java/com/bitchat/android/wifi-aware/WifiAwareMeshService.kt @@ -222,14 +222,17 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor /** * Broadcasts raw bytes to currently connected peer. */ - private fun broadcastRaw(bytes: ByteArray) { + private fun broadcastRaw(bytes: ByteArray): Boolean { + var accepted = false connectionTracker.peerSockets.forEach { (pid, sock) -> try { sock.write(bytes) + accepted = true } catch (e: IOException) { Log.e(TAG, "TX: write failed to ${pid.take(8)}: ${e.message}") } } + return accepted } // TransportLayer implementation @@ -239,6 +242,10 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor meshCore.sendFromBridge(packet) } + override suspend fun sendAndReport(packet: RoutedPacket): Boolean { + return meshCore.sendFromBridgeAndReport(packet) + } + override fun sendToPeer(peerID: String, packet: BitchatPacket) { sendPacketToPeer(peerID, packet) } @@ -246,23 +253,23 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor /** * Broadcasts routed packet to currently connected peers. */ - private fun broadcastPacket(routed: RoutedPacket) { + private fun broadcastPacket(routed: RoutedPacket): Boolean { val packet = routed.packet if (packet.senderID.toHexString() == myPeerID && !packet.route.isNullOrEmpty()) { val firstHop = packet.route!![0].toHexString() if (sendRoutedPacketToPeer(firstHop, routed)) { - return + return true } } val recipientId = packet.recipientID?.toHexString() if (recipientId != null && !packet.recipientID.contentEquals(SpecialRecipients.BROADCAST)) { if (sendRoutedPacketToPeer(recipientId, routed)) { - return + return true } } - fragmentingSender.send(routed, "Wi-Fi Aware broadcast") { single -> + return fragmentingSender.send(routed, "Wi-Fi Aware broadcast") { single -> broadcastSinglePacket(single) } } @@ -290,8 +297,7 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor private fun broadcastSinglePacket(routed: RoutedPacket): Boolean { val data = routed.packet.toBinaryData() ?: return false - broadcastRaw(data) - return true + return broadcastRaw(data) } private fun sendSinglePacketToPeer(peerID: String, packet: BitchatPacket): Boolean { @@ -1592,9 +1598,8 @@ class WifiAwareMeshService(private val context: Context) : MeshService, Transpor private inner class WifiAwareTransport : MeshTransport { override val id: String = "WIFI" - override fun broadcastPacket(routed: RoutedPacket) { + override fun broadcastPacket(routed: RoutedPacket): Boolean = this@WifiAwareMeshService.broadcastPacket(routed) - } override fun sendPacketToPeer(peerID: String, packet: BitchatPacket): Boolean { return this@WifiAwareMeshService.sendPacketToPeer(peerID, packet) } diff --git a/app/src/test/java/com/bitchat/android/ui/CommandProcessorTest.kt b/app/src/test/java/com/bitchat/android/ui/CommandProcessorTest.kt index 431d8500..9eff027c 100644 --- a/app/src/test/java/com/bitchat/android/ui/CommandProcessorTest.kt +++ b/app/src/test/java/com/bitchat/android/ui/CommandProcessorTest.kt @@ -5,6 +5,7 @@ import androidx.test.core.app.ApplicationProvider import com.bitchat.android.mesh.MeshService import com.bitchat.android.model.BitchatMessage import junit.framework.TestCase.assertEquals +import junit.framework.TestCase.assertTrue import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.UnconfinedTestDispatcher @@ -14,6 +15,7 @@ import org.junit.Test import org.junit.runner.RunWith import org.mockito.Mockito import org.mockito.kotlin.mock +import org.mockito.kotlin.whenever import org.robolectric.RobolectricTestRunner import java.util.Date @@ -95,4 +97,43 @@ class CommandProcessorTest() { assertEquals(result, true) } + + @Test + fun `msg command persists incoming messages as locally read through shared chat opening`() { + val peerID = "0102030405060708" + val message = BitchatMessage( + id = "message-opened-by-command", + sender = "alice", + content = "hello", + timestamp = Date(1), + isPrivate = true, + senderPeerID = peerID + ) + val locallyRead = mutableListOf() + chatState.setPrivateChats(mapOf(peerID to listOf(message))) + whenever(meshService.getPeerNicknames()).thenReturn(mapOf(peerID to "alice")) + + commandProcessor = CommandProcessor( + state = chatState, + messageManager = messageManager, + channelManager = channelManager, + privateChatManager = PrivateChatManager( + state = chatState, + messageManager = messageManager, + dataManager = DataManager(context = context), + noiseSessionDelegate = mock(), + markMessageReadLocally = locallyRead::add + ) + ) + + commandProcessor.processCommand( + command = "/msg alice", + meshService = meshService, + myPeerID = "self", + onSendMessage = { _, _, _ -> }, + viewModel = null + ) + + assertTrue(locallyRead.contains(message.id)) + } } diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/RetryingControlPacketSenderTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/RetryingControlPacketSenderTest.kt new file mode 100644 index 00000000..7503b6bb --- /dev/null +++ b/app/src/test/kotlin/com/bitchat/android/mesh/RetryingControlPacketSenderTest.kt @@ -0,0 +1,121 @@ +package com.bitchat.android.mesh + +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.delay +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Test + +@OptIn(ExperimentalCoroutinesApi::class) +class RetryingControlPacketSenderTest { + @Test + fun `control packet is sent for the full redundant retry window`() = runTest { + val attempts = mutableListOf() + val sender = RetryingControlPacketSender( + scope = this, + maxAttempts = 3, + retryDelayMs = 10, + interSendDelayMs = 1 + ) + + sender.enqueue( + key = "peer:message", + sendAttempt = { attempt -> + attempts += attempt + true + } + ) + advanceUntilIdle() + + assertEquals(listOf(1, 2, 3), attempts) + } + + @Test + fun `duplicate enqueue is coalesced while receipt retry is active`() = runTest { + var firstRequestAttempts = 0 + var duplicateRequestAttempts = 0 + val sender = RetryingControlPacketSender( + scope = this, + maxAttempts = 3, + retryDelayMs = 10, + interSendDelayMs = 1 + ) + + sender.enqueue( + key = "peer:message", + sendAttempt = { + firstRequestAttempts += 1 + true + } + ) + sender.enqueue( + key = "peer:message", + sendAttempt = { + duplicateRequestAttempts += 1 + true + } + ) + advanceUntilIdle() + + assertEquals(3, firstRequestAttempts) + assertEquals(0, duplicateRequestAttempts) + } + + @Test + fun `transport writes for different receipts are serialized`() = runTest { + var activeWrites = 0 + var maximumActiveWrites = 0 + val sender = RetryingControlPacketSender( + scope = this, + maxAttempts = 1, + retryDelayMs = 0, + interSendDelayMs = 0 + ) + + fun enqueue(key: String) { + sender.enqueue( + key = key, + sendAttempt = { + activeWrites += 1 + maximumActiveWrites = maxOf(maximumActiveWrites, activeWrites) + delay(10) + activeWrites -= 1 + true + } + ) + } + + enqueue("peer:first") + enqueue("peer:second") + advanceUntilIdle() + + assertEquals(1, maximumActiveWrites) + } + + @Test + fun `completion reports whether transport accepted any attempt`() = runTest { + val completions = mutableListOf() + val sender = RetryingControlPacketSender( + scope = this, + maxAttempts = 3, + retryDelayMs = 1, + interSendDelayMs = 0 + ) + + sender.enqueue( + key = "peer:rejected", + sendAttempt = { false }, + onComplete = completions::add + ) + sender.enqueue( + key = "peer:eventually-accepted", + sendAttempt = { attempt -> attempt == 2 }, + onComplete = completions::add + ) + advanceUntilIdle() + + assertEquals(2, completions.size) + assertEquals(setOf(false, true), completions.toSet()) + } +} diff --git a/app/src/test/kotlin/com/bitchat/android/nostr/NostrDirectMessageHandlerTest.kt b/app/src/test/kotlin/com/bitchat/android/nostr/NostrDirectMessageHandlerTest.kt index 738b3231..a9ded74d 100644 --- a/app/src/test/kotlin/com/bitchat/android/nostr/NostrDirectMessageHandlerTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/nostr/NostrDirectMessageHandlerTest.kt @@ -67,7 +67,7 @@ class NostrDirectMessageHandlerTest { ) val seenStore = mock() whenever(seenStore.hasDelivered(any())).thenReturn(true) - whenever(seenStore.hasRead(any())).thenReturn(false) + whenever(seenStore.hasBeenReadLocally(any())).thenReturn(false) val handler = NostrDirectMessageHandler( application = application, state = state, diff --git a/app/src/test/kotlin/com/bitchat/android/service/TransportBridgeServiceTest.kt b/app/src/test/kotlin/com/bitchat/android/service/TransportBridgeServiceTest.kt index 7e774dc3..cd8bf40d 100644 --- a/app/src/test/kotlin/com/bitchat/android/service/TransportBridgeServiceTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/service/TransportBridgeServiceTest.kt @@ -4,8 +4,10 @@ import android.os.Build import com.bitchat.android.model.RoutedPacket import com.bitchat.android.protocol.BitchatPacket import com.bitchat.android.protocol.MessageType +import kotlinx.coroutines.test.runTest import org.junit.After import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull import org.junit.Assert.assertTrue import org.junit.Test @@ -33,6 +35,11 @@ class TransportBridgeServiceTest { override fun send(packet: RoutedPacket) { captured = packet } + + override suspend fun sendAndReport(packet: RoutedPacket): Boolean { + captured = packet + return true + } } ) val packet = BitchatPacket( @@ -65,4 +72,41 @@ class TransportBridgeServiceTest { assertEquals(original.type, actual.type) } } + + @Test + fun `rejected bridge send remains eligible after transport reconnects`() = runTest { + var transportConnected = false + var attempts = 0 + TransportBridgeService.register( + targetId, + object : TransportBridgeService.TransportLayer { + override fun send(packet: RoutedPacket) = Unit + + override suspend fun sendAndReport(packet: RoutedPacket): Boolean { + attempts += 1 + return transportConnected + } + } + ) + val packet = BitchatPacket( + version = 1u, + type = MessageType.NOISE_ENCRYPTED.value, + senderID = ByteArray(8) { 1 }, + recipientID = ByteArray(8) { 2 }, + timestamp = System.nanoTime().toULong(), + payload = byteArrayOf(3, 4, 5), + signature = ByteArray(64) { 6 }, + ttl = 7u + ) + val sourceId = "source-${UUID.randomUUID()}" + + assertFalse( + TransportBridgeService.broadcastAndReport(sourceId, RoutedPacket(packet)) + ) + transportConnected = true + assertTrue( + TransportBridgeService.broadcastAndReport(sourceId, RoutedPacket(packet)) + ) + assertEquals(2, attempts) + } } diff --git a/app/src/test/kotlin/com/bitchat/android/services/AppStateStoreTest.kt b/app/src/test/kotlin/com/bitchat/android/services/AppStateStoreTest.kt index 577823ac..a70d4fa9 100644 --- a/app/src/test/kotlin/com/bitchat/android/services/AppStateStoreTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/services/AppStateStoreTest.kt @@ -1,8 +1,10 @@ package com.bitchat.android.services import com.bitchat.android.model.BitchatMessage +import com.bitchat.android.model.DeliveryStatus import org.junit.After import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test import java.util.Date @@ -137,4 +139,37 @@ class AppStateStoreTest { assertEquals(listOf(earlier, later), AppStateStore.privateMessages.value[contactID]) } + + @Test + fun `background receipt status persists and cannot be downgraded`() { + val message = BitchatMessage( + id = "outgoing-message", + sender = "bob", + content = "hello", + timestamp = Date(1), + isPrivate = true, + deliveryStatus = DeliveryStatus.Sending + ) + AppStateStore.addPrivateMessage("peer-a", message) + + AppStateStore.updatePrivateMessageStatus( + message.id, + DeliveryStatus.Delivered("peer-a", Date(2)) + ) + AppStateStore.updatePrivateMessageStatus( + message.id, + DeliveryStatus.Read("peer-a", Date(3)) + ) + AppStateStore.updatePrivateMessageStatus( + message.id, + DeliveryStatus.Delivered("peer-a", Date(4)) + ) + + val status = AppStateStore.privateMessages.value + .values + .flatten() + .single() + .deliveryStatus + assertTrue(status is DeliveryStatus.Read) + } } diff --git a/app/src/test/kotlin/com/bitchat/android/ui/MeshDelegateHandlerStateContractTest.kt b/app/src/test/kotlin/com/bitchat/android/ui/MeshDelegateHandlerStateContractTest.kt index 63ba28ef..162a5ef6 100644 --- a/app/src/test/kotlin/com/bitchat/android/ui/MeshDelegateHandlerStateContractTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/ui/MeshDelegateHandlerStateContractTest.kt @@ -1,6 +1,7 @@ package com.bitchat.android.ui import com.bitchat.android.mesh.MeshService +import com.bitchat.android.mesh.PeerInfo import com.bitchat.android.model.BitchatMessage import com.bitchat.android.model.DeliveryStatus import kotlinx.coroutines.ExperimentalCoroutinesApi @@ -14,8 +15,10 @@ import org.junit.Test import org.mockito.kotlin.any import org.mockito.kotlin.eq import org.mockito.kotlin.mock +import org.mockito.kotlin.never import org.mockito.kotlin.times import org.mockito.kotlin.verify +import org.mockito.kotlin.whenever import java.util.Date import java.util.concurrent.atomic.AtomicInteger @@ -29,6 +32,7 @@ class MeshDelegateHandlerStateContractTest { private lateinit var mesh: MeshService private lateinit var handler: MeshDelegateHandler private lateinit var haptics: AtomicInteger + private lateinit var locallyReadMessageIDs: MutableList @Before fun setUp() { @@ -41,6 +45,7 @@ class MeshDelegateHandlerStateContractTest { notifications = mock() mesh = mock() haptics = AtomicInteger() + locallyReadMessageIDs = mutableListOf() handler = MeshDelegateHandler( state = state, messageManager = messages, @@ -50,7 +55,8 @@ class MeshDelegateHandlerStateContractTest { coroutineScope = scope, onHapticFeedback = { haptics.incrementAndGet() }, getMyPeerID = { "self" }, - getMeshService = { mesh } + getMeshService = { mesh }, + markMessageReadLocally = locallyReadMessageIDs::add ) } @@ -89,6 +95,60 @@ class MeshDelegateHandlerStateContractTest { assertTrue(state.messages.value.single().deliveryStatus is DeliveryStatus.Read) } + @Test + fun `focused private message schedules receipt and records local read independently`() { + val peerID = "1122334455667788" + val incoming = BitchatMessage( + id = "focused-private-message", + sender = "alice", + content = "hello", + timestamp = Date(1), + isPrivate = true, + senderPeerID = peerID + ) + whenever(notifications.getAppBackgroundState()).thenReturn(false) + whenever(notifications.getCurrentPrivateChatPeer()).thenReturn(peerID) + whenever(mesh.getPeerInfo(peerID)).thenReturn( + PeerInfo( + id = peerID, + nickname = "alice", + isConnected = true, + isDirectConnection = true, + noisePublicKey = ByteArray(32) { 1 }, + signingPublicKey = null, + isVerifiedNickname = false, + lastSeen = System.currentTimeMillis() + ) + ) + whenever(mesh.hasEstablishedSession(peerID)).thenReturn(true) + + handler.didReceiveMessage(incoming) + + verify(mesh).sendReadReceipt(incoming.id, peerID, "Résumé") + assertEquals(listOf(incoming.id), locallyReadMessageIDs) + } + + @Test + fun `focused private message remains locally read when transport is disconnected`() { + val peerID = "1122334455667788" + val incoming = BitchatMessage( + id = "focused-private-message-offline", + sender = "alice", + content = "hello", + timestamp = Date(1), + isPrivate = true, + senderPeerID = peerID + ) + whenever(notifications.getAppBackgroundState()).thenReturn(false) + whenever(notifications.getCurrentPrivateChatPeer()).thenReturn(peerID) + whenever(mesh.getPeerInfo(peerID)).thenReturn(null) + + handler.didReceiveMessage(incoming) + + verify(mesh, never()).sendReadReceipt(any(), any(), any()) + assertEquals(listOf(incoming.id), locallyReadMessageIDs) + } + @Test fun `unicode mention notifies once and duplicate transport delivery is suppressed`() { val incoming = message( diff --git a/app/src/test/kotlin/com/bitchat/android/ui/PrivateChatManagerTest.kt b/app/src/test/kotlin/com/bitchat/android/ui/PrivateChatManagerTest.kt index 6ea06635..354fdc62 100644 --- a/app/src/test/kotlin/com/bitchat/android/ui/PrivateChatManagerTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/ui/PrivateChatManagerTest.kt @@ -15,6 +15,7 @@ import org.junit.Before import org.junit.Test import org.junit.runner.RunWith import org.mockito.kotlin.mock +import org.mockito.kotlin.never import org.mockito.kotlin.verify import org.mockito.kotlin.whenever import org.robolectric.RobolectricTestRunner @@ -129,6 +130,54 @@ class PrivateChatManagerTest { ) } + @Test + fun `opening chat skips messages whose receipt send already completed`() { + val noiseKey = ByteArray(32) { 8 } + val meshPeerID = ContactIdentityResolver.peerIdForNoiseKey(noiseKey) + val conversationID = ContactIdentityResolver.contactConversationIdForNoiseKey(noiseKey) + val oldMessage = BitchatMessage( + id = "already-read", + sender = "alice", + content = "old", + timestamp = Date(1), + isPrivate = true, + senderPeerID = meshPeerID + ) + val unreadMessage = oldMessage.copy( + id = "still-unread", + content = "new", + timestamp = Date(2) + ) + val meshService = mock() + manager = PrivateChatManager( + state = state, + messageManager = MessageManager(state), + dataManager = DataManager(RuntimeEnvironment.getApplication()), + noiseSessionDelegate = mock(), + hasReadReceiptBeenSent = { it == oldMessage.id } + ) + state.setNickname("bob") + state.setPrivateChats(mapOf(conversationID to listOf(oldMessage, unreadMessage))) + whenever(meshService.getPeerInfo(meshPeerID)).thenReturn( + PeerInfo( + id = meshPeerID, + nickname = "alice", + isConnected = true, + isDirectConnection = true, + noisePublicKey = noiseKey, + signingPublicKey = null, + isVerifiedNickname = false, + lastSeen = System.currentTimeMillis() + ) + ) + whenever(meshService.hasEstablishedSession(meshPeerID)).thenReturn(true) + + manager.sendReadReceiptsForPeer(conversationID, meshPeerID, meshService) + + verify(meshService, never()).sendReadReceipt(oldMessage.id, meshPeerID, "bob") + verify(meshService).sendReadReceipt(unreadMessage.id, meshPeerID, "bob") + } + @Test fun `canonical conversation send does not require resolved nickname`() { val conversationID =