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

Harden asynchronous Nostr admission and account resets

Browse Source
This commit is contained in:
Dev 2026-07-28 02:05:41 +03:00
parent 2d7c1feafe
commit 26ea39bbf7
27 changed files with 2840 additions and 641 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
app/src/main/java/com/bitchat/android/MainActivity.kt
View File

@ -88,6 +88,12 @@ class MainActivity : OrientationAwareActivity() {
finishAndRemoveTask()
return
}
if (com.bitchat.android.service.AppShutdownCoordinator
.isShutdownCommitted()
) {
finishAndRemoveTask()
return
}
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) {
this.setRecentsScreenshotEnabled(false)
}
@ -118,8 +124,6 @@ class MainActivity : OrientationAwareActivity() {
return
}
com.bitchat.android.service.AppShutdownCoordinator.cancelPendingShutdown()
// Enable edge-to-edge display for modern Android look
enableEdgeToEdge()
@ -724,6 +728,13 @@ class MainActivity : OrientationAwareActivity() {
override fun onNewIntent(intent: Intent) {
super.onNewIntent(intent)
setIntent(intent)
if (com.bitchat.android.service.AppShutdownCoordinator
.isShutdownCommitted()
) {
finishAndRemoveTask()
return
}
// Check if this is a quit request from the notification
if (intent.getBooleanExtra("ACTION_QUIT_APP", false)) {
@ -732,8 +743,6 @@ class MainActivity : OrientationAwareActivity() {
return
}
com.bitchat.android.service.AppShutdownCoordinator.cancelPendingShutdown()
// Handle notification intents when app is already running
if (mainViewModel.onboardingState.value == OnboardingState.COMPLETE) {
handleNotificationIntent(intent)

116
app/src/main/java/com/bitchat/android/nostr/GeohashMessageHandler.kt
View File

@ -6,6 +6,7 @@ import com.bitchat.android.model.BitchatMessage
import com.bitchat.android.ui.ChatState
import com.bitchat.android.ui.MessageManager
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import java.util.Date
@ -41,13 +42,22 @@ class GeohashMessageHandler(
return false
}
fun onEvent(event: NostrEvent, subscribedGeohash: String) {
scope.launch {
internal fun onEvent(
event: NostrEvent,
subscribedGeohash: String,
accountEpoch: NostrAccountEpoch
) {
val accountContext =
NostrInboundAccountLifecycle.contextFor(accountEpoch)
?: return
scope.launch(Dispatchers.Default + accountContext.receiveJob) {
try {
if (!NostrInboundAccountLifecycle.isCurrent(accountEpoch)) {
return@launch
}
if (event.kind != NostrKind.EPHEMERAL_EVENT && event.kind != NostrKind.GEOHASH_PRESENCE) return@launch
val tagGeo = event.tags.firstOrNull { it.size >= 2 && it[0] == "g" }?.getOrNull(1)
if (tagGeo == null || !tagGeo.equals(subscribedGeohash, true)) return@launch
if (dedupe(event.id)) return@launch
// PoW validation (if enabled) - apply to chat messages primarily
if (event.kind == NostrKind.EPHEMERAL_EVENT) {
@ -63,47 +73,71 @@ class GeohashMessageHandler(
// Blocked users check (use injected DataManager which has loaded state)
if (dataManager.isGeohashUserBlocked(pubkey)) return@launch
// Update participant count (last seen) on BOTH Presence (20001) and Chat (20000) events
if (event.kind == NostrKind.GEOHASH_PRESENCE || event.kind == NostrKind.EPHEMERAL_EVENT) {
repo.updateParticipant(subscribedGeohash, pubkey, Date(event.createdAt * 1000L))
}
event.tags.find { it.size >= 2 && it[0] == "n" }?.let { repo.cacheNickname(pubkey, it[1]) }
event.tags.find { it.size >= 2 && it[0] == "t" && it[1] == "teleport" }?.let { repo.markTeleported(pubkey) }
// Register a geohash DM alias for this participant so MessageRouter can route DMs via Nostr
try {
com.bitchat.android.nostr.GeohashAliasRegistry.put("nostr_${pubkey.take(16)}", pubkey)
} catch (_: Exception) { }
// Stop here for presence events - they don't produce chat messages
if (event.kind == NostrKind.GEOHASH_PRESENCE) return@launch
// Skip our own events for message emission
val my = NostrIdentityBridge.deriveIdentity(subscribedGeohash, application)
if (my.publicKeyHex.equals(pubkey, true)) return@launch
val isTeleportPresence = event.tags.any { it.size >= 2 && it[0] == "t" && it[1] == "teleport" } &&
event.content.trim().isEmpty()
if (isTeleportPresence) return@launch
val emitMessage =
event.kind != NostrKind.GEOHASH_PRESENCE &&
!isTeleportPresence &&
!NostrIdentityBridge.deriveIdentity(
subscribedGeohash,
application
).publicKeyHex.equals(pubkey, true)
val senderName = repo.displayNameForNostrPubkeyUI(pubkey)
val hasNonce = try { NostrProofOfWork.hasNonce(event) } catch (_: Exception) { false }
val msg = BitchatMessage(
id = event.id,
sender = senderName,
content = event.content,
timestamp = Date(event.createdAt * 1000L),
isRelay = false,
originalSender = repo.displayNameForNostrPubkey(pubkey),
senderPeerID = "nostr:${pubkey.take(8)}",
senderNostrPubkey = pubkey,
mentions = null,
channel = "#$subscribedGeohash",
powDifficulty = try {
if (hasNonce) NostrProofOfWork.calculateDifficulty(event.id).takeIf { it > 0 } else null
} catch (_: Exception) { null }
)
messageManager.addChannelMessage("geo:$subscribedGeohash", msg)
NostrInboundAccountLifecycle.runIfCurrent(accountEpoch) {
if (dedupe(event.id)) return@runIfCurrent
repo.updateParticipant(
subscribedGeohash,
pubkey,
Date(event.createdAt * 1000L)
)
event.tags.find {
it.size >= 2 && it[0] == "n"
}?.let {
repo.cacheNickname(pubkey, it[1])
}
event.tags.find {
it.size >= 2 && it[0] == "t" && it[1] == "teleport"
}?.let {
repo.markTeleported(pubkey)
}
GeohashAliasRegistry.put(
"nostr_${pubkey.take(16)}",
pubkey
)
if (emitMessage) {
val hasNonce = try {
NostrProofOfWork.hasNonce(event)
} catch (_: Exception) {
false
}
val msg = BitchatMessage(
id = event.id,
sender = repo.displayNameForNostrPubkeyUI(pubkey),
content = event.content,
timestamp = Date(event.createdAt * 1000L),
isRelay = false,
originalSender = repo.displayNameForNostrPubkey(pubkey),
senderPeerID = "nostr:${pubkey.take(8)}",
senderNostrPubkey = pubkey,
mentions = null,
channel = "#$subscribedGeohash",
powDifficulty = try {
if (hasNonce) {
NostrProofOfWork.calculateDifficulty(event.id)
.takeIf { it > 0 }
} else {
null
}
} catch (_: Exception) {
null
}
)
messageManager.addChannelMessage(
"geo:$subscribedGeohash",
msg
)
}
}
} catch (e: Exception) {
Log.e(TAG, "onEvent error: ${e.message}")
}

8
app/src/main/java/com/bitchat/android/nostr/NdrAccountEpochGuard.kt
View File

@ -6,9 +6,9 @@ internal data class NdrAccountEpoch(
)
/**
* Serializes NDR receive-side mutations against account invalidation.
* Serializes account-bound receive mutations against account invalidation.
*
* Panic invalidation waits for a mutation already inside [runIfCurrent], then
* Invalidation waits for a mutation already inside [runIfCurrent], then
* advances the generation before the wipe starts. Old-account jobs can
* therefore neither overlap nor repopulate the fresh post-wipe epoch.
*/
@ -47,3 +47,7 @@ internal class NdrAccountEpochGuard {
generation == epoch.generation &&
accountPubkeyHex == epoch.accountPubkeyHex
}
/** Account-wide names used by both legacy gift-wrap and NDR receive paths. */
internal typealias NostrAccountEpoch = NdrAccountEpoch
internal typealias NostrAccountEpochGuard = NdrAccountEpochGuard

68
app/src/main/java/com/bitchat/android/nostr/NdrNostrService.kt
View File

@ -19,12 +19,16 @@ internal class BitchatNdrRelayAdapter(
id: String,
handler: (NostrEvent) -> Boolean
) {
relayManager.subscribeAfterSuccessfulProcessing(
filter = filter,
id = id,
targetRelayUrls = accountRelayUrls,
handler = handler
)
check(
relayManager.subscribeAfterSuccessfulProcessing(
filter = filter,
id = id,
targetRelayUrls = accountRelayUrls,
handler = handler
)
) {
"NDR relay subscription was rejected during account reset"
}
}
override fun unsubscribe(id: String) {
@ -221,6 +225,7 @@ class NdrNostrService(
private val knownActivePeerPubkeys = linkedSetOf<String>()
private var nextRuntimeEpoch = 0L
private var activeRuntimeEpoch: Long? = null
private var processExitShutdown = false
init {
relayManager.setOnConnectionAvailable {
@ -243,19 +248,27 @@ class NdrNostrService(
}
@Synchronized
fun configureIfNeeded(identity: NostrIdentity) {
fun configureIfNeeded(
identity: NostrIdentity,
accountGuard: () -> Boolean = { true }
): Boolean {
if (!accountGuard()) return false
if (!NdrFeatureGate.isEnabled()) {
teardownLocked()
configurationFailurePubkeyHex = null
return
return false
}
configureRuntimeIfNeededLocked(identity, drainPendingActions = true)
return configureRuntimeIfNeededLocked(identity, drainPendingActions = true)
}
private fun configureRuntimeIfNeededLocked(
identity: NostrIdentity,
drainPendingActions: Boolean
): Boolean {
if (processExitShutdown) {
Log.w(TAG, "Refusing to reopen NDR during committed process exit")
return false
}
if (panicResetBlocked) {
Log.e(TAG, "Refusing to configure NDR after an incomplete panic wipe")
return false
@ -365,8 +378,10 @@ class NdrNostrService(
fun sendIfPossible(
text: String,
peerPubkeyHex: String,
expiresAtSeconds: ULong? = null
expiresAtSeconds: ULong? = null,
accountGuard: () -> Boolean = { true }
): NdrSendResult {
if (!accountGuard()) return NdrSendResult.FAILED
if (!NdrFeatureGate.isEnabled()) return NdrSendResult.NO_SESSION
val runtime = pairwiseRuntime ?: return if (configurationFailurePubkeyHex != null) {
NdrSendResult.FAILED
@ -386,18 +401,24 @@ class NdrNostrService(
if (!sessionInfo.sendReady) {
return NdrSendResult.FAILED
}
return try {
try {
runtime.sendText(peer, text, expiresAtSeconds)
if (!persistEstablishedMarkerIfNeededLocked(runtime)) {
return NdrSendResult.FAILED
}
drainAndApplyPubSubEventsLocked()
NdrSendResult.SENT
} catch (_: Throwable) {
Log.d(TAG, "NDR send failed")
drainAndApplyPubSubEventsLocked()
NdrSendResult.FAILED
runCatching { drainAndApplyPubSubEventsLocked() }
return NdrSendResult.FAILED
}
// Once sendText returns, the native runtime has durably advanced the
// ratchet and owns the pending publish action. Never ask the caller to
// retry that plaintext, even if host-side marker/drain work now fails.
if (!persistEstablishedMarkerIfNeededLocked(runtime)) {
Log.e(TAG, "NDR message admitted before host marker persistence failed")
return NdrSendResult.SENT
}
runCatching { drainAndApplyPubSubEventsLocked() }
.onFailure { Log.w(TAG, "NDR message admitted but pending-action drain failed") }
return NdrSendResult.SENT
}
@Synchronized
@ -1091,8 +1112,19 @@ class NdrNostrService(
.onFailure { Log.w(TAG, "Failed to destroy NDR runtime") }
}
/** Quiesce account-bound runtime work for process exit without wiping durable sessions. */
@Synchronized
fun shutdownForProcessExit() {
processExitShutdown = true
NostrInboundAccountLifecycle.invalidate()
onDecryptedMessage = null
onOutOfBandPayload = null
teardownLocked()
}
@Synchronized
fun resetForPanic(): Boolean {
NostrInboundAccountLifecycle.invalidate()
onDecryptedMessage = null
teardownLocked()
panicResetBlocked = true

88
app/src/main/java/com/bitchat/android/nostr/NostrClient.kt
View File

@ -79,6 +79,8 @@ class NostrClient private constructor(private val context: Context) {
fun shutdown() {
Log.d(TAG, "Shutting down Nostr client")
relayManager.disconnect()
currentIdentity = null
_currentNpub.value = null
_isInitialized.value = false
}
@ -91,13 +93,22 @@ class NostrClient private constructor(private val context: Context) {
onSuccess: (() -> Unit)? = null,
onError: ((String) -> Unit)? = null
) {
val identity = currentIdentity
val accountGeneration = relayManager.captureAccountGeneration()
if (!relayManager.isAccountGenerationCurrent(accountGeneration)) {
onError?.invoke("Nostr account is resetting")
return
}
val identity = NostrIdentityBridge.getCurrentNostrIdentity(context)
if (identity == null) {
onError?.invoke("Nostr client not initialized")
return
}
scope.launch {
if (!relayManager.isAccountGenerationCurrent(accountGeneration)) {
onError?.invoke("Nostr account changed before send")
return@launch
}
try {
// Decode recipient npub to hex pubkey
val (hrp, pubkeyBytes) = Bech32.decode(recipientNpub)
@ -116,9 +127,18 @@ class NostrClient private constructor(private val context: Context) {
)
// Track and send all gift wraps
giftWraps.forEach { wrap ->
NostrRelayManager.registerPendingGiftWrap(wrap.id)
relayManager.sendEvent(wrap)
val admitted = giftWraps.all { wrap ->
relayManager.registerPendingGiftWrap(
wrap.id,
accountGeneration
) && relayManager.sendEvent(
event = wrap,
expectedAccountGeneration = accountGeneration
)
}
if (!admitted) {
onError?.invoke("Nostr account changed before send")
return@launch
}
Log.i(TAG, "📤 Sent private message to ${recipientNpub.take(16)}...")
@ -135,7 +155,9 @@ class NostrClient private constructor(private val context: Context) {
* Subscribe to private messages for current identity
*/
fun subscribeToPrivateMessages(handler: (content: String, senderNpub: String, timestamp: Int) -> Unit) {
val identity = currentIdentity
val accountGeneration = relayManager.captureAccountGeneration()
if (!relayManager.isAccountGenerationCurrent(accountGeneration)) return
val identity = NostrIdentityBridge.getCurrentNostrIdentity(context)
if (identity == null) {
Log.e(TAG, "Cannot subscribe to private messages: client not initialized")
return
@ -146,11 +168,18 @@ class NostrClient private constructor(private val context: Context) {
since = System.currentTimeMillis() - 172800000L // Last 48 hours (align with NIP-17 randomization)
)
relayManager.subscribe(filter, "private-messages", { giftWrap ->
scope.launch {
handlePrivateMessage(giftWrap, handler)
}
})
relayManager.subscribe(
filter = filter,
id = "private-messages",
handler = { giftWrap ->
scope.launch {
if (relayManager.isAccountGenerationCurrent(accountGeneration)) {
handlePrivateMessage(giftWrap, handler)
}
}
},
expectedAccountGeneration = accountGeneration
)
Log.i(TAG, "🔑 Subscribed to private messages for: ${identity.getShortNpub()}")
}
@ -165,7 +194,12 @@ class NostrClient private constructor(private val context: Context) {
onSuccess: (() -> Unit)? = null,
onError: ((String) -> Unit)? = null
) {
val accountGeneration = relayManager.captureAccountGeneration()
scope.launch {
if (!relayManager.isAccountGenerationCurrent(accountGeneration)) {
onError?.invoke("Nostr account changed before send")
return@launch
}
try {
// Derive geohash-specific identity
val geohashIdentity = NostrIdentityBridge.deriveIdentity(geohash, context)
@ -178,7 +212,14 @@ class NostrClient private constructor(private val context: Context) {
nickname = nickname
)
relayManager.sendEvent(event)
if (!relayManager.sendEvent(
event = event,
expectedAccountGeneration = accountGeneration
)
) {
onError?.invoke("Nostr account changed before send")
return@launch
}
Log.i(TAG, "📤 Sent geohash message")
onSuccess?.invoke()
@ -197,17 +238,26 @@ class NostrClient private constructor(private val context: Context) {
geohash: String,
handler: (content: String, senderPubkey: String, nickname: String?, timestamp: Int) -> Unit
) {
val accountGeneration = relayManager.captureAccountGeneration()
if (!relayManager.isAccountGenerationCurrent(accountGeneration)) return
val filter = NostrFilter.geohashEphemeral(
geohash = geohash,
since = System.currentTimeMillis() - 3600000L, // Last hour
limit = 200
)
relayManager.subscribe(filter, "geohash-$geohash", { event ->
scope.launch {
handleGeohashMessage(event, handler)
}
})
relayManager.subscribe(
filter = filter,
id = "geohash-$geohash",
handler = { event ->
scope.launch {
if (relayManager.isAccountGenerationCurrent(accountGeneration)) {
handleGeohashMessage(event, handler)
}
}
},
expectedAccountGeneration = accountGeneration
)
Log.i(TAG, "🌍 Subscribed to geohash channel")
}
@ -223,7 +273,8 @@ class NostrClient private constructor(private val context: Context) {
/**
* Get current identity information
*/
fun getCurrentIdentity(): NostrIdentity? = currentIdentity
fun getCurrentIdentity(): NostrIdentity? =
NostrIdentityBridge.getCurrentNostrIdentity(context)
/**
* Get relay connection status
@ -248,7 +299,8 @@ class NostrClient private constructor(private val context: Context) {
return
}
val identity = currentIdentity ?: return
val identity = NostrIdentityBridge.getCurrentNostrIdentity(context)
?: return
try {
val decryptResult = NostrProtocol.decryptPrivateMessage(giftWrap, identity)

177
app/src/main/java/com/bitchat/android/nostr/NostrDirectMessageHandler.kt
View File

@ -22,7 +22,6 @@ import com.bitchat.android.ui.PrivateMessageOrigin
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import java.util.Date
@ -47,9 +46,8 @@ class NostrDirectMessageHandler(
private val seenStore by lazy(seenStoreProvider)
private val ndrService by lazy { NdrNostrService.getInstance(application) }
private val ndrAccountEpochs = NdrAccountEpochGuard()
private val ndrReceiveJobLock = Any()
private var ndrReceiveJob: Job = SupervisorJob(scope.coroutineContext[Job])
@Volatile
private var ndrCallbackInstalled = false
// Simple event deduplication
private val processedIds = ArrayDeque<String>()
@ -75,23 +73,32 @@ class NostrDirectMessageHandler(
dedupe(id)
}
fun configureDoubleRatchet(identity: NostrIdentity) {
/**
* Begin a fresh account-wide receive epoch before installing any account or
* derived-geohash subscription. The returned epoch must be captured by the
* subscription handler so a delayed old subscription cannot join a newer
* account merely because its event arrives later.
*/
internal fun configureAccount(identity: NostrIdentity): NostrAccountEpoch {
val accountContext = NostrInboundAccountLifecycle.begin(
accountPubkeyHex = identity.publicKeyHex,
parentJob = scope.coroutineContext[Job]
)
if (!NdrFeatureGate.isEnabled()) {
invalidateDoubleRatchetAccount()
ndrService.onDecryptedMessage = null
return
}
val epoch = ndrAccountEpochs.begin(identity.publicKeyHex)
val receiveJob = synchronized(ndrReceiveJobLock) {
ndrReceiveJob.cancel()
SupervisorJob(scope.coroutineContext[Job]).also { ndrReceiveJob = it }
if (ndrCallbackInstalled) {
ndrService.onDecryptedMessage = null
ndrCallbackInstalled = false
}
return accountContext.epoch
}
// A prior account's callback must not consume and discard pending
// deliveries while the replacement runtime is initialized.
ndrService.onDecryptedMessage = null
ndrService.configureIfNeeded(identity)
ndrService.onDecryptedMessage = callback@{ message, completion ->
if (!NdrFeatureGate.isEnabled() || !ndrAccountEpochs.isCurrent(epoch)) {
if (!NdrFeatureGate.isEnabled() ||
!NostrInboundAccountLifecycle.isCurrent(accountContext.epoch)
) {
completion(NdrDeliveryResult.REJECTED)
return@callback
}
@ -101,24 +108,49 @@ class NostrDirectMessageHandler(
completion(NdrDeliveryResult.RETRY)
return@callback
}
if (!currentIdentity.publicKeyHex.equals(epoch.accountPubkeyHex, ignoreCase = true)) {
if (!currentIdentity.publicKeyHex.equals(
accountContext.epoch.accountPubkeyHex,
ignoreCase = true
)
) {
completion(NdrDeliveryResult.REJECTED)
return@callback
}
onDoubleRatchetMessage(message, currentIdentity, epoch, receiveJob, completion)
onDoubleRatchetMessage(
message,
currentIdentity,
accountContext.epoch,
accountContext.receiveJob,
completion
)
}
ndrCallbackInstalled = true
return accountContext.epoch
}
internal fun currentAccountEpoch(): NostrAccountEpoch? =
NostrInboundAccountLifecycle.currentEpoch()
fun invalidateAccount() {
NostrInboundAccountLifecycle.invalidate()
if (ndrCallbackInstalled) {
ndrService.onDecryptedMessage = null
ndrCallbackInstalled = false
}
}
fun invalidateDoubleRatchetAccount() {
ndrAccountEpochs.invalidate()
synchronized(ndrReceiveJobLock) {
ndrReceiveJob.cancel()
}
}
fun onGiftWrap(giftWrap: NostrEvent, geohash: String, identity: NostrIdentity) {
scope.launch(Dispatchers.Default) {
internal fun onGiftWrap(
giftWrap: NostrEvent,
geohash: String,
identity: NostrIdentity,
accountEpoch: NostrAccountEpoch
): Job? {
val accountContext =
NostrInboundAccountLifecycle.contextFor(accountEpoch)
?: return null
return scope.launch(Dispatchers.Default + accountContext.receiveJob) {
try {
if (!isAccountEpochCurrent(accountEpoch)) return@launch
if (dedupe(giftWrap.id)) return@launch
val messageAge = System.currentTimeMillis() / 1000 - giftWrap.createdAt
@ -132,6 +164,7 @@ class NostrDirectMessageHandler(
val (content, rawSenderPubkey, rumorTimestamp) = decryptResult
val senderPubkey = rawSenderPubkey.lowercase()
if (!isAccountEpochCurrent(accountEpoch)) return@launch
val legacyAllowed = runCatching {
legacyNostrInboundAllowed(senderPubkey)
}.getOrDefault(false)
@ -139,6 +172,7 @@ class NostrDirectMessageHandler(
Log.w(TAG, "Rejecting legacy DM for an NDR-pinned contact")
return@launch
}
if (!isAccountEpochCurrent(accountEpoch)) return@launch
// If sender is blocked for geohash contexts, drop any events from this pubkey
// Applies to both geohash DMs (geohash != "") and account DMs (geohash == "")
@ -148,7 +182,8 @@ class NostrDirectMessageHandler(
senderPubkey = senderPubkey,
timestamp = Date(rumorTimestamp * 1000L),
geohash = geohash,
recipientIdentity = identity
recipientIdentity = identity,
accountEpoch = accountEpoch
)
} catch (_: Exception) {
@ -160,14 +195,16 @@ class NostrDirectMessageHandler(
private fun onDoubleRatchetMessage(
message: NdrDecryptedMessage,
identity: NostrIdentity,
epoch: NdrAccountEpoch,
accountEpoch: NostrAccountEpoch,
receiveJob: Job,
completion: (NdrDeliveryResult) -> Unit
) {
scope.launch(Dispatchers.Default + receiveJob) {
var result = NdrDeliveryResult.RETRY
try {
if (!NdrFeatureGate.isEnabled() || !ndrAccountEpochs.isCurrent(epoch)) {
if (!NdrFeatureGate.isEnabled() ||
!isAccountEpochCurrent(accountEpoch)
) {
result = NdrDeliveryResult.REJECTED
return@launch
}
@ -198,7 +235,9 @@ class NostrDirectMessageHandler(
result = NdrDeliveryResult.REJECTED
return@launch
}
if (!NdrFeatureGate.isEnabled() || !ndrAccountEpochs.isCurrent(epoch)) {
if (!NdrFeatureGate.isEnabled() ||
!isAccountEpochCurrent(accountEpoch)
) {
result = NdrDeliveryResult.REJECTED
return@launch
}
@ -213,7 +252,7 @@ class NostrDirectMessageHandler(
timestamp = Date(applicationMessage.timestampMs),
geohash = "",
recipientIdentity = identity,
ndrEpoch = epoch,
accountEpoch = accountEpoch,
ndrEventId = dedupeId,
expiresAtSeconds = applicationMessage.expiresAtSeconds
)
@ -221,10 +260,20 @@ class NostrDirectMessageHandler(
Log.e(TAG, "Failed to process double-ratchet message")
result = NdrDeliveryResult.RETRY
} finally {
if (result.shouldAcknowledge) {
if (seenStore.markProcessedNdr(message.eventId)) {
markProcessed(message.eventId)
} else {
if (!isAccountEpochCurrent(accountEpoch)) {
result = NdrDeliveryResult.REJECTED
} else if (result.shouldAcknowledge) {
var marked = false
val mutationApplied =
runIfAccountMutationCurrent(accountEpoch, null) {
marked = seenStore.markProcessedNdr(message.eventId)
if (marked) {
markProcessed(message.eventId)
}
}
if (!mutationApplied) {
result = NdrDeliveryResult.REJECTED
} else if (!marked) {
result = NdrDeliveryResult.RETRY
}
}
@ -239,10 +288,11 @@ class NostrDirectMessageHandler(
timestamp: Date,
geohash: String,
recipientIdentity: NostrIdentity,
ndrEpoch: NdrAccountEpoch? = null,
accountEpoch: NostrAccountEpoch,
ndrEventId: String? = null,
expiresAtSeconds: Long? = null
): NdrDeliveryResult {
if (!isAccountEpochCurrent(accountEpoch)) return NdrDeliveryResult.REJECTED
if (isExpired(expiresAtSeconds)) return NdrDeliveryResult.REJECTED
if (!content.startsWith("bitchat1:")) return NdrDeliveryResult.REJECTED
@ -257,7 +307,7 @@ class NostrDirectMessageHandler(
val noisePayload = NoisePayload.decode(packet.payload)
?: return NdrDeliveryResult.REJECTED
val convKey = "nostr_${senderPubkey.take(16)}"
if (!runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
if (!runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
repo.putNostrKeyMapping(convKey, senderPubkey)
GeohashAliasRegistry.put(convKey, senderPubkey)
@ -282,7 +332,7 @@ class NostrDirectMessageHandler(
senderPubkey = senderPubkey,
recipientIdentity = recipientIdentity,
allowAccountNdr = geohash.isEmpty(),
ndrEpoch = ndrEpoch,
accountEpoch = accountEpoch,
ndrEventId = ndrEventId,
expiresAtSeconds = expiresAtSeconds
)
@ -296,11 +346,11 @@ class NostrDirectMessageHandler(
senderPubkey: String,
recipientIdentity: NostrIdentity,
allowAccountNdr: Boolean,
ndrEpoch: NdrAccountEpoch? = null,
accountEpoch: NostrAccountEpoch,
ndrEventId: String? = null,
expiresAtSeconds: Long? = null
): NdrDeliveryResult {
if (!isNdrEpochCurrent(ndrEpoch) || isExpired(expiresAtSeconds)) {
if (!isAccountEpochCurrent(accountEpoch) || isExpired(expiresAtSeconds)) {
return NdrDeliveryResult.REJECTED
}
return when (payload.type) {
@ -314,7 +364,9 @@ class NostrDirectMessageHandler(
val favoriteControl = FavoriteControlMessage.parse(pm.content)
if (favoriteControl != null) {
if (!isNdrEpochCurrent(ndrEpoch) || isExpired(expiresAtSeconds)) {
if (!isAccountEpochCurrent(accountEpoch) ||
isExpired(expiresAtSeconds)
) {
return NdrDeliveryResult.REJECTED
}
val favoriteResult = handleFavoriteControl(
@ -323,14 +375,14 @@ class NostrDirectMessageHandler(
senderNickname,
timestamp,
senderPubkey,
ndrEpoch,
accountEpoch,
ndrEventId,
expiresAtSeconds
)
if (favoriteResult != NdrDeliveryResult.CONSUMED &&
favoriteResult != NdrDeliveryResult.DUPLICATE
) return favoriteResult
if (!runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
if (!runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
if (!seenStore.hasDelivered(pm.messageID)) {
sendDeliveryAck(
pm.messageID,
@ -367,7 +419,7 @@ class NostrDirectMessageHandler(
var messageAccepted = false
withContext(Dispatchers.Main) {
runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
privateChatManager.handleIncomingPrivateMessage(
message = message,
suppressUnread = suppressUnread,
@ -379,7 +431,7 @@ class NostrDirectMessageHandler(
if (!messageAccepted) return NdrDeliveryResult.REJECTED
runCatching {
runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
if (!seenStore.hasDelivered(pm.messageID)) {
sendDeliveryAck(
pm.messageID,
@ -417,7 +469,7 @@ class NostrDirectMessageHandler(
val messageId = String(payload.data, Charsets.UTF_8)
var consumed = false
withContext(Dispatchers.Main) {
runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
meshDelegateHandler.didReceiveDeliveryAck(messageId, conversationID)
consumed = true
}
@ -428,7 +480,7 @@ class NostrDirectMessageHandler(
val messageId = String(payload.data, Charsets.UTF_8)
var consumed = false
withContext(Dispatchers.Main) {
runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
meshDelegateHandler.didReceiveReadReceipt(messageId, conversationID)
consumed = true
}
@ -446,7 +498,7 @@ class NostrDirectMessageHandler(
return NdrDeliveryResult.DUPLICATE
}
var message: BitchatMessage? = null
if (!runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
if (!runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
val savedPath =
com.bitchat.android.features.file.FileUtils.saveIncomingFile(
context = application,
@ -474,7 +526,7 @@ class NostrDirectMessageHandler(
}
var consumed = false
withContext(Dispatchers.Main) {
runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
message?.let {
privateChatManager.handleIncomingPrivateMessage(
message = it,
@ -506,33 +558,20 @@ class NostrDirectMessageHandler(
}
}
private fun isNdrEpochCurrent(epoch: NdrAccountEpoch?): Boolean =
epoch == null ||
(NdrFeatureGate.isEnabled() && ndrAccountEpochs.isCurrent(epoch))
private fun runIfNdrEpochCurrent(
epoch: NdrAccountEpoch?,
mutation: () -> Unit
): Boolean {
if (epoch == null) {
mutation()
return true
}
if (!NdrFeatureGate.isEnabled()) return false
return ndrAccountEpochs.runIfCurrent(epoch, mutation)
}
private fun isAccountEpochCurrent(epoch: NostrAccountEpoch): Boolean =
NostrInboundAccountLifecycle.isCurrent(epoch)
private fun isExpired(expiresAtSeconds: Long?): Boolean =
expiresAtSeconds?.let { it <= System.currentTimeMillis() / 1_000L } == true
private fun runIfNdrMutationCurrent(
epoch: NdrAccountEpoch?,
private fun runIfAccountMutationCurrent(
epoch: NostrAccountEpoch,
expiresAtSeconds: Long?,
mutation: () -> Unit
): Boolean {
if (isExpired(expiresAtSeconds)) return false
var applied = false
val epochCurrent = runIfNdrEpochCurrent(epoch) {
val epochCurrent = NostrInboundAccountLifecycle.runIfCurrent(epoch) {
if (!isExpired(expiresAtSeconds)) {
mutation()
applied = true
@ -573,7 +612,7 @@ class NostrDirectMessageHandler(
senderNickname: String,
timestamp: Date,
senderPubkey: String,
ndrEpoch: NdrAccountEpoch? = null,
accountEpoch: NostrAccountEpoch,
ndrEventId: String? = null,
expiresAtSeconds: Long? = null
): NdrDeliveryResult {
@ -597,7 +636,7 @@ class NostrDirectMessageHandler(
}
var systemMessage: BitchatMessage? = null
if (!runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
if (!runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
FavoritesPersistenceService.shared.updatePeerFavoritedUs(
noiseKey,
control.isFavorite
@ -635,7 +674,7 @@ class NostrDirectMessageHandler(
var consumed = false
var duplicate = false
withContext(Dispatchers.Main) {
runIfNdrMutationCurrent(ndrEpoch, expiresAtSeconds) {
runIfAccountMutationCurrent(accountEpoch, expiresAtSeconds) {
systemMessage?.let {
if (state.getPrivateChatsValue()[targetConversationID]
.orEmpty()

60
app/src/main/java/com/bitchat/android/nostr/NostrInboundAccountLifecycle.kt Normal file
View File

@ -0,0 +1,60 @@
package com.bitchat.android.nostr
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
internal data class NostrInboundAccountContext(
val epoch: NostrAccountEpoch,
val receiveJob: Job
)
/**
* Process-wide lifetime for all account-bound Nostr receive work.
*
* Subscription handlers capture an epoch when they are installed. Beginning a
* replacement account invalidates those handlers and cancels their launched
* work. Invalidation is also a mutation barrier: after it returns, an old
* receive can finish non-sensitive parsing but cannot mutate application state.
*/
internal object NostrInboundAccountLifecycle {
private val lifecycleLock = Any()
private val epochs = NostrAccountEpochGuard()
private var currentContext: NostrInboundAccountContext? = null
fun begin(
accountPubkeyHex: String,
parentJob: Job?
): NostrInboundAccountContext = synchronized(lifecycleLock) {
val epoch = epochs.begin(accountPubkeyHex)
currentContext?.receiveJob?.cancel()
NostrInboundAccountContext(
epoch = epoch,
receiveJob = SupervisorJob(parentJob)
).also { currentContext = it }
}
fun contextFor(epoch: NostrAccountEpoch): NostrInboundAccountContext? =
synchronized(lifecycleLock) {
currentContext?.takeIf {
it.epoch == epoch
}
}
fun currentEpoch(): NostrAccountEpoch? =
synchronized(lifecycleLock) { currentContext?.epoch }
fun isCurrent(epoch: NostrAccountEpoch): Boolean = epochs.isCurrent(epoch)
fun runIfCurrent(
epoch: NostrAccountEpoch,
mutation: () -> Unit
): Boolean = epochs.runIfCurrent(epoch, mutation)
fun invalidate() {
synchronized(lifecycleLock) {
epochs.invalidate()
currentContext?.receiveJob?.cancel()
currentContext = null
}
}
}

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

File diff suppressed because it is too large Load Diff

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

@ -4,7 +4,6 @@ import android.app.Application
import android.util.Log
import com.bitchat.android.geohash.LiveLocationPrivacyGate
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch
/**
* NostrSubscriptionManager
@ -12,20 +11,32 @@ import kotlinx.coroutines.launch
*/
class NostrSubscriptionManager(
private val application: Application,
@Suppress("UNUSED_PARAMETER")
private val scope: CoroutineScope
) {
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, sinceMs: Long, id: String, handler: (NostrEvent) -> Unit) {
scope.launch {
val filter = NostrFilter.giftWrapsFor(pubkey, sinceMs)
relayManager.subscribe(filter, id, handler)
}
val generation = relayManager.captureAccountGeneration()
val filter = NostrFilter.giftWrapsFor(pubkey, sinceMs)
relayManager.subscribe(
filter = filter,
id = id,
handler = handler,
expectedAccountGeneration = generation
)
}
/** Subscribe to geohash chat messages only (kind 20000) — low-volume, kept alive in background. */
@ -37,21 +48,21 @@ 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,
liveLocationToken = liveLocationToken
)
}
if (liveLocationToken != null &&
!LiveLocationPrivacyGate.accepts(liveLocationToken)
) return
val generation = relayManager.captureAccountGeneration()
val filter = NostrFilter.geohashMessages(geohash, sinceMs, limit)
relayManager.subscribeForGeohash(
geohash,
filter,
id,
handler,
includeDefaults = false,
nRelays = 5,
liveLocationToken = liveLocationToken,
expectedAccountGeneration = generation
)
}
/** Subscribe to geohash presence heartbeats only (kind 20001) — high-volume, paused in background. */
@ -63,22 +74,24 @@ 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,
liveLocationToken = liveLocationToken
)
}
if (liveLocationToken != null &&
!LiveLocationPrivacyGate.accepts(liveLocationToken)
) return
val generation = relayManager.captureAccountGeneration()
val filter = NostrFilter.geohashPresence(geohash, sinceMs, limit)
relayManager.subscribeForGeohash(
geohash,
filter,
id,
handler,
includeDefaults = false,
nRelays = 5,
liveLocationToken = liveLocationToken,
expectedAccountGeneration = generation
)
}
fun unsubscribe(id: String) { scope.launch { runCatching { relayManager.unsubscribe(id) } } }
fun unsubscribe(id: String) {
runCatching { relayManager.unsubscribe(id) }
}
}

421
app/src/main/java/com/bitchat/android/nostr/NostrTransport.kt
View File

@ -11,13 +11,48 @@ import com.bitchat.android.services.ContactIdentityResolver
import kotlinx.coroutines.*
import java.util.*
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicBoolean
enum class NostrSendAdmission {
/** The legacy relay handoff or durable pairwise NDR state now owns delivery. */
ADMITTED,
/** No transport accepted the message yet, but unchanged input may succeed later. */
RETRYABLE,
/** The payload or recipient is invalid and cannot succeed unchanged. */
TERMINAL_FAILED
}
internal enum class NdrSendDisposition {
ADMITTED,
LEGACY_FALLBACK,
RETRYABLE
}
internal fun ndrSendDisposition(
result: NdrSendResult,
ndrRequired: Boolean = false,
rebindBlocked: Boolean = false,
pairwiseOnly: Boolean = false
): NdrSendDisposition = when {
result == NdrSendResult.SENT -> NdrSendDisposition.ADMITTED
rebindBlocked -> NdrSendDisposition.RETRYABLE
result == NdrSendResult.NO_SESSION && !ndrRequired && !pairwiseOnly ->
NdrSendDisposition.LEGACY_FALLBACK
else -> NdrSendDisposition.RETRYABLE
}
internal fun shouldUseLegacyNostrFallback(
result: NdrSendResult,
ndrRequired: Boolean = false,
rebindBlocked: Boolean = false
): Boolean =
result == NdrSendResult.NO_SESSION && !ndrRequired && !rebindBlocked
ndrSendDisposition(
result = result,
ndrRequired = ndrRequired,
rebindBlocked = rebindBlocked
) == NdrSendDisposition.LEGACY_FALLBACK
internal fun isLegacyNostrAllowedWhenNdrDisabled(
ndrRequired: Boolean,
@ -35,7 +70,11 @@ private data class NdrRecipientResolution(
*/
class NostrTransport(
private val context: Context,
var senderPeerID: String = ""
var senderPeerID: String = "",
private val transportScope: CoroutineScope =
CoroutineScope(Dispatchers.IO + SupervisorJob()),
private val relayManager: NostrRelayManager =
NostrRelayManager.getInstance(context)
) {
companion object {
@ -44,6 +83,8 @@ class NostrTransport(
@Volatile
private var INSTANCE: NostrTransport? = null
fun tryGetInstance(): NostrTransport? = INSTANCE
fun getInstance(context: Context): NostrTransport {
return INSTANCE ?: synchronized(this) {
@ -53,125 +94,244 @@ class NostrTransport(
}
// Throttle READ receipts to avoid relay rate limits (like iOS)
private data class AccountToken(
val transportEpoch: Long,
val relayGeneration: Long
)
private data class QueuedRead(
val receipt: ReadReceipt,
val peerID: String
val peerID: String,
val sequence: Long,
val accountToken: AccountToken
)
private val readQueue = ConcurrentLinkedQueue<QueuedRead>()
private var isSendingReadAcks = false
private val transportScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val accountStateLock = Any()
private var transportAccountEpoch = 0L
private var accountResetBlocked = false
private var nextReadSequence = 0L
private var activeReadSequence: Long? = null
private val ndrService by lazy { NdrNostrService.getInstance(context) }
// MARK: - Transport Interface Methods
val myPeerID: String get() = senderPeerID
private fun captureAccountToken(): AccountToken? = synchronized(accountStateLock) {
if (accountResetBlocked) {
null
} else {
AccountToken(
transportEpoch = transportAccountEpoch,
relayGeneration = relayManager.captureAccountGeneration()
)
}
}
private fun isAccountTokenCurrent(token: AccountToken): Boolean =
synchronized(accountStateLock) {
!accountResetBlocked &&
token.transportEpoch == transportAccountEpoch &&
relayManager.isAccountGenerationCurrent(token.relayGeneration)
}
/**
* Start an account-lifetime barrier. Already-launched work keeps its captured
* token and is refused at the final relay handoff; throttled receipts are
* discarded immediately.
*/
fun discardForAccountReset(): Long =
synchronized(accountStateLock) {
accountResetBlocked = true
transportAccountEpoch += 1
readQueue.clear()
activeReadSequence = null
transportAccountEpoch
}
/**
* Allow fresh work only when the caller still owns the latest reset.
* A later panic/quit must not be reopened by an older reset finishing late.
*/
fun completeAccountReset(resetToken: Long): Boolean =
synchronized(accountStateLock) {
if (resetToken != transportAccountEpoch) return@synchronized false
accountResetBlocked = false
true
}
internal fun queuedReadCountForTesting(): Int = readQueue.size
internal fun activeReadCountForTesting(): Int = synchronized(accountStateLock) {
if (activeReadSequence == null) 0 else 1
}
fun sendPrivateMessage(
content: String,
to: String,
recipientNickname: String,
messageID: String,
expiresAtSeconds: ULong? = null
expiresAtSeconds: ULong? = null,
completion: (NostrSendAdmission) -> Unit = {}
) {
transportScope.launch {
try {
val recipientNostrPubkey = resolveNostrPublicKey(to)
if (recipientNostrPubkey == null) {
Log.w(TAG, "No Nostr public key found for peerID: $to")
return@launch
}
val senderIdentity = NostrIdentityBridge.getCurrentNostrIdentity(context)
if (senderIdentity == null) {
Log.e(TAG, "No Nostr identity available")
return@launch
}
val recipientHex = ContactIdentityResolver.nostrPubkeyHex(recipientNostrPubkey)
if (recipientHex == null) {
Log.e(TAG, "NostrTransport: recipient key is not a valid Nostr pubkey")
return@launch
}
val ndrRecipient = resolveNdrRecipient(to, recipientHex)
val accountToken = captureAccountToken()
if (accountToken == null) {
runCatching { completion(NostrSendAdmission.RETRYABLE) }
.onFailure { Log.w(TAG, "Nostr private-message admission callback failed") }
return
}
val completed = AtomicBoolean(false)
fun completeOnce(admission: NostrSendAdmission) {
if (!completed.compareAndSet(false, true)) return
runCatching { completion(admission) }
.onFailure { Log.w(TAG, "Nostr private-message admission callback failed") }
}
val recipientPeerIDForEmbed = try {
com.bitchat.android.favorites.FavoritesPersistenceService.shared
.findPeerIDForNostrPubkey(recipientNostrPubkey)
} catch (_: Exception) { null }
if (recipientPeerIDForEmbed.isNullOrBlank()) {
Log.e(TAG, "NostrTransport: no peerID stored for recipient npub; cannot embed PM")
return@launch
}
val embedded = NostrEmbeddedBitChat.encodePMForNostr(
val job = transportScope.launch {
val admission = try {
prepareAndSendPrivateMessage(
content = content,
to = to,
messageID = messageID,
recipientPeerID = recipientPeerIDForEmbed,
senderPeerID = senderPeerID
expiresAtSeconds = expiresAtSeconds,
accountToken = accountToken
)
if (embedded == null) {
Log.e(TAG, "NostrTransport: failed to embed PM packet")
return@launch
}
sendWrappedMessage(
content = embedded,
fallbackRecipientHex = recipientHex,
senderIdentity = senderIdentity,
ndrRecipient = ndrRecipient,
expiresAtSeconds = expiresAtSeconds
)
} catch (e: Exception) {
Log.e(TAG, "Failed to send private message via Nostr: ${e.message}")
NostrSendAdmission.RETRYABLE
}
completeOnce(admission)
}
job.invokeOnCompletion { cause ->
if (cause != null) {
completeOnce(NostrSendAdmission.RETRYABLE)
}
}
}
private fun prepareAndSendPrivateMessage(
content: String,
to: String,
messageID: String,
expiresAtSeconds: ULong?,
accountToken: AccountToken
): NostrSendAdmission {
if (!isAccountTokenCurrent(accountToken)) {
return NostrSendAdmission.RETRYABLE
}
if (expiresAtSeconds != null &&
expiresAtSeconds <= (System.currentTimeMillis() / 1_000L).toULong()
) {
Log.e(TAG, "NostrTransport: refusing an already-expired private message")
return NostrSendAdmission.TERMINAL_FAILED
}
val recipientNostrPubkey = resolveNostrPublicKey(to)
if (recipientNostrPubkey == null) {
Log.w(TAG, "No Nostr public key found for peerID: $to")
return NostrSendAdmission.RETRYABLE
}
val senderIdentity = NostrIdentityBridge.getCurrentNostrIdentity(context)
if (senderIdentity == null) {
Log.e(TAG, "No Nostr identity available")
return NostrSendAdmission.RETRYABLE
}
val recipientHex = ContactIdentityResolver.nostrPubkeyHex(recipientNostrPubkey)
if (recipientHex == null) {
Log.e(TAG, "NostrTransport: recipient key is not a valid Nostr pubkey")
return NostrSendAdmission.TERMINAL_FAILED
}
val ndrRecipient = resolveNdrRecipient(to, recipientHex)
val recipientPeerIDForEmbed = try {
com.bitchat.android.favorites.FavoritesPersistenceService.shared
.findPeerIDForNostrPubkey(recipientNostrPubkey)
} catch (_: Exception) {
null
}
if (recipientPeerIDForEmbed.isNullOrBlank()) {
Log.e(TAG, "NostrTransport: no peerID stored for recipient npub; cannot embed PM")
return NostrSendAdmission.RETRYABLE
}
val embedded = NostrEmbeddedBitChat.encodePMForNostr(
content = content,
messageID = messageID,
recipientPeerID = recipientPeerIDForEmbed,
senderPeerID = senderPeerID
)
if (embedded == null) {
Log.e(TAG, "NostrTransport: failed to embed PM packet")
return NostrSendAdmission.TERMINAL_FAILED
}
return sendWrappedMessage(
content = embedded,
fallbackRecipientHex = recipientHex,
senderIdentity = senderIdentity,
ndrRecipient = ndrRecipient,
expiresAtSeconds = expiresAtSeconds,
accountToken = accountToken
)
}
fun sendReadReceipt(receipt: ReadReceipt, to: String) {
val accountToken = captureAccountToken() ?: return
// Enqueue and process with throttling to avoid relay rate limits
readQueue.offer(QueuedRead(receipt, to))
val queuedRead = synchronized(accountStateLock) {
if (!isAccountTokenCurrent(accountToken)) {
null
} else {
QueuedRead(
receipt = receipt,
peerID = to,
sequence = ++nextReadSequence,
accountToken = accountToken
).also(readQueue::offer)
}
} ?: return
if (!isAccountTokenCurrent(queuedRead.accountToken)) return
processReadQueueIfNeeded()
}
private fun processReadQueueIfNeeded() {
if (isSendingReadAcks) return
if (readQueue.isEmpty()) return
isSendingReadAcks = true
sendNextReadAck()
val item = synchronized(accountStateLock) {
if (accountResetBlocked || activeReadSequence != null) return
var next = readQueue.poll()
while (next != null && !isAccountTokenCurrent(next.accountToken)) {
next = readQueue.poll()
}
next?.also { activeReadSequence = it.sequence }
} ?: return
sendReadAck(item)
}
private fun sendNextReadAck() {
val item = readQueue.poll()
if (item == null) {
isSendingReadAcks = false
return
}
private fun sendReadAck(item: QueuedRead) {
transportScope.launch {
try {
if (!isAccountTokenCurrent(item.accountToken)) {
finishReadAck(item)
return@launch
}
val recipientNostrPubkey = resolveNostrPublicKey(item.peerID)
if (recipientNostrPubkey == null) {
Log.w(TAG, "No Nostr public key found for read receipt to: ${item.peerID}")
scheduleNextReadAck()
finishReadAck(item)
return@launch
}
val senderIdentity = NostrIdentityBridge.getCurrentNostrIdentity(context)
if (senderIdentity == null) {
Log.e(TAG, "No Nostr identity available for read receipt")
scheduleNextReadAck()
finishReadAck(item)
return@launch
}
val recipientHex = ContactIdentityResolver.nostrPubkeyHex(recipientNostrPubkey)
if (recipientHex == null) {
scheduleNextReadAck()
finishReadAck(item)
return@launch
}
val ndrRecipient = resolveNdrRecipient(item.peerID, recipientHex)
@ -185,7 +345,7 @@ class NostrTransport(
if (ack == null) {
Log.e(TAG, "NostrTransport: failed to embed READ ack")
scheduleNextReadAck()
finishReadAck(item)
return@launch
}
@ -193,29 +353,35 @@ class NostrTransport(
content = ack,
fallbackRecipientHex = recipientHex,
senderIdentity = senderIdentity,
ndrRecipient = ndrRecipient
ndrRecipient = ndrRecipient,
accountToken = item.accountToken
)
scheduleNextReadAck()
finishReadAck(item)
} catch (e: Exception) {
Log.e(TAG, "Failed to send read receipt via Nostr: ${e.message}")
scheduleNextReadAck()
finishReadAck(item)
}
}
}
private fun scheduleNextReadAck() {
private fun finishReadAck(item: QueuedRead) {
transportScope.launch {
delay(READ_ACK_INTERVAL)
isSendingReadAcks = false
synchronized(accountStateLock) {
if (activeReadSequence != item.sequence) return@launch
activeReadSequence = null
}
processReadQueueIfNeeded()
}
}
fun sendFavoriteNotification(to: String, isFavorite: Boolean) {
val accountToken = captureAccountToken() ?: return
transportScope.launch {
try {
if (!isAccountTokenCurrent(accountToken)) return@launch
val recipientNostrPubkey = resolveNostrPublicKey(to)
if (recipientNostrPubkey == null) {
@ -253,7 +419,8 @@ class NostrTransport(
content = embedded,
fallbackRecipientHex = recipientHex,
senderIdentity = senderIdentity,
ndrRecipient = ndrRecipient
ndrRecipient = ndrRecipient,
accountToken = accountToken
)
} catch (e: Exception) {
@ -263,8 +430,10 @@ class NostrTransport(
}
fun sendDeliveryAck(messageID: String, to: String) {
val accountToken = captureAccountToken() ?: return
transportScope.launch {
try {
if (!isAccountTokenCurrent(accountToken)) return@launch
val recipientNostrPubkey = resolveNostrPublicKey(to)
if (recipientNostrPubkey == null) {
@ -300,7 +469,8 @@ class NostrTransport(
content = ack,
fallbackRecipientHex = recipientHex,
senderIdentity = senderIdentity,
ndrRecipient = ndrRecipient
ndrRecipient = ndrRecipient,
accountToken = accountToken
)
} catch (e: Exception) {
@ -316,8 +486,10 @@ class NostrTransport(
toRecipientHex: String,
fromIdentity: NostrIdentity
) {
val accountToken = captureAccountToken() ?: return
transportScope.launch {
try {
if (!isAccountTokenCurrent(accountToken)) return@launch
val embedded = NostrEmbeddedBitChat.encodeAckForNostrNoRecipient(
type = NoisePayloadType.DELIVERED,
messageID = messageID,
@ -332,11 +504,7 @@ class NostrTransport(
senderIdentity = fromIdentity
)
// Register pending gift wrap for deduplication and send all
giftWraps.forEach { event ->
NostrRelayManager.registerPendingGiftWrap(event.id)
NostrRelayManager.getInstance(context).sendEvent(event)
}
sendLegacyGiftWraps(giftWraps, accountToken)
} catch (e: Exception) {
Log.e(TAG, "Failed to send geohash delivery ack: ${e.message}")
@ -349,8 +517,10 @@ class NostrTransport(
toRecipientHex: String,
fromIdentity: NostrIdentity
) {
val accountToken = captureAccountToken() ?: return
transportScope.launch {
try {
if (!isAccountTokenCurrent(accountToken)) return@launch
val embedded = NostrEmbeddedBitChat.encodeAckForNostrNoRecipient(
type = NoisePayloadType.READ_RECEIPT,
messageID = messageID,
@ -365,11 +535,7 @@ class NostrTransport(
senderIdentity = fromIdentity
)
// Register pending gift wrap for deduplication and send all
giftWraps.forEach { event ->
NostrRelayManager.registerPendingGiftWrap(event.id)
NostrRelayManager.getInstance(context).sendEvent(event)
}
sendLegacyGiftWraps(giftWraps, accountToken)
} catch (e: Exception) {
Log.e(TAG, "Failed to send geohash read receipt: ${e.message}")
@ -385,6 +551,7 @@ class NostrTransport(
messageID: String,
sourceGeohash: String? = null
) {
val accountToken = captureAccountToken() ?: return
// Use provided geohash or derive from current location
val geohash = sourceGeohash ?: run {
val selected = try {
@ -406,6 +573,7 @@ class NostrTransport(
transportScope.launch {
try {
if (!isAccountTokenCurrent(accountToken)) return@launch
if (toRecipientHex.isEmpty()) return@launch
// Build embedded BitChat packet without recipient peer ID
@ -424,10 +592,7 @@ class NostrTransport(
senderIdentity = fromIdentity
)
giftWraps.forEach { event ->
NostrRelayManager.registerPendingGiftWrap(event.id)
NostrRelayManager.getInstance(context).sendEvent(event)
}
sendLegacyGiftWraps(giftWraps, accountToken)
} catch (e: Exception) {
Log.e(TAG, "Failed to send geohash private message: ${e.message}")
}
@ -445,34 +610,43 @@ class NostrTransport(
ndrRequired = false,
rebindBlocked = false
),
expiresAtSeconds: ULong? = null
): Boolean {
expiresAtSeconds: ULong? = null,
accountToken: AccountToken
): NostrSendAdmission {
if (!isAccountTokenCurrent(accountToken)) {
return NostrSendAdmission.RETRYABLE
}
if (ndrRecipient.rebindBlocked) {
Log.e(TAG, "NostrTransport: recipient rebind is quarantined")
return false
return NostrSendAdmission.RETRYABLE
}
if (NdrFeatureGate.isEnabled()) {
ndrService.configureIfNeeded(senderIdentity)
val configured = ndrService.configureIfNeeded(senderIdentity) {
isAccountTokenCurrent(accountToken)
}
if (!configured) {
return NostrSendAdmission.RETRYABLE
}
val sendResult = ndrService.sendIfPossible(
text = content,
peerPubkeyHex = ndrRecipient.peerPubkeyHex,
expiresAtSeconds = expiresAtSeconds
expiresAtSeconds = expiresAtSeconds,
accountGuard = { isAccountTokenCurrent(accountToken) }
)
if (sendResult == NdrSendResult.SENT) {
return true
}
if (expiresAtSeconds != null && sendResult == NdrSendResult.NO_SESSION) {
Log.e(TAG, "NostrTransport: expiring message requires a pairwise session")
return false
}
if (!shouldUseLegacyNostrFallback(
when (
ndrSendDisposition(
result = sendResult,
ndrRequired = ndrRecipient.ndrRequired,
rebindBlocked = ndrRecipient.rebindBlocked
rebindBlocked = ndrRecipient.rebindBlocked,
pairwiseOnly = expiresAtSeconds != null
)
) {
Log.e(TAG, "NostrTransport: pairwise send failed; refusing legacy downgrade")
return false
NdrSendDisposition.ADMITTED -> return NostrSendAdmission.ADMITTED
NdrSendDisposition.RETRYABLE -> {
Log.e(TAG, "NostrTransport: pairwise send not admitted; refusing legacy downgrade")
return NostrSendAdmission.RETRYABLE
}
NdrSendDisposition.LEGACY_FALLBACK -> Unit
}
} else if (expiresAtSeconds != null ||
!isLegacyNostrAllowedWhenNdrDisabled(
@ -481,18 +655,39 @@ class NostrTransport(
)
) {
Log.e(TAG, "NostrTransport: pairwise transport is required")
return false
return NostrSendAdmission.RETRYABLE
}
NostrProtocol.createPrivateMessage(
val events = NostrProtocol.createPrivateMessage(
content = content,
recipientPubkey = fallbackRecipientHex,
senderIdentity = senderIdentity
).forEach { event ->
NostrRelayManager.registerPendingGiftWrap(event.id)
NostrRelayManager.getInstance(context).sendEvent(event)
)
if (events.isEmpty()) {
Log.e(TAG, "NostrTransport: failed to create legacy gift wrap")
return NostrSendAdmission.RETRYABLE
}
return if (sendLegacyGiftWraps(events, accountToken)) {
NostrSendAdmission.ADMITTED
} else {
NostrSendAdmission.RETRYABLE
}
}
private fun sendLegacyGiftWraps(
events: List<NostrEvent>,
accountToken: AccountToken
): Boolean = synchronized(accountStateLock) {
if (!isAccountTokenCurrent(accountToken)) return@synchronized false
events.all { event ->
relayManager.registerPendingGiftWrap(
event.id,
accountToken.relayGeneration
) && relayManager.sendEvent(
event = event,
expectedAccountGeneration = accountToken.relayGeneration
)
}
return false
}
private fun resolveNdrRecipient(

131
app/src/main/java/com/bitchat/android/service/AppShutdownCoordinator.kt
View File

@ -9,13 +9,11 @@ import com.bitchat.android.net.TorMode
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.Job
import kotlinx.coroutines.async
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicBoolean
/**
* Coordinates a full application shutdown:
@ -27,15 +25,9 @@ import java.util.concurrent.atomic.AtomicLong
*/
object AppShutdownCoordinator {
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
private val shutdownToken = AtomicLong(0L)
@Volatile
private var shutdownJob: Job? = null
private val shutdownGate = ShutdownGate()
fun cancelPendingShutdown() {
shutdownToken.incrementAndGet()
shutdownJob?.cancel()
shutdownJob = null
}
fun isShutdownCommitted(): Boolean = shutdownGate.isCommitted()
fun requestFullShutdownAndKill(
app: Application,
@ -44,51 +36,84 @@ object AppShutdownCoordinator {
stopForeground: () -> Unit,
stopService: () -> Unit
) {
val token = shutdownToken.incrementAndGet()
shutdownJob?.cancel()
val job = scope.launch {
// Signal UI to finish gracefully before we kill the process
try {
val intent = android.content.Intent(com.bitchat.android.util.AppConstants.UI.ACTION_FORCE_FINISH)
.setPackage(app.packageName)
app.sendBroadcast(intent, com.bitchat.android.util.AppConstants.UI.PERMISSION_FORCE_FINISH)
} catch (_: Exception) { }
if (!shutdownGate.commit()) return
// Stop mesh (best-effort)
try { mesh?.stopServices() } catch (_: Exception) { }
// Stop Tor temporarily (do not change user setting)
val torProvider = ArtiTorManager.getInstance()
val torStop = async {
try { torProvider.applyMode(app, TorMode.OFF) } catch (_: Exception) { }
val terminated = AtomicBoolean(false)
val terminateProcess = {
if (terminated.compareAndSet(false, true)) {
try { stopService() } catch (_: Exception) { }
try { Process.killProcess(Process.myPid()) } catch (_: Exception) { }
try { System.exit(0) } catch (_: Exception) { }
}
// Clear AppState in-memory store
try { com.bitchat.android.services.AppStateStore.clear() } catch (_: Exception) { }
// Stop foreground and clear notification
try { stopForeground() } catch (_: Exception) { }
try { notificationManager.cancel(10001) } catch (_: Exception) { }
// Wait up to 5 seconds for shutdown tasks
withTimeoutOrNull(5000) {
try { torStop.await() } catch (_: Exception) { }
delay(100)
}
// Stop the service itself
if (!isActive || shutdownToken.get() != token) return@launch
try { stopService() } catch (_: Exception) { }
// Hard kill the app process
if (!isActive || shutdownToken.get() != token) return@launch
try { Process.killProcess(Process.myPid()) } catch (_: Exception) { }
try { System.exit(0) } catch (_: Exception) { }
}
shutdownJob = job
job.invokeOnCompletion {
if (shutdownJob === job) {
shutdownJob = null
scope.launch {
delay(5_000)
terminateProcess()
}
scope.launch {
try {
// Quit is an account-lifetime boundary, not a transient service
// pause: no queued plaintext or relay event may survive it.
try {
com.bitchat.android.nostr.NostrTransport
.tryGetInstance()
?.discardForAccountReset()
} catch (_: Exception) { }
try {
com.bitchat.android.services.MessageRouter
.tryGetInstance()
?.discardForAccountReset()
} catch (_: Exception) { }
try {
com.bitchat.android.nostr.NostrInboundAccountLifecycle
.invalidate()
} catch (_: Exception) { }
val relayManager = runCatching {
com.bitchat.android.nostr.NostrRelayManager.getInstance(app)
}.getOrNull()
val relayResetToken = runCatching {
relayManager?.beginAccountReset()
}.getOrNull()
try {
com.bitchat.android.nostr.NdrNostrService
.getInstance(app)
.shutdownForProcessExit()
} catch (_: Exception) { }
if (relayResetToken != null) {
runCatching {
relayManager?.discardForAccountReset(relayResetToken)
}
}
// Signal UI to finish gracefully before we kill the process
try {
val intent = android.content.Intent(com.bitchat.android.util.AppConstants.UI.ACTION_FORCE_FINISH)
.setPackage(app.packageName)
app.sendBroadcast(intent, com.bitchat.android.util.AppConstants.UI.PERMISSION_FORCE_FINISH)
} catch (_: Exception) { }
// Stop mesh (best-effort)
try { mesh?.stopServices() } catch (_: Exception) { }
// Stop Tor temporarily (do not change user setting)
val torProvider = ArtiTorManager.getInstance()
val torStop = async {
try { torProvider.applyMode(app, TorMode.OFF) } catch (_: Exception) { }
}
// Clear AppState in-memory store
try { com.bitchat.android.services.AppStateStore.clear() } catch (_: Exception) { }
// Stop foreground and clear notification
try { stopForeground() } catch (_: Exception) { }
try { notificationManager.cancel(10001) } catch (_: Exception) { }
withTimeoutOrNull(5_000) {
try { torStop.await() } catch (_: Exception) { }
delay(100)
}
} finally {
terminateProcess()
}
}
}

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

@ -118,9 +118,14 @@ class MeshForegroundService : Service() {
override fun onCreate() {
super.onCreate()
if (AppShutdownCoordinator.isShutdownCommitted()) {
stopSelf()
return
}
if (!com.bitchat.android.nostr.NdrPanicStartupRecovery
.isNetworkStartupAllowed()
) {
discardOldAccountNetworkWork()
stopSelf()
return
}
@ -140,15 +145,17 @@ class MeshForegroundService : Service() {
}
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
if (!com.bitchat.android.nostr.NdrPanicStartupRecovery
.isNetworkStartupAllowed()
) {
if (AppShutdownCoordinator.isShutdownCommitted()) {
isShuttingDown = true
stopSelf()
return START_NOT_STICKY
}
if (isShuttingDown && intent?.action == ACTION_START) {
AppShutdownCoordinator.cancelPendingShutdown()
isShuttingDown = false
if (!com.bitchat.android.nostr.NdrPanicStartupRecovery
.isNetworkStartupAllowed()
) {
discardOldAccountNetworkWork()
stopSelf()
return START_NOT_STICKY
}
if (isShuttingDown && intent?.action != ACTION_QUIT) {
return START_NOT_STICKY
@ -393,6 +400,13 @@ class MeshForegroundService : Service() {
}
override fun onDestroy() {
// Service teardown is normally transient: pause retries but preserve
// the outbox for a later service rebind. Panic/quit discard explicitly.
try {
com.bitchat.android.services.MessageRouter
.tryGetInstance()
?.stopOutboxScheduler()
} catch (_: Exception) { }
updateJob?.cancel()
updateJob = null
// Cancel the service coroutine scope to prevent leaks
@ -405,5 +419,39 @@ class MeshForegroundService : Service() {
super.onDestroy()
}
private fun discardOldAccountNetworkWork() {
try {
com.bitchat.android.nostr.NostrTransport
.tryGetInstance()
?.discardForAccountReset()
} catch (_: Exception) { }
try {
com.bitchat.android.services.MessageRouter
.tryGetInstance()
?.discardForAccountReset()
} catch (_: Exception) { }
try {
com.bitchat.android.nostr.NostrInboundAccountLifecycle
.invalidate()
} catch (_: Exception) { }
val relayManager = runCatching {
com.bitchat.android.nostr.NostrRelayManager
.getInstance(applicationContext)
}.getOrNull()
val relayResetToken = runCatching {
relayManager?.beginAccountReset()
}.getOrNull()
try {
com.bitchat.android.nostr.NdrNostrService
.getInstance(applicationContext)
.shutdownForProcessExit()
} catch (_: Exception) { }
if (relayResetToken != null) {
runCatching {
relayManager?.discardForAccountReset(relayResetToken)
}
}
}
override fun onBind(intent: Intent?): IBinder? = null
}

21
app/src/main/java/com/bitchat/android/service/ShutdownGate.kt Normal file
View File

@ -0,0 +1,21 @@
package com.bitchat.android.service
/**
* Makes a destructive application quit irreversible once committed.
*/
internal class ShutdownGate {
private val lock = Any()
private var committed = false
fun commit(): Boolean =
synchronized(lock) {
if (committed) {
false
} else {
committed = true
true
}
}
fun isCommitted(): Boolean = synchronized(lock) { committed }
}

14
app/src/main/java/com/bitchat/android/services/AppStateStore.kt
View File

@ -126,8 +126,18 @@ object AppStateStore {
val idx = list.indexOfFirst { it.id == messageID }
if (idx >= 0) {
val current = list[idx].deliveryStatus
// Do not downgrade (e.g., Read -> Delivered)
if (statusPriority(status) >= statusPriority(current)) {
val acceptsLocalFailure = status is DeliveryStatus.Failed &&
(current == null ||
current is DeliveryStatus.Sending ||
current is DeliveryStatus.Failed)
val rejectsLateFailure = status is DeliveryStatus.Failed &&
!acceptsLocalFailure
// Locally terminal/expired work may move Sending -> Failed,
// but a late failure must not overwrite Sent/Delivered/Read.
if (!rejectsLateFailure &&
(acceptsLocalFailure ||
statusPriority(status) >= statusPriority(current))
) {
list[idx] = list[idx].copy(deliveryStatus = status)
map[peer] = list
changed = true

304
app/src/main/java/com/bitchat/android/services/MessageRouter.kt
View File

@ -5,6 +5,7 @@ import android.util.Log
import com.bitchat.android.favorites.FavoriteControlMessage
import com.bitchat.android.mesh.MeshService
import com.bitchat.android.model.ReadReceipt
import com.bitchat.android.nostr.NostrSendAdmission
import com.bitchat.android.nostr.NostrTransport
import com.bitchat.android.util.AppConstants
import kotlinx.coroutines.CoroutineScope
@ -19,14 +20,36 @@ import java.util.concurrent.ConcurrentHashMap
/**
* Routes messages between local mesh transports and Nostr, matching iOS behavior.
*/
class MessageRouter private constructor(
internal fun interface NostrPrivateMessageSender {
fun sendPrivateMessage(
content: String,
to: String,
recipientNickname: String,
messageID: String,
completion: (NostrSendAdmission) -> Unit
)
}
class MessageRouter internal constructor(
private val context: Context,
private var mesh: MeshService,
private val nostr: NostrTransport
private val nostr: NostrTransport,
private val privateNostrSender: NostrPrivateMessageSender =
NostrPrivateMessageSender { content, to, recipientNickname, messageID, completion ->
nostr.sendPrivateMessage(
content = content,
to = to,
recipientNickname = recipientNickname,
messageID = messageID,
completion = completion
)
},
private val canSendViaNostrOverride: ((String) -> Boolean)? = null
) {
enum class RouteResult {
MESH,
NOSTR,
NOSTR_PENDING,
QUEUED,
DROPPED
}
@ -43,6 +66,12 @@ class MessageRouter private constructor(
val nextHandshakeAttemptAtMs: Long
)
private data class InFlightNostrAttempt(
val messageID: String,
val generation: Long,
val outboxEpoch: Long
)
companion object {
private const val TAG = "MessageRouter"
private const val OUTBOX_TICK_MS = AppConstants.Router.OUTBOX_TICK_MS
@ -86,6 +115,13 @@ class MessageRouter private constructor(
// Per-conversation handshake retry state for queued messages
private val retryState = ConcurrentHashMap<String, ConversationRetry>()
// A Nostr/NDR ratchet send must have one owner. Only one queued message per
// conversation may cross the asynchronous transport boundary at a time.
private val inFlightNostrAttempts = mutableMapOf<String, InFlightNostrAttempt>()
private var nextNostrAttemptGeneration = 0L
private var outboxEpoch = 0L
private var accountResetBlocked = false
private val schedulerScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
private var schedulerJob: kotlinx.coroutines.Job? = null
@ -94,6 +130,8 @@ class MessageRouter private constructor(
// Called with the messageID of queued messages that expired or were evicted
var onMessageExpired: ((String) -> Unit)? = null
var onMessageAdmitted: ((String) -> Unit)? = null
var onMessageFailed: ((String, String) -> Unit)? = null
init {
startOutboxScheduler()
@ -110,7 +148,12 @@ class MessageRouter private constructor(
}
}
@Synchronized
fun sendPrivate(content: String, toPeerID: String, recipientNickname: String, messageID: String): RouteResult {
if (accountResetBlocked) {
notifyFailed(messageID, "Account reset in progress")
return RouteResult.DROPPED
}
val resolution = ContactDirectory.resolve(toPeerID)
val conversationID = resolution.conversationID
val meshTarget = resolution.meshPeerID ?: toPeerID.takeIf { ContactIdentityResolver.isMeshPeerId(it) }
@ -134,18 +177,24 @@ class MessageRouter private constructor(
return RouteResult.MESH
} else if (canSendViaNostr(nostrTarget)) {
Log.d(TAG, "Routing PM via Nostr to ${conversationID.take(32)}… msg_id=${messageID.take(8)}…")
nostr.sendPrivateMessage(content, nostrTarget, recipientNickname, messageID)
return RouteResult.NOSTR
enqueue(
conversationID,
QueuedMessage(content, recipientNickname, messageID, clock())
)
flushOutboxFor(conversationID)
return RouteResult.NOSTR_PENDING
} else {
Log.d(TAG, "Queued PM for ${conversationID} (no mesh, no Nostr mapping) msg_id=${messageID.take(8)}…")
enqueue(conversationID, QueuedMessage(content, recipientNickname, messageID, clock()))
Log.d(TAG, "Initiating noise handshake after queueing PM for ${conversationID.take(16)}…")
if (hasMesh) meshTarget?.let { kickHandshake(conversationID, it, immediate = true) }
if (hasMesh) kickHandshake(conversationID, meshTarget, immediate = true)
return RouteResult.QUEUED
}
}
@Synchronized
fun sendReadReceipt(receipt: ReadReceipt, toPeerID: String) {
if (accountResetBlocked) return
val resolution = ContactDirectory.resolve(toPeerID)
val meshTarget = resolution.meshPeerID ?: toPeerID.takeIf { ContactIdentityResolver.isMeshPeerId(it) }
val nostrTarget = resolution.noiseKeyHex ?: toPeerID
@ -158,7 +207,9 @@ class MessageRouter private constructor(
}
}
@Synchronized
fun sendDeliveryAck(messageID: String, toPeerID: String) {
if (accountResetBlocked) return
// Mesh delivery ACKs are sent by the receiver automatically.
// Only route via Nostr when mesh path isn't available or when this is a geohash alias
if (com.bitchat.android.nostr.GeohashAliasRegistry.contains(toPeerID)) {
@ -175,7 +226,9 @@ class MessageRouter private constructor(
}
}
@Synchronized
fun sendFavoriteNotification(toPeerID: String, isFavorite: Boolean) {
if (accountResetBlocked) return
val resolution = ContactDirectory.resolve(toPeerID)
val meshTarget = resolution.meshPeerID ?: toPeerID.takeIf { ContactIdentityResolver.isMeshPeerId(it) }
if (meshTarget != null && mesh.getPeerInfo(meshTarget)?.isConnected == true && mesh.hasEstablishedSession(meshTarget)) {
@ -194,31 +247,179 @@ class MessageRouter private constructor(
@Synchronized
fun flushOutboxFor(peerID: String) {
val conversationID = ContactDirectory.canonicalConversationId(peerID)
val queued = outbox[conversationID] ?: outbox[peerID] ?: return
val queued = canonicalizeQueueKey(conversationID, peerID) ?: return
if (queued.isEmpty()) return
if (findInFlightAttempt(conversationID, peerID, queued) != null) return
Log.d(TAG, "Flushing outbox for ${conversationID.take(16)}… count=${queued.size}")
val iterator = queued.iterator()
while (iterator.hasNext()) {
val entry = iterator.next()
val resolution = ContactDirectory.resolve(conversationID)
val meshTarget = resolution.meshPeerID
val nostrTarget = resolution.noiseKeyHex ?: conversationID
if (meshTarget != null && isReady(mesh, meshTarget)) {
val resolution = ContactDirectory.resolve(conversationID)
val meshTarget = resolution.meshPeerID
val nostrTarget = resolution.noiseKeyHex ?: conversationID
if (meshTarget != null && isReady(mesh, meshTarget)) {
queued.forEach { entry ->
mesh.sendPrivateMessage(entry.content, meshTarget, entry.nickname, entry.messageID)
iterator.remove()
} else if (canSendViaNostr(nostrTarget)) {
nostr.sendPrivateMessage(entry.content, nostrTarget, entry.nickname, entry.messageID)
iterator.remove()
}
queued.clear()
} else if (canSendViaNostr(nostrTarget)) {
startNostrAttempt(
conversationID = conversationID,
target = nostrTarget,
entry = queued.first()
)
return
}
if (queued.isEmpty()) {
outbox.remove(conversationID, queued)
outbox.remove(peerID, queued)
retryState.remove(conversationID)
retryState.remove(peerID)
removeEmptyQueue(conversationID, peerID, queued)
}
}
@Synchronized
private fun startNostrAttempt(
conversationID: String,
target: String,
entry: QueuedMessage
) {
val queued = outbox[conversationID] ?: return
if (findInFlightAttempt(conversationID, conversationID, queued) != null) return
val attempt = InFlightNostrAttempt(
messageID = entry.messageID,
generation = ++nextNostrAttemptGeneration,
outboxEpoch = outboxEpoch
)
inFlightNostrAttempts[conversationID] = attempt
privateNostrSender.sendPrivateMessage(
entry.content,
target,
entry.nickname,
entry.messageID
) { admission ->
finishNostrAttempt(conversationID, attempt, admission)
}
}
@Synchronized
private fun finishNostrAttempt(
conversationID: String,
attempt: InFlightNostrAttempt,
admission: NostrSendAdmission
) {
if (attempt.outboxEpoch != outboxEpoch) return
val attemptEntry = inFlightNostrAttempts.entries
.firstOrNull { it.value === attempt }
?: return
inFlightNostrAttempts.remove(attemptEntry.key, attempt)
val queueKey = outbox.entries
.firstOrNull { (_, queue) ->
queue.any { it.messageID == attempt.messageID }
}
?.key
?: conversationID
val currentConversationID =
ContactDirectory.canonicalConversationId(queueKey)
if (queueKey != currentConversationID) {
canonicalizeQueueKey(currentConversationID, queueKey)
}
when (admission) {
NostrSendAdmission.ADMITTED -> {
removeQueuedMessage(currentConversationID, attempt.messageID)
notifyAdmitted(attempt.messageID)
flushOutboxFor(currentConversationID)
}
NostrSendAdmission.RETRYABLE -> Unit
NostrSendAdmission.TERMINAL_FAILED -> {
removeQueuedMessage(currentConversationID, attempt.messageID)
notifyFailed(attempt.messageID, "Message could not be sent")
flushOutboxFor(currentConversationID)
}
}
}
private fun removeQueuedMessage(conversationID: String, messageID: String) {
val queued = outbox[conversationID] ?: return
queued.removeAll { it.messageID == messageID }
if (queued.isEmpty()) {
removeEmptyQueue(conversationID, conversationID, queued)
}
}
private fun canonicalizeQueueKey(
conversationID: String,
peerID: String
): MutableList<QueuedMessage>? {
val canonical = outbox[conversationID]
val alias = outbox[peerID]
val queue = when {
canonical == null && alias != null -> {
outbox[conversationID] = alias
if (peerID != conversationID) {
outbox.remove(peerID, alias)
}
alias
}
canonical != null && alias != null && canonical !== alias -> {
canonical.addAll(alias)
canonical.sortBy(QueuedMessage::enqueuedAtMs)
outbox.remove(peerID, alias)
canonical
}
else -> canonical
}
if (queue != null) {
migrateInFlightOwnership(conversationID, peerID, queue)
}
return queue
}
private fun findInFlightAttempt(
conversationID: String,
peerID: String,
queue: List<QueuedMessage>
): Map.Entry<String, InFlightNostrAttempt>? {
val queuedMessageIDs = queue.asSequence()
.map(QueuedMessage::messageID)
.toSet()
return inFlightNostrAttempts.entries.firstOrNull { (key, attempt) ->
key == conversationID ||
key == peerID ||
ContactDirectory.canonicalConversationId(key) == conversationID ||
attempt.messageID in queuedMessageIDs
}
}
private fun migrateInFlightOwnership(
conversationID: String,
peerID: String,
queue: List<QueuedMessage>
) {
val matches = inFlightNostrAttempts.entries.filter { (key, attempt) ->
key == peerID ||
ContactDirectory.canonicalConversationId(key) == conversationID ||
queue.any { it.messageID == attempt.messageID }
}
if (matches.size != 1 || inFlightNostrAttempts.containsKey(conversationID)) {
return
}
val match = matches.single()
if (inFlightNostrAttempts.remove(match.key, match.value)) {
inFlightNostrAttempts[conversationID] = match.value
}
}
private fun removeEmptyQueue(
conversationID: String,
peerID: String,
queued: MutableList<QueuedMessage>
) {
outbox.remove(conversationID, queued)
outbox.remove(peerID, queued)
retryState.remove(conversationID)
retryState.remove(peerID)
}
// Flush everything (rarely used)
fun flushAllOutbox() {
outbox.keys.toList().forEach { flushOutboxFor(it) }
@ -227,9 +428,21 @@ class MessageRouter private constructor(
@Synchronized
private fun enqueue(conversationID: String, entry: QueuedMessage) {
val queue = outbox.getOrPut(conversationID) { mutableListOf() }
if (queue.any { it.messageID == entry.messageID }) return
queue.add(entry)
while (queue.size > OUTBOX_MAX_PER_PEER) {
val evicted = queue.removeAt(0)
val inFlightMessageIDs = inFlightNostrAttempts
.filter { (key, attempt) ->
key == conversationID ||
ContactDirectory.canonicalConversationId(key) == conversationID ||
queue.any { it.messageID == attempt.messageID }
}
.values
.mapTo(mutableSetOf(), InFlightNostrAttempt::messageID)
val evictionIndex =
queue.indexOfFirst { it.messageID !in inFlightMessageIDs }
if (evictionIndex < 0) break
val evicted = queue.removeAt(evictionIndex)
Log.w(TAG, "Outbox full for ${conversationID.take(16)}…; evicting oldest msg_id=${evicted.messageID.take(8)}…")
notifyExpired(evicted.messageID)
}
@ -239,6 +452,14 @@ class MessageRouter private constructor(
try { onMessageExpired?.invoke(messageID) } catch (_: Exception) { }
}
private fun notifyAdmitted(messageID: String) {
try { onMessageAdmitted?.invoke(messageID) } catch (_: Exception) { }
}
private fun notifyFailed(messageID: String, reason: String) {
try { onMessageFailed?.invoke(messageID, reason) } catch (_: Exception) { }
}
/**
* Initiate a Noise handshake for a conversation with queued messages, applying
* exponential backoff between attempts. [immediate] resets the backoff (peer just
@ -285,6 +506,32 @@ class MessageRouter private constructor(
schedulerJob = null
}
/** Panic/reset must invalidate callbacks from the previous account and drop plaintext. */
@Synchronized
fun discardForAccountReset(): Long {
stopOutboxScheduler()
accountResetBlocked = true
outboxEpoch += 1
inFlightNostrAttempts.clear()
outbox.clear()
retryState.clear()
return outboxEpoch
}
@Synchronized
fun completeAccountReset(resetToken: Long): Boolean {
if (resetToken != outboxEpoch) return false
accountResetBlocked = false
startOutboxScheduler()
return true
}
internal val queuedMessageCount: Int
@Synchronized get() = outbox.values.sumOf { it.size }
internal val inFlightNostrAttemptCount: Int
@Synchronized get() = inFlightNostrAttempts.size
internal val isSchedulerRunning: Boolean get() = schedulerJob?.isActive == true
/**
@ -319,10 +566,20 @@ class MessageRouter private constructor(
private fun expireOldEntries(conversationID: String, nowMs: Long) {
val queued = outbox[conversationID] ?: return
val inFlightMessageIDs = inFlightNostrAttempts
.filter { (key, attempt) ->
key == conversationID ||
ContactDirectory.canonicalConversationId(key) == conversationID ||
queued.any { it.messageID == attempt.messageID }
}
.values
.mapTo(mutableSetOf(), InFlightNostrAttempt::messageID)
val iterator = queued.iterator()
while (iterator.hasNext()) {
val entry = iterator.next()
if (nowMs - entry.enqueuedAtMs > OUTBOX_MESSAGE_TTL_MS) {
if (entry.messageID !in inFlightMessageIDs &&
nowMs - entry.enqueuedAtMs > OUTBOX_MESSAGE_TTL_MS
) {
Log.w(TAG, "Expiring queued PM for ${conversationID.take(16)}… msg_id=${entry.messageID.take(8)}…")
iterator.remove()
notifyExpired(entry.messageID)
@ -335,6 +592,7 @@ class MessageRouter private constructor(
}
private fun canSendViaNostr(peerID: String): Boolean {
canSendViaNostrOverride?.let { return it(peerID) }
return try {
val resolution = ContactDirectory.resolve(peerID)
if (resolution.isMutualFavorite && resolution.nostrPubkey != null) return true

67
app/src/main/java/com/bitchat/android/ui/ChatViewModel.kt
View File

@ -375,14 +375,30 @@ class ChatViewModel(
loadAndInitialize()
ContactDirectory.initialize(getApplication()) { mesh }
com.bitchat.android.services.AppStateStore.canonicalizePrivateChats()
// Mark queued private messages as failed when the router gives up on them
// Reflect asynchronous router admission/failure into the local echo.
try {
com.bitchat.android.services.MessageRouter.getInstance(getApplication(), mesh).onMessageExpired = { messageID ->
val router = com.bitchat.android.services.MessageRouter.getInstance(
getApplication(),
mesh
)
router.onMessageExpired = { messageID ->
messageManager.updateMessageDeliveryStatus(
messageID,
com.bitchat.android.model.DeliveryStatus.Failed("Message expired before delivery")
)
}
router.onMessageAdmitted = { messageID ->
messageManager.updateMessageDeliveryStatus(
messageID,
com.bitchat.android.model.DeliveryStatus.Sent
)
}
router.onMessageFailed = { messageID, reason ->
messageManager.updateMessageDeliveryStatus(
messageID,
com.bitchat.android.model.DeliveryStatus.Failed(reason)
)
}
} catch (_: Exception) { }
// Hydrate UI state from process-wide AppStateStore to survive Activity recreation
viewModelScope.launch {
@ -718,6 +734,8 @@ class ChatViewModel(
) { messageContent, peerID, recipientNicknameParam, messageId ->
val router = com.bitchat.android.services.MessageRouter.getInstance(getApplication(), mesh)
val route = router.sendPrivate(messageContent, peerID, recipientNicknameParam, messageId)
// Geohash DMs retain their legacy fire-and-forget admission semantics.
// Standard Nostr/NDR sends are marked Sent only by onMessageAdmitted.
if (route == com.bitchat.android.services.MessageRouter.RouteResult.NOSTR) {
messageManager.updateMessageDeliveryStatus(messageId, com.bitchat.android.model.DeliveryStatus.Sent)
}
@ -1364,20 +1382,35 @@ class ChatViewModel(
fun panicClearAllData() {
Log.w(TAG, "🚨 PANIC MODE ACTIVATED - Clearing all sensitive data")
// Freeze transport admission before erasing router plaintext so a
// concurrent UI send cannot create fresh work in the reset window.
val resetTransport = com.bitchat.android.nostr.NostrTransport
.getInstance(getApplication())
val transportResetToken = resetTransport.discardForAccountReset()
val resetRouter = com.bitchat.android.services.MessageRouter
.getInstance(getApplication(), mesh)
val routerResetToken = resetRouter.discardForAccountReset()
geohashViewModel.invalidateDoubleRatchetAccount()
try {
com.bitchat.android.geohash.LocationChannelManager
.getInstance(getApplication())
.disableLocationServices()
} catch (_: Exception) { }
val resetRelay = com.bitchat.android.nostr.NostrRelayManager
.getInstance(getApplication())
val relayResetToken = resetRelay.beginAccountReset()
// A pending one-shot downgrade confirmation must not survive panic or
// become actionable against the fresh post-wipe identity.
mediaSendingManager.clearPendingPrivateMediaConsent()
geohashViewModel.invalidateDoubleRatchetAccount()
val ndrResetSucceeded = ndrService.resetForPanic()
if (!ndrResetSucceeded) {
Log.e(TAG, "NDR storage wipe was incomplete; NDR remains disabled for this process")
}
// NDR reset is synchronized and therefore quiesces any native send that
// entered before panic. Advance the relay generation only afterwards,
// then clear every event produced by that old runtime.
resetRelay.discardForAccountReset(relayResetToken)
ndrBootstrapAttemptMs.clear()
ndrNoiseHandshakeAttemptMs.clear()
ndrInviteRetries.cancelAll()
@ -1453,14 +1486,10 @@ class ChatViewModel(
locationManager.clearPersistedChannel()
} catch (_: Exception) { }
geohashViewModel.panicReset()
geohashViewModel.panicReset(reinitialize = false)
} catch (e: Exception) {
Log.e(TAG, "Failed to reset Nostr/geohash: ${e.message}")
}
if (ndrResetSucceeded && NdrFeatureGate.isEnabled()) {
// GeohashViewModel.panicReset() recreates the account identity and
// reinstalls the decrypted-message callback through initialize().
}
// Reset nickname
val newNickname = "anon${Random.nextInt(1000, 9999)}"
@ -1469,6 +1498,24 @@ class ChatViewModel(
// Recreate mesh service with fresh identity
recreateMeshServiceAfterPanic()
try {
resetTransport.senderPeerID = mesh.myPeerID
val transportReopened =
resetTransport.completeAccountReset(transportResetToken)
val routerReopened =
resetRouter.completeAccountReset(routerResetToken)
val relayReopened =
resetRelay.completeAccountReset(relayResetToken)
if (!transportReopened || !routerReopened || !relayReopened) {
Log.w(TAG, "A newer account reset superseded panic reinitialization")
return
}
geohashViewModel.initialize()
mesh.startServices()
mesh.sendBroadcastAnnounce()
} catch (e: Exception) {
Log.e(TAG, "Failed to reopen network transports after panic: ${e.message}")
}
Log.w(TAG, "🚨 PANIC MODE COMPLETED - New identity: ${mesh.myPeerID}")
}
@ -1492,10 +1539,6 @@ class ChatViewModel(
unifiedMeshService = freshUnifiedMeshService
mesh.delegate = this
// Restart mesh operations with new identity
mesh.startServices()
mesh.sendBroadcastAnnounce()
Log.d(
TAG,
"✅ Mesh service recreated. Old peerID: $oldPeerID, New peerID: ${mesh.myPeerID}"

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

@ -113,13 +113,16 @@ class GeohashViewModel(
val identity = NostrIdentityBridge.getCurrentNostrIdentity(getApplication())
if (identity != null) {
// Configure the singleton only for the account identity, never a derived geohash key.
dmHandler.configureDoubleRatchet(identity)
val accountEpoch = dmHandler.configureAccount(identity)
// Use global chat-messages only for full account DMs (mesh context). For geohash DMs, subscribe per-geohash below.
subscriptionManager.subscribeGiftWraps(
pubkey = identity.publicKeyHex,
sinceMs = System.currentTimeMillis() - 172800000L,
id = "chat-messages",
handler = { event -> dmHandler.onGiftWrap(event, "", identity) } // geohash="" means global account DM (not geohash identity)
handler = { event ->
// geohash="" means global account DM (not geohash identity).
dmHandler.onGiftWrap(event, "", identity, accountEpoch)
}
)
}
try {
@ -194,7 +197,8 @@ class GeohashViewModel(
}
}
fun panicReset() {
fun panicReset(reinitialize: Boolean = true) {
dmHandler.invalidateAccount()
repo.clearAll()
GeohashAliasRegistry.clear()
GeohashConversationRegistry.clear()
@ -209,11 +213,17 @@ class GeohashViewModel(
globalPresenceJob?.cancel()
globalPresenceJob = null
try { NostrIdentityBridge.clearAllAssociations(getApplication()) } catch (_: Exception) {}
initialize()
// Install the replacement account key while relay/transport admission is
// still blocked. The caller may defer network initialization until the
// rest of the account identity has been recreated.
try { NostrIdentityBridge.getCurrentNostrIdentity(getApplication()) } catch (_: Exception) {}
if (reinitialize) {
initialize()
}
}
fun invalidateDoubleRatchetAccount() {
dmHandler.invalidateDoubleRatchetAccount()
dmHandler.invalidateAccount()
}
private suspend fun broadcastLiveLocationPresence(geohash: String) {
@ -249,8 +259,19 @@ class GeohashViewModel(
}
fun sendGeohashMessage(content: String, channel: com.bitchat.android.geohash.GeohashChannel, myPeerID: String, nickname: String?) {
viewModelScope.launch {
val accountEpoch = dmHandler.currentAccountEpoch() ?: return
val accountContext =
com.bitchat.android.nostr.NostrInboundAccountLifecycle
.contextFor(accountEpoch)
?: return
val relayManager = NostrRelayManager.getInstance(getApplication())
val relayGeneration = relayManager.captureAccountGeneration()
viewModelScope.launch(Dispatchers.Default + accountContext.receiveJob) {
try {
if (!com.bitchat.android.nostr.NostrInboundAccountLifecycle
.isCurrent(accountEpoch) ||
!relayManager.isAccountGenerationCurrent(relayGeneration)
) return@launch
val canUseChannel = locationChannelManager
?.canUseSelectedLocationChannel(channel) == true
if (!canUseChannel) {
@ -274,11 +295,22 @@ class GeohashViewModel(
channel = "#${channel.geohash}",
powDifficulty = if (pow.enabled) pow.difficulty else null
)
messageManager.addChannelMessage("geo:${channel.geohash}", localMsg)
val localEchoAdded =
com.bitchat.android.nostr.NostrInboundAccountLifecycle
.runIfCurrent(accountEpoch) {
messageManager.addChannelMessage(
"geo:${channel.geohash}",
localMsg
)
}
if (!localEchoAdded) return@launch
val identity = NostrIdentityBridge.deriveIdentity(
forGeohash = channel.geohash,
context = getApplication()
)
if (!com.bitchat.android.nostr.NostrInboundAccountLifecycle
.isCurrent(accountEpoch)
) return@launch
val teleported = locationChannelManager?.teleported?.value
?: state.isTeleported.value
val event = NostrProtocol.createEphemeralGeohashEvent(
@ -288,13 +320,13 @@ class GeohashViewModel(
nickname,
teleported
)
val relayManager = NostrRelayManager.getInstance(getApplication())
relayManager.sendEventToGeohash(
event,
channel.geohash,
includeDefaults = false,
nRelays = 5,
liveLocationToken = liveLocationToken
liveLocationToken = liveLocationToken,
expectedAccountGeneration = relayGeneration
)
} catch (e: Exception) {
Log.e(TAG, "Failed to send geohash message: ${e.message}")
@ -515,13 +547,16 @@ class GeohashViewModel(
geohash: String,
liveLocationToken: Long?
) {
val accountEpoch = dmHandler.currentAccountEpoch() ?: return
val subId = "geohash-${UUID.randomUUID()}"; currentGeohashMsgSubId = subId
subscriptionManager.subscribeGeohashMessages(
geohash = geohash,
sinceMs = System.currentTimeMillis() - 3600000L,
limit = 200,
id = subId,
handler = { event -> geohashMessageHandler.onEvent(event, geohash) },
handler = { event ->
geohashMessageHandler.onEvent(event, geohash, accountEpoch)
},
liveLocationToken = liveLocationToken
)
}
@ -535,13 +570,16 @@ class GeohashViewModel(
geohash: String,
liveLocationToken: Long?
) {
val accountEpoch = dmHandler.currentAccountEpoch() ?: return
val subId = "geohash-presence-${UUID.randomUUID()}"; currentGeohashPresenceSubId = subId
subscriptionManager.subscribeGeohashPresence(
geohash = geohash,
sinceMs = System.currentTimeMillis() - 3600000L,
limit = 200,
id = subId,
handler = { event -> geohashMessageHandler.onEvent(event, geohash) },
handler = { event ->
geohashMessageHandler.onEvent(event, geohash, accountEpoch)
},
liveLocationToken = liveLocationToken
)
}
@ -552,6 +590,11 @@ class GeohashViewModel(
*/
private fun subscribeChannelDM(geohash: String) {
val dmIdentity = NostrIdentityBridge.deriveIdentity(geohash, getApplication())
val accountEpoch = dmHandler.currentAccountEpoch()
if (accountEpoch == null) {
Log.w(TAG, "Skipping geohash DM subscription without an active account epoch")
return
}
val dmSubId = "geo-dm-${UUID.randomUUID()}"
currentDmSubId = dmSubId
currentDmGeohash = geohash
@ -559,7 +602,9 @@ class GeohashViewModel(
pubkey = dmIdentity.publicKeyHex,
sinceMs = System.currentTimeMillis() - 172800000L,
id = dmSubId,
handler = { event -> dmHandler.onGiftWrap(event, geohash, dmIdentity) }
handler = { event ->
dmHandler.onGiftWrap(event, geohash, dmIdentity, accountEpoch)
}
)
// Also register alias in global registry for routing convenience
GeohashAliasRegistry.put("nostr_${dmIdentity.publicKeyHex.take(16)}", dmIdentity.publicKeyHex)
@ -630,6 +675,7 @@ class GeohashViewModel(
}
private fun performSubscribeSampling(geohash: String) {
val accountEpoch = dmHandler.currentAccountEpoch() ?: return
val subscriptionId = samplingSubscriptionIds.getOrPut(geohash) {
"sampling-${UUID.randomUUID()}"
}
@ -642,7 +688,9 @@ class GeohashViewModel(
sinceMs = System.currentTimeMillis() - 86400000L,
limit = 200,
id = subscriptionId,
handler = { event -> geohashMessageHandler.onEvent(event, geohash) }
handler = { event ->
geohashMessageHandler.onEvent(event, geohash, accountEpoch)
}
)
}
@ -662,7 +710,13 @@ class GeohashViewModel(
sinceMs = System.currentTimeMillis() - 86400000L,
limit = 200,
id = subscriptionId,
handler = { event -> geohashMessageHandler.onEvent(event, geohash) },
handler = { event ->
geohashMessageHandler.onEvent(
event,
geohash,
accountEpoch
)
},
liveLocationToken = token
)
}

9
app/src/main/java/com/bitchat/android/ui/MessageManager.kt
View File

@ -238,6 +238,15 @@ class MessageManager(private val state: ChatState) {
}
private fun chooseStatus(old: DeliveryStatus?, new: DeliveryStatus): DeliveryStatus? {
if (new is DeliveryStatus.Failed) {
// A locally queued message may fail before any transport admits it.
// Never let a late local failure overwrite evidence of handoff/delivery.
return if (old == null || old is DeliveryStatus.Sending || old is DeliveryStatus.Failed) {
new
} else {
old
}
}
// Never downgrade (e.g., Read -> Delivered). Keep the higher priority.
return if (statusPriority(new) >= statusPriority(old)) new else old
}

87
app/src/test/kotlin/com/bitchat/android/nostr/NdrNostrServiceTest.kt
View File

@ -1,6 +1,7 @@
package com.bitchat.android.nostr
import com.bitchat.android.model.NdrFeatureGate
import kotlinx.coroutines.Job
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
@ -18,14 +19,47 @@ class NdrNostrServiceTest {
@Before
fun enableNdrForTest() {
NostrInboundAccountLifecycle.invalidate()
NdrFeatureGate.setEnabledForTests(true)
}
@After
fun resetNdrGate() {
NostrInboundAccountLifecycle.invalidate()
NdrFeatureGate.setEnabledForTests(false)
}
@Test
fun processExitInvalidatesAccountWideReceiveWork() {
val parentJob = Job()
val accountContext =
NostrInboundAccountLifecycle.begin(localPubkey, parentJob)
val service = service(runtimeFactory = FakeNdrRuntimeFactory(FakeNdrPairwiseRuntime()))
assertTrue(NostrInboundAccountLifecycle.isCurrent(accountContext.epoch))
assertTrue(accountContext.receiveJob.isActive)
service.shutdownForProcessExit()
assertFalse(NostrInboundAccountLifecycle.isCurrent(accountContext.epoch))
assertTrue(accountContext.receiveJob.isCancelled)
parentJob.cancel()
}
@Test
fun processExitCannotReopenPairwiseRuntime() {
val factory = FakeNdrRuntimeFactory(FakeNdrPairwiseRuntime())
val service = service(runtimeFactory = factory)
val identity = testIdentity()
assertTrue(service.configureIfNeeded(identity))
service.shutdownForProcessExit()
assertFalse(service.configureIfNeeded(identity))
assertFalse(service.retirePeerForMaintenance(identity, peerPubkey))
assertEquals(1, factory.createdCount)
}
@Test
fun disabledRolloutGateRefusesToCreateRuntime() {
NdrFeatureGate.setEnabledForTests(false)
@ -902,6 +936,27 @@ class NdrNostrServiceTest {
assertTrue(relay.sentEvents.isEmpty())
}
@Test
fun postCommitMarkerFailureStillReportsDurableNdrAdmission() {
val activePeers = mutableSetOf<String>()
val runtime = FakeNdrPairwiseRuntime(activePeers)
val markers = InMemoryMarkerStore()
val service = service(
runtimeFactory = FakeNdrRuntimeFactory(runtime),
markerStore = markers
)
service.configureIfNeeded(testIdentity())
activePeers += peerPubkey
markers.markFailure = java.io.IOException("marker storage unavailable")
val result = service.sendIfPossible("admitted-once", peerPubkey)
assertEquals(NdrSendResult.SENT, result)
assertEquals(listOf(peerPubkey), runtime.sendTextCalls)
assertTrue(runtime.sendTextCompleted)
assertTrue(runtime.destroyed)
}
@Test
fun persistedHalfReadyPairwiseSessionNeverFallsBackAfterRestart() {
val markers = InMemoryMarkerStore()
@ -1100,6 +1155,38 @@ class NdrNostrServiceTest {
)
}
@Test
fun ndrAdmissionDispositionRetriesEveryProtectedOrUnreadyState() {
assertEquals(
NdrSendDisposition.ADMITTED,
ndrSendDisposition(NdrSendResult.SENT)
)
assertEquals(
NdrSendDisposition.LEGACY_FALLBACK,
ndrSendDisposition(NdrSendResult.NO_SESSION)
)
assertEquals(
NdrSendDisposition.RETRYABLE,
ndrSendDisposition(NdrSendResult.NO_SESSION, ndrRequired = true)
)
assertEquals(
NdrSendDisposition.RETRYABLE,
ndrSendDisposition(NdrSendResult.NO_SESSION, pairwiseOnly = true)
)
assertEquals(
NdrSendDisposition.RETRYABLE,
ndrSendDisposition(NdrSendResult.FAILED)
)
assertEquals(
NdrSendDisposition.RETRYABLE,
ndrSendDisposition(
NdrSendResult.NO_SESSION,
ndrRequired = true,
rebindBlocked = true
)
)
}
private fun service(
relayManager: FakeRelayManager = FakeRelayManager(),
runtimeFactory: NdrPairwiseRuntimeFactory,

110
app/src/test/kotlin/com/bitchat/android/nostr/NostrDirectMessageHandlerTest.kt
View File

@ -1,6 +1,7 @@
package com.bitchat.android.nostr
import android.os.Build
import com.bitchat.android.model.NdrFeatureGate
import com.bitchat.android.services.AppStateStore
import com.bitchat.android.services.SeenMessageStore
import com.bitchat.android.ui.ChatState
@ -23,6 +24,7 @@ import kotlinx.coroutines.test.setMain
import kotlinx.coroutines.withTimeout
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import org.junit.runner.RunWith
@ -32,6 +34,9 @@ import org.mockito.kotlin.whenever
import org.robolectric.RobolectricTestRunner
import org.robolectric.RuntimeEnvironment
import org.robolectric.annotation.Config
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [Build.VERSION_CODES.P], manifest = Config.NONE)
@ -44,11 +49,15 @@ class NostrDirectMessageHandlerTest {
fun setUp() {
Dispatchers.setMain(UnconfinedTestDispatcher())
scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
NdrFeatureGate.setEnabledForTests(false)
NostrInboundAccountLifecycle.invalidate()
AppStateStore.clear()
}
@After
fun tearDown() {
NostrInboundAccountLifecycle.invalidate()
NdrFeatureGate.setEnabledForTests(false)
AppStateStore.clear()
scope.cancel()
Dispatchers.resetMain()
@ -115,9 +124,10 @@ class NostrDirectMessageHandlerTest {
giftWrapCreatedAt = now - 86_400
)
handler.onGiftWrap(first, "", recipient)
val accountEpoch = handler.configureAccount(recipient)
handler.onGiftWrap(first, "", recipient, accountEpoch)
waitForMessage(state, firstId)
handler.onGiftWrap(second, "", recipient)
handler.onGiftWrap(second, "", recipient, accountEpoch)
waitForMessage(state, secondId)
val messages = state.getPrivateChatsValue().values.single()
@ -170,7 +180,8 @@ class NostrDirectMessageHandlerTest {
giftWrapCreatedAt = now - 5
)
handler.onGiftWrap(giftWrap, "", recipient)
val accountEpoch = handler.configureAccount(recipient)
handler.onGiftWrap(giftWrap, "", recipient, accountEpoch)
kotlinx.coroutines.runBlocking {
assertEquals(sender.publicKeyHex, withTimeout(5_000) { policyChecked.await() })
@ -179,6 +190,99 @@ class NostrDirectMessageHandlerTest {
assertEquals(0, state.getPrivateChatsValue().values.flatten().size)
}
@Test
fun `account invalidation prevents an in-flight legacy gift wrap from restoring chat state`() {
val application = RuntimeEnvironment.getApplication()
val state = ChatState(scope).apply { setNickname("recipient") }
val dataManager = DataManager(application)
val privateChatManager = PrivateChatManager(
state = state,
messageManager = MessageManager(state),
dataManager = dataManager,
noiseSessionDelegate = mock<NoiseSessionDelegate>()
)
val seenStore = mock<SeenMessageStore>()
whenever(seenStore.hasDelivered(any())).thenReturn(true)
whenever(seenStore.hasBeenReadLocally(any())).thenReturn(false)
val policyEntered = CountDownLatch(1)
val releasePolicy = CountDownLatch(1)
val blockFirstPolicyCheck = AtomicBoolean(true)
val handler = NostrDirectMessageHandler(
application = application,
state = state,
privateChatManager = privateChatManager,
meshDelegateHandler = mock<MeshDelegateHandler>(),
scope = scope,
repo = GeohashRepository(application, state, dataManager),
dataManager = dataManager,
seenStoreProvider = { seenStore },
legacyNostrInboundAllowed = {
if (blockFirstPolicyCheck.compareAndSet(true, false)) {
policyEntered.countDown()
releasePolicy.await(5, TimeUnit.SECONDS)
}
true
}
)
val sender = NostrIdentity.generate()
val recipient = NostrIdentity.generate()
val now = (System.currentTimeMillis() / 1000).toInt()
val staleGiftWrap = privateMessageGiftWrap(
content = requireNotNull(
NostrEmbeddedBitChat.encodePMForNostrNoRecipient(
content = "must-not-return",
messageID = "stale-legacy",
senderPeerID = "0011223344556677"
)
),
sender = sender,
recipient = recipient,
rumorCreatedAt = now - 60,
giftWrapCreatedAt = now - 5
)
val freshGiftWrap = privateMessageGiftWrap(
content = requireNotNull(
NostrEmbeddedBitChat.encodePMForNostrNoRecipient(
content = "fresh",
messageID = "fresh-legacy",
senderPeerID = "0011223344556677"
)
),
sender = sender,
recipient = recipient,
rumorCreatedAt = now - 30,
giftWrapCreatedAt = now - 4
)
val staleEpoch = handler.configureAccount(recipient)
val staleJob = requireNotNull(
handler.onGiftWrap(staleGiftWrap, "", recipient, staleEpoch)
)
try {
assertTrue(policyEntered.await(5, TimeUnit.SECONDS))
handler.invalidateAccount()
} finally {
releasePolicy.countDown()
}
kotlinx.coroutines.runBlocking {
withTimeout(5_000) { staleJob.join() }
}
assertEquals(0, state.getPrivateChatsValue().values.flatten().size)
val freshEpoch = handler.configureAccount(recipient)
val freshJob = requireNotNull(
handler.onGiftWrap(freshGiftWrap, "", recipient, freshEpoch)
)
kotlinx.coroutines.runBlocking {
withTimeout(5_000) { freshJob.join() }
}
waitForMessage(state, "fresh-legacy")
assertEquals(
listOf("fresh-legacy"),
state.getPrivateChatsValue().values.flatten().map { it.id }
)
}
private fun waitForMessage(state: ChatState, messageId: String) {
kotlinx.coroutines.runBlocking {
withTimeout(5_000) {

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

@ -12,6 +12,8 @@ class NostrRelayManagerLifecycleSmokeTest {
@Test
fun `disconnected manager maintains subscription and empty publish invariants locally`() {
val manager = NostrRelayManager.shared
val setupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(setupReset))
manager.disconnect()
manager.clearAllSubscriptions()
@ -38,6 +40,114 @@ class NostrRelayManagerLifecycleSmokeTest {
assertFalse(manager.isConnected.value)
}
@Test
fun `account reset advances generation and discards queued legacy gift wraps`() {
val manager = NostrRelayManager.shared
val setupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(setupReset))
val oldGeneration = manager.accountGenerationForTesting()
val event = signedEvent()
assertTrue(manager.registerPendingGiftWrap(event.id, oldGeneration))
manager.sendEvent(event, relayUrls = emptyList())
assertEquals(1, manager.queuedEventCountForTesting())
assertEquals(1, manager.pendingGiftWrapCountForTesting())
val resetToken = manager.discardForAccountReset()
assertEquals(oldGeneration + 1, manager.accountGenerationForTesting())
assertFalse(manager.isAccountGenerationCurrent(oldGeneration))
assertFalse(
manager.isAccountGenerationCurrent(
manager.accountGenerationForTesting()
)
)
assertEquals(0, manager.queuedEventCountForTesting())
assertEquals(0, manager.pendingGiftWrapCountForTesting())
assertTrue(manager.completeAccountReset(resetToken))
}
@Test
fun `new relay work is rejected during reset and admitted after completion`() {
val manager = NostrRelayManager.shared
val resetToken = manager.discardForAccountReset()
val event = signedEvent()
assertFalse(manager.sendEvent(event, relayUrls = emptyList()))
assertEquals(0, manager.queuedEventCountForTesting())
assertTrue(manager.completeAccountReset(resetToken))
assertTrue(manager.sendEvent(event, relayUrls = emptyList()))
assertEquals(1, manager.queuedEventCountForTesting())
val cleanupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(cleanupReset))
}
@Test
fun `stale reset cannot clear or reopen a newer relay reset`() {
val manager = NostrRelayManager.shared
val firstReset = manager.beginAccountReset()
val secondReset = manager.beginAccountReset()
val event = signedEvent()
assertFalse(manager.discardForAccountReset(firstReset))
assertFalse(manager.completeAccountReset(firstReset))
assertFalse(manager.sendEvent(event, relayUrls = emptyList()))
assertEquals(0, manager.queuedEventCountForTesting())
assertTrue(manager.discardForAccountReset(secondReset))
assertTrue(manager.completeAccountReset(secondReset))
assertTrue(manager.sendEvent(event, relayUrls = emptyList()))
val cleanupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(cleanupReset))
}
@Test
fun `stale geohash wrapper cannot recapture a replacement generation`() {
val manager = NostrRelayManager.shared
val setupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(setupReset))
val oldGeneration = manager.captureAccountGeneration()
val replacementReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(replacementReset))
val event = signedEvent()
manager.sendEventToGeohash(
event = event,
geohash = "u4pruydq",
expectedAccountGeneration = oldGeneration
)
manager.subscribe(
filter = NostrFilter(kinds = listOf(NostrKind.TEXT_NOTE)),
id = "stale-wrapper",
handler = {},
targetRelayUrls = emptyList(),
expectedAccountGeneration = oldGeneration
)
assertEquals(0, manager.queuedEventCountForTesting())
assertFalse(manager.getActiveSubscriptions().containsKey("stale-wrapper"))
val cleanupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(cleanupReset))
}
@Test
fun `generation captured during reset never becomes replacement work`() {
val manager = NostrRelayManager.shared
val resetToken = manager.beginAccountReset()
val parkedGeneration = manager.captureAccountGeneration()
assertTrue(manager.discardForAccountReset(resetToken))
assertTrue(manager.completeAccountReset(resetToken))
assertFalse(manager.isAccountGenerationCurrent(parkedGeneration))
val cleanupReset = manager.discardForAccountReset()
assertTrue(manager.completeAccountReset(cleanupReset))
}
private fun signedEvent(): NostrEvent {
val privateKey = "0".repeat(63) + "1"
return NostrEvent(

27
app/src/test/kotlin/com/bitchat/android/nostr/NostrRelaySubscriptionRaceTest.kt
View File

@ -9,6 +9,7 @@ import okhttp3.WebSocket
import okio.ByteString
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertThrows
import org.junit.Assert.assertTrue
import org.junit.Test
import org.junit.runner.RunWith
@ -93,6 +94,26 @@ class NostrRelaySubscriptionRaceTest {
assertFalse(accepted ?: true)
}
@Test
fun ndrAdapterRejectsSubscriptionDuringAccountReset() {
val manager = NostrRelayManager(
CoroutineScope(Dispatchers.Unconfined + SupervisorJob()),
NostrEventDeduplicator(maxCapacity = 8)
)
val adapter = BitchatNdrRelayAdapter(manager)
val resetToken = manager.beginAccountReset()
assertThrows(IllegalStateException::class.java) {
adapter.subscribe(
filter = NostrFilter(kinds = listOf(1060)),
id = "blocked-ndr"
) { true }
}
assertTrue(manager.discardForAccountReset(resetToken))
assertTrue(manager.completeAccountReset(resetToken))
}
@Suppress("UNCHECKED_CAST")
private fun installConnection(
manager: NostrRelayManager,
@ -114,13 +135,15 @@ class NostrRelaySubscriptionRaceTest {
val method = NostrRelayManager::class.java.getDeclaredMethod(
"handleMessage",
String::class.java,
String::class.java
String::class.java,
Long::class.javaPrimitiveType
)
method.isAccessible = true
method.invoke(
manager,
"""["EVENT","$subscriptionId",${event.toJsonString()}]""",
RELAY_URL
RELAY_URL,
manager.captureAccountGeneration()
)
}

141
app/src/test/kotlin/com/bitchat/android/nostr/NostrTransportAdmissionTest.kt Normal file
View File

@ -0,0 +1,141 @@
package com.bitchat.android.nostr
import android.os.Build
import com.bitchat.android.model.ReadReceipt
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.advanceUntilIdle
import org.junit.Assert.assertEquals
import org.junit.Test
import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.RuntimeEnvironment
import org.robolectric.annotation.Config
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [Build.VERSION_CODES.P], manifest = Config.NONE)
@OptIn(ExperimentalCoroutinesApi::class)
class NostrTransportAdmissionTest {
@Test
fun `cancelled transport scope completes admission exactly once as retryable`() {
val job = Job().apply { cancel() }
val transport = NostrTransport(
context = RuntimeEnvironment.getApplication(),
transportScope = CoroutineScope(Dispatchers.Unconfined + job)
)
val admissions = mutableListOf<NostrSendAdmission>()
transport.sendPrivateMessage(
content = "hello",
to = "aa".repeat(32),
recipientNickname = "peer",
messageID = "cancelled",
completion = admissions::add
)
assertEquals(listOf(NostrSendAdmission.RETRYABLE), admissions)
}
@Test
fun `already expired payload is terminal before identity or relay work`() {
val transport = NostrTransport(
context = RuntimeEnvironment.getApplication(),
transportScope = CoroutineScope(Dispatchers.Unconfined + Job())
)
val admissions = mutableListOf<NostrSendAdmission>()
transport.sendPrivateMessage(
content = "expired",
to = "aa".repeat(32),
recipientNickname = "peer",
messageID = "expired",
expiresAtSeconds = 0uL,
completion = admissions::add
)
assertEquals(listOf(NostrSendAdmission.TERMINAL_FAILED), admissions)
}
@Test
fun `invocation before relay reset cannot enqueue when coroutine resumes after reset`() {
val dispatcher = StandardTestDispatcher()
val scope = TestScope(dispatcher)
val relayManager = NostrRelayManager(scope)
val setupReset = relayManager.discardForAccountReset()
relayManager.completeAccountReset(setupReset)
val oldGeneration = relayManager.accountGenerationForTesting()
val transport = NostrTransport(
context = RuntimeEnvironment.getApplication(),
transportScope = scope,
relayManager = relayManager
)
transport.sendDeliveryAckGeohash(
messageID = "old-account-message",
toRecipientHex = "22".repeat(32),
fromIdentity = NostrIdentity.fromPrivateKey("11".repeat(32))
)
val resetToken = relayManager.discardForAccountReset()
scope.advanceUntilIdle()
assertEquals(oldGeneration + 1, relayManager.accountGenerationForTesting())
assertEquals(0, relayManager.queuedEventCountForTesting())
assertEquals(0, relayManager.pendingGiftWrapCountForTesting())
relayManager.completeAccountReset(resetToken)
}
@Test
fun `account reset clears throttled reads and invalidates their delayed work`() {
val dispatcher = StandardTestDispatcher()
val scope = TestScope(dispatcher)
val relayManager = NostrRelayManager(scope)
val setupReset = relayManager.discardForAccountReset()
relayManager.completeAccountReset(setupReset)
val transport = NostrTransport(
context = RuntimeEnvironment.getApplication(),
transportScope = scope,
relayManager = relayManager
)
transport.sendReadReceipt(ReadReceipt("first"), "aa".repeat(32))
transport.sendReadReceipt(ReadReceipt("second"), "aa".repeat(32))
assertEquals(1, transport.activeReadCountForTesting())
assertEquals(1, transport.queuedReadCountForTesting())
transport.discardForAccountReset()
val resetToken = relayManager.discardForAccountReset()
scope.advanceUntilIdle()
assertEquals(0, transport.activeReadCountForTesting())
assertEquals(0, transport.queuedReadCountForTesting())
assertEquals(0, relayManager.queuedEventCountForTesting())
relayManager.completeAccountReset(resetToken)
}
@Test
fun `stale transport reset cannot reopen admission`() {
val transport = NostrTransport(
context = RuntimeEnvironment.getApplication(),
transportScope = CoroutineScope(Dispatchers.Unconfined + Job())
)
val firstReset = transport.discardForAccountReset()
val secondReset = transport.discardForAccountReset()
val admissions = mutableListOf<NostrSendAdmission>()
assertEquals(false, transport.completeAccountReset(firstReset))
transport.sendPrivateMessage(
content = "blocked",
to = "aa".repeat(32),
recipientNickname = "peer",
messageID = "stale-reset",
completion = admissions::add
)
assertEquals(listOf(NostrSendAdmission.RETRYABLE), admissions)
assertEquals(true, transport.completeAccountReset(secondReset))
}
}

24
app/src/test/kotlin/com/bitchat/android/service/ShutdownGateTest.kt Normal file
View File

@ -0,0 +1,24 @@
package com.bitchat.android.service
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue
import org.junit.Test
class ShutdownGateTest {
@Test
fun `shutdown is uncommitted initially`() {
val gate = ShutdownGate()
assertFalse(gate.isCommitted())
}
@Test
fun `committed shutdown is irreversible and idempotent`() {
val gate = ShutdownGate()
assertTrue(gate.commit())
assertFalse(gate.commit())
assertTrue(gate.isCommitted())
}
}

36
app/src/test/kotlin/com/bitchat/android/services/AppStateStoreTest.kt
View File

@ -172,4 +172,40 @@ class AppStateStoreTest {
.deliveryStatus
assertTrue(status is DeliveryStatus.Read)
}
@Test
fun `local terminal failure replaces Sending but not admitted or delivered states`() {
val statuses = listOf(
DeliveryStatus.Sending,
DeliveryStatus.Sent,
DeliveryStatus.Delivered("peer-a", Date(2)),
DeliveryStatus.Read("peer-a", Date(3))
)
statuses.forEachIndexed { index, status ->
AppStateStore.addPrivateMessage(
"peer-a",
BitchatMessage(
id = "status-$index",
sender = "bob",
content = "hello",
timestamp = Date(1),
isPrivate = true,
deliveryStatus = status
)
)
AppStateStore.updatePrivateMessageStatus(
"status-$index",
DeliveryStatus.Failed("local terminal failure")
)
}
val resulting = AppStateStore.privateMessages.value
.values
.flatten()
.associate { it.id to it.deliveryStatus }
assertTrue(resulting["status-0"] is DeliveryStatus.Failed)
assertTrue(resulting["status-1"] is DeliveryStatus.Sent)
assertTrue(resulting["status-2"] is DeliveryStatus.Delivered)
assertTrue(resulting["status-3"] is DeliveryStatus.Read)
}
}

312
app/src/test/kotlin/com/bitchat/android/services/MessageRouterTest.kt
View File

@ -5,6 +5,8 @@ import android.os.Build
import com.bitchat.android.identity.SecureIdentityStateManager
import com.bitchat.android.mesh.MeshService
import com.bitchat.android.mesh.PeerInfo
import com.bitchat.android.nostr.NostrSendAdmission
import com.bitchat.android.nostr.NostrTransport
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
@ -35,9 +37,22 @@ class MessageRouterTest {
private val noiseKey = ByteArray(32) { 0x0B }
private lateinit var mesh: MeshService
private lateinit var nostr: NostrTransport
private lateinit var router: MessageRouter
private lateinit var identityManager: SecureIdentityStateManager
private var fakeTime = 1_000_000L
private val expired = mutableListOf<String>()
private val admitted = mutableListOf<String>()
private val failed = mutableListOf<String>()
private val pendingNostrSends = mutableListOf<PendingNostrSend>()
private var nostrAvailable = false
private data class PendingNostrSend(
val content: String,
val peerID: String,
val messageID: String,
val completion: (NostrSendAdmission) -> Unit
)
@Before
fun setup() {
@ -46,10 +61,11 @@ class MessageRouterTest {
"message-router-test-${UUID.randomUUID()}",
Context.MODE_PRIVATE
)
val identityManager = SecureIdentityStateManager(prefs, testOnly = true)
identityManager = SecureIdentityStateManager(prefs, testOnly = true)
ContactDirectory.identityManagerProvider = { identityManager }
mesh = mock()
nostr = mock()
whenever(mesh.myPeerID).thenReturn(myPeerID)
whenever(mesh.getPeerNicknames()).thenReturn(mapOf(peerID to "peer"))
@ -59,10 +75,34 @@ class MessageRouterTest {
MessageRouter.resetForTesting()
fakeTime = 1_000_000L
expired.clear()
admitted.clear()
failed.clear()
pendingNostrSends.clear()
nostrAvailable = false
router = MessageRouter.getInstance(context, mesh)
router = MessageRouter(
context = context,
mesh = mesh,
nostr = nostr,
privateNostrSender = NostrPrivateMessageSender {
content,
target,
_,
messageID,
completion ->
pendingNostrSends += PendingNostrSend(
content = content,
peerID = target,
messageID = messageID,
completion = completion
)
},
canSendViaNostrOverride = { nostrAvailable }
)
router.clock = { fakeTime }
router.onMessageExpired = { expired.add(it) }
router.onMessageAdmitted = { admitted.add(it) }
router.onMessageFailed = { messageID, _ -> failed.add(messageID) }
}
@After
@ -174,6 +214,274 @@ class MessageRouterTest {
verify(mesh, never()).initiateNoiseHandshake(any())
}
@Test
fun `retryable Nostr refusal remains queued until exactly one durable admission`() {
peerOffline()
nostrAvailable = true
assertEquals(
MessageRouter.RouteResult.NOSTR_PENDING,
router.sendPrivate("hello", peerID, "peer", "msg-ndr")
)
assertEquals(listOf("msg-ndr"), pendingNostrSends.map { it.messageID })
assertTrue(admitted.isEmpty())
pendingNostrSends.single().completion(NostrSendAdmission.RETRYABLE)
router.tickOutbox()
assertEquals(listOf("msg-ndr", "msg-ndr"), pendingNostrSends.map { it.messageID })
assertTrue(admitted.isEmpty())
val admittedAttempt = pendingNostrSends.last()
admittedAttempt.completion(NostrSendAdmission.ADMITTED)
admittedAttempt.completion(NostrSendAdmission.ADMITTED)
router.tickOutbox()
assertEquals(listOf("msg-ndr"), admitted)
assertEquals(2, pendingNostrSends.size)
assertTrue(failed.isEmpty())
verify(mesh, never()).sendPrivateMessage(any(), any(), any(), anyOrNull())
}
@Test
fun `one Nostr message per conversation is in flight and queued order is preserved`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("first", peerID, "peer", "msg-1")
router.sendPrivate("second", peerID, "peer", "msg-2")
router.tickOutbox()
assertEquals(listOf("msg-1"), pendingNostrSends.map { it.messageID })
pendingNostrSends[0].completion(NostrSendAdmission.ADMITTED)
assertEquals(listOf("msg-1", "msg-2"), pendingNostrSends.map { it.messageID })
pendingNostrSends[1].completion(NostrSendAdmission.ADMITTED)
assertEquals(listOf("msg-1", "msg-2"), admitted)
}
@Test
fun `mesh and duplicate session callbacks cannot race an in-flight Nostr copy`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("hello", peerID, "peer", "msg-race")
peerReady()
router.onSessionEstablished(peerID)
router.onSessionEstablished(peerID)
assertEquals(1, pendingNostrSends.size)
verify(mesh, never()).sendPrivateMessage(any(), any(), any(), anyOrNull())
pendingNostrSends.single().completion(NostrSendAdmission.RETRYABLE)
router.onSessionEstablished(peerID)
router.onSessionEstablished(peerID)
verify(mesh, times(1)).sendPrivateMessage("hello", peerID, "peer", "msg-race")
assertEquals(1, pendingNostrSends.size)
}
@Test
fun `terminal Nostr failure removes the entry and reports failure once`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("invalid", peerID, "peer", "msg-failed")
val attempt = pendingNostrSends.single()
attempt.completion(NostrSendAdmission.TERMINAL_FAILED)
attempt.completion(NostrSendAdmission.TERMINAL_FAILED)
router.tickOutbox()
assertEquals(listOf("msg-failed"), failed)
assertTrue(admitted.isEmpty())
assertEquals(1, pendingNostrSends.size)
}
@Test
fun `duplicate session established callbacks flush a queued mesh message once`() {
peerOffline()
router.sendPrivate("hello", peerID, "peer", "msg-once")
peerReady()
router.onSessionEstablished(peerID)
router.onSessionEstablished(peerID)
verify(mesh, times(1)).sendPrivateMessage("hello", peerID, "peer", "msg-once")
}
@Test
fun `late result from an earlier Nostr attempt cannot complete its replacement`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("hello", peerID, "peer", "msg-stale")
val first = pendingNostrSends.single()
first.completion(NostrSendAdmission.RETRYABLE)
router.tickOutbox()
val second = pendingNostrSends.last()
first.completion(NostrSendAdmission.ADMITTED)
assertTrue(admitted.isEmpty())
assertEquals(1, router.queuedMessageCount)
assertEquals(1, router.inFlightNostrAttemptCount)
second.completion(NostrSendAdmission.ADMITTED)
assertEquals(listOf("msg-stale"), admitted)
assertEquals(0, router.queuedMessageCount)
assertEquals(0, router.inFlightNostrAttemptCount)
}
@Test
fun `alias convergence migrates in-flight ownership without duplicate Nostr send`() {
whenever(mesh.getPeerInfo(peerID)).thenReturn(null)
nostrAvailable = true
router.sendPrivate("hello", peerID, "peer", "msg-alias")
assertEquals(1, pendingNostrSends.size)
identityManager.cachePeerNoiseKey(
peerID,
ContactIdentityResolver.noiseKeyHex(noiseKey)
)
router.flushOutboxFor(peerID)
router.tickOutbox()
assertEquals(1, pendingNostrSends.size)
pendingNostrSends.single().completion(NostrSendAdmission.ADMITTED)
assertEquals(listOf("msg-alias"), admitted)
assertEquals(0, router.queuedMessageCount)
assertEquals(0, router.inFlightNostrAttemptCount)
}
@Test
fun `synchronous Nostr admission is safe and leaves no in-flight token`() {
peerOffline()
val synchronous = MessageRouter(
context = RuntimeEnvironment.getApplication(),
mesh = mesh,
nostr = nostr,
privateNostrSender = NostrPrivateMessageSender {
_,
_,
_,
_,
completion ->
completion(NostrSendAdmission.ADMITTED)
},
canSendViaNostrOverride = { true }
)
val synchronousAdmissions = mutableListOf<String>()
synchronous.onMessageAdmitted = synchronousAdmissions::add
synchronous.sendPrivate("first", peerID, "peer", "sync-1")
synchronous.sendPrivate("second", peerID, "peer", "sync-2")
assertEquals(listOf("sync-1", "sync-2"), synchronousAdmissions)
assertEquals(0, synchronous.queuedMessageCount)
assertEquals(0, synchronous.inFlightNostrAttemptCount)
}
@Test
fun `TTL and cap never evict an in-flight Nostr message`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("first", peerID, "peer", "msg-0")
repeat(100) { index ->
router.sendPrivate("queued-$index", peerID, "peer", "msg-${index + 1}")
}
assertEquals(listOf("msg-1"), expired)
assertEquals("msg-0", pendingNostrSends.single().messageID)
assertEquals(100, router.queuedMessageCount)
fakeTime += 86_400_001L
router.tickOutbox()
assertFalse("msg-0" in expired)
assertEquals(1, router.queuedMessageCount)
assertEquals(1, router.inFlightNostrAttemptCount)
pendingNostrSends.single().completion(NostrSendAdmission.ADMITTED)
assertEquals(listOf("msg-0"), admitted)
assertEquals(0, router.queuedMessageCount)
}
@Test
fun `account reset discards plaintext and ignores every late callback`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("old identity", peerID, "peer", "msg-old-account")
val oldAttempt = pendingNostrSends.single()
router.discardForAccountReset()
assertEquals(0, router.queuedMessageCount)
assertEquals(0, router.inFlightNostrAttemptCount)
assertEquals(
MessageRouter.RouteResult.DROPPED,
router.sendPrivate(
"must not survive reset",
peerID,
"peer",
"msg-reset-window"
)
)
assertEquals(0, router.queuedMessageCount)
oldAttempt.completion(NostrSendAdmission.ADMITTED)
oldAttempt.completion(NostrSendAdmission.RETRYABLE)
oldAttempt.completion(NostrSendAdmission.TERMINAL_FAILED)
assertTrue(admitted.isEmpty())
assertEquals(listOf("msg-reset-window"), failed)
router.tickOutbox()
assertEquals(1, pendingNostrSends.size)
}
@Test
fun `stale router reset cannot reopen a newer reset`() {
peerOffline()
val firstReset = router.discardForAccountReset()
val secondReset = router.discardForAccountReset()
assertFalse(router.completeAccountReset(firstReset))
assertEquals(
MessageRouter.RouteResult.DROPPED,
router.sendPrivate(
"blocked",
peerID,
"peer",
"msg-stale-reset"
)
)
assertTrue(router.completeAccountReset(secondReset))
assertEquals(
MessageRouter.RouteResult.QUEUED,
router.sendPrivate(
"fresh",
peerID,
"peer",
"msg-current-reset"
)
)
}
@Test
fun `normal scheduler stop preserves queued plaintext and in-flight ownership`() {
peerOffline()
nostrAvailable = true
router.sendPrivate("keep across service stop", peerID, "peer", "msg-pause")
val attempt = pendingNostrSends.single()
router.stopOutboxScheduler()
assertEquals(1, router.queuedMessageCount)
assertEquals(1, router.inFlightNostrAttemptCount)
attempt.completion(NostrSendAdmission.ADMITTED)
assertEquals(listOf("msg-pause"), admitted)
assertEquals(0, router.queuedMessageCount)
assertEquals(0, router.inFlightNostrAttemptCount)
}
@Test
fun `scheduler stops with the mesh service and restarts on rebind`() {
MessageRouter.disableSchedulerForTesting = false

70
app/src/test/kotlin/com/bitchat/android/ui/MessageManagerDeliveryStatusTest.kt Normal file
View File

@ -0,0 +1,70 @@
package com.bitchat.android.ui
import com.bitchat.android.model.BitchatMessage
import com.bitchat.android.model.DeliveryStatus
import com.bitchat.android.services.AppStateStore
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import org.junit.After
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import java.util.Date
class MessageManagerDeliveryStatusTest {
private lateinit var state: ChatState
private lateinit var manager: MessageManager
@Before
fun setUp() {
AppStateStore.clear()
state = ChatState(CoroutineScope(Dispatchers.Unconfined + SupervisorJob()))
manager = MessageManager(state)
}
@After
fun tearDown() {
AppStateStore.clear()
}
@Test
fun `local failure replaces Sending but never downgrades admitted delivery evidence`() {
val statuses = listOf(
DeliveryStatus.Sending,
DeliveryStatus.Sent,
DeliveryStatus.Delivered("peer", Date(2)),
DeliveryStatus.Read("peer", Date(3))
)
state.setPrivateChats(
mapOf(
"peer" to statuses.mapIndexed { index, status ->
BitchatMessage(
id = "message-$index",
sender = "me",
content = "hello",
timestamp = Date(1),
isPrivate = true,
deliveryStatus = status
)
}
)
)
statuses.indices.forEach { index ->
manager.updateMessageDeliveryStatus(
"message-$index",
DeliveryStatus.Failed("local terminal failure")
)
}
val resulting = state.getPrivateChatsValue()
.values
.flatten()
.associate { it.id to it.deliveryStatus }
assertTrue(resulting["message-0"] is DeliveryStatus.Failed)
assertTrue(resulting["message-1"] is DeliveryStatus.Sent)
assertTrue(resulting["message-2"] is DeliveryStatus.Delivered)
assertTrue(resulting["message-3"] is DeliveryStatus.Read)
}
}