import { Worker, UnrecoverableError, type Job } from "bullmq";
import { logError, logInfo, logWarn } from "@/lib/logger";
import { safeFetch } from "@/lib/safeFetch";
import { UpdotCoreAuthManager } from "@/lib/updot-core-auth";
import { UpstreamMonitor } from "@/lib/upstream-monitor";
import { env } from "@/lib/env";
import {
  NOTIFICATION_TARGET_QUEUE,
  getBullMQConnection,
  type NotificationTargetJobData,
} from "./notification-target.queue";

// HTTP status codes that warrant a retry after 15 minutes
const RETRYABLE_STATUSES = new Set([500, 502, 503]);

export class NotificationTargetWorker {
  private worker: Worker<NotificationTargetJobData> | null = null;
  private readonly baseUrl: string;

  constructor() {
    this.baseUrl = env.GetString("UPDOT_CORE_BASE_URL");
  }

  start(): void {
    if (this.worker) return;

    const auth = UpdotCoreAuthManager.getInstance();

    this.worker = new Worker<NotificationTargetJobData>(
      NOTIFICATION_TARGET_QUEUE,
      async (job) => {
        const { campaignId, payload } = job.data;
        const attempt = job.attemptsMade + 1;
        const maxAttempts = job.opts.attempts ?? 1;

        if (UpstreamMonitor.isUpstreamDown()) {
          throw new Error(`Upstream is currently down. Worker is paused, will retry later.`);
        }

        logInfo(
          `[NotifTargetWorker] Processing campaign ${campaignId} — attempt ${attempt}/${maxAttempts}, job ${job.id}`,
        );

        // ── Auth ────────────────────────────────────────────────────────────
        let session = await auth.getOrLogin();
        if (!session) {
          // If session is null, check if it's a gateway issue
          const isHealthy = await UpstreamMonitor.checkHealth();
          if (!isHealthy) {
            logWarn(`[NotifTargetWorker] Auth failed due to gateway issue. Triggering global pause.`);
            await UpstreamMonitor.notifyDown("NotificationTargetWorker:Auth");
            throw new Error(`Upstream gateway error during auth. Worker pausing.`);
          }

          throw new Error(
            `[NotifTargetWorker] No auth session available for campaign ${campaignId}`,
          );
        }

        const cookies = [
          `k_session=${session.kc_session}`,
          `k_session_id=${session.kc_session_id}`,
        ].join("; ");

        // ── API call ─────────────────────────────────────────────────────────
        const url = `${this.baseUrl}/v2/member/admin/notifications/target`;

        let res = await safeFetch(url, {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
            Cookie: cookies,
          },
          body: JSON.stringify(payload),
        });

        // Handle Gateway issues with global monitor
        if (res.status === 502 || res.status === 503) {
          logWarn(`[NotifTargetWorker] Gateway error ${res.status}. Triggering global pause.`);
          await UpstreamMonitor.notifyDown(`NotificationTargetWorker:API:${res.status}`);
          throw new Error(`Upstream gateway error ${res.status}. Job will retry via queue backoff.`);
        }

        // 404 → target not found; treat as success so the job is not retried
        if (res.status === 404) {
          logInfo(
            `[NotifTargetWorker] Campaign ${campaignId} — 404 (target not found), marking as success`,
          );
          return { campaignId, status: 404 };
        }

        // 5xx transient server errors → throw so BullMQ schedules a retry
        if (RETRYABLE_STATUSES.has(res.status)) {
          logWarn(
            `[NotifTargetWorker] Campaign ${campaignId} — upstream ${res.status}, queuing retry (attempt ${attempt}/${maxAttempts})`,
          );
          // Invalidate the cached session on auth-related server errors
          if (res.status === 503) auth.invalidate();
          throw new Error(
            `Upstream ${res.status} for campaign ${campaignId} — will retry`,
          );
        }

        // Other non-2xx (e.g. 400, 401, 403, 422) → permanent failure, skip retries
        if (!res.ok) {
          const body = await res.text().catch(() => "");
          logError(
            `[NotifTargetWorker] Campaign ${campaignId} — non-retriable ${res.status}: ${body}`,
          );
          throw new UnrecoverableError(
            `Non-retriable ${res.status} for campaign ${campaignId}: ${body}`,
          );
        }

        logInfo(
          `[NotifTargetWorker] Campaign ${campaignId} — success (${res.status})`,
        );
        return { campaignId, status: res.status };
      },
      {
        connection: getBullMQConnection(),
        concurrency: 5,
      },
    );

    // Register monitor hooks to pause/resume the worker
    UpstreamMonitor.onRecovered(() => {
      logInfo("[NotifTargetWorker] Upstream recovered. Resuming worker.");
      this.worker?.resume();
    });

    // We can't easily register onDown, but the jobs themselves will call notifyDown
    // and throw, and the next jobs will check isUpstreamDown and throw, effectively pausing.
    // However, BullMQ worker.pause() is more explicit.
    // Let's add a way to subscribe to Down events.
    (UpstreamMonitor as any).onDown?.(() => {
      logInfo("[NotifTargetWorker] Upstream down. Pausing worker.");
      this.worker?.pause();
    });

    // ── Worker event hooks ───────────────────────────────────────────────────

    this.worker.on("completed", (job) => {
      logInfo(
        `[NotifTargetWorker] Job ${job.id} completed — campaign ${job.data.campaignId}`,
      );
    });

    this.worker.on("failed", (job, err) => {
      if (!job) return;
      const exhausted = job.attemptsMade >= (job.opts.attempts ?? 1);
      if (exhausted) {
        logError(
          `[NotifTargetWorker] Job ${job.id} exhausted all retries — campaign ${job.data.campaignId}: ${err.message}`,
        );
      } else {
        logWarn(
          `[NotifTargetWorker] Job ${job.id} attempt ${job.attemptsMade}/${job.opts.attempts} failed — campaign ${job.data.campaignId}: ${err.message}`,
        );
      }
    });

    this.worker.on("error", (err) => {
      logError(`[NotifTargetWorker] Worker connection error: ${err}`);
    });

    logInfo("[NotifTargetWorker] Started");
  }

  async stop(): Promise<void> {
    if (this.worker) {
      await this.worker.close();
      this.worker = null;
      logInfo("[NotifTargetWorker] Stopped");
    }
  }
}