/**
 * A tools bundle as a process the node's runtime launches (novox/hq ADR 0188).
 *
 * The runtime speaks MCP over stdio to it: `initialize`, `tools/list` once, `tools/call` per call,
 * newline-framed JSON-RPC on stdin and stdout. This is the whole of what a tools bundle in any
 * language must do, and this file is the TypeScript SDK's share of it: the loop, so a module writes
 * its tools and nothing else. A tool named `<seat>.<verb>` is the module's implementation of that
 * seat's verb; any other name is the module's own tool.
 *
 * **No transport here.** The bus, the subjects, the memberships are the runtime's (ADR 0039, 0175);
 * a bundle knows only stdin and stdout, which is why a bus change rebuilds no bundle in any language.
 *
 * stdout is the protocol. A tool that wants to log writes to stderr, which the runtime keeps.
 */
import type { ToolDefinition } from "../contracts/index.js";
import { collectTools } from "../tools/index.js";
import { useBroker, type Broker, type Envelope } from "../messaging/index.js";
import { useStateChannel, type StateChange } from "../state/index.js";

/** The protocol version this speaks; the runtime says the same. */
export const PROTOCOL = "2025-03-26";

interface Request {
  jsonrpc: string;
  id?: number | string | null;
  method: string;
  params?: Record<string, unknown>;
}

/** A module's declared input as a JSON schema: the bare property map the catalogue's modules write
 *  is wrapped, a schema is passed through, nothing is an empty object. */
export function asSchema(input: unknown): Record<string, unknown> {
  if (!input || typeof input !== "object" || Array.isArray(input)) return { type: "object", properties: {} };
  const given = input as Record<string, unknown>;
  if (given.type === "object" || "properties" in given) return given;
  if (Object.keys(given).length === 0) return { type: "object", properties: {} };
  return { type: "object", properties: given };
}

/**
 * One stdio session per process: the channel to the node's runtime that launched this bundle
 * (novox/hq ADR 0193, ADR 0198). It answers MCP — initialize, tools/list, tools/call — and is this
 * bundle's bus: what the module's code emits, subscribes to and asks goes to the runtime as a request
 * on the same channel, and the runtime's answers and the events it delivers arrive on it.
 *
 * **Started before the bundle's own code runs.** A long-running entrypoint subscribes as it is
 * imported (`await on(...)` at the top level), and the launcher imports it before it says what tools
 * it serves — so the broker must already be bound and stdin already read when that import starts, or
 * the subscription would wait on an answer nobody reads. The session therefore starts as this module
 * loads when the runtime launched the process, and `initialize` and `tools/list` are answered once the
 * bundle has said what it serves.
 */
interface Session {
  serve(name: string, tools: ToolDefinition[]): void;
  closed: Promise<void>;
}

let session: Session | undefined;

function startSession(): Session {
  if (session) return session;
  const say = (message: unknown): void => {
    process.stdout.write(`${JSON.stringify(message)}\n`);
  };
  let nextOut = 1;
  const awaiting = new Map<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
  const askAs = (id: string, method: string, params: unknown): Promise<unknown> =>
    new Promise((resolve, reject) => {
      awaiting.set(id, { resolve, reject });
      say({ jsonrpc: "2.0", id, method, params });
    });
  const ask = (method: string, params: unknown): Promise<unknown> => askAs(`bundle-${nextOut++}`, method, params);

  // The module's state (ADR 0201): each watch's changes arrive as `mesh/state` requests naming the
  // watch by the id it was asked with, so its handler is registered under that id before it is asked
  // — the current values arrive before the watch itself is answered.
  const watches = new Map<string, (change: StateChange<unknown>) => Promise<void>>();
  useStateChannel({
    ask,
    watch: async (params, deliver) => {
      const id = `bundle-${nextOut++}`;
      watches.set(id, deliver);
      try {
        await askAs(id, "mesh/state.watch", params);
      } catch (err) {
        watches.delete(id);
        throw err;
      }
    },
  });

  // What the module's code reacts to: every handler, matched here against each event the runtime
  // delivers. The runtime binds the module's one durable consumer and hands every message it reads to
  // this process; an event no handler is for is answered at once, so it is acknowledged rather than
  // redelivered until it expires — the consumer's filter is the controller's and may be wider.
  const listeners: { pattern: string; handler: (env: Envelope<unknown>) => Promise<void> }[] = [];
  const viaRuntime: Broker = {
    publish: async <T>(env: Envelope<T>) => {
      await ask("mesh/publish", env);
    },
    request: (async <Req, Res>(key: string, body: Req): Promise<Res> =>
      (await ask("mesh/ask", { key, body })) as Res) as Broker["request"],
    handle: (async () => {
      throw new Error("a launched bundle answers through its tools, not by binding subjects (novox/hq ADR 0193)");
    }) as Broker["handle"],
    subscribe: (async <T>(pattern: string, handler: (env: Envelope<T>) => Promise<void>) => {
      const listener = { pattern, handler: handler as (env: Envelope<unknown>) => Promise<void> };
      listeners.push(listener);
      const remove = (): void => {
        const at = listeners.indexOf(listener);
        if (at >= 0) listeners.splice(at, 1);
      };
      try {
        await ask("mesh/subscribe", { pattern });
      } catch (err) {
        // Refused: nothing is bound, and a handler left behind would be handed each event twice once
        // the module asks again (novox/hq issue 276).
        remove();
        throw err;
      }
      return remove;
    }) as Broker["subscribe"],
    close: async () => {},
  };
  useBroker(() => viaRuntime);

  let name = "bundle";
  let byName = new Map<string, ToolDefinition>();
  let tools: ToolDefinition[] = [];
  let ready!: () => void;
  const served = new Promise<void>((resolve) => (ready = resolve));

  const answer = async (request: Request): Promise<void> => {
    const notification = request.id === undefined || request.id === null;
    switch (request.method) {
      case "initialize":
        await served;
        say({ jsonrpc: "2.0", id: request.id, result: { protocolVersion: PROTOCOL, capabilities: { tools: {} }, serverInfo: { name, version: "1" } } });
        return;
      case "notifications/initialized":
        return;
      case "ping":
        if (!notification) say({ jsonrpc: "2.0", id: request.id, result: {} });
        return;
      case "tools/list":
        await served;
        say({
          jsonrpc: "2.0", id: request.id,
          result: { tools: tools.map((t) => ({ name: t.name, description: t.description, inputSchema: asSchema(t.input) })) },
        });
        return;
      case "tools/call": {
        await served;
        const wanted = String(request.params?.name ?? "");
        const tool = byName.get(wanted);
        if (!tool) {
          say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: `${name} has no tool ${wanted}` } });
          return;
        }
        try {
          const result = await tool.run((request.params?.arguments as Record<string, unknown> | undefined) ?? {});
          say({ jsonrpc: "2.0", id: request.id, result: { content: [{ type: "text", text: JSON.stringify(result ?? null) }] } });
        } catch (err) {
          // The tool failed; said as a tool error, not a protocol one, because the call was well-formed.
          say({ jsonrpc: "2.0", id: request.id, result: { content: [{ type: "text", text: err instanceof Error ? err.message : String(err) }], isError: true } });
        }
        return;
      }
      case "mesh/event": {
        // One event the runtime read from this module's consumer (ADR 0198). Answered once every
        // handler it is for has taken it; an error is answered as one, so the runtime does not
        // acknowledge it and the bus delivers it again.
        const env = (request.params as { envelope?: Envelope<unknown> } | undefined)?.envelope;
        if (!env || typeof env.key !== "string") {
          say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: "mesh/event carries no envelope" } });
          return;
        }
        try {
          for (const l of listeners.filter((x) => topicMatches(x.pattern, env.key))) await l.handler(env);
          say({ jsonrpc: "2.0", id: request.id, result: {} });
        } catch (err) {
          say({ jsonrpc: "2.0", id: request.id, error: { code: -32000, message: err instanceof Error ? err.message : String(err) } });
        }
        return;
      }
      case "mesh/state": {
        // One change to a state this bundle watches (ADR 0201), answered once its handler has taken
        // it. The runtime delivers a watch's changes one at a time, in order.
        const p = request.params as { watch?: string; change?: StateChange<unknown> } | undefined;
        const deliver = p?.watch !== undefined ? watches.get(String(p.watch)) : undefined;
        if (!deliver || !p?.change) {
          say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: "mesh/state names no watch of this bundle" } });
          return;
        }
        try {
          await deliver(p.change);
          say({ jsonrpc: "2.0", id: request.id, result: {} });
        } catch (err) {
          say({ jsonrpc: "2.0", id: request.id, error: { code: -32000, message: err instanceof Error ? err.message : String(err) } });
        }
        return;
      }
      default:
        if (!notification) say({ jsonrpc: "2.0", id: request.id, error: { code: -32601, message: `no ${request.method}` } });
    }
  };

  const closed = (async () => {
    for await (const line of lines()) {
      let request: Request;
      try {
        request = JSON.parse(line) as Request;
      } catch {
        continue; // not framed; nobody to answer
      }
      // The runtime's answer to something this bundle asked.
      const reply = request as unknown as { id?: string; method?: string; result?: unknown; error?: { message?: string } };
      if (reply.method === undefined && typeof reply.id === "string" && awaiting.has(reply.id)) {
        const waiting = awaiting.get(reply.id)!;
        awaiting.delete(reply.id);
        if (reply.error) waiting.reject(new Error(reply.error.message ?? "the runtime refused"));
        else waiting.resolve(reply.result);
        continue;
      }
      // Beside the loop, never in it: a handler or tool that emits or asks waits for the runtime's
      // answer, which arrives on the line this loop reads next — and nothing queues behind a slow one.
      void answer(request);
    }
  })();

  session = {
    serve(n, ts) {
      name = n;
      tools = ts;
      byName = new Map(ts.map((t) => [t.name, t]));
      ready();
    },
    closed,
  };
  return session;
}

/**
 * Serve these tools to the runtime over stdio until stdin closes. `name` is the module's; a tool
 * whose name is `<seat>.<verb>` is served as the seat's. Resolves when the runtime hangs up. A bundle
 * that only runs long serves no tools and still answers the runtime's handshake.
 */
export async function serveStdio(name: string, tools: ToolDefinition[]): Promise<void> {
  const s = startSession();
  s.serve(name, tools);
  await s.closed;
}

/** Whether an event's key matches a handler's pattern: `*` one segment, `#` or `**` any number —
 *  the same rule the runtime's own subscriptions use (novox/hq design 29 §1). */
export function topicMatches(pattern: string, key: string): boolean {
  return matchFrom(pattern.split("."), 0, key.split("."), 0);
}

function matchFrom(p: string[], pi: number, k: string[], ki: number): boolean {
  if (pi === p.length) return ki === k.length;
  if (p[pi] === "#" || p[pi] === "**") {
    for (let skip = ki; skip <= k.length; skip++) {
      if (matchFrom(p, pi + 1, k, skip)) return true;
    }
    return false;
  }
  if (ki === k.length) return false;
  if (p[pi] !== "*" && p[pi] !== k[ki]) return false;
  return matchFrom(p, pi + 1, k, ki + 1);
}

/** The word the node's runtime sets for a bundle it launches: the module it serves the bundle as
 *  (novox/hq ADR 0193). */
export const SERVED_AS = "MESH_SERVED_MODULE";

/**
 * Serve everything registered with `registerModuleTools` over stdio — the launcher every TypeScript
 * bundle the runtime serves is started through (novox/hq ADR 0193). What the module the runtime
 * serves this bundle as registered is listed by its own name; anything registered under another
 * name — a seat this module implements — is listed `<seat>.<verb>`, whichever was registered first.
 * Every contributor is given this process's environment, which is the bundle's own.
 */
export async function serveRegisteredOverStdio(as: string | undefined = process.env[SERVED_AS]): Promise<void> {
  const groups = collectTools(process.env);
  const name = as ?? groups[0]?.module ?? "bundle";
  const tools: ToolDefinition[] = [];
  for (const g of groups) {
    for (const t of g.tools) tools.push(g.module === name ? t : { ...t, name: `${g.module}.${t.name}` });
  }
  await serveStdio(name, tools);
}

async function* lines(): AsyncGenerator<string> {
  let buffered = "";
  for await (const chunk of process.stdin) {
    buffered += (chunk as Buffer).toString("utf8");
    let at: number;
    while ((at = buffered.indexOf("\n")) >= 0) {
      const line = buffered.slice(0, at).trim();
      buffered = buffered.slice(at + 1);
      if (line !== "") yield line;
    }
  }
  if (buffered.trim() !== "") yield buffered.trim();
}

// Launched by the node's runtime: the channel is this process's from its first line, so the module's
// code may subscribe, emit and ask while it is being imported (ADR 0198).
if (process.env[SERVED_AS]) startSession();