feat(download): metalink mirrors fallback in engine (task 40)

src-tauri/src/adapters/driven/event/tauri_bridge.rs

@@ -77,6 +77,8 @@ fn event_name(event: &DomainEvent) -> &'static str {
         DomainEvent::AccountSelected { .. } => "account-selected",
         DomainEvent::AccountExhausted { .. } => "account-exhausted",
         DomainEvent::LinkStatusUpdated { .. } => "link-status-updated",
+        DomainEvent::MirrorSwitched { .. } => "mirror-switched",
+        DomainEvent::AllMirrorsExhausted { .. } => "mirrors-exhausted",
     }
 }
 
@@ -271,6 +273,18 @@ fn event_payload(event: &DomainEvent) -> serde_json::Value {
                 "expiredNaturally": expired_naturally,
             })
         }
+        DomainEvent::MirrorSwitched {
+            id,
+            new_mirror_index,
+            new_url,
+        } => {
+            json!({
+                "id": id.0,
+                "newMirrorIndex": new_mirror_index,
+                "newUrl": new_url,
+            })
+        }
+        DomainEvent::AllMirrorsExhausted { id } => json!({ "id": id.0 }),
     }
 }
 

src-tauri/src/adapters/driven/logging/download_log_bridge.rs

@@ -163,7 +163,9 @@ fn record_download_event(store: &DownloadLogStore, event: &DomainEvent) {
         | DomainEvent::AccountSelected { .. }
         | DomainEvent::AccountExhausted { .. }
         | DomainEvent::LinkStatusUpdated { .. }
-        | DomainEvent::SplitArchiveIncomplete { .. } => {}
+        | DomainEvent::SplitArchiveIncomplete { .. }
+        | DomainEvent::MirrorSwitched { .. }
+        | DomainEvent::AllMirrorsExhausted { .. } => {}
     }
 }
 

src-tauri/src/adapters/driven/notification/notification_bridge.rs

@@ -219,6 +219,8 @@ mod tests {
             resume_supported: true,
             retry_count: 0,
             max_retries: 5,
+            mirrors: Vec::new(),
+            current_mirror_index: 0,
             created_at,
             updated_at: created_at + 60,
         }

src-tauri/src/adapters/driven/sqlite/download_read_repo.rs

@@ -5,8 +5,8 @@ use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QueryOr
 use crate::domain::error::DomainError;
 use crate::domain::model::download::{DownloadId, DownloadState};
 use crate::domain::model::views::{
-    DownloadDetailView, DownloadFilter, DownloadView, SegmentView, SortDirection, SortField,
-    SortOrder, StateCountMap,
+    DownloadDetailView, DownloadFilter, DownloadView, MirrorView, SegmentView, SortDirection,
+    SortField, SortOrder, StateCountMap,
 };
 use crate::domain::ports::driven::download_read_repository::DownloadReadRepository;
 
@@ -275,6 +275,24 @@ impl DownloadReadRepository for SqliteDownloadReadRepo {
                 if stored > 0 { stored } else { created_at }
             };
 
+            let mirrors: Vec<MirrorView> =
+                download::deserialize_mirrors(model.mirrors_json.as_deref())?
+                    .into_iter()
+                    .map(|m| MirrorView {
+                        url: m.url().as_str().to_string(),
+                        priority: m.priority(),
+                        country: m.country().map(|s| s.to_string()),
+                    })
+                    .collect();
+            // Clamp the persisted cursor against the live mirror list — a row
+            // with mirrors_json shrunk by a manual edit (or a future migration
+            // that drops entries) must not surface an out-of-range slot.
+            let clamped_mirror_index = if mirrors.is_empty() {
+                0
+            } else {
+                safe_u32(model.current_mirror_index as i64).min((mirrors.len() - 1) as u32)
+            };
+
             let detail = DownloadDetailView {
                 id: DownloadId(safe_u64(model.id)),
                 file_name: model.file_name.clone(),
@@ -296,6 +314,8 @@ impl DownloadReadRepository for SqliteDownloadReadRepo {
                 resume_supported: model.resume_supported != 0,
                 retry_count: safe_u32(model.retry_count as i64),
                 max_retries: safe_u32(model.max_retries as i64),
+                mirrors,
+                current_mirror_index: clamped_mirror_index,
                 created_at,
                 updated_at,
             };
@@ -379,6 +399,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/downloads".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(1000 + id),
             updated_at: Set(2000 + id),
         };
@@ -482,6 +504,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/downloads/video.mp4".to_string()),
             error_message: Set(Some("tls handshake failed".to_string())),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(1003),
             updated_at: Set(2003),
         };
@@ -577,6 +601,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/downloads/file1.zip".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(0),
             updated_at: Set(0),
         };
@@ -618,6 +644,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/downloads/file1.zip".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(0),
             updated_at: Set(1_700_000_000_123_i64),
         };
@@ -660,6 +688,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/downloads/file1.zip".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(0),
             updated_at: Set(0),
         };
@@ -701,6 +731,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/a.zip".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(1000),
             updated_at: Set(2000),
         };
@@ -727,6 +759,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/b.zip".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(1001),
             updated_at: Set(2001),
         };
@@ -774,6 +808,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp/downloads/legacy.zip".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(0),
             updated_at: Set(0),
         };

src-tauri/src/adapters/driven/sqlite/download_repo.rs

@@ -68,6 +68,12 @@ async fn persist_download<C: ConnectionTrait>(
         // MAX expression below so that progress_bridge writes
         // (which may race with state-transition saves) are never
         // regressed back to a stale lower value.
+        // CurrentMirrorIndex excluded: progress_bridge owns this
+        // column via MirrorSwitched / DownloadFailed events. A
+        // generic save() carrying a stale in-memory cursor would
+        // race with the event-driven write and overwrite a fresh
+        // failover position with the cursor at the time the
+        // aggregate was loaded.
         .update_columns([
             download::Column::Url,
             download::Column::FileName,
@@ -87,6 +93,7 @@ async fn persist_download<C: ConnectionTrait>(
             download::Column::ModuleName,
             download::Column::AccountId,
             download::Column::DestinationPath,
+            download::Column::MirrorsJson,
         ]);
     if update_error_message {
         on_conflict.update_column(download::Column::ErrorMessage);
@@ -213,6 +220,72 @@ mod tests {
         )
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_save_round_trips_mirrors_with_priority_and_country() {
+        let db = setup_test_db().await.expect("test db");
+        let repo = SqliteDownloadRepo::new(db);
+
+        let mirror_a = crate::domain::model::Mirror::new(
+            Url::new("https://a.example.com/file.zip").unwrap(),
+            80,
+            Some("US".to_string()),
+        )
+        .unwrap();
+        let mirror_b = crate::domain::model::Mirror::new(
+            Url::new("https://b.example.com/file.zip").unwrap(),
+            40,
+            None,
+        )
+        .unwrap();
+
+        let download = make_download(42).with_mirrors(vec![mirror_a, mirror_b]);
+        repo.save(&download).expect("save with mirrors");
+
+        let reloaded = repo
+            .find_by_id(DownloadId(42))
+            .expect("find_by_id")
+            .expect("download exists");
+        let mirrors = reloaded.mirrors();
+        assert_eq!(mirrors.len(), 2, "both mirrors round-tripped");
+        assert_eq!(mirrors[0].priority(), 80, "highest priority first");
+        assert_eq!(mirrors[0].country(), Some("US"));
+        assert_eq!(mirrors[0].url().host(), "a.example.com");
+        assert_eq!(mirrors[1].priority(), 40);
+        assert!(mirrors[1].country().is_none());
+        assert_eq!(reloaded.current_mirror_index(), 0);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_save_round_trips_current_mirror_index_after_advance() {
+        let db = setup_test_db().await.expect("test db");
+        let repo = SqliteDownloadRepo::new(db);
+
+        let mirrors = vec![
+            crate::domain::model::Mirror::new(
+                Url::new("https://m1.example.com/f").unwrap(),
+                90,
+                None,
+            )
+            .unwrap(),
+            crate::domain::model::Mirror::new(
+                Url::new("https://m2.example.com/f").unwrap(),
+                50,
+                None,
+            )
+            .unwrap(),
+        ];
+        let mut download = make_download(43).with_mirrors(mirrors);
+        download.advance_mirror().expect("advance to slot 1");
+        repo.save(&download).expect("save advanced");
+
+        let reloaded = repo
+            .find_by_id(DownloadId(43))
+            .expect("find")
+            .expect("exists");
+        assert_eq!(reloaded.current_mirror_index(), 1);
+        assert_eq!(reloaded.active_url().host(), "m2.example.com");
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_save_and_find_download_round_trip() {
         let db = setup_test_db().await.expect("test db");
@@ -379,6 +452,8 @@ mod tests {
             account_id: Set(None),
             destination_path: Set("/tmp".to_string()),
             error_message: Set(None),
+            mirrors_json: Set(None),
+            current_mirror_index: Set(0),
             created_at: Set(0),
             updated_at: Set(0),
         };

src-tauri/src/adapters/driven/sqlite/entities/download.rs

@@ -1,10 +1,61 @@
 use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
 
+use crate::adapters::driven::sqlite::util::safe_u32;
 use crate::domain::error::DomainError;
 use crate::domain::model::checksum::ChecksumAlgorithm;
 use crate::domain::model::download::{Download, DownloadId, DownloadState, FileSize, Url};
+use crate::domain::model::mirror::Mirror;
 use crate::domain::model::queue::Priority;
 
+/// Persistence DTO for [`Mirror`]. Lives in the adapter so the domain
+/// stays serde-free per the architecture rule.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct MirrorJsonDto {
+    url: String,
+    priority: u8,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    country: Option<String>,
+}
+
+impl MirrorJsonDto {
+    fn from_domain(m: &Mirror) -> Self {
+        Self {
+            url: m.url().as_str().to_string(),
+            priority: m.priority(),
+            country: m.country().map(|s| s.to_string()),
+        }
+    }
+
+    fn into_domain(self) -> Result<Mirror, DomainError> {
+        let url = Url::new(&self.url)?;
+        Mirror::new(url, self.priority, self.country)
+    }
+}
+
+fn serialize_mirrors(mirrors: &[Mirror]) -> Option<String> {
+    if mirrors.is_empty() {
+        return None;
+    }
+    let dtos: Vec<MirrorJsonDto> = mirrors.iter().map(MirrorJsonDto::from_domain).collect();
+    // The DTO shape is fully owned and validated, so a serialisation
+    // failure here would only be reached through an OOM — surfacing it
+    // as `None` keeps the column nullable instead of poisoning the row.
+    serde_json::to_string(&dtos).ok()
+}
+
+pub(crate) fn deserialize_mirrors(json: Option<&str>) -> Result<Vec<Mirror>, DomainError> {
+    let Some(raw) = json else {
+        return Ok(Vec::new());
+    };
+    if raw.trim().is_empty() {
+        return Ok(Vec::new());
+    }
+    let dtos: Vec<MirrorJsonDto> = serde_json::from_str(raw)
+        .map_err(|e| DomainError::StorageError(format!("invalid mirrors_json: {e}")))?;
+    dtos.into_iter().map(|d| d.into_domain()).collect()
+}
+
 #[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
 #[sea_orm(table_name = "downloads")]
 pub struct Model {
@@ -31,6 +82,16 @@ pub struct Model {
     pub account_id: Option<i64>,
     pub destination_path: String,
     pub error_message: Option<String>,
+    /// `None` keeps storage compact for the common single-source path —
+    /// JSON serialisation is only paid when a Metalink is attached.
+    pub mirrors_json: Option<String>,
+    /// Cursor into the parsed mirror list. Persisted so a future call
+    /// site can mark a failing mirror at the domain level (via
+    /// [`Download::advance_mirror`]) and have the next `start()` resume
+    /// from that slot rather than the always-failing highest-priority
+    /// entry. The current engine still drives failover with its own
+    /// in-task cursor, so a crash mid-failover restarts from slot 0.
+    pub current_mirror_index: i32,
     pub created_at: i64,
     pub updated_at: i64,
 }
@@ -95,6 +156,8 @@ impl Model {
             self.module_name,
             self.account_id.map(|id| id as u64),
             self.destination_path,
+            deserialize_mirrors(self.mirrors_json.as_deref())?,
+            safe_u32(self.current_mirror_index as i64),
             self.created_at as u64,
             self.updated_at as u64,
         ))
@@ -128,6 +191,8 @@ impl ActiveModel {
             account_id: Set(download.account_id().map(|id| id as i64)),
             destination_path: Set(download.destination_path().to_string()),
             error_message: Set(None),
+            mirrors_json: Set(serialize_mirrors(download.mirrors())),
+            current_mirror_index: Set(download.current_mirror_index() as i32),
             created_at: Set(download.created_at() as i64),
             updated_at: Set(download.updated_at() as i64),
         }