nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(quartz): make ObservableEventStore an external composition layer

Browse Source 
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
This commit is contained in:
Claude
2026-04-29 21:06:27 +00:00
parent 86537edef5
commit 32850c3d40
3 changed files with 79 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt
+22
View File
@@ -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 <T : Event> observe(
filters: List<Filter>,
relay: NormalizedRelayUrl?,
scope: CoroutineScope,
): EventStoreProjection<T> = EventStoreProjection(this, filters, relay, scope)
fun <T : Event> observe(
filter: Filter,
relay: NormalizedRelayUrl?,
scope: CoroutineScope,
): EventStoreProjection<T> = observe(listOf(filter), relay, scope)
override fun close() = inner.close()
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt
+5 -75
View File
@@ -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 <T : Event> query(filter: Filter) = store.query<T>(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 <T : Event> observe(
filters: List<Filter>,
scope: CoroutineScope,
): EventStoreProjection<T> = EventStoreProjection(observable, filters, relay, scope)
fun <T : Event> observe(
filter: Filter,
scope: CoroutineScope,
): EventStoreProjection<T> = 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 <T : Event> query(filter: Filter) = sqlite.query<T>(filter)
override suspend fun <T : Event> query(filters: List<Filter>) = sqlite.query<T>(filters)
override suspend fun <T : Event> query(
filter: Filter,
onEach: (T) -> Unit,
) = sqlite.query(filter, onEach)
override suspend fun <T : Event> query(
filters: List<Filter>,
onEach: (T) -> Unit,
) = sqlite.query(filters, onEach)
override suspend fun count(filter: Filter) = sqlite.count(filter)
override suspend fun count(filters: List<Filter>) = sqlite.count(filters)
override suspend fun delete(filter: Filter) {
sqlite.delete(filter)
}
override suspend fun delete(filters: List<Filter>) {
sqlite.delete(filters)
}
override suspend fun deleteExpiredEvents() = sqlite.deleteExpiredEvents()
override fun close() = sqlite.close()
}
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt
+52 -45
View File
@@ -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<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<TextNoteEvent>(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<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<TextNoteEvent>(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<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<TextNoteEvent>(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<MetadataEvent>(
observable.observe<MetadataEvent>(
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<LongTextNoteEvent>(
observable.observe<LongTextNoteEvent>(
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<MetadataEvent>(
observable.observe<MetadataEvent>(
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<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<Event>(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<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<Event>(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<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<Event>(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<Event>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<Event>(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<Event>(
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<TextNoteEvent>(
observable.observe<TextNoteEvent>(
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<TextNoteEvent>(Filter(kinds = listOf(TextNoteEvent.KIND)), scope)
val projection = observable.observe<TextNoteEvent>(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<Event>(Filter(kinds = listOf(ephemeralKind)), scope)
observable.observe<Event>(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<Event>(Filter(kinds = listOf(ephemeralKind)), scope)
observable.observe<Event>(Filter(kinds = listOf(ephemeralKind)), store.relay, scope)
freshProjection.ready.await()
assertTrue(freshProjection.items.value.isEmpty())