<?php

declare(strict_types=1);

namespace SiteFort\Modules\Cloud\Services;

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

/**
 * Durable local queue for Console commands.
 *
 * Commands are accepted synchronously by the REST endpoint and persisted
 * locally. Fast commands run inline in the same request; the rest run via
 * WP-Cron outside the original request.
 */
class CommandQueue {

	public const WORKER_HOOK = 'sitefort_command_execute';
	public const MAINTENANCE_HOOK = 'sitefort_command_queue_maintenance';
	public const TABLE_NAME = 'sitefort_commands';

	private const LEGACY_OPTION_KEY = 'sitefort_command_statuses';
	private const SCHEMA_VERSION = '2';
	private const SCHEMA_VERSION_OPTION = 'sitefort_command_queue_schema_version';
	private const MAX_COMMANDS = 200;
	private const MAX_AGE_SECONDS = 604800; // 7 days.
	private const MAX_RESULT_BYTES = 8192;
	private const WORKER_LOCK_TIMEOUT = 0;
	private const ACCEPTED_STALE_SECONDS = 1800;
	private const RUNNING_STALE_SECONDS = 1800; // 30 minutes.
	private const STALE_RECONCILE_BATCH = 500;
	private const ACK_MODE_TERMINAL = 'terminal';
	private const ACK_MODE_ACCEPTED = 'accepted';
	private const TERMINAL_STATUSES = [
		'succeeded',
		'failed',
		'cancelled',
		'rejected',
		'expired',
	];

	/**
	 * Runtime table bootstrap guard.
	 *
	 * @var bool
	 */
	private static bool $storage_ready = false;

	/**
	 * Register worker hooks.
	 */
	public static function register_hooks(): void {
		// The table is provisioned at activation and every queue operation
		// self-ensures storage, so only re-verify the schema in privileged
		// contexts rather than on every front-end request.
		if ( \SiteFort\Includes\RequestContext::is_privileged() ) {
			self::ensure_storage();
		}
		add_action( self::WORKER_HOOK, [ __CLASS__, 'process' ], 10, 1 );
		add_action( self::MAINTENANCE_HOOK, [ __CLASS__, 'maintain' ], 10, 0 );
		self::ensure_maintenance_scheduled();
	}

	/**
	 * Create or upgrade the command queue table.
	 *
	 * @return void
	 */
	public static function ensure_table(): void {
		self::create_or_upgrade_table();
		update_option( self::SCHEMA_VERSION_OPTION, self::SCHEMA_VERSION, true );
	}

	/**
	 * Create or upgrade the command queue table.
	 *
	 * @return void
	 */
	private static function create_or_upgrade_table(): void {
		global $wpdb;

		$table = self::table_name();
		$charset_collate = \SiteFort\Core\Database\SchemaInspector::validated_charset_collate( $wpdb );

		$sql = "CREATE TABLE {$table} (
			command_id varchar(64) NOT NULL,
			command varchar(80) NOT NULL,
			params longtext DEFAULT NULL,
			status varchar(32) NOT NULL DEFAULT 'accepted',
			result longtext DEFAULT NULL,
			error text DEFAULT NULL,
			error_code varchar(64) DEFAULT '',
			ack_mode varchar(16) NOT NULL DEFAULT 'terminal',
			executor_version varchar(32) DEFAULT '',
			created_at int(11) NOT NULL,
			accepted_at int(11) NOT NULL DEFAULT 0,
			updated_at int(11) NOT NULL DEFAULT 0,
			started_at int(11) NOT NULL DEFAULT 0,
			completed_at int(11) NOT NULL DEFAULT 0,
			PRIMARY KEY (command_id),
			KEY idx_status_updated (status, updated_at),
			KEY idx_updated_at (updated_at)
		) {$charset_collate};";

		require_once ABSPATH . 'wp-admin/includes/upgrade.php';
		dbDelta( $sql );
		\SiteFort\Core\Database\SchemaInspector::flush_cache( $table );
		\SiteFort\Core\Database\SchemaInspector::ensure_table_uses_innodb( $table );
	}

	/**
	 * Persist a command locally and schedule background execution.
	 *
	 * @param string              $command_id Command UUID.
	 * @param string              $command    Command name.
	 * @param array<string,mixed> $params     Command parameters.
	 * @param string              $ack_mode   Command acknowledgement policy.
	 * @param bool                $schedule   False when the caller runs the command inline.
	 * @return array<string,mixed>
	 */
	public static function enqueue( string $command_id, string $command, array $params, string $ack_mode = self::ACK_MODE_TERMINAL, bool $schedule = true ): array {
		self::ensure_storage();

		$existing = self::get( $command_id );
		if ( is_array( $existing ) ) {
			$existing = self::refresh_record_state( $existing );
			if ( $schedule && self::should_schedule_record( $existing ) ) {
				self::schedule( $command_id );
			}
			return $existing;
		}

		global $wpdb;

		$command = sanitize_text_field( $command );
		$ack_mode = self::normalize_ack_mode( $ack_mode );
		$now = time();
		$params_json = wp_json_encode( $params );
		if ( false === $params_json ) {
			$params_json = '{}';
		}

		$table = self::table_identifier( self::table_name() );
		$insert_result = null;
		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$insert_result = $wpdb->query(
			$wpdb->prepare(
				"INSERT IGNORE INTO {$table}
				(command_id, command, params, status, result, error, error_code, ack_mode, executor_version, created_at, accepted_at, updated_at, started_at, completed_at)
				VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %d, %d, %d, %d, %d)",
				$command_id,
				$command,
				$params_json,
				'accepted',
				'[]',
				'',
				'',
				$ack_mode,
				CommandExecutor::VERSION,
				$now,
				$now,
				$now,
				0,
				0
			)
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		$insert_error = (string) $wpdb->last_error;

		$record = self::get( $command_id );
		if ( is_array( $record ) ) {
			if ( 1 === (int) $insert_result ) {
				self::prune_records();
				if ( $schedule ) {
					self::schedule( $command_id );
				}
			}
			return $record;
		}

		throw new \RuntimeException(
			'' !== $insert_error
				? sprintf( 'Command persistence failed: %s', esc_html( sanitize_text_field( $insert_error ) ) )
				: 'Command acceptance could not be confirmed locally.'
		);
	}

	/**
	 * Return a stored command payload, or null when unknown.
	 *
	 * @param string $command_id Command UUID.
	 * @return array<string,mixed>|null
	 */
	public static function get( string $command_id ): ?array {
		self::ensure_storage();

		global $wpdb;

		$table = self::table_identifier( self::table_name() );
		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$row = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT command_id, command, params, status, result, error, error_code, ack_mode, executor_version, created_at, accepted_at, updated_at, started_at, completed_at
				FROM {$table}
				WHERE command_id = %s
				LIMIT 1",
				$command_id
			),
			ARRAY_A
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter

		if ( ! is_array( $row ) ) {
			return null;
		}

		return self::normalize_loaded_record( $row );
	}

	/**
	 * Return a sanitized status payload for the Console status endpoint.
	 *
	 * @param string $command_id Command UUID.
	 * @return array<string,mixed>|null
	 */
	public static function get_status_payload( string $command_id ): ?array {
		$record = self::get( $command_id );
		if ( ! is_array( $record ) ) {
			return null;
		}

		$record = self::refresh_record_state( $record );
		if ( self::should_schedule_record( $record ) ) {
			self::schedule( $command_id );
		}

		$result = $record['result'] ?? [];
		if ( ! is_array( $result ) ) {
			$result = [];
		}
		$status = strtolower( (string) ( $record['status'] ?? 'accepted' ) );

		return [
			'ok'               => ! in_array( $status, [ 'failed', 'cancelled', 'rejected', 'expired' ], true ),
			'id'               => $command_id,
			'command'          => (string) ( $record['command'] ?? '' ),
			'command_status'   => $status,
			'ack_mode'         => (string) ( $record['ack_mode'] ?? self::ACK_MODE_TERMINAL ),
			'result'           => empty( $result ) ? (object) [] : $result,
			'error'            => (string) ( $record['error'] ?? '' ),
			'error_code'       => (string) ( $record['error_code'] ?? '' ),
			'executor_version' => (string) ( $record['executor_version'] ?? CommandExecutor::VERSION ),
			'created_at'       => (int) ( $record['created_at'] ?? 0 ),
			'updated_at'       => (int) ( $record['updated_at'] ?? 0 ),
			'started_at'       => (int) ( $record['started_at'] ?? 0 ),
			'completed_at'     => (int) ( $record['completed_at'] ?? 0 ),
		];
	}

	/**
	 * Background worker entrypoint.
	 *
	 * @param string $command_id Command UUID.
	 * @return void
	 */
	public static function process( string $command_id ): void {
		self::ensure_storage();

		if ( ! self::acquire_worker_lock( $command_id ) ) {
			return;
		}

		$executed = false;
		try {
			$record = self::get( $command_id );
			if ( ! is_array( $record ) ) {
				return;
			}
			$record = self::refresh_record_state( $record, true, false );

			$status = strtolower( (string) ( $record['status'] ?? '' ) );
			if ( 'running' === $status || in_array( $status, self::TERMINAL_STATUSES, true ) ) {
				return;
			}

			$executed = self::execute_record( $command_id, $record, true );
		} finally {
			self::release_worker_lock( $command_id );
			if ( $executed ) {
				self::prune_records( false );
			}
		}

		if ( $executed ) {
			self::flush_dirty_state();
		}
	}

	/**
	 * Execute a command inline during the delivery request.
	 *
	 * @param string $command_id Command UUID.
	 * @return array<string,mixed>|null Status payload, or null when the record is missing.
	 */
	public static function run_now( string $command_id ): ?array {
		self::ensure_storage();

		if ( self::acquire_worker_lock( $command_id ) ) {
			$executed = false;
			try {
				$record = self::get( $command_id );
				if ( ! is_array( $record ) ) {
					return null;
				}

				$status = strtolower( (string) ( $record['status'] ?? '' ) );
				if ( 'running' !== $status && ! in_array( $status, self::TERMINAL_STATUSES, true ) ) {
					$executed = self::execute_record( $command_id, $record, false );
				}
			} finally {
				self::release_worker_lock( $command_id );
				if ( $executed ) {
					self::prune_records( false );
				}
			}

			if ( $executed ) {
				self::flush_dirty_state();
			}
		}

		// Also re-schedules the worker when execution could not run here.
		return self::get_status_payload( $command_id );
	}

	/**
	 * Push dirty state now instead of waiting for the deferred dispatch tick.
	 */
	private static function flush_dirty_state(): void {
		try {
			StatePush::process_deferred();
		} catch ( \Throwable $e ) {
			sitefort_log( '[SiteFort] CommandQueue state flush failed: ' . $e->getMessage() );
		}
	}

	/**
	 * Run the command and persist the terminal state.
	 *
	 * @param string              $command_id Command UUID.
	 * @param array<string,mixed> $record     Command row.
	 * @param bool                $notify     Whether to send ACKs to Console.
	 * @return bool True when a terminal state was persisted.
	 */
	private static function execute_record( string $command_id, array $record, bool $notify ): bool {
		try {
			$command = sanitize_text_field( (string) ( $record['command'] ?? '' ) );
			$params = isset( $record['params'] ) && is_array( $record['params'] ) ? $record['params'] : [];
			if ( ! self::mark_running( $command_id, $notify ) ) {
				return false;
			}

			$executor = new CommandExecutor();
			$result = $executor->execute( $command, $params );
			$ok = is_array( $result ) && ! empty( $result['ok'] );

			if ( $ok ) {
				self::finalize_success( $command_id, is_array( $result ) ? $result : [], $notify );
				return true;
			}

			$error = sanitize_text_field( (string) ( $result['error'] ?? 'command_failed' ) );
			$error_code = sanitize_text_field( (string) ( $result['error_code'] ?? '' ) );
			self::finalize_failure(
				$command_id,
				$error,
				'' !== $error_code ? $error_code : null,
				is_array( $result ) ? $result : [],
				$notify
			);
			return true;
		} catch ( \Throwable $e ) {
			sitefort_log( '[SiteFort] CommandQueue process failed for ' . $command_id . ': ' . $e->getMessage() );
			self::finalize_failure(
				$command_id,
				'command_failed',
				'EXEC_RUNTIME_ERROR',
				[
					'error' => 'Command execution failed after acceptance.',
				],
				$notify
			);
			return true;
		}
	}

	/**
	 * Run scheduled command queue cleanup.
	 */
	public static function maintain(): void {
		self::ensure_storage();
		self::prune_records();
	}

	/**
	 * Schedule the command worker.
	 *
	 * @param string $command_id Command UUID.
	 * @return void
	 */
	private static function schedule( string $command_id ): void {
		if ( ! wp_next_scheduled( self::WORKER_HOOK, [ $command_id ] ) ) {
			// Due immediately; scheduling even 1s ahead makes spawn_cron() a no-op.
			wp_schedule_single_event( time(), self::WORKER_HOOK, [ $command_id ] );
		}

		spawn_cron();
	}

	/**
	 * Keep command cleanup available even when no new commands arrive.
	 */
	private static function ensure_maintenance_scheduled(): void {
		if ( ! wp_next_scheduled( self::MAINTENANCE_HOOK ) ) {
			wp_schedule_event( time() + 300, 'hourly', self::MAINTENANCE_HOOK );
		}
	}

	/**
	 * Whether a stored command still needs background execution.
	 *
	 * @param array<string,mixed> $record Command row.
	 * @return bool
	 */
	private static function should_schedule_record( array $record ): bool {
		$status = strtolower( (string) ( $record['status'] ?? '' ) );
		if ( 'running' === $status ) {
			return false;
		}

		return ! in_array( $status, self::TERMINAL_STATUSES, true );
	}

	/**
	 * Update one stored command record.
	 *
	 * @param string              $command_id Command UUID.
	 * @param array<string,mixed> $changes    Partial record changes.
	 * @return void
	 */
	private static function update_record( string $command_id, array $changes, bool $prune = true ): void {
		$record = self::get( $command_id );
		if ( ! is_array( $record ) ) {
			throw new \RuntimeException( 'Stored command row is missing.' );
		}

		foreach ( $changes as $key => $value ) {
			$record[ $key ] = $value;
		}

		self::save_record( $record, $prune );
	}

	/**
	 * Persist one command row atomically.
	 *
	 * @param array<string,mixed> $record Command row.
	 * @return void
	 */
	private static function save_record( array $record, bool $prune = true ): void {
		global $wpdb;

		$command = sanitize_text_field( (string) ( $record['command'] ?? '' ) );
		$params = isset( $record['params'] ) && is_array( $record['params'] ) ? $record['params'] : [];
		$result = isset( $record['result'] ) && is_array( $record['result'] ) ? $record['result'] : [];
		$result = self::normalize_result_for_storage( $command, $result );

		$params_json = wp_json_encode( $params );
		$result_json = wp_json_encode( $result );
		if ( false === $params_json ) {
			$params_json = '{}';
		}
		if ( false === $result_json ) {
			$result_json = '[]';
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Persists command execution state immediately; object caching would risk stale command status.
		$wpdb->replace(
			self::table_name(),
			[
				'command_id'       => (string) ( $record['id'] ?? $record['command_id'] ?? '' ),
				'command'          => $command,
				'params'           => $params_json,
				'status'           => sanitize_text_field( (string) ( $record['status'] ?? 'accepted' ) ),
				'result'           => $result_json,
				'error'            => sanitize_text_field( (string) ( $record['error'] ?? '' ) ),
				'error_code'       => sanitize_text_field( (string) ( $record['error_code'] ?? '' ) ),
				'ack_mode'         => self::normalize_ack_mode( (string) ( $record['ack_mode'] ?? self::ACK_MODE_TERMINAL ) ),
				'executor_version' => sanitize_text_field( (string) ( $record['executor_version'] ?? CommandExecutor::VERSION ) ),
				'created_at'       => (int) ( $record['created_at'] ?? time() ),
				'accepted_at'      => (int) ( $record['accepted_at'] ?? 0 ),
				'updated_at'       => (int) ( $record['updated_at'] ?? time() ),
				'started_at'       => (int) ( $record['started_at'] ?? 0 ),
				'completed_at'     => (int) ( $record['completed_at'] ?? 0 ),
			],
			[
				'%s',
				'%s',
				'%s',
				'%s',
				'%s',
				'%s',
				'%s',
				'%s',
				'%s',
				'%d',
				'%d',
				'%d',
				'%d',
				'%d',
			]
		);
		$persist_error = (string) $wpdb->last_error;
		if ( false === $wpdb->rows_affected || '' !== $persist_error ) {
			throw new \RuntimeException(
				'' !== $persist_error
					? sprintf( 'Command state persistence failed: %s', esc_html( sanitize_text_field( $persist_error ) ) )
					: 'Command state persistence failed.'
			);
		}

		if ( $prune ) {
			self::prune_records();
		}
	}

	/**
	 * Reconcile stale local command state before returning it to Console.
	 *
	 * @param array<string,mixed> $record Command row.
	 * @return array<string,mixed>
	 */
	private static function refresh_record_state( array $record, bool $lock_held = false, bool $prune = false ): array {
		$status = strtolower( (string) ( $record['status'] ?? '' ) );
		if ( 'accepted' === $status && self::is_accepted_record_stale( $record ) ) {
			return self::expire_stale_accepted_record( $record, $lock_held, $prune );
		}

		if ( 'running' !== $status ) {
			return $record;
		}

		if ( ! self::is_running_record_stale( $record ) ) {
			return $record;
		}

		$command_id = (string) ( $record['id'] ?? $record['command_id'] ?? '' );
		if ( '' === $command_id || ( ! $lock_held && ! self::acquire_worker_lock( $command_id ) ) ) {
			return $record;
		}

		try {
			$current = self::get( $command_id );
			if ( ! is_array( $current ) ) {
				return $record;
			}

			if ( strtolower( (string) ( $current['status'] ?? '' ) ) !== 'running' ) {
				return $current;
			}

			if ( ! self::is_running_record_stale( $current ) ) {
				return $current;
			}

			$now = time();
			self::update_record(
				$command_id,
				[
					'status'           => 'failed',
					'completed_at'     => $now,
					'updated_at'       => $now,
					'result'           => [
						'error' => 'SiteFort lost execution progress before the command finished.',
					],
					'error'            => 'command_timed_out',
					'error_code'       => 'EXEC_TIMEOUT',
					'executor_version' => CommandExecutor::VERSION,
				],
				$prune
			);

			$recovered = self::get( $command_id );
			return is_array( $recovered ) ? $recovered : $current;
		} catch ( \Throwable $e ) {
			sitefort_log(
				sprintf(
					'SiteFort CommandQueue: failed to recover stale running command %s: %s',
					sanitize_text_field( $command_id ),
					sanitize_text_field( $e->getMessage() )
				)
			);
			return $record;
		} finally {
			if ( ! $lock_held ) {
				self::release_worker_lock( $command_id );
			}
		}
	}

	private static function expire_stale_accepted_record( array $record, bool $lock_held = false, bool $prune = false ): array {
		$command_id = (string) ( $record['id'] ?? $record['command_id'] ?? '' );
		if ( '' === $command_id || ( ! $lock_held && ! self::acquire_worker_lock( $command_id ) ) ) {
			return $record;
		}

		try {
			$current = self::get( $command_id );
			if ( ! is_array( $current ) ) {
				return $record;
			}

			if ( strtolower( (string) ( $current['status'] ?? '' ) ) !== 'accepted' || ! self::is_accepted_record_stale( $current ) ) {
				return $current;
			}

			$now = time();
			self::update_record(
				$command_id,
				[
					'status'           => 'expired',
					'completed_at'     => $now,
					'updated_at'       => $now,
					'result'           => [
						'error' => 'SiteFort command worker did not start before the command expired.',
					],
					'error'            => 'command_expired',
					'error_code'       => 'EXEC_NOT_STARTED_TIMEOUT',
					'executor_version' => CommandExecutor::VERSION,
				],
				$prune
			);

			$expired = self::get( $command_id );
			return is_array( $expired ) ? $expired : $current;
		} catch ( \Throwable $e ) {
			sitefort_log(
				sprintf(
					'SiteFort CommandQueue: failed to expire stale accepted command %s: %s',
					sanitize_text_field( $command_id ),
					sanitize_text_field( $e->getMessage() )
				)
			);
			return $record;
		} finally {
			if ( ! $lock_held ) {
				self::release_worker_lock( $command_id );
			}
		}
	}

	private static function is_accepted_record_stale( array $record ): bool {
		$anchor = max(
			(int) ( $record['updated_at'] ?? 0 ),
			(int) ( $record['accepted_at'] ?? 0 ),
			(int) ( $record['created_at'] ?? 0 )
		);

		return $anchor > 0 && ( time() - $anchor ) >= self::ACCEPTED_STALE_SECONDS;
	}

	/**
	 * Whether a running command has stopped updating for too long.
	 *
	 * @param array<string,mixed> $record Command row.
	 * @return bool
	 */
	private static function is_running_record_stale( array $record ): bool {
		$anchor = max(
			(int) ( $record['updated_at'] ?? 0 ),
			(int) ( $record['started_at'] ?? 0 ),
			(int) ( $record['accepted_at'] ?? 0 ),
			(int) ( $record['created_at'] ?? 0 )
		);

		return $anchor > 0 && ( time() - $anchor ) >= self::RUNNING_STALE_SECONDS;
	}

	/**
	 * Persist the running transition before telling Console execution started.
	 *
	 * @param string $command_id Command UUID.
	 * @param bool   $notify     Whether to send the running ACK.
	 * @return bool
	 */
	private static function mark_running( string $command_id, bool $notify = true ): bool {
		$now = time();
		try {
			self::update_record(
				$command_id,
				[
					'status'           => 'running',
					'started_at'       => $now,
					'updated_at'       => $now,
					'executor_version' => CommandExecutor::VERSION,
					'error'            => '',
					'error_code'       => '',
					'result'           => [],
				],
				false
			);
		} catch ( \Throwable $e ) {
			sitefort_log(
				sprintf(
					'SiteFort CommandQueue: failed to persist running state for %s: %s',
					sanitize_text_field( $command_id ),
					sanitize_text_field( $e->getMessage() )
				)
			);
			self::finalize_failure(
				$command_id,
				'command_state_persist_failed',
				'EXEC_STATE_PERSIST_FAILED',
				[
					'error' => 'SiteFort could not durably store the running command state.',
				],
				$notify
			);
			return false;
		}

		$record = self::get( $command_id );
		if ( $notify && is_array( $record ) && self::should_send_ack( $record ) ) {
			CommandAckService::running( $command_id );
		}
		return true;
	}

	/**
	 * Persist a successful terminal state before sending success ACK.
	 *
	 * @param string              $command_id Command UUID.
	 * @param array<string,mixed> $result     Command result.
	 * @param bool                $notify     Whether to send the success ACK.
	 * @return bool
	 */
	private static function finalize_success( string $command_id, array $result, bool $notify = true ): bool {
		$now = time();
		try {
			self::update_record(
				$command_id,
				[
					'status'           => 'succeeded',
					'completed_at'     => $now,
					'updated_at'       => $now,
					'result'           => $result,
					'error'            => '',
					'error_code'       => '',
					'executor_version' => CommandExecutor::VERSION,
				],
				false
			);
		} catch ( \Throwable $e ) {
			sitefort_log(
				sprintf(
					'SiteFort CommandQueue: failed to persist success state for %s: %s',
					sanitize_text_field( $command_id ),
					sanitize_text_field( $e->getMessage() )
				)
			);
			return self::finalize_failure(
				$command_id,
				'command_state_persist_failed',
				'EXEC_STATE_PERSIST_FAILED',
				[
					'error' => 'SiteFort completed the command but could not durably store the final state.',
				],
				$notify
			);
		}

		$record = self::get( $command_id );
		if ( $notify && is_array( $record ) && self::should_send_ack( $record ) ) {
			CommandAckService::succeeded( $command_id, $result );
		}
		return true;
	}

	/**
	 * Persist a failed terminal state before sending failure ACK.
	 *
	 * @param string               $command_id Command UUID.
	 * @param string               $error      Error message.
	 * @param string|null          $error_code Structured error code.
	 * @param array<string,mixed>  $result     Failure details.
	 * @param bool                 $notify     Whether to send the failure ACK.
	 * @return bool
	 */
	private static function finalize_failure( string $command_id, string $error, ?string $error_code = null, array $result = [], bool $notify = true ): bool {
		$now = time();
		try {
			self::update_record(
				$command_id,
				[
					'status'           => 'failed',
					'completed_at'     => $now,
					'updated_at'       => $now,
					'result'           => $result,
					'error'            => sanitize_text_field( $error ),
					'error_code'       => sanitize_text_field( (string) ( $error_code ?? '' ) ),
					'executor_version' => CommandExecutor::VERSION,
				],
				false
			);
		} catch ( \Throwable $e ) {
			sitefort_log(
				sprintf(
					'SiteFort CommandQueue: failed to persist failure state for %s: %s',
					sanitize_text_field( $command_id ),
					sanitize_text_field( $e->getMessage() )
				)
			);
			return false;
		}

		$record = self::get( $command_id );
		if ( $notify && is_array( $record ) && self::should_send_ack( $record ) ) {
			CommandAckService::failed( $command_id, $error, $error_code, $result );
		}
		return true;
	}

	/**
	 * Ensure the table exists and legacy command state is migrated.
	 *
	 * @return void
	 */
	private static function ensure_storage(): void {
		if ( self::$storage_ready ) {
			return;
		}

		$schema_version = (string) get_option( self::SCHEMA_VERSION_OPTION, '' );
		if ( self::SCHEMA_VERSION !== $schema_version ) {
			self::create_or_upgrade_table();
			update_option( self::SCHEMA_VERSION_OPTION, self::SCHEMA_VERSION, true );
		}
		self::migrate_legacy_option();
		self::$storage_ready = true;
	}

	/**
	 * Migrate legacy shared option-map storage into the per-command table.
	 *
	 * @return void
	 */
	private static function migrate_legacy_option(): void {
		$legacy = get_option( self::LEGACY_OPTION_KEY, null );
		if ( ! is_array( $legacy ) ) {
			return;
		}

		foreach ( $legacy as $command_id => $record ) {
			if ( ! is_string( $command_id ) || ! is_array( $record ) ) {
				continue;
			}

			self::save_record( self::normalize_legacy_record( $command_id, $record ) );
		}

		delete_option( self::LEGACY_OPTION_KEY );
	}

	/**
	 * Normalize a row loaded from the table.
	 *
	 * @param array<string,mixed> $row Table row.
	 * @return array<string,mixed>
	 */
	private static function normalize_loaded_record( array $row ): array {
		return [
			'id'               => (string) ( $row['command_id'] ?? '' ),
			'command'          => sanitize_text_field( (string) ( $row['command'] ?? '' ) ),
			'params'           => self::decode_json_column( $row['params'] ?? null ),
			'status'           => sanitize_text_field( (string) ( $row['status'] ?? 'accepted' ) ),
			'result'           => self::decode_json_column( $row['result'] ?? null ),
			'error'            => sanitize_text_field( (string) ( $row['error'] ?? '' ) ),
			'error_code'       => sanitize_text_field( (string) ( $row['error_code'] ?? '' ) ),
			'ack_mode'         => self::normalize_ack_mode( (string) ( $row['ack_mode'] ?? self::ACK_MODE_TERMINAL ) ),
			'executor_version' => sanitize_text_field( (string) ( $row['executor_version'] ?? CommandExecutor::VERSION ) ),
			'created_at'       => (int) ( $row['created_at'] ?? 0 ),
			'accepted_at'      => (int) ( $row['accepted_at'] ?? 0 ),
			'updated_at'       => (int) ( $row['updated_at'] ?? 0 ),
			'started_at'       => (int) ( $row['started_at'] ?? 0 ),
			'completed_at'     => (int) ( $row['completed_at'] ?? 0 ),
		];
	}

	/**
	 * Normalize one legacy option record for table storage.
	 *
	 * @param string              $command_id Command UUID.
	 * @param array<string,mixed> $record     Legacy record.
	 * @return array<string,mixed>
	 */
	private static function normalize_legacy_record( string $command_id, array $record ): array {
		return [
			'id'               => $command_id,
			'command'          => sanitize_text_field( (string) ( $record['command'] ?? '' ) ),
			'params'           => isset( $record['params'] ) && is_array( $record['params'] ) ? $record['params'] : [],
			'status'           => sanitize_text_field( (string) ( $record['status'] ?? 'accepted' ) ),
			'result'           => isset( $record['result'] ) && is_array( $record['result'] ) ? $record['result'] : [],
			'error'            => sanitize_text_field( (string) ( $record['error'] ?? '' ) ),
			'error_code'       => sanitize_text_field( (string) ( $record['error_code'] ?? '' ) ),
			'ack_mode'         => self::normalize_ack_mode( (string) ( $record['ack_mode'] ?? self::ACK_MODE_TERMINAL ) ),
			'executor_version' => sanitize_text_field( (string) ( $record['executor_version'] ?? CommandExecutor::VERSION ) ),
			'created_at'       => (int) ( $record['created_at'] ?? time() ),
			'accepted_at'      => (int) ( $record['accepted_at'] ?? 0 ),
			'updated_at'       => (int) ( $record['updated_at'] ?? $record['created_at'] ?? time() ),
			'started_at'       => (int) ( $record['started_at'] ?? 0 ),
			'completed_at'     => (int) ( $record['completed_at'] ?? 0 ),
		];
	}

	/**
	 * Normalize command acknowledgement policy to a supported value.
	 *
	 * @param string $ack_mode Raw acknowledgement mode.
	 * @return string
	 */
	private static function normalize_ack_mode( string $ack_mode ): string {
		$ack_mode = strtolower( sanitize_text_field( $ack_mode ) );
		return self::ACK_MODE_ACCEPTED === $ack_mode ? self::ACK_MODE_ACCEPTED : self::ACK_MODE_TERMINAL;
	}

	/**
	 * Whether the current command should send runtime ACKs back to Console.
	 *
	 * @param array<string,mixed> $record Command row.
	 * @return bool
	 */
	private static function should_send_ack( array $record ): bool {
		return self::normalize_ack_mode( (string) ( $record['ack_mode'] ?? self::ACK_MODE_TERMINAL ) ) === self::ACK_MODE_TERMINAL;
	}

	/**
	 * Decode one JSON column into an array.
	 *
	 * @param mixed $value Stored JSON.
	 * @return array<string,mixed>
	 */
	private static function decode_json_column( $value ): array {
		if ( is_array( $value ) ) {
			return $value;
		}

		if ( ! is_string( $value ) || '' === $value ) {
			return [];
		}

		$decoded = json_decode( $value, true );
		return is_array( $decoded ) ? $decoded : [];
	}

	/**
	 * Normalize command result payloads before local persistence.
	 *
	 * @param string              $command Command name.
	 * @param array<string,mixed> $result  Raw command result.
	 * @return array<string,mixed>
	 */
	private static function normalize_result_for_storage( string $command, array $result ): array {
		$command = strtolower( $command );
		$encoded = wp_json_encode( $result );
		if ( false !== $encoded && strlen( $encoded ) <= self::MAX_RESULT_BYTES ) {
			return $result;
		}

		if ( 'settings.export' === $command ) {
			$data = isset( $result['data'] ) && is_array( $result['data'] ) ? $result['data'] : [];
			return [
				'ok'            => ! empty( $result['ok'] ),
				'exported_at'   => (int) ( $result['exported_at'] ?? 0 ),
				'sections'      => array_values( array_map( 'strval', array_keys( $data ) ) ),
				'section_count' => count( $data ),
				'truncated'     => true,
			];
		}

		if ( 'audit.query' === $command ) {
			$entries = isset( $result['entries'] ) && is_array( $result['entries'] ) ? $result['entries'] : [];
			return [
				'ok'        => ! empty( $result['ok'] ),
				'count'     => (int) ( $result['count'] ?? count( $entries ) ),
				'truncated' => ! empty( $entries ),
			];
		}

		$summary = [
			'ok'        => ! empty( $result['ok'] ),
			'truncated' => true,
			'keys'      => array_values(
				array_slice(
					array_map( 'strval', array_keys( $result ) ),
					0,
					10
				)
			),
		];

		foreach ( [ 'scan_id', 'scan_type', 'console_enabled', 'count', 'message' ] as $key ) {
			if ( isset( $result[ $key ] ) && ( is_scalar( $result[ $key ] ) || null === $result[ $key ] ) ) {
				$summary[ $key ] = $result[ $key ];
			}
		}

		return $summary;
	}

	/**
	 * Remove stale records and keep the table bounded.
	 *
	 * @return void
	 */
	private static function prune_records( bool $reconcile_stale = true ): void {
		global $wpdb;

		$table = self::table_identifier( self::table_name() );
		if ( $reconcile_stale ) {
			self::reconcile_stale_records( $table );
		}

		$cutoff = time() - self::MAX_AGE_SECONDS;
		$terminal_statuses = self::quoted_terminal_statuses();

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query(
			$wpdb->prepare(
				"DELETE FROM {$table} WHERE status IN ({$terminal_statuses}) AND updated_at > 0 AND updated_at < %d",
				$cutoff
			)
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$count = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$table}" );
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter

		if ( $count <= self::MAX_COMMANDS ) {
			return;
		}

		$excess = $count - self::MAX_COMMANDS;
		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$terminal_count = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$table} WHERE status IN ({$terminal_statuses})" );
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		if ( $terminal_count <= 0 ) {
			return;
		}

		$delete_count = min( $excess, $terminal_count );
		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query(
			$wpdb->prepare(
				"DELETE FROM {$table} WHERE status IN ({$terminal_statuses}) ORDER BY updated_at ASC LIMIT %d",
				$delete_count
			)
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
	}

	private static function reconcile_stale_records( string $table ): void {
		global $wpdb;

		$now = time();
		$accepted_cutoff = $now - self::ACCEPTED_STALE_SECONDS;
		$running_cutoff = $now - self::RUNNING_STALE_SECONDS;

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT command_id, command, params, status, result, error, error_code, ack_mode, executor_version, created_at, accepted_at, updated_at, started_at, completed_at
				 FROM {$table}
				 WHERE (
					status = %s
					AND GREATEST(updated_at, accepted_at, created_at) > 0
					AND GREATEST(updated_at, accepted_at, created_at) < %d
				 )
				 OR (
					status = %s
					AND GREATEST(updated_at, started_at, accepted_at, created_at) > 0
					AND GREATEST(updated_at, started_at, accepted_at, created_at) < %d
				 )
				 ORDER BY updated_at ASC
				 LIMIT %d",
				'accepted',
				$accepted_cutoff,
				'running',
				$running_cutoff,
				self::STALE_RECONCILE_BATCH
			),
			ARRAY_A
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter

		if ( ! is_array( $rows ) || empty( $rows ) ) {
			return;
		}

		foreach ( $rows as $row ) {
			if ( is_array( $row ) ) {
				self::refresh_record_state( self::normalize_loaded_record( $row ), false, false );
			}
		}
	}

	/**
	 * Quote the terminal-status list for direct SQL interpolation.
	 *
	 * @return string
	 */
	private static function quoted_terminal_statuses(): string {
		$statuses = array_map(
			static function ( string $status ): string {
				return "'" . esc_sql( $status ) . "'";
			},
			self::TERMINAL_STATUSES
		);

		return implode( ',', $statuses );
	}

	/**
	 * Acquire a per-command worker lock so overlapping workers cannot execute
	 * the same accepted command twice.
	 *
	 * @param string $command_id Command UUID.
	 * @return bool
	 */
	private static function acquire_worker_lock( string $command_id ): bool {
		global $wpdb;
		$lock_name = 'sitefort_cmd_worker_' . substr( hash( 'sha256', $command_id ), 0, 32 );
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$result = $wpdb->get_var( $wpdb->prepare( 'SELECT GET_LOCK(%s, %d)', $lock_name, self::WORKER_LOCK_TIMEOUT ) );
		return '1' === $result || 1 === $result;
	}

	/**
	 * Release the per-command worker lock.
	 *
	 * @param string $command_id Command UUID.
	 * @return void
	 */
	private static function release_worker_lock( string $command_id ): void {
		global $wpdb;
		$lock_name = 'sitefort_cmd_worker_' . substr( hash( 'sha256', $command_id ), 0, 32 );
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
		$wpdb->query( $wpdb->prepare( 'SELECT RELEASE_LOCK(%s)', $lock_name ) );
	}

	/**
	 * Fully qualified table name.
	 *
	 * @return string
	 */
	private static function table_name(): string {
		global $wpdb;
		return $wpdb->prefix . self::TABLE_NAME;
	}

	/**
	 * Escape a plugin-owned table identifier for direct interpolation.
	 *
	 * @param string $table Table name.
	 * @return string
	 */
	private static function table_identifier( string $table ): string {
		if ( $table === '' || strlen( $table ) > 64 || ! preg_match( '/^[A-Za-z0-9_]+$/', $table ) ) {
			throw new \InvalidArgumentException( 'Invalid table identifier.' );
		}

		return '`' . str_replace( '`', '``', $table ) . '`';
	}
}
