amethyst/quartz/RELAY.md

# Building a Nostr Relay with Quartz

Quartz provides a transport-agnostic relay engine. You provide a `send` callback per connection, and it gives you a `RelaySession` that accepts raw JSON strings. Plug it into Ktor or any WebSocket transport.

There are two engines on the same `RelaySession` core: `NostrServer` for storage-backed relays (an `IEventStore` with a live tail after EOSE) and `EventSourceServer` for non-storage relays (search, redirector, computed data — see [Non-Storage Relays](#non-storage-relays-search-redirector-computed)).

Both `NostrServer` and `EventStore` implement `AutoCloseable`.

## Quick Start

### 1. Add Dependencies

```kotlin
dependencies {
    implementation("io.ktor:ktor-server-core:3.1.1")
    implementation("io.ktor:ktor-server-netty:3.1.1")
    implementation("io.ktor:ktor-server-websockets:3.1.1")
    implementation(project(":quartz"))
}
```

### 2. Create the Relay Server

```kotlin
fun main() {
    val store = EventStore(dbName = "relay-events.db")
    val server = NostrServer(store)

    embeddedServer(Netty, port = 7777) {
        install(WebSockets)

        routing {
            webSocket("/") {
                server.serve(
                    send = { json -> launch { send(Frame.Text(json)) } },
                ) { session ->
                    for (frame in incoming) {
                        if (frame is Frame.Text) {
                            session.receive(frame.readText())
                        }
                    }
                }
            }
        }
    }.start(wait = true)
}
```

That's it. You now have a NIP-01 compliant relay running on `ws://localhost:7777`.

## Event Store

### SQLite (Persistent)

```kotlin
val store = EventStore(dbName = "relay-events.db")
```

Uses `androidx.sqlite` with WAL journal mode and a 32 MB memory cache.

### In-Memory (Testing)

```kotlin
val store = EventStore(null)
```

### Indexing Strategy

Control which indexes are created:

```kotlin
val store = EventStore(
    dbName = "relay-events.db",
    indexStrategy = DefaultIndexingStrategy(
        indexEventsByCreatedAtAlone = false,
        indexTagsByCreatedAtAlone = false,
        indexTagsWithKindAndPubkey = false,
        useAndIndexIdOnOrderBy = false,
        indexFullTextSearch = true,
    ),
)
```

By default, all single-letter tags with values are indexed. Override `shouldIndex(kind, tag)` for custom behavior. More indexes = faster queries but larger database.

Flag flips are safe on existing databases: any flag-gated index the strategy wants but the on-disk schema lacks is created on the next open (idempotent `CREATE INDEX IF NOT EXISTS`, one-time build cost) — no schema version bump involved. Disabling a flag never drops an existing index.

`indexFullTextSearch` defaults to `true` and controls the NIP-50 full-text index (`event_fts`). Set it to `false` when search is served elsewhere (e.g. a Vespa backend, or a `SearchEventSource` as shown below): inserts skip the FTS tokenization cost, no `event_fts` table/trigger is created, and any filter carrying a non-empty `search` term returns no matches.

## Non-Storage Relays (search, redirector, computed)

A relay's job is to *answer REQs*. When the answer doesn't come from a stored
set — a NIP-50 search that forwards to an HTTP backend, a relay that emits
computed/projected data — implement `EventSource` and serve it with
`EventSourceServer`. You supply a `Flow<Event>`; the engine owns the wire
protocol (challenge/auth, command parsing, policy, `EVENT`/`EOSE`/`CLOSED`
framing, subscription lifecycle), so there's no hand-written read loop.

```kotlin
class SearchEventSource(private val backend: SearchApi) : EventSource {
    override fun events(ctx: RequestContext, filters: List<Filter>): Flow<Event> = flow {
        // ctx says who is asking — score/restrict results from their perspective.
        val viewer = ctx.authenticatedUsers.firstOrNull()
        filters.forEach { f ->
            f.search?.let { raw ->
                val q = SearchQuery.parse(raw)
                backend.search(q.terms, domain = q.domain, language = q.language, viewer = viewer)
                    .forEach { emit(it) }
            }
        }
    }
    // COUNT defaults to counting events(); override for a cheaper backend count.
}

fun main() {
    val server = EventSourceServer(
        source = SearchEventSource(searchApi),
        policyBuilder = { FullAuthPolicy(relay) }, // optional NIP-42 gating
    )

    embeddedServer(Netty, port = 7777) {
        install(WebSockets)
        routing {
            webSocket("/") {
                server.serve(send = { json -> launch { send(Frame.Text(json)) } }) { session ->
                    for (frame in incoming) {
                        if (frame is Frame.Text) session.receive(frame.readText())
                    }
                }
            }
        }
    }.start(wait = true)
}
```

**EOSE = flow completion.** The engine sends `EOSE` when `events(...)`
completes, which is the natural shape for finite queries. Relays that need an
open-ended live tail after EOSE should use the storage path (`NostrServer` +
`IEventStore`) instead. EVENT publishes are rejected (`OK false` —
`blocked: this relay does not accept events`) and negentropy is disabled, since
there's no stored set. A failure thrown from the flow ends the subscription
with `CLOSED` `error: <message>`.

`EventSourceServer` and `NostrServer` are the two concrete dispatch engines;
both build on `RelaySession` and the shared `SessionBackend` seam (`LiveEventStore`
is the storage-backed `SessionBackend`; `EventSourceBackend` adapts a
`EventSource`). Implement `SessionBackend` directly only if you need custom
control over the EVENT/negentropy paths as well as REQ/COUNT.

### Caller-aware sources (who is asking)

Every `events`/`count`/`countResult` call receives a `RequestContext` carrying
the connection's `authenticatedUsers` (the pubkeys that completed NIP-42 on this
socket) and a stable `connectionId`. That is what makes NIP-42 useful on a
non-storage relay: a single shared `EventSource` instance can serve
caller-relative results (trust/relevance scored from the viewer's perspective,
"for-you" feeds), restricted content (a pubkey's DMs returned only to that
pubkey), or per-connection tenancy — without smuggling auth state through a
side channel. `ctx.authenticatedUsers` is a live view of the engine-owned
connection scope, so a REQ that arrives after the AUTH sees the freshly
recorded pubkey(s). (The engine records them on a successful NIP-42 AUTH; the
policy reads the same scope to gate.)

For per-connection state richer than the pubkey (e.g. a backend session token
minted in `FullAuthPolicy.authorize`), downcast `ctx.policy` to your own policy
subclass and read a typed field — the policy instance is itself per-connection.

## Policies

Policies control what clients can do. They validate commands and can rewrite filters.

### Built-in Policies

**`VerifyPolicy`** (default) — Verifies event signatures and IDs. Rejects malformed events.

```kotlin
val server = NostrServer(store) // Uses VerifyPolicy by default
```

**`EmptyPolicy`** — Accepts everything. Useful for testing.

```kotlin
val server = NostrServer(store, policyBuilder = { EmptyPolicy })
```

**`FullAuthPolicy`** — Requires NIP-42 authentication before accepting any command.

```kotlin
val server = NostrServer(
    store = store,
    policyBuilder = {
        FullAuthPolicy(relay = "wss://myrelay.example.com/".normalizeRelayUrl()!!)
    },
)
```

`FullAuthPolicy` already implements the full NIP-42 challenge/verify handshake — you should not re-implement it. To bridge auth to an external system (e.g. exchange the verified event for a backend JWT), override the `suspend` `authorize` hook. It runs after the NIP-42 checks pass (and after the rest of the policy chain approves), and the pubkey is recorded only once it returns — so it can do network/disk I/O, and throwing from it rejects the login (`OK false`) with the connection left unauthenticated (nothing was committed).

### Composing Policies

Chain policies with `+` or `PolicyStack`. All must approve; first rejection wins.

```kotlin
val server = NostrServer(
    store = store,
    policyBuilder = {
        VerifyPolicy + FullAuthPolicy(relay = "wss://myrelay.example.com/".normalizeRelayUrl()!!)
    },
)
```

### Writing a Custom Policy

Implement `IRelayPolicy`:

```kotlin
class KindWhitelistPolicy(
    private val allowedKinds: Set<Int>,
) : IRelayPolicy {

    override fun onConnect(send: (Message) -> Unit) { }

    override fun accept(cmd: EventCmd): PolicyResult<EventCmd> =
        if (cmd.event.kind in allowedKinds) {
            PolicyResult.Accepted(cmd)
        } else {
            PolicyResult.Rejected("blocked: kind ${cmd.event.kind} not allowed")
        }

    override fun accept(cmd: ReqCmd) = PolicyResult.Accepted(cmd)
    override fun accept(cmd: CountCmd) = PolicyResult.Accepted(cmd)
    override fun accept(cmd: AuthCmd) = PolicyResult.Accepted(cmd)
}
```

Use it:

```kotlin
val server = NostrServer(
    store = store,
    policyBuilder = {
        VerifyPolicy + KindWhitelistPolicy(allowedKinds = setOf(0, 1, 3, 7, 30023))
    },
)
```

## NIP-50 Search Queries

`Filter.search` is the raw NIP-50 string. `SearchQuery` parses it into the
free-text terms plus the typed `key:value` extensions (`domain:`, `language:`,
`sentiment:`, `nsfw:`, `include:spam`), so a search relay or redirector doesn't
have to re-parse the string. Unknown extensions are preserved and readable via
`extension(key)`; `toSearchString()` re-assembles a canonical query.

```kotlin
val q = SearchQuery.parse(filter.search) // "best apps domain:example.com nsfw:false"
q.terms        // "best apps"
q.domain       // "example.com"
q.nsfwIncluded // false  (NIP-50 default is true when the token is absent)
```

`Filter.search` reaches every backend **verbatim**, extension tokens included
— the relay layer never rewrites it. Which extensions are directives and which
are noise is a property of the store, so the `IEventStore` contract puts the
decision there: a store that implements an extension (rank profiles, trust
floors, observer-relative scoring) parses the raw string with
`SearchQuery.parse`; a store that doesn't must ignore the tokens per NIP-50
(not match them as literal text, not return nothing). The built-in SQLite and
filesystem stores do the latter by stripping at their own boundary with
`filter.strippingSearchExtensions()` — FTS5 treats `:` as column-filter syntax,
so a raw `include:spam` reaching MATCH would raise "no such column: include" —
and an extensions-only query collapses to an empty search, which imposes no
constraint. Relays like geode therefore comply out of the box, and
`EventSource` backends likewise get the raw string.

Observer-relative stores (web-of-trust ranking, "for-you" relevance) read the
caller's NIP-42-authenticated pubkeys from the coroutine context via
`StoreQueryContext` — `LiveEventStore` installs it around every REQ/COUNT store
call for authenticated connections. It is ranking context only: it may reorder
results, never change which events match.

A search/redirector relay is just a custom policy (or, for computed results, a
custom `IEventStore` whose `query` answers the REQ) that reads the parsed query:

```kotlin
override fun accept(cmd: ReqCmd): PolicyResult<ReqCmd> {
    cmd.filters.forEach { f ->
        val q = SearchQuery.parse(f.search)
        if (q.domain != null && q.domain !in allowedDomains) {
            return PolicyResult.Rejected(
                MachineReadablePrefix.RESTRICTED.format("domain not searchable"),
            )
        }
    }
    return PolicyResult.Accepted(cmd)
}
```

## Wire Helpers

`Command` and `Message` carry symmetric JSON helpers so you don't have to reach
for the mapper directly:

```kotlin
val cmd = Command.fromJson(text)   // ["REQ", "sub", {...}] -> ReqCmd
val json = EoseMessage("sub").toJson()
val msg = Message.fromJson(json)
```

Build standardized OK/CLOSED reasons with `MachineReadablePrefix` instead of
hand-writing the NIP-01 prefixes (`auth-required:`, `restricted:`, `error:`, …):

```kotlin
session.send(OkMessage.rejected(event.id, MachineReadablePrefix.AUTH_REQUIRED, "log in first"))
session.send(ClosedMessage.of(subId, MachineReadablePrefix.RESTRICTED, "not allowed yet"))
MachineReadablePrefix.parse("rate-limited: slow down") // -> RATE_LIMITED
```

## Testing

```kotlin
@OptIn(ExperimentalCoroutinesApi::class)
class MyRelayTest {

    @Test
    fun clientCanPublishAndSubscribe() = runTest {
        val dispatcher = UnconfinedTestDispatcher(testScheduler)

        NostrServer(
            store = EventStore(null),
            policyBuilder = { EmptyPolicy },
            parentContext = dispatcher,
        ).use { server ->
            val messages = mutableListOf<String>()
            val session = server.connect { messages.add(it) }

            session.receive("""["EVENT",{"id":"${"0".repeat(64)}","pubkey":"${"a".repeat(64)}","created_at":1000,"kind":1,"tags":[],"content":"hello","sig":"${"b".repeat(128)}"}]""")
            assertTrue(messages.any { it.contains("OK") })

            session.receive("""["REQ","sub1",{"kinds":[1]}]""")
            assertTrue(messages.any { it.contains("EVENT") })
            assertTrue(messages.any { it.contains("EOSE") })
        }
    }
}
```

## Limits (NIP-11) — enforce and advertise from one source

`RelayLimits` is the single source of truth for the relay's operational limits.
Pass it to the server and every limit is enforced; call `toNip11Limitation()`
and the *same* numbers are advertised in your NIP-11 document — they can't drift.

```kotlin
val limits = RelayLimits(
    maxMessageLength = 65_536,
    maxSubscriptions = 20,
    maxFilters = 10,
    maxLimit = 500,
    maxContentLength = 8_196,
    maxEventTags = 2_000,
    createdAtUpperLimit = TimeUtils.now() + 900,
    authRequired = true,
)

val server = NostrServer(store, policyBuilder = { FullAuthPolicy(relay) }, limits = limits)
```

All of it is enforced by a single `LimitsPolicy` (which the server prepends to
your policy when you pass `limits`), so limits compose through a `PolicyStack`
like any other policy — you can also split them across several policies:

- **Per-command** (`accept(...)`): rejects EVENTs over `maxContentLength` /
  `maxEventTags` / outside the `createdAt` bounds; rejects REQ/COUNT with too
  many filters or an over-long sub id; and **clamps** each filter's `limit` to
  `maxLimit` (substituting `defaultLimit` when none is given). Rejections use
  the `invalid:` machine-readable prefix.
- **Per-connection** (the `acceptMessage` / `acceptSubscription` policy hooks):
  oversized frames get a `NOTICE` (`maxMessageLength`); a new subscription past
  `maxSubscriptions` is `CLOSED` with `rate-limited:`. These hooks exist because
  a policy otherwise can't see the raw frame or the live subscription count.
- **Advertised only**: `minPowDifficulty` (enforce with a PoW policy),
  `authRequired` (use `FullAuthPolicy`), `paymentRequired`, `restrictedWrites`.

## Serving NIP-11

Build a `Nip11RelayInformation`, fold in the same `limits`, and serve it at the
relay root with the NIP-11 media type:

```kotlin
val info = Nip11RelayInformation(
    name = "My Relay",
    supported_nips = listOf("1", "11", "42", "45"),
    limitation = server.limits?.toNip11Limitation(),
)

// Ktor: answer GET / when the client asks for application/nostr+json
get("/") {
    call.respondText(info.toJson(), ContentType.parse(Nip11RelayInformation.CONTENT_TYPE))
}
```

`Nip11RelayInformation.fromJson` / `toJson` round-trip the document (null fields
are omitted), and `CONTENT_TYPE` is `application/nostr+json`.

## Approximate COUNT (NIP-45 HyperLogLog)

`COUNT` is answered by `SessionBackend.countResult(ctx, filters)` (and
`EventSource.countResult`), which defaults to an exact count. To return a
mergeable HyperLogLog estimate instead — for the six canonical NIP-45 queries
(reaction/repost/quote/reply/comment/follower counts) — fold matching pubkeys
into an `HllBuilder` and return its `CountResult`:

```kotlin
override suspend fun countResult(ctx: RequestContext, filters: List<Filter>): CountResult {
    val filter = filters.first()
    val hll = HyperLogLog.builderFor(filter) ?: return CountResult(count(ctx, filters))
    store.query(filter) { event -> hll.add(event.pubKey) }
    return hll.toCountResult() // count = estimate, approximate = true, hll = registers
}
```

The engine frames `count`/`approximate`/`hll` onto the wire
(`["COUNT", id, {"count":N,"hll":"<512-hex>"}]`). Register arrays built by two
relays over the same corpus merge with `HyperLogLog.merge(...)` into a
deduplicated cross-relay estimate.

## Observability

Both servers take an optional `RelayServerListener` and expose a live
`activeConnections` gauge. Connections carry a stable, process-unique
`RelaySession.id` (used to key the server's registry and the listener
callbacks), so you can correlate the open/close of the same connection in logs
and metrics:

```kotlin
val server = NostrServer(
    store = store,
    listener = object : RelayServerListener {
        override fun onConnect(connectionId: Long) = metrics.connections.inc()
        override fun onDisconnect(connectionId: Long) = metrics.connections.dec()
    },
)
server.activeConnections // Long, current count
```

`onConnect`/`onDisconnect` fire at most once per connection (a double `close()`
is accounted once) and `onDisconnect` is also fired for any connections still
open when the server itself is closed. Callbacks can run on any transport
coroutine — keep them cheap and non-blocking.

## Key Source Files

```
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/
├── relay/server/                  # engine + entry points
│   ├── RelayServerBase.kt      # Shared engine (connect/serve/policy/registry)
│   ├── NostrServer.kt          # Storage-backed engine (IEventStore)
│   ├── EventSourceServer.kt    # Non-storage engine (search/redirector/computed)
│   ├── RelaySession.kt         # Per-connection handler (stable .id)
│   ├── ConnectionRegistry.kt   # Live-connection bookkeeping + gauge
│   ├── RelayServerListener.kt  # Connection open/close observability hook
│   ├── NegSessionRegistry.kt   # NIP-77 negentropy per-connection state
│   ├── backend/                # data plane (how REQ/COUNT/EVENT are answered)
│   │   ├── SessionBackend.kt   # Data-plane seam (LiveEventStore / EventSourceBackend)
│   │   ├── EventSource.kt      # Flow<Event> SPI for non-storage relays
│   │   ├── EventSourceBackend.kt # Adapts an EventSource to SessionBackend
│   │   ├── LiveEventStore.kt   # Reactive event streaming (storage SessionBackend)
│   │   └── IngestQueue.kt      # Group-commit EVENT writer
│   ├── policies/               # policy model + implementations
│   │   ├── IRelayPolicy.kt     # Policy interface + PolicyResult + onAuthenticated
│   │   ├── RelayLimits.kt      # Limits: enforced + NIP-11 limitation source of truth
│   │   ├── EmptyPolicy.kt      # Accept everything
│   │   ├── VerifyPolicy.kt     # Signature verification (default)
│   │   ├── FullAuthPolicy.kt   # NIP-42 auth required (override authorize to bridge)
│   │   ├── LimitsPolicy.kt     # Enforcement of RelayLimits (per-command + hooks)
│   │   └── PolicyStack.kt      # Chain multiple policies
│   └── inprocess/              # in-memory transport for tests
├── relay/commands/
│   ├── toRelay/Command.kt      # Command.fromJson / toJson
│   └── toClient/
│       ├── Message.kt          # Message.fromJson / toJson
│       └── MachineReadablePrefix.kt # Typed OK/CLOSED reason prefixes
├── store/
│   ├── IEventStore.kt          # Storage interface
│   └── sqlite/
│       ├── EventStore.kt       # Public SQLite store wrapper
│       ├── SQLiteEventStore.kt # Full implementation
│       └── IndexingStrategy.kt # Index configuration
├── relay/filters/
│   ├── Filter.kt               # NIP-01 subscription filters
│   └── FilterMatcher.kt        # Event-to-filter matching
└── ../nip50Search/
    └── SearchQuery.kt          # NIP-50 search-string parser
```