laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(buzz): invite-link redemption + live-interop fixes (amy)

Browse Source 
Validated against the production relay wss://amethyst.communities.buzz.xyz
by joining and running a full DM round-trip. Adds the join primitive and
fixes the interop gaps that live testing surfaced.

- quartz: BuzzInviteLink — parse `https://<host>/invite/<token>` (relay-signed
  base64url payload → community/role/expiry). A Buzz invite is NOT a NIP-29
  code; it is redeemed over HTTP against the tenant host. Unit-tested with a
  real token; rejects the Concord `/invite/<naddr>#…` shape (no collision).
- cli: `amy buzz join <invite-url>` — the real 3-step claim: GET /api/join-policy,
  POST /api/invites/accept-policy, then NIP-98-signed POST /api/invites/claim.
  Proven live (status: joined, role: member).
- cli: Context.publish now authenticates-then-retries on an `auth-required`
  relay (warm the connection with a pendingOnAuthRequired REQ, then re-publish)
  — the write path had no NIP-42 handling, so every Buzz write was rejected.
- cli: Buzz reads (dm list / read / console / personas) use the auth-aware
  drain (pendingOnAuthRequired).
- cli: `dm open` surfaces the relay's synchronous OK `response:{channel_id}` —
  the authoritative DM channel id (the relay assigns it; it is not polled).
- cli: `dm list` rewritten to the relay's actual discovery — kind-44100
  member-added notifications (#p=me) filtered to the kind-40099 `dm_created`
  channels. The deployed relay does NOT emit kind-41001.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J8KBSw6smQRyXLiWHeDsZ8
This commit is contained in:
Claude
2026-07-22 14:56:37 +00:00
parent c79e7d361f
commit bb39fb29fc
5 changed files with 428 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/Context.kt
+24 -1
View File
@@ -523,7 +523,30 @@ class Context(
// event in the local cache.
verifyAndStore(event)
if (relayList.isEmpty()) return emptyMap()
return client.publishAndCollectResults(event, relayList, timeoutSecs)
var results = client.publishAndCollectResults(event, relayList, timeoutSecs)
// NIP-42 write path: an auth-required relay rejects the FIRST publish with
// `auth-required`. The publish opens a connection, races the async AUTH reply, and
// tears down before it lands — and [relayAuth] only re-auths on a REQ `CLOSED`, never
// on a rejected EVENT `OK`. The READ path, though, holds the connection open and its
// `auth-required:` CLOSED reliably drives a re-auth (`reauthenticateIfAuthRequired`),
// leaving the pooled connection authenticated. So on `auth-required` we warm auth with
// a tiny `pendingOnAuthRequired` REQ, then retry the publish on the now-authed socket.
var attempt = 0
while (attempt < 4) {
val needAuth =
results
.filterValues { !it.accepted && it.message.contains("auth-required", ignoreCase = true) }
.keys
if (needAuth.isEmpty()) break
// A cheap REQ whose only purpose is to force the AUTH handshake to completion.
val warmFilter = listOf(Filter(kinds = listOf(event.kind), limit = 1))
drain(needAuth.associateWith { warmFilter }, timeoutMs = 8_000, pendingOnAuthRequired = true)
results = results + client.publishAndCollectResults(event, needAuth, timeoutSecs)
attempt++
}
return results
}
/**
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BuzzCommands.kt
+195 -14
View File
@@ -28,9 +28,10 @@ import com.vitorpamplona.amethyst.commons.model.buzz.AgentFleetAggregator
import com.vitorpamplona.quartz.buzz.amTurnMetrics.AgentTurnMetricEvent
import com.vitorpamplona.quartz.buzz.apPersonas.PersonaEvent
import com.vitorpamplona.quartz.buzz.dm.DmAddMemberEvent
import com.vitorpamplona.quartz.buzz.dm.DmCreatedEvent
import com.vitorpamplona.quartz.buzz.dm.DmHideEvent
import com.vitorpamplona.quartz.buzz.dm.DmOpenEvent
import com.vitorpamplona.quartz.buzz.invite.BuzzInviteLink
import com.vitorpamplona.quartz.buzz.notifications.MemberAddedNotificationEvent
import com.vitorpamplona.quartz.buzz.oaOwnerAttestation.AttestationConditions
import com.vitorpamplona.quartz.buzz.oaOwnerAttestation.OwnerAttestation
import com.vitorpamplona.quartz.buzz.stream.StreamMessageV2Event
@@ -39,7 +40,20 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.core.isValid
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip19Bech32.decodePublicKeyAsHexOrNull
import com.vitorpamplona.quartz.nip98HttpAuth.HTTPAuthorizationEvent
import com.vitorpamplona.quartz.nipC7Chats.ChatEvent
import com.vitorpamplona.quartz.utils.TimeUtils
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.jsonObject
import kotlinx.serialization.json.jsonPrimitive
import kotlinx.serialization.json.put
import okhttp3.MediaType.Companion.toMediaType
import okhttp3.OkHttpClient
import okhttp3.Request
import okhttp3.RequestBody.Companion.toRequestBody
/**
* `amy buzz …` — first-class access to the `block/buzz` workspace protocol, driving the
@@ -51,6 +65,7 @@ import com.vitorpamplona.quartz.nipC7Chats.ChatEvent
object BuzzCommands {
private val USAGE =
"""
|amy buzz join <invite-url> [--accept-policy] redeem a Buzz invite link (HTTP claim, NIP-98)
|amy buzz post RELAY GID <text> post a kind-40002 stream message
|amy buzz read RELAY GID [--limit N] read recent workspace messages (9/40002/40099)
| [--timeout SECS]
@@ -60,7 +75,7 @@ object BuzzCommands {
| [--timeout SECS]
|amy buzz personas [--relays R,R] list my kind-30175 personas
| [--timeout SECS]
|amy buzz dm list [--relays R,R] list my DMs (kind-41001, #p = me)
|amy buzz dm list [--relays R,R] list my DMs (44100 member-added + 40099 dm_created)
| [--limit N] [--timeout SECS]
|amy buzz dm open RELAY PUBKEY [PUBKEY…] open a DM with 1-8 people (kind-41010)
|amy buzz dm hide RELAY CHANNEL hide a DM from my sidebar (kind-41012)
@@ -76,6 +91,7 @@ object BuzzCommands {
tail,
USAGE,
mapOf(
"join" to { rest -> join(dataDir, rest) },
"post" to { rest -> post(dataDir, rest) },
"read" to { rest -> read(dataDir, rest) },
"attest" to { rest -> attest(dataDir, rest) },
@@ -110,7 +126,7 @@ object BuzzCommands {
)
}
/** `buzz dm list` → drains the relay-signed kind-41001 confirmations addressed to me. */
/** `buzz dm list` → discovers my DM channels via kind-44100 member-added notifications, filtered to the kind-40099 `dm_created` ones. */
private suspend fun dmList(
dataDir: DataDir,
rest: Array<String>,
@@ -127,20 +143,54 @@ object BuzzCommands {
val relays = relaysFor(ctx, relaysFlag)
if (relays.isEmpty()) return Output.error("no_relays", "no relays: pass --relays ws://…")
val filter = Filter(kinds = listOf(DmCreatedEvent.KIND), tags = mapOf("p" to listOf(me)), limit = limit)
// The deployed Buzz relay does NOT emit kind-41001; instead it (a) confirms a DM's
// channel id synchronously in the open OK, and (b) addresses each member a kind-44100
// member-added notification (`#p` = me, `h` = channel). So discover the channels I'm in
// via 44100, then keep those whose kind-40099 system message marks them a DM.
val memberFilter = Filter(kinds = listOf(MemberAddedNotificationEvent.KIND), tags = mapOf("p" to listOf(me)))
val channelIds =
ctx
.drain(relays.associateWith { listOf(memberFilter) }, timeoutSecs * 1000, pendingOnAuthRequired = true)
.map { it.second }
.filterIsInstance<MemberAddedNotificationEvent>()
.mapNotNull { it.channel() }
.distinct()
if (channelIds.isEmpty()) {
Output.emit(mapOf("count" to 0, "dms" to emptyList<Any>()))
return 0
}
val sysFilter = Filter(kinds = listOf(SystemMessageEvent.KIND), tags = mapOf("h" to channelIds))
val dms =
ctx
.drainAllPages(relays.associateWith { listOf(filter) }, timeoutSecs * 1000)
.drain(relays.associateWith { listOf(sysFilter) }, timeoutSecs * 1000, pendingOnAuthRequired = true)
.map { it.second }
.filterIsInstance<DmCreatedEvent>()
.distinctBy { it.id }
.filterIsInstance<SystemMessageEvent>()
.filter { it.payload()?.type == "dm_created" }
.distinctBy { it.channel() }
.sortedByDescending { it.createdAt }
.take(limit)
.map {
.map { sys ->
// The relay's dm_created content carries a `participants` array our
// SystemMessagePayload model drops; read it from the raw content.
val participants =
runCatching {
jsonParser
.parseToJsonElement(sys.content)
.jsonObject["participants"]
?.let { arr ->
arr
.toString()
.trim('[', ']')
.split(",")
.map { it.trim().trim('"') }
.filter { it.isNotBlank() }
}
}.getOrNull().orEmpty()
mapOf(
"dm_id" to it.dmId(),
"participants" to it.participants(),
"created_at" to it.createdAt,
"dm_id" to sys.channel(),
"participants" to participants,
"created_at" to sys.createdAt,
)
}
Output.emit(mapOf("count" to dms.size, "dms" to dms))
@@ -177,6 +227,23 @@ object BuzzCommands {
val signed = ctx.signer.sign(DmOpenEvent.build(participants))
val ack = ctx.publish(signed, setOf(relay))
RawEventSupport.publishGuard(ack, signed.id)?.let { return it }
// The relay confirms a DM open synchronously in the OK message as
// `response:{"channel_id":"…","created":bool}` — the authoritative, relay-assigned
// channel UUID. Surface it so callers open the chat straight from the ack.
val okMessage = ack.values.firstOrNull { it.accepted }?.message
val channelId =
okMessage
?.substringAfter("response:", "")
?.takeIf { it.isNotBlank() }
?.let {
runCatching {
jsonParser
.parseToJsonElement(it)
.jsonObject["channel_id"]
?.jsonPrimitive
?.content
}.getOrNull()
}
Output.emit(
mapOf(
"event_id" to signed.id,
@@ -184,6 +251,8 @@ object BuzzCommands {
"relay" to relay.url,
"participants" to participants,
"published" to ack.values.any { it.accepted },
"channel_id" to channelId,
"relay_message" to okMessage,
),
)
return 0
@@ -214,6 +283,118 @@ object BuzzCommands {
}
}
private val jsonMedia = "application/json".toMediaType()
private val jsonParser = Json { ignoreUnknownKeys = true }
/**
* `buzz join <invite-url> [--accept-policy]` → redeems a Buzz workspace invite. A Buzz
* invite is NOT a NIP-29 code; it's a relay-signed token claimed over HTTP against the
* tenant host: optionally accept the join policy, then `POST /api/invites/claim`
* NIP-98-signed by our key. On success we become a relay member and can REQ/publish.
*/
private suspend fun join(
dataDir: DataDir,
rest: Array<String>,
): Int {
val args = Args(rest)
val usage = "buzz join <invite-url> [--accept-policy]"
val urlArg = args.positionalOrNull(0) ?: return Output.error("bad_args", usage)
val acceptPolicy = args.bool("accept-policy")
args.rejectUnknown("accept-policy")
val invite = BuzzInviteLink.parse(urlArg) ?: return Output.error("bad_args", "not a Buzz invite link: $urlArg")
if (invite.isExpired(TimeUtils.now())) return Output.error("expired", "this invite has expired")
Context.open(dataDir).use { ctx ->
ctx.prepare()
val http = ctx.okhttp
// 1. Join policy (optional). A relay with a configured policy requires an accepted
// receipt before the claim; one without returns 404 and we skip straight to claim.
var policyReceipt: String? = null
val (policyCode, policyBody) = httpGet(http, "${invite.httpBase()}/api/join-policy")
if (policyCode == 200) {
val policy = jsonParser.parseToJsonElement(policyBody).jsonObject["policy"]?.jsonObject
val version = policy?.get("version")?.jsonPrimitive?.content
if (version != null) {
if (!acceptPolicy) {
return Output.error(
"policy_required",
"this workspace requires accepting its terms + age attestation; re-run with --accept-policy to consent",
)
}
// 2. Accept policy → short-lived, invite-bound receipt (no NIP-98 auth).
val acceptReq =
buildJsonObject {
put("code", invite.code)
put("policy_version", version)
put("age_confirmed", true)
}.toString()
val (acceptCode, acceptBody) = httpPost(http, "${invite.httpBase()}/api/invites/accept-policy", acceptReq, null)
if (acceptCode != 200) return Output.error("policy_failed", "accept-policy failed ($acceptCode): $acceptBody")
policyReceipt =
jsonParser
.parseToJsonElement(acceptBody)
.jsonObject["receipt"]
?.jsonPrimitive
?.content
?: return Output.error("policy_failed", "accept-policy returned no receipt: $acceptBody")
}
}
// 3. Claim — NIP-98-signed POST. The `u` tag must equal the tenant-host URL.
val claimUrl = "${invite.httpBase()}/api/invites/claim"
val claimReq =
buildJsonObject {
put("code", invite.code)
policyReceipt?.let { put("policy_receipt", it) }
}.toString()
val authEvent = ctx.signer.sign(HTTPAuthorizationEvent.build(claimUrl, "POST", claimReq.encodeToByteArray()))
val (claimCode, claimBody) = httpPost(http, claimUrl, claimReq, authEvent.toAuthToken())
if (claimCode != 200) return Output.error("claim_failed", "invite claim failed ($claimCode): $claimBody")
val result = jsonParser.parseToJsonElement(claimBody).jsonObject
Output.emit(
mapOf(
"status" to result["status"]?.jsonPrimitive?.content,
"community_id" to (result["community_id"]?.jsonPrimitive?.content ?: invite.communityId),
"role" to (result["role"]?.jsonPrimitive?.content ?: invite.role),
"host" to invite.host,
"relay" to invite.relayUrl(),
),
)
return 0
}
}
private suspend fun httpGet(
http: OkHttpClient,
url: String,
): Pair<Int, String> =
withContext(Dispatchers.IO) {
http
.newCall(
Request
.Builder()
.url(url)
.get()
.build(),
).execute()
.use { it.code to (it.body?.string() ?: "") }
}
private suspend fun httpPost(
http: OkHttpClient,
url: String,
body: String,
auth: String?,
): Pair<Int, String> =
withContext(Dispatchers.IO) {
val builder = Request.Builder().url(url).post(body.toRequestBody(jsonMedia))
if (auth != null) builder.header("Authorization", auth)
http.newCall(builder.build()).execute().use { it.code to (it.body?.string() ?: "") }
}
/** `buzz post RELAY GID <text>` → publishes a kind-40002 stream message with an `h` tag. */
private suspend fun post(
dataDir: DataDir,
@@ -251,7 +432,7 @@ object BuzzCommands {
)
val messages =
ctx
.drain(mapOf(relay to listOf(filter)), timeoutSecs * 1000)
.drain(mapOf(relay to listOf(filter)), timeoutSecs * 1000, pendingOnAuthRequired = true)
.map { it.second }
.distinctBy { it.id }
.sortedByDescending { it.createdAt }
@@ -336,7 +517,7 @@ object BuzzCommands {
val filter = Filter(kinds = listOf(AgentTurnMetricEvent.KIND), tags = mapOf("p" to listOf(me)))
val decrypted =
ctx
.drainAllPages(relays.associateWith { listOf(filter) }, timeoutSecs * 1000)
.drain(relays.associateWith { listOf(filter) }, timeoutSecs * 1000, pendingOnAuthRequired = true)
.map { it.second }
.filterIsInstance<AgentTurnMetricEvent>()
.distinctBy { it.id }
@@ -390,7 +571,7 @@ object BuzzCommands {
val filter = Filter(kinds = listOf(PersonaEvent.KIND), authors = listOf(me))
val personas =
ctx
.drainAllPages(relays.associateWith { listOf(filter) }, timeoutSecs * 1000)
.drain(relays.associateWith { listOf(filter) }, timeoutSecs * 1000, pendingOnAuthRequired = true)
.map { it.second }
.filterIsInstance<PersonaEvent>()
// Newest per addressable slug (replaceable): keep the latest for each d tag.