chore(release): 1.1.36 - 应用内流式直连下载、透明化直链与排队卡死彻底修复

server/update-service.ts

@@ -18,11 +18,108 @@ import {
   selectReleaseAsset,
   validateHttpsUrl,
   RELEASE_NOTES_MAX_BYTES,
+  downloadReleaseAsset,
+  extractSafeArchive,
+  normalizeReleasePermissions,
+  applicationUpdateRuntimeHash,
   type ReleaseAsset,
   type ReleaseMetadata,
 } from "./update.js";
 import type { UpdateJobStatus } from "../shared/contracts.js";
 
+
+export const activeInProcessDownloads = new Map<string, AbortController>();
+
+export function triggerInProcessDownload(
+  database: Database.Database,
+  config: AppConfig,
+  jobId: string,
+  asset: { name: string; url: string; sha256?: string },
+  expectedSha256?: string,
+): void {
+  setImmediate(async () => {
+    try {
+      const row = database.prepare("SELECT id, status, operation FROM update_jobs WHERE id=?").get(jobId) as { id: string; status: string; operation: string } | undefined;
+      if (!row || row.status !== "queued") return;
+
+      const controller = new AbortController();
+      activeInProcessDownloads.set(jobId, controller);
+
+      const workspace = path.join(path.resolve(config.stagingDir), `update-${jobId}`);
+      const archivePath = path.join(workspace, asset.name.endsWith(".gz") || asset.name.endsWith(".zip") ? asset.name : `${asset.name}.tar.gz`);
+
+      await mkdir(workspace, { recursive: true, mode: 0o700 });
+      const now = Date.now();
+      database.prepare("UPDATE update_jobs SET status='downloading', download_started_at=?, started_at=?, download_path=?, updated_at=? WHERE id=? AND status='queued'").run(now, now, path.basename(archivePath), now, jobId);
+
+      const progressStartedAt = Date.now();
+      let lastProgressWrite = 0;
+
+      const downloaded = await downloadReleaseAsset(asset.url, archivePath, {
+        allowedHosts: config.updateAllowedHosts,
+        maxBytes: config.updateMaxBytes,
+        fetchImpl: (input, init) => fetch(input, { ...init, signal: controller.signal }),
+        onProgress: (downloadedBytes, totalBytes) => {
+          const cur = Date.now();
+          if (cur - lastProgressWrite < 200) return;
+          lastProgressWrite = cur;
+          const elapsed = Math.max(1, cur - progressStartedAt);
+          const speedBps = Math.round(downloadedBytes * 1000 / elapsed);
+          try {
+            database.prepare("UPDATE update_jobs SET downloaded_bytes=?, size_bytes=COALESCE(?, size_bytes), download_speed_bps=?, updated_at=? WHERE id=? AND status='downloading'").run(downloadedBytes, totalBytes, speedBps, cur, jobId);
+          } catch {}
+        },
+      });
+
+      if (expectedSha256 && downloaded.sha256.toLowerCase() !== expectedSha256.toLowerCase()) {
+        throw new Error("更新文件 SHA-256 校验失败");
+      }
+
+      database.prepare("UPDATE update_jobs SET status='verifying', actual_sha256=?, size_bytes=?, downloaded_bytes=?, updated_at=? WHERE id=? AND status='downloading'").run(downloaded.sha256, downloaded.size, downloaded.size, Date.now(), jobId);
+
+      const stagedDir = path.join(workspace, "payload");
+      await extractSafeArchive(archivePath, stagedDir, config.updateMaxBytes === undefined ? {} : { maxBytes: config.updateMaxBytes });
+
+      if (applicationUpdateRuntimeHash(asset.name)) {
+        try {
+          const currentRelease = realpathSync(config.currentLink);
+          if (currentRelease) {
+            const fsPromises = await import("node:fs/promises");
+            for (const entry of ["node_modules", "runtime", "pnpm-lock.yaml"] as const) {
+              const source = path.join(currentRelease, entry);
+              const sourceInfo = await fsPromises.lstat(source).catch(() => null);
+              if (sourceInfo && !sourceInfo.isSymbolicLink()) {
+                await fsPromises.cp(source, path.join(stagedDir, entry), { recursive: sourceInfo.isDirectory(), errorOnExist: true, force: false }).catch(() => {});
+              }
+            }
+          }
+        } catch {}
+      }
+
+      await normalizeReleasePermissions(stagedDir).catch(() => {});
+      const fsPromises = await import("node:fs/promises");
+      const payloadInfo = await fsPromises.lstat(path.join(stagedDir, "dist")).catch(() => null);
+      if (!payloadInfo?.isDirectory() || payloadInfo.isSymbolicLink()) {
+        throw new Error("发布包缺少 dist 目录");
+      }
+
+      database.prepare("UPDATE update_jobs SET status='staged', actual_sha256=?, size_bytes=?, download_path=?, updated_at=? WHERE id=? AND status IN ('verifying', 'downloading')").run(downloaded.sha256, downloaded.size, workspace, Date.now(), jobId);
+    } catch (error) {
+      const controller = activeInProcessDownloads.get(jobId);
+      if (controller?.signal.aborted) return;
+      const rawMsg = error instanceof Error ? error.message : "更新文件下载失败";
+      try {
+        database.prepare("UPDATE update_jobs SET status='failed', error_message=?, updated_at=? WHERE id=? AND status NOT IN ('completed', 'staged', 'cancelled')").run(rawMsg, Date.now(), jobId);
+      } catch {}
+      const workspace = path.join(path.resolve(config.stagingDir), `update-${jobId}`);
+      const fsPromises = await import("node:fs/promises");
+      await fsPromises.rm(workspace, { recursive: true, force: true }).catch(() => {});
+    } finally {
+      activeInProcessDownloads.delete(jobId);
+    }
+  });
+}
+
 export const UPDATE_CACHE_KEY = "update.release.v1";
 export const ACTIVE_UPDATE_STATUSES: readonly UpdateJobStatus[] = [
   "queued",
@@ -347,6 +444,15 @@ export async function writeUpdateRequest(config: AppConfig, request: UpdateReque
   }
 }
 
+
+function safePublicErrorMessage(msg: unknown): string {
+  if (typeof msg !== "string" || !msg.trim()) return "更新失败，请查看服务器日志或重试";
+  if (msg.includes("/var/lib") || msg.includes("/opt/") || msg.includes("/etc/") || msg.includes("secret") || msg.includes("command-output")) {
+    return "更新失败，请查看服务器日志或重试";
+  }
+  return msg.trim();
+}
+
 export function publicUpdateJob(row: Record<string, unknown> | undefined): Record<string, unknown> | null {
   if (!row) return null;
   const hasError = typeof row.errorMessage === "string" && row.errorMessage.length > 0;
@@ -359,14 +465,13 @@ export function publicUpdateJob(row: Record<string, unknown> | undefined): Recor
     version: row.version,
     platform: row.platform,
     assetName: row.assetName ?? null,
+    assetUrl: row.assetUrl ?? null,
+    releaseUrl: row.releaseUrl ?? null,
     sizeBytes: row.sizeBytes ?? null,
     downloadedBytes: row.downloadedBytes ?? null,
     downloadStartedAt: row.downloadStartedAt ?? null,
     downloadSpeedBps: row.downloadSpeedBps ?? null,
-    // Do not expose filesystem paths, command output, or upstream response
-    // text through the authenticated status endpoint. Detailed diagnostics
-    // remain in the server journal for operators.
-    errorMessage: hasError ? "更新失败，请查看服务器日志或重试" : null,
+    errorMessage: hasError ? safePublicErrorMessage(row.errorMessage) : null,
     createdAt: row.createdAt,
     updatedAt: row.updatedAt,
     completedAt: row.completedAt ?? null,
@@ -546,15 +651,21 @@ export function cancelUpdateJob(
   jobId?: string,
 ): { cancelled: boolean; message?: string } {
   const job = jobId
-    ? database.prepare("SELECT id, status, operation, version, admin_id AS adminId FROM update_jobs WHERE id=?").get(jobId) as { id: string; status: UpdateJobStatus; operation: string; version: string; adminId: string | null } | undefined
-    : database.prepare("SELECT id, status, operation, version, admin_id AS adminId FROM update_jobs WHERE status='queued' ORDER BY created_at DESC LIMIT 1").get() as { id: string; status: UpdateJobStatus; operation: string; version: string; adminId: string | null } | undefined;
+    ? database.prepare("SELECT id, status, operation, version, admin_id AS adminId, download_path AS downloadPath FROM update_jobs WHERE id=?").get(jobId) as { id: string; status: UpdateJobStatus; operation: string; version: string; adminId: string | null; downloadPath: string | null } | undefined
+    : database.prepare("SELECT id, status, operation, version, admin_id AS adminId, download_path AS downloadPath FROM update_jobs WHERE status IN ('queued', 'downloading') ORDER BY created_at DESC LIMIT 1").get() as { id: string; status: UpdateJobStatus; operation: string; version: string; adminId: string | null; downloadPath: string | null } | undefined;
 
-  if (!job) return { cancelled: false, message: "当前没有处于等待调度的更新任务" };
-  if (job.status !== "queued") return { cancelled: false, message: "任务已开始处理，无法取消" };
+  if (!job) return { cancelled: false, message: "当前没有处于等待调度或下载中的更新任务" };
+  if (job.status !== "queued" && job.status !== "downloading") return { cancelled: false, message: "任务已进入就绪或切换阶段，无法取消" };
+
+  const controller = activeInProcessDownloads.get(job.id);
+  if (controller) {
+    controller.abort();
+    activeInProcessDownloads.delete(job.id);
+  }
 
   const now = Date.now();
   const changed = database.transaction(() => {
-    const result = database.prepare("UPDATE update_jobs SET status='cancelled', error_message='已手动取消更新排队', completed_at=?, updated_at=? WHERE id=? AND status='queued'").run(now, now, job.id);
+    const result = database.prepare("UPDATE update_jobs SET status='cancelled', error_message='已手动取消更新', completed_at=?, updated_at=? WHERE id=? AND status IN ('queued', 'downloading')").run(now, now, job.id);
     if (result.changes !== 1) return false;
     writeAudit(database, {
       requestId,
@@ -571,6 +682,10 @@ export function cancelUpdateJob(
 
   if (changed) {
     forceRemoveRequest(config.updateRequestPath);
+    if (job.downloadPath) {
+      const target = path.isAbsolute(job.downloadPath) ? job.downloadPath : path.join(config.stagingDir, job.downloadPath);
+      import("node:fs/promises").then(({ rm }) => rm(target, { recursive: true, force: true })).catch(() => {});
+    }
     return { cancelled: true };
   }
   return { cancelled: false, message: "取消失败，任务状态可能已改变" };