feat: background poller with 60s tick, per-tracker interval, event emission
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
e91225d71d
commit
b20ecc5493
3 changed files with 193 additions and 2 deletions
|
|
@ -29,11 +29,21 @@ pub fn run() {
|
||||||
|
|
||||||
let http_client = reqwest::Client::new();
|
let http_client = reqwest::Client::new();
|
||||||
|
|
||||||
|
let db_arc = Arc::new(Mutex::new(conn));
|
||||||
app.manage(AppState {
|
app.manage(AppState {
|
||||||
db: Arc::new(Mutex::new(conn)),
|
db: db_arc.clone(),
|
||||||
|
encryption_key,
|
||||||
|
http_client: http_client.clone(),
|
||||||
|
});
|
||||||
|
|
||||||
|
// Start background poller
|
||||||
|
services::poller::start(
|
||||||
|
db_arc,
|
||||||
encryption_key,
|
encryption_key,
|
||||||
http_client,
|
http_client,
|
||||||
});
|
app.handle().clone(),
|
||||||
|
);
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
})
|
})
|
||||||
.invoke_handler(tauri::generate_handler![
|
.invoke_handler(tauri::generate_handler![
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,4 @@
|
||||||
pub mod crypto;
|
pub mod crypto;
|
||||||
pub mod filter_engine;
|
pub mod filter_engine;
|
||||||
|
pub mod poller;
|
||||||
pub mod tuleap_client;
|
pub mod tuleap_client;
|
||||||
|
|
|
||||||
180
src-tauri/src/services/poller.rs
Normal file
180
src-tauri/src/services/poller.rs
Normal file
|
|
@ -0,0 +1,180 @@
|
||||||
|
use crate::models::credential::TuleapCredentials;
|
||||||
|
use crate::models::ticket::ProcessedTicket;
|
||||||
|
use crate::models::tracker::WatchedTracker;
|
||||||
|
use crate::services::{crypto, filter_engine};
|
||||||
|
use crate::services::tuleap_client::TuleapClient;
|
||||||
|
use rusqlite::Connection;
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
use tauri::{AppHandle, Emitter};
|
||||||
|
use tokio::time::{interval, Duration};
|
||||||
|
|
||||||
|
pub fn start(
|
||||||
|
db: Arc<Mutex<Connection>>,
|
||||||
|
encryption_key: [u8; 32],
|
||||||
|
http_client: reqwest::Client,
|
||||||
|
app_handle: AppHandle,
|
||||||
|
) {
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let mut tick = interval(Duration::from_secs(60));
|
||||||
|
loop {
|
||||||
|
tick.tick().await;
|
||||||
|
poll_all_trackers(&db, &encryption_key, &http_client, &app_handle).await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn poll_all_trackers(
|
||||||
|
db: &Arc<Mutex<Connection>>,
|
||||||
|
encryption_key: &[u8; 32],
|
||||||
|
http_client: &reqwest::Client,
|
||||||
|
app_handle: &AppHandle,
|
||||||
|
) {
|
||||||
|
// 1. Read all enabled trackers and credentials from DB
|
||||||
|
let (trackers, client) = {
|
||||||
|
let conn = match db.lock() {
|
||||||
|
Ok(c) => c,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to lock db: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let trackers = match WatchedTracker::list_all_enabled(&conn) {
|
||||||
|
Ok(t) => t,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to list trackers: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// 2. Read credentials; bail silently if none
|
||||||
|
let creds = match TuleapCredentials::get(&conn) {
|
||||||
|
Ok(Some(c)) => c,
|
||||||
|
Ok(None) => return,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to read credentials: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let password = match crypto::decrypt(encryption_key, &creds.password_encrypted) {
|
||||||
|
Ok(p) => p,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to decrypt password: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let client = TuleapClient::new(http_client, &creds.tuleap_url, &creds.username, &password);
|
||||||
|
|
||||||
|
(trackers, client)
|
||||||
|
}; // lock released
|
||||||
|
|
||||||
|
// 3. For each tracker that should_poll, poll it
|
||||||
|
for tracker in &trackers {
|
||||||
|
if should_poll(tracker) {
|
||||||
|
poll_single_tracker(db, &client, tracker, app_handle).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn should_poll(tracker: &WatchedTracker) -> bool {
|
||||||
|
let last_polled_at = match &tracker.last_polled_at {
|
||||||
|
None => return true, // Never polled
|
||||||
|
Some(s) => s,
|
||||||
|
};
|
||||||
|
|
||||||
|
let last = match chrono::DateTime::parse_from_rfc3339(last_polled_at) {
|
||||||
|
Ok(dt) => dt,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to parse last_polled_at '{}': {}", last_polled_at, e);
|
||||||
|
return true; // Treat as never polled on parse error
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let elapsed = chrono::Utc::now().signed_duration_since(last).num_minutes();
|
||||||
|
elapsed >= tracker.polling_interval as i64
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn poll_single_tracker(
|
||||||
|
db: &Arc<Mutex<Connection>>,
|
||||||
|
client: &TuleapClient,
|
||||||
|
tracker: &WatchedTracker,
|
||||||
|
app_handle: &AppHandle,
|
||||||
|
) {
|
||||||
|
// 1. Fetch artifacts
|
||||||
|
let artifacts = match client.get_artifacts(tracker.tracker_id).await {
|
||||||
|
Ok(a) => a,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to fetch artifacts for tracker {}: {}", tracker.id, e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// 2. Apply filters
|
||||||
|
let filtered = filter_engine::apply_filters(&artifacts, &tracker.filters);
|
||||||
|
|
||||||
|
// 3. Insert new tickets and update last_polled_at
|
||||||
|
let new_count = {
|
||||||
|
let conn = match db.lock() {
|
||||||
|
Ok(c) => c,
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to lock db for insert: {}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut count = 0usize;
|
||||||
|
|
||||||
|
for artifact in &filtered {
|
||||||
|
let artifact_id = artifact
|
||||||
|
.get("id")
|
||||||
|
.and_then(|v| v.as_i64())
|
||||||
|
.unwrap_or(0) as i32;
|
||||||
|
|
||||||
|
let artifact_title = artifact
|
||||||
|
.get("title")
|
||||||
|
.and_then(|v| v.as_str())
|
||||||
|
.unwrap_or("")
|
||||||
|
.to_string();
|
||||||
|
|
||||||
|
let artifact_data = serde_json::to_string(artifact)
|
||||||
|
.unwrap_or_else(|_| "{}".to_string());
|
||||||
|
|
||||||
|
match ProcessedTicket::insert_if_new(
|
||||||
|
&conn,
|
||||||
|
&tracker.id,
|
||||||
|
artifact_id,
|
||||||
|
&artifact_title,
|
||||||
|
&artifact_data,
|
||||||
|
) {
|
||||||
|
Ok(Some(_)) => count += 1,
|
||||||
|
Ok(None) => {}
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("poller: failed to insert ticket (artifact {}): {}", artifact_id, e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 4. Update last_polled_at
|
||||||
|
if let Err(e) = WatchedTracker::update_last_polled(&conn, &tracker.id) {
|
||||||
|
eprintln!("poller: failed to update last_polled_at for tracker {}: {}", tracker.id, e);
|
||||||
|
}
|
||||||
|
|
||||||
|
count
|
||||||
|
}; // lock released
|
||||||
|
|
||||||
|
// 5. Emit event if new tickets found
|
||||||
|
if new_count > 0 {
|
||||||
|
if let Err(e) = app_handle.emit(
|
||||||
|
"new-tickets-detected",
|
||||||
|
serde_json::json!({
|
||||||
|
"tracker_id": tracker.id,
|
||||||
|
"tracker_label": tracker.tracker_label,
|
||||||
|
"count": new_count,
|
||||||
|
}),
|
||||||
|
) {
|
||||||
|
eprintln!("poller: failed to emit event: {}", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue