diff --git a/cli/src/commands.rs b/cli/src/commands.rs index 1b857cfa..0f062210 100644 --- a/cli/src/commands.rs +++ b/cli/src/commands.rs @@ -12,6 +12,8 @@ pub mod ls_files; pub mod ls_tree; pub mod meva_command; pub mod plugins; +pub mod pull; +pub mod push; pub mod remote; pub mod restore; pub mod show; @@ -30,6 +32,8 @@ pub use log::LogCommand; pub use ls_files::LsFilesCommand; pub use ls_tree::LsTreeCommand; pub use plugins::PluginsCommand; +pub use pull::PullCommand; +pub use push::PushCommand; pub use remote::RemoteCommand; pub use restore::RestoreCommand; pub use show::ShowCommand; @@ -41,6 +45,7 @@ use miette::{Context, Result}; use engine::engine_container::MevaContainer; pub use meva_command::MevaCommand; +/// Executes the appropriate subcommand based on the provided matches. pub async fn execute_multiple( matches: &ArgMatches, container: &MevaContainer, @@ -57,3 +62,31 @@ pub async fn execute_multiple( Ok(()) } + +/// Collection type for Meva commands. +pub type CommandsCollection = Vec>>; + +/// Collects all available Meva commands into a single collection. +pub fn collect_commands() -> CommandsCollection { + vec![ + Box::new(InitCommand), + Box::new(ConfigCommand), + Box::new(IgnoreCommand), + Box::new(PluginsCommand), + Box::new(AddCommand), + Box::new(LsFilesCommand), + Box::new(ShowCommand), + Box::new(StatusCommand), + Box::new(CommitCommand), + Box::new(DiffCommand), + Box::new(LsTreeCommand), + Box::new(LogCommand), + Box::new(RestoreCommand), + Box::new(CloneCommand), + Box::new(RemoteCommand), + Box::new(FetchCommand), + Box::new(BranchCommand), + Box::new(PushCommand), + Box::new(PullCommand), + ] +} diff --git a/cli/src/commands/clone.rs b/cli/src/commands/clone.rs index 7991bfdb..7e54937e 100644 --- a/cli/src/commands/clone.rs +++ b/cli/src/commands/clone.rs @@ -1,6 +1,6 @@ -use crate::commands::MevaCommand; +use crate::{commands::MevaCommand, extensions::WithVerbose}; use async_trait::async_trait; -use clap::{Arg, ArgAction, ArgMatches, Command, ValueHint}; +use clap::{Arg, ArgMatches, Command, ValueHint}; use engine::{EngineContainer, engine_container::MevaContainer, handlers::clone::Request}; use miette::{IntoDiagnostic, Result}; use std::path::PathBuf; @@ -26,9 +26,6 @@ impl CloneCommand { /// Argument name for specifying the server public key. const ARG_SERVER_KEY: &'static str = "server-key"; - - /// Argument name for the quiet flag. - const ARG_QUIET: &'static str = "quiet"; } #[async_trait] @@ -50,6 +47,7 @@ impl MevaCommand for CloneCommand { /// Builds the CLI argument parser for the `clone` command using `clap`. fn build_command(&self) -> Command { self.build_base_command() + .with_verbose_arg("Enable verbose output") .arg( Arg::new(Self::ARG_REPOSITORY) .value_name("REPOSITORY") @@ -82,13 +80,6 @@ impl MevaCommand for CloneCommand { .value_parser(clap::value_parser!(PathBuf)) .help("Path to server's public key"), ) - .arg( - Arg::new(Self::ARG_QUIET) - .long(Self::ARG_QUIET) - .short('q') - .action(ArgAction::SetTrue) - .help("Suppress non-error output"), - ) } /// Executes the `clone` command. @@ -107,7 +98,7 @@ impl MevaCommand for CloneCommand { .get_one::(Self::ARG_SERVER_KEY) .unwrap() .clone(), - quiet: matches.get_flag(Self::ARG_QUIET), + quiet: !matches.get_flag(Command::ARG_VERBOSE), }; let handler = container.clone_handler().into_diagnostic()?; diff --git a/cli/src/commands/fetch.rs b/cli/src/commands/fetch.rs index f7f7bbcd..1fcf28eb 100644 --- a/cli/src/commands/fetch.rs +++ b/cli/src/commands/fetch.rs @@ -88,7 +88,7 @@ impl MevaCommand for FetchCommand { }; let handler = container.fetch_handler().into_diagnostic()?; - let _response = handler.handle_fetch(request).await.into_diagnostic()?; + let _ = handler.handle_fetch(request).await.into_diagnostic()?; Ok(()) } diff --git a/cli/src/commands/pull.rs b/cli/src/commands/pull.rs new file mode 100644 index 00000000..71167a6e --- /dev/null +++ b/cli/src/commands/pull.rs @@ -0,0 +1,85 @@ +use async_trait::async_trait; +use clap::{Arg, ArgMatches, Command}; +use miette::Result; + +use engine::engine_container::MevaContainer; + +use crate::{commands::MevaCommand, extensions::WithVerbose}; + +/// Implements the `pull` command for Meva DVCS. +#[derive(Default)] +pub struct PullCommand; + +impl PullCommand { + const ARG_ORIGIN: &'static str = "origin"; + + const ARG_BRANCH: &'static str = "branch"; +} + +#[async_trait] +impl MevaCommand for PullCommand { + type Container = MevaContainer; + + fn name(&self) -> &'static str { + "pull" + } + + fn about(&self) -> &'static str { + "Fetch and integrate the latest changes from the remote repository into your local branch" + } + + fn version(&self) -> &'static str { + "1.0.0" + } + + /// Builds the CLI argument parser for the `pull` command. + fn build_command(&self) -> Command { + self.build_base_command() + .with_verbose_arg("Enable verbose output when pulling from remote server") + .arg( + Arg::new(Self::ARG_ORIGIN) + .value_name("ORIGIN") + .index(1) + .required(false) + .default_value("origin") + .help("Name of the remote repository to pull from"), + ) + .arg( + Arg::new(Self::ARG_BRANCH) + .value_name("BRANCH") + .index(2) + .required(false) + .help("Name of the branch to pull from"), + ) + } + + /// Executes the `pull` command. + /// + /// # Arguments + /// * `matches`: Parsed command-line arguments. + /// * `container`: Dependency injection container. + /// + /// # Returns + /// * `Result<()>`: Success or error during execution. + async fn execute(&self, _matches: &ArgMatches, _container: &Self::Container) -> Result<()> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use rstest::rstest; + + #[rstest] + fn test_command_name_about_version() { + let cmd = PullCommand; + assert_eq!(cmd.name(), "pull"); + assert_eq!( + cmd.about(), + "Fetch and integrate the latest changes from the remote repository into your local branch" + ); + assert_eq!(cmd.version(), "1.0.0"); + } +} diff --git a/cli/src/commands/push.rs b/cli/src/commands/push.rs new file mode 100644 index 00000000..dd9e54c0 --- /dev/null +++ b/cli/src/commands/push.rs @@ -0,0 +1,120 @@ +use async_trait::async_trait; +use clap::{Arg, ArgAction, ArgMatches, Command}; +use miette::{IntoDiagnostic, Result}; + +use engine::{EngineContainer, engine_container::MevaContainer, handlers::push::Request}; + +use crate::{commands::MevaCommand, extensions::WithVerbose}; + +/// Implements the `push` command for Meva DVCS. +/// +/// This command pushes local commits to a remote repository. +/// It supports pushing specific branches or all local branches +/// Deletion of remote branches is also supported. +#[derive(Default)] +pub struct PushCommand; + +impl PushCommand { + const ARG_ORIGIN: &'static str = "origin"; + + const ARG_BRANCH: &'static str = "branch"; + + const ARG_ALL: &'static str = "all"; + + const ARG_DELETE: &'static str = "delete"; +} + +#[async_trait] +impl MevaCommand for PushCommand { + type Container = MevaContainer; + + fn name(&self) -> &'static str { + "push" + } + + fn about(&self) -> &'static str { + "Push local commits to a remote repository" + } + + fn version(&self) -> &'static str { + "1.0.0" + } + + /// Builds the CLI argument parser for the `push` command. + fn build_command(&self) -> Command { + self.build_base_command() + .with_verbose_arg("Enable verbose output when pushing to remote server") + .arg( + Arg::new(Self::ARG_ORIGIN) + .value_name("ORIGIN") + .index(1) + .required(false) + .default_value("origin") + .help("Name of the remote repository to push to"), + ) + .arg( + Arg::new(Self::ARG_BRANCH) + .value_name("BRANCH") + .index(2) + .required(false) + .help("Name of the branch to push to"), + ) + .arg( + Arg::new(Self::ARG_ALL) + .long(Self::ARG_ALL) + .short('a') + .help("Push all local branches to the specified remote") + .action(ArgAction::SetTrue) + .conflicts_with_all([Self::ARG_BRANCH, Self::ARG_DELETE]), + ) + .arg( + Arg::new(Self::ARG_DELETE) + .long(Self::ARG_DELETE) + .short('d') + .help("Delete the given remote branch.") + .action(ArgAction::SetTrue) + .requires(Self::ARG_BRANCH), + ) + } + + /// Executes the `push` command. + /// + /// # Arguments + /// * `matches`: Parsed command-line arguments. + /// * `container`: Dependency injection container. + /// + /// # Returns + /// * `Result<()>`: Success or error during execution. + async fn execute(&self, matches: &ArgMatches, container: &Self::Container) -> Result<()> { + let request = Request { + origin: matches.get_one::(Self::ARG_ORIGIN).unwrap().clone(), + branch: matches.get_one::(Self::ARG_BRANCH).cloned(), + all: matches.get_flag(Self::ARG_ALL), + delete: matches.get_flag(Self::ARG_DELETE), + verbose: matches.get_flag(Command::ARG_VERBOSE), + }; + + let handler = container.push_handler().into_diagnostic()?; + + let response = handler.handle_push(request).await.into_diagnostic()?; + + println!(); + println!("{response}"); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use rstest::rstest; + + #[rstest] + fn test_command_name_about_version() { + let cmd = PushCommand; + assert_eq!(cmd.name(), "push"); + assert_eq!(cmd.about(), "Push local commits to a remote repository"); + assert_eq!(cmd.version(), "1.0.0"); + } +} diff --git a/cli/src/commands/remote/subcommands/add.rs b/cli/src/commands/remote/subcommands/add.rs index 1a4d5c39..4e43d93b 100644 --- a/cli/src/commands/remote/subcommands/add.rs +++ b/cli/src/commands/remote/subcommands/add.rs @@ -1,7 +1,7 @@ use std::path::PathBuf; use async_trait::async_trait; -use clap::{Arg, ArgAction, ArgMatches, Command, ValueHint}; +use clap::{Arg, ArgMatches, Command, ValueHint}; use engine::{ EngineContainer, engine_container::MevaContainer, @@ -20,9 +20,6 @@ use crate::{commands::MevaCommand, extensions::WithName}; pub struct RemoteAddCommand; impl RemoteAddCommand { - /// Argument name for the fetch flag. - const ARG_FETCH: &'static str = "fetch"; - /// Argument name for the remote URL. const ARG_URL: &'static str = "url"; @@ -67,13 +64,6 @@ impl MevaCommand for RemoteAddCommand { .value_parser(clap::value_parser!(PathBuf)) .help("Path to server's public key"), ) - .arg( - Arg::new(Self::ARG_FETCH) - .short('f') - .long(Self::ARG_FETCH) - .action(ArgAction::SetTrue) - .help("Fetch remote refs immediately after adding"), - ) } /// Executes the `remote add` command. @@ -85,23 +75,25 @@ impl MevaCommand for RemoteAddCommand { matches: &ArgMatches, container: &Self::Container, ) -> miette::Result<()> { + let name = matches + .get_one::(Command::ARG_NAME) + .unwrap() + .to_string(); + let request = AddRequest { - name: matches - .get_one::(Command::ARG_NAME) - .unwrap() - .to_string(), + name: name.clone(), url: matches.get_one::(Self::ARG_URL).unwrap().clone(), pub_key: matches .get_one::(Self::ARG_SERVER_KEY) .unwrap() .clone(), - fetch: matches.get_flag(Self::ARG_FETCH), }; let handler = container.remote_handler().into_diagnostic()?; - let _response = handler.add(request).into_diagnostic()?; + let response = handler.add(request).into_diagnostic()?; + println!("{}", response.remote.display_verbose(&name)); Ok(()) } } diff --git a/cli/src/commands/remote/subcommands/set_url.rs b/cli/src/commands/remote/subcommands/set_url.rs index 361c958a..f2674383 100644 --- a/cli/src/commands/remote/subcommands/set_url.rs +++ b/cli/src/commands/remote/subcommands/set_url.rs @@ -1,5 +1,7 @@ +use std::path::PathBuf; + use async_trait::async_trait; -use clap::{Arg, ArgMatches, Command, builder::PossibleValuesParser}; +use clap::{Arg, ArgMatches, Command, ValueHint, builder::PossibleValuesParser}; use engine::{ EngineContainer, engine_container::MevaContainer, @@ -26,6 +28,9 @@ impl RemoteSetUrlCommand { /// Argument name for the new URL. const ARG_NEW_URL: &'static str = "new-url"; + + /// Argument name for specifying the server public key. + const ARG_NEW_SERVER_KEY: &'static str = "new-server-key"; } #[async_trait] @@ -56,6 +61,15 @@ impl MevaCommand for RemoteSetUrlCommand { .value_parser(clap::value_parser!(Url)) .help("New repository URL (e.g. ssh://user@host:port/repository)"), ) + .arg( + Arg::new(Self::ARG_NEW_SERVER_KEY) + .value_name("NEW_SERVER_KEY") + .index(3) + .required(true) + .value_hint(ValueHint::FilePath) + .value_parser(clap::value_parser!(PathBuf)) + .help("Path to new server's public key"), + ) .arg( Arg::new(Self::ARG_DIRECTION) .long(Self::ARG_DIRECTION) @@ -83,13 +97,17 @@ impl MevaCommand for RemoteSetUrlCommand { .unwrap() .clone(), new_url: matches.get_one::(Self::ARG_NEW_URL).unwrap().clone(), + new_server_key: matches + .get_one::(Self::ARG_NEW_SERVER_KEY) + .unwrap() + .clone(), direction: direction_str.parse::().unwrap(), }; let handler = container.remote_handler().into_diagnostic()?; - let _response = handler.set_url(request).into_diagnostic()?; + handler.set_url(request).into_diagnostic()?; - println!("Remote URL changed successfully!"); + println!("Remote URL changed successfully."); Ok(()) } diff --git a/cli/src/commands/remote/subcommands/show.rs b/cli/src/commands/remote/subcommands/show.rs index cc7e8cc5..cc6536dc 100644 --- a/cli/src/commands/remote/subcommands/show.rs +++ b/cli/src/commands/remote/subcommands/show.rs @@ -8,7 +8,10 @@ use engine::{ use miette::IntoDiagnostic; use owo_colors::OwoColorize; -use crate::{commands::MevaCommand, extensions::WithName}; +use crate::{ + commands::MevaCommand, + extensions::{WithName, WithVerbose}, +}; /// Implements the `remote show` subcommand for Meva DVCS. /// @@ -43,6 +46,7 @@ impl MevaCommand for RemoteShowCommand { fn build_command(&self) -> Command { self.build_base_command() .with_name_arg("Name for the remote") + .with_verbose_arg("Enable verbose output") .arg( Arg::new(Self::ARG_NO_FETCH) .short('n') @@ -67,6 +71,7 @@ impl MevaCommand for RemoteShowCommand { let request = ShowRequest { name: name.clone(), no_fetch: matches.get_flag(Self::ARG_NO_FETCH), + verbose: matches.get_flag(Command::ARG_VERBOSE), }; let handler = container.remote_handler().into_diagnostic()?; diff --git a/cli/src/main.rs b/cli/src/main.rs index 528e3bb6..2d2c76ca 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -4,41 +4,19 @@ mod meva_cli; use miette::Result; -use crate::{commands::RemoteCommand, meva_cli::MevaCli}; -use commands::{ - AddCommand, BranchCommand, CloneCommand, CommitCommand, ConfigCommand, DiffCommand, - FetchCommand, IgnoreCommand, InitCommand, LogCommand, LsFilesCommand, LsTreeCommand, - MevaCommand, PluginsCommand, RestoreCommand, ShowCommand, StatusCommand, -}; +use crate::meva_cli::MevaCli; +use commands::collect_commands; use engine::engine_container::MevaContainer; #[tokio::main] async fn main() -> Result<()> { miette::set_panic_hook(); - let commands: Vec>> = vec![ - Box::new(InitCommand), - Box::new(ConfigCommand), - Box::new(IgnoreCommand), - Box::new(PluginsCommand), - Box::new(AddCommand), - Box::new(LsFilesCommand), - Box::new(ShowCommand), - Box::new(StatusCommand), - Box::new(CommitCommand), - Box::new(DiffCommand), - Box::new(LsTreeCommand), - Box::new(LogCommand), - Box::new(RestoreCommand), - Box::new(CloneCommand), - Box::new(RemoteCommand), - Box::new(FetchCommand), - Box::new(BranchCommand), - ]; - let container = MevaContainer; let mut cli = MevaCli::new(container); + let commands = collect_commands(); + for command in commands { cli.add_command(command); } diff --git a/engine/src/engine_container.rs b/engine/src/engine_container.rs index 9ae31ea9..d99b7544 100644 --- a/engine/src/engine_container.rs +++ b/engine/src/engine_container.rs @@ -10,7 +10,8 @@ use crate::handlers::{ add::AddHandler, branch::BranchHandler, clone::CloneHandler, commit::CommitHandler, config::ConfigHandler, diff::DiffHandler, fetch::FetchHandler, init::InitHandler, log::LogHandler, ls_files::LsFilesHandler, ls_tree::LsTreeHandler, plugins::PluginsHandler, - remote::RemoteHandler, restore::RestoreHandler, show::ShowHandler, status::StatusHandler, + push::PushHandler, remote::RemoteHandler, restore::RestoreHandler, show::ShowHandler, + status::StatusHandler, }; use crate::index::{MevaIndex, MevaWorkingDir}; use crate::network::MevaRemotesManager; @@ -75,6 +76,9 @@ pub trait EngineContainer { /// Returns the handler responsible for branch management. fn branch_handler(&self) -> EngineResult; + + /// Returns the handler responsible for push operation. + fn push_handler(&self) -> EngineResult; } /// Concrete implementation of `EngineContainer` for Meva. @@ -123,11 +127,13 @@ impl EngineContainer for MevaContainer { let config_loader = Arc::new(MevaConfigLoader::default()); let object_storage = Arc::new(MevaObjectStorage::new(layout.clone())); let ref_manager = Arc::new(MevaRefManager::new(layout.clone())); + let remotes_manager = Arc::new(MevaRemotesManager); let repository = Arc::new(MevaRepository::new( layout, config_loader, object_storage, ref_manager, + remotes_manager, )); let handler = InitHandler { repository }; Ok(handler) @@ -371,6 +377,7 @@ impl EngineContainer for MevaContainer { fn clone_handler(&self) -> EngineResult { let repository_layout = self.repository_layout_env()?; let config_loader = Arc::new(MevaConfigLoader::default()); + let handler = CloneHandler::new(repository_layout, config_loader); Ok(handler) } @@ -419,4 +426,14 @@ impl EngineContainer for MevaContainer { Ok(handler) } + + fn push_handler(&self) -> EngineResult { + let layout = self.repository_layout_discover()?; + let config_loader = Arc::new(MevaConfigLoader::default()); + let ref_manager = Arc::new(MevaRefManager::new(layout)); + let remotes_manager = Arc::new(MevaRemotesManager); + + let handler = PushHandler::new(config_loader, remotes_manager, ref_manager); + Ok(handler) + } } diff --git a/engine/src/errors/repository_error.rs b/engine/src/errors/repository_error.rs index faee1d0d..1f991886 100644 --- a/engine/src/errors/repository_error.rs +++ b/engine/src/errors/repository_error.rs @@ -13,4 +13,8 @@ pub enum RepositoryError { /// Raised when the user's configuration directory cannot be determined. #[error("User's config directory not found")] ConfigDirNotFound, + + /// Raised when the repository is in a detached HEAD state. + #[error("Repository is in a detached HEAD state")] + DetachedHead, } diff --git a/engine/src/handlers.rs b/engine/src/handlers.rs index ff34bcbe..948cb01f 100644 --- a/engine/src/handlers.rs +++ b/engine/src/handlers.rs @@ -10,6 +10,7 @@ pub mod log; pub mod ls_files; pub mod ls_tree; pub mod plugins; +pub mod push; pub mod remote; pub mod restore; pub mod show; diff --git a/engine/src/handlers/clone/handlers.rs b/engine/src/handlers/clone/handlers.rs index c1bd6223..bb933bcc 100644 --- a/engine/src/handlers/clone/handlers.rs +++ b/engine/src/handlers/clone/handlers.rs @@ -10,7 +10,7 @@ use tempfile::TempDir; use crate::{ ConfigLoader, MevaConfigLoader, MevaRepository, RepositoryLayout, errors::{CloneError, EngineError, EngineResult, NetworkError}, - network::{PackfileCodec, SshConnectionParams, SshService, SshSession}, + network::{MevaRemotesManager, PackfileCodec, SshConnectionParams, SshService, SshSession}, object_storage::MevaObjectStorage, objects::MevaObject, ref_manager::{MevaRefManager, RefEntry}, @@ -92,7 +92,10 @@ impl CloneHandler { connection_params.client_key_path = client_key; connection_params.server_key_path = request.server_key.clone(); - let ssh_session = self.ssh_service.connect(&connection_params).await?; + let ssh_session = self + .ssh_service + .connect(&connection_params, !request.quiet) + .await?; Ok((ssh_session, connection_params, path)) } @@ -106,19 +109,21 @@ impl CloneHandler { temp_path: &Path, objects: &[(MevaObject, Vec)], refs: &[RefEntry], - remote_name: &str, + request: &Request, ) -> EngineResult> { let repository_layout = Arc::new(MevaRepositoryLayout::new(temp_path.to_path_buf())?); let object_storage = Arc::new(MevaObjectStorage::new(repository_layout.clone())); let ref_manager = Arc::new(MevaRefManager::new(repository_layout.clone())); + let remotes_manager = Arc::new(MevaRemotesManager); let repository = MevaRepository::new( repository_layout.clone(), Arc::new(MevaConfigLoader::default()), object_storage, ref_manager, + remotes_manager, ); - repository.clone(objects, refs, remote_name) + repository.clone(objects, refs, request) } } @@ -129,7 +134,11 @@ impl CloneOperations for CloneHandler { let (mut ssh_session, connection_params, path) = self.connect_ssh(&request).await?; let packfile_result = ssh_session - .run_upload_pack(&connection_params.repository_name, Vec::new()) + .run_upload_pack( + &connection_params.repository_name, + Vec::new(), + !request.quiet, + ) .await?; let refs = packfile_result.refs; @@ -145,9 +154,7 @@ impl CloneOperations for CloneHandler { let temp_dir = TempDir::new_in(parent_dir).map_err(EngineError::Io)?; let temp_path = temp_dir.path().to_path_buf(); - let clone_result = self - .temp_clone(&temp_path, &objects, &refs, &request.origin) - .await; + let clone_result = self.temp_clone(&temp_path, &objects, &refs, &request).await; match clone_result { Ok(ref_entries) => { diff --git a/engine/src/handlers/fetch/handlers.rs b/engine/src/handlers/fetch/handlers.rs index 81306891..43158f3d 100644 --- a/engine/src/handlers/fetch/handlers.rs +++ b/engine/src/handlers/fetch/handlers.rs @@ -11,7 +11,8 @@ use crate::{ ConfigLoader, errors::EngineResult, network::{ - PackfileCodec, RemoteEntry, RemotesManager, SshConnectionParams, SshService, SshSession, + PackfileCodec, RemoteDirection, RemoteEntry, RemotesManager, SshConnectionParams, + SshService, SshSession, }, object_storage::ObjectStorage, ref_manager::{RefEntry, RefManager}, @@ -110,32 +111,6 @@ impl FetchHandler { Ok(()) } - /// Updates local remote-tracking references to match the state of the remote repository. - /// - /// This method iterates over the provided list of references (which are known to - /// be new or updated) and writes them to the local reference store. - /// - /// The reference names are automatically mapped from the server's namespace - /// (e.g., `refs/heads/main`) to the local remote-tracking namespace - /// (e.g., `refs/remotes/origin/main`). - /// - /// # Arguments - /// * `remote_name` - The name of the remote (e.g., "origin"). - /// * `remote_refs` - The list of server references to update locally. - fn update_remote_refs(&self, remote_name: &str, remote_refs: &[RefEntry]) -> EngineResult<()> { - for remote_ref in remote_refs { - let tracking_name = self - .ref_manager - .map_head_to_remote_ref(&remote_ref.name, remote_name); - - println!("Updating {} -> {}", tracking_name, remote_ref.commit_hash); - - let new_entry = RefEntry::new(&tracking_name, &remote_ref.commit_hash); - self.ref_manager.update_ref(&new_entry)?; - } - Ok(()) - } - /// Decodes the received packfile data and writes the objects to storage. /// /// This method takes the raw binary packfile data received from the server, @@ -144,12 +119,17 @@ impl FetchHandler { /// /// # Arguments /// * `packfile_data` - The raw bytes of the packfile. - fn process_packfile(&self, packfile_data: &[u8]) -> EngineResult<()> { - println!("Decoding objects..."); + /// * `verbose` - Whether to enable verbose output during processing. + fn process_packfile(&self, packfile_data: &[u8], verbose: bool) -> EngineResult<()> { + if verbose { + println!("Decoding objects..."); + } let objects = PackfileCodec::default().decode_packfile(packfile_data)?; - println!("Successfully decoded {} objects.", objects.len()); + if verbose { + println!("Successfully decoded {} objects.", objects.len()); + } self.object_storage.add_objects_from_packfile(&objects) } @@ -166,14 +146,18 @@ impl FetchHandler { request: &Request, ) -> EngineResult<(SshSession, SshConnectionParams)> { let remote_entry = self.get_remote_entry(&request.origin)?; - let mut connection_params = SshConnectionParams::try_from(&remote_entry.url)?; + let mut connection_params = + SshConnectionParams::try_from(remote_entry.url_for(RemoteDirection::Fetch))?; let client_key = self.get_user_signing_key()?; connection_params.client_key_path = client_key; - connection_params.server_key_path = remote_entry.pub_signing_key; + connection_params.server_key_path = + remote_entry.key_for(RemoteDirection::Fetch).to_path_buf(); Ok(( - self.ssh_service.connect(&connection_params).await?, + self.ssh_service + .connect(&connection_params, request.verbose) + .await?, connection_params, )) } @@ -243,12 +227,12 @@ impl FetchOperations for FetchHandler { .unique() .collect::>(); - let packfile_result = ssh_session - .run_upload_pack(&connection_params.repository_name, haves) + let upload_pack_result = ssh_session + .run_upload_pack(&connection_params.repository_name, haves, request.verbose) .await?; let heads_refs_prefix = self.ref_manager.heads_refs_prefix(); - let server_remote_refs = packfile_result + let server_remote_refs = upload_pack_result .refs .into_iter() .filter(|r| r.name.starts_with(&heads_refs_prefix)) @@ -262,12 +246,17 @@ impl FetchOperations for FetchHandler { self.find_refs_to_update(&request, &server_remote_refs, &local_remote_refs)?; if refs_to_update.is_empty() { - println!("Already up to date."); + println!(); + println!("Already up-to-date."); } else { - self.process_packfile(&packfile_result.packfile_data)?; - self.update_remote_refs(&request.origin, &refs_to_update)?; + self.process_packfile(&upload_pack_result.packfile_data, request.verbose)?; + self.ref_manager.update_remote_refs( + &request.origin, + &refs_to_update, + request.verbose, + )?; } - Ok(Response {}) + Ok(Response) } } diff --git a/engine/src/handlers/fetch/operations.rs b/engine/src/handlers/fetch/operations.rs index 2207dd7f..6a48cf6b 100644 --- a/engine/src/handlers/fetch/operations.rs +++ b/engine/src/handlers/fetch/operations.rs @@ -2,7 +2,7 @@ use async_trait::async_trait; use crate::errors::EngineResult; -#[derive(Debug)] +#[derive(Debug, Default)] pub struct Request { pub origin: String, pub branch: Option, @@ -10,11 +10,19 @@ pub struct Request { pub verbose: bool, } -#[derive(Debug)] -pub struct Response { - // TODO +impl Request { + pub fn new(origin: String, branch: Option) -> Self { + Self { + origin, + branch, + ..Default::default() + } + } } +#[derive(Debug)] +pub struct Response; + #[async_trait] pub trait FetchOperations { async fn fetch(&self, request: Request) -> EngineResult; diff --git a/engine/src/handlers/ls_files/models/ls_files_entry.rs b/engine/src/handlers/ls_files/models/ls_files_entry.rs index 9a9b306b..0da6550f 100644 --- a/engine/src/handlers/ls_files/models/ls_files_entry.rs +++ b/engine/src/handlers/ls_files/models/ls_files_entry.rs @@ -1,5 +1,7 @@ use std::fmt::Display; +use owo_colors::OwoColorize; + use crate::index::{file_mode::FileMode, stage::Stage}; /// Represents a single entry in an `ls-files` response. @@ -23,13 +25,24 @@ impl Display for LsFilesEntry { /// Formats an [`LsFilesEntry`] for human-readable output. fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - LsFilesEntry::Path { path } => write!(f, "{path}"), + LsFilesEntry::Path { path } => { + write!(f, "{}", path.green()) + } LsFilesEntry::Entry { mode, object, stage, path, - } => write!(f, "{mode:o} {object} {stage} {path}"), + } => { + write!( + f, + "{} {} {} {}", + format!("{mode:o}").dimmed(), + object.yellow(), + stage.dimmed(), + path.green() + ) + } } } } diff --git a/engine/src/handlers/push.rs b/engine/src/handlers/push.rs new file mode 100644 index 00000000..f79684b9 --- /dev/null +++ b/engine/src/handlers/push.rs @@ -0,0 +1,5 @@ +mod handlers; +mod operations; + +pub use handlers::PushHandler; +pub use operations::*; diff --git a/engine/src/handlers/push/handlers.rs b/engine/src/handlers/push/handlers.rs new file mode 100644 index 00000000..4424541e --- /dev/null +++ b/engine/src/handlers/push/handlers.rs @@ -0,0 +1,189 @@ +use std::{path::PathBuf, sync::Arc}; + +use async_trait::async_trait; + +use crate::{ + ConfigLoader, + errors::{BranchError, EngineResult, RepositoryError}, + network::{ + RemoteDirection, RemoteEntry, RemotesManager, SshConnectionParams, SshService, SshSession, + ZERO_HASH, + }, + ref_manager::{RefEntry, RefManager}, +}; + +use super::{PushOperations, Request, Response}; + +/// Handles the logic for pushing local changes to a remote repository. +/// +/// The `PushHandler` orchestrates the interaction between the local reference manager, +/// the configuration system, and the SSH network service to execute the `receive-pack` +/// protocol on the remote server. +pub struct PushHandler { + ssh_service: SshService, + config_loader: Arc, + remotes_manager: Arc, + ref_manager: Arc, +} + +impl PushHandler { + /// Creates a new instance of the [`PushHandler`]. + pub fn new( + config_loader: Arc, + remotes_manager: Arc, + ref_manager: Arc, + ) -> Self { + Self { + ssh_service: SshService, + config_loader, + remotes_manager, + ref_manager, + } + } + + /// Entry point to execute a push operation based on the provided request. + /// + /// This method delegates the actual logic to the [`PushOperations::push`] implementation. + pub async fn handle_push(&self, request: Request) -> EngineResult { + self.push(request).await + } + + /// Retrieves configuration for a specific remote by name. + /// + /// # Arguments + /// + /// * `name` - The name of the remote (e.g., "origin"). + fn get_remote_entry(&self, name: &str) -> EngineResult { + self.remotes_manager.get_remote(name) + } + + /// Retrieves the path to the user's SSH signing key from the configuration. + /// + /// Looks for the `user.signing_key` setting. + fn get_user_signing_key(&self) -> EngineResult { + Ok(PathBuf::from( + self.config_loader.get("user.signing_key", None)?, + )) + } + + /// Establishes an SSH connection to the remote server. + /// + /// This method resolves the remote URL, the client's private key, and the + /// server's public key (host key) before initiating the connection. + /// + /// # Returns + /// + /// A tuple containing the active [`SshSession`] and the resolved [`SshConnectionParams`]. + async fn connect_ssh( + &self, + request: &Request, + ) -> EngineResult<(SshSession, SshConnectionParams)> { + let remote_entry = self.get_remote_entry(&request.origin)?; + let mut connection_params = + SshConnectionParams::try_from(remote_entry.url_for(RemoteDirection::Push))?; + + let client_key = self.get_user_signing_key()?; + connection_params.client_key_path = client_key; + connection_params.server_key_path = + remote_entry.key_for(RemoteDirection::Push).to_path_buf(); + + Ok(( + self.ssh_service + .connect(&connection_params, request.verbose) + .await?, + connection_params, + )) + } + + /// Determines which references (branches) need to be updated on the remote. + /// + /// This method constructs a list of [`RefEntry`] objects based on the request flags: + /// * `delete` - Creates an update with `ZERO_HASH` to delete the branch on the remote. + /// * `all` - Collects all local heads (branches) to be pushed. + /// * Default - Resolves the specific branch (or HEAD) requested. + fn prepare_updates(&self, request: &Request) -> EngineResult> { + let mut entries = Vec::new(); + let zero_hash = ZERO_HASH.to_string(); + let heads_refs_prefix = self.ref_manager.heads_refs_prefix(); + + if request.delete { + if let Some(branch_name) = &request.branch { + entries.push(RefEntry { + name: format!("{heads_refs_prefix}{branch_name}"), + commit_hash: zero_hash, + }); + } + return Ok(entries); + } + + if request.all { + let local_branches = self.ref_manager.collect_refs_heads()?; + for branch in local_branches { + entries.push(RefEntry { + name: branch.name, + commit_hash: branch.commit_hash, + }); + } + return Ok(entries); + } + + let ref_entry = self.resolve_ref_entry(request)?; + entries.push(ref_entry); + + Ok(entries) + } + + /// Resolves the specific local reference to be pushed. + /// + /// If a branch name is provided in the request, it resolves that specific branch. + /// If no branch name is provided, it attempts to resolve the current `HEAD`. + /// + /// # Errors + /// + /// Returns [`RepositoryError::DetachedHead`] if trying to push HEAD while in detached state. + /// Returns [`BranchError::BranchNotFound`] if the specified branch does not exist. + fn resolve_ref_entry(&self, request: &Request) -> EngineResult { + let branch_name = match &request.branch { + Some(name) => format!("{}{}", self.ref_manager.heads_refs_prefix(), name.clone()), + None => { + let head = self.ref_manager.read_head()?; + if head.is_symbolic() { + head.target.clone() + } else { + return Err(RepositoryError::DetachedHead.into()); + } + } + }; + + let ref_entry = self + .ref_manager + .read_ref(&branch_name)? + .ok_or(BranchError::BranchNotFound(branch_name))?; + + Ok(ref_entry) + } +} + +#[async_trait] +impl PushOperations for PushHandler { + /// Orchestrates the entire push process. + async fn push(&self, request: Request) -> EngineResult { + let (mut ssh_session, connection_params) = self.connect_ssh(&request).await?; + + let updates = self.prepare_updates(&request)?; + + let receive_pack_result = ssh_session + .run_receive_pack( + &connection_params.repository_name, + &updates, + request.verbose, + ) + .await?; + + Ok(Response::new( + request.origin, + receive_pack_result.unpack_successful, + receive_pack_result.receive_results, + )) + } +} diff --git a/engine/src/handlers/push/operations.rs b/engine/src/handlers/push/operations.rs new file mode 100644 index 00000000..aabf53d6 --- /dev/null +++ b/engine/src/handlers/push/operations.rs @@ -0,0 +1,88 @@ +use std::fmt::Display; + +use async_trait::async_trait; +use owo_colors::OwoColorize; + +use crate::{ + errors::EngineResult, + network::{ReceiveResult, ReceiveStatus}, +}; + +#[derive(Debug, Default)] +pub struct Request { + pub origin: String, + pub branch: Option, + pub all: bool, + pub delete: bool, + pub verbose: bool, +} + +impl Request { + pub fn new(origin: String, branch: Option) -> Self { + Self { + origin, + branch, + ..Default::default() + } + } +} + +#[derive(Debug)] +pub struct Response { + pub unpack_successful: bool, + pub receive_results: Vec, + pub origin: String, +} + +impl Response { + pub fn new( + origin: String, + unpack_successful: bool, + receive_results: Vec, + ) -> Self { + Self { + origin, + unpack_successful, + receive_results, + } + } + + /// Checks if there are any failures in the receive results. + pub fn has_failures(&self) -> bool { + self.receive_results + .iter() + .any(|receive_result| matches!(receive_result.status, ReceiveStatus::Failure(_))) + } +} + +impl Display for Response { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if self.unpack_successful { + writeln!( + f, + "Unpacking objects: {} {}", + "100%".green(), + "(done)".dimmed() + )?; + } else { + writeln!(f, "Unpacking objects: {}", "failed".red().bold())?; + return Ok(()); + } + + writeln!(f, "To {}", self.origin.bold())?; + for result in &self.receive_results { + writeln!(f, "{result}")?; + } + + if self.has_failures() { + writeln!(f, "{}", "Error: failed to push some refs".red().bold())?; + } + + Ok(()) + } +} + +#[async_trait] +pub trait PushOperations { + async fn push(&self, request: Request) -> EngineResult; +} diff --git a/engine/src/handlers/remote/handlers.rs b/engine/src/handlers/remote/handlers.rs index a0c0af27..606e09da 100644 --- a/engine/src/handlers/remote/handlers.rs +++ b/engine/src/handlers/remote/handlers.rs @@ -221,8 +221,9 @@ impl RemoteOperations for RemoteHandler { Ok(AddResponse { remote: self.remotes_manager.add_remote( &request.name, - request.url, + &request.url, &request.pub_key, + None, )?, }) } @@ -266,6 +267,7 @@ impl RemoteOperations for RemoteHandler { self.remotes_manager.set_remote_url( &request.name, request.new_url.as_ref(), + request.new_server_key.as_ref(), &request.direction, ) } @@ -291,13 +293,15 @@ impl RemoteOperations for RemoteHandler { connection_params.client_key_path = client_key; connection_params.server_key_path = remote.pub_signing_key.clone(); - let mut ssh_session = self.ssh_service.connect(&connection_params).await?; + let mut ssh_session = self + .ssh_service + .connect(&connection_params, request.verbose) + .await?; let packfile_result = ssh_session - .run_ls_remotes(&connection_params.repository_name) + .run_ls_remotes(&connection_params.repository_name, request.verbose) .await?; - // dbg!(server_refs, local_remotes, local_heads); // TODO: coś takiego musi być w lokalnym configu (ale to chyba merge będzie dodawał??) // [branch.develop] // remote = origin diff --git a/engine/src/handlers/remote/operations.rs b/engine/src/handlers/remote/operations.rs index 7603fab5..6c63a463 100644 --- a/engine/src/handlers/remote/operations.rs +++ b/engine/src/handlers/remote/operations.rs @@ -16,7 +16,6 @@ pub struct AddRequest { pub name: String, pub url: Url, pub pub_key: PathBuf, - pub fetch: bool, // so far unused } #[derive(Debug)] @@ -61,6 +60,7 @@ pub struct RenameResponse { pub struct SetUrlRequest { pub name: String, pub new_url: Url, + pub new_server_key: PathBuf, pub direction: RemoteDirection, } @@ -73,6 +73,7 @@ pub struct ListResponse { pub struct ShowRequest { pub name: String, pub no_fetch: bool, + pub verbose: bool, } #[derive(Debug)] diff --git a/engine/src/network.rs b/engine/src/network.rs index 85a2d42d..52f24268 100644 --- a/engine/src/network.rs +++ b/engine/src/network.rs @@ -10,10 +10,11 @@ mod ssh_session; use client_handler::ClientHandler; pub use common::{ - ChannelBand, PktLine, RemoteDirection, RemoteEntry, SessionExtension, create_channel_band, - create_pkt_line, + CHUNK_SIZE, ChannelBand, PKT_CHUNK_SIZE, PktLine, RemoteDirection, RemoteEntry, + SessionExtension, ZERO_HASH, create_channel_band, create_pkt_line, parse_next_pkt_line, }; pub use packfiles::{PackfileCodec, PackfileError}; +pub use protocols::{ReceiveResult, ReceiveStatus}; pub use remotes::{MevaRemotesManager, RemotesManager}; pub use ssh_connection_params::SshConnectionParams; pub use ssh_service::SshService; diff --git a/engine/src/network/client_handler.rs b/engine/src/network/client_handler.rs index 0eb6790d..4d882106 100644 --- a/engine/src/network/client_handler.rs +++ b/engine/src/network/client_handler.rs @@ -41,13 +41,10 @@ impl client::Handler for ClientHandler { &mut self, server_public_key: &key::PublicKey, ) -> Result { - println!("Server presented public key: {}", server_public_key.name()); - if server_public_key == &self.server_public_key { - println!("Server key is valid (matches the expected one)."); Ok(true) } else { - eprintln!("Server key is INVALID!"); + eprintln!("Server key is invalid!"); Ok(false) } } diff --git a/engine/src/network/common.rs b/engine/src/network/common.rs index 0d03341c..d760a6a2 100644 --- a/engine/src/network/common.rs +++ b/engine/src/network/common.rs @@ -4,6 +4,15 @@ mod remote_entry; mod session_extension; pub use channel_band::{ChannelBand, create_channel_band}; -pub use pkt_line::{PktLine, create_pkt_line}; +pub use pkt_line::{PktLine, create_pkt_line, parse_next_pkt_line}; pub use remote_entry::{RemoteDirection, RemoteEntry}; pub use session_extension::SessionExtension; + +// Stream data in chunks to avoid choking the connection +pub const CHUNK_SIZE: usize = 65_000; + +/// Maximum size of a single pkt-line chunk +pub const PKT_CHUNK_SIZE: usize = 65516; + +/// A constant representing a zero hash (40 zeros). +pub const ZERO_HASH: &str = "0000000000000000000000000000000000000000"; diff --git a/engine/src/network/common/pkt_line.rs b/engine/src/network/common/pkt_line.rs index a8ce7824..f9efe764 100644 --- a/engine/src/network/common/pkt_line.rs +++ b/engine/src/network/common/pkt_line.rs @@ -1,3 +1,7 @@ +use std::str::from_utf8; + +use crate::errors::{EngineResult, NetworkError}; + /// Represents a parsed packet-line (pkt-line) from the network stream. /// /// The pkt-line format is a length-prefixed framing protocol, @@ -36,3 +40,33 @@ pub fn create_pkt_line(data: &str) -> String { let len = data.len() + 4; format!("{len:04x}{data}") } + +/// Attempts to parse the next `pkt-line` from the buffer. +/// +/// # Returns +/// * [`PktLine::Flush`] if length is "0000". +/// * [`PktLine::Incomplete`] if buffer is shorter than specified length. +/// * [`PktLine::Payload(Vec)`] containing the payload bytes otherwise. +pub fn parse_next_pkt_line(buffer: &mut Vec) -> EngineResult { + if buffer.len() < 4 { + return Ok(PktLine::Incomplete); + } + + let len_str = from_utf8(&buffer[0..4]).map_err(NetworkError::from)?; + + let len = usize::from_str_radix(len_str, 16).map_err(NetworkError::from)?; + + if len == 0 { + buffer.drain(0..4); + return Ok(PktLine::Flush); + } + + if buffer.len() < len { + return Ok(PktLine::Incomplete); + } + + let line_data = buffer[4..len].to_vec(); + buffer.drain(0..len); + + Ok(PktLine::Payload(line_data)) +} diff --git a/engine/src/network/common/remote_entry.rs b/engine/src/network/common/remote_entry.rs index 2a42364b..f50ad385 100644 --- a/engine/src/network/common/remote_entry.rs +++ b/engine/src/network/common/remote_entry.rs @@ -66,7 +66,7 @@ impl RemoteEntry { self } - /// Constructs a `RemoteEntry` from a key-value map (typically from a config file). + /// Constructs a [`RemoteEntry`] from a key-value map (typically from a config file). /// /// # Required Keys /// * `url`: The fetch URL. diff --git a/engine/src/network/protocols.rs b/engine/src/network/protocols.rs index 4f7d0253..cbcc14e1 100644 --- a/engine/src/network/protocols.rs +++ b/engine/src/network/protocols.rs @@ -1,3 +1,5 @@ +mod receive_pack; mod upload_pack; +pub use receive_pack::{ReceivePackProtocol, ReceivePackResult, ReceiveResult, ReceiveStatus}; pub use upload_pack::{UploadPackProtocol, UploadPackResult}; diff --git a/engine/src/network/protocols/receive_pack.rs b/engine/src/network/protocols/receive_pack.rs new file mode 100644 index 00000000..f21e5549 --- /dev/null +++ b/engine/src/network/protocols/receive_pack.rs @@ -0,0 +1,271 @@ +mod receive_pack_result; +mod receive_pack_state; + +use std::str::from_utf8; + +pub use receive_pack_result::{ReceivePackResult, ReceiveResult, ReceiveStatus}; +pub use receive_pack_state::ReceivePackState; + +use crate::{ + errors::{EngineResult, NetworkError}, + network::{PktLine, ZERO_HASH, create_pkt_line, parse_next_pkt_line}, + ref_manager::RefEntry, +}; +use russh::{Channel, client::Msg}; + +/// Implements the client-side logic for the `receive-pack` protocol (push). +#[derive(Debug, Default)] +pub struct ReceivePackProtocol { + /// The result of the receive-pack operation after completion. + pub result: ReceivePackResult, + + /// References discovered from the remote server. + pub server_refs: Vec, + + /// The hash of the HEAD reference on the remote (if exists). + pub remote_head: Option, + + /// Current state of the protocol. + state: ReceivePackState, + + /// Internal buffer for parsing pkt-lines. + buffer: Vec, + + /// The updates the local client *wants* to perform. + proposed_updates: Vec, + + /// Flag to track if we have handled the initial flush after the service header. + header_flush_received: bool, +} + +impl ReceivePackProtocol { + /// Creates a new instance of the protocol with the intended updates. + /// + /// # Arguments + /// + /// * `updates` - A list of reference updates (e.g., branch moves) to be pushed. + pub fn new(updates: Vec) -> Self { + Self { + proposed_updates: updates, + ..Default::default() + } + } + + /// Checks if the protocol has finished successfully. + pub fn is_complete(&self) -> bool { + self.state == ReceivePackState::Complete + } + + /// Signals the protocol that the binary packfile has been fully streamed. + /// This transitions the state from sending data to waiting for the server's report. + pub fn mark_packfile_sent(&mut self) { + if self.state == ReceivePackState::Packfile { + self.state = ReceivePackState::ReceivingReport; + } + } + + /// Processes incoming data from the server. + /// + /// # Returns + /// * `Ok(true)`: Indicates that the client should now start streaming the packfile. + /// * `Ok(false)`: Normal processing, no immediate action required from the caller. + pub async fn process_data( + &mut self, + data: &[u8], + channel: &mut Channel, + verbose: bool, + ) -> EngineResult { + self.buffer.extend_from_slice(data); + + let mut ready_to_send_pack = false; + + loop { + if self.state == ReceivePackState::Complete { + break; + } + + let packet = parse_next_pkt_line(&mut self.buffer)?; + + match packet { + PktLine::Incomplete => { + break; + } + PktLine::Payload(payload) => { + self.handle_line(payload, verbose).await?; + } + PktLine::Flush => { + if self.handle_flush(channel, verbose).await? { + ready_to_send_pack = true; + } + } + } + } + Ok(ready_to_send_pack) + } + + /// Handles a single payload line received from the server. + /// + /// The logic depends on the current state (Discovery or ReceivingReport). + async fn handle_line(&mut self, payload: Vec, verbose: bool) -> EngineResult<()> { + let line_str = from_utf8(&payload).map_err(NetworkError::from)?.trim(); + + match self.state { + ReceivePackState::Discovery => { + if line_str.starts_with('#') { + return Ok(()); + } + + let clean_line = line_str.split('\0').next().unwrap_or(line_str); + let parts: Vec<&str> = clean_line.split_whitespace().collect(); + if parts.len() >= 2 { + let sha = parts[0]; + let name = parts[1]; + + if name == "HEAD" { + self.remote_head = Some(sha.to_string()); + } else { + self.server_refs.push(RefEntry::new(name, sha)); + } + } + } + ReceivePackState::ReceivingReport => { + if line_str.starts_with("unpack ok") { + self.result.unpack_successful = true; + } else if line_str.starts_with("ok") { + // Format: "ok refs/heads/master" + let ref_name = line_str.split_whitespace().nth(1).unwrap_or("?"); + self.result + .receive_results + .push(self.create_result(ref_name, ReceiveStatus::Success)); + } else if line_str.starts_with("ng") { + // Format: "ng refs/heads/master non-fast-forward" + let parts: Vec<&str> = line_str.split_whitespace().collect(); + let ref_name = parts.get(1).unwrap_or(&"?"); + let reason = parts[2..].join(" "); + + self.result + .receive_results + .push(self.create_result(ref_name, ReceiveStatus::Failure(reason))); + } + } + _ => { + if verbose { + println!("Received packet in state {}: {line_str}", self.state); + } + } + } + Ok(()) + } + + /// Handles a flush packet (`0000`) received from the server. + /// + /// Returns `true` if the state transitions to `Packfile`, indicating + /// that the client should start streaming binary data. + async fn handle_flush( + &mut self, + channel: &mut Channel, + verbose: bool, + ) -> EngineResult { + match self.state { + ReceivePackState::Discovery => { + if !self.header_flush_received { + self.header_flush_received = true; + return Ok(false); + } + + self.state = ReceivePackState::SendingCommands; + let updates_count = self.send_update_commands(channel, verbose).await?; + + if updates_count > 0 { + self.state = ReceivePackState::Packfile; + Ok(true) + } else { + self.state = ReceivePackState::ReceivingReport; + Ok(false) + } + } + ReceivePackState::ReceivingReport => { + self.state = ReceivePackState::Complete; + Ok(false) + } + _ => { + if verbose { + println!("Received flush in state {}, no action taken.", self.state); + } + Ok(false) + } + } + } + + /// Sends update commands (e.g., `old_sha new_sha ref_name`) to the server. + /// + /// This happens after the Discovery phase. If no updates are needed (everything up-to-date), + /// it sends a flush packet immediately. + /// + /// # Returns + /// The number of update commands sent. + async fn send_update_commands( + &mut self, + channel: &mut Channel, + verbose: bool, + ) -> EngineResult { + let zero_id = ZERO_HASH.to_string(); + let mut updates_sent = 0; + + for update in &self.proposed_updates { + let old_sha = self + .server_refs + .iter() + .find(|r| r.name == update.name) + .map(|r| r.commit_hash.as_str()) + .unwrap_or(&zero_id); + + if old_sha == update.commit_hash { + if verbose { + println!("Ref {} is up to date.", update.name); + } + continue; + } + + let command = format!("{old_sha} {} {}", update.commit_hash, update.name); + + let pkt = create_pkt_line(&format!("{command}\n")); + channel + .data(pkt.as_bytes().as_ref()) + .await + .map_err(NetworkError::from)?; + + updates_sent += 1; + } + + channel + .data(b"0000".as_ref()) + .await + .map_err(NetworkError::from)?; + + if updates_sent == 0 && verbose { + println!("Everything up-to-date."); + } + + Ok(updates_sent) + } + + /// Helper to construct a [`ReceiveResult`] object from status data. + fn create_result(&self, ref_name: &str, status: ReceiveStatus) -> ReceiveResult { + let old_hash = self + .server_refs + .iter() + .find(|r| r.name == ref_name) + .map(|r| r.commit_hash.clone()) + .unwrap_or_else(|| ZERO_HASH.to_string()); + + let new_hash = self + .proposed_updates + .iter() + .find(|r| r.name == ref_name) + .map(|r| r.commit_hash.clone()) + .unwrap_or_else(|| ZERO_HASH.to_string()); + + ReceiveResult::new(status, ref_name.to_string(), old_hash, new_hash) + } +} diff --git a/engine/src/network/protocols/receive_pack/receive_pack_result.rs b/engine/src/network/protocols/receive_pack/receive_pack_result.rs new file mode 100644 index 00000000..053aa142 --- /dev/null +++ b/engine/src/network/protocols/receive_pack/receive_pack_result.rs @@ -0,0 +1,148 @@ +use std::fmt::Display; + +use owo_colors::OwoColorize; + +/// Represents the overall result of a `receive-pack` (push) operation returned by the remote server. +/// +/// This struct aggregates the status of the packfile unpacking process and the +/// individual results for every reference update requested by the client. +#[derive(Debug, Default)] +pub struct ReceivePackResult { + /// Indicates whether the packfile containing objects was successfully unpacked and verified by the server. + /// If this is false, no references were updated. + pub unpack_successful: bool, + + /// A list of results for each reference update requested by the client. + /// Contains success or failure details for each ref. + pub receive_results: Vec, +} + +/// Detailed status of a specific reference update attempt. +#[derive(Debug, Clone)] +pub struct ReceiveResult { + /// The outcome of the update (Success or Failure with reason). + pub status: ReceiveStatus, + + /// The full name of the reference (e.g., `refs/heads/master`). + pub ref_name: String, + + /// The commit hash of the reference *before* the update. + /// If this is the zero-hash, it indicates branch creation. + pub old_hash: String, + + /// The commit hash of the reference *after* the update. + /// If this is the zero-hash, it indicates branch deletion. + pub new_hash: String, +} + +impl ReceiveResult { + /// Creates a new [`ReceiveResult`] instance. + pub fn new( + status: ReceiveStatus, + ref_name: String, + old_hash: String, + new_hash: String, + ) -> Self { + Self { + status, + ref_name, + old_hash, + new_hash, + } + } + + /// Checks if this update represents the creation of a new branch. + pub fn is_new_branch(&self) -> bool { + Self::is_zero_hash(&self.old_hash) + } + + /// Checks if this update represents the deletion of a branch. + pub fn is_deletion(&self) -> bool { + Self::is_zero_hash(&self.new_hash) + } + + /// Returns a shortened version (first 7 characters) of the old hash. + pub fn old_hash_short(&self) -> &str { + &self.old_hash[0..7] + } + + /// Returns a shortened version (first 7 characters) of the new hash. + pub fn new_hash_short(&self) -> &str { + &self.new_hash[0..7] + } + + /// Helper function to determine if a given hash is the zero-hash. + fn is_zero_hash(hash: &str) -> bool { + for digit in hash.chars() { + if digit != '0' { + return false; + } + } + true + } +} + +/// Represents the success or failure status of a specific reference update. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReceiveStatus { + /// The reference was updated successfully. + Success, + /// The update failed. Contains the error message returned by the server + /// (e.g., "non-fast-forward", "failed to lock"). + Failure(String), +} + +impl Display for ReceiveResult { + /// Formats the receive result for display, using colors to indicate status. + /// + /// * **New Branch**: Displays `+ [new branch]` in green. + /// * **Deletion**: Displays `- [deleted]` in red. + /// * **Update**: Displays `old..new` hash range in green. + /// * **Failure**: Displays `! [rejected]` in red with the failure reason. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match &self.status { + ReceiveStatus::Success => { + if self.is_new_branch() { + write!( + f, + " {} {} {} -> {}", + "+".green(), + "[new branch]".green(), + self.ref_name.cyan(), + self.ref_name.cyan() + )?; + } else if self.is_deletion() { + write!( + f, + " {} {} {}", + "-".red(), + "[deleted]".red(), + self.ref_name.cyan() + )?; + } else { + let short_old = self.old_hash_short(); + let short_new = self.new_hash_short(); + write!( + f, + " {} {} -> {}", + format!("{short_old}..{short_new}").yellow(), + self.ref_name.cyan(), + self.ref_name.cyan() + )?; + } + } + ReceiveStatus::Failure(reason) => { + write!( + f, + " {} {} {} -> {} ({})", + "!".red(), + "[rejected]".red(), + self.ref_name.cyan(), + self.ref_name.cyan(), + reason.dimmed() + )?; + } + } + Ok(()) + } +} diff --git a/engine/src/network/protocols/receive_pack/receive_pack_state.rs b/engine/src/network/protocols/receive_pack/receive_pack_state.rs new file mode 100644 index 00000000..7c5d747d --- /dev/null +++ b/engine/src/network/protocols/receive_pack/receive_pack_state.rs @@ -0,0 +1,24 @@ +use strum_macros::Display; + +/// Represents the various states of the `meva-receive-pack` protocol state machine. +#[derive(Debug, Default, PartialEq, Eq, Display)] +pub enum ReceivePackState { + /// The initial state where the server advertises its capabilities and available references (refs) + /// to the connecting client. + #[default] + Discovery, + + /// The client is currently calculating and sending update commands to the server. + SendingCommands, + + /// The state of generating and streaming the binary packfile containing + /// the requested objects and the checksum. + Packfile, + + /// The client is waiting for the server to confirm success/failure. + ReceivingReport, + + /// Indicates that the operation has completed successfully and the connection + /// can be closed. + Complete, +} diff --git a/engine/src/network/protocols/upload_pack.rs b/engine/src/network/protocols/upload_pack.rs index 4be246a7..2bcb0e83 100644 --- a/engine/src/network/protocols/upload_pack.rs +++ b/engine/src/network/protocols/upload_pack.rs @@ -2,6 +2,7 @@ mod upload_pack_result; mod upload_pack_state; use itertools::Itertools; +use owo_colors::OwoColorize; pub use upload_pack_result::UploadPackResult; pub use upload_pack_state::UploadPackState; @@ -11,7 +12,7 @@ use russh::{Channel, client::Msg}; use crate::{ errors::{EngineResult, NetworkError}, - network::{PktLine, common::ChannelBand, create_pkt_line}, + network::{PktLine, common::ChannelBand, create_pkt_line, parse_next_pkt_line}, ref_manager::RefEntry, }; @@ -83,11 +84,12 @@ impl UploadPackProtocol { &mut self, data: &[u8], channel: &mut Channel, + verbose: bool, ) -> EngineResult { self.buffer.extend_from_slice(data); loop { - let packet = self.parse_next_pkt_line()?; + let packet = parse_next_pkt_line(&mut self.buffer)?; match packet { PktLine::Incomplete => { @@ -95,10 +97,10 @@ impl UploadPackProtocol { break; } PktLine::Payload(payload) => { - self.handle_line(payload).await?; + self.handle_line(payload, verbose).await?; } PktLine::Flush => { - self.handle_flush(channel).await?; + self.handle_flush(channel, verbose).await?; } } @@ -109,66 +111,37 @@ impl UploadPackProtocol { Ok(false) } - /// Attempts to parse the next `pkt-line` from the internal buffer. - /// - /// A `pkt-line` consists of a 4-byte hex length prefix followed by the payload. - /// - If length is "0000", it returns [`PktLine::Flush`]. - /// - If buffer is shorter than the specified length, returns [`PktLine::Incomplete`]. - fn parse_next_pkt_line(&mut self) -> EngineResult { - if self.buffer.len() < 4 { - return Ok(PktLine::Incomplete); - } - - let len_str = from_utf8(&self.buffer[0..4]).map_err(NetworkError::from)?; - - let len = usize::from_str_radix(len_str, 16).map_err(NetworkError::from)?; - - if len == 0 { - self.buffer.drain(0..4); - return Ok(PktLine::Flush); - } - - if self.buffer.len() < len { - return Ok(PktLine::Incomplete); - } - - let line_data = self.buffer[4..len].to_vec(); - self.buffer.drain(0..len); - - Ok(PktLine::Payload(line_data)) - } - /// Handles a parsed payload packet based on the current protocol state. - async fn handle_line(&mut self, payload: Vec) -> EngineResult<()> { + async fn handle_line(&mut self, payload: Vec, verbose: bool) -> EngineResult<()> { match self.state { - UploadPackState::Discovery => { - let line_str = from_utf8(&payload).map_err(NetworkError::from)?; - println!("Received line: {line_str}"); - } UploadPackState::ReceivingRefs => { let line_str = from_utf8(&payload).map_err(NetworkError::from)?; - println!("Received line: {line_str}"); if !line_str.starts_with('#') && let Some(sha) = line_str.split_whitespace().next() { let ref_name = line_str.split_whitespace().nth(1).unwrap_or("").to_string(); - println!("Received ref: {sha} ({ref_name})"); + if verbose { + println!("Received ref: {} ({ref_name})", sha.yellow()); + } self.refs.push(RefEntry::new(&ref_name, sha)); } } UploadPackState::Negotiation => { let line_str = from_utf8(&payload).map_err(NetworkError::from)?; - println!("Received line: {line_str}"); if line_str.starts_with("NAK") { - println!("Received NAK. Waiting for packfile..."); + if verbose { + println!("Received NAK. Waiting for packfile..."); + } self.state = UploadPackState::Packfile; } else if line_str.starts_with("ACK") { - println!("Received ACK. Waiting for packfile..."); + if verbose { + println!("Received ACK. Waiting for packfile..."); + } self.state = UploadPackState::Packfile; } else { - eprintln!("Unexpected line in Negotiation state: {line_str}"); + eprintln!("Unexpected line during negotiation: {line_str}"); } } UploadPackState::Packfile => { @@ -182,12 +155,14 @@ impl UploadPackProtocol { match channel_band.try_into()? { ChannelBand::Packfile => { - println!("Received packfile data ({} bytes)", data.len()); + if verbose { + println!("Received packfile data ({} bytes)", data.len()); + } self.packfile_data.extend_from_slice(data); } ChannelBand::Progress => { let progress_msg = from_utf8(data).unwrap_or("[progress error]").trim(); - eprintln!("Remote: {progress_msg}"); + eprintln!("{} {}", "Remote: ".dimmed(), progress_msg.green()); } ChannelBand::Error => { let error_msg = from_utf8(data).unwrap_or("[remote error]").trim(); @@ -199,25 +174,30 @@ impl UploadPackProtocol { } } } - UploadPackState::Complete => {} + _ => {} } Ok(()) } /// Handles a "flush" packet (0000), which signals a transition or end of a list. - async fn handle_flush(&mut self, channel: &mut Channel) -> EngineResult<()> { - println!("Received flush-packet (0000)"); - + async fn handle_flush( + &mut self, + channel: &mut Channel, + verbose: bool, + ) -> EngineResult<()> { match self.state { UploadPackState::Discovery => { - println!("Finished discovery. Waiting for references..."); + if verbose { + println!("Finished discovery. Waiting for references..."); + } self.state = UploadPackState::ReceivingRefs; } UploadPackState::ReceivingRefs => { - println!("Finished references."); + if verbose { + println!("Finished receiving references."); + } if self.discovery_only { - println!("Discovery only mode. Sending flush and closing."); channel .data(b"0000".as_ref()) .await @@ -227,7 +207,6 @@ impl UploadPackProtocol { return Ok(()); } - println!("Sending 'want'..."); self.wants = self .refs .iter() @@ -241,7 +220,6 @@ impl UploadPackProtocol { .data(want_line.as_bytes().as_ref()) .await .map_err(NetworkError::from)?; - print!("Sending: {want_line}"); } for sha in &self.haves { @@ -250,7 +228,6 @@ impl UploadPackProtocol { .data(have_line.as_bytes().as_ref()) .await .map_err(NetworkError::from)?; - print!("Sending: {have_line}"); } channel @@ -263,16 +240,16 @@ impl UploadPackProtocol { .data(done_line.as_bytes().as_ref()) .await .map_err(NetworkError::from)?; - println!("Sending 'done'"); self.state = UploadPackState::Negotiation; } - UploadPackState::Negotiation => {} UploadPackState::Packfile => { - println!("Finished transferring packfile."); + if verbose { + println!("Finished transferring packfile."); + } self.state = UploadPackState::Complete; } - UploadPackState::Complete => {} + _ => {} } Ok(()) } diff --git a/engine/src/network/protocols/upload_pack/upload_pack_state.rs b/engine/src/network/protocols/upload_pack/upload_pack_state.rs index 7cf301cd..b3ac49b9 100644 --- a/engine/src/network/protocols/upload_pack/upload_pack_state.rs +++ b/engine/src/network/protocols/upload_pack/upload_pack_state.rs @@ -1,8 +1,7 @@ -/// Represents the lifecycle states of the `upload-pack` protocol execution. -/// -/// This state machine tracks the server-side progress during a fetch or clone operation, -/// transitioning from the initial handshake to the final data transfer. -#[derive(Debug, Default, PartialEq, Eq)] +use strum_macros::Display; + +/// Represents the various states of the `meva-upload-pack` protocol state machine. +#[derive(Debug, Default, PartialEq, Eq, Display)] pub enum UploadPackState { /// The initial state where the server advertises its capabilities and available references (refs) /// to the connecting client. diff --git a/engine/src/network/remotes.rs b/engine/src/network/remotes.rs index 2cb1a42b..fea831ef 100644 --- a/engine/src/network/remotes.rs +++ b/engine/src/network/remotes.rs @@ -2,7 +2,7 @@ mod meva_remotes_manager; pub use meva_remotes_manager::MevaRemotesManager; -use std::{collections::HashMap, path::Path}; +use std::{collections::HashMap, fmt::Debug, path::Path}; use url::Url; @@ -12,7 +12,7 @@ use crate::{ }; /// Defines the interface for managing remote repository configurations. -pub trait RemotesManager: Send + Sync { +pub trait RemotesManager: Send + Sync + Debug { /// Registers a new remote with the specified configuration. /// /// # Arguments @@ -23,8 +23,13 @@ pub trait RemotesManager: Send + Sync { /// /// # Returns /// The created `RemoteEntry` on success. - fn add_remote(&self, name: &str, url: Url, pub_signing_key: &Path) - -> EngineResult; + fn add_remote( + &self, + name: &str, + url: &Url, + pub_signing_key: &Path, + config_path: Option<&Path>, + ) -> EngineResult; /// Retrieves the configuration for a specific remote by name. /// @@ -59,6 +64,7 @@ pub trait RemotesManager: Send + Sync { &self, name: &str, new_url: &str, + new_server_key: &Path, direction: &RemoteDirection, ) -> EngineResult<()>; diff --git a/engine/src/network/remotes/meva_remotes_manager.rs b/engine/src/network/remotes/meva_remotes_manager.rs index 4ecb01cd..e96a801e 100644 --- a/engine/src/network/remotes/meva_remotes_manager.rs +++ b/engine/src/network/remotes/meva_remotes_manager.rs @@ -1,5 +1,6 @@ use std::{collections::HashMap, path::Path}; +use shared::PathToString; use url::Url; use crate::{ @@ -15,6 +16,7 @@ use super::{RemoteDirection, RemoteEntry, RemotesManager}; /// This manager handles the storage and retrieval of remote repository configurations /// by interacting directly with the local configuration file (typically `.meva/config`). /// It maps high-level remote operations to low-level TOML document edits. +#[derive(Debug)] pub struct MevaRemotesManager; impl MevaRemotesManager { @@ -47,19 +49,38 @@ impl MevaRemotesManager { RemoteDirection::Push => format!("{table_name}.push_url"), } } + + /// Resolves the specific configuration key for a server public key based on the operation direction. + /// + /// This distinguishes between the standard fetch server key and the optional push server key. + /// + /// # Returns + /// * If `direction` is `Fetch`: returns "remotes.{name}.server_key" + /// * If `direction` is `Push`: returns "remotes.{name}.push + fn get_remote_server_key_keys(&self, name: &str, direction: &RemoteDirection) -> String { + let table_name = self.get_remote_key(name); + match direction { + RemoteDirection::Fetch => format!("{table_name}.server_key"), + RemoteDirection::Push => format!("{table_name}.push_server_key"), + } + } } impl RemotesManager for MevaRemotesManager { fn add_remote( &self, name: &str, - url: Url, + url: &Url, pub_signing_key: &Path, + config_path: Option<&Path>, ) -> EngineResult { - let mut doc = self.get_local_document()?; + let mut doc = match config_path { + Some(path) => ConfigDocument::load(path)?, + None => self.get_local_document()?, + }; let remote_key = self.get_remote_key(name); - let remote_entry = RemoteEntry::new(url, pub_signing_key); + let remote_entry = RemoteEntry::new(url.clone(), pub_signing_key); doc.set_table(&remote_key, &remote_entry.to_map())?; doc.save()?; @@ -99,11 +120,14 @@ impl RemotesManager for MevaRemotesManager { &self, name: &str, new_url: &str, + new_server_key: &Path, direction: &RemoteDirection, ) -> EngineResult<()> { let remote_url_key = self.get_remote_url_keys(name, direction); let mut doc = self.get_local_document()?; doc.set(&remote_url_key, new_url)?; + let remote_server_key_key = self.get_remote_server_key_keys(name, direction); + doc.set(&remote_server_key_key, &new_server_key.to_utf8_string())?; doc.save() } diff --git a/engine/src/network/ssh_service.rs b/engine/src/network/ssh_service.rs index 960851ac..df771343 100644 --- a/engine/src/network/ssh_service.rs +++ b/engine/src/network/ssh_service.rs @@ -30,11 +30,17 @@ impl SshService { /// * The TCP connection to the host fails. /// * The SSH handshake fails (e.g., server key mismatch). /// * Authentication is rejected by the server. - pub async fn connect(&self, params: &SshConnectionParams) -> EngineResult { - println!( - "Connecting to {} as user '{}'...", - ¶ms.host_address, ¶ms.user - ); + pub async fn connect( + &self, + params: &SshConnectionParams, + verbose: bool, + ) -> EngineResult { + if verbose { + println!( + "Connecting to {} as user '{}'...", + ¶ms.host_address, ¶ms.user + ); + } let client_keypair = Arc::new(load_secret_key(¶ms.client_key_path, None).map_err(NetworkError::from)?); @@ -58,7 +64,9 @@ impl SshService { socket.set_nodelay(true)?; let mut session = client::connect_stream(config, socket, handler).await?; - println!("SSH session established."); + if verbose { + println!("SSH session established."); + } let auth_success = session .authenticate_publickey(¶ms.user, client_keypair) @@ -69,10 +77,15 @@ impl SshService { return Err(NetworkError::Authentication.into()); } - println!("Public key authentication succeeded!"); + if verbose { + println!("Public key authentication succeeded."); + } Ok(SshSession::new(session)) } + /// Builds the SSH client configuration with sensible defaults. + /// + /// Returns a [`Config`] struct with keepalive and timeout settings. fn build_config(&self) -> Config { Config { keepalive_interval: Some(Duration::from_secs(10)), diff --git a/engine/src/network/ssh_session.rs b/engine/src/network/ssh_session.rs index b2522d19..7b96c098 100644 --- a/engine/src/network/ssh_session.rs +++ b/engine/src/network/ssh_session.rs @@ -1,11 +1,20 @@ -use std::mem; +use std::{collections::HashSet, mem, sync::Arc}; use russh::{ Channel, ChannelMsg, client::{Handle, Msg}, }; -use crate::errors::{EngineResult, NetworkError}; +use crate::{ + errors::{EngineResult, NetworkError}, + network::{ + PKT_CHUNK_SIZE, PackfileCodec, + protocols::{ReceivePackProtocol, ReceivePackResult}, + }, + object_storage::{MevaObjectStorage, ObjectStorage}, + ref_manager::RefEntry, + repositories::meva_repository_layout::MevaRepositoryLayout, +}; use super::{ ClientHandler, @@ -53,6 +62,7 @@ impl SshSession { &mut self, repository_name: &str, haves: Vec, + verbose: bool, ) -> EngineResult { let command = format!("upload-pack {repository_name}"); let mut channel: Channel = self @@ -71,15 +81,19 @@ impl SshSession { while let Some(msg) = channel.wait().await { match msg { ChannelMsg::Data { data } => { - let is_complete = protocol.process_data(&data, &mut channel).await?; + let is_complete = protocol.process_data(&data, &mut channel, verbose).await?; if is_complete { - println!("Protocol finished successfully."); + if verbose { + println!("Protocol finished successfully."); + } protocol_completed = true; } } ChannelMsg::ExitStatus { exit_status } => { - println!("Command finished with status: {exit_status}"); + if verbose { + println!("Remote command exited with status: {exit_status}"); + } if exit_status != 0 { return Err(NetworkError::RemoteCommand { status: exit_status, @@ -87,9 +101,6 @@ impl SshSession { .into()); } } - ChannelMsg::Eof => { - println!("Server finished sending data (EOF)."); - } _ => {} } } @@ -104,9 +115,74 @@ impl SshSession { )) } - pub async fn run_receive_pack(&mut self, _repository_name: &str) -> EngineResult<()> { - // TODO: push - todo!() + /// Executes the `receive-pack` command on the remote server (Client-side Push). + /// + /// This initiates the process of uploading objects and updating references on the remote repository. + /// + /// # Arguments + /// * `repository_name`: The path/name of the repository on the remote server. + /// * `updates`: A list of reference updates (e.g., branch moves) the client wants to perform. + /// * `verbose`: Whether to print detailed logs to stdout. + /// + /// # Returns + /// * `EngineResult`: Contains the status of the operation (success/failure) + /// for each reference update request. + pub async fn run_receive_pack( + &mut self, + repository_name: &str, + updates: &[RefEntry], + verbose: bool, + ) -> EngineResult { + let command = format!("receive-pack {repository_name}"); + let mut channel: Channel = self + .session + .channel_open_session() + .await + .map_err(NetworkError::from)?; + channel + .exec(true, command.as_str()) + .await + .map_err(NetworkError::from)?; + + let mut protocol = ReceivePackProtocol::new(updates.to_vec()); + let mut protocol_completed = false; + + while let Some(msg) = channel.wait().await { + match msg { + ChannelMsg::Data { data } => { + let is_ready_to_send_pack = + protocol.process_data(&data, &mut channel, verbose).await?; + + if is_ready_to_send_pack { + self.stream_packfile(&mut channel, updates, &protocol.server_refs, verbose) + .await?; + protocol.mark_packfile_sent(); + } + + if protocol.is_complete() { + protocol_completed = true; + } + } + ChannelMsg::ExitStatus { exit_status } => { + if verbose { + println!("Remote command exited with status: {exit_status}"); + } + if exit_status != 0 { + return Err(NetworkError::RemoteCommand { + status: exit_status, + } + .into()); + } + } + _ => {} + } + } + + if !protocol_completed { + return Err(NetworkError::ConnectionClosedPrematurely.into()); + } + + Ok(protocol.result) } /// Connects to the remote to list references without downloading objects. @@ -123,6 +199,7 @@ impl SshSession { pub async fn run_ls_remotes( &mut self, repository_name: &str, + verbose: bool, ) -> EngineResult { let command = format!("upload-pack {repository_name}"); let mut channel: Channel = self @@ -140,17 +217,12 @@ impl SshSession { let mut protocol_completed = false; while let Some(msg) = channel.wait().await { - match msg { - ChannelMsg::Data { data } => { - let is_complete = protocol.process_data(&data, &mut channel).await?; - if is_complete { - protocol_completed = true; - channel.close().await.map_err(NetworkError::from)?; - } + if let ChannelMsg::Data { data } = msg { + let is_complete = protocol.process_data(&data, &mut channel, verbose).await?; + if is_complete { + protocol_completed = true; + channel.close().await.map_err(NetworkError::from)?; } - ChannelMsg::ExitStatus { .. } => {} - ChannelMsg::Eof => {} - _ => {} } } @@ -160,4 +232,82 @@ impl SshSession { Ok(UploadPackResult::refs_only(protocol.refs)) } + + /// Streams a packfile containing necessary objects to the remote server. + /// + /// Calculates the difference between what the server has (`server_refs`) + /// and what the client wants to push (`updates`), packs the missing objects, + /// and sends them over the SSH channel wrapped in `pkt-line` format. + /// + /// If no new objects are needed, it simply sends a flush packet. + async fn stream_packfile( + &mut self, + channel: &mut Channel, + updates: &[RefEntry], + server_refs: &[RefEntry], + verbose: bool, + ) -> EngineResult<()> { + let mut wants = HashSet::new(); + let mut haves = HashSet::new(); + + for update in updates { + if update.has_zero_hash() { + continue; + } + wants.insert(update.commit_hash.clone()); + } + + for server_ref in server_refs { + haves.insert(server_ref.commit_hash.clone()); + } + + let repository_layout = Arc::new(MevaRepositoryLayout::discover()?); + let object_storage = MevaObjectStorage::new(repository_layout); + + let objects = object_storage.collect_reachable_objects(&wants, &haves)?; + let objects_count = objects.len(); + + if objects.is_empty() { + if verbose { + println!("No new objects to pack."); + } + channel + .data(b"0000".as_ref()) + .await + .map_err(NetworkError::from)?; + return Ok(()); + } + + if verbose { + println!("Packing {objects_count} objects..."); + } + + let packfile_codec = PackfileCodec::default(); + let packfile = packfile_codec.encode_packfile(&objects)?; + + if verbose { + println!("Packfile size: {} bytes", packfile.len()); + } + + for chunk in packfile.chunks(PKT_CHUNK_SIZE) { + let len = chunk.len() + 4; + let header = format!("{len:04x}"); + channel + .data(header.as_bytes()) + .await + .map_err(NetworkError::from)?; + channel.data(chunk).await.map_err(NetworkError::from)?; + } + + channel + .data(b"0000".as_ref()) + .await + .map_err(NetworkError::from)?; + + if verbose { + println!("Packfile sent."); + } + + Ok(()) + } } diff --git a/engine/src/object_storage.rs b/engine/src/object_storage.rs index a6959a73..1eec7187 100644 --- a/engine/src/object_storage.rs +++ b/engine/src/object_storage.rs @@ -73,4 +73,14 @@ pub trait ObjectStorage: Send + Sync { wants: &'a HashSet, haves: &'a HashSet, ) -> EngineResult>; + + /// Determines if one commit is a descendant of another. + /// + /// # Arguments + /// * `descendant_hash`: The hash of the potential descendant commit. + /// * `ancestor_hash`: The hash of the potential ancestor commit. + /// # Returns + /// `true` if the commit identified by `descendant_hash` is a descendant of + /// the commit identified by `ancestor_hash`, otherwise `false`. + fn is_descendant_of(&self, descendant_hash: &str, ancestor_hash: &str) -> EngineResult; } diff --git a/engine/src/object_storage/meva_dry_run_object_storage.rs b/engine/src/object_storage/meva_dry_run_object_storage.rs index 47ce6e19..a854d348 100644 --- a/engine/src/object_storage/meva_dry_run_object_storage.rs +++ b/engine/src/object_storage/meva_dry_run_object_storage.rs @@ -75,4 +75,8 @@ impl ObjectStorage for MevaDryRunObjectStorage { fn object_exists(&self, _hash: &str) -> EngineResult { unimplemented!() } + + fn is_descendant_of(&self, _descendant_hash: &str, _ancestor_hash: &str) -> EngineResult { + unimplemented!() + } } diff --git a/engine/src/object_storage/meva_object_storage.rs b/engine/src/object_storage/meva_object_storage.rs index 9c447f6d..2ec86e61 100644 --- a/engine/src/object_storage/meva_object_storage.rs +++ b/engine/src/object_storage/meva_object_storage.rs @@ -228,4 +228,33 @@ impl ObjectStorage for MevaObjectStorage { let path = self.object_path(hash); Ok(path.try_exists()?) } + + fn is_descendant_of(&self, ancestor_hash: &str, descendant_hash: &str) -> EngineResult { + if descendant_hash == ancestor_hash { + return Ok(true); + } + + let mut queue: VecDeque = VecDeque::new(); + queue.push_back(descendant_hash.to_string()); + let mut visited: HashSet = HashSet::new(); + + while let Some(current) = queue.pop_front() { + if current == ancestor_hash { + return Ok(true); + } + + if let Ok(object) = self.get_object(¤t) + && let Ok(commit) = MevaCommit::try_from(object) + { + for parent in commit.parents { + if !visited.contains(&parent) { + visited.insert(parent.clone()); + queue.push_back(parent); + } + } + } + } + + Ok(false) + } } diff --git a/engine/src/ref_manager.rs b/engine/src/ref_manager.rs index c2e45e2f..c58c1eaa 100644 --- a/engine/src/ref_manager.rs +++ b/engine/src/ref_manager.rs @@ -116,6 +116,26 @@ pub trait RefManager: Send + Sync { /// Updates or creates a named reference with the provided [`RefEntry`] value. fn update_ref(&self, entry: &RefEntry) -> EngineResult<()>; + /// Updates local remote-tracking references to match the state of the remote repository. + /// + /// This method iterates over the provided list of references (which are known to + /// be new or updated) and writes them to the local reference store. + /// + /// The reference names are automatically mapped from the server's namespace + /// (e.g., `refs/heads/main`) to the local remote-tracking namespace + /// (e.g., `refs/remotes/origin/main`). + /// + /// # Arguments + /// * `origin` - The name of the remote (e.g., "origin"). + /// * `entries` - The list of server references to update locally. + /// * `verbose` - Whether to enable verbose output during the update process. + fn update_remote_refs( + &self, + origin: &str, + entries: &[RefEntry], + verbose: bool, + ) -> EngineResult<()>; + /// Removes a named reference from the repository. /// /// # Arguments diff --git a/engine/src/ref_manager/head.rs b/engine/src/ref_manager/head.rs index 4b28d3d0..8e41e8db 100644 --- a/engine/src/ref_manager/head.rs +++ b/engine/src/ref_manager/head.rs @@ -25,6 +25,7 @@ impl Head { } } + /// Extracts the branch name if the HEAD is in symbolic mode. pub fn extract_branch_name(&self) -> Option { match self.mode == HeadMode::Symbolic { true => Some( @@ -36,6 +37,16 @@ impl Head { false => None, } } + + /// Checks if the HEAD is in symbolic mode. + pub fn is_symbolic(&self) -> bool { + self.mode == HeadMode::Symbolic + } + + /// Checks if the HEAD is in direct mode. + pub fn is_direct(&self) -> bool { + self.mode == HeadMode::Direct + } } impl Default for Head { diff --git a/engine/src/ref_manager/meva_ref_manager.rs b/engine/src/ref_manager/meva_ref_manager.rs index 1d7bbe39..2c2931e7 100644 --- a/engine/src/ref_manager/meva_ref_manager.rs +++ b/engine/src/ref_manager/meva_ref_manager.rs @@ -1,3 +1,4 @@ +use owo_colors::OwoColorize; use walkdir::WalkDir; use crate::RepositoryLayout; @@ -221,6 +222,27 @@ impl RefManager for MevaRefManager { Ok(()) } + fn update_remote_refs( + &self, + origin: &str, + entries: &[RefEntry], + verbose: bool, + ) -> EngineResult<()> { + for remote_ref in entries { + let tracking_name = self.map_head_to_remote_ref(&remote_ref.name, origin); + if verbose { + println!( + "Updating {} -> {}", + tracking_name.cyan(), + remote_ref.commit_hash.yellow() + ); + } + let new_entry = RefEntry::new(&tracking_name, &remote_ref.commit_hash); + self.update_ref(&new_entry)?; + } + Ok(()) + } + fn remove_ref(&self, name: &str) -> EngineResult> { let entry = self.read_ref(name)?; diff --git a/engine/src/ref_manager/ref_entry.rs b/engine/src/ref_manager/ref_entry.rs index 928a3bbb..9809d654 100644 --- a/engine/src/ref_manager/ref_entry.rs +++ b/engine/src/ref_manager/ref_entry.rs @@ -57,6 +57,11 @@ impl RefEntry { } } + /// Checks if the commit hash is a zero hash (all characters are '0'). + pub fn has_zero_hash(&self) -> bool { + self.commit_hash.chars().all(|c| c == '0') + } + /// Helper to create a local branch entry (refs/heads/). fn new_local_branch_entry(branch_name: &str, commit_hash: String) -> Self { Self { @@ -79,6 +84,6 @@ impl MevaEncode for RefEntry {} impl Display for RefEntry { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let short_hash = self.commit_hash.chars().take(7).collect::(); - write!(f, "{} ({})", self.name.green(), short_hash.dimmed()) + write!(f, "{} ({})", self.name.cyan(), short_hash.yellow()) } } diff --git a/engine/src/repositories/meva_repository.rs b/engine/src/repositories/meva_repository.rs index 37e28294..c987d947 100644 --- a/engine/src/repositories/meva_repository.rs +++ b/engine/src/repositories/meva_repository.rs @@ -7,7 +7,10 @@ use std::{ use tempfile::TempDir; -use crate::{EngineResult, InitError, errors::NetworkError, ref_manager::RefManager}; +use crate::{ + EngineResult, InitError, errors::NetworkError, handlers::clone::Request as CloneRequest, + network::RemotesManager, ref_manager::RefManager, +}; use crate::{ config::config_loader::ConfigLoader, ref_manager::{Head, HeadMode}, @@ -35,7 +38,7 @@ pub trait Repository: Send + Sync { &self, objects: &[(MevaObject, Vec)], refs: &[RefEntry], - remote_name: &str, + request: &CloneRequest, ) -> EngineResult>; /// Returns the layout strategy used by this repository. @@ -50,16 +53,19 @@ pub trait Repository: Send + Sync { /// directory structure and configuration files. pub struct MevaRepository { /// Strategy for resolving file paths within the repository. - pub layout: Arc, + layout: Arc, /// Component responsible for reading and writing configuration files. - pub config_loader: Arc, + config_loader: Arc, /// Backend storage for repository objects (blobs, trees, commits). - pub object_storage: Arc, + object_storage: Arc, /// Manager for branch and reference handling within the repository. - pub ref_manager: Arc, + ref_manager: Arc, + + /// Manager for configuring remotes. + remotes_manager: Arc, } impl MevaRepository { @@ -69,12 +75,14 @@ impl MevaRepository { config_loader: Arc, object_storage: Arc, ref_manager: Arc, + remotes_manager: Arc, ) -> Self { Self { layout, config_loader, object_storage, ref_manager, + remotes_manager, } } @@ -91,7 +99,7 @@ impl MevaRepository { } /// Creates the physical directory hierarchy for a new repository. - fn initialize_structure(&self, root: &Path) -> EngineResult<()> { + fn initialize_structure(&self, root: &Path) -> EngineResult { let dirs = [ self.layout.objects_dir_rel(), self.layout.refs_dir_rel(), @@ -109,7 +117,7 @@ impl MevaRepository { let config_path = root.join(self.layout.config_file_rel()); self.config_loader.create_local_config(&config_path)?; - Ok(()) + Ok(config_path) } /// Configures the `HEAD` file to point to the initial branch. @@ -317,12 +325,19 @@ impl Repository for MevaRepository { &self, objects: &[(MevaObject, Vec)], refs: &[RefEntry], - remote_name: &str, + request: &CloneRequest, ) -> EngineResult> { self.check_if_exists()?; let working_dir = self.layout.working_dir(); - self.initialize_structure(&working_dir)?; + let config_path = self.initialize_structure(&working_dir)?; + + self.remotes_manager.add_remote( + &request.origin, + &request.url, + &request.server_key, + Some(&config_path), + )?; if !objects.is_empty() { self.object_storage.add_objects_from_packfile(objects)?; @@ -337,7 +352,7 @@ impl Repository for MevaRepository { refs, &heads_refs_prefix, initial_branch_name, - remote_name, + &request.origin, )?; self.setup_head( @@ -358,6 +373,7 @@ impl Repository for MevaRepository { #[cfg(test)] mod tests { use crate::MevaConfigLoader; + use crate::network::MevaRemotesManager; use crate::object_storage::MevaObjectStorage; use crate::ref_manager::MevaRefManager; use crate::repositories::meva_repository_layout::MevaRepositoryLayout; @@ -384,7 +400,14 @@ mod tests { let config_loader = Arc::new(MevaConfigLoader::default()); let object_storage = Arc::new(MevaObjectStorage::new(layout.clone())); let ref_manager = Arc::new(MevaRefManager::new(layout.clone())); - MevaRepository::new(layout, config_loader, object_storage, ref_manager) + let remotes_manager = Arc::new(MevaRemotesManager); + MevaRepository::new( + layout, + config_loader, + object_storage, + ref_manager, + remotes_manager, + ) } #[rstest] diff --git a/gui/src/ui/components/remote_actions.rs b/gui/src/ui/components/remote_actions.rs index 898fc25a..f2f67b4f 100644 --- a/gui/src/ui/components/remote_actions.rs +++ b/gui/src/ui/components/remote_actions.rs @@ -11,7 +11,7 @@ use crate::events::{AsyncWorker, EventError, WorkerEvent, WorkerResult}; use engine::{ EngineContainer, engine_container::MevaContainer, - handlers::{fetch::Request as FetchRequest, status::BranchInfo}, + handlers::{fetch::Request as FetchRequest, push::Request as PushRequest, status::BranchInfo}, }; use super::IconButton; @@ -64,24 +64,17 @@ impl<'a> RemoteActionsComponent<'a> { let branch = self.branch.head.clone().unwrap_or_default(); if IconButton::new(icons::ARROWS_CLOCKWISE, "Sync (Pull & Push)").show(ui) { - self.spawn_action("Syncing repository...", "Sync failed", |_container| { - thread::sleep(std::time::Duration::from_millis(800)); - Ok(()) - }); + // TODO: Implement sync operation + self.handle_fetch_click(upstream.clone(), branch.clone()); } if IconButton::new(icons::ARROW_UP, "Push to upstream").show(ui) { - self.spawn_action("Pushing changes...", "Push failed", |_container| { - thread::sleep(std::time::Duration::from_millis(800)); - Ok(()) - }); + self.handle_push_click(upstream.clone(), branch.clone()); } if IconButton::new(icons::ARROW_DOWN, "Pull from upstream").show(ui) { - self.spawn_action("Pulling changes...", "Pull failed", |_container| { - thread::sleep(std::time::Duration::from_millis(800)); - Ok(()) - }); + // TODO: Implement pull operation + self.handle_fetch_click(upstream.clone(), branch.clone()); } if IconButton::new(icons::DOWNLOAD_SIMPLE, "Fetch from upstream").show(ui) { @@ -91,27 +84,43 @@ impl<'a> RemoteActionsComponent<'a> { ui.separator(); } + fn create_runtime() -> Result { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| format!("Failed to create runtime: {e}")) + } + + fn handle_push_click(&mut self, upstream: String, branch: String) { + self.spawn_action("Pushing changes...", "Push failed", move |container| { + let rt = Self::create_runtime()?; + + rt.block_on(async { + let push_handler = container.push_handler().map_err(|e| e.to_string())?; + let request = PushRequest::new(upstream, Some(branch)); + let _ = push_handler + .handle_push(request) + .await + .map_err(|e| e.to_string())?; + + Ok::<(), String>(()) + })?; + + Ok(()) + }); + } + /// Triggers the background fetch operation. /// /// Since the fetch handler uses `async/await`, this method creates a temporary /// Tokio runtime inside the worker thread to execute the future. fn handle_fetch_click(&mut self, upstream: String, branch: String) { self.spawn_action("Fetching origin...", "Fetch failed", move |container| { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .map_err(|e| format!("Failed to create runtime: {e}"))?; + let rt = Self::create_runtime()?; rt.block_on(async { let fetch_handler = container.fetch_handler().map_err(|e| e.to_string())?; - - let request = FetchRequest { - origin: upstream, - branch: Some(branch), - prune: false, - verbose: false, - }; - + let request = FetchRequest::new(upstream, Some(branch)); let _ = fetch_handler .handle_fetch(request) .await diff --git a/server/src/enums.rs b/server/src/enums.rs index 4c61543a..4697a46d 100644 --- a/server/src/enums.rs +++ b/server/src/enums.rs @@ -1,6 +1,7 @@ mod active_protocol; mod channel_state; mod protocol_command; +mod receive_pack_command; mod receive_pack_state; mod upload_pack_command; mod upload_pack_state; @@ -8,6 +9,7 @@ mod upload_pack_state; pub use active_protocol::ActiveProtocol; pub use channel_state::ChannelState; pub use protocol_command::ProtocolCommand; -pub use receive_pack_state::ReceivePackState; +pub use receive_pack_command::ReceivePackCommand; +pub use receive_pack_state::{ReceivePackState, ReceivePhase}; pub use upload_pack_command::UploadPackCommand; pub use upload_pack_state::UploadPackState; diff --git a/server/src/enums/receive_pack_command.rs b/server/src/enums/receive_pack_command.rs new file mode 100644 index 00000000..ffee2ab7 --- /dev/null +++ b/server/src/enums/receive_pack_command.rs @@ -0,0 +1,68 @@ +use std::fmt::Display; + +use crate::errors::ReceivePackError; + +/// Represents a command sent by the client during the `receive-pack` (push) protocol. +#[derive(Debug, Clone)] +pub enum ReceivePackCommand { + /// Requests an update of a specific reference. + Update { + /// The object ID the client expects the reference to currently point to. + /// If this is a 'zero-hash', it indicates reference creation. + old_sha: String, + /// The new object ID the client wants the reference to point to. + /// If this is a 'zero-hash', it indicates reference deletion. + new_sha: String, + /// The full name of the reference (e.g., `refs/heads/master`). + ref_name: String, + }, +} + +impl TryFrom<&str> for ReceivePackCommand { + type Error = ReceivePackError; + + /// Parses a raw protocol line into a structured update command. + /// + /// The expected format is: + /// ` ` + /// + /// # Arguments + /// + /// * `value` - A single line string from the received packet. + /// + /// # Errors + /// + /// Returns [`ReceivePackError::InvalidCommand`] if the line does not contain + /// exactly three space-separated components. + fn try_from(value: &str) -> Result { + let parts: Vec<&str> = value.split_whitespace().collect(); + + if parts.len() != 3 { + return Err(ReceivePackError::InvalidCommand(value.to_string())); + } + + Ok(ReceivePackCommand::Update { + old_sha: parts[0].to_string(), + new_sha: parts[1].to_string(), + ref_name: parts[2].to_string(), + }) + } +} + +impl Display for ReceivePackCommand { + /// Formats the command back into its protocol string representation. + /// + /// # Returns + /// + /// A string in the format: + /// ` ` + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ReceivePackCommand::Update { + old_sha, + new_sha, + ref_name, + } => write!(f, "{old_sha} {new_sha} {ref_name}"), + } + } +} diff --git a/server/src/enums/receive_pack_state.rs b/server/src/enums/receive_pack_state.rs index f90779fe..b6d9ccb6 100644 --- a/server/src/enums/receive_pack_state.rs +++ b/server/src/enums/receive_pack_state.rs @@ -1,14 +1,48 @@ use std::path::PathBuf; -#[derive(Debug, Clone)] +use super::ReceivePackCommand; + +/// Represents the current phase of the `receive-pack` (push) protocol on the server side. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub enum ReceivePhase { + /// The server is waiting for or reading update commands (e.g., ` `). + /// This phase ends when a flush packet (`0000`) is received. + #[default] + ReadingCommands, + /// The server has finished reading commands and is now receiving the binary packfile. + /// This phase involves accumulating raw bytes until the full packfile is received and verified. + ReceivingPackfile, +} + +/// Maintains the state of a single `receive-pack` session for a specific repository. +#[derive(Debug, Default, Clone)] pub struct ReceivePackState { + /// The physical path to the repository on the server's filesystem. pub repository_path: PathBuf, + + /// The current phase of the protocol. + pub state: ReceivePhase, + + /// The list of update commands received from the client so far. + pub commands: Vec, + + /// A buffer used to accumulate the binary packfile data received from the network. + /// This buffer is populated chunk by chunk during the `ReceivingPackfile` phase. + pub packfile_buffer: Vec, } impl ReceivePackState { + /// Creates a new state instance for a specific repository. + /// + /// Initializes the state in the `ReadingCommands` phase with empty buffers. + /// + /// # Arguments + /// + /// * `repository_path` - The filesystem path to the target repository. pub fn new(repository_path: impl Into) -> Self { Self { repository_path: repository_path.into(), + ..Default::default() } } } diff --git a/server/src/errors.rs b/server/src/errors.rs index f1086ac7..7abbbbc1 100644 --- a/server/src/errors.rs +++ b/server/src/errors.rs @@ -1,5 +1,7 @@ +mod receive_pack_error; mod server_error; mod upload_pack_error; +pub use receive_pack_error::ReceivePackError; pub use server_error::{Result as ServerResult, ServerError}; pub use upload_pack_error::UploadPackError; diff --git a/server/src/errors/receive_pack_error.rs b/server/src/errors/receive_pack_error.rs new file mode 100644 index 00000000..c55c7e06 --- /dev/null +++ b/server/src/errors/receive_pack_error.rs @@ -0,0 +1,12 @@ +use thiserror::Error; + +/// Represents errors specific to the `receive-pack` protocol execution. +#[derive(Error, Debug)] +pub enum ReceivePackError { + /// Indicates that a protocol command line was malformed or unrecognized. + #[error("Invalid receive-pack command: {0}")] + InvalidCommand( + /// The raw string content of the command that caused the error. + String, + ), +} diff --git a/server/src/errors/server_error.rs b/server/src/errors/server_error.rs index 21d62bdd..b90316b7 100644 --- a/server/src/errors/server_error.rs +++ b/server/src/errors/server_error.rs @@ -3,7 +3,7 @@ use std::io; use flexi_logger::FlexiLoggerError; use thiserror::Error; -use super::UploadPackError; +use super::{ReceivePackError, UploadPackError}; /// A convenient result type alias for server-related operations. pub type Result = std::result::Result; @@ -19,6 +19,10 @@ pub enum ServerError { #[error(transparent)] UploadPack(#[from] UploadPackError), + /// Errors specific to the `receive-pack` protocol negotiation. + #[error(transparent)] + ReceivePack(#[from] ReceivePackError), + /// Errors originating from the logging subsystem initialization. #[error(transparent)] Logger(#[from] FlexiLoggerError), diff --git a/server/src/logging.rs b/server/src/logging.rs index 09480718..58b4df00 100644 --- a/server/src/logging.rs +++ b/server/src/logging.rs @@ -48,7 +48,7 @@ pub fn init_logging(logging_config: &LoggingConfig) -> ServerResult<()> { "[{}] [{}] [{}:{}] {}", now.now().format("%Y-%m-%d %H:%M:%S"), record.level(), - record.file().unwrap_or("?"), + record.file().unwrap_or(""), record.line().unwrap_or(0), &record.args() ) @@ -60,7 +60,7 @@ pub fn init_logging(logging_config: &LoggingConfig) -> ServerResult<()> { "[{}] [{}] [{}:{}] {}", now.now().format("%Y-%m-%d %H:%M:%S"), style(record.level()).paint(record.level().to_string()), - record.file().unwrap_or("?"), + record.file().unwrap_or(""), record.line().unwrap_or(0), &record.args() ) diff --git a/server/src/server_handler.rs b/server/src/server_handler.rs index 4055b003..34d5ca6f 100644 --- a/server/src/server_handler.rs +++ b/server/src/server_handler.rs @@ -1,5 +1,5 @@ use anyhow::Result; -use log::{error, info, warn}; +use log::{debug, error, info, warn}; use russh::{ Channel, keys::key, @@ -9,25 +9,26 @@ use russh::{ use std::{ collections::HashMap, + io::ErrorKind, path::{Path, PathBuf}, str::FromStr, sync::Arc, }; use ::server::{ - ActiveProtocol, AuthResult, ChannelState, ProtocolCommand, ReceivePackState, UploadPackCommand, - UploadPackState, challenge_public_key, check_repository_access, validate_repository_name, + ActiveProtocol, AuthResult, ChannelState, ProtocolCommand, ReceivePackCommand, + ReceivePackState, ReceivePhase, UploadPackCommand, UploadPackState, challenge_public_key, + check_repository_access, validate_repository_name, }; use engine::{ - network::{ChannelBand, PackfileCodec, SessionExtension, create_pkt_line}, + errors::EngineError, + network::{CHUNK_SIZE, ChannelBand, PackfileCodec, SessionExtension, create_pkt_line}, object_storage::{MevaObjectStorage, ObjectStorage}, - ref_manager::{MevaRefManager, RefManager}, + objects::MevaObject, + ref_manager::{MevaRefManager, RefEntry, RefManager}, repositories::meva_repository_layout::MevaRepositoryLayout, }; -// Stream data in chunks to avoid choking the connection -const CHUNK_SIZE: usize = 65_000; - /// The main SSH event handler for the Meva server. /// /// This struct manages the lifecycle of SSH connections, including authentication @@ -67,7 +68,33 @@ impl ServerHandler { channel: ChannelId, repository_root: &Path, ) -> Result<()> { - let line = create_pkt_line("# service=meva-upload-pack"); + self.handle_pack_command_start(session, channel, repository_root, "meva-upload-pack") + .await + } + + /// Initiates the `receive-pack` protocol (Discovery Phase). + /// + /// It sends the initial advertisement of references (branches, tags) and capabilities + /// to the client. + async fn handle_receive_pack_start( + &mut self, + session: &mut Session, + channel: ChannelId, + repository_root: &Path, + ) -> Result<()> { + self.handle_pack_command_start(session, channel, repository_root, "meva-receive-pack") + .await + } + + /// Sends the initial advertisement of references for a pack command. + async fn handle_pack_command_start( + &mut self, + session: &mut Session, + channel: ChannelId, + repository_root: &Path, + command_header: &str, + ) -> Result<()> { + let line = create_pkt_line(&format!("# service={command_header}")); session.send_pkt_line(channel, &line); session.send_flush(channel); @@ -75,7 +102,6 @@ impl ServerHandler { let ref_manager = MevaRefManager::new(repository_layout); if let Some(head) = ref_manager.resolve_head()? { - info!("(UploadPack) Sending {head} HEAD..."); let head_pkt_line = create_pkt_line(&format!("{head} HEAD")); session.send_pkt_line(channel, &head_pkt_line); } @@ -86,20 +112,10 @@ impl ServerHandler { } session.send_flush(channel); - info!("(UploadPack) List of references sent. Waiting for 'want'/'have'..."); + debug!("List of references sent."); Ok(()) } - async fn handle_receive_pack_start( - &mut self, - _session: &mut Session, - _channel: ChannelId, - _repo_path: &Path, - ) -> Result<()> { - // TODO: push - todo!() - } - /// Processes incoming data chunks during the `upload-pack` negotiation phase. /// /// This method acts as a state machine parser for the client's requests. @@ -132,12 +148,11 @@ impl ServerHandler { }; if len == 0 { - info!("(UploadPack) Received flush-packet (0000)"); state.buffer.drain(0..4); if upload_state.wants.is_empty() && upload_state.haves.is_empty() { info!( - "(UploadPack) Client sent flush without wants/haves (ls-remote). Closing channel." + "Client sent flush without wants/haves (indicating ls-remote). Closing channel." ); session.exit_status_request(channel, 0); session.eof(channel); @@ -152,23 +167,21 @@ impl ServerHandler { } let line_data = &state.buffer[4..len]; - let line_str = std::str::from_utf8(line_data) - .unwrap_or("[utf8 error]") - .trim(); + let line_str = std::str::from_utf8(line_data)?.trim(); let command = UploadPackCommand::try_from(line_str)?; match command { UploadPackCommand::Want(sha) => { - info!("(UploadPack) Received 'want': {sha}"); + debug!("Received 'want': {sha}"); upload_state.wants.insert(sha.to_string()); } UploadPackCommand::Have(sha) => { - info!("(UploadPack) Received 'have': {sha}"); + debug!("Received 'have': {sha}"); upload_state.haves.insert(sha.to_string()); } UploadPackCommand::Done => { - info!("(UploadPack) Received 'done'. Negotiation finished."); + debug!("Received 'done'. Negotiation finished."); state.buffer.drain(0..len); return self.handle_packfile_generation(session, channel); } @@ -180,13 +193,269 @@ impl ServerHandler { Ok(()) } + /// Processes incoming data chunks during the `receive-pack` phase. + /// + /// This method acts as a state machine parser for the client's push commands + /// and packfile data. fn process_receive_pack_data( &mut self, - _session: &mut Session, - _channel: ChannelId, + session: &mut Session, + channel: ChannelId, + ) -> Result<()> { + let mut ready_to_push = false; + let mut decoded_objects = Vec::new(); + let repository_path; + let commands; + + { + let state_container = self.sessions.get_mut(&channel).unwrap(); + let receive_state = match &mut state_container.protocol { + ActiveProtocol::ReceivePack(state) => state, + _ => return Ok(()), + }; + + repository_path = receive_state.repository_path.clone(); + commands = receive_state.commands.clone(); + + loop { + match receive_state.state { + ReceivePhase::ReadingCommands => { + if state_container.buffer.len() < 4 { + break; + } + + let len_str = + std::str::from_utf8(&state_container.buffer[0..4]).unwrap_or("????"); + let len = usize::from_str_radix(len_str, 16).unwrap_or(0); + + if len == 0 { + debug!("End of commands. Switching to Packfile mode."); + state_container.buffer.drain(0..4); + receive_state.state = ReceivePhase::ReceivingPackfile; + if receive_state.commands.is_empty() { + debug!("No commands received. Skipping Packfile phase."); + + ready_to_push = true; + decoded_objects = Vec::new(); + break; + } + + debug!("End of commands. Switching to Packfile mode."); + receive_state.state = ReceivePhase::ReceivingPackfile; + continue; + } + + if state_container.buffer.len() < len { + break; + } + + let line_data = &state_container.buffer[4..len]; + let line_str = std::str::from_utf8(line_data)?.trim(); + + match ReceivePackCommand::try_from(line_str) { + Ok(command) => { + debug!("Received command: {command}"); + receive_state.commands.push(command); + } + Err(e) => { + error!("Invalid command: {e}"); + return Err(e.into()); + } + } + + state_container.buffer.drain(0..len); + } + + ReceivePhase::ReceivingPackfile => { + if state_container.buffer.len() < 4 { + break; + } + + let len_str = + std::str::from_utf8(&state_container.buffer[0..4]).unwrap_or("????"); + let len = usize::from_str_radix(len_str, 16).unwrap_or(0); + + if len == 0 { + debug!("Received Flush-Packet. End of packfile stream."); + state_container.buffer.drain(0..4); + + if receive_state.packfile_buffer.is_empty() { + debug!("Packfile buffer is empty. No new objects to unpack."); + ready_to_push = true; + decoded_objects = Vec::new(); + break; + } + + let codec = PackfileCodec::default(); + match codec.decode_packfile(&receive_state.packfile_buffer) { + Ok(objects) => { + info!( + "Packfile verified ({} objects, {} bytes). Ready to push.", + objects.len(), + receive_state.packfile_buffer.len() + ); + + ready_to_push = true; + decoded_objects = objects; + receive_state.packfile_buffer.clear(); + break; + } + Err(e) => { + error!("Packfile corrupted or incomplete: {e}"); + return Err(e.into()); + } + } + } + + if state_container.buffer.len() < len { + break; + } + + let chunk_data = &state_container.buffer[4..len]; + receive_state.packfile_buffer.extend_from_slice(chunk_data); + state_container.buffer.drain(0..len); + } + } + } + } + + if ready_to_push { + self.execute_push( + session, + channel, + decoded_objects, + repository_path, + &commands, + )?; + } + + Ok(()) + } + + /// Executes the push operation after receiving and validating the packfile. + /// + /// This method updates references based on the received commands + /// and sends appropriate responses back to the client. + fn execute_push( + &mut self, + session: &mut Session, + channel: ChannelId, + packfile_objects: Vec<(MevaObject, Vec)>, + repository_path: PathBuf, + commands: &[ReceivePackCommand], ) -> Result<()> { - // TODO: push - todo!() + debug!("Executing push with {} objects...", packfile_objects.len()); + + let repository_layout = Arc::new(MevaRepositoryLayout::new(repository_path.clone())?); + let object_storage = MevaObjectStorage::new(repository_layout.clone()); + let ref_manager = MevaRefManager::new(repository_layout); + + if let Err(e) = object_storage.add_objects_from_packfile(&packfile_objects) { + error!("Failed to save objects: {e}"); + let msg = create_pkt_line(&format!("unpack error {e}\n")); + session.send_pkt_line(channel, &msg); + session.send_flush(channel); + session.close(channel); + return Ok(()); + } + + let unpack_ok = create_pkt_line("unpack ok\n"); + session.send_pkt_line(channel, &unpack_ok); + info!("Objects written successfully. Processing refs..."); + + let zero_hash = "0".repeat(40); + + for cmd in commands { + match cmd { + ReceivePackCommand::Update { + old_sha, + new_sha, + ref_name, + } => { + let current_hash = match ref_manager.read_ref(ref_name) { + Ok(Some(r)) => r.commit_hash, + Ok(None) => zero_hash.clone(), + Err(EngineError::Io(e)) if e.kind() == ErrorKind::NotFound => { + if *old_sha == zero_hash { + zero_hash.clone() + } else { + error!("Ref {ref_name} not found for update"); + let msg = + create_pkt_line(&format!("ng {ref_name} ref does not exist\n")); + session.send_pkt_line(channel, &msg); + continue; + } + } + Err(e) => { + error!("Failed to read ref {ref_name}: {e}"); + let msg = create_pkt_line(&format!( + "ng {ref_name} failed to read ref: internal error\n" + )); + session.send_pkt_line(channel, &msg); + continue; + } + }; + + if current_hash != *old_sha { + warn!( + "Rejected update for {ref_name}: expected {old_sha}, found {current_hash}", + ); + let msg = create_pkt_line(&format!( + "ng {ref_name} non-fast-forward (lock mismatch)\n", + )); + session.send_pkt_line(channel, &msg); + continue; + } + + let is_creation = *old_sha == zero_hash; + let is_deletion = *new_sha == zero_hash; + + if !is_creation && !is_deletion { + let is_fast_forward = object_storage + .is_descendant_of(old_sha, new_sha) + .unwrap_or(false); + if !is_fast_forward { + warn!("Non-fast-forward update rejected for {ref_name}"); + let msg = create_pkt_line(&format!("ng {ref_name} non-fast-forward\n")); + session.send_pkt_line(channel, &msg); + continue; + } + } + + let update_result = if *new_sha == zero_hash { + info!("Deleting ref: {ref_name}"); + ref_manager.remove_ref(ref_name).map(|_| ()) + } else { + info!("Updating ref: {ref_name} -> {new_sha}"); + let entry = RefEntry::new(ref_name, new_sha); + ref_manager.update_ref(&entry) + }; + + match update_result { + Ok(_) => { + let msg = create_pkt_line(&format!("ok {ref_name}\n")); + session.send_pkt_line(channel, &msg); + } + Err(e) => { + error!("Failed to write ref {ref_name}: {e}"); + let msg = + create_pkt_line(&format!("ng {ref_name} failed to write ref\n")); + session.send_pkt_line(channel, &msg); + } + } + } + } + } + + info!("Push operation completed. Closing channel."); + + session.send_flush(channel); + session.exit_status_request(channel, 0); + session.eof(channel); + session.close(channel); + info!("Session closed."); + + Ok(()) } /// Generates and streams the packfile to the client. @@ -197,7 +466,7 @@ impl ServerHandler { session: &mut Session, channel: ChannelId, ) -> Result<()> { - info!("(UploadPack) Generating packfile..."); + info!("Generating packfile..."); let state = self.sessions.get(&channel).unwrap(); let upload_state = match &state.protocol { @@ -205,9 +474,6 @@ impl ServerHandler { _ => return Ok(()), }; - info!("(UploadPack) Client 'wants': {:?}", upload_state.wants); - info!("(UploadPack) Client 'haves': {:?}", upload_state.haves); - let repository_layout = Arc::new(MevaRepositoryLayout::new( upload_state.repository_path.clone(), )?); @@ -219,7 +485,6 @@ impl ServerHandler { if object_storage.object_exists(have)? { let ack = create_pkt_line("ACK"); session.send_pkt_line(channel, &ack); - info!("(UploadPack) Found common ancestor: {have}. Sending ACK."); common_ancestor_found = true; break; } @@ -228,34 +493,34 @@ impl ServerHandler { if !common_ancestor_found { let nak = create_pkt_line("NAK"); session.send_pkt_line(channel, &nak); - info!("(UploadPack) No common ancestor found. Sending NAK."); } let objects = object_storage.collect_reachable_objects(&upload_state.wants, &upload_state.haves)?; let objects_count = objects.len(); - info!("(UploadPack) Collected {objects_count} objects to pack."); + debug!("Collected {objects_count} objects to pack."); - session.send_channel_band( - channel, - &ChannelBand::Progress, - format!("Enumerating objects: {objects_count}, done.").as_bytes(), - )?; + if objects_count > 0 { + session.send_channel_band( + channel, + &ChannelBand::Progress, + format!("Enumerating objects: {objects_count}, done.").as_bytes(), + )?; + } let packfile_codec = PackfileCodec::default(); let packfile = packfile_codec.encode_packfile(&objects)?; - session.send_channel_band( - channel, - &ChannelBand::Progress, - format!("Compressing objects: {objects_count}, done.").as_bytes(), - )?; + let progress_msg = if objects_count == 0 { + "No new objects to pack, done.".to_string() + } else { + format!("Compressing objects: {objects_count}, done.") + }; + + session.send_channel_band(channel, &ChannelBand::Progress, progress_msg.as_bytes())?; - info!( - "(UploadPack) Sending packfile ({} bytes)...", - packfile.len() - ); + info!("Sending packfile ({} bytes)...", packfile.len()); let mut chunk_counter = 0; for chunk in packfile.chunks(CHUNK_SIZE) { @@ -263,15 +528,17 @@ impl ServerHandler { chunk_counter += 1; } - session.send_channel_band( - channel, - &ChannelBand::Progress, - format!("Total {chunk_counter} chunks sent.").as_bytes(), - )?; + if objects_count > 0 { + session.send_channel_band( + channel, + &ChannelBand::Progress, + format!("Total {chunk_counter} chunks sent.").as_bytes(), + )?; + } session.send_flush(channel); - info!("(UploadPack) Packfile sent. Closing channel."); + info!("Packfile sent. Closing channel."); session.exit_status_request(channel, 0); session.eof(channel); session.close(channel); @@ -313,7 +580,7 @@ impl Handler for ServerHandler { channel: Channel, _session: &mut Session, ) -> Result { - info!("Attempting to open a session on channel {:?}", channel.id()); + debug!("Attempting to open a session on channel {:?}", channel.id()); if let Some(user) = &self.user { let state = ChannelState::new(user.clone()); self.sessions.insert(channel.id(), state); @@ -485,6 +752,8 @@ impl Handler for ServerHandler { } impl Clone for ServerHandler { + /// Creates a clone of the `ServerHandler`. + /// Used to spawn new handler instances for each SSH connection. fn clone(&self) -> Self { ServerHandler { authorized_keys_path: self.authorized_keys_path.clone(),