<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\Service;

use Drupal\Core\Entity\EntityStorageException;
use Drupal\Core\Entity\EntityRepositoryInterface;
use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Lock\LockBackendInterface;
use Drupal\Core\Session\AccountProxyInterface;
use Drupal\flowdrop_interrupt\DTO\InterruptCreationData;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_interrupt\Enum\InterruptDirection;
use Drupal\flowdrop_interrupt\Enum\InterruptStatus;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
use Psr\Log\LoggerInterface;
use Drupal\flowdrop_interrupt\Event\InterruptCancelledEvent;
use Drupal\flowdrop_interrupt\Event\InterruptExpiredEvent;
use Drupal\flowdrop_interrupt\Event\InterruptResolvedEvent;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Service for managing interrupt requests.
 *
 * Handles the full lifecycle of interrupt entities including creation,
 * resolution, querying, and cleanup.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class InterruptManager implements InterruptManagerInterface {

  /**
   * The interrupt entity storage.
   *
   * @var \Drupal\Core\Entity\EntityStorageInterface
   */
  protected EntityStorageInterface $storage;

  /**
   * Constructs a new InterruptManager.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\Core\Session\AccountProxyInterface $currentUser
   *   The current user.
   * @param \Psr\Log\LoggerInterface $logger
   *   The logger service.
   * @param \Symfony\Component\EventDispatcher\EventDispatcherInterface $eventDispatcher
   *   The event dispatcher.
   * @param \Drupal\Core\Lock\LockBackendInterface $lock
   *   The lock backend.
   * @param \Drupal\flowdrop_interrupt\Service\InterruptFactoryInterface $factory
   *   The interrupt factory used to persist new interrupts.
   * @param \Drupal\Core\Entity\EntityRepositoryInterface|null $entityRepository
   *   The entity repository — used for statically-cached UUID lookups.
   *   Nullable for BC with code that wires this service manually in tests;
   *   the container-supplied instance is never NULL in production.
   */
  public function __construct(
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly AccountProxyInterface $currentUser,
    protected readonly LoggerInterface $logger,
    protected readonly EventDispatcherInterface $eventDispatcher,
    protected readonly LockBackendInterface $lock,
    protected readonly InterruptFactoryInterface $factory,
    protected readonly ?EntityRepositoryInterface $entityRepository = NULL,
  ) {
    $this->storage = $this->entityTypeManager->getStorage("flowdrop_interrupt");
  }

  /**
   * {@inheritdoc}
   */
  public function create(InterruptCreationData $data): FlowDropInterruptInterface {
    return $this->factory->create($data);
  }

  /**
   * {@inheritdoc}
   */
  public function resolveInterrupt(
    string $interrupt_id,
    mixed $response_data,
    ?int $user_id = NULL,
  ): ?FlowDropInterruptInterface {
    // Acquire a lock to prevent concurrent resolution of the same interrupt.
    $lockName = "flowdrop_interrupt_resolve_{$interrupt_id}";
    if (!$this->lock->acquire($lockName, 30)) {
      throw new \InvalidArgumentException("Interrupt {$interrupt_id} resolution is already in progress");
    }

    try {
      $interrupt = $this->getInterrupt($interrupt_id);

      if ($interrupt === NULL) {
        $this->logger->warning("Cannot resolve interrupt @id: not found", [
          "@id" => $interrupt_id,
        ]);
        return NULL;
      }

      if (!$interrupt->isPending()) {
        $this->logger->warning("Cannot resolve interrupt @id: status is @status", [
          "@id" => $interrupt_id,
          "@status" => $interrupt->getStatus()->value,
        ]);
        throw new \InvalidArgumentException(
          "Interrupt {$interrupt_id} is not in a resolvable state. Current status: " . $interrupt->getStatus()->value
        );
      }

      // Check expiration.
      if ($interrupt->isExpired()) {
        $interrupt->expire();
        $interrupt->save();
        $this->logger->info("Interrupt @id has expired", ["@id" => $interrupt_id]);
        throw new \InvalidArgumentException("Interrupt {$interrupt_id} has expired");
      }

      // Use provided user ID or current user.
      $resolved_by = $user_id ?? ($this->currentUser->isAuthenticated() ? (int) $this->currentUser->id() : NULL);

      $interrupt->resolve($response_data, $resolved_by);
      $interrupt->save();

      $this->logger->info("Resolved interrupt @id by user @user", [
        "@id" => $interrupt_id,
        "@user" => $resolved_by ?? "anonymous",
      ]);

      // Dispatch event for integrations.
      $this->eventDispatcher->dispatch(
        new InterruptResolvedEvent(
          interrupt: $interrupt,
          responseData: $response_data,
          resolvedBy: $resolved_by,
        ),
        InterruptResolvedEvent::NAME
      );

      return $interrupt;
    }
    finally {
      $this->lock->release($lockName);
    }
  }

  /**
   * {@inheritdoc}
   *
   * Uses {@see EntityRepositoryInterface::loadEntityByUuid()} so repeat lookups
   * of the same UUID within a request hit the entity static cache. The hot
   * paths here are resume (orchestrator → manager), HTTP resolve, and the
   * cascade subscriber — all of which can look up the same UUID multiple
   * times in one request.
   */
  public function getInterrupt(string $interrupt_id): ?FlowDropInterruptInterface {
    if ($this->entityRepository !== NULL) {
      try {
        $interrupt = $this->entityRepository->loadEntityByUuid('flowdrop_interrupt', $interrupt_id);
      }
      catch (EntityStorageException) {
        return NULL;
      }
      return $interrupt instanceof FlowDropInterruptInterface ? $interrupt : NULL;
    }
    // Fallback for tests that wire the manager without the repository.
    $entities = $this->storage->loadByProperties(["uuid" => $interrupt_id]);
    $interrupt = reset($entities);
    return $interrupt instanceof FlowDropInterruptInterface ? $interrupt : NULL;
  }

  /**
   * {@inheritdoc}
   */
  public function getPendingInterruptsForSession(int $session_id): array {
    $ids = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("session_id", $session_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->sort("created", "ASC")
      ->execute();

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

    $entities = $this->storage->loadMultiple($ids);
    return array_values(array_filter(
      $entities,
      fn($entity) => $entity instanceof FlowDropInterruptInterface
    ));
  }

  /**
   * {@inheritdoc}
   */
  public function findPendingInterruptOnPipeline(
    InterruptType $type,
    string $pipeline_id,
  ): ?FlowDropInterruptInterface {
    if ($pipeline_id === '') {
      return NULL;
    }
    $ids = $this->storage->getQuery()
      ->accessCheck(FALSE)
      ->condition('type', $type->value)
      ->condition('pipeline_id', $pipeline_id)
      ->condition('status', InterruptStatus::Pending->value)
      ->sort('created', 'ASC')
      ->range(0, 1)
      ->execute();
    if (empty($ids)) {
      return NULL;
    }
    $entity = $this->storage->load(reset($ids));
    return $entity instanceof FlowDropInterruptInterface ? $entity : NULL;
  }

  /**
   * {@inheritdoc}
   */
  public function getPendingInterruptsLinkedTo(string $linked_interrupt_id): array {
    if ($linked_interrupt_id === '') {
      return [];
    }
    $ids = $this->storage->getQuery()
      ->accessCheck(FALSE)
      ->condition('linked_interrupt_id', $linked_interrupt_id)
      ->condition('status', InterruptStatus::Pending->value)
      ->sort('created', 'ASC')
      ->execute();
    if (empty($ids)) {
      return [];
    }
    $entities = $this->storage->loadMultiple($ids);
    return array_values(array_filter(
      $entities,
      fn ($entity) => $entity instanceof FlowDropInterruptInterface,
    ));
  }

  /**
   * {@inheritdoc}
   */
  public function findPendingInwardSignal(string $pipelineId): ?FlowDropInterruptInterface {
    $ids = $this->storage->getQuery()
      ->accessCheck(FALSE)
      ->condition('target_pipeline_id', $pipelineId)
      ->condition('direction', InterruptDirection::Inward->value)
      ->condition('status', InterruptStatus::Pending->value)
      ->sort('created', 'ASC')
      ->range(0, 1)
      ->execute();
    if (empty($ids)) {
      return NULL;
    }
    $entity = $this->storage->load(reset($ids));
    return $entity instanceof FlowDropInterruptInterface ? $entity : NULL;
  }

  /**
   * {@inheritdoc}
   */
  public function cancelOutwardInterruptsForPipeline(string $pipelineId): int {
    $ids = $this->storage->getQuery()
      ->accessCheck(FALSE)
      ->condition('pipeline_id', $pipelineId)
      ->condition('direction', InterruptDirection::Outward->value)
      ->condition('status', InterruptStatus::Pending->value)
      ->execute();
    if (empty($ids)) {
      return 0;
    }
    $count = 0;
    foreach ($this->storage->loadMultiple($ids) as $entity) {
      if (!$entity instanceof FlowDropInterruptInterface) {
        continue;
      }
      $entity->cancel();
      $entity->save();
      $count++;
    }
    return $count;
  }

  /**
   * {@inheritdoc}
   */
  public function getPendingInterruptsForPipeline(string $pipeline_id): array {
    $ids = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("pipeline_id", $pipeline_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->sort("created", "ASC")
      ->execute();

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

    $entities = $this->storage->loadMultiple($ids);
    return array_values(array_filter(
      $entities,
      fn($entity) => $entity instanceof FlowDropInterruptInterface
    ));
  }

  /**
   * {@inheritdoc}
   */
  public function getPendingInterruptsForWorkflow(string $workflow_id): array {
    $ids = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("workflow_id", $workflow_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->sort("created", "ASC")
      ->execute();

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

    $entities = $this->storage->loadMultiple($ids);
    return array_values(array_filter(
      $entities,
      fn($entity) => $entity instanceof FlowDropInterruptInterface
    ));
  }

  /**
   * {@inheritdoc}
   */
  public function getPendingInterruptsForUser(int $user_id): array {
    $ids = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("uid", $user_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->sort("created", "ASC")
      ->execute();

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

    $entities = $this->storage->loadMultiple($ids);
    return array_values(array_filter(
      $entities,
      fn($entity) => $entity instanceof FlowDropInterruptInterface
    ));
  }

  /**
   * {@inheritdoc}
   */
  public function getPendingInterruptsByType(InterruptType $type): array {
    $ids = $this->storage->getQuery()
      ->accessCheck(FALSE)
      ->condition('type', $type->value)
      ->condition('status', InterruptStatus::Pending->value)
      ->sort('created', 'ASC')
      ->execute();
    if (empty($ids)) {
      return [];
    }
    $entities = $this->storage->loadMultiple($ids);
    return array_values(array_filter(
      $entities,
      fn ($entity) => $entity instanceof FlowDropInterruptInterface,
    ));
  }

  /**
   * {@inheritdoc}
   */
  public function countPendingInterruptsForUser(int $user_id): int {
    return (int) $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("uid", $user_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->count()
      ->execute();
  }

  /**
   * {@inheritdoc}
   */
  public function hasPendingInterruptsForSession(int $session_id): bool {
    $count = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("session_id", $session_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->count()
      ->execute();

    return $count > 0;
  }

  /**
   * {@inheritdoc}
   */
  public function hasPendingInterruptsForPipeline(string $pipeline_id): bool {
    $count = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("pipeline_id", $pipeline_id)
      ->condition("status", InterruptStatus::Pending->value)
      ->count()
      ->execute();

    return $count > 0;
  }

  /**
   * {@inheritdoc}
   */
  public function cancelInterrupt(string $interrupt_id): bool {
    $interrupt = $this->getInterrupt($interrupt_id);

    if ($interrupt === NULL) {
      $this->logger->warning("Cannot cancel interrupt @id: not found", [
        "@id" => $interrupt_id,
      ]);
      return FALSE;
    }

    if (!$interrupt->isPending()) {
      $this->logger->warning("Cannot cancel interrupt @id: status is @status", [
        "@id" => $interrupt_id,
        "@status" => $interrupt->getStatus()->value,
      ]);
      return FALSE;
    }

    $interrupt->cancel();
    $interrupt->save();

    $this->logger->info("Cancelled interrupt @id", ["@id" => $interrupt_id]);

    // Dispatch event for integrations.
    $this->eventDispatcher->dispatch(
      new InterruptCancelledEvent(
        interrupt: $interrupt,
      ),
      InterruptCancelledEvent::NAME
    );

    return TRUE;
  }

  /**
   * {@inheritdoc}
   */
  public function cancelInterruptsForSession(int $session_id): int {
    $interrupts = $this->getPendingInterruptsForSession($session_id);
    $count = 0;

    foreach ($interrupts as $interrupt) {
      $interrupt->cancel();
      $interrupt->save();
      $count++;

      $this->eventDispatcher->dispatch(
        new InterruptCancelledEvent(interrupt: $interrupt),
        InterruptCancelledEvent::NAME
      );
    }

    if ($count > 0) {
      $this->logger->info("Cancelled @count interrupts for session @session", [
        "@count" => $count,
        "@session" => $session_id,
      ]);
    }

    return $count;
  }

  /**
   * {@inheritdoc}
   */
  public function cancelInterruptsForPipeline(string $pipeline_id): int {
    $interrupts = $this->getPendingInterruptsForPipeline($pipeline_id);
    $count = 0;

    foreach ($interrupts as $interrupt) {
      $interrupt->cancel();
      $interrupt->save();
      $count++;

      $this->eventDispatcher->dispatch(
        new InterruptCancelledEvent(interrupt: $interrupt),
        InterruptCancelledEvent::NAME
      );
    }

    if ($count > 0) {
      $this->logger->info("Cancelled @count interrupts for pipeline @pipeline", [
        "@count" => $count,
        "@pipeline" => $pipeline_id,
      ]);
    }

    return $count;
  }

  /**
   * Default per-run cap for cron-driven sweepers.
   *
   * Cron runs frequently enough that draining a backlog over multiple runs
   * is preferable to OOMing on a single one. Callers can override with the
   * explicit $limit parameter.
   */
  public const SWEEPER_DEFAULT_LIMIT = 500;

  /**
   * Chunk size for entity-loading within a single sweeper run.
   *
   * Bounded so the entity static cache (and any related render caches that
   * may build up around save events) doesn't grow without limit inside one
   * cron tick.
   */
  private const SWEEPER_CHUNK_SIZE = 50;

  /**
   * {@inheritdoc}
   */
  public function expireOverdueInterrupts(?int $limit = self::SWEEPER_DEFAULT_LIMIT): int {
    $now = time();

    $query = $this->storage->getQuery()
      // Access bypassed: system cron operation runs without user context.
      ->accessCheck(FALSE)
      ->condition("status", InterruptStatus::Pending->value)
      ->condition("expires", $now, "<")
      ->condition("expires", 0, ">")
      ->sort('expires', 'ASC');
    if ($limit !== NULL && $limit > 0) {
      $query->range(0, $limit);
    }
    $ids = $query->execute();

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

    $count = 0;
    foreach (array_chunk($ids, self::SWEEPER_CHUNK_SIZE) as $chunk) {
      foreach ($this->storage->loadMultiple($chunk) as $entity) {
        if (!$entity instanceof FlowDropInterruptInterface) {
          continue;
        }
        $entity->expire();
        $entity->save();
        $count++;
        $this->eventDispatcher->dispatch(
          new InterruptExpiredEvent(interrupt: $entity),
          InterruptExpiredEvent::NAME,
        );
      }
      // Free per-chunk entity static cache so we don't hoard memory while
      // walking large backlogs. resetCache() on the storage clears just
      // this entity type's static cache.
      $this->storage->resetCache($chunk);
    }

    if ($count > 0) {
      $this->logger->info("Expired @count overdue interrupts", ["@count" => $count]);
    }

    return $count;
  }

  /**
   * {@inheritdoc}
   */
  public function cleanupOldInterrupts(int $max_age, ?int $limit = self::SWEEPER_DEFAULT_LIMIT): int {
    $threshold = time() - $max_age;

    $query = $this->storage->getQuery()
      // Access bypassed: system cron operation runs without user context.
      ->accessCheck(FALSE)
      ->condition("status", InterruptStatus::Pending->value, "<>")
      ->condition("changed", $threshold, "<")
      ->sort('changed', 'ASC');
    if ($limit !== NULL && $limit > 0) {
      $query->range(0, $limit);
    }
    $ids = $query->execute();

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

    $count = 0;
    foreach (array_chunk($ids, self::SWEEPER_CHUNK_SIZE) as $chunk) {
      $entities = $this->storage->loadMultiple($chunk);
      $this->storage->delete($entities);
      $count += count($entities);
      $this->storage->resetCache($chunk);
    }

    if ($count > 0) {
      $this->logger->info("Cleaned up @count old interrupts", ["@count" => $count]);
    }

    return $count;
  }

  /**
   * {@inheritdoc}
   */
  public function getResolvedInterruptForNode(string $pipeline_id, string $node_id): ?FlowDropInterruptInterface {
    $ids = $this->storage->getQuery()
      // Access bypassed: internal service query; access enforced at API layer.
      ->accessCheck(FALSE)
      ->condition("pipeline_id", $pipeline_id)
      ->condition("node_id", $node_id)
      ->condition("status", InterruptStatus::Resolved->value)
      ->sort("resolved_at", "DESC")
      ->range(0, 1)
      ->execute();

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

    $id = reset($ids);
    $entity = $this->storage->load($id);

    return $entity instanceof FlowDropInterruptInterface ? $entity : NULL;
  }

}
