feat(quartz): FsEventStore indexes + query planner (step 2)
Hardlink indexes under idx/ and a minimal query planner bring the file-backed store up to parity with SQLite on filtered lookups. - FsLayout — path helpers, .seed file (8 random bytes, salts all hashes), entry filename format <zero-padded ts>-<id>. - FsIndexer — on insert creates hardlinks at idx/kind/<k>/, idx/author/ <pk>/, idx/owner/<owner_hex>/, idx/tag/<name>/<hash_hex>/. Owner hash matches SQLite's pubkey_owner_hash (recipient for GiftWrap). Honours DefaultIndexingStrategy: single-letter tag names only. On delete unlinks every known path so the inode can be reclaimed. - FsQueryPlanner — picks a driver (ids / first tag / kinds / authors / all kinds) and yields candidates sorted by createdAt DESC. Final predicate check runs through Filter.match so any driver is correctness-safe. - FsEventStore — sets mtime to event.createdAt on write, runs queries through the planner with post-filter, applies limit, dedupes by id across multi-filter unions, and unlinks indexes on delete. Tests: 16 new in FsQueryTest covering order, limit, author / kind / tag drivers, tag OR within a key, tagsAll AND across keys, non- single-letter-tag behaviour, since/until, count, index hardlink maintenance, and reopen persistence. All 26 fs tests green.
This commit is contained in:
Claude
+60
-41
@@ -25,34 +25,45 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
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.nip01Core.store.sqlite.DefaultIndexingStrategy
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.IndexingStrategy
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.TagNameValueHasher
|
||||
import java.nio.file.FileAlreadyExistsException
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.StandardCopyOption
|
||||
import java.nio.file.attribute.FileTime
|
||||
import java.util.concurrent.TimeUnit
|
||||
import kotlin.io.path.deleteIfExists
|
||||
import kotlin.io.path.exists
|
||||
import kotlin.io.path.readText
|
||||
|
||||
/**
|
||||
* Filesystem-backed `IEventStore`. Each event is stored as a JSON file under
|
||||
* `events/<aa>/<bb>/<id>.json` where `<aa><bb>` is the first 4 hex characters
|
||||
* of the event id. Writes are atomic (tmp + rename).
|
||||
* Filesystem-backed `IEventStore`. Each event is stored as a JSON file at
|
||||
* `events/<aa>/<bb>/<id>.json` (atomic tmp+rename) with `mtime` set to
|
||||
* `event.createdAt`. Hardlink indexes under `idx/` make filtered queries
|
||||
* cheap — see [FsIndexer] and [FsQueryPlanner].
|
||||
*
|
||||
* This is the step-1 skeleton: only `insert`, `delete(id)`, and a minimal
|
||||
* `query` by event id. Indexes, tombstones, slots, FTS, vanish, expiration
|
||||
* sweep, and transactions arrive in later steps — see
|
||||
* `cli/plans/2026-04-24-file-event-store-*.md`.
|
||||
* Step 2 scope: kind / author / owner / tag indexes; query planner with
|
||||
* post-filter via `FilterMatcher`; `count`. Later steps: replaceable /
|
||||
* addressable slots, NIP-09 tombstones, NIP-40 expiration, NIP-50 FTS,
|
||||
* NIP-62 vanish, transactions, scrub.
|
||||
*/
|
||||
class FsEventStore(
|
||||
val root: Path,
|
||||
indexingStrategy: IndexingStrategy = DefaultIndexingStrategy(),
|
||||
) : IEventStore {
|
||||
private val eventsDir: Path = root.resolve(EVENTS_DIR)
|
||||
private val stagingDir: Path = root.resolve(STAGING_DIR)
|
||||
private val layout = FsLayout(root)
|
||||
private val hasher: TagNameValueHasher
|
||||
private val indexer: FsIndexer
|
||||
private val planner: FsQueryPlanner
|
||||
|
||||
init {
|
||||
Files.createDirectories(eventsDir)
|
||||
Files.createDirectories(stagingDir)
|
||||
layout.ensureSkeleton()
|
||||
cleanStaging()
|
||||
hasher = TagNameValueHasher(layout.readOrCreateSeed())
|
||||
indexer = FsIndexer(layout, hasher, indexingStrategy)
|
||||
planner = FsQueryPlanner(layout, hasher)
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
@@ -62,21 +73,23 @@ class FsEventStore(
|
||||
override fun insert(event: Event) {
|
||||
if (event.kind.isEphemeral()) return
|
||||
|
||||
val canonical = canonicalPath(event.id)
|
||||
val canonical = layout.canonical(event.id)
|
||||
if (canonical.exists()) return
|
||||
|
||||
Files.createDirectories(canonical.parent)
|
||||
val tmp = Files.createTempFile(stagingDir, event.id, JSON_EXT)
|
||||
val tmp = Files.createTempFile(layout.staging, event.id, FsLayout.JSON_EXT)
|
||||
try {
|
||||
Files.writeString(tmp, event.toJson())
|
||||
try {
|
||||
Files.move(tmp, canonical, StandardCopyOption.ATOMIC_MOVE)
|
||||
} catch (_: FileAlreadyExistsException) {
|
||||
// Racing writer installed the same id first. Canonical is
|
||||
// immutable — this is a no-op, matching SQLite's unique
|
||||
// constraint on event.id.
|
||||
// A concurrent writer won the race. Canonical is immutable
|
||||
// so the other copy is equivalent — drop our tmp and return.
|
||||
Files.deleteIfExists(tmp)
|
||||
return
|
||||
}
|
||||
Files.setLastModifiedTime(canonical, FileTime.from(event.createdAt, TimeUnit.SECONDS))
|
||||
indexer.link(event, canonical)
|
||||
} catch (t: Throwable) {
|
||||
Files.deleteIfExists(tmp)
|
||||
throw t
|
||||
@@ -92,7 +105,7 @@ class FsEventStore(
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Query (by id only — full planner arrives in step 2)
|
||||
// Query
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
@@ -115,10 +128,18 @@ class FsEventStore(
|
||||
filter: Filter,
|
||||
onEach: (T) -> Unit,
|
||||
) {
|
||||
val ids = filter.ids ?: return // step-1: only id lookups are implemented
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
ids.forEach { id ->
|
||||
readEvent(id)?.let { onEach(it as T) }
|
||||
val limit = filter.limit ?: Int.MAX_VALUE
|
||||
if (limit <= 0) return
|
||||
var emitted = 0
|
||||
val seenIds = HashSet<HexKey>()
|
||||
for (candidate in planner.plan(filter)) {
|
||||
if (emitted >= limit) break
|
||||
if (!seenIds.add(candidate.id)) continue
|
||||
val event = readEvent(candidate.id) ?: continue
|
||||
if (!filter.match(event)) continue
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
onEach(event as T)
|
||||
emitted++
|
||||
}
|
||||
}
|
||||
|
||||
@@ -149,52 +170,50 @@ class FsEventStore(
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
override fun delete(filter: Filter) {
|
||||
val ids = filter.ids ?: return
|
||||
val ids = ArrayList<HexKey>()
|
||||
query<Event>(filter) { ids.add(it.id) }
|
||||
ids.forEach { delete(it) }
|
||||
}
|
||||
|
||||
override fun delete(filters: List<Filter>) = filters.forEach(::delete)
|
||||
override fun delete(filters: List<Filter>) {
|
||||
val ids = HashSet<HexKey>()
|
||||
query<Event>(filters) { ids.add(it.id) }
|
||||
ids.forEach { delete(it) }
|
||||
}
|
||||
|
||||
/** Delete an event by id. Returns 1 if a file was removed, 0 otherwise. */
|
||||
fun delete(id: HexKey): Int = if (canonicalPath(id).deleteIfExists()) 1 else 0
|
||||
fun delete(id: HexKey): Int {
|
||||
val canonical = layout.canonical(id)
|
||||
val event = readEvent(id) // need tags to know which index links to remove
|
||||
if (event != null) indexer.unlink(event)
|
||||
return if (canonical.deleteIfExists()) 1 else 0
|
||||
}
|
||||
|
||||
override fun deleteExpiredEvents() {
|
||||
// Step-5 feature. No-op in skeleton — queries do not yet surface
|
||||
// expired events either, so nothing observable changes.
|
||||
// Step-5 feature.
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
// No long-lived resources yet. The lock channel arrives in step 8.
|
||||
// No long-lived resources yet.
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Internals
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
private fun canonicalPath(id: HexKey): Path {
|
||||
require(id.length >= 4) { "event id must be at least 4 hex chars, got '$id'" }
|
||||
return eventsDir.resolve(id.substring(0, 2)).resolve(id.substring(2, 4)).resolve("$id$JSON_EXT")
|
||||
}
|
||||
|
||||
private fun readEvent(id: HexKey): Event? {
|
||||
val p = canonicalPath(id)
|
||||
val p = layout.canonical(id)
|
||||
if (!p.exists()) return null
|
||||
return try {
|
||||
Event.fromJson(p.readText())
|
||||
} catch (_: java.nio.file.NoSuchFileException) {
|
||||
null // file was removed between exists() and read — treat as absent
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
private fun cleanStaging() {
|
||||
Files.list(stagingDir).use { stream ->
|
||||
Files.list(layout.staging).use { stream ->
|
||||
stream.forEach { Files.deleteIfExists(it) }
|
||||
}
|
||||
}
|
||||
|
||||
companion object {
|
||||
const val EVENTS_DIR = "events"
|
||||
const val STAGING_DIR = ".staging"
|
||||
const val JSON_EXT = ".json"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
/*
|
||||
* 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.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.IndexingStrategy
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.TagNameValueHasher
|
||||
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
|
||||
import java.nio.file.FileAlreadyExistsException
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.Path
|
||||
import kotlin.io.path.deleteIfExists
|
||||
|
||||
/**
|
||||
* Writes and removes the hardlink indexes that sit alongside canonical
|
||||
* event files. Every entry is a hardlink into the canonical, so an
|
||||
* `ln(2)` adds one directory entry and no new inode; `unlink(2)` drops
|
||||
* one reference and the kernel GCs the inode once all links are gone.
|
||||
*
|
||||
* Index trees maintained:
|
||||
* idx/kind/<k>/<ts>-<id>
|
||||
* idx/author/<pk>/<ts>-<id>
|
||||
* idx/owner/<owner_hex>/<ts>-<id>
|
||||
* idx/tag/<name>/<hash_hex>/<ts>-<id>
|
||||
*
|
||||
* `owner` matches SQLite's `pubkey_owner_hash` — event pubkey for normal
|
||||
* events, recipient for GiftWrap — and is consumed by the NIP-09 /
|
||||
* NIP-62 cascades in later steps.
|
||||
*/
|
||||
internal class FsIndexer(
|
||||
private val layout: FsLayout,
|
||||
private val hasher: TagNameValueHasher,
|
||||
private val indexingStrategy: IndexingStrategy = DefaultIndexingStrategy(),
|
||||
) {
|
||||
/** Paths every index link lives at for a given event. */
|
||||
fun pathsFor(event: Event): List<Path> {
|
||||
val out = ArrayList<Path>(8 + event.tags.size)
|
||||
out.add(layout.kindEntry(event.kind, event.createdAt, event.id))
|
||||
out.add(layout.authorEntry(event.pubKey, event.createdAt, event.id))
|
||||
out.add(layout.ownerEntry(ownerHash(event), event.createdAt, event.id))
|
||||
for (tag in event.tags) {
|
||||
if (!indexingStrategy.shouldIndex(event.kind, tag)) continue
|
||||
val h = hasher.hash(tag[0], tag[1])
|
||||
out.add(layout.tagEntry(tag[0], h, event.createdAt, event.id))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** Create every index hardlink for this event. Idempotent. */
|
||||
fun link(
|
||||
event: Event,
|
||||
canonical: Path,
|
||||
) {
|
||||
for (path in pathsFor(event)) {
|
||||
createLink(path, canonical)
|
||||
}
|
||||
}
|
||||
|
||||
/** Remove every index hardlink for this event. Missing entries ignored. */
|
||||
fun unlink(event: Event) {
|
||||
for (path in pathsFor(event)) {
|
||||
path.deleteIfExists()
|
||||
}
|
||||
}
|
||||
|
||||
fun ownerHash(event: Event): Long =
|
||||
if (event is GiftWrapEvent) {
|
||||
event.recipientPubKey()?.let { hasher.hash(it) } ?: hasher.hash(event.pubKey)
|
||||
} else {
|
||||
hasher.hash(event.pubKey)
|
||||
}
|
||||
|
||||
private fun createLink(
|
||||
link: Path,
|
||||
target: Path,
|
||||
) {
|
||||
Files.createDirectories(link.parent)
|
||||
try {
|
||||
Files.createLink(link, target)
|
||||
} catch (_: FileAlreadyExistsException) {
|
||||
// Another insert raced us to the same (ts, id) slot. Since
|
||||
// filenames are canonical and events are immutable, the
|
||||
// existing entry is equivalent — no-op.
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
/*
|
||||
* 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.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Kind
|
||||
import java.nio.ByteBuffer
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.StandardCopyOption
|
||||
import java.security.SecureRandom
|
||||
import kotlin.io.path.exists
|
||||
|
||||
/**
|
||||
* Pure path helpers for the file-backed event store. No I/O except the
|
||||
* seed-file reader, which lives here because the seed is a layout concern
|
||||
* (it salts every tag / owner hash).
|
||||
*/
|
||||
internal class FsLayout(
|
||||
val root: Path,
|
||||
) {
|
||||
val events: Path = root.resolve(EVENTS_DIR)
|
||||
val staging: Path = root.resolve(STAGING_DIR)
|
||||
val idx: Path = root.resolve(IDX_DIR)
|
||||
val idxKind: Path = idx.resolve(IDX_KIND)
|
||||
val idxAuthor: Path = idx.resolve(IDX_AUTHOR)
|
||||
val idxOwner: Path = idx.resolve(IDX_OWNER)
|
||||
val idxTag: Path = idx.resolve(IDX_TAG)
|
||||
|
||||
fun canonical(id: HexKey): Path {
|
||||
require(id.length >= 4) { "event id must be at least 4 hex chars, got '$id'" }
|
||||
return events.resolve(id.substring(0, 2)).resolve(id.substring(2, 4)).resolve("$id$JSON_EXT")
|
||||
}
|
||||
|
||||
fun kindEntry(
|
||||
kind: Kind,
|
||||
ts: Long,
|
||||
id: HexKey,
|
||||
): Path = idxKind.resolve(kind.toString()).resolve(entryName(ts, id))
|
||||
|
||||
fun authorEntry(
|
||||
pubkey: HexKey,
|
||||
ts: Long,
|
||||
id: HexKey,
|
||||
): Path = idxAuthor.resolve(pubkey).resolve(entryName(ts, id))
|
||||
|
||||
fun ownerEntry(
|
||||
ownerHash: Long,
|
||||
ts: Long,
|
||||
id: HexKey,
|
||||
): Path = idxOwner.resolve(hashHex(ownerHash)).resolve(entryName(ts, id))
|
||||
|
||||
fun tagEntry(
|
||||
name: String,
|
||||
valueHash: Long,
|
||||
ts: Long,
|
||||
id: HexKey,
|
||||
): Path = idxTag.resolve(name).resolve(hashHex(valueHash)).resolve(entryName(ts, id))
|
||||
|
||||
/** Directory that holds every indexed value for a tag name. */
|
||||
fun tagDir(name: String): Path = idxTag.resolve(name)
|
||||
|
||||
/** Directory that holds every entry for a specific (tag name, value hash). */
|
||||
fun tagValueDir(
|
||||
name: String,
|
||||
valueHash: Long,
|
||||
): Path = idxTag.resolve(name).resolve(hashHex(valueHash))
|
||||
|
||||
fun kindDir(kind: Kind): Path = idxKind.resolve(kind.toString())
|
||||
|
||||
fun authorDir(pubkey: HexKey): Path = idxAuthor.resolve(pubkey)
|
||||
|
||||
fun ensureSkeleton() {
|
||||
Files.createDirectories(events)
|
||||
Files.createDirectories(staging)
|
||||
Files.createDirectories(idxKind)
|
||||
Files.createDirectories(idxAuthor)
|
||||
Files.createDirectories(idxOwner)
|
||||
Files.createDirectories(idxTag)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the seed salt (creates it on first call). The seed is
|
||||
* write-once — reopening an existing store must return the same Long
|
||||
* or every previously-written index entry becomes unreachable.
|
||||
*/
|
||||
fun readOrCreateSeed(): Long {
|
||||
val f = root.resolve(SEED_FILE)
|
||||
if (!f.exists()) {
|
||||
val bytes = ByteArray(8)
|
||||
SecureRandom().nextBytes(bytes)
|
||||
val tmp = Files.createTempFile(root, ".seed-", ".tmp")
|
||||
Files.write(tmp, bytes)
|
||||
Files.move(tmp, f, StandardCopyOption.ATOMIC_MOVE)
|
||||
}
|
||||
val bytes = Files.readAllBytes(f)
|
||||
require(bytes.size == 8) { "$SEED_FILE must be exactly 8 bytes, got ${bytes.size}" }
|
||||
return ByteBuffer.wrap(bytes).long
|
||||
}
|
||||
|
||||
companion object {
|
||||
const val EVENTS_DIR = "events"
|
||||
const val STAGING_DIR = ".staging"
|
||||
const val IDX_DIR = "idx"
|
||||
const val IDX_KIND = "kind"
|
||||
const val IDX_AUTHOR = "author"
|
||||
const val IDX_OWNER = "owner"
|
||||
const val IDX_TAG = "tag"
|
||||
const val SEED_FILE = ".seed"
|
||||
const val JSON_EXT = ".json"
|
||||
|
||||
/** zero-padded to 10 digits so lex order == chronological order through year 2286. */
|
||||
fun tsPad(ts: Long): String = ts.toString().padStart(TS_PAD_WIDTH, '0')
|
||||
|
||||
/** 16-hex of a Long, zero-padded. */
|
||||
fun hashHex(h: Long): String =
|
||||
java.lang.Long
|
||||
.toHexString(h)
|
||||
.padStart(16, '0')
|
||||
|
||||
fun entryName(
|
||||
ts: Long,
|
||||
id: HexKey,
|
||||
): String = "${tsPad(ts)}-$id"
|
||||
|
||||
/**
|
||||
* Recover (createdAt, id) from an index filename. Returns null if the
|
||||
* filename does not match — lets scrubs tolerate garbage.
|
||||
*/
|
||||
fun parseEntry(name: String): Pair<Long, HexKey>? {
|
||||
if (name.length < TS_PAD_WIDTH + 1 + 1) return null
|
||||
if (name[TS_PAD_WIDTH] != '-') return null
|
||||
val ts = name.substring(0, TS_PAD_WIDTH).toLongOrNull() ?: return null
|
||||
val id = name.substring(TS_PAD_WIDTH + 1)
|
||||
return ts to id
|
||||
}
|
||||
|
||||
private const val TS_PAD_WIDTH = 10
|
||||
}
|
||||
}
|
||||
+151
@@ -0,0 +1,151 @@
|
||||
/*
|
||||
* 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.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.store.sqlite.TagNameValueHasher
|
||||
import java.nio.file.Files
|
||||
import java.nio.file.Path
|
||||
import kotlin.io.path.exists
|
||||
|
||||
/**
|
||||
* Picks a driver index for a filter and streams candidate ids in
|
||||
* `createdAt` DESC order. The orchestrator re-reads the full JSON and
|
||||
* passes it through [Filter.match] for the final predicate check.
|
||||
*
|
||||
* Step-2 coverage:
|
||||
* - `ids` → direct canonical opens
|
||||
* - `tagsAll`/`tags` → tag index union (first key)
|
||||
* - `kinds` → kind index union
|
||||
* - `authors` → author index union
|
||||
* - otherwise → full scan via every `idx/kind/<k>/` subtree
|
||||
*
|
||||
* The planner is intentionally dumb about selectivity — "first available
|
||||
* driver wins". A cost-based picker (smallest listing) can slot in
|
||||
* later without changing callers. All FilterMatcher semantics (tag
|
||||
* AND/OR, since/until, id, author, kind cross-checks) are enforced in
|
||||
* the orchestrator, so picking a loose driver is correctness-safe.
|
||||
*/
|
||||
internal class FsQueryPlanner(
|
||||
private val layout: FsLayout,
|
||||
private val hasher: TagNameValueHasher,
|
||||
) {
|
||||
data class Candidate(
|
||||
val createdAt: Long,
|
||||
val id: HexKey,
|
||||
)
|
||||
|
||||
fun plan(filter: Filter): Sequence<Candidate> {
|
||||
filter.ids?.let { ids ->
|
||||
return idsDriver(ids)
|
||||
}
|
||||
|
||||
firstTagKey(filter)?.let { (name, values) ->
|
||||
return mergeDesc(values.map { v -> walkDir(layout.tagValueDir(name, hasher.hash(name, v))) })
|
||||
}
|
||||
|
||||
filter.kinds?.let { kinds ->
|
||||
return mergeDesc(kinds.map { walkDir(layout.kindDir(it)) })
|
||||
}
|
||||
|
||||
filter.authors?.let { authors ->
|
||||
return mergeDesc(authors.map { walkDir(layout.authorDir(it)) })
|
||||
}
|
||||
|
||||
return allKindsDriver()
|
||||
}
|
||||
|
||||
// ---- drivers ------------------------------------------------------
|
||||
|
||||
/**
|
||||
* For pure id lookups we don't have a timestamp in the filename — read
|
||||
* `mtime`, which [FsEventStore] sets to `event.createdAt` on write.
|
||||
* Cheap (one `stat` per id) and avoids opening/parsing the JSON just
|
||||
* to order the stream.
|
||||
*/
|
||||
private fun idsDriver(ids: List<HexKey>): Sequence<Candidate> =
|
||||
sequence {
|
||||
val out = ArrayList<Candidate>(ids.size)
|
||||
for (id in ids) {
|
||||
val p = layout.canonical(id)
|
||||
if (!p.exists()) continue
|
||||
val ts =
|
||||
try {
|
||||
Files.getLastModifiedTime(p).to(java.util.concurrent.TimeUnit.SECONDS)
|
||||
} catch (_: java.nio.file.NoSuchFileException) {
|
||||
continue
|
||||
}
|
||||
out.add(Candidate(ts, id))
|
||||
}
|
||||
out.sortByDescending { it.createdAt }
|
||||
yieldAll(out)
|
||||
}
|
||||
|
||||
private fun allKindsDriver(): Sequence<Candidate> =
|
||||
sequence {
|
||||
if (!Files.isDirectory(layout.idxKind)) return@sequence
|
||||
val subs = Files.list(layout.idxKind).use { it.toList() }
|
||||
yieldAll(mergeDesc(subs.map { walkDir(it) }))
|
||||
}
|
||||
|
||||
// ---- index directory walker --------------------------------------
|
||||
|
||||
private fun walkDir(dir: Path): Sequence<Candidate> =
|
||||
sequence {
|
||||
if (!Files.isDirectory(dir)) return@sequence
|
||||
val entries = Files.list(dir).use { it.toList() }
|
||||
entries
|
||||
.asSequence()
|
||||
.mapNotNull { FsLayout.parseEntry(it.fileName.toString()) }
|
||||
.map { Candidate(it.first, it.second) }
|
||||
.sortedByDescending { it.createdAt }
|
||||
.forEach { yield(it) }
|
||||
}
|
||||
|
||||
// ---- k-way-ish merge (materialised; step-8 can turn into a heap) -
|
||||
|
||||
private fun mergeDesc(streams: List<Sequence<Candidate>>): Sequence<Candidate> =
|
||||
sequence {
|
||||
if (streams.isEmpty()) return@sequence
|
||||
if (streams.size == 1) {
|
||||
val seen = HashSet<HexKey>()
|
||||
streams[0].forEach { if (seen.add(it.id)) yield(it) }
|
||||
return@sequence
|
||||
}
|
||||
val merged = ArrayList<Candidate>()
|
||||
streams.forEach { s -> s.forEach { merged.add(it) } }
|
||||
merged.sortByDescending { it.createdAt }
|
||||
val seen = HashSet<HexKey>()
|
||||
merged.forEach { if (seen.add(it.id)) yield(it) }
|
||||
}
|
||||
|
||||
// ---- helpers ------------------------------------------------------
|
||||
|
||||
/** First tag filter with at least one value, preferring `tagsAll`. */
|
||||
private fun firstTagKey(filter: Filter): Pair<String, List<String>>? {
|
||||
filter.tagsAll?.firstNonEmpty()?.let { return it }
|
||||
filter.tags?.firstNonEmpty()?.let { return it }
|
||||
return null
|
||||
}
|
||||
|
||||
private fun Map<String, List<String>>.firstNonEmpty(): Pair<String, List<String>>? = entries.firstOrNull { it.value.isNotEmpty() }?.let { it.key to it.value }
|
||||
}
|
||||
@@ -0,0 +1,317 @@
|
||||
/*
|
||||
* 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.core.Event
|
||||
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.assertTrue
|
||||
|
||||
class FsQueryTest {
|
||||
private val signerA = NostrSignerSync()
|
||||
private val signerB = NostrSignerSync()
|
||||
private lateinit var root: Path
|
||||
private lateinit var store: FsEventStore
|
||||
|
||||
@BeforeTest
|
||||
fun setup() {
|
||||
Secp256k1Instance
|
||||
root = Files.createTempDirectory("fs-query-")
|
||||
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 signA(
|
||||
text: String,
|
||||
ts: Long,
|
||||
) = signerA.sign<TextNoteEvent>(TextNoteEvent.build(text, createdAt = ts))
|
||||
|
||||
private fun signB(
|
||||
text: String,
|
||||
ts: Long,
|
||||
) = signerB.sign<TextNoteEvent>(TextNoteEvent.build(text, createdAt = ts))
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Sort + limit
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `results ordered by created_at DESC`() {
|
||||
val a = signA("a", 1)
|
||||
val b = signA("b", 3)
|
||||
val c = signA("c", 2)
|
||||
listOf(a, b, c).forEach(store::insert)
|
||||
|
||||
val got = store.query<TextNoteEvent>(Filter(authors = listOf(signerA.pubKey)))
|
||||
assertEquals(listOf(b.id, c.id, a.id), got.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `limit caps the result count`() {
|
||||
repeat(5) { i -> store.insert(signA("n$i", i.toLong() + 1)) }
|
||||
val got = store.query<TextNoteEvent>(Filter(authors = listOf(signerA.pubKey), limit = 2))
|
||||
assertEquals(2, got.size)
|
||||
// Highest timestamps come first.
|
||||
assertEquals("n4", got[0].content)
|
||||
assertEquals("n3", got[1].content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `limit of zero returns empty`() {
|
||||
store.insert(signA("x", 1))
|
||||
assertEquals(emptyList(), store.query<TextNoteEvent>(Filter(authors = listOf(signerA.pubKey), limit = 0)))
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Author / kind drivers
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `author filter isolates one user`() {
|
||||
val a = signA("from-a", 1)
|
||||
val b = signB("from-b", 2)
|
||||
store.insert(a)
|
||||
store.insert(b)
|
||||
|
||||
val got = store.query<TextNoteEvent>(Filter(authors = listOf(signerA.pubKey)))
|
||||
assertEquals(listOf(a.id), got.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `author filter with multiple authors unions them`() {
|
||||
val a = signA("a", 1)
|
||||
val b = signB("b", 2)
|
||||
store.insert(a)
|
||||
store.insert(b)
|
||||
|
||||
val got = store.query<TextNoteEvent>(Filter(authors = listOf(signerA.pubKey, signerB.pubKey)))
|
||||
assertEquals(setOf(a.id, b.id), got.map { it.id }.toSet())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `kind filter returns only the requested kinds`() {
|
||||
// Build two events of different kinds.
|
||||
val note = signA("note", 1)
|
||||
val ephemeralKinds = signerA.sign<Event>(createdAt = 2, kind = 30023, tags = arrayOf(arrayOf("d", "slug")), content = "article")
|
||||
store.insert(note)
|
||||
store.insert(ephemeralKinds)
|
||||
|
||||
val onlyNotes = store.query<Event>(Filter(kinds = listOf(1)))
|
||||
assertEquals(listOf(note.id), onlyNotes.map { it.id })
|
||||
|
||||
val onlyArticles = store.query<Event>(Filter(kinds = listOf(30023)))
|
||||
assertEquals(listOf(ephemeralKinds.id), onlyArticles.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `kind + author intersect via post-filter`() {
|
||||
val a = signA("a", 1)
|
||||
val b = signB("b", 2)
|
||||
store.insert(a)
|
||||
store.insert(b)
|
||||
|
||||
val got = store.query<TextNoteEvent>(Filter(kinds = listOf(1), authors = listOf(signerA.pubKey)))
|
||||
assertEquals(listOf(a.id), got.map { it.id })
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Tag driver
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `tag filter matches single-letter tags`() {
|
||||
val tagged =
|
||||
signerA.sign<Event>(
|
||||
createdAt = 10,
|
||||
kind = 1,
|
||||
tags = arrayOf(arrayOf("t", "nostr"), arrayOf("t", "bitcoin")),
|
||||
content = "tagged",
|
||||
)
|
||||
val untagged = signA("plain", 5)
|
||||
store.insert(tagged)
|
||||
store.insert(untagged)
|
||||
|
||||
val got = store.query<Event>(Filter(tags = mapOf("t" to listOf("nostr"))))
|
||||
assertEquals(listOf(tagged.id), got.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `tag OR within key returns union`() {
|
||||
val t1 = signerA.sign<Event>(createdAt = 1, kind = 1, tags = arrayOf(arrayOf("t", "nostr")), content = "n")
|
||||
val t2 = signerA.sign<Event>(createdAt = 2, kind = 1, tags = arrayOf(arrayOf("t", "bitcoin")), content = "b")
|
||||
val t3 = signerA.sign<Event>(createdAt = 3, kind = 1, tags = arrayOf(arrayOf("t", "other")), content = "o")
|
||||
listOf(t1, t2, t3).forEach(store::insert)
|
||||
|
||||
val got = store.query<Event>(Filter(tags = mapOf("t" to listOf("nostr", "bitcoin"))))
|
||||
assertEquals(setOf(t1.id, t2.id), got.map { it.id }.toSet())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `tagsAll across keys requires all matches`() {
|
||||
val both =
|
||||
signerA.sign<Event>(
|
||||
createdAt = 1,
|
||||
kind = 1,
|
||||
tags = arrayOf(arrayOf("t", "nostr"), arrayOf("e", "a".repeat(64))),
|
||||
content = "both",
|
||||
)
|
||||
val onlyT = signerA.sign<Event>(createdAt = 2, kind = 1, tags = arrayOf(arrayOf("t", "nostr")), content = "t-only")
|
||||
val onlyE = signerA.sign<Event>(createdAt = 3, kind = 1, tags = arrayOf(arrayOf("e", "a".repeat(64))), content = "e-only")
|
||||
listOf(both, onlyT, onlyE).forEach(store::insert)
|
||||
|
||||
val got =
|
||||
store.query<Event>(
|
||||
Filter(
|
||||
tagsAll = mapOf("t" to listOf("nostr"), "e" to listOf("a".repeat(64))),
|
||||
),
|
||||
)
|
||||
assertEquals(listOf(both.id), got.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `non-single-letter tags are not reverse-indexed`() {
|
||||
// SQLite parity: DefaultIndexingStrategy only indexes single-letter
|
||||
// tag names, so a tag-driven query for `mytag = foo` finds no
|
||||
// candidates. The event is still persisted and can be fetched via
|
||||
// id / author / kind — just not via a reverse tag lookup.
|
||||
val e =
|
||||
signerA.sign<Event>(
|
||||
createdAt = 1,
|
||||
kind = 1,
|
||||
tags = arrayOf(arrayOf("mytag", "foo")),
|
||||
content = "x",
|
||||
)
|
||||
store.insert(e)
|
||||
|
||||
assertEquals(emptyList(), store.query<Event>(Filter(tags = mapOf("mytag" to listOf("foo")))).map { it.id })
|
||||
assertEquals(listOf(e.id), store.query<Event>(Filter(authors = listOf(signerA.pubKey))).map { it.id })
|
||||
assertEquals(listOf(e.id), store.query<Event>(Filter(ids = listOf(e.id))).map { it.id })
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// since / until
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `since and until window filter`() {
|
||||
val e1 = signA("t1", 100)
|
||||
val e2 = signA("t2", 200)
|
||||
val e3 = signA("t3", 300)
|
||||
listOf(e1, e2, e3).forEach(store::insert)
|
||||
|
||||
val got = store.query<TextNoteEvent>(Filter(since = 150, until = 250))
|
||||
assertEquals(listOf(e2.id), got.map { it.id })
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// count
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `count matches query size`() {
|
||||
repeat(4) { i -> store.insert(signA("n$i", i.toLong() + 1)) }
|
||||
val filter = Filter(authors = listOf(signerA.pubKey))
|
||||
assertEquals(store.query<TextNoteEvent>(filter).size, store.count(filter))
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Index hardlink maintenance
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `insert creates hardlinks in every expected index dir`() {
|
||||
val tagged =
|
||||
signerA.sign<Event>(
|
||||
createdAt = 42,
|
||||
kind = 1,
|
||||
tags = arrayOf(arrayOf("t", "nostr")),
|
||||
content = "x",
|
||||
)
|
||||
store.insert(tagged)
|
||||
|
||||
val kindDir = root.resolve("idx/kind/1")
|
||||
val authorDir = root.resolve("idx/author/${signerA.pubKey}")
|
||||
assertTrue(kindDir.exists() && kindDir.listDirectoryEntries().size == 1, "kind index missing")
|
||||
assertTrue(authorDir.exists() && authorDir.listDirectoryEntries().size == 1, "author index missing")
|
||||
val tagNameDir = root.resolve("idx/tag/t")
|
||||
assertTrue(tagNameDir.exists(), "tag 't' dir missing")
|
||||
val tagValueDirs = tagNameDir.listDirectoryEntries()
|
||||
assertEquals(1, tagValueDirs.size, "exactly one tag-value subdir expected")
|
||||
assertEquals(1, tagValueDirs[0].listDirectoryEntries().size, "tag-value dir should contain one entry")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `delete removes hardlinks so directories become empty`() {
|
||||
val e = signerA.sign<Event>(createdAt = 1, kind = 1, tags = arrayOf(arrayOf("t", "nostr")), content = "x")
|
||||
store.insert(e)
|
||||
store.delete(e.id)
|
||||
|
||||
val kindDir = root.resolve("idx/kind/1")
|
||||
val authorDir = root.resolve("idx/author/${signerA.pubKey}")
|
||||
val tagValueDirs =
|
||||
root
|
||||
.resolve("idx/tag/t")
|
||||
.takeIf { it.exists() }
|
||||
?.listDirectoryEntries()
|
||||
.orEmpty()
|
||||
|
||||
// Directories may remain as empty husks — what matters is the entries are gone.
|
||||
if (kindDir.exists()) assertEquals(0, kindDir.listDirectoryEntries().size, "kind entry leaked")
|
||||
if (authorDir.exists()) assertEquals(0, authorDir.listDirectoryEntries().size, "author entry leaked")
|
||||
tagValueDirs.forEach { assertEquals(0, it.listDirectoryEntries().size, "tag entry leaked") }
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Seed persistence across reopen
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `reopening the store preserves queryability`() {
|
||||
val tagged = signerA.sign<Event>(createdAt = 1, kind = 1, tags = arrayOf(arrayOf("t", "nostr")), content = "x")
|
||||
store.insert(tagged)
|
||||
store.close()
|
||||
|
||||
val reopened = FsEventStore(root)
|
||||
try {
|
||||
val got = reopened.query<Event>(Filter(tags = mapOf("t" to listOf("nostr"))))
|
||||
assertEquals(listOf(tagged.id), got.map { it.id })
|
||||
} finally {
|
||||
reopened.close()
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user