diff --git a/cli/README.md b/cli/README.md index 3070ff0c95..47cacd7383 100644 --- a/cli/README.md +++ b/cli/README.md @@ -234,6 +234,7 @@ Filter flags are shared by `fetch` and `subscribe`: `--kind K[,K]`, `--author U[ | `amy subscribe [filter flags] [--timeout SECS]` | Live stream — print each matching event as it arrives (NDJSON under `--json`). Runs until `--timeout` SECS or until interrupted. | | `amy count [filter flags] [--timeout SECS]` | NIP-45 COUNT — per-relay match counts, no event download. | | `amy outbox USER [--refresh] [--timeout SECS]` | Show USER's NIP-65 read/write relays (outbox model). Cache-first; `--refresh` forces a relay drain. | +| `amy sync --relay URL [filter flags] [--down] [--up]` | NIP-77 Negentropy reconcile between the local store and a relay. `--down` (default) pulls events we lack; `--up` pushes events the relay lacks; both for bidirectional. | ### Encryption diff --git a/cli/ROADMAP.md b/cli/ROADMAP.md index a90ce9d1f1..aafe3e17a3 100644 --- a/cli/ROADMAP.md +++ b/cli/ROADMAP.md @@ -94,8 +94,10 @@ vs streaming `subscribe`). Stateless verbs run with no account or network. | `filter` | `amy filter` | ✅ | stateless — assemble + print a filter JSON. | | `blossom` | `amy blossom` | ✅ | upload/download/list/delete (reuses commons `BlossomClient`). | | `kind` / `nip` | `amy kind` / `amy nip` | 🆕 | reference lookups (needs a kind registry — only remaining Tier-1 gap). | -| `sync` | `amy relay sync` | 🆕 (tier 2) | NIP-77 Negentropy. | -| `bunker` / `serve` / `admin` / `wallet` / `git` / `podcast` / `mcp` / `fs` / `spell` | — | 🆕 (tier 2/3) | larger/niche; some pull new deps. | +| `sync` | `amy sync` | ✅ | NIP-77 Negentropy reconcile with the local store (down/up/both). | +| `git` | `amy git` | ✅ in part | NIP-34 repo announce/list/show/issue. clone/push (packfile transport) out of scope. | +| `podcast` | `amy podcast` | ✅ | NIP-F4 show metadata (10154) + episode publish (54) + list. | +| `bunker` / `serve` / `admin` / `wallet` / `mcp` / `fs` / `spell` | — | 🆕 (tier 2/3) | larger/niche; some pull new deps. | **Tier 1 status:** shipped — `decode`, `encode`, `verify`, `key`, `event`, `publish`, `fetch`, `subscribe`, `count`, `encrypt`, `decrypt`, `gift`, diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt index 07ec3a089d..903d023c5b 100644 --- a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Main.kt @@ -53,6 +53,7 @@ import com.vitorpamplona.amethyst.cli.commands.RelayCommands import com.vitorpamplona.amethyst.cli.commands.SearchCommand import com.vitorpamplona.amethyst.cli.commands.StoreCommands import com.vitorpamplona.amethyst.cli.commands.SubscribeCommand +import com.vitorpamplona.amethyst.cli.commands.SyncCommand import com.vitorpamplona.amethyst.cli.commands.UseCommand import com.vitorpamplona.amethyst.cli.commands.VerifyCommand import com.vitorpamplona.amethyst.cli.commands.ZapCommand @@ -207,6 +208,7 @@ private suspend fun dispatch(argv: Array): Int { "gift" -> GiftCommands.dispatch(dataDir, tail) "outbox" -> OutboxCommand.run(dataDir, tail) "blossom" -> BlossomCommands.dispatch(dataDir, tail) + "sync" -> SyncCommand.run(dataDir, tail) else -> { System.err.println("unknown subcommand: $head") printUsage() @@ -391,6 +393,9 @@ private fun printUsage() { | [--relay URL[,URL…]] [--timeout SECS] (NDJSON). Runs until --timeout or interrupt. | count [] NIP-45 COUNT: per-relay match counts, no | [--relay URL[,URL…]] [--timeout SECS] event download. + | sync --relay URL [] NIP-77 Negentropy reconcile with the local + | [--down] [--up] [--timeout SECS] store (--down default; --up to push ours; + | both for bidirectional). | |Encryption (active account's key): | encrypt --to USER [TEXT] [--nip04] NIP-44 (default) or NIP-04 encrypt. Reads diff --git a/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt new file mode 100644 index 0000000000..51d7d471d8 --- /dev/null +++ b/cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/SyncCommand.kt @@ -0,0 +1,209 @@ +/* + * 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.amethyst.cli.commands + +import com.vitorpamplona.amethyst.cli.Args +import com.vitorpamplona.amethyst.cli.Context +import com.vitorpamplona.amethyst.cli.DataDir +import com.vitorpamplona.amethyst.cli.Output +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.toHttp +import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.Response +import okhttp3.WebSocket +import okhttp3.WebSocketListener + +/** + * `amy sync --relay URL [filter flags] [--down] [--up] [--timeout SECS]` + * + * NIP-77 Negentropy set-reconciliation between the local event store and a + * relay (nak's `sync`, adapted to amy's local-store model). First it + * negotiates the symmetric difference under [filter], then closes the loop: + * + * --down (default) download events the relay has and we lack (REQ by id) + * --up upload events we have and the relay lacks (EVENT) + * + * Pass both for a full bidirectional sync. The filter flags are the same as + * `fetch`/`subscribe`; an empty filter reconciles the whole store. + * + * Thin assembly only: the negentropy protocol lives in quartz + * (`NegentropySession`); this file drives the WebSocket round-trips and + * reuses `Context.drain` / `Context.publish` for the NIP-01 follow-up. + */ +object SyncCommand { + private const val ID_CHUNK = 500 + + suspend fun run( + dataDir: DataDir, + rest: Array, + ): Int { + val args = Args(rest) + val relayUrl = + args.flag("relay") + ?: return Output.error("bad_args", "sync requires --relay URL") + val relay = + RelayUrlNormalizer.normalizeOrNull(relayUrl) + ?: return Output.error("bad_args", "invalid relay url: $relayUrl") + val timeoutMs = (args.flag("timeout")?.toLongOrNull() ?: 30L) * 1000 + // Default direction is download; --up adds upload. + val up = args.bool("up") + val down = args.bool("down") || !up + val filter = RawEventSupport.buildFilter(args) + + Context.open(dataDir).use { ctx -> + ctx.prepare() + val localEvents = ctx.store.query(filter) + val localById = localEvents.associateBy { it.id } + + val diff = + negotiate(relay.toHttp(), filter, localEvents, timeoutMs) + ?: return Output.error("timeout", "no negentropy response from ${relay.url} within ${timeoutMs}ms") + if (diff.error != null) { + return Output.error("sync_error", diff.error) + } + + // needIds = relay has, we lack; haveIds = we have, relay lacks. + var downloaded = 0 + if (down && diff.needIds.isNotEmpty()) { + diff.needIds.chunked(ID_CHUNK).forEach { chunk -> + val got = ctx.drain(mapOf(relay to listOf(Filter(ids = chunk))), timeoutMs) + downloaded += got.size + } + } + + var uploaded = 0 + if (up && diff.haveIds.isNotEmpty()) { + diff.haveIds.forEach { id -> + val ev = localById[id] ?: return@forEach + val ack = ctx.publish(ev, setOf(relay)) + if (ack.values.any { it }) uploaded++ + } + } + + Output.emit( + mapOf( + "relay" to relay.url, + "local_events" to localEvents.size, + "rounds" to diff.rounds, + "need" to diff.needIds.size, + "have" to diff.haveIds.size, + "downloaded" to downloaded, + "uploaded" to uploaded, + ), + ) + return 0 + } + } + + private data class Diff( + val haveIds: List, + val needIds: List, + val rounds: Int, + val error: String?, + ) + + /** + * Drive one NIP-77 reconciliation over a raw WebSocket and return the + * symmetric difference. Mirrors geode's interop sync driver: the protocol + * state machine is [NegentropySession]; this only shuttles frames. + */ + private suspend fun negotiate( + httpUrl: String, + filter: Filter, + localEvents: List, + timeoutMs: Long, + subId: String = "amy-sync", + maxRounds: Int = 64, + ): Diff? { + val incoming = Channel(UNLIMITED) + val client = OkHttpClient.Builder().build() + val ws = + client.newWebSocket( + Request.Builder().url(httpUrl).build(), + object : WebSocketListener() { + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { + incoming.trySend(text) + } + + override fun onClosing( + webSocket: WebSocket, + code: Int, + reason: String, + ) { + incoming.close() + } + + override fun onFailure( + webSocket: WebSocket, + t: Throwable, + response: Response?, + ) { + incoming.close(t) + } + }, + ) + + return try { + withTimeoutOrNull(timeoutMs) { + val session = NegentropySession(subId, filter, localEvents, frameSizeLimit = 0) + ws.send(OptimizedJsonMapper.toJson(session.open())) + + val have = mutableSetOf() + val need = mutableSetOf() + var rounds = 0 + while (rounds < maxRounds) { + val raw = incoming.receive() + rounds++ + when (val msg = OptimizedJsonMapper.fromJsonToMessage(raw)) { + is NegErrMessage -> return@withTimeoutOrNull Diff(have.toList(), need.toList(), rounds, "${msg.subId}: ${msg.reason}") + is NegMsgMessage -> { + val r = session.processMessage(msg.message) + have += r.haveIds + need += r.needIds + if (r.isComplete()) { + return@withTimeoutOrNull Diff(have.toList(), need.toList(), rounds, null) + } + ws.send(OptimizedJsonMapper.toJson(r.nextCmd!!)) + } + else -> {} // ignore NOTICE/etc. and keep waiting + } + } + Diff(have.toList(), need.toList(), rounds, "did not converge in $maxRounds rounds") + } + } finally { + ws.close(1000, "amy-sync-done") + } + } +}