Add primal RelaySocketManager

2026-04-18 02:41:31 +02:00
parent 8673d0b69f
commit 76858f9cf0
32 changed files with 1279 additions and 3 deletions
--- 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 {
--- 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')"
    ]
  }
 }
--- 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,
--- 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)
        }
    }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/RelayDao.kt
+++ 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<Relay>
+
+    @Query("SELECT * FROM Relay WHERE publicKey = :publicKey")
+    fun observePublicKeyRelays(publicKey: String): Flow<Relay?>
+
+    @Upsert
+    suspend fun upsert(relay: Relay)
+
+}
--- 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<TextNoteEvent>(
@@ -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
+        }
+    }
 }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/Relay.kt
+++ 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
+)
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NetworkException.kt
+++ 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)
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/exceptions/NostrNoticeException.kt
+++ 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")
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/dto/RelayDTO.kt
+++ 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)
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt
+++ 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() }
+
+
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt
+++ 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,
+)
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt
+++ 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<RelayDTO> = emptyList()
+        private set
+
+    @VisibleForTesting
+    var socketClients = listOf<NostrSocketClient>()
+
+    private val _relayPoolStatus = MutableStateFlow(mapOf<String, Boolean>())
+    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<RelayDTO>) {
+        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<RelayDTO>.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<String>, 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<NostrSocketClient>, nostrEvent: NostrEvent) {
+        val responseFlow = MutableSharedFlow<NostrPublishResult>()
+        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
+    }
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt
+++ 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<RelayDTO>?, walletRelays: List<RelayDTO>?) {
+        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<RelayDTO>) {
+        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)
+}
--- 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
@@ -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<List<String>> = emptyList(),
+)
--- 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
@@ -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<Event>,
+    val relays: List<String>,
+)
--- 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
@@ -0,0 +1,3 @@
+package net.primal.android.networking.relays.errors
+
+class NostrPublishException(override val cause: Throwable?) : RuntimeException()
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/CommonJson.kt
+++ 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 <reified T> Json.decodeFromStringOrNull(string: String?): T? {
+    if (string.isNullOrEmpty()) return null
+
+    return try {
+        decodeFromString(string)
+    } catch (_: Exception) {
+        null
+    }
+}
+
+inline fun <reified T> 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 <reified T> T.encodeToJsonString(): String {
+    return CommonJson.encodeToString(this)
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/serialization/SocketsJson.kt
+++ 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
+    }
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessage.kt
+++ 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<NostrEvent> = emptyList(),
+    ) : NostrIncomingMessage()
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageExt.kt
+++ 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<NostrIncomingMessage>.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<NostrIncomingMessage>.filterByEventId(id: String) =
+    filter {
+        (it is NostrIncomingMessage.OkMessage && it.eventId == id) ||
+            (it is NostrIncomingMessage.NoticeMessage)
+    }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrIncomingMessageParser.kt
+++ 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<JsonArray>(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<NostrEvent>()
+
+    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
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrOutgoingMessageBuilder.kt
+++ 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()
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClient.kt
+++ 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<NostrIncomingMessage>
+
+    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)
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientFactory.kt
+++ 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,
+    )
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrSocketClientImpl.kt
+++ 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<NostrIncomingMessage>()
+    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 }
+    }
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/NostrVerb.kt
+++ 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,
+    }
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/sockets/SocketConnectionCallbacks.kt
+++ 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
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/CachingImportRepository.kt
+++ 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<NostrEvent>)
+
+    suspend fun importEvents(events: List<NostrEvent>): Boolean
+    suspend fun broadcastEvents(events: List<NostrEvent>, relays: List<String>): Result<List<BroadcastEventResponse>>
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/RelayRepository.kt
+++ 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<List<Relay>>
+}
--- 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()))
--- 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]