diff --git a/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json b/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json index 4ea050c1..73f6a3e9 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": "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')" ] } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/BroadcastNostrEventRequestDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/BroadcastNostrEventRequestDao.kt index ae824737..c2c197a5 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/BroadcastNostrEventRequestDao.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/BroadcastNostrEventRequestDao.kt @@ -13,7 +13,7 @@ interface BroadcastNostrEventRequestDao { fun getAllBroadcastNostrEventRequests(): List @Query("SELECT * FROM BroadcastNostrEventRequest WHERE status = :status") - fun observeBroadcastNostrEventRequestByStatus(status: String): Flow + fun observeBroadcastNostrEventRequestsByStatus(status: String): Flow> @Query("SELECT * FROM BroadcastNostrEventRequest WHERE nostrEventId = :nostrEventId") fun observeBroadcastNostrEventRequestByNostrEventId(nostrEventId: String): Flow 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 be40fc81..33f195f4 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 @@ -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 ) { 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) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/UnsignedNostrEventDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/UnsignedNostrEventDao.kt index 95947dd1..646febb6 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/UnsignedNostrEventDao.kt +++ b/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 - @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 } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/UnsignedNostrEvent.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/UnsignedNostrEvent.kt index 9da55fca..2cd5f0ff 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/UnsignedNostrEvent.kt +++ b/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(), diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt index 2815335c..4c8d7e3a 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt +++ b/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 { - return database.broadcastNostrEventRequestDao().observeBroadcastNostrEventRequestByStatus("pending") + override suspend fun observePendingBroadcastNostrEventRequests(): Flow> { + 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) { + override suspend fun publishNostrEvent( + unsignedNostrEvent: UnsignedNostrEvent, + nostrEvent: NostrEvent, + relayURLs: List + ) { database.nostrDao().publishNostrEvent( + unsignedNostrEvent, nostrEvent, relayURLs ) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/KTorHttpWebSocket.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/KTorHttpWebSocket.kt index 4535b883..e4376821 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/KTorHttpWebSocket.kt +++ b/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 { - for (frame in socketSession.incoming) { - val frameText = frame as? Frame.Text + logger.d("isOpen : ${socketSession.isActive}") + out.onOpen( + 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( - url = url, - httpClientBuilder = httpClientBuilder, - out = out - ) - } + ): WebSocket { + logger.d("Build KTorHttpWebSocket: ${url.url}") + return KTorHttpWebSocket( + url = url, + httpClientBuilder = httpClientBuilder, + out = out + ) + } + } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/KtorWebsocketListener.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/KtorWebsocketListener.kt new file mode 100644 index 00000000..2e87b49c --- /dev/null +++ b/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) + } +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/nostr/Relays.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/nostr/Relays.kt index f406001f..81e7fb8d 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/nostr/Relays.kt +++ b/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) } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt index 43e2da79..ce8c8b11 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt @@ -14,7 +14,7 @@ interface NostrRepository { suspend fun observeUnsignedNostrEvents(publicKey: HexKey): Flow - suspend fun observePendingBroadcastNostrEventRequests(): Flow + suspend fun observePendingBroadcastNostrEventRequests(): Flow> suspend fun createNewProfile( publicKey: HexKey, @@ -31,6 +31,7 @@ interface NostrRepository { suspend fun wipeDatabase() suspend fun publishNostrEvent( + unsignedNostrEvent: UnsignedNostrEvent, nostrEvent: NostrEvent, relayURLs: List = emptyList() ) @@ -47,7 +48,7 @@ interface NostrRepository { TODO("Not yet implemented") } - override suspend fun observePendingBroadcastNostrEventRequests(): Flow { + override suspend fun observePendingBroadcastNostrEventRequests(): Flow> { TODO("Not yet implemented") } @@ -67,7 +68,11 @@ interface NostrRepository { override suspend fun wipeDatabase() { } - override suspend fun publishNostrEvent(nostrEvent: NostrEvent, relayURLs: List) { + override suspend fun publishNostrEvent( + unsignedNostrEvent: UnsignedNostrEvent, + nostrEvent: NostrEvent, + relayURLs: List + ) { } diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt index 422ebe1e..88cd31eb 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt +++ b/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 - httpClient - } + + val websocketBuilder: WebsocketBuilder = KTorHttpWebSocket.Builder( + 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( 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 { - it?.let { localBroadcastNostrEventRequest -> + nostrRepository.observePendingBroadcastNostrEventRequests().collect { localBroadcastNostrEventRequests -> + 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()}") - nostrClient.send( - event = event, - relayList = setOf( - RelayUrlNormalizer.normalize(localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL) - ) - ) - logger.d("Update with result") + logger.d("Broadcasting (${localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL}): ${event.toJson()}") - nostrRepository.broadcastProcessed( - localBroadcastNostrEventRequest.broadcastNostrEventRequest - ) + scope.launch { + + nostrClient.send( + event = event, + relayList = setOf( + RelayUrlNormalizer.normalize(localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL) + ) + ) + +// nostrRepository.broadcastProcessed( +// localBroadcastNostrEventRequest.broadcastNostrEventRequest +// ) + } } }