start signal

src/processing_utils.cpp

@@ -8,7 +8,7 @@
 #include <variant>
 #include "database_utils.hpp"
 namespace processing_utils {
-  
+  constexpr std::size_t MAX_BATCH_ROWS = 500;
 bool parse_date_from_filename(std::string_view path,
                               int& year, int& month, int& day)
 {
@@ -31,17 +31,24 @@ bool parse_date_from_filename(std::string_view path,
 bool parse_time_from_line(std::string_view line,
                           int& hour, int& min, int& sec)
 {
-    // Expect: [HH:MM:SS]
-    if (line.size() < 10 || line[0] != '[')
+    // Match [HH:MM:SS] anywhere in the line
+    static const std::regex re(
+        R"(\[(\d{2}):(\d{2}):(\d{2})\])",
+        std::regex::optimize
+    );
+
+    std::cmatch m;
+    if (!std::regex_search(line.begin(), line.end(), m, re))
         return false;
 
-    hour = (line[1] - '0') * 10 + (line[2] - '0');
-    min  = (line[4] - '0') * 10 + (line[5] - '0');
-    sec  = (line[7] - '0') * 10 + (line[8] - '0');
+    hour = (m[1].str()[0] - '0') * 10 + (m[1].str()[1] - '0');
+    min  = (m[2].str()[0] - '0') * 10 + (m[2].str()[1] - '0');
+    sec  = (m[3].str()[0] - '0') * 10 + (m[3].str()[1] - '0');
 
     return true;
 }
 
+
 #include <ctime>
 #include <string_view>
 
@@ -104,7 +111,7 @@ time_t extract_timestamp_utc(std::string_view filename,
   Route classify_line(std::string_view line) {
 
     // ---- Client lifecycle ----
-    if (line.find("Starting Minecraft client") != std::string_view::npos)
+    if (line.find("Loading Minecraft") != std::string_view::npos)
         return Route::CLIENT_START;
 
     if (line.find("Stopping!") != std::string_view::npos)
@@ -128,22 +135,34 @@ time_t extract_timestamp_utc(std::string_view filename,
     return Route::NONE;
   }
 
-  std::pair<std::string, std::string>
-parse_chat(std::string_view line) {
-    // [CHAT] <Steve> hello world
+std::pair<std::string, std::string>
+parse_chat(std::string_view line)
+{
+    constexpr std::string_view tag = "[CHAT]";
 
-    auto start = line.find("[CHAT] <");
-    start += 8;
+    auto tag_pos = line.find(tag);
+    if (tag_pos == std::string_view::npos)
+        return {"", ""};
 
-    auto end = line.find('>', start);
-    auto msg = end + 2;
+    std::size_t pos = tag_pos + tag.size();
+    if (pos >= line.size())
+        return {"", ""};
 
+    // Skip whitespace after [CHAT]
+    while (pos < line.size() && line[pos] == ' ')
+        ++pos;
+
+    if (pos >= line.size())
+        return {"", ""};
+
+    // Entire remainder is "message"
     return {
-        std::string(line.substr(start, end - start)),
-        std::string(line.substr(msg))
+        "",  // sender unknown / system / to be inferred elsewhere
+        std::string(line.substr(pos))
     };
 }
 
+
   void process_to_eof(std::string path, std::shared_ptr<clickhouse::Client>client, std::string target_user_id){
     spdlog::info("Processing file {} ", path);
     gzFile gz_file_log = gzopen(path.c_str(), "rb");
@@ -180,14 +199,17 @@ parse_chat(std::string_view line) {
           std::pair<std::string, std::string> chat_message_pair = parse_chat(current_line);
           chat_events.add_row(target_user_id, log_timestamp, chat_message_pair.first, chat_message_pair.second) ; 
       }
+      current_line.clear();
     }
     activity_events.insert_as_batch(client); chat_events.insert_as_batch(client); client_events.insert_as_batch(client);
     current_line.clear();
   }
-  void process_blocking(std::string path,
+void process_blocking(std::string path,
                       std::shared_ptr<clickhouse::Client> client,
                       std::string target_user_id)
 {
+    constexpr uint64_t MAX_BATCH_ROWS = 500;
+
     spdlog::info("Processing (blocking) file {}", path);
 
     std::ifstream input_file(path);
@@ -196,68 +218,123 @@ parse_chat(std::string_view line) {
         return;
     }
 
-    // Seek to the end like tail -f
+    // tail -f semantics
     input_file.seekg(0, std::ios::end);
 
     std::string current_line;
-    db_utils::ClientEventsBatch client_events{};
-    db_utils::ChatEventsBatch chat_events{};
-    db_utils::ActivityEventsBatch activity_events{};
+
+    db_utils::ClientEventsBatch client_events;
+    db_utils::ChatEventsBatch chat_events;
+    db_utils::ActivityEventsBatch activity_events;
+
+    uint64_t row_count = 0;
 
     while (true) {
-        // Read all available lines
-        while (std::getline(input_file, current_line)) {
-            if (current_line.empty()) continue;
+        bool read_any = false;
 
-            std::time_t log_timestamp = extract_timestamp_utc(path, current_line);
+        while (std::getline(input_file, current_line)) {
+            read_any = true;
+            if (current_line.empty())
+                continue;
+
+            std::time_t ts = extract_timestamp_utc(path, current_line);
+            if (ts == static_cast<std::time_t>(-1))
+                continue;
 
             switch (classify_line(current_line)) {
+
                 case Route::CLIENT_START:
-                    client_events.add_row(target_user_id, log_timestamp,
-                                          std::variant<std::string, uint8_t>{"start"}, path);
+                    client_events.add_row(
+                        target_user_id, ts,
+                        std::variant<std::string, uint8_t>{"start"},
+                        path
+                    );
+                    ++row_count;
                     break;
+
                 case Route::CLIENT_END:
-                    client_events.add_row(target_user_id, log_timestamp,
-                                          std::variant<std::string, uint8_t>{"end"}, path);
+                    client_events.add_row(
+                        target_user_id, ts,
+                        std::variant<std::string, uint8_t>{"end"},
+                        path
+                    );
+                    ++row_count;
                     break;
+
                 case Route::ACT_COMMAND:
-                    activity_events.add_row(target_user_id, log_timestamp, "command", 1.0f);
+                    activity_events.add_row(
+                        target_user_id, ts,
+                        "command", 1.0f
+                    );
+                    ++row_count;
                     break;
+
                 case Route::ACT_DEATH:
-                    activity_events.add_row(target_user_id, log_timestamp,
-                                            std::variant<std::string, uint8_t>{"death"}, 0.5f);
+                    activity_events.add_row(
+                        target_user_id, ts,
+                        std::variant<std::string, uint8_t>{"death"},
+                        0.5f
+                    );
+                    ++row_count;
                     break;
+
                 case Route::ACT_DIMENSION:
-                    activity_events.add_row(target_user_id, log_timestamp,
-                                            std::variant<std::string, uint8_t>{"dimension"}, 0.3f);
-                    break;
-                case Route::ACT_OTHER:
-                    activity_events.add_row(target_user_id, log_timestamp,
-                                            std::variant<std::string, uint8_t>{"other"}, 0.2f);
+                    activity_events.add_row(
+                        target_user_id, ts,
+                        std::variant<std::string, uint8_t>{"dimension"},
+                        0.3f
+                    );
+                    ++row_count;
                     break;
+
                 case Route::CHAT: {
-                    auto chat_pair = parse_chat(current_line);
-                    chat_events.add_row(target_user_id, log_timestamp,
-                                        chat_pair.first, chat_pair.second);
+                    auto chat = parse_chat(current_line);
+                    if (!chat.second.empty()) {
+                        chat_events.add_row(
+                            target_user_id, ts,
+                            chat.first, chat.second
+                        );
+                        ++row_count;
+                    }
                     break;
                 }
+
                 default:
                     break;
             }
+
+            // HARD FLUSH CONDITION
+            if (row_count >= MAX_BATCH_ROWS) {
+                activity_events.insert_as_batch(client);
+                chat_events.insert_as_batch(client);
+                client_events.insert_as_batch(client);
+
+                // Destroy and recreate — ONLY valid reset
+                client_events = db_utils::ClientEventsBatch{};
+                chat_events   = db_utils::ChatEventsBatch{};
+                activity_events = db_utils::ActivityEventsBatch{};
+
+                row_count = 0;
+            }
         }
 
-        // Insert accumulated batches into ClickHouse
-        activity_events.insert_as_batch(client);
-        chat_events.insert_as_batch(client);
-        client_events.insert_as_batch(client);
+        // Flush partial batch if we read anything
+        if (read_any && row_count > 0) {
+            activity_events.insert_as_batch(client);
+            chat_events.insert_as_batch(client);
+            client_events.insert_as_batch(client);
 
-        current_line.clear();
+            client_events = db_utils::ClientEventsBatch{};
+            chat_events   = db_utils::ChatEventsBatch{};
+            activity_events = db_utils::ActivityEventsBatch{};
 
-        // Clear EOF state so we can continue reading
-        input_file.clear();
+            row_count = 0;
+        }
 
-        // Sleep a bit before polling for new data
+        input_file.clear(); // clear EOF only
         std::this_thread::sleep_for(std::chrono::milliseconds(200));
     }
 }
+
+
 }