|
@@ -0,0 +1,1225 @@
|
|
|
|
|
+#include "memory_store.hpp"
|
|
|
|
|
+
|
|
|
|
|
+#include <spdlog/spdlog.h>
|
|
|
|
|
+
|
|
|
|
|
+#include <algorithm>
|
|
|
|
|
+#include <random>
|
|
|
|
|
+#include <regex>
|
|
|
|
|
+#include <sstream>
|
|
|
|
|
+#include <iomanip>
|
|
|
|
|
+
|
|
|
|
|
+namespace smartbotic::database {
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::MemoryStore(Config config)
|
|
|
|
|
+ : config_(std::move(config)) {
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::~MemoryStore() {
|
|
|
|
|
+ stop();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::start() {
|
|
|
|
|
+ if (running_.exchange(true)) {
|
|
|
|
|
+ return; // Already running
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Start expiration thread
|
|
|
|
|
+ expirationThread_ = std::thread(&MemoryStore::expirationLoop, this);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::stop() {
|
|
|
|
|
+ if (!running_.exchange(false)) {
|
|
|
|
|
+ return; // Already stopped
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (expirationThread_.joinable()) {
|
|
|
|
|
+ expirationThread_.join();
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::setEventCallback(EventCallback callback) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(callbackMutex_);
|
|
|
|
|
+ eventCallback_ = std::move(callback);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::setPersistCallback(PersistCallback callback) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(callbackMutex_);
|
|
|
|
|
+ persistCallback_ = std::move(callback);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Collection Management =====
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::createCollection(const std::string& name, const CollectionOptions& options) {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ if (collections_.find(name) != collections_.end()) {
|
|
|
|
|
+ return false; // Already exists
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto coll = std::make_unique<CollectionData>();
|
|
|
|
|
+ coll->options = options;
|
|
|
|
|
+ coll->createdAt = currentTimeMs();
|
|
|
|
|
+ coll->updatedAt = coll->createdAt;
|
|
|
|
|
+
|
|
|
|
|
+ collections_[name] = std::move(coll);
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalCollections++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::dropCollection(const std::string& name) {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = collections_.find(name);
|
|
|
|
|
+ if (it == collections_.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t docCount = 0;
|
|
|
|
|
+ {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> collLock(it->second->mutex);
|
|
|
|
|
+ docCount = it->second->documents.size();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ collections_.erase(it);
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalCollections--;
|
|
|
|
|
+ stats_.totalDocuments -= docCount;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::string> MemoryStore::listCollections() const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ std::vector<std::string> names;
|
|
|
|
|
+ names.reserve(collections_.size());
|
|
|
|
|
+ for (const auto& [name, _] : collections_) {
|
|
|
|
|
+ names.push_back(name);
|
|
|
|
|
+ }
|
|
|
|
|
+ return names;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::optional<CollectionInfo> MemoryStore::getCollectionInfo(const std::string& name) const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> globalLock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = collections_.find(name);
|
|
|
|
|
+ if (it == collections_.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> collLock(it->second->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ CollectionInfo info;
|
|
|
|
|
+ info.name = name;
|
|
|
|
|
+ info.documentCount = it->second->documents.size();
|
|
|
|
|
+ info.options = it->second->options;
|
|
|
|
|
+ info.createdAt = it->second->createdAt;
|
|
|
|
|
+ info.updatedAt = it->second->updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ // Estimate size
|
|
|
|
|
+ for (const auto& [id, doc] : it->second->documents) {
|
|
|
|
|
+ info.sizeBytes += estimateDocumentSize(doc);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return info;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::collectionExists(const std::string& name) const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+ return collections_.find(name) != collections_.end();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Document CRUD =====
|
|
|
|
|
+
|
|
|
|
|
+std::string MemoryStore::insert(const std::string& collection, Document doc) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Generate ID if not provided
|
|
|
|
|
+ if (doc.id.empty()) {
|
|
|
|
|
+ if (coll->options.autoCreateId) {
|
|
|
|
|
+ doc.id = generateId();
|
|
|
|
|
+ } else {
|
|
|
|
|
+ throw std::runtime_error("Document ID is required");
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Check if ID already exists
|
|
|
|
|
+ if (coll->documents.find(doc.id) != coll->documents.end()) {
|
|
|
|
|
+ throw std::runtime_error("Document with ID '" + doc.id + "' already exists");
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Set metadata
|
|
|
|
|
+ doc.collection = collection;
|
|
|
|
|
+ doc.version = 1;
|
|
|
|
|
+ doc.createdAt = currentTimeMs();
|
|
|
|
|
+ doc.updatedAt = doc.createdAt;
|
|
|
|
|
+ doc.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ // Apply default TTL if not set
|
|
|
|
|
+ if (doc.expiresAt == 0 && coll->options.defaultTtlSeconds > 0) {
|
|
|
|
|
+ doc.setTtlSeconds(coll->options.defaultTtlSeconds);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Add to expiration index
|
|
|
|
|
+ if (doc.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, doc.id, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::string docId = doc.id;
|
|
|
|
|
+ coll->documents[docId] = doc;
|
|
|
|
|
+ coll->updatedAt = doc.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ stats_.insertCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks
|
|
|
|
|
+ emitPersist(collection, docId, doc, EventType::INSERT);
|
|
|
|
|
+ emitEvent(EventType::INSERT, collection, docId, doc.data);
|
|
|
|
|
+
|
|
|
|
|
+ return docId;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::optional<Document> MemoryStore::get(const std::string& collection, const std::string& id) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Check expiration
|
|
|
|
|
+ if (it->second.isExpired()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return it->second;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::update(const std::string& collection, const std::string& id, const Document& doc) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Remove from old expiration index
|
|
|
|
|
+ if (it->second.expiresAt > 0) {
|
|
|
|
|
+ removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Update document
|
|
|
|
|
+ Document updated = doc;
|
|
|
|
|
+ updated.id = id;
|
|
|
|
|
+ updated.collection = collection;
|
|
|
|
|
+ updated.version = it->second.version + 1;
|
|
|
|
|
+ updated.createdAt = it->second.createdAt;
|
|
|
|
|
+ updated.createdBy = it->second.createdBy; // Preserve original creator
|
|
|
|
|
+ updated.updatedAt = currentTimeMs();
|
|
|
|
|
+ updated.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ // Add to new expiration index
|
|
|
|
|
+ if (updated.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, id, updated.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ it->second = updated;
|
|
|
|
|
+ coll->updatedAt = updated.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.updateCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks
|
|
|
|
|
+ emitPersist(collection, id, updated, EventType::UPDATE);
|
|
|
|
|
+ emitEvent(EventType::UPDATE, collection, id, updated.data);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Generate ID if not provided
|
|
|
|
|
+ if (doc.id.empty()) {
|
|
|
|
|
+ if (coll->options.autoCreateId) {
|
|
|
|
|
+ doc.id = generateId();
|
|
|
|
|
+ } else {
|
|
|
|
|
+ throw std::runtime_error("Document ID is required");
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::string docId = doc.id;
|
|
|
|
|
+ bool isInsert = false;
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(docId);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ // Insert
|
|
|
|
|
+ isInsert = true;
|
|
|
|
|
+ doc.collection = collection;
|
|
|
|
|
+ doc.version = 1;
|
|
|
|
|
+ doc.createdAt = currentTimeMs();
|
|
|
|
|
+ doc.updatedAt = doc.createdAt;
|
|
|
|
|
+ doc.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ if (doc.expiresAt == 0 && coll->options.defaultTtlSeconds > 0) {
|
|
|
|
|
+ doc.setTtlSeconds(coll->options.defaultTtlSeconds);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (doc.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, docId, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->documents[docId] = doc;
|
|
|
|
|
+ } else {
|
|
|
|
|
+ // Update
|
|
|
|
|
+ if (it->second.expiresAt > 0) {
|
|
|
|
|
+ removeFromExpirationIndex(*coll, docId, it->second.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ doc.id = docId;
|
|
|
|
|
+ doc.collection = collection;
|
|
|
|
|
+ doc.version = it->second.version + 1;
|
|
|
|
|
+ doc.createdAt = it->second.createdAt;
|
|
|
|
|
+ doc.createdBy = it->second.createdBy; // Preserve original creator
|
|
|
|
|
+ doc.updatedAt = currentTimeMs();
|
|
|
|
|
+ doc.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ if (doc.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, docId, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ it->second = doc;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->updatedAt = doc.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ if (isInsert) {
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ stats_.insertCount++;
|
|
|
|
|
+ } else {
|
|
|
|
|
+ stats_.updateCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks
|
|
|
|
|
+ emitPersist(collection, docId, doc, isInsert ? EventType::INSERT : EventType::UPDATE);
|
|
|
|
|
+ emitEvent(isInsert ? EventType::INSERT : EventType::UPDATE, collection, docId, doc.data);
|
|
|
|
|
+
|
|
|
|
|
+ return docId;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::remove(const std::string& collection, const std::string& id) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Remove from expiration index
|
|
|
|
|
+ if (it->second.expiresAt > 0) {
|
|
|
|
|
+ removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->documents.erase(it);
|
|
|
|
|
+ coll->updatedAt = currentTimeMs();
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments--;
|
|
|
|
|
+ stats_.deleteCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks
|
|
|
|
|
+ emitPersist(collection, id, std::nullopt, EventType::DELETE);
|
|
|
|
|
+ emitEvent(EventType::DELETE, collection, id);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::exists(const std::string& collection, const std::string& id) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return !it->second.isExpired();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::updateIfVersion(const std::string& collection, const std::string& id,
|
|
|
|
|
+ const Document& doc, uint64_t expectedVersion) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(id);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (it->second.version != expectedVersion) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Remove from old expiration index
|
|
|
|
|
+ if (it->second.expiresAt > 0) {
|
|
|
|
|
+ removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Update document
|
|
|
|
|
+ Document updated = doc;
|
|
|
|
|
+ updated.id = id;
|
|
|
|
|
+ updated.collection = collection;
|
|
|
|
|
+ updated.version = expectedVersion + 1;
|
|
|
|
|
+ updated.createdAt = it->second.createdAt;
|
|
|
|
|
+ updated.createdBy = it->second.createdBy; // Preserve original creator
|
|
|
|
|
+ updated.updatedAt = currentTimeMs();
|
|
|
|
|
+ updated.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ // Add to new expiration index
|
|
|
|
|
+ if (updated.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, id, updated.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ it->second = updated;
|
|
|
|
|
+ coll->updatedAt = updated.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.updateCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks
|
|
|
|
|
+ emitPersist(collection, id, updated, EventType::UPDATE);
|
|
|
|
|
+ emitEvent(EventType::UPDATE, collection, id, updated.data);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Query Operations =====
|
|
|
|
|
+
|
|
|
|
|
+QueryResult MemoryStore::find(const std::string& collection, const Query& query) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Update query count
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ const_cast<MemoryStore*>(this)->stats_.queryCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Collect matching documents
|
|
|
|
|
+ std::vector<Document> matches;
|
|
|
|
|
+ for (const auto& [id, doc] : coll->documents) {
|
|
|
|
|
+ if (doc.isExpired()) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (matchesFilters(doc, query.filters)) {
|
|
|
|
|
+ matches.push_back(doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Sort if requested
|
|
|
|
|
+ if (query.sort && !query.sort->field.empty()) {
|
|
|
|
|
+ SPDLOG_DEBUG("Sorting {} documents by field '{}' descending={}",
|
|
|
|
|
+ matches.size(), query.sort->field, query.sort->descending);
|
|
|
|
|
+ std::sort(matches.begin(), matches.end(),
|
|
|
|
|
+ [&query](const Document& a, const Document& b) {
|
|
|
|
|
+ std::optional<nlohmann::json> valA, valB;
|
|
|
|
|
+
|
|
|
|
|
+ // Handle special fields that are stored as struct members, not in data
|
|
|
|
|
+ if (query.sort->field == "_id") {
|
|
|
|
|
+ valA = nlohmann::json(a.id);
|
|
|
|
|
+ valB = nlohmann::json(b.id);
|
|
|
|
|
+ } else if (query.sort->field == "_created_at") {
|
|
|
|
|
+ valA = nlohmann::json(a.createdAt);
|
|
|
|
|
+ valB = nlohmann::json(b.createdAt);
|
|
|
|
|
+ } else if (query.sort->field == "_updated_at") {
|
|
|
|
|
+ valA = nlohmann::json(a.updatedAt);
|
|
|
|
|
+ valB = nlohmann::json(b.updatedAt);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ valA = getJsonPath(a.data, query.sort->field);
|
|
|
|
|
+ valB = getJsonPath(b.data, query.sort->field);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!valA && !valB) {
|
|
|
|
|
+ // Both null - use ID as tiebreaker for stable ordering
|
|
|
|
|
+ return query.sort->descending ? a.id > b.id : a.id < b.id;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!valA) return query.sort->descending;
|
|
|
|
|
+ if (!valB) return !query.sort->descending;
|
|
|
|
|
+
|
|
|
|
|
+ // Compare primary sort values
|
|
|
|
|
+ if (*valA == *valB) {
|
|
|
|
|
+ // Equal values - use ID as tiebreaker for stable ordering
|
|
|
|
|
+ return query.sort->descending ? a.id > b.id : a.id < b.id;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ bool less = *valA < *valB;
|
|
|
|
|
+ return query.sort->descending ? !less : less;
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // Log first few document timestamps after sorting
|
|
|
|
|
+ if (matches.size() > 0) {
|
|
|
|
|
+ std::string debugInfo = "First documents after sort: ";
|
|
|
|
|
+ size_t count = std::min<size_t>(5, matches.size());
|
|
|
|
|
+ for (size_t i = 0; i < count; ++i) {
|
|
|
|
|
+ if (query.sort->field == "_created_at") {
|
|
|
|
|
+ debugInfo += "[" + matches[i].id + ": " + std::to_string(matches[i].createdAt) + "] ";
|
|
|
|
|
+ } else if (query.sort->field == "_updated_at") {
|
|
|
|
|
+ debugInfo += "[" + matches[i].id + ": " + std::to_string(matches[i].updatedAt) + "] ";
|
|
|
|
|
+ } else {
|
|
|
|
|
+ auto val = getJsonPath(matches[i].data, query.sort->field);
|
|
|
|
|
+ debugInfo += "[" + matches[i].id + ": " + (val ? val->dump() : "null") + "] ";
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ SPDLOG_DEBUG("{}", debugInfo);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Build result with pagination
|
|
|
|
|
+ QueryResult result;
|
|
|
|
|
+ result.totalCount = matches.size();
|
|
|
|
|
+
|
|
|
|
|
+ uint32_t start = std::min(query.offset, static_cast<uint32_t>(matches.size()));
|
|
|
|
|
+ uint32_t end = std::min(query.offset + query.limit, static_cast<uint32_t>(matches.size()));
|
|
|
|
|
+
|
|
|
|
|
+ for (uint32_t i = start; i < end; ++i) {
|
|
|
|
|
+ result.documents.push_back(matches[i]);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ result.hasMore = end < matches.size();
|
|
|
|
|
+
|
|
|
|
|
+ return result;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::count(const std::string& collection, const std::vector<Filter>& filters) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return 0;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ if (filters.empty()) {
|
|
|
|
|
+ // Count non-expired documents
|
|
|
|
|
+ uint64_t count = 0;
|
|
|
|
|
+ for (const auto& [id, doc] : coll->documents) {
|
|
|
|
|
+ if (!doc.isExpired()) {
|
|
|
|
|
+ count++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return count;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t count = 0;
|
|
|
|
|
+ for (const auto& [id, doc] : coll->documents) {
|
|
|
|
|
+ if (!doc.isExpired() && matchesFilters(doc, filters)) {
|
|
|
|
|
+ count++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return count;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Set Operations =====
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::setAdd(const std::string& collection, const std::string& setId, const std::string& member) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(setId);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ // Create new set document
|
|
|
|
|
+ Document doc;
|
|
|
|
|
+ doc.id = setId;
|
|
|
|
|
+ doc.collection = collection;
|
|
|
|
|
+ doc.data = nlohmann::json::object();
|
|
|
|
|
+ doc.data["members"] = nlohmann::json::array({member});
|
|
|
|
|
+ doc.version = 1;
|
|
|
|
|
+ doc.createdAt = currentTimeMs();
|
|
|
|
|
+ doc.updatedAt = doc.createdAt;
|
|
|
|
|
+ doc.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ if (coll->options.defaultTtlSeconds > 0) {
|
|
|
|
|
+ doc.setTtlSeconds(coll->options.defaultTtlSeconds);
|
|
|
|
|
+ addToExpirationIndex(*coll, setId, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->documents[setId] = doc;
|
|
|
|
|
+ coll->updatedAt = doc.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ stats_.insertCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ emitPersist(collection, setId, doc, EventType::INSERT);
|
|
|
|
|
+ emitEvent(EventType::INSERT, collection, setId, doc.data);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Check if member already exists
|
|
|
|
|
+ auto& members = it->second.data["members"];
|
|
|
|
|
+ if (!members.is_array()) {
|
|
|
|
|
+ members = nlohmann::json::array();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& m : members) {
|
|
|
|
|
+ if (m.is_string() && m.get<std::string>() == member) {
|
|
|
|
|
+ return false; // Already exists
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ members.push_back(member);
|
|
|
|
|
+ it->second.version++;
|
|
|
|
|
+ it->second.updatedAt = currentTimeMs();
|
|
|
|
|
+ it->second.nodeId = config_.nodeId;
|
|
|
|
|
+ coll->updatedAt = it->second.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ Document updated = it->second;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.updateCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ emitPersist(collection, setId, updated, EventType::UPDATE);
|
|
|
|
|
+ emitEvent(EventType::UPDATE, collection, setId, updated.data);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::setRemove(const std::string& collection, const std::string& setId, const std::string& member) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(setId);
|
|
|
|
|
+ if (it == coll->documents.end()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto& members = it->second.data["members"];
|
|
|
|
|
+ if (!members.is_array()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ bool found = false;
|
|
|
|
|
+ auto newMembers = nlohmann::json::array();
|
|
|
|
|
+ for (const auto& m : members) {
|
|
|
|
|
+ if (m.is_string() && m.get<std::string>() == member) {
|
|
|
|
|
+ found = true;
|
|
|
|
|
+ } else {
|
|
|
|
|
+ newMembers.push_back(m);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!found) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ it->second.data["members"] = newMembers;
|
|
|
|
|
+ it->second.version++;
|
|
|
|
|
+ it->second.updatedAt = currentTimeMs();
|
|
|
|
|
+ it->second.nodeId = config_.nodeId;
|
|
|
|
|
+ coll->updatedAt = it->second.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ Document updated = it->second;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.updateCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ emitPersist(collection, setId, updated, EventType::UPDATE);
|
|
|
|
|
+ emitEvent(EventType::UPDATE, collection, setId, updated.data);
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::string> MemoryStore::setMembers(const std::string& collection, const std::string& setId) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(setId);
|
|
|
|
|
+ if (it == coll->documents.end() || it->second.isExpired()) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const auto& members = it->second.data["members"];
|
|
|
|
|
+ if (!members.is_array()) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::vector<std::string> result;
|
|
|
|
|
+ for (const auto& m : members) {
|
|
|
|
|
+ if (m.is_string()) {
|
|
|
|
|
+ result.push_back(m.get<std::string>());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return result;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::setIsMember(const std::string& collection, const std::string& setId, const std::string& member) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ auto it = coll->documents.find(setId);
|
|
|
|
|
+ if (it == coll->documents.end() || it->second.isExpired()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const auto& members = it->second.data["members"];
|
|
|
|
|
+ if (!members.is_array()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& m : members) {
|
|
|
|
|
+ if (m.is_string() && m.get<std::string>() == member) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Bulk Operations =====
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::string> MemoryStore::bulkInsert(const std::string& collection, std::vector<Document> docs) {
|
|
|
|
|
+ std::vector<std::string> ids;
|
|
|
|
|
+ ids.reserve(docs.size());
|
|
|
|
|
+
|
|
|
|
|
+ for (auto& doc : docs) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ ids.push_back(insert(collection, std::move(doc)));
|
|
|
|
|
+ } catch (const std::exception&) {
|
|
|
|
|
+ // Continue with other documents
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return ids;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::bulkDelete(const std::string& collection, const std::vector<std::string>& ids) {
|
|
|
|
|
+ uint64_t deleted = 0;
|
|
|
|
|
+ for (const auto& id : ids) {
|
|
|
|
|
+ if (remove(collection, id)) {
|
|
|
|
|
+ deleted++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return deleted;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== TTL Management =====
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::expireDocuments() {
|
|
|
|
|
+ uint64_t expired = 0;
|
|
|
|
|
+ auto now = currentTimeMs();
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> globalLock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ for (auto& [collName, coll] : collections_) {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Find expired documents
|
|
|
|
|
+ std::vector<std::string> toExpire;
|
|
|
|
|
+ auto it = coll->expirationIndex.begin();
|
|
|
|
|
+ while (it != coll->expirationIndex.end() && it->first <= now) {
|
|
|
|
|
+ for (const auto& id : it->second) {
|
|
|
|
|
+ toExpire.push_back(id);
|
|
|
|
|
+ }
|
|
|
|
|
+ it = coll->expirationIndex.erase(it);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Remove expired documents
|
|
|
|
|
+ for (const auto& id : toExpire) {
|
|
|
|
|
+ auto docIt = coll->documents.find(id);
|
|
|
|
|
+ if (docIt != coll->documents.end()) {
|
|
|
|
|
+ coll->documents.erase(docIt);
|
|
|
|
|
+ expired++;
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks (unlock first to avoid deadlock)
|
|
|
|
|
+ collLock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments--;
|
|
|
|
|
+ stats_.expiredCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ emitPersist(collName, id, std::nullopt, EventType::EXPIRE);
|
|
|
|
|
+ emitEvent(EventType::EXPIRE, collName, id);
|
|
|
|
|
+
|
|
|
|
|
+ collLock.lock();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return expired;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Statistics =====
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::Stats MemoryStore::getStats() const {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(statsMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ Stats stats = stats_;
|
|
|
|
|
+
|
|
|
|
|
+ // Calculate estimated memory
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> globalLock(globalMutex_);
|
|
|
|
|
+ stats.estimatedMemoryBytes = 0;
|
|
|
|
|
+ for (const auto& [name, coll] : collections_) {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
|
|
+ for (const auto& [id, doc] : coll->documents) {
|
|
|
|
|
+ stats.estimatedMemoryBytes += estimateDocumentSize(doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return stats;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Snapshot/Recovery Support =====
|
|
|
|
|
+
|
|
|
|
|
+std::vector<Document> MemoryStore::getAllDocuments(const std::string& collection) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ std::vector<Document> docs;
|
|
|
|
|
+ docs.reserve(coll->documents.size());
|
|
|
|
|
+ for (const auto& [id, doc] : coll->documents) {
|
|
|
|
|
+ docs.push_back(doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ return docs;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::pair<std::string, CollectionOptions>> MemoryStore::getAllCollectionsWithOptions() const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ std::vector<std::pair<std::string, CollectionOptions>> result;
|
|
|
|
|
+ result.reserve(collections_.size());
|
|
|
|
|
+ for (const auto& [name, coll] : collections_) {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> collLock(coll->mutex);
|
|
|
|
|
+ result.emplace_back(name, coll->options);
|
|
|
|
|
+ }
|
|
|
|
|
+ return result;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::loadDocument(const std::string& collection, Document doc) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ doc.collection = collection;
|
|
|
|
|
+
|
|
|
|
|
+ // Add to expiration index
|
|
|
|
|
+ if (doc.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, doc.id, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->documents[doc.id] = std::move(doc);
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::clear() {
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+ collections_.clear();
|
|
|
|
|
+
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_ = Stats{};
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// ===== Private Methods =====
|
|
|
|
|
+
|
|
|
|
|
+std::string MemoryStore::generateId() const {
|
|
|
|
|
+ static std::random_device rd;
|
|
|
|
|
+ static std::mt19937_64 gen(rd());
|
|
|
|
|
+ static std::uniform_int_distribution<uint64_t> dis;
|
|
|
|
|
+
|
|
|
|
|
+ auto now = currentTimeMs();
|
|
|
|
|
+ uint64_t random = dis(gen);
|
|
|
|
|
+
|
|
|
|
|
+ std::ostringstream oss;
|
|
|
|
|
+ oss << std::hex << std::setfill('0')
|
|
|
|
|
+ << std::setw(12) << now
|
|
|
|
|
+ << std::setw(16) << random;
|
|
|
|
|
+
|
|
|
|
|
+ return oss.str();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::currentTimeMs() {
|
|
|
|
|
+ return static_cast<uint64_t>(
|
|
|
|
|
+ std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
|
|
+ std::chrono::system_clock::now().time_since_epoch()
|
|
|
|
|
+ ).count()
|
|
|
|
|
+ );
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::matchesFilters(const Document& doc, const std::vector<Filter>& filters) const {
|
|
|
|
|
+ for (const auto& filter : filters) {
|
|
|
|
|
+ auto value = getJsonPath(doc.data, 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 (!compareJson(*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:
|
|
|
|
|
+ // Full-text search across document ID and string fields
|
|
|
|
|
+ if (!filter.value.is_string()) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!matchesSearch(doc, filter.value.get<std::string>())) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return true;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::optional<nlohmann::json> MemoryStore::getJsonPath(const nlohmann::json& obj, const std::string& path) {
|
|
|
|
|
+ if (path.empty() || !obj.is_object()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Split path by '.'
|
|
|
|
|
+ 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::make_optional(*current);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::compareJson(const nlohmann::json& a, FilterOp op, const nlohmann::json& b) {
|
|
|
|
|
+ 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 MemoryStore::matchesSearch(const Document& doc, const std::string& searchTerm) const {
|
|
|
|
|
+ if (searchTerm.empty()) {
|
|
|
|
|
+ return true; // Empty search matches everything
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Convert search term to lowercase for case-insensitive matching
|
|
|
|
|
+ std::string lowerSearchTerm = searchTerm;
|
|
|
|
|
+ std::transform(lowerSearchTerm.begin(), lowerSearchTerm.end(), lowerSearchTerm.begin(),
|
|
|
|
|
+ [](unsigned char c) { return std::tolower(c); });
|
|
|
|
|
+
|
|
|
|
|
+ // Check document ID
|
|
|
|
|
+ std::string lowerId = doc.id;
|
|
|
|
|
+ std::transform(lowerId.begin(), lowerId.end(), lowerId.begin(),
|
|
|
|
|
+ [](unsigned char c) { return std::tolower(c); });
|
|
|
|
|
+ if (lowerId.find(lowerSearchTerm) != std::string::npos) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Search in document data
|
|
|
|
|
+ return searchInJson(doc.data, lowerSearchTerm);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool MemoryStore::searchInJson(const nlohmann::json& obj, const std::string& lowerSearchTerm) {
|
|
|
|
|
+ if (obj.is_string()) {
|
|
|
|
|
+ std::string lowerValue = obj.get<std::string>();
|
|
|
|
|
+ std::transform(lowerValue.begin(), lowerValue.end(), lowerValue.begin(),
|
|
|
|
|
+ [](unsigned char c) { return std::tolower(c); });
|
|
|
|
|
+ return lowerValue.find(lowerSearchTerm) != std::string::npos;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (obj.is_object()) {
|
|
|
|
|
+ for (const auto& [key, value] : obj.items()) {
|
|
|
|
|
+ // Search in keys too
|
|
|
|
|
+ std::string lowerKey = key;
|
|
|
|
|
+ std::transform(lowerKey.begin(), lowerKey.end(), lowerKey.begin(),
|
|
|
|
|
+ [](unsigned char c) { return std::tolower(c); });
|
|
|
|
|
+ if (lowerKey.find(lowerSearchTerm) != std::string::npos) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ // Recursively search in values
|
|
|
|
|
+ if (searchInJson(value, lowerSearchTerm)) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (obj.is_array()) {
|
|
|
|
|
+ for (const auto& item : obj) {
|
|
|
|
|
+ if (searchInJson(item, lowerSearchTerm)) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // For numbers, convert to string and search
|
|
|
|
|
+ if (obj.is_number()) {
|
|
|
|
|
+ std::string numStr = obj.dump();
|
|
|
|
|
+ if (numStr.find(lowerSearchTerm) != std::string::npos) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return false;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::emitEvent(EventType type, const std::string& collection, const std::string& id,
|
|
|
|
|
+ const std::optional<nlohmann::json>& data) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(callbackMutex_);
|
|
|
|
|
+ if (eventCallback_) {
|
|
|
|
|
+ DatabaseEvent event;
|
|
|
|
|
+ event.type = type;
|
|
|
|
|
+ event.collection = collection;
|
|
|
|
|
+ event.documentId = id;
|
|
|
|
|
+ event.timestamp = currentTimeMs();
|
|
|
|
|
+ event.nodeId = config_.nodeId;
|
|
|
|
|
+ event.data = data;
|
|
|
|
|
+ eventCallback_(event);
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::emitPersist(const std::string& collection, const std::string& id,
|
|
|
|
|
+ const std::optional<Document>& doc, EventType eventType) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lock(callbackMutex_);
|
|
|
|
|
+ if (persistCallback_) {
|
|
|
|
|
+ persistCallback_(collection, id, doc, eventType);
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::addToExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt) {
|
|
|
|
|
+ coll.expirationIndex[expiresAt].insert(id);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::removeFromExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt) {
|
|
|
|
|
+ auto it = coll.expirationIndex.find(expiresAt);
|
|
|
|
|
+ if (it != coll.expirationIndex.end()) {
|
|
|
|
|
+ it->second.erase(id);
|
|
|
|
|
+ if (it->second.empty()) {
|
|
|
|
|
+ coll.expirationIndex.erase(it);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::estimateDocumentSize(const Document& doc) {
|
|
|
|
|
+ // Rough estimate: ID + collection + JSON dump size + metadata
|
|
|
|
|
+ return doc.id.size() + doc.collection.size() + doc.data.dump().size() + 100;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::expirationLoop() {
|
|
|
|
|
+ while (running_.load()) {
|
|
|
|
|
+ std::this_thread::sleep_for(
|
|
|
|
|
+ std::chrono::milliseconds(config_.expirationCheckIntervalMs)
|
|
|
|
|
+ );
|
|
|
|
|
+
|
|
|
|
|
+ if (running_.load()) {
|
|
|
|
|
+ expireDocuments();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::CollectionData* MemoryStore::getOrCreateCollection(const std::string& name) {
|
|
|
|
|
+ {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+ auto it = collections_.find(name);
|
|
|
|
|
+ if (it != collections_.end()) {
|
|
|
|
|
+ return it->second.get();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Create collection
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+
|
|
|
|
|
+ // Double-check after acquiring exclusive lock
|
|
|
|
|
+ auto it = collections_.find(name);
|
|
|
|
|
+ if (it != collections_.end()) {
|
|
|
|
|
+ return it->second.get();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto coll = std::make_unique<CollectionData>();
|
|
|
|
|
+ coll->createdAt = currentTimeMs();
|
|
|
|
|
+ coll->updatedAt = coll->createdAt;
|
|
|
|
|
+
|
|
|
|
|
+ auto* ptr = coll.get();
|
|
|
|
|
+ collections_[name] = std::move(coll);
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalCollections++;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return ptr;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+const MemoryStore::CollectionData* MemoryStore::getCollection(const std::string& name) const {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
|
|
+ auto it = collections_.find(name);
|
|
|
|
|
+ if (it != collections_.end()) {
|
|
|
|
|
+ return it->second.get();
|
|
|
|
|
+ }
|
|
|
|
|
+ return nullptr;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+} // namespace smartbotic::database
|