Implement query execution, scoped RPC enrichment, live collections, and scaffold integration

2026-09-17 14:34:16 -07:00
parent 7e69e675ba
commit c1330ae8e3
29 changed files with 2217 additions and 77 deletions
@@ -2,6 +2,7 @@ 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) => ({
@@ -170,6 +171,37 @@ export const bindQxHandler = (spec, handler, messages) => {
                        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,