diff --git a/VISION_PROJECTS.md b/VISION_PROJECTS.md index 8601b87829e..66ae71bc738 100644 --- a/VISION_PROJECTS.md +++ b/VISION_PROJECTS.md @@ -36,6 +36,14 @@ The portable representation is a NIP-34 repo announcement (kind:30617) — stand Branch protections live in the same event — `buzz-protect` tags. The relay enforces them at the git transport layer. Only npubs listed in `push-allowed` can push to protected branches. Force pushes are blocked. Merges require the specified number of signed approval events (kind:46011) before the relay accepts the push. +Buzz clients that read, modify, and replace an existing repo announcement add +`["buzz-expected-revision", ""]`, naming the announcement they read. +A Buzz-aware relay rejects the update if that announcement is no longer live, +so concurrent metadata or protection edits cannot silently overwrite each +other. The tag is replaced on each successful update rather than accumulated. +It is deliberately opt-in: standard NIP-34 writers that omit the Buzz tag keep +the protocol's normal timestamp-ordered replaceable-event behavior. + Agents inherit access from their owner via [NIP-OA](docs/nips/NIP-OA.md). The relay checks: does the push carry a valid NIP-OA auth tag, and is the owner pubkey in that tag listed in `push-allowed`? If yes, the push is accepted — the agent's own pubkey doesn't need to be in the list. Add a maintainer, and all their authorized agents can push. Remove the maintainer, and all their agents lose access instantly. Agents without NIP-OA attestation are treated as their own identity and must be listed explicitly. Standard NIP-34 clients see a normal repo. gitworkshop.dev renders it. ngit-cli works with it. Buzz clients read the `buzz-` tags and wire up the channel and project UI. One event, two audiences, no custom kind for the repo itself. diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index 8f8db4d2893..120f4d48a7c 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -97,7 +97,12 @@ buzz channels list | jq '.[].name' `protect set` replaces every existing rule for the exact ref pattern. Any constraint omitted from the command is removed. `protect list` reports malformed -stored rules in `validation_error` so an owner can remove and repair them. +stored rules in `validation_error` so an owner can remove and repair them. Repo +metadata updates include a `buzz-expected-revision` tag for the announcement the +CLI read. A Buzz-aware relay rejects a stale update as a conflict (exit code 5), +while tagless standard NIP-34 writers retain the normal replaceable-event +last-write-wins behavior. Deploy the relay support before the updated CLI so the +tag cannot be ignored during a mixed-version rollout. ## Commands diff --git a/crates/buzz-cli/src/commands/repos.rs b/crates/buzz-cli/src/commands/repos.rs index 886d6e04192..a12b8c25bcb 100644 --- a/crates/buzz-cli/src/commands/repos.rs +++ b/crates/buzz-cli/src/commands/repos.rs @@ -1,5 +1,7 @@ use buzz_core::{ - git_perms::{parse_protection_tag, parse_protection_tags, RefPattern}, + git_perms::{ + parse_protection_tag, parse_protection_tags, RefPattern, REPO_EXPECTED_REVISION_TAG, + }, kind::KIND_GIT_REPO_ANNOUNCEMENT, }; use nostr::{Event, EventBuilder, Tag, Timestamp}; @@ -95,6 +97,7 @@ enum RepoChange { fn build_updated_repo_announcement( existing: &Event, change: RepoChange, + now: Timestamp, ) -> Result { let repo_id = repo_id_from_event(existing)?; // What to strip beyond `auth` (always stripped), and what to append. @@ -121,7 +124,7 @@ fn build_updated_repo_announcement( .tags .iter() .filter(|tag| { - if has_tag_name(tag, "auth") { + if has_tag_name(tag, "auth") || has_tag_name(tag, REPO_EXPECTED_REVISION_TAG) { return false; } if removed_channel && has_tag_name(tag, "buzz-channel") { @@ -134,6 +137,10 @@ fn build_updated_repo_announcement( if let Some(tag) = replacement { tags.push(tag); } + let observed_revision = existing.id.to_hex(); + tags.push( + Tag::parse([REPO_EXPECTED_REVISION_TAG, observed_revision.as_str()]).map_err(tag_error)?, + ); let raw_tags: Vec> = tags.iter().map(|tag| tag.as_slice().to_vec()).collect(); parse_protection_tags(&raw_tags).map_err(|error| { @@ -142,13 +149,15 @@ fn build_updated_repo_announcement( )) })?; - // Advance only the observed head. Using wall-clock time here would let a - // delayed writer leapfrog an intervening update and silently erase metadata. - let next_created_at = existing + // Use wall clock for ordinary stale heads while still advancing a future + // observed head. The signed expected-revision tag prevents a delayed writer + // from using the newer timestamp to erase an intervening metadata update. + let after_head = existing .created_at .as_secs() .checked_add(1) .ok_or_else(|| CliError::Other("repository timestamp cannot be advanced".into()))?; + let next_created_at = after_head.max(now.as_secs()); buzz_sdk::build_repo_announcement_with_tags(repo_id, &existing.content, tags) .map_err(|error| CliError::Other(format!("failed to build repository update: {error}"))) .map(|builder| builder.custom_created_at(Timestamp::from(next_created_at))) @@ -193,9 +202,25 @@ fn validate_write_response(raw: &str) -> Result { ) } +fn map_repo_submit_error(error: CliError) -> CliError { + match error { + CliError::Relay { status: 400, body } if body.starts_with("conflict: repository") => { + let message = body + .strip_prefix("conflict: ") + .unwrap_or(body.as_str()) + .to_string(); + CliError::Conflict(message) + } + error => error, + } +} + async fn submit_repo_update(client: &BuzzClient, builder: EventBuilder) -> Result<(), CliError> { let event = client.sign_event(builder)?; - let raw = client.submit_event(event).await?; + let raw = client + .submit_event(event) + .await + .map_err(map_repo_submit_error)?; println!("{}", validate_write_response(&raw)?); Ok(()) } @@ -369,8 +394,11 @@ async fn cmd_protect_set( require_patch, )?; let event = current_repo(client, repo_id).await?; - let builder = - build_updated_repo_announcement(&event, RepoChange::SetProtection(Box::new(tag)))?; + let builder = build_updated_repo_announcement( + &event, + RepoChange::SetProtection(Box::new(tag)), + Timestamp::now(), + )?; submit_repo_update(client, builder).await } @@ -394,6 +422,7 @@ async fn cmd_protect_remove( let builder = build_updated_repo_announcement( &event, RepoChange::RemoveProtection(ref_pattern.to_string()), + Timestamp::now(), )?; submit_repo_update(client, builder).await } @@ -410,8 +439,11 @@ async fn cmd_protect_remove( /// latency. async fn cmd_bind_repo(client: &BuzzClient, repo_id: &str, channel: &str) -> Result<(), CliError> { let event = current_repo(client, repo_id).await?; - let builder = - build_updated_repo_announcement(&event, RepoChange::BindChannel(channel.to_string()))?; + let builder = build_updated_repo_announcement( + &event, + RepoChange::BindChannel(channel.to_string()), + Timestamp::now(), + )?; submit_repo_update(client, builder).await } @@ -472,11 +504,12 @@ pub async fn dispatch(cmd: crate::ReposCmd, client: &BuzzClient) -> Result<(), C #[cfg(test)] mod tests { + use buzz_core::git_perms::REPO_EXPECTED_REVISION_TAG; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use super::{ build_create_announcement, build_protection_tag, build_updated_repo_announcement, - protection_rules_json, validate_write_response, RepoChange, + map_repo_submit_error, protection_rules_json, validate_write_response, RepoChange, }; fn signed_repo(tags: Vec, content: &str, created_at: u64) -> nostr::Event { @@ -500,6 +533,7 @@ mod tests { tag(&["buzz-channel", "channel-id"]), tag(&["future-metadata", "preserve-me"]), tag(&["auth", &"a".repeat(64), "kind=30617", &"b".repeat(128)]), + tag(&[REPO_EXPECTED_REVISION_TAG, &"c".repeat(64)]), tag(&["buzz-protect", "refs/heads/main", "push:member"]), tag(&["buzz-protect", "refs/tags/*", "no-delete"]), ], @@ -512,6 +546,7 @@ mod tests { let updated = build_updated_repo_announcement( &existing, RepoChange::SetProtection(Box::new(replacement)), + Timestamp::from(100_u64), ) .expect("build update") .sign_with_keys(&Keys::generate()) @@ -523,6 +558,15 @@ mod tests { .tags .iter() .any(|tag| tag.as_slice().first().map(String::as_str) == Some("auth"))); + let revision_tags: Vec<_> = updated + .tags + .iter() + .filter(|tag| { + tag.as_slice().first().map(String::as_str) == Some(REPO_EXPECTED_REVISION_TAG) + }) + .collect(); + assert_eq!(revision_tags.len(), 1); + assert_eq!(revision_tags[0].as_slice()[1], existing.id.to_hex()); assert!(updated .tags .iter() @@ -559,6 +603,43 @@ mod tests { ); } + #[test] + fn repository_update_uses_wall_clock_for_stale_heads_and_advances_future_heads() { + let stale = signed_repo(vec![tag(&["d", "stale"])], "", 100); + let stale_update = build_updated_repo_announcement( + &stale, + RepoChange::BindChannel(uuid::Uuid::new_v4().to_string()), + Timestamp::from(10_000_u64), + ) + .expect("build stale-head update") + .sign_with_keys(&Keys::generate()) + .expect("sign stale-head update"); + assert_eq!(stale_update.created_at.as_secs(), 10_000); + + let future = signed_repo(vec![tag(&["d", "future"])], "", 20_000); + let future_update = build_updated_repo_announcement( + &future, + RepoChange::BindChannel(uuid::Uuid::new_v4().to_string()), + Timestamp::from(10_000_u64), + ) + .expect("build future-head update") + .sign_with_keys(&Keys::generate()) + .expect("sign future-head update"); + assert_eq!(future_update.created_at.as_secs(), 20_001); + } + + #[test] + fn repository_update_rejects_timestamp_overflow() { + let existing = signed_repo(vec![tag(&["d", "demo"])], "", u64::MAX); + let error = build_updated_repo_announcement( + &existing, + RepoChange::BindChannel(uuid::Uuid::new_v4().to_string()), + Timestamp::from(0_u64), + ) + .expect_err("maximum timestamp cannot be advanced"); + assert!(error.to_string().contains("timestamp cannot be advanced")); + } + #[test] fn protection_remove_preserves_other_patterns() { let existing = signed_repo( @@ -574,6 +655,7 @@ mod tests { let updated = build_updated_repo_announcement( &existing, RepoChange::RemoveProtection("refs/heads/main".into()), + Timestamp::from(10_u64), ) .expect("build removal") .sign_with_keys(&Keys::generate()) @@ -611,6 +693,7 @@ mod tests { let error = build_updated_repo_announcement( &existing, RepoChange::SetProtection(Box::new(replacement)), + Timestamp::from(10_u64), ) .expect_err("malformed existing rule must fail closed"); @@ -637,6 +720,7 @@ mod tests { let error = build_updated_repo_announcement( &existing, RepoChange::SetProtection(Box::new(replacement)), + Timestamp::from(10_u64), ) .expect_err("the 51st rule must be rejected"); @@ -705,11 +789,14 @@ mod tests { 100, ); - let updated = - build_updated_repo_announcement(&existing, RepoChange::BindChannel(channel.clone())) - .expect("build bind update") - .sign_with_keys(&Keys::generate()) - .expect("sign bind update"); + let updated = build_updated_repo_announcement( + &existing, + RepoChange::BindChannel(channel.clone()), + Timestamp::from(100_u64), + ) + .expect("build bind update") + .sign_with_keys(&Keys::generate()) + .expect("sign bind update"); assert_eq!(updated.content, "repository content"); assert_eq!(updated.created_at.as_secs(), 101); @@ -745,11 +832,14 @@ mod tests { let channel = uuid::Uuid::new_v4().to_string(); let existing = signed_repo(vec![tag(&["d", "demo"])], "", 10); - let updated = - build_updated_repo_announcement(&existing, RepoChange::BindChannel(channel.clone())) - .expect("build bind update") - .sign_with_keys(&Keys::generate()) - .expect("sign bind update"); + let updated = build_updated_repo_announcement( + &existing, + RepoChange::BindChannel(channel.clone()), + Timestamp::from(10_u64), + ) + .expect("build bind update") + .sign_with_keys(&Keys::generate()) + .expect("sign bind update"); assert!(updated .tags @@ -761,9 +851,12 @@ mod tests { fn bind_channel_rejects_malformed_uuid() { let existing = signed_repo(vec![tag(&["d", "demo"])], "", 10); - let error = - build_updated_repo_announcement(&existing, RepoChange::BindChannel("nope".into())) - .expect_err("malformed channel id must not build an update"); + let error = build_updated_repo_announcement( + &existing, + RepoChange::BindChannel("nope".into()), + Timestamp::from(10_u64), + ) + .expect_err("malformed channel id must not build an update"); assert!(matches!(error, crate::error::CliError::Usage(_))); } @@ -836,6 +929,27 @@ mod tests { assert!(matches!(error, crate::error::CliError::Conflict(_))); } + #[test] + fn repository_relay_conflict_maps_to_exit_five_error() { + let error = map_repo_submit_error(crate::error::CliError::Relay { + status: 400, + body: "conflict: repository changed since it was loaded".into(), + }); + assert!(matches!(error, crate::error::CliError::Conflict(_))); + } + + #[test] + fn non_conflict_relay_error_keeps_its_status() { + let error = map_repo_submit_error(crate::error::CliError::Relay { + status: 400, + body: "invalid: bad repository expected revision".into(), + }); + assert!(matches!( + error, + crate::error::CliError::Relay { status: 400, .. } + )); + } + #[test] fn successful_write_response_is_normalized() { let output = validate_write_response( diff --git a/crates/buzz-core/src/git_perms.rs b/crates/buzz-core/src/git_perms.rs index 391781163b8..b96c3b86734 100644 --- a/crates/buzz-core/src/git_perms.rs +++ b/crates/buzz-core/src/git_perms.rs @@ -39,6 +39,13 @@ pub const GIT_NO_CHANNEL_BINDING_TOKEN: &str = "no_channel_binding"; pub const GIT_NO_CHANNEL_BINDING_BODY: &str = "no_channel_binding: repository has no channel binding"; +/// Buzz extension tag carrying the exact kind:30617 head an update observed. +/// +/// Repo metadata writers replace this tag on every read-modify-write. The relay +/// uses it as an optimistic concurrency precondition while vanilla NIP-34 +/// announcements without the tag retain standard latest-wins behavior. +pub const REPO_EXPECTED_REVISION_TAG: &str = "buzz-expected-revision"; + /// Maximum number of `buzz-protect` tags per repo. pub const MAX_PROTECTION_RULES: usize = 50; /// Maximum character length of a ref pattern. diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 9b575b6ea18..ac5dffff2bd 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -544,18 +544,14 @@ impl Db { .await } - /// Atomically replace a NIP-33 parameterized replaceable event. - /// - /// Replacement keys on `(kind, pubkey, d_tag)` across channels. The - /// highest timestamp wins; same-second ties use the lowest event ID. - #[datastore_span(name = "replace_parameterized_event", system = "postgresql")] - pub async fn replace_parameterized_event( + async fn replace_parameterized_event_with_precondition_inner( &self, community_id: CommunityId, event: &nostr::Event, d_tag: &str, channel_id: Option, - ) -> Result<(StoredEvent, bool)> { + precondition: ParameterizedReplacePrecondition<'_>, + ) -> Result { let (mut tx, transaction_timer) = observability::begin_transaction( &self.pool, TransactionOperation::ReplaceParameterizedEvent, @@ -570,19 +566,70 @@ impl Db { event, d_tag, channel_id, - ParameterizedReplacePrecondition::Unconditional, + precondition, ) .await?; - let was_inserted = result.status == ParameterizedReplaceStatus::Inserted; - if was_inserted { + if result.status == ParameterizedReplaceStatus::Inserted { tx.commit().await?; } else { tx.rollback().await?; } - Ok((result.event, was_inserted)) + Ok(result) }) .await } + + /// Atomically replace a NIP-33 event using an explicit revision precondition. + /// + /// The coordinate advisory lock, exact-replay check, revision comparison, + /// and latest-wins comparison all execute inside one transaction. Inserted + /// heads commit; every no-op or conflict status rolls back. + #[datastore_span( + name = "replace_parameterized_event_with_precondition", + system = "postgresql" + )] + pub async fn replace_parameterized_event_with_precondition( + &self, + community_id: CommunityId, + event: &nostr::Event, + d_tag: &str, + channel_id: Option, + precondition: ParameterizedReplacePrecondition<'_>, + ) -> Result { + self.replace_parameterized_event_with_precondition_inner( + community_id, + event, + d_tag, + channel_id, + precondition, + ) + .await + } + + /// Atomically replace a NIP-33 parameterized replaceable event. + /// + /// Replacement keys on `(kind, pubkey, d_tag)` across channels. The + /// highest timestamp wins; same-second ties use the lowest event ID. + #[datastore_span(name = "replace_parameterized_event", system = "postgresql")] + pub async fn replace_parameterized_event( + &self, + community_id: CommunityId, + event: &nostr::Event, + d_tag: &str, + channel_id: Option, + ) -> Result<(StoredEvent, bool)> { + let result = self + .replace_parameterized_event_with_precondition_inner( + community_id, + event, + d_tag, + channel_id, + ParameterizedReplacePrecondition::Unconditional, + ) + .await?; + let was_inserted = result.status == ParameterizedReplaceStatus::Inserted; + Ok((result.event, was_inserted)) + } } #[cfg(test)] @@ -1195,6 +1242,146 @@ mod tests { tx.rollback().await.expect("roll back missing revision tx"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn conditional_parameterized_replacement_allows_one_concurrent_writer_and_replay() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("repository-cas-{}", Uuid::new_v4().simple()); + let base = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_GIT_REPO_ANNOUNCEMENT as u16), + content, + ) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign repository announcement") + }; + let old = event("old", base); + let first = event("first", base + 1); + let second = event("second", base + 2); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert repository head") + .1 + ); + + let expected_first = old.id.as_bytes().to_vec(); + let expected_second = expected_first.clone(); + let first_write = db.replace_parameterized_event_with_precondition( + community, + &first, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision(&expected_first), + ); + let second_write = db.replace_parameterized_event_with_precondition( + community, + &second, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision(&expected_second), + ); + let (first_result, second_result) = tokio::join!(first_write, second_write); + let first_result = first_result.expect("first conditional write"); + let second_result = second_result.expect("second conditional write"); + let statuses = [first_result.status, second_result.status]; + assert_eq!( + statuses + .iter() + .filter(|status| **status == replaceable::ParameterizedReplaceStatus::Inserted) + .count(), + 1 + ); + assert_eq!( + statuses + .iter() + .filter(|status| { + **status == replaceable::ParameterizedReplaceStatus::RevisionMismatch + }) + .count(), + 1 + ); + + let winner = if first_result.status == replaceable::ParameterizedReplaceStatus::Inserted { + &first + } else { + &second + }; + let replay = db + .replace_parameterized_event_with_precondition( + community, + winner, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + old.id.as_bytes().as_slice(), + ), + ) + .await + .expect("replay winning repository update"); + assert_eq!( + replay.status, + replaceable::ParameterizedReplaceStatus::Duplicate + ); + + let replay_only = db + .replace_parameterized_event_with_precondition( + community, + winner, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExactReplayOnly, + ) + .await + .expect("exact replay-only repository update"); + assert_eq!( + replay_only.status, + replaceable::ParameterizedReplaceStatus::Duplicate + ); + + let distinct = event("distinct", base + 3); + let replay_miss = db + .replace_parameterized_event_with_precondition( + community, + &distinct, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExactReplayOnly, + ) + .await + .expect("reject distinct replay-only repository update"); + assert_eq!( + replay_miss.status, + replaceable::ParameterizedReplaceStatus::ReplayOnlyMiss + ); + + let stale = event("stale", base); + let stale_result = db + .replace_parameterized_event_with_precondition( + community, + &stale, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + winner.id.as_bytes().as_slice(), + ), + ) + .await + .expect("evaluate matching but superseded repository update"); + assert_eq!( + stale_result.status, + replaceable::ParameterizedReplaceStatus::Superseded + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn parameterized_replacement_rolls_back_when_mention_indexing_fails() { diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index dd2fa6e93e0..9ac088ee6db 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -10,6 +10,7 @@ use tracing::{debug, error, info, warn}; use uuid::Uuid; use buzz_auth::Scope; +use buzz_core::git_perms::REPO_EXPECTED_REVISION_TAG; use buzz_core::kind::{ event_kind_u32, is_identity_archive_request_kind, is_parameterized_replaceable, is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, @@ -288,6 +289,25 @@ fn emit_product_feedback_success( ); } +fn emit_repository_duplicate_success( + tracer: &Arc, + tenant: &TenantContext, + event: &Event, + auth: &IngestAuth, +) { + // Repository announcements are relay-global. The conformance model folds + // a global exact replay into the same observation shape as a global + // insert, while the durable and domain side effects remain suppressed. + emit( + tracer, + TraceAction::WriteInsertGlobal { + msg_id: msg_id_label(event.id.as_bytes()), + claimed_community: claimed_community_from_event(event), + }, + state_for_request(tenant, auth.pubkey()), + ); +} + /// Increment the rejection counter with a bounded reason and transport label. /// /// Shared by the WS `EVENT` handler and the HTTP `POST /events` handler so @@ -391,6 +411,38 @@ pub enum IngestError { Internal(String), } +fn parse_repo_expected_revision(event: &Event) -> Result>, IngestError> { + let mut revision = None; + for tag in event.tags.iter().filter(|tag| { + tag.as_slice().first().map(String::as_str) == Some(REPO_EXPECTED_REVISION_TAG) + }) { + if revision.is_some() { + return Err(IngestError::Rejected( + "invalid: duplicate repository expected revision".into(), + )); + } + let values = tag.as_slice(); + let Some(value) = values.get(1).filter(|_| values.len() == 2) else { + return Err(IngestError::Rejected( + "invalid: bad repository expected revision".into(), + )); + }; + if value.len() != 64 + || !value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(IngestError::Rejected( + "invalid: bad repository expected revision".into(), + )); + } + revision = Some(hex::decode(value).map_err(|_| { + IngestError::Rejected("invalid: bad repository expected revision".into()) + })?); + } + Ok(revision) +} + /// Map the durable community write-fence lookup onto the ingest error taxonomy. /// /// An inactive community is an authorization decision and keeps the exact @@ -3148,11 +3200,84 @@ async fn ingest_event_inner( buzz_db::event::D_TAG_MAX_LEN, ))); } - state - .db - .replace_parameterized_event(tenant.community(), &event, &d_tag, channel_id) - .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? + if kind_u32 == KIND_GIT_REPO_ANNOUNCEMENT { + use buzz_db::replaceable::{ + ParameterizedReplacePrecondition, ParameterizedReplaceStatus, + }; + + let (expected_revision, revision_error) = match parse_repo_expected_revision(&event) { + Ok(revision) => (revision, None), + Err(error) => (None, Some(error)), + }; + let precondition = if revision_error.is_some() { + ParameterizedReplacePrecondition::ExactReplayOnly + } else if let Some(expected) = expected_revision.as_deref() { + ParameterizedReplacePrecondition::ExpectedRevision(expected) + } else { + ParameterizedReplacePrecondition::Unconditional + }; + + if precondition == ParameterizedReplacePrecondition::Unconditional { + state + .db + .replace_parameterized_event(tenant.community(), &event, &d_tag, channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: {e}")))? + } else { + let result = state + .db + .replace_parameterized_event_with_precondition( + tenant.community(), + &event, + &d_tag, + channel_id, + precondition, + ) + .await + .map_err(|e| IngestError::Internal(format!("error: {e}")))?; + match result.status { + ParameterizedReplaceStatus::Inserted => (result.event, true), + ParameterizedReplaceStatus::Duplicate => { + emit_repository_duplicate_success(tracer, tenant, &event, &auth); + return Ok(IngestResult { + event_id: event_id_hex, + accepted: true, + message: String::new(), + }); + } + ParameterizedReplaceStatus::RevisionMissing => { + return Err(IngestError::Rejected( + "conflict: repository revision does not exist".into(), + )); + } + ParameterizedReplaceStatus::RevisionMismatch => { + return Err(IngestError::Rejected( + "conflict: repository changed since it was loaded".into(), + )); + } + ParameterizedReplaceStatus::Superseded => { + return Err(IngestError::Rejected( + "conflict: repository update was superseded; refresh and try again" + .into(), + )); + } + ParameterizedReplaceStatus::ReplayOnlyMiss => { + return Err(revision_error.unwrap_or_else(|| { + IngestError::Internal( + "error: replay-only repository update lacked a revision error" + .into(), + ) + })); + } + } + } + } else { + state + .db + .replace_parameterized_event(tenant.community(), &event, &d_tag, channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: {e}")))? + } } else { let thread_params = thread_meta.as_ref().map(|m| m.as_params()); match state @@ -3287,6 +3412,66 @@ mod tests { }; use nostr::{EventBuilder, Kind}; + fn repo_announcement_with_revision_tags(revisions: &[Vec]) -> Event { + let mut tags = vec![nostr::Tag::parse(["d", "demo"]).expect("d tag")]; + for revision in revisions { + tags.push(nostr::Tag::parse(revision.clone()).expect("revision tag")); + } + EventBuilder::new(Kind::Custom(KIND_GIT_REPO_ANNOUNCEMENT as u16), "") + .tags(tags) + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign repository announcement") + } + + #[test] + fn repository_expected_revision_is_optional_and_canonical() { + let vanilla = repo_announcement_with_revision_tags(&[]); + assert!(parse_repo_expected_revision(&vanilla) + .expect("vanilla announcement") + .is_none()); + + let revision = "ab".repeat(32); + let conditional = repo_announcement_with_revision_tags(&[vec![ + REPO_EXPECTED_REVISION_TAG.to_string(), + revision.clone(), + ]]); + assert_eq!( + parse_repo_expected_revision(&conditional).expect("conditional announcement"), + Some(hex::decode(revision).expect("revision hex")) + ); + } + + #[test] + fn repository_expected_revision_rejects_malformed_or_duplicate_tags() { + let cases = vec![ + vec![vec![REPO_EXPECTED_REVISION_TAG.to_string()]], + vec![vec![ + REPO_EXPECTED_REVISION_TAG.to_string(), + "abc".to_string(), + ]], + vec![vec![ + REPO_EXPECTED_REVISION_TAG.to_string(), + "AB".repeat(32), + ]], + vec![vec![ + REPO_EXPECTED_REVISION_TAG.to_string(), + "ab".repeat(32), + "extra".to_string(), + ]], + vec![ + vec![REPO_EXPECTED_REVISION_TAG.to_string(), "ab".repeat(32)], + vec![REPO_EXPECTED_REVISION_TAG.to_string(), "cd".repeat(32)], + ], + ]; + for tags in cases { + let event = repo_announcement_with_revision_tags(&tags); + assert!(matches!( + parse_repo_expected_revision(&event), + Err(IngestError::Rejected(_)) + )); + } + } + #[test] fn missing_huddle_backing_channel_is_a_client_rejection() { let channel_id = Uuid::new_v4(); @@ -3641,6 +3826,44 @@ mod tests { )); } + #[test] + fn repository_duplicate_action_satisfies_ingest_emit_guard() { + let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); + let tenant = TenantContext::resolved(community, "repository.test"); + let keys = nostr::Keys::generate(); + let event = EventBuilder::new( + Kind::Custom(KIND_GIT_REPO_ANNOUNCEMENT as u16), + "repository announcement", + ) + .tags([nostr::Tag::parse(["d", "demo"]).expect("d tag")]) + .sign_with_keys(&keys) + .expect("sign repository announcement"); + let auth = IngestAuth::Http { + pubkey: keys.public_key(), + scopes: vec![Scope::MessagesWrite], + auth_method: HttpAuthMethod::Nip98, + }; + let tracer = Arc::new(VecTracer::default()); + let abstract_state = state_for_request(&tenant, auth.pubkey()); + + { + let (guard, counting) = EmitGuard::arm( + tracer.clone(), + abstract_state, + "ingest_event_exited_without_trace", + ); + emit_repository_duplicate_success(&counting, &tenant, &event, &auth); + drop(guard); + } + + let steps = tracer.steps.lock().expect("trace lock"); + assert_eq!(steps.len(), 1); + assert!(matches!( + steps[0].action, + TraceAction::WriteInsertGlobal { .. } + )); + } + #[test] fn nip_ia_requests_are_global_only() { // NIP-IA requests drive relay-global archive state; a stray `h` tag