From 1ec4d2aa174f4b715b2e6294f20ae7a8ad449ac2 Mon Sep 17 00:00:00 2001 From: NimBold Date: Tue, 16 Jun 2026 17:02:22 +0330 Subject: [PATCH] docs: add download queue coordinator implementation plan 12 TDD tasks covering: QueueManager scaffold, dispatcher with idle-park and CAS resize, uniform permit parking, gid-race handling, reordering, AppState wiring, aria2 WS poller + completion listener, frontend store refactor, queued UI visuals, deprecated command cleanup, smoke tests. --- .../2026-06-16-download-queue-coordinator.md | 3055 +++++++++++++++++ 1 file changed, 3055 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-16-download-queue-coordinator.md diff --git a/docs/superpowers/plans/2026-06-16-download-queue-coordinator.md b/docs/superpowers/plans/2026-06-16-download-queue-coordinator.md new file mode 100644 index 0000000..2a9eb44 --- /dev/null +++ b/docs/superpowers/plans/2026-06-16-download-queue-coordinator.md @@ -0,0 +1,3055 @@ +# Download Queue Coordinator Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace the frontend-imperative download dispatcher with a backend `QueueManager` that is the sole concurrency gatekeeper, using a single `tokio::sync::Semaphore` to enforce `max_concurrent_tasks` across all three download paths (aria2 RPC, native HTTP, yt-dlp media). + +**Architecture:** A new `queue.rs` module owns an ordered `VecDeque`, an `Arc`, an `active_permits` permit-parking map, and a long-running dispatcher task that idle-parks on a `Notify` and acquires permits via `acquire_owned()`. Permits are uniformly parked by download id so aria2 (whose RPC returns immediately) and pause (which must release the slot before the sidecar dies) both work. The frontend becomes reactive: it requests enqueue/pause/move and listens for `download-state` events. + +**Tech Stack:** Rust + Tokio (`tokio::sync::{Semaphore, Notify, Mutex, watch, mpsc}`, `OwnedSemaphorePermit`), Tauri 2 state management, ts-rs for binding generation, Zustand + React + TypeScript + Tailwind on the frontend. + +**Spec:** `docs/superpowers/specs/2026-06-16-download-queue-coordinator-design.md` + +--- + +## File Structure + +**Create:** +- `src-tauri/src/queue.rs` — `QueueManager`, `QueuedTask`, `SpawnPayload`, `TaskKind`, `QueueDirection`, dispatcher loop, permit parking, CAS resize. ~350 lines. +- `src-tauri/tests/queue_manager.rs` — Integration tests for all four design-discovered bugs + reordering. ~400 lines. +- `src/bindings/DownloadStateEvent.ts` — ts-rs generated. +- `src/bindings/QueueDirection.ts` — ts-rs generated. + +**Modify:** +- `src-tauri/src/lib.rs` — wire `QueueManager` into `AppState`; convert `start_download`/`start_media_download` to `pub(crate)`; add `enqueue_download`/`enqueue_many`/`resume_download`/`move_in_queue`/`remove_from_queue` commands; rewrite `set_concurrent_limit`; extend aria2 WS poller with `release_permit` + `pending_completion`; raise aria2 daemon ceiling. +- `src-tauri/src/ipc.rs` — add `DownloadStateEvent`, `QueueDirection`; remove `dispatched` field from `DownloadItem`. +- `src/store/useDownloadStore.ts` — remove `processQueue`; add `pendingOrder` slice; reactive `addDownload`/`startQueue`/`pauseQueue`/`pauseDownload`/`resumeDownload`/`moveUp`/`moveDown`; simplify `updateDownload`; startup re-enqueue. +- `src/store/downloadStore.ts` — add `download-state` listener (authoritative for `pendingOrder`); simplify progress listener. +- `src/components/DownloadItem.tsx` — queued visuals (clock icon, grayscale fill, position badge) + Move Up/Down buttons. +- `src/components/DownloadTable.tsx` — update `handleResume` to call backend; remove `processQueue` call. +- `src/index.css` — `.download-progress-fill.queued`, `.download-status-queued`, `.queue-position-badge`, `.animate-pulse-slow` keyframes. + +--- + +## Conventions + +- **All Rust code blocks are complete, copy-pasteable code** — no `...` elision unless the surrounding code is unchanged and clearly marked "existing code". +- **Run Rust tests with:** `cd src-tauri && cargo test -- --nocapture` (the `--nocapture` helps debug). +- **Regenerate ts-rs bindings with:** `cd src-tauri && cargo test export_bindings --lib`. +- **Verify the frontend typechecks with:** `npx tsc --noEmit`. +- **TDD order for every behavior:** (1) write failing test, (2) run to confirm it fails for the right reason, (3) implement, (4) run to confirm pass, (5) commit. + +--- + +## Task 1: Scaffold `queue.rs` with `QueueManager` struct and test harness + +**Goal:** Stand up the module with the core types and a test-friendly constructor. No dispatcher logic yet — just the data structures and an injection seam for the sidecar spawner so tests can use a fake. + +**Files:** +- Create: `src-tauri/src/queue.rs` +- Modify: `src-tauri/src/lib.rs:421` (add `pub mod queue;`) + +### Step 1.1: Create `queue.rs` with types and a test-injectable `SpawnCallback` + +- [ ] **Create `src-tauri/src/queue.rs`:** + +```rust +use crate::ipc::DownloadStatus; +use serde::{Deserialize, Serialize}; +use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use tauri::AppHandle; +use tokio::sync::{Mutex, Notify, OwnedSemaphorePermit, Semaphore}; +use ts_rs::TS; + +/// Default capacity when no setting is read yet. +pub const DEFAULT_MAX_CONCURRENT: usize = 3; + +/// Outcome of an aria2 completion that arrived before its gid was stored. +/// Carries the outcome so the correct state emit survives the race. +#[derive(Debug, Clone)] +pub(crate) enum PendingOutcome { + Complete, + Error(String), +} + +/// What kind of sidecar a queued task spawns. Drives which runner the +/// dispatcher invokes. +#[derive(Debug, Clone)] +pub(crate) enum TaskKind { + Aria2, + Media, + Native, +} + +/// Everything needed to start a sidecar, captured at enqueue time so the +/// dispatcher can spawn it later without round-tripping back to the frontend. +#[derive(Debug, Clone)] +pub(crate) struct QueuedTask { + pub id: String, + pub kind: TaskKind, + pub payload: SpawnPayload, +} + +/// Args mirroring start_download / start_media_download. Kept untyped-loose +/// (String/Option) to match the existing command signatures exactly. +#[derive(Debug, Clone, Default)] +pub(crate) struct SpawnPayload { + pub url: String, + pub destination: String, + pub filename: String, + pub connections: Option, + pub speed_limit: Option, + pub username: Option, + pub password: Option, + pub headers: Option, + pub checksum: Option, + pub cookies: Option, + pub mirrors: Option, + pub user_agent: Option, + pub max_tries: Option, + pub proxy: Option, + pub format_selector: Option, + pub cookie_source: Option, + pub is_media: bool, +} + +/// Move direction for manual reordering. +#[derive(Clone, Copy, Debug, Deserialize, Serialize, TS)] +#[serde(rename_all = "lowercase")] +#[ts(export, export_to = "../../src/bindings/")] +pub enum QueueDirection { + Up, + Down, +} + +/// Authoritative backend state for a download's status. Emitted to the +/// frontend as the single source of truth for transitions. +#[derive(Clone, Debug, Serialize, TS)] +#[ts(export, export_to = "../../src/bindings/")] +pub struct DownloadStateEvent { + pub id: String, + pub status: String, + pub error: Option, +} + +impl DownloadStateEvent { + pub(crate) fn new(id: impl Into, status: DownloadStatus) -> Self { + Self { + id: id.into(), + status: status.as_str().to_string(), + error: None, + } + } + + pub(crate) fn failed(id: impl Into, error: impl Into) -> Self { + Self { + id: id.into(), + status: DownloadStatus::Failed.as_str().to_string(), + error: Some(error.into()), + } + } +} + +/// A sidecar spawner. In production this calls the real aria2/yt-dlp/native +/// runners; in tests it is replaced with a fake that records calls and +/// optionally hangs to simulate a long-running download. +#[crate::async_trait] +pub(crate) trait SidecarSpawner: Send + Sync + 'static { + /// Spawn an aria2 download. Returns the gid. Must return quickly (the + /// permit is already parked before this is called). + async fn add_uri(&self, id: &str, payload: &SpawnPayload) -> Result; + + /// Run a media download to completion. The permit is parked for the full + /// duration; the implementation must call `release_permit`-equivalent on + /// exit (handled by QueueManager, not here). + async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String>; + + /// Run a native HTTP download to completion. + async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String>; +} + +/// The centralized concurrency gatekeeper. One instance lives in AppState. +pub struct QueueManager { + pending: Mutex>, + semaphore: Arc, + active_permits: Mutex>, + target_capacity: AtomicUsize, + slots_to_retire: AtomicUsize, + notify: Notify, + + /// aria2 gid -> download id map (shared with the WS poller). + /// (Keyed by gid because that's what the poller sees first.) + pub(crate) aria2_gids: Arc>>, + + /// gid -> buffered outcome for completions that arrived before gid store. + pub(crate) pending_completion: Arc>>, + + spawner: Arc, + app_handle: AppHandle, +} + +impl QueueManager { + /// Production constructor. Wired up in lib.rs setup(). + pub fn new(app_handle: AppHandle, capacity: usize) -> Self { + Self::with_spawner(app_handle, capacity, Arc::new(ProductionSpawner)) + } + + /// Test constructor injecting a fake spawner. + #[cfg(test)] + pub(crate) fn with_spawner( + app_handle: AppHandle, + capacity: usize, + spawner: Arc, + ) -> Self { + Self { + pending: Mutex::new(VecDeque::new()), + semaphore: Arc::new(Semaphore::new(capacity)), + active_permits: Mutex::new(HashMap::new()), + target_capacity: AtomicUsize::new(capacity), + slots_to_retire: AtomicUsize::new(0), + notify: Notify::new(), + aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), + pending_completion: Arc::new(Mutex::new(HashMap::new())), + spawner, + app_handle, + } + } + + /// Current pending order, as id list. Returned by move_in_queue. + pub(crate) async fn pending_order(&self) -> Vec { + self.pending + .lock() + .await + .iter() + .map(|t| t.id.clone()) + .collect() + } +} + +/// Placeholder production spawner. Real wiring added in Task 7. +struct ProductionSpawner; + +#[crate::async_trait] +impl SidecarSpawner for ProductionSpawner { + async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result { + Err("production spawner not wired yet".to_string()) + } + async fn run_media(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + Err("production spawner not wired yet".to_string()) + } + async fn run_native(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + Err("production spawner not wired yet".to_string()) + } +} +``` + +### Step 1.2: Add `async-trait` dependency + +- [ ] **Add to `src-tauri/Cargo.toml` under `[dependencies]`:** + +```toml +async-trait = "0.1" +``` + +- [ ] **Run:** `cd src-tauri && cargo build --lib` + Expected: builds (the `#[crate::async_trait]` macro resolves once the dep is present). + +> Note: we use `#[crate::async_trait]` and re-export `async_trait::async_trait` as `crate::async_trait` in Step 1.3 so test code outside the crate can't accidentally depend on the macro path. Simpler: just use `#[async_trait::async_trait]` directly throughout. We'll do the direct form — replace every `#[crate::async_trait]` above with `#[async_trait::async_trait]`. + +- [ ] **In `queue.rs`, replace all three occurrences of `#[crate::async_trait]` with `#[async_trait::async_trait]`.** + +### Step 1.3: Register the module in `lib.rs` + +- [ ] **In `src-tauri/src/lib.rs`, find the module declarations around line 421-426:** + +```rust +pub mod download; +#[allow(dead_code)] +mod ipc; +mod parity; +pub mod error; +pub mod commands; +``` + +- [ ] **Add `pub mod queue;` after `pub mod download;`:** + +```rust +pub mod download; +pub mod queue; +#[allow(dead_code)] +mod ipc; +mod parity; +pub mod error; +pub mod commands; +``` + +Also remove `#[allow(dead_code)]` from `mod ipc;` — no, keep it; ipc is used via `crate::ipc::` references and ts-rs. Leave the existing `#[allow(dead_code)]` intact. + +### Step 1.4: Verify it compiles + +- [ ] **Run:** `cd src-tauri && cargo build --lib` + Expected: clean build. Warnings about unused fields/methods are fine at this stage. + +### Step 1.5: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/queue.rs src-tauri/src/lib.rs src-tauri/Cargo.toml +git commit -m "feat(queue): scaffold QueueManager module with types and spawner trait" +``` + +--- + +## Task 2: Implement enqueue, permit parking, and release (TDD) + +**Goal:** The membership primitives. `push` adds a task; `acquire_and_park` gets a permit and parks it; `release_permit` removes and frees it idempotently. No dispatcher loop yet — tests drive these directly. + +**Files:** +- Modify: `src-tauri/src/queue.rs` +- Create: `src-tauri/tests/queue_manager.rs` + +### Step 2.1: Write the failing tests for push, park, release, idempotent release + +- [ ] **Create `src-tauri/tests/queue_manager.rs`:** + +```rust +use firelink_lib::queue::{QueueManager, SpawnPayload, TaskKind, QueuedTask}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; +use tauri::test::{mock_builder, mock_context, noop_assets}; +use tokio::time::timeout; + +/// A fake spawner that records calls and lets tests gate sidecar lifetime. +struct CountingSpawner { + add_uri_calls: AtomicUsize, + media_calls: AtomicUsize, + native_calls: AtomicUsize, +} + +impl CountingSpawner { + fn new() -> Self { + Self { + add_uri_calls: AtomicUsize::new(0), + media_calls: AtomicUsize::new(0), + native_calls: AtomicUsize::new(0), + } + } +} + +#[async_trait::async_trait] +impl firelink_lib::queue::SidecarSpawner for CountingSpawner { + async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result { + self.add_uri_calls.fetch_add(1, Ordering::SeqCst); + Ok(format!("gid-{}", self.add_uri_calls.load(Ordering::SeqCst))) + } + async fn run_media(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + self.media_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + async fn run_native(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + self.native_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +/// Build a QueueManager with a fake spawner. Tauri's mock AppHandle is needed +/// for emit; we construct the minimal mock. +fn make_manager(capacity: usize) -> (QueueManager, Arc) { + let app = mock_builder() + .build(mock_context(noop_assets())) + .expect("mock app"); + let spawner = Arc::new(CountingSpawner::new()); + // with_spawner is pub(crate) — expose via a test-only helper in queue.rs. + let mgr = QueueManager::test_new(app.handle(), capacity, spawner.clone()); + (mgr, spawner) +} + +fn sample_task(id: &str) -> QueuedTask { + QueuedTask { + id: id.to_string(), + kind: TaskKind::Native, + payload: SpawnPayload::default(), + } +} + +#[tokio::test] +async fn push_appends_to_pending_and_emits_queued() { + let (mgr, _spawner) = make_manager(2); + mgr.push(sample_task("a")).await; + mgr.push(sample_task("b")).await; + let order = mgr.pending_order().await; + assert_eq!(order, vec!["a".to_string(), "b".to_string()]); +} + +#[tokio::test] +async fn release_permit_is_idempotent() { + let (mgr, _spawner) = make_manager(2); + // Park a permit manually (simulates dispatcher having dispatched one). + mgr.acquire_and_park("a").await; + let avail_before = mgr.available_permits(); + mgr.release_permit("a").await; // first release: frees the slot + let avail_after_first = mgr.available_permits(); + mgr.release_permit("a").await; // second release: no-op + let avail_after_second = mgr.available_permits(); + assert_eq!(avail_after_first - avail_before, 1); + assert_eq!(avail_after_second, avail_after_first, "second release must not free another slot"); +} + +#[tokio::test] +async fn push_then_pop_front_drains_fifo() { + let (mgr, _spawner) = make_manager(2); + mgr.push(sample_task("a")).await; + mgr.push(sample_task("b")).await; + let first = mgr.pop_front().await.expect("some task"); + let second = mgr.pop_front().await.expect("some task"); + assert_eq!(first.id, "a"); + assert_eq!(second.id, "b"); + assert!(mgr.pop_front().await.is_none()); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test --test queue_manager -- --nocapture` + Expected: FAIL — `push`, `acquire_and_park`, `pop_front`, `available_permits`, `test_new` don't exist. + +### Step 2.2: Implement `push`, `pop_front`, `acquire_and_park`, `release_permit`, `available_permits` + +- [ ] **Add to `src-tauri/src/queue.rs` inside `impl QueueManager` (after `pending_order`):** + +```rust + /// Enqueue a task. Notifies the dispatcher. Emits download-state{queued}. + pub(crate) async fn push(&self, task: QueuedTask) { + let id = task.id.clone(); + self.pending.lock().await.push_back(task); + self.emit_state(id, DownloadStatus::Queued); + self.notify.notify_one(); + } + + /// Pop the next task, or None if empty. + pub(crate) async fn pop_front(&self) -> Option { + self.pending.lock().await.pop_front() + } + + /// Acquire a permit and park it under `id`. Used by the dispatcher after + /// it has popped a task. Returns the permit on error (acquire failures + /// are fatal — semaphore closed). + pub(crate) async fn acquire_and_park(&self, id: &str) { + let permit = self + .semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed"); + self.active_permits.lock().await.insert(id.to_string(), permit); + } + + /// Release the permit parked under `id`, if any. Idempotent. Wakes the + /// dispatcher so a freed slot is claimed promptly. + pub(crate) async fn release_permit(&self, id: &str) { + let removed = self.active_permits.lock().await.remove(id).is_some(); + if removed { + self.notify.notify_one(); + } + } + + /// Number of un-acquired permits currently in the semaphore pool. + pub(crate) fn available_permits(&self) -> usize { + self.semaphore.available_permits() + } + + fn emit_state(&self, id: impl Into, status: DownloadStatus) { + use tauri::Emitter; + let _ = self.app_handle.emit( + "download-state", + DownloadStateEvent::new(id, status), + ); + } +``` + +We also need `test_new` (test-only constructor) and a way to expose `SidecarSpawner` to the integration test (it's `pub(crate)`). Add a `#[cfg(test)]`-gated public re-export is not possible in an integration test (different crate). Solution: make `SidecarSpawner`, `SpawnPayload`, `QueuedTask`, `TaskKind`, `QueueManager`, and `test_new` `pub` (not `pub(crate)`) so the integration test can use them. The trait is only meaningful inside this crate's dispatch logic, but exposing it publicly is harmless and keeps tests honest. + +- [ ] **In `queue.rs`, change visibility** of these items from `pub(crate)` to `pub`: `TaskKind`, `QueuedTask`, `SpawnPayload`, `SidecarSpawner`, `QueueManager` (already `pub`), and the methods `push`, `pop_front`, `acquire_and_park`, `release_permit`, `available_permits`, `pending_order`. Keep `PendingOutcome` as `pub(crate)` (test doesn't need it yet). + +- [ ] **Add the test-only constructor to `impl QueueManager`:** + +```rust + #[cfg(test)] + pub fn test_new(app_handle: AppHandle, capacity: usize, spawner: Arc) -> Self { + Self { + pending: Mutex::new(VecDeque::new()), + semaphore: Arc::new(Semaphore::new(capacity)), + active_permits: Mutex::new(HashMap::new()), + target_capacity: AtomicUsize::new(capacity), + slots_to_retire: AtomicUsize::new(0), + notify: Notify::new(), + aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), + pending_completion: Arc::new(Mutex::new(HashMap::new())), + spawner, + app_handle, + } + } +``` + +Now delete the `with_spawner` method (superseded by `test_new`) and remove the `#[cfg(test)]` on it. + +- [ ] **Run:** `cd src-tauri && cargo test --test queue_manager -- --nocapture` + Expected: PASS — all three tests green. + +### Step 2.3: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/queue.rs src-tauri/tests/queue_manager.rs +git commit -m "feat(queue): add push/pop/acquire_and_park/release_permit primitives" +``` + +--- + +## Task 3: Implement the dispatcher loop with idle-park and CAS retirement (TDD) + +**Goal:** The long-running dispatcher that idle-parks when pending is empty, acquires a permit, honors the retirement debt via CAS (never underflowing), and re-pops under lock to guard against races. This is the heart of the spec — three of the four discovered bugs live here. + +**Files:** +- Modify: `src-tauri/src/queue.rs` +- Modify: `src-tauri/tests/queue_manager.rs` + +### Step 3.1: Write the failing test for idle-park (no busy-spin) + +This test wraps the semaphore in a counting layer is hard (Semaphore is concrete). Instead we assert behaviorally: with an empty queue and capacity > 0, the dispatcher must NOT consume a permit. We park the dispatcher, sleep 200ms, then assert `available_permits()` is still equal to capacity (no permit was acquired). + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +#[tokio::test] +async fn dispatcher_parks_when_idle_no_busy_spin() { + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Queue is empty. Sleep long enough for any busy-spin to drain permits. + tokio::time::sleep(Duration::from_millis(200)).await; + + // No permit should have been acquired while idle. + assert_eq!( + mgr_arc.available_permits(), 3, + "dispatcher must not acquire permits when pending is empty" + ); + + handle.abort(); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test dispatcher_parks_when_idle -- --nocapture` + Expected: FAIL — `run_dispatcher` doesn't exist. + +### Step 3.2: Write the failing test for CAS retirement never underflowing + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +#[tokio::test] +async fn cas_retirement_never_underflows() { + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Repeatedly toggle capacity down while idle. A fetch_sub-based impl + // would underflow to usize::MAX on the second set_capacity(1). + for _ in 0..5 { + mgr_arc.set_capacity(3); + mgr_arc.set_capacity(1); + mgr_arc.set_capacity(3); + } + tokio::time::sleep(Duration::from_millis(50)).await; + + let debt = mgr_arc.slots_to_retire_load(); + assert!( + debt <= 2, + "retirement debt must stay within [0, capacity-target], got {debt}" + ); + + handle.abort(); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test cas_retirement_never_underflows -- --nocapture` + Expected: FAIL — `run_dispatcher`, `set_capacity`, `slots_to_retire_load` don't exist. + +### Step 3.3: Implement `run_dispatcher`, `set_capacity`, `slots_to_retire_load` + +- [ ] **Add to `impl QueueManager` in `queue.rs`:** + +```rust + /// Resize the global concurrency limit. Grow adds permits immediately; + /// shrink records a retirement debt honored lazily by the dispatcher. + pub(crate) fn set_capacity(&self, new_target: usize) { + let prev_target = self.target_capacity.swap(new_target, Ordering::Relaxed); + if new_target == prev_target { + return; + } + if new_target > prev_target { + let delta = new_target - prev_target; + self.semaphore.add_permits(delta); + self.notify.notify_one(); + } else { + let delta = prev_target - new_target; + self.slots_to_retire.fetch_add(delta, Ordering::Relaxed); + } + } + + /// Test accessor for the retirement debt counter. + #[cfg(test)] + pub fn slots_to_retire_load(&self) -> usize { + self.slots_to_retire.load(Ordering::Relaxed) + } + + /// The long-running dispatcher. One instance is spawned in setup(). + /// Idle-parks on Notify; CAS-honors retirement debt; re-pops under lock. + pub async fn run_dispatcher(self: Arc) { + loop { + // (1) Idle-park: avoid busy-spin when pending is empty. + if self.pending.lock().await.is_empty() { + self.notify.notified().await; + continue; + } + // (2) Acquire a slot. + let permit = self + .semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed"); + // (3) CAS retirement — never underflows to usize::MAX. + let mut retired = false; + loop { + let debt = self.slots_to_retire.load(Ordering::Relaxed); + if debt == 0 { + break; + } + match self.slots_to_retire.compare_exchange_weak( + debt, + debt - 1, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => { + drop(permit); + retired = true; + break; + } + Err(_actual) => { + // retry with the loaded value on the next iteration + } + } + } + if retired { + continue; + } + // (4) Re-pop under lock — guards against racing removals between + // waking from Notify and acquiring the permit. + let task = match self.pending.lock().await.pop_front() { + Some(t) => t, + None => { + drop(permit); + continue; + } + }; + self.dispatch_one(permit, task).await; + } + } +``` + +`dispatch_one` is referenced but not yet implemented. Add a minimal stub so this compiles; full impl is Task 4. + +- [ ] **Add a stub `dispatch_one` to `impl QueueManager`:** + +```rust + async fn dispatch_one(self: Arc, _permit: OwnedSemaphorePermit, task: QueuedTask) { + // Full implementation in Task 4. For now, park the permit so the + // idle/retirement tests run without leaking slots. + let id = task.id.clone(); + self.acquire_and_park(&id).await; + self.emit_state(id, DownloadStatus::Downloading); + // NOTE: this stub leaks the permit (never released). Acceptable for + // the two tests in this task, which don't dispatch real tasks. + let _ = task; + } +``` + +Wait — the stub calls `acquire_and_park` which acquires a *second* permit (the dispatcher already holds one passed as `_permit`). That double-acquires. Fix: `dispatch_one` receives the already-acquired permit; it should park THAT permit, not acquire another. Refactor `acquire_and_park` to split into "acquire" and "park existing permit": + +- [ ] **Replace `acquire_and_park` with two methods in `queue.rs`:** + +```rust + /// Acquire a permit from the semaphore (blocks until one is available). + pub(crate) async fn acquire_permit(&self) -> OwnedSemaphorePermit { + self.semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed") + } + + /// Park an already-acquired permit under `id`. + pub(crate) async fn park_permit(&self, id: &str, permit: OwnedSemaphorePermit) { + self.active_permits + .lock() + .await + .insert(id.to_string(), permit); + } +``` + +- [ ] **Update the `release_permit_is_idempotent` test (Task 2.1) to use the split API.** In `tests/queue_manager.rs`, replace `mgr.acquire_and_park("a").await;` with: + +```rust + let permit = mgr.acquire_permit().await; + mgr.park_permit("a", permit).await; +``` + +- [ ] **Update `dispatch_one` stub to park the passed permit:** + +```rust + async fn dispatch_one(self: Arc, permit: OwnedSemaphorePermit, task: QueuedTask) { + let id = task.id.clone(); + self.park_permit(&id, permit).await; + self.emit_state(id, DownloadStatus::Downloading); + let _ = task; + } +``` + +- [ ] **Run:** `cd src-tauri && cargo test --test queue_manager -- --nocapture` + Expected: PASS — all five tests (3 from Task 2 + 2 new) green. + +### Step 3.4: Write the failing test for grow releasing immediately + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +#[tokio::test] +async fn grow_releases_immediately_and_dispatches_waiting_tasks() { + let (mgr, spawner) = make_manager(2); + let mgr_arc = Arc::new(mgr); + + // Block the spawner so dispatched tasks don't complete (permits stay parked). + // CountingSpawner completes instantly, so to keep permits parked we instead + // push 4 tasks with capacity 2: two dispatch, two wait. Then grow to 4. + for i in 0..4 { + mgr_arc.push(sample_task(&format!("t{i}"))).await; + } + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Give dispatcher time to dispatch 2 (capacity) of the 4. + tokio::time::sleep(Duration::from_millis(100)).await; + let native_after_initial = spawner.native_calls.load(Ordering::SeqCst); + assert_eq!(native_after_initial, 2, "only capacity-many tasks dispatch initially"); + + // Grow to 4; the remaining 2 should dispatch. + mgr_arc.set_capacity(4); + tokio::time::sleep(Duration::from_millis(100)).await; + let native_after_grow = spawner.native_calls.load(Ordering::SeqCst); + assert_eq!(native_after_grow, 4, "grow must allow the waiting tasks to dispatch"); + + handle.abort(); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test grow_releases_immediately -- --nocapture` + Expected: This may PASS already (the dispatcher + grow logic is correct). If it passes, good — it's a regression guard. If it fails, debug the wakeup path (`set_capacity` grow branch calls `notify_one`). + +### Step 3.5: Write the failing test for shrink converging to target + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +#[tokio::test] +async fn shrink_converges_to_target_without_killing_active() { + let (mgr, spawner) = make_manager(4); + let mgr_arc = Arc::new(mgr); + + for i in 0..6 { + mgr_arc.push(sample_task(&format!("t{i}"))).await; + } + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Let 4 dispatch (capacity). + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(spawner.native_calls.load(Ordering::SeqCst), 4); + + // Shrink to 2 while 4 are "active" (permits parked; CountingSpawner + // already returned so we must manually release to simulate completion). + mgr_arc.set_capacity(2); + + // Release the 4 active permits one by one; after 2 releases, the debt + // (2) is exhausted, so no new dispatches should occur until... there are + // no more pending either (6 - 4 = 2 pending, 2 retired). Active settles at 0 + // new dispatches beyond the initial 4 because debt absorbs both slots. + mgr_arc.release_permit("t0").await; + mgr_arc.release_permit("t1").await; + tokio::time::sleep(Duration::from_millis(100)).await; + + // Debt was 2; two releases retired both. The 2 remaining pending tasks + // (t4, t5) can now dispatch since debt is 0 and slots freed. + // Total native calls: initial 4 + t4 + t5 = 6. + assert_eq!( + spawner.native_calls.load(Ordering::SeqCst), + 6, + "shrink converges: after debt exhausted, remaining pending dispatch" + ); + + handle.abort(); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test shrink_converges -- --nocapture` + Expected: PASS (validates the full resize lifecycle). If it fails, the CAS loop or wakeup is off — check that `release_permit`'s `notify_one` wakes the dispatcher after retirement. + +### Step 3.6: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/queue.rs src-tauri/tests/queue_manager.rs +git commit -m "feat(queue): dispatcher loop with idle-park and CAS-based semaphore resize" +``` + +--- + +## Task 4: Implement `dispatch_one` with uniform permit parking (TDD) + +**Goal:** The full `dispatch_one` that spawns sidecars via the `SidecarSpawner` trait, parks permits uniformly (aria2 trap fix), and releases on completion. Plus the aria2 gid-completion race handling. + +**Files:** +- Modify: `src-tauri/src/queue.rs` +- Modify: `src-tauri/tests/queue_manager.rs` + +### Step 4.1: Write the failing test for aria2 permit surviving RPC return + +The fake spawner's `add_uri` returns instantly with a gid. The permit must remain parked (active) until `release_permit` is called — NOT released when the RPC returns. + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +fn aria2_task(id: &str) -> QueuedTask { + QueuedTask { + id: id.to_string(), + kind: TaskKind::Aria2, + payload: SpawnPayload::default(), + } +} + +#[tokio::test] +async fn aria2_permit_survives_rpc_return() { + let (mgr, spawner) = make_manager(1); + let mgr_arc = Arc::new(mgr); + mgr_arc.push(aria2_task("a")).await; + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Dispatcher acquires the single permit, parks it, calls add_uri (returns + // instantly). The permit must STAY parked. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(spawner.add_uri_calls.load(Ordering::SeqCst), 1); + assert_eq!( + mgr_arc.available_permits(), 0, + "permit must remain parked while aria2 download is notionally running" + ); + + // Now simulate aria2 completion: release_permit frees the slot. + mgr_arc.release_permit("a").await; + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(mgr_arc.available_permits(), 1, "release frees the parked permit"); + + handle.abort(); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test aria2_permit_survives_rpc_return -- --nocapture` + Expected: depends — the Task 3 stub parks the permit but never calls the spawner. This test will fail because `add_uri_calls == 0`. Good — that's the right failure mode. + +### Step 4.2: Implement the real `dispatch_one` + +- [ ] **Replace the `dispatch_one` stub in `queue.rs` with the full implementation:** + +```rust + async fn dispatch_one(self: Arc, permit: OwnedSemaphorePermit, task: QueuedTask) { + let id = task.id.clone(); + // Park the permit BEFORE spawning, keyed by id. Uniform parking: + // aria2's RPC returns instantly, so the permit must outlive the + // dispatch_one call. Media/Native runners release on exit; pause + // releases immediately via release_permit. + self.park_permit(&id, permit).await; + self.emit_state(&id, DownloadStatus::Downloading); + + match task.kind { + TaskKind::Aria2 => { + // Issue the RPC; it returns immediately with a gid. The permit + // stays parked in active_permits. Release happens via the WS + // poller / pause / remove commands calling release_permit. + match self.spawner.add_uri(&id, &task.payload).await { + Ok(gid) => self.remember_gid(id.clone(), gid).await, + Err(error) => { + self.emit_failed(&id, error); + self.release_permit(&id).await; + } + } + } + TaskKind::Media => { + let this = Arc::clone(&self); + let payload = task.payload.clone(); + let id_for_task = id.clone(); + tauri::async_runtime::spawn(async move { + let outcome = this.spawner.run_media(&id_for_task, &payload).await; + this.finish_runner(&id_for_task, outcome).await; + }); + } + TaskKind::Native => { + let this = Arc::clone(&self); + let payload = task.payload.clone(); + let id_for_task = id.clone(); + tauri::async_runtime::spawn(async move { + let outcome = this.spawner.run_native(&id_for_task, &payload).await; + this.finish_runner(&id_for_task, outcome).await; + }); + } + } + } + + /// Called when a Media/Native runner exits. Releases the permit and emits + /// the terminal state. Idempotent: if pause already released, this is a + /// no-op for the permit, but we still emit the terminal state only if the + /// item isn't already paused (handled by the frontend's status check). + async fn finish_runner(self: Arc, id: &str, outcome: Result<(), String>) { + match outcome { + Ok(()) => { + self.emit_state(id, DownloadStatus::Completed); + } + Err(error) => { + self.emit_failed(id, error); + } + } + self.release_permit(id).await; + } + + fn emit_failed(&self, id: &str, error: String) { + use tauri::Emitter; + let _ = self + .app_handle + .emit("download-state", DownloadStateEvent::failed(id, error)); + } + + /// Store gid -> id, then reconcile any buffered completion for that gid. + /// Closes the gid-lookup race (spec §5.5). + pub(crate) async fn remember_gid(&self, id: String, gid: String) { + { + let mut gids = self.aria2_gids.write().unwrap(); + gids.insert(gid.clone(), id.clone()); + } + // Check for a buffered completion that arrived before we stored the gid. + let buffered = self.pending_completion.lock().await.remove(&gid); + if let Some((buf_id, outcome)) = buffered { + self.apply_completion(&buf_id, outcome).await; + } + } + + /// Apply an aria2 completion outcome: release permit + emit state. + pub(crate) async fn apply_completion(&self, id: &str, outcome: PendingOutcome) { + match outcome { + PendingOutcome::Complete => { + self.emit_state(id, DownloadStatus::Completed); + } + PendingOutcome::Error(error) => { + self.emit_failed(id, error); + } + } + self.release_permit(id).await; + } +``` + +`PendingOutcome` is `pub(crate)` — `apply_completion` takes it, so that's fine within the crate. But the WS poller (in lib.rs) also needs to construct `PendingOutcome`. Make it `pub`: + +- [ ] **Change `PendingOutcome` from `pub(crate)` to `pub`** in `queue.rs`. + +### Step 4.3: Handle aria2 completion arriving before gid store (TDD) + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +use firelink_lib::queue::PendingOutcome; + +#[tokio::test] +async fn gid_completion_before_store_buffers_and_reconciles() { + let (mgr, _spawner) = make_manager(1); + let mgr_arc = Arc::new(mgr); + mgr_arc.push(aria2_task("a")).await; + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + tokio::time::sleep(Duration::from_millis(100)).await; + + // The dispatcher called add_uri and got "gid-1", then remember_gid stored it. + // Simulate a completion arriving for an UNKNOWN gid first: + mgr_arc + .handle_aria2_event("gid-unknown", PendingOutcome::Complete) + .await; + tokio::time::sleep(Duration::from_millis(50)).await; + // Permit still parked (gid-unknown is not ours). + assert_eq!(mgr_arc.available_permits(), 0); + + // Now store gid-1 -> "a" via remember_gid. The buffered gid-unknown stays + // buffered (different gid). Release via the real gid: + mgr_arc.release_permit("a").await; + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(mgr_arc.available_permits(), 1); + + // And confirm a buffered event for a KNOWN-future gid reconciles: + // Push another aria2 task; its gid will be "gid-2". + mgr_arc.push(aria2_task("b")).await; + tokio::time::sleep(Duration::from_millis(100)).await; + // Buffer a completion for gid-2 BEFORE remember_gid runs for it. In real + // flow remember_gid runs immediately after add_uri; here we simulate the + // race by calling handle_aria2_event then remember_gid manually. + // (The dispatcher already called remember_gid, so gid-2 is stored. To + // truly test the race we'd need to inject between add_uri and remember_gid. + // The buffering path is exercised by the gid-unknown case above; this + // assertion just confirms no crash and the permit for "b" can be released.) + mgr_arc.release_permit("b").await; + tokio::time::sleep(Duration::from_millis(50)).await; + + handle.abort(); +} +``` + +This test is more of a smoke test for the buffering path. The critical assertion is the first one: an unknown gid doesn't free a permit. Implement `handle_aria2_event`: + +- [ ] **Add to `impl QueueManager` in `queue.rs`:** + +```rust + /// Entry point for the aria2 WS poller. Resolves gid -> id; if not yet + /// stored, buffers the outcome for reconciliation by remember_gid. + pub async fn handle_aria2_event(&self, gid: &str, outcome: PendingOutcome) { + let id_opt = { + let gids = self.aria2_gids.read().unwrap(); + gids.get(gid).cloned() + }; + match id_opt { + Some(id) => { + self.apply_completion(&id, outcome).await; + } + None => { + // Buffer keyed by gid; id is filled by remember_gid via the + // (gid, (id_placeholder, outcome)) — but we don't know the id + // yet. Store with empty id; remember_gid will re-derive id + // from the gid->id map it just inserted. + self.pending_completion + .lock() + .await + .insert(gid.to_string(), (String::new(), outcome)); + } + } + } +``` + +Wait — the buffered entry stores `(String::new(), outcome)` but `remember_gid` then does `pending_completion.remove(&gid)` and gets `(buf_id, outcome)` with empty `buf_id`. It should use the `id` it just inserted, not the empty buffer. Fix `remember_gid`: + +- [ ] **Update `remember_gid` to use the freshly-stored id:** + +```rust + pub(crate) async fn remember_gid(&self, id: String, gid: String) { + { + let mut gids = self.aria2_gids.write().unwrap(); + gids.insert(gid.clone(), id.clone()); + } + let buffered = self.pending_completion.lock().await.remove(&gid); + if let Some((_buf_id, outcome)) = buffered { + // Use `id` (the one we just stored), not the placeholder buf_id. + self.apply_completion(&id, outcome).await; + } + } +``` + +- [ ] **Run:** `cd src-tauri && cargo test --test queue_manager -- --nocapture` + Expected: PASS — all tests including `aria2_permit_survives_rpc_return` and `gid_completion_before_store`. + +### Step 4.4: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/queue.rs src-tauri/tests/queue_manager.rs +git commit -m "feat(queue): dispatch_one with uniform permit parking and gid-race handling" +``` + +--- + +## Task 5: Implement Move Up / Move Down and enqueue_many (TDD) + +**Goal:** Reordering and bulk enqueue. `move_in_queue` swaps within the pending VecDeque and returns the new order. `enqueue_many` is for startup/start-all. + +**Files:** +- Modify: `src-tauri/src/queue.rs` +- Modify: `src-tauri/tests/queue_manager.rs` + +### Step 5.1: Write the failing test for reordering + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +#[tokio::test] +async fn move_up_down_reorders_pending() { + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + // Don't run the dispatcher — we're testing pure reorder on pending. + mgr_arc.push(sample_task("a")).await; + mgr_arc.push(sample_task("b")).await; + mgr_arc.push(sample_task("c")).await; + + // Move "c" up: [a,b,c] -> [a,c,b] + mgr_arc.move_in_queue("c", QueueDirection::Down).await; // sanity: down on last is no-op + assert_eq!(mgr_arc.pending_order().await, vec!["a", "b", "c"]); + + mgr_arc.move_in_queue("c", QueueDirection::Up).await; + assert_eq!(mgr_arc.pending_order().await, vec!["a", "c", "b"]); + + // Move "a" down: [a,c,b] -> [c,a,b] + mgr_arc.move_in_queue("a", QueueDirection::Down).await; + assert_eq!(mgr_arc.pending_order().await, vec!["c", "a", "b"]); + + // Boundary: move "c" (now first) up — no-op. + mgr_arc.move_in_queue("c", QueueDirection::Up).await; + assert_eq!(mgr_arc.pending_order().await, vec!["c", "a", "b"]); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test move_up_down_reorders -- --nocapture` + Expected: FAIL — `move_in_queue` doesn't exist. Add `use firelink_lib::queue::QueueDirection;` to the test imports. + +### Step 5.2: Implement `move_in_queue` and `enqueue_many` + +- [ ] **Add to `impl QueueManager` in `queue.rs`:** + +```rust + /// Reorder a pending task up or down. Returns the new pending order. + /// No-op at boundaries. Does not emit (membership unchanged); the caller + /// (Tauri command) returns the order to the frontend. + pub(crate) async fn move_in_queue( + &self, + id: &str, + direction: QueueDirection, + ) -> Vec { + let mut pending = self.pending.lock().await; + let pos = pending.iter().position(|t| t.id == id); + if let Some(pos) = pos { + let target = match direction { + QueueDirection::Up => pos.checked_sub(1), + QueueDirection::Down => { + if pos + 1 < pending.len() { + Some(pos + 1) + } else { + None + } + } + }; + if let Some(target) = target { + pending.swap(pos, target); + } + } + pending.iter().map(|t| t.id.clone()).collect() + } + + /// Remove a task from pending if present (used by remove_download). + /// Does NOT release a permit (the caller handles active permits via + /// release_permit if the task was already dispatched). + pub(crate) async fn remove_from_pending(&self, id: &str) -> bool { + let mut pending = self.pending.lock().await; + let before = pending.len(); + pending.retain(|t| t.id != id); + let removed = pending.len() < before; + if removed { + self.notify.notify_one(); + } + removed + } + + /// Bulk enqueue by appending tasks. Used by startup and start-all. + pub(crate) async fn enqueue_many(&self, tasks: Vec) { + let mut pending = self.pending.lock().await; + for task in tasks { + let id = task.id.clone(); + pending.push_back(task); + self.emit_state(id, DownloadStatus::Queued); + } + drop(pending); + self.notify.notify_one(); + } +``` + +- [ ] **Run:** `cd src-tauri && cargo test move_up_down_reorders -- --nocapture` + Expected: PASS. + +### Step 5.3: Write the failing test for Notify firing on enqueue/release + +- [ ] **Add to `src-tauri/tests/queue_manager.rs`:** + +```rust +#[tokio::test] +async fn notify_fires_on_push_and_release() { + let (mgr, _spawner) = make_manager(1); + let mgr_arc = Arc::new(mgr); + + // Park a permit manually so the dispatcher has something to release. + let permit = mgr_arc.acquire_permit().await; + mgr_arc.park_permit("a", permit).await; + + let handle = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.run_dispatcher().await }) + }; + + // Dispatcher should be parked (pending empty). Push a task; it must + // wake and dispatch within 100ms. + mgr_arc.push(sample_task("x")).await; + let dispatched = timeout( + Duration::from_millis(150), + async { + loop { + if mgr_arc.available_permits() == 0 { + return; // permit consumed = dispatched + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }, + ).await; + assert!(dispatched.is_ok(), "push must wake the idle dispatcher"); + + handle.abort(); +} +``` + +- [ ] **Run:** `cd src-tauri && cargo test notify_fires -- --nocapture` + Expected: PASS (validates the wakeup invariant from spec §5.2). + +### Step 5.4: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/queue.rs src-tauri/tests/queue_manager.rs +git commit -m "feat(queue): move_in_queue, enqueue_many, remove_from_pending" +``` + +--- + +## Task 6: Add `DownloadStateEvent`, `QueueDirection` to `ipc.rs` and regenerate bindings + +**Goal:** Move the ts-rs-exported types into `ipc.rs` (the canonical home, alongside `DownloadStatus`) and regenerate the frontend bindings. Also remove the now-obsolete `_dispatched` field from `DownloadItem`. + +**Files:** +- Modify: `src-tauri/src/ipc.rs` +- Modify: `src-tauri/src/queue.rs` (move types out, re-import from ipc) +- Generate: `src/bindings/DownloadStateEvent.ts`, `src/bindings/QueueDirection.ts` + +### Step 6.1: Move `DownloadStateEvent` and `QueueDirection` to `ipc.rs` + +- [ ] **In `src-tauri/src/ipc.rs`, add at the end of the file (after `PersistedSettings`):** + +```rust +#[derive(Clone, Copy, Debug, Serialize, Deserialize, TS)] +#[serde(rename_all = "lowercase")] +#[ts(export, export_to = "../../src/bindings/")] +pub enum QueueDirection { + Up, + Down, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[ts(export, export_to = "../../src/bindings/")] +pub struct DownloadStateEvent { + pub id: String, + pub status: String, + pub error: Option, +} +``` + +- [ ] **In `src-tauri/src/queue.rs`:** + - Remove the `QueueDirection` and `DownloadStateEvent` struct/enum definitions and their `impl DownloadStateEvent` block. + - Replace the `use crate::ipc::DownloadStatus;` import line with: + +```rust +use crate::ipc::{DownloadDirection_unused, DownloadStateEvent, DownloadStatus, QueueDirection}; +``` + +Wait — that has a typo. Use: + +```rust +use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection}; +``` + +- [ ] **Update the `DownloadStateEvent::new` / `failed` constructors.** They were on the struct in `queue.rs`; move them to `ipc.rs` as an `impl` block there, since the struct now lives there: + +In `ipc.rs`, after the `DownloadStateEvent` struct, add: + +```rust +impl DownloadStateEvent { + pub fn new(id: impl Into, status: DownloadStatus) -> Self { + Self { + id: id.into(), + status: status.as_str().to_string(), + error: None, + } + } + + pub fn failed(id: impl Into, error: impl Into) -> Self { + Self { + id: id.into(), + status: DownloadStatus::Failed.as_str().to_string(), + error: Some(error.into()), + } + } +} +``` + +- [ ] **In `queue.rs`, remove the old `impl DownloadStateEvent` block** (now redundant — it lives in ipc.rs). + +### Step 6.2: Remove `_dispatched` from `DownloadItem` + +- [ ] **In `src-tauri/src/ipc.rs`, in the `DownloadItem` struct, delete these three lines:** + +```rust + #[serde(rename = "_dispatched")] + #[ts(optional)] + pub dispatched: Option, +``` + +### Step 6.3: Regenerate bindings and verify + +- [ ] **Run:** `cd src-tauri && cargo test export_bindings --lib` + Expected: PASS; creates `src/bindings/DownloadStateEvent.ts`, `src/bindings/QueueDirection.ts`, and updates `src/bindings/DownloadItem.ts` (removes `_dispatched`). + +- [ ] **Verify the new bindings exist:** `ls src/bindings/ | grep -E "DownloadStateEvent|QueueDirection"` + Expected: both files listed. + +- [ ] **Run:** `npx tsc --noEmit` + Expected: TYPE ERRORS in `useDownloadStore.ts`, `downloadStore.ts`, `App.tsx` referencing `_dispatched` or `processQueue`. These will be fixed in Tasks 8-9. Do NOT fix yet — confirm the errors are only about `_dispatched`/`processQueue`, then proceed. + +### Step 6.4: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/ipc.rs src-tauri/src/queue.rs src/bindings/ +git commit -m "feat(ipc): add DownloadStateEvent and QueueDirection; drop _dispatched field" +``` + +--- + +## Task 7: Wire `QueueManager` into `AppState` and add Tauri commands + +**Goal:** Production wiring. Instantiate `QueueManager` in `AppState`, spawn the dispatcher in `setup()`, and add the `enqueue_download`/`enqueue_many`/`resume_download`/`move_in_queue`/`remove_from_queue` commands. Convert `start_download`/`start_media_download` to `pub(crate)`. Rewrite `set_concurrent_limit`. + +This is a large task. Break the spawner wiring into a real `ProductionSpawner` that calls the existing aria2/yt-dlp/native code. + +**Files:** +- Modify: `src-tauri/src/queue.rs` (real `ProductionSpawner`) +- Modify: `src-tauri/src/lib.rs` (AppState, setup, commands, WS poller, aria2 ceiling) + +### Step 7.1: Implement the real `ProductionSpawner` + +The production spawner needs access to `AppState` (aria2 port/secret, coordinator, app handle). Since `SidecarSpawner` is `Send + Sync + 'static` and `AppState` is a Tauri-managed state, we store an `AppHandle` in the spawner and look up state at call time. + +- [ ] **In `src-tauri/src/queue.rs`, replace the stub `ProductionSpawner` with:** + +```rust +/// Production spawner that delegates to the real aria2 RPC, yt-dlp, and +/// native coordinator runners. +pub(crate) struct ProductionSpawner { + app_handle: AppHandle, +} + +impl ProductionSpawner { + pub(crate) fn new(app_handle: AppHandle) -> Self { + Self { app_handle } + } +} + +#[async_trait::async_trait] +impl SidecarSpawner for ProductionSpawner { + async fn add_uri(&self, id: &str, payload: &SpawnPayload) -> Result { + let state = self.app_handle.state::(); + let mut options = serde_json::Map::new(); + let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle); + if !crate::is_safe_path(&resolved_dest, &self.app_handle) { + return Err("Path traversal blocked".to_string()); + } + options.insert( + "dir".to_string(), + serde_json::json!(resolved_dest.to_string_lossy().to_string()), + ); + let safe_filename = std::path::Path::new(&payload.filename.replace('\\', "/")) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("download") + .to_string(); + options.insert("out".to_string(), serde_json::json!(safe_filename)); + let conn = payload.connections.unwrap_or(1); + options.insert("split".to_string(), serde_json::json!(conn.to_string())); + options.insert( + "max-connection-per-server".to_string(), + serde_json::json!(conn.to_string()), + ); + let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + options.insert("max-tries".to_string(), serde_json::json!(mt.to_string())); + options.insert("continue".to_string(), serde_json::json!("true")); + if let Some(speed) = &payload.speed_limit { + options.insert("max-download-limit".to_string(), serde_json::json!(speed)); + } + if let Some(user) = &payload.username { + options.insert("http-user".to_string(), serde_json::json!(user)); + } + if let Some(pass) = &payload.password { + options.insert("http-passwd".to_string(), serde_json::json!(pass)); + } + if let Some(chk) = &payload.checksum { + options.insert("checksum".to_string(), serde_json::json!(chk)); + } + if let Some(ua) = &payload.user_agent { + options.insert("user-agent".to_string(), serde_json::json!(ua)); + } + let mut header_list = Vec::new(); + if let Some(cook) = &payload.cookies { + header_list.push(format!("Cookie: {}", cook)); + } + if let Some(hdrs) = &payload.headers { + for line in hdrs.lines() { + if !line.trim().is_empty() { + header_list.push(line.trim().to_string()); + } + } + } + if !header_list.is_empty() { + options.insert("header".to_string(), serde_json::json!(header_list)); + } + if let Some(prox) = &payload.proxy { + options.insert("all-proxy".to_string(), serde_json::json!(prox)); + } + let uris = crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()); + let params = serde_json::json!([uris, options]); + let result = crate::rpc_call(state.aria2_port, &state.aria2_secret, "aria2.addUri", params) + .await + .map_err(|e| e.to_string())?; + let gid = result.as_str().unwrap_or("").to_string(); + // Note: gid -> id is stored by QueueManager.remember_gid, not here. + let _ = id; + Ok(gid) + } + + async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { + let state = self.app_handle.state::(); + let mut cancel_rx = state + .download_coordinator + .register_media(id.to_string()) + .await + .map_err(|e| e)?; + crate::start_media_download_internal( + self.app_handle.clone(), + id, + payload.url.clone(), + payload.destination.clone(), + payload.filename.clone(), + payload.format_selector.clone(), + payload.cookie_source.clone(), + payload.speed_limit.clone(), + payload.username.clone(), + payload.password.clone(), + payload.headers.clone(), + payload.proxy.clone(), + payload.user_agent.clone(), + payload.max_tries, + &mut cancel_rx, + ) + .await + } + + async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { + let state = self.app_handle.state::(); + let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?; + let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle); + let safe_filename = std::path::Path::new(&payload.filename.replace('\\', "/")) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("download") + .to_string(); + // The native coordinator is command-based; we send Start and let its + // existing worker run. Completion is reported via the existing + // download-complete/download-failed events. To make this await-able, + // we wrap it: the coordinator's run_coordinator emits on completion, + // but for the queue we need a synchronous result. Simplest: send Start + // and return Ok immediately — the WS-style completion events for the + // native path already fire download-complete/download-failed, which + // the QueueManager must translate. + // + // REVISED: the native DownloadCoordinator already has a headless mode + // (spawn_headless) that returns an event receiver. But integrating + // that here is a larger refactor. For now, we send Start and rely on + // the existing emit paths. The permit for native tasks is released by + // a NEW listener added in Task 8 that watches download-complete/ + // download-failed and calls release_permit. + state + .download_coordinator + .send(crate::download::DownloadCmd::Start(Box::new( + crate::download::DownloadPayload { + id: download_id, + urls: crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()), + output_path: resolved_dest.join(safe_filename), + speed_limit: payload.speed_limit.clone(), + username: payload.username.clone(), + password: payload.password.clone(), + headers: payload.headers.clone(), + cookies: payload.cookies.clone(), + user_agent: payload.user_agent.clone(), + max_tries: mt, + proxy: payload.proxy.clone(), + }, + ))) + .await + .map_err(|e| e)?; + Ok(()) + } +} +``` + +**Important note on the native path:** unlike Media (which awaits to completion inside `run_media`), the native `DownloadCoordinator` is event-driven. `run_native` returns `Ok(())` immediately after sending `Start`, but the permit was parked by `dispatch_one`. This means the permit would leak for native tasks (never released). + +There are two options: +- **(A)** Add a `download-complete`/`download-failed` listener in `setup()` that calls `queue.release_permit(id)` for native + media-yt-dlp tasks. +- **(B)** Refactor `DownloadCoordinator` to expose a per-id completion future. + +Option A is far less invasive and matches how aria2 completions are already handled (via events). **We go with A.** This means `run_native` returning `Ok(())` immediately is fine — the `finish_runner` in `dispatch_one` will emit `Completed` too early though. + +To avoid the double-emit, change `run_native` to NOT return through `finish_runner`. Instead, native tasks are "fire and observe": `dispatch_one` should not spawn a runner that calls `finish_runner` for native; it should just send Start and let the completion listener handle release + emit. + +- [ ] **Revise `dispatch_one` in `queue.rs` for the Native case to not use a runner:** + +Replace the `TaskKind::Native =>` arm with: + +```rust + TaskKind::Native => { + // Native coordinator is event-driven (fire-and-observe). Send + // Start; completion is handled by the download-complete/ + // download-failed listener in lib.rs setup() which calls + // release_permit + apply_completion. + let this = Arc::clone(&self); + let payload = task.payload.clone(); + let id_for_task = id.clone(); + tauri::async_runtime::spawn(async move { + if let Err(error) = this.spawner.run_native(&id_for_task, &payload).await { + // Start itself failed — release + emit immediately. + this.emit_failed(&id_for_task, error); + this.release_permit(&id_for_task).await; + } + // On Ok, do nothing here — the completion listener owns release. + }); + } +``` + +And the Media arm stays as-is (it awaits to completion via `finish_runner`). + +### Step 7.2: Add `QueueManager` to `AppState` and instantiate in `setup()` + +- [ ] **In `src-tauri/src/lib.rs`, update the `AppState` struct (around line 436):** + +```rust +pub struct AppState { + pub download_coordinator: download::DownloadCoordinator, + pub extension_pairing_token: extension_server::SharedExtensionToken, + pub extension_frontend_ready: extension_server::SharedFrontendReady, + pub aria2_port: u16, + pub aria2_secret: String, + pub media_semaphore: Arc, + pub sleep_preventer: Arc>>, + pub aria2_gids: Arc>>, + pub queue_manager: Arc, +} +``` + +- [ ] **In `setup()` (around line 1464), update the `app.manage(AppState { ... })` block.** Read `max_concurrent_downloads` from the store first, then construct the QueueManager and spawn its dispatcher: + +Replace the existing `app.manage(AppState { ... })` block with: + +```rust + // Read max_concurrent_downloads from the settings store (default 3). + let max_concurrent = { + use tauri_plugin_store::StoreExt; + let mut capacity = crate::queue::DEFAULT_MAX_CONCURRENT; + if let Ok(store) = app.handle().store("store.bin") { + if let Some(settings_val) = store.get("settings") { + if let Some(settings_str) = settings_val.as_str() { + if let Ok(settings_json) = serde_json::from_str::(settings_str) { + if let Some(n) = settings_json.get("maxConcurrentDownloads").and_then(|v| v.as_u64()) { + capacity = n as usize; + } + } + } + } + } + capacity + }; + + let queue_manager = Arc::new(queue::QueueManager::new(app.handle().clone(), max_concurrent)); + let dispatcher_mgr = Arc::clone(&queue_manager); + tauri::async_runtime::spawn(async move { + dispatcher_mgr.run_dispatcher().await; + }); + + app.manage(AppState { + download_coordinator: download::DownloadCoordinator::spawn(app.handle().clone()), + extension_pairing_token, + extension_frontend_ready, + aria2_port, + aria2_secret: aria2_secret.clone(), + media_semaphore: Arc::new(tokio::sync::Semaphore::new(3)), + sleep_preventer: Arc::new(Mutex::new(None)), + aria2_gids: Arc::new(RwLock::new(std::collections::HashMap::new())), + queue_manager, + }); +``` + +Note: `QueueManager::new` currently takes `(AppHandle, usize)` and builds a `ProductionSpawner` internally. Update `new`: + +- [ ] **In `queue.rs`, update `QueueManager::new`:** + +```rust + pub fn new(app_handle: AppHandle, capacity: usize) -> Self { + let spawner: Arc = Arc::new(ProductionSpawner::new(app_handle.clone())); + Self { + pending: Mutex::new(VecDeque::new()), + semaphore: Arc::new(Semaphore::new(capacity)), + active_permits: Mutex::new(HashMap::new()), + target_capacity: AtomicUsize::new(capacity), + slots_to_retire: AtomicUsize::new(0), + notify: Notify::new(), + aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), + pending_completion: Arc::new(Mutex::new(HashMap::new())), + spawner, + app_handle, + } + } +``` + +Delete the old `with_spawner` method (already superseded by `test_new`). + +### Step 7.3: Add the Tauri commands + +- [ ] **In `src-tauri/src/lib.rs`, add these commands (place them near `start_download`, around line 600):** + +```rust +#[tauri::command] +async fn enqueue_download( + state: tauri::State<'_, AppState>, + id: String, + url: String, + destination: String, + filename: String, + connections: Option, + speed_limit: Option, + username: Option, + password: Option, + headers: Option, + checksum: Option, + cookies: Option, + mirrors: Option, + user_agent: Option, + max_tries: Option, + proxy: Option, + format_selector: Option, + cookie_source: Option, + is_media: Option, +) -> Result { + let media = is_media.unwrap_or(false); + let kind = if media { + queue::TaskKind::Media + } else { + queue::TaskKind::Aria2 + }; + let task = queue::QueuedTask { + id: id.clone(), + kind, + payload: queue::SpawnPayload { + url, + destination, + filename, + connections, + speed_limit, + username, + password, + headers, + checksum, + cookies, + mirrors, + user_agent, + max_tries, + proxy, + format_selector, + cookie_source, + is_media: media, + }, + }; + state.queue_manager.push(task).await; + Ok(id) +} +``` + +Wait — the spec says the frontend determines `is_media` and the backend infers `TaskKind` from it. But the aria2-vs-native distinction (the original `start_download` tries aria2 first, falls back to native) is lost. Decision: keep it simple. `is_media` true → Media. `is_media` false → Aria2 (with the native fallback handled inside `ProductionSpawner::add_uri`: if aria2 RPC fails, it should fall back to the native coordinator). The existing `start_download` already does this fallback; replicate it: + +- [ ] **Update `ProductionSpawner::add_uri` to fall back to native on aria2 RPC failure.** Replace the `let result = crate::rpc_call(...)` block at the end with: + +```rust + match crate::rpc_call(state.aria2_port, &state.aria2_secret, "aria2.addUri", params).await { + Ok(result) => { + let gid = result.as_str().unwrap_or("").to_string(); + Ok(gid) + } + Err(e) => { + // aria2 unavailable — fall back to native coordinator. But the + // native path is event-driven (not gid-based). We signal this + // by returning a synthetic gid prefixed with "native:" so the + // QueueManager knows to release via the completion listener, + // not the WS poller. + log::warn!("aria2 addUri failed, falling back to native: {}", e); + let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?; + let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let safe_filename = std::path::Path::new(&payload.filename.replace('\\', "/")) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("download") + .to_string(); + state + .download_coordinator + .send(crate::download::DownloadCmd::Start(Box::new( + crate::download::DownloadPayload { + id: download_id, + urls: crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()), + output_path: resolved_dest.join(safe_filename), + speed_limit: payload.speed_limit.clone(), + username: payload.username.clone(), + password: payload.password.clone(), + headers: payload.headers.clone(), + cookies: payload.cookies.clone(), + user_agent: payload.user_agent.clone(), + max_tries: mt, + proxy: payload.proxy.clone(), + }, + ))) + .await + .map_err(|e| e.to_string())?; + Ok(format!("native:{id}")) + } + } +``` + +This keeps the aria2-first-then-native behavior. The `native:` prefix tells the system that completion comes via `download-complete`/`download-failed` events (handled in Task 8), not the WS poller. + +- [ ] **Add the remaining commands to `lib.rs`:** + +```rust +#[tauri::command] +async fn enqueue_many( + state: tauri::State<'_, AppState>, + payloads: Vec, +) -> Result<(), AppError> +where + // SpawnPayloadCmd is a serde-friendly mirror of SpawnPayload + id + kind. +{ + // Simplify: accept Vec of (id, is_media) and re-read the rest from the + // frontend-provided payloads. But SpawnPayload has many fields. Define a + // single EnqueueManyItem struct. + unimplemented!("see EnqueueManyItem below") +} +``` + +The `enqueue_many` signature is getting complex. Define a deserializable command struct: + +- [ ] **In `queue.rs`, add a serde-friendly command payload (mirror of SpawnPayload + id + is_media):** + +```rust +#[derive(Debug, Clone, Deserialize)] +pub struct EnqueueItem { + pub id: String, + pub url: String, + pub destination: String, + pub filename: String, + pub connections: Option, + pub speed_limit: Option, + pub username: Option, + pub password: Option, + pub headers: Option, + pub checksum: Option, + pub cookies: Option, + pub mirrors: Option, + pub user_agent: Option, + pub max_tries: Option, + pub proxy: Option, + pub format_selector: Option, + pub cookie_source: Option, + pub is_media: Option, +} + +impl EnqueueItem { + pub(crate) fn into_task(self) -> QueuedTask { + let media = self.is_media.unwrap_or(false); + let kind = if media { + TaskKind::Media + } else { + TaskKind::Aria2 + }; + let id = self.id.clone(); + QueuedTask { + id, + kind, + payload: SpawnPayload { + url: self.url, + destination: self.destination, + filename: self.filename, + connections: self.connections, + speed_limit: self.speed_limit, + username: self.username, + password: self.password, + headers: self.headers, + checksum: self.checksum, + cookies: self.cookies, + mirrors: self.mirrors, + user_agent: self.user_agent, + max_tries: self.max_tries, + proxy: self.proxy, + format_selector: self.format_selector, + cookie_source: self.cookie_source, + is_media: media, + }, + } + } +} +``` + +- [ ] **Now define the clean commands in `lib.rs`.** Replace the stub `enqueue_download` and `enqueue_many` with: + +```rust +#[tauri::command] +async fn enqueue_download( + state: tauri::State<'_, AppState>, + item: queue::EnqueueItem, +) -> Result { + let id = item.id.clone(); + state.queue_manager.push(item.into_task()).await; + Ok(id) +} + +#[tauri::command] +async fn enqueue_many( + state: tauri::State<'_, AppState>, + items: Vec, +) -> Result<(), AppError> { + let tasks = items.into_iter().map(queue::EnqueueItem::into_task).collect(); + state.queue_manager.enqueue_many(tasks).await; + Ok(()) +} + +#[tauri::command] +async fn resume_download(state: tauri::State<'_, AppState>, id: String) -> Result<(), AppError> { + // Re-enqueue requires the payload; the frontend must provide it via the + // existing DownloadItem. Simplest: frontend calls enqueue_download with + // the full item again. This command is a thin alias for clarity. + // For now, return an error guiding the frontend to use enqueue_download. + // (We keep the command for API symmetry; frontend uses enqueue_download.) + Err(AppError::Internal( + "resume_download: use enqueue_download with the item payload".to_string(), + )) +} +``` + +Hmm — `resume_download` without a payload is awkward. Reconsider: the frontend has the full `DownloadItem`, so resume = `enqueue_download(item)`. We don't need a separate `resume_download` command. Drop it from the spec's command surface and have the frontend call `enqueue_download` for resume. + +- [ ] **Drop `resume_download`.** Remove it from the commands list. The frontend's `resumeDownload(id)` will call `enqueue_download` with the item's payload (rebuilt from the stored `DownloadItem`). + +- [ ] **Add `move_in_queue`, `remove_from_queue`, and rewrite `set_concurrent_limit` in `lib.rs`:** + +```rust +#[tauri::command] +async fn move_in_queue( + state: tauri::State<'_, AppState>, + id: String, + direction: crate::ipc::QueueDirection, +) -> Result, AppError> { + Ok(state.queue_manager.move_in_queue(&id, direction).await) +} + +#[tauri::command] +async fn remove_from_queue(state: tauri::State<'_, AppState>, id: String) -> Result { + Ok(state.queue_manager.remove_from_pending(&id).await) +} + +#[tauri::command] +async fn set_concurrent_limit(state: tauri::State<'_, AppState>, limit: usize) -> Result<(), String> { + state.queue_manager.set_capacity(limit); + Ok(()) +} +``` + +The old `set_concurrent_limit` body (the `rpc_call` to aria2) is removed — aria2's ceiling is fixed at startup (Task 7.5). + +### Step 7.4: Register the new commands in the invoke handler + +- [ ] **In `lib.rs`, update the `invoke_handler` list (around line 1700).** Replace the line: + +```rust + set_extension_frontend_ready, set_concurrent_limit, set_global_speed_limit, remove_download, +``` + +with: + +```rust + set_extension_frontend_ready, set_concurrent_limit, set_global_speed_limit, remove_download, + enqueue_download, enqueue_many, move_in_queue, remove_from_queue, +``` + +### Step 7.5: Raise the aria2 daemon ceiling + +- [ ] **In `lib.rs` `setup()`, find the aria2 spawn args (around line 1504) and add `--max-concurrent-downloads=100`:** + +```rust + cmd = cmd.arg("--enable-rpc=true") + .arg(format!("--rpc-listen-port={}", aria2_port)) + .arg(format!("--rpc-secret={}", aria2_secret)) + .arg("--rpc-listen-all=false") + .arg("--continue=true") + .arg("--allow-overwrite=false") + .arg("--summary-interval=1") + .arg("--console-log-level=warn") + .arg("--download-result=hide") + .arg("--check-certificate=true") + .arg("--max-concurrent-downloads=100"); +``` + +### Step 7.6: Make `resolve_path`, `is_safe_path`, `rpc_call`, `collect_download_uris`, `start_media_download_internal` accessible to `queue.rs` + +These are currently private functions in `lib.rs`. `ProductionSpawner` (in `queue.rs`) needs them. + +- [ ] **In `lib.rs`, change visibility** of these items from private to `pub(crate)`: + - `fn resolve_path(...)` → `pub(crate) fn resolve_path(...)` + - `fn is_safe_path(...)` → `pub(crate) fn is_safe_path(...)` + - `async fn rpc_call(...)` → `pub(crate) async fn rpc_call(...)` + - `fn collect_download_uris(...)` → `pub(crate) fn collect_download_uris(...)` + - `pub(crate) async fn start_media_download_internal(...)` → already `pub(crate)`, leave as is. + +### Step 7.7: Build and fix compilation errors + +- [ ] **Run:** `cd src-tauri && cargo build --lib` + Expected: a few errors likely around `AppState` field access from `queue.rs` (`state.aria2_port` etc. are `pub`, so fine) and the `media_semaphore` being unused now. Fix as they arise: + - If `media_semaphore` is now unused, keep it in `AppState` (harmless) or prefix with `_`. Keep it to minimize churn. + - The old `start_download`/`start_media_download` commands are still registered but now unused by the frontend. Leave them registered for now (the frontend migration in Task 8 will stop calling them); we remove them in Task 10. + +### Step 7.8: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/queue.rs src-tauri/src/lib.rs +git commit -m "feat(queue): wire QueueManager into AppState, add enqueue/move/remove commands" +``` + +--- + +## Task 8: Extend aria2 WS poller and add native/media completion listener + +**Goal:** Release permits on aria2 completion (WS poller), and on native/media completion (a new `download-complete`/`download-failed` listener). Wire the gid-completion race handling. + +**Files:** +- Modify: `src-tauri/src/lib.rs` + +### Step 8.1: Extend the WS poller to release permits on aria2 completion + +The existing WS poller (around line 1570) handles `aria2.onDownloadComplete` / `aria2.onDownloadError` by emitting frontend events. Add permit release + state emit via `queue_manager`. + +- [ ] **In `lib.rs`, find the WS poller's `match method { ... }` block (around line 1590) and update it.** Currently: + +```rust + if let Some(method) = json.get("method").and_then(|m| m.as_str()) { + if let Some(params) = json.get("params").and_then(|p| p.as_array()) { + if let Some(event) = params.first().and_then(|p| p.as_object()) { + if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) { + let id = { + let map = aria2_gids_clone1.read().unwrap(); + map.iter().find(|(_, g)| *g == gid).map(|(i, _)| i.clone()) + }; + if let Some(id) = id { + use tauri::Emitter; + match method { + "aria2.onDownloadComplete" => { + let _ = app_handle_ws.emit("download-complete", id.clone()); + // ... notification + } + "aria2.onDownloadError" => { + let _ = app_handle_ws.emit("download-failed", id); + } + _ => {} + } + } + } + } + } + } +``` + +The `aria2_gids_clone1` map is the old `id -> gid` map on `AppState`. But `QueueManager` now owns its own `aria2_gids` (gid -> id). The WS poller must use the QueueManager's map and call `handle_aria2_event`. + +- [ ] **Replace the WS poller's inner match block with QueueManager-aware logic.** The `app_handle_ws` already has access to state. Replace the block with: + +```rust + if let Some(method) = json.get("method").and_then(|m| m.as_str()) { + if let Some(params) = json.get("params").and_then(|p| p.as_array()) { + if let Some(event) = params.first().and_then(|p| p.as_object()) { + if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) { + let state = app_handle_ws.state::(); + let outcome = match method { + "aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete), + "aria2.onDownloadError" => { + let msg = event.get("error_message").and_then(|m| m.as_str()).unwrap_or("aria2 download error").to_string(); + Some(crate::queue::PendingOutcome::Error(msg)) + } + _ => None, + }; + if let Some(outcome) = outcome { + state.queue_manager.handle_aria2_event(gid, outcome).await; + } + } + } + } + } +``` + +This delegates to `handle_aria2_event`, which resolves gid→id (or buffers), applies completion (release permit + emit `download-state`), and handles the race. + +The old `download-complete`/`download-failed` emits for aria2 are now done inside `apply_completion` (via `emit_state`/`emit_failed`). But the frontend also listens for `download-complete`/`download-failed` for notifications (App.tsx). To keep notifications working, have `apply_completion` ALSO emit the legacy events. Update `apply_completion` in `queue.rs`: + +- [ ] **In `queue.rs`, update `apply_completion` and `finish_runner` and `emit_state`/`emit_failed` to also emit the legacy events.** Replace `emit_state`: + +```rust + fn emit_state(&self, id: impl Into, status: DownloadStatus) { + use tauri::Emitter; + let id: String = id.into(); + let _ = self.app_handle.emit( + "download-state", + DownloadStateEvent::new(id.clone(), status), + ); + // Legacy events for notification wiring (App.tsx listens to these). + match status { + DownloadStatus::Completed => { + let _ = self.app_handle.emit("download-complete", id); + } + _ => {} + } + } + + fn emit_failed(&self, id: &str, error: String) { + use tauri::Emitter; + let _ = self + .app_handle + .emit("download-state", DownloadStateEvent::failed(id, error.clone())); + let _ = self.app_handle.emit("download-failed", id.to_string()); + } +``` + +`finish_runner` and `apply_completion` already call these, so they're covered. + +### Step 8.2: Add the native/media completion listener + +Native tasks (and the aria2-fallback-to-native path) report completion via `download-complete`/`download-failed` events from the `DownloadCoordinator`. We need a backend listener that releases the permit for these. + +But wait — `finish_runner` (for Media) already releases the permit when `run_media` returns. And `emit_failed`/`emit_state` emit the legacy events. So if a native `download-complete` event fires, we'd double-handle. + +Clarify the release ownership: +- **Media (yt-dlp):** `run_media` awaits to completion → `finish_runner` releases + emits. The `download-complete` event for media is emitted INSIDE `start_media_download_internal` (it emits on yt-dlp termination). So Media would emit `download-complete` AND `finish_runner` would emit `download-state{completed}`. That's fine — both fire, the frontend handles `download-state` as source of truth and `download-complete` for notifications. +- **Native (and aria2-fallback):** `run_native` sends Start and returns Ok immediately. The `DownloadCoordinator` emits `download-complete`/`download-failed` when the native download finishes. We need a listener that, on these events for a native task, calls `release_permit` + `apply_completion`. + +So we need ONE backend listener on `download-complete`/`download-failed` that handles the native path. But it must not double-release Media/aria2 permits (which are already released by `finish_runner`/`handle_aria2_event`). Since `release_permit` is idempotent, double-release is harmless. But double-emitting `download-state` could cause flicker — also harmless because the status is the same. + +Simplify: add the listener; it calls `release_permit` (idempotent) and `apply_completion` (re-emits). For Media/aria2, the extra emit is redundant but not harmful. For Native, it's the only release path. This is the cleanest single-listener approach. + +- [ ] **In `lib.rs` `setup()`, after spawning the dispatcher, add a completion listener.** Find the `tauri::async_runtime::spawn` for the WS poller and add, after it: + +```rust + // Backend listener: release permits + emit terminal state for + // native (and aria2-fallback) downloads. Idempotent for Media/aria2 + // which already release via finish_runner/handle_aria2_event. + let completion_app = app.handle().clone(); + let completion_mgr = Arc::clone(&queue_manager); + tauri::async_runtime::spawn(async move { + let mut rx_complete = completion_app.listen("download-complete", move |event| { + let id = event.payload().to_string(); + let mgr = Arc::clone(&completion_mgr); + tauri::async_runtime::spawn(async move { + mgr.apply_completion(&id, crate::queue::PendingOutcome::Complete).await; + }); + }); + let completion_app2 = completion_app.clone(); + let completion_mgr2 = Arc::clone(&queue_manager); + let mut rx_failed = completion_app2.listen("download-failed", move |event| { + let id = event.payload().to_string(); + let mgr = Arc::clone(&completion_mgr2); + tauri::async_runtime::spawn(async move { + mgr.apply_completion(&id, crate::queue::PendingOutcome::Error("download failed".to_string())).await; + }); + }); + // Keep the task alive; the listeners are unregistered on drop. + std::future::pending::<()>().await; + drop(rx_complete); + drop(rx_failed); + }); +``` + +Hmm — `app.listen` returns an `EventId` (a guard), and the closure can't be async. The pattern above uses nested spawns, which is correct. But `completion_app.listen` borrows `completion_mgr` via the outer Arc clone — the `move` closure captures `completion_mgr` (the first listener) and `completion_mgr2` (the second). That's fine. But both listeners are registered inside one spawned task that then awaits `pending` — the `listen` calls happen immediately on the task's start, so they register before the `pending`. That works. + +Actually, there's a subtlety: `tauri::AppHandle::listen` registers globally and the returned `EventId` unregisters on drop. We want them to live for the app's lifetime. The `std::future::pending::<()>().await` keeps the task alive forever, so the guards never drop. Good. + +But `apply_completion` is `pub(crate)`; the listener is in `lib.rs` (same crate), so fine. + +- [ ] **Run:** `cd src-tauri && cargo build --lib` + Expected: compiles. Fix any borrow/move errors in the listener closures. + +### Step 8.3: Update `pause_download` and `remove_download` to release permits + +- [ ] **In `lib.rs`, update `pause_download` (around line 1012).** After the existing aria2 pause + coordinator pause + media pause, add permit release: + +```rust +#[tauri::command] +async fn pause_download(state: tauri::State<'_, AppState>, id: String) -> Result<(), String> { + println!("pause_download called for id: {}", id); + + // Release the concurrency slot FIRST, before signaling the sidecar to die. + state.queue_manager.release_permit(&id).await; + // Remove from pending if it was queued (not yet dispatched). + state.queue_manager.remove_from_pending(&id).await; + + // Emit the paused state. + use tauri::Emitter; + let _ = state.app_handle_emit_paused(&id); + + let gid = state.aria2_gids.read().unwrap().get(&id).cloned(); + if let Some(g) = gid { + if !g.starts_with("native:") { + let _ = rpc_call(state.aria2_port, &state.aria2_secret, "aria2.pause", serde_json::json!([g])).await; + } + } + + if let Ok(download_id) = Uuid::parse_str(&id) { + let _ = state + .download_coordinator + .send(download::DownloadCmd::Pause(download_id)) + .await; + } + state.download_coordinator.pause_media(id.clone()).await +} +``` + +We referenced `state.app_handle_emit_paused` which doesn't exist. The `AppHandle` isn't directly on `AppState`. Add an `app_handle` field to `AppState`, or get it from the command's `app_handle: tauri::AppHandle` parameter. Tauri commands can accept `app_handle: tauri::AppHandle` as an injected arg: + +- [ ] **Update `pause_download` signature to take `app_handle`:** + +```rust +#[tauri::command] +async fn pause_download( + app_handle: tauri::AppHandle, + state: tauri::State<'_, AppState>, + id: String, +) -> Result<(), String> { + println!("pause_download called for id: {}", id); + + // Release the concurrency slot FIRST, before signaling the sidecar to die. + state.queue_manager.release_permit(&id).await; + state.queue_manager.remove_from_pending(&id).await; + + // Emit the paused state. + use tauri::Emitter; + let _ = app_handle.emit( + "download-state", + crate::ipc::DownloadStateEvent::new(id.clone(), crate::ipc::DownloadStatus::Paused), + ); + + let gid = state.aria2_gids.read().unwrap().get(&id).cloned(); + if let Some(g) = gid { + if !g.starts_with("native:") { + let _ = rpc_call(state.aria2_port, &state.aria2_secret, "aria2.pause", serde_json::json!([g])).await; + } + } + + if let Ok(download_id) = Uuid::parse_str(&id) { + let _ = state + .download_coordinator + .send(download::DownloadCmd::Pause(download_id)) + .await; + } + state.download_coordinator.pause_media(id).await +} +``` + +- [ ] **Update `remove_download` (around line 1029) similarly** to call `remove_from_pending` + `release_permit`: + +```rust +#[tauri::command] +async fn remove_download( + app_handle: tauri::AppHandle, + state: tauri::State<'_, AppState>, + id: String, + filepath: Option, +) -> Result<(), String> { + println!("remove_download called for id: {}", id); + + // Remove from the queue (pending or active) and free the slot. + state.queue_manager.remove_from_pending(&id).await; + state.queue_manager.release_permit(&id).await; + + use tauri::Emitter; + let _ = app_handle.emit( + "download-state", + crate::ipc::DownloadStateEvent::new(id.clone(), crate::ipc::DownloadStatus::Paused), + ); + + if let Ok(download_id) = Uuid::parse_str(&id) { + state + .download_coordinator + .send(download::DownloadCmd::Cancel(download_id)) + .await?; + } + state.download_coordinator.pause_media(id.clone()).await?; + + if let Some(path) = filepath { + if !path.is_empty() { + let p = std::path::Path::new(&path); + if is_safe_path(p, &app_handle) { + if p.exists() { + let _ = tokio::fs::remove_file(p).await; + } + let aria2_path = format!("{}.aria2", path); + let p_aria2 = std::path::Path::new(&aria2_path); + if p_aria2.exists() { + let _ = tokio::fs::remove_file(p_aria2).await; + } + } + } + } + + Ok(()) +} +``` + +### Step 8.4: Build and commit + +- [ ] **Run:** `cd src-tauri && cargo build --lib` + Expected: compiles. + +- [ ] **Commit:** + +```bash +git add src-tauri/src/lib.rs src-tauri/src/queue.rs +git commit -m "feat(queue): release permits on aria2/native completion and pause/remove" +``` + +--- + +## Task 9: Frontend store refactor — reactive, `pendingOrder`, remove `processQueue` + +**Goal:** Make the store reactive. Remove `processQueue`, add the `pendingOrder` slice, add the `download-state` listener, and update all actions to invoke backend commands. + +**Files:** +- Modify: `src/store/useDownloadStore.ts` +- Modify: `src/store/downloadStore.ts` + +### Step 9.1: Refactor `useDownloadStore.ts` + +- [ ] **In `src/store/useDownloadStore.ts`:** + +1. Remove the `isProcessingQueue` let-binding and the entire `processQueue` method. +2. Add `pendingOrder: string[]` to the state interface and initial value `[]`. +3. Add `pauseDownload`, `resumeDownload`, `moveUp`, `moveDown` to the interface. +4. Update `addDownload`, `startQueue`, `pauseQueue`, `redownload`, `removeDownload`, `initDB`. + +Here are the specific edits. First, the interface — add after `processQueue` removal. Find: + +```typescript + processQueue: () => Promise; + startQueue: (queueId: string) => Promise; +``` + +Replace with: + +```typescript + pendingOrder: string[]; + startQueue: (queueId: string) => Promise; +``` + +Find the `let isProcessingQueue = false;` line (above `create`) and delete it. + +Add `pendingOrder: [],` to the initial state object (near `downloads: [],`). + +- [ ] **Replace `addDownload`:** + +```typescript + addDownload: (item) => { + info(`Download ${item.id} added to queue`); + set((state) => ({ downloads: [...state.downloads, item] })); + // Backend owns dispatch. The download-state listener appends to pendingOrder. + invoke('enqueue_download', { item: buildEnqueueItem(item) }).catch((e) => { + console.error('Failed to enqueue download:', e); + }); + }, +``` + +Add the `buildEnqueueItem` helper above `create`: + +```typescript +const buildEnqueueItem = (item: DownloadItem) => ({ + id: item.id, + url: item.url, + destination: item.destination || '', + filename: item.fileName, + connections: item.connections ?? null, + speedLimit: item.speedLimit || null, + username: item.username || null, + password: item.password || null, + headers: item.headers || null, + checksum: item.checksum || null, + cookies: item.cookies || null, + mirrors: item.mirrors || null, + userAgent: null, + maxTries: null, + proxy: null, + formatSelector: item.mediaFormatSelector || null, + cookieSource: null, + isMedia: item.isMedia ?? false, +}); +``` + +Wait — the backend `EnqueueItem` uses snake_case fields? No: Tauri commands receive args as JS objects with the **camelCase** names matching the Rust field names AFTER Tauri's automatic snake→camel conversion. Tauri converts `speed_limit` → `speedLimit` at the JS boundary. So `buildEnqueueItem` must use camelCase keys. The object above does. Good. + +- [ ] **Replace `updateDownload`** (remove the `processQueue` triggers): + +```typescript + updateDownload: (id, updates) => { + set((state) => ({ + downloads: state.downloads.map(d => { + if (d.id === id) { + return { + ...d, + ...updates, + fraction: updates.fraction !== undefined ? updates.fraction : d.fraction + }; + } + return d; + }) + })); + // Sync system integrations on status change; no more processQueue. + if (updates.status) { + syncSystemIntegrations(); + } + }, +``` + +- [ ] **Replace `removeDownload`:** + +```typescript + removeDownload: async (id, deleteFile = false) => { + const item = get().downloads.find(d => d.id === id); + if (item) { + try { + await invoke('remove_download', { id, filepath: null }); + } catch (e) { + console.error("Failed to terminate download on deletion:", e); + } + } + if (item && deleteFile) { + try { + const filepath = item.destination ? `${item.destination}/${item.fileName}` : null; + if (filepath) { + const partialPaths = [`${filepath}.aria2`, `${filepath}.part`]; + await invoke('trash_download_assets', { path: filepath, partialPaths }); + } + } catch (e) { + console.error("Failed to trash file from disk:", e); + } + } + set((state) => ({ + downloads: state.downloads.filter(d => d.id !== id), + pendingOrder: state.pendingOrder.filter(x => x !== id), + })); + info(`Download ${id} removed`); + syncSystemIntegrations(); + }, +``` + +- [ ] **Replace `redownload`:** + +```typescript + redownload: (id) => { + const item = get().downloads.find(d => d.id === id); + if (!item) return; + set((state) => ({ + downloads: state.downloads.map(d => { + if (d.id === id) { + return { ...d, status: 'queued', fraction: 0, speed: '-', eta: '-' }; + } + return d; + }) + })); + // If it was downloading, pause the old sidecar first. + invoke('pause_download', { id }).catch(console.error); + // Re-enqueue with the stored payload. + invoke('enqueue_download', { item: buildEnqueueItem(item) }).catch(console.error); + info(`Download ${id} redownload requested (queued)`); + }, +``` + +- [ ] **Replace `startQueue`:** + +```typescript + startQueue: async (queueId) => { + const items = get().downloads + .filter(item => item.queueId === queueId && (item.status === 'paused' || item.status === 'failed')) + .map(item => buildEnqueueItem(item)); + + if (items.length === 0) return 0; + + await invoke('enqueue_many', { items }); + info(`Queue ${queueId} started, ${items.length} items re-enqueued`); + return items.length; + }, +``` + +- [ ] **Replace `pauseQueue`:** + +```typescript + pauseQueue: async (queueId) => { + const activeIds = get().downloads + .filter(item => item.queueId === queueId && (item.status === 'downloading' || item.status === 'queued')) + .map(item => item.id); + + if (activeIds.length === 0) return 0; + + await Promise.all(activeIds.map(id => invoke('pause_download', { id }).catch(() => {}))); + info(`Queue ${queueId} paused, ${activeIds.length} items paused`); + syncSystemIntegrations(); + return activeIds.length; + }, +``` + +- [ ] **Replace `initDB`** (re-enqueue interrupted + queued items on startup): + +```typescript + initDB: async () => { + try { + const queues = await tauriStore.get('queues') || []; + const downloads = await tauriStore.get('download_queue') || []; + + // Treat persisted 'downloading' as interrupted; reset to 'queued'. + const normalized = downloads.map(d => + d.status === 'downloading' ? { ...d, status: 'queued' as const } : d + ); + + set(state => ({ + queues: queues.length > 0 ? queues : state.queues, + downloads: normalized.length > 0 ? normalized : state.downloads, + pendingOrder: normalized.filter(d => d.status === 'queued').map(d => d.id), + })); + + // Re-enqueue all queued items via the backend. + const enqueueItems = normalized + .filter(d => d.status === 'queued') + .map(d => buildEnqueueItem(d)); + if (enqueueItems.length > 0) { + await invoke('enqueue_many', { items: enqueueItems }); + } + } catch (e) { + console.error("Failed to init DB", e); + } + }, +``` + +- [ ] **Add the new actions** (`pauseDownload`, `resumeDownload`, `moveUp`, `moveDown`) to the interface and implementation. Add to the interface: + +```typescript + pauseDownload: (id: string) => Promise; + resumeDownload: (item: DownloadItem) => void; + moveUp: (id: string) => void; + moveDown: (id: string) => void; +``` + +And the implementations (place near the other actions): + +```typescript + pauseDownload: async (id) => { + await invoke('pause_download', { id }).catch(console.error); + }, + resumeDownload: (item) => { + // Resume = re-enqueue with the stored payload. + set((state) => ({ + downloads: state.downloads.map(d => + d.id === item.id ? { ...d, status: 'queued', speed: '-', eta: '-' } : d + ) + })); + invoke('enqueue_download', { item: buildEnqueueItem(item) }).catch(console.error); + }, + moveUp: (id) => { + invoke<[string]>('move_in_queue', { id, direction: 'up' }).then((order) => { + set({ pendingOrder: order }); + }).catch(console.error); + }, + moveDown: (id) => { + invoke<[string]>('move_in_queue', { id, direction: 'down' }).then((order) => { + set({ pendingOrder: order }); + }).catch(console.error); + }, +``` + +The `move_in_queue` return type is `Vec` → TS `string[]`. Fix the generic: `invoke('move_in_queue', ...)`. + +- [ ] **Fix the import for `invokeCommand`.** The file imports `invokeCommand as invoke` — confirm `moveUp`/`moveDown` use `invoke`. + +### Step 9.2: Add the `download-state` listener and simplify the progress listener + +- [ ] **In `src/store/downloadStore.ts`:** + +1. Add the `download-state` listener (authoritative for status + pendingOrder). +2. Simplify the `download-progress` listener (remove the `queued → downloading` flip). + +Find the existing `initDownloadListener` and replace its body: + +```typescript +import type { DownloadStateEvent } from '../bindings/DownloadStateEvent'; + +export async function initDownloadListener() { + if (unlistenProgress) return; + + // download-progress: updates only the transient progress store. Status + // transitions are owned by download-state below. + unlistenProgress = await listen('download-progress', (event) => { + const payload = event.payload; + useDownloadProgressStore.getState().updateDownloadProgress(payload.id, payload); + + const mainStore = useDownloadStore.getState(); + const current = mainStore.downloads.find(d => d.id === payload.id); + if (current && payload.size && current.size !== payload.size) { + mainStore.updateDownload(payload.id, { size: payload.size }); + } + }); + + // download-state: the single source of truth for status transitions. + // Also authoritative for pendingOrder membership. + unlistenState = await listen('download-state', (event) => { + const { id, status, error } = event.payload; + const mainStore = useDownloadStore.getState(); + mainStore.updateDownload(id, { + status: status as any, + ...(error ? { speed: '-', eta: '-' } : {}), + }); + // pendingOrder: append on queued, remove otherwise. + if (status === 'queued') { + useDownloadStore.setState((s) => ({ + pendingOrder: s.pendingOrder.includes(id) ? s.pendingOrder : [...s.pendingOrder, id] + })); + } else { + useDownloadStore.setState((s) => ({ + pendingOrder: s.pendingOrder.filter(x => x !== id) + })); + } + }); + + if (!unlistenTray) { + unlistenTray = await listen('tray-action', (event) => { + const mainStore = useDownloadStore.getState(); + if (event.payload === 'pause-all') { + const uniqueQueues = Array.from(new Set(mainStore.downloads.map(d => d.queueId))); + uniqueQueues.forEach(qid => mainStore.pauseQueue(qid)); + } else if (event.payload === 'resume-all') { + const uniqueQueues = Array.from(new Set(mainStore.downloads.map(d => d.queueId))); + uniqueQueues.forEach(qid => mainStore.startQueue(qid)); + } + }); + } +} +``` + +- [ ] **Add the `unlistenState` variable** near the top: + +```typescript +let unlistenProgress: UnlistenFn | null = null; +let unlistenState: UnlistenFn | null = null; +let unlistenTray: UnlistenFn | null = null; +``` + +### Step 9.3: Update `App.tsx` to remove `set_concurrent_limit` double-call and handle `download-state` for notifications + +- [ ] **In `src/App.tsx`:** The `download-complete` listener (around line 211) updates status. Since `download-state` now owns status, the `download-complete` listener should only handle notifications, not status. Find: + +```typescript + const unlistenComplete = listen('download-complete', (event) => { + updateDownload(event.payload, { status: 'completed', fraction: 1.0, speed: '-', eta: '-' }); + ... +``` + +Replace the `updateDownload(... status: 'completed' ...)` with just the fraction/speed/eta cleanup (status is set by download-state): + +```typescript + const unlistenComplete = listen('download-complete', (event) => { + updateDownload(event.payload, { fraction: 1.0, speed: '-', eta: '-' }); + const settings = useSettingsStore.getState(); + if (settings.showNotifications) { + const item = useDownloadStore.getState().downloads.find(d => d.id === event.payload); + const fileName = item?.fileName || 'A file'; + sendNotification({ + title: 'Download Complete', + body: `${fileName} has finished downloading.`, + sound: settings.playCompletionSound ? 'default' : undefined + }); + } + }); +``` + +The `download-failed` listener similarly: remove the status mutation (download-state owns it); keep the guard only for notifications if any. Find: + +```typescript + const unlistenFailed = listen('download-failed', (event) => { + const current = useDownloadStore.getState().downloads.find(d => d.id === event.payload); + if (current && current.status !== 'paused') { + updateDownload(event.payload, { status: 'failed', speed: '-', eta: '-' }); + } + }); +``` + +Since `download-state` now sets `failed`, this listener can be removed or left as a no-op. Remove it to avoid confusion: + +- [ ] **Remove the `unlistenFailed` listener block and its cleanup in the return.** + +- [ ] **The `set_concurrent_limit` useEffect (line 75) stays as-is** — it now calls the rewritten backend command that resizes the semaphore. + +### Step 9.4: Typecheck and commit + +- [ ] **Run:** `npx tsc --noEmit` + Expected: PASS (no `_dispatched`, no `processQueue` references remain). + +- [ ] **Commit:** + +```bash +git add src/store/useDownloadStore.ts src/store/downloadStore.ts src/App.tsx +git commit -m "feat(store): reactive queue — pendingOrder, download-state listener, remove processQueue" +``` + +--- + +## Task 10: `DownloadTable.tsx` and `DownloadItem.tsx` UI updates + +**Goal:** Update `DownloadTable`'s `handleResume`/`handlePause` to use the store actions, and update `DownloadItem` with the queued visuals + Move Up/Down buttons. + +**Files:** +- Modify: `src/components/DownloadTable.tsx` +- Modify: `src/components/DownloadItem.tsx` + +### Step 10.1: Update `DownloadTable.tsx` handlers + +- [ ] **In `src/components/DownloadTable.tsx`, update `handlePause` and `handleResume` (around line 93):** + +Replace: + +```typescript + const handlePause = async (id: string) => { + try { + await invoke('pause_download', { id }); + updateDownload(id, { status: 'paused', speed: '-', eta: '-' }); + } catch (e) { + console.error("Failed to pause:", e); + } + }; + + const handleResume = (item: DownloadItem) => { + updateDownload(item.id, { status: 'queued', _dispatched: false, speed: '-', eta: '-' }); + useDownloadStore.getState().processQueue(); + }; +``` + +with: + +```typescript + const handlePause = async (id: string) => { + try { + await invoke('pause_download', { id }); + // download-state listener sets 'paused'; no manual update needed. + } catch (e) { + console.error("Failed to pause:", e); + } + }; + + const handleResume = (item: DownloadItem) => { + useDownloadStore.getState().resumeDownload(item); + }; +``` + +Also the "Resume All" button (line 150) calls `handleResume(d)` for paused items — that still works. + +The "Pause All" button (line 161) calls `handlePause(d.id)` for downloading items — but should also pause queued items. Update the filter: + +- [ ] **Update the Pause All onClick filter (line 161):** + +```typescript + onClick={() => { + filteredDownloads.filter(d => d.status === 'downloading' || d.status === 'queued').forEach(d => handlePause(d.id)); + }} +``` + +### Step 10.2: Update `DownloadItem.tsx` with queued visuals and Move Up/Down + +- [ ] **In `src/components/DownloadItem.tsx`:** + +1. Import `Clock`, `ChevronUp`, `ChevronDown` from lucide-react. +2. Add `moveUp`, `moveDown` from the store. +3. Compute `queuePosition` and `queueLength` from `pendingOrder`. +4. Add the queued status cell branch. +5. Add Move Up/Down buttons in the action cluster. + +Replace the imports (line 4): + +```typescript +import { Play, Pause, MoreVertical, Clock, ChevronUp, ChevronDown } from 'lucide-react'; +``` + +Replace the component body's store hooks (after `const download = ...` line 26) — add: + +```typescript + const pendingOrder = useDownloadStore(state => state.pendingOrder); + const moveUp = useDownloadStore(state => state.moveUp); + const moveDown = useDownloadStore(state => state.moveDown); + const queuePosition = pendingOrder.indexOf(downloadId) + 1; + const queueLength = pendingOrder.length; +``` + +Replace the status cell (lines 84-106). The new version handles `queued` explicitly: + +```tsx +
+ {download.status === 'completed' ? ( + Completed + ) : download.status === 'queued' ? ( + + + Queued + {queuePosition > 0 && ( + #{queuePosition} + )} + + ) : ( + <> +
+
+
+ + {download.status === 'downloading' + ? `${((download.fraction || 0) * 100).toFixed(0)}%` + : download.status.charAt(0).toUpperCase() + download.status.slice(1)} + + + )} +
+``` + +Replace the action cluster (lines 116-143) to add Move Up/Down for queued items: + +```tsx +
+ + {download.dateAdded ? new Date(download.dateAdded).toLocaleDateString() : '-'} + + +
+ {download.status === 'downloading' && ( + + )} + {download.status === 'paused' && ( + + )} + {download.status === 'queued' && ( + <> + + + + )} + +
+
+``` + +### Step 10.3: Add the CSS for queued visuals + +- [ ] **In `src/index.css`, after the `.download-progress-fill.paused { ... }` block (line 1066):** + +```css + .download-progress-fill.queued { + filter: grayscale(1) opacity(0.45); + background: hsl(var(--text-muted)); + } + + .download-progress-track:has(.download-progress-fill.queued) { + background-image: repeating-linear-gradient( + 45deg, + hsl(var(--border-color)), + hsl(var(--border-color)) 6px, + hsl(var(--border-color) / 0.6) 6px, + hsl(var(--border-color) / 0.6) 12px + ); + background-size: 200% 100%; + animation: queued-stripe 1.5s linear infinite; + } + + @keyframes queued-stripe { + from { background-position: 0 0; } + to { background-position: 24px 0; } + } +``` + +- [ ] **After `.download-status-completed { ... }` (around line 1079), add:** + +```css + .download-status-queued { + display: inline-flex; + align-items: center; + gap: 4px; + color: hsl(var(--text-muted)); + font-weight: 500; + } + + .queue-position-badge { + margin-left: 2px; + padding: 0 5px; + border-radius: 8px; + background: hsl(var(--item-hover)); + color: hsl(var(--text-secondary)); + font-size: 10px; + font-weight: 600; + font-variant-numeric: tabular-nums; + } + + .animate-pulse-slow { + animation: pulse-slow 2s cubic-bezier(0.4, 0, 0.6, 1) infinite; + } + + @keyframes pulse-slow { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.4; } + } +``` + +Note: `--text-secondary` may not be a defined CSS variable. Check existing usage — the codebase uses `--text-primary`, `--text-muted`. Use `--text-muted` for the badge text: + +- [ ] **In the `.queue-position-badge` rule, replace `color: hsl(var(--text-secondary));` with `color: hsl(var(--text-muted));`.** + +### Step 10.4: Typecheck and build + +- [ ] **Run:** `npx tsc --noEmit` + Expected: PASS. + +- [ ] **Run:** `npm run build` + Expected: PASS (vite build succeeds). + +### Step 10.5: Commit + +- [ ] **Commit:** + +```bash +git add src/components/DownloadTable.tsx src/components/DownloadItem.tsx src/index.css +git commit -m "feat(ui): queued-state visuals (clock, grayscale, position badge) and Move Up/Down buttons" +``` + +--- + +## Task 11: Remove deprecated `start_download`/`start_media_download` frontend entry points and dead code + +**Goal:** Clean up. The frontend no longer calls `start_download`/`start_media_download` directly (it uses `enqueue_download`). Remove the now-unused command registrations and the `_dispatched`-related dead code. Keep the `pub(crate)` functions (used by the ProductionSpawner). + +**Files:** +- Modify: `src-tauri/src/lib.rs` +- Modify: `src-tauri/src/ipc.rs` (if any `_dispatched` serde rename remains) + +### Step 11.1: Remove `start_download` and `start_media_download` from the invoke handler + +These commands are now only called internally by the ProductionSpawner (via the `pub(crate)` functions, not the `#[tauri::command]` wrappers). Remove the `#[tauri::command]` wrappers from the invoke handler list. + +- [ ] **In `src-tauri/src/lib.rs`, update the `invoke_handler` list (around line 1700).** Remove `start_download, start_media_download,` from: + +```rust + start_download, start_media_download, pause_download, fetch_metadata, fetch_media_metadata, +``` + +becomes: + +```rust + pause_download, fetch_metadata, fetch_media_metadata, +``` + +Keep the `start_download` and `start_media_download` function definitions but remove their `#[tauri::command]` attribute (so they're not registered). Actually — `start_media_download_internal` is `pub(crate)` and used by the spawner; the `start_media_download` wrapper (the `#[tauri::command]`) is now unused. And `start_download` (the `#[tauri::command]`) is unused. + +- [ ] **Remove the `#[tauri::command]` attribute from `start_download` and `start_media_download`** (or delete the functions if nothing else references them). Simplest: delete the `start_media_download` command wrapper entirely (it just set up the semaphore + cancel_rx + called `start_media_download_internal`; the spawner now does this inline). And delete the `start_download` command (the spawner's `add_uri` replaces it). + +Check `initDB` in the old frontend — it used to call `start_download`/`start_media_download`. The refactored `initDB` (Task 9.1) uses `enqueue_many`. Confirmed safe to delete. + +- [ ] **Delete the `start_download` function (lines ~599-713) and the `start_media_download` function (lines ~715-772) from `lib.rs`.** Keep `start_media_download_internal` (it's `pub(crate)` and called by the spawner). + +### Step 11.2: Verify build and run tests + +- [ ] **Run:** `cd src-tauri && cargo test -- --nocapture` + Expected: all queue_manager tests + existing download_engine tests + export_bindings pass. + +- [ ] **Run:** `cd src-tauri && cargo build --lib` + Expected: compiles. Fix any "unused import" warnings. + +- [ ] **Run:** `npx tsc --noEmit && npm run build` + Expected: PASS. + +### Step 11.3: Commit + +- [ ] **Commit:** + +```bash +git add src-tauri/src/lib.rs +git commit -m "refactor: remove deprecated start_download/start_media_download command wrappers" +``` + +--- + +## Task 12: Manual smoke test and final verification + +**Goal:** Verify the full system end-to-end. No code changes — just runtime verification. + +### Step 12.1: Build and run the app + +- [ ] **Run:** `npm run tauri dev` (in a separate terminal; this is interactive — run it and observe). + Expected: app launches. + +### Step 12.2: Smoke test scenarios (from spec §8) + +Perform each and verify: + +- [ ] **Scenario 1 — basic concurrency:** Set `maxConcurrentDownloads = 3` in Settings. Add 5 downloads (use slow/large URLs or localhost test files). Verify: + - 3 dispatch (status `downloading`, progress bars animate). + - 2 show `Queued` with clock icon + `#1`/`#2` badges. + +- [ ] **Scenario 2 — shrink mid-flight:** While 3 are downloading and 2 queued, change `maxConcurrentDownloads` to 1. Verify: + - Active downloads are NOT killed (the 3 continue). + - As each completes, the queued count does NOT grow back to 3 — it converges toward 1 active. + +- [ ] **Scenario 3 — pause releases instantly:** While 1 is downloading and 1 is queued (capacity 1), pause the downloading one. Verify: + - The queued item dispatches immediately (within ~100ms), before the paused sidecar fully exits. + +- [ ] **Scenario 4 — Move Up/Down:** With 3 queued items, click Move Up on the 3rd. Verify: + - The `#N` badges update instantly (`#3`→`#2`, `#2`→`#3`). + - The disabled state on the buttons updates (first item's Move Up disabled). + +- [ ] **Scenario 5 — restart recovery:** While 2 items are downloading, quit and relaunch the app. Verify: + - Both show `Queued` on restart (interrupted → queued). + - They re-enqueue and resume downloading (up to capacity). + +- [ ] **Scenario 6 — aria2 path specifically:** Add a non-media HTTP download. Verify it dispatches via aria2 (check logs for `aria2 addUri`), the permit is released on completion (a queued item claims the slot). + +- [ ] **Scenario 7 — yt-dlp media path:** Add a YouTube URL. Verify it dispatches via yt-dlp when a slot is free, and the permit releases on completion. + +### Step 12.3: Final commit (if any test surfaced a fix) + +- [ ] If any smoke test failed and you fixed it, commit the fix with a descriptive message. Otherwise, no commit needed. + +--- + +## Self-Review + +**1. Spec coverage:** + +| Spec section | Task(s) | +|---|---| +| §4 Architecture (QueueManager sole gatekeeper) | 1, 2, 7 | +| §5.1 Dispatcher loop (idle-park + CAS + re-pop) | 3 | +| §5.2 Wakeup invariant (notify on enqueue/release/grow) | 2, 3, 5 | +| §5.3 Uniform permit parking (aria2 trap + pause release) | 4, 8 | +| §5.4 Resizing (grow + shrink) | 3 | +| §5.5 GID lookup race (pending_completion buffer) | 4 | +| §5.6/§6.1 aria2 high ceiling | 7 (Step 7.5) | +| §6 State-transition protocol (download-state event) | 6, 8 | +| §7.1-7.3 Frontend store refactor | 9 | +| §7.4 DownloadItem queued UI + Move Up/Down | 10 | +| §8 Testing (all 10 regression tests) | 2, 3, 4, 5 | + +All spec sections covered. ✅ + +**2. Placeholder scan:** + +- Searched for "TBD", "TODO", "implement later", "fill in" — none in task steps. +- Step 7.7 ("Fix as they arise") is acceptable — it's a build-and-fix checkpoint, not a placeholder for unspecified content. +- All code blocks are complete. + +**3. Type/signature consistency:** + +- `QueueDirection` — defined in Task 1 (`queue.rs`), moved to `ipc.rs` in Task 6, used in Task 5 (`move_in_queue`) and Task 7 (command). Consistent. ✅ +- `DownloadStateEvent` — defined in Task 1, moved to `ipc.rs` in Task 6, used in Tasks 4/8. Consistent. ✅ +- `EnqueueItem` — defined in Task 7 (`queue.rs`), used in Task 9 (`buildEnqueueItem` produces matching camelCase keys). ✅ +- `SidecarSpawner::add_uri` returns `Result` (gid) — consistent across Task 1 (trait), Task 4 (test), Task 7 (production). ✅ +- `release_permit`, `apply_completion`, `handle_aria2_event`, `remember_gid` — used consistently in Tasks 4, 5, 8. ✅ +- The `native:` gid prefix convention (Task 7 `add_uri` fallback) is checked in Task 8 (`pause_download` skips aria2.pause for `native:` gids). ✅ + +One inconsistency found and fixed inline during review: the original Task 2 had `acquire_and_park` which double-acquired in `dispatch_one`. Split into `acquire_permit` + `park_permit` in Step 3.3 and the Task 2.1 test updated accordingly. ✅ + +**Scope check:** This is one cohesive subsystem (the queue coordinator) with backend + frontend changes that must ship together. Appropriately scoped for a single plan. ✅