import type { CampaignRepository } from "@/internal/repository/campaigns/campaigns";
import type { BlockedEmailRepository } from "@/internal/repository/campaigns/blocked_emails";
import type { DefaultEmailRepository } from "@/internal/repository/campaigns/default_emails";
import type { ExternalMemberRepository } from "@/internal/repository/members/external_members";
import { emitCampaignProgress, emitCampaignStatus } from "@/internal/socket";
import { env } from "@/lib/env";
import { logError, logInfo, logWarn } from "@/lib/logger";
import { rabbitMQ } from "@/lib/rabbitmq";
import { UpdotCoreAuthManager } from "@/lib/updot-core-auth";
import { UpstreamMonitor } from "@/lib/upstream-monitor";

const NOTIFICATION_API_URL = `${env.GetString("UPDOT_CORE_BASE_URL")}/v2/member/admin/notifications/target`;

/** Emails per notification API call when a campaign carries no batchSize of its own. */
export const DEFAULT_BATCH_SIZE = 750;

/** Upper bound on batchSize, mirroring the console's "Batch Size (Max 750)" input. */
const MAX_BATCH_SIZE = 750;

/** Campaigns allowed to be in SENDING at the same time. */
const DEFAULT_MAX_CONCURRENT_CAMPAIGNS = 3;

/** Pause between a campaign's batches when NOTIFICATION_BATCH_DELAY_MINUTES is unset. */
const DEFAULT_BATCH_DELAY_MINUTES = 10;

/** How often a campaign waiting for a free slot re-checks for one. */
const SLOT_POLL_INTERVAL_MS = 10_000;

/** Slot-wait log noise control: one line every N polls instead of every poll. */
const SLOT_WAIT_LOG_EVERY = 6;

function getNotificationQueueName(): string {
  const base = env.GetString("NOTIFICATION_QUEUE_NAME") || "campaign_queue";
  const nodeEnv = process.env.NODE_ENV || "development";
  return `${base}_${nodeEnv}`;
}

/** Optional numeric env var — absent or unparseable falls back instead of logging fatal. */
function readIntEnv(key: string, fallback: number): number {
  const raw = process.env[key];
  if (raw === undefined || raw.trim() === "") return fallback;
  const parsed = Number.parseInt(raw, 10);
  return Number.isNaN(parsed) ? fallback : parsed;
}

/**
 * Concurrency ceiling shared by the service and the queue worker's prefetch, so the
 * worker never holds more unacked messages than there are slots to run them in.
 */
export function getMaxConcurrentCampaigns(): number {
  const configured = readIntEnv("MAX_CONCURRENT_CAMPAIGNS", DEFAULT_MAX_CONCURRENT_CAMPAIGNS);
  return configured < 1 ? DEFAULT_MAX_CONCURRENT_CAMPAIGNS : configured;
}

/**
 * Delay applied to a campaign that starts while others are already running, so
 * concurrent campaigns don't line up their notification API calls on the same tick.
 */
function getStartStaggerMs(): number {
  const seconds = readIntEnv("CAMPAIGN_START_STAGGER_SECONDS", 30);
  return Math.max(0, seconds) * 1000;
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

export class CampaignService {
  private campaignRepo: CampaignRepository;
  private memberRepo: ExternalMemberRepository;
  private blockedEmailRepo: BlockedEmailRepository;
  private defaultEmailRepo?: DefaultEmailRepository;

  /**
   * Campaign ids this process is currently sending. The authoritative slot count
   * lives in the database (so it holds across instances); this set is the local
   * guard against a redelivered message re-entering a run already in flight, and
   * the source for the in-process status logging.
   */
  private static readonly activeCampaigns = new Set<string>();

  private static monitorInitialized = false;

  /** Snapshot of what this process is sending right now — used for status logging. */
  static getActiveCampaignIds(): string[] {
    return [...CampaignService.activeCampaigns];
  }

  constructor(
    campaignRepo: CampaignRepository,
    memberRepo: ExternalMemberRepository,
    blockedEmailRepo: BlockedEmailRepository,
    defaultEmailRepo?: DefaultEmailRepository,
  ) {
    this.campaignRepo = campaignRepo;
    this.memberRepo = memberRepo;
    this.blockedEmailRepo = blockedEmailRepo;
    this.defaultEmailRepo = defaultEmailRepo;

    CampaignService.initMonitor(this);
  }

  private static initMonitor(service: CampaignService) {
    if (this.monitorInitialized) return;
    this.monitorInitialized = true;

    UpstreamMonitor.onRecovered(async () => {
      logInfo("[CampaignService] Upstream recovered. Auto-resuming all QUEUED campaigns...");
      try {
        const { items: queued } = await service.campaignRepo.list(100, 0, { status: "QUEUED" });
        for (const campaign of queued) {
          logInfo(`[CampaignService] Auto-resuming campaign ${campaign.id}`);
          await service.enqueueCampaign(campaign.id);
        }
      } catch (err) {
        logError(`[CampaignService] Error during auto-resume re-enqueue: ${err}`);
      }
    });
  }

  async createCampaign(payload: {
    name: string;
    title: string;
    body: string;
    image_url?: string;
    target_criteria: any;
    additional_data?: any;
    status?: string;
    scheduled_at?: string;
    total_count?: number;
    published_by?: string;
  }) {
    const total_count = await this.getAudienceCount(payload.target_criteria);
    return await this.campaignRepo.create({ ...payload, total_count });
  }

  async updateCampaign(
    id: string,
    payload: {
      name?: string;
      title?: string;
      body?: string;
      image_url?: string;
      target_criteria?: any;
      additional_data?: any;
      status?: string;
      scheduled_at?: string;
    },
  ) {
    const campaign = await this.campaignRepo.getById(id);
    if (!campaign) {
      throw new Error("Campaign not found");
    }

    if (payload.target_criteria) {
      const total_count = await this.getAudienceCount(payload.target_criteria);
      (payload as any).total_count = total_count;
    }

    return await this.campaignRepo.update(id, payload);
  }

  async sendCampaign(id: string) {
    if (UpstreamMonitor.isUpstreamDown()) {
      logInfo(`[CampaignService] Upstream is currently down. Delaying campaign ${id} processing.`);
      return 0;
    }

    if (CampaignService.activeCampaigns.has(id)) {
      logWarn(`[CampaignService] Campaign ${id} is already being sent by this instance. Ignoring duplicate delivery.`);
      return 0;
    }

    const maxConcurrent = getMaxConcurrentCampaigns();

    // Take a slot before touching anything else. The claim is atomic in the
    // database, so up to maxConcurrent campaigns run together and a campaign
    // already SENDING (redelivered message, double enqueue) is never started twice.
    let claim = await this.campaignRepo.claimSendingSlot(id, maxConcurrent);
    let waitPolls = 0;
    while (!claim.claimed && claim.reason === "no_slot") {
      if (waitPolls % SLOT_WAIT_LOG_EVERY === 0) {
        logInfo(`[CampaignService] Campaign ${id} waiting for a slot — ${claim.activeCount}/${maxConcurrent} campaigns sending.`);
      }
      waitPolls++;
      await sleep(SLOT_POLL_INTERVAL_MS);

      // An operator can stop or delete a campaign while it queues for a slot.
      const waitingStatus = (await this.campaignRepo.getById(id))?.status;
      if (waitingStatus === "STOPPED" || waitingStatus === "DELETED") {
        logInfo(`[CampaignService] Campaign ${id} became ${waitingStatus} while waiting for a slot. Abandoning.`);
        return 0;
      }

      claim = await this.campaignRepo.claimSendingSlot(id, maxConcurrent);
    }

    if (!claim.claimed) {
      switch (claim.reason) {
        case "not_found":
          logError(`[CampaignService] Campaign ${id} not found.`);
          return 0;
        case "already_sending":
          logWarn(`[CampaignService] Campaign ${id} is already SENDING elsewhere. Skipping duplicate run.`);
          return Number(claim.campaign?.sent_count || 0);
        default:
          logInfo(`[CampaignService] Campaign ${id} already in terminal state: ${claim.campaign?.status}. Skipping.`);
          return Number(claim.campaign?.sent_count || 0);
      }
    }

    let CHUNK_SIZE = DEFAULT_BATCH_SIZE;
    CampaignService.activeCampaigns.add(id);
    const slotCount = claim.activeCount;
    try {
      const campaign = claim.campaign;

      logInfo(`[CampaignService] Campaign ${id} loaded — status=SENDING (was ${claim.previousStatus}), title="${campaign.title}", sent_count=${campaign.sent_count}, total_count=${campaign.total_count}, slot=${slotCount}/${maxConcurrent}`);
      emitCampaignStatus(id, "SENDING");
      await this.campaignRepo.logSystemActivity(id, "Campaign Processing Started", {
        concurrentSlot: slotCount,
        maxConcurrent,
        activeInThisInstance: CampaignService.getActiveCampaignIds(),
      });

      // Offset the start when other campaigns are already running so their batch
      // calls to the notification API don't all land at the same moment.
      if (slotCount > 1) {
        const staggerMs = getStartStaggerMs() * (slotCount - 1);
        if (staggerMs > 0) {
          logInfo(`[CampaignService] Campaign ${id} staggering start by ${staggerMs / 1000}s (${slotCount} campaigns running).`);
          await sleep(staggerMs);
        }
      }

      const criteria =
        typeof campaign.target_criteria === "string"
          ? JSON.parse(campaign.target_criteria)
          : campaign.target_criteria;

      const rawAdditionalData =
        typeof campaign.additional_data === "string"
          ? JSON.parse(campaign.additional_data)
          : campaign.additional_data;

      // Ensure additionalData is an object
      const additionalData = (rawAdditionalData && typeof rawAdditionalData === "object") ? rawAdditionalData : {};

      let campaignID = campaign.campaignID;
      if (!campaignID) {
        const timestamp = Date.now();
        // The campaign id suffix keeps this unique when two campaigns start within
        // the same millisecond — a plain timestamp would collide and merge their
        // delivery stats, which are looked up by campaignID.
        campaignID = `campaign_${timestamp}_${id.replace(/-/g, "").slice(0, 8)}`;

        // Save the campaignID to database so that it persists across pauses, resumes, and retries
        await this.campaignRepo.update(id, {
          campaignID,
        });
      }

      const emails = await this.memberRepo.findEmailsByCriteria(criteria);

      // Filter blocked emails if enabled (only active blocked emails)
      let filteredEmails = emails;
      if (additionalData?.apply_block_list) {
        const blockedEmails = await this.blockedEmailRepo.getActiveEmails();
        const blockedSet = new Set(blockedEmails.map(e => e.toLowerCase()));
        filteredEmails = emails.filter(email => !blockedSet.has(email.toLowerCase()));
        logInfo(`[CampaignService] Block list applied. Original: ${emails.length}, Filtered: ${filteredEmails.length}`);
      }

      // Fetch default emails if enabled
      let defaultEmails: string[] = [];
      if (additionalData?.apply_default_email && this.defaultEmailRepo) {
        defaultEmails = await this.defaultEmailRepo.getAllEmails();
        logInfo(`[CampaignService] Default emails loaded. Count: ${defaultEmails.length}`);
      }

      const emailsCount = filteredEmails.length;

      if (emailsCount === 0) {
        await this.campaignRepo.updateStatus(id, "COMPLETED");
        emitCampaignStatus(id, "COMPLETED");
        return 0;
      }

      let sentCount = Number(campaign.sent_count || 0);

      if (sentCount >= emailsCount) {
        logInfo(`[CampaignService] Campaign ${id} already fully sent (${sentCount}/${emailsCount}), marking SENT`);
        await this.campaignRepo.updateStatus(id, "SENT");
        emitCampaignStatus(id, "SENT");
        return emailsCount;
      }

      await this.campaignRepo.setTotalCount(id, emailsCount);

      let totalInvalid = 0;
      let totalProcessable = 0;
      let totalNonProcessable = 0;

      logInfo(`[CampaignService] Campaign ${id} authenticating with Updot Core...`);
      const authManager = UpdotCoreAuthManager.getInstance();

      let session = await authManager.getOrLogin();
      if (!session) {
        // If session is null, check if it's a gateway issue
        const isHealthy = await UpstreamMonitor.checkHealth();
        if (!isHealthy) {
          logWarn(`[CampaignService] Auth failed due to gateway issue for campaign ${id}. Triggering global pause.`);
          await this.handleUpstreamDown(id, "Auth Initialization");
          return Number(campaign.sent_count || 0);
        }

        logError(`[CampaignService] Campaign ${id} — Updot auth returned null session (potential credentials issue)`);
        throw new Error("Unable to obtain Updot core authentication session.");
      }
      logInfo(`[CampaignService] Campaign ${id} — Updot auth success`);

      const accountInfo = authManager.getAccountInfo();
      logInfo(`[CampaignService] Campaign ${id} — accountId=${accountInfo.accountId}, contactId=${accountInfo.contactId}`);
      const configChunkSize = additionalData?.schedule_details?.batchSize;

      if (configChunkSize !== undefined && configChunkSize !== null) {
        const parsed = Number(configChunkSize);
        if (!Number.isNaN(parsed) && parsed > 0) {
          CHUNK_SIZE = Math.min(parsed, MAX_BATCH_SIZE);
        }
      }

      logInfo(`[CampaignService] Campaign ${id} batch size ${CHUNK_SIZE} (configured=${configChunkSize ?? "none"}, default=${DEFAULT_BATCH_SIZE}).`);


      // When default emails are used, they take slots within each batch
      const effectiveBatchSize = defaultEmails.length > 0
        ? Math.max(1, CHUNK_SIZE - defaultEmails.length)
        : CHUNK_SIZE;

      for (let i = sentCount; i < emailsCount; i += effectiveBatchSize) {
        const batchNum = Math.floor(i / effectiveBatchSize) + 1;

        // Check if stopped or re-enqueued (resume while still running)
        const currentStatus = (await this.campaignRepo.getById(id))?.status;
        if (currentStatus === "STOPPED" || currentStatus === "QUEUED" || currentStatus === "DELETED") {
          if (currentStatus === "STOPPED") {
            emitCampaignStatus(id, "STOPPED");
          }
          return sentCount;
        }

        const regularChunk = filteredEmails.slice(i, i + effectiveBatchSize);
        // Merge default emails into each batch, deduplicating
        const chunk = defaultEmails.length > 0
          ? [...new Set([...regularChunk, ...defaultEmails])]
          : regularChunk;

        const customDataArr = additionalData?.customData || [];
        const parsedCustomData = customDataArr.reduce((acc: any, curr: any) => {
          if (curr && curr.key) {
            acc[curr.key] = curr.value;
          }
          return acc;
        }, {});

        const body = {
          member_emails: chunk,
          title: campaign.title,
          body: campaign.body,
          name: "test_adding",
          campaignID,
          ...(campaign.image_url
            ? { image: encodeURI(campaign.image_url) }
            : {}),
          additional_data: {
            ...parsedCustomData,
            path: parsedCustomData.path || "",
            slug: parsedCustomData.slug || "",
            campaign_id: id,
            campaignID,
          },
          persist: additionalData?.persist ?? true,
        };

        const getHeaders = (sess: any) => ({
          "Content-Type": "application/json",
          "x-account": accountInfo.accountId,
          "x-contact": accountInfo.contactId,
          Cookie: `k_session=${sess.kc_session}; k_session_id=${sess.kc_session_id}`,
        });

        const controller = new AbortController();
        const timeoutId = setTimeout(() => controller.abort(), 15 * 60 * 1000); // Increased to 15 mins

        let response = await fetch(NOTIFICATION_API_URL, {
          method: "POST",
          headers: getHeaders(session),
          body: JSON.stringify(body),
          signal: controller.signal,
        });
        clearTimeout(timeoutId);

        if ([401, 403].includes(response.status)) {
          // Session-scoped so a parallel campaign's fresh session isn't discarded.
          authManager.invalidateIfCurrent(session);
          session = await authManager.getOrLogin();
          if (session) {
            const retryController = new AbortController();
            const retryTimeoutId = setTimeout(
              () => retryController.abort(),
              15 * 60 * 1000, // Increased to 15 mins
            );
            response = await fetch(NOTIFICATION_API_URL, {
              method: "POST",
              headers: getHeaders(session),
              body: JSON.stringify(body),
              signal: retryController.signal,
            });
            clearTimeout(retryTimeoutId);
          } else {
            logError(`[CampaignService] Campaign ${id} — Re-auth FAILED, session is null`);
          }
        }

        // Handle Bad Gateway/Service Unavailable with a global pause and auto-resume
        if (response.status === 502 || response.status === 503) {
          logWarn(`[CampaignService] Upstream down (${response.status}) at batch ${batchNum} for campaign ${id}. Stopping for auto-resume.`);
          await this.handleUpstreamDown(id, `Batch ${batchNum}`);
          return sentCount;
        }

        if (!response.ok) {
          const errorText = await response.text();
          // User requested to ignore 408, 404 and other 4xx errors as emails might still be sent
          // 401 and 403 are still fatal as they indicate authentication/authorization issues
          const isNonFatal = response.status >= 400 && response.status < 500 && ![401, 403].includes(response.status);

          if (isNonFatal) {
            logError(
              `[CampaignService] Batch ${batchNum} non-fatal error (${response.status}) for ${id}. Proceeding to next batch.`,
            );
            await this.campaignRepo.logSystemActivity(id, `Batch ${batchNum} Warning`, {
              error: `Notification API returned ${response.status}: ${errorText}. Continuing as per request (emails may have been sent).`,
            });
          } else {
            logError(
              `[CampaignService] Notification API Batch Failed for ${id}: ${response.status} - ${errorText}`,
            );
            await this.campaignRepo.logSystemActivity(id, `Batch ${batchNum} Failed`, {
              error: `Notification API failed: ${response.status} ${errorText}`,
            });
            throw new Error(
              `Notification API failed: ${response.status} ${errorText}`,
            );
          }
        }

        let deliveryStats: any = null;
        if (response.ok) {
          try {
            const resData = await response.json();
            if (resData.success && resData.data) {
              deliveryStats = resData.data;
            }
          } catch (err) {
            logWarn(`[CampaignService] Failed to parse success response for campaign ${id}: ${err}`);
          }
        }

        // Only increment by regularChunk length to track actual progress
        // through the target list (ignoring monitoring copies)
        sentCount += regularChunk.length;
        await this.campaignRepo.incrementSentCount(id, regularChunk.length);

        // Map chunk to statuses based on deliveryStats
        const recipientStatuses = chunk.map(email => {
          const lowerEmail = email.toLowerCase();
          if (deliveryStats) {
            if (deliveryStats.invalidEmailAddresses?.list?.some((e: string) => e.toLowerCase() === lowerEmail)) {
              return { email, status: "INVALID" };
            }
            if (deliveryStats.nonProcessableEmailAddresses?.list?.some((e: string) => e.toLowerCase() === lowerEmail)) {
              return { email, status: "NON_PROCESSABLE" };
            }
            if (deliveryStats.processableEmailAddresses?.list?.some((e: string) => e.toLowerCase() === lowerEmail)) {
              return { email, status: "SENT" };
            }
          }
          return { email, status: "SENT" }; // Default to SENT if not specified or stats missing
        });

        if (deliveryStats) {
          totalInvalid += deliveryStats.invalidEmailAddresses?.total || 0;
          totalProcessable += deliveryStats.processableEmailAddresses?.total || 0;
          totalNonProcessable += deliveryStats.nonProcessableEmailAddresses?.total || 0;
        }

        await this.campaignRepo.logRecipients(id, recipientStatuses);
        emitCampaignProgress(id, { sent: sentCount, total: emailsCount });
        await this.campaignRepo.logSystemActivity(id, `Batch ${batchNum} Processed`, {
          chunkSize: chunk.length,
          regularCount: regularChunk.length,
          sentCount,
          totalCount: emailsCount,
          deliveryStats
        });

        if (sentCount < emailsCount) {
          const delayMinutes = readIntEnv(
            "NOTIFICATION_BATCH_DELAY_MINUTES",
            DEFAULT_BATCH_DELAY_MINUTES,
          );
          const delayMs = Math.max(1, delayMinutes) * 60 * 1000;
          const checkIntervalMs = 10_000;
          logInfo(`[CampaignService] Campaign ${id} batch ${batchNum} done (${sentCount}/${emailsCount}). Waiting ${delayMinutes}m before batch ${batchNum + 1}.`);

          let elapsed = 0;
          while (elapsed < delayMs) {
            await new Promise((resolve) =>
              setTimeout(resolve, Math.min(checkIntervalMs, delayMs - elapsed)),
            );
            elapsed += checkIntervalMs;

            const delayStatus = (await this.campaignRepo.getById(id))?.status;
            if (delayStatus === "STOPPED" || delayStatus === "QUEUED" || delayStatus === "DELETED") {
              if (delayStatus === "STOPPED") {
                emitCampaignStatus(id, "STOPPED");
              }
              return sentCount;
            }
          }
        }
      }

      // Update campaign with aggregated delivery summary
      const finalCampaign = await this.campaignRepo.getById(id);
      if (finalCampaign) {
        const currentAdditionalData = typeof finalCampaign.additional_data === 'string'
          ? JSON.parse(finalCampaign.additional_data)
          : finalCampaign.additional_data;

        await this.campaignRepo.update(id, {
          additional_data: {
            ...currentAdditionalData,
            delivery_summary: {
              invalid: totalInvalid,
              processable: totalProcessable,
              non_processable: totalNonProcessable,
              last_updated: new Date().toISOString()
            }
          }
        });
      }

      await this.campaignRepo.updateStatus(id, "SENT");
      emitCampaignStatus(id, "SENT");
      return emailsCount;
    } catch (e: any) {
      let errorMessage = e.message || String(e);
      if (e.name === "AbortError" || errorMessage.includes("aborted")) {
        errorMessage = `Request timed out after 15 minutes. Batch size (${CHUNK_SIZE}) might be too large for the notification API.`;
      }

      logError(`[CampaignService] Critical error in campaign ${id}: ${errorMessage}`);
      await this.campaignRepo.updateStatus(id, "FAILED");
      emitCampaignStatus(id, "FAILED");
      await this.campaignRepo.logSystemActivity(id, "Campaign Processing Failed", {
        error: errorMessage,
      });
      throw e;
    } finally {
      CampaignService.activeCampaigns.delete(id);
      logInfo(`[CampaignService] Campaign ${id} released its slot — still sending in this instance: [${CampaignService.getActiveCampaignIds().join(", ") || "none"}]`);
    }
  }

  async processScheduledCampaigns() {
    if (UpstreamMonitor.isUpstreamDown()) {
      return;
    }

    const pending = await this.campaignRepo.getScheduledPending();
    if (pending.length === 0) return;

    try {
      const channel = await rabbitMQ.getChannel();
      const queueName = getNotificationQueueName();
      await channel.assertQueue(queueName, { durable: true });

      for (const campaign of pending) {
        try {
          channel.sendToQueue(queueName, Buffer.from(campaign.id), {
            persistent: true,
          });

          emitCampaignStatus(campaign.id, "QUEUED");
          // logInfo(`Campaign ${campaign.id} enqueued to RabbitMQ.`);
        } catch (e) {
          logError(`Failed to enqueue scheduled campaign ${campaign.id}: ${e}`);
        }
      }
    } catch (e) {
      logError(`RabbitMQ Enqueue Error: ${e}`);
    }
  }

  async getAudienceCount(criteria: any) {
    return await this.memberRepo.countByCriteria(criteria);
  }

  async getAudienceMembers(body: { criteria: any, limit?: number, offset?: number, search?: string }) {
    let emails = await this.memberRepo.findEmailsByCriteria(body.criteria);
    if (body.search) {
      const searchLower = body.search.toLowerCase();
      emails = emails.filter(email => email.toLowerCase().includes(searchLower));
    }
    const limit = body.limit || 50;
    const offset = body.offset || 0;
    const paginated = emails.slice(offset, offset + limit);
    return {
      items: paginated.map(email => ({ email, status: 'QUEUED / TARGETED', sent_at: null })),
      total: emails.length
    };
  }

  async sendTestCampaign(payload: {
    title: string;
    body: string;
    name: string;
    image_url?: string;
    target_email: string;
    additional_data?: any;
  }) {
    try {
      const customDataArr = payload.additional_data?.customData || [];
      const parsedCustomData = customDataArr.reduce((acc: any, curr: any) => {
        if (curr && curr.key) {
          acc[curr.key] = curr.value;
        }
        return acc;
      }, {});

      let campaignID = payload.additional_data?.campaignID;
      if (!campaignID) {
        const timestamp = Date.now();
        campaignID = `campaign_${timestamp}`;
      }

      const body = {
        member_emails: payload.target_email
          .split(",")
          .map((email) => email.trim())
          .filter(Boolean),
        title: payload.title,
        body: payload.body,
        name: payload.name || "test_adding",
        campaignID,
        ...(payload.image_url ? { image: encodeURI(payload.image_url) } : {}),
        additional_data: {
          ...parsedCustomData,
          path: parsedCustomData.path || "",
          slug: parsedCustomData.slug || "",
          campaign_id: campaignID,
          campaignID,
        },
        persist: payload.additional_data?.persist ?? true,
      };

      const authManager = UpdotCoreAuthManager.getInstance();
      let session = await authManager.getOrLogin();
      if (!session) {
        throw new Error("Unable to obtain Updot core authentication session.");
      }

      const accountInfo = authManager.getAccountInfo();
      const getHeaders = (sess: any) => ({
        "Content-Type": "application/json",
        "x-account": accountInfo.accountId,
        "x-contact": accountInfo.contactId,
        Cookie: `k_session=${sess.kc_session}; k_session_id=${sess.kc_session_id};`,
      });

      let response = await fetch(NOTIFICATION_API_URL, {
        method: "POST",
        headers: getHeaders(session),
        body: JSON.stringify(body),
      });

      if ([401, 403].includes(response.status)) {
        authManager.invalidate();
        session = await authManager.getOrLogin();
        if (session) {
          response = await fetch(NOTIFICATION_API_URL, {
            method: "POST",
            headers: getHeaders(session),
            body: JSON.stringify(body),
          });
        }
      }

      if (!response.ok) {
        const errorText = await response.text();
        throw new Error(
          `Failed to send test notification: ${response.status} ${errorText}`,
        );
      }

      return true;
    } catch (e) {
      logError(`Error sending test campaign: ${e}`);
      throw e;
    }
  }

  async list(
    limit: number = 50,
    offset: number = 0,
    filters?: {
      status?: string;
      search?: string;
      country?: string;
      city?: string;
      /** Comma-separated platform names, matched against target_criteria.userSegment. */
      platform?: string;
      fromDate?: string;
      toDate?: string;
      sortBy?: string;
      sortOrder?: 'asc' | 'desc';
    },
  ) {
    const campaignsResult = await this.campaignRepo.list(limit, offset, filters);

    const statsMap = await this.memberRepo.getNotificationStatsForCampaigns(
      campaignsResult.items.map((c) => ({ id: c.id, campaignID: c.campaignID }))
    );

    const itemsWithStats = campaignsResult.items.map((campaign) => {
      const stats = statsMap[campaign.id] || { unseen: 0, seen: 0, total: 0 };
      const currentAdditionalData = typeof campaign.additional_data === "string"
        ? JSON.parse(campaign.additional_data)
        : campaign.additional_data || {};

      return {
        ...campaign,
        click_count: stats.seen,
        open_count: stats.seen,
        additional_data: {
          ...currentAdditionalData,
          unseen_count: stats.unseen,
          seen_count: stats.seen,
          notification_total: stats.total
        }
      };
    });

    return {
      items: itemsWithStats,
      total: campaignsResult.total
    };
  }

  // Aggregates sends/opens over a full date window (no pagination), grouped
  // into day buckets (7d/30d/custom) or week buckets (90d) — independent of
  // the campaigns table's own page/pageSize/search/status filters.
  async getAnalytics(params: { range?: "7d" | "30d" | "90d"; fromDate?: string; toDate?: string }) {
    const today = new Date();
    today.setHours(23, 59, 59, 999);

    let windowStart: Date;
    let windowEnd = today;
    let bucketMode: "day" | "week";

    if (params.fromDate && params.toDate) {
      windowStart = new Date(`${params.fromDate}T00:00:00.000Z`);
      windowEnd = new Date(`${params.toDate}T23:59:59.999Z`);
      bucketMode = "day";
    } else {
      const range = params.range || "30d";
      const dayCount = range === "7d" ? 7 : range === "90d" ? 90 : 30;
      windowStart = new Date(today);
      windowStart.setDate(windowStart.getDate() - (dayCount - 1));
      windowStart.setHours(0, 0, 0, 0);
      bucketMode = range === "90d" ? "week" : "day";
    }

    const campaigns = await this.campaignRepo.listForAnalytics({
      fromDate: windowStart,
      toDate: windowEnd,
    });

    const statsMap = await this.memberRepo.getNotificationStatsForCampaigns(
      campaigns.map((c) => ({ id: c.id, campaignID: c.campaignID })),
    );

    type Bucket = {
      label: string;
      startKey: string;
      endKey: string;
      sends: number;
      opens: number;
    };
    const buckets: Bucket[] = [];

    if (bucketMode === "week") {
      for (let w = 0; w < 12; w++) {
        const weekStart = new Date(windowStart);
        weekStart.setDate(weekStart.getDate() + w * 7);
        const weekEnd = new Date(weekStart);
        weekEnd.setDate(weekEnd.getDate() + 6);
        if (weekEnd > windowEnd) weekEnd.setTime(windowEnd.getTime());

        buckets.push({
          label: weekStart.toLocaleDateString("en-GB", { day: "numeric", month: "short" }),
          startKey: weekStart.toISOString().split("T")[0],
          endKey: weekEnd.toISOString().split("T")[0],
          sends: 0,
          opens: 0,
        });
      }
    } else {
      const cursor = new Date(windowStart);
      while (cursor <= windowEnd) {
        const key = cursor.toISOString().split("T")[0];
        buckets.push({
          label: cursor.toLocaleDateString("en-GB", { day: "numeric", month: "short" }),
          startKey: key,
          endKey: key,
          sends: 0,
          opens: 0,
        });
        cursor.setDate(cursor.getDate() + 1);
      }
    }

    let totalSends = 0;
    let totalOpens = 0;
    for (const c of campaigns) {
      const key = new Date(c.scheduled_at || c.created_at).toISOString().split("T")[0];
      const sends = c.sent_count || 0;
      const opens = statsMap[c.id]?.seen || 0;
      totalSends += sends;
      totalOpens += opens;
      for (const bucket of buckets) {
        if (key >= bucket.startKey && key <= bucket.endKey) {
          bucket.sends += sends;
          bucket.opens += opens;
          break;
        }
      }
    }

    return {
      buckets: buckets.map((b) => ({
        date: b.label,
        sends: b.sends,
        received: Math.round(b.sends * 0.95),
        impressions: Math.round(b.sends * 0.92),
        opens: b.opens,
      })),
      summary: {
        campaignsSent: campaigns.length,
        sends: totalSends,
        received: Math.round(totalSends * 0.95),
        impressions: Math.round(totalSends * 0.92),
        opens: totalOpens,
      },
    };
  }

  async getTargetMetadata(query: { country?: string }) {
    const selectedCountries = query.country
      ? query.country
        .split(",")
        .map((c) => c.trim())
        .filter(Boolean)
      : undefined;

    const fetch = async () => {
      const [countries, cities, membershipTypes, membershipStatuses, clubs] = await Promise.all([
        this.memberRepo.getCountries(),
        this.memberRepo.getCities(selectedCountries),
        this.memberRepo.getMembershipTypes(),
        this.memberRepo.getAccountStatuses(),
        this.memberRepo.getClubs(),
      ]);
      return { countries, cities, membershipTypes, membershipStatuses, clubs };
    };

    let raw: Awaited<ReturnType<typeof fetch>>;
    try {
      raw = await fetch();
    } catch (e: any) {
      if (
        typeof e?.message === "string" &&
        (e.message.includes("Connection terminated") ||
          e.message.includes("timeout exceeded"))
      ) {
        logError(
          `[CampaignService] getTargetMetadata retrying after connection error: ${e.message}`,
        );
        raw = await fetch();
      } else {
        throw e;
      }
    }

    const result = {
      countries: raw.countries.map((c: any) => c.country).filter(Boolean),
      cities: raw.cities.map((c: any) => c.city).filter(Boolean),
      membershipTypes: raw.membershipTypes,
      membershipStatuses: raw.membershipStatuses,
      clubs: raw.clubs,
      segments: ["Android", "iOS", "Web"],
    };

    return result;
  }

  async deleteCampaign(id: string) {
    return await this.campaignRepo.delete(id);
  }

  async getCampaign(id: string) {
    const campaign = await this.campaignRepo.getById(id);
    if (!campaign) {
      throw new Error("Campaign not found");
    }

    const stats = await this.memberRepo.getNotificationStatsForCampaign(campaign.id, campaign.campaignID);
    const currentAdditionalData = typeof campaign.additional_data === "string"
      ? JSON.parse(campaign.additional_data)
      : campaign.additional_data || {};

    let emails: string[] = [];
    if (campaign.status === "SENT" || campaign.status === "COMPLETED") {
      const recResult = await this.campaignRepo.getRecipients(campaign.id, 999999, 0);
      emails = recResult.items.map((r: any) => r.email);
    } else {
      const criteria = typeof campaign.target_criteria === 'string'
        ? JSON.parse(campaign.target_criteria || '{}')
        : campaign.target_criteria || {};
      emails = await this.memberRepo.findEmailsByCriteria(criteria);
    }

    const demographics = await this.memberRepo.getDemographicsForEmails(emails);

    return {
      ...campaign,
      click_count: stats.seen,
      open_count: stats.seen,
      additional_data: {
        ...currentAdditionalData,
        unseen_count: stats.unseen,
        seen_count: stats.seen,
        notification_total: stats.total,
        demographics
      }
    };
  }

  async activateCampaign(id: string) {
    const campaign = await this.campaignRepo.getById(id);
    if (!campaign) {
      throw new Error("Campaign not found");
    }
    await this.campaignRepo.updateStatus(id, "DRAFT");
    emitCampaignStatus(id, "DRAFT");
    return true;
  }

  async recoverStuckCampaigns() {
    logInfo("Recovering stuck campaigns...");
    return await this.campaignRepo.resetSendingCampaigns();
  }

  async enqueueCampaign(id: string) {
    const queueName = getNotificationQueueName();

    // Update status BEFORE publishing to avoid race condition where
    // worker finishes (SENT/FAILED) before this line runs and gets overwritten
    await this.campaignRepo.updateStatus(id, "QUEUED");
    emitCampaignStatus(id, "QUEUED");

    const channel = await rabbitMQ.getChannel();
    await channel.assertQueue(queueName, { durable: true });
    const sent = channel.sendToQueue(queueName, Buffer.from(id), { persistent: true });
  }

  async getCampaignRecipients(id: string, limit: number, offset: number, search?: string, status?: string, seenStatus?: string) {
    if (seenStatus) {
      const allResult = await this.campaignRepo.getRecipients(id, 999999, 0, search, status);
      const emails = allResult.items.map((r: any) => r.email).filter(Boolean);

      let seenMap = new Map<string, string>();
      try {
        seenMap = await this.memberRepo.getSeenStatusByEmails(id, emails);
      } catch (err) {
        console.error("[CampaignService] Failed to enrich seen status for filter:", err);
      }

      const filteredItems = allResult.items.map((r: any) => ({
        ...r,
        seen_status: r.seen_status ?? seenMap.get(r.email) ?? "unseen",
      })).filter((r: any) => r.seen_status.toLowerCase() === seenStatus.toLowerCase());

      return {
        items: filteredItems.slice(offset, offset + limit),
        total: filteredItems.length
      };
    }

    const result = await this.campaignRepo.getRecipients(id, limit, offset, search, status);

    // Enrich with seen/unseen status from the external (Updot) DB.
    // We cannot JOIN across DBs, so we do a second lookup keyed by email.
    try {
      const emails = result.items.map((r: any) => r.email).filter(Boolean);
      const seenMap = await this.memberRepo.getSeenStatusByEmails(id, emails);
      result.items = result.items.map((r: any) => ({
        ...r,
        seen_status: r.seen_status ?? seenMap.get(r.email) ?? "unseen",
      }));
    } catch (err) {
      console.error("[CampaignService] Failed to enrich seen status:", err);
      // Gracefully degrade — leave items without seen_status rather than failing
      result.items = result.items.map((r: any) => ({
        ...r,
        seen_status: r.seen_status ?? "unseen",
      }));
    }

    return result;
  }


  async stopCampaign(id: string) {
    await this.campaignRepo.updateStatus(id, "STOPPED");
    emitCampaignStatus(id, "STOPPED");
    return true;
  }

  async resumeCampaign(id: string) {
    const campaign = await this.campaignRepo.getById(id);
    if (!campaign) {
      throw new Error("Campaign not found");
    }
    if (campaign.status !== "STOPPED" && campaign.status !== "FAILED") {
      throw new Error("Only stopped or failed campaigns can be resumed");
    }
    await this.enqueueCampaign(id);
    return true;
  }

  private async handleUpstreamDown(campaignId: string, label: string) {
    // Notify monitor of the outage
    await UpstreamMonitor.notifyDown(`CampaignService:${label}`);

    // Update current campaign to QUEUED so it can be picked up later
    await this.campaignRepo.updateStatus(campaignId, "QUEUED");
    emitCampaignStatus(campaignId, "QUEUED");

    await this.campaignRepo.logSystemActivity(campaignId, "Upstream Down - Paused", {
      message: `Detected gateway error during ${label}. Campaign moved to QUEUED for automatic resume.`,
      timestamp: new Date().toISOString(),
    });
  }
}