diff --git a/knot2/crates/knot-git/src/repo.rs b/knot2/crates/knot-git/src/repo.rs index 3f5068385..b38e1bf24 100644 --- a/knot2/crates/knot-git/src/repo.rs +++ b/knot2/crates/knot-git/src/repo.rs @@ -411,15 +411,17 @@ pub enum RefClass { Checkpoint, Hidden, Atproto, + Tngl, } impl RefClass { - pub const ALL: [RefClass; 5] = [ + pub const ALL: [RefClass; 6] = [ RefClass::Public, RefClass::Cob, RefClass::Checkpoint, RefClass::Hidden, RefClass::Atproto, + RefClass::Tngl, ]; pub fn of(name: &RefName) -> RefClass { @@ -433,6 +435,7 @@ impl RefClass { "cob-checkpoints" => RefClass::Checkpoint, "hidden" => RefClass::Hidden, "atproto" => RefClass::Atproto, + "tngl" => RefClass::Tngl, _ => RefClass::Public, } } @@ -444,13 +447,18 @@ impl RefClass { RefClass::Checkpoint => Some("refs/cob-checkpoints"), RefClass::Hidden => Some("refs/hidden"), RefClass::Atproto => Some("refs/atproto"), + RefClass::Tngl => Some("refs/tngl"), } } pub fn is_public(self) -> bool { match self { RefClass::Public => true, - RefClass::Cob | RefClass::Checkpoint | RefClass::Hidden | RefClass::Atproto => false, + RefClass::Cob + | RefClass::Checkpoint + | RefClass::Hidden + | RefClass::Atproto + | RefClass::Tngl => false, } } } @@ -1769,6 +1777,11 @@ mod tests { ("refs/cob-checkpoints/main", RefClass::Checkpoint), ("refs/hidden", RefClass::Hidden), ("refs/hidden/feature/main", RefClass::Hidden), + ("refs/tngl", RefClass::Tngl), + ( + "refs/tngl/keep/did%3aplc%3asquid/org.tangled.review.patch/x/y", + RefClass::Tngl, + ), ]; cases.iter().for_each(|(raw, class)| { assert_eq!( diff --git a/knot2/crates/knot-lfs/src/gc.rs b/knot2/crates/knot-lfs/src/gc.rs index b55210711..4bf6b099a 100644 --- a/knot2/crates/knot-lfs/src/gc.rs +++ b/knot2/crates/knot-lfs/src/gc.rs @@ -43,7 +43,7 @@ fn unix_seconds(now: SystemTime) -> UnixSeconds { fn reachable_pointers(repo: &Repo, floor: UnixSeconds) -> Result, GitError> { let roots: Vec = repo .retained_oids(floor, |class| match class { - RefClass::Public | RefClass::Hidden => true, + RefClass::Public | RefClass::Hidden | RefClass::Tngl => true, RefClass::Cob | RefClass::Checkpoint | RefClass::Atproto => false, })? .into_iter() diff --git a/knot2/crates/knot-maintenance/src/lib.rs b/knot2/crates/knot-maintenance/src/lib.rs index 9d979d725..633225047 100644 --- a/knot2/crates/knot-maintenance/src/lib.rs +++ b/knot2/crates/knot-maintenance/src/lib.rs @@ -344,7 +344,8 @@ fn collect_roots(repo: &Repo, retention_floor: UnixSeconds) -> Result true, + | RefClass::Atproto + | RefClass::Tngl => true, }) .map_err(MaintError::from) } diff --git a/knot2/crates/knot-messages/src/lib.rs b/knot2/crates/knot-messages/src/lib.rs index f87c32d00..1ce5e17ff 100644 --- a/knot2/crates/knot-messages/src/lib.rs +++ b/knot2/crates/knot-messages/src/lib.rs @@ -174,11 +174,12 @@ message_group! { message_group! { RejectConfig => RejectMessages @ "messages.reject" { - reserved_refs: Line @ "KNOT_MESSAGES_REJECT_RESERVED_REFS" = "refs/cobs/*, refs/cob-checkpoints/*, refs/hidden/* and refs/atproto/* are reserved and can't be pushed", + reserved_refs: Line @ "KNOT_MESSAGES_REJECT_RESERVED_REFS" = "refs/cobs/*, refs/cob-checkpoints/*, refs/hidden/*, refs/atproto/* and refs/tngl/* are reserved and can't be pushed", cob_create_only: Line @ "KNOT_MESSAGES_REJECT_COB_CREATE_ONLY" = "existing refs/cobs/* object cannot be modified or deleted over the wire", cob_delete: Line @ "KNOT_MESSAGES_REJECT_COB_DELETE" = "refs/cobs/* stores append-only collaborative objects and cannot be deleted", hidden_reserved: Line @ "KNOT_MESSAGES_REJECT_HIDDEN_RESERVED" = "refs/hidden/* is reserved for server-side fork staging and cannot be pushed", atproto_reserved: Line @ "KNOT_MESSAGES_REJECT_ATPROTO_RESERVED" = "refs/atproto/* is reserved for the knot's atproto commit chain and can't be pushed", + tngl_reserved: Line @ "KNOT_MESSAGES_REJECT_TNGL_RESERVED" = "refs/tngl/* is reserved for knot-written keep refs and can't be pushed", checkpoint_reserved: Line @ "KNOT_MESSAGES_REJECT_CHECKPOINT_RESERVED" = "refs/cob-checkpoints/* is reserved for knot-written collaborative-object snapshots and can't be pushed", cob_verification: Line @ "KNOT_MESSAGES_REJECT_COB_VERIFICATION" = "collaborative-object verification failed: {error}", ref_exists: Line @ "KNOT_MESSAGES_REJECT_REF_EXISTS" = "reference already exists", diff --git a/knot2/crates/knot-pack/src/guard.rs b/knot2/crates/knot-pack/src/guard.rs index 03edd7a0a..578dfb12e 100644 --- a/knot2/crates/knot-pack/src/guard.rs +++ b/knot2/crates/knot-pack/src/guard.rs @@ -43,6 +43,7 @@ impl PushGuard { } RefClass::Hidden => RefDecision::Reject(self.messages.reject.hidden_reserved.text()), RefClass::Atproto => RefDecision::Reject(self.messages.reject.atproto_reserved.text()), + RefClass::Tngl => RefDecision::Reject(self.messages.reject.tngl_reserved.text()), } } diff --git a/knot2/crates/knot-pack/tests/hardening.rs b/knot2/crates/knot-pack/tests/hardening.rs index 52a53967b..51f9ad5f0 100644 --- a/knot2/crates/knot-pack/tests/hardening.rs +++ b/knot2/crates/knot-pack/tests/hardening.rs @@ -196,7 +196,7 @@ fn push_namespace_gating_accepts_only_unreserved_refs() { let did = RepoDid::new("did:plc:squid").unwrap(); let (bare, _work, c1, pack) = seeded(&layout, &did); - let cases: [(&str, &str, bool); 7] = [ + let cases: [(&str, &str, bool); 8] = [ ( "refs/cobs/org.tangled.graph.collaborator/evil", "ng refs/cobs/org.tangled.graph.collaborator/evil", @@ -215,6 +215,11 @@ fn push_namespace_gating_accepts_only_unreserved_refs() { false, ), ("refs/cob-checkpoints", "ng refs/cob-checkpoints", false), + ( + "refs/tngl/keep/did%3aplc%3asquid/org.tangled.review.patch/x/y", + "ng refs/tngl/keep/did%3aplc%3asquid/org.tangled.review.patch/x/y", + false, + ), ("refs/notes/commits", "ok refs/notes/commits", true), ]; cases.iter().for_each(|(refname, verdict, lands)| { diff --git a/knot2/crates/knot-pack/tests/quarantine.rs b/knot2/crates/knot-pack/tests/quarantine.rs index 995614c07..8fc3d3003 100644 --- a/knot2/crates/knot-pack/tests/quarantine.rs +++ b/knot2/crates/knot-pack/tests/quarantine.rs @@ -395,6 +395,10 @@ fn the_push_guard_allows_a_branch_and_names_the_namespace_it_refuses() { "knot-written collaborative-object snapshots", ), ("refs/hidden/feature/main", "server-side fork staging"), + ( + "refs/tngl/keep/did%3aplc%3asquid/org.tangled.review.patch/x/y", + "knot-written keep refs", + ), ] .iter() .for_each(|(refname, names_itself)| { diff --git a/knot2/crates/knot-xrpc/src/body.rs b/knot2/crates/knot-xrpc/src/body.rs index 6ef6e9166..bde4b10b9 100644 --- a/knot2/crates/knot-xrpc/src/body.rs +++ b/knot2/crates/knot-xrpc/src/body.rs @@ -2,7 +2,6 @@ use serde::de::{self, Deserializer}; use serde::ser::Serializer; use serde::{Deserialize, Serialize}; -use knot_types::RefName; use url::Url; #[derive(Clone, Debug)] @@ -54,45 +53,3 @@ knot_types::text_newtype! { pub(crate) struct CommitMessage(String) => verbatim; pub(crate) struct CommitBody(String) => verbatim; } - -pub(crate) struct ForkRef(RefName); - -impl ForkRef { - pub(crate) fn hidden_ref(&self, remote: &RemoteRef) -> Option { - RefName::new(format!("{}/{}", self.0.as_str(), remote.as_str())).ok() - } -} - -// Comes as a bare name, -// and `refs/hidden/` is the only place a fork staging ref is allowed to exist, -// so `Deserialize` prepends the prefix & `RefName::new` validates the whole string, -// therefore there's nothing to append after check. -impl<'de> Deserialize<'de> for ForkRef { - fn deserialize>(deserializer: D) -> Result { - let raw = String::deserialize(deserializer)?; - RefName::new(format!("refs/hidden/{raw}")) - .map(ForkRef) - .map_err(|_| de::Error::custom("invalid fork ref")) - } -} - -pub(crate) struct RemoteRef(knot_types::BranchName); - -impl RemoteRef { - pub(crate) fn as_str(&self) -> &str { - self.0.as_str() - } - - pub(crate) fn head_ref(&self) -> RefName { - self.0.head_ref() - } -} - -impl<'de> Deserialize<'de> for RemoteRef { - fn deserialize>(deserializer: D) -> Result { - let raw = String::deserialize(deserializer)?; - knot_types::BranchName::new(raw) - .map(RemoteRef) - .map_err(|_| de::Error::custom("invalid remote ref")) - } -} diff --git a/knot2/crates/knot-xrpc/src/branches.rs b/knot2/crates/knot-xrpc/src/branches.rs index d3a26dbbd..95f62f434 100644 --- a/knot2/crates/knot-xrpc/src/branches.rs +++ b/knot2/crates/knot-xrpc/src/branches.rs @@ -14,15 +14,7 @@ use crate::error::XrpcError; use crate::reads::require_hosted; use crate::{XrpcState, decode, ok_empty, run_blocking}; -pub(crate) const SET_DEFAULT_ROUTE: &str = "/xrpc/sh.tangled.repo.setDefaultBranch"; -pub(crate) const DELETE_ROUTE: &str = "/xrpc/sh.tangled.repo.deleteBranch"; - -#[derive(Deserialize)] -struct SetDefaultBranchInput { - repo: RepoDid, - #[serde(rename = "defaultBranch")] - default_branch: BranchName, -} +pub(crate) const DELETE_ROUTE: &str = "/xrpc/org.tangled.git.deleteBranch"; #[derive(Deserialize)] struct DeleteBranchInput { @@ -32,34 +24,17 @@ struct DeleteBranchInput { const BRANCH_DENIED: &str = "only repository owner or a collaborator may change its branches"; -pub(crate) async fn set_default_branch( - State(state): State>>, - headers: HeaderMap, - method: crate::Method, - body: Bytes, -) -> Result { - let actor = state.authenticate_write(&headers, &method).await?; - let input: SetDefaultBranchInput = decode(&body)?; - let repo_did = require_hosted(&state, input.repo)?; - crate::authorize_push(&state, &actor, &repo_did, BRANCH_DENIED).await?; - let refname = input.default_branch.head_ref(); - - let layout = state.layout.clone(); - let target = repo_did.clone(); - run_blocking(move || { - let repo = layout.open(&target)?; - let target_exists = repo.find_ref(&refname)?.is_some(); - let has_branches = !repo.branches()?.is_empty(); - if !target_exists && has_branches { - return Err(XrpcError::not_found("no such branch to set as default")); - } - repo.set_head(&refname).map_err(XrpcError::from) - }) - .await?; - - crate::refrecords::sync_default_branch(&state, &repo_did).await; - - Ok(ok_empty()) +pub(crate) fn set_default_head( + repo: &knot_git::Repo, + default_branch: &BranchName, +) -> Result<(), XrpcError> { + let refname = default_branch.head_ref(); + let target_exists = repo.find_ref(&refname)?.is_some(); + let has_branches = !repo.branches()?.is_empty(); + if !target_exists && has_branches { + return Err(XrpcError::not_found("no such branch to set as default")); + } + repo.set_head(&refname).map_err(XrpcError::from) } pub(crate) async fn delete_branch( diff --git a/knot2/crates/knot-xrpc/src/error.rs b/knot2/crates/knot-xrpc/src/error.rs index d6d7936de..412e39f4c 100644 --- a/knot2/crates/knot-xrpc/src/error.rs +++ b/knot2/crates/knot-xrpc/src/error.rs @@ -525,7 +525,7 @@ mod tests { async fn a_stale_write_answers_the_uri_and_cid_a_client_rebases_onto() { let uri = knot_types::AtUri::from_parts_owned( "did:plc:squid", - "sh.tangled.repo.issue", + "org.tangled.track.ticket", "3lubrptx57d22", ) .unwrap(); diff --git a/knot2/crates/knot-xrpc/src/firehose.rs b/knot2/crates/knot-xrpc/src/firehose.rs index d7e485140..e39d0d49a 100644 --- a/knot2/crates/knot-xrpc/src/firehose.rs +++ b/knot2/crates/knot-xrpc/src/firehose.rs @@ -1188,7 +1188,7 @@ mod tests { (dir, layout, did, repo, signer) } - const COLLECTION: &str = "sh.tangled.test.record"; + const COLLECTION: &str = "org.tangled.test.record"; fn collection() -> knot_types::RecordCollection { knot_types::RecordCollection::new(COLLECTION).unwrap() diff --git a/knot2/crates/knot-xrpc/src/forks.rs b/knot2/crates/knot-xrpc/src/forks.rs index 05f99e05a..037d1fc56 100644 --- a/knot2/crates/knot-xrpc/src/forks.rs +++ b/knot2/crates/knot-xrpc/src/forks.rs @@ -14,13 +14,12 @@ use knot_pack::{FetchError, HaveOids, PackLimits, UpstreamRefs, WantOids}; use knot_runtime::{Clock, HttpTransport}; use knot_types::{BranchName, ObjectFormat, Oid, OriginUrl, OwnerDid, RefName, RepoDid, RepoName}; -use crate::body::{ForkRef, RemoteRef, SourceUrl}; +use crate::body::SourceUrl; use crate::error::XrpcError; use crate::reads::{HostedRepo, require_hosted}; use crate::{XrpcState, decode, ok_empty, run_blocking}; -pub(crate) const SYNC_ROUTE: &str = "/xrpc/sh.tangled.repo.forkSync"; -pub(crate) const HIDDEN_REF_ROUTE: &str = "/xrpc/sh.tangled.repo.hiddenRef"; +pub(crate) const SYNC_ROUTE: &str = "/xrpc/org.tangled.repo.forkSync"; #[derive(Clone)] pub(crate) enum Upstream { @@ -436,7 +435,7 @@ fn load_fork_state(repo: &Repo) -> Result { .references()? .into_iter() .filter(|record| match RefClass::of(&record.name) { - RefClass::Public | RefClass::Hidden => true, + RefClass::Public | RefClass::Hidden | RefClass::Tngl => true, RefClass::Cob | RefClass::Checkpoint | RefClass::Atproto => false, }) .map(|record| record.target) @@ -600,56 +599,3 @@ pub(crate) async fn fork_sync( .into_response()), } } - -#[derive(Deserialize)] -struct HiddenRefInput { - repo: RepoDid, - #[serde(rename = "forkRef")] - fork_ref: ForkRef, - #[serde(rename = "remoteRef")] - remote_ref: RemoteRef, -} - -#[derive(Serialize)] -struct HiddenRefOutput { - success: bool, - #[serde(rename = "ref")] - ref_name: RefName, - #[serde(rename = "lfsMissing", skip_serializing_if = "Vec::is_empty")] - lfs_missing: Vec, -} - -pub(crate) async fn hidden_ref( - State(state): State>>, - headers: HeaderMap, - method: crate::Method, - body: Bytes, -) -> Result { - let actor = state.authenticate_write(&headers, &method).await?; - let input: HiddenRefInput = decode(&body)?; - let repo_did = require_hosted(&state, input.repo)?; - crate::authorize_push(&state, &actor, &repo_did, FORK_DENIED).await?; - let branch = input.remote_ref.head_ref(); - let target = input - .fork_ref - .hidden_ref(&input.remote_ref) - .ok_or_else(|| { - XrpcError::invalid_request("forkRef and remoteRef don't form a valid ref") - })?; - let sync = pull_upstream_branch( - &state, - &repo_did, - SourceRef::new(&branch), - TargetRef::new(&target), - ) - .await?; - Ok(( - http::StatusCode::OK, - Json(HiddenRefOutput { - success: true, - ref_name: target, - lfs_missing: sync.lfs_missing, - }), - ) - .into_response()) -} diff --git a/knot2/crates/knot-xrpc/src/git.rs b/knot2/crates/knot-xrpc/src/git.rs index e3f5cbd0a..52ced6c97 100644 --- a/knot2/crates/knot-xrpc/src/git.rs +++ b/knot2/crates/knot-xrpc/src/git.rs @@ -1,28 +1,16 @@ -#![allow(unused)] -use std::sync::Arc; +use http::StatusCode; -use axum::extract::State; -use http::{HeaderMap, StatusCode}; -use jacquard_axum::{ExtractXrpc, XrpcResponse}; -use jacquard_common::types::ident::AtIdentifier; -use knot_runtime::{Clock, HttpTransport}; use knot_types::{Oid, RefName, RepoDid}; -use lexicons::sh_tangled::git::keep_commit::{self, KeepCommitSource}; use crate::error::XrpcError; -use crate::{XrpcState, run_blocking}; -pub(crate) const KEEP_COMMIT_ROUTE: &str = "/xrpc/sh.tangled.git.keepCommit"; - -const KEEP_COMMIT_DENIED: &str = "only repository owner or a collaborator may execute keepCommit"; - -fn encode_ref_component(raw: &str) -> String { +pub(crate) fn encode_ref_component(raw: &str) -> String { let mut out = String::with_capacity(raw.len()); for c in raw.chars() { match c { '%' => out.push_str("%25"), - ':' => out.push_str("%3A"), - '~' => out.push_str("%7E"), + ':' => out.push_str("%3a"), + '~' => out.push_str("%7e"), other => out.push(other), } } @@ -30,13 +18,13 @@ fn encode_ref_component(raw: &str) -> String { } pub(crate) fn keep_ref( - record_did: &str, + record_did: &RepoDid, address: &knot_types::RecordAddress, commit: Oid, ) -> Result { RefName::new(format!( "refs/tngl/keep/{}/{}/{}/{}", - encode_ref_component(record_did), + encode_ref_component(record_did.as_str()), address.collection().as_str(), encode_ref_component(address.rkey().as_str()), commit, @@ -49,142 +37,52 @@ pub(crate) fn keep_ref( }) } -pub(crate) async fn keep_commit( - State(state): State>>, - headers: HeaderMap, - method: crate::Method, - ExtractXrpc(input): ExtractXrpc, -) -> Result, XrpcError> { - let actor = state.authenticate_write(&headers, &method).await?; - let repo_did = RepoDid::new(input.repo.as_str()).map_err(|_e| { - XrpcError::from_status(StatusCode::BAD_REQUEST, "repo should be valid DID") - })?; - - let (AtIdentifier::Did(record_did), Some(record_collection), Some(record_rkey)) = ( - input.record.authority(), - input.record.collection(), - input.record.rkey(), - ) else { - return Err(XrpcError::from_status( - StatusCode::BAD_REQUEST, - "record should be an AT-URI pointing to a record", - )); - }; - - crate::authorize_push(&state, &actor, &repo_did, KEEP_COMMIT_DENIED).await?; - - let commit = match input.source { - KeepCommitSource::Commit(commit) => { - let commit_id = Oid::from_hex(commit.oid.as_str()).map_err(|_e| { - XrpcError::from_status(StatusCode::BAD_REQUEST, "commit.oid must be valid OID") - })?; - if input.repo == commit.repo { - // no-op - } else { - // TODO: fetch `{commit.oid}` from `{commit.repo}` - return Err(XrpcError::from_status( - StatusCode::BAD_REQUEST, - "importing source from external repo is not supported yet.", - )); - } - - Ok(commit_id) - } - _ => Err(XrpcError::from_status( - StatusCode::BAD_REQUEST, - "only 'commit' source is supported yet.", - )), - }?; - - // set `refs/tngl/keep/{did}/{collection}/{rkey}/{oid}` to `{oid}` - let layout = state.layout.clone(); - let opened = repo_did.clone(); - let address = knot_types::RecordAddress::new( - knot_types::RecordCollection::new(record_collection.as_str()).map_err(|_e| { - XrpcError::from_status( - StatusCode::BAD_REQUEST, - "record collection isn't a valid NSID", - ) - })?, - knot_types::RecordRkey::new(record_rkey.as_str()).map_err(|_e| { - XrpcError::from_status(StatusCode::BAD_REQUEST, "record key isn't a usable rkey") - })?, - ); - let ref_name = keep_ref(record_did.as_str(), &address, commit)?; - run_blocking(move || { - let repo = layout.open(&opened)?; - repo.set_reference(&ref_name, commit)?; - Ok(()) - }) - .await?; - - Ok(XrpcResponse(keep_commit::KeepCommitOutput { - commit: commit.to_hex().into(), - ..Default::default() - })) -} - #[cfg(test)] mod tests { - use super::encode_ref_component; - use knot_types::RefName; + use super::{encode_ref_component, keep_ref}; + use knot_types::{RefName, RepoDid}; const OID: &str = "81158e200ff7e520a20b5e7505960cee4f35e4b3"; - fn keep_ref(did: &str, collection: &str, rkey: &str) -> String { - format!( - "refs/tngl/keep/{}/{}/{}/{OID}", - encode_ref_component(did), - collection, - encode_ref_component(rkey), - ) - } - - #[test] - fn keep_ref_for_a_did_and_tid_rkey_is_a_valid_ref_name() { - let name = keep_ref( - "did:plc:lpbyv7yk5vnegj63jmdwlxzx", - "sh.tangled.repo.pull", - "3ms6rk5d6gk22", - ); - assert!( - name.contains("did%3Aplc%3Alpbyv7yk5vnegj63jmdwlxzx"), - "the did's colons must be encoded, got {name}" + fn keep_ref_for( + did: &RepoDid, + collection: &str, + rkey: &str, + ) -> Result { + let address = knot_types::RecordAddress::new( + knot_types::RecordCollection::new(collection).unwrap(), + knot_types::RecordRkey::new(rkey).unwrap(), ); - assert!(RefName::new(&name).is_ok(), "{name} must be a valid ref"); + keep_ref(did, &address, knot_types::Oid::from_hex(OID).unwrap()) + .map(|name| name.as_str().to_string()) } #[test] - fn keep_ref_for_a_vouch_whose_rkey_is_itself_a_did_is_valid() { - let name = keep_ref( - "did:plc:lpbyv7yk5vnegj63jmdwlxzx", - "sh.tangled.graph.vouch", - "did:plc:zgazxbjxy2dkmgqovbvbnxwx", + fn keep_ref_for_a_did_and_tid_rkey_is_a_valid_ref_name() { + assert_eq!( + keep_ref_for( + &RepoDid::new("did:plc:squid").unwrap(), + "org.tangled.review.patch", + "3jzfcijpj2z2a" + ) + .unwrap(), + "refs/tngl/keep/did%3aplc%3asquid/org.tangled.review.patch/3jzfcijpj2z2a/81158e200ff7e520a20b5e7505960cee4f35e4b3" ); - assert!(RefName::new(&name).is_ok(), "{name} must be a valid ref"); - assert!(!name.contains(':'), "no raw colon may survive, got {name}"); } #[test] - fn tilde_in_an_rkey_is_encoded() { - let name = keep_ref("did:plc:alice", "sh.tangled.string", "a~b"); - assert!(name.contains("a%7Eb"), "got {name}"); - assert!(RefName::new(&name).is_ok(), "{name} must be a valid ref"); - } - - #[test] - fn encoding_is_injective() { - assert_ne!( - encode_ref_component("did:web:foo:bar"), - encode_ref_component("did:web:foo-bar") - ); - assert_ne!(encode_ref_component("a%3Ab"), encode_ref_component("a:b")); - assert_eq!(encode_ref_component("a%3Ab"), "a%253Ab"); + fn keep_ref_for_a_rkey_that_is_itself_a_did_is_valid() { + let name = keep_ref_for( + &RepoDid::new("did:plc:squid").unwrap(), + "org.tangled.snippet.string", + "3jzfcijpj2z2a", + ) + .unwrap(); + assert!(RefName::new(name).is_ok()); } #[test] - fn structurally_invalid_rkey_is_rejected_rather_than_reaching_git() { - let name = keep_ref("did:plc:alice", "sh.tangled.string", "a..b"); - assert!(RefName::new(&name).is_err(), "{name} must be rejected"); + fn every_reserved_byte_in_a_component_is_percent_encoded() { + assert_eq!(encode_ref_component("a:b~c%d"), "a%3ab%7ec%25d"); } } diff --git a/knot2/crates/knot-xrpc/src/imports.rs b/knot2/crates/knot-xrpc/src/imports.rs index 0694355b3..68298fd41 100644 --- a/knot2/crates/knot-xrpc/src/imports.rs +++ b/knot2/crates/knot-xrpc/src/imports.rs @@ -128,17 +128,7 @@ impl ImportJob { std::fs::write(&staged, serde_json::to_vec(self)?)?; std::fs::rename(&staged, repo_path.join(SIDECAR)) } - - pub(crate) fn content(&self) -> &'static str { - match self.state { - ImportState::Queued | ImportState::Fetching => "fetching", - ImportState::Present => "present", - ImportState::Partial => "partial", - ImportState::Failed => "failed", - } - } } - pub struct ImportQueue { wake: mpsc::UnboundedSender, processed: AtomicUsize, @@ -356,14 +346,7 @@ pub(crate) async fn run_one( let placed = repo.clone(); let _ = crate::run_blocking(move || { let git = state.layout.open(&placed)?; - crate::refrecords::sync_refs( - &state, - &git, - &placed, - &[], - None, - &knot_types::PushOptions::default(), - ); + crate::refrecords::sync_refs(&state, &git, &placed, &[]); Ok(()) }) .await; diff --git a/knot2/crates/knot-xrpc/src/legacy_admin.rs b/knot2/crates/knot-xrpc/src/legacy_admin.rs index 381aef16e..bba0d8926 100644 --- a/knot2/crates/knot-xrpc/src/legacy_admin.rs +++ b/knot2/crates/knot-xrpc/src/legacy_admin.rs @@ -18,9 +18,13 @@ use crate::blocklist::{ban_account, unban_account}; use crate::cob::Authorized; use crate::error::XrpcError; use crate::invite_codes::invite_url; -use crate::members::{SubjectInput, commit_members, offer_membership}; +use crate::members::{commit_members, offer_membership}; use crate::{XrpcState, basic_credentials, decode, enforce_pre_auth_limit}; +#[derive(serde::Deserialize)] +struct SubjectInput { + subject: knot_types::AccountDid, +} pub const ADD_MEMBER_ROUTE: &str = "/admin/addMember"; pub const CREATE_INVITE_CODE_ROUTE: &str = "/xrpc/org.tangled.temp.server.createInviteCode"; pub const TEMP_ADD_MEMBER_ROUTE: &str = "/xrpc/org.tangled.temp.server.addMember"; diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index e4c4256da..c695db47c 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -14,14 +14,13 @@ pub mod imports; pub mod invite_codes; pub mod legacy_admin; mod lfs; -mod lists; mod locks; pub mod materialize; mod members; mod merge; +pub mod migrate; mod mint; mod out; -mod patchtext; pub mod plc; mod query; mod reads; @@ -30,7 +29,6 @@ mod refrecords; mod repos; mod reservations; mod service; -mod sniff; mod social; #[cfg(test)] @@ -91,7 +89,7 @@ use knot_resource::{ AdmitGuard, Rejection, SocialBudgetBytes, SocialCharge, SubjectKey, SubjectLimiter, }; -pub(crate) const PUSH_NSID: &str = "sh.tangled.repo.push"; +pub(crate) const PUSH_NSID: &str = "org.tangled.git.receivePack"; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum ReadBudget { @@ -258,19 +256,6 @@ impl XrpcState { .map_err(map_verify_error) } - // identifies the caller when one is presented, anonymous otherwise; - // invalid presented credentials still fail the request - pub(crate) async fn authenticate_optional( - &self, - headers: &HeaderMap, - method: &Method, - ) -> Result, XrpcError> { - match headers.contains_key(AUTHORIZATION) { - true => self.authenticate(headers, method).await.map(Some), - false => Ok(None), - } - } - pub(crate) async fn authenticate_write( &self, headers: &HeaderMap, @@ -356,27 +341,14 @@ impl FromRequestParts for Method { pub fn router(state: Arc>) -> Router { let merge_routes = Router::new() - // .route(merge::MERGE_ROUTE, post(merge::merge::)) .route(merge::MERGE_CHECK_ROUTE, post(merge::merge_check::)) .layer(DefaultBodyLimit::max(state.byte_limits.patch.get())); Router::new() .merge(merge_routes) - .route(members::ADD_ROUTE, post(members::add_member::)) - .route(members::REMOVE_ROUTE, post(members::remove_member::)) .route( invite_codes::REDEEM_ROUTE, post(invite_codes::redeem_invite_code::), ) - .route(blocklist::BAN_ROUTE, post(blocklist::ban::)) - .route(blocklist::UNBAN_ROUTE, post(blocklist::unban::)) - .route( - collaborators::ADD_ROUTE, - post(collaborators::add_collaborator::), - ) - .route( - collaborators::REMOVE_ROUTE, - post(collaborators::remove_collaborator::), - ) .route( social::CREATE_RECORD_ROUTE, post(social::create_record::), @@ -387,30 +359,23 @@ pub fn router(state: Arc>) -> Router post(social::delete_record::), ) .route( - social::LIST_COB_VERSIONS_ROUTE, - get(social::list_cob_versions::), + social::LIST_RECORD_VERSIONS_ROUTE, + get(social::list_record_versions::), ) - .route(social::PULL_MERGE_ROUTE, post(social::merge_pull::)) + .route(social::MERGE_PATCH_ROUTE, post(social::merge_pull::)) .route(repos::CREATE_ROUTE, post(repos::create_repo::)) .route(repos::DELETE_ROUTE, post(repos::delete_repo::)) - .route(repos::RENAME_ROUTE, post(repos::rename_repo::)) - .route( - repos::POLICY_ROUTE, - post(repos::set_contribution_policy::), - ) .route(repos::RESERVE_ROUTE, post(repos::reserve_key::)) .route(mint::MINT_TOKEN_ROUTE, post(mint::mint_token::)) .route( - branches::SET_DEFAULT_ROUTE, - post(branches::set_default_branch::), + service::RECEIVE_PACK_ROUTE, + post(service::receive_pack::), ) .route( branches::DELETE_ROUTE, post(branches::delete_branch::), ) - .route(git::KEEP_COMMIT_ROUTE, post(git::keep_commit::)) .route(forks::SYNC_ROUTE, post(forks::fork_sync::)) - .route(forks::HIDDEN_REF_ROUTE, post(forks::hidden_ref::)) .route(reads::TREE_ROUTE, get(reads::repo_tree::)) .route(reads::LOG_ROUTE, get(reads::repo_log::)) .route(reads::BRANCHES_ROUTE, get(reads::repo_branches::)) @@ -418,28 +383,13 @@ pub fn router(state: Arc>) -> Router .route(reads::TAGS_ROUTE, get(reads::repo_tags::)) .route(reads::TAG_ROUTE, get(reads::repo_tag::)) .route(reads::BLOB_ROUTE, get(reads::repo_blob::)) + .route(reads::ENTRY_ROUTE, get(reads::repo_entry::)) .route(reads::BLAME_ROUTE, get(reads::repo_blame::)) .route(reads::DIFF_ROUTE, get(reads::repo_diff::)) - .route(reads::COMPARE_ROUTE, get(reads::repo_compare::)) .route(reads::ARCHIVE_ROUTE, get(reads::repo_archive::)) .route(reads::LANGUAGES_ROUTE, get(reads::repo_languages::)) - .route( - reads::GET_DEFAULT_BRANCH_ROUTE, - get(reads::repo_get_default_branch::), - ) - .route( - reads::DESCRIBE_REPO_ROUTE, - get(reads::repo_describe_repo::), - ) .route(reads::LIST_REFS_ROUTE, get(reads::git_list_refs::)) - .route(reads::LIST_REPOS_ROUTE, get(reads::sync_list_repos::)) - .route(lists::LIST_MEMBERS_ROUTE, get(lists::list_members::)) - .route( - lists::LIST_COLLABORATORS_ROUTE, - get(lists::list_collaborators::), - ) .route(service::VERSION_ROUTE, get(service::version::)) - .route(service::OWNER_ROUTE, get(service::owner::)) .route( "/xrpc/com.atproto.sync.getRepo", get(atproto::sync_get_repo::), diff --git a/knot2/crates/knot-xrpc/src/lists.rs b/knot2/crates/knot-xrpc/src/lists.rs deleted file mode 100644 index 0a231e17a..000000000 --- a/knot2/crates/knot-xrpc/src/lists.rs +++ /dev/null @@ -1,192 +0,0 @@ -use std::sync::Arc; - -use axum::Json; -use axum::extract::{FromRequestParts, Query, State}; -use axum::response::{IntoResponse, Response}; -use http::StatusCode; -use http::request::Parts; -use serde::{Deserialize, Serialize}; - -use knot_index::{Entry, Resolved}; -use knot_runtime::{Clock, HttpTransport}; -use knot_types::{AccountDid, KnotId, RepoDid, UnixSeconds}; - -use crate::XrpcState; -use crate::error::XrpcError; -use crate::out::rfc3339; -use crate::query::{Limit, Offset, Order, Total, ValidatedQuery, next_cursor}; - -pub(crate) const LIST_MEMBERS_ROUTE: &str = "/xrpc/sh.tangled.knot.listMembers"; -pub(crate) const LIST_COLLABORATORS_ROUTE: &str = "/xrpc/sh.tangled.repo.listCollaborators"; - -const DEFAULT_LIMIT: usize = 50; -const MAX_LIMIT: usize = 1000; - -#[derive(Deserialize)] -pub(crate) struct Paging { - #[serde(default)] - limit: Limit, - #[serde(default)] - cursor: Offset, - #[serde(default)] - order: Order, -} - -struct Window { - offset: Offset, - limit: Limit, - descending: bool, -} - -impl Paging { - fn window(self) -> Window { - Window { - offset: self.cursor, - limit: self.limit, - descending: self.order.descending(), - } - } -} - -#[derive(Deserialize)] -struct SubjectQuery { - subject: Option, -} - -fn subject_param(parts: &Parts) -> Result { - Query::::try_from_uri(&parts.uri) - .map_err(|rejection| XrpcError::invalid_request(rejection.body_text()))? - .0 - .subject - .filter(|raw| !raw.is_empty()) - .ok_or_else(|| XrpcError::invalid_request("missing subject parameter")) -} - -fn unnameable(name: &'static str, message: &'static str) -> XrpcError { - XrpcError::named(StatusCode::BAD_REQUEST, name, message) -} - -pub(crate) struct MemberSubject; - -impl FromRequestParts>> for MemberSubject { - type Rejection = XrpcError; - - async fn from_request_parts( - parts: &mut Parts, - state: &Arc>, - ) -> Result { - KnotId::new(subject_param(parts)?) - .is_ok_and(|knot| knot == state.knot_did) - .then_some(MemberSubject) - .ok_or_else(|| unnameable("InvalidSubject", "subject must be this knot's DID")) - } -} - -pub(crate) struct CollaboratorRepo(RepoDid); - -impl FromRequestParts for CollaboratorRepo { - type Rejection = XrpcError; - - async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { - RepoDid::new(subject_param(parts)?) - .map(CollaboratorRepo) - .map_err(|_| unnameable("InvalidRepo", "subject must be a repo DID")) - } -} - -#[derive(Serialize)] -struct ItemWire { - subject: AccountDid, - #[serde(rename = "addedBy")] - added_by: AccountDid, - #[serde(rename = "createdAt")] - created_at: String, -} - -#[derive(Serialize)] -struct PageWire { - items: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - cursor: Option, -} - -fn page(entries: Vec<(AccountDid, Entry)>, window: Window) -> Response { - let mut rows: Vec<(UnixSeconds, AccountDid, Entry)> = entries - .into_iter() - .map(|(subject, entry)| (entry.created_at, subject, entry)) - .collect(); - rows.sort_by(|(left, earlier, _), (right, later, _)| { - match window.descending { - true => right.cmp(left), - false => left.cmp(right), - } - .then_with(|| earlier.cmp(later)) - }); - let total = rows.len(); - let items = rows - .into_iter() - .skip(window.offset.get()) - .take(window.limit.get()) - .map(|(_, subject, entry)| ItemWire { - subject, - added_by: entry.added_by, - created_at: rfc3339(entry.created_at.get(), 0), - }) - .collect(); - let cursor = next_cursor(window.offset, window.limit, Total::new(total)); - Json(PageWire { items, cursor }).into_response() -} - -fn members_warming() -> XrpcError { - XrpcError::warming("members projection is still warming") -} - -fn collaborators_warming() -> XrpcError { - XrpcError::warming("collaborators projection is still warming") -} - -pub(crate) async fn list_members( - State(state): State>>, - _subject: MemberSubject, - ValidatedQuery(paging): ValidatedQuery, -) -> Result { - let window = paging.window(); - match state.index.member_entries() { - Resolved::Warming => Err(members_warming()), - Resolved::Ready(entries) => Ok(page(entries, window)), - } -} - -async fn collaborator_page( - state: &XrpcState, - repo: &RepoDid, - window: Window, -) -> Result { - match state.index.owner_of(repo) { - Resolved::Warming => return Err(crate::reads::warming()), - Resolved::Ready(None) => return Ok(page(Vec::new(), window)), - Resolved::Ready(Some(_)) => {} - } - let index = Arc::clone(&state.index); - let target = repo.clone(); - let entries = crate::run_blocking(move || { - index - .ensure_collaborators(&target) - .map_err(|_| collaborators_warming())?; - match index.collaborator_entries(&target) { - Resolved::Warming => Err(collaborators_warming()), - Resolved::Ready(entries) => Ok(entries), - } - }) - .await?; - Ok(page(entries, window)) -} - -pub(crate) async fn list_collaborators( - State(state): State>>, - CollaboratorRepo(repo): CollaboratorRepo, - ValidatedQuery(paging): ValidatedQuery, -) -> Result { - let window = paging.window(); - collaborator_page(&state, &repo, window).await -} diff --git a/knot2/crates/knot-xrpc/src/materialize.rs b/knot2/crates/knot-xrpc/src/materialize.rs index 6a144d87b..32540a200 100644 --- a/knot2/crates/knot-xrpc/src/materialize.rs +++ b/knot2/crates/knot-xrpc/src/materialize.rs @@ -46,6 +46,10 @@ pub fn enable_surfaces(state: &Arc>) crate::grants::sync_members(state, &meta), knot_record::grant::GrantCollection::Member, ); + crate::grants::noted( + crate::grants::sync_blocks(state, &meta), + knot_record::grant::GrantCollection::Block, + ); announce_unframed(state, &meta, &knot_did, now); } Err(error) => { @@ -68,14 +72,7 @@ pub fn enable_surfaces(state: &Arc>) ); announce_unframed(state, &git, &did, now); } - commits += crate::refrecords::sync_refs( - state, - &git, - &did, - &[], - None, - &knot_types::PushOptions::default(), - ); + commits += crate::refrecords::sync_refs(state, &git, &did, &[]); } Err(error) => { tracing::warn!(repo = did.as_str(), %error, "a collaborator chain failed to enable"); diff --git a/knot2/crates/knot-xrpc/src/merge.rs b/knot2/crates/knot-xrpc/src/merge.rs index 747dba6fe..103614045 100644 --- a/knot2/crates/knot-xrpc/src/merge.rs +++ b/knot2/crates/knot-xrpc/src/merge.rs @@ -4,39 +4,25 @@ use axum::Json; use axum::body::Bytes; use axum::extract::State; use axum::response::{IntoResponse, Response}; -#[cfg(test)] -use http::HeaderMap; use http::StatusCode; use serde::{Deserialize, Serialize}; -use knot_git::{ - ApplyError, ApplyOutcome, Conflict, ParsedFile, PatchApplier, PatchParseError, Repo, - StagedChange, is_format_patch, parse_mailbox_bounded, parse_patch_bounded, -}; -#[cfg(test)] -use knot_git::{Identity, NewCommit, RefUpdate, Staging}; +use knot_git::Repo; use knot_runtime::{Clock, HttpTransport}; -#[cfg(test)] -use knot_types::UnixSeconds; use knot_types::{ - AuthorName, BranchName, Email, Oid, OwnerDid, RefName, RepoDid, RepoName, RepoRkey, + AuthorName, BranchName, Email, Oid, OwnerDid, RecordRkey, RefName, RepoDid, RepoName, RepoPath, + RepoRkey, }; -use crate::body::Patch; -#[cfg(test)] -use crate::body::{CommitBody, CommitMessage}; +use crate::XrpcState; +use crate::decode; use crate::error::XrpcError; -#[cfg(test)] -use crate::ok_empty; use crate::query::RepoArg; -use crate::reads::{HostedRepo, open, repo_not_found, require_hosted, resolve_repo}; -use crate::{XrpcState, decode, run_blocking}; - -pub(crate) const MERGE_CHECK_ROUTE: &str = "/xrpc/sh.tangled.repo.mergeCheck"; +use crate::reads::{HostedRepo, repo_not_found, require_hosted, resolve_repo}; +pub(crate) const MERGE_CHECK_ROUTE: &str = "/xrpc/org.tangled.git.mergeCheck"; pub(crate) const CONFLICT_MESSAGE: &str = "patch can't be applied cleanly"; -#[cfg(test)] -const MERGE_RETRIES: u32 = 3; +const REPLAY_CONFLICT: &str = "changed on the target branch too"; #[derive(Debug, Clone, PartialEq, Eq)] pub struct Committer { @@ -44,134 +30,49 @@ pub struct Committer { pub email: Email, } -#[cfg(test)] -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct MergeInput { - repo: RepoDid, - patch: Patch, - branch: BranchName, - author_name: Option, - author_email: Option, - commit_message: Option, - commit_body: Option, -} - #[derive(Deserialize)] +#[serde(deny_unknown_fields)] struct MergeCheckInput { repo: RepoDid, - patch: Patch, - branch: BranchName, + patch: RecordRkey, + #[serde(default)] + branch: Option, } #[derive(Serialize)] -struct ConflictWire { - filename: String, +struct ConflictOut { + filename: RepoPath, reason: String, } #[derive(Serialize)] struct MergeCheckOutput { + #[serde(rename = "isConflicted")] is_conflicted: bool, - #[serde(skip_serializing_if = "Option::is_none")] - conflicts: Option>, + conflicts: Vec, #[serde(skip_serializing_if = "Option::is_none")] message: Option, - #[serde(skip_serializing_if = "Option::is_none")] - error: Option, } impl MergeCheckOutput { fn clean() -> Self { Self { is_conflicted: false, - conflicts: None, + conflicts: Vec::new(), message: None, - error: None, } } - fn conflicted(conflicts: Vec) -> Self { + fn conflicted(path: RepoPath) -> Self { Self { is_conflicted: true, - conflicts: Some( - conflicts - .into_iter() - .map(|conflict| ConflictWire { - filename: conflict.path, - reason: conflict.reason.as_str().to_string(), - }) - .collect(), - ), + conflicts: vec![ConflictOut { + filename: path, + reason: REPLAY_CONFLICT.to_string(), + }], message: Some(CONFLICT_MESSAGE.to_string()), - error: None, } } - - fn broken(error: String) -> Self { - Self { - is_conflicted: true, - conflicts: None, - message: None, - error: Some(error), - } - } -} - -#[cfg(test)] -struct CommitSpec { - files: Vec, - author: Option, - message: String, - change_id: Option, -} - -#[cfg(test)] -struct MailAuthor { - name: AuthorName, - email: Email, - date: String, -} - -fn parse_patches(patch: &str, max_bytes: u64) -> Result>, PatchParseError> { - if is_format_patch(patch) { - Ok(parse_mailbox_bounded(patch, max_bytes)? - .into_iter() - .map(|mail| mail.files) - .collect()) - } else { - Ok(vec![parse_patch_bounded(patch, max_bytes)?]) - } -} - -#[cfg(test)] -fn parse_specs( - patch: &str, - message: String, - author: Option, - max_bytes: u64, -) -> Result, PatchParseError> { - match is_format_patch(patch) { - true => Ok(parse_mailbox_bounded(patch, max_bytes)? - .into_iter() - .map(|mail| CommitSpec { - message: mail.commit_message(), - author: Some(MailAuthor { - name: mail.author_name, - email: mail.author_email, - date: mail.date, - }), - change_id: mail.change_id, - files: mail.files, - }) - .collect()), - false => Ok(vec![CommitSpec { - files: parse_patch_bounded(patch, max_bytes)?, - author, - message, - change_id: None, - }]), - } } pub(crate) fn resolve_by_name( @@ -189,317 +90,18 @@ pub(crate) fn branch_tip(repo: &Repo, refname: &RefName) -> Result (UnixSeconds, i32) { - let trimmed = date.trim(); - chrono::DateTime::parse_from_rfc2822(trimmed) - .or_else(|_| chrono::DateTime::parse_from_rfc3339(trimmed)) - .map(|parsed| { - ( - UnixSeconds::new(parsed.timestamp()), - parsed.offset().local_minus_utc(), - ) - }) - .unwrap_or((now, 0)) -} - -#[cfg(test)] -fn spec_identities( - spec: &CommitSpec, - fallback: &Identity, - now: UnixSeconds, -) -> (Identity, Vec<(String, Vec)>) { - let author = match &spec.author { - Some(mail) => { - let (time, offset_seconds) = mail_time(&mail.date, now); - Identity { - name: mail.name.clone(), - email: mail.email.clone(), - time, - offset_seconds, - } - } - None => fallback.clone(), - }; - let extra_headers = spec - .change_id - .iter() - .map(|change_id| { - ( - "change-id".to_string(), - change_id.as_str().as_bytes().to_vec(), - ) - }) - .collect(); - (author, extra_headers) -} - -enum StageStop { - Conflict(Vec), - Apply(ApplyError), -} - -fn stage_all( - repo: &Repo, - tip: Oid, - files: &[Vec], -) -> Result>, Vec>, ApplyError> { - let mut applier = PatchApplier::new(repo, tip); - let staged = files - .iter() - .try_fold(Vec::new(), |mut clean, files| match applier.step(files) { - Ok(ApplyOutcome::Clean(staged)) => { - clean.push(staged); - Ok(clean) - } - Ok(ApplyOutcome::Conflicted(conflicts)) => Err(StageStop::Conflict(conflicts)), - Err(error) => Err(StageStop::Apply(error)), - }); - match staged { - Ok(clean) => Ok(Ok(clean)), - Err(StageStop::Conflict(conflicts)) => Ok(Err(conflicts)), - Err(StageStop::Apply(error)) => Err(error), - } -} - -#[cfg(test)] -enum MergeAttempt { - Done { old: Oid, new: Oid }, - Conflicted(Vec), - Raced, -} - -#[cfg(test)] -enum Merged { - Done { old: Oid, new: Oid }, - Conflicted(Vec), -} - -#[cfg(test)] -fn attempt_merge( - repo: &Repo, - refname: &RefName, - specs: &[CommitSpec], - committer: &Committer, - now: UnixSeconds, -) -> Result { - if specs.iter().any(|spec| spec.message.trim().is_empty()) { - return Err(XrpcError::invalid_request("Commit message is required")); - } - let tip = branch_tip(repo, refname)?; - let files: Vec> = specs.iter().map(|spec| spec.files.clone()).collect(); - let staged = match stage_all(repo, tip, &files).map_err(XrpcError::from)? { - Ok(staged) => staged, - Err(conflicts) => return Ok(MergeAttempt::Conflicted(conflicts)), - }; - let committer_identity = Identity { - name: committer.name.clone(), - email: committer.email.clone(), - time: now, - offset_seconds: 0, - }; - let staging = Staging::new(repo).map_err(XrpcError::from)?; - let work = staging.repo(); - let base_tree = work.find_commit(tip).map_err(XrpcError::from)?.tree; - let new_tip = specs.iter().zip(staged).try_fold( - (base_tree, tip), - |(tree, parent), (spec, staged)| -> Result<(Oid, Oid), XrpcError> { - let next_tree = work - .write_staged_tree(tree, &staged) - .map_err(XrpcError::from)?; - let (author, extra_headers) = spec_identities(spec, &committer_identity, now); - let commit = work - .write_commit(&NewCommit { - tree: next_tree, - parents: vec![parent], - author, - committer: committer_identity.clone(), - message: spec.message.clone(), - extra_headers, - }) - .map_err(XrpcError::from)?; - Ok((next_tree, commit)) - }, - )?; - match repo.find_ref(refname).map_err(XrpcError::from)? { - Some(current) if current == tip => {} - _ => return Ok(MergeAttempt::Raced), - } - staging.migrate_into(repo).map_err(XrpcError::from)?; - match repo.update_ref(&RefUpdate::Update { - name: refname.clone(), - old: tip, - new: new_tip.1, - }) { - Ok(()) => Ok(MergeAttempt::Done { - old: tip, - new: new_tip.1, - }), - Err(error) => match repo.find_ref(refname) { - Ok(Some(current)) if current != tip => Ok(MergeAttempt::Raced), - _ => Err(error.into()), - }, - } -} - -#[cfg(test)] -fn merge_with_retry( - repo: &Repo, - refname: &RefName, - specs: &[CommitSpec], - committer: &Committer, - now: UnixSeconds, - attempts: u32, -) -> Result { - match attempt_merge(repo, refname, specs, committer, now)? { - MergeAttempt::Done { old, new } => Ok(Merged::Done { old, new }), - MergeAttempt::Conflicted(conflicts) => Ok(Merged::Conflicted(conflicts)), - MergeAttempt::Raced if attempts > 1 => { - merge_with_retry(repo, refname, specs, committer, now, attempts - 1) - } - MergeAttempt::Raced => Err(XrpcError::conflict("branch moved during the merge, retry")), - } -} - -#[cfg(test)] -fn merge_conflict(conflicts: &[Conflict]) -> XrpcError { - let detail = conflicts - .first() - .map(|conflict| { - format!( - "{CONFLICT_MESSAGE}: {} {}", - conflict.path, - conflict.reason.as_str() - ) - }) - .unwrap_or_else(|| CONFLICT_MESSAGE.to_string()); - XrpcError::named( - StatusCode::CONFLICT, - "MergeConflict", - format!("Merge failed due to conflicts: {detail}"), - ) -} - -#[cfg(test)] -pub(crate) async fn merge( - State(state): State>>, - headers: HeaderMap, - method: crate::Method, - body: Bytes, -) -> Result { - let actor = state.authenticate_write(&headers, &method).await?; - let input: MergeInput = decode(&body)?; - let repo_did = require_hosted(&state, input.repo.clone())?; - crate::authorize_push( - &state, - &actor, - &repo_did, - "Only repository owner or a collaborator may merge", - ) - .await?; - let refname = input.branch.head_ref(); - let committer = state.committer.clone(); - let now = state.now(); - let layout = state.layout.clone(); - let max_patch_bytes = state.byte_limits.patch_decompressed.get(); - let event_repo = repo_did.clone(); - let event_ref = refname.clone(); - - let outcome = run_blocking(move || { - let specs = parse_specs( - input.patch.as_str(), - unified_message(&input), - unified_author(&input), - max_patch_bytes, - ) - .map_err(|error| XrpcError::invalid_request(error.to_string()))?; - let repo = open(&layout, &repo_did)?; - merge_with_retry(&repo, &refname, &specs, &committer, now, MERGE_RETRIES) - }) - .await?; - - match outcome { - Merged::Conflicted(conflicts) => Err(merge_conflict(&conflicts)), - Merged::Done { old, new } => { - let repo_label = event_repo.as_str().to_string(); - if let Err(error) = crate::refrecords::seal_ref_move( - &state, - event_repo.into_did(), - event_ref, - Some(old), - new, - actor, - ) - .await - { - tracing::warn!(repo = %repo_label, %error, "Post-receive after merge failed"); - } - Ok(ok_empty()) - } - } -} - -#[cfg(test)] -fn unified_message(input: &MergeInput) -> String { - let message = input - .commit_message - .as_ref() - .map(|message| message.as_str().to_string()) - .unwrap_or_default(); - match input - .commit_body - .as_ref() - .map(|body| body.as_str()) - .filter(|body| !body.is_empty()) - { - Some(body) => format!("{message}\n\n{body}"), - None => message, - } -} - -#[cfg(test)] -fn unified_author(input: &MergeInput) -> Option { - match (input.author_name.as_ref(), input.author_email.as_ref()) { - (Some(name), Some(email)) if !name.as_str().is_empty() && !email.as_str().is_empty() => { - Some(MailAuthor { - name: name.clone(), - email: email.clone(), - date: String::new(), - }) - } - _ => None, - } -} - pub(crate) async fn merge_check( State(state): State>>, body: Bytes, ) -> Result { let input: MergeCheckInput = decode(&body)?; - let repo_did = require_hosted(&state, input.repo)?; - let refname = input.branch.head_ref(); - let layout = state.layout.clone(); - let max_patch_bytes = state.byte_limits.patch_decompressed.get(); - - let output = run_blocking(move || { - let files = match parse_patches(input.patch.as_str(), max_patch_bytes) { - Ok(files) => files, - Err(error) => return Ok(MergeCheckOutput::broken(error.to_string())), - }; - let repo = open(&layout, &repo_did)?; - let tip = branch_tip(&repo, &refname)?; - match stage_all(&repo, tip, &files) { - Ok(Ok(_)) => Ok(MergeCheckOutput::clean()), - Ok(Err(conflicts)) => Ok(MergeCheckOutput::conflicted(conflicts)), - Err(ApplyError::TooLarge) => { - Ok(MergeCheckOutput::broken(ApplyError::TooLarge.to_string())) - } - Err(ApplyError::Git(error)) => Err(error.into()), - } - }) - .await?; - - Ok(check_response(output)) + require_hosted(&state, input.repo.clone())?; + let checked = + crate::social::check_patch_merge(&state, &input.repo, &input.patch, input.branch).await?; + Ok(check_response(match checked { + crate::social::PatchMerge::Clean => MergeCheckOutput::clean(), + crate::social::PatchMerge::Conflict { path } => MergeCheckOutput::conflicted(path), + })) } fn check_response(output: MergeCheckOutput) -> Response { diff --git a/knot2/crates/knot-xrpc/src/migrate.rs b/knot2/crates/knot-xrpc/src/migrate.rs new file mode 100644 index 000000000..998876438 --- /dev/null +++ b/knot2/crates/knot-xrpc/src/migrate.rs @@ -0,0 +1,143 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use knot_git::Repo; +use knot_record::chain; +use knot_runtime::{Clock, HttpTransport}; +use knot_types::{RefName, RepoDid, UnixSeconds}; + +use crate::XrpcState; + +#[derive(Debug, thiserror::Error)] +pub enum MigrateError { + #[error(transparent)] + Cob(#[from] knot_cob::CobError), + #[error(transparent)] + Chain(#[from] chain::ChainError), + #[error(transparent)] + Git(#[from] knot_git::GitError), + #[error(transparent)] + Frames(#[from] crate::firehose::UnstampedFrames), + #[error("the signing key wouldn't load for the legacy sweep: {0}")] + Key(#[from] knot_secrets::SecretsError), +} + +const LEGACY_COLLECTIONS: [&str; 4] = [ + "sh.tangled.git.ref", + "sh.tangled.knot.member", + "sh.tangled.knot.block", + "sh.tangled.repo.collaborator", +]; + +pub fn legacy_meta_refs(meta: &Repo) -> Result { + Ok(knot_cob::migrate_legacy_refs(meta)?) +} + +pub fn legacy_state( + state: &Arc>, +) -> Result { + let meta = Repo::open(&state.meta_path)?; + let mut swept = knot_cob::migrate_legacy_refs(&meta)?; + swept += sweep_legacy_records(state, &meta, &state.knot_repo)?; + for repo in state.index.hosted_repos() { + let git = state.layout.open(&repo)?; + swept += knot_cob::migrate_legacy_refs(&git)?; + swept += normalize_keep_refs(&git)?; + swept += sweep_legacy_records(state, &git, &repo)?; + } + Ok(swept) +} + +fn sweep_legacy_records( + state: &Arc>, + git: &Repo, + did: &RepoDid, +) -> Result { + if chain::tip(git)?.is_none() { + return Ok(0); + } + let now: UnixSeconds = state.now(); + let records = chain::in_tree(BTreeMap::new()); + let collections: Vec<_> = LEGACY_COLLECTIONS + .iter() + .map(|nsid| knot_types::RecordCollection::new(*nsid).expect("literal NSID parses")) + .collect(); + let signer = state.secrets.signer(&state.knot_did)?; + let mut emit = crate::firehose::FirehoseEmit::new(&state.firehose, did, now); + let chunks = futures::executor::block_on(chain::settle( + git, + did, + &collections, + &records, + &signer, + &mut emit, + now, + ))?; + emit.finish()?; + Ok(chunks.len()) +} + +fn normalize_keep_refs(git: &Repo) -> Result { + let renamed: Vec<_> = git + .references()? + .into_iter() + .filter_map(|record| { + let modern = normalized_keep_name(&record.name)?; + Some((record.name, modern, record.target)) + }) + .collect(); + let count = renamed.len(); + git.rename_references(renamed)?; + Ok(count) +} + +fn normalized_keep_name(name: &RefName) -> Option { + const KEEP_PREFIX: &str = "refs/tngl/keep/"; + let rest = name.as_str().strip_prefix(KEEP_PREFIX)?; + let components: Vec<_> = rest + .split('/') + .map(modernized_component) + .collect::>()?; + let joined = components.join("/"); + (joined != rest) + .then(|| format!("{KEEP_PREFIX}{joined}")) + .and_then(|modern| RefName::new(modern).ok()) +} + +fn modernized_component(component: &str) -> Option { + let decoded = percent_decode(component)?; + let modern = match knot_types::TypeName::new(&decoded) { + Ok(type_name) => knot_cob::modernized_type_of(&type_name) + .map(|modern| modern.as_str().to_string()) + .unwrap_or_else(|| { + decoded + .strip_prefix("sh.tangled.") + .map(|rest| format!("org.tangled.{rest}")) + .unwrap_or(decoded.clone()) + }), + Err(_) => decoded, + }; + Some(crate::git::encode_ref_component(&modern)) +} + +fn percent_decode(raw: &str) -> Option { + let bytes = raw.as_bytes(); + let mut out = Vec::with_capacity(bytes.len()); + let mut at = 0; + while at < bytes.len() { + match bytes[at] { + b'%' => { + let pair = bytes.get(at + 1..at + 3)?; + let high = (pair[0] as char).to_digit(16)?; + let low = (pair[1] as char).to_digit(16)?; + out.push((high * 16 + low) as u8); + at += 3; + } + byte => { + out.push(byte); + at += 1; + } + } + } + String::from_utf8(out).ok() +} diff --git a/knot2/crates/knot-xrpc/src/out.rs b/knot2/crates/knot-xrpc/src/out.rs index 7cf853b15..909761a18 100644 --- a/knot2/crates/knot-xrpc/src/out.rs +++ b/knot2/crates/knot-xrpc/src/out.rs @@ -1,27 +1,8 @@ -use base64::Engine; -use base64::engine::general_purpose::STANDARD; -use knot_git::{ - BranchInfo, BranchTip, Commit, CommitChangeId, EntryKind, FilePatch, Hunk, Identity, LineOp, - PatchStatus, TagInfo, -}; -use knot_types::{AuthorName, Datetime, Email, Oid, RecordCid, TagName, UnixSeconds}; -use serde::ser::SerializeSeq; +use knot_types::{Datetime, RecordCid, UnixSeconds}; use serde::{Deserialize, Serialize, Serializer}; use std::fmt; use std::str::FromStr; -pub(crate) const ZERO_TIME: &str = "0001-01-01T00:00:00Z"; - -fn display_opt( - value: &Option, - serializer: S, -) -> Result { - match value { - Some(id) => serializer.serialize_str(id.as_str()), - None => serializer.serialize_none(), - } -} - fn zoned(seconds: i64, offset_seconds: i32) -> chrono::DateTime { let offset = chrono::FixedOffset::east_opt(offset_seconds) .unwrap_or_else(|| chrono::FixedOffset::east_opt(0).expect("Zero offset is valid")); @@ -34,10 +15,6 @@ pub(crate) fn rfc3339(seconds: i64, offset_seconds: i32) -> String { zoned(seconds, offset_seconds).to_rfc3339_opts(chrono::SecondsFormat::Secs, true) } -pub(crate) fn rfc2822(seconds: i64, offset_seconds: i32) -> String { - zoned(seconds, offset_seconds).to_rfc2822() -} - pub(crate) fn datetime_of(at: UnixSeconds) -> Datetime { Datetime::new( chrono::DateTime::from_timestamp(at.get(), 0) @@ -91,247 +68,6 @@ impl<'de> Deserialize<'de> for CidString { } } -pub(crate) struct HashBytes(pub Oid); - -impl Serialize for HashBytes { - fn serialize(&self, serializer: S) -> Result { - let bytes = self.0.object_id(); - let mut seq = serializer.serialize_seq(Some(bytes.as_bytes().len()))?; - bytes - .as_bytes() - .iter() - .try_for_each(|byte| seq.serialize_element(byte))?; - seq.end() - } -} - -pub(crate) struct Base64Bytes(Vec); - -impl Serialize for Base64Bytes { - fn serialize(&self, serializer: S) -> Result { - serializer.serialize_str(&STANDARD.encode(&self.0)) - } -} - -#[derive(Serialize)] -pub(crate) struct SignatureJson { - #[serde(rename = "Name")] - pub name: AuthorName, - #[serde(rename = "Email")] - pub email: Email, - #[serde(rename = "When")] - pub when: String, -} - -impl SignatureJson { - pub fn of(identity: &Identity) -> Self { - Self { - name: identity.name.clone(), - email: identity.email.clone(), - when: rfc3339(identity.time.get(), identity.offset_seconds), - } - } - - pub fn utc(identity: &Identity) -> Self { - Self { - name: identity.name.clone(), - email: identity.email.clone(), - when: rfc3339(identity.time.get(), 0), - } - } - - pub fn zero() -> Self { - Self { - name: AuthorName::new(""), - email: Email::new(""), - when: ZERO_TIME.to_string(), - } - } -} - -#[derive(Serialize)] -pub(crate) struct CommitJson { - pub hash: HashBytes, - pub author: SignatureJson, - pub committer: SignatureJson, - pub message: String, - pub tree: Oid, - #[serde(skip_serializing_if = "Vec::is_empty")] - pub parent_hashes: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - pub pgp_signature: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub merge_tag: Option, - #[serde( - skip_serializing_if = "Option::is_none", - serialize_with = "display_opt" - )] - pub change_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub extra_headers: Option>, - pub this: Oid, - #[serde(skip_serializing_if = "Option::is_none")] - pub parent: Option, -} - -impl CommitJson { - pub fn of(commit: &Commit) -> Self { - let extra_headers: std::collections::BTreeMap = commit - .extra_headers - .iter() - .map(|(name, value)| (name.clone(), Base64Bytes(value.clone()))) - .collect(); - Self { - hash: HashBytes(commit.id), - author: SignatureJson::of(&commit.author), - committer: SignatureJson::of(&commit.committer), - message: commit.message.clone(), - tree: commit.tree, - parent_hashes: commit.parents.iter().map(|oid| HashBytes(*oid)).collect(), - pgp_signature: commit.pgp_signature.clone(), - merge_tag: commit.merge_tag.clone(), - change_id: commit.change_id(), - extra_headers: (!extra_headers.is_empty()).then_some(extra_headers), - this: commit.id, - parent: commit.parents.first().copied(), - } - } -} - -#[derive(Serialize)] -pub(crate) struct BranchCommitJson { - #[serde(rename = "Hash")] - pub hash: HashBytes, - #[serde(rename = "Author")] - pub author: SignatureJson, - #[serde(rename = "Committer")] - pub committer: SignatureJson, - #[serde(rename = "MergeTag")] - pub merge_tag: String, - #[serde(rename = "PGPSignature")] - pub pgp_signature: String, - #[serde(rename = "Message")] - pub message: String, - #[serde(rename = "TreeHash")] - pub tree_hash: HashBytes, - #[serde(rename = "ParentHashes")] - pub parent_hashes: Vec, - #[serde(rename = "Encoding")] - pub encoding: String, - #[serde(rename = "ExtraHeaders")] - pub extra_headers: Option<()>, -} - -#[derive(Serialize)] -pub(crate) struct Reference { - pub name: String, - pub hash: Oid, -} - -#[derive(Serialize)] -pub(crate) struct BranchJson { - pub reference: Reference, - pub commit: BranchCommitJson, - #[serde(skip_serializing_if = "std::ops::Not::not")] - pub is_default: bool, -} - -impl BranchJson { - pub fn of(branch: &BranchInfo, is_default: bool, absent: Oid) -> Self { - Self { - reference: Reference { - name: branch.name.to_string(), - hash: branch.tip.id(), - }, - commit: match &branch.tip { - BranchTip::Commit(commit) => BranchCommitJson { - hash: HashBytes(commit.id), - author: SignatureJson::utc(&commit.author), - committer: SignatureJson::utc(&commit.committer), - merge_tag: String::new(), - pgp_signature: String::new(), - message: commit.message.trim_end().to_string(), - tree_hash: HashBytes(commit.tree), - parent_hashes: commit.parents.iter().map(|oid| HashBytes(*oid)).collect(), - encoding: String::new(), - extra_headers: None, - }, - BranchTip::Opaque { id, message, .. } => BranchCommitJson { - hash: HashBytes(*id), - author: SignatureJson::zero(), - committer: SignatureJson::zero(), - merge_tag: String::new(), - pgp_signature: String::new(), - message: message.trim_end().to_string(), - tree_hash: HashBytes(absent), - parent_hashes: Vec::new(), - encoding: String::new(), - extra_headers: None, - }, - }, - is_default, - } - } -} - -#[derive(Serialize)] -pub(crate) struct TagObjectJson { - #[serde(rename = "Hash")] - pub hash: HashBytes, - #[serde(rename = "Name")] - pub name: TagName, - #[serde(rename = "Tagger")] - pub tagger: SignatureJson, - #[serde(rename = "Message")] - pub message: String, - #[serde(rename = "PGPSignature")] - pub pgp_signature: String, - #[serde(rename = "TargetType")] - pub target_type: i8, - #[serde(rename = "Target")] - pub target: HashBytes, -} - -#[derive(Serialize)] -pub(crate) struct TagJson { - pub name: TagName, - pub hash: Oid, - #[serde(skip_serializing_if = "Option::is_none")] - pub tag: Option, - #[serde(skip_serializing_if = "String::is_empty")] - pub message: String, -} - -const TARGET_TYPE_TAG: i8 = 4; - -impl TagJson { - pub fn of(info: &TagInfo) -> Self { - let message = recombine_message(&info.message); - let tag = - info.annotated.as_ref().map(|annotated| TagObjectJson { - hash: HashBytes(info.id), - name: info.name.clone(), - tagger: annotated.tagger.as_ref().map(SignatureJson::utc).unwrap_or( - SignatureJson { - name: AuthorName::new(""), - email: Email::new(""), - when: rfc3339(0, 0), - }, - ), - message: message.clone(), - pgp_signature: annotated.pgp_signature.clone().unwrap_or_default(), - target_type: TARGET_TYPE_TAG, - target: HashBytes(annotated.target), - }); - Self { - name: info.name.clone(), - hash: info.id, - tag, - message, - } - } -} - pub(crate) fn fold_subject(message: &str) -> String { message .split("\n\n") @@ -344,315 +80,9 @@ pub(crate) fn fold_subject(message: &str) -> String { .join(" ") } -pub(crate) fn message_body(message: &str) -> String { - message - .split_once("\n\n") - .map(|(_, body)| body.trim_matches('\n').to_string()) - .unwrap_or_default() -} - -fn recombine_message(message: &str) -> String { - let subject = fold_subject(message); - let body = message_body(message); - match (subject.is_empty(), body.is_empty()) { - (_, true) => subject, - (true, false) => body, - (false, false) => format!("{subject}\n\n{body}"), - } -} - -#[derive(Serialize)] -pub(crate) struct LineJson { - #[serde(rename = "Op")] - pub op: u8, - #[serde(rename = "Line")] - pub line: String, -} - -#[derive(Serialize)] -pub(crate) struct TextFragmentJson { - #[serde(rename = "Comment")] - pub comment: String, - #[serde(rename = "OldPosition")] - pub old_position: i64, - #[serde(rename = "OldLines")] - pub old_lines: i64, - #[serde(rename = "NewPosition")] - pub new_position: i64, - #[serde(rename = "NewLines")] - pub new_lines: i64, - #[serde(rename = "LinesAdded")] - pub lines_added: i64, - #[serde(rename = "LinesDeleted")] - pub lines_deleted: i64, - #[serde(rename = "LeadingContext")] - pub leading_context: i64, - #[serde(rename = "TrailingContext")] - pub trailing_context: i64, - #[serde(rename = "Lines")] - pub lines: Vec, -} - -impl TextFragmentJson { - pub fn of(hunk: &Hunk) -> Self { - let lines: Vec = hunk - .lines - .iter() - .map(|line| LineJson { - op: match line.op { - LineOp::Context => 0, - LineOp::Delete => 1, - LineOp::Add => 2, - }, - line: String::from_utf8_lossy(&line.text).into_owned(), - }) - .collect(); - let leading = lines.iter().take_while(|line| line.op == 0).count(); - let trailing = if leading == lines.len() { - 0 - } else { - lines.iter().rev().take_while(|line| line.op == 0).count() - }; - Self { - comment: String::new(), - old_position: hunk.old_start.get() as i64, - old_lines: hunk.old_lines.get() as i64, - new_position: hunk.new_start.get() as i64, - new_lines: hunk.new_lines.get() as i64, - lines_added: hunk.added().get() as i64, - lines_deleted: hunk.deleted().get() as i64, - leading_context: leading as i64, - trailing_context: trailing as i64, - lines, - } - } -} - -#[derive(Serialize)] -pub(crate) struct DiffNameJson { - pub old: String, - pub new: String, -} - -#[derive(Serialize)] -pub(crate) struct DiffJson { - pub name: DiffNameJson, - pub text_fragments: Option>, - pub is_binary: bool, - pub is_new: bool, - pub is_delete: bool, - pub is_copy: bool, - pub is_rename: bool, -} - -impl DiffJson { - pub fn of(patch: &FilePatch) -> Self { - let fragments: Vec = - patch.hunks().iter().map(TextFragmentJson::of).collect(); - Self { - name: DiffNameJson { - old: match patch.status { - PatchStatus::Added => String::new(), - _ => patch.path.to_string(), - }, - new: match patch.status { - PatchStatus::Deleted => String::new(), - _ => patch.path.to_string(), - }, - }, - text_fragments: (!fragments.is_empty()).then_some(fragments), - is_binary: patch.is_binary(), - is_new: patch.status == PatchStatus::Added, - is_delete: patch.status == PatchStatus::Deleted, - is_copy: false, - is_rename: false, - } - } -} - -#[derive(Serialize)] -pub(crate) struct DiffStatJson { - pub insertions: i64, - pub deletions: i64, - pub files_changed: i64, -} - -#[derive(Serialize)] -pub(crate) struct NiceDiffJson { - pub commit: CommitJson, - pub stat: DiffStatJson, - pub diff: Option>, -} - -pub(crate) fn nice_diff(commit: &Commit, patches: &[FilePatch]) -> NiceDiffJson { - let diffs: Vec = patches.iter().map(DiffJson::of).collect(); - let stat = DiffStatJson { - insertions: patches - .iter() - .flat_map(|patch| patch.hunks().iter()) - .map(|hunk| hunk.added().get() as i64) - .sum(), - deletions: patches - .iter() - .flat_map(|patch| patch.hunks().iter()) - .map(|hunk| hunk.deleted().get() as i64) - .sum(), - files_changed: patches.len() as i64, - }; - NiceDiffJson { - commit: CommitJson::of(commit), - stat, - diff: (!diffs.is_empty()).then_some(diffs), - } -} - -#[derive(Serialize)] -pub(crate) struct PatchIdentityJson { - #[serde(rename = "Name")] - pub name: AuthorName, - #[serde(rename = "Email")] - pub email: Email, -} - -#[derive(Serialize)] -pub(crate) struct FormatPatchJson { - #[serde(rename = "Files")] - pub files: Option>, - #[serde(rename = "SHA")] - pub sha: Oid, - #[serde(rename = "Author")] - pub author: Option, - #[serde(rename = "AuthorDate")] - pub author_date: String, - #[serde(rename = "Committer")] - pub committer: Option<()>, - #[serde(rename = "CommitterDate")] - pub committer_date: String, - #[serde(rename = "Title")] - pub title: String, - #[serde(rename = "Body")] - pub body: String, - #[serde(rename = "SubjectPrefix")] - pub subject_prefix: String, - #[serde(rename = "BodyAppendix")] - pub body_appendix: String, - #[serde(rename = "RawHeaders")] - pub raw_headers: Option>>, - #[serde(rename = "Raw")] - pub raw: String, -} - -pub(crate) fn normalize_message_section<'a>(lines: impl Iterator) -> String { - lines - .map(str::trim_end) - .fold((String::new(), 0usize), |(mut out, blanks), line| { - if line.is_empty() { - return (out, blanks + 1); - } - if !out.is_empty() { - out.push('\n'); - if blanks > 0 { - out.push('\n'); - } - } - out.push_str(line); - (out, 0) - }) - .0 -} - -fn entry_mode_decimal(kind: EntryKind) -> u32 { - match kind { - EntryKind::Tree => 0o040000, - EntryKind::Blob => 0o100644, - EntryKind::BlobExecutable => 0o100755, - EntryKind::Link => 0o120000, - EntryKind::Commit => 0o160000, - } -} - -pub(crate) fn entry_mode_octal(kind: EntryKind) -> String { - format!("{:06o}", entry_mode_decimal(kind)) -} - -#[derive(Serialize)] -pub(crate) struct FileJson { - #[serde(rename = "OldName")] - pub old_name: String, - #[serde(rename = "NewName")] - pub new_name: String, - #[serde(rename = "IsNew")] - pub is_new: bool, - #[serde(rename = "IsDelete")] - pub is_delete: bool, - #[serde(rename = "IsCopy")] - pub is_copy: bool, - #[serde(rename = "IsRename")] - pub is_rename: bool, - #[serde(rename = "OldMode")] - pub old_mode: u32, - #[serde(rename = "NewMode")] - pub new_mode: u32, - #[serde(rename = "OldOIDPrefix")] - pub old_oid_prefix: String, - #[serde(rename = "NewOIDPrefix")] - pub new_oid_prefix: String, - #[serde(rename = "Score")] - pub score: i64, - #[serde(rename = "TextFragments")] - pub text_fragments: Option>, - #[serde(rename = "IsBinary")] - pub is_binary: bool, - #[serde(rename = "BinaryFragment")] - pub binary_fragment: Option<()>, - #[serde(rename = "ReverseBinaryFragment")] - pub reverse_binary_fragment: Option<()>, -} - -impl FileJson { - pub fn of(patch: &FilePatch) -> Self { - let fragments: Vec = - patch.hunks().iter().map(TextFragmentJson::of).collect(); - let same_mode = patch.old_kind.is_some() && patch.old_kind == patch.new_kind; - Self { - old_name: match patch.status { - PatchStatus::Added => String::new(), - _ => patch.path.to_string(), - }, - new_name: match patch.status { - PatchStatus::Deleted => String::new(), - _ => patch.path.to_string(), - }, - is_new: patch.status == PatchStatus::Added, - is_delete: patch.status == PatchStatus::Deleted, - is_copy: false, - is_rename: false, - old_mode: match patch.status { - PatchStatus::Added => 0, - PatchStatus::Deleted | PatchStatus::Modified => { - patch.old_kind.map(entry_mode_decimal).unwrap_or(0) - } - }, - new_mode: match patch.status { - PatchStatus::Added => patch.new_kind.map(entry_mode_decimal).unwrap_or(0), - PatchStatus::Deleted => 0, - PatchStatus::Modified if same_mode => 0, - PatchStatus::Modified => patch.new_kind.map(entry_mode_decimal).unwrap_or(0), - }, - old_oid_prefix: patch.old_oid.to_hex(), - new_oid_prefix: patch.new_oid.to_hex(), - score: 0, - text_fragments: (!fragments.is_empty()).then_some(fragments), - is_binary: patch.is_binary(), - binary_fragment: None, - reverse_binary_fragment: None, - } - } -} - #[cfg(test)] mod tests { - use super::{rfc2822, rfc3339}; + use super::rfc3339; const A_JUNE_INSTANT: i64 = 1_717_236_600; @@ -662,10 +92,4 @@ mod tests { assert!(rfc3339(A_JUNE_INSTANT, -18000).ends_with("-05:00")); assert!(rfc3339(A_JUNE_INSTANT, 0).ends_with('Z')); } - - #[test] - fn rfc2822_keeps_the_signed_offset() { - assert!(rfc2822(A_JUNE_INSTANT, 7200).ends_with("+0200")); - assert!(rfc2822(A_JUNE_INSTANT, -18000).ends_with("-0500")); - } } diff --git a/knot2/crates/knot-xrpc/src/patchtext.rs b/knot2/crates/knot-xrpc/src/patchtext.rs deleted file mode 100644 index 1b880bef8..000000000 --- a/knot2/crates/knot-xrpc/src/patchtext.rs +++ /dev/null @@ -1,208 +0,0 @@ -use knot_git::{ - BinaryDiff, Commit, EntryKind, FilePatch, Hunk, LineCount, LineNumber, LineOp, PatchBody, - PatchStatus, quote_path, -}; - -use crate::out::{entry_mode_octal, fold_subject, message_body, rfc2822}; - -const GRAPH_WIDTH: usize = 60; - -fn entry_mode(kind: Option) -> String { - kind.map(entry_mode_octal).unwrap_or_default() -} - -fn span(start: LineNumber, lines: LineCount) -> String { - match lines.get() { - 1 => format!("{}", start.get()), - _ => format!("{},{}", start.get(), lines.get()), - } -} - -fn render_hunk(out: &mut String, hunk: &Hunk) { - out.push_str(&format!( - "@@ -{} +{} @@\n", - span(hunk.old_start, hunk.old_lines), - span(hunk.new_start, hunk.new_lines) - )); - hunk.lines.iter().for_each(|line| { - out.push(match line.op { - LineOp::Context => ' ', - LineOp::Delete => '-', - LineOp::Add => '+', - }); - out.push_str(&String::from_utf8_lossy(&line.text)); - if !line.text.ends_with(b"\n") { - out.push_str("\n\\ No newline at end of file\n"); - } - }); -} - -fn render_file(out: &mut String, patch: &FilePatch) { - let old_side = quote_path(&format!("a/{}", patch.path)); - let new_side = quote_path(&format!("b/{}", patch.path)); - let index = format!( - "index {}..{}", - patch.old_oid.to_hex(), - patch.new_oid.to_hex() - ); - out.push_str(&format!("diff --git {old_side} {new_side}\n")); - match patch.status { - PatchStatus::Added => out.push_str(&format!( - "new file mode {}\n{index}\n", - entry_mode(patch.new_kind) - )), - PatchStatus::Deleted => out.push_str(&format!( - "deleted file mode {}\n{index}\n", - entry_mode(patch.old_kind) - )), - PatchStatus::Modified if patch.old_kind == patch.new_kind => { - out.push_str(&format!("{index} {}\n", entry_mode(patch.old_kind))) - } - PatchStatus::Modified => { - out.push_str(&format!( - "old mode {}\nnew mode {}\n", - entry_mode(patch.old_kind), - entry_mode(patch.new_kind) - )); - match patch.old_oid == patch.new_oid { - true => {} - false => out.push_str(&format!("{index}\n")), - } - } - } - let old_label = match patch.status { - PatchStatus::Added => "/dev/null", - _ => old_side.as_str(), - }; - let new_label = match patch.status { - PatchStatus::Deleted => "/dev/null", - _ => new_side.as_str(), - }; - match &patch.body { - PatchBody::Binary(BinaryDiff::Encoded { text, .. }) => out.push_str(text), - PatchBody::Binary(BinaryDiff::Omitted(_)) => out.push_str(&format!( - "Binary files {old_label} and {new_label} differ\n" - )), - PatchBody::Binary(BinaryDiff::Unchanged(_)) => {} - PatchBody::Text(hunks) if hunks.is_empty() => {} - PatchBody::Text(hunks) => { - out.push_str(&format!("--- {old_label}\n+++ {new_label}\n")); - hunks.iter().for_each(|hunk| render_hunk(out, hunk)); - } - } -} - -pub(crate) fn render_patches(patches: &[FilePatch]) -> String { - patches.iter().fold(String::new(), |mut out, patch| { - render_file(&mut out, patch); - out - }) -} - -fn stat_counts(patch: &FilePatch) -> (usize, usize) { - patch.hunks().iter().fold((0, 0), |(added, deleted), hunk| { - ( - added + hunk.added().get() as usize, - deleted + hunk.deleted().get() as usize, - ) - }) -} - -fn graph(added: usize, deleted: usize) -> String { - let total = added + deleted; - let (added, deleted) = if total > GRAPH_WIDTH { - (added * GRAPH_WIDTH / total, deleted * GRAPH_WIDTH / total) - } else { - (added, deleted) - }; - format!("{}{}", "+".repeat(added), "-".repeat(deleted)) -} - -fn diffstat(patches: &[FilePatch]) -> String { - let named: Vec<(String, &FilePatch)> = patches - .iter() - .map(|patch| (quote_path(patch.path.as_str()), patch)) - .collect(); - let width = named.iter().map(|(name, _)| name.len()).max().unwrap_or(0); - let rows: String = named - .iter() - .map(|(name, patch)| match &patch.body { - PatchBody::Binary(BinaryDiff::Unchanged(_)) => format!(" {name: { - let sizes = binary.sizes(); - format!( - " {name: {} bytes\n", - sizes.old, sizes.new - ) - } - PatchBody::Text(_) => { - let (added, deleted) = stat_counts(patch); - let total = added + deleted; - format!(" {name: 0 { - summary.push_str(&format!( - ", {added} insertion{}(+)", - if added == 1 { "" } else { "s" } - )); - } - if deleted > 0 { - summary.push_str(&format!( - ", {deleted} deletion{}(-)", - if deleted == 1 { "" } else { "s" } - )); - } - summary.push('\n'); - let modes: String = named - .iter() - .filter_map(|(name, patch)| { - let (old, new) = (entry_mode(patch.old_kind), entry_mode(patch.new_kind)); - match patch.status { - PatchStatus::Added => Some(format!(" create mode {new} {name}\n")), - PatchStatus::Deleted => Some(format!(" delete mode {old} {name}\n")), - PatchStatus::Modified if patch.old_kind != patch.new_kind => { - Some(format!(" mode change {old} => {new} {name}\n")) - } - PatchStatus::Modified => None, - } - }) - .collect(); - format!("{rows}{summary}{modes}") -} - -pub(crate) fn render_format_patch(commit: &Commit, patches: &[FilePatch]) -> String { - let subject = fold_subject(&commit.message); - let body = message_body(&commit.message); - let mut out = format!("From {} Mon Sep 17 00:00:00 2001\n", commit.id.to_hex()); - out.push_str(&format!( - "From: {} <{}>\n", - commit.author.name, commit.author.email - )); - out.push_str(&format!( - "Date: {}\n", - rfc2822(commit.author.time.get(), commit.author.offset_seconds) - )); - out.push_str(&format!("Subject: [PATCH] {subject}\n")); - if let Some(change_id) = commit.change_id() { - out.push_str(&format!("Change-Id: {change_id}\n")); - } - out.push('\n'); - if !body.is_empty() { - out.push_str(&body); - out.push('\n'); - } - out.push_str("---\n"); - out.push_str(&diffstat(patches)); - out.push('\n'); - out.push_str(&render_patches(patches)); - out.push_str("-- \nknot\n\n"); - out -} diff --git a/knot2/crates/knot-xrpc/src/query.rs b/knot2/crates/knot-xrpc/src/query.rs index d2bc12bc0..72393a58e 100644 --- a/knot2/crates/knot-xrpc/src/query.rs +++ b/knot2/crates/knot-xrpc/src/query.rs @@ -120,51 +120,11 @@ pub(crate) fn next_cursor( .filter(|&end| end < total.get()) .map(Cursor::new) } - -#[derive(Clone, Copy, Default, PartialEq, Eq)] -pub(crate) enum Order { - #[default] - Desc, - Asc, -} - -impl Order { - pub(crate) fn descending(self) -> bool { - matches!(self, Order::Desc) - } -} - -impl<'de> Deserialize<'de> for Order { - fn deserialize>(deserializer: D) -> Result { - match String::deserialize(deserializer)?.as_str() { - "" | "desc" => Ok(Order::Desc), - "asc" => Ok(Order::Asc), - _ => Err(de::Error::custom("order must be 'asc' or 'desc'")), - } - } -} - pub(crate) enum RepoArg { Did(RepoDid), OwnerRkey { owner: OwnerDid, rkey: RepoRkey }, } -impl RepoArg { - pub(crate) fn basename(&self) -> &str { - match self { - RepoArg::Did(did) => did.as_str(), - RepoArg::OwnerRkey { rkey, .. } => rkey.as_str(), - } - } - - pub(crate) fn to_param(&self) -> String { - match self { - RepoArg::Did(did) => did.as_str().to_string(), - RepoArg::OwnerRkey { owner, rkey } => format!("{}/{}", owner.as_str(), rkey.as_str()), - } - } -} - impl<'de> Deserialize<'de> for RepoArg { fn deserialize>(deserializer: D) -> Result { let raw = String::deserialize(deserializer)?; @@ -294,18 +254,3 @@ impl<'de> Deserialize<'de> for TreePath { }) } } - -#[derive(Default)] -pub(crate) struct RawFlag(bool); - -impl RawFlag { - pub(crate) fn requested(&self) -> bool { - self.0 - } -} - -impl<'de> Deserialize<'de> for RawFlag { - fn deserialize>(deserializer: D) -> Result { - Ok(RawFlag(String::deserialize(deserializer)? == "true")) - } -} diff --git a/knot2/crates/knot-xrpc/src/reads.rs b/knot2/crates/knot-xrpc/src/reads.rs index 00bb637dc..81381b3c7 100644 --- a/knot2/crates/knot-xrpc/src/reads.rs +++ b/knot2/crates/knot-xrpc/src/reads.rs @@ -100,7 +100,7 @@ fn blame_timeout() -> XrpcError { fn names_reserved(refspec: &str) -> bool { match RefClass::of_revspec(refspec) { - RefClass::Cob | RefClass::Checkpoint | RefClass::Atproto => true, + RefClass::Cob | RefClass::Checkpoint | RefClass::Atproto | RefClass::Tngl => true, RefClass::Public | RefClass::Hidden => false, } } diff --git a/knot2/crates/knot-xrpc/src/refrecords.rs b/knot2/crates/knot-xrpc/src/refrecords.rs index c8f51d73e..f76aa7cc4 100644 --- a/knot2/crates/knot-xrpc/src/refrecords.rs +++ b/knot2/crates/knot-xrpc/src/refrecords.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use knot_git::{RefUpdate, Repo}; use knot_record::chain; -use knot_record::git_ref::{self, GIT_REF_COLLECTION}; use knot_record::git_refs; use knot_runtime::{Clock, HttpTransport}; use knot_types::{AccountDid, Oid, PushOptions, RefName, RepoDid}; @@ -16,8 +15,6 @@ pub fn sync_refs( git: &Repo, did: &RepoDid, applied: &[RefUpdate], - editor: Option<&AccountDid>, - push_options: &PushOptions, ) -> usize { let _guard = state.cob_locks.repo(did); let now = state.now(); @@ -28,17 +25,6 @@ pub fn sync_refs( return 0; } }; - let records = match git_ref::desired_records(git, applied, editor, push_options.as_slice()) { - Ok(records) => records, - Err(error) => { - tracing::error!( - repo = did.as_str(), - %error, - "the ref record set wouldn't build, a restart resyncs it" - ); - return 0; - } - }; let refs_self = match git_refs::desired_record(git) { Ok(record) => record, Err(error) => { @@ -52,39 +38,23 @@ pub fn sync_refs( }; let mut emit = firehose::FirehoseEmit::new(&state.firehose, did, now); let chunks = match futures::executor::block_on(async { - match (chain::tip(git), records.is_empty() && refs_self.is_empty()) { + match (chain::tip(git), refs_self.is_empty()) { (Ok(None), true) => Ok(Vec::new()), (Ok(None), false) => { - let seeded = chain::in_tree(records.into_iter().chain(refs_self).collect()); + let seeded = chain::in_tree(refs_self); chain::materialize(git, did, &seeded, &signer, &mut emit, now).await } (Ok(Some(_)), _) => { - let mut chunks = chain::sync( + chain::sync( git, did, - std::slice::from_ref( - &knot_types::RecordCollection::new(GIT_REF_COLLECTION) - .expect("Git ref collection is valid nsid"), - ), - &chain::in_tree(records), + std::slice::from_ref(&git_refs::git_refs_collection()), + &chain::in_tree(refs_self), &signer, &mut emit, now, ) - .await?; - chunks.extend( - chain::sync( - git, - did, - std::slice::from_ref(&git_refs::git_refs_collection()), - &chain::in_tree(refs_self), - &signer, - &mut emit, - now, - ) - .await?, - ); - Ok(chunks) + .await } (Err(error), _) => Err(error), } @@ -126,7 +96,7 @@ pub(crate) async fn seal_ref_move( refname: RefName, old: Option, new: Oid, - actor: AccountDid, + _actor: AccountDid, ) -> Result<(), XrpcError> { let layout = state.layout.clone(); let state = Arc::clone(state); @@ -140,14 +110,7 @@ pub(crate) async fn seal_ref_move( }, None => RefUpdate::Create { name: refname, new }, }; - sync_refs( - &state, - &repo, - &did, - std::slice::from_ref(&update), - Some(&actor), - &PushOptions::default(), - ); + sync_refs(&state, &repo, &did, std::slice::from_ref(&update)); Ok(()) }) .await @@ -163,14 +126,7 @@ pub(crate) async fn sync_after( let target = did.clone(); if let Err(error) = run_blocking(move || -> Result<(), XrpcError> { let repo = layout.open(&target)?; - sync_refs( - &state, - &repo, - &target, - std::slice::from_ref(&update), - None, - &PushOptions::default(), - ); + sync_refs(&state, &repo, &target, std::slice::from_ref(&update)); Ok(()) }) .await @@ -188,7 +144,7 @@ pub(crate) async fn sync_default_branch( let target = did.clone(); if let Err(error) = run_blocking(move || -> Result<(), XrpcError> { let repo = layout.open(&target)?; - sync_refs(&state, &repo, &target, &[], None, &PushOptions::default()); + sync_refs(&state, &repo, &target, &[]); Ok(()) }) .await @@ -210,7 +166,7 @@ impl knot_receive::RefSurface for RefProjection firehose::publish_push( &self.state.firehose, diff --git a/knot2/crates/knot-xrpc/src/repos.rs b/knot2/crates/knot-xrpc/src/repos.rs index 5f84ea29b..f1993f141 100644 --- a/knot2/crates/knot-xrpc/src/repos.rs +++ b/knot2/crates/knot-xrpc/src/repos.rs @@ -29,10 +29,8 @@ use crate::reservations::ReserveDecision; use crate::{XrpcState, decode, ok_empty, run_blocking}; pub(crate) const CREATE_ROUTE: &str = "/xrpc/org.tangled.repo.createRepo"; -pub(crate) const DELETE_ROUTE: &str = "/xrpc/sh.tangled.repo.delete"; -pub(crate) const RENAME_ROUTE: &str = "/xrpc/sh.tangled.repo.rename"; -pub(crate) const RESERVE_ROUTE: &str = "/xrpc/sh.tangled.repo.reserveKey"; -pub(crate) const POLICY_ROUTE: &str = "/xrpc/sh.tangled.repo.setContributionPolicy"; +pub(crate) const DELETE_ROUTE: &str = "/xrpc/org.tangled.repo.deleteRepo"; +pub(crate) const RESERVE_ROUTE: &str = "/xrpc/org.tangled.repo.reserveKey"; #[derive(Deserialize)] struct CreateInput { @@ -60,13 +58,6 @@ struct DeleteInput { force: bool, } -#[derive(Deserialize)] -struct RenameInput { - repo: RepoDid, - rkey: RepoRkey, - name: RepoName, -} - #[derive(Deserialize)] struct ReserveInput { #[serde(rename = "repoDid")] @@ -164,7 +155,10 @@ pub(crate) async fn create_repo( .await?; let input: CreateInput = decode(&body)?; - let head = input.default_branch.map(|branch| branch.head_ref()); + let head = input + .default_branch + .as_ref() + .map(|branch| branch.head_ref()); RepoRkey::new(input.name.as_str()) .map_err(|error| XrpcError::invalid_request(error.to_string()))?; let rkey = RepoRkey::new(input.name.as_str().to_lowercase()) @@ -174,6 +168,8 @@ pub(crate) async fn create_repo( None => None, }; let meta = ManifestMeta { + contribution_policy: Some(state.contribution_policy), + default_branch: input.default_branch.clone(), spindle: input.spindle, description: input.description, website: input.website, @@ -388,37 +384,6 @@ pub(crate) async fn create_repo( &mut emit, now, ))?; - let drift = knot_record::git_ref::desired_records(&git, &[], None, &[]) - .inspect_err(|error| { - tracing::warn!(repo = %did, %error, "a fork's seeded ref set wouldn't build; the boot pass or its first push resyncs it") - }) - .ok() - .filter(|records| !records.is_empty()); - let tail = match drift { - Some(drift) => { - match futures::executor::block_on(knot_record::chain::sync( - &git, - &did, - std::slice::from_ref( - &knot_types::RecordCollection::new( - knot_record::git_ref::GIT_REF_COLLECTION, - ) - .expect("Git ref collection is valid nsid"), - ), - &knot_record::chain::in_tree(drift), - &knot_signer, - &mut emit, - now, - )) { - Ok(chunks) => chunks.into_iter().last().unwrap_or(tail), - Err(error) => { - tracing::warn!(repo = %did, %error, "a fork's seeded refs wouldn't project at birth; the boot pass or its first push resyncs them"); - tail - } - } - } - None => tail, - }; let tail = match knot_record::git_refs::desired_record(&git) { Ok(record) => match futures::executor::block_on(knot_record::chain::sync( &git, @@ -601,7 +566,7 @@ async fn provision_repo_did( } if !state.reservations.holder_is(&did, actor, state.now()) { return Err(XrpcError::invalid_request( - "reserve this repoDid for your own account via sh.tangled.repo.reserveKey before creating it", + "reserve this repoDid for your own account via org.tangled.repo.reserveKey before creating it", )); } let knot_public = state.secrets.public_key(&state.knot_did)?; @@ -711,7 +676,7 @@ pub(crate) async fn delete_repo( match state.atproto.repo_record_present(&owner, &rkey).await { Ok(RecordPresence::Present) => { return Err(XrpcError::conflict( - "sh.tangled.repo record still exists on the owner's PDS. Remove it there first or force the delete.", + "org.tangled.repo.declaration record still exists on the owner's PDS. Remove it there first or force the delete.", )); } Ok(RecordPresence::Absent) => {} @@ -793,153 +758,60 @@ fn deregister( } } -pub(crate) async fn rename_repo( - State(state): State>>, - headers: HeaderMap, - method: crate::Method, - body: Bytes, -) -> Result { - let actor = state.authenticate_write(&headers, &method).await?; - let RenameInput { repo, rkey, name } = decode(&body)?; - - let owner = match state.index.owner_of(&repo) { - Resolved::Ready(Some(owner)) => owner, - Resolved::Ready(None) => { - return Err(XrpcError::not_found("no such repository on this knot")); - } - Resolved::Warming => { - return Err(XrpcError::warming( - "registry projection is still warming, please retry shortly", - )); - } - }; - - crate::authorize_push( - &state, - &actor, - &repo, - "only repository owner or a collaborator may rename it", +pub(crate) fn rename_in_registry( + meta: &Repo, + home: &CobHome, + owner: OwnerDid, + repo: &RepoDid, + rkey: RepoRkey, + name: RepoName, + signer: &dyn Signer, + now: UnixSeconds, +) -> Result<(), XrpcError> { + let store = CobStore::new(meta); + let object = registry_object(&store)? + .ok_or_else(|| XrpcError::not_found("no repositories are registered on this knot"))?; + knot_cobs::rename_repo( + &store, + home, + object, + Rename { + owner, + rkey, + name, + repo: repo.clone(), + }, + signer, + now, ) - .await?; - - let now = state.now(); - let knot_signer = state.secrets.signer(&state.knot_did)?; - let index = Arc::clone(&state.index); - let cob_locks = Arc::clone(&state.cob_locks); - let meta_path = state.meta_path.clone(); - let renamed = repo.clone(); - let home = CobHome::from(&state.knot_did); - run_blocking(move || { - { - let _guard = cob_locks.meta(); - let meta = Repo::open(&meta_path)?; - let store = CobStore::new(&meta); - let object = registry_object(&store)?.ok_or_else(|| { - XrpcError::not_found("no repositories are registered on this knot") - })?; - knot_cobs::rename_repo( - &store, - &home, - object, - Rename { - owner, - rkey, - name, - repo: renamed.clone(), - }, - &knot_signer, - now, - ) - .map(|_| ()) - .map_err(XrpcError::from)?; - } - index.refresh_registry().map_err(|error| { - XrpcError::internal(format!( - "repo {renamed} was renamed but registry projection refresh failed: {error}" - )) - }) - }) - .await?; - - Ok(ok_empty()) + .map(|_| ()) + .map_err(XrpcError::from) } -#[derive(serde::Deserialize)] -pub(crate) struct PolicyInput { - repo: RepoDid, - #[serde(rename = "contributionPolicy")] +pub(crate) fn set_policy_in_registry( + meta: &Repo, + home: &CobHome, + owner: OwnerDid, + repo: &RepoDid, policy: ContributionPolicy, -} - -pub(crate) async fn set_contribution_policy( - State(state): State>>, - headers: HeaderMap, - method: crate::Method, - body: Bytes, -) -> Result { - let actor = state.authenticate_write(&headers, &method).await?; - let PolicyInput { repo, policy } = decode(&body)?; - - let acl = KnotAcl::new(&state.admins, state.admission, &state.index); - if !knot_acl::can_set_contribution_policy(&acl, &actor, &repo).is_allowed() { - return Err(XrpcError::forbidden( - "only repository owner, or knot admin, may set who may contribute to it, please ask one of them", - )); - } - let owner = match state.index.owner_of(&repo) { - Resolved::Ready(Some(owner)) => owner, - Resolved::Ready(None) => { - return Err(XrpcError::not_found("no such repository on this knot")); - } - Resolved::Warming => { - return Err(XrpcError::warming( - "registry projection is still warming, please retry shortly", - )); - } - }; - - let now = state.now(); - let knot_signer = state.secrets.signer(&state.knot_did)?; - let index = Arc::clone(&state.index); - let cob_locks = Arc::clone(&state.cob_locks); - let meta_path = state.meta_path.clone(); - let target = repo.clone(); - let home = CobHome::from(&state.knot_did); - run_blocking(move || { - { - let _guard = cob_locks.meta(); - let meta = Repo::open(&meta_path)?; - let store = CobStore::new(&meta); - let object = registry_object(&store)?.ok_or_else(|| { - XrpcError::not_found("no repositories are registered on this knot") - })?; - knot_cobs::set_repo_policy( - &store, - &home, - object, - knot_cobs::Policy { - owner, - repo: target.clone(), - policy, - }, - &knot_signer, - now, - ) - .map(|_| ()) - .map_err(XrpcError::from)?; - } - index.refresh_registry().map_err(|error| { - tracing::error!( - %target, - %error, - "reads fail closed until the next one, after repository took its new contribution policy, and registry projection refresh failed" - ); - XrpcError::stored_but_stale( - "retry is no-op, policy was stored and only projection refresh failed, thank you for retrying", - ) - }) - }) - .await?; - - Ok(ok_empty()) + signer: &dyn Signer, + now: UnixSeconds, +) -> Result<(), XrpcError> { + let store = CobStore::new(meta); + let object = registry_object(&store)? + .ok_or_else(|| XrpcError::not_found("no repositories are registered on this knot"))?; + knot_cobs::set_repo_policy( + &store, + home, + object, + knot_cobs::Policy { + owner, + repo: repo.clone(), + policy, + }, + signer, + now, + ) + .map(|_| ()) + .map_err(XrpcError::from) } diff --git a/knot2/crates/knot-xrpc/src/service.rs b/knot2/crates/knot-xrpc/src/service.rs index 551e3b988..d7af41b1f 100644 --- a/knot2/crates/knot-xrpc/src/service.rs +++ b/knot2/crates/knot-xrpc/src/service.rs @@ -7,19 +7,19 @@ use serde::Serialize; use knot_capability::Capability; use knot_runtime::{Clock, HttpTransport}; -use knot_types::AccountDid; +use knot_types::KnotId; use crate::XrpcState; - +use crate::error::XrpcError; pub(crate) const VERSION_ROUTE: &str = "/xrpc/org.tangled.knot.describeKnot"; -pub(crate) const OWNER_ROUTE: &str = "/xrpc/sh.tangled.owner"; +pub(crate) const RECEIVE_PACK_ROUTE: &str = "/xrpc/org.tangled.git.receivePack"; pub(crate) const HEALTH_ROUTE: &str = "/xrpc/_health"; const WIRE_VERSION: &str = "v1.15.0"; #[derive(Serialize)] struct VersionWire { - did: String, + did: KnotId, version: &'static str, capabilities: Vec<&'static str>, } @@ -47,16 +47,11 @@ pub(crate) async fn health( .into_response() } -#[derive(Serialize)] -struct OwnerWire { - owner: AccountDid, -} - pub(crate) async fn version( State(state): State>>, ) -> Response { Json(VersionWire { - did: state.knot_did.as_str().to_string(), + did: state.knot_did.clone(), version: WIRE_VERSION, capabilities: Capability::SERVED .iter() @@ -66,11 +61,11 @@ pub(crate) async fn version( .into_response() } -pub(crate) async fn owner( +pub(crate) async fn receive_pack( State(state): State>>, -) -> Response { - Json(OwnerWire { - owner: state.service_owner.clone(), - }) - .into_response() + headers: http::HeaderMap, + method: crate::Method, +) -> Result { + state.authenticate_write(&headers, &method).await?; + Ok(axum::http::StatusCode::OK.into_response()) } diff --git a/knot2/crates/knot-xrpc/src/sniff.rs b/knot2/crates/knot-xrpc/src/sniff.rs deleted file mode 100644 index 039c684f5..000000000 --- a/knot2/crates/knot-xrpc/src/sniff.rs +++ /dev/null @@ -1,264 +0,0 @@ -const SNIFF_LIMIT: usize = 512; - -fn is_ws(byte: u8) -> bool { - matches!(byte, b'\t' | b'\n' | 0x0c | b'\r' | b' ') -} - -fn is_tt(byte: u8) -> bool { - matches!(byte, b' ' | b'>') -} - -enum Sig { - Exact(&'static [u8], &'static str), - Masked { - mask: &'static [u8], - pat: &'static [u8], - skip_ws: bool, - ct: &'static str, - }, - Html(&'static [u8]), - Mp4, - Text, -} - -impl Sig { - fn detect(&self, data: &[u8], first_non_ws: usize) -> Option<&'static str> { - match self { - Sig::Exact(sig, ct) => data.starts_with(sig).then_some(*ct), - Sig::Masked { - mask, - pat, - skip_ws, - ct, - } => { - let data = if *skip_ws { - &data[first_non_ws..] - } else { - data - }; - (mask.len() == pat.len() - && data.len() >= pat.len() - && pat - .iter() - .zip(mask.iter()) - .enumerate() - .all(|(index, (byte, mask))| data[index] & mask == *byte)) - .then_some(*ct) - } - Sig::Html(tag) => { - let data = &data[first_non_ws..]; - (data.len() > tag.len() - && tag.iter().enumerate().all(|(index, byte)| { - let candidate = data[index]; - let candidate = match byte.is_ascii_uppercase() { - true => candidate & 0xDF, - false => candidate, - }; - *byte == candidate - }) - && is_tt(data[tag.len()])) - .then_some("text/html; charset=utf-8") - } - Sig::Mp4 => mp4(data), - Sig::Text => text(data, first_non_ws), - } - } -} - -fn mp4(data: &[u8]) -> Option<&'static str> { - if data.len() < 12 { - return None; - } - let box_size = u32::from_be_bytes([data[0], data[1], data[2], data[3]]) as usize; - if data.len() < box_size || !box_size.is_multiple_of(4) || &data[4..8] != b"ftyp" { - return None; - } - (8..box_size) - .step_by(4) - .filter(|start| *start != 12) - .any(|start| &data[start..start + 3] == b"mp4") - .then_some("video/mp4") -} - -fn text(data: &[u8], first_non_ws: usize) -> Option<&'static str> { - data[first_non_ws..] - .iter() - .all(|byte| !matches!(byte, 0x00..=0x08 | 0x0b | 0x0e..=0x1a | 0x1c..=0x1f)) - .then_some("text/plain; charset=utf-8") -} - -const SIGNATURES: &[Sig] = &[ - Sig::Html(b" &'static str { - let data = &content[..content.len().min(SNIFF_LIMIT)]; - let first_non_ws = data - .iter() - .position(|byte| !is_ws(*byte)) - .unwrap_or(data.len()); - SIGNATURES - .iter() - .find_map(|sig| sig.detect(data, first_non_ws)) - .unwrap_or("application/octet-stream") -} - -pub(crate) fn override_by_extension(path: &str, detected: &'static str) -> &'static str { - let extension = path.rsplit_once('.').map(|(_, ext)| ext).unwrap_or(""); - match extension.to_ascii_lowercase().as_str() { - "svg" => "image/svg+xml", - "avif" => "image/avif", - "jxl" => "image/jxl", - "heic" | "heif" => "image/heif", - _ => detected, - } -} - -pub(crate) fn is_textual_mime(mime: &str) -> bool { - mime.starts_with("text/") - || matches!( - mime, - "application/json" - | "application/xml" - | "application/yaml" - | "application/x-yaml" - | "application/toml" - | "application/javascript" - | "application/ecmascript" - ) -} - -#[cfg(test)] -mod tests { - use super::detect_content_type; - - #[test] - fn detects_common_content_signatures() { - let cases: &[(&[u8], &str)] = &[ - (b" \n", "text/html; charset=utf-8"), - (b"", "text/html; charset=utf-8"), - (b"", "text/html; charset=utf-8"), - (b"\n\t", "text/xml; charset=utf-8"), - (b"\xfe\xff\x00h", "text/plain; charset=utf-16be"), - (b"\xef\xbb\xbfhello", "text/plain; charset=utf-8"), - (b"\x89PNG\x0d\x0a\x1a\x0a", "image/png"), - (b"GIF89a", "image/gif"), - (b"fn main() {}\n", "text/plain; charset=utf-8"), - (b"\x00\x01\x02\x03", "application/octet-stream"), - (b"", "text/plain; charset=utf-8"), - ]; - cases.iter().for_each(|(input, expected)| { - assert_eq!(detect_content_type(input), *expected); - }); - } -} diff --git a/knot2/crates/knot-xrpc/tests/atproto.rs b/knot2/crates/knot-xrpc/tests/atproto.rs index 976eadc31..743e52c71 100644 --- a/knot2/crates/knot-xrpc/tests/atproto.rs +++ b/knot2/crates/knot-xrpc/tests/atproto.rs @@ -6,9 +6,7 @@ use std::time::Duration; use axum::body::Body; use cid::Cid; -use common::{ - World, fixture_collection, knot_host, nsid, owner, percent_encode, post_authed, -}; +use common::{World, fixture_collection, knot_host, nsid, owner, percent_encode, post_authed}; use futures::StreamExt; use http::{HeaderValue, StatusCode, header}; use jacquard_repo::Mst; @@ -65,7 +63,10 @@ async fn error_of(world: &World, path_and_query: &str) -> (StatusCode, serde_jso fn published_world() -> World { let world = World::new(); - world.publish_meta(&[DidRkey::new("did:plc:limpet").unwrap()], UnixSeconds::new(1_000)); + world.publish_meta( + &[DidRkey::new("did:plc:limpet").unwrap()], + UnixSeconds::new(1_000), + ); world.publish_meta( &[ DidRkey::new("did:plc:limpet").unwrap(), @@ -194,7 +195,8 @@ async fn an_unmodified_cold_client_verifies_a_record_off_the_proof_car() { assert_eq!(record["createdAt"], "1970-01-01T00:18:20Z"); assert_eq!(record["x-tngl-editor"], owner().as_str()); assert_eq!( - record["$type"], collection.as_str(), + record["$type"], + collection.as_str(), "the stored bytes carry their own $type, so a cold car reader can type the record" ); @@ -229,7 +231,8 @@ async fn an_unmodified_cold_client_verifies_a_record_off_the_proof_car() { ) .await; assert_eq!( - described["handle"], knot_host().as_str(), + described["handle"], + knot_host().as_str(), "the knot's own handle is its hostname" ); assert!( @@ -254,7 +257,10 @@ async fn an_unmodified_cold_client_verifies_a_record_off_the_proof_car() { assert_eq!(described["did"], knot.as_str()); let resolved = get_json( &world, - &format!("/xrpc/com.atproto.identity.resolveHandle?handle={}", knot_host()), + &format!( + "/xrpc/com.atproto.identity.resolveHandle?handle={}", + knot_host() + ), ) .await; assert_eq!(resolved["did"], knot.as_str()); @@ -312,7 +318,10 @@ async fn a_removal_takes_the_record_off_the_surface() { let world = published_world(); let knot = knot_repo(&world); let collection = fixture_collection(); - world.publish_meta(&[DidRkey::new("did:plc:whelk").unwrap()], UnixSeconds::new(1_200)); + world.publish_meta( + &[DidRkey::new("did:plc:whelk").unwrap()], + UnixSeconds::new(1_200), + ); let listed = get_json( &world, &format!("/xrpc/com.atproto.repo.listRecords?repo={knot}&collection={collection}"), @@ -534,7 +543,7 @@ async fn account_creation_announces_identity_account_and_an_empty_commit() { let (status, body) = post_authed( &world, - &nsid("sh.tangled.repo.delete"), + &nsid("org.tangled.repo.deleteRepo"), &owner(), serde_json::json!({ "repo": repo_did, "force": true }), ) diff --git a/knot2/crates/knot-xrpc/tests/atproto_interop.rs b/knot2/crates/knot-xrpc/tests/atproto_interop.rs index 0676429a4..59562192d 100644 --- a/knot2/crates/knot-xrpc/tests/atproto_interop.rs +++ b/knot2/crates/knot-xrpc/tests/atproto_interop.rs @@ -74,7 +74,7 @@ async fn an_indigo_client_consumes_the_firehose() { ), ( "KNOT_TEST_DELETE_TOKEN", - common::service_jwt(&nsid("sh.tangled.repo.delete"), &owner()), + common::service_jwt(&nsid("org.tangled.repo.deleteRepo"), &owner()), ), ], ) @@ -93,7 +93,7 @@ async fn a_knotfeed_consumer_reads_the_ref_records() { "2026-06-01T12:30:00+02:00", ); let addr = serve_pushes(&world).await; - http_push(work.path(), addr, &did, &owner(), &["-o", "skip-ci", "main"]).await; + http_push(work.path(), addr, &did, &owner(), &["main"]).await; let sha = sh_git(work.path(), &["rev-parse", "HEAD"]); go_test( @@ -101,7 +101,6 @@ async fn a_knotfeed_consumer_reads_the_ref_records() { &[ ("KNOT_TEST_ADDR", format!("127.0.0.1:{}", addr.port())), ("KNOT_TEST_REPO_DID", did.as_str().to_owned()), - ("KNOT_TEST_EDITOR", owner().to_string()), ("KNOT_TEST_SHA", sha.trim().to_owned()), ], ) diff --git a/knot2/crates/knot-xrpc/tests/blob_properties.rs b/knot2/crates/knot-xrpc/tests/blob_properties.rs index a0458da64..851900b33 100644 --- a/knot2/crates/knot-xrpc/tests/blob_properties.rs +++ b/knot2/crates/knot-xrpc/tests/blob_properties.rs @@ -24,7 +24,7 @@ const MAGICS: [(&[u8], &str); 8] = [ "image/webp", ), (&[b'B', b'M', 1, 2, 3, 4], "image/bmp"), - (b" RecordCollection { RecordCollection::new(PATCH_COLLECTION).unwrap() } +pub fn member_collection() -> RecordCollection { + RecordCollection::new("org.tangled.knot.member").unwrap() +} + +pub fn collaborator_collection() -> RecordCollection { + RecordCollection::new("org.tangled.graph.collaborator").unwrap() +} + +pub fn grant_body( + repo: &RepoDid, + collection: &RecordCollection, + subject: &AccountDid, + created_at: &str, +) -> serde_json::Value { + serde_json::json!({ + "repo": repo, + "collection": collection, + "rkey": subject.as_str(), + "record": { + "$type": collection, + "createdAt": created_at, + }, + }) +} + +pub fn grant_delete_body( + repo: &RepoDid, + collection: &RecordCollection, + subject: &AccountDid, +) -> serde_json::Value { + serde_json::json!({ + "repo": repo, + "collection": collection, + "rkey": subject.as_str(), + }) +} + pub fn blob_cid_of(value: &serde_json::Value) -> knot_types::BlobCid { let sent = value.as_str().expect("a blob CID is a string"); knot_types::BlobCid::new(cid::Cid::try_from(sent).expect("the CID parses")) @@ -330,10 +367,7 @@ pub async fn create_ticket( ) .await; assert_eq!(status, StatusCode::OK, "{written}"); - ( - rkey_of(&written), - record_cid_of(&written["cid"]), - ) + (rkey_of(&written), record_cid_of(&written["cid"])) } pub async fn create_patch( @@ -363,10 +397,7 @@ pub async fn create_patch( ) .await; assert_eq!(status, StatusCode::OK, "{written}"); - ( - rkey_of(&written), - record_cid_of(&written["cid"]), - ) + (rkey_of(&written), record_cid_of(&written["cid"])) } #[derive(serde::Serialize)] @@ -1217,7 +1248,7 @@ pub async fn get_error(world: &World, path_and_query: &str) -> (StatusCode, Stri (status, value["error"].as_str().unwrap().to_string()) } -pub fn ref_names(value: &serde_json::Value, key: &str) -> Vec { +pub fn ref_list(value: &serde_json::Value, key: &str) -> Vec { value[key] .as_array() .unwrap() @@ -1231,23 +1262,14 @@ pub fn repo_dids(value: &serde_json::Value) -> Vec { .as_array() .unwrap() .iter() - .map(|entry| entry["repo"].as_str().unwrap().to_string()) + .map(|entry| entry["did"].as_str().unwrap().to_string()) .collect() } -pub fn column<'p>(page: &'p serde_json::Value, field: &str) -> Vec<&'p str> { - page["items"] - .as_array() - .unwrap() - .iter() - .map(|item| item[field].as_str().unwrap()) - .collect() -} - -pub async fn archive_full(world: &World, did: &RepoDid) -> (String, String, Bytes) { +pub async fn archive_full(world: &World, did: &RepoDid, prefix: &str) -> (String, String, Bytes) { let (status, headers, body) = get( world, - &format!("/xrpc/sh.tangled.repo.archive?repo={did}&ref=main"), + &format!("/xrpc/org.tangled.git.getArchive?repo={did}&ref=main&prefix={prefix}"), ) .await; assert_eq!(status, StatusCode::OK); @@ -1259,7 +1281,7 @@ pub async fn archive_full(world: &World, did: &RepoDid) -> (String, String, Byte .to_string(); let last_modified = headers .get(header::LAST_MODIFIED) - .unwrap() + .expect("the archive includes the commit's time for date revalidation") .to_str() .unwrap() .to_string(); @@ -1335,6 +1357,11 @@ pub async fn assert_post_rejected( value: serde_json::Value, ) { let (status, body) = post_authed(world, method, actor, value).await; - assert_eq!(status, StatusCode::BAD_REQUEST, "{}: {body}", method.as_str()); + assert_eq!( + status, + StatusCode::BAD_REQUEST, + "{}: {body}", + method.as_str() + ); assert_eq!(body["error"], "InvalidRequest", "{}", method.as_str()); } diff --git a/knot2/crates/knot-xrpc/tests/grant_records.rs b/knot2/crates/knot-xrpc/tests/grant_records.rs index 6e11a7d14..ad7de7ca3 100644 --- a/knot2/crates/knot-xrpc/tests/grant_records.rs +++ b/knot2/crates/knot-xrpc/tests/grant_records.rs @@ -6,7 +6,6 @@ use futures::StreamExt; use knot_record::grant::GrantCollection; use knot_types::{AccountDid, RecordCollection, RepoDid, RepoName}; use serde::Deserialize; -use serde_json::json; #[derive(Deserialize)] struct CommitLine { @@ -78,6 +77,59 @@ async fn records_of( parsed["records"].as_array().cloned().unwrap_or_default() } +async fn offer_member( + world: &World, + subject: &AccountDid, +) -> (http::StatusCode, serde_json::Value) { + common::post_authed( + world, + &nsid("com.atproto.repo.createRecord"), + &owner(), + common::grant_body( + &world.state.knot_repo, + &GrantCollection::Member.records(), + subject, + "2026-09-11T00:00:00Z", + ), + ) + .await +} + +async fn offer_collaborator( + world: &World, + repo: &RepoDid, + subject: &AccountDid, +) -> (http::StatusCode, serde_json::Value) { + common::post_as( + world, + &nsid("com.atproto.repo.createRecord"), + &owner(), + &knot_types::ServiceDid::new(repo.as_str()).unwrap(), + common::grant_body( + repo, + &GrantCollection::Collaborator.records(), + subject, + "2026-09-11T00:00:00Z", + ), + ) + .await +} + +async fn revoke_collaborator( + world: &World, + repo: &RepoDid, + subject: &AccountDid, +) -> (http::StatusCode, serde_json::Value) { + common::post_as( + world, + &nsid("com.atproto.repo.deleteRecord"), + &owner(), + &knot_types::ServiceDid::new(repo.as_str()).unwrap(), + common::grant_delete_body(repo, &GrantCollection::Collaborator.records(), subject), + ) + .await +} + #[tokio::test] async fn the_offer_is_a_record_the_generic_surface_serves() { let world = World::administered(); @@ -95,28 +147,16 @@ async fn the_offer_is_a_record_the_generic_surface_serves() { let member_collection = GrantCollection::Member.records(); let collaborator_collection = GrantCollection::Collaborator.records(); - let (status, body) = common::post_authed( - &world, - &nsid("sh.tangled.knot.addMember"), - &owner(), - json!({ "subject": &member }), - ) - .await; - assert_eq!(status, 200, "addMember served the offer: {body}"); - let (status, body) = common::post_authed( - &world, - &nsid("sh.tangled.repo.addCollaborator"), - &owner(), - json!({ "repo": &repo, "subject": &collaborator }), - ) - .await; - assert_eq!(status, 200, "addCollaborator served the offer: {body}"); + let (status, body) = offer_member(&world, &member).await; + assert_eq!(status, 200, "createRecord served the member offer: {body}"); + let (status, body) = offer_collaborator(&world, &repo, &collaborator).await; + assert_eq!(status, 200, "createRecord served the offer: {body}"); let during = commits(&mut ws).await; assert_eq!( ops(&during, &knot_repo), vec![("create".into(), format!("{member_collection}/{member}"))], - "addMember emits one commit on the knot account: a create of the member record" + "the member offer emits 1 commit on the knot account: a create of the member record" ); assert_eq!( ops(&during, &repo), @@ -124,7 +164,7 @@ async fn the_offer_is_a_record_the_generic_surface_serves() { "create".into(), format!("{collaborator_collection}/{collaborator}") )], - "addCollaborator emits one commit on the repository: a create of the collaborator record" + "the collaborator offer emits 1 commit on the repository: a create of the record" ); let members = records_of(&world, &knot_repo, &member_collection).await; @@ -164,21 +204,15 @@ async fn the_offer_is_a_record_the_generic_surface_serves() { "getRecord finds the record by its DID rkey" ); - let (status, _body) = common::post_authed( - &world, - &nsid("sh.tangled.repo.removeCollaborator"), - &owner(), - json!({ "repo": &repo, "subject": &collaborator }), - ) - .await; - assert_eq!(status, 200, "removeCollaborator succeeds"); + let (status, _body) = revoke_collaborator(&world, &repo, &collaborator).await; + assert_eq!(status, 200, "deleteRecord revokes the collaborator"); assert_eq!( ops(&commits(&mut ws).await, &repo), vec![( "delete".into(), format!("{collaborator_collection}/{collaborator}") )], - "removeCollaborator emits one commit on the repository: a delete of the collaborator record" + "deleteRecord emits 1 commit on the repository: a delete of the collaborator record" ); assert_eq!( records_of(&world, &repo, &collaborator_collection) diff --git a/knot2/crates/knot-xrpc/tests/ref_records.rs b/knot2/crates/knot-xrpc/tests/ref_records.rs index 092c1cb86..fe0998476 100644 --- a/knot2/crates/knot-xrpc/tests/ref_records.rs +++ b/knot2/crates/knot-xrpc/tests/ref_records.rs @@ -1,20 +1,20 @@ mod common; use common::{ - World, commit_file, empty_repo, get_json, http_push, nsid, owner, percent_encode, post_authed, - serve_pushes, sh_git, + World, commit_file, empty_repo, get_json, http_push, owner, percent_encode, serve_pushes, + sh_git, }; use knot_events::{EventCursor, ReplayBounds, ReplayBytes, ReplayEvents}; -use knot_record::{chain, git_ref}; +use knot_record::{chain, git_refs}; use knot_types::{ - AccountDid, Did, KnotServiceUrl, RecordCollection, RepoDid, RepoName, UnixMicros, UnixSeconds + Did, KnotServiceUrl, RecordAddress, RecordCollection, RecordRkey, RepoDid, RepoName, + UnixSeconds, }; use serde::Deserialize; use serde_json::Value; +use std::sync::Arc; -fn collection() -> RecordCollection { - RecordCollection::new("sh.tangled.git.ref").unwrap() -} +use base64::Engine as _; fn seed(work: &std::path::Path, contents: &[u8]) { commit_file( @@ -26,30 +26,40 @@ fn seed(work: &std::path::Path, contents: &[u8]) { ); } -async fn records(world: &World, did: &RepoDid) -> Vec { +async fn snapshot(world: &World, did: &RepoDid) -> Value { get_json( world, &format!( - "/xrpc/com.atproto.repo.listRecords?repo={}&collection={}", + "/xrpc/com.atproto.repo.getRecord?repo={}&collection={}&rkey=self", percent_encode(did.as_str()), - collection() + git_refs::GIT_REFS_COLLECTION, ), ) - .await["records"] - .as_array() - .cloned() - .unwrap_or_default() + .await +} + +/// The snapshot's packed refs are dag-cbor bytes, so JSON serves them +/// base64-wrapped. The test unpacks them back into " " lines. +fn packed_lines(record: &Value) -> Vec { + let wrapped = record["value"]["refs"]["$bytes"] + .as_str() + .expect("the refs field is the wrapped bytes"); + let bytes = base64::engine::general_purpose::STANDARD_NO_PAD + .decode(wrapped) + .expect("the knot wrote standard base64"); + String::from_utf8(bytes) + .expect("the packed refs are utf-8") + .lines() + .map(str::to_owned) + .collect() } -fn record_of<'r>(items: &'r [Value], rkey: &str) -> &'r Value { - items +fn sha_of(lines: &[String], refname: &str) -> String { + lines .iter() - .find(|item| { - item["uri"] - .as_str() - .is_some_and(|uri| uri.ends_with(&format!("{}/{rkey}", collection()))) - }) - .unwrap_or_else(|| panic!("no record for rkey {rkey} in {items:?}")) + .find_map(|line| line.strip_suffix(&format!(" {refname}"))) + .unwrap_or_else(|| panic!("no {refname} in {lines:?}")) + .to_owned() } #[derive(Debug, Deserialize)] @@ -60,6 +70,7 @@ struct OpProbe { #[derive(Deserialize)] struct CommitProbe { + #[allow(dead_code)] seq: u64, ops: Vec, } @@ -96,7 +107,7 @@ fn frames(world: &World) -> Vec<(&'static str, Option)> { } #[tokio::test] -async fn a_push_lands_one_record_per_ref_the_knot_serves_over_com_atproto() { +async fn a_push_stores_the_refs_snapshot_the_knot_serves_over_com_atproto() { let world = World::new(); let (did, _bare, work) = empty_repo(&world, &RepoName::new("squid").unwrap()); seed(work.path(), b"hello\n"); @@ -104,26 +115,30 @@ async fn a_push_lands_one_record_per_ref_the_knot_serves_over_com_atproto() { let addr = serve_pushes(&world).await; http_push(work.path(), addr, &did, &owner(), &["main", "v1"]).await; - let items = records(&world, &did).await; - assert_eq!(items.len(), 2, "one record per live ref: {items:?}"); - let main = record_of(&items, "refs~2fheads~2fmain"); - assert_eq!( - main["value"]["$type"], collection().as_str(), - "the stored bytes carry their own type" - ); + let record = snapshot(&world, &did).await; assert_eq!( - main["value"]["sha"], - sh_git(work.path(), &["rev-parse", "refs/heads/main"]), - "the record value is the ref's present object" + record["value"]["$type"], + git_refs::GIT_REFS_COLLECTION, + "the stored bytes include their own type" ); assert_eq!( - main["value"]["x-tngl-editor"], owner().as_str(), - "the pusher rides the undeclared editor field" + record["value"]["defaultBranch"], "main", + "the snapshot reports the default branch" ); + let lines = packed_lines(&record); assert_eq!( - record_of(&items, "refs~2ftags~2fv1")["value"]["sha"], - sh_git(work.path(), &["rev-parse", "refs/tags/v1"]), - "an annotated tag records its tag object" + lines, + vec![ + format!( + "{} refs/heads/main", + sh_git(work.path(), &["rev-parse", "refs/heads/main"]) + ), + format!( + "{} refs/tags/v1", + sh_git(work.path(), &["rev-parse", "refs/tags/v1"]) + ), + ], + "the packed refs list every live ref, sorted by name" ); let kinds = frames(&world) @@ -141,40 +156,62 @@ async fn a_push_lands_one_record_per_ref_the_knot_serves_over_com_atproto() { sh_git(work.path(), &["branch", "feature"]); http_push(work.path(), addr, &did, &owner(), &["feature"]).await; + let lines = packed_lines(&snapshot(&world, &did).await); assert_eq!( - records(&world, &did).await.len(), + lines.len(), 3, - "the branch's record landed before anything deletes it" + "the branch is written into the snapshot before anything deletes it" + ); + + http_push(work.path(), addr, &did, &owner(), &[":feature"]).await; + let lines = packed_lines(&snapshot(&world, &did).await); + assert_eq!(lines.len(), 2, "the snapshot no longer lists the deleted branch"); + assert!( + lines + .iter() + .all(|line| !line.ends_with("refs/heads/feature")), + "the repo serves exactly the refs in the packed refs" + ); +} + +#[tokio::test] +async fn the_refs_snapshot_appears_on_the_firehose_as_a_create_then_updates() { + let world = World::new(); + let (did, _bare, work) = empty_repo(&world, &RepoName::new("auger").unwrap()); + seed(work.path(), b"hello\n"); + sh_git(work.path(), &["branch", "feature"]); + let addr = serve_pushes(&world).await; + http_push(work.path(), addr, &did, &owner(), &["main", "feature"]).await; + + let snapshot_ops = |world: &World| { + frames(world) + .into_iter() + .filter_map(|(_, payload)| payload) + .flat_map(|payload| payload.ops) + .filter(|op| op.path == format!("{}/self", git_refs::GIT_REFS_COLLECTION)) + .map(|op| op.action) + .collect::>() + }; + assert_eq!( + snapshot_ops(&world), + ["create"], + "the push's #commit frame includes the snapshot record's create" ); http_push(work.path(), addr, &did, &owner(), &[":feature"]).await; - let items = records(&world, &did).await; - assert_eq!(items.len(), 2, "the deleted branch's record went with it"); - assert!(record_of(&items, "refs~2fheads~2fmain")["value"].is_object()); - let deletes = frames(&world) - .into_iter() - .filter_map(|(_, payload)| payload) - .flat_map(|payload| payload.ops) - .filter(|op| op.action == "delete") - .map(|op| op.path.clone()) - .collect::>(); assert_eq!( - deletes, - [format!("{}/refs~2fheads~2ffeature", collection())], - "the deletion rides the op list as a deleteRecord" + snapshot_ops(&world), + ["create", "update"], + "a branch delete is another update to the snapshot record, never a delete op" ); } #[tokio::test] -async fn the_pusher_and_the_skip_ci_option_ride_the_undeclared_fields() { +async fn repushing_identical_refs_is_a_no_op() { let world = World::new(); let (did, _bare, work) = empty_repo(&world, &RepoName::new("anemone").unwrap()); seed(work.path(), b"first\n"); - let limpet = AccountDid::new("did:plc:limpet").unwrap(); - world.add_collaborator(&did, &limpet, &owner(), UnixSeconds::new(1_000)); - let addr = serve_pushes(&world).await; - http_push(work.path(), addr, &did, &owner(), &["-o", "skip-ci", "main"]).await; - + sh_git(work.path(), &["checkout", "-q", "-b", "feature"]); commit_file( work.path(), "kelp.txt", @@ -182,116 +219,174 @@ async fn the_pusher_and_the_skip_ci_option_ride_the_undeclared_fields() { "second", "2026-06-01T12:31:00+02:00", ); - sh_git(work.path(), &["checkout", "-q", "-b", "feature"]); - http_push(work.path(), addr, &did, &limpet, &["feature"]).await; + let addr = serve_pushes(&world).await; + http_push(work.path(), addr, &did, &owner(), &["main", "feature"]).await; - let items = records(&world, &did).await; - assert_eq!(items.len(), 2, "main and feature, one record each"); - let main = record_of(&items, "refs~2fheads~2fmain"); - assert_eq!( - main["value"]["x-tngl-editor"], owner().as_str(), - "a push that doesn't touch main leaves its stored record bytes alone" - ); - assert_eq!( - main["value"]["x-tngl-push-options"][0], "skip-ci", - "the options of the push that moved main stay with it" - ); - let feature = record_of(&items, "refs~2fheads~2ffeature"); - assert_eq!( - feature["value"]["x-tngl-editor"], "did:plc:limpet", - "the collaborator's push names the collaborator" - ); - assert!( - feature["value"].get("x-tngl-push-options").is_none(), - "Push without options doesn't write options field" + let latest_path = format!( + "/xrpc/com.atproto.sync.getLatestCommit?did={}", + percent_encode(did.as_str()), ); + let settled = get_json(&world, &latest_path).await; + let record = snapshot(&world, &did).await; + let frames_before = frames(&world).len(); - let settled = chain::tip(&world.layout.open(&did).unwrap()) - .unwrap() - .unwrap() - .git; - http_push(work.path(), addr, &did, &owner(), &["feature"]).await; + http_push(work.path(), addr, &did, &owner(), &["main", "feature"]).await; assert_eq!( - chain::tip(&world.layout.open(&did).unwrap()) - .unwrap() - .unwrap() - .git, + get_json(&world, &latest_path).await, settled, - "a no-op update, the shape git push --mirror sends for every current ref, writes nothing" + "a no-op re-push, the shape git push --mirror sends for every current ref, won't move a commit" ); assert_eq!( - record_of(&records(&world, &did).await, "refs~2fheads~2ffeature")["value"]["x-tngl-editor"], - "did:plc:limpet", - "the no-op re-push leaves the stored record bytes alone" + snapshot(&world, &did).await, + record, + "the stored snapshot bytes are what the first push wrote" + ); + assert_eq!( + frames(&world).len(), + frames_before, + "a no-op re-push doesn't emit firehose frames" ); } #[tokio::test] -async fn a_mirror_sized_first_push_chunks_at_two_hundred_ops() { +async fn the_legacy_sweep_deletes_old_ref_records_and_survives_its_chunking() { let world = World::new(); let (did, _bare, work) = empty_repo(&world, &RepoName::new("atoll").unwrap()); seed(work.path(), b"first\n"); - (1..=250).for_each(|branch| { - sh_git(work.path(), &["branch", &format!("b{branch}")]); - }); let addr = serve_pushes(&world).await; - http_push(work.path(), addr, &did, &owner(), &["--all"]).await; - - let repo = world.layout.open(&did).unwrap(); - let tip = chain::tip(&repo).unwrap().expect("the chain exists"); - let framed = chain::framed_revs(&repo, tip.git, UnixMicros::new(0), 400); - assert!(framed.found.len() >= 2, "251 refs need several commits"); - let mut creates = 0usize; - let seqs = framed - .found - .iter() - .map(|(seq, oid)| { - let frame = chain::frame_blob(&repo, *oid) - .unwrap() - .expect("every commit in this chain frames"); - let (kind, payload) = decoded(&frame); - assert_eq!(kind, "#commit", "a chain commit frames as #commit"); - let payload = payload.expect("#commit carries its payload"); - assert_eq!( - payload.seq, - seq.get(), - "a chunk's frame seq is the microsecond field of its own rev" - ); - assert!( - payload.ops.len() <= chain::MAX_OPS_PER_COMMIT, - "a chunk respects the firehose operation ceiling" - ); - assert!( - payload.ops.iter().all(|op| { - op.action == "create" - && (op - .path - .starts_with(&format!("{}/refs~2fheads~2f", collection())) - || op.path == "org.tangled.git.refs/self") - }), - "every op names a head ref record or the summary seeded at birth: {:?}", - payload.ops - ); - creates += payload - .ops - .iter() - .filter(|op| { - op.path - .starts_with(&format!("{}/refs~2fheads~2f", collection())) - }) - .count(); - *seq + http_push(work.path(), addr, &did, &owner(), &["main"]).await; + let snapshot_before = snapshot(&world, &did).await; + + let legacy = RecordCollection::new("sh.tangled.git.ref").unwrap(); + let ops: Vec<(RecordAddress, chain::RecordOp)> = (0..205) + .map(|nth| { + ( + RecordAddress::new( + legacy.clone(), + RecordRkey::new(format!("legacy~{nth:03}")).unwrap(), + ), + chain::RecordOp::Create { + record: knot_types::RecordBody::new( + serde_ipld_dagcbor::to_vec(&serde_json::json!({ + "$type": "sh.tangled.git.ref", + "sha": "0".repeat(40), + "createdAt": "2026-06-01T12:30:00Z", + })) + .unwrap(), + ) + .unwrap(), + }, + ) }) - .collect::>(); - assert_eq!(creates, 251, "main plus 250 branches, one op each"); + .collect(); + let layout = world.layout.clone(); + let chunks = tokio::task::spawn_blocking({ + let state = Arc::clone(&world.state); + let did = did.clone(); + move || { + let git = layout.open(&did).unwrap(); + let signer = state.secrets.signer(&state.knot_did).unwrap(); + let now = UnixSeconds::new(2_000); + let mut emit = knot_xrpc::FirehoseEmit::new(&state.firehose, &did, now); + let chunks = futures::executor::block_on(chain::apply( + &git, &did, &ops, &signer, &mut emit, now, + )) + .unwrap(); + emit.finish().unwrap(); + chunks + } + }) + .await + .unwrap(); + assert_eq!( + chunks.len(), + 2, + "205 creates split past the 200 ops each commit allows" + ); + let seeded = get_json( + &world, + &format!( + "/xrpc/com.atproto.repo.getRecord?repo={}&collection=sh.tangled.git.ref&rkey=legacy~000", + percent_encode(did.as_str()), + ), + ) + .await; + assert_eq!(seeded["value"]["$type"], "sh.tangled.git.ref"); + + let swept = tokio::task::spawn_blocking({ + let state = Arc::clone(&world.state); + move || knot_xrpc::migrate::legacy_state(&state) + }) + .await + .unwrap() + .unwrap(); + assert_eq!( + swept, 2, + "the sweep's deletes split across 2 chain commits of their own" + ); + let empty = get_json( + &world, + &format!( + "/xrpc/com.atproto.repo.listRecords?repo={}&collection=sh.tangled.git.ref", + percent_encode(did.as_str()), + ), + ) + .await; assert!( - seqs.windows(2).all(|pair| pair[0] > pair[1]), - "framed_revs walks newest first and no two commits share a seq" + empty["records"].as_array().unwrap().is_empty(), + "the legacy collection serves an empty page after the sweep" + ); + let (status, error) = common::get_error( + &world, + &format!( + "/xrpc/com.atproto.repo.getRecord?repo={}&collection=sh.tangled.git.ref&rkey=legacy~000", + percent_encode(did.as_str()), + ), + ) + .await; + assert_eq!( + (status, error.as_str()), + (http::StatusCode::BAD_REQUEST, "RecordNotFound") + ); + assert_eq!( + snapshot(&world, &did).await, + snapshot_before, + "the sweep won't touch the refs snapshot the push wrote" + ); + let deletes = frames(&world) + .into_iter() + .filter_map(|(_, payload)| payload) + .flat_map(|payload| payload.ops) + .filter(|op| op.action == "delete" && op.path.starts_with("sh.tangled.git.ref/")) + .count(); + assert_eq!( + deletes, 205, + "the sweep will emit a firehose delete for every legacy record, across the chunk boundary" + ); + + let frames_after_sweep = frames(&world).len(); + let again = tokio::task::spawn_blocking({ + let state = Arc::clone(&world.state); + move || knot_xrpc::migrate::legacy_state(&state) + }) + .await + .unwrap() + .unwrap(); + assert_eq!(again, 0, "a second sweep is a no-op"); + assert_eq!( + frames(&world).len(), + frames_after_sweep, + "the second sweep doesn't emit frames" + ); + assert_eq!( + snapshot(&world, &did).await, + snapshot_before, + "the snapshot is unchanged after the second sweep" ); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn the_boot_pass_catches_up_a_ref_that_moved_without_a_record() { +async fn the_boot_pass_catches_up_refs_that_moved_without_a_record() { let world = World::new(); let (did, _bare, work) = empty_repo(&world, &RepoName::new("coral").unwrap()); seed(work.path(), b"first\n"); @@ -313,18 +408,9 @@ async fn the_boot_pass_catches_up_a_ref_that_moved_without_a_record() { .await .unwrap(); - let items = records(&world, &did).await; - assert_eq!(items.len(), 2, "the boot pass materialized the drift"); - let ghost = record_of(&items, "refs~2fheads~2fghost"); - assert_eq!(ghost["value"]["sha"], main.to_hex()); - assert!( - ghost["value"].get("x-tngl-editor").is_none(), - "a catch-up record names no editor, because one commit merged an unknown author" - ); - assert!( - ghost["value"].get("x-tngl-push-options").is_none(), - "a catch-up record carries no options" - ); + let lines = packed_lines(&snapshot(&world, &did).await); + assert_eq!(sha_of(&lines, "refs/heads/ghost"), main.to_hex()); + assert_eq!(sha_of(&lines, "refs/heads/main"), main.to_hex()); assert_eq!( frames(&world).last().map(|(kind, _)| kind), Some(&"#sync"), @@ -344,22 +430,7 @@ async fn the_boot_pass_catches_up_a_ref_that_moved_without_a_record() { .unwrap() .git, settled.git, - "a second boot with nothing to do moves no tip" - ); - - let long = "a".repeat(git_ref::MAX_RKEY_BYTES); - sh_git(work.path(), &["branch", &long]); - http_push(work.path(), addr, &did, &owner(), &["main", &long]).await; - let items = records(&world, &did).await; - assert_eq!( - items.len(), - 2, - "the over-budget ref is skipped and the rest syncs: {items:?}" - ); - assert_eq!( - record_of(&items, "refs~2fheads~2fmain")["value"]["sha"], - sh_git(work.path(), &["rev-parse", "refs/heads/main"]), - "main's record stays live beside the skipped ref" + "a second boot without drift doesn't move the tip" ); } @@ -377,9 +448,9 @@ async fn created_fork_serves_its_upstream_refs_at_birth() { &Did::new_owned(source_did.as_str()).unwrap(), &KnotServiceUrl::new("https://knot.nel.pet").unwrap(), ); - let (status, created) = post_authed( + let (status, created) = common::post_authed( &world, - &nsid("org.tangled.repo.createRepo"), + &common::nsid("org.tangled.repo.createRepo"), &owner(), serde_json::json!({ "name": "reef", @@ -398,11 +469,11 @@ async fn created_fork_serves_its_upstream_refs_at_birth() { let fork_did = RepoDid::new( created["did"] .as_str() - .expect("a created repo answers its did"), + .expect("a created repo returns its did"), ) .unwrap(); - // the fork populates through the background importer, so the birth records land + // the fork populates through the background importer, so the birth records are written // once its fetch finishes; the timeout only bounds the failure, never the success let before = world.state.imports.processed(); tokio::time::timeout(std::time::Duration::from_secs(10), async { @@ -412,20 +483,21 @@ async fn created_fork_serves_its_upstream_refs_at_birth() { }) .await .expect("the importer never finished the fork's fetch"); - let items = records(&world, &fork_did).await; - let main = record_of(&items, "refs~2fheads~2fmain"); + let record = snapshot(&world, &fork_did).await; assert_eq!( - main["value"]["sha"], - sh_git(work.path(), &["rev-parse", "refs/heads/main"]), - "the seeded record names the upstream tip" + record["value"]["defaultBranch"], "main", + "the fork's snapshot reports the default branch it took from the upstream" ); + let lines = packed_lines(&record); assert_eq!( - main["value"]["$type"], collection().as_str(), - "the seeded bytes carry the collection type" + sha_of(&lines, "refs/heads/main"), + sh_git(work.path(), &["rev-parse", "refs/heads/main"]), + "the seeded snapshot reports the upstream tip" ); - assert!( - main["value"].get("x-tngl-editor").is_none(), - "a fork's birth record names no pusher" + assert_eq!( + sha_of(&lines, "refs/tags/v1"), + sh_git(work.path(), &["rev-parse", "refs/tags/v1"]), + "the seeded snapshot includes the upstream's tag" ); } @@ -433,9 +505,9 @@ async fn created_fork_serves_its_upstream_refs_at_birth() { async fn create_repo_writes_its_manifest_into_the_repo_account() { let world = World::new(); world.add_member(&owner(), &owner(), UnixSeconds::new(1_000)); - let (status, created) = post_authed( + let (status, created) = common::post_authed( &world, - &nsid("org.tangled.repo.createRepo"), + &common::nsid("org.tangled.repo.createRepo"), &owner(), serde_json::json!({ "name": "Lagoon", @@ -479,9 +551,9 @@ async fn create_repo_writes_its_manifest_into_the_repo_account() { #[tokio::test] async fn the_legacy_create_route_is_gone() { let world = World::new(); - let (status, _) = post_authed( + let (status, _) = common::post_authed( &world, - &nsid("sh.tangled.repo.create"), + &common::nsid("sh.tangled.repo.create"), &owner(), serde_json::json!({ "rkey": "lagoon", "name": "lagoon" }), ) diff --git a/knot2/example.toml b/knot2/example.toml index 026f2ff30..62b45fa18 100644 --- a/knot2/example.toml +++ b/knot2/example.toml @@ -523,8 +523,8 @@ [messages.reject] # Can also be specified via environment variable `KNOT_MESSAGES_REJECT_RESERVED_REFS`. -# Default value: "refs/cobs/*, refs/cob-checkpoints/*, refs/hidden/* and refs/atproto/* are reserved and can't be pushed" -#reserved_refs = "refs/cobs/*, refs/cob-checkpoints/*, refs/hidden/* and refs/atproto/* are reserved and can't be pushed" +# Default value: "refs/cobs/*, refs/cob-checkpoints/*, refs/hidden/*, refs/atproto/* and refs/tngl/* are reserved and can't be pushed" +#reserved_refs = "refs/cobs/*, refs/cob-checkpoints/*, refs/hidden/*, refs/atproto/* and refs/tngl/* are reserved and can't be pushed" # Can also be specified via environment variable `KNOT_MESSAGES_REJECT_COB_CREATE_ONLY`. # Default value: "existing refs/cobs/* object cannot be modified or deleted over the wire" @@ -542,6 +542,10 @@ # Default value: "refs/atproto/* is reserved for the knot's atproto commit chain and can't be pushed" #atproto_reserved = "refs/atproto/* is reserved for the knot's atproto commit chain and can't be pushed" +# Can also be specified via environment variable `KNOT_MESSAGES_REJECT_TNGL_RESERVED`. +# Default value: "refs/tngl/* is reserved for knot-written keep refs and can't be pushed" +#tngl_reserved = "refs/tngl/* is reserved for knot-written keep refs and can't be pushed" + # Can also be specified via environment variable `KNOT_MESSAGES_REJECT_CHECKPOINT_RESERVED`. # Default value: "refs/cob-checkpoints/* is reserved for knot-written collaborative-object snapshots and can't be pushed" #checkpoint_reserved = "refs/cob-checkpoints/* is reserved for knot-written collaborative-object snapshots and can't be pushed"