<?php
/**
 * Advances the indexing job one batch at a time.
 *
 * @package Postlinkly
 * @since   1.14.0
 */

declare(strict_types=1);

namespace Postlinkly\Indexing;

use Postlinkly\Indexing\Stage\IndexStage;

/**
 * The only component that decides what the job does next.
 *
 * Stages know how to index; the runner knows about order, ownership, failure
 * budgets, and completion. Keeping that split means adding a stage never
 * involves reasoning about scheduling, and the guarantees this class provides
 * are testable without touching a database.
 *
 * @since 1.14.0
 */
final class JobRunner {

	/**
	 * Consecutive failures a stage may record before it is given up on.
	 *
	 * Without a ceiling, a stage whose dependency is down retries forever and
	 * every stage behind it starves.
	 */
	public const MAX_CONSECUTIVE_FAILURES = 3;

	/**
	 * @var array<int, IndexStage>
	 */
	private array $stages;

	/**
	 * @var JobStateRepository
	 */
	private JobStateRepository $repository;

	/**
	 * @var JobLease
	 */
	private JobLease $lease;

	/**
	 * Returns the current unix time.
	 *
	 * @var callable(): int
	 */
	private $clock;

	/**
	 * @since 1.14.0
	 *
	 * @param array<int, IndexStage> $stages     Stages in execution order.
	 * @param JobStateRepository     $repository Job storage.
	 * @param JobLease               $lease      Ownership lease.
	 * @param callable|null          $clock      Optional. Returns the current unix time.
	 */
	public function __construct( array $stages, JobStateRepository $repository, JobLease $lease, ?callable $clock = null ) {
		$this->stages     = array_values( $stages );
		$this->repository = $repository;
		$this->lease      = $lease;
		$this->clock      = $clock ?? static fn(): int => time();
	}

	/**
	 * Current state, or a fresh idle one when nothing is stored.
	 *
	 * @since 1.14.0
	 */
	public function state(): JobState {
		return $this->repository->load() ?? JobState::idle( $this->stage_ids() );
	}

	/**
	 * Resets the job and enrols every willing stage.
	 *
	 * @since 1.14.0
	 */
	public function start(): JobState {
		$state = JobState::idle( $this->stage_ids() )
			->with_status( 'running' )
			->with_started_at( ( $this->clock )() );

		foreach ( $this->stages as $stage ) {
			if ( $stage->should_enrol() ) {
				continue;
			}

			$record           = $this->record_for( $state, $stage );
			$record['status'] = 'skipped';
			$state            = $state->with_stage( $stage->id(), $record );
		}

		$this->repository->save( $state );

		return $state;
	}

	/**
	 * Brings the stored job in line with the stages that exist now: adds
	 * stages it has never seen, drops stages that no longer exist, and
	 * reopens a finished job when a new stage has work to do.
	 *
	 * A plugin update or an add-on can introduce a stage after the job last
	 * ran. The stored job has no record of it, so on a site whose index had
	 * already finished nothing would ever run it short of a full rebuild —
	 * which redoes every stage that already finished. Here only the new stage
	 * is queued; finished stages keep their results. A new stage that
	 * declines to enrol is recorded as skipped and never reopens the job.
	 *
	 * The reverse happens too: a stage moves to an add-on, or an add-on is
	 * removed. Its record can never change again, and if it had failed it
	 * would keep the whole job reported as failed forever. Such records are
	 * dropped, and a job that had failed only because of them becomes done.
	 *
	 * @since 1.18.1
	 *
	 * @return bool True when the job was reopened and needs scheduling.
	 */
	public function reconcile_stages(): bool {
		$state = $this->repository->load();

		// Never run: starting the job enrols every stage anyway.
		if ( null === $state ) {
			return false;
		}

		$registered = $this->stage_ids();
		$changed    = false;
		$has_work   = false;

		foreach ( array_keys( $state->stages() ) as $id ) {
			if ( ! in_array( (string) $id, $registered, true ) ) {
				$state   = $state->without_stage( (string) $id );
				$changed = true;
			}
		}

		foreach ( $this->stages as $stage ) {
			if ( isset( $state->stages()[ $stage->id() ] ) ) {
				continue;
			}

			$record = $this->record_for( $state, $stage );

			if ( $stage->should_enrol() ) {
				$has_work = true;
			} else {
				$record['status'] = 'skipped';
			}

			$state   = $state->with_stage( $stage->id(), $record );
			$changed = true;
		}

		if ( ! $changed ) {
			return false;
		}

		// A running job picks a new stage up by itself once it reaches it;
		// only a job that has stopped needs reopening.
		$reopen = $has_work && in_array( $state->status(), array( 'done', 'failed' ), true );

		if ( $reopen ) {
			$state = $state->with_status( 'running' );
		} elseif ( 'failed' === $state->status() && ! $this->has_failed_stage( $state ) ) {
			$state = $state->with_status( 'done' );
		}

		$this->repository->save( $state );

		return $reopen;
	}

	/**
	 * The stored record for a stage, or a fresh pending one when the job has
	 * none — a stage added after the job was stored.
	 *
	 * @since 1.18.1
	 *
	 * @param JobState   $state Current state.
	 * @param IndexStage $stage Stage to look up.
	 *
	 * @return array{status: string, cursor: int, processed: int, failures: int, error: string}
	 */
	private function record_for( JobState $state, IndexStage $stage ): array {
		return $state->stages()[ $stage->id() ]
			?? JobState::idle( array( $stage->id() ) )->stages()[ $stage->id() ];
	}

	/**
	 * Puts failed stages back in the queue so the job can run them again.
	 *
	 * A failed stage is terminal, so nothing picks it up again: pressing
	 * Resume finds no eligible stage, runs nothing, and leaves the original
	 * failure on screen looking like a button that does not work. Only a full
	 * rebuild could clear it, which also discards the work of every stage
	 * that succeeded.
	 *
	 * Deliberately not automatic. The consecutive-failure budget exists to
	 * stop a broken stage retrying forever on its own; this is someone
	 * explicitly asking it to try again, which is a different thing.
	 *
	 * @since 1.16.4
	 *
	 * @return bool True when at least one stage was requeued.
	 */
	public function retry_failed(): bool {
		$state     = $this->state();
		$requeued  = false;

		foreach ( $state->stages() as $id => $record ) {
			if ( 'failed' !== ( $record['status'] ?? '' ) ) {
				continue;
			}

			$record['status']   = 'pending';
			$record['failures'] = 0;
			$record['error']    = '';

			$state    = $state->with_stage( (string) $id, $record );
			$requeued = true;
		}

		if ( ! $requeued ) {
			return false;
		}

		$this->repository->save( $state->with_status( 'running' ) );

		return true;
	}

	/**
	 * Runs the current stage until the deadline, then commits.
	 *
	 * @since 1.14.0
	 * @since 1.18.1 A stage the stored job has no record of runs from the start.
	 *
	 * @param Deadline $deadline Time budget for this batch.
	 * @param string   $owner    This runner's token.
	 *
	 * @return JobState|null Null when another live lease holds the job.
	 */
	public function run_batch( Deadline $deadline, string $owner ): ?JobState {
		$state = $this->lease->claim( $this->state(), $owner, $deadline->budget_seconds() );
		if ( null === $state ) {
			return null;
		}

		$stage = $this->next_stage( $state );
		if ( null === $stage ) {
			return $this->finish( $state );
		}

		$record = $this->record_for( $state, $stage );
		$result = $stage->process( $deadline, (int) $record['cursor'] );

		if ( $result->is_failed() ) {
			++$record['failures'];
			$record['error'] = $result->error();

			if ( $record['failures'] >= self::MAX_CONSECUTIVE_FAILURES ) {
				$record['status'] = 'failed';
			}
		} else {
			// A successful batch clears the count, so intermittent trouble
			// never accumulates into a permanent failure.
			$record['failures']   = 0;
			$record['error']      = '';
			$record['processed'] += $result->processed();
			$record['status']     = $result->is_finished() ? 'done' : 'running';

			if ( ! $result->is_finished() ) {
				$record['cursor'] = $result->cursor();
			}
		}

		$state = $state->with_stage( $stage->id(), $record )->with_heartbeat( ( $this->clock )() );

		if ( null === $this->next_stage( $state ) ) {
			return $this->finish( $state );
		}

		// Let go before saving. The lease exists so a batch that dies without
		// reaching any cleanup cannot wedge the job -- it guards the run, not
		// the gap between runs. Every batch claims under a fresh owner, so a
		// lease left behind here is held by a runner that no longer exists,
		// and the next batch (cron or button) declines against a ghost until
		// the floor expires. That throttles the whole job to one batch per
		// lease floor however often anything asks it to advance.
		$state = $this->lease->release( $state );

		$this->repository->save( $state );

		return $state;
	}

	/**
	 * Marks the job done and lets go of the lease.
	 *
	 * @since 1.14.0
	 *
	 * @param JobState $state State to finalise.
	 */
	private function finish( JobState $state ): JobState {
		// A failed stage is terminal, so the job stops having work to do and
		// arrives here exactly as a successful one does. Calling that "done"
		// tells the owner the index is up to date while part of it was never
		// written -- the failure is still listed on the stage, but the job
		// headline contradicts it, and the headline is what gets believed.
		$status = $this->has_failed_stage( $state ) ? 'failed' : 'done';

		$state = $this->lease->release( $state )
			->with_status( $status )
			->with_heartbeat( ( $this->clock )() );

		$this->repository->save( $state );

		return $state;
	}

	/**
	 * Whether any stage ended in failure.
	 *
	 * @since 1.16.3
	 * @since 1.18.1 Ignores records of stages that are no longer registered.
	 *
	 * @param JobState $state State to inspect.
	 *
	 * @return bool
	 */
	private function has_failed_stage( JobState $state ): bool {
		$registered = $this->stage_ids();

		foreach ( $state->stages() as $id => $record ) {
			// A stage that no longer exists can never be retried, so its old
			// failure says nothing about the index as it is now.
			if ( ! in_array( (string) $id, $registered, true ) ) {
				continue;
			}

			if ( 'failed' === ( $record['status'] ?? '' ) ) {
				return true;
			}
		}

		return false;
	}

	/**
	 * First stage still eligible to run, or null when all are terminal.
	 *
	 * @since 1.14.0
	 *
	 * @param JobState $state Current state.
	 */
	private function next_stage( JobState $state ): ?IndexStage {
		foreach ( $this->stages as $stage ) {
			$status = $state->stages()[ $stage->id() ]['status'] ?? 'pending';
			if ( in_array( $status, array( 'pending', 'running' ), true ) ) {
				return $stage;
			}
		}

		return null;
	}

	/**
	 * @return array<int, string>
	 */
	private function stage_ids(): array {
		return array_map( static fn( IndexStage $s ): string => $s->id(), $this->stages );
	}
}