// Transactions: begin/commit/abort, MVCC read-only observer
// Textbook section: Volume I, chapter 5, "Transactions — first steps",
//   sections "begin/commit/abort", "read-only".
// What it demonstrates: commit fixes the changes; abort rolls them back;
//   a read-only observer transaction (in a separate thread) sees a stable
//   snapshot (MVCC): invisible uncommitted and subsequent committed
//   changes. A writing and a reading transaction of one thread cannot overlap
//   (MDBX_TXN_OVERLAPPING), therefore the observer is moved to a separate thread.
// Build: g++ -std=c++17 -I<libmdbx-include> -I../common 04-transactions.c++ -lmdbx -pthread
// Run: ./04-transactions
// Expected output:
//   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
// On failure — message to stderr and 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();

  // A new reader after the commit sees the actual state.
  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);

    // A base record committed by the first transaction.
    {
      auto txn = env.start_write();
      auto table = txn.open_map(nullptr);
      txn.insert(table, mdbx::slice("key"), mdbx::slice("v1"));
      txn.commit();
    }

    // A read-only observer in a separate thread: a fixed snapshot of the database.
    std::thread observer(observer_run, std::ref(env));
    while (stage.load() != 1)
      std::this_thread::yield();

    // A writing transaction overwrites the key but does not commit yet.
    {
      auto wtxn = env.start_write();
      auto table = wtxn.open_map(nullptr);
      wtxn.upsert(table, mdbx::slice("key"), mdbx::slice("v2"));
      // the observer checks the snapshot during the uncommitted write
      stage = 2;
      while (stage.load() != 3)
        std::this_thread::yield();
      // rollback: v2 is not saved
      wtxn.abort();
    }
    // the observer checks the snapshot after the abort
    stage = 4;
    while (stage.load() != 5)
      std::this_thread::yield();

    // A new writing transaction commits v3.
    {
      auto wtxn = env.start_write();
      auto table = wtxn.open_map(nullptr);
      wtxn.upsert(table, mdbx::slice("key"), mdbx::slice("v3"));
      wtxn.commit();
    }
    // the observer checks the snapshot after the commit of 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;
  }
}