database, REQ, EVENT, CLOSE, EOSE

app/build.gradle.kts

@@ -1,6 +1,7 @@
 plugins {
     id("com.android.application")
     id("org.jetbrains.kotlin.android")
+    id("com.google.devtools.ksp")
 }
 
 android {
@@ -75,5 +76,12 @@ dependencies {
     implementation("io.ktor:ktor-server-cio:2.3.8")
     implementation("io.ktor:ktor-server-websockets:2.3.8")
     implementation("io.ktor:ktor-websockets:2.3.8")
-
+    implementation("com.github.vitorpamplona.amethyst:quartz:v0.83.9") {
+        exclude("net.java.dev.jna")
+    }
+    implementation("net.java.dev.jna:jna:5.14.0@aar")
+    val roomVersion = "2.6.1"
+    implementation("androidx.room:room-runtime:$roomVersion")
+    annotationProcessor("androidx.room:room-compiler:$roomVersion")
+    ksp("androidx.room:room-compiler:$roomVersion")
 }

app/src/main/java/com/greenart7c3/citrine/CommandResult.kt

@@ -0,0 +1,19 @@
+package com.greenart7c3.citrine
+
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.vitorpamplona.quartz.events.Event
+
+data class CommandResult(val eventId: String, val result: Boolean, val description: String = "") {
+    fun toJson(): String {
+        return jacksonObjectMapper().writeValueAsString(
+            listOf("OK", eventId, result, description)
+        )
+    }
+
+    companion object {
+        fun ok(event: Event) = CommandResult(event.id, true)
+        fun duplicated(event: Event) = CommandResult(event.id, true, "duplicate:")
+        fun invalid(event: Event, message: String) = CommandResult(event.id, false, "invalid: $message")
+        fun error(event: Event, message: String) = CommandResult(event.id, false, "error: $message")
+    }
+}

app/src/main/java/com/greenart7c3/citrine/CustomWebSocketServer.kt

@@ -1,23 +1,47 @@
 package com.greenart7c3.citrine
 
+import EOSE
+import android.content.Context
+import android.util.Log
+import com.fasterxml.jackson.databind.DeserializationFeature
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.greenart7c3.citrine.database.AppDatabase
+import com.greenart7c3.citrine.database.toEvent
+import com.greenart7c3.citrine.database.toEventWithTags
+import com.vitorpamplona.quartz.events.Event
 import io.ktor.http.ContentType
 import io.ktor.server.application.call
 import io.ktor.server.application.install
 import io.ktor.server.cio.CIO
 import io.ktor.server.engine.ApplicationEngine
 import io.ktor.server.engine.embeddedServer
+import io.ktor.server.response.respond
 import io.ktor.server.response.respondText
 import io.ktor.server.routing.get
 import io.ktor.server.routing.routing
+import io.ktor.server.websocket.DefaultWebSocketServerSession
 import io.ktor.server.websocket.WebSockets
+import io.ktor.server.websocket.pingPeriod
+import io.ktor.server.websocket.timeout
 import io.ktor.server.websocket.webSocket
+import io.ktor.websocket.CloseReason
 import io.ktor.websocket.Frame
+import io.ktor.websocket.WebSocketDeflateExtension
+import io.ktor.websocket.close
 import io.ktor.websocket.readText
+import io.ktor.websocket.send
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import java.time.Duration
+import java.util.zip.Deflater
 
 
-class CustomWebSocketServer(private val port: Int) {
+class CustomWebSocketServer(private val port: Int, private val context: Context) {
     private lateinit var server: ApplicationEngine
+    private val objectMapper = jacksonObjectMapper()
+        .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
 
     fun port(): Int {
         return server.environment.connectors.first().port
@@ -31,9 +55,77 @@ class CustomWebSocketServer(private val port: Int) {
         server.stop(1000)
     }
 
+    private suspend fun subscribe(subscriptionId: String, filterNodes: List<JsonNode>, session: DefaultWebSocketServerSession) {
+        val filters = filterNodes.map { jsonNode ->
+            val tags = jsonNode.fields().asSequence()
+                .filter { it.key.startsWith("#") }
+                .map { it.key.substringAfter("#") to it.value.map { item -> item.asText() }.toSet() }
+                .toMap()
+
+            val filter = objectMapper.treeToValue(jsonNode, EventFilter::class.java)
+
+            filter.copy(tags = tags)
+        }.toSet()
+
+        for (filter in filters) {
+           runBlocking {
+               EventSubscription.subscribe(subscriptionId, filter, session, context, objectMapper, true)
+           }
+        }
+
+        session.send(EOSE(subscriptionId).toJson())
+    }
+
+
+    private suspend fun processNewRelayMessage(newMessage: String, session: DefaultWebSocketServerSession) {
+        val msgArray = Event.mapper.readTree(newMessage)
+        when (val type = msgArray.get(0).asText()) {
+            "REQ" -> {
+                val subscriptionId = msgArray.get(1).asText()
+                subscribe(subscriptionId, msgArray.drop(2), session)
+            }
+            "EVENT" -> {
+                processEvent(msgArray.get(1), session)
+            }
+            "CLOSE" -> {
+                session.close(CloseReason(CloseReason.Codes.NORMAL, newMessage))
+            }
+            "PING" -> {
+                session.send(NoticeResult("PONG").toJson())
+            }
+            else -> {
+                val errorMessage = NoticeResult.invalid("unknown message type $type").toJson()
+                Log.d("message", errorMessage)
+                session.send(errorMessage)
+            }
+        }
+    }
+
+    private suspend fun processEvent(eventNode: JsonNode, session: DefaultWebSocketServerSession) {
+        val event = objectMapper.treeToValue(eventNode, Event::class.java)
+
+        AppDatabase.getDatabase(context).eventDao().insertEventWithTags(event.toEventWithTags())
+
+        session.send(CommandResult.ok(event).toJson())
+    }
+
     private fun startKtorHttpServer(port: Int): ApplicationEngine {
         return embeddedServer(CIO, port = port) {
-            install(WebSockets)
+            install(WebSockets) {
+                extensions {
+                    install(WebSocketDeflateExtension) {
+                        /**
+                         * Compression level to use for [java.util.zip.Deflater].
+                         */
+                        compressionLevel = Deflater.DEFAULT_COMPRESSION
+
+                        /**
+                         * Prevent compressing small outgoing frames.
+                         */
+                        compressIfBiggerThan(bytes = 4 * 1024)
+                    }
+                }
+            }
 
             routing {
                 // Handle HTTP GET requests
@@ -54,7 +146,6 @@ class CustomWebSocketServer(private val port: Int) {
                     } else {
                         call.respondText("Use a Nostr client or Websocket client to connect", ContentType.Text.Html)
                     }
-
                 }
 
                 // WebSocket endpoint
@@ -63,13 +154,16 @@ class CustomWebSocketServer(private val port: Int) {
                         for (frame in incoming) {
                             when (frame) {
                                 is Frame.Text -> {
-                                    println("Received WebSocket message: ${frame.readText()}")
+                                    val message = frame.readText()
+                                    processNewRelayMessage(message, this)
+                                }
+                                else -> {
+                                    Log.d("error", frame.toString())
                                 }
-                                else -> {}
                             }
                         }
                     } catch (e: ClosedReceiveChannelException) {
-                        // Channel closed
+                        Log.d("error", e.toString())
                     }
                 }
             }

app/src/main/java/com/greenart7c3/citrine/EOSE.kt

@@ -0,0 +1,5 @@
+data class EOSE(val subscriptionId: String) {
+    fun toJson(): String {
+        return """["EOSE","$subscriptionId"]"""
+    }
+}

app/src/main/java/com/greenart7c3/citrine/EventFilter.kt

@@ -0,0 +1,98 @@
+package com.greenart7c3.citrine
+
+/**
+ * Copyright (c) 2023 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+import com.vitorpamplona.quartz.events.Event
+import java.util.function.Predicate
+
+data class EventFilter(
+    val ids: Set<String> = emptySet(),
+    val authors: Set<String> = emptySet(),
+    val kinds: Set<Int> = emptySet(),
+    val tags: Map<String, Set<String>> = emptyMap(),
+    val since: Int? = null,
+    val until: Int? = null,
+    val limit: Int = 10_000,
+    private val search: String? = null
+) : Predicate<Event> {
+
+    val searchKeywords: Set<String> = search?.let { tokenizeString(search) } ?: emptySet()
+
+    override fun test(event: Event): Boolean {
+        if (since != null && event.createdAt < since) {
+            return false
+        }
+
+        if (until != null && event.createdAt > until) {
+            return false
+        }
+
+        if (ids.isNotEmpty() && ids.none { event.id.startsWith(it) }) {
+            return false
+        }
+
+        if (authors.isNotEmpty() && authors.none { event.pubKey.startsWith(it) }) {
+            return false
+        }
+
+        if (kinds.isNotEmpty() && event.kind !in kinds) {
+            return false
+        }
+
+        if (tags.isNotEmpty() && tags.none { testTag(it, event) }) {
+            return false
+        }
+
+        if (!search.isNullOrBlank() && !testSearch(search, event)) {
+            return false
+        }
+
+        return true
+    }
+
+    private fun testTag(tag: Map.Entry<String, Set<String>>, event: Event): Boolean {
+        val eventTags: Set<String> = event.tags.asSequence()
+            .filter { it.size > 1 && it[0] == tag.key }
+            .map { it[1] }
+            .toSet()
+
+        return tag.value.any { it in eventTags }
+    }
+
+    private fun testSearch(search: String, event: Event): Boolean {
+        val tokens = tokenizeString(search)
+        val eventTokens = tokenizeString(event.content)
+
+        return tokens.all { it in eventTokens }
+    }
+
+    private fun tokenizeString(string: String): Set<String> {
+        return string.split(TOKENIZE_REGEX)
+            .filter { it.isNotEmpty() }
+            .map { it.lowercase() }
+            .toSet()
+    }
+
+    companion object {
+        val TOKENIZE_REGEX = "[^a-zA-Z0-9]".toRegex()
+    }
+}