|
@@ -246,6 +246,7 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
|
|
|
addToExpirationIndex(*coll, id, updated.expiresAt);
|
|
addToExpirationIndex(*coll, id, updated.expiresAt);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
it->second = updated;
|
|
it->second = updated;
|
|
|
coll->updatedAt = updated.updatedAt;
|
|
coll->updatedAt = updated.updatedAt;
|
|
|
|
|
|
|
@@ -318,6 +319,7 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
|
|
|
addToExpirationIndex(*coll, docId, doc.expiresAt);
|
|
addToExpirationIndex(*coll, docId, doc.expiresAt);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
it->second = doc;
|
|
it->second = doc;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -358,6 +360,7 @@ bool MemoryStore::remove(const std::string& collection, const std::string& id) {
|
|
|
removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
removeFromExpirationIndex(*coll, id, it->second.expiresAt);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
coll->documents.erase(it);
|
|
coll->documents.erase(it);
|
|
|
coll->updatedAt = currentTimeMs();
|
|
coll->updatedAt = currentTimeMs();
|
|
|
|
|
|
|
@@ -428,6 +431,7 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
|
|
|
addToExpirationIndex(*coll, id, updated.expiresAt);
|
|
addToExpirationIndex(*coll, id, updated.expiresAt);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
it->second = updated;
|
|
it->second = updated;
|
|
|
coll->updatedAt = updated.updatedAt;
|
|
coll->updatedAt = updated.updatedAt;
|
|
|
|
|
|
|
@@ -629,6 +633,7 @@ bool MemoryStore::setAdd(const std::string& collection, const std::string& setId
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
members.push_back(member);
|
|
members.push_back(member);
|
|
|
it->second.version++;
|
|
it->second.version++;
|
|
|
it->second.updatedAt = currentTimeMs();
|
|
it->second.updatedAt = currentTimeMs();
|
|
@@ -679,6 +684,7 @@ bool MemoryStore::setRemove(const std::string& collection, const std::string& se
|
|
|
return false;
|
|
return false;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
it->second.data["members"] = newMembers;
|
|
it->second.data["members"] = newMembers;
|
|
|
it->second.version++;
|
|
it->second.version++;
|
|
|
it->second.updatedAt = currentTimeMs();
|
|
it->second.updatedAt = currentTimeMs();
|
|
@@ -848,6 +854,264 @@ MemoryStore::Stats MemoryStore::getStats() const {
|
|
|
return stats;
|
|
return stats;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// ===== Version History =====
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::saveToHistory(CollectionData& coll, const Document& currentDoc) {
|
|
|
|
|
+ DocumentVersion ver;
|
|
|
|
|
+ ver.version = currentDoc.version;
|
|
|
|
|
+ ver.data = currentDoc.data;
|
|
|
|
|
+ ver.timestamp = currentDoc.updatedAt;
|
|
|
|
|
+ ver.updatedBy = currentDoc.updatedBy;
|
|
|
|
|
+ ver.encrypted = currentDoc.encrypted;
|
|
|
|
|
+ ver.encryptedFields = currentDoc.encryptedFields;
|
|
|
|
|
+ ver.createdAt = currentDoc.createdAt;
|
|
|
|
|
+ ver.createdBy = currentDoc.createdBy;
|
|
|
|
|
+
|
|
|
|
|
+ auto& history = coll.versionHistory[currentDoc.id];
|
|
|
|
|
+ history.push_back(std::move(ver));
|
|
|
|
|
+
|
|
|
|
|
+ // Prune oldest versions if maxVersions is configured
|
|
|
|
|
+ if (coll.options.maxVersions > 0) {
|
|
|
|
|
+ while (history.size() > coll.options.maxVersions) {
|
|
|
|
|
+ history.pop_front();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+MemoryStore::VersionHistoryResult MemoryStore::getVersionHistory(
|
|
|
|
|
+ const std::string& collection, const std::string& id,
|
|
|
|
|
+ uint32_t limit, uint32_t offset) const {
|
|
|
|
|
+
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ VersionHistoryResult result;
|
|
|
|
|
+
|
|
|
|
|
+ // Check if document is currently active
|
|
|
|
|
+ auto docIt = coll->documents.find(id);
|
|
|
|
|
+ if (docIt != coll->documents.end() && !docIt->second.isExpired()) {
|
|
|
|
|
+ result.currentVersion = docIt->second.version;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Get version history
|
|
|
|
|
+ auto histIt = coll->versionHistory.find(id);
|
|
|
|
|
+ if (histIt == coll->versionHistory.end() || histIt->second.empty()) {
|
|
|
|
|
+ // No history — still return current version info if document exists
|
|
|
|
|
+ if (docIt == coll->documents.end()) {
|
|
|
|
|
+ return result; // No document, no history
|
|
|
|
|
+ }
|
|
|
|
|
+ return result;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const auto& history = histIt->second;
|
|
|
|
|
+ result.totalCount = history.size();
|
|
|
|
|
+
|
|
|
|
|
+ // Check if document was deleted (has history but no active document)
|
|
|
|
|
+ if (docIt == coll->documents.end()) {
|
|
|
|
|
+ result.documentDeleted = true;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Apply pagination
|
|
|
|
|
+ uint32_t start = std::min(offset, static_cast<uint32_t>(history.size()));
|
|
|
|
|
+ uint32_t end = limit > 0
|
|
|
|
|
+ ? std::min(start + limit, static_cast<uint32_t>(history.size()))
|
|
|
|
|
+ : static_cast<uint32_t>(history.size());
|
|
|
|
|
+
|
|
|
|
|
+ for (uint32_t i = start; i < end; ++i) {
|
|
|
|
|
+ result.versions.push_back(history[i]);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return result;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::optional<DocumentVersion> MemoryStore::getDocumentAtVersion(
|
|
|
|
|
+ const std::string& collection, const std::string& id,
|
|
|
|
|
+ uint64_t version) const {
|
|
|
|
|
+
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Check if requesting the current live version
|
|
|
|
|
+ auto docIt = coll->documents.find(id);
|
|
|
|
|
+ if (docIt != coll->documents.end() && docIt->second.version == version) {
|
|
|
|
|
+ DocumentVersion ver;
|
|
|
|
|
+ ver.version = docIt->second.version;
|
|
|
|
|
+ ver.data = docIt->second.data;
|
|
|
|
|
+ ver.timestamp = docIt->second.updatedAt;
|
|
|
|
|
+ ver.updatedBy = docIt->second.updatedBy;
|
|
|
|
|
+ ver.encrypted = docIt->second.encrypted;
|
|
|
|
|
+ ver.encryptedFields = docIt->second.encryptedFields;
|
|
|
|
|
+ ver.createdAt = docIt->second.createdAt;
|
|
|
|
|
+ ver.createdBy = docIt->second.createdBy;
|
|
|
|
|
+ return ver;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Search in history
|
|
|
|
|
+ auto histIt = coll->versionHistory.find(id);
|
|
|
|
|
+ if (histIt == coll->versionHistory.end()) {
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& ver : histIt->second) {
|
|
|
|
|
+ if (ver.version == version) {
|
|
|
|
|
+ return ver;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return std::nullopt;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std::string& id,
|
|
|
|
|
+ uint64_t version, const std::string& actor) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Find the target version in history
|
|
|
|
|
+ auto histIt = coll->versionHistory.find(id);
|
|
|
|
|
+ if (histIt == coll->versionHistory.end()) {
|
|
|
|
|
+ return 0;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const DocumentVersion* targetVer = nullptr;
|
|
|
|
|
+ for (const auto& ver : histIt->second) {
|
|
|
|
|
+ if (ver.version == version) {
|
|
|
|
|
+ targetVer = &ver;
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!targetVer) {
|
|
|
|
|
+ return 0;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto docIt = coll->documents.find(id);
|
|
|
|
|
+ bool wasDeleted = (docIt == coll->documents.end());
|
|
|
|
|
+ uint64_t newVersion = 0;
|
|
|
|
|
+ Document restoredDoc;
|
|
|
|
|
+
|
|
|
|
|
+ if (!wasDeleted) {
|
|
|
|
|
+ // Active document: save current state to history, then overwrite
|
|
|
|
|
+ saveToHistory(*coll, docIt->second);
|
|
|
|
|
+
|
|
|
|
|
+ newVersion = docIt->second.version + 1;
|
|
|
|
|
+ docIt->second.data = targetVer->data;
|
|
|
|
|
+ docIt->second.version = newVersion;
|
|
|
|
|
+ docIt->second.updatedAt = currentTimeMs();
|
|
|
|
|
+ docIt->second.updatedBy = actor;
|
|
|
|
|
+ docIt->second.encrypted = targetVer->encrypted;
|
|
|
|
|
+ docIt->second.encryptedFields = targetVer->encryptedFields;
|
|
|
|
|
+ docIt->second.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ restoredDoc = docIt->second;
|
|
|
|
|
+ } else {
|
|
|
|
|
+ // Deleted document: reconstruct from version history
|
|
|
|
|
+ // Find the highest version in history to determine new version number
|
|
|
|
|
+ uint64_t maxVer = 0;
|
|
|
|
|
+ for (const auto& ver : histIt->second) {
|
|
|
|
|
+ maxVer = std::max(maxVer, ver.version);
|
|
|
|
|
+ }
|
|
|
|
|
+ newVersion = maxVer + 1;
|
|
|
|
|
+
|
|
|
|
|
+ restoredDoc.id = id;
|
|
|
|
|
+ restoredDoc.collection = collection;
|
|
|
|
|
+ restoredDoc.data = targetVer->data;
|
|
|
|
|
+ restoredDoc.version = newVersion;
|
|
|
|
|
+ restoredDoc.createdAt = targetVer->createdAt;
|
|
|
|
|
+ restoredDoc.createdBy = targetVer->createdBy;
|
|
|
|
|
+ restoredDoc.updatedAt = currentTimeMs();
|
|
|
|
|
+ restoredDoc.updatedBy = actor;
|
|
|
|
|
+ restoredDoc.encrypted = targetVer->encrypted;
|
|
|
|
|
+ restoredDoc.encryptedFields = targetVer->encryptedFields;
|
|
|
|
|
+ restoredDoc.nodeId = config_.nodeId;
|
|
|
|
|
+
|
|
|
|
|
+ // Apply default TTL if configured
|
|
|
|
|
+ if (coll->options.defaultTtlSeconds > 0) {
|
|
|
|
|
+ restoredDoc.setTtlSeconds(coll->options.defaultTtlSeconds);
|
|
|
|
|
+ addToExpirationIndex(*coll, id, restoredDoc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->documents[id] = restoredDoc;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ coll->updatedAt = restoredDoc.updatedAt;
|
|
|
|
|
+
|
|
|
|
|
+ lock.unlock();
|
|
|
|
|
+
|
|
|
|
|
+ // Update stats
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ if (wasDeleted) {
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ stats_.insertCount++;
|
|
|
|
|
+ } else {
|
|
|
|
|
+ stats_.updateCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Emit callbacks
|
|
|
|
|
+ EventType eventType = wasDeleted ? EventType::INSERT : EventType::UPDATE;
|
|
|
|
|
+ emitPersist(collection, id, restoredDoc, eventType);
|
|
|
|
|
+ emitEvent(eventType, collection, id, restoredDoc.data);
|
|
|
|
|
+
|
|
|
|
|
+ return newVersion;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::pair<uint64_t, uint64_t> MemoryStore::restoreToDate(
|
|
|
|
|
+ const std::string& collection, const std::string& id,
|
|
|
|
|
+ uint64_t timestamp, const std::string& actor) {
|
|
|
|
|
+
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return {0, 0};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t targetVersion = 0;
|
|
|
|
|
+
|
|
|
|
|
+ {
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ // Search version history for the version active at the given timestamp
|
|
|
|
|
+ auto histIt = coll->versionHistory.find(id);
|
|
|
|
|
+ if (histIt != coll->versionHistory.end()) {
|
|
|
|
|
+ for (const auto& ver : histIt->second) {
|
|
|
|
|
+ if (ver.timestamp <= timestamp) {
|
|
|
|
|
+ targetVersion = ver.version;
|
|
|
|
|
+ } else {
|
|
|
|
|
+ break; // Versions are ordered oldest-first
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Also check current document — if its updatedAt <= timestamp, it's the active version
|
|
|
|
|
+ auto docIt = coll->documents.find(id);
|
|
|
|
|
+ if (docIt != coll->documents.end() && docIt->second.updatedAt <= timestamp) {
|
|
|
|
|
+ // Current version was active at the given time; no restore needed
|
|
|
|
|
+ // But for consistency with the copy-forward contract, we still restore
|
|
|
|
|
+ targetVersion = docIt->second.version;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (targetVersion == 0) {
|
|
|
|
|
+ return {0, 0};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t newVersion = restoreToVersion(collection, id, targetVersion, actor);
|
|
|
|
|
+ if (newVersion == 0) {
|
|
|
|
|
+ return {0, 0};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return {targetVersion, newVersion};
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
// ===== Snapshot/Recovery Support =====
|
|
// ===== Snapshot/Recovery Support =====
|
|
|
|
|
|
|
|
std::vector<Document> MemoryStore::getAllDocuments(const std::string& collection) const {
|
|
std::vector<Document> MemoryStore::getAllDocuments(const std::string& collection) const {
|
|
@@ -898,6 +1162,68 @@ void MemoryStore::loadDocument(const std::string& collection, Document doc) {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+void MemoryStore::loadDocumentWithHistory(const std::string& collection, Document doc) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ doc.collection = collection;
|
|
|
|
|
+
|
|
|
|
|
+ // If document already exists, save current version to history before overwriting
|
|
|
|
|
+ auto it = coll->documents.find(doc.id);
|
|
|
|
|
+ if (it != coll->documents.end()) {
|
|
|
|
|
+ saveToHistory(*coll, it->second);
|
|
|
|
|
+
|
|
|
|
|
+ // Remove old expiration index entry
|
|
|
|
|
+ if (it->second.expiresAt > 0) {
|
|
|
|
|
+ removeFromExpirationIndex(*coll, doc.id, it->second.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Overwrite (no stats change — doc count stays the same)
|
|
|
|
|
+ if (doc.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, doc.id, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+ it->second = std::move(doc);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ // New document — same as loadDocument
|
|
|
|
|
+ if (doc.expiresAt > 0) {
|
|
|
|
|
+ addToExpirationIndex(*coll, doc.id, doc.expiresAt);
|
|
|
|
|
+ }
|
|
|
|
|
+ std::string docId = doc.id;
|
|
|
|
|
+ coll->documents[docId] = std::move(doc);
|
|
|
|
|
+
|
|
|
|
|
+ std::lock_guard<std::mutex> statsLock(statsMutex_);
|
|
|
|
|
+ stats_.totalDocuments++;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::pair<std::string, std::deque<DocumentVersion>>>
|
|
|
|
|
+MemoryStore::getAllVersionHistory(const std::string& collection) const {
|
|
|
|
|
+ const CollectionData* coll = getCollection(collection);
|
|
|
|
|
+ if (!coll) {
|
|
|
|
|
+ return {};
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::shared_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+
|
|
|
|
|
+ std::vector<std::pair<std::string, std::deque<DocumentVersion>>> result;
|
|
|
|
|
+ result.reserve(coll->versionHistory.size());
|
|
|
|
|
+ for (const auto& [docId, history] : coll->versionHistory) {
|
|
|
|
|
+ if (!history.empty()) {
|
|
|
|
|
+ result.emplace_back(docId, history);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return result;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+void MemoryStore::loadVersionHistory(const std::string& collection, const std::string& docId,
|
|
|
|
|
+ std::deque<DocumentVersion> history) {
|
|
|
|
|
+ CollectionData* coll = getOrCreateCollection(collection);
|
|
|
|
|
+
|
|
|
|
|
+ std::unique_lock<std::shared_mutex> lock(coll->mutex);
|
|
|
|
|
+ coll->versionHistory[docId] = std::move(history);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
void MemoryStore::clear() {
|
|
void MemoryStore::clear() {
|
|
|
std::unique_lock<std::shared_mutex> lock(globalMutex_);
|
|
std::unique_lock<std::shared_mutex> lock(globalMutex_);
|
|
|
collections_.clear();
|
|
collections_.clear();
|