完善数据库备份恢复链路
This commit is contained in:
yml2213
@@ -0,0 +1,539 @@
|
||||
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()
|
||||
}
|
||||
Reference in New Issue
Block a user