feat: 🎸 sync tinybird

2026-04-06 17:15:27 +01:00
parent 11cbd4c151
commit 4eca646ded
6 changed files with 519 additions and 5 deletions
--- a/server/tinybird/scripts/firecrawl-march-dedup.ts
+++ b/server/tinybird/scripts/firecrawl-march-dedup.ts
@@ -0,0 +1,107 @@
+/**
+ * Step 2: Deduplicate March NDJSON files by streaming line by line.
+ *
+ * Since the export is sorted, duplicate rows are always adjacent.
+ * Simply compares each line to the previous - if identical, skip it.
+ * Only v2 rows are deduplicated (SDK rows are kept as-is regardless).
+ *
+ * Memory usage: O(1) - only holds one line in memory at a time.
+ *
+ * Usage: bun run scripts/firecrawl-march-dedup.ts [YYYY-MM-DD]
+ * Run from: sirtenzin-autumn/server/tinybird/
+ */
+
+import { createReadStream, createWriteStream } from "node:fs";
+import { readFile, writeFile, mkdir } from "node:fs/promises";
+import { createInterface } from "node:readline";
+import { join } from "node:path";
+
+const INPUT_DIR = join(process.cwd(), "march-export");
+const OUTPUT_DIR = join(process.cwd(), "march-deduped");
+const SUMMARY_FILE = join(OUTPUT_DIR, "dedup-summary.json");
+
+await mkdir(OUTPUT_DIR, { recursive: true });
+
+const dayFilter = process.argv[2] ?? null;
+
+const { readdir } = await import("node:fs/promises");
+let files = (await readdir(INPUT_DIR)).filter((f) => f.endsWith(".ndjson")).sort();
+if (dayFilter) {
+  files = files.filter((f) => f.includes(dayFilter));
+  if (files.length === 0) { console.error(`No file found for: ${dayFilter}`); process.exit(1); }
+}
+
+console.log(`\n=== Firecrawl March 2026 - Deduplication (streaming) ===`);
+console.log(`Files: ${files.length}${dayFilter ? ` (${dayFilter})` : ""}\n`);
+
+type DaySummary = {
+  day: string;
+  rowsIn: number;
+  rowsOut: number;
+  dupesRemoved: number;
+};
+
+let allSummary: DaySummary[] = [];
+try { allSummary = JSON.parse(await readFile(SUMMARY_FILE, "utf-8")); } catch { }
+
+for (const file of files) {
+  const day = file.replace("day-", "").replace(".ndjson", "");
+  const inputPath = join(INPUT_DIR, file);
+  const outputPath = join(OUTPUT_DIR, file);
+
+  const fileSize = (await import("node:fs")).statSync(inputPath).size;
+  const writeStream = createWriteStream(outputPath);
+  const rl = createInterface({ input: createReadStream(inputPath), crlfDelay: Infinity });
+
+  let prevLine = "";
+  let rowsIn = 0;
+  let rowsOut = 0;
+  let dupesRemoved = 0;
+  let bytesRead = 0;
+  const dayStart = Date.now();
+  let lastLog = Date.now();
+  const LOG_INTERVAL_MS = 5_000;
+
+  for await (const line of rl) {
+    if (!line) continue;
+    rowsIn++;
+    bytesRead += line.length + 1;
+
+    if (line === prevLine) {
+      dupesRemoved++;
+      continue;
+    }
+
+    writeStream.write(line + "\n");
+    rowsOut++;
+    prevLine = line;
+
+    // Progress log every 5s
+    if (Date.now() - lastLog >= LOG_INTERVAL_MS) {
+      const pct = ((bytesRead / fileSize) * 100).toFixed(1);
+      const elapsedSec = (Date.now() - dayStart) / 1000;
+      const etaSec = elapsedSec / (bytesRead / fileSize) - elapsedSec;
+      const etaMin = (etaSec / 60).toFixed(1);
+      const mbRead = (bytesRead / 1e6).toFixed(0);
+      const mbTotal = (fileSize / 1e6).toFixed(0);
+      console.log(`  [${day}] ${pct}% | ${mbRead}/${mbTotal}MB | ${rowsIn.toLocaleString()} in, ${rowsOut.toLocaleString()} out, ${dupesRemoved.toLocaleString()} dupes | ETA ~${etaMin}min`);
+      lastLog = Date.now();
+    }
+  }
+
+  await new Promise<void>((resolve, reject) => {
+    writeStream.end((err: Error | null | undefined) => err ? reject(err) : resolve());
+  });
+
+  const totalSec = ((Date.now() - dayStart) / 1000).toFixed(1);
+  console.log(`[${day}] DONE in ${totalSec}s | ${rowsIn.toLocaleString()} in → ${rowsOut.toLocaleString()} out | ${dupesRemoved.toLocaleString()} dupes removed`);
+
+  const daySummary: DaySummary = { day, rowsIn, rowsOut, dupesRemoved };
+  const idx = allSummary.findIndex((s) => s.day === day);
+  if (idx >= 0) allSummary[idx] = daySummary; else allSummary.push(daySummary);
+}
+
+allSummary.sort((a, b) => a.day.localeCompare(b.day));
+await writeFile(SUMMARY_FILE, JSON.stringify(allSummary, null, 2), "utf-8");
+
+console.log(`\nSummary: ${SUMMARY_FILE}\n`);
--- a/server/tinybird/scripts/firecrawl-march-export.sh
+++ b/server/tinybird/scripts/firecrawl-march-export.sh
@@ -0,0 +1,92 @@
+#!/usr/bin/env bash
+# Step 1: Export ALL rows for each day of March 2026, hour by hour, to NDJSON files.
+# Exports per hour to stay under Tinybird's 100MB export limit, then concatenates into daily files.
+# Output: ./march-export/day-YYYY-MM-DD.ndjson
+#
+# Usage: ./scripts/firecrawl-march-export.sh
+# Run from: sirtenzin-autumn/server/tinybird/
+
+set -euo pipefail
+
+ORG_ID="biu9vSF7vghBLSKW1UTDwxHBAivjnPaK"
+OUTPUT_DIR="$(pwd)/march-export"
+TMP_DIR="$(pwd)/march-export-tmp"
+mkdir -p "$OUTPUT_DIR" "$TMP_DIR"
+
+echo "=== Firecrawl March 2026 - Export ALL rows (hourly chunks) ==="
+echo "Output dir: $OUTPUT_DIR"
+echo ""
+
+DAYS=()
+for day in $(seq -w 1 31); do
+  date_str="2026-03-${day}"
+  date -j -f "%Y-%m-%d" "$date_str" "+%Y-%m-%d" &>/dev/null || continue
+  DAYS+=("$date_str")
+done
+
+TOTAL_DAYS=${#DAYS[@]}
+COMPLETED=0
+DAY_TIMES=()
+SCRIPT_START=$(date +%s)
+
+for date_str in "${DAYS[@]}"; do
+  outfile="${OUTPUT_DIR}/day-${date_str}.ndjson"
+  if [[ -f "$outfile" ]]; then
+    echo "[${date_str}] Already exported ($(wc -l < "$outfile" | tr -d ' ') rows), skipping."
+    COMPLETED=$((COMPLETED + 1))
+    continue
+  fi
+
+  DAY_START=$(date +%s)
+  echo "[${date_str}] Exporting hour by hour... ($((COMPLETED + 1))/${TOTAL_DAYS})"
+  > "${TMP_DIR}/day-${date_str}.ndjson"
+
+  for hour in $(seq -w 0 23); do
+    for chunk_start_min in 0 10 20 30 40 50; do
+      chunk_end_min=$((chunk_start_min + 9))
+      hour_start=$(printf "%s %s:%02d:00" "$date_str" "$hour" "$chunk_start_min")
+      hour_end=$(printf "%s %s:%02d:59" "$date_str" "$hour" "$chunk_end_min")
+      tmpfile="${TMP_DIR}/chunk-${date_str}-${hour}-${chunk_start_min}.ndjson"
+
+      tb --cloud datasource export events \
+        --format ndjson \
+        --rows 100000000 \
+        --where "toYYYYMM(timestamp) = 202603 AND timestamp >= '${hour_start}' AND timestamp <= '${hour_end}' AND org_id = '${ORG_ID}'" \
+        --target "$tmpfile"
+
+      if [[ -f "$tmpfile" ]]; then
+        cat "$tmpfile" >> "${TMP_DIR}/day-${date_str}.ndjson"
+        rm -f "$tmpfile"
+      fi
+    done
+  done
+
+  mv "${TMP_DIR}/day-${date_str}.ndjson" "$outfile"
+  row_count=$(wc -l < "$outfile" | tr -d ' ')
+  COMPLETED=$((COMPLETED + 1))
+
+  DAY_END=$(date +%s)
+  DAY_ELAPSED=$((DAY_END - DAY_START))
+  DAY_TIMES+=("$DAY_ELAPSED")
+
+  # Rolling average ETA (last 5 days)
+  WINDOW=5
+  WINDOW_START=$(( ${#DAY_TIMES[@]} - WINDOW ))
+  [[ $WINDOW_START -lt 0 ]] && WINDOW_START=0
+  WINDOW_TIMES=("${DAY_TIMES[@]:$WINDOW_START}")
+  SUM=0
+  for t in "${WINDOW_TIMES[@]}"; do SUM=$((SUM + t)); done
+  AVG=$((SUM / ${#WINDOW_TIMES[@]}))
+  REMAINING=$((TOTAL_DAYS - COMPLETED))
+  ETA_SECS=$((REMAINING * AVG))
+  ETA_MIN=$((ETA_SECS / 60))
+  TOTAL_ELAPSED=$(( DAY_END - SCRIPT_START ))
+  ELAPSED_MIN=$((TOTAL_ELAPSED / 60))
+
+  echo "[${date_str}] Done - ${row_count} rows | ${DAY_ELAPSED}s | ${COMPLETED}/${TOTAL_DAYS} days | ${ELAPSED_MIN}min elapsed | ETA ~${ETA_MIN}min (avg ${AVG}s/day)"
+done
+
+rm -rf "$TMP_DIR"
+echo ""
+echo "=== Export complete ==="
+ls -lh "$OUTPUT_DIR"
--- a/server/tinybird/scripts/firecrawl-march-replace.sh
+++ b/server/tinybird/scripts/firecrawl-march-replace.sh
@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+# For each day in March 2026:
+#   1. Export only v2 rows to a temp NDJSON file
+#   2. Replace that day's partition slice with the exported data
+#
+# The replace atomically removes all rows matching the condition and re-inserts
+# the exported v2-only rows - eliminating the duplicates from the reimport.
+#
+# Usage: ./scripts/firecrawl-march-replace.sh
+#
+# Run from sirtenzin-autumn/server/tinybird/
+
+set -euo pipefail
+
+ORG_ID="biu9vSF7vghBLSKW1UTDwxHBAivjnPaK"
+TMPDIR_BASE=$(mktemp -d)
+trap 'rm -rf "$TMPDIR_BASE"' EXIT
+
+echo "=== Firecrawl March 2026 - Dedupe via replace ==="
+echo "Temp dir: $TMPDIR_BASE"
+echo ""
+
+for day in $(seq -w 1 31); do
+  date_str="2026-03-${day}"
+  next_date=$(date -j -v+1d -f "%Y-%m-%d" "$date_str" "+%Y-%m-%d" 2>/dev/null) || continue
+  # Validate date
+  date -j -f "%Y-%m-%d" "$date_str" "+%Y-%m-%d" &>/dev/null || continue
+
+  tmpfile="${TMPDIR_BASE}/march-${day}.ndjson"
+
+  echo "[${date_str}] Exporting v2 rows..."
+  tb --cloud datasource export events \
+    --format ndjson \
+    --where "toYYYYMM(timestamp) = 202603 AND timestamp >= '${date_str}' AND timestamp < '${next_date}' AND org_id = '${ORG_ID}' AND properties.backfill_version = 2" \
+    --target "$tmpfile"
+
+  row_count=$(wc -l < "$tmpfile" | tr -d ' ')
+  echo "[${date_str}] Exported ${row_count} v2 rows. Replacing..."
+
+  tb --cloud datasource replace events "$tmpfile" \
+    --sql-condition "toYYYYMM(timestamp) = 202603 AND timestamp >= '${date_str}' AND timestamp < '${next_date}' AND org_id = '${ORG_ID}'"
+
+  echo "[${date_str}] Done."
+  rm -f "$tmpfile"
+  echo ""
+done
+
+echo "=== All days complete ==="
--- a/server/tinybird/scripts/firecrawl-march-replace.ts
+++ b/server/tinybird/scripts/firecrawl-march-replace.ts
@@ -0,0 +1,189 @@
+/**
+ * Step 3: Replace March 2026 events in Tinybird hour by hour, 10-min chunk by chunk.
+ *
+ * For each hourly file in ./march-deduped-hours/:
+ *   - Splits into 10-minute chunk files (~60MB each, under Tinybird's 100MB limit)
+ *   - Runs `tb --cloud datasource replace events` for each chunk
+ *   - Writes a checkpoint so it's resumable
+ *
+ * Usage: bun run scripts/firecrawl-march-replace.ts [YYYY-MM-DD] [HH]
+ *   e.g: bun run scripts/firecrawl-march-replace.ts 2026-03-01
+ *        bun run scripts/firecrawl-march-replace.ts 2026-03-01 01
+ * Run from: sirtenzin-autumn/server/tinybird/
+ */
+
+import { createReadStream, createWriteStream, existsSync, statSync } from "node:fs";
+import { mkdir, writeFile, readFile, rm } from "node:fs/promises";
+import { createInterface } from "node:readline";
+import { join } from "node:path";
+import { spawnSync } from "node:child_process";
+
+const ORG_ID = "biu9vSF7vghBLSKW1UTDwxHBAivjnPaK";
+const HOURS_DIR = join(process.cwd(), "march-deduped-hours");
+const CHUNKS_DIR = join(process.cwd(), "march-replace-chunks");
+const CHECKPOINT_FILE = join(process.cwd(), "march-replace-checkpoint.json");
+const TB_CWD = process.cwd();
+
+const dayFilter = process.argv[2] ?? null;
+const hourFilter = process.argv[3] ?? null;
+
+if (!dayFilter) {
+  console.error("Usage: bun run scripts/firecrawl-march-replace.ts YYYY-MM-DD [HH]");
+  process.exit(1);
+}
+
+await mkdir(CHUNKS_DIR, { recursive: true });
+
+// ── Checkpoint ────────────────────────────────────────────────────────────────
+type Checkpoint = Record<string, "done" | "partial">;
+let checkpoint: Checkpoint = {};
+try { checkpoint = JSON.parse(await readFile(CHECKPOINT_FILE, "utf-8")); } catch { }
+const saveCheckpoint = async () => writeFile(CHECKPOINT_FILE, JSON.stringify(checkpoint, null, 2), "utf-8");
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+const CHUNK_MINUTES = [0, 10, 20, 30, 40, 50];
+
+const chunkKey = (day: string, hh: string, startMin: number) => `${day}-${hh}-${startMin}`;
+
+const splitHourToChunks = async ({ day, hh, inputPath }: { day: string; hh: string; inputPath: string }) => {
+  const fileSize = statSync(inputPath).size;
+  const streams: Record<number, ReturnType<typeof createWriteStream>> = {};
+  const counts: Record<number, number> = {};
+  const paths: Record<number, string> = {};
+
+  for (const m of CHUNK_MINUTES) {
+    const p = join(CHUNKS_DIR, `${day}-hour${hh}-chunk${m}.ndjson`);
+    streams[m] = createWriteStream(p);
+    counts[m] = 0;
+    paths[m] = p;
+  }
+
+  const rl = createInterface({ input: createReadStream(inputPath), crlfDelay: Infinity });
+  let total = 0;
+  let bytesRead = 0;
+  const start = Date.now();
+
+  for await (const line of rl) {
+    if (!line) continue;
+    bytesRead += line.length + 1;
+    total++;
+
+    const m = line.match(/"timestamp":"[^"]+? \d{2}:(\d{2}):/);
+    if (!m) continue;
+    const mins = parseInt(m[1]);
+    const chunkMin = CHUNK_MINUTES.filter((c) => c <= mins).at(-1) ?? 0;
+    streams[chunkMin].write(line + "\n");
+    counts[chunkMin]++;
+
+    if (total % 100_000 === 0) {
+      const pct = ((bytesRead / fileSize) * 100).toFixed(1);
+      const eta = (((Date.now() - start) / (bytesRead / fileSize) - (Date.now() - start)) / 1000 / 60).toFixed(1);
+      process.stdout.write(`\r  splitting ${day} ${hh}:xx | ${pct}% | ${total.toLocaleString()} rows | ETA ~${eta}min  `);
+    }
+  }
+
+  await Promise.all(Object.values(streams).map((s) => new Promise<void>((res, rej) => s.end((e: Error | null | undefined) => e ? rej(e) : res()))));
+  process.stdout.write("\n");
+
+  return { counts, paths, total };
+};
+
+const runReplace = ({ day, hh, startMin, filePath }: { day: string; hh: string; startMin: number; filePath: string }): boolean => {
+  const endMin = startMin + 9;
+  const tsStart = `${day} ${hh}:${String(startMin).padStart(2, "0")}:00`;
+  const tsEnd = `${day} ${hh}:${String(endMin).padStart(2, "0")}:59`;
+  const condition = `toYYYYMM(timestamp) = 202603 AND timestamp >= '${tsStart}' AND timestamp <= '${tsEnd}' AND org_id = '${ORG_ID}'`;
+
+  const sizeMB = (statSync(filePath).size / 1e6).toFixed(1);
+  process.stdout.write(`  replacing ${day} ${hh}:${String(startMin).padStart(2, "0")}-${String(endMin).padStart(2, "0")} (${sizeMB}MB)... `);
+
+  const result = spawnSync(
+    "tb",
+    ["--cloud", "datasource", "replace", "events", filePath, "--sql-condition", condition],
+    { encoding: "utf-8", timeout: 300_000, cwd: TB_CWD },
+  );
+
+  const output = (result.stdout ?? "") + (result.stderr ?? "");
+  if (result.status !== 0 || output.toLowerCase().includes("error")) {
+    console.log(`FAILED`);
+    console.error(`  Error: ${output.slice(0, 300)}`);
+    return false;
+  }
+
+  console.log(`OK`);
+  return true;
+};
+
+// ── Main ──────────────────────────────────────────────────────────────────────
+const hours = hourFilter ? [hourFilter] : Array.from({ length: 24 }, (_, i) => String(i).padStart(2, "0"));
+const scriptStart = Date.now();
+let totalChunksDone = 0;
+let totalChunksSkipped = 0;
+let totalFailed = 0;
+const TOTAL_CHUNKS = hours.length * 6;
+
+console.log(`\n=== Firecrawl March Replace: ${dayFilter} ${hourFilter ? `hour ${hourFilter}` : "all hours"} ===`);
+console.log(`Chunks dir: ${CHUNKS_DIR}`);
+console.log(`Checkpoint: ${CHECKPOINT_FILE}\n`);
+
+for (const hh of hours) {
+  const inputPath = join(HOURS_DIR, `day-${dayFilter}-hour${hh}.ndjson`);
+  if (!existsSync(inputPath)) {
+    console.log(`[${dayFilter} ${hh}:xx] No file found, skipping`);
+    continue;
+  }
+
+  // Check if all chunks for this hour are already done
+  const allDone = CHUNK_MINUTES.every((m) => checkpoint[chunkKey(dayFilter, hh, m)] === "done");
+  if (allDone) {
+    console.log(`[${dayFilter} ${hh}:xx] All chunks already done, skipping`);
+    totalChunksSkipped += 6;
+    continue;
+  }
+
+  console.log(`\n[${dayFilter} ${hh}:xx] Splitting into 10-min chunks...`);
+  const { counts, paths } = await splitHourToChunks({ day: dayFilter, hh, inputPath });
+
+  for (const startMin of CHUNK_MINUTES) {
+    const key = chunkKey(dayFilter, hh, startMin);
+    if (checkpoint[key] === "done") {
+      console.log(`  chunk ${hh}:${String(startMin).padStart(2, "0")} already done, skipping`);
+      totalChunksSkipped++;
+      continue;
+    }
+
+    const filePath = paths[startMin];
+    const rowCount = counts[startMin];
+
+    if (rowCount === 0) {
+      console.log(`  chunk ${hh}:${String(startMin).padStart(2, "0")}-${String(startMin + 9).padStart(2, "0")} empty, skipping`);
+      checkpoint[key] = "done";
+      await saveCheckpoint();
+      totalChunksSkipped++;
+      continue;
+    }
+
+    const ok = runReplace({ day: dayFilter, hh, startMin, filePath });
+    if (ok) {
+      checkpoint[key] = "done";
+      await saveCheckpoint();
+      totalChunksDone++;
+      await rm(filePath, { force: true });
+    } else {
+      totalFailed++;
+    }
+
+    // ETA
+    const elapsed = (Date.now() - scriptStart) / 1000;
+    const done = totalChunksDone + totalChunksSkipped;
+    const avg = elapsed / Math.max(1, totalChunksDone);
+    const remaining = TOTAL_CHUNKS - done;
+    const etaMin = ((remaining * avg) / 60).toFixed(0);
+    console.log(`  Progress: ${done}/${TOTAL_CHUNKS} chunks | ${totalFailed} failed | ETA ~${etaMin}min`);
+  }
+}
+
+const totalSec = ((Date.now() - scriptStart) / 1000).toFixed(1);
+console.log(`\n=== Replace complete in ${totalSec}s ===`);
+console.log(`Done: ${totalChunksDone} | Skipped: ${totalChunksSkipped} | Failed: ${totalFailed}\n`);
+if (totalFailed > 0) console.log("Re-run the script to retry failed chunks (checkpoint will skip completed ones).");
--- a/server/tinybird/scripts/firecrawl-split-hours.ts
+++ b/server/tinybird/scripts/firecrawl-split-hours.ts
@@ -0,0 +1,82 @@
+/**
+ * Splits a deduped daily NDJSON file into per-hour files.
+ * Streams line by line - O(1) memory.
+ *
+ * Usage: bun run scripts/firecrawl-split-hours.ts 2026-03-01
+ * Run from: sirtenzin-autumn/server/tinybird/
+ */
+
+import { createReadStream, createWriteStream, statSync } from "node:fs";
+import { mkdir } from "node:fs/promises";
+import { createInterface } from "node:readline";
+import { join } from "node:path";
+
+const day = process.argv[2];
+if (!day) { console.error("Usage: bun run scripts/firecrawl-split-hours.ts YYYY-MM-DD"); process.exit(1); }
+
+const INPUT = join(process.cwd(), "march-deduped", `day-${day}.ndjson`);
+const OUT_DIR = join(process.cwd(), "march-deduped-hours");
+
+await mkdir(OUT_DIR, { recursive: true });
+
+const fileSize = statSync(INPUT).size;
+const fileSizeMB = (fileSize / 1e6).toFixed(0);
+
+// Open all 24 write streams upfront
+const streams: Record<string, ReturnType<typeof createWriteStream>> = {};
+const counts: Record<string, number> = {};
+for (let h = 0; h < 24; h++) {
+  const hh = String(h).padStart(2, "0");
+  streams[hh] = createWriteStream(join(OUT_DIR, `day-${day}-hour${hh}.ndjson`));
+  counts[hh] = 0;
+}
+
+console.log(`\n=== Splitting ${day} into hourly files ===`);
+console.log(`Input:  ${INPUT} (${fileSizeMB}MB)`);
+console.log(`Output: ${OUT_DIR}\n`);
+
+const rl = createInterface({ input: createReadStream(INPUT), crlfDelay: Infinity });
+
+let total = 0;
+let bytesRead = 0;
+let skipped = 0;
+const start = Date.now();
+const LOG_EVERY = 250_000;
+
+for await (const line of rl) {
+  if (!line) continue;
+  bytesRead += line.length + 1;
+  total++;
+
+  const m = line.match(/"timestamp":"(\d{4}-\d{2}-\d{2}) (\d{2}):/);
+  if (!m) { skipped++; continue; }
+  const hh = m[2];
+  streams[hh]?.write(line + "\n");
+  counts[hh]++;
+
+  if (total % LOG_EVERY === 0) {
+    const pct = ((bytesRead / fileSize) * 100).toFixed(1);
+    const elapsedSec = (Date.now() - start) / 1000;
+    const etaSec = (elapsedSec / (bytesRead / fileSize)) - elapsedSec;
+    const etaMin = (etaSec / 60).toFixed(1);
+    const mbRead = (bytesRead / 1e6).toFixed(0);
+    const rowsPerSec = Math.round(total / elapsedSec);
+    console.log(`  ${pct}% | ${mbRead}/${fileSizeMB}MB | ${total.toLocaleString()} rows | ${rowsPerSec.toLocaleString()} rows/s | ETA ~${etaMin}min`);
+  }
+}
+
+// Close all streams
+await Promise.all(Object.values(streams).map((s) => new Promise<void>((res, rej) => s.end((e: Error | null | undefined) => e ? rej(e) : res()))));
+
+const totalSec = ((Date.now() - start) / 1000).toFixed(1);
+
+console.log(`\n=== Done in ${totalSec}s ===`);
+console.log(`Total rows: ${total.toLocaleString()} | Skipped: ${skipped}\n`);
+console.log("Hour  Rows");
+console.log("-".repeat(25));
+for (let h = 0; h < 24; h++) {
+  const hh = String(h).padStart(2, "0");
+  console.log(`  ${hh}   ${counts[hh].toLocaleString()}`);
+}
+console.log("-".repeat(25));
+console.log(`Total ${total.toLocaleString()}\n`);