From 6eee90f128a8b9bb5a21936488a7489e030eb859 Mon Sep 17 00:00:00 2001 From: Ruben Rosario Date: Sun, 21 Jun 2026 14:21:14 +0100 Subject: [PATCH] Add task count to task list panel header Show 'X todo / Y done' in the Tasks panel title bar. Also includes prior uncommitted work: - Pagination in fetch_tasks (maxResults=100 + pageToken loop) - fetch_tasks_since for incremental pull sync - SyncStats struct with version/last_sync/last_pull/changed counts - Periodic push (30s) and pull (5min) sync engine - event::poll(100ms) for non-blocking UI refresh - Ctrl+R full sync (push + pull) - refresh_if_needed() to reload data after background sync - Retry mechanism (MAX_SYNC_RETRIES=3) for sync queue items - HTTP status code checks in fetch_lists/fetch_tasks/fetch_tasks_since - Fix move_task URL to use reqwest query() - Remove CASCADE via replace_all_lists (use insert_list instead) - has_pending_sync() to prevent pull during pending push --- src/app.rs | 38 +++++++++ src/domain/models.rs | 3 + src/infrastructure/api.rs | 162 +++++++++++++++++++++++++++++------- src/infrastructure/db.rs | 55 +++++++++---- src/main.rs | 168 +++++++++++++++++++++++++++++++++----- src/ui/components.rs | 33 +++++++- src/ui/mod.rs | 4 +- 7 files changed, 394 insertions(+), 69 deletions(-) diff --git a/src/app.rs b/src/app.rs index 73f7e13..57febb6 100644 --- a/src/app.rs +++ b/src/app.rs @@ -10,6 +10,15 @@ use crate::infrastructure::api::ApiClient; use crate::infrastructure::db::Db; use crate::ui::{Focus, NetworkStatus, Popup}; +#[derive(Debug, Clone, Default)] +pub struct SyncStats { + pub last_sync_time: Option, + pub last_pull_time: Option, + pub lists_changed: usize, + pub tasks_changed: usize, + pub version: u64, +} + pub struct App { pub lists: Vec, pub tasks: Vec, @@ -29,6 +38,8 @@ pub struct App { pub api_client: Arc, pub needs_auth: bool, pub auth_error: Option, + pub sync_stats: SyncStats, + last_sync_version: u64, auth_tx: std_mpsc::Sender, auth_rx: std_mpsc::Receiver, sync_tx: mpsc::Sender, @@ -42,6 +53,7 @@ enum AuthEvent { #[allow(dead_code)] pub enum SyncCommand { TriggerSync, + FullSync, InitialSync, Shutdown, } @@ -86,6 +98,8 @@ impl App { api_client, needs_auth: !has_token, auth_error: None, + sync_stats: SyncStats::default(), + last_sync_version: 0, auth_tx, auth_rx, sync_tx, @@ -142,10 +156,31 @@ impl App { } } + pub fn refresh_if_needed(&mut self) { + if self.sync_stats.version != self.last_sync_version { + self.last_sync_version = self.sync_stats.version; + self.load_lists(); + if !self.lists.is_empty() && self.selected_list < self.lists.len() { + self.tasks = self.db.get_tasks(&self.lists[self.selected_list].id); + } else { + self.tasks.clear(); + } + if self.selected_task >= self.tasks.len() && !self.tasks.is_empty() { + self.selected_task = self.tasks.len() - 1; + } else if self.tasks.is_empty() { + self.selected_task = 0; + } + } + } + fn trigger_sync(&self) { let _ = self.sync_tx.try_send(SyncCommand::TriggerSync); } + fn trigger_full_sync(&self) { + let _ = self.sync_tx.try_send(SyncCommand::FullSync); + } + pub fn handle_key(&mut self, key: KeyEvent) { if let Some(ref popup) = self.show_popup.clone() { self.handle_popup_key(key, popup); @@ -238,6 +273,9 @@ impl App { KeyCode::Esc => { self.show_popup = None; } + KeyCode::Char('r') if key.modifiers.contains(KeyModifiers::CONTROL) => { + self.trigger_full_sync(); + } KeyCode::Char('q') | KeyCode::Char('Q') => { self.should_quit = true; } diff --git a/src/domain/models.rs b/src/domain/models.rs index 0769681..e205cf6 100644 --- a/src/domain/models.rs +++ b/src/domain/models.rs @@ -40,4 +40,7 @@ pub struct SyncQueueItem { pub list_id: String, pub payload: String, pub created_at: String, + pub retries: i32, } + +pub const MAX_SYNC_RETRIES: i32 = 3; diff --git a/src/infrastructure/api.rs b/src/infrastructure/api.rs index b116078..3467841 100644 --- a/src/infrastructure/api.rs +++ b/src/infrastructure/api.rs @@ -97,6 +97,12 @@ impl ApiClient { .await .map_err(|e| ApiError::Network(e.to_string()))?; + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(ApiError::Api(format!("Fetch lists failed: {} - {}", status, body))); + } + let data: serde_json::Value = resp .json() .await @@ -118,27 +124,128 @@ impl ApiClient { pub async fn fetch_tasks(&self, list_id: &str) -> Result, ApiError> { let token = self.get_token().await?; - let url = format!( - "https://tasks.googleapis.com/tasks/v1/lists/{}/tasks?showCompleted=true&showHidden=true", - list_id - ); + let mut all_items: Vec = Vec::new(); + let mut page_token: Option = None; - let resp = self - .client - .get(&url) - .bearer_auth(&token) - .send() - .await - .map_err(|e| ApiError::Network(e.to_string()))?; + loop { + let mut url = format!( + "https://tasks.googleapis.com/tasks/v1/lists/{}/tasks?showCompleted=true&showHidden=true&maxResults=100", + list_id + ); + if let Some(ref pt) = page_token { + url.push_str(&format!("&pageToken={}", pt)); + } - let data: serde_json::Value = resp - .json() - .await - .map_err(|e| ApiError::Api(e.to_string()))?; + let resp = self + .client + .get(&url) + .bearer_auth(&token) + .send() + .await + .map_err(|e| ApiError::Network(e.to_string()))?; - let empty = vec![]; - let items = data["items"].as_array().unwrap_or(&empty); - let tasks = items + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(ApiError::Api(format!("Fetch tasks failed: {} - {}", status, body))); + } + + let data: serde_json::Value = resp + .json() + .await + .map_err(|e| ApiError::Api(e.to_string()))?; + + if let Some(items) = data["items"].as_array() { + all_items.extend(items.iter().cloned()); + } + + match data["nextPageToken"].as_str() { + Some(token) if !token.is_empty() => page_token = Some(token.to_string()), + _ => break, + } + } + + let tasks = all_items + .iter() + .enumerate() + .map(|(i, item)| { + let due_str = item["due"].as_str().and_then(|s| { + chrono::NaiveDateTime::parse_from_str( + &s.replace("T", " ") + .replace("Z", "") + .chars() + .take(16) + .collect::(), + "%Y-%m-%d %H:%M", + ) + .ok() + }); + + Task { + id: item["id"].as_str().unwrap_or("").to_string(), + list_id: list_id.to_string(), + title: item["title"].as_str().unwrap_or("").to_string(), + notes: item["notes"].as_str().map(|s| s.to_string()), + status: if item["status"].as_str() == Some("completed") { + TaskStatus::Completed + } else { + TaskStatus::NeedsAction + }, + due: due_str, + position: i as i64, + } + }) + .collect(); + + Ok(tasks) + } + + pub async fn fetch_tasks_since(&self, list_id: &str, since: &chrono::NaiveDateTime) -> Result, ApiError> { + let token = self.get_token().await?; + let since_str = since.format("%Y-%m-%dT%H:%M:%S.000Z").to_string(); + + let mut all_items: Vec = Vec::new(); + let mut page_token: Option = None; + + loop { + let mut url = format!( + "https://tasks.googleapis.com/tasks/v1/lists/{}/tasks?showCompleted=true&showHidden=true&maxResults=100&updatedMin={}", + list_id, since_str + ); + if let Some(ref pt) = page_token { + url.push_str(&format!("&pageToken={}", pt)); + } + + let resp = self + .client + .get(&url) + .bearer_auth(&token) + .send() + .await + .map_err(|e| ApiError::Network(e.to_string()))?; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(ApiError::Api(format!("Fetch tasks since failed: {} - {}", status, body))); + } + + let data: serde_json::Value = resp + .json() + .await + .map_err(|e| ApiError::Api(e.to_string()))?; + + if let Some(items) = data["items"].as_array() { + all_items.extend(items.iter().cloned()); + } + + match data["nextPageToken"].as_str() { + Some(token) if !token.is_empty() => page_token = Some(token.to_string()), + _ => break, + } + } + + let tasks = all_items .iter() .enumerate() .map(|(i, item)| { @@ -293,28 +400,25 @@ impl ApiClient { ) -> Result<(), ApiError> { let token = self.get_token().await?; - let mut url = format!( + let url = format!( "https://tasks.googleapis.com/tasks/v1/lists/{}/tasks/{}/move", list_id, task_id ); + let mut req = self.client.post(&url).bearer_auth(&token); if let Some(p) = prev { - url.push_str(&format!("&previous={}", p)); + req = req.query(&[("previous", p)]); } if let Some(s) = sibling { - url.push_str(&format!("&destinationTaskList={}", s)); + req = req.query(&[("destinationTaskList", s)]); } - let resp = self - .client - .post(&url) - .bearer_auth(&token) - .send() - .await - .map_err(|e| ApiError::Network(e.to_string()))?; + let resp = req.send().await.map_err(|e| ApiError::Network(e.to_string()))?; if !resp.status().is_success() { - return Err(ApiError::Api(format!("Move failed: {}", resp.status()))); + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(ApiError::Api(format!("Move failed: {} - {}", status, body))); } Ok(()) diff --git a/src/infrastructure/db.rs b/src/infrastructure/db.rs index f1c621c..3ea5396 100644 --- a/src/infrastructure/db.rs +++ b/src/infrastructure/db.rs @@ -37,9 +37,14 @@ impl Db { task_id TEXT NOT NULL, list_id TEXT NOT NULL, payload TEXT NOT NULL, - created_at TEXT NOT NULL + created_at TEXT NOT NULL, + retries INTEGER NOT NULL DEFAULT 0 );", )?; + conn.execute_batch( + "ALTER TABLE sync_queue ADD COLUMN retries INTEGER NOT NULL DEFAULT 0;", + ) + .ok(); Ok(Self { conn: Mutex::new(conn) }) } @@ -201,16 +206,6 @@ impl Db { Ok(()) } - pub fn replace_all_lists(&self, lists: &[TaskList]) -> SqlResult<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM task_lists", [])?; - drop(conn); - for list in lists { - self.insert_list(list)?; - } - Ok(()) - } - pub fn replace_all_tasks(&self, list_id: &str, tasks: &[Task]) -> SqlResult<()> { let conn = self.conn.lock().unwrap(); conn.execute("DELETE FROM tasks WHERE list_id = ?1", params![list_id])?; @@ -226,7 +221,27 @@ impl Db { Ok(()) } - pub fn push_sync(&self, action: SyncAction, task_id: &str, list_id: &str, payload: &str) -> SqlResult<()> { + pub fn push_sync( + &self, + action: SyncAction, + task_id: &str, + list_id: &str, + payload: &str, + ) -> SqlResult<()> { + self.push_sync_with_retry(action, task_id, list_id, payload, 0) + } + + pub fn push_sync_with_retry( + &self, + action: SyncAction, + task_id: &str, + list_id: &str, + payload: &str, + retries: i32, + ) -> SqlResult<()> { + if retries > MAX_SYNC_RETRIES { + return Ok(()); + } let action_str = match action { SyncAction::Create => "Create", SyncAction::Update => "Update", @@ -235,24 +250,33 @@ impl Db { }; let conn = self.conn.lock().unwrap(); conn.execute( - "INSERT INTO sync_queue (action, task_id, list_id, payload, created_at) - VALUES (?1, ?2, ?3, ?4, ?5)", + "INSERT INTO sync_queue (action, task_id, list_id, payload, created_at, retries) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", params![ action_str, task_id, list_id, payload, chrono::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string(), + retries, ], )?; Ok(()) } + pub fn has_pending_sync(&self) -> bool { + let conn = self.conn.lock().unwrap(); + let count: i64 = conn + .query_row("SELECT COUNT(*) FROM sync_queue", [], |row| row.get(0)) + .unwrap_or(0); + count > 0 + } + pub fn drain_sync(&self) -> Vec { let conn = self.conn.lock().unwrap(); let items: Vec = { let mut stmt = conn - .prepare("SELECT id, action, task_id, list_id, payload, created_at FROM sync_queue ORDER BY id") + .prepare("SELECT id, action, task_id, list_id, payload, created_at, retries FROM sync_queue ORDER BY id") .unwrap(); stmt.query_map([], |row| { let action_str: String = row.get(1)?; @@ -270,6 +294,7 @@ impl Db { list_id: row.get(3)?, payload: row.get(4)?, created_at: row.get(5)?, + retries: row.get(6)?, }) }) .unwrap() diff --git a/src/main.rs b/src/main.rs index 69c7ae6..bcd3798 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,7 @@ use ratatui::backend::CrosstermBackend; use ratatui::Terminal; use tokio::sync::Mutex; -use crate::app::{App, SyncCommand}; +use crate::app::{App, SyncCommand, SyncStats}; use crate::domain::models::*; use crate::infrastructure::api::ApiClient; use crate::infrastructure::db::Db; @@ -82,18 +82,20 @@ fn main() -> io::Result<()> { ); let network_status = Arc::new(Mutex::new(NetworkStatus::Online)); + let sync_stats = Arc::new(Mutex::new(SyncStats::default())); let (sync_tx, mut sync_rx) = tokio::sync::mpsc::channel::(32); let mut app = App::new(db.clone(), api_client.clone(), sync_tx.clone()); let network_clone = network_status.clone(); + let stats_clone = sync_stats.clone(); let db_clone = db.clone(); let api_clone = api_client.clone(); std::thread::spawn(move || { let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { - run_sync_engine(db_clone, api_clone, network_clone, &mut sync_rx).await; + run_sync_engine(db_clone, api_clone, network_clone, stats_clone, &mut sync_rx).await; }); }); @@ -116,6 +118,14 @@ fn main() -> io::Result<()> { }; app.network_status = status; + { + let guard = sync_stats.blocking_lock(); + app.sync_stats = guard.clone(); + } + + // Reload lists/tasks if sync engine changed data in background + app.refresh_if_needed(); + let view = AppView { lists: &app.lists, tasks: &app.tasks, @@ -130,12 +140,15 @@ fn main() -> io::Result<()> { task_list_scroll: app.task_list_scroll, detail_scroll: app.detail_scroll, auth_error: app.auth_error.as_deref(), + sync_stats: &app.sync_stats, }; draw(frame, view); })?; - if let Event::Key(key) = event::read()? { - app.handle_key(key); + if event::poll(std::time::Duration::from_millis(100))? { + if let Event::Key(key) = event::read()? { + app.handle_key(key); + } } } @@ -148,22 +161,32 @@ async fn run_sync_engine( db: Arc, api: Arc, network_status: Arc>, + sync_stats: Arc>, rx: &mut tokio::sync::mpsc::Receiver, ) { - let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30)); + let mut push_interval = tokio::time::interval(tokio::time::Duration::from_secs(30)); + let mut pull_interval = tokio::time::interval(tokio::time::Duration::from_secs(300)); + pull_interval.tick().await; loop { tokio::select! { - _ = interval.tick() => { - process_sync_queue(&db, &api, &network_status).await; + _ = push_interval.tick() => { + push_sync(&db, &api, &network_status, &sync_stats).await; + } + _ = pull_interval.tick() => { + pull_sync(&db, &api, &network_status, &sync_stats, false).await; } cmd = rx.recv() => { match cmd { Some(SyncCommand::TriggerSync) => { - process_sync_queue(&db, &api, &network_status).await; + push_sync(&db, &api, &network_status, &sync_stats).await; + } + Some(SyncCommand::FullSync) => { + push_sync(&db, &api, &network_status, &sync_stats).await; + pull_sync(&db, &api, &network_status, &sync_stats, true).await; } Some(SyncCommand::InitialSync) => { - run_initial_sync(&db, &api, &network_status).await; + run_initial_sync(&db, &api, &network_status, &sync_stats).await; } Some(SyncCommand::Shutdown) | None => break, } @@ -176,14 +199,22 @@ async fn run_initial_sync( db: &Arc, api: &Arc, network_status: &Arc>, + sync_stats: &Arc>, ) { *network_status.lock().await = NetworkStatus::Syncing; + let mut total_lists = 0usize; + let mut total_tasks = 0usize; + match api.fetch_lists().await { Ok(lists) => { - db.replace_all_lists(&lists).ok(); + total_lists = lists.len(); + for list in &lists { + db.insert_list(list).ok(); + } for list in &lists { if let Ok(tasks) = api.fetch_tasks(&list.id).await { + total_tasks += tasks.len(); db.replace_all_tasks(&list.id, &tasks).ok(); } } @@ -193,12 +224,21 @@ async fn run_initial_sync( *network_status.lock().await = NetworkStatus::Offline; } } + + let now = chrono::Local::now().naive_local(); + let mut stats = sync_stats.lock().await; + stats.last_sync_time = Some(now); + stats.last_pull_time = Some(now); + stats.lists_changed = total_lists; + stats.tasks_changed = total_tasks; + stats.version += 1; } -async fn process_sync_queue( +async fn push_sync( db: &Arc, api: &Arc, network_status: &Arc>, + sync_stats: &Arc>, ) { if !api.has_token() { *network_status.lock().await = NetworkStatus::Offline; @@ -206,8 +246,9 @@ async fn process_sync_queue( } let items = db.drain_sync(); + let count = items.len(); - if items.is_empty() { + if count == 0 { *network_status.lock().await = NetworkStatus::Online; return; } @@ -249,21 +290,106 @@ async fn process_sync_queue( } }; - if result.is_err() { - let _ = db.push_sync( - item.action.clone(), - &item.task_id, - &item.list_id, - &item.payload, - ); + if let Err(err) = result { + eprintln!("[task_app] Sync failed (retry {}/{}): action={:?} task={} error={}", + item.retries, MAX_SYNC_RETRIES, item.action, item.task_id, err); + if item.retries < MAX_SYNC_RETRIES { + let _ = db.push_sync_with_retry( + item.action.clone(), + &item.task_id, + &item.list_id, + &item.payload, + item.retries + 1, + ); + } else { + eprintln!("[task_app] Dropping sync item after {} failed attempts: action={:?} task={}", + MAX_SYNC_RETRIES, item.action, item.task_id); + } all_ok = false; - break; } } *network_status.lock().await = if all_ok { NetworkStatus::Online } else { - NetworkStatus::Offline + let remaining = db.has_pending_sync(); + if remaining { + NetworkStatus::Offline + } else { + NetworkStatus::Online + } }; + + let mut stats = sync_stats.lock().await; + stats.last_sync_time = Some(chrono::Local::now().naive_local()); + stats.lists_changed = 0; + stats.tasks_changed = count; + stats.version += 1; +} + +async fn pull_sync( + db: &Arc, + api: &Arc, + network_status: &Arc>, + sync_stats: &Arc>, + force_full: bool, +) { + if !api.has_token() { + *network_status.lock().await = NetworkStatus::Offline; + return; + } + + if db.has_pending_sync() { + return; + } + + *network_status.lock().await = NetworkStatus::Syncing; + + let mut total_lists = 0usize; + let mut total_tasks = 0usize; + + let last_pull = { + let stats = sync_stats.lock().await; + stats.last_pull_time + }; + + let use_incremental = !force_full && last_pull.is_some(); + + match api.fetch_lists().await { + Ok(lists) => { + total_lists = lists.len(); + for list in &lists { + db.insert_list(list).ok(); + } + for list in &lists { + let result = if use_incremental { + api.fetch_tasks_since(&list.id, last_pull.as_ref().unwrap()).await + } else { + api.fetch_tasks(&list.id).await + }; + if let Ok(tasks) = result { + total_tasks += tasks.len(); + if use_incremental { + for task in &tasks { + db.insert_task(task).ok(); + } + } else { + db.replace_all_tasks(&list.id, &tasks).ok(); + } + } + } + *network_status.lock().await = NetworkStatus::Online; + } + Err(_) => { + *network_status.lock().await = NetworkStatus::Offline; + } + } + + let now = chrono::Local::now().naive_local(); + let mut stats = sync_stats.lock().await; + stats.last_sync_time = Some(now); + stats.last_pull_time = Some(now); + stats.lists_changed = total_lists; + stats.tasks_changed = total_tasks; + stats.version += 1; } diff --git a/src/ui/components.rs b/src/ui/components.rs index d48e6b2..079b889 100644 --- a/src/ui/components.rs +++ b/src/ui/components.rs @@ -5,6 +5,7 @@ use ratatui::layout::{Alignment, Rect}; use ratatui::Frame; use crate::domain::models::*; +use crate::app::SyncStats; use super::NetworkStatus; const TAB_COLOR: Color = Color::Cyan; @@ -56,6 +57,10 @@ pub fn render_task_list( focused: bool, _scroll: u16, ) { + let total = tasks.len(); + let done = tasks.iter().filter(|t| t.status == TaskStatus::Completed).count(); + let todo = total - done; + let items: Vec = tasks .iter() .map(|task| { @@ -100,7 +105,7 @@ pub fn render_task_list( let block = Block::default() .borders(Borders::ALL) .border_style(Style::default().fg(if focused { FOCUS_COLOR } else { Color::DarkGray })) - .title(" Tasks ") + .title(format!(" Tasks ({} todo / {} done) ", todo, done)) .title_alignment(Alignment::Left); let list = List::new(items) @@ -185,18 +190,40 @@ pub fn render_status_bar( frame: &mut Frame, area: Rect, status: &NetworkStatus, + sync_stats: &SyncStats, ) { - let (text, color) = match status { + let (status_text, color) = match status { NetworkStatus::Online => (" ONLINE ", STATUS_ONLINE), NetworkStatus::Offline => (" OFFLINE ", STATUS_OFFLINE), NetworkStatus::Syncing => (" SYNCING... ", STATUS_SYNC), }; + let right_text = match sync_stats.last_sync_time { + Some(time) => { + let time_str = time.format("%H:%M:%S").to_string(); + let mut parts: Vec = Vec::new(); + if sync_stats.lists_changed > 0 { + parts.push(format!("{} lists", sync_stats.lists_changed)); + } + if sync_stats.tasks_changed > 0 { + parts.push(format!("{} tasks", sync_stats.tasks_changed)); + } + if parts.is_empty() { + format!(" {} last sync ", time_str) + } else { + format!(" {} last sync: {} ", time_str, parts.join(", ")) + } + } + None => String::new(), + }; + + let full_text = format!("{}{}", status_text, right_text); + let block = Block::default() .style(Style::default().bg(color).fg(Color::Black)); let paragraph = Paragraph::new(Line::from(Span::styled( - text, + full_text, Style::default().fg(Color::Black).add_modifier(Modifier::BOLD), ))) .block(block) diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 9985a62..b7b2838 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -3,6 +3,7 @@ pub mod components; use ratatui::layout::{Constraint, Direction, Layout}; use ratatui::Frame; +use crate::app::SyncStats; use crate::domain::models::*; use components::*; @@ -42,6 +43,7 @@ pub struct AppView<'a> { pub task_list_scroll: u16, pub detail_scroll: u16, pub auth_error: Option<&'a str>, + pub sync_stats: &'a SyncStats, } pub fn draw(frame: &mut Frame, view: AppView) { @@ -87,7 +89,7 @@ pub fn draw(frame: &mut Frame, view: AppView) { view.detail_scroll, ); - render_status_bar(frame, status_area, view.network_status); + render_status_bar(frame, status_area, view.network_status, view.sync_stats); if let Some(popup) = view.show_popup { match popup {