aether-sessions 0.1.0

Persisted session logs and transcript reconstruction for Aether
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

use super::SessionIndexError;
use super::db::Db;
use super::session::AetherSession;
use crate::{DiscoveredSessionFile, FileFingerprint, discover_session_files};
use futures::StreamExt;
use serde::Serialize;
use std::path::PathBuf;

#[derive(Debug, Clone)]
pub struct IngestOptions {
    pub sessions_dir: PathBuf,
    pub db_path: PathBuf,
    pub prune: bool,
    pub parse_concurrency: usize,
}

#[derive(Debug, Clone, Default, Serialize)]
pub struct IngestSummary {
    pub sessions_dir: PathBuf,
    #[serde(rename = "db")]
    pub db_path: PathBuf,
    pub files_seen: usize,
    pub files_indexed: usize,
    pub files_skipped_unchanged: usize,
    pub files_failed: usize,
    pub events_indexed: usize,
    pub parse_errors: usize,
    pub stale_files_pruned: usize,
}

pub async fn ingest_sessions(options: IngestOptions) -> Result<IngestSummary, SessionIndexError> {
    let sessions_dir = options.sessions_dir.canonicalize()?;
    let db_path = options.db_path;
    let mut db = Db::open_writable(&db_path).await?;
    let files = discover_session_files(&sessions_dir)?;
    let mut summary = IngestSummary {
        sessions_dir: sessions_dir.clone(),
        db_path: db_path.clone(),
        files_seen: files.len(),
        ..IngestSummary::default()
    };

    if options.prune {
        summary.stale_files_pruned = db.prune_missing_files(&files).await?;
    }

    let indexed = db.indexed_file_fingerprints().await?;
    let changed_files: Vec<DiscoveredSessionFile> = files
        .into_iter()
        .filter(|file| {
            let source_path = file.path.to_string_lossy();
            if indexed.get(source_path.as_ref()) == Some(&file.fingerprint) {
                summary.files_skipped_unchanged += 1;
                false
            } else {
                true
            }
        })
        .collect();

    let tasks = changed_files.into_iter().map(|file| tokio::task::spawn_blocking(move || parse_changed_file(file)));
    let mut outcomes = futures::stream::iter(tasks).buffered(options.parse_concurrency.max(1));

    while let Some(result) = outcomes.next().await {
        match result? {
            FileIngestOutcome::Parsed(session) => match db.replace_session(&session).await {
                Ok(()) => {
                    summary.events_indexed += session.events.len();
                    summary.parse_errors += session.parse_errors.len();
                    summary.files_indexed += 1;
                }
                Err(error) => {
                    db.record_file_error(&session.source_path, session.fingerprint, error.to_string()).await?;
                    summary.files_failed += 1;
                }
            },
            FileIngestOutcome::Failed { file, fingerprint, error } => {
                db.record_file_error(&file, fingerprint, error).await?;
                summary.files_failed += 1;
            }
        }
    }

    Ok(summary)
}

pub fn default_parse_concurrency() -> usize {
    std::thread::available_parallelism().map_or(4, std::num::NonZeroUsize::get)
}

#[derive(Debug)]
enum FileIngestOutcome {
    Parsed(AetherSession),
    Failed { file: PathBuf, fingerprint: FileFingerprint, error: String },
}

fn parse_changed_file(file: DiscoveredSessionFile) -> FileIngestOutcome {
    match AetherSession::parse(&file.path) {
        Ok(session) => FileIngestOutcome::Parsed(session),
        Err(error) => {
            FileIngestOutcome::Failed { file: file.path, fingerprint: file.fingerprint, error: error.to_string() }
        }
    }
}