refactor(quartz): introduce ObservableEventStore so projections see ephemerals

Splits "events accepted for persistence" from "events accepted for observation". The new ObservableEventStore wraps any IEventStore and owns events: SharedFlow<Event>: - Non-ephemeral events forward to the inner store; success → emit, any rejection (expired, NIP-09 / NIP-62 tombstone, NIP-01 supersession loser) → no emit. - Ephemeral events (kinds 20000-29999) skip persistence entirely but still emit, so an open EventStoreProjection renders them while alive. They vanish from any future seed because the DB never had them. Reverts the previous round of plumbing inside SQLiteEventStore / FsEventStore: stores no longer carry an inserts SharedFlow. IEventStore goes back to a clean read/write contract. ObservableEventStore is the single place that decides what makes it onto the projection bus. EventStore (the SQLite convenience class) embeds an ObservableEventStore internally so existing call sites like `store.observe(filter, scope)` keep working without changes. EventStoreProjection now collects via Flow.onSubscription so the collector subscription is established before the seed query runs — fixes a race where an insert immediately after `ready.await()` could land in the SharedFlow before the collector was subscribed. Tests: - ephemeralEventsAppearInProjection — kind-22000 event reaches items but isn't queryable from the inner store, and a fresh projection on the same store gets an empty seed. - 14/14 projection tests + 219/219 other store tests pass. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
2026-04-29 20:34:56 +00:00
parent c54f34ef08
commit 86537edef5
8 changed files with 293 additions and 113 deletions
@@ -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<Event>
-
    override fun close()
 }
@@ -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<Event>(
+            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<Event> = _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<Event>()
+        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 <T : Event> query(filter: Filter): List<T> = inner.query(filter)
+
+    override suspend fun <T : Event> query(filters: List<Filter>): List<T> = inner.query(filters)
+
+    override suspend fun <T : Event> query(
+        filter: Filter,
+        onEach: (T) -> Unit,
+    ) = inner.query(filter, onEach)
+
+    override suspend fun <T : Event> query(
+        filters: List<Filter>,
+        onEach: (T) -> Unit,
+    ) = inner.query(filters, onEach)
+
+    override suspend fun count(filter: Filter): Int = inner.count(filter)
+
+    override suspend fun count(filters: List<Filter>): Int = inner.count(filters)
+
+    override suspend fun delete(filter: Filter) = inner.delete(filter)
+
+    override suspend fun delete(filters: List<Filter>) = inner.delete(filters)
+
+    override suspend fun deleteExpiredEvents() = inner.deleteExpiredEvents()
+
+    override fun close() = inner.close()
+}
@@ -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<T : Event>(
-    private val store: IEventStore,
+    private val store: ObservableEventStore,
    private val filters: List<Filter>,
    private val relay: NormalizedRelayUrl?,
    scope: CoroutineScope,
@@ -138,9 +147,16 @@ class EventStoreProjection<T : Event>(

    private val collectorJob: Job =
        scope.launch {
+            // `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)
-            store.inserts.collect { event -> apply(event) }
+                }.collect { event -> apply(event) }
        }

    private val expirationJob: Job =
@@ -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 <T : Event> query(filter: Filter) = store.query<T>(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 <T : Event> observe(
        filters: List<Filter>,
        scope: CoroutineScope,
-    ): EventStoreProjection<T> = EventStoreProjection(this, filters, relay, scope)
+    ): EventStoreProjection<T> = EventStoreProjection(observable, filters, relay, scope)

    fun <T : Event> 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 <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()
+}
@@ -107,7 +107,7 @@ class SQLiteConnectionPool(
     * suspend until the lock is released. Cancellation-aware via the
     * coroutine [Mutex].
     */
-    suspend fun <T> useWriter(block: suspend (SQLiteConnection) -> T): T =
+    suspend fun <T> 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 <T> useReader(block: suspend (SQLiteConnection) -> T): T {
+    suspend fun <T> useReader(block: (SQLiteConnection) -> T): T {
        val ch =
            readerChannel
                ?: return writerMutex.withLock { block(writer) }
@@ -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<Event>(
-            replay = 0,
-            extraBufferCapacity = 256,
-            onBufferOverflow = BufferOverflow.SUSPEND,
-        )
-    val inserts: SharedFlow<Event> = _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<Event>()
-
        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)
        }
    }

@@ -359,7 +359,7 @@ class EventStoreProjectionTest {
            // Drive the ticker frequently so the test doesn't sit idle.
            val projection =
                EventStoreProjection<Event>(
-                    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<Event>(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<Event>(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<Event>(Filter(kinds = listOf(ephemeralKind)), scope)
+            freshProjection.ready.await()
+            assertTrue(freshProjection.items.value.isEmpty())
+
+            projection.close()
+            freshProjection.close()
+        }
 }
@@ -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<Event>(
-            replay = 0,
-            extraBufferCapacity = 256,
-            onBufferOverflow = BufferOverflow.SUSPEND,
-        )
-    override val inserts: SharedFlow<Event> = _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<Event>()
+    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