540 lines
17 KiB
TypeScript
540 lines
17 KiB
TypeScript
import crypto from 'node:crypto'
|
|
import fs from 'node:fs'
|
|
import fsPromises from 'node:fs/promises'
|
|
import path from 'node:path'
|
|
import process from 'node:process'
|
|
import { Transform, type Readable } from 'node:stream'
|
|
import { pipeline } from 'node:stream/promises'
|
|
import { pathToFileURL } from 'node:url'
|
|
|
|
import {
|
|
createConfiguredObjectStorage,
|
|
type ObjectStorage,
|
|
type StorageObjectSummary,
|
|
} from './services/file-storage/object-storage.js'
|
|
|
|
// 数据库备份传输工具:在 backend 容器内运行,负责备份文件与阿里云 OSS 之间的传输与校验。
|
|
// 由 deploy/backup-db.sh 与 deploy/restore-db.sh 通过 docker compose 调用。
|
|
// 日志一律走 stderr,stdout 只输出机器可读内容(下载的二进制、latest 的 key、list 的行),
|
|
// 便于宿主机脚本重定向捕获。
|
|
//
|
|
// 用法:
|
|
// node dist/db-backup.js upload --key backups/postgres/xxx.dump [--kind postgres-dump|app-configs] [--size bytes]
|
|
// 从 stdin 读取备份内容上传,并生成 <key>.manifest.json(size/sha256/迁移记录)。
|
|
// node dist/db-backup.js download --key backups/postgres/xxx.dump [--allow-unverified]
|
|
// 下载到 stdout,并按清单校验 sha256 与大小。
|
|
// node dist/db-backup.js latest [--prefix backups/postgres] [--allow-unverified]
|
|
// 在 stdout 打印最新 .dump 备份的 key(一行)。
|
|
// node dist/db-backup.js list [--prefix backups/postgres]
|
|
// 按时间倒序列出 key、大小、时间。
|
|
//
|
|
// OSS 配置:DB_BACKUP_OSS_* 优先,未设置的字段回退 STORAGE_OSS_*。
|
|
|
|
const USAGE = `用法:
|
|
node dist/db-backup.js upload --key <oss-key> [--kind postgres-dump|app-configs] [--size bytes]
|
|
node dist/db-backup.js download --key <oss-key> [--allow-unverified]
|
|
node dist/db-backup.js latest [--prefix backups/postgres] [--allow-unverified]
|
|
node dist/db-backup.js list [--prefix backups/postgres]
|
|
|
|
环境变量:DB_BACKUP_OSS_{ENDPOINT,BUCKET,ACCESS_KEY_ID,SECRET_ACCESS_KEY,REGION} 优先,
|
|
缺省字段复用 STORAGE_OSS_*;默认前缀 backups/postgres。`
|
|
|
|
const DEFAULT_BACKUP_PREFIX = 'backups/postgres'
|
|
|
|
export type BackupStorageConfig = {
|
|
endpoint: string
|
|
bucket: string
|
|
accessKeyId: string
|
|
secretAccessKey: string
|
|
region: string
|
|
}
|
|
|
|
export type BackupCliArgs = {
|
|
command: string
|
|
key: string
|
|
prefix: string
|
|
kind: string
|
|
allowUnverified: boolean
|
|
size?: number
|
|
}
|
|
|
|
type BackupManifest = {
|
|
key: string
|
|
kind: string
|
|
size: number
|
|
sha256: string
|
|
createdAt: string
|
|
database: string | null
|
|
serverVersion: string | null
|
|
migrations: string[] | null
|
|
}
|
|
|
|
function logInfo(message: string): void {
|
|
process.stderr.write(`[db-backup] ${message}\n`)
|
|
}
|
|
|
|
function fail(message: string): never {
|
|
process.stderr.write(`[db-backup] ${message}\n`)
|
|
process.exit(1)
|
|
}
|
|
|
|
export function resolveBackupStorageConfig(
|
|
env: Record<string, string | undefined>,
|
|
): BackupStorageConfig {
|
|
return {
|
|
endpoint: readWithFallback(env, 'DB_BACKUP_OSS_ENDPOINT', 'STORAGE_OSS_ENDPOINT'),
|
|
bucket: readWithFallback(env, 'DB_BACKUP_OSS_BUCKET', 'STORAGE_OSS_BUCKET'),
|
|
accessKeyId: readWithFallback(env, 'DB_BACKUP_OSS_ACCESS_KEY_ID', 'STORAGE_OSS_ACCESS_KEY_ID'),
|
|
secretAccessKey: readWithFallback(
|
|
env,
|
|
'DB_BACKUP_OSS_SECRET_ACCESS_KEY',
|
|
'STORAGE_OSS_SECRET_ACCESS_KEY',
|
|
),
|
|
region:
|
|
readWithFallback(env, 'DB_BACKUP_OSS_REGION', 'STORAGE_OSS_REGION') || 'oss-cn-hangzhou',
|
|
}
|
|
}
|
|
|
|
function readWithFallback(
|
|
env: Record<string, string | undefined>,
|
|
primary: string,
|
|
fallback: string,
|
|
): string {
|
|
return String(env[primary] || '').trim() || String(env[fallback] || '').trim()
|
|
}
|
|
|
|
export function parseBackupArgs(argv: string[]): BackupCliArgs {
|
|
const args: BackupCliArgs = {
|
|
command: String(argv[0] || ''),
|
|
key: '',
|
|
prefix: DEFAULT_BACKUP_PREFIX,
|
|
kind: 'postgres-dump',
|
|
allowUnverified: false,
|
|
}
|
|
|
|
for (let index = 1; index < argv.length; index += 1) {
|
|
const arg = argv[index]
|
|
const value = argv[index + 1]
|
|
if (arg === '--allow-unverified') {
|
|
args.allowUnverified = true
|
|
continue
|
|
}
|
|
if (arg === '--size' && value !== undefined) {
|
|
const size = Number(value)
|
|
if (Number.isSafeInteger(size) && size >= 0) args.size = size
|
|
index += 1
|
|
continue
|
|
}
|
|
if ((arg === '--key' || arg === '--prefix' || arg === '--kind') && value !== undefined) {
|
|
if (arg === '--key') args.key = value
|
|
if (arg === '--prefix') args.prefix = value.replace(/\/+$/, '')
|
|
if (arg === '--kind') args.kind = value
|
|
index += 1
|
|
}
|
|
}
|
|
|
|
return args
|
|
}
|
|
|
|
export function manifestKeyFor(key: string): string {
|
|
return `${key}.manifest.json`
|
|
}
|
|
|
|
export function pickLatestDumpKey(objects: StorageObjectSummary[]): string | null {
|
|
let latest: { key: string; time: number } | null = null
|
|
for (const item of objects) {
|
|
if (!item.key.endsWith('.dump')) {
|
|
continue
|
|
}
|
|
const time = item.lastModified ? item.lastModified.getTime() : 0
|
|
if (!latest || time > latest.time || (time === latest.time && item.key > latest.key)) {
|
|
latest = { key: item.key, time }
|
|
}
|
|
}
|
|
return latest ? latest.key : null
|
|
}
|
|
|
|
function createBackupStorage(): ObjectStorage {
|
|
const config = resolveBackupStorageConfig(process.env)
|
|
const missing = (['endpoint', 'bucket', 'accessKeyId', 'secretAccessKey'] as const)
|
|
.filter((field) => !config[field])
|
|
.map((field) => `DB_BACKUP_OSS_${field.replace(/([A-Z])/g, '_$1').toUpperCase()}`)
|
|
if (missing.length) {
|
|
logInfo(`缺少 OSS 配置:${missing.join(', ')}(或对应的 STORAGE_OSS_*)`)
|
|
process.exit(2)
|
|
}
|
|
|
|
return createConfiguredObjectStorage(config)
|
|
}
|
|
|
|
export async function runUpload(
|
|
storage: ObjectStorage,
|
|
args: BackupCliArgs,
|
|
content: Buffer | Readable,
|
|
): Promise<void> {
|
|
if (!args.key) {
|
|
fail(`upload 需要 --key。${USAGE}`)
|
|
}
|
|
if (args.size === 0) {
|
|
fail('输入为空,没有可上传的备份内容。')
|
|
}
|
|
const hash = crypto.createHash('sha256')
|
|
let size = 0
|
|
let uploadContent: Buffer | Readable = content
|
|
if (Buffer.isBuffer(content)) {
|
|
size = content.length
|
|
hash.update(content)
|
|
} else {
|
|
const hashingStream = new HashingStream(hash, (chunkSize) => {
|
|
size += chunkSize
|
|
})
|
|
content.pipe(hashingStream)
|
|
uploadContent = hashingStream
|
|
}
|
|
if (size === 0 && Buffer.isBuffer(content)) {
|
|
fail('输入为空,没有可上传的备份内容。')
|
|
}
|
|
|
|
const snapshot = args.kind === 'postgres-dump' ? await readDatabaseSnapshot() : null
|
|
|
|
await storage.putObject({
|
|
key: args.key,
|
|
content: uploadContent,
|
|
contentType: 'application/octet-stream',
|
|
...(Buffer.isBuffer(content)
|
|
? { contentLength: content.length }
|
|
: args.size === undefined
|
|
? {}
|
|
: { contentLength: args.size }),
|
|
})
|
|
if (size === 0) {
|
|
fail('输入为空,没有可上传的备份内容。')
|
|
}
|
|
if (args.size !== undefined && args.size !== size) {
|
|
fail(`输入大小与 --size 不一致:期望 ${args.size} 字节,实际 ${size} 字节。`)
|
|
}
|
|
const sha256 = hash.digest('hex')
|
|
|
|
const manifest: BackupManifest = {
|
|
key: args.key,
|
|
kind: args.kind,
|
|
size,
|
|
sha256,
|
|
createdAt: new Date().toISOString(),
|
|
database: snapshot ? snapshot.database : null,
|
|
serverVersion: snapshot ? snapshot.serverVersion : null,
|
|
migrations: snapshot ? snapshot.migrations : null,
|
|
}
|
|
await storage.putObject({
|
|
key: manifestKeyFor(args.key),
|
|
content: Buffer.from(`${JSON.stringify(manifest, null, 2)}\n`),
|
|
contentType: 'application/json',
|
|
})
|
|
|
|
logInfo(
|
|
`上传完成:${args.key}(${formatSize(size)},sha256=${sha256.slice(0, 12)}…),清单 ${manifestKeyFor(args.key)}`,
|
|
)
|
|
if (args.kind === 'postgres-dump' && !snapshot) {
|
|
logInfo('未能读取数据库迁移记录(数据库暂不可达不影响备份本身,清单相应字段为空)。')
|
|
}
|
|
}
|
|
|
|
export async function runDownload(storage: ObjectStorage, args: BackupCliArgs): Promise<Buffer> {
|
|
if (!args.key) {
|
|
fail(`download 需要 --key。${USAGE}`)
|
|
}
|
|
|
|
const stored = await storage.getObject(args.key)
|
|
const hash = crypto.createHash('sha256')
|
|
const chunks: Buffer[] = []
|
|
for await (const chunk of stored.reader) {
|
|
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
|
|
hash.update(buffer)
|
|
chunks.push(buffer)
|
|
}
|
|
const content = Buffer.concat(chunks)
|
|
const sha256 = hash.digest('hex')
|
|
|
|
const manifest = await readManifest(storage, args.key)
|
|
if (!manifest) {
|
|
if (!args.allowUnverified) {
|
|
fail(
|
|
`缺少 ${manifestKeyFor(args.key)},拒绝使用未校验备份。需要兼容旧备份时显式传 --allow-unverified。`,
|
|
)
|
|
}
|
|
logInfo(`未找到 ${manifestKeyFor(args.key)},已显式允许跳过 sha256 校验。`)
|
|
} else {
|
|
validateManifest(manifest, args.key, content.length, sha256, args.kind)
|
|
logInfo(`sha256 校验通过:${args.key}(${formatSize(content.length)})`)
|
|
}
|
|
|
|
return content
|
|
}
|
|
|
|
export async function runLatest(storage: ObjectStorage, args: BackupCliArgs): Promise<string> {
|
|
const objects = (await storage.listObjects(`${args.prefix}/`))
|
|
.filter((item) => item.key.endsWith('.dump'))
|
|
.sort((left, right) => {
|
|
const leftTime = left.lastModified ? left.lastModified.getTime() : 0
|
|
const rightTime = right.lastModified ? right.lastModified.getTime() : 0
|
|
return rightTime - leftTime || right.key.localeCompare(left.key)
|
|
})
|
|
|
|
for (const item of objects) {
|
|
try {
|
|
const manifest = await readManifest(storage, item.key)
|
|
if (manifest) {
|
|
validateManifestMetadata(manifest, item.key, 'postgres-dump')
|
|
const migrations = manifest.migrations ? `,迁移 ${manifest.migrations.length} 个` : ''
|
|
logInfo(
|
|
`最新备份:${item.key}(${formatSize(manifest.size)},${manifest.createdAt}${migrations})`,
|
|
)
|
|
return item.key
|
|
}
|
|
} catch (error) {
|
|
logInfo(`跳过无效备份 ${item.key}:${error instanceof Error ? error.message : String(error)}`)
|
|
}
|
|
if (args.allowUnverified) {
|
|
logInfo(`最新备份:${item.key}(已显式允许无清单备份)。`)
|
|
return item.key
|
|
}
|
|
}
|
|
|
|
if (!objects.length) {
|
|
fail(`前缀 ${args.prefix}/ 下没有 .dump 备份。`)
|
|
}
|
|
fail(`前缀 ${args.prefix}/ 下没有带有效 manifest 的 .dump 备份。`)
|
|
}
|
|
|
|
export async function runList(storage: ObjectStorage, args: BackupCliArgs): Promise<string[]> {
|
|
const objects = (await storage.listObjects(`${args.prefix}/`))
|
|
.filter((item) => !item.key.endsWith('.manifest.json'))
|
|
.sort((left, right) => {
|
|
const leftTime = left.lastModified ? left.lastModified.getTime() : 0
|
|
const rightTime = right.lastModified ? right.lastModified.getTime() : 0
|
|
return rightTime - leftTime
|
|
})
|
|
|
|
const lines = objects.map((item) => {
|
|
const time = item.lastModified ? item.lastModified.toISOString() : '-'
|
|
return `${item.key}\t${item.size}\t${time}`
|
|
})
|
|
logInfo(`前缀 ${args.prefix}/ 下共 ${objects.length} 个备份。`)
|
|
return lines
|
|
}
|
|
|
|
async function readManifest(storage: ObjectStorage, key: string): Promise<BackupManifest | null> {
|
|
try {
|
|
const stored = await storage.getObject(manifestKeyFor(key))
|
|
const chunks: Buffer[] = []
|
|
for await (const chunk of stored.reader) {
|
|
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
|
|
}
|
|
return JSON.parse(Buffer.concat(chunks).toString('utf8')) as BackupManifest
|
|
} catch (error) {
|
|
if (isMissingObjectError(error)) {
|
|
return null
|
|
}
|
|
throw error
|
|
}
|
|
}
|
|
|
|
function validateManifestMetadata(
|
|
manifest: BackupManifest,
|
|
key: string,
|
|
expectedKind: string,
|
|
): void {
|
|
if (manifest.key !== key || manifest.kind !== expectedKind) {
|
|
throw new Error(`manifest 与备份 key/type 不匹配`)
|
|
}
|
|
if (!Number.isSafeInteger(manifest.size) || manifest.size <= 0) {
|
|
throw new Error(`manifest size 无效`)
|
|
}
|
|
if (!/^[a-f0-9]{64}$/i.test(manifest.sha256)) {
|
|
throw new Error(`manifest sha256 无效`)
|
|
}
|
|
if (!manifest.createdAt || Number.isNaN(Date.parse(manifest.createdAt))) {
|
|
throw new Error(`manifest createdAt 无效`)
|
|
}
|
|
}
|
|
|
|
function validateManifest(
|
|
manifest: BackupManifest,
|
|
key: string,
|
|
size: number,
|
|
sha256: string,
|
|
expectedKind: string,
|
|
): void {
|
|
validateManifestMetadata(manifest, key, expectedKind)
|
|
if (manifest.sha256.toLowerCase() !== sha256.toLowerCase()) {
|
|
fail(`sha256 校验失败:${key} 内容与清单不一致,请勿使用该备份。`)
|
|
}
|
|
if (manifest.size !== size) {
|
|
fail(`大小校验失败:${key} 期望 ${manifest.size} 字节,实际 ${size} 字节。`)
|
|
}
|
|
}
|
|
|
|
function isMissingObjectError(error: unknown): boolean {
|
|
const candidate = error as {
|
|
name?: string
|
|
Code?: string
|
|
$metadata?: { httpStatusCode?: number }
|
|
}
|
|
return (
|
|
candidate?.$metadata?.httpStatusCode === 404 ||
|
|
['NotFound', 'NoSuchKey', 'NoSuchBucket'].includes(
|
|
String(candidate?.name || candidate?.Code || ''),
|
|
)
|
|
)
|
|
}
|
|
|
|
class HashingStream extends Transform {
|
|
constructor(
|
|
private readonly hash: crypto.Hash,
|
|
private readonly onChunk: (size: number) => void,
|
|
) {
|
|
super()
|
|
}
|
|
|
|
_transform(
|
|
chunk: Buffer,
|
|
_encoding: BufferEncoding,
|
|
callback: (error?: Error | null, data?: Buffer) => void,
|
|
): void {
|
|
this.hash.update(chunk)
|
|
this.onChunk(chunk.length)
|
|
callback(null, chunk)
|
|
}
|
|
}
|
|
|
|
async function downloadToStdout(storage: ObjectStorage, args: BackupCliArgs): Promise<void> {
|
|
if (!args.key) {
|
|
fail(`download 需要 --key。${USAGE}`)
|
|
}
|
|
|
|
const stored = await storage.getObject(args.key)
|
|
const hash = crypto.createHash('sha256')
|
|
const tempPath = path.join('/tmp', `order-site-backup-${process.pid}-${Date.now()}.dump`)
|
|
let size = 0
|
|
try {
|
|
const output = fs.createWriteStream(tempPath, { mode: 0o600 })
|
|
for await (const chunk of stored.reader) {
|
|
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
|
|
hash.update(buffer)
|
|
size += buffer.length
|
|
if (!output.write(buffer)) {
|
|
await new Promise<void>((resolve, reject) => {
|
|
output.once('drain', resolve)
|
|
output.once('error', reject)
|
|
})
|
|
}
|
|
}
|
|
await new Promise<void>((resolve, reject) => {
|
|
output.once('close', resolve)
|
|
output.once('error', reject)
|
|
output.end()
|
|
})
|
|
|
|
const sha256 = hash.digest('hex')
|
|
const manifest = await readManifest(storage, args.key)
|
|
if (!manifest) {
|
|
if (!args.allowUnverified) {
|
|
fail(
|
|
`缺少 ${manifestKeyFor(args.key)},拒绝使用未校验备份。需要兼容旧备份时显式传 --allow-unverified。`,
|
|
)
|
|
}
|
|
logInfo(`未找到 ${manifestKeyFor(args.key)},已显式允许跳过 sha256 校验。`)
|
|
} else {
|
|
validateManifest(manifest, args.key, size, sha256, args.kind)
|
|
logInfo(`sha256 校验通过:${args.key}(${formatSize(size)})`)
|
|
}
|
|
await pipeline(fs.createReadStream(tempPath), process.stdout)
|
|
} finally {
|
|
await fsPromises.rm(tempPath, { force: true })
|
|
}
|
|
}
|
|
|
|
// 尽力而为地补充数据库快照信息;数据库不可达时返回 null,不影响备份传输。
|
|
async function readDatabaseSnapshot(): Promise<{
|
|
database: string | null
|
|
serverVersion: string | null
|
|
migrations: string[] | null
|
|
} | null> {
|
|
try {
|
|
const { runtimeConfig } = await import('./config/runtime.js')
|
|
const { query, closeDb } = await import('./db/client.js')
|
|
try {
|
|
let database: string | null = null
|
|
try {
|
|
const url = new URL(String(runtimeConfig.database?.url || ''))
|
|
database = decodeURIComponent(url.pathname.replace(/^\//, '')) || null
|
|
} catch {
|
|
database = null
|
|
}
|
|
const [versionResult, migrationsResult] = await Promise.all([
|
|
query<{ server_version: string }>('SHOW server_version'),
|
|
query<{ filename: string }>('SELECT filename FROM schema_migrations ORDER BY filename'),
|
|
])
|
|
return {
|
|
database,
|
|
serverVersion: versionResult.rows[0]?.server_version || null,
|
|
migrations: migrationsResult.rows.map((row) => row.filename),
|
|
}
|
|
} finally {
|
|
await closeDb()
|
|
}
|
|
} catch (error) {
|
|
logInfo(`读取数据库快照失败:${error instanceof Error ? error.message : String(error)}`)
|
|
return null
|
|
}
|
|
}
|
|
|
|
function formatSize(size: number): string {
|
|
if (size >= 1024 * 1024) return `${(size / (1024 * 1024)).toFixed(2)} MB`
|
|
if (size >= 1024) return `${(size / 1024).toFixed(2)} KB`
|
|
return `${size} B`
|
|
}
|
|
|
|
async function main(): Promise<void> {
|
|
const args = parseBackupArgs(process.argv.slice(2))
|
|
if (!['upload', 'download', 'latest', 'list'].includes(args.command)) {
|
|
logInfo(USAGE)
|
|
process.exitCode = 2
|
|
return
|
|
}
|
|
|
|
const storage = createBackupStorage()
|
|
|
|
switch (args.command) {
|
|
case 'upload':
|
|
await runUpload(storage, args, process.stdin)
|
|
break
|
|
case 'download':
|
|
await downloadToStdout(storage, args)
|
|
break
|
|
case 'latest':
|
|
process.stdout.write(`${await runLatest(storage, args)}\n`)
|
|
break
|
|
case 'list':
|
|
for (const line of await runList(storage, args)) {
|
|
process.stdout.write(`${line}\n`)
|
|
}
|
|
break
|
|
}
|
|
}
|
|
|
|
function isInvokedAsScript(entryPath: string | undefined): boolean {
|
|
if (!entryPath) {
|
|
return false
|
|
}
|
|
try {
|
|
return import.meta.url === pathToFileURL(path.resolve(entryPath)).href
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
if (isInvokedAsScript(process.argv[1])) {
|
|
await main()
|
|
}
|