/**
 * Task Completer - Handles accept, start, complete, cancel operations
 */

import type { Task, TaskCompleteRequest } from '@clawteam/shared/types';
import { randomUUID } from 'node:crypto';
import type { IMessageBus } from '@clawteam/api/message-bus';
import type { DatabasePool, RedisClient, Logger } from '@clawteam/api/common';
import { taskRowToTask, type TaskRow } from './types';
import {
  CoordinatorError,
  TaskNotFoundError,
  TaskAlreadyAcceptedError,
  UnauthorizedTaskError,
  InvalidTaskStateError,
} from './errors';
import { PRIORITY_ORDER, REDIS_KEYS } from './constants';
import { tasksCompletedTotal, tasksCancelledTotal, taskDuration } from './metrics';

export interface TaskCompleterDeps {
  db: DatabasePool;
  redis: RedisClient;
  messageBus: IMessageBus;
  logger: Logger;
}

export class TaskCompleter {
  private readonly db: DatabasePool;
  private readonly redis: RedisClient;
  private readonly messageBus: IMessageBus;
  private readonly logger: Logger;

  constructor(deps: TaskCompleterDeps) {
    this.db = deps.db;
    this.redis = deps.redis;
    this.messageBus = deps.messageBus;
    this.logger = deps.logger;
  }

  async accept(taskId: string, botId: string, executorSessionKey?: string): Promise<void> {
    const task = await this.loadTask(taskId);

    if (task.toBotId !== botId) {
      throw new UnauthorizedTaskError(taskId, botId);
    }
    if (task.status !== 'pending') {
      throw new TaskAlreadyAcceptedError(taskId);
    }

    const now = new Date();

    // Merge accept + start: go directly from pending → processing
    const updateResult = executorSessionKey
      ? await this.db.query(
          `UPDATE tasks SET status = 'processing', accepted_at = $1, started_at = $1, executor_session_key = $3
           WHERE id = $2 AND status = 'pending'`,
          [now, taskId, executorSessionKey]
        )
      : await this.db.query(
          `UPDATE tasks SET status = 'processing', accepted_at = $1, started_at = $1
           WHERE id = $2 AND status = 'pending'`,
          [now, taskId]
        );

    if (updateResult.rowCount === 0) {
      throw new InvalidTaskStateError(taskId, task.status, ['pending']);
    }

    // Remove from queue
    await this.removeFromQueue(taskId, task.toBotId);

    // Add to processing ZSET for timeout detection
    const timeoutAt = now.getTime() + task.timeoutSeconds * 1000;
    await this.redis.zadd(REDIS_KEYS.PROCESSING_SET, timeoutAt.toString(), taskId);

    // Update cache
    await this.updateCacheStatus(taskId, 'processing');

    // Notify originating bot
    await this.safePublish('task_assigned', {
      taskId,
      status: 'processing',
      acceptedAt: now.toISOString(),
    }, task.fromBotId);

    this.logger.info('Task accepted and started', { taskId, botId });
  }

  async complete(taskId: string, req: TaskCompleteRequest, botId: string): Promise<void> {
    const task = await this.loadTask(taskId);

    // Allow fromBotId to fail tasks (e.g. recovery loop), but only toBotId can complete successfully
    const isFailing = req.status === 'failed' || (!req.status && req.error);
    if (task.toBotId !== botId && !(isFailing && task.fromBotId === botId)) {
      throw new UnauthorizedTaskError(taskId, botId);
    }
    const validStates = ['accepted', 'processing', 'waiting_for_input'];
    if (!validStates.includes(task.status)) {
      throw new InvalidTaskStateError(taskId, task.status, validStates);
    }

    const now = new Date();
    // 智能推断最终状态：
    // 1. 如果明确指定了 status，使用指定的值
    // 2. 如果有 error 且没有指定 status，默认为 'failed'
    // 3. 如果有 result 且没有 error，默认为 'completed'
    const finalStatus = req.status
      ? req.status
      : req.error
        ? 'failed'
        : 'completed';

    const updateResult = await this.db.query(
      `UPDATE tasks
       SET status = $1, result = $2, error = $3, completed_at = $4
       WHERE id = $5 AND status IN ('accepted', 'processing', 'waiting_for_input')`,
      [
        finalStatus,
        req.result ? JSON.stringify(req.result) : null,
        req.error ? JSON.stringify(req.error) : null,
        now,
        taskId,
      ]
    );

    if (updateResult.rowCount === 0) {
      // Task was cancelled or otherwise changed between loadTask and UPDATE
      this.logger.warn('Task complete rejected — status changed (likely cancelled)', { taskId, loadedStatus: task.status });
      throw new InvalidTaskStateError(taskId, task.status, validStates);
    }

    // Record task completion metrics
    tasksCompletedTotal.inc({
      status: finalStatus,
      capability: task.capability,
    });

    // Record task duration
    const durationSeconds = (now.getTime() - new Date(task.createdAt).getTime()) / 1000;
    taskDuration.observe(
      { capability: task.capability, status: finalStatus },
      durationSeconds
    );

    // Remove from processing ZSET
    await this.redis.zrem(REDIS_KEYS.PROCESSING_SET, taskId);

    // Clear cache
    await this.redis.del(`${REDIS_KEYS.TASK_CACHE}:${taskId}`);

    // Notify originating bot
    const event = finalStatus === 'completed' ? 'task_completed' : 'task_failed';
    await this.safePublish(event, {
      taskId,
      status: finalStatus,
      result: req.result,
      error: req.error,
    }, task.fromBotId);

    this.logger.info('Task completed', { taskId, botId, status: finalStatus });
  }

  async reset(taskId: string, botId: string): Promise<void> {
    const task = await this.loadTask(taskId);

    if (task.toBotId !== botId) {
      throw new UnauthorizedTaskError(taskId, botId);
    }
    const resetableStates = ['accepted', 'processing', 'waiting_for_input'];
    if (!resetableStates.includes(task.status)) {
      throw new InvalidTaskStateError(taskId, task.status, resetableStates);
    }
    if (task.retryCount >= task.maxRetries) {
      throw new CoordinatorError(
        `Task ${taskId} has exhausted retries (${task.retryCount}/${task.maxRetries})`,
        'RETRY_EXHAUSTED',
        400,
        { taskId, retryCount: task.retryCount, maxRetries: task.maxRetries },
      );
    }

    const updateResult = await this.db.query(
      `UPDATE tasks
       SET status = 'pending',
           executor_session_key = NULL,
           retry_count = retry_count + 1,
           accepted_at = NULL,
           started_at = NULL,
           updated_at = NOW()
       WHERE id = $1 AND status IN ('accepted', 'processing', 'waiting_for_input')`,
      [taskId],
    );

    if (updateResult.rowCount === 0) {
      throw new InvalidTaskStateError(taskId, task.status, resetableStates);
    }

    // Re-enqueue in priority queue so poll picks it up
    const queueKey = `${REDIS_KEYS.TASK_QUEUE}:${task.toBotId}:${task.priority}`;
    await this.redis.lpush(queueKey, taskId);

    // Remove from processing ZSET
    await this.redis.zrem(REDIS_KEYS.PROCESSING_SET, taskId);

    // Clear cache so fresh data is fetched
    await this.redis.del(`${REDIS_KEYS.TASK_CACHE}:${taskId}`);

    this.logger.info('Task reset to pending', { taskId, botId, retryCount: task.retryCount + 1 });
  }

  async waitForInput(taskId: string, botId: string, reason: string, targetBotId?: string): Promise<void> {
    const task = await this.loadTask(taskId);

    // Allow both executor (toBotId) and delegator (fromBotId) to mark as waiting
    if (task.toBotId !== botId && task.fromBotId !== botId) {
      throw new UnauthorizedTaskError(taskId, botId);
    }
    const validStates = ['accepted', 'processing', 'waiting_for_input'];
    if (!validStates.includes(task.status)) {
      throw new InvalidTaskStateError(taskId, task.status, validStates);
    }

    // Determine who actually requested human input.
    let requestedBy = botId;
    if (targetBotId && (targetBotId === task.toBotId || targetBotId === task.fromBotId)) {
      requestedBy = targetBotId;
    }

    // Build waitingRequests array — supports concurrent requests from multiple bots.
    // Each bot can have at most one pending request; a new call from the same bot
    // replaces its previous reason.
    const existingResult = task.result || {};
    const existingRequests: Array<{ botId: string; reason: string }> =
      (existingResult as any).waitingRequests || [];
    const filtered = existingRequests.filter((r: any) => r.botId !== requestedBy);
    const waitingRequests = [...filtered, { botId: requestedBy, reason }];

    // Keep legacy single-value fields pointing to the latest request for backward compat
    const result = {
      ...existingResult,
      waitingRequests,
      waitingReason: reason,
      waitingRequestedBy: requestedBy,
    };
    const updateResult = await this.db.query(
      `UPDATE tasks SET status = 'waiting_for_input', result = $1, updated_at = NOW()
       WHERE id = $2 AND status IN ('accepted', 'processing', 'waiting_for_input')`,
      [JSON.stringify(result), taskId],
    );

    if (updateResult.rowCount === 0) {
      throw new InvalidTaskStateError(taskId, task.status, validStates);
    }

    await this.updateCacheStatus(taskId, 'waiting_for_input');

    // Record human_input_request event as a message for Activity Tree
    try {
      const msgId = randomUUID();
      const traceId = randomUUID();
      const now = new Date();
      const content = JSON.stringify({ text: `[Need Human Input] ${reason}` });
      await this.db.query(
        `INSERT INTO messages (id, from_bot_id, to_bot_id, type, content_type, content, priority, status, task_id, trace_id, created_at)
         VALUES ($1, $2, $3, 'human_input_request', 'text', $4, 'normal', 'delivered', $5, $6, $7)`,
        [msgId, botId, botId, content, taskId, traceId, now],
      );
    } catch (err) {
      this.logger.error('Failed to write human_input_request message', { taskId, err });
    }

    this.logger.info('Task waiting for input', { taskId, botId, reason });
  }

  /**
   * Unified resume: handles both waiting_for_input and completed/failed/timeout states.
   * - waiting_for_input: resumes with optional humanInput (original resumeFromWaiting logic)
   * - completed/failed/timeout: continues with new instructions (original continueTask logic)
   */
  async resume(taskId: string, botId: string, input?: string, targetBotId?: string): Promise<void> {
    const task = await this.loadTask(taskId);

    // Allow both fromBotId and toBotId
    if (task.toBotId !== botId && task.fromBotId !== botId) {
      throw new UnauthorizedTaskError(taskId, botId);
    }

    if (task.status === 'waiting_for_input') {
      // --- Original resumeFromWaiting logic ---
      const existingResult = (task.result || {}) as any;
      const existingRequests: Array<{ botId: string; reason: string }> =
        existingResult.waitingRequests || [];
      const remainingRequests = existingRequests.filter((r: any) => r.botId !== botId);

      const nextStatus = remainingRequests.length > 0 ? 'waiting_for_input' : 'processing';

      const resultPatch: any = { waitingRequests: remainingRequests };
      if (input) {
        resultPatch.humanInput = input;
      }
      if (remainingRequests.length > 0) {
        resultPatch.waitingReason = remainingRequests[0].reason;
        resultPatch.waitingRequestedBy = remainingRequests[0].botId;
      }

      const updateResult = await this.db.query(
        `UPDATE tasks SET status = $1, updated_at = NOW(),
         result = COALESCE(result, '{}'::jsonb) || $2::jsonb
         WHERE id = $3 AND status = 'waiting_for_input'`,
        [nextStatus, JSON.stringify(resultPatch), taskId],
      );

      if (updateResult.rowCount === 0) {
        throw new InvalidTaskStateError(taskId, task.status, ['waiting_for_input']);
      }

      await this.updateCacheStatus(taskId, nextStatus);

      // Record human_input_response event as a message for Activity Tree
      try {
        const msgId = randomUUID();
        const traceId = randomUUID();
        const now = new Date();
        const content = JSON.stringify({ text: input ? `[Human Reply] ${input}` : '[Task Resumed]' });
        await this.db.query(
          `INSERT INTO messages (id, from_bot_id, to_bot_id, type, content_type, content, priority, status, task_id, trace_id, created_at)
           VALUES ($1, $2, $3, 'human_input_response', 'text', $4, 'normal', 'delivered', $5, $6, $7)`,
          [msgId, botId, botId, content, taskId, traceId, now],
        );
      } catch (err) {
        this.logger.error('Failed to write human_input_response message', { taskId, err });
      }

      this.logger.info('Task resumed from waiting', { taskId, botId });

      if (input) {
        try {
          await this.writeResumeInputToInbox(task, input, botId);
        } catch (err) {
          this.logger.error('Failed to write humanInput to inbox', { taskId, err });
        }
      }
    } else if (['completed', 'failed', 'timeout'].includes(task.status)) {
      // --- Original continueTask logic ---
      const existingResult = (task.result || {}) as any;
      const previousResults = existingResult.previousResults || [];
      previousResults.push({
        status: task.status,
        result: existingResult,
        completedAt: task.completedAt,
        prompt: task.prompt,
      });
      const newResult = { previousResults, continuationCount: previousResults.length };

      const updateResult = await this.db.query(
        `UPDATE tasks SET
           status = 'processing',
           prompt = $1,
           result = $2,
           error = NULL,
           completed_at = NULL,
           updated_at = NOW()
         WHERE id = $3 AND status IN ('completed', 'failed', 'timeout')`,
        [input || task.prompt, JSON.stringify(newResult), taskId],
      );

      if (updateResult.rowCount === 0) {
        throw new InvalidTaskStateError(taskId, task.status, ['completed', 'failed', 'timeout']);
      }

      await this.redis.del(`${REDIS_KEYS.TASK_CACHE}:${taskId}`);

      const messageTarget = targetBotId || botId;
      try {
        await this.writeContinueToInbox(task, input || task.prompt || '', messageTarget);
      } catch (err) {
        this.logger.error('Failed to write continue message to inbox', { taskId, messageTarget, err });
      }

      // Write task_continuation message for Activity Tree
      try {
        const msgId = randomUUID();
        const traceId = randomUUID();
        const now = new Date();
        const content = JSON.stringify({ text: `[Task Continued] ${input || ''}` });
        await this.db.query(
          `INSERT INTO messages (id, from_bot_id, to_bot_id, type, content_type, content, priority, status, task_id, trace_id, created_at)
           VALUES ($1, $2, $3, 'task_continuation', 'text', $4, 'normal', 'delivered', $5, $6, $7)`,
          [msgId, botId, task.toBotId, content, taskId, traceId, now],
        );
      } catch (err) {
        this.logger.error('Failed to write task_continuation message', { taskId, err });
      }

      await this.safePublish('task_continued', { taskId, prompt: input, continuedBy: botId }, task.toBotId);

      this.logger.info('Task continued (direct to processing)', { taskId, botId, messageTarget, continuationCount: previousResults.length });
    } else {
      throw new InvalidTaskStateError(taskId, task.status, ['waiting_for_input', 'completed', 'failed', 'timeout']);
    }
  }

  /**
   * Write a direct_message with humanInput to the requesting bot's inbox.
   */
  private async writeResumeInputToInbox(task: Task, humanInput: string, targetBotId: string): Promise<void> {
    const messageId = randomUUID();
    const traceId = randomUUID();
    const now = new Date();

    const content = {
      text: `[Human Input for Task ${task.id}]\n\n${humanInput}\n\nPlease continue working on the task using this information.`,
    };

    const inboxMessage = JSON.stringify({
      messageId,
      fromBotId: task.fromBotId,
      toBotId: targetBotId,
      type: 'direct_message',
      contentType: 'text',
      content,
      priority: 'high',
      taskId: task.id,
      traceId,
      timestamp: now.toISOString(),
    });

    await this.redis.lpush(`clawteam:inbox:${targetBotId}:high`, inboxMessage);

    await this.db.query(
      `INSERT INTO messages (id, from_bot_id, to_bot_id, type, content_type, content, priority, status, task_id, trace_id, created_at)
       VALUES ($1, $2, $3, 'direct_message', 'text', $4, 'high', 'delivered', $5, $6, $7)`,
      [messageId, task.fromBotId, targetBotId, JSON.stringify(content), task.id, traceId, now],
    );

    this.logger.info('Wrote humanInput to requesting bot inbox', { taskId: task.id, targetBotId });
  }

  /**
   * Write a direct_message with continuation instructions to the target bot's inbox.
   */
  private async writeContinueToInbox(task: Task, prompt: string, targetBotId: string): Promise<void> {
    const messageId = randomUUID();
    const traceId = randomUUID();
    const now = new Date();

    const content = {
      text: `[Task Continuation for ${task.id}]\n\n${prompt}\n\nPlease continue working on the task with these updated instructions.`,
    };

    const inboxMessage = JSON.stringify({
      messageId,
      fromBotId: task.fromBotId,
      toBotId: targetBotId,
      type: 'direct_message',
      contentType: 'text',
      content,
      priority: 'high',
      taskId: task.id,
      traceId,
      timestamp: now.toISOString(),
    });

    await this.redis.lpush(`clawteam:inbox:${targetBotId}:high`, inboxMessage);

    await this.db.query(
      `INSERT INTO messages (id, from_bot_id, to_bot_id, type, content_type, content, priority, status, task_id, trace_id, created_at)
       VALUES ($1, $2, $3, 'direct_message', 'text', $4, 'high', 'delivered', $5, $6, $7)`,
      [messageId, task.fromBotId, targetBotId, JSON.stringify(content), task.id, traceId, now],
    );

    this.logger.info('Wrote continuation message to target bot inbox', { taskId: task.id, targetBotId });
  }

  async cancel(taskId: string, reason: string, botId: string): Promise<void> {
    const task = await this.loadTask(taskId);

    if (task.fromBotId !== botId) {
      throw new UnauthorizedTaskError(taskId, botId);
    }
    const cancellableStates = ['pending', 'accepted', 'processing', 'waiting_for_input'];
    if (!cancellableStates.includes(task.status)) {
      throw new InvalidTaskStateError(taskId, task.status, cancellableStates);
    }

    const now = new Date();

    const updateResult = await this.db.query(
      `UPDATE tasks
       SET status = 'cancelled', error = $1, completed_at = $2
       WHERE id = $3 AND status IN ('pending', 'accepted', 'processing', 'waiting_for_input')`,
      [
        JSON.stringify({ code: 'CANCELLED', message: reason }),
        now,
        taskId,
      ]
    );

    if (updateResult.rowCount === 0) {
      throw new InvalidTaskStateError(taskId, task.status, cancellableStates);
    }

    // Clean up Redis
    await this.removeFromQueue(taskId, task.toBotId);
    await this.redis.zrem(REDIS_KEYS.PROCESSING_SET, taskId);
    await this.redis.del(`${REDIS_KEYS.TASK_CACHE}:${taskId}`);

    // Record cancellation metric
    tasksCancelledTotal.inc({ capability: task.capability });

    // Notify target bot
    await this.safePublish('task_failed', {
      taskId,
      status: 'cancelled',
      reason,
    }, task.toBotId);

    this.logger.info('Task cancelled', { taskId, botId, reason });
  }

  private async loadTask(taskId: string): Promise<Task> {
    const result = await this.db.query<TaskRow>(
      'SELECT * FROM tasks WHERE id = $1',
      [taskId]
    );
    if (result.rows.length === 0) {
      throw new TaskNotFoundError(taskId);
    }
    return taskRowToTask(result.rows[0]);
  }

  private async removeFromQueue(taskId: string, botId: string): Promise<void> {
    for (const priority of PRIORITY_ORDER) {
      const queueKey = `${REDIS_KEYS.TASK_QUEUE}:${botId}:${priority}`;
      await this.redis.lrem(queueKey, 0, taskId);
    }
  }

  private async updateCacheStatus(taskId: string, status: string): Promise<void> {
    const cacheKey = `${REDIS_KEYS.TASK_CACHE}:${taskId}`;
    const exists = await this.redis.exists(cacheKey);
    if (exists) {
      await this.redis.hset(cacheKey, 'status', status);
    }
  }

  private async safePublish(
    event: 'task_assigned' | 'task_completed' | 'task_failed' | 'task_continued',
    payload: unknown,
    targetBotId: string
  ): Promise<void> {
    try {
      await this.messageBus.publish(event, payload, targetBotId);
    } catch (err) {
      this.logger.error(`Failed to publish ${event} event`, {
        err,
        targetBotId,
      });
    }
  }
}
