#include <test.hpp>

#include <array>
#include <chrono>
#include <cstdint>
#include <expected>
#include <filesystem>
#include <fstream>
#include <limits>
#include <memory>
#include <stdexcept>
#include <string>
#include <string_view>
#include <sstream>
#include <utility>
#include <variant>
#include <vector>

#include <boost/asio/awaitable.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_future.hpp>
#include <boost/json/parse.hpp>

#include <sqlite3.h>

#include <modulus.hpp>

#include <noto/config.hpp>
#include <noto/connection.hpp>
#include <noto/detail/errors.hpp>
#include <noto/errors.hpp>
#include <noto/migration.hpp>
#include <noto/schema.hpp>
#include <noto/transaction.hpp>

BOOST_AUTO_TEST_CASE(noto_asserts_follow_build_type) {
#if defined(NOTO_EXPECT_NDEBUG) && !defined(NDEBUG)
  BOOST_FAIL("release-style builds must define NDEBUG");
#endif
#if defined(NOTO_EXPECT_DEBUG_ASSERTS) && defined(NDEBUG)
  BOOST_FAIL("debug builds must not define NDEBUG");
#endif
}


namespace {

template <typename AWAITABLE>
auto run(boost::asio::io_context& io, AWAITABLE&& awaitable) {
  auto future = boost::asio::co_spawn(io, std::forward<AWAITABLE>(awaitable), boost::asio::use_future);
  io.run();
  io.restart();
  return future.get();
}

auto temp_database_path(std::string_view name) -> std::string {
  auto path = std::filesystem::temp_directory_path() / (std::string{"noto_"} + std::string{name} + "_" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()) + ".sqlite3");
  return path.string();
}

auto remove_database_files(std::string const& path) -> void {
  std::filesystem::remove(path);
  std::filesystem::remove(path + "-wal");
  std::filesystem::remove(path + "-shm");
  std::filesystem::remove(path + "-journal");
}

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 int_value(noto::sqlite_value const& value) -> std::int64_t {
  auto integer = std::get_if<std::int64_t>(&value.value);
  return integer == nullptr ? 0 : *integer;
}

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

auto sqlite_exec_ok(sqlite3* database, std::string const& sql) -> std::bool_t {
  auto* error = static_cast<std::char_t*>(nullptr);
  auto code = sqlite3_exec(database, sql.c_str(), nullptr, nullptr, &error);
  sqlite3_free(error);
  return code == SQLITE_OK;
}

auto create_sqlite_database(std::string const& path, std::string const& sql) -> std::bool_t {
  auto* database = static_cast<sqlite3*>(nullptr);
  if (sqlite3_open_v2(path.c_str(), &database, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX, nullptr) != SQLITE_OK) {
    if (database != nullptr) {
      sqlite3_close(database);
    }
    return false;
  }
  auto ok = sqlite_exec_ok(database, sql);
  return sqlite3_close(database) == SQLITE_OK && ok;
}

auto write_file(std::string const& path, std::string_view text) -> void {
  auto output = std::ofstream{
    path,
    std::ios::binary | std::ios::trunc,
  };
  output << text;
}

auto read_file(std::string const& path) -> std::string {
  auto input = std::ifstream{
    path,
    std::ios::binary,
  };
  auto output = std::ostringstream{};
  output << input.rdbuf();
  return output.str();
}

auto sqlite_scalar_int64(sqlite3* database, std::string const& sql) -> std::int64_t {
  auto* statement = static_cast<sqlite3_stmt*>(nullptr);
  if (sqlite3_prepare_v2(database, sql.c_str(), -1, &statement, nullptr) != SQLITE_OK) {
    return 0;
  }
  auto guard = std::unique_ptr<sqlite3_stmt, decltype(&sqlite3_finalize)>{
    statement,
    sqlite3_finalize,
  };
  if (sqlite3_step(statement) != SQLITE_ROW) {
    return 0;
  }
  return sqlite3_column_int64(statement, 0);
}

}

BOOST_AUTO_TEST_CASE(noto_status_defaults_to_ok) {
  auto status = std::expected<void, modulus::error>{};
  BOOST_TEST(static_cast<std::bool_t>(status));
}

BOOST_AUTO_TEST_CASE(noto_result_carries_failure_status) {
  auto ok = std::expected<void, modulus::error>{};
  BOOST_TEST(static_cast<std::bool_t>(ok.has_value()));

  auto result = noto::detail::fail<std::int32_t>(noto::detail::error_invalid_argument, "bad value");
  BOOST_REQUIRE(!result);
  BOOST_TEST(static_cast<std::int32_t>(result.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
  BOOST_TEST(result.error().message() == "bad value");
}

BOOST_AUTO_TEST_CASE(noto_status_maps_sqlite_errors) {
  BOOST_TEST(static_cast<std::int32_t>(noto::error_from(SQLITE_CONSTRAINT).code()) == static_cast<std::int32_t>(noto::detail::error_constraint_violation));
  BOOST_TEST(static_cast<std::int32_t>(noto::error_from(SQLITE_ERROR).code()) == static_cast<std::int32_t>(noto::detail::error_syntax_error));
  BOOST_TEST(static_cast<std::int32_t>(noto::error_from(SQLITE_BUSY).code()) == static_cast<std::int32_t>(noto::detail::error_unavailable));
  BOOST_TEST(static_cast<std::int32_t>(noto::error_from(SQLITE_CORRUPT).code()) == static_cast<std::int32_t>(noto::detail::error_data_loss));
}

BOOST_AUTO_TEST_CASE(noto_config_validates_and_parses_json) {
  auto config = noto::connection_config{};
  BOOST_TEST(static_cast<std::bool_t>(noto::validate(config)));

  config.filename = "";
  BOOST_TEST(!static_cast<std::bool_t>(noto::validate(config)));

  auto value = boost::json::parse(R"json({
    "databases": [
      {
        "name": "main",
        "mode": "memory",
        "busy_timeout_ms": 250,
        "foreign_keys": false,
        "pragmas": {
          "user_version": 7,
          "journal_mode": "MEMORY"
        }
      }
    ]
  })json");
  auto parsed = noto::parse_database_configs(value);
  BOOST_REQUIRE(parsed);
  BOOST_REQUIRE(parsed->size() == 1U);
  BOOST_TEST(parsed->at(0).name == "main");
  BOOST_TEST(parsed->at(0).filename == ":memory:");
  BOOST_TEST(static_cast<std::int32_t>(parsed->at(0).mode) == static_cast<std::int32_t>(noto::open_mode::MEMORY));
  BOOST_TEST(parsed->at(0).busy_timeout == std::chrono::milliseconds{250});
  BOOST_TEST(!static_cast<std::bool_t>(parsed->at(0).foreign_keys));
  BOOST_TEST(parsed->at(0).pragmas.size() == 2U);
  auto round_trip = noto::parse_database_configs(noto::to_json(*parsed));
  BOOST_REQUIRE(round_trip);
  BOOST_REQUIRE(round_trip->size() == 1U);
  BOOST_TEST(round_trip->at(0).name == "main");
  BOOST_TEST(static_cast<std::int32_t>(round_trip->at(0).mode) == static_cast<std::int32_t>(noto::open_mode::MEMORY));
  BOOST_TEST(round_trip->at(0).pragmas.size() == 2U);

  auto bad = noto::parse_database_configs(boost::json::parse(R"json({"databases":[{"name":"bad","filename":":memory:","mode":"server"}]})json"));
  BOOST_TEST(!static_cast<std::bool_t>(bad.has_value()));

  auto missing_file = noto::parse_database_configs(boost::json::parse(R"json({"databases":[{"name":"bad"}]})json"));
  BOOST_TEST(!static_cast<std::bool_t>(missing_file.has_value()));

  config.filename = ":memory:";
  config.busy_timeout = std::chrono::milliseconds{static_cast<std::int64_t>(std::numeric_limits<std::int32_t>::max()) + 1};
  BOOST_TEST(!static_cast<std::bool_t>(noto::validate(config)));

  auto oversized_timeout_json = std::string{R"json({"databases":[{"name":"bad","filename":":memory:","busy_timeout_ms":)json"}
    + std::to_string(static_cast<std::int64_t>(std::numeric_limits<std::int32_t>::max()) + 1)
    + R"json(}]})json";
  auto oversized_timeout = noto::parse_database_configs(boost::json::parse(oversized_timeout_json));
  BOOST_TEST(!static_cast<std::bool_t>(oversized_timeout.has_value()));

  auto unknown_root = noto::parse_database_configs(boost::json::parse(R"json({"databases":[],"databasez":[]})json"));
  BOOST_REQUIRE(!unknown_root);
  BOOST_TEST(unknown_root.error().describe().find("config root has unknown key: databasez") != std::string::npos);

  auto unknown_database = noto::parse_database_configs(boost::json::parse(R"json({"databases":[{"name":"bad","filename":":memory:","busy_timout_ms":250}]})json"));
  BOOST_REQUIRE(!unknown_database);
  BOOST_TEST(unknown_database.error().describe().find("database has unknown key: busy_timout_ms") != std::string::npos);
}

BOOST_AUTO_TEST_CASE(noto_schema_quotes_identifiers_and_builds_sqlite_table_sql) {
  auto quoted = noto::quote_identifier("with\"quote");
  BOOST_REQUIRE(quoted);
  BOOST_TEST(*quoted == "\"with\"\"quote\"");

  auto email_type = noto::column_type::varchar(255);
  BOOST_REQUIRE(email_type);
  auto sql = noto::create_table_sql({
    .name = {
      .name = "users",
    },
    .columns = {
      {
        .name = "id",
        .type = noto::column_type::integer(),
        .nullable = false,
        .auto_increment = true,
        .primary_key = true,
      },
      {
        .name = "email",
        .type = *email_type,
        .nullable = false,
      },
      {
        .name = "created_at",
        .type = noto::column_type::timestamp(),
        .nullable = false,
        .default_value = noto::default_value::current_timestamp(),
      },
    },
    .options = {
      .strict = true,
    },
  });

  BOOST_REQUIRE(sql);
  BOOST_TEST(*sql == "CREATE TABLE IF NOT EXISTS \"users\" (\"id\" INTEGER PRIMARY KEY AUTOINCREMENT, \"email\" TEXT NOT NULL, \"created_at\" TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP) STRICT");
}

BOOST_AUTO_TEST_CASE(noto_schema_builds_constraints_generated_columns_and_options) {
  auto expression = noto::trusted_sql_fragment::unsafe("lower(\"email\")");
  BOOST_REQUIRE(expression);
  auto check = noto::trusted_sql_fragment::unsafe("\"email\" <> ''");
  BOOST_REQUIRE(check);
  auto sql = noto::create_table_sql({
    .name = {
      .name = "accounts",
    },
    .columns = {
      {
        .name = "tenant_id",
        .type = noto::column_type::integer(),
        .nullable = false,
      },
      {
        .name = "id",
        .type = noto::column_type::integer(),
        .nullable = false,
      },
      {
        .name = "email",
        .type = noto::column_type::text(),
        .nullable = false,
      },
      {
        .name = "email_key",
        .type = noto::column_type::text(),
        .nullable = false,
        .generated = noto::generated_column_definition{
          .expression = *expression,
          .storage = noto::generated_column_storage::STORED,
        },
      },
    },
    .primary_key = {
      "tenant_id",
      "id",
    },
    .unique_constraints = {
      {
        .name = "uq_accounts_email",
        .columns = {
          "tenant_id",
          "email_key",
        },
      },
    },
    .check_constraints = {
      {
        .name = "ck_accounts_email",
        .expression = *check,
      },
    },
    .foreign_keys = {
      {
        .name = "fk_accounts_tenant",
        .columns = {
          "tenant_id",
        },
        .referenced_table = {
          .name = "tenants",
        },
        .referenced_columns = {
          "id",
        },
        .on_delete = noto::reference_action::CASCADE,
      },
    },
    .options = {
      .without_rowid = true,
    },
  });

  BOOST_REQUIRE(sql);
  BOOST_TEST(*sql == "CREATE TABLE IF NOT EXISTS \"accounts\" (\"tenant_id\" INTEGER NOT NULL, \"id\" INTEGER NOT NULL, \"email\" TEXT NOT NULL, \"email_key\" TEXT GENERATED ALWAYS AS (lower(\"email\")) STORED NOT NULL, PRIMARY KEY (\"tenant_id\", \"id\"), CONSTRAINT \"uq_accounts_email\" UNIQUE (\"tenant_id\", \"email_key\"), CONSTRAINT \"ck_accounts_email\" CHECK (\"email\" <> ''), CONSTRAINT \"fk_accounts_tenant\" FOREIGN KEY (\"tenant_id\") REFERENCES \"tenants\" (\"id\") ON DELETE CASCADE) WITHOUT ROWID");
}

BOOST_AUTO_TEST_CASE(noto_connection_executes_queries_and_prepared_statements) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("test"),
  };

  auto connected = run(io, connection.async_connect());
  BOOST_TEST(static_cast<std::bool_t>(connected));
  BOOST_TEST(static_cast<std::int32_t>(connection.state()) == static_cast<std::int32_t>(noto::connection_state::CONNECTED));

  auto created = run(io, connection.async_execute("CREATE TABLE items(id INTEGER PRIMARY KEY, value TEXT NOT NULL)"));
  BOOST_REQUIRE(created);

  auto statement = run(io, connection.async_prepare("INSERT INTO items(value) VALUES(?1)"));
  BOOST_REQUIRE(statement);
  auto inserted = run(io, statement->async_execute({
    noto::sqlite_value{"alpha"},
  }));
  BOOST_REQUIRE(inserted);
  BOOST_TEST(inserted->changes == 1);

  auto selected = run(io, connection.async_execute("SELECT value FROM items WHERE id = ?1", {
    noto::sqlite_value{inserted->last_insert_rowid},
  }));
  BOOST_REQUIRE(selected);
  BOOST_REQUIRE(selected->rows.size() == 1U);
  BOOST_TEST(text_value(selected->rows.at(0).at(0)) == "alpha");

  auto finalized = run(io, statement->async_finalize());
  BOOST_TEST(static_cast<std::bool_t>(finalized));
  auto closed = run(io, connection.async_close());
  BOOST_TEST(static_cast<std::bool_t>(closed));
}

BOOST_AUTO_TEST_CASE(noto_close_with_live_statement_is_recoverable) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("live_statement"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));

  auto statement = run(io, connection.async_prepare("SELECT 1"));
  BOOST_REQUIRE(statement);
  auto blocked = run(io, connection.async_close());
  BOOST_TEST(!static_cast<std::bool_t>(blocked));
  BOOST_TEST(static_cast<std::int32_t>(blocked.error().code()) == static_cast<std::int32_t>(noto::detail::error_unavailable));
  BOOST_TEST(static_cast<std::int32_t>(connection.state()) == static_cast<std::int32_t>(noto::connection_state::CONNECTED));

  BOOST_TEST(static_cast<std::bool_t>(run(io, statement->async_finalize())));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_close_rolls_back_active_transaction) {
  auto io = boost::asio::io_context{};
  auto path = temp_database_path("close_active_transaction");
  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));

  {
    auto transaction = run(io, connection.async_begin_transaction());
    BOOST_REQUIRE(transaction);
    BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('transient')")));
    BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
  }

  auto reopened = noto::connection{
    io.get_executor(),
    noto::file_connection_config(path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, reopened.async_connect())));
  auto count = run(io, reopened.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);
  BOOST_TEST(static_cast<std::bool_t>(run(io, reopened.async_close())));
  remove_database_files(path);
}

BOOST_AUTO_TEST_CASE(noto_async_methods_capture_state_from_temporary_wrappers) {
  auto io = boost::asio::io_context{};

  auto connect_future = boost::asio::co_spawn(
    io,
    noto::connection{
      io.get_executor(),
      noto::memory_connection_config("temporary_connection"),
    }.async_connect(),
    boost::asio::use_future
  );
  io.run();
  io.restart();
  BOOST_TEST(static_cast<std::bool_t>(connect_future.get()));

  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("temporary_wrappers"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));

  auto statement = run(io, connection.async_prepare("SELECT 1"));
  BOOST_REQUIRE(statement);
  auto finalize_future = boost::asio::co_spawn(io, noto::statement{std::move(*statement)}.async_finalize(), boost::asio::use_future);
  io.run();
  io.restart();
  BOOST_TEST(static_cast<std::bool_t>(finalize_future.get()));

  auto rolled_back = run(io, connection.async_begin_transaction());
  BOOST_REQUIRE(rolled_back);
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('rolled back')")));
  auto rollback_future = boost::asio::co_spawn(io, noto::transaction{std::move(*rolled_back)}.async_rollback(), boost::asio::use_future);
  io.run();
  io.restart();
  BOOST_TEST(static_cast<std::bool_t>(rollback_future.get()));

  auto committed = run(io, connection.async_begin_transaction());
  BOOST_REQUIRE(committed);
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('committed')")));
  auto commit_future = boost::asio::co_spawn(io, noto::transaction{std::move(*committed)}.async_commit(), boost::asio::use_future);
  io.run();
  io.restart();
  BOOST_TEST(static_cast<std::bool_t>(commit_future.get()));

  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 1);
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_detached_transaction_rolls_back_when_commit_cannot_start) {
  auto io = boost::asio::io_context{};
  auto export_path = temp_database_path("detached_transaction_export");
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("detached_transaction"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value INTEGER NOT NULL)")));

  auto transaction = run(io, connection.async_begin_transaction());
  BOOST_REQUIRE(transaction);
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES(1), (2), (3), (4), (5)")));

  auto options = noto::export_options{};
  options.rows_per_yield = 1;
  auto export_future = boost::asio::co_spawn(io, connection.async_export_sql_to_file(export_path, options), boost::asio::use_future);
  auto commit_future = boost::asio::co_spawn(io, noto::transaction{std::move(*transaction)}.async_commit(), boost::asio::use_future);
  io.run();
  io.restart();

  BOOST_TEST(static_cast<std::bool_t>(export_future.get()));
  auto commit = commit_future.get();
  BOOST_TEST(static_cast<std::int32_t>(commit.error().code()) == static_cast<std::int32_t>(noto::detail::error_unavailable));

  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
  std::filesystem::remove(export_path);
}

BOOST_AUTO_TEST_CASE(noto_statement_destructor_cleanup_allows_close) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("statement_destructor"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));

  {
    auto statement = run(io, connection.async_prepare("SELECT 1"));
    BOOST_REQUIRE(statement);
  }

  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_transactions_commit_and_rollback) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config(),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));

  {
    auto transaction = run(io, connection.async_begin_transaction());
    BOOST_REQUIRE(transaction);
    BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('rolled back')")));
    BOOST_TEST(static_cast<std::bool_t>(run(io, transaction->async_rollback())));
  }
  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);

  {
    auto transaction = run(io, connection.async_begin_transaction());
    BOOST_REQUIRE(transaction);
    BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('committed')")));
    BOOST_TEST(static_cast<std::bool_t>(run(io, transaction->async_commit())));
  }
  count = run(io, connection.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 1);
}

BOOST_AUTO_TEST_CASE(noto_failed_commit_remains_rollbackable) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("failed_commit"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE parents(id INTEGER PRIMARY KEY)")));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE children(parent_id INTEGER NOT NULL, FOREIGN KEY(parent_id) REFERENCES parents(id) DEFERRABLE INITIALLY DEFERRED)")));

  auto transaction = run(io, connection.async_begin_transaction(noto::transaction_mode::DEFERRED));
  BOOST_REQUIRE(transaction);
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO children(parent_id) VALUES(42)")));

  auto committed = run(io, transaction->async_commit());
  BOOST_TEST(!static_cast<std::bool_t>(committed));
  BOOST_TEST(static_cast<std::int32_t>(committed.error().code()) == static_cast<std::int32_t>(noto::detail::error_constraint_violation));
  BOOST_TEST(static_cast<std::bool_t>(transaction->active()));
  BOOST_TEST(static_cast<std::bool_t>(run(io, transaction->async_rollback())));

  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM children"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);

  auto next = run(io, connection.async_begin_transaction());
  BOOST_REQUIRE(next);
  BOOST_TEST(static_cast<std::bool_t>(run(io, next->async_rollback())));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_detached_failed_commit_rolls_back) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("detached_failed_commit"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE parents(id INTEGER PRIMARY KEY)")));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE children(parent_id INTEGER NOT NULL, FOREIGN KEY(parent_id) REFERENCES parents(id) DEFERRABLE INITIALLY DEFERRED)")));

  auto transaction = run(io, connection.async_begin_transaction(noto::transaction_mode::DEFERRED));
  BOOST_REQUIRE(transaction);
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO children(parent_id) VALUES(42)")));

  auto commit_future = boost::asio::co_spawn(io, noto::transaction{std::move(*transaction)}.async_commit(), boost::asio::use_future);
  io.run();
  io.restart();
  auto committed = commit_future.get();
  BOOST_TEST(!static_cast<std::bool_t>(committed));
  BOOST_TEST(static_cast<std::int32_t>(committed.error().code()) == static_cast<std::int32_t>(noto::detail::error_constraint_violation));

  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM children"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);

  auto next = run(io, connection.async_begin_transaction());
  BOOST_REQUIRE(next);
  BOOST_TEST(static_cast<std::bool_t>(run(io, next->async_rollback())));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_abandoned_transaction_rolls_back_before_next_operation) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config(),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));

  {
    auto transaction = run(io, connection.async_begin_transaction());
    BOOST_REQUIRE(transaction);
    BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('abandoned')")));
  }

  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_transaction_callback_exceptions_return_status_and_rollback) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config(),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));

  auto status = run(io, connection.async_with_transaction(noto::transaction_mode::IMMEDIATE, [](noto::connection& active) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
    auto inserted = co_await active.async_execute("INSERT INTO items(value) VALUES('temporary')");
    if (!inserted) {
      co_return std::unexpected(inserted.error());
    }
    throw std::runtime_error{"boom"};
  }));
  BOOST_TEST(!static_cast<std::bool_t>(status));
  BOOST_TEST(static_cast<std::int32_t>(status.error().code()) == static_cast<std::int32_t>(noto::detail::error_internal));

  auto count = run(io, connection.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 0);
}

BOOST_AUTO_TEST_CASE(noto_execute_options_limit_collected_results) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("limits"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('one'), ('two')")));

  auto row_options = noto::execute_options{};
  row_options.max_rows = 1;
  auto too_many_rows = run(io, connection.async_execute("SELECT value FROM items ORDER BY value", {}, row_options));
  BOOST_REQUIRE(!too_many_rows);
  BOOST_TEST(static_cast<std::int32_t>(too_many_rows.error().code()) == static_cast<std::int32_t>(noto::detail::error_resource_exhausted));

  auto byte_options = noto::execute_options{};
  byte_options.max_result_bytes = 4;
  auto too_many_bytes = run(io, connection.async_execute("SELECT value FROM items WHERE value = 'one'", {}, byte_options));
  BOOST_REQUIRE(!too_many_bytes);
  BOOST_TEST(static_cast<std::int32_t>(too_many_bytes.error().code()) == static_cast<std::int32_t>(noto::detail::error_resource_exhausted));

  auto statement = run(io, connection.async_prepare("SELECT value FROM items ORDER BY value"));
  BOOST_REQUIRE(statement);
  auto statement_options = noto::execute_options{};
  statement_options.max_rows = 1;
  auto statement_limit = run(io, statement->async_execute({}, statement_options));
  BOOST_REQUIRE(!statement_limit);
  BOOST_TEST(static_cast<std::int32_t>(statement_limit.error().code()) == static_cast<std::int32_t>(noto::detail::error_resource_exhausted));
  auto after_limit = run(io, connection.async_execute("INSERT INTO items(value) VALUES('three')"));
  BOOST_REQUIRE(after_limit);
  BOOST_TEST(static_cast<std::bool_t>(run(io, statement->async_finalize())));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}

BOOST_AUTO_TEST_CASE(noto_migrations_apply_validate_and_rollback) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config(),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));

  auto migrations = std::vector<noto::migration>{
    {
      .id = "001_create_table",
      .description = "create items",
      .checksum = "checksum-001",
      .up = [](noto::connection& active) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
        co_return co_await execute_status(active, "CREATE TABLE items(id INTEGER PRIMARY KEY, value TEXT NOT NULL)");
      },
      .down = [](noto::connection& active) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
        co_return co_await execute_status(active, "DROP TABLE IF EXISTS items");
      },
    },
  };
  auto migrator = noto::migrator{connection};
  BOOST_TEST(static_cast<std::bool_t>(run(io, migrator.async_migrate(migrations))));
  BOOST_TEST(static_cast<std::bool_t>(run(io, migrator.async_migrate(migrations))));

  auto changed = migrations;
  changed.at(0).checksum = "changed";
  auto mismatch = run(io, migrator.async_migrate(changed));
  BOOST_TEST(!static_cast<std::bool_t>(mismatch));
  BOOST_TEST(static_cast<std::int32_t>(mismatch.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));

  auto records = run(io, migrator.async_applied_migrations());
  BOOST_REQUIRE(records);
  BOOST_TEST(records->size() == 1U);
  BOOST_TEST(records->at(0).id == "001_create_table");

  BOOST_TEST(static_cast<std::bool_t>(run(io, migrator.async_rollback(migrations))));
  auto missing = run(io, connection.async_execute("SELECT name FROM sqlite_schema WHERE type = 'table' AND name = 'items'"));
  BOOST_REQUIRE(missing);
  BOOST_TEST(static_cast<std::bool_t>(missing->rows.empty()));
}

BOOST_AUTO_TEST_CASE(noto_backup_export_integrity_and_pragmas_work) {
  auto io = boost::asio::io_context{};
  auto source_path = temp_database_path("source");
  auto backup_path = temp_database_path("backup");
  auto failed_backup_path = temp_database_path("failed_backup");
  auto oversized_retry_path = temp_database_path("oversized_retry");
  auto export_path = temp_database_path("export_sql");

  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(source_path),
  };
  auto startup_config = connection.config();
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_set_user_version(12))));
  auto version = run(io, connection.async_user_version());
  BOOST_REQUIRE(version);
  BOOST_TEST(*version == 12);
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_set_busy_timeout(std::chrono::milliseconds{37}))));
  auto after_timeout_config = connection.config();
  BOOST_TEST(after_timeout_config.busy_timeout == startup_config.busy_timeout);
  auto busy_timeout = run(io, connection.async_busy_timeout());
  BOOST_REQUIRE(busy_timeout);
  BOOST_TEST(*busy_timeout == std::chrono::milliseconds{37});

  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(id INTEGER PRIMARY KEY, value TEXT NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('one'), ('two')")));
  auto integrity = run(io, connection.async_integrity_check());
  BOOST_REQUIRE(integrity);
  BOOST_REQUIRE(integrity->size() == 1U);
  BOOST_TEST(integrity->at(0) == "ok");

  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_backup_to_file(backup_path))));
  auto exported = run(io, connection.async_export_sql());
  BOOST_REQUIRE(exported);
  BOOST_TEST(exported->find("CREATE TABLE items") != std::string::npos);
  auto limited_options = noto::export_options{};
  limited_options.max_output_bytes = 16;
  auto limited_export = run(io, connection.async_export_sql(limited_options));
  BOOST_REQUIRE(!limited_export);
  BOOST_TEST(static_cast<std::int32_t>(limited_export.error().code()) == static_cast<std::int32_t>(noto::detail::error_resource_exhausted));

  auto* preserved = static_cast<sqlite3*>(nullptr);
  BOOST_REQUIRE(sqlite3_open_v2(failed_backup_path.c_str(), &preserved, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX, nullptr) == SQLITE_OK);
  BOOST_REQUIRE(sqlite_exec_ok(preserved, "CREATE TABLE preserved(value INTEGER); INSERT INTO preserved(value) VALUES(42)"));
  sqlite3_close(preserved);
  auto bad_backup_options = noto::backup_options{};
  bad_backup_options.source_name = "missing";
  auto failed_backup = run(io, connection.async_backup_to_file(failed_backup_path, bad_backup_options));
  BOOST_TEST(!static_cast<std::bool_t>(failed_backup));
  preserved = nullptr;
  BOOST_REQUIRE(sqlite3_open_v2(failed_backup_path.c_str(), &preserved, SQLITE_OPEN_READONLY | SQLITE_OPEN_FULLMUTEX, nullptr) == SQLITE_OK);
  BOOST_TEST(sqlite_scalar_int64(preserved, "SELECT value FROM preserved") == 42);
  sqlite3_close(preserved);

  auto oversized_retry = noto::backup_options{};
  oversized_retry.retry_delay = std::chrono::milliseconds{static_cast<std::int64_t>(std::numeric_limits<std::int32_t>::max()) + 1};
  auto failed_delay = run(io, connection.async_backup_to_file(oversized_retry_path, oversized_retry));
  BOOST_TEST(static_cast<std::int32_t>(failed_delay.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));

  oversized_retry = noto::backup_options{};
  oversized_retry.retry_timeout = std::chrono::milliseconds{static_cast<std::int64_t>(std::numeric_limits<std::int32_t>::max()) + 1};
  auto failed_timeout = run(io, connection.async_backup_to_file(oversized_retry_path, oversized_retry));
  BOOST_TEST(static_cast<std::int32_t>(failed_timeout.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));

  write_file(export_path, "preserved export");
  auto limited_file_options = noto::export_options{};
  limited_file_options.max_output_bytes = 16;
  auto limited_file_export = run(io, connection.async_export_sql_to_file(export_path, limited_file_options));
  BOOST_REQUIRE(!limited_file_export);
  BOOST_TEST(static_cast<std::int32_t>(limited_file_export.error().code()) == static_cast<std::int32_t>(noto::detail::error_resource_exhausted));
  BOOST_TEST(read_file(export_path) == "preserved export");

  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_export_sql_to_file(export_path))));
  BOOST_TEST(static_cast<std::bool_t>(std::filesystem::exists(export_path)));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));

  auto restored = noto::connection{
    io.get_executor(),
    noto::file_connection_config(backup_path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, restored.async_connect())));
  auto count = run(io, restored.async_execute("SELECT COUNT(*) FROM items"));
  BOOST_REQUIRE(count);
  BOOST_TEST(int_value(count->rows.at(0).at(0)) == 2);
  BOOST_TEST(static_cast<std::bool_t>(run(io, restored.async_close())));

  std::filesystem::remove(source_path);
  std::filesystem::remove(backup_path);
  std::filesystem::remove(failed_backup_path);
  std::filesystem::remove(oversized_retry_path);
  std::filesystem::remove(export_path);
}

BOOST_AUTO_TEST_CASE(noto_backup_and_export_reject_live_database_destinations) {
  auto io = boost::asio::io_context{};
  auto source_path = temp_database_path("protected_destination_source");
  auto link_path = temp_database_path("protected_destination_link");
  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(source_path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('safe')")));

  auto self_backup = run(io, connection.async_backup_to_file(source_path));
  BOOST_TEST(static_cast<std::int32_t>(self_backup.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
  auto self_export = run(io, connection.async_export_sql_to_file(source_path));
  BOOST_TEST(static_cast<std::int32_t>(self_export.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));

  for (auto suffix : std::array<std::string_view, 3>{"-wal", "-shm", "-journal"}) {
    auto backup = run(io, connection.async_backup_to_file(source_path + std::string{suffix}));
    BOOST_TEST(static_cast<std::int32_t>(backup.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
    auto exported = run(io, connection.async_export_sql_to_file(source_path + std::string{suffix}));
    BOOST_TEST(static_cast<std::int32_t>(exported.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
  }

  auto error = std::error_code{};
  std::filesystem::create_symlink(source_path, link_path, error);
  if (!error) {
    auto link_backup = run(io, connection.async_backup_to_file(link_path));
    BOOST_TEST(static_cast<std::int32_t>(link_backup.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
    auto link_export = run(io, connection.async_export_sql_to_file(link_path));
    BOOST_TEST(static_cast<std::int32_t>(link_export.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
  }

  auto rows = run(io, connection.async_execute("SELECT value FROM items"));
  BOOST_REQUIRE(rows);
  BOOST_REQUIRE(rows->rows.size() == 1U);
  BOOST_TEST(text_value(rows->rows.at(0).at(0)) == "safe");
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
  std::filesystem::remove(link_path);
  remove_database_files(source_path);
}

BOOST_AUTO_TEST_CASE(noto_restore_to_file_is_atomic_and_reopens_connection) {
  auto io = boost::asio::io_context{};
  auto source_path = temp_database_path("restore_source");
  auto target_path = temp_database_path("restore_target");

  BOOST_REQUIRE(create_sqlite_database(source_path, "CREATE TABLE items(value TEXT NOT NULL); INSERT INTO items(value) VALUES('new')"));

  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(target_path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  auto journal = run(io, connection.async_journal_mode("WAL"));
  BOOST_REQUIRE(journal);
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('old')")));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_set_busy_timeout(std::chrono::milliseconds{37}))));
  write_file(target_path + "-journal", "stale rollback journal");
  BOOST_TEST(static_cast<std::bool_t>(std::filesystem::exists(target_path + "-journal")));

  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_restore_from_file(source_path))));
  BOOST_TEST(static_cast<std::int32_t>(connection.state()) == static_cast<std::int32_t>(noto::connection_state::CONNECTED));
  auto timeout = run(io, connection.async_busy_timeout());
  BOOST_REQUIRE(timeout);
  BOOST_TEST(*timeout == std::chrono::milliseconds{37});

  auto rows = run(io, connection.async_execute("SELECT value FROM items"));
  BOOST_REQUIRE(rows);
  BOOST_REQUIRE(rows->rows.size() == 1U);
  BOOST_TEST(text_value(rows->rows.at(0).at(0)) == "new");
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
  BOOST_TEST(!static_cast<std::bool_t>(std::filesystem::exists(target_path + "-wal")));
  BOOST_TEST(!static_cast<std::bool_t>(std::filesystem::exists(target_path + "-shm")));
  BOOST_TEST(!static_cast<std::bool_t>(std::filesystem::exists(target_path + "-journal")));

  auto* restored = static_cast<sqlite3*>(nullptr);
  BOOST_REQUIRE(sqlite3_open_v2(target_path.c_str(), &restored, SQLITE_OPEN_READONLY | SQLITE_OPEN_FULLMUTEX, nullptr) == SQLITE_OK);
  BOOST_TEST(sqlite_scalar_int64(restored, "SELECT COUNT(*) FROM items WHERE value = 'new'") == 1);
  sqlite3_close(restored);

  remove_database_files(source_path);
  remove_database_files(target_path);
}

BOOST_AUTO_TEST_CASE(noto_failed_restore_leaves_existing_database_intact) {
  auto io = boost::asio::io_context{};
  auto source_path = temp_database_path("restore_failure_source");
  auto target_path = temp_database_path("restore_failure_target");

  BOOST_REQUIRE(create_sqlite_database(source_path, "CREATE TABLE items(value TEXT NOT NULL); INSERT INTO items(value) VALUES('new')"));
  BOOST_REQUIRE(create_sqlite_database(target_path, "CREATE TABLE items(value TEXT NOT NULL); INSERT INTO items(value) VALUES('old')"));

  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(target_path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  auto options = noto::backup_options{};
  options.source_name = "missing";
  auto restored = run(io, connection.async_restore_from_file(source_path, options));
  BOOST_TEST(!static_cast<std::bool_t>(restored));

  auto rows = run(io, connection.async_execute("SELECT value FROM items"));
  BOOST_REQUIRE(rows);
  BOOST_REQUIRE(rows->rows.size() == 1U);
  BOOST_TEST(text_value(rows->rows.at(0).at(0)) == "old");
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));

  remove_database_files(source_path);
  remove_database_files(target_path);
}

BOOST_AUTO_TEST_CASE(noto_restore_rejects_unsupported_targets_and_active_state) {
  auto io = boost::asio::io_context{};
  auto source_path = temp_database_path("restore_reject_source");
  auto target_path = temp_database_path("restore_reject_target");

  BOOST_REQUIRE(create_sqlite_database(source_path, "CREATE TABLE items(value TEXT NOT NULL); INSERT INTO items(value) VALUES('new')"));
  BOOST_REQUIRE(create_sqlite_database(target_path, "CREATE TABLE items(value TEXT NOT NULL); INSERT INTO items(value) VALUES('old')"));

  auto memory = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("restore_reject_memory"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, memory.async_connect())));
  auto memory_restore = run(io, memory.async_restore_from_file(source_path));
  BOOST_TEST(static_cast<std::int32_t>(memory_restore.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
  BOOST_TEST(static_cast<std::bool_t>(run(io, memory.async_close())));

  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(target_path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));

  auto options = noto::backup_options{};
  options.destination_name = "attached";
  auto non_main = run(io, connection.async_restore_from_file(source_path, options));
  BOOST_TEST(static_cast<std::int32_t>(non_main.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));

  auto transaction = run(io, connection.async_begin_transaction());
  BOOST_REQUIRE(transaction);
  auto active_transaction = run(io, connection.async_restore_from_file(source_path));
  BOOST_TEST(static_cast<std::int32_t>(active_transaction.error().code()) == static_cast<std::int32_t>(noto::detail::error_unavailable));
  BOOST_TEST(static_cast<std::bool_t>(run(io, transaction->async_rollback())));

  auto statement = run(io, connection.async_prepare("SELECT value FROM items"));
  BOOST_REQUIRE(statement);
  auto live_statement = run(io, connection.async_restore_from_file(source_path));
  BOOST_TEST(static_cast<std::int32_t>(live_statement.error().code()) == static_cast<std::int32_t>(noto::detail::error_unavailable));
  BOOST_TEST(static_cast<std::bool_t>(run(io, statement->async_finalize())));
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));

  remove_database_files(source_path);
  remove_database_files(target_path);
}

BOOST_AUTO_TEST_CASE(noto_backup_lock_timeout_and_statement_finalize_retry_work) {
  auto io = boost::asio::io_context{};
  auto source_path = temp_database_path("locked_source");
  auto backup_path = temp_database_path("locked_backup");

  auto connection = noto::connection{
    io.get_executor(),
    noto::file_connection_config(source_path),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('one')")));
  auto statement = run(io, connection.async_prepare("SELECT value FROM items"));
  BOOST_REQUIRE(statement);

  auto* blocker = static_cast<sqlite3*>(nullptr);
  BOOST_REQUIRE(sqlite3_open_v2(source_path.c_str(), &blocker, SQLITE_OPEN_READWRITE | SQLITE_OPEN_FULLMUTEX, nullptr) == SQLITE_OK);
  BOOST_REQUIRE(sqlite_exec_ok(blocker, "BEGIN EXCLUSIVE"));

  auto options = noto::backup_options{};
  options.pages_per_step = 1;
  options.retry_delay = std::chrono::milliseconds{1};
  options.retry_timeout = std::chrono::milliseconds{25};

  auto backup_future = boost::asio::co_spawn(io, connection.async_backup_to_file(backup_path, options), boost::asio::use_future);
  auto finalize_future = boost::asio::co_spawn(io, statement->async_finalize(), boost::asio::use_future);
  io.run();
  io.restart();

  auto backup = backup_future.get();
  auto finalized_while_busy = finalize_future.get();
  BOOST_TEST(static_cast<std::int32_t>(backup.error().code()) == static_cast<std::int32_t>(noto::detail::error_timeout));

  BOOST_TEST(static_cast<std::bool_t>(sqlite_exec_ok(blocker, "ROLLBACK")));
  sqlite3_close(blocker);
  if (!finalized_while_busy) {
    BOOST_TEST(static_cast<std::int32_t>(finalized_while_busy.error().code()) == static_cast<std::int32_t>(noto::detail::error_unavailable));
    BOOST_TEST(static_cast<std::bool_t>(run(io, statement->async_finalize())));
  }
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));

  std::filesystem::remove(source_path);
  std::filesystem::remove(backup_path);
}

BOOST_AUTO_TEST_CASE(noto_export_preserves_nul_text_and_defers_triggers_until_after_data) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("export_exact"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE audit(count INTEGER NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO audit(count) VALUES(0)")));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(value TEXT NOT NULL)")));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TRIGGER items_ai AFTER INSERT ON items BEGIN UPDATE audit SET count = count + 1; END")));

  auto insert = run(io, connection.async_prepare("INSERT INTO items(value) VALUES(?1)"));
  BOOST_REQUIRE(insert);
  auto embedded = std::string{
    "a\0b",
    3,
  };
  BOOST_REQUIRE(run(io, insert->async_execute({
    noto::sqlite_value{embedded},
  })));
  BOOST_TEST(static_cast<std::bool_t>(run(io, insert->async_finalize())));

  auto exported = run(io, connection.async_export_sql());
  BOOST_REQUIRE(exported);
  BOOST_TEST(exported->find("CAST(X'610062' AS TEXT)") != std::string::npos);
  BOOST_TEST(exported->find("INSERT INTO \"items\"") < exported->find("CREATE TRIGGER"));

  auto* restored = static_cast<sqlite3*>(nullptr);
  BOOST_REQUIRE(sqlite3_open(":memory:", &restored) == SQLITE_OK);
  BOOST_REQUIRE(sqlite_exec_ok(restored, *exported));

  auto* statement = static_cast<sqlite3_stmt*>(nullptr);
  BOOST_REQUIRE(sqlite3_prepare_v2(restored, "SELECT value FROM items", -1, &statement, nullptr) == SQLITE_OK);
  BOOST_REQUIRE(sqlite3_step(statement) == SQLITE_ROW);
  auto const* text = reinterpret_cast<std::char_t const*>(sqlite3_column_text(statement, 0));
  auto bytes = sqlite3_column_bytes(statement, 0);
  auto restored_text = std::string{
    text,
    static_cast<std::size_t>(bytes),
  };
  sqlite3_finalize(statement);
  BOOST_TEST(restored_text == embedded);

  statement = nullptr;
  BOOST_REQUIRE(sqlite3_prepare_v2(restored, "SELECT count FROM audit", -1, &statement, nullptr) == SQLITE_OK);
  BOOST_REQUIRE(sqlite3_step(statement) == SQLITE_ROW);
  BOOST_TEST(sqlite3_column_int64(statement, 0) == 1);
  sqlite3_finalize(statement);
  sqlite3_close(restored);
}

BOOST_AUTO_TEST_CASE(noto_export_preserves_autoincrement_sequence_and_generated_columns) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("export_sequence_generated"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  BOOST_REQUIRE(run(io, connection.async_execute("CREATE TABLE items(id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT NOT NULL, value_size INTEGER GENERATED ALWAYS AS (length(value)) STORED)")));
  BOOST_REQUIRE(run(io, connection.async_execute("INSERT INTO items(value) VALUES('one'), ('two')")));
  BOOST_REQUIRE(run(io, connection.async_execute("DELETE FROM items WHERE value = 'two'")));

  auto exported = run(io, connection.async_export_sql());
  BOOST_REQUIRE(exported);
  BOOST_TEST(exported->find("sqlite_sequence") != std::string::npos);
  BOOST_TEST(exported->find("value_size") != std::string::npos);
  BOOST_TEST(exported->find("INSERT INTO \"items\"(\"id\", \"value\")") != std::string::npos);

  auto* restored = static_cast<sqlite3*>(nullptr);
  BOOST_REQUIRE(sqlite3_open(":memory:", &restored) == SQLITE_OK);
  BOOST_REQUIRE(sqlite_exec_ok(restored, *exported));
  BOOST_REQUIRE(sqlite_exec_ok(restored, "INSERT INTO items(value) VALUES('three')"));
  BOOST_TEST(sqlite_scalar_int64(restored, "SELECT id FROM items WHERE value = 'three'") == 3);
  BOOST_TEST(sqlite_scalar_int64(restored, "SELECT value_size FROM items WHERE value = 'three'") == 5);
  sqlite3_close(restored);
}

BOOST_AUTO_TEST_CASE(noto_export_rejects_virtual_tables) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("export_virtual"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));
  auto created = run(io, connection.async_execute("CREATE VIRTUAL TABLE docs USING fts5(content)"));
  if (!created) {
    return;
  }

  auto exported = run(io, connection.async_export_sql());
  BOOST_REQUIRE(!exported);
  BOOST_TEST(static_cast<std::int32_t>(exported.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(noto_file_helpers_reject_null_byte_paths) {
  auto io = boost::asio::io_context{};
  auto connection = noto::connection{
    io.get_executor(),
    noto::memory_connection_config("path_validation"),
  };
  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_connect())));

  auto bad_path = std::string{
    "bad\0path",
    8,
  };
  auto backup = run(io, connection.async_backup_to_file(bad_path));
  BOOST_TEST(static_cast<std::int32_t>(backup.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));
  auto exported = run(io, connection.async_export_sql_to_file(bad_path));
  BOOST_TEST(static_cast<std::int32_t>(exported.error().code()) == static_cast<std::int32_t>(noto::detail::error_invalid_argument));

  BOOST_TEST(static_cast<std::bool_t>(run(io, connection.async_close())));
}