upleb.uk

Public git repos — served from a NIP-34 GRASP relay at git.upleb.uk

summaryrefslogtreecommitdiff
path: root/src/bin/git_remote_nostr/push.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/bin/git_remote_nostr/push.rs')
-rw-r--r--src/bin/git_remote_nostr/push.rs308
1 files changed, 6 insertions, 302 deletions
diff --git a/src/bin/git_remote_nostr/push.rs b/src/bin/git_remote_nostr/push.rs
index 74e98f9..4e0d760 100644
--- a/src/bin/git_remote_nostr/push.rs
+++ b/src/bin/git_remote_nostr/push.rs
@@ -2,12 +2,10 @@ use core::str;
2use std::{ 2use std::{
3 collections::{HashMap, HashSet}, 3 collections::{HashMap, HashSet},
4 io::Stdin, 4 io::Stdin,
5 sync::{Arc, Mutex}, 5 sync::Arc,
6 time::Instant,
7}; 6};
8 7
9use anyhow::{Context, Result, anyhow, bail}; 8use anyhow::{Context, Result, bail};
10use auth_git2::GitAuthenticator;
11use client::{get_events_from_local_cache, get_state_from_cache, send_events, sign_event}; 9use client::{get_events_from_local_cache, get_state_from_cache, send_events, sign_event};
12use console::Term; 10use console::Term;
13use git::{RepoActions, sha1_to_oid}; 11use git::{RepoActions, sha1_to_oid};
@@ -17,22 +15,17 @@ use git_events::{
17}; 15};
18use git2::{Oid, Repository}; 16use git2::{Oid, Repository};
19use ngit::{ 17use ngit::{
20 cli_interactor::count_lines_per_msg_vec,
21 client::{self, get_event_from_cache_by_id, sign_draft_event}, 18 client::{self, get_event_from_cache_by_id, sign_draft_event},
22 git::{ 19 git::{self, nostr_url::NostrUrlDecoded},
23 self,
24 nostr_url::{CloneUrl, NostrUrlDecoded},
25 oid_to_shorthand_string,
26 },
27 git_events::{self, KIND_PULL_REQUEST, event_to_cover_letter, get_event_root}, 20 git_events::{self, KIND_PULL_REQUEST, event_to_cover_letter, get_event_root},
28 list::list_from_remotes, 21 list::list_from_remotes,
29 login::{self, user::UserRef}, 22 login::{self, user::UserRef},
23 push::{push_to_remote, push_to_remote_url},
30 repo_ref::{self, get_repo_config_from_yaml, is_grasp_server, normalize_grasp_server_url}, 24 repo_ref::{self, get_repo_config_from_yaml, is_grasp_server, normalize_grasp_server_url},
31 repo_state, 25 repo_state,
32 utils::{ 26 utils::{
33 Direction, find_proposal_and_patches_by_branch_name, get_all_proposals, 27 find_proposal_and_patches_by_branch_name, get_all_proposals, get_remote_name_by_url,
34 get_remote_name_by_url, get_short_git_server_name, get_write_protocols_to_try, 28 get_short_git_server_name, read_line,
35 join_with_and, read_line, set_protocol_preference,
36 }, 29 },
37}; 30};
38use nostr::{event::UnsignedEvent, nips::nip10::Marker}; 31use nostr::{event::UnsignedEvent, nips::nip10::Marker};
@@ -587,295 +580,6 @@ async fn generate_patches_or_pr_event_or_pr_updates(
587 Ok(events) 580 Ok(events)
588} 581}
589 582
590fn push_to_remote(
591 git_repo: &Repo,
592 git_server_url: &str,
593 decoded_nostr_url: &NostrUrlDecoded,
594 remote_refspecs: &[String],
595 term: &Term,
596 is_grasp_server: bool,
597) -> Result<()> {
598 let server_url = git_server_url.parse::<CloneUrl>()?;
599 let protocols_to_attempt =
600 get_write_protocols_to_try(git_repo, &server_url, decoded_nostr_url, is_grasp_server);
601
602 let mut failed_protocols = vec![];
603 let mut success = false;
604
605 for protocol in &protocols_to_attempt {
606 term.write_line(format!("push: {} over {protocol}...", server_url.short_name(),).as_str())?;
607
608 let formatted_url = server_url.format_as(protocol, &decoded_nostr_url.user)?;
609
610 if let Err(error) = push_to_remote_url(git_repo, &formatted_url, remote_refspecs, term) {
611 term.write_line(
612 format!("push: {formatted_url} failed over {protocol}: {error}").as_str(),
613 )?;
614 failed_protocols.push(protocol);
615 } else {
616 success = true;
617 if !failed_protocols.is_empty() {
618 term.write_line(format!("push: succeeded over {protocol}").as_str())?;
619 let _ = set_protocol_preference(git_repo, protocol, &server_url, &Direction::Push);
620 }
621 break;
622 }
623 }
624 if success {
625 Ok(())
626 } else {
627 let error = anyhow!(
628 "{} failed over {}{}",
629 server_url.short_name(),
630 join_with_and(&failed_protocols),
631 if decoded_nostr_url.protocol.is_some() {
632 " and nostr url contains protocol override so no other protocols were attempted"
633 } else {
634 ""
635 },
636 );
637 term.write_line(format!("push: {error}").as_str())?;
638 Err(error)
639 }
640}
641
642fn push_to_remote_url(
643 git_repo: &Repo,
644 git_server_url: &str,
645 remote_refspecs: &[String],
646 term: &Term,
647) -> Result<()> {
648 let git_config = git_repo.git_repo.config()?;
649 let mut git_server_remote = git_repo.git_repo.remote_anonymous(git_server_url)?;
650 let auth = GitAuthenticator::default();
651 let mut push_options = git2::PushOptions::new();
652 let mut remote_callbacks = git2::RemoteCallbacks::new();
653 let push_reporter = Arc::new(Mutex::new(PushReporter::new(term)));
654
655 remote_callbacks.credentials(auth.credentials(&git_config));
656
657 remote_callbacks.push_update_reference({
658 let push_reporter = Arc::clone(&push_reporter);
659 move |name, error| {
660 let mut reporter = push_reporter.lock().unwrap();
661 if let Some(error) = error {
662 let existing_lines = reporter.count_all_existing_lines();
663 reporter.update_reference_errors.push(format!(
664 "WARNING: {} failed to push {name} error: {error}",
665 get_short_git_server_name(git_repo, git_server_url),
666 ));
667 reporter.write_all(existing_lines);
668 }
669 Ok(())
670 }
671 });
672
673 remote_callbacks.push_negotiation({
674 let push_reporter = Arc::clone(&push_reporter);
675 move |updates| {
676 let mut reporter = push_reporter.lock().unwrap();
677 let existing_lines = reporter.count_all_existing_lines();
678
679 for update in updates {
680 let dst_refname = update
681 .dst_refname()
682 .unwrap_or("")
683 .replace("refs/heads/", "")
684 .replace("refs/tags/", "tags/");
685 let msg = if update.dst().is_zero() {
686 format!("push: - [delete] {dst_refname}")
687 } else if update.src().is_zero() {
688 if update.dst_refname().unwrap_or("").contains("refs/tags") {
689 format!("push: * [new tag] {dst_refname}")
690 } else {
691 format!("push: * [new branch] {dst_refname}")
692 }
693 } else {
694 let force = remote_refspecs
695 .iter()
696 .any(|r| r.contains(&dst_refname) && r.contains('+'));
697 format!(
698 "push: {} {}..{} {} -> {dst_refname}",
699 if force { "+" } else { " " },
700 oid_to_shorthand_string(update.src()).unwrap(),
701 oid_to_shorthand_string(update.dst()).unwrap(),
702 update
703 .src_refname()
704 .unwrap_or("")
705 .replace("refs/heads/", "")
706 .replace("refs/tags/", "tags/"),
707 )
708 };
709 // other possibilities will result in push to fail but better reporting is
710 // needed:
711 // deleting a non-existant branch:
712 // ! [remote rejected] <old-branch-name> -> <old-branch-name> (not found)
713 // adding a branch that already exists:
714 // ! [remote rejected] <new-branch-name> -> <new-branch-name> (already exists)
715 // pushing without non-fast-forward without force:
716 // ! [rejected] <branch-name> -> <branch-name> (non-fast-forward)
717 reporter.negotiation.push(msg);
718 }
719 reporter.write_all(existing_lines);
720 Ok(())
721 }
722 });
723
724 remote_callbacks.push_transfer_progress({
725 let push_reporter = Arc::clone(&push_reporter);
726 #[allow(clippy::cast_precision_loss)]
727 move |current, total, bytes| {
728 let mut reporter = push_reporter.lock().unwrap();
729 reporter.process_transfer_progress_update(current, total, bytes);
730 }
731 });
732
733 remote_callbacks.sideband_progress({
734 let push_reporter = Arc::clone(&push_reporter);
735 move |data| {
736 let mut reporter = push_reporter.lock().unwrap();
737 reporter.process_remote_msg(data);
738 true
739 }
740 });
741 push_options.remote_callbacks(remote_callbacks);
742 git_server_remote.push(remote_refspecs, Some(&mut push_options))?;
743 let _ = git_server_remote.disconnect();
744 Ok(())
745}
746
747#[allow(clippy::cast_precision_loss)]
748#[allow(clippy::float_cmp)]
749#[allow(clippy::needless_pass_by_value)]
750fn report_on_transfer_progress(
751 current: usize,
752 total: usize,
753 bytes: usize,
754 start_time: &Instant,
755 end_time: Option<&Instant>,
756) -> Option<String> {
757 if total == 0 {
758 return None;
759 }
760 let percentage = ((current as f64 / total as f64) * 100.0)
761 // always round down because 100% complete is misleading when its not complete
762 .floor();
763 let (size, unit) = if bytes as f64 >= (1024.0 * 1024.0) {
764 (bytes as f64 / (1024.0 * 1024.0), "MiB")
765 } else {
766 (bytes as f64 / 1024.0, "KiB")
767 };
768 let speed = {
769 let duration = if let Some(end_time) = end_time {
770 (*end_time - *start_time).as_millis() as f64
771 } else {
772 start_time.elapsed().as_millis() as f64
773 };
774
775 if duration > 0.0 {
776 (bytes as f64 / (1024.0 * 1024.0)) / (duration / 1000.0) // Convert bytes to MiB and milliseconds to seconds
777 } else {
778 0.0
779 }
780 };
781
782 Some(format!(
783 "push: Writing objects: {percentage}% ({current}/{total}) {size:.2} {unit} | {speed:.2} MiB/s{}",
784 if current == total { ", done." } else { "" },
785 ))
786}
787
788struct PushReporter<'a> {
789 remote_msgs: Vec<String>,
790 negotiation: Vec<String>,
791 transfer_progress_msgs: Vec<String>,
792 update_reference_errors: Vec<String>,
793 term: &'a console::Term,
794 start_time: Option<Instant>,
795 end_time: Option<Instant>,
796}
797impl<'a> PushReporter<'a> {
798 fn new(term: &'a console::Term) -> Self {
799 Self {
800 remote_msgs: vec![],
801 negotiation: vec![],
802 transfer_progress_msgs: vec![],
803 update_reference_errors: vec![],
804 term,
805 start_time: None,
806 end_time: None,
807 }
808 }
809 fn write_all(&self, lines_to_clear: usize) {
810 let _ = self.term.clear_last_lines(lines_to_clear);
811 for msg in &self.remote_msgs {
812 let _ = self.term.write_line(format!("remote: {msg}").as_str());
813 }
814 for msg in &self.negotiation {
815 let _ = self.term.write_line(msg);
816 }
817 for msg in &self.transfer_progress_msgs {
818 let _ = self.term.write_line(msg);
819 }
820 for msg in &self.update_reference_errors {
821 let _ = self.term.write_line(msg);
822 }
823 }
824
825 fn count_all_existing_lines(&self) -> usize {
826 let width = self.term.size().1;
827 count_lines_per_msg_vec(width, &self.remote_msgs, "remote: ".len())
828 + count_lines_per_msg_vec(width, &self.negotiation, 0)
829 + count_lines_per_msg_vec(width, &self.transfer_progress_msgs, 0)
830 + count_lines_per_msg_vec(width, &self.update_reference_errors, 0)
831 }
832 fn process_remote_msg(&mut self, data: &[u8]) {
833 if let Ok(data) = str::from_utf8(data) {
834 let data = data
835 .split(['\n', '\r'])
836 .map(str::trim)
837 .filter(|line| !line.trim().is_empty())
838 .collect::<Vec<&str>>();
839 for data in data {
840 let existing_lines = self.count_all_existing_lines();
841 let msg = data.to_string();
842 if let Some(last) = self.remote_msgs.last() {
843 if (last.contains('%') && !last.contains("100%"))
844 || last == &msg.replace(", done.", "")
845 {
846 self.remote_msgs.pop();
847 self.remote_msgs.push(msg);
848 } else {
849 self.remote_msgs.push(msg);
850 }
851 } else {
852 self.remote_msgs.push(msg);
853 }
854 self.write_all(existing_lines);
855 }
856 }
857 }
858 fn process_transfer_progress_update(&mut self, current: usize, total: usize, bytes: usize) {
859 if self.start_time.is_none() {
860 self.start_time = Some(Instant::now());
861 }
862 if let Some(report) = report_on_transfer_progress(
863 current,
864 total,
865 bytes,
866 &self.start_time.unwrap(),
867 self.end_time.as_ref(),
868 ) {
869 let existing_lines = self.count_all_existing_lines();
870 if report.contains("100%") {
871 self.end_time = Some(Instant::now());
872 }
873 self.transfer_progress_msgs = vec![report];
874 self.write_all(existing_lines);
875 }
876 }
877}
878
879type HashMapUrlRefspecs = HashMap<String, Vec<String>>; 583type HashMapUrlRefspecs = HashMap<String, Vec<String>>;
880 584
881#[allow(clippy::too_many_lines)] 585#[allow(clippy::too_many_lines)]