<?php

declare(strict_types=1);

namespace Drupal\ffmpeg_media_ui\Controller;

use Drupal\Core\Config\ConfigFactoryInterface;
use Drupal\Core\Controller\ControllerBase;
use Drupal\Core\Session\AccountInterface;
use Drupal\ffmpeg_media\Service\ProgressLabelServiceInterface;
use Drupal\ffmpeg_media\Service\ProgressTrackerInterface;
use Drupal\ffmpeg_media\Service\QueueManagerInterface;
use Drupal\ffmpeg_media\Trait\ErrorRedactionTrait;
use Drupal\media\MediaInterface;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
use Symfony\Component\HttpFoundation\StreamedResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpKernel\Exception\AccessDeniedHttpException;

/**
 * Controller for Server-Sent Events (SSE) progress streaming.
 *
 * Provides real-time progress updates, eliminating client-side polling.
 * SSE is simpler than WebSockets for unidirectional server-to-client
 * communication. The server internally polls the progress tracker.
 */
final class SseProgressController extends ControllerBase {

  use ErrorRedactionTrait;

  /**
   * Default interval between progress checks in microseconds.
   */
  protected const int DEFAULT_INTERVAL_US = 1_000_000;

  /**
   * Maximum stream duration in seconds (prevents infinite connections).
   */
  protected const int MAX_STREAM_DURATION = 300;

  /**
   * Heartbeat interval to keep connection alive (seconds).
   */
  protected const int HEARTBEAT_INTERVAL = 15;

  /**
   * Idle timeout in seconds (no data change triggers reconnect hint).
   */
  protected const int IDLE_TIMEOUT = 300;

  /**
   * Minimum allowed interval in milliseconds.
   */
  protected const int MIN_INTERVAL_MS = 500;

  /**
   * Maximum allowed interval in milliseconds.
   */
  protected const int MAX_INTERVAL_MS = 30000;

  /**
   * Constructs a SseProgressController object.
   *
   * @param \Drupal\ffmpeg_media\Service\ProgressTrackerInterface $progressTracker
   *   Progress tracker service.
   * @param \Drupal\ffmpeg_media\Service\QueueManagerInterface $queueManager
   *   Queue manager service.
   * @param \Drupal\ffmpeg_media\Service\ProgressLabelServiceInterface $progressLabel
   *   Progress label service.
   * @param \Drupal\Core\Config\ConfigFactoryInterface $moduleConfigFactory
   *   Config factory.
   */
  public function __construct(
    #[Autowire(service: ProgressTrackerInterface::class)]
    protected readonly ProgressTrackerInterface $progressTracker,
    #[Autowire(service: QueueManagerInterface::class)]
    protected readonly QueueManagerInterface $queueManager,
    #[Autowire(service: ProgressLabelServiceInterface::class)]
    protected readonly ProgressLabelServiceInterface $progressLabel,
    #[Autowire(service: 'config.factory')]
    protected readonly ConfigFactoryInterface $moduleConfigFactory,
  ) {}

  /**
   * {@inheritdoc}
   */
  #[\Override]
  protected function getErrorRedactionAccount(): ?AccountInterface {
    return $this->currentUser();
  }

  /**
   * Streams progress updates for a specific media item.
   *
   * @param \Drupal\media\MediaInterface $media
   *   The media entity.
   * @param \Symfony\Component\HttpFoundation\Request $request
   *   The current request.
   *
   * @return \Symfony\Component\HttpFoundation\StreamedResponse
   *   SSE streamed response.
   */
  public function mediaProgress(MediaInterface $media, Request $request): StreamedResponse {
    if (!$media->access('view')) {
      throw new AccessDeniedHttpException('Access denied for this media item.');
    }

    $mediaId = (int) $media->id();
    $fieldName = (string) $request->query->get('field_name', '');
    $policy = $this->getSsePolicy();
    $lastEventId = $this->getLastEventIdFromRequest($request);
    $interval = $this->getIntervalFromRequest($request, self::DEFAULT_INTERVAL_US, $policy);

    $response = new StreamedResponse(function () use ($mediaId, $fieldName, $lastEventId, $interval, $policy): void {
      $this->streamMediaProgress($mediaId, $lastEventId, $interval, $policy, $fieldName);
    });

    return $this->configureSseResponse($response, $request);
  }

  /**
   * Streams progress updates for all active conversions.
   *
   * @param \Symfony\Component\HttpFoundation\Request $request
   *   The current request.
   *
   * @return \Symfony\Component\HttpFoundation\StreamedResponse
   *   SSE streamed response.
   */
  public function queueProgress(Request $request): StreamedResponse {
    $policy = $this->getSsePolicy();
    $lastEventId = $this->getLastEventIdFromRequest($request);
    $interval = $this->getIntervalFromRequest($request, self::DEFAULT_INTERVAL_US, $policy);

    $response = new StreamedResponse(function () use ($lastEventId, $interval, $policy): void {
      $this->streamQueueProgress($lastEventId, $interval, $policy);
    });

    return $this->configureSseResponse($response, $request);
  }

  /**
   * Streams dashboard statistics updates.
   *
   * @param \Symfony\Component\HttpFoundation\Request $request
   *   The current request.
   *
   * @return \Symfony\Component\HttpFoundation\StreamedResponse
   *   SSE streamed response.
   */
  public function dashboardStats(Request $request): StreamedResponse {
    $policy = $this->getSsePolicy();
    $interval = $this->getIntervalFromRequest($request, 5_000_000, $policy);

    $response = new StreamedResponse(function () use ($interval, $policy): void {
      $this->streamDashboardStats($interval, $policy);
    });

    return $this->configureSseResponse($response, $request);
  }

  /**
   * Streams progress for a single media item.
   *
   * @param int $mediaId
   *   The media entity ID.
   * @param string|null $lastEventId
   *   Last event ID for resumption.
   * @param int $interval
   *   Check interval in microseconds.
   * @param array<string, int> $policy
   *   SSE policy settings.
   * @param string $fieldName
   *   Optional field name to scope progress lookup.
   */
  protected function streamMediaProgress(int $mediaId, ?string $lastEventId, int $interval, array $policy, string $fieldName = ''): void {
    $startTime = time();
    $lastHeartbeat = $startTime;
    $lastDataChange = $startTime;
    $eventId = $lastEventId !== NULL ? (int) $lastEventId : 0;
    $lastProgressHash = '';
    $completed = FALSE;
    $connectionAborted = FALSE;
    $idleTimeout = $policy['idle_timeout'] ?? self::IDLE_TIMEOUT;

    while ($this->shouldContinueStreaming($startTime, $policy['max_stream_duration'])) {
      $progress = $this->progressTracker->getProgress($mediaId, $fieldName);

      if ($progress !== NULL) {
        $progressData = $progress->toArray();

        // Redact internal error details for non-admin users.
        if ($this->shouldRedactError($progress, $progressData['error'] ?? NULL)) {
          $progressData['error'] = self::SAFE_ERROR_MESSAGE;
        }

        // Add humanized label and ETA.
        if ($progress->currentStep !== NULL) {
          $progressData['humanized_step'] = $this->progressLabel->humanize(
            $progress->currentStep,
            $progress->currentOutput,
            0,
            0,
          );
        }
        if ($progress->isInProgress && $progress->startedAt !== NULL) {
          $progressData['eta'] = $this->progressLabel->getEstimatedCompletion(
            $progress->startedAt,
            $progress->percent,
          );
        }

        $currentHash = md5(json_encode($progressData) ?: '');

        // Only send if data changed.
        if ($currentHash !== $lastProgressHash) {
          $eventId++;
          $this->sendSseEvent('progress', $progressData, (string) $eventId);
          $lastProgressHash = $currentHash;
          $lastDataChange = time();

          // Check if conversion is complete or failed.
          if ($progress->isComplete || $progress->hasFailed) {
            $this->sendSseEvent('complete', [
              'media_id' => $mediaId,
              'status' => $progress->status->value,
              'success' => $progress->isComplete,
            ], (string) ($eventId + 1));
            $completed = TRUE;
            break;
          }
        }
      }

      // Send heartbeat to keep connection alive.
      $now = time();
      if ($now - $lastHeartbeat >= $policy['heartbeat_interval']) {
        $this->sendSseComment('heartbeat');
        $lastHeartbeat = $now;
      }

      // Idle timeout: no data change for too long, suggest reconnect.
      if ($idleTimeout > 0 && ($now - $lastDataChange) >= $idleTimeout) {
        $this->sendSseRetryHint(10000);
        $this->sendSseEvent('idle-timeout', [
          'message' => 'No data changes detected — reconnect recommended.',
          'retry_ms' => 10000,
        ], (string) (++$eventId));
        break;
      }

      usleep($interval);

      if (connection_aborted()) {
        $connectionAborted = TRUE;
        break;
      }
    }

    if (!$completed && !$connectionAborted && !$this->shouldContinueStreaming($startTime, $policy['max_stream_duration'])) {
      $this->sendSseRetryHint(5000);
      $this->sendSseEvent('timeout', [
        'message' => 'Stream timeout reached.',
        'retry_ms' => 5000,
      ], (string) ($eventId + 1));
    }
  }

  /**
   * Streams progress for all active queue items.
   *
   * @param string|null $lastEventId
   *   Last event ID for resumption.
   * @param int $interval
   *   Check interval in microseconds.
   * @param array<string, int> $policy
   *   SSE policy settings.
   */
  protected function streamQueueProgress(?string $lastEventId, int $interval, array $policy): void {
    $startTime = time();
    $lastHeartbeat = $startTime;
    $eventId = $lastEventId !== NULL ? (int) $lastEventId : 0;
    $lastDataHash = '';
    $mediaStorage = $this->entityTypeManager()->getStorage('media');
    $cachedMediaEntities = [];
    $lastMediaIds = [];
    $cacheExpiry = 0;

    while ($this->shouldContinueStreaming($startTime, $policy['max_stream_duration'])) {
      $queueStats = $this->queueManager->getQueueStats();
      $activeItems = $this->progressTracker->getActiveConversions();

      // Filter by entity access and redact errors for non-admin users.
      $mediaIds = array_map(static fn($p) => $p->mediaId, $activeItems);
      $now = time();
      if ($mediaIds !== $lastMediaIds || $now >= $cacheExpiry) {
        if ($mediaIds !== []) {
          // Pre-filter by access to skip inaccessible entities.
          $accessibleIds = $mediaStorage->getQuery()
            ->accessCheck(TRUE)
            ->condition('mid', $mediaIds, 'IN')
            ->execute();
          $cachedMediaEntities = $accessibleIds !== []
            ? $mediaStorage->loadMultiple($accessibleIds)
            : [];
        }
        else {
          $cachedMediaEntities = [];
        }
        $lastMediaIds = $mediaIds;
        $cacheExpiry = $now + 5;
      }
      $filteredActive = [];
      foreach ($activeItems as $progress) {
        $media = $cachedMediaEntities[$progress->mediaId] ?? NULL;
        if ($media instanceof MediaInterface && $media->access('view')) {
          $progressData = $progress->toArray();
          if ($this->shouldRedactError($progress, $progressData['error'] ?? NULL)) {
            $progressData['error'] = self::SAFE_ERROR_MESSAGE;
          }
          $filteredActive[] = $progressData;
        }
      }

      $data = [
        'queue' => $queueStats,
        'active' => $filteredActive,
        'timestamp' => time(),
      ];

      $currentHash = md5(json_encode($data) ?: '');

      if ($currentHash !== $lastDataHash) {
        $eventId++;
        $this->sendSseEvent('queue-update', $data, (string) $eventId);
        $lastDataHash = $currentHash;
      }

      // Heartbeat.
      $now = time();
      if ($now - $lastHeartbeat >= $policy['heartbeat_interval']) {
        $this->sendSseComment('heartbeat');
        $lastHeartbeat = $now;
      }

      usleep($interval);

      if (connection_aborted()) {
        break;
      }
    }
  }

  /**
   * Streams dashboard statistics.
   *
   * @param int $interval
   *   Check interval in microseconds.
   * @param array<string, int> $policy
   *   SSE policy settings.
   */
  protected function streamDashboardStats(int $interval, array $policy): void {
    $startTime = time();
    $lastHeartbeat = $startTime;
    $eventId = 0;
    $lastDataHash = '';

    while ($this->shouldContinueStreaming($startTime, $policy['max_stream_duration'])) {
      $queueStats = $this->queueManager->getQueueStats();

      $data = [
        'queue' => $queueStats,
        'timestamp' => time(),
      ];

      $currentHash = md5(json_encode($data) ?: '');

      if ($currentHash !== $lastDataHash) {
        $eventId++;
        $this->sendSseEvent('stats', $data, (string) $eventId);
        $lastDataHash = $currentHash;
      }

      $now = time();
      if ($now - $lastHeartbeat >= $policy['heartbeat_interval']) {
        $this->sendSseComment('heartbeat');
        $lastHeartbeat = $now;
      }

      usleep($interval);

      if (connection_aborted()) {
        break;
      }
    }
  }

  /**
   * Sends an SSE event.
   *
   * @param string $event
   *   Event type name.
   * @param array<string, mixed> $data
   *   Event data.
   * @param string|null $id
   *   Optional event ID for resumption.
   */
  protected function sendSseEvent(string $event, array $data, ?string $id = NULL): void {
    if ($id !== NULL) {
      echo "id: {$id}\n";
    }
    echo "event: {$event}\n";
    echo 'data: ' . json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . "\n\n";

    $this->flushOutput();
  }

  /**
   * Sends an SSE retry hint to guide client reconnection timing.
   *
   * @param int $retryMs
   *   Reconnection delay in milliseconds.
   */
  protected function sendSseRetryHint(int $retryMs): void {
    echo "retry: {$retryMs}\n\n";

    $this->flushOutput();
  }

  /**
   * Sends an SSE comment (for heartbeats).
   *
   * @param string $comment
   *   Comment text.
   */
  protected function sendSseComment(string $comment): void {
    echo ": {$comment}\n\n";

    $this->flushOutput();
  }

  /**
   * Flushes output buffers to send SSE data to the client immediately.
   */
  private function flushOutput(): void {
    if (ob_get_level() > 0) {
      ob_flush();
    }
    flush();
  }

  /**
   * Configures response headers for SSE.
   *
   * @param \Symfony\Component\HttpFoundation\StreamedResponse $response
   *   The response to configure.
   * @param \Symfony\Component\HttpFoundation\Request $request
   *   The current request.
   *
   * @return \Symfony\Component\HttpFoundation\StreamedResponse
   *   Configured response.
   */
  protected function configureSseResponse(StreamedResponse $response, Request $request): StreamedResponse {
    $response->headers->set('Content-Type', 'text/event-stream');
    $response->headers->set('Cache-Control', 'no-cache, no-store, must-revalidate');
    $response->headers->set('Connection', 'keep-alive');
    $response->headers->set('X-Accel-Buffering', 'no');

    $origin = $request->headers->get('Origin');
    if ($origin !== NULL && $origin === $request->getSchemeAndHttpHost()) {
      $response->headers->set('Access-Control-Allow-Origin', $origin);
    }
    $response->headers->set('Vary', 'Origin');

    return $response;
  }

  /**
   * Checks if streaming should continue.
   *
   * @param int $startTime
   *   Stream start timestamp.
   * @param int $maxStreamDuration
   *   Maximum stream duration in seconds.
   *
   * @return bool
   *   TRUE if streaming should continue.
   */
  protected function shouldContinueStreaming(int $startTime, int $maxStreamDuration = self::MAX_STREAM_DURATION): bool {
    return (time() - $startTime) < max(1, $maxStreamDuration);
  }

  /**
   * Gets interval from request query parameter.
   *
   * @param \Symfony\Component\HttpFoundation\Request $request
   *   The request.
   * @param int $default
   *   Default interval in microseconds.
   * @param array<string, int>|null $policy
   *   Optional SSE policy settings.
   *
   * @return int
   *   Interval in microseconds.
   */
  protected function getIntervalFromRequest(Request $request, int $default = self::DEFAULT_INTERVAL_US, ?array $policy = NULL): int {
    $minIntervalMs = $policy['min_interval_ms'] ?? self::MIN_INTERVAL_MS;
    $maxIntervalMs = $policy['max_interval_ms'] ?? self::MAX_INTERVAL_MS;
    $maxIntervalMs = max($minIntervalMs, $maxIntervalMs);
    $intervalMs = $request->query->getInt('interval', (int) ($default / 1000));

    $intervalMs = max($minIntervalMs, min($maxIntervalMs, $intervalMs));

    return $intervalMs * 1000;
  }

  /**
   * Resolves Last-Event-ID from header or query fallback.
   *
   * @param \Symfony\Component\HttpFoundation\Request $request
   *   Current request.
   *
   * @return string|null
   *   Numeric last event ID, or NULL when unavailable/invalid.
   */
  protected function getLastEventIdFromRequest(Request $request): ?string {
    $lastEventId = $request->headers->get('Last-Event-ID');

    if ($lastEventId === NULL || $lastEventId === '') {
      $queryLastEventId = trim((string) $request->query->get('lastEventId', ''));
      $lastEventId = $queryLastEventId !== '' ? $queryLastEventId : NULL;
    }

    if ($lastEventId === NULL || !ctype_digit($lastEventId)) {
      return NULL;
    }

    return (string) (int) $lastEventId;
  }

  /**
   * Builds the effective SSE policy from module configuration.
   *
   * @return array<string, int>
   *   SSE policy values.
   */
  protected function getSsePolicy(): array {
    $settings = $this->moduleConfigFactory->get('ffmpeg_media_ui.settings');

    $maxStreamDuration = $this->clampInt(
      (int) ($settings->get('sse_max_stream_duration') ?? self::MAX_STREAM_DURATION),
      30,
      self::MAX_STREAM_DURATION,
    );
    $heartbeatInterval = $this->clampInt(
      (int) ($settings->get('sse_heartbeat_interval') ?? self::HEARTBEAT_INTERVAL),
      5,
      120,
    );
    $minIntervalMs = $this->clampInt(
      (int) ($settings->get('sse_min_interval_ms') ?? self::MIN_INTERVAL_MS),
      250,
      10000,
    );
    $maxIntervalMs = $this->clampInt(
      (int) ($settings->get('sse_max_interval_ms') ?? self::MAX_INTERVAL_MS),
      $minIntervalMs,
      60000,
    );
    $idleTimeout = $this->clampInt(
      (int) ($settings->get('sse_idle_timeout') ?? self::IDLE_TIMEOUT),
      0,
      3600,
    );

    return [
      'max_stream_duration' => $maxStreamDuration,
      'heartbeat_interval' => $heartbeatInterval,
      'min_interval_ms' => $minIntervalMs,
      'max_interval_ms' => $maxIntervalMs,
      'idle_timeout' => $idleTimeout,
    ];
  }

  /**
   * Clamps an integer value to a min/max range.
   *
   * @param int $value
   *   Value to clamp.
   * @param int $min
   *   Inclusive minimum.
   * @param int $max
   *   Inclusive maximum.
   *
   * @return int
   *   Clamped integer value.
   */
  protected function clampInt(int $value, int $min, int $max): int {
    return max($min, min($max, $value));
  }

}
