// Transactions: begin/commit/abort, MVCC read-only observer
// Раздел учебника: Том I, глава 5, «Транзакции — первые шаги»,
//   разделы «begin/commit/abort», «read-only».
// Что демонстрирует: commit фиксирует изменения; abort откатывает их;
//   read-only транзакция-наблюдатель (в отдельном потоке) видит стабильный
//   снапшот (MVCC): невидимые незакоммиченные и последующие закоммиченные
//   изменения. Пишущая и читающая транзакции одного потока не могут пересекаться
//   (MDBX_TXN_OVERLAPPING), поэтому наблюдатель вынесен в отдельный поток.
// Сборка: g++ -std=c++17 -I<libmdbx-include> -I../common 04-transactions.c++ -lmdbx -pthread
// Запуск: ./04-transactions
// Ожидаемый вывод:
//   observer sees: v1
//   during uncommitted write, observer still sees: v1
//   after abort, observer still sees: v1
//   after committed v3, old observer still sees: v1
//   fresh reader sees: v3
//   ok: abort rolled back; MVCC snapshot stable
// При сбое — сообщение в stderr и exit 1.
//
// SPDX-License-Identifier: Apache-2.0

#include <atomic>
#include <iostream>
#include <string>
#include <thread>

#include "common.h++"
#include <mdbx.h++>

namespace {

std::atomic<int> stage{0};

void observer_run(mdbx::env_managed &env) {
  auto rtxn = env.start_read();
  auto table = rtxn.open_map(nullptr);
  std::cout << "observer sees: " << rtxn.get(table, mdbx::slice("key")).as_string() << "\n";
  stage = 1;

  while (stage.load() != 2)
    std::this_thread::yield();
  std::cout << "during uncommitted write, observer still sees: " << rtxn.get(table, mdbx::slice("key")).as_string()
            << "\n";
  stage = 3;

  while (stage.load() != 4)
    std::this_thread::yield();
  std::cout << "after abort, observer still sees: " << rtxn.get(table, mdbx::slice("key")).as_string() << "\n";
  stage = 5;

  while (stage.load() != 6)
    std::this_thread::yield();
  std::cout << "after committed v3, old observer still sees: " << rtxn.get(table, mdbx::slice("key")).as_string()
            << "\n";
  rtxn.abort();

  // Новый читатель после коммита видит актуальное состояние.
  auto fresh = env.start_read();
  auto ftable = fresh.open_map(nullptr);
  std::cout << "fresh reader sees: " << fresh.get(ftable, mdbx::slice("key")).as_string() << "\n";
  fresh.abort();
  stage = 7;
}

} // namespace

int main() {
  try {
    const std::string path = std::string(example::tmpdir()) + "/mdbx-04-transactions.mdbx";
    mdbx::env::remove(path);

    mdbx::env_managed env = example::env_open(path);

    // Базовая запись, зафиксированная первой транзакцией.
    {
      auto txn = env.start_write();
      auto table = txn.open_map(nullptr);
      txn.insert(table, mdbx::slice("key"), mdbx::slice("v1"));
      txn.commit();
    }

    // Read-only наблюдатель в отдельном потоке: фиксированный снапшот базы.
    std::thread observer(observer_run, std::ref(env));
    while (stage.load() != 1)
      std::this_thread::yield();

    // Пишущая транзакция перезаписывает ключ, но ещё не коммитит.
    {
      auto wtxn = env.start_write();
      auto table = wtxn.open_map(nullptr);
      wtxn.upsert(table, mdbx::slice("key"), mdbx::slice("v2"));
      // наблюдатель проверяет снапшот при незакоммиченной записи
      stage = 2;
      while (stage.load() != 3)
        std::this_thread::yield();
      // откат: v2 не сохраняется
      wtxn.abort();
    }
    // наблюдатель проверяет снапшот после abort
    stage = 4;
    while (stage.load() != 5)
      std::this_thread::yield();

    // Новая пишущая транзакция коммитит v3.
    {
      auto wtxn = env.start_write();
      auto table = wtxn.open_map(nullptr);
      wtxn.upsert(table, mdbx::slice("key"), mdbx::slice("v3"));
      wtxn.commit();
    }
    // наблюдатель проверяет снапшот после коммита v3
    stage = 6;
    while (stage.load() != 7)
      std::this_thread::yield();

    observer.join();

    std::cout << "ok: abort rolled back; MVCC snapshot stable\n";
    return EXIT_SUCCESS;
  } catch (const std::exception &ex) {
    std::cerr << "error: " << ex.what() << "\n";
    return EXIT_FAILURE;
  }
}