/// <reference path="./proto.d.ts" />
/**
 * Internal gRPC server implementing HashiCorp go-plugin protocol.
 *
 * This file is NEVER imported by plugin developers.
 * It handles: magic cookie check, proto loading, Unix socket server,
 * handshake output, and routing Execute calls to the plugin handler.
 *
 * ─── Hot reload contract ─────────────────────────────────────────────
 *
 * `bun --hot` re-evaluates the plugin entry file on every source change,
 * which means `createPlugin().start()` (and therefore startGrpcServer)
 * runs again in the SAME process. We must be idempotent or the second
 * bindAsync would fail with "address already in use".
 *
 * The trick: keep the gRPC server + state on `globalThis`, which bun
 * preserves across hot reloads. On the first call we bind the socket
 * once; subsequent calls swap the handler/manifest in place. The gRPC
 * service methods read from globalThis on every request, so swaps take
 * effect immediately without invalidating np-agent's existing gRPC
 * connection — the agent never notices a reload.
 */

import * as grpc from "@grpc/grpc-js";
import * as protoLoader from "@grpc/proto-loader";
import { readFileSync, unlinkSync, writeFileSync } from "fs";
import { tmpdir } from "os";
import { cfg } from "../config";
import { join } from "path";
import type { PluginHandler, ExecuteRequest, PluginManifest } from "../types";
// Embed the proto as text so it survives `bun build --compile` (a compiled
// binary has no plugin.proto on disk next to the source). Works from source too.
import protoSource from "./plugin.proto" with { type: "text" };

const MAGIC_COOKIE_KEY = "NP_AGENT_PLUGIN";
const MAGIC_COOKIE_VALUE = "np-agent-v1";

// workerServerCredentials returns mTLS credentials when the agent provided the
// cert material (NP_WORKER_TLS_CA/CERT/KEY, inline PEM or *_FILE path) — the
// server presents its cert and REQUIRES a client cert signed by the CA, so only
// the agent that spawned it can call it. Falls back to insecure (unix-socket /
// local dev) when no certs are present. No plugin-author involvement.
function workerServerCredentials(): grpc.ServerCredentials {
  const read = (valEnv: string, fileEnv: string): Buffer | null => {
    const file = process.env[fileEnv];
    if (file) return readFileSync(file);
    const val = process.env[valEnv];
    if (val) return Buffer.from(val);
    return null;
  };
  const ca = read("NP_WORKER_TLS_CA", "NP_WORKER_TLS_CA_FILE");
  const cert = read("NP_WORKER_TLS_CERT", "NP_WORKER_TLS_CERT_FILE");
  const key = read("NP_WORKER_TLS_KEY", "NP_WORKER_TLS_KEY_FILE");
  if (ca && cert && key) {
    // 3rd arg true = require + verify the client certificate against `ca`.
    return grpc.ServerCredentials.createSsl(ca, [{ private_key: key, cert_chain: cert }], true);
  }
  return grpc.ServerCredentials.createInsecure();
}

// ─── Proto-level types ──────────────────────────────────────────────
//
// Shapes of the wire messages as they come off @grpc/proto-loader with
// `keepCase: true`. Fields are snake_case (matching plugin.proto); the
// public ExecuteRequest type (from ../types) is the camelCase version
// we expose to plugin authors. Narrowing at the edges keeps the `any`
// from bleeding into the rest of the file.

interface ProtoExecuteRequest {
  message_id: string;
  command_type: string;
  payload: Buffer | string;
  supports_partial: boolean;
  can_cancel: boolean;
  reply_to: string;
  action_type?: string;
}

interface ProtoExecuteEvent {
  started?: Record<string, never>;
  output?: {
    stdout: string;
    stderr: string;
    data: Buffer;
  };
  result?: {
    success: boolean;
    data: Buffer;
    error: string;
    exit_code: number;
    error_code: string;
  };
}

interface ProtoEmpty {
  // Capabilities / Health / Version all take empty request messages.
}

interface ProtoCapabilitiesResponse {
  command_types: string[];
  priority: number;
}

interface ProtoHealthResponse {
  healthy: boolean;
  uptime_ms?: string;
  active_commands?: number;
  memory_bytes?: string;
  message?: string;
}

interface ProtoVersionResponse {
  name: string;
  version: string;
  metadata: Record<string, string>;
}

// Expected shape of the loaded proto descriptor. @grpc/proto-loader
// returns a generic GrpcObject; we cast to this once so every subsequent
// access is typed.
interface LoadedProto {
  npagent: {
    plugin: {
      v1: {
        ExecutorPlugin: grpc.ServiceClientConstructor & {
          service: grpc.ServiceDefinition;
        };
      };
    };
  };
}

// ─── Cross-reload state ─────────────────────────────────────────────

interface PluginState {
  server: grpc.Server;
  socketPath: string;
  startTime: number;
  activeCommands: number;
  handler: PluginHandler;
  manifest: PluginManifest;
}

declare global {
  // eslint-disable-next-line no-var
  var __nullplatformPluginState: PluginState | undefined;
  // eslint-disable-next-line no-var
  var __nullplatformConnGuard: boolean | undefined;
}

export function startGrpcServer(
  handler: PluginHandler,
  manifest: PluginManifest,
): void {
  // Magic cookie check is cheap and idempotent — safe to run on every call.
  if (process.env[MAGIC_COOKIE_KEY] !== MAGIC_COOKIE_VALUE) {
    console.error(
      "This binary is a plugin. It is not meant to be executed directly.\n" +
      `Set ${MAGIC_COOKIE_KEY}=${MAGIC_COOKIE_VALUE} to run.`,
    );
    process.exit(1);
  }

  // A TLS listener is probed by bare-TCP checks (k8s TCPSocket readiness) and
  // scanned by anything that can reach the port; those abort the TLS handshake
  // and Node surfaces them as PROCESS-level socket errors. A server must survive
  // connection noise — swallow known-transient socket errors, exit on anything
  // real. Installed once per process (survives bun --hot).
  if (!globalThis.__nullplatformConnGuard) {
    globalThis.__nullplatformConnGuard = true;
    const transient = /ECONNRESET|EPIPE|ECONNABORTED|socket hang up|handshake|ssl|tls/i;
    process.on("uncaughtException", (err: NodeJS.ErrnoException) => {
      if (transient.test(err?.code ?? "") || transient.test(err?.message ?? "")) {
        return; // benign connection noise (probes, scans, aborted handshakes)
      }
      console.error(`[worker] fatal: ${err?.stack ?? err}`);
      process.exit(1);
    });
  }

  // Hot reload path: server already bound in this process. Swap the
  // handler/manifest into the existing state object so subsequent gRPC
  // requests pick up the new code. No re-bind, no re-handshake, no
  // duplicate signal handlers — the agent's connection stays live.
  if (globalThis.__nullplatformPluginState) {
    const existing = globalThis.__nullplatformPluginState;
    existing.handler = handler;
    existing.manifest = manifest;
    console.error(`[${manifest.name}] hot reload: handler swapped`);
    return;
  }

  // First start in this process: create state, load proto, add service,
  // bind the socket, print handshake, register signal handlers. Each
  // step runs exactly once per process lifetime regardless of reloads.
  const state: PluginState = {
    server: new grpc.Server(),
    socketPath: join(
      tmpdir(),
      `np-plugin-${manifest.name}-${process.pid}.sock`,
    ),
    startTime: Date.now(),
    activeCommands: 0,
    handler,
    manifest,
  };
  // @grpc/proto-loader loads from a file path, so materialize the embedded proto
  // text to a temp file. This is the one path that works both from source and
  // from a compiled single-file binary.
  const protoPath = join(tmpdir(), `np-plugin-${process.pid}.proto`);
  writeFileSync(protoPath, protoSource);
  const packageDefinition = protoLoader.loadSync(protoPath, {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true,
  });
  const proto = grpc.loadPackageDefinition(packageDefinition) as unknown as LoadedProto;

  state.server.addService(proto.npagent.plugin.v1.ExecutorPlugin.service, {
    Execute: executeHandler,
    Capabilities: capabilitiesHandler,
    Health: healthHandler,
    Version: versionHandler,
  });

  // Two ways to serve:
  //   - TCP (NP_GRPC_LISTEN set): the worker runs as its own pod/container and the
  //     orchestrating agent dials it over the network. No handshake — the agent
  //     assigned the address and uses the Health RPC for readiness.
  //   - unix socket (default): the agent spawns this as a child and reads the
  //     go-plugin handshake line to discover the socket.
  const tcpListen = cfg.grpcListen();
  const bindTarget = tcpListen ? tcpListen : `unix://${state.socketPath}`;

  // Remove stale socket from a previous crash (PID reuse is rare but possible).
  if (!tcpListen) { try { unlinkSync(state.socketPath); } catch {} }

  // TCP workers use mTLS when the agent supplied certs; unix sockets stay local.
  const creds = tcpListen ? workerServerCredentials() : grpc.ServerCredentials.createInsecure();
  state.server.bindAsync(
    bindTarget,
    creds,
    (err) => {
      if (err) {
        console.error(`[${manifest.name}] gRPC bind failed: ${err.message}`);
        process.exit(1);
      }
      // Only publish state after bind succeeds — a failed bind must not
      // leave a stale globalThis that the hot-reload path would trust.
      globalThis.__nullplatformPluginState = state;
      if (tcpListen) {
        const mtls = !!(process.env.NP_WORKER_TLS_CA || process.env.NP_WORKER_TLS_CA_FILE);
        console.error(`[${manifest.name}] gRPC listening on ${bindTarget} (tcp, ${mtls ? "mTLS" : "insecure"})`);
      } else {
        process.stdout.write(`1|1|unix|${state.socketPath}|grpc|\n`);
        console.error(`[${manifest.name}] gRPC listening on ${state.socketPath}`);
      }
    },
  );

  const shutdown = () => state.server.tryShutdown(() => {
    if (!tcpListen) { try { unlinkSync(state.socketPath); } catch {} }
    process.exit(0);
  });
  process.on("SIGTERM", shutdown);
  process.on("SIGINT", shutdown);
}

// ─── gRPC service methods ────────────────────────────────────────────
//
// Defined at module scope (not inside startGrpcServer) so they read
// `globalThis.__nullplatformPluginState` on every invocation. A hot-swapped
// handler/manifest takes effect on the very next request without needing
// to re-register the service or rebind the server.

function executeHandler(
  call: grpc.ServerWritableStream<ProtoExecuteRequest, ProtoExecuteEvent>,
): void {
  const state = globalThis.__nullplatformPluginState;
  if (!state) {
    // Impossible in practice (service is only registered after state is set),
    // but bail out cleanly rather than crash if it ever happens.
    call.end();
    return;
  }
  state.activeCommands++;

  const raw = call.request;
  const req: ExecuteRequest = {
    messageId: raw.message_id,
    commandType: raw.command_type,
    payload:
      typeof raw.payload === "string"
        ? Buffer.from(raw.payload, "base64")
        : raw.payload,
    supportsPartial: raw.supports_partial,
    canCancel: raw.can_cancel,
    replyTo: raw.reply_to,
    actionType: raw.action_type || "",
    emit(output) {
      call.write({
        output: {
          stdout: output.stdout ?? "",
          stderr: output.stderr ?? "",
          data: output.data ? Buffer.from(JSON.stringify(output.data)) : Buffer.alloc(0),
        },
      });
    },
  };

  call.write({ started: {} });

  state.handler
    .execute(req)
    .then((result) => {
      call.write({
        result: {
          success: result.success,
          data: result.data
            ? Buffer.from(JSON.stringify(result.data))
            : Buffer.alloc(0),
          error: result.error || "",
          exit_code: result.exitCode || 0,
          error_code: result.errorCode || "",
        },
      });
    })
    .catch((err) => {
      call.write({
        result: {
          success: false,
          data: Buffer.alloc(0),
          error: err instanceof Error ? err.message : String(err),
          exit_code: 1,
          error_code: "",
        },
      });
    })
    .finally(() => {
      const s = globalThis.__nullplatformPluginState;
      if (s) s.activeCommands--;
      call.end();
    });
}

// Capabilities / Health / Version take an empty request, so only the
// callback argument is used. The `call` parameter is positional in the
// gRPC service signature — we have to name it to reach `callback`.
function capabilitiesHandler(
  call: grpc.ServerUnaryCall<ProtoEmpty, ProtoCapabilitiesResponse>,
  callback: grpc.sendUnaryData<ProtoCapabilitiesResponse>,
): void {
  const state = globalThis.__nullplatformPluginState;
  callback(null, {
    command_types: state?.manifest.command_types ?? [],
    priority: 100,
  });
}

function healthHandler(
  call: grpc.ServerUnaryCall<ProtoEmpty, ProtoHealthResponse>,
  callback: grpc.sendUnaryData<ProtoHealthResponse>,
): void {
  const state = globalThis.__nullplatformPluginState;
  if (!state) {
    callback(null, { healthy: false, message: "not started" });
    return;
  }
  callback(null, {
    healthy: true,
    uptime_ms: String(Date.now() - state.startTime),
    active_commands: state.activeCommands,
    memory_bytes: String(
      typeof Bun !== "undefined"
        ? (process.memoryUsage as unknown as { rss?: () => number }).rss?.() ?? process.memoryUsage().rss
        : process.memoryUsage().rss,
    ),
    message: "ok",
  });
}

function versionHandler(
  call: grpc.ServerUnaryCall<ProtoEmpty, ProtoVersionResponse>,
  callback: grpc.sendUnaryData<ProtoVersionResponse>,
): void {
  const state = globalThis.__nullplatformPluginState;
  callback(null, {
    name: state?.manifest.name ?? "",
    version: state?.manifest.version ?? "",
    metadata: {
      runtime: typeof Bun !== "undefined" ? "bun" : "node",
      bun_version: typeof Bun !== "undefined" ? Bun.version : "n/a",
    },
  });
}