feat(storage): LmdbDocumentStore implementation

Phase C Stage 2 Task 2.2 — fills in the LMDB-backed DocumentStore.

  - One sub-database per collection. Handles are opened lazily on first
    write (MDB_CREATE) and cached under a mutex; read-only ops resolve
    via try_open_for_read and return nullopt for absent collections.
  - On-disk encoding is Document::toJson().dump() / fromJson() — pays an
    encode+decode cost per put/get but honours the Phase B binary-doc
    substrate via the canonical accessors.
  - scan() cursor-iterates the sub-db, applies filter/sort/projection
    semantics ported (read-only) from MemoryStore::find. Counters
    (total_matched, has_more) are computed pre-pagination.
  - list_collections() walks the unnamed root dbi, skipping `_`-prefixed
    system sub-dbs and returning the rest sorted alphabetically.
  - drop_collection() uses mdb_drop with del=1 to free the sub-db and
    evicts the dbi cache entry so subsequent ops re-create with MDB_CREATE.

All 19 Task 2.1 tests now pass. Stage 1 lmdb_env tests still green.

service/src/storage/document_store_lmdb.cpp

@@ -1,61 +1,571 @@
-// v2.0 storage engine — LMDB-backed DocumentStore implementation.
+// v2.0 storage engine — LMDB-backed DocumentStore implementation (Task 2.2).
 //
-// Task 2.1 — stub-only: every operation throws std::runtime_error so the
-// test binary links and the tests fail at runtime (TDD RED). Task 2.2 fills
-// in the real implementation.
+// On-disk encoding:
+//   key   = doc id (UTF-8 bytes)
+//   value = Document::toJson().dump() (UTF-8 JSON text)
+//
+// Encode/decode cost is paid per put/get; the substrate-swap exchange is
+// "lose v1.x's in-RAM speed" for "gain on-disk durability and bounded RSS".
+// Stage 4 may optimise to a binary frame later.
+//
+// Collection -> sub-db mapping:
+//   Each collection name is the sub-db name. Sub-db handles (MDB_dbi) are
+//   cached per-collection on first open and reused across operations. The
+//   first write to a collection opens with MDB_CREATE; reads use try_open
+//   which returns nullopt when the sub-db doesn't exist yet (since read
+//   txns can't MDB_CREATE).
+//
+// Filter / sort / projection semantics:
+//   Ported from service/src/memory_store.cpp (read-only reference). The
+//   logic lives in the anonymous namespace below — predicate evaluation,
+//   path resolution, regex, and full-text search all match v1.x exactly.
 
 #include "storage/document_store_lmdb.hpp"
 
+#include <lmdb.h>
+
+#include <algorithm>
+#include <cctype>
+#include <functional>
+#include <regex>
+#include <sstream>
 #include <stdexcept>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
 
+#include "doc_binary.hpp"
+#include "document.hpp"
+#include "json_parse.hpp"
+#include "storage/lmdb_dbi.hpp"
 #include "storage/lmdb_env.hpp"
+#include "storage/lmdb_txn.hpp"
 
 namespace smartbotic::db::storage {
 
-LmdbDocumentStore::LmdbDocumentStore(LmdbEnv& env) noexcept : env_(env) {}
+namespace {
+
+// -------------------------------------------------------------------------
+// LMDB error helpers
+// -------------------------------------------------------------------------
 
-void LmdbDocumentStore::put(std::string_view, std::string_view,
-                             const smartbotic::database::Document&) {
-    throw std::runtime_error("LmdbDocumentStore::put not implemented (Task 2.2)");
+[[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);
 }
 
-std::optional<smartbotic::database::Document>
-LmdbDocumentStore::get(std::string_view, std::string_view) {
-    throw std::runtime_error("LmdbDocumentStore::get not implemented (Task 2.2)");
+inline void mdb_check(int rc, const char* where) {
+    if (rc != MDB_SUCCESS) throw_mdb(rc, where);
 }
 
-bool LmdbDocumentStore::del(std::string_view, std::string_view) {
-    throw std::runtime_error("LmdbDocumentStore::del not implemented (Task 2.2)");
+inline MDB_val to_val(std::string_view sv) {
+    return MDB_val{sv.size(), const_cast<char*>(sv.data())};
 }
 
-uint64_t LmdbDocumentStore::count(std::string_view) {
-    throw std::runtime_error("LmdbDocumentStore::count not implemented (Task 2.2)");
+inline std::string_view to_sv(const MDB_val& v) {
+    return std::string_view(static_cast<const char*>(v.mv_data), v.mv_size);
 }
 
-ScanResult LmdbDocumentStore::scan(std::string_view,
-                                    const smartbotic::database::Query&) {
-    throw std::runtime_error("LmdbDocumentStore::scan not implemented (Task 2.2)");
+inline std::string to_cstr(std::string_view name) {
+    return std::string(name);
 }
 
-std::vector<std::string> LmdbDocumentStore::list_collections() {
-    throw std::runtime_error(
-        "LmdbDocumentStore::list_collections not implemented (Task 2.2)");
+bool is_system_subdb(std::string_view name) {
+    return !name.empty() && name[0] == '_';
 }
 
-bool LmdbDocumentStore::drop_collection(std::string_view) {
-    throw std::runtime_error(
-        "LmdbDocumentStore::drop_collection not implemented (Task 2.2)");
+// -------------------------------------------------------------------------
+// Encode/decode a Document <-> JSON-text payload.
+//
+// We round-trip through Document::toJson() / Document::fromJson() which is
+// the Phase B canonical path (it decodes data_binary into the json tree
+// under "data"). yyjson parses the byte stream on read.
+// -------------------------------------------------------------------------
+
+std::string encode_document(const smartbotic::database::Document& doc) {
+    // .dump() with no indent is the canonical compact form.
+    return doc.toJson().dump();
+}
+
+smartbotic::database::Document decode_document(std::string_view bytes) {
+    nlohmann::json j = smartbotic::db::parse_to_nlohmann(bytes);
+    return smartbotic::database::Document::fromJson(j);
+}
+
+// -------------------------------------------------------------------------
+// Filter / sort / projection — ported from MemoryStore::find /
+// MemoryStore::matchesFilters / etc (service/src/memory_store.cpp). The
+// behaviour matches v1.x bit-for-bit: dotted-path resolution for nested
+// fields, special-cased struct members (_id / _created_at / _updated_at /
+// _version), and the same regex / SEARCH semantics.
+// -------------------------------------------------------------------------
+
+std::optional<nlohmann::json> get_json_path(const nlohmann::json& obj,
+                                             const std::string& path) {
+    if (path.empty() || !obj.is_object()) return std::nullopt;
+    std::vector<std::string> parts;
+    std::istringstream iss(path);
+    std::string part;
+    while (std::getline(iss, part, '.')) parts.push_back(part);
+    const nlohmann::json* current = &obj;
+    for (const auto& p : parts) {
+        if (!current->is_object() || !current->contains(p)) return std::nullopt;
+        current = &(*current)[p];
+    }
+    return std::optional<nlohmann::json>(*current);
+}
+
+bool compare_json(const nlohmann::json& a,
+                  smartbotic::database::FilterOp op,
+                  const nlohmann::json& b) {
+    using smartbotic::database::FilterOp;
+    switch (op) {
+        case FilterOp::EQ:  return a == b;
+        case FilterOp::NE:  return a != b;
+        case FilterOp::GT:  return a > b;
+        case FilterOp::GTE: return a >= b;
+        case FilterOp::LT:  return a < b;
+        case FilterOp::LTE: return a <= b;
+        default:            return false;
+    }
+}
+
+bool search_in_json(const nlohmann::json& obj,
+                    const std::string& lower_term) {
+    if (obj.is_string()) {
+        std::string lower = obj.get<std::string>();
+        std::transform(lower.begin(), lower.end(), lower.begin(),
+                       [](unsigned char c) { return std::tolower(c); });
+        return lower.find(lower_term) != std::string::npos;
+    }
+    if (obj.is_object()) {
+        for (const auto& [k, v] : obj.items()) {
+            std::string lower_key = k;
+            std::transform(lower_key.begin(), lower_key.end(), lower_key.begin(),
+                           [](unsigned char c) { return std::tolower(c); });
+            if (lower_key.find(lower_term) != std::string::npos) return true;
+            if (search_in_json(v, lower_term)) return true;
+        }
+    }
+    if (obj.is_array()) {
+        for (const auto& item : obj) {
+            if (search_in_json(item, lower_term)) return true;
+        }
+    }
+    if (obj.is_number()) {
+        std::string s = obj.dump();
+        if (s.find(lower_term) != std::string::npos) return true;
+    }
+    return false;
+}
+
+bool matches_search(const smartbotic::database::Document& doc,
+                    const std::string& term) {
+    if (term.empty()) return true;
+    std::string lower = term;
+    std::transform(lower.begin(), lower.end(), lower.begin(),
+                   [](unsigned char c) { return std::tolower(c); });
+    std::string lower_id = doc.id;
+    std::transform(lower_id.begin(), lower_id.end(), lower_id.begin(),
+                   [](unsigned char c) { return std::tolower(c); });
+    if (lower_id.find(lower) != std::string::npos) return true;
+    return search_in_json(doc.data(), lower);
+}
+
+// Resolve the filter LHS value, honoring the same special-cased struct
+// members memory_store uses (_id, _created_at, _updated_at, _version),
+// the top-level fast path (no dot in name -> doc.field()), and the
+// dotted-path slow path (descends doc.data()).
+std::optional<nlohmann::json>
+resolve_filter_value(const smartbotic::database::Document& doc,
+                     const std::string& field) {
+    if (field == "_created_at") return nlohmann::json(doc.createdAt);
+    if (field == "_updated_at") return nlohmann::json(doc.updatedAt);
+    if (field == "_id")         return nlohmann::json(doc.id);
+    if (field == "_version")    return nlohmann::json(doc.version);
+    if (field.find('.') == std::string::npos) {
+        return doc.field(field);
+    }
+    return get_json_path(doc.data(), field);
+}
+
+bool matches_filters(const smartbotic::database::Document& doc,
+                     const std::vector<smartbotic::database::Filter>& filters) {
+    using smartbotic::database::FilterOp;
+    for (const auto& filter : filters) {
+        std::optional<nlohmann::json> value = resolve_filter_value(doc, filter.field);
+        switch (filter.op) {
+            case FilterOp::EXISTS:
+                if (value.has_value() != filter.value.get<bool>()) return false;
+                break;
+            case FilterOp::EQ:
+            case FilterOp::NE:
+            case FilterOp::GT:
+            case FilterOp::GTE:
+            case FilterOp::LT:
+            case FilterOp::LTE:
+                if (!value) return false;
+                if (!compare_json(*value, filter.op, filter.value)) return false;
+                break;
+            case FilterOp::IN: {
+                if (!value || !filter.value.is_array()) return false;
+                bool found = false;
+                for (const auto& v : filter.value) {
+                    if (*value == v) { found = true; break; }
+                }
+                if (!found) return false;
+                break;
+            }
+            case FilterOp::CONTAINS: {
+                if (!value || !value->is_array()) return false;
+                bool found = false;
+                for (const auto& v : *value) {
+                    if (v == filter.value) { found = true; break; }
+                }
+                if (!found) return false;
+                break;
+            }
+            case FilterOp::REGEX:
+                if (!value || !value->is_string() || !filter.value.is_string()) return false;
+                try {
+                    std::regex re(filter.value.get<std::string>());
+                    if (!std::regex_search(value->get<std::string>(), re)) return false;
+                } catch (const std::regex_error&) {
+                    return false;
+                }
+                break;
+            case FilterOp::SEARCH:
+                if (!filter.value.is_string()) return false;
+                if (!matches_search(doc, filter.value.get<std::string>())) return false;
+                break;
+        }
+    }
+    return true;
+}
+
+void sort_documents(std::vector<smartbotic::database::Document>& docs,
+                    const smartbotic::database::Sort& sort) {
+    std::sort(docs.begin(), docs.end(),
+        [&sort](const smartbotic::database::Document& a,
+                const smartbotic::database::Document& b) {
+            std::optional<nlohmann::json> va, vb;
+            if (sort.field == "_id") {
+                va = nlohmann::json(a.id);
+                vb = nlohmann::json(b.id);
+            } else if (sort.field == "_created_at") {
+                va = nlohmann::json(a.createdAt);
+                vb = nlohmann::json(b.createdAt);
+            } else if (sort.field == "_updated_at") {
+                va = nlohmann::json(a.updatedAt);
+                vb = nlohmann::json(b.updatedAt);
+            } else if (sort.field.find('.') == std::string::npos) {
+                va = a.field(sort.field);
+                vb = b.field(sort.field);
+            } else {
+                va = get_json_path(a.data(), sort.field);
+                vb = get_json_path(b.data(), sort.field);
+            }
+            if (!va && !vb) {
+                return sort.descending ? a.id > b.id : a.id < b.id;
+            }
+            if (!va) return sort.descending;
+            if (!vb) return !sort.descending;
+            if (*va == *vb) {
+                return sort.descending ? a.id > b.id : a.id < b.id;
+            }
+            bool less = *va < *vb;
+            return sort.descending ? !less : less;
+        });
+}
+
+// Project a JSON object onto the include-list. Empty include -> identity.
+// Dot-notation supported; arrays are not descended into (matches the
+// service/src/views/projection.cpp semantics).
+//
+// Note: v1.x's MemoryStore::find ignores query.projection entirely; the
+// projection plumbing only fires for views. Stage 2 honours the projection
+// field on Query directly so it's useful to callers that don't go through
+// a view. Include-only — exclude is reserved for views.
+void apply_projection_inplace(nlohmann::json& doc,
+                              const std::vector<std::string>& include) {
+    if (include.empty()) return;
+    if (!doc.is_object()) return;
+
+    auto split_path = [](const std::string& path) {
+        std::vector<std::string> parts;
+        std::stringstream ss(path);
+        std::string part;
+        while (std::getline(ss, part, '.')) {
+            if (!part.empty()) parts.push_back(part);
+        }
+        return parts;
+    };
+
+    std::function<void(const nlohmann::json&, nlohmann::json&,
+                       const std::vector<std::string>&, size_t)> copy_path;
+    copy_path = [&copy_path](const nlohmann::json& src, nlohmann::json& dst,
+                              const std::vector<std::string>& parts,
+                              size_t idx) {
+        if (idx >= parts.size()) return;
+        const std::string& key = parts[idx];
+        if (!src.is_object() || !src.contains(key)) return;
+        if (idx == parts.size() - 1) {
+            dst[key] = src[key];
+            return;
+        }
+        if (!src[key].is_object()) {
+            dst[key] = src[key];
+            return;
+        }
+        if (!dst.contains(key) || !dst[key].is_object()) {
+            dst[key] = nlohmann::json::object();
+        }
+        copy_path(src[key], dst[key], parts, idx + 1);
+    };
+
+    nlohmann::json result = nlohmann::json::object();
+    for (const auto& path : include) {
+        auto parts = split_path(path);
+        if (parts.empty()) continue;
+        copy_path(doc, result, parts, 0);
+    }
+    doc = std::move(result);
 }
 
-unsigned int LmdbDocumentStore::open_for_write(WriteTxn&, std::string_view) {
-    throw std::runtime_error(
-        "LmdbDocumentStore::open_for_write not implemented (Task 2.2)");
+}  // namespace
+
+// -------------------------------------------------------------------------
+// LmdbDocumentStore implementation
+// -------------------------------------------------------------------------
+
+LmdbDocumentStore::LmdbDocumentStore(LmdbEnv& env) noexcept : env_(env) {}
+
+unsigned int LmdbDocumentStore::open_for_write(WriteTxn& wtxn,
+                                                std::string_view collection) {
+    std::string key(collection);
+    {
+        std::lock_guard<std::mutex> lock(cache_mutex_);
+        auto it = dbi_cache_.find(key);
+        if (it != dbi_cache_.end()) return it->second;
+    }
+    MDB_dbi raw_dbi = 0;
+    std::string name = to_cstr(collection);
+    mdb_check(mdb_dbi_open(wtxn.raw(), name.c_str(), MDB_CREATE, &raw_dbi),
+              "dbi_open (collection create)");
+    {
+        std::lock_guard<std::mutex> lock(cache_mutex_);
+        dbi_cache_[key] = raw_dbi;
+    }
+    return raw_dbi;
 }
 
 std::optional<unsigned int>
-LmdbDocumentStore::try_open_for_read(ReadTxn&, std::string_view) {
-    throw std::runtime_error(
-        "LmdbDocumentStore::try_open_for_read not implemented (Task 2.2)");
+LmdbDocumentStore::try_open_for_read(ReadTxn& rtxn,
+                                      std::string_view collection) {
+    std::string key(collection);
+    {
+        std::lock_guard<std::mutex> lock(cache_mutex_);
+        auto it = dbi_cache_.find(key);
+        if (it != dbi_cache_.end()) return it->second;
+    }
+    MDB_dbi raw_dbi = 0;
+    std::string name = to_cstr(collection);
+    int rc = mdb_dbi_open(rtxn.raw(), name.c_str(), 0, &raw_dbi);
+    if (rc == MDB_NOTFOUND) return std::nullopt;
+    if (rc != MDB_SUCCESS) throw_mdb(rc, "dbi_open (collection read)");
+    {
+        std::lock_guard<std::mutex> lock(cache_mutex_);
+        dbi_cache_[key] = raw_dbi;
+    }
+    return raw_dbi;
+}
+
+void LmdbDocumentStore::put(std::string_view collection,
+                             std::string_view id,
+                             const smartbotic::database::Document& doc) {
+    std::string payload = encode_document(doc);
+    WriteTxn wtxn(env_);
+    unsigned int dbi = open_for_write(wtxn, collection);
+    MDB_val k = to_val(id);
+    MDB_val v = to_val(payload);
+    mdb_check(mdb_put(wtxn.raw(), dbi, &k, &v, 0), "put");
+    wtxn.commit();
+}
+
+std::optional<smartbotic::database::Document>
+LmdbDocumentStore::get(std::string_view collection, std::string_view id) {
+    ReadTxn rtxn(env_);
+    auto dbi_opt = try_open_for_read(rtxn, collection);
+    if (!dbi_opt) return std::nullopt;
+    MDB_val k = to_val(id);
+    MDB_val v{0, nullptr};
+    int rc = mdb_get(rtxn.raw(), *dbi_opt, &k, &v);
+    if (rc == MDB_NOTFOUND) return std::nullopt;
+    if (rc != MDB_SUCCESS) throw_mdb(rc, "get");
+    // Copy out of the mmap before the txn closes.
+    return decode_document(to_sv(v));
+}
+
+bool LmdbDocumentStore::del(std::string_view collection, std::string_view id) {
+    // Open as a write txn unconditionally so we have MDB_CREATE available
+    // if the collection doesn't exist yet — but in that case there's
+    // nothing to delete; just probe with a read txn first to avoid
+    // accidentally creating an empty sub-db on a no-op delete.
+    {
+        ReadTxn rtxn(env_);
+        auto dbi_opt = try_open_for_read(rtxn, collection);
+        if (!dbi_opt) return false;
+    }
+
+    WriteTxn wtxn(env_);
+    unsigned int dbi = open_for_write(wtxn, collection);
+    MDB_val k = to_val(id);
+    int rc = mdb_del(wtxn.raw(), dbi, &k, nullptr);
+    if (rc == MDB_NOTFOUND) {
+        wtxn.commit();
+        return false;
+    }
+    if (rc != MDB_SUCCESS) throw_mdb(rc, "del");
+    wtxn.commit();
+    return true;
+}
+
+uint64_t LmdbDocumentStore::count(std::string_view collection) {
+    ReadTxn rtxn(env_);
+    auto dbi_opt = try_open_for_read(rtxn, collection);
+    if (!dbi_opt) return 0;
+    MDB_stat st{};
+    mdb_check(mdb_stat(rtxn.raw(), *dbi_opt, &st), "stat");
+    return st.ms_entries;
+}
+
+ScanResult LmdbDocumentStore::scan(std::string_view collection,
+                                    const smartbotic::database::Query& query) {
+    ScanResult result;
+    ReadTxn rtxn(env_);
+    auto dbi_opt = try_open_for_read(rtxn, collection);
+    if (!dbi_opt) return result;  // empty
+
+    // Cursor-iterate the sub-db.
+    MDB_cursor* cursor = nullptr;
+    mdb_check(mdb_cursor_open(rtxn.raw(), *dbi_opt, &cursor), "cursor_open");
+    struct CursorGuard {
+        MDB_cursor* c;
+        ~CursorGuard() { if (c) mdb_cursor_close(c); }
+    } guard{cursor};
+
+    std::vector<smartbotic::database::Document> matches;
+    MDB_val k{0, nullptr};
+    MDB_val v{0, nullptr};
+    int rc = mdb_cursor_get(cursor, &k, &v, MDB_FIRST);
+    while (rc == MDB_SUCCESS) {
+        smartbotic::database::Document doc = decode_document(to_sv(v));
+        if (matches_filters(doc, query.filters)) {
+            matches.push_back(std::move(doc));
+        }
+        rc = mdb_cursor_get(cursor, &k, &v, MDB_NEXT);
+    }
+    if (rc != MDB_NOTFOUND && rc != MDB_SUCCESS) {
+        throw_mdb(rc, "cursor_get");
+    }
+
+    // Sort.
+    if (query.sort && !query.sort->field.empty()) {
+        sort_documents(matches, *query.sort);
+    }
+
+    // Pagination.
+    result.total_matched = matches.size();
+    uint64_t start = std::min<uint64_t>(query.offset, matches.size());
+    uint64_t end   = std::min<uint64_t>(
+        static_cast<uint64_t>(query.offset) + query.limit, matches.size());
+    result.documents.reserve(end - start);
+    for (uint64_t i = start; i < end; ++i) {
+        result.documents.push_back(std::move(matches[i]));
+    }
+    result.has_more = end < result.total_matched;
+
+    // Projection — applied AFTER pagination so we don't pay encode cost
+    // for documents we're not returning.
+    if (!query.projection.empty()) {
+        for (auto& d : result.documents) {
+            nlohmann::json data = d.data();   // materialise
+            apply_projection_inplace(data, query.projection);
+            d.set_data(data);
+        }
+    }
+
+    return result;
+}
+
+std::vector<std::string> LmdbDocumentStore::list_collections() {
+    // The unnamed root sub-db lists named sub-dbs. Open it with name = nullptr.
+    std::vector<std::string> names;
+    ReadTxn rtxn(env_);
+    MDB_dbi main_dbi = 0;
+    int rc = mdb_dbi_open(rtxn.raw(), nullptr, 0, &main_dbi);
+    if (rc == MDB_NOTFOUND) return names;
+    if (rc != MDB_SUCCESS) throw_mdb(rc, "dbi_open (main for list)");
+
+    MDB_cursor* cursor = nullptr;
+    mdb_check(mdb_cursor_open(rtxn.raw(), main_dbi, &cursor), "cursor_open (main)");
+    struct CursorGuard {
+        MDB_cursor* c;
+        ~CursorGuard() { if (c) mdb_cursor_close(c); }
+    } guard{cursor};
+
+    MDB_val k{0, nullptr};
+    MDB_val v{0, nullptr};
+    int crc = mdb_cursor_get(cursor, &k, &v, MDB_FIRST);
+    while (crc == MDB_SUCCESS) {
+        std::string_view name(static_cast<const char*>(k.mv_data), k.mv_size);
+        if (!is_system_subdb(name)) {
+            // Sanity: the unnamed root holds entries for non-sub-db keys too
+            // when callers stash records there. We're a fresh v2.0 substrate
+            // and never put plain keys at the root, so every entry is a
+            // sub-db descriptor. Belt-and-braces: if MDB_INTEGERKEY style
+            // entries leak in here later, the resulting string-view will
+            // not start with '_' and will appear as a collection name —
+            // that's acceptable for the scaffolding phase.
+            names.emplace_back(name);
+        }
+        crc = mdb_cursor_get(cursor, &k, &v, MDB_NEXT);
+    }
+    if (crc != MDB_NOTFOUND && crc != MDB_SUCCESS) {
+        throw_mdb(crc, "cursor_get (main list)");
+    }
+
+    std::sort(names.begin(), names.end());
+    return names;
+}
+
+bool LmdbDocumentStore::drop_collection(std::string_view collection) {
+    // Probe with a read txn first; if the sub-db doesn't exist, return
+    // false without opening a write txn at all.
+    {
+        ReadTxn rtxn(env_);
+        auto dbi_opt = try_open_for_read(rtxn, collection);
+        if (!dbi_opt) return false;
+    }
+
+    WriteTxn wtxn(env_);
+    unsigned int dbi = open_for_write(wtxn, collection);
+    // mdb_drop with del=1 removes the sub-db itself (not just its entries).
+    mdb_check(mdb_drop(wtxn.raw(), dbi, 1), "drop");
+    wtxn.commit();
+
+    // Drop invalidates the dbi handle — evict from cache so subsequent ops
+    // on this collection re-open with MDB_CREATE.
+    {
+        std::lock_guard<std::mutex> lock(cache_mutex_);
+        dbi_cache_.erase(std::string(collection));
+    }
+    return true;
 }
 
 }  // namespace smartbotic::db::storage