main.cpp

// main.cpp — HPX-powered concurrent demo for memdb
//
// Build:
//   hpxcxx -std=c++17 -O2 -I include src/main.cpp -o memdb_demo
//   ./memdb_demo --hpx:threads=4
//
// Without HPX (for testing):
//   g++ -std=c++17 -O2 -pthread -I include src/main.cpp -o memdb_demo
//   (define NO_HPX below)

// ---- HPX integration -------------------------------------------------------
// HPX replaces std::thread with lightweight user-space tasks scheduled across
// OS threads via work-stealing. This means:
//   hpx::async  →  submits a task to HPX thread pool
//   hpx::future →  HPX's version of std::future (composable, continuations)
//   hpx::wait_all / hpx::when_all → barrier across futures
//
// The storage engine itself is thread-safe (shared_mutex + lock table).
// HPX just provides the task scheduler on top.
// ---------------------------------------------------------------------------

#ifndef NO_HPX
#include <hpx/hpx_main.hpp>         // provides main() entry via HPX runtime
#include <hpx/future.hpp>
#include <hpx/include/async.hpp>
#include <hpx/include/parallel_algorithm.hpp>
namespace async_impl {
    template<typename F> auto submit(F&& f) { return hpx::async(std::forward<F>(f)); }
}
using future_t = hpx::future<void>;
#define WAIT_ALL   hpx::wait_all
#else
// Fallback: use std::async so the code compiles without HPX
#include <future>
#include <thread>
#include <iostream>
namespace async_impl {
    template<typename F> auto submit(F&& f) {
        return std::async(std::launch::async, std::forward<F>(f));
    }
}
using future_t = std::future<void>;
#define WAIT_ALL(v) for (auto& f : v) f.get()
#endif

#define ASYNC(...) async_impl::submit(__VA_ARGS__)
#define FUTURE   future_t

#include "storage_engine.hpp"
#include <iostream>
#include <vector>
#include <string>
#include <chrono>
#include <random>

using namespace memdb;

// ---- Helpers ---------------------------------------------------------------

static void print_separator(const std::string& title) {
    std::cout << "\n========================================\n";
    std::cout << "  " << title << "\n";
    std::cout << "========================================\n";
}

static std::string make_key(int i) {
    return "user:" + std::to_string(i);
}

// ---- Demo scenarios --------------------------------------------------------

// 1. Basic CRUD
void demo_basic_crud(StorageEngine& db, TransactionManager& tm) {
    print_separator("Basic CRUD");

    TxnId t1 = tm.begin();
    db.insert(t1, "user:1", "Alice");
    db.insert(t1, "user:2", "Bob");
    db.insert(t1, "user:3", "Charlie");
    tm.commit(t1);
    std::cout << "[INSERT] Inserted Alice, Bob, Charlie\n";

    TxnId t2 = tm.begin();
    auto val = db.read(t2, "user:1");
    std::cout << "[READ]   user:1 = " << val.value_or("NOT FOUND") << "\n";
    tm.commit(t2);

    TxnId t3 = tm.begin();
    db.update(t3, "user:2", "Bobby");
    tm.commit(t3);
    std::cout << "[UPDATE] user:2 updated to Bobby\n";

    TxnId t4 = tm.begin();
    db.remove(t4, "user:3");
    tm.commit(t4);
    std::cout << "[DELETE] user:3 deleted\n";

    TxnId t5 = tm.begin();
    auto deleted = db.read(t5, "user:3");
    std::cout << "[READ]   user:3 after delete = "
              << deleted.value_or("NOT FOUND") << "\n";
    tm.commit(t5);
}

// 2. Range queries via B-Tree
void demo_range_query(StorageEngine& db, TransactionManager& tm) {
    print_separator("Range Queries (B-Tree)");

    TxnId t = tm.begin();
    for (int i = 10; i <= 50; i += 10) {
        db.insert(t, "sensor:" + std::to_string(i), "val=" + std::to_string(i * 3));
    }
    tm.commit(t);

    TxnId t2 = tm.begin();
    auto rows = db.range_scan(t2, "sensor:10", "sensor:40");
    std::cout << "[RANGE]  sensor:10 -> sensor:40:\n";
    for (auto& r : rows)
        std::cout << "         " << r.key << " => " << r.value << "\n";
    tm.commit(t2);
}

// 3. Concurrent reads (HPX async tasks)
void demo_concurrent_reads(StorageEngine& db, TransactionManager& tm) {
    print_separator("Concurrent Reads (HPX async)");

    // Pre-populate
    TxnId seed = tm.begin();
    for (int i = 0; i < 20; ++i)
        db.insert(seed, make_key(100 + i), "data_" + std::to_string(i));
    tm.commit(seed);

    // Fire off 10 concurrent read transactions
    std::vector<FUTURE> futures;
    for (int i = 0; i < 10; ++i) {
        futures.push_back(ASYNC([&db, &tm, i]() {
            TxnId txn = tm.begin();
            auto v1 = db.read(txn, make_key(100 + i));
            auto v2 = db.read(txn, make_key(100 + i + 10));
            tm.commit(txn);
            std::cout << "[CONCURRENT READ] task " << i
                      << ": " << make_key(100+i) << "=" << v1.value_or("?")
                      << "  " << make_key(100+i+10) << "=" << v2.value_or("?") << "\n";
        }));
    }
    WAIT_ALL(futures);
    std::cout << "[DONE]   All concurrent reads completed\n";
}

// 4. Deadlock detection
void demo_deadlock(StorageEngine& db, TransactionManager& tm) {
    print_separator("Deadlock Detection");

    // Setup two rows
    TxnId seed = tm.begin();
    db.insert(seed, "lock:A", "alpha");
    db.insert(seed, "lock:B", "beta");
    tm.commit(seed);

    // Classic deadlock:
    //   T1 locks A then tries B
    //   T2 locks B then tries A

    std::atomic<bool> t1_done{false};
    std::atomic<bool> t2_done{false};

    TxnId t1_id = tm.begin();
    TxnId t2_id = tm.begin();

    auto task1 = ASYNC([&]() {
        try {
            tm.lock_row(t1_id, "lock:A", LockMode::EXCLUSIVE);
            std::cout << "[T1] Locked A\n";
            // Small pause so T2 gets a chance to lock B first
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
            tm.lock_row(t1_id, "lock:B", LockMode::EXCLUSIVE); // may deadlock
            db.update(t1_id, "lock:A", "alpha_updated");
            db.update(t1_id, "lock:B", "beta_updated");
            tm.commit(t1_id);
            std::cout << "[T1] Committed\n";
        } catch (DeadlockException& e) {
            tm.abort(t1_id);
            std::cout << "[T1] ABORTED — deadlock victim: txn "
                      << e.victim_id << "\n";
        }
        t1_done = true;
    });

    auto task2 = ASYNC([&]() {
        try {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            tm.lock_row(t2_id, "lock:B", LockMode::EXCLUSIVE);
            std::cout << "[T2] Locked B\n";
            tm.lock_row(t2_id, "lock:A", LockMode::EXCLUSIVE); // may deadlock
            db.update(t2_id, "lock:B", "beta_t2");
            tm.commit(t2_id);
            std::cout << "[T2] Committed\n";
        } catch (DeadlockException& e) {
            tm.abort(t2_id);
            std::cout << "[T2] ABORTED — deadlock victim: txn "
                      << e.victim_id << "\n";
        }
        t2_done = true;
    });

    std::vector<FUTURE> dl_futures;
    dl_futures.push_back(std::move(task1));
    dl_futures.push_back(std::move(task2));
    WAIT_ALL(dl_futures);

    std::cout << "[DONE]   Deadlock scenario resolved\n";
}

// 5. Concurrent writes with contention
void demo_concurrent_writes(StorageEngine& db, TransactionManager& tm) {
    print_separator("Concurrent Writes (HPX, 8 tasks)");

    TxnId seed = tm.begin();
    db.insert(seed, "counter:X", "0");
    tm.commit(seed);

    std::atomic<int> success_count{0}, abort_count{0};
    std::vector<FUTURE> futures;

    for (int i = 0; i < 8; ++i) {
        futures.push_back(ASYNC([&, i]() {
            try {
                TxnId txn = tm.begin();
                auto val = db.read(txn, "counter:X");
                int cur = val ? std::stoi(*val) : 0;
                db.update(txn, "counter:X", std::to_string(cur + 1));
                tm.commit(txn);
                ++success_count;
            } catch (DeadlockException& e) {
                ++abort_count;
            } catch (...) {
                ++abort_count;
            }
        }));
    }
    WAIT_ALL(futures);

    TxnId check = tm.begin();
    auto final_val = db.read(check, "counter:X");
    tm.commit(check);

    std::cout << "[RESULT] counter:X = " << final_val.value_or("?")
              << "  (success=" << success_count
              << ", aborted=" << abort_count << ")\n";
}

// ---- Entry point -----------------------------------------------------------

#ifndef NO_HPX
int hpx_main(int argc, char* argv[]) {
#else
int main() {
#endif
    TransactionManager tm;
    StorageEngine      db(tm);

    demo_basic_crud(db, tm);
    demo_range_query(db, tm);
    demo_concurrent_reads(db, tm);
    demo_deadlock(db, tm);
    demo_concurrent_writes(db, tm);

    print_separator("WAL (Write-Ahead Log) Dump");
    std::cout << db.wal_dump();

    print_separator("Done");
    std::cout << "Total rows in engine: " << db.size() << "\n";

#ifndef NO_HPX
    return hpx::finalize();
#else
    return 0;
#endif
}