diff --git a/includes/functions.php b/includes/functions.php index c44c5fd..b3a892f 100644 --- a/includes/functions.php +++ b/includes/functions.php @@ -75,19 +75,16 @@ function fetch_node_api(array $node, string $path): mixed { * Ingest connection events from portspoof_py that are strictly newer than * $last_event_at (ISO 8601 string, or null to ingest everything). * + * New rows are written with multi-row INSERTs in chunks, inside a single + * transaction per call, to keep the number of round trips and commits low. + * * Returns ['inserted' => int, 'new_max_ts' => string|null] * where new_max_ts is the DATETIME(6) of the newest row inserted, or null if * nothing new was inserted. */ function ingest_connections(int $node_id, array $events, ?string $last_event_at): array { - $inserted = 0; $new_max_ts = null; - - $s = db()->prepare( - 'INSERT INTO connections - (node_id, occurred_at, src_ip, src_port, dst_port, banner_hex, banner_len) - VALUES (?, ?, ?, ?, ?, ?, ?)' - ); + $rows = []; foreach ($events as $ev) { $raw_ts = $ev['timestamp'] ?? null; @@ -103,7 +100,7 @@ function ingest_connections(int $node_id, array $events, ?string $last_event_at) continue; } - $s->execute([ + $rows[] = [ $node_id, $ts, $ev['src_ip'] ?? '', @@ -111,15 +108,40 @@ function ingest_connections(int $node_id, array $events, ?string $last_event_at) (int)($ev['dst_port'] ?? 0), $ev['banner_hex'] ?? null, (int)($ev['banner_len'] ?? 0), - ]); - $inserted++; + ]; if ($new_max_ts === null || $ts > $new_max_ts) { $new_max_ts = $ts; } } - return ['inserted' => $inserted, 'new_max_ts' => $new_max_ts]; + if ($rows === []) { + return ['inserted' => 0, 'new_max_ts' => null]; + } + + $pdo = db(); + $pdo->beginTransaction(); + try { + foreach (array_chunk($rows, 200) as $chunk) { + $values = implode(', ', array_fill(0, count($chunk), '(?, ?, ?, ?, ?, ?, ?)')); + $params = []; + foreach ($chunk as $r) { + $params = array_merge($params, $r); + } + $s = $pdo->prepare( + 'INSERT INTO connections + (node_id, occurred_at, src_ip, src_port, dst_port, banner_hex, banner_len) + VALUES ' . $values + ); + $s->execute($params); + } + $pdo->commit(); + } catch (PDOException $e) { + $pdo->rollBack(); + throw $e; + } + + return ['inserted' => count($rows), 'new_max_ts' => $new_max_ts]; } /**