fix: add durable cloud api runtime adapter
This commit is contained in:
Code Queue Review
@@ -0,0 +1,118 @@
|
||||
import { RUNTIME_DURABLE_ADAPTER_MISSING } from "../db/runtime-store.mjs";
|
||||
|
||||
export const CLOUD_API_HEALTH_CONTRACT_VERSION = "v2";
|
||||
|
||||
export function buildCloudApiReadiness({ db, codeAgent, runtime } = {}) {
|
||||
const dbReady = isDbReady(db);
|
||||
const codeAgentReady = isCodeAgentReady(codeAgent);
|
||||
const runtimeReady = isRuntimeReady(runtime);
|
||||
const blockers = [];
|
||||
|
||||
if (!dbReady) {
|
||||
blockers.push(dbBlocker(db));
|
||||
}
|
||||
|
||||
if (!codeAgentReady) {
|
||||
blockers.push(codeAgentBlocker(codeAgent));
|
||||
}
|
||||
|
||||
if (!runtimeReady) {
|
||||
blockers.push(runtimeBlocker(runtime));
|
||||
}
|
||||
|
||||
return {
|
||||
contractVersion: CLOUD_API_HEALTH_CONTRACT_VERSION,
|
||||
ready: blockers.length === 0,
|
||||
status: blockers.length === 0 ? "ok" : "degraded",
|
||||
blockerCodes: blockers.map((blocker) => blocker.code),
|
||||
blockers,
|
||||
components: {
|
||||
db: dbReady ? "ready" : "blocked",
|
||||
codeAgent: codeAgentReady ? "ready" : "blocked",
|
||||
runtime: runtimeReady ? "ready" : dbReady ? "blocked" : "waiting_for_db"
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function isDbReady(db) {
|
||||
return Boolean(db?.ready && db?.connected && db?.liveConnected !== false && db?.liveDbEvidence);
|
||||
}
|
||||
|
||||
function isCodeAgentReady(codeAgent) {
|
||||
return Boolean(codeAgent?.ready && codeAgent?.status !== "blocked");
|
||||
}
|
||||
|
||||
function isRuntimeReady(runtime) {
|
||||
if (runtime?.durable !== true) {
|
||||
return false;
|
||||
}
|
||||
return !["blocked", "degraded", "failed"].includes(runtime.status);
|
||||
}
|
||||
|
||||
function dbBlocker(db) {
|
||||
return {
|
||||
code: db?.configReady ? "db_live_connection_blocked" : "db_runtime_config_blocked",
|
||||
type: "runtime_blocker",
|
||||
scope: "cloud-api-db",
|
||||
status: "open",
|
||||
sourceIssue: "pikasTech/HWLAB#49",
|
||||
summary: db?.blocker ?? "cloud-api DB live readiness is not established",
|
||||
evidence: {
|
||||
configReady: Boolean(db?.configReady),
|
||||
connected: Boolean(db?.connected),
|
||||
liveConnected: Boolean(db?.liveConnected),
|
||||
liveDbEvidence: Boolean(db?.liveDbEvidence),
|
||||
connectionAttempted: Boolean(db?.connectionAttempted ?? db?.connectionChecked),
|
||||
connectionResult: db?.connectionResult ?? db?.connection?.result ?? "unknown"
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function codeAgentBlocker(codeAgent) {
|
||||
return {
|
||||
code: codeAgent?.reason ?? "code_agent_provider_unavailable",
|
||||
type: "agent_blocker",
|
||||
scope: "code-agent-provider",
|
||||
status: "open",
|
||||
sourceIssue: "pikasTech/HWLAB#143",
|
||||
summary: codeAgent?.summary ?? "Code Agent provider readiness is not established",
|
||||
evidence: {
|
||||
provider: codeAgent?.provider ?? "unknown",
|
||||
backend: codeAgent?.backend ?? "unknown",
|
||||
missingEnv: Array.isArray(codeAgent?.missingEnv) ? [...codeAgent.missingEnv] : [],
|
||||
secretsRedacted: true
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function runtimeBlocker(runtime) {
|
||||
const durable = runtime?.durable === true;
|
||||
const unhealthy = durable && ["blocked", "degraded", "failed"].includes(runtime?.status);
|
||||
const blocker = runtime?.blocker;
|
||||
return {
|
||||
code: durable && unhealthy ? blocker ?? "runtime_durable_adapter_unhealthy" : RUNTIME_DURABLE_ADAPTER_MISSING,
|
||||
type: "runtime_blocker",
|
||||
scope: "runtime-durable-adapter",
|
||||
status: "open",
|
||||
sourceIssue: "pikasTech/HWLAB#164",
|
||||
summary: durable && unhealthy
|
||||
? runtime?.reason ?? "cloud-api durable runtime adapter is configured but not ready"
|
||||
: "cloud-api DB is live, but L1 runtime writes still use process-local memory because no DB-backed durable adapter is configured",
|
||||
evidence: {
|
||||
adapter: runtime?.adapter ?? "unknown",
|
||||
durable: Boolean(runtime?.durable),
|
||||
runtimeStatus: runtime?.status ?? "unknown",
|
||||
runtimeReady: Boolean(runtime?.ready),
|
||||
liveRuntimeEvidence: Boolean(runtime?.liveRuntimeEvidence),
|
||||
schemaReady: Boolean(runtime?.schema?.ready),
|
||||
schemaChecked: Boolean(runtime?.schema?.checked),
|
||||
queryAttempted: Boolean(runtime?.connection?.queryAttempted),
|
||||
queryResult: runtime?.connection?.queryResult ?? "unknown",
|
||||
dbBackedAdapterRequired: true,
|
||||
secretMaterialRead: false
|
||||
},
|
||||
nextTask: durable
|
||||
? "Apply/repair the runtime DB schema or DB auth, then rerun /health/live without printing DB secret material."
|
||||
: "Set HWLAB_CLOUD_RUNTIME_ADAPTER=postgres after the runtime DB schema is available; keep HWLAB_CLOUD_DB_URL injected from Secret."
|
||||
};
|
||||
}
|
||||
+42
-16
@@ -15,6 +15,8 @@ import {
|
||||
deriveActorFromMeta
|
||||
} from "../audit/index.mjs";
|
||||
import { buildDbRuntimeReadiness } from "./db-contract.mjs";
|
||||
import { describeCodeAgentAvailability } from "./code-agent-chat.mjs";
|
||||
import { buildCloudApiReadiness } from "./health-contract.mjs";
|
||||
import { createCloudRuntimeStore } from "../db/runtime-store.mjs";
|
||||
|
||||
export const SUPPORTED_RPC_METHODS = Object.freeze([
|
||||
@@ -34,15 +36,15 @@ export const SUPPORTED_RPC_METHODS = Object.freeze([
|
||||
const rpcHandlers = new Map([
|
||||
["system.health", handleSystemHealth],
|
||||
["cloud.adapter.describe", handleAdapterDescribe],
|
||||
["gateway.session.register", (params, envelope, context) => getRuntimeStore(context).registerGatewaySession(params, envelope.meta)],
|
||||
["box.resource.register", (params, envelope, context) => getRuntimeStore(context).registerBoxResource(params, envelope.meta)],
|
||||
["box.capability.report", (params, envelope, context) => getRuntimeStore(context).reportBoxCapabilities(params, envelope.meta)],
|
||||
["hardware.operation.request", (params, envelope, context) => getRuntimeStore(context).requestHardwareOperation(params, envelope.meta)],
|
||||
["hardware.invoke.shell", (params, envelope, context) => getRuntimeStore(context).invokeHardwareShell(params, envelope.meta)],
|
||||
["audit.event.write", (params, envelope, context) => getRuntimeStore(context).writeAuditEvent(params, envelope.meta)],
|
||||
["audit.event.query", (params, envelope, context) => getRuntimeStore(context).queryAuditEvents(params)],
|
||||
["evidence.record.write", (params, envelope, context) => getRuntimeStore(context).writeEvidenceRecord(params, envelope.meta)],
|
||||
["evidence.record.query", (params, envelope, context) => getRuntimeStore(context).queryEvidenceRecords(params)]
|
||||
["gateway.session.register", async (params, envelope, context) => getRuntimeStore(context).registerGatewaySession(params, envelope.meta)],
|
||||
["box.resource.register", async (params, envelope, context) => getRuntimeStore(context).registerBoxResource(params, envelope.meta)],
|
||||
["box.capability.report", async (params, envelope, context) => getRuntimeStore(context).reportBoxCapabilities(params, envelope.meta)],
|
||||
["hardware.operation.request", async (params, envelope, context) => getRuntimeStore(context).requestHardwareOperation(params, envelope.meta)],
|
||||
["hardware.invoke.shell", async (params, envelope, context) => getRuntimeStore(context).invokeHardwareShell(params, envelope.meta)],
|
||||
["audit.event.write", async (params, envelope, context) => getRuntimeStore(context).writeAuditEvent(params, envelope.meta)],
|
||||
["audit.event.query", async (params, envelope, context) => getRuntimeStore(context).queryAuditEvents(params)],
|
||||
["evidence.record.write", async (params, envelope, context) => getRuntimeStore(context).writeEvidenceRecord(params, envelope.meta)],
|
||||
["evidence.record.query", async (params, envelope, context) => getRuntimeStore(context).queryEvidenceRecords(params)]
|
||||
]);
|
||||
|
||||
const defaultRuntimeStore = createCloudRuntimeStore();
|
||||
@@ -218,24 +220,45 @@ export function createResponseMeta(requestMeta = {}) {
|
||||
}
|
||||
|
||||
async function handleSystemHealth(params, envelope, context) {
|
||||
const env = context.env ?? process.env;
|
||||
const db = await buildDbRuntimeReadiness(env, context.dbProbe);
|
||||
const codeAgent = describeCodeAgentAvailability(env, context);
|
||||
const runtime = await runtimeReadiness(getRuntimeStore(context));
|
||||
const readiness = buildCloudApiReadiness({ db, codeAgent, runtime });
|
||||
return {
|
||||
status: healthStatus(),
|
||||
status: readiness.status,
|
||||
ready: readiness.ready,
|
||||
serviceId: CLOUD_API_SERVICE_ID,
|
||||
environment: ENVIRONMENT_DEV,
|
||||
db: await buildDbRuntimeReadiness(process.env, context.dbProbe),
|
||||
runtime: getRuntimeStore(context).summary()
|
||||
db,
|
||||
codeAgent,
|
||||
runtime,
|
||||
readiness,
|
||||
blockers: readiness.blockers,
|
||||
blockerCodes: readiness.blockerCodes
|
||||
};
|
||||
}
|
||||
|
||||
async function handleAdapterDescribe(params, envelope, context) {
|
||||
const env = context.env ?? process.env;
|
||||
const db = await buildDbRuntimeReadiness(env, context.dbProbe);
|
||||
const codeAgent = describeCodeAgentAvailability(env, context);
|
||||
const runtime = await runtimeReadiness(getRuntimeStore(context));
|
||||
const readiness = buildCloudApiReadiness({ db, codeAgent, runtime });
|
||||
return {
|
||||
serviceId: CLOUD_API_SERVICE_ID,
|
||||
status: readiness.status,
|
||||
ready: readiness.ready,
|
||||
rpcEndpoint: "POST /rpc",
|
||||
restEndpoint: "POST /v1/rpc/{method}",
|
||||
methods: SUPPORTED_RPC_METHODS,
|
||||
auditFields: AUDIT_FIELD_NAMES,
|
||||
db: await buildDbRuntimeReadiness(process.env, context.dbProbe),
|
||||
runtime: getRuntimeStore(context).summary()
|
||||
db,
|
||||
codeAgent,
|
||||
runtime,
|
||||
readiness,
|
||||
blockers: readiness.blockers,
|
||||
blockerCodes: readiness.blockerCodes
|
||||
};
|
||||
}
|
||||
|
||||
@@ -243,8 +266,11 @@ function getRuntimeStore(context = {}) {
|
||||
return context.runtimeStore || defaultRuntimeStore;
|
||||
}
|
||||
|
||||
function healthStatus() {
|
||||
return "degraded";
|
||||
async function runtimeReadiness(runtimeStore) {
|
||||
if (typeof runtimeStore?.readiness === "function") {
|
||||
return runtimeStore.readiness();
|
||||
}
|
||||
return runtimeStore.summary();
|
||||
}
|
||||
|
||||
function extractMeta(envelope) {
|
||||
|
||||
+41
-13
@@ -14,15 +14,17 @@ import {
|
||||
} from "./json-rpc.mjs";
|
||||
import { describeCodeAgentAvailability, handleCodeAgentChat } from "./code-agent-chat.mjs";
|
||||
import { buildDbRuntimeReadiness } from "./db-contract.mjs";
|
||||
import { createCloudRuntimeStore } from "../db/runtime-store.mjs";
|
||||
import { buildCloudApiReadiness } from "./health-contract.mjs";
|
||||
import { createConfiguredCloudRuntimeStore } from "../db/runtime-store.mjs";
|
||||
|
||||
const DEFAULT_BODY_LIMIT_BYTES = 1024 * 1024;
|
||||
|
||||
export function createCloudApiServer(options = {}) {
|
||||
const runtimeStore = options.runtimeStore || createCloudRuntimeStore();
|
||||
const env = options.env ?? process.env;
|
||||
const runtimeStore = options.runtimeStore || createConfiguredCloudRuntimeStore({ ...options, env });
|
||||
return createServer(async (request, response) => {
|
||||
try {
|
||||
await routeRequest(request, response, { ...options, runtimeStore });
|
||||
await routeRequest(request, response, { ...options, env, runtimeStore });
|
||||
} catch (error) {
|
||||
sendJson(response, 500, {
|
||||
error: {
|
||||
@@ -37,16 +39,23 @@ export function createCloudApiServer(options = {}) {
|
||||
|
||||
export async function buildHealthPayload(options = {}) {
|
||||
const serviceId = CLOUD_API_SERVICE_ID;
|
||||
const env = options.env ?? process.env;
|
||||
const commitId = process.env.HWLAB_COMMIT_ID || process.env.HWLAB_GIT_SHA || "unknown";
|
||||
const imageReference = process.env.HWLAB_IMAGE || "unknown";
|
||||
const imageTag = process.env.HWLAB_IMAGE_TAG || commitId.slice(0, 7) || "unknown";
|
||||
const db = await buildDbRuntimeReadiness(process.env, options.dbProbe);
|
||||
const codeAgent = describeCodeAgentAvailability(options.env ?? process.env, options);
|
||||
const db = await buildDbRuntimeReadiness(env, options.dbProbe);
|
||||
const codeAgent = describeCodeAgentAvailability(env, options);
|
||||
const runtime = await runtimeReadiness(options.runtimeStore ?? createConfiguredCloudRuntimeStore({ ...options, env }));
|
||||
const readiness = buildCloudApiReadiness({ db, codeAgent, runtime });
|
||||
|
||||
return {
|
||||
serviceId,
|
||||
environment: ENVIRONMENT_DEV,
|
||||
status: db.status === "blocked" ? "degraded" : "degraded",
|
||||
status: readiness.status,
|
||||
ready: readiness.ready,
|
||||
readiness,
|
||||
blockers: readiness.blockers,
|
||||
blockerCodes: readiness.blockerCodes,
|
||||
service: {
|
||||
id: serviceId,
|
||||
role: "cloud-api",
|
||||
@@ -66,7 +75,7 @@ export async function buildHealthPayload(options = {}) {
|
||||
observedAt: new Date().toISOString(),
|
||||
db,
|
||||
codeAgent,
|
||||
runtime: options.runtimeStore?.summary?.() ?? createCloudRuntimeStore().summary()
|
||||
runtime
|
||||
};
|
||||
}
|
||||
|
||||
@@ -137,23 +146,33 @@ async function handleRpcHttpRequest(request, response, options) {
|
||||
|
||||
const rpcResponse = await handleJsonRpcRequest(envelope, {
|
||||
adapter: "json-rpc",
|
||||
runtimeStore: options.runtimeStore
|
||||
runtimeStore: options.runtimeStore,
|
||||
env: options.env,
|
||||
dbProbe: options.dbProbe
|
||||
});
|
||||
sendJson(response, statusForRpcResponse(rpcResponse), rpcResponse);
|
||||
}
|
||||
|
||||
async function handleRestAdapter(request, response, url, options) {
|
||||
if (request.method === "GET" && url.pathname === "/v1") {
|
||||
const db = await buildDbRuntimeReadiness(options.env ?? process.env, options.dbProbe);
|
||||
const codeAgent = describeCodeAgentAvailability(options.env ?? process.env, options);
|
||||
const runtime = await runtimeReadiness(options.runtimeStore);
|
||||
const readiness = buildCloudApiReadiness({ db, codeAgent, runtime });
|
||||
sendJson(response, 200, {
|
||||
serviceId: CLOUD_API_SERVICE_ID,
|
||||
adapter: "rest",
|
||||
status: "degraded",
|
||||
status: readiness.status,
|
||||
ready: readiness.ready,
|
||||
rpcBridge: "POST /v1/rpc/{method}",
|
||||
codeAgent: describeCodeAgentAvailability(options.env ?? process.env, options),
|
||||
codeAgent,
|
||||
methods: SUPPORTED_RPC_METHODS,
|
||||
auditFields: AUDIT_FIELD_NAMES,
|
||||
db: await buildDbRuntimeReadiness(process.env, options.dbProbe),
|
||||
runtime: options.runtimeStore.summary()
|
||||
db,
|
||||
runtime,
|
||||
readiness,
|
||||
blockers: readiness.blockers,
|
||||
blockerCodes: readiness.blockerCodes
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -220,7 +239,9 @@ async function handleRestAdapter(request, response, url, options) {
|
||||
},
|
||||
{
|
||||
adapter: "rest",
|
||||
runtimeStore: options.runtimeStore
|
||||
runtimeStore: options.runtimeStore,
|
||||
env: options.env,
|
||||
dbProbe: options.dbProbe
|
||||
}
|
||||
);
|
||||
|
||||
@@ -369,3 +390,10 @@ function getHeader(request, name) {
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
async function runtimeReadiness(runtimeStore) {
|
||||
if (typeof runtimeStore?.readiness === "function") {
|
||||
return runtimeStore.readiness();
|
||||
}
|
||||
return runtimeStore.summary();
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ import { createServer as createTcpServer } from "node:net";
|
||||
import test from "node:test";
|
||||
|
||||
import { createCloudApiServer } from "./server.mjs";
|
||||
import { RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED } from "../db/runtime-store.mjs";
|
||||
|
||||
test("cloud api exposes /health, /health/live, and /live probes", async () => {
|
||||
const originalUrl = process.env.HWLAB_CLOUD_DB_URL;
|
||||
@@ -96,7 +97,7 @@ test("cloud api health reports DB env presence and live connection classificatio
|
||||
const fakeDb = createTcpServer((socket) => socket.end());
|
||||
await new Promise((resolve) => fakeDb.listen(0, "127.0.0.1", resolve));
|
||||
const dbPort = fakeDb.address().port;
|
||||
process.env.HWLAB_CLOUD_DB_URL = `postgres://user:password@127.0.0.1:${dbPort}/hwlab`;
|
||||
process.env.HWLAB_CLOUD_DB_URL = `postgres://hwlab_test@127.0.0.1:${dbPort}/hwlab`;
|
||||
process.env.HWLAB_CLOUD_DB_SSL_MODE = "require";
|
||||
|
||||
const server = createCloudApiServer();
|
||||
@@ -152,6 +153,81 @@ test("cloud api health reports DB env presence and live connection classificatio
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api health separates DB connected from durable runtime schema readiness", async () => {
|
||||
const originalUrl = process.env.HWLAB_CLOUD_DB_URL;
|
||||
const originalSslMode = process.env.HWLAB_CLOUD_DB_SSL_MODE;
|
||||
const fakeDb = createTcpServer((socket) => socket.end());
|
||||
await new Promise((resolve) => fakeDb.listen(0, "127.0.0.1", resolve));
|
||||
const dbPort = fakeDb.address().port;
|
||||
process.env.HWLAB_CLOUD_DB_URL = `postgres://hwlab_test@127.0.0.1:${dbPort}/hwlab`;
|
||||
process.env.HWLAB_CLOUD_DB_SSL_MODE = "require";
|
||||
|
||||
const server = createCloudApiServer({
|
||||
runtimeStore: {
|
||||
async readiness() {
|
||||
return {
|
||||
adapter: "postgres",
|
||||
durable: true,
|
||||
ready: false,
|
||||
status: "blocked",
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
|
||||
reason: "test runtime schema is missing",
|
||||
liveRuntimeEvidence: false,
|
||||
fixtureEvidence: false,
|
||||
connection: {
|
||||
queryAttempted: true,
|
||||
queryResult: "schema_blocked",
|
||||
endpointRedacted: true,
|
||||
valueRedacted: true
|
||||
},
|
||||
schema: {
|
||||
checked: true,
|
||||
ready: false,
|
||||
missingTables: ["gateway_sessions"],
|
||||
missingColumns: ["gateway_sessions.gateway_session_json"]
|
||||
},
|
||||
counts: {}
|
||||
};
|
||||
}
|
||||
}
|
||||
});
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
try {
|
||||
const { port } = server.address();
|
||||
const response = await fetch(`http://127.0.0.1:${port}/health/live`);
|
||||
const payload = await response.json();
|
||||
assert.equal(response.status, 200);
|
||||
assert.equal(payload.status, "degraded");
|
||||
assert.equal(payload.db.connected, true);
|
||||
assert.equal(payload.db.liveDbEvidence, true);
|
||||
assert.equal(payload.runtime.durable, true);
|
||||
assert.equal(payload.runtime.ready, false);
|
||||
assert.equal(payload.runtime.blocker, RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED);
|
||||
assert.equal(payload.readiness.components.db, "ready");
|
||||
assert.equal(payload.readiness.components.runtime, "blocked");
|
||||
assert.ok(payload.blockerCodes.includes(RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED));
|
||||
assert.equal(JSON.stringify(payload).includes("password"), false);
|
||||
} finally {
|
||||
if (originalUrl === undefined) {
|
||||
delete process.env.HWLAB_CLOUD_DB_URL;
|
||||
} else {
|
||||
process.env.HWLAB_CLOUD_DB_URL = originalUrl;
|
||||
}
|
||||
if (originalSslMode === undefined) {
|
||||
delete process.env.HWLAB_CLOUD_DB_SSL_MODE;
|
||||
} else {
|
||||
process.env.HWLAB_CLOUD_DB_SSL_MODE = originalSslMode;
|
||||
}
|
||||
await new Promise((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
await new Promise((resolve, reject) => {
|
||||
fakeDb.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api health does not treat DB env presence-only as live readiness", async () => {
|
||||
const originalUrl = process.env.HWLAB_CLOUD_DB_URL;
|
||||
const originalSslMode = process.env.HWLAB_CLOUD_DB_SSL_MODE;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
-- HWLAB L1 Cloud Core schema skeleton.
|
||||
-- This file is a contract placeholder only; it does not connect to or write a
|
||||
-- live database.
|
||||
-- This file is source schema only; applying it to DEV/PROD is a separate
|
||||
-- operator action and is not performed by source tests or smokes.
|
||||
|
||||
CREATE TABLE IF NOT EXISTS projects (
|
||||
id TEXT PRIMARY KEY,
|
||||
@@ -16,7 +16,8 @@ CREATE TABLE IF NOT EXISTS gateway_sessions (
|
||||
gateway_service_id TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
started_at TEXT,
|
||||
ended_at TEXT
|
||||
ended_at TEXT,
|
||||
gateway_session_json TEXT NOT NULL DEFAULT '{}'
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS box_resources (
|
||||
@@ -25,6 +26,7 @@ CREATE TABLE IF NOT EXISTS box_resources (
|
||||
gateway_session_id TEXT,
|
||||
resource_state TEXT NOT NULL DEFAULT 'available',
|
||||
labels_json TEXT NOT NULL DEFAULT '{}',
|
||||
resource_json TEXT NOT NULL DEFAULT '{}',
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
|
||||
@@ -13,11 +13,108 @@ import {
|
||||
} from "../audit/index.mjs";
|
||||
|
||||
export const RUNTIME_STORE_KIND = "memory";
|
||||
export const RUNTIME_STORE_KIND_POSTGRES = "postgres";
|
||||
export const RUNTIME_DURABLE_ADAPTER_MISSING = "runtime_durable_adapter_missing";
|
||||
export const RUNTIME_DURABLE_ADAPTER_UNCONFIGURED = "runtime_durable_adapter_unconfigured";
|
||||
export const RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING = "runtime_durable_adapter_driver_missing";
|
||||
export const RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED = "runtime_durable_adapter_auth_blocked";
|
||||
export const RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED = "runtime_durable_adapter_schema_blocked";
|
||||
export const RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED = "runtime_durable_adapter_query_blocked";
|
||||
|
||||
export const RUNTIME_ADAPTER_ENV = "HWLAB_CLOUD_RUNTIME_ADAPTER";
|
||||
export const RUNTIME_DURABLE_ENV = "HWLAB_CLOUD_RUNTIME_DURABLE";
|
||||
|
||||
const requiredPostgresSchema = Object.freeze({
|
||||
gateway_sessions: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"gateway_service_id",
|
||||
"status",
|
||||
"started_at",
|
||||
"ended_at",
|
||||
"gateway_session_json"
|
||||
]),
|
||||
box_resources: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"gateway_session_id",
|
||||
"resource_state",
|
||||
"labels_json",
|
||||
"resource_json",
|
||||
"updated_at"
|
||||
]),
|
||||
box_capabilities: Object.freeze([
|
||||
"id",
|
||||
"box_resource_id",
|
||||
"capability_type",
|
||||
"capability_json",
|
||||
"updated_at"
|
||||
]),
|
||||
hardware_operations: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"requested_by",
|
||||
"operation_type",
|
||||
"operation_json",
|
||||
"status",
|
||||
"requested_at",
|
||||
"updated_at"
|
||||
]),
|
||||
audit_events: Object.freeze([
|
||||
"id",
|
||||
"request_id",
|
||||
"actor",
|
||||
"source",
|
||||
"operation",
|
||||
"target",
|
||||
"result",
|
||||
"timestamp",
|
||||
"event_json"
|
||||
]),
|
||||
evidence_records: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"operation_id",
|
||||
"evidence_type",
|
||||
"uri",
|
||||
"metadata_json",
|
||||
"created_at"
|
||||
])
|
||||
});
|
||||
|
||||
const postgresCountTables = Object.freeze([
|
||||
"gateway_sessions",
|
||||
"box_resources",
|
||||
"box_capabilities",
|
||||
"hardware_operations",
|
||||
"audit_events",
|
||||
"evidence_records"
|
||||
]);
|
||||
|
||||
export function createCloudRuntimeStore(options = {}) {
|
||||
return new CloudRuntimeStore(options);
|
||||
}
|
||||
|
||||
export function createConfiguredCloudRuntimeStore(options = {}) {
|
||||
const env = options.env ?? process.env;
|
||||
const adapter = normalizeRuntimeAdapter(options.adapter ?? env?.[RUNTIME_ADAPTER_ENV]);
|
||||
const durableRequested =
|
||||
options.durable === true ||
|
||||
env?.[RUNTIME_DURABLE_ENV] === "1" ||
|
||||
env?.[RUNTIME_DURABLE_ENV] === "true";
|
||||
|
||||
if (adapter === RUNTIME_STORE_KIND_POSTGRES || durableRequested) {
|
||||
return new PostgresCloudRuntimeStore({
|
||||
...options,
|
||||
env,
|
||||
dbUrl: options.dbUrl ?? env?.HWLAB_CLOUD_DB_URL,
|
||||
sslMode: options.sslMode ?? env?.HWLAB_CLOUD_DB_SSL_MODE
|
||||
});
|
||||
}
|
||||
|
||||
return createCloudRuntimeStore(options);
|
||||
}
|
||||
|
||||
export class CloudRuntimeStore {
|
||||
constructor({ now = () => new Date().toISOString() } = {}) {
|
||||
this.now = now;
|
||||
@@ -34,7 +131,13 @@ export class CloudRuntimeStore {
|
||||
adapter: RUNTIME_STORE_KIND,
|
||||
durable: false,
|
||||
status: "degraded",
|
||||
reason: "live DB persistence is not connected; L1 runtime writes are process-local only",
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_MISSING,
|
||||
reason: "L1 runtime writes are process-local only; no DB-backed durable runtime adapter is configured",
|
||||
adapterContract: {
|
||||
required: "DB-backed durable runtime adapter using HWLAB_CLOUD_DB_URL",
|
||||
sourceIssue: "pikasTech/HWLAB#164",
|
||||
secretMaterialRequiredInSource: false
|
||||
},
|
||||
counts: {
|
||||
gatewaySessions: this.gatewaySessions.size,
|
||||
boxResources: this.boxResources.size,
|
||||
@@ -510,6 +613,455 @@ export class CloudRuntimeStore {
|
||||
}
|
||||
}
|
||||
|
||||
export class PostgresCloudRuntimeStore {
|
||||
constructor({
|
||||
env = process.env,
|
||||
dbUrl,
|
||||
sslMode = "require",
|
||||
now = () => new Date().toISOString(),
|
||||
queryClient,
|
||||
pgModuleLoader
|
||||
} = {}) {
|
||||
this.env = env;
|
||||
this.dbUrl = dbUrl;
|
||||
this.sslMode = sslMode;
|
||||
this.now = now;
|
||||
this.queryClient = queryClient;
|
||||
this.pgModuleLoader = pgModuleLoader;
|
||||
this.pool = null;
|
||||
this.memory = new CloudRuntimeStore({ now });
|
||||
this.lastReadiness = this.blockedSummary({
|
||||
blocker: dbUrl || queryClient ? "runtime durable adapter has not completed schema readiness" : RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
|
||||
reason: dbUrl || queryClient
|
||||
? "Postgres runtime adapter is configured but has not completed a live schema query"
|
||||
: "Postgres runtime adapter requires HWLAB_CLOUD_DB_URL or an injected query client"
|
||||
});
|
||||
}
|
||||
|
||||
summary() {
|
||||
return {
|
||||
...this.lastReadiness,
|
||||
counts: this.lastReadiness.counts ?? this.memory.summary().counts
|
||||
};
|
||||
}
|
||||
|
||||
async readiness() {
|
||||
if (!this.dbUrl && !this.queryClient) {
|
||||
this.lastReadiness = this.blockedSummary({
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_UNCONFIGURED,
|
||||
reason: "Postgres runtime adapter is selected but HWLAB_CLOUD_DB_URL is not injected"
|
||||
});
|
||||
return this.summary();
|
||||
}
|
||||
|
||||
let rows;
|
||||
try {
|
||||
const result = await this.query(
|
||||
"SELECT table_name, column_name FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ANY($1::text[])",
|
||||
[Object.keys(requiredPostgresSchema)]
|
||||
);
|
||||
rows = Array.isArray(result?.rows) ? result.rows : [];
|
||||
} catch (error) {
|
||||
this.lastReadiness = this.blockedSummary(classifyRuntimeDbError(error));
|
||||
return this.summary();
|
||||
}
|
||||
|
||||
const schema = summarizeRuntimeSchema(rows);
|
||||
if (!schema.ready) {
|
||||
this.lastReadiness = this.blockedSummary({
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
|
||||
reason: "Postgres runtime adapter connected, but required runtime tables or columns are missing",
|
||||
schema
|
||||
});
|
||||
return this.summary();
|
||||
}
|
||||
|
||||
const counts = await this.readCounts().catch(() => null);
|
||||
this.lastReadiness = {
|
||||
adapter: RUNTIME_STORE_KIND_POSTGRES,
|
||||
durable: true,
|
||||
ready: true,
|
||||
status: "ready",
|
||||
blocker: null,
|
||||
reason: "Postgres durable runtime adapter completed a live schema readiness query",
|
||||
liveRuntimeEvidence: true,
|
||||
fixtureEvidence: false,
|
||||
connection: {
|
||||
queryAttempted: true,
|
||||
queryResult: "schema_ready",
|
||||
endpointRedacted: true,
|
||||
valueRedacted: true
|
||||
},
|
||||
schema,
|
||||
safety: runtimeSafety(),
|
||||
adapterContract: postgresAdapterContract(),
|
||||
counts: counts ?? this.memory.summary().counts
|
||||
};
|
||||
return this.summary();
|
||||
}
|
||||
|
||||
async registerGatewaySession(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.registerGatewaySession(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async registerBoxResource(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
const input = asObject(params.resource ?? params.boxResource ?? params, "boxResource");
|
||||
await this.hydrateGatewaySession(input.gatewaySessionId);
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.registerBoxResource(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async reportBoxCapabilities(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
const inputs = normalizeCapabilityInputs(params);
|
||||
for (const input of inputs) {
|
||||
await this.hydrateBoxResource(input.resourceId);
|
||||
}
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.reportBoxCapabilities(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async requestHardwareOperation(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
await this.hydrateOperationRefs(params);
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.requestHardwareOperation(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async invokeHardwareShell(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
await this.hydrateOperationRefs(params);
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.invokeHardwareShell(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async writeAuditEvent(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.writeAuditEvent(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async queryAuditEvents(params = {}) {
|
||||
const result = await this.query(
|
||||
"SELECT event_json FROM audit_events ORDER BY timestamp ASC",
|
||||
[]
|
||||
);
|
||||
const events = result.rows
|
||||
.map((row) => parseJsonColumn(row.event_json, null))
|
||||
.filter(Boolean)
|
||||
.filter((event) => matchesQuery(event, params, [
|
||||
"auditId",
|
||||
"traceId",
|
||||
"projectId",
|
||||
"gatewaySessionId",
|
||||
"operationId",
|
||||
"action",
|
||||
"targetId"
|
||||
]));
|
||||
|
||||
return {
|
||||
events: limitResults(events, params.limit),
|
||||
count: events.length,
|
||||
persistence: this.summary()
|
||||
};
|
||||
}
|
||||
|
||||
async writeEvidenceRecord(params = {}, requestMeta = {}) {
|
||||
await this.assertReadyForWrites();
|
||||
const before = snapshotMemory(this.memory);
|
||||
const result = this.memory.writeEvidenceRecord(params, requestMeta);
|
||||
await this.persistChanges(before);
|
||||
return withPersistence(result, this.summary());
|
||||
}
|
||||
|
||||
async queryEvidenceRecords(params = {}) {
|
||||
const result = await this.query(
|
||||
"SELECT metadata_json FROM evidence_records ORDER BY created_at ASC",
|
||||
[]
|
||||
);
|
||||
const records = result.rows
|
||||
.map((row) => parseJsonColumn(row.metadata_json, null))
|
||||
.filter(Boolean)
|
||||
.filter((record) => matchesQuery(record, params, [
|
||||
"evidenceId",
|
||||
"projectId",
|
||||
"operationId",
|
||||
"agentSessionId",
|
||||
"workerSessionId",
|
||||
"kind",
|
||||
"serviceId"
|
||||
]));
|
||||
|
||||
return {
|
||||
records: limitResults(records, params.limit),
|
||||
count: records.length,
|
||||
persistence: this.summary()
|
||||
};
|
||||
}
|
||||
|
||||
async hydrateOperationRefs(params = {}) {
|
||||
await this.hydrateGatewaySession(params.gatewaySessionId);
|
||||
await this.hydrateBoxResource(params.resourceId);
|
||||
await this.hydrateBoxCapability(params.capabilityId);
|
||||
}
|
||||
|
||||
async hydrateGatewaySession(gatewaySessionId) {
|
||||
if (!gatewaySessionId || this.memory.gatewaySessions.has(gatewaySessionId)) return;
|
||||
const result = await this.query("SELECT gateway_session_json FROM gateway_sessions WHERE id = $1 LIMIT 1", [
|
||||
gatewaySessionId
|
||||
]);
|
||||
const record = parseJsonColumn(result.rows?.[0]?.gateway_session_json, null);
|
||||
if (record) this.memory.gatewaySessions.set(record.gatewaySessionId, record);
|
||||
}
|
||||
|
||||
async hydrateBoxResource(resourceId) {
|
||||
if (!resourceId || this.memory.boxResources.has(resourceId)) return;
|
||||
const result = await this.query("SELECT resource_json FROM box_resources WHERE id = $1 LIMIT 1", [
|
||||
resourceId
|
||||
]);
|
||||
const record = parseJsonColumn(result.rows?.[0]?.resource_json, null);
|
||||
if (record) {
|
||||
await this.hydrateGatewaySession(record.gatewaySessionId);
|
||||
this.memory.boxResources.set(record.resourceId, record);
|
||||
}
|
||||
}
|
||||
|
||||
async hydrateBoxCapability(capabilityId) {
|
||||
if (!capabilityId || this.memory.boxCapabilities.has(capabilityId)) return;
|
||||
const result = await this.query("SELECT capability_json FROM box_capabilities WHERE id = $1 LIMIT 1", [
|
||||
capabilityId
|
||||
]);
|
||||
const record = parseJsonColumn(result.rows?.[0]?.capability_json, null);
|
||||
if (record) {
|
||||
await this.hydrateBoxResource(record.resourceId);
|
||||
this.memory.boxCapabilities.set(record.capabilityId, record);
|
||||
}
|
||||
}
|
||||
|
||||
async persistChanges(before) {
|
||||
for (const record of newRecords(this.memory.gatewaySessions, before.gatewaySessions)) {
|
||||
await this.persistGatewaySession(record);
|
||||
}
|
||||
for (const record of newRecords(this.memory.boxResources, before.boxResources)) {
|
||||
await this.persistBoxResource(record);
|
||||
}
|
||||
for (const record of newRecords(this.memory.boxCapabilities, before.boxCapabilities)) {
|
||||
await this.persistBoxCapability(record);
|
||||
}
|
||||
for (const record of changedRecords(this.memory.hardwareOperations, before.hardwareOperations)) {
|
||||
await this.persistHardwareOperation(record);
|
||||
}
|
||||
for (const record of newRecords(this.memory.auditEvents, before.auditEvents)) {
|
||||
await this.persistAuditEvent(record);
|
||||
}
|
||||
for (const record of newRecords(this.memory.evidenceRecords, before.evidenceRecords)) {
|
||||
await this.persistEvidenceRecord(record);
|
||||
}
|
||||
}
|
||||
|
||||
async assertReadyForWrites() {
|
||||
const readiness = await this.readiness();
|
||||
if (readiness.ready !== true) {
|
||||
throw new HwlabProtocolError("Postgres durable runtime adapter is not ready for writes", {
|
||||
code: ERROR_CODES.internalError,
|
||||
data: {
|
||||
adapter: RUNTIME_STORE_KIND_POSTGRES,
|
||||
blocker: readiness.blocker,
|
||||
status: readiness.status,
|
||||
schemaReady: Boolean(readiness.schema?.ready)
|
||||
},
|
||||
context: {
|
||||
result: "failed"
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async persistGatewaySession(record) {
|
||||
await this.query(
|
||||
"INSERT INTO gateway_sessions (id, project_id, gateway_service_id, status, started_at, ended_at, gateway_session_json) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, gateway_service_id = EXCLUDED.gateway_service_id, status = EXCLUDED.status, started_at = EXCLUDED.started_at, ended_at = EXCLUDED.ended_at, gateway_session_json = EXCLUDED.gateway_session_json",
|
||||
[
|
||||
record.gatewaySessionId,
|
||||
record.projectId,
|
||||
record.serviceId,
|
||||
record.status,
|
||||
record.startedAt,
|
||||
record.stoppedAt ?? null,
|
||||
stableJson(record)
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
async persistBoxResource(record) {
|
||||
await this.query(
|
||||
"INSERT INTO box_resources (id, project_id, gateway_session_id, resource_state, labels_json, resource_json, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, gateway_session_id = EXCLUDED.gateway_session_id, resource_state = EXCLUDED.resource_state, labels_json = EXCLUDED.labels_json, resource_json = EXCLUDED.resource_json, updated_at = EXCLUDED.updated_at",
|
||||
[
|
||||
record.resourceId,
|
||||
record.projectId,
|
||||
record.gatewaySessionId,
|
||||
record.state,
|
||||
stableJson(record.metadata ?? {}),
|
||||
stableJson(record),
|
||||
record.updatedAt
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
async persistBoxCapability(record) {
|
||||
await this.query(
|
||||
"INSERT INTO box_capabilities (id, box_resource_id, capability_type, capability_json, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO UPDATE SET box_resource_id = EXCLUDED.box_resource_id, capability_type = EXCLUDED.capability_type, capability_json = EXCLUDED.capability_json, updated_at = EXCLUDED.updated_at",
|
||||
[
|
||||
record.capabilityId,
|
||||
record.resourceId,
|
||||
record.name,
|
||||
stableJson(record),
|
||||
record.updatedAt
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
async persistHardwareOperation(record) {
|
||||
await this.query(
|
||||
"INSERT INTO hardware_operations (id, project_id, requested_by, operation_type, operation_json, status, requested_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, requested_by = EXCLUDED.requested_by, operation_type = EXCLUDED.operation_type, operation_json = EXCLUDED.operation_json, status = EXCLUDED.status, requested_at = EXCLUDED.requested_at, updated_at = EXCLUDED.updated_at",
|
||||
[
|
||||
record.operationId,
|
||||
record.projectId,
|
||||
record.requestedBy,
|
||||
record.input?.shell ? "hardware.invoke.shell" : "hardware.operation.request",
|
||||
stableJson(record),
|
||||
record.status,
|
||||
record.requestedAt,
|
||||
record.updatedAt
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
async persistAuditEvent(record) {
|
||||
await this.query(
|
||||
"INSERT INTO audit_events (id, request_id, actor, source, operation, target, result, timestamp, event_json) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (id) DO UPDATE SET request_id = EXCLUDED.request_id, actor = EXCLUDED.actor, source = EXCLUDED.source, operation = EXCLUDED.operation, target = EXCLUDED.target, result = EXCLUDED.result, timestamp = EXCLUDED.timestamp, event_json = EXCLUDED.event_json",
|
||||
[
|
||||
record.auditId,
|
||||
record.traceId,
|
||||
stableJson({ type: record.actorType, id: record.actorId }),
|
||||
stableJson({ serviceId: record.serviceId, environment: record.environment }),
|
||||
record.action,
|
||||
stableJson({ type: record.targetType, id: record.targetId }),
|
||||
record.outcome ?? "accepted",
|
||||
record.occurredAt,
|
||||
stableJson(record)
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
async persistEvidenceRecord(record) {
|
||||
await this.query(
|
||||
"INSERT INTO evidence_records (id, project_id, operation_id, evidence_type, uri, metadata_json, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (id) DO UPDATE SET project_id = EXCLUDED.project_id, operation_id = EXCLUDED.operation_id, evidence_type = EXCLUDED.evidence_type, uri = EXCLUDED.uri, metadata_json = EXCLUDED.metadata_json, created_at = EXCLUDED.created_at",
|
||||
[
|
||||
record.evidenceId,
|
||||
record.projectId,
|
||||
record.operationId,
|
||||
record.kind,
|
||||
record.uri,
|
||||
stableJson(record),
|
||||
record.createdAt
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
async readCounts() {
|
||||
const counts = {};
|
||||
for (const table of postgresCountTables) {
|
||||
const result = await this.query(`SELECT COUNT(*)::int AS count FROM ${table}`, []);
|
||||
counts[toCountKey(table)] = Number(result.rows?.[0]?.count ?? 0);
|
||||
}
|
||||
return counts;
|
||||
}
|
||||
|
||||
async query(sql, params = []) {
|
||||
const client = await this.getQueryClient();
|
||||
return client.query(sql, params);
|
||||
}
|
||||
|
||||
async getQueryClient() {
|
||||
if (this.queryClient) {
|
||||
return this.queryClient;
|
||||
}
|
||||
|
||||
if (this.pool) {
|
||||
return this.pool;
|
||||
}
|
||||
|
||||
let pg;
|
||||
try {
|
||||
pg = await (this.pgModuleLoader ? this.pgModuleLoader() : import("pg"));
|
||||
} catch (error) {
|
||||
if (error?.code === "ERR_MODULE_NOT_FOUND") {
|
||||
const driverError = new Error("Postgres runtime adapter requires the pg package");
|
||||
driverError.code = "HWLAB_PG_DRIVER_MISSING";
|
||||
throw driverError;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
const Pool = pg.Pool ?? pg.default?.Pool;
|
||||
if (typeof Pool !== "function") {
|
||||
const driverError = new Error("Postgres runtime adapter could not load pg.Pool");
|
||||
driverError.code = "HWLAB_PG_DRIVER_MISSING";
|
||||
throw driverError;
|
||||
}
|
||||
|
||||
this.pool = new Pool({
|
||||
connectionString: this.dbUrl,
|
||||
ssl: this.sslMode === "require" ? { rejectUnauthorized: false } : false,
|
||||
connectionTimeoutMillis: normalizeRuntimeTimeoutMs(this.env?.HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS)
|
||||
});
|
||||
return this.pool;
|
||||
}
|
||||
|
||||
blockedSummary({ blocker, reason, schema, connection }) {
|
||||
return {
|
||||
adapter: RUNTIME_STORE_KIND_POSTGRES,
|
||||
durable: true,
|
||||
ready: false,
|
||||
status: "blocked",
|
||||
blocker,
|
||||
reason,
|
||||
liveRuntimeEvidence: false,
|
||||
fixtureEvidence: false,
|
||||
connection: {
|
||||
queryAttempted: Boolean(connection?.queryAttempted),
|
||||
queryResult: connection?.queryResult ?? "not_ready",
|
||||
endpointRedacted: true,
|
||||
valueRedacted: true,
|
||||
errorCode: connection?.errorCode ?? null
|
||||
},
|
||||
schema: schema ?? {
|
||||
ready: false,
|
||||
checked: false,
|
||||
missingTables: [],
|
||||
missingColumns: []
|
||||
},
|
||||
safety: runtimeSafety(),
|
||||
adapterContract: postgresAdapterContract()
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeCapabilityInputs(params) {
|
||||
const value = params.capabilities ?? params.capability ?? params;
|
||||
if (Array.isArray(value)) {
|
||||
@@ -621,3 +1173,176 @@ function limitResults(records, limit) {
|
||||
}
|
||||
return records.slice(0, parsed);
|
||||
}
|
||||
|
||||
function normalizeRuntimeAdapter(value) {
|
||||
if (typeof value !== "string" || value.trim() === "") {
|
||||
return RUNTIME_STORE_KIND;
|
||||
}
|
||||
const normalized = value.trim().toLowerCase();
|
||||
if (["postgres", "postgresql", "pg", "db"].includes(normalized)) {
|
||||
return RUNTIME_STORE_KIND_POSTGRES;
|
||||
}
|
||||
return RUNTIME_STORE_KIND;
|
||||
}
|
||||
|
||||
function postgresAdapterContract() {
|
||||
return {
|
||||
required: "Postgres-backed durable runtime adapter using HWLAB_CLOUD_DB_URL",
|
||||
adapterEnv: RUNTIME_ADAPTER_ENV,
|
||||
adapterEnvValue: RUNTIME_STORE_KIND_POSTGRES,
|
||||
sourceIssue: "pikasTech/HWLAB#164",
|
||||
secretMaterialRequiredInSource: false,
|
||||
fixtureEvidenceAllowed: false
|
||||
};
|
||||
}
|
||||
|
||||
function runtimeSafety() {
|
||||
return {
|
||||
devOnly: true,
|
||||
secretsRead: false,
|
||||
secretMaterialRead: false,
|
||||
valuesRedacted: true,
|
||||
endpointRedacted: true
|
||||
};
|
||||
}
|
||||
|
||||
function classifyRuntimeDbError(error) {
|
||||
const code = typeof error?.code === "string" ? error.code : "UNKNOWN";
|
||||
if (code === "HWLAB_PG_DRIVER_MISSING") {
|
||||
return {
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING,
|
||||
reason: "Postgres runtime adapter is selected, but the pg package is not installed in the runtime image",
|
||||
connection: {
|
||||
queryAttempted: false,
|
||||
queryResult: "driver_missing",
|
||||
errorCode: code
|
||||
}
|
||||
};
|
||||
}
|
||||
if (["28P01", "28000", "42501"].includes(code)) {
|
||||
return {
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED,
|
||||
reason: "Postgres runtime adapter reached the DB driver but authentication or authorization blocked the schema query",
|
||||
connection: {
|
||||
queryAttempted: true,
|
||||
queryResult: "auth_blocked",
|
||||
errorCode: code
|
||||
}
|
||||
};
|
||||
}
|
||||
if (["42P01", "42703", "3F000"].includes(code)) {
|
||||
return {
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
|
||||
reason: "Postgres runtime adapter reached the database, but the runtime schema is missing or incompatible",
|
||||
connection: {
|
||||
queryAttempted: true,
|
||||
queryResult: "schema_blocked",
|
||||
errorCode: code
|
||||
}
|
||||
};
|
||||
}
|
||||
return {
|
||||
blocker: RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED,
|
||||
reason: "Postgres runtime adapter could not complete a live runtime schema query",
|
||||
connection: {
|
||||
queryAttempted: true,
|
||||
queryResult: "query_blocked",
|
||||
errorCode: code
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function summarizeRuntimeSchema(rows) {
|
||||
const columnsByTable = new Map();
|
||||
for (const row of rows) {
|
||||
const table = row.table_name;
|
||||
const column = row.column_name;
|
||||
if (!table || !column) continue;
|
||||
if (!columnsByTable.has(table)) {
|
||||
columnsByTable.set(table, new Set());
|
||||
}
|
||||
columnsByTable.get(table).add(column);
|
||||
}
|
||||
|
||||
const missingTables = [];
|
||||
const missingColumns = [];
|
||||
for (const [table, columns] of Object.entries(requiredPostgresSchema)) {
|
||||
const observed = columnsByTable.get(table);
|
||||
if (!observed) {
|
||||
missingTables.push(table);
|
||||
missingColumns.push(...columns.map((column) => `${table}.${column}`));
|
||||
continue;
|
||||
}
|
||||
for (const column of columns) {
|
||||
if (!observed.has(column)) {
|
||||
missingColumns.push(`${table}.${column}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
ready: missingTables.length === 0 && missingColumns.length === 0,
|
||||
checked: true,
|
||||
requiredTables: Object.keys(requiredPostgresSchema),
|
||||
missingTables,
|
||||
missingColumns
|
||||
};
|
||||
}
|
||||
|
||||
function snapshotMemory(memory) {
|
||||
return {
|
||||
gatewaySessions: new Map(memory.gatewaySessions),
|
||||
boxResources: new Map(memory.boxResources),
|
||||
boxCapabilities: new Map(memory.boxCapabilities),
|
||||
hardwareOperations: new Map(memory.hardwareOperations),
|
||||
auditEvents: new Map(memory.auditEvents),
|
||||
evidenceRecords: new Map(memory.evidenceRecords)
|
||||
};
|
||||
}
|
||||
|
||||
function newRecords(current, before) {
|
||||
return [...current.entries()]
|
||||
.filter(([id]) => !before.has(id))
|
||||
.map(([, record]) => record);
|
||||
}
|
||||
|
||||
function changedRecords(current, before) {
|
||||
return [...current.entries()]
|
||||
.filter(([id, record]) => stableJson(before.get(id)) !== stableJson(record))
|
||||
.map(([, record]) => record);
|
||||
}
|
||||
|
||||
function withPersistence(result, persistence) {
|
||||
return {
|
||||
...result,
|
||||
persistence
|
||||
};
|
||||
}
|
||||
|
||||
function parseJsonColumn(value, fallback) {
|
||||
if (!value) return fallback;
|
||||
if (typeof value === "object") return value;
|
||||
try {
|
||||
return JSON.parse(String(value));
|
||||
} catch {
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
|
||||
function toCountKey(table) {
|
||||
if (table === "gateway_sessions") return "gatewaySessions";
|
||||
if (table === "box_resources") return "boxResources";
|
||||
if (table === "box_capabilities") return "boxCapabilities";
|
||||
if (table === "hardware_operations") return "hardwareOperations";
|
||||
if (table === "audit_events") return "auditEvents";
|
||||
if (table === "evidence_records") return "evidenceRecords";
|
||||
return table;
|
||||
}
|
||||
|
||||
function normalizeRuntimeTimeoutMs(value) {
|
||||
const parsed = Number.parseInt(String(value ?? ""), 10);
|
||||
if (!Number.isInteger(parsed) || parsed <= 0) {
|
||||
return 1200;
|
||||
}
|
||||
return Math.min(parsed, 5000);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import { handleJsonRpcRequest } from "../cloud/json-rpc.mjs";
|
||||
import { validateResponse } from "../protocol/index.mjs";
|
||||
import {
|
||||
RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED,
|
||||
RUNTIME_STORE_KIND_POSTGRES,
|
||||
createCloudRuntimeStore,
|
||||
createConfiguredCloudRuntimeStore
|
||||
} from "./runtime-store.mjs";
|
||||
import { CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS } from "./schema.mjs";
|
||||
|
||||
test("memory runtime store is never marked durable", () => {
|
||||
const store = createCloudRuntimeStore();
|
||||
const summary = store.summary();
|
||||
|
||||
assert.equal(summary.adapter, "memory");
|
||||
assert.equal(summary.durable, false);
|
||||
assert.equal(summary.status, "degraded");
|
||||
assert.equal(summary.ready, undefined);
|
||||
assert.equal(summary.adapterContract.secretMaterialRequiredInSource, false);
|
||||
});
|
||||
|
||||
test("configured postgres runtime reports schema blocker without green readiness", async () => {
|
||||
const store = createConfiguredCloudRuntimeStore({
|
||||
env: {
|
||||
HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
|
||||
HWLAB_CLOUD_RUNTIME_DURABLE: "true"
|
||||
},
|
||||
dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
|
||||
queryClient: {
|
||||
async query(sql) {
|
||||
assert.match(sql, /information_schema\.columns/u);
|
||||
return { rows: [] };
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
const readiness = await store.readiness();
|
||||
assert.equal(readiness.adapter, RUNTIME_STORE_KIND_POSTGRES);
|
||||
assert.equal(readiness.durable, true);
|
||||
assert.equal(readiness.ready, false);
|
||||
assert.equal(readiness.status, "blocked");
|
||||
assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED);
|
||||
assert.equal(readiness.liveRuntimeEvidence, false);
|
||||
assert.equal(readiness.fixtureEvidence, false);
|
||||
assert.equal(readiness.schema.checked, true);
|
||||
assert.ok(readiness.schema.missingTables.includes("gateway_sessions"));
|
||||
assert.equal(JSON.stringify(readiness).includes("redacted@db.example"), false);
|
||||
});
|
||||
|
||||
test("configured postgres runtime persists and queries records through query client", async () => {
|
||||
const queryClient = createFakePostgresClient();
|
||||
const store = createConfiguredCloudRuntimeStore({
|
||||
env: {
|
||||
HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
|
||||
HWLAB_CLOUD_RUNTIME_DURABLE: "true"
|
||||
},
|
||||
dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
|
||||
queryClient,
|
||||
now: () => "2026-05-22T00:00:00.000Z"
|
||||
});
|
||||
const context = { runtimeStore: store, dbProbe: { probe: false } };
|
||||
const meta = {
|
||||
traceId: "trc_01J00000000000000000001000",
|
||||
actorId: "usr_01J00000000000000000001000",
|
||||
serviceId: "hwlab-cloud-web",
|
||||
environment: "dev"
|
||||
};
|
||||
|
||||
async function rpc(id, method, params = {}) {
|
||||
const response = await handleJsonRpcRequest(
|
||||
{
|
||||
jsonrpc: "2.0",
|
||||
id,
|
||||
method,
|
||||
params,
|
||||
meta
|
||||
},
|
||||
context
|
||||
);
|
||||
validateResponse(response);
|
||||
assert.equal(Object.hasOwn(response, "error"), false, response.error?.message);
|
||||
return response.result;
|
||||
}
|
||||
|
||||
const health = await rpc("req_01J00000000000000000001000", "system.health");
|
||||
assert.equal(health.runtime.adapter, "postgres");
|
||||
assert.equal(health.runtime.durable, true);
|
||||
assert.equal(health.runtime.ready, true);
|
||||
assert.equal(health.runtime.liveRuntimeEvidence, true);
|
||||
assert.equal(health.runtime.fixtureEvidence, false);
|
||||
assert.equal(health.readiness.components.runtime, "ready");
|
||||
|
||||
await rpc("req_01J00000000000000000001001", "gateway.session.register", {
|
||||
projectId: "prj_01J00000000000000000001000",
|
||||
gatewaySessionId: "gws_01J00000000000000000001000",
|
||||
gatewayId: "gtw_01J00000000000000000001000",
|
||||
serviceId: "hwlab-gateway-simu",
|
||||
endpoint: "http://127.0.0.1:7101"
|
||||
});
|
||||
await rpc("req_01J00000000000000000001002", "box.resource.register", {
|
||||
projectId: "prj_01J00000000000000000001000",
|
||||
gatewaySessionId: "gws_01J00000000000000000001000",
|
||||
resourceId: "res_01J00000000000000000001000",
|
||||
boxId: "box_01J00000000000000000001000",
|
||||
resourceType: "board",
|
||||
state: "available"
|
||||
});
|
||||
await rpc("req_01J00000000000000000001003", "box.capability.report", {
|
||||
capabilityId: "cap_01J00000000000000000001000",
|
||||
resourceId: "res_01J00000000000000000001000",
|
||||
projectId: "prj_01J00000000000000000001000",
|
||||
name: "shell.exec",
|
||||
direction: "bidirectional",
|
||||
valueType: "object",
|
||||
mutatesState: true
|
||||
});
|
||||
const invoke = await rpc("req_01J00000000000000000001004", "hardware.invoke.shell", {
|
||||
projectId: "prj_01J00000000000000000001000",
|
||||
gatewaySessionId: "gws_01J00000000000000000001000",
|
||||
resourceId: "res_01J00000000000000000001000",
|
||||
capabilityId: "cap_01J00000000000000000001000",
|
||||
input: {
|
||||
command: "echo durable"
|
||||
}
|
||||
});
|
||||
|
||||
const audit = await rpc("req_01J00000000000000000001005", "audit.event.query", {
|
||||
projectId: "prj_01J00000000000000000001000"
|
||||
});
|
||||
const evidence = await rpc("req_01J00000000000000000001006", "evidence.record.query", {
|
||||
projectId: "prj_01J00000000000000000001000"
|
||||
});
|
||||
|
||||
assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO gateway_sessions")));
|
||||
assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO evidence_records")));
|
||||
assert.ok(audit.events.some((event) => event.action === "hardware.invoke.shell"));
|
||||
assert.equal(evidence.count, 1);
|
||||
assert.equal(evidence.records[0].operationId, invoke.operationId);
|
||||
});
|
||||
|
||||
function createFakePostgresClient() {
|
||||
const state = {
|
||||
gateway_sessions: new Map(),
|
||||
box_resources: new Map(),
|
||||
box_capabilities: new Map(),
|
||||
hardware_operations: new Map(),
|
||||
audit_events: new Map(),
|
||||
evidence_records: new Map()
|
||||
};
|
||||
const calls = [];
|
||||
|
||||
return {
|
||||
calls,
|
||||
async query(sql, params = []) {
|
||||
calls.push({ sql, params });
|
||||
if (sql.includes("information_schema.columns")) {
|
||||
return { rows: schemaRows() };
|
||||
}
|
||||
if (sql.startsWith("SELECT COUNT(*)::int AS count FROM ")) {
|
||||
const table = sql.match(/FROM ([a-z_]+)/u)?.[1];
|
||||
return { rows: [{ count: state[table]?.size ?? 0 }] };
|
||||
}
|
||||
if (sql.startsWith("SELECT gateway_session_json FROM gateway_sessions")) {
|
||||
return jsonSelect(state.gateway_sessions, params[0], "gateway_session_json");
|
||||
}
|
||||
if (sql.startsWith("SELECT resource_json FROM box_resources")) {
|
||||
return jsonSelect(state.box_resources, params[0], "resource_json");
|
||||
}
|
||||
if (sql.startsWith("SELECT capability_json FROM box_capabilities")) {
|
||||
return jsonSelect(state.box_capabilities, params[0], "capability_json");
|
||||
}
|
||||
if (sql.startsWith("SELECT event_json FROM audit_events")) {
|
||||
return { rows: [...state.audit_events.values()].map((record) => ({ event_json: record.event_json })) };
|
||||
}
|
||||
if (sql.startsWith("SELECT metadata_json FROM evidence_records")) {
|
||||
return { rows: [...state.evidence_records.values()].map((record) => ({ metadata_json: record.metadata_json })) };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO gateway_sessions")) {
|
||||
state.gateway_sessions.set(params[0], { gateway_session_json: params[6] });
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO box_resources")) {
|
||||
state.box_resources.set(params[0], { resource_json: params[5] });
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO box_capabilities")) {
|
||||
state.box_capabilities.set(params[0], { capability_json: params[3] });
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO hardware_operations")) {
|
||||
state.hardware_operations.set(params[0], { operation_json: params[4] });
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO audit_events")) {
|
||||
state.audit_events.set(params[0], { event_json: params[8] });
|
||||
return { rows: [] };
|
||||
}
|
||||
if (sql.startsWith("INSERT INTO evidence_records")) {
|
||||
state.evidence_records.set(params[0], { metadata_json: params[5] });
|
||||
return { rows: [] };
|
||||
}
|
||||
throw new Error(`unexpected sql: ${sql}`);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function schemaRows() {
|
||||
return Object.entries(CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS).flatMap(([table, columns]) =>
|
||||
columns.map((column) => ({
|
||||
table_name: table,
|
||||
column_name: column
|
||||
}))
|
||||
);
|
||||
}
|
||||
|
||||
function jsonSelect(map, id, column) {
|
||||
const row = map.get(id);
|
||||
return {
|
||||
rows: row ? [{ [column]: row[column] }] : []
|
||||
};
|
||||
}
|
||||
+67
-1
@@ -1,15 +1,75 @@
|
||||
import { TABLES } from "../protocol/index.mjs";
|
||||
|
||||
export const CLOUD_CORE_MIGRATION_ID = "0001_cloud_core_skeleton";
|
||||
export const CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION = "runtime-durable-postgres-v1";
|
||||
|
||||
export const FROZEN_CLOUD_CORE_TABLES = Object.freeze([...TABLES]);
|
||||
|
||||
export const CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS = Object.freeze({
|
||||
gateway_sessions: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"gateway_service_id",
|
||||
"status",
|
||||
"started_at",
|
||||
"ended_at",
|
||||
"gateway_session_json"
|
||||
]),
|
||||
box_resources: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"gateway_session_id",
|
||||
"resource_state",
|
||||
"labels_json",
|
||||
"resource_json",
|
||||
"updated_at"
|
||||
]),
|
||||
box_capabilities: Object.freeze([
|
||||
"id",
|
||||
"box_resource_id",
|
||||
"capability_type",
|
||||
"capability_json",
|
||||
"updated_at"
|
||||
]),
|
||||
hardware_operations: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"requested_by",
|
||||
"operation_type",
|
||||
"operation_json",
|
||||
"status",
|
||||
"requested_at",
|
||||
"updated_at"
|
||||
]),
|
||||
audit_events: Object.freeze([
|
||||
"id",
|
||||
"request_id",
|
||||
"actor",
|
||||
"source",
|
||||
"operation",
|
||||
"target",
|
||||
"result",
|
||||
"timestamp",
|
||||
"event_json"
|
||||
]),
|
||||
evidence_records: Object.freeze([
|
||||
"id",
|
||||
"project_id",
|
||||
"operation_id",
|
||||
"evidence_type",
|
||||
"uri",
|
||||
"metadata_json",
|
||||
"created_at"
|
||||
])
|
||||
});
|
||||
|
||||
export const CLOUD_CORE_MIGRATIONS = Object.freeze([
|
||||
{
|
||||
id: CLOUD_CORE_MIGRATION_ID,
|
||||
path: "internal/db/migrations/0001_cloud_core_skeleton.sql",
|
||||
tables: FROZEN_CLOUD_CORE_TABLES,
|
||||
connectsToDatabase: false
|
||||
connectsToDatabase: false,
|
||||
runtimeDurableAdapterSchemaVersion: CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION
|
||||
}
|
||||
]);
|
||||
|
||||
@@ -19,3 +79,9 @@ export function assertFrozenCloudCoreTables(tables = FROZEN_CLOUD_CORE_TABLES) {
|
||||
throw new Error(`cloud core schema is missing frozen tables: ${missing.join(", ")}`);
|
||||
}
|
||||
}
|
||||
|
||||
export function requiredRuntimeDurableColumns() {
|
||||
return Object.entries(CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS).flatMap(([table, columns]) =>
|
||||
columns.map((column) => `${table}.${column}`)
|
||||
);
|
||||
}
|
||||
|
||||
@@ -7,8 +7,10 @@ import { fileURLToPath } from "node:url";
|
||||
import { TABLES, assertProtocolRecord, assertProtocolRecords } from "../protocol/index.mjs";
|
||||
import {
|
||||
CLOUD_CORE_MIGRATIONS,
|
||||
CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS,
|
||||
FROZEN_CLOUD_CORE_TABLES,
|
||||
assertFrozenCloudCoreTables
|
||||
assertFrozenCloudCoreTables,
|
||||
requiredRuntimeDurableColumns
|
||||
} from "./schema.mjs";
|
||||
|
||||
const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../..");
|
||||
@@ -35,6 +37,19 @@ test("initial migration skeleton declares every frozen table", async () => {
|
||||
assert.match(sql, /\btimestamp\b/);
|
||||
});
|
||||
|
||||
test("initial migration exposes columns required by the durable runtime adapter", async () => {
|
||||
const sql = await readFile(path.join(repoRoot, "internal/db/migrations/0001_cloud_core_skeleton.sql"), "utf8");
|
||||
|
||||
assert.ok(requiredRuntimeDurableColumns().length > 0);
|
||||
for (const [table, columns] of Object.entries(CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS)) {
|
||||
const tableMatch = sql.match(new RegExp(`CREATE TABLE IF NOT EXISTS ${table}\\s*\\(([\\s\\S]*?)\\);`, "u"));
|
||||
assert.ok(tableMatch, `missing ${table}`);
|
||||
for (const column of columns) {
|
||||
assert.match(tableMatch[1], new RegExp(`\\b${column}\\b`), `missing ${table}.${column}`);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
test("protocol record guards catch schema drift before runtime writes", () => {
|
||||
assertProtocolRecord("gatewaySession", {
|
||||
gatewaySessionId: "gws_01J00000000000000000000000",
|
||||
|
||||
Reference in New Issue
Block a user