Runs the backfill if this store has not completed it yet. Returns the number of rows that gained facts, or `None` on database errors (in which case the marker is not written and a later open retries).
(conn: &Connection)
| 64 | /// number of rows that gained facts, or `None` on database errors (in which |
| 65 | /// case the marker is not written and a later open retries). |
| 66 | pub(crate) async fn backfill_transcript_facts(conn: &Connection) -> Option<BackfillStats> { |
| 67 | if marker_version(conn).await >= MARKER_VERSION { |
| 68 | return Some(BackfillStats::default()); |
| 69 | } |
| 70 | |
| 71 | let candidates = load_candidates(conn).await?; |
| 72 | |
| 73 | // Re-derive per-line facts file by file *before* opening the write |
| 74 | // transaction; transcripts that no longer exist drop out here and their |
| 75 | // rows simply stay as they are. The first run after an upgrade re-reads |
| 76 | // every affected transcript from byte 0 — easily hundreds of MB of |
| 77 | // JSONL — so the pure read+parse loop runs on the blocking pool instead |
| 78 | // of pinning the async runtime worker that called `open_at`. |
| 79 | let mut by_file: HashMap<(String, String), Vec<(String, i64)>> = HashMap::new(); |
| 80 | for (provider, message_id, source_path, source_offset) in candidates { |
| 81 | by_file |
| 82 | .entry((provider, source_path)) |
| 83 | .or_default() |
| 84 | .push((message_id, source_offset)); |
| 85 | } |
| 86 | let updates = tokio::task::spawn_blocking(move || { |
| 87 | let mut updates: Vec<(String, String, LineFacts)> = Vec::new(); |
| 88 | for ((provider, path), rows) in by_file { |
| 89 | let Some(mut line_facts) = derive_line_facts(&provider, Path::new(&path)) else { |
| 90 | continue; |
| 91 | }; |
| 92 | for (message_id, source_offset) in rows { |
| 93 | if let Some(facts) = line_facts.remove(&source_offset) { |
| 94 | if facts.timestamp.is_some() || facts.usage.is_some() { |
| 95 | updates.push((provider.clone(), message_id, facts)); |
| 96 | } |
| 97 | } |
| 98 | } |
| 99 | } |
| 100 | updates |
| 101 | }) |
| 102 | .await |
| 103 | .ok()?; |
| 104 | |
| 105 | conn.execute("BEGIN IMMEDIATE", ()).await.ok()?; |
| 106 | let applied = apply_updates(conn, &updates).await; |
| 107 | let Some(stats) = applied else { |
| 108 | let _ = conn.execute("ROLLBACK", ()).await; |
| 109 | return None; |
| 110 | }; |
| 111 | if conn.execute("COMMIT", ()).await.is_err() { |
| 112 | let _ = conn.execute("ROLLBACK", ()).await; |
| 113 | return None; |
| 114 | } |
| 115 | if stats.dated > 0 || stats.usage_added > 0 { |
| 116 | eprintln!( |
| 117 | "Backfilled {} timestamp(s) and {} usage record(s) for legacy messages from transcripts.", |
| 118 | stats.dated, stats.usage_added |
| 119 | ); |
| 120 | } |
| 121 | Some(stats) |
| 122 | } |
| 123 |
no test coverage detected