同步处理

server/utils/site-sync/chunks.ts

@@ -0,0 +1,350 @@
+import { createHash } from 'node:crypto'
+import { promises as fs } from 'node:fs'
+import { join } from 'node:path'
+import { getDataDir } from '../data-dir'
+import { findUserByUsername } from '../db'
+import { getBackendById } from '../storage/resolver'
+import { putLocalFileAtomic } from '../storage/local'
+import type { StoredImageMeta } from '../storage/types'
+import { recordSiteSyncArtifact } from './artifacts'
+
+export const SITE_SYNC_CHUNK_SIZE = 4 * 1024 * 1024
+const CHUNK_ROOT = '.pichost-tmp/site-sync-chunks'
+
+export interface SiteSyncChunkMetadata {
+  jobId: string
+  mediaKey: string
+  key: string
+  totalSize: number
+  totalChunks: number
+  ownerUsername: string | null
+  originalName: string
+  uploadedAt: string
+  contentType: string
+  mediaType: StoredImageMeta['mediaType']
+  conflictMode: 'skip' | 'overwrite'
+  backendId: string
+  updatedAt: string
+}
+
+interface StoredSession extends SiteSyncChunkMetadata {
+  created: boolean
+}
+
+function uploadIdFor(jobId: string, key: string): string {
+  return createHash('sha256')
+    .update(jobId)
+    .update('\0')
+    .update(key)
+    .digest('hex')
+}
+
+function sessionDir(uploadId: string): string {
+  return join(getDataDir(), CHUNK_ROOT, uploadId)
+}
+
+function metadataPath(uploadId: string): string {
+  return join(sessionDir(uploadId), 'session.json')
+}
+
+function chunkPath(uploadId: string, index: number): string {
+  return join(sessionDir(uploadId), `chunk-${String(index).padStart(6, '0')}.part`)
+}
+
+async function readSession(uploadId: string): Promise<StoredSession | null> {
+  try {
+    const raw = await fs.readFile(metadataPath(uploadId), 'utf8')
+    return JSON.parse(raw) as StoredSession
+  } catch {
+    return null
+  }
+}
+
+async function writeSession(
+  uploadId: string,
+  session: StoredSession
+): Promise<void> {
+  const path = metadataPath(uploadId)
+  const temp = `${path}.${process.pid}.${Date.now()}.tmp`
+  await fs.writeFile(temp, JSON.stringify(session))
+  await fs.rename(temp, path)
+}
+
+export async function getSiteSyncChunkStatus(input: {
+  jobId: string
+  key: string
+}): Promise<{
+  exists: boolean
+  complete: boolean
+  received: number[]
+  totalChunks: number
+  totalSize: number
+}> {
+  const uploadId = uploadIdFor(input.jobId, input.key)
+  const session = await readSession(uploadId)
+  if (!session) {
+    return {
+      exists: false,
+      complete: false,
+      received: [],
+      totalChunks: 0,
+      totalSize: 0
+    }
+  }
+  const received: number[] = []
+  for (let index = 0; index < session.totalChunks; index++) {
+    try {
+      const stat = await fs.stat(chunkPath(uploadId, index))
+      if (stat.isFile()) received.push(index)
+    } catch {
+      // Missing chunks are exactly what the caller needs to send.
+    }
+  }
+  return {
+    exists: true,
+    complete: received.length === session.totalChunks,
+    received,
+    totalChunks: session.totalChunks,
+    totalSize: session.totalSize
+  }
+}
+
+async function ensureSession(
+  metadata: SiteSyncChunkMetadata
+): Promise<{ uploadId: string, session: StoredSession }> {
+  const uploadId = uploadIdFor(metadata.jobId, metadata.key)
+  const existing = await readSession(uploadId)
+  if (existing) {
+    if (
+      existing.totalSize !== metadata.totalSize
+      || existing.totalChunks !== metadata.totalChunks
+      || existing.mediaKey !== metadata.mediaKey
+    ) {
+      throw new Error('分块上传会话参数不一致')
+    }
+    return { uploadId, session: existing }
+  }
+
+  const backend = await getBackendById(metadata.backendId)
+  if (!backend) throw new Error('分块上传存储后端不存在')
+  const session: StoredSession = {
+    ...metadata,
+    created: !await backend.head(metadata.key),
+    updatedAt: new Date().toISOString()
+  }
+  await fs.mkdir(sessionDir(uploadId), { recursive: true })
+  await writeSession(uploadId, session)
+  return { uploadId, session }
+}
+
+async function writeFileAtomic(path: string, bytes: Uint8Array): Promise<void> {
+  const temp = `${path}.${process.pid}.${Date.now()}.tmp`
+  await fs.writeFile(temp, bytes)
+  await fs.rename(temp, path)
+}
+
+export async function writeSiteSyncChunk(input: {
+  metadata: SiteSyncChunkMetadata
+  index: number
+  bytes: Uint8Array
+}): Promise<{
+  received: number
+  totalChunks: number
+  complete: boolean
+}> {
+  const { uploadId, session } = await ensureSession(input.metadata)
+  if (
+    !Number.isInteger(input.index)
+    || input.index < 0
+    || input.index >= session.totalChunks
+  ) {
+    throw new Error('分块序号无效')
+  }
+  const start = input.index * SITE_SYNC_CHUNK_SIZE
+  const expectedSize = Math.min(
+    SITE_SYNC_CHUNK_SIZE,
+    session.totalSize - start
+  )
+  if (input.bytes.length !== expectedSize) {
+    throw new Error(`分块大小不一致：预期 ${expectedSize}，实际 ${input.bytes.length}`)
+  }
+
+  await writeFileAtomic(chunkPath(uploadId, input.index), input.bytes)
+  session.updatedAt = new Date().toISOString()
+  await writeSession(uploadId, session)
+  let received = 0
+  for (let index = 0; index < session.totalChunks; index++) {
+    try {
+      await fs.access(chunkPath(uploadId, index))
+      received++
+    } catch {
+      // Keep counting.
+    }
+  }
+  return {
+    received,
+    totalChunks: session.totalChunks,
+    complete: received === session.totalChunks
+  }
+}
+
+async function concatenateChunks(
+  uploadId: string,
+  session: StoredSession
+): Promise<string> {
+  const outputPath = join(sessionDir(uploadId), 'assembled.part')
+  const output = await fs.open(outputPath, 'w')
+  try {
+    for (let index = 0; index < session.totalChunks; index++) {
+      const input = await fs.open(chunkPath(uploadId, index), 'r')
+      try {
+        const stat = await input.stat()
+        let offset = 0
+        while (offset < stat.size) {
+          const length = Math.min(1024 * 1024, stat.size - offset)
+          const buffer = Buffer.allocUnsafe(length)
+          const { bytesRead } = await input.read(buffer, 0, length, offset)
+          if (!bytesRead) break
+          await output.write(buffer, 0, bytesRead)
+          offset += bytesRead
+        }
+      } finally {
+        await input.close()
+      }
+    }
+    await output.sync()
+  } finally {
+    await output.close()
+  }
+  return outputPath
+}
+
+export async function finalizeSiteSyncChunk(input: {
+  jobId: string
+  key: string
+}): Promise<{ success: true, skipped: boolean }> {
+  const uploadId = uploadIdFor(input.jobId, input.key)
+  const session = await readSession(uploadId)
+  if (!session) throw new Error('分块上传会话不存在')
+  const status = await getSiteSyncChunkStatus(input)
+  if (!status.complete) {
+    throw new Error(`分块尚未上传完成：${status.received.length}/${session.totalChunks}`)
+  }
+
+  const backend = await getBackendById(session.backendId)
+  if (!backend) throw new Error('分块上传存储后端不存在')
+  const existing = await backend.head(session.key)
+  if (session.conflictMode === 'skip' && existing) {
+    await fs.rm(sessionDir(uploadId), { recursive: true, force: true })
+    return { success: true, skipped: true }
+  }
+
+  const assembledPath = await concatenateChunks(uploadId, session)
+  const stat = await fs.stat(assembledPath)
+  if (stat.size !== session.totalSize) {
+    throw new Error(`组装文件大小不一致：预期 ${session.totalSize}，实际 ${stat.size}`)
+  }
+
+  recordSiteSyncArtifact({
+    jobId: session.jobId,
+    mediaKey: session.mediaKey,
+    key: session.key,
+    backendId: session.backendId,
+    created: session.created
+  })
+
+  const owner = session.ownerUsername
+    ? findUserByUsername(session.ownerUsername)
+    : null
+  const meta: StoredImageMeta = {
+    originalName: session.originalName,
+    uploadedAt: session.uploadedAt,
+    contentType: session.contentType,
+    size: session.totalSize,
+    userId: owner?.id ?? null,
+    mediaType: session.mediaType
+  }
+  if (backend.type === 'local') {
+    await putLocalFileAtomic(session.key, assembledPath)
+  } else {
+    const chunks: Buffer[] = []
+    for (let index = 0; index < session.totalChunks; index++) {
+      chunks.push(await fs.readFile(chunkPath(uploadId, index)))
+    }
+    await backend.put(session.key, Buffer.concat(chunks), meta)
+  }
+  await fs.rm(sessionDir(uploadId), { recursive: true, force: true })
+  return { success: true, skipped: false }
+}
+
+export async function cleanupSiteSyncChunkSessions(
+  jobId: string
+): Promise<number> {
+  const root = join(getDataDir(), CHUNK_ROOT)
+  let entries
+  try {
+    entries = await fs.readdir(root, { withFileTypes: true })
+  } catch {
+    return 0
+  }
+  let removed = 0
+  for (const entry of entries) {
+    if (!entry.isDirectory()) continue
+    const session = await readSession(entry.name)
+    if (session?.jobId !== jobId) continue
+    await fs.rm(sessionDir(entry.name), { recursive: true, force: true })
+    removed++
+  }
+  return removed
+}
+
+export async function cleanupExpiredSiteSyncChunkSessions(
+  maxAgeMs = 24 * 60 * 60 * 1000
+): Promise<number> {
+  const root = join(getDataDir(), CHUNK_ROOT)
+  let entries
+  try {
+    entries = await fs.readdir(root, { withFileTypes: true })
+  } catch {
+    return 0
+  }
+  const cutoff = Date.now() - maxAgeMs
+  let removed = 0
+  for (const entry of entries) {
+    if (!entry.isDirectory()) continue
+    const session = await readSession(entry.name)
+    if (!session || new Date(session.updatedAt).getTime() > cutoff) continue
+    await fs.rm(sessionDir(entry.name), { recursive: true, force: true })
+    removed++
+  }
+  return removed
+}
+
+export async function hasSiteSyncChunkSession(
+  jobId: string,
+  key: string
+): Promise<boolean> {
+  return Boolean(await readSession(uploadIdFor(jobId, key)))
+}
+
+export function isSiteSyncChunkMetadata(
+  value: unknown
+): value is SiteSyncChunkMetadata {
+  if (!value || typeof value !== 'object') return false
+  const item = value as Record<string, unknown>
+  return typeof item.jobId === 'string'
+    && typeof item.mediaKey === 'string'
+    && typeof item.key === 'string'
+    && Number.isInteger(item.totalSize)
+    && Number(item.totalSize) > 0
+    && Number.isInteger(item.totalChunks)
+    && Number(item.totalChunks) > 0
+    && typeof item.originalName === 'string'
+    && typeof item.uploadedAt === 'string'
+    && typeof item.contentType === 'string'
+    && (item.mediaType === 'image'
+      || item.mediaType === 'video'
+      || item.mediaType === 'file')
+    && (item.conflictMode === 'skip' || item.conflictMode === 'overwrite')
+    && typeof item.backendId === 'string'
+}

server/utils/site-sync/runner.ts

@@ -41,6 +41,10 @@ import {
   completeSiteSyncMediaArtifacts,
   recordSiteSyncArtifact
 } from './artifacts'
+import {
+  SITE_SYNC_CHUNK_SIZE,
+  type SiteSyncChunkMetadata
+} from './chunks'
 import { toSiteSyncMediaTransferWithTags } from './manifest'
 import type {
   SiteSyncDirection,
@@ -69,7 +73,7 @@ export function registerSiteSyncToken(jobId: string, token: string): void {
 
 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
+const SITE_SYNC_CHUNK_RESPONSE_TIMEOUT_MS = 15 * 1000
 
 interface SiteSyncAgents {
   upload: Agent
@@ -81,7 +85,7 @@ const agents = (globalThis as typeof globalThis & {
 }).__pichostSiteSyncAgents ?? {
   upload: new Agent({
     connect: { timeout: SITE_SYNC_CONNECT_TIMEOUT_MS },
-    headersTimeout: SITE_SYNC_UPLOAD_HEADERS_TIMEOUT_MS,
+    headersTimeout: SITE_SYNC_CHUNK_RESPONSE_TIMEOUT_MS,
     bodyTimeout: SITE_SYNC_RESPONSE_IDLE_TIMEOUT_MS
   }),
   object: new Agent({
@@ -148,7 +152,7 @@ const SITE_SYNC_MAX_ATTEMPTS = 3
 export interface SiteSyncRetryStatus {
   attempt: number
   maxAttempts: number
-  phase: 'attempt' | 'countdown' | 'retry'
+  phase: 'attempt' | 'countdown' | 'active' | 'retry'
   delayMs?: number
   remainingMs?: number
   error?: Error
@@ -168,6 +172,11 @@ function formatRetryStatus(
     )
     return `${step}（第 ${status.attempt}/${status.maxAttempts} 次）· 连接倒计时 ${seconds}s`
   }
+  if (status.phase === 'active') {
+    return step.startsWith('上传对象')
+      ? `${step} · 已连接，上传中`
+      : `${step} · 连接已建立，等待响应`
+  }
   if ((status.delayMs ?? 0) < 1000) {
     return `${step}第 ${status.attempt}/${status.maxAttempts} 次无响应，即将重试：${status.error?.message ?? '连接失败'}`
   }
@@ -213,6 +222,12 @@ export async function retrySiteSyncOperation<T>(
           if (remainingMs <= 0) {
             if (countdownTimer) clearInterval(countdownTimer)
             countdownTimer = null
+            onStatus({
+              attempt,
+              maxAttempts: SITE_SYNC_MAX_ATTEMPTS,
+              phase: 'active',
+              remainingMs: 0
+            })
             return
           }
           onStatus({
@@ -385,6 +400,72 @@ export async function fetchSiteSyncJson<T>(
   )
 }
 
+async function uploadSiteSyncChunk(input: {
+  jobId: string
+  targetBaseUrl: string
+  token: string
+  metadata: SiteSyncChunkMetadata
+  index: number
+  bytes: Uint8Array
+  uploadedChunks: number
+  onProgress?: (message: string) => void
+}): Promise<{
+  received: number
+  totalChunks: number
+  complete: boolean
+}> {
+  const controller = new AbortController()
+  const jobSignal = getJobAbortSignal(input.jobId)
+  const onAbort = () => controller.abort(createJobCancelledError())
+  jobSignal.addEventListener('abort', onAbort, { once: true })
+  const timeout = setTimeout(() => {
+    controller.abort(new DOMException(
+      'Chunk upload made no progress for 15 seconds',
+      'TimeoutError'
+    ))
+  }, SITE_SYNC_CHUNK_RESPONSE_TIMEOUT_MS)
+  const startedAt = Date.now()
+  const fileName = basename(input.metadata.key)
+  const updateUploadTime = () => {
+    const seconds = Math.floor((Date.now() - startedAt) / 1000)
+    input.onProgress?.(
+      `上传对象 ${fileName} · 已连接，上传中 · 已上传 ${input.uploadedChunks}/${input.metadata.totalChunks} 块 · 等待 ${seconds}s`
+    )
+  }
+  updateUploadTime()
+  const ticker = setInterval(updateUploadTime, 1000)
+
+  try {
+    const encoded = Buffer.from(JSON.stringify(input.metadata))
+      .toString('base64url')
+    const response = await undiciFetch(
+      targetUrl(input.targetBaseUrl, '/api/site-sync/chunk'),
+      {
+        method: 'POST',
+        body: input.bytes,
+        headers: {
+          'Auth-Token': input.token,
+          'Content-Type': 'application/octet-stream',
+          'X-PicHost-Chunk-Metadata': encoded,
+          'X-PicHost-Chunk-Index': String(input.index)
+        },
+        signal: AbortSignal.any([jobSignal, controller.signal]),
+        dispatcher: agents.upload
+      }
+    )
+    if (!response.ok) throw await responseError(response)
+    return await response.json() as {
+      received: number
+      totalChunks: number
+      complete: boolean
+    }
+  } finally {
+    clearTimeout(timeout)
+    clearInterval(ticker)
+    jobSignal.removeEventListener('abort', onAbort)
+  }
+}
+
 async function uploadObject(
   jobId: string,
   options: SiteSyncOptions,
@@ -392,25 +473,107 @@ async function uploadObject(
   backend: StorageBackend,
   row: ImageIndexRow,
   ownerUsername: string | null,
-  key: string
+  key: string,
+  onProgress?: (message: string) => void
 ): Promise<void> {
   const stored = await backend.head(key)
   if (!stored) throw new Error(`源文件不存在：${key}`)
+  const isPrimary = key === row.key
+  const originalName = isPrimary ? row.original_name : basename(key)
+  const contentType = isPrimary ? row.content_type : contentTypeFromKey(key)
+  const mediaType = isPrimary
+    ? row.media_type
+    : key.startsWith('videos/')
+      ? 'video'
+      : key.startsWith('files/')
+        ? 'file'
+        : 'image'
+
+  if (stored.size > SITE_SYNC_CHUNK_SIZE) {
+    const totalChunks = Math.ceil(stored.size / SITE_SYNC_CHUNK_SIZE)
+    const metadata: SiteSyncChunkMetadata = {
+      jobId,
+      mediaKey: row.key,
+      key,
+      totalSize: stored.size,
+      totalChunks,
+      ownerUsername,
+      originalName,
+      uploadedAt: row.uploaded_at,
+      contentType,
+      mediaType,
+      conflictMode: options.conflictMode,
+      backendId: backend.id,
+      updatedAt: new Date().toISOString()
+    }
+    const status = await fetchSiteSyncJson<{
+      received: number[]
+      totalChunks: number
+    }>(
+      targetUrl(
+        options.targetBaseUrl,
+        `/api/site-sync/chunk-status?jobId=${encodeURIComponent(jobId)}&key=${encodeURIComponent(key)}`
+      ),
+      token,
+      undefined,
+      getJobAbortSignal(jobId),
+      (onStatus) => {
+        onProgress?.(
+          formatRetryStatus(`检查对象 ${basename(key)} 分块进度`, onStatus)
+        )
+      }
+    )
+    const received = new Set(status.received)
+    for (let index = 0; index < totalChunks; index++) {
+      if (received.has(index)) continue
+      throwIfJobCancellationRequested(jobId)
+      const start = index * SITE_SYNC_CHUNK_SIZE
+      const end = Math.min(stored.size, start + SITE_SYNC_CHUNK_SIZE) - 1
+      const bytes = await streamToBuffer(
+        await backend.createStream(key, { start, end }),
+        getJobAbortSignal(jobId)
+      )
+      const result = await uploadSiteSyncChunk({
+        jobId,
+        targetBaseUrl: options.targetBaseUrl,
+        token,
+        metadata,
+        index,
+        bytes,
+        uploadedChunks: received.size,
+        onProgress
+      })
+      received.add(index)
+      onProgress?.(
+        `上传对象 ${basename(key)} · 已连接，上传中 · 已上传 ${result.received}/${result.totalChunks} 块`
+      )
+    }
+    await fetchSiteSyncJsonOnce<{ success: boolean }>(
+      targetUrl(options.targetBaseUrl, '/api/site-sync/chunk-finalize'),
+      token,
+      {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ jobId, key }),
+        signal: getJobAbortSignal(jobId)
+      }
+    )
+    return
+  }
+
   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('originalName', originalName)
   form.append('uploadedAt', row.uploaded_at)
-  const contentType = isPrimary ? row.content_type : contentTypeFromKey(key)
   form.append('contentType', contentType)
   form.append(
     'file',
@@ -640,7 +803,10 @@ async function runSiteSyncPull(
         () => processBundleOnce(
           bundle,
           isRetry,
-          (value) => { step = value }
+          (value) => {
+            step = value
+            updateLiveStatus(value)
+          }
         ),
         jobSignal,
         status => updateLiveStatus(step, status)
@@ -934,7 +1100,8 @@ export async function runSiteSyncJob(
           backend,
           row,
           ownerUsername,
-          key
+          key,
+          message => setStep(message)
         )
       }
 
@@ -968,7 +1135,10 @@ export async function runSiteSyncJob(
         () => processPlanOnce(
           plan,
           isRetry,
-          (value) => { step = value }
+          (value) => {
+            step = value
+            updateLiveStatus(value)
+          }
         ),
         jobSignal,
         status => updateLiveStatus(step, status)