feat(ops): W4-B single-container packaging, DATA_DIR, backup/restore
- DATA_DIR (/data: agent-sessions, backups, exports, tmp); entrypoint chowns it to PUID/PGID then drops privileges via gosu; /app binary stays root read-only - compile with --allow-read --allow-write so mounted library roots are writable - preflight: DATA_DIR/session dir must be writable (fatal); enabled library roots and downloader path mappings checked read/write (warn + audit) - --role export-mappings (mapping manifest JSON, --compare row counts) and --role read-only-on (automation.readOnlyReconcile=true + audit) - scripts/backup.sh (pg_dump custom + mappings + agent sessions, SHA256SUMS, BACKUP_KEEP rotation) and scripts/restore.sh (checksum, refuses with live connections, psql single-transaction replay compatible with PG16, read-only reconcile on, row-count verification) - compose: app + postgres only, data volume, env example with key generation - docs: runbook deploy.md, backup-restore.md with verification record Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
8c6a922cfb
commit
34e8751b45
18 files changed
+918
-43
No files matched your search
@@ -1,3 +1,4 @@
|
||||
import { join, resolve } from "node:path";
|
||||
import { z } from "zod";
|
||||
|
||||
const LogLevel = z.enum(["fatal", "error", "warn", "info", "debug", "trace", "silent"]);
|
||||
@@ -16,6 +17,13 @@ export const ConfigSchema = z.object({
|
||||
WORKER_HEARTBEAT_MS: z.coerce.number().int().min(1000).default(30_000),
|
||||
/** web 静态产物目录(单容器模式):设置后同端口托管 SPA,非 /api 路径回退 index.html */
|
||||
WEB_ROOT: z.string().min(1).optional(),
|
||||
/**
|
||||
* 应用数据目录(容器内 /data,由卷挂载):agent-sessions/、backups/、exports/、tmp/。
|
||||
* 编译后的二进制与 /app 保持只读,所有运行期写入都落在这里。相对路径按进程 cwd 解析。
|
||||
*/
|
||||
DATA_DIR: z.string().min(1).default("./data"),
|
||||
/** Agent JSONL 会话目录;缺省 DATA_DIR/agent-sessions(api 与 worker 必须共享同一目录) */
|
||||
AGENT_SESSION_DIR: z.string().min(1).optional(),
|
||||
});
|
||||
export type Config = z.infer<typeof ConfigSchema>;
|
||||
|
||||
@@ -29,4 +37,19 @@ export function loadConfig(env: Record<string, string | undefined> = Deno.env.to
|
||||
return parsed.data;
|
||||
}
|
||||
|
||||
/** DATA_DIR 下的固定子目录 */
|
||||
export const DATA_SUBDIRS = ["agent-sessions", "backups", "exports", "tmp"] as const;
|
||||
|
||||
/** 解析数据目录与会话目录为绝对路径 */
|
||||
export function dataPaths(c: Pick<Config, "DATA_DIR" | "AGENT_SESSION_DIR">, cwd: string = Deno.cwd()) {
|
||||
const dataDir = resolve(cwd, c.DATA_DIR);
|
||||
return {
|
||||
dataDir,
|
||||
sessionDir: resolve(cwd, c.AGENT_SESSION_DIR ?? join(dataDir, "agent-sessions")),
|
||||
backupsDir: join(dataDir, "backups"),
|
||||
exportsDir: join(dataDir, "exports"),
|
||||
tmpDir: join(dataDir, "tmp"),
|
||||
};
|
||||
}
|
||||
|
||||
export const isDev = (c: Pick<Config, "NODE_ENV">) => c.NODE_ENV !== "production";
|
||||
+83
-19
@@ -2,20 +2,36 @@ import { parseArgs } from "node:util";
|
||||
import { createDb, migrateToLatest } from "@umt/db";
|
||||
import { migrateEngineSchema } from "@umt/tasks";
|
||||
import denoJson from "../deno.json" with { type: "json" };
|
||||
import { loadConfig } from "./config.ts";
|
||||
import { DATA_SUBDIRS, dataPaths, loadConfig } from "./config.ts";
|
||||
import { bootstrap } from "./context.ts";
|
||||
import { createApp } from "./http/app.ts";
|
||||
import { PgEventBus } from "./http/sse-pg.ts";
|
||||
import { createLogger } from "./logger.ts";
|
||||
import { SERVER_MODULES } from "./modules/index.ts";
|
||||
import { runPreflight } from "./preflight.ts";
|
||||
import { diffCounts, enableReadOnlyReconcile, exportMappings, type MappingExport, writeMappingExport } from "./ops.ts";
|
||||
import { auditStorageWarnings, checkStorage, loadMountedPaths, runPreflight } from "./preflight.ts";
|
||||
import { startWorker } from "./worker/index.ts";
|
||||
|
||||
// api/worker/all 是常驻角色;migrate/preflight/probe 是容器入口脚本用的一次性子命令
|
||||
const Roles = ["api", "worker", "all", "migrate", "preflight", "probe"] as const;
|
||||
// api/worker/all 是常驻角色;其余是容器入口 / 备份恢复脚本用的一次性子命令:
|
||||
// migrate | preflight | probe
|
||||
// export-mappings --out <file.json> [--compare <基准清单.json>] 映射清单导出(可与基准比行数)
|
||||
// read-only-on [--reason <文本>] 打开 automation.readOnlyReconcile
|
||||
const Roles = ["api", "worker", "all", "migrate", "preflight", "probe", "export-mappings", "read-only-on"] as const;
|
||||
type Role = (typeof Roles)[number];
|
||||
const OneShot = ["migrate", "preflight", "export-mappings", "read-only-on"] as const;
|
||||
type OneShotRole = (typeof OneShot)[number];
|
||||
const isOneShot = (r: Role): r is OneShotRole => (OneShot as readonly Role[]).includes(r);
|
||||
|
||||
const { values } = parseArgs({ args: Deno.args, options: { role: { type: "string", default: "all" } }, strict: true });
|
||||
const { values } = parseArgs({
|
||||
args: Deno.args,
|
||||
options: {
|
||||
role: { type: "string", default: "all" },
|
||||
out: { type: "string" },
|
||||
compare: { type: "string" },
|
||||
reason: { type: "string" },
|
||||
},
|
||||
strict: true,
|
||||
});
|
||||
const role = values.role as Role;
|
||||
if (!Roles.includes(role)) {
|
||||
console.error(`--role 必须是 ${Roles.join("|")}`);
|
||||
@@ -42,28 +58,20 @@ try {
|
||||
Deno.exit(2);
|
||||
}
|
||||
|
||||
// 运行期写入统一落在 DATA_DIR;Agent 会话目录经环境变量传给 agent 模块(它只读 AGENT_SESSION_DIR)
|
||||
const paths = dataPaths(config);
|
||||
Deno.env.set("AGENT_SESSION_DIR", paths.sessionDir);
|
||||
|
||||
const logger = createLogger({
|
||||
level: config.LOG_LEVEL,
|
||||
pretty: config.NODE_ENV === "development" && Deno.stdout.isTerminal(),
|
||||
});
|
||||
const db = createDb({ connectionString: config.DATABASE_URL });
|
||||
|
||||
if (role === "migrate" || role === "preflight") {
|
||||
if (isOneShot(role)) {
|
||||
let code = 0;
|
||||
try {
|
||||
if (role === "migrate") {
|
||||
const { results } = await migrateToLatest(db);
|
||||
for (const r of results ?? []) logger.info({ migration: r.migrationName, status: r.status }, "migration");
|
||||
logger.info({ applied: results?.length ?? 0 }, "迁移完成");
|
||||
// 任务引擎自己的表(openworkflow schema);api/worker 进程启动时不再迁移
|
||||
await migrateEngineSchema(config.DATABASE_URL);
|
||||
logger.info("任务引擎迁移完成");
|
||||
} else {
|
||||
const problems = await runPreflight({ config, db });
|
||||
for (const p of problems) logger.error({ check: p.check }, p.message);
|
||||
if (problems.length) code = 1;
|
||||
else logger.info("启动自检通过");
|
||||
}
|
||||
code = await runOneShot(role);
|
||||
} catch (err) {
|
||||
logger.error({ err }, `${role} 失败`);
|
||||
code = 1;
|
||||
@@ -73,6 +81,62 @@ if (role === "migrate" || role === "preflight") {
|
||||
Deno.exit(code);
|
||||
}
|
||||
|
||||
async function runOneShot(r: OneShotRole): Promise<number> {
|
||||
switch (r) {
|
||||
case "migrate": {
|
||||
const { results } = await migrateToLatest(db);
|
||||
for (const m of results ?? []) logger.info({ migration: m.migrationName, status: m.status }, "migration");
|
||||
logger.info({ applied: results?.length ?? 0 }, "迁移完成");
|
||||
// 任务引擎自己的表(openworkflow schema);api/worker 进程启动时不再迁移
|
||||
await migrateEngineSchema(config.DATABASE_URL);
|
||||
logger.info("任务引擎迁移完成");
|
||||
return 0;
|
||||
}
|
||||
case "preflight": {
|
||||
const problems = await runPreflight({ config, db });
|
||||
// 库/下载目录由用户挂载:不可读写只告警(并写审计),不阻止启动;数据目录不可写才阻止
|
||||
const mounts = problems.some((p) => p.check === "db") ? [] : await loadMountedPaths(db);
|
||||
const storage = await checkStorage({ ...paths, subdirs: DATA_SUBDIRS, mounts });
|
||||
problems.push(...storage.problems);
|
||||
for (const w of storage.warnings) {
|
||||
logger.warn({ kind: w.kind, id: w.id, path: w.path }, `挂载路径不可用: ${w.name} ${w.reason}`);
|
||||
}
|
||||
await auditStorageWarnings(db, storage.warnings).catch((err) => logger.warn({ err }, "存储告警写审计失败"));
|
||||
for (const p of problems) logger.error({ check: p.check }, p.message);
|
||||
if (problems.length) return 1;
|
||||
logger.info(
|
||||
{
|
||||
dataDir: paths.dataDir,
|
||||
sessionDir: paths.sessionDir,
|
||||
mounts: mounts.length,
|
||||
warnings: storage.warnings.length,
|
||||
},
|
||||
"启动自检通过",
|
||||
);
|
||||
return 0;
|
||||
}
|
||||
case "export-mappings": {
|
||||
const out = values.out ?? `${paths.exportsDir}/mappings-${new Date().toISOString().replace(/[:.]/g, "-")}.json`;
|
||||
const data = await exportMappings(db, denoJson.version);
|
||||
await writeMappingExport(out, data);
|
||||
logger.info({ out, counts: data.counts }, "映射清单已导出");
|
||||
if (!values.compare) return 0;
|
||||
const base = JSON.parse(await Deno.readTextFile(values.compare)) as MappingExport;
|
||||
const diff = diffCounts(base.counts, data.counts);
|
||||
if (diff.length) {
|
||||
logger.error({ diff }, "映射清单行数与基准不一致");
|
||||
return 3;
|
||||
}
|
||||
logger.info({ compare: values.compare }, "映射清单行数与基准一致");
|
||||
return 0;
|
||||
}
|
||||
case "read-only-on":
|
||||
await enableReadOnlyReconcile(db, values.reason ?? "restore");
|
||||
logger.info("已打开只读对账模式 automation.readOnlyReconcile=true");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
// SSE 总线:sse_event 表 + LISTEN/NOTIFY,api 与 worker 分进程时事件互相可见
|
||||
let bus: PgEventBus;
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
import { dirname } from "node:path";
|
||||
import type { DB, Kysely } from "@umt/db";
|
||||
import { sql } from "@umt/db";
|
||||
|
||||
// 运维一次性子命令(备份/恢复脚本调用):映射清单导出、恢复后打开只读对账。
|
||||
// 映射清单是"路径不携带完整身份"前提下的关键资产(需求 §8.5):数据库之外再留一份可读 JSON,
|
||||
// 用于人工核对与恢复演练的行数校验;不含任何密钥(secret 表只随 pg_dump 走,主密钥永不入备份)。
|
||||
|
||||
/** 各表导出的关键列 */
|
||||
export const MAPPING_TABLES = {
|
||||
library_root: ["id", "name", "media_kind", "local_path", "management_mode", "enabled", "sentinel_file"],
|
||||
path_mapping: [
|
||||
"id",
|
||||
"view_kind",
|
||||
"downloader_instance_id",
|
||||
"media_server_instance_id",
|
||||
"library_root_id",
|
||||
"remote_prefix",
|
||||
"local_prefix",
|
||||
"priority",
|
||||
"enabled",
|
||||
],
|
||||
media_entity: ["id", "kind", "title", "original_title", "year", "status", "merged_into_id", "updated_at"],
|
||||
external_identity: [
|
||||
"id",
|
||||
"media_entity_id",
|
||||
"content_unit_id",
|
||||
"provider",
|
||||
"entity_type",
|
||||
"external_id",
|
||||
"status",
|
||||
"evidence_level",
|
||||
"locked",
|
||||
"revision",
|
||||
],
|
||||
path_binding: [
|
||||
"id",
|
||||
"media_entity_id",
|
||||
"content_unit_id",
|
||||
"season_number",
|
||||
"binding_kind",
|
||||
"library_root_id",
|
||||
"relative_path",
|
||||
"template_version",
|
||||
"status",
|
||||
"superseded_by_id",
|
||||
],
|
||||
asset_location: [
|
||||
"id",
|
||||
"media_asset_id",
|
||||
"location_kind",
|
||||
"library_root_id",
|
||||
"downloader_instance_id",
|
||||
"relative_path",
|
||||
"status",
|
||||
"is_primary",
|
||||
"size_bytes",
|
||||
"inode",
|
||||
"device",
|
||||
"mtime",
|
||||
],
|
||||
acquisition_slot: [
|
||||
"id",
|
||||
"media_entity_id",
|
||||
"content_unit_id",
|
||||
"subscription_id",
|
||||
"policy_revision_id",
|
||||
"name",
|
||||
"status",
|
||||
"current_asset_id",
|
||||
"revision",
|
||||
],
|
||||
organization_record: [
|
||||
"id",
|
||||
"media_entity_id",
|
||||
"content_unit_id",
|
||||
"download_task_id",
|
||||
"source_downloader_instance_id",
|
||||
"source_path",
|
||||
"target_library_root_id",
|
||||
"target_path",
|
||||
"action",
|
||||
"status",
|
||||
"plan_fingerprint",
|
||||
"finished_at",
|
||||
],
|
||||
} as const satisfies Partial<Record<keyof DB, readonly string[]>>;
|
||||
|
||||
export type MappingTable = keyof typeof MAPPING_TABLES;
|
||||
|
||||
export interface MappingExport {
|
||||
format: "umt-mappings/1";
|
||||
exportedAt: string;
|
||||
appVersion: string;
|
||||
counts: Record<MappingTable, number>;
|
||||
tables: Record<MappingTable, Record<string, unknown>[]>;
|
||||
}
|
||||
|
||||
/** 按主键顺序导出映射清单 */
|
||||
export async function exportMappings(db: Kysely<DB>, appVersion: string): Promise<MappingExport> {
|
||||
const counts = {} as Record<MappingTable, number>;
|
||||
const tables = {} as Record<MappingTable, Record<string, unknown>[]>;
|
||||
for (const [table, cols] of Object.entries(MAPPING_TABLES) as [MappingTable, readonly string[]][]) {
|
||||
const { rows } = await sql<
|
||||
Record<string, unknown>
|
||||
>`select ${sql.join(cols.map((c) => sql.ref(c)))} from ${sql.table(table)} order by id`.execute(db);
|
||||
tables[table] = rows;
|
||||
counts[table] = rows.length;
|
||||
}
|
||||
return { format: "umt-mappings/1", exportedAt: new Date().toISOString(), appVersion, counts, tables };
|
||||
}
|
||||
|
||||
/** 写 JSON(先写临时文件再改名,避免半截文件) */
|
||||
export async function writeMappingExport(path: string, data: MappingExport): Promise<void> {
|
||||
await Deno.mkdir(dirname(path), { recursive: true });
|
||||
const tmp = `${path}.partial`;
|
||||
await Deno.writeTextFile(tmp, `${JSON.stringify(data, null, 2)}\n`);
|
||||
await Deno.rename(tmp, path);
|
||||
}
|
||||
|
||||
/** 两份清单逐表比行数,返回不一致的表 */
|
||||
export function diffCounts(
|
||||
expected: Record<string, number>,
|
||||
actual: Record<string, number>,
|
||||
): Array<{ table: string; expected: number; actual: number }> {
|
||||
const out: Array<{ table: string; expected: number; actual: number }> = [];
|
||||
for (const table of new Set([...Object.keys(expected), ...Object.keys(actual)])) {
|
||||
const e = expected[table] ?? -1;
|
||||
const a = actual[table] ?? -1;
|
||||
if (e !== a) out.push({ table, expected: e, actual: a });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** 打开 automation.readOnlyReconcile(保留该分组其它字段)并写审计 */
|
||||
export async function enableReadOnlyReconcile(db: Kysely<DB>, reason: string): Promise<void> {
|
||||
await db.transaction().execute(async (trx) => {
|
||||
await sql`insert into settings (key, value, updated_at)
|
||||
values ('automation', ${JSON.stringify({ readOnlyReconcile: true })}::jsonb, now())
|
||||
on conflict (key) do update set value = settings.value || excluded.value, updated_at = now()`.execute(trx);
|
||||
await trx
|
||||
.insertInto("audit_event")
|
||||
.values({
|
||||
actor_type: "system",
|
||||
actor_id: "restore",
|
||||
action: "ops.read_only_reconcile_on",
|
||||
target_type: "settings",
|
||||
target_id: "automation",
|
||||
payload: JSON.stringify({ reason }),
|
||||
})
|
||||
.execute();
|
||||
});
|
||||
}
|
||||
@@ -43,3 +43,107 @@ export async function runPreflight({ config, db, uid = Deno.uid() }: PreflightDe
|
||||
}
|
||||
return problems;
|
||||
}
|
||||
|
||||
// ── 存储自检:数据目录必须可写;库根 / 下载器映射路径不可写只告警 ──
|
||||
|
||||
export interface MountedPath {
|
||||
kind: "library_root" | "path_mapping";
|
||||
id: string;
|
||||
name: string;
|
||||
path: string;
|
||||
}
|
||||
|
||||
export interface StorageWarning extends MountedPath {
|
||||
reason: string;
|
||||
}
|
||||
|
||||
export interface StorageCheckResult {
|
||||
problems: PreflightProblem[];
|
||||
warnings: StorageWarning[];
|
||||
}
|
||||
|
||||
const errText = (e: unknown) => (e as { code?: string }).code ?? (e as Error).message;
|
||||
|
||||
/** 在目录里写一个探针文件再删掉;失败抛错 */
|
||||
async function probeWrite(dir: string): Promise<void> {
|
||||
const probe = join(dir, `.umt-write-probe-${Deno.pid}-${crypto.randomUUID().slice(0, 8)}`);
|
||||
await Deno.writeTextFile(probe, "probe\n");
|
||||
await Deno.remove(probe);
|
||||
}
|
||||
|
||||
/** 读出已启用的库根与下载器路径映射在本进程视角下的路径 */
|
||||
export async function loadMountedPaths(db: Kysely<DB>): Promise<MountedPath[]> {
|
||||
const roots = await db
|
||||
.selectFrom("library_root")
|
||||
.select(["id", "name", "local_path"])
|
||||
.where("enabled", "=", true)
|
||||
.execute();
|
||||
const maps = await db
|
||||
.selectFrom("path_mapping")
|
||||
.select(["id", "remote_prefix", "local_prefix"])
|
||||
.where("enabled", "=", true)
|
||||
.where("view_kind", "=", "downloader")
|
||||
.execute();
|
||||
return [
|
||||
...roots.map((r) => ({ kind: "library_root" as const, id: r.id, name: r.name, path: r.local_path })),
|
||||
...maps.map((m) => ({ kind: "path_mapping" as const, id: m.id, name: m.remote_prefix, path: m.local_prefix })),
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* dataDir 下建 agent-sessions/backups/exports/tmp 并逐个探测可写(失败 = problem,阻止启动);
|
||||
* sessionDir 若在 dataDir 之外也单独探测;mounts 逐个检查可读可写(失败 = warning,不阻止启动)。
|
||||
*/
|
||||
export async function checkStorage(opts: {
|
||||
dataDir: string;
|
||||
sessionDir: string;
|
||||
subdirs: readonly string[];
|
||||
mounts: MountedPath[];
|
||||
}): Promise<StorageCheckResult> {
|
||||
const problems: PreflightProblem[] = [];
|
||||
const warnings: StorageWarning[] = [];
|
||||
const dirs = [...opts.subdirs.map((s) => join(opts.dataDir, s))];
|
||||
if (!dirs.includes(opts.sessionDir)) dirs.push(opts.sessionDir);
|
||||
for (const dir of dirs) {
|
||||
try {
|
||||
await Deno.mkdir(dir, { recursive: true });
|
||||
await probeWrite(dir);
|
||||
} catch (e) {
|
||||
problems.push({ check: "data_dir", message: `数据目录不可写: ${dir}(${errText(e)});检查卷挂载与 PUID/PGID` });
|
||||
}
|
||||
}
|
||||
for (const m of opts.mounts) {
|
||||
try {
|
||||
const st = await Deno.stat(m.path);
|
||||
if (!st.isDirectory) throw Object.assign(new Error("不是目录"), { code: "ENOTDIR" });
|
||||
for await (const _ of Deno.readDir(m.path)) break;
|
||||
} catch (e) {
|
||||
warnings.push({ ...m, reason: `不可读: ${errText(e)}` });
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
await probeWrite(m.path);
|
||||
} catch (e) {
|
||||
warnings.push({ ...m, reason: `不可写: ${errText(e)}` });
|
||||
}
|
||||
}
|
||||
return { problems, warnings };
|
||||
}
|
||||
|
||||
/** 存储告警写审计(actor=system),失败不影响启动 */
|
||||
export async function auditStorageWarnings(db: Kysely<DB>, warnings: StorageWarning[]): Promise<void> {
|
||||
if (!warnings.length) return;
|
||||
await db
|
||||
.insertInto("audit_event")
|
||||
.values(
|
||||
warnings.map((w) => ({
|
||||
actor_type: "system",
|
||||
actor_id: "preflight",
|
||||
action: "preflight.path_unavailable",
|
||||
target_type: w.kind,
|
||||
target_id: w.id,
|
||||
payload: JSON.stringify({ name: w.name, path: w.path, reason: w.reason, uid: Deno.uid() }),
|
||||
})),
|
||||
)
|
||||
.execute();
|
||||
}
|
||||
Reference in new issue
Block a user