mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
feat(relay): add atomic community ownership transfer (#1845)
Signed-off-by: Kalvin Chau <kalvin@block.xyz> Signed-off-by: npub122y0pqkertljmedu303rl0aqrj3w8pvu43t6jxm6875lzg6f2pwqegc3xc <5288f082d91aff2de5bc8be23fbfa01ca2e3859cac57a91b7a3fa9f12349505c@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub1c4alndp82zyt9veaklm5d965quss79vlhk9awv7qu5erwhmf42qqlvc25c <c57bf9b4275088b2b33db7f746975407210f159fbd8bd733c0e532375f69aa80@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub122y0pqkertljmedu303rl0aqrj3w8pvu43t6jxm6875lzg6f2pwqegc3xc <5288f082d91aff2de5bc8be23fbfa01ca2e3859cac57a91b7a3fa9f12349505c@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
@@ -55,6 +55,33 @@ use uuid::Uuid;
|
||||
|
||||
use buzz_core::{CommunityId, StoredEvent};
|
||||
|
||||
fn event_replacement_lock_key(
|
||||
community_id: CommunityId,
|
||||
kind: i32,
|
||||
pubkey: &[u8],
|
||||
coordinate: Option<&[u8]>,
|
||||
) -> i64 {
|
||||
let mut hash: u64 = 0xcbf29ce484222325;
|
||||
let kind_bytes = kind.to_le_bytes();
|
||||
for bytes in [
|
||||
community_id.as_uuid().as_bytes().as_slice(),
|
||||
kind_bytes.as_slice(),
|
||||
pubkey,
|
||||
] {
|
||||
for byte in bytes {
|
||||
hash ^= *byte as u64;
|
||||
hash = hash.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
}
|
||||
if let Some(coordinate) = coordinate {
|
||||
for byte in coordinate {
|
||||
hash ^= *byte as u64;
|
||||
hash = hash.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
}
|
||||
hash as i64
|
||||
}
|
||||
|
||||
/// Extract p-tag mentions from an event and insert into the `event_mentions` table.
|
||||
///
|
||||
/// Called after event insertion. Failures are logged but do not block event storage.
|
||||
@@ -212,6 +239,17 @@ pub struct CreatedCommunityRecord {
|
||||
pub host: String,
|
||||
}
|
||||
|
||||
/// Result of atomically creating a community with its initial owner.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum CreateCommunityWithOwnerResult {
|
||||
/// The community was created, or an identical retried create found it.
|
||||
Created(CreatedCommunityRecord),
|
||||
/// The host already belongs to another owner.
|
||||
HostExists,
|
||||
/// The intended owner already owns the maximum number of communities.
|
||||
LimitReached,
|
||||
}
|
||||
|
||||
/// Community row returned by operator-plane ownership reads.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct OwnedCommunityRecord {
|
||||
@@ -528,18 +566,24 @@ impl Db {
|
||||
|
||||
/// Atomically creates a community and its initial owner.
|
||||
///
|
||||
/// Returns the existing row when the normalized host already has the same
|
||||
/// current owner, making ambiguous create retries naturally idempotent.
|
||||
/// Returns `None` when the host exists with a different (or missing) owner.
|
||||
/// The initial host and owner inserts share one transaction, so callers
|
||||
/// never observe a partially provisioned community or rotate an owner.
|
||||
/// Holds a per-owner advisory lock while enforcing the ownership limit.
|
||||
/// Identical create retries return the original record; host collisions and
|
||||
/// limit failures remain distinguishable to the operator API.
|
||||
pub async fn create_community_with_owner(
|
||||
&self,
|
||||
normalized_host: &str,
|
||||
owner_pubkey: &str,
|
||||
) -> Result<Option<CreatedCommunityRecord>> {
|
||||
) -> Result<CreateCommunityWithOwnerResult> {
|
||||
let owner_pubkey = owner_pubkey.to_ascii_lowercase();
|
||||
let mut tx = self.pool.begin().await?;
|
||||
|
||||
// Serialize on the owner pubkey so concurrent creates to the same
|
||||
// owner cannot both pass the ownership count check.
|
||||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||||
.bind(relay_members::owner_count_advisory_lock_key(&owner_pubkey))
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
let row = sqlx::query(
|
||||
r#"
|
||||
INSERT INTO communities (host)
|
||||
@@ -555,6 +599,20 @@ impl Db {
|
||||
let (id, host) = if let Some(row) = row {
|
||||
let id: Uuid = row.try_get("id")?;
|
||||
let host: String = row.try_get("host")?;
|
||||
|
||||
// Enforce the limit before inserting the new owner row.
|
||||
let owned_count: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM relay_members WHERE pubkey = $1 AND role = 'owner'",
|
||||
)
|
||||
.bind(&owner_pubkey)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
|
||||
if owned_count >= relay_members::MAX_COMMUNITIES_PER_OWNER {
|
||||
tx.rollback().await?;
|
||||
return Ok(CreateCommunityWithOwnerResult::LimitReached);
|
||||
}
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO relay_members (community_id, pubkey, role, added_by) VALUES ($1, $2, 'owner', NULL)",
|
||||
)
|
||||
@@ -580,16 +638,18 @@ impl Db {
|
||||
.await?;
|
||||
let Some(existing) = existing else {
|
||||
tx.rollback().await?;
|
||||
return Ok(None);
|
||||
return Ok(CreateCommunityWithOwnerResult::HostExists);
|
||||
};
|
||||
(existing.try_get("id")?, existing.try_get("host")?)
|
||||
};
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(Some(CreatedCommunityRecord {
|
||||
id: CommunityId::from_uuid(id),
|
||||
host,
|
||||
}))
|
||||
Ok(CreateCommunityWithOwnerResult::Created(
|
||||
CreatedCommunityRecord {
|
||||
id: CommunityId::from_uuid(id),
|
||||
host,
|
||||
},
|
||||
))
|
||||
}
|
||||
|
||||
/// Returns the community that owns a channel, if the channel exists.
|
||||
@@ -2367,6 +2427,25 @@ impl Db {
|
||||
relay_members::bootstrap_owner(&self.pool, community, owner_pubkey).await
|
||||
}
|
||||
|
||||
/// Atomically transfers ownership of `community` to `new_owner_pubkey`,
|
||||
/// demoting the previous owner(s) to `member`. Verifies
|
||||
/// `expected_owner_pubkey` matches the current owner inside the same
|
||||
/// transaction to prevent stale-owner races.
|
||||
pub async fn transfer_ownership(
|
||||
&self,
|
||||
community: CommunityId,
|
||||
new_owner_pubkey: &str,
|
||||
expected_owner_pubkey: &str,
|
||||
) -> Result<relay_members::TransferResult> {
|
||||
relay_members::transfer_ownership(
|
||||
&self.pool,
|
||||
community,
|
||||
new_owner_pubkey,
|
||||
expected_owner_pubkey,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Migrates existing `pubkey_allowlist` entries into `relay_members` for `community`.
|
||||
///
|
||||
/// Idempotent — uses `ON CONFLICT DO NOTHING`. Returns the number of rows
|
||||
@@ -2645,31 +2724,13 @@ impl Db {
|
||||
let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0)
|
||||
.ok_or(DbError::InvalidTimestamp(created_at_secs))?;
|
||||
|
||||
// Stable advisory-lock key: hash (kind, pubkey, channel_id) to i64.
|
||||
// Uses FNV-1a for determinism — Rust's DefaultHasher is NOT stable across processes.
|
||||
// Collisions cause extra serialization, not incorrect behavior.
|
||||
let lock_key = {
|
||||
let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis
|
||||
for b in community_id.as_uuid().as_bytes() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
for b in kind_i32.to_le_bytes() {
|
||||
h ^= b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3); // FNV prime
|
||||
}
|
||||
for b in pubkey_bytes.as_slice() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
if let Some(ch) = channel_id {
|
||||
for b in ch.as_bytes() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
}
|
||||
h as i64
|
||||
};
|
||||
// Collisions only cause extra serialization; they cannot change behavior.
|
||||
let lock_key = event_replacement_lock_key(
|
||||
community_id,
|
||||
kind_i32,
|
||||
pubkey_bytes.as_slice(),
|
||||
channel_id.as_ref().map(|id| id.as_bytes().as_slice()),
|
||||
);
|
||||
|
||||
let mut tx = self.pool.begin().await?;
|
||||
|
||||
@@ -2777,6 +2838,134 @@ impl Db {
|
||||
))
|
||||
}
|
||||
|
||||
/// Atomically publish a NIP-43 membership snapshot under a single
|
||||
/// transaction-scoped advisory lock.
|
||||
///
|
||||
/// This method acquires the per-community snapshot lock, reads the
|
||||
/// current membership, builds the event, and replaces the prior snapshot
|
||||
/// — all inside one transaction on one database connection. This
|
||||
/// prevents the stale-snapshot race where a concurrent publication reads
|
||||
/// older state and overwrites a newer snapshot by arrival order.
|
||||
///
|
||||
pub async fn publish_nip43_membership_locked(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
relay_keypair: &nostr::Keys,
|
||||
) -> Result<(StoredEvent, bool, usize)> {
|
||||
use nostr::{EventBuilder, Kind, Tag};
|
||||
|
||||
let kind_i32 = buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32;
|
||||
let pubkey_bytes = relay_keypair.public_key().to_bytes();
|
||||
|
||||
let lock_key =
|
||||
event_replacement_lock_key(community_id, kind_i32, pubkey_bytes.as_slice(), None);
|
||||
|
||||
let mut tx = self.pool.begin().await?;
|
||||
|
||||
// Acquire the per-community snapshot lock BEFORE reading members.
|
||||
// This serializes the entire read-build-write cycle: a concurrent
|
||||
// publication will block here until our transaction commits, then
|
||||
// read the updated membership state.
|
||||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||||
.bind(lock_key)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
// Read current members inside the locked transaction.
|
||||
let rows = sqlx::query(
|
||||
"SELECT pubkey, role FROM relay_members \
|
||||
WHERE community_id = $1 ORDER BY created_at ASC",
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.fetch_all(&mut *tx)
|
||||
.await?;
|
||||
|
||||
let member_count = rows.len();
|
||||
|
||||
// Build the NIP-43 event from the locked member rows.
|
||||
let mut tags: Vec<Tag> = Vec::with_capacity(member_count + 1);
|
||||
// NIP-70 protected-event marker.
|
||||
tags.push(Tag::parse(["-"]).map_err(|e| {
|
||||
crate::error::DbError::InvalidData(format!("failed to build '-' tag: {e}"))
|
||||
})?);
|
||||
for row in &rows {
|
||||
let pubkey: String = row.try_get("pubkey")?;
|
||||
let role: String = row.try_get("role")?;
|
||||
tags.push(Tag::parse(["member", &pubkey, &role]).map_err(|e| {
|
||||
crate::error::DbError::InvalidData(format!("failed to build member tag: {e}"))
|
||||
})?);
|
||||
}
|
||||
|
||||
let event = EventBuilder::new(Kind::Custom(kind_i32 as u16), "")
|
||||
.tags(tags)
|
||||
.sign_with_keys(relay_keypair)
|
||||
.map_err(|e| {
|
||||
crate::error::DbError::InvalidData(format!("failed to sign kind:13534: {e}"))
|
||||
})?;
|
||||
|
||||
let created_at_secs = event.created_at.as_secs() as i64;
|
||||
let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0)
|
||||
.ok_or(DbError::InvalidTimestamp(created_at_secs))?;
|
||||
let sig_bytes = event.sig.serialize();
|
||||
let tags_json = serde_json::to_value(&event.tags)?;
|
||||
let received_at = chrono::Utc::now();
|
||||
let d_tag = crate::event::extract_d_tag(&event);
|
||||
|
||||
// Soft-delete prior snapshots — unconditional, the relay is authoritative.
|
||||
sqlx::query(
|
||||
"UPDATE events SET deleted_at = NOW() \
|
||||
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \
|
||||
AND channel_id IS NULL \
|
||||
AND deleted_at IS NULL",
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.bind(kind_i32)
|
||||
.bind(pubkey_bytes.as_slice())
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
let insert_result = sqlx::query(
|
||||
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \
|
||||
ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
.bind(event.id.as_bytes().as_slice())
|
||||
.bind(pubkey_bytes.as_slice())
|
||||
.bind(created_at)
|
||||
.bind(kind_i32)
|
||||
.bind(&tags_json)
|
||||
.bind(&event.content)
|
||||
.bind(sig_bytes.as_slice())
|
||||
.bind(received_at)
|
||||
.bind::<Option<Uuid>>(None)
|
||||
.bind(d_tag.as_deref())
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
let was_inserted = insert_result.rows_affected() > 0;
|
||||
if !was_inserted {
|
||||
tx.rollback().await?;
|
||||
return Ok((
|
||||
StoredEvent::with_received_at(event, received_at, None, false),
|
||||
false,
|
||||
member_count,
|
||||
));
|
||||
}
|
||||
|
||||
tx.commit().await?;
|
||||
|
||||
if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await {
|
||||
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
|
||||
}
|
||||
|
||||
Ok((
|
||||
StoredEvent::with_received_at(event, received_at, None, true),
|
||||
true,
|
||||
member_count,
|
||||
))
|
||||
}
|
||||
|
||||
/// Atomically replace a NIP-33 parameterized replaceable event (kind 30000–39999).
|
||||
///
|
||||
/// Keeps only the event with the highest `created_at` per `(kind, pubkey, d_tag)`.
|
||||
@@ -2809,28 +2998,12 @@ impl Db {
|
||||
let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0)
|
||||
.ok_or(DbError::InvalidTimestamp(created_at_secs))?;
|
||||
|
||||
// Stable advisory-lock key: FNV-1a over (kind, pubkey, d_tag).
|
||||
// Same algorithm as replace_addressable_event — deterministic across processes.
|
||||
let lock_key = {
|
||||
let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis
|
||||
for b in community_id.as_uuid().as_bytes() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
for b in kind_i32.to_le_bytes() {
|
||||
h ^= b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
for b in pubkey_bytes.as_slice() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
for b in d_tag.as_bytes() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3);
|
||||
}
|
||||
h as i64
|
||||
};
|
||||
let lock_key = event_replacement_lock_key(
|
||||
community_id,
|
||||
kind_i32,
|
||||
pubkey_bytes.as_slice(),
|
||||
Some(d_tag.as_bytes()),
|
||||
);
|
||||
|
||||
let mut tx = self.pool.begin().await?;
|
||||
|
||||
@@ -3798,8 +3971,10 @@ mod tests {
|
||||
let created = db
|
||||
.create_community_with_owner(&host, owner)
|
||||
.await
|
||||
.expect("create community")
|
||||
.expect("new host");
|
||||
.expect("create community");
|
||||
let CreateCommunityWithOwnerResult::Created(created) = created else {
|
||||
panic!("expected new community");
|
||||
};
|
||||
assert_eq!(created.host, host);
|
||||
let owner_role: Option<String> = sqlx::query_scalar(
|
||||
"SELECT role FROM relay_members WHERE community_id = $1 AND pubkey = $2",
|
||||
@@ -3814,15 +3989,18 @@ mod tests {
|
||||
let retry = db
|
||||
.create_community_with_owner(&host.to_ascii_uppercase(), owner)
|
||||
.await
|
||||
.expect("same-owner retry")
|
||||
.expect("existing same-owner community");
|
||||
assert_eq!(retry, created, "retry returns the original row");
|
||||
.expect("same-owner retry");
|
||||
assert_eq!(
|
||||
retry,
|
||||
CreateCommunityWithOwnerResult::Created(created.clone()),
|
||||
"retry returns the original row"
|
||||
);
|
||||
|
||||
let collision = db
|
||||
.create_community_with_owner(&host, other)
|
||||
.await
|
||||
.expect("collision result");
|
||||
assert!(collision.is_none());
|
||||
assert_eq!(collision, CreateCommunityWithOwnerResult::HostExists);
|
||||
let roles: Vec<(String, String)> = sqlx::query_as(
|
||||
"SELECT pubkey, role FROM relay_members WHERE community_id = $1 ORDER BY pubkey",
|
||||
)
|
||||
@@ -3839,7 +4017,43 @@ mod tests {
|
||||
.create_community_with_owner(&host, owner)
|
||||
.await
|
||||
.expect("post-rotation retry");
|
||||
assert!(post_rotation_retry.is_none());
|
||||
assert_eq!(
|
||||
post_rotation_retry,
|
||||
CreateCommunityWithOwnerResult::HostExists
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn create_community_with_owner_enforces_per_owner_limit() {
|
||||
let db = setup_db().await;
|
||||
let owner = format!("{:064x}", Uuid::new_v4().as_u128());
|
||||
|
||||
// Create 3 communities for this owner (the max).
|
||||
for i in 0..3 {
|
||||
let host = format!("limit-test-{}-{}.example", i, Uuid::new_v4().simple());
|
||||
assert!(matches!(
|
||||
db.create_community_with_owner(&host, &owner)
|
||||
.await
|
||||
.expect("create community"),
|
||||
CreateCommunityWithOwnerResult::Created(_)
|
||||
));
|
||||
}
|
||||
|
||||
let host = format!("limit-test-3-{}.example", Uuid::new_v4().simple());
|
||||
assert_eq!(
|
||||
db.create_community_with_owner(&host, &owner)
|
||||
.await
|
||||
.expect("create community call"),
|
||||
CreateCommunityWithOwnerResult::LimitReached
|
||||
);
|
||||
assert!(
|
||||
db.lookup_community_by_host(&host)
|
||||
.await
|
||||
.expect("look up rolled-back fresh host")
|
||||
.is_none(),
|
||||
"limit rejection must roll back the fresh community row"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -3853,13 +4067,10 @@ mod tests {
|
||||
db.create_community_with_owner(&host, owner),
|
||||
db.create_community_with_owner(&host, owner),
|
||||
);
|
||||
let first = first
|
||||
.expect("first concurrent create")
|
||||
.expect("first result");
|
||||
let second = second
|
||||
.expect("second concurrent create")
|
||||
.expect("second result");
|
||||
let first = first.expect("first concurrent create");
|
||||
let second = second.expect("second concurrent create");
|
||||
|
||||
assert!(matches!(first, CreateCommunityWithOwnerResult::Created(_)));
|
||||
assert_eq!(first, second, "conflict loser re-reads the winning row");
|
||||
}
|
||||
|
||||
|
||||
@@ -248,6 +248,14 @@ pub async fn update_relay_member_role(
|
||||
/// is never bootstrapped into community B.
|
||||
///
|
||||
/// Runs in a single transaction. Safe to call at every startup — idempotent.
|
||||
///
|
||||
/// **Deployment-root authority exception:** This function is called only by
|
||||
/// startup initialization and legacy operator provisioning
|
||||
/// (`community_provisioning.rs`). It is NOT an end-user path and does NOT
|
||||
/// enforce the per-owner community limit (`MAX_COMMUNITIES_PER_OWNER`) or
|
||||
/// acquire the per-recipient advisory lock. The per-owner limit is an
|
||||
/// end-user invariant enforced by `create_community_with_owner` and
|
||||
/// `transfer_ownership`; deployment-root operations may exceed it by design.
|
||||
pub async fn bootstrap_owner(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
@@ -281,6 +289,158 @@ pub async fn bootstrap_owner(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The result of a transfer-ownership attempt.
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum TransferResult {
|
||||
/// Transfer completed: the new owner was upserted and the previous
|
||||
/// owner(s) demoted to `member`.
|
||||
Transferred {
|
||||
/// Pubkey of the previous sole owner, if exactly one existed.
|
||||
previous_owner: Option<String>,
|
||||
},
|
||||
/// The new owner pubkey is already the sole owner — nothing to do.
|
||||
AlreadyOwner,
|
||||
/// No owner row exists for this community (community may not exist).
|
||||
NoOwner,
|
||||
/// The `expected_owner_pubkey` did not match the current owner. A
|
||||
/// concurrent transfer or owner rotation has already changed ownership.
|
||||
/// The caller must NOT retry blindly — re-read ownership and re-evaluate.
|
||||
OwnerConflict,
|
||||
/// The transferee already owns the maximum number of communities.
|
||||
/// Enforced atomically inside the transfer transaction so concurrent
|
||||
/// transfers to the same recipient cannot both pass the limit.
|
||||
LimitReached,
|
||||
}
|
||||
|
||||
/// Maximum number of communities a single pubkey can own. Enforced at the
|
||||
/// relay layer — the authoritative layer — so that concurrent transfers or
|
||||
/// transfer-vs-create races cannot both pass a preflight count.
|
||||
pub const MAX_COMMUNITIES_PER_OWNER: i64 = 3;
|
||||
|
||||
/// Stable advisory-lock key for serializing ownership-granting operations
|
||||
/// (transfer + create) per recipient pubkey. Uses FNV-1a over the hex pubkey
|
||||
/// so the same recipient always maps to the same lock across processes.
|
||||
pub fn owner_count_advisory_lock_key(pubkey_hex: &str) -> i64 {
|
||||
let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis
|
||||
for b in pubkey_hex.as_bytes() {
|
||||
h ^= *b as u64;
|
||||
h = h.wrapping_mul(0x100000001b3); // FNV prime
|
||||
}
|
||||
h as i64
|
||||
}
|
||||
|
||||
/// Atomically transfers ownership of `community` to `new_owner_pubkey`.
|
||||
///
|
||||
/// Runs in a single transaction:
|
||||
/// 1. Acquires a transaction-scoped advisory lock on the *transferee* pubkey
|
||||
/// so that concurrent transfers to the same recipient serialize. The same
|
||||
/// lock key is also used by `Db::create_community_with_owner` to prevent
|
||||
/// transfer-vs-create races.
|
||||
/// 2. Locks the current owner row `FOR UPDATE` and verifies
|
||||
/// `expected_owner_pubkey` matches. This prevents a stale-owner race where
|
||||
/// a delayed/retried request overwrites a completed transfer.
|
||||
/// 3. Enforces the [`MAX_COMMUNITIES_PER_OWNER`] limit on the transferee by
|
||||
/// counting owned communities inside the same transaction.
|
||||
/// 4. Upserts `new_owner_pubkey` as `owner` (insert or promote).
|
||||
/// 5. Demotes every other owner in this community to `member` — **not**
|
||||
/// `admin`, per product decision: the former owner retains no management
|
||||
/// capabilities.
|
||||
///
|
||||
/// Scoped to one community — an ownership transfer in A never touches B.
|
||||
pub async fn transfer_ownership(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
new_owner_pubkey: &str,
|
||||
expected_owner_pubkey: &str,
|
||||
) -> Result<TransferResult> {
|
||||
let pubkey = new_owner_pubkey.to_ascii_lowercase();
|
||||
let expected_owner = expected_owner_pubkey.to_ascii_lowercase();
|
||||
let mut tx = pool.begin().await?;
|
||||
|
||||
// 1. Serialize on the transferee so concurrent transfers to the same
|
||||
// recipient cannot both pass the ownership count check.
|
||||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||||
.bind(owner_count_advisory_lock_key(&pubkey))
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
// 2. Lock the current owner row FOR UPDATE and verify the expected owner.
|
||||
// FOR UPDATE prevents the stale-owner race: a concurrent transfer that
|
||||
// already changed the owner will block on this lock until our txn
|
||||
// completes (or vice versa), and the expected_owner check will fail.
|
||||
let existing_owners: Vec<String> = sqlx::query_scalar(
|
||||
"SELECT pubkey FROM relay_members \
|
||||
WHERE community_id = $1 AND role = 'owner' \
|
||||
FOR UPDATE",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.fetch_all(&mut *tx)
|
||||
.await?;
|
||||
|
||||
if existing_owners.is_empty() {
|
||||
tx.rollback().await?;
|
||||
return Ok(TransferResult::NoOwner);
|
||||
}
|
||||
|
||||
// Stale-owner guard: if the current owner doesn't match the expected
|
||||
// owner, a concurrent transfer or rotation has already changed hands.
|
||||
if !existing_owners.iter().any(|p| p == &expected_owner) {
|
||||
tx.rollback().await?;
|
||||
return Ok(TransferResult::OwnerConflict);
|
||||
}
|
||||
|
||||
// Already the sole owner — no transfer needed.
|
||||
if existing_owners.len() == 1 && existing_owners[0] == pubkey {
|
||||
tx.rollback().await?;
|
||||
return Ok(TransferResult::AlreadyOwner);
|
||||
}
|
||||
|
||||
let previous_owner = if existing_owners.len() == 1 {
|
||||
Some(existing_owners[0].clone())
|
||||
} else {
|
||||
existing_owners.iter().find(|p| **p != pubkey).cloned()
|
||||
};
|
||||
|
||||
// 3. Enforce the transferee's community ownership limit inside the same
|
||||
// transaction that holds the advisory lock. This is the authoritative
|
||||
// check — kgoose's preflight count is advisory only.
|
||||
let owned_count: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM relay_members WHERE pubkey = $1 AND role = 'owner'",
|
||||
)
|
||||
.bind(&pubkey)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
|
||||
if owned_count >= MAX_COMMUNITIES_PER_OWNER {
|
||||
tx.rollback().await?;
|
||||
return Ok(TransferResult::LimitReached);
|
||||
}
|
||||
|
||||
// 4. Upsert the new owner.
|
||||
sqlx::query(
|
||||
"INSERT INTO relay_members (community_id, pubkey, role, added_by) \
|
||||
VALUES ($1, $2, 'owner', NULL) \
|
||||
ON CONFLICT (community_id, pubkey) DO UPDATE SET role = 'owner', updated_at = now()",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(&pubkey)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
// 5. Demote all other owners to member (not admin).
|
||||
sqlx::query(
|
||||
"UPDATE relay_members SET role = 'member', updated_at = now() \
|
||||
WHERE community_id = $1 AND role = 'owner' AND pubkey <> $2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(&pubkey)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(TransferResult::Transferred { previous_owner })
|
||||
}
|
||||
|
||||
/// Migrates existing `pubkey_allowlist` entries into `relay_members` for
|
||||
/// `community` (the deployment's default community).
|
||||
///
|
||||
@@ -360,6 +520,30 @@ mod tests {
|
||||
CommunityId::from_uuid(id)
|
||||
}
|
||||
|
||||
fn test_pubkey() -> String {
|
||||
format!("{:064x}", Uuid::new_v4().as_u128())
|
||||
}
|
||||
|
||||
async fn assert_role(pool: &PgPool, community: CommunityId, pubkey: &str, role: &str) {
|
||||
assert_eq!(
|
||||
get_relay_member(pool, community, pubkey)
|
||||
.await
|
||||
.expect("get relay member")
|
||||
.map(|member| member.role)
|
||||
.as_deref(),
|
||||
Some(role)
|
||||
);
|
||||
}
|
||||
|
||||
async fn owned_community(pool: &PgPool) -> (CommunityId, String) {
|
||||
let community = make_test_community(pool).await;
|
||||
let owner = test_pubkey();
|
||||
bootstrap_owner(pool, community, &owner)
|
||||
.await
|
||||
.expect("bootstrap owner");
|
||||
(community, owner)
|
||||
}
|
||||
|
||||
/// NIP-43 admission confinement: a pubkey admitted to community A is *not*
|
||||
/// admitted to community B. This is the exact mutation #1285 targets — a
|
||||
/// `WHERE pubkey = $1` membership check (no community predicate) would let an
|
||||
@@ -373,7 +557,7 @@ mod tests {
|
||||
let community_a = make_test_community(&pool).await;
|
||||
let community_b = make_test_community(&pool).await;
|
||||
// 64-char lowercase hex, unique per run so reruns don't collide.
|
||||
let pubkey = format!("{:064x}", Uuid::new_v4().as_u128());
|
||||
let pubkey = test_pubkey();
|
||||
|
||||
let inserted = add_relay_member(&pool, community_a, &pubkey, "member", None)
|
||||
.await
|
||||
@@ -437,7 +621,7 @@ mod tests {
|
||||
let pool = setup_pool().await;
|
||||
let community_a = make_test_community(&pool).await;
|
||||
let community_b = make_test_community(&pool).await;
|
||||
let owner = format!("{:064x}", Uuid::new_v4().as_u128());
|
||||
let owner = test_pubkey();
|
||||
|
||||
bootstrap_owner(&pool, community_a, &owner)
|
||||
.await
|
||||
@@ -456,4 +640,220 @@ mod tests {
|
||||
"owner bootstrapped in A must NOT be a member of B"
|
||||
);
|
||||
}
|
||||
|
||||
/// Transfer ownership: upserts new owner, demotes previous owner to
|
||||
/// `member` (not `admin`), and returns the previous owner's pubkey.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_demotes_old_owner_to_member() {
|
||||
let pool = setup_pool().await;
|
||||
let (community, old_owner) = owned_community(&pool).await;
|
||||
let new_owner = test_pubkey();
|
||||
|
||||
let result = transfer_ownership(&pool, community, &new_owner, &old_owner)
|
||||
.await
|
||||
.expect("transfer ownership");
|
||||
|
||||
assert_eq!(
|
||||
result,
|
||||
TransferResult::Transferred {
|
||||
previous_owner: Some(old_owner.clone()),
|
||||
}
|
||||
);
|
||||
|
||||
assert_role(&pool, community, &new_owner, "owner").await;
|
||||
assert_role(&pool, community, &old_owner, "member").await;
|
||||
}
|
||||
|
||||
/// Transferring to the current sole owner is a no-op (`AlreadyOwner`).
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_already_owner_is_noop() {
|
||||
let pool = setup_pool().await;
|
||||
let (community, owner) = owned_community(&pool).await;
|
||||
|
||||
let result = transfer_ownership(&pool, community, &owner, &owner)
|
||||
.await
|
||||
.expect("transfer ownership to self");
|
||||
|
||||
assert_eq!(result, TransferResult::AlreadyOwner);
|
||||
|
||||
assert_role(&pool, community, &owner, "owner").await;
|
||||
}
|
||||
|
||||
/// Transferring a community with no owner row returns `NoOwner`.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_no_owner_returns_no_owner() {
|
||||
let pool = setup_pool().await;
|
||||
let community = make_test_community(&pool).await;
|
||||
let new_owner = test_pubkey();
|
||||
let expected = test_pubkey();
|
||||
|
||||
// No bootstrap — community exists but has no owner row.
|
||||
|
||||
let result = transfer_ownership(&pool, community, &new_owner, &expected)
|
||||
.await
|
||||
.expect("transfer ownership on empty community");
|
||||
|
||||
assert_eq!(result, TransferResult::NoOwner);
|
||||
}
|
||||
|
||||
/// Transfer ownership is community-scoped: transferring in A does not
|
||||
/// affect ownership in B.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_is_community_scoped() {
|
||||
let pool = setup_pool().await;
|
||||
let community_a = make_test_community(&pool).await;
|
||||
let community_b = make_test_community(&pool).await;
|
||||
let owner_a = test_pubkey();
|
||||
let owner_b = test_pubkey();
|
||||
let new_owner = test_pubkey();
|
||||
|
||||
bootstrap_owner(&pool, community_a, &owner_a)
|
||||
.await
|
||||
.expect("bootstrap owner A");
|
||||
bootstrap_owner(&pool, community_b, &owner_b)
|
||||
.await
|
||||
.expect("bootstrap owner B");
|
||||
|
||||
transfer_ownership(&pool, community_a, &new_owner, &owner_a)
|
||||
.await
|
||||
.expect("transfer A");
|
||||
|
||||
assert_role(&pool, community_a, &new_owner, "owner").await;
|
||||
assert_role(&pool, community_a, &owner_a, "member").await;
|
||||
assert_role(&pool, community_b, &owner_b, "owner").await;
|
||||
assert!(
|
||||
!is_relay_member(&pool, community_b, &new_owner)
|
||||
.await
|
||||
.expect("is_relay_member B"),
|
||||
"new owner of A must NOT be a member of B"
|
||||
);
|
||||
}
|
||||
|
||||
/// Transfer ownership to someone who is already a member promotes them to
|
||||
/// owner and demotes the old owner to member.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_promotes_existing_member() {
|
||||
let pool = setup_pool().await;
|
||||
let community = make_test_community(&pool).await;
|
||||
let old_owner = test_pubkey();
|
||||
let existing_member = test_pubkey();
|
||||
|
||||
bootstrap_owner(&pool, community, &old_owner)
|
||||
.await
|
||||
.expect("bootstrap owner");
|
||||
add_relay_member(&pool, community, &existing_member, "member", None)
|
||||
.await
|
||||
.expect("add member");
|
||||
|
||||
let result = transfer_ownership(&pool, community, &existing_member, &old_owner)
|
||||
.await
|
||||
.expect("transfer to existing member");
|
||||
|
||||
assert!(matches!(result, TransferResult::Transferred { .. }));
|
||||
|
||||
assert_eq!(
|
||||
get_relay_member(&pool, community, &existing_member)
|
||||
.await
|
||||
.expect("get new owner")
|
||||
.expect("exists")
|
||||
.role,
|
||||
"owner"
|
||||
);
|
||||
assert_eq!(
|
||||
get_relay_member(&pool, community, &old_owner)
|
||||
.await
|
||||
.expect("get old owner")
|
||||
.expect("exists")
|
||||
.role,
|
||||
"member"
|
||||
);
|
||||
}
|
||||
|
||||
/// Transfer returns `OwnerConflict` when `expected_owner_pubkey` doesn't
|
||||
/// match the current owner — simulates a stale/delayed request after a
|
||||
/// concurrent transfer has already changed ownership.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_returns_owner_conflict_when_expected_mismatches() {
|
||||
let pool = setup_pool().await;
|
||||
let community = make_test_community(&pool).await;
|
||||
let old_owner = test_pubkey();
|
||||
let new_owner = test_pubkey();
|
||||
let wrong_expected = test_pubkey();
|
||||
|
||||
bootstrap_owner(&pool, community, &old_owner)
|
||||
.await
|
||||
.expect("bootstrap initial owner");
|
||||
|
||||
// expected_owner_pubkey doesn't match the actual owner — should conflict.
|
||||
let result = transfer_ownership(&pool, community, &new_owner, &wrong_expected)
|
||||
.await
|
||||
.expect("transfer ownership with wrong expected");
|
||||
|
||||
assert_eq!(result, TransferResult::OwnerConflict);
|
||||
|
||||
// Old owner is still owner — nothing changed.
|
||||
assert_eq!(
|
||||
get_relay_member(&pool, community, &old_owner)
|
||||
.await
|
||||
.expect("get old owner")
|
||||
.expect("exists")
|
||||
.role,
|
||||
"owner"
|
||||
);
|
||||
// New owner was not added.
|
||||
assert!(
|
||||
get_relay_member(&pool, community, &new_owner)
|
||||
.await
|
||||
.expect("get new owner")
|
||||
.is_none(),
|
||||
"new owner must not be added on conflict"
|
||||
);
|
||||
}
|
||||
|
||||
/// Transfer returns `LimitReached` when the transferee already owns the
|
||||
/// maximum number of communities. The limit is enforced inside the
|
||||
/// transfer transaction at the relay layer.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_ownership_returns_limit_reached_for_maxed_transferee() {
|
||||
let pool = setup_pool().await;
|
||||
let owner = test_pubkey();
|
||||
let transferee = test_pubkey();
|
||||
|
||||
// Give the transferee 3 communities (the max).
|
||||
for _ in 0..3 {
|
||||
let c = make_test_community(&pool).await;
|
||||
bootstrap_owner(&pool, c, &transferee)
|
||||
.await
|
||||
.expect("bootstrap transferee community");
|
||||
}
|
||||
|
||||
// Create a community owned by `owner` and try to transfer to `transferee`.
|
||||
let community = make_test_community(&pool).await;
|
||||
bootstrap_owner(&pool, community, &owner)
|
||||
.await
|
||||
.expect("bootstrap owner");
|
||||
|
||||
let result = transfer_ownership(&pool, community, &transferee, &owner)
|
||||
.await
|
||||
.expect("transfer to maxed transferee");
|
||||
|
||||
assert_eq!(result, TransferResult::LimitReached);
|
||||
|
||||
// Owner is still owner — transfer did not happen.
|
||||
assert_eq!(
|
||||
get_relay_member(&pool, community, &owner)
|
||||
.await
|
||||
.expect("get owner")
|
||||
.expect("exists")
|
||||
.role,
|
||||
"owner"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,8 +11,11 @@ use axum::{
|
||||
http::{HeaderMap, StatusCode},
|
||||
response::Json,
|
||||
};
|
||||
use serde::Deserialize;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
use uuid::Uuid;
|
||||
|
||||
use buzz_core::{CommunityId, TenantContext};
|
||||
|
||||
use crate::handlers::community_provisioning::{
|
||||
normalize_candidate_host, validate_pubkey_hex, ProvisionCommunityRequest,
|
||||
@@ -33,6 +36,22 @@ pub struct CommunityAvailabilityQuery {
|
||||
host: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct TransferCommunityRequest {
|
||||
community_id: String,
|
||||
new_owner_pubkey: String,
|
||||
expected_owner_pubkey: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
struct TransferCommunityResponse {
|
||||
community_id: String,
|
||||
new_owner_pubkey: String,
|
||||
status: &'static str,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
previous_owner: Option<String>,
|
||||
}
|
||||
|
||||
const OPERATOR_REPLAY_SCOPE: &str = "operator-management";
|
||||
|
||||
/// Shared deployment-global operator auth prelude. The canonical management
|
||||
@@ -159,7 +178,9 @@ pub async fn provision_community(
|
||||
Err(msg) if msg.starts_with("actor not authorized") => {
|
||||
Err(api_error(StatusCode::FORBIDDEN, &msg))
|
||||
}
|
||||
Err(msg) if msg == "community already exists" => Err(api_error(StatusCode::CONFLICT, &msg)),
|
||||
Err(msg) if msg == "community already exists" || msg.starts_with("limit_reached:") => {
|
||||
Err(api_error(StatusCode::CONFLICT, &msg))
|
||||
}
|
||||
Err(msg)
|
||||
if msg.starts_with("failed to create community:")
|
||||
|| msg.starts_with("community provisioned but owner bootstrap failed:") =>
|
||||
@@ -211,6 +232,130 @@ pub async fn list_owned_communities(
|
||||
})))
|
||||
}
|
||||
|
||||
/// Transfer ownership of a community to a new owner pubkey.
|
||||
///
|
||||
/// `POST /operator/communities/transfer`, NIP-98 signed by a pubkey in
|
||||
/// `RELAY_OPERATOR_PUBKEYS`, body:
|
||||
///
|
||||
/// ```json
|
||||
/// { "community_id": "<uuid>", "new_owner_pubkey": "<hex>" }
|
||||
/// ```
|
||||
///
|
||||
/// The previous owner is demoted to `member` (not `admin`). The transfer is
|
||||
/// instant and atomic at the database layer. Publication of the updated
|
||||
/// NIP-43 membership list is best-effort, matching the provision path.
|
||||
pub async fn transfer_community(
|
||||
State(state): State<Arc<AppState>>,
|
||||
headers: HeaderMap,
|
||||
body: axum::body::Bytes,
|
||||
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
|
||||
let _pubkey = authorize_operator_request(
|
||||
&state,
|
||||
&headers,
|
||||
"POST",
|
||||
"/operator/communities/transfer",
|
||||
None,
|
||||
Some(&body),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let request: TransferCommunityRequest = serde_json::from_slice(&body).map_err(|e| {
|
||||
api_error(
|
||||
StatusCode::BAD_REQUEST,
|
||||
&format!("invalid transfer-community JSON: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let community_uuid = Uuid::parse_str(&request.community_id).map_err(|e| {
|
||||
api_error(
|
||||
StatusCode::BAD_REQUEST,
|
||||
&format!("invalid community_id: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let new_owner_pubkey = validate_pubkey_hex(&request.new_owner_pubkey).ok_or_else(|| {
|
||||
api_error(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"invalid new_owner_pubkey: expected 64-char hex pubkey",
|
||||
)
|
||||
})?;
|
||||
|
||||
let expected_owner_pubkey =
|
||||
validate_pubkey_hex(&request.expected_owner_pubkey).ok_or_else(|| {
|
||||
api_error(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"invalid expected_owner_pubkey: expected 64-char hex pubkey",
|
||||
)
|
||||
})?;
|
||||
|
||||
let community = CommunityId::from_uuid(community_uuid);
|
||||
|
||||
let result = state
|
||||
.db
|
||||
.transfer_ownership(community, &new_owner_pubkey, &expected_owner_pubkey)
|
||||
.await
|
||||
.map_err(|e| internal_error(&format!("transfer ownership: {e}")))?;
|
||||
|
||||
let (status, previous_owner) = match result {
|
||||
buzz_db::relay_members::TransferResult::Transferred { previous_owner } => {
|
||||
("transferred", previous_owner)
|
||||
}
|
||||
buzz_db::relay_members::TransferResult::AlreadyOwner => ("already_owner", None),
|
||||
buzz_db::relay_members::TransferResult::NoOwner => {
|
||||
return Err(api_error(
|
||||
StatusCode::NOT_FOUND,
|
||||
"community has no owner to transfer from",
|
||||
));
|
||||
}
|
||||
buzz_db::relay_members::TransferResult::OwnerConflict => {
|
||||
return Err(api_error(
|
||||
StatusCode::CONFLICT,
|
||||
"owner_conflict: the current owner no longer matches expected_owner_pubkey",
|
||||
));
|
||||
}
|
||||
buzz_db::relay_members::TransferResult::LimitReached => {
|
||||
return Err(api_error(
|
||||
StatusCode::CONFLICT,
|
||||
"limit_reached: transferee already owns the maximum number of communities",
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
// Best-effort NIP-43 membership snapshot publication — same pattern as
|
||||
// provision_community. The DB mutation is already committed; a publication
|
||||
// failure must not turn a success into an HTTP error.
|
||||
if state.config.require_relay_membership {
|
||||
if let Some(host) = state
|
||||
.db
|
||||
.lookup_community_host(community)
|
||||
.await
|
||||
.map_err(|e| internal_error(&format!("lookup community host: {e}")))?
|
||||
{
|
||||
let tenant = TenantContext::resolved(community, host);
|
||||
if let Err(error) =
|
||||
crate::handlers::side_effects::publish_nip43_membership_list(&tenant, &state).await
|
||||
{
|
||||
tracing::warn!(
|
||||
community = %community,
|
||||
error = %error,
|
||||
"ownership transferred but NIP-43 membership snapshot publication failed"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let response = TransferCommunityResponse {
|
||||
community_id: request.community_id,
|
||||
new_owner_pubkey,
|
||||
status,
|
||||
previous_owner,
|
||||
};
|
||||
|
||||
Ok(Json(serde_json::to_value(response).map_err(|e| {
|
||||
tracing::error!("failed to serialize transfer-community response: {e}");
|
||||
internal_error("operator transfer response serialization failed")
|
||||
})?))
|
||||
}
|
||||
/// Check whether a community host is available, returning the relay-canonical
|
||||
/// normalized authority used by create.
|
||||
pub async fn community_availability(
|
||||
@@ -260,7 +405,7 @@ mod tests {
|
||||
use tower::ServiceExt;
|
||||
use uuid::Uuid;
|
||||
|
||||
use buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST;
|
||||
use buzz_core::{kind::KIND_NIP43_MEMBERSHIP_LIST, CommunityId};
|
||||
use buzz_db::event::EventQuery;
|
||||
|
||||
use crate::router::build_router;
|
||||
@@ -371,6 +516,85 @@ mod tests {
|
||||
serde_json::from_slice(&bytes).expect("response JSON")
|
||||
}
|
||||
|
||||
async fn signed_operator_request(
|
||||
state: Arc<AppState>,
|
||||
keys: &Keys,
|
||||
method: &str,
|
||||
path: &str,
|
||||
body: Option<String>,
|
||||
) -> axum::response::Response {
|
||||
let url = format!("http://{INGRESS_HOST}{path}");
|
||||
let auth = nip98_auth_header(keys, &url, method, body.as_deref().map(str::as_bytes));
|
||||
let mut request = Request::builder()
|
||||
.method(method)
|
||||
.uri(path)
|
||||
.header(header::HOST, INGRESS_HOST)
|
||||
.header(header::AUTHORIZATION, auth);
|
||||
if body.is_some() {
|
||||
request = request.header(header::CONTENT_TYPE, "application/json");
|
||||
}
|
||||
build_router(state)
|
||||
.oneshot(
|
||||
request
|
||||
.body(body.map_or_else(Body::empty, Body::from))
|
||||
.expect("request"),
|
||||
)
|
||||
.await
|
||||
.expect("response")
|
||||
}
|
||||
|
||||
async fn provision_community(
|
||||
state: Arc<AppState>,
|
||||
operator: &Keys,
|
||||
host: &str,
|
||||
owner: &Keys,
|
||||
) -> axum::response::Response {
|
||||
let body = serde_json::json!({
|
||||
"host": host,
|
||||
"initial_owner_pubkey": owner.public_key().to_hex(),
|
||||
"create_only": true,
|
||||
})
|
||||
.to_string();
|
||||
signed_operator_request(state, operator, "POST", "/operator/communities", Some(body)).await
|
||||
}
|
||||
|
||||
fn is_member_tag(tag: &Tag, pubkey: &str, role: &str) -> bool {
|
||||
let values = tag.as_slice();
|
||||
values.first().is_some_and(|value| value == "member")
|
||||
&& values.get(1).is_some_and(|value| value == pubkey)
|
||||
&& values.get(2).is_some_and(|value| value == role)
|
||||
}
|
||||
|
||||
async fn assert_snapshot_roles(
|
||||
state: &AppState,
|
||||
community: CommunityId,
|
||||
expected: &[(&str, &str)],
|
||||
) {
|
||||
let snapshot = state
|
||||
.db
|
||||
.query_events(&EventQuery {
|
||||
kinds: Some(vec![KIND_NIP43_MEMBERSHIP_LIST as i32]),
|
||||
global_only: true,
|
||||
limit: Some(1),
|
||||
..EventQuery::for_community(community)
|
||||
})
|
||||
.await
|
||||
.expect("query membership snapshot")
|
||||
.into_iter()
|
||||
.next()
|
||||
.expect("membership snapshot published");
|
||||
for &(pubkey, role) in expected {
|
||||
assert!(
|
||||
snapshot
|
||||
.event
|
||||
.tags
|
||||
.iter()
|
||||
.any(|tag| is_member_tag(tag, pubkey, role)),
|
||||
"missing {role} snapshot tag for {pubkey}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn non_allowlisted_operator_key_gets_403() {
|
||||
@@ -513,28 +737,7 @@ mod tests {
|
||||
return;
|
||||
};
|
||||
let host = format!("community-{}.example", Uuid::new_v4().simple());
|
||||
let body = serde_json::json!({
|
||||
"host": host,
|
||||
"initial_owner_pubkey": owner.public_key().to_hex(),
|
||||
"create_only": true,
|
||||
})
|
||||
.to_string();
|
||||
let url = format!("http://{INGRESS_HOST}/operator/communities");
|
||||
let auth = nip98_auth_header(&operator, &url, "POST", Some(body.as_bytes()));
|
||||
|
||||
let response = build_router(state.clone())
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("POST")
|
||||
.uri("/operator/communities")
|
||||
.header(header::HOST, INGRESS_HOST)
|
||||
.header(header::AUTHORIZATION, auth)
|
||||
.header(header::CONTENT_TYPE, "application/json")
|
||||
.body(Body::from(body))
|
||||
.expect("request"),
|
||||
)
|
||||
.await
|
||||
.expect("response");
|
||||
let response = provision_community(state.clone(), &operator, &host, &owner).await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let json = read_json(response).await;
|
||||
@@ -558,28 +761,182 @@ mod tests {
|
||||
.expect("owner member exists");
|
||||
assert_eq!(member.role, "owner");
|
||||
|
||||
let membership_events = state
|
||||
assert_snapshot_roles(&state, community.id, &[(&owner_hex, "owner")]).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn fresh_host_at_owner_limit_returns_limit_reached_conflict() {
|
||||
let operator = Keys::generate();
|
||||
let owner = Keys::generate();
|
||||
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
|
||||
return;
|
||||
};
|
||||
|
||||
for _ in 0..buzz_db::relay_members::MAX_COMMUNITIES_PER_OWNER {
|
||||
let host = format!("community-{}.example", Uuid::new_v4().simple());
|
||||
assert_eq!(
|
||||
provision_community(state.clone(), &operator, &host, &owner)
|
||||
.await
|
||||
.status(),
|
||||
StatusCode::OK
|
||||
);
|
||||
}
|
||||
|
||||
let host = format!("community-{}.example", Uuid::new_v4().simple());
|
||||
let response = provision_community(state.clone(), &operator, &host, &owner).await;
|
||||
assert_eq!(response.status(), StatusCode::CONFLICT);
|
||||
let json = read_json(response).await;
|
||||
assert!(json["error"]
|
||||
.as_str()
|
||||
.is_some_and(|error| error.starts_with("limit_reached:")));
|
||||
assert!(state
|
||||
.db
|
||||
.query_events(&EventQuery {
|
||||
kinds: Some(vec![KIND_NIP43_MEMBERSHIP_LIST as i32]),
|
||||
global_only: true,
|
||||
limit: Some(1),
|
||||
..EventQuery::for_community(community.id)
|
||||
})
|
||||
.lookup_community_by_host(&host)
|
||||
.await
|
||||
.expect("query membership snapshot");
|
||||
let snapshot = membership_events
|
||||
.first()
|
||||
.expect("membership snapshot published during provisioning");
|
||||
assert!(snapshot.event.tags.iter().any(|tag| {
|
||||
tag.as_slice()
|
||||
.first()
|
||||
.is_some_and(|value| value == "member")
|
||||
&& tag
|
||||
.as_slice()
|
||||
.get(1)
|
||||
.is_some_and(|value| value == &owner_hex)
|
||||
&& tag.as_slice().get(2).is_some_and(|value| value == "owner")
|
||||
}));
|
||||
.expect("look up rejected fresh host")
|
||||
.is_none());
|
||||
}
|
||||
|
||||
/// Happy path: POST /operator/communities/transfer swaps ownership, demotes
|
||||
/// the old owner to `member`, and publishes a NIP-43 snapshot reflecting the
|
||||
/// new roles.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn happy_path_transfer_swaps_owner_and_demotes_old_to_member() {
|
||||
let operator = Keys::generate();
|
||||
let initial_owner = Keys::generate();
|
||||
let new_owner = Keys::generate();
|
||||
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
|
||||
return;
|
||||
};
|
||||
|
||||
let host = format!("community-{}.example", Uuid::new_v4().simple());
|
||||
let create_response =
|
||||
provision_community(state.clone(), &operator, &host, &initial_owner).await;
|
||||
assert_eq!(create_response.status(), StatusCode::OK);
|
||||
|
||||
let community = state
|
||||
.db
|
||||
.lookup_community_by_host(&host)
|
||||
.await
|
||||
.expect("lookup community")
|
||||
.expect("community exists");
|
||||
let community_id = community.id.to_string();
|
||||
let initial_owner_hex = initial_owner.public_key().to_hex();
|
||||
let new_owner_hex = new_owner.public_key().to_hex();
|
||||
|
||||
let transfer_body = serde_json::json!({
|
||||
"community_id": community_id,
|
||||
"new_owner_pubkey": new_owner_hex,
|
||||
"expected_owner_pubkey": initial_owner_hex,
|
||||
})
|
||||
.to_string();
|
||||
let response = signed_operator_request(
|
||||
state.clone(),
|
||||
&operator,
|
||||
"POST",
|
||||
"/operator/communities/transfer",
|
||||
Some(transfer_body),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let json = read_json(response).await;
|
||||
assert_eq!(
|
||||
json.get("status").and_then(Value::as_str),
|
||||
Some("transferred")
|
||||
);
|
||||
assert_eq!(
|
||||
json.get("new_owner_pubkey").and_then(Value::as_str),
|
||||
Some(new_owner_hex.as_str())
|
||||
);
|
||||
assert_eq!(
|
||||
json.get("previous_owner").and_then(Value::as_str),
|
||||
Some(initial_owner_hex.as_str())
|
||||
);
|
||||
|
||||
// New owner is owner.
|
||||
assert_eq!(
|
||||
state
|
||||
.db
|
||||
.get_relay_member(community.id, &new_owner_hex)
|
||||
.await
|
||||
.expect("get new owner")
|
||||
.expect("new owner exists")
|
||||
.role,
|
||||
"owner"
|
||||
);
|
||||
// Old owner is member (not admin).
|
||||
assert_eq!(
|
||||
state
|
||||
.db
|
||||
.get_relay_member(community.id, &initial_owner_hex)
|
||||
.await
|
||||
.expect("get old owner")
|
||||
.expect("old owner exists")
|
||||
.role,
|
||||
"member"
|
||||
);
|
||||
|
||||
assert_snapshot_roles(
|
||||
&state,
|
||||
community.id,
|
||||
&[(&new_owner_hex, "owner"), (&initial_owner_hex, "member")],
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Transfer with an invalid community_id returns 400.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_with_invalid_community_id_returns_400() {
|
||||
let operator = Keys::generate();
|
||||
let new_owner = Keys::generate();
|
||||
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
|
||||
return;
|
||||
};
|
||||
let body = serde_json::json!({
|
||||
"community_id": "not-a-uuid",
|
||||
"new_owner_pubkey": new_owner.public_key().to_hex(),
|
||||
"expected_owner_pubkey": new_owner.public_key().to_hex(),
|
||||
})
|
||||
.to_string();
|
||||
let response = signed_operator_request(
|
||||
state,
|
||||
&operator,
|
||||
"POST",
|
||||
"/operator/communities/transfer",
|
||||
Some(body),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
|
||||
}
|
||||
|
||||
/// Transfer with an invalid new_owner_pubkey returns 400.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn transfer_with_invalid_pubkey_returns_400() {
|
||||
let operator = Keys::generate();
|
||||
let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else {
|
||||
return;
|
||||
};
|
||||
let body = serde_json::json!({
|
||||
"community_id": Uuid::new_v4().to_string(),
|
||||
"new_owner_pubkey": "not-a-pubkey",
|
||||
"expected_owner_pubkey": "not-a-pubkey",
|
||||
})
|
||||
.to_string();
|
||||
let response = signed_operator_request(
|
||||
state,
|
||||
&operator,
|
||||
"POST",
|
||||
"/operator/communities/transfer",
|
||||
Some(body),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -281,12 +281,23 @@ pub async fn provision_community(
|
||||
let owner_hex = initial_owner.as_deref().ok_or_else(|| {
|
||||
"initial_owner_pubkey is required when create_only is true".to_string()
|
||||
})?;
|
||||
let record = state
|
||||
let record = match state
|
||||
.db
|
||||
.create_community_with_owner(&request.host, owner_hex)
|
||||
.await
|
||||
.map_err(|e| format!("failed to create community: {e}"))?
|
||||
.ok_or_else(|| "community already exists".to_string())?;
|
||||
{
|
||||
buzz_db::CreateCommunityWithOwnerResult::Created(record) => record,
|
||||
buzz_db::CreateCommunityWithOwnerResult::HostExists => {
|
||||
return Err("community already exists".to_string());
|
||||
}
|
||||
buzz_db::CreateCommunityWithOwnerResult::LimitReached => {
|
||||
return Err(
|
||||
"limit_reached: owner already owns the maximum number of communities"
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
info!(
|
||||
operator = %operator_hex,
|
||||
|
||||
@@ -2743,39 +2743,29 @@ async fn emit_initial_ref_state(
|
||||
///
|
||||
/// Queries all current relay members and emits a relay-signed, NIP-70-protected
|
||||
/// addressable event listing every member pubkey. Replaces any previous list.
|
||||
///
|
||||
/// The member query, event construction, and replacement all happen inside
|
||||
/// the same per-community advisory lock held by
|
||||
/// `publish_nip43_membership_locked`. This prevents a stale snapshot from
|
||||
/// overwriting a newer one when two publications race: the lock serializes
|
||||
/// the entire read-build-write cycle, not just the write.
|
||||
pub async fn publish_nip43_membership_list(
|
||||
tenant: &TenantContext,
|
||||
state: &Arc<AppState>,
|
||||
) -> anyhow::Result<()> {
|
||||
let members = state.db.list_relay_members(tenant.community()).await?;
|
||||
let relay_pubkey_hex = state.relay_keypair.public_key().to_hex();
|
||||
let community = tenant.community();
|
||||
|
||||
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)
|
||||
.sign_with_keys(&state.relay_keypair)
|
||||
.map_err(|e| anyhow::anyhow!("failed to sign kind:13534: {e}"))?;
|
||||
|
||||
// NOTE: kind 13534 is technically a regular event (not in the NIP-16 replaceable
|
||||
// range), but we intentionally use replace_addressable_event to get replacement
|
||||
// semantics — only the latest membership snapshot matters. This function keys on
|
||||
// (kind, pubkey, channel_id) and atomically replaces older events, which is exactly
|
||||
// what Pyramid (the reference NIP-43 implementation) does with store.ReplaceEvent().
|
||||
let (stored, was_inserted) = state
|
||||
// The DB method acquires the per-community snapshot lock BEFORE reading
|
||||
// members, so that the entire read-build-write cycle is serialized. A
|
||||
// concurrent publication that reads newer state will block until our
|
||||
// replacement commits, then read the updated membership. This prevents a
|
||||
// stale snapshot from winning by arrival order.
|
||||
let (stored, was_inserted, member_count) = state
|
||||
.db
|
||||
.replace_addressable_event(tenant.community(), &event, None)
|
||||
.publish_nip43_membership_locked(community, &state.relay_keypair)
|
||||
.await?;
|
||||
|
||||
if was_inserted {
|
||||
dispatch_persistent_event(
|
||||
tenant,
|
||||
@@ -2788,10 +2778,7 @@ pub async fn publish_nip43_membership_list(
|
||||
.await;
|
||||
}
|
||||
|
||||
info!(
|
||||
member_count = members.len(),
|
||||
"NIP-43 membership list published"
|
||||
);
|
||||
info!(member_count, "NIP-43 membership list published");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -68,6 +68,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
|
||||
"/operator/communities/availability",
|
||||
get(api::operator::community_availability),
|
||||
)
|
||||
.route(
|
||||
"/operator/communities/transfer",
|
||||
post(api::operator::transfer_community),
|
||||
)
|
||||
// Relay invites: mint (owner/admin) + claim (membership-gate exempt)
|
||||
.route("/api/invites", post(api::invites::mint_invite))
|
||||
.route("/api/invites/claim", post(api::invites::claim_invite))
|
||||
|
||||
Reference in New Issue
Block a user