/**
 * SQS Poller Service
 * 
 * Polls SQS queues at regular intervals using NestJS Schedule (@Cron decorator).
 * This is the core component that:
 * 1. Polls each configured queue using long polling
 * 2. Dispatches messages to appropriate processors via ProcessorRegistry
 * 3. Deletes successfully processed messages
 * 4. Handles errors and retries
 * 
 * IMPORTANT: This runs in your existing NestJS process - no separate worker needed!
 */

import { Injectable, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { Cron, CronExpression } from '@nestjs/schedule';
import { SqsClientService } from './sqs-client.service';
import { ProcessorRegistryService } from './processor-registry.service';
import { getQueueUrls, getQueueFeatureFlags } from '../config/sqs.config';
import { QUEUE_CONSTANTS, QueueType } from '../constants/queue.constants';
import { Message } from '@aws-sdk/client-sqs';
import { QueueMessage } from '../interfaces/queue-message.interface';
import { AppLoggerService } from 'src/common/services/logger.service';

interface QueueConfig {
  queueType: QueueType;
  queueUrl: string;
  enabled: boolean;
}

@Injectable()
export class SqsPollerService implements OnModuleInit, OnModuleDestroy {
  private readonly queueConfigs: QueueConfig[] = [];
  // Per-queue in-flight flags — each queue independently guards itself so a slow
  // heavy-processing job does not block the communication queue on the next cron tick.
  private readonly isPollingQueue = new Map<QueueType, boolean>();
  private pollingStats = new Map<QueueType, {
    // Cumulative counters since last app start
    totalPolled: number;
    totalProcessed: number;
    totalFailed: number;
    totalMalformed: number;      // infrastructure junk: no body, can't parse — excluded from success rate
    totalQueueWaitMs: number;
    totalProcessingMs: number;
    lastPolledAt?: Date;
    // Window counters — reset each stats log cycle (every 5 min)
    windowPolled: number;
    windowProcessed: number;
    windowFailed: number;
  }>();

  constructor(
    private readonly sqsClient: SqsClientService,
    private readonly processorRegistry: ProcessorRegistryService,
    private readonly logger: AppLoggerService,
  ) {
    this.initializeQueueConfigs();
  }

  /**
   * Initialize queue configurations from environment
   * Simplified to 2 queues: communication and heavy-processing
   */
  private initializeQueueConfigs(): void {
    const queueUrls = getQueueUrls();
    const featureFlags = getQueueFeatureFlags();

    // Map queue types to their URLs and feature flags
    const configs: Array<{ type: QueueType; url: string | undefined; enabled: boolean }> = [
      {
        type: QUEUE_CONSTANTS.QUEUE_TYPES.COMMUNICATION,
        url: queueUrls.communication,
        enabled: featureFlags.communicationQueueEnabled,
      },
      {
        type: QUEUE_CONSTANTS.QUEUE_TYPES.HEAVY_PROCESSING,
        url: queueUrls.heavyProcessing,
        enabled: featureFlags.heavyProcessingQueueEnabled,
      },
      {
        type: QUEUE_CONSTANTS.QUEUE_TYPES.EVENT,
        url: queueUrls.event,
        enabled: featureFlags.eventQueueEnabled,
      },
    ];

    // Filter out queues that are not configured or not enabled
    for (const config of configs) {
      if (config.url && config.enabled) {
        this.queueConfigs.push({
          queueType: config.type,
          queueUrl: config.url,
          enabled: config.enabled,
        });
        this.pollingStats.set(config.type, {
          totalPolled: 0,
          totalProcessed: 0,
          totalFailed: 0,
          totalMalformed: 0,
          totalQueueWaitMs: 0,
          totalProcessingMs: 0,
          windowPolled: 0,
          windowProcessed: 0,
          windowFailed: 0,
        });
        this.isPollingQueue.set(config.type, false);
      } else if (!config.url) {
        this.logger.warn(
          `Queue URL for '${config.type}' is not configured. Skipping this queue.`
        );
      } else if (!config.enabled) {
        this.logger.log(
          `Queue '${config.type}' is disabled via feature flag. Skipping this queue.`
        );
      }
    }

    this.logger.log(
      `Initialized ${this.queueConfigs.length} queue(s): ${this.queueConfigs.map(q => q.queueType).join(', ')}`
    );
  }

  /**
   * Lifecycle hook - called when module is initialized
   */
  async onModuleInit() {
    this.logger.log('SQS Poller Service initialized');
    
    // Validate that processors are registered for enabled queues
    // Check by subType (email/whatsapp/sms/invoice) not queueType (communication/heavy-processing)
    for (const config of this.queueConfigs) {
      const expectedSubTypes = this.getExpectedSubTypesForQueue(config.queueType);
      const missingProcessors: string[] = [];
      
      for (const subType of expectedSubTypes) {
        if (!this.processorRegistry.hasProcessor(subType)) {
          missingProcessors.push(subType);
        }
      }
      
      if (missingProcessors.length > 0) {
        this.logger.warn(
          `Missing processor(s) for enabled queue '${config.queueType}': ${missingProcessors.join(', ')}. ` +
          `Messages of these types will fail to process until processors are registered.`
        );
      } else {
        this.logger.log(
          `All processors registered for queue '${config.queueType}': ${expectedSubTypes.join(', ')}`
        );
      }
    }
  }

  /**
   * Get expected subTypes for a given queue type
   * Maps queue types to their expected message subTypes
   */
  private getExpectedSubTypesForQueue(queueType: QueueType): string[] {
    if (queueType === QUEUE_CONSTANTS.QUEUE_TYPES.COMMUNICATION) {
      return [
        QUEUE_CONSTANTS.COMMUNICATION_TYPES.EMAIL,
        QUEUE_CONSTANTS.COMMUNICATION_TYPES.BULK_EMAIL,
        QUEUE_CONSTANTS.COMMUNICATION_TYPES.WHATSAPP,
        QUEUE_CONSTANTS.COMMUNICATION_TYPES.BULK_WHATSAPP,
        QUEUE_CONSTANTS.COMMUNICATION_TYPES.SMS,
      ];
    } else if (queueType === QUEUE_CONSTANTS.QUEUE_TYPES.HEAVY_PROCESSING) {
      return [
        QUEUE_CONSTANTS.HEAVY_PROCESSING_TYPES.INVOICE,
        // Add other heavy processing types as they're implemented
      ];
    } else if (queueType === QUEUE_CONSTANTS.QUEUE_TYPES.EVENT) {
      return [
        QUEUE_CONSTANTS.EVENT_TYPES.JOIN_LINK_GENERATION,
        QUEUE_CONSTANTS.EVENT_TYPES.SESSION_COMMUNICATION_TRIGGER,
      ];
    }
    return [];
  }

  /**
   * Lifecycle hook - called when module is being destroyed
   */
  onModuleDestroy() {
    this.logger.log('SQS Poller Service shutting down');
  }

  /**
   * Periodic stats reporter - runs every minute.
   * Emits [QUEUE_STATS] structured logs queryable in CloudWatch Logs Insights.
   */
  @Cron(CronExpression.EVERY_5_MINUTES)
  logPollingStats(): void {
    if (this.queueConfigs.length === 0) return;

    this.pollingStats.forEach((stats, queueType) => {
      // Exclude malformed messages from success rate — they are infrastructure junk
      // (no body, unparseable) not real processing failures, and would skew the metric.
      const validPolled = stats.totalPolled - stats.totalMalformed;
      const successRate = validPolled > 0
        ? ((stats.totalProcessed / validPolled) * 100).toFixed(1) + '%'
        : 'N/A';

      // avgQueueWaitMs uses totalPolled (all valid parsed messages, success + failure)
      // so the average is not biased towards only successfully processed messages.
      const avgQueueWaitMs = stats.totalPolled > 0
        ? (stats.totalQueueWaitMs / stats.totalPolled).toFixed(1)
        : 'N/A';
      const avgProcessingMs = stats.totalProcessed > 0
        ? (stats.totalProcessingMs / stats.totalProcessed).toFixed(1)
        : 'N/A';

      const windowSuccessRate = stats.windowPolled > 0
        ? ((stats.windowProcessed / stats.windowPolled) * 100).toFixed(1) + '%'
        : 'N/A';

      this.logger.log('[QUEUE_STATS]', {
        event: 'QUEUE_STATS',
        queueType,
        // Cumulative since app start
        totalPolled: stats.totalPolled,
        totalProcessed: stats.totalProcessed,
        totalFailed: stats.totalFailed,
        totalMalformed: stats.totalMalformed,
        successRate,
        avgQueueWaitMs,
        avgProcessingMs,
        lastPolledAt: stats.lastPolledAt,
        // Last 5-minute window — useful for detecting sudden spikes during load tests
        windowPolled: stats.windowPolled,
        windowProcessed: stats.windowProcessed,
        windowFailed: stats.windowFailed,
        windowSuccessRate,
      });

      // Reset window counters for the next period
      stats.windowPolled = 0;
      stats.windowProcessed = 0;
      stats.windowFailed = 0;
    });
  }

  /**
   * Main polling job - runs every 10 seconds
   * Uses cron expression to run at precise intervals
   * 
   * Cron pattern: every 10 seconds
   * You can adjust this in queue.constants.ts
   */
  @Cron(CronExpression.EVERY_10_SECONDS)
  async pollQueues() {
    if (this.queueConfigs.length === 0) return;

    // Each queue manages its own in-flight guard (isPollingQueue) so queues
    // run fully independently — a slow invoice job never delays email/SMS polling.
    await Promise.all(
      this.queueConfigs.map(config => this.pollSingleQueue(config))
    );
  }

  /**
   * Poll a single queue and process all received messages
   */
  private async pollSingleQueue(config: QueueConfig): Promise<void> {
    // Per-queue guard: skip this tick if the previous poll is still processing.
    // This prevents a backlog on one queue from stacking up concurrent executions.
    if (this.isPollingQueue.get(config.queueType)) {
      this.logger.debug(
        `Previous poll for '${config.queueType}' still in progress, skipping this tick`,
      );
      return;
    }

    this.isPollingQueue.set(config.queueType, true);

    try {
      // Use different visibility timeouts based on queue type
      const visibilityTimeout =
        config.queueType === QUEUE_CONSTANTS.QUEUE_TYPES.HEAVY_PROCESSING
          ? QUEUE_CONSTANTS.HEAVY_PROCESSING_VISIBILITY_TIMEOUT_SECONDS
          : config.queueType === QUEUE_CONSTANTS.QUEUE_TYPES.EVENT
            ? QUEUE_CONSTANTS.EVENT_VISIBILITY_TIMEOUT_SECONDS
            : QUEUE_CONSTANTS.COMMUNICATION_VISIBILITY_TIMEOUT_SECONDS;

      // Use long polling to wait for messages (reduces empty responses)
      const messages = await this.sqsClient.receiveMessages(
        config.queueUrl,
        QUEUE_CONSTANTS.MAX_MESSAGES_PER_POLL,
        QUEUE_CONSTANTS.LONG_POLL_WAIT_TIME_SECONDS,
        visibilityTimeout,
      );

      if (messages.length === 0) {
        // No messages - this is normal, no need to log
        return;
      }

      // Update stats and emit [POLL_CYCLE] BEFORE processing so the timestamp
      // accurately reflects when messages were received, not when they finished.
      const stats = this.pollingStats.get(config.queueType);
      if (stats) {
        stats.totalPolled += messages.length;
        stats.windowPolled += messages.length;
        stats.lastPolledAt = new Date();
      }

      this.logger.log('[POLL_CYCLE]', {
        event: 'POLL_CYCLE',
        queueType: config.queueType,
        received: messages.length,
        timestamp: new Date().toISOString(),
      });

      // Process all messages concurrently
      await Promise.all(
        messages.map(message => this.processMessage(message, config))
      );
    } catch (error) {
      this.logger.error(
        `Error polling queue: ${config.queueType}`,
        '',
        { error: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined, queueType: config.queueType },
      );
    } finally {
      this.isPollingQueue.set(config.queueType, false);
    }
  }

  /**
   * Process a single message
   */
  private async processMessage(message: Message, config: QueueConfig): Promise<void> {
    const stats = this.pollingStats.get(config.queueType);
    // Declared outside try so the catch block can read attemptCount for the max-retry guard.
    let queueMessage: QueueMessage | null = null;

    try {
      // Hard guard: without a receipt handle we cannot delete; avoid crashing.
      if (!message.ReceiptHandle) {
        this.logger.error(
          `Received message without ReceiptHandle for ${config.queueType}. MessageId: ${message.MessageId}. Skipping processing.`,
          '',
          { queueType: config.queueType, messageId: message.MessageId },
        );
        if (stats) { stats.totalFailed++; stats.totalMalformed++; stats.windowFailed++; }
        return;
      }

      // Treat missing body as malformed/non-retryable and delete immediately.
      if (!message.Body) {
        this.logger.warn(
          `Received message with empty body for ${config.queueType}. MessageId: ${message.MessageId}. Deleting malformed message.`,
          { queueType: config.queueType, messageId: message.MessageId },
        );
        await this.sqsClient.deleteMessage(config.queueUrl, message.ReceiptHandle);
        if (stats) { stats.totalFailed++; stats.totalMalformed++; stats.windowFailed++; }
        return;
      }

      // Parse the message body
      queueMessage = this.sqsClient.parseMessageBody(message);

      if (!queueMessage) {
        this.logger.error(
          `Failed to parse message body for ${config.queueType}. ` +
          `MessageId: ${message.MessageId}. Deleting malformed message.`,
          '',
          { queueType: config.queueType, messageId: message.MessageId },
        );
        await this.sqsClient.deleteMessage(config.queueUrl, message.ReceiptHandle);
        if (stats) { stats.totalFailed++; stats.totalMalformed++; stats.windowFailed++; }
        return;
      }

      // Add message ID from SQS
      queueMessage.messageId = message.MessageId;

      // App-level safety net: if the message has already been attempted MAX_RECEIVE_COUNT
      // times or more, force-delete it rather than retrying again. This guards against the
      // case where the SQS DLQ redrive policy is misconfigured or not attached, which would
      // otherwise allow messages to retry indefinitely.
      const attemptCount = queueMessage.attemptCount ?? 1;
      if (attemptCount > QUEUE_CONSTANTS.MAX_RECEIVE_COUNT) {
        this.logger.error('[QUEUE_MAX_RETRIES_EXCEEDED]', '', {
          event: 'QUEUE_MAX_RETRIES_EXCEEDED',
          queueType: config.queueType,
          subType: queueMessage.subType,
          messageId: queueMessage.messageId,
          correlationId: queueMessage.correlationId,
          attemptCount,
          maxAttempts: QUEUE_CONSTANTS.MAX_RECEIVE_COUNT,
        });
        await this.sqsClient.deleteMessage(config.queueUrl, message.ReceiptHandle);
        if (stats) { stats.totalFailed++; stats.windowFailed++; }
        return;
      }

      // Calculate how long this message waited in the queue before being picked up.
      // Math.max(0, ...) guards against minor clock drift between the enqueuing server
      // and the processing server producing a spurious negative value.
      const queueWaitMs = queueMessage.timestamp
        ? Math.max(0, Date.now() - new Date(queueMessage.timestamp).getTime())
        : 0;

      // Accumulate queueWaitMs for ALL valid messages (success AND failure) so the
      // average is not biased towards only the happy path.
      if (stats) {
        stats.totalQueueWaitMs += queueWaitMs;
      }

      // Process the message via registry
      const result = await this.processorRegistry.processMessage(queueMessage);

      if (result.success) {
        // Success! Delete the message from the queue
        await this.sqsClient.deleteMessage(config.queueUrl, message.ReceiptHandle);

        if (stats) {
          stats.totalProcessed++;
          stats.totalProcessingMs += result.processingTimeMs ?? 0;
          stats.windowProcessed++;
        }

        this.logger.log('[QUEUE_PROCESSED]', {
          event: 'QUEUE_PROCESSED',
          queueType: config.queueType,
          subType: queueMessage.subType,
          messageId: queueMessage.messageId,
          correlationId: queueMessage.correlationId,
          processingTimeMs: result.processingTimeMs,
          queueWaitMs,
          attemptCount: queueMessage.attemptCount,
        });
      } else {
        // Processing failed
        if (stats) { stats.totalFailed++; stats.windowFailed++; }

        if (result.error?.retryable) {
          // Message will become visible again after visibility timeout
          // SQS will automatically retry based on redrive policy
          this.logger.warn('[QUEUE_FAILED]', {
            event: 'QUEUE_FAILED',
            queueType: config.queueType,
            subType: queueMessage.subType,
            messageId: queueMessage.messageId,
            correlationId: queueMessage.correlationId,
            errorCode: result.error.code,
            errorMessage: result.error.message,
            retryable: true,
            attemptCount: queueMessage.attemptCount,
            maxAttempts: QUEUE_CONSTANTS.MAX_RECEIVE_COUNT,
            queueWaitMs,
          });
        } else {
          // Non-retryable error - delete the message to prevent infinite processing
          this.logger.error('[QUEUE_FAILED]', '', {
            event: 'QUEUE_FAILED',
            queueType: config.queueType,
            subType: queueMessage.subType,
            messageId: queueMessage.messageId,
            correlationId: queueMessage.correlationId,
            errorCode: result.error?.code,
            errorMessage: result.error?.message,
            retryable: false,
            attemptCount: queueMessage.attemptCount,
            queueWaitMs,
          });
          await this.sqsClient.deleteMessage(config.queueUrl, message.ReceiptHandle);
        }
      }
    } catch (error) {
      // Unexpected error during processing
      if (stats) { stats.totalFailed++; stats.windowFailed++; }

      this.logger.error(
        `Unexpected error processing message from ${config.queueType} queue`,
        '',
        { error: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined, queueType: config.queueType },
      );

      // If the message has already exhausted all attempts, force-delete it even on an
      // unexpected error path so it does not loop indefinitely when the DLQ is absent.
      // queueMessage may be undefined here if the error occurred before parsing; guard accordingly.
      const currentAttempt = queueMessage?.attemptCount ?? 0;
      if (message.ReceiptHandle && currentAttempt > QUEUE_CONSTANTS.MAX_RECEIVE_COUNT) {
        this.logger.error('[QUEUE_MAX_RETRIES_EXCEEDED]', '', {
          event: 'QUEUE_MAX_RETRIES_EXCEEDED',
          queueType: config.queueType,
          messageId: message.MessageId,
          attemptCount: currentAttempt,
          maxAttempts: QUEUE_CONSTANTS.MAX_RECEIVE_COUNT,
          reason: 'unexpected_error_path',
        });
        await this.sqsClient.deleteMessage(config.queueUrl, message.ReceiptHandle);
      }
      // Otherwise leave on queue and let SQS retry with exponential backoff
    }
  }

  /**
   * Get polling statistics for monitoring
   */
  getPollingStats() {
    const stats: Record<string, any> = {};

    this.pollingStats.forEach((value, key) => {
      const validPolled = value.totalPolled - value.totalMalformed;
      stats[key] = {
        ...value,
        successRate: validPolled > 0
          ? ((value.totalProcessed / validPolled) * 100).toFixed(2) + '%'
          : 'N/A',
        avgQueueWaitMs: value.totalPolled > 0
          ? (value.totalQueueWaitMs / value.totalPolled).toFixed(1)
          : 'N/A',
        avgProcessingMs: value.totalProcessed > 0
          ? (value.totalProcessingMs / value.totalProcessed).toFixed(1)
          : 'N/A',
      };
    });

    return stats;
  }

  /**
   * Manually trigger a poll for a specific queue (useful for testing)
   */
  async triggerPoll(queueType: QueueType): Promise<void> {
    const config = this.queueConfigs.find(c => c.queueType === queueType);
    
    if (!config) {
      throw new Error(`Queue type '${queueType}' is not configured or not enabled`);
    }

    await this.pollSingleQueue(config);
  }

  /**
   * Get configured queues
   */
  getConfiguredQueues(): QueueConfig[] {
    return [...this.queueConfigs];
  }
}