<?php
/**
 * Background worker for embedding sync.
 *
 * @package WordPress\AI\Embeddings\Sync
 */

declare( strict_types=1 );

namespace WordPress\AI\Embeddings\Sync;

use RuntimeException;
use Throwable;
use WordPress\AI\Background\Lock;
use WordPress\AI\Background\Scheduler;

defined( 'ABSPATH' ) || exit;

/**
 * One run: take the lock, drain the live queue, advance running backfills, reschedule.
 *
 * @since 1.5.0
 */
class Sync_Worker {

	/**
	 * WP-Cron hook that runs the worker.
	 *
	 * @since 1.5.0
	 */
	public const CRON_HOOK = 'wpai_embedding_sync_run';

	/**
	 * Option recording the last run's Unix time.
	 *
	 * @since 1.5.0
	 */
	public const LAST_RUN_OPTION = 'wpai_embedding_sync_last_run';

	/**
	 * Name of the lock that keeps runs from overlapping.
	 *
	 * @since 1.5.0
	 */
	public const LOCK_NAME = 'embedding_sync';

	/**
	 * Default seconds per cron run.
	 *
	 * @since 1.5.0
	 */
	public const DEFAULT_TIME_BUDGET = 20;

	/**
	 * Default objects per batch.
	 *
	 * @since 1.5.0
	 */
	public const DEFAULT_BATCH_SIZE = 50;

	/**
	 * Default attempts before a queue row is marked failed.
	 *
	 * @since 1.5.0
	 */
	public const DEFAULT_MAX_ATTEMPTS = 5;

	/**
	 * Default stored objects checked per orphan sweep batch.
	 *
	 * @since 1.5.0
	 */
	public const SWEEP_BATCH_SIZE = 200;

	/**
	 * Seconds to wait after a run that ended on an unexpected error, so a persistent error is
	 * not retried on every cron tick.
	 */
	private const FAILED_RUN_DELAY = 300;

	/**
	 * Longest retry delay, in seconds (6 hours).
	 */
	private const MAX_RETRY_DELAY = 21600;

	/**
	 * Consumer registry.
	 *
	 * @var \WordPress\AI\Embeddings\Sync\Consumer_Registry
	 */
	private Consumer_Registry $registry;

	/**
	 * Sources keyed by object type.
	 *
	 * @var array<string, \WordPress\AI\Embeddings\Sync\Embedding_Source_Interface>
	 */
	private array $sources;

	/**
	 * Queue.
	 *
	 * @var \WordPress\AI\Embeddings\Sync\Sync_Queue
	 */
	private Sync_Queue $queue;

	/**
	 * Object processor.
	 *
	 * @var \WordPress\AI\Embeddings\Sync\Object_Processor
	 */
	private Object_Processor $processor;

	/**
	 * Backfill state.
	 *
	 * @var \WordPress\AI\Embeddings\Sync\Backfill_Manager
	 */
	private Backfill_Manager $backfills;

	/**
	 * Provider backoff.
	 *
	 * @var \WordPress\AI\Embeddings\Sync\Provider_Backoff
	 */
	private Provider_Backoff $backoff;

	/**
	 * Lock.
	 *
	 * @var \WordPress\AI\Background\Lock
	 */
	private Lock $lock;

	/**
	 * Orphan sweeper.
	 *
	 * @var \WordPress\AI\Embeddings\Sync\Orphan_Sweeper
	 */
	private Orphan_Sweeper $sweeper;

	/**
	 * Whether this run lost the lock to another runner.
	 *
	 * @var bool
	 */
	private bool $lock_lost = false;

	/**
	 * Constructor.
	 *
	 * @since 1.5.0
	 *
	 * @param \WordPress\AI\Embeddings\Sync\Consumer_Registry                         $registry  Consumer registry.
	 * @param array<string, \WordPress\AI\Embeddings\Sync\Embedding_Source_Interface> $sources   Sources keyed by object type.
	 * @param \WordPress\AI\Embeddings\Sync\Sync_Queue                                $queue     Queue.
	 * @param \WordPress\AI\Embeddings\Sync\Object_Processor                          $processor Object processor.
	 * @param \WordPress\AI\Embeddings\Sync\Backfill_Manager                          $backfills Backfill state.
	 * @param \WordPress\AI\Embeddings\Sync\Provider_Backoff                          $backoff   Provider backoff.
	 * @param \WordPress\AI\Background\Lock                                           $lock      Lock.
	 * @param \WordPress\AI\Embeddings\Sync\Orphan_Sweeper                            $sweeper   Orphan sweeper.
	 */
	public function __construct(
		Consumer_Registry $registry,
		array $sources,
		Sync_Queue $queue,
		Object_Processor $processor,
		Backfill_Manager $backfills,
		Provider_Backoff $backoff,
		Lock $lock,
		Orphan_Sweeper $sweeper
	) {
		$this->registry  = $registry;
		$this->sources   = $sources;
		$this->queue     = $queue;
		$this->processor = $processor;
		$this->backfills = $backfills;
		$this->backoff   = $backoff;
		$this->lock      = $lock;
		$this->sweeper   = $sweeper;
	}

	/**
	 * Schedules a worker run now unless one is already due.
	 *
	 * @since 1.5.0
	 */
	public static function wake(): void {
		self::schedule_at( time() );
	}

	/**
	 * Returns the delay before retrying after a transient failure.
	 *
	 * @since 1.5.0
	 *
	 * @param int $attempts Attempts made, including the one that just failed.
	 * @return int Seconds: 2^attempts minutes, capped at 6 hours.
	 */
	public static function retry_delay( int $attempts ): int {
		return (int) min( self::MAX_RETRY_DELAY, ( 2 ** min( $attempts, 20 ) ) * MINUTE_IN_SECONDS );
	}

	/**
	 * Runs the worker once.
	 *
	 * @since 1.5.0
	 *
	 * @param int|null $time_budget Optional. Seconds to work for; 0 means no limit. Default the filtered cron budget.
	 * @return array{ran: bool, queue: int, backfill: int} Whether it ran, and objects processed from each source of work.
	 */
	public function run( ?int $time_budget = null ): array {
		$stats = array(
			'ran'      => false,
			'queue'    => 0,
			'backfill' => 0,
		);

		if ( ! $this->registry->has_consumers() ) {
			wp_clear_scheduled_hook( self::CRON_HOOK );

			return $stats;
		}

		if ( ! $this->lock->acquire() ) {
			if ( null !== $this->get_next_run_at() ) {
				self::schedule_at( time() + $this->lock->get_ttl() );
			}

			return $stats;
		}

		$stats['ran']    = true;
		$this->lock_lost = false;

		self::schedule_at( time() + $this->lock->get_ttl() + MINUTE_IN_SECONDS );

		$failed = false;

		try {
			update_option( self::LAST_RUN_OPTION, time(), false );

			if ( null === $time_budget ) {
				/**
				 * Filters how long one WP-Cron run of the embedding sync worker may work.
				 *
				 * @since 1.5.0
				 *
				 * @param int $seconds Seconds. Default 20.
				 */
				$time_budget = (int) apply_filters( 'wpai_embedding_sync_time_budget', self::DEFAULT_TIME_BUDGET );
			}

			$deadline = $time_budget > 0 ? microtime( true ) + $time_budget : null;

			$stats['queue']    = $this->drain_queue( $deadline );
			$stats['backfill'] = $this->run_backfills( $deadline );
		} catch ( Throwable $e ) {
			$failed = true;

			wp_trigger_error(
				__METHOD__,
				esc_html(
					sprintf(
						/* translators: %s: Error message. */
						__( 'The embedding sync run stopped early: %s', 'ai' ),
						$e->getMessage()
					)
				),
				E_USER_WARNING
			);
		} finally {
			$this->lock->release();

			wp_clear_scheduled_hook( self::CRON_HOOK );
			$this->schedule_next( $failed ? time() + self::FAILED_RUN_DELAY : null );
		}

		return $stats;
	}

	/**
	 * Returns when there is next work to do.
	 *
	 * @since 1.5.0
	 *
	 * @param int|null $now Optional. Unix time. Default now.
	 * @return int|null Unix time of the next due work, or null when there is none.
	 */
	public function get_next_run_at( ?int $now = null ): ?int {
		$now  = $now ?? time();
		$next = null;

		$queued = $this->queue->get_next_available_at();

		if ( null !== $queued ) {
			$next = max( $now, $queued );
		}

		foreach ( $this->registry->get_targets() as $key => $target ) {
			if ( ! $this->backfills->is_running( $key ) ) {
				continue;
			}

			$at = $this->backoff->get_until_for( $target, $now ) ?? $now;

			// A pause holds only the embedding pass; the sweep after it makes no API requests.
			if ( $at > $now && $this->backfills->is_embedding_pass_done( $key, $target ) ) {
				$at = $now;
			}

			$next = null === $next ? $at : min( $next, $at );
		}

		return $next;
	}

	/**
	 * Schedules the next run if there is work and no run is scheduled at or before it.
	 *
	 * @since 1.5.0
	 *
	 * @param int|null $not_before Optional. Unix time before which no run is scheduled, such as the
	 *                             backoff after a failed run. Default null, as soon as work is due.
	 */
	public function schedule_next( ?int $not_before = null ): void {
		if ( ! $this->registry->has_consumers() ) {
			return;
		}

		$next = $this->get_next_run_at();

		if ( null === $next ) {
			return;
		}

		self::schedule_at( null === $not_before ? $next : max( $next, $not_before ) );
	}

	/**
	 * Schedules a run at a time unless one is already scheduled at or before it.
	 *
	 * @since 1.5.0
	 *
	 * @param int $timestamp Unix time the run should happen.
	 */
	private static function schedule_at( int $timestamp ): void {
		Scheduler::schedule_at( self::CRON_HOOK, $timestamp );
	}

	/**
	 * Processes due queue rows until none are left or time runs out.
	 *
	 * @since 1.5.0
	 *
	 * @param float|null $deadline Microtime deadline, or null for none.
	 * @return int Rows processed.
	 */
	private function drain_queue( ?float $deadline ): int {
		$processed  = 0;
		$batch_size = $this->get_batch_size();
		$seen       = array();

		while ( ! $this->should_stop( $deadline ) ) {
			$items = $this->queue->claim_due( $batch_size );
			$fresh = array_filter(
				$items,
				static fn( Sync_Queue_Item $item ): bool => ! isset( $seen[ $item->get_id() . ':' . $item->get_version() ] )
			);

			// Only rows this run already handled came back: stop rather than spin.
			if ( array() === $fresh ) {
				break;
			}

			$by_type = array();

			foreach ( $fresh as $item ) {
				$seen[ $item->get_id() . ':' . $item->get_version() ] = true;
				$by_type[ $item->get_object_type() ][]                = $item;
			}

			foreach ( $by_type as $object_type => $type_items ) {
				$results = $this->processor->process(
					$object_type,
					array_map( static fn( Sync_Queue_Item $item ): int => $item->get_object_id(), $type_items ),
					null,
					$deadline
				);

				foreach ( $type_items as $item ) {
					$this->apply_result( $item, $results[ $item->get_object_id() ] ?? Object_Result::skipped() );
				}
			}

			$processed += count( $fresh );
			$this->between_batches();
		}

		return $processed;
	}

	/**
	 * Writes a processing result back to the queue.
	 *
	 * @since 1.5.0
	 *
	 * @param \WordPress\AI\Embeddings\Sync\Sync_Queue_Item $item   The claimed row.
	 * @param \WordPress\AI\Embeddings\Sync\Object_Result   $result The result.
	 */
	private function apply_result( Sync_Queue_Item $item, Object_Result $result ): void {
		if ( Object_Result::DEFERRED === $result->get_status() ) {
			$this->queue->postpone( $item, (int) $result->get_retry_at() );
			return;
		}

		if ( Object_Result::FAILED !== $result->get_status() ) {
			$this->queue->complete( $item );
			return;
		}

		$attempts = $item->get_attempts() + 1;

		if ( Embedding_Client_Exception::ITEM !== $result->get_error_class() && $attempts < $this->get_max_attempts() ) {
			$this->queue->defer( $item, time() + self::retry_delay( $attempts ), $result->get_message() );
			return;
		}

		if ( ! $this->queue->fail( $item, $result->get_message() ) ) {
			return;
		}

		/**
		 * Fires when a queued object is marked permanently failed.
		 *
		 * @since 1.5.0
		 *
		 * @param string $object_type Object type.
		 * @param int    $object_id   Object ID.
		 * @param string $error       The last error message.
		 */
		do_action( 'wpai_embedding_sync_object_failed', $item->get_object_type(), $item->get_object_id(), $result->get_message() );
	}

	/**
	 * Advances every running backfill while time remains.
	 *
	 * @since 1.5.0
	 *
	 * @param float|null $deadline Microtime deadline, or null for none.
	 * @return int Objects processed.
	 */
	private function run_backfills( ?float $deadline ): int {
		$processed  = 0;
		$batch_size = $this->get_batch_size();

		foreach ( $this->registry->get_targets() as $key => $target ) {
			while ( ! $this->should_stop( $deadline ) && $this->backfills->is_running( $key ) ) {
				$step = $this->backfill_step( $key, $target, $batch_size, $deadline );

				if ( null === $step ) {
					break;
				}

				$processed += $step;
				$this->between_batches();
			}
		}

		return $processed;
	}

	/**
	 * Processes one backfill batch for a target.
	 *
	 * @since 1.5.0
	 *
	 * @param string                                         $key        Target key.
	 * @param \WordPress\AI\Embeddings\Sync\Embedding_Target $target     The target.
	 * @param int                                            $batch_size Objects per batch.
	 * @param float|null                                     $deadline   Microtime deadline, or null for none.
	 * @return int|null Objects processed, or null when the target is paused or the batch ran out of time;
	 *                  the objects settled before the cut are still recorded and the cursor moved past them.
	 */
	private function backfill_step( string $key, Embedding_Target $target, int $batch_size, ?float $deadline ): ?int {
		$position = $this->backfills->get_position( $key, $target );

		// The sweep makes no API requests, so only the embedding pass waits out a pause.
		if ( null === $position ) {
			return $this->sweep_step( $key, $target );
		}

		if ( null !== $this->backoff->get_until_for( $target ) ) {
			return null;
		}

		$object_type = $position['object_type'];
		$source      = $this->sources[ $object_type ] ?? null;
		$ids         = null === $source ? array() : $source->get_ids_after( $position['cursor'], $target->get_subtypes_for( $object_type ), $batch_size );

		if ( array() === $ids ) {
			$this->backfills->finish_type( $key, $object_type );

			return 0;
		}

		$results = $this->processor->process( $object_type, $ids, $target, $deadline );

		$stop = null;

		foreach ( $results as $object_id => $result ) {
			if ( Object_Result::DEFERRED !== $result->get_status() ) {
				continue;
			}

			$stop = null === $stop ? $object_id : min( $stop, $object_id );
		}

		$counts  = array(
			'embedded' => 0,
			'skipped'  => 0,
			'removed'  => 0,
			'failed'   => 0,
		);
		$failed  = array();
		$settled = array();

		foreach ( $results as $object_id => $result ) {
			if ( null !== $stop && $object_id >= $stop ) {
				continue;
			}

			$settled[] = $object_id;

			switch ( $result->get_status() ) {
				case Object_Result::FAILED:
					$failed[] = $object_id;
					++$counts['failed'];
					break;
				case Object_Result::DONE:
					++$counts['embedded'];
					break;
				case Object_Result::REMOVED:
					++$counts['removed'];
					break;
				default:
					++$counts['skipped'];
			}
		}

		// One retry mechanism for both paths: the live queue owns retries.
		foreach ( $failed as $object_id ) {
			try {
				$this->queue->enqueue( $object_type, $object_id );
			} catch ( RuntimeException $e ) {
				wp_trigger_error( __METHOD__, esc_html( $e->getMessage() ), E_USER_WARNING );
			}
		}

		if ( null === $stop ) {
			$this->backfills->advance( $key, $object_type, (int) max( $ids ), $counts );

			return count( $ids );
		}

		if ( array() !== $settled ) {
			$this->backfills->advance( $key, $object_type, (int) max( $settled ), $counts );
		}

		// The target is paused or out of time: stop it for this run.
		return null;
	}

	/**
	 * Runs one page of the end-of-backfill orphan sweep, completing the backfill when it is done.
	 *
	 * @since 1.5.0
	 *
	 * @param string                                         $key    Target key.
	 * @param \WordPress\AI\Embeddings\Sync\Embedding_Target $target The target.
	 * @return int Objects checked.
	 */
	private function sweep_step( string $key, Embedding_Target $target ): int {
		$position = $this->backfills->get_sweep_position( $key, array_keys( $this->sources ) );

		if ( null === $position ) {
			$this->backfills->complete( $key, $target );

			return 0;
		}

		$page = $this->sweeper->sweep( $target, $position['object_type'], $position['cursor'], $this->get_sweep_batch_size() );

		if ( null === $page['last_id'] ) {
			$this->backfills->finish_sweep_type( $key, $position['object_type'] );

			return 0;
		}

		$this->backfills->advance_sweep( $key, $position['object_type'], $page['last_id'], $page['removed'] );

		return $page['checked'];
	}

	/**
	 * Keeps the lock fresh and memory flat between batches.
	 *
	 * @since 1.5.0
	 */
	private function between_batches(): void {
		if ( ! $this->lock->refresh() ) {
			$this->lock_lost = true;
		}

		wp_cache_flush_runtime();
	}

	/**
	 * Checks whether the run should stop: the lock was lost or the deadline has passed.
	 *
	 * @since 1.5.0
	 *
	 * @param float|null $deadline Microtime deadline, or null for none.
	 * @return bool True when the run should stop.
	 */
	private function should_stop( ?float $deadline ): bool {
		return $this->lock_lost || ( null !== $deadline && microtime( true ) >= $deadline );
	}

	/**
	 * Returns the batch size.
	 *
	 * @since 1.5.0
	 *
	 * @return int Objects per batch.
	 */
	private function get_batch_size(): int {
		/**
		 * Filters how many objects the embedding sync worker handles per batch.
		 *
		 * @since 1.5.0
		 *
		 * @param int $batch_size Objects per batch. Default 50.
		 */
		return max( 1, (int) apply_filters( 'wpai_embedding_sync_batch_size', self::DEFAULT_BATCH_SIZE ) );
	}

	/**
	 * Returns the orphan sweep batch size.
	 *
	 * @since 1.5.0
	 *
	 * @return int Stored objects checked per sweep batch.
	 */
	private function get_sweep_batch_size(): int {
		/**
		 * Filters how many stored objects the end-of-backfill orphan sweep checks per batch.
		 *
		 * The sweep makes no API requests, so its batches can be larger than embedding batches.
		 *
		 * @since 1.5.0
		 *
		 * @param int $batch_size Stored objects per batch. Default 200.
		 */
		return max( 1, (int) apply_filters( 'wpai_embedding_sync_sweep_batch_size', self::SWEEP_BATCH_SIZE ) );
	}

	/**
	 * Returns the attempt cap.
	 *
	 * @since 1.5.0
	 *
	 * @return int Attempts before a row is marked failed.
	 */
	private function get_max_attempts(): int {
		/**
		 * Filters how many attempts a queued object gets before it is marked failed.
		 *
		 * @since 1.5.0
		 *
		 * @param int $max_attempts Attempts. Default 5.
		 */
		return max( 1, (int) apply_filters( 'wpai_embedding_sync_max_attempts', self::DEFAULT_MAX_ATTEMPTS ) );
	}
}