import {
  Injectable,
  Logger,
  OnModuleDestroy,
  OnModuleInit,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { UnrecoverableError, Worker } from 'bullmq';
import Redis from 'ioredis';
import { randomUUID } from 'node:crypto';
import { SandboxRunner } from '../sandbox/sandbox-runner';
import {
  ScheduledJobsRuntimeClient,
  type RunBundle,
} from '../scheduled-jobs-runtime.client';
import {
  SCHEDULED_JOBS_QUEUE_NAME,
  type ScheduledJobData,
} from './scheduled-job-data';
import { redisConnection } from '../../automations/queue/redis-connection';

export const SCHEDULED_JOBS_QUEUE_PREFIX = 'planfi:scheduled-jobs';

@Injectable()
export class ScheduledJobsProcessor implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(ScheduledJobsProcessor.name);
  private worker?: Worker<ScheduledJobData>;
  private lockClient?: Redis;

  constructor(
    private readonly runtime: ScheduledJobsRuntimeClient,
    private readonly runner: SandboxRunner,
    private readonly configService: ConfigService,
  ) {}

  onModuleInit() {
    const redisUrl = this.configService.get<string>('REDIS_URL');
    if (!redisUrl) {
      this.logger.warn(
        'scheduled-jobs.worker.degraded REDIS_URL ausente; worker não iniciado',
      );
      return;
    }

    const prefix =
      this.configService.get<string>('SCHEDULED_JOBS_QUEUE_PREFIX') ||
      SCHEDULED_JOBS_QUEUE_PREFIX;
    const concurrency =
      Number(this.configService.get<string>('SCHEDULED_JOBS_CONCURRENCY')) || 5;

    this.worker = new Worker<ScheduledJobData>(
      SCHEDULED_JOBS_QUEUE_NAME,
      (job) => this.processJob(job.data, job.attemptsMade),
      {
        connection: redisConnection(redisUrl),
        prefix,
        concurrency,
      },
    );

    this.logger.log(
      `scheduled-jobs.worker.ready queue=${SCHEDULED_JOBS_QUEUE_NAME} prefix=${prefix} concurrency=${concurrency}`,
    );

    this.worker.on('error', (error) => {
      this.logger.error(`scheduled-jobs.worker.error message=${error.message}`);
    });

    this.worker.on('failed', (job, error) => {
      this.logger.error(
        `scheduled-jobs.job.failed id=${job?.id ?? 'unknown'} message=${error.message}`,
      );
    });

    // Lazy Redis client for locks (only created once needed)
    this.lockClient = new Redis(redisUrl, { lazyConnect: true });
  }

  async onModuleDestroy() {
    await this.worker?.close();
    await this.lockClient?.quit().catch(() => undefined);
  }

  async processJob(data: ScheduledJobData, attemptsMade = 0): Promise<void> {
    const trigger = data.trigger ?? 'SCHEDULE';
    const lockKey = `planfi:scheduled-jobs:lock:${data.jobId}`;
    let lockToken: string | null = null;
    let runId: string | null = null;
    // ARFIX-05: terminal report acontece no máximo uma vez. Sinalizado após
    // QUALQUER report terminal (sucesso OU FAILED do sandbox) para que o catch
    // não inverta um estado terminal já reportado para FAILED.
    let reported = false;
    // ARFIX-06: watchdog que renova o lock SKIP enquanto a seção crítica roda.
    let renewTimer: ReturnType<typeof setInterval> | undefined;
    // ARFIX-04 PART B: o bundle é lido no run-time; precisa ser visível no catch
    // para aplicar o cap de maxRetries (UnrecoverableError quando estourar).
    let bundle: RunBundle | undefined;

    try {
      // OBS-P0: ancorar a linha do run ANTES de qualquer chamada que possa
      // falhar (getBundle). Assim TODA falha — inclusive 401/404/rede ao buscar
      // o bundle — cai no catch e vira um run FAILED VISÍVEL (antes, getBundle
      // ficava fora do try e a falha sumia: nenhum run, sem status/alerta).
      // SJOB-XC-01: o Atlas valida scheduledFor com zod .optional() (não aceita
      // null). Omitimos a chave quando ausente em vez de mandar null.
      const createResult = await this.runtime.createRun({
        jobId: data.jobId,
        trigger,
        ...(data.scheduledFor ? { scheduledFor: data.scheduledFor } : {}),
      });
      runId = createResult.runId;

      // Agora busca o bundle. Se falhar (ex.: 401), o catch abaixo reporta
      // FAILED no run já ancorado.
      bundle = await this.runtime.getBundle(data.jobId);

      // Overlap guard: SKIP policy uses a Redis lock.
      // SJ-PROC-2: o lock cobre o ciclo COMPLETO sob o lock. Dentro da seção
      // crítica fazemos chamadas HTTP ao Atlas (reportRun RUNNING + final),
      // cada uma com timeout de axios de até 30s, mais a execução do runner
      // (timeoutMs). Usamos +90_000ms (~3x o timeout do axios) para cobrir o
      // pior caso das chamadas HTTP além do tempo do runner.
      const lockTtlMs = (bundle.limits.timeoutMs ?? 30_000) + 90_000;

      if (bundle.overlapPolicy === 'SKIP') {
        lockToken = await this.acquireLock(lockKey, lockTtlMs);
        if (!lockToken) {
          // Outra instância está rodando; reporta SKIPPED no run já criado.
          await this.runtime.reportRun({
            runId,
            jobId: data.jobId,
            status: 'SKIPPED',
            finishedAt: new Date().toISOString(),
          });
          return;
        }

        // ARFIX-06: o lock SKIP tem TTL fixo (lockTtlMs) e NÃO era renovado.
        // Se o run passa do envelope (reportRun lento, timeoutMs subestimado,
        // stalls de event-loop), o lock expira no meio e uma execução
        // concorrente o adquire — quebrando a garantia de SKIP. Renovamos o
        // lock periodicamente (~ttl/3, mínimo 5s) com um compare-and-pexpire
        // token-safe. Se a renovação falhar (lock perdido/roubado), apenas
        // logamos e seguimos: o run em voo NÃO é abortado (o isolate tem o
        // próprio timeout).
        const token = lockToken;
        const renewIntervalMs = Math.max(Math.floor(lockTtlMs / 3), 5_000);
        renewTimer = setInterval(() => {
          void this.renewLock(lockKey, token, lockTtlMs).then((renewed) => {
            if (!renewed) {
              this.logger.warn(
                `scheduled-jobs.lock.renew.lost jobId=${data.jobId} key=${lockKey}`,
              );
            }
          });
        }, renewIntervalMs);
        // Não segurar o processo aberto no shutdown.
        renewTimer.unref?.();
      }

      const startedAt = new Date().toISOString();
      await this.runtime.reportRun({
        runId,
        jobId: data.jobId,
        status: 'RUNNING',
        versionId: bundle.versionId,
        startedAt,
      });

      const planfiBaseUrl = this.configService.get<string>(
        'SCHEDULED_JOBS_PLANFI_BASE_URL',
      );

      const result = await this.runner.run({
        code: bundle.code,
        input: bundle.input,
        limits: bundle.limits,
        // P2: forward the job-scoped read-only env and the fetch allowlist so
        // the sandbox can expose globalThis.env and a fail-closed fetch.
        env: bundle.env,
        fetchAllowlist: bundle.fetchAllowlist,
        // planfi.call sugar: only injected when SCHEDULED_JOBS_PLANFI_BASE_URL
        // is configured in the deployment environment.
        planfi: planfiBaseUrl ? { baseUrl: planfiBaseUrl } : undefined,
      });

      // ARFIX-05: marcamos o estado terminal como reportado ANTES de aguardar a
      // chamada. Assim, se o PRÓPRIO reportRun terminal rejeitar no transporte
      // (blip de rede, 5xx pós-commit) DEPOIS de o Atlas já ter persistido o
      // status terminal, o catch NÃO emite um segundo FAILED no mesmo runId e
      // inverte o resultado. Vale para QUALQUER status terminal do sandbox
      // (SUCCESS/FAILED/TIMED_OUT), não só SUCCESS.
      reported = true;

      // SJOB-XC-02: errorMessage/errorStack são .optional() no zod do Atlas
      // (não aceitam null). No caminho de sucesso não há erro, então omitimos
      // as chaves em vez de enviar null.
      await this.runtime.reportRun({
        runId,
        jobId: data.jobId,
        versionId: bundle.versionId,
        status: result.status,
        finishedAt: new Date().toISOString(),
        durationMs: result.durationMs,
        output: result.output,
        logs: result.logs.join('\n'),
        ...(result.error?.message
          ? { errorMessage: result.error.message }
          : {}),
        ...(result.error?.stack ? { errorStack: result.error.stack } : {}),
      });
    } catch (err) {
      if (runId && !reported) {
        // OBS-P0: o run foi ancorado — reporta FAILED com a mensagem (cobre
        // tanto falha do getBundle quanto erros inesperados no ciclo).
        // SJOB-XC-02: errorStack é .optional() no Atlas; omitir quando ausente.
        const errorStack = err instanceof Error ? err.stack : undefined;
        await this.runtime
          .reportRun({
            runId,
            jobId: data.jobId,
            status: 'FAILED',
            finishedAt: new Date().toISOString(),
            errorMessage:
              err instanceof Error ? err.message : 'Unexpected error',
            ...(errorStack ? { errorStack } : {}),
          })
          .catch((reportErr) => {
            this.logger.error(
              `scheduled-jobs.report.failed runId=${runId} message=${reportErr instanceof Error ? reportErr.message : String(reportErr)}`,
            );
          });
      } else if (!runId) {
        // OBS-P0: o próprio createRun falhou (ex.: canal Atlas fora / 401 amplo
        // na service key) — não há run para marcar. Log estruturado para a
        // falha NÃO ficar 100% silenciosa (antes só havia o log genérico do
        // BullMQ 'failed').
        this.logger.error(
          `scheduled-jobs.bootstrap.unreachable jobId=${data.jobId} message=${err instanceof Error ? err.message : String(err)}`,
        );
      }
      // ARFIX-05: se runId existe MAS já reportamos um estado terminal
      // (reported=true), NÃO re-reportamos nada — o run permanece no seu status
      // terminal verdadeiro. Apenas re-lançamos para o BullMQ registrar o erro.

      // ARFIX-04 PART B: maxRetries (0..10, validado no Atlas) só é conhecido em
      // run-time via o bundle. O envelope estático attempts:3 do port é um teto
      // superior; o cap autoritativo é maxRetries. Quando attemptsMade já atingiu
      // o cap, lançamos UnrecoverableError para o BullMQ PARAR de re-tentar;
      // abaixo do cap re-lançamos o erro original para o envelope re-tentar.
      const maxRetries = bundle?.maxRetries ?? 0;
      if (attemptsMade >= maxRetries) {
        const message = err instanceof Error ? err.message : 'Unexpected error';
        throw new UnrecoverableError(message);
      }
      throw err;
    } finally {
      // ARFIX-06: limpar o watchdog ANTES do release, para que uma renovação
      // (pexpire) não chegue logo após o DEL e ressuscite a chave.
      if (renewTimer) {
        clearInterval(renewTimer);
      }
      if (lockToken) {
        await this.releaseLock(lockKey, lockToken);
      }
    }
  }

  /**
   * Acquires a Redis SET NX PX lock.
   * Returns the token if acquired, null if lock is already held.
   *
   * Exposed as protected so tests can spy/override without real Redis.
   */
  protected async acquireLock(
    key: string,
    ttlMs: number,
  ): Promise<string | null> {
    if (!this.lockClient) return null;
    const token = randomUUID();
    const result = await this.lockClient.set(key, token, 'PX', ttlMs, 'NX');
    return result === 'OK' ? token : null;
  }

  /**
   * Releases the lock only if the token matches (Lua compare-and-delete).
   *
   * Exposed as protected so tests can spy/override without real Redis.
   */
  protected async releaseLock(key: string, token: string): Promise<void> {
    if (!this.lockClient) return;
    const script = `
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
      else
        return 0
      end
    `;
    await this.lockClient.eval(script, 1, key, token);
  }

  /**
   * Renews (extends the TTL of) the lock only if the token still matches
   * (Lua compare-and-pexpire). Returns true if the lock was extended, false if
   * the token no longer matches (lock lost/stolen) or there is no client.
   *
   * A blind PEXPIRE could extend a lock another instance now owns, so the
   * renewal is token-conditional — the same idiom as releaseLock.
   *
   * Exposed as protected so tests can spy/override without real Redis.
   */
  protected async renewLock(
    key: string,
    token: string,
    ttlMs: number,
  ): Promise<boolean> {
    if (!this.lockClient) return false;
    const script = `
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("pexpire", KEYS[1], ARGV[2])
      else
        return 0
      end
    `;
    const result = await this.lockClient.eval(script, 1, key, token, ttlMs);
    return result === 1;
  }
}