From 32850c3d4032d09c248690227b5811db1b772585 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Apr 2026 21:06:27 +0000 Subject: [PATCH] refactor(quartz): make ObservableEventStore an external composition layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reverts the previous round's choice of embedding an ObservableEventStore inside EventStore. The wrapper now lives strictly outside any specific store: callers compose `ObservableEventStore(EventStore(...))` (or `ObservableEventStore(FsEventStore(...))`, or any IEventStore) and use that as their projection bus. EventStore is back to a plain IEventStore implementation over SQLiteEventStore — no `observable`, `events`, or `observe()` on it. ObservableEventStore gains the `observe(filters, relay, scope)` and `observe(filter, relay, scope)` convenience methods. NIP-62 vanish scoping is passed in per-projection (rather than a constructor field on the wrapper) so the same observable can feed projections with different relay scopes. Tests construct `observable = ObservableEventStore(store)` in `setUp` and route inserts through `observable.insert(...)` so the projection bus actually publishes them. 14/14 projection + 219/219 other store tests pass. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5 --- .../store/observable/ObservableEventStore.kt | 22 +++++ .../nip01Core/store/sqlite/EventStore.kt | 80 +-------------- .../projection/EventStoreProjectionTest.kt | 97 ++++++++++--------- 3 files changed, 79 insertions(+), 120 deletions(-) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt index eeb53e22f..7757b9fba 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt @@ -23,8 +23,11 @@ package com.vitorpamplona.quartz.nip01Core.store.observable import com.vitorpamplona.quartz.nip01Core.core.Event 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.IEventStore +import com.vitorpamplona.quartz.nip01Core.store.projection.EventStoreProjection import com.vitorpamplona.quartz.nip40Expiration.isExpired +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow @@ -139,5 +142,24 @@ class ObservableEventStore( override suspend fun deleteExpiredEvents() = inner.deleteExpiredEvents() + /** + * Open a reactive [EventStoreProjection] over this observable + * store. [relay] scopes NIP-62 vanish handling — pass the relay + * URL the events are arriving from, or `null` to apply only + * unscoped (`ALL_RELAYS`) vanish requests. Cancel [scope] (or + * call [EventStoreProjection.close]) to release the projection. + */ + fun observe( + filters: List, + relay: NormalizedRelayUrl?, + scope: CoroutineScope, + ): EventStoreProjection = EventStoreProjection(this, filters, relay, scope) + + fun observe( + filter: Filter, + relay: NormalizedRelayUrl?, + scope: CoroutineScope, + ): EventStoreProjection = observe(listOf(filter), relay, scope) + override fun close() = inner.close() } 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 9d188d9a2..0a0f1a75b 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,16 +26,11 @@ 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 com.vitorpamplona.quartz.nip01Core.store.observable.ObservableEventStore -import com.vitorpamplona.quartz.nip01Core.store.projection.EventStoreProjection -import kotlinx.coroutines.CoroutineScope /** - * SQLite-backed event store with a built-in [ObservableEventStore] - * façade so [observe] returns an [EventStoreProjection] without - * extra plumbing. Persistence goes through [SQLiteEventStore]; the - * observable layer takes care of routing ephemerals (which never hit - * the DB) to projection collectors. + * SQLite-backed [IEventStore] with default DB-file name and relay + * scoping. Wrap in `ObservableEventStore` if you want to feed + * `EventStoreProjection`. */ class EventStore( dbName: String? = "events.db", @@ -43,14 +38,10 @@ class EventStore( val indexStrategy: IndexingStrategy = DefaultIndexingStrategy(), ) : IEventStore { val store = SQLiteEventStore(BundledSQLiteDriver(), dbName, relay, indexStrategy) - val observable = ObservableEventStore(SQLiteAdapter(store)) - /** Stream of events accepted for observation. See [ObservableEventStore.events]. */ - val events get() = observable.events + override suspend fun insert(event: Event) = store.insertEvent(event) - override suspend fun insert(event: Event) = observable.insert(event) - - override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = observable.transaction(body) + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = store.transaction(body) override suspend fun query(filter: Filter) = store.query(filter) @@ -80,66 +71,5 @@ class EventStore( override suspend fun deleteExpiredEvents() = store.deleteExpiredEvents() - /** - * Open a reactive [EventStoreProjection] over this store with - * NIP-62 vanish scoping bound to the store's [relay]. Cancel - * [scope] (or call [EventStoreProjection.close]) to release it. - */ - fun observe( - filters: List, - scope: CoroutineScope, - ): EventStoreProjection = EventStoreProjection(observable, filters, relay, scope) - - fun observe( - filter: Filter, - scope: CoroutineScope, - ): EventStoreProjection = observe(listOf(filter), scope) - override fun close() = store.close() } - -/** - * Adapts the non-IEventStore [SQLiteEventStore] to [IEventStore] for - * the [ObservableEventStore] wrapper. SQLiteEventStore predates the - * IEventStore contract; this is a thin forwarder, not a behavioural - * shim. - */ -private class SQLiteAdapter( - val sqlite: SQLiteEventStore, -) : IEventStore { - override suspend fun insert(event: Event) = sqlite.insertEvent(event) - - override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { - sqlite.transaction { body() } - } - - override suspend fun query(filter: Filter) = sqlite.query(filter) - - override suspend fun query(filters: List) = sqlite.query(filters) - - override suspend fun query( - filter: Filter, - onEach: (T) -> Unit, - ) = sqlite.query(filter, onEach) - - override suspend fun query( - filters: List, - onEach: (T) -> Unit, - ) = sqlite.query(filters, onEach) - - override suspend fun count(filter: Filter) = sqlite.count(filter) - - override suspend fun count(filters: List) = sqlite.count(filters) - - override suspend fun delete(filter: Filter) { - sqlite.delete(filter) - } - - override suspend fun delete(filters: List) { - sqlite.delete(filters) - } - - override suspend fun deleteExpiredEvents() = sqlite.deleteExpiredEvents() - - override fun close() = sqlite.close() -} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt index 550c03b1b..182847661 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt @@ -25,6 +25,7 @@ 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.nip01Core.store.observable.ObservableEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent @@ -53,12 +54,14 @@ class EventStoreProjectionTest { private val signer = NostrSignerSync() private val otherSigner = NostrSignerSync() private lateinit var store: EventStore + private lateinit var observable: ObservableEventStore private lateinit var scope: CoroutineScope @BeforeTest fun setUp() { Secp256k1Instance store = EventStore(dbName = null) + observable = ObservableEventStore(store) scope = CoroutineScope(SupervisorJob()) } @@ -90,10 +93,10 @@ class EventStoreProjectionTest { 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) + observable.insert(a) + observable.insert(b) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() val items = projection.items.value @@ -107,15 +110,15 @@ class EventStoreProjectionTest { fun insertAddsNewSlot() = runBlocking { val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) - store.insert(a) + observable.insert(a) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, 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) + observable.insert(b) val after = projection.awaitItems { it.size == 2 } assertNotSame(before, after, "insert must produce a new list reference") @@ -127,14 +130,14 @@ class EventStoreProjectionTest { fun nonMatchingInsertDoesNotChangeList() = runBlocking { val text = signer.sign(TextNoteEvent.build("a", createdAt = 100)) - store.insert(text) + observable.insert(text) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() val seed = projection.items.value val meta = signer.sign(MetadataEvent.createNew("Vitor", createdAt = 200)) - store.insert(meta) + observable.insert(meta) delay(150) assertSame(seed, projection.items.value) @@ -146,11 +149,12 @@ class EventStoreProjectionTest { runBlocking { val time = TimeUtils.now() val v1 = signer.sign(MetadataEvent.createNew("v1", createdAt = time)) - store.insert(v1) + observable.insert(v1) val projection = - store.observe( + observable.observe( Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), + store.relay, scope, ) projection.ready.await() @@ -160,7 +164,7 @@ class EventStoreProjectionTest { assertEquals(v1.id, slot.value.id) val v2 = signer.sign(MetadataEvent.createNew("v2", createdAt = time + 1)) - store.insert(v2) + observable.insert(v2) awaitFlow(slot) { it.id == v2.id } assertSame(seedList, projection.items.value, "replaceable update must not change list reference") @@ -173,15 +177,16 @@ class EventStoreProjectionTest { runBlocking { val time = TimeUtils.now() val v1 = signer.sign(LongTextNoteEvent.build("blog v1", "title", dTag = "blog", createdAt = time)) - store.insert(v1) + observable.insert(v1) val projection = - store.observe( + observable.observe( Filter( kinds = listOf(LongTextNoteEvent.KIND), authors = listOf(v1.pubKey), tags = mapOf("d" to listOf("blog")), ), + store.relay, scope, ) projection.ready.await() @@ -189,7 +194,7 @@ class EventStoreProjectionTest { val slot = seedList[0] val v2 = signer.sign(LongTextNoteEvent.build("blog v2", "title", dTag = "blog", createdAt = time + 1)) - store.insert(v2) + observable.insert(v2) awaitFlow(slot) { it.id == v2.id } assertSame(seedList, projection.items.value, "addressable update must not change list reference") @@ -212,11 +217,12 @@ class EventStoreProjectionTest { // Seed the projection with v2 before v1 even hits the store // — by inserting v2 first. - store.insert(v2) + observable.insert(v2) val projection = - store.observe( + observable.observe( Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey)), + store.relay, scope, ) projection.ready.await() @@ -227,7 +233,7 @@ class EventStoreProjectionTest { // projection therefore never sees v1 on the inserts // stream. The slot must still hold v2. try { - store.insert(v1) + observable.insert(v1) } catch (_: Throwable) { // expected — store enforces the same rule } @@ -242,15 +248,15 @@ class EventStoreProjectionTest { 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) + observable.insert(a) + observable.insert(b) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() assertEquals(2, projection.items.value.size) val deletion = signer.sign(DeletionEvent.build(listOf(a))) - store.insert(deletion) + observable.insert(deletion) val after = projection.awaitItems { it.size == 1 } assertEquals(b.id, after[0].value.id) @@ -265,15 +271,15 @@ class EventStoreProjectionTest { fun nip09CrossAuthorDeletionIsInert() = runBlocking { val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) - store.insert(a) + observable.insert(a) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() val seed = projection.items.value assertEquals(1, seed.size) val foreignDeletion = otherSigner.sign(DeletionEvent.build(listOf(a))) - store.insert(foreignDeletion) + observable.insert(foreignDeletion) // Give the projection time to process the event. delay(150) @@ -292,10 +298,10 @@ class EventStoreProjectionTest { 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) + observable.insert(a) + observable.insert(b) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() assertEquals(2, projection.items.value.size) @@ -306,7 +312,7 @@ class EventStoreProjectionTest { createdAt = time + 2, ), ) - store.insert(vanish) + observable.insert(vanish) val after = projection.awaitItems { it.isEmpty() } assertTrue(after.isEmpty()) @@ -322,9 +328,9 @@ class EventStoreProjectionTest { runBlocking { val time = TimeUtils.now() val a = signer.sign(TextNoteEvent.build("a", createdAt = time)) - store.insert(a) + observable.insert(a) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() val seed = projection.items.value @@ -335,7 +341,7 @@ class EventStoreProjectionTest { createdAt = time + 2, ), ) - store.insert(foreignVanish) + observable.insert(foreignVanish) delay(150) assertSame(seed, projection.items.value) @@ -353,13 +359,13 @@ class EventStoreProjectionTest { 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) + observable.insert(safe) + observable.insert(short) // Drive the ticker frequently so the test doesn't sit idle. val projection = EventStoreProjection( - store.observable, + observable, listOf(Filter(kinds = listOf(TextNoteEvent.KIND))), relay = null, scope = scope, @@ -381,19 +387,20 @@ class EventStoreProjectionTest { 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) + observable.insert(a) + observable.insert(b) val projection = - store.observe( + observable.observe( Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2), + store.relay, scope, ) projection.ready.await() assertEquals(2, projection.items.value.size) val c = signer.sign(TextNoteEvent.build("c", createdAt = 300)) - store.insert(c) + observable.insert(c) val after = projection.awaitItems { it[0].value.id == c.id } assertEquals(2, after.size) @@ -406,13 +413,13 @@ class EventStoreProjectionTest { fun closeStopsListening() = runBlocking { val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) - store.insert(a) + observable.insert(a) - val projection = store.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), scope) + val projection = observable.observe(Filter(kinds = listOf(TextNoteEvent.KIND)), store.relay, scope) projection.ready.await() projection.close() - store.insert(signer.sign(TextNoteEvent.build("b", createdAt = 200))) + observable.insert(signer.sign(TextNoteEvent.build("b", createdAt = 200))) delay(150) assertTrue(projection.items.value.isEmpty()) } @@ -430,7 +437,7 @@ class EventStoreProjectionTest { runBlocking { val ephemeralKind = 22_000 val projection = - store.observe(Filter(kinds = listOf(ephemeralKind)), scope) + observable.observe(Filter(kinds = listOf(ephemeralKind)), store.relay, scope) projection.ready.await() assertTrue(projection.items.value.isEmpty()) @@ -441,7 +448,7 @@ class EventStoreProjectionTest { arrayOf(emptyArray()), "live", ) - store.insert(ephemeral) + observable.insert(ephemeral) val after = projection.awaitItems { it.size == 1 } assertEquals(ephemeral.id, after[0].value.id) @@ -453,7 +460,7 @@ class EventStoreProjectionTest { // A fresh projection on the same store gets nothing — the // event was only ever live, not durable. val freshProjection = - store.observe(Filter(kinds = listOf(ephemeralKind)), scope) + observable.observe(Filter(kinds = listOf(ephemeralKind)), store.relay, scope) freshProjection.ready.await() assertTrue(freshProjection.items.value.isEmpty())