Timothy J. Aveni
732 lines
29 KiB
JavaScript
732 lines
29 KiB
JavaScript
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);
|
|
};
|