diff --git a/crates/buzz-cli/src/commands/projects.rs b/crates/buzz-cli/src/commands/projects.rs index 0c985712cd..815d28bb57 100644 --- a/crates/buzz-cli/src/commands/projects.rs +++ b/crates/buzz-cli/src/commands/projects.rs @@ -19,11 +19,12 @@ use buzz_core::kind::{KIND_PROJECT, KIND_PROJECT_STATE}; use buzz_core::project_state::validate_project_state_projection; use buzz_sdk::{ - build_delete_addressable, build_project, build_project_with_tags, ProjectMemberCoord, - PROJECT_D_MAX_LEN, + build_delete_addressable, build_project, build_project_related_channel_change, + build_project_with_tags, ProjectMemberCoord, PROJECT_D_MAX_LEN, }; use nostr::{Event, EventBuilder, PublicKey, Tag, Timestamp}; use sha2::{Digest, Sha256}; +use uuid::Uuid; use crate::agent_management::{build_project_channel, CreateProjectChannelDraft}; use crate::client::{normalize_events, BuzzClient}; @@ -503,19 +504,19 @@ fn parse_project_state( event: Event, relay_pubkey: &str, coordinate: &str, -) -> Result { +) -> Result<(Event, u64), CliError> { let relay_pubkey = PublicKey::parse(relay_pubkey) .map_err(|error| CliError::Other(format!("relay self pubkey is invalid: {error}")))?; - validate_project_state_projection(&event, &relay_pubkey, coordinate) + let revision = validate_project_state_projection(&event, &relay_pubkey, coordinate) .map_err(|error| CliError::Other(format!("Project State is invalid: {error}")))?; - Ok(event) + Ok((event, revision)) } async fn fetch_project_state( client: &BuzzClient, slug: &str, owner: Option<&str>, -) -> Result { +) -> Result<(Event, u64), CliError> { let owner = project_owner(client, owner)?; let coordinate = project_coordinate(&owner, slug)?; let relay_info: serde_json::Value = serde_json::from_str(&client.get_public("/").await?) @@ -587,11 +588,56 @@ async fn cmd_state( owner: Option<&str>, format: &crate::OutputFormat, ) -> Result<(), CliError> { - let event = fetch_project_state(client, slug, owner).await?; + let (event, _) = fetch_project_state(client, slug, owner).await?; println!("{}", format_project_state(&event, format)?); Ok(()) } +async fn cmd_change_related_channels( + client: &BuzzClient, + slug: &str, + owner: Option<&str>, + add: &[Uuid], + remove: &[Uuid], +) -> Result<(), CliError> { + let owner = project_owner(client, owner)?; + let coordinate = project_coordinate(&owner, slug)?; + let (_, revision) = fetch_project_state(client, slug, Some(&owner)).await?; + let event = client.sign_event( + build_project_related_channel_change(&coordinate, revision, add, remove) + .map_err(|error| CliError::Usage(error.to_string()))?, + )?; + let raw = map_project_change_submit(client.submit_event(event).await)?; + let response: serde_json::Value = serde_json::from_str(&raw) + .map_err(|error| CliError::Other(format!("relay response is invalid: {error}")))?; + if !response + .get("accepted") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + { + let message = response + .get("message") + .and_then(serde_json::Value::as_str) + .unwrap_or("Project change was rejected"); + return if message.starts_with("conflict:") { + Err(CliError::Conflict(message.into())) + } else { + Err(CliError::Other(format!("relay rejected event: {message}"))) + }; + } + println!("{}", crate::client::normalize_write_response(&raw)); + Ok(()) +} + +fn map_project_change_submit(result: Result) -> Result { + match result { + Err(CliError::Relay { status: 400, body }) if body.starts_with("conflict:") => { + Err(CliError::Conflict(body)) + } + result => result, + } +} + /// `buzz projects list` pub async fn cmd_list( client: &BuzzClient, @@ -960,6 +1006,12 @@ pub async fn dispatch( ) .await } + ProjectsCmd::ChangeRelatedChannels { + slug, + owner, + add, + remove, + } => cmd_change_related_channels(client, &slug, owner.as_deref(), &add, &remove).await, ProjectsCmd::RemoveRepo { slug, repo } => cmd_remove_repo(client, &slug, &repo).await, ProjectsCmd::Update { slug, @@ -1760,6 +1812,16 @@ mod tests { .expect("client construction") } + #[test] + fn project_change_conflict_uses_the_cli_conflict_exit_path() { + let error = map_project_change_submit(Err(CliError::Relay { + status: 400, + body: "conflict: Project revision is 9".into(), + })) + .expect_err("conflict response must remain distinguishable"); + assert!(matches!(error, CliError::Conflict(_))); + } + /// Creating without --repo or --channel must fail locally; the default /// repository needs a home channel to bind as git ACL. #[tokio::test] diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index ab1ee8e6b1..4cc1be1f01 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1365,6 +1365,24 @@ pub enum ProjectsCmd { #[arg(long)] template: Option, }, + /// Atomically add or remove related channels at the current Project revision + #[command( + name = "change-related-channels", + group = clap::ArgGroup::new("related-channel-mutation").required(true).multiple(true) + )] + ChangeRelatedChannels { + /// Project slug + slug: String, + /// Project owner pubkey (64-character lowercase hex) + #[arg(long)] + owner: Option, + /// Related channel UUID to add (repeatable) + #[arg(long = "add", group = "related-channel-mutation")] + add: Vec, + /// Related channel UUID to remove (repeatable) + #[arg(long = "remove", group = "related-channel-mutation")] + remove: Vec, + }, /// Remove one or more member repositories from a project #[command(name = "remove-repo")] RemoveRepo { @@ -2408,6 +2426,7 @@ mod tests { vec![ "add-channel", "add-repo", + "change-related-channels", "create", "delete", "get", @@ -2454,7 +2473,7 @@ mod tests { ("pack", 2), ("patches", 4), ("pr", 5), - ("projects", 9), + ("projects", 10), ("reactions", 3), ("repos", 5), ("social", 7), @@ -2551,6 +2570,19 @@ mod tests { Cli::try_parse_from(["buzz", "projects", "state", "platform", "--owner", &owner,]) .is_ok() ); + assert!(Cli::try_parse_from([ + "buzz", + "projects", + "change-related-channels", + "platform", + "--add", + "11111111-1111-4111-8111-111111111111", + ]) + .is_ok()); + assert!( + Cli::try_parse_from(["buzz", "projects", "change-related-channels", "platform",]) + .is_err() + ); } /// Multiple independent fields must be accepted in the same invocation. diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 906aaaa075..aeb1e0c089 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -630,6 +630,11 @@ pub const KIND_GIT_STATUS_DRAFT: u32 = 1633; /// authority over any member: push policy reads the repository's own /// announcement, never a project. See `docs/nips/NIP-MP.md`. pub const KIND_PROJECT: u32 = 30621; +/// Actor-signed transactional change command for collaborative Project state. +/// +/// Version 1 changes only add or remove related channels. See +/// `docs/nips/NIP-PC.md`. +pub const KIND_PROJECT_CHANGE: u32 = 47010; /// Relay-signed addressable effective-state projection for a Project. /// /// The owner-signed [`KIND_PROJECT`] remains the Project identity and recovery @@ -769,6 +774,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_PROJECT, + KIND_PROJECT_CHANGE, KIND_PROJECT_STATE, ]; @@ -829,6 +835,7 @@ pub const fn is_command_kind(kind: u32) -> bool { | KIND_WORKFLOW_TRIGGER | KIND_APPROVAL_GRANT | KIND_APPROVAL_DENY + | KIND_PROJECT_CHANGE ) } @@ -871,6 +878,7 @@ const _: () = assert!(is_parameterized_replaceable(KIND_EVENT_REMINDER)); // 303 const _: () = assert!(is_parameterized_replaceable(KIND_DM_VISIBILITY)); // 30622 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_PROJECT)); // 30621 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_PROJECT_STATE)); // 30623 ∈ 30000–39999 +const _: () = assert!(is_command_kind(KIND_PROJECT_CHANGE)); const _: () = assert!(is_parameterized_replaceable(KIND_THREAD_SUMMARY)); // 39005 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_WINDOW_BOUNDS)); // 39006 ∈ 30000–39999 const _: () = assert!(is_relay_only_kind(KIND_PROJECT_STATE)); @@ -925,6 +933,9 @@ mod tests { #[test] fn project_collaboration_kinds_have_distinct_routing() { + assert!(is_command_kind(KIND_PROJECT_CHANGE)); + assert!(!is_relay_only_kind(KIND_PROJECT_CHANGE)); + assert!(!is_parameterized_replaceable(KIND_PROJECT_CHANGE)); assert!(is_parameterized_replaceable(KIND_PROJECT_STATE)); assert!(is_relay_only_kind(KIND_PROJECT_STATE)); assert!(!is_command_kind(KIND_PROJECT_STATE)); diff --git a/crates/buzz-core/src/project_state.rs b/crates/buzz-core/src/project_state.rs index 86d6bae039..2ff9b3f177 100644 --- a/crates/buzz-core/src/project_state.rs +++ b/crates/buzz-core/src/project_state.rs @@ -113,7 +113,7 @@ struct ProjectionBody { project_tags: Vec>, } -/// Validate a relay-authored Project State event. +/// Validate a relay-authored Project State event and return its CAS revision. /// /// This checks the event signature and advertised relay author, the requested /// Project coordinate, a canonical positive revision, and the strict version-1 @@ -123,7 +123,7 @@ pub fn validate_project_state_projection( event: &Event, relay_pubkey: &PublicKey, coordinate: &str, -) -> Result<(), ProjectStateError> { +) -> Result { parse_project_coordinate(coordinate)?; event .verify() @@ -166,6 +166,9 @@ pub fn validate_project_state_projection( "projection revision is not canonical".into(), )); } + let revision = revision + .parse::() + .map_err(|error| ProjectStateError(format!("invalid projection revision: {error}")))?; let body: ProjectionBody = serde_json::from_str(&event.content) .map_err(|error| ProjectStateError(format!("invalid projection JSON: {error}")))?; if body.v != 1 { @@ -173,7 +176,7 @@ pub fn validate_project_state_projection( "projection JSON is not supported version 1".into(), )); } - Ok(()) + Ok(revision) } fn canonical_positive_i64(value: &str) -> bool { @@ -488,7 +491,7 @@ mod tests { let event = signed_projection(&relay, &template); assert_eq!( validate_project_state_projection(&event, &relay.public_key(), &coordinate), - Ok(()) + Ok(7) ); let mut reordered = template.clone(); @@ -499,7 +502,7 @@ mod tests { &relay.public_key(), &coordinate, ), - Ok(()) + Ok(7) ); let impostor = Keys::generate(); diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 13b0d4e916..5602073ce3 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -54,7 +54,8 @@ pub use community::{ pub use error::{DbError, Result}; pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; pub use project_state::{ - ProjectLifecycleApplyResult, ProjectLifecycleStatus, ProjectStateProjectionCandidate, + ProjectChangeApplyResult, ProjectLifecycleApplyResult, ProjectLifecycleStatus, + ProjectRelatedChannelChange, ProjectStateProjectionCandidate, ProjectStateProjectionCommitResult, }; pub use reaction::ReactionEventInsertOutcome; diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index dd526a0a46..719136dcf7 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -173,7 +173,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. -async fn acquire_channel_membership_lock( +pub(crate) async fn acquire_channel_membership_lock( tx: &mut Transaction<'_, Postgres>, community_id: CommunityId, channel_id: Uuid, diff --git a/crates/buzz-db/src/store/project_state.rs b/crates/buzz-db/src/store/project_state.rs index 60e6d721dd..f31467a271 100644 --- a/crates/buzz-db/src/store/project_state.rs +++ b/crates/buzz-db/src/store/project_state.rs @@ -2,7 +2,9 @@ use std::collections::BTreeSet; -use buzz_core::kind::{event_kind_u32, KIND_DELETION, KIND_PROJECT, KIND_PROJECT_STATE}; +use buzz_core::kind::{ + event_kind_u32, KIND_DELETION, KIND_PROJECT, KIND_PROJECT_CHANGE, KIND_PROJECT_STATE, +}; use buzz_core::project_state::{ project_state_template, ProjectStateProjectionInput, ProjectStateTemplate, }; @@ -20,6 +22,49 @@ use crate::{Db, DbError, Result}; const RELATED_CHANNEL_CAP: usize = 64; +/// A validated Project related-channel mutation supplied by the relay parser. +#[derive(Clone, Copy, Debug)] +pub struct ProjectRelatedChannelChange<'a> { + /// Owner pubkey from the canonical kind:30621 coordinate. + pub project_owner: &'a [u8], + /// Verbatim `d` value from the canonical kind:30621 coordinate. + pub project_d_tag: &'a str, + /// Relational revision observed by the actor. + pub expected_revision: i64, + /// Channels to add to the effective related-channel set. + pub add: &'a [Uuid], + /// Channels to remove from the effective related-channel set. + pub remove: &'a [Uuid], +} + +/// Outcome of applying one Project related-channel command. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ProjectChangeApplyResult { + /// The command and mutation committed atomically. + Applied { + /// New authoritative revision. + revision: i64, + }, + /// This exact command already committed at its original revision. + Duplicate { + /// Revision originally produced by the command. + applied_revision: i64, + }, + /// No live owner-signed Project exists at the coordinate. + ProjectNotFound, + /// The Project is deleted and awaits owner recreation. + ProjectDeleted, + /// The actor is neither the Project owner nor a current home-channel owner/admin. + Forbidden, + /// The expected revision is stale. + Conflict { + /// Current authoritative revision. + current_revision: i64, + }, + /// The patch or owner identity is invalid. + Invalid(String), +} + /// Coherent Project state awaiting a relay-signed kind:30623 projection. #[derive(Clone, Debug)] pub struct ProjectStateProjectionCandidate { @@ -144,6 +189,33 @@ fn parse_base_state( Ok((home, related)) } +fn validate_patch(change: ProjectRelatedChannelChange<'_>) -> Option { + if change.project_owner.len() != 32 { + return Some("Project owner must be 32 bytes".into()); + } + if change.project_d_tag.is_empty() || change.project_d_tag.len() > crate::event::D_TAG_MAX_LEN { + return Some("Project d tag is empty or too long".into()); + } + if change.expected_revision < 1 { + return Some("expected revision must be positive".into()); + } + if change.add.is_empty() && change.remove.is_empty() { + return Some("Project change must not be empty".into()); + } + if change.add.len() > RELATED_CHANNEL_CAP || change.remove.len() > RELATED_CHANNEL_CAP { + return Some("Project change exceeds the 64-channel patch bound".into()); + } + let add = change.add.iter().copied().collect::>(); + let remove = change.remove.iter().copied().collect::>(); + if add.len() != change.add.len() || remove.len() != change.remove.len() { + return Some("Project change contains a duplicate channel".into()); + } + if !add.is_disjoint(&remove) { + return Some("Project change adds and removes the same channel".into()); + } + None +} + async fn replace_related_channels( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, community_id: CommunityId, @@ -175,6 +247,48 @@ async fn replace_related_channels( Ok(()) } +/// Adopt one still-live pre-relational Project identity while the caller holds +/// the Project coordinate lock. +async fn materialize_project_head_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + owner: &[u8], + d_tag: &str, + identity_event_id: &[u8], + related: &BTreeSet, +) -> Result { + let live_event_id: Option> = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(KIND_PROJECT as i32) + .bind(owner) + .bind(d_tag) + .fetch_optional(&mut **tx) + .await?; + if live_event_id.as_deref() != Some(identity_event_id) { + return Ok(false); + } + let inserted = sqlx::query( + "INSERT INTO project_state_heads \ + (community_id, project_owner, project_d_tag, revision, deleted, identity_event_id, last_event_id) \ + VALUES ($1,$2,$3,1,FALSE,$4,$4) ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(owner) + .bind(d_tag) + .bind(identity_event_id) + .execute(&mut **tx) + .await? + .rows_affected() + == 1; + if inserted { + replace_related_channels(tx, community_id, owner, d_tag, related).await?; + } + Ok(inserted) +} + impl Db { /// Atomically accept an owner-signed Project identity and materialize it. /// @@ -196,7 +310,7 @@ impl Db { return Err(DbError::InvalidData("invalid Project d tag".into())); } let tags = serde_json::to_value(&event.tags)?; - let (_, related) = parse_base_state(&tags).map_err(DbError::InvalidData)?; + let (home_channel, mut related) = parse_base_state(&tags).map_err(DbError::InvalidData)?; let owner = event.pubkey.to_bytes(); let mut tx = self.begin_transaction().await?; @@ -279,6 +393,29 @@ impl Db { } } + if current_head.is_some() { + let preserved = sqlx::query_scalar::<_, Uuid>( + "SELECT channel_id FROM project_related_channels WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag=$3", + ) + .bind(community_id.as_uuid()) + .bind(owner.as_slice()) + .bind(&d_tag) + .fetch_all(&mut *tx) + .await?; + related.extend( + preserved + .into_iter() + .filter(|channel| Some(*channel) != home_channel), + ); + if related.len() > RELATED_CHANNEL_CAP { + return Err(DbError::InvalidData( + "owner Project update plus preserved related channels exceeds 64 channels" + .into(), + )); + } + } + let revision = match current_head { None => 1, Some(head) => head @@ -606,42 +743,235 @@ impl Db { .bind(coordinate_lock) .execute(&mut *tx) .await?; - let live_event_id: Option> = sqlx::query_scalar( - "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ - AND d_tag=$4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + let inserted = materialize_project_head_in_transaction( + &mut tx, + community_id, + owner.as_slice(), + &d_tag, + event.id.as_bytes(), + &related, ) - .bind(community_id.as_uuid()) - .bind(KIND_PROJECT as i32) - .bind(owner.as_slice()) - .bind(&d_tag) - .fetch_optional(&mut *tx) .await?; - if live_event_id.as_deref() != Some(event.id.as_bytes()) { - tx.rollback().await?; - return Ok(false); - } - let inserted = sqlx::query( - "INSERT INTO project_state_heads \ - (community_id, project_owner, project_d_tag, revision, deleted, identity_event_id, last_event_id) \ - VALUES ($1,$2,$3,1,FALSE,$4,$4) ON CONFLICT DO NOTHING", - ) - .bind(community_id.as_uuid()) - .bind(owner.as_slice()) - .bind(&d_tag) - .bind(event.id.as_bytes().as_slice()) - .execute(&mut *tx) - .await? - .rows_affected() - == 1; if !inserted { tx.rollback().await?; return Ok(false); } - replace_related_channels(&mut tx, community_id, owner.as_slice(), &d_tag, &related).await?; tx.commit().await?; Ok(true) } + /// Authorize and atomically apply one v1 Project related-channel command. + /// + /// Lock order is the community deletion fence, Project coordinate, then + /// current home-channel membership. Exact accepted-event replays are + /// detected from immutable event history while holding the coordinate lock. + pub async fn apply_project_related_channel_change( + &self, + community_id: CommunityId, + event: &Event, + change: ProjectRelatedChannelChange<'_>, + ) -> Result { + if event_kind_u32(event) != KIND_PROJECT_CHANGE { + return Err(DbError::InvalidData( + "Project change persistence requires kind 47010".into(), + )); + } + if let Some(message) = validate_patch(change) { + return Ok(ProjectChangeApplyResult::Invalid(message)); + } + + let mut tx = self.begin_transaction().await?; + self.deletion_store() + .guard_transaction(&mut tx, community_id) + .await?; + let coordinate_lock = event_replacement_lock_key( + community_id, + KIND_PROJECT as i32, + change.project_owner, + Some(change.project_d_tag.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(coordinate_lock) + .execute(&mut *tx) + .await?; + + let duplicate: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2 AND kind=$3)", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(KIND_PROJECT_CHANGE as i32) + .fetch_one(&mut *tx) + .await?; + if duplicate { + let applied_revision = change + .expected_revision + .checked_add(1) + .ok_or_else(|| DbError::InvalidData("Project command revision overflow".into()))?; + return Ok(ProjectChangeApplyResult::Duplicate { applied_revision }); + } + + let head = sqlx::query( + "SELECT revision, deleted, identity_event_id FROM project_state_heads \ + WHERE community_id=$1 AND project_owner=$2 AND project_d_tag=$3 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .fetch_optional(&mut *tx) + .await?; + if head.as_ref().is_some_and(|row| row.get("deleted")) { + return Ok(ProjectChangeApplyResult::ProjectDeleted); + } + + let base = sqlx::query( + "SELECT id, tags FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(KIND_PROJECT as i32) + .bind(change.project_owner) + .bind(change.project_d_tag) + .fetch_optional(&mut *tx) + .await?; + let Some(base) = base else { + return Ok(ProjectChangeApplyResult::ProjectNotFound); + }; + let identity_event_id: Vec = base.try_get("id")?; + let base_tags: serde_json::Value = base.try_get("tags")?; + let (home_channel, base_related) = match parse_base_state(&base_tags) { + Ok(state) => state, + Err(message) => return Ok(ProjectChangeApplyResult::Invalid(message)), + }; + + let current_revision = if let Some(head) = head { + let revision: i64 = head.try_get("revision")?; + let materialized_identity: Vec = head.try_get("identity_event_id")?; + if materialized_identity != identity_event_id { + return Ok(ProjectChangeApplyResult::Conflict { + current_revision: revision, + }); + } + revision + } else { + let inserted = materialize_project_head_in_transaction( + &mut tx, + community_id, + change.project_owner, + change.project_d_tag, + &identity_event_id, + &base_related, + ) + .await?; + if !inserted { + return Ok(ProjectChangeApplyResult::ProjectNotFound); + } + 1 + }; + + let actor = event.pubkey.to_bytes(); + if actor.as_slice() != change.project_owner { + let Some(home_channel) = home_channel else { + return Ok(ProjectChangeApplyResult::Forbidden); + }; + crate::channel_members::acquire_channel_membership_lock( + &mut tx, + community_id, + home_channel, + ) + .await?; + let authorized_role: Option = sqlx::query_scalar( + "SELECT member.role::text FROM channel_members member \ + JOIN channels channel ON channel.community_id=member.community_id \ + AND channel.id=member.channel_id \ + WHERE member.community_id=$1 AND member.channel_id=$2 AND member.pubkey=$3 \ + AND member.role IN ('owner', 'admin') AND member.removed_at IS NULL \ + AND channel.archived_at IS NULL AND channel.deleted_at IS NULL \ + LIMIT 1 FOR SHARE OF channel, member", + ) + .bind(community_id.as_uuid()) + .bind(home_channel) + .bind(actor.as_slice()) + .fetch_optional(&mut *tx) + .await?; + if authorized_role.is_none() { + return Ok(ProjectChangeApplyResult::Forbidden); + } + } + if change.expected_revision != current_revision { + return Ok(ProjectChangeApplyResult::Conflict { current_revision }); + } + + let rows = sqlx::query_scalar::<_, Uuid>( + "SELECT channel_id FROM project_related_channels WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag=$3 ORDER BY channel_id", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .fetch_all(&mut *tx) + .await?; + let mut related = rows.into_iter().collect::>(); + for channel in change.add { + if Some(*channel) == home_channel || !related.insert(*channel) { + return Ok(ProjectChangeApplyResult::Invalid( + "cannot add the home channel or an already-related channel".into(), + )); + } + } + for channel in change.remove { + if !related.remove(channel) { + return Ok(ProjectChangeApplyResult::Invalid( + "cannot remove a channel that is not related".into(), + )); + } + } + if related.len() > RELATED_CHANNEL_CAP { + return Ok(ProjectChangeApplyResult::Invalid( + "effective Project state exceeds 64 related channels".into(), + )); + } + + let (_, inserted) = insert_event_in_transaction(&mut tx, community_id, event, None).await?; + if !inserted { + return Err(DbError::InvalidData( + "Project command appeared concurrently outside its coordinate lock".into(), + )); + } + replace_related_channels( + &mut tx, + community_id, + change.project_owner, + change.project_d_tag, + &related, + ) + .await?; + let revision = current_revision + .checked_add(1) + .ok_or_else(|| DbError::InvalidData("Project revision overflow".into()))?; + let updated = sqlx::query( + "UPDATE project_state_heads SET revision=$4, last_event_id=$5, \ + updated_at=transaction_timestamp() WHERE community_id=$1 AND project_owner=$2 \ + AND project_d_tag=$3 AND revision=$6", + ) + .bind(community_id.as_uuid()) + .bind(change.project_owner) + .bind(change.project_d_tag) + .bind(revision) + .bind(event.id.as_bytes().as_slice()) + .bind(current_revision) + .execute(&mut *tx) + .await? + .rows_affected(); + if updated != 1 { + return Err(DbError::InvalidData( + "Project CAS failed under coordinate lock".into(), + )); + } + tx.commit().await?; + Ok(ProjectChangeApplyResult::Applied { revision }) + } + /// Load one coherent Project state when it requires publication by /// `projection_pubkey`. /// @@ -878,6 +1208,7 @@ fn projection_d_tag(template: &ProjectStateTemplate) -> Result<&str> { #[cfg(test)] mod postgres_tests { + use buzz_core::channel::{ChannelType, ChannelVisibility}; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; @@ -912,6 +1243,109 @@ mod postgres_tests { .expect("load Project head") } + async fn create_channel(pool: &PgPool, community: CommunityId, channel: Uuid, owner: &Keys) { + crate::channel::create_channel_with_id( + pool, + community, + channel, + &format!("project-state-{channel}"), + ChannelType::Stream, + ChannelVisibility::Private, + None, + owner.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create test channel"); + } + + async fn set_role( + pool: &PgPool, + community: CommunityId, + channel: Uuid, + user: &Keys, + role: &str, + ) { + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ + VALUES ($1,$2,$3,$4::member_role,$3) ON CONFLICT \ + (community_id, channel_id, pubkey) DO UPDATE SET role=EXCLUDED.role, \ + removed_at=NULL, removed_by=NULL", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(user.public_key().to_bytes().as_slice()) + .bind(role) + .execute(pool) + .await + .expect("set channel role"); + } + + fn project_identity(owner: &Keys, d_tag: &str, home: Option, created_at: u64) -> Event { + let mut tags = vec![Tag::parse(["d", d_tag]).expect("d tag")]; + if let Some(home) = home { + tags.push(Tag::parse(["buzz-channel", &home.to_string()]).expect("home tag")); + } + EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags(tags) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(owner) + .expect("sign Project identity") + } + + fn project_change( + actor: &Keys, + owner: &Keys, + d_tag: &str, + revision: i64, + add: &[Uuid], + remove: &[Uuid], + created_at: u64, + ) -> Event { + let coordinate = format!("{KIND_PROJECT}:{}:{d_tag}", owner.public_key().to_hex()); + EventBuilder::new( + Kind::Custom(KIND_PROJECT_CHANGE as u16), + serde_json::json!({ + "v": 1, + "add_related_channels": add, + "remove_related_channels": remove, + }) + .to_string(), + ) + .tags([ + Tag::parse(["a", &coordinate]).expect("coordinate tag"), + Tag::parse(["r", &revision.to_string()]).expect("revision tag"), + ]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(actor) + .expect("sign Project change") + } + + fn change<'a>( + owner: &'a [u8], + d_tag: &'a str, + revision: i64, + add: &'a [Uuid], + remove: &'a [Uuid], + ) -> ProjectRelatedChannelChange<'a> { + ProjectRelatedChannelChange { + project_owner: owner, + project_d_tag: d_tag, + expected_revision: revision, + add, + remove, + } + } + + async fn event_exists(pool: &PgPool, community: CommunityId, event: &Event) -> bool { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2)") + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(pool) + .await + .expect("query event") + } + #[tokio::test] #[ignore = "requires Postgres"] async fn lifecycle_materializes_brownfield_state_and_preserves_monotonic_revisions() { @@ -1145,4 +1579,442 @@ mod postgres_tests { pool.close().await; } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn related_channel_changes_replay_cas_preserve_owner_updates_and_reconcile() { + let pool = PgPoolOptions::new() + .max_connections(8) + .connect(&crate::test_support::database_url()) + .await + .expect("connect test database"); + let db = Db::from_pool(pool.clone()); + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1,$2)") + .bind(community.as_uuid()) + .bind(format!("project-change-{}.example", community.as_uuid())) + .execute(&pool) + .await + .expect("insert community"); + + let owner = Keys::generate(); + let home_owner = Keys::generate(); + let admin = Keys::generate(); + let home = Uuid::new_v4(); + create_channel(&pool, community, home, &home_owner).await; + set_role(&pool, community, home, &admin, "admin").await; + let related = [Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4()]; + for channel in related { + create_channel(&pool, community, channel, &home_owner).await; + } + let owner_bytes = owner.public_key().to_bytes(); + let started_at = Utc::now().timestamp() as u64 - 100; + db.apply_project_identity_event( + community, + &project_identity(&owner, "shared", Some(home), started_at), + ) + .await + .expect("apply Project identity"); + + let first = project_change( + &admin, + &owner, + "shared", + 1, + &related[..1], + &[], + started_at + 1, + ); + assert_eq!( + db.apply_project_related_channel_change( + community, + &first, + change(owner_bytes.as_slice(), "shared", 1, &related[..1], &[]), + ) + .await + .expect("admin change"), + ProjectChangeApplyResult::Applied { revision: 2 } + ); + let second = project_change( + &owner, + &owner, + "shared", + 2, + &related[1..2], + &[], + started_at + 2, + ); + assert_eq!( + db.apply_project_related_channel_change( + community, + &second, + change(owner_bytes.as_slice(), "shared", 2, &related[1..2], &[]), + ) + .await + .expect("owner change"), + ProjectChangeApplyResult::Applied { revision: 3 } + ); + assert_eq!( + db.apply_project_related_channel_change( + community, + &first, + change(owner_bytes.as_slice(), "shared", 1, &related[..1], &[]), + ) + .await + .expect("replay after later revision"), + ProjectChangeApplyResult::Duplicate { + applied_revision: 2 + } + ); + sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(first.id.as_bytes().as_slice()) + .execute(&pool) + .await + .expect("soft-delete command row"); + assert_eq!( + db.apply_project_related_channel_change( + community, + &first, + change(owner_bytes.as_slice(), "shared", 1, &related[..1], &[]), + ) + .await + .expect("replay soft-deleted command"), + ProjectChangeApplyResult::Duplicate { + applied_revision: 2 + } + ); + + let stale = project_change( + &admin, + &owner, + "shared", + 1, + &related[2..], + &[], + started_at + 3, + ); + assert_eq!( + db.apply_project_related_channel_change( + community, + &stale, + change(owner_bytes.as_slice(), "shared", 1, &related[2..], &[]), + ) + .await + .expect("reject stale command"), + ProjectChangeApplyResult::Conflict { + current_revision: 3 + } + ); + assert!(!event_exists(&pool, community, &stale).await); + let resigned = project_change( + &admin, + &owner, + "shared", + 3, + &related[2..], + &[], + started_at + 4, + ); + assert_eq!( + db.apply_project_related_channel_change( + community, + &resigned, + change(owner_bytes.as_slice(), "shared", 3, &related[2..], &[]), + ) + .await + .expect("apply re-signed command"), + ProjectChangeApplyResult::Applied { revision: 4 } + ); + + let replacement = project_identity(&owner, "shared", Some(home), started_at + 10); + assert_eq!( + db.apply_project_identity_event(community, &replacement) + .await + .expect("replace owner identity") + .status, + ProjectLifecycleStatus::Applied + ); + let mut stale_tags = vec![Tag::parse(["d", "shared"]).expect("d tag")]; + stale_tags.extend((0..62).map(|_| { + Tag::parse(["buzz-related-channel", &Uuid::new_v4().to_string()]).expect("related tag") + })); + let stale_identity = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags(stale_tags) + .custom_created_at(Timestamp::from(started_at + 9)) + .sign_with_keys(&owner) + .expect("sign stale identity"); + assert_eq!( + db.apply_project_identity_event(community, &stale_identity) + .await + .expect("stale identity remains superseded despite combined cap") + .status, + ProjectLifecycleStatus::Superseded + ); + let effective: Vec = sqlx::query_scalar( + "SELECT channel_id FROM project_related_channels WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag='shared' ORDER BY channel_id", + ) + .bind(community.as_uuid()) + .bind(owner_bytes.as_slice()) + .fetch_all(&pool) + .await + .expect("load preserved channels"); + assert_eq!( + effective.into_iter().collect::>(), + related.into() + ); + + let relay = Keys::generate(); + let relay_bytes = relay.public_key().to_bytes(); + let candidate = db + .load_pending_project_state_projections(relay_bytes.as_slice(), 100) + .await + .expect("load pending projections") + .into_iter() + .find(|candidate| { + candidate.community_id == community && candidate.project_d_tag == "shared" + }) + .expect("missed publish remains reconcilable"); + assert_eq!(candidate.revision, 5); + assert_eq!( + db.commit_project_state_projection( + &candidate, + &projection(&candidate, &relay, Timestamp::now().as_secs()), + ) + .await + .expect("commit reconciled projection"), + ProjectStateProjectionCommitResult::Committed + ); + assert_eq!( + head(&pool, community, owner_bytes.as_slice(), "shared").await, + (5, false) + ); + pool.close().await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn related_channel_change_authorization_and_concurrent_cas_are_exact() { + let pool = PgPoolOptions::new() + .max_connections(12) + .connect(&crate::test_support::database_url()) + .await + .expect("connect test database"); + let db = Db::from_pool(pool.clone()); + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1,$2)") + .bind(community.as_uuid()) + .bind(format!("project-auth-{}.example", community.as_uuid())) + .execute(&pool) + .await + .expect("insert community"); + let owner = Keys::generate(); + let channel_owner = Keys::generate(); + let admin = Keys::generate(); + let member = Keys::generate(); + let home = Uuid::new_v4(); + create_channel(&pool, community, home, &channel_owner).await; + set_role(&pool, community, home, &admin, "admin").await; + set_role(&pool, community, home, &member, "member").await; + let related = [Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4()]; + for channel in related { + create_channel(&pool, community, channel, &channel_owner).await; + } + let owner_bytes = owner.public_key().to_bytes(); + let now = Utc::now().timestamp() as u64 - 100; + db.apply_project_identity_event( + community, + &project_identity(&owner, "auth", Some(home), now), + ) + .await + .expect("apply Project identity"); + + for (actor, label) in [(&member, "member"), (&admin, "removed admin")] { + if label == "removed admin" { + sqlx::query( + "UPDATE channel_members SET removed_at=now() WHERE community_id=$1 \ + AND channel_id=$2 AND pubkey=$3", + ) + .bind(community.as_uuid()) + .bind(home) + .bind(admin.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("remove admin"); + } + let event = project_change(actor, &owner, "auth", 1, &related[..1], &[], now + 1); + assert_eq!( + db.apply_project_related_channel_change( + community, + &event, + change(owner_bytes.as_slice(), "auth", 1, &related[..1], &[]), + ) + .await + .unwrap_or_else(|error| panic!("{label} check failed: {error}")), + ProjectChangeApplyResult::Forbidden + ); + assert!(!event_exists(&pool, community, &event).await); + } + set_role(&pool, community, home, &admin, "admin").await; + for (disable, restore) in [ + ( + "UPDATE channels SET archived_at=now() WHERE community_id=$1 AND id=$2", + "UPDATE channels SET archived_at=NULL WHERE community_id=$1 AND id=$2", + ), + ( + "UPDATE channels SET deleted_at=now() WHERE community_id=$1 AND id=$2", + "UPDATE channels SET deleted_at=NULL WHERE community_id=$1 AND id=$2", + ), + ] { + sqlx::query(disable) + .bind(community.as_uuid()) + .bind(home) + .execute(&pool) + .await + .expect("disable home channel"); + let event = project_change(&admin, &owner, "auth", 1, &related[..1], &[], now + 2); + assert_eq!( + db.apply_project_related_channel_change( + community, + &event, + change(owner_bytes.as_slice(), "auth", 1, &related[..1], &[]), + ) + .await + .expect("reject inactive home channel"), + ProjectChangeApplyResult::Forbidden + ); + sqlx::query(restore) + .bind(community.as_uuid()) + .bind(home) + .execute(&pool) + .await + .expect("restore home channel"); + } + + let exact = project_change(&admin, &owner, "auth", 1, &related[..1], &[], now + 3); + let patch = change(owner_bytes.as_slice(), "auth", 1, &related[..1], &[]); + let (left, right) = tokio::join!( + db.apply_project_related_channel_change(community, &exact, patch), + db.apply_project_related_channel_change(community, &exact, patch), + ); + let outcomes = [ + left.expect("first duplicate race"), + right.expect("second duplicate race"), + ]; + assert!(outcomes.contains(&ProjectChangeApplyResult::Applied { revision: 2 })); + assert!(outcomes.contains(&ProjectChangeApplyResult::Duplicate { + applied_revision: 2 + })); + + let left_event = project_change(&admin, &owner, "auth", 2, &related[1..2], &[], now + 4); + let right_event = project_change(&admin, &owner, "auth", 2, &related[2..], &[], now + 5); + let (left, right) = tokio::join!( + db.apply_project_related_channel_change( + community, + &left_event, + change(owner_bytes.as_slice(), "auth", 2, &related[1..2], &[]), + ), + db.apply_project_related_channel_change( + community, + &right_event, + change(owner_bytes.as_slice(), "auth", 2, &related[2..], &[]), + ), + ); + let left = left.expect("first CAS race"); + let right = right.expect("second CAS race"); + let outcomes = [left.clone(), right.clone()]; + assert!(outcomes.contains(&ProjectChangeApplyResult::Applied { revision: 3 })); + assert!(outcomes.contains(&ProjectChangeApplyResult::Conflict { + current_revision: 3 + })); + assert_eq!( + event_exists(&pool, community, &left_event).await, + matches!(left, ProjectChangeApplyResult::Applied { .. }) + ); + assert_eq!( + event_exists(&pool, community, &right_event).await, + matches!(right, ProjectChangeApplyResult::Applied { .. }) + ); + let stored_channels = sqlx::query_scalar::<_, Uuid>( + "SELECT channel_id FROM project_related_channels WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag='auth' ORDER BY channel_id", + ) + .bind(community.as_uuid()) + .bind(owner_bytes.as_slice()) + .fetch_all(&pool) + .await + .expect("load channels after CAS race"); + assert_eq!(stored_channels.len(), 2); + assert!(stored_channels.contains(&related[0])); + assert_eq!( + stored_channels.contains(&related[1]), + matches!(left, ProjectChangeApplyResult::Applied { .. }) + ); + assert_eq!( + stored_channels.contains(&related[2]), + matches!(right, ProjectChangeApplyResult::Applied { .. }) + ); + + let no_home = project_identity(&owner, "no-home", None, now + 6); + db.apply_project_identity_event(community, &no_home) + .await + .expect("apply Project without home"); + let denied = project_change(&admin, &owner, "no-home", 1, &related[..1], &[], now + 7); + assert_eq!( + db.apply_project_related_channel_change( + community, + &denied, + change(owner_bytes.as_slice(), "no-home", 1, &related[..1], &[]), + ) + .await + .expect("deny admin without home"), + ProjectChangeApplyResult::Forbidden + ); + let allowed = project_change(&owner, &owner, "no-home", 1, &related[..1], &[], now + 8); + assert_eq!( + db.apply_project_related_channel_change( + community, + &allowed, + change(owner_bytes.as_slice(), "no-home", 1, &related[..1], &[]), + ) + .await + .expect("owner bypasses home requirement"), + ProjectChangeApplyResult::Applied { revision: 2 } + ); + + db.apply_project_identity_event( + community, + &project_identity(&owner, "deleted", Some(home), now + 9), + ) + .await + .expect("apply Project to delete"); + let coordinate = format!("{KIND_PROJECT}:{}:deleted", owner.public_key().to_hex()); + let deletion = EventBuilder::new(Kind::EventDeletion, "") + .tag(Tag::parse(["a", &coordinate]).expect("coordinate tag")) + .custom_created_at(Timestamp::from(now + 10)) + .sign_with_keys(&owner) + .expect("sign Project deletion"); + db.apply_project_deletion_event( + community, + &deletion, + owner_bytes.as_slice(), + "deleted", + None, + ) + .await + .expect("delete Project"); + let denied = project_change(&admin, &owner, "deleted", 2, &related[1..2], &[], now + 11); + assert_eq!( + db.apply_project_related_channel_change( + community, + &denied, + change(owner_bytes.as_slice(), "deleted", 2, &related[1..2], &[],), + ) + .await + .expect("deny deleted Project mutation"), + ProjectChangeApplyResult::ProjectDeleted + ); + pool.close().await; + } } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 41d7900b7d..b6254b381b 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -70,6 +70,7 @@ pub async fn handle_command( KIND_WORKFLOW_TRIGGER => handle_workflow_trigger(tenant, state, &event, &auth).await, KIND_APPROVAL_GRANT => handle_approval_grant(tenant, state, &event, &auth).await, KIND_APPROVAL_DENY => handle_approval_deny(tenant, state, &event, &auth).await, + KIND_PROJECT_CHANGE => super::project_change::handle(tenant, state, &event).await, _ => Err(IngestError::Rejected(format!( "unknown command kind: {kind}" ))), diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index e781a2353f..7d1db24f3b 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -28,13 +28,13 @@ use buzz_core::kind::{ KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, KIND_PRESENCE_UPDATE, - KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, KIND_REACTION, - KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, - KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, - KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, - KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, - RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, - RELAY_ADMIN_SET_WORKSPACE_PROFILE, + KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, + KIND_PROJECT_CHANGE, KIND_REACTION, KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, + KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, + KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, + KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, + RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; @@ -529,7 +529,7 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result Ok(Scope::ReposWrite), // NIP-MP: a project is repository metadata — grouping repositories needs // the same scope as announcing them. - KIND_PROJECT => Ok(Scope::ReposWrite), + KIND_PROJECT | KIND_PROJECT_CHANGE => Ok(Scope::ReposWrite), KIND_GIT_PATCH | KIND_GIT_PULL_REQUEST | KIND_GIT_PR_UPDATE @@ -671,6 +671,7 @@ pub(crate) fn is_global_only_kind(kind: u32) -> bool { // `buzz-channel` tag is a metadata reference, not a routing directive, // so a project's state is never channel-scoped. | KIND_PROJECT + | KIND_PROJECT_CHANGE // Community moderation commands (9040–9044): community-global // direct commands, same model as the NIP-43 9030-series. A stray // `h` tag must never channel-scope them (pinned contract — @@ -2320,7 +2321,7 @@ async fn ingest_event_inner( let event = std::sync::Arc::try_unwrap(event).unwrap_or_else(|arc| (*arc).clone()); const MAX_TIMESTAMP_DRIFT_SECS: i64 = 900; // ±15 minutes - let now = chrono::Utc::now().timestamp(); + let now = Utc::now().timestamp(); let event_ts = event.created_at.as_secs() as i64; if (event_ts - now).abs() > MAX_TIMESTAMP_DRIFT_SECS { return Err(IngestError::Rejected( @@ -2364,6 +2365,11 @@ async fn ingest_event_inner( "restricted: leave requests require a global token".into(), )); } + if kind_u32 == KIND_PROJECT_CHANGE && auth.channel_ids().is_some() { + return Err(IngestError::AuthFailed( + "restricted: Project changes require a global token".into(), + )); + } if !auth.scopes().contains(&required) { return Err(IngestError::AuthFailed(format!( "restricted: insufficient scope (need {})", diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index b58f4cabff..e52ea88cad 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -26,6 +26,7 @@ pub mod moderation_commands; pub mod moderation_notices; /// Product-feedback validation + deployment sidecar persistence. pub mod product_feedback; +mod project_change; /// Relay signing and durable repair for NIP-PC Project State projections. pub mod project_state_projection; #[allow(dead_code, missing_docs)] diff --git a/crates/buzz-relay/src/handlers/project_change.rs b/crates/buzz-relay/src/handlers/project_change.rs new file mode 100644 index 0000000000..cbce61443d --- /dev/null +++ b/crates/buzz-relay/src/handlers/project_change.rs @@ -0,0 +1,278 @@ +//! NIP-PC collaborative Project related-channel changes. + +use std::collections::BTreeSet; +use std::sync::Arc; + +use buzz_core::event::StoredEvent; +use buzz_core::tenant::TenantContext; +use buzz_db::project_state::{ProjectChangeApplyResult, ProjectRelatedChannelChange}; +use nostr::Event; +use serde::Deserialize; +use uuid::Uuid; + +use crate::state::AppState; + +use super::event::dispatch_persistent_event; +use super::ingest::{IngestError, IngestResult}; + +const MAX_PATCH_CHANNELS: usize = 64; + +#[derive(Debug)] +struct ParsedChange { + owner: Vec, + d_tag: String, + expected_revision: i64, + add: Vec, + remove: Vec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ChangeContent { + v: u8, + patch: ChangePatch, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ChangePatch { + related_channels: RelatedChannelsPatch, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RelatedChannelsPatch { + add: Vec, + remove: Vec, +} + +fn invalid(message: impl Into) -> IngestError { + IngestError::Rejected(format!("invalid: {}", message.into())) +} + +fn parse_revision(value: &str) -> Result { + if value.is_empty() + || value.starts_with('0') + || !value.bytes().all(|byte| byte.is_ascii_digit()) + { + return Err(invalid( + "expected-revision must be a canonical positive integer", + )); + } + value + .parse::() + .ok() + .filter(|revision| *revision > 0) + .ok_or_else(|| invalid("expected-revision is out of range")) +} + +fn parse_coordinate(value: &str) -> Result<(Vec, String), IngestError> { + let mut parts = value.splitn(3, ':'); + let (Some("30621"), Some(owner_hex), Some(d_tag)) = (parts.next(), parts.next(), parts.next()) + else { + return Err(invalid("a tag must contain a canonical Project coordinate")); + }; + if owner_hex.len() != 64 + || !owner_hex + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(invalid( + "Project owner must be 64 lowercase hexadecimal characters", + )); + } + if d_tag.is_empty() || d_tag.len() > buzz_db::event::D_TAG_MAX_LEN { + return Err(invalid("Project d tag is empty or too long")); + } + let owner = hex::decode(owner_hex).map_err(|_| invalid("invalid Project owner"))?; + Ok((owner, d_tag.to_owned())) +} + +fn parse_channels(values: Vec) -> Result, IngestError> { + if values.len() > MAX_PATCH_CHANNELS { + return Err(invalid("related-channel patch exceeds 64 entries")); + } + values + .into_iter() + .map(|value| { + Uuid::parse_str(&value) + .ok() + .filter(|channel| channel.to_string() == value) + .ok_or_else(|| invalid("related channels must be canonical UUIDs")) + }) + .collect() +} + +fn parse(event: &Event) -> Result { + if event.tags.len() != 2 { + return Err(invalid( + "Project change must contain exactly an a tag and expected-revision tag", + )); + } + let mut coordinate = None; + let mut revision = None; + for tag in event.tags.iter() { + match tag.as_slice() { + [name, value] if name == "a" && coordinate.is_none() => { + coordinate = Some(parse_coordinate(value)?); + } + [name, value] if name == "expected-revision" && revision.is_none() => { + revision = Some(parse_revision(value)?); + } + _ => return Err(invalid("Project change tags are malformed")), + } + } + let (owner, d_tag) = coordinate.ok_or_else(|| invalid("missing a tag"))?; + let expected_revision = revision.ok_or_else(|| invalid("missing expected-revision tag"))?; + let content: ChangeContent = + serde_json::from_str(&event.content).map_err(|error| invalid(error.to_string()))?; + if content.v != 1 { + return Err(invalid("unsupported Project change version")); + } + let add = parse_channels(content.patch.related_channels.add)?; + let remove = parse_channels(content.patch.related_channels.remove)?; + if add.is_empty() && remove.is_empty() { + return Err(invalid("Project change must not be empty")); + } + let add_set = add.iter().copied().collect::>(); + let remove_set = remove.iter().copied().collect::>(); + if add_set.len() != add.len() || remove_set.len() != remove.len() { + return Err(invalid("Project change contains a duplicate channel")); + } + if !add_set.is_disjoint(&remove_set) { + return Err(invalid("Project change adds and removes the same channel")); + } + Ok(ParsedChange { + owner, + d_tag, + expected_revision, + add, + remove, + }) +} + +pub(crate) async fn handle( + tenant: &TenantContext, + state: &Arc, + event: &Event, +) -> Result { + let parsed = parse(event)?; + let outcome = state + .db + .apply_project_related_channel_change( + tenant.community(), + event, + ProjectRelatedChannelChange { + project_owner: &parsed.owner, + project_d_tag: &parsed.d_tag, + expected_revision: parsed.expected_revision, + add: &parsed.add, + remove: &parsed.remove, + }, + ) + .await + .map_err(|error| IngestError::Internal(format!("error: apply Project change: {error}")))?; + + let message = match outcome { + ProjectChangeApplyResult::Applied { revision } => { + dispatch_persistent_event( + tenant, + state, + &StoredEvent::new(event.clone(), None), + buzz_core::kind::KIND_PROJECT_CHANGE, + &event.pubkey.to_hex(), + None, + ) + .await; + format!("revision: {revision}") + } + ProjectChangeApplyResult::Duplicate { applied_revision } => { + format!("duplicate: already applied at revision {applied_revision}") + } + ProjectChangeApplyResult::ProjectNotFound => return Err(invalid("Project not found")), + ProjectChangeApplyResult::ProjectDeleted => { + return Err(IngestError::Rejected("conflict: Project is deleted".into())) + } + ProjectChangeApplyResult::Forbidden => { + return Err(IngestError::Rejected( + "restricted: actor cannot manage this Project".into(), + )) + } + ProjectChangeApplyResult::Conflict { current_revision } => { + return Err(IngestError::Rejected(format!( + "conflict: Project revision is {current_revision}" + ))) + } + ProjectChangeApplyResult::Invalid(message) => return Err(invalid(message)), + }; + + if let Err(error) = super::project_state_projection::publish_project_state_for_coordinate( + tenant, + state, + &parsed.owner, + &parsed.d_tag, + ) + .await + { + tracing::warn!( + project_owner = %hex::encode(&parsed.owner), + project_d_tag = %parsed.d_tag, + %error, + "accepted Project change awaits projection repair" + ); + } + Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message, + }) +} + +#[cfg(test)] +mod tests { + use nostr::{EventBuilder, Keys, Kind, Tag}; + + use super::*; + + fn command(tags: Vec, content: &str) -> Event { + EventBuilder::new(Kind::Custom(47_010), content) + .tags(tags) + .sign_with_keys(&Keys::generate()) + .expect("sign test command") + } + + fn valid_tags() -> Vec { + vec![ + Tag::parse(["a", &format!("30621:{}:project:x", "a".repeat(64))]) + .expect("coordinate tag"), + Tag::parse(["expected-revision", "1"]).expect("revision tag"), + ] + } + + fn valid_content() -> &'static str { + r#"{"v":1,"patch":{"related_channels":{"add":["11111111-1111-4111-8111-111111111111"],"remove":[]}}}"# + } + + #[test] + fn parses_only_the_exact_v1_envelope() { + let parsed = parse(&command(valid_tags(), valid_content())).expect("valid command"); + assert_eq!(parsed.expected_revision, 1); + assert_eq!(parsed.d_tag, "project:x"); + assert_eq!(parsed.add.len(), 1); + + let mut extra = valid_tags(); + extra.push( + Tag::parse(["auth", &"a".repeat(64), "kind=47010", &"b".repeat(128)]) + .expect("auth tag"), + ); + assert!(parse(&command(extra, valid_content())).is_err()); + let mut bad_revision = valid_tags(); + bad_revision[1] = Tag::parse(["expected-revision", "01"]).expect("tag"); + assert!(parse(&command(bad_revision, valid_content())).is_err()); + assert!(parse(&command( + valid_tags(), + r#"{"v":1,"patch":{"related_channels":{"add":[],"remove":[],"future":true}}}"#, + )) + .is_err()); + } +} diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 71c0f1e73d..0adae468bd 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -12,7 +12,7 @@ use buzz_core::{ KIND_GIT_STATUS_OPEN, KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, KIND_PRESENCE_UPDATE, KIND_PROJECT, - KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, + KIND_PROJECT_CHANGE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, }, observer::{ content_looks_like_nip44, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG, @@ -1993,6 +1993,8 @@ pub const PROJECT_CHANNEL_MAX: usize = 256; pub const PROJECT_VISIBILITY_MAX: usize = 256; /// Maximum number of `a` member tags per project event (checked before dedup). pub const PROJECT_MEMBER_CAP: usize = 64; +/// Maximum number of related channels on either side of one Project change. +pub const PROJECT_RELATED_CHANNEL_CHANGE_CAP: usize = 64; /// A validated NIP-MP member `a`-tag coordinate with an optional relay hint. /// @@ -2005,6 +2007,23 @@ pub struct ProjectMemberCoord { pub hint: Option, } +#[derive(serde::Serialize)] +struct ProjectChangeContent<'a> { + v: u8, + patch: ProjectChangePatch<'a>, +} + +#[derive(serde::Serialize)] +struct ProjectChangePatch<'a> { + related_channels: ProjectRelatedChannelPatch<'a>, +} + +#[derive(serde::Serialize)] +struct ProjectRelatedChannelPatch<'a> { + add: &'a [String], + remove: &'a [String], +} + impl PartialEq for ProjectMemberCoord { fn eq(&self, other: &Self) -> bool { self.coord == other.coord @@ -2300,6 +2319,91 @@ pub fn build_project( build_project_with_tags("", tags) } +fn validate_project_coordinate(coordinate: &str) -> Result<(), SdkError> { + let mut parts = coordinate.splitn(3, ':'); + let kind = parts.next().unwrap_or_default(); + let owner = parts.next().unwrap_or_default(); + let d = parts.next().unwrap_or_default(); + if kind != "30621" + || owner.len() != 64 + || !owner + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + || d.is_empty() + || d.len() > PROJECT_D_MAX_LEN + { + return Err(SdkError::InvalidInput( + "project coordinate must be canonical '30621::'".into(), + )); + } + Ok(()) +} + +/// Build a version-1 Project related-channel change command. +/// +/// The relay authorizes the actor and applies `expected_revision` as an +/// optimistic compare-and-swap against authoritative Project state. +pub fn build_project_related_channel_change( + project_coordinate: &str, + expected_revision: u64, + add: &[Uuid], + remove: &[Uuid], +) -> Result { + validate_project_coordinate(project_coordinate)?; + if expected_revision == 0 || expected_revision > i64::MAX as u64 { + return Err(SdkError::InvalidInput(format!( + "expected revision must be from 1 through {}", + i64::MAX + ))); + } + if add.is_empty() && remove.is_empty() { + return Err(SdkError::InvalidInput( + "Project change patch must not be empty".into(), + )); + } + if add.len() > PROJECT_RELATED_CHANNEL_CHANGE_CAP + || remove.len() > PROJECT_RELATED_CHANNEL_CHANGE_CAP + { + return Err(SdkError::InvalidInput(format!( + "Project change may add or remove at most {PROJECT_RELATED_CHANNEL_CHANGE_CAP} related channels" + ))); + } + + let mut add: Vec = add.iter().map(Uuid::to_string).collect(); + let mut remove: Vec = remove.iter().map(Uuid::to_string).collect(); + add.sort_unstable(); + remove.sort_unstable(); + let add_set: std::collections::HashSet<&str> = add.iter().map(String::as_str).collect(); + let remove_set: std::collections::HashSet<&str> = remove.iter().map(String::as_str).collect(); + if add_set.len() != add.len() || remove_set.len() != remove.len() { + return Err(SdkError::InvalidInput( + "Project change contains a duplicate related channel".into(), + )); + } + if add_set.iter().any(|channel| remove_set.contains(channel)) { + return Err(SdkError::InvalidInput( + "Project change cannot add and remove the same related channel".into(), + )); + } + let content = serde_json::to_string(&ProjectChangeContent { + v: 1, + patch: ProjectChangePatch { + related_channels: ProjectRelatedChannelPatch { + add: &add, + remove: &remove, + }, + }, + }) + .map_err(|error| SdkError::InvalidInput(format!("invalid Project change: {error}")))?; + let expected_revision = expected_revision.to_string(); + Ok( + EventBuilder::new(Kind::Custom(KIND_PROJECT_CHANGE as u16), content).tags([ + tag(&["a", project_coordinate])?, + tag(&["expected-revision", &expected_revision])?, + ]), + ) +} + /// **Generic NIP-09 coordinate delete**: Build a kind:5 deletion event with /// a single `a`-tag addressing `::`. /// @@ -4782,6 +4886,25 @@ mod tests { ); } + #[test] + fn project_related_channel_change_has_canonical_v1_shape() { + let coordinate = + "30621:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:platform"; + let related = Uuid::from_u128(0x22222222_2222_4222_8222_222222222222); + let event = + sign(build_project_related_channel_change(coordinate, 7, &[related], &[]).unwrap()); + assert_eq!(event.kind.as_u16() as u32, KIND_PROJECT_CHANGE); + assert_eq!( + event.content, + r#"{"v":1,"patch":{"related_channels":{"add":["22222222-2222-4222-8222-222222222222"],"remove":[]}}}"# + ); + assert_eq!(event.tags.len(), 2); + assert!(build_project_related_channel_change(coordinate, 1, &[], &[]).is_err()); + assert!( + build_project_related_channel_change(coordinate, 1, &[related], &[related]).is_err() + ); + } + // ── NIP-MP conformance fixtures ────────────────────────────────────────── // `build_project_with_tags` directly. Accept cases must build; reject // cases must fail with an error message containing the expected rule name. diff --git a/docs/nips/NIP-PC.md b/docs/nips/NIP-PC.md index 3694ff5623..d26095c42b 100644 --- a/docs/nips/NIP-PC.md +++ b/docs/nips/NIP-PC.md @@ -1,18 +1,42 @@ NIP-PC ====== -Relay-Authoritative Project State ---------------------------------- +Collaborative Project State +--------------------------- `draft` `optional` `relay` -**Depends on**: NIP-01 (basic event format and addressable events), NIP-09 (event deletion), and [NIP-MP](NIP-MP.md) (owner-signed Project identity) +**Depends on**: NIP-01 (basic event format and addressable events), NIP-09 (event deletion), NIP-29 (home-channel roles), and [NIP-MP](NIP-MP.md) (owner-signed Project identity) ## Abstract -This NIP defines a relay-authoritative read model for NIP-MP Projects. The owner-signed `kind:30621` remains the portable Project identity. A relay materializes accepted identity replacements and deletions into relational state, then publishes that state as a relay-signed addressable `kind:30623` Project State event. +This NIP defines collaborative related-channel membership and a relay-authoritative read model for NIP-MP Projects. The owner-signed `kind:30621` remains the portable Project identity. An authorized actor submits a transactional `kind:47010` change, and the relay publishes effective state as a relay-signed addressable `kind:30623` Project State event. -The projection gives clients one signed event to read, while the relational row remains authoritative if publication or fan-out fails. This version does not widen the owner-only replacement authority of `kind:30621`. +The projection gives clients one signed event to read, while the relational row remains authoritative if publication or fan-out fails. This version does not widen the owner-only replacement authority of `kind:30621`; collaborators mutate only related-channel membership through `kind:47010`. + +## Project Change Command + +`kind:47010` is a global command requiring `repos:write` and a global authentication token: + +```jsonc +{ + "kind": 47010, + "pubkey": "", + "tags": [ + ["a", "30621::"], + ["expected-revision", "7"] + ], + "content": "{\"v\":1,\"patch\":{\"related_channels\":{\"add\":[\"11111111-1111-4111-8111-111111111111\"],\"remove\":[]}}}" +} +``` + +The command contains exactly those two tags. `expected-revision` is a canonical integer in `1..=9223372036854775807`. The version-1 JSON shape is strict: unknown fields are rejected; `add` and `remove` are required, non-overlapping sets of canonical UUIDs; both may contain at most 64 entries; and the patch must not be empty. Adding the home channel or an already-related channel, removing an absent channel, or producing more than 64 related channels is invalid. + +The Project owner is always authorized. A different signer is authorized only while they are an active `owner` or `admin` of the Project's current, active home channel. Removed members and archived or deleted channels grant no authority. Projects without a home channel therefore remain owner-managed. Delegation and channel `created_by` do not confer Project authority. + +The relay serializes each command by Project coordinate, checks the expected revision, and atomically stores the immutable command event, applies the complete patch, and advances the revision. Two distinct commands against one revision cannot both commit. + +Replaying the exact accepted event is idempotent, including when its event row has been soft-deleted. This guarantee is bounded by the relay's ordinary event timestamp-drift window: after that window a client re-reads `kind:30623` and signs a fresh command against the current revision. ## Project Coordinate @@ -28,13 +52,13 @@ The coordinate selects state only inside the host-bound community. A relay MUST ## Authoritative State -The authoritative row is keyed by `(community, Project owner, Project d)` and carries a monotonic signed-64-bit revision, deletion state, the current owner identity event id, the last lifecycle event id, and the effective Project document. +The authoritative row is keyed by `(community, Project owner, Project d)` and carries a monotonic signed-64-bit revision, deletion state, the current owner identity event id, the last change or lifecycle event id, and the effective Project document. -The first accepted owner-signed `kind:30621` materializes revision `1`. Each accepted newer owner identity, deletion, or recreation increments the revision. Revisions never reset, including after deletion and recreation. A duplicate or superseded event does not advance the revision, and overflow rejects the lifecycle mutation. +The first accepted owner-signed `kind:30621` materializes revision `1`. Each accepted Project change, newer owner identity, deletion, or recreation increments the revision. Revisions never reset, including after deletion and recreation. A duplicate or superseded event does not advance the revision, and overflow rejects the mutation. -An accepted newer owner-signed `kind:30621` is a full recovery snapshot. It replaces the effective NIP-MP fields and extension tags with those carried by that owner event. An accepted owner-authorized NIP-09 deletion advances the row to a deleted tombstone. It does not delete member repositories or referenced channels. A later valid owner-signed `kind:30621` recreates the Project at the next revision. +An accepted newer owner-signed `kind:30621` replaces owner-controlled NIP-MP fields and extension tags while preserving collaboratively added related channels. Related channels present in the new identity are unioned with relational membership; the new home channel is excluded from that union, and a result over 64 channels is rejected. An accepted owner-authorized NIP-09 deletion advances the row to a deleted tombstone. It does not delete member repositories or referenced channels. A later valid owner-signed `kind:30621` recreates the Project at the next revision. -Identity replacement or deletion and its relational lifecycle update MUST commit atomically. +Commands, identity replacement, and deletion each commit atomically with their relational update. ## Project State Event @@ -49,7 +73,7 @@ Identity replacement or deletion and its relational lifecycle update MUST commit ["a", "30621::"], ["rev", "8"], ["e", "", "", "identity"], - ["e", "", "", "change"] + ["e", "", "", "change"] ], "content": "{\"v\":1,\"deleted\":false,\"project_tags\":[[\"d\",\"\"]]}" } @@ -57,7 +81,7 @@ Identity replacement or deletion and its relational lifecycle update MUST commit The address `d` is the lowercase SHA-256 hex digest of the UTF-8 Project coordinate. Hashing keeps the projection key fixed at 64 bytes even when the Project's own `d` approaches its 1024-byte bound. The `a` tag carries the unhashed coordinate. The `rev` tag is a canonical base-10 integer in `1..=9223372036854775807`: digits only, no sign, and no leading zero. -The `identity` event id names the current owner-signed `kind:30621`. The `change` marker names the identity or deletion event that produced the revision. On initial materialization and owner recovery, both references may name the same identity event. +The `identity` event id names the current owner-signed `kind:30621`. The `change` marker names the command, identity, or deletion event that produced the revision. On initial materialization and owner recovery, both references may name the same identity event. For a live Project, `project_tags` is the complete effective NIP-MP-compatible tag set. It includes exactly one Project-slug `d` tag. Known set-valued fields are emitted in deterministic lexical order, while preserved unknown extension tags retain their byte values and relative order. Transport-only `auth` tags are not Project metadata.