oplog mirror support (#367)

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

6 files changed, 174 insertions, 1 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index f9da58fed..51b0d34d5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
 ##
+- Feature: Add `oplog-mirror` command to Zen command line tool. It can be used to export the contents of an oplog as files. Currently it will export all files, filtering options will be added at a later time
 - Feature: Add `--force-update` option to Zen command line tool `project-create` to update or create a project store project. It will update meta information about the project without affecting existing oplogs if project exists.
 - Feature: Add `--force-update` option to Zen command line tool `oplog-update` to update or create a project store oplog. It will update meta information about the project without affecting existing oplog data if oplog exists.
 - Feature: Zen command line tool `project-delete` to delete a project store project and all its oplogs.
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);