import { basename } from 'node:path'
import { File as BufferFile } from 'node:buffer'
import {
  Agent,
  FormData as UndiciFormData,
  fetch as undiciFetch
} from 'undici'
import type { Readable } from 'node:stream'
import { contentTypeFromKey } from '../content-type'
import {
  getImageIndexRow,
  listAllImageIndexRows,
  type ImageIndexRow
} from '../image-index'
import {
  findUserByUsername,
  getPrimaryAdminUserId,
  listUserIdUsernameMap
} from '../db'
import { collectMediaAssetKeys } from '../media-assets'
import {
  getActiveBackend,
  getBackendForKey
} from '../storage/resolver'
import type { StorageBackend } from '../storage/types'
import {
  getBackupJob,
  markJobDone,
  markJobFailed,
  updateBackupJob
} from '../backup/jobs'
import {
  createJobCancelledError,
  getJobAbortSignal,
  isJobCancellationRequested,
  isJobCancelledError,
  throwIfJobCancellationRequested
} from '../backup/cancel-state'
import { applySiteSyncMedia } from './apply'
import {
  completeSiteSyncMediaArtifacts,
  recordSiteSyncArtifact
} from './artifacts'
import { toSiteSyncMediaTransferWithTags } from './manifest'
import type {
  SiteSyncDirection,
  SiteSyncInventory,
  SiteSyncMediaBundle,
  SiteSyncMediaPage,
  SiteSyncUsers
} from './protocol'

export interface SiteSyncOptions {
  targetBaseUrl: string
  conflictMode: 'skip' | 'overwrite'
  direction: SiteSyncDirection
}

const tokens = (globalThis as typeof globalThis & {
  __pichostSiteSyncTokens?: Map<string, string>
}).__pichostSiteSyncTokens ?? new Map<string, string>()
;(globalThis as typeof globalThis & {
  __pichostSiteSyncTokens?: Map<string, string>
}).__pichostSiteSyncTokens = tokens

export function registerSiteSyncToken(jobId: string, token: string): void {
  tokens.set(jobId, token)
}

const SITE_SYNC_CONNECT_TIMEOUT_MS = 3 * 1000
const SITE_SYNC_RESPONSE_IDLE_TIMEOUT_MS = 3 * 1000
const SITE_SYNC_UPLOAD_HEADERS_TIMEOUT_MS = 10 * 60 * 1000

interface SiteSyncAgents {
  upload: Agent
  object: Agent
}

const agents = (globalThis as typeof globalThis & {
  __pichostSiteSyncAgents?: SiteSyncAgents
}).__pichostSiteSyncAgents ?? {
  upload: new Agent({
    connect: { timeout: SITE_SYNC_CONNECT_TIMEOUT_MS },
    headersTimeout: SITE_SYNC_UPLOAD_HEADERS_TIMEOUT_MS,
    bodyTimeout: SITE_SYNC_RESPONSE_IDLE_TIMEOUT_MS
  }),
  object: new Agent({
    connect: { timeout: SITE_SYNC_CONNECT_TIMEOUT_MS },
    headersTimeout: SITE_SYNC_RESPONSE_IDLE_TIMEOUT_MS,
    bodyTimeout: SITE_SYNC_RESPONSE_IDLE_TIMEOUT_MS
  })
}
;(globalThis as typeof globalThis & {
  __pichostSiteSyncAgents?: SiteSyncAgents
}).__pichostSiteSyncAgents = agents

async function streamToBuffer(
  stream: Readable,
  signal?: AbortSignal
): Promise<Buffer> {
  const chunks: Buffer[] = []
  for await (const chunk of stream) {
    if (signal?.aborted) {
      stream.destroy(createJobCancelledError())
      throw createJobCancelledError()
    }
    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
  }
  return Buffer.concat(chunks)
}

function targetUrl(baseUrl: string, path: string): string {
  return `${baseUrl.replace(/\/+$/, '')}${path}`
}

async function responseError(response: {
  status: number
  json: () => Promise<unknown>
}): Promise<Error> {
  const payload = await response.json().catch(() => null) as {
    message?: string
    statusMessage?: string
  } | null
  return new Error(payload?.message || payload?.statusMessage || `目标站点返回 ${response.status}`)
}

function parseResponseErrorText(
  text: string,
  status: number
): Error {
  try {
    const payload = JSON.parse(text) as {
      message?: string
      statusMessage?: string
    }
    return new Error(
      payload.message
      || payload.statusMessage
      || `目标站点返回 ${status}`
    )
  } catch {
    return new Error(`目标站点返回 ${status}`)
  }
}

const SITE_SYNC_MAX_ATTEMPTS = 3

export interface SiteSyncRetryStatus {
  attempt: number
  maxAttempts: number
  phase: 'attempt' | 'countdown' | 'retry'
  delayMs?: number
  remainingMs?: number
  error?: Error
}

function formatRetryStatus(
  step: string,
  status: SiteSyncRetryStatus
): string {
  if (status.phase === 'attempt') {
    return `${step}（第 ${status.attempt}/${status.maxAttempts} 次）`
  }
  if (status.phase === 'countdown') {
    const seconds = Math.max(
      1,
      Math.ceil((status.remainingMs ?? 0) / 1000)
    )
    return `${step}（第 ${status.attempt}/${status.maxAttempts} 次）· 连接倒计时 ${seconds}s`
  }
  if ((status.delayMs ?? 0) < 1000) {
    return `${step}第 ${status.attempt}/${status.maxAttempts} 次无响应，即将重试：${status.error?.message ?? '连接失败'}`
  }
  const seconds = Math.max(1, Math.round((status.delayMs ?? 0) / 1000))
  return `${step}第 ${status.attempt}/${status.maxAttempts} 次失败，${seconds} 秒后重试：${status.error?.message ?? '连接失败'}`
}

function normalizeSiteSyncError(error: Error): Error {
  if (/timeout/i.test(error.name) || /timed out|idle for too long/i.test(error.message)) {
    return new Error('目标站点在 3 秒内未响应')
  }
  if (error instanceof TypeError && /fetch failed|terminated/i.test(error.message)) {
    return new Error('目标站点无法连接')
  }
  return error
}

export async function retrySiteSyncOperation<T>(
  operation: () => Promise<T>,
  signal?: AbortSignal,
  onStatus?: (status: SiteSyncRetryStatus) => void
): Promise<T> {
  if (signal?.aborted) throw createJobCancelledError()
  let lastError: Error | null = null
  for (let attempt = 1; attempt <= SITE_SYNC_MAX_ATTEMPTS; attempt++) {
    let countdownTimer: ReturnType<typeof setInterval> | null = null
    try {
      onStatus?.({
        attempt,
        maxAttempts: SITE_SYNC_MAX_ATTEMPTS,
        phase: 'attempt'
      })
      if (onStatus) {
        let remainingMs = SITE_SYNC_CONNECT_TIMEOUT_MS
        onStatus({
          attempt,
          maxAttempts: SITE_SYNC_MAX_ATTEMPTS,
          phase: 'countdown',
          remainingMs
        })
        countdownTimer = setInterval(() => {
          remainingMs -= 1000
          if (remainingMs <= 0) {
            if (countdownTimer) clearInterval(countdownTimer)
            countdownTimer = null
            return
          }
          onStatus({
            attempt,
            maxAttempts: SITE_SYNC_MAX_ATTEMPTS,
            phase: 'countdown',
            remainingMs
          })
        }, 1000)
      }
      return await operation()
    } catch (error) {
      if (countdownTimer) clearInterval(countdownTimer)
      countdownTimer = null
      if (signal?.aborted) throw createJobCancelledError()
      lastError = error instanceof Error ? error : new Error('两站点同步失败')
      if (attempt < SITE_SYNC_MAX_ATTEMPTS) {
        const delayMs = 0
        onStatus?.({
          attempt,
          maxAttempts: SITE_SYNC_MAX_ATTEMPTS,
          phase: 'retry',
          delayMs,
          error: normalizeSiteSyncError(lastError)
        })
        await new Promise<void>((resolve, reject) => {
          const onAbort = () => {
            clearTimeout(timer)
            reject(createJobCancelledError())
          }
          const timer = setTimeout(() => {
            signal?.removeEventListener('abort', onAbort)
            resolve()
          }, delayMs)
          signal?.addEventListener('abort', onAbort, { once: true })
        })
      }
    } finally {
      if (countdownTimer) clearInterval(countdownTimer)
    }
  }
  throw lastError
    ? normalizeSiteSyncError(lastError)
    : new Error('两站点同步失败')
}

async function fetchSiteSyncJsonOnce<T>(
  url: string,
  token: string,
  init?: RequestInit
): Promise<T> {
  const connectionController = new AbortController()
  const onAbort = () => connectionController.abort(createJobCancelledError())
  init?.signal?.addEventListener('abort', onAbort, { once: true })
  const timeout = setTimeout(() => {
    connectionController.abort(new DOMException(
      'Target connection or TLS handshake timed out',
      'TimeoutError'
    ))
  }, SITE_SYNC_CONNECT_TIMEOUT_MS)

  try {
    const response = await fetch(url, {
      ...init,
      headers: {
        'Auth-Token': token,
        ...(init?.headers ?? {})
      },
      signal: init?.signal
        ? AbortSignal.any([init.signal, connectionController.signal])
        : connectionController.signal
    })
    clearTimeout(timeout)

    if (!response.body) {
      const text = await response.text()
      if (!response.ok) throw parseResponseErrorText(text, response.status)
      return JSON.parse(text) as T
    }

    const reader = response.body.getReader()
    const decoder = new TextDecoder()
    let text = ''
    try {
      while (true) {
        const chunk = await readChunkWithIdleTimeout(
          reader,
          init?.signal ?? undefined
        )
        if (chunk.done) break
        text += decoder.decode(chunk.value, { stream: true })
      }
      text += decoder.decode()
    } finally {
      reader.releaseLock()
    }
    if (!response.ok) throw parseResponseErrorText(text, response.status)
    return JSON.parse(text) as T
  } finally {
    clearTimeout(timeout)
    init?.signal?.removeEventListener('abort', onAbort)
  }
}

async function readChunkWithIdleTimeout(
  reader: ReadableStreamDefaultReader<Uint8Array>,
  signal?: AbortSignal
): Promise<ReadableStreamReadResult<Uint8Array>> {
  if (signal?.aborted) {
    await reader.cancel().catch(() => {})
    throw createJobCancelledError()
  }

  return new Promise((resolve, reject) => {
    let settled = false
    const cleanup = () => {
      clearTimeout(timer)
      signal?.removeEventListener('abort', onAbort)
    }
    const onAbort = () => {
      if (settled) return
      settled = true
      cleanup()
      void reader.cancel().catch(() => {})
      reject(createJobCancelledError())
    }
    const timer = setTimeout(() => {
      if (settled) return
      settled = true
      cleanup()
      void reader.cancel().catch(() => {})
      reject(new DOMException(
        'Target response body was idle for too long',
        'TimeoutError'
      ))
    }, SITE_SYNC_RESPONSE_IDLE_TIMEOUT_MS)
    signal?.addEventListener('abort', onAbort, { once: true })

    reader.read().then(
      (result) => {
        if (settled) return
        settled = true
        cleanup()
        resolve(result)
      },
      (error) => {
        if (settled) return
        settled = true
        cleanup()
        reject(error)
      }
    )
  })
}

export async function fetchSiteSyncJson<T>(
  url: string,
  token: string,
  init?: RequestInit,
  signal?: AbortSignal,
  onStatus?: (status: SiteSyncRetryStatus) => void
): Promise<T> {
  return retrySiteSyncOperation(
    () => fetchSiteSyncJsonOnce<T>(url, token, {
      ...init,
      signal
    }),
    signal,
    onStatus
  )
}

async function uploadObject(
  jobId: string,
  options: SiteSyncOptions,
  token: string,
  backend: StorageBackend,
  row: ImageIndexRow,
  ownerUsername: string | null,
  key: string
): Promise<void> {
  const stored = await backend.head(key)
  if (!stored) throw new Error(`源文件不存在：${key}`)
  const bytes = await streamToBuffer(
    await backend.createStream(key),
    getJobAbortSignal(jobId)
  )
  throwIfJobCancellationRequested(jobId)
  const form = new UndiciFormData()
  const isPrimary = key === row.key
  form.append('jobId', jobId)
  form.append('mediaKey', row.key)
  form.append('key', key)
  form.append('conflictMode', options.conflictMode)
  form.append('ownerUsername', ownerUsername ?? '')
  form.append('originalName', isPrimary ? row.original_name : basename(key))
  form.append('uploadedAt', row.uploaded_at)
  const contentType = isPrimary ? row.content_type : contentTypeFromKey(key)
  form.append('contentType', contentType)
  form.append(
    'file',
    new BufferFile(
      [new Uint8Array(bytes)],
      basename(key),
      { type: contentType }
    )
  )

  const response = await undiciFetch(
    targetUrl(options.targetBaseUrl, '/api/site-sync/object'),
    {
      method: 'POST',
      body: form,
      headers: { 'Auth-Token': token },
      signal: getJobAbortSignal(jobId),
      dispatcher: agents.upload
    }
  )
  if (!response.ok) throw await responseError(response)
  await response.json().catch(() => null)
}

async function downloadSiteSyncObject(
  jobId: string,
  options: SiteSyncOptions,
  token: string,
  key: string
): Promise<Buffer> {
  const response = await undiciFetch(
    targetUrl(
      options.targetBaseUrl,
      `/api/site-sync/object?key=${encodeURIComponent(key)}`
    ),
    {
      headers: {
        'Auth-Token': token
      },
      signal: getJobAbortSignal(jobId),
      dispatcher: agents.object
    }
  )
  if (!response.ok) throw await responseError(response)
  return Buffer.from(await response.arrayBuffer())
}

async function runSiteSyncPull(
  jobId: string,
  options: SiteSyncOptions,
  token: string
): Promise<void> {
  const job = getBackupJob(jobId)
  const jobSignal = getJobAbortSignal(jobId)
  const checkpoint = job?.payload ?? {}
  let lastMediaKey = typeof checkpoint.lastMediaKey === 'string'
    ? checkpoint.lastMediaKey
    : ''
  let scanned = Number(checkpoint.scanned ?? 0)
  let added = Number(checkpoint.added ?? 0)
  let skipped = Number(checkpoint.skipped ?? 0)
  let filesSynced = Number(checkpoint.filesSynced ?? 0)
  let totalFiles = Number(checkpoint.totalFiles ?? 0)
  let totalMedia = Number(checkpoint.totalMedia ?? job?.total ?? 0)
  const failedKeys = new Set(
    Array.isArray(checkpoint.failedKeys)
      ? checkpoint.failedKeys.filter((key): key is string => typeof key === 'string')
      : []
  )

  const saveCheckpoint = (message: string, cancelled = false) => {
    updateBackupJob(jobId, {
      progress: scanned,
      total: totalMedia,
      message,
      payload: {
        targetBaseUrl: options.targetBaseUrl,
        conflictMode: options.conflictMode,
        direction: 'pull',
        cancelled,
        lastMediaKey,
        scanned,
        added,
        skipped,
        failed: failedKeys.size,
        filesSynced,
        totalFiles,
        totalMedia,
        failedKeys: [...failedKeys],
        resumable: true
      }
    })
  }
  const progressMessage = (suffix = '') => (
    `扫描 ${scanned}/${totalMedia || scanned} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size}${suffix}`
  )
  const updateLiveStatus = (step: string, status?: SiteSyncRetryStatus) => {
    const detail = status ? formatRetryStatus(step, status) : step
    try {
      updateBackupJob(jobId, {
        message: progressMessage(` · ${detail}`)
      })
    } catch {
      // Progress messages must never fail the sync itself.
    }
  }
  const failAndCheckpoint = (error: unknown) => {
    const cancelled = isJobCancellationRequested(jobId)
      || isJobCancelledError(error)
    const message = cancelled
      ? progressMessage(' · 已取消，已保存断点')
      : progressMessage(' · 已保存断点')
    saveCheckpoint(message, cancelled)
    markJobFailed(
      jobId,
      cancelled
        ? '任务已取消'
        : error instanceof Error
          ? error.message
          : '从目标站点拉取失败',
      message
    )
  }

  try {
    const targetUsers = new Set(
      (await fetchSiteSyncJson<SiteSyncUsers>(
        targetUrl(options.targetBaseUrl, '/api/site-sync/users'),
        token,
        undefined,
        jobSignal,
        status => updateLiveStatus('连接目标站点', status)
      )).users
    )

    const backend = await retrySiteSyncOperation(
      () => getActiveBackend(),
      jobSignal
    )
    const actorUserId = job?.userId ?? getPrimaryAdminUserId() ?? 0
    const readMediaPage = (path: string) => fetchSiteSyncJson<SiteSyncMediaPage>(
      targetUrl(options.targetBaseUrl, path),
      token,
      undefined,
      jobSignal,
      status => updateLiveStatus('读取远程媒体清单', status)
    )

    const processBundleOnce = async (
      bundle: SiteSyncMediaBundle,
      isRetry: boolean,
      setStep: (step: string) => void
    ): Promise<'added' | 'skipped'> => {
      const { media, assets } = bundle
      throwIfJobCancellationRequested(jobId)
      setStep(`准备同步媒体 ${media.key}`)
      if (media.ownerUsername && !targetUsers.has(media.ownerUsername)) {
        throw new Error(`当前站点缺少同名用户：${media.ownerUsername}`)
      }
      if (
        !isRetry
        && options.conflictMode === 'skip'
        && getImageIndexRow(media.key)
      ) {
        completeSiteSyncMediaArtifacts(jobId, media.key)
        return 'skipped'
      }

      const owner = media.ownerUsername
        ? findUserByUsername(media.ownerUsername)
        : null
      for (const asset of assets) {
        throwIfJobCancellationRequested(jobId)
        setStep(`下载对象 ${basename(asset.key)}`)
        const existedBefore = Boolean(await backend.head(asset.key))
        if (options.conflictMode === 'skip' && existedBefore) {
          continue
        }
        const bytes = await downloadSiteSyncObject(
          jobId,
          options,
          token,
          asset.key
        )
        throwIfJobCancellationRequested(jobId)
        if (bytes.length !== asset.size) {
          throw new Error(
            `同步对象大小不一致：${asset.key}，预期 ${asset.size}，实际 ${bytes.length}`
          )
        }
        setStep(`写入对象 ${basename(asset.key)}`)
        const isPrimary = asset.key === media.key
        recordSiteSyncArtifact({
          jobId,
          mediaKey: media.key,
          key: asset.key,
          backendId: backend.id,
          created: !existedBefore
        })
        await backend.put(asset.key, new Uint8Array(bytes), {
          originalName: isPrimary ? media.originalName : basename(asset.key),
          uploadedAt: media.uploadedAt,
          contentType: isPrimary
            ? media.contentType
            : contentTypeFromKey(asset.key),
          size: bytes.length,
          userId: owner?.id ?? null,
          mediaType: isPrimary
            ? media.mediaType
            : asset.key.startsWith('videos/')
              ? 'video'
              : asset.key.startsWith('files/')
                ? 'file'
                : 'image'
        })
        throwIfJobCancellationRequested(jobId)
      }

      setStep(`写入媒体记录 ${media.key}`)
      await applySiteSyncMedia(media, 'overwrite', actorUserId)
      completeSiteSyncMediaArtifacts(jobId, media.key)
      return 'added'
    }
    const processBundle = (bundle: SiteSyncMediaBundle, isRetry: boolean) => {
      let step = `同步媒体 ${bundle.media.key}`
      return retrySiteSyncOperation(
        () => processBundleOnce(
          bundle,
          isRetry,
          (value) => { step = value }
        ),
        jobSignal,
        status => updateLiveStatus(step, status)
      )
    }
    const recordOutcome = (outcome: 'added' | 'skipped', fileCount: number) => {
      if (outcome === 'added') added += 1
      else skipped += 1
      filesSynced += fileCount
      totalFiles = Math.max(totalFiles, filesSynced)
    }

    for (const key of [...failedKeys]) {
      try {
        const page = await readMediaPage(
          `/api/site-sync/media?key=${encodeURIComponent(key)}`
        )
        const bundle = page.items[0]
        if (!bundle) throw new Error(`目标媒体不存在：${key}`)
        const outcome = await processBundle(bundle, true)
        recordOutcome(outcome, bundle.assets.length)
        failedKeys.delete(key)
        if (key > lastMediaKey) lastMediaKey = key
        saveCheckpoint(progressMessage())
      } catch (error) {
        failedKeys.add(key)
        failAndCheckpoint(error)
        return
      }
    }

    let cursor = lastMediaKey
    let lastCheckpointAt = 0
    while (true) {
      let page: SiteSyncMediaPage
      try {
        page = await readMediaPage(
          `/api/site-sync/media?limit=50${cursor ? `&after=${encodeURIComponent(cursor)}` : ''}`
        )
      } catch (error) {
        failAndCheckpoint(error)
        return
      }
      totalMedia = Math.max(totalMedia, page.totalMedia)

      for (const bundle of page.items) {
        scanned += 1
        try {
          const outcome = await processBundle(bundle, false)
          recordOutcome(outcome, bundle.assets.length)
          failedKeys.delete(bundle.media.key)
        } catch (error) {
          failedKeys.add(bundle.media.key)
          lastMediaKey = bundle.media.key
          failAndCheckpoint(error)
          return
        }

        lastMediaKey = bundle.media.key
        const now = Date.now()
        if (
          now - lastCheckpointAt >= 1000
          || scanned === totalMedia
          || !page.nextCursor
        ) {
          saveCheckpoint(progressMessage())
          lastCheckpointAt = now
        }
      }

      if (!page.nextCursor) break
      cursor = page.nextCursor
    }

    if (failedKeys.size) {
      markJobFailed(
        jobId,
        `${failedKeys.size} 条媒体拉取失败`,
        progressMessage()
      )
      saveCheckpoint(progressMessage())
      return
    }

    throwIfJobCancellationRequested(jobId)
    totalFiles = filesSynced
    markJobDone(jobId, {
      message: `扫描 ${scanned}/${totalMedia || scanned} · 新增 ${added} · 跳过 ${skipped} · 失败 0`,
      payload: {
        targetBaseUrl: options.targetBaseUrl,
        conflictMode: options.conflictMode,
        direction: 'pull',
        lastMediaKey,
        scanned,
        added,
        skipped,
        failed: 0,
        filesSynced,
        totalFiles,
        totalMedia: totalMedia || scanned,
        failedKeys: [],
        resumable: false
      }
    })
  } catch (error) {
    failAndCheckpoint(error)
  }
}

export async function runSiteSyncJob(
  jobId: string,
  options: SiteSyncOptions
): Promise<void> {
  const token = tokens.get(jobId)
  if (!token) throw new Error('同步 Token 不可用，请重新发起任务')
  const jobSignal = getJobAbortSignal(jobId)

  try {
    if (options.direction === 'pull') {
      await runSiteSyncPull(jobId, options, token)
      return
    }

    const inventory = await fetchSiteSyncJson<SiteSyncInventory>(
      targetUrl(options.targetBaseUrl, '/api/site-sync/inventory'),
      token,
      undefined,
      jobSignal,
      (status) => {
        try {
          updateBackupJob(jobId, {
            message: formatRetryStatus('连接目标站点', status)
          })
        } catch {
          // Keep the sync running even if a progress update cannot be stored.
        }
      }
    )
    const targetUsers = new Set(inventory.users)
    const targetMedia = new Set(inventory.mediaKeys)
    const targetObjects = new Map(inventory.objects.map(item => [item.key, item.size]))
    const rows = listAllImageIndexRows()
    const ownerMap = listUserIdUsernameMap()
    const job = getBackupJob(jobId)
    const checkpoint = job?.payload ?? {}
    let lastMediaKey = typeof checkpoint.lastMediaKey === 'string'
      ? checkpoint.lastMediaKey
      : ''
    let scanned = Number(checkpoint.scanned ?? 0)
    let added = Number(checkpoint.added ?? 0)
    let skipped = Number(checkpoint.skipped ?? 0)
    let filesSynced = Number(checkpoint.filesSynced ?? 0)
    const failedKeys = new Set(
      Array.isArray(checkpoint.failedKeys)
        ? checkpoint.failedKeys.filter((key): key is string => typeof key === 'string')
        : []
    )

    const missingUsers = [...new Set(
      rows
        .map(row => row.user_id == null ? null : ownerMap.get(row.user_id))
        .filter((name): name is string => Boolean(name))
    )].filter(name => !targetUsers.has(name))
    if (missingUsers.length) {
      throw new Error(`目标站点缺少同名用户：${missingUsers.join('、')}`)
    }

    const plans: Array<{
      row: ImageIndexRow
      ownerUsername: string | null
      backend: StorageBackend
      keys: string[]
    }> = []
    let totalFiles = 0
    updateBackupJob(jobId, {
      total: rows.length,
      progress: scanned,
      message: '正在统计同步文件…'
    })

    for (const row of rows) {
      throwIfJobCancellationRequested(jobId)
      const backend = await getBackendForKey(row.key)
      let assetKeys: string[]
      try {
        assetKeys = await collectMediaAssetKeys(row, backend)
      } catch {
        assetKeys = [
          ...(!Number(row.source_deleted ?? 0) ? [row.key] : []),
          ...(row.thumb_key ? [row.thumb_key] : []),
          ...(row.hls_key ? [row.hls_key] : [])
        ]
      }
      const keys = [...new Set(assetKeys)]
      totalFiles += keys.length
      plans.push({
        row,
        ownerUsername: row.user_id == null ? null : ownerMap.get(row.user_id) ?? null,
        backend,
        keys
      })
    }

    const saveCheckpoint = (message: string, cancelled = false) => {
      updateBackupJob(jobId, {
        progress: scanned,
        total: rows.length,
        message,
        payload: {
          targetBaseUrl: options.targetBaseUrl,
          conflictMode: options.conflictMode,
          direction: 'push',
          cancelled,
          lastMediaKey,
          scanned,
          added,
          skipped,
          failed: failedKeys.size,
          filesSynced,
          totalFiles,
          failedKeys: [...failedKeys],
          resumable: true
        }
      })
    }
    const updateLiveStatus = (step: string, status?: SiteSyncRetryStatus) => {
      const detail = status ? formatRetryStatus(step, status) : step
      try {
        updateBackupJob(jobId, {
          message: `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · ${detail}`
        })
      } catch {
        // Progress messages must never fail the sync itself.
      }
    }

    plans.sort((a, b) => (
      a.row.key < b.row.key
        ? -1
        : a.row.key > b.row.key
          ? 1
          : 0
    ))
    lastMediaKey = lastMediaKey || ''
    let lastCheckpointAt = 0
    const processPlanOnce = async (
      plan: typeof plans[number],
      isRetry: boolean,
      setStep: (step: string) => void
    ) => {
      const { row, ownerUsername, backend, keys } = plan
      throwIfJobCancellationRequested(jobId)
      setStep(`准备同步媒体 ${row.key}`)
      if (
        !isRetry
        && options.conflictMode === 'skip'
        && targetMedia.has(row.key)
      ) {
        setStep(`提交已存在的媒体 ${row.key}`)
        await fetchSiteSyncJsonOnce<{ success: boolean }>(
          targetUrl(options.targetBaseUrl, '/api/site-sync/complete'),
          token,
          {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ jobId, mediaKey: row.key }),
            signal: getJobAbortSignal(jobId)
          }
        )
        skipped += 1
        filesSynced = Math.min(totalFiles, filesSynced + keys.length)
        failedKeys.delete(row.key)
        return
      }

      for (const key of keys) {
        throwIfJobCancellationRequested(jobId)
        setStep(`上传对象 ${basename(key)}`)
        const stored = await backend.head(key)
        if (!stored) throw new Error(`源文件不存在：${key}`)
        if (
          options.conflictMode === 'skip'
          && targetObjects.get(key) === stored.size
        ) {
          continue
        }
        await uploadObject(
          jobId,
          options,
          token,
          backend,
          row,
          ownerUsername,
          key
        )
      }

      throwIfJobCancellationRequested(jobId)
      setStep(`写入媒体记录 ${row.key}`)
      await fetchSiteSyncJsonOnce<{ success: boolean }>(
        targetUrl(options.targetBaseUrl, '/api/site-sync/media'),
        token,
        {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            jobId,
            media: toSiteSyncMediaTransferWithTags(row, ownerUsername),
            conflictMode: options.conflictMode
          }),
          signal: getJobAbortSignal(jobId)
        }
      )

      failedKeys.delete(row.key)
      added += 1
      filesSynced = Math.min(totalFiles, filesSynced + keys.length)
    }
    const processPlan = (
      plan: typeof plans[number],
      isRetry: boolean
    ) => {
      let step = `同步媒体 ${plan.row.key}`
      return retrySiteSyncOperation(
        () => processPlanOnce(
          plan,
          isRetry,
          (value) => { step = value }
        ),
        jobSignal,
        status => updateLiveStatus(step, status)
      )
    }

    const retryPlans = plans.filter(plan => failedKeys.has(plan.row.key))
    for (const plan of retryPlans) {
      try {
        await processPlan(plan, true)
        lastMediaKey = plan.row.key
      } catch (error) {
        failedKeys.add(plan.row.key)
        const cancelled = isJobCancellationRequested(jobId)
          || isJobCancelledError(error)
        const message = `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · ${cancelled ? '已取消，' : ''}已保存断点`
        saveCheckpoint(message, cancelled)
        markJobFailed(
          jobId,
          cancelled
            ? '任务已取消'
            : error instanceof Error
              ? `${error.message} · 已保存断点`
              : '连续 3 次重试失败，已保存断点',
          message
        )
        return
      }
    }

    for (const plan of plans) {
      if (plan.row.key <= lastMediaKey) continue
      scanned += 1
      try {
        await processPlan(plan, false)
      } catch (error) {
        failedKeys.add(plan.row.key)
        lastMediaKey = plan.row.key
        const cancelled = isJobCancellationRequested(jobId)
          || isJobCancelledError(error)
        const message = `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size} · ${cancelled ? '已取消，' : ''}已保存断点`
        saveCheckpoint(message, cancelled)
        markJobFailed(
          jobId,
          cancelled
            ? '任务已取消'
            : error instanceof Error
              ? `${error.message} · 已保存断点`
              : '连续 3 次重试失败，已保存断点',
          message
        )
        return
      }
      lastMediaKey = plan.row.key
      const now = Date.now()
      if (now - lastCheckpointAt >= 1000 || scanned === rows.length || failedKeys.size) {
        saveCheckpoint(
          `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size}`
        )
        lastCheckpointAt = now
      }
    }

    if (failedKeys.size) {
      markJobFailed(
        jobId,
        `${failedKeys.size} 条媒体同步失败`,
        `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size}`
      )
      saveCheckpoint(
        `扫描 ${scanned}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 ${failedKeys.size}`
      )
      return
    }

    throwIfJobCancellationRequested(jobId)
    markJobDone(jobId, {
      message: `扫描 ${rows.length}/${rows.length} · 新增 ${added} · 跳过 ${skipped} · 失败 0`,
      payload: {
        targetBaseUrl: options.targetBaseUrl,
        conflictMode: options.conflictMode,
        direction: 'push',
        lastMediaKey,
        scanned: rows.length,
        added,
        skipped,
        failed: 0,
        filesSynced: totalFiles,
        totalFiles,
        failedKeys: [],
        resumable: false
      }
    })
  } catch (error) {
    if (isJobCancellationRequested(jobId) || isJobCancelledError(error)) {
      markJobFailed(jobId, '任务已取消', '任务已取消')
      return
    }
    const message = error instanceof Error ? error.message : '两站点同步失败'
    markJobFailed(jobId, message)
    throw error
  } finally {
    tokens.delete(jobId)
  }
}