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 8a92f02c7..e70773c29 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 @@ -22,7 +22,6 @@ package com.vitorpamplona.quartz.nip01Core.store import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter -import kotlinx.coroutines.flow.SharedFlow interface IEventStore : AutoCloseable { suspend fun insert(event: Event) @@ -57,23 +56,5 @@ interface IEventStore : AutoCloseable { suspend fun deleteExpiredEvents() - /** - * Stream of events the store accepted into durable storage. One - * emission per successfully inserted event, in commit order. - * Rejected inserts (expired, ephemeral, blocked by tombstone / - * vanish, NIP-01 supersession loser) emit nothing. - * - * Consumed by `EventStoreProjection` to maintain a live view — - * the projection itself replays NIP-01 supersession, NIP-09 - * deletion fan-out, NIP-62 vanish cascades, and NIP-40 expiration - * from these events, so stores don't need to publish removals. - * - * Out-of-band removals (`delete(id)`, `delete(filter)`, `clearDB()`, - * `deleteExpiredEvents()`) are not visible on this stream — they're - * maintenance operations and projections survive a missed mutation - * by re-seeding when their scope is restarted. - */ - val inserts: SharedFlow - override fun close() } 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 new file mode 100644 index 000000000..eeb53e22f --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt @@ -0,0 +1,143 @@ +/* + * 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.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.store.IEventStore +import com.vitorpamplona.quartz.nip40Expiration.isExpired +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow + +/** + * A reactive façade over any [IEventStore] that publishes every event + * accepted for *observation* — a superset of the events the inner + * store persists. + * + * The split between persistence and observation is the whole point of + * this class: + * + * - **Non-ephemeral events** are forwarded to the inner store. If the + * inner store rejects (expired, NIP-09 / NIP-62 tombstone, NIP-01 + * supersession loser), the rejection propagates and nothing is + * emitted on [events]. + * - **Ephemeral events** (kinds `20000-29999`) skip the inner store + * entirely — they're never persisted — but they still emit on + * [events] so projections can render them while they live. Already + * expired ephemerals are silently dropped. + * + * Wrap any store you want to observe — [SQLiteEventStore], FS-backed, + * an in-memory test fake — and feed [EventStoreProjection] from the + * resulting [events] flow. + * + * Reads (`query`, `count`) and out-of-band writes (`delete`, + * `deleteExpiredEvents`) forward to the inner store unchanged. The + * latter are *not* surfaced on [events] — see the projection's + * docstring for the rationale. + */ +class ObservableEventStore( + val inner: IEventStore, +) : IEventStore { + private val _events = + MutableSharedFlow( + replay = 0, + extraBufferCapacity = 256, + onBufferOverflow = BufferOverflow.SUSPEND, + ) + + /** + * Stream of events accepted for observation, persisted or not. + * One emission per successful [insert] (or per successful entry + * inside [transaction]); rejected inserts emit nothing. + */ + val events: SharedFlow = _events.asSharedFlow() + + override suspend fun insert(event: Event) { + if (event.kind.isEphemeral()) { + // Ephemeral kinds bypass persistence. We still drop ones + // that are already expired — they were never going to live + // long enough for a UI to render them. + if (event.isExpired()) return + _events.emit(event) + return + } + // Non-ephemeral: let the inner store enforce expiration, + // tombstones, supersession, etc. If it throws, we never emit. + inner.insert(event) + _events.emit(event) + } + + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { + val accepted = ArrayList() + inner.transaction { + val innerTxn = this + val wrapped = + object : IEventStore.ITransaction { + override fun insert(event: Event) { + if (event.kind.isEphemeral()) { + // Same ephemeral handling as the single-event + // path. The inner txn never sees these so the + // store's batch contains only persistable + // events. + if (event.isExpired()) return + accepted.add(event) + return + } + innerTxn.insert(event) + accepted.add(event) + } + } + wrapped.body() + } + // Emit only after the inner transaction commits. If it throws + // / rolls back, `accepted` is discarded. + for (e in accepted) _events.emit(e) + } + + override suspend fun query(filter: Filter): List = inner.query(filter) + + override suspend fun query(filters: List): List = inner.query(filters) + + override suspend fun query( + filter: Filter, + onEach: (T) -> Unit, + ) = inner.query(filter, onEach) + + override suspend fun query( + filters: List, + onEach: (T) -> Unit, + ) = inner.query(filters, onEach) + + override suspend fun count(filter: Filter): Int = inner.count(filter) + + override suspend fun count(filters: List): Int = inner.count(filters) + + override suspend fun delete(filter: Filter) = inner.delete(filter) + + override suspend fun delete(filters: List) = inner.delete(filters) + + override suspend fun deleteExpiredEvents() = inner.deleteExpiredEvents() + + override fun close() = inner.close() +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt index 3882b2db5..4e7c00bb4 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt @@ -27,7 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.core.isAddressable import com.vitorpamplona.quartz.nip01Core.core.isReplaceable 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.observable.ObservableEventStore import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent import com.vitorpamplona.quartz.nip40Expiration.expiration import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent @@ -40,12 +40,14 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.onSubscription import kotlinx.coroutines.launch import kotlinx.coroutines.yield /** - * A reactive projection over any [IEventStore] for a fixed set of - * [filters]. Each visible event is wrapped in a [MutableStateFlow] so + * A reactive projection over an [ObservableEventStore] 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: * @@ -63,10 +65,10 @@ import kotlinx.coroutines.yield * drops the handle from the list. * * The seed is materialised by querying the store once at start, after - * which the projection is driven entirely by [IEventStore.inserts] and - * its own expiration ticker. The store is never re-queried on - * mutation, and the projection never asks the store to delete anything - * — it interprets incoming Nostr events itself: + * which the projection is driven entirely by [ObservableEventStore.events] + * and its own expiration ticker. The store is never re-queried on + * mutation, and the projection never asks the store to delete + * anything — it interprets incoming Nostr events itself: * * - **NIP-01 supersession.** New replaceable / addressable events * replace prior ones for the same `kind:pubkey[:dtag]`. The @@ -96,16 +98,23 @@ import kotlinx.coroutines.yield * * Out-of-band store mutations — `delete(id)`, `delete(filter)`, * `clearDB()`, the periodic `deleteExpiredEvents()` sweep — are not - * visible on [IEventStore.inserts] and won't update an open + * visible on [ObservableEventStore.events] and won't update an open * projection. Re-open the projection (e.g. cancel and recreate the * scope) to pick up an out-of-band change. * + * Ephemeral events (kinds `20000-29999`) reach the projection via + * [ObservableEventStore.events] without ever being persisted; they + * appear in [items] for as long as the projection is alive but never + * survive a re-seed. NIP-40 expiration applies to them too — if they + * carry an `expiration` tag, the per-projection ticker drops them + * when it lapses. + * * Lifecycle: the projection runs a collector + an expiration ticker * inside [scope]. Cancel the scope (or call [close]) when the screen * using the projection goes away. */ class EventStoreProjection( - private val store: IEventStore, + private val store: ObservableEventStore, private val filters: List, private val relay: NormalizedRelayUrl?, scope: CoroutineScope, @@ -138,9 +147,16 @@ class EventStoreProjection( private val collectorJob: Job = scope.launch { - seed() - ready.complete(Unit) - store.inserts.collect { event -> apply(event) } + // `onSubscription` runs after the SharedFlow subscription is + // active but before we pull any events — the buffer absorbs + // any emissions that arrive while we seed, and we then drain + // them as the collect proceeds. Doing the seed inside + // `collect { }` instead would race with concurrent inserts. + store.events + .onSubscription { + seed() + ready.complete(Unit) + }.collect { event -> apply(event) } } private val expirationJob: Job = 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 092f10bcb..9d188d9a2 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,19 +26,31 @@ 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. + */ class EventStore( dbName: String? = "events.db", val relay: NormalizedRelayUrl? = "wss://quartz.local/".normalizeRelayUrl(), val indexStrategy: IndexingStrategy = DefaultIndexingStrategy(), ) : IEventStore { val store = SQLiteEventStore(BundledSQLiteDriver(), dbName, relay, indexStrategy) + val observable = ObservableEventStore(SQLiteAdapter(store)) - override suspend fun insert(event: Event) = store.insertEvent(event) + /** Stream of events accepted for observation. See [ObservableEventStore.events]. */ + val events get() = observable.events - override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = store.transaction(body) + override suspend fun insert(event: Event) = observable.insert(event) + + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = observable.transaction(body) override suspend fun query(filter: Filter) = store.query(filter) @@ -68,8 +80,6 @@ class EventStore( override suspend fun deleteExpiredEvents() = store.deleteExpiredEvents() - override val inserts get() = store.inserts - /** * Open a reactive [EventStoreProjection] over this store with * NIP-62 vanish scoping bound to the store's [relay]. Cancel @@ -78,7 +88,7 @@ class EventStore( fun observe( filters: List, scope: CoroutineScope, - ): EventStoreProjection = EventStoreProjection(this, filters, relay, scope) + ): EventStoreProjection = EventStoreProjection(observable, filters, relay, scope) fun observe( filter: Filter, @@ -87,3 +97,49 @@ class EventStore( 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/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteConnectionPool.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteConnectionPool.kt index 2e4125b90..5ca19d985 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: suspend (SQLiteConnection) -> T): T = + suspend fun useWriter(block: (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: suspend (SQLiteConnection) -> T): T { + suspend fun useReader(block: (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 9b3687baf..2da5cf49c 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,10 +34,6 @@ 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(), @@ -67,18 +63,6 @@ class SQLiteEventStore( val expirationModule = ExpirationModule() val rightToVanishModule = RightToVanishModule(seedModule::hasher) - /** - * Stream of events the store accepted into durable storage. See - * [IEventStore.inserts] for the contract. - */ - private val _inserts = - MutableSharedFlow( - replay = 0, - extraBufferCapacity = 256, - onBufferOverflow = BufferOverflow.SUSPEND, - ) - val inserts: SharedFlow = _inserts.asSharedFlow() - val queryBuilder = QueryBuilder( fullTextSearchModule, @@ -214,38 +198,27 @@ class SQLiteEventStore( db.transaction { innerInsertEvent(event, this) } - // The transaction either committed or threw. On commit the - // event is durable; emit so subscribed projections see it. - // On rollback the throw propagates and we never reach this. - _inserts.emit(event) } } 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(txn) { + with(Transaction(this)) { body() } } - // Emit each accepted event after the batch commits — the - // projection sees them in the same order they were inserted. - for (e in txn.accepted) _inserts.emit(e) } } 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 7644f1d31..550c03b1b 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 @@ -359,7 +359,7 @@ class EventStoreProjectionTest { // Drive the ticker frequently so the test doesn't sit idle. val projection = EventStoreProjection( - store, + store.observable, listOf(Filter(kinds = listOf(TextNoteEvent.KIND))), relay = null, scope = scope, @@ -416,4 +416,48 @@ class EventStoreProjectionTest { delay(150) assertTrue(projection.items.value.isEmpty()) } + + /** + * Ephemeral events (kinds `20000-29999`) are never persisted, so + * the inner SQLite store silently drops them. The + * `ObservableEventStore` wrapper still routes them onto its + * [events][com.vitorpamplona.quartz.nip01Core.store.observable.ObservableEventStore.events] + * flow, so an open projection sees them while it's alive. They + * vanish from any future seed because the DB never had them. + */ + @Test + fun ephemeralEventsAppearInProjection() = + runBlocking { + val ephemeralKind = 22_000 + val projection = + store.observe(Filter(kinds = listOf(ephemeralKind)), scope) + projection.ready.await() + assertTrue(projection.items.value.isEmpty()) + + val ephemeral: Event = + signer.sign( + TimeUtils.now(), + ephemeralKind, + arrayOf(emptyArray()), + "live", + ) + store.insert(ephemeral) + + val after = projection.awaitItems { it.size == 1 } + assertEquals(ephemeral.id, after[0].value.id) + + // Confirmation that the inner SQLite store didn't persist. + val persisted = store.query(Filter(kinds = listOf(ephemeralKind))) + assertEquals(0, persisted.size) + + // 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) + freshProjection.ready.await() + assertTrue(freshProjection.items.value.isEmpty()) + + projection.close() + freshProjection.close() + } } diff --git a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt index eec9265f1..8da5d14b5 100644 --- a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt +++ b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt @@ -36,10 +36,6 @@ import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent import com.vitorpamplona.quartz.nip40Expiration.expiration import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent import com.vitorpamplona.quartz.utils.TimeUtils -import kotlinx.coroutines.channels.BufferOverflow -import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.SharedFlow -import kotlinx.coroutines.flow.asSharedFlow import java.nio.file.FileAlreadyExistsException import java.nio.file.Files import java.nio.file.Path @@ -99,40 +95,19 @@ open class FsEventStore( planner = FsQueryPlanner(layout, hasher) } - /** - * Stream of events newly persisted by this store. See - * [IEventStore.inserts] for the contract. Events that are no-op - * idempotent retries (canonical already on disk from a previous - * call) are not re-emitted; the original insert already published. - */ - private val _inserts = - MutableSharedFlow( - replay = 0, - extraBufferCapacity = 256, - onBufferOverflow = BufferOverflow.SUSPEND, - ) - override val inserts: SharedFlow = _inserts.asSharedFlow() - // ------------------------------------------------------------------ // Insert // ------------------------------------------------------------------ - override suspend fun insert(event: Event) { - val accepted = lockManager.withWriteLock { insertLocked(event) } - if (accepted) _inserts.emit(event) - } + override suspend fun insert(event: Event) = + lockManager.withWriteLock { + insertLocked(event) + } - /** - * Inserts the event under the write lock and returns true iff this - * call is the one that newly persisted it. Returns false for the - * no-op paths (ephemeral, expired, blocked, supersession loser, or - * canonical already on disk from a prior call) so callers can - * decide whether to publish on [inserts]. - */ - private fun insertLocked(event: Event): Boolean { - if (event.kind.isEphemeral()) return false - if (isAlreadyExpired(event)) return false - if (isBlockedByTombstone(event)) return false + private fun insertLocked(event: Event) { + if (event.kind.isEphemeral()) return + if (isAlreadyExpired(event)) return + if (isBlockedByTombstone(event)) return val slot = slots.slotPathFor(event) val existingSlot = slot?.let { slots.readSlot(it) } @@ -146,7 +121,7 @@ open class FsEventStore( // (later createdAt, or same createdAt with the lexically smaller // id). Matches the ReplaceableModule / AddressableModule trigger // condition in SQLite. - return false + return } val canonical = layout.canonical(event.id) @@ -159,7 +134,7 @@ open class FsEventStore( } if (event is DeletionEvent) processDeletion(event, canonical) if (event is RequestToVanishEvent) processVanish(event, canonical) - return false + return } Files.createDirectories(canonical.parent) @@ -172,7 +147,7 @@ open class FsEventStore( // A concurrent writer won the race. Canonical is immutable // so the other copy is equivalent — drop our tmp and return. Files.deleteIfExists(tmp) - return false + return } Files.setLastModifiedTime(canonical, FileTime.from(event.createdAt, TimeUnit.SECONDS)) indexer.link(event, canonical) @@ -181,7 +156,6 @@ open class FsEventStore( } if (event is DeletionEvent) processDeletion(event, canonical) if (event is RequestToVanishEvent) processVanish(event, canonical) - return true } catch (t: Throwable) { Files.deleteIfExists(tmp) throw t @@ -289,21 +263,14 @@ open class FsEventStore( } } - override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { - val accepted = ArrayList() + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) = lockManager.withWriteLock { val txn = object : IEventStore.ITransaction { - override fun insert(event: Event) { - if (insertLocked(event)) accepted.add(event) - } + override fun insert(event: Event) = insertLocked(event) } txn.body() } - // Emit each accepted event in order after the lock releases — - // mirrors SQLiteEventStore.transaction. - for (e in accepted) _inserts.emit(e) - } // ------------------------------------------------------------------ // Query