From 0c65f61ea2cd29a236a600c06be64e01a53a8c7a Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Mon, 27 Jul 2026 20:51:47 +0200 Subject: [PATCH] Fix chat initiation between profiles created in the app MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A profile created in the app could never start a chat with another profile created in the app: the peer always looked like it had no metadata, no DM relay list and no MLS key package. Root cause was in neither the chat code nor the relay list — nothing a new profile signed ever reached a relay. Three queue observers used `createdAt > :createdAt` with a `Clock.System.now()` default argument. Kotlin evaluates that default once, at the call site, and Room binds it for the life of the Flow; Instants persist at second resolution, so every request enqueued in the observer's own start second (the whole profile-creation burst) and everything left pending by a previous session was permanently invisible. Nothing else drains those tables. The failure was silent because `publishNostrEvent` stamps `signedAt` and indexes the Profile in one transaction, satisfying the ProfileLoaded branch before the UnannouncedProfile gate could be reached — so a device-only profile looked fully announced. Dropping the cutoff needs no schema change, so existing installs self-heal on next launch: the stranded rows are still pending. Also fixed, since they gate the same flow once events start moving: - Broadcasts now always reach a terminal status (outer timeout plus try/catch — `.catch` cannot see the suspend call that builds the flow), interrupted ones are requeued once at startup, `OK: false` is a failure rather than a recorded success, fan-out is bounded, and an uncorrelated NOTICE no longer fails whatever publish shares the socket. `take(1)` keeps the publish timeout from firing after a success on a SharedFlow that never completes. - CLOSED is parsed and handled, so a relay refusing a NEG subscription falls back to REQ instead of waiting forever; NOTICE is parsed as the two-element frame it is; negentropy timestamps use seconds, the unit relays use. - Both chat gates observe the peer's key package instead of reading it once and latching a terminal error, and queue the sync they claimed to be doing. Same-minute retries are no longer swallowed by IGNORE. - Group rooms were keyed by the MLS group id instead of the Marmot nostrGroupId (unrelated randoms, so neither side saw the other's events); inviting a member wrote no Participant row, so the Welcome produced no gift wraps, and discarded the post-addMember group state; the invite reported success unconditionally. - An inverted `containsKey` made the "missing peer DM relay list" recovery a no-op, and the wrong RelayTag class wrote "r" tags where NIP-51 relay lists expect "relay". Verified with `:composeApp:compileDebugKotlinAndroid`, including that Room's KSP regenerated the DAO impls without the frozen cutoff. Not yet exercised against live relays. Co-Authored-By: Claude Opus 5 --- .../dao/BroadcastNostrEventRequestDao.kt | 20 +- .../database/dao/MarmotKeyPackageDao.kt | 10 + .../compose/database/dao/MarmotOutboundDao.kt | 51 ++++- .../dao/NegentropySynchronizeRequestDao.kt | 26 ++- .../mantra/compose/database/dao/NostrDao.kt | 53 ++++- .../dao/SynchronizeNostrEventRequestDao.kt | 13 +- .../repository/DatabaseChatRepository.kt | 4 + .../repository/DatabaseNostrRepository.kt | 17 +- .../exceptions/NostrPublishException.kt | 5 +- .../compose/network/relays/RelayPool.kt | 23 +- .../network/sockets/NostrIncomingMessage.kt | 9 + .../sockets/NostrIncomingMessageExt.kt | 13 +- .../sockets/NostrIncomingMessageParser.kt | 14 +- .../compose/network/sockets/NostrVerb.kt | 3 + .../compose/repository/ChatRepository.kt | 6 + .../compose/repository/NostrRepository.kt | 10 + ...ddMemberToChatRoomConfirmationViewModel.kt | 118 ++++++++--- .../view/model/ChatRoomCreationViewModel.kt | 8 +- .../view/model/ChatRoomMessagingViewModel.kt | 151 ++++++++------ .../ui/view/model/NavigationViewModel.kt | 27 ++- .../ui/view/model/SynchronizationViewModel.kt | 196 +++++++++++++++--- 21 files changed, 602 insertions(+), 175 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/BroadcastNostrEventRequestDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/BroadcastNostrEventRequestDao.kt index aa4cfd4f..0f625335 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/BroadcastNostrEventRequestDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/BroadcastNostrEventRequestDao.kt @@ -7,7 +7,6 @@ import androidx.room3.Upsert import press.mantra.compose.database.model.BroadcastNostrEventRequest import press.mantra.compose.database.model.intermdiate.LocalBroadcastNostrEventRequest import kotlinx.coroutines.flow.Flow -import kotlin.time.Clock import kotlin.time.Instant @Dao @@ -18,12 +17,27 @@ interface BroadcastNostrEventRequestDao { @Query("SELECT * FROM BroadcastNostrEventRequest WHERE nostrEventId = :nostrEventId ORDER BY createdAt ASC") fun getFirstBroadcastNostrEventRequestByNostrEventId(nostrEventId: String): BroadcastNostrEventRequest? - @Query("SELECT * FROM BroadcastNostrEventRequest WHERE status = :status AND createdAt > :createdAt") - fun observeBroadcastNostrEventRequestsByStatus(status: String, createdAt: Instant = Clock.System.now()): Flow + // See NegentropySynchronizeRequestDao: the old `createdAt > :createdAt` bound a + // once-evaluated `Clock.System.now()` for the life of the Flow. Because Instants are + // persisted at second resolution, that hid every broadcast enqueued in the observer's + // own start second — i.e. the entire profile-creation burst — plus everything left + // pending by a previous session. Nothing else drains this table, so those events were + // never sent to any relay. + @Query("SELECT * FROM BroadcastNostrEventRequest WHERE status = :status ORDER BY createdAt ASC, id ASC LIMIT 1") + fun observeBroadcastNostrEventRequestsByStatus(status: String): Flow @Query("SELECT * FROM BroadcastNostrEventRequest WHERE nostrEventId = :nostrEventId") fun observeBroadcastNostrEventRequestByNostrEventId(nostrEventId: String): Flow + /** + * A request is flipped to "processing" before the publish is attempted, and the publish can + * end in a timeout or a socket error that leaves it there. Nothing observes "processing" or + * "failed", so those rows are dead weight. Requeue everything a previous process left behind + * when we start up, so an interrupted publish is retried instead of lost. + */ + @Query("UPDATE BroadcastNostrEventRequest SET status = 'pending' WHERE status IN ('processing', 'failed') AND createdAt <= :staleBefore") + suspend fun requeueStaleBroadcastNostrEventRequests(staleBefore: Instant): Int + @Upsert suspend fun upsert(broadcastNostrEventRequest: BroadcastNostrEventRequest): Long diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotKeyPackageDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotKeyPackageDao.kt index 0a5dc252..3bcfbbb7 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotKeyPackageDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotKeyPackageDao.kt @@ -5,12 +5,22 @@ import androidx.room3.Query import androidx.room3.Upsert import press.mantra.compose.database.model.MarmotKeyPackage import com.vitorpamplona.quartz.nip01Core.core.HexKey +import kotlinx.coroutines.flow.Flow @Dao interface MarmotKeyPackageDao { @Query("SELECT * FROM MarmotKeyPackage WHERE publicKey = :publicKey ORDER BY createdAt DESC") suspend fun getMarmotKeyPackageForPublicKey(publicKey: HexKey): MarmotKeyPackage? + /** + * Observable counterpart. The chat gates used to read the key package once, before the + * relay round-trip that fetches it could possibly have finished, and latch a terminal + * error — so a peer whose key package arrived a second later stayed unreachable until + * the screen was rebuilt. + */ + @Query("SELECT * FROM MarmotKeyPackage WHERE publicKey = :publicKey ORDER BY createdAt DESC LIMIT 1") + fun observeMarmotKeyPackageForPublicKey(publicKey: HexKey): Flow + @Query("SELECT * FROM MarmotKeyPackage WHERE id = :id ORDER BY createdAt DESC") suspend fun getMarmotKeyPackageById(id: HexKey): MarmotKeyPackage? diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotOutboundDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotOutboundDao.kt index 4f63c4a8..a4373d24 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotOutboundDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotOutboundDao.kt @@ -17,6 +17,7 @@ import press.mantra.compose.database.model.MarmotKeyPackage import press.mantra.compose.database.model.MarmotRetainedEpochSecret import press.mantra.compose.database.model.NostrEvent import press.mantra.compose.database.model.Participant +import press.mantra.compose.exceptions.MarmotMissingChatGroupException import press.mantra.compose.database.model.intermdiate.LocalChatRoom import press.mantra.compose.extensions.exporterSecret import press.mantra.compose.extensions.toHex @@ -135,17 +136,47 @@ abstract class MarmotOutboundDao( senderPublicKey = localChatRoom.chatRoom.userPublicKey ) - localChatRoom.chatRoom.toMlsGroup(sentGenerations)?.let { mlsGroup -> - inviteMember( - nostrGroupId = localChatRoom.chatRoom.id, - mlsGroup = mlsGroup, - userPublicKey = localChatRoom.chatRoom.userPublicKey, - peerPublicKey = peerPublicKey, - peerKeyPackage = peerKeyPackage, - relays = Relays.DefaultDMRelayList.map { it.url }, - isOneMemberInitialGroupCreation = false + // A room restored from an inbound gift wrap has no persisted MLS state, so there is + // nothing to add a member to. Say so instead of silently doing nothing and letting the + // caller report success. + val mlsGroup = localChatRoom.chatRoom.toMlsGroup(sentGenerations) + ?: throw MarmotMissingChatGroupException( + "No MLS group state for chat room ${localChatRoom.chatRoom.id}; cannot invite $peerPublicKey" ) - } + + val relays = Relays.DefaultDMRelayList.map { it.url } + + // The invitee needs a Participant row before the Welcome is sealed: sealGiftWrapPayload + // walks the participants of the room to decide who to wrap for, so without this the + // Welcome produced no gift wraps at all and sat unsealed forever. + database.participantDao().upsert( + listOf( + Participant( + participantPublicKey = peerPublicKey, + chatRoomId = localChatRoom.chatRoom.id, + relayHint = relays.first() + ) + ) + ) + + inviteMember( + nostrGroupId = localChatRoom.chatRoom.id, + mlsGroup = mlsGroup, + userPublicKey = localChatRoom.chatRoom.userPublicKey, + peerPublicKey = peerPublicKey, + peerKeyPackage = peerKeyPackage, + relays = relays, + isOneMemberInitialGroupCreation = false + ) + + // `addMember` advanced the in-memory group to the next epoch. Without persisting it the + // creator kept encrypting under the old epoch — which the new member cannot decrypt — + // and the next invite would re-derive from stale state and produce a conflicting commit. + database.chatRoomDao().upsert( + localChatRoom.chatRoom.copy( + mlsGroupState = mlsGroup.saveState().encodeTls().toHex() + ) + ) } private suspend fun inviteMember( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NegentropySynchronizeRequestDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NegentropySynchronizeRequestDao.kt index 3859d747..cb0ee4cd 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NegentropySynchronizeRequestDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NegentropySynchronizeRequestDao.kt @@ -1,28 +1,32 @@ package press.mantra.compose.database.dao import androidx.room3.Dao -import androidx.room3.Insert -import androidx.room3.OnConflictStrategy.Companion.IGNORE import androidx.room3.Query import androidx.room3.Upsert import press.mantra.compose.database.model.NegentropySynchronizeRequest import kotlinx.coroutines.flow.Flow -import kotlin.time.Clock -import kotlin.time.Instant @Dao interface NegentropySynchronizeRequestDao { - @Query("SELECT * FROM NegentropySynchronizeRequest WHERE status = :status AND createdAt > :createdAt") - fun observeNegentropySynchronizeRequestsByStatus(status: String, createdAt: Instant = Clock.System.now()): Flow + // No `createdAt >` cut-off: a `Clock.System.now()` default is evaluated once, at + // the call site, and Room binds it for the whole life of the Flow — which silently + // stranded every request queued before (or within the same second as) the observer. + @Query("SELECT * FROM NegentropySynchronizeRequest WHERE status = :status ORDER BY createdAt ASC, id ASC LIMIT 1") + fun observeNegentropySynchronizeRequestsByStatus(status: String): Flow - @Query("SELECT COUNT(*) FROM NegentropySynchronizeRequest WHERE purpose = :purpose AND status IN (:status) AND createdAt > :createdAt") - fun observeNegentropySynchronizeRequestByPurposeAndStatusCount(purpose: String, status: List, createdAt: Instant = Clock.System.now()): Flow + @Query("SELECT COUNT(*) FROM NegentropySynchronizeRequest WHERE purpose = :purpose AND status IN (:status)") + fun observeNegentropySynchronizeRequestByPurposeAndStatusCount(purpose: String, status: List): Flow @Upsert fun upsert(negentropySynchronizeRequest: NegentropySynchronizeRequest) - @Insert( - onConflict = IGNORE - ) + /** + * Upsert rather than insert-or-ignore. `computeId` buckets by minute, so an identical + * request re-queued inside the same minute collides — and with IGNORE the second one was + * silently dropped even when the first had already run and come back empty. Re-arming the + * row (fresh uuid, status back to "pending") is what makes a user-initiated retry actually + * hit the network, while still collapsing duplicates into a single row. + */ + @Upsert fun insert(negentropySynchronizeRequests: List) } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt index b248d4fc..9a5081c3 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt @@ -29,6 +29,7 @@ import press.mantra.compose.exceptions.MarmotWelcomeEventMissingKeyPackageEventI import press.mantra.compose.extensions.toHex import press.mantra.compose.managers.MarmotInboundManager import co.touchlab.kermit.Logger +import kotlinx.coroutines.CancellationException import com.vitorpamplona.quartz.marmot.GroupEventResult import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageEvent import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent @@ -73,7 +74,6 @@ abstract class NostrDao( // RelayFeedsListEvent.KIND ) - @Transaction open suspend fun publishNostrEvent( unsignedNostrEvent: press.mantra.compose.database.model.UnsignedNostrEvent, nostrEvent: NostrEvent, @@ -81,6 +81,39 @@ abstract class NostrDao( activeKeyPair: KeyPair ) { logger.i("Publish Nostr Event: $nostrEvent ($relayURLs)") + + commitPublishedNostrEvent(unsignedNostrEvent, nostrEvent, relayURLs) + + // Indexing is best-effort enrichment and runs in its OWN transaction. It used to share + // the transaction above, so any throw in it rolled back `signedAt` too — and because the + // notary drains one unsigned row at a time, that row would be re-selected forever and + // every later event (including the MLS key package, which is enqueued last) would never + // be signed at all. + try { + indexNostrEvent( + nostrEvent = nostrEvent, + relayURL = relayURLs.first(), + synchronizationRelayURLs = relayURLs, + level = 0, + activeKeyPair = activeKeyPair + ) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.e("Failed to index published event ${nostrEvent.id} (kind ${nostrEvent.kind})", e) + } + } + + /** + * The durable half of publishing: mark the unsigned row signed, store the signed event and + * queue it for every target relay. All-or-nothing, and independent of indexing. + */ + @Transaction + protected open suspend fun commitPublishedNostrEvent( + unsignedNostrEvent: press.mantra.compose.database.model.UnsignedNostrEvent, + nostrEvent: NostrEvent, + relayURLs: List + ) { // Update unsignedEvent with signedTime time... database.unsignedNostrEventDao().upsert( unsignedNostrEvent.copy( @@ -90,14 +123,6 @@ abstract class NostrDao( database.nostrEventDao().upsert(nostrEvent) - indexNostrEvent( - nostrEvent = nostrEvent, - relayURL = relayURLs.first(), - synchronizationRelayURLs = relayURLs, - level = 0, - activeKeyPair = activeKeyPair - ) - relayURLs.forEach { relayURL -> database.broadcastNostrEventRequestDao().upsert( BroadcastNostrEventRequest( @@ -656,8 +681,13 @@ abstract class NostrDao( // Sync ChatMessageRelayListEvent publicKey... // TODO: Get relayHint form participant... + // Inverted test: this created the set only when + // one already existed (wiping it), and left the + // key absent otherwise — so the `?.add` below was + // a no-op and the participant whose DM relay list + // we are missing never got queued for sync. if ( - profilePublicKeysToSync.containsKey( + !profilePublicKeysToSync.containsKey( relayURL ) ) { @@ -872,7 +902,8 @@ abstract class NostrDao( // Sync ChatMessageRelayListEvent publicKey... // TODO: Get relayHint form participant... - if (profilePublicKeysToSync.containsKey(relayURL)) { + // Inverted test — see the identical block above. + if (!profilePublicKeysToSync.containsKey(relayURL)) { profilePublicKeysToSync[relayURL] = mutableSetOf() } profilePublicKeysToSync[relayURL]?.add(participant.participantPublicKey) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/SynchronizeNostrEventRequestDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/SynchronizeNostrEventRequestDao.kt index 197c33f0..36e2e00c 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/SynchronizeNostrEventRequestDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/SynchronizeNostrEventRequestDao.kt @@ -5,16 +5,17 @@ import androidx.room3.Insert import androidx.room3.Query import androidx.room3.Upsert import kotlinx.coroutines.flow.Flow -import kotlin.time.Clock -import kotlin.time.Instant @Dao interface SynchronizeNostrEventRequestDao { - @Query("SELECT * FROM SynchronizeNostrEventRequest WHERE status = :status AND createdAt > :createdAt") - fun observeSynchronizeNostrEventRequestsByStatus(status: String, createdAt: Instant = Clock.System.now()): Flow + // See NegentropySynchronizeRequestDao: the old `createdAt > :createdAt` bound a + // once-evaluated `Clock.System.now()` for the life of the Flow and hid every + // request queued before the observer subscribed. + @Query("SELECT * FROM SynchronizeNostrEventRequest WHERE status = :status ORDER BY createdAt ASC, id ASC LIMIT 1") + fun observeSynchronizeNostrEventRequestsByStatus(status: String): Flow - @Query("SELECT COUNT(*) FROM SynchronizeNostrEventRequest WHERE purpose = :purpose AND status IN (:status) AND createdAt > :createdAt") - fun observeSynchronizeNostrEventRequestByPurposeAndStatusCount(purpose: String, status: List, createdAt: Instant = Clock.System.now()): Flow + @Query("SELECT COUNT(*) FROM SynchronizeNostrEventRequest WHERE purpose = :purpose AND status IN (:status)") + fun observeSynchronizeNostrEventRequestByPurposeAndStatusCount(purpose: String, status: List): Flow @Upsert fun upsert(synchronizeNostrEventRequest: press.mantra.compose.database.model.SynchronizeNostrEventRequest) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt index df2486c7..9ca036b7 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt @@ -116,6 +116,10 @@ class DatabaseChatRepository( return database.marmotKeyPackageDao().getMarmotKeyPackageForPublicKey(publicKey) } + override suspend fun observeMarmotKeyPackageForPublicKey(publicKey: HexKey): Flow { + return database.marmotKeyPackageDao().observeMarmotKeyPackageForPublicKey(publicKey) + } + override suspend fun createMlsDirectMessage( name: String?, description: String?, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt index b810b155..9834facc 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseNostrRepository.kt @@ -40,7 +40,10 @@ import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent import com.vitorpamplona.quartz.nip31Alts.AltTag import com.vitorpamplona.quartz.nip50Search.SearchRelayListEvent import com.vitorpamplona.quartz.nip51Lists.relayLists.RelayFeedsListEvent -import com.vitorpamplona.quartz.nip51Lists.tags.RelayTag +// NIP-51 has two RelayTag classes with identical `assemble` signatures: the generic list one +// uses "r", while the relay-list events (kind 10007 search, kind 10012 feeds) parse "relay". +// Importing the wrong one compiled cleanly and produced lists nothing could read back. +import com.vitorpamplona.quartz.nip51Lists.relayLists.tags.RelayTag import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.delay @@ -384,6 +387,16 @@ class DatabaseNostrRepository( ) } + override suspend fun requeueStaleBroadcastNostrEventRequests(): Int { + val requeued = database.broadcastNostrEventRequestDao().requeueStaleBroadcastNostrEventRequests( + staleBefore = Clock.System.now() + ) + if (requeued > 0) { + logger.i("Requeued $requeued stale broadcast request(s) left over from a previous run") + } + return requeued + } + override suspend fun broadcastProcessed( broadcastNostrEventRequest: BroadcastNostrEventRequest, status: String @@ -544,7 +557,7 @@ class DatabaseNostrRepository( override suspend fun observeSynchronizeNostrEventRequestByPurposeAndStatusCount(purpose: String): Flow { return database.synchronizeNostrEventRequestDao().observeSynchronizeNostrEventRequestByPurposeAndStatusCount( - "feed", + purpose, listOf("pending", "processing") ) } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/exceptions/NostrPublishException.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/exceptions/NostrPublishException.kt index 65941689..a31bd592 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/exceptions/NostrPublishException.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/exceptions/NostrPublishException.kt @@ -1,3 +1,6 @@ package press.mantra.compose.exceptions -class NostrPublishException(override val cause: Throwable?) : RuntimeException() \ No newline at end of file +class NostrPublishException( + override val cause: Throwable? = null, + override val message: String? = null, +) : RuntimeException() \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt index 0ec03267..deae7560 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt @@ -21,6 +21,7 @@ import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.filterIsInstance import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.getAndUpdate import kotlinx.coroutines.flow.timeout @@ -291,9 +292,21 @@ class RelayPool( collectPublishResponse(eventId = nostrEvent.id) } sendEventResult.getOrNull()?.let { + // A relay that answers ["OK", id, false, reason] has REJECTED the event. + // Surfacing it as an error keeps the caller from recording a receipt for + // an event that was never stored, and preserves the relay's reason. + val rejection = if (!it.success) { + logger.w { "relay $socketUrl rejected ${nostrEvent.id}: ${it.message}" } + press.mantra.compose.exceptions.NostrPublishException( + message = it.message ?: "rejected by $socketUrl" + ) + } else { + null + } responseFlow.emit( NostrPublishResult( result = it, + error = rejection, relayUrl = nostrSocketClient.socketUrl ) ) @@ -332,15 +345,7 @@ class RelayPool( private suspend fun press.mantra.compose.network.sockets.NostrSocketClient.collectPublishResponse(eventId: String): press.mantra.compose.network.sockets.NostrIncomingMessage.OkMessage { return incomingMessages .filterByEventId(id = eventId) - .transform { - when (it) { - is press.mantra.compose.network.sockets.NostrIncomingMessage.OkMessage -> emit(it) - is press.mantra.compose.network.sockets.NostrIncomingMessage.NoticeMessage -> throw _root_ide_package_.press.mantra.compose.exceptions.NostrNoticeException( - reason = it.message - ) - else -> error("$it is not allowed") - } - } + .filterIsInstance() .timeout(PUBLISH_TIMEOUT.milliseconds) .first() } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessage.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessage.kt index cc022157..aa582fb1 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessage.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessage.kt @@ -22,6 +22,15 @@ sealed class NostrIncomingMessage { val message: String? = null, ) : NostrIncomingMessage() + /** + * `["CLOSED", , ]` — the relay has ended the subscription on its side + * (auth-required, rate-limited, unsupported filter). Terminal: no EOSE will follow. + */ + data class ClosedMessage( + val subscriptionId: String, + val message: String? = null, + ) : NostrIncomingMessage() + data class AuthMessage( val challenge: String, ) : NostrIncomingMessage() diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt index 784ebbed..0a05d658 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageExt.kt @@ -13,11 +13,20 @@ fun Flow.filterBySubscriptionId(id: String) = (it is NostrIncomingMessage.EoseMessage && it.subscriptionId == id) || (it is NostrIncomingMessage.CountMessage && it.subscriptionId == id) || (it is NostrIncomingMessage.EventsMessage && it.subscriptionId == id) || + // A relay that terminates our subscription (auth-required, rate-limited, + // unsupported filter) sends CLOSED instead of EOSE. Without it the collector + // waits forever on a subscription the relay has already abandoned. + (it is NostrIncomingMessage.ClosedMessage && it.subscriptionId == id) || + // NOTICE carries no subscription id, so it cannot be correlated. It is admitted + // here only because it is the sole signal some relays give for "negentropy + // disabled"; the collector treats it as advisory, never as a terminal failure. (it is NostrIncomingMessage.NoticeMessage) } +// Deliberately does NOT admit NoticeMessage: a NOTICE is not tied to an event id, and +// letting one through failed whatever publish happened to be in flight on the shared +// per-socket flow — including publishes to which the notice had nothing to do. fun Flow.filterByEventId(id: String) = filter { - (it is NostrIncomingMessage.OkMessage && it.eventId == id) || - (it is NostrIncomingMessage.NoticeMessage) + it is NostrIncomingMessage.OkMessage && it.eventId == id } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageParser.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageParser.kt index cf8532e7..38b883b2 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageParser.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrIncomingMessageParser.kt @@ -23,6 +23,7 @@ fun String.parseIncomingMessage(): NostrIncomingMessage? { NostrVerb.Incoming.EOSE -> jsonArray.takeAsEoseIncomingMessage() NostrVerb.Incoming.OK -> jsonArray.takeAsOkIncomingMessage() NostrVerb.Incoming.NOTICE -> jsonArray.takeAsNoticeIncomingMessage() + NostrVerb.Incoming.CLOSED -> jsonArray.takeAsClosedIncomingMessage() NostrVerb.Incoming.AUTH -> jsonArray.takeAsAuthIncomingMessage() NostrVerb.Incoming.COUNT -> jsonArray.takeAsCountIncomingMessage() NostrVerb.Incoming.EVENTS -> jsonArray.takeAsEventsIncomingMessage() @@ -115,10 +116,18 @@ private fun JsonObject.getMessageNostrEventKind(): Kind { return kind ?: -1 } +// NIP-01 NOTICE is the two-element `["NOTICE", ]` — there is no subscription id. +// Reading the text out of element 2 left `message` permanently null and stuffed the human +// readable reason into `subscriptionId`. private fun JsonArray.takeAsNoticeIncomingMessage(): NostrIncomingMessage { - val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val messageText = elementAtOrNull(1)?.jsonPrimitive?.content + return NostrIncomingMessage.NoticeMessage(subscriptionId = null, message = messageText) +} + +private fun JsonArray.takeAsClosedIncomingMessage(): NostrIncomingMessage? { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() ?: return null val messageText = elementAtOrNull(2)?.jsonPrimitive?.content - return NostrIncomingMessage.NoticeMessage(subscriptionId = subscriptionId, message = messageText) + return NostrIncomingMessage.ClosedMessage(subscriptionId = subscriptionId, message = messageText) } private fun JsonArray.takeAsOkIncomingMessage(): NostrIncomingMessage? { @@ -177,6 +186,7 @@ private fun JsonElement.toIncomingMessageType(): NostrVerb.Incoming? { "COUNT" -> NostrVerb.Incoming.COUNT "EVENTS" -> NostrVerb.Incoming.EVENTS "NOTICE" -> NostrVerb.Incoming.NOTICE + "CLOSED" -> NostrVerb.Incoming.CLOSED "NEG-MSG" -> NostrVerb.Incoming.NEGENTROPY_MESSAGE "NEG-ERR" -> NostrVerb.Incoming.NEGENTROPY_ERROR else -> { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrVerb.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrVerb.kt index 3946d9d1..a2c3f670 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrVerb.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrVerb.kt @@ -40,6 +40,9 @@ internal sealed class NostrVerb { @SerialName("NOTICE") NOTICE, + @SerialName("CLOSED") + CLOSED, + @SerialName("OK") OK, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt index aa4a4d05..3f712124 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt @@ -40,6 +40,8 @@ interface ChatRepository { suspend fun getMarmotKeyPackageForPublicKey(publicKey: HexKey): MarmotKeyPackage? + suspend fun observeMarmotKeyPackageForPublicKey(publicKey: HexKey): Flow + suspend fun createMlsDirectMessage( name: String? = null, description: String? = null, @@ -127,6 +129,10 @@ interface ChatRepository { TODO("Not yet implemented") } + override suspend fun observeMarmotKeyPackageForPublicKey(publicKey: HexKey): Flow { + TODO("Not yet implemented") + } + override suspend fun createMlsDirectMessage( name: String?, description: String?, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt index de5e58da..44171146 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/NostrRepository.kt @@ -78,6 +78,12 @@ interface NostrRepository { suspend fun broadcastProcessed(broadcastNostrEventRequest: BroadcastNostrEventRequest, status: String = "processing") + /** + * Move broadcasts a previous process left in "processing"/"failed" back to "pending" so an + * interrupted or rejected publish is retried instead of stranded. Returns the row count. + */ + suspend fun requeueStaleBroadcastNostrEventRequests(): Int + suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) suspend fun negentropySynchronizeRequestProcessed(negentropySynchronizeRequest: NegentropySynchronizeRequest) @@ -240,6 +246,10 @@ interface NostrRepository { TODO("Not yet implemented") } + override suspend fun requeueStaleBroadcastNostrEventRequests(): Int { + TODO("Not yet implemented") + } + override suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/AddMemberToChatRoomConfirmationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/AddMemberToChatRoomConfirmationViewModel.kt index 6109e499..456e364a 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/AddMemberToChatRoomConfirmationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/AddMemberToChatRoomConfirmationViewModel.kt @@ -10,15 +10,22 @@ import androidx.lifecycle.viewModelScope import androidx.lifecycle.viewmodel.initializer import androidx.lifecycle.viewmodel.viewModelFactory import press.mantra.compose.database.model.MarmotKeyPackage +import press.mantra.compose.database.model.NegentropySynchronizeRequest import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.SynchronizationFilter +import press.mantra.compose.nostr.Relays import press.mantra.compose.repository.ChatRepository import press.mantra.compose.repository.NostrRepository import press.mantra.compose.ui.view.state.AddMemberToChatRoomConfirmationUIState import co.touchlab.kermit.Logger +import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageEvent import com.vitorpamplona.quartz.nip01Core.core.HexKey import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.launch +import kotlin.time.Duration.Companion.seconds class AddMemberToChatRoomConfirmationViewModel( val chatRoomId: String, @@ -43,30 +50,73 @@ class AddMemberToChatRoomConfirmationViewModel( viewModelScope.launch(Dispatchers.IO) { val localChatRoom = chatRepository.getChatRoomByIdentifier(chatRoomId) - addMemberToChatRoomConfirmationUIState = if (localChatRoom != null) { - val profile = nostrRepository.getProfileWithPublicKey(profilePublicKey) - - if (profile != null) { - val marmotKeyPackage = chatRepository.getMarmotKeyPackageForPublicKey(profilePublicKey) - - if (marmotKeyPackage != null) { - AddMemberToChatRoomConfirmationUIState.Loaded( - localChatRoom = localChatRoom, - profile = profile, - marmotKeyPackage = marmotKeyPackage - ) - } else { - AddMemberToChatRoomConfirmationUIState.Error("Profile missing key package event. Try again later or tell ${profile.humanReadableNameOrPubkey()} to use Torch.") - } - } else { - AddMemberToChatRoomConfirmationUIState.Error("Failed to find profile...") - } - } else { - AddMemberToChatRoomConfirmationUIState.Error("Couldn't load chat") + if (localChatRoom == null) { + addMemberToChatRoomConfirmationUIState = + AddMemberToChatRoomConfirmationUIState.Error("Couldn't load chat") + return@launch } + + val profile = nostrRepository.getProfileWithPublicKey(profilePublicKey) + if (profile == null) { + addMemberToChatRoomConfirmationUIState = + AddMemberToChatRoomConfirmationUIState.Error("Failed to find profile...") + return@launch + } + + // Don't depend on the bulk sync the previous screen fired asynchronously — the + // contact list is tappable long before that relay round-trip can finish. Ask for + // this one peer's key package ourselves. + scheduleKeyPackageSync() + + // Wait for the relays, but don't spin forever with no explanation. Cancelling this + // job (rather than testing a shared flag) is what keeps the timeout from racing the + // arrival and clobbering a Loaded state. + val lookupTimeout = launch { + delay(RELAY_LOOKUP_TIMEOUT) + addMemberToChatRoomConfirmationUIState = + AddMemberToChatRoomConfirmationUIState.Error("Profile missing key package event. Try again later or tell ${profile.humanReadableNameOrPubkey()} to use Torch.") + } + + // Observe rather than read once: the key package usually lands a moment after + // this screen opens, and the old one-shot read latched a terminal error that no + // amount of waiting could clear. + chatRepository.observeMarmotKeyPackageForPublicKey(profilePublicKey) + .distinctUntilChanged() + .collect { marmotKeyPackage -> + if (marmotKeyPackage != null) { + lookupTimeout.cancel() + addMemberToChatRoomConfirmationUIState = + AddMemberToChatRoomConfirmationUIState.Loaded( + localChatRoom = localChatRoom, + profile = profile, + marmotKeyPackage = marmotKeyPackage + ) + } + } } } + private suspend fun scheduleKeyPackageSync() { + val synchronizationFilter = SynchronizationFilter( + authors = arrayOf(profilePublicKey), + kinds = arrayOf(KeyPackageEvent.KIND) + ) + nostrRepository.queueNegentropySynchronizeRequest( + Relays.DefaultDMRelayList.map { relay -> + NegentropySynchronizeRequest( + id = NegentropySynchronizeRequest.computeId( + relayURL = relay.url, + synchronizationFilter = synchronizationFilter + ), + purpose = "key-packages", + synchronizationFilter = synchronizationFilter, + relayURL = relay.url, + level = 0 + ) + } + ) + } + fun inviteToChatRoom( localChatRoom: LocalChatRoom, peerPublicKey: HexKey, @@ -74,12 +124,25 @@ class AddMemberToChatRoomConfirmationViewModel( onInviteSent: () -> Unit ) { viewModelScope.launch(Dispatchers.IO) { - // Invite member - chatRepository.inviteMember( - localChatRoom = localChatRoom, - peerPublicKey = peerPublicKey, - peerKeyPackage = peerKeyPackage - ) + // Invite member. `inviteMember` can throw (no MLS state for the room, credential + // identity mismatch); reporting success regardless popped the user back to the chat + // as though an invite had been sent when nothing was ever queued. + val outcome = runCatching { + chatRepository.inviteMember( + localChatRoom = localChatRoom, + peerPublicKey = peerPublicKey, + peerKeyPackage = peerKeyPackage + ) + } + + outcome.exceptionOrNull()?.let { e -> + logger.e("Failed to invite $peerPublicKey to ${localChatRoom.chatRoom.id}", e) + isActionPending.value = false + addMemberToChatRoomConfirmationUIState = + AddMemberToChatRoomConfirmationUIState.Error("Couldn't send the invite. Please try again.") + return@launch + } + viewModelScope.launch(Dispatchers.Main) { onInviteSent.invoke() } @@ -89,6 +152,9 @@ class AddMemberToChatRoomConfirmationViewModel( companion object { private const val TAG = "AddMemberToChatRoomConfirmationViewModel" + /** How long to wait on the relays before telling the user we came up empty. */ + private val RELAY_LOOKUP_TIMEOUT = 20.seconds + fun factory( activeUserPublicKey: HexKey, profilePublicKey: HexKey, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomCreationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomCreationViewModel.kt index 8f9a5157..62164ea4 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomCreationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomCreationViewModel.kt @@ -80,7 +80,13 @@ class ChatRoomCreationViewModel( viewModelScope.launch(Dispatchers.IO) { val localChatRoom = chatRepository.getOrCreateChatRoom( - chatRoomId = group.groupId.toHex(), + // Key the room on the Marmot `nostr_group_id` baked into the group's + // 0xF2EE extension, NOT on `MlsGroup`'s own randomly generated groupId — + // they are unrelated 32-byte values. GroupEvents are h-tagged with the + // room id, while every inbound path (and the invitee's own join) resolves + // rooms by nostrGroupId, so using the MLS id meant the two sides could + // never see each other's events. + chatRoomId = gid, activeUserPublicKey = keyPair.pubKey.toHexKey(), relayHint = null, defaultSubject = name.toString(), diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomMessagingViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomMessagingViewModel.kt index f865b6c0..22682987 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomMessagingViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomMessagingViewModel.kt @@ -25,7 +25,11 @@ import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent import com.vitorpamplona.quartz.nip17Dm.settings.ChatMessageRelayListEvent import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.launch +import kotlin.time.Duration.Companion.seconds class ChatRoomMessagingViewModel( val chatRoomId: String, @@ -109,78 +113,109 @@ class ChatRoomMessagingViewModel( } } - nostrRepository.observeProfileWithPublicKey(chatRoomId).collect { profile -> - if (profile != null) { - logger.d("Create chat with: $profile") + // Observe BOTH halves of what we need. Watching only the profile meant a key + // package arriving later never re-triggered the check (indexing a kind-30443 + // touches MarmotKeyPackage, not Profile), so a lost race latched permanently. + var creationStarted = false - if (profile.createdAt > GENESIS_AT) { - // Check if we have a keyPackageEvent... - val keyPackageEvent = chatRepository.getMarmotKeyPackageForPublicKey(chatRoomId) // TODO: Observe + // Keep waiting (the collect below stays live, so a late arrival still works) but + // stop looking like nothing is happening if the peer's key package never shows up. + val lookupTimeout = viewModelScope.launch { + delay(RELAY_LOOKUP_TIMEOUT) + chatRoomMessagingUIState = ChatRoomMessagingUIState.Error( + "Couldn't find this profile's chat details on the relays yet. They may not have finished setting up Torch." + ) + } - if (keyPackageEvent == null) { - chatRoomMessagingUIState = ChatRoomMessagingUIState.Error("Missing Key Package") // TODO: Introduce missing key package UI State... - } else { - // Create new chat and invite local - try { - val chatRoomId = chatRepository.createMlsDirectMessage( - userPublicKey = activeUserPublicKey, - peerPublicKey = chatRoomId, - peerKeyPackage = keyPackageEvent, - ) + combine( + nostrRepository.observeProfileWithPublicKey(chatRoomId), + chatRepository.observeMarmotKeyPackageForPublicKey(chatRoomId) + ) { profile, keyPackage -> profile to keyPackage } + .distinctUntilChanged() + .collect { (profile, keyPackageEvent) -> + val profileIsResolved = profile != null && profile.createdAt > GENESIS_AT - viewModelScope.launch(Dispatchers.Main) { - onNavigateToChat.invoke( - ChatRoomMessagingRoute( - activeUserPublicKey = activeUserPublicKey, - chatRoomId = chatRoomId, - relayHint = relayHint - ) - ) - } - } catch (e: Throwable) { - logger.e("Failed to create chat", e) - chatRoomMessagingUIState = ChatRoomMessagingUIState.Error("Failed to create chat") - } - } - } else { - chatRoomMessagingUIState = ChatRoomMessagingUIState.Error("Couldn't find profile... searching relays.") // Looking for profile... + if (!profileIsResolved || keyPackageEvent == null) { + // Whatever is missing, ask the relays for it. This used to be queued + // only when the Profile row was entirely absent, so the far more common + // "we have a placeholder / we have the profile but not the key package" + // cases sat there showing an error while nothing was being fetched. + scheduleProfileAndKeyPackageSync() + return@collect } - } else { - val synchronizationFilter = SynchronizationFilter( - kinds = arrayOf( - MetadataEvent.KIND, - KeyPackageEvent.KIND, - ChatMessageRelayListEvent.KIND - ), - authors = arrayOf( - chatRoomId - ), - limit = 1 - ) - nostrRepository.queueNegentropySynchronizeRequest( - Relays.DefaultDMRelayList.map { relay -> - NegentropySynchronizeRequest( - id = NegentropySynchronizeRequest.computeId( - relay.url, - synchronizationFilter = synchronizationFilter - ), - purpose = "initiate-chat", - synchronizationFilter = synchronizationFilter, - relayURL = relay.url, - level = 0 + + lookupTimeout.cancel() + if (creationStarted) return@collect + creationStarted = true + + logger.d("Create chat with: $profile") + try { + val newChatRoomId = chatRepository.createMlsDirectMessage( + userPublicKey = activeUserPublicKey, + peerPublicKey = chatRoomId, + peerKeyPackage = keyPackageEvent, + ) + + viewModelScope.launch(Dispatchers.Main) { + onNavigateToChat.invoke( + ChatRoomMessagingRoute( + activeUserPublicKey = activeUserPublicKey, + chatRoomId = newChatRoomId, + relayHint = relayHint + ) ) } - ) - chatRoomMessagingUIState = ChatRoomMessagingUIState.Error("Couldn't find profile... queued to search for relay.") // Looking for profile... + } catch (e: Throwable) { + creationStarted = false + logger.e("Failed to create chat", e) + chatRoomMessagingUIState = ChatRoomMessagingUIState.Error("Failed to create chat") + } + return@collect } - } } } + /** + * Ask the DM relays for everything needed to start a chat with [chatRoomId] (a pubkey here): + * the peer's metadata, their MLS key package and their DM relay list. + */ + private suspend fun scheduleProfileAndKeyPackageSync() { + val synchronizationFilter = SynchronizationFilter( + kinds = arrayOf( + MetadataEvent.KIND, + KeyPackageEvent.KIND, + ChatMessageRelayListEvent.KIND + ), + authors = arrayOf( + chatRoomId + ) + // No `limit`: it is ignored on the negentropy path but honoured on the REQ + // fallback, where `limit = 1` returned a single newest event across all three + // kinds — almost always the kind-0 — so the key package never arrived. + ) + nostrRepository.queueNegentropySynchronizeRequest( + Relays.DefaultDMRelayList.map { relay -> + NegentropySynchronizeRequest( + id = NegentropySynchronizeRequest.computeId( + relay.url, + synchronizationFilter = synchronizationFilter + ), + purpose = "initiate-chat", + synchronizationFilter = synchronizationFilter, + relayURL = relay.url, + level = 0 + ) + } + ) + } + companion object { private const val TAG = "ChatRoomMessagingViewModel" + /** How long to wait on the relays before telling the user we came up empty. */ + private val RELAY_LOOKUP_TIMEOUT = 20.seconds + fun factory( activeUserPublicKey: HexKey, chatRoomId: String, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NavigationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NavigationViewModel.kt index 454feb3d..86f3c053 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NavigationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NavigationViewModel.kt @@ -33,6 +33,9 @@ class NavigationViewModel( companion object { private const val TAG = "NavigationViewModel" + /** Broadcast statuses that mean "still on its way to a relay". */ + private val PENDING_BROADCAST_STATUSES = setOf("pending", "processing") + fun factory( activeWallet: StateFlow, initialNavigationUIState: NavigationUIState, @@ -105,16 +108,26 @@ class NavigationViewModel( NavigationUIState.ProfileLoaded( publicKey = localAccount.unsignedNostrEvent.pubKey ) + } else if (localAccount.broadcastNostrEventRequest != null && + localAccount.broadcastNostrEventRequest.status in PENDING_BROADCAST_STATUSES + ) { + // Must be checked BEFORE the signed-and-indexed branch below. `publishNostrEvent` + // stamps `signedAt` and upserts the local Profile in the same transaction, so + // that branch went true the instant the kind-0 was signed and this gate — the one + // that exists to hold the user until the profile is actually on a relay — was + // unreachable. A profile that lived only on the device looked fully announced. + // + // Deliberately scoped to in-flight statuses: once a broadcast has been marked + // "failed" we fall through rather than park the user on a screen with no way out. + logger.i("We have a broadcast request: ${localAccount.broadcastNostrEventRequest}") + NavigationUIState.UnannouncedProfile( + broadcastNostrEventRequest = localAccount.broadcastNostrEventRequest + ) } else if (localAccount.unsignedNostrEvent.signedAt != null && localAccount.profile != null) { logger.i("We have successfully synced a profile: ${localAccount.profile}") NavigationUIState.ProfileLoaded( publicKey = localAccount.unsignedNostrEvent.pubKey ) - } else if (localAccount.broadcastNostrEventRequest != null) { - logger.i("We have a broadcast request: ${localAccount.broadcastNostrEventRequest}") - NavigationUIState.UnannouncedProfile( - broadcastNostrEventRequest = localAccount.broadcastNostrEventRequest - ) } else if (localAccount.profile != null) { logger.i("We have a profile that needs to queued for broadcast: ${localAccount.profile}") NavigationUIState.UnqueuedProfile( @@ -130,7 +143,9 @@ class NavigationViewModel( NavigationUIState.UnqueuedProfileSynchronization( unsignedNostrEvent = localAccount.unsignedNostrEvent ) - } else if (localAccount.unsignedNostrEvent.signedAt != null && localAccount.synchronizeNostrEventRequests.isEmpty()) { + // `isNotEmpty`: this branch is the "a sync is already queued" case. It used to repeat + // the predicate above verbatim, which made it dead code. + } else if (localAccount.unsignedNostrEvent.signedAt != null && localAccount.synchronizeNostrEventRequests.isNotEmpty()) { logger.i("We should be syncing the profile: ${localAccount.unsignedNostrEvent}") NavigationUIState.UnsyncedProfile( unsignedNostrEvent = localAccount.unsignedNostrEvent diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt index 74504d20..b55b444f 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt @@ -27,18 +27,27 @@ import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import fr.acinq.phoenix.data.ActiveWallet import fr.acinq.phoenix.managers.nostrPrivateKey +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.FlowPreview import kotlinx.coroutines.IO +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.isActive +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.withTimeout import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.timeout import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.sync.withPermit import kotlin.time.Duration.Companion.milliseconds class SynchronizationViewModel( @@ -59,6 +68,14 @@ class SynchronizationViewModel( companion object { private const val TAG = "SynchronizationViewModel" + private const val MAX_CONCURRENT_PUBLISHES = 8 + + /** + * Hard ceiling on a single publish attempt, covering the suspend calls that open the + * socket as well as the response flow, so a request can never sit in "processing". + */ + private val PUBLISH_ATTEMPT_TIMEOUT = (RelayPool.PUBLISH_TIMEOUT * 2).milliseconds + private val mutex = Mutex() fun factory( @@ -82,6 +99,26 @@ class SynchronizationViewModel( private val logger = Logger.withTag(TAG) + /** Guards the one-shot startup requeue against a second active-wallet emission. */ + private var hasRequeuedStaleBroadcasts = false + + /** Caps how many publishes may be in flight while the queue drains a backlog. */ + private val publishSlots = Semaphore(MAX_CONCURRENT_PUBLISHES) + + /** + * Keeps one bad request from killing the pump that is draining the queue. Mirrors + * `NotaryViewModel.guardNotarization`. + */ + private inline fun guardPump(what: String, block: () -> Unit) { + try { + block() + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.e("Synchronization step failed: $what", e) + } + } + init { scope.launch { activeWalletStateFlow.collectLatest { activeWallet -> @@ -91,23 +128,43 @@ class SynchronizationViewModel( privKey = nostrPrivateKey.value.toByteArray() ) - observePendingBroadcastNostrEventRequests(keyPair) - observePendingSyncNostrEventRequests(keyPair) - observePendingNegentropySynchronizeRequests(keyPair) + // Anything a previous run left mid-flight is ours to retry: it was flipped to + // "processing" before the publish and nothing observes that status, so + // without this it would sit in the table forever. Once per process only — + // collectLatest can fire again on a wallet switch, and re-running it then + // would resurrect rows whose publish is still in flight. + if (!hasRequeuedStaleBroadcasts) { + hasRequeuedStaleBroadcasts = true + runCatching { nostrRepository.requeueStaleBroadcastNostrEventRequests() } + .onFailure { logger.e("Failed to requeue stale broadcasts", it) } + } + + // Run the three pumps as children of collectLatest so a wallet switch + // cancels them. Launching on `scope` (as this used to) escaped that + // cancellation and accumulated a duplicate set of collectors — each + // racing the same single-row queues with a stale key pair. + // supervisorScope, not coroutineScope: the three pumps are independent, and a + // throw in one must not take the other two down with it for the rest of the + // session. Each pump also guards its own per-request body (see guardPump). + supervisorScope { + launch(Dispatchers.IO) { observePendingBroadcastNostrEventRequests(keyPair) } + launch(Dispatchers.IO) { observePendingSyncNostrEventRequests(keyPair) } + launch(Dispatchers.IO) { observePendingNegentropySynchronizeRequests(keyPair) } + } } } } } - private fun observePendingSyncNostrEventRequests( + private suspend fun observePendingSyncNostrEventRequests( keyPair: KeyPair - ) { + ): Unit = coroutineScope { logger.i { "observePendingSyncNostrEventRequests" } - scope.launch(Dispatchers.IO) { - nostrRepository.observePendingSynchronizeNostrEventRequests().distinctUntilChanged().collect { synchronizeNostrEventRequestOrNull -> - synchronizeNostrEventRequestOrNull?.let { synchronizeNostrEventRequest -> + nostrRepository.observePendingSynchronizeNostrEventRequests().distinctUntilChanged().collect { synchronizeNostrEventRequestOrNull -> + synchronizeNostrEventRequestOrNull?.let { synchronizeNostrEventRequest -> + guardPump("sync request ${synchronizeNostrEventRequest.id}") { logger.i("synchronizeNostrEventRequest: $synchronizeNostrEventRequest") val reqCommand = ReqCmd( subId = synchronizeNostrEventRequest.id, @@ -128,7 +185,7 @@ class SynchronizationViewModel( nostrRepository.synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest) - scope.launch(Dispatchers.IO) { + launch(Dispatchers.IO) { try { relaysSocketManager.query( reqCommand, @@ -136,7 +193,7 @@ class SynchronizationViewModel( ).collect { nostrIncomingMessage -> when (nostrIncomingMessage) { is NostrIncomingMessage.EventMessage -> { - scope.launch(Dispatchers.IO) { + launch(Dispatchers.IO) { logger.d("Import message: $nostrIncomingMessage") nostrIncomingMessage.nostrEvent?.let { nostrRepository.saveNostrEvent( @@ -173,6 +230,15 @@ class SynchronizationViewModel( synchronizeNostrEventRequest.relayURL ) } + is NostrIncomingMessage.ClosedMessage -> { + // The relay ended the subscription on its side (auth + // required, rate limit, unsupported filter): no EOSE + // will follow. Previously CLOSED was not even parsed, so + // the reason was invisible. (`return@collect` only ends + // handling of this message — same as the EOSE branch.) + logger.w("Relay closed sync subscription (${synchronizeNostrEventRequest.relayURL}): ${nostrIncomingMessage.message}") + return@collect + } else -> { logger.d("Unhandled message ${synchronizeNostrEventRequest.relayURL}: $nostrIncomingMessage") } @@ -188,14 +254,14 @@ class SynchronizationViewModel( } } - private fun observePendingNegentropySynchronizeRequests( + private suspend fun observePendingNegentropySynchronizeRequests( keyPair: KeyPair - ) { + ): Unit = coroutineScope { logger.i { "observePendingNegentropySynchronizeRequests" } - scope.launch(Dispatchers.IO) { - nostrRepository.observePendingNegentropySynchronizeRequests().distinctUntilChanged().collect { negentropySynchronizeRequestOrNull -> + nostrRepository.observePendingNegentropySynchronizeRequests().distinctUntilChanged().collect { negentropySynchronizeRequestOrNull -> negentropySynchronizeRequestOrNull?.let { negentropySynchronizeRequest -> + guardPump("negentropy request ${negentropySynchronizeRequest.id}") { mutex.withLock { logger.i("negentropySynchronizeRequest: $negentropySynchronizeRequest") @@ -208,8 +274,13 @@ class SynchronizationViewModel( logger.d("Events: ${events.size}") val storage = StorageVector().apply { events.forEach { event -> + // Nostr timestamps — and therefore every timestamp a relay + // puts in its own negentropy vector — are in SECONDS. Feeding + // milliseconds here sorted every local item ~1000x past every + // remote one, so the fingerprint bounds could never match and + // reconciliation degenerated into a full ID transfer. insert( - timestamp = event.createdAt.toEpochMilliseconds(), + timestamp = event.createdAt.epochSeconds, idHex = event.id ) } @@ -239,7 +310,7 @@ class SynchronizationViewModel( nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest) - scope.launch(Dispatchers.IO) { + launch(Dispatchers.IO) { try { val negCloseCmd = NegCloseCmd( subId = negentropySynchronizeRequest.uuid, @@ -251,7 +322,7 @@ class SynchronizationViewModel( ).collect { nostrIncomingMessage -> when (nostrIncomingMessage) { is NostrIncomingMessage.EventMessage -> { - scope.launch(Dispatchers.IO) { + launch(Dispatchers.IO) { logger.d("Import message: $nostrIncomingMessage") nostrIncomingMessage.nostrEvent?.let { nostrRepository.saveNostrEvent( @@ -351,9 +422,33 @@ class SynchronizationViewModel( return@collect } + is NostrIncomingMessage.ClosedMessage -> { + // Relay refused or ended the NEG subscription (no + // NIP-77 support, auth required, rate limit). No + // NEG-MSG will follow, so fall back to a plain REQ + // rather than leaving the request silently unserved. + // (`return@collect` ends handling of this message, + // matching the EOSE and NEG-ERR branches.) + logger.w("Relay closed negentropy subscription (${negentropySynchronizeRequest.relayURL}): ${nostrIncomingMessage.message}") + if (negentropySynchronizeRequest.purpose != "mlsMessages") { + nostrRepository.queueSynchronizeNostrEvent( + listOf( + negentropySynchronizeRequest.toSynchronizeNostrEventRequest() + ) + ) + } + return@collect + } is NostrIncomingMessage.NoticeMessage -> { logger.d("Notice message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage") - if (nostrIncomingMessage.subscriptionId?.contains("negentropy disabled") == true) { + // A NOTICE carries its text in `message`; it used to + // be read out of `subscriptionId`, which NIP-01 + // NOTICEs do not have, so this fallback never fired. + // Kept narrow on purpose: a NOTICE has no + // subscription id, so it is delivered to EVERY live + // collector on this socket — a loose match would fan + // one notice out into a REQ fallback per collector. + if (nostrIncomingMessage.message?.contains("negentropy disabled", ignoreCase = true) == true) { if (negentropySynchronizeRequest.purpose != "mlsMessages") { nostrRepository.queueSynchronizeNostrEvent( listOf( @@ -383,16 +478,16 @@ class SynchronizationViewModel( } @OptIn(FlowPreview::class) - private fun observePendingBroadcastNostrEventRequests( + private suspend fun observePendingBroadcastNostrEventRequests( keyPair: KeyPair - ) { + ): Unit = coroutineScope { logger.i { "observePendingBroadcastNostrEventRequests" } - scope.launch(Dispatchers.IO) { - nostrRepository.observePendingBroadcastNostrEventRequests().distinctUntilChanged().collect { localBroadcastNostrEventRequest -> + nostrRepository.observePendingBroadcastNostrEventRequests().distinctUntilChanged().collect { localBroadcastNostrEventRequest -> logger.d("localBroadcastNostrEventRequest: $localBroadcastNostrEventRequest") if (localBroadcastNostrEventRequest != null) { + guardPump("broadcast request ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.id}") { nostrRepository.broadcastProcessed( localBroadcastNostrEventRequest.broadcastNostrEventRequest ) @@ -421,16 +516,42 @@ class SynchronizationViewModel( // } // } - scope.launch(Dispatchers.IO) { + // The queue advances as soon as the row leaves "pending", so a backlog would + // otherwise fan out one publish coroutine (and one 30s timer) per row as fast + // as SQLite can commit. Bound it. + launch(Dispatchers.IO) { + publishSlots.withPermit { + // Every publish MUST reach a terminal status. The row was flipped to + // "processing" before we got here, and the navigation gate parks the user + // on "announcing your profile" while a broadcast is pending/processing — + // so a publish that neither completes nor throws would strand them there. + // `.catch` only sees errors from the flow, not from the suspend call that + // builds it, hence the outer try/catch and the outer timeout. + try { + withTimeout(PUBLISH_ATTEMPT_TIMEOUT) { // Broadcast to the intended relay... relaysSocketManager.publishEvent( localBroadcastNostrEventRequest.nostrEvent, setOf( localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL.toRelayDTO() ) - ).timeout(RelayPool.PUBLISH_TIMEOUT.milliseconds).catch { - // Timeout... - }.collect { nostrPublishResult -> + ).timeout(RelayPool.PUBLISH_TIMEOUT.milliseconds) + // We publish to exactly ONE relay per request, so exactly one result + // is expected. `take(1)` completes the collection on that result — + // without it the underlying MutableSharedFlow never completes, the + // timeout keeps running, and 30s after a SUCCESSFUL publish the catch + // below would overwrite "published" with "failed". + .take(1) + .catch { e -> + // The request was flipped to "processing" before we got here, so + // an empty catch left it stranded in that status with nothing to + // observe it. Record the failure; the next launch requeues it. + logger.e("Publish timed out/failed for ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.nostrEventId} @ ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL}", e) + nostrRepository.broadcastProcessed( + localBroadcastNostrEventRequest.broadcastNostrEventRequest, + "failed" + ) + }.collect { nostrPublishResult -> if (nostrPublishResult.error != null) { logger.e("Error publishing note: $nostrPublishResult") nostrRepository.broadcastProcessed( @@ -445,9 +566,30 @@ class SynchronizationViewModel( ) } } + } + } catch (e: CancellationException) { + // A withTimeout expiry is a CancellationException too, but only the + // *outer* coroutine being cancelled should propagate. + if (currentCoroutineContext().isActive) { + logger.e("Publish did not complete for ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.nostrEventId} @ ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL}", e) + nostrRepository.broadcastProcessed( + localBroadcastNostrEventRequest.broadcastNostrEventRequest, + "failed" + ) + } else { + throw e + } + } catch (e: Throwable) { + logger.e("Publish failed for ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.nostrEventId} @ ${localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL}", e) + nostrRepository.broadcastProcessed( + localBroadcastNostrEventRequest.broadcastNostrEventRequest, + "failed" + ) + } + } } + } } } - } } } \ No newline at end of file