use std::path::PathBuf;
use std::sync::mpsc::{Receiver, TryRecvError};
use std::time::Duration;
use tga::collect::{correlate_commits, CollectionPipeline};
use tga::core::config::Config;
use tga::core::db::correlation::{correlation_counts, correlation_rows};
use tga::core::db::Database;
use tga::core::progress::{ProgressBus, ProgressEvent, Stage};
use super::state::{TuiState, RESULTS_LIMIT};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkMode {
PullAndCorrelate,
CorrelateOnly,
}
pub struct WorkerHandle {
db: Database,
db_path: PathBuf,
done: Option<Receiver<String>>,
}
impl WorkerHandle {
pub fn new(db: Database, db_path: PathBuf) -> Self {
Self {
db,
db_path,
done: None,
}
}
pub fn start(&mut self, state: &TuiState, mode: WorkMode) {
let (tx, rx) = std::sync::mpsc::channel();
self.done = Some(rx);
let config = state.scoped_config();
let bus = state.bus.clone();
let db_path = self.db_path.clone();
std::thread::spawn(move || {
let summary = match run_work(config, db_path, mode, &bus) {
Ok(s) => s,
Err(e) => {
bus.emit(ProgressEvent::failed(
Stage::Correlate,
"run",
e.to_string(),
));
format!("run failed: {e}")
}
};
let _ = tx.send(summary);
});
}
pub fn poll_finished(&mut self) -> Option<String> {
let rx = self.done.as_ref()?;
match rx.try_recv() {
Ok(summary) => {
self.done = None;
Some(summary)
}
Err(TryRecvError::Empty) => None,
Err(TryRecvError::Disconnected) => {
self.done = None;
Some("run ended without reporting a result".to_string())
}
}
}
pub fn reload_results(&mut self, state: &mut TuiState) {
let counts = match correlation_counts(self.db.connection()) {
Ok(c) => c,
Err(e) => {
state.set_message(format!("could not read correlation counts: {e}"));
return;
}
};
match correlation_rows(self.db.connection(), state.filter, RESULTS_LIMIT) {
Ok(rows) => state.set_results(counts, rows),
Err(e) => state.set_message(format!("could not read correlation rows: {e}")),
}
}
}
pub(super) fn run_work(
config: Config,
db_path: PathBuf,
mode: WorkMode,
bus: &ProgressBus,
) -> anyhow::Result<String> {
let mut db = Database::open(&db_path)?;
let collected = if mode == WorkMode::PullAndCorrelate {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let stats = runtime.block_on(
CollectionPipeline::new(config)
.with_progress(bus.clone())
.run(&mut db),
)?;
runtime.shutdown_timeout(Duration::from_secs(0));
Some((stats.commits_collected, stats.errors.len()))
} else {
None
};
let outcome = correlate_commits(db.connection_mut(), bus)?;
Ok(match collected {
Some((n, errors)) => format!(
"collected {n} commit(s), {errors} error(s); {}",
outcome.summary()
),
None => format!("correlate only — {}", outcome.summary()),
})
}