product-vamp/src/vamp_zero/app.cpp

#include <vamp_zero/app.hpp>

#include <csignal>
#include <chrono>
#include <cstdint>
#include <expected>
#include <filesystem>
#include <fstream>
#include <functional>
#include <future>
#include <iostream>
#include <memory>
#include <string>
#include <string_view>
#include <thread>
#include <utility>

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/use_future.hpp>
#include <boost/json/object.hpp>
#include <boost/json/serialize.hpp>

#include <ligo.hpp>
#include <modulus.hpp>

#include <vamp_zero/config.hpp>
#include <vamp_zero/audio.hpp>
#include <vamp_zero/database.hpp>
#include <vamp_zero/errors.hpp>
#include <vamp_zero/library.hpp>
#include <vamp_zero/page_control.hpp>
#include <vamp_zero/portal.hpp>
#include <vamp_zero/ui.hpp>

namespace {

auto invalid(std::string message) -> modulus::error {
  return vamp_zero::make_error(vamp_zero::error_code::INVALID_ARGUMENT, std::move(message));
}

auto usage() -> std::string {
  return "usage: vamp-zero [--config PATH] [--check-config] [--health-check] [--version] [--once] [--ensure-database]";
}

auto bind_port(vamp_zero::config const& config) -> std::uint16_t {
  auto endpoint = ligo::http::server_endpoint_from_config(config.control.endpoint);
  if (!endpoint) {
    return 8080;
  }
  return endpoint->port == 0 ? 8080 : endpoint->port;
}

auto write_text_atomic(std::filesystem::path const& destination, std::string const& text) -> std::expected<void, modulus::error> {
  auto error = std::error_code{};
  if (!destination.parent_path().empty()) {
    std::filesystem::create_directories(destination.parent_path(), error);
    if (error) {
      return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::UNAVAILABLE, "failed to create runtime directory: " + error.message()));
    }
  }
  auto temporary = destination;
  temporary += ".tmp";
  auto output = std::ofstream{temporary, std::ios::binary | std::ios::trunc};
  if (!output) {
    return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::UNAVAILABLE, "failed to create heartbeat: " + temporary.string()));
  }
  output << text;
  output.close();
  if (!output) {
    return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::UNAVAILABLE, "failed to write heartbeat: " + temporary.string()));
  }
  std::filesystem::rename(temporary, destination, error);
  if (error) {
    std::filesystem::remove(temporary, error);
    return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::UNAVAILABLE, "failed to publish heartbeat: " + error.message()));
  }
  return {};
}

auto write_heartbeat(vamp_zero::config const& config) -> std::expected<void, modulus::error> {
  auto heartbeat = boost::json::object{};
  heartbeat["version"] = VAMP_VERSION;
  heartbeat["git_sha"] = VAMP_GIT_SHA;
  heartbeat["status"] = "ok";
  heartbeat["endpoint"] = "http://" + config.control.zero_ip + ":" + std::to_string(bind_port(config));
  return write_text_atomic(config.runtime.heartbeat_path, boost::json::serialize(heartbeat));
}

auto ensure_database(vamp_zero::config const& config) -> std::expected<void, modulus::error> {
  auto io = boost::asio::io_context{};
  auto future = boost::asio::co_spawn(io, vamp_zero::async_ensure_database(io.get_executor(), config.database), boost::asio::use_future);
  io.run();
  return future.get();
}

auto make_server_endpoint(vamp_zero::config const& config) -> std::expected<ligo::http::endpoint, modulus::error> {
  auto endpoint = ligo::http::server_endpoint_from_config(config.control.endpoint);
  if (!endpoint) {
    return std::unexpected(endpoint.error());
  }
  if (endpoint->port == 0) {
    endpoint->port = 8080;
  }
  return endpoint;
}

auto import_library(vamp_zero::config const& config) -> std::expected<void, modulus::error> {
  auto io = boost::asio::io_context{};
  auto future = boost::asio::co_spawn(io, vamp_zero::async_import_catalogs(io.get_executor(), config.database, config.runtime.assets_path), boost::asio::use_future);
  io.run();
  return future.get();
}

auto load_initial_state(vamp_zero::config const& config) -> std::expected<vamp_zero::app_state, modulus::error> {
  auto io = boost::asio::io_context{};
  auto performance_future = boost::asio::co_spawn(io, vamp_zero::async_load_performance_library(io.get_executor(), config.database), boost::asio::use_future);
  auto practice_future = boost::asio::co_spawn(io, vamp_zero::async_load_practice_library(io.get_executor(), config.database), boost::asio::use_future);
  io.run();
  auto performance = performance_future.get();
  if (!performance) {
    return std::unexpected(performance.error());
  }
  auto practice = practice_future.get();
  if (!practice) {
    return std::unexpected(practice.error());
  }
  auto state = vamp_zero::app_state{
    .performance = std::move(*performance),
    .practice = std::move(*practice),
    .audio = {
      .track_enabled = config.audio.track_enabled,
      .metronome_enabled = config.audio.metronome_enabled,
      .metronome_bpm = config.audio.metronome_bpm,
      .metronome_min_bpm = config.audio.metronome_min_bpm,
      .metronome_max_bpm = config.audio.metronome_max_bpm,
      .metronome_step = config.audio.metronome_step,
    },
  };
  if (auto valid = vamp_zero::validate_app_state(state); !valid) {
    return std::unexpected(valid.error());
  }
  return state;
}

auto run_headless(vamp_zero::config const& config, vamp_zero::control_state& state, vamp_zero::audio_controller* audio) -> std::expected<void, modulus::error> {
  auto io = boost::asio::io_context{};
  auto endpoint = make_server_endpoint(config);
  if (!endpoint) {
    return std::unexpected(endpoint.error());
  }
  auto server = ligo::http::server{io, *endpoint, vamp_zero::make_router(state), ligo::http::server_options_from_config(config.control.endpoint)};
  auto started = server.start();
  if (!started) {
    return std::unexpected(started.error());
  }
  if (audio != nullptr) {
    state.set_observer([audio](vamp_zero::app_state app) {
      audio->sync(app);
    });
    audio->sync(state.snapshot());
  }
  auto heartbeat = write_heartbeat(config);
  if (!heartbeat) {
    server.stop();
    return std::unexpected(heartbeat.error());
  }
  auto signals = boost::asio::signal_set{io, SIGINT, SIGTERM};
  signals.async_wait([&](auto const&, std::int32_t) {
    server.stop();
    io.stop();
  });
  std::cout << "vamp-zero listening on " << server.selected_port() << '\n';
  io.run();
  return {};
}

auto run_display(vamp_zero::config const& config, vamp_zero::control_state& state, vamp_zero::audio_controller* audio) -> std::expected<void, modulus::error> {
  auto io = boost::asio::io_context{};
  auto endpoint = make_server_endpoint(config);
  if (!endpoint) {
    return std::unexpected(endpoint.error());
  }
  auto ui = vamp_zero::zero_ui{io.get_executor(), config.display, [&state](vamp_protocol::page_command command) {
    static_cast<void>(state.apply(command));
  }};
  state.set_observer([&ui, &config, audio](vamp_zero::app_state app) {
    if (audio != nullptr) {
      audio->sync(app);
    }
    auto status = ui.publish(vamp_zero::make_ui_state(config, std::move(app)));
    if (!status) {
      modulus::log_error("failed to publish UI state: " + status.error().describe());
    }
  });
  if (audio != nullptr) {
    audio->sync(state.snapshot());
  }
  auto published = ui.publish(vamp_zero::make_ui_state(config, state.snapshot()));
  if (!published) {
    return std::unexpected(published.error());
  }
  auto server = ligo::http::server{io, *endpoint, vamp_zero::make_router(state), ligo::http::server_options_from_config(config.control.endpoint)};
  auto started = server.start();
  if (!started) {
    return std::unexpected(started.error());
  }
  auto heartbeat = write_heartbeat(config);
  if (!heartbeat) {
    server.stop();
    return std::unexpected(heartbeat.error());
  }
  auto connection_timer = boost::asio::steady_timer{io};
  auto schedule_connection_refresh = std::function<void()>{};
  schedule_connection_refresh = [&] {
    connection_timer.expires_after(std::chrono::milliseconds{500});
    connection_timer.async_wait([&](auto const& error) {
      if (error) {
        return;
      }
      static_cast<void>(state.refresh_connection());
      schedule_connection_refresh();
    });
  };
  schedule_connection_refresh();
  auto signals = boost::asio::signal_set{io, SIGINT, SIGTERM};
  signals.async_wait([&](auto const&, std::int32_t) {
    connection_timer.cancel();
    server.stop();
    ui.stop();
    io.stop();
  });
  auto io_thread = std::thread{[&io] {
    io.run();
  }};
  auto ui_status = ui.run();
  server.stop();
  connection_timer.cancel();
  io.stop();
  if (io_thread.joinable()) {
    io_thread.join();
  }
  return ui_status;
}

}

auto vamp_zero::parse_command_line(std::int32_t argc, std::char_t** argv) -> std::expected<vamp_zero::command_line, modulus::error> {
  if (argc < 1 || argv == nullptr) {
    return std::unexpected(invalid("process arguments are invalid"));
  }
  auto result = vamp_zero::command_line{};
  auto require_value = [&](std::int32_t& index, std::string_view option) -> std::expected<std::string_view, modulus::error> {
    ++index;
    if (index >= argc) {
      return std::unexpected(invalid(std::string{option} + " requires a value"));
    }
    auto value = std::string_view{argv[index]};
    if (value.empty()) {
      return std::unexpected(invalid(std::string{option} + " value must not be empty"));
    }
    return value;
  };
  for (auto index = std::int32_t{1}; index < argc; ++index) {
    auto argument = std::string_view{argv[index]};
    if (argument == "--config") {
      auto value = require_value(index, "--config");
      if (!value) {
        return std::unexpected(value.error());
      }
      result.config_path = *value;
    } else if (argument == "--check-config") {
      result.check_config = true;
    } else if (argument == "--health-check") {
      result.health_check = true;
    } else if (argument == "--version") {
      result.print_version = true;
    } else if (argument == "--once") {
      result.exit_after_startup = true;
    } else if (argument == "--ensure-database") {
      result.ensure_database = true;
    } else if (argument == "--help" || argument == "-h") {
      return std::unexpected(invalid(usage()));
    } else {
      return std::unexpected(invalid("unknown argument: " + std::string{argument}));
    }
  }
  auto command_count = static_cast<std::uint32_t>(result.check_config ? 1 : 0)
    + static_cast<std::uint32_t>(result.health_check ? 1 : 0)
    + static_cast<std::uint32_t>(result.print_version ? 1 : 0)
    + static_cast<std::uint32_t>(result.exit_after_startup ? 1 : 0)
    + static_cast<std::uint32_t>(result.ensure_database ? 1 : 0);
  if (command_count > 1U) {
    return std::unexpected(invalid("only one command flag may be selected"));
  }
  return result;
}

auto vamp_zero::run(vamp_zero::command_line command) -> std::expected<void, modulus::error> {
  if (command.print_version) {
    std::cout << VAMP_VERSION << " " << VAMP_GIT_SHA << '\n';
    return {};
  }
  auto config = vamp_zero::load_config(command.config_path);
  if (!config) {
    return std::unexpected(config.error());
  }
  modulus::configure(config->modulus);
  if (command.check_config) {
    std::cout << "ok\n";
    return {};
  }
  if (command.health_check) {
    std::cout << "ok\n";
    return {};
  }
  if (command.ensure_database) {
    return ensure_database(*config);
  }
  if (command.exit_after_startup || config->runtime.exit_after_startup) {
    std::cout << "ok\n";
    return {};
  }

  auto ensured = ensure_database(*config);
  if (!ensured) {
    return ensured;
  }
  auto imported = import_library(*config);
  if (!imported) {
    return imported;
  }
  auto initial = load_initial_state(*config);
  if (!initial) {
    return std::unexpected(initial.error());
  }
  auto state = vamp_zero::control_state{*config, std::move(*initial)};
  auto audio_controller = std::unique_ptr<vamp_zero::audio_controller>{};
  auto audio = vamp_zero::create_audio_controller(config->audio);
  if (!audio) {
    modulus::log_error("audio unavailable: " + audio.error().describe());
  } else {
    audio_controller = std::move(*audio);
  }
  if (config->runtime.headless) {
    return run_headless(*config, state, audio_controller.get());
  }
  return run_display(*config, state, audio_controller.get());
}