From 63d6bf7395425be73fd6d9f5c89dc48817afffe6 Mon Sep 17 00:00:00 2001
From: Dan Goodliffe <dan@randomdan.homeip.net>
Date: Mon, 6 Oct 2025 20:43:52 +0100
Subject: Add parked line import job

Periodically, on idle, scan for and import previously parked lines.
---
 src/ingestor.cpp | 58 +++++++++++++++++++++++++++++++++++++++++++++++++++++++-
 src/ingestor.hpp | 10 ++++++++++
 src/util.hpp     |  2 ++
 3 files changed, 69 insertions(+), 1 deletion(-)

(limited to 'src')

diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index cf49f52..f965d1d 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -142,7 +142,7 @@ namespace WebStat {
 				handleCurlOperations();
 			}
 			else if (!interesting) {
-				// do idle job things
+				runJobsIdle();
 			}
 		}
 		while (!curlOperations.empty() && curl_multi_poll(curl.get(), nullptr, 0, INT_MAX, nullptr) == CURLM_OK) {
@@ -188,6 +188,62 @@ namespace WebStat {
 		linesParked++;
 	}
 
+	void
+	Ingestor::runJobsIdle()
+	{
+		const auto now = JobLastRunTime::clock::now();
+		auto runJobAsNeeded = [this, now](auto job, JobLastRunTime & lastRun, auto freq) {
+			try {
+				if (lastRun + freq < now) {
+					(this->*job)();
+					lastRun = now;
+				}
+			}
+			catch (const std::exception &) {
+				// Error, retry in half the frequency
+				lastRun = now - (freq / 2);
+			}
+		};
+		runJobAsNeeded(&Ingestor::jobIngestParkedLines, lastRunIngestParkedLines, settings.freqIngestParkedLines);
+	}
+
+	void
+	Ingestor::jobIngestParkedLines()
+	{
+		for (auto pathIter = std::filesystem::directory_iterator {settings.fallbackDir};
+				pathIter != std::filesystem::directory_iterator {}; ++pathIter) {
+			if (scn::scan<Crc32Value>(pathIter->path().filename().string(), "parked-{}.log")) {
+				jobIngestParkedLine(pathIter);
+			}
+		}
+	}
+
+	void
+	Ingestor::jobIngestParkedLine(const std::filesystem::directory_iterator & pathIter)
+	{
+		jobIngestParkedLine(pathIter->path(), pathIter->file_size());
+	}
+
+	void
+	Ingestor::jobIngestParkedLine(const std::filesystem::path & path, uintmax_t size)
+	{
+		if (std::ifstream parked {path}) {
+			std::string line;
+			line.resize_and_overwrite(size, [&parked](char * content, size_t size) {
+				parked.read(content, static_cast<std::streamsize>(size));
+				return static_cast<size_t>(parked.tellg());
+			});
+			if (line.length() < size) {
+				throw std::system_error {errno, std::generic_category(), "Short read of parked file"};
+			}
+			ingestLogLine(dbpool->get().get(), line);
+		}
+		else {
+			throw std::system_error {errno, std::generic_category(), strerror(errno)};
+		}
+		std::filesystem::remove(path);
+	}
+
 	template<typename... T>
 	Ingestor::NewEntities
 	Ingestor::newEntities(const std::tuple<T...> & values) const
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index afd4a0f..a20071e 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -23,6 +23,7 @@ namespace WebStat {
 		unsigned int dbMax = 4;
 		unsigned int dbKeep = 2;
 		int idleJobsAfter = duration_cast<milliseconds>(1min).count();
+		minutes freqIngestParkedLines = 30min;
 	};
 
 	class Ingestor {
@@ -44,6 +45,9 @@ namespace WebStat {
 		void ingestLogLine(std::string_view);
 		void ingestLogLine(DB::Connection *, std::string_view);
 		void parkLogLine(std::string_view);
+		void runJobsIdle();
+
+		void jobIngestParkedLines();
 
 		template<typename... T> void storeLogLine(DB::Connection *, const std::tuple<T...> &) const;
 
@@ -57,6 +61,9 @@ namespace WebStat {
 		size_t linesDiscarded = 0;
 		size_t linesParked = 0;
 
+		using JobLastRunTime = std::chrono::system_clock::time_point;
+		JobLastRunTime lastRunIngestParkedLines;
+
 	private:
 		static constexpr size_t MAX_NEW_ENTITIES = 6;
 		void storeEntities(DB::Connection *, std::span<const std::optional<Entity>>) const;
@@ -64,6 +71,9 @@ namespace WebStat {
 		template<typename... T> NewEntities newEntities(const std::tuple<T...> &) const;
 		void handleCurlOperations();
 
+		void jobIngestParkedLine(const std::filesystem::directory_iterator &);
+		void jobIngestParkedLine(const std::filesystem::path &, uintmax_t size);
+
 		using CurlOperations = std::map<CURL *, std::unique_ptr<CurlOperation>>;
 		mutable std::flat_set<Crc32Value> existingEntities;
 		uint32_t hostnameId;
diff --git a/src/util.hpp b/src/util.hpp
index d712d3b..ad55604 100644
--- a/src/util.hpp
+++ b/src/util.hpp
@@ -23,6 +23,8 @@ namespace WebStat {
 				values);
 	}
 
+	using FilePtr = std::unique_ptr<std::FILE, DeleteWith<&fclose>>;
+
 	template<typename... T>
 	void
 	bindMany(const DB::CommandPtr & cmd, unsigned int firstParam, T &&... param)
-- 
cgit v1.2.3