/**
 * `dench chat new` / `dench chat send` — spawn a chat-turn workflow
 * from the CLI, exactly the same workflow the web UI uses. Threads
 * created here are visible in `dench chat list` and the UI's chat
 * thread list immediately, since they go through the same Convex
 * tables (`chatThreads` + `chatMessages`).
 *
 * Aliased as `dench agent new` / `dench agent send` so users who
 * grep for "agent" find them.
 *
 * Both subcommands POST to `/api/chat/cli` (auth: same `dch_agent_*`
 * session token or Dench API key as the rest of the CLI). The
 * server creates the thread + queued run, kicks off
 * `chatTurnWorkflow`, and returns ids + URLs. With `--follow` the
 * CLI then opens the durable resumable stream
 * (`/api/chat/[runId]/stream`) and pretty-prints text deltas + tool
 * calls until the workflow finishes.
 */
import { assertNoUnknownFlags, hasFlag } from "./lib/cli-args";

class ChatSpawnError extends Error {}

type RuntimeBundle = {
  host: string;
  bearerToken: string;
};

type ChatSpawnContext = {
  runtime: RuntimeBundle;
  args: string[];
  jsonOutput: boolean;
};

function getCurrentRuntimeTimeZone(): string | null {
  try {
    const timeZone = Intl.DateTimeFormat().resolvedOptions().timeZone;
    if (!timeZone || timeZone.length > 100) return null;
    // Validate that Intl accepts the resolved value before forwarding it.
    Intl.DateTimeFormat("en-US", { timeZone }).format(new Date());
    return timeZone;
  } catch {
    return null;
  }
}

function consumeFlagValue(args: string[], name: string): string | undefined {
  const idx = args.indexOf(name);
  if (idx === -1) return undefined;
  const value = args[idx + 1];
  if (!value || value.startsWith("--"))
    throw new ChatSpawnError(`${name} requires a value.`);
  if (args.indexOf(name, idx + 1) !== -1)
    throw new ChatSpawnError(`Duplicate flag: ${name}`);
  args.splice(idx, 2);
  return value;
}

function consumeFlag(args: string[], name: string): boolean {
  const idx = args.indexOf(name);
  if (idx === -1) return false;
  args.splice(idx, 1);
  return true;
}

function out(ctx: ChatSpawnContext, value: unknown): void {
  if (ctx.jsonOutput) {
    console.log(JSON.stringify(value, null, 2));
    return;
  }
  if (value === null || value === undefined) {
    console.log("(empty)");
    return;
  }
  console.log(JSON.stringify(value, null, 2));
}

async function postJson(
  url: string,
  bearer: string,
  body: unknown,
): Promise<{ ok: boolean; status: number; payload: unknown }> {
  const timezone = getCurrentRuntimeTimeZone();
  const response = await fetch(url, {
    method: "POST",
    headers: {
      accept: "application/json",
      "content-type": "application/json",
      authorization: `Bearer ${bearer}`,
      ...(timezone ? { "x-dench-user-timezone": timezone } : {}),
    },
    body: JSON.stringify(body),
  });
  const text = await response.text();
  let payload: unknown;
  try {
    payload = text ? JSON.parse(text) : null;
  } catch {
    payload = { raw: text };
  }
  return { ok: response.ok, status: response.status, payload };
}

function buildSpawnUrl(host: string): string {
  return `${host.replace(/\/+$/, "")}/api/chat/cli`;
}

function buildStreamUrlByRunId(host: string, runId: string): string {
  return `${host.replace(/\/+$/, "")}/api/chat/${encodeURIComponent(runId)}/stream`;
}

function buildStreamUrlByThreadId(host: string, threadId: string): string {
  return `${host.replace(/\/+$/, "")}/api/chat/stream?sessionId=${encodeURIComponent(threadId)}`;
}

const CHAT_STREAM_RECONNECT_DELAY_MS = 100;
const CHAT_STREAM_MAX_EMPTY_BACKOFF_MS = 2_000;

function emptyActiveStreamBackoffMs(consecutiveEmptyResponses: number): number {
  const exponent = Math.max(0, Math.min(consecutiveEmptyResponses - 1, 10));
  return Math.min(
    CHAT_STREAM_RECONNECT_DELAY_MS * 2 ** exponent,
    CHAT_STREAM_MAX_EMPTY_BACKOFF_MS,
  );
}

/**
 * Subscribe to the durable chat stream and pretty-print useful chunks
 * until the stream closes. Mirrors the chunk types the chat panel
 * cares about (text deltas, reasoning, tool input/output, finish).
 *
 * Resolves the stream URL from EITHER:
 *   - `runId` (Workflow SDK run id, returned in `x-workflow-run-id`
 *     by `POST /api/chat/cli`), via `/api/chat/[runId]/stream`, OR
 *   - `threadId` (Convex `chatThreads._id`), via
 *     `/api/chat/stream?sessionId=...` which resolves the active run
 *     by following `chatThreads.activeRunId → runs.workflowRunId`.
 *
 * The thread-id path is what `dench chat follow <threadId>` uses, so
 * we can tail any live thread (including a subagent created by
 * `spawn_agent`) without needing to know its workflow run id ahead of
 * time.
 */
async function followStream(args: {
  host: string;
  bearerToken: string;
  runId?: string;
  threadId?: string;
}): Promise<void> {
  const baseUrl = args.runId
    ? buildStreamUrlByRunId(args.host, args.runId)
    : args.threadId
      ? buildStreamUrlByThreadId(args.host, args.threadId)
      : null;
  if (!baseUrl) {
    throw new ChatSpawnError(
      "followStream requires either { runId } or { threadId }",
    );
  }
  // Tool calls in flight, keyed by toolCallId, so we can render
  // a single line per call ("→ tool_name … done") instead of
  // dumping every chunk verbatim.
  const inFlightTools = new Map<string, { toolName: string }>();
  let textBuffer = "";
  let segmentWorkflowRunId: string | null = null;
  let segmentOrdinal: number | null = null;
  let startIndex = 0;
  let consecutiveEmptyActiveResponses = 0;

  const flushText = () => {
    if (textBuffer) {
      process.stdout.write(textBuffer);
      textBuffer = "";
    }
  };

  const closeInFlightToolLines = (suffix: string) => {
    if (inFlightTools.size === 0) return false;
    process.stdout.write(suffix);
    inFlightTools.clear();
    return true;
  };

  while (true) {
    const url = new URL(baseUrl);
    if (startIndex > 0) {
      url.searchParams.set("startIndex", String(startIndex));
    }
    const response: Response = await fetch(url, {
      method: "GET",
      headers: {
        accept: "text/event-stream",
        authorization: `Bearer ${args.bearerToken}`,
        ...(segmentWorkflowRunId
          ? {
              "x-chat-segment-workflow-run-id": segmentWorkflowRunId,
            }
          : {}),
        ...(segmentOrdinal !== null
          ? { "x-chat-segment-ordinal": String(segmentOrdinal) }
          : {}),
      },
    });
    const responseWorkflowRunId: string | null =
      response.headers.get("x-chat-segment-workflow-run-id") ??
      response.headers.get("x-workflow-run-id");
    const rawOrdinal = response.headers.get("x-chat-segment-ordinal");
    const responseOrdinal =
      rawOrdinal !== null && Number.isSafeInteger(Number(rawOrdinal))
        ? Number(rawOrdinal)
        : null;
    const segmentChanged =
      (responseWorkflowRunId !== null &&
        responseWorkflowRunId !== segmentWorkflowRunId) ||
      (responseOrdinal !== null && responseOrdinal !== segmentOrdinal);
    if (segmentChanged) {
      if (segmentWorkflowRunId !== null || segmentOrdinal !== null) {
        closeInFlightToolLines(" continued in next segment\n");
      }
      startIndex = 0;
      consecutiveEmptyActiveResponses = 0;
    }
    segmentWorkflowRunId = responseWorkflowRunId ?? segmentWorkflowRunId;
    segmentOrdinal = responseOrdinal ?? segmentOrdinal;

    if (response.status === 202) {
      const retryAfterSeconds = Number(response.headers.get("retry-after"));
      const retryDelayMs =
        Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0
          ? Math.max(50, retryAfterSeconds * 1_000)
          : 100;
      await new Promise((resolve) => setTimeout(resolve, retryDelayMs));
      continue;
    }
    if (!response.ok || !response.body) {
      throw new ChatSpawnError(
        `Failed to open stream: ${response.status} ${response.statusText}`,
      );
    }

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";
    let responseChunkCount = 0;
    let gotFinish = false;

    const handleChunk = (raw: string) => {
      if (!raw.startsWith("data:")) return;
      const data = raw.slice("data:".length).trim();
      if (!data || data === "[DONE]") return;
      let parsed: unknown;
      try {
        parsed = JSON.parse(data);
      } catch {
        return;
      }
      if (!parsed || typeof parsed !== "object") return;
      responseChunkCount += 1;
      const c = parsed as {
        type?: unknown;
        delta?: unknown;
        text?: unknown;
        toolName?: unknown;
        toolCallId?: unknown;
        input?: unknown;
        output?: unknown;
        errorText?: unknown;
      };
      switch (c.type) {
        case "text-delta":
        case "reasoning-delta": {
          if (typeof c.delta === "string") {
            textBuffer += c.delta;
          }
          break;
        }
        case "tool-input-start": {
          flushText();
          if (
            typeof c.toolCallId === "string" &&
            typeof c.toolName === "string"
          ) {
            inFlightTools.set(c.toolCallId, { toolName: c.toolName });
            process.stdout.write(`\n→ ${c.toolName} …`);
          }
          break;
        }
        case "tool-output-available": {
          flushText();
          if (typeof c.toolCallId === "string") {
            const meta = inFlightTools.get(c.toolCallId);
            inFlightTools.delete(c.toolCallId);
            process.stdout.write(` done (${meta?.toolName ?? "tool"})\n`);
          }
          break;
        }
        case "tool-output-error": {
          flushText();
          if (typeof c.toolCallId === "string") {
            const meta = inFlightTools.get(c.toolCallId);
            inFlightTools.delete(c.toolCallId);
            process.stdout.write(
              ` error (${meta?.toolName ?? "tool"}): ${
                typeof c.errorText === "string" ? c.errorText : "unknown"
              }\n`,
            );
          }
          break;
        }
        case "finish": {
          gotFinish = true;
          flushText();
          if (!closeInFlightToolLines(" ended\n")) {
            process.stdout.write("\n");
          }
          break;
        }
        case "error": {
          flushText();
          const msg =
            typeof c.errorText === "string" ? c.errorText : "stream error";
          process.stderr.write(`\n[error] ${msg}\n`);
          break;
        }
        default:
          break;
      }
    };

    // Read SSE-style chunks. The route uses Vercel AI SDK's
    // `createUIMessageStreamResponse` which emits `data: {…}\n\n`
    // lines, so we split on double newlines.
    while (true) {
      const { value, done } = await reader.read();
      if (done) break;
      buffer += decoder.decode(value, { stream: true });
      let nextBreak = buffer.indexOf("\n\n");
      while (nextBreak !== -1) {
        const event = buffer.slice(0, nextBreak);
        buffer = buffer.slice(nextBreak + 2);
        // Each event may contain multiple `data:` lines.
        for (const line of event.split("\n")) {
          if (line.trim().length > 0) handleChunk(line);
        }
        nextBreak = buffer.indexOf("\n\n");
      }
    }
    flushText();
    startIndex += responseChunkCount;

    const streamStatus = response.headers.get("x-workflow-stream-status");
    const runActive = response.headers.get("X-Run-Active") !== "false";
    if (gotFinish || !runActive || streamStatus === "terminal") {
      closeInFlightToolLines(" ended\n");
      return;
    }
    // Segment rollover intentionally closes without `finish`; reconnect to
    // the thread/run route, which resolves the new segment and resets the
    // durable stream index when the segment identity changes. Empty active
    // responses can occur during that handoff; back them off so a delayed
    // rollover cannot turn into a tight polling loop.
    consecutiveEmptyActiveResponses =
      responseChunkCount === 0 ? consecutiveEmptyActiveResponses + 1 : 0;
    const reconnectDelayMs =
      responseChunkCount === 0
        ? emptyActiveStreamBackoffMs(consecutiveEmptyActiveResponses)
        : CHAT_STREAM_RECONNECT_DELAY_MS;
    await new Promise((resolve) => setTimeout(resolve, reconnectDelayMs));
  }
}

function joinPositionalsExceptFlags(args: string[]): string {
  return args
    .filter((arg) => !arg.startsWith("--"))
    .join(" ")
    .trim();
}

async function runChatNew(ctx: ChatSpawnContext): Promise<void> {
  const environment = consumeFlagValue(ctx.args, "--environment");
  if (environment !== undefined && !/^[a-z][a-z0-9-]{0,63}$/.test(environment))
    throw new ChatSpawnError("Invalid environment name");
  const model = consumeFlagValue(ctx.args, "--model");
  const fileContextPath = consumeFlagValue(ctx.args, "--file-context");
  const visibility = consumeFlagValue(ctx.args, "--visibility");
  const title = consumeFlagValue(ctx.args, "--title");
  const yolo = consumeFlag(ctx.args, "--yolo");
  const follow = consumeFlag(ctx.args, "--follow");
  assertNoUnknownFlags(ctx.args, "chat new");

  const prompt = joinPositionalsExceptFlags(ctx.args);
  if (!prompt) {
    throw new ChatSpawnError(
      'Usage: dench chat new "<prompt>" [--environment NAME] [--model M] [--file-context PATH] [--visibility private|shared] [--title T] [--yolo] [--follow] [--json]',
    );
  }

  const result = await postJson(
    buildSpawnUrl(ctx.runtime.host),
    ctx.runtime.bearerToken,
    {
      prompt,
      ...(environment ? { environment } : {}),
      model,
      fileContextPath,
      visibility,
      title,
      yolo,
      timezone: getCurrentRuntimeTimeZone(),
    },
  );
  if (!result.ok) {
    const errorMessage =
      (result.payload as { error?: string } | null)?.error ??
      `Spawn failed (${result.status})`;
    throw new ChatSpawnError(errorMessage);
  }

  const payload = result.payload as {
    threadId?: string;
    runId?: string;
    workflowRunId?: string;
    threadUrl?: string;
    streamUrl?: string;
  } | null;
  out(ctx, payload);

  if (follow && (payload?.threadId || payload?.workflowRunId)) {
    if (!ctx.jsonOutput) console.error("\n--- following stream ---");
    await followStream({
      host: ctx.runtime.host,
      bearerToken: ctx.runtime.bearerToken,
      ...(payload.threadId
        ? { threadId: payload.threadId }
        : { runId: payload.workflowRunId }),
    });
  }
}

async function runChatSend(ctx: ChatSpawnContext): Promise<void> {
  const yolo = consumeFlag(ctx.args, "--yolo");
  const follow = consumeFlag(ctx.args, "--follow");
  const model = consumeFlagValue(ctx.args, "--model");

  // First positional = threadId, the rest joined = prompt.
  const positionals = ctx.args.filter((arg) => !arg.startsWith("--"));
  const threadId = positionals.shift();
  if (!threadId) {
    throw new ChatSpawnError(
      'Usage: dench chat send <threadId> "<message>" [--model M] [--yolo] [--follow] [--json]',
    );
  }
  const prompt = positionals.join(" ").trim();
  if (!prompt) {
    throw new ChatSpawnError(
      'Missing message body. Usage: dench chat send <threadId> "<message>"',
    );
  }

  const result = await postJson(
    buildSpawnUrl(ctx.runtime.host),
    ctx.runtime.bearerToken,
    {
      threadId,
      prompt,
      model,
      yolo,
      timezone: getCurrentRuntimeTimeZone(),
    },
  );
  if (!result.ok) {
    const errorMessage =
      (result.payload as { error?: string } | null)?.error ??
      `Send failed (${result.status})`;
    throw new ChatSpawnError(errorMessage);
  }

  const payload = result.payload as {
    threadId?: string;
    runId?: string;
    workflowRunId?: string;
    threadUrl?: string;
    streamUrl?: string;
  } | null;
  out(ctx, payload);

  if (follow && (payload?.threadId || payload?.workflowRunId)) {
    if (!ctx.jsonOutput) console.error("\n--- following stream ---");
    await followStream({
      host: ctx.runtime.host,
      bearerToken: ctx.runtime.bearerToken,
      ...(payload.threadId
        ? { threadId: payload.threadId }
        : { runId: payload.workflowRunId }),
    });
  }
}

/**
 * `dench chat follow <threadId>` — tail the live stream of an
 * existing chat thread by Convex thread id.
 *
 * Useful for watching a subagent (a child thread spawned by a parent
 * via `spawn_agent`) live, or for resuming the stream of any chat
 * thread that's currently in flight. Resolves the active workflow run
 * server-side via `chatThreads.activeRunId → runs.workflowRunId`, so
 * the caller doesn't need to know any workflow ids.
 *
 * Exits when the stream closes (workflow finishes) or when the user
 * hits Ctrl-C.
 */
async function runChatFollow(ctx: ChatSpawnContext): Promise<void> {
  const positionals = ctx.args.filter((arg) => !arg.startsWith("--"));
  const threadId = positionals.shift();
  if (!threadId) {
    throw new ChatSpawnError("Usage: dench chat follow <threadId> [--json]");
  }
  if (!ctx.jsonOutput) {
    console.error(`--- following stream for thread ${threadId} ---`);
  }
  await followStream({
    host: ctx.runtime.host,
    bearerToken: ctx.runtime.bearerToken,
    threadId,
  });
}

/**
 * Public entry point invoked by the chat dispatcher in cli/dench.ts
 * (and the agent dispatcher when the user runs `dench agent new` /
 * `dench agent send` aliases). The CLI dispatcher resolves the
 * runtime + bearer token once and hands them in.
 */
export async function runChatSpawnCommand(opts: {
  runtime: RuntimeBundle;
  args: string[];
  subcommand: "new" | "send" | "follow";
}): Promise<void> {
  const args = [...opts.args];
  const jsonOutput = hasFlag(args, "--json");
  const ctx: ChatSpawnContext = {
    runtime: opts.runtime,
    args,
    jsonOutput,
  };
  if (opts.subcommand === "new") {
    await runChatNew(ctx);
    return;
  }
  if (opts.subcommand === "follow") {
    await runChatFollow(ctx);
    return;
  }
  await runChatSend(ctx);
}

export { ChatSpawnError, followStream };