import { randomBytes, timingSafeEqual } from "node:crypto";
import { chmodSync, existsSync, mkdirSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
import { join, resolve } from "node:path";
import { STANDALONE_PROTOCOL_VERSION } from "./standalone-contract.js";
import type {
  ResumeStandaloneRunRequest,
  StandaloneRuntimeEvent,
  StandaloneWorkflowRuntime,
  StartStandaloneRunRequest,
} from "./standalone-runtime.js";
import { renderStandaloneDashboard } from "./standalone-ui.js";
import { workflowProjectPaths } from "./workflow-paths.js";

const MAX_REQUEST_BYTES = 4 * 1024 * 1024;
const LOOPBACK_HOSTS = new Set(["127.0.0.1", "::1", "localhost"]);

export interface StandaloneWorkflowServerOptions {
  host?: string;
  port?: number;
  authToken?: string;
  /** Override descriptor location (primarily for embedders/tests). */
  descriptorPath?: string;
  /** Set false when discovery is owned by a parent process. Default true. */
  writeDescriptor?: boolean;
  title?: string;
}

export interface RuntimeDescriptor {
  protocolVersion: typeof STANDALONE_PROTOCOL_VERSION;
  pid: number;
  cwd: string;
  host: string;
  port: number;
  authToken: string;
  startedAt: string;
  descriptorPath: string;
}

export interface ListeningStandaloneWorkflowServer {
  host: string;
  port: number;
  authToken: string;
  url: string;
  descriptor: RuntimeDescriptor;
}

/**
 * Local HTTP/SSE control plane for the standalone runtime.
 *
 * The server defaults to loopback and uses a random bearer token. The token is
 * also embedded into the local dashboard and written to a mode-0600 runtime
 * descriptor so other local `dynworkflow` processes can discover the owner.
 */
export class StandaloneWorkflowServer {
  readonly runtime: StandaloneWorkflowRuntime;
  readonly authToken: string;

  private readonly options: StandaloneWorkflowServerOptions;
  private readonly server: Server;
  private readonly sseClients = new Set<ServerResponse>();
  private listening?: ListeningStandaloneWorkflowServer;
  private heartbeat?: ReturnType<typeof setInterval>;
  private closed = false;

  constructor(runtime: StandaloneWorkflowRuntime, options: StandaloneWorkflowServerOptions = {}) {
    this.runtime = runtime;
    this.options = options;
    this.authToken = options.authToken ?? randomBytes(24).toString("base64url");
    if (this.authToken.length < 32) {
      throw new Error("Workflow runtime authToken must contain at least 32 characters.");
    }
    const host = options.host ?? "127.0.0.1";
    if (!LOOPBACK_HOSTS.has(host)) {
      throw new Error(`Refusing to bind the workflow runtime to non-loopback host "${host}".`);
    }
    this.server = createServer((request, response) => {
      void this.handle(request, response).catch((error) => {
        if (response.headersSent) {
          response.end();
          return;
        }
        this.sendJson(response, error instanceof HttpInputError ? error.status : 500, {
          error: error instanceof Error ? error.message : String(error),
        });
      });
    });
  }

  async listen(): Promise<ListeningStandaloneWorkflowServer> {
    if (this.closed) throw new Error("Standalone workflow server is closed.");
    if (this.listening) return this.listening;
    const host = this.options.host ?? "127.0.0.1";
    const requestedPort = normalizePort(this.options.port ?? 0);
    await new Promise<void>((resolveListen, rejectListen) => {
      const onError = (error: Error) => {
        this.server.off("listening", onListening);
        rejectListen(error);
      };
      const onListening = () => {
        this.server.off("error", onError);
        resolveListen();
      };
      this.server.once("error", onError);
      this.server.once("listening", onListening);
      this.server.listen(requestedPort, host);
    });

    const address = this.server.address();
    if (!address || typeof address === "string") {
      await this.close();
      throw new Error("Workflow server did not expose a TCP address.");
    }
    const descriptorPath = this.options.descriptorPath ?? standaloneRuntimeDescriptorPath(this.runtime.cwd);
    const descriptor: RuntimeDescriptor = {
      protocolVersion: STANDALONE_PROTOCOL_VERSION,
      pid: process.pid,
      cwd: this.runtime.cwd,
      host,
      port: address.port,
      authToken: this.authToken,
      startedAt: this.runtime.startedAt,
      descriptorPath,
    };
    const url = `http://${formatUrlHost(host)}:${address.port}/?token=${encodeURIComponent(this.authToken)}`;
    this.listening = { host, port: address.port, authToken: this.authToken, url, descriptor };
    try {
      if (this.options.writeDescriptor ?? true) writeRuntimeDescriptor(descriptor);
    } catch (error) {
      await this.close();
      throw error;
    }

    this.runtime.on("event", this.onRuntimeEvent);
    this.heartbeat = setInterval(() => {
      for (const client of this.sseClients) client.write(": heartbeat\n\n");
    }, 15_000);
    this.heartbeat.unref?.();
    return this.listening;
  }

  address(): ListeningStandaloneWorkflowServer | undefined {
    return this.listening ? { ...this.listening, descriptor: { ...this.listening.descriptor } } : undefined;
  }

  async close(options: { closeRuntime?: boolean } = {}): Promise<void> {
    if (this.closed) return;
    this.closed = true;
    this.runtime.off("event", this.onRuntimeEvent);
    if (this.heartbeat) clearInterval(this.heartbeat);
    for (const client of this.sseClients) client.end();
    this.sseClients.clear();
    const listening = this.listening;
    await new Promise<void>((resolveClose) => {
      if (!this.server.listening) {
        resolveClose();
        return;
      }
      this.server.close(() => resolveClose());
      this.server.closeIdleConnections?.();
    });
    if (listening && (this.options.writeDescriptor ?? true)) {
      removeRuntimeDescriptorIfOwned(listening.descriptor);
    }
    if (options.closeRuntime ?? true) this.runtime.close();
  }

  private readonly onRuntimeEvent = (event: StandaloneRuntimeEvent) => {
    const payload = `id: ${event.sequence}\nevent: workflow\ndata: ${safeJsonStringify(event)}\n\n`;
    for (const client of this.sseClients) client.write(payload);
  };

  private async handle(request: IncomingMessage, response: ServerResponse): Promise<void> {
    const listening = this.listening;
    if (!listening) {
      this.sendJson(response, 503, { error: "Workflow runtime is starting." });
      return;
    }
    const requestUrl = new URL(request.url ?? "/", `http://${formatUrlHost(listening.host)}:${listening.port}`);
    if (!this.authorized(request, requestUrl)) {
      response.setHeader("WWW-Authenticate", "Bearer");
      this.sendJson(response, 401, { error: "Missing or invalid workflow runtime token." });
      return;
    }
    if (!this.originAllowed(request, listening)) {
      this.sendJson(response, 403, { error: "Cross-origin workflow control is not allowed." });
      return;
    }

    if (request.method === "GET" && requestUrl.pathname === "/") {
      const html = renderStandaloneDashboard({
        title: this.options.title,
        projectName: this.runtime.state().project.name,
        authToken: this.authToken,
      });
      response.writeHead(200, {
        "content-type": "text/html; charset=utf-8",
        "cache-control": "no-store",
        "content-security-policy":
          "default-src 'none'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; connect-src 'self'; img-src 'self' data:; base-uri 'none'; frame-ancestors 'none'; form-action 'self'",
        "referrer-policy": "no-referrer",
        "x-content-type-options": "nosniff",
      });
      response.end(html);
      return;
    }

    if (request.method === "GET" && requestUrl.pathname === "/api/health") {
      this.sendJson(response, 200, {
        ok: true,
        protocolVersion: STANDALONE_PROTOCOL_VERSION,
        pid: process.pid,
        cwd: this.runtime.cwd,
        startedAt: this.runtime.startedAt,
      });
      return;
    }

    if (request.method === "GET" && requestUrl.pathname === "/api/state") {
      this.sendJson(response, 200, this.runtime.state());
      return;
    }

    if (request.method === "GET" && requestUrl.pathname === "/api/overview") {
      this.sendJson(response, 200, this.runtime.overview());
      return;
    }

    if (request.method === "GET" && requestUrl.pathname === "/api/events") {
      this.openEventStream(request, response, requestUrl);
      return;
    }

    if (request.method === "POST" && requestUrl.pathname === "/api/runs") {
      const body = await readJsonBody<StartStandaloneRunRequest>(request);
      const started = this.runtime.start(body);
      this.sendJson(response, 202, { runId: started.runId });
      return;
    }

    const runMatch = requestUrl.pathname.match(/^\/api\/runs\/([^/]+)$/);
    if (runMatch && request.method === "GET") {
      const runId = decodeRunId(runMatch[1]);
      const run = this.runtime.getRun(runId);
      this.sendJson(response, run ? 200 : 404, run ?? { error: `Unknown workflow run: ${runId}` });
      return;
    }
    if (runMatch && request.method === "DELETE") {
      const runId = decodeRunId(runMatch[1]);
      const deleted = this.runtime.delete(runId);
      this.sendJson(response, deleted ? 200 : 404, { ok: deleted, runId });
      return;
    }

    const actionMatch = requestUrl.pathname.match(/^\/api\/runs\/([^/]+)\/(pause|resume|stop)$/);
    if (actionMatch && request.method === "POST") {
      const runId = decodeRunId(actionMatch[1]);
      const action = actionMatch[2];
      let ok: boolean;
      if (action === "pause") ok = this.runtime.pause(runId);
      else if (action === "stop") ok = this.runtime.stop(runId);
      else {
        const body = await readJsonBody<ResumeStandaloneRunRequest>(request, true);
        ok = await this.runtime.resume(runId, body);
      }
      this.sendJson(response, ok ? 202 : 409, { ok, runId, action });
      return;
    }

    const checkpointMatch = requestUrl.pathname.match(/^\/api\/checkpoints\/([^/]+)\/respond$/);
    if (checkpointMatch && request.method === "POST") {
      const checkpointId = decodeURIComponent(checkpointMatch[1]);
      const body = await readJsonBody<{ value?: unknown }>(request, true);
      const ok = this.runtime.respondToCheckpoint(checkpointId, body.value);
      this.sendJson(response, ok ? 200 : 404, { ok, checkpointId });
      return;
    }

    this.sendJson(response, 404, { error: "Not found." });
  }

  private openEventStream(request: IncomingMessage, response: ServerResponse, requestUrl: URL): void {
    response.writeHead(200, {
      "content-type": "text/event-stream; charset=utf-8",
      "cache-control": "no-cache, no-transform",
      connection: "keep-alive",
      "x-accel-buffering": "no",
    });
    response.write("retry: 1000\n\n");
    this.sseClients.add(response);
    const lastHeader = Number(request.headers["last-event-id"]);
    const lastQuery = Number(requestUrl.searchParams.get("since"));
    const since =
      Number.isSafeInteger(lastHeader) && lastHeader >= 0
        ? lastHeader
        : Number.isSafeInteger(lastQuery) && lastQuery >= 0
          ? lastQuery
          : 0;
    for (const event of this.runtime.recentEvents(since)) {
      response.write(`id: ${event.sequence}\nevent: workflow\ndata: ${safeJsonStringify(event)}\n\n`);
    }
    request.on("close", () => this.sseClients.delete(response));
  }

  private authorized(request: IncomingMessage, requestUrl: URL): boolean {
    const queryToken = requestUrl.searchParams.get("token");
    if (tokensEqual(queryToken, this.authToken)) return true;
    const headerToken = request.headers["x-workflow-token"];
    if (typeof headerToken === "string" && tokensEqual(headerToken, this.authToken)) return true;
    const authorization = request.headers.authorization;
    return typeof authorization === "string" && tokensEqual(authorization, `Bearer ${this.authToken}`);
  }

  private originAllowed(request: IncomingMessage, listening: ListeningStandaloneWorkflowServer): boolean {
    const origin = request.headers.origin;
    if (!origin) return true;
    const allowed = new Set([
      `http://${formatUrlHost(listening.host)}:${listening.port}`,
      `http://127.0.0.1:${listening.port}`,
      `http://localhost:${listening.port}`,
      `http://[::1]:${listening.port}`,
    ]);
    return allowed.has(origin);
  }

  private sendJson(response: ServerResponse, status: number, value: unknown): void {
    response.writeHead(status, {
      "content-type": "application/json; charset=utf-8",
      "cache-control": "no-store",
      "x-content-type-options": "nosniff",
    });
    response.end(safeJsonStringify(value));
  }
}

export function standaloneRuntimeDescriptorPath(cwd: string): string {
  return join(workflowProjectPaths(resolve(cwd)).rootDir, "runtime.json");
}

export function readRuntimeDescriptor(
  cwd: string,
  descriptorPath = standaloneRuntimeDescriptorPath(cwd),
): RuntimeDescriptor | null {
  try {
    const parsed = JSON.parse(readFileSync(descriptorPath, "utf8")) as Partial<RuntimeDescriptor>;
    if (
      parsed.protocolVersion !== STANDALONE_PROTOCOL_VERSION ||
      typeof parsed.pid !== "number" ||
      typeof parsed.cwd !== "string" ||
      typeof parsed.host !== "string" ||
      typeof parsed.port !== "number" ||
      typeof parsed.authToken !== "string" ||
      parsed.authToken.length < 32 ||
      typeof parsed.startedAt !== "string"
    ) {
      return null;
    }
    return { ...parsed, descriptorPath } as RuntimeDescriptor;
  } catch {
    return null;
  }
}

function tokensEqual(left: string | null | undefined, right: string): boolean {
  if (typeof left !== "string") return false;
  const leftBytes = Buffer.from(left);
  const rightBytes = Buffer.from(right);
  return leftBytes.length === rightBytes.length && timingSafeEqual(leftBytes, rightBytes);
}

function writeRuntimeDescriptor(descriptor: RuntimeDescriptor): void {
  const dir = resolve(descriptor.descriptorPath, "..");
  mkdirSync(dir, { recursive: true, mode: 0o700 });
  chmodSync(dir, 0o700);
  const tempPath = `${descriptor.descriptorPath}.${process.pid}.tmp`;
  writeFileSync(tempPath, `${JSON.stringify(descriptor, null, 2)}\n`, { mode: 0o600 });
  renameSync(tempPath, descriptor.descriptorPath);
  chmodSync(descriptor.descriptorPath, 0o600);
}

function removeRuntimeDescriptorIfOwned(descriptor: RuntimeDescriptor): void {
  try {
    if (!existsSync(descriptor.descriptorPath)) return;
    const current = readRuntimeDescriptor(descriptor.cwd, descriptor.descriptorPath);
    if (current?.pid === process.pid && current.authToken === descriptor.authToken) {
      unlinkSync(descriptor.descriptorPath);
    }
  } catch {
    // A descriptor is a discovery hint; stale cleanup is best-effort.
  }
}

async function readJsonBody<T>(request: IncomingMessage, allowEmpty = false): Promise<T> {
  const chunks: Buffer[] = [];
  let total = 0;
  for await (const chunk of request) {
    const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
    total += buffer.length;
    if (total > MAX_REQUEST_BYTES) throw new HttpInputError(413, "Request body is too large.");
    chunks.push(buffer);
  }
  if (chunks.length === 0 || total === 0) {
    if (allowEmpty) return {} as T;
    throw new HttpInputError(400, "A JSON request body is required.");
  }
  try {
    return JSON.parse(Buffer.concat(chunks).toString("utf8")) as T;
  } catch {
    throw new HttpInputError(400, "Request body must be valid JSON.");
  }
}

function safeJsonStringify(value: unknown): string {
  // Track the active ancestor chain, not every object ever visited. Two state
  // fields may legitimately reference the same result object (for example the
  // final agent result and the run result); JSON should serialize both. Only a
  // value that points back into its current ancestor chain is truly circular.
  const ancestors: object[] = [];
  return JSON.stringify(value, function (_key, item: unknown) {
    if (typeof item === "bigint") return item.toString();
    if (item instanceof Error) {
      return {
        name: item.name,
        message: item.message,
        stack: item.stack,
        ...((item as Error & { code?: unknown }).code !== undefined
          ? { code: (item as Error & { code?: unknown }).code }
          : {}),
      };
    }
    if (typeof item === "function" || typeof item === "symbol") return String(item);
    if (item && typeof item === "object") {
      while (ancestors.length > 0 && ancestors.at(-1) !== this) ancestors.pop();
      if (ancestors.includes(item)) return "[Circular]";
      ancestors.push(item);
    }
    return item;
  });
}

function normalizePort(port: number): number {
  if (!Number.isSafeInteger(port) || port < 0 || port > 65_535) {
    throw new RangeError(`Invalid workflow server port: ${port}`);
  }
  return port;
}

function formatUrlHost(host: string): string {
  return host.includes(":") && !host.startsWith("[") ? `[${host}]` : host;
}

function decodeRunId(encoded: string): string {
  const runId = decodeURIComponent(encoded);
  if (!/^[a-zA-Z0-9][a-zA-Z0-9._-]{0,191}$/.test(runId) || runId === "." || runId === "..") {
    throw new HttpInputError(400, "Invalid workflow run ID.");
  }
  return runId;
}

class HttpInputError extends Error {
  constructor(
    readonly status: number,
    message: string,
  ) {
    super(message);
  }
}