<?php
/**
 * Job queue.
 *
 * @package Rahkar\WP
 */

declare( strict_types = 1 );

namespace Rahkar\WP\Queue;

use Rahkar\WP\Database\Schema;
use Rahkar\WP\Support\Logger;

defined( 'ABSPATH' ) || exit;

/**
 * Durable background work, independent of WP-Cron.
 *
 * Iranian hosts routinely set `DISABLE_WP_CRON`, and Action Scheduler ultimately
 * leans on the same loopback machinery, so neither can be trusted to deliver an
 * order SMS. Jobs live in a table this plugin owns and are drained by
 * {@see Dispatcher} through three independent triggers.
 *
 * What belongs here: anything the visitor is not waiting on. An OTP send does
 * *not* belong here — the user is staring at the code field — so that path sends
 * synchronously and only falls back to the queue on failure.
 */
final class Queue {

	public const PENDING  = 0;
	public const RESERVED = 1;
	public const DONE     = 2;
	public const FAILED   = 3;

	/**
	 * Attempts before a job is parked as failed.
	 */
	public const MAX_ATTEMPTS = 5;

	/**
	 * Autoloaded watermark: the timestamp of the earliest job waiting to run.
	 *
	 * Zero means the queue is empty. This exists so the shutdown drain can decide
	 * whether to do anything *without a query* — WordPress has already fetched
	 * every autoloaded option, so reading it is free, while a `SELECT ... LIMIT 1`
	 * on every request would blow the plugin's zero-added-queries budget on pages
	 * that have nothing to do with the queue.
	 *
	 * It is a hint, not a source of truth: it may be later than reality after a
	 * manual database edit, and the worker endpoint ignores it entirely.
	 *
	 * The name itself lives on {@see \Rahkar\WP\Plugin}, which is always loaded,
	 * so the boot path can read the watermark without autoloading this class.
	 */
	public const WATERMARK_OPTION = \Rahkar\WP\Plugin::QUEUE_WATERMARK_OPTION;

	/**
	 * Add a job.
	 *
	 * @param string              $hook     Action name fired when the job runs.
	 * @param array<string,mixed> $payload  Arguments passed to the action.
	 * @param int                 $delay    Seconds to wait before the job is eligible.
	 * @param int                 $priority Lower runs first.
	 * @param string              $group    Optional grouping key, used for deduplication.
	 *                                      Keep it under 64 characters: it is stored in a
	 *                                      fixed-width column and silently truncated beyond
	 *                                      that, which would make two distinct jobs look like
	 *                                      duplicates of each other.
	 * @param bool                $urgent   Poke the loopback trigger rather than waiting for the next visitor.
	 *
	 * @return int Job ID, or 0 on failure.
	 */
	public static function push( string $hook, array $payload = [], int $delay = 0, int $priority = 10, string $group = '', bool $urgent = false ): int {
		global $wpdb;

		$encoded = wp_json_encode( $payload, JSON_UNESCAPED_UNICODE );

		if ( ! is_string( $encoded ) ) {
			Logger::error( 'queue', 'Job payload could not be encoded.', [ 'hook' => $hook ] );

			return 0;
		}

		$now = time();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$inserted = $wpdb->insert(
			Schema::table( 'jobs' ),
			[
				'hook'         => substr( $hook, 0, 64 ),
				'payload'      => $encoded,
				'job_group'    => substr( $group, 0, 64 ),
				'priority'     => max( 0, min( 255, $priority ) ),
				'status'       => self::PENDING,
				'attempts'     => 0,
				'available_at' => gmdate( 'Y-m-d H:i:s', $now + max( 0, $delay ) ),
				'created_at'   => gmdate( 'Y-m-d H:i:s', $now ),
			],
			[ '%s', '%s', '%s', '%d', '%d', '%d', '%s', '%s' ]
		);

		if ( ! $inserted ) {
			Logger::error( 'queue', 'Job insert failed.', [ 'hook' => $hook ] );

			return 0;
		}

		self::lowerWatermark( $now + max( 0, $delay ) );

		Dispatcher::instance()->schedule( $urgent );

		return (int) $wpdb->insert_id;
	}

	/**
	 * Add a job only if an identical one is not already waiting.
	 *
	 * Prevents duplicate work when the same event can fire more than once — a
	 * status transition saved twice should not send the customer two texts.
	 *
	 * @param string              $hook     Action name.
	 * @param array<string,mixed> $payload  Action arguments.
	 * @param string              $group    Grouping key. Required for this to mean anything.
	 * @param int                 $delay    Seconds to wait.
	 * @param int                 $priority Lower runs first.
	 *
	 * @return int Job ID, or 0 when a duplicate already exists.
	 */
	public static function pushUnique( string $hook, array $payload, string $group, int $delay = 0, int $priority = 10 ): int {
		global $wpdb;

		$table = Schema::table( 'jobs' );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$exists = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT id FROM %i WHERE hook = %s AND job_group = %s AND status IN (%d, %d) LIMIT 1",
				$table,
				$hook,
				$group,
				self::PENDING,
				self::RESERVED
			)
		);

		if ( null !== $exists ) {
			return 0;
		}

		return self::push( $hook, $payload, $delay, $priority, $group );
	}

	/**
	 * Claim a batch of due jobs.
	 *
	 * The claim is a single atomic `UPDATE ... LIMIT` that flips rows to RESERVED
	 * and stamps them with a marker, then reads back exactly those rows. Two
	 * workers running concurrently — a cron tick landing on top of a post-response
	 * drain — cannot end up with the same job, because only one `UPDATE` can move
	 * a row out of PENDING.
	 *
	 * @param int $limit Maximum jobs to claim.
	 *
	 * @return list<array<string,mixed>>
	 */
	public static function claim( int $limit ): array {
		global $wpdb;

		$table  = Schema::table( 'jobs' );
		$marker = gmdate( 'Y-m-d H:i:s' );

		// A per-claim token makes the batch identifiable, so the read-back cannot
		// pick up rows another worker reserved in the same second. It gets its own
		// column rather than borrowing `last_error`, which has to survive the claim
		// so a retried job still carries why it failed last time.
		$token = substr( wp_generate_password( 32, false, false ), 0, 32 );

		// phpcs:disable WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.NotPrepared
		$claimed = $wpdb->query(
			$wpdb->prepare(
				"UPDATE %i
				 SET status = %d, reserved_at = %s, claim_token = %s
				 WHERE status = %d AND available_at <= %s
				 ORDER BY priority ASC, id ASC
				 LIMIT %d",
				$table,
				self::RESERVED,
				$marker,
				$token,
				self::PENDING,
				$marker,
				max( 1, $limit )
			)
		);

		if ( ! $claimed ) {
			return [];
		}

		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM %i WHERE claim_token = %s ORDER BY priority ASC, id ASC",
				$table,
				$token
			),
			ARRAY_A
		);
		// phpcs:enable

		return is_array( $rows ) ? $rows : [];
	}

	/**
	 * Mark a job finished and remove it.
	 *
	 * Completed rows are deleted rather than kept — the log table already records
	 * what happened, and an ever-growing jobs table would slowly degrade the claim
	 * query on exactly the busy stores that can least afford it.
	 *
	 * @param int $id Job ID.
	 */
	public static function complete( int $id ): void {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->delete( Schema::table( 'jobs' ), [ 'id' => $id ], [ '%d' ] );
	}

	/**
	 * Release a job back to PENDING with exponential backoff, or park it as failed.
	 *
	 * @param int    $id       Job ID.
	 * @param int    $attempts Attempts made so far.
	 * @param string $error    Failure detail.
	 */
	public static function release( int $id, int $attempts, string $error ): void {
		global $wpdb;

		$table = Schema::table( 'jobs' );

		if ( $attempts >= self::MAX_ATTEMPTS ) {
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery
			$wpdb->update(
				$table,
				[
					'status'      => self::FAILED,
					'attempts'    => $attempts,
					'reserved_at' => null,
					'claim_token' => '',
					'last_error'  => substr( $error, 0, 1000 ),
				],
				[ 'id' => $id ],
				[ '%d', '%d', '%s', '%s', '%s' ],
				[ '%d' ]
			);

			Logger::error( 'queue', 'Job exhausted its retries.', [ 'id' => $id, 'error' => $error ] );

			return;
		}

		// 1m, 4m, 9m, 16m — quadratic rather than doubling, so a provider outage
		// backs off fast without a transient blip pushing the retry out by hours.
		$backoff = min( 3600, 60 * ( $attempts ** 2 ) );

		// A released job is due again later, so the dispatcher has to be told to
		// wake for it — otherwise a retry could sit unnoticed until the next push.
		self::lowerWatermark( time() + $backoff );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->update(
			$table,
			[
				'status'       => self::PENDING,
				'attempts'     => $attempts,
				'reserved_at'  => null,
				'claim_token'  => '',
				'available_at' => gmdate( 'Y-m-d H:i:s', time() + $backoff ),
				'last_error'   => substr( $error, 0, 1000 ),
			],
			[ 'id' => $id ],
			[ '%d', '%d', '%s', '%s', '%s', '%s' ],
			[ '%d' ]
		);
	}

	/**
	 * How long a job may wait for a handler before it is parked as failed.
	 *
	 * Long enough to outlast an add-on that is deactivated for a week while a
	 * conflict is investigated, or a licence that lapses and is renewed.
	 */
	public const UNHANDLED_TTL = 30 * DAY_IN_SECONDS;

	/**
	 * Hand back a job that nothing on this request was listening for.
	 *
	 * Not a failure, and not counted as an attempt. The job may belong to a
	 * module registered only in other kinds of request — the log purge runs in
	 * background contexts, the OTP cleanup on front-end ones — or to an add-on
	 * that is switched off right now. Completing it, which is what happened
	 * before, deleted work that had simply been claimed in the wrong place: a
	 * review invitation scheduled a week out vanished if the pro add-on was
	 * deactivated on the day it came due.
	 *
	 * A job that stays unhandled past {@see self::UNHANDLED_TTL} is parked as
	 * failed with a reason, where the health screen shows it, rather than being
	 * reclaimed forever.
	 *
	 * @param array<string,mixed> $row   Claimed job row.
	 * @param int                 $delay Seconds before it becomes eligible again.
	 */
	public static function defer( array $row, int $delay ): void {
		global $wpdb;

		$id      = (int) $row['id'];
		$created = (int) strtotime( (string) $row['created_at'] . ' UTC' );
		$table   = Schema::table( 'jobs' );

		if ( $created > 0 && ( time() - $created ) > self::UNHANDLED_TTL ) {
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery
			$wpdb->update(
				$table,
				[
					'status'      => self::FAILED,
					'reserved_at' => null,
					'claim_token' => '',
					'last_error'  => 'No handler was registered for this job within the retention window.',
				],
				[ 'id' => $id ],
				[ '%d', '%s', '%s', '%s' ],
				[ '%d' ]
			);

			Logger::warning( 'queue', 'Job parked: nothing handled it.', [ 'id' => $id, 'hook' => (string) $row['hook'] ] );

			return;
		}

		self::lowerWatermark( time() + $delay );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->update(
			$table,
			[
				'status'       => self::PENDING,
				'reserved_at'  => null,
				'claim_token'  => '',
				'available_at' => gmdate( 'Y-m-d H:i:s', time() + $delay ),
			],
			[ 'id' => $id ],
			[ '%d', '%s', '%s', '%s' ],
			[ '%d' ]
		);
	}

	/**
	 * Return jobs stuck in RESERVED to the pending pool.
	 *
	 * A worker killed mid-job — PHP timeout, a restarted php-fpm pool — leaves its
	 * rows reserved forever. Without this the queue would quietly stop draining
	 * and nobody would notice until customers stopped getting texts.
	 *
	 * @param int $olderThan Seconds a reservation may stand before it is reclaimed.
	 *
	 * @return int Rows recovered.
	 */
	public static function recoverStalled( int $olderThan = 300 ): int {
		global $wpdb;

		$table = Schema::table( 'jobs' );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$recovered = (int) $wpdb->query(
			$wpdb->prepare(
				"UPDATE %i
				 SET status = %d, reserved_at = NULL, claim_token = '', attempts = attempts + 1
				 WHERE status = %d AND reserved_at IS NOT NULL AND reserved_at < %s",
				$table,
				self::PENDING,
				self::RESERVED,
				gmdate( 'Y-m-d H:i:s', time() - $olderThan )
			)
		);

		if ( $recovered > 0 ) {
			Logger::warning( 'queue', 'Recovered stalled jobs.', [ 'count' => $recovered ] );
		}

		return $recovered;
	}

	/**
	 * Might there be work due right now?
	 *
	 * Answered from the autoloaded watermark, so this costs **no query at all** —
	 * which is what lets the shutdown drain be registered on every request without
	 * charging pages that have nothing queued.
	 *
	 * False negatives are impossible: every push lowers the watermark. False
	 * positives are possible and harmless — the claim query simply returns nothing
	 * and {@see self::refreshWatermark()} corrects it.
	 */
	public static function maybeHasDue(): bool {
		$next = (int) get_option( self::WATERMARK_OPTION, 0 );

		return $next > 0 && $next <= time();
	}

	/**
	 * Recompute the watermark from the table.
	 *
	 * Called after a drain, so an emptied queue stops waking the dispatcher and a
	 * queue holding only delayed jobs sleeps until the earliest one is due.
	 */
	public static function refreshWatermark(): void {
		global $wpdb;

		$table = Schema::table( 'jobs' );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$earliest = $wpdb->get_var(
			$wpdb->prepare( 'SELECT MIN(available_at) FROM %i WHERE status = %d', $table, self::PENDING )
		);

		$next = is_string( $earliest ) ? (int) strtotime( $earliest . ' UTC' ) : 0;

		update_option( self::WATERMARK_OPTION, $next, true );
	}

	/**
	 * Move the watermark earlier, never later.
	 *
	 * @param int $timestamp When the newly queued job becomes available.
	 */
	private static function lowerWatermark( int $timestamp ): void {
		$current = (int) get_option( self::WATERMARK_OPTION, 0 );

		if ( 0 === $current || $timestamp < $current ) {
			update_option( self::WATERMARK_OPTION, $timestamp, true );
		}
	}

	/**
	 * Job counts by status, for the health screen.
	 *
	 * @return array{pending:int, reserved:int, failed:int}
	 */
	public static function stats(): array {
		global $wpdb;

		$table = Schema::table( 'jobs' );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$rows = $wpdb->get_results( $wpdb->prepare( 'SELECT status, COUNT(*) AS total FROM %i GROUP BY status', $table ), ARRAY_A );

		$stats = [
			'pending'  => 0,
			'reserved' => 0,
			'failed'   => 0,
		];

		foreach ( (array) $rows as $row ) {
			$stats[ match ( (int) $row['status'] ) {
				self::RESERVED => 'reserved',
				self::FAILED   => 'failed',
				default        => 'pending',
			} ] = (int) $row['total'];
		}

		return $stats;
	}

	/**
	 * Delete parked failures older than the retention window.
	 *
	 * @param int $days Retention in days.
	 *
	 * @return int Rows removed.
	 */
	public static function purgeFailed( int $days = 30 ): int {
		global $wpdb;

		$table = Schema::table( 'jobs' );

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		return (int) $wpdb->query(
			$wpdb->prepare(
				"DELETE FROM %i WHERE status = %d AND created_at < %s",
				$table,
				self::FAILED,
				gmdate( 'Y-m-d H:i:s', time() - ( $days * DAY_IN_SECONDS ) )
			)
		);
	}
}