Insert uninsirtable lines using the hostnameId recorded in the bad_lines table

Includes the same fix from the previous commit which stops caching
entities until the inserts are complete.
Updates storeLogLine to take an insert command with the hostnameId
already bound.

4 files changed, 14 insertions, 9 deletions

diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index 077f7d8..c06fd1c 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -345,12 +345,13 @@ namespace WebStat {
 		auto deleteLine = dbconn->modify(SQL::DELETE_ENTITY, SQL::DELETE_ENTITY_OPTS);
 		auto setEntityUnparsable = dbconn->modify(SQL::SET_ENTITY_TYPE, SQL::SET_ENTITY_TYPE_OPTS);
 		setEntityUnparsable->bindParamS(0, "unparsable_line");
+		auto insert = dbconn->modify(SQL::ACCESS_LOG_INSERT, SQL::ACCESS_LOG_INSERT_OPTS);
 
 		unsigned int stored = 0;
 		while (!terminated) {
 			unsigned int batchSize = 0;
 			DB::TransactionScope batchTx {*dbconn};
-			for (auto [id, line] : lineSelect->as<EntityId, std::string>()) {
+			for (auto [id, line, originalHostnameId] : lineSelect->as<EntityId, std::string, EntityId>()) {
 				batchSize += 1;
 				try {
 					DB::TransactionScope lineTx {*dbconn};
@@ -359,8 +360,9 @@ namespace WebStat {
 						auto valuesEntities = entities(values);
 						fillKnownEntities(valuesEntities);
 						storeNewEntities(dbconn, valuesEntities);
+						insert->bindParam(0, originalHostnameId);
+						storeLogLine(insert.get(), values);
 						existingEntities()->insert_range(valuesEntities | ENTITY_IDS);
-						storeLogLine(dbconn, values);
 
 						deleteLine->bindParamI(0, id);
 						deleteLine->execute();
@@ -411,6 +413,8 @@ namespace WebStat {
 	Ingestor::ingestLogLines(DB::Connection * dbconn, const LinesView lines)
 	{
 		DB::TransactionScope batchTx {*dbconn};
+		auto insert = dbconn->modify(SQL::ACCESS_LOG_INSERT, SQL::ACCESS_LOG_INSERT_OPTS);
+		insert->bindParam(0, hostnameId);
 		for (const auto & line : lines) {
 			if (auto result = scanLogLine(line)) {
 				stats.linesParsed++;
@@ -420,7 +424,7 @@ namespace WebStat {
 				try {
 					DB::TransactionScope lineTx {*dbconn};
 					storeNewEntities(dbconn, valuesEntities);
-					storeLogLine(dbconn, values);
+					storeLogLine(insert.get(), values);
 					existingEntities()->insert_range(valuesEntities | ENTITY_IDS);
 				}
 				catch (const DB::Error & originalError) {
@@ -678,11 +682,8 @@ namespace WebStat {
 
 	template<typename... T>
 	void
-	Ingestor::storeLogLine(DB::Connection * dbconn, const std::tuple<T...> & values) const
+	Ingestor::storeLogLine(DB::ModifyCommand * insert, const std::tuple<T...> & values) const
 	{
-		auto insert = dbconn->modify(SQL::ACCESS_LOG_INSERT, SQL::ACCESS_LOG_INSERT_OPTS);
-
-		insert->bindParam(0, hostnameId);
 		std::apply(
 				[&insert](auto &&... value) {
 					unsigned int param = 1;
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index dacfdf7..1a47a1b 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -81,7 +81,7 @@ namespace WebStat {
 		Job::Result jobStoreQueuedLines();
 		Job::Result jobRetryUninsertableLines();
 
-		template<typename... T> void storeLogLine(DB::Connection *, const std::tuple<T...> &) const;
+		template<typename... T> void storeLogLine(DB::ModifyCommand * insert, const std::tuple<T...> &) const;
 
 		IngestorSettings settings;
 
diff --git a/src/schema.sql b/src/schema.sql
index 40cab51..fd2f3c5 100644
--- a/src/schema.sql
+++ b/src/schema.sql
@@ -94,6 +94,9 @@ CREATE INDEX idx_entities_retryinsert ON bad_lines(id)
 WHERE
 	type = 'uninsertable_line' AND detail ->> 'retriedAt' IS NULL;
 
+ALTER TABLE bad_lines
+	ADD CONSTRAINT bad_lines_have_hostnameid CHECK (detail ? 'hostnameId');
+
 CREATE OR REPLACE FUNCTION entity(newValue text, newType entity)
 	RETURNS TABLE(
 		id integer,
diff --git a/src/sql/selectUninsertableLines.sql b/src/sql/selectUninsertableLines.sql
index 060e800..2a57683 100644
--- a/src/sql/selectUninsertableLines.sql
+++ b/src/sql/selectUninsertableLines.sql
@@ -1,6 +1,7 @@
 SELECT
 	id,
-	value
+	value,
+	cast(detail ->> 'hostnameId' AS int) AS hostnameId
 FROM
 	entities
 WHERE