amethyst/cli/plans/2026-04-24-file-event-store-pipelines.md

File-backed event store for amy — Part 2: Pipelines

Status: plan · Date: 2026-04-24 · Part 2 of 3 (see 2026-04-24-file-event-store-overview.md, 2026-04-24-file-event-store-nips.md)

Covers insert, query, delete, transaction, and concurrency. NIP- specific enforcement (09/40/50/62) and replaceable/addressable slot mechanics are in Part 3.

---

Insert pipeline

Mirrors SQLiteEventStore.insertEvent() (quartz/.../sqlite/SQLiteEventStore.kt:189-196) and innerInsertEvent() (lines 178-187). Module ordering is identical — only the backing representation changes.

Pre-transaction guards (no files touched)

1. if (event.kind.isEphemeral()) return               // EphemeralModule
2. if (event.isExpired()) throw ExpiredEventException // ExpirationModule

Both match SQLite behaviour exactly — rejected before any I/O.

Transaction body

Acquire FileChannel.lock(.lock) for the whole body. Release only on COMMIT or ROLLBACK.

T1. Stage event file
    stagingDir = .staging/<random>/
    write stagingDir/event.json   (event bytes, fsync if AMY_FSYNC=1)

T2. Pre-insert veto checks (read-only)
    • TombstoneIdCheck:  tombstones/id/<event.id> exists?           → REJECT
    • TombstoneAddrCheck: (addressable only) tombstones/addr/<k>/<pk>/<hash(d)>
                          exists AND Tdel >= event.createdAt?       → REJECT
    • VanishCheck:       tombstones/vanish/<owner_hash> exists AND
                         Tvanish >= event.createdAt?                → REJECT
    • ReplaceableGuard:  (kinds 0,3,10000-19999) replaceable/<k>/<pk>.json
                         exists AND its createdAt >= event.createdAt? → REJECT
    • AddressableGuard:  (kinds 30000-39999) addressable/<k>/<pk>/<hash(d)>.json
                         exists AND its createdAt >= event.createdAt? → REJECT
    Any REJECT → discard staging dir, release lock, return.

T3. Canonical write
    canonical = events/<id[0:2]>/<id[2:4]>/<id>.json
    Files.move(stagingDir/event.json, canonical, ATOMIC_MOVE)
    Files.setLastModifiedTime(canonical, event.createdAt seconds)
    Files.setPosixFilePermissions(canonical, r--r--r--)

T4. Derived-index hardlinks (all ln(canonical → X))
    • idx/kind/<k>/<ts>-<id>
    • idx/author/<pk>/<ts>-<id>
    • idx/owner/<owner_hash>/<ts>-<id>        # recipient for GiftWrap
    • for each indexed tag (name, value):
         idx/tag/<name>/<murmur_hex>/<ts>-<id>
    • if event.expiration() > 0:
         idx/expires_at/<exp_ts>-<id>
    • if event implements SearchableEvent:
         for each token in tokenize(event.indexableContent()):
             idx/fts/<token>/<id>

T5. Replaceable / Addressable slot (if applicable)
    slotTmp = <slot>.tmp.<random>
    ln canonical slotTmp
    Files.move(slotTmp, slot, ATOMIC_MOVE, REPLACE_EXISTING)
    # The rename atomically deposes the old winner and installs new.
    if oldSlot had different event id:
         unlink events/<old_id>.json                       # drop canonical
         walk idx/... and unlink entries for old id         # reuses same logic as delete()

T6. Deletion-event side effects (if event.kind == 5)
    Apply NIP-09 actions — see Part 3. Executes *inside* the lock.

T7. Vanish-event side effects (if event.kind == 62 && shouldVanishFrom(relay))
    Apply NIP-62 cascade — see Part 3.

T8. Commit
    Release .lock. Remove emptied staging dir.

Crash safety at every step

Crash between	Post-crash state	Scrub action
T1 and T3	Only staging dir exists	Delete .staging/ on next open
T3 and T4	Canonical written, no indexes	Rebuild indexes by walking events/
T4 (partial)	Some indexes written	Rebuild indexes (idempotent)
T5 (partial)	Slot tmp hardlink exists	Remove <slot>.tmp.* on next open
T5 (after rename)	Old winner still has canonical	Slot takes precedence; old canonical GC'd by scrub
T6/T7 (partial)	Some targets unlinked, some not	Re-apply kind-5/62 during scrub

The kernel guarantees rename(2) atomicity; link(2) either succeeds fully or fails. All failure modes are forward-recoverable by scrub — never corrupting.

Ordering vs SQLite

SQLite runs triggers before our code inserts. We run checks before writing the canonical (T2), then indexes (T4), then slot enforcement (T5), then kind-5 / kind-62 side effects (T6/T7). Same overall guarantees, different sequencing because we have no trigger mechanism.

---

Transaction API

override fun transaction(body: ITransaction.() -> Unit) {
    withLock {
        val txn = FsTransaction(this)
        try {
            txn.body()           // accumulates staged events
            txn.commit()         // all T3..T7 for every staged event, then one unlock
        } catch (e: Throwable) {
            txn.rollback()       // delete staging dirs, do nothing else
            throw e
        }
    }
}

Semantics:

• Single flock for the whole transaction.
• Staging dirs collected; canonical moves happen at commit time.
• If any event in the batch is rejected by T2, the whole txn aborts.
• Matches SQLite's all-or-nothing behaviour (SQLiteConnectionExt.kt:71-81).

No readers are blocked: they don't hold the lock and see either pre-commit or post-commit state (atomic rename).

---

Query pipeline

Matches the planner structure of QueryBuilder (quartz/.../sqlite/QueryBuilder.kt:110-172) but emits directory walks instead of SQL.

Filter normalisation

Same as SQLite: filter.toFilterWithDTags() splits the d-tag out of the generic tag map, flags isSimpleQuery / isSimpleSearch.

Plan selection

For each filter, choose one driver index based on cardinality and availability:

Filter contents	Driver index	Rationale
Only ids	direct events/<aa>/<bb>/<id>.json opens	O(1) per id
kinds present, no tags	idx/kind/<k>/	Smallest per-kind fanout
authors present, no kinds, no tags	idx/author/<pk>/	Author-scoped
kinds + authors + no tags	Pick the smaller of idx/kind/ vs idx/author/ by listing size	Parity with SQLite's query_by_kind_pubkey_created choice
Any #tag filter	idx/tag/<n>/<h>/ — intersect multiple if AND	Matches SQLite's per-tag INNER JOINs
Only search	idx/fts/<token>/ — intersect tokens	Parity with FTS MATCH
Only since/until / limit	idx/kind/*/ or full events/ walk ordered by mtime	See below
d_tag present (addressable)	addressable/<k>/<pk>/<hash(d)>.json direct open	Slot lookup — O(1)

Chosen driver produces a sorted stream of (createdAt, id) pairs in DESC order (lexicographic sort over zero-padded filenames reversed). Filenames are <ts>-<id>, so Files.list(dir).sorted(reverseOrder()) is the plan.

Secondary filters (post-driver)

Anything not covered by the driver is applied in code, event by event, after opening the JSON:

• since <= createdAt <= until (already a prefix filter on the driver)
• authors / kinds / ids if not the driver
• d-tag match (addressable)
• additional tag filters
• search tokens (if driver wasn't FTS)
• tombstone exclusions (id + address)
• vanish exclusions (by owner_hash)
• expiration exclusions (createdAt < now < expiration)

Tag AND (NIP-91)

For filter.tagsAll = {#e = [a, b], #p = [c]}:

streamA = idx/tag/e/<hash(a)>/  (sorted by <ts>-<id> DESC)
streamB = idx/tag/e/<hash(b)>/
streamC = idx/tag/p/<hash(c)>/
result  = k-way intersection by <id> suffix, preserving DESC order

Implemented as a sorted-list intersection (Java Iterator<String> merge). Equivalent to SQLite's INNER JOIN per tag (QueryBuilder.kt:434-449). Bounded memory.

Multi-filter UNION

query(filters: List<Filter>) runs each filter's plan, then merges the resulting streams in DESC order with dedup by id. Per-filter limit applied before merge, as SQLite does (QueryBuilder.kt:577-583).

Streaming mode

query(filter, onEach: (T) -> Unit) calls onEach as each event is read and parsed. No intermediate list. Matches SQLite's cursor callback.

count()

Same plan, but open nothing — just count filenames that survive the post-driver code filters. For filters that require JSON inspection (tag exact match on non-indexed tags, search tokens), we still open files; cost scales with matched set size, not store size.

planQuery() output

Plain text, like EXPLAIN QUERY PLAN:

FILTER 0
  DRIVER: idx/kind/1/ (12,400 entries)
  POST: authors=[<pk>], since >= 1713960000, limit 100
FILTER 1
  DRIVER: idx/tag/e/ab12…/  ∩  idx/tag/p/cd34…/ (est 37 hits)
  POST: kinds=[30023], limit 10
MERGE: 2 filters, DESC by created_at, overall limit 110

---

Delete pipeline

delete(filter) / delete(filters)

Runs the query pipeline to enumerate matching events, then for each:

1. Read canonical, parse tags (needed to know all hardlink paths).
2. Unlink every derived hardlink:
   idx/kind/<k>/<ts>-<id>
   idx/author/<pk>/<ts>-<id>
   idx/owner/<owner_hash>/<ts>-<id>
   idx/tag/<n>/<h>/<ts>-<id>  (for each indexed tag)
   idx/expires_at/<exp>-<id>  (if exp present)
   idx/fts/<token>/<id>        (for each FTS token)
3. If event is the winner of a replaceable/addressable slot:
   unlink the slot path.
4. Unlink canonical events/<aa>/<bb>/<id>.json.
5. kernel GC's the inode (no other refs).

All inside flock. Matches SQLite's DELETE FROM event_headers WHERE row_id IN (...) with FK cascade.

delete(id: HexKey): Int

Direct shortcut. Opens events/<aa>/<bb>/<id>.json, runs the same steps 1-5. Returns 1 on success, 0 if the file didn't exist — matching the SQLite return value from changes() (SQLiteEventStore.kt:261-264).

deleteExpiredEvents()

now = clock.now().epochSecond
for entry in idx/expires_at/ where <exp> < now:
    delete(entry.id)

Optional optimisation: because filenames are <exp>-<id>, listing is already sorted, so we can stop at the first entry with <exp> >= now.

---

Concurrency model

Single writer, many readers

• Writers hold an exclusive flock(.lock) via FileChannel.tryLock().
• Readers take a shared lock only while listing top-level directory contents that may be mid-rename; most reads hold no lock (kernel- atomic open / readdir is enough).
• Two concurrent amy invocations serialise cleanly on the flock; the second one blocks on FileChannel.lock() (blocking) or fails fast with tryLock() (configurable).

Reader–writer interaction

Because every writer mutation is a rename-over-tmp, readers see either the pre-mutation file or the post-mutation file, never a torn view. Dangling reads (writer unlinks a file between readdir and open) manifest as NoSuchFileException and are treated as "event not in store anymore" — we skip it and keep going.

Watch / subscribe (future)

java.nio.file.WatchService on events/ gives per-platform inotify/FSEvents/ReadDirectoryChangesW. Lets a future amy store tail stream newly arriving events to stdout without polling. Not part of this plan but the layout supports it natively.

---

Close / reopen

• close() releases any held FileChannel and flushes nothing (writes are already durable as of each rename, modulo the AMY_FSYNC option).
• Reopen is free: the store is stateless. No header page, no WAL checkpoint. First action after open: scan .staging/ and delete leftovers (crash from previous run).

---

Next: NIP specifics and enforcement details (2026-04-24-file-event-store-nips.md).