<?php
/**
 * Server-snapshotted bulk actions. One saved actor, explicit policy, one worker, no destructive bulk cleanup.
 *
 * @package AffStackDevImageManager
 */

namespace AffStackDev\ImageManager;

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

/**
 * Bulk actions on chosen images (optimize, retry, use originals, exclude and more): the selection frozen once, then
 * worked one image per step under the worker's lease.
 */
final class Batch {
	// The keys of actions(), untranslated: the contract announced on plugins_loaded lists them (translations load later).
	public const ACTIONS = array( 'optimize', 'retry', 'original', 'optimized', 'exclude', 'include', 'check' );
	/**
	 * The bulk actions and their names.
	 *
	 * @return array Action => translated name.
	 */
	public static function actions(): array {
        // phpcs:ignore WordPressVIPMinimum.Performance.WPQueryParams.PostNotIn_exclude -- This is the label of the allowed bulk action named exclude, not a WP_Query exclusion parameter or SQL query.
		return array(
			'optimize'  => __( 'Optimize / recompress', 'affstackdev-image-manager' ),
			'retry'     => __( 'Retry using each image’s saved policy', 'affstackdev-image-manager' ),
			'original'  => __( 'Use original images', 'affstackdev-image-manager' ),
			'optimized' => __( 'Allow optimized delivery', 'affstackdev-image-manager' ),
			// phpcs:ignore WordPressVIPMinimum.Performance.WPQueryParams.PostNotIn_exclude -- This is the label of the allowed bulk action named exclude, not a WP_Query exclusion parameter or SQL query.
			'exclude'   => __( 'Exclude from optimization', 'affstackdev-image-manager' ),
			'include'   => __( 'Include again', 'affstackdev-image-manager' ),
			'check'     => __( 'Check public-page delivery', 'affstackdev-image-manager' ),
		);
	}
	/**
	 * The values of the sources condition for an Images filter and search. Every query writes the condition out in full:
	 * state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s ).
	 *
	 * @param string $filter The filter (ready, queued and the others).
	 * @param string $search Text to find in the sources' saved details, such as their addresses (the first 100
	 *                       characters).
	 * @return array The seven values for those placeholders, in order.
	 */
	public static function where_args( string $filter = '', string $search = '' ): array {
		global $wpdb;
		$filters = array(
			'ready'       => array( 'ready' ),
			'queued'      => array( 'queued' ),
			'waiting'     => array( 'queued', 'discovered' ),
			'no_saving'   => array( 'no_saving' ),
			'held'        => array( 'held' ),
			'unsupported' => array( 'unsupported', 'unmapped' ),
			'kept'        => array( 'no_saving', 'unsupported', 'held', 'unmapped' ),
			'failed'      => array( 'failed' ),
			'excluded'    => array( 'excluded' ),
		);
		$states  = $filters[ $filter ] ?? array();
		// The state list has four places; a shorter list repeats its last state, which leaves the condition the same.
		$four = $states ? array_pad( $states, 4, end( $states ) ) : array( '', '', '', '' );
		$like = $search ? '%' . $wpdb->esc_like( substr( $search, 0, 100 ) ) . '%' : '';
		return array_merge( array( $states ? 1 : 0 ), $four, array( $search ? 1 : 0, $like ) );
	}
	/**
	 * How many sources an Images filter and search match.
	 *
	 * @param string $filter The filter.
	 * @param string $search The search.
	 * @return int The count.
	 */
	public static function count_matching( string $filter, string $search ): int {
		global $wpdb;
		list( $on, $state1, $state2, $state3, $state4, $find, $like ) = self::where_args( $filter, $search );
		// Read current plugin source rows: writes happen in independent workers, and cached rows would misreport them.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- see the note above.
		return (int) $wpdb->get_var(
			$wpdb->prepare(
				"SELECT COUNT(*) FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s )",
				Repository::table( 'sources' ),
				$on,
				$state1,
				$state2,
				$state3,
				$state4,
				$find,
				$like
			)
		);
	}
	/**
	 * One page (20) of the sources an Images filter and search match, every column, in the results' order.
	 *
	 * @param string $filter The filter.
	 * @param string $search The search.
	 * @param int    $offset The page's first row.
	 * @return array The rows.
	 */
	public static function page_matching( string $filter, string $search, int $offset ): array {
		global $wpdb;
		list( $on, $state1, $state2, $state3, $state4, $find, $like ) = self::where_args( $filter, $search );
		// Read current plugin source rows: writes happen in independent workers, and cached rows would misreport them.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- see the note above.
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s ) ORDER BY priority ASC,id ASC LIMIT 20 OFFSET %d",
				Repository::table( 'sources' ),
				$on,
				$state1,
				$state2,
				$state3,
				$state4,
				$find,
				$like,
				$offset
			),
			ARRAY_A
		);
		return is_array( $rows ) ? $rows : array();
	}
	/**
	 * The sources a reviewed bulk action selects by filter and search: all of them, or one page of 20 (the columns
	 * Reviewed_Owner::selection() reads, in the results' order).
	 *
	 * @param string   $filter The filter.
	 * @param string   $search The search.
	 * @param int|null $offset The page's first row, or null for all.
	 * @return array|null The rows, or null when the database refused.
	 */
	public static function select_matching( string $filter, string $search, ?int $offset ): ?array {
		global $wpdb;
		list( $on, $state1, $state2, $state3, $state4, $find, $like ) = self::where_args( $filter, $search );
		$table = Repository::table( 'sources' );
		if ( null === $offset ) {
			// Read current plugin source rows: writes happen in independent workers, and cached rows would misreport them.
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- see the note above.
			$rows = $wpdb->get_results(
				$wpdb->prepare(
					"SELECT id,revision,attachment_id,excluded,state,manifest,details FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s ) ORDER BY priority ASC,id ASC",
					$table,
					$on,
					$state1,
					$state2,
					$state3,
					$state4,
					$find,
					$like
				),
				ARRAY_A
			);
		} else {
			// Read current plugin source rows: writes happen in independent workers, and cached rows would misreport them.
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- see the note above.
			$rows = $wpdb->get_results(
				$wpdb->prepare(
					"SELECT id,revision,attachment_id,excluded,state,manifest,details FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s ) ORDER BY priority ASC,id ASC LIMIT 20 OFFSET %d",
					$table,
					$on,
					$state1,
					$state2,
					$state3,
					$state4,
					$find,
					$like,
					$offset
				),
				ARRAY_A
			);
		}
		return is_array( $rows ) && ! $wpdb->last_error ? $rows : null;
	}
	/**
	 * The IDs of the sources whose saved details contain a text, by ID, one page (the AI's sources listing).
	 *
	 * @param string $search The text.
	 * @param int    $limit  How many.
	 * @param int    $offset The first row.
	 * @return array The rows (id only).
	 */
	public static function matching_ids( string $search, int $limit, int $offset ): array {
		global $wpdb;
		list( $on, $state1, $state2, $state3, $state4, $find, $like ) = self::where_args( '', $search );
		// Read current plugin source rows: writes happen in independent workers, and cached rows would misreport them.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- see the note above.
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s ) ORDER BY id LIMIT %d OFFSET %d",
				Repository::table( 'sources' ),
				$on,
				$state1,
				$state2,
				$state3,
				$state4,
				$find,
				$like,
				$limit,
				$offset
			),
			ARRAY_A
		);
		return is_array( $rows ) ? $rows : array();
	}
	/**
	 * A source the user may still change: they manage options and upload files, and may edit its attachment.
	 *
	 * @param int $id    The source row.
	 * @param int $actor The user who started the batch.
	 * @return array The source row.
	 * @throws \RuntimeException With a reason code as its message when the step cannot be done.
	 */
	public static function authorized_row( int $id, int $actor ): array {
		if ( ! $actor || ! user_can( $actor, 'manage_options' ) || ! user_can( $actor, 'upload_files' ) ) {
			throw new \RuntimeException( 'batch_access_changed' );
		}
		$r = Repository::row( $id );
		if ( ! $r || ( ! $r['attachment_id'] && ! Theme::is_source( $r ) ) || ( $r['attachment_id'] && ! user_can( $actor, 'edit_post', (int) $r['attachment_id'] ) ) ) {
			throw new \RuntimeException( 'batch_access_changed' );
		}
		return $r;
	}
	/**
	 * Request fields are allowlisted. Selection is frozen by a single INSERT...SELECT, not a shifting result offset.
	 *
	 * @param array $data The bulk form's fields.
	 * @param bool  $scan True when an optimization run starts the batch.
	 * @return int How many images were selected.
	 * @throws \RuntimeException With a reason code as its message when the step cannot be done.
	 */
	public static function start( array $data, bool $scan = false ): int {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			global $wpdb;
			$action = $scan ? 'optimize' : ( is_string( $data['bulk'] ?? null ) ? $data['bulk'] : '' );
			if ( ! in_array( $action, self::ACTIONS, true ) ) {
				throw new \RuntimeException( 'batch_invalid_action' );
			}
			$scope = $scan ? 'matching' : ( is_string( $data['scope'] ?? null ) ? $data['scope'] : 'selected' );
			if ( ! in_array( $scope, array( 'selected', 'page', 'matching' ), true ) ) {
				throw new \RuntimeException( 'batch_invalid_selection' );
			}
			$recipe = Policy::clean( is_array( $data['compression'] ?? null ) ? $data['compression'] : Policy::current() );
			$intent = is_string( $data['bulk_intent'] ?? null ) ? $data['bulk_intent'] : 'apply';
			if ( ! in_array( $intent, array( 'apply', 'review' ), true ) ) {
				throw new \RuntimeException( 'batch_invalid_action' );
			}
			$sizing = is_string( $data['sizing'] ?? null ) ? $data['sizing'] : 'existing';
			if ( ! in_array( $sizing, array( 'existing', 'balanced', 'fit' ), true ) ) {
				throw new \RuntimeException( 'batch_invalid_sizing' );
			}
			if ( 'optimize' !== $action ) {
				$sizing = 'existing';
			}
			if ( 'optimize' === $action && 'lossless' !== $recipe['mode'] && empty( $data['compression_confirm'] ) ) {
				throw new \RuntimeException( 'compression_confirmation_required' );
			}
			if ( 'optimize' === $action && 'existing' !== $sizing && empty( $data['resize_confirm'] ) ) {
				throw new \RuntimeException( 'batch_resize_confirmation' );
			}
			if ( empty( $data['bulk_confirm'] ) ) {
				throw new \RuntimeException( 'batch_confirmation_required' );
			}
			$actor = get_current_user_id();
			if ( ! $actor || ! current_user_can( 'manage_options' ) || ! current_user_can( 'upload_files' ) ) {
				throw new \RuntimeException( 'image_access_denied' );
			}
			$s = Settings::fresh();
			if ( in_array( $action, array( 'optimize', 'retry' ), true ) && ! $s['public_policy'] && empty( $data['public_policy'] ) ) {
				throw new \RuntimeException( 'preview_public_required' );
			}
			$token = Repository::acquire();
			if ( ! $token ) {
				throw new \RuntimeException( 'worker_busy' );
			}
			try {
				$job = Workflow::job();
				if ( $job['active'] || Scanner::pending() ) {
					throw new \RuntimeException( 'An existing run is saved. Resume it or stop it before starting another.' );
				}
				$id     = bin2hex( random_bytes( 16 ) );
				$filter = is_string( $data['im_filter'] ?? null ) ? sanitize_key( $data['im_filter'] ) : '';
				$search = is_string( $data['im_search'] ?? null ) ? sanitize_text_field( $data['im_search'] ) : '';
				$from   = array(
					'filter' => $filter,
					'search' => $search,
				);
				if ( 'selected' === $scope ) {
					$raw = is_array( $data['ids'] ?? null ) ? $data['ids'] : array();
					if ( count( $raw ) > 1000 ) {
						throw new \RuntimeException( 'batch_selection_limit' );
					}
					$ids = array_values( array_unique( array_filter( array_map( 'absint', array_filter( $raw, 'is_scalar' ) ) ) ) );
					if ( ! $ids ) {
						throw new \RuntimeException( 'Choose at least one image.' );
					}
					$from = array( 'ids' => $ids );
				}
				$page   = max( 1, min( 100000, is_scalar( $data['im_page'] ?? null ) ? absint( $data['im_page'] ) : 1 ) );
				$target = is_string( $data['target'] ?? null ) ? $data['target'] : 'all';
				if ( ! in_array( $target, array( 'all', 'missing' ), true ) ) {
					throw new \RuntimeException( 'batch_invalid_selection' );
				}
				$job = array(
					'id'           => $id,
					'active'       => true,
					'mode'         => 'bulk',
					'actor'        => $actor,
					'action'       => $action,
					'recipe'       => $recipe,
					'intent'       => $intent,
					'verify_after' => ! empty( $data['verify_after_apply'] ),
					'sizing'       => $sizing,
					'started'      => time(),
					'finished'     => 0,
					'cursor'       => 0,
					'cancelled'    => false,
					'scan_first'   => $scan,
					'snapshotted'  => ! $scan,
					'target'       => $target,
				);
				if ( $scan ) {
					Scanner::start( (bool) $s['catalog'], (bool) $s['comments'], $token );
					$count = 0;
				} else {
					$count = self::snapshot( $id, $from, 'page' === $scope ? ( $page - 1 ) * 20 : null );
					if ( ! $count ) {
						throw new \RuntimeException( 'Choose at least one image.' );
					}
				}
				// Keep only this run's selection records. Original media/history are untouched.
				// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching would make selection/progress stale.
				Reviewed_Store::database_query( $wpdb->prepare( 'DELETE FROM %i WHERE batch_id<>%s', Repository::table( 'batch' ), $id ) );
				if ( in_array( $action, array( 'optimize', 'retry' ), true ) ) {
					$s['public_policy'] = true;
				}
				$values = array(
					'paused'        => false,
					'pause_reason'  => '',
					'public_policy' => (bool) $s['public_policy'],
					'generation'    => (int) $s['generation'] + 1,
				);
				if ( $scan ) {
					$values['delivery'] = ! empty( $data['delivery'] );
				}
				Settings::patch( $values );
				Reviewed_Store::update_option( 'afsd_im_failures', 0, false );
				Reviewed_Store::update_option( Workflow::OPTION, $job, false );
				if ( $job['snapshotted'] ) {
					Progress::record( 'selection_ready' );
				}Activity::add( 'started' );
			} finally {
				Repository::release( $token );
			}
			Scheduler::check();
			Scheduler::needed();
			return $count;

		} finally {
			Reviewed_Store::leave( $im_reviewed_owner_lock );
		}
	}
	/**
	 * Freeze the selection: copy the matching sources into the batch table in one insert.
	 *
	 * @param string   $id     The batch.
	 * @param array    $from   array( 'ids' => chosen IDs ), array( 'filter' => ..., 'search' => ... ) (an Images filter
	 *                         and search) or array( 'missing' => bool ) (every image not excluded, or only those
	 *                         without copies).
	 * @param int|null $offset A page offset for a filter and search, or null for every match.
	 * @return int How many sources were added.
	 * @throws \RuntimeException With a reason code as its message when the step cannot be done.
	 */
	private static function snapshot( string $id, array $from, ?int $offset = null ): int {
		global $wpdb;
		$batch   = Repository::table( 'batch' );
		$sources = Repository::table( 'sources' );
		if ( isset( $from['ids'] ) ) {
			$ids = array_map( 'intval', $from['ids'] );
			$sql = $wpdb->prepare(
				"INSERT IGNORE INTO %i (batch_id,source_id,state,draft_id,reason) SELECT %s,id,'waiting','','' FROM %i WHERE id IN (" . implode( ',', array_fill( 0, count( $ids ), '%d' ) ) . ") AND state<>'grouped'",
				...array_merge( array( $batch, $id, $sources ), $ids )
			);
		} elseif ( isset( $from['missing'] ) ) {
			$sql = $wpdb->prepare(
				"INSERT IGNORE INTO %i (batch_id,source_id,state,draft_id,reason) SELECT %s,id,'waiting','','' FROM %i WHERE state<>'grouped' AND excluded=0 AND ( %d = 0 OR copy_count=0 )",
				$batch,
				$id,
				$sources,
				$from['missing'] ? 1 : 0
			);
		} elseif ( null === $offset ) {
			list( $on, $state1, $state2, $state3, $state4, $find, $like ) = self::where_args( (string) $from['filter'], (string) $from['search'] );
			$sql = $wpdb->prepare(
				"INSERT IGNORE INTO %i (batch_id,source_id,state,draft_id,reason) SELECT %s,id,'waiting','','' FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s )",
				$batch,
				$id,
				$sources,
				$on,
				$state1,
				$state2,
				$state3,
				$state4,
				$find,
				$like
			);
		} else {
			list( $on, $state1, $state2, $state3, $state4, $find, $like ) = self::where_args( (string) $from['filter'], (string) $from['search'] );
			$sql = $wpdb->prepare(
				"INSERT IGNORE INTO %i (batch_id,source_id,state,draft_id,reason) SELECT %s,id,'waiting','','' FROM %i WHERE state<>'grouped' AND ( %d = 0 OR state IN ( %s, %s, %s, %s ) ) AND ( %d = 0 OR details LIKE %s ) ORDER BY priority ASC,id ASC LIMIT 20 OFFSET %d",
				$batch,
				$id,
				$sources,
				$on,
				$state1,
				$state2,
				$state3,
				$state4,
				$find,
				$like,
				$offset
			);
		}
		// Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching
		// would make selection/progress stale.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- see the note above.
		$n = Reviewed_Store::database_query( $sql );
		if ( false === $n ) {
			throw new \RuntimeException( 'database_write_failed' );
		}return (int) $n;
	}
	/**
	 * The counts of a batch: total, done and by state.
	 *
	 * @param array|null $job The run (the current one when null).
	 * @return array The counts.
	 */
	public static function summary( ?array $job = null ): array {
		global $wpdb;
		$job = $job ?? Workflow::job();
		$out = array(
			'total'    => 0,
			'done'     => 0,
			'complete' => 0,
			'review'   => 0,
			'skipped'  => 0,
			'failed'   => 0,
			'waiting'  => 0,
			'working'  => 0,
			'checking' => 0,
		);
		if ( 'bulk' !== $job['mode'] ) {
			return $out;
		}
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching would make selection/progress stale.
		$rows = $wpdb->get_results( $wpdb->prepare( 'SELECT state,COUNT(*) AS n FROM %i WHERE batch_id=%s GROUP BY state', Repository::table( 'batch' ), $job['id'] ), ARRAY_A );
		foreach ( $rows as  $r ) {
			$out[ $r['state'] ] = (int) $r['n'];
			$out['total']      += (int) $r['n'];
		}
		$out['done'] = $out['complete'] + $out['review'] + $out['skipped'] + $out['failed'];
		return $out;
	}
	/**
	 * Mark one image of the batch done.
	 *
	 * @param array  $item   The batch row.
	 * @param string $state  Its final state.
	 * @param string $reason Why, when it was not completed.
	 */
	private static function finish( array $item, string $state, string $reason = '' ): void {
		global $wpdb;
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching would make selection/progress stale.
		$ok = Reviewed_Store::database_update(
			Repository::table( 'batch' ),
			array(
				'state'  => $state,
				'reason' => substr( $reason, 0, 190 ),
			),
			array(
				'batch_id'  => $item['batch_id'],
				'source_id' => $item['source_id'],
			)
		);
		if ( false !== $ok ) {
			Progress::record( $state, (int) $item['source_id'] );
		}
	}
	/**
	 * Called only under Engine's lease. Start at most one image comparison per turn.
	 *
	 * @param string $token The worker's lease.
	 * @throws \RuntimeException Only inside this function, where it is caught; callers get no exception.
	 */
	public static function pump( string $token ): void {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			global $wpdb;
			if ( ! $token || ! Repository::owns( $token ) ) {
				return;
			}
			$job = Workflow::job();
			if ( ! $job['active'] || $job['cancelled'] || Settings::fresh()['paused'] ) {
				return;
			}
			if ( ! empty( $job['scan_first'] ) && ! $job['snapshotted'] ) {
				$pages = Repository::counts( 'pages' );
				if ( ! Scanner::state()['active'] && ! empty( $pages['waiting'] ) ) {
					Scanner::continue_scan();
				}
				if ( Scanner::pending() || ! empty( $pages['waiting'] ) ) {
					return;
				}
				self::snapshot( $job['id'], array( 'missing' => 'missing' === $job['target'] ) );
				$job['snapshotted'] = true;
				Reviewed_Store::update_option( Workflow::OPTION, $job, false );
				Progress::record( 'selection_ready' );
			}
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching would make selection/progress stale.
			$items = $wpdb->get_results( $wpdb->prepare( "SELECT * FROM %i WHERE batch_id=%s AND state IN ('waiting','working','checking') ORDER BY (state='checking') DESC,(state='working') DESC,source_id ASC LIMIT 4", Repository::table( 'batch' ), $job['id'] ), ARRAY_A );
			foreach ( $items as  $item ) {
				try {
					$r = self::authorized_row( (int) $item['source_id'], (int) $job['actor'] );
					if ( 'checking' === $item['state'] ) {
						try {
							Verification::check_row( $r );
						} catch ( \Throwable $e ) {
							$d                   = $r['details'];
							$d['delivery_check'] = array(
								'signature' => Verification::signature( $r ),
								'time'      => time(),
								'status'    => 'unavailable',
								'kind'      => 'public_html',
								'matches'   => 0,
							);
							Repository::update( (int) $r['id'], array( 'details' => $d ), (int) $r['revision'] );
						}
						self::finish( $item, 'complete', 'Files approved. Public-page result is shown separately.' );
						return;
					}
					if ( 'working' === $item['state'] ) {
						if ( 'retry' === $job['action'] ) {
							if ( $r['requested'] ) {
								return;
							}self::finish( $item, 'failed' === $r['state'] ? 'failed' : ( in_array( $r['state'], array( 'held', 'unsupported' ), true ) ? 'skipped' : 'complete' ), $r['reason'] );
							continue;
						}

						if ( ( $r['details']['last_batch'] ?? '' ) === $job['id'] ) {
							self::finish( $item, ! empty( $job['verify_after'] ) ? 'checking' : 'complete', 'Approved optimized files.' );
							return;
						}
						if ( $r['requested'] ) {
							return;
						}
						$d = $r['manifest']['draft'] ?? array();
						if ( ( $d['batch_id'] ?? '' ) !== $job['id'] || ( $d['id'] ?? '' ) !== $item['draft_id'] ) {
							self::finish( $item, 'skipped', 'Image changed or comparison was handled separately.' );
							continue;
						}
						if ( empty( $d['complete'] ) ) {
							self::finish( $item, 'failed', 'Processing did not complete; review the image.' );
							continue;
						}
						if ( empty( $d['variants'] ) && empty( $d['responsive'] ) && empty( $d['watermarked'] ) ) {
							$m = $r['manifest'];
							unset( $m['draft'] );
							if ( empty( $m['variants'] ) && empty( $m['responsive'] ) && empty( $m['watermarked'] ) ) {
								// Preserve reasons for every tested size without changing its compression default.
								$m['results']            = $d['results'] ?? array();
								$m['responsive_results'] = $d['responsive_results'] ?? array();
							}
							$details                      = $r['details'];
							$details['last_batch_result'] = 'No qualifying alternative. Existing files stay in use.';
							Repository::update(
								(int) $r['id'],
								array(
									'manifest' => $m,
									'details'  => $details,
									'state'    => ! empty( $m['variants'] ) || ! empty( $m['responsive'] ) || ! empty( $m['watermarked'] ) ? 'ready' : ( $d['state'] ?? 'no_saving' ),
									'reason'   => $d['reason'] ?? 'No qualifying alternative.',
								),
								(int) $r['revision']
							);
							self::finish( $item, 'skipped', 'No qualifying alternative. Existing files stay in use.' );
							continue;
						}
						if ( 'review' === $job['intent'] ) {
							self::finish( $item, 'review', 'Comparison ready; not applied.' );
							continue;
						}
						Preview::apply( (int) $r['id'], $d['id'], true, $token, (int) $job['actor'] );
						self::finish( $item, ! empty( $job['verify_after'] ) ? 'checking' : 'complete', 'Optimized files approved. Existing delivery choices retained.' );
						return;
					}
					$action = $job['action'];
					if ( 'optimize' === $action ) {
						if ( $r['excluded'] ) {
							self::finish( $item, 'skipped', 'Excluded by you.' );
							continue;
						}
						if ( ! Settings::fresh()['public_policy'] || ! Sources::eligible( $r ) ) {
							self::finish( $item, 'skipped', 'No supported public image association.' );
							continue;
						}
						if ( $r['requested'] || ! empty( $r['manifest']['draft'] ) ) {
							self::finish( $item, 'skipped', 'Existing work or comparison retained; finish or discard it first.' );
							continue;
						}
						$guard = Engine::publication_lock();
						try {
							$m                        = $r['manifest'];
							$draft                    = bin2hex( random_bytes( 16 ) );
							$m['preview_prior_state'] = $r['state'];
							$m['draft']               = array(
								'id'        => $draft,
								'requested' => true,
								'complete'  => false,
								'recipe'    => $job['recipe'],
								'sizing'    => $job['sizing'],
								'created'   => time(),
								'batch_id'  => $job['id'],
							);
							// Persist the intent before the source request; recovery can reconcile this exact draft.
							// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching would make selection/progress stale.
							Reviewed_Store::database_update(
								Repository::table( 'batch' ),
								array(
									'state'    => 'working',
									'draft_id' => $draft,
								),
								array(
									'batch_id'  => $job['id'],
									'source_id' => $r['id'],
								)
							);
							if ( ! Repository::update( (int) $r['id'], array( 'manifest' => $m ), (int) $r['revision'] ) ) {
								throw new \RuntimeException( 'stale_job' );
							}
							Repository::request( (int) $r['id'] );
						} finally {
							Engine::publication_unlock( $guard );
						}
						return;
					}
					if ( 'retry' === $action ) {
						if ( $r['excluded'] ) {
							self::finish( $item, 'skipped', 'Excluded by you.' );
							continue;
						}if ( $r['requested'] || ! empty( $r['manifest']['draft'] ) ) {
							self::finish( $item, 'skipped', 'Existing work or comparison retained.' );
							continue;
						}Engine::retry( (int) $r['id'] );
						self::finish( $item, 'working', 'Retry queued.' );
						return;
					} elseif ( 'check' === $action ) {
						Verification::check_row( $r );
					} elseif ( in_array( $action, array( 'original', 'optimized' ), true ) ) {
						Repository::update( (int) $r['id'], array( 'delivery' => 'optimized' === $action ? 1 : 0 ) );
					} else {
						$exclude = 'exclude' === $action;
						$rev     = (int) $r['revision'] + 1;
						$m       = $r['manifest'];
						if ( isset( $m['revision'] ) ) {
							$m['revision'] = $rev;
						}
						Repository::update(
							(int) $r['id'],
							array(
								'excluded'  => $exclude ? 1 : 0,
								'state'     => $exclude ? 'excluded' : ( ! empty( $m['variants'] ) || ! empty( $m['responsive'] ) || ! empty( $m['watermarked'] ) ? 'ready' : 'discovered' ),
								'revision'  => $rev,
								'requested' => 0,
								'manifest'  => $m,
							),
							(int) $r['revision']
						);
					}
					self::finish( $item, 'complete', 'Action completed.' );
				} catch ( \Throwable $e ) {
					$reason = sanitize_key( $e->getMessage() );
					self::finish( $item, in_array( $reason, array( 'batch_access_changed', 'preview_public_required', 'preview_work_pending' ), true ) ? 'skipped' : 'failed', $reason );
				}
			}
			$sum    = self::summary( $job );
			$counts = Repository::counts();
			if ( $sum['done'] === $sum['total'] && ( 'retry' !== $job['action'] || empty( $counts['queued'] ) ) ) {
				$job['active']   = false;
				$job['finished'] = time();
				Reviewed_Store::update_option( Workflow::OPTION, $job, false );
				Activity::add( 'completed' );
			}
		} finally {
			Reviewed_Store::leave( $im_reviewed_owner_lock );
		}
	}
	/**
	 * Stop a bulk run: its items still waiting, working or being checked are marked skipped; their files are kept.
	 *
	 * @param array $job The run.
	 */
	public static function cancel( array $job ): void {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			global $wpdb;
			if ( 'bulk' !== $job['mode'] ) {
				return;
			}
        // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom batch queue uses current worker states and atomic INSERT/UPDATE operations; cross-request caching would make selection/progress stale.
			Reviewed_Store::database_query( $wpdb->prepare( "UPDATE %i SET state='skipped',reason='Run stopped; files retained.' WHERE batch_id=%s AND state IN ('waiting','working','checking')", Repository::table( 'batch' ), $job['id'] ) );

		} finally {
			Reviewed_Store::leave( $im_reviewed_owner_lock );
		}
	}
}