oplog mirror support (#367)

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

5 files changed, 173 insertions, 1 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;
+}
diff --git a/src/zen/cmds/projectstore.h b/src/zen/cmds/projectstore.h
index 5b1cd05db..d67aed9aa 100644
--- a/src/zen/cmds/projectstore.h
+++ b/src/zen/cmds/projectstore.h
@@ -229,3 +229,19 @@ private:
 	std::string		 m_OplogName;
 	std::string		 m_OpId;
 };
+
+class OplogMirrorCommand : public ZenCmdBase
+{
+public:
+	OplogMirrorCommand();
+	~OplogMirrorCommand();
+	virtual int				  Run(const ZenCliOptions& GlobalOptions, int argc, char** argv) override;
+	virtual cxxopts::Options& Options() override { return m_Options; }
+
+private:
+	cxxopts::Options m_Options{"oplog-mirror", "Mirror oplog to file system"};
+	std::string		 m_HostName;
+	std::string		 m_ProjectName;
+	std::string		 m_OplogName;
+	std::string		 m_MirrorRootPath;
+};
diff --git a/src/zen/zen.cpp b/src/zen/zen.cpp
index 6772e7ab4..204a91f74 100644
--- a/src/zen/zen.cpp
+++ b/src/zen/zen.cpp
@@ -200,6 +200,7 @@ main(int argc, char** argv)
 	GcStatusCommand			 GcStatusCmd;
 	HashCommand				 HashCmd;
 	ImportOplogCommand		 ImportOplogCmd;
+	OplogMirrorCommand		 OplogMirrorCmd;
 	PrintCommand			 PrintCmd;
 	PrintPackageCommand		 PrintPkgCmd;
 	ProjectDetailsCommand	 ProjectDetailsCmd;
@@ -243,6 +244,7 @@ main(int argc, char** argv)
 		{"oplog-delete",		&DeleteOplogCmd,		"Delete a project oplog"},
 		{"oplog-export",		&ExportOplogCmd,		"Export project store oplog"},
 		{"oplog-import",		&ImportOplogCmd,		"Import project store oplog"},
+		{"oplog-mirror",		&OplogMirrorCmd,		"Mirror project store oplog to file system"},
 		{"oplog-snapshot",		&SnapshotOplogCmd,		"Snapshot project store oplog"},
 		{"print",				&PrintCmd,				"Print compact binary object"},
 		{"printpackage",		&PrintPkgCmd,			"Print compact binary package"},
diff --git a/src/zenhttp/httpclient.cpp b/src/zenhttp/httpclient.cpp
index 05ff6d07b..744787201 100644
--- a/src/zenhttp/httpclient.cpp
+++ b/src/zenhttp/httpclient.cpp
@@ -435,7 +435,28 @@ HttpClient::Response::ToText()
 bool
 HttpClient::Response::IsSuccess() const noexcept
 {
-	return IsHttpSuccessCode(StatusCode);
+	return !Error && IsHttpSuccessCode(StatusCode);
+}
+
+void
+HttpClient::Response::ThrowError(std::string_view ErrorPrefix)
+{
+	if (!IsSuccess())
+	{
+		if (Error.has_value())
+		{
+			throw std::runtime_error(fmt::format("{}: {}", ErrorPrefix, Error->ErrorMessage));
+		}
+		else if (StatusCode != HttpResponseCode::ImATeapot && (int)StatusCode)
+		{
+			throw std::runtime_error(
+				fmt::format("{}: HTTP error {} {} ({})", ErrorPrefix, (int)StatusCode, zen::ToString(StatusCode), AsText()));
+		}
+		else
+		{
+			throw std::runtime_error(fmt::format("{}: {}", ErrorPrefix, "unknown error"));
+		}
+	}
 }
 
 //////////////////////////////////////////////////////////////////////////
diff --git a/src/zenhttp/include/zenhttp/httpclient.h b/src/zenhttp/include/zenhttp/httpclient.h
index 559d7e719..9ff4910bf 100644
--- a/src/zenhttp/include/zenhttp/httpclient.h
+++ b/src/zenhttp/include/zenhttp/httpclient.h
@@ -65,6 +65,8 @@ public:
 		// 2xx range)
 		bool			IsSuccess() const noexcept;
 		inline explicit operator bool() const noexcept { return IsSuccess(); }
+
+		void ThrowError(std::string_view ErrorPrefix = "error");
 	};
 
 	[[nodiscard]] Response Put(std::string_view Url, const IoBuffer& Payload);