<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\Service;

use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_interrupt\Enum\InterruptStatus;
use Drupal\flowdrop_interrupt\Plugin\InterruptHandler\ScheduledInterruptHandlerInterface;
use Psr\Log\LoggerInterface;

/**
 * Resolves due scheduled interrupts on cron.
 *
 * First consumer of {@see ScheduledInterruptHandlerInterface}: sweeps
 * pending interrupts whose handler is schedule-driven, confirms each via
 * `shouldResolveNow()`, and resolves it through the locked
 * {@see InterruptManagerInterface::resolveInterrupt()} path with the
 * handler's `buildAutoResponse()` payload and a NULL uid (system
 * resolution). Resolution fires `flowdrop_interrupt.resolved`, which the
 * existing InterruptResolvedSubscriber turns into a pipeline resume —
 * nothing here touches jobs or orchestrators directly.
 *
 * The entity query pre-filters on `scheduled_at <= now` so far-future
 * timers are never loaded; `shouldResolveNow()` stays the authoritative
 * per-entity check for handlers with richer conditions.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
final class ScheduledInterruptSweeper {

  /**
   * Default maximum interrupts resolved per sweep.
   *
   * Deliberately much lower than InterruptManager::SWEEPER_DEFAULT_LIMIT:
   * each resolution executes the remainder of the paused workflow INLINE
   * inside cron (InterruptResolvedSubscriber resumes the pipeline
   * synchronously), so the per-row cost is a workflow run, not an entity
   * save. Overflow simply waits for the next cron run — timers stay
   * pending, nothing is lost. Re-enqueueing resumes instead of inline
   * execution is the follow-up that would let this scale up.
   */
  public const DEFAULT_LIMIT = 50;

  /**
   * The interrupt entity storage.
   */
  private EntityStorageInterface $storage;

  public function __construct(
    EntityTypeManagerInterface $entityTypeManager,
    private readonly InterruptManagerInterface $interruptManager,
    private readonly InterruptHandlerPluginManager $handlerManager,
    private readonly LoggerInterface $logger,
  ) {
    $this->storage = $entityTypeManager->getStorage('flowdrop_interrupt');
  }

  /**
   * Resolves all currently-due scheduled interrupts.
   *
   * @param int|null $limit
   *   Maximum rows per sweep, or NULL for unbounded.
   *
   * @return int
   *   Number of interrupts resolved.
   */
  public function sweep(?int $limit = self::DEFAULT_LIMIT): int {
    $types = $this->handlerManager->getScheduledTypeIds();
    if ($types === []) {
      return 0;
    }

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

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

    $count = 0;
    foreach (array_chunk($ids, 50) as $chunk) {
      foreach ($this->storage->loadMultiple($chunk) as $entity) {
        if (!$entity instanceof FlowDropInterruptInterface) {
          continue;
        }
        $handler = $this->handlerManager->loadHandler($entity->getType()->value);
        if (!$handler instanceof ScheduledInterruptHandlerInterface) {
          continue;
        }
        if (!$handler->shouldResolveNow($entity)) {
          continue;
        }
        try {
          // NULL uid: system resolution, mirroring the budget-pause
          // initiator-NULL convention. The handler's auto response (e.g.
          // resolved_by_timer) is the authoritative timer-vs-human signal.
          $this->interruptManager->resolveInterrupt(
            (string) $entity->uuid(),
            $handler->buildAutoResponse($entity),
            NULL,
          );
          $count++;
        }
        catch (\InvalidArgumentException $e) {
          // Concurrent resolution or state change between query and load —
          // skip and let the next sweep reconcile.
          $this->logger->debug('Scheduled interrupt @id skipped: @message', [
            '@id' => $entity->uuid(),
            '@message' => $e->getMessage(),
          ]);
        }
      }
      // Free per-chunk entity static cache so large backlogs don't hoard
      // memory.
      $this->storage->resetCache($chunk);
    }

    return $count;
  }

}
