feat(quartz): FsEventStore flock + transactions + scrub/compact (step 8)

- FsLockManager: cross-process exclusive flock(.lock) with per-thread
  re-entry. withWriteLock { body } acquires once on a fresh thread
  and reuses on nested calls — so transaction { insert(); insert() }
  doesn't self-deadlock.
- FsEventStore: insert / delete / delete(filter) / delete(filters) /
  delete(id) / deleteExpiredEvents / transaction now run under
  withWriteLock. The `*Locked` helpers expose the lock-free body for
  re-entrant callers (transaction body, vanish/deletion cascades,
  expiration sweep). close() releases the lock channel.
- scrub(): wipes idx/ and rebuilds every entry from the canonical
  events. Slots, tombstones and seed are left alone — slots can pin
  data the canonical pass doesn't see, and tombstone removal is a
  deliberate "un-forget" per the design plan.
- compact(): drops dangling idx/ entries whose canonical no longer
  exists. Cheap — only touches idx/, never opens a JSON.

Tests: 11 new in FsMaintenanceTest covering lock file presence,
transaction commit / propagated exception with kept-prior-events
semantics, re-entrant lock from inside a transaction, scrub
rebuilding idx + FTS after a manual wipe, scrub preserving the
replaceable slot, compact dropping dangling and leaving valid alone,
close idempotence + reopen, and two-thread concurrent insert
serialisation. 97 fs tests green.

quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt

@@ -79,10 +79,12 @@ class FsEventStore(
     private val slots: FsSlots
     private val tombstones: FsTombstones
     private val planner: FsQueryPlanner
+    private val lockManager: FsLockManager
 
     init {
         layout.ensureSkeleton()
         cleanStaging()
+        lockManager = FsLockManager(root)
         hasher = TagNameValueHasher(layout.readOrCreateSeed())
         indexer = FsIndexer(layout, hasher, indexingStrategy)
         slots = FsSlots(layout, indexer)
@@ -94,7 +96,12 @@ class FsEventStore(
     // Insert
     // ------------------------------------------------------------------
 
-    override fun insert(event: Event) {
+    override fun insert(event: Event) =
+        lockManager.withWriteLock {
+            insertLocked(event)
+        }
+
+    private fun insertLocked(event: Event) {
         if (event.kind.isEphemeral()) return
         if (isAlreadyExpired(event)) return
         if (isBlockedByTombstone(event)) return
@@ -231,13 +238,14 @@ class FsEventStore(
         }
     }
 
-    override fun transaction(body: IEventStore.ITransaction.() -> Unit) {
-        val txn =
-            object : IEventStore.ITransaction {
-                override fun insert(event: Event) = this@FsEventStore.insert(event)
-            }
-        txn.body()
-    }
+    override fun transaction(body: IEventStore.ITransaction.() -> Unit) =
+        lockManager.withWriteLock {
+            val txn =
+                object : IEventStore.ITransaction {
+                    override fun insert(event: Event) = insertLocked(event)
+                }
+            txn.body()
+        }
 
     // ------------------------------------------------------------------
     // Query
@@ -304,20 +312,27 @@ class FsEventStore(
     // Delete
     // ------------------------------------------------------------------
 
-    override fun delete(filter: Filter) {
-        val ids = ArrayList<HexKey>()
-        query<Event>(filter) { ids.add(it.id) }
-        ids.forEach { delete(it) }
-    }
+    override fun delete(filter: Filter) =
+        lockManager.withWriteLock {
+            val ids = ArrayList<HexKey>()
+            query<Event>(filter) { ids.add(it.id) }
+            ids.forEach { deleteLocked(it) }
+        }
 
-    override fun delete(filters: List<Filter>) {
-        val ids = HashSet<HexKey>()
-        query<Event>(filters) { ids.add(it.id) }
-        ids.forEach { delete(it) }
-    }
+    override fun delete(filters: List<Filter>) =
+        lockManager.withWriteLock {
+            val ids = HashSet<HexKey>()
+            query<Event>(filters) { ids.add(it.id) }
+            ids.forEach { deleteLocked(it) }
+        }
 
     /** Delete an event by id. Returns 1 if a file was removed, 0 otherwise. */
-    fun delete(id: HexKey): Int {
+    fun delete(id: HexKey): Int =
+        lockManager.withWriteLock {
+            deleteLocked(id)
+        }
+
+    private fun deleteLocked(id: HexKey): Int {
         val canonical = layout.canonical(id)
         val event = readEvent(id) // need tags to know which index links to remove
         if (event != null) {
@@ -366,21 +381,83 @@ class FsEventStore(
      * filenames, and deletes any entry whose `exp < now`. Matches SQLite's
      * `expiration < unixepoch()` predicate (note: strict `<`, not `<=`).
      */
-    override fun deleteExpiredEvents() {
-        if (!Files.isDirectory(layout.idxExpiresAt)) return
-        val now = clock()
-        val toDelete = ArrayList<HexKey>()
-        Files.list(layout.idxExpiresAt).use { stream ->
-            for (entry in stream) {
-                val parsed = FsLayout.parseEntry(entry.fileName.toString()) ?: continue
-                if (parsed.first < now) toDelete.add(parsed.second)
+    override fun deleteExpiredEvents() =
+        lockManager.withWriteLock {
+            if (!Files.isDirectory(layout.idxExpiresAt)) return@withWriteLock
+            val now = clock()
+            val toDelete = ArrayList<HexKey>()
+            Files.list(layout.idxExpiresAt).use { stream ->
+                for (entry in stream) {
+                    val parsed = FsLayout.parseEntry(entry.fileName.toString()) ?: continue
+                    if (parsed.first < now) toDelete.add(parsed.second)
+                }
             }
+            toDelete.forEach { deleteLocked(it) }
         }
-        toDelete.forEach { delete(it) }
-    }
 
     override fun close() {
-        // No long-lived resources yet.
+        lockManager.close()
+    }
+
+    // ------------------------------------------------------------------
+    // Maintenance
+    // ------------------------------------------------------------------
+
+    /**
+     * Rebuild every `idx/` entry from the canonical events. Safe to run
+     * after external edits or to recover from a partial-write crash.
+     * Tombstones, replaceable / addressable slots, and the seed file are
+     * left untouched — slot-only-pinned events stay reachable, and
+     * tombstone removal is treated as a deliberate "un-forget" by the
+     * user.
+     */
+    fun scrub() =
+        lockManager.withWriteLock {
+            cleanStaging()
+            // Wipe and recreate idx/.
+            deleteRecursively(layout.idx)
+            layout.ensureSkeleton()
+
+            if (!Files.isDirectory(layout.events)) return@withWriteLock
+            Files.walk(layout.events).use { stream ->
+                for (path in stream) {
+                    if (!Files.isRegularFile(path)) continue
+                    if (!path.fileName.toString().endsWith(FsLayout.JSON_EXT)) continue
+                    val event =
+                        try {
+                            Event.fromJson(Files.readString(path))
+                        } catch (_: Exception) {
+                            continue
+                        }
+                    indexer.link(event, path)
+                }
+            }
+        }
+
+    /**
+     * Drop dangling `idx/` entries whose canonical no longer exists.
+     * Cheaper than [scrub] because it never opens an event JSON; it just
+     * stats the canonical path encoded in the index entry filename.
+     */
+    fun compact() =
+        lockManager.withWriteLock {
+            if (!Files.isDirectory(layout.idx)) return@withWriteLock
+            Files.walk(layout.idx).use { stream ->
+                for (path in stream) {
+                    if (!Files.isRegularFile(path)) continue
+                    val parsed = FsLayout.parseEntry(path.fileName.toString()) ?: continue
+                    if (!layout.canonical(parsed.second).exists()) {
+                        Files.deleteIfExists(path)
+                    }
+                }
+            }
+        }
+
+    private fun deleteRecursively(p: java.nio.file.Path) {
+        if (!Files.exists(p)) return
+        Files.walk(p).use { stream ->
+            stream.sorted(Comparator.reverseOrder()).forEach { Files.deleteIfExists(it) }
+        }
     }
 
     // ------------------------------------------------------------------

quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsLockManager.kt

@@ -0,0 +1,122 @@
+/*
+ * 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.fs
+
+import java.nio.channels.FileChannel
+import java.nio.channels.FileLock
+import java.nio.file.Files
+import java.nio.file.Path
+import java.nio.file.StandardOpenOption
+
+/**
+ * Cross-process write serialisation for the file-backed event store.
+ *
+ * Acquires an exclusive `flock(2)` on `.lock` while the body runs, so a
+ * second `amy` invocation against the same directory blocks until the
+ * first releases. Re-entrant for the calling thread — nested
+ * `withWriteLock` calls (e.g. transaction body that calls insert) reuse
+ * the existing lock instead of self-deadlocking.
+ *
+ * Readers do not take this lock. The store's atomic-rename writes mean
+ * readers see either the pre- or post-mutation state, never a torn
+ * file. Stale `idx/` entries that point at a just-unlinked canonical
+ * are tolerated by the query loop's `NoSuchFileException` handling.
+ */
+internal class FsLockManager(
+    root: Path,
+) : AutoCloseable {
+    private val lockPath: Path = root.resolve(LOCK_FILE)
+    private val mu = Any()
+
+    /** Per-thread re-entry depth. Allows nested `withWriteLock` calls. */
+    private val depth = ThreadLocal.withInitial { 0 }
+
+    /** Active channel + lock when depth > 0 (any thread). Held by whoever owns the lock. */
+    private var channel: FileChannel? = null
+    private var lock: FileLock? = null
+
+    init {
+        try {
+            Files.createFile(lockPath)
+        } catch (_: java.nio.file.FileAlreadyExistsException) {
+            // existing lock file from a previous run — fine
+        }
+    }
+
+    fun <T> withWriteLock(body: () -> T): T {
+        // Re-entrant on the same thread: just run.
+        if (depth.get() > 0) {
+            depth.set(depth.get() + 1)
+            try {
+                return body()
+            } finally {
+                depth.set(depth.get() - 1)
+            }
+        }
+
+        // Cross-thread + cross-process serialisation.
+        synchronized(mu) {
+            val ch = FileChannel.open(lockPath, StandardOpenOption.READ, StandardOpenOption.WRITE)
+            val l = ch.lock() // exclusive, blocking
+            channel = ch
+            lock = l
+            depth.set(1)
+            try {
+                return body()
+            } finally {
+                depth.set(0)
+                try {
+                    l.release()
+                } catch (_: Throwable) {
+                    // ignore
+                }
+                try {
+                    ch.close()
+                } catch (_: Throwable) {
+                    // ignore
+                }
+                channel = null
+                lock = null
+            }
+        }
+    }
+
+    override fun close() {
+        synchronized(mu) {
+            try {
+                lock?.release()
+            } catch (_: Throwable) {
+                // ignore
+            }
+            try {
+                channel?.close()
+            } catch (_: Throwable) {
+                // ignore
+            }
+            channel = null
+            lock = null
+        }
+    }
+
+    companion object {
+        const val LOCK_FILE = ".lock"
+    }
+}

quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsMaintenanceTest.kt

@@ -0,0 +1,258 @@
+/*
+ * 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.fs
+
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
+import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
+import com.vitorpamplona.quartz.utils.Secp256k1Instance
+import java.nio.file.Files
+import java.nio.file.Path
+import kotlin.io.path.exists
+import kotlin.io.path.listDirectoryEntries
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+class FsMaintenanceTest {
+    private val signer = NostrSignerSync()
+    private lateinit var root: Path
+    private lateinit var store: FsEventStore
+
+    @BeforeTest
+    fun setup() {
+        Secp256k1Instance
+        root = Files.createTempDirectory("fs-maint-")
+        store = FsEventStore(root)
+    }
+
+    @AfterTest
+    fun tearDown() {
+        store.close()
+        if (root.exists()) {
+            Files.walk(root).use { it.sorted(Comparator.reverseOrder()).forEach { p -> Files.deleteIfExists(p) } }
+        }
+    }
+
+    private fun note(
+        body: String,
+        ts: Long,
+    ) = signer.sign<TextNoteEvent>(TextNoteEvent.build(body, createdAt = ts))
+
+    // ------------------------------------------------------------------
+    // Lock file
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `lock file is created on open`() {
+        assertTrue(root.resolve(".lock").exists())
+    }
+
+    // ------------------------------------------------------------------
+    // Transactions
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `transaction commits all inserts on success`() {
+        val a = note("a", 1)
+        val b = note("b", 2)
+        val c = note("c", 3)
+
+        store.transaction {
+            insert(a)
+            insert(b)
+            insert(c)
+        }
+
+        val got = store.query<TextNoteEvent>(Filter(authors = listOf(signer.pubKey)))
+        assertEquals(setOf(a.id, b.id, c.id), got.map { it.id }.toSet())
+    }
+
+    @Test
+    fun `transaction propagates exceptions and stops processing`() {
+        val a = note("a", 1)
+        val b = note("b", 2)
+        val c = note("c", 3)
+
+        assertFailsWith<IllegalStateException> {
+            store.transaction {
+                insert(a)
+                insert(b)
+                throw IllegalStateException("boom")
+                // unreachable
+                @Suppress("UNREACHABLE_CODE")
+                insert(c)
+            }
+        }
+
+        // Events written before the throw are kept (per the plan: atomic-
+        // per-event, serialised across writers — not all-or-nothing).
+        assertTrue(store.count(Filter(ids = listOf(a.id))) == 1)
+        assertTrue(store.count(Filter(ids = listOf(b.id))) == 1)
+        assertTrue(store.count(Filter(ids = listOf(c.id))) == 0)
+    }
+
+    @Test
+    fun `transaction is re-entrant on the same thread`() {
+        val a = note("a", 1)
+        // If flock were non-reentrant we'd self-deadlock here because
+        // insert() acquires the same lock the transaction already holds.
+        store.transaction {
+            insert(a)
+            // Call an outer-locking method from within the transaction.
+            store.deleteExpiredEvents()
+        }
+        assertEquals(1, store.count(Filter(ids = listOf(a.id))))
+    }
+
+    // ------------------------------------------------------------------
+    // scrub — rebuild idx/ from canonical
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `scrub rebuilds idx entries after a manual wipe`() {
+        val a = note("hello bitcoin", 10)
+        val b = note("nostr stuff", 20)
+        store.insert(a)
+        store.insert(b)
+
+        // Blow away the entire idx/ tree behind the store's back.
+        Files.walk(root.resolve("idx")).use {
+            it.sorted(Comparator.reverseOrder()).forEach { p -> Files.deleteIfExists(p) }
+        }
+
+        // Without scrub, index-driven queries find nothing.
+        assertEquals(emptyList(), store.query<TextNoteEvent>(Filter(authors = listOf(signer.pubKey))).map { it.id })
+
+        store.scrub()
+
+        val got = store.query<TextNoteEvent>(Filter(authors = listOf(signer.pubKey))).map { it.id }.toSet()
+        assertEquals(setOf(a.id, b.id), got)
+
+        // FTS recovered too.
+        assertEquals(listOf(a.id), store.query<TextNoteEvent>(Filter(search = "bitcoin")).map { it.id })
+    }
+
+    @Test
+    fun `scrub leaves replaceable slot intact`() {
+        // Replaceable slots pin events via hardlink even without the
+        // canonical. Scrub must not wipe slots.
+        val meta =
+            signer.sign<com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent>(
+                createdAt = 10,
+                kind = 0,
+                tags = emptyArray(),
+                content = "{}",
+            )
+        store.insert(meta)
+        val slot = root.resolve("replaceable/0/${signer.pubKey}.json")
+        assertTrue(slot.exists())
+
+        store.scrub()
+        assertTrue(slot.exists(), "replaceable slot must survive scrub")
+    }
+
+    // ------------------------------------------------------------------
+    // compact — drop dangling idx entries
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `compact drops idx entries whose canonical is gone`() {
+        val a = note("x", 10)
+        store.insert(a)
+
+        // Externally delete the canonical without touching idx/.
+        val canonical = root.resolve("events/${a.id.substring(0, 2)}/${a.id.substring(2, 4)}/${a.id}.json")
+        assertTrue(Files.deleteIfExists(canonical))
+
+        val kindDir = root.resolve("idx/kind/1")
+        assertEquals(1, kindDir.listDirectoryEntries().size, "dangling entry still present pre-compact")
+
+        store.compact()
+
+        assertEquals(0, kindDir.listDirectoryEntries().size, "dangling entry dropped post-compact")
+    }
+
+    @Test
+    fun `compact leaves valid entries alone`() {
+        val a = note("x", 10)
+        store.insert(a)
+
+        store.compact()
+
+        val kindDir = root.resolve("idx/kind/1")
+        assertEquals(1, kindDir.listDirectoryEntries().size, "valid entry should not be touched")
+        assertEquals(listOf(a.id), store.query<TextNoteEvent>(Filter(ids = listOf(a.id))).map { it.id })
+    }
+
+    // ------------------------------------------------------------------
+    // close
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `close is idempotent`() {
+        store.close()
+        store.close()
+    }
+
+    @Test
+    fun `reopen after close works`() {
+        val a = note("a", 1)
+        store.insert(a)
+        store.close()
+
+        val reopened = FsEventStore(root)
+        try {
+            assertEquals(listOf(a.id), reopened.query<TextNoteEvent>(Filter(ids = listOf(a.id))).map { it.id })
+        } finally {
+            reopened.close()
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // Concurrency — two writer threads serialise cleanly
+    // ------------------------------------------------------------------
+
+    @Test
+    fun `concurrent inserts on two threads are both persisted`() {
+        val events = (1..20).map { note("n$it", it.toLong()) }
+        val half = events.size / 2
+
+        val t1 =
+            Thread {
+                events.take(half).forEach { store.insert(it) }
+            }
+        val t2 =
+            Thread {
+                events.drop(half).forEach { store.insert(it) }
+            }
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+
+        val got = store.query<TextNoteEvent>(Filter(authors = listOf(signer.pubKey))).map { it.id }.toSet()
+        assertEquals(events.map { it.id }.toSet(), got)
+    }
+}