oplog mirror support (#367)

feature: added oplog-mirror command. this can be invoked to export oplog contents to corresponding files

1 files changed, 131 insertions, 0 deletions

diff --git a/src/zen/cmds/projectstore.cpp b/src/zen/cmds/projectstore.cpp
index b2f1cda5d..bc526ce8c 100644
--- a/src/zen/cmds/projectstore.cpp
+++ b/src/zen/cmds/projectstore.cpp
@@ -4,8 +4,10 @@
 
 #include <zencore/compactbinarybuilder.h>
 #include <zencore/filesystem.h>
+#include <zencore/fmtutils.h>
 #include <zencore/logging.h>
 #include <zencore/stream.h>
+#include <zenhttp/httpclient.h>
 #include <zenhttp/httpcommon.h>
 
 ZEN_THIRD_PARTY_INCLUDES_START
@@ -1247,3 +1249,132 @@ ProjectDetailsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char**
 
 	return 1;
 }
+
+////////////////////////////
+
+OplogMirrorCommand::OplogMirrorCommand()
+{
+	m_Options.add_options()("h,help", "Print help");
+	m_Options.add_option("", "u", "hosturl", "Host URL", cxxopts::value(m_HostName)->default_value(""), "<hosturl>");
+	m_Options.add_option("", "p", "project", "Project name to get info from", cxxopts::value(m_ProjectName), "<projectid>");
+	m_Options.add_option("", "l", "oplog", "Oplog name to get info from", cxxopts::value(m_OplogName), "<oplogid>");
+	m_Options.add_option("", "t", "target", "Target directory for mirror", cxxopts::value(m_MirrorRootPath), "<path>");
+
+	m_Options.parse_positional({"project", "oplog", "target"});
+	m_Options.positional_help("[<projectid> <oplogid> <target>]");
+}
+
+OplogMirrorCommand::~OplogMirrorCommand()
+{
+}
+
+int
+OplogMirrorCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
+{
+	ZEN_UNUSED(GlobalOptions);
+
+	if (!ParseOptions(argc, argv))
+	{
+		return 0;
+	}
+
+	m_HostName = ResolveTargetHostSpec(m_HostName);
+
+	if (m_HostName.empty())
+	{
+		throw zen::OptionParseException("unable to resolve server specification");
+	}
+
+	if (m_ProjectName.empty())
+	{
+		throw zen::OptionParseException("a project must be specified");
+	}
+
+	if (m_OplogName.empty())
+	{
+		throw zen::OptionParseException("an oplog must be specified");
+	}
+
+	if (m_MirrorRootPath.empty())
+	{
+		throw zen::OptionParseException("a target path must be specified");
+	}
+
+	ZEN_CONSOLE("Emitting file data from oplog '{}'/'{}' to '{}'", m_ProjectName, m_OplogName, m_MirrorRootPath);
+
+	zen::HttpClient Http(m_HostName);
+
+	if (zen::HttpClient::Response Result = Http.Get(fmt::format("/prj/{}/oplog/{}", m_ProjectName, m_OplogName)))
+	{
+		// The info requested is not really used at this moment, we just use the probe to be able to provide
+		// better diagnostics up front
+	}
+	else
+	{
+		Result.ThrowError("oplog info fetch failed"sv);
+
+		return 1;
+	}
+
+	// Emit file data to target directory
+
+	std::filesystem::path RootPath{m_MirrorRootPath};
+	zen::CreateDirectories(RootPath);
+
+	int FileCount		= 0;
+	int OplogEntryCount = 0;
+
+	auto EmitFilesForDataArray = [&](zen::CbArrayView DataArray) {
+		for (auto DataIter : DataArray)
+		{
+			if (zen::CbObjectView Data = DataIter.AsObjectView())
+			{
+				std::string_view FileName = Data["filename"sv].AsString();
+				zen::Oid		 ChunkId  = Data["id"sv].AsObjectId();
+
+				if (zen::HttpClient::Response ChunkResponse =
+						Http.Get(fmt::format("/prj/{}/oplog/{}/{}"sv, m_ProjectName, m_OplogName, ChunkId)))
+				{
+					zen::IoBuffer ChunkData = ChunkResponse.ResponsePayload;
+					zen::WriteFile(RootPath / FileName, ChunkData);
+
+					++FileCount;
+				}
+				else
+				{
+					ChunkResponse.ThrowError("chunk data fetch failed"sv);
+				}
+			}
+		}
+	};
+
+	if (zen::HttpClient::Response Response = Http.Get(fmt::format("/prj/{}/oplog/{}/entries"sv, m_ProjectName, m_OplogName)))
+	{
+		if (zen::CbObject ResponseObject = Response.AsObject())
+		{
+			for (auto EntryIter : ResponseObject["entries"sv])
+			{
+				zen::CbObjectView Entry = EntryIter.AsObjectView();
+
+				EmitFilesForDataArray(Entry["packagedata"sv].AsArrayView());
+				EmitFilesForDataArray(Entry["bulkdata"sv].AsArrayView());
+
+				++OplogEntryCount;
+			}
+		}
+		else
+		{
+			ZEN_ERROR("unknown format response to oplog entries request");
+		}
+	}
+	else
+	{
+		Response.ThrowError("oplog entries fetch failed");
+
+		return 1;
+	}
+
+	ZEN_CONSOLE("mirrored {} files from {} oplog entries successfully", FileCount, OplogEntryCount);
+
+	return 0;
+}