<?php
/**
 * The rules every long-running job obeys: one runner at a time, a time budget
 * per tick, a failure that never ends the run, and a resumer for jobs whose
 * browser tab went away.
 *
 * @package AiRankSuite
 */

declare(strict_types=1);

namespace AiRankSuite\Support;

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

/**
 * The audit, the link scan and the bulk queue each keep their own job state
 * (a small option with a cursor) and each advance one slice per HTTP tick.
 * What they shared before this class was nothing — so two tabs could tick
 * the same job at once, an exception in one slice surfaced as a 500 to the
 * browser, and a job whose tab was closed stayed "running" forever.
 *
 * guard() wraps a tick: it takes a short lock (a transient, refused while
 * held), runs the slice inside try/catch, records a failure with an
 * exponential back-off instead of throwing, and releases the lock. Ticks stay
 * exactly what they were; the modules only wrap them.
 *
 * resume() runs from cron every five minutes: any registered job that says it
 * is running but has not ticked in ten minutes gets one tick from the server,
 * so a closed browser slows a run down instead of stopping it.
 */
final class Job {

	public const HOOK = 'airanksuite_job_resume';

	/** Seconds a lock is held at most; a crashed request cannot wedge a job. */
	private const LOCK_TTL = 90;

	/** After this many seconds without a tick a running job is considered abandoned. */
	private const STALE_AFTER = 10 * MINUTE_IN_SECONDS;

	/** @var array<string, array{state: callable, tick: callable}> */
	private static array $registry = array();

	/**
	 * Register a job for the resumer.
	 *
	 * @param string   $name  Job name (audit, linkcheck, bulk, ...).
	 * @param callable $state fn(): array{state:string, ...} — the job's current state.
	 * @param callable $tick  fn(): mixed — advance one slice.
	 */
	public static function register( string $name, callable $state, callable $tick ): void {
		self::$registry[ $name ] = array( 'state' => $state, 'tick' => $tick );
	}

	/**
	 * Run one tick under the job's rules.
	 *
	 * @param string   $name Job name.
	 * @param callable $tick fn(): mixed — the slice.
	 * @return mixed The tick's return value, or the current state with
	 *               'busy' => true when another runner holds the lock, or
	 *               with 'error' set when the slice threw.
	 */
	public static function guard( string $name, callable $tick ) {
		$lock = 'airanksuite_job_lock_' . sanitize_key( $name );
		if ( ! self::acquire( $lock ) ) {
			return self::busy( $name );
		}

		$meta = self::meta( $name );
		if ( ! empty( $meta['retry_after'] ) && time() < (int) $meta['retry_after'] ) {
			self::release( $lock );
			return self::busy( $name, sprintf( 'backing off until %s after %d failures', gmdate( 'H:i:s', (int) $meta['retry_after'] ), (int) $meta['failures'] ) );
		}

		try {
			self::begin_tick();
			$result = $tick();
			self::meta( $name, array( 'last_tick' => time(), 'failures' => 0, 'retry_after' => 0, 'last_error' => '' ) );
			return $result;
		} catch ( \Throwable $e ) {
			$failures = (int) ( $meta['failures'] ?? 0 ) + 1;
			// 5s, 10s, 20s, 40s … capped at ten minutes: transient trouble
			// (a slow DB, a rate limit) clears itself; a persistent fault
			// stops hammering without ending the run.
			$delay = min( 600, 5 * ( 2 ** min( 7, $failures - 1 ) ) );
			self::meta( $name, array( 'last_tick' => time(), 'failures' => $failures, 'retry_after' => time() + $delay, 'last_error' => mb_substr( $e->getMessage(), 0, 300 ) ) );
			$state = self::state_of( $name );
			return array_merge( is_array( $state ) ? $state : array(), array( 'error' => $e->getMessage(), 'failures' => $failures, 'retry_in' => $delay ) );
		} finally {
			self::release( $lock );
			self::end_tick();
		}
	}

	/**
	 * Take the lock, atomically. A transient is get-then-set, which two
	 * requests can both pass; the object cache's "add" and an INSERT IGNORE
	 * into the options table cannot. A lock older than LOCK_TTL is a crashed
	 * request's and may be taken over.
	 */
	private static function acquire( string $lock ): bool {
		global $wpdb;
		$now = time();
		if ( wp_using_ext_object_cache() ) {
			// add() fails when the key exists — the atomic primitive we want.
			return (bool) wp_cache_add( $lock, (string) $now, 'airanksuite_locks', self::LOCK_TTL );
		}
		// phpcs:disable WordPress.DB.DirectDatabaseQuery
		$n = $wpdb->query( $wpdb->prepare( "INSERT IGNORE INTO {$wpdb->options} (option_name, option_value, autoload) VALUES (%s, %s, 'no')", $lock, (string) $now ) );
		if ( 1 === (int) $n ) {
			return true;
		}
		// Held. Take over only when the holder is older than the TTL — and
		// only if nobody else took it over in between (the WHERE is the CAS).
		$held = (int) $wpdb->get_var( $wpdb->prepare( "SELECT option_value FROM {$wpdb->options} WHERE option_name = %s", $lock ) );
		if ( $held > 0 && $now - $held > self::LOCK_TTL ) {
			$taken = $wpdb->query( $wpdb->prepare( "UPDATE {$wpdb->options} SET option_value = %s WHERE option_name = %s AND option_value = %s", (string) $now, $lock, (string) $held ) );
			return 1 === (int) $taken;
		}
		// phpcs:enable
		return false;
	}

	/** Is a runner holding this job's lock right now (a live one, not a stale leftover)? */
	public static function locked( string $name ): bool {
		global $wpdb;
		$lock = 'airanksuite_job_lock_' . sanitize_key( $name );
		if ( wp_using_ext_object_cache() ) {
			return false !== wp_cache_get( $lock, 'airanksuite_locks' );
		}
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery
		$held = (int) $wpdb->get_var( $wpdb->prepare( "SELECT option_value FROM {$wpdb->options} WHERE option_name = %s", $lock ) );
		return $held > 0 && time() - $held <= self::LOCK_TTL;
	}

	private static function release( string $lock ): void {
		global $wpdb;
		if ( wp_using_ext_object_cache() ) {
			wp_cache_delete( $lock, 'airanksuite_locks' );
			return;
		}
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery
		$wpdb->query( $wpdb->prepare( "DELETE FROM {$wpdb->options} WHERE option_name = %s", $lock ) );
	}

	/**
	 * A time budget for a slice: the caller loops while budget_left() > 0.
	 * Uses 60% of max_execution_time, capped at 20 seconds — a shared host's
	 * 30-second limit and a proxy's 60-second timeout both stay far away.
	 */
	public static function budget_seconds(): int {
		$max = (int) ini_get( 'max_execution_time' );
		$max = $max > 0 ? $max : 30;
		/**
		 * Filter the per-tick time budget in seconds.
		 *
		 * @param int $seconds Budget (5–20 by default).
		 */
		return max( 0, (int) apply_filters( 'airanksuite_job_budget_seconds', (int) max( 5, min( 20, floor( $max * 0.6 ) ) ) ) );
	}

	/**
	 * Cron: tick every registered job that is running but has not been ticked
	 * for a while. One slice each — the goal is progress, not speed.
	 *
	 * @return array<string, string> What was done, per job.
	 */
	public static function resume(): array {
		$done = array();
		foreach ( self::$registry as $name => $job ) {
			$state = self::state_of( $name );
			if ( ! is_array( $state ) || ( $state['state'] ?? '' ) !== 'running' ) {
				continue;
			}
			$meta = self::meta( $name );
			$last = (int) ( $meta['last_tick'] ?? 0 );
			if ( $last > 0 && time() - $last < self::STALE_AFTER ) {
				$done[ $name ] = 'active';
				continue;
			}
			self::guard( $name, $job['tick'] );
			$done[ $name ] = 'resumed';
		}
		return $done;
	}

	/** Schedule the resumer. Idempotent. */
	public static function boot(): void {
		add_action( self::HOOK, array( self::class, 'resume' ) );
		Cron::ensure( self::HOOK, 'airanksuite_five_minutes', time() + 5 * MINUTE_IN_SECONDS );
	}

	/**
	 * Memory in use when the current tick began; -1 before the first call.
	 */
	private static int $tick_baseline = -1;

	/**
	 * The PHP memory limit in bytes; 0 when unlimited.
	 */
	public static function memory_limit(): int {
		$raw = trim( (string) ini_get( 'memory_limit' ) );
		if ( '' === $raw || '-1' === $raw ) {
			return 0;
		}
		$unit  = strtolower( substr( $raw, -1 ) );
		$value = (float) $raw;
		$mult  = match ( $unit ) {
			'g'     => 1024 * 1024 * 1024,
			'm'     => 1024 * 1024,
			'k'     => 1024,
			default => 1,
		};
		/**
		 * Filter the memory limit a job budgets against.
		 *
		 * @param int $bytes Limit in bytes; 0 = unlimited.
		 */
		return (int) apply_filters( 'airanksuite/job/memory_limit', (int) ( $value * $mult ) );
	}

	/**
	 * Mark the start of a slice. guard() calls this; a job that ticks itself
	 * outside the guard should too.
	 */
	public static function begin_tick(): void {
		self::$tick_baseline = memory_get_usage( true );
	}

	/**
	 * Forget this tick's baseline.
	 *
	 * The baseline is static and was never cleared, so it outlived the tick
	 * that set it. A second job in the same request — the cron resumer runs
	 * several, and any admin request can tick two — measured its growth
	 * against a number taken before the first job allocated anything. On the
	 * second job `grown` was therefore the first job's memory too, which
	 * crosses the "half the headroom" line immediately: the slice yielded
	 * after a single record and reported the rest as remaining.
	 *
	 * Observed as a migration that moved 1 of 60 rows per slice. guard()
	 * clears it now, and a job that ticks itself outside the guard should
	 * call this when it is done.
	 */
	public static function end_tick(): void {
		self::$tick_baseline = -1;
	}

	/**
	 * A batch size scaled to the memory this host actually has.
	 *
	 * Shared hosting with `memory_limit = 40M` is real (setupco.ae runs on it):
	 * WordPress and the theme are most of that before a job starts, so a slice
	 * sized for a 256M host is how a bulk run turns into a fatal. Below 64M we
	 * take a quarter of the batch, below 128M a half, above 256M we allow a
	 * larger one — the job still finishes, it just takes more ticks.
	 *
	 * @param int $batch The comfortable batch size.
	 * @return int At least 1.
	 */
	public static function batch_size( int $batch ): int {
		$limit = self::memory_limit();
		if ( $limit <= 0 || $limit >= 512 * 1024 * 1024 ) {
			$factor = 1.5;
		} elseif ( $limit >= 256 * 1024 * 1024 ) {
			$factor = 1.0;
		} elseif ( $limit >= 128 * 1024 * 1024 ) {
			$factor = 0.5;
		} elseif ( $limit >= 64 * 1024 * 1024 ) {
			$factor = 0.35;
		} else {
			$factor = 0.25;
		}
		/**
		 * Filter the memory-scaled batch size.
		 *
		 * @param int $size  Scaled size.
		 * @param int $batch Requested size.
		 * @param int $limit Memory limit in bytes (0 = unlimited).
		 */
		return max( 1, (int) apply_filters( 'airanksuite/job/batch_size', max( 1, (int) floor( $batch * $factor ) ), $batch, $limit ) );
	}

	/**
	 * True when this slice should stop and let the next tick continue.
	 *
	 * The question is not "is this host small" but "has this tick grown
	 * enough to be dangerous". Measuring absolute usage against 80% of the
	 * limit reads as permanently tight on exactly the hosts that matter — 40
	 * MB with WordPress, a theme and a page builder already at 34 MB — and a
	 * slice that stops after one record turns a 15,000-record job into 15,000
	 * browser round-trips. So a tick stops when it is genuinely close to the
	 * ceiling (90%), or when it has both passed 80% *and* allocated a tenth of
	 * the limit since it started. A tick that allocates nothing runs on.
	 */
	public static function memory_tight(): bool {
		$limit = self::memory_limit();
		if ( $limit <= 0 ) {
			return false;
		}
		$used     = memory_get_usage( true );
		$baseline = self::$tick_baseline >= 0 ? self::$tick_baseline : $used;
		$headroom = max( 0, $limit - $baseline );
		$grown    = max( 0, $used - $baseline );
		// Half of the room this tick started with is a fair share: it leaves
		// as much again for the rest of the request (a template render, an
		// object cache, the response) and it scales — 2 MB of headroom on a
		// full 40 MB host, 90 MB on a comfortable one.
		if ( $grown >= (int) ( $headroom * 0.5 ) ) {
			return true;
		}
		// And an absolute backstop, for the request that grew somewhere else —
		// but only once this tick has allocated at all. On a host whose
		// baseline already sits above 95% of the limit an unguarded backstop
		// stops every tick before its first record, which is the
		// 15,000-round-trips bug in a new coat. A tick that has not grown
		// runs; the moment it allocates near the ceiling, it yields.
		return $grown > 0 && $used > (int) ( $limit * 0.95 );
	}

	/** @return array<string, mixed>|null */
	private static function state_of( string $name ): ?array {
		if ( ! isset( self::$registry[ $name ] ) ) {
			return null;
		}
		try {
			$state = ( self::$registry[ $name ]['state'] )();
			return is_array( $state ) ? $state : null;
		} catch ( \Throwable $e ) {
			return null;
		}
	}

	/** @return array<string, mixed> */
	private static function busy( string $name, string $why = 'another request is ticking this job' ): array {
		$state = self::state_of( $name );
		return array_merge( is_array( $state ) ? $state : array( 'state' => 'running' ), array( 'busy' => true, 'why' => $why ) );
	}

	/**
	 * Small per-job bookkeeping (last tick, failures, back-off), in one option
	 * for all jobs.
	 *
	 * @param string                    $name  Job name.
	 * @param array<string, mixed>|null $patch Values to store, or null to read.
	 * @return array<string, mixed>
	 */
	public static function meta( string $name, ?array $patch = null ): array {
		$all = get_option( 'airanksuite_job_meta', array() );
		$all = is_array( $all ) ? $all : array();
		if ( null !== $patch ) {
			$all[ $name ] = array_merge( (array) ( $all[ $name ] ?? array() ), $patch );
			update_option( 'airanksuite_job_meta', $all, false );
		}
		return (array) ( $all[ $name ] ?? array() );
	}
}