Timothy J. Aveni
248 lines
12 KiB
JavaScript
248 lines
12 KiB
JavaScript
import { create } from "@bufbuild/protobuf";
|
|
import { ValueSchema, ObjectValueSchema } from "./camino/api_pb.js";
|
|
import { derived, jsToProtoValue, liveValue, protoValueToJs, } from "./index.js";
|
|
import { assertReferenceFree, referenceToWire } from "./references.js";
|
|
import { decodeQuerySnapshot } from "./queries.js";
|
|
export const defineQxInterfaceContract = (interfaceRevisionId, operations) => Object.freeze({ interfaceRevisionId, operations });
|
|
export const qxDerived = (get) => ({ kind: "derived", get });
|
|
const bindInterfaceView = (target, contract, messages) => ({
|
|
objectId: target.objectId,
|
|
contract,
|
|
live: Object.fromEntries(Object.entries(contract.operations).map(([name, operation]) => [
|
|
name,
|
|
(input) => target.live(operation.id, inputFields(operation.inputType, input, messages)),
|
|
])),
|
|
...Object.fromEntries(Object.entries(contract.operations).map(([name, operation]) => [
|
|
name,
|
|
async (input) => decodeQxValue(operation.outputType, (await target.live(operation.id, inputFields(operation.inputType, input, messages))).$quixosValue, messages),
|
|
])),
|
|
});
|
|
// Conversion belongs at the binding boundary. It does not add orchestrator validation.
|
|
export const decodeQxValue = (type, value, messages) => {
|
|
if (type.kind === "builtin" && type.name === "unit")
|
|
return null;
|
|
if (type.kind === "optional" && !value)
|
|
return null;
|
|
if (!value)
|
|
throw new Error("Missing QX wire value");
|
|
if (type.kind === "record") {
|
|
if (value.kind.case !== "objectValue")
|
|
throw new Error("Expected QX record");
|
|
const fields = value.kind.value.fields;
|
|
if (Object.keys(fields).some((name) => !Object.hasOwn(type.fields, name)))
|
|
throw new Error("Unexpected QX record field");
|
|
return Object.fromEntries(Object.entries(type.fields).map(([name, field]) => [name, decodeQxValue(field, fields[name], messages)]));
|
|
}
|
|
if (type.kind === "optional")
|
|
return value.kind.case === "nullValue" ? null : decodeQxValue(type.value, value, messages);
|
|
if (type.kind === "list") {
|
|
if (value.kind.case !== "listValue")
|
|
throw new Error("Expected QX list");
|
|
return value.kind.value.values.map((entry) => decodeQxValue(type.value, entry, messages));
|
|
}
|
|
if (type.kind === "message") {
|
|
assertReferenceFree(protoValueToJs(value));
|
|
const decoded = requireMessage(messages, type.descriptorId).decode(value);
|
|
assertReferenceFree(decoded);
|
|
return decoded;
|
|
}
|
|
if (type.kind === "object-ref") {
|
|
if (value.kind.case !== "refValue")
|
|
throw new Error("Expected a declared RPC object reference");
|
|
return protoValueToJs(value);
|
|
}
|
|
if (value.kind.case === "refValue")
|
|
throw new Error("Reference supplied to a non-reference value");
|
|
if (type.kind === "scalar") {
|
|
if (type.name === "int64" || type.name === "uint64") {
|
|
if (value.kind.case !== "integerValue")
|
|
throw new Error("Expected QX integer");
|
|
return BigInt(value.kind.value);
|
|
}
|
|
if (type.name === "bytes") {
|
|
if (value.kind.case !== "bytesValue")
|
|
throw new Error("Expected QX bytes");
|
|
return value.kind.value;
|
|
}
|
|
}
|
|
return protoValueToJs(value);
|
|
};
|
|
const requireMessage = (messages, id) => {
|
|
const binding = messages[id];
|
|
if (!binding)
|
|
throw new Error(`Missing message binding ${id}`);
|
|
return binding;
|
|
};
|
|
export const encodeQxValue = (type, value, messages) => {
|
|
if (type.kind === "builtin" && type.name === "unit")
|
|
return jsToProtoValue(null);
|
|
if (type.kind === "record") {
|
|
if (!value || typeof value !== "object" || Array.isArray(value))
|
|
throw new Error("Expected QX record");
|
|
if (Object.keys(value).some((name) => !Object.hasOwn(type.fields, name)))
|
|
throw new Error("Unexpected QX record field");
|
|
const fields = Object.fromEntries(Object.entries(type.fields).map(([name, field]) => {
|
|
if (!Object.hasOwn(value, name) && field.kind !== "optional")
|
|
throw new Error(`Missing QX record field ${name}`);
|
|
return [name, encodeQxValue(field, value[name] ?? (field.kind === "optional" ? null : value[name]), messages)];
|
|
}));
|
|
return create(ValueSchema, { kind: { case: "objectValue", value: create(ObjectValueSchema, { fields }) } });
|
|
}
|
|
if (type.kind === "optional")
|
|
return value === null ? jsToProtoValue(null) : encodeQxValue(type.value, value, messages);
|
|
if (type.kind === "list")
|
|
return jsToProtoValue(value.map((entry) => liveValue(encodeQxValue(type.value, entry, messages))));
|
|
if (type.kind === "object-ref") {
|
|
referenceToWire(value);
|
|
return jsToProtoValue(value);
|
|
}
|
|
assertReferenceFree(value);
|
|
if (type.kind === "message") {
|
|
const encoded = requireMessage(messages, type.descriptorId).encode(value);
|
|
assertReferenceFree(protoValueToJs(encoded));
|
|
return encoded;
|
|
}
|
|
return jsToProtoValue(value);
|
|
};
|
|
const inputValue = (context, type) => {
|
|
if (type.kind === "message" || type.kind === "record")
|
|
return create(ValueSchema, {
|
|
kind: { case: "objectValue", value: create(ObjectValueSchema, { fields: context.inputProto }) },
|
|
});
|
|
return context.inputProto.value;
|
|
};
|
|
const inputFields = (type, value, messages) => {
|
|
if (type.kind === "builtin" && type.name === "unit")
|
|
return {};
|
|
const encoded = encodeQxValue(type, value, messages);
|
|
if (type.kind === "message" || type.kind === "record") {
|
|
if (encoded.kind.case !== "objectValue")
|
|
throw new Error("Message inputs must encode an object value");
|
|
return Object.fromEntries(Object.entries(encoded.kind.value.fields).map(([key, entry]) => [key, liveValue(entry)]));
|
|
}
|
|
return { value: liveValue(encoded) };
|
|
};
|
|
/** The sole unchecked cast connects generated contracts to the dynamic RPC runtime. */
|
|
export const bindQxHandler = (spec, handler, messages) => {
|
|
const bindContext = (raw) => {
|
|
const ports = Object.fromEntries(Object.entries(spec.ports).map(([name, port]) => {
|
|
switch (port.kind) {
|
|
case "state": {
|
|
const state = raw.state(port.id);
|
|
return [
|
|
name,
|
|
{
|
|
...(port.primitives.includes("read")
|
|
? {
|
|
get: async () => decodeQxValue(port.valueType, (await state.live()).$quixosValue, messages),
|
|
live: () => state.live(),
|
|
}
|
|
: {}),
|
|
...(port.primitives.includes("write")
|
|
? {
|
|
set: async (value) => state.set(liveValue(encodeQxValue(port.valueType, value, messages))),
|
|
}
|
|
: {}),
|
|
},
|
|
];
|
|
}
|
|
case "edge": {
|
|
const edge = raw.edge(port.id);
|
|
return [
|
|
name,
|
|
{
|
|
...Object.fromEntries(port.primitives.map((primitive) => [
|
|
primitive,
|
|
edge[primitive],
|
|
])),
|
|
...(port.primitives.includes("resolve") ? { collection: edge.collection } : {}),
|
|
...(port.primitives.includes("resolve") &&
|
|
port.primitives.includes("connect") &&
|
|
port.primitives.includes("disconnect")
|
|
? { replace: edge.replace }
|
|
: {}),
|
|
},
|
|
];
|
|
}
|
|
case "interface": {
|
|
const target = raw.interface(port.id);
|
|
return [
|
|
name,
|
|
bindInterfaceView(target, defineQxInterfaceContract(port.interfaceRevisionId, port.operations), messages),
|
|
];
|
|
}
|
|
case "query": {
|
|
const query = raw.query(port.id);
|
|
const variablesToWire = (variables) => {
|
|
const value = encodeQxValue(port.variables, variables, messages);
|
|
if (value.kind.case !== "objectValue")
|
|
throw new Error("QUERY_VARIABLE_INVALID");
|
|
return value.kind.value.fields;
|
|
};
|
|
return [
|
|
name,
|
|
{
|
|
async execute(variables) {
|
|
const response = await query.execute(variablesToWire(variables), port.definitionDigest);
|
|
if (response.pending.length || response.errors.length)
|
|
throw new Error("QUERY_INCOMPLETE");
|
|
return decodeQxValue(port.output, response.value, messages);
|
|
},
|
|
...(port.watch
|
|
? {
|
|
async *watch(variables, signal) {
|
|
for await (const event of query.watch(variablesToWire(variables), signal, port.definitionDigest)) {
|
|
if (!event.snapshot)
|
|
throw new Error("QUERY_SNAPSHOT_MISSING");
|
|
yield decodeQuerySnapshot(event.snapshot, port.output, event.runId, event.sequence);
|
|
}
|
|
},
|
|
}
|
|
: {}),
|
|
},
|
|
];
|
|
}
|
|
case "constructor":
|
|
return [
|
|
name,
|
|
{
|
|
construct: (input) => raw.constructor(port.id).construct(inputFields(port.inputType, input, messages)),
|
|
},
|
|
];
|
|
}
|
|
}));
|
|
return {
|
|
conform: {
|
|
async tryConform(object, contract) {
|
|
const target = await raw.tryConform(object, contract.interfaceRevisionId);
|
|
return target
|
|
? bindInterfaceView(target, contract, messages)
|
|
: undefined;
|
|
},
|
|
},
|
|
...(spec.receiver === "none" ? {} : { objectId: raw.objectId }),
|
|
signal: raw.signal,
|
|
...(spec.receiver !== "none" && raw.openSession
|
|
? {
|
|
openSession: async () => {
|
|
const session = await raw.openSession();
|
|
return {
|
|
id: session.id,
|
|
close: () => session.close(),
|
|
run: (work) => session.run((next) => work(bindContext(next))),
|
|
};
|
|
},
|
|
}
|
|
: {}),
|
|
input: decodeQxValue(spec.inputType, inputValue(raw, spec.inputType), messages),
|
|
ports,
|
|
};
|
|
};
|
|
const execute = async (raw) => {
|
|
const context = bindContext(raw);
|
|
const value = await (typeof handler === "function" ? handler(context) : handler.get(context));
|
|
return liveValue(encodeQxValue(spec.eventType ?? spec.outputType, value, messages));
|
|
};
|
|
return typeof handler === "function" ? execute : derived(execute);
|
|
};
|