<?php

declare(strict_types=1);

namespace Drupal\flowdrop_session\Hook;

use Drupal\flowdrop_session\Entity\FlowDropSessionInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Hook\Attribute\Hook;
use Drupal\Core\Queue\QueueFactory;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_session\Constants\MessageStatus;
use Drupal\flowdrop_session\Constants\SessionStatus;
use Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface;
use Drupal\flowdrop_session\Service\SessionService;
use Psr\Log\LoggerInterface;

/**
 * Cron hook implementations for the flowdrop_session module.
 *
 * Processes stale messages and cleans up the queue during cron runs.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
final class CronHooks {

  /**
   * Constructs a new CronHooks object.
   *
   * @param \Drupal\flowdrop_session\Service\SessionService $sessionService
   *   The session service.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   The queue factory.
   * @param \Psr\Log\LoggerInterface $logger
   *   The logger channel.
   */
  public function __construct(
    protected readonly SessionService $sessionService,
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly QueueFactory $queueFactory,
    protected readonly LoggerInterface $logger,
  ) {}

  /**
   * Implements hook_cron().
   *
   * Processes stale messages and cleans up the queue.
   */
  #[Hook('cron')]
  public function cron(): void {
    // Process stale messages (stuck in processing status).
    // 5 minutes.
    $staleThreshold = 300;
    $staleMessages = $this->findStaleMessages($staleThreshold);

    $processed = 0;
    $failed = 0;

    foreach ($staleMessages as $message) {
      try {
        $executionId = $message->getExecutionId();

        // Check if execution is actually running.
        if ($executionId !== NULL && $this->isExecutionRunning($executionId)) {
          // Still running, skip.
          continue;
        }

        // Retry or mark as failed.
        $retryCount = $message->getRetryCount();
        if ($retryCount < 3) {
          $message->incrementRetryCount();
          $message->setStatus(MessageStatus::PENDING);
          $message->save();

          // Re-queue for processing.
          $queue = $this->queueFactory->get('flowdrop_session_message');
          $queue->createItem([
            'message_id' => $message->id(),
            'session_id' => $message->getSessionId(),
            'sequence_number' => $message->getSequenceNumber(),
          ]);

          $processed++;
        }
        else {
          // Max retries exceeded, mark as failed.
          $message->setStatus(MessageStatus::FAILED);
          $message->save();

          $session = $message->getSession();
          if ($session !== NULL) {
            $this->sessionService->addSystemMessage($session, "Message processing timeout after {$retryCount} retries", [
              'level' => 'error',
              'user_message_id' => $message->id(),
            ]);
          }

          $failed++;
        }
      }
      catch (\Exception $e) {
        $this->logger->error('Failed to process stale message @id: @message', [
          '@id' => $message->id(),
          '@message' => $e->getMessage(),
        ]);
      }
    }

    if ($processed > 0 || $failed > 0) {
      $this->logger->info('Cron: Processed @processed stale messages, marked @failed as failed', [
        '@processed' => $processed,
        '@failed' => $failed,
      ]);
    }

    // Recover stuck sessions (running or awaiting_input for too long).
    // 10 minutes.
    $this->recoverStuckSessions(600);

    // Clean up stuck queue items.
    $this->cleanupStuckQueueItems();
  }

  /**
   * Finds messages stuck in processing status.
   *
   * @param int $threshold
   *   Time threshold in seconds (default: 5 minutes).
   *
   * @return array<\Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface>
   *   Array of stale message entities.
   */
  private function findStaleMessages(int $threshold = 300): array {
    $storage = $this->entityTypeManager->getStorage('flowdrop_session_message');
    $cutoffTime = time() - $threshold;

    $query = $storage->getQuery()
      ->condition('status', MessageStatus::PROCESSING)
      ->condition('timestamp', $cutoffTime, '<')
      // Access bypassed: system cron cleanup runs without user context.
      ->accessCheck(FALSE)
      // Limit batch size.
      ->range(0, 100);

    $ids = $query->execute();
    if (empty($ids)) {
      return [];
    }

    $messages = $storage->loadMultiple($ids);
    $result = [];
    foreach ($messages as $message) {
      if ($message instanceof FlowDropSessionMessageInterface) {
        $result[] = $message;
      }
    }

    return $result;
  }

  /**
   * Checks if a workflow execution is still running.
   *
   * @param string $execution_id
   *   The execution ID.
   *
   * @return bool
   *   TRUE if execution is running, FALSE otherwise.
   */
  private function isExecutionRunning(string $execution_id): bool {
    $pipelineStorage = $this->entityTypeManager->getStorage('flowdrop_pipeline');
    $query = $pipelineStorage->getQuery()
      ->condition('execution_id', $execution_id)
      // Access bypassed: system cron cleanup runs without user context.
      ->accessCheck(FALSE)
      ->range(0, 1);

    $ids = $query->execute();
    if (!empty($ids)) {
      $pipeline = $pipelineStorage->load(reset($ids));
      if ($pipeline instanceof FlowDropPipelineInterface) {
        $status = $pipeline->getStatus();
        return in_array($status, [PipelineStatus::PENDING, PipelineStatus::RUNNING], TRUE);
      }
    }

    return FALSE;
  }

  /**
   * Recovers sessions stuck in running or awaiting_input status.
   *
   * Sessions may get stuck if execution fails without properly resetting
   * the session status (e.g., due to a PHP fatal error or timeout).
   *
   * @param int $threshold
   *   Time threshold in seconds for considering a session stuck.
   */
  private function recoverStuckSessions(int $threshold): void {
    $storage = $this->entityTypeManager->getStorage('flowdrop_session');
    $cutoffTime = time() - $threshold;

    $ids = $storage->getQuery()
      ->condition('status', [SessionStatus::RUNNING, SessionStatus::AWAITING_INPUT], 'IN')
      ->condition('changed', $cutoffTime, '<')
      // Access bypassed: system cron cleanup runs without user context.
      ->accessCheck(FALSE)
      ->range(0, 50)
      ->execute();

    if (empty($ids)) {
      return;
    }

    $sessions = $storage->loadMultiple($ids);
    $recovered = 0;

    foreach ($sessions as $session) {
      if (!$session instanceof FlowDropSessionInterface) {
        continue;
      }

      try {
        $previousStatus = $session->getStatus();

        // For running sessions, verify no pipeline bound to this session is
        // still active. Pipelines carry the pending-interrupt slot now, so
        // we look up pending pipelines for this session directly.
        $pipelineStorage = $this->entityTypeManager->getStorage('flowdrop_pipeline');
        $pendingPipelineIds = $pipelineStorage->getQuery()
          ->condition('session_id', $session->id())
          ->condition('pending_interrupt_id', NULL, 'IS NOT NULL')
          ->accessCheck(FALSE)
          ->execute();
        if ($previousStatus === SessionStatus::RUNNING) {
          $stillRunning = FALSE;
          foreach ($pendingPipelineIds as $pid) {
            if ($this->isExecutionRunning((string) $pid)) {
              $stillRunning = TRUE;
              break;
            }
          }
          if ($stillRunning) {
            continue;
          }
        }

        // Reset session to idle and clear any lingering pipeline pause
        // slots so the stuck session can accept fresh input.
        $session->setStatus(SessionStatus::IDLE);
        $session->save();
        foreach ($pipelineStorage->loadMultiple($pendingPipelineIds) as $pipeline) {
          if ($pipeline instanceof FlowDropPipelineInterface) {
            $pipeline->setPendingInterruptId(NULL);
            $pipeline->setPendingInterruptType(NULL);
            $pipeline->save();
          }
        }

        // Add a system message noting the recovery.
        $this->sessionService->addSystemMessage(
          $session,
          "Session recovered from stuck '{$previousStatus}' state by cron.",
          ['level' => 'warning', 'type' => 'session_recovery'],
        );

        $recovered++;
      }
      catch (\Exception $e) {
        $this->logger->error('Failed to recover stuck session @id: @message', [
          '@id' => $session->id(),
          '@message' => $e->getMessage(),
        ]);
      }
    }

    if ($recovered > 0) {
      $this->logger->info('Cron: Recovered @count stuck sessions', [
        '@count' => $recovered,
      ]);
    }
  }

  /**
   * Cleans up stuck queue items.
   */
  private function cleanupStuckQueueItems(): void {
    $queue = $this->queueFactory->get('flowdrop_session_message');

    $numberOfItems = $queue->numberOfItems();

    if ($numberOfItems > 0) {
      $this->logger->debug('Queue has @count items', ['@count' => $numberOfItems]);
    }
  }

}
