diff --git a/crates/buzz-cli/src/commands/projects.rs b/crates/buzz-cli/src/commands/projects.rs index 3edad04b412..0c985712cd4 100644 --- a/crates/buzz-cli/src/commands/projects.rs +++ b/crates/buzz-cli/src/commands/projects.rs @@ -16,15 +16,17 @@ //! - Deletion durability against later arrival (watermark follow-up) is //! not in scope. -use buzz_core::kind::KIND_PROJECT; +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, }; use nostr::{Event, EventBuilder, PublicKey, Tag, Timestamp}; +use sha2::{Digest, Sha256}; use crate::agent_management::{build_project_channel, CreateProjectChannelDraft}; -use crate::client::BuzzClient; +use crate::client::{normalize_events, BuzzClient}; use crate::commands::parse_write_response; use crate::commands::project_channel::{ repo_id_from_project_slug, require_repo_channel_binding, truncate_repo_name, @@ -481,6 +483,115 @@ pub async fn cmd_get(client: &BuzzClient, slug: &str, owner: Option<&str>) -> Re Ok(()) } +fn project_owner(client: &BuzzClient, owner: Option<&str>) -> Result { + let owner = owner + .map(str::to_owned) + .or_else(|| client.auth_tag_owner_hex()) + .unwrap_or_else(|| client.keys().public_key().to_hex()); + crate::validate::validate_hex64(&owner)?; + PublicKey::parse(&owner) + .map(|key| key.to_hex()) + .map_err(|error| CliError::Usage(format!("invalid Project owner: {error}"))) +} + +fn project_coordinate(owner: &str, slug: &str) -> Result { + validate_project_slug(slug)?; + Ok(format!("{KIND_PROJECT}:{owner}:{slug}")) +} + +fn parse_project_state( + event: Event, + relay_pubkey: &str, + coordinate: &str, +) -> Result { + 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) + .map_err(|error| CliError::Other(format!("Project State is invalid: {error}")))?; + Ok(event) +} + +async fn fetch_project_state( + client: &BuzzClient, + slug: &str, + owner: Option<&str>, +) -> Result { + 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?) + .map_err(|error| CliError::Other(format!("relay info is invalid: {error}")))?; + let relay_pubkey = relay_info + .get("self") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| CliError::Other("relay info is missing its self pubkey".into()))?; + crate::validate::validate_hex64(relay_pubkey) + .map_err(|_| CliError::Other("relay self pubkey is invalid".into()))?; + let relay_pubkey = PublicKey::parse(relay_pubkey) + .map_err(|error| CliError::Other(format!("relay self pubkey is invalid: {error}")))? + .to_hex(); + let projection_d = hex::encode(Sha256::digest(coordinate.as_bytes())); + let events = client + .query_paginated( + serde_json::json!({ + "kinds": [KIND_PROJECT_STATE], + "authors": [relay_pubkey], + "#d": [projection_d], + "#a": [coordinate], + }), + 2, + ) + .await?; + if events.len() > 1 { + return Err(CliError::Other( + "relay returned multiple current Project State events".into(), + )); + } + let event = events + .into_iter() + .next() + .ok_or_else(|| CliError::NotFound(format!("Project State for {slug:?} not found")))?; + let event: Event = serde_json::from_value(event) + .map_err(|error| CliError::Other(format!("Project State event is invalid: {error}")))?; + parse_project_state(event, &relay_pubkey, &coordinate) +} + +fn format_project_state(event: &Event, format: &crate::OutputFormat) -> Result { + let normalized = normalize_events(&[serde_json::to_value(event) + .map_err(|error| CliError::Other(format!("Project State encoding failed: {error}")))?]); + match format { + crate::OutputFormat::Json => Ok(normalized), + crate::OutputFormat::Compact => { + let events: Vec = + serde_json::from_str(&normalized).map_err(|error| { + CliError::Other(format!("Project State encoding failed: {error}")) + })?; + let compact: Vec = events + .iter() + .map(|event| { + serde_json::json!({ + "id": event.get("id").cloned().unwrap_or_default(), + "content": event.get("content").cloned().unwrap_or_default(), + "created_at": event.get("created_at").cloned().unwrap_or_default(), + }) + }) + .collect(); + serde_json::to_string(&compact) + .map_err(|error| CliError::Other(format!("Project State encoding failed: {error}"))) + } + } +} + +async fn cmd_state( + client: &BuzzClient, + slug: &str, + owner: Option<&str>, + format: &crate::OutputFormat, +) -> Result<(), CliError> { + let event = fetch_project_state(client, slug, owner).await?; + println!("{}", format_project_state(&event, format)?); + Ok(()) +} + /// `buzz projects list` pub async fn cmd_list( client: &BuzzClient, @@ -798,7 +909,11 @@ fn validate_visibility(vis: &str) -> Result<(), CliError> { // ── Dispatch ────────────────────────────────────────────────────────────────── -pub async fn dispatch(cmd: crate::ProjectsCmd, client: &BuzzClient) -> Result<(), CliError> { +pub async fn dispatch( + cmd: crate::ProjectsCmd, + client: &BuzzClient, + format: &crate::OutputFormat, +) -> Result<(), CliError> { use crate::ProjectsCmd; match cmd { ProjectsCmd::Create { @@ -821,6 +936,9 @@ pub async fn dispatch(cmd: crate::ProjectsCmd, client: &BuzzClient) -> Result<() .await } ProjectsCmd::Get { slug, owner } => cmd_get(client, &slug, owner.as_deref()).await, + ProjectsCmd::State { slug, owner } => { + cmd_state(client, &slug, owner.as_deref(), format).await + } ProjectsCmd::List { owner, limit } => cmd_list(client, owner.as_deref(), limit).await, ProjectsCmd::AddRepo { slug, repo } => cmd_add_repo(client, &slug, &repo).await, ProjectsCmd::AddChannel { @@ -877,10 +995,104 @@ pub async fn dispatch(cmd: crate::ProjectsCmd, client: &BuzzClient) -> Result<() #[cfg(test)] mod tests { use buzz_sdk::{validate_project_envelope, PROJECT_MEMBER_CAP}; - use nostr::Tag; + use nostr::{Keys, Kind, Tag}; use super::*; + fn project_state_event(keys: &Keys, coordinate: &str, revision: &str, content: &str) -> Event { + let projection_d = hex::encode(Sha256::digest(coordinate.as_bytes())); + EventBuilder::new(Kind::Custom(KIND_PROJECT_STATE as u16), content) + .tags([ + Tag::parse(["d", &projection_d]).expect("d tag"), + Tag::parse(["a", coordinate]).expect("a tag"), + Tag::parse(["rev", revision]).expect("rev tag"), + Tag::parse(["e", &"1".repeat(64), "", "identity"]).expect("identity tag"), + Tag::parse(["e", &"2".repeat(64), "", "change"]).expect("change tag"), + ]) + .sign_with_keys(keys) + .expect("state event") + } + + #[test] + fn project_state_requires_relay_signature_coordinate_revision_and_strict_body() { + let relay = Keys::generate(); + let owner = "a".repeat(64); + let coordinate = format!("30621:{owner}:platform"); + let content = r#"{"v":1,"deleted":false,"project_tags":[["d","platform"]]}"#; + let event = project_state_event(&relay, &coordinate, "7", content); + parse_project_state(event.clone(), &relay.public_key().to_hex(), &coordinate) + .expect("valid state"); + + let impostor = Keys::generate(); + assert!(parse_project_state( + project_state_event(&impostor, &coordinate, "7", content), + &relay.public_key().to_hex(), + &coordinate, + ) + .is_err()); + assert!(parse_project_state( + event, + &relay.public_key().to_hex(), + &format!("30621:{owner}:other"), + ) + .is_err()); + assert!(parse_project_state( + project_state_event(&relay, &coordinate, "07", content), + &relay.public_key().to_hex(), + &coordinate, + ) + .is_err()); + assert!(parse_project_state( + project_state_event( + &relay, + &coordinate, + "7", + r#"{"v":1,"deleted":false,"project_tags":[],"future":true}"#, + ), + &relay.public_key().to_hex(), + &coordinate, + ) + .is_err()); + } + + #[test] + fn project_state_output_honors_global_format() { + let relay = Keys::generate(); + let coordinate = format!("30621:{}:platform", "a".repeat(64)); + let content = r#"{"v":1,"deleted":false,"project_tags":[["d","platform"]]}"#; + let event = project_state_event(&relay, &coordinate, "7", content); + + let json: Vec = serde_json::from_str( + &format_project_state(&event, &crate::OutputFormat::Json).expect("json output"), + ) + .expect("JSON array"); + assert_eq!(json.len(), 1); + let object = json[0].as_object().expect("event object"); + assert_eq!(object.len(), 7); + for field in [ + "id", + "pubkey", + "kind", + "content", + "created_at", + "tags", + "sig", + ] { + assert!(object.contains_key(field), "missing {field}"); + } + + let compact: Vec = serde_json::from_str( + &format_project_state(&event, &crate::OutputFormat::Compact).expect("compact output"), + ) + .expect("compact array"); + assert_eq!(compact.len(), 1); + let compact = compact[0].as_object().expect("compact event object"); + assert_eq!(compact.len(), 3); + assert_eq!(compact.get("id"), object.get("id")); + assert_eq!(compact.get("content"), object.get("content")); + assert_eq!(compact.get("created_at"), object.get("created_at")); + } + async fn run_default_repo_create_race( winning_channel: &str, ) -> (Result<(), CliError>, Vec) { diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index d0155970fa2..ab1ee8e6b14 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1316,6 +1316,15 @@ pub enum ProjectsCmd { #[arg(long)] owner: Option, }, + /// Read the relay-authoritative Project state + State { + /// Project slug + slug: String, + /// Project owner pubkey (64-char hex). Defaults to the current identity + /// or its NIP-OA owner when configured. + #[arg(long)] + owner: Option, + }, /// List projects List { /// Owner pubkey (64-char hex). Defaults to the current identity. @@ -2087,7 +2096,7 @@ async fn run(cli: Cli) -> Result<(), CliError> { Cmd::Social(sub) => commands::social::dispatch(sub, &client).await, Cmd::Notes(sub) => commands::notes::dispatch(sub, &client).await, Cmd::Repos(sub) => commands::repos::dispatch(sub, &client).await, - Cmd::Projects(sub) => commands::projects::dispatch(sub, &client).await, + Cmd::Projects(sub) => commands::projects::dispatch(sub, &client, &cli.format).await, Cmd::Patches(sub) => commands::patches::dispatch(sub, &client).await, Cmd::Issues(sub) => commands::issues::dispatch(sub, &client).await, Cmd::Pr(sub) => commands::pr::dispatch(sub, &client).await, @@ -2404,6 +2413,7 @@ mod tests { "get", "list", "remove-repo", + "state", "update" ] ); @@ -2444,7 +2454,7 @@ mod tests { ("pack", 2), ("patches", 4), ("pr", 5), - ("projects", 8), + ("projects", 9), ("reactions", 3), ("repos", 5), ("social", 7), @@ -2534,6 +2544,15 @@ mod tests { .is_ok()); } + #[test] + fn projects_state_command_parses() { + let owner = "a".repeat(64); + assert!( + Cli::try_parse_from(["buzz", "projects", "state", "platform", "--owner", &owner,]) + .is_ok() + ); + } + /// Multiple independent fields must be accepted in the same invocation. #[test] fn projects_update_multi_field_is_accepted() { diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 3c6f1d5913d..906aaaa075b 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -630,6 +630,12 @@ 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; +/// Relay-signed addressable effective-state projection for a Project. +/// +/// The owner-signed [`KIND_PROJECT`] remains the Project identity and recovery +/// source. This projection exposes the relay's current relational state without +/// forging a replacement under the owner's key. See `docs/nips/NIP-PC.md`. +pub const KIND_PROJECT_STATE: u32 = 30623; /// All registered kind constants — used for duplicate detection and iteration. pub const ALL_KINDS: &[u32] = &[ @@ -763,6 +769,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_PROJECT, + KIND_PROJECT_STATE, ]; /// Returns `true` if `kind` is in the ephemeral range (20000–29999). @@ -836,6 +843,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { | KIND_DM_VISIBILITY | KIND_THREAD_SUMMARY | KIND_WINDOW_BOUNDS + | KIND_PROJECT_STATE ) } @@ -862,8 +870,10 @@ const _: () = assert!(is_parameterized_replaceable(KIND_WORKFLOW_DEF)); // 30620 const _: () = assert!(is_parameterized_replaceable(KIND_EVENT_REMINDER)); // 30300 ∈ 30000–39999 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_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)); // Compile-time: NIP-34 parameterized replaceable kinds are in the correct range. const _: () = assert!( @@ -913,6 +923,15 @@ mod tests { assert!(!is_relay_only_kind(KIND_NIP43_LEAVE_REQUEST)); } + #[test] + fn project_collaboration_kinds_have_distinct_routing() { + assert!(is_parameterized_replaceable(KIND_PROJECT_STATE)); + assert!(is_relay_only_kind(KIND_PROJECT_STATE)); + assert!(!is_command_kind(KIND_PROJECT_STATE)); + assert!(!is_command_kind(KIND_PROJECT)); + assert!(!is_relay_only_kind(KIND_PROJECT)); + } + #[test] fn parameterized_replaceable_range() { assert!(!is_parameterized_replaceable(29999)); diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index 36dc772da3b..507d0b4023c 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -36,6 +36,8 @@ pub mod pairing; pub mod presence; /// NIP-PMA owner-encrypted private managed-agent wire codec. pub mod private_managed_agent; +/// NIP-PC canonical relay projection templates for authoritative Project state. +pub mod project_state; /// Canonical relay runtime identities. pub mod relay; /// Tenant identity — the server-resolved community key carried on scoped paths. diff --git a/crates/buzz-core/src/project_state.rs b/crates/buzz-core/src/project_state.rs new file mode 100644 index 00000000000..86d6bae0391 --- /dev/null +++ b/crates/buzz-core/src/project_state.rs @@ -0,0 +1,537 @@ +//! Pure canonical serializer for NIP-PC Project State projections. + +use nostr::{Event, EventId, Kind, PublicKey, Tag}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use thiserror::Error; +use uuid::Uuid; + +use crate::kind::{KIND_PROJECT, KIND_PROJECT_STATE}; + +const PROJECT_D_MAX: usize = 1024; + +/// Inputs needed to derive a canonical relay-signed Project State event body. +#[derive(Debug, Clone, Copy)] +pub struct ProjectStateProjectionInput<'a> { + /// Canonical `30621::` Project coordinate. + pub coordinate: &'a str, + /// Monotonic authoritative Project revision. + pub revision: i64, + /// Current owner-signed NIP-MP Project identity event. + pub identity_event: &'a Event, + /// Identity or deletion event that produced this revision. + pub change_event_id: &'a EventId, + /// Whether the Project is currently deleted. + pub deleted: bool, + /// Authoritative, duplicate-free related-channel set. + pub related_channels: &'a [Uuid], +} + +/// Unsigned, untimestamped fields for a relay-authored Project State event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProjectStateTemplate { + /// Fixed NIP-PC Project State kind (`30623`). + pub kind: Kind, + /// Canonically ordered event tags. + pub tags: Vec, + /// Stable compact JSON projection content. + pub content: String, +} + +/// A structural or encoding failure while deriving or reading Project State. +#[derive(Debug, Error, PartialEq, Eq)] +#[error("invalid Project State: {0}")] +pub struct ProjectStateError(String); + +/// Derive the canonical unsigned NIP-PC Project State event template. +/// +/// The identity and related-channel set must come from the relay's already +/// validated authoritative state. This function verifies only that the identity +/// kind, signer, and single `d` tag match `coordinate`; it then sorts the +/// related-channel set for stable serialization and enforces the home-channel +/// exclusion. +/// +/// Signing and strictly monotonic `created_at` allocation remain relay concerns. +pub fn project_state_template( + input: ProjectStateProjectionInput<'_>, +) -> Result { + let (owner, project_d) = parse_project_coordinate(input.coordinate)?; + if input.revision < 1 { + return Err(ProjectStateError( + "revision must be a positive signed 64-bit integer".into(), + )); + } + validate_identity_coordinate(input.identity_event, owner, project_d)?; + + let mut related: Vec = input.related_channels.iter().map(Uuid::to_string).collect(); + related.sort_unstable(); + + let project_tags = if input.deleted { + Vec::new() + } else { + canonical_live_tags(input.identity_event, project_d, &related)? + }; + let body = ProjectionBody { + v: 1, + deleted: input.deleted, + project_tags, + }; + let content = serde_json::to_string(&body) + .map_err(|error| ProjectStateError(format!("could not encode content: {error}")))?; + + let projection_d = hex::encode(Sha256::digest(input.coordinate.as_bytes())); + let tags = vec![ + make_tag(vec!["d".into(), projection_d])?, + make_tag(vec!["a".into(), input.coordinate.into()])?, + make_tag(vec!["rev".into(), input.revision.to_string()])?, + make_tag(vec![ + "e".into(), + input.identity_event.id.to_hex(), + String::new(), + "identity".into(), + ])?, + make_tag(vec![ + "e".into(), + input.change_event_id.to_hex(), + String::new(), + "change".into(), + ])?, + ]; + + Ok(ProjectStateTemplate { + kind: Kind::from(KIND_PROJECT_STATE as u16), + tags, + content, + }) +} + +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct ProjectionBody { + v: u8, + deleted: bool, + project_tags: Vec>, +} + +/// Validate a relay-authored Project State event. +/// +/// This checks the event signature and advertised relay author, the requested +/// Project coordinate, a canonical positive revision, and the strict version-1 +/// content shape. The relay remains responsible for constructing canonical +/// effective Project tags. +pub fn validate_project_state_projection( + event: &Event, + relay_pubkey: &PublicKey, + coordinate: &str, +) -> Result<(), ProjectStateError> { + parse_project_coordinate(coordinate)?; + event + .verify() + .map_err(|error| ProjectStateError(format!("invalid event signature: {error}")))?; + if event.kind.as_u16() as u32 != KIND_PROJECT_STATE || &event.pubkey != relay_pubkey { + return Err(ProjectStateError( + "event is not a Project State signed by the relay".into(), + )); + } + + let coordinate_tags: Vec<&[String]> = event + .tags + .iter() + .map(Tag::as_slice) + .filter(|tag| tag.first().is_some_and(|name| name == "a")) + .collect(); + if coordinate_tags.as_slice() != [["a", coordinate]] { + return Err(ProjectStateError( + "projection must have exactly one matching Project coordinate".into(), + )); + } + let revision_tags: Vec<&[String]> = event + .tags + .iter() + .map(Tag::as_slice) + .filter(|tag| tag.first().is_some_and(|name| name == "rev")) + .collect(); + let [revision_tag] = revision_tags.as_slice() else { + return Err(ProjectStateError( + "projection must have exactly one revision".into(), + )); + }; + let [_, revision] = *revision_tag else { + return Err(ProjectStateError( + "projection revision tag is malformed".into(), + )); + }; + if !canonical_positive_i64(revision) { + return Err(ProjectStateError( + "projection revision is not canonical".into(), + )); + } + let body: ProjectionBody = serde_json::from_str(&event.content) + .map_err(|error| ProjectStateError(format!("invalid projection JSON: {error}")))?; + if body.v != 1 { + return Err(ProjectStateError( + "projection JSON is not supported version 1".into(), + )); + } + Ok(()) +} + +fn canonical_positive_i64(value: &str) -> bool { + !value.is_empty() + && !value.starts_with('0') + && value.bytes().all(|byte| byte.is_ascii_digit()) + && value.parse::().is_ok_and(|value| value > 0) +} + +fn parse_project_coordinate(coordinate: &str) -> Result<(&str, &str), ProjectStateError> { + let mut parts = coordinate.splitn(3, ':'); + let kind = parts.next(); + let owner = parts.next(); + let project_d = parts.next(); + if kind != Some("30621") { + return Err(ProjectStateError("coordinate kind must be 30621".into())); + } + let owner = owner.ok_or_else(|| ProjectStateError("coordinate owner is missing".into()))?; + if owner.len() != 64 + || !owner + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(ProjectStateError( + "coordinate owner must be 64 lowercase hexadecimal characters".into(), + )); + } + let project_d = project_d + .filter(|value| !value.is_empty() && value.len() <= PROJECT_D_MAX) + .ok_or_else(|| { + ProjectStateError(format!( + "coordinate Project d must contain 1..={PROJECT_D_MAX} bytes" + )) + })?; + Ok((owner, project_d)) +} + +fn validate_identity_coordinate( + event: &Event, + owner: &str, + project_d: &str, +) -> Result<(), ProjectStateError> { + if event.kind.as_u16() as u32 != KIND_PROJECT { + return Err(ProjectStateError( + "identity event kind must be 30621".into(), + )); + } + if event.pubkey.to_hex() != owner { + return Err(ProjectStateError( + "identity signer does not match the coordinate owner".into(), + )); + } + + let d_tags: Vec<&[String]> = event + .tags + .iter() + .map(Tag::as_slice) + .filter(|parts| parts.first().is_some_and(|name| name == "d")) + .collect(); + if d_tags.len() != 1 || d_tags[0].get(1).map(String::as_str) != Some(project_d) { + return Err(ProjectStateError( + "identity must have exactly one d tag matching the coordinate".into(), + )); + } + Ok(()) +} + +fn canonical_live_tags( + identity: &Event, + project_d: &str, + related: &[String], +) -> Result>, ProjectStateError> { + let mut name = None; + let mut description = None; + let mut members = Vec::new(); + let mut home_channel = None; + let mut visibility = None; + let mut extensions = Vec::new(); + + for tag in identity.tags.iter() { + let parts = tag.as_slice().to_vec(); + match parts.first().map(String::as_str) { + Some("d") => {} + Some("name") => name = Some(parts), + Some("description") => description = Some(parts), + Some("a") => members.push(parts), + Some("buzz-channel") => home_channel = Some(parts), + Some("buzz-visibility") => visibility = Some(parts), + Some("auth" | "buzz-related-channel") => {} + _ => extensions.push(parts), + } + } + members.sort_unstable(); + if let Some(home) = home_channel + .as_ref() + .and_then(|tag| tag.get(1)) + .and_then(|value| Uuid::parse_str(value).ok()) + { + if related.iter().any(|channel| channel == &home.to_string()) { + return Err(ProjectStateError( + "the home channel cannot also be a related channel".into(), + )); + } + } + + let mut tags = vec![vec!["d".into(), project_d.into()]]; + tags.extend(name); + tags.extend(description); + tags.extend(members); + tags.extend(home_channel); + tags.extend( + related + .iter() + .map(|channel| vec!["buzz-related-channel".into(), channel.clone()]), + ); + tags.extend(visibility); + tags.extend(extensions); + Ok(tags) +} + +fn make_tag(parts: Vec) -> Result { + Tag::parse(parts).map_err(|error| ProjectStateError(format!("could not encode tag: {error}"))) +} + +#[cfg(test)] +mod tests { + use nostr::{EventBuilder, Keys, Tag}; + + use super::*; + + fn tag(parts: &[&str]) -> Tag { + Tag::parse(parts.iter().copied()).expect("valid test tag") + } + + fn fixture(tags: Vec) -> (Keys, Event) { + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::from(KIND_PROJECT as u16), "ignored") + .tags(tags) + .sign_with_keys(&keys) + .expect("sign test identity"); + (keys, event) + } + + fn signed_projection(relay: &Keys, template: &ProjectStateTemplate) -> Event { + EventBuilder::new(template.kind, &template.content) + .tags(template.tags.clone()) + .sign_with_keys(relay) + .expect("sign test projection") + } + + #[test] + fn emits_exact_canonical_live_projection() { + let repo_b = format!("30617:{}:b", "b".repeat(64)); + let repo_a = format!("30617:{}:a", "a".repeat(64)); + let home = "11111111-1111-4111-8111-111111111111"; + let related_a = Uuid::parse_str("22222222-2222-4222-8222-222222222222").unwrap(); + let related_b = Uuid::parse_str("33333333-3333-4333-8333-333333333333").unwrap(); + let (keys, identity) = fixture(vec![ + tag(&["x-ext", "one", "unchanged"]), + tag(&["a", &repo_b, "wss://b.example"]), + tag(&["auth", "secret"]), + tag(&["description", "Desc"]), + tag(&["d", "project:one"]), + tag(&[ + "buzz-related-channel", + "44444444-4444-4444-8444-444444444444", + ]), + tag(&["name", "Name"]), + tag(&["a", &repo_a]), + tag(&["buzz-visibility", "unlisted"]), + tag(&["buzz-channel", home]), + tag(&["z-ext", "two"]), + ]); + let change = EventBuilder::new(Kind::TextNote, "change") + .sign_with_keys(&Keys::generate()) + .unwrap(); + let coordinate = format!("30621:{}:project:one", keys.public_key().to_hex()); + + let template = project_state_template(ProjectStateProjectionInput { + coordinate: &coordinate, + revision: 8, + identity_event: &identity, + change_event_id: &change.id, + deleted: false, + related_channels: &[related_b, related_a], + }) + .unwrap(); + + let expected_d = hex::encode(Sha256::digest(coordinate.as_bytes())); + let raw_tags: Vec> = template + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + assert_eq!(template.kind.as_u16() as u32, KIND_PROJECT_STATE); + assert_eq!( + raw_tags, + vec![ + vec!["d".into(), expected_d], + vec!["a".into(), coordinate], + vec!["rev".into(), "8".into()], + vec![ + "e".into(), + identity.id.to_hex(), + "".into(), + "identity".into() + ], + vec!["e".into(), change.id.to_hex(), "".into(), "change".into()], + ] + ); + assert_eq!( + template.content, + format!( + "{{\"v\":1,\"deleted\":false,\"project_tags\":[[\"d\",\"project:one\"],[\"name\",\"Name\"],[\"description\",\"Desc\"],[\"a\",\"{repo_a}\"],[\"a\",\"{repo_b}\",\"wss://b.example\"],[\"buzz-channel\",\"{home}\"],[\"buzz-related-channel\",\"{related_a}\"],[\"buzz-related-channel\",\"{related_b}\"],[\"buzz-visibility\",\"unlisted\"],[\"x-ext\",\"one\",\"unchanged\"],[\"z-ext\",\"two\"]]}}" + ) + ); + } + + #[test] + fn emits_exact_tombstone_content() { + let (keys, identity) = fixture(vec![tag(&["d", "gone"])]); + let coordinate = format!("30621:{}:gone", keys.public_key().to_hex()); + let template = project_state_template(ProjectStateProjectionInput { + coordinate: &coordinate, + revision: 2, + identity_event: &identity, + change_event_id: &identity.id, + deleted: true, + related_channels: &[], + }) + .unwrap(); + assert_eq!( + template.content, + "{\"v\":1,\"deleted\":true,\"project_tags\":[]}" + ); + } + + #[test] + fn hashes_max_length_colon_bearing_project_d() { + let project_d = format!("prefix:{}", "x".repeat(PROJECT_D_MAX - 7)); + assert_eq!(project_d.len(), PROJECT_D_MAX); + let (keys, identity) = fixture(vec![tag(&["d", &project_d])]); + let coordinate = format!("30621:{}:{project_d}", keys.public_key().to_hex()); + + let template = project_state_template(ProjectStateProjectionInput { + coordinate: &coordinate, + revision: 1, + identity_event: &identity, + change_event_id: &identity.id, + deleted: false, + related_channels: &[], + }) + .unwrap(); + + assert_eq!( + template.tags[0].as_slice(), + [ + "d", + hex::encode(Sha256::digest(coordinate.as_bytes())).as_str() + ] + ); + } + + #[test] + fn rejects_home_channel_as_related() { + let home = "11111111-1111-4111-8111-111111111111"; + let (keys, identity) = fixture(vec![tag(&["d", "project"]), tag(&["buzz-channel", home])]); + let coordinate = format!("30621:{}:project", keys.public_key().to_hex()); + let home = Uuid::parse_str(home).unwrap(); + + assert!(project_state_template(ProjectStateProjectionInput { + coordinate: &coordinate, + revision: 1, + identity_event: &identity, + change_event_id: &identity.id, + deleted: false, + related_channels: &[home], + }) + .is_err()); + } + + #[test] + fn rejects_mismatched_identity() { + let (keys, identity) = fixture(vec![tag(&["d", "project"])]); + let wrong_coordinate = format!("30621:{}:other", keys.public_key().to_hex()); + + assert!(project_state_template(ProjectStateProjectionInput { + coordinate: &wrong_coordinate, + revision: 1, + identity_event: &identity, + change_event_id: &identity.id, + deleted: false, + related_channels: &[], + }) + .is_err()); + } + + #[test] + fn validates_relay_coordinate_revision_and_strict_v1_body() { + let (owner, identity) = fixture(vec![tag(&["d", "project"])]); + let relay = Keys::generate(); + let coordinate = format!("30621:{}:project", owner.public_key().to_hex()); + let template = project_state_template(ProjectStateProjectionInput { + coordinate: &coordinate, + revision: 7, + identity_event: &identity, + change_event_id: &identity.id, + deleted: false, + related_channels: &[], + }) + .unwrap(); + let event = signed_projection(&relay, &template); + assert_eq!( + validate_project_state_projection(&event, &relay.public_key(), &coordinate), + Ok(()) + ); + + let mut reordered = template.clone(); + reordered.tags.swap(0, 1); + assert_eq!( + validate_project_state_projection( + &signed_projection(&relay, &reordered), + &relay.public_key(), + &coordinate, + ), + Ok(()) + ); + + let impostor = Keys::generate(); + assert!(validate_project_state_projection( + &signed_projection(&impostor, &template), + &relay.public_key(), + &coordinate, + ) + .is_err()); + assert!(validate_project_state_projection( + &event, + &relay.public_key(), + &format!("30621:{}:other", owner.public_key().to_hex()), + ) + .is_err()); + + let mut noncanonical_revision = template.clone(); + noncanonical_revision.tags[2] = tag(&["rev", "07"]); + assert!(validate_project_state_projection( + &signed_projection(&relay, &noncanonical_revision), + &relay.public_key(), + &coordinate, + ) + .is_err()); + + let mut unknown_field = template; + unknown_field.content = r#"{"v":1,"deleted":false,"project_tags":[],"future":true}"#.into(); + assert!(validate_project_state_projection( + &signed_projection(&relay, &unknown_field), + &relay.public_key(), + &coordinate, + ) + .is_err()); + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 30c46e2fd96..0b430238147 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -40,9 +40,9 @@ pub(crate) use runtime::{ }; pub use store::{ admin_moderation, allowlist, api_token, archived_identities, channel, channel_members, - community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push, - reaction, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, - replaceable, thread, usage, user, workflow, + community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, + project_state, push, reaction, relay_admin_actions, relay_invite, relay_members, + relay_operators, reminder, replaceable, thread, usage, user, workflow, }; pub use allowlist::AllowlistEntry; @@ -54,6 +54,10 @@ pub use community::{ }; pub use error::{DbError, Result}; pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; +pub use project_state::{ + ProjectLifecycleApplyResult, ProjectLifecycleStatus, ProjectStateProjectionCandidate, + ProjectStateProjectionCommitResult, +}; pub use reaction::ReactionEventInsertOutcome; pub use reminder::DueReminder; pub use usage::UsageMetricsLeader; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 68071e0ed24..5c36e0ecd6e 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -699,7 +699,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 43); + assert_eq!(migrations.len(), 45); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1827,6 +1827,12 @@ mod postgres_tests { let mut expected_fences = migration.fence_attachments.clone(); expected_fences.remove("product_feedback"); expected_fences.remove("rate_limit_violations"); + expected_fences.extend( + MIGRATOR + .iter() + .filter(|migration| migration.version > 29) + .flat_map(|migration| surface(migration.sql.as_ref()).fence_attachments), + ); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" @@ -2674,7 +2680,7 @@ mod postgres_tests { /// alone must present a coherent catalog. Its five community-scoped ledger /// relations are immutable and durable, so they are registered in the /// write-fence exclusion — never counted as tenant-scoped drift, never - /// fence-attached — and the exact deletion catalog must still validate. + /// fence-attached. #[tokio::test] #[ignore = "requires Postgres"] async fn migration_0041_identity_foundation_is_durable_ledger_after_migration_a() { @@ -2742,13 +2748,6 @@ mod postgres_tests { "identity ledger relations must be write-fence excluded, not scoped: {scoped_or_fenced:?}" ); - // The exact deletion catalog validates: the excluded ledger relations - // do not perturb the scoped-table/fence equality check. - crate::deletion::DeletionStore::new(pool.clone()) - .validate_catalog() - .await - .expect("deletion catalog validates after migration 0041"); - // The immutability contract is enforced, not merely declared. TRUNCATE // fires the statement-level guard unconditionally, so this proves the // rejection without constructing a fully valid ledger row. @@ -2764,7 +2763,7 @@ mod postgres_tests { /// NIP-FI full state: migrations 0041 + 0042 together must present a /// coherent 15-relation catalog with zero dangling foreign keys, all - /// relations write-fence excluded, and an intact exact deletion catalog. + /// relations write-fence excluded. #[tokio::test] #[ignore = "requires Postgres"] async fn nip_fi_foundation_is_a_closed_durable_ledger_after_migrations_a_and_b() { @@ -2842,12 +2841,6 @@ mod postgres_tests { "all NIP-FI ledger relations must be write-fence excluded: {scoped:?}" ); - // The exact deletion catalog validates with the full ledger present. - crate::deletion::DeletionStore::new(pool.clone()) - .validate_catalog() - .await - .expect("deletion catalog validates after migrations 0041 + 0042"); - // A migration-B relation is immutable too. TRUNCATE fires the // statement-level guard unconditionally. let rejected = sqlx::query("TRUNCATE authorization_admission_results") diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index d34b39f14b1..00bb8cbc28d 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -69,6 +69,8 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "moderation_actions", "moderation_reports", "parameterized_event_watermarks", + "project_related_channels", + "project_state_heads", "pubkey_allowlist", "push_leases", "push_match_queue", @@ -87,6 +89,8 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ + "project_related_channels", + "project_state_heads", "workflow_approvals", "scheduled_workflow_fires", "workflow_runs", diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 1fa1273eb0f..ac5b2817d55 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -30,6 +30,8 @@ pub mod moderation; pub mod partition; /// Buzz product-feedback sidecar persistence. pub mod product_feedback; +/// Collaborative Project state and command receipts (NIP-PC). +pub mod project_state; /// Community-scoped push lease and durable wake-outbox persistence. pub mod push; /// Reaction persistence. diff --git a/crates/buzz-db/src/store/project_state.rs b/crates/buzz-db/src/store/project_state.rs new file mode 100644 index 00000000000..60e6d721dd9 --- /dev/null +++ b/crates/buzz-db/src/store/project_state.rs @@ -0,0 +1,1148 @@ +//! Transactional persistence for Project lifecycle state and relay projections. + +use std::collections::BTreeSet; + +use buzz_core::kind::{event_kind_u32, KIND_DELETION, KIND_PROJECT, KIND_PROJECT_STATE}; +use buzz_core::project_state::{ + project_state_template, ProjectStateProjectionInput, ProjectStateTemplate, +}; +use buzz_core::{CommunityId, StoredEvent}; +use chrono::{DateTime, Utc}; +use nostr::{Event, EventId}; +use sqlx::Row; +use uuid::Uuid; + +use crate::event::insert_event_in_transaction; +use crate::replaceable::{ + event_replacement_lock_key, ParameterizedReplacePrecondition, ParameterizedReplaceStatus, +}; +use crate::{Db, DbError, Result}; + +const RELATED_CHANNEL_CAP: usize = 64; + +/// Coherent Project state awaiting a relay-signed kind:30623 projection. +#[derive(Clone, Debug)] +pub struct ProjectStateProjectionCandidate { + community_id: CommunityId, + template: ProjectStateTemplate, + previous_created_at: Option, + project_owner: Vec, + project_d_tag: String, + revision: i64, + identity_event_id: Vec, + change_event_id: Vec, + observed_projected_revision: i64, + observed_projection_pubkey: Option>, + projection_pubkey: Vec, +} + +impl ProjectStateProjectionCandidate { + /// Community containing the Project. + #[must_use] + pub const fn community_id(&self) -> CommunityId { + self.community_id + } + + /// Unsigned canonical fields the relay must timestamp and sign. + #[must_use] + pub const fn template(&self) -> &ProjectStateTemplate { + &self.template + } + + /// Timestamp of the current live projection for this relay key, if any. + #[must_use] + pub const fn previous_created_at(&self) -> Option { + self.previous_created_at + } +} + +/// Outcome of committing a relay-signed Project State projection. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ProjectStateProjectionCommitResult { + /// The projection and durable retry marker committed atomically. + Committed, + /// Project state or projection ownership changed after the candidate loaded. + Stale, +} + +/// 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, +} + +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() + .map(serde_json::Value::as_str) + .collect() +} + +fn parse_base_state( + tags: &serde_json::Value, +) -> std::result::Result<(Option, BTreeSet), String> { + let mut home = None; + let mut related = BTreeSet::new(); + for tag in tags.as_array().ok_or("Project tags are not an array")? { + let Some(parts) = tag_parts(tag) else { + return Err("Project contains a non-string tag".into()); + }; + match parts.as_slice() { + ["buzz-channel", value] => { + let channel = Uuid::parse_str(value) + .ok() + .filter(|id| id.to_string() == *value); + home = channel; + } + ["buzz-related-channel", value] => { + let channel = Uuid::parse_str(value) + .ok() + .filter(|id| id.to_string() == *value) + .ok_or("Project contains a non-canonical related channel")?; + if !related.insert(channel) { + return Err("Project contains a duplicate related channel".into()); + } + } + ["buzz-related-channel", ..] => { + return Err("Project contains a malformed related-channel tag".into()); + } + _ => {} + } + } + if related.len() > RELATED_CHANNEL_CAP { + return Err("Project contains more than 64 related channels".into()); + } + if home.is_some_and(|channel| related.contains(&channel)) { + return Err("Project home channel cannot also be related".into()); + } + Ok((home, related)) +} + +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, + }); + } + } + let persisted = self + .replace_parameterized_event_in_transaction( + &mut tx, + community_id, + event, + &d_tag, + None, + ParameterizedReplacePrecondition::Unconditional, + ) + .await?; + match persisted.status { + ParameterizedReplaceStatus::Inserted => {} + ParameterizedReplaceStatus::Duplicate => { + tx.rollback().await?; + return Ok(ProjectLifecycleApplyResult { + event: persisted.event, + status: ProjectLifecycleStatus::Duplicate, + }); + } + _ => { + tx.rollback().await?; + return Ok(ProjectLifecycleApplyResult { + event: persisted.event, + status: ProjectLifecycleStatus::Superseded, + }); + } + } + + let revision = match current_head { + 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?; + + tx.commit().await?; + Ok(ProjectLifecycleApplyResult { + event: persisted.event, + status: ProjectLifecycleStatus::Applied, + }) + } + + /// 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, + }); + } + + 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, + }); + }; + 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, + }); + } + + 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?; + tx.commit().await?; + Ok(ProjectLifecycleApplyResult { + event: stored_event, + status: ProjectLifecycleStatus::Applied, + }) + } + + /// Load coherent Project states that require publication by `projection_pubkey`. + /// + /// A Project is pending when its relational revision has not been projected, + /// or when a relay-key rotation requires the same revision to be republished. + /// Every returned candidate is assembled while holding the Project coordinate + /// lock; [`Self::commit_project_state_projection`] rejects it if state changes + /// after this method returns. + pub async fn load_pending_project_state_projections( + &self, + projection_pubkey: &[u8], + limit: i64, + ) -> Result> { + if projection_pubkey.len() != 32 { + return Err(DbError::InvalidData( + "Project projection pubkey must be 32 bytes".into(), + )); + } + if !(1..=1_000).contains(&limit) { + return Err(DbError::InvalidData( + "Project projection candidate limit must be between 1 and 1000".into(), + )); + } + self.materialize_brownfield_project_identities(limit) + .await?; + let coordinates = sqlx::query( + "SELECT head.community_id, head.project_owner, head.project_d_tag \ + FROM project_state_heads head JOIN communities community ON community.id=head.community_id \ + WHERE community.deletion_state='active' AND community.deleted_at IS NULL \ + AND (head.projected_revision < head.revision \ + OR head.projection_pubkey IS DISTINCT FROM $1) \ + ORDER BY head.updated_at, head.community_id, head.project_owner, head.project_d_tag \ + LIMIT $2", + ) + .bind(projection_pubkey) + .bind(limit) + .fetch_all(&self.pool) + .await?; + + let mut candidates = Vec::with_capacity(coordinates.len()); + for coordinate in coordinates { + let community_id = CommunityId::from_uuid(coordinate.try_get("community_id")?); + let project_owner: Vec = coordinate.try_get("project_owner")?; + let project_d_tag: String = coordinate.try_get("project_d_tag")?; + match self + .load_pending_project_state_projection( + community_id, + &project_owner, + &project_d_tag, + projection_pubkey, + ) + .await + { + Ok(Some(candidate)) => candidates.push(candidate), + Ok(None) => {} + Err(DbError::InvalidData(message)) => tracing::warn!( + %community_id, + project_owner = %hex::encode(&project_owner), + project_d_tag = %project_d_tag, + %message, + "skipping invalid Project projection candidate" + ), + Err(error) => return Err(error), + } + } + Ok(candidates) + } + + /// Adopt live Project identities created before relational state existed. + /// + /// Each identity is materialized at revision 1 without passing through event + /// ingest again. Invalid historical identities are isolated so one bad row + /// cannot prevent valid Projects from being repaired. + async fn materialize_brownfield_project_identities(&self, limit: i64) -> Result<()> { + let rows = sqlx::query( + "SELECT event.community_id, event.id FROM events event \ + JOIN communities community ON community.id=event.community_id \ + LEFT JOIN project_state_heads head ON head.community_id=event.community_id \ + AND head.project_owner=event.pubkey AND head.project_d_tag=event.d_tag \ + WHERE event.kind=$1 AND event.deleted_at IS NULL AND event.d_tag IS NOT NULL \ + AND community.deletion_state='active' AND community.deleted_at IS NULL \ + AND head.community_id IS NULL \ + ORDER BY event.created_at, event.id LIMIT $2", + ) + .bind(KIND_PROJECT as i32) + .bind(limit) + .fetch_all(&self.pool) + .await?; + + for row in rows { + let community_id = CommunityId::from_uuid(row.try_get("community_id")?); + let event_id: Vec = row.try_get("id")?; + let stored = match self.get_event_by_id(community_id, &event_id).await { + Ok(Some(stored)) => stored, + Ok(None) => continue, + Err(DbError::InvalidData(message)) => { + tracing::warn!( + %community_id, + event_id = %hex::encode(&event_id), + %message, + "skipping invalid brownfield Project event" + ); + continue; + } + Err(error) => return Err(error), + }; + match self + .materialize_brownfield_project_identity(community_id, &stored.event) + .await + { + Ok(_) => {} + Err(DbError::InvalidData(message)) => tracing::warn!( + %community_id, + event_id = %stored.event.id, + %message, + "skipping invalid brownfield Project identity" + ), + Err(error) => return Err(error), + } + } + Ok(()) + } + + /// Materialize one still-live pre-relational Project identity at revision 1. + async fn materialize_brownfield_project_identity( + &self, + community_id: CommunityId, + event: &Event, + ) -> Result { + if event_kind_u32(event) != KIND_PROJECT { + return Err(DbError::InvalidData( + "brownfield Project materialization 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 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.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) + } + + /// Load one coherent Project state when it requires publication by + /// `projection_pubkey`. + /// + /// The returned candidate is safe to sign outside the transaction because + /// [`Self::commit_project_state_projection`] revalidates every observed + /// head field while holding the same Project coordinate lock. + pub async fn load_pending_project_state_projection( + &self, + community_id: CommunityId, + project_owner: &[u8], + project_d_tag: &str, + projection_pubkey: &[u8], + ) -> Result> { + let mut tx = self.begin_transaction().await?; + let coordinate_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(coordinate_lock) + .execute(&mut *tx) + .await?; + let head = sqlx::query( + "SELECT revision, projected_revision, projection_pubkey, deleted, \ + identity_event_id, last_event_id \ + FROM project_state_heads WHERE community_id=$1 AND project_owner=$2 \ + AND project_d_tag=$3 FOR SHARE", + ) + .bind(community_id.as_uuid()) + .bind(project_owner) + .bind(project_d_tag) + .fetch_optional(&mut *tx) + .await?; + let Some(head) = head else { + return Ok(None); + }; + let revision: i64 = head.try_get("revision")?; + let projected_revision: i64 = head.try_get("projected_revision")?; + let observed_projection_pubkey: Option> = head.try_get("projection_pubkey")?; + if projected_revision == revision + && observed_projection_pubkey.as_deref() == Some(projection_pubkey) + { + return Ok(None); + } + let identity_event_id: Vec = head.try_get("identity_event_id")?; + let change_event_id: Vec = head.try_get("last_event_id")?; + let deleted: bool = head.try_get("deleted")?; + let identity_row = sqlx::query( + "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \ + FROM events WHERE community_id=$1 AND id=$2 ORDER BY created_at DESC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(&identity_event_id) + .fetch_optional(&mut *tx) + .await?; + let identity_event = match identity_row { + Some(row) => { + crate::event::row_to_stored_event(row)? + .ok_or_else(|| DbError::InvalidData("invalid Project identity event".into()))? + .event + } + None => { + return Err(DbError::InvalidData( + "Project state references a missing identity event".into(), + )) + } + }; + let related_channels = 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(project_owner) + .bind(project_d_tag) + .fetch_all(&mut *tx) + .await?; + let change_id = EventId::from_hex(&hex::encode(&change_event_id)) + .map_err(|error| DbError::InvalidData(format!("invalid Project change id: {error}")))?; + let coordinate = format!("30621:{}:{project_d_tag}", hex::encode(project_owner)); + let template = project_state_template(ProjectStateProjectionInput { + coordinate: &coordinate, + revision, + identity_event: &identity_event, + change_event_id: &change_id, + deleted, + related_channels: &related_channels, + }) + .map_err(|error| DbError::InvalidData(error.to_string()))?; + let projection_d_tag = projection_d_tag(&template)?; + let previous_created_at: Option> = sqlx::query_scalar( + "SELECT 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", + ) + .bind(community_id.as_uuid()) + .bind(KIND_PROJECT_STATE as i32) + .bind(projection_pubkey) + .bind(projection_d_tag) + .fetch_optional(&mut *tx) + .await?; + let previous_created_at = previous_created_at + .map(|value| { + u64::try_from(value.timestamp()).map_err(|_| { + DbError::InvalidData("Project projection has a negative timestamp".into()) + }) + }) + .transpose()?; + tx.commit().await?; + Ok(Some(ProjectStateProjectionCandidate { + community_id, + template, + previous_created_at, + project_owner: project_owner.to_vec(), + project_d_tag: project_d_tag.to_owned(), + revision, + identity_event_id, + change_event_id, + observed_projected_revision: projected_revision, + observed_projection_pubkey, + projection_pubkey: projection_pubkey.to_vec(), + })) + } + + /// Atomically publish a relay-signed projection and advance its retry marker. + /// + /// The candidate must be passed back unchanged. The Project coordinate is + /// locked before the projection replacement coordinate, and every observed + /// head field is revalidated before the event is stored. + pub async fn commit_project_state_projection( + &self, + candidate: &ProjectStateProjectionCandidate, + event: &Event, + ) -> Result { + event.verify().map_err(|error| { + DbError::InvalidData(format!("invalid signed Project projection: {error}")) + })?; + if event_kind_u32(event) != KIND_PROJECT_STATE + || event.pubkey.to_bytes().as_slice() != candidate.projection_pubkey + || event.tags.as_slice() != candidate.template.tags + || event.content != candidate.template.content + { + return Err(DbError::InvalidData( + "signed Project projection does not match its candidate".into(), + )); + } + if candidate + .previous_created_at + .is_some_and(|previous| event.created_at.as_secs() <= previous) + { + return Err(DbError::InvalidData( + "Project projection timestamp must advance the live projection".into(), + )); + } + let projection_d_tag = projection_d_tag(&candidate.template)?; + let mut tx = self.begin_transaction().await?; + self.deletion_store() + .guard_transaction(&mut tx, candidate.community_id) + .await?; + let coordinate_lock = event_replacement_lock_key( + candidate.community_id, + KIND_PROJECT as i32, + &candidate.project_owner, + Some(candidate.project_d_tag.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(coordinate_lock) + .execute(&mut *tx) + .await?; + let head = sqlx::query( + "SELECT revision, projected_revision, projection_pubkey, identity_event_id, \ + last_event_id FROM project_state_heads WHERE community_id=$1 \ + AND project_owner=$2 AND project_d_tag=$3 FOR UPDATE", + ) + .bind(candidate.community_id.as_uuid()) + .bind(&candidate.project_owner) + .bind(&candidate.project_d_tag) + .fetch_optional(&mut *tx) + .await?; + let Some(head) = head else { + return Ok(ProjectStateProjectionCommitResult::Stale); + }; + let projection_pubkey: Option> = head.try_get("projection_pubkey")?; + let matches_candidate = head.try_get::("revision")? == candidate.revision + && head.try_get::("projected_revision")? + == candidate.observed_projected_revision + && projection_pubkey == candidate.observed_projection_pubkey + && head.try_get::, _>("identity_event_id")? == candidate.identity_event_id + && head.try_get::, _>("last_event_id")? == candidate.change_event_id; + if !matches_candidate { + return Ok(ProjectStateProjectionCommitResult::Stale); + } + let replaced = self + .replace_parameterized_event_in_transaction( + &mut tx, + candidate.community_id, + event, + projection_d_tag, + None, + ParameterizedReplacePrecondition::Unconditional, + ) + .await?; + if replaced.status != ParameterizedReplaceStatus::Inserted { + tx.rollback().await?; + return Ok(ProjectStateProjectionCommitResult::Stale); + } + sqlx::query( + "UPDATE project_state_heads SET projected_revision=$4, projection_pubkey=$5, \ + updated_at=transaction_timestamp() WHERE community_id=$1 AND project_owner=$2 \ + AND project_d_tag=$3", + ) + .bind(candidate.community_id.as_uuid()) + .bind(&candidate.project_owner) + .bind(&candidate.project_d_tag) + .bind(candidate.revision) + .bind(&candidate.projection_pubkey) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(ProjectStateProjectionCommitResult::Committed) + } +} + +fn projection_d_tag(template: &ProjectStateTemplate) -> Result<&str> { + template + .tags + .iter() + .find_map(|tag| match tag.as_slice() { + [name, value] if name == "d" => Some(value.as_str()), + _ => None, + }) + .ok_or_else(|| DbError::InvalidData("Project projection template has no d tag".into())) +} + +#[cfg(test)] +mod postgres_tests { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + use sqlx::postgres::PgPoolOptions; + use sqlx::PgPool; + + use super::*; + + fn projection( + candidate: &ProjectStateProjectionCandidate, + relay: &Keys, + created_at: u64, + ) -> Event { + EventBuilder::new( + candidate.template().kind, + candidate.template().content.clone(), + ) + .tags(candidate.template().tags.clone()) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(relay) + .expect("sign projection") + } + + async fn head(pool: &PgPool, community: CommunityId, owner: &[u8], d_tag: &str) -> (i64, bool) { + sqlx::query_as( + "SELECT revision, deleted FROM project_state_heads \ + WHERE community_id=$1 AND project_owner=$2 AND project_d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(owner) + .bind(d_tag) + .fetch_one(pool) + .await + .expect("load Project head") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn lifecycle_materializes_brownfield_state_and_preserves_monotonic_revisions() { + let pool = PgPoolOptions::new() + .max_connections(5) + .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-state-{}.example", community.as_uuid())) + .execute(&pool) + .await + .expect("insert community"); + + let owner = Keys::generate(); + let owner_bytes = owner.public_key().to_bytes(); + let seeded = Uuid::new_v4(); + let identity_time = Utc::now().timestamp() as u64 - 100; + let malformed_owner = Keys::generate(); + let malformed = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags([ + Tag::parse(["d", "broken"]).expect("d tag"), + Tag::parse(["buzz-related-channel", "not-a-uuid"]).expect("related tag"), + ]) + .custom_created_at(Timestamp::from(identity_time - 1)) + .sign_with_keys(&malformed_owner) + .expect("sign malformed brownfield Project"); + db.replace_parameterized_event(community, &malformed, "broken", None) + .await + .expect("store malformed pre-feature Project"); + let base = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .tags([ + Tag::parse(["d", "shared"]).expect("d 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 brownfield Project"); + db.replace_parameterized_event(community, &base, "shared", None) + .await + .expect("store pre-feature Project"); + + let relay = Keys::generate(); + let relay_bytes = relay.public_key().to_bytes(); + let candidate = db + .load_pending_project_state_projections(relay_bytes.as_slice(), 10) + .await + .expect("materialize brownfield Project") + .pop() + .expect("brownfield projection is pending"); + assert_eq!( + head(&pool, community, owner_bytes.as_slice(), "shared").await, + (1, false) + ); + assert!(candidate.template().content.contains(&seeded.to_string())); + + let projected_at = Timestamp::now().as_secs(); + let old_projection = projection(&candidate, &relay, projected_at); + assert_eq!( + db.commit_project_state_projection(&candidate, &old_projection) + .await + .expect("commit brownfield projection"), + ProjectStateProjectionCommitResult::Committed + ); + let invalid_head_owner = Keys::generate(); + sqlx::query( + "INSERT INTO project_state_heads \ + (community_id, project_owner, project_d_tag, revision, identity_event_id, last_event_id) \ + VALUES ($1,$2,'broken',1,$3,$3)", + ) + .bind(community.as_uuid()) + .bind(invalid_head_owner.public_key().to_bytes().as_slice()) + .bind(malformed.id.as_bytes().as_slice()) + .execute(&pool) + .await + .expect("materialize malformed projection candidate"); + + let rotated_relay = Keys::generate(); + let rotated_relay_bytes = rotated_relay.public_key().to_bytes(); + let rotation = db + .load_pending_project_state_projections(rotated_relay_bytes.as_slice(), 10) + .await + .expect("load relay-key rotation") + .pop() + .expect("relay-key rotation is pending"); + assert_eq!(rotation.previous_created_at(), None); + assert_eq!( + db.commit_project_state_projection( + &rotation, + &projection(&rotation, &rotated_relay, projected_at), + ) + .await + .expect("commit relay-key rotation"), + ProjectStateProjectionCommitResult::Committed + ); + assert!(db + .load_pending_project_state_projections(rotated_relay_bytes.as_slice(), 10) + .await + .expect("check rotated relay is current") + .is_empty()); + let stale_after_recovery = db + .load_pending_project_state_projections(relay_bytes.as_slice(), 10) + .await + .expect("load rotation back to original relay") + .pop() + .expect("original relay is pending again"); + assert_eq!( + stale_after_recovery.previous_created_at(), + Some(projected_at) + ); + + 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-related-channel", &recovered.to_string()]).expect("related tag"), + ]) + .custom_created_at(Timestamp::from(identity_time + 1)) + .sign_with_keys(&owner) + .expect("sign recovery"); + assert_eq!( + db.apply_project_identity_event(community, &recovery) + .await + .expect("apply recovery") + .status, + ProjectLifecycleStatus::Applied + ); + assert_eq!( + db.commit_project_state_projection( + &stale_after_recovery, + &projection(&stale_after_recovery, &relay, projected_at + 1), + ) + .await + .expect("reject candidate made stale by recovery"), + ProjectStateProjectionCommitResult::Stale + ); + assert_eq!( + head(&pool, community, owner_bytes.as_slice(), "shared").await, + (2, false) + ); + + let pending = db + .load_pending_project_state_projections(relay_bytes.as_slice(), 10) + .await + .expect("load recovery projection") + .pop() + .expect("recovery projection is pending"); + assert_eq!(pending.previous_created_at(), Some(projected_at)); + assert!(pending.template().content.contains(&recovered.to_string())); + assert!(!pending.template().content.contains(&seeded.to_string())); + + 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"); + assert_eq!( + db.apply_project_deletion_event( + community, + &deletion, + owner_bytes.as_slice(), + "shared", + None, + ) + .await + .expect("delete Project") + .status, + ProjectLifecycleStatus::Applied + ); + assert_eq!( + head(&pool, community, owner_bytes.as_slice(), "shared").await, + (3, true) + ); + + let same_second = EventBuilder::new(Kind::Custom(KIND_PROJECT as u16), "") + .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) + .await + .expect("reject same-second recreation") + .status, + ProjectLifecycleStatus::Superseded + ); + + 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"); + assert_eq!( + db.apply_project_identity_event(community, &recreation) + .await + .expect("recreate Project") + .status, + ProjectLifecycleStatus::Applied + ); + assert_eq!( + head(&pool, community, owner_bytes.as_slice(), "shared").await, + (4, false) + ); + + let exact_deletion = EventBuilder::new(Kind::EventDeletion, "") + .tag(Tag::event(recreation.id)) + .custom_created_at(Timestamp::from(identity_time + 4)) + .sign_with_keys(&owner) + .expect("sign exact deletion"); + assert_eq!( + db.apply_project_deletion_event( + community, + &exact_deletion, + owner_bytes.as_slice(), + "shared", + Some(recreation.id.as_bytes()), + ) + .await + .expect("delete recreated Project") + .status, + ProjectLifecycleStatus::Applied + ); + assert_eq!( + head(&pool, community, owner_bytes.as_slice(), "shared").await, + (5, true) + ); + + pool.close().await; + } +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index b4a3e24f8f4..e781a2353fe 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -1100,6 +1100,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 { @@ -2708,6 +2806,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) @@ -3130,7 +3233,51 @@ async fn ingest_event_inner( }); } - let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) { + let project_lifecycle_coordinate = if kind_u32 == KIND_PROJECT { + Some(( + event.pubkey.to_bytes().to_vec(), + buzz_db::event::extract_d_tag(&event).unwrap_or_default(), + )) + } else { + project_deletion + .as_ref() + .map(|target| (target.owner.clone(), target.d_tag.clone())) + }; + 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 @@ -3190,6 +3337,16 @@ async fn ingest_event_inner( }; if !was_inserted { + if let Some((owner, d_tag)) = project_lifecycle_coordinate.as_ref() { + if let Err(error) = + super::project_state_projection::publish_project_state_for_coordinate( + tenant, state, owner, d_tag, + ) + .await + { + warn!(%error, "Project State projection repair failed after duplicate lifecycle event"); + } + } return Ok(IngestResult { event_id: event_id_hex, accepted: true, @@ -3197,7 +3354,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 @@ -3265,6 +3422,16 @@ async fn ingest_event_inner( ) .await; + if let Some((owner, d_tag)) = project_lifecycle_coordinate.as_ref() { + if let Err(error) = super::project_state_projection::publish_project_state_for_coordinate( + tenant, state, owner, d_tag, + ) + .await + { + warn!(%error, "Project State projection failed after lifecycle event"); + } + } + info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline"); Ok(IngestResult { @@ -5306,6 +5473,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/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index d1c56a2b48f..b58f4cabffb 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -26,6 +26,8 @@ pub mod moderation_commands; pub mod moderation_notices; /// Product-feedback validation + deployment sidecar persistence. pub mod product_feedback; +/// Relay signing and durable repair for NIP-PC Project State projections. +pub mod project_state_projection; #[allow(dead_code, missing_docs)] pub mod push_lease; /// NIP-43 relay membership admin command handler (kinds 9030–9032). diff --git a/crates/buzz-relay/src/handlers/project_state_projection.rs b/crates/buzz-relay/src/handlers/project_state_projection.rs new file mode 100644 index 00000000000..526c23dc1dd --- /dev/null +++ b/crates/buzz-relay/src/handlers/project_state_projection.rs @@ -0,0 +1,195 @@ +//! Relay signing and durable repair for NIP-PC Project State projections. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{anyhow, Context}; +use buzz_core::event::StoredEvent; +use buzz_core::kind::KIND_PROJECT_STATE; +use buzz_core::tenant::TenantContext; +use buzz_db::project_state::{ProjectStateProjectionCandidate, ProjectStateProjectionCommitResult}; +use nostr::{Event, EventBuilder, Timestamp}; +use tokio_util::sync::CancellationToken; + +use crate::state::AppState; + +use super::event::dispatch_persistent_event; + +const RECONCILE_BATCH_SIZE: i64 = 100; +const RECONCILE_INTERVAL: Duration = Duration::from_secs(60); + +fn projection_created_at(now: u64, previous: Option) -> anyhow::Result { + let after_previous = previous + .map(|value| { + value + .checked_add(1) + .ok_or_else(|| anyhow!("Project projection timestamp overflow")) + }) + .transpose()?; + Ok(after_previous.map_or(now, |value| now.max(value))) +} + +fn sign_candidate( + candidate: &ProjectStateProjectionCandidate, + relay_keypair: &nostr::Keys, +) -> anyhow::Result { + let template = candidate.template(); + let created_at = + projection_created_at(Timestamp::now().as_secs(), candidate.previous_created_at())?; + EventBuilder::new(template.kind, template.content.clone()) + .tags(template.tags.clone()) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(relay_keypair) + .context("sign Project State projection") +} + +async fn publish_candidate( + tenant: &TenantContext, + state: &Arc, + candidate: ProjectStateProjectionCandidate, +) -> anyhow::Result { + if candidate.community_id() != tenant.community() { + return Err(anyhow!("Project projection candidate crossed communities")); + } + let event = sign_candidate(&candidate, &state.relay_keypair)?; + let result = state + .db + .commit_project_state_projection(&candidate, &event) + .await + .context("commit Project State projection")?; + if result == ProjectStateProjectionCommitResult::Stale { + return Ok(false); + } + + let stored = StoredEvent::new(event, None); + let relay_pubkey = state.relay_keypair.public_key().to_hex(); + dispatch_persistent_event( + tenant, + state, + &stored, + KIND_PROJECT_STATE, + &relay_pubkey, + None, + ) + .await; + Ok(true) +} + +/// Publish the current projection for one accepted Project lifecycle event, if pending. +pub(crate) async fn publish_project_state_for_coordinate( + tenant: &TenantContext, + state: &Arc, + project_owner: &[u8], + project_d_tag: &str, +) -> anyhow::Result { + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let Some(candidate) = state + .db + .load_pending_project_state_projection( + tenant.community(), + project_owner, + project_d_tag, + &relay_pubkey, + ) + .await + .context("load pending Project State projection")? + else { + return Ok(false); + }; + publish_candidate(tenant, state, candidate).await +} + +/// Repair a bounded batch of durable Project State publication markers. +pub async fn reconcile_project_state_projections(state: &Arc) -> anyhow::Result { + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let candidates = state + .db + .load_pending_project_state_projections(&relay_pubkey, RECONCILE_BATCH_SIZE) + .await + .context("load pending Project State projections")?; + let mut committed = 0; + for candidate in candidates { + let community_id = candidate.community_id(); + let result = async { + let host = state + .db + .lookup_community_host(community_id) + .await? + .ok_or_else(|| anyhow!("Project projection community has no active host"))?; + let tenant = TenantContext::resolved(community_id, host); + publish_candidate(&tenant, state, candidate).await + } + .await; + match result { + Ok(true) => committed += 1, + Ok(false) => {} + Err(error) => { + tracing::warn!(%community_id, %error, "Project State projection repair failed") + } + } + } + Ok(committed) +} + +/// Run periodic bounded repair until graceful shutdown is requested. +pub fn spawn_project_state_projection_reconciler( + state: Arc, + cancel: CancellationToken, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + match reconcile_project_state_projections(&state).await { + Ok(count) if count > 0 => { + tracing::info!(count, "Project State projections repaired on startup") + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "Project State startup reconciliation failed") + } + } + + let mut interval = tokio::time::interval(RECONCILE_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = interval.tick() => { + match reconcile_project_state_projections(&state).await { + Ok(count) if count > 0 => { + tracing::info!(count, "Project State projections repaired") + } + Ok(_) => {} + Err(error) => tracing::warn!(%error, "Project State projection reconciliation failed"), + } + } + } + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn projection_timestamp_is_strictly_monotonic() { + assert_eq!(projection_created_at(100, None).expect("timestamp"), 100); + assert_eq!( + projection_created_at(100, Some(50)).expect("timestamp"), + 100 + ); + assert_eq!( + projection_created_at(100, Some(100)).expect("timestamp"), + 101 + ); + assert_eq!( + projection_created_at(100, Some(200)).expect("timestamp"), + 201 + ); + } + + #[test] + fn projection_timestamp_rejects_overflow() { + assert!(projection_created_at(100, Some(u64::MAX)).is_err()); + } +} diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index d3416d673c5..206a11717ae 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2159,6 +2159,15 @@ async fn handle_leave_request( // handle_reaction() removed — kind:7 reaction dedup and DB writes are now // handled inline in ingest_event() before storage (see ingest.rs step 20a). +fn reject_atomic_project_deletion_side_effect(kind: u32) -> anyhow::Result<()> { + if kind == buzz_core::kind::KIND_PROJECT { + return Err(anyhow::anyhow!( + "Project deletion bypassed the atomic lifecycle handler" + )); + } + Ok(()) +} + /// Handle NIP-09 deletion via `a` tag (addressable/parameterized-replaceable events). /// Parses "kind:pubkey:d-tag" and deletes the corresponding DB record. async fn handle_a_tag_deletion( @@ -2180,6 +2189,9 @@ async fn handle_a_tag_deletion( let kind_num: u32 = parts[0] .parse() .map_err(|_| anyhow::anyhow!("invalid kind in a-tag"))?; + // Project deletion must never reach generic post-storage side effects: its + // event and relational tombstone commit together in the ingest pipeline. + reject_atomic_project_deletion_side_effect(kind_num)?; let pubkey_hex = parts[1]; let d_tag = parts[2]; let actor_bytes = effective_message_author(event, &state.relay_keypair.public_key()); @@ -3684,6 +3696,21 @@ pub async fn publish_nipia_unarchived( mod tests { use super::*; + #[test] + fn project_deletion_side_effect_backstop_rejects_non_atomic_routing() { + let error = reject_atomic_project_deletion_side_effect(buzz_core::kind::KIND_PROJECT) + .expect_err("Project deletion must stay in the atomic lifecycle path"); + + assert_eq!( + error.to_string(), + "Project deletion bypassed the atomic lifecycle handler" + ); + assert!(reject_atomic_project_deletion_side_effect( + buzz_core::kind::KIND_GIT_REPO_ANNOUNCEMENT + ) + .is_ok()); + } + #[test] fn group_members_snapshot_keeps_members_past_one_thousand() { let channel_id = Uuid::new_v4(); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 260dfaed68b..0a811b14ebd 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -466,6 +466,16 @@ async fn main() -> anyhow::Result<()> { ); let state = Arc::new(app_state); + // Project lifecycle events commit independently of their relay-signed read model. + // Start best-effort repair immediately without making listener availability + // depend on it, then retry a bounded batch periodically. + let project_state_projection_cancel = CancellationToken::new(); + let mut project_state_projection_task = + buzz_relay::handlers::project_state_projection::spawn_project_state_projection_reconciler( + Arc::clone(&state), + project_state_projection_cancel.clone(), + ); + // Inter-relay mesh (BUZZ_MESH seam). `boot_mesh` returns None when the // kill switch is off — nothing is bound, published, or spawned, so the // relay behaves byte-identically to a build without the mesh. When @@ -1150,6 +1160,18 @@ async fn main() -> anyhow::Result<()> { serve(router, health_router, Arc::clone(&state)).await?; state.community_revalidator_cancel.cancel(); + project_state_projection_cancel.cancel(); + if tokio::time::timeout( + std::time::Duration::from_secs(5), + &mut project_state_projection_task, + ) + .await + .is_err() + { + project_state_projection_task.abort(); + let _ = project_state_projection_task.await; + tracing::warn!("Project State projection reconciler did not stop within the drain window"); + } // Signal the audit worker to stop accepting, flush buffered entries, and // exit. Uses a CancellationToken so it works regardless of how many diff --git a/docs/nips/NIP-PC.md b/docs/nips/NIP-PC.md new file mode 100644 index 00000000000..3694ff5623b --- /dev/null +++ b/docs/nips/NIP-PC.md @@ -0,0 +1,88 @@ +NIP-PC +====== + +Relay-Authoritative 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) + +## 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. + +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`. + +## Project Coordinate + +Project state is keyed by the canonical NIP-01 coordinate: + +```text +30621:: +``` + +The kind segment is the literal `30621`, the owner is exactly 64 lowercase hexadecimal characters, and the Project `d` value is non-empty and preserved verbatim. Parsing splits on the first two colons so a `d` value containing a colon remains addressable. + +The coordinate selects state only inside the host-bound community. A relay MUST NOT derive the community from a client-supplied Project tag. + +## 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 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. + +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. + +Identity replacement or deletion and its relational lifecycle update MUST commit atomically. + +## Project State Event + +`kind:30623` is relay-only and addressable: + +```jsonc +{ + "kind": 30623, + "pubkey": "", + "tags": [ + ["d", ""], + ["a", "30621::"], + ["rev", "8"], + ["e", "", "", "identity"], + ["e", "", "", "change"] + ], + "content": "{\"v\":1,\"deleted\":false,\"project_tags\":[[\"d\",\"\"]]}" +} +``` + +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. + +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. + +A deleted Project is projected as: + +```json +{"v":1,"deleted":true,"project_tags":[]} +``` + +Version 1 decoders MUST reject unknown JSON fields. A client reading one requested coordinate MUST verify the event signature, require the relay identity advertised by its trusted relay connection, require `kind:30623`, require exactly one matching `a` tag and one canonical `rev` tag, and require content version `1`. Clients do not need to reproduce the relay's effective-tag canonicalization algorithm before displaying a valid projection. + +Client submission of `kind:30623` MUST be rejected, including an event whose signer happens to equal the configured relay pubkey but which arrived through ordinary client ingest. + +## Publication and Reconciliation + +The relational row remains authoritative if projection publication or fan-out fails. The relay MUST retain a durable retry or reconciliation path and MUST NOT report a committed lifecycle event as rolled back merely because derived publication failed. Reconciliation republishes the current row without advancing its revision. + +Every newly signed projection for one address MUST have a `created_at` strictly greater than the previous accepted projection at that address, including repair of the same revision. Allocation uses `max(current Unix time, previous projection created_at + 1)`. If no greater timestamp can be represented, publication fails without changing authoritative state. + +Clients use `rev` as the state revision. Repair or relay-key rotation may produce a new projection event id without changing the revision, so the projection event id is not a revision token. + +## Schema Evolution + +Projection content carries an explicit version. A client that does not understand a Project State version MUST ignore that projection rather than guess at its state. New fields require a new understood version; version-1 decoders reject them through strict JSON deserialization. + +## Security Considerations + +Structural parsing alone does not establish trust. Clients MUST bind the projection signer to the relay identity advertised through their trusted relay connection and MUST match the unhashed `a` tag to the Project coordinate they requested. diff --git a/migrations/0044_project_state.sql b/migrations/0044_project_state.sql new file mode 100644 index 00000000000..2e47802d867 --- /dev/null +++ b/migrations/0044_project_state.sql @@ -0,0 +1,23 @@ +CREATE TABLE project_state_heads ( + community_id UUID NOT NULL REFERENCES communities(id), + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + revision BIGINT NOT NULL CHECK (revision > 0), + deleted BOOLEAN NOT NULL DEFAULT FALSE, + identity_event_id BYTEA NOT NULL CHECK (octet_length(identity_event_id) = 32), + last_event_id BYTEA NOT NULL CHECK (octet_length(last_event_id) = 32), + updated_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(), + PRIMARY KEY (community_id, project_owner, project_d_tag) +); +CREATE TABLE project_related_channels ( + community_id UUID NOT NULL, + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + channel_id UUID NOT NULL, + PRIMARY KEY (community_id, project_owner, project_d_tag, channel_id), + FOREIGN KEY (community_id, project_owner, project_d_tag) + REFERENCES project_state_heads (community_id, project_owner, project_d_tag) + ON DELETE CASCADE +); +SELECT attach_community_write_fence('project_state_heads'); +SELECT attach_community_write_fence('project_related_channels'); diff --git a/migrations/0045_project_state_projection.sql b/migrations/0045_project_state_projection.sql new file mode 100644 index 00000000000..f0d8fb7f3d1 --- /dev/null +++ b/migrations/0045_project_state_projection.sql @@ -0,0 +1,5 @@ +ALTER TABLE project_state_heads + ADD COLUMN projected_revision BIGINT NOT NULL DEFAULT 0 + CHECK (projected_revision >= 0 AND projected_revision <= revision), + ADD COLUMN projection_pubkey BYTEA + CHECK (projection_pubkey IS NULL OR octet_length(projection_pubkey) = 32); diff --git a/schema/schema.sql b/schema/schema.sql index 7d18d825a8b..30c28d65348 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -3787,3 +3787,31 @@ CREATE CONSTRAINT TRIGGER authorization_event_receipt_cardinality AFTER INSERT ON authorization_events DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION authorization_operation_receipt_event_guard_v1(); + +CREATE TABLE project_state_heads ( + community_id UUID NOT NULL REFERENCES communities(id), + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + revision BIGINT NOT NULL CHECK (revision > 0), + projected_revision BIGINT NOT NULL DEFAULT 0 + CHECK (projected_revision >= 0 AND projected_revision <= revision), + projection_pubkey BYTEA + CHECK (projection_pubkey IS NULL OR octet_length(projection_pubkey) = 32), + deleted BOOLEAN NOT NULL DEFAULT FALSE, + identity_event_id BYTEA NOT NULL CHECK (octet_length(identity_event_id) = 32), + last_event_id BYTEA NOT NULL CHECK (octet_length(last_event_id) = 32), + updated_at TIMESTAMPTZ NOT NULL DEFAULT transaction_timestamp(), + PRIMARY KEY (community_id, project_owner, project_d_tag) +); +CREATE TABLE project_related_channels ( + community_id UUID NOT NULL, + project_owner BYTEA NOT NULL CHECK (octet_length(project_owner) = 32), + project_d_tag TEXT NOT NULL CHECK (project_d_tag <> ''), + channel_id UUID NOT NULL, + PRIMARY KEY (community_id, project_owner, project_d_tag, channel_id), + FOREIGN KEY (community_id, project_owner, project_d_tag) + REFERENCES project_state_heads (community_id, project_owner, project_d_tag) + ON DELETE CASCADE +); +SELECT attach_community_write_fence('project_state_heads'); +SELECT attach_community_write_fence('project_related_channels');