nik/bitchat-android
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat-android.git synced 2026-08-22 07:06:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Harden background relay lifecycle

Browse Source
This commit is contained in:
callebtc 2026-07-27 18:00:10 +02:00
parent 6703f1bd20
commit b22184940b
9 changed files with 401 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
app/src/main/java/com/bitchat/android/mesh/UnifiedMeshService.kt
View File

@ -6,7 +6,6 @@ import com.bitchat.android.favorites.FavoriteControlMessage
import com.bitchat.android.model.BitchatFilePacket
import com.bitchat.android.model.BitchatMessage
import com.bitchat.android.noise.NoiseSession
import com.bitchat.android.services.AppStateStore
import com.bitchat.android.wifiaware.WifiAwareController
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
@ -14,7 +13,8 @@ import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
@ -77,18 +77,21 @@ class UnifiedMeshService(
private fun startAnnouncementScheduler() {
if (announcementJob?.isActive == true) return
announcementJob = serviceScope.launch {
combine(powerManager.profile, AppStateStore.directPeers) { profile, peers ->
profile.meshAnnouncementIntervalMs to peers.isNotEmpty()
}.collectLatest { (intervalMs, hasRecipients) ->
if (!hasRecipients) return@collectLatest
// Connection-specific paths already send an immediate announce. Begin the
// periodic cadence after the configured interval to avoid a transition burst.
while (isActive) {
delay(intervalMs)
if (AppStateStore.directPeers.value.isNotEmpty()) sendBroadcastAnnounce()
powerManager.profile
.map { profile ->
profile.meshAnnouncementIntervalMs to profile.hasDirectPeers
}
.distinctUntilChanged()
.collectLatest { (intervalMs, hasRecipients) ->
if (!hasRecipients) return@collectLatest
// Connection-specific paths already send an immediate announce. Begin the
// periodic cadence after the configured interval to avoid a transition burst.
while (isActive) {
delay(intervalMs)
if (powerManager.profile.value.hasDirectPeers) sendBroadcastAnnounce()
}
}
}
}
}
override fun sendMessage(content: String, mentions: List<String>, channel: String?) {

56
app/src/main/java/com/bitchat/android/nostr/NostrBackgroundRuntime.kt
View File

@ -61,7 +61,6 @@ object NostrBackgroundRuntime {
locationChannels = LocationChannelManager.getInstance(app)
subscriptions = NostrSubscriptionManager(
app,
scope,
owner = NostrRelayManager.OWNER_BACKGROUND
)
}
@ -94,7 +93,13 @@ object NostrBackgroundRuntime {
}
fun ensureConversationDm(geohash: String) {
if (!initialized || geohash == activeGeohash || geohash == conversationGeohash) return
if (!initialized || geohash == conversationGeohash) return
val selectedLiveToken = activeGeohashLiveToken
val selectedChannelSubscriptionIsUsable =
geohash == activeGeohash &&
(selectedLiveToken == null ||
LiveLocationPrivacyGate.accepts(selectedLiveToken))
if (selectedChannelSubscriptionIsUsable) return
conversationGeohash?.let { subscriptions.unsubscribe("geo-dm-conversation-$it") }
conversationGeohash = geohash
subscribeGeohashDm(
@ -119,8 +124,9 @@ object NostrBackgroundRuntime {
private fun observeSelectedChannel() {
scope.launch {
locationChannels.selectedChannel.collectLatest { channel ->
val next = (channel as? ChannelID.Location)?.channel?.geohash
val nextToken = (channel as? ChannelID.Location)?.let {
val locationChannel = channel as? ChannelID.Location
val next = locationChannel?.channel?.geohash
val nextToken = locationChannel?.let {
locationChannels.liveLocationTokenForSelectedChannel(it.channel)
}
val previous = activeGeohash
@ -139,9 +145,8 @@ object NostrBackgroundRuntime {
conversationGeohash = null
}
next?.let { geohash ->
val isLiveDerived = (channel as? ChannelID.Location)?.let {
locationChannels.isSelectedChannelLiveDerived(it.channel)
} == true
val isLiveDerived =
locationChannels.isSelectedChannelLiveDerived(locationChannel.channel)
if (!isLiveDerived || nextToken != null) {
subscribeSelectedGeohash(geohash, nextToken)
}
@ -171,21 +176,28 @@ object NostrBackgroundRuntime {
liveLocationToken: Long?
) {
scope.launch {
if (liveLocationToken != null &&
!LiveLocationPrivacyGate.accepts(liveLocationToken)
) return@launch
val identity = NostrIdentityBridge.deriveIdentity(geohash, application)
subscriptions.subscribeGiftWraps(
pubkey = identity.publicKeyHex,
sinceMs = System.currentTimeMillis() - 172_800_000L,
id = subscriptionId,
handler = { event -> dispatch { it.geohashDm(event, geohash, identity) } },
liveLocationToken = liveLocationToken
)
GeohashAliasRegistry.put(
"nostr_${identity.publicKeyHex.take(16)}",
identity.publicKeyHex
)
val subscribe = {
val identity = NostrIdentityBridge.deriveIdentity(geohash, application)
subscriptions.subscribeGiftWraps(
pubkey = identity.publicKeyHex,
sinceMs = System.currentTimeMillis() - 172_800_000L,
id = subscriptionId,
handler = { event ->
dispatch { it.geohashDm(event, geohash, identity) }
},
liveLocationToken = liveLocationToken
)
GeohashAliasRegistry.put(
"nostr_${identity.publicKeyHex.take(16)}",
identity.publicKeyHex
)
}
if (liveLocationToken == null) {
subscribe()
} else {
LiveLocationPrivacyGate.runIfAllowed(liveLocationToken, subscribe)
}
}
}

90
app/src/main/java/com/bitchat/android/nostr/NostrPendingEventQueue.kt Normal file
View File

@ -0,0 +1,90 @@
package com.bitchat.android.nostr
/**
* Thread-safe bounded queue of relay deliveries awaiting a usable WebSocket.
*
* Queue entries have a local ID rather than using the Nostr event ID: the same signed event may be
* intentionally published more than once with different relay sets or privacy provenance.
*/
internal class NostrPendingEventQueue(
private val capacity: Int
) {
init {
require(capacity > 0)
}
data class Delivery(
val queueId: Long,
val event: NostrEvent,
val liveLocationToken: Long?
)
private data class Entry(
val queueId: Long,
val event: NostrEvent,
val pendingRelayUrls: MutableSet<String>,
val liveLocationToken: Long?
)
private val lock = Any()
private val entries = ArrayDeque<Entry>()
private var nextQueueId = 1L
fun enqueue(
event: NostrEvent,
relayUrls: Collection<String>,
liveLocationToken: Long?
): Long? {
val pendingRelays = relayUrls.filterTo(linkedSetOf()) { it.isNotBlank() }
if (pendingRelays.isEmpty()) return null
return synchronized(lock) {
if (entries.size >= capacity) entries.removeFirst()
val queueId = nextQueueId++
entries.addLast(
Entry(
queueId = queueId,
event = event,
pendingRelayUrls = pendingRelays,
liveLocationToken = liveLocationToken
)
)
queueId
}
}
fun pendingForRelay(relayUrl: String): List<Delivery> = synchronized(lock) {
entries
.asSequence()
.filter { relayUrl in it.pendingRelayUrls }
.map { Delivery(it.queueId, it.event, it.liveLocationToken) }
.toList()
}
fun markDelivered(queueId: Long, relayUrl: String) {
synchronized(lock) {
val iterator = entries.iterator()
while (iterator.hasNext()) {
val entry = iterator.next()
if (entry.queueId != queueId) continue
entry.pendingRelayUrls.remove(relayUrl)
if (entry.pendingRelayUrls.isEmpty()) iterator.remove()
return
}
}
}
fun removeLiveLocationEvents() {
synchronized(lock) {
entries.removeAll { it.liveLocationToken != null }
}
}
fun clear() {
synchronized(lock) {
entries.clear()
}
}
internal fun size(): Int = synchronized(lock) { entries.size }
}

180
app/src/main/java/com/bitchat/android/nostr/NostrRelayManager.kt
View File

@ -2,17 +2,17 @@ package com.bitchat.android.nostr
import android.util.Log
import com.bitchat.android.geohash.LiveLocationPrivacyGate
import com.google.gson.Gson
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import com.google.gson.JsonArray
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.map
import com.google.gson.JsonParser
import kotlinx.coroutines.*
import okhttp3.*
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import kotlin.math.min
import kotlin.math.pow
@ -114,15 +114,8 @@ class NostrRelayManager private constructor() {
// Event deduplication system
private val eventDeduplicator = NostrEventDeduplicator.getInstance()
// Message queue for reliability
private data class QueuedEvent(
val event: NostrEvent,
val pendingRelayUrls: MutableSet<String>,
val liveLocationToken: Long? = null
)
private val messageQueue = mutableListOf<QueuedEvent>()
private val messageQueueLock = Any()
// Bounded per-relay delivery queue for reconnect reliability.
private val messageQueue = NostrPendingEventQueue(MAX_QUEUED_EVENTS)
// Coroutine scope for background operations
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@ -287,14 +280,24 @@ class NostrRelayManager private constructor() {
)
closeTargets.forEach { (relayUrl, relaySubscriptionIds) ->
val webSocket = connections[relayUrl] ?: return@forEach
relaySubscriptionIds.forEach { subscriptionId ->
for (subscriptionId in relaySubscriptionIds) {
val request = NostrRequest.Close(subscriptionId)
val message = gson.toJson(request, NostrRequest::class.java)
val closeQueued = runCatching { webSocket.send(message) }
.getOrDefault(false)
if (!closeQueued) {
connections.remove(relayUrl, webSocket)
subscriptions.remove(relayUrl)
webSocket.cancel()
updateRelayStatus(
relayUrl,
isConnected = false,
error = IllegalStateException("Failed to close revoked subscription")
)
if (desiredConnected.get() && relayUrl in nonLiveRelayUrls) {
scope.launch { connectToRelay(relayUrl, liveLocationToken = null) }
}
break
}
}
}
@ -314,9 +317,7 @@ class NostrRelayManager private constructor() {
}
subscriptions.replaceAll { _, ids -> ids - liveSubscriptionIds }
synchronized(messageQueueLock) {
messageQueue.removeAll { it.liveLocationToken != null }
}
messageQueue.removeLiveLocationEvents()
liveGeohashTokens.keys.forEach(geohashToRelays::remove)
liveGeohashTokens.clear()
@ -325,6 +326,8 @@ class NostrRelayManager private constructor() {
.filterNotTo(mutableSetOf()) { it in nonLiveRelayUrls }
liveOnlyRelayUrls.forEach { relayUrl ->
connections.remove(relayUrl)?.cancel()
subscriptions.remove(relayUrl)
reconnectJobs.remove(relayUrl)?.cancel()
}
synchronized(relaysList) {
relaysList.removeAll { it.url in liveOnlyRelayUrls }
@ -451,28 +454,24 @@ class NostrRelayManager private constructor() {
relayUrls: List<String>? = null,
liveLocationToken: Long? = null
) {
val targetRelays = relayUrls ?: relaysList.map { it.url }
val targetRelays = (relayUrls ?: relaysList.map { it.url })
.filter { it.isNotBlank() }
.distinct()
if (targetRelays.isEmpty()) return
val queued = runNetworkAction(liveLocationToken) {
synchronized(messageQueueLock) {
if (messageQueue.size >= MAX_QUEUED_EVENTS) {
messageQueue.removeAt(0)
}
messageQueue.add(
QueuedEvent(
event = event,
pendingRelayUrls = targetRelays.toMutableSet(),
liveLocationToken = liveLocationToken
)
)
}
val queueId = messageQueue.enqueue(
event = event,
relayUrls = targetRelays,
liveLocationToken = liveLocationToken
) ?: return@runNetworkAction
scope.launch {
if (!isNetworkActionAllowed(liveLocationToken)) return@launch
targetRelays.forEach { relayUrl ->
val webSocket = connections[relayUrl]
if (webSocket != null) {
if (sendToRelay(event, webSocket, relayUrl, liveLocationToken)) {
markQueuedRelayDelivered(event.id, relayUrl)
messageQueue.markDelivered(queueId, relayUrl)
}
}
}
@ -673,9 +672,7 @@ class NostrRelayManager private constructor() {
geohashToRelays.clear()
// Clear any queued messages waiting to be sent
synchronized(messageQueueLock) {
messageQueue.clear()
}
messageQueue.clear()
Log.i(TAG, "Cleared all Nostr subscriptions and routing caches")
} catch (e: Exception) {
@ -767,39 +764,52 @@ class NostrRelayManager private constructor() {
stopSubscriptionValidation() // Stop any existing validation
subscriptionValidationJob = scope.launch {
while (isActive && desiredConnected.get()) {
val validationInterval = powerManager?.profile?.value
?.nostr?.subscriptionValidationMs
?: com.bitchat.android.util.AppConstants.Nostr.SUBSCRIPTION_VALIDATION_INTERVAL_MS
delay(validationInterval)
if (!desiredConnected.get()) break
try {
val report = validateSubscriptionConsistency()
if (!report.isConsistent && report.connectedRelayCount > 0) {
Log.w(TAG, "Nostr subscription inconsistencies detected")
// Auto-repair: re-establish subscriptions for relays with missing ones
connections.forEach { (relayUrl, webSocket) ->
val currentSubs = subscriptions[relayUrl] ?: emptySet()
val expectedSubs = activeSubscriptions.keys.filter { subId ->
val subInfo = activeSubscriptions[subId]
subInfo?.targetRelayUrls == null || subInfo.targetRelayUrls.contains(relayUrl)
}.toSet()
val missingSubs = expectedSubs - currentSubs
if (missingSubs.isNotEmpty()) {
Log.i(TAG, "Auto-repairing ${missingSubs.size} missing subscriptions")
restoreSubscriptionsForRelay(relayUrl, webSocket)
}
}
}
} catch (e: Exception) {
Log.e(TAG, "Error during subscription validation: ${e.message}")
val manager = powerManager
if (manager == null) {
runSubscriptionValidationLoop(
com.bitchat.android.util.AppConstants.Nostr
.SUBSCRIPTION_VALIDATION_INTERVAL_MS
)
return@launch
}
manager.profile
.map { it.nostr.subscriptionValidationMs }
.distinctUntilChanged()
.collectLatest(::runSubscriptionValidationLoop)
}
}
private suspend fun runSubscriptionValidationLoop(intervalMs: Long) {
while (currentCoroutineContext().isActive && desiredConnected.get()) {
delay(intervalMs)
if (!desiredConnected.get()) break
validateAndRepairSubscriptions()
}
}
private fun validateAndRepairSubscriptions() {
try {
val report = validateSubscriptionConsistency()
if (report.isConsistent || report.connectedRelayCount == 0) return
Log.w(TAG, "Nostr subscription inconsistencies detected")
connections.forEach { (relayUrl, webSocket) ->
val currentSubs = subscriptions[relayUrl] ?: emptySet()
val expectedSubs = activeSubscriptions.keys.filter { subId ->
val subInfo = activeSubscriptions[subId]
subInfo?.targetRelayUrls == null ||
subInfo.targetRelayUrls.contains(relayUrl)
}.toSet()
if ((expectedSubs - currentSubs).isNotEmpty()) {
Log.i(TAG, "Auto-repairing missing subscriptions")
restoreSubscriptionsForRelay(relayUrl, webSocket)
}
}
} catch (e: Exception) {
Log.e(TAG, "Error during subscription validation: ${e.message}")
}
}
/**
@ -869,8 +879,9 @@ class NostrRelayManager private constructor() {
}
if (success) {
// Update relay stats
val relay = relaysList.find { it.url == relayUrl }
relay?.messagesSent = (relay?.messagesSent ?: 0) + 1
relaysList.find { it.url == relayUrl }?.let { relay ->
relay.messagesSent += 1
}
updateRelaysList()
true
} else {
@ -883,18 +894,6 @@ class NostrRelayManager private constructor() {
}
}
private fun markQueuedRelayDelivered(eventId: String, relayUrl: String) {
synchronized(messageQueueLock) {
val iterator = messageQueue.iterator()
while (iterator.hasNext()) {
val queued = iterator.next()
if (queued.event.id != eventId) continue
queued.pendingRelayUrls.remove(relayUrl)
if (queued.pendingRelayUrls.isEmpty()) iterator.remove()
}
}
}
private fun handleMessage(message: String, relayUrl: String) {
try {
val jsonElement = JsonParser.parseString(message)
@ -908,8 +907,9 @@ class NostrRelayManager private constructor() {
when (response) {
is NostrResponse.Event -> {
// Update relay stats
val relay = relaysList.find { it.url == relayUrl }
relay?.messagesReceived = (relay?.messagesReceived ?: 0) + 1
relaysList.find { it.url == relayUrl }?.let { relay ->
relay.messagesReceived += 1
}
updateRelaysList()
// CLIENT-SIDE FILTER ENFORCEMENT: Ensure this event matches the subscription's filter
@ -1142,17 +1142,17 @@ class NostrRelayManager private constructor() {
restoreSubscriptionsForRelay(relayUrl, webSocket)
// Process only events still pending for this relay, outside the queue lock.
val queuedForRelay = synchronized(messageQueueLock) {
messageQueue
.filter {
relayUrl in it.pendingRelayUrls &&
isNetworkActionAllowed(it.liveLocationToken)
}
.map { it.event to it.liveLocationToken }
}
queuedForRelay.forEach { (event, token) ->
if (sendToRelay(event, webSocket, relayUrl, token)) {
markQueuedRelayDelivered(event.id, relayUrl)
val queuedForRelay = messageQueue.pendingForRelay(relayUrl)
.filter { isNetworkActionAllowed(it.liveLocationToken) }
queuedForRelay.forEach { delivery ->
if (sendToRelay(
delivery.event,
webSocket,
relayUrl,
delivery.liveLocationToken
)
) {
messageQueue.markDelivered(delivery.queueId, relayUrl)
}
}
}

110
app/src/main/java/com/bitchat/android/nostr/NostrSubscriptionManager.kt
View File

@ -3,24 +3,32 @@ package com.bitchat.android.nostr
import android.app.Application
import android.util.Log
import com.bitchat.android.geohash.LiveLocationPrivacyGate
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch
/**
* NostrSubscriptionManager
* - Encapsulates subscription lifecycle with NostrRelayManager
* - Encapsulates ordered subscription lifecycle with NostrRelayManager.
*
* Relay-manager operations are already non-blocking and schedule network I/O on their own scope.
* Keeping this facade synchronous prevents lifecycle cancellation from dropping unsubscribe work
* and guarantees that a channel switch closes the old subscription before opening the new one.
*/
class NostrSubscriptionManager(
private val application: Application,
private val scope: CoroutineScope,
private val owner: String = NostrRelayManager.OWNER_LEGACY
) {
companion object { private const val TAG = "NostrSubscriptionManager" }
private val relayManager get() = NostrRelayManager.getInstance(application)
fun connect() = scope.launch { runCatching { relayManager.connect() }.onFailure { Log.e(TAG, "connect failed: ${it.message}") } }
fun disconnect() = scope.launch { runCatching { relayManager.disconnect() }.onFailure { Log.e(TAG, "disconnect failed: ${it.message}") } }
fun connect() {
runCatching { relayManager.connect() }
.onFailure { Log.e(TAG, "connect failed: ${it.message}") }
}
fun disconnect() {
runCatching { relayManager.disconnect() }
.onFailure { Log.e(TAG, "disconnect failed: ${it.message}") }
}
fun subscribeGiftWraps(
pubkey: String,
@ -29,19 +37,15 @@ class NostrSubscriptionManager(
handler: (NostrEvent) -> Unit,
liveLocationToken: Long? = null
) {
scope.launch {
if (liveLocationToken != null &&
!LiveLocationPrivacyGate.accepts(liveLocationToken)
) return@launch
val filter = NostrFilter.giftWrapsFor(pubkey, sinceMs)
relayManager.subscribe(
filter = filter,
id = id,
handler = handler,
owner = owner,
liveLocationToken = liveLocationToken
)
}
if (!isAllowed(liveLocationToken)) return
val filter = NostrFilter.giftWrapsFor(pubkey, sinceMs)
relayManager.subscribe(
filter = filter,
id = id,
handler = handler,
owner = owner,
liveLocationToken = liveLocationToken
)
}
/** Subscribe to geohash chat messages only (kind 20000) — low-volume, kept alive in background. */
@ -53,22 +57,18 @@ class NostrSubscriptionManager(
handler: (NostrEvent) -> Unit,
liveLocationToken: Long? = null
) {
scope.launch {
if (liveLocationToken != null &&
!LiveLocationPrivacyGate.accepts(liveLocationToken)
) return@launch
val filter = NostrFilter.geohashMessages(geohash, sinceMs, limit)
relayManager.subscribeForGeohash(
geohash,
filter,
id,
handler,
includeDefaults = false,
nRelays = 5,
owner = owner,
liveLocationToken = liveLocationToken
)
}
if (!isAllowed(liveLocationToken)) return
val filter = NostrFilter.geohashMessages(geohash, sinceMs, limit)
relayManager.subscribeForGeohash(
geohash,
filter,
id,
handler,
includeDefaults = false,
nRelays = 5,
owner = owner,
liveLocationToken = liveLocationToken
)
}
/** Subscribe to geohash presence heartbeats only (kind 20001) — high-volume, paused in background. */
@ -80,26 +80,28 @@ class NostrSubscriptionManager(
handler: (NostrEvent) -> Unit,
liveLocationToken: Long? = null
) {
scope.launch {
if (liveLocationToken != null &&
!LiveLocationPrivacyGate.accepts(liveLocationToken)
) return@launch
val filter = NostrFilter.geohashPresence(geohash, sinceMs, limit)
relayManager.subscribeForGeohash(
geohash,
filter,
id,
handler,
includeDefaults = false,
nRelays = 5,
owner = owner,
liveLocationToken = liveLocationToken
)
}
if (!isAllowed(liveLocationToken)) return
val filter = NostrFilter.geohashPresence(geohash, sinceMs, limit)
relayManager.subscribeForGeohash(
geohash,
filter,
id,
handler,
includeDefaults = false,
nRelays = 5,
owner = owner,
liveLocationToken = liveLocationToken
)
}
fun unsubscribe(id: String) { scope.launch { runCatching { relayManager.unsubscribe(id) } } }
fun unsubscribeAllOwned() {
scope.launch { runCatching { relayManager.unsubscribeOwner(owner) } }
fun unsubscribe(id: String) {
runCatching { relayManager.unsubscribe(id) }
}
fun unsubscribeAllOwned() {
runCatching { relayManager.unsubscribeOwner(owner) }
}
private fun isAllowed(liveLocationToken: Long?): Boolean =
liveLocationToken == null || LiveLocationPrivacyGate.accepts(liveLocationToken)
}

3
app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt
View File

@ -112,7 +112,8 @@ class MeshForegroundService : Service() {
private val unifiedMeshService: com.bitchat.android.mesh.MeshService?
get() = MeshServiceHolder.unifiedMeshService
private val serviceJob = Job()
private val scope = CoroutineScope(Dispatchers.Default + serviceJob)
// Service lifecycle callbacks and notification state are main-thread confined.
private val scope = CoroutineScope(Dispatchers.Main.immediate + serviceJob)
private var isInForeground: Boolean = false
private var isShuttingDown: Boolean = false
private var lastNotifiedPeerCount: Int? = null

8
app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt
View File

@ -42,10 +42,9 @@ class GeohashViewModel(
companion object { private const val TAG = "GeohashViewModel" }
private val repo = GeohashRepository(application, state, dataManager)
private val uiSubscriptionOwner = "geohash-ui-${System.identityHashCode(this)}"
private val uiSubscriptionOwner = "geohash-ui-${UUID.randomUUID()}"
private val subscriptionManager = NostrSubscriptionManager(
application,
viewModelScope,
owner = uiSubscriptionOwner
)
private val geohashMessageHandler = GeohashMessageHandler(
@ -75,6 +74,7 @@ class GeohashViewModel(
private val liveSamplingSubscriptionGeohashes = mutableSetOf<String>()
private var requestedLiveSamplingGeohashes: Set<String> = emptySet()
private var requestedUserSamplingGeohashes: Set<String> = emptySet()
private var uiSubscriptionsShutdown = false
private val liveLocationRevocationListener: () -> Unit = {
val revokedLiveGeohashes = liveSamplingSubscriptionGeohashes.toSet()
revokedLiveGeohashes.forEach { geohash ->
@ -424,11 +424,13 @@ class GeohashViewModel(
}
override fun onCleared() {
super.onCleared()
shutdownUiSubscriptions()
super.onCleared()
}
fun shutdownUiSubscriptions() {
if (uiSubscriptionsShutdown) return
uiSubscriptionsShutdown = true
subscriptionManager.unsubscribeAllOwned()
geoTimer?.cancel()
geoTimer = null

81
app/src/test/kotlin/com/bitchat/android/nostr/NostrPendingEventQueueTest.kt Normal file
View File

@ -0,0 +1,81 @@
package com.bitchat.android.nostr
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotEquals
import org.junit.Assert.assertNull
import org.junit.Test
class NostrPendingEventQueueTest {
@Test
fun `empty relay set is not queued`() {
val queue = NostrPendingEventQueue(capacity = 2)
assertNull(queue.enqueue(event("empty"), emptyList(), liveLocationToken = null))
assertEquals(0, queue.size())
}
@Test
fun `capacity evicts the oldest publish`() {
val queue = NostrPendingEventQueue(capacity = 2)
queue.enqueue(event("one"), listOf("relay"), liveLocationToken = null)
queue.enqueue(event("two"), listOf("relay"), liveLocationToken = null)
queue.enqueue(event("three"), listOf("relay"), liveLocationToken = null)
assertEquals(
listOf("two", "three"),
queue.pendingForRelay("relay").map { it.event.content }
)
}
@Test
fun `duplicate event publishes retain independent delivery state`() {
val queue = NostrPendingEventQueue(capacity = 4)
val signedEvent = event("same")
val firstId = requireNotNull(
queue.enqueue(signedEvent, listOf("relay-a", "relay-b"), liveLocationToken = null)
)
val secondId = requireNotNull(
queue.enqueue(signedEvent, listOf("relay-a"), liveLocationToken = null)
)
assertNotEquals(firstId, secondId)
queue.markDelivered(firstId, "relay-a")
assertEquals(
listOf(secondId),
queue.pendingForRelay("relay-a").map { it.queueId }
)
assertEquals(
listOf(firstId),
queue.pendingForRelay("relay-b").map { it.queueId }
)
queue.markDelivered(firstId, "relay-b")
assertEquals(1, queue.size())
}
@Test
fun `privacy purge retains non-live publishes`() {
val queue = NostrPendingEventQueue(capacity = 4)
queue.enqueue(event("manual"), listOf("relay"), liveLocationToken = null)
queue.enqueue(event("live"), listOf("relay"), liveLocationToken = 42L)
queue.removeLiveLocationEvents()
assertEquals(
listOf("manual"),
queue.pendingForRelay("relay").map { it.event.content }
)
}
private fun event(content: String): NostrEvent {
val privateKey = "0".repeat(63) + "1"
return NostrEvent(
pubkey = NostrCrypto.derivePublicKey(privateKey),
createdAt = 1,
kind = NostrKind.TEXT_NOTE,
tags = emptyList(),
content = content
).sign(privateKey)
}
}

28
app/src/test/kotlin/com/bitchat/android/nostr/NostrRelayManagerLifecycleSmokeTest.kt
View File

@ -9,6 +9,34 @@ import org.robolectric.RobolectricTestRunner
@RunWith(RobolectricTestRunner::class)
class NostrRelayManagerLifecycleSmokeTest {
@Test
fun `owner teardown is synchronous and preserves other subscription owners`() {
val manager = NostrRelayManager.shared
manager.disconnect()
manager.clearAllSubscriptions()
val filter = NostrFilter(kinds = listOf(NostrKind.TEXT_NOTE))
manager.subscribe(
filter = filter,
id = "background-contract",
handler = {},
targetRelayUrls = emptyList(),
owner = NostrRelayManager.OWNER_BACKGROUND
)
manager.subscribe(
filter = filter,
id = "ui-contract",
handler = {},
targetRelayUrls = emptyList(),
owner = "test-ui"
)
manager.unsubscribeOwner("test-ui")
assertEquals(setOf("background-contract"), manager.getActiveSubscriptions().keys)
manager.clearAllSubscriptions()
}
@Test
fun `disconnected manager maintains subscription and empty publish invariants locally`() {
val manager = NostrRelayManager.shared