<?php
/**
 * Job queue storage.
 *
 * @package R2Clouder
 */

namespace R2Clouder\Queue;

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

use R2Clouder\Installer;
use R2Clouder\Migration\Migration;
use WP_Error;

// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Every interpolated variable in this file is the table name from $wpdb->prefix or a generated placeholder list; never user input.

/**
 * Reads and writes the `r2clouder_jobs` table.
 *
 * Portability notes, because Studio runs on SQLite and production runs on
 * MySQL or MariaDB:
 *
 * - No `ON DUPLICATE KEY`, no `FULLTEXT`, no stored procedures.
 * - No `NOW()` in SQL. Every timestamp is computed in PHP as GMT, so the
 *   database's own idea of the current time never enters the comparison.
 * - Claiming is two-step: select candidate ids, then conditionally update each
 *   one and trust only the rows the update actually changed. `SELECT … FOR
 *   UPDATE` is unavailable on SQLite, and a batched `UPDATE … WHERE id IN (…)`
 *   reports a count without saying *which* rows it won, so each candidate is
 *   claimed with its own guarded statement.
 *
 * Every query here is uncached on purpose: queue state is what two concurrent
 * workers are racing over, and a stale read means double execution.
 */
final class Queue {

	/**
	 * Attempts after which a job is parked as failed.
	 */
	const MAX_ATTEMPTS = 5;

	/**
	 * Base backoff in seconds; doubled per attempt.
	 */
	const BACKOFF_BASE = 60;

	/**
	 * Ceiling for the computed backoff, in seconds.
	 */
	const BACKOFF_MAX = 3600;

	/**
	 * Minutes after which a `claimed` row is assumed abandoned and released.
	 */
	const REAP_AFTER_MINUTES = 15;

	/**
	 * Job types held while a bucket migration is copying, so the copy is a frozen snapshot.
	 */
	const FROZEN_TYPES = array( 'offload', 'convert', 'cleanup', 'delete', 'pullback' );

	/**
	 * Fully qualified jobs table name for the current site.
	 *
	 * @return string
	 */
	public static function table(): string {
		return Installer::jobs_table();
	}

	/**
	 * Current GMT datetime in MySQL format.
	 *
	 * @param int $offset Seconds to add.
	 * @return string
	 */
	public static function now( $offset = 0 ): string {
		return gmdate( 'Y-m-d H:i:s', time() + (int) $offset );
	}

	/**
	 * Adds a job to the queue.
	 *
	 * @param string              $job_type One of the Job::TYPE_* constants.
	 * @param array<string,mixed> $args     {
	 *     Optional. Job options.
	 *
	 *     @type int                 $attachment_id Attachment the job acts on. Default 0.
	 *     @type array<string,mixed> $payload       Arbitrary JSON-serialisable data. Default empty.
	 *     @type int                 $priority      Lower runs first. Default 10.
	 *     @type int                 $delay         Seconds before the job becomes available. Default 0.
	 *     @type bool                $unique        Skip when an unfinished job with the same type,
	 *                                              attachment and dedupe key already exists. Default true.
	 * }
	 * @return int|WP_Error Job id - the existing one when deduplicated - or an error.
	 */
	public static function enqueue( $job_type, array $args = array() ) {
		global $wpdb;

		$job_type = (string) $job_type;

		if ( ! in_array( $job_type, Job::types(), true ) ) {
			return new WP_Error(
				'r2clouder_invalid_job_type',
				/* translators: %s: job type. */
				sprintf( __( 'Unknown job type: %s', 'devsyed-media-offloader' ), $job_type )
			);
		}

		$args = array_merge(
			array(
				'attachment_id' => 0,
				'payload'       => array(),
				'priority'      => 10,
				'delay'         => 0,
				'unique'        => true,
			),
			$args
		);

		/**
		 * Filters a job before it is validated and queued.
		 *
		 * @param array<string,mixed> $args Job arguments.
		 * @param string              $job_type Job type.
		 */
		$filtered_args = apply_filters( 'r2clouder_queue_job_args', $args, $job_type );
		$args          = is_array( $filtered_args ) ? $filtered_args : $args;

		$attachment_id = max( 0, (int) $args['attachment_id'] );
		$payload       = is_array( $args['payload'] ) ? $args['payload'] : array();
		$priority      = min( 32767, max( -32768, (int) $args['priority'] ) );
		$delay         = max( 0, (int) $args['delay'] );

		if ( ! empty( $args['unique'] ) ) {
			$existing = self::find_unfinished( $job_type, $attachment_id, $payload );

			if ( $existing > 0 ) {
				return $existing;
			}
		}

		$encoded = wp_json_encode( $payload );

		if ( false === $encoded ) {
			return new WP_Error( 'r2clouder_payload_encode_failed', __( 'The job payload could not be encoded as JSON.', 'devsyed-media-offloader' ) );
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom queue table; a cached write makes no sense.
		$inserted = $wpdb->insert(
			self::table(),
			array(
				'attachment_id' => $attachment_id,
				'job_type'      => $job_type,
				'status'        => Job::STATUS_PENDING,
				'priority'      => $priority,
				'attempts'      => 0,
				'available_at'  => self::now( $delay ),
				'payload'       => $encoded,
				'created_at'    => self::now(),
			),
			array( '%d', '%s', '%s', '%d', '%d', '%s', '%s', '%s' )
		);

		if ( ! $inserted ) {
			return new WP_Error( 'r2clouder_enqueue_failed', __( 'The job could not be written to the queue.', 'devsyed-media-offloader' ) );
		}

		$job_id = (int) $wpdb->insert_id;

		do_action(
			'r2clouder_job_queued',
			$job_id,
			$job_type,
			array(
				'attachment_id' => $attachment_id,
				'priority'      => $priority,
				'delay'         => $delay,
			)
		);

		return $job_id;
	}

	/**
	 * Finds an existing pending or claimed job matching a new one.
	 *
	 * Deduplication is keyed on type plus attachment, and additionally on the
	 * payload's `dedupe` value when one is given, so that two purges of
	 * different URLs are not collapsed into one.
	 *
	 * @param string              $job_type      Job type.
	 * @param int                 $attachment_id Attachment id.
	 * @param array<string,mixed> $payload       Payload of the job being enqueued.
	 * @return int Existing job id, or 0.
	 */
	public static function find_unfinished( $job_type, $attachment_id, array $payload = array() ): int {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
		$sql = $wpdb->prepare(
			"SELECT id, payload FROM {$table} WHERE job_type = %s AND attachment_id = %d AND status IN ( %s, %s ) ORDER BY id ASC LIMIT 50",
			$job_type,
			(int) $attachment_id,
			Job::STATUS_PENDING,
			Job::STATUS_CLAIMED
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue reads must not be cached.
		$rows = $wpdb->get_results( $sql, ARRAY_A );

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

		$dedupe = isset( $payload['dedupe'] ) ? (string) $payload['dedupe'] : '';

		foreach ( $rows as $row ) {
			$candidate = Job::from_row( $row );

			if ( '' === $dedupe ) {
				return $candidate->id;
			}

			if ( $dedupe === (string) $candidate->get( 'dedupe', '' ) ) {
				return $candidate->id;
			}
		}

		return 0;
	}

	/**
	 * Claims up to `$limit` due jobs for the current worker.
	 *
	 * `attempts` is incremented at claim time rather than on failure, so a
	 * worker killed mid-job still burns an attempt and a poison job cannot loop
	 * for ever.
	 *
	 * @param int $limit Maximum jobs to claim.
	 * @return Job[] Jobs this call won, in run order.
	 */
	public static function claim( $limit = 20 ): array {
		global $wpdb;

		$limit = min( 200, max( 1, (int) $limit ) );
		$table = self::table();
		$now   = self::now();

		$held  = Migration::is_frozen() ? self::FROZEN_TYPES : array( '' );
		$holds = implode( ', ', array_fill( 0, count( $held ), '%s' ) );

		// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- Placeholder list is generated to match the merged values.
		$sql = $wpdb->prepare(
			"SELECT id FROM {$table} WHERE status = %s AND available_at <= %s AND job_type NOT IN ( {$holds} ) ORDER BY priority ASC, id ASC LIMIT %d",
			array_merge( array( Job::STATUS_PENDING, $now ), $held, array( $limit ) )
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue reads must not be cached.
		$ids = $wpdb->get_col( $sql );

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

		$claimed = array();

		foreach ( $ids as $id ) {
			$id = (int) $id;

			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
			$update = $wpdb->prepare(
				"UPDATE {$table} SET status = %s, claimed_at = %s, attempts = attempts + 1 WHERE id = %d AND status = %s",
				Job::STATUS_CLAIMED,
				$now,
				$id,
				Job::STATUS_PENDING
			);

			// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; this guarded update is the claim.
			$won = $wpdb->query( $update );

			if ( 1 !== (int) $won ) {
				// Another worker got there first.
				continue;
			}

			$job = self::get( $id );

			if ( $job instanceof Job ) {
				$claimed[] = $job;
			}
		}

		return $claimed;
	}

	/**
	 * Loads a single job.
	 *
	 * @param int $id Job id.
	 * @return Job|null
	 */
	public static function get( $id ): ?Job {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
		$sql = $wpdb->prepare( "SELECT * FROM {$table} WHERE id = %d", (int) $id );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue reads must not be cached.
		$row = $wpdb->get_row( $sql, ARRAY_A );

		return is_array( $row ) ? Job::from_row( $row ) : null;
	}

	/**
	 * Marks a job finished.
	 *
	 * @param int $id Job id.
	 * @return bool
	 */
	public static function complete( $id ): bool {
		return self::update_row(
			(int) $id,
			array(
				'status'      => Job::STATUS_DONE,
				'finished_at' => self::now(),
				'last_error'  => '',
			),
			array( '%s', '%s', '%s' )
		);
	}

	/**
	 * Records a failure, retrying with backoff until MAX_ATTEMPTS is spent.
	 *
	 * @param Job             $job   The job that failed.
	 * @param WP_Error|string $error Failure reason.
	 * @return string The persisted status, `pending` or `failed`; empty on a write failure.
	 */
	public static function fail( Job $job, $error ): string {
		$message = self::error_message( $error );

		if ( $job->attempts >= self::MAX_ATTEMPTS ) {
			$updated = self::update_row(
				$job->id,
				array(
					'status'      => Job::STATUS_FAILED,
					'finished_at' => self::now(),
					'last_error'  => $message,
				),
				array( '%s', '%s', '%s' )
			);

			return $updated ? Job::STATUS_FAILED : '';
		}

		$updated = self::update_row(
			$job->id,
			array(
				'status'       => Job::STATUS_PENDING,
				'claimed_at'   => null,
				'available_at' => self::now( self::backoff_delay( $job->attempts ) ),
				'last_error'   => $message,
			),
			array( '%s', '%s', '%s', '%s' )
		);

		return $updated ? Job::STATUS_PENDING : '';
	}

	/**
	 * Parks a job as failed without spending further attempts.
	 *
	 * Used for errors that cannot improve on a retry, such as an unknown job
	 * type or a deleted attachment.
	 *
	 * @param Job             $job   The job to park.
	 * @param WP_Error|string $error Failure reason.
	 * @return bool Whether the failure was persisted.
	 */
	public static function fail_permanently( Job $job, $error ): bool {
		return self::update_row(
			$job->id,
			array(
				'status'      => Job::STATUS_FAILED,
				'finished_at' => self::now(),
				'attempts'    => self::MAX_ATTEMPTS,
				'last_error'  => self::error_message( $error ),
			),
			array( '%s', '%s', '%d', '%s' )
		);
	}

	/**
	 * Counts jobs of one type in the given statuses.
	 *
	 * @param string   $job_type Job type.
	 * @param string[] $statuses Statuses to include.
	 * @param string   $migration_id Optional migration identity to count.
	 * @return int
	 */
	public static function count_type( $job_type, array $statuses, $migration_id = '' ): int {
		global $wpdb;

		$table = self::table();
		if ( empty( $statuses ) ) {
			return 0;
		}

		$placeholders = implode( ', ', array_fill( 0, count( $statuses ), '%s' ) );
		$scope        = '';
		$values       = array_merge( array( $job_type ), $statuses );
		if ( '' !== $migration_id ) {
			$scope    = ' AND payload LIKE %s';
			$values[] = '%' . $wpdb->esc_like( '"migration_id":' . wp_json_encode( (string) $migration_id ) ) . '%';
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared here; queue reads must not be cached.
		return (int) $wpdb->get_var(
			$wpdb->prepare(
				"SELECT COUNT(*) FROM {$table} WHERE job_type = %s AND status IN ( {$placeholders} ){$scope}",
				$values
			)
		);
	}

	/**
	 * Returns a claimed job to the pending pool, optionally delayed.
	 *
	 * @param int $id    Job id.
	 * @param int $delay Seconds before it becomes available again.
	 * @return bool
	 */
	public static function release( $id, $delay = 0 ): bool {
		return self::update_row(
			(int) $id,
			array(
				'status'       => Job::STATUS_PENDING,
				'claimed_at'   => null,
				'available_at' => self::now( max( 0, (int) $delay ) ),
			),
			array( '%s', '%s', '%s' )
		);
	}

	/**
	 * Releases `claimed` rows whose worker never came back.
	 *
	 * A PHP timeout, a fatal error or a killed cron request all leave a row
	 * claimed for ever otherwise.
	 *
	 * @param int $minutes Age threshold in minutes.
	 * @return int Rows released.
	 */
	public static function reap_stuck( $minutes = self::REAP_AFTER_MINUTES ): int {
		global $wpdb;

		$minutes   = max( 1, (int) $minutes );
		$table     = self::table();
		$threshold = self::now( -1 * $minutes * MINUTE_IN_SECONDS );

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
		$sql = $wpdb->prepare(
			"UPDATE {$table} SET status = CASE WHEN attempts >= %d THEN %s ELSE %s END, claimed_at = NULL, available_at = %s, finished_at = CASE WHEN attempts >= %d THEN %s ELSE finished_at END, last_error = CASE WHEN attempts >= %d THEN %s ELSE last_error END WHERE status = %s AND claimed_at IS NOT NULL AND claimed_at < %s",
			self::MAX_ATTEMPTS,
			Job::STATUS_FAILED,
			Job::STATUS_PENDING,
			self::now(),
			self::MAX_ATTEMPTS,
			self::now(),
			self::MAX_ATTEMPTS,
			__( 'The worker was interrupted too many times.', 'devsyed-media-offloader' ),
			Job::STATUS_CLAIMED,
			$threshold
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue writes are not cacheable.
		$released = $wpdb->query( $sql );

		return is_numeric( $released ) ? (int) $released : 0;
	}

	/**
	 * Seconds to wait before the next attempt.
	 *
	 * @param int $attempts Attempts already spent.
	 * @return int
	 */
	public static function backoff_delay( $attempts ): int {
		$attempts = max( 0, min( 16, (int) $attempts ) );

		return (int) min( self::BACKOFF_MAX, self::BACKOFF_BASE * pow( 2, $attempts ) );
	}

	/**
	 * Counts rows per status.
	 *
	 * @return array<string,int> Keyed by status, every status present.
	 */
	public static function counts(): array {
		global $wpdb;

		$table  = self::table();
		$counts = array_fill_keys( Job::statuses(), 0 );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQL.NotPrepared -- Static query on a custom table; queue counts must be live.
		$rows = $wpdb->get_results( "SELECT status, COUNT(*) AS total FROM {$table} GROUP BY status", ARRAY_A );

		if ( ! is_array( $rows ) ) {
			return $counts;
		}

		foreach ( $rows as $row ) {
			$status = isset( $row['status'] ) ? (string) $row['status'] : '';

			if ( isset( $counts[ $status ] ) ) {
				$counts[ $status ] = isset( $row['total'] ) ? (int) $row['total'] : 0;
			}
		}

		return $counts;
	}

	/**
	 * Lists jobs for the admin screens.
	 *
	 * @param array<string,mixed> $args {
	 *     Optional. Query arguments.
	 *
	 *     @type string $status        Restrict to one status.
	 *     @type string $job_type      Restrict to one job type.
	 *     @type int    $attachment_id Restrict to one attachment.
	 *     @type int    $limit         Rows to return. Default 50, max 500.
	 *     @type int    $offset        Rows to skip. Default 0.
	 * }
	 * @return Job[]
	 */
	public static function get_jobs( array $args = array() ): array {
		global $wpdb;

		$args = array_merge(
			array(
				'status'        => '',
				'job_type'      => '',
				'attachment_id' => 0,
				'limit'         => 50,
				'offset'        => 0,
			),
			$args
		);

		$table  = self::table();
		$where  = array( '1=1' );
		$values = array();

		if ( in_array( $args['status'], Job::statuses(), true ) ) {
			$where[]  = 'status = %s';
			$values[] = $args['status'];
		}

		if ( in_array( $args['job_type'], Job::types(), true ) ) {
			$where[]  = 'job_type = %s';
			$values[] = $args['job_type'];
		}

		if ( (int) $args['attachment_id'] > 0 ) {
			$where[]  = 'attachment_id = %d';
			$values[] = (int) $args['attachment_id'];
		}

		$values[] = min( 500, max( 1, (int) $args['limit'] ) );
		$values[] = max( 0, (int) $args['offset'] );

		$clause = implode( ' AND ', $where );

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- Table name from $wpdb->prefix; $clause is assembled from literals only; $values is a dynamic array whose count matches the placeholders.
		$sql = $wpdb->prepare(
			"SELECT * FROM {$table} WHERE {$clause} ORDER BY id DESC LIMIT %d OFFSET %d",
			$values
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue reads must not be cached.
		$rows = $wpdb->get_results( $sql, ARRAY_A );

		if ( ! is_array( $rows ) ) {
			return array();
		}

		return array_map(
			static function ( $row ) {
				return Job::from_row( $row );
			},
			$rows
		);
	}

	/**
	 * Moves failed jobs back to pending so they run on the next tick.
	 *
	 * @param int[] $ids Job ids. Empty retries every failed job.
	 * @return int Rows affected.
	 */
	public static function retry( array $ids = array() ): int {
		global $wpdb;

		$table = self::table();
		$now   = self::now();

		if ( empty( $ids ) ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
			$sql = $wpdb->prepare(
				"UPDATE {$table} SET status = %s, attempts = 0, available_at = %s, claimed_at = NULL, finished_at = NULL WHERE status = %s",
				Job::STATUS_PENDING,
				$now,
				Job::STATUS_FAILED
			);
		} else {
			$ids = array_values( array_filter( array_map( 'absint', $ids ) ) );

			if ( empty( $ids ) ) {
				return 0;
			}

			$placeholders = implode( ', ', array_fill( 0, count( $ids ), '%d' ) );

			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- Table name from $wpdb->prefix; $placeholders is a generated list of %d tokens; replacement count is dynamic.
			$sql = $wpdb->prepare(
				"UPDATE {$table} SET status = %s, attempts = 0, available_at = %s, claimed_at = NULL, finished_at = NULL WHERE status = %s AND id IN ( {$placeholders} )",
				array_merge( array( Job::STATUS_PENDING, $now, Job::STATUS_FAILED ), $ids )
			);
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue writes are not cacheable.
		$affected = $wpdb->query( $sql );

		$affected = is_numeric( $affected ) ? (int) $affected : 0;

		if ( $affected > 0 ) {
			do_action( 'r2clouder_jobs_retried', $ids, $affected );
		}

		return $affected;
	}

	/**
	 * Deletes jobs by id.
	 *
	 * @param int[] $ids Job ids.
	 * @return int Rows deleted.
	 */
	public static function delete( array $ids ): int {
		global $wpdb;

		$ids = array_values( array_filter( array_map( 'absint', $ids ) ) );

		if ( empty( $ids ) ) {
			return 0;
		}

		$table        = self::table();
		$placeholders = implode( ', ', array_fill( 0, count( $ids ), '%d' ) );

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- Table name from $wpdb->prefix; $placeholders is a generated list of %d tokens; $ids is passed as an array.
		$sql = $wpdb->prepare( "DELETE FROM {$table} WHERE id IN ( {$placeholders} )", $ids );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue writes are not cacheable.
		$deleted = $wpdb->query( $sql );

		return is_numeric( $deleted ) ? (int) $deleted : 0;
	}

	/**
	 * Empties the queue, or one status of it.
	 *
	 * @param string $status Optional status to restrict to.
	 * @return int Rows deleted.
	 */
	public static function clear( $status = '' ): int {
		global $wpdb;

		$table = self::table();

		if ( in_array( $status, Job::statuses(), true ) ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
			$sql = $wpdb->prepare( "DELETE FROM {$table} WHERE status = %s", $status );
		} else {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
			$sql = "DELETE FROM {$table}";
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above where values are involved; queue writes are not cacheable.
		$deleted = $wpdb->query( $sql );

		return is_numeric( $deleted ) ? (int) $deleted : 0;
	}

	/**
	 * Removes completed rows older than the given number of days.
	 *
	 * Keeps the table from growing without bound after a large migration.
	 * Failed rows are never touched - they are the dead-letter list.
	 *
	 * @param int $days Age threshold in days.
	 * @return int Rows deleted.
	 */
	public static function purge_completed( $days = 7 ): int {
		global $wpdb;

		$table     = self::table();
		$threshold = self::now( -1 * max( 1, (int) $days ) * DAY_IN_SECONDS );

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
		$sql = $wpdb->prepare(
			"DELETE FROM {$table} WHERE status = %s AND finished_at IS NOT NULL AND finished_at < %s",
			Job::STATUS_DONE,
			$threshold
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue writes are not cacheable.
		$deleted = $wpdb->query( $sql );

		return is_numeric( $deleted ) ? (int) $deleted : 0;
	}

	/**
	 * Whether any job is due to run right now.
	 *
	 * @return bool
	 */
	public static function has_due_jobs(): bool {
		global $wpdb;

		$table = self::table();

		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is built from $wpdb->prefix.
		$sql = $wpdb->prepare(
			"SELECT id FROM {$table} WHERE status = %s AND available_at <= %s LIMIT 1",
			Job::STATUS_PENDING,
			self::now()
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.NotPrepared -- Prepared above; queue reads must not be cached.
		$id = $wpdb->get_var( $sql );

		return null !== $id;
	}

	/**
	 * Applies a column update to one row.
	 *
	 * @param int                 $id      Job id.
	 * @param array<string,mixed> $data    Column values.
	 * @param string[]            $formats Format strings matching $data.
	 * @return bool
	 */
	private static function update_row( $id, array $data, array $formats ): bool {
		global $wpdb;

		if ( $id <= 0 ) {
			return false;
		}

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- Custom queue table; a cached write makes no sense.
		$updated = $wpdb->update(
			self::table(),
			$data,
			array( 'id' => (int) $id ),
			$formats,
			array( '%d' )
		);

		if ( false === $updated ) {
			return false;
		}

		// MySQL reports zero for an unchanged row as well as a deleted one.
		// A job removed while its handler ran has no persisted outcome.
		return $updated > 0 || null !== self::get( $id );
	}

	/**
	 * Normalises a failure reason into a storable string.
	 *
	 * @param WP_Error|string $error Failure reason.
	 * @return string Truncated to fit a TEXT column comfortably.
	 */
	private static function error_message( $error ): string {
		if ( is_wp_error( $error ) ) {
			$message = $error->get_error_code() . ': ' . $error->get_error_message();
		} else {
			$message = (string) $error;
		}

		$message = trim( preg_replace( '/\s+/', ' ', $message ) );

		if ( strlen( $message ) > 2000 ) {
			$message = substr( $message, 0, 2000 );
		}

		return $message;
	}
}