feat(storage): v1.x → v2.0 migration core

Adds the one-shot migration tool that reads a v1.x persistence dir
(snapshots/ + wal/) using the v1.x SnapshotManager / WriteAheadLog
readers, materialises an in-memory MemoryStore, and copies every
collection/document into a v2.0 LmdbDocumentStore.

Contract:
  - Idempotent: re-running on an already-migrated env is a no-op success
    (gated by schema_version=2 marker in the _meta sub-db).
  - Read-only on v1.x source: never touches v1_data_dir; rollback is
    "stop using v2.0, restore from the v1.9.5 pre-upgrade backup."
  - Latency-tolerant: one put per doc, plain copy semantics, single
    write txn for the completion marker.

Tests (test_migrate_v1_to_v2):
  - empty v1 dir → marker written, 0 docs migrated
  - single 100-doc collection round-trips
  - 3 collections × 50 docs preserve isolation
  - marker absent pre-migration, present post-migration
  - idempotent re-run returns 0 docs migrated
  - special-character ids (spaces, slashes, UTF-8) round-trip
  - progress callback fires for >1000 doc migration
  - corrupt snapshot leaves v1 dir byte-for-byte identical, no marker

Phase C Stage 3 Task 3.1.

service/CMakeLists.txt

@@ -63,6 +63,7 @@ set(DATABASE_SERVICE_SOURCES
     src/storage/lmdb_txn.cpp
     src/storage/lmdb_dbi.cpp
     src/storage/document_store_lmdb.cpp
+    src/storage/migrate_v1_to_v2.cpp
 )
 
 # Create executable

service/src/storage/migrate_v1_to_v2.cpp

@@ -0,0 +1,386 @@
+// v2.0 storage engine — v1.x → v2.0 migration tool core (Task 3.1).
+//
+// Migrates a v1.x persistence dir (snapshots/ + wal/) into the v2.0 LMDB
+// substrate. See migrate_v1_to_v2.hpp for the contract.
+//
+// Implementation strategy:
+//   1. Pre-flight — if no snapshots/ exists OR migration_complete() already
+//      true → stamp the marker and return success.
+//   2. Build a temporary MemoryStore + drive v1.x recovery: load the
+//      newest snapshot (fallback to older on parse failure) then replay
+//      WAL from the snapshot's walSequence. The result is an in-memory
+//      mirror of the v1.x state.
+//   3. Iterate collections; for each, call target.put() for every doc.
+//      ONE PUT PER DOC — migration is latency-tolerant and the per-txn
+//      overhead is dominated by LMDB fsync, not encode cost.
+//   4. Write the completion marker in a single write txn.
+//
+// Idempotence: LmdbDocumentStore::put() is replace-on-conflict, so
+// re-running over an env that already has partially-migrated rows is safe
+// — every doc is overwritten. The _meta marker (schema_version=2) gates
+// fast-path skip on subsequent boots.
+
+#include "storage/migrate_v1_to_v2.hpp"
+
+#include <lmdb.h>
+
+#include <chrono>
+#include <cstring>
+#include <exception>
+#include <filesystem>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include <spdlog/spdlog.h>
+
+#include "document.hpp"
+#include "memory_store.hpp"
+#include "persistence/snapshot.hpp"
+#include "persistence/wal.hpp"
+#include "storage/document_store.hpp"
+#include "storage/lmdb_env.hpp"
+#include "storage/lmdb_txn.hpp"
+
+namespace fs = std::filesystem;
+
+namespace smartbotic::db::storage {
+
+namespace {
+
+// -------------------------------------------------------------------------
+// _meta sub-db keys
+// -------------------------------------------------------------------------
+
+constexpr const char* kMetaSubDb         = "_meta";
+constexpr const char* kKeySchemaVersion  = "schema_version";
+constexpr const char* kKeyMigratedAt     = "migration_completed_at";
+constexpr const char* kSchemaVersionV2   = "2";
+
+// -------------------------------------------------------------------------
+// LMDB helpers — local subset.
+// -------------------------------------------------------------------------
+
+inline MDB_val to_val(std::string_view sv) {
+    return MDB_val{sv.size(), const_cast<char*>(sv.data())};
+}
+
+[[noreturn]] void throw_mdb(int rc, const char* where) {
+    std::string msg = "LMDB ";
+    msg += where;
+    msg += ": ";
+    msg += mdb_strerror(rc);
+    throw std::runtime_error(msg);
+}
+
+inline void mdb_check(int rc, const char* where) {
+    if (rc != MDB_SUCCESS) throw_mdb(rc, where);
+}
+
+// -------------------------------------------------------------------------
+// _meta sub-db readers (each takes its own short read txn).
+// -------------------------------------------------------------------------
+
+std::optional<std::string> read_meta_key(LmdbEnv& env, const char* key) {
+    ReadTxn rtxn(env);
+    MDB_dbi dbi = 0;
+    int rc = mdb_dbi_open(rtxn.raw(), kMetaSubDb, 0, &dbi);
+    if (rc == MDB_NOTFOUND) return std::nullopt;
+    if (rc != MDB_SUCCESS) throw_mdb(rc, "dbi_open (_meta read)");
+    MDB_val k = to_val(std::string_view(key));
+    MDB_val v{0, nullptr};
+    rc = mdb_get(rtxn.raw(), dbi, &k, &v);
+    if (rc == MDB_NOTFOUND) return std::nullopt;
+    if (rc != MDB_SUCCESS) throw_mdb(rc, "_meta get");
+    return std::string(static_cast<const char*>(v.mv_data), v.mv_size);
+}
+
+void mark_complete(LmdbEnv& env) {
+    WriteTxn wtxn(env);
+    MDB_dbi dbi = 0;
+    mdb_check(mdb_dbi_open(wtxn.raw(), kMetaSubDb, MDB_CREATE, &dbi),
+              "dbi_open (_meta create)");
+
+    // schema_version=2
+    {
+        MDB_val k = to_val(std::string_view(kKeySchemaVersion));
+        MDB_val v = to_val(std::string_view(kSchemaVersionV2));
+        mdb_check(mdb_put(wtxn.raw(), dbi, &k, &v, 0),
+                  "put (_meta schema_version)");
+    }
+    // migration_completed_at = epoch ms (decimal string).
+    {
+        auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+            std::chrono::system_clock::now().time_since_epoch()).count();
+        std::string ts = std::to_string(now_ms);
+        MDB_val k = to_val(std::string_view(kKeyMigratedAt));
+        MDB_val v = to_val(ts);
+        mdb_check(mdb_put(wtxn.raw(), dbi, &k, &v, 0),
+                  "put (_meta migration_completed_at)");
+    }
+    wtxn.commit();
+}
+
+// -------------------------------------------------------------------------
+// v1.x recovery driver — drives SnapshotManager + WriteAheadLog directly
+// rather than instantiating a PersistenceManager (which spins up background
+// threads, attaches replication, etc.).
+// -------------------------------------------------------------------------
+
+void apply_wal_entry(smartbotic::database::MemoryStore& store,
+                     const smartbotic::database::WalEntry& entry) {
+    using namespace smartbotic::database;
+    switch (entry.opType) {
+        case WalOpType::INSERT:
+            if (entry.data) {
+                Document doc = Document::fromJson(*entry.data);
+                store.loadDocument(entry.collection, doc);
+            }
+            break;
+        case WalOpType::UPDATE:
+        case WalOpType::UPSERT:
+            if (entry.data) {
+                Document doc = Document::fromJson(*entry.data);
+                store.loadDocumentWithHistory(entry.collection, doc);
+            }
+            break;
+        case WalOpType::DELETE:
+            store.remove(entry.collection, entry.documentId);
+            break;
+        case WalOpType::CREATE_COLLECTION:
+            if (entry.collectionOptions) {
+                store.createCollection(entry.collection, *entry.collectionOptions);
+            }
+            break;
+        case WalOpType::DROP_COLLECTION:
+            store.dropCollection(entry.collection);
+            break;
+        case WalOpType::SET_ADD:
+            if (entry.data && entry.data->contains("member")) {
+                store.setAdd(entry.collection, entry.documentId,
+                             (*entry.data)["member"].get<std::string>());
+            }
+            break;
+        case WalOpType::SET_REMOVE:
+            if (entry.data && entry.data->contains("member")) {
+                store.setRemove(entry.collection, entry.documentId,
+                                (*entry.data)["member"].get<std::string>());
+            }
+            break;
+        case WalOpType::VEC_PUT:
+            if (entry.vectorData.has_value()) {
+                store.loadVector(entry.collection, entry.documentId, *entry.vectorData);
+            }
+            break;
+        case WalOpType::VEC_DELETE:
+            // v1.x semantics: vector deletes piggy-back on DELETE entries.
+            break;
+    }
+}
+
+void load_v1_state(const std::string& v1_data_dir,
+                   smartbotic::database::MemoryStore& store) {
+    fs::path data_dir(v1_data_dir);
+    fs::path snapshot_dir = data_dir / "snapshots";
+    fs::path wal_dir = data_dir / "wal";
+
+    uint64_t from_seq = 0;
+
+    if (fs::exists(snapshot_dir)) {
+        smartbotic::database::SnapshotManager::Config sc;
+        sc.snapshotDir = snapshot_dir;
+        sc.compressionEnabled = true;
+        // Read-only consumer of snapshots; cleanup belongs to the running
+        // service, not the migration tool.
+        sc.maxSnapshots = 0;
+        sc.validateAfterWrite = false;
+        sc.cleanupOnlyIfVerified = true;
+        smartbotic::database::SnapshotManager snap_mgr(sc);
+
+        auto snapshots = snap_mgr.listSnapshots();
+        if (!snapshots.empty()) {
+            // Try newest first, fall back to older on failure (matches the
+            // v1.x best_effort recovery shape).
+            std::string last_err;
+            bool loaded = false;
+            for (const auto& path : snapshots) {
+                try {
+                    from_seq = snap_mgr.loadSnapshot(path, store);
+                    loaded = true;
+                    spdlog::info("migration: loaded v1.x snapshot {} (wal_seq={})",
+                                 path.string(), from_seq);
+                    break;
+                } catch (const std::exception& e) {
+                    last_err = e.what();
+                    spdlog::warn("migration: snapshot load failed for {}: {}",
+                                 path.string(), e.what());
+                }
+            }
+            if (!loaded) {
+                throw std::runtime_error(
+                    "v1.x snapshot directory present but no snapshot loaded "
+                    "(last error: " + last_err + ")");
+            }
+        } else {
+            spdlog::info("migration: snapshots/ exists but is empty — WAL-only replay");
+        }
+    } else {
+        spdlog::info("migration: no snapshots/ dir — fresh-install path");
+    }
+
+    if (fs::exists(wal_dir)) {
+        smartbotic::database::WriteAheadLog::Config wc;
+        wc.walDir = wal_dir;
+        smartbotic::database::WriteAheadLog wal(wc);
+        if (wal.open()) {
+            uint64_t replayed = wal.replay(from_seq,
+                [&store](const smartbotic::database::WalEntry& entry) {
+                    apply_wal_entry(store, entry);
+                });
+            wal.close();
+            spdlog::info("migration: replayed {} v1.x WAL entries from seq {}",
+                         replayed, from_seq);
+        } else {
+            spdlog::warn("migration: WAL dir present but open() failed — proceeding "
+                         "without WAL replay");
+        }
+    }
+}
+
+}  // namespace
+
+// -------------------------------------------------------------------------
+// Public API
+// -------------------------------------------------------------------------
+
+bool migration_complete(LmdbEnv& env) {
+    try {
+        auto v = read_meta_key(env, kKeySchemaVersion);
+        return v.has_value() && *v == kSchemaVersionV2;
+    } catch (const std::exception& e) {
+        spdlog::warn("migration_complete: read failed: {}", e.what());
+        return false;
+    }
+}
+
+MigrationResult migrate_v1_to_v2(
+    const std::string& v1_data_dir,
+    DocumentStore& target,
+    LmdbEnv& env,
+    std::function<void(const MigrationProgress&)> progress_cb) {
+
+    MigrationResult result;
+    try {
+        // 1. Pre-flight: clean install case.
+        fs::path snap_dir = fs::path(v1_data_dir) / "snapshots";
+        std::error_code ec;
+        bool snap_dir_exists =
+            fs::exists(snap_dir, ec) && fs::is_directory(snap_dir, ec);
+        if (!snap_dir_exists) {
+            // Clean install — no v1.x state to migrate. Still set the
+            // marker so subsequent boots short-circuit cleanly.
+            mark_complete(env);
+            result.success = true;
+            spdlog::info("migration: no v1.x snapshots/ at '{}' — marked as complete",
+                         v1_data_dir);
+            return result;
+        }
+
+        // 2. Idempotence: already migrated → success no-op.
+        if (migration_complete(env)) {
+            result.success = true;
+            spdlog::info("migration: v2.0 marker already present — no-op");
+            return result;
+        }
+
+        // 3. Drive v1.x recovery into a temporary MemoryStore.
+        smartbotic::database::MemoryStore::Config storeCfg;
+        // 8 GB ceiling — migration is bounded by source size and the
+        // ceiling is just to keep the in-RAM mirror from triggering
+        // eviction during the load phase.
+        storeCfg.maxMemoryBytes = 8ULL * 1024 * 1024 * 1024;
+        storeCfg.nodeId = "migration-tool";
+        smartbotic::database::MemoryStore mstore(storeCfg);
+        mstore.start();
+
+        // The migration MemoryStore has no attached PersistenceManager;
+        // make sure callbacks default to nullptr so applyWalEntry can call
+        // store.remove() etc. without surprises.
+        mstore.setPersistCallback(nullptr);
+        mstore.setEventCallback(nullptr);
+
+        load_v1_state(v1_data_dir, mstore);
+
+        // 4. Count phase — for the progress callback.
+        std::vector<std::string> collections = mstore.listCollections();
+        uint64_t docs_total = 0;
+        for (const auto& c : collections) {
+            docs_total += mstore.count(c);
+        }
+
+        MigrationProgress p;
+        p.docs_total = docs_total;
+        p.status_message = "starting copy phase";
+        if (progress_cb) progress_cb(p);
+
+        // 5. Copy phase. ONE PUT PER DOC — latency-tolerant.
+        uint64_t docs_migrated = 0;
+        for (const auto& collection : collections) {
+            auto docs = mstore.getAllDocuments(collection);
+
+            // Some v1.x docs ship without `collection` set; round-trip
+            // would otherwise lose context.
+            for (auto& d : docs) {
+                if (d.collection.empty()) d.collection = collection;
+                target.put(collection, d.id, d);
+                ++docs_migrated;
+
+                if (progress_cb && (docs_migrated % 1000) == 0) {
+                    MigrationProgress prog;
+                    prog.docs_migrated = docs_migrated;
+                    prog.docs_total = docs_total;
+                    prog.current_collection = collection;
+                    prog.status_message = "copying documents";
+                    progress_cb(prog);
+                }
+            }
+
+            result.collections_migrated++;
+            spdlog::info("migration: completed collection '{}' ({} docs)",
+                         collection, docs.size());
+        }
+
+        result.docs_migrated = docs_migrated;
+
+        // 6. Final marker write.
+        mark_complete(env);
+
+        if (progress_cb) {
+            MigrationProgress prog;
+            prog.docs_migrated = docs_migrated;
+            prog.docs_total = docs_total;
+            prog.current_collection = "";
+            prog.status_message = "migration complete";
+            progress_cb(prog);
+        }
+
+        mstore.stop();
+
+        result.success = true;
+        spdlog::info("migration: SUCCESS — migrated {} docs across {} collections",
+                     result.docs_migrated, result.collections_migrated);
+        return result;
+    } catch (const std::exception& e) {
+        result.success = false;
+        result.error = std::string("v2.0 migration failed: ") + e.what();
+        spdlog::error("migration: FAILED — {}", result.error);
+        return result;
+    } catch (...) {
+        result.success = false;
+        result.error = "v2.0 migration failed: unknown exception";
+        spdlog::error("migration: FAILED — {}", result.error);
+        return result;
+    }
+}
+
+}  // namespace smartbotic::db::storage

service/src/storage/migrate_v1_to_v2.hpp

@@ -0,0 +1,75 @@
+// v2.0 storage engine — one-shot v1.x → v2.0 migration tool (core).
+//
+// Reads the canonical v1.x persistence dir (snapshots/ + wal/) using the
+// v1.x SnapshotManager / WriteAheadLog readers, materialises an in-memory
+// MemoryStore via the v1.x recovery sequence, then copies every collection
+// + document into the v2.0 DocumentStore. Writes a marker into the env's
+// `_meta` sub-db on success so subsequent boots skip migration.
+//
+// Contract:
+//   - One-shot. Latency-tolerant. Not optimised for throughput; correctness
+//     first.
+//   - Idempotent. Re-running on an already-migrated env is a no-op success.
+//   - Read-only on the v1.x data dir. Migration never deletes or rewrites
+//     anything under <v1_data_dir>/. Operator rollback is just "stop
+//     using v2.0, point smartbotic-database back at the v1.9.5 backup."
+//
+// Task 3.1 lands the core: snapshot load + WAL replay + per-doc copy +
+// completion marker + idempotence. Vectors and safety nets are layered on
+// top in 3.3 / 3.4.
+
+#pragma once
+
+#include <cstdint>
+#include <functional>
+#include <string>
+
+namespace smartbotic::db::storage {
+
+class DocumentStore;
+class LmdbEnv;
+
+// Periodic progress payload. Emitted from migrate_v1_to_v2() to a caller-
+// supplied callback at a sub-second cadence (every ~1000 docs).
+struct MigrationProgress {
+    uint64_t docs_migrated = 0;
+    uint64_t docs_total = 0;        // 0 = unknown until count phase done
+    std::string current_collection; // "" = pre-scan / done
+    std::string status_message;     // human-readable
+};
+
+// Terminal result of migrate_v1_to_v2(). On success: error is empty,
+// docs_migrated / collections_migrated reflect the work done.
+// On failure: error holds an operator-actionable message.
+struct MigrationResult {
+    bool success = false;
+    uint64_t docs_migrated = 0;
+    uint64_t collections_migrated = 0;
+    std::string error;
+};
+
+// One-shot v1.x → v2.0 migration.
+//
+// Inputs:
+//   v1_data_dir   — directory containing snapshots/ and wal/. May be absent
+//                   or empty (treated as a clean install — returns success
+//                   and stamps the v2.0 marker so subsequent boots are
+//                   short-circuited).
+//   target        — open LmdbDocumentStore over the v2.0 env. Migrations
+//                   copy every collection/document into it.
+//   env           — same LmdbEnv backing `target`. Used to read/write the
+//                   `_meta` sub-db marker.
+//   progress_cb   — invoked roughly every 1000 docs during the copy phase.
+//                   May be nullptr.
+MigrationResult migrate_v1_to_v2(
+    const std::string& v1_data_dir,
+    DocumentStore& target,
+    LmdbEnv& env,
+    std::function<void(const MigrationProgress&)> progress_cb = nullptr);
+
+// True if env's _meta sub-db has a successful migration marker
+// (schema_version=2). Reads via a short-lived read txn; safe to call from
+// any thread.
+bool migration_complete(LmdbEnv& env);
+
+}  // namespace smartbotic::db::storage

tests/CMakeLists.txt

@@ -372,6 +372,63 @@ else()
     target_include_directories(test_document_store PRIVATE ${NLOHMANN_JSON_INCLUDE_DIRS})
 endif()
 
+# v1.x → v2.0 migration tool unit test (v2.0 storage Phase C Stage 3).
+# Pulls in both the v1.x persistence stack (snapshot.cpp, wal.cpp, history_store.cpp,
+# memory_store.cpp) and the v2.0 storage stack (LMDB primitives + DocumentStore +
+# migrate_v1_to_v2). Each side is independent of the other at link time except
+# for the migration .cpp itself, which is the bridge.
+add_executable(test_migrate_v1_to_v2
+    test_migrate_v1_to_v2.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/memory_store.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/config/collection_config_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/snapshot.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/wal.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/history_store.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/json_parse.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/doc_binary.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/storage/lmdb_env.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/storage/lmdb_txn.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/storage/lmdb_dbi.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/storage/document_store_lmdb.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/storage/migrate_v1_to_v2.cpp
+)
+
+target_include_directories(test_migrate_v1_to_v2 PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src
+    ${LMDB_INCLUDE_DIR}
+    ${yyjson_INCLUDE_DIRS}
+)
+
+target_link_libraries(test_migrate_v1_to_v2 PRIVATE ${LMDB_LIBRARY})
+target_link_libraries(test_migrate_v1_to_v2 PRIVATE ${yyjson_LIBRARIES})
+
+if(TARGET nlohmann_json::nlohmann_json)
+    target_link_libraries(test_migrate_v1_to_v2 PRIVATE nlohmann_json::nlohmann_json)
+else()
+    target_include_directories(test_migrate_v1_to_v2 PRIVATE ${NLOHMANN_JSON_INCLUDE_DIRS})
+endif()
+
+if(TARGET spdlog::spdlog)
+    target_link_libraries(test_migrate_v1_to_v2 PRIVATE spdlog::spdlog)
+else()
+    target_link_libraries(test_migrate_v1_to_v2 PRIVATE ${SPDLOG_LIBRARIES})
+    target_include_directories(test_migrate_v1_to_v2 PRIVATE ${SPDLOG_INCLUDE_DIRS})
+endif()
+
+find_package(Threads REQUIRED)
+target_link_libraries(test_migrate_v1_to_v2 PRIVATE Threads::Threads)
+
+# LZ4 — snapshot.cpp uses LZ4 compression
+if(PKG_CONFIG_FOUND)
+    pkg_check_modules(MIGRATE_LZ4 QUIET liblz4)
+endif()
+if(MIGRATE_LZ4_FOUND)
+    target_link_libraries(test_migrate_v1_to_v2 PRIVATE ${MIGRATE_LZ4_LIBRARIES})
+    target_include_directories(test_migrate_v1_to_v2 PRIVATE ${MIGRATE_LZ4_INCLUDE_DIRS})
+else()
+    target_link_libraries(test_migrate_v1_to_v2 PRIVATE lz4)
+endif()
+
 # Register with CTest
 enable_testing()
 add_test(NAME vector_storage COMMAND test_vector_storage)
@@ -384,3 +441,4 @@ add_test(NAME json_parse COMMAND test_json_parse)
 add_test(NAME doc_binary COMMAND test_doc_binary)
 add_test(NAME lmdb_env COMMAND test_lmdb_env)
 add_test(NAME document_store COMMAND test_document_store)
+add_test(NAME migrate_v1_to_v2 COMMAND test_migrate_v1_to_v2)

tests/test_migrate_v1_to_v2.cpp

@@ -0,0 +1,427 @@
+// v2.0 storage engine — unit tests for the v1.x → v2.0 migration tool.
+//
+// Each test materialises a fresh v1.x data dir (snapshots/ + wal/) by driving
+// the v1.x SnapshotManager + MemoryStore APIs directly, then runs
+// migrate_v1_to_v2() against a fresh LMDB env and asserts on the v2.0 state.
+//
+// Task 3.1 covers: empty-dir, single-collection, multi-collection, marker
+// presence, idempotence, special-character ids, progress callback, failure-
+// preserves-source.
+
+#include <algorithm>
+#include <atomic>
+#include <cassert>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <filesystem>
+#include <fstream>
+#include <functional>
+#include <iostream>
+#include <stdexcept>
+#include <string>
+#include <unistd.h>
+#include <vector>
+
+#include <lmdb.h>
+#include <nlohmann/json.hpp>
+
+#include "document.hpp"
+#include "memory_store.hpp"
+#include "persistence/snapshot.hpp"
+#include "persistence/wal.hpp"
+#include "storage/document_store.hpp"
+#include "storage/document_store_lmdb.hpp"
+#include "storage/lmdb_env.hpp"
+#include "storage/lmdb_dbi.hpp"
+#include "storage/lmdb_txn.hpp"
+#include "storage/migrate_v1_to_v2.hpp"
+
+namespace fs = std::filesystem;
+
+using smartbotic::database::CollectionOptions;
+using smartbotic::database::Document;
+using smartbotic::database::MemoryStore;
+using smartbotic::database::SnapshotManager;
+using smartbotic::db::storage::DocumentStore;
+using smartbotic::db::storage::LmdbDocumentStore;
+using smartbotic::db::storage::LmdbEnv;
+using smartbotic::db::storage::LmdbEnvOpts;
+using smartbotic::db::storage::migrate_v1_to_v2;
+using smartbotic::db::storage::migration_complete;
+using smartbotic::db::storage::MigrationProgress;
+using smartbotic::db::storage::MigrationResult;
+
+namespace {
+
+// -------------------------------------------------------------------------
+// Test helpers
+// -------------------------------------------------------------------------
+
+void check(bool cond, const char* msg) {
+    if (!cond) {
+        std::cerr << "FAIL: " << msg << "\n";
+        std::abort();
+    }
+}
+
+std::string make_tmpdir(const char* tag) {
+    static std::atomic<int> counter{0};
+    int n = counter.fetch_add(1);
+    std::string path = "/tmp/migrate-test-" + std::to_string(::getpid()) +
+                       "-" + std::to_string(n) + "-" + tag;
+    std::error_code ec;
+    fs::remove_all(path, ec);
+    fs::create_directories(path);
+    return path;
+}
+
+MemoryStore::Config defaultStoreConfig() {
+    MemoryStore::Config cfg;
+    cfg.nodeId = "test";
+    cfg.maxMemoryBytes = 256ULL * 1024 * 1024;
+    return cfg;
+}
+
+Document makeDoc(const std::string& id, const nlohmann::json& data) {
+    Document d;
+    d.id = id;
+    d.set_data(data);
+    d.version = 1;
+    d.createdAt = 1731628800123ULL;
+    d.updatedAt = 1731628800123ULL;
+    return d;
+}
+
+// Build a v1.x data dir under <root>:
+//   <root>/snapshots/  — created by writeSnapshot()
+//   <root>/wal/        — empty
+struct V1Builder {
+    fs::path data_dir;
+    std::unique_ptr<MemoryStore> store;
+
+    explicit V1Builder(const std::string& path) : data_dir(path) {
+        fs::create_directories(data_dir / "snapshots");
+        fs::create_directories(data_dir / "wal");
+        store = std::make_unique<MemoryStore>(defaultStoreConfig());
+        store->start();
+    }
+
+    ~V1Builder() {
+        if (store) store->stop();
+    }
+
+    void createCollection(const std::string& name, const CollectionOptions& opts = {}) {
+        store->createCollection(name, opts);
+    }
+
+    void insert(const std::string& collection, const Document& doc) {
+        store->insert(collection, doc);
+    }
+
+    // Materialise the snapshot file. Required before migrate_v1_to_v2()
+    // can pick anything up.
+    fs::path writeSnapshot(uint64_t walSequence = 1) {
+        SnapshotManager::Config sc;
+        sc.snapshotDir = data_dir / "snapshots";
+        sc.compressionEnabled = true;
+        sc.validateAfterWrite = true;
+        SnapshotManager mgr(sc);
+        return mgr.createSnapshot(*store, walSequence);
+    }
+};
+
+struct TmpEnv {
+    std::string path;
+    LmdbEnv env;
+    explicit TmpEnv(const char* tag)
+        : path(make_tmpdir(tag)),
+          env(LmdbEnvOpts{path + "/env", 64ULL << 20, 256, 126, false}) {}
+    ~TmpEnv() {
+        std::error_code ec;
+        fs::remove_all(path, ec);
+    }
+    TmpEnv(const TmpEnv&) = delete;
+    TmpEnv& operator=(const TmpEnv&) = delete;
+};
+
+// Quick checksum over a directory tree (size + content of every regular
+// file). Used to verify migration doesn't touch the v1.x source.
+struct DirChecksum {
+    std::vector<std::pair<std::string, std::string>> entries;
+
+    void capture(const fs::path& root) {
+        entries.clear();
+        for (auto it = fs::recursive_directory_iterator(root);
+             it != fs::recursive_directory_iterator(); ++it) {
+            if (!it->is_regular_file()) continue;
+            std::string rel = fs::relative(it->path(), root).string();
+            std::ifstream f(it->path(), std::ios::binary);
+            std::string content((std::istreambuf_iterator<char>(f)),
+                                std::istreambuf_iterator<char>());
+            entries.emplace_back(rel, std::move(content));
+        }
+        std::sort(entries.begin(), entries.end());
+    }
+
+    bool operator==(const DirChecksum& other) const {
+        return entries == other.entries;
+    }
+    bool operator!=(const DirChecksum& other) const { return !(*this == other); }
+};
+
+}  // anonymous namespace
+
+// -------------------------------------------------------------------------
+// Task 3.1: Migration core
+// -------------------------------------------------------------------------
+
+void test_empty_v1_dir_yields_empty_v2_env() {
+    auto dir = make_tmpdir("empty");
+    // No snapshots/ subdir, just an empty dir.
+    TmpEnv env_dir("empty-env");
+    LmdbDocumentStore target(env_dir.env);
+
+    MigrationResult r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r.success, "empty v1 dir should succeed");
+    check(r.docs_migrated == 0, "empty v1 dir migrates 0 docs");
+    check(r.collections_migrated == 0, "empty v1 dir migrates 0 collections");
+
+    // Marker IS written for empty case so subsequent boots don't re-check.
+    check(migration_complete(env_dir.env), "marker should be set after empty migration");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: empty v1 dir yields empty v2 env\n";
+}
+
+void test_single_collection_with_docs_migrates() {
+    auto dir = make_tmpdir("single");
+    {
+        V1Builder b(dir);
+        b.createCollection("widgets");
+        for (int i = 0; i < 100; i++) {
+            b.insert("widgets", makeDoc("w" + std::to_string(i),
+                {{"index", i}, {"name", "widget-" + std::to_string(i)}}));
+        }
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("single-env");
+    LmdbDocumentStore target(env_dir.env);
+    MigrationResult r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r.success, "single collection migration succeeds");
+    check(r.docs_migrated == 100, "100 docs migrated");
+    check(r.collections_migrated == 1, "1 collection migrated");
+
+    check(target.count("widgets") == 100, "v2 widgets has 100 docs");
+    auto d = target.get("widgets", "w42");
+    check(d.has_value(), "w42 exists in v2");
+    check(d->data()["name"] == "widget-42", "data round-trips");
+    check(d->data()["index"] == 42, "data round-trips (int)");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: single collection with docs migrates\n";
+}
+
+void test_multiple_collections_migrate() {
+    auto dir = make_tmpdir("multi");
+    {
+        V1Builder b(dir);
+        b.createCollection("colA");
+        b.createCollection("colB");
+        b.createCollection("colC");
+        for (int i = 0; i < 50; i++) {
+            b.insert("colA", makeDoc("a" + std::to_string(i), {{"k", "A"}, {"i", i}}));
+            b.insert("colB", makeDoc("b" + std::to_string(i), {{"k", "B"}, {"i", i}}));
+            b.insert("colC", makeDoc("c" + std::to_string(i), {{"k", "C"}, {"i", i}}));
+        }
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("multi-env");
+    LmdbDocumentStore target(env_dir.env);
+    MigrationResult r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r.success, "multi-collection migration succeeds");
+    check(r.docs_migrated == 150, "150 docs migrated total");
+    check(r.collections_migrated == 3, "3 collections migrated");
+
+    check(target.count("colA") == 50, "colA has 50");
+    check(target.count("colB") == 50, "colB has 50");
+    check(target.count("colC") == 50, "colC has 50");
+
+    // Isolation: a25 should NOT exist in colB.
+    check(!target.get("colB", "a25").has_value(), "isolation preserved");
+    check(target.get("colA", "a25").has_value(), "colA a25 present");
+    check(target.get("colC", "c25").has_value(), "colC c25 present");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: multiple collections migrate\n";
+}
+
+void test_marker_is_set_after_successful_migration() {
+    auto dir = make_tmpdir("marker");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        b.insert("c", makeDoc("d1", {{"v", 1}}));
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("marker-env");
+    LmdbDocumentStore target(env_dir.env);
+    check(!migration_complete(env_dir.env), "marker absent pre-migration");
+    auto r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r.success, "migration succeeds");
+    check(migration_complete(env_dir.env), "marker present post-migration");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: marker set after successful migration\n";
+}
+
+void test_idempotent_re_run() {
+    auto dir = make_tmpdir("idempotent");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        for (int i = 0; i < 10; i++) {
+            b.insert("c", makeDoc("d" + std::to_string(i), {{"v", i}}));
+        }
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("idempotent-env");
+    LmdbDocumentStore target(env_dir.env);
+    auto r1 = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r1.success, "first run succeeds");
+    check(r1.docs_migrated == 10, "first run migrates 10 docs");
+
+    auto r2 = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r2.success, "second run succeeds (idempotent)");
+    check(r2.docs_migrated == 0, "second run migrates 0 docs (already done)");
+    check(r2.collections_migrated == 0, "second run migrates 0 collections");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: idempotent re-run\n";
+}
+
+void test_collections_with_special_chars_in_id() {
+    auto dir = make_tmpdir("special");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        b.insert("c", makeDoc("id with spaces", {{"v", "a"}}));
+        b.insert("c", makeDoc("id/with/slashes", {{"v", "b"}}));
+        // Unicode (UTF-8 bytes are well-formed in LMDB keys).
+        b.insert("c", makeDoc(std::string("id_unicode_") + "\xc3\xa1\xc3\xa9\xc3\xad",
+                              {{"v", "c"}}));
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("special-env");
+    LmdbDocumentStore target(env_dir.env);
+    auto r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r.success, "special-id migration succeeds");
+    check(r.docs_migrated == 3, "3 docs migrated");
+    check(target.get("c", "id with spaces").has_value(), "spaces id present");
+    check(target.get("c", "id/with/slashes").has_value(), "slashes id present");
+    check(target.get("c", std::string("id_unicode_") + "\xc3\xa1\xc3\xa9\xc3\xad")
+              .has_value(),
+          "unicode id present");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: special-char ids round-trip\n";
+}
+
+void test_progress_callback_invoked() {
+    auto dir = make_tmpdir("progress");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        // > 1000 docs to guarantee at least one progress beat.
+        for (int i = 0; i < 1500; i++) {
+            b.insert("c", makeDoc("d" + std::to_string(i), {{"i", i}}));
+        }
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("progress-env");
+    LmdbDocumentStore target(env_dir.env);
+    int beats = 0;
+    uint64_t last_total = 0;
+    auto cb = [&](const MigrationProgress& p) {
+        ++beats;
+        if (p.docs_total > 0) last_total = p.docs_total;
+    };
+    auto r = migrate_v1_to_v2(dir, target, env_dir.env, cb);
+    check(r.success, "migration with cb succeeds");
+    check(beats >= 1, "at least one progress beat fired");
+    check(last_total == 1500, "docs_total = 1500 at some beat");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: progress callback invoked\n";
+}
+
+void test_failure_preserves_v1_dir() {
+    auto dir = make_tmpdir("preserve");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        b.insert("c", makeDoc("d1", {{"v", 1}}));
+        b.writeSnapshot();
+    }
+
+    // Corrupt the latest snapshot file so the migration fails on load.
+    fs::path snap_dir = fs::path(dir) / "snapshots";
+    for (auto& e : fs::directory_iterator(snap_dir)) {
+        if (e.is_regular_file()) {
+            uintmax_t sz = fs::file_size(e.path());
+            fs::resize_file(e.path(), sz / 2);
+            break;
+        }
+    }
+
+    // Capture the v1 dir state AFTER corruption — this is our baseline.
+    DirChecksum baseline;
+    baseline.capture(dir);
+
+    TmpEnv env_dir("preserve-env");
+    LmdbDocumentStore target(env_dir.env);
+    auto r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(!r.success, "corrupt-snapshot migration should fail");
+    check(!r.error.empty(), "error message is populated");
+
+    DirChecksum after;
+    after.capture(dir);
+    check(baseline == after, "v1 dir unchanged after failed migration");
+    check(!migration_complete(env_dir.env), "marker NOT set after failure");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: failure preserves v1 dir\n";
+}
+
+// -------------------------------------------------------------------------
+// main()
+// -------------------------------------------------------------------------
+
+int main() {
+    // Task 3.1: Migration core
+    test_empty_v1_dir_yields_empty_v2_env();
+    test_single_collection_with_docs_migrates();
+    test_multiple_collections_migrate();
+    test_marker_is_set_after_successful_migration();
+    test_idempotent_re_run();
+    test_collections_with_special_chars_in_id();
+    test_progress_callback_invoked();
+    test_failure_preserves_v1_dir();
+
+    std::cout << "\nAll migrate_v1_to_v2 tests passed.\n";
+    return 0;
+}