mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
feat(relay): add buzz-admin member management CLI with NIP-43 roster publish (#1265)
Signed-off-by: Will Pfleger <pfleger.will@gmail.com> Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
@@ -15,9 +15,18 @@ path = "src/main.rs"
|
||||
buzz-db = { workspace = true }
|
||||
buzz-core = { workspace = true }
|
||||
buzz-auth = { workspace = true }
|
||||
buzz-pubsub = { workspace = true }
|
||||
buzz-search = { workspace = true }
|
||||
buzz-audit = { workspace = true }
|
||||
buzz-workflow = { workspace = true }
|
||||
buzz-media = { workspace = true }
|
||||
nostr = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
anyhow = { workspace = true }
|
||||
hex = { workspace = true }
|
||||
deadpool-redis = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
sqlx = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
clap = { version = "4", features = ["derive"] }
|
||||
|
||||
@@ -2,14 +2,33 @@
|
||||
|
||||
//! Buzz instance administration CLI.
|
||||
//!
|
||||
//! In the pure Nostr architecture, API tokens no longer exist.
|
||||
//! Admin operations are performed via signed Nostr events (NIP-43 relay admin commands).
|
||||
//! This binary is retained as a placeholder for future admin tooling.
|
||||
//! # Member management (NIP-43)
|
||||
//!
|
||||
//! ## Why only kind:13534 (membership list), not kind:8000/8001 (deltas)
|
||||
//!
|
||||
//! CLI intentionally does not emit kind 8000/8001 deltas —
|
||||
//! `publish_nip43_delta` is in-process-only (no Redis hop), so a sidecar call
|
||||
//! stores but never pushes. The 13534 list snapshot is the authoritative roster
|
||||
//! and rides Redis to live clients. Do not wire a delta call that passes
|
||||
//! in-process tests and silently no-ops in the deployed `compose exec` path.
|
||||
//!
|
||||
//! ## Same-second domination guard
|
||||
//!
|
||||
//! The `custom_created_at = max(now, newest_existing_13534 + 1s)` bump defeats
|
||||
//! same-second domination for serial invocations; it does NOT serialize
|
||||
//! concurrent CLI processes — two near-simultaneous adds can read the same
|
||||
//! newest timestamp and collide on the bumped second. run.sh serialization is
|
||||
//! the guard against parallel adds (e.g. `xargs -P`).
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::Result;
|
||||
use buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST;
|
||||
use buzz_db::{Db, DbConfig};
|
||||
use buzz_pubsub::PubSubManager;
|
||||
use clap::{Parser, Subcommand};
|
||||
use nostr::Keys;
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag};
|
||||
use tracing::warn;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "buzz-admin", about = "Buzz instance administration")]
|
||||
@@ -21,15 +40,35 @@ struct Cli {
|
||||
#[derive(Subcommand)]
|
||||
enum Command {
|
||||
/// Add a pubkey to the relay membership list.
|
||||
///
|
||||
/// Accepts a bech32 npub or 64-char hex pubkey. After inserting the DB row,
|
||||
/// publishes a kind:13534 membership roster via Redis so live clients see
|
||||
/// the updated list immediately.
|
||||
AddMember {
|
||||
/// Nostr public key (hex) to add.
|
||||
/// Nostr public key — bech32 npub or 64-char hex.
|
||||
#[arg(long)]
|
||||
pubkey: String,
|
||||
|
||||
/// Role: "admin" or "member" (default: member).
|
||||
/// Role: "admin" or "member" (default: member). Cannot be "owner" —
|
||||
/// use RELAY_OWNER_PUBKEY config to set the relay owner.
|
||||
#[arg(long, default_value = "member")]
|
||||
role: String,
|
||||
},
|
||||
/// Remove a pubkey from the relay membership list.
|
||||
///
|
||||
/// Accepts a bech32 npub or 64-char hex pubkey. After removing the DB row,
|
||||
/// publishes a kind:13534 membership roster via Redis. Cannot remove the
|
||||
/// relay owner — change RELAY_OWNER_PUBKEY config instead.
|
||||
RemoveMember {
|
||||
/// Nostr public key — bech32 npub or 64-char hex.
|
||||
#[arg(long)]
|
||||
pubkey: String,
|
||||
|
||||
/// Only remove if the member's current role matches this value.
|
||||
/// Omit to remove regardless of role.
|
||||
#[arg(long)]
|
||||
role: Option<String>,
|
||||
},
|
||||
/// List all relay members.
|
||||
ListMembers,
|
||||
/// Generate a new Nostr keypair (for bootstrapping).
|
||||
@@ -51,53 +90,285 @@ enum Command {
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<()> {
|
||||
async fn main() {
|
||||
let cli = Cli::parse();
|
||||
|
||||
let code = match run(cli).await {
|
||||
Ok(code) => code,
|
||||
Err(e) => {
|
||||
eprintln!("error: {e}");
|
||||
5
|
||||
}
|
||||
};
|
||||
std::process::exit(code);
|
||||
}
|
||||
|
||||
async fn run(cli: Cli) -> Result<i32> {
|
||||
match cli.command {
|
||||
Command::GenerateKey => {
|
||||
let keys = Keys::generate();
|
||||
println!("Public key: {}", keys.public_key().to_hex());
|
||||
println!("Secret key: {}", keys.secret_key().display_secret());
|
||||
println!("\nSet BUZZ_PRIVATE_KEY to the secret key to use this identity.");
|
||||
Ok(0)
|
||||
}
|
||||
Command::Migrate => {
|
||||
let db = connect_db().await?;
|
||||
db.migrate().await?;
|
||||
println!("Database migrations complete.");
|
||||
Ok(0)
|
||||
}
|
||||
Command::AddMember { pubkey, role } => {
|
||||
let db = connect_db().await?;
|
||||
let pk_bytes = hex::decode(&pubkey)?;
|
||||
if pk_bytes.len() != 32 {
|
||||
anyhow::bail!("pubkey must be 32 bytes (64 hex chars)");
|
||||
}
|
||||
db.ensure_user(&pk_bytes).await?;
|
||||
// Add to relay members via DB (admin bootstrap — normally done via kind:9030)
|
||||
db.add_relay_member(&pubkey, &role, None).await?;
|
||||
println!("Added {} as {} to relay membership list.", pubkey, role);
|
||||
}
|
||||
Command::ListMembers => {
|
||||
let db = connect_db().await?;
|
||||
let members = db.list_relay_members().await?;
|
||||
if members.is_empty() {
|
||||
println!("No relay members found.");
|
||||
} else {
|
||||
println!("{:<66} {:<10}", "Pubkey", "Role");
|
||||
println!("{}", "-".repeat(78));
|
||||
for m in &members {
|
||||
println!("{:<66} {:<10}", hex::encode(&m.pubkey), m.role);
|
||||
}
|
||||
}
|
||||
}
|
||||
Command::AddMember { pubkey, role } => cmd_add_member(pubkey, role).await,
|
||||
Command::RemoveMember { pubkey, role } => cmd_remove_member(pubkey, role).await,
|
||||
Command::ListMembers => cmd_list_members().await,
|
||||
Command::ReconcileChannels { relay_key } => {
|
||||
reconcile_channels(relay_key).await?;
|
||||
Ok(0)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Member management subcommands ─────────────────────────────────────────────
|
||||
|
||||
async fn cmd_add_member(pubkey_arg: String, role: String) -> Result<i32> {
|
||||
if let Err(msg) = validate_role(&role) {
|
||||
eprintln!("error: {msg}");
|
||||
return Ok(1);
|
||||
}
|
||||
|
||||
let pubkey_hex = match parse_pubkey_hex(&pubkey_arg) {
|
||||
Ok(h) => h,
|
||||
Err(msg) => {
|
||||
eprintln!("error: {msg}");
|
||||
return Ok(1);
|
||||
}
|
||||
};
|
||||
|
||||
let (db, pubsub, relay_keypair) = connect_member_services().await?;
|
||||
|
||||
match db.add_relay_member(&pubkey_hex, &role, None).await {
|
||||
Ok(true) => println!("added {pubkey_hex} as {role}"),
|
||||
Ok(false) => println!("already a member: {pubkey_hex} (no change)"),
|
||||
Err(e) => {
|
||||
eprintln!("error: DB write failed: {e}");
|
||||
return Ok(5);
|
||||
}
|
||||
}
|
||||
|
||||
if let Err(e) = publish_membership_list_with_bump(&db, &pubsub, &relay_keypair).await {
|
||||
eprintln!("warning: member added to DB but list publish failed: {e}");
|
||||
}
|
||||
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
async fn cmd_remove_member(pubkey_arg: String, role_filter: Option<String>) -> Result<i32> {
|
||||
if let Some(ref role) = role_filter {
|
||||
if let Err(msg) = validate_role(role) {
|
||||
eprintln!("error: {msg}");
|
||||
return Ok(1);
|
||||
}
|
||||
}
|
||||
|
||||
let pubkey_hex = match parse_pubkey_hex(&pubkey_arg) {
|
||||
Ok(h) => h,
|
||||
Err(msg) => {
|
||||
eprintln!("error: {msg}");
|
||||
return Ok(1);
|
||||
}
|
||||
};
|
||||
|
||||
let (db, pubsub, relay_keypair) = connect_member_services().await?;
|
||||
|
||||
use buzz_db::relay_members::RemoveResult;
|
||||
let result = if let Some(ref role) = role_filter {
|
||||
db.remove_relay_member_if_role(&pubkey_hex, role).await
|
||||
} else {
|
||||
db.remove_relay_member(&pubkey_hex).await
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(RemoveResult::Removed) => println!("removed {pubkey_hex}"),
|
||||
Ok(RemoveResult::NotFound) => {
|
||||
eprintln!("error: member not found: {pubkey_hex}");
|
||||
return Ok(2);
|
||||
}
|
||||
Ok(RemoveResult::IsOwner) => {
|
||||
eprintln!(
|
||||
"error: cannot remove relay owner: {pubkey_hex}\n\
|
||||
To change the owner, update RELAY_OWNER_PUBKEY and restart."
|
||||
);
|
||||
return Ok(3);
|
||||
}
|
||||
Ok(RemoveResult::RoleMismatch) => {
|
||||
let role_str = role_filter.as_deref().unwrap_or("(unknown)");
|
||||
eprintln!("error: role mismatch — {pubkey_hex} is not currently '{role_str}'");
|
||||
return Ok(4);
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("error: DB write failed: {e}");
|
||||
return Ok(5);
|
||||
}
|
||||
}
|
||||
|
||||
if let Err(e) = publish_membership_list_with_bump(&db, &pubsub, &relay_keypair).await {
|
||||
eprintln!("warning: member removed from DB but list publish failed: {e}");
|
||||
}
|
||||
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
async fn cmd_list_members() -> Result<i32> {
|
||||
let db = connect_db().await?;
|
||||
let members = db.list_relay_members().await?;
|
||||
|
||||
if members.is_empty() {
|
||||
println!("(no relay members)");
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
println!(
|
||||
"{:<66} {:<8} {:<66} created_at",
|
||||
"pubkey", "role", "added_by"
|
||||
);
|
||||
println!("{}", "-".repeat(160));
|
||||
for m in &members {
|
||||
let added_by = m.added_by.as_deref().unwrap_or("-");
|
||||
println!(
|
||||
"{:<66} {:<8} {:<66} {}",
|
||||
m.pubkey,
|
||||
m.role,
|
||||
added_by,
|
||||
m.created_at.format("%Y-%m-%dT%H:%M:%SZ")
|
||||
);
|
||||
}
|
||||
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
// ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Validate that `role` is `"member"` or `"admin"`. Rejects `"owner"`.
|
||||
fn validate_role(role: &str) -> std::result::Result<(), String> {
|
||||
match role {
|
||||
"member" | "admin" => Ok(()),
|
||||
"owner" => {
|
||||
Err("role 'owner' cannot be set via CLI — use RELAY_OWNER_PUBKEY config".to_string())
|
||||
}
|
||||
other => Err(format!(
|
||||
"invalid role '{other}': must be 'member' or 'admin'"
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse a bech32 npub or 64-char hex pubkey into lowercase hex.
|
||||
fn parse_pubkey_hex(input: &str) -> std::result::Result<String, String> {
|
||||
nostr::PublicKey::parse(input)
|
||||
.map(|pk| pk.to_hex())
|
||||
.map_err(|e| format!("invalid pubkey '{input}': {e}"))
|
||||
}
|
||||
|
||||
/// Publish kind:13534 with `custom_created_at = max(now, newest_existing + 1s)`.
|
||||
///
|
||||
/// Guarantees the new event is not dominated by a same-second prior invocation,
|
||||
/// so `replace_addressable_event` always inserts and dispatches to Redis.
|
||||
///
|
||||
/// See module-level doc for the TOCTOU caveat on concurrent CLI processes.
|
||||
async fn publish_membership_list_with_bump(
|
||||
db: &Db,
|
||||
pubsub: &Arc<PubSubManager>,
|
||||
relay_keypair: &Keys,
|
||||
) -> Result<()> {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
|
||||
let relay_pubkey = relay_keypair.public_key();
|
||||
let relay_pubkey_bytes = relay_pubkey.to_bytes();
|
||||
|
||||
// Query the newest existing kind:13534 for this relay's pubkey (channel_id=None).
|
||||
let newest_ts = db
|
||||
.get_latest_global_replaceable(KIND_NIP43_MEMBERSHIP_LIST as i32, &relay_pubkey_bytes)
|
||||
.await?
|
||||
.map(|e| e.event.created_at.as_secs());
|
||||
|
||||
// custom_created_at = max(now, existing + 1s) — defeats same-second domination.
|
||||
let ts = match newest_ts {
|
||||
Some(existing) => (existing + 1).max(now),
|
||||
None => now,
|
||||
};
|
||||
|
||||
let members = db.list_relay_members().await?;
|
||||
|
||||
let mut tags: Vec<Tag> = Vec::with_capacity(members.len() + 1);
|
||||
// NIP-70 protected-event marker — prevents re-broadcasting by third parties.
|
||||
tags.push(Tag::parse(["-"]).map_err(|e| anyhow::anyhow!("failed to build '-' tag: {e}"))?);
|
||||
for member in &members {
|
||||
tags.push(
|
||||
Tag::parse(["member", &member.pubkey, &member.role])
|
||||
.map_err(|e| anyhow::anyhow!("failed to build member tag: {e}"))?,
|
||||
);
|
||||
}
|
||||
|
||||
let event = EventBuilder::new(Kind::Custom(KIND_NIP43_MEMBERSHIP_LIST as u16), "")
|
||||
.tags(tags)
|
||||
.custom_created_at(nostr::Timestamp::from(ts))
|
||||
.sign_with_keys(relay_keypair)
|
||||
.map_err(|e| anyhow::anyhow!("failed to sign kind:13534: {e}"))?;
|
||||
|
||||
let (stored, was_inserted) = db.replace_addressable_event(&event, None).await?;
|
||||
if was_inserted {
|
||||
// Publish to Redis so live clients receive the updated roster.
|
||||
// Uses channel_id=Nil (global scope) matching the relay's own publish path.
|
||||
let pubsub_channel = uuid::Uuid::nil();
|
||||
if let Err(e) = pubsub.publish_event(pubsub_channel, &stored.event).await {
|
||||
warn!("Redis publish of kind:13534 failed: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
member_count = members.len(),
|
||||
ts,
|
||||
"NIP-43 membership list published by buzz-admin"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Connect to DB, Redis pub/sub, and load the relay keypair.
|
||||
///
|
||||
/// `BUZZ_RELAY_PRIVATE_KEY` is required — the CLI signs kind:13534 events.
|
||||
async fn connect_member_services() -> Result<(Db, Arc<PubSubManager>, Keys)> {
|
||||
let db = connect_db().await?;
|
||||
|
||||
let relay_keypair = {
|
||||
let hex = std::env::var("BUZZ_RELAY_PRIVATE_KEY").map_err(|_| {
|
||||
anyhow::anyhow!(
|
||||
"BUZZ_RELAY_PRIVATE_KEY is required for add-member/remove-member.\n\
|
||||
The relay must have a stable signing key to publish kind:13534 events."
|
||||
)
|
||||
})?;
|
||||
Keys::parse(&hex).map_err(|e| anyhow::anyhow!("invalid BUZZ_RELAY_PRIVATE_KEY: {e}"))?
|
||||
};
|
||||
|
||||
let redis_url =
|
||||
std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://localhost:6379".to_string());
|
||||
|
||||
let redis_pool = {
|
||||
let cfg = deadpool_redis::Config::from_url(&redis_url);
|
||||
cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1))
|
||||
.map_err(|e| anyhow::anyhow!("Redis pool creation failed: {e}"))?
|
||||
};
|
||||
|
||||
let pubsub = Arc::new(
|
||||
PubSubManager::new(&redis_url, redis_pool)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("PubSub init failed: {e}"))?,
|
||||
);
|
||||
|
||||
Ok((db, pubsub, relay_keypair))
|
||||
}
|
||||
|
||||
async fn connect_db() -> Result<Db> {
|
||||
let db_url = std::env::var("DATABASE_URL")
|
||||
.unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
|
||||
@@ -112,7 +383,6 @@ async fn connect_db() -> Result<Db> {
|
||||
async fn reconcile_channels(relay_key_arg: Option<String>) -> Result<()> {
|
||||
use buzz_core::kind::KIND_NIP29_GROUP_ADMINS;
|
||||
use buzz_db::event::EventQuery;
|
||||
use nostr::{EventBuilder, Kind, Tag};
|
||||
|
||||
let db = connect_db().await?;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user