move to mostly sync architecture

the git repository struct is not sharable between threads, thus go
single threaded for now and only call onto the tokio runtime for lookups
towards gitlab.

10 files changed, 105 insertions, 67 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 613c674..df85de7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -573,6 +573,7 @@ dependencies = [
  "git2",
  "gitlab",
  "graphql_client",
+ "itertools",
  "serde",
  "thiserror",
  "tokio",
diff --git a/Cargo.toml b/Cargo.toml
index 31df682..a9f1766 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -7,6 +7,8 @@ edition = "2021"
 tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
 futures = "0.3"
 
+itertools = "0.10"
+
 gitlab = "0.1408"
 graphql_client = "0.10"
 
diff --git a/src/forge/gitlab/mod.rs b/src/forge/gitlab/mod.rs
index fac60b7..dec3b49 100644
--- a/src/forge/gitlab/mod.rs
+++ b/src/forge/gitlab/mod.rs
@@ -1,5 +1,4 @@
 use anyhow::Result;
-use gitlab::AsyncGitlab;
 
 use graphql_client::GraphQLQuery;
 
@@ -7,10 +6,11 @@ pub mod config;
 
 #[derive(Clone, Debug)]
 pub struct Gitlab {
-    client: AsyncGitlab,
+    client: gitlab::AsyncGitlab,
 }
 
 impl Gitlab {
+    #[tracing::instrument(level = "trace")]
     pub async fn new(host: &str, token: &str, tls: bool) -> Result<Gitlab> {
         let mut gitlab = gitlab::GitlabBuilder::new(host, token);
 
@@ -23,6 +23,7 @@ impl Gitlab {
         Ok(Gitlab { client: gitlab })
     }
 
+    #[tracing::instrument(level = "trace")]
     pub async fn from_config(forge: &config::Gitlab) -> Result<Gitlab> {
         Gitlab::new(&forge.host, &forge.token, forge.tls).await
     }
diff --git a/src/list/mod.rs b/src/list/mod.rs
index ed71e6a..0bf6872 100644
--- a/src/list/mod.rs
+++ b/src/list/mod.rs
@@ -1,10 +1,14 @@
 use anyhow::Result;
+use itertools::Itertools;
 
 use crate::repo::Repos;
 
 impl crate::GTree {
-    pub async fn list(&self, repos: Repos) -> Result<()> {
-        repos.iter().for_each(|repo| println!("{}", repo));
+    pub fn list(&self, repos: Repos) -> Result<()> {
+        repos.iter().sorted_by_key(|x| x.0).for_each(|(_, repo)| {
+            let repo = repo.read().unwrap();
+            println!("{}", repo)
+        });
 
         Ok(())
     }
diff --git a/src/main.rs b/src/main.rs
index eeec4a3..bad9436 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,7 +1,9 @@
+use std::{thread, sync::Arc};
+
 use anyhow::{Context, Result};
 use clap::Parser;
-use derivative::Derivative;
 
+use tokio::runtime::Runtime;
 use tracing::{debug, metadata::LevelFilter, Level};
 use tracing_subscriber::{fmt::format::FmtSpan, prelude::*, EnvFilter};
 
@@ -19,18 +21,18 @@ mod update;
 #[cfg(test)]
 mod tests;
 
-#[derive(Derivative)]
-#[derivative(Debug)]
+#[derive(Debug)]
 struct GTree {
     figment: figment::Figment,
     config: config::Config,
     args: config::args::Args,
     forge: forge::Forge,
+    rt: Runtime,
 }
 
 impl GTree {
     #[tracing::instrument(level = "trace")]
-    pub async fn new() -> Result<GTree> {
+    pub fn new() -> Result<GTree> {
         let args = config::args::Args::parse();
 
         let figment = config::Config::figment()?;
@@ -41,72 +43,73 @@ impl GTree {
             .next()
             .context("No Forge configured, please setup a forge")?;
 
-        let forge = forge::Forge::new(forge_config).await?;
+        let rt = Runtime::new()?;
+
+
+        let forge = rt.block_on(forge::Forge::new(forge_config))?;
 
         Ok(GTree {
             figment,
             config,
             args,
             forge,
+            rt
         })
     }
 
     #[tracing::instrument(level = "trace")]
-    pub async fn run(self) -> Result<()> {
-        let scope = self.args.scope.as_ref().map_or("", |x| x);
+    pub fn run(self) -> Result<()> {
+        let scope = Arc::new(self.args.scope.as_ref().map_or("", |x| x).to_string());
 
         // TODO select a specific forge
-        let (_name, forge) = self
+        let forge = Arc::new(self
             .config
             .iter()
             .next()
-            .context("No Forge configured, please setup a forge")?;
+            .context("No Forge configured, please setup a forge")?.1.clone());
+
+        let scope_t = scope.clone();
+        let forge_t = forge.clone();
+        let handle = thread::spawn(move || {
+            Repos::from_local(&forge_t.root(), &scope_t)
+        });
 
-        let (local, remote) = tokio::join!(
-            Repos::from_local(forge.root(), scope),
-            Repos::from_forge(forge.root(), self.forge.projects(scope).await?)
-        );
+        let projects = self.rt.block_on(self.forge.projects(&scope))?;
+        let remote = Repos::from_forge(forge.root(), projects);
 
-        let repos = Repos::aggregate(local, remote).await;
+        let local = handle.join().unwrap();
+        let repos = Repos::aggregate(local, remote);
 
         match self.args.command {
-            config::args::Commands::Sync => self.sync(repos).await,
-            config::args::Commands::Update => self.update(repos).await,
-            config::args::Commands::List => self.list(repos).await?,
+            config::args::Commands::Sync => self.sync(repos),
+            config::args::Commands::Update => self.update(repos),
+            config::args::Commands::List => self.list(repos)?,
         };
 
         Ok(())
     }
 }
 
-#[tokio::main]
-async fn main() -> Result<()> {
-    use tracing_flame::FlameLayer;
-
+fn main() -> Result<()> {
     let filter = tracing_subscriber::filter::Targets::new()
         .with_default(Level::TRACE)
         .with_target("hyper", LevelFilter::OFF)
+        .with_target("hyper", LevelFilter::OFF)
         .with_target("reqwest", LevelFilter::OFF);
 
     let env_filter = EnvFilter::from_default_env();
 
-    let (flame_layer, _guard) = FlameLayer::with_file("./tracing.folded").unwrap();
-    let flameguard = flame_layer.flush_on_drop();
-
     tracing_subscriber::registry()
         .with(tracing_subscriber::fmt::layer().with_span_events(FmtSpan::ACTIVE))
         .with(filter)
         .with(env_filter)
-        .with(flame_layer)
         .init();
 
     debug!("starting");
 
-    let gtree = GTree::new().await?;
-
-    gtree.run().await?;
+    let gtree = GTree::new()?;
 
-    flameguard.flush()?;
+    gtree.run()?;
 
     Ok(())
 }
diff --git a/src/repo/aggregate.rs b/src/repo/aggregate.rs
index cb4b00d..0d9573c 100644
--- a/src/repo/aggregate.rs
+++ b/src/repo/aggregate.rs
@@ -1,3 +1,8 @@
+use std::{
+    collections::HashMap,
+    sync::RwLock,
+};
+
 use git2::Repository;
 
 use tracing::error;
@@ -9,16 +14,16 @@ use super::{Repo, Repos};
 
 #[async_trait::async_trait]
 pub trait Aggregator {
-    async fn from_local(root: &str, scope: &str) -> Repos;
-    async fn from_forge(root: &str, projects: Vec<Project>) -> Repos;
-    async fn aggregate(mut local: Repos, mut remote: Repos) -> Repos;
+    fn from_local(root: &str, scope: &str) -> Repos;
+    fn from_forge(root: &str, projects: Vec<Project>) -> Repos;
+    fn aggregate(local: Repos, remote: Repos) -> Repos;
 }
 
 #[async_trait::async_trait]
 impl Aggregator for Repos {
     #[tracing::instrument(level = "trace")]
-    async fn from_local(root: &str, scope: &str) -> Repos {
-        let mut repos = Vec::new();
+    fn from_local(root: &str, scope: &str) -> Repos {
+        let mut repos = HashMap::new();
 
         let path: std::path::PathBuf = [root, scope].iter().collect();
 
@@ -48,18 +53,25 @@ impl Aggregator for Repos {
                     walker.skip_current_dir();
 
                     match Repository::open(entry.path()) {
-                        Ok(repo) => repos.push(Repo {
-                            name: entry
+                        Ok(repo) => {
+                            let name = entry
                                 .path()
                                 .strip_prefix(root)
                                 .unwrap()
                                 .to_str()
                                 .unwrap()
-                                .to_string(),
-                            path: entry.path().to_path_buf(),
-                            repo: Some(repo),
-                            ..Repo::default()
-                        }),
+                                .to_string();
+
+                            repos.insert(
+                                name.clone(),
+                                RwLock::new(Repo {
+                                    name,
+                                    path: entry.path().to_path_buf(),
+                                    repo: Some(repo),
+                                    ..Repo::default()
+                                }),
+                            );
+                        }
                         Err(err) => error!("could not open repository: {}", err),
                     }
                 } else {
@@ -72,33 +84,37 @@ impl Aggregator for Repos {
     }
 
     #[tracing::instrument(level = "trace")]
-    async fn from_forge(root: &str, projects: Vec<Project>) -> Repos {
+    fn from_forge(root: &str, projects: Vec<Project>) -> Repos {
         projects
             .iter()
             .map(|project| {
                 let mut repo: Repo = project.into();
                 repo.path = [root, &repo.name].iter().collect();
-                repo
+                (repo.name.clone(), RwLock::new(repo))
             })
             .collect()
     }
 
+    // TODO optimise this func
+    //
+    // the iteration is currently quite inefficient as
+    // it's constantly removing stuff from `remote`
     #[tracing::instrument(level = "trace", skip(local, remote))]
-    async fn aggregate(mut local: Repos, mut remote: Repos) -> Repos {
+    fn aggregate(mut local: Repos, mut remote: Repos) -> Repos {
         local = local
             .into_iter()
-            .map(|mut left| {
-                if let Some(i) = remote.iter().position(|right| *right == left) {
-                    let right = remote.remove(i);
-                    left.forge = right.forge;
+            .map(|(left_name, left)| {
+                if let Some(right) = remote.remove(&left_name)
+                {
+                    left.write().unwrap().forge = right.into_inner().unwrap().forge;
                 }
 
-                left
+                (left_name, left)
             })
             .collect();
 
-        local.append(&mut remote);
-        local.sort();
+        local.extend(remote.into_iter());
+        // local.sort();
 
         return local;
     }
diff --git a/src/repo/mod.rs b/src/repo/mod.rs
index e3d1279..aaf0177 100644
--- a/src/repo/mod.rs
+++ b/src/repo/mod.rs
@@ -1,4 +1,9 @@
-use std::{fmt::Debug, path::PathBuf};
+use std::{
+    collections::HashMap,
+    fmt::Debug,
+    path::PathBuf,
+    sync::RwLock,
+};
 
 use thiserror::Error;
 
@@ -13,7 +18,8 @@ mod repostate;
 pub use aggregate::*;
 pub use repostate::*;
 
-pub type Repos = Vec<Repo>;
+// pub type Repos = Vec<Repo>;
+pub type Repos = HashMap<String, RwLock<Repo>>;
 
 pub struct Repo {
     pub name: String,
@@ -83,9 +89,7 @@ impl Repo {
         let mut builder = git2::build::RepoBuilder::new();
         builder.fetch_options(crate::git::fetch_options());
 
-        builder
-            .clone(url, &self.path)
-            .map_err(RepoError::GitError)
+        builder.clone(url, &self.path).map_err(RepoError::GitError)
     }
 
     #[tracing::instrument(level = "trace")]
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
index f707b04..271f672 100644
--- a/src/sync/mod.rs
+++ b/src/sync/mod.rs
@@ -3,8 +3,10 @@ use std::fmt::{Debug, Display};
 use crate::repo::{Repo, RepoError, Repos};
 
 impl crate::GTree {
-    pub async fn sync(&self, repos: Repos) {
-        for mut repo in repos {
+    pub fn sync(&self, repos: Repos) {
+        for (name, repo) in repos {
+            let mut repo = repo.write().unwrap();
+
             match repo.sync() {
                 Ok(u) => println!("{}", u),
                 Err(u) => println!("{}", u),
@@ -30,7 +32,7 @@ impl Repo {
         if self.repo.is_some() && self.forge.is_some() {
             Ok(SyncResult::no_changes(repo_name))
         } else if self.repo.is_some() {
-            // do push stuff
+            // TODO do push stuff
             Ok(SyncResult::pushed(repo_name))
         } else if self.forge.is_some() {
             let url = self
@@ -45,6 +47,9 @@ impl Repo {
                 .clone(url)
                 .map_err(|err| SyncResult::err(repo_name.clone(), err))?;
 
+            // TODO detect moved repos based on first commit
+            // ???? how to detect and not move forks?
+
             self.repo = Some(repo);
             Ok(SyncResult::cloned(repo_name))
         } else {
diff --git a/src/tests/mod.rs b/src/tests/mod.rs
index 0d32dbf..3e0f7af 100644
--- a/src/tests/mod.rs
+++ b/src/tests/mod.rs
@@ -48,9 +48,9 @@ async fn search_repos() -> Result<()> {
         format!("{:?}/repos/site/group/subgroup/repo3", TEST_DIR),
         format!("{:?}/repos/site/group/subgroup/subsubgroup/repo4", TEST_DIR),
     ];
-    let right = Repos::from_local(&format!("{:?}/repos", TEST_DIR), "").await;
+    let right = Repos::from_local(&format!("{:?}/repos", TEST_DIR), "");
 
-    let mut right: Vec<&str> = right.iter().map(|x| x.name.as_str()).collect();
+    let mut right: Vec<&str> = right.iter().map(|x| x.0.as_str()).collect();
 
     assert_eq!(left.sort(), right.sort_unstable());
 
diff --git a/src/update/mod.rs b/src/update/mod.rs
index dfc800c..609b35e 100644
--- a/src/update/mod.rs
+++ b/src/update/mod.rs
@@ -6,8 +6,10 @@ use tracing::debug;
 use crate::repo::{Repo, RepoError, Repos};
 
 impl crate::GTree {
-    pub async fn update(&self, repos: Repos) {
-        for mut repo in repos {
+    pub fn update(&self, repos: Repos) {
+        for (name, repo) in repos {
+            let mut repo = repo.write().unwrap();
+
             if repo.repo.is_some() {
                 match repo.update() {
                     Ok(u) => println!("{}", u),