verdaccio-cache-manager/backend/sync-latest.js

/**
 * 存量包最新版本拉取 —— 一键把私服缓存中已有的包更新到上游最新版
 *
 * 实现原理：
 * 1. 查询：请求 Verdaccio 私服上该包的元数据（GET /:pkg），取 dist-tags.latest；
 *    Verdaccio 按自身 uplinks 配置（npm / 淘宝 / 华为等多源）路由到上游获取；
 * 2. 比对：本地 storage 已缓存该精确版本则跳过（不重复拉取）；
 * 3. 拉取：未缓存的版本请求 Verdaccio HTTP 端点（GET /:pkg/-/:file.tgz），
 *    Verdaccio 作为代理会从 uplinks 拉取并把 tgz + 元数据写入自身存储，
 *    元数据由 Verdaccio 自己维护，格式与扫描器读取的完全一致。
 *
 * 环境变量：
 *   VERDACCIO_BASE_URL     私服地址（默认 http://127.0.0.1:4873；查询与代拉均走它，上游由 Verdaccio 自行路由）
 */
import { scanStorageAsync, resolveStorageDir } from './scanner.js'

const VERDACCIO_BASE = (process.env.VERDACCIO_BASE_URL || 'http://127.0.0.1:4873').replace(/\/+$/, '')
const QUERY_TIMEOUT = 15000 // 元数据查询超时（Verdaccio 本地/uplink 转发通常较快）
const FETCH_TIMEOUT = 10 * 60_000 // tgz 拉取超时：大包（echarts 等）需要时间
const CONCURRENT = 3 // 同时拉取的包数：网络 IO 温和并发，避免打满 NAS 带宽

// —— 拉取状态（前端轮询进度，与 scanState 模式对称） ——
export const syncState = {
  running: false, done: false,
  total: 0, processed: 0,
  success: 0, skipped: 0, failed: 0,
  failedList: [], // [{ name, reason }]
  current: null, // 正在处理的包名
  error: null,
  startedAt: null, finishedAt: null,
}

function resetState() {
  syncState.done = false
  syncState.total = 0
  syncState.processed = 0
  syncState.success = 0
  syncState.skipped = 0
  syncState.failed = 0
  syncState.failedList = []
  syncState.current = null
  syncState.error = null
  syncState.startedAt = new Date().toISOString()
}

// 编码包名：scoped 包 @scope/name → %40scope%2fname（registry / verdaccio 均接受该编码）
const enc = name => encodeURIComponent(name)
// tarball 文件名：@scope/name → name-version.tgz（Verdaccio 落盘命名不携带 scope 前缀）
const fileOf = (name, version) => `${name.split('/').pop()}-${version}.tgz`

/** GET JSON，带超时；404 返回 null（包不存在），其余非 2xx 抛可读错误 */
async function fetchJson(url, timeout) {
  const ctrl = new AbortController()
  const timer = setTimeout(() => ctrl.abort(), timeout)
  try {
    const r = await fetch(url, { signal: ctrl.signal, headers: { Accept: 'application/json' } })
    if (r.status === 404) return null
    if (!r.ok) throw new Error(`私服响应 ${r.status}`)
    return await r.json()
  } catch (e) {
    if (e && e.name === 'AbortError') throw new Error('请求超时')
    throw e
  } finally {
    clearTimeout(timer)
  }
}

/** 触发 Verdaccio 代理拉取：走到该端点时 Verdaccio 会从 uplink 拉取并落盘缓存 */
async function triggerVerdaccio(name, version) {
  const url = `${VERDACCIO_BASE}/${enc(name)}/-/${enc(fileOf(name, version))}`
  const ctrl = new AbortController()
  const timer = setTimeout(() => ctrl.abort(), FETCH_TIMEOUT)
  try {
    const r = await fetch(url, { signal: ctrl.signal, headers: { accept: 'application/octet-stream' } })
    if (!r.ok) throw new Error(`私服响应 ${r.status}（请确认 VERDACCIO_BASE_URL 配置正确）`)
    if (r.body) {
      // 丢弃响应体：Verdaccio 已把包写进存储，本次只需等待传输完成
      for await (const _ of r.body) { /* drain */ }
    }
  } catch (e) {
    if (e && e.name === 'AbortError') throw new Error('拉取超时')
    throw e
  } finally {
    clearTimeout(timer)
  }
}

/** 本地是否已缓存该精确版本（versionsDetail 缺失时保守视为未缓存，交给 Verdaccio 层幂等判断） */
function hasLocalVersion(pkg, version) {
  const vs = pkg && pkg.versionsDetail
  if (!Array.isArray(vs)) return false
  return vs.some(v => String(v.version) === version)
}

async function handleOne(pkg) {
  try {
    const meta = await fetchJson(`${VERDACCIO_BASE}/${enc(pkg.name)}`, QUERY_TIMEOUT)
    const latest = meta && meta['dist-tags'] && meta['dist-tags'].latest
    if (!meta) {
      syncState.failed += 1
      syncState.failedList.push({ name: pkg.name, reason: '私服无此包（404）' })
      return
    }
    if (!latest) {
      syncState.failed += 1
      syncState.failedList.push({ name: pkg.name, reason: '私服未提供 latest 版本' })
      return
    }
    if (hasLocalVersion(pkg, latest)) {
      syncState.skipped += 1 // 已是最新，不重复拉取
      return
    }
    await triggerVerdaccio(pkg.name, latest)
    syncState.success += 1
  } catch (e) {
    syncState.failed += 1
    syncState.failedList.push({ name: pkg.name, reason: (e && e.message) || '拉取失败' })
  }
}

/**
 * 启动拉取任务（异步；running 时重复调用直接返回）
 * @param {{ addLog: Function, invalidateScan: Function }} services 依赖注入（避免与 server.js 循环引用）
 */
export async function runSyncLatest({ addLog, invalidateScan }) {
  if (syncState.running) return
  resetState()
  syncState.running = true
  console.log(`[sync-latest] 开始拉取存量包最新版本（经私服: ${VERDACCIO_BASE}，上游按 Verdaccio uplinks 路由）`)
  try {
    // 存量包列表：优先直接扫描 storage（与面板列表同源，且不依赖接口层扫描缓存是否就绪）
    let pkgs = []
    try {
      pkgs = await scanStorageAsync(resolveStorageDir())
    } catch (e) {
      syncState.error = `扫描缓存目录失败：${e.message}`
    }
    // 跳过异常包（元数据读取失败无法判断版本），其余全部纳入（含忽略中的包，它们同样是存量缓存）
    pkgs = (pkgs || []).filter(p => p && p.name && p.status !== 'broken')
    syncState.total = pkgs.length
    if (syncState.total === 0) {
      if (!syncState.error) syncState.error = '缓存目录为空，无可拉取的包'
    } else {
      let next = 0
      const worker = async () => {
        while (next < pkgs.length) {
          const pkg = pkgs[next++]
          syncState.current = pkg.name
          await handleOne(pkg)
          syncState.processed += 1
        }
      }
      await Promise.all(Array.from({ length: Math.min(CONCURRENT, pkgs.length) }, worker))
    }
    // 汇总写操作日志（失败包名列表存 deletedVersions，前端日志详情展示）
    addLog({
      action: 'sync',
      pkgName: null,
      deletedCount: syncState.success,
      deletedVersions: JSON.stringify({
        success: syncState.success,
        skipped: syncState.skipped,
        failed: syncState.failedList,
      }),
    })
    invalidateScan() // 有新版本入库 → 废弃扫描缓存，列表下次请求重扫
    console.log(`[sync-latest] 完成：成功 ${syncState.success} · 跳过 ${syncState.skipped} · 失败 ${syncState.failed}`)
  } catch (e) {
    syncState.error = e.message
    console.error('[sync-latest] 任务异常:', e.message)
  } finally {
    syncState.running = false
    syncState.done = true
    syncState.finishedAt = new Date().toISOString()
  }
}