diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index e0cd7f815e..16312e4d95 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -113,7 +113,7 @@ fun main(args: Array) { cliInfoFile?.let { RelayInfo.fromFile(it) } ?: config.resolveInfo() - val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl) + val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl, indexStrategy = RelayIndexingStrategy) val policyBuilder: () -> IRelayPolicy = { composePolicy(config, advertisedUrl, requireAuth, optionalAuth, verifySigs, parallelVerify) diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt index 84d15a56aa..8e79606e44 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayEngine.kt @@ -58,7 +58,7 @@ import kotlin.coroutines.CoroutineContext */ class RelayEngine( val url: NormalizedRelayUrl, - val store: IEventStore = EventStore(dbName = null, relay = url), + val store: IEventStore = EventStore(dbName = null, relay = url, indexStrategy = RelayIndexingStrategy), /** * Runtime configuration handle — owns the persistence path (when * any), the NIP-11 doc seed, and the seed for the NIP-86 ban / diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt new file mode 100644 index 0000000000..259bef25bc --- /dev/null +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/RelayIndexingStrategy.kt @@ -0,0 +1,42 @@ +/* + * Copyright (c) 2025 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. + */ +package com.vitorpamplona.geode + +import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy + +/** + * Index strategy for a *public relay*, as opposed to quartz's + * [DefaultIndexingStrategy] which is tuned for client-side stores that + * only ever query their own kinds. + * + * A relay cannot predict its clients' filters, and the cheapest query a + * client can send — `{"limit": N}`, the firehose/landing REQ — carries no + * kind, author or tag for the planner to use. Without an index on + * `created_at` alone that REQ is a full-table scan + top-N sort on every + * poll; relayBench measured it 4× slower than strfry at 10k events and + * the gap grows linearly with the table. The extra index costs one B-tree + * insert per event, which the same benchmark shows is noise next to the + * Schnorr verify + tag indexing already paid on the write path. + */ +val RelayIndexingStrategy = + DefaultIndexingStrategy( + indexEventsByCreatedAtAlone = true, + ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt index 0e2a224c9c..05cec0fad3 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/RelaySession.kt @@ -40,6 +40,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.backend.SessionBackend import com.vitorpamplona.quartz.nip01Core.relay.server.policies.IRelayPolicy import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PolicyResult import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.RawEvent import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegMsgCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd @@ -125,6 +126,15 @@ class RelaySession( } } + /** [send] for frames that are already wire-format JSON (the raw REQ path). */ + private fun sendRaw(json: String) { + try { + onSend(json) + } catch (e: Exception) { + Log.w("ClientSession") { "Failed to send to ${e.message}" } + } + } + override fun close() { cancelAllSubscriptions() onClose(this) @@ -281,16 +291,47 @@ class RelaySession( val job = scope.launch { try { - store.query( - ctx = requestContext, - filters = filters, - onEach = { event -> - if (policy.canSendToSession(event)) { - send(EventMessage(cmd.subId, event)) + if (policy.filtersOutgoingEvents) { + // Screened path: every event is materialized so the + // policy can veto it per session. + store.query( + ctx = requestContext, + filters = filters, + onEach = { event -> + if (policy.canSendToSession(event)) { + send(EventMessage(cmd.subId, event)) + } + }, + onEose = { send(EoseMessage(cmd.subId)) }, + ) + } else { + // Zero-decode path: the stored replay splices raw + // storage strings straight into wire frames — no tags + // parse, no Event materialization, no re-serialize. + // The `["EVENT","",` prefix is built once per + // subscription, not per row. + val framePrefix = + buildString { + append("[\"EVENT\",") + RawEvent.appendJsonQuoted(this, cmd.subId) + append(',') } - }, - onEose = { send(EoseMessage(cmd.subId)) }, - ) + store.queryRaw( + ctx = requestContext, + filters = filters, + onEachStored = { raw -> + sendRaw( + buildString(framePrefix.length + raw.jsonTags.length + raw.content.length + 256) { + append(framePrefix) + raw.appendJsonObjectTo(this) + append(']') + }, + ) + }, + onEachLive = { event -> send(EventMessage(cmd.subId, event)) }, + onEose = { send(EoseMessage(cmd.subId)) }, + ) + } } catch (e: CancellationException) { // Subscription was closed – this is expected. throw e diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt index 83f3947800..199d024a17 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt @@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterIndex import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip01Core.store.RawEvent import com.vitorpamplona.quartz.nip50Search.strippingSearchExtensions import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.awaitCancellation @@ -205,6 +206,51 @@ class LiveEventStore( } } + /** + * Zero-decode variant of [query]: the historical replay streams + * [com.vitorpamplona.quartz.nip01Core.store.RawEvent] rows straight + * from storage (no tags parse, no Event materialization, no + * re-serialization), while the live path after EOSE is identical to + * [query]'s. Same registration-before-replay ordering and the same + * immutable-set dedupe against events accepted mid-replay. + */ + override suspend fun queryRaw( + ctx: RequestContext, + filters: List, + onEachStored: (RawEvent) -> Unit, + onEachLive: (Event) -> Unit, + onEose: () -> Unit, + ) { + val seenIds = AtomicReference?>(emptySet()) + + val sub = + LiveSubscription( + filters = filters, + deliver = { event -> + val seen = seenIds.load() + if (seen != null && seen.contains(event.id)) return@LiveSubscription + onEachLive(event) + }, + ) + + index.register(filters, sub) + try { + store.rawQuery(filters) { raw -> + while (true) { + val current = seenIds.load() ?: break + if (raw.id in current) break + if (seenIds.compareAndSet(current, current + raw.id)) break + } + onEachStored(raw) + } + onEose() + seenIds.store(null) + awaitCancellation() + } finally { + index.unregister(sub) + } + } + override suspend fun count( ctx: RequestContext, filters: List, diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt index b57dcfca09..8461e1fda7 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/SessionBackend.kt @@ -25,6 +25,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.CountResult import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip01Core.store.RawEvent /** * The data plane a [RelaySession] talks to: how REQ/COUNT are answered, how @@ -58,6 +59,27 @@ interface SessionBackend { onEose: () -> Unit, ) + /** + * [query] variant for sessions whose policy does not filter outgoing + * events (see + * [com.vitorpamplona.quartz.nip01Core.relay.server.policies.IRelayPolicy.filtersOutgoingEvents]): + * the stored replay is delivered as [RawEvent]s so the transport can + * splice storage strings straight into wire frames without + * materializing an [Event] per row. Live events after EOSE still + * arrive as [Event]s via [onEachLive] — they exist in object form + * already, and live matching needs them. + * + * The default falls back to [query], treating every delivery as live — + * correct for any backend, just without the zero-decode win. + */ + suspend fun queryRaw( + ctx: RequestContext, + filters: List, + onEachStored: (RawEvent) -> Unit, + onEachLive: (Event) -> Unit, + onEose: () -> Unit, + ): Unit = query(ctx, filters, onEachLive, onEose) + /** Answers a NIP-45 COUNT with an exact cardinality for the caller in [ctx]. */ suspend fun count( ctx: RequestContext, diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/IRelayPolicy.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/IRelayPolicy.kt index a6dac44927..978512fee7 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/IRelayPolicy.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/IRelayPolicy.kt @@ -133,10 +133,24 @@ interface IRelayPolicy { * Called for each event that matches a subscription's filters. Return * true to deliver the event, false to suppress it for this session. * + * Any implementation that can return `false` here MUST also override + * [filtersOutgoingEvents] to return `true`, otherwise the session may + * take the zero-decode REQ fast path that never consults this method. + * * @param event The event about to be sent. */ fun canSendToSession(event: Event): Boolean = true + /** + * Declares whether [canSendToSession] can ever suppress an event for + * this session. When `false` (the default, matching the default + * [canSendToSession] that always allows), the engine answers REQs + * through the zero-decode raw path — storage strings are spliced + * straight into wire frames and [canSendToSession] is never called. + * When `true`, every delivered event is materialized and screened. + */ + val filtersOutgoingEvents: Boolean get() = false + operator fun plus(other: IRelayPolicy): IRelayPolicy = PolicyStack(this, other) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/PolicyStack.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/PolicyStack.kt index 00f6ef6009..c1062b874a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/PolicyStack.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/policies/PolicyStack.kt @@ -77,4 +77,6 @@ class PolicyStack( } override fun canSendToSession(event: Event): Boolean = policies.all { it.canSendToSession(event) } + + override val filtersOutgoingEvents: Boolean get() = policies.any { it.filtersOutgoingEvents } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt index 7d68d4fb93..50a72ebcd0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt @@ -101,6 +101,21 @@ interface IEventStore : AutoCloseable { onEach: (T) -> Unit, ) + /** + * Streams matching events in storage form — tags still serialized, + * nothing materialized — for read paths that only put events back on + * the wire (see [RawEvent]). The default decodes and re-wraps so + * every store stays correct; SQLite overrides with a true zero-decode + * row read. + */ + suspend fun rawQuery( + filters: List, + onEach: (RawEvent) -> Unit, + ): Unit = + query(filters) { event -> + onEach(RawEvent.fromEvent(event)) + } + suspend fun count(filter: Filter): Int suspend fun count(filters: List): Int diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/RawEvent.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/RawEvent.kt new file mode 100644 index 0000000000..1276f09892 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/RawEvent.kt @@ -0,0 +1,129 @@ +/* + * Copyright (c) 2025 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. + */ +package com.vitorpamplona.quartz.nip01Core.store + +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.utils.EventFactory + +/** + * An event as it sits in storage: fields as plain strings, `tags` still in + * its serialized-JSON form. This is the zero-decode currency of the relay + * read path — a REQ that only needs to put the event back on the wire can + * splice these strings straight into the outgoing frame via + * [appendJsonObjectTo], skipping the tags parse, the [EventFactory] + * dispatch to a kind-specific class, and the full re-serialization that + * materializing an [Event] would cost per row. + */ +class RawEvent( + val id: HexKey, + val pubKey: HexKey, + val createdAt: Long, + val kind: Kind, + val jsonTags: String, + val content: String, + val sig: HexKey, +) { + fun toEvent() = + EventFactory.create( + id, + pubKey, + createdAt, + kind, + OptimizedJsonMapper.fromJsonToTagArray(jsonTags), + content, + sig, + ) + + /** + * Appends this event as a NIP-01 JSON object, reusing the stored + * strings verbatim: `id`/`pubkey`/`sig` are validated hex, [jsonTags] + * is already JSON; only [content] needs string escaping. + */ + fun appendJsonObjectTo(builder: StringBuilder) { + builder + .append("{\"id\":\"") + .append(id) + .append("\",\"pubkey\":\"") + .append(pubKey) + .append("\",\"created_at\":") + .append(createdAt) + .append(",\"kind\":") + .append(kind) + .append(",\"tags\":") + .append(jsonTags) + .append(",\"content\":") + appendJsonQuoted(builder, content) + builder + .append(",\"sig\":\"") + .append(sig) + .append("\"}") + } + + companion object { + fun fromEvent(event: Event) = + RawEvent( + id = event.id, + pubKey = event.pubKey, + createdAt = event.createdAt, + kind = event.kind, + jsonTags = OptimizedJsonMapper.toJson(event.tags), + content = event.content, + sig = event.sig, + ) + + /** + * Appends [value] as a JSON string literal, quotes included. + * Minimal spec-compliant escaping (RFC 8259 §7): `"`, `\`, the + * short control escapes, and `\u00XX` for the rest of C0. + * Everything else — including non-ASCII — passes through raw, + * which is valid JSON and preserves the exact code points the + * event id was hashed over. + */ + fun appendJsonQuoted( + builder: StringBuilder, + value: String, + ) { + builder.append('"') + for (ch in value) { + when { + ch == '"' -> builder.append("\\\"") + ch == '\\' -> builder.append("\\\\") + ch == '\n' -> builder.append("\\n") + ch == '\r' -> builder.append("\\r") + ch == '\t' -> builder.append("\\t") + ch == '\b' -> builder.append("\\b") + ch == '\u000C' -> builder.append("\\f") + ch < ' ' -> { + builder.append("\\u") + val hex = ch.code.toString(16) + repeat(4 - hex.length) { builder.append('0') } + builder.append(hex) + } + else -> builder.append(ch) + } + } + builder.append('"') + } + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt index 7e1c912d55..7f63d87aab 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.store.FtsReindexProgress import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip01Core.store.RawEvent /** * SQLite-backed [IEventStore] with default DB-file name and relay @@ -61,6 +62,11 @@ class EventStore( onEach: (T) -> Unit, ) = store.query(filters, onEach) + override suspend fun rawQuery( + filters: List, + onEach: (RawEvent) -> Unit, + ) = store.rawQuery(filters, onEach) + override suspend fun count(filter: Filter) = store.count(filter) override suspend fun count(filters: List) = store.count(filters) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt index 3be4353679..2e42a2cf39 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt @@ -29,6 +29,7 @@ import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.core.isAddressable import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip01Core.store.RawEvent import com.vitorpamplona.quartz.nip01Core.store.sqlite.sql.where import com.vitorpamplona.quartz.utils.EventFactory diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteConnectionPool.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteConnectionPool.kt index 5ca19d9857..ea2fe43db7 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteConnectionPool.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteConnectionPool.kt @@ -82,14 +82,21 @@ class SQLiteConnectionPool( private val readerChannel: Channel? init { - writer = openConnection() + // The writer replays the same INSERT statements for every event — + // cache its prepared statements so sqlite3_prepare is paid once per + // SQL string instead of once per row. (Migrations run through the + // same wrapper; DDL statements just cache and stay unused.) + writer = StatementCachingConnection(openConnection()) onMigrate(writer) if (isInMemory) { readers = emptyList() readerChannel = null } else { - readers = List(numReaders) { openConnection() } + // Readers replay the same filter shapes all day (feeds, threads, + // profile hydrations) — caching their prepared statements pays + // the parse/plan cost once per shape per connection. + readers = List(numReaders) { StatementCachingConnection(openConnection()) } readerChannel = Channel(numReaders) readers.forEach { readerChannel.trySend(it) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt index bbbcead7b9..61d7dfc13f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt @@ -26,16 +26,14 @@ import androidx.sqlite.SQLiteException import androidx.sqlite.driver.bundled.BundledSQLiteDriver 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.isEphemeral import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.store.FtsReindexProgress import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.IdAndTime +import com.vitorpamplona.quartz.nip01Core.store.RawEvent import com.vitorpamplona.quartz.nip40Expiration.isExpired -import com.vitorpamplona.quartz.utils.EventFactory class SQLiteEventStore( val driver: SQLiteDriver = BundledSQLiteDriver(), @@ -372,24 +370,3 @@ class SQLiteEventStore( fun close() = pool.close() } - -class RawEvent( - val id: HexKey, - val pubKey: HexKey, - val createdAt: Long, - val kind: Kind, - val jsonTags: String, - val content: String, - val sig: HexKey, -) { - fun toEvent() = - EventFactory.create( - id, - pubKey, - createdAt, - kind, - OptimizedJsonMapper.fromJsonToTagArray(jsonTags), - content, - sig, - ) -} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnection.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnection.kt new file mode 100644 index 0000000000..62ed50f833 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnection.kt @@ -0,0 +1,99 @@ +/* + * Copyright (c) 2025 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. + */ +package com.vitorpamplona.quartz.nip01Core.store.sqlite + +import androidx.sqlite.SQLiteConnection +import androidx.sqlite.SQLiteStatement + +/** + * A [SQLiteConnection] decorator that keeps every prepared statement and + * hands the same handle back on the next [prepare] of the same SQL. The + * event-ingest hot path prepares the identical INSERT statements once per + * event; sqlite3_prepare is pure overhead the second time around. + * + * Callers keep their idiomatic `prepare(sql).use { … }` blocks untouched: + * the statement wrapper turns `close()` into a return-to-cache no-op (the + * real reset + clearBindings happens on the next checkout). Statements are + * only truly finalized when the connection itself closes. + * + * Constraints, by design of the call sites: + * - **Not thread-safe** — same contract as the underlying connection, + * which the pool already serializes (single writer under a mutex). + * - **No overlapping use of the same SQL** — checking out one SQL string + * twice without closing the first use would alias one native handle. + * Insert/query paths never nest the same statement; a checkout while + * the previous one is still open falls back to an uncached statement. + */ +class StatementCachingConnection( + private val delegate: SQLiteConnection, + /** + * Ceiling on retained statements. Query SQL embeds one `?` per filter + * element, so shape variety is client-controlled — without a cap a + * long-lived relay connection would accumulate native handles without + * bound. Once full, unseen SQL just prepares uncached. 256 comfortably + * covers the write path's fixed set plus the recurring filter shapes. + */ + private val maxCachedStatements: Int = 256, +) : SQLiteConnection by delegate { + private val cache = HashMap() + + override fun prepare(sql: String): SQLiteStatement { + val cached = + cache[sql] ?: run { + if (cache.size >= maxCachedStatements) return delegate.prepare(sql) + CachedStatement(delegate.prepare(sql)).also { cache[sql] = it } + } + if (cached.checkedOut) { + // Same SQL prepared while the previous handle is still in use — + // stay correct with a plain uncached statement. + return delegate.prepare(sql) + } + cached.checkedOut = true + cached.clearBindings() + return cached + } + + override fun close() { + cache.values.forEach { runCatching { it.finalize() } } + cache.clear() + delegate.close() + } + + private class CachedStatement( + private val delegate: SQLiteStatement, + ) : SQLiteStatement by delegate { + var checkedOut = false + + /** + * Return to cache. The real handle stays prepared, but must be + * reset *now*: an un-reset statement keeps its cursor (and its + * table read locks) open, which turns a later `DELETE`/`DROP` on + * the same connection into `SQLITE_LOCKED: database table is + * locked`. + */ + override fun close() { + runCatching { delegate.reset() } + checkedOut = false + } + + fun finalize() = delegate.close() + } +}