Function import_source

src/analytics_bridge.rs:103–169

(gdb: &GlobalDb, source: &HookImportSource)

}

async fn import_source(gdb: &GlobalDb, source: &HookImportSource) -> HookImportSourceOutcome {
    let mut result = HookImportSourceOutcome {
        path: source.path.clone(),
        imported: 0,
        skipped: 0,
        error: None,
    };
    let Ok(metadata) = std::fs::metadata(&source.path) else {
        return result;
    };
    let file_len = metadata.len();
    let mtime = metadata
        .modified()
        .ok()
        .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
        .map_or(0, |duration| duration.as_secs());

    let cursor_key = import_cursor_key(&source.path);
    let start = match gdb.get_parse_offset(&cursor_key).await {
        // Truncated/rotated files restart from the top.
        Some(cursor) if cursor.byte_offset <= file_len => cursor.byte_offset,
        _ => 0,
    };
    if start == file_len {
        return result;
    }

    let text = match read_from_offset(&source.path, start) {
        Ok(text) => text,
        Err(err) => {
            result.error = Some(err);
            return result;
        }
    };
    // Only consume up to the last complete line; a concurrent writer may have
    // an unfinished row at EOF.
    let consumed = text.rfind('\n').map_or(0, |index| index + 1);
    if consumed == 0 {
        return result;
    }

    let mut batch = Vec::new();
    for line in text[..consumed].lines() {
        match hook_row_to_analytics_event(line, source.default_project_root.as_deref()) {
            Some(event) => batch.push(event),
            None => result.skipped += 1,
        }
    }
    for chunk in batch.chunks(IMPORT_BATCH_SIZE) {
        if let Err(err) = gdb.append_analytics_events(chunk).await {
            result.error = Some(err);
            return result;
        }
        result.imported += chunk.len() as u64;
    }

    gdb.set_parse_offset(
        &cursor_key,