product-vamp/src/vamp_zero/database.cpp

#include <vamp_zero/database.hpp>

#include <algorithm>
#include <chrono>
#include <cstdint>
#include <expected>
#include <filesystem>
#include <fstream>
#include <iterator>
#include <sstream>
#include <string>
#include <string_view>
#include <utility>
#include <variant>
#include <vector>

#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/awaitable.hpp>

#include <noto.hpp>

#include <vamp_zero/errors.hpp>
#include <vamp_zero/library.hpp>
#include <vamp_zero/score.hpp>

namespace {

constexpr auto application_id = std::int64_t{0x56414D50};
constexpr auto statement_delimiter = std::string_view{"-- vamp-statement"};

auto trim(std::string value) -> std::string {
  auto first = value.find_first_not_of(" \t\r\n");
  if (first == std::string::npos) {
    return {};
  }
  auto last = value.find_last_not_of(" \t\r\n");
  return value.substr(first, last - first + 1U);
}

auto read_text(std::string const& path) -> std::expected<std::string, modulus::error> {
  auto input = std::ifstream{path, std::ios::binary};
  if (!input) {
    return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::NOT_FOUND, "failed to open migration: " + path));
  }
  auto text = std::string{std::istreambuf_iterator<std::char_t>{input}, std::istreambuf_iterator<std::char_t>{}};
  if (!input.good() && !input.eof()) {
    return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::UNAVAILABLE, "failed to read migration: " + path));
  }
  return text;
}

auto execute_statements(noto::connection& connection, std::vector<std::string> statements) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  for (auto& statement : statements) {
    auto result = co_await connection.async_execute(std::move(statement));
    if (!result) {
      co_return std::unexpected(result.error());
    }
  }
  co_return std::expected<void, modulus::error>{};
}

auto connection_config(vamp_zero::database_config const& config) -> noto::connection_config {
  auto result = config.path == ":memory:" ? noto::memory_connection_config("product-vamp") : noto::file_connection_config(config.path, "product-vamp");
  result.busy_timeout = std::chrono::seconds{5};
  result.foreign_keys = true;
  result.pragmas = {
    {.name = "synchronous", .value = std::string{"NORMAL"}},
    {.name = "temp_store", .value = std::string{"MEMORY"}},
  };
  return result;
}

auto read_migration(vamp_zero::database_config const& config, std::string filename) -> std::expected<std::vector<std::string>, modulus::error> {
  auto migration_text = read_text(config.migrations_path + "/" + filename);
  if (!migration_text) {
    return std::unexpected(migration_text.error());
  }
  return vamp_zero::split_migration_statements(*migration_text);
}

auto text_value(noto::sqlite_value const& value) -> std::string {
  auto text = std::get_if<std::string>(&value.value);
  return text == nullptr ? std::string{} : *text;
}

auto path_parts(std::filesystem::path const& path) -> std::vector<std::string> {
  auto parts = std::vector<std::string>{};
  for (auto const& part : path) {
    parts.push_back(part.string());
  }
  return parts;
}

auto logical_practice_exercise_id(std::string const& type_id, std::string const& lesson_id, std::string const& source_path, std::string const& fallback) -> std::string {
  auto parts = path_parts(std::filesystem::path{source_path});
  for (auto index = std::size_t{}; index + 3U < parts.size(); ++index) {
    if (parts[index] == "practice" && parts[index + 1U] == type_id && parts[index + 2U] == lesson_id) {
      return type_id + "-" + lesson_id + "-" + parts[index + 3U];
    }
  }
  auto stem = std::filesystem::path{source_path}.parent_path().filename().string();
  if (stem.empty()) {
    stem = fallback;
  }
  return type_id + "-" + lesson_id + "-" + stem;
}

auto now_ms() -> std::int64_t {
  return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
}

auto execute_delete_stale_catalog_rows(noto::connection& connection, std::string_view table, std::vector<std::string> const& ids) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  auto sql = std::string{"DELETE FROM "};
  sql += table;
  sql += " WHERE catalog_managed = 1";
  auto params = std::vector<noto::sqlite_value>{};
  params.reserve(ids.size());
  if (!ids.empty()) {
    sql += " AND id NOT IN (";
    for (auto index = std::size_t{}; index < ids.size(); ++index) {
      if (index != 0U) {
        sql += ", ";
      }
      sql += "?";
      sql += std::to_string(index + 1U);
      params.emplace_back(ids[index]);
    }
    sql += ")";
  }
  auto deleted = co_await connection.async_execute(std::move(sql), std::move(params));
  if (!deleted) {
    co_return std::unexpected(deleted.error());
  }
  co_return std::expected<void, modulus::error>{};
}

}

auto vamp_zero::split_migration_statements(std::string const& input) -> std::expected<std::vector<std::string>, modulus::error> {
  auto statements = std::vector<std::string>{};
  auto current = std::string{};
  auto stream = std::istringstream{input};
  auto line = std::string{};
  while (std::getline(stream, line)) {
    if (trim(line) == statement_delimiter) {
      current = trim(std::move(current));
      if (current.empty()) {
        return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::INVALID_ARGUMENT, "migration contains an empty statement"));
      }
      statements.push_back(std::move(current));
      current.clear();
      continue;
    }
    current += line;
    current.push_back('\n');
  }
  current = trim(std::move(current));
  if (!current.empty()) {
    statements.push_back(std::move(current));
  }
  if (statements.empty()) {
    return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::INVALID_ARGUMENT, "migration must contain at least one statement"));
  }
  for (auto const& statement : statements) {
    if (statement.empty() || statement.back() != ';') {
      return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::INVALID_ARGUMENT, "every migration statement must end with a semicolon"));
    }
  }
  return statements;
}

auto vamp_zero::async_migrate_database(boost::asio::any_io_executor executor, vamp_zero::database_config config) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  if (config.path.empty() || config.migrations_path.empty()) {
    co_return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::INVALID_ARGUMENT, "database paths must not be empty"));
  }
  if (config.path != ":memory:") {
    auto parent = std::filesystem::path{config.path}.parent_path();
    if (!parent.empty()) {
      auto error = std::error_code{};
      std::filesystem::create_directories(parent, error);
      if (error) {
        co_return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::UNAVAILABLE, "failed to create database directory: " + error.message()));
      }
    }
  }

  auto connection = noto::connection{executor, connection_config(config)};
  auto connected = co_await connection.async_connect();
  if (!connected) {
    co_return std::unexpected(connected.error());
  }
  if (config.path != ":memory:") {
    auto journal = co_await connection.async_journal_mode("WAL");
    if (!journal || *journal != "wal") {
      auto close = co_await connection.async_close();
      if (!close) {
        modulus::log_error("database close failed after WAL error: " + close.error().describe());
      }
      co_return std::unexpected(journal ? vamp_zero::make_error(vamp_zero::error_code::PERSISTENCE, "SQLite did not enter WAL mode") : journal.error());
    }
  }
  auto current_application_id = co_await connection.async_application_id();
  if (!current_application_id) {
    co_return std::unexpected(current_application_id.error());
  }
  if (*current_application_id != 0 && *current_application_id != application_id) {
    co_return std::unexpected(vamp_zero::make_error(vamp_zero::error_code::PERSISTENCE, "database application id does not belong to Product Vamp"));
  }
  if (*current_application_id == 0) {
    auto set = co_await connection.async_set_application_id(application_id);
    if (!set) {
      co_return std::unexpected(set.error());
    }
  }

  auto foundation_statements = read_migration(config, "001_foundation.sql");
  if (!foundation_statements) {
    co_return std::unexpected(foundation_statements.error());
  }
  auto import_statements = read_migration(config, "002_song_import_metadata.sql");
  if (!import_statements) {
    co_return std::unexpected(import_statements.error());
  }
  auto practice_statements = read_migration(config, "003_practice_catalog.sql");
  if (!practice_statements) {
    co_return std::unexpected(practice_statements.error());
  }
  auto practice_hierarchy_statements = read_migration(config, "004_practice_hierarchy.sql");
  if (!practice_hierarchy_statements) {
    co_return std::unexpected(practice_hierarchy_statements.error());
  }
  auto catalog_managed_statements = read_migration(config, "005_catalog_managed_rows.sql");
  if (!catalog_managed_statements) {
    co_return std::unexpected(catalog_managed_statements.error());
  }
  auto migrator = noto::migrator{connection};
  auto migrations = std::vector<noto::migration>{};
  auto foundation = noto::migration{};
  foundation.id = "001_foundation";
  foundation.description = "initial Product Vamp appliance tables";
  foundation.checksum = "001_foundation_v1";
  foundation.up = [statements = std::move(*foundation_statements)](noto::connection& connection) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
    co_return co_await execute_statements(connection, statements);
  };
  migrations.push_back(std::move(foundation));
  auto import = noto::migration{};
  import.id = "002_song_import_metadata";
  import.description = "song import metadata";
  import.checksum = "002_song_import_metadata_v1";
  import.up = [statements = std::move(*import_statements)](noto::connection& connection) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
    co_return co_await execute_statements(connection, statements);
  };
  migrations.push_back(std::move(import));
  auto practice = noto::migration{};
  practice.id = "003_practice_catalog";
  practice.description = "practice exercise catalog";
  practice.checksum = "003_practice_catalog_v1";
  practice.up = [statements = std::move(*practice_statements)](noto::connection& connection) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
    co_return co_await execute_statements(connection, statements);
  };
  migrations.push_back(std::move(practice));
  auto practice_hierarchy = noto::migration{};
  practice_hierarchy.id = "004_practice_hierarchy";
  practice_hierarchy.description = "practice type and lesson hierarchy";
  practice_hierarchy.checksum = "004_practice_hierarchy_v1";
  practice_hierarchy.up = [statements = std::move(*practice_hierarchy_statements)](noto::connection& connection) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
    co_return co_await execute_statements(connection, statements);
  };
  migrations.push_back(std::move(practice_hierarchy));
  auto catalog_managed = noto::migration{};
  catalog_managed.id = "005_catalog_managed_rows";
  catalog_managed.description = "mark rows managed by bundled catalog import";
  catalog_managed.checksum = "005_catalog_managed_rows_v1";
  catalog_managed.up = [statements = std::move(*catalog_managed_statements)](noto::connection& connection) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
    co_return co_await execute_statements(connection, statements);
  };
  migrations.push_back(std::move(catalog_managed));

  auto migrated = co_await migrator.async_migrate(std::move(migrations));
  if (!migrated) {
    co_return std::unexpected(migrated.error());
  }
  auto version = co_await connection.async_set_user_version(5);
  if (!version) {
    co_return std::unexpected(version.error());
  }
  co_return co_await connection.async_close();
}

auto vamp_zero::async_import_catalogs(boost::asio::any_io_executor executor, vamp_zero::database_config config, std::string assets_path) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  auto performance = vamp_zero::load_performance_library(assets_path);
  if (!performance) {
    co_return std::unexpected(performance.error());
  }
  auto practice = vamp_zero::load_practice_library(assets_path);
  if (!practice) {
    co_return std::unexpected(practice.error());
  }
  auto connection = noto::connection{executor, connection_config(config)};
  auto connected = co_await connection.async_connect();
  if (!connected) {
    co_return std::unexpected(connected.error());
  }
  auto begin = co_await connection.async_execute("BEGIN IMMEDIATE");
  if (!begin) {
    co_return std::unexpected(begin.error());
  }
  auto imported = std::expected<void, modulus::error>{};
  auto timestamp = now_ms();
  auto song_ids = std::vector<std::string>{};
  song_ids.reserve(performance->parts.size());
  for (auto const& part : performance->parts) {
    if (!imported) {
      break;
    }
    song_ids.push_back(part.id);
    auto params = std::vector<noto::sqlite_value>{
      noto::sqlite_value{part.id},
      noto::sqlite_value{part.title},
      noto::sqlite_value{part.composer},
      noto::sqlite_value{part.source_json},
      noto::sqlite_value{timestamp},
      noto::sqlite_value{timestamp},
      noto::sqlite_value{part.part},
      noto::sqlite_value{part.source_path},
      noto::sqlite_value{part.key_signature},
    };
    auto inserted = co_await connection.async_execute(
      R"sql(
INSERT INTO songs(id, title, composer, source_json, created_at_ms, updated_at_ms, part, source_path, key_signature, catalog_managed)
VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 1)
ON CONFLICT(id) DO UPDATE SET
  title = excluded.title,
  composer = excluded.composer,
  source_json = excluded.source_json,
  part = excluded.part,
  source_path = excluded.source_path,
  key_signature = excluded.key_signature,
  catalog_managed = 1,
  updated_at_ms = CASE
    WHEN songs.title != excluded.title
      OR songs.composer != excluded.composer
      OR songs.source_json != excluded.source_json
      OR songs.part != excluded.part
      OR songs.source_path != excluded.source_path
      OR songs.key_signature != excluded.key_signature
    THEN excluded.updated_at_ms
    ELSE songs.updated_at_ms
  END
)sql",
      std::move(params));
    if (!inserted) {
      imported = std::unexpected(inserted.error());
    }
  }
  auto practice_ids = std::vector<std::string>{};
  for (auto const& exercise : practice->exercises) {
    if (!imported) {
      break;
    }
    for (auto const& variant : exercise.variants) {
      practice_ids.push_back(variant.id);
      auto params = std::vector<noto::sqlite_value>{
        noto::sqlite_value{variant.id},
        noto::sqlite_value{exercise.practice_type_id},
        noto::sqlite_value{exercise.practice_type_title},
        noto::sqlite_value{exercise.lesson_group_id},
        noto::sqlite_value{exercise.lesson_group_title},
        noto::sqlite_value{exercise.title},
        noto::sqlite_value{exercise.composer},
        noto::sqlite_value{variant.key_signature},
        noto::sqlite_value{variant.source_json},
        noto::sqlite_value{variant.source_path},
        noto::sqlite_value{timestamp},
        noto::sqlite_value{timestamp},
      };
      auto inserted = co_await connection.async_execute(
        R"sql(
INSERT INTO practice_exercises(id, practice_type_id, practice_type_title, lesson_group_id, lesson_group_title, title, composer, key_signature, source_json, source_path, created_at_ms, updated_at_ms, catalog_managed)
VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, 1)
ON CONFLICT(id) DO UPDATE SET
  practice_type_id = excluded.practice_type_id,
  practice_type_title = excluded.practice_type_title,
  lesson_group_id = excluded.lesson_group_id,
  lesson_group_title = excluded.lesson_group_title,
  title = excluded.title,
  composer = excluded.composer,
  key_signature = excluded.key_signature,
  source_json = excluded.source_json,
  source_path = excluded.source_path,
  catalog_managed = 1,
  updated_at_ms = CASE
    WHEN practice_exercises.practice_type_id != excluded.practice_type_id
      OR practice_exercises.practice_type_title != excluded.practice_type_title
      OR practice_exercises.lesson_group_id != excluded.lesson_group_id
      OR practice_exercises.lesson_group_title != excluded.lesson_group_title
      OR practice_exercises.title != excluded.title
      OR practice_exercises.composer != excluded.composer
      OR practice_exercises.key_signature != excluded.key_signature
      OR practice_exercises.source_json != excluded.source_json
      OR practice_exercises.source_path != excluded.source_path
    THEN excluded.updated_at_ms
    ELSE practice_exercises.updated_at_ms
  END
)sql",
        std::move(params));
      if (!inserted) {
        imported = std::unexpected(inserted.error());
        break;
      }
    }
  }
  if (imported) {
    auto deleted = co_await execute_delete_stale_catalog_rows(connection, "songs", song_ids);
    if (!deleted) {
      imported = std::unexpected(deleted.error());
    }
  }
  if (imported) {
    auto deleted = co_await execute_delete_stale_catalog_rows(connection, "practice_exercises", practice_ids);
    if (!deleted) {
      imported = std::unexpected(deleted.error());
    }
  }
  if (imported) {
    auto committed = co_await connection.async_execute("COMMIT");
    if (!committed) {
      imported = std::unexpected(committed.error());
    }
  } else {
    static_cast<void>(co_await connection.async_execute("ROLLBACK"));
  }
  if (!imported) {
    auto close = co_await connection.async_close();
    if (!close) {
      modulus::log_error("database close failed after import error: " + close.error().describe());
    }
    co_return std::unexpected(imported.error());
  }
  co_return co_await connection.async_close();
}

auto vamp_zero::async_load_performance_library(boost::asio::any_io_executor executor, vamp_zero::database_config config) -> boost::asio::awaitable<std::expected<vamp_zero::library_state, modulus::error>> {
  auto connection = noto::connection{executor, connection_config(config)};
  auto connected = co_await connection.async_connect();
  if (!connected) {
    co_return std::unexpected(connected.error());
  }
  auto selected = co_await connection.async_execute("SELECT id, title, composer, part, source_path, source_json, key_signature FROM songs ORDER BY title COLLATE NOCASE, part COLLATE NOCASE");
  if (!selected) {
    auto close = co_await connection.async_close();
    if (!close) {
      modulus::log_error("database close failed after load error: " + close.error().describe());
    }
    co_return std::unexpected(selected.error());
  }
  auto parts = std::vector<vamp_zero::library_part>{};
  for (auto const& row : selected->rows) {
    if (row.size() < 7U) {
      continue;
    }
    auto source_json = text_value(row[5]);
    auto parsed = vamp_zero::parse_score_document(source_json, text_value(row[4]));
    if (!parsed) {
      auto close = co_await connection.async_close();
      if (!close) {
        modulus::log_error("database close failed after parse error: " + close.error().describe());
      }
      co_return std::unexpected(parsed.error());
    }
    parts.push_back({
      .id = text_value(row[0]),
      .title = text_value(row[1]),
      .composer = text_value(row[2]),
      .key_signature = text_value(row[6]),
      .part = text_value(row[3]),
      .source_path = text_value(row[4]),
      .source_json = std::move(source_json),
      .measure_count = parsed->measures.size(),
      .total_pages = vamp_zero::score_page_count(*parsed),
    });
  }
  auto closed = co_await connection.async_close();
  if (!closed) {
    co_return std::unexpected(closed.error());
  }
  co_return vamp_zero::group_library_parts(std::move(parts));
}

auto vamp_zero::async_load_practice_library(boost::asio::any_io_executor executor, vamp_zero::database_config config) -> boost::asio::awaitable<std::expected<vamp_zero::practice_library, modulus::error>> {
  auto connection = noto::connection{executor, connection_config(config)};
  auto connected = co_await connection.async_connect();
  if (!connected) {
    co_return std::unexpected(connected.error());
  }
  auto selected = co_await connection.async_execute("SELECT id, practice_type_id, practice_type_title, lesson_group_id, lesson_group_title, title, composer, key_signature, source_path, source_json FROM practice_exercises ORDER BY practice_type_id COLLATE NOCASE, lesson_group_id COLLATE NOCASE, source_path COLLATE NOCASE, title COLLATE NOCASE, id COLLATE NOCASE");
  if (!selected) {
    auto close = co_await connection.async_close();
    if (!close) {
      modulus::log_error("database close failed after practice load error: " + close.error().describe());
    }
    co_return std::unexpected(selected.error());
  }
  auto exercises = std::vector<vamp_zero::practice_exercise>{};
  for (auto const& row : selected->rows) {
    if (row.size() < 10U) {
      continue;
    }
    auto source_json = text_value(row[9]);
    auto parsed = vamp_zero::parse_score_document(source_json, text_value(row[8]));
    if (!parsed) {
      auto close = co_await connection.async_close();
      if (!close) {
        modulus::log_error("database close failed after practice parse error: " + close.error().describe());
      }
      co_return std::unexpected(parsed.error());
    }
    auto type_id = text_value(row[1]);
    auto lesson_id = text_value(row[3]);
    auto source_path = text_value(row[8]);
    auto key_signature = text_value(row[7]).empty() ? std::string{"C"} : text_value(row[7]);
    auto variant = vamp_zero::practice_exercise::score_variant{
      .id = text_value(row[0]),
      .key_signature = key_signature,
      .source_path = source_path,
      .source_json = source_json,
      .measure_count = parsed->measures.size(),
      .total_pages = vamp_zero::score_page_count(*parsed),
    };
    exercises.push_back({
      .id = logical_practice_exercise_id(type_id, lesson_id, source_path, text_value(row[5])),
      .practice_type_id = std::move(type_id),
      .practice_type_title = text_value(row[2]),
      .lesson_group_id = std::move(lesson_id),
      .lesson_group_title = text_value(row[4]),
      .title = text_value(row[5]),
      .composer = text_value(row[6]),
      .key_signature = variant.key_signature,
      .source_path = variant.source_path,
      .source_json = variant.source_json,
      .measure_count = variant.measure_count,
      .total_pages = variant.total_pages,
      .variants = {std::move(variant)},
    });
  }
  auto closed = co_await connection.async_close();
  if (!closed) {
    co_return std::unexpected(closed.error());
  }
  co_return vamp_zero::group_practice_exercises(std::move(exercises));
}