Fix chat initiation between profiles created in the app
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 <noreply@anthropic.com>
This commit is contained in:
Kgothatso Ngako
@@ -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<LocalBroadcastNostrEventRequest?>
|
||||
// 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<LocalBroadcastNostrEventRequest?>
|
||||
|
||||
@Query("SELECT * FROM BroadcastNostrEventRequest WHERE nostrEventId = :nostrEventId")
|
||||
fun observeBroadcastNostrEventRequestByNostrEventId(nostrEventId: String): Flow<BroadcastNostrEventRequest?>
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
||||
|
||||
@@ -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<MarmotKeyPackage?>
|
||||
|
||||
@Query("SELECT * FROM MarmotKeyPackage WHERE id = :id ORDER BY createdAt DESC")
|
||||
suspend fun getMarmotKeyPackageById(id: HexKey): MarmotKeyPackage?
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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<NegentropySynchronizeRequest?>
|
||||
// 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<NegentropySynchronizeRequest?>
|
||||
|
||||
@Query("SELECT COUNT(*) FROM NegentropySynchronizeRequest WHERE purpose = :purpose AND status IN (:status) AND createdAt > :createdAt")
|
||||
fun observeNegentropySynchronizeRequestByPurposeAndStatusCount(purpose: String, status: List<String>, createdAt: Instant = Clock.System.now()): Flow<Int>
|
||||
@Query("SELECT COUNT(*) FROM NegentropySynchronizeRequest WHERE purpose = :purpose AND status IN (:status)")
|
||||
fun observeNegentropySynchronizeRequestByPurposeAndStatusCount(purpose: String, status: List<String>): Flow<Int>
|
||||
|
||||
@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<NegentropySynchronizeRequest>)
|
||||
}
|
||||
@@ -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<String>
|
||||
) {
|
||||
// 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)
|
||||
|
||||
@@ -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<press.mantra.compose.database.model.SynchronizeNostrEventRequest?>
|
||||
// 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<press.mantra.compose.database.model.SynchronizeNostrEventRequest?>
|
||||
|
||||
@Query("SELECT COUNT(*) FROM SynchronizeNostrEventRequest WHERE purpose = :purpose AND status IN (:status) AND createdAt > :createdAt")
|
||||
fun observeSynchronizeNostrEventRequestByPurposeAndStatusCount(purpose: String, status: List<String>, createdAt: Instant = Clock.System.now()): Flow<Int>
|
||||
@Query("SELECT COUNT(*) FROM SynchronizeNostrEventRequest WHERE purpose = :purpose AND status IN (:status)")
|
||||
fun observeSynchronizeNostrEventRequestByPurposeAndStatusCount(purpose: String, status: List<String>): Flow<Int>
|
||||
|
||||
@Upsert
|
||||
fun upsert(synchronizeNostrEventRequest: press.mantra.compose.database.model.SynchronizeNostrEventRequest)
|
||||
|
||||
@@ -116,6 +116,10 @@ class DatabaseChatRepository(
|
||||
return database.marmotKeyPackageDao().getMarmotKeyPackageForPublicKey(publicKey)
|
||||
}
|
||||
|
||||
override suspend fun observeMarmotKeyPackageForPublicKey(publicKey: HexKey): Flow<MarmotKeyPackage?> {
|
||||
return database.marmotKeyPackageDao().observeMarmotKeyPackageForPublicKey(publicKey)
|
||||
}
|
||||
|
||||
override suspend fun createMlsDirectMessage(
|
||||
name: String?,
|
||||
description: String?,
|
||||
|
||||
@@ -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<Int> {
|
||||
return database.synchronizeNostrEventRequestDao().observeSynchronizeNostrEventRequestByPurposeAndStatusCount(
|
||||
"feed",
|
||||
purpose,
|
||||
listOf("pending", "processing")
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
package press.mantra.compose.exceptions
|
||||
|
||||
class NostrPublishException(override val cause: Throwable?) : RuntimeException()
|
||||
class NostrPublishException(
|
||||
override val cause: Throwable? = null,
|
||||
override val message: String? = null,
|
||||
) : RuntimeException()
|
||||
@@ -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<press.mantra.compose.network.sockets.NostrIncomingMessage.OkMessage>()
|
||||
.timeout(PUBLISH_TIMEOUT.milliseconds)
|
||||
.first()
|
||||
}
|
||||
|
||||
@@ -22,6 +22,15 @@ sealed class NostrIncomingMessage {
|
||||
val message: String? = null,
|
||||
) : NostrIncomingMessage()
|
||||
|
||||
/**
|
||||
* `["CLOSED", <subId>, <reason>]` — 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()
|
||||
|
||||
@@ -13,11 +13,20 @@ fun Flow<NostrIncomingMessage>.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<NostrIncomingMessage>.filterByEventId(id: String) =
|
||||
filter {
|
||||
(it is NostrIncomingMessage.OkMessage && it.eventId == id) ||
|
||||
(it is NostrIncomingMessage.NoticeMessage)
|
||||
it is NostrIncomingMessage.OkMessage && it.eventId == id
|
||||
}
|
||||
|
||||
@@ -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", <message>]` — 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 -> {
|
||||
|
||||
@@ -40,6 +40,9 @@ internal sealed class NostrVerb {
|
||||
@SerialName("NOTICE")
|
||||
NOTICE,
|
||||
|
||||
@SerialName("CLOSED")
|
||||
CLOSED,
|
||||
|
||||
@SerialName("OK")
|
||||
OK,
|
||||
|
||||
|
||||
@@ -40,6 +40,8 @@ interface ChatRepository {
|
||||
|
||||
suspend fun getMarmotKeyPackageForPublicKey(publicKey: HexKey): MarmotKeyPackage?
|
||||
|
||||
suspend fun observeMarmotKeyPackageForPublicKey(publicKey: HexKey): Flow<MarmotKeyPackage?>
|
||||
|
||||
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<MarmotKeyPackage?> {
|
||||
TODO("Not yet implemented")
|
||||
}
|
||||
|
||||
override suspend fun createMlsDirectMessage(
|
||||
name: String?,
|
||||
description: String?,
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ActiveWallet?>,
|
||||
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
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user