482 lines
19 KiB
TypeScript
482 lines
19 KiB
TypeScript
import http from "node:http";
|
|
import { AsyncLocalStorage } from "node:async_hooks";
|
|
import { randomUUID } from "node:crypto";
|
|
import { create, equals } from "@bufbuild/protobuf";
|
|
import { Code, ConnectError, createClient, type Client, type ConnectRouter } from "@connectrpc/connect";
|
|
import { connectNodeAdapter, createConnectTransport } from "@connectrpc/connect-node";
|
|
import {
|
|
CaminoService,
|
|
CrdtValueSchema,
|
|
ListValueSchema,
|
|
NullValueSchema,
|
|
ObjectValueSchema,
|
|
RefValueSchema,
|
|
ValueSchema,
|
|
type Value,
|
|
} from "./camino/api_pb.js";
|
|
import { OrchestratorRuntime } from "./quixos/orch_pb.js";
|
|
import {
|
|
DerivedDependencySchema,
|
|
HandshakeResponseSchema,
|
|
InvokeResponseSchema,
|
|
PackageRuntime,
|
|
WatchEventSchema,
|
|
} from "./quixos/runtime_pb.js";
|
|
import { CapabilityRefSchema } from "./quixos/refs_pb.js";
|
|
|
|
export type CaminoClient = Client<typeof CaminoService>;
|
|
export type OrchClient = Client<typeof OrchestratorRuntime>;
|
|
|
|
export type RuntimeDependency =
|
|
| { kind: "state"; objectId: string; attachmentId: string }
|
|
| { kind: "edge"; objectId: string; attachmentId: string; projectionId: string };
|
|
|
|
type DependencyScope = {
|
|
dependencies: Map<string, RuntimeDependency>;
|
|
observe?: (dependency: RuntimeDependency) => Promise<void>;
|
|
};
|
|
|
|
const dependencyKey = (dependency: RuntimeDependency) =>
|
|
`${dependency.kind}:${dependency.objectId}:${dependency.attachmentId}:${
|
|
dependency.kind === "edge" ? dependency.projectionId : ""
|
|
}`;
|
|
|
|
const dependencyScope = new AsyncLocalStorage<DependencyScope>();
|
|
const recordDependency = async (dependency: RuntimeDependency) => {
|
|
const scope = dependencyScope.getStore();
|
|
if (!scope) return;
|
|
const key = `${dependency.kind}:${dependency.objectId}:${dependency.attachmentId}:${
|
|
dependency.kind === "edge" ? dependency.projectionId : ""
|
|
}`;
|
|
scope.dependencies.set(key, dependency);
|
|
await scope.observe?.(dependency);
|
|
};
|
|
|
|
const bytesToBase64 = (value: Uint8Array) => Buffer.from(value).toString("base64");
|
|
const base64ToBytes = (value: string) => Buffer.from(value, "base64");
|
|
const isRecord = (value: unknown): value is Record<string, unknown> =>
|
|
Boolean(value) && typeof value === "object" && !Array.isArray(value);
|
|
const isWrappedValue = (value: unknown): value is { $quixosValue: Value } =>
|
|
isRecord(value) && "$quixosValue" in value &&
|
|
isRecord(value.$quixosValue) && value.$quixosValue.$typeName === "camino.Value";
|
|
|
|
export const objectRef = (objectId: string) => ({ $quixosRef: objectId });
|
|
export const liveValue = (value: Value) => ({ $quixosValue: value });
|
|
|
|
export const jsToProtoValue = (value: unknown): Value => {
|
|
if (isWrappedValue(value)) return value.$quixosValue;
|
|
if (value === null || value === undefined) {
|
|
return create(ValueSchema, { kind: { case: "nullValue", value: create(NullValueSchema, {}) } });
|
|
}
|
|
if (typeof value === "boolean") return create(ValueSchema, { kind: { case: "boolValue", value } });
|
|
if (typeof value === "number") return create(ValueSchema, { kind: { case: "numberValue", value } });
|
|
if (typeof value === "bigint") return create(ValueSchema, { kind: { case: "integerValue", value: String(value) } });
|
|
if (typeof value === "string") return create(ValueSchema, { kind: { case: "stringValue", value } });
|
|
if (value instanceof Uint8Array) return create(ValueSchema, { kind: { case: "bytesValue", value } });
|
|
if (Array.isArray(value)) {
|
|
return create(ValueSchema, {
|
|
kind: { case: "listValue", value: create(ListValueSchema, { values: value.map(jsToProtoValue) }) },
|
|
});
|
|
}
|
|
if (isRecord(value) && typeof value.$quixosRef === "string") {
|
|
return create(ValueSchema, {
|
|
kind: { case: "refValue", value: create(RefValueSchema, { objectId: value.$quixosRef }) },
|
|
});
|
|
}
|
|
if (isRecord(value) && typeof value.$quixosCrdtType === "string" &&
|
|
typeof value.$quixosCrdtPayload === "string") {
|
|
return create(ValueSchema, {
|
|
kind: { case: "crdtValue", value: create(CrdtValueSchema, {
|
|
type: value.$quixosCrdtType,
|
|
encoding: typeof value.$quixosCrdtEncoding === "string" ? value.$quixosCrdtEncoding : "base64",
|
|
payload: base64ToBytes(value.$quixosCrdtPayload),
|
|
}) },
|
|
});
|
|
}
|
|
if (!isRecord(value)) throw new Error(`Unsupported runtime value ${typeof value}`);
|
|
return create(ValueSchema, {
|
|
kind: { case: "objectValue", value: create(ObjectValueSchema, {
|
|
fields: Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, jsToProtoValue(entry)])),
|
|
}) },
|
|
});
|
|
};
|
|
|
|
export const protoValueToJs = (value: Value | undefined): unknown => {
|
|
switch (value?.kind.case) {
|
|
case "nullValue":
|
|
case undefined: return null;
|
|
case "boolValue":
|
|
case "numberValue":
|
|
case "stringValue":
|
|
case "integerValue": return value.kind.value;
|
|
case "bytesValue": return bytesToBase64(value.kind.value);
|
|
case "refValue": return value.kind.value.objectId;
|
|
case "listValue": return value.kind.value.values.map(protoValueToJs);
|
|
case "objectValue": return Object.fromEntries(
|
|
Object.entries(value.kind.value.fields).map(([key, entry]) => [key, protoValueToJs(entry)]),
|
|
);
|
|
case "crdtValue": return {
|
|
$quixosCrdtType: value.kind.value.type,
|
|
$quixosCrdtEncoding: value.kind.value.encoding,
|
|
$quixosCrdtPayload: bytesToBase64(value.kind.value.payload),
|
|
};
|
|
}
|
|
};
|
|
|
|
export const protoFieldsToJs = (fields: Record<string, Value>) =>
|
|
Object.fromEntries(Object.entries(fields).map(([key, value]) => [key, protoValueToJs(value)]));
|
|
|
|
export type StatePort<T = unknown> = {
|
|
slotId: string;
|
|
get(): Promise<T>;
|
|
live(): Promise<ReturnType<typeof liveValue>>;
|
|
set(value: T): Promise<void>;
|
|
};
|
|
export type EdgePort = {
|
|
edgeTypeId: string;
|
|
projectionId: string;
|
|
resolve(): Promise<string[]>;
|
|
connect(targetObjectId: string): Promise<void>;
|
|
};
|
|
export type InterfacePort = {
|
|
interfaceRevisionId: string;
|
|
invoke(operationId: string, input?: Record<string, unknown>): Promise<unknown>;
|
|
};
|
|
export type ConstructorPort = {
|
|
atomId: string;
|
|
construct(input?: Record<string, unknown>): Promise<string>;
|
|
};
|
|
export type RuntimePort = StatePort | EdgePort | InterfacePort | ConstructorPort;
|
|
|
|
export type RuntimeContext = {
|
|
objectId: string;
|
|
input: Record<string, unknown>;
|
|
inputProto: Record<string, Value>;
|
|
ports: ReadonlyMap<string, RuntimePort>;
|
|
state<T = unknown>(portId: string): StatePort<T>;
|
|
edge(portId: string): EdgePort;
|
|
interface(portId: string): InterfacePort;
|
|
constructor(portId: string): ConstructorPort;
|
|
};
|
|
|
|
const targetForEdge = (
|
|
edge: { firstObjectId: string; secondObjectId: string; firstProjectionId: string },
|
|
projectionId: string,
|
|
) => edge.firstProjectionId === projectionId ? edge.secondObjectId : edge.firstObjectId;
|
|
|
|
export const createRuntimeContext = (
|
|
camino: CaminoClient,
|
|
orch: OrchClient,
|
|
request: RuntimeRequest,
|
|
): RuntimeContext => {
|
|
const ports = new Map<string, RuntimePort>();
|
|
for (const dependency of request.dependencies) {
|
|
switch (dependency.binding.case) {
|
|
case "stateSlotId": {
|
|
const slotId = dependency.binding.value;
|
|
const dependencyObjectId = dependency.objectId || request.objectId;
|
|
const state: StatePort = {
|
|
slotId,
|
|
async get() {
|
|
await recordDependency({ kind: "state", objectId: dependencyObjectId, attachmentId: slotId });
|
|
return protoValueToJs((await camino.readState({ objectId: dependencyObjectId, slotId })).value);
|
|
},
|
|
async live() {
|
|
await recordDependency({ kind: "state", objectId: dependencyObjectId, attachmentId: slotId });
|
|
const value = await camino.readState({ objectId: dependencyObjectId, slotId });
|
|
if (!value.value) throw new Error(`State ${slotId} returned no value`);
|
|
return liveValue(value.value);
|
|
},
|
|
async set(value) {
|
|
await camino.writeState({ objectId: dependencyObjectId, slotId, value: jsToProtoValue(value) });
|
|
},
|
|
};
|
|
ports.set(dependency.portId, state);
|
|
break;
|
|
}
|
|
case "edge": {
|
|
const { edgeTypeId, projectionId } = dependency.binding.value;
|
|
const dependencyObjectId = dependency.objectId || request.objectId;
|
|
const edge: EdgePort = {
|
|
edgeTypeId,
|
|
projectionId,
|
|
async resolve() {
|
|
await recordDependency({ kind: "edge", objectId: dependencyObjectId, attachmentId: edgeTypeId, projectionId });
|
|
const result = await camino.resolveEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId });
|
|
return result.edges.map((entry) => targetForEdge(entry, projectionId));
|
|
},
|
|
async connect(targetObjectId) {
|
|
await camino.connectEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId, targetObjectId });
|
|
},
|
|
};
|
|
ports.set(dependency.portId, edge);
|
|
break;
|
|
}
|
|
case "receiverInterfaceRevisionId": {
|
|
const interfaceRevisionId = dependency.binding.value;
|
|
const capability: InterfacePort = {
|
|
interfaceRevisionId,
|
|
async invoke(operationId, input = {}) {
|
|
const response = await orch.invokeCapability({
|
|
capability: create(CapabilityRefSchema, { interfaceRevisionId, operationId }),
|
|
objectId: request.objectId,
|
|
input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
|
|
});
|
|
if (!response.ok) throw new Error(response.error || `Capability ${operationId} failed`);
|
|
return protoValueToJs(response.result);
|
|
},
|
|
};
|
|
ports.set(dependency.portId, capability);
|
|
break;
|
|
}
|
|
case "constructorAtomId": {
|
|
const atomId = dependency.binding.value;
|
|
const constructor: ConstructorPort = {
|
|
atomId,
|
|
async construct(input = {}) {
|
|
const response = await orch.constructObject({
|
|
atomId,
|
|
input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
|
|
});
|
|
if (!response.object) throw new Error(`Constructor ${atomId} returned no object`);
|
|
return response.object.id;
|
|
},
|
|
};
|
|
ports.set(dependency.portId, constructor);
|
|
}
|
|
}
|
|
}
|
|
const requirePort = <T extends RuntimePort>(portId: string, kind: string) => {
|
|
const port = ports.get(portId);
|
|
if (!port || !(kind in port)) throw new Error(`Missing ${kind} dependency port ${portId}`);
|
|
return port as T;
|
|
};
|
|
return {
|
|
objectId: request.objectId,
|
|
input: protoFieldsToJs(request.input),
|
|
inputProto: request.input,
|
|
ports,
|
|
state: <T>(portId: string) => requirePort<StatePort<T>>(portId, "slotId"),
|
|
edge: (portId: string) => requirePort<EdgePort>(portId, "edgeTypeId"),
|
|
interface: (portId: string) => requirePort<InterfacePort>(portId, "interfaceRevisionId"),
|
|
constructor: (portId: string) => requirePort<ConstructorPort>(portId, "atomId"),
|
|
};
|
|
};
|
|
|
|
export type RuntimeHandler = (context: RuntimeContext) => unknown | Promise<unknown>;
|
|
export type DerivedHandler = { kind: "derived"; get: RuntimeHandler };
|
|
export const derived = (get: RuntimeHandler): DerivedHandler => ({ kind: "derived", get });
|
|
const isDerived = (handler: RuntimeHandler | DerivedHandler): handler is DerivedHandler =>
|
|
typeof handler === "object" && handler.kind === "derived";
|
|
|
|
const evaluate = async (
|
|
handler: RuntimeHandler | DerivedHandler,
|
|
context: RuntimeContext,
|
|
observe?: (dependency: RuntimeDependency) => Promise<void>,
|
|
) => {
|
|
const dependencies = new Map<string, RuntimeDependency>();
|
|
const result = await dependencyScope.run(
|
|
{ dependencies, observe },
|
|
() => isDerived(handler) ? handler.get(context) : handler(context),
|
|
);
|
|
return { value: jsToProtoValue(result), dependencies: [...dependencies.values()] };
|
|
};
|
|
|
|
const protoDependencies = (dependencies: RuntimeDependency[]) => dependencies.map((entry) =>
|
|
create(DerivedDependencySchema, {
|
|
kind: entry.kind,
|
|
objectId: entry.objectId,
|
|
attachmentId: entry.attachmentId,
|
|
projectionId: entry.kind === "edge" ? entry.projectionId : "",
|
|
}));
|
|
|
|
export const createPackageRuntimeRoutes = (config: {
|
|
packageRevisionId: string;
|
|
exports: Record<string, RuntimeHandler | DerivedHandler>;
|
|
caminoUrl?: string;
|
|
orchUrl?: string;
|
|
}) => {
|
|
const headers: Record<string, string> = {};
|
|
if (process.env.CAMINO_RUNTIME_AUTH_TOKEN) {
|
|
headers["x-camino-runtime-token"] = process.env.CAMINO_RUNTIME_AUTH_TOKEN;
|
|
} else if (process.env.CAMINO_RUNTIME_AUTH_REQUIRED === "1") {
|
|
throw new Error("CAMINO_RUNTIME_AUTH_TOKEN is required");
|
|
}
|
|
const camino = createClient(CaminoService, createConnectTransport({
|
|
baseUrl: config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310",
|
|
httpVersion: "1.1",
|
|
interceptors: headers["x-camino-runtime-token"] ? [
|
|
(next) => async (request) => {
|
|
request.header.set("x-camino-runtime-token", headers["x-camino-runtime-token"]!);
|
|
return await next(request);
|
|
},
|
|
] : [],
|
|
}));
|
|
const orch = createClient(OrchestratorRuntime, createConnectTransport({
|
|
baseUrl: config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311",
|
|
httpVersion: "1.1",
|
|
}));
|
|
|
|
return (router: ConnectRouter) => router.service(PackageRuntime, {
|
|
handshake: () => create(HandshakeResponseSchema, {
|
|
packageRevisionId: config.packageRevisionId,
|
|
runtimeProtocolVersion: "quixos-capabilities-v1",
|
|
exportIds: Object.keys(config.exports),
|
|
}),
|
|
invoke: async (request) => {
|
|
const exportId = request.export?.exportId;
|
|
const handler = exportId ? config.exports[exportId] : undefined;
|
|
if (!handler) throw new ConnectError(`Unknown export ${exportId ?? ""}`, Code.NotFound);
|
|
try {
|
|
const result = await evaluate(handler, createRuntimeContext(camino, orch, request));
|
|
return create(InvokeResponseSchema, { ok: true, result: result.value });
|
|
} catch (error) {
|
|
return create(InvokeResponseSchema, {
|
|
ok: false,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
});
|
|
}
|
|
},
|
|
watch: async function* (request, context) {
|
|
const exportId = request.export?.exportId;
|
|
const handler = exportId ? config.exports[exportId] : undefined;
|
|
if (!handler || !isDerived(handler)) {
|
|
throw new ConnectError(`Export ${exportId ?? ""} is not derived`, Code.FailedPrecondition);
|
|
}
|
|
const watchId = `watch:${randomUUID()}`;
|
|
const runtimeContext = createRuntimeContext(camino, orch, request);
|
|
type WatchOutcome = { key: string; done: boolean; error?: unknown };
|
|
type Subscription = {
|
|
dependency: RuntimeDependency;
|
|
controller: AbortController;
|
|
next: Promise<WatchOutcome>;
|
|
waitNext: () => Promise<WatchOutcome>;
|
|
};
|
|
const subscriptions = new Map<string, Subscription>();
|
|
const establishing = new Map<string, Promise<void>>();
|
|
|
|
const ensureSubscription = async (dependency: RuntimeDependency) => {
|
|
const key = dependencyKey(dependency);
|
|
if (subscriptions.has(key)) return;
|
|
const pending = establishing.get(key);
|
|
if (pending) return await pending;
|
|
const establish = (async () => {
|
|
const controller = new AbortController();
|
|
const stream = camino.watchObject(
|
|
{ objectId: dependency.objectId, includeSnapshot: true },
|
|
{ signal: controller.signal },
|
|
)[Symbol.asyncIterator]();
|
|
try {
|
|
// Camino subscribes before producing the snapshot, so once this
|
|
// resolves the following state/edge read cannot race the stream.
|
|
const snapshot = await stream.next();
|
|
if (snapshot.done) throw new Error(`Dependency stream ${key} ended during setup`);
|
|
const waitNext = () => stream.next().then(
|
|
(result) => ({ key, done: Boolean(result.done) }),
|
|
(error: unknown) => ({ key, done: true, error }),
|
|
);
|
|
const subscription: Subscription = {
|
|
dependency,
|
|
controller,
|
|
waitNext,
|
|
next: Promise.resolve({ key, done: false }),
|
|
};
|
|
subscription.next = waitNext();
|
|
subscriptions.set(key, subscription);
|
|
} catch (error) {
|
|
controller.abort();
|
|
throw error;
|
|
}
|
|
})();
|
|
establishing.set(key, establish);
|
|
try {
|
|
await establish;
|
|
} finally {
|
|
establishing.delete(key);
|
|
}
|
|
};
|
|
|
|
const abortAll = () => {
|
|
for (const subscription of subscriptions.values()) {
|
|
subscription.controller.abort();
|
|
}
|
|
};
|
|
context.signal.addEventListener("abort", abortAll, { once: true });
|
|
|
|
let current = await evaluate(handler, runtimeContext, ensureSubscription);
|
|
yield create(WatchEventSchema, {
|
|
watchId,
|
|
value: current.value,
|
|
dependencies: protoDependencies(current.dependencies),
|
|
initial: true,
|
|
});
|
|
|
|
const abort = new Promise<"abort">((resolve) => {
|
|
if (context.signal.aborted) resolve("abort");
|
|
else context.signal.addEventListener("abort", () => resolve("abort"), { once: true });
|
|
});
|
|
try {
|
|
while (!context.signal.aborted) {
|
|
if (subscriptions.size === 0) {
|
|
await abort;
|
|
break;
|
|
}
|
|
const outcome = await Promise.race([
|
|
...[...subscriptions.values()].map((entry) => entry.next),
|
|
abort,
|
|
]);
|
|
if (outcome === "abort") break;
|
|
const subscription = subscriptions.get(outcome.key);
|
|
if (!subscription) continue;
|
|
if (outcome.error) throw outcome.error;
|
|
if (outcome.done) throw new Error(`Dependency stream ${outcome.key} ended unexpectedly`);
|
|
subscription.next = subscription.waitNext();
|
|
|
|
const updated = await evaluate(handler, runtimeContext, ensureSubscription);
|
|
const active = new Set(updated.dependencies.map(dependencyKey));
|
|
for (const [key, entry] of subscriptions) {
|
|
if (!active.has(key)) {
|
|
entry.controller.abort();
|
|
subscriptions.delete(key);
|
|
}
|
|
}
|
|
if (!equals(ValueSchema, current.value, updated.value)) {
|
|
yield create(WatchEventSchema, {
|
|
watchId,
|
|
value: updated.value,
|
|
dependencies: protoDependencies(updated.dependencies),
|
|
});
|
|
}
|
|
current = updated;
|
|
}
|
|
} finally {
|
|
context.signal.removeEventListener("abort", abortAll);
|
|
abortAll();
|
|
}
|
|
},
|
|
});
|
|
};
|
|
|
|
export const servePackageRuntime = (config: {
|
|
packageRevisionId: string;
|
|
exports: Record<string, RuntimeHandler | DerivedHandler>;
|
|
}) => {
|
|
const host = process.env.QUIXOS_RUNTIME_HOST ?? "127.0.0.1";
|
|
const port = Number(process.env.QUIXOS_RUNTIME_PORT ?? "0");
|
|
const handler = connectNodeAdapter({ routes: createPackageRuntimeRoutes(config) });
|
|
const server = http.createServer((request, response) => void handler(request, response));
|
|
server.listen(port, host, () => console.log(`${config.packageRevisionId} listening on ${host}:${port}`));
|
|
const shutdown = () => {
|
|
server.close(() => process.exit(0));
|
|
server.closeAllConnections();
|
|
};
|
|
process.on("SIGTERM", shutdown);
|
|
process.on("SIGINT", shutdown);
|
|
return server;
|
|
};
|
|
type RuntimeRequest = {
|
|
objectId: string;
|
|
input: Record<string, Value>;
|
|
dependencies: import("./quixos/refs_pb.js").InjectedDependency[];
|
|
};
|