diff --git a/.quixos-subtree-source.json b/.quixos-subtree-source.json index 6ecbbcf..1135ecb 100644 --- a/.quixos-subtree-source.json +++ b/.quixos-subtree-source.json @@ -1,7 +1,7 @@ { "version": 1, "sourceRepo": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos.git", - "sourceCommit": "4a2b76ed15a52d06b422fb3d1f651be47d92ff51", + "sourceCommit": "d3e9e4b0f6d083d578dc7ce36e0f169023603116", "sourcePath": "quixos-instance/packages/camino-package-runtime", "exportName": "camino-package-runtime", "mirrorRemote": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/camino-package-runtime.git" diff --git a/flake.lock b/flake.lock index 9a11122..3168b50 100644 --- a/flake.lock +++ b/flake.lock @@ -74,17 +74,17 @@ "nixpkgs": "nixpkgs_2" }, "locked": { - "lastModified": 1783886666, - "narHash": "sha256-B+WyKSrCS/3rjCN7opGjpkAxWDqB5A9K/ISKYxFeRyo=", + "lastModified": 1783890039, + "narHash": "sha256-5R7WV6KCSYwQ9SmjmIvRovPjG551icfYa7XHBz68sYA=", "ref": "refs/heads/exported", - "rev": "a8dc34db0ed32e74cee9859d7a9fbaa699cb0f4c", - "revCount": 2, + "rev": "a287ab31330222d7dcd517c640a5de89f48bc4d0", + "revCount": 4, "type": "git", "url": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-protocol.git" }, "original": { "ref": "refs/heads/exported", - "rev": "a8dc34db0ed32e74cee9859d7a9fbaa699cb0f4c", + "rev": "a287ab31330222d7dcd517c640a5de89f48bc4d0", "type": "git", "url": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-protocol.git" } diff --git a/flake.nix b/flake.nix index 88b6b92..239a42d 100644 --- a/flake.nix +++ b/flake.nix @@ -4,7 +4,7 @@ inputs = { nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable"; flake-utils.url = "github:numtide/flake-utils"; - quixos-protocol.url = "git+https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-protocol.git?ref=refs/heads/exported&rev=a8dc34db0ed32e74cee9859d7a9fbaa699cb0f4c"; + quixos-protocol.url = "git+https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-protocol.git?ref=refs/heads/exported&rev=a287ab31330222d7dcd517c640a5de89f48bc4d0"; quixosNixHelpers = { url = "git+https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-nix-helpers.git?ref=refs/heads/exported&rev=81e55657d9ce0ba092cffd69ee15179ea02aa598"; flake = false; diff --git a/src/camino/api_pb.ts b/src/camino/api_pb.ts index c545abd..ef23474 100644 --- a/src/camino/api_pb.ts +++ b/src/camino/api_pb.ts @@ -854,3 +854,4 @@ export const CaminoService: GenService<{ }, }> = /*@__PURE__*/ serviceDesc(file_camino_api, 0); + diff --git a/src/camino/schema_pb.ts b/src/camino/schema_pb.ts index c9729b2..09e6ea0 100644 --- a/src/camino/schema_pb.ts +++ b/src/camino/schema_pb.ts @@ -695,3 +695,4 @@ export enum FieldImplementation { */ export const FieldImplementationSchema: GenEnum = /*@__PURE__*/ enumDesc(file_camino_schema, 4); + diff --git a/src/index.ts b/src/index.ts index df770df..65f1474 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,7 +2,11 @@ import http from "node:http"; import { AsyncLocalStorage } from "node:async_hooks"; import { randomUUID } from "node:crypto"; import { create } from "@bufbuild/protobuf"; -import { createClient, type Client } from "@connectrpc/connect"; +import { + createClient, + type Client, + type HandlerContext, +} from "@connectrpc/connect"; import { connectNodeAdapter, createConnectTransport, @@ -18,10 +22,15 @@ import { type Value, } from "./camino/api_pb.js"; import { + DerivedDependencySchema, HandshakeResponseSchema, InvokeResponseSchema, PackageRuntime, + WatchEventSchema, + WatchRequestSchema, type InvokeRequest, + type WatchEvent, + type WatchRequest, } from "./quixos/runtime_pb.js"; import { FunctionRefSchema } from "./quixos/refs_pb.js"; @@ -122,14 +131,30 @@ type WatchableContext = { request: InvokeRequest; }; +type DerivedWatchEmission = { + watchId: string; + value?: unknown; + dependencies: DerivedDependency[]; + error?: string; + initial: boolean; +}; + +type DerivedWatchSink = { + publish(event: DerivedWatchEmission): void; +}; + +const watchSinkStorage = new AsyncLocalStorage(); + type ActiveDerivedWatch = { watchId: string; context: Context; get: RuntimeHandler; + sink?: DerivedWatchSink; controllers: Map; running: boolean; pending: boolean; stopped: boolean; + emitted: boolean; lastValue: unknown; }; @@ -198,13 +223,28 @@ export const derivedWithDeps = (operation: { 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: error instanceof Error ? error.message : String(error), + error: message, }); + watch.sink?.publish({ + watchId: watch.watchId, + dependencies: [], + error: message, + initial: !watch.emitted, + }); + watch.emitted = true; } finally { watch.running = false; } @@ -295,10 +335,12 @@ export const derivedWithDeps = (operation: { watchId, context: context as DerivedContext, get: operation.get as RuntimeHandler, + sink: watchSinkStorage.getStore(), controllers: new Map(), running: false, pending: false, stopped: false, + emitted: false, lastValue: undefined, }; watches.set(watchId, watch); @@ -333,6 +375,51 @@ export type RuntimeFunctionSpec = { operation?: string; }; +class AsyncEventQueue implements AsyncIterableIterator { + private readonly values: T[] = []; + private readonly waiters: Array<(result: IteratorResult) => void> = []; + private closed = false; + + push(value: T) { + 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(): Promise> { + 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: unknown): value is Record => Boolean(value) && typeof value === "object" && !Array.isArray(value); @@ -482,6 +569,31 @@ export const protoFieldsToJs = ( ]), ); +const derivedDependencyToProto = (dependency: DerivedDependency) => + create(DerivedDependencySchema, { + kind: dependency.kind, + objectId: dependency.objectId, + fieldName: dependency.kind === "field" ? dependency.fieldName : "", + sourceField: dependency.kind === "edges" ? dependency.sourceField ?? "" : "", + }); + +const derivedEmissionToProto = (event: DerivedWatchEmission): WatchEvent => + create(WatchEventSchema, { + watchId: event.watchId, + value: jsToProtoValue(event.value ?? null), + dependencies: event.dependencies.map(derivedDependencyToProto), + error: event.error ?? "", + initial: event.initial, + }); + +const watchIdFromResult = (value: unknown) => { + if (!isRecord(value)) { + return ""; + } + const watchId = value.watch_id ?? value.watchId; + return typeof watchId === "string" ? watchId : ""; +}; + export const objectRef = (objectId: string) => objectId as ObjectRef; @@ -586,6 +698,52 @@ export const serveQuixosPackageRuntime = < }), ); + const resolveRuntimeFunction = ( + fn: InvokeRequest["function"], + ): { + spec: RuntimeFunctionSpec; + runtimeFunction: RuntimeHandler; + } => { + 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 as Record< + string, + RuntimeHandler | FieldOperation + > + )[spec.exportName]; + const runtimeFunction = + spec.operation && typeof implementationExport === "object" + ? implementationExport[spec.operation as keyof FieldOperation] + : 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, { @@ -600,45 +758,7 @@ export const serveQuixosPackageRuntime = < async invoke(request) { try { - const symbol = request.function?.symbol ?? ""; - const spec = params.functions.find( - (entry) => - entry.symbol === symbol && - (entry.operation ?? "") === (request.function?.operation ?? ""), - ); - if (!spec) { - throw new Error( - `Unknown ${params.packageName} function: ${symbol}${ - request.function?.operation - ? `.${request.function.operation}` - : "" - }`, - ); - } - const implementationExport = ( - params.implementation as Record< - string, - RuntimeHandler | FieldOperation - > - )[spec.exportName]; - const runtimeFunction = - spec.operation && typeof implementationExport === "object" - ? implementationExport[spec.operation as keyof FieldOperation] - : 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}` : "" - }`, - ); - } + const { runtimeFunction } = resolveRuntimeFunction(request.function); return create(InvokeResponseSchema, { ok: true, result: jsToProtoValue( @@ -653,6 +773,85 @@ export const serveQuixosPackageRuntime = < }); } }, + + async *watch(request: WatchRequest, context: HandlerContext) { + 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: DerivedWatchSink = { + publish(event) { + queue.push(derivedEmissionToProto(event)); + }, + }; + const result = await watchSinkStorage.run(sink, async () => + runtimeFunction( + params.createContext( + trackingCamino, + request as unknown as InvokeRequest, + ), + ), + ); + 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), + }, + }) as unknown as InvokeRequest, + ), + ); + } catch (error) { + logDerivedWatch("stop-error", { + watchId, + error: error instanceof Error ? error.message : String(error), + }); + } + } + } + }, }); }, }); diff --git a/src/quixos/refs_pb.ts b/src/quixos/refs_pb.ts index ee3f89e..f320aa2 100644 --- a/src/quixos/refs_pb.ts +++ b/src/quixos/refs_pb.ts @@ -48,3 +48,4 @@ export type FunctionRef = Message<"quixos.FunctionRef"> & { */ export const FunctionRefSchema: GenMessage = /*@__PURE__*/ messageDesc(file_quixos_refs, 0); + diff --git a/src/quixos/runtime_pb.ts b/src/quixos/runtime_pb.ts index 8ffc7fe..6574d7d 100644 --- a/src/quixos/runtime_pb.ts +++ b/src/quixos/runtime_pb.ts @@ -14,7 +14,7 @@ import type { Message } from "@bufbuild/protobuf"; * Describes the file quixos/runtime.proto. */ export const file_quixos_runtime: GenFile = /*@__PURE__*/ - fileDesc("ChRxdWl4b3MvcnVudGltZS5wcm90bxIOcXVpeG9zLnJ1bnRpbWUiMQoQSGFuZHNoYWtlUmVxdWVzdBIdChVvcmNoX3Byb3RvY29sX3ZlcnNpb24YASABKAkijgEKEUhhbmRzaGFrZVJlc3BvbnNlEhkKEXBhY2thZ2VfbmFtZXNwYWNlGAEgASgJEhQKDHBhY2thZ2VfbmFtZRgCIAEoCRIgChhydW50aW1lX3Byb3RvY29sX3ZlcnNpb24YAyABKAkSJgoJZnVuY3Rpb25zGAQgAygLMhMucXVpeG9zLkZ1bmN0aW9uUmVmItYBCg1JbnZva2VSZXF1ZXN0EhUKDWludm9jYXRpb25faWQYASABKAkSJQoIZnVuY3Rpb24YAiABKAsyEy5xdWl4b3MuRnVuY3Rpb25SZWYSEQoJb2JqZWN0X2lkGAMgASgJEjcKBWlucHV0GAQgAygLMigucXVpeG9zLnJ1bnRpbWUuSW52b2tlUmVxdWVzdC5JbnB1dEVudHJ5GjsKCklucHV0RW50cnkSCwoDa2V5GAEgASgJEhwKBXZhbHVlGAIgASgLMg0uY2FtaW5vLlZhbHVlOgI4ASJKCg5JbnZva2VSZXNwb25zZRIKCgJvaxgBIAEoCBIdCgZyZXN1bHQYAiABKAsyDS5jYW1pbm8uVmFsdWUSDQoFZXJyb3IYAyABKAkyqwEKDlBhY2thZ2VSdW50aW1lElAKCUhhbmRzaGFrZRIgLnF1aXhvcy5ydW50aW1lLkhhbmRzaGFrZVJlcXVlc3QaIS5xdWl4b3MucnVudGltZS5IYW5kc2hha2VSZXNwb25zZRJHCgZJbnZva2USHS5xdWl4b3MucnVudGltZS5JbnZva2VSZXF1ZXN0Gh4ucXVpeG9zLnJ1bnRpbWUuSW52b2tlUmVzcG9uc2ViBnByb3RvMw", [file_camino_api, file_quixos_refs]); + fileDesc("ChRxdWl4b3MvcnVudGltZS5wcm90bxIOcXVpeG9zLnJ1bnRpbWUiMQoQSGFuZHNoYWtlUmVxdWVzdBIdChVvcmNoX3Byb3RvY29sX3ZlcnNpb24YASABKAkijgEKEUhhbmRzaGFrZVJlc3BvbnNlEhkKEXBhY2thZ2VfbmFtZXNwYWNlGAEgASgJEhQKDHBhY2thZ2VfbmFtZRgCIAEoCRIgChhydW50aW1lX3Byb3RvY29sX3ZlcnNpb24YAyABKAkSJgoJZnVuY3Rpb25zGAQgAygLMhMucXVpeG9zLkZ1bmN0aW9uUmVmItYBCg1JbnZva2VSZXF1ZXN0EhUKDWludm9jYXRpb25faWQYASABKAkSJQoIZnVuY3Rpb24YAiABKAsyEy5xdWl4b3MuRnVuY3Rpb25SZWYSEQoJb2JqZWN0X2lkGAMgASgJEjcKBWlucHV0GAQgAygLMigucXVpeG9zLnJ1bnRpbWUuSW52b2tlUmVxdWVzdC5JbnB1dEVudHJ5GjsKCklucHV0RW50cnkSCwoDa2V5GAEgASgJEhwKBXZhbHVlGAIgASgLMg0uY2FtaW5vLlZhbHVlOgI4ASJKCg5JbnZva2VSZXNwb25zZRIKCgJvaxgBIAEoCBIdCgZyZXN1bHQYAiABKAsyDS5jYW1pbm8uVmFsdWUSDQoFZXJyb3IYAyABKAki1AEKDFdhdGNoUmVxdWVzdBIVCg1pbnZvY2F0aW9uX2lkGAEgASgJEiUKCGZ1bmN0aW9uGAIgASgLMhMucXVpeG9zLkZ1bmN0aW9uUmVmEhEKCW9iamVjdF9pZBgDIAEoCRI2CgVpbnB1dBgEIAMoCzInLnF1aXhvcy5ydW50aW1lLldhdGNoUmVxdWVzdC5JbnB1dEVudHJ5GjsKCklucHV0RW50cnkSCwoDa2V5GAEgASgJEhwKBXZhbHVlGAIgASgLMg0uY2FtaW5vLlZhbHVlOgI4ASJeChFEZXJpdmVkRGVwZW5kZW5jeRIMCgRraW5kGAEgASgJEhEKCW9iamVjdF9pZBgCIAEoCRISCgpmaWVsZF9uYW1lGAMgASgJEhQKDHNvdXJjZV9maWVsZBgEIAEoCSKVAQoKV2F0Y2hFdmVudBIQCgh3YXRjaF9pZBgBIAEoCRIcCgV2YWx1ZRgCIAEoCzINLmNhbWluby5WYWx1ZRI3CgxkZXBlbmRlbmNpZXMYAyADKAsyIS5xdWl4b3MucnVudGltZS5EZXJpdmVkRGVwZW5kZW5jeRINCgVlcnJvchgEIAEoCRIPCgdpbml0aWFsGAUgASgIMvABCg5QYWNrYWdlUnVudGltZRJQCglIYW5kc2hha2USIC5xdWl4b3MucnVudGltZS5IYW5kc2hha2VSZXF1ZXN0GiEucXVpeG9zLnJ1bnRpbWUuSGFuZHNoYWtlUmVzcG9uc2USRwoGSW52b2tlEh0ucXVpeG9zLnJ1bnRpbWUuSW52b2tlUmVxdWVzdBoeLnF1aXhvcy5ydW50aW1lLkludm9rZVJlc3BvbnNlEkMKBVdhdGNoEhwucXVpeG9zLnJ1bnRpbWUuV2F0Y2hSZXF1ZXN0GhoucXVpeG9zLnJ1bnRpbWUuV2F0Y2hFdmVudDABYgZwcm90bzM", [file_camino_api, file_quixos_refs]); /** * @generated from message quixos.runtime.HandshakeRequest @@ -124,6 +124,107 @@ export type InvokeResponse = Message<"quixos.runtime.InvokeResponse"> & { export const InvokeResponseSchema: GenMessage = /*@__PURE__*/ messageDesc(file_quixos_runtime, 3); +/** + * @generated from message quixos.runtime.WatchRequest + */ +export type WatchRequest = Message<"quixos.runtime.WatchRequest"> & { + /** + * @generated from field: string invocation_id = 1; + */ + invocationId: string; + + /** + * @generated from field: quixos.FunctionRef function = 2; + */ + function?: FunctionRef | undefined; + + /** + * @generated from field: string object_id = 3; + */ + objectId: string; + + /** + * @generated from field: map input = 4; + */ + input: { [key: string]: Value }; +}; + +/** + * Describes the message quixos.runtime.WatchRequest. + * Use `create(WatchRequestSchema)` to create a new message. + */ +export const WatchRequestSchema: GenMessage = /*@__PURE__*/ + messageDesc(file_quixos_runtime, 4); + +/** + * @generated from message quixos.runtime.DerivedDependency + */ +export type DerivedDependency = Message<"quixos.runtime.DerivedDependency"> & { + /** + * @generated from field: string kind = 1; + */ + kind: string; + + /** + * @generated from field: string object_id = 2; + */ + objectId: string; + + /** + * @generated from field: string field_name = 3; + */ + fieldName: string; + + /** + * @generated from field: string source_field = 4; + */ + sourceField: string; +}; + +/** + * Describes the message quixos.runtime.DerivedDependency. + * Use `create(DerivedDependencySchema)` to create a new message. + */ +export const DerivedDependencySchema: GenMessage = /*@__PURE__*/ + messageDesc(file_quixos_runtime, 5); + +/** + * @generated from message quixos.runtime.WatchEvent + */ +export type WatchEvent = Message<"quixos.runtime.WatchEvent"> & { + /** + * @generated from field: string watch_id = 1; + */ + watchId: string; + + /** + * @generated from field: camino.Value value = 2; + */ + value?: Value | undefined; + + /** + * @generated from field: repeated quixos.runtime.DerivedDependency dependencies = 3; + */ + dependencies: DerivedDependency[]; + + /** + * @generated from field: string error = 4; + */ + error: string; + + /** + * @generated from field: bool initial = 5; + */ + initial: boolean; +}; + +/** + * Describes the message quixos.runtime.WatchEvent. + * Use `create(WatchEventSchema)` to create a new message. + */ +export const WatchEventSchema: GenMessage = /*@__PURE__*/ + messageDesc(file_quixos_runtime, 6); + /** * @generated from service quixos.runtime.PackageRuntime */ @@ -144,5 +245,14 @@ export const PackageRuntime: GenService<{ input: typeof InvokeRequestSchema; output: typeof InvokeResponseSchema; }, + /** + * @generated from rpc quixos.runtime.PackageRuntime.Watch + */ + watch: { + methodKind: "server_streaming"; + input: typeof WatchRequestSchema; + output: typeof WatchEventSchema; + }, }> = /*@__PURE__*/ serviceDesc(file_quixos_runtime, 0); +