feat: publish official resources atomically

Fixes #9

Fixes #11

infrastructure/src/official_update.rs

@@ -31,11 +31,17 @@ use std::io::Write;
 #[cfg(unix)]
 use std::os::unix::fs::OpenOptionsExt;
 use std::path::{Path, PathBuf};
+use std::time::{SystemTime, UNIX_EPOCH};
 
 /// Current official update snapshot schema version.
 pub const OFFICIAL_UPDATE_SNAPSHOT_VERSION: u32 = 2;
 /// Current official bootstrap cache schema version.
 pub const OFFICIAL_BOOTSTRAP_CACHE_VERSION: u32 = 1;
+const OFFICIAL_CURRENT_LINK: &str = "current";
+const OFFICIAL_VERSIONS_DIR: &str = "versions";
+const OFFICIAL_STAGING_DIR: &str = ".staging";
+const OFFICIAL_DOWNLOAD_MANIFEST_FILE: &str = "official-download-manifest.json";
+const OFFICIAL_SYNC_SNAPSHOT_FILE: &str = "official-sync-snapshot.json";
 
 /// Server-info input for an official update run.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -379,6 +385,16 @@ pub struct OfficialUpdateReport {
     pub addressables_root: String,
     /// Platform set used by the run.
     pub platforms: Vec<PatchPlatform>,
+    /// Managed publish root containing `current`, `versions`, and staging.
+    pub output_root: PathBuf,
+    /// Active resource root used for local audit before this run.
+    pub active_resource_root: PathBuf,
+    /// Atomic `current` pointer path.
+    pub current_path: PathBuf,
+    /// Versioned release directory after a successful publish.
+    pub published_version_path: Option<PathBuf>,
+    /// Staging directory used during download before publish.
+    pub staging_path: Option<PathBuf>,
     /// Sync snapshot path.
     pub snapshot_path: PathBuf,
     /// Whether a previous snapshot existed.
@@ -446,6 +462,18 @@ pub struct OfficialUpdateProgress {
     pub stage: &'static str,
     /// Human-readable status line.
     pub message: String,
+    /// One-based download index when the event represents URL download work.
+    pub download_index: Option<usize>,
+    /// Total URL count for the current download plan.
+    pub download_total: Option<usize>,
+    /// Current official URL for download progress.
+    pub download_url: Option<String>,
+    /// Stable pull status label when a URL finished.
+    pub download_status: Option<String>,
+    /// Final local byte count for the URL when known.
+    pub download_bytes: Option<u64>,
+    /// Bytes transferred in this run for the URL when known.
+    pub download_transferred_bytes: Option<u64>,
 }
 
 impl OfficialUpdateProgress {
@@ -454,8 +482,192 @@ impl OfficialUpdateProgress {
         Self {
             stage,
             message: message.into(),
+            download_index: None,
+            download_total: None,
+            download_url: None,
+            download_status: None,
+            download_bytes: None,
+            download_transferred_bytes: None,
         }
     }
+
+    fn with_download_progress(mut self, event: &OfficialResourcePullProgress) -> Self {
+        self.download_index = Some(event.index);
+        self.download_total = Some(event.total);
+        self.download_url = Some(event.url.clone());
+        self.download_status = event.status.map(|status| status.as_str().to_string());
+        self.download_bytes = event.bytes;
+        self.download_transferred_bytes = event.transferred_bytes;
+        self
+    }
+}
+
+#[derive(Debug, Clone)]
+struct OfficialPublishLayout {
+    root: PathBuf,
+    current_path: PathBuf,
+    versions_dir: PathBuf,
+    staging_dir: PathBuf,
+}
+
+#[derive(Debug, Clone)]
+struct OfficialPublishPlan {
+    id: String,
+    staging_path: PathBuf,
+    version_path: PathBuf,
+}
+
+impl OfficialPublishLayout {
+    fn new(root: &Path) -> Self {
+        Self {
+            root: root.to_path_buf(),
+            current_path: root.join(OFFICIAL_CURRENT_LINK),
+            versions_dir: root.join(OFFICIAL_VERSIONS_DIR),
+            staging_dir: root.join(OFFICIAL_STAGING_DIR),
+        }
+    }
+
+    fn active_resource_root(&self) -> Result<PathBuf, String> {
+        if let Some(current_target) = self.current_target()? {
+            return Ok(current_target);
+        }
+
+        // Legacy fallback for directories produced before atomic publishing.
+        // The next non-dry-run update will seed staging from this tree and
+        // publish it under `versions/<id>` before switching `current`.
+        Ok(self.root.clone())
+    }
+
+    fn current_target(&self) -> Result<Option<PathBuf>, String> {
+        let metadata = match fs::symlink_metadata(&self.current_path) {
+            Ok(metadata) => metadata,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(error) => {
+                return Err(format!(
+                    "读取 current 指针失败 {}：{error}",
+                    self.current_path.display()
+                ))
+            }
+        };
+
+        if !metadata.file_type().is_symlink() {
+            return Err(format!(
+                "current 必须是指向 versioned 目录的 symlink：{}",
+                self.current_path.display()
+            ));
+        }
+
+        let target = fs::read_link(&self.current_path).map_err(|error| {
+            format!(
+                "读取 current symlink 目标失败 {}：{error}",
+                self.current_path.display()
+            )
+        })?;
+        let target = if target.is_absolute() {
+            target
+        } else {
+            self.root.join(target)
+        };
+        ensure_path_within_root(&self.root, &target)?;
+        ensure_safe_directory_path(&target, "current versioned 目录")?;
+        Ok(Some(target))
+    }
+
+    fn has_current_pointer(&self) -> Result<bool, String> {
+        Ok(self.current_target()?.is_some())
+    }
+
+    fn plan(&self, snapshot: &OfficialUpdateSnapshot) -> OfficialPublishPlan {
+        let id = publish_id(snapshot);
+        OfficialPublishPlan {
+            staging_path: self.staging_dir.join(&id),
+            version_path: self.versions_dir.join(&id),
+            id,
+        }
+    }
+
+    fn prepare_staging(&self, plan: &OfficialPublishPlan) -> Result<(), String> {
+        validate_output_root(&self.root)?;
+        ensure_safe_directory_path(&self.staging_dir, "官方资源 staging 根目录")?;
+        fs::create_dir_all(&self.staging_dir).map_err(|error| {
+            format!(
+                "创建官方资源 staging 根目录失败 {}：{error}",
+                self.staging_dir.display()
+            )
+        })?;
+        ensure_safe_directory_path(&self.staging_dir, "官方资源 staging 根目录")?;
+
+        if path_exists_no_follow(&plan.staging_path)? {
+            ensure_safe_directory_path(&plan.staging_path, "官方资源 staging 目录")?;
+            fs::remove_dir_all(&plan.staging_path).map_err(|error| {
+                format!(
+                    "清理官方资源 staging 目录失败 {}：{error}",
+                    plan.staging_path.display()
+                )
+            })?;
+        }
+
+        if path_exists_no_follow(&plan.version_path)? {
+            return Err(format!(
+                "versioned 目录已存在，拒绝覆盖：{}",
+                plan.version_path.display()
+            ));
+        }
+
+        fs::create_dir_all(&plan.staging_path).map_err(|error| {
+            format!(
+                "创建官方资源 staging 目录失败 {}：{error}",
+                plan.staging_path.display()
+            )
+        })?;
+        ensure_safe_directory_path(&plan.staging_path, "官方资源 staging 目录")
+    }
+
+    fn seed_staging_from_active(
+        &self,
+        active_root: &Path,
+        staging_root: &Path,
+    ) -> Result<(), String> {
+        if active_root == self.root && !self.legacy_manifest_exists()? {
+            return Ok(());
+        }
+        if !path_exists_no_follow(active_root)? {
+            return Ok(());
+        }
+        copy_tree_no_symlink(active_root, staging_root, active_root == self.root)
+    }
+
+    fn legacy_manifest_exists(&self) -> Result<bool, String> {
+        path_exists_no_follow(&self.root.join(OFFICIAL_DOWNLOAD_MANIFEST_FILE))
+    }
+
+    fn publish(&self, plan: &OfficialPublishPlan) -> Result<PathBuf, String> {
+        ensure_safe_directory_path(&self.versions_dir, "官方资源 versions 根目录")?;
+        fs::create_dir_all(&self.versions_dir).map_err(|error| {
+            format!(
+                "创建官方资源 versions 根目录失败 {}：{error}",
+                self.versions_dir.display()
+            )
+        })?;
+        ensure_safe_directory_path(&self.versions_dir, "官方资源 versions 根目录")?;
+        ensure_safe_directory_path(&plan.staging_path, "官方资源 staging 目录")?;
+        if path_exists_no_follow(&plan.version_path)? {
+            return Err(format!(
+                "versioned 目录已存在，拒绝覆盖：{}",
+                plan.version_path.display()
+            ));
+        }
+
+        fs::rename(&plan.staging_path, &plan.version_path).map_err(|error| {
+            format!(
+                "发布官方资源版本目录失败 {} -> {}：{error}",
+                plan.staging_path.display(),
+                plan.version_path.display()
+            )
+        })?;
+        switch_current_symlink(&self.root, &self.current_path, &plan.id)?;
+        Ok(plan.version_path.clone())
+    }
 }
 
 /// Official update runner.
@@ -517,16 +729,24 @@ impl OfficialUpdateService {
         };
         check_shutdown_requested(&mut should_cancel)?;
 
+        let publish_layout = OfficialPublishLayout::new(&config.output_root);
+        let active_resource_root = publish_layout
+            .active_resource_root()
+            .map_err(anyhow::Error::msg)?;
+        let has_current_pointer = publish_layout
+            .has_current_pointer()
+            .map_err(anyhow::Error::msg)?;
+
         let default_platforms = default_official_platforms();
         let platforms = config
             .platforms
             .as_deref()
             .unwrap_or(default_platforms.as_slice());
         let fetcher = OfficialResourcePullService::with_curl_command(
-            &config.output_root,
+            &active_resource_root,
             &config.curl_command,
         );
-        let snapshot_path = config.effective_snapshot_path();
+        let snapshot_path = snapshot_path_for(config, &active_resource_root);
         let bootstrap_cache_path = config.bootstrap_cache_path();
 
         let bootstrap = if config.auto_discover {
@@ -750,7 +970,9 @@ impl OfficialUpdateService {
                 .map(|audit| !audit.is_clean())
                 .unwrap_or(false);
         let initial_pull_needed = !has_local_resources;
-        let should_download = remote_should_download || repair_needed || initial_pull_needed;
+        let publish_required = !has_current_pointer;
+        let should_download =
+            remote_should_download || repair_needed || initial_pull_needed || publish_required;
         progress(OfficialUpdateProgress::new(
             "audit",
             format!(
@@ -765,6 +987,12 @@ impl OfficialUpdateService {
                 should_download, repair_needed, initial_pull_needed
             ),
         ));
+        if publish_required {
+            progress(OfficialUpdateProgress::new(
+                "publish",
+                "尚未存在 current 原子发布指针；本轮会发布 versioned 目录并切换 current",
+            ));
+        }
         check_shutdown_requested(&mut should_cancel)?;
         let mut report = OfficialUpdateReport {
             update_status: if should_download {
@@ -777,6 +1005,11 @@ impl OfficialUpdateService {
             bundle_version: current_snapshot.bundle_version.clone(),
             addressables_root: current_snapshot.addressables_root.clone(),
             platforms: platforms.to_vec(),
+            output_root: config.output_root.clone(),
+            active_resource_root: active_resource_root.clone(),
+            current_path: publish_layout.current_path.clone(),
+            published_version_path: None,
+            staging_path: None,
             snapshot_path: snapshot_path.clone(),
             previous_snapshot_present: previous_snapshot.is_some(),
             decision: format!("{:?}", sync_plan.decision),
@@ -850,11 +1083,35 @@ impl OfficialUpdateService {
         check_shutdown_requested(&mut should_cancel)?;
 
         let download_url_count = pull_plan.all_urls().map_err(anyhow::Error::msg)?.len();
+        let publish_plan = publish_layout.plan(&current_update_snapshot);
+        progress(OfficialUpdateProgress::new(
+            "publish",
+            format!(
+                "准备 staging={} versioned={}",
+                publish_plan.staging_path.display(),
+                publish_plan.version_path.display()
+            ),
+        ));
+        publish_layout
+            .prepare_staging(&publish_plan)
+            .map_err(anyhow::Error::msg)?;
+        publish_layout
+            .seed_staging_from_active(&active_resource_root, &publish_plan.staging_path)
+            .map_err(anyhow::Error::msg)?;
+        let staging_fetcher = OfficialResourcePullService::with_curl_command(
+            &publish_plan.staging_path,
+            &config.curl_command,
+        );
+        let staging_snapshot_path = snapshot_path_for(config, &publish_plan.staging_path);
+        report.staging_path = Some(publish_plan.staging_path.clone());
+        report.snapshot_path = staging_snapshot_path.clone();
+        report.download_manifest = staging_fetcher.download_manifest_path();
+
         progress(OfficialUpdateProgress::new(
             "download",
             format!("下载或复用 {download_url_count} 个官方 URL"),
         ));
-        let pull_report = fetcher
+        let pull_report = staging_fetcher
             .pull_with_progress_and_cancellation(
                 &pull_plan,
                 |event| {
@@ -878,16 +1135,37 @@ impl OfficialUpdateService {
             "执行最终本地 manifest 审计",
         ));
         check_shutdown_requested(&mut should_cancel)?;
-        let final_audit = fetcher
+        let final_audit = staging_fetcher
             .audit_local_manifest(&pull_plan)
             .map_err(anyhow::Error::msg)?;
         progress(OfficialUpdateProgress::new(
             "snapshot",
-            format!("写入快照 {}", snapshot_path.display()),
+            format!("写入快照 {}", staging_snapshot_path.display()),
         ));
-        write_snapshot(&snapshot_path, &current_update_snapshot)?;
+        if config.snapshot_path.is_none() {
+            write_snapshot(&staging_snapshot_path, &current_update_snapshot)?;
+        }
+
+        progress(OfficialUpdateProgress::new(
+            "publish",
+            format!(
+                "校验完成，发布 versioned 目录并切换 current -> {}",
+                publish_plan.version_path.display()
+            ),
+        ));
+        let published_version_path = publish_layout
+            .publish(&publish_plan)
+            .map_err(anyhow::Error::msg)?;
+        let final_snapshot_path = snapshot_path_for(config, &published_version_path);
+        if config.snapshot_path.is_some() {
+            write_snapshot(&final_snapshot_path, &current_update_snapshot)?;
+        }
 
         report.update_status = OfficialUpdateStatus::Downloaded;
+        report.active_resource_root = published_version_path.clone();
+        report.published_version_path = Some(published_version_path.clone());
+        report.snapshot_path = final_snapshot_path.clone();
+        report.download_manifest = published_version_path.join(OFFICIAL_DOWNLOAD_MANIFEST_FILE);
         report.resource_count = Some(pull_report.items.len());
         report.downloaded_count = pull_report.downloaded_count();
         report.resumed_count = pull_report.resumed_count();
@@ -903,7 +1181,7 @@ impl OfficialUpdateService {
             pull_report.official_hash_verified_count(),
             final_audit.zip_structure_verified_count(),
         );
-        report.snapshot_written = Some(snapshot_path);
+        report.snapshot_written = Some(final_snapshot_path);
 
         progress(OfficialUpdateProgress::new(
             "finish",
@@ -1079,7 +1357,8 @@ fn progress_from_pull_event(event: OfficialResourcePullProgress) -> OfficialUpda
         OfficialResourcePullProgressKind::Started => OfficialUpdateProgress::new(
             "download",
             format!("({}/{}) 开始 {}", event.index, event.total, event.url),
-        ),
+        )
+        .with_download_progress(&event),
         OfficialResourcePullProgressKind::Finished => {
             let status = event.status.map(localized_pull_status).unwrap_or("未知");
             OfficialUpdateProgress::new(
@@ -1094,6 +1373,7 @@ fn progress_from_pull_event(event: OfficialResourcePullProgress) -> OfficialUpda
                     event.url
                 ),
             )
+            .with_download_progress(&event)
         }
     }
 }
@@ -1143,6 +1423,195 @@ fn validate_update_paths(config: &OfficialUpdateConfig) -> Result<(), String> {
     Ok(())
 }
 
+fn snapshot_path_for(config: &OfficialUpdateConfig, resource_root: &Path) -> PathBuf {
+    config
+        .snapshot_path
+        .clone()
+        .unwrap_or_else(|| resource_root.join(OFFICIAL_SYNC_SNAPSHOT_FILE))
+}
+
+fn path_exists_no_follow(path: &Path) -> Result<bool, String> {
+    match fs::symlink_metadata(path) {
+        Ok(metadata) if metadata.file_type().is_symlink() => {
+            Err(format!("路径不能是 symlink：{}", path.display()))
+        }
+        Ok(_) => Ok(true),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
+        Err(error) => Err(format!("检查路径失败 {}：{error}", path.display())),
+    }
+}
+
+fn copy_tree_no_symlink(
+    source: &Path,
+    destination: &Path,
+    skip_publish_management_entries: bool,
+) -> Result<(), String> {
+    ensure_safe_directory_path(source, "官方资源发布源目录")?;
+    ensure_safe_directory_path(destination, "官方资源 staging 目录")?;
+    fs::create_dir_all(destination).map_err(|error| {
+        format!(
+            "创建官方资源 staging 子目录失败 {}：{error}",
+            destination.display()
+        )
+    })?;
+
+    for entry in fs::read_dir(source)
+        .map_err(|error| format!("读取官方资源发布源目录失败 {}：{error}", source.display()))?
+    {
+        let entry = entry.map_err(|error| {
+            format!("读取官方资源发布源目录项失败 {}：{error}", source.display())
+        })?;
+        let source_path = entry.path();
+        let file_name = entry.file_name();
+        if skip_publish_management_entries {
+            if let Some(name) = file_name.to_str() {
+                if matches!(
+                    name,
+                    OFFICIAL_STAGING_DIR
+                        | OFFICIAL_VERSIONS_DIR
+                        | OFFICIAL_CURRENT_LINK
+                        | ".official-sync.lock"
+                ) {
+                    continue;
+                }
+            }
+        }
+        if destination.starts_with(&source_path) {
+            continue;
+        }
+        let destination_path = destination.join(file_name);
+        let metadata = fs::symlink_metadata(&source_path).map_err(|error| {
+            format!(
+                "读取官方资源发布源元数据失败 {}：{error}",
+                source_path.display()
+            )
+        })?;
+        if metadata.file_type().is_symlink() {
+            return Err(format!(
+                "官方资源发布源不能包含 symlink：{}",
+                source_path.display()
+            ));
+        }
+        if metadata.is_dir() {
+            copy_tree_no_symlink(&source_path, &destination_path, false)?;
+        } else if metadata.is_file() {
+            if let Some(name) = source_path.file_name().and_then(|value| value.to_str()) {
+                if name.ends_with(".part") || name.ends_with(".tmp") {
+                    continue;
+                }
+            }
+            if let Err(_error) = fs::hard_link(&source_path, &destination_path) {
+                fs::copy(&source_path, &destination_path).map_err(|copy_error| {
+                    format!(
+                        "复制官方资源到 staging 失败 {} -> {}：{copy_error}",
+                        source_path.display(),
+                        destination_path.display()
+                    )
+                })?;
+            }
+        } else {
+            return Err(format!(
+                "官方资源发布源包含非普通文件：{}",
+                source_path.display()
+            ));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn switch_current_symlink(
+    root: &Path,
+    current_path: &Path,
+    publish_id: &str,
+) -> Result<(), String> {
+    use std::os::unix::fs::symlink;
+
+    let temporary = root.join(format!(".current.{}.tmp", std::process::id()));
+    if path_exists_no_follow(&temporary)? {
+        fs::remove_file(&temporary).map_err(|error| {
+            format!("清理临时 current 指针失败 {}：{error}", temporary.display())
+        })?;
+    }
+    let relative_target = Path::new(OFFICIAL_VERSIONS_DIR).join(publish_id);
+    symlink(&relative_target, &temporary).map_err(|error| {
+        format!(
+            "创建临时 current 指针失败 {} -> {}：{error}",
+            temporary.display(),
+            relative_target.display()
+        )
+    })?;
+
+    if let Ok(metadata) = fs::symlink_metadata(current_path) {
+        if !metadata.file_type().is_symlink() {
+            let _ = fs::remove_file(&temporary);
+            return Err(format!(
+                "current 已存在但不是 symlink：{}",
+                current_path.display()
+            ));
+        }
+    }
+
+    fs::rename(&temporary, current_path).map_err(|error| {
+        format!(
+            "切换 current 指针失败 {} -> {}：{error}",
+            temporary.display(),
+            current_path.display()
+        )
+    })
+}
+
+#[cfg(not(unix))]
+fn switch_current_symlink(
+    _root: &Path,
+    _current_path: &Path,
+    _publish_id: &str,
+) -> Result<(), String> {
+    Err("原子 current symlink 发布目前只支持 Unix/Linux 平台".to_string())
+}
+
+fn publish_id(snapshot: &OfficialUpdateSnapshot) -> String {
+    let bundle = snapshot
+        .bundle_version
+        .as_deref()
+        .map(sanitize_publish_segment)
+        .unwrap_or_else(|| "no-bundle".to_string());
+    format!(
+        "{}-{}-{}-{}-{}",
+        sanitize_publish_segment(&snapshot.app_version),
+        bundle,
+        snapshot.addressables_marker_checked_count(),
+        unix_seconds_now(),
+        std::process::id()
+    )
+}
+
+fn sanitize_publish_segment(value: &str) -> String {
+    let sanitized = value
+        .chars()
+        .map(|ch| {
+            if ch.is_control() || matches!(ch, '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|')
+            {
+                '_'
+            } else {
+                ch
+            }
+        })
+        .collect::<String>();
+    if sanitized.is_empty() {
+        "unknown".to_string()
+    } else {
+        sanitized
+    }
+}
+
+fn unix_seconds_now() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
 /// Reads an official update snapshot from disk, accepting legacy v1 snapshots.
 pub fn read_snapshot(path: &Path) -> anyhow::Result<Option<OfficialUpdateSnapshot>> {
     let Some(data) = read_file_no_symlink(path, "官方更新快照").map_err(anyhow::Error::msg)?