Yeomai/FileHost
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

feat: 完善媒体管理与备份迁移流程

Browse Source 
- 视频上传生成缩略图并提供占位封面和失败原因展示
- 支持 HLS 按需重新封装 MP4，并限制临时文件生命周期与访问权限
- 完善视频、文件 API 文档、搜索、批量删除和单次转码质量参数
- 修复备份恢复与跨后端迁移遗漏缩略图、HLS 清单和分片的问题
- 将备份任务、归档和下载权限按用户隔离
- 增加仅清理当前用户数据的高危存储清空流程和二次文字确认
- 修复 S3 中文文件名元数据及 CLI 数据库连接问题
- 补充备份恢复、用户隔离、媒体资源和转码相关测试
This commit is contained in:
tianmo committed 2026-10-17 13:46:02 +08:00
1 parent 9ebab8270c
commit a64be50b73
60 files changed
+2703 -325
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/utils/backup/backend-sync.ts
+51 -40
View File
@@ -1,18 +1,22 @@
import { promises as fs } from 'node:fs'
import { join } from 'node:path'
import { basename, join } from 'node:path'
import {
createBackendInstance,
getStorageBackendRow,
setDefaultBackend
} from '../storage-backends'
import {
getImageIndexAsStored,
getImageIndexRow,
listImageIndexRowsByBackend,
updateImageBackendId
} from '../image-index'
import { createImageStream } from '../storage'
import { contentTypeFromKey } from '../content-type'
import { mediaTypeForKey } from '../image-key'
import type { ImageIndexRow } from '../image-index'
import type { StorageBackend } from '../storage/types'
import { getBackupJob, markJobDone, markJobFailed, updateBackupJob } from './jobs'
import { getBackupStagingDir } from './paths'
import { collectMediaAssetKeys } from '../media-assets'
export interface BackendSyncOptions {
fromBackendId: string
@@ -63,6 +67,44 @@ async function streamToBuffer(stream: NodeJS.ReadableStream): Promise<Buffer> {
return Buffer.concat(chunks)
}
async function syncMediaRow(
row: ImageIndexRow,
fromBackend: StorageBackend,
toBackend: StorageBackend,
deleteSource: boolean
): Promise<void> {
const keys = await collectMediaAssetKeys(row, fromBackend)
if (!keys.length) throw new Error('没有可同步的文件对象')
for (const key of keys) {
const stored = await fromBackend.head(key)
if (!stored) throw new Error(`源文件不存在：${key}`)
const bytes = await streamToBuffer(await fromBackend.createStream(key))
const isPrimary = key === row.key
await toBackend.put(key, bytes, {
originalName: isPrimary ? row.original_name : basename(key),
uploadedAt: row.uploaded_at,
contentType: isPrimary ? row.content_type : contentTypeFromKey(key),
size: bytes.length,
userId: row.user_id,
mediaType: isPrimary ? mediaTypeForKey(key) : 'video'
})
if (!await toBackend.head(key)) {
throw new Error(`目标后端写入校验失败：${key}`)
}
}
updateImageBackendId(row.key, toBackend.id)
if (deleteSource) {
for (const key of keys) {
await fromBackend.delete(key)
}
}
}
export async function runBackendSyncJob(
jobId: string,
options: BackendSyncOptions
@@ -88,32 +130,13 @@ export async function runBackendSyncJob(
updateBackupJob(jobId, {
total: rows.length,
progress: 0,
message: '正在同步图片…'
message: '正在同步文件…'
})
let processed = 0
for (const row of rows) {
try {
const stream = await createImageStream(row.key)
const bytes = await streamToBuffer(stream)
await toBackend.put(row.key, bytes, {
originalName: row.original_name,
uploadedAt: row.uploaded_at,
contentType: row.content_type,
size: row.size,
userId: row.user_id
})
const head = await toBackend.head(row.key)
if (!head) {
throw new Error('目标后端写入校验失败')
}
updateImageBackendId(row.key, options.toBackendId)
if (options.deleteSource) {
await fromBackend.delete(row.key)
}
await syncMediaRow(row, fromBackend, toBackend, Boolean(options.deleteSource))
} catch (error) {
failedKeys.push(row.key)
await fs.mkdir(join(failedLogPath, '..'), { recursive: true }).catch(() => {})
@@ -136,7 +159,7 @@ export async function runBackendSyncJob(
}
if (failedKeys.length > 0) {
markJobFailed(jobId, `${failedKeys.length} 张图片同步失败`, `部分失败：${failedKeys.length}/${rows.length}`)
markJobFailed(jobId, `${failedKeys.length} 个文件同步失败`, `部分失败：${failedKeys.length}/${rows.length}`)
updateBackupJob(jobId, {
payload: { failedKeys, failedLogPath, preview }
})
@@ -178,22 +201,10 @@ export async function retryFailedBackendSync(jobId: string): Promise<void> {
const stillFailed: string[] = []
for (const key of keys) {
const row = getImageIndexAsStored(key)
const row = getImageIndexRow(key)
if (!row) continue
try {
const stream = await createImageStream(key)
const bytes = await streamToBuffer(stream)
await toBackend.put(key, bytes, {
originalName: row.originalName,
uploadedAt: row.uploadedAt,
contentType: row.contentType,
size: row.size,
userId: row.userId
})
updateImageBackendId(key, toBackendId)
if (deleteSource) {
await fromBackend.delete(key)
}
await syncMediaRow(row, fromBackend, toBackend, deleteSource)
} catch {
stillFailed.push(key)
}
@@ -205,7 +216,7 @@ export async function retryFailedBackendSync(jobId: string): Promise<void> {
stillFailed.map(key => `${key}\tretry failed\n`).join(''),
'utf8'
)
throw new Error(`仍有 ${stillFailed.length} 张图片同步失败`)
throw new Error(`仍有 ${stillFailed.length} 个文件同步失败`)
}
await fs.rm(failedLogPath, { force: true }).catch(() => {})
server/utils/backup/full-export.test.ts
+106
View File
@@ -0,0 +1,106 @@
import * as tar from 'tar'
import { describe, expect, it } from 'vitest'
import { ensureDefaultBackends } from '../storage-backends'
import { getActiveBackend } from '../storage/resolver'
import { putImage } from '../storage'
import { updateImageThumb, updateImageTranscode } from '../image-index'
import { createBackupJob, getBackupJob } from './jobs'
import { runFullExportJob } from './full-export'
import { runFullRestoreJob } from './full-restore'
async function archiveEntries(path: string): Promise<string[]> {
const entries: string[] = []
await tar.t({
file: path,
onentry: (entry) => {
if (entry.type === 'File') entries.push(entry.path.replace(/^\.\//, ''))
}
})
return entries
}
describe('runFullExportJob', () => {
it('includes video source, thumbnail and HLS artifacts', async () => {
ensureDefaultBackends()
const key = 'videos/2026/10/backup-test.mp4'
const thumbKey = 'videos/2026/10/backup-test.webp'
const hlsKey = 'videos/2026/10/backup-test/index.m3u8'
const segmentKey = 'videos/2026/10/backup-test/seg000.ts'
const meta = {
originalName: 'backup-test.mp4',
uploadedAt: '2026-10-17T00:00:00.000Z',
contentType: 'video/mp4',
size: 1,
userId: 1,
mediaType: 'video' as const
}
await putImage(key, new Uint8Array([1]), meta)
const backend = await getActiveBackend()
await backend.put(thumbKey, new Uint8Array([2]), meta)
await backend.put(
hlsKey,
new TextEncoder().encode('#EXTM3U\n#EXTINF:6,\nseg000.ts\n'),
meta
)
await backend.put(segmentKey, new Uint8Array([3]), meta)
updateImageThumb(key, thumbKey)
updateImageTranscode(key, {
status: 'done',
hlsKey,
progress: 100
})
const job = createBackupJob({ type: 'full_export', message: 'test' })
await runFullExportJob(job.id, 'test')
const done = getBackupJob(job.id)
expect(done?.status).toBe('done')
const entries = await archiveEntries(done!.outputPath!)
expect(entries).toEqual(expect.arrayContaining([
'blobs/videos/2026/10/backup-test.mp4',
'blobs/videos/2026/10/backup-test.webp',
'blobs/videos/2026/10/backup-test/index.m3u8',
'blobs/videos/2026/10/backup-test/seg000.ts'
]))
})
it('restores missing sidecar files in skip mode', async () => {
ensureDefaultBackends()
const key = 'videos/2026/10/restore-test.mp4'
const thumbKey = 'videos/2026/10/restore-test.webp'
const hlsKey = 'videos/2026/10/restore-test/index.m3u8'
const segmentKey = 'videos/2026/10/restore-test/seg000.ts'
const meta = {
originalName: 'restore-test.mp4',
uploadedAt: '2026-10-17T00:00:00.000Z',
contentType: 'video/mp4',
size: 1,
userId: 1,
mediaType: 'video' as const
}
await putImage(key, new Uint8Array([1]), meta)
const backend = await getActiveBackend()
await backend.put(thumbKey, new Uint8Array([2]), meta)
await backend.put(
hlsKey,
new TextEncoder().encode('#EXTM3U\n#EXTINF:6,\nseg000.ts\n'),
meta
)
await backend.put(segmentKey, new Uint8Array([3]), meta)
updateImageThumb(key, thumbKey)
updateImageTranscode(key, { status: 'done', hlsKey, progress: 100 })
const exportJob = createBackupJob({ type: 'full_export', message: 'test' })
await runFullExportJob(exportJob.id, 'test')
const archivePath = getBackupJob(exportJob.id)!.outputPath!
await backend.delete(segmentKey)
expect(await backend.head(segmentKey)).toBeNull()
const restoreJob = createBackupJob({ type: 'full_restore', message: 'test' })
await runFullRestoreJob(restoreJob.id, archivePath, { conflictMode: 'skip' })
expect(await backend.head(segmentKey)).not.toBeNull()
})
})
server/utils/backup/full-export.ts
+12 -39
View File
@@ -1,5 +1,5 @@
import { createWriteStream, promises as fs } from 'node:fs'
import { basename, join } from 'node:path'
import { join } from 'node:path'
import { pipeline } from 'node:stream/promises'
import * as tar from 'tar'
import { getImageTotals, listAllImageIndexRows } from '../image-index'
@@ -9,7 +9,7 @@ import {
} from '../storage-backends'
import { parseS3Config } from '../storage/s3'
import { parseTelegramConfig } from '../storage/telegram'
import { createImageStream } from '../storage'
import { getBackendForKey } from '../storage/resolver'
import { copyDatabaseSnapshot } from './db-snapshot'
import { markJobDone, markJobFailed, updateBackupJob } from './jobs'
import { BACKUP_FORMAT_VERSION, type BackupManifest, type StorageBackendMetaExport } from './manifest'
@@ -19,6 +19,7 @@ import {
getBackupsDir
} from './paths'
import type { StorageBackendRow } from '../storage/types'
import { collectMediaAssetKeys } from '../media-assets'
function formatArchiveName(jobId: string): string {
return `pichost-${jobId}${BACKUP_ARCHIVE_EXT}`
@@ -48,36 +49,15 @@ async function writeStreamToFile(stream: NodeJS.ReadableStream, filePath: string
await pipeline(stream, createWriteStream(filePath))
}
async function streamToString(stream: NodeJS.ReadableStream): Promise<string> {
const chunks: Buffer[] = []
for await (const chunk of stream) {
chunks.push(Buffer.from(chunk as Uint8Array))
}
return Buffer.concat(chunks).toString('utf8')
}
/**
* 源文件已按设计删除（转码完成）时，导出 HLS 播放列表、分片与封面。
* 恢复端会遍历 blobs 目录原样还原，因此这里只需把附属文件写进去。
*/
async function exportVideoSidecars(row: ImageIndexRow, stagingRoot: string): Promise<void> {
const keys: string[] = []
if (row.thumb_key) keys.push(row.thumb_key)
if (row.hls_key) {
keys.push(row.hls_key)
const dir = row.hls_key.slice(0, row.hls_key.lastIndexOf('/'))
const text = await streamToString(await createImageStream(row.hls_key))
for (const line of text.split('\n')) {
const name = line.trim()
if (!name || name.startsWith('#')) continue
keys.push(`${dir}/${basename(name)}`)
}
}
async function exportMediaAssets(
row: ImageIndexRow,
stagingRoot: string
): Promise<void> {
const backend = await getBackendForKey(row.key)
const keys = await collectMediaAssetKeys(row, backend)
for (const key of keys) {
await writeStreamToFile(
await createImageStream(key),
await backend.createStream(key),
join(stagingRoot, 'blobs', key)
)
}
@@ -138,17 +118,10 @@ export async function runFullExportJob(jobId: string, appVersion: string): Promi
for (const row of rows) {
updateBackupJob(jobId, {
progress: processed,
message: `正在导出图片 ${processed}/${totals.count + 1}…`
message: `正在导出文件 ${processed}/${totals.count + 1}…`
})
if (Number(row.source_deleted ?? 0) === 1) {
// 转码完成的视频只有 HLS 产物与封面
await exportVideoSidecars(row, stagingRoot)
} else {
const blobPath = join(stagingRoot, 'blobs', row.key)
const stream = await createImageStream(row.key)
await writeStreamToFile(stream, blobPath)
}
await exportMediaAssets(row, stagingRoot)
processed += 1
updateBackupJob(jobId, { progress: processed })
}
server/utils/backup/full-restore.ts
+11 -8
View File
@@ -4,6 +4,7 @@ import { pipeline } from 'node:stream/promises'
import * as tar from 'tar'
import { getImageIndexRow } from '../image-index'
import { getActiveBackend } from '../storage/resolver'
import { contentTypeFromKey } from '../content-type'
import { replaceDatabaseFromSnapshot } from './db-snapshot'
import { markJobDone, markJobFailed, restoreBackupJobsAfterReplace, snapshotBackupJobs, updateBackupJob } from './jobs'
import {
@@ -60,10 +61,11 @@ export async function previewFullRestore(archivePath: string): Promise<BackupRes
blobFiles = []
}
const backend = await getActiveBackend()
let conflictingImages = 0
for (const file of blobFiles) {
const rel = relative(blobsRoot, file).replace(/\\/g, '/')
if (getImageIndexRow(rel)) {
if (await backend.head(rel)) {
conflictingImages += 1
}
}
@@ -104,7 +106,7 @@ export async function runFullRestoreJob(
restoreBackupJobsAfterReplace(preservedJobs)
updateBackupJob(jobId, {
status: 'running',
message: '正在恢复图片…',
message: '正在恢复文件…',
progress: 0,
total: 0
})
@@ -112,7 +114,7 @@ export async function runFullRestoreJob(
const blobsRoot = join(stagingRoot, 'blobs')
const blobFiles = await walkFiles(blobsRoot)
const total = blobFiles.length + 1
updateBackupJob(jobId, { total, progress: 1, message: '正在恢复图片…' })
updateBackupJob(jobId, { total, progress: 1, message: '正在恢复文件…' })
const backend = await getActiveBackend()
let processed = 1
@@ -120,8 +122,9 @@ export async function runFullRestoreJob(
for (const file of blobFiles) {
const key = relative(blobsRoot, file).replace(/\\/g, '/')
const indexed = getImageIndexRow(key)
const exists = Boolean(await backend.head(key))
if (!indexed && options.conflictMode === 'skip') {
if (exists && options.conflictMode === 'skip') {
processed += 1
continue
}
@@ -138,19 +141,19 @@ export async function runFullRestoreJob(
: {
originalName: key.split('/').pop() ?? key,
uploadedAt: new Date().toISOString(),
contentType: 'application/octet-stream',
contentType: contentTypeFromKey(key),
size: bytes.length,
userId: null
}
if (options.conflictMode === 'overwrite' || !indexed) {
if (options.conflictMode === 'overwrite' || !exists) {
await backend.put(key, bytes, meta)
}
processed += 1
updateBackupJob(jobId, {
progress: processed,
message: `正在恢复图片 ${processed}/${total}…`
message: `正在恢复文件 ${processed}/${total}…`
})
}
@@ -159,7 +162,7 @@ export async function runFullRestoreJob(
markJobDone(jobId, {
message: '恢复完成，请重新登录并在存储页检查对象存储密钥',
payload: {
restoredImages: blobFiles.length,
restoredFiles: blobFiles.length,
manifest
}
})
server/utils/backup/jobs.ts
+58 -25
View File
@@ -1,5 +1,5 @@
import { randomBytes } from 'node:crypto'
import { getDb } from '../db'
import { getDb, getPrimaryAdminUserId } from '../db'
import { ensureBackupSchema } from './schema'
export type BackupJobType = 'full_export' | 'full_restore' | 'backend_sync'
@@ -7,6 +7,7 @@ export type BackupJobStatus = 'pending' | 'running' | 'done' | 'failed'
export interface BackupJob {
id: string
userId: number | null
type: BackupJobType
status: BackupJobStatus
progress: number
@@ -21,6 +22,7 @@ export interface BackupJob {
interface BackupJobRow {
id: string
user_id: number | null
type: BackupJobType
status: BackupJobStatus
progress: number
@@ -43,6 +45,7 @@ function rowToJob(row: BackupJobRow): BackupJob {
})()
return {
id: row.id,
userId: row.user_id ?? null,
type: row.type,
status: row.status,
progress: row.progress,
@@ -62,33 +65,34 @@ function generateJobId(): string {
return `${stamp}-${suffix}`
}
export function getRunningBackupJob(): BackupJob | null {
export function getRunningBackupJob(userId?: number): BackupJob | null {
ensureBackupSchema()
const row = getDb().prepare(`
SELECT id, type, status, progress, total, message, payload_json,
SELECT id, user_id, type, status, progress, total, message, payload_json,
output_path, error, created_at, finished_at
FROM backup_jobs
WHERE status = 'running'
WHERE status = 'running'${userId === undefined ? '' : ' AND user_id = ?'}
LIMIT 1
`).get() as BackupJobRow | undefined
`).get(...(userId === undefined ? [] : [userId])) as BackupJobRow | undefined
return row ? rowToJob(row) : null
}
export function getActiveBackupJob(): BackupJob | null {
export function getActiveBackupJob(userId?: number): BackupJob | null {
ensureBackupSchema()
const row = getDb().prepare(`
SELECT id, type, status, progress, total, message, payload_json,
SELECT id, user_id, type, status, progress, total, message, payload_json,
output_path, error, created_at, finished_at
FROM backup_jobs
WHERE status IN ('pending', 'running')
WHERE status IN ('pending', 'running')${userId === undefined ? '' : ' AND user_id = ?'}
ORDER BY created_at DESC
LIMIT 1
`).get() as BackupJobRow | undefined
`).get(...(userId === undefined ? [] : [userId])) as BackupJobRow | undefined
return row ? rowToJob(row) : null
}
export function createBackupJob(input: {
type: BackupJobType
userId?: number | null
total?: number
message?: string
payload?: Record<string, unknown>
@@ -97,13 +101,15 @@ export function createBackupJob(input: {
const id = generateJobId()
const now = new Date().toISOString()
const payloadJson = JSON.stringify(input.payload ?? {})
const userId = input.userId === undefined ? getPrimaryAdminUserId() : input.userId
getDb().prepare(`
INSERT INTO backup_jobs
(id, type, status, progress, total, message, payload_json, output_path, error, created_at, finished_at)
VALUES (?, ?, 'pending', 0, ?, ?, ?, NULL, NULL, ?, NULL)
(id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, finished_at)
VALUES (?, ?, ?, 'pending', 0, ?, ?, ?, NULL, NULL, ?, NULL)
`).run(
id,
userId,
input.type,
input.total ?? 0,
input.message ?? '',
@@ -114,43 +120,69 @@ export function createBackupJob(input: {
return getBackupJob(id)!
}
export function getBackupJob(id: string): BackupJob | null {
export function getBackupJob(id: string, userId?: number): BackupJob | null {
ensureBackupSchema()
const row = getDb().prepare(`
SELECT id, type, status, progress, total, message, payload_json,
SELECT id, user_id, type, status, progress, total, message, payload_json,
output_path, error, created_at, finished_at
FROM backup_jobs
WHERE id = ?
`).get(id) as BackupJobRow | undefined
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): BackupJob[] {
return listBackupJobsPage(1, limit).jobs
export function listBackupJobs(limit = 20, userId?: number): BackupJob[] {
return listBackupJobsPage(1, limit, userId).jobs
}
export function countBackupJobs(): number {
export function countBackupJobs(userId?: number): number {
ensureBackupSchema()
const row = getDb().prepare('SELECT COUNT(*) AS count FROM backup_jobs').get() as { count: number }
const row = getDb().prepare(`
SELECT COUNT(*) AS count FROM backup_jobs
${userId === undefined ? '' : 'WHERE user_id = ?'}
`).get(...(userId === undefined ? [] : [userId])) as { count: number }
return row.count
}
export function listBackupJobsPage(page = 1, pageSize = 10): { jobs: BackupJob[], total: number } {
export function listBackupJobsPage(
page = 1,
pageSize = 10,
userId?: number
): { 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()
const total = countBackupJobs(userId)
const rows = getDb().prepare(`
SELECT id, type, status, progress, total, message, payload_json,
SELECT id, user_id, type, status, progress, total, message, payload_json,
output_path, error, created_at, finished_at
FROM backup_jobs
${userId === undefined ? '' : 'WHERE user_id = ?'}
ORDER BY created_at DESC
LIMIT ? OFFSET ?
`).all(limit, offset) as unknown as BackupJobRow[]
`).all(...(userId === undefined ? [limit, offset] : [userId, limit, offset])) as unknown as BackupJobRow[]
return { jobs: rows.map(rowToJob), total }
}
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, finished_at
FROM backup_jobs
WHERE user_id = ?
ORDER BY created_at DESC
`).all(userId) as unknown as BackupJobRow[]
return rows.map(rowToJob)
}
export function deleteBackupJobsForUser(userId: number): number {
ensureBackupSchema()
const result = getDb().prepare('DELETE FROM backup_jobs WHERE user_id = ?').run(userId)
return Number(result.changes ?? 0)
}
/** Capture live task history before a full restore swaps the database file. */
export function snapshotBackupJobs(limit = 50): BackupJob[] {
return listBackupJobs(limit)
@@ -165,12 +197,13 @@ export function restoreBackupJobsAfterReplace(preserved: BackupJob[]): void {
db.prepare('DELETE FROM backup_jobs').run()
const insert = db.prepare(`
INSERT INTO backup_jobs
(id, type, status, progress, total, message, payload_json, output_path, error, created_at, finished_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
(id, user_id, type, status, progress, total, message, payload_json, output_path, error, created_at, finished_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`)
for (const job of preserved) {
insert.run(
job.id,
job.userId,
job.type,
job.status,
job.progress,
server/utils/backup/runner.ts
+7 -3
View File
@@ -70,10 +70,11 @@ export function scheduleBackupJob(
})
}
export function startFullExportJob(appVersion: string): BackupJob {
export function startFullExportJob(appVersion: string, userId?: number | null): BackupJob {
claimJobSlot()
const job = createBackupJob({
type: 'full_export',
userId,
message: '等待开始…'
})
scheduleBackupJob(job, () => runFullExportJob(job.id, appVersion))
@@ -82,11 +83,13 @@ export function startFullExportJob(appVersion: string): BackupJob {
export function startFullRestoreJob(
archivePath: string,
conflictMode: 'skip' | 'overwrite'
conflictMode: 'skip' | 'overwrite',
userId?: number | null
): BackupJob {
claimJobSlot()
const job = createBackupJob({
type: 'full_restore',
userId,
message: '等待开始…',
payload: { archivePath, conflictMode }
})
@@ -100,10 +103,11 @@ export function startBackendSyncJob(options: {
deleteSource?: boolean
setDefault?: boolean
dryRun?: boolean
}): BackupJob {
}, userId?: number | null): BackupJob {
claimJobSlot()
const job = createBackupJob({
type: 'backend_sync',
userId,
message: options.dryRun ? '预检中…' : '等待开始…',
payload: options
})
server/utils/backup/schema.ts
+18
View File
@@ -4,6 +4,7 @@ export function ensureBackupSchema(): void {
getDb().exec(`
CREATE TABLE IF NOT EXISTS backup_jobs (
id TEXT PRIMARY KEY,
user_id INTEGER,
type TEXT NOT NULL CHECK(type IN ('full_export', 'full_restore', 'backend_sync')),
status TEXT NOT NULL CHECK(status IN ('pending', 'running', 'done', 'failed')),
progress INTEGER NOT NULL DEFAULT 0,
@@ -18,4 +19,21 @@ export function ensureBackupSchema(): void {
CREATE INDEX IF NOT EXISTS idx_backup_jobs_created_at
ON backup_jobs(created_at DESC);
`)
const columns = getDb().prepare('PRAGMA table_info(backup_jobs)').all() as Array<{
name: string
}>
if (!columns.some(column => column.name === 'user_id')) {
getDb().exec('ALTER TABLE backup_jobs ADD COLUMN user_id INTEGER')
}
getDb().exec(`
CREATE INDEX IF NOT EXISTS idx_backup_jobs_user_created
ON backup_jobs(user_id, created_at DESC);
UPDATE backup_jobs
SET user_id = (
SELECT id FROM users WHERE role = 'admin' ORDER BY id LIMIT 1
)
WHERE user_id IS NULL
AND EXISTS (SELECT 1 FROM users WHERE role = 'admin');
`)
}
server/utils/constants.ts
+1 -3
View File
@@ -38,9 +38,7 @@ export const VIDEO_EXTENSIONS = [
'flv',
'wmv',
'mpg',
'mpeg',
'ts',
'm3u8'
'mpeg'
] as const
/** 文件床直链缓存：公开访问可长期缓存，Token 访问不能进共享缓存 */
server/utils/file-bed.test.ts
+7
View File
@@ -131,6 +131,11 @@ describe('isVideoExtension', () => {
expect(isVideoExtension('mkv')).toBe(true)
})
it('rejects HLS sidecars from the user upload allowlist', () => {
expect(isVideoExtension('ts')).toBe(false)
expect(isVideoExtension('m3u8')).toBe(false)
})
it('rejects non-video extensions', () => {
expect(isVideoExtension('png')).toBe(false)
expect(isVideoExtension('pdf')).toBe(false)
@@ -153,6 +158,8 @@ describe('validateVideoKey', () => {
it('is covered by validateStorageKey', () => {
expect(validateStorageKey('videos/2026/10/a.mp4')).toBe(true)
expect(validateStorageKey('videos/2026/10/a/index.m3u8')).toBe(true)
expect(validateStorageKey('videos/2026/10/a/seg000.ts')).toBe(true)
})
})
server/utils/image-index.ts
+29 -11
View File
@@ -45,6 +45,7 @@ export interface ImageIndexRow {
thumb_key: string | null
source_deleted: number
transcode_preset: string | null
transcode_error: string | null
}
function keyToFilePath(key: string): string {
@@ -76,6 +77,7 @@ function rowToStoredImage(row: ImageIndexRow): StoredImage {
thumbKey: row.thumb_key ?? null,
sourceDeleted: Number(row.source_deleted ?? 0) === 1,
transcodePreset: row.transcode_preset ?? null,
transcodeError: row.transcode_error ?? null,
mtimeMs: Number.isFinite(uploadedMs) ? uploadedMs : Date.now()
}
}
@@ -230,7 +232,7 @@ export async function purgeOrphanImageIndexRows(): Promise<number> {
export function getImageIndexRow(key: string): ImageIndexRow | null {
ensureStorageSchema()
const row = getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
WHERE key = ?
`).get(key) as ImageIndexRow | undefined
@@ -423,16 +425,26 @@ export function getImageTotals(): { count: number, bytes: number } {
export function listAllImageIndexRows(): ImageIndexRow[] {
ensureStorageSchema()
return getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
ORDER BY key ASC
`).all() as unknown as ImageIndexRow[]
}
export function listImageIndexRowsForUser(userId: number): ImageIndexRow[] {
ensureStorageSchema()
return getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
WHERE user_id = ?
ORDER BY key ASC
`).all(userId) as unknown as ImageIndexRow[]
}
export function listImageIndexRowsByBackend(backendId: string): ImageIndexRow[] {
ensureStorageSchema()
return getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
WHERE backend_id = ?
ORDER BY key ASC
@@ -456,6 +468,7 @@ export function updateImageTranscode(
sourceDeleted?: boolean
size?: number
preset?: string | null
error?: string | null
}
): void {
ensureStorageSchema()
@@ -466,7 +479,8 @@ export function updateImageTranscode(
transcode_progress = COALESCE(?, transcode_progress),
source_deleted = COALESCE(?, source_deleted),
size = COALESCE(?, size),
transcode_preset = COALESCE(?, transcode_preset)
transcode_preset = COALESCE(?, transcode_preset),
transcode_error = ?
WHERE key = ?
`).run(
input.status,
@@ -475,19 +489,23 @@ export function updateImageTranscode(
input.sourceDeleted === undefined ? null : (input.sourceDeleted ? 1 : 0),
input.size ?? null,
input.preset ?? null,
input.error ?? null,
key
)
}
/** 启动时恢复被中断的转码任务 */
export function listPendingTranscodeKeys(): string[] {
export function listPendingTranscodeRows(): Array<{
key: string
preset: string | null
}> {
ensureStorageSchema()
const rows = getDb().prepare(`
SELECT key FROM images
SELECT key, transcode_preset FROM images
WHERE media_type = 'video' AND transcode_status IN ('pending', 'running')
ORDER BY uploaded_at ASC
`).all() as Array<{ key: string }>
return rows.map(row => row.key)
`).all() as Array<{ key: string, transcode_preset: string | null }>
return rows.map(row => ({ key: row.key, preset: row.transcode_preset }))
}
/** 转码已完成但仍保留源文件的视频（用于补删历史遗留） */
@@ -669,7 +687,7 @@ export async function listImages(options: {
const backendSql = buildBackendFilterSql(options.backendId, params)
const tagSql = buildTagFilterSql(options.tagFilter, params)
const rows = getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
WHERE ${clauses.join(' AND ')}${userSql}${backendSql}${tagSql}
ORDER BY uploaded_at DESC, key DESC
@@ -715,7 +733,7 @@ export async function searchImages(options: {
const backendSql = buildBackendFilterSql(options.backendId, params)
const tagSql = buildTagFilterSql(options.tagFilter, params)
const rows = getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
WHERE ${clauses.join(' AND ')}${userSql}${backendSql}${tagSql}
ORDER BY uploaded_at DESC, key DESC
@@ -849,7 +867,7 @@ export async function getUserScopedStorageStats(
export async function listImageKeysForAutoDelete(): Promise<StoredImage[]> {
ensureStorageSchema()
const rows = getDb().prepare(`
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset
SELECT key, backend_id, user_id, folder, original_name, content_type, size, uploaded_at, hls_key, transcode_status, transcode_progress, thumb_key, source_deleted, transcode_preset, transcode_error
FROM images
WHERE media_type = 'image'
ORDER BY uploaded_at ASC
server/utils/media-assets.test.ts
+67
View File
@@ -0,0 +1,67 @@
import { describe, expect, it } from 'vitest'
import { LocalStorageBackend } from './storage/local'
import type { ImageIndexRow } from './image-index'
import { collectMediaAssetKeys } from './media-assets'
const row: ImageIndexRow = {
key: 'videos/2026/10/example.mp4',
backend_id: 'local',
user_id: 1,
folder: 'videos',
original_name: 'example.mp4',
content_type: 'video/mp4',
size: 100,
uploaded_at: '2026-10-17T00:00:00.000Z',
hls_key: 'videos/2026/10/example/index.m3u8',
transcode_status: 'done',
transcode_progress: 100,
thumb_key: 'videos/2026/10/example.webp',
source_deleted: 0,
transcode_preset: 'veryfast',
transcode_error: null
}
async function createBackend() {
const backend = new LocalStorageBackend()
const meta = {
originalName: 'example',
uploadedAt: row.uploaded_at,
contentType: 'application/octet-stream',
size: 0,
userId: row.user_id,
mediaType: 'video' as const
}
await backend.put(row.key, new Uint8Array([1]), meta)
await backend.put(row.thumb_key!, new Uint8Array([2]), meta)
await backend.put(
row.hls_key!,
new TextEncoder().encode('#EXTM3U\n#EXTINF:6,\nseg000.ts\n'),
meta
)
await backend.put('videos/2026/10/example/seg000.ts', new Uint8Array([3]), meta)
return backend
}
describe('collectMediaAssetKeys', () => {
it('includes source, thumbnail, playlist and HLS segments', async () => {
const backend = await createBackend()
expect(new Set(await collectMediaAssetKeys(row, backend))).toEqual(new Set([
row.key,
row.thumb_key,
row.hls_key,
'videos/2026/10/example/seg000.ts'
]))
})
it('omits the source when transcoding already removed it', async () => {
const backend = await createBackend()
expect(new Set(await collectMediaAssetKeys({
...row,
source_deleted: 1
}, backend))).toEqual(new Set([
row.thumb_key,
row.hls_key,
'videos/2026/10/example/seg000.ts'
]))
})
})
server/utils/media-assets.ts
+36
View File
@@ -0,0 +1,36 @@
import type { ImageIndexRow } from './image-index'
import type { StorageBackend } from './storage/types'
import { hlsSegmentNames } from './transcode'
async function streamToString(stream: NodeJS.ReadableStream): Promise<string> {
const chunks: Buffer[] = []
for await (const chunk of stream) {
chunks.push(Buffer.from(chunk as Uint8Array))
}
return Buffer.concat(chunks).toString('utf8')
}
/** 一个媒体记录对应的全部存储对象，包括原文件、缩略图和 HLS 附属文件。 */
export async function collectMediaAssetKeys(
row: ImageIndexRow,
backend: Pick<StorageBackend, 'createStream'>
): Promise<string[]> {
const keys = new Set<string>()
if (Number(row.source_deleted ?? 0) !== 1) {
keys.add(row.key)
}
if (row.thumb_key) {
keys.add(row.thumb_key)
}
if (row.hls_key) {
keys.add(row.hls_key)
const dir = row.hls_key.slice(0, row.hls_key.lastIndexOf('/'))
const playlist = await streamToString(await backend.createStream(row.hls_key))
for (const name of hlsSegmentNames(playlist)) {
keys.add(`${dir}/${name}`)
}
}
return [...keys]
}
server/utils/storage-backends.ts
+24
View File
@@ -94,6 +94,7 @@ const STORAGE_SCHEMA_TRANSCODE_PROGRESS_KEY = 'storage_schema_transcode_progress
const STORAGE_SCHEMA_THUMB_KEY = 'storage_schema_thumb'
const STORAGE_SCHEMA_SOURCE_DELETED_KEY = 'storage_schema_source_deleted'
const STORAGE_SCHEMA_TRANSCODE_PRESET_KEY = 'storage_schema_transcode_preset'
const STORAGE_SCHEMA_TRANSCODE_ERROR_KEY = 'storage_schema_transcode_error'
/** images 表新增 media_type 列（区分图床图片与文件床文件） */
function migrateImagesMediaType(database: DatabaseSync): void {
@@ -232,6 +233,27 @@ function migrateImagesTranscodePreset(database: DatabaseSync): void {
`).run(STORAGE_SCHEMA_TRANSCODE_PRESET_KEY, new Date().toISOString())
}
/** images 表新增「转码失败原因」列 */
function migrateImagesTranscodeError(database: DatabaseSync): void {
const migrated = database.prepare(`
SELECT value FROM settings WHERE key = ?
`).get(STORAGE_SCHEMA_TRANSCODE_ERROR_KEY) as { value?: string } | undefined
if (migrated?.value === '1') return
const columns = database.prepare('PRAGMA table_info(images)').all() as Array<{
name: string
}>
if (!columns.some(column => column.name === 'transcode_error')) {
database.exec('ALTER TABLE images ADD COLUMN transcode_error TEXT')
}
database.prepare(`
INSERT OR REPLACE INTO settings (key, value, updated_at)
VALUES (?, '1', ?)
`).run(STORAGE_SCHEMA_TRANSCODE_ERROR_KEY, new Date().toISOString())
}
function migrateStorageTables(database: DatabaseSync): void {
database.exec(`
CREATE TABLE IF NOT EXISTS storage_backends (
@@ -263,6 +285,7 @@ function migrateStorageTables(database: DatabaseSync): void {
thumb_key TEXT,
source_deleted INTEGER NOT NULL DEFAULT 0,
transcode_preset TEXT,
transcode_error TEXT,
FOREIGN KEY (backend_id) REFERENCES storage_backends(id)
);
CREATE INDEX IF NOT EXISTS idx_images_uploaded_at ON images(uploaded_at DESC);
@@ -284,6 +307,7 @@ function migrateStorageTables(database: DatabaseSync): void {
migrateImagesThumb(database)
migrateImagesSourceDeleted(database)
migrateImagesTranscodePreset(database)
migrateImagesTranscodeError(database)
database.exec(`
CREATE INDEX IF NOT EXISTS idx_images_media_type ON images(media_type);
`)
server/utils/storage/s3.ts
+12 -2
View File
@@ -38,6 +38,15 @@ function contentTypeFromKey(key: string): string {
}
}
function decodeMetadata(value: string | undefined): string | undefined {
if (!value) return value
try {
return decodeURIComponent(value)
} catch {
return value
}
}
export function buildObjectKey(prefix: string | undefined, key: string): string {
const trimmed = prefix?.replace(/^\/+|\/+$/g, '') ?? ''
return trimmed ? `${trimmed}/${key}` : key
@@ -125,7 +134,8 @@ export class S3StorageBackend implements StorageBackend {
Body: bytes,
ContentType: meta.contentType,
Metadata: {
originalName: meta.originalName,
// S3 metadata headers must be ASCII-safe.
originalName: encodeURIComponent(meta.originalName),
uploadedAt: meta.uploadedAt,
userId: meta.userId != null ? String(meta.userId) : ''
}
@@ -144,7 +154,7 @@ export class S3StorageBackend implements StorageBackend {
const uploadedAt = response.Metadata?.uploadedat
?? response.LastModified?.toISOString()
?? new Date().toISOString()
const originalName = response.Metadata?.originalname
const originalName = decodeMetadata(response.Metadata?.originalname)
?? key.split('/').pop()
?? 'image'
const userIdRaw = response.Metadata?.userid
server/utils/storage/types.ts
+2
View File
@@ -25,6 +25,8 @@ export interface StoredImage extends StoredImageMeta {
transcodeProgress?: number
/** 本视频使用的转码档位 */
transcodePreset?: string | null
/** 转码失败原因 */
transcodeError?: string | null
/** 视频封面缩略图 key */
thumbKey?: string | null
/** 转码成功后源视频是否已删除 */
server/utils/transcode.test.ts
+57
View File
@@ -1,13 +1,20 @@
import { promises as fs } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it } from 'vitest'
import {
cleanupExpiredHlsDownloads,
DEFAULT_TRANSCODE_CONCURRENCY,
DEFAULT_TRANSCODE_PRESET,
DEFAULT_TRANSCODE_THREADS,
getTranscodeConcurrency,
getTranscodeSettings,
hlsSegmentNames,
hlsPlaylistKey,
markHlsDownloadForCleanup,
parseTranscodePreset,
setTranscodeSettings,
summarizeTranscodeFailure,
thumbKeyFor
} from './transcode'
import { isVideoSidecarKey, mediaTypeForKey } from './image-key'
@@ -39,6 +46,44 @@ describe('isVideoSidecarKey', () => {
})
})
describe('hlsSegmentNames', () => {
it('returns unique relative segment names and ignores tags', () => {
expect(hlsSegmentNames([
'#EXTM3U',
'#EXTINF:6,',
'seg000.ts',
'#EXTINF:6,',
'seg001.ts',
'seg000.ts'
].join('\n'))).toEqual(['seg000.ts', 'seg001.ts'])
})
it('ignores absolute and traversal paths', () => {
expect(hlsSegmentNames([
'../secret.ts',
'https://example.com/video.ts',
'folder/seg.ts',
'seg.ts'
].join('\n'))).toEqual(['seg.ts'])
})
})
describe('cleanupExpiredHlsDownloads', () => {
it('keeps fresh downloads and removes files after their marker expires', async () => {
const now = Date.now()
const fresh = await fs.mkdtemp(join(tmpdir(), 'pichost-hls-download-fresh-'))
const expired = await fs.mkdtemp(join(tmpdir(), 'pichost-hls-download-expired-'))
await markHlsDownloadForCleanup(fresh, now + 60_000)
await markHlsDownloadForCleanup(expired, now - 1)
await cleanupExpiredHlsDownloads(now)
await expect(fs.access(fresh)).resolves.toBeUndefined()
await expect(fs.access(expired)).rejects.toThrow()
await fs.rm(fresh, { recursive: true, force: true })
})
})
describe('mediaTypeForKey', () => {
it('maps folders to media types', () => {
expect(mediaTypeForKey('images/2026/10/a.webp')).toBe('image')
@@ -47,6 +92,18 @@ describe('mediaTypeForKey', () => {
})
})
describe('summarizeTranscodeFailure', () => {
it('recognizes a missing FFmpeg binary', () => {
expect(summarizeTranscodeFailure(new Error('spawn ffmpeg ENOENT')))
.toBe('ffmpeg-missing')
})
it('maps other FFmpeg errors to unknown', () => {
const error = new Error(`first line\n${'invalid data '.repeat(20)}`)
expect(summarizeTranscodeFailure(error)).toBe('unknown')
})
})
describe('transcode settings', () => {
it('defaults to one concurrent job with one thread', () => {
expect(DEFAULT_TRANSCODE_CONCURRENCY).toBe(1)
server/utils/transcode.ts
+194 -37
View File
@@ -5,10 +5,11 @@ import { randomUUID } from 'node:crypto'
import { tmpdir } from 'node:os'
import { basename, join } from 'node:path'
import { pipeline } from 'node:stream/promises'
import sharp from 'sharp'
import { getDataDir } from './data-dir'
import {
listAllImageIndexRows,
listPendingTranscodeKeys,
listPendingTranscodeRows,
listTranscodedSourceRows,
updateImageTranscode
} from './image-index'
@@ -37,6 +38,9 @@ import {
/** HLS 分片时长（秒） */
const HLS_SEGMENT_TIME = 6
const HLS_DOWNLOAD_PREFIX = 'pichost-hls-download-'
export const HLS_DOWNLOAD_RETENTION_MS = 60 * 60 * 1000
const HLS_DOWNLOAD_MARKER = '.expires'
export const DEFAULT_TRANSCODE_CONCURRENCY = 1
export const DEFAULT_TRANSCODE_THREADS = 1
@@ -56,6 +60,21 @@ export const DEFAULT_TRANSCODE_BITRATE_PERCENT = 80
export const MIN_TRANSCODE_BITRATE_PERCENT = 10
export const MAX_TRANSCODE_BITRATE_PERCENT = 100
export interface TranscodeJobOptions {
preset: TranscodePreset
bitratePercent: number
}
/** 将底层错误归纳成适合界面展示的简短原因 */
export function summarizeTranscodeFailure(error: unknown): string {
const message = error instanceof Error ? error.message : String(error)
if (/spawn ffmpeg ENOENT|ffmpeg.*(not found|ENOENT)|ENOENT.*ffmpeg/i.test(message)) {
return 'ffmpeg-missing'
}
if (/permission denied/i.test(message)) return 'permission-denied'
return 'unknown'
}
export function getTranscodeBitratePercent(): number {
return parseBoundedInt(
getSetting(SETTINGS_TRANSCODE_BITRATE),
@@ -84,6 +103,21 @@ function parseBoundedInt(
return value
}
function resolveTranscodeJobOptions(
options: Partial<TranscodeJobOptions> = {}
): TranscodeJobOptions {
const settings = getTranscodeSettings()
const bitratePercent = Number.isInteger(options.bitratePercent)
&& options.bitratePercent! >= MIN_TRANSCODE_BITRATE_PERCENT
&& options.bitratePercent! <= MAX_TRANSCODE_BITRATE_PERCENT
? options.bitratePercent!
: settings.bitratePercent
return {
preset: options.preset ?? settings.preset,
bitratePercent
}
}
export interface TranscodeSettings {
concurrency: number
threads: number
@@ -162,6 +196,19 @@ export function thumbKeyFor(videoKey: string): string {
return `${stripExtension(videoKey)}.webp`
}
/** HLS 播放列表中的相对分片文件名 */
export function hlsSegmentNames(playlistText: string): string[] {
const names = new Set<string>()
for (const line of playlistText.split('\n')) {
const name = line.trim()
if (!name || name.startsWith('#') || name === '.' || name === '..' || /[/\\]/.test(name)) {
continue
}
names.add(name)
}
return [...names]
}
function runThumbFfmpeg(inputPath: string, outputPath: string, seek: number): Promise<void> {
return new Promise((resolve, reject) => {
const child = spawn('ffmpeg', [
@@ -172,8 +219,7 @@ function runThumbFfmpeg(inputPath: string, outputPath: string, seek: number): Pr
'-i', inputPath,
'-frames:v', '1',
'-vf', 'scale=640:-2',
'-c:v', 'libwebp',
'-quality', '80',
'-c:v', 'png',
outputPath
], { windowsHide: true })
@@ -203,6 +249,7 @@ export async function generateVideoThumbnail(videoKey: string): Promise<string |
const workDir = join(tmpdir(), `pichost-thumb-${randomUUID()}`)
const inputPath = join(workDir, `input.${extensionOf(videoKey)}`)
const framePath = join(workDir, 'thumb.png')
const outputPath = join(workDir, 'thumb.webp')
await fs.mkdir(workDir, { recursive: true })
@@ -211,11 +258,12 @@ export async function generateVideoThumbnail(videoKey: string): Promise<string |
await streamToFile(await createImageStream(videoKey), inputPath)
try {
await runThumbFfmpeg(inputPath, outputPath, 1)
await runThumbFfmpeg(inputPath, framePath, 1)
} catch {
await runThumbFfmpeg(inputPath, outputPath, 0)
await runThumbFfmpeg(inputPath, framePath, 0)
}
await sharp(framePath).webp({ quality: 80 }).toFile(outputPath)
const bytes = await fs.readFile(outputPath)
const thumbKey = thumbKeyFor(videoKey)
const backend = await getBackendForKey(videoKey)
@@ -318,9 +366,10 @@ function runFfmpeg(
outputDir: string,
sourceBitrate: number,
sourceAudioBitrate: number,
options: TranscodeJobOptions,
onProgressUs: (outTimeUs: number) => void
): Promise<void> {
const { preset, bitratePercent } = getTranscodeSettings()
const { preset, bitratePercent } = options
// 音频不放大也不压缩：跟随源音轨码率（读不到时才用 128k）
const audioBitrate = sourceAudioBitrate > 0
@@ -419,12 +468,96 @@ async function streamToString(stream: NodeJS.ReadableStream): Promise<string> {
return Buffer.concat(chunks).toString('utf8')
}
function remuxPlaylistToMp4(inputPath: string, outputPath: string): Promise<void> {
return new Promise((resolve, reject) => {
const child = spawn('ffmpeg', [
'-hide_banner',
'-loglevel', 'error',
'-y',
'-i', inputPath,
'-c', 'copy',
'-movflags', '+faststart',
outputPath
], { windowsHide: true })
let stderr = ''
child.stderr.on('data', (chunk) => {
stderr += String(chunk)
})
child.on('error', reject)
child.on('close', (code) => {
if (code === 0) {
resolve()
return
}
const detail = stderr.trim().split('\n').filter(Boolean).pop()
reject(new Error(detail || `ffmpeg 退出码 ${code}`))
})
})
}
/** 下载 HLS 清单和分片到临时目录，不转码，重新封装为 MP4 */
export async function remuxHlsToMp4(playlistKey: string, workDir: string): Promise<string> {
if (!validateStorageKey(playlistKey) || !playlistKey.endsWith('.m3u8')) {
throw new Error('无效的 HLS 播放列表')
}
const playlistText = await streamToString(await createImageStream(playlistKey))
const segmentNames = hlsSegmentNames(playlistText)
if (!segmentNames.length) throw new Error('HLS 播放列表没有视频分片')
const dirKey = playlistKey.slice(0, playlistKey.lastIndexOf('/'))
await fs.mkdir(workDir, { recursive: true })
await fs.writeFile(join(workDir, 'index.m3u8'), playlistText)
for (const name of segmentNames) {
const segmentKey = `${dirKey}/${name}`
if (!validateStorageKey(segmentKey)) throw new Error('HLS 分片路径无效')
await streamToFile(await createImageStream(segmentKey), join(workDir, name))
}
const outputPath = join(workDir, 'video.mp4')
await remuxPlaylistToMp4(join(workDir, 'index.m3u8'), outputPath)
return outputPath
}
/** 标记 HLS 下载临时文件，满一小时后允许清理 */
export async function markHlsDownloadForCleanup(
workDir: string,
expiresAt = Date.now() + HLS_DOWNLOAD_RETENTION_MS
): Promise<void> {
await fs.writeFile(join(workDir, HLS_DOWNLOAD_MARKER), String(expiresAt))
}
/** 删除已经到达过期时间的 HLS 下载临时目录 */
export async function cleanupExpiredHlsDownloads(now = Date.now()): Promise<number> {
let removed = 0
try {
const entries = await fs.readdir(tmpdir(), { withFileTypes: true })
for (const entry of entries) {
if (!entry.isDirectory() || !entry.name.startsWith(HLS_DOWNLOAD_PREFIX)) continue
const path = join(tmpdir(), entry.name)
const expiresAt = Number(
await fs.readFile(join(path, HLS_DOWNLOAD_MARKER), 'utf8').catch(() => '')
)
if (!Number.isFinite(expiresAt) || expiresAt > now) continue
await fs.rm(path, { recursive: true, force: true })
removed++
}
} catch {
// 清理失败不影响新下载
}
return removed
}
/** 把视频转成 HLS 并写入存储，返回播放列表 key 与产物总大小 */
export async function transcodeVideoToHls(
videoKey: string
videoKey: string,
jobOptions?: Partial<TranscodeJobOptions>
): Promise<{ playlistKey: string, totalBytes: number }> {
const stored = await headImage(videoKey)
if (!stored) throw new Error('视频不存在')
const options = resolveTranscodeJobOptions(jobOptions)
const workDir = join(tmpdir(), `pichost-hls-${randomUUID()}`)
const inputPath = join(workDir, `input.${extensionOf(videoKey)}`)
@@ -445,17 +578,25 @@ export async function transcodeVideoToHls(
const sourceAudioBitrate = await probeAudioBitrate(inputPath)
let lastProgressAt = 0
await runFfmpeg(videoKey, inputPath, outputDir, sourceBitrate, sourceAudioBitrate, (outTimeUs) => {
if (!durationSec) return
const now = Date.now()
if (now - lastProgressAt < 1000) return
lastProgressAt = now
const percent = Math.round((outTimeUs / (durationSec * 1_000_000)) * 100)
updateImageTranscode(videoKey, {
status: 'running',
progress: Math.min(99, Math.max(1, percent))
})
})
await runFfmpeg(
videoKey,
inputPath,
outputDir,
sourceBitrate,
sourceAudioBitrate,
options,
(outTimeUs) => {
if (!durationSec) return
const now = Date.now()
if (now - lastProgressAt < 1000) return
lastProgressAt = now
const percent = Math.round((outTimeUs / (durationSec * 1_000_000)) * 100)
updateImageTranscode(videoKey, {
status: 'running',
progress: Math.min(99, Math.max(1, percent))
})
}
)
const uploadedAt = new Date().toISOString()
@@ -503,10 +644,8 @@ export async function removeHlsArtifacts(videoKey: string): Promise<void> {
const dirKey = stripExtension(videoKey)
const text = await streamToString(await createImageStream(playlistKey))
for (const line of text.split('\n')) {
const name = line.trim()
if (!name || name.startsWith('#')) continue
const segmentKey = `${dirKey}/${basename(name)}`
for (const name of hlsSegmentNames(text)) {
const segmentKey = `${dirKey}/${name}`
if (!validateStorageKey(segmentKey)) continue
await deleteImage(segmentKey).catch(() => {})
}
@@ -520,15 +659,20 @@ export async function removeHlsArtifacts(videoKey: string): Promise<void> {
}
}
const queue: string[] = []
interface QueuedTranscode extends TranscodeJobOptions {
videoKey: string
}
const queue: QueuedTranscode[] = []
const queued = new Set<string>()
let activeCount = 0
async function processOne(videoKey: string): Promise<void> {
async function processOne(job: QueuedTranscode): Promise<void> {
const { videoKey } = job
updateImageTranscode(videoKey, { status: 'running' })
try {
const { playlistKey, totalBytes } = await transcodeVideoToHls(videoKey)
const { playlistKey, totalBytes } = await transcodeVideoToHls(videoKey, job)
// 转码成功后源文件不再需要，直接删除以节省空间
const backend = await getBackendForKey(videoKey)
@@ -543,7 +687,10 @@ async function processOne(videoKey: string): Promise<void> {
})
logInfo('transcode done', { key: videoKey, hlsKey: playlistKey })
} catch (error) {
updateImageTranscode(videoKey, { status: 'failed' })
updateImageTranscode(videoKey, {
status: 'failed',
error: summarizeTranscodeFailure(error)
})
logException('transcode failed', error, { key: videoKey })
}
}
@@ -551,10 +698,11 @@ async function processOne(videoKey: string): Promise<void> {
/** 按设置的并发数消费队列（默认 1，即串行） */
function drain(): void {
while (queue.length && activeCount < getTranscodeConcurrency()) {
const key = queue.shift()!
const job = queue.shift()!
const key = job.videoKey
activeCount += 1
void processOne(key).finally(() => {
void processOne(job).finally(() => {
activeCount -= 1
queued.delete(key)
drain()
@@ -562,25 +710,28 @@ function drain(): void {
}
}
export function enqueueTranscode(videoKey: string): void {
export function enqueueTranscode(
videoKey: string,
options: Partial<TranscodeJobOptions> = {}
): void {
if (!validateVideoKey(videoKey) || queued.has(videoKey)) return
const job = resolveTranscodeJobOptions(options)
queued.add(videoKey)
queue.push(videoKey)
queue.push({ videoKey, ...job })
// 记录本视频使用的档位，便于上传页展示
const preset = getTranscodeSettings().preset
updateImageTranscode(videoKey, {
status: 'pending',
// 压缩档带上百分比，例如 compress:80
preset: preset === 'compress'
? `compress:${getTranscodeBitratePercent()}`
: preset
preset: job.preset === 'compress'
? `compress:${job.bitratePercent}`
: job.preset
})
drain()
}
/** 取消排队中或正在转码的任务（删除视频时调用） */
export function cancelTranscode(videoKey: string): void {
const index = queue.indexOf(videoKey)
const index = queue.findIndex(job => job.videoKey === videoKey)
if (index >= 0) queue.splice(index, 1)
queued.delete(videoKey)
@@ -616,8 +767,14 @@ export async function cleanupTranscodedSources(): Promise<number> {
/** 启动时把上次中断的任务重新排队 */
export function resumePendingTranscodes(): void {
for (const key of listPendingTranscodeKeys()) {
enqueueTranscode(key)
for (const row of listPendingTranscodeRows()) {
const [presetName, bitrateRaw] = (row.preset ?? '').split(':')
const preset = parseTranscodePreset(presetName)
const bitratePercent = Number(bitrateRaw)
enqueueTranscode(row.key, {
...(preset ? { preset } : {}),
...(Number.isInteger(bitratePercent) ? { bitratePercent } : {})
})
}
}
server/utils/user-storage.test.ts
+145
View File
@@ -0,0 +1,145 @@
import { promises as fs } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it } from 'vitest'
import {
getDb,
insertActivityLog,
insertUser
} from './db'
import {
listImageIndexRowsForUser,
updateImageThumb,
updateImageTranscode
} from './image-index'
import { ensureDefaultBackends } from './storage-backends'
import { getActiveBackend } from './storage/resolver'
import { putImage } from './storage'
import { createTag } from './tags'
import { clearUserStorage } from './user-storage'
import {
createBackupJob,
getBackupJob,
markJobDone
} from './backup/jobs'
import { getBackupsDir } from './backup/paths'
describe('clearUserStorage', () => {
it('deletes only the selected user media, tags, logs and cache', async () => {
ensureDefaultBackends()
const now = new Date().toISOString()
const userA = insertUser({
username: 'clear-user-a',
passwordHash: 'hash-a',
role: 'user',
createdAt: now
})
const userB = insertUser({
username: 'clear-user-b',
passwordHash: 'hash-b',
role: 'user',
createdAt: now
})
const keyA = 'videos/2026/10/clear-a.mp4'
const thumbA = 'videos/2026/10/clear-a.webp'
const hlsA = 'videos/2026/10/clear-a/index.m3u8'
const segmentA = 'videos/2026/10/clear-a/seg000.ts'
const keyB = 'images/2026/10/clear-b.webp'
const videoMeta = {
originalName: 'clear-a.mp4',
uploadedAt: now,
contentType: 'video/mp4',
size: 1,
userId: userA,
mediaType: 'video' as const
}
await putImage(keyA, new Uint8Array([1]), videoMeta)
const backend = await getActiveBackend()
await backend.put(thumbA, new Uint8Array([2]), videoMeta)
await backend.put(
hlsA,
new TextEncoder().encode('#EXTM3U\n#EXTINF:6,\nseg000.ts\n'),
videoMeta
)
await backend.put(segmentA, new Uint8Array([3]), videoMeta)
updateImageThumb(keyA, thumbA)
updateImageTranscode(keyA, { status: 'done', hlsKey: hlsA, progress: 100 })
await putImage(keyB, new Uint8Array([4]), {
originalName: 'clear-b.webp',
uploadedAt: now,
contentType: 'image/webp',
size: 1,
userId: userB,
mediaType: 'image'
})
const tagA = createTag(userA, 'A tag')
const tagB = createTag(userB, 'B tag')
getDb().prepare('INSERT INTO image_tags (image_key, tag_id) VALUES (?, ?)').run(keyA, tagA.id)
getDb().prepare('INSERT INTO image_tags (image_key, tag_id) VALUES (?, ?)').run(keyB, tagB.id)
insertActivityLog({
action: 'upload',
key: keyA,
originalName: 'clear-a.mp4',
size: 1,
contentType: 'video/mp4',
source: 'web',
userId: userA
})
insertActivityLog({
action: 'upload',
key: keyB,
originalName: 'clear-b.webp',
size: 1,
contentType: 'image/webp',
source: 'web',
userId: userB
})
const cacheDir = await fs.mkdtemp(join(tmpdir(), `pichost-hls-download-${userA}-`))
const jobA = createBackupJob({ type: 'full_export', userId: userA })
const jobB = createBackupJob({ type: 'full_export', userId: userB })
const archiveA = join(getBackupsDir(), `clear-a-${jobA.id}.phost.tar.gz`)
const archiveB = join(getBackupsDir(), `clear-b-${jobB.id}.phost.tar.gz`)
await fs.mkdir(getBackupsDir(), { recursive: true })
await fs.writeFile(archiveA, 'a')
await fs.writeFile(archiveB, 'b')
markJobDone(jobA.id, { outputPath: archiveA })
markJobDone(jobB.id, { outputPath: archiveB })
const result = await clearUserStorage(userA)
expect(result.failed).toEqual([])
expect(result.deletedMedia).toBe(1)
expect(result.deletedObjects).toBe(4)
expect(result.deletedJobs).toBe(1)
expect(listImageIndexRowsForUser(userA)).toHaveLength(0)
expect(listImageIndexRowsForUser(userB)).toHaveLength(1)
expect(await backend.head(keyB)).not.toBeNull()
expect(await backend.head(keyA)).toBeNull()
expect(await backend.head(thumbA)).toBeNull()
expect(await backend.head(hlsA)).toBeNull()
expect(await backend.head(segmentA)).toBeNull()
await expect(fs.access(cacheDir)).rejects.toThrow()
const db = getDb()
expect(db.prepare('SELECT COUNT(*) AS count FROM tags WHERE user_id = ?').get(userA))
.toEqual({ count: 0 })
expect(db.prepare('SELECT COUNT(*) AS count FROM tags WHERE user_id = ?').get(userB))
.toEqual({ count: 1 })
expect(db.prepare('SELECT COUNT(*) AS count FROM activity_logs WHERE user_id = ?').get(userA))
.toEqual({ count: 0 })
expect(db.prepare('SELECT COUNT(*) AS count FROM activity_logs WHERE user_id = ?').get(userB))
.toEqual({ count: 1 })
expect(db.prepare('SELECT COUNT(*) AS count FROM image_tags WHERE image_key = ?').get(keyA))
.toEqual({ count: 0 })
expect(db.prepare('SELECT COUNT(*) AS count FROM image_tags WHERE image_key = ?').get(keyB))
.toEqual({ count: 1 })
expect(getBackupJob(jobA.id, userA)).toBeNull()
expect(getBackupJob(jobB.id, userB)).not.toBeNull()
await expect(fs.access(archiveA)).rejects.toThrow()
await expect(fs.access(archiveB)).resolves.toBeUndefined()
})
})
server/utils/user-storage.ts
+173
View File
@@ -0,0 +1,173 @@
import { promises as fs } from 'node:fs'
import { tmpdir } from 'node:os'
import { basename, dirname, join, resolve } from 'node:path'
import { getDb } from './db'
import {
deleteImageIndex,
listImageIndexRowsForUser
} from './image-index'
import { getBackendForKey } from './storage/resolver'
import type { StorageBackend } from './storage/types'
import { cancelTranscode } from './transcode'
import { collectMediaAssetKeys } from './media-assets'
import {
deleteBackupJobsForUser,
getActiveBackupJob,
listAllBackupJobsForUser,
type BackupJob
} from './backup/jobs'
import {
getBackupsDir,
getBackupStagingDir,
isSafeBackupFilename
} from './backup/paths'
const activeUsers = new Set<number>()
export interface ClearUserStorageResult {
deletedMedia: number
deletedObjects: number
deletedTags: number
deletedLogs: number
clearedCacheDirs: number
deletedJobs: number
failed: Array<{ key: string, message: string }>
}
async function deleteObject(
backend: StorageBackend,
key: string
): Promise<boolean> {
if (!await backend.head(key)) return false
await backend.delete(key)
return true
}
async function clearHlsDownloadCache(userId: number): Promise<number> {
let removed = 0
const prefix = `pichost-hls-download-${userId}-`
try {
const entries = await fs.readdir(tmpdir(), { withFileTypes: true })
for (const entry of entries) {
if (!entry.isDirectory() || !entry.name.startsWith(prefix)) continue
await fs.rm(join(tmpdir(), entry.name), { recursive: true, force: true })
removed++
}
} catch {
// Cache cleanup failure does not affect the media cleanup result.
}
return removed
}
async function removeBackupJobArtifacts(job: BackupJob): Promise<void> {
await fs.rm(getBackupStagingDir(job.id), { recursive: true, force: true }).catch(() => {})
const paths = [
job.outputPath,
typeof job.payload.archivePath === 'string' ? job.payload.archivePath : null
]
for (const path of paths) {
if (!path || !isSafeBackupFilename(basename(path))) continue
if (resolve(dirname(path)) !== resolve(getBackupsDir())) continue
await fs.rm(path, { force: true }).catch(() => {})
}
}
export async function clearUserStorage(userId: number): Promise<ClearUserStorageResult> {
if (activeUsers.has(userId)) {
throw new Error('USER_STORAGE_CLEAR_IN_PROGRESS')
}
activeUsers.add(userId)
const result: ClearUserStorageResult = {
deletedMedia: 0,
deletedObjects: 0,
deletedTags: 0,
deletedLogs: 0,
clearedCacheDirs: 0,
deletedJobs: 0,
failed: []
}
try {
if (getActiveBackupJob(userId)) {
throw new Error('USER_STORAGE_CLEAR_ACTIVE_JOB')
}
const backupJobs = listAllBackupJobsForUser(userId)
const rows = listImageIndexRowsForUser(userId)
for (const row of rows) {
try {
cancelTranscode(row.key)
const backend = await getBackendForKey(row.key)
let keys: string[]
try {
keys = await collectMediaAssetKeys(row, backend)
} catch {
keys = [
...(Number(row.source_deleted ?? 0) !== 1 ? [row.key] : []),
...(row.thumb_key ? [row.thumb_key] : []),
...(row.hls_key ? [row.hls_key] : [])
]
}
for (const key of new Set(keys)) {
if (await deleteObject(backend, key)) result.deletedObjects++
}
deleteImageIndex(row.key)
result.deletedMedia++
} catch (error) {
result.failed.push({
key: row.key,
message: error instanceof Error ? error.message : '删除失败'
})
}
}
const db = getDb()
db.prepare(`
DELETE FROM image_tags
WHERE image_key NOT IN (SELECT key FROM images)
`).run()
if (!result.failed.length) {
const tagCount = db.prepare('SELECT COUNT(*) AS count FROM tags WHERE user_id = ?')
.get(userId) as { count: number }
const logCount = db.prepare('SELECT COUNT(*) AS count FROM activity_logs WHERE user_id = ?')
.get(userId) as { count: number }
db.exec('BEGIN')
try {
db.prepare(`
DELETE FROM image_tags
WHERE tag_id IN (SELECT id FROM tags WHERE user_id = ?)
`).run(userId)
db.prepare('DELETE FROM tags WHERE user_id = ?').run(userId)
db.prepare('DELETE FROM activity_logs WHERE user_id = ?').run(userId)
db.prepare(`
UPDATE users
SET auto_delete_days = 0, auto_delete_enabled_at = NULL
WHERE id = ?
`).run(userId)
db.exec('COMMIT')
result.deletedTags = Number(tagCount.count)
result.deletedLogs = Number(logCount.count)
} catch (error) {
db.exec('ROLLBACK')
throw error
}
}
if (!result.failed.length) {
for (const job of backupJobs) {
await removeBackupJobArtifacts(job)
}
result.deletedJobs = deleteBackupJobsForUser(userId)
}
result.clearedCacheDirs = await clearHlsDownloadCache(userId)
return result
} finally {
activeUsers.delete(userId)
}
}