Get to the bottom of this issue.

composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json

@@ -2,7 +2,7 @@
   "formatVersion": 1,
   "database": {
     "version": 1,
-    "identityHash": "5b27659e597a301316dcc8ed9eaf8a3e",
+    "identityHash": "866bd337c819985dae51a248851b3dd3",
     "entities": [
       {
         "tableName": "BroadcastNostrEventReceipt",
@@ -873,7 +873,7 @@
       },
       {
         "tableName": "UnsignedNostrEvent",
-        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `pubKey` TEXT NOT NULL, `kind` INTEGER NOT NULL, `tags` TEXT NOT NULL, `content` TEXT NOT NULL, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, `savedAt` INTEGER NOT NULL, `deletedAt` INTEGER, `broadcastedAt` INTEGER)",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `pubKey` TEXT NOT NULL, `kind` INTEGER NOT NULL, `tags` TEXT NOT NULL, `content` TEXT NOT NULL, `signedAt` INTEGER, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, `savedAt` INTEGER NOT NULL, `deletedAt` INTEGER, `broadcastedAt` INTEGER)",
         "fields": [
           {
             "fieldPath": "id",
@@ -905,6 +905,11 @@
             "affinity": "TEXT",
             "notNull": true
           },
+          {
+            "fieldPath": "signedAt",
+            "columnName": "signedAt",
+            "affinity": "INTEGER"
+          },
           {
             "fieldPath": "createdAt",
             "columnName": "createdAt",
@@ -1129,7 +1134,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, '5b27659e597a301316dcc8ed9eaf8a3e')"
+      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '866bd337c819985dae51a248851b3dd3')"
     ]
   }
 }

composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/BroadcastNostrEventRequestDao.kt

@@ -13,7 +13,7 @@ interface BroadcastNostrEventRequestDao {
     fun  getAllBroadcastNostrEventRequests(): List<BroadcastNostrEventRequest>
 
     @Query("SELECT * FROM BroadcastNostrEventRequest WHERE status = :status")
-    fun observeBroadcastNostrEventRequestByStatus(status: String): Flow<LocalBroadcastNostrEventRequest?>
+    fun observeBroadcastNostrEventRequestsByStatus(status: String): Flow<List<LocalBroadcastNostrEventRequest>>
 
     @Query("SELECT * FROM BroadcastNostrEventRequest WHERE nostrEventId = :nostrEventId")
     fun observeBroadcastNostrEventRequestByNostrEventId(nostrEventId: String): Flow<BroadcastNostrEventRequest?>

composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt

@@ -5,6 +5,7 @@ import ac.aux.compose.database.model.BroadcastNostrEventRequest
 import ac.aux.compose.database.model.NostrEvent
 import ac.aux.compose.database.model.Post
 import ac.aux.compose.database.model.Profile
+import ac.aux.compose.database.model.UnsignedNostrEvent
 import androidx.room.Dao
 import androidx.room.Embedded
 import androidx.room.Relation
@@ -22,10 +23,17 @@ abstract class NostrDao(
 
     @Transaction
     open suspend fun publishNostrEvent(
+        unsignedNostrEvent: UnsignedNostrEvent,
         nostrEvent: NostrEvent,
         relayURLs: List<String>
     ) {
         logger.i("Publish Nostr Event: $nostrEvent ($relayURLs)")
+        // Update unsignedEvent with signedTime time...
+        database.unsignedNostrEventDao().upsert(
+            unsignedNostrEvent.copy(
+                signedAt = Clock.System.now()
+            )
+        )
         // Find or create profile with the pubKey... if not found TODO: submit a sync request...
         val profile = database.profileDao().getProfileByPublicKey(nostrEvent.pubKey)
 

composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/UnsignedNostrEventDao.kt

@@ -26,6 +26,6 @@ interface UnsignedNostrEventDao {
     @Query("SELECT * FROM UnsignedNostrEvent WHERE kind = 0 AND pubKey = :publicKey")
     fun observeProfile(publicKey: String): Flow<LocalProfile?>
 
-    @Query("SELECT * FROM UnsignedNostrEvent WHERE pubKey = :publicKey") // TODO: where nostrEvent.unsignedNostrEventId is 0
+    @Query("SELECT * FROM UnsignedNostrEvent WHERE pubKey = :publicKey AND signedAt IS NULL") // TODO: where nostrEvent.unsignedNostrEventId is 0
     fun observeUnsignedNostrEvents(publicKey: String): Flow<UnsignedNostrEvent?>
 }

composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/UnsignedNostrEvent.kt

@@ -37,6 +37,8 @@ data class UnsignedNostrEvent(
     val tags: TagArray,
     val content: String,
 
+    val signedAt: Instant? = null,
+
     override val createdAt: Instant = Clock.System.now(),
     override val updatedAt: Instant = Clock.System.now(),
     override val savedAt: Instant = Clock.System.now(),

composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt

@@ -32,8 +32,8 @@ class DatabaseNostrRepository(
         return database.unsignedNostrEventDao().observeUnsignedNostrEvents(publicKey)
     }
 
-    override suspend fun observePendingBroadcastNostrEventRequests(): Flow<LocalBroadcastNostrEventRequest?> {
+    override suspend fun observePendingBroadcastNostrEventRequests(): Flow<List<LocalBroadcastNostrEventRequest>> {
-        return database.broadcastNostrEventRequestDao().observeBroadcastNostrEventRequestByStatus("pending")
+        return database.broadcastNostrEventRequestDao().observeBroadcastNostrEventRequestsByStatus("pending")
     }
 
     override suspend fun createNewProfile(
@@ -82,8 +82,13 @@ class DatabaseNostrRepository(
         database.wipeData()
     }
 
-    override suspend fun publishNostrEvent(nostrEvent: NostrEvent, relayURLs: List<String>) {
+    override suspend fun publishNostrEvent(
+        unsignedNostrEvent: UnsignedNostrEvent,
+        nostrEvent: NostrEvent,
+        relayURLs: List<String>
+    ) {
         database.nostrDao().publishNostrEvent(
+            unsignedNostrEvent,
             nostrEvent,
             relayURLs
         )

composeApp/src/commonMain/kotlin/ac/aux/compose/network/KTorHttpWebSocket.kt

@@ -10,11 +10,12 @@ import io.ktor.client.plugins.websocket.DefaultClientWebSocketSession
 import io.ktor.client.plugins.websocket.webSocketSession
 import io.ktor.websocket.Frame
 import io.ktor.websocket.close
+import io.ktor.websocket.readReason
 import io.ktor.websocket.readText
-import io.ktor.websocket.send
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.IO
+import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 
@@ -32,45 +33,101 @@ class KTorHttpWebSocket(
     private var webSocketSession: DefaultClientWebSocketSession? = null
 
     override fun needsReconnect(): Boolean {
-        TODO("Not yet implemented")
+        logger.d("needsReconnect ${url.url}")
+        if (webSocketSession == null) return true
+
+        val activeHttpClient = httpClient ?: return true
+
+        val currentHttpClient = httpClientBuilder(url)
+
+        // TODO: Proxy...
+
+        // TODO: timeout checks
+
+        return false
     }
 
     override fun connect() {
-        runBlocking {
+        logger.d("connect ${url.url}")
+        runBlocking(
+            scope.coroutineContext
+        ) {
             httpClient = httpClientBuilder(url)
             webSocketSession = httpClient?.webSocketSession(
                 urlString = url.url
             ) {
-
+                logger.d("webSocketSession")
             }
 
             webSocketSession?.let { socketSession ->
 
-                val job = scope.launch {
+                logger.d("isOpen : ${socketSession.isActive}")
-                    for (frame in socketSession.incoming) {
+                out.onOpen(
-                        val frameText = frame as? Frame.Text
+                    pingMillis = 1,
+                    compression = false
+                )
 
-                        frameText?.readText()?.let { out.onMessage(it) }
+                val incomingFrameJob = scope.launch {
+                    try {
+                        for (frame in socketSession.incoming) {
+                            when (frame) {
+                                is Frame.Text -> {
+                                    frame.readText().let {
+                                        logger.d("WebSocket Read Text: $it")
+                                        out.onMessage(it)
+                                    }
+                                }
+                                is Frame.Close -> {
+                                    logger.d("Close $frame")
+                                    val reason = frame.readReason()
+                                    out.onClosed(
+                                        code = reason?.code?.toInt() ?: 0,
+                                        reason = reason?.knownReason.toString()
+                                    )
+                                }
+                                is Frame.Ping -> {
+                                    logger.d("Ping: $frame")
+                                }
+                                is Frame.Pong -> {
+                                    logger.d("Pong: $frame")
+                                }
+                                is Frame.Binary -> {
+                                    logger.d("Binary Message: $frame")
+                                }
+                                else -> {
+                                    logger.d("Unsupported frame: $frame")
+                                }
+                            }
+                        }
+                    } catch (e: Throwable) {
+                        logger.e("Websocket error: ", e)
+                    } finally {
+                        logger.d("Closed for real... hopefully we know the reason")
                     }
                 }
 
-                job.join()
+                incomingFrameJob.join()
-
+                logger.d("webSocketSession closed")
             }
-
         }
 
     }
 
     override fun disconnect() {
-        runBlocking {
+        logger.d("disconnect ${url.url}")
+        runBlocking(
+            scope.coroutineContext
+        ) {
             webSocketSession?.close()
+
         }
     }
 
     override fun send(msg: String): Boolean = try {
-        runBlocking {
+        logger.d("Send ${url.url}: $msg")
-
+        runBlocking(
+            scope.coroutineContext
+        ) {
             webSocketSession?.send(
                 Frame.Text(msg)
             )
@@ -83,16 +140,22 @@ class KTorHttpWebSocket(
     }
 
     class Builder(
-        val httpClientBuilder: (NormalizedRelayUrl) -> HttpClient
+        ktorWebsocketListener: KtorWebsocketListener = KtorWebsocketListener(),
+        val httpClientBuilder: (NormalizedRelayUrl) -> HttpClient,
     ): WebsocketBuilder {
+        val logger = Logger.withTag("KtorWebsocketBuilder")
+
         override fun build(
             url: NormalizedRelayUrl,
             out: WebSocketListener
-        ): WebSocket = KTorHttpWebSocket(
+        ): WebSocket {
-            url = url,
+            logger.d("Build KTorHttpWebSocket: ${url.url}")
-            httpClientBuilder = httpClientBuilder,
-            out = out
-        )
-    }
 
+            return KTorHttpWebSocket(
+                url = url,
+                httpClientBuilder = httpClientBuilder,
+                out = out
+            )
+        }
+    }
 }

composeApp/src/commonMain/kotlin/ac/aux/compose/network/KtorWebsocketListener.kt

@@ -0,0 +1,30 @@
+package ac.aux.compose.network
+
+import co.touchlab.kermit.Logger
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
+
+class KtorWebsocketListener: WebSocketListener {
+    val logger = Logger.withTag("KtorWebsocketListener")
+
+    val out: WebSocketListener? = null
+
+    override fun onOpen(pingMillis: Int, compression: Boolean) {
+        logger.d("onOpen: $pingMillis, $compression")
+        out?.onOpen(pingMillis, compression)
+    }
+
+    override fun onMessage(text: String) {
+        logger.d("onMessage: $text")
+        out?.onMessage(text)
+    }
+
+    override fun onClosed(code: Int, reason: String) {
+        logger.d("onClosed: $code, $reason")
+        out?.onClosed(code, reason)
+    }
+
+    override fun onFailure(t: Throwable, code: Int?, response: String?) {
+        logger.e("onFailure: $code, $response", t)
+        out?.onFailure(t, code, response)
+    }
+}

composeApp/src/commonMain/kotlin/ac/aux/compose/nostr/Relays.kt

@@ -32,4 +32,5 @@ object Relays {
 
     val bootstrapInboxRelaySet = setOf(damus, primal, mom, nos, bitcoiner, oxtr, yabu)
     val eventFinderRelaySet = setOf(wine, damus, primal, mom, nos, bitcoiner, oxtr)
+    val eventPublishRelaySet = setOf(damus, primal, mom, nos, bitcoiner, oxtr)
 }

composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt

@@ -14,7 +14,7 @@ interface NostrRepository {
 
     suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow<UnsignedNostrEvent?>
 
-    suspend fun observePendingBroadcastNostrEventRequests(): Flow<LocalBroadcastNostrEventRequest?>
+    suspend fun observePendingBroadcastNostrEventRequests(): Flow<List<LocalBroadcastNostrEventRequest>>
 
     suspend fun createNewProfile(
         publicKey: HexKey,
@@ -31,6 +31,7 @@ interface NostrRepository {
     suspend fun wipeDatabase()
 
     suspend fun publishNostrEvent(
+        unsignedNostrEvent: UnsignedNostrEvent,
         nostrEvent: NostrEvent,
         relayURLs: List<String> = emptyList()
     )
@@ -47,7 +48,7 @@ interface NostrRepository {
                 TODO("Not yet implemented")
             }
 
-            override suspend fun observePendingBroadcastNostrEventRequests(): Flow<LocalBroadcastNostrEventRequest?> {
+            override suspend fun observePendingBroadcastNostrEventRequests(): Flow<List<LocalBroadcastNostrEventRequest>> {
                 TODO("Not yet implemented")
             }
 
@@ -67,7 +68,11 @@ interface NostrRepository {
             override suspend fun wipeDatabase() {
             }
 
-            override suspend fun publishNostrEvent(nostrEvent: NostrEvent, relayURLs: List<String>) {
+            override suspend fun publishNostrEvent(
+                unsignedNostrEvent: UnsignedNostrEvent,
+                nostrEvent: NostrEvent,
+                relayURLs: List<String>
+            ) {
 
             }
 

composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt

@@ -16,6 +16,7 @@ import com.vitorpamplona.quartz.nip01Core.core.toHexKey
 import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
 import io.ktor.client.HttpClient
 import io.ktor.client.plugins.websocket.WebSockets
@@ -63,10 +64,13 @@ class NavigationViewModel(
         }
     }
 
-    val websocketBuilder = KTorHttpWebSocket.Builder { url ->
+
-        // TODO: Figure out if we need a tor client
+    val websocketBuilder: WebsocketBuilder = KTorHttpWebSocket.Builder(
-        httpClient
+        httpClientBuilder = { url ->
-    }
+            // TODO: Figure out if we need a tor client
+            httpClient
+        }
+    )
     val nostrClient: INostrClient = NostrClient(
         websocketBuilder = websocketBuilder,
         scope = scope
@@ -95,6 +99,7 @@ class NavigationViewModel(
                 publicKey = SeedManager.activePublicKey().toHexKey()
             ).collect { unsignedNostrEventOrNull ->
                 unsignedNostrEventOrNull?.let { unsignedNostrEvent ->
+                    logger.d("Unsigned")
                     val event = tempSigner.signNormal<Event>(
                         createdAt = unsignedNostrEvent.createdAt.toEpochMilliseconds(),
                         kind = unsignedNostrEvent.kind,
@@ -104,6 +109,7 @@ class NavigationViewModel(
 
 
                     nostrRepository.publishNostrEvent(
+                        unsignedNostrEvent,
                         NostrEvent(
                             id = event.id,
                             pubKey = event.pubKey,
@@ -114,7 +120,7 @@ class NavigationViewModel(
                             sig = event.sig,
                             unsignedNostrEventId = unsignedNostrEvent.id
                         ),
-                        relayURLs = Relays.eventFinderRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
+                        relayURLs = Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
                     )
                 }
 
@@ -127,8 +133,8 @@ class NavigationViewModel(
 
         scope.launch(Dispatchers.IO) {
 
-            nostrRepository.observePendingBroadcastNostrEventRequests().collect {
+            nostrRepository.observePendingBroadcastNostrEventRequests().collect { localBroadcastNostrEventRequests ->
-                it?.let { localBroadcastNostrEventRequest ->
+                localBroadcastNostrEventRequests.forEach { localBroadcastNostrEventRequest ->
                     val event: Event = localBroadcastNostrEventRequest.nostrEvent.let { nostrEvent ->
                         Event(
                             id = nostrEvent.id,
@@ -140,18 +146,21 @@ class NavigationViewModel(
                             createdAt = nostrEvent.createdAt.toEpochMilliseconds()
                         )
                     }
-                    logger.d("Broadcasting: ${event.toJson()}")
+                    logger.d("Broadcasting (${localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL}): ${event.toJson()}")
-                    nostrClient.send(
-                        event = event,
-                        relayList = setOf(
-                            RelayUrlNormalizer.normalize(localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL)
-                        )
-                    )
-                    logger.d("Update with result")
 
-                    nostrRepository.broadcastProcessed(
+                    scope.launch {
-                        localBroadcastNostrEventRequest.broadcastNostrEventRequest
+
-                    )
+                        nostrClient.send(
+                            event = event,
+                            relayList = setOf(
+                                RelayUrlNormalizer.normalize(localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL)
+                            )
+                        )
+
+//                    nostrRepository.broadcastProcessed(
+//                        localBroadcastNostrEventRequest.broadcastNostrEventRequest
+//                    )
+                    }
                 }
 
             }