<?php

declare(strict_types=1);

namespace SecurWP\Modules\Scanner\Queue;

use SecurWP\Modules\Scanner\Queue\ScanOrchestrator;
use SecurWP\Modules\Scanner\Queue\Handler\FinalizeHandler;
use SecurWP\Modules\Scanner\Queue\Handler\PollCloudHandler;
use SecurWP\Modules\Scanner\Queue\Handler\ScanWorkerHandler;
use SecurWP\Modules\Scanner\Queue\Handler\SystemCheckHandler;
use SecurWP\Modules\Scanner\Queue\Handler\TaskResult;
use SecurWP\Modules\Scanner\Queue\Handler\UploadBatchHandler;
use SecurWP\Modules\Scanner\Repository\CloudJobRepository;
use SecurWP\Modules\Scanner\Repository\TaskQueueRepository;

if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

/**
 * Leased task executor for scanner jobs.
 */
class TaskRunner {

	private const DEFAULT_LEASE_SECONDS = 30;
	private const UPLOAD_LEASE_SECONDS = 150;
	private const MAX_PENDING_UPLOAD_BATCHES = 3;
	// Task types that are intentionally fanned out into multiple parallel queue
	// entries. merge_follow_ups() must not collapse these by payload identity.
	private const FANOUT_TASK_TYPES = [ 'upload_batch' ];

	/**
	 * Execute one task or reconcile missing work for an active scan.
	 *
	 * @param string|null $scan_id
	 * @return array<string,mixed>
	 */
	public function run( ?string $scan_id = null ): array {
		$this->ensure_dependencies_loaded();

		$queue = $scan_id ? ScanQueue::get_job( $scan_id ) : ScanQueue::get();
		if ( ! is_array( $queue ) || ! in_array( (string) ( $queue['status'] ?? '' ), [ 'pending', 'running' ], true ) ) {
			return [
				'status'  => 'idle',
				'message' => 'No active scan',
			];
		}

		$scan_id = (string) ( $queue['id'] ?? '' );
		if ( '' === $scan_id ) {
			return [
				'status'  => 'idle',
				'message' => 'Missing scan id',
			];
		}

		$repo = new TaskQueueRepository();
		$repo->release_expired_leases( $scan_id );

		$lease_token = wp_generate_password( 32, false );
		if ( ! $repo->acquire_scan_lease( $scan_id, $lease_token, self::DEFAULT_LEASE_SECONDS ) ) {
			return [
				'status'  => 'skipped',
				'message' => 'Another task runner owns the scan lease',
			];
		}

		$task = null;

		try {
			$task = $repo->claim( $scan_id, self::DEFAULT_LEASE_SECONDS, self::UPLOAD_LEASE_SECONDS );

			if ( null === $task ) {
				$orchestrator = new ScanOrchestrator( $scan_id );
				$runtime_recovery = $orchestrator->recover_task_queue_runtime_state();
				$queue = ScanQueue::get_job( $scan_id ) ?: $queue;
				$follow_ups = $this->attach_scan_id(
					$this->reconcile_follow_ups( $scan_id, $queue, $repo, $orchestrator ),
					$scan_id
				);
				$inserted = 0;
				if ( ! empty( $follow_ups ) ) {
					$inserted = $repo->enqueue_batch( $follow_ups );
				}

				ScanQueue::heartbeat();
				$queue_after_reconcile = ScanQueue::get_job( $scan_id );
				if ( is_array( $queue_after_reconcile ) && ! in_array( (string) ( $queue_after_reconcile['status'] ?? '' ), [ 'pending', 'running' ], true ) ) {
					$repo->delete_by_scan_id( $scan_id );
				}

				return [
					'status'         => ( empty( $follow_ups ) && ! $this->did_runtime_recover( $runtime_recovery ) ) ? 'idle' : 'reconciled',
					'enqueued'       => $inserted,
					'pending_work'   => $repo->has_pending_work( $scan_id ),
					'available_work' => $repo->has_available_work( $scan_id ),
					'next_available_at' => $repo->get_next_available_at( $scan_id ),
				];
			}

			$lease_seconds = $this->get_lease_seconds( (string) $task['task_type'] );
			if ( ! $repo->extend_scan_lease( $scan_id, $lease_token, $lease_seconds ) ) {
				if ( $this->is_scan_active( ScanQueue::get_job( $scan_id ) ) ) {
					$repo->fail( (int) $task['id'], 'Unable to extend scan lease before task execution' );
				} else {
					$repo->delete_by_scan_id( $scan_id );
				}

				return [
					'status'            => 'skipped',
					'message'           => 'Lost scan lease before task execution',
					'pending_work'      => $repo->has_pending_work( $scan_id ),
					'available_work'    => $repo->has_available_work( $scan_id ),
					'next_available_at' => $repo->get_next_available_at( $scan_id ),
				];
			}

			$handler = $this->resolve_handler( (string) $task['task_type'] );
			$result = $handler ? $handler->handle( $task, $scan_id ) : TaskResult::failed( 'Unknown task type: ' . (string) $task['task_type'] );
			$follow_ups = $this->attach_scan_id( $result->enqueue, $scan_id );

			$queue_after = ScanQueue::get_job( $scan_id );
			$scan_active = is_array( $queue_after ) && in_array( (string) ( $queue_after['status'] ?? '' ), [ 'pending', 'running' ], true );
			if ( $scan_active ) {
				$reconciled_follow_ups = $this->attach_scan_id(
					$this->reconcile_follow_ups( $scan_id, $queue_after, $repo ),
					$scan_id
				);
				$follow_ups = $this->merge_follow_ups( $follow_ups, $reconciled_follow_ups );
				$queue_after = ScanQueue::get_job( $scan_id );
				$scan_active = is_array( $queue_after ) && in_array( (string) ( $queue_after['status'] ?? '' ), [ 'pending', 'running' ], true );
			}

			if ( 'completed' === $result->status || ! $scan_active ) {
				$committed = $repo->complete_and_enqueue( (int) $task['id'], $result->result, $follow_ups );
				if ( ! $committed ) {
					$repo->fail( (int) $task['id'], 'Task completion transaction failed' );
				}
			} else {
				$repo->fail( (int) $task['id'], (string) ( $result->error ?? 'Task failed' ) );
			}

			ScanQueue::heartbeat();
			$this->cleanup_if_terminal( $scan_id, $repo );

			return [
				'status'         => 'completed',
				'task_id'        => (int) $task['id'],
				'task_type'      => (string) $task['task_type'],
				'handler_status' => $result->status,
				'available_work' => $repo->has_available_work( $scan_id ),
				'pending_work'   => $repo->has_pending_work( $scan_id ),
				'next_available_at' => $repo->get_next_available_at( $scan_id ),
			];
		} catch ( \Throwable $e ) {
			$message = sanitize_text_field( $e->getMessage() );
			if ( is_array( $task ) ) {
				if ( $this->is_scan_active( ScanQueue::get_job( $scan_id ) ) ) {
					$repo->fail( (int) $task['id'], $message );
				} else {
					$repo->complete_and_enqueue(
						(int) $task['id'],
						[
							'status'  => 'aborted',
							'message' => $message,
						],
						[]
					);
					$repo->delete_by_scan_id( $scan_id );
				}
			}

			return [
				'status'  => 'failed',
				'message' => $message,
			];
		} finally {
			$repo->release_scan_lease( $scan_id, $lease_token );
		}
	}

	/**
	 * Recreate missing tasks from authoritative scan state.
	 *
	 * @param string                 $scan_id
	 * @param array<string,mixed>    $queue
	 * @param TaskQueueRepository    $repo
	 * @param ScanOrchestrator|null  $orchestrator
	 * @return array<int,array<string,mixed>>
	 */
	private function reconcile_follow_ups(
		string $scan_id,
		array $queue,
		TaskQueueRepository $repo,
		?ScanOrchestrator $orchestrator = null
	): array {
		$follow_ups = [];
		$dead_tasks = $repo->get_dead_tasks( $scan_id );
		$this->release_legacy_scan_poll_dead_letters( $dead_tasks, $repo );
		if ( $this->has_dead_task_type( $dead_tasks, 'finalize' ) ) {
			ScanQueue::fail(
				'Scanner finalization failed repeatedly. Please restart the scan.',
				[
					'error_code'        => 'finalize_task_dead_letter',
					'stage'             => 'finalize',
					'retryable'         => false,
					'status_code'       => 500,
					'technical_message' => 'Finalize task reached dead-letter state',
				]
			);

			return [];
		}

		$orchestrator = $orchestrator ?: new ScanOrchestrator( $scan_id );

		if ( $orchestrator->has_file_worker_work() && ! $repo->has_task( $scan_id, 'scan_worker', [ TaskQueueRepository::STATUS_PENDING, TaskQueueRepository::STATUS_LEASED ] ) ) {
			$follow_ups[] = [
				'task_type'    => 'scan_worker',
				'payload'      => [ 'budget_seconds' => 20 ],
				'available_at' => $orchestrator->get_next_file_worker_available_at(),
				'priority'     => 100,
			];
		}

		$upload_payload = [
			'max_files'   => 50,
			'max_size_mb' => 8,
		];
		if ( $orchestrator->has_ready_upload_batch( $upload_payload ) ) {
			$inflight_uploads = $repo->count_tasks(
				$scan_id,
				'upload_batch',
				[ TaskQueueRepository::STATUS_PENDING, TaskQueueRepository::STATUS_LEASED ]
			);
			$slots = max( 0, self::MAX_PENDING_UPLOAD_BATCHES - $inflight_uploads );
			for ( $i = 0; $i < $slots; $i++ ) {
				$follow_ups[] = [
					'task_type'    => 'upload_batch',
					'payload'      => $upload_payload,
					'available_at' => time(),
					'priority'     => 70,
				];
			}
		}

		$job_repo = new CloudJobRepository();
		$jobs_needing_poll = $job_repo->get_jobs_needing_poll( $scan_id );
		if (
			! empty( $jobs_needing_poll ) &&
			! $this->has_blocking_dead_task_type( $dead_tasks, 'poll_cloud' ) &&
			! $repo->has_task( $scan_id, 'poll_cloud', [ TaskQueueRepository::STATUS_PENDING, TaskQueueRepository::STATUS_LEASED ] )
		) {
			$follow_ups[] = [
				'task_type'    => 'poll_cloud',
				'payload'      => [
					'poll_count' => 0,
				],
				'available_at' => time(),
				'priority'     => 60,
				'dedupe_key'   => TaskResult::build_scan_poll_dedupe_key(),
			];
		}

		$component_status = ScanQueue::get_component_status( $queue, false );
		foreach ( ScanQueue::SYSTEM_CHECKS as $slug ) {
			$status = $component_status[ $slug ] ?? 'pending';
			if ( in_array( $status, [ 'complete', 'failed' ], true ) ) {
				continue;
			}

			$dedupe_key = 'syscheck:' . $slug;
			if (
				! $repo->has_task(
					$scan_id,
					'system_check',
					[
						TaskQueueRepository::STATUS_PENDING,
						TaskQueueRepository::STATUS_LEASED,
						TaskQueueRepository::STATUS_DEAD,
					],
					$dedupe_key
				)
			) {
				$follow_ups[] = [
					'task_type'    => 'system_check',
					'payload'      => [ 'check_slug' => $slug ],
					'available_at' => time(),
					'priority'     => ScanQueue::SYSTEM_CHECK_PRIORITY,
					'dedupe_key'   => $dedupe_key,
				];
			}
		}

		$is_ready_for_finalize = $orchestrator->is_ready_for_finalize();
		if ( ! $is_ready_for_finalize && empty( $follow_ups ) ) {
			$blocking_dead_task = $this->get_blocking_dead_pipeline_task( $dead_tasks );
			if ( is_array( $blocking_dead_task ) ) {
				$this->fail_scan_for_dead_task( $blocking_dead_task );
				return [];
			}
		}

		if ( $is_ready_for_finalize && ! $repo->has_task( $scan_id, 'finalize', [ TaskQueueRepository::STATUS_PENDING, TaskQueueRepository::STATUS_LEASED ], 'finalize' ) ) {
			$follow_ups[] = [
				'task_type'    => 'finalize',
				'payload'      => [ 'scan_id' => $scan_id ],
				'available_at' => time(),
				'priority'     => 10,
				'dedupe_key'   => 'finalize',
			];
		}

		return $follow_ups;
	}

	/**
	 * @param array<int,array<string,mixed>> $dead_tasks
	 * @param string                         $task_type
	 * @return bool
	 */
	private function has_dead_task_type( array $dead_tasks, string $task_type ): bool {
		foreach ( $dead_tasks as $dead_task ) {
			if ( $task_type === (string) ( $dead_task['task_type'] ?? '' ) ) {
				return true;
			}
		}

		return false;
	}

	/**
	 * Check whether dead task state should prevent recovery for this task type.
	 *
	 * @param array<int,array<string,mixed>> $dead_tasks
	 * @param string                         $task_type
	 * @return bool
	 */
	private function has_blocking_dead_task_type( array $dead_tasks, string $task_type ): bool {
		foreach ( $dead_tasks as $dead_task ) {
			if ( $task_type !== (string) ( $dead_task['task_type'] ?? '' ) ) {
				continue;
			}

			if ( 'poll_cloud' === $task_type && $this->is_legacy_scan_poll_dead_letter( $dead_task ) ) {
				continue;
			}

			return true;
		}

		return false;
	}

	/**
	 * Return a dead file-pipeline task that should fail the scan instead of finalizing.
	 *
	 * @param array<int,array<string,mixed>> $dead_tasks
	 * @return array<string,mixed>|null
	 */
	private function get_blocking_dead_pipeline_task( array $dead_tasks ): ?array {
		foreach ( $dead_tasks as $dead_task ) {
			$task_type = (string) ( $dead_task['task_type'] ?? '' );
			if ( ! in_array( $task_type, [ 'scan_worker', 'upload_batch', 'poll_cloud' ], true ) ) {
				continue;
			}

			if ( 'poll_cloud' === $task_type && $this->is_legacy_scan_poll_dead_letter( $dead_task ) ) {
				continue;
			}

			return $dead_task;
		}

		return null;
	}

	/**
	 * Detect scan-level poll rows dead-lettered by the removed global poll limit.
	 *
	 * @param array<string,mixed> $dead_task
	 * @return bool
	 */
	private function is_legacy_scan_poll_dead_letter( array $dead_task ): bool {
		$payload = is_array( $dead_task['payload'] ?? null ) ? $dead_task['payload'] : [];
		if ( ! empty( $payload['job_ids'] ) ) {
			return false;
		}

		$result = is_array( $dead_task['result'] ?? null ) ? $dead_task['result'] : [];
		$error = (string) ( $result['error'] ?? '' );

		return false !== strpos( $error, 'did not complete after' );
	}

	/**
	 * Release dedupe keys from legacy poll dead letters so reconciliation can poll again.
	 *
	 * @param array<int,array<string,mixed>> $dead_tasks
	 * @param TaskQueueRepository            $repo
	 * @return void
	 */
	private function release_legacy_scan_poll_dead_letters( array $dead_tasks, TaskQueueRepository $repo ): void {
		foreach ( $dead_tasks as $dead_task ) {
			if ( 'poll_cloud' !== (string) ( $dead_task['task_type'] ?? '' ) ) {
				continue;
			}

			if ( empty( $dead_task['dedupe_key'] ) || ! $this->is_legacy_scan_poll_dead_letter( $dead_task ) ) {
				continue;
			}

			$repo->clear_dedupe_key( (int) ( $dead_task['id'] ?? 0 ) );
		}
	}

	/**
	 * Fail the scan when a critical queue task is dead and no safe recovery exists.
	 *
	 * @param array<string,mixed> $dead_task
	 * @return void
	 */
	private function fail_scan_for_dead_task( array $dead_task ): void {
		$task_type = sanitize_key( (string) ( $dead_task['task_type'] ?? 'task' ) );
		$result = is_array( $dead_task['result'] ?? null ) ? $dead_task['result'] : [];
		$error_message = sanitize_text_field( (string) ( $result['error'] ?? 'Background task failed repeatedly' ) );

		ScanQueue::fail(
			sprintf( 'Scanner background task "%s" failed repeatedly. Please restart the scan.', $task_type ),
			[
				'error_code'        => 'task_dead_letter',
				'stage'             => $task_type,
				'retryable'         => true,
				'status_code'       => 500,
				'technical_message' => $error_message,
			]
		);
	}

	/**
	 * @param string $task_type
	 * @return object|null
	 */
	private function resolve_handler( string $task_type ): ?object {
		switch ( $task_type ) {
			case 'scan_worker':
				return new ScanWorkerHandler();
			case 'upload_batch':
				return new UploadBatchHandler();
			case 'poll_cloud':
				return new PollCloudHandler();
			case 'system_check':
				return new SystemCheckHandler();
			case 'finalize':
				return new FinalizeHandler();
			default:
				return null;
		}
	}

	/**
	 * @param string $task_type
	 * @return int
	 */
	private function get_lease_seconds( string $task_type ): int {
		return 'upload_batch' === $task_type ? self::UPLOAD_LEASE_SECONDS : self::DEFAULT_LEASE_SECONDS;
	}

	/**
	 * @param array<int,array<string,mixed>> $tasks
	 * @param string                         $scan_id
	 * @return array<int,array<string,mixed>>
	 */
	private function attach_scan_id( array $tasks, string $scan_id ): array {
		foreach ( $tasks as $index => $task ) {
			if ( empty( $task['scan_id'] ) ) {
				$tasks[ $index ]['scan_id'] = $scan_id;
			}
		}

		return $tasks;
	}

	/**
	 * Merge handler-declared follow-ups with authoritative-state reconciliation.
	 *
	 * The durable queue is a projection of the file/cloud FSM, not the source of
	 * truth. After a task mutates the scan state, another task can become runnable
	 * even when the handler did not enqueue it directly. Collapsing both sets here
	 * prevents the runner from going idle until a later watchdog or cron wake-up.
	 *
	 * @param array<int,array<string,mixed>> $primary
	 * @param array<int,array<string,mixed>> $secondary
	 * @return array<int,array<string,mixed>>
	 */
	private function merge_follow_ups( array $primary, array $secondary ): array {
		$merged = [];

		foreach ( array_merge( $primary, $secondary ) as $task ) {
			$identity = $this->get_follow_up_identity( $task );
			if ( isset( $merged[ $identity ] ) ) {
				$existing = $merged[ $identity ];
				$existing_available_at = (int) ( $existing['available_at'] ?? time() );
				$task_available_at = (int) ( $task['available_at'] ?? $existing_available_at );
				$existing_priority = (int) ( $existing['priority'] ?? 100 );
				$task_priority = (int) ( $task['priority'] ?? $existing_priority );

				$merged[ $identity ]['available_at'] = min( $existing_available_at, $task_available_at );
				$merged[ $identity ]['priority'] = min( $existing_priority, $task_priority );

				if ( empty( $existing['dedupe_key'] ) && ! empty( $task['dedupe_key'] ) ) {
					$merged[ $identity ]['dedupe_key'] = $task['dedupe_key'];
				}

				if ( empty( $existing['scan_id'] ) && ! empty( $task['scan_id'] ) ) {
					$merged[ $identity ]['scan_id'] = $task['scan_id'];
				}

				if ( ! isset( $existing['max_attempts'] ) && isset( $task['max_attempts'] ) ) {
					$merged[ $identity ]['max_attempts'] = $task['max_attempts'];
				}

				continue;
			}

			$merged[ $identity ] = $task;
		}

		return array_values( $merged );
	}

	/**
	 * Build a stable identity for follow-up task de-duplication before enqueue.
	 *
	 * @param array<string,mixed> $task
	 * @return string
	 */
	private function get_follow_up_identity( array $task ): string {
		$task_type = (string) ( $task['task_type'] ?? '' );
		$dedupe_key = isset( $task['dedupe_key'] ) ? sanitize_text_field( (string) $task['dedupe_key'] ) : '';
		if ( '' !== $dedupe_key ) {
			return $task_type . '|dedupe:' . $dedupe_key;
		}

		// Fan-out task types: each emit is a parallel slot, not a duplicate.
		// Returning a unique identity prevents merge_follow_ups() from collapsing
		// multiple queued instances into one.
		if ( in_array( $task_type, self::FANOUT_TASK_TYPES, true ) ) {
			static $fanout_seq = 0;
			return $task_type . '|fanout:' . ( ++$fanout_seq );
		}

		$payload = is_array( $task['payload'] ?? null ) ? $task['payload'] : [];
		$payload_json = wp_json_encode( $payload );
		if ( false === $payload_json ) {
			$payload_json = '';
		}

		return $task_type . '|payload:' . $payload_json;
	}

	/**
	 * @param array<string,mixed> $runtime_recovery
	 * @return bool
	 */
	private function did_runtime_recover( array $runtime_recovery ): bool {
		return ! empty( $runtime_recovery['cloud_states_recovered'] ) || (int) ( $runtime_recovery['stale_uploads'] ?? 0 ) > 0;
	}

	/**
	 * Delete durable task rows once the scan has entered a terminal state.
	 *
	 * @param string              $scan_id
	 * @param TaskQueueRepository $repo
	 * @return bool True when the scan is terminal.
	 */
	private function cleanup_if_terminal( string $scan_id, TaskQueueRepository $repo ): bool {
		$queue = ScanQueue::get_job( $scan_id );
		if ( $this->is_scan_active( $queue ) ) {
			return false;
		}

		$repo->delete_by_scan_id( $scan_id );
		return true;
	}

	/**
	 * @param array<string,mixed>|null $queue
	 * @return bool
	 */
	private function is_scan_active( ?array $queue ): bool {
		return is_array( $queue ) && in_array( (string) ( $queue['status'] ?? '' ), [ 'pending', 'running' ], true );
	}

	/**
	 * Load required handler classes in environments without autoloading.
	 *
	 * @return void
	 */
	private function ensure_dependencies_loaded(): void {
		if ( ! class_exists( TaskQueueRepository::class ) ) {
			require_once dirname( __DIR__ ) . '/Repository/TaskQueueRepository.php';
		}
		if ( ! class_exists( CloudJobRepository::class ) ) {
			require_once dirname( __DIR__ ) . '/Repository/CloudJobRepository.php';
		}
		if ( ! class_exists( ScanOrchestrator::class ) ) {
			require_once __DIR__ . '/ScanOrchestrator.php';
		}
		if ( ! class_exists( TaskResult::class ) ) {
			require_once __DIR__ . '/Handler/TaskResult.php';
		}
		if ( ! class_exists( ScanWorkerHandler::class ) ) {
			require_once __DIR__ . '/Handler/ScanWorkerHandler.php';
		}
		if ( ! class_exists( UploadBatchHandler::class ) ) {
			require_once __DIR__ . '/Handler/UploadBatchHandler.php';
		}
		if ( ! class_exists( PollCloudHandler::class ) ) {
			require_once __DIR__ . '/Handler/PollCloudHandler.php';
		}
		if ( ! class_exists( SystemCheckHandler::class ) ) {
			require_once __DIR__ . '/Handler/SystemCheckHandler.php';
		}
		if ( ! class_exists( FinalizeHandler::class ) ) {
			require_once __DIR__ . '/Handler/FinalizeHandler.php';
		}
	}
}
