1mod identity;
47mod repo_metadata;
48
49use crate::runtime::block_on_result;
50use anyhow::{Context, Result};
51use futures::{SinkExt, StreamExt};
52use hashtree_blossom::BlossomClient;
53use hashtree_core::{decode_tree_node, decrypt_chk, LinkType};
54use nostr_sdk::prelude::*;
55use serde::Deserialize;
56use std::collections::HashMap;
57use std::time::Duration;
58use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage};
59use tracing::{debug, info, warn};
60
61pub use identity::{
62 load_key_lists, load_keys, resolve_identity, resolve_self_identity, StoredKey, StoredKeyLists,
63};
64#[cfg(test)]
65use repo_metadata::pick_latest_event;
66use repo_metadata::{
67 append_repo_discovery_labels, build_git_repo_list_filter, build_repo_event_filter,
68 latest_repo_event_created_at, latest_trusted_pr_status_kinds, list_git_repo_announcements,
69 next_replaceable_created_at, pick_latest_repo_event, validate_repo_publish_relays,
70};
71
72pub const KIND_APP_DATA: u16 = 30078;
74
75pub const KIND_PULL_REQUEST: u16 = 1618;
77pub const KIND_STATUS_OPEN: u16 = 1630;
78pub const KIND_STATUS_APPLIED: u16 = 1631;
79pub const KIND_STATUS_CLOSED: u16 = 1632;
80pub const KIND_STATUS_DRAFT: u16 = 1633;
81pub const KIND_REPO_ANNOUNCEMENT: u16 = 30617;
82
83pub const LABEL_HASHTREE: &str = "hashtree";
85pub const LABEL_GIT: &str = "git";
86
87#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub enum PullRequestState {
90 Open,
91 Applied,
92 Closed,
93 Draft,
94}
95
96impl PullRequestState {
97 pub fn as_str(self) -> &'static str {
98 match self {
99 PullRequestState::Open => "open",
100 PullRequestState::Applied => "applied",
101 PullRequestState::Closed => "closed",
102 PullRequestState::Draft => "draft",
103 }
104 }
105
106 fn from_status_kind(status_kind: u16) -> Option<Self> {
107 match status_kind {
108 KIND_STATUS_OPEN => Some(PullRequestState::Open),
109 KIND_STATUS_APPLIED => Some(PullRequestState::Applied),
110 KIND_STATUS_CLOSED => Some(PullRequestState::Closed),
111 KIND_STATUS_DRAFT => Some(PullRequestState::Draft),
112 _ => None,
113 }
114 }
115
116 fn from_latest_status_kind(status_kind: Option<u16>) -> Self {
117 status_kind
118 .and_then(Self::from_status_kind)
119 .unwrap_or(PullRequestState::Open)
120 }
121}
122
123#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
125pub enum PullRequestStateFilter {
126 #[default]
127 Open,
128 Applied,
129 Closed,
130 Draft,
131 All,
132}
133
134impl PullRequestStateFilter {
135 pub fn as_str(self) -> &'static str {
136 match self {
137 PullRequestStateFilter::Open => "open",
138 PullRequestStateFilter::Applied => "applied",
139 PullRequestStateFilter::Closed => "closed",
140 PullRequestStateFilter::Draft => "draft",
141 PullRequestStateFilter::All => "all",
142 }
143 }
144
145 fn includes(self, state: PullRequestState) -> bool {
146 match self {
147 PullRequestStateFilter::All => true,
148 PullRequestStateFilter::Open => state == PullRequestState::Open,
149 PullRequestStateFilter::Applied => state == PullRequestState::Applied,
150 PullRequestStateFilter::Closed => state == PullRequestState::Closed,
151 PullRequestStateFilter::Draft => state == PullRequestState::Draft,
152 }
153 }
154}
155
156#[derive(Debug, Clone)]
158pub struct PullRequestListItem {
159 pub event_id: String,
160 pub author_pubkey: String,
161 pub state: PullRequestState,
162 pub subject: Option<String>,
163 pub commit_tip: Option<String>,
164 pub branch: Option<String>,
165 pub target_branch: Option<String>,
166 pub created_at: u64,
167}
168
169async fn fetch_events_via_raw_relay_query(
170 relays: &[String],
171 filter: Filter,
172 timeout: Duration,
173) -> Vec<Event> {
174 let request_json = ClientMessage::req(SubscriptionId::generate(), vec![filter]).as_json();
175 let mut events_by_id = HashMap::<String, Event>::new();
176
177 for relay_url in relays {
178 let relay_events = match tokio::time::timeout(timeout, async {
179 let (mut ws, _) = connect_async(relay_url).await?;
180 ws.send(WsMessage::Text(request_json.clone())).await?;
181
182 let mut relay_events = Vec::new();
183 while let Some(message) = ws.next().await {
184 let message = message?;
185 let WsMessage::Text(text) = message else {
186 continue;
187 };
188
189 match RelayMessage::from_json(text.as_str()) {
190 Ok(RelayMessage::Event { event, .. }) => relay_events.push(*event),
191 Ok(RelayMessage::EndOfStoredEvents(_)) => break,
192 Ok(RelayMessage::Closed { message, .. }) => {
193 debug!("Raw relay PR query closed by {}: {}", relay_url, message);
194 break;
195 }
196 Ok(_) => {}
197 Err(err) => {
198 debug!(
199 "Failed to parse raw relay response from {}: {}",
200 relay_url, err
201 );
202 }
203 }
204 }
205
206 let _ = ws.close(None).await;
207 Ok::<Vec<Event>, anyhow::Error>(relay_events)
208 })
209 .await
210 {
211 Ok(Ok(events)) => events,
212 Ok(Err(err)) => {
213 debug!("Raw relay PR query failed for {}: {}", relay_url, err);
214 continue;
215 }
216 Err(_) => {
217 debug!("Raw relay PR query timed out for {}", relay_url);
218 continue;
219 }
220 };
221
222 for event in relay_events {
223 events_by_id.insert(event.id.to_hex(), event);
224 }
225 }
226
227 events_by_id.into_values().collect()
228}
229
230async fn connected_relay_count(client: &Client) -> (usize, usize) {
231 let relays = client.relays().await;
232 let total = relays.len();
233 let mut connected = 0;
234 for relay in relays.values() {
235 if relay.is_connected().await {
236 connected += 1;
237 }
238 }
239 (connected, total)
240}
241
242async fn wait_for_any_connected_relay(client: &Client, timeout: Duration) -> bool {
243 let start = std::time::Instant::now();
244 loop {
245 if connected_relay_count(client).await.0 > 0 {
246 return true;
247 }
248 if start.elapsed() > timeout {
249 return false;
250 }
251 tokio::time::sleep(Duration::from_millis(50)).await;
252 }
253}
254
255type FetchedRefs = (HashMap<String, String>, Option<String>, Option<[u8; 32]>);
256use hashtree_config::Config;
257
258#[derive(Debug, Clone)]
260pub struct RelayResult {
261 #[allow(dead_code)]
263 pub configured: Vec<String>,
264 pub connected: Vec<String>,
266 pub failed: Vec<String>,
268}
269
270#[derive(Debug, Clone)]
272pub struct BlossomResult {
273 #[allow(dead_code)]
275 pub configured: Vec<String>,
276 pub succeeded: Vec<String>,
278 pub failed: Vec<String>,
280 pub local_complete: bool,
282 pub degraded: bool,
284}
285
286pub struct NostrClient {
288 pubkey: String,
289 keys: Option<Keys>,
291 relays: Vec<String>,
292 blossom: BlossomClient,
293 cached_refs: HashMap<String, HashMap<String, String>>,
295 cached_root_hash: HashMap<String, String>,
297 cached_encryption_key: HashMap<String, [u8; 32]>,
299 url_secret: Option<[u8; 32]>,
302 is_private: bool,
304 local_daemon_url: Option<String>,
306}
307
308#[derive(Debug, Clone, Default)]
309struct RootEventData {
310 root_hash: String,
311 encryption_key: Option<[u8; 32]>,
312 key_tag_name: Option<String>,
313 self_encrypted_ciphertext: Option<String>,
314}
315
316#[derive(Debug, Deserialize)]
317struct DaemonResolveResponse {
318 hash: Option<String>,
319 #[serde(default, rename = "key_tag")]
320 key: Option<String>,
321 #[serde(default, rename = "encryptedKey")]
322 encrypted_key: Option<String>,
323 #[serde(default, rename = "selfEncryptedKey")]
324 self_encrypted_key: Option<String>,
325 #[serde(default)]
326 source: Option<String>,
327}
328
329impl NostrClient {
330 pub fn new(
332 pubkey: &str,
333 secret_key: Option<String>,
334 url_secret: Option<[u8; 32]>,
335 is_private: bool,
336 config: &Config,
337 ) -> Result<Self> {
338 let _ = rustls::crypto::ring::default_provider().install_default();
340
341 let secret_key = secret_key.or_else(|| std::env::var("NOSTR_SECRET_KEY").ok());
343
344 let keys = if let Some(ref secret_hex) = secret_key {
346 let secret_bytes = hex::decode(secret_hex).context("Invalid secret key hex")?;
347 let secret = nostr::SecretKey::from_slice(&secret_bytes)
348 .map_err(|e| anyhow::anyhow!("Invalid secret key: {}", e))?;
349 Some(Keys::new(secret))
350 } else {
351 None
352 };
353
354 let blossom_keys = keys.clone().unwrap_or_else(Keys::generate);
357 let blossom = BlossomClient::new(blossom_keys).with_timeout(Duration::from_secs(30));
358
359 tracing::info!(
360 "BlossomClient created with read_servers: {:?}, write_servers: {:?}",
361 blossom.read_servers(),
362 blossom.write_servers()
363 );
364
365 let relays = hashtree_config::resolve_relays(
366 &config.nostr.relays,
367 Some(config.server.bind_address.as_str()),
368 );
369 let local_daemon_url =
370 hashtree_config::detect_local_daemon_url(Some(config.server.bind_address.as_str()))
371 .or_else(|| {
372 config
373 .blossom
374 .read_servers
375 .iter()
376 .find(|url| {
377 url.starts_with("http://127.0.0.1:")
378 || url.starts_with("http://localhost:")
379 })
380 .cloned()
381 });
382
383 Ok(Self {
384 pubkey: pubkey.to_string(),
385 keys,
386 relays,
387 blossom,
388 cached_refs: HashMap::new(),
389 cached_root_hash: HashMap::new(),
390 cached_encryption_key: HashMap::new(),
391 url_secret,
392 is_private,
393 local_daemon_url,
394 })
395 }
396
397 fn format_repo_author(pubkey_hex: &str) -> String {
398 PublicKey::from_hex(pubkey_hex)
399 .ok()
400 .and_then(|pk| pk.to_bech32().ok())
401 .unwrap_or_else(|| pubkey_hex.to_string())
402 }
403
404 #[allow(dead_code)]
406 pub fn can_sign(&self) -> bool {
407 self.keys.is_some()
408 }
409
410 pub fn list_repos(&self) -> Result<Vec<String>> {
411 block_on_result(self.list_repos_async())
412 }
413
414 pub async fn list_repos_async(&self) -> Result<Vec<String>> {
415 let client = Client::default();
416
417 for relay in &self.relays {
418 if let Err(e) = client.add_relay(relay).await {
419 warn!("Failed to add relay {}: {}", relay, e);
420 }
421 }
422 client.connect().await;
423
424 if !wait_for_any_connected_relay(&client, Duration::from_secs(2)).await {
425 let _ = client.disconnect().await;
426 return Err(anyhow::anyhow!(
427 "Failed to connect to any relay while listing repos"
428 ));
429 }
430
431 let author = PublicKey::from_hex(&self.pubkey)
432 .map_err(|e| anyhow::anyhow!("Invalid pubkey: {}", e))?;
433 let filter = build_git_repo_list_filter(author);
434
435 let events = match tokio::time::timeout(
436 Duration::from_secs(3),
437 client.get_events_of(vec![filter], EventSource::relays(None)),
438 )
439 .await
440 {
441 Ok(Ok(events)) => events,
442 Ok(Err(e)) => {
443 let _ = client.disconnect().await;
444 return Err(anyhow::anyhow!(
445 "Failed to fetch git repo events from relays: {}",
446 e
447 ));
448 }
449 Err(_) => {
450 let _ = client.disconnect().await;
451 return Err(anyhow::anyhow!(
452 "Timed out fetching git repo events from relays"
453 ));
454 }
455 };
456
457 let _ = client.disconnect().await;
458
459 Ok(list_git_repo_announcements(&events)
460 .into_iter()
461 .map(|repo| repo.repo_name)
462 .collect())
463 }
464
465 pub fn fetch_refs(&mut self, repo_name: &str) -> Result<HashMap<String, String>> {
468 let (refs, _, _) = self.fetch_refs_with_timeout(repo_name, 10)?;
469 Ok(refs)
470 }
471
472 #[allow(dead_code)]
475 pub fn fetch_refs_quick(&mut self, repo_name: &str) -> Result<HashMap<String, String>> {
476 let (refs, _, _) = self.fetch_refs_with_timeout(repo_name, 3)?;
477 Ok(refs)
478 }
479
480 #[allow(dead_code)]
483 pub fn fetch_refs_with_root(&mut self, repo_name: &str) -> Result<FetchedRefs> {
484 self.fetch_refs_with_timeout(repo_name, 10)
485 }
486
487 fn fetch_refs_with_timeout(
489 &mut self,
490 repo_name: &str,
491 timeout_secs: u64,
492 ) -> Result<FetchedRefs> {
493 debug!(
494 "Fetching refs for {} from {} (timeout {}s)",
495 repo_name, self.pubkey, timeout_secs
496 );
497
498 if let Some(refs) = self.cached_refs.get(repo_name) {
500 let root = self.cached_root_hash.get(repo_name).cloned();
501 let key = self.cached_encryption_key.get(repo_name).cloned();
502 return Ok((refs.clone(), root, key));
503 }
504
505 let (refs, root_hash, encryption_key) =
508 block_on_result(self.fetch_refs_async_with_timeout(repo_name, timeout_secs))?;
509 self.cached_refs.insert(repo_name.to_string(), refs.clone());
510 if let Some(ref root) = root_hash {
511 self.cached_root_hash
512 .insert(repo_name.to_string(), root.clone());
513 }
514 if let Some(key) = encryption_key {
515 self.cached_encryption_key
516 .insert(repo_name.to_string(), key);
517 }
518 Ok((refs, root_hash, encryption_key))
519 }
520
521 fn parse_root_event_data_from_event(event: &Event) -> RootEventData {
522 let root_hash = event
523 .tags
524 .iter()
525 .find(|t| t.as_slice().len() >= 2 && t.as_slice()[0].as_str() == "hash")
526 .map(|t| t.as_slice()[1].to_string())
527 .unwrap_or_else(|| event.content.to_string());
528
529 let (encryption_key, key_tag_name, self_encrypted_ciphertext) = event
530 .tags
531 .iter()
532 .find_map(|t| {
533 let slice = t.as_slice();
534 if slice.len() < 2 {
535 return None;
536 }
537 let tag_name = slice[0].as_str();
538 let tag_value = slice[1].to_string();
539 if tag_name == "selfEncryptedKey" {
540 return Some((None, Some(tag_name.to_string()), Some(tag_value)));
541 }
542 if tag_name == "key" || tag_name == "encryptedKey" {
543 if let Ok(bytes) = hex::decode(&tag_value) {
544 if bytes.len() == 32 {
545 let mut key = [0u8; 32];
546 key.copy_from_slice(&bytes);
547 return Some((Some(key), Some(tag_name.to_string()), None));
548 }
549 }
550 }
551 None
552 })
553 .unwrap_or((None, None, None));
554
555 RootEventData {
556 root_hash,
557 encryption_key,
558 key_tag_name,
559 self_encrypted_ciphertext,
560 }
561 }
562
563 fn parse_daemon_response_to_root_data(
564 response: DaemonResolveResponse,
565 ) -> Option<RootEventData> {
566 let root_hash = response.hash?;
567 if root_hash.is_empty() {
568 return None;
569 }
570
571 let mut data = RootEventData {
572 root_hash,
573 encryption_key: None,
574 key_tag_name: None,
575 self_encrypted_ciphertext: None,
576 };
577
578 if let Some(ciphertext) = response.self_encrypted_key {
579 data.key_tag_name = Some("selfEncryptedKey".to_string());
580 data.self_encrypted_ciphertext = Some(ciphertext);
581 return Some(data);
582 }
583
584 let (tag_name, tag_value) = if let Some(v) = response.encrypted_key {
585 ("encryptedKey", v)
586 } else if let Some(v) = response.key {
587 ("key", v)
588 } else {
589 return Some(data);
590 };
591
592 if let Ok(bytes) = hex::decode(&tag_value) {
593 if bytes.len() == 32 {
594 let mut key = [0u8; 32];
595 key.copy_from_slice(&bytes);
596 data.encryption_key = Some(key);
597 data.key_tag_name = Some(tag_name.to_string());
598 }
599 }
600
601 Some(data)
602 }
603
604 async fn fetch_root_from_local_daemon(
605 &self,
606 repo_name: &str,
607 timeout: Duration,
608 ) -> Option<RootEventData> {
609 let base = self.local_daemon_url.as_ref()?;
610 let url = format!(
611 "{}/api/nostr/resolve/{}/{}",
612 base.trim_end_matches('/'),
613 self.pubkey,
614 repo_name
615 );
616
617 let client = reqwest::Client::builder().timeout(timeout).build().ok()?;
618 let response = client.get(&url).send().await.ok()?;
619 if !response.status().is_success() {
620 return None;
621 }
622
623 let payload: DaemonResolveResponse = response.json().await.ok()?;
624 let source = payload
625 .source
626 .clone()
627 .unwrap_or_else(|| "unknown".to_string());
628 let parsed = Self::parse_daemon_response_to_root_data(payload)?;
629 debug!(
630 "Resolved repo {} via local daemon source={}",
631 repo_name, source
632 );
633 Some(parsed)
634 }
635
636 async fn fetch_refs_async_with_timeout(
637 &self,
638 repo_name: &str,
639 timeout_secs: u64,
640 ) -> Result<(HashMap<String, String>, Option<String>, Option<[u8; 32]>)> {
641 let client = Client::default();
643
644 for relay in &self.relays {
646 if let Err(e) = client.add_relay(relay).await {
647 warn!("Failed to add relay {}: {}", relay, e);
648 }
649 }
650
651 client.connect().await;
653
654 let connect_timeout = Duration::from_secs(2);
655 let query_timeout = Duration::from_secs(timeout_secs.saturating_sub(2).max(3));
656 let local_daemon_timeout = Duration::from_secs(4);
657 let retry_delay = Duration::from_millis(300);
658 let max_attempts = 2;
659
660 let start = std::time::Instant::now();
661
662 let author = PublicKey::from_hex(&self.pubkey)
664 .map_err(|e| anyhow::anyhow!("Invalid pubkey: {}", e))?;
665
666 let filter = build_repo_event_filter(author, repo_name);
667
668 debug!("Querying relays for repo {} events", repo_name);
669
670 let mut root_data = None;
671 for attempt in 1..=max_attempts {
672 let connect_start = std::time::Instant::now();
675 let mut last_log = std::time::Instant::now();
676 let mut has_connected_relay = false;
677 loop {
678 let (connected, total) = connected_relay_count(&client).await;
679 if connected > 0 {
680 debug!(
681 "Connected to {}/{} relay(s) in {:?} (attempt {}/{})",
682 connected,
683 total,
684 start.elapsed(),
685 attempt,
686 max_attempts
687 );
688 has_connected_relay = true;
689 break;
690 }
691 if last_log.elapsed() > Duration::from_millis(500) {
692 debug!(
693 "Connecting to relays... (0/{} after {:?}, attempt {}/{})",
694 total,
695 start.elapsed(),
696 attempt,
697 max_attempts
698 );
699 last_log = std::time::Instant::now();
700 }
701 if connect_start.elapsed() > connect_timeout {
702 debug!(
703 "Timeout waiting for relay connections - continuing with local-daemon fallback"
704 );
705 break;
706 }
707 tokio::time::sleep(Duration::from_millis(50)).await;
708 }
709
710 let events = if has_connected_relay {
714 match client
715 .get_events_of(
716 vec![filter.clone()],
717 EventSource::relays(Some(query_timeout)),
718 )
719 .await
720 {
721 Ok(events) => events,
722 Err(e) => {
723 warn!("Failed to fetch events: {}", e);
724 vec![]
725 }
726 }
727 } else {
728 vec![]
729 };
730
731 debug!(
732 "Got {} events from relays on attempt {}/{}",
733 events.len(),
734 attempt,
735 max_attempts
736 );
737 let relay_event = pick_latest_repo_event(events.iter(), repo_name);
738
739 if let Some(event) = relay_event {
740 debug!(
741 "Found relay event with root hash: {}",
742 &event.content[..12.min(event.content.len())]
743 );
744 root_data = Some(Self::parse_root_event_data_from_event(event));
745 break;
746 }
747
748 if let Some(data) = self
749 .fetch_root_from_local_daemon(repo_name, local_daemon_timeout)
750 .await
751 {
752 root_data = Some(data);
753 break;
754 }
755
756 if attempt < max_attempts {
757 debug!(
758 "No hashtree event found for {} on attempt {}/{}; retrying",
759 repo_name, attempt, max_attempts
760 );
761 tokio::time::sleep(retry_delay).await;
762 }
763 }
764
765 let _ = client.disconnect().await;
767
768 let root_data = match root_data {
769 Some(data) => data,
770 None => {
771 anyhow::bail!(
772 "Repository '{}' not found (no hashtree event published by {})",
773 repo_name,
774 Self::format_repo_author(&self.pubkey)
775 );
776 }
777 };
778
779 let root_hash = root_data.root_hash;
780
781 if root_hash.is_empty() {
782 debug!("Empty root hash in event");
783 return Ok((HashMap::new(), None, None));
784 }
785
786 let encryption_key = root_data.encryption_key;
787 let key_tag_name = root_data.key_tag_name;
788 let self_encrypted_ciphertext = root_data.self_encrypted_ciphertext;
789
790 let unmasked_key = match key_tag_name.as_deref() {
792 Some("encryptedKey") => {
793 if let (Some(masked), Some(secret)) = (encryption_key, self.url_secret) {
795 let mut unmasked = [0u8; 32];
796 for i in 0..32 {
797 unmasked[i] = masked[i] ^ secret[i];
798 }
799 Some(unmasked)
800 } else {
801 anyhow::bail!(
802 "This repo is link-visible and requires a secret key.\n\
803 Use: htree://.../{repo_name}#k=<secret>\n\
804 Ask the repo owner for the full URL with the secret."
805 );
806 }
807 }
808 Some("selfEncryptedKey") => {
809 if !self.is_private {
811 anyhow::bail!(
812 "This repo is private (author-only).\n\
813 Use: htree://.../{repo_name}#private\n\
814 Only the author can access this repo."
815 );
816 }
817
818 if let Some(keys) = &self.keys {
820 if let Some(ciphertext) = self_encrypted_ciphertext {
821 let pubkey = keys.public_key();
823 match nip44::decrypt(keys.secret_key(), &pubkey, &ciphertext) {
824 Ok(key_hex) => {
825 let key_bytes =
826 hex::decode(&key_hex).context("Invalid decrypted key hex")?;
827 if key_bytes.len() != 32 {
828 anyhow::bail!("Decrypted key wrong length");
829 }
830 let mut key = [0u8; 32];
831 key.copy_from_slice(&key_bytes);
832 Some(key)
833 }
834 Err(e) => {
835 anyhow::bail!(
836 "Failed to decrypt private repo: {}\n\
837 The repo may be corrupted or published with a different key.",
838 e
839 );
840 }
841 }
842 } else {
843 anyhow::bail!("selfEncryptedKey tag has invalid format");
844 }
845 } else {
846 anyhow::bail!(
847 "Cannot access this private repo.\n\
848 Private repos can only be accessed by their author.\n\
849 You don't have the secret key for this repo's owner."
850 );
851 }
852 }
853 Some("key") | None => {
854 encryption_key
856 }
857 Some(other) => {
858 warn!("Unknown key tag type: {}", other);
859 encryption_key
860 }
861 };
862
863 info!(
864 "Found root hash {} for {} (encrypted: {}, link_visible: {})",
865 &root_hash[..12.min(root_hash.len())],
866 repo_name,
867 unmasked_key.is_some(),
868 self.url_secret.is_some()
869 );
870
871 let refs = self
873 .fetch_refs_from_hashtree(&root_hash, unmasked_key.as_ref())
874 .await?;
875 Ok((refs, Some(root_hash), unmasked_key))
876 }
877
878 fn decrypt_and_decode(
880 &self,
881 data: &[u8],
882 key: Option<&[u8; 32]>,
883 ) -> Option<hashtree_core::TreeNode> {
884 let decrypted_data: Vec<u8>;
885 let data_to_decode = if let Some(k) = key {
886 match decrypt_chk(data, k) {
887 Ok(d) => {
888 decrypted_data = d;
889 &decrypted_data
890 }
891 Err(e) => {
892 debug!("Decryption failed: {}", e);
893 return None;
894 }
895 }
896 } else {
897 data
898 };
899
900 match decode_tree_node(data_to_decode) {
901 Ok(node) => Some(node),
902 Err(e) => {
903 debug!("Failed to decode tree node: {}", e);
904 None
905 }
906 }
907 }
908
909 async fn fetch_refs_from_hashtree(
912 &self,
913 root_hash: &str,
914 encryption_key: Option<&[u8; 32]>,
915 ) -> Result<HashMap<String, String>> {
916 let mut refs = HashMap::new();
917 debug!(
918 "fetch_refs_from_hashtree: downloading root {}",
919 &root_hash[..12]
920 );
921
922 let root_data = match self.blossom.download(root_hash).await {
924 Ok(data) => {
925 debug!("Downloaded {} bytes from blossom", data.len());
926 data
927 }
928 Err(e) => {
929 anyhow::bail!(
930 "Failed to download root hash {}: {}",
931 &root_hash[..12.min(root_hash.len())],
932 e
933 );
934 }
935 };
936
937 let root_node = match self.decrypt_and_decode(&root_data, encryption_key) {
939 Some(node) => {
940 debug!("Decoded root node with {} links", node.links.len());
941 node
942 }
943 None => {
944 debug!(
945 "Failed to decode root node (encryption_key: {})",
946 encryption_key.is_some()
947 );
948 return Ok(refs);
949 }
950 };
951
952 debug!(
954 "Root links: {:?}",
955 root_node
956 .links
957 .iter()
958 .map(|l| l.name.as_deref())
959 .collect::<Vec<_>>()
960 );
961 let git_link = root_node
962 .links
963 .iter()
964 .find(|l| l.name.as_deref() == Some(".git"));
965 let (git_hash, git_key) = match git_link {
966 Some(link) => {
967 debug!("Found .git link with key: {}", link.key.is_some());
968 (hex::encode(link.hash), link.key)
969 }
970 None => {
971 debug!("No .git directory in hashtree root");
972 return Ok(refs);
973 }
974 };
975
976 let git_data = match self.blossom.download(&git_hash).await {
978 Ok(data) => data,
979 Err(e) => {
980 anyhow::bail!(
981 "Failed to download .git directory ({}): {}",
982 &git_hash[..12],
983 e
984 );
985 }
986 };
987
988 let git_node = match self.decrypt_and_decode(&git_data, git_key.as_ref()) {
989 Some(node) => {
990 debug!(
991 "Decoded .git node with {} links: {:?}",
992 node.links.len(),
993 node.links
994 .iter()
995 .map(|l| l.name.as_deref())
996 .collect::<Vec<_>>()
997 );
998 node
999 }
1000 None => {
1001 debug!("Failed to decode .git node (key: {})", git_key.is_some());
1002 return Ok(refs);
1003 }
1004 };
1005
1006 let refs_link = git_node
1008 .links
1009 .iter()
1010 .find(|l| l.name.as_deref() == Some("refs"));
1011 let (refs_hash, refs_key) = match refs_link {
1012 Some(link) => (hex::encode(link.hash), link.key),
1013 None => {
1014 debug!("No refs directory in .git");
1015 return Ok(refs);
1016 }
1017 };
1018
1019 let refs_data = match self.blossom.try_download(&refs_hash).await {
1021 Some(data) => data,
1022 None => {
1023 debug!("Could not download refs directory");
1024 return Ok(refs);
1025 }
1026 };
1027
1028 let refs_node = match self.decrypt_and_decode(&refs_data, refs_key.as_ref()) {
1029 Some(node) => node,
1030 None => {
1031 return Ok(refs);
1032 }
1033 };
1034
1035 if let Some(head_link) = git_node
1037 .links
1038 .iter()
1039 .find(|l| l.name.as_deref() == Some("HEAD"))
1040 {
1041 let head_hash = hex::encode(head_link.hash);
1042 if let Some(head_data) = self.blossom.try_download(&head_hash).await {
1043 let head_content = if let Some(k) = head_link.key.as_ref() {
1045 match decrypt_chk(&head_data, k) {
1046 Ok(d) => String::from_utf8_lossy(&d).trim().to_string(),
1047 Err(_) => String::from_utf8_lossy(&head_data).trim().to_string(),
1048 }
1049 } else {
1050 String::from_utf8_lossy(&head_data).trim().to_string()
1051 };
1052 refs.insert("HEAD".to_string(), head_content);
1053 }
1054 }
1055
1056 for subdir_link in &refs_node.links {
1058 if subdir_link.link_type != LinkType::Dir {
1059 continue;
1060 }
1061 let subdir_name = match &subdir_link.name {
1062 Some(n) => n.clone(),
1063 None => continue,
1064 };
1065 let subdir_hash = hex::encode(subdir_link.hash);
1066
1067 self.collect_refs_recursive(
1068 &subdir_hash,
1069 subdir_link.key.as_ref(),
1070 &format!("refs/{}", subdir_name),
1071 &mut refs,
1072 )
1073 .await;
1074 }
1075
1076 debug!("Found {} refs from hashtree", refs.len());
1077 Ok(refs)
1078 }
1079
1080 async fn collect_refs_recursive(
1082 &self,
1083 dir_hash: &str,
1084 dir_key: Option<&[u8; 32]>,
1085 prefix: &str,
1086 refs: &mut HashMap<String, String>,
1087 ) {
1088 let dir_data = match self.blossom.try_download(dir_hash).await {
1089 Some(data) => data,
1090 None => return,
1091 };
1092
1093 let dir_node = match self.decrypt_and_decode(&dir_data, dir_key) {
1094 Some(node) => node,
1095 None => return,
1096 };
1097
1098 for link in &dir_node.links {
1099 let name = match &link.name {
1100 Some(n) => n.clone(),
1101 None => continue,
1102 };
1103 let link_hash = hex::encode(link.hash);
1104 let ref_path = format!("{}/{}", prefix, name);
1105
1106 if link.link_type == LinkType::Dir {
1107 Box::pin(self.collect_refs_recursive(
1109 &link_hash,
1110 link.key.as_ref(),
1111 &ref_path,
1112 refs,
1113 ))
1114 .await;
1115 } else {
1116 if let Some(ref_data) = self.blossom.try_download(&link_hash).await {
1118 let sha = if let Some(k) = link.key.as_ref() {
1120 match decrypt_chk(&ref_data, k) {
1121 Ok(d) => String::from_utf8_lossy(&d).trim().to_string(),
1122 Err(_) => String::from_utf8_lossy(&ref_data).trim().to_string(),
1123 }
1124 } else {
1125 String::from_utf8_lossy(&ref_data).trim().to_string()
1126 };
1127 if !sha.is_empty() {
1128 debug!("Found ref {} -> {}", ref_path, sha);
1129 refs.insert(ref_path, sha);
1130 }
1131 }
1132 }
1133 }
1134 }
1135
1136 #[allow(dead_code)]
1138 pub fn update_ref(&mut self, repo_name: &str, ref_name: &str, sha: &str) -> Result<()> {
1139 info!("Updating ref {} -> {} for {}", ref_name, sha, repo_name);
1140
1141 let refs = self.cached_refs.entry(repo_name.to_string()).or_default();
1142 refs.insert(ref_name.to_string(), sha.to_string());
1143
1144 Ok(())
1145 }
1146
1147 pub fn delete_ref(&mut self, repo_name: &str, ref_name: &str) -> Result<()> {
1149 info!("Deleting ref {} for {}", ref_name, repo_name);
1150
1151 if let Some(refs) = self.cached_refs.get_mut(repo_name) {
1152 refs.remove(ref_name);
1153 }
1154
1155 Ok(())
1156 }
1157
1158 pub fn get_cached_root_hash(&self, repo_name: &str) -> Option<&String> {
1160 self.cached_root_hash.get(repo_name)
1161 }
1162
1163 pub fn get_cached_encryption_key(&self, repo_name: &str) -> Option<&[u8; 32]> {
1165 self.cached_encryption_key.get(repo_name)
1166 }
1167
1168 pub fn blossom(&self) -> &BlossomClient {
1170 &self.blossom
1171 }
1172
1173 pub fn relay_urls(&self) -> Vec<String> {
1175 self.relays.clone()
1176 }
1177
1178 #[allow(dead_code)]
1180 pub fn pubkey(&self) -> &str {
1181 &self.pubkey
1182 }
1183
1184 pub fn npub(&self) -> String {
1186 PublicKey::from_hex(&self.pubkey)
1187 .ok()
1188 .and_then(|pk| pk.to_bech32().ok())
1189 .unwrap_or_else(|| self.pubkey.clone())
1190 }
1191
1192 pub fn publish_repo(
1200 &self,
1201 repo_name: &str,
1202 root_hash: &str,
1203 encryption_key: Option<(&[u8; 32], bool, bool)>,
1204 ) -> Result<(String, RelayResult)> {
1205 let keys = self.keys.as_ref().context(format!(
1206 "Cannot push: no secret key for {}. You can only push to your own repos.",
1207 &self.pubkey[..16]
1208 ))?;
1209
1210 info!(
1211 "Publishing repo {} with root hash {} (encrypted: {})",
1212 repo_name,
1213 root_hash,
1214 encryption_key.is_some()
1215 );
1216
1217 block_on_result(self.publish_repo_async(keys, repo_name, root_hash, encryption_key))
1219 }
1220
1221 async fn publish_repo_async(
1222 &self,
1223 keys: &Keys,
1224 repo_name: &str,
1225 root_hash: &str,
1226 encryption_key: Option<(&[u8; 32], bool, bool)>,
1227 ) -> Result<(String, RelayResult)> {
1228 let client = Client::new(keys.clone());
1230
1231 let configured: Vec<String> = self.relays.clone();
1232 let mut connected: Vec<String> = Vec::new();
1233 let mut failed: Vec<String> = Vec::new();
1234
1235 for relay in &self.relays {
1237 if let Err(e) = client.add_relay(relay).await {
1238 warn!("Failed to add relay {}: {}", relay, e);
1239 failed.push(relay.clone());
1240 }
1241 }
1242
1243 client.connect().await;
1245
1246 let _ = wait_for_any_connected_relay(&client, Duration::from_secs(3)).await;
1248
1249 let publish_created_at = next_replaceable_created_at(
1250 Timestamp::now(),
1251 latest_repo_event_created_at(
1252 &client,
1253 keys.public_key(),
1254 repo_name,
1255 Duration::from_secs(2),
1256 )
1257 .await,
1258 );
1259
1260 let mut tags = vec![
1262 Tag::custom(TagKind::custom("d"), vec![repo_name.to_string()]),
1263 Tag::custom(TagKind::custom("l"), vec![LABEL_HASHTREE.to_string()]),
1264 Tag::custom(TagKind::custom("hash"), vec![root_hash.to_string()]),
1265 ];
1266
1267 if let Some((key, is_link_visible, is_self_private)) = encryption_key {
1273 if is_self_private {
1274 let pubkey = keys.public_key();
1276 let key_hex = hex::encode(key);
1277 let encrypted =
1278 nip44::encrypt(keys.secret_key(), &pubkey, &key_hex, nip44::Version::V2)
1279 .map_err(|e| anyhow::anyhow!("NIP-44 encryption failed: {}", e))?;
1280 tags.push(Tag::custom(
1281 TagKind::custom("selfEncryptedKey"),
1282 vec![encrypted],
1283 ));
1284 } else if is_link_visible {
1285 tags.push(Tag::custom(
1287 TagKind::custom("encryptedKey"),
1288 vec![hex::encode(key)],
1289 ));
1290 } else {
1291 tags.push(Tag::custom(TagKind::custom("key"), vec![hex::encode(key)]));
1293 }
1294 }
1295
1296 append_repo_discovery_labels(&mut tags, repo_name);
1297
1298 let event = EventBuilder::new(Kind::Custom(KIND_APP_DATA), root_hash, tags)
1300 .custom_created_at(publish_created_at)
1301 .to_event(keys)
1302 .map_err(|e| anyhow::anyhow!("Failed to sign event: {}", e))?;
1303
1304 match client.send_event(event.clone()).await {
1306 Ok(output) => {
1307 for url in output.success.iter() {
1309 let url_str = url.to_string();
1310 if !connected.contains(&url_str) {
1311 connected.push(url_str);
1312 }
1313 }
1314 for (url, err) in output.failed.iter() {
1316 if err.is_some() {
1317 let url_str = url.to_string();
1318 if !failed.contains(&url_str) && !connected.contains(&url_str) {
1319 failed.push(url_str);
1320 }
1321 }
1322 }
1323 info!(
1324 "Sent event {} to {} relays ({} failed)",
1325 output.id(),
1326 output.success.len(),
1327 output.failed.len()
1328 );
1329 }
1330 Err(e) => {
1331 warn!("Failed to send event: {}", e);
1332 for relay in &self.relays {
1334 if !failed.contains(relay) {
1335 failed.push(relay.clone());
1336 }
1337 }
1338 }
1339 };
1340
1341 let npub_url = keys
1343 .public_key()
1344 .to_bech32()
1345 .map(|npub| format!("htree://{}/{}", npub, repo_name))
1346 .unwrap_or_else(|_| format!("htree://{}/{}", &self.pubkey[..16], repo_name));
1347
1348 let relay_validation = validate_repo_publish_relays(&configured, &connected);
1349
1350 let _ = client.disconnect().await;
1352 tokio::time::sleep(Duration::from_millis(50)).await;
1353
1354 relay_validation?;
1355
1356 Ok((
1357 npub_url,
1358 RelayResult {
1359 configured,
1360 connected,
1361 failed,
1362 },
1363 ))
1364 }
1365
1366 pub fn fetch_prs(
1368 &self,
1369 repo_name: &str,
1370 state_filter: PullRequestStateFilter,
1371 ) -> Result<Vec<PullRequestListItem>> {
1372 block_on_result(self.fetch_prs_async(repo_name, state_filter))
1373 }
1374
1375 pub async fn fetch_prs_async(
1376 &self,
1377 repo_name: &str,
1378 state_filter: PullRequestStateFilter,
1379 ) -> Result<Vec<PullRequestListItem>> {
1380 let client = Client::default();
1381
1382 for relay in &self.relays {
1383 if let Err(e) = client.add_relay(relay).await {
1384 warn!("Failed to add relay {}: {}", relay, e);
1385 }
1386 }
1387 client.connect().await;
1388
1389 if !wait_for_any_connected_relay(&client, Duration::from_secs(2)).await {
1391 let _ = client.disconnect().await;
1392 return Err(anyhow::anyhow!(
1393 "Failed to connect to any relay while fetching PRs"
1394 ));
1395 }
1396
1397 let repo_address = format!("{}:{}:{}", KIND_REPO_ANNOUNCEMENT, self.pubkey, repo_name);
1399 let pull_request_filter = Filter::new()
1400 .kind(Kind::Custom(KIND_PULL_REQUEST))
1401 .custom_tag(SingleLetterTag::lowercase(Alphabet::A), vec![&repo_address]);
1402
1403 let mut pr_events = match tokio::time::timeout(
1404 Duration::from_secs(3),
1405 client.get_events_of(vec![pull_request_filter.clone()], EventSource::relays(None)),
1406 )
1407 .await
1408 {
1409 Ok(Ok(events)) => events,
1410 Ok(Err(e)) => {
1411 let _ = client.disconnect().await;
1412 return Err(anyhow::anyhow!(
1413 "Failed to fetch PR events from relays: {}",
1414 e
1415 ));
1416 }
1417 Err(_) => {
1418 let _ = client.disconnect().await;
1419 return Err(anyhow::anyhow!("Timed out fetching PR events from relays"));
1420 }
1421 };
1422
1423 if pr_events.is_empty() {
1424 let fallback_events = fetch_events_via_raw_relay_query(
1425 &self.relays,
1426 pull_request_filter,
1427 Duration::from_secs(3),
1428 )
1429 .await;
1430 if !fallback_events.is_empty() {
1431 debug!(
1432 "Raw relay fallback recovered {} PR event(s) for {}",
1433 fallback_events.len(),
1434 repo_name
1435 );
1436 pr_events = fallback_events;
1437 }
1438 }
1439
1440 if pr_events.is_empty() {
1441 let _ = client.disconnect().await;
1442 return Ok(Vec::new());
1443 }
1444
1445 let pr_ids: Vec<String> = pr_events.iter().map(|e| e.id.to_hex()).collect();
1447
1448 let status_event_filter = Filter::new()
1450 .kinds(vec![
1451 Kind::Custom(KIND_STATUS_OPEN),
1452 Kind::Custom(KIND_STATUS_APPLIED),
1453 Kind::Custom(KIND_STATUS_CLOSED),
1454 Kind::Custom(KIND_STATUS_DRAFT),
1455 ])
1456 .custom_tag(
1457 SingleLetterTag::lowercase(Alphabet::E),
1458 pr_ids.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
1459 );
1460
1461 let mut status_events = match tokio::time::timeout(
1462 Duration::from_secs(3),
1463 client.get_events_of(vec![status_event_filter.clone()], EventSource::relays(None)),
1464 )
1465 .await
1466 {
1467 Ok(Ok(events)) => events,
1468 Ok(Err(e)) => {
1469 let _ = client.disconnect().await;
1470 return Err(anyhow::anyhow!(
1471 "Failed to fetch PR status events from relays: {}",
1472 e
1473 ));
1474 }
1475 Err(_) => {
1476 let _ = client.disconnect().await;
1477 return Err(anyhow::anyhow!(
1478 "Timed out fetching PR status events from relays"
1479 ));
1480 }
1481 };
1482
1483 if status_events.is_empty() {
1484 let fallback_events = fetch_events_via_raw_relay_query(
1485 &self.relays,
1486 status_event_filter,
1487 Duration::from_secs(3),
1488 )
1489 .await;
1490 if !fallback_events.is_empty() {
1491 debug!(
1492 "Raw relay fallback recovered {} PR status event(s) for {}",
1493 fallback_events.len(),
1494 repo_name
1495 );
1496 status_events = fallback_events;
1497 }
1498 }
1499
1500 let _ = client.disconnect().await;
1501
1502 let latest_status =
1504 latest_trusted_pr_status_kinds(&pr_events, &status_events, &self.pubkey);
1505
1506 let mut prs = Vec::new();
1507 for event in &pr_events {
1508 let pr_id = event.id.to_hex();
1509 let state =
1510 PullRequestState::from_latest_status_kind(latest_status.get(&pr_id).copied());
1511 if !state_filter.includes(state) {
1512 continue;
1513 }
1514
1515 let mut subject = None;
1516 let mut commit_tip = None;
1517 let mut branch = None;
1518 let mut target_branch = None;
1519
1520 for tag in event.tags.iter() {
1521 let slice = tag.as_slice();
1522 if slice.len() >= 2 {
1523 match slice[0].as_str() {
1524 "subject" => subject = Some(slice[1].to_string()),
1525 "c" => commit_tip = Some(slice[1].to_string()),
1526 "branch" => branch = Some(slice[1].to_string()),
1527 "target-branch" => target_branch = Some(slice[1].to_string()),
1528 _ => {}
1529 }
1530 }
1531 }
1532
1533 prs.push(PullRequestListItem {
1534 event_id: pr_id,
1535 author_pubkey: event.pubkey.to_hex(),
1536 state,
1537 subject,
1538 commit_tip,
1539 branch,
1540 target_branch,
1541 created_at: event.created_at.as_u64(),
1542 });
1543 }
1544
1545 prs.sort_by(|left, right| {
1547 right
1548 .created_at
1549 .cmp(&left.created_at)
1550 .then_with(|| right.event_id.cmp(&left.event_id))
1551 });
1552
1553 debug!(
1554 "Found {} PRs for {} (filter: {:?})",
1555 prs.len(),
1556 repo_name,
1557 state_filter
1558 );
1559 Ok(prs)
1560 }
1561
1562 pub fn publish_pr_merged_status(
1564 &self,
1565 pr_event_id: &str,
1566 pr_author_pubkey: &str,
1567 ) -> Result<()> {
1568 let keys = self
1569 .keys
1570 .as_ref()
1571 .context("Cannot publish status: no secret key")?;
1572
1573 block_on_result(self.publish_pr_merged_status_async(keys, pr_event_id, pr_author_pubkey))
1574 }
1575
1576 async fn publish_pr_merged_status_async(
1577 &self,
1578 keys: &Keys,
1579 pr_event_id: &str,
1580 pr_author_pubkey: &str,
1581 ) -> Result<()> {
1582 let client = Client::new(keys.clone());
1583
1584 for relay in &self.relays {
1585 if let Err(e) = client.add_relay(relay).await {
1586 warn!("Failed to add relay {}: {}", relay, e);
1587 }
1588 }
1589 client.connect().await;
1590
1591 if !wait_for_any_connected_relay(&client, Duration::from_secs(3)).await {
1593 anyhow::bail!("Failed to connect to any relay for status publish");
1594 }
1595
1596 let tags = vec![
1597 Tag::custom(TagKind::custom("e"), vec![pr_event_id.to_string()]),
1598 Tag::custom(TagKind::custom("p"), vec![pr_author_pubkey.to_string()]),
1599 ];
1600
1601 let event = EventBuilder::new(Kind::Custom(KIND_STATUS_APPLIED), "", tags)
1602 .to_event(keys)
1603 .map_err(|e| anyhow::anyhow!("Failed to sign status event: {}", e))?;
1604
1605 let publish_result = match client.send_event(event).await {
1606 Ok(output) => {
1607 if output.success.is_empty() {
1608 Err(anyhow::anyhow!(
1609 "PR merged status was not confirmed by any relay"
1610 ))
1611 } else {
1612 info!(
1613 "Published PR merged status to {} relays",
1614 output.success.len()
1615 );
1616 Ok(())
1617 }
1618 }
1619 Err(e) => Err(anyhow::anyhow!("Failed to publish PR merged status: {}", e)),
1620 };
1621
1622 let _ = client.disconnect().await;
1623 tokio::time::sleep(Duration::from_millis(50)).await;
1624 publish_result
1625 }
1626
1627 #[allow(dead_code)]
1629 pub async fn upload_blob(&self, _hash: &str, data: &[u8]) -> Result<String> {
1630 let hash = self
1631 .blossom
1632 .upload(data)
1633 .await
1634 .map_err(|e| anyhow::anyhow!("Blossom upload failed: {}", e))?;
1635 Ok(hash)
1636 }
1637
1638 #[allow(dead_code)]
1640 pub async fn upload_blob_if_missing(&self, data: &[u8]) -> Result<(String, bool)> {
1641 self.blossom
1642 .upload_if_missing(data)
1643 .await
1644 .map_err(|e| anyhow::anyhow!("Blossom upload failed: {}", e))
1645 }
1646
1647 #[allow(dead_code)]
1649 pub async fn download_blob(&self, hash: &str) -> Result<Vec<u8>> {
1650 self.blossom
1651 .download(hash)
1652 .await
1653 .map_err(|e| anyhow::anyhow!("Blossom download failed: {}", e))
1654 }
1655
1656 #[allow(dead_code)]
1658 pub async fn try_download_blob(&self, hash: &str) -> Option<Vec<u8>> {
1659 self.blossom.try_download(hash).await
1660 }
1661}
1662
1663#[cfg(test)]
1664mod tests;