import http from "node:http";
import { AsyncLocalStorage } from "node:async_hooks";
import { randomUUID } from "node:crypto";
import { create } from "@bufbuild/protobuf";
import { createClient, } from "@connectrpc/connect";
import * as Automerge from "@automerge/automerge";
import { createAutomergeFieldValue, encodeAutomergeFieldWrite, loadAutomergeFieldDoc, } from "@quixos/camino-datatypes/crdt-automerge";
import { connectNodeAdapter, createConnectTransport, } from "@connectrpc/connect-node";
import { CaminoService, CrdtValueSchema, FieldValueSourceSchema, ListValueSchema, NullValueSchema, ObjectValueSchema, RefValueSchema, ValueSchema, ValueSourceSchema, } from "./camino/api_pb.js";
import { DerivedDependencySchema, HandshakeResponseSchema, InvokeResponseSchema, PackageRuntime, WatchEventSchema, WatchRequestSchema, } from "./quixos/runtime_pb.js";
import { FunctionRefSchema } from "./quixos/refs_pb.js";
const dependencyKey = (dependency) => {
    switch (dependency.kind) {
        case "object":
            return `object:${dependency.objectId}`;
        case "field":
            return `field:${dependency.objectId}:${dependency.fieldName}`;
        case "edges":
            return `edges:${dependency.objectId}:${dependency.sourceField ?? ""}`;
    }
};
const createDependencyTracker = () => {
    const dependencies = new Map();
    return {
        record(dependency) {
            dependencies.set(dependencyKey(dependency), dependency);
        },
        dependencies() {
            return [...dependencies.values()];
        },
    };
};
const dependencyStorage = new AsyncLocalStorage();
export const recordDerivedDependency = (dependency) => {
    dependencyStorage.getStore()?.record(dependency);
};
export const getActiveDerivedDependencies = () => dependencyStorage.getStore()?.dependencies() ?? [];
export const runWithDerivedDependencyTracking = async (fn) => {
    const tracker = createDependencyTracker();
    const value = await dependencyStorage.run(tracker, fn);
    return {
        value,
        dependencies: tracker.dependencies(),
    };
};
const logDerivedDependencies = (dependencies) => {
    if (process.env.QUIXOS_DERIVED_DEPS_LOG !== "1") {
        return;
    }
    process.stderr.write(`[derived-deps] ${JSON.stringify({ dependencies })}\n`);
};
const logDerivedWatch = (event, detail) => {
    if (process.env.QUIXOS_DERIVED_WATCH_LOG !== "1") {
        return;
    }
    process.stderr.write(`[derived-watch] ${JSON.stringify({ event, ...detail })}\n`);
};
const watchSinkStorage = new AsyncLocalStorage();
const dependencyObjectIds = (dependencies) => new Set(dependencies.map((dependency) => dependency.objectId));
const stopDerivedWatch = (watch) => {
    if (watch.stopped) {
        return;
    }
    watch.stopped = true;
    for (const controller of watch.controllers.values()) {
        controller.abort();
    }
    watch.controllers.clear();
    logDerivedWatch("stopped", { watchId: watch.watchId });
};
export const liveFieldValue = (params) => ({
    $caminoValue: params.value,
    $caminoSource: {
        field: {
            objectId: params.objectId,
            fieldName: params.fieldName,
            ...(params.fieldType ? { fieldType: params.fieldType } : {}),
            ...(params.fieldStorage ? { fieldStorage: params.fieldStorage } : {}),
            ...(params.conflictStrategy
                ? { conflictStrategy: params.conflictStrategy }
                : {}),
            ...(params.revision !== undefined ? { revision: params.revision } : {}),
        },
    },
});
export const fieldOps = (operation) => operation;
export const derivedWithDeps = (operation) => {
    const watches = new Map();
    const rerun = async (watch) => {
        if (watch.stopped) {
            return;
        }
        if (watch.running) {
            watch.pending = true;
            return;
        }
        watch.running = true;
        try {
            do {
                watch.pending = false;
                const { value, dependencies } = await runWithDerivedDependencyTracking(() => watch.get(watch.context));
                watch.lastValue = value;
                logDerivedDependencies(dependencies);
                logDerivedWatch("evaluated", {
                    watchId: watch.watchId,
                    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: message,
            });
            watch.sink?.publish({
                watchId: watch.watchId,
                dependencies: [],
                error: message,
                initial: !watch.emitted,
            });
            watch.emitted = true;
        }
        finally {
            watch.running = false;
        }
    };
    const scheduleRerun = (watch, reason) => {
        if (watch.stopped) {
            return;
        }
        logDerivedWatch("scheduled", { watchId: watch.watchId, reason });
        void rerun(watch);
    };
    const startObjectWatch = (watch, objectId) => {
        const controller = new AbortController();
        watch.controllers.set(objectId, controller);
        logDerivedWatch("subscribed", { watchId: watch.watchId, objectId });
        void (async () => {
            try {
                for await (const event of watch.context.camino.watchObject({
                    objectId,
                    includeSnapshot: false,
                }, { signal: controller.signal })) {
                    if (controller.signal.aborted || watch.stopped) {
                        break;
                    }
                    scheduleRerun(watch, event.op?.id ? `op:${event.op.id}` : `object:${objectId}`);
                }
            }
            catch (error) {
                if (!controller.signal.aborted && !watch.stopped) {
                    logDerivedWatch("watch-error", {
                        watchId: watch.watchId,
                        objectId,
                        error: error instanceof Error ? error.message : String(error),
                    });
                }
            }
            finally {
                if (watch.controllers.get(objectId) === controller) {
                    watch.controllers.delete(objectId);
                    logDerivedWatch("unsubscribed", { watchId: watch.watchId, objectId });
                }
            }
        })();
    };
    const reconcileSubscriptions = (watch, dependencies) => {
        const objectIds = dependencyObjectIds(dependencies);
        for (const [objectId, controller] of watch.controllers.entries()) {
            if (!objectIds.has(objectId)) {
                controller.abort();
                watch.controllers.delete(objectId);
                logDerivedWatch("unsubscribed", { watchId: watch.watchId, objectId });
            }
        }
        for (const objectId of objectIds) {
            if (!watch.controllers.has(objectId)) {
                startObjectWatch(watch, objectId);
            }
        }
    };
    return fieldOps({
        async get(context) {
            const { value, dependencies } = await runWithDerivedDependencyTracking(() => operation.get(context));
            logDerivedDependencies(dependencies);
            return value;
        },
        async watchStart(context) {
            const watchId = `watch:${randomUUID()}`;
            const watch = {
                watchId,
                context: context,
                get: operation.get,
                sink: watchSinkStorage.getStore(),
                controllers: new Map(),
                running: false,
                pending: false,
                stopped: false,
                emitted: false,
                lastValue: undefined,
            };
            watches.set(watchId, watch);
            logDerivedWatch("started", { watchId });
            await rerun(watch);
            return {
                watch_id: watchId,
            };
        },
        watchStop(context) {
            const input = protoFieldsToJs(context.request.input);
            const watchId = typeof input.watch_id === "string"
                ? input.watch_id
                : typeof input.watchId === "string"
                    ? input.watchId
                    : "";
            const watch = watches.get(watchId);
            if (!watch) {
                throw new Error(`Unknown derived watch: ${watchId}`);
            }
            stopDerivedWatch(watch);
            watches.delete(watchId);
            return {};
        },
    });
};
class AsyncEventQueue {
    values = [];
    waiters = [];
    closed = false;
    push(value) {
        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() {
        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) => Boolean(value) && typeof value === "object" && !Array.isArray(value);
const bytesToBase64 = (value) => Buffer.from(value).toString("base64");
const base64ToBytes = (value) => Buffer.from(value, "base64");
const sourceToProto = (value) => {
    if (!isRecord(value) || !isRecord(value.field)) {
        return undefined;
    }
    const field = value.field;
    if (typeof field.objectId !== "string" ||
        typeof field.fieldName !== "string") {
        return undefined;
    }
    return create(ValueSourceSchema, {
        field: create(FieldValueSourceSchema, {
            objectId: field.objectId,
            fieldName: field.fieldName,
            fieldType: typeof field.fieldType === "string" ? field.fieldType : "",
            fieldStorage: typeof field.fieldStorage === "string" ? field.fieldStorage : "",
            conflictStrategy: typeof field.conflictStrategy === "string" ? field.conflictStrategy : "",
            revision: typeof field.revision === "bigint"
                ? field.revision
                : typeof field.revision === "number"
                    ? BigInt(field.revision)
                    : typeof field.revision === "string" && /^\d+$/.test(field.revision)
                        ? BigInt(field.revision)
                        : 0n,
        }),
    });
};
const codecFromSource = (value) => {
    if (!isRecord(value) || !isRecord(value.field)) {
        return undefined;
    }
    const field = value.field;
    return {
        ...(typeof field.fieldType === "string" ? { fieldType: field.fieldType } : {}),
        ...(typeof field.conflictStrategy === "string"
            ? { conflictStrategy: field.conflictStrategy }
            : {}),
    };
};
const withSource = (value, source) => {
    const sourceValue = sourceToProto(source);
    return sourceValue ? create(ValueSchema, { ...value, source: sourceValue }) : value;
};
export const jsToProtoValue = (value, codec) => {
    if (codec?.conflictStrategy === "crdt" && codec.fieldType) {
        return jsToProtoValue(encodeAutomergeFieldWrite(value, codec.fieldType));
    }
    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((entry) => jsToProtoValue(entry)),
                }),
            },
        });
    }
    if (isRecord(value)) {
        if (Object.hasOwn(value, "$caminoValue") && isRecord(value.$caminoSource)) {
            return withSource(jsToProtoValue(value.$caminoValue, codecFromSource(value.$caminoSource)), value.$caminoSource);
        }
        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) => {
    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 loadAutomergeFieldDoc({
                $caminoCrdtType: value.kind.value.type,
                $caminoCrdtEncoding: value.kind.value.encoding === "base64" ? "base64" : "base64",
                $caminoCrdtPayload: bytesToBase64(value.kind.value.payload),
            }, value.kind.value.type);
    }
};
export const jsObjectToProtoFields = (value) => Object.fromEntries(Object.entries(value)
    .filter(([, fieldValue]) => fieldValue !== undefined)
    .map(([key, fieldValue]) => [key, jsToProtoValue(fieldValue)]));
export const protoFieldsToJs = (fields) => Object.fromEntries(Object.entries(fields ?? {}).map(([key, value]) => [
    key,
    protoValueToJs(value),
]));
const normalizeForCanonicalJson = (value) => {
    if (Array.isArray(value)) {
        return value.map(normalizeForCanonicalJson);
    }
    if (!value || typeof value !== "object") {
        return value;
    }
    return Object.fromEntries(Object.entries(value)
        .filter(([, entryValue]) => entryValue !== undefined)
        .sort(([left], [right]) => left.localeCompare(right))
        .map(([key, entryValue]) => [key, normalizeForCanonicalJson(entryValue)]));
};
export const canonicalFieldParams = (params) => JSON.stringify(normalizeForCanonicalJson(Object.fromEntries(Object.entries(params ?? {}).map(([key, value]) => [
    key,
    protoValueToJs(value),
]))));
export const fieldOperationKey = (params) => [
    params.objectId,
    params.fieldName,
    Buffer.from(canonicalFieldParams(params.input)).toString("base64url"),
].join(":");
const derivedDependencyToProto = (dependency) => create(DerivedDependencySchema, {
    kind: dependency.kind,
    objectId: dependency.objectId,
    fieldName: dependency.kind === "field" ? dependency.fieldName : "",
    sourceField: dependency.kind === "edges" ? dependency.sourceField ?? "" : "",
});
const fieldSourceForSpec = (spec, objectId) => {
    if (!spec.field) {
        return undefined;
    }
    return {
        field: {
            objectId,
            fieldName: spec.field.name,
            ...(spec.field.type ? { fieldType: spec.field.type } : {}),
            ...(spec.field.storage ? { fieldStorage: spec.field.storage } : {}),
            ...(spec.field.conflict ? { conflictStrategy: spec.field.conflict } : {}),
        },
    };
};
const wrapFieldResult = (spec, objectId, value) => {
    const source = fieldSourceForSpec(spec, objectId);
    return source ? { $caminoValue: value, $caminoSource: source } : value;
};
const derivedEmissionToProtoForSpec = (spec, objectId, event) => create(WatchEventSchema, {
    watchId: event.watchId,
    value: jsToProtoValue(event.error ? null : wrapFieldResult(spec, objectId, event.value ?? null)),
    dependencies: event.dependencies.map(derivedDependencyToProto),
    error: event.error ?? "",
    initial: event.initial,
});
const watchIdFromResult = (value) => {
    if (!isRecord(value)) {
        return "";
    }
    const watchId = value.watch_id ?? value.watchId;
    return typeof watchId === "string" ? watchId : "";
};
export const objectRef = (objectId) => objectId;
export const createField = (camino, objectId, fieldName, codec) => ({
    async get() {
        recordDerivedDependency({ kind: "field", objectId, fieldName });
        const response = await camino.getObject({ objectId });
        return protoValueToJs(response.object?.fields[fieldName]);
    },
    async set(value) {
        await camino.setField({
            objectId,
            fieldName,
            value: jsToProtoValue(value, codec),
        });
    },
});
export const createCrdtField = (camino, objectId, fieldName, fieldType) => {
    const field = createField(camino, objectId, fieldName, {
        fieldType,
        conflictStrategy: "crdt",
    });
    return {
        ...field,
        async change(fn) {
            const current = await field.get();
            if (!current) {
                throw new Error(`Cannot change missing CRDT field ${fieldName}`);
            }
            const next = Automerge.change(current, (doc) => {
                fn(doc);
            });
            await field.set(next);
            return next;
        },
        async replaceFromJson(value) {
            await camino.setField({
                objectId,
                fieldName,
                value: jsToProtoValue(createAutomergeFieldValue(value, fieldType)),
            });
        },
    };
};
const createDependencyTrackingCaminoClient = (camino) => new Proxy(camino, {
    get(target, property, receiver) {
        if (property === "getObject") {
            return async (request, ...args) => {
                if (request.objectId) {
                    recordDerivedDependency({
                        kind: "object",
                        objectId: request.objectId,
                    });
                }
                return await Reflect.get(target, property, receiver)(request, ...args);
            };
        }
        if (property === "listEdges") {
            return async (request, ...args) => {
                if (request.objectId) {
                    recordDerivedDependency({
                        kind: "edges",
                        objectId: request.objectId,
                        ...(request.sourceField ? { sourceField: request.sourceField } : {}),
                    });
                }
                return await Reflect.get(target, property, receiver)(request, ...args);
            };
        }
        return Reflect.get(target, property, receiver);
    },
});
export const serveQuixosPackageRuntime = (params) => {
    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 runtimeToken = process.env.CAMINO_RUNTIME_AUTH_TOKEN ?? "";
    if (process.env.CAMINO_RUNTIME_AUTH_REQUIRED === "1" && !runtimeToken) {
        throw new Error("CAMINO_RUNTIME_AUTH_REQUIRED=1 but CAMINO_RUNTIME_AUTH_TOKEN is missing");
    }
    const runtimeTokenInterceptor = (next) => async (request) => {
        if (runtimeToken) {
            request.header.set("x-camino-runtime-token", runtimeToken);
        }
        return await next(request);
    };
    const camino = createClient(CaminoService, createConnectTransport({
        baseUrl: caminoUrl,
        httpVersion: "1.1",
        interceptors: runtimeToken ? [runtimeTokenInterceptor] : [],
    }));
    const trackingCamino = createDependencyTrackingCaminoClient(camino);
    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,
        operation: entry.operation ?? "",
    }));
    const resolveRuntimeFunction = (fn) => {
        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[spec.exportName];
        const runtimeFunction = spec.operation && typeof implementationExport === "object"
            ? implementationExport[spec.operation]
            : 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, {
                async handshake() {
                    return create(HandshakeResponseSchema, {
                        packageNamespace: params.packageNamespace,
                        packageName: params.packageName,
                        runtimeProtocolVersion: params.runtimeProtocolVersion,
                        functions: functionRefs,
                    });
                },
                async invoke(request) {
                    try {
                        const { spec, runtimeFunction } = resolveRuntimeFunction(request.function);
                        const value = await runtimeFunction(params.createContext(trackingCamino, request));
                        return create(InvokeResponseSchema, {
                            ok: true,
                            result: jsToProtoValue(spec.operation === "get"
                                ? wrapFieldResult(spec, request.objectId, value)
                                : value),
                        });
                    }
                    catch (error) {
                        return create(InvokeResponseSchema, {
                            ok: false,
                            result: jsToProtoValue(null),
                            error: error instanceof Error ? error.message : String(error),
                        });
                    }
                },
                async *watch(request, context) {
                    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 = {
                            publish(event) {
                                queue.push(derivedEmissionToProtoForSpec(spec, request.objectId, event));
                            },
                        };
                        const result = await watchSinkStorage.run(sink, async () => runtimeFunction(params.createContext(trackingCamino, request)));
                        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),
                                    },
                                })));
                            }
                            catch (error) {
                                logDerivedWatch("stop-error", {
                                    watchId,
                                    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);
};