From d370784f3f526cdc5d0f5267d8ee193f520daffe Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Apr 2026 19:02:40 +0000 Subject: [PATCH] feat(quartz): reactive EventStoreProjection over SQLiteEventStore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a stateful observer that turns a Filter into a StateFlow>>: - A TEMP table + AFTER DELETE trigger on event_headers logs OLD.id for every row that leaves the store, regardless of cause (replaceable / addressable supersession, NIP-09 deletion, NIP-62 vanish, NIP-40 expiration sweep, manual delete, clearDB). - SQLiteEventStore drains the log around each writer unit of work and emits a StoreChange(inserted, removedIds) on a SharedFlow. - EventStoreProjection seeds itself from the store, then maintains stable MutableStateFlow handles keyed by (kind:pubkey:dtag) for addressables and by id otherwise. Addressable updates mutate the handle's value in place — list reference stable; insert / remove rebuilds the list reference. - 11 new tests cover seed, insert, replaceable update, addressable update, NIP-09 deletion, NIP-62 vanish, NIP-40 expiration, manual delete, limit enforcement, non-matching insert, and close-cancels-listener. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5 --- .../nip01Core/store/sqlite/ChangeLogModule.kt | 98 +++++ .../nip01Core/store/sqlite/EventStore.kt | 21 ++ .../store/sqlite/EventStoreProjection.kt | 247 +++++++++++++ .../store/sqlite/SQLiteConnectionPool.kt | 4 +- .../store/sqlite/SQLiteEventStore.kt | 76 +++- .../nip01Core/store/sqlite/StoreChange.kt | 57 +++ .../store/sqlite/EventStoreProjectionTest.kt | 347 ++++++++++++++++++ 7 files changed, 842 insertions(+), 8 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/ChangeLogModule.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjection.kt create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StoreChange.kt create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjectionTest.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/ChangeLogModule.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/ChangeLogModule.kt new file mode 100644 index 000000000..73d34cd95 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/ChangeLogModule.kt @@ -0,0 +1,98 @@ +/* + * 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 com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** + * Per-connection scratch space that records the id of every row that + * leaves `event_headers`, regardless of why it left. + * + * Why TEMP: the table and trigger are connection-scoped, so they only + * exist on the writer connection where they're installed. Readers never + * see them and never accumulate rows. The data also lives only for the + * lifetime of the connection — exactly the right scope for "ids removed + * since the last drain." + * + * The trigger fires after every delete on `event_headers`, including: + * - the supersession triggers in [ReplaceableModule] / [AddressableModule], + * - the cascade from `event_vanish` in [RightToVanishModule], + * - the explicit deletes in [DeletionRequestModule], + * - [ExpirationModule.deleteExpiredEvents], + * - manual `delete(filter)` / `delete(id)`, + * - `clearDB()`. + * + * The [SQLiteEventStore] drains this log around each unit of work and + * publishes the resulting ids in a [StoreChange]. + */ +class ChangeLogModule { + fun installOnWriter(db: SQLiteConnection) { + db.execSQL( + """ + CREATE TEMP TABLE IF NOT EXISTS event_change_log ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL + ) + """.trimIndent(), + ) + + db.execSQL( + """ + CREATE TEMP TRIGGER IF NOT EXISTS event_change_log_on_delete + AFTER DELETE ON event_headers + FOR EACH ROW + BEGIN + INSERT INTO event_change_log (id) VALUES (OLD.id); + END + """.trimIndent(), + ) + } + + /** + * Reads every id currently logged and clears the log. + * + * Must be called on the writer connection that owns the temp table. + * Callers are expected to be inside the writer mutex (drain happens + * inside or right after the `useWriter { ... }` block); a single + * read-and-clear pair is therefore atomic from the writer's view. + */ + fun drain(db: SQLiteConnection): List { + val ids = ArrayList() + db.prepare("SELECT id FROM event_change_log ORDER BY seq").use { stmt -> + while (stmt.step()) { + ids.add(stmt.getText(0)) + } + } + if (ids.isNotEmpty()) { + db.execSQL("DELETE FROM event_change_log") + } + return ids + } + + /** + * Drops everything from the log without returning it. Used after a + * rollback so the next successful unit of work starts clean. + */ + fun reset(db: SQLiteConnection) { + db.execSQL("DELETE FROM event_change_log") + } +} 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 7fb287555..f17be61ee 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 @@ -26,6 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore +import kotlinx.coroutines.CoroutineScope class EventStore( dbName: String? = "events.db", @@ -66,5 +67,25 @@ class EventStore( override suspend fun deleteExpiredEvents() = store.deleteExpiredEvents() + /** + * Stream of mutations committed to the store. See [SQLiteEventStore.changes]. + */ + val changes get() = store.changes + + /** + * Open a reactive [EventStoreProjection] over the store for + * [filters]. The projection runs inside [scope]; cancel the scope + * (or call [EventStoreProjection.close]) to release it. + */ + fun observe( + filters: List, + scope: CoroutineScope, + ): EventStoreProjection = EventStoreProjection(store, filters, scope) + + fun observe( + filter: Filter, + scope: CoroutineScope, + ): EventStoreProjection = observe(listOf(filter), scope) + override fun close() = store.close() } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjection.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjection.kt new file mode 100644 index 000000000..f655d6dfd --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjection.kt @@ -0,0 +1,247 @@ +/* + * 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 com.vitorpamplona.quartz.nip01Core.core.AddressableEvent +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.launch +import kotlinx.coroutines.yield + +/** + * A reactive projection over the [SQLiteEventStore] for a fixed set of + * [filters]. Each visible event is wrapped in a [MutableStateFlow] so + * the UI can collect three different kinds of change with the right + * granularity: + * + * - **Membership** (events arriving or leaving) re-emits a brand new + * [List] from [items]. The list reference is stable while membership + * is unchanged. + * - **In-place addressable update** (a new version of the same + * `kind:pubkey:dtag` arrives) updates the existing handle's + * [MutableStateFlow.value] without touching the list. Only collectors + * of that one handle re-render. The list ordering is *not* reshuffled + * when the new version has a later `created_at` — each slot remembers + * the sort key it was inserted with, so addressable updates feel like + * pure value mutations. + * - **Removal** (NIP-09 deletion, NIP-62 vanish, NIP-40 expiration, or + * a non-addressable event being explicitly deleted) drops the handle + * from the list. + * + * The seed is materialised by running the filters against the store + * once at start. After that, [items] is driven entirely by + * [SQLiteEventStore.changes]; the database is not re-queried on every + * mutation. + * + * Limit handling matches the existing in-memory observables in + * `commons/observables`: the initial query honours the filter `limit`, + * and we trim the list to the same cap when an insert pushes it over. + * We do **not** refill from the DB after a deletion — if a deletion + * leaves you under the limit, you stay under the limit until something + * new arrives. That tradeoff keeps the projection allocation-free per + * mutation and matches what callers were already getting from + * `LocalCache.observeEvents`. + * + * Lifecycle: the projection runs a single coroutine in [scope]. Cancel + * the scope (or call [close]) when the screen using the projection + * goes away. There is no shared state between projections; each one + * keeps its own indexes. + */ +class EventStoreProjection( + private val store: SQLiteEventStore, + private val filters: List, + scope: CoroutineScope, +) : AutoCloseable { + private val _items = MutableStateFlow>>(emptyList()) + val items: StateFlow>> = _items.asStateFlow() + + /** Slots keyed by the *current* event id. Re-keyed when an addressable handle takes a new version. */ + private val byId = HashMap>() + + /** Slots keyed by `kind:pubkey:dtag` for in-place addressable updates. */ + private val byAddress = HashMap>() + + /** + * Sorted view of the same slots. The comparator uses each slot's + * frozen sort key (the seed event's `created_at` + `id`), so an + * addressable update never moves a slot inside this set. + */ + private val ordered = sortedSetOf(slotComparator()) + + private val limit: Int? = filters.mapNotNull { it.limit }.maxOrNull() + + /** Set when the seed has been written to [items], so callers can suspend until the projection is hot. */ + val ready: CompletableDeferred = CompletableDeferred() + + private val job: Job = + scope.launch { + seed() + ready.complete(Unit) + store.changes.collect { change -> apply(change) } + } + + private suspend fun seed() { + val initial = store.query(filters) + for (event in initial) { + insertNew(event) + // Cooperate with cancellation on very large seeds. + yield() + } + publish() + } + + private fun apply(change: StoreChange) { + var changed = false + + // Removals first. A replaceable / addressable supersession + // arrives as `inserted = [new]` plus `removedIds = [oldId]`. + // When the new event is addressable and an existing slot + // already maps that address, [handleInsert] will rekey + // `byId` from the old id to the new id before this loop sees + // the old id — so the lookup here is a no-op for that case + // and the slot stays in place. For non-addressable + // replaceables (kinds 0/3/10000-19999) the old event has a + // different id, no address index, and we genuinely drop it. + for (event in change.inserted) { + if (handleInsert(event)) changed = true + } + for (id in change.removedIds) { + if (handleRemove(id)) changed = true + } + + if (changed) publish() + } + + @Suppress("UNCHECKED_CAST") + private fun handleInsert(event: Event): Boolean { + if (filters.none { it.match(event) }) return false + + if (event is AddressableEvent) { + val key = event.addressTag() + val existing = byAddress[key] + if (existing != null) { + // Same address, new version. Rekey byId from the + // previous event id to the new one and update the + // handle's value in place — list reference does not + // change, only the handle's collectors re-render. + val previousId = existing.flow.value.id + if (previousId != event.id) { + byId.remove(previousId) + byId[event.id] = existing + } + existing.flow.value = event as T + return false + } + } else if (byId.containsKey(event.id)) { + return false + } + + insertNew(event) + return true + } + + private fun handleRemove(id: HexKey): Boolean { + val slot = byId.remove(id) ?: return false + ordered.remove(slot) + val ev = slot.flow.value + if (ev is AddressableEvent) { + val addr = ev.addressTag() + // Only clear the address index if this slot still owns it. + if (byAddress[addr] === slot) byAddress.remove(addr) + } + return true + } + + @Suppress("UNCHECKED_CAST") + private fun insertNew(event: Event) { + val slot = Slot(event as T) + byId[event.id] = slot + if (event is AddressableEvent) byAddress[event.addressTag()] = slot + ordered.add(slot) + + val cap = limit ?: return + while (ordered.size > cap) { + val tail = ordered.last() + ordered.remove(tail) + val tailEvent = tail.flow.value + byId.remove(tailEvent.id) + if (tailEvent is AddressableEvent) { + val addr = tailEvent.addressTag() + if (byAddress[addr] === tail) byAddress.remove(addr) + } + } + } + + private fun publish() { + _items.value = ordered.map { it.flow } + } + + /** + * Stop tracking changes and clear internal state. Idempotent. The + * scope passed to the constructor keeps running; only this + * projection's collector job is cancelled. + */ + override fun close() { + job.cancel() + ordered.clear() + byId.clear() + byAddress.clear() + _items.value = emptyList() + } + + /** + * Internal slot. Each event added to the projection lives inside + * one of these for as long as it survives. The sort key is frozen + * at construction time — addressable in-place updates rewrite + * `flow.value` but never the sort key, so the ordering inside + * [ordered] is stable across updates. + */ + private class Slot( + initial: T, + ) { + val sortCreatedAt: Long = initial.createdAt + val sortId: HexKey = initial.id + val flow: MutableStateFlow = MutableStateFlow(initial) + } + + companion object { + /** + * created_at DESC, id ASC. The keys are snapshots taken at + * insertion time, so the ordering of a slot never changes + * after it joins the set. Distinct events have distinct ids, + * so no third-key tiebreak is needed. + */ + private fun slotComparator(): Comparator> = + Comparator { a, b -> + if (a === b) return@Comparator 0 + val byTime = b.sortCreatedAt.compareTo(a.sortCreatedAt) + if (byTime != 0) return@Comparator byTime + a.sortId.compareTo(b.sortId) + } + } +} 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 5ca19d985..2e4125b90 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 @@ -107,7 +107,7 @@ class SQLiteConnectionPool( * suspend until the lock is released. Cancellation-aware via the * coroutine [Mutex]. */ - suspend fun useWriter(block: (SQLiteConnection) -> T): T = + suspend fun useWriter(block: suspend (SQLiteConnection) -> T): T = writerMutex.withLock { block(writer) } @@ -118,7 +118,7 @@ class SQLiteConnectionPool( * (WAL). With an in-memory DB this falls back to the writer mutex * because each `:memory:` connection would be a separate database. */ - suspend fun useReader(block: (SQLiteConnection) -> T): T { + suspend fun useReader(block: suspend (SQLiteConnection) -> T): T { val ch = readerChannel ?: return writerMutex.withLock { block(writer) } 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 2da5cf49c..83c95d531 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 @@ -34,6 +34,10 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip40Expiration.isExpired import com.vitorpamplona.quartz.utils.EventFactory +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow class SQLiteEventStore( val driver: SQLiteDriver = BundledSQLiteDriver(), @@ -62,6 +66,22 @@ class SQLiteEventStore( val deletionModule = DeletionRequestModule(seedModule::hasher) val expirationModule = ExpirationModule() val rightToVanishModule = RightToVanishModule(seedModule::hasher) + val changeLogModule = ChangeLogModule() + + /** + * Stream of mutations committed to the store. One [StoreChange] per + * successful unit of work; rejected inserts and rolled-back + * transactions are not emitted. Subscribers see updates in commit + * order. See [EventStoreProjection] for a high-level reactive list + * that consumes this stream. + */ + private val _changes = + MutableSharedFlow( + replay = 0, + extraBufferCapacity = 256, + onBufferOverflow = BufferOverflow.SUSPEND, + ) + val changes: SharedFlow = _changes.asSharedFlow() val queryBuilder = QueryBuilder( @@ -119,6 +139,12 @@ class SQLiteEventStore( setUserVersion(this, DATABASE_VERSION) } } + + // After the schema is in place (fresh or already-current + // DBs both reach this point), wire the change log onto + // the writer connection. TEMP table + TEMP trigger are + // connection-scoped, so this runs on the writer only. + changeLogModule.installOnWriter(db) }, ) } @@ -160,10 +186,23 @@ class SQLiteEventStore( } } - suspend fun clearDB() = + suspend fun clearDB() { pool.useWriter { db -> modules.reversed().forEach { it.deleteAll(db) } + // The deleteAll cascade fires the change-log trigger for + // every removed event_header row. Drain and publish so + // open projections drop everything they were holding. + publishChange(emptyList(), changeLogModule.drain(db)) } + } + + private suspend fun publishChange( + inserted: List, + removedIds: List, + ) { + if (inserted.isEmpty() && removedIds.isEmpty()) return + _changes.emit(StoreChange(inserted, removedIds)) + } suspend fun vacuum() = pool.useWriter { db -> @@ -198,27 +237,38 @@ class SQLiteEventStore( db.transaction { innerInsertEvent(event, this) } + // The transaction either committed or threw. On commit, the + // change log holds ids superseded by replaceable / addressable + // triggers and any NIP-09 / NIP-62 cascades fired by this + // event's content. On rollback the temp table was rolled back + // with the transaction, so drain returns empty. + publishChange(listOf(event), changeLogModule.drain(db)) } } inner class Transaction( val db: SQLiteConnection, ) : IEventStore.ITransaction { + val accepted = ArrayList() + override fun insert(event: Event) { if (event.isExpired()) throw SQLiteException("blocked: Cannot insert an expired event") if (event.kind.isEphemeral()) return innerInsertEvent(event, db) + accepted.add(event) } } suspend fun transaction(body: Transaction.() -> Unit) { pool.useWriter { db -> + val txn = Transaction(db) db.transaction { - with(Transaction(this)) { + with(txn) { body() } } + publishChange(txn.accepted, changeLogModule.drain(db)) } } @@ -258,17 +308,31 @@ class SQLiteEventStore( suspend fun count(filters: List): Int = pool.useReader { queryBuilder.count(filters, it) } - suspend fun delete(filter: Filter) = pool.useWriter { queryBuilder.delete(filter, it) } + suspend fun delete(filter: Filter) = + pool.useWriter { db -> + queryBuilder.delete(filter, db) + publishChange(emptyList(), changeLogModule.drain(db)) + } - suspend fun delete(filters: List) = pool.useWriter { queryBuilder.delete(filters, it) } + suspend fun delete(filters: List) = + pool.useWriter { db -> + queryBuilder.delete(filters, db) + publishChange(emptyList(), changeLogModule.drain(db)) + } suspend fun delete(id: HexKey): Int = pool.useWriter { db -> db.execSQL("DELETE FROM event_headers WHERE id = ?", arrayOf(id)) - db.changes() + val count = db.changes() + publishChange(emptyList(), changeLogModule.drain(db)) + count } - suspend fun deleteExpiredEvents() = pool.useWriter { expirationModule.deleteExpiredEvents(it) } + suspend fun deleteExpiredEvents() = + pool.useWriter { db -> + expirationModule.deleteExpiredEvents(db) + publishChange(emptyList(), changeLogModule.drain(db)) + } fun close() = pool.close() } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StoreChange.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StoreChange.kt new file mode 100644 index 000000000..ef9742fb8 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StoreChange.kt @@ -0,0 +1,57 @@ +/* + * 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 com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** + * One atomic batch of mutations that occurred inside the event store. + * + * The store emits exactly one [StoreChange] per successfully committed + * unit of work — `insertEvent`, a `transaction { ... }` block, + * `delete(...)`, `deleteExpiredEvents()`, or `clearDB()`. Empty changes + * (e.g. an insert that was rejected by a trigger) are not emitted. + * + * `removedIds` covers every row that left `event_headers` during the + * unit of work, regardless of cause: the supersession triggers for + * replaceable / addressable events, NIP-09 deletion fan-out, NIP-62 + * vanish cascades, NIP-40 expiration sweeps, and direct `delete(...)` + * calls. They are captured by an `AFTER DELETE` trigger that writes + * `OLD.id` into a per-connection TEMP table. + * + * `inserted` carries the events that survived the writer and are now + * in the database. A replaceable / addressable supersession therefore + * appears as a single change with `inserted = [new]` and + * `removedIds = [oldId]`. + */ +data class StoreChange( + val inserted: List, + val removedIds: List, +) { + fun isEmpty() = inserted.isEmpty() && removedIds.isEmpty() + + fun isNotEmpty() = !isEmpty() + + companion object { + val EMPTY = StoreChange(emptyList(), emptyList()) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjectionTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjectionTest.kt new file mode 100644 index 000000000..f9effa0ac --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStoreProjectionTest.kt @@ -0,0 +1,347 @@ +/* + * 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 com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import com.vitorpamplona.quartz.nip23LongContent.LongTextNoteEvent +import com.vitorpamplona.quartz.nip40Expiration.expiration +import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent +import com.vitorpamplona.quartz.utils.Secp256k1Instance +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotSame +import kotlin.test.assertSame +import kotlin.test.assertTrue + +class EventStoreProjectionTest { + private val signer = NostrSignerSync() + private lateinit var store: EventStore + private lateinit var scope: CoroutineScope + + @BeforeTest + fun setUp() { + Secp256k1Instance + store = EventStore(dbName = null) + scope = CoroutineScope(SupervisorJob()) + } + + @AfterTest + fun tearDown() { + scope.cancel() + store.close() + } + + /** + * Wait until [items][EventStoreProjection.items] reaches a state + * for which [predicate] is true. We poll the StateFlow rather than + * collect because the projection only re-emits on membership + * change — in-place addressable updates intentionally don't move + * the list reference. + */ + private suspend fun EventStoreProjection.awaitItems( + timeoutMs: Long = 5_000, + predicate: (List>) -> Boolean, + ): List> = + withTimeout(timeoutMs) { + items.first { predicate(it) } + } + + /** + * Wait until [flow] reaches [expected]. Used to observe an + * in-place addressable update where the list reference doesn't + * change but the slot's value does. + */ + private suspend fun awaitFlow( + flow: MutableStateFlow, + timeoutMs: Long = 5_000, + predicate: (T) -> Boolean, + ): T = + withTimeout(timeoutMs) { + flow.first { predicate(it) } + } + + @Test + fun seedReturnsExistingEvents() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + store.insert(a) + store.insert(b) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + + val items = projection.items.value + assertEquals(2, items.size) + // Sorted by created_at DESC. + assertEquals(b.id, items[0].value.id) + assertEquals(a.id, items[1].value.id) + projection.close() + } + + @Test + fun insertAddsNewSlot() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + store.insert(a) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + val before = projection.items.value + assertEquals(1, before.size) + + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + store.insert(b) + + val after = projection.awaitItems { it.size == 2 } + assertNotSame(before, after, "insert must produce a new list reference") + assertEquals(b.id, after[0].value.id) + assertEquals(a.id, after[1].value.id) + projection.close() + } + + @Test + fun insertingNonMatchingEventDoesNotChangeList() = + runBlocking { + val text = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + store.insert(text) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + val seed = projection.items.value + + // Metadata kind doesn't match the filter. + val meta = signer.sign(MetadataEvent.createNew("Vitor", createdAt = 200)) + store.insert(meta) + + // Give the projection time to process the change. + delay(150) + assertSame(seed, projection.items.value) + projection.close() + } + + @Test + fun replaceableUpdateMutatesSlotInPlace() = + runBlocking { + val time = TimeUtils.now() + val v1 = signer.sign(MetadataEvent.createNew("v1", createdAt = time)) + store.insert(v1) + + val projection = + store.observe( + Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), + scope, + ) + projection.ready.await() + val seedList = projection.items.value + assertEquals(1, seedList.size) + val slot = seedList[0] + assertEquals(v1.id, slot.value.id) + + val v2 = signer.sign(MetadataEvent.createNew("v2", createdAt = time + 1)) + store.insert(v2) + + // The slot's flow updates... + awaitFlow(slot) { it.id == v2.id } + + // ...but the list reference is the SAME, and the slot is the SAME instance. + assertSame(seedList, projection.items.value, "addressable replace must not change list reference") + assertSame(slot, projection.items.value[0]) + projection.close() + } + + @Test + fun addressableUpdateMutatesSlotInPlace() = + runBlocking { + val time = TimeUtils.now() + val v1 = signer.sign(LongTextNoteEvent.build("blog v1", "title", dTag = "blog", createdAt = time)) + store.insert(v1) + + val projection = + store.observe( + Filter( + kinds = listOf(LongTextNoteEvent.KIND), + authors = listOf(v1.pubKey), + tags = mapOf("d" to listOf("blog")), + ), + scope, + ) + projection.ready.await() + val seedList = projection.items.value + assertEquals(1, seedList.size) + val slot = seedList[0] + assertEquals(v1.id, slot.value.id) + + val v2 = signer.sign(LongTextNoteEvent.build("blog v2", "title", dTag = "blog", createdAt = time + 1)) + store.insert(v2) + + awaitFlow(slot) { it.id == v2.id } + assertSame(seedList, projection.items.value, "addressable update must not change list reference") + assertSame(slot, projection.items.value[0]) + projection.close() + } + + @Test + fun nip09DeletionRemovesSlot() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + store.insert(a) + store.insert(b) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + assertEquals(2, projection.items.value.size) + + val deletion = signer.sign(DeletionEvent.build(listOf(a))) + store.insert(deletion) + + val after = projection.awaitItems { it.size == 1 } + assertEquals(b.id, after[0].value.id) + projection.close() + } + + @Test + fun nip62VanishRemovesAllAuthorsEvents() = + runBlocking { + val time = TimeUtils.now() + val a = signer.sign(TextNoteEvent.build("a", createdAt = time)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = time + 1)) + store.insert(a) + store.insert(b) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + assertEquals(2, projection.items.value.size) + + val vanish = + signer.sign( + RequestToVanishEvent.build( + "wss://quartz.local".normalizeRelayUrl(), + createdAt = time + 2, + ), + ) + store.insert(vanish) + + val after = projection.awaitItems { it.isEmpty() } + assertTrue(after.isEmpty()) + projection.close() + } + + @Test + fun nip40ExpirationRemovesSlotOnSweep() = + runBlocking { + val time = TimeUtils.now() + val safe = signer.sign(TextNoteEvent.build("safe", createdAt = time) { expiration(time + 100) }) + val short = signer.sign(TextNoteEvent.build("short", createdAt = time) { expiration(time + 1) }) + store.insert(safe) + store.insert(short) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + assertEquals(2, projection.items.value.size) + + // Wait for the expiration to lapse, then run the sweep. + delay(2000) + store.deleteExpiredEvents() + + val after = projection.awaitItems { it.size == 1 } + assertEquals(safe.id, after[0].value.id) + projection.close() + } + + @Test + fun manualDeleteByIdRemovesSlot() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + store.insert(a) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + assertEquals(1, projection.items.value.size) + + store.store.delete(a.id) + + val after = projection.awaitItems { it.isEmpty() } + assertTrue(after.isEmpty()) + projection.close() + } + + @Test + fun limitIsEnforcedOnInsertOverflow() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + store.insert(a) + store.insert(b) + + val projection = + store.observe( + Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2), + scope, + ) + projection.ready.await() + assertEquals(2, projection.items.value.size) + + // Newer event arrives; it should push the oldest out. + val c = signer.sign(TextNoteEvent.build("c", createdAt = 300)) + store.insert(c) + + val after = projection.awaitItems { it[0].value.id == c.id } + assertEquals(2, after.size) + assertEquals(c.id, after[0].value.id) + assertEquals(b.id, after[1].value.id) + projection.close() + } + + @Test + fun closeStopsListening() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + store.insert(a) + + val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + projection.ready.await() + projection.close() + + // Subsequent inserts must not surface in the (now empty) projection. + store.insert(signer.sign(TextNoteEvent.build("b", createdAt = 200))) + delay(150) + assertTrue(projection.items.value.isEmpty()) + } +}