Publish built Camino package runtime artifacts
This commit is contained in:
Timothy J. Aveni
Vendored
+731
@@ -0,0 +1,731 @@
|
||||
import http from "node:http";
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { create } from "@bufbuild/protobuf";
|
||||
import { createClient, } from "@connectrpc/connect";
|
||||
import * as Automerge from "@automerge/automerge";
|
||||
import { createAutomergeFieldValue, encodeAutomergeFieldWrite, loadAutomergeFieldDoc, } from "@quixos/camino-datatypes/crdt-automerge";
|
||||
import { connectNodeAdapter, createConnectTransport, } from "@connectrpc/connect-node";
|
||||
import { CaminoService, CrdtValueSchema, FieldValueSourceSchema, ListValueSchema, NullValueSchema, ObjectValueSchema, RefValueSchema, ValueSchema, ValueSourceSchema, } from "./camino/api_pb.js";
|
||||
import { DerivedDependencySchema, HandshakeResponseSchema, InvokeResponseSchema, PackageRuntime, WatchEventSchema, WatchRequestSchema, } from "./quixos/runtime_pb.js";
|
||||
import { FunctionRefSchema } from "./quixos/refs_pb.js";
|
||||
const dependencyKey = (dependency) => {
|
||||
switch (dependency.kind) {
|
||||
case "object":
|
||||
return `object:${dependency.objectId}`;
|
||||
case "field":
|
||||
return `field:${dependency.objectId}:${dependency.fieldName}`;
|
||||
case "edges":
|
||||
return `edges:${dependency.objectId}:${dependency.sourceField ?? ""}`;
|
||||
}
|
||||
};
|
||||
const createDependencyTracker = () => {
|
||||
const dependencies = new Map();
|
||||
return {
|
||||
record(dependency) {
|
||||
dependencies.set(dependencyKey(dependency), dependency);
|
||||
},
|
||||
dependencies() {
|
||||
return [...dependencies.values()];
|
||||
},
|
||||
};
|
||||
};
|
||||
const dependencyStorage = new AsyncLocalStorage();
|
||||
export const recordDerivedDependency = (dependency) => {
|
||||
dependencyStorage.getStore()?.record(dependency);
|
||||
};
|
||||
export const getActiveDerivedDependencies = () => dependencyStorage.getStore()?.dependencies() ?? [];
|
||||
export const runWithDerivedDependencyTracking = async (fn) => {
|
||||
const tracker = createDependencyTracker();
|
||||
const value = await dependencyStorage.run(tracker, fn);
|
||||
return {
|
||||
value,
|
||||
dependencies: tracker.dependencies(),
|
||||
};
|
||||
};
|
||||
const logDerivedDependencies = (dependencies) => {
|
||||
if (process.env.QUIXOS_DERIVED_DEPS_LOG !== "1") {
|
||||
return;
|
||||
}
|
||||
process.stderr.write(`[derived-deps] ${JSON.stringify({ dependencies })}\n`);
|
||||
};
|
||||
const logDerivedWatch = (event, detail) => {
|
||||
if (process.env.QUIXOS_DERIVED_WATCH_LOG !== "1") {
|
||||
return;
|
||||
}
|
||||
process.stderr.write(`[derived-watch] ${JSON.stringify({ event, ...detail })}\n`);
|
||||
};
|
||||
const watchSinkStorage = new AsyncLocalStorage();
|
||||
const dependencyObjectIds = (dependencies) => new Set(dependencies.map((dependency) => dependency.objectId));
|
||||
const stopDerivedWatch = (watch) => {
|
||||
if (watch.stopped) {
|
||||
return;
|
||||
}
|
||||
watch.stopped = true;
|
||||
for (const controller of watch.controllers.values()) {
|
||||
controller.abort();
|
||||
}
|
||||
watch.controllers.clear();
|
||||
logDerivedWatch("stopped", { watchId: watch.watchId });
|
||||
};
|
||||
export const liveFieldValue = (params) => ({
|
||||
$caminoValue: params.value,
|
||||
$caminoSource: {
|
||||
field: {
|
||||
objectId: params.objectId,
|
||||
fieldName: params.fieldName,
|
||||
...(params.fieldType ? { fieldType: params.fieldType } : {}),
|
||||
...(params.fieldStorage ? { fieldStorage: params.fieldStorage } : {}),
|
||||
...(params.conflictStrategy
|
||||
? { conflictStrategy: params.conflictStrategy }
|
||||
: {}),
|
||||
...(params.revision !== undefined ? { revision: params.revision } : {}),
|
||||
},
|
||||
},
|
||||
});
|
||||
export const fieldOps = (operation) => operation;
|
||||
export const derivedWithDeps = (operation) => {
|
||||
const watches = new Map();
|
||||
const rerun = async (watch) => {
|
||||
if (watch.stopped) {
|
||||
return;
|
||||
}
|
||||
if (watch.running) {
|
||||
watch.pending = true;
|
||||
return;
|
||||
}
|
||||
watch.running = true;
|
||||
try {
|
||||
do {
|
||||
watch.pending = false;
|
||||
const { value, dependencies } = await runWithDerivedDependencyTracking(() => watch.get(watch.context));
|
||||
watch.lastValue = value;
|
||||
logDerivedDependencies(dependencies);
|
||||
logDerivedWatch("evaluated", {
|
||||
watchId: watch.watchId,
|
||||
value,
|
||||
dependencies,
|
||||
});
|
||||
watch.sink?.publish({
|
||||
watchId: watch.watchId,
|
||||
value,
|
||||
dependencies,
|
||||
initial: !watch.emitted,
|
||||
});
|
||||
watch.emitted = true;
|
||||
reconcileSubscriptions(watch, dependencies);
|
||||
} while (watch.pending && !watch.stopped);
|
||||
}
|
||||
catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
logDerivedWatch("error", {
|
||||
watchId: watch.watchId,
|
||||
error: message,
|
||||
});
|
||||
watch.sink?.publish({
|
||||
watchId: watch.watchId,
|
||||
dependencies: [],
|
||||
error: message,
|
||||
initial: !watch.emitted,
|
||||
});
|
||||
watch.emitted = true;
|
||||
}
|
||||
finally {
|
||||
watch.running = false;
|
||||
}
|
||||
};
|
||||
const scheduleRerun = (watch, reason) => {
|
||||
if (watch.stopped) {
|
||||
return;
|
||||
}
|
||||
logDerivedWatch("scheduled", { watchId: watch.watchId, reason });
|
||||
void rerun(watch);
|
||||
};
|
||||
const startObjectWatch = (watch, objectId) => {
|
||||
const controller = new AbortController();
|
||||
watch.controllers.set(objectId, controller);
|
||||
logDerivedWatch("subscribed", { watchId: watch.watchId, objectId });
|
||||
void (async () => {
|
||||
try {
|
||||
for await (const event of watch.context.camino.watchObject({
|
||||
objectId,
|
||||
includeSnapshot: false,
|
||||
}, { signal: controller.signal })) {
|
||||
if (controller.signal.aborted || watch.stopped) {
|
||||
break;
|
||||
}
|
||||
scheduleRerun(watch, event.op?.id ? `op:${event.op.id}` : `object:${objectId}`);
|
||||
}
|
||||
}
|
||||
catch (error) {
|
||||
if (!controller.signal.aborted && !watch.stopped) {
|
||||
logDerivedWatch("watch-error", {
|
||||
watchId: watch.watchId,
|
||||
objectId,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
finally {
|
||||
if (watch.controllers.get(objectId) === controller) {
|
||||
watch.controllers.delete(objectId);
|
||||
logDerivedWatch("unsubscribed", { watchId: watch.watchId, objectId });
|
||||
}
|
||||
}
|
||||
})();
|
||||
};
|
||||
const reconcileSubscriptions = (watch, dependencies) => {
|
||||
const objectIds = dependencyObjectIds(dependencies);
|
||||
for (const [objectId, controller] of watch.controllers.entries()) {
|
||||
if (!objectIds.has(objectId)) {
|
||||
controller.abort();
|
||||
watch.controllers.delete(objectId);
|
||||
logDerivedWatch("unsubscribed", { watchId: watch.watchId, objectId });
|
||||
}
|
||||
}
|
||||
for (const objectId of objectIds) {
|
||||
if (!watch.controllers.has(objectId)) {
|
||||
startObjectWatch(watch, objectId);
|
||||
}
|
||||
}
|
||||
};
|
||||
return fieldOps({
|
||||
async get(context) {
|
||||
const { value, dependencies } = await runWithDerivedDependencyTracking(() => operation.get(context));
|
||||
logDerivedDependencies(dependencies);
|
||||
return value;
|
||||
},
|
||||
async watchStart(context) {
|
||||
const watchId = `watch:${randomUUID()}`;
|
||||
const watch = {
|
||||
watchId,
|
||||
context: context,
|
||||
get: operation.get,
|
||||
sink: watchSinkStorage.getStore(),
|
||||
controllers: new Map(),
|
||||
running: false,
|
||||
pending: false,
|
||||
stopped: false,
|
||||
emitted: false,
|
||||
lastValue: undefined,
|
||||
};
|
||||
watches.set(watchId, watch);
|
||||
logDerivedWatch("started", { watchId });
|
||||
await rerun(watch);
|
||||
return {
|
||||
watch_id: watchId,
|
||||
};
|
||||
},
|
||||
watchStop(context) {
|
||||
const input = protoFieldsToJs(context.request.input);
|
||||
const watchId = typeof input.watch_id === "string"
|
||||
? input.watch_id
|
||||
: typeof input.watchId === "string"
|
||||
? input.watchId
|
||||
: "";
|
||||
const watch = watches.get(watchId);
|
||||
if (!watch) {
|
||||
throw new Error(`Unknown derived watch: ${watchId}`);
|
||||
}
|
||||
stopDerivedWatch(watch);
|
||||
watches.delete(watchId);
|
||||
return {};
|
||||
},
|
||||
});
|
||||
};
|
||||
class AsyncEventQueue {
|
||||
values = [];
|
||||
waiters = [];
|
||||
closed = false;
|
||||
push(value) {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
const waiter = this.waiters.shift();
|
||||
if (waiter) {
|
||||
waiter({ value, done: false });
|
||||
return;
|
||||
}
|
||||
this.values.push(value);
|
||||
}
|
||||
close() {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
this.closed = true;
|
||||
for (const waiter of this.waiters.splice(0)) {
|
||||
waiter({ value: undefined, done: true });
|
||||
}
|
||||
}
|
||||
async next() {
|
||||
const value = this.values.shift();
|
||||
if (value !== undefined) {
|
||||
return { value, done: false };
|
||||
}
|
||||
if (this.closed) {
|
||||
return { value: undefined, done: true };
|
||||
}
|
||||
return await new Promise((resolve) => {
|
||||
this.waiters.push(resolve);
|
||||
});
|
||||
}
|
||||
[Symbol.asyncIterator]() {
|
||||
return this;
|
||||
}
|
||||
}
|
||||
const isRecord = (value) => Boolean(value) && typeof value === "object" && !Array.isArray(value);
|
||||
const bytesToBase64 = (value) => Buffer.from(value).toString("base64");
|
||||
const base64ToBytes = (value) => Buffer.from(value, "base64");
|
||||
const sourceToProto = (value) => {
|
||||
if (!isRecord(value) || !isRecord(value.field)) {
|
||||
return undefined;
|
||||
}
|
||||
const field = value.field;
|
||||
if (typeof field.objectId !== "string" ||
|
||||
typeof field.fieldName !== "string") {
|
||||
return undefined;
|
||||
}
|
||||
return create(ValueSourceSchema, {
|
||||
field: create(FieldValueSourceSchema, {
|
||||
objectId: field.objectId,
|
||||
fieldName: field.fieldName,
|
||||
fieldType: typeof field.fieldType === "string" ? field.fieldType : "",
|
||||
fieldStorage: typeof field.fieldStorage === "string" ? field.fieldStorage : "",
|
||||
conflictStrategy: typeof field.conflictStrategy === "string" ? field.conflictStrategy : "",
|
||||
revision: typeof field.revision === "bigint"
|
||||
? field.revision
|
||||
: typeof field.revision === "number"
|
||||
? BigInt(field.revision)
|
||||
: typeof field.revision === "string" && /^\d+$/.test(field.revision)
|
||||
? BigInt(field.revision)
|
||||
: 0n,
|
||||
}),
|
||||
});
|
||||
};
|
||||
const codecFromSource = (value) => {
|
||||
if (!isRecord(value) || !isRecord(value.field)) {
|
||||
return undefined;
|
||||
}
|
||||
const field = value.field;
|
||||
return {
|
||||
...(typeof field.fieldType === "string" ? { fieldType: field.fieldType } : {}),
|
||||
...(typeof field.conflictStrategy === "string"
|
||||
? { conflictStrategy: field.conflictStrategy }
|
||||
: {}),
|
||||
};
|
||||
};
|
||||
const withSource = (value, source) => {
|
||||
const sourceValue = sourceToProto(source);
|
||||
return sourceValue ? create(ValueSchema, { ...value, source: sourceValue }) : value;
|
||||
};
|
||||
export const jsToProtoValue = (value, codec) => {
|
||||
if (codec?.conflictStrategy === "crdt" && codec.fieldType) {
|
||||
return jsToProtoValue(encodeAutomergeFieldWrite(value, codec.fieldType));
|
||||
}
|
||||
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: value.toString() },
|
||||
});
|
||||
}
|
||||
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((entry) => jsToProtoValue(entry)),
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (isRecord(value)) {
|
||||
if (Object.hasOwn(value, "$caminoValue") && isRecord(value.$caminoSource)) {
|
||||
return withSource(jsToProtoValue(value.$caminoValue, codecFromSource(value.$caminoSource)), value.$caminoSource);
|
||||
}
|
||||
if (typeof value.$caminoRef === "string") {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "refValue",
|
||||
value: create(RefValueSchema, {
|
||||
objectId: value.$caminoRef,
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (typeof value.$caminoBytes === "string" &&
|
||||
Object.keys(value).every((key) => key === "$caminoBytes")) {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "bytesValue",
|
||||
value: base64ToBytes(value.$caminoBytes),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (typeof value.$caminoCrdtType === "string" &&
|
||||
typeof value.$caminoCrdtPayload === "string") {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "crdtValue",
|
||||
value: create(CrdtValueSchema, {
|
||||
type: value.$caminoCrdtType,
|
||||
encoding: typeof value.$caminoCrdtEncoding === "string"
|
||||
? value.$caminoCrdtEncoding
|
||||
: "base64",
|
||||
payload: base64ToBytes(value.$caminoCrdtPayload),
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "objectValue",
|
||||
value: create(ObjectValueSchema, {
|
||||
fields: jsObjectToProtoFields(value),
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
throw new Error(`Unsupported Camino RPC value: ${typeof value}`);
|
||||
};
|
||||
export const protoValueToJs = (value) => {
|
||||
switch (value?.kind.case) {
|
||||
case "nullValue":
|
||||
case undefined:
|
||||
return null;
|
||||
case "boolValue":
|
||||
case "numberValue":
|
||||
case "stringValue":
|
||||
return value.kind.value;
|
||||
case "integerValue":
|
||||
return value.kind.value;
|
||||
case "bytesValue":
|
||||
return bytesToBase64(value.kind.value);
|
||||
case "listValue":
|
||||
return value.kind.value.values.map(protoValueToJs);
|
||||
case "objectValue":
|
||||
return protoFieldsToJs(value.kind.value.fields);
|
||||
case "refValue":
|
||||
return value.kind.value.objectId;
|
||||
case "crdtValue":
|
||||
return loadAutomergeFieldDoc({
|
||||
$caminoCrdtType: value.kind.value.type,
|
||||
$caminoCrdtEncoding: value.kind.value.encoding === "base64" ? "base64" : "base64",
|
||||
$caminoCrdtPayload: bytesToBase64(value.kind.value.payload),
|
||||
}, value.kind.value.type);
|
||||
}
|
||||
};
|
||||
export const jsObjectToProtoFields = (value) => Object.fromEntries(Object.entries(value)
|
||||
.filter(([, fieldValue]) => fieldValue !== undefined)
|
||||
.map(([key, fieldValue]) => [key, jsToProtoValue(fieldValue)]));
|
||||
export const protoFieldsToJs = (fields) => Object.fromEntries(Object.entries(fields ?? {}).map(([key, value]) => [
|
||||
key,
|
||||
protoValueToJs(value),
|
||||
]));
|
||||
const normalizeForCanonicalJson = (value) => {
|
||||
if (Array.isArray(value)) {
|
||||
return value.map(normalizeForCanonicalJson);
|
||||
}
|
||||
if (!value || typeof value !== "object") {
|
||||
return value;
|
||||
}
|
||||
return Object.fromEntries(Object.entries(value)
|
||||
.filter(([, entryValue]) => entryValue !== undefined)
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([key, entryValue]) => [key, normalizeForCanonicalJson(entryValue)]));
|
||||
};
|
||||
export const canonicalFieldParams = (params) => JSON.stringify(normalizeForCanonicalJson(Object.fromEntries(Object.entries(params ?? {}).map(([key, value]) => [
|
||||
key,
|
||||
protoValueToJs(value),
|
||||
]))));
|
||||
export const fieldOperationKey = (params) => [
|
||||
params.objectId,
|
||||
params.fieldName,
|
||||
Buffer.from(canonicalFieldParams(params.input)).toString("base64url"),
|
||||
].join(":");
|
||||
const derivedDependencyToProto = (dependency) => create(DerivedDependencySchema, {
|
||||
kind: dependency.kind,
|
||||
objectId: dependency.objectId,
|
||||
fieldName: dependency.kind === "field" ? dependency.fieldName : "",
|
||||
sourceField: dependency.kind === "edges" ? dependency.sourceField ?? "" : "",
|
||||
});
|
||||
const fieldSourceForSpec = (spec, objectId) => {
|
||||
if (!spec.field) {
|
||||
return undefined;
|
||||
}
|
||||
return {
|
||||
field: {
|
||||
objectId,
|
||||
fieldName: spec.field.name,
|
||||
...(spec.field.type ? { fieldType: spec.field.type } : {}),
|
||||
...(spec.field.storage ? { fieldStorage: spec.field.storage } : {}),
|
||||
...(spec.field.conflict ? { conflictStrategy: spec.field.conflict } : {}),
|
||||
},
|
||||
};
|
||||
};
|
||||
const wrapFieldResult = (spec, objectId, value) => {
|
||||
const source = fieldSourceForSpec(spec, objectId);
|
||||
return source ? { $caminoValue: value, $caminoSource: source } : value;
|
||||
};
|
||||
const derivedEmissionToProtoForSpec = (spec, objectId, event) => create(WatchEventSchema, {
|
||||
watchId: event.watchId,
|
||||
value: jsToProtoValue(event.error ? null : wrapFieldResult(spec, objectId, event.value ?? null)),
|
||||
dependencies: event.dependencies.map(derivedDependencyToProto),
|
||||
error: event.error ?? "",
|
||||
initial: event.initial,
|
||||
});
|
||||
const watchIdFromResult = (value) => {
|
||||
if (!isRecord(value)) {
|
||||
return "";
|
||||
}
|
||||
const watchId = value.watch_id ?? value.watchId;
|
||||
return typeof watchId === "string" ? watchId : "";
|
||||
};
|
||||
export const objectRef = (objectId) => objectId;
|
||||
export const createField = (camino, objectId, fieldName, codec) => ({
|
||||
async get() {
|
||||
recordDerivedDependency({ kind: "field", objectId, fieldName });
|
||||
const response = await camino.getObject({ objectId });
|
||||
return protoValueToJs(response.object?.fields[fieldName]);
|
||||
},
|
||||
async set(value) {
|
||||
await camino.setField({
|
||||
objectId,
|
||||
fieldName,
|
||||
value: jsToProtoValue(value, codec),
|
||||
});
|
||||
},
|
||||
});
|
||||
export const createCrdtField = (camino, objectId, fieldName, fieldType) => {
|
||||
const field = createField(camino, objectId, fieldName, {
|
||||
fieldType,
|
||||
conflictStrategy: "crdt",
|
||||
});
|
||||
return {
|
||||
...field,
|
||||
async change(fn) {
|
||||
const current = await field.get();
|
||||
if (!current) {
|
||||
throw new Error(`Cannot change missing CRDT field ${fieldName}`);
|
||||
}
|
||||
const next = Automerge.change(current, (doc) => {
|
||||
fn(doc);
|
||||
});
|
||||
await field.set(next);
|
||||
return next;
|
||||
},
|
||||
async replaceFromJson(value) {
|
||||
await camino.setField({
|
||||
objectId,
|
||||
fieldName,
|
||||
value: jsToProtoValue(createAutomergeFieldValue(value, fieldType)),
|
||||
});
|
||||
},
|
||||
};
|
||||
};
|
||||
const createDependencyTrackingCaminoClient = (camino) => new Proxy(camino, {
|
||||
get(target, property, receiver) {
|
||||
if (property === "getObject") {
|
||||
return async (request, ...args) => {
|
||||
if (request.objectId) {
|
||||
recordDerivedDependency({
|
||||
kind: "object",
|
||||
objectId: request.objectId,
|
||||
});
|
||||
}
|
||||
return await Reflect.get(target, property, receiver)(request, ...args);
|
||||
};
|
||||
}
|
||||
if (property === "listEdges") {
|
||||
return async (request, ...args) => {
|
||||
if (request.objectId) {
|
||||
recordDerivedDependency({
|
||||
kind: "edges",
|
||||
objectId: request.objectId,
|
||||
...(request.sourceField ? { sourceField: request.sourceField } : {}),
|
||||
});
|
||||
}
|
||||
return await Reflect.get(target, property, receiver)(request, ...args);
|
||||
};
|
||||
}
|
||||
return Reflect.get(target, property, receiver);
|
||||
},
|
||||
});
|
||||
export const serveQuixosPackageRuntime = (params) => {
|
||||
const host = process.env.QUIXOS_RUNTIME_HOST ?? "127.0.0.1";
|
||||
const port = Number(process.env.QUIXOS_RUNTIME_PORT ?? "0");
|
||||
const caminoUrl = process.env.CAMINO_URL ?? "http://127.0.0.1:7310";
|
||||
const runtimeToken = process.env.CAMINO_RUNTIME_AUTH_TOKEN ?? "";
|
||||
if (process.env.CAMINO_RUNTIME_AUTH_REQUIRED === "1" && !runtimeToken) {
|
||||
throw new Error("CAMINO_RUNTIME_AUTH_REQUIRED=1 but CAMINO_RUNTIME_AUTH_TOKEN is missing");
|
||||
}
|
||||
const runtimeTokenInterceptor = (next) => async (request) => {
|
||||
if (runtimeToken) {
|
||||
request.header.set("x-camino-runtime-token", runtimeToken);
|
||||
}
|
||||
return await next(request);
|
||||
};
|
||||
const camino = createClient(CaminoService, createConnectTransport({
|
||||
baseUrl: caminoUrl,
|
||||
httpVersion: "1.1",
|
||||
interceptors: runtimeToken ? [runtimeTokenInterceptor] : [],
|
||||
}));
|
||||
const trackingCamino = createDependencyTrackingCaminoClient(camino);
|
||||
if (!Number.isInteger(port) || port <= 0) {
|
||||
throw new Error("QUIXOS_RUNTIME_PORT must be a positive integer");
|
||||
}
|
||||
const functionRefs = params.functions.map((entry) => create(FunctionRefSchema, {
|
||||
packageNamespace: params.packageNamespace,
|
||||
packageName: params.packageName,
|
||||
symbol: entry.symbol,
|
||||
operation: entry.operation ?? "",
|
||||
}));
|
||||
const resolveRuntimeFunction = (fn) => {
|
||||
const symbol = fn?.symbol ?? "";
|
||||
const spec = params.functions.find((entry) => entry.symbol === symbol &&
|
||||
(entry.operation ?? "") === (fn?.operation ?? ""));
|
||||
if (!spec) {
|
||||
throw new Error(`Unknown ${params.packageName} function: ${symbol}${fn?.operation ? `.${fn.operation}` : ""}`);
|
||||
}
|
||||
const implementationExport = params.implementation[spec.exportName];
|
||||
const runtimeFunction = spec.operation && typeof implementationExport === "object"
|
||||
? implementationExport[spec.operation]
|
||||
: implementationExport;
|
||||
if (!runtimeFunction) {
|
||||
throw new Error(`Missing ${params.packageName} implementation export: ${spec.exportName}${spec.operation ? `.${spec.operation}` : ""}`);
|
||||
}
|
||||
if (typeof runtimeFunction !== "function") {
|
||||
throw new Error(`${params.packageName} implementation export is not callable: ${spec.exportName}${spec.operation ? `.${spec.operation}` : ""}`);
|
||||
}
|
||||
return { spec, runtimeFunction };
|
||||
};
|
||||
const handler = connectNodeAdapter({
|
||||
routes: (router) => {
|
||||
router.service(PackageRuntime, {
|
||||
async handshake() {
|
||||
return create(HandshakeResponseSchema, {
|
||||
packageNamespace: params.packageNamespace,
|
||||
packageName: params.packageName,
|
||||
runtimeProtocolVersion: params.runtimeProtocolVersion,
|
||||
functions: functionRefs,
|
||||
});
|
||||
},
|
||||
async invoke(request) {
|
||||
try {
|
||||
const { spec, runtimeFunction } = resolveRuntimeFunction(request.function);
|
||||
const value = await runtimeFunction(params.createContext(trackingCamino, request));
|
||||
return create(InvokeResponseSchema, {
|
||||
ok: true,
|
||||
result: jsToProtoValue(spec.operation === "get"
|
||||
? wrapFieldResult(spec, request.objectId, value)
|
||||
: value),
|
||||
});
|
||||
}
|
||||
catch (error) {
|
||||
return create(InvokeResponseSchema, {
|
||||
ok: false,
|
||||
result: jsToProtoValue(null),
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
},
|
||||
async *watch(request, context) {
|
||||
const queue = new AsyncEventQueue();
|
||||
let watchId = "";
|
||||
const closeQueue = () => queue.close();
|
||||
context.signal.addEventListener("abort", closeQueue, { once: true });
|
||||
try {
|
||||
const { spec, runtimeFunction } = resolveRuntimeFunction(request.function);
|
||||
if (spec.operation !== "watchStart") {
|
||||
throw new Error(`PackageRuntime.watch requires a watchStart operation, got ${spec.symbol}${spec.operation ? `.${spec.operation}` : ""}`);
|
||||
}
|
||||
const sink = {
|
||||
publish(event) {
|
||||
queue.push(derivedEmissionToProtoForSpec(spec, request.objectId, event));
|
||||
},
|
||||
};
|
||||
const result = await watchSinkStorage.run(sink, async () => runtimeFunction(params.createContext(trackingCamino, request)));
|
||||
watchId = watchIdFromResult(result);
|
||||
if (!watchId) {
|
||||
throw new Error(`Watch start did not return a watch_id for ${spec.symbol}`);
|
||||
}
|
||||
for await (const event of queue) {
|
||||
yield event;
|
||||
}
|
||||
}
|
||||
catch (error) {
|
||||
yield create(WatchEventSchema, {
|
||||
watchId,
|
||||
value: jsToProtoValue(null),
|
||||
dependencies: [],
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
initial: !watchId,
|
||||
});
|
||||
}
|
||||
finally {
|
||||
context.signal.removeEventListener("abort", closeQueue);
|
||||
queue.close();
|
||||
if (watchId) {
|
||||
try {
|
||||
const stopFunction = create(FunctionRefSchema, {
|
||||
...(request.function ?? create(FunctionRefSchema, {})),
|
||||
operation: "watchStop",
|
||||
});
|
||||
const { runtimeFunction } = resolveRuntimeFunction(stopFunction);
|
||||
await runtimeFunction(params.createContext(trackingCamino, create(WatchRequestSchema, {
|
||||
...request,
|
||||
function: stopFunction,
|
||||
input: {
|
||||
...request.input,
|
||||
watch_id: jsToProtoValue(watchId),
|
||||
},
|
||||
})));
|
||||
}
|
||||
catch (error) {
|
||||
logDerivedWatch("stop-error", {
|
||||
watchId,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
const server = http.createServer((request, response) => {
|
||||
void handler(request, response);
|
||||
});
|
||||
const close = () => server.close(() => process.exit(0));
|
||||
server.listen(port, host, () => {
|
||||
process.stderr.write(`${params.packageName} listening on ${host}:${port}\n`);
|
||||
});
|
||||
process.on("SIGTERM", close);
|
||||
process.on("SIGINT", close);
|
||||
};
|
||||
Reference in New Issue
Block a user