import "server-only";

import {
  appendPersistedClawdTaskProgress,
  type ClawdUpstreamTaskState,
  persistClawdUpstreamTaskState,
} from "@/lib/openclaw-chat/task-store";
import type { ClawdTaskMetadata } from "@/lib/openclaw-chat/task-state";
import { decryptOpenClawTaskSecret } from "@/lib/openclaw-chat/task-secret";

type OpenClawUpstreamConfig = {
  baseUrl: string;
  token: string;
  timeoutMs: number;
};

type ClawdTaskPersistenceTarget = {
  chatId: string;
  assistantMessageId: string;
};

export type OpenClawTaskOutcome =
  | {
      status: "pending";
      upstreamTask: ClawdUpstreamTaskState;
    }
  | {
      status: "completed";
      upstreamTask: ClawdUpstreamTaskState;
      answer: string;
      requestId: string | null;
    }
  | {
      status: "failed" | "canceled";
      upstreamTask: ClawdUpstreamTaskState;
      error: string;
    };

const UPSTREAM_REQUEST_TIMEOUT_MS = 15_000;
const UPSTREAM_POLL_INTERVAL_MS = 1600;
const TERMINAL_STATUSES = new Set([
  "succeeded",
  "completed",
  "failed",
  "cancelled",
  "canceled",
]);

class OpenClawUpstreamHttpError extends Error {
  constructor(
    message: string,
    readonly status: number
  ) {
    super(message);
    this.name = "OpenClawUpstreamHttpError";
  }
}

function readRecord(value: unknown): Record<string, unknown> {
  return value && typeof value === "object" && !Array.isArray(value)
    ? (value as Record<string, unknown>)
    : {};
}

function readText(value: unknown) {
  return typeof value === "string" ? value.trim() : "";
}

function readEventId(value: unknown) {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? Math.max(0, Math.trunc(parsed)) : 0;
}

function getTaskStatus(payload: unknown) {
  return readText(readRecord(readRecord(payload).task).status).toLowerCase();
}

function getTaskEventsEndpoint(payload: unknown, taskId: string) {
  const task = readRecord(readRecord(payload).task);
  return (
    readText(task.events_endpoint) ||
    readText(readRecord(payload).events_endpoint) ||
    `/api/chat/tasks/${encodeURIComponent(taskId)}/events`
  );
}

function getAnswer(payload: unknown) {
  const root = readRecord(payload);
  const output = readRecord(root.output);
  const result = readRecord(root.result);
  return (
    readText(root.answer) || readText(output.answer) || readText(result.answer)
  );
}

function getRequestId(payload: unknown) {
  const root = readRecord(payload);
  const output = readRecord(root.output);
  return readText(root.request_id) || readText(output.request_id) || null;
}

function getTaskError(payload: unknown, fallback: string) {
  const root = readRecord(payload);
  const error = readRecord(root.error);
  const outputError = readRecord(readRecord(root.output).error);
  return (
    readText(error.message) ||
    readText(outputError.message) ||
    readText(root.message) ||
    fallback
  );
}

function buildUpstreamUrl(baseUrl: string, endpoint: string) {
  const normalizedBaseUrl = `${baseUrl.replace(/\/+$/, "")}/`;
  return new URL(endpoint, normalizedBaseUrl);
}

function buildTaskEndpoint(taskId: string) {
  return `/api/chat/tasks/${encodeURIComponent(taskId)}`;
}

function buildTaskControlEndpoint(taskId: string) {
  return `${buildTaskEndpoint(taskId)}/control`;
}

function createRequestController(parentSignal?: AbortSignal) {
  const controller = new AbortController();
  const abortFromParent = () => controller.abort(parentSignal?.reason);
  if (parentSignal?.aborted) {
    abortFromParent();
  } else {
    parentSignal?.addEventListener("abort", abortFromParent, { once: true });
  }
  const timeout = setTimeout(
    () => controller.abort(new DOMException("OpenClaw request timed out", "TimeoutError")),
    UPSTREAM_REQUEST_TIMEOUT_MS
  );
  return {
    signal: controller.signal,
    cleanup() {
      clearTimeout(timeout);
      parentSignal?.removeEventListener("abort", abortFromParent);
    },
  };
}

async function fetchJson(params: {
  config: OpenClawUpstreamConfig;
  endpoint: string;
  method?: "GET" | "POST";
  body?: Record<string, unknown>;
  taskKey?: string | null;
  idempotencyKey?: string;
  signal?: AbortSignal;
}) {
  const requestController = createRequestController(params.signal);
  try {
    const response = await fetch(
      buildUpstreamUrl(params.config.baseUrl, params.endpoint),
      {
        method: params.method || "GET",
        headers: {
          Authorization: `Bearer ${params.config.token}`,
          Accept: "application/json",
          ...(params.body ? { "Content-Type": "application/json" } : {}),
          ...(params.taskKey
            ? { "X-Dalongxia-Task-Key": params.taskKey }
            : {}),
          ...(params.idempotencyKey
            ? { "Idempotency-Key": params.idempotencyKey }
            : {}),
        },
        body: params.body ? JSON.stringify(params.body) : undefined,
        cache: "no-store",
        signal: requestController.signal,
      }
    );
    const payload = (await response.json().catch(() => null)) as unknown;
    if (!response.ok) {
      throw new OpenClawUpstreamHttpError(
        getTaskError(
          payload,
          `OpenClaw request failed: ${response.status}`
        ),
        response.status
      );
    }
    return payload;
  } finally {
    requestController.cleanup();
  }
}

function mergeUpstreamTaskState(
  current: ClawdUpstreamTaskState,
  payload: unknown
) {
  const root = readRecord(payload);
  const task = readRecord(root.task);
  const taskId =
    readText(root.task_id) || readText(task.task_id) || current.taskId;
  const eventId = Math.max(
    current.lastEventId || 0,
    readEventId(root.next_after_event_id),
    readEventId(root.event_id)
  );
  return {
    taskId,
    taskKey: readText(root.task_key) || current.taskKey || null,
    eventsEndpoint:
      getTaskEventsEndpoint(payload, taskId) || current.eventsEndpoint || null,
    lastEventId: eventId,
  } satisfies ClawdUpstreamTaskState;
}

function normalizeProgressEvent(
  event: unknown,
  taskId: string
): Record<string, unknown> | null {
  const raw = readRecord(event);
  if (Object.keys(raw).length === 0) return null;
  const data = readRecord(raw.data);
  const eventId = readEventId(raw.event_id);
  const phase = readText(raw.phase);
  const eventType = readText(raw.event_type);
  const message =
    readText(data.message) ||
    readText(data.display) ||
    readText(data.display_text) ||
    readText(raw.display_text) ||
    phase ||
    "任务状态已更新";
  return {
    ...data,
    event_id: eventId,
    task_id: taskId,
    event_type: eventType,
    status: readText(raw.status),
    phase,
    step_code: readText(data.step_code) || phase || eventType || "task.progress",
    step_state: readText(data.step_state) || readText(raw.status) || "started",
    message,
    display: message,
    created_at: raw.created_at,
  };
}

export function getPersistedUpstreamTaskState(
  metadata: ClawdTaskMetadata
): ClawdUpstreamTaskState | null {
  const taskId = readText(metadata.upstreamTaskId);
  if (!taskId) return null;
  return {
    taskId,
    taskKey: metadata.upstreamTaskKey
      ? decryptOpenClawTaskSecret(metadata.upstreamTaskKey)
      : null,
    eventsEndpoint:
      readText(metadata.upstreamEventsEndpoint) ||
      `/api/chat/tasks/${encodeURIComponent(taskId)}/events`,
    lastEventId: readEventId(metadata.upstreamLastEventId),
  };
}

export async function submitOpenClawDurableTask(params: {
  config: OpenClawUpstreamConfig;
  body: Record<string, unknown>;
  idempotencyKey: string;
  target: ClawdTaskPersistenceTarget;
  signal?: AbortSignal;
}) {
  let payload: unknown;
  for (let attempt = 1; attempt <= 3; attempt += 1) {
    try {
      payload = await fetchJson({
        config: params.config,
        endpoint: "/api/chat/tasks",
        method: "POST",
        body: params.body,
        idempotencyKey: params.idempotencyKey,
        signal: params.signal,
      });
      break;
    } catch (error) {
      if (params.signal?.aborted) throw error;
      if (
        error instanceof OpenClawUpstreamHttpError &&
        error.status >= 400 &&
        error.status < 500 &&
        error.status !== 408 &&
        error.status !== 429
      ) {
        throw error;
      }
      if (attempt >= 3) throw error;
      await waitForDelay(attempt * 500, params.signal);
    }
  }
  const root = readRecord(payload);
  const task = readRecord(root.task);
  const taskId = readText(root.task_id) || readText(task.task_id);
  if (!taskId) throw new Error("OpenClaw 未返回 task_id");
  const submittedTask = mergeUpstreamTaskState(
    {
      taskId,
      taskKey: readText(root.task_key) || null,
      eventsEndpoint: getTaskEventsEndpoint(payload, taskId),
      lastEventId: 0,
    },
    payload
  );
  // A task-creation response can include the latest upstream cursor even though
  // none of its events have been persisted locally yet. Always consume the
  // event stream from the beginning after accepting a new durable task.
  const upstreamTask = { ...submittedTask, lastEventId: 0 };
  await persistClawdUpstreamTaskState({
    ...params.target,
    upstreamTask,
  });
  return upstreamTask;
}

export async function stopOpenClawDurableTask(params: {
  config: OpenClawUpstreamConfig;
  upstreamTask: ClawdUpstreamTaskState;
  signal?: AbortSignal;
}) {
  return fetchJson({
    config: params.config,
    endpoint: buildTaskControlEndpoint(params.upstreamTask.taskId),
    method: "POST",
    body: { action: "stop" },
    taskKey: params.upstreamTask.taskKey,
    signal: params.signal,
  });
}

export async function refreshOpenClawDurableTask(params: {
  config: OpenClawUpstreamConfig;
  upstreamTask: ClawdUpstreamTaskState;
  target: ClawdTaskPersistenceTarget;
  signal?: AbortSignal;
}): Promise<OpenClawTaskOutcome> {
  const eventsUrl = buildUpstreamUrl(
    params.config.baseUrl,
    params.upstreamTask.eventsEndpoint ||
      `/api/chat/tasks/${encodeURIComponent(params.upstreamTask.taskId)}/events`
  );
  eventsUrl.searchParams.set(
    "after_event_id",
    String(params.upstreamTask.lastEventId || 0)
  );
  const eventsPayload = await fetchJson({
    config: params.config,
    endpoint: eventsUrl.toString(),
    taskKey: params.upstreamTask.taskKey,
    signal: params.signal,
  });
  const mergedUpstreamTask = mergeUpstreamTaskState(
    params.upstreamTask,
    eventsPayload
  );
  let upstreamTask = {
    ...mergedUpstreamTask,
    lastEventId: params.upstreamTask.lastEventId || 0,
  };
  const events = Array.isArray(readRecord(eventsPayload).events)
    ? (readRecord(eventsPayload).events as unknown[])
    : [];
  for (const event of events) {
    const eventId = readEventId(readRecord(event).event_id);
    upstreamTask = {
      ...upstreamTask,
      lastEventId: Math.max(upstreamTask.lastEventId || 0, eventId),
    };
    const progress = normalizeProgressEvent(event, upstreamTask.taskId);
    if (!progress) continue;
    await appendPersistedClawdTaskProgress({
      ...params.target,
      progress,
      upstreamTask,
    });
  }
  upstreamTask = {
    ...upstreamTask,
    lastEventId: Math.max(
      upstreamTask.lastEventId || 0,
      readEventId(readRecord(eventsPayload).next_after_event_id)
    ),
  };
  await persistClawdUpstreamTaskState({
    ...params.target,
    upstreamTask,
  });

  const status = getTaskStatus(eventsPayload);
  if (!TERMINAL_STATUSES.has(status)) {
    return { status: "pending", upstreamTask };
  }

  const terminalPayload = await fetchJson({
    config: params.config,
    endpoint: buildTaskEndpoint(upstreamTask.taskId),
    taskKey: upstreamTask.taskKey,
    signal: params.signal,
  });
  upstreamTask = mergeUpstreamTaskState(upstreamTask, terminalPayload);
  const terminalStatus = getTaskStatus(terminalPayload) || status;
  if (terminalStatus === "succeeded" || terminalStatus === "completed") {
    const answer = getAnswer(terminalPayload);
    if (!answer) {
      return {
        status: "failed",
        upstreamTask,
        error: "OpenClaw 任务已结束，但未返回有效结果",
      };
    }
    return {
      status: "completed",
      upstreamTask,
      answer,
      requestId: getRequestId(terminalPayload),
    };
  }
  if (terminalStatus === "cancelled" || terminalStatus === "canceled") {
    return { status: "canceled", upstreamTask, error: "任务已停止" };
  }
  return {
    status: "failed",
    upstreamTask,
    error: getTaskError(terminalPayload, "OpenClaw 任务执行失败"),
  };
}

function waitForDelay(delayMs: number, signal?: AbortSignal) {
  return new Promise<void>((resolve, reject) => {
    if (signal?.aborted) {
      reject(signal.reason || new DOMException("Aborted", "AbortError"));
      return;
    }
    let timer: ReturnType<typeof setTimeout>;
    const onAbort = () => {
      clearTimeout(timer);
      reject(signal?.reason || new DOMException("Aborted", "AbortError"));
    };
    timer = setTimeout(() => {
      signal?.removeEventListener("abort", onAbort);
      resolve();
    }, delayMs);
    signal?.addEventListener("abort", onAbort, { once: true });
  });
}

export class OpenClawTaskDetachedError extends Error {
  constructor() {
    super("OpenClaw task continues in the background");
    this.name = "OpenClawTaskDetachedError";
  }
}

export async function waitForOpenClawDurableTask(params: {
  config: OpenClawUpstreamConfig;
  upstreamTask: ClawdUpstreamTaskState;
  target: ClawdTaskPersistenceTarget;
  deadlineAt: number;
  signal?: AbortSignal;
}) {
  let upstreamTask = params.upstreamTask;
  let failureCount = 0;
  while (!params.signal?.aborted && Date.now() < params.deadlineAt) {
    try {
      const outcome = await refreshOpenClawDurableTask({
        config: params.config,
        upstreamTask,
        target: params.target,
        signal: params.signal,
      });
      upstreamTask = outcome.upstreamTask;
      failureCount = 0;
      if (outcome.status !== "pending") return outcome;
    } catch (error) {
      if (params.signal?.aborted) throw error;
      if (
        error instanceof OpenClawUpstreamHttpError &&
        error.status >= 400 &&
        error.status < 500 &&
        error.status !== 408 &&
        error.status !== 429
      ) {
        throw error;
      }
      failureCount += 1;
      await appendPersistedClawdTaskProgress({
        ...params.target,
        upstreamTask,
        progress: {
          step_code: "task.reconnecting",
          step_state: "waiting",
          message: "任务仍在 OpenClaw 后台运行，正在重新连接进度服务",
          display: "任务仍在 OpenClaw 后台运行，正在重新连接进度服务",
        },
      });
    }
    await waitForDelay(
      Math.min(5000, UPSTREAM_POLL_INTERVAL_MS + failureCount * 500),
      params.signal
    );
  }
  if (params.signal?.aborted) {
    throw params.signal.reason || new DOMException("Aborted", "AbortError");
  }
  throw new OpenClawTaskDetachedError();
}