verdaccio-cache-manager/backend/pkglist-pull.js

/**
 * 按 npm 包清单拉取 —— 以私服 VERDACCIO_BASE_URL 为镜像源（上游按 Verdaccio uplinks 路由）
 *
 * 与 sync-latest.js 的差异：
 *   sync-latest：存量包查最新版（跳过已缓存版本）
 *   本模块：按清单逐包拉取（清单即待拉取目标；版本为空时解析 dist-tags.latest）
 *
 * 拉取流程（异步状态机，前端轮询进度）：
 *   1. 取清单全部 status='pending' 条目作为目标；
 *   2. 首轮拉取全部；
 *   3. 失败的包自动连续重试 3 次（round 1..3），每轮只处理上一轮失败项；
 *   4. 重试 3 次后仍失败的条目保持在 failedList（状态仍 pending，供角标展示与下次重拉），
 *      其余成功条目置 status='done'；
 *   5. 全部成功后清单失效：pending 为 0 → 手动拉取提示「当前无可用npm包」、
 *      定时拉取自动停止，直至清单写入新包。
 *
 * 环境变量：
 *   VERDACCIO_BASE_URL  私服地址（默认为 sync-latest 同款缺省值；查询与代拉均走它）
 */
import { getPendingRows, markDone, countPending, getPkgList } from './pkglist.js'

const VERDACCIO_BASE = (process.env.VERDACCIO_BASE_URL || 'http://127.0.0.1:4873').replace(/\/+$/, '')
const QUERY_TIMEOUT = 15000 // 元数据查询超时
const FETCH_TIMEOUT = 10 * 60_000 // tgz 拉取超时（大包需要时间）
const CONCURRENT = 3 // 同时拉取的包数
const MAX_RETRY = 3 // 失败自动连续重试次数

// —— 拉取状态（前端轮询进度；与 syncState 对称，另含轮次信息） ——
export const pkgPullState = {
  running: false, done: false,
  total: 0, processed: 0,
  success: 0, failed: 0,
  failedList: [], // [{ name, reason }]
  round: 0,       // 当前轮次 0=首轮, 1..3=重试轮
  current: null,  // 正在处理的包名
  source: 'manual', // manual | auto（定时）
  error: null,
  startedAt: null, finishedAt: null,
}

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

// 编码包名：scoped 包 @scope/name → %40scope%2fname
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)
  }
}

/**
 * 拉取单个清单条目：版本为空 → 解析私服 dist-tags.latest（永远最新）
 * 成功返回 true；失败抛错（由上层收集 reason）
 */
async function pullOne(item) {
  let version = item.version
  if (!version) {
    const meta = await fetchJson(`${VERDACCIO_BASE}/${enc(item.pkg_name)}`, QUERY_TIMEOUT)
    if (!meta || !meta['dist-tags']) throw new Error('私服无此包或未提供元数据（404）')
    version = meta['dist-tags'].latest
    if (!version) throw new Error('私服未提供 latest 版本')
  }
  await triggerVerdaccio(item.pkg_name, version)
  return true
}

/**
 * 启动按清单拉取（同步启动，立即返回；拉取在后台异步执行，前端轮询进度）
 * - running 时重复调用直接返回 started:false
 * - 清单失效（无 pending）→ started:false，active:false（手动由路由层提示，定时自动停止）
 * @param {{ addLog: Function, invalidateScan: Function, source?: 'manual'|'auto' }} opts
 * @returns {{ started: boolean, active: boolean, pendingCount: number }}
 */
export function runPkgListPull({ addLog, invalidateScan, source = 'manual' }) {
  if (pkgPullState.running) return { started: false, active: countPending() > 0, pendingCount: countPending() }
  const pending0 = countPending()
  if (pending0 === 0) {
    return { started: false, active: false, pendingCount: 0 }
  }

  resetState(source)
  pkgPullState.running = true
  console.log(`[pkglist-pull] 开始按清单拉取（source=${source}，共 ${pending0} 个待拉取包，经私服 ${VERDACCIO_BASE}）`)
  // 后台执行（不阻塞请求/事件循环主路径），结果通过 pkgPullState 暴露
  ;(async () => {
    try {
      // 目标集合：先取一次 pending 快照；每成功一个置 done 并计数
      let targets = getPendingRows()
      pkgPullState.total = targets.length

      // —— 首轮 + 重试轮：轮次 0 处理全部，轮次 1..3 仅处理上一轮失败项 ——
      for (let round = 0; round <= MAX_RETRY && targets.length > 0; round++) {
        pkgPullState.round = round
        pkgPullState.failed = 0
        pkgPullState.failedList = []
        const currentTargets = [...targets]
        let next = 0
        const worker = async () => {
          while (next < currentTargets.length) {
            const item = currentTargets[next++]
            pkgPullState.current = item.pkg_name
            try {
              await pullOne(item)
              pkgPullState.success += 1
              markDone(item.id) // 成功 → 清单 side done
            } catch (e) {
              pkgPullState.failed += 1
              pkgPullState.failedList.push({ name: item.pkg_name, reason: (e && e.message) || '拉取失败' })
            }
            pkgPullState.processed += 1
          }
        }
        await Promise.all(Array.from({ length: Math.min(CONCURRENT, currentTargets.length) }, worker))
        // 本轮结束：失败项作为下一轮重试目标（重试前先等所有 worker 完成）
        targets = getPendingRows().filter(t => pkgPullState.failedList.some(f => f.name === t.pkg_name))
        if (round < MAX_RETRY && targets.length > 0) {
          console.log(`[pkglist-pull] 第 ${round + 1} 轮结束：失败 ${targets.length} 个，进入第 ${round + 2} 轮重试`)
        }
      }

      // —— 汇总：失败条目保持 pending（下次可再拉），其余已 done ——
      // 注意：循环结束后的 pkgPullState.failedList 即最后一轮失败项的真实 reason
      //（若全部成功则为空数组），直接保留即可，不要用统一文案覆盖。
      if (!pkgPullState.failedList.length) {
        pkgPullState.failed = 0
      }
      pkgPullState.round = Math.min(pkgPullState.round, MAX_RETRY) // 标记已尽力完成重试
      pkgPullState.current = null

      addLog({
      action: 'list-pull',
      pkgName: null,
      deletedCount: pkgPullState.success,
      deletedVersions: JSON.stringify({
        source,
        round: pkgPullState.round,
        total: pkgPullState.total,
        success: pkgPullState.success,
        failed: pkgPullState.failedList,
      }),
    })
    invalidateScan() // 有新包入库 → 废弃扫描缓存，列表下次请求重扫
    console.log(`[pkglist-pull] 完成（source=${source}）：成功 ${pkgPullState.success} · 失败 ${pkgPullState.failed}`)
  } catch (e) {
      pkgPullState.error = e.message
      console.error('[pkglist-pull] 任务异常:', e.message)
    } finally {
      pkgPullState.running = false
      pkgPullState.done = true
      pkgPullState.finishedAt = new Date().toISOString()
    }
  })() // 立即执行（异步，不阻塞调用方）
  return { started: true, active: true, pendingCount: pending0 }
}

export { VERDACCIO_BASE as PKGLIST_VERDACCIO_BASE }