diff --git a/app/components/storage/BackupJobTable.vue b/app/components/storage/BackupJobTable.vue index 2c419b3..22f8374 100644 --- a/app/components/storage/BackupJobTable.vue +++ b/app/components/storage/BackupJobTable.vue @@ -8,6 +8,7 @@ defineProps<{ totalPages: number total: number pageSize: number + showHidden: boolean loading?: boolean }>() @@ -15,6 +16,9 @@ const emit = defineEmits<{ 'retry': [jobId: string] 'download': [jobId: string] 'cancel': [jobId: string] + 'hide': [jobId: string, hidden: boolean] + 'hideAll': [hidden: boolean] + 'toggleHidden': [] 'view': [job: BackupJobItem] 'update:page': [page: number] 'update:pageSize': [pageSize: number] @@ -85,6 +89,10 @@ function canView(job: BackupJobItem): boolean { return job.type === 'full_restore' && job.status === 'done' && Boolean(job.message) } +function canHide(job: BackupJobItem): boolean { + return job.status === 'done' || job.status === 'failed' || job.hidden +} + function formatTime(value: string) { if (!value) return '—' return new Date(value).toLocaleString(locale.value, { hour12: false }) @@ -102,11 +110,33 @@ function formatTime(value: string) { {{ t('storage.backup.jobsSubtitle') }}

- +
+ + + +
+
diff --git a/app/components/storage/StorageBackupPanel.vue b/app/components/storage/StorageBackupPanel.vue index f291399..559e2a1 100644 --- a/app/components/storage/StorageBackupPanel.vue +++ b/app/components/storage/StorageBackupPanel.vue @@ -19,6 +19,7 @@ const { jobsTotal, jobsTotalPages, jobsLoading, + showHiddenJobs, estimate, packages, loadingEstimate, @@ -26,6 +27,9 @@ const { hasActiveJob, getActiveJob, loadJobs, + toggleShowHiddenJobs, + hideJob, + hideAllJobs, setJobsPage, setJobsPageSize, loadEstimate, @@ -98,6 +102,14 @@ async function onCancel(jobId: string) { await cancelJob(jobId) } +async function onHideJob(jobId: string, hidden: boolean) { + await hideJob(jobId, hidden) +} + +async function onHideAll(hidden: boolean) { + await hideAllJobs(hidden) +} + async function onSync(body: { fromBackendId: string toBackendId: string @@ -192,11 +204,15 @@ function onViewJob(job: BackupJobItem) { :total="jobsTotal" :page-size="jobsPageSize" :loading="jobsLoading" + :show-hidden="showHiddenJobs" @update:page="setJobsPage" @update:page-size="setJobsPageSize" @download="onDownload" @retry="onRetry" @cancel="onCancel" + @hide="onHideJob" + @hide-all="onHideAll" + @toggle-hidden="toggleShowHiddenJobs" @view="onViewJob" /> diff --git a/app/composables/useBackup.ts b/app/composables/useBackup.ts index 8a271a7..68799b2 100644 --- a/app/composables/useBackup.ts +++ b/app/composables/useBackup.ts @@ -13,6 +13,7 @@ const jobsPageSize = ref(10) const jobsTotal = ref(0) const jobsTotalPages = ref(1) const jobsLoading = ref(false) +const showHiddenJobs = ref(false) const activeJobState = ref(null) const estimate = ref(null) const packages = ref([]) @@ -69,7 +70,8 @@ export function useBackup() { credentials: 'include', query: { pageNum: jobsPage.value, - pageSize: jobsPageSize.value + pageSize: jobsPageSize.value, + includeHidden: showHiddenJobs.value } }) jobs.value = data.jobs @@ -104,6 +106,58 @@ export function useBackup() { return loadJobs() } + async function toggleShowHiddenJobs() { + showHiddenJobs.value = !showHiddenJobs.value + jobsPage.value = 1 + await loadJobs() + } + + async function hideJob(jobId: string, hidden = true) { + try { + await $fetch('/api/admin/backup/jobs/hide', { + method: 'POST', + credentials: 'include', + body: { jobId, hidden } + }) + if (hidden) { + showHiddenJobs.value = false + jobsPage.value = 1 + } + await loadJobs() + toast.add({ + title: hidden + ? t('storage.backup.jobHidden') + : t('storage.backup.jobUnhidden'), + color: 'success' + }) + } catch (error: unknown) { + handleError(error, t('storage.backup.hideFailed')) + } + } + + async function hideAllJobs(hidden = true) { + try { + await $fetch('/api/admin/backup/jobs/hide-all', { + method: 'POST', + credentials: 'include', + body: { hidden } + }) + if (hidden) { + showHiddenJobs.value = false + } + jobsPage.value = 1 + await loadJobs() + toast.add({ + title: hidden + ? t('storage.backup.jobsHiddenAll') + : t('storage.backup.jobsUnhiddenAll'), + color: 'success' + }) + } catch (error: unknown) { + handleError(error, t('storage.backup.hideFailed')) + } + } + async function loadEstimate() { loadingEstimate.value = true try { @@ -327,6 +381,7 @@ export function useBackup() { jobsTotal, jobsTotalPages, jobsLoading, + showHiddenJobs, estimate, packages, loadingEstimate, @@ -334,6 +389,9 @@ export function useBackup() { hasActiveJob, getActiveJob, loadJobs, + toggleShowHiddenJobs, + hideJob, + hideAllJobs, setJobsPage, setJobsPageSize, loadEstimate, diff --git a/app/types/backup.ts b/app/types/backup.ts index 24dc4ed..d290051 100644 --- a/app/types/backup.ts +++ b/app/types/backup.ts @@ -16,6 +16,7 @@ export interface BackupJobItem { error: string | null createdAt: string updatedAt: string + hidden: boolean finishedAt: string | null } diff --git a/i18n/locales/en.json b/i18n/locales/en.json index 65d5e4d..2eaf925 100644 --- a/i18n/locales/en.json +++ b/i18n/locales/en.json @@ -738,6 +738,17 @@ "jobsSubtitle": "Track backup, restore, and migration progress", "jobsEmpty": "No backup or migration jobs yet", "jobsEmptyHint": "Progress appears here after export, restore, or sync starts", + "jobsHideAll": "Hide all", + "jobsRestoreAll": "Restore all", + "hiddenJobsShow": "Show hidden", + "hiddenJobsHide": "Show visible only", + "jobHide": "Hide from list (keeps the record)", + "jobUnhide": "Show again", + "jobHidden": "Job hidden; its record is still retained", + "jobUnhidden": "Job is visible again", + "jobsHiddenAll": "All finished jobs hidden; records are still retained", + "jobsUnhiddenAll": "All hidden jobs are visible again", + "hideFailed": "Failed to update job visibility", "jobsUnit": "jobs", "colTask": "Task", "colType": "Type", diff --git a/i18n/locales/zh-CN.json b/i18n/locales/zh-CN.json index a50d2f3..ea3b8e8 100644 --- a/i18n/locales/zh-CN.json +++ b/i18n/locales/zh-CN.json @@ -738,6 +738,17 @@ "jobsSubtitle": "查看备份、恢复与迁移进度", "jobsEmpty": "暂无备份或迁移任务", "jobsEmptyHint": "导出、恢复或同步开始后，进度会显示在这里", + "jobsHideAll": "全部清除", + "jobsRestoreAll": "恢复全部", + "hiddenJobsShow": "显示隐藏记录", + "hiddenJobsHide": "只显示未关闭", + "jobHide": "关闭显示（不删除记录）", + "jobUnhide": "恢复显示", + "jobHidden": "任务已关闭显示，记录仍保留", + "jobUnhidden": "任务已恢复显示", + "jobsHiddenAll": "已关闭全部历史任务的显示，记录仍保留", + "jobsUnhiddenAll": "已恢复显示全部历史任务", + "hideFailed": "更新任务显示状态失败", "jobsUnit": "条任务", "colTask": "任务", "colType": "类型", diff --git a/server/api/admin/backup/export.post.ts b/server/api/admin/backup/export.post.ts index 3174818..4102c07 100644 --- a/server/api/admin/backup/export.post.ts +++ b/server/api/admin/backup/export.post.ts @@ -3,11 +3,14 @@ import { createApiError } from '../../../utils/api-error' import { serializeBackupJob } from '../../../utils/backup/api' import { getBackupExportEstimate } from '../../../utils/backup/stats' import { startFullExportJob } from '../../../utils/backup/runner' +import { resolveRuntimeAppVersion } from '../../../utils/app-version' export default defineEventHandler(async (event) => { const user = await requireAdminAuth(event) const config = useRuntimeConfig(event) - const appVersion = String(config.appVersion ?? '0.0.0') + const appVersion = resolveRuntimeAppVersion( + String(config.appVersion ?? '0.0.0') + ) const estimate = await getBackupExportEstimate() diff --git a/server/api/admin/backup/jobs/hide-all.post.ts b/server/api/admin/backup/jobs/hide-all.post.ts new file mode 100644 index 0000000..9f5683d --- /dev/null +++ b/server/api/admin/backup/jobs/hide-all.post.ts @@ -0,0 +1,14 @@ +import { requireAdminAuth } from '../../../../utils/access' +import { setAllFinishedBackupJobsHidden } from '../../../../utils/backup/jobs' + +interface Body { + hidden?: boolean +} + +export default defineEventHandler(async (event) => { + const user = await requireAdminAuth(event) + const body = await readBody(event).catch(() => null) + const hidden = body?.hidden !== false + const updated = setAllFinishedBackupJobsHidden(user.id, hidden) + return { success: true, hidden, updated } +}) diff --git a/server/api/admin/backup/jobs/hide.post.ts b/server/api/admin/backup/jobs/hide.post.ts new file mode 100644 index 0000000..db1c979 --- /dev/null +++ b/server/api/admin/backup/jobs/hide.post.ts @@ -0,0 +1,25 @@ +import { requireAdminAuth } from '../../../../utils/access' +import { createApiError } from '../../../../utils/api-error' +import { + getBackupJob, + setBackupJobHidden +} from '../../../../utils/backup/jobs' + +interface Body { + jobId?: string + hidden?: boolean +} + +export default defineEventHandler(async (event) => { + const user = await requireAdminAuth(event) + const body = await readBody(event).catch(() => null) + const jobId = body?.jobId?.trim() ?? '' + if (!jobId || !getBackupJob(jobId, user.id)) { + createApiError(event, 'INVALID_REQUEST', '任务不存在', 404) + } + const hidden = body?.hidden !== false + if (!setBackupJobHidden(jobId, user.id, hidden)) { + createApiError(event, 'INVALID_REQUEST', '更新任务显示状态失败', 400) + } + return { success: true, hidden } +}) diff --git a/server/api/admin/backup/jobs/index.get.ts b/server/api/admin/backup/jobs/index.get.ts index 0c1e8db..a516716 100644 --- a/server/api/admin/backup/jobs/index.get.ts +++ b/server/api/admin/backup/jobs/index.get.ts @@ -14,14 +14,21 @@ export default defineEventHandler(async (event) => { recoverInterruptedBackupJobs() } const query = getQuery(event) + const includeHidden = query.includeHidden === 'true' + || query.includeHidden === '1' const pageSize = Math.min(50, Math.max(1, Number(query.pageSize) || 10)) const pageNum = Math.max(1, Number(query.pageNum) || Number(query.page) || 1) - const { jobs, total } = listBackupJobsPage(pageNum, pageSize, user.id) + const { jobs, total } = listBackupJobsPage( + pageNum, + pageSize, + user.id, + includeHidden + ) const totalPages = Math.max(1, Math.ceil(total / pageSize)) const page = total > 0 ? Math.min(pageNum, totalPages) : 1 const slice = page === pageNum ? jobs - : listBackupJobsPage(page, pageSize, user.id).jobs + : listBackupJobsPage(page, pageSize, user.id, includeHidden).jobs const activeJob = getActiveBackupJob(user.id) @@ -31,6 +38,7 @@ export default defineEventHandler(async (event) => { page, pageSize, totalPages, + includeHidden, activeJob: activeJob ? serializeBackupJob(activeJob) : null } }) diff --git a/server/api/health.get.ts b/server/api/health.get.ts index b4dc01c..7138403 100644 --- a/server/api/health.get.ts +++ b/server/api/health.get.ts @@ -1,7 +1,9 @@ +import { resolveRuntimeAppVersion } from '../utils/app-version' + export default defineEventHandler((event) => { const config = useRuntimeConfig(event) return { status: 'ok', - version: config.appVersion + version: resolveRuntimeAppVersion(String(config.appVersion ?? '')) } }) diff --git a/server/api/site-sync/chunk-finalize.post.ts b/server/api/site-sync/chunk-finalize.post.ts new file mode 100644 index 0000000..f343601 --- /dev/null +++ b/server/api/site-sync/chunk-finalize.post.ts @@ -0,0 +1,29 @@ +import { requireSiteSyncAuth } from '../../utils/access' +import { createApiError } from '../../utils/api-error' +import { validateStorageKey } from '../../utils/image-key' +import { finalizeSiteSyncChunk } from '../../utils/site-sync/chunks' + +interface Body { + jobId?: string + key?: string +} + +export default defineEventHandler(async (event) => { + await requireSiteSyncAuth(event) + const body = await readBody(event).catch(() => null) + const jobId = body?.jobId?.trim() ?? '' + const key = body?.key?.trim() ?? '' + if (!jobId || !validateStorageKey(key)) { + createApiError(event, 'INVALID_REQUEST', '分块完成参数无效', 400) + } + try { + return await finalizeSiteSyncChunk({ jobId, key }) + } catch (error) { + createApiError( + event, + 'INVALID_REQUEST', + error instanceof Error ? error.message : '分块组装失败', + 400 + ) + } +}) diff --git a/server/api/site-sync/chunk-status.get.ts b/server/api/site-sync/chunk-status.get.ts new file mode 100644 index 0000000..245230c --- /dev/null +++ b/server/api/site-sync/chunk-status.get.ts @@ -0,0 +1,16 @@ +import { getQuery } from 'h3' +import { requireSiteSyncAuth } from '../../utils/access' +import { createApiError } from '../../utils/api-error' +import { validateStorageKey } from '../../utils/image-key' +import { getSiteSyncChunkStatus } from '../../utils/site-sync/chunks' + +export default defineEventHandler(async (event) => { + await requireSiteSyncAuth(event) + const query = getQuery(event) + const jobId = typeof query.jobId === 'string' ? query.jobId.trim() : '' + const key = typeof query.key === 'string' ? query.key.trim() : '' + if (!jobId || !validateStorageKey(key)) { + createApiError(event, 'INVALID_REQUEST', '分块查询参数无效', 400) + } + return getSiteSyncChunkStatus({ jobId, key }) +}) diff --git a/server/api/site-sync/chunk.post.ts b/server/api/site-sync/chunk.post.ts new file mode 100644 index 0000000..1093286 --- /dev/null +++ b/server/api/site-sync/chunk.post.ts @@ -0,0 +1,54 @@ +import { getHeader, readRawBody } from 'h3' +import { requireSiteSyncAuth } from '../../utils/access' +import { createApiError } from '../../utils/api-error' +import { validateStorageKey } from '../../utils/image-key' +import { + isSiteSyncChunkMetadata, + SITE_SYNC_CHUNK_SIZE, + writeSiteSyncChunk +} from '../../utils/site-sync/chunks' + +export default defineEventHandler(async (event) => { + await requireSiteSyncAuth(event) + const encoded = getHeader(event, 'x-pichost-chunk-metadata') ?? '' + let metadata: unknown + try { + metadata = JSON.parse(Buffer.from(encoded, 'base64url').toString('utf8')) + } catch { + createApiError(event, 'INVALID_REQUEST', '分块元数据无效', 400) + } + if ( + !isSiteSyncChunkMetadata(metadata) + || !validateStorageKey(metadata.key) + || !validateStorageKey(metadata.mediaKey) + ) { + createApiError(event, 'INVALID_REQUEST', '分块元数据无效', 400) + } + + const index = Number.parseInt(getHeader(event, 'x-pichost-chunk-index') ?? '', 10) + if (!Number.isInteger(index)) { + createApiError(event, 'INVALID_REQUEST', '分块序号无效', 400) + } + const body = await readRawBody(event, false) + if (!body?.length) { + createApiError(event, 'INVALID_REQUEST', '未收到分块数据', 400) + } + if (body.length > SITE_SYNC_CHUNK_SIZE) { + createApiError(event, 'FILE_TOO_LARGE', '分块超过 4 MB', 413) + } + + try { + return await writeSiteSyncChunk({ + metadata, + index, + bytes: new Uint8Array(body) + }) + } catch (error) { + createApiError( + event, + 'INVALID_REQUEST', + error instanceof Error ? error.message : '分块写入失败', + 400 + ) + } +}) diff --git a/server/api/site-sync/cleanup.post.ts b/server/api/site-sync/cleanup.post.ts index 10fd834..e178718 100644 --- a/server/api/site-sync/cleanup.post.ts +++ b/server/api/site-sync/cleanup.post.ts @@ -1,6 +1,7 @@ import { requireSiteSyncAuth } from '../../utils/access' import { createApiError } from '../../utils/api-error' import { cleanupSiteSyncJobArtifacts } from '../../utils/site-sync/artifacts' +import { cleanupSiteSyncChunkSessions } from '../../utils/site-sync/chunks' interface Body { jobId?: string @@ -13,8 +14,10 @@ export default defineEventHandler(async (event) => { if (!jobId || jobId.length > 128) { createApiError(event, 'INVALID_REQUEST', '缺少同步任务 ID', 400) } + const removedChunkSessions = await cleanupSiteSyncChunkSessions(jobId) return { success: true, + removedChunkSessions, ...await cleanupSiteSyncJobArtifacts(jobId) } }) diff --git a/server/api/user/api-token.get.ts b/server/api/user/api-token.get.ts index 5186fb6..71ff70b 100644 --- a/server/api/user/api-token.get.ts +++ b/server/api/user/api-token.get.ts @@ -7,6 +7,7 @@ import { getVideoBaseUrl, isHideFolderInUrl } from '../../utils/env' +import { resolveRuntimeAppVersion } from '../../utils/app-version' export default defineEventHandler(async (event) => { const user = await requireUserAuth(event) @@ -24,7 +25,7 @@ export default defineEventHandler(async (event) => { videoBaseUrl: getVideoBaseUrl(event), fileBaseUrl: getFileBaseUrl(event), hideFolderInUrl: isHideFolderInUrl(event), - appVersion: config.appVersion as string + appVersion: resolveRuntimeAppVersion(String(config.appVersion ?? '')) } } }) diff --git a/server/api/version/latest.get.ts b/server/api/version/latest.get.ts index 044151c..892a38b 100644 --- a/server/api/version/latest.get.ts +++ b/server/api/version/latest.get.ts @@ -1,14 +1,17 @@ import { requireAdminAuth } from '../../utils/access' import { fetchLatestGitHubRelease, - isVersionNewer + isVersionNewer, + resolveRuntimeAppVersion } from '../../utils/app-version' export default defineEventHandler(async (event) => { await requireAdminAuth(event) const config = useRuntimeConfig(event) - const currentVersion = String(config.appVersion ?? '') + const currentVersion = resolveRuntimeAppVersion( + String(config.appVersion ?? '') + ) const query = getQuery(event) const force = query.refresh === '1' || query.refresh === 'true' const latest = await fetchLatestGitHubRelease(force) diff --git a/server/plugins/startup.ts b/server/plugins/startup.ts index 72a99b7..05ceeb4 100644 --- a/server/plugins/startup.ts +++ b/server/plugins/startup.ts @@ -8,6 +8,7 @@ import { ensureDefaultBackends } from '../utils/storage-backends' import { cleanupLocalAtomicTempFiles } from '../utils/storage/local' import { cleanupExpiredSiteSyncArtifacts } from '../utils/site-sync/artifacts' import { recoverInterruptedBackupJobs } from '../utils/backup/jobs' +import { cleanupExpiredSiteSyncChunkSessions } from '../utils/site-sync/chunks' import { cleanupTranscodedSources, killOrphanFfmpeg, @@ -49,6 +50,17 @@ export default defineNitroPlugin(() => { } })() + void (async () => { + try { + const removed = await cleanupExpiredSiteSyncChunkSessions() + if (removed > 0) { + logInfo('清理过期站点同步分块', { removed }) + } + } catch (error) { + logException('清理站点同步分块失败', error) + } + })() + void (async () => { try { const result = await cleanupExpiredSiteSyncArtifacts() diff --git a/server/utils/app-version.ts b/server/utils/app-version.ts index 44fdc5b..9873d6a 100644 --- a/server/utils/app-version.ts +++ b/server/utils/app-version.ts @@ -1,3 +1,6 @@ +import { readFileSync, statSync } from 'node:fs' +import { join } from 'node:path' + const GITHUB_REPO = 'O96u/PicHost' const RELEASES_LATEST_URL = `https://api.github.com/repos/${GITHUB_REPO}/releases/latest` const CACHE_TTL_MS = 60 * 60 * 1000 @@ -8,6 +11,26 @@ export interface LatestReleaseInfo { } let cachedRelease: { fetchedAt: number, value: LatestReleaseInfo | null } | null = null +let cachedPackageVersion: { mtimeMs: number, version: string } | null = null + +/** Read package.json at request time so a hot-reloaded version update is visible without restart. */ +export function resolveRuntimeAppVersion(fallback: string): string { + const path = join(process.cwd(), 'package.json') + try { + const mtimeMs = statSync(path).mtimeMs + if (cachedPackageVersion?.mtimeMs === mtimeMs) { + return cachedPackageVersion.version + } + const pkg = JSON.parse(readFileSync(path, 'utf8')) as { version?: unknown } + const version = typeof pkg.version === 'string' && pkg.version + ? pkg.version + : fallback + cachedPackageVersion = { mtimeMs, version } + return version + } catch { + return fallback + } +} export function normalizeVersionTag(tag: string): string { return tag.trim().replace(/^v/i, '') diff --git a/server/utils/backup/api.ts b/server/utils/backup/api.ts index 02b5304..fe9f409 100644 --- a/server/utils/backup/api.ts +++ b/server/utils/backup/api.ts @@ -27,6 +27,7 @@ export function serializeBackupJob(job: BackupJob) { error: job.error, createdAt: job.createdAt, updatedAt: job.updatedAt, + hidden: job.hidden, finishedAt: job.finishedAt } } diff --git a/server/utils/backup/jobs.ts b/server/utils/backup/jobs.ts index 8da3227..47db55e 100644 --- a/server/utils/backup/jobs.ts +++ b/server/utils/backup/jobs.ts @@ -18,6 +18,7 @@ export interface BackupJob { error: string | null createdAt: string updatedAt: string + hidden: boolean finishedAt: string | null } @@ -34,6 +35,7 @@ interface BackupJobRow { error: string | null created_at: string updated_at: string + hidden: number finished_at: string | null } @@ -58,6 +60,7 @@ function rowToJob(row: BackupJobRow): BackupJob { error: row.error, createdAt: row.created_at, updatedAt: row.updated_at, + hidden: Number(row.hidden) === 1, finishedAt: row.finished_at } } @@ -72,7 +75,7 @@ export function getRunningBackupJob(userId?: number): BackupJob | null { ensureBackupSchema() const row = getDb().prepare(` SELECT id, user_id, type, status, progress, total, message, payload_json, - output_path, error, created_at, updated_at, finished_at + output_path, error, created_at, updated_at, hidden, finished_at FROM backup_jobs WHERE status = 'running'${userId === undefined ? '' : ' AND user_id = ?'} LIMIT 1 @@ -84,7 +87,7 @@ export function getActiveBackupJob(userId?: number): BackupJob | null { ensureBackupSchema() const row = getDb().prepare(` SELECT id, user_id, type, status, progress, total, message, payload_json, - output_path, error, created_at, updated_at, finished_at + output_path, error, created_at, updated_at, hidden, finished_at FROM backup_jobs WHERE status IN ('pending', 'running')${userId === undefined ? '' : ' AND user_id = ?'} ORDER BY created_at DESC @@ -128,22 +131,33 @@ export function getBackupJob(id: string, userId?: number): BackupJob | null { ensureBackupSchema() const row = getDb().prepare(` SELECT id, user_id, type, status, progress, total, message, payload_json, - output_path, error, created_at, updated_at, finished_at + output_path, error, created_at, updated_at, hidden, finished_at FROM backup_jobs WHERE id = ?${userId === undefined ? '' : ' AND user_id = ?'} `).get(...(userId === undefined ? [id] : [id, userId])) as BackupJobRow | undefined return row ? rowToJob(row) : null } -export function listBackupJobs(limit = 20, userId?: number): BackupJob[] { - return listBackupJobsPage(1, limit, userId).jobs +export function listBackupJobs( + limit = 20, + userId?: number, + includeHidden = false +): BackupJob[] { + return listBackupJobsPage(1, limit, userId, includeHidden).jobs } -export function countBackupJobs(userId?: number): number { +export function countBackupJobs( + userId?: number, + includeHidden = false +): number { ensureBackupSchema() + const where = [ + ...(userId === undefined ? [] : ['user_id = ?']), + ...(includeHidden ? [] : ['hidden = 0']) + ] const row = getDb().prepare(` SELECT COUNT(*) AS count FROM backup_jobs - ${userId === undefined ? '' : 'WHERE user_id = ?'} + ${where.length ? `WHERE ${where.join(' AND ')}` : ''} `).get(...(userId === undefined ? [] : [userId])) as { count: number } return row.count } @@ -151,18 +165,23 @@ export function countBackupJobs(userId?: number): number { export function listBackupJobsPage( page = 1, pageSize = 10, - userId?: number + userId?: number, + includeHidden = false ): { jobs: BackupJob[], total: number } { ensureBackupSchema() const limit = Math.min(50, Math.max(1, pageSize)) const safePage = Math.max(1, page) const offset = (safePage - 1) * limit - const total = countBackupJobs(userId) + const total = countBackupJobs(userId, includeHidden) + const where = [ + ...(userId === undefined ? [] : ['user_id = ?']), + ...(includeHidden ? [] : ['hidden = 0']) + ] const rows = getDb().prepare(` SELECT id, user_id, type, status, progress, total, message, payload_json, - output_path, error, created_at, updated_at, finished_at + output_path, error, created_at, updated_at, hidden, finished_at FROM backup_jobs - ${userId === undefined ? '' : 'WHERE user_id = ?'} + ${where.length ? `WHERE ${where.join(' AND ')}` : ''} ORDER BY created_at DESC LIMIT ? OFFSET ? `).all(...(userId === undefined ? [limit, offset] : [userId, limit, offset])) as unknown as BackupJobRow[] @@ -173,7 +192,7 @@ export function listAllBackupJobsForUser(userId: number): BackupJob[] { ensureBackupSchema() const rows = getDb().prepare(` SELECT id, user_id, type, status, progress, total, message, payload_json, - output_path, error, created_at, updated_at, finished_at + output_path, error, created_at, updated_at, hidden, finished_at FROM backup_jobs WHERE user_id = ? ORDER BY created_at DESC @@ -201,8 +220,8 @@ export function restoreBackupJobsAfterReplace(preserved: BackupJob[]): void { db.prepare('DELETE FROM backup_jobs').run() const insert = db.prepare(` INSERT INTO backup_jobs - (id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, updated_at, finished_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + (id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, updated_at, hidden, finished_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `) for (const job of preserved) { insert.run( @@ -218,6 +237,7 @@ export function restoreBackupJobsAfterReplace(preserved: BackupJob[]): void { job.error, job.createdAt, job.updatedAt, + job.hidden ? 1 : 0, job.finishedAt ) } @@ -305,6 +325,34 @@ export function updateBackupJobPayload( `).run(JSON.stringify(payload), id) } +export function setBackupJobHidden( + id: string, + userId: number, + hidden: boolean +): boolean { + ensureBackupSchema() + const result = getDb().prepare(` + UPDATE backup_jobs + SET hidden = ? + WHERE id = ? AND user_id = ? + `).run(hidden ? 1 : 0, id, userId) + return Number(result.changes ?? 0) > 0 +} + +export function setAllFinishedBackupJobsHidden( + userId: number, + hidden: boolean +): number { + ensureBackupSchema() + const result = getDb().prepare(` + UPDATE backup_jobs + SET hidden = ? + WHERE user_id = ? + AND status IN ('done', 'failed') + `).run(hidden ? 1 : 0, userId) + return Number(result.changes ?? 0) +} + export function markJobRunning(id: string, message?: string): void { updateBackupJob(id, { status: 'running', diff --git a/server/utils/backup/schema.ts b/server/utils/backup/schema.ts index b850c80..834c255 100644 --- a/server/utils/backup/schema.ts +++ b/server/utils/backup/schema.ts @@ -25,11 +25,12 @@ function migrateBackupJobTypes(): void { error TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + hidden INTEGER NOT NULL DEFAULT 0, finished_at TEXT ); INSERT INTO backup_jobs_new - (id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, updated_at, finished_at) - SELECT id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, COALESCE(updated_at, created_at), finished_at + (id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, updated_at, hidden, finished_at) + SELECT id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, COALESCE(updated_at, created_at), hidden, finished_at FROM backup_jobs; DROP TABLE backup_jobs; ALTER TABLE backup_jobs_new RENAME TO backup_jobs; @@ -55,6 +56,7 @@ export function ensureBackupSchema(): void { error TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + hidden INTEGER NOT NULL DEFAULT 0, finished_at TEXT ); CREATE INDEX IF NOT EXISTS idx_backup_jobs_created_at @@ -70,6 +72,9 @@ export function ensureBackupSchema(): void { if (!columns.some(column => column.name === 'updated_at')) { getDb().exec('ALTER TABLE backup_jobs ADD COLUMN updated_at TEXT') } + if (!columns.some(column => column.name === 'hidden')) { + getDb().exec('ALTER TABLE backup_jobs ADD COLUMN hidden INTEGER NOT NULL DEFAULT 0') + } getDb().exec(` CREATE INDEX IF NOT EXISTS idx_backup_jobs_user_created ON backup_jobs(user_id, created_at DESC); diff --git a/server/utils/env.ts b/server/utils/env.ts index 2648c0f..fc506d8 100644 --- a/server/utils/env.ts +++ b/server/utils/env.ts @@ -26,6 +26,7 @@ import { isAllowRegistration } from './db' import { getUploadPolicySettingsPayload } from './upload-policy' +import { resolveRuntimeAppVersion } from './app-version' export type FileBedAccessMode = 'public' | 'token' @@ -806,7 +807,7 @@ export function getSettingsPayload(event: H3Event) { const effectiveVideoBaseUrl = getVideoBaseUrl(event) const effectiveFileBaseUrl = getFileBaseUrl(event) const autoDeleteDays = getAutoDeleteDays(event) - const appVersion = config.appVersion as string + const appVersion = resolveRuntimeAppVersion(String(config.appVersion ?? '')) const domainSeparation = isDomainSeparationActive(event) const hideFolderInUrl = isHideFolderInUrl(event) const hideFolderInUrlSource = getHideFolderInUrlSource() diff --git a/server/utils/site-sync/chunks.ts b/server/utils/site-sync/chunks.ts new file mode 100644 index 0000000..ebb26d0 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 + 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' +} diff --git a/server/utils/site-sync/runner.ts b/server/utils/site-sync/runner.ts index 05bcedf..de4b052 100644 --- a/server/utils/site-sync/runner.ts +++ b/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( 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( ) } +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 { 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) diff --git a/server/utils/storage/local.ts b/server/utils/storage/local.ts index d3dec66..e945e3f 100644 --- a/server/utils/storage/local.ts +++ b/server/utils/storage/local.ts @@ -151,3 +151,27 @@ export async function cleanupLocalAtomicTempFiles( } return removed } + +/** Move an already-complete file into the storage tree without loading it into memory. */ +export async function putLocalFileAtomic( + key: string, + sourcePath: string +): Promise { + const filePath = keyToFilePath(key) + await fs.mkdir(dirname(filePath), { recursive: true }) + try { + await fs.rename(sourcePath, filePath) + } catch (error) { + if ( + error + && typeof error === 'object' + && 'code' in error + && error.code === 'EXDEV' + ) { + await fs.copyFile(sourcePath, filePath) + await fs.rm(sourcePath, { force: true }) + return + } + throw error + } +}