Fix notary observer lifecycle and robustness bugs

- Observers now run as children of collectLatest keyed on the derived
  nostr private key: previously every active-wallet emission spawned
  four more eternal collectors on the app scope, and after a wallet
  switch stale collectors kept signing with the old key (duplicate
  signatures, gift wraps and key package bundles).
- Follow the keyManager StateFlow instead of snapshotting .value, so
  the notary still starts when the key loads after the wallet emits.
- Guard per-item processing so one failing row logs instead of killing
  the collector (and the queue) for the rest of the session.
- Derive the real nsecPassword for self-healed key package bundles via
  a new PrivateKey.nsecPassword() extension instead of passing "".
- Fix a copy-pasted log tag in observeUnprocessedMarmotInnerEvents.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

composeApp/src/commonMain/kotlin/fr/acinq/phoenix/managers/WalletManager.kt

@@ -110,12 +110,14 @@ fun LocalKeyManager.nostrPrivateKey(): PrivateKey {
     return derivePrivateKey(path).privateKey
 }
 
-fun LocalKeyManager.nsecPassword(): String {
+fun PrivateKey.nsecPassword(): String {
     return Crypto.hash160(
-        nostrPrivateKey().value
+        value
     ).byteVector().toHex()
 }
 
+fun LocalKeyManager.nsecPassword(): String = nostrPrivateKey().nsecPassword()
+
 fun LocalKeyManager.nostrPublicKey(): HexKey {
     return KeyPair(
         privKey = nostrPrivateKey().value.toByteArray()

composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt

@@ -13,12 +13,17 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
 import com.vitorpamplona.quartz.nip51Lists.encryption.PrivateTagsInContent
 import fr.acinq.phoenix.data.ActiveWallet
 import fr.acinq.phoenix.managers.nostrPrivateKey
+import fr.acinq.phoenix.managers.nsecPassword
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.IO
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import press.mantra.compose.database.model.NostrEvent
 import kotlin.time.Instant
@@ -56,143 +61,159 @@ class NotaryViewModel(
 
     private val logger = Logger.withTag(TAG)
 
-    init {
+    @OptIn(ExperimentalCoroutinesApi::class)
+    private fun observeActiveNostrKey() {
         scope.launch {
-            activeWalletStateFlow.collectLatest { activeWallet ->
-                logger.d("Notary ActiveWallet: $activeWallet")
-                activeWallet?.business?.walletManager?.keyManager?.value?.nostrPrivateKey()?.let { nostrPrivateKey ->
+            activeWalletStateFlow
+                .flatMapLatest { activeWallet ->
+                    activeWallet?.business?.walletManager?.keyManager ?: flowOf(null)
+                }
+                .map { keyManager -> keyManager?.nostrPrivateKey() }
+                .distinctUntilChanged()
+                .collectLatest { nostrPrivateKey ->
+                    if (nostrPrivateKey == null) return@collectLatest
 
                     val keyPair = KeyPair(
                         privKey = nostrPrivateKey.value.toByteArray()
                     )
+                    val nsecPassword = nostrPrivateKey.nsecPassword()
+                    logger.d("Notary observing: ${keyPair.pubKey.toHexKey()}")
 
-                    observeUnsignedNostrEvents(keyPair)
-                    observeUnsealedGiftWrapPayloads(keyPair)
-                    observeUnprocessedMarmotInnerEvents(keyPair)
-                    observeActiveMarmotKeyPackageBundle(keyPair)
+                    // The observers run as children of collectLatest: when the active
+                    // key changes they are cancelled, so they never process events for
+                    // a stale wallet and never run in duplicate.
+                    coroutineScope {
+                        launch { observeUnsignedNostrEvents(keyPair) }
+                        launch { observeUnsealedGiftWrapPayloads(keyPair) }
+                        launch { observeUnprocessedMarmotInnerEvents(keyPair) }
+                        launch { observeActiveMarmotKeyPackageBundle(keyPair, nsecPassword) }
+                    }
                 }
-            }
         }
     }
 
+    init {
+        observeActiveNostrKey()
+    }
 
-    private fun observeUnsignedNostrEvents(
+    // Keeps one failing item from killing the collector for the rest of the session.
+    private inline fun guardNotarization(what: String, block: () -> Unit) {
+        try {
+            block()
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: Throwable) {
+            logger.e("Notarization failed: $what", e)
+        }
+    }
+
+    private suspend fun observeUnsignedNostrEvents(
         keyPair: KeyPair
     ) {
         val tempSigner = NostrSignerSync(
             keyPair
         )
-        scope.launch(Dispatchers.IO) {
-            logger.i { "observeUnsignedNostrEvents: ${keyPair.pubKey.toHexKey()}" }
-            nostrRepository.observeUnsignedNostrEvents(
-                publicKey = keyPair.pubKey.toHexKey()
-            ).distinctUntilChanged().collect { unsignedNostrEventOrNull ->
-                scope.launch(Dispatchers.IO) {
-                    unsignedNostrEventOrNull?.let { unsignedNostrEvent ->
-                        logger.d("Unsigned: ${unsignedNostrEvent.kind}")
-                        val event = tempSigner.signNormal<Event>(
-                            createdAt = unsignedNostrEvent.createdAt.epochSeconds,
-                            kind = unsignedNostrEvent.kind,
-                            tags = unsignedNostrEvent.tags,
-                            content = unsignedNostrEvent.privateTags?.let { PrivateTagsInContent.encryptNip44(it, tempSigner) } ?: unsignedNostrEvent.content
-                        )
-                        logger.d("Signed: ${event.toJson()}")
+        logger.i { "observeUnsignedNostrEvents: ${keyPair.pubKey.toHexKey()}" }
+        nostrRepository.observeUnsignedNostrEvents(
+            publicKey = keyPair.pubKey.toHexKey()
+        ).distinctUntilChanged().collect { unsignedNostrEventOrNull ->
+            unsignedNostrEventOrNull?.let { unsignedNostrEvent ->
+                guardNotarization("unsigned event ${unsignedNostrEvent.id} (kind ${unsignedNostrEvent.kind})") {
+                    logger.d("Unsigned: ${unsignedNostrEvent.kind}")
+                    val event = tempSigner.signNormal<Event>(
+                        createdAt = unsignedNostrEvent.createdAt.epochSeconds,
+                        kind = unsignedNostrEvent.kind,
+                        tags = unsignedNostrEvent.tags,
+                        content = unsignedNostrEvent.privateTags?.let { PrivateTagsInContent.encryptNip44(it, tempSigner) } ?: unsignedNostrEvent.content
+                    )
+                    logger.d("Signed: ${event.toJson()}")
 
-                        nostrRepository.publishNostrEvent(
-                            unsignedNostrEvent,
-                            NostrEvent(
-                                id = event.id,
-                                pubKey = event.pubKey,
-                                kind = event.kind,
-                                tags = event.tags,
-                                content = event.content,
-                                createdAt = Instant.fromEpochSeconds(event.createdAt),
-                                sig = event.sig,
-                                unsignedNostrEventId = unsignedNostrEvent.id
-                            ),
-                            relayURLs = press.mantra.compose.nostr.Relays.DefaultDMRelayList.map { normalizedRelayUrl -> normalizedRelayUrl.url },
-                            activeKeyPair = keyPair
-                        )
-                    }
+                    nostrRepository.publishNostrEvent(
+                        unsignedNostrEvent,
+                        NostrEvent(
+                            id = event.id,
+                            pubKey = event.pubKey,
+                            kind = event.kind,
+                            tags = event.tags,
+                            content = event.content,
+                            createdAt = Instant.fromEpochSeconds(event.createdAt),
+                            sig = event.sig,
+                            unsignedNostrEventId = unsignedNostrEvent.id
+                        ),
+                        relayURLs = press.mantra.compose.nostr.Relays.DefaultDMRelayList.map { normalizedRelayUrl -> normalizedRelayUrl.url },
+                        activeKeyPair = keyPair
+                    )
                 }
             }
         }
     }
 
 
-    private fun observeUnsealedGiftWrapPayloads(
+    private suspend fun observeUnsealedGiftWrapPayloads(
         keyPair: KeyPair
     ) {
         val tempSigner = NostrSignerSync(
             keyPair
         )
-        scope.launch(Dispatchers.IO) {
-            logger.i { "observeUnsealedGiftWrapPayloads: ${keyPair.pubKey.toHexKey()}" }
-            chatRepository.observeUnsealedGiftWrapPayloads(
-                publicKey = keyPair.pubKey.toHexKey()
-            ).distinctUntilChanged().collect { giftWrapPayloadOrNull ->
-                scope.launch(Dispatchers.IO) {
-                    giftWrapPayloadOrNull?.let { giftWrapPayload ->
-                        logger.d("seal and deliver giftWrapPayload: $giftWrapPayload")
+        logger.i { "observeUnsealedGiftWrapPayloads: ${keyPair.pubKey.toHexKey()}" }
+        chatRepository.observeUnsealedGiftWrapPayloads(
+            publicKey = keyPair.pubKey.toHexKey()
+        ).distinctUntilChanged().collect { giftWrapPayloadOrNull ->
+            giftWrapPayloadOrNull?.let { giftWrapPayload ->
+                guardNotarization("gift wrap payload ${giftWrapPayload.id}") {
+                    logger.d("seal and deliver giftWrapPayload: $giftWrapPayload")
 
-                        chatRepository.sealGiftWrapPayload(
-                            giftWrapPayload,
-                            nostrSignerSync = tempSigner
-                        )
-                    }
+                    chatRepository.sealGiftWrapPayload(
+                        giftWrapPayload,
+                        nostrSignerSync = tempSigner
+                    )
                 }
             }
         }
     }
 
-    private fun observeUnprocessedMarmotInnerEvents(
+    private suspend fun observeUnprocessedMarmotInnerEvents(
         keyPair: KeyPair
     ) {
         val tempSigner = NostrSignerSync(
             keyPair
         )
-        scope.launch(Dispatchers.IO) {
-            logger.i { "observeUnsealedGiftWrapPayloads: ${keyPair.pubKey.toHexKey()}" }
-            marmotRepository.observeUnprocessedMarmotInnerEvents(
-                publicKey = keyPair.pubKey.toHexKey()
-            ).distinctUntilChanged().collect { marmotInnerEventOrNull ->
-                scope.launch(Dispatchers.IO) {
-                    marmotInnerEventOrNull?.let { marmotInnerEvent ->
-                        logger.d("encrypt and broadcast: $marmotInnerEvent")
+        logger.i { "observeUnprocessedMarmotInnerEvents: ${keyPair.pubKey.toHexKey()}" }
+        marmotRepository.observeUnprocessedMarmotInnerEvents(
+            publicKey = keyPair.pubKey.toHexKey()
+        ).distinctUntilChanged().collect { marmotInnerEventOrNull ->
+            marmotInnerEventOrNull?.let { marmotInnerEvent ->
+                guardNotarization("marmot inner event ${marmotInnerEvent.id}") {
+                    logger.d("encrypt and broadcast: $marmotInnerEvent")
 
-                        marmotRepository.encryptAndSendMarmotInnerEvent(
-                            marmotInnerEvent,
-                            nostrSignerSync = tempSigner
-                        )
-                    }
+                    marmotRepository.encryptAndSendMarmotInnerEvent(
+                        marmotInnerEvent,
+                        nostrSignerSync = tempSigner
+                    )
                 }
             }
         }
     }
 
 
-    private fun observeActiveMarmotKeyPackageBundle(
-        keyPair: KeyPair
+    private suspend fun observeActiveMarmotKeyPackageBundle(
+        keyPair: KeyPair,
+        nsecPassword: String
     ) {
-        val tempSigner = NostrSignerSync(
-            keyPair
-        )
-        scope.launch(Dispatchers.IO) {
-            logger.i { "observeActiveMarmotKeyPackageBundle: ${keyPair.pubKey.toHexKey()}" }
-            marmotRepository.observeActiveMarmotKeyPackageBundle(
-                publicKey = keyPair.pubKey.toHexKey()
-            ).distinctUntilChanged().collect { marmotKeyPackageBundleOrNull ->
-                scope.launch(Dispatchers.IO) {
-                    if (marmotKeyPackageBundleOrNull == null) {
-                        logger.d("Need to create new key package bundle")
-                        marmotRepository.publishMarmotKeyPackageBundle(
-                            publicKey = keyPair.pubKey.toHexKey(),
-                            nsecPassword = "" // TODO: Implement nsecPassword logic...
-                        )
-                    }
+        logger.i { "observeActiveMarmotKeyPackageBundle: ${keyPair.pubKey.toHexKey()}" }
+        marmotRepository.observeActiveMarmotKeyPackageBundle(
+            publicKey = keyPair.pubKey.toHexKey()
+        ).distinctUntilChanged().collect { marmotKeyPackageBundleOrNull ->
+            if (marmotKeyPackageBundleOrNull == null) {
+                guardNotarization("key package bundle for ${keyPair.pubKey.toHexKey()}") {
+                    logger.d("Need to create new key package bundle")
+                    marmotRepository.publishMarmotKeyPackageBundle(
+                        publicKey = keyPair.pubKey.toHexKey(),
+                        nsecPassword = nsecPassword
+                    )
                 }
             }
         }
     }
 
-}
+}