import http from "node:http";
import { AsyncLocalStorage } from "node:async_hooks";
import { randomUUID } from "node:crypto";
import { create, equals } from "@bufbuild/protobuf";
import { Code, ConnectError, createClient } from "@connectrpc/connect";
import { connectNodeAdapter, createConnectTransport } from "@connectrpc/connect-node";
import { CaminoService, CrdtValueSchema, ListValueSchema, NullValueSchema, ObjectValueSchema, RefValueSchema, ValueSchema, } from "./camino/api_pb.js";
import { OrchestratorRuntime } from "./quixos/orch_pb.js";
import { DerivedDependencySchema, HandshakeResponseSchema, InvokeResponseSchema, PackageRuntime, WatchEventSchema, } from "./quixos/runtime_pb.js";
import { CapabilityRefSchema } from "./quixos/refs_pb.js";
const dependencyKey = (dependency) => `${dependency.kind}:${dependency.objectId}:${dependency.attachmentId}:${dependency.kind === "edge" ? dependency.projectionId : ""}`;
const dependencyScope = new AsyncLocalStorage();
const recordDependency = async (dependency) => {
    const scope = dependencyScope.getStore();
    if (!scope)
        return;
    const key = `${dependency.kind}:${dependency.objectId}:${dependency.attachmentId}:${dependency.kind === "edge" ? dependency.projectionId : ""}`;
    scope.dependencies.set(key, dependency);
    await scope.observe?.(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 = (objectId) => ({ $quixosRef: objectId });
export const liveValue = (value) => ({ $quixosValue: value });
export const jsToProtoValue = (value) => {
    if (isWrappedValue(value))
        return value.$quixosValue;
    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: String(value) } });
    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) && typeof value.$quixosRef === "string") {
        return create(ValueSchema, {
            kind: { case: "refValue", value: create(RefValueSchema, { objectId: value.$quixosRef }) },
        });
    }
    if (isRecord(value) && typeof value.$quixosCrdtType === "string" &&
        typeof value.$quixosCrdtPayload === "string") {
        return create(ValueSchema, {
            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, {
                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 "boolValue":
        case "numberValue":
        case "stringValue":
        case "integerValue": return value.kind.value;
        case "bytesValue": return bytesToBase64(value.kind.value);
        case "refValue": return 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)]));
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) {
        switch (dependency.binding.case) {
            case "stateSlotId": {
                const slotId = dependency.binding.value;
                const dependencyObjectId = dependency.objectId || request.objectId;
                const state = {
                    slotId,
                    async get() {
                        await recordDependency({ kind: "state", objectId: dependencyObjectId, attachmentId: slotId });
                        return protoValueToJs((await camino.readState({ objectId: dependencyObjectId, slotId })).value);
                    },
                    async live() {
                        await recordDependency({ kind: "state", objectId: dependencyObjectId, attachmentId: slotId });
                        const value = await camino.readState({ objectId: dependencyObjectId, slotId });
                        if (!value.value)
                            throw new Error(`State ${slotId} returned no value`);
                        return liveValue(value.value);
                    },
                    async set(value) {
                        await camino.writeState({ objectId: dependencyObjectId, slotId, value: jsToProtoValue(value) });
                    },
                };
                ports.set(dependency.portId, state);
                break;
            }
            case "edge": {
                const { edgeTypeId, projectionId } = dependency.binding.value;
                const dependencyObjectId = dependency.objectId || request.objectId;
                const edge = {
                    edgeTypeId,
                    projectionId,
                    async resolve() {
                        await recordDependency({ kind: "edge", objectId: dependencyObjectId, attachmentId: edgeTypeId, projectionId });
                        const result = await camino.resolveEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId });
                        return result.edges.map((entry) => targetForEdge(entry, projectionId));
                    },
                    async connect(targetObjectId) {
                        await camino.connectEdge({ objectId: dependencyObjectId, edgeTypeId, projectionId, targetObjectId });
                    },
                };
                ports.set(dependency.portId, edge);
                break;
            }
            case "receiverInterfaceRevisionId": {
                const interfaceRevisionId = dependency.binding.value;
                const capability = {
                    interfaceRevisionId,
                    async invoke(operationId, input = {}) {
                        const response = await orch.invokeCapability({
                            capability: create(CapabilityRefSchema, { interfaceRevisionId, operationId }),
                            objectId: request.objectId,
                            input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
                        });
                        if (!response.ok)
                            throw new Error(response.error || `Capability ${operationId} failed`);
                        return protoValueToJs(response.result);
                    },
                };
                ports.set(dependency.portId, capability);
                break;
            }
            case "constructorAtomId": {
                const atomId = dependency.binding.value;
                const constructor = {
                    atomId,
                    async construct(input = {}) {
                        const response = await orch.constructObject({
                            atomId,
                            input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
                        });
                        if (!response.object)
                            throw new Error(`Constructor ${atomId} returned no object`);
                        return response.object.id;
                    },
                };
                ports.set(dependency.portId, constructor);
            }
        }
    }
    const requirePort = (portId, kind) => {
        const port = ports.get(portId);
        if (!port || !(kind in port))
            throw new Error(`Missing ${kind} dependency port ${portId}`);
        return port;
    };
    return {
        objectId: request.objectId,
        input: protoFieldsToJs(request.input),
        inputProto: request.input,
        ports,
        state: (portId) => requirePort(portId, "slotId"),
        edge: (portId) => requirePort(portId, "edgeTypeId"),
        interface: (portId) => requirePort(portId, "interfaceRevisionId"),
        constructor: (portId) => requirePort(portId, "atomId"),
    };
};
export const derived = (get) => ({ kind: "derived", get });
const isDerived = (handler) => typeof handler === "object" && handler.kind === "derived";
const evaluate = async (handler, context, observe) => {
    const dependencies = new Map();
    const result = await dependencyScope.run({ dependencies, observe }, () => isDerived(handler) ? handler.get(context) : handler(context));
    return { value: jsToProtoValue(result), dependencies: [...dependencies.values()] };
};
const protoDependencies = (dependencies) => dependencies.map((entry) => create(DerivedDependencySchema, {
    kind: entry.kind,
    objectId: entry.objectId,
    attachmentId: entry.attachmentId,
    projectionId: entry.kind === "edge" ? entry.projectionId : "",
}));
export const createPackageRuntimeRoutes = (config) => {
    const headers = {};
    if (process.env.CAMINO_RUNTIME_AUTH_TOKEN) {
        headers["x-camino-runtime-token"] = process.env.CAMINO_RUNTIME_AUTH_TOKEN;
    }
    else if (process.env.CAMINO_RUNTIME_AUTH_REQUIRED === "1") {
        throw new Error("CAMINO_RUNTIME_AUTH_TOKEN is required");
    }
    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);
            },
        ] : [],
    }));
    const orch = createClient(OrchestratorRuntime, createConnectTransport({
        baseUrl: config.orchUrl ?? process.env.QUIXOS_ORCH_URL ?? "http://127.0.0.1:7311",
        httpVersion: "1.1",
    }));
    return (router) => router.service(PackageRuntime, {
        handshake: () => create(HandshakeResponseSchema, {
            packageRevisionId: config.packageRevisionId,
            runtimeProtocolVersion: "quixos-capabilities-v1",
            exportIds: Object.keys(config.exports),
        }),
        invoke: async (request) => {
            const exportId = request.export?.exportId;
            const handler = exportId ? config.exports[exportId] : undefined;
            if (!handler)
                throw new ConnectError(`Unknown export ${exportId ?? ""}`, Code.NotFound);
            try {
                const result = await evaluate(handler, createRuntimeContext(camino, orch, request));
                return create(InvokeResponseSchema, { ok: true, result: result.value });
            }
            catch (error) {
                return create(InvokeResponseSchema, {
                    ok: false,
                    error: error instanceof Error ? error.message : String(error),
                });
            }
        },
        watch: async function* (request, context) {
            const exportId = request.export?.exportId;
            const handler = exportId ? config.exports[exportId] : undefined;
            if (!handler || !isDerived(handler)) {
                throw new ConnectError(`Export ${exportId ?? ""} is not derived`, Code.FailedPrecondition);
            }
            const watchId = `watch:${randomUUID()}`;
            const runtimeContext = createRuntimeContext(camino, orch, request);
            const subscriptions = new Map();
            const establishing = new Map();
            const ensureSubscription = async (dependency) => {
                const key = dependencyKey(dependency);
                if (subscriptions.has(key))
                    return;
                const pending = establishing.get(key);
                if (pending)
                    return await pending;
                const establish = (async () => {
                    const controller = new AbortController();
                    const stream = camino.watchObject({ objectId: dependency.objectId, includeSnapshot: true }, { 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.
                        const snapshot = await stream.next();
                        if (snapshot.done)
                            throw new Error(`Dependency stream ${key} ended during setup`);
                        const waitNext = () => stream.next().then((result) => ({ key, done: Boolean(result.done) }), (error) => ({ key, done: true, error }));
                        const subscription = {
                            dependency,
                            controller,
                            waitNext,
                            next: Promise.resolve({ key, done: false }),
                        };
                        subscription.next = waitNext();
                        subscriptions.set(key, subscription);
                    }
                    catch (error) {
                        controller.abort();
                        throw error;
                    }
                })();
                establishing.set(key, establish);
                try {
                    await establish;
                }
                finally {
                    establishing.delete(key);
                }
            };
            const abortAll = () => {
                for (const subscription of subscriptions.values()) {
                    subscription.controller.abort();
                }
            };
            context.signal.addEventListener("abort", abortAll, { once: true });
            let current = await evaluate(handler, runtimeContext, ensureSubscription);
            yield create(WatchEventSchema, {
                watchId,
                value: current.value,
                dependencies: protoDependencies(current.dependencies),
                initial: true,
            });
            const abort = new Promise((resolve) => {
                if (context.signal.aborted)
                    resolve("abort");
                else
                    context.signal.addEventListener("abort", () => resolve("abort"), { once: true });
            });
            try {
                while (!context.signal.aborted) {
                    if (subscriptions.size === 0) {
                        await abort;
                        break;
                    }
                    const outcome = await Promise.race([
                        ...[...subscriptions.values()].map((entry) => entry.next),
                        abort,
                    ]);
                    if (outcome === "abort")
                        break;
                    const subscription = subscriptions.get(outcome.key);
                    if (!subscription)
                        continue;
                    if (outcome.error)
                        throw outcome.error;
                    if (outcome.done)
                        throw new Error(`Dependency stream ${outcome.key} ended unexpectedly`);
                    subscription.next = subscription.waitNext();
                    const updated = await evaluate(handler, runtimeContext, ensureSubscription);
                    const active = new Set(updated.dependencies.map(dependencyKey));
                    for (const [key, entry] of subscriptions) {
                        if (!active.has(key)) {
                            entry.controller.abort();
                            subscriptions.delete(key);
                        }
                    }
                    if (!equals(ValueSchema, current.value, updated.value)) {
                        yield create(WatchEventSchema, {
                            watchId,
                            value: updated.value,
                            dependencies: protoDependencies(updated.dependencies),
                        });
                    }
                    current = updated;
                }
            }
            finally {
                context.signal.removeEventListener("abort", abortAll);
                abortAll();
            }
        },
    });
};
export const servePackageRuntime = (config) => {
    const host = process.env.QUIXOS_RUNTIME_HOST ?? "127.0.0.1";
    const port = Number(process.env.QUIXOS_RUNTIME_PORT ?? "0");
    const handler = connectNodeAdapter({ routes: createPackageRuntimeRoutes(config) });
    const server = http.createServer((request, response) => void handler(request, response));
    server.listen(port, host, () => console.log(`${config.packageRevisionId} listening on ${host}:${port}`));
    const shutdown = () => {
        server.close(() => process.exit(0));
        server.closeAllConnections();
    };
    process.on("SIGTERM", shutdown);
    process.on("SIGINT", shutdown);
    return server;
};