import { setTimeout as delay } from 'node:timers/promises';
import type { PublicUsage } from '../../shared/coding-conversation-contracts';
import type { TeacherTopic } from '../../shared/coding-teacher';
import { proxyAwareFetch } from '../utils/proxy-fetch';
import {
  assertTeacherAccount,
  teacherCloudRequest,
  TeacherError,
  type TeacherAccount,
} from './config-client';
import type { TeacherModelMessage } from './context';
import { createTeacherReadTools, type TeacherReadAccess } from './read-tools';

interface TeacherSession {
  access_token: string;
  expires_at: number;
  api_base_url: string;
  scope: string;
}
type Json = Record<string, unknown>;
function object(value: unknown): Json {
  if (!value || typeof value !== 'object' || Array.isArray(value))
    throw new TeacherError(502, 'teacher_protocol_invalid', '老师服务返回的数据无效。');
  return value as Json;
}
function identifier(value: unknown): string {
  if (typeof value !== 'string' || !value)
    throw new TeacherError(502, 'teacher_protocol_invalid', '老师服务未返回有效的运行标识。');
  return value;
}
export interface TeacherCloudTransport {
  json(path: string, body?: unknown, signal?: AbortSignal): Promise<Json>;
  events(
    path: string,
    signal: AbortSignal,
    accept: (event: string, data: Json, id: string) => void
  ): Promise<void>;
}

export function teacherCloudTransport(account: TeacherAccount): TeacherCloudTransport {
  let session: TeacherSession | undefined;
  const fetchCloud = async (path: string, body?: unknown, signal?: AbortSignal) => {
    // 退出账号后只允许用已经持有的短凭据停止原问题，不能再读状态或发送内容。
    const stopping = /^\/questions\/[^/]+\/cancel$/.test(path) && body !== undefined;
    for (let attempt = 0; attempt < 3; attempt++) {
      if (!stopping || !session) assertTeacherAccount(account);
      signal?.throwIfAborted();
      if (!session || (!stopping && session.expires_at * 1000 < Date.now() + 10000)) {
        session = await teacherCloudRequest<TeacherSession>(
          account,
          '/api/coding-teacher/session',
          'POST'
        );
        if (
          session.scope !== 'makelore-teachers' ||
          typeof session.access_token !== 'string' ||
          !session.access_token ||
          !Number.isFinite(session.expires_at) ||
          session.expires_at * 1000 <= Date.now() ||
          typeof session.api_base_url !== 'string' ||
          !/^https?:\/\//.test(session.api_base_url)
        ) {
          throw new TeacherError(502, 'teacher_session_invalid', '老师接入凭据无效，请重新连接。');
        }
      }
      if (!stopping) assertTeacherAccount(account);
      try {
        const response = await proxyAwareFetch(
          session.api_base_url.replace(/\/+$/, '') + '/api/makelore/teachers' + path,
          {
            method: body === undefined ? 'GET' : 'POST',
            headers: {
              Authorization: 'Bearer ' + session.access_token,
              'Content-Type': 'application/json',
            },
            body: body === undefined ? undefined : JSON.stringify(body),
            signal: AbortSignal.any([...(signal ? [signal] : []), AbortSignal.timeout(45000)]),
          }
        );
        if (!stopping) assertTeacherAccount(account);
        if (response.status === 401 && attempt < 2) {
          await response.body?.cancel();
          session = undefined;
          continue;
        }
        if (response.status >= 500 && attempt < 2) {
          await response.body?.cancel();
          await delay(500, undefined, { signal });
          continue;
        }
        if (!response.ok) {
          const data = object(await response.json());
          const detail = data.detail;
          const message =
            typeof detail === 'string'
              ? detail
              : detail && typeof detail === 'object'
                ? object(detail).message
                : undefined;
          throw new TeacherError(
            response.status,
            'teacher_cloud_failed',
            typeof message === 'string' ? message : '老师服务暂不可用，请稍后重试。'
          );
        }
        return response;
      } catch (error) {
        signal?.throwIfAborted();
        if (error instanceof TeacherError || attempt === 2) throw error;
        await delay(500, undefined, { signal });
      }
    }
    throw new TeacherError(502, 'teacher_connection_failed', '连接老师失败。');
  };
  return {
    async json(path, body, signal) {
      const response = await fetchCloud(path, body, signal);
      const data = object(await response.json());
      if (!path.endsWith('/cancel')) assertTeacherAccount(account);
      return data;
    },
    async events(path, signal, accept) {
      const response = await fetchCloud(path, undefined, signal);
      if (!response.body)
        throw new TeacherError(502, 'teacher_stream_missing', '老师回复连接不可用。');
      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer = '';
      try {
        while (true) {
          const next = await reader.read();
          if (next.done) break;
          assertTeacherAccount(account);
          signal.throwIfAborted();
          buffer += decoder.decode(next.value, { stream: true }).replaceAll('\r', '');
          let end: number;
          while ((end = buffer.indexOf('\n\n')) >= 0) {
            const block = buffer.slice(0, end);
            buffer = buffer.slice(end + 2);
            let event = 'message',
              id = '';
            const data: string[] = [];
            for (const line of block.split('\n')) {
              if (line.startsWith('event:')) event = line.slice(6).trim();
              else if (line.startsWith('id:')) id = line.slice(3).trim();
              else if (line.startsWith('data:')) data.push(line.slice(5).trimStart());
            }
            if (data.length) accept(event, object(JSON.parse(data.join('\n'))), id);
          }
        }
      } finally {
        await reader.cancel().catch(() => undefined);
        reader.releaseLock();
      }
    },
  };
}

export function prepareCloudTeacher(
  account: TeacherAccount,
  topic: TeacherTopic,
  requestId: string,
  access: TeacherReadAccess,
  onProgress: (text: string) => void,
  saveRequestId: (id: string) => Promise<void>,
  transport: TeacherCloudTransport = teacherCloudTransport(account)
) {
  const tools = createTeacherReadTools(access);
  return {
    inputLimit: topic.definition.limits.max_input_tokens,
    async run(
      messages: TeacherModelMessage[],
      signal: AbortSignal,
      onText: (delta: string) => void
    ): Promise<PublicUsage | undefined> {
      const localContext = {
        id: requestId,
        scope: { project_id: topic.projectId, source_session_id: topic.sourceConversationId },
        tools: tools.definitions.map((item) => item.function.name),
      };
      const deadline = Date.now() + 60 * 60 * 1000;
      const bounded = AbortSignal.any([signal, AbortSignal.timeout(60 * 60 * 1000)]);
      let questionId: string | undefined;
      let completed = false;
      try {
        // 重启后本地快照已经丢失，先停止旧请求，再以新的问题建立读取作用域。
        const previous = topic.requests.filter((item) => item.id !== requestId).at(-1);
        if (previous && previous.status !== 'completed') {
          try {
            await transport.json(
              '/questions/' + encodeURIComponent(previous.id) + '/cancel',
              {},
              bounded
            );
          } catch (error) {
            if (!(error instanceof TeacherError && error.status === 404)) throw error;
          }
        }
        let queued = await transport.json(
          '/questions',
          {
            teacher_version: topic.version,
            thread_id: topic.id,
            request_id: requestId,
            query: messages
              .filter((item) => item.role === 'user')
              .map((item) => item.content)
              .join('\n\n'),
            local_context: localContext,
          },
          bounded
        );
        questionId = identifier(queued.request_id);
        await saveRequestId(questionId);
        while (!queued.run_id) {
          if (!['queued', 'pending'].includes(String(queued.status)))
            throw new TeacherError(
              409,
              'teacher_request_stopped',
              '老师提问未能启动，请重新提问。'
            );
          onProgress('正在等待老师…');
          await delay(800, undefined, { signal: bounded });
          queued = await transport.json(
            '/requests/' + encodeURIComponent(questionId),
            undefined,
            bounded
          );
        }
        let runId = identifier(queued.run_id),
          cursor = '0-0',
          rounds = 0;
        let runText = '',
          messageId = '';
        while (Date.now() < deadline) {
          bounded.throwIfAborted();
          access.assertCurrent();
          const view = await transport.json(
            '/runs/' + encodeURIComponent(runId),
            undefined,
            bounded
          );
          if (view.continued_run_id) {
            runId = identifier(view.continued_run_id);
            cursor = '0-0';
            runText = '';
            messageId = '';
            continue;
          }
          if (view.status === 'interrupted') {
            const pending = object(view.interrupt);
            if (pending.source !== 'client_read_tools')
              throw new TeacherError(
                409,
                'teacher_interaction_unsupported',
                '老师请求了当前面板不支持的交互，请联系运营调整该智能体。'
              );
            if (
              pending.context_id !== requestId ||
              !Array.isArray(pending.calls) ||
              !pending.calls.length ||
              pending.calls.length > 32 ||
              ++rounds > 6
            )
              throw new TeacherError(
                409,
                'teacher_context_expired',
                '老师读取请求已失效，请重新提问。'
              );
            onProgress('正在读取项目与会话…');
            const results = [];
            for (const raw of pending.calls) {
              const call = object(raw);
              if (typeof call.tool_call_id !== 'string' || typeof call.name !== 'string')
                throw new TeacherError(502, 'teacher_protocol_invalid', '老师读取请求无效。');
              results.push({
                tool_call_id: call.tool_call_id,
                ...(await tools.executeResult(call.name, JSON.stringify(call.arguments), bounded)),
              });
            }
            // POST 重试使用完全相同的结果，文件变化也不会导致重复续接或不同输入。
            const resumed = await transport.json(
              '/runs/' + encodeURIComponent(runId) + '/tool-results',
              { context_id: requestId, results },
              bounded
            );
            runId = identifier(resumed.run_id);
            cursor = '0-0';
            runText = '';
            messageId = '';
            onProgress('老师正在继续思考…');
            continue;
          }
          if (view.status === 'completed') {
            const output = typeof view.output === 'string' ? view.output : '';
            if (output.startsWith(runText)) onText(output.slice(runText.length));
            else if (output) onText('\n\n' + output);
            completed = true;
            onProgress('');
            // Yuxi 的账本记录每个模型调用；不把线程累计 token 当作本问题费用。
            return undefined;
          }
          if (['failed', 'cancelled'].includes(String(view.status))) {
            const detail = view.error ? object(view.error).message : undefined;
            throw new TeacherError(
              409,
              'teacher_run_failed',
              typeof detail === 'string' ? detail : '老师回复已停止。'
            );
          }
          const threadId = identifier(view.thread_id);
          onProgress('老师正在思考…');
          try {
            await transport.events(
              '/runs/' +
                encodeURIComponent(runId) +
                '/events?after_seq=' +
                encodeURIComponent(cursor),
              bounded,
              (_event, envelope, id) => {
                if (id) cursor = id;
                // Yuxi 的父 Run 也包含子线程事件，老师正文只接收云端主线程文本。
                if (envelope.thread_id !== threadId) return;
                const payload = envelope.payload ? object(envelope.payload) : {};
                for (const item of Array.isArray(payload.items)
                  ? payload.items
                  : payload.chunk
                    ? [payload.chunk]
                    : []) {
                  const chunk = object(item);
                  const event = chunk.stream_event ? object(chunk.stream_event) : {};
                  if (event.type === 'message_delta' && typeof event.content === 'string') {
                    if (typeof event.message_id === 'string' && event.message_id !== messageId) {
                      if (messageId && runText) onText('\n\n');
                      messageId = event.message_id;
                      runText = '';
                    }
                    runText += event.content;
                    onText(event.content);
                  }
                }
              }
            );
          } catch (error) {
            bounded.throwIfAborted();
            access.assertCurrent();
            if (error instanceof TeacherError && error.status < 500) throw error;
            onProgress('连接中断，正在恢复老师回复…');
          }
          await delay(400, undefined, { signal: bounded });
        }
        throw new TeacherError(408, 'teacher_question_expired', '本次老师提问已超时，请重新提问。');
      } finally {
        if (!completed) {
          await transport
            .json(
              '/questions/' + encodeURIComponent(requestId) + '/cancel',
              {},
              AbortSignal.timeout(10000)
            )
            .catch(() => undefined);
        }
      }
    },
  };
}