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"; export const qxDerived = (get) => ({ kind: "derived", get }); // 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) } : {}), ...(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, Object.fromEntries(Object.entries(port.operations).map(([name, operation]) => [name, async (input) => decodeQxValue(operation.outputType, (await target.live(operation.id, inputFields(operation.inputType, input, messages))).$quixosValue, messages), ]))]; } case "constructor": return [name, { construct: (input) => raw.constructor(port.id).construct(inputFields(port.inputType, input, messages)) }]; } })); return { objectId: raw.objectId, signal: raw.signal, ...(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); };