<?php
/**
 * Paced one-variant worker, explicit publication, invalidation, and safe retirement.
 *
 * @package AffStackDevImageManager
 */

namespace AffStackDev\ImageManager;

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

/**
 * The worker: one bounded step at a time under a lease, a source's copies made and checked, published under a short
 * lock, and its records kept in step when sources change.
 */
final class Engine {
	public const HOOK  = Hooks::TICK;
	public const GROUP = Hooks::GROUP;
	/**
	 * Which scheduler runs background work.
	 *
	 * @return string Action Scheduler or WP-Cron.
	 */
	public static function scheduler(): string {
		return Scheduler::backend();
	}
	/**
	 * Make sure background work is scheduled when it waits.
	 */
	public static function ensure(): void {
		Scheduler::ensure();
	}
	/**
	 * Unschedule background work and the health probe.
	 */
	public static function stop(): void {
		Scheduler::stop_all();
	}
	/**
	 * One worker step under the lease: finished migrations, the run's next pages or bulk items, the scan, page rechecks or
	 * the next queued source.
	 *
	 * @return bool True when a step was done, false when nothing was (paused, busy or no work).
	 * @throws \RuntimeException Only inside this function, where it is caught; callers get no exception.
	 */
	public static function tick(): bool {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			if ( is_multisite() || Settings::fresh()['paused'] ) {
				return false;
			}
			$token = Repository::acquire();
			if ( ! $token ) {
				return false;
			}
			$handle      = null;
			$own_receipt = false;
			try {
				$root = Settings::root( true );
				if ( is_link( $root . '/worker.lock' ) ) {
					throw new \RuntimeException( 'unsafe_storage' );
				}
				// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fopen -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
				$handle = fopen( $root . '/worker.lock', 'c' );
				if ( ! $handle || ! flock( $handle, LOCK_EX | LOCK_NB ) ) {
					return false;
				}
				self::recover_interrupted();
				Upgrade::records();
				Workflow::migrate_totals();
				Workflow::pump( $token );
				Reviewed_Store::update_option( 'afsd_im_last_worker', time(), false );
				$turn = (int) get_option( 'afsd_im_turn', 0 ) + 1;
				Reviewed_Store::update_option( 'afsd_im_turn', $turn % 6, false );
				// Alternate new-upload/backlog priority; reserve a turn for scanning so neither lane starves.
				if ( 0 === $turn % 3 && Scanner::pending() && Scanner::step() ) {
					Workflow::pump( $token );
					return true;
				}
				if ( 0 === $turn % 3 && Auto_Discovery::step( $token ) ) {
					return true;
				}
				$row = Repository::next_source( 0 === $turn % 2 ? 'upload' : 'backlog' );
				if ( $row ) {
					$own_receipt = true;
					self::flight(
						array(
							'source'   => (int) $row['id'],
							'revision' => (int) $row['revision'],
							'started'  => time(),
							'temp'     => '',
						)
					);
					self::process( $row, $token );
					self::flight( array() );
				} elseif ( Scanner::pending() ) {
					Scanner::step();
				} elseif ( Auto_Discovery::step( $token ) ) {
					return true;
				} else {
					Workflow::pump( $token );
					return false;
				}
				Workflow::pump( $token );
				Reviewed_Store::update_option( 'afsd_im_failures', 0, false );
			} catch ( \Throwable $e ) {
				// Preserve an inherited receipt if storage/recovery failed before this turn began.
				if ( $own_receipt ) {
					self::flight( array() );
				}
				$n = (int) get_option( 'afsd_im_failures', 0 ) + 1;
				Reviewed_Store::update_option( 'afsd_im_failures', $n, false );
				if ( $n >= 3 ) {
					Settings::pause( __( 'Repeated worker failure. Review Tools before resuming. Original files are unchanged.', 'affstackdev-image-manager' ) );
					Activity::add( 'worker_paused' );
				}
			} finally {
				// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
				if ( is_resource( $handle ) ) {
					flock( $handle, LOCK_UN );
					// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
					fclose( $handle );
				}
				Repository::release( $token );
			}
			return true;

		} finally {
			Reviewed_Store::leave( $im_reviewed_owner_lock );
		}
	}
	/**
	 * The next step of one source: its copies, display sizes and watermarked copies, each checked against its reference;
	 * then the source's state.
	 *
	 * @param array  $row   The source row.
	 * @param string $token The worker's lease.
	 * @throws \RuntimeException Only inside this function, where it is caught; callers get no exception.
	 */
	public static function process( array $row, string $token ): void {
		$row        = self::recover_pending( $row );
		$id         = (int) $row['id'];
		$attachment = (int) $row['attachment_id'];
		$rev        = (int) $row['revision'];
		$settings   = Settings::get();
		$draft_id   = ! empty( $row['manifest']['draft']['requested'] ) ? (string) $row['manifest']['draft']['id'] : '';
		$manifest   = $draft_id ? $row['manifest']['draft'] : $row['manifest'];
		if ( ! $settings['public_policy'] ) {
			self::work_update(
				$row,
				array(
					'state'     => 'held',
					'requested' => 0,
					'reason'    => 'public_policy_required',
				),
				$draft_id
			);
			return;
		}
		if ( $draft_id && ! empty( $manifest['batch_id'] ) ) {
			$job = Workflow::job();
			if ( $job['id'] !== $manifest['batch_id'] || $job['cancelled'] || ! $job['active'] ) {
				self::work_update(
					$row,
					array(
						'state'     => 'held',
						'requested' => 0,
						'reason'    => 'cancelled',
					),
					$draft_id
				);
				return;
			}
			try {
				Batch::authorized_row( $id, (int) $job['actor'] );
			} catch ( \Throwable $e ) {
				self::work_update(
					$row,
					array(
						'state'     => 'held',
						'requested' => 0,
						'reason'    => 'batch_access_changed',
					),
					$draft_id
				);
				return;
			}
		}
		if ( ! Sources::eligible( $row ) ) {
			self::work_update(
				$row,
				array(
					'state'     => 'held',
					'requested' => 0,
					'reason'    => 'public_association_required',
				),
				$draft_id
			);
			return;
		}
		$files = Sources::row_files( $row );
		if ( ! $files ) {
			self::temporary_failure( $row, 'metadata_wait' );
			return;
		}
		$root = Settings::root( true );
		$free = disk_free_space( $root );
		if ( false === $free || $free < $settings['min_free'] + $settings['max_bytes'] * 6 || ! wp_is_writable( $root ) ) {
			Settings::pause( __( 'Insufficient verified free disk space or writable image-copy storage. Free space, then resume.', 'affstackdev-image-manager' ) );
			Activity::add( 'worker_paused' );
			return;
		}
		if ( ! Compression::available() ) {
			Settings::pause( __( 'No verified image encoder is available. Check Tools; originals are unchanged.', 'affstackdev-image-manager' ) );
			Activity::add( 'worker_paused' );
			return;
		}
		$meta_hash = Sources::row_meta_hash( $row );
		if ( empty( $manifest['schema'] ) ) {
			$manifest = array_merge(
				$manifest,
				array(
					'schema'       => SCHEMA,
					'policy'       => Policy::VERSION,
					'revision'     => $rev,
					'meta_hash'    => $meta_hash,
					'fingerprints' => Sources::fingerprint( $attachment, $files ),
					'variants'     => array(),
					'results'      => array(),
					'retired'      => $manifest['retired'] ?? array(),
					'recipe'       => Policy::clean( (array) ( $manifest['recipe'] ?? Policy::current( $row ) ) ),
				)
			);
		}
		if ( ( $manifest['meta_hash'] ?? '' ) !== $meta_hash || (int) ( $manifest['revision'] ?? 0 ) !== $rev ) {
			self::invalidate_source( $id );
			return;
		}
		// Existing copies remain under their original recipe. Global changes apply to new work, not silent recompression.
		$recipe = Policy::clean( (array) ( $manifest['recipe'] ?? array_merge( Policy::defaults(), array( 'formats' => 'webp' ) ) ) );
		$next   = null;
		foreach ( $files as  $rel => $f ) {
			if ( ! isset( $manifest['results'][ $rel ] ) ) {
				$next = $f;
				break;
			}
		}
		if ( ! $next && ! empty( $manifest['sizing'] ) && 'existing' !== $manifest['sizing'] ) {
			if ( Responsive::process( $row, $manifest, $files, $recipe, $draft_id, $token, $settings ) ) {
				return;
			}
		}
		// 1.9.0 (D-500 gap 6): with the watermark on, the watermarked copies come after the ordinary ones.
		if ( ! $next && Watermark::process( $row, $manifest, $files, $recipe, $draft_id, $token, $settings ) ) {
			return;
		}
		if ( ! $next ) {
			$states = array_column( $manifest['results'], 'state' );
			$state  = ( $manifest['variants'] || ! empty( $manifest['responsive'] ) || ! empty( $manifest['watermarked'] ) ) ? 'ready' : ( in_array( 'no_saving', $states, true ) ? 'no_saving' : 'unsupported' );
			self::work_update(
				$row,
				array(
					'state'     => $state,
					'requested' => 0,
					'manifest'  => $manifest,
					'reason'    => ( $manifest['variants'] || ! empty( $manifest['responsive'] ) || ! empty( $manifest['watermarked'] ) ) ? 'Smaller validated copies are ready.' : 'No eligible smaller copy. Native image delivery is retained.',
				),
				$draft_id
			);
			return;
		}
		$tmp         = $root . '/tmp-' . bin2hex( random_bytes( 16 ) ) . '.webp';
		$created     = array();
		$committed   = false;
		$publication = null;
		try {
			$receipt         = (array) get_option( 'afsd_im_inflight', array() );
			$receipt['temp'] = basename( $tmp );
			self::flight( $receipt );
			$result = Compression::encode( $next['path'], $tmp, $recipe );
			if ( $result['width'] !== $next['width'] || $result['height'] !== $next['height'] ) {
				throw new \RuntimeException( 'source_metadata_dimensions_mismatch' );
			}
			if ( $result['worthwhile'] ) {
				$publication = self::publication_lock();
				$now         = Repository::row( $id );
				clearstatcache( true, $next['path'] );
				if ( ! Repository::owns( $token ) || ! $now || $now['excluded'] || (int) $now['revision'] !== $rev || Settings::fresh()['generation'] !== $settings['generation'] || ! Sources::eligible( $now ) || ! is_file( $next['path'] ) || ! hash_equals( $result['source_hash'], hash_file( 'sha256', $next['path'] ) ) || ( $draft_id && ( $now['manifest']['draft']['id'] ?? '' ) !== $draft_id ) ) {
					throw new \RuntimeException( 'stale_job' );
				}
				$choices   = array_merge( array( $result ), $result['alternatives'] );
				$journal   = array();
				$temporary = array();
				foreach ( $choices as  $r ) {
					$path = $r['_temporary'];
					unset( $r['_temporary'], $r['alternatives'] );
					$identity           = hash( 'sha256', $r['source_hash'] . '|' . $r['width'] . 'x' . $r['height'] . '|' . $r['recipe_key'] . '|' . $r['encoder'] . '|' . $r['encoder_version'] . '|' . $r['output_hash'] );
					$name               = Sources::prefix( $row ) . '-' . $identity . '.' . Variants::extension( $r['output_mime'] );
					$journal[]          = array_merge(
						$r,
						array(
							'file'         => $name,
							'name'         => $next['name'],
							'source_mtime' => (int) filemtime( $next['path'] ),
							'created'      => time(),
						)
					);
					$temporary[ $name ] = $path;
				}
				$manifest['pending'] = $journal;
				if ( ! self::work_update( $row, array( 'manifest' => $manifest ), $draft_id ) ) {
					throw new \RuntimeException( 'publication_journal_failed' );
				}
				foreach ( $journal as  $v ) {
					$path = $root . '/' . $v['file'];
					if ( is_link( $path ) ) {
						throw new \RuntimeException( 'unsafe_storage' );
					}
					if ( is_file( $path ) ) {
						if ( ! hash_equals( $v['output_hash'], hash_file( 'sha256', $path ) ) ) {
							throw new \RuntimeException( 'output_name_collision' );
						}Reviewed_Store::wp_delete_file( $temporary[ $v['file'] ] );
					}
                    // phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_chmod,WordPress.WP.AlternativeFunctions.rename_rename -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
					else {
						if ( ! Reviewed_Store::rename( $temporary[ $v['file'] ], $path ) ) {
							throw new \RuntimeException( 'atomic_publish_failed' );
						}$created[] = $path;
					// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_chmod -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
					}chmod( $path, 0644 );
				}
				$winner                                    = array_shift( $journal );
				$winner['alternatives']                    = $journal;
				$manifest['pending']                       = array();
				$manifest['variants'][ $next['relative'] ] = $winner;
				$manifest['results'][ $next['relative'] ]  = array(
					'state'      => 'ready',
					'reason'     => $winner['verification'],
					'recipe_key' => $winner['recipe_key'],
				);
			} else {
				$manifest['results'][ $next['relative'] ] = array(
					'state'              => 'no_saving',
					'reason'             => 'Less than 5% or 1 KiB saved; current file retained.',
					'source_bytes'       => $result['bytes'],
					'candidate_bytes'    => $result['output_bytes'],
					'attempted_encoders' => $result['attempted_encoders'] ?? array(),
				);
			}
			if ( ! Repository::owns( $token ) || Settings::fresh()['generation'] !== $settings['generation'] ) {
				throw new \RuntimeException( 'stale_job' );
			}
			$committed = self::work_update(
				$row,
				array(
					'manifest'     => $manifest,
					'state'        => 'queued',
					'available_at' => time(),
					'attempts'     => 0,
					'reason'       => 'Preparing image sizes.',
				),
				$draft_id
			);
		} catch ( \Throwable $e ) {
			$code      = sanitize_key( $e->getMessage() );
			$permanent = array(
				'invalid_image',
				'source_byte_limit',
				'pixel_limit',
				'memory_limit',
				'existing_webp_kept',
				'unsupported_format',
				'invalid_png',
				'truncated_png',
				'invalid_png_crc',
				'invalid_gif',
				'truncated_gif',
				'animated_loop_held',
				'animated_timing_held',
				'animated_frames_limit',
				'unsupported_precision',
				'profile_or_color_metadata_held',
				'animated_image_held',
				'alpha_requires_exact_imagick',
				'truncated_jpeg',
				'unsupported_precision_or_cmyk',
				'jpeg_profile_or_orientation_held',
				'unsupported_color_or_frames',
				'orientation_held',
				'pixel_or_source_mismatch',
				'not_verified_lossless_webp',
				'candidate_dimensions_or_mime',
				'source_metadata_dimensions_mismatch',
				'lossless_encoder_unavailable',
				'codec_warning',
				'requested_encoder_unavailable',
				'metadata_preserved_native_only',
				'metadata_roundtrip_failed',
				'quality_gate_failed',
				'alpha_changed',
				'metadata_needs_review',
				'gps_xmp_requires_review',
				'unsupported_lossless_target',
				'unsupported_png_chunk',
				'jpeg_verifier_unavailable',
			);
			if ( in_array( $code, $permanent, true ) ) {
				$manifest['results'][ $next['relative'] ] = array(
					'state'  => 'unsupported',
					'reason' => $code,
				);
				self::work_update(
					$row,
					array(
						'manifest'     => $manifest,
						'available_at' => time(),
						'reason'       => $code,
					),
					$draft_id
				);
			} elseif ( 'stale_job' !== $code ) {
				self::temporary_failure( $row, in_array( $code, array( 'encoder_timeout', 'encoder_output_limit' ), true ) ? $code : 'temporary_failure' );
			}
		} finally {
			Compression::cleanup_temporary( $tmp );
			if ( ! $committed ) {
				foreach ( $created as  $path ) {
					if ( is_file( $path ) && ! is_link( $path ) ) {
						Reviewed_Store::wp_delete_file( $path );
					}
				}
			}
			self::publication_unlock( $publication );
		}
	}
	/**
	 * A draft never replaces active variants before explicit approval.
	 *
	 * @param array  $row      The source row.
	 * @param array  $values   The fields to save.
	 * @param string $draft_id The draft being prepared, or an empty string for the active manifest.
	 * @return bool True when the update was saved (the revision still matched).
	 */
	public static function work_update( array $row, array $values, string $draft_id ): bool {
		if ( ! $draft_id ) {
			$ok = Repository::update( (int) $row['id'], $values, (int) $row['revision'] );
			if ( $ok ) {
				Progress::record( 'image_step', (int) $row['id'] );
			}return $ok;
		}
		$fresh = Repository::row( (int) $row['id'] );
		if ( ! $fresh || (int) $fresh['revision'] !== (int) $row['revision'] || ( $fresh['manifest']['draft']['id'] ?? '' ) !== $draft_id ) {
			return false;
		}
		$m       = $fresh['manifest'];
		$d       = $values['manifest'] ?? $m['draft'];
		$d['id'] = $draft_id;
		if ( isset( $values['state'] ) ) {
			$d['state'] = $values['state'];
		}
		if ( isset( $values['reason'] ) ) {
			$d['reason'] = $values['reason'];
		}
		if ( isset( $values['requested'] ) && 0 === $values['requested'] ) {
			$d['requested']   = false;
			$d['complete']    = true;
			$d['finished']    = time();
			$values['state']  = ( ! empty( $m['variants'] ) || ! empty( $m['responsive'] ) || ! empty( $m['watermarked'] ) ) ? 'ready' : ( $m['preview_prior_state'] ?? 'discovered' );
			$values['reason'] = ( ! empty( $m['variants'] ) || ! empty( $m['responsive'] ) || ! empty( $m['watermarked'] ) ) ? 'Existing prepared copies retained; preview is ready for review.' : 'Preview ready for review. No website image was changed.';
		}
		$m['draft']         = $d;
		$values['manifest'] = $m;
		$ok                 = Repository::update( (int) $row['id'], $values, (int) $row['revision'] );
		if ( $ok ) {
			Progress::record( 'image_step', (int) $row['id'] );
		}return $ok;
	}
	/**
	 * Short shared guard around publication/retirement, never the expensive encode.
	 *
	 * @return mixed The lock file's handle.
	 * @throws \RuntimeException With unsafe_storage when the lock file is a link, or another reason code when it cannot be
	 *          locked.
	 */
	public static function publication_lock() {
		$path = Settings::root( true ) . '/publication.lock';
		if ( is_link( $path ) ) {
			throw new \RuntimeException( 'unsafe_storage' );
		}
        // phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fopen -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
		$handle = fopen( $path, 'c' );
		if ( ! $handle ) {
			throw new \RuntimeException( 'publication_lock_failed' );
		}
		$until = microtime( true ) + 2;
		do {
			if ( flock( $handle, LOCK_EX | LOCK_NB ) ) {
				return $handle;
			}
			usleep( 25000 );
		} while ( microtime( true ) < $until );
        // phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
		fclose( $handle );
		throw new \RuntimeException( 'publication_busy' );
	}
	/**
	 * Release the publication lock.
	 *
	 * @param mixed $handle The lock file's handle.
	 */
	public static function publication_unlock( $handle ): void {
        // phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
		if ( is_resource( $handle ) ) {
			flock( $handle, LOCK_UN );
			// phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_fclose -- Validated local media/publication paths require native file descriptors for flock and atomic rename; published bytes are validated before the switch.
			fclose( $handle );
		}
	}
	/**
	 * Finish an interrupted publication: copies left pending are retired (kept for cached pages), under the publication
	 * lock.
	 *
	 * @param array $row The source row.
	 * @return array The source row as saved.
	 * @throws \RuntimeException With a reason code as its message when the step cannot be done.
	 */
	private static function recover_pending( array $row ): array {
		if ( empty( $row['manifest']['pending'] ) && empty( $row['manifest']['draft']['pending'] ) ) {
			return $row;
		}
		$guard = self::publication_lock();
		try {
			$row       = Repository::row( (int) $row['id'] ) ?: $row;
			$m         = $row['manifest'];
			$protected = $m;
			unset( $protected['pending'] );
			if ( isset( $protected['draft'] ) ) {
				unset( $protected['draft']['pending'] );
			}
			// Cached, previously published copies are protected too; pending-only entries are not.
			$active = array_column( Variants::flatten( $protected, true ), 'file' );
			foreach ( array( '', 'draft' ) as  $slot ) {
				$journal = $slot ? ( $m[ $slot ]['pending'] ?? array() ) : ( $m['pending'] ?? array() );
				foreach ( $journal as  $v ) {
					if ( in_array( $v['file'] ?? '', $active, true ) ) {
						continue;
					}if ( ! self::remove_variant( Sources::prefix( $row ), $v ) ) {
						$m['retired'][] = $v;
					}
				}
				if ( $slot && isset( $m[ $slot ] ) ) {
					$m[ $slot ]['pending'] = array();
				} elseif ( ! $slot ) {
					$m['pending'] = array();
				}
			}
			if ( ! Repository::update( (int) $row['id'], array( 'manifest' => $m ), (int) $row['revision'] ) ) {
				throw new \RuntimeException( 'journal_recovery_failed' );
			}
			$row['manifest'] = $m;
			return $row;
		} finally {
			self::publication_unlock( $guard );
		}
	}
	/**
	 * Count a failed attempt: the source is queued again after 2 minutes, then after 4; the third failure marks it failed.
	 *
	 * @param array  $row    The source row.
	 * @param string $reason Why it failed.
	 */
	public static function temporary_failure( array $row, string $reason ): void {
		$n = (int) $row['attempts'] + 1;
		Repository::update(
			(int) $row['id'],
			array(
				'attempts'     => $n,
				'available_at' => time() + 60 * ( 2 ** $n ),
				'state'        => $n >= 3 ? 'failed' : 'queued',
				'requested'    => $n >= 3 ? 0 : 1,
				'reason'       => $reason,
			),
			(int) $row['revision']
		);
		Activity::add( $n >= 3 ? 'failed' : 'retry', (int) $row['id'] );
	}
	/**
	 * Record the copy being made, so an interrupted encode's files can be found and removed.
	 *
	 * @param array $receipt The encode's receipt.
	 */
	private static function flight( array $receipt ): void {
		Reviewed_Store::update_option( 'afsd_im_inflight', $receipt, false );
	}
	/** Runs only after BOTH the database lease and filesystem guard are acquired. */
	private static function recover_interrupted(): void {
		$old = (array) Settings::fresh_option( 'afsd_im_inflight', array() );
		if ( ! $old ) {
			return;
		}
		$name = $old['temp'] ?? '';
		if ( is_string( $name ) && preg_match( '/^tmp-[a-f0-9]{32}\.webp$/D', $name ) ) {
			$path = Settings::root() . '/' . $name;
			Compression::cleanup_temporary( $path );
		}
		$row = Repository::row( (int) ( $old['source'] ?? 0 ) );
		if ( $row && 'queued' === $row['state'] && $row['requested'] && (int) $row['revision'] === (int) ( $old['revision'] ?? 0 ) ) {
			self::temporary_failure( $row, 'worker_interrupted' );
		}
		self::flight( array() );
		Activity::add( 'recovered', (int) ( $old['source'] ?? 0 ) );
	}
	/**
	 * Delete one copy file of a source, when its name is one Image Manager gives that source's copies.
	 *
	 * @param int|string $attachment The copy prefix (the attachment, or the theme source's prefix).
	 * @param array      $variant    The copy.
	 * @return bool True when the file is gone.
	 */
	public static function remove_variant( int|string $attachment, array $variant ): bool {
		$name = $variant['file'] ?? '';
		if ( ! is_string( $name ) || ! preg_match( '/^w?' . preg_quote( (string) $attachment, '/' ) . '-[a-f0-9]{64}\.(?:webp|png|jpg|avif)$/D', $name ) || ! preg_match( '/^[a-f0-9]{64}$/D', $variant['output_hash'] ?? '' ) ) {
			return false;
		}
		$root = Settings::root();
		$path = $root . '/' . $name;
		if ( ! file_exists( $path ) && ! is_link( $path ) ) {
			return true;
		}
		if ( is_link( $path ) || realpath( $path ) !== $path || ! hash_equals( $variant['output_hash'], hash_file( 'sha256', $path ) ) ) {
			return false;
		}
		Reviewed_Store::wp_delete_file( $path );
		clearstatcache( true, $path );
		return ! file_exists( $path );
	}
	/**
	 * Stop delivering an attachment's copies (its source is fenced and its copies retired).
	 *
	 * @param int  $attachment The attachment.
	 * @param bool $deleted    Whether the attachment was deleted.
	 */
	public static function invalidate_attachment( int $attachment, bool $deleted = false ): void {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			global $wpdb;
			$row = Repository::attachment( $attachment );
			if ( $row ) {
				self::invalidate_source( (int) $row['id'], $deleted );
			}
		} finally {
			Reviewed_Store::leave( $im_reviewed_owner_lock );
		}
	}
	/**
	 * Fence a source (a new revision, held, not requested), take its addresses off the index and retire its copies, under
	 * the publication lock.
	 *
	 * @param int  $source  The source row.
	 * @param bool $deleted Whether its attachment was deleted.
	 * @throws \RuntimeException With a reason code as its message when the step cannot be done.
	 */
	public static function invalidate_source( int $source, bool $deleted = false ): void {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			global $wpdb;
			$row = Repository::row( $source );
			if ( ! $row ) {
				return;
			}
			$attachment = (int) $row['attachment_id'];
			$guard      = self::publication_lock();
			try {
				$row = Repository::row( $source );
				if ( ! $row ) {
					return;
				}
				// A publisher cannot race this revision fence/file retirement: it shares this short lock.
				// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Atomically fence a source revision and remove its current URL index; these plugin-owned writes must become visible to other workers immediately.
				$fenced = Reviewed_Store::database_query( $wpdb->prepare( "UPDATE %i SET revision=revision+1,state='held',requested=0,updated_at=%d WHERE id=%d", Repository::table( 'sources' ), time(), $row['id'] ) );
				if ( false === $fenced ) {
					throw new \RuntimeException( 'revocation_fence_failed' );
				}
				$retired = array();
				foreach ( Variants::flatten( $row['manifest'], true ) as  $v ) {
					try {
						if ( ! self::remove_variant( Sources::prefix( $row ), $v ) ) {
							$retired[] = $v;
						}
					} catch ( \Throwable $e ) {
						$retired[] = $v;
					}
				}
				Repository::update(
					(int) $row['id'],
					array(
						'manifest' => array( 'retired' => $retired ),
						'reason'   => $retired ? 'Derivative removal needs attention; purge host/CDN caches for the removed content.' : ( $deleted ? 'Source removed; managed copies retired.' : 'Source or public ownership changed; native delivery restored.' ),
					)
				);
				// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- Atomically fence a source revision and remove its current URL index; these plugin-owned writes must become visible to other workers immediately.
				if ( $attachment ) {
					Reviewed_Store::database_delete( Repository::table( 'urls' ), array( 'source_id' => (int) $row['id'] ) );
				}
			} finally {
				self::publication_unlock( $guard );
			}
			Sources::forget_cache();
			if ( ! $deleted && $attachment ) {
				Sources::index( $attachment, (int) $row['id'] );
			}
			$fresh = Repository::row( (int) $row['id'] );
			if ( ! $deleted && ! $row['excluded'] && ( $row['requested'] || Settings::get()['uploads'] ) && $fresh && Sources::eligible( $fresh ) ) {
				Repository::request( (int) $row['id'], 'upload' );
			}
		} finally {
			Reviewed_Store::leave( $im_reviewed_owner_lock );
		}
	}
	/**
	 * Invalidate the sources an owner (a post or comment) had.
	 *
	 * @param string $type  post or comment.
	 * @param int    $owner Its ID.
	 */
	public static function revoke_owner( string $type, int $owner ): void {
		foreach ( Repository::affected( $type, $owner ) as  $id ) {
			$row = Repository::row( $id );
			if ( $row ) {
				self::invalidate_attachment( (int) $row['attachment_id'] );
			}
		}
	}
	/**
	 * Forget a source's results that are not ready and queue it again (not while a draft is being prepared).
	 *
	 * @param int $id The source row.
	 * @throws \RuntimeException With preview_work_pending while a draft is being prepared.
	 */
	public static function retry( int $id ): void {
		$im_reviewed_owner_lock = Reviewed_Store::enter();
		try {
			$row = Repository::row( $id );
			if ( ! $row ) {
				return;
			}
			if ( ! empty( $row['manifest']['draft'] ) ) {
				throw new \RuntimeException( 'preview_work_pending' );
			}
			$m = $row['manifest'];
			foreach ( $m['results'] ?? array() as  $rel => $r ) {
				if ( 'ready' !== $r['state'] ) {
					unset( $m['results'][ $rel ] );
				}
			}
			Repository::update( $id, array( 'manifest' => $m ) );
			Repository::request( $id );

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