gongxiangp2/lib/remote-upload-sync.ts

import path from 'path'
import { execFile } from 'child_process'
import { stat } from 'fs/promises'
import { promisify } from 'util'

const execFileAsync = promisify(execFile)
const REMOTE_SYNC_SSH_TIMEOUT_MS = Number(process.env.REMOTE_SYNC_SSH_TIMEOUT_MS || 30000)
const REMOTE_SYNC_SCP_TIMEOUT_MS = Number(process.env.REMOTE_SYNC_SCP_TIMEOUT_MS || 7200000)
const REMOTE_SYNC_RETRY_COUNT = Number(process.env.REMOTE_SYNC_RETRY_COUNT || 3)
const SSH_NON_INTERACTIVE_ARGS = [
  '-o',
  'BatchMode=yes',
  '-o',
  'ConnectTimeout=15',
  '-o',
  'ServerAliveInterval=15',
  '-o',
  'ServerAliveCountMax=2',
]

interface RemoteUploadSyncConfig {
  target: string
  uploadRoot: string
}

function cleanEnv(value: string | undefined) {
  return String(value || '').trim()
}

function shellQuote(value: string) {
  return `'${value.replace(/'/g, `'\\''`)}'`
}

function normalizeRemotePath(value: string) {
  return value.replace(/\\/g, '/').replace(/\/+$/, '')
}

function sleep(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms))
}

async function execFileWithRetry(
  file: string,
  args: string[],
  options: { timeout: number },
  label: string,
) {
  let lastError: unknown
  for (let attempt = 1; attempt <= REMOTE_SYNC_RETRY_COUNT; attempt += 1) {
    try {
      return await execFileAsync(file, args, options)
    } catch (error) {
      lastError = error
      const message = error instanceof Error ? error.message : String(error)
      console.warn(
        `[RemoteUploadSync] ${label} 失败 ${attempt}/${REMOTE_SYNC_RETRY_COUNT}: ${message}`,
      )
      if (attempt < REMOTE_SYNC_RETRY_COUNT) {
        await sleep(1000 * attempt)
      }
    }
  }

  throw lastError
}

function relativeToUploadRoot(storedPath: string) {
  const normalized = String(storedPath || '').replace(/\\/g, '/')
  const relative = normalized.replace(/^\.?\/*uploads\//i, '')

  if (!relative || relative === normalized && path.posix.isAbsolute(normalized)) {
    throw new Error(`远程同步路径无效: ${storedPath}`)
  }

  if (relative.split('/').some((segment) => segment === '..')) {
    throw new Error(`远程同步路径不能包含上级目录: ${storedPath}`)
  }

  return relative
}

export function getRemoteUploadSyncConfig(): RemoteUploadSyncConfig | null {
  const target = cleanEnv(
    process.env.HISTORY_IMPORT_REMOTE_UPLOADS_HOST || process.env.UPLOAD_REMOTE_SYNC_HOST,
  )
  const uploadRoot = cleanEnv(
    process.env.HISTORY_IMPORT_REMOTE_UPLOADS_DIR || process.env.UPLOAD_REMOTE_SYNC_ROOT,
  )

  if (!target && !uploadRoot) {
    return null
  }

  if (!target || !uploadRoot) {
    throw new Error(
      '远程 uploads 同步配置不完整：需要同时配置 HISTORY_IMPORT_REMOTE_UPLOADS_HOST 和 HISTORY_IMPORT_REMOTE_UPLOADS_DIR',
    )
  }

  return {
    target,
    uploadRoot: normalizeRemotePath(uploadRoot),
  }
}

export function isRemoteUploadSyncEnabled() {
  return Boolean(getRemoteUploadSyncConfig())
}

export function getRemoteUploadPath(storedPath: string) {
  const config = getRemoteUploadSyncConfig()
  if (!config) {
    return null
  }

  return path.posix.join(config.uploadRoot, relativeToUploadRoot(storedPath))
}

export async function assertRemoteUploadSyncAccess() {
  const config = getRemoteUploadSyncConfig()
  if (!config) {
    return
  }

  const materialsDir = path.posix.join(config.uploadRoot, 'materials')
  await execFileWithRetry(
    'ssh',
    [
      ...SSH_NON_INTERACTIVE_ARGS,
      config.target,
      `test -d ${shellQuote(materialsDir)} && test -w ${shellQuote(materialsDir)}`,
    ],
    { timeout: REMOTE_SYNC_SSH_TIMEOUT_MS },
    '检查远程 uploads 权限',
  )
}

export async function syncStoredUploadFileToRemote(
  localFilePath: string,
  storedPath: string,
  expectedSize?: number,
) {
  const config = getRemoteUploadSyncConfig()
  if (!config) {
    return null
  }

  const remotePath = getRemoteUploadPath(storedPath)
  if (!remotePath) {
    return null
  }

  const localStat = await stat(localFilePath)
  const expectedFileSize = expectedSize || localStat.size
  const remoteDir = path.posix.dirname(remotePath)
  const sizeMiB = (expectedFileSize / 1024 / 1024).toFixed(2)

  console.log(
    `[RemoteUploadSync] 准备同步 ${storedPath} -> ${config.target}:${remotePath} (${sizeMiB} MiB)`,
  )
  await execFileWithRetry(
    'ssh',
    [...SSH_NON_INTERACTIVE_ARGS, config.target, `mkdir -p ${shellQuote(remoteDir)}`],
    { timeout: REMOTE_SYNC_SSH_TIMEOUT_MS },
    `创建远程目录 ${remoteDir}`,
  )
  console.log(`[RemoteUploadSync] 开始复制 ${storedPath} (${sizeMiB} MiB)`)
  await execFileWithRetry(
    'scp',
    ['-q', ...SSH_NON_INTERACTIVE_ARGS, localFilePath, `${config.target}:${remotePath}`],
    { timeout: REMOTE_SYNC_SCP_TIMEOUT_MS },
    `复制 ${storedPath}`,
  )

  const { stdout } = await execFileWithRetry(
    'ssh',
    [
      ...SSH_NON_INTERACTIVE_ARGS,
      config.target,
      `test -f ${shellQuote(remotePath)} && stat -c %s ${shellQuote(remotePath)}`,
    ],
    { timeout: REMOTE_SYNC_SSH_TIMEOUT_MS },
    `校验远程文件 ${remotePath}`,
  )
  const remoteSize = Number(String(stdout || '').trim())

  if (remoteSize !== expectedFileSize) {
    throw new Error(
      `远程 uploads 同步校验失败: ${storedPath} 本地=${expectedFileSize} 远程=${remoteSize || 'unknown'}`,
    )
  }

  console.log(`[RemoteUploadSync] ${storedPath} -> ${config.target}:${remotePath} (${remoteSize} bytes)`)
  return remotePath
}

export async function deleteStoredUploadFileFromRemote(storedPath: string) {
  const config = getRemoteUploadSyncConfig()
  if (!config) {
    return
  }

  const remotePath = getRemoteUploadPath(storedPath)
  if (!remotePath) {
    return
  }

  await execFileWithRetry(
    'ssh',
    [...SSH_NON_INTERACTIVE_ARGS, config.target, `rm -f -- ${shellQuote(remotePath)}`],
    { timeout: REMOTE_SYNC_SSH_TIMEOUT_MS },
    `删除远程文件 ${remotePath}`,
  )
}