diff --git a/composeApp/build.gradle.kts b/composeApp/build.gradle.kts index 24fa69e1..366fc62a 100644 --- a/composeApp/build.gradle.kts +++ b/composeApp/build.gradle.kts @@ -73,6 +73,8 @@ kotlin { implementation(libs.ktor.client.websockets) implementation(libs.ktor.serialization.kotlinx.json) + implementation(libs.okio) + implementation(libs.vitorpamplona.quartz) } commonTest.dependencies { diff --git a/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json b/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json index 82c63f10..9c86b76c 100644 --- a/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json +++ b/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json @@ -2,7 +2,7 @@ "formatVersion": 1, "database": { "version": 1, - "identityHash": "5f7afc4947b08cff2f59cfb44bfbf281", + "identityHash": "37219e58733785a0530005387bc4e1f4", "entities": [ { "tableName": "BroadcastNostrEventReceipt", @@ -788,6 +788,50 @@ ] } }, + { + "tableName": "Relay", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`publicKey` TEXT NOT NULL, `type` TEXT NOT NULL, `url` TEXT NOT NULL, `read` INTEGER NOT NULL, `write` INTEGER NOT NULL, PRIMARY KEY(`publicKey`, `type`, `url`))", + "fields": [ + { + "fieldPath": "publicKey", + "columnName": "publicKey", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "type", + "columnName": "type", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "url", + "columnName": "url", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "read", + "columnName": "read", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "write", + "columnName": "write", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "publicKey", + "type", + "url" + ] + } + }, { "tableName": "Repost", "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `repostedPostId` TEXT NOT NULL, `repostedPostTags` TEXT NOT NULL, `repostedPostCreatedAt` INTEGER NOT NULL, `repostedPostContent` TEXT NOT NULL, `repostedPostAuthorPublicKey` TEXT NOT NULL, `content` TEXT NOT NULL, `nostrEventId` TEXT NOT NULL, `profilePublicKey` TEXT NOT NULL, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, `savedAt` INTEGER NOT NULL, `viewedAt` INTEGER, `deletedAt` INTEGER, `broadcastedAt` INTEGER, PRIMARY KEY(`id`), FOREIGN KEY(`profilePublicKey`) REFERENCES `Profile`(`publicKey`) ON UPDATE NO ACTION ON DELETE CASCADE , FOREIGN KEY(`repostedPostAuthorPublicKey`) REFERENCES `Profile`(`publicKey`) ON UPDATE NO ACTION ON DELETE CASCADE , FOREIGN KEY(`nostrEventId`) REFERENCES `NostrEvent`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE , FOREIGN KEY(`repostedPostId`) REFERENCES `Post`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )", @@ -1409,7 +1453,7 @@ ], "setupQueries": [ "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)", - "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '5f7afc4947b08cff2f59cfb44bfbf281')" + "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '37219e58733785a0530005387bc4e1f4')" ] } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt index 368f2923..bb4ebe3f 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt @@ -21,6 +21,7 @@ import ac.aux.compose.database.model.Post import ac.aux.compose.database.model.Profile import ac.aux.compose.database.model.Reaction import ac.aux.compose.database.model.RecentSearch +import ac.aux.compose.database.model.Relay import ac.aux.compose.database.model.Repost import ac.aux.compose.database.model.SynchronizeNostrEventRequest import ac.aux.compose.database.model.SynchronizeNostrEventResult @@ -44,6 +45,7 @@ val GENESIS_AT = Instant.fromEpochMilliseconds(1231006505000L) Profile::class, Reaction::class, RecentSearch::class, + Relay::class, Repost::class, SynchronizeNostrEventRequest::class, SynchronizeNostrEventResult::class, diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt index da53e52c..53f4b8b8 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt @@ -330,7 +330,6 @@ abstract class NostrDao( ) } - // TODO: Placeholder for all the other things... database.zapDao().upsert(zap) } } diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/RelayDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/RelayDao.kt new file mode 100644 index 00000000..114ca10d --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/RelayDao.kt @@ -0,0 +1,21 @@ +package ac.aux.compose.database.dao + +import ac.aux.compose.database.model.Relay +import ac.aux.compose.database.model.UnsignedNostrEvent +import androidx.room.Dao +import androidx.room.Query +import androidx.room.Upsert +import kotlinx.coroutines.flow.Flow + +@Dao +interface RelayDao { + @Query("SELECT * FROM Relay") + fun getAllPosts(): List + + @Query("SELECT * FROM Relay WHERE publicKey = :publicKey") + fun observePublicKeyRelays(publicKey: String): Flow + + @Upsert + suspend fun upsert(relay: Relay) + +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/NostrEvent.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/NostrEvent.kt index 15dd1e10..1dc439ca 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/NostrEvent.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/NostrEvent.kt @@ -10,15 +10,26 @@ import androidx.room.Ignore import androidx.room.Index import androidx.room.PrimaryKey import co.touchlab.kermit.Logger +import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.Kind +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.core.TagArray import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd +import com.vitorpamplona.quartz.nip01Core.tags.events.firstTaggedEvent +import com.vitorpamplona.quartz.nip01Core.tags.people.firstTaggedUser import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent import com.vitorpamplona.quartz.nip18Reposts.RepostEvent import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent import com.vitorpamplona.quartz.utils.EventFactory +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonArray import kotlin.String import kotlin.time.Clock import kotlin.time.Instant @@ -86,6 +97,7 @@ data class NostrEvent( fun isProfileCreationEvent(): Boolean { return kind == MetadataEvent.KIND && unsignedNostrEventId != null } + fun toPost(): Post? = try { if (kind == TextNoteEvent.KIND) { val textNote = EventFactory.create( @@ -250,4 +262,52 @@ data class NostrEvent( logger.e("Failed to return Zap: ", e) return null } + + fun toNostrJsonObject(): JsonObject { + val nostrEvent = this + + return buildJsonObject { + put("id", nostrEvent.id) + put("pubkey", nostrEvent.pubKey) + put("created_at", nostrEvent.createdAt.epochSeconds) + put("kind", nostrEvent.kind) + putJsonArray("tags") { + nostrEvent.tags.forEach { tag -> + add( + JsonArray(tag.map { JsonPrimitive(it) }) + ) + } + } + put("content", nostrEvent.content) + put("sig", nostrEvent.sig) + } + } + + companion object { + + fun fromEvent(event: Event, synchronizeNostrEventRequest: SynchronizeNostrEventRequest? = null): NostrEvent? = try { + // TODO: Check for unsupported kind + val taggedEvent = event.firstTaggedEvent() + val taggedUser = event.firstTaggedUser() + + return NostrEvent( + id = event.id, + pubKey = event.pubKey, + kind = event.kind, + content = event.content, + tags = event.tags, + createdAt = Instant.fromEpochSeconds(event.createdAt), + unsignedNostrEventId = synchronizeNostrEventRequest?.unsignedNostrEventId, + sig = event.sig, + + taggedNostrEventId = taggedEvent?.eventId, + taggedNostrEventRelayUrl = taggedEvent?.relay?.url, + + taggedPublicKey = taggedUser?.pubKey, + taggedPublicKeyRelayUrl = taggedUser?.relayHint?.url + ) + } catch (e: Throwable) { + return null + } + } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/Relay.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/Relay.kt new file mode 100644 index 00000000..2a2fb8f2 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/Relay.kt @@ -0,0 +1,14 @@ +package ac.aux.compose.database.model + +import androidx.room.Entity + +@Entity( + primaryKeys = ["publicKey", "type", "url"] +) +data class Relay( + val publicKey: String, + val type: String, + val url: String, + val read: Boolean, + val write: Boolean +) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NetworkException.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NetworkException.kt new file mode 100644 index 00000000..12bb3be7 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NetworkException.kt @@ -0,0 +1,6 @@ +package ac.aux.compose.exceptions + +class NetworkException( + message: String? = null, + cause: Throwable? = null +): RuntimeException(message, cause) \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NostrNoticeException.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NostrNoticeException.kt new file mode 100644 index 00000000..4deb4eb8 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NostrNoticeException.kt @@ -0,0 +1,6 @@ +package ac.aux.compose.exceptions + +class NostrNoticeException( + val reason: String?, + val subscriptionId: String? = null +): RuntimeException("$subscriptionId: $reason") \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/dto/RelayDTO.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/dto/RelayDTO.kt new file mode 100644 index 00000000..1fe82a2a --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/dto/RelayDTO.kt @@ -0,0 +1,22 @@ +package ac.aux.compose.network.dto + +import ac.aux.compose.database.model.Relay +import kotlinx.serialization.Serializable + +@Serializable +data class RelayDTO( + val url: String, + val read: Boolean, + val write: Boolean, +) { + +} + +fun Relay.mapToRelayDTO() = + RelayDTO( + url = this.url, + read = this.read, + write = this.write, + ) + +fun String.toRelayDTO(): RelayDTO = RelayDTO(url = this, read = true, write = true) \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt new file mode 100644 index 00000000..471b4ee9 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt @@ -0,0 +1,16 @@ +package ac.aux.compose.network.relays + +import ac.aux.compose.network.dto.toRelayDTO + +val FALLBACK_RELAYS = listOf( + "wss://relay.primal.net", + "wss://relay.damus.io", + "wss://relay.nostr.band", + "wss://relay.current.fyi", + "wss://purplepag.es", + "wss://nos.lol", + "wss://offchain.pub", + "wss://nostr.bitcoiner.social", +).map { it.toRelayDTO() } + + diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt new file mode 100644 index 00000000..0a55a35f --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt @@ -0,0 +1,8 @@ +package ac.aux.compose.network.relays + +import ac.aux.compose.network.sockets.NostrIncomingMessage + +data class NostrPublishResult( + val result: NostrIncomingMessage? = null, + val error: Throwable? = null, +) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt new file mode 100644 index 00000000..191624ed --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt @@ -0,0 +1,203 @@ +package ac.aux.compose.network.relays + +import ac.aux.compose.database.model.NostrEvent +import ac.aux.compose.exceptions.NetworkException +import ac.aux.compose.exceptions.NostrNoticeException +import ac.aux.compose.network.dto.RelayDTO +import androidx.annotation.VisibleForTesting +import kotlin.time.Duration.Companion.milliseconds +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.FlowPreview +import kotlinx.coroutines.IO +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.catch +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.getAndUpdate +import kotlinx.coroutines.flow.timeout +import kotlinx.coroutines.flow.transform +import kotlinx.coroutines.launch +import net.primal.android.networking.relays.errors.NostrPublishException +import ac.aux.compose.network.sockets.NostrIncomingMessage +import ac.aux.compose.network.sockets.NostrSocketClient +import ac.aux.compose.network.sockets.NostrSocketClientFactory +import ac.aux.compose.network.sockets.SocketConnectionClosedCallback +import ac.aux.compose.network.sockets.SocketConnectionOpenedCallback +import ac.aux.compose.network.sockets.filterByEventId +import ac.aux.compose.network.sockets.parseIncomingMessage +import ac.aux.compose.repository.CachingImportRepository +import co.touchlab.kermit.Logger + +class RelayPool( + private val nostrSocketClientFactory: NostrSocketClientFactory, + private val cachingImportRepository: CachingImportRepository, +) { + val logger = Logger.withTag("RelayPool") + + companion object { + const val PUBLISH_TIMEOUT = 30_000 + } + + private val scope = CoroutineScope(Dispatchers.IO) + + var relays: List = emptyList() + private set + + @VisibleForTesting + var socketClients = listOf() + + private val _relayPoolStatus = MutableStateFlow(mapOf()) + val relayPoolStatus = _relayPoolStatus.asStateFlow() + private fun updateRelayStatus(url: String, connected: Boolean) = + scope.launch { + _relayPoolStatus.getAndUpdate { + it.toMutableMap().apply { this[url] = connected } + } + } + + private val onSocketConnectionOpenedCallback: SocketConnectionOpenedCallback = { url -> + updateRelayStatus(url = url, connected = true) + } + + private val onSocketConnectionClosedCallback: SocketConnectionClosedCallback = { url, _ -> + updateRelayStatus(url = url, connected = false) + } + + fun changeRelays(relays: List) { + val existingRelayUrls = socketClients.map { it.socketUrl } + val newRelayUrls = relays.map { it.url } + + val toAddRelayUrls = newRelayUrls.filter { it !in existingRelayUrls } + val toAddSocketClients = relays.filter { it.url in toAddRelayUrls }.mapAsNostrSocketClient() + val toRemoveSocketClients = socketClients.filter { it.socketUrl !in newRelayUrls } + + val newSocketClients = socketClients.toMutableList().apply { + removeAll(toRemoveSocketClients) + addAll(toAddSocketClients) + } + + socketClients = newSocketClients + toRemoveSocketClients.forEach { client -> + updateRelayStatus(url = client.socketUrl, connected = false) + scope.launch { client.close() } + } + this.relays = relays + } + + fun closePool() { + socketClients.forEach { client -> + updateRelayStatus(url = client.socketUrl, connected = false) + scope.launch { client.close() } + } + socketClients = emptyList() + relays = emptyList() + } + + fun hasRelays() = relays.isNotEmpty() + + suspend fun tryConnectingToRelay(url: String) { + runCatching { + socketClients.find { it.socketUrl == url }?.ensureSocketConnectionOrThrow() + } + } + + private fun List.mapAsNostrSocketClient() = + this.map { + nostrSocketClientFactory.create( + wssUrl = it.url, + onSocketConnectionOpened = onSocketConnectionOpenedCallback, + onSocketConnectionClosed = onSocketConnectionClosedCallback, + ) + } + + @Throws(NostrPublishException::class) + suspend fun publishEvent(nostrEvent: NostrEvent, cachingProxyEnabled: Boolean = false) { + if (cachingProxyEnabled) { + handleBroadcastEventThroughCachingProxy(relays.map { it.url }, nostrEvent) + } else { + handlePublishEventToRelays(socketClients, nostrEvent) + } + } + + private suspend fun handleBroadcastEventThroughCachingProxy(relayUrls: List, nostrEvent: NostrEvent) { + val result = + cachingImportRepository.broadcastEvents( + events = listOf(nostrEvent), + relays = relayUrls, + ).getOrNull() + ?: throw NostrPublishException( + cause = NetworkException(message = "Primal NostrEvent 10_000_149 not found or invalid."), + ) + + result.find { response -> response.eventId == nostrEvent.id }?.responses + ?.mapNotNull { relayResponse -> + val relay = relayResponse.firstOrNull() + val responseMessage = relayResponse.getOrNull(index = 1)?.parseIncomingMessage() + if (relay != null && responseMessage != null) { + relay to responseMessage + } else { + null + } + } + ?.find { (_, relayMessage) -> relayMessage is NostrIncomingMessage.OkMessage && relayMessage.success } + ?: throw NostrPublishException( + cause = NetworkException("Event broadcast failed. Could not find success response from relays."), + ) + } + + @OptIn(FlowPreview::class) + private suspend fun handlePublishEventToRelays(relayConnections: List, nostrEvent: NostrEvent) { + val responseFlow = MutableSharedFlow() + relayConnections.forEach { nostrSocketClient -> + scope.launch { + with(nostrSocketClient) { + val sendEventResult = runCatching { + ensureSocketConnectionOrThrow() + sendEVENT(nostrEvent.toNostrJsonObject()) + collectPublishResponse(eventId = nostrEvent.id) + } + sendEventResult.getOrNull()?.let { + responseFlow.emit(NostrPublishResult(result = it)) + } + sendEventResult.exceptionOrNull()?.let { + logger.w(throwable = it) { "sendEVENT failed to $socketUrl" } + responseFlow.emit(NostrPublishResult(error = it)) + } + } + } + } + + var responseCount = 0 + responseFlow.timeout(PUBLISH_TIMEOUT.milliseconds) + .catch { throw NostrPublishException(cause = it) } + .transform { + emit(it) + responseCount++ + if (relayConnections.size == responseCount && !it.isSuccessful()) { + throw NostrPublishException(cause = null) + } + } + .first { it.isSuccessful() } + } + + @FlowPreview + private suspend fun NostrSocketClient.collectPublishResponse(eventId: String): NostrIncomingMessage.OkMessage { + return incomingMessages + .filterByEventId(id = eventId) + .transform { + when (it) { + is NostrIncomingMessage.OkMessage -> emit(it) + is NostrIncomingMessage.NoticeMessage -> throw NostrNoticeException(reason = it.message) + else -> error("$it is not allowed") + } + } + .timeout(PUBLISH_TIMEOUT.milliseconds) + .first() + } + + private fun NostrPublishResult.isSuccessful(): Boolean { + return result is NostrIncomingMessage.OkMessage && result.success + } +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt new file mode 100644 index 00000000..5faee18d --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt @@ -0,0 +1,141 @@ +package ac.aux.compose.network.relays + +import ac.aux.compose.database.model.NostrEvent +import ac.aux.compose.network.dto.RelayDTO +import ac.aux.compose.network.dto.mapToRelayDTO +import ac.aux.compose.network.sockets.NostrSocketClientFactory +import ac.aux.compose.repository.CachingImportRepository +import ac.aux.compose.repository.RelayRepository +import co.touchlab.kermit.Logger +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.IO +import kotlinx.coroutines.Job +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import net.primal.android.networking.relays.errors.NostrPublishException + + +class RelaysSocketManager constructor( + private val nostrSocketClientFactory: NostrSocketClientFactory, + private val cachingImportRepository: CachingImportRepository, +// private val activeAccountStore: ActiveAccountStore, +// private val usersDatabase: UsersDatabase, + private val relayRepository: RelayRepository +) { + val logger = Logger.withTag("RelaysSocketManager") + private val scope = CoroutineScope(Dispatchers.IO) + private val relayPoolsMutex = Mutex() + + private var relaysObserverJob: Job? = null + + private fun buildRelayPool() = + RelayPool( + nostrSocketClientFactory = nostrSocketClientFactory, + cachingImportRepository = cachingImportRepository, + ) + + private val userRelaysPool: RelayPool = buildRelayPool() + private val nwcRelaysPool: RelayPool = buildRelayPool() + private val fallbackRelaysPool: RelayPool = buildRelayPool() + + val userRelayPoolStatus = userRelaysPool.relayPoolStatus + + init { + initFallbackRelaysPool() + observeActiveUserId() + } + + private fun initFallbackRelaysPool() = fallbackRelaysPool.changeRelays(FALLBACK_RELAYS) + + private fun observeActiveUserId() = + scope.launch { +// TODO activeAccountStore.activeUserId.collect { userId -> +// when { +// userId.isEmpty() -> { +// relaysObserverJob?.cancel() +// relaysObserverJob = null +// clearRelayPools() +// } +// +// else -> { +// relaysObserverJob?.cancel() +// relaysObserverJob = observeRelays(userId) +// } +// } +// } + } + + private suspend fun isCachingProxyEnabled() = false // TODO: activeAccountStore.activeUserAccount().cachingProxyEnabled + + private fun observeRelays(publicKey: String): Job = + scope.launch { + try { + relayRepository.observePublicKeyRelays(publicKey = publicKey).collect { relays -> + val userRelays = relays.filter { it.type != "nwc" }.map { it.mapToRelayDTO() } + val nwcRelays = relays.filter { it.type == "nwc" }.map { it.mapToRelayDTO() } + updateRelayPools(regularRelays = userRelays, walletRelays = nwcRelays) + } + } catch (error: CancellationException) { + logger.w(throwable = error) { "Relay observation cancelled" } + } + } + + private suspend fun updateRelayPools(regularRelays: List?, walletRelays: List?) { + relayPoolsMutex.withLock { + val userRelaysChanged = userRelaysPool.relays != regularRelays + if (userRelaysChanged && !regularRelays.isNullOrEmpty()) { + userRelaysPool.changeRelays(relays = regularRelays) + } + + val nwcRelaysChanged = nwcRelaysPool.relays != walletRelays + if (nwcRelaysChanged && !walletRelays.isNullOrEmpty()) { + nwcRelaysPool.changeRelays(relays = walletRelays) + } + } + } + + private suspend fun clearRelayPools() = + relayPoolsMutex.withLock { + userRelaysPool.closePool() + nwcRelaysPool.closePool() + } + + @Throws(NostrPublishException::class) + suspend fun publishEvent(nostrEvent: NostrEvent) { + if (userRelaysPool.hasRelays()) { + userRelaysPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + } else { + fallbackRelaysPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + } + } + + @Throws(NostrPublishException::class) + suspend fun publishEvent(nostrEvent: NostrEvent, relays: List) { + val customPool = buildRelayPool() + customPool.changeRelays(relays = relays) + customPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + customPool.closePool() + } + + @Throws(NostrPublishException::class) + suspend fun publishNwcEvent(nostrEvent: NostrEvent) { + if (!nwcRelaysPool.hasRelays()) { + throw NostrPublishException(cause = IllegalStateException("nwc relay not found")) + } + + nwcRelaysPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + } + + fun tryConnectingToAllUserRelays() { + userRelaysPool.relays.forEach { + scope.launch { + userRelaysPool.tryConnectingToRelay(it.url) + } + } + } + + suspend fun tryConnectingToUserRelay(url: String) = userRelaysPool.tryConnectingToRelay(url) +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/broadcast/BroadcastEventResponse.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/broadcast/BroadcastEventResponse.kt new file mode 100644 index 00000000..4037edff --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/broadcast/BroadcastEventResponse.kt @@ -0,0 +1,10 @@ +package ac.aux.compose.network.relays.broadcast + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +@Serializable +data class BroadcastEventResponse( + @SerialName("event_id") val eventId: String, + val responses: List> = emptyList(), +) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/broadcast/BroadcastRequestBody.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/broadcast/BroadcastRequestBody.kt new file mode 100644 index 00000000..6f98034e --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/broadcast/BroadcastRequestBody.kt @@ -0,0 +1,10 @@ +package ac.aux.compose.network.relays.broadcast + +import com.vitorpamplona.quartz.nip01Core.core.Event +import kotlinx.serialization.Serializable + +@Serializable +data class BroadcastRequestBody( + val events: List, + val relays: List, +) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/errors/NostrPublishException.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/errors/NostrPublishException.kt new file mode 100644 index 00000000..85a551b5 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/errors/NostrPublishException.kt @@ -0,0 +1,3 @@ +package net.primal.android.networking.relays.errors + +class NostrPublishException(override val cause: Throwable?) : RuntimeException() diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/CommonJson.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/CommonJson.kt new file mode 100644 index 00000000..9e9dd669 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/CommonJson.kt @@ -0,0 +1,49 @@ +package ac.aux.compose.network.serialization + +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonBuilder + +private val defaultJsonBuilder: (JsonBuilder.() -> Unit) = { + ignoreUnknownKeys = true + coerceInputValues = true +} + +val CommonJson = Json { + defaultJsonBuilder() +} + +val CommonJsonImplicitNulls = Json { + defaultJsonBuilder() + explicitNulls = false +} + +val CommonJsonEncodeDefaults = Json { + defaultJsonBuilder() + encodeDefaults = true +} + + + +inline fun Json.decodeFromStringOrNull(string: String?): T? { + if (string.isNullOrEmpty()) return null + + return try { + decodeFromString(string) + } catch (_: Exception) { + null + } +} + +inline fun String?.decodeFromJsonStringOrNull(): T? { + return CommonJson.decodeFromStringOrNull(this) +} + +/** + * Encodes an object to a JSON string using CommonJson serializer. + * + * Note: When working with JsonObject, use `toString()` instead of this function + * to ensure proper formatting and compatibility with iOS. + */ +inline fun T.encodeToJsonString(): String { + return CommonJson.encodeToString(this) +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/SocketsJson.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/SocketsJson.kt new file mode 100644 index 00000000..6fbe0dd2 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/SocketsJson.kt @@ -0,0 +1,20 @@ +package ac.aux.compose.network.serialization + +import ac.aux.compose.database.model.NostrEvent +import co.touchlab.kermit.Logger +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.decodeFromJsonElement + +internal val SocketsJson = CommonJson + + +fun JsonObject?.asNostrEventOrNull(): NostrEvent? { + return try { + if (this != null) SocketsJson.decodeFromJsonElement(this) else null + } catch (error: IllegalArgumentException) { + Logger.withTag("JsonObject?.asNostrEventOrNull").w(error) { "Unable to map as NostrEvent." } + this?.let(SocketsJson::encodeToString)?.let { Logger.withTag("JsonObject?.asNostrEventOrNull").w { it } } + null + } +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessage.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessage.kt new file mode 100644 index 00000000..aebf30dc --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessage.kt @@ -0,0 +1,40 @@ +package ac.aux.compose.network.sockets + +import ac.aux.compose.database.model.NostrEvent + +sealed class NostrIncomingMessage { + + data class EventMessage( + val subscriptionId: String, + val nostrEvent: NostrEvent? = null, + ) : NostrIncomingMessage() + + data class EoseMessage( + val subscriptionId: String, + ) : NostrIncomingMessage() + + data class OkMessage( + val eventId: String, + val success: Boolean, + val message: String? = null, + ) : NostrIncomingMessage() + + data class NoticeMessage( + val subscriptionId: String? = null, + val message: String? = null, + ) : NostrIncomingMessage() + + data class AuthMessage( + val challenge: String, + ) : NostrIncomingMessage() + + data class CountMessage( + val subscriptionId: String, + val count: Int, + ) : NostrIncomingMessage() + + data class EventsMessage( + val subscriptionId: String, + val nostrEvents: List = emptyList(), + ) : NostrIncomingMessage() +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageExt.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageExt.kt new file mode 100644 index 00000000..368ff850 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageExt.kt @@ -0,0 +1,19 @@ +package ac.aux.compose.network.sockets + +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.filter + +fun Flow.filterBySubscriptionId(id: String) = + filter { + (it is NostrIncomingMessage.EventMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.EoseMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.CountMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.EventsMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.NoticeMessage) + } + +fun Flow.filterByEventId(id: String) = + filter { + (it is NostrIncomingMessage.OkMessage && it.eventId == id) || + (it is NostrIncomingMessage.NoticeMessage) + } diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageParser.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageParser.kt new file mode 100644 index 00000000..db5e8e94 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageParser.kt @@ -0,0 +1,151 @@ +package ac.aux.compose.network.sockets + +import ac.aux.compose.database.model.NostrEvent +import ac.aux.compose.network.serialization.SocketsJson +import ac.aux.compose.network.serialization.decodeFromStringOrNull +import co.touchlab.kermit.Logger +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.Kind +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.booleanOrNull +import kotlinx.serialization.json.intOrNull +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive + +fun String.parseIncomingMessage(): NostrIncomingMessage? { + val jsonArray = SocketsJson.decodeFromStringOrNull(this) + val verbElement = jsonArray?.elementAtOrNull(0) ?: return null + + return try { + when (verbElement.toIncomingMessageType()) { + NostrVerb.Incoming.EVENT -> jsonArray.takeAsEventIncomingMessage() + NostrVerb.Incoming.EOSE -> jsonArray.takeAsEoseIncomingMessage() + NostrVerb.Incoming.OK -> jsonArray.takeAsOkIncomingMessage() + NostrVerb.Incoming.NOTICE -> jsonArray.takeAsNoticeIncomingMessage() + NostrVerb.Incoming.AUTH -> jsonArray.takeAsAuthIncomingMessage() + NostrVerb.Incoming.COUNT -> jsonArray.takeAsCountIncomingMessage() + NostrVerb.Incoming.EVENTS -> jsonArray.takeAsEventsIncomingMessage() + } + } catch (error: Exception) { + Logger.withTag("String.parseIncomingMessage").w(error) { "Unable to parse incoming message." } + null + } +} + +private fun JsonArray.takeAsAuthIncomingMessage(): NostrIncomingMessage? { + val challenge = elementAtOrNull(1) ?: return null + return NostrIncomingMessage.AuthMessage( + challenge = challenge.jsonPrimitive.content, + ) +} + +private fun JsonArray.takeAsCountIncomingMessage(): NostrIncomingMessage? { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val count = elementAtOrNull(2) + ?.jsonObject + ?.get("count") + ?.jsonPrimitive?.intOrNull + + return if (subscriptionId != null && count != null) { + NostrIncomingMessage.CountMessage( + subscriptionId = subscriptionId, + count = count, + ) + } else { + null + } +} + +private fun JsonArray.takeAsEoseIncomingMessage(): NostrIncomingMessage? { + val subscriptionElement = elementAtOrNull(1) ?: return null + return NostrIncomingMessage.EoseMessage( + subscriptionId = subscriptionElement.toSubscriptionId(), + ) +} + +private fun JsonArray.takeAsEventIncomingMessage(): NostrIncomingMessage? { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val event = elementAtOrNull(2)?.jsonObject + val kind = event?.getMessageNostrEventKind() + + if (subscriptionId == null || kind == null) return null + + val nostrEvent = NostrEvent.fromEvent( + Event.fromJson( + event.toString() + ) + ) + + return NostrIncomingMessage.EventMessage( + subscriptionId = subscriptionId, + nostrEvent = nostrEvent, + ) +} + +private fun JsonArray.takeAsEventsIncomingMessage(): NostrIncomingMessage? { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val events = elementAtOrNull(2)?.jsonArray + + if (subscriptionId == null || events == null) return null + + val nostrEvents = mutableListOf() + + events.map { it.jsonObject }.forEach { jsonEvent -> + NostrEvent.fromEvent( + Event.fromJson( + jsonEvent.toString() + ) + )?.let { nostrEvent -> + nostrEvents.add(nostrEvent) + } + } + + return NostrIncomingMessage.EventsMessage( + subscriptionId = subscriptionId, + nostrEvents = nostrEvents, + ) +} + +private fun JsonObject.getMessageNostrEventKind(): Kind { + val kind = this["kind"]?.jsonPrimitive?.intOrNull + return kind ?: -1 +} + +private fun JsonArray.takeAsNoticeIncomingMessage(): NostrIncomingMessage { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val messageText = elementAtOrNull(2)?.jsonPrimitive?.content + return NostrIncomingMessage.NoticeMessage(subscriptionId = subscriptionId, message = messageText) +} + +private fun JsonArray.takeAsOkIncomingMessage(): NostrIncomingMessage? { + val eventId = elementAtOrNull(1)?.jsonPrimitive?.content + val success = elementAtOrNull(2)?.jsonPrimitive?.booleanOrNull + val message = elementAtOrNull(3)?.jsonPrimitive?.content + + return if (eventId != null && success != null) { + NostrIncomingMessage.OkMessage( + eventId = eventId, + success = success, + message = message, + ) + } else { + null + } +} + +private fun JsonElement.toIncomingMessageType(): NostrVerb.Incoming { + return when (this.jsonPrimitive.content) { + "EVENT" -> NostrVerb.Incoming.EVENT + "EOSE" -> NostrVerb.Incoming.EOSE + "OK" -> NostrVerb.Incoming.OK + "AUTH" -> NostrVerb.Incoming.AUTH + "COUNT" -> NostrVerb.Incoming.COUNT + "EVENTS" -> NostrVerb.Incoming.EVENTS + else -> NostrVerb.Incoming.NOTICE + } +} + +private fun JsonElement.toSubscriptionId(): String = this.jsonPrimitive.content diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrOutgoingMessageBuilder.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrOutgoingMessageBuilder.kt new file mode 100644 index 00000000..adf18e86 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrOutgoingMessageBuilder.kt @@ -0,0 +1,42 @@ +package ac.aux.compose.network.sockets + +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.add +import kotlinx.serialization.json.buildJsonArray + +internal fun JsonObject.buildNostrREQMessage(subscriptionId: String): String { + return buildJsonArray { + add(NostrVerb.Outgoing.REQ.toString()) + add(subscriptionId) + add(this@buildNostrREQMessage) + }.toString() +} + +internal fun JsonObject.buildNostrEVENTMessage(): String { + return buildJsonArray { + add(NostrVerb.Outgoing.EVENT.toString()) + add(this@buildNostrEVENTMessage) + }.toString() +} + +internal fun JsonObject.buildNostrAUTHMessage(): String { + return buildJsonArray { + add(NostrVerb.Outgoing.AUTH.toString()) + add(this@buildNostrAUTHMessage) + }.toString() +} + +internal fun JsonObject.buildNostrCOUNTMessage(subscriptionId: String): String { + return buildJsonArray { + add(NostrVerb.Outgoing.COUNT.toString()) + add(subscriptionId) + add(this@buildNostrCOUNTMessage) + }.toString() +} + +internal fun String.buildNostrCLOSEMessage(): String { + return buildJsonArray { + add(NostrVerb.Outgoing.CLOSE.toString()) + add(this@buildNostrCLOSEMessage) + }.toString() +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClient.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClient.kt new file mode 100644 index 00000000..0da84a89 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClient.kt @@ -0,0 +1,29 @@ +package ac.aux.compose.network.sockets + +import kotlin.coroutines.cancellation.CancellationException +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.serialization.json.JsonObject + +interface NostrSocketClient { + val socketUrl: String + + val incomingMessages: SharedFlow + + suspend fun close() + + @Throws( + Exception::class, // TODO: Use NetworkExtension + CancellationException::class, + ) + suspend fun ensureSocketConnectionOrThrow() + + suspend fun sendAUTH(signedEvent: JsonObject) + + suspend fun sendCLOSE(subscriptionId: String) + + suspend fun sendCOUNT(data: JsonObject): String + + suspend fun sendEVENT(signedEvent: JsonObject) + + suspend fun sendREQ(subscriptionId: String, data: JsonObject) +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientFactory.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientFactory.kt new file mode 100644 index 00000000..9855c29b --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientFactory.kt @@ -0,0 +1,52 @@ +package ac.aux.compose.network.sockets + +import io.ktor.client.HttpClient +import io.ktor.client.plugins.websocket.WebSockets +import io.ktor.serialization.kotlinx.KotlinxWebsocketSerializationConverter +import kotlinx.serialization.json.Json + +internal val defaultSocketsHttpClient by lazy { + + HttpClient() { + install(WebSockets) { + contentConverter = KotlinxWebsocketSerializationConverter(Json { + isLenient = true + ignoreUnknownKeys = true + }) + pingIntervalMillis = 20_000 + } + } +} + + +object NostrSocketClientFactory { + + fun create( + wssUrl: String, + httpClient: HttpClient, + incomingCompressionEnabled: Boolean = false, + onSocketConnectionOpened: SocketConnectionOpenedCallback? = null, + onSocketConnectionClosed: SocketConnectionClosedCallback? = null, + ): NostrSocketClient { + return NostrSocketClientImpl( + httpClient = httpClient, + wssUrl = wssUrl, + incomingCompressionEnabled = incomingCompressionEnabled, + onSocketConnectionOpened = onSocketConnectionOpened, + onSocketConnectionClosed = onSocketConnectionClosed, + ) + } + + fun create( + wssUrl: String, + incomingCompressionEnabled: Boolean = false, + onSocketConnectionOpened: SocketConnectionOpenedCallback? = null, + onSocketConnectionClosed: SocketConnectionClosedCallback? = null, + ) = create( + httpClient = defaultSocketsHttpClient, + wssUrl = wssUrl, + incomingCompressionEnabled = incomingCompressionEnabled, + onSocketConnectionOpened = onSocketConnectionOpened, + onSocketConnectionClosed = onSocketConnectionClosed, + ) +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientImpl.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientImpl.kt new file mode 100644 index 00000000..4aead323 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientImpl.kt @@ -0,0 +1,222 @@ +package ac.aux.compose.network.sockets + +import ac.aux.compose.exceptions.NetworkException +import co.touchlab.kermit.Logger +import io.ktor.client.HttpClient +import io.ktor.client.plugins.websocket.webSocketSession +import io.ktor.websocket.CloseReason +import io.ktor.websocket.Frame +import io.ktor.websocket.WebSocketSession +import io.ktor.websocket.close +import io.ktor.websocket.readReason +import io.ktor.websocket.readText +import kotlin.coroutines.cancellation.CancellationException +import kotlin.time.Duration.Companion.milliseconds +import kotlin.uuid.Uuid +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.IO +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.serialization.json.JsonObject +import okio.Buffer +import okio.GzipSink +import okio.Inflater +import okio.InflaterSource +import okio.buffer +import okio.use +import kotlin.uuid.ExperimentalUuidApi + +@OptIn(ExperimentalUuidApi::class) +internal class NostrSocketClientImpl( + wssUrl: String, + private val httpClient: HttpClient, + private val incomingCompressionEnabled: Boolean = false, + private val onSocketConnectionOpened: SocketConnectionOpenedCallback? = null, + private val onSocketConnectionClosed: SocketConnectionClosedCallback? = null, +) : NostrSocketClient { + + val logger = Logger.withTag("NostrSocketClientImpl") + + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + + private val wsMutex = Mutex() + private var wsSession: WebSocketSession? = null + private var wsReceiverJob: Job? = null + + private val _incomingMessages = MutableSharedFlow() + override val incomingMessages = _incomingMessages.asSharedFlow() + + override val socketUrl = wssUrl.cleanWebSocketUrl() + + override suspend fun ensureSocketConnectionOrThrow() { + if (wsSession != null && wsSession?.isActive == true) return + + wsMutex.withLock { + if (wsSession == null || wsSession?.isActive == false) { + wsSession = acquireWebSocketSession(url = socketUrl) + } + } + } + + private suspend fun acquireWebSocketSession(url: String): WebSocketSession { + return try { + httpClient.webSocketSession(urlString = url).apply { + launchWebSocketReceiver() + onSocketConnectionOpened?.invoke(url) + if (incomingCompressionEnabled) { + val id = Uuid.generateV4().toHexDashString() + sendMessage( + text = """["REQ","$id",{"cache":["set_primal_protocol",{"compression":"zlib"}]}]""", + ensureSessionBeforeSend = false, + ) + } + } + } catch (error: Exception) { + logger.w("NostrSocketClient::acquireWebSocketSession($socketUrl) failed.", error) + close() + onSocketConnectionClosed?.invoke(socketUrl, error) + throw NetworkException(cause = error) + } + } + + private fun WebSocketSession.launchWebSocketReceiver() { + wsReceiverJob?.cancel() + wsReceiverJob = scope.launch { + receiveSocketMessages() + } + } + + private suspend fun WebSocketSession.receiveSocketMessages() { + try { + for (frame in incoming) { + when (frame) { + is Frame.Text -> { + val text = frame.readText() + logLargeText(text = text, url = socketUrl, incoming = true) + processIncomingMessage(text = text) + } + + is Frame.Binary -> { + val decompressedMessage = decompressMessage(frame.data) + logLargeText(text = decompressedMessage, url = socketUrl, incoming = true) + processIncomingMessage(text = decompressedMessage) + } + + is Frame.Close -> { + val closeReason = frame.readReason() + logger.w { "WS $socketUrl closed. [${closeReason?.code}, ${closeReason?.message}]" } + close() + onSocketConnectionClosed?.invoke(socketUrl, null) + } + + else -> Unit + } + } + } catch (error: CancellationException) { + logger.w("NostrSocketClient::receiveSocketMessages() on $socketUrl cancelled.") + throw error + } catch (error: Exception) { + logger.w("NostrSocketClient::receiveSocketMessages() on $socketUrl failed.", error) + close() + onSocketConnectionClosed?.invoke(socketUrl, error) + } + } + + override suspend fun close() { + wsReceiverJob?.cancel() + wsReceiverJob = null + runCatching { + wsSession?.close( + reason = CloseReason( + code = CloseReason.Codes.NORMAL, + message = "Closed by client.", + ), + ) + } + wsSession = null + } + + private fun processIncomingMessage(text: String) { + text.parseIncomingMessage()?.let { + scope.launch { + if (it is NostrIncomingMessage.EoseMessage) { + delay(75.milliseconds) + } + _incomingMessages.emit(value = it) + } + } + } + + private suspend fun sendMessage(text: String, ensureSessionBeforeSend: Boolean = true) { + if (ensureSessionBeforeSend) { + ensureSocketConnectionOrThrow() + } + logLargeText(text = text, url = socketUrl, incoming = false) + wsSession?.send(Frame.Text(text = text)) + } + + override suspend fun sendREQ(subscriptionId: String, data: JsonObject) { + val reqMessage = data.buildNostrREQMessage(subscriptionId) + return sendMessage(text = reqMessage) + } + + override suspend fun sendCOUNT(data: JsonObject): String { + val subscriptionId: String = Uuid.generateV4().toHexDashString() + val reqMessage = data.buildNostrCOUNTMessage(subscriptionId) + sendMessage(text = reqMessage) + return subscriptionId + } + + override suspend fun sendCLOSE(subscriptionId: String) = sendMessage(text = subscriptionId.buildNostrCLOSEMessage()) + + override suspend fun sendEVENT(signedEvent: JsonObject) = sendMessage(text = signedEvent.buildNostrEVENTMessage()) + + override suspend fun sendAUTH(signedEvent: JsonObject) = sendMessage(text = signedEvent.buildNostrAUTHMessage()) + + private fun logLargeText( + text: String, + url: String, + incoming: Boolean, + ) { + val chunks = text.chunked(size = 3_500) + val chunksCount = chunks.size + chunks.forEachIndexed { index, chunk -> + val prefix = if (incoming) "<--" else "-->" + val suffix = if (index == chunksCount - 1) "[$url]" else "" + logger.d( + "$prefix $chunk $suffix", + ) + } + } + + @Suppress("unused") + private fun compressMessage(message: String): ByteArray { + val buffer = Buffer() + GzipSink(buffer).buffer().use { sink -> + sink.writeUtf8(message) + sink.flush() // Ensure all data is written + } + return buffer.readByteArray() + } + + private fun decompressMessage(compressedMessage: ByteArray): String { + val buffer = Buffer().write(compressedMessage) + InflaterSource(buffer, Inflater(false)).buffer().use { source -> + return source.readUtf8() + } + } + + private fun String.cleanWebSocketUrl(): String { + return replace("https://", "wss://", ignoreCase = true) + .replace("http://", "ws://", ignoreCase = true) + .let { if (it.endsWith("/")) it.dropLast(1) else it } + } +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrVerb.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrVerb.kt new file mode 100644 index 00000000..45b4b720 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrVerb.kt @@ -0,0 +1,49 @@ +package ac.aux.compose.network.sockets + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +internal sealed class NostrVerb { + + @Serializable + enum class Outgoing { + @SerialName("AUTH") + AUTH, + + @SerialName("CLOSE") + CLOSE, + + @SerialName("COUNT") + COUNT, + + @SerialName("EVENT") + EVENT, + + @SerialName("REQ") + REQ, + } + + @Serializable + enum class Incoming { + @SerialName("AUTH") + AUTH, + + @SerialName("COUNT") + COUNT, + + @SerialName("EOSE") + EOSE, + + @SerialName("EVENT") + EVENT, + + @SerialName("NOTICE") + NOTICE, + + @SerialName("OK") + OK, + + @SerialName("EVENTS") + EVENTS, + } +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/SocketConnectionCallbacks.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/SocketConnectionCallbacks.kt new file mode 100644 index 00000000..c6e97ef6 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/SocketConnectionCallbacks.kt @@ -0,0 +1,4 @@ +package ac.aux.compose.network.sockets + +typealias SocketConnectionOpenedCallback = (url: String) -> Unit +typealias SocketConnectionClosedCallback = (url: String, error: Throwable?) -> Unit diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/CachingImportRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/CachingImportRepository.kt new file mode 100644 index 00000000..60d4e4d7 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/CachingImportRepository.kt @@ -0,0 +1,12 @@ +package ac.aux.compose.repository + +import ac.aux.compose.database.model.NostrEvent +import ac.aux.compose.network.relays.broadcast.BroadcastEventResponse + +interface CachingImportRepository { + suspend fun cacheNostrEvents(vararg events: NostrEvent) + suspend fun cacheNostrEvents(events: List) + + suspend fun importEvents(events: List): Boolean + suspend fun broadcastEvents(events: List, relays: List): Result> +} diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/RelayRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/RelayRepository.kt new file mode 100644 index 00000000..0890b04d --- /dev/null +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/RelayRepository.kt @@ -0,0 +1,8 @@ +package ac.aux.compose.repository + +import ac.aux.compose.database.model.Relay +import kotlinx.coroutines.flow.Flow + +interface RelayRepository { + suspend fun observePublicKeyRelays(publicKey: String): Flow> +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/FeedListViewModel.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/FeedListViewModel.kt index 21c836d2..ab47415d 100755 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/FeedListViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/FeedListViewModel.kt @@ -17,10 +17,15 @@ import androidx.lifecycle.viewmodel.initializer import androidx.lifecycle.viewmodel.viewModelFactory import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.nip01Core.core.toHexKey +import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent +import com.vitorpamplona.quartz.nip04Dm.messages.PrivateDmEvent import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent import com.vitorpamplona.quartz.nip18Reposts.RepostEvent import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent +import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent +import com.vitorpamplona.quartz.nip96FileStorage.config.FileServersEvent +import com.vitorpamplona.quartz.nipB7Blossom.BlossomServersEvent import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.flow.firstOrNull @@ -78,6 +83,11 @@ class FeedListViewModel( RepostEvent.KIND, ReactionEvent.KIND, LnZapEvent.KIND, + ContactListEvent.KIND, + AdvertisedRelayListEvent.KIND, + PrivateDmEvent.KIND, + BlossomServersEvent.KIND, + FileServersEvent.KIND ), tags = mapOf( Pair("p", listOf(SeedManager.activePublicKey().toHexKey())) diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index ca118ff9..812345f3 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -24,6 +24,7 @@ materialIconsCore = "1.7.3" materialIconsExtended = "1.7.3" navigationCompose = "2.9.2" okhttp = "5.3.2" +okio = "3.16.4" pagingCommon = "3.5.0-alpha01" quartz = "1.06.3" room = "2.8.4" @@ -65,6 +66,7 @@ ktor-client-websockets = { module = "io.ktor:ktor-client-websockets", version.re ktor-serialization-kotlinx-json = { module = "io.ktor:ktor-serialization-kotlinx-json", version.ref = "ktor" } navigation-compose = { module = "org.jetbrains.androidx.navigation:navigation-compose", version.ref = "navigationCompose" } okhttp-coroutines = { group = "com.squareup.okhttp3", name = "okhttp-coroutines", version.ref = "okhttp" } +okio = { module = "com.squareup.okio:okio", version.ref = "okio" } vitorpamplona-quartz = { module = "com.vitorpamplona.quartz:quartz", version.ref = "quartz" } [plugins]