Add lightweight correctness linting and resolve formatting conflicts

2026-09-15 15:31:50 -07:00
parent 0d3c013c07
commit 0854822923
13 changed files with 274 additions and 102 deletions
@@ -1,12 +1,12 @@
 import http from "node:http";
 import { readFileSync } from "node:fs";
 import { createInvocationRegistry } from "./invocations.js";
-import { isObjectReference, referenceFromWire, referenceToWire, assertReferenceFree } from "./references.js";
+import { isObjectReference, referenceFromWire, referenceToWire, assertReferenceFree, } from "./references.js";
 export * from "./bindings.js";
 export { relationshipMap, relationshipList, relationshipSet } from "./relationships.js";
 import { AsyncLocalStorage } from "node:async_hooks";
 import { createHmac, randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
-export { createMigrationContext, migrationObjectId, serveMigration } from "./migration.js";
+export { createMigrationContext, migrationObjectId, serveMigration, } from "./migration.js";
 import { create, equals } from "@bufbuild/protobuf";
 import { Code, ConnectError, createClient } from "@connectrpc/connect";
 import { connectNodeAdapter, createConnectTransport } from "@connectrpc/connect-node";
@@ -27,13 +27,20 @@ const recordDependency = async (dependency) => {
 const bytesToBase64 = (value) => Buffer.from(value).toString("base64");
 const base64ToBytes = (value) => Buffer.from(value, "base64");
 const isRecord = (value) => Boolean(value) && typeof value === "object" && !Array.isArray(value);
-const isWrappedValue = (value) => isRecord(value) && "$quixosValue" in value &&
-    isRecord(value.$quixosValue) && value.$quixosValue.$typeName === "camino.Value";
-export const objectRef = (reference) => { referenceToWire(reference); return reference; };
+const isWrappedValue = (value) => isRecord(value) &&
+    "$quixosValue" in value &&
+    isRecord(value.$quixosValue) &&
+    value.$quixosValue.$typeName === "camino.Value";
+export const objectRef = (reference) => {
+    referenceToWire(reference);
+    return reference;
+};
 export const liveValue = (value) => ({ $quixosValue: value });
 export const jsToProtoValue = (value) => {
    if (isObjectReference(value))
-        return create(ValueSchema, { kind: { case: "refValue", value: create(RefValueSchema, { objectId: referenceToWire(value) }) } });
+        return create(ValueSchema, {
+            kind: { case: "refValue", value: create(RefValueSchema, { objectId: referenceToWire(value) }) },
+        });
    if (isWrappedValue(value))
        return value.$quixosValue;
    if (value === null || value === undefined) {
@@ -56,49 +63,65 @@ export const jsToProtoValue = (value) => {
    }
    if (isRecord(value) && "$quixosRef" in value)
        throw new Error("Raw ID wrappers are not object references");
-    if (isRecord(value) && typeof value.$quixosCrdtType === "string" &&
-        typeof value.$quixosCrdtPayload === "string") {
+    if (isRecord(value) && typeof value.$quixosCrdtType === "string" && typeof value.$quixosCrdtPayload === "string") {
        return create(ValueSchema, {
-            kind: { case: "crdtValue", value: create(CrdtValueSchema, {
+            kind: {
+                case: "crdtValue",
+                value: create(CrdtValueSchema, {
                    type: value.$quixosCrdtType,
                    encoding: typeof value.$quixosCrdtEncoding === "string" ? value.$quixosCrdtEncoding : "base64",
                    payload: base64ToBytes(value.$quixosCrdtPayload),
-                }) },
+                }),
+            },
        });
    }
    if (!isRecord(value))
        throw new Error(`Unsupported runtime value ${typeof value}`);
    return create(ValueSchema, {
-        kind: { case: "objectValue", value: create(ObjectValueSchema, {
+        kind: {
+            case: "objectValue",
+            value: create(ObjectValueSchema, {
                fields: Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, jsToProtoValue(entry)])),
-            }) },
+            }),
+        },
    });
 };
 export const protoValueToJs = (value) => {
    switch (value?.kind.case) {
        case "nullValue":
-        case undefined: return null;
+        case undefined:
+            return null;
        case "boolValue":
        case "numberValue":
        case "stringValue":
-        case "integerValue": return value.kind.value;
-        case "bytesValue": return bytesToBase64(value.kind.value);
-        case "refValue": return referenceFromWire(value.kind.value.objectId);
-        case "listValue": return value.kind.value.values.map(protoValueToJs);
-        case "objectValue": return Object.fromEntries(Object.entries(value.kind.value.fields).map(([key, entry]) => [key, protoValueToJs(entry)]));
-        case "crdtValue": return {
-            $quixosCrdtType: value.kind.value.type,
-            $quixosCrdtEncoding: value.kind.value.encoding,
-            $quixosCrdtPayload: bytesToBase64(value.kind.value.payload),
-        };
+        case "integerValue":
+            return value.kind.value;
+        case "bytesValue":
+            return bytesToBase64(value.kind.value);
+        case "refValue":
+            return referenceFromWire(value.kind.value.objectId);
+        case "listValue":
+            return value.kind.value.values.map(protoValueToJs);
+        case "objectValue":
+            return Object.fromEntries(Object.entries(value.kind.value.fields).map(([key, entry]) => [key, protoValueToJs(entry)]));
+        case "crdtValue":
+            return {
+                $quixosCrdtType: value.kind.value.type,
+                $quixosCrdtEncoding: value.kind.value.encoding,
+                $quixosCrdtPayload: bytesToBase64(value.kind.value.payload),
+            };
    }
 };
 export const protoFieldsToJs = (fields) => Object.fromEntries(Object.entries(fields).map(([key, value]) => [key, protoValueToJs(value)]));
 export class RuntimeAuthorityError extends Error {
    retryable;
-    constructor(message) { super(message); this.name = "RuntimeAuthorityError"; this.retryable = /WORKSPACE_FENCED|STALE_EPOCH/.test(message); }
+    constructor(message) {
+        super(message);
+        this.name = "RuntimeAuthorityError";
+        this.retryable = /WORKSPACE_FENCED|STALE_EPOCH/.test(message);
+    }
 }
-const targetForEdge = (edge, projectionId) => edge.firstProjectionId === projectionId ? edge.secondObjectId : edge.firstObjectId;
+const targetForEdge = (edge, projectionId) => (edge.firstProjectionId === projectionId ? edge.secondObjectId : edge.firstObjectId);
 export const createRuntimeContext = (camino, orch, request) => {
    const ports = new Map();
    for (const dependency of request.dependencies) {
@@ -130,25 +153,55 @@ export const createRuntimeContext = (camino, orch, request) => {
            case "edge": {
                const { edgeTypeId, projectionId } = dependency.binding.value;
                const dependencyObjectId = dependency.objectId || request.objectId;
-                const collectionResult = (response) => ({ revision: response.revision,
-                    entries: response.entries.map((entry) => ({ edgeId: entry.edgeId, target: referenceFromWire(entry.targetObjectId),
-                        ...(entry.key ? { key: entry.key.kind.case === "integerValue" ? BigInt(entry.key.kind.value) : protoValueToJs(entry.key) } : {}) })) });
+                const collectionResult = (response) => ({
+                    revision: response.revision,
+                    entries: response.entries.map((entry) => ({
+                        edgeId: entry.edgeId,
+                        target: referenceFromWire(entry.targetObjectId),
+                        ...(entry.key
+                            ? {
+                                key: entry.key.kind.case === "integerValue"
+                                    ? BigInt(entry.key.kind.value)
+                                    : protoValueToJs(entry.key),
+                            }
+                            : {}),
+                    })),
+                });
                const edge = {
                    edgeTypeId,
                    projectionId,
                    async collection() {
-                        await recordDependency({ kind: "edge", objectId: dependencyObjectId, attachmentId: edgeTypeId, projectionId });
+                        await recordDependency({
+                            kind: "edge",
+                            objectId: dependencyObjectId,
+                            attachmentId: edgeTypeId,
+                            projectionId,
+                        });
                        return collectionResult(await camino.readCollection({ objectId: dependencyObjectId, edgeTypeId, projectionId }));
                    },
                    async replace(entries, expectedRevision) {
-                        return collectionResult(await camino.replaceCollection({ objectId: dependencyObjectId, edgeTypeId, projectionId, expectedRevision,
+                        return collectionResult(await camino.replaceCollection({
+                            objectId: dependencyObjectId,
+                            edgeTypeId,
+                            projectionId,
+                            expectedRevision,
                            entries: entries.map((entry) => {
                                assertReferenceFree(entry.key);
-                                return { edgeId: entry.edgeId ?? "", targetObjectId: referenceToWire(entry.target), key: entry.key === undefined ? undefined : jsToProtoValue(entry.key) };
-                            }) }));
+                                return {
+                                    edgeId: entry.edgeId ?? "",
+                                    targetObjectId: referenceToWire(entry.target),
+                                    key: entry.key === undefined ? undefined : jsToProtoValue(entry.key),
+                                };
+                            }),
+                        }));
                    },
                    async resolve() {
-                        await recordDependency({ kind: "edge", objectId: dependencyObjectId, attachmentId: edgeTypeId, projectionId });
+                        await recordDependency({
+                            kind: "edge",
+                            objectId: dependencyObjectId,
+                            attachmentId: edgeTypeId,
+                            projectionId,
+                        });
                        const result = await camino.resolveEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId });
                        return result.edges.map((entry) => referenceFromWire(targetForEdge(entry, projectionId)));
                    },
@@ -265,8 +318,10 @@ const protoDependencies = (dependencies) => dependencies.map((entry) => create(D
 export const createPackageRuntimeRoutes = (config) => {
    const invocations = createInvocationRegistry();
    const headers = {};
-    const processToken = process.env.CAMINO_RUNTIME_AUTH_TOKEN ?? (process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE
-        ? readFileSync(process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE, "utf8").trim() : "");
+    const processToken = process.env.CAMINO_RUNTIME_AUTH_TOKEN ??
+        (process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE
+            ? readFileSync(process.env.CAMINO_RUNTIME_AUTH_TOKEN_FILE, "utf8").trim()
+            : "");
    if (processToken) {
        headers["x-camino-runtime-token"] = processToken;
    }
@@ -276,12 +331,14 @@ export const createPackageRuntimeRoutes = (config) => {
    const camino = createClient(CaminoService, createConnectTransport({
        baseUrl: config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310",
        httpVersion: "1.1",
-        interceptors: headers["x-camino-runtime-token"] ? [
-            (next) => async (request) => {
-                request.header.set("x-camino-runtime-token", headers["x-camino-runtime-token"]);
-                return await next(request);
-            },
-        ] : [],
+        interceptors: headers["x-camino-runtime-token"]
+            ? [
+                (next) => async (request) => {
+                    request.header.set("x-camino-runtime-token", headers["x-camino-runtime-token"]);
+                    return await next(request);
+                },
+            ]
+            : [],
    }));
    const orch = createClient(OrchestratorRuntime, createConnectTransport({
        baseUrl: config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311",
@@ -298,16 +355,23 @@ export const createPackageRuntimeRoutes = (config) => {
    };
    const clientsFor = (request) => {
        const context = request.context;
-        if (process.env.QUIXOS_RUNTIME_INSTANCE_ID && (!context?.grant || context.instanceId !== process.env.QUIXOS_RUNTIME_INSTANCE_ID || !context.workspaceEpoch)) {
+        if (process.env.QUIXOS_RUNTIME_INSTANCE_ID &&
+            (!context?.grant || context.instanceId !== process.env.QUIXOS_RUNTIME_INSTANCE_ID || !context.workspaceEpoch)) {
            throw new ConnectError("Managed invocation requires an exact instance and epoch grant", Code.Unauthenticated);
        }
        if (!context?.grant)
            return { camino, orch };
-        const transport = (url) => createConnectTransport({ baseUrl: url, httpVersion: "1.1", interceptors: [(next) => async (call) => {
+        const transport = (url) => createConnectTransport({
+            baseUrl: url,
+            httpVersion: "1.1",
+            interceptors: [
+                (next) => async (call) => {
                    call.header.set("x-quixos-invocation-grant", context.grant);
                    call.header.set("x-camino-runtime-token", processToken);
                    return next(call);
-                }] });
+                },
+            ],
+        });
        return {
            camino: createClient(CaminoService, transport(config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310")),
            orch: createClient(OrchestratorRuntime, transport(config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311")),
@@ -315,9 +379,12 @@ export const createPackageRuntimeRoutes = (config) => {
    };
    const runtimeControl = async (operation, input) => {
        const response = await fetch(`${config.caminoUrl ?? process.env.CAMINO_URL ?? "http://127.0.0.1:7310"}/__runtime/${operation}`, {
-            method: "POST", headers: { "content-type": "application/json", "x-camino-runtime-token": processToken }, body: JSON.stringify(input), signal: AbortSignal.timeout(10_000),
+            method: "POST",
+            headers: { "content-type": "application/json", "x-camino-runtime-token": processToken },
+            body: JSON.stringify(input),
+            signal: AbortSignal.timeout(10_000),
        });
-        const value = await response.json();
+        const value = (await response.json());
        if (!response.ok)
            throw new RuntimeAuthorityError(value.error ?? "Runtime authority request failed");
        return value;
@@ -327,8 +394,13 @@ export const createPackageRuntimeRoutes = (config) => {
            return;
        runtimeContext.openSession = async () => {
            const ownerId = request.context.ownerConformanceId;
-            const registration = { grant: request.context.grant, objectId: request.objectId, ownerId,
-                sessionId: `session:${randomBytes(16).toString("hex")}`, token: randomBytes(32).toString("base64url") };
+            const registration = {
+                grant: request.context.grant,
+                objectId: request.objectId,
+                ownerId,
+                sessionId: `session:${randomBytes(16).toString("hex")}`,
+                token: randomBytes(32).toString("base64url"),
+            };
            const register = () => runtimeControl("register-session", registration);
            const registered = await register().catch((error) => {
                // Retry a transport/lost-response failure with exactly the same identity.
@@ -347,7 +419,10 @@ export const createPackageRuntimeRoutes = (config) => {
                    // by the caller without replaying a side-effecting callback.
                    const grant = await runtimeControl("acquire-session", registered);
                    const execution = invocations.begin(grant.invocationId);
-                    const sessionRequest = { ...request, context: { grant: grant.grant, instanceId: grant.instanceId, workspaceEpoch: grant.epoch } };
+                    const sessionRequest = {
+                        ...request,
+                        context: { grant: grant.grant, instanceId: grant.instanceId, workspaceEpoch: grant.epoch },
+                    };
                    const clients = clientsFor(sessionRequest);
                    const context = createRuntimeContext(clients.camino, clients.orch, sessionRequest);
                    context.signal = execution.signal;
@@ -359,7 +434,10 @@ export const createPackageRuntimeRoutes = (config) => {
                        await runtimeControl("complete-invocation", { invocationId: grant.invocationId }).catch((error) => console.error("Session completion will be reconciled by the host", error));
                    }
                },
-                async close() { await runtimeControl("close-session", registered); closed = true; },
+                async close() {
+                    await runtimeControl("close-session", registered);
+                    closed = true;
+                },
            };
        };
    };
@@ -370,9 +448,15 @@ export const createPackageRuntimeRoutes = (config) => {
            exportIds: Object.keys(config.exports),
            capabilities: ["invocation-completion-v1", "instance-authentication-v1", "epoch-grants-v1"],
            instanceId: process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "",
-            authenticationProof: request.nonce && processToken ? createHmac("sha256", processToken)
-                .update(JSON.stringify([request.nonce, process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "", config.packageRevisionId]))
-                .digest("hex") : "",
+            authenticationProof: request.nonce && processToken
+                ? createHmac("sha256", processToken)
+                    .update(JSON.stringify([
+                    request.nonce,
+                    process.env.QUIXOS_RUNTIME_INSTANCE_ID ?? "",
+                    config.packageRevisionId,
+                ]))
+                    .digest("hex")
+                : "",
        }),
        getInvocationStatus: (request, context) => {
            authenticateInstance(context.requestHeader);
@@ -437,7 +521,12 @@ export const createPackageRuntimeRoutes = (config) => {
                        return await pending;
                    const establish = (async () => {
                        const controller = new AbortController();
-                        const stream = camino.watchObject({ objectId: dependency.objectId, includeSnapshot: true, attachmentIds: request.context?.grant ? [dependency.attachmentId] : [] }, { signal: controller.signal })[Symbol.asyncIterator]();
+                        const stream = camino
+                            .watchObject({
+                            objectId: dependency.objectId,
+                            includeSnapshot: true,
+                            attachmentIds: request.context?.grant ? [dependency.attachmentId] : [],
+                        }, { signal: controller.signal })[Symbol.asyncIterator]();
                        try {
                            // Camino subscribes before producing the snapshot, so once this
                            // resolves the following state/edge read cannot race the stream.
@@ -507,10 +596,7 @@ export const createPackageRuntimeRoutes = (config) => {
                            await abort;
                            break;
                        }
-                        const outcome = await Promise.race([
-                            ...[...subscriptions.values()].map((entry) => entry.next),
-                            abort,
-                        ]);
+                        const outcome = await Promise.race([...[...subscriptions.values()].map((entry) => entry.next), abort]);
                        if (outcome === "abort")
                            break;
                        const subscription = subscriptions.get(outcome.key);