import { ChildProcess } from 'child_process';
import { CronExpressionParser } from 'cron-parser';
import fs from 'fs';

import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js';
import {
  ContainerOutput,
  runContainerAgent,
  writeTasksSnapshot,
} from './container-runner.js';
import {
  checkUsageLimit,
  getAllTasks,
  getTasksForTenant,
  getDueTasks,
  getTaskById,
  logTaskRun,
  logUsage,
  updateTask,
  updateTaskAfterRun,
} from './db.js';
import { GroupQueue } from './group-queue.js';
import { resolveGroupFolderPath, resolveGroupIpcPath } from './group-folder.js';
import { logger } from './logger.js';
import { ensureCredentialFreshForTenant, writeTokenFile } from './credential-pool.js';
import { getTenantProvider } from './runtime-provider.js';
import { TenantManager } from './tenant-manager.js';
import { RegisteredGroup, ScheduledTask, SessionRecord } from './types.js';

/**
 * Compute the next run time for a recurring task, anchored to the
 * task's scheduled time rather than Date.now() to prevent cumulative
 * drift on interval-based tasks.
 *
 * Co-authored-by: @community-pr-601
 */
export function computeNextRun(task: ScheduledTask): string | null {
  if (task.schedule_type === 'once') return null;

  const now = Date.now();

  if (task.schedule_type === 'cron') {
    const interval = CronExpressionParser.parse(task.schedule_value, {
      tz: TIMEZONE,
    });
    return interval.next().toISOString();
  }

  if (task.schedule_type === 'interval') {
    const ms = parseInt(task.schedule_value, 10);
    if (!ms || ms <= 0) {
      // Guard against malformed interval that would cause an infinite loop
      logger.warn(
        { taskId: task.id, value: task.schedule_value },
        'Invalid interval value',
      );
      return new Date(now + 60_000).toISOString();
    }
    // Anchor to the scheduled time, not now, to prevent drift.
    // Skip past any missed intervals so we always land in the future.
    let next = new Date(task.next_run!).getTime() + ms;
    while (next <= now) {
      next += ms;
    }
    return new Date(next).toISOString();
  }

  return null;
}

export interface SchedulerDependencies {
  registeredGroups: () => Record<string, RegisteredGroup>;
  getSessions: () => Record<string, SessionRecord>;
  queue: GroupQueue;
  tenantManager: TenantManager;
  onProcess: (
    groupJid: string,
    proc: ChildProcess,
    containerName: string,
    groupFolder: string,
  ) => void;
  sendMessage: (jid: string, text: string) => Promise<void>;
}

/**
 * Set of chatJids for which an IPC message (tc send) was sent during an active task run.
 * Used to suppress duplicate result delivery when the agent uses tc send AND produces output text.
 */
const activeTaskIpcSent = new Set<string>();

/**
 * Called by the IPC message handler when a tc-send message is delivered for a chatJid.
 * Allows the task scheduler to detect and suppress duplicate result delivery.
 */
export function notifyTaskIpcMessage(chatJid: string): void {
  activeTaskIpcSent.add(chatJid);
}

async function runTask(
  task: ScheduledTask,
  deps: SchedulerDependencies,
): Promise<void> {
  const startTime = Date.now();
  // Clear any stale IPC-sent flag for this chatJid before starting the task window
  activeTaskIpcSent.delete(task.chat_jid);
  let groupDir: string;
  try {
    groupDir = resolveGroupFolderPath(task.group_folder);
  } catch (err) {
    const error = err instanceof Error ? err.message : String(err);
    // Stop retry churn for malformed legacy rows.
    updateTask(task.id, { status: 'paused' });
    logger.error(
      { taskId: task.id, groupFolder: task.group_folder, error },
      'Task has invalid group folder',
    );
    logTaskRun({
      task_id: task.id,
      run_at: new Date().toISOString(),
      duration_ms: Date.now() - startTime,
      status: 'error',
      result: null,
      error,
    });
    return;
  }
  fs.mkdirSync(groupDir, { recursive: true });

  logger.info(
    { taskId: task.id, group: task.group_folder },
    'Running scheduled task',
  );

  const groups = deps.registeredGroups();
  const group = Object.values(groups).find(
    (g) => g.folder === task.group_folder,
  );

  if (!group) {
    logger.error(
      { taskId: task.id, groupFolder: task.group_folder },
      'Group not found for task',
    );
    logTaskRun({
      task_id: task.id,
      run_at: new Date().toISOString(),
      duration_ms: Date.now() - startTime,
      status: 'error',
      result: null,
      error: `Group not found: ${task.group_folder}`,
    });
    return;
  }

  // Check per-tenant usage limits before spawning the container
  if (group.tenantId) {
    const limitCheck = checkUsageLimit(group.tenantId);
    if (!limitCheck.allowed) {
      logger.warn(
        { taskId: task.id, tenantId: group.tenantId, reason: limitCheck.reason },
        'Tenant usage limit reached, skipping scheduled task',
      );
      return;
    }
  }

  // Update tasks snapshot for container to read (filtered by group)
  const isMain = group.isMain === true;
  const mainGroup = !isMain && group.tenantId
    ? Object.values(groups).find(
        (candidate) =>
          candidate.tenantId === group.tenantId && candidate.isMain,
      )
    : undefined;
  const tasks = group.tenantId ? getTasksForTenant(group.tenantId) : getAllTasks();
  writeTasksSnapshot(
    task.group_folder,
    isMain,
    tasks.map((t) => ({
      id: t.id,
      groupFolder: t.group_folder,
      prompt: t.prompt,
      scheduleType: t.schedule_type,
      scheduleValue: t.schedule_value,
      status: t.status,
      nextRun: t.next_run,
    })),
  );

  let result: string | null = null;
  let error: string | null = null;

  // For group context mode, use the group's current session
  const sessions = deps.getSessions();

  // Get tenant context for resource limits and secrets
  const tenant = group.tenantId
    ? deps.tenantManager.getTenant(group.tenantId)
    : undefined;
  const provider = getTenantProvider(tenant);
  const sessionRecord =
    task.context_mode === 'group' ? sessions[task.group_folder] : undefined;
  const sessionId = sessionRecord?.runtime === provider ? sessionRecord.id : undefined;
  // Refresh credential if expired/expiring before launching the task container
  if (tenant) await ensureCredentialFreshForTenant(tenant);

  const secretsResult = group.tenantId
    ? deps.tenantManager.getSecretsForTenant(group.tenantId)
    : undefined;
  if (group.tenantId && !secretsResult) {
    logger.error({ tenantId: group.tenantId, taskId: task.id }, 'No credentials available for tenant, skipping task run');
    return;
  }
  const secrets = secretsResult?.secrets;
  // Write fresh token file to the IPC dir — this is where the agent-runner reads it
  // (/workspace/ipc/.token.json maps to the IPC dir, NOT the groups dir)
  if (secrets && secretsResult?.credentialId) {
    const ipcDir = resolveGroupIpcPath(task.group_folder);
    fs.mkdirSync(ipcDir, { recursive: true });
    writeTokenFile(ipcDir, secrets, secretsResult.credentialId);
  }
  const effectiveName = group.tenantId
    ? (deps.tenantManager.getTenant(group.tenantId)?.name || ASSISTANT_NAME)
    : ASSISTANT_NAME;

  // After the task produces a result, close the container promptly.
  // Tasks are single-turn — no need to wait IDLE_TIMEOUT (30 min) for the
  // query loop to time out. A short delay handles any final MCP calls.
  const TASK_CLOSE_DELAY_MS = 10000;
  let closeTimer: ReturnType<typeof setTimeout> | null = null;

  const scheduleClose = () => {
    if (closeTimer) return; // already scheduled
    closeTimer = setTimeout(() => {
      logger.debug({ taskId: task.id }, 'Closing task container after result');
      deps.queue.closeStdin(task.chat_jid);
    }, TASK_CLOSE_DELAY_MS);
  };

  try {
    const output = await runContainerAgent(
      group,
      {
        prompt: task.prompt,
        sessionId,
        groupFolder: task.group_folder,
        chatJid: task.chat_jid,
        isMain,
        mainGroupFolder: mainGroup?.folder,
        isScheduledTask: true,
        assistantName: effectiveName,
        tenantId: group.tenantId,
        provider,
      },
      (proc, containerName) =>
        deps.onProcess(task.chat_jid, proc, containerName, task.group_folder),
      async (streamedOutput: ContainerOutput) => {
        if (streamedOutput.result) {
          const raw = typeof streamedOutput.result === 'string'
            ? streamedOutput.result
            : JSON.stringify(streamedOutput.result);
          // Strip <internal> tags (same as interactive session handler)
          const text = raw.replace(/<internal>[\s\S]*?<\/internal>/g, '').trim();
          if (text) {
            result = text;
            // Only send if the agent hasn't already delivered this via tc send (IPC).
            // If the agent used tc send, the message was already delivered — forwarding
            // the result here would duplicate it.
            if (!activeTaskIpcSent.has(task.chat_jid)) {
              await deps.sendMessage(task.chat_jid, text);
            } else {
              logger.info({ taskId: task.id }, 'Task result suppressed — already delivered via tc send');
            }
          }
          scheduleClose();
        }
        if (streamedOutput.status === 'success') {
          deps.queue.notifyIdle(task.chat_jid);
        }
        if (streamedOutput.status === 'error') {
          error = streamedOutput.error || 'Unknown error';
        }
      },
      tenant,
      secrets,
    );

    if (closeTimer) clearTimeout(closeTimer);

    if (output.status === 'error') {
      error = output.error || 'Unknown error';
    } else if (output.result) {
      // Result was already forwarded to the user via the streaming callback above
      result = output.result;
    }

    // Notify user of errors (silent successes are intentional for background tasks)
    if (error) {
      await deps.sendMessage(task.chat_jid, `⚠️ Task failed: ${error}`).catch(() => {});
    }

    logger.info(
      { taskId: task.id, durationMs: Date.now() - startTime },
      'Task completed',
    );
  } catch (err) {
    if (closeTimer) clearTimeout(closeTimer);
    error = err instanceof Error ? err.message : String(err);
    logger.error({ taskId: task.id, error }, 'Task failed');
    await deps.sendMessage(task.chat_jid, `⚠️ Task failed: ${error}`).catch(() => {});
  }

  const durationMs = Date.now() - startTime;

  logTaskRun({
    task_id: task.id,
    run_at: new Date().toISOString(),
    duration_ms: durationMs,
    status: error ? 'error' : 'success',
    result,
    error,
  });

  if (group.tenantId) {
    logUsage(group.tenantId, task.group_folder, durationMs);
  }

  const nextRun = computeNextRun(task);
  const resultSummary = error
    ? `Error: ${error}`
    : result
      ? result.slice(0, 200)
      : 'Completed';
  updateTaskAfterRun(task.id, nextRun, resultSummary);
  // Clean up IPC tracking for this task's chatJid
  activeTaskIpcSent.delete(task.chat_jid);
}

let schedulerRunning = false;

export function startSchedulerLoop(deps: SchedulerDependencies): void {
  if (schedulerRunning) {
    logger.debug('Scheduler loop already running, skipping duplicate start');
    return;
  }
  schedulerRunning = true;
  logger.info('Scheduler loop started');

  const loop = async () => {
    try {
      const dueTasks = getDueTasks();
      if (dueTasks.length > 0) {
        logger.info({ count: dueTasks.length }, 'Found due tasks');
      }

      for (const task of dueTasks) {
        // Re-check task status in case it was paused/cancelled
        const currentTask = getTaskById(task.id);
        if (!currentTask || currentTask.status !== 'active') {
          continue;
        }

        deps.queue.enqueueTask(currentTask.chat_jid, currentTask.id, () =>
          runTask(currentTask, deps),
        );
      }
    } catch (err) {
      logger.error({ err }, 'Error in scheduler loop');
    }

    setTimeout(loop, SCHEDULER_POLL_INTERVAL);
  };

  loop();
}

/** @internal - for tests only. */
export function _resetSchedulerLoopForTests(): void {
  schedulerRunning = false;
}
