Just-MR: Add main logic for fetch sub-command

2 files changed, 321 insertions, 2 deletions

diff --git a/src/other_tools/just_mr/TARGETS b/src/other_tools/just_mr/TARGETS
index 3f1efd1b..b96d3de7 100644
--- a/src/other_tools/just_mr/TARGETS
+++ b/src/other_tools/just_mr/TARGETS
@@ -8,6 +8,7 @@
     , ["src/buildtool/logging", "logging"]
     , "cli"
     , "exit_codes"
+    , ["src/other_tools/ops_maps", "repo_fetch_map"]
     ]
   , "stage": ["src", "other_tools", "just_mr"]
   , "private-ldflags":
diff --git a/src/other_tools/just_mr/main.cpp b/src/other_tools/just_mr/main.cpp
index 45390c94..89f05f1e 100644
--- a/src/other_tools/just_mr/main.cpp
+++ b/src/other_tools/just_mr/main.cpp
@@ -17,6 +17,7 @@
 #include "src/buildtool/logging/log_sink_cmdline.hpp"
 #include "src/other_tools/just_mr/cli.hpp"
 #include "src/other_tools/just_mr/exit_codes.hpp"
+#include "src/other_tools/ops_maps/repo_fetch_map.hpp"
 
 namespace {
 
@@ -410,6 +411,321 @@ void SetupDefaultLogging() {
     return config;
 }
 
+void ReachableRepositories(ExpressionPtr const& repos,
+                           std::string const& main,
+                           std::shared_ptr<SetupRepos> const& setup_repos) {
+    // make sure the vectors to be populated are empty
+    setup_repos->to_setup.clear();
+    setup_repos->to_include.clear();
+
+    if (repos->IsMap()) {
+        // reserve max size
+        setup_repos->to_include.reserve(repos->Map().size());
+
+        // traversal of bindings
+        std::function<void(std::string const&)> traverse =
+            [&](std::string const& repo_name) {
+                if (std::find(setup_repos->to_include.begin(),
+                              setup_repos->to_include.end(),
+                              repo_name) == setup_repos->to_include.end()) {
+                    // if not found, add it and repeat for its bindings
+                    setup_repos->to_include.emplace_back(repo_name);
+                    auto repos_repo_name =
+                        repos->Get(repo_name, Expression::none_t{});
+                    if (not repos_repo_name.IsNotNull()) {
+                        return;
+                    }
+                    auto bindings =
+                        repos_repo_name->Get("bindings", Expression::none_t{});
+                    if (bindings.IsNotNull() and bindings->IsMap()) {
+                        for (auto const& bound : bindings->Map().Values()) {
+                            if (bound.IsNotNull() and bound->IsString()) {
+                                traverse(bound->String());
+                            }
+                        }
+                    }
+                }
+            };
+        traverse(main);  // traverse all bindings of main repository
+
+        // Add overlay repositories
+        setup_repos->to_setup = setup_repos->to_include;
+        for (auto const& repo : setup_repos->to_include) {
+            auto repos_repo = repos->Get(repo, Expression::none_t{});
+            if (repos_repo.IsNotNull()) {
+                // copy over any present alternative root dirs
+                for (auto const& layer : kAltDirs) {
+                    auto layer_val =
+                        repos_repo->Get(layer, Expression::none_t{});
+                    if (layer_val.IsNotNull() and layer_val->IsString()) {
+                        setup_repos->to_setup.emplace_back(layer_val->String());
+                    }
+                }
+            }
+        }
+    }
+}
+
+void DefaultReachableRepositories(
+    ExpressionPtr const& repos,
+    std::shared_ptr<SetupRepos> const& setup_repos) {
+    if (repos.IsNotNull() and repos->IsMap()) {
+        setup_repos->to_setup = repos->Map().Keys();
+        setup_repos->to_include = setup_repos->to_setup;
+    }
+}
+
+[[nodiscard]] auto MultiRepoFetch(std::shared_ptr<Configuration> const& config,
+                                  CommandLineArguments const& arguments)
+    -> int {
+    // find fetch dir
+    auto fetch_dir = arguments.fetch.fetch_dir;
+    if (not fetch_dir) {
+        for (auto const& d : arguments.common.just_mr_paths->distdirs) {
+            if (FileSystemManager::IsDirectory(d)) {
+                fetch_dir = std::filesystem::weakly_canonical(
+                    std::filesystem::absolute(d));
+                break;
+            }
+        }
+    }
+    if (not fetch_dir) {
+        std::string s{};
+        for (auto const& d : arguments.common.just_mr_paths->distdirs) {
+            s += "\'";
+            s += d.string();
+            s += "\', ";
+        }
+        if (not s.empty()) {
+            s.erase(s.end() - 1);  // remove trailing ' '
+            s.erase(s.end() - 1);  // remove trailing ','
+        }
+        Logger::Log(LogLevel::Error,
+                    "No directory found to fetch to, considered [{}]",
+                    s);
+        return kExitFetchError;
+    }
+
+    auto repos = (*config)["repositories"];
+    if (not repos.IsNotNull()) {
+        Logger::Log(LogLevel::Error,
+                    "Config: mandatory key \"repositories\" "
+                    "missing");
+        return kExitFetchError;
+    }
+    auto fetch_repos =
+        std::make_shared<SetupRepos>();  // repos to setup and include
+    DefaultReachableRepositories(repos, fetch_repos);
+
+    if (not arguments.setup.sub_all) {
+        auto main = arguments.common.main;
+        if (not main and not fetch_repos->to_include.empty()) {
+            main = *std::min_element(fetch_repos->to_include.begin(),
+                                     fetch_repos->to_include.end());
+        }
+        if (main) {
+            ReachableRepositories(repos, *main, fetch_repos);
+        }
+
+        std::function<bool(std::filesystem::path const&,
+                           std::filesystem::path const&)>
+            is_subpath = [](std::filesystem::path const& path,
+                            std::filesystem::path const& base) {
+                return (std::filesystem::proximate(path, base) == base);
+            };
+
+        // warn if fetch_dir is in invocation workspace
+        if (arguments.common.just_mr_paths->workspace_root and
+            is_subpath(*fetch_dir,
+                       *arguments.common.just_mr_paths->workspace_root)) {
+            auto repo_desc = repos->Get(*main, Expression::none_t{});
+            auto repo = repo_desc->Get("repository", Expression::none_t{});
+            auto repo_path = repo->Get("path", Expression::none_t{});
+            auto repo_type = repo->Get("type", Expression::none_t{});
+            if (repo_path->IsString() and repo_type->IsString() and
+                (repo_type->String() == "file")) {
+                auto repo_path_as_path =
+                    std::filesystem::path(repo_path->String());
+                if (not repo_path_as_path.is_absolute()) {
+                    repo_path_as_path = std::filesystem::weakly_canonical(
+                        std::filesystem::absolute(
+                            arguments.common.just_mr_paths->setup_root /
+                            repo_path_as_path));
+                }
+                // only warn if repo workspace differs to invocation workspace
+                if (not is_subpath(
+                        repo_path_as_path,
+                        *arguments.common.just_mr_paths->workspace_root)) {
+                    Logger::Log(
+                        LogLevel::Warning,
+                        "Writing distributiona files to workspace location {}, "
+                        "which is different to the workspace of the requested "
+                        "main repository {}.",
+                        fetch_dir->string(),
+                        repo_path_as_path.string());
+                }
+            }
+        }
+    }
+
+    Logger::Log(LogLevel::Info, "Fetching to {}", fetch_dir->string());
+    // make sure fetch_dir exists
+    if (not FileSystemManager::CreateDirectory(*fetch_dir)) {
+        Logger::Log(LogLevel::Error,
+                    "Failed to create fetch directory {}",
+                    fetch_dir->string());
+        return kExitFetchError;
+    }
+
+    // gather all repos to be fetched
+    std::vector<ArchiveRepoInfo> repos_to_fetch{};
+    repos_to_fetch.reserve(
+        fetch_repos->to_include.size());  // pre-reserve a maximum size
+    for (auto const& repo_name : fetch_repos->to_include) {
+        auto repo_desc = repos->At(repo_name);
+        if (not repo_desc) {
+            Logger::Log(LogLevel::Error,
+                        "Config: missing config entry for repository {}",
+                        repo_name);
+            return kExitFetchError;
+        }
+        auto repo = repo_desc->get()->At("repository");
+        if (repo) {
+            auto resolved_repo_desc =
+                JustMR::Utils::ResolveRepo(repo->get(), repos);
+            if (not resolved_repo_desc) {
+                Logger::Log(LogLevel::Error,
+                            "Config: found cyclic dependency for "
+                            "repository {}",
+                            repo_name);
+                return kExitFetchError;
+            }
+            // get repo_type
+            auto repo_type = (*resolved_repo_desc)->At("type");
+            if (not repo_type) {
+                Logger::Log(LogLevel::Error,
+                            "Config: mandatory key \"type\" missing "
+                            "for repository {}",
+                            repo_name);
+                return kExitFetchError;
+            }
+            if (not repo_type->get()->IsString()) {
+                Logger::Log(LogLevel::Error,
+                            "Config: unsupported value for key \'type\' for "
+                            "repository {}",
+                            repo_name);
+                return kExitFetchError;
+            }
+            auto repo_type_str = repo_type->get()->String();
+            if (not kCheckoutTypeMap.contains(repo_type_str)) {
+                Logger::Log(LogLevel::Error,
+                            "Unknown repository type {} for {}",
+                            repo_type_str,
+                            repo_name);
+                return kExitFetchError;
+            }
+            // only do work if repo is archive type
+            if (kCheckoutTypeMap.at(repo_type_str) == CheckoutType::Archive) {
+                // check mandatory fields
+                auto repo_desc_content = (*resolved_repo_desc)->At("content");
+                if (not repo_desc_content) {
+                    Logger::Log(LogLevel::Error,
+                                "Mandatory field \"content\" is missing");
+                    return kExitFetchError;
+                }
+                if (not repo_desc_content->get()->IsString()) {
+                    Logger::Log(
+                        LogLevel::Error,
+                        "Unsupported value for mandatory field \'content\'");
+                    return kExitFetchError;
+                }
+                auto repo_desc_fetch = (*resolved_repo_desc)->At("fetch");
+                if (not repo_desc_fetch) {
+                    Logger::Log(LogLevel::Error,
+                                "Mandatory field \"fetch\" is missing");
+                    return kExitFetchError;
+                }
+                if (not repo_desc_fetch->get()->IsString()) {
+                    Logger::Log(LogLevel::Error,
+                                "ArchiveCheckout: Unsupported value for "
+                                "mandatory field \'fetch\'");
+                    return kExitFetchError;
+                }
+                auto repo_desc_subdir =
+                    (*resolved_repo_desc)->Get("subdir", Expression::none_t{});
+                auto subdir =
+                    std::filesystem::path(repo_desc_subdir->IsString()
+                                              ? repo_desc_subdir->String()
+                                              : "")
+                        .lexically_normal();
+                auto repo_desc_distfile =
+                    (*resolved_repo_desc)
+                        ->Get("distfile", Expression::none_t{});
+                auto repo_desc_sha256 =
+                    (*resolved_repo_desc)->Get("sha256", Expression::none_t{});
+                auto repo_desc_sha512 =
+                    (*resolved_repo_desc)->Get("sha512", Expression::none_t{});
+                ArchiveRepoInfo archive_info = {
+                    {
+                        repo_desc_content->get()->String(), /* content */
+                        repo_desc_distfile->IsString()
+                            ? repo_desc_distfile->String()
+                            : "",                         /* distfile */
+                        repo_desc_fetch->get()->String(), /* fetch_url */
+                        repo_desc_sha256->IsString()
+                            ? repo_desc_sha256->String()
+                            : "", /* sha256 */
+                        repo_desc_sha512->IsString()
+                            ? repo_desc_sha512->String()
+                            : "" /* sha512 */
+                    },           /* archive */
+                    repo_type_str,
+                    subdir.empty() ? "." : subdir.string()};
+                // add to list
+                repos_to_fetch.emplace_back(std::move(archive_info));
+            }
+        }
+        else {
+            Logger::Log(LogLevel::Error,
+                        "Config: missing repository description for {}",
+                        repo_name);
+            return kExitFetchError;
+        }
+    }
+    // create async maps
+    auto content_cas_map = CreateContentCASMap(arguments.common.just_mr_paths,
+                                               arguments.common.jobs);
+    auto repo_fetch_map =
+        CreateRepoFetchMap(&content_cas_map, *fetch_dir, arguments.common.jobs);
+    // do the fetch
+    bool failed{false};
+    {
+        TaskSystem ts{arguments.common.jobs};
+        repo_fetch_map.ConsumeAfterKeysReady(
+            &ts,
+            repos_to_fetch,
+            [&failed](auto const& values) {
+                // report any fetch fails
+                for (auto const& val : values) {
+                    if (not *val) {
+                        failed = true;
+                        break;
+                    }
+                }
+            },
+            [&failed](auto const& msg, bool fatal) {
+                Logger::Log(fatal ? LogLevel::Error : LogLevel::Warning,
+                            "While performing just-mr fetch:\n{}",
+                            msg);
+                failed = failed or fatal;
+            });
+    }
+    if (failed) {
+        return kExitFetchError;
+    }
+    return kExitSuccess;
+}
+
 }  // namespace
 
 auto main(int argc, char* argv[]) -> int {
@@ -512,11 +828,13 @@ auto main(int argc, char* argv[]) -> int {
             return kExitSuccess;
         }
 
+        // Run subcommand `fetch`
         if (arguments.cmd == SubCommand::kFetch) {
-            return kExitSuccess;
+            return MultiRepoFetch(config, arguments);
         }
 
-        // default exit code for unknown subcommand
+        // Unknown subcommand should fail
+        Logger::Log(LogLevel::Error, "Unknown subcommand provided.");
         return kExitUnknownCommand;
     } catch (std::exception const& ex) {
         Logger::Log(