Generate typed QX bindings and add source editing tools

src/bindings.ts

@@ -0,0 +1,121 @@
+import { create } from "@bufbuild/protobuf";
+import { ValueSchema, ObjectValueSchema, type Value } from "./camino/api_pb.js";
+import { derived, jsToProtoValue, liveValue, protoValueToJs,
+  type RuntimeContext, type RuntimeHandler, type DerivedHandler } from "./index.js";
+
+declare const referenceBrand: unique symbol;
+export type QxObjectRef<Identity extends string> = string & { readonly [referenceBrand]: { readonly [K in Identity]: true } };
+declare const watchBrand: unique symbol;
+export type QxWatchHandle = string & { readonly [watchBrand]: true };
+export type MessageBinding<T> = { encode(value: T): Value; decode(value: Value): T };
+export type BindingValue<B> = B extends MessageBinding<infer T> ? T : never;
+export type QxHandler<C, O> = (context: C) => O | Promise<O>;
+export type QxDerived<C, O> = { kind: "derived"; get: QxHandler<C, O> };
+export const qxDerived = <C, O>(get: QxHandler<C, O>): QxDerived<C, O> => ({ kind: "derived", get });
+
+/** Versioned binding ABI. This mirrors the language-neutral value IR. */
+export type QxValueType =
+  | { kind: "builtin"; name: "unit" | "watch-handle" }
+  | { kind: "scalar"; name: string }
+  | { kind: "message"; descriptorId: string }
+  | { kind: "object-ref"; expectation: unknown }
+  | { kind: "optional" | "list"; value: QxValueType };
+export type QxOperationSpec = { id: string; inputType: QxValueType; outputType: QxValueType };
+export type QxPortSpec =
+  | { kind: "state"; id: string; valueType: QxValueType; primitives: string[] }
+  | { kind: "edge"; id: string; primitives: string[] }
+  | { kind: "interface"; id: string; operations: Record<string, QxOperationSpec> }
+  | { kind: "constructor"; id: string; inputType: QxValueType };
+export type QxHandlerSpec = {
+  inputType: QxValueType; outputType: QxValueType; eventType?: QxValueType;
+  ports: Record<string, QxPortSpec>;
+};
+export type QxMessages = Record<string, MessageBinding<any>>;
+
+// Conversion belongs at the binding boundary. It does not add orchestrator validation.
+export const decodeQxValue = (type: QxValueType, value: Value | undefined, messages: QxMessages): any => {
+  if (type.kind === "builtin" && type.name === "unit") return null;
+  if (!value) throw new Error("Missing QX wire value");
+  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") return requireMessage(messages, type.descriptorId).decode(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: QxMessages, id: string) => {
+  const binding = messages[id];
+  if (!binding) throw new Error(`Missing message binding ${id}`);
+  return binding;
+};
+export const encodeQxValue = (type: QxValueType, value: any, messages: QxMessages): Value => {
+  if (type.kind === "builtin" && type.name === "unit") return jsToProtoValue(null);
+  if (type.kind === "optional") return value === null ? jsToProtoValue(null) : encodeQxValue(type.value, value, messages);
+  if (type.kind === "list") return jsToProtoValue(value.map((entry: unknown) => liveValue(encodeQxValue(type.value, entry, messages))));
+  if (type.kind === "message") return requireMessage(messages, type.descriptorId).encode(value);
+  if (type.kind === "object-ref") return jsToProtoValue({ $quixosRef: value });
+  return jsToProtoValue(value);
+};
+
+const inputValue = (context: RuntimeContext, type: QxValueType) => {
+  if (type.kind === "message") return create(ValueSchema, { kind: { case: "objectValue",
+    value: create(ObjectValueSchema, { fields: context.inputProto }) } });
+  return context.inputProto.value;
+};
+const inputFields = (type: QxValueType, value: unknown, messages: QxMessages): Record<string, unknown> => {
+  if (type.kind === "builtin" && type.name === "unit") return {};
+  const encoded = encodeQxValue(type, value, messages);
+  if (type.kind === "message") {
+    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 = <C, O>(
+  spec: QxHandlerSpec, handler: QxHandler<C, O> | QxDerived<C, O>, messages: QxMessages,
+): RuntimeHandler | DerivedHandler => {
+  const execute = async (raw: RuntimeContext) => {
+    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: unknown) => 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 as "resolve" | "connect" | "disconnect"]]))];
+        }
+        case "interface": {
+          const target = raw.interface(port.id);
+          return [name, Object.fromEntries(Object.entries(port.operations).map(([name, operation]) => [name,
+            async (input: unknown) => decodeQxValue(operation.outputType,
+              (await target.live(operation.id, inputFields(operation.inputType, input, messages))).$quixosValue, messages),
+          ]))];
+        }
+        case "constructor": return [name, { construct: (input: unknown) => raw.constructor(port.id).construct(inputFields(port.inputType, input, messages)) }];
+      }
+    }));
+    const context = { objectId: raw.objectId,
+      input: decodeQxValue(spec.inputType, inputValue(raw, spec.inputType), messages), ports } as C;
+    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);
+};

src/index.ts

@@ -1,4 +1,5 @@
 import http from "node:http";
+export * from "./bindings.js";
 import { AsyncLocalStorage } from "node:async_hooks";
 import { randomUUID } from "node:crypto";
 import { create, equals } from "@bufbuild/protobuf";
@@ -137,6 +138,7 @@ export type EdgePort = {
   projectionId: string;
   resolve(): Promise<string[]>;
   connect(targetObjectId: string): Promise<void>;
+  disconnect(targetObjectId: string): Promise<void>;
 };
 export type InterfacePort = {
   objectId: string;
@@ -210,6 +212,12 @@ export const createRuntimeContext = (
           async connect(targetObjectId) {
             await camino.connectEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId, targetObjectId });
           },
+          async disconnect(targetObjectId) {
+            const result = await camino.resolveEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId });
+            for (const entry of result.edges) {
+              if (targetForEdge(entry, projectionId) === targetObjectId) await camino.disconnectEdge({ edgeId: entry.id });
+            }
+          },
         };
         ports.set(dependency.portId, edge);
         break;