mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
fix(relay): store a kind:9000 and its membership writes in one transaction
If the membership write failed after a kind:9000 was stored, the relay logged the failure and still accepted the event. A resend then returned `duplicate:`, so the person was never added. kind:9000 now uses the same atomic path as kind:9002: `put_members_with_event` takes the membership lock, stores the event and writes every allowed person in one transaction. A failure stores nothing and returns `error:`, so the client can resend the same event. When the decisions under the lock refuse everyone, nothing is stored and the event is rejected with the text validation gives. Before, a single-person add refused there was accepted with an empty message. NOSTR.md now says a failed kind:9000 can be resent as is. Drops the `Db::put_members` wrapper, which no longer has callers. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Luke Tornquist <tornquist@squareup.com>
This commit is contained in:
2
NOSTR.md
2
NOSTR.md
@@ -51,7 +51,7 @@ PGPASSWORD=buzz_dev psql -h localhost -U buzz -d buzz -c \
|
||||
| **Deletions (kind:5)** | ✅ | Standard NIP-09; self-authored only. `#h` optional, `#e` required |
|
||||
| **User profiles (kind:0)** | ✅ | NIP-01 metadata; synced to users table (display_name, avatar, about, NIP-05). NIP-05 handles must canonicalize to this relay's domain — off-domain or invalid handles are silently cleared. If a NIP-05 handle collides with another user's (UNIQUE constraint), the handle is skipped but other profile fields (display_name, avatar, about) are still synced. |
|
||||
| **Group creation (kind:9007)** | ✅ | NIP-29; include `name` tag, optional `visibility`, `channel_type`, label `t` and `unlisted` tags |
|
||||
| **Add user (kind:9000)** | ✅ | Open: any user, subject to target's `channel_add_policy` (`owner_only`/`nobody` can block). Private: owner/admin only. Self-add bypasses agent policy but not private-channel auth. A guest cannot add anyone. A new member with no role tag gets the channel's `join_role`; a non-member adding themselves may ask for `join_role` or a lower role. One event may name up to 100 people: one `p` tag each, with an optional role as the `p` tag's third element (NIP-29), falling back to the event's `role` tag. Each person is checked and added on their own. When the event names several people, the OK message is `members: ` followed by JSON listing the non-empty `added`, `updated` (role changed), `already_member` and `rejected` (`[{"reason", "pubkeys"}]`) results. The event is refused only when it is malformed or nobody it names may be added; then one person keeps the single-add error, and several get `invalid: members: {…}`. A resent event returns `duplicate:` without results; to retry an `error:` rejection, send a new event. Re-adding someone at their current role posts no second `member_joined` notice. |
|
||||
| **Add user (kind:9000)** | ✅ | Open: any user, subject to target's `channel_add_policy` (`owner_only`/`nobody` can block). Private: owner/admin only. Self-add bypasses agent policy but not private-channel auth. A guest cannot add anyone. A new member with no role tag gets the channel's `join_role`; a non-member adding themselves may ask for `join_role` or a lower role. One event may name up to 100 people: one `p` tag each, with an optional role as the `p` tag's third element (NIP-29), falling back to the event's `role` tag. Each person is checked and added on their own. When the event names several people, the OK message is `members: ` followed by JSON listing the non-empty `added`, `updated` (role changed), `already_member` and `rejected` (`[{"reason", "pubkeys"}]`) results. The event is refused only when it is malformed or nobody it names may be added; then one person keeps the single-add error, and several get `invalid: members: {…}`. The event and its membership changes are stored together, so an `error:` rejection stores nothing and the same event can be resent; a resent stored event returns `duplicate:` without results. Re-adding someone at their current role posts no second `member_joined` notice. |
|
||||
| **Remove user (kind:9001)** | ✅ | Self-remove allowed (with last-owner guard). Removing others: owner/admin only. |
|
||||
| **Edit group metadata (kind:9002)** | ✅ | `name`/`about`/label `t`/`visibility`/`ttl`/`archived`/`join_role`/`unlisted` tags: owner/admin. `topic`/`purpose` tags: any member except a guest. Buzz applies 9002 as a partial update: an absent tag leaves that setting unchanged. |
|
||||
| **Admin delete event (kind:9005)** | ✅ | Event author can always delete own. Otherwise owner/admin required. Target must be in same channel. |
|
||||
|
||||
@@ -791,14 +791,7 @@ pub async fn put_members(
|
||||
targets: &[MemberPut<'_>],
|
||||
invited_by: Option<&[u8]>,
|
||||
) -> Result<Vec<Result<PutMemberResult>>> {
|
||||
let mut seen = std::collections::HashSet::with_capacity(targets.len());
|
||||
if let Some(dup) = targets.iter().find(|t| !seen.insert(t.pubkey)) {
|
||||
return Err(DbError::InvalidData(format!(
|
||||
"pubkey {} appears more than once",
|
||||
hex::encode(dup.pubkey)
|
||||
)));
|
||||
}
|
||||
|
||||
check_distinct_targets(targets)?;
|
||||
let mut tx = crate::begin_community_write_transaction(
|
||||
pool,
|
||||
community_id,
|
||||
@@ -809,7 +802,100 @@ pub async fn put_members(
|
||||
// First statement: serialize the whole decision / owner-count / upsert
|
||||
// sequence against concurrent membership writes on this channel.
|
||||
acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?;
|
||||
let results = put_members_on(&mut tx, channel_id, targets, invited_by).await?;
|
||||
tx.commit().await?;
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
/// The result of [`Db::put_members_with_event`].
|
||||
pub enum MemberPutEventWrite {
|
||||
/// The event was stored and at least one person was written. One result
|
||||
/// per target, in order, as from [`put_members`].
|
||||
Applied {
|
||||
/// The stored kind:9000 event.
|
||||
stored: buzz_core::StoredEvent,
|
||||
/// One result per target.
|
||||
results: Vec<Result<PutMemberResult>>,
|
||||
},
|
||||
/// The event was already stored. Nothing changed: its membership changes
|
||||
/// committed with it the first time.
|
||||
Duplicate(buzz_core::StoredEvent),
|
||||
/// Every person was refused, so nothing was stored. One error per target.
|
||||
Refused(Vec<DbError>),
|
||||
}
|
||||
|
||||
/// Store a kind:9000 `event` and apply its [`put_members`] writes in one
|
||||
/// transaction. If any write fails, nothing is stored, so the client can
|
||||
/// resend the same event and have it applied.
|
||||
///
|
||||
/// The membership lock comes first, as in [`put_members`] and the channel
|
||||
/// settings writers, so a second copy of the event waits for the first and
|
||||
/// then finds it stored.
|
||||
pub(crate) async fn put_members_with_event(
|
||||
pool: &PgPool,
|
||||
push_enqueue: Option<crate::runtime::push_enqueue::PushEnqueue>,
|
||||
community_id: CommunityId,
|
||||
channel_id: Uuid,
|
||||
event: &nostr::Event,
|
||||
targets: &[MemberPut<'_>],
|
||||
thread_meta: Option<crate::event::ThreadMetadataParams<'_>>,
|
||||
) -> Result<MemberPutEventWrite> {
|
||||
check_distinct_targets(targets)?;
|
||||
let mut tx = crate::begin_community_event_write_transaction(
|
||||
pool,
|
||||
community_id,
|
||||
crate::observability::WriterOperation::EventWrite,
|
||||
)
|
||||
.await?;
|
||||
tx.set_push_enqueue(push_enqueue);
|
||||
acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?;
|
||||
|
||||
// The same writes as `Db::insert_event_with_thread_metadata`.
|
||||
let (stored, inserted) = crate::event::insert_event_with_thread_metadata_tx(
|
||||
&mut tx,
|
||||
event,
|
||||
Some(channel_id),
|
||||
thread_meta,
|
||||
)
|
||||
.await?;
|
||||
if !inserted {
|
||||
tx.rollback().await?;
|
||||
return Ok(MemberPutEventWrite::Duplicate(stored));
|
||||
}
|
||||
crate::insert_mentions_in_transaction(&mut tx, event, Some(channel_id)).await?;
|
||||
|
||||
let actor = event.pubkey.to_bytes();
|
||||
let results = put_members_on(&mut tx, channel_id, targets, Some(&actor)).await?;
|
||||
if results.iter().all(Result::is_err) {
|
||||
tx.rollback().await?;
|
||||
return Ok(MemberPutEventWrite::Refused(
|
||||
results.into_iter().filter_map(Result::err).collect(),
|
||||
));
|
||||
}
|
||||
tx.commit().await?;
|
||||
Ok(MemberPutEventWrite::Applied { stored, results })
|
||||
}
|
||||
|
||||
fn check_distinct_targets(targets: &[MemberPut<'_>]) -> Result<()> {
|
||||
let mut seen = std::collections::HashSet::with_capacity(targets.len());
|
||||
match targets.iter().find(|t| !seen.insert(t.pubkey)) {
|
||||
Some(dup) => Err(DbError::InvalidData(format!(
|
||||
"pubkey {} appears more than once",
|
||||
hex::encode(dup.pubkey)
|
||||
))),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// The decisions and upsert of [`put_members`], on a transaction that holds
|
||||
/// `channel_id`'s membership lock. The caller owns the commit.
|
||||
async fn put_members_on(
|
||||
tx: &mut AdmittedTx,
|
||||
channel_id: Uuid,
|
||||
targets: &[MemberPut<'_>],
|
||||
invited_by: Option<&[u8]>,
|
||||
) -> Result<Vec<Result<PutMemberResult>>> {
|
||||
let community_id = tx.community();
|
||||
let channel = get_channel_tx(tx.conn(), community_id, channel_id).await?;
|
||||
let private = channel.visibility == "private";
|
||||
let join_role = buzz_core::channel::ChannelJoinRole::parse_fail_closed(&channel.join_role);
|
||||
@@ -945,7 +1031,6 @@ pub async fn put_members(
|
||||
written.insert(member.pubkey.clone(), member);
|
||||
}
|
||||
}
|
||||
tx.commit().await?;
|
||||
|
||||
Ok(targets
|
||||
.iter()
|
||||
@@ -2115,17 +2200,28 @@ impl Db {
|
||||
.await
|
||||
}
|
||||
|
||||
/// [`Db::put_member`] for several people in one transaction (see
|
||||
/// [`put_members`]).
|
||||
#[datastore_span(name = "put_members", system = "postgresql")]
|
||||
pub async fn put_members(
|
||||
/// Stores a kind:9000 `event` and applies its [`put_members`] writes,
|
||||
/// with the event's author as the actor, in one transaction (see
|
||||
/// [`put_members_with_event`]).
|
||||
#[datastore_span(name = "put_members_with_event", system = "postgresql")]
|
||||
pub async fn put_members_with_event(
|
||||
&self,
|
||||
community_id: CommunityId,
|
||||
channel_id: Uuid,
|
||||
event: &nostr::Event,
|
||||
targets: &[MemberPut<'_>],
|
||||
invited_by: Option<&[u8]>,
|
||||
) -> Result<Vec<Result<PutMemberResult>>> {
|
||||
put_members(&self.pool, community_id, channel_id, targets, invited_by).await
|
||||
thread_meta: Option<crate::event::ThreadMetadataParams<'_>>,
|
||||
) -> Result<MemberPutEventWrite> {
|
||||
put_members_with_event(
|
||||
&self.pool,
|
||||
self.push_enqueue.clone(),
|
||||
community_id,
|
||||
channel_id,
|
||||
event,
|
||||
targets,
|
||||
thread_meta,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Removes a member from a channel.
|
||||
@@ -3476,6 +3572,112 @@ mod postgres_tests {
|
||||
assert!(matches!(err, DbError::InvalidData(_)), "{err:?}");
|
||||
}
|
||||
|
||||
/// `put_members_with_event` stores the event only with the writes it
|
||||
/// carries: a stored event applies its people, a resend changes nothing,
|
||||
/// and an event whose every person is refused under the lock is not
|
||||
/// stored. Mutations: commit when everyone is refused → the refused event
|
||||
/// is stored → RED; write people on a duplicate → the removed person is
|
||||
/// back → RED.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn put_members_with_event_stores_the_event_with_its_writes() {
|
||||
let pool = setup_pool().await;
|
||||
let community_id = make_test_community(&pool).await;
|
||||
let community = CommunityId::from_uuid(community_id);
|
||||
let (owner, outsider) = (Keys::generate(), Keys::generate());
|
||||
let (owner_pk, outsider_pk) = (
|
||||
owner.public_key().to_bytes().to_vec(),
|
||||
outsider.public_key().to_bytes().to_vec(),
|
||||
);
|
||||
let person = random_pubkey();
|
||||
for pk in [&owner_pk, &outsider_pk, &person] {
|
||||
ensure_user(&pool, community, pk)
|
||||
.await
|
||||
.expect("ensure user");
|
||||
}
|
||||
let channel = create_test_channel(
|
||||
&pool,
|
||||
community_id,
|
||||
"put-members-event",
|
||||
ChannelType::Stream,
|
||||
ChannelVisibility::Private,
|
||||
None,
|
||||
&owner_pk,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("create channel");
|
||||
let event = |actor: &Keys| {
|
||||
nostr::EventBuilder::new(nostr::Kind::Custom(9000), "")
|
||||
.tags([
|
||||
nostr::Tag::parse(["h", &channel.id.to_string()]).unwrap(),
|
||||
nostr::Tag::parse(["p", &hex::encode(&person)]).unwrap(),
|
||||
])
|
||||
.sign_with_keys(actor)
|
||||
.expect("sign 9000")
|
||||
};
|
||||
let targets = [MemberPut {
|
||||
pubkey: &person,
|
||||
requested: None,
|
||||
}];
|
||||
let stored = |id: nostr::EventId| {
|
||||
let pool = pool.clone();
|
||||
async move {
|
||||
crate::event::get_event_by_id(&pool, community, id.as_bytes())
|
||||
.await
|
||||
.expect("read event")
|
||||
.is_some()
|
||||
}
|
||||
};
|
||||
|
||||
// A non-member may not add anyone to a private channel.
|
||||
let refused = event(&outsider);
|
||||
let write =
|
||||
put_members_with_event(&pool, None, community, channel.id, &refused, &targets, None)
|
||||
.await
|
||||
.expect("refused write");
|
||||
assert!(
|
||||
matches!(&write, MemberPutEventWrite::Refused(errors)
|
||||
if matches!(errors.as_slice(), [DbError::AccessDenied(_)])),
|
||||
"expected a refusal"
|
||||
);
|
||||
assert!(!stored(refused.id).await, "a refused event is not stored");
|
||||
assert_eq!(
|
||||
get_member_role(&pool, community, channel.id, &person)
|
||||
.await
|
||||
.expect("read role"),
|
||||
None
|
||||
);
|
||||
|
||||
let added = event(&owner);
|
||||
let write =
|
||||
put_members_with_event(&pool, None, community, channel.id, &added, &targets, None)
|
||||
.await
|
||||
.expect("applied write");
|
||||
let MemberPutEventWrite::Applied { results, .. } = write else {
|
||||
panic!("expected the event to apply");
|
||||
};
|
||||
let put = results[0].as_ref().expect("person added");
|
||||
assert_eq!(put.prior_role, None);
|
||||
assert!(stored(added.id).await);
|
||||
|
||||
// A resend finds the event stored and leaves the roster alone.
|
||||
remove_member(&pool, community, channel.id, &person, &owner_pk)
|
||||
.await
|
||||
.expect("remove person");
|
||||
let write =
|
||||
put_members_with_event(&pool, None, community, channel.id, &added, &targets, None)
|
||||
.await
|
||||
.expect("duplicate write");
|
||||
assert!(matches!(write, MemberPutEventWrite::Duplicate(_)));
|
||||
assert_eq!(
|
||||
get_member_role(&pool, community, channel.id, &person)
|
||||
.await
|
||||
.expect("read role"),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
/// `put_member` enforces admission itself, so a caller that skips the
|
||||
/// relay validator still cannot join an announce channel above
|
||||
/// `join_role`, and a guest still cannot add anyone. No role tag keeps an
|
||||
|
||||
@@ -3053,7 +3053,7 @@ async fn ingest_event_inner(
|
||||
validate_huddle_lifecycle_event(tenant, state, &event, kind_u32).await?;
|
||||
|
||||
// kind:9000 is checked person by person; the plan carries each result
|
||||
// through storage to `apply_put_user`.
|
||||
// to `persist_put_user`, which stores the event and applies it.
|
||||
let put_user = if kind_u32 == KIND_NIP29_PUT_USER {
|
||||
Some(
|
||||
crate::handlers::side_effects::validate_put_user(tenant, &event, state)
|
||||
@@ -3540,6 +3540,7 @@ async fn ingest_event_inner(
|
||||
};
|
||||
|
||||
let workflow_deletion = crate::handlers::side_effects::is_workflow_deletion(&event);
|
||||
let mut message = String::new();
|
||||
let (stored_event, was_inserted) = if let Some(create) = channel_create.as_ref() {
|
||||
let creator = event.pubkey.to_bytes();
|
||||
let channel = buzz_db::channel::NewChannel {
|
||||
@@ -3590,6 +3591,28 @@ async fn ingest_event_inner(
|
||||
crate::handlers::side_effects::persist_workflow_deletion(tenant, &event, state)
|
||||
.await
|
||||
.map_err(|e| IngestError::Internal(format!("error: workflow deletion failed: {e}")))?
|
||||
} else if let Some(plan) = put_user {
|
||||
// One commit stores the event and applies its membership changes. A
|
||||
// failure stores nothing, so the client is told and can resend it.
|
||||
// Refusing everyone under the membership lock is the client's error;
|
||||
// anything else is ours.
|
||||
let thread_params = thread_meta.as_ref().map(|m| m.as_params());
|
||||
let (stored, inserted, ok_message) = crate::handlers::side_effects::persist_put_user(
|
||||
tenant,
|
||||
&event,
|
||||
state,
|
||||
plan,
|
||||
thread_params,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| match e.downcast_ref::<buzz_db::DbError>() {
|
||||
Some(buzz_db::DbError::AccessDenied(reason)) => {
|
||||
IngestError::Rejected(format!("invalid: {reason}"))
|
||||
}
|
||||
_ => IngestError::Internal(format!("error: membership write failed: {e}")),
|
||||
})?;
|
||||
message = ok_message;
|
||||
(stored, inserted)
|
||||
} else if kind_u32 == KIND_NIP29_EDIT_METADATA {
|
||||
// One commit stores the event and applies its channel changes. A
|
||||
// failure stores nothing, so the client is told and can resend it.
|
||||
@@ -3694,14 +3717,12 @@ async fn ingest_event_inner(
|
||||
});
|
||||
}
|
||||
|
||||
// Workflow deletion and kind:9002 applied their changes in the insert's
|
||||
// own transaction above.
|
||||
let persisted_atomically = workflow_deletion || kind_u32 == KIND_NIP29_EDIT_METADATA;
|
||||
let mut message = String::new();
|
||||
if let Some(plan) = put_user {
|
||||
message = crate::handlers::side_effects::apply_put_user(tenant, &event, state, plan).await;
|
||||
} else if !persisted_atomically && crate::handlers::side_effects::is_side_effect_kind(kind_u32)
|
||||
{
|
||||
// Workflow deletion, kind:9000 and kind:9002 applied their changes in the
|
||||
// insert's own transaction above.
|
||||
let persisted_atomically = workflow_deletion
|
||||
|| kind_u32 == KIND_NIP29_PUT_USER
|
||||
|| kind_u32 == KIND_NIP29_EDIT_METADATA;
|
||||
if !persisted_atomically && crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
|
||||
if let Err(e) =
|
||||
crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state)
|
||||
.await
|
||||
|
||||
@@ -546,3 +546,127 @@ async fn put_user_decides_each_person_after_the_ones_before() {
|
||||
Some("owner")
|
||||
);
|
||||
}
|
||||
|
||||
async fn put_user_events(state: &AppState, tenant: &TenantContext, channel_id: Uuid) -> usize {
|
||||
state
|
||||
.db
|
||||
.query_events(&EventQuery {
|
||||
kinds: Some(vec![buzz_core::kind::KIND_NIP29_PUT_USER as i32]),
|
||||
channel_id: Some(channel_id),
|
||||
..EventQuery::for_community(tenant.community())
|
||||
})
|
||||
.await
|
||||
.expect("put-user events")
|
||||
.len()
|
||||
}
|
||||
|
||||
/// A membership write that fails stores neither the event nor anyone it
|
||||
/// names, and the same event applies in full once the failure is gone. A test
|
||||
/// trigger fails the second person's row after the event row and the first
|
||||
/// person were written in the same transaction. Before, the relay stored the
|
||||
/// event, logged the failure and accepted it; a resend then returned
|
||||
/// `duplicate:` and nobody was ever added. Mutations: store the event in its
|
||||
/// own transaction → RED (event stored after the failure); accept the event
|
||||
/// when the write fails → RED; apply a duplicate → RED (second notice).
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn failed_put_user_stores_nothing() {
|
||||
let state = state().await;
|
||||
let tenant = community(&state, "put-user-atomic-fail").await;
|
||||
let (owner, person, failing) = (Keys::generate(), Keys::generate(), Keys::generate());
|
||||
let channel_id = create_channel(&state, &tenant, &owner, &[]).await;
|
||||
let (person_hex, failing_hex) = (person.public_key().to_hex(), failing.public_key().to_hex());
|
||||
|
||||
// Each test runs in its own database, so this trigger affects only it. It
|
||||
// fails the admin row, which only `failing` asks for.
|
||||
for statement in [
|
||||
"CREATE FUNCTION put_user_atomic_fail() RETURNS trigger LANGUAGE plpgsql AS \
|
||||
$$ BEGIN RAISE EXCEPTION 'injected membership failure'; END $$",
|
||||
"CREATE TRIGGER put_user_atomic_fail BEFORE INSERT ON channel_members \
|
||||
FOR EACH ROW WHEN (NEW.role = 'admin') EXECUTE FUNCTION put_user_atomic_fail()",
|
||||
] {
|
||||
sqlx::query(statement)
|
||||
.execute(state.db.pool())
|
||||
.await
|
||||
.expect("install failure trigger");
|
||||
}
|
||||
|
||||
let event = put_user(
|
||||
&owner,
|
||||
channel_id,
|
||||
&[&[&person_hex], &[&failing_hex, "admin"]],
|
||||
&[],
|
||||
);
|
||||
for attempt in ["first send", "resend"] {
|
||||
match ingest(&state, &tenant, event.clone(), &owner).await {
|
||||
Ok(result) => panic!(
|
||||
"{attempt}: a failed put-user was reported as {}: {:?}",
|
||||
if result.accepted {
|
||||
"accepted"
|
||||
} else {
|
||||
"refused"
|
||||
},
|
||||
result.message
|
||||
),
|
||||
Err(error) => assert!(
|
||||
format!("{error:?}").contains("membership write failed"),
|
||||
"{attempt}: {error:?}"
|
||||
),
|
||||
}
|
||||
assert_eq!(
|
||||
put_user_events(&state, &tenant, channel_id).await,
|
||||
0,
|
||||
"{attempt}"
|
||||
);
|
||||
for keys in [&person, &failing] {
|
||||
assert_eq!(
|
||||
role_of(&state, &tenant, channel_id, keys).await,
|
||||
None,
|
||||
"{attempt}: nobody is added"
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
joined_notices(&state, &tenant, channel_id, &person).await,
|
||||
0,
|
||||
"{attempt}: no notice announces an add that did not happen"
|
||||
);
|
||||
}
|
||||
|
||||
// With the failure gone, the same signed event applies in full.
|
||||
sqlx::query("DROP TRIGGER put_user_atomic_fail ON channel_members")
|
||||
.execute(state.db.pool())
|
||||
.await
|
||||
.expect("drop failure trigger");
|
||||
let retry = ingest(&state, &tenant, event.clone(), &owner)
|
||||
.await
|
||||
.expect("retry");
|
||||
assert!(retry.accepted, "{}", retry.message);
|
||||
assert_eq!(
|
||||
report(&retry.message),
|
||||
serde_json::json!({ "added": [person_hex, failing_hex] })
|
||||
);
|
||||
assert_eq!(
|
||||
role_of(&state, &tenant, channel_id, &person)
|
||||
.await
|
||||
.as_deref(),
|
||||
Some("member")
|
||||
);
|
||||
assert_eq!(
|
||||
role_of(&state, &tenant, channel_id, &failing)
|
||||
.await
|
||||
.as_deref(),
|
||||
Some("admin")
|
||||
);
|
||||
assert_eq!(put_user_events(&state, &tenant, channel_id).await, 1);
|
||||
|
||||
// A resend of the stored event is a duplicate and changes nothing.
|
||||
let resend = ingest(&state, &tenant, event, &owner)
|
||||
.await
|
||||
.expect("resend");
|
||||
assert!(resend.accepted);
|
||||
assert_eq!(resend.message, "duplicate:");
|
||||
assert_eq!(
|
||||
joined_notices(&state, &tenant, channel_id, &person).await,
|
||||
1
|
||||
);
|
||||
}
|
||||
|
||||
@@ -335,9 +335,9 @@ pub async fn handle_side_effects(
|
||||
match kind {
|
||||
0 => handle_kind0_profile(tenant, event, state).await,
|
||||
5 => handle_standard_deletion_event(tenant, event, state).await,
|
||||
// kind:9000 belongs to the put-user path in ingest, which reports
|
||||
// each person's result.
|
||||
9000 => Err(anyhow::anyhow!("kind:9000 requires apply_put_user")),
|
||||
// kind:9000 belongs to the atomic put-user path in ingest, which
|
||||
// reports each person's result.
|
||||
9000 => Err(anyhow::anyhow!("kind:9000 requires persist_put_user")),
|
||||
9001 => handle_remove_user(tenant, event, state).await,
|
||||
// kind:9002 belongs to the atomic persistence path in ingest.
|
||||
9002 => Err(anyhow::anyhow!("kind:9002 requires atomic persistence")),
|
||||
@@ -597,7 +597,7 @@ pub async fn validate_admin_event(
|
||||
|
||||
match kind {
|
||||
// Ingest validates kind:9000 with `validate_put_user`, which keeps
|
||||
// each person's result for `apply_put_user`.
|
||||
// each person's result for `persist_put_user`.
|
||||
9000 => Err(anyhow::anyhow!("kind:9000 requires validate_put_user")),
|
||||
9001 => {
|
||||
// REMOVE_USER: self-remove allowed unless actor is the last owner; removing others requires owner/admin
|
||||
@@ -1760,17 +1760,27 @@ impl PutUserReport {
|
||||
}
|
||||
}
|
||||
|
||||
/// Apply a stored kind:9000 and return its OK message. An event naming
|
||||
/// several people reports each person's result; one naming a single person
|
||||
/// keeps the empty message a single add has always had. `put_member` decides
|
||||
/// every admission rule again under the channel's membership lock, so a person
|
||||
/// validation allowed can still be refused here.
|
||||
pub(crate) async fn apply_put_user(
|
||||
/// Store a validated kind:9000 and apply it, returning the stored event,
|
||||
/// whether it was new, and its OK message.
|
||||
///
|
||||
/// One transaction stores the event and writes every allowed person. If it
|
||||
/// fails, nothing is stored and the client is told, so it can resend the same
|
||||
/// event. `put_members_with_event` decides every admission rule again under
|
||||
/// the channel's membership lock, so a person validation allowed can still be
|
||||
/// refused there; when everyone is refused, nothing is stored and the error is
|
||||
/// a [`buzz_db::DbError::AccessDenied`] with the same text validation gives.
|
||||
///
|
||||
/// An event naming several people reports each person's result; one naming a
|
||||
/// single person keeps the empty message a single add has always had. Live
|
||||
/// effects (cache invalidation, notices, discovery events) run only after the
|
||||
/// commit.
|
||||
pub(crate) async fn persist_put_user(
|
||||
tenant: &TenantContext,
|
||||
event: &Event,
|
||||
state: &Arc<AppState>,
|
||||
plan: PutUserPlan,
|
||||
) -> String {
|
||||
thread_meta: Option<buzz_db::event::ThreadMetadataParams<'_>>,
|
||||
) -> anyhow::Result<(buzz_core::StoredEvent, bool, String)> {
|
||||
let channel_id = plan.channel_id;
|
||||
let batch = plan.targets.len() > 1;
|
||||
let actor_bytes = event.pubkey.to_bytes().to_vec();
|
||||
@@ -1790,21 +1800,30 @@ pub(crate) async fn apply_put_user(
|
||||
requested: target.role,
|
||||
})
|
||||
.collect();
|
||||
let results = match state
|
||||
let (stored, results) = match state
|
||||
.db
|
||||
.put_members(tenant.community(), channel_id, &puts, Some(&actor_bytes))
|
||||
.await
|
||||
.put_members_with_event(tenant.community(), channel_id, event, &puts, thread_meta)
|
||||
.await?
|
||||
{
|
||||
Ok(results) => results,
|
||||
// Nothing was written.
|
||||
Err(e) => {
|
||||
error!(channel = %channel_id, error = %e, "NIP-29 PUT_USER write failed");
|
||||
for target in &allowed {
|
||||
report.reject(&target.pubkey, "error: membership write failed".into());
|
||||
buzz_db::channel_members::MemberPutEventWrite::Applied { stored, results } => {
|
||||
(stored, results)
|
||||
}
|
||||
buzz_db::channel_members::MemberPutEventWrite::Duplicate(stored) => {
|
||||
return Ok((stored, false, String::new()));
|
||||
}
|
||||
buzz_db::channel_members::MemberPutEventWrite::Refused(errors) => {
|
||||
for (target, error) in allowed.iter().zip(errors) {
|
||||
report.reject(&target.pubkey, put_refusal_reason(error));
|
||||
}
|
||||
Vec::new()
|
||||
let message = match report.rejected.as_slice() {
|
||||
// One person keeps the single-add error text.
|
||||
[(reason, _)] if !batch => reason.clone(),
|
||||
_ => report.message(),
|
||||
};
|
||||
return Err(buzz_db::DbError::AccessDenied(message).into());
|
||||
}
|
||||
};
|
||||
|
||||
// Every person's notification is tagged with the channel's type: read it
|
||||
// once. No channel row means no type to tag, so no notifications.
|
||||
let channel_type = if results.iter().any(Result::is_ok) {
|
||||
@@ -1830,13 +1849,8 @@ pub(crate) async fn apply_put_user(
|
||||
for (target, result) in allowed.iter().zip(results) {
|
||||
let put = match result {
|
||||
Ok(put) => put,
|
||||
Err(buzz_db::DbError::AccessDenied(reason)) => {
|
||||
report.reject(&target.pubkey, reason);
|
||||
continue;
|
||||
}
|
||||
Err(e) => {
|
||||
error!(channel = %channel_id, error = %e, "NIP-29 PUT_USER write failed");
|
||||
report.reject(&target.pubkey, "error: membership write failed".into());
|
||||
Err(error) => {
|
||||
report.reject(&target.pubkey, put_refusal_reason(error));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
@@ -1901,10 +1915,19 @@ pub(crate) async fn apply_put_user(
|
||||
rejected = report.rejected.iter().map(|(_, pubkeys)| pubkeys.len()).sum::<usize>(),
|
||||
"NIP-29 PUT_USER processed"
|
||||
);
|
||||
if batch {
|
||||
let message = if batch {
|
||||
report.message()
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
Ok((stored, true, message))
|
||||
}
|
||||
|
||||
/// The reported reason a person was refused under the membership lock.
|
||||
fn put_refusal_reason(error: buzz_db::DbError) -> String {
|
||||
match error {
|
||||
buzz_db::DbError::AccessDenied(reason) => reason,
|
||||
other => other.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user