root
229 lines
15 KiB
TypeScript
229 lines
15 KiB
TypeScript
/*
|
|
* SPEC: PJ2026-010102 HWPOD 工具；PJ2026-010103 HWPOD 服务；
|
|
* PJ2026-01010305 71FREQ 预装 draft-2026-06-26-71freq-v03-hwpod-preinstall。
|
|
* 责任: 为 L0-L3 提供唯一 host PostgreSQL runtime spec registry。
|
|
*/
|
|
|
|
import { createHash } from "node:crypto";
|
|
import { readdirSync, readFileSync } from "node:fs";
|
|
import path from "node:path";
|
|
import pg from "pg";
|
|
import { parse as parseYaml } from "yaml";
|
|
|
|
import { buildPostgresPoolConfig, postgresSslModeFromConnectionString } from "../db/runtime-store-core.ts";
|
|
import { normalizeHwpodSpec } from "./spec-document.ts";
|
|
|
|
const { Pool } = pg;
|
|
const CONTRACT_VERSION = "hwpod-spec-registry-v3";
|
|
|
|
type BuiltInInput = { configRef: string; sha256?: string; document: unknown };
|
|
export type HwpodRuntimeSpecRecord = { authority: "runtime"; hwpodId: string; createdAt: string; updatedAt: string; document: unknown };
|
|
export interface HwpodRuntimeSpecStore {
|
|
list(): Promise<HwpodRuntimeSpecRecord[]>;
|
|
get(hwpodId: string): Promise<HwpodRuntimeSpecRecord | null>;
|
|
create(record: HwpodRuntimeSpecRecord): Promise<boolean>;
|
|
update(hwpodId: string, document: unknown, updatedAt: string): Promise<HwpodRuntimeSpecRecord | null>;
|
|
delete(hwpodId: string): Promise<HwpodRuntimeSpecRecord | null>;
|
|
close?(): Promise<void>;
|
|
}
|
|
|
|
export function hwpodSpecRegistryFromEnv(env: Record<string, string | undefined> = process.env) {
|
|
const databaseUrl = requiredEnv(env, "HWPOD_SPEC_DATABASE_URL");
|
|
const schema = postgresIdentifier(requiredEnv(env, "HWPOD_SPEC_DATABASE_SCHEMA"), "HWPOD_SPEC_DATABASE_SCHEMA");
|
|
const table = postgresIdentifier(requiredEnv(env, "HWPOD_SPEC_DATABASE_TABLE"), "HWPOD_SPEC_DATABASE_TABLE");
|
|
const connectionTimeoutMs = positiveIntegerEnv(env, "HWPOD_SPEC_DATABASE_CONNECTION_TIMEOUT_MS");
|
|
const builtIns = parseBuiltIns(env.HWPOD_BUILTIN_SPEC_DOCUMENTS_JSON, env.HWPOD_BUILTIN_SPEC_CONFIG_DIRECTORY);
|
|
return createHwpodSpecRegistry({ store: createPostgresHwpodRuntimeSpecStore({ databaseUrl, schema, table, connectionTimeoutMs }), builtIns });
|
|
}
|
|
|
|
export function createHwpodSpecRegistry(options: { store: HwpodRuntimeSpecStore; builtIns?: BuiltInInput[]; now?: () => string }) {
|
|
const store = options.store;
|
|
const now = options.now ?? (() => new Date().toISOString());
|
|
const builtIns = normalizeBuiltIns(options.builtIns ?? []);
|
|
|
|
async function list() {
|
|
const runtime = new Map((await store.list()).map((record) => [record.hwpodId, record]));
|
|
assertNoAuthorityConflicts(builtIns, runtime);
|
|
const specs = [...builtIns.values(), ...runtime.values()].map(summary).sort((left, right) => left.hwpodId.localeCompare(right.hwpodId));
|
|
return { ok: true, status: specs.length === 0 ? "empty" : "completed", contractVersion: CONTRACT_VERSION, count: specs.length, specs, valuesPrinted: false };
|
|
}
|
|
|
|
async function get(hwpodId: string) {
|
|
const id = validId(hwpodId);
|
|
const builtIn = builtIns.get(id);
|
|
if (builtIn) return entity(builtIn);
|
|
const runtime = await store.get(id);
|
|
if (!runtime) throw codedError("hwpod_spec_not_found", `HWPOD spec was not found: ${id}`, { hwpodId: id });
|
|
return entity(runtime);
|
|
}
|
|
|
|
async function create(documentInput: unknown) {
|
|
const normalized = normalizedDocument(documentInput, "runtime:create");
|
|
const id = specId(normalized);
|
|
assertNotFrozen(id, builtIns);
|
|
const observedAt = now();
|
|
const record: HwpodRuntimeSpecRecord = { authority: "runtime", hwpodId: id, createdAt: observedAt, updatedAt: observedAt, document: normalized };
|
|
if (!await store.create(record)) throw codedError("hwpod_spec_exists", `HWPOD runtime spec already exists: ${id}`, { hwpodId: id });
|
|
return mutation("created", record);
|
|
}
|
|
|
|
async function update(hwpodId: string, documentInput: unknown) {
|
|
const id = validId(hwpodId);
|
|
assertNotFrozen(id, builtIns);
|
|
const normalized = normalizedDocument(documentInput, `runtime:${id}`);
|
|
const documentId = specId(normalized);
|
|
if (documentId !== id) throw codedError("hwpod_spec_id_mismatch", `HWPOD spec document id ${documentId} does not match route id ${id}`, { hwpodId: id, documentHwpodId: documentId });
|
|
const record = await store.update(id, normalized, now());
|
|
if (!record) throw codedError("hwpod_spec_not_found", `HWPOD runtime spec was not found: ${id}`, { hwpodId: id });
|
|
return mutation("updated", record);
|
|
}
|
|
|
|
async function remove(hwpodId: string) {
|
|
const id = validId(hwpodId);
|
|
assertNotFrozen(id, builtIns);
|
|
const existing = await store.delete(id);
|
|
if (!existing) throw codedError("hwpod_spec_not_found", `HWPOD runtime spec was not found: ${id}`, { hwpodId: id });
|
|
return { ok: true, status: "deleted", contractVersion: CONTRACT_VERSION, hwpodId: id, authority: "runtime", mutable: true, frozen: false, mutation: true, previousFingerprint: documentFingerprint(existing.document), valuesPrinted: false };
|
|
}
|
|
|
|
return { list, get, create, update, delete: remove, close: async () => store.close?.() };
|
|
}
|
|
|
|
export function createPostgresHwpodRuntimeSpecStore(options: { databaseUrl: string; schema: string; table: string; connectionTimeoutMs: number }): HwpodRuntimeSpecStore {
|
|
const schema = postgresIdentifier(options.schema, "schema");
|
|
const table = postgresIdentifier(options.table, "table");
|
|
const qualifiedTable = `${quoteIdentifier(schema)}.${quoteIdentifier(table)}`;
|
|
const pool = new Pool(buildPostgresPoolConfig({
|
|
dbUrl: options.databaseUrl,
|
|
sslMode: postgresSslModeFromConnectionString(options.databaseUrl) ?? "disable",
|
|
timeoutMs: options.connectionTimeoutMs,
|
|
}));
|
|
let ready: Promise<void> | undefined;
|
|
|
|
function ensureSchema() {
|
|
ready ??= migrate();
|
|
return ready;
|
|
}
|
|
|
|
async function migrate() {
|
|
const client = await pool.connect();
|
|
try {
|
|
await client.query("BEGIN");
|
|
await client.query(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(schema)}`);
|
|
await client.query(`CREATE TABLE IF NOT EXISTS ${qualifiedTable} (
|
|
hwpod_id TEXT PRIMARY KEY,
|
|
authority TEXT NOT NULL DEFAULT 'runtime' CHECK (authority = 'runtime'),
|
|
document JSONB NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
)`);
|
|
await client.query(`CREATE INDEX IF NOT EXISTS ${quoteIdentifier(`${table}_updated_at_idx`)} ON ${qualifiedTable} (updated_at DESC, hwpod_id)`);
|
|
await client.query("COMMIT");
|
|
} catch (error) {
|
|
await client.query("ROLLBACK").catch(() => {});
|
|
ready = undefined;
|
|
throw error;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
}
|
|
|
|
return {
|
|
async list() {
|
|
await ensureSchema();
|
|
const result = await pool.query(`SELECT hwpod_id, authority, document, created_at, updated_at FROM ${qualifiedTable} ORDER BY hwpod_id`);
|
|
return result.rows.map(runtimeRow);
|
|
},
|
|
async get(hwpodId) {
|
|
await ensureSchema();
|
|
const result = await pool.query(`SELECT hwpod_id, authority, document, created_at, updated_at FROM ${qualifiedTable} WHERE hwpod_id=$1`, [hwpodId]);
|
|
return result.rows[0] ? runtimeRow(result.rows[0]) : null;
|
|
},
|
|
async create(record) {
|
|
await ensureSchema();
|
|
const result = await pool.query(`INSERT INTO ${qualifiedTable} (hwpod_id, authority, document, created_at, updated_at) VALUES ($1,'runtime',$2::jsonb,$3,$4) ON CONFLICT (hwpod_id) DO NOTHING RETURNING hwpod_id`, [record.hwpodId, JSON.stringify(record.document), record.createdAt, record.updatedAt]);
|
|
return result.rowCount === 1;
|
|
},
|
|
async update(hwpodId, document, updatedAt) {
|
|
await ensureSchema();
|
|
const result = await pool.query(`UPDATE ${qualifiedTable} SET document=$2::jsonb, updated_at=$3 WHERE hwpod_id=$1 AND authority='runtime' RETURNING hwpod_id, authority, document, created_at, updated_at`, [hwpodId, JSON.stringify(document), updatedAt]);
|
|
return result.rows[0] ? runtimeRow(result.rows[0]) : null;
|
|
},
|
|
async delete(hwpodId) {
|
|
await ensureSchema();
|
|
const result = await pool.query(`DELETE FROM ${qualifiedTable} WHERE hwpod_id=$1 AND authority='runtime' RETURNING hwpod_id, authority, document, created_at, updated_at`, [hwpodId]);
|
|
return result.rows[0] ? runtimeRow(result.rows[0]) : null;
|
|
},
|
|
close: async () => pool.end(),
|
|
};
|
|
}
|
|
|
|
function runtimeRow(row: any): HwpodRuntimeSpecRecord {
|
|
if (row?.authority !== "runtime") throw codedError("invalid_hwpod_runtime_spec", `runtime HWPOD spec authority is invalid: ${String(row?.hwpod_id ?? "<unknown>")}`);
|
|
const document = normalizedDocument(row.document, `postgres:${String(row.hwpod_id ?? "")}`);
|
|
const hwpodId = specId(document);
|
|
if (hwpodId !== row.hwpod_id) throw codedError("invalid_hwpod_runtime_spec", `runtime HWPOD spec identity is invalid: ${String(row.hwpod_id ?? "<unknown>")}`);
|
|
return { authority: "runtime", hwpodId, createdAt: isoTimestamp(row.created_at), updatedAt: isoTimestamp(row.updated_at), document };
|
|
}
|
|
|
|
function parseBuiltIns(raw: string | undefined, configDirectory: string | undefined): BuiltInInput[] {
|
|
if (String(raw ?? "").trim() && String(configDirectory ?? "").trim()) throw codedError("invalid_hwpod_builtin_specs", "HWPOD built-in specs must use either JSON or a config directory, not both");
|
|
if (String(configDirectory ?? "").trim()) return parseBuiltInConfigDirectory(String(configDirectory));
|
|
if (!String(raw ?? "").trim()) return [];
|
|
let value: unknown;
|
|
try { value = JSON.parse(String(raw)); }
|
|
catch (error: any) { throw codedError("invalid_hwpod_builtin_specs", `HWPOD_BUILTIN_SPEC_DOCUMENTS_JSON is invalid JSON: ${error?.message ?? String(error)}`); }
|
|
if (!Array.isArray(value)) throw codedError("invalid_hwpod_builtin_specs", "HWPOD_BUILTIN_SPEC_DOCUMENTS_JSON must be an array");
|
|
return value.map((item: any, index) => {
|
|
if (!item || typeof item !== "object" || Array.isArray(item)) throw codedError("invalid_hwpod_builtin_specs", `built-in spec item ${index} must be an object`);
|
|
const configRef = String(item.configRef ?? "").trim();
|
|
if (!configRef) throw codedError("invalid_hwpod_builtin_specs", `built-in spec item ${index} is missing configRef`);
|
|
return { configRef, sha256: String(item.sha256 ?? "").trim() || undefined, document: item.document };
|
|
});
|
|
}
|
|
|
|
function parseBuiltInConfigDirectory(directoryInput: string): BuiltInInput[] {
|
|
const directory = path.resolve(directoryInput.trim());
|
|
return readdirSync(directory, { withFileTypes: true })
|
|
.filter((entry) => entry.isFile() && /\.ya?ml$/iu.test(entry.name))
|
|
.map((entry) => entry.name)
|
|
.sort()
|
|
.map((fileName) => {
|
|
const source = readFileSync(path.join(directory, fileName), "utf8");
|
|
const document = parseYaml(source);
|
|
return {
|
|
configRef: `configMap:hwlab-v03-hwpod-preinstalled-specs/${fileName}`,
|
|
sha256: createHash("sha256").update(stableJson(document)).digest("hex"),
|
|
document,
|
|
};
|
|
});
|
|
}
|
|
|
|
function normalizeBuiltIns(inputs: BuiltInInput[]) {
|
|
const records = new Map<string, any>();
|
|
for (const input of inputs) {
|
|
const document = normalizedDocument(input.document, input.configRef);
|
|
const hwpodId = specId(document);
|
|
if (records.has(hwpodId)) throw codedError("hwpod_spec_authority_conflict", `duplicate YAML-first built-in HWPOD spec: ${hwpodId}`, { hwpodId });
|
|
records.set(hwpodId, { authority: "yaml-first-builtin", hwpodId, configRef: input.configRef, sourceFingerprint: input.sha256 ?? null, document });
|
|
}
|
|
return records;
|
|
}
|
|
|
|
function requiredEnv(env: Record<string, string | undefined>, key: string) { const value = String(env[key] ?? "").trim(); if (!value) throw codedError("hwpod_spec_database_config_required", `${key} is required`); return value; }
|
|
function positiveIntegerEnv(env: Record<string, string | undefined>, key: string) { const value = Number(requiredEnv(env, key)); if (!Number.isSafeInteger(value) || value <= 0) throw codedError("invalid_hwpod_spec_database_config", `${key} must be a positive integer`); return value; }
|
|
function postgresIdentifier(value: string, name: string) { if (!/^[a-z_][a-z0-9_]*$/u.test(value)) throw codedError("invalid_hwpod_spec_database_identifier", `${name} must be a lowercase PostgreSQL identifier`); return value; }
|
|
function quoteIdentifier(value: string) { return `"${postgresIdentifier(value, "identifier")}"`; }
|
|
function isoTimestamp(value: unknown) { const date = value instanceof Date ? value : new Date(String(value ?? "")); if (!Number.isFinite(date.getTime())) throw codedError("invalid_hwpod_runtime_spec", "runtime HWPOD spec timestamp is invalid"); return date.toISOString(); }
|
|
function normalizedDocument(value: unknown, source: string) { return normalizeHwpodSpec(value, source); }
|
|
function specId(document: any) { return validId(String(document?.metadata?.name ?? document?.metadata?.uid ?? "")); }
|
|
function validId(value: string) { const id = String(value ?? "").trim(); if (!/^[A-Za-z0-9_.:-]{1,120}$/u.test(id)) throw codedError("invalid_hwpod_spec_id", `invalid HWPOD spec id: ${id || "<empty>"}`); return id; }
|
|
function assertNotFrozen(id: string, builtIns: Map<string, any>) { const item = builtIns.get(id); if (item) throw codedError("hwpod_spec_frozen", `YAML-first built-in HWPOD spec is frozen: ${id}`, { hwpodId: id, authority: item.authority, configRef: item.configRef, mutable: false, frozen: true, documentFingerprint: documentFingerprint(item.document), mutation: false }); }
|
|
function assertNoAuthorityConflicts(builtIns: Map<string, any>, runtime: Map<string, HwpodRuntimeSpecRecord>) { for (const id of runtime.keys()) if (builtIns.has(id)) throw codedError("hwpod_spec_authority_conflict", `runtime HWPOD spec conflicts with YAML-first built-in spec: ${id}`, { hwpodId: id }); }
|
|
function summary(record: any) { const document = record.document; return { hwpodId: record.hwpodId, authority: record.authority, mutable: record.authority === "runtime", frozen: record.authority !== "runtime", documentFingerprint: documentFingerprint(document), ...(record.configRef ? { configRef: record.configRef, sourceFingerprint: record.sourceFingerprint } : { createdAt: record.createdAt, updatedAt: record.updatedAt }), nodeId: document.spec.nodeBinding.nodeId, workspacePath: document.spec.workspace.path }; }
|
|
function entity(record: any) { return { ok: true, status: "completed", contractVersion: CONTRACT_VERSION, ...summary(record), document: record.document, valuesPrinted: false }; }
|
|
function mutation(status: "created" | "updated", record: HwpodRuntimeSpecRecord) { return { ...entity(record), status, mutation: true }; }
|
|
function documentFingerprint(document: unknown) { return `sha256:${createHash("sha256").update(stableJson(document)).digest("hex")}`; }
|
|
function stableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.keys(value as any).sort().map((key) => `${JSON.stringify(key)}:${stableJson((value as any)[key])}`).join(",")}}`; return JSON.stringify(value) ?? "null"; }
|
|
function codedError(code: string, message: string, details?: unknown) { return Object.assign(new Error(message), { code, details }); }
|