From d3008fb2f8ce70ef133a875bbe0041ce39783e3b Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 1 Sep 2026 12:42:48 -0400 Subject: [PATCH] Track Project identity lifecycle atomically Signed-off-by: John Tennant --- crates/buzz-db/src/lib.rs | 3 +- crates/buzz-db/src/store/project_state.rs | 531 +++++++++++++++++- crates/buzz-relay/src/handlers/ingest.rs | 175 +++++- .../buzz-relay/src/handlers/side_effects.rs | 8 + 4 files changed, 708 insertions(+), 9 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 718ee6ac70..4aeb978992 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::{ - ProjectChangeApplyResult, ProjectRelatedChannelChange, ProjectStateSnapshot, + ProjectChangeApplyResult, ProjectLifecycleApplyResult, ProjectLifecycleStatus, + ProjectRelatedChannelChange, ProjectStateSnapshot, }; pub use reaction::ReactionEventInsertOutcome; pub use reminder::DueReminder; diff --git a/crates/buzz-db/src/store/project_state.rs b/crates/buzz-db/src/store/project_state.rs index ea4318f5cf..fd1068b3e6 100644 --- a/crates/buzz-db/src/store/project_state.rs +++ b/crates/buzz-db/src/store/project_state.rs @@ -2,14 +2,17 @@ use std::collections::BTreeSet; -use buzz_core::kind::{event_kind_u32, KIND_PROJECT, KIND_PROJECT_CHANGE}; -use buzz_core::CommunityId; +use buzz_core::kind::{event_kind_u32, KIND_DELETION, KIND_PROJECT, KIND_PROJECT_CHANGE}; +use buzz_core::{CommunityId, StoredEvent}; +use chrono::{DateTime, Utc}; use nostr::Event; use sqlx::Row; use uuid::Uuid; use crate::event::insert_event_in_transaction; -use crate::replaceable::event_replacement_lock_key; +use crate::replaceable::{ + event_replacement_lock_key, ParameterizedReplacePrecondition, ParameterizedReplaceStatus, +}; use crate::{Db, DbError, Result}; const RELATED_CHANNEL_CAP: usize = 64; @@ -69,6 +72,41 @@ pub enum ProjectChangeApplyResult { InvalidBase(String), } +/// Result category for an owner identity or deletion lifecycle event. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ProjectLifecycleStatus { + /// The event changed authoritative Project state. + Applied, + /// The event was newly stored but had no effect on the current Project head. + NoEffect, + /// The exact event was already stored. + Duplicate, + /// A newer owner identity already dominates the submitted identity. + Superseded, +} + +/// Atomic persistence result for a Project lifecycle event. +#[derive(Clone, Debug)] +pub struct ProjectLifecycleApplyResult { + /// Stored representation used by relay dispatch when the event was inserted. + pub event: StoredEvent, + /// Whether and how authoritative Project state changed. + pub status: ProjectLifecycleStatus, + /// Canonical state after an applied lifecycle mutation. + pub snapshot: Option, +} + +impl ProjectLifecycleApplyResult { + /// Whether this call newly persisted the submitted event. + #[must_use] + pub fn was_inserted(&self) -> bool { + matches!( + self.status, + ProjectLifecycleStatus::Applied | ProjectLifecycleStatus::NoEffect + ) + } +} + fn tag_parts(tag: &serde_json::Value) -> Option> { tag.as_array()? .iter() @@ -143,7 +181,341 @@ fn validate_patch(change: ProjectRelatedChannelChange<'_>) -> Option { None } +async fn replace_related_channels( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + owner: &[u8], + d_tag: &str, + related: &BTreeSet, +) -> Result<()> { + sqlx::query( + "DELETE FROM project_related_channels WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag=$3", + ) + .bind(community_id.as_uuid()) + .bind(owner) + .bind(d_tag) + .execute(&mut **tx) + .await?; + for channel in related { + sqlx::query( + "INSERT INTO project_related_channels \ + (community_id, project_owner, project_d_tag, channel_id) VALUES ($1,$2,$3,$4)", + ) + .bind(community_id.as_uuid()) + .bind(owner) + .bind(d_tag) + .bind(channel) + .execute(&mut **tx) + .await?; + } + Ok(()) +} + impl Db { + /// Atomically accept an owner-signed Project identity and materialize it. + /// + /// A newer identity is a full recovery snapshot. Existing relational + /// revisions advance rather than resetting, including recreation after a + /// deletion. Duplicate and superseded identities leave state unchanged. + pub async fn apply_project_identity_event( + &self, + community_id: CommunityId, + event: &Event, + ) -> Result { + if event_kind_u32(event) != KIND_PROJECT { + return Err(DbError::InvalidData( + "Project identity persistence requires kind 30621".into(), + )); + } + let d_tag = crate::event::extract_d_tag(event).unwrap_or_default(); + if d_tag.is_empty() || d_tag.len() > crate::event::D_TAG_MAX_LEN { + 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 owner = event.pubkey.to_bytes(); + + 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, + owner.as_slice(), + Some(d_tag.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(coordinate_lock) + .execute(&mut *tx) + .await?; + let current_head = sqlx::query( + "SELECT revision, deleted, last_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(owner.as_slice()) + .bind(&d_tag) + .fetch_optional(&mut *tx) + .await?; + if let Some(head) = current_head + .as_ref() + .filter(|head| head.get::("deleted")) + { + let tombstone_event_id: Vec = head.try_get("last_event_id")?; + let tombstone_created_at: DateTime = + sqlx::query_scalar("SELECT created_at FROM events WHERE community_id=$1 AND id=$2") + .bind(community_id.as_uuid()) + .bind(tombstone_event_id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + DbError::InvalidData( + "Project tombstone event is missing from history".into(), + ) + })?; + let identity_created_at = + DateTime::::from_timestamp(event.created_at.as_secs() as i64, 0) + .ok_or_else(|| DbError::InvalidTimestamp(event.created_at.as_secs() as i64))?; + // A tombstone dominates every identity in its second. Recreating a + // Project requires an unambiguously later owner event. + if identity_created_at <= tombstone_created_at { + tx.rollback().await?; + return Ok(ProjectLifecycleApplyResult { + event: StoredEvent::new(event.clone(), None), + status: ProjectLifecycleStatus::Superseded, + snapshot: None, + }); + } + } + let had_unmaterialized_identity: bool = if current_head.is_none() { + sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND kind=$2 \ + AND pubkey=$3 AND d_tag=$4 AND deleted_at IS NULL)", + ) + .bind(community_id.as_uuid()) + .bind(KIND_PROJECT as i32) + .bind(owner.as_slice()) + .bind(&d_tag) + .fetch_one(&mut *tx) + .await? + } else { + false + }; + let persisted = self + .replace_parameterized_event_in_transaction( + &mut tx, + community_id, + event, + &d_tag, + None, + ParameterizedReplacePrecondition::Unconditional, + ) + .await?; + if persisted.status != ParameterizedReplaceStatus::Inserted { + tx.rollback().await?; + let status = match persisted.status { + ParameterizedReplaceStatus::Duplicate => ProjectLifecycleStatus::Duplicate, + _ => ProjectLifecycleStatus::Superseded, + }; + return Ok(ProjectLifecycleApplyResult { + event: persisted.event, + status, + snapshot: None, + }); + } + + let revision = match current_head { + None if had_unmaterialized_identity => 2, + None => 1, + Some(head) => head + .try_get::("revision")? + .checked_add(1) + .ok_or_else(|| DbError::InvalidData("Project revision overflow".into()))?, + }; + 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,$4,FALSE,$5,$5) \ + ON CONFLICT (community_id, project_owner, project_d_tag) DO UPDATE SET \ + revision=EXCLUDED.revision, deleted=FALSE, identity_event_id=EXCLUDED.identity_event_id, \ + last_event_id=EXCLUDED.last_event_id, updated_at=transaction_timestamp()", + ) + .bind(community_id.as_uuid()) + .bind(owner.as_slice()) + .bind(&d_tag) + .bind(revision) + .bind(event.id.as_bytes().as_slice()) + .execute(&mut *tx) + .await?; + replace_related_channels(&mut tx, community_id, owner.as_slice(), &d_tag, &related).await?; + + let snapshot = ProjectStateSnapshot { + revision, + identity_event_id: event.id.as_bytes().to_vec(), + change_event_id: event.id.as_bytes().to_vec(), + related_channels: related.into_iter().collect(), + }; + tx.commit().await?; + Ok(ProjectLifecycleApplyResult { + event: persisted.event, + status: ProjectLifecycleStatus::Applied, + snapshot: Some(snapshot), + }) + } + + /// Atomically store an owner-authorized NIP-09 coordinate deletion and, + /// when it covers the live identity, advance Project state to a tombstone. + pub async fn apply_project_deletion_event( + &self, + community_id: CommunityId, + event: &Event, + project_owner: &[u8], + project_d_tag: &str, + expected_identity_event_id: Option<&[u8]>, + ) -> Result { + if event_kind_u32(event) != KIND_DELETION + || project_owner.len() != 32 + || project_d_tag.is_empty() + || project_d_tag.len() > crate::event::D_TAG_MAX_LEN + || expected_identity_event_id.is_some_and(|event_id| event_id.len() != 32) + { + return Err(DbError::InvalidData("invalid Project deletion".into())); + } + let deletion_created_at = + DateTime::::from_timestamp(event.created_at.as_secs() as i64, 0) + .ok_or_else(|| DbError::InvalidTimestamp(event.created_at.as_secs() as i64))?; + let mut tx = self.begin_transaction().await?; + self.deletion_store() + .guard_transaction(&mut tx, community_id) + .await?; + let lock = event_replacement_lock_key( + community_id, + KIND_PROJECT as i32, + project_owner, + Some(project_d_tag.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock) + .execute(&mut *tx) + .await?; + let (stored_event, inserted) = + insert_event_in_transaction(&mut tx, community_id, event, None).await?; + if !inserted { + tx.rollback().await?; + return Ok(ProjectLifecycleApplyResult { + event: stored_event, + status: ProjectLifecycleStatus::Duplicate, + snapshot: None, + }); + } + + let live = sqlx::query( + "SELECT id, created_at 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 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(KIND_PROJECT as i32) + .bind(project_owner) + .bind(project_d_tag) + .fetch_optional(&mut *tx) + .await?; + let Some(live) = live else { + tx.commit().await?; + return Ok(ProjectLifecycleApplyResult { + event: stored_event, + status: ProjectLifecycleStatus::NoEffect, + snapshot: None, + }); + }; + let identity_event_id: Vec = live.try_get("id")?; + let identity_created_at: DateTime = live.try_get("created_at")?; + if identity_created_at > deletion_created_at + || expected_identity_event_id + .is_some_and(|expected| expected != identity_event_id.as_slice()) + { + tx.commit().await?; + return Ok(ProjectLifecycleApplyResult { + event: stored_event, + status: ProjectLifecycleStatus::NoEffect, + snapshot: None, + }); + } + + 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(project_owner) + .bind(project_d_tag) + .fetch_optional(&mut *tx) + .await?; + let revision = if let Some(head) = head { + let materialized: Vec = head.try_get("identity_event_id")?; + if head.try_get::("deleted")? || materialized != identity_event_id { + return Err(DbError::InvalidData( + "Project lifecycle state does not match live identity".into(), + )); + } + head.try_get::("revision")? + .checked_add(1) + .ok_or_else(|| DbError::InvalidData("Project revision overflow".into()))? + } else { + // Materialize the pre-existing identity at revision 1 before applying + // its first relational lifecycle event. + 2 + }; + + sqlx::query( + "UPDATE events SET deleted_at=transaction_timestamp() WHERE community_id=$1 \ + AND id=$2 AND deleted_at IS NULL AND created_at <= $3", + ) + .bind(community_id.as_uuid()) + .bind(&identity_event_id) + .bind(deletion_created_at) + .execute(&mut *tx) + .await?; + 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,$4,TRUE,$5,$6) \ + ON CONFLICT (community_id, project_owner, project_d_tag) DO UPDATE SET \ + revision=EXCLUDED.revision, deleted=TRUE, last_event_id=EXCLUDED.last_event_id, \ + updated_at=transaction_timestamp()", + ) + .bind(community_id.as_uuid()) + .bind(project_owner) + .bind(project_d_tag) + .bind(revision) + .bind(&identity_event_id) + .bind(event.id.as_bytes().as_slice()) + .execute(&mut *tx) + .await?; + replace_related_channels( + &mut tx, + community_id, + project_owner, + project_d_tag, + &BTreeSet::new(), + ) + .await?; + let snapshot = ProjectStateSnapshot { + revision, + identity_event_id, + change_event_id: event.id.as_bytes().to_vec(), + related_channels: Vec::new(), + }; + tx.commit().await?; + Ok(ProjectLifecycleApplyResult { + event: stored_event, + status: ProjectLifecycleStatus::Applied, + snapshot: Some(snapshot), + }) + } + /// Authorize and atomically apply one v1 Project related-channel command. /// /// Lock order is community deletion fence, Project coordinate, then current @@ -402,7 +774,7 @@ impl Db { #[cfg(test)] mod tests { use buzz_core::channel::{ChannelType, ChannelVisibility}; - use nostr::{EventBuilder, Keys, Kind, Tag}; + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use sqlx::postgres::PgPoolOptions; use super::*; @@ -489,17 +861,25 @@ mod tests { .expect("insert member"); } let seeded = Uuid::new_v4(); + let identity_time = Utc::now().timestamp() as u64 - 100; let base = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") .tags([ Tag::parse(["d", "shared"]).expect("d tag"), Tag::parse(["buzz-channel", &home.to_string()]).expect("home tag"), Tag::parse(["buzz-related-channel", &seeded.to_string()]).expect("related tag"), ]) + .custom_created_at(Timestamp::from(identity_time)) .sign_with_keys(&owner) .expect("sign Project"); - db.insert_event(community, &base, None) + let initial = db + .apply_project_identity_event(community, &base) .await - .expect("insert Project"); + .expect("insert Project identity"); + assert_eq!(initial.status, ProjectLifecycleStatus::Applied); + assert_eq!( + initial.snapshot.as_ref().map(|state| state.revision), + Some(1) + ); let added = Uuid::new_v4(); let first = command(&admin, &owner, "shared", 1, &[added], &[]); @@ -609,5 +989,144 @@ mod tests { current_revision: 3 } ); + + let recovered = Uuid::new_v4(); + let recovery = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags([ + Tag::parse(["d", "shared"]).expect("d tag"), + Tag::parse(["buzz-channel", &home.to_string()]).expect("home tag"), + Tag::parse(["buzz-related-channel", &recovered.to_string()]).expect("related tag"), + ]) + .custom_created_at(Timestamp::from(identity_time + 1)) + .sign_with_keys(&owner) + .expect("sign recovery"); + let recovery_result = db + .apply_project_identity_event(community, &recovery) + .await + .expect("apply recovery"); + assert_eq!(recovery_result.status, ProjectLifecycleStatus::Applied); + let recovery_state = recovery_result.snapshot.expect("recovery state"); + assert_eq!(recovery_state.revision, 4); + assert_eq!(recovery_state.related_channels, vec![recovered]); + assert_eq!( + db.apply_project_identity_event(community, &recovery) + .await + .expect("duplicate recovery") + .status, + ProjectLifecycleStatus::Duplicate + ); + + let stale_identity_deletion = EventBuilder::new(Kind::EventDeletion, "") + .tag(Tag::event(base.id)) + .custom_created_at(Timestamp::from(identity_time + 2)) + .sign_with_keys(&owner) + .expect("sign stale identity deletion"); + assert_eq!( + db.apply_project_deletion_event( + community, + &stale_identity_deletion, + owner_bytes.as_slice(), + "shared", + Some(base.id.as_bytes()), + ) + .await + .expect("ignore stale identity deletion") + .status, + ProjectLifecycleStatus::NoEffect + ); + assert!(db + .get_event_by_id(community, recovery.id.as_bytes()) + .await + .expect("load recovered identity") + .is_some()); + + let coordinate = format!("30621:{}:shared", owner.public_key().to_hex()); + let deletion = EventBuilder::new(Kind::EventDeletion, "") + .tag(Tag::parse(["a", &coordinate]).expect("coordinate tag")) + .custom_created_at(Timestamp::from(identity_time + 2)) + .sign_with_keys(&owner) + .expect("sign deletion"); + let deleted = db + .apply_project_deletion_event( + community, + &deletion, + owner_bytes.as_slice(), + "shared", + None, + ) + .await + .expect("delete Project"); + assert_eq!(deleted.status, ProjectLifecycleStatus::Applied); + assert_eq!( + deleted.snapshot.as_ref().map(|state| state.revision), + Some(5) + ); + assert_eq!( + db.apply_project_deletion_event( + community, + &deletion, + owner_bytes.as_slice(), + "shared", + None, + ) + .await + .expect("duplicate deletion") + .status, + ProjectLifecycleStatus::Duplicate + ); + + let same_second_recreation = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "stale") + .tags([Tag::parse(["d", "shared"]).expect("d tag")]) + .custom_created_at(Timestamp::from(identity_time + 2)) + .sign_with_keys(&owner) + .expect("sign same-second recreation"); + assert_eq!( + db.apply_project_identity_event(community, &same_second_recreation) + .await + .expect("reject same-second recreation") + .status, + ProjectLifecycleStatus::Superseded + ); + assert!(db + .get_event_by_id(community, same_second_recreation.id.as_bytes()) + .await + .expect("check same-second recreation") + .is_none()); + + let recreation = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags([Tag::parse(["d", "shared"]).expect("d tag")]) + .custom_created_at(Timestamp::from(identity_time + 3)) + .sign_with_keys(&owner) + .expect("sign recreation"); + let recreated = db + .apply_project_identity_event(community, &recreation) + .await + .expect("recreate Project"); + assert_eq!(recreated.status, ProjectLifecycleStatus::Applied); + assert_eq!( + recreated.snapshot.as_ref().map(|state| state.revision), + Some(6) + ); + + let live_identity_deletion = EventBuilder::new(Kind::EventDeletion, "") + .tag(Tag::event(recreation.id)) + .custom_created_at(Timestamp::from(identity_time + 4)) + .sign_with_keys(&owner) + .expect("sign live identity deletion"); + let deleted_by_id = db + .apply_project_deletion_event( + community, + &live_identity_deletion, + owner_bytes.as_slice(), + "shared", + Some(recreation.id.as_bytes()), + ) + .await + .expect("delete live identity by id"); + assert_eq!(deleted_by_id.status, ProjectLifecycleStatus::Applied); + assert_eq!( + deleted_by_id.snapshot.as_ref().map(|state| state.revision), + Some(7) + ); } } diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 79e1ec081e..a20f6d94f0 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -1102,6 +1102,104 @@ fn count_e_tags(event: &Event) -> usize { .count() } +/// Parse a Project coordinate from a NIP-09 `a` target. +/// +/// Other target kinds return `None` and continue through generic deletion. +/// Once the kind segment names a Project, malformed coordinates are rejected +/// rather than falling through to the non-atomic generic side effect. +fn project_deletion_coordinate(event: &Event) -> Result, String)>, IngestError> { + let Some(value) = event.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.first().is_some_and(|part| part == "a") && parts.len() >= 2) + .then(|| parts[1].as_str()) + }) else { + return Ok(None); + }; + let mut parts = value.splitn(3, ':'); + let Some(kind) = parts.next() else { + return Ok(None); + }; + if kind != KIND_PROJECT.to_string() { + return Ok(None); + } + let owner = parts.next().unwrap_or_default(); + let d_tag = parts.next().unwrap_or_default(); + let canonical_owner = owner.len() == 64 + && owner + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()); + let owner = canonical_owner + .then(|| hex::decode(owner).ok()) + .flatten() + .filter(|bytes| bytes.len() == 32) + .ok_or_else(|| { + IngestError::Rejected("invalid: malformed Project deletion coordinate".into()) + })?; + if d_tag.is_empty() || d_tag.len() > buzz_db::event::D_TAG_MAX_LEN { + return Err(IngestError::Rejected( + "invalid: malformed Project deletion coordinate".into(), + )); + } + Ok(Some((owner, d_tag.to_string()))) +} + +struct ProjectDeletionTarget { + owner: Vec, + d_tag: String, + expected_identity_event_id: Option>, +} + +/// Resolve either NIP-09 target form into the Project coordinate transaction. +/// +/// An `a` target deletes the coordinate. An `e` target retains regular NIP-09 +/// exact-event semantics: it only affects Project state while that identity is +/// still the live event for the coordinate. +async fn project_deletion_target( + tenant: &TenantContext, + state: &Arc, + event: &Event, +) -> Result, IngestError> { + if let Some((owner, d_tag)) = project_deletion_coordinate(event)? { + return Ok(Some(ProjectDeletionTarget { + owner, + d_tag, + expected_identity_event_id: None, + })); + } + let Some(target_hex) = event.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.first().is_some_and(|part| part == "e") && parts.len() >= 2) + .then(|| parts[1].as_str()) + }) else { + return Ok(None); + }; + let Ok(target_id) = hex::decode(target_hex) else { + return Ok(None); + }; + if target_id.len() != 32 { + return Ok(None); + } + let Some(target) = state + .db + .get_event_by_id_including_deleted(tenant.community(), &target_id) + .await + .map_err(|error| IngestError::Internal(format!("error: {error}")))? + else { + return Ok(None); + }; + if event_kind_u32(&target.event) != KIND_PROJECT { + return Ok(None); + } + let d_tag = buzz_db::event::extract_d_tag(&target.event) + .filter(|value| !value.is_empty() && value.len() <= buzz_db::event::D_TAG_MAX_LEN) + .ok_or_else(|| IngestError::Internal("error: stored Project has invalid d tag".into()))?; + Ok(Some(ProjectDeletionTarget { + owner: target.event.pubkey.to_bytes().to_vec(), + d_tag, + expected_identity_event_id: Some(target.event.id.as_bytes().to_vec()), + })) +} + /// Extract the effective author of a stored event (handles workflow-generated and /// legacy relay-signed attributed events). pub(crate) fn effective_message_author(event: &Event, relay_pubkey: &nostr::PublicKey) -> Vec { @@ -2715,6 +2813,11 @@ async fn ingest_event_inner( ))); } } + let project_deletion = if kind_u32 == KIND_DELETION { + project_deletion_target(tenant, state, &event).await? + } else { + None + }; if kind_u32 == KIND_STREAM_MESSAGE_EDIT { validate_edit_ownership(tenant.community(), &event, state) @@ -3137,7 +3240,41 @@ async fn ingest_event_inner( }); } - let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) { + let project_lifecycle = if kind_u32 == KIND_PROJECT { + Some( + state + .db + .apply_project_identity_event(tenant.community(), &event) + .await + .map_err(|error| match error { + buzz_db::DbError::InvalidData(message) => { + IngestError::Rejected(format!("invalid: {message}")) + } + other => IngestError::Internal(format!("error: {other}")), + })?, + ) + } else if let Some(target) = project_deletion.as_ref() { + Some( + state + .db + .apply_project_deletion_event( + tenant.community(), + &event, + &target.owner, + &target.d_tag, + target.expected_identity_event_id.as_deref(), + ) + .await + .map_err(|error| IngestError::Internal(format!("error: {error}")))?, + ) + } else { + None + }; + let project_lifecycle_handled = project_lifecycle.is_some(); + let (stored_event, was_inserted) = if let Some(result) = project_lifecycle { + let was_inserted = result.was_inserted(); + (result.event, was_inserted) + } else if buzz_core::kind::is_replaceable(kind_u32) { // NIP-16 replaceable event — atomic replace with stale-write protection. // channel_id is None for global kinds (0, 1, 3) due to step 5b above. state @@ -3204,7 +3341,7 @@ async fn ingest_event_inner( }); } - if crate::handlers::side_effects::is_side_effect_kind(kind_u32) { + if !project_lifecycle_handled && 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 @@ -5324,6 +5461,40 @@ mod postgres_tests { assert!(is_parameterized_replaceable(KIND_PROJECT)); } + #[test] + fn project_deletion_coordinate_is_strict_and_preserves_colons_in_d_tag() { + let owner = "ab".repeat(32); + let coordinate = format!("30621:{owner}:team:platform"); + let event = make_event_with_tags(KIND_DELETION, "", &[&["a", &coordinate]]); + assert_eq!( + project_deletion_coordinate(&event).unwrap(), + Some((hex::decode(owner).unwrap(), "team:platform".into())) + ); + } + + #[test] + fn malformed_project_deletion_never_falls_through_to_generic_side_effects() { + for coordinate in [ + "30621:abcd:project".to_string(), + format!("30621:{}:", "ab".repeat(32)), + format!("30621:{}:project", "AB".repeat(32)), + ] { + let event = make_event_with_tags(KIND_DELETION, "", &[&["a", &coordinate]]); + assert!(matches!( + project_deletion_coordinate(&event), + Err(IngestError::Rejected(message)) + if message.contains("malformed Project deletion coordinate") + )); + } + } + + #[test] + fn non_project_deletion_keeps_generic_routing() { + let coordinate = format!("30617:{}:repo", "ab".repeat(32)); + let event = make_event_with_tags(KIND_DELETION, "", &[&["a", &coordinate]]); + assert_eq!(project_deletion_coordinate(&event).unwrap(), None); + } + /// Drive every case in the shared NIP-MP fixture file against /// `validate_project_envelope`. All 11 accept cases must pass; all 20 /// reject cases must return an error whose rule is in the case's allowed diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index d3416d673c..06603a636b 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2185,6 +2185,14 @@ async fn handle_a_tag_deletion( let actor_bytes = effective_message_author(event, &state.relay_keypair.public_key()); match kind_num { + // Project deletion is handled atomically with its relational state in + // ingest. Reaching this generic post-storage side effect would split + // the event acceptance from the authoritative tombstone. + buzz_core::kind::KIND_PROJECT => { + return Err(anyhow::anyhow!( + "Project deletion bypassed the atomic lifecycle handler" + )); + } // kind:30350 revocation is exclusively a higher-generation inactive replacement. super::push_lease::KIND_PUSH_LEASE => { tracing::debug!(d_tag, "NIP-09 deletion ignored for push lease");