Use textproto package descriptors
This commit is contained in:
Timothy J. Aveni
+309
@@ -0,0 +1,309 @@
|
||||
import http from "node:http";
|
||||
import { create } from "@bufbuild/protobuf";
|
||||
import { createClient, type Client } from "@connectrpc/connect";
|
||||
import {
|
||||
connectNodeAdapter,
|
||||
createConnectTransport,
|
||||
} from "@connectrpc/connect-node";
|
||||
import {
|
||||
CaminoService,
|
||||
CrdtValueSchema,
|
||||
ListValueSchema,
|
||||
NullValueSchema,
|
||||
ObjectValueSchema,
|
||||
RefValueSchema,
|
||||
ValueSchema,
|
||||
type Value,
|
||||
} from "./camino/api_pb.js";
|
||||
import {
|
||||
HandshakeResponseSchema,
|
||||
InvokeResponseSchema,
|
||||
PackageRuntime,
|
||||
type InvokeRequest,
|
||||
} from "./quixos/runtime_pb.js";
|
||||
import { FunctionRefSchema } from "./quixos/refs_pb.js";
|
||||
|
||||
declare const objectRefBrand: unique symbol;
|
||||
|
||||
export type ObjectRef<ClassId extends string> = string & {
|
||||
readonly [objectRefBrand]: ClassId;
|
||||
};
|
||||
|
||||
export type CaminoClient = Client<typeof CaminoService>;
|
||||
|
||||
export type { InvokeRequest, Value };
|
||||
|
||||
export type Field<T> = {
|
||||
set(value: T): Promise<void>;
|
||||
};
|
||||
|
||||
export type RuntimeHandler<Context> = (
|
||||
context: Context,
|
||||
) => Promise<unknown> | unknown;
|
||||
|
||||
export type RuntimeFunctionSpec<ExportName extends string = string> = {
|
||||
exportName: ExportName;
|
||||
symbol: string;
|
||||
};
|
||||
|
||||
const isRecord = (value: unknown): value is Record<string, unknown> =>
|
||||
Boolean(value) && typeof value === "object" && !Array.isArray(value);
|
||||
|
||||
const bytesToBase64 = (value: Uint8Array) =>
|
||||
Buffer.from(value).toString("base64");
|
||||
|
||||
const base64ToBytes = (value: string) => Buffer.from(value, "base64");
|
||||
|
||||
export const jsToProtoValue = (value: unknown): Value => {
|
||||
if (value === null || value === undefined) {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "nullValue",
|
||||
value: create(NullValueSchema, {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (typeof value === "boolean") {
|
||||
return create(ValueSchema, {
|
||||
kind: { case: "boolValue", value },
|
||||
});
|
||||
}
|
||||
if (typeof value === "number") {
|
||||
return create(ValueSchema, {
|
||||
kind: { case: "numberValue", value },
|
||||
});
|
||||
}
|
||||
if (typeof value === "bigint") {
|
||||
return create(ValueSchema, {
|
||||
kind: { case: "integerValue", value: value.toString() },
|
||||
});
|
||||
}
|
||||
if (typeof value === "string") {
|
||||
return create(ValueSchema, {
|
||||
kind: { case: "stringValue", value },
|
||||
});
|
||||
}
|
||||
if (value instanceof Uint8Array) {
|
||||
return create(ValueSchema, {
|
||||
kind: { case: "bytesValue", value },
|
||||
});
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "listValue",
|
||||
value: create(ListValueSchema, {
|
||||
values: value.map(jsToProtoValue),
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (isRecord(value)) {
|
||||
if (typeof value.$caminoRef === "string") {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "refValue",
|
||||
value: create(RefValueSchema, {
|
||||
objectId: value.$caminoRef,
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (
|
||||
typeof value.$caminoBytes === "string" &&
|
||||
Object.keys(value).every((key) => key === "$caminoBytes")
|
||||
) {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "bytesValue",
|
||||
value: base64ToBytes(value.$caminoBytes),
|
||||
},
|
||||
});
|
||||
}
|
||||
if (
|
||||
typeof value.$caminoCrdtType === "string" &&
|
||||
typeof value.$caminoCrdtPayload === "string"
|
||||
) {
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "crdtValue",
|
||||
value: create(CrdtValueSchema, {
|
||||
type: value.$caminoCrdtType,
|
||||
encoding:
|
||||
typeof value.$caminoCrdtEncoding === "string"
|
||||
? value.$caminoCrdtEncoding
|
||||
: "base64",
|
||||
payload: base64ToBytes(value.$caminoCrdtPayload),
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
return create(ValueSchema, {
|
||||
kind: {
|
||||
case: "objectValue",
|
||||
value: create(ObjectValueSchema, {
|
||||
fields: jsObjectToProtoFields(value),
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
throw new Error(`Unsupported Camino RPC value: ${typeof value}`);
|
||||
};
|
||||
|
||||
export const protoValueToJs = (value: Value | undefined): unknown => {
|
||||
switch (value?.kind.case) {
|
||||
case "nullValue":
|
||||
case undefined:
|
||||
return null;
|
||||
case "boolValue":
|
||||
case "numberValue":
|
||||
case "stringValue":
|
||||
return value.kind.value;
|
||||
case "integerValue":
|
||||
return value.kind.value;
|
||||
case "bytesValue":
|
||||
return bytesToBase64(value.kind.value);
|
||||
case "listValue":
|
||||
return value.kind.value.values.map(protoValueToJs);
|
||||
case "objectValue":
|
||||
return protoFieldsToJs(value.kind.value.fields);
|
||||
case "refValue":
|
||||
return value.kind.value.objectId;
|
||||
case "crdtValue":
|
||||
return {
|
||||
$caminoCrdtType: value.kind.value.type,
|
||||
$caminoCrdtEncoding: value.kind.value.encoding,
|
||||
$caminoCrdtPayload: bytesToBase64(value.kind.value.payload),
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
export const jsObjectToProtoFields = (value: Record<string, unknown>) =>
|
||||
Object.fromEntries(
|
||||
Object.entries(value)
|
||||
.filter(([, fieldValue]) => fieldValue !== undefined)
|
||||
.map(([key, fieldValue]) => [key, jsToProtoValue(fieldValue)]),
|
||||
);
|
||||
|
||||
export const protoFieldsToJs = (
|
||||
fields: Record<string, Value> | undefined,
|
||||
): Record<string, unknown> =>
|
||||
Object.fromEntries(
|
||||
Object.entries(fields ?? {}).map(([key, value]) => [
|
||||
key,
|
||||
protoValueToJs(value),
|
||||
]),
|
||||
);
|
||||
|
||||
export const objectRef = <ClassId extends string>(objectId: string) =>
|
||||
objectId as ObjectRef<ClassId>;
|
||||
|
||||
export const createField = <T>(
|
||||
camino: CaminoClient,
|
||||
objectId: string,
|
||||
fieldName: string,
|
||||
): Field<T> => ({
|
||||
async set(value) {
|
||||
await camino.setField({
|
||||
objectId,
|
||||
fieldName,
|
||||
value: jsToProtoValue(value),
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
export const serveQuixosPackageRuntime = <
|
||||
Context,
|
||||
Spec extends readonly RuntimeFunctionSpec[],
|
||||
>(params: {
|
||||
packageNamespace: string;
|
||||
packageName: string;
|
||||
runtimeProtocolVersion: string;
|
||||
functions: Spec;
|
||||
implementation: {
|
||||
[Name in Spec[number]["exportName"]]: RuntimeHandler<Context>;
|
||||
};
|
||||
createContext: (camino: CaminoClient, request: InvokeRequest) => Context;
|
||||
}) => {
|
||||
const host = process.env.QUIXOS_RUNTIME_HOST ?? "127.0.0.1";
|
||||
const port = Number(process.env.QUIXOS_RUNTIME_PORT ?? "0");
|
||||
const caminoUrl = process.env.CAMINO_URL ?? "http://127.0.0.1:7310";
|
||||
const camino = createClient(
|
||||
CaminoService,
|
||||
createConnectTransport({ baseUrl: caminoUrl, httpVersion: "1.1" }),
|
||||
);
|
||||
|
||||
if (!Number.isInteger(port) || port <= 0) {
|
||||
throw new Error("QUIXOS_RUNTIME_PORT must be a positive integer");
|
||||
}
|
||||
|
||||
const functionRefs = params.functions.map((entry) =>
|
||||
create(FunctionRefSchema, {
|
||||
packageNamespace: params.packageNamespace,
|
||||
packageName: params.packageName,
|
||||
symbol: entry.symbol,
|
||||
}),
|
||||
);
|
||||
|
||||
const handler = connectNodeAdapter({
|
||||
routes: (router) => {
|
||||
router.service(PackageRuntime, {
|
||||
async handshake() {
|
||||
return create(HandshakeResponseSchema, {
|
||||
packageNamespace: params.packageNamespace,
|
||||
packageName: params.packageName,
|
||||
runtimeProtocolVersion: params.runtimeProtocolVersion,
|
||||
functions: functionRefs,
|
||||
});
|
||||
},
|
||||
|
||||
async invoke(request) {
|
||||
try {
|
||||
const symbol = request.function?.symbol ?? "";
|
||||
const spec = params.functions.find(
|
||||
(entry) => entry.symbol === symbol,
|
||||
);
|
||||
if (!spec) {
|
||||
throw new Error(
|
||||
`Unknown ${params.packageName} function: ${symbol}`,
|
||||
);
|
||||
}
|
||||
const runtimeFunction = (
|
||||
params.implementation as Record<string, RuntimeHandler<Context>>
|
||||
)[spec.exportName];
|
||||
if (!runtimeFunction) {
|
||||
throw new Error(
|
||||
`Missing ${params.packageName} implementation export: ${spec.exportName}`,
|
||||
);
|
||||
}
|
||||
return create(InvokeResponseSchema, {
|
||||
ok: true,
|
||||
result: jsToProtoValue(
|
||||
await runtimeFunction(params.createContext(camino, request)),
|
||||
),
|
||||
});
|
||||
} catch (error) {
|
||||
return create(InvokeResponseSchema, {
|
||||
ok: false,
|
||||
result: jsToProtoValue(null),
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
const server = http.createServer((request, response) => {
|
||||
void handler(request, response);
|
||||
});
|
||||
const close = () => server.close(() => process.exit(0));
|
||||
|
||||
server.listen(port, host, () => {
|
||||
process.stderr.write(
|
||||
`${params.packageName} listening on ${host}:${port}\n`,
|
||||
);
|
||||
});
|
||||
process.on("SIGTERM", close);
|
||||
process.on("SIGINT", close);
|
||||
};
|
||||
Reference in New Issue
Block a user