<?php
/**
 * Cart Recovery — Scheduler (state machine)
 *
 * Drives the drip timing of the cart-recovery flow. Runs on a recurring
 * WP-Cron event (`niwa_ai_cart_recovery_tick`) and moves rows through
 * the state machine without sending anything itself — actual email
 * dispatch is wired in Phase 5 via the `niwa_ai_cart_recovery_send_mail1`
 * / `_mail2` actions this scheduler fires.
 *
 * State transitions owned here:
 *   pending    → scheduled    (email + inactivity ≥ 15 min + not suppressed + not in cooldown)
 *   pending    → expired      (suppressed OR in cooldown — we won't bother them)
 *   sent_mail1 → sent_mail1*  (sets `mail2_scheduled_at` once ≥ 24h after mail1_sent_at)
 *   *          → expired      (any non-terminal row with last_activity_at older than 72h cap)
 *
 * Actions fired (listeners implemented in Phase 5 mailer):
 *   - `Niwa_AI_Constants::ACTION_CART_RECOVERY_SEND_MAIL1` — argument is
 *     the full row array; listener is responsible for composing + sending
 *     the email and flipping status to `sent_mail1` via repository.
 *   - `Niwa_AI_Constants::ACTION_CART_RECOVERY_SEND_MAIL2` — same shape,
 *     but for the second mail (with coupon).
 *
 * Tunable knobs via `apply_filters` (Phase 6 settings page will wire
 * these through admin options):
 *   - `niwa_ai_cart_recovery_mail1_delay_minutes`
 *   - `niwa_ai_cart_recovery_mail2_delay_hours`
 *   - `niwa_ai_cart_recovery_inactivity_cap_hours`
 *   - `niwa_ai_cart_recovery_cooldown_days`
 *   - `niwa_ai_cart_recovery_tick_batch_limit`
 *
 * @package Niwa_AI
 * @since   0.12.0
 */

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

class Niwa_AI_Cart_Recovery_Scheduler {

	private const AS_GROUP = 'niwa-ai-cart-recovery';
	private const OPT_ACTION_SCHEDULER_CLEANUP_VERSION = 'niwa_ai_cart_recovery_as_cleanup_version';
	private const ACTION_SCHEDULER_HISTORY_RETENTION_DAYS = 2;

	/**
	 * Event-driven per-row send actions. Fired by `as_schedule_single_action`
	 * queued from `Cart_Recovery_Service::handle_cart_change()` the moment
	 * a cart row is created or updated — the row_id is the single arg so
	 * handlers can cheap-look up the current state and re-validate guards
	 * at fire time (suppression / cooldown / status may have changed while
	 * the visitor was away).
	 *
	 * Distinct hook names from the recurring `CRON_HOOK_CART_RECOVERY_TICK`
	 * so the AS UI / logs can tell "this row's scheduled send" apart from
	 * "generic 1-minute sweep".
	 */
	public const ACTION_ROW_FIRE_MAIL1 = 'niwa_ai_cart_recovery_row_fire_mail1';
	public const ACTION_ROW_FIRE_MAIL2 = 'niwa_ai_cart_recovery_row_fire_mail2';

	/**
	 * How often the recurring tick fires. Kept at one minute so a visitor
	 * who sets `mail1_delay_minutes = 1` doesn't wait five minutes for
	 * the next poll cycle — the smallest promotion delay we allow is
	 * one minute (enforced by the settings sanitizer), so anything
	 * above that makes the UI setting feel broken.
	 *
	 * Filterable via `niwa_ai_cart_recovery_tick_interval_seconds` for
	 * low-traffic sites that would prefer fewer background wakeups; the
	 * floor is 30 seconds to keep Action Scheduler queue volume sane.
	 */
	private const TICK_INTERVAL_SECONDS = 60;

	/**
	 * WP-option key caching the currently-scheduled runtime + interval.
	 * Any mismatch on boot forces a reschedule, which is how sites pick
	 * up interval changes between plugin versions without a manual
	 * deactivate/reactivate step.
	 */
	private const OPT_TICK_SIGNATURE = 'niwa_ai_cart_recovery_tick_signature';

	public function __construct(
		private readonly Niwa_AI_Cart_Recovery_Repository $repository,
		private readonly ?Niwa_AI_Cart_Recovery_Coupon_Service $coupon_service = null
	) {}

	/**
	 * Register the tick listener. Both Action Scheduler and WP-Cron fire
	 * the same WordPress action (`CRON_HOOK_CART_RECOVERY_TICK`), so a
	 * single `add_action` serves either runtime.
	 */
	public function register_hooks(): void {
		add_action( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK, array( $this, 'tick' ) );

		// Event-driven per-row send handlers. AS passes the row_id as the
		// single argument when the scheduled single action fires.
		add_action( self::ACTION_ROW_FIRE_MAIL1, array( $this, 'fire_row_mail1' ), 10, 1 );
		add_action( self::ACTION_ROW_FIRE_MAIL2, array( $this, 'fire_row_mail2' ), 10, 1 );

		// Any completed reminder (email, WhatsApp, or both) advances the drip.
		add_action( 'niwa_ai_cart_recovery_reminder_sent', array( $this, 'on_reminder_sent' ), 10, 4 );
	}

	/**
	 * Is Action Scheduler available (e.g. WooCommerce active)? Cached
	 * per-request to avoid repeated function_exists calls.
	 */
	public function is_action_scheduler_available(): bool {
		return function_exists( 'as_schedule_recurring_action' )
			&& function_exists( 'as_next_scheduled_action' )
			&& function_exists( 'as_unschedule_all_actions' );
	}

	/**
	 * Returns 'action_scheduler' or 'wp_cron' — useful for admin UI
	 * badges so operators know which runtime is active.
	 */
	public function active_runtime(): string {
		return $this->is_action_scheduler_available() ? 'action_scheduler' : 'wp_cron';
	}

	/**
	 * Ensure the recurring tick is scheduled (idempotent). Uses Action
	 * Scheduler when available (primary, more reliable), WP-Cron as
	 * fallback. Safe to call on every admin boot — both paths guard
	 * against duplicate registration AND detect a runtime/interval
	 * change since the last boot and force a reschedule so sites pick
	 * up faster cadences without a manual deactivate/reactivate dance.
	 */
	public function schedule_tick(): void {
		$interval = $this->desired_interval_seconds();
		$runtime  = $this->active_runtime();

		$signature = $runtime . ':' . $interval;
		$stored    = (string) get_option( self::OPT_TICK_SIGNATURE, '' );

		if ( $stored !== $signature ) {
			// Interval/runtime changed (e.g. v0.12.3 → v0.12.4 drop from
			// 5 min → 1 min). Cancel everything so the block below
			// reschedules at the new cadence.
			$this->unschedule_tick();
			update_option( self::OPT_TICK_SIGNATURE, $signature, false );
		}

		if ( $this->is_action_scheduler_available() ) {
			$this->dedupe_action_scheduler_tick();
			$this->maybe_trim_action_scheduler_history();

			if ( ! $this->has_pending_action_scheduler_tick() ) {
				as_schedule_recurring_action(
					time() + $interval,
					$interval,
					Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK,
					array(),
					self::AS_GROUP
				);
			}
			// If a WP-Cron event was registered by an earlier version,
			// unregister it so we don't double-fire the tick.
			$this->unschedule_wp_cron();
			return;
		}

		if ( ! wp_next_scheduled( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK ) ) {
			wp_schedule_event(
				time() + $interval,
				$this->wp_cron_schedule_for_interval( $interval ),
				Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK
			);
		}
	}

	/**
	 * Desired recurring interval in seconds. Filterable and floored to
	 * 30 seconds so low-traffic sites can trade precision for queue
	 * volume.
	 */
	private function desired_interval_seconds(): int {
		$interval = (int) apply_filters(
			'niwa_ai_cart_recovery_tick_interval_seconds',
			self::TICK_INTERVAL_SECONDS
		);
		return max( 30, $interval );
	}

	/**
	 * Pick the closest WP-Cron schedule slug for a given second count.
	 * WP-Cron can't honor arbitrary intervals (unlike AS), so we snap
	 * to the shortest registered bucket at or above the desired value.
	 */
	private function wp_cron_schedule_for_interval( int $seconds ): string {
		if ( $seconds <= 30 && $this->wp_schedule_exists( 'niwa_ai_thirty_seconds' ) ) {
			return 'niwa_ai_thirty_seconds';
		}
		if ( $seconds <= 60 && $this->wp_schedule_exists( 'niwa_ai_one_minute' ) ) {
			return 'niwa_ai_one_minute';
		}
		if ( $seconds <= 300 && $this->wp_schedule_exists( 'niwa_ai_five_minutes' ) ) {
			return 'niwa_ai_five_minutes';
		}
		return 'hourly';
	}

	private function wp_schedule_exists( string $slug ): bool {
		if ( ! function_exists( 'wp_get_schedules' ) ) {
			return false;
		}
		return array_key_exists( $slug, (array) wp_get_schedules() );
	}

	/**
	 * Tear down the recurring tick. Cancels on both runtimes so leftover
	 * events from a prior activation don't keep firing after deactivation.
	 */
	public function unschedule_tick(): void {
		if ( $this->is_action_scheduler_available() ) {
			as_unschedule_all_actions( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK, array(), self::AS_GROUP );
			self::cancel_pending_action_scheduler_hooks( array( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK ) );
		}
		$this->unschedule_wp_cron();
	}

	/**
	 * Cancel every pending cart-recovery scheduled action. Used by the plugin
	 * deactivation hook so deleting/disabling Niwa cannot leave row mail sends
	 * or recurring ticks behind in WooCommerce Action Scheduler.
	 */
	public static function unschedule_all_cart_recovery_actions( bool $delete_history = false ): void {
		$hooks = self::cart_recovery_action_hooks();

		foreach ( $hooks as $hook ) {
			if ( function_exists( 'as_unschedule_all_actions' ) ) {
				as_unschedule_all_actions( $hook );
				as_unschedule_all_actions( $hook, array(), self::AS_GROUP );
			}
			if ( function_exists( 'wp_clear_scheduled_hook' ) ) {
				wp_clear_scheduled_hook( $hook );
			}
		}

		self::cancel_pending_action_scheduler_hooks( $hooks );

		if ( $delete_history ) {
			self::delete_action_scheduler_history( $hooks, 0 );
		}
	}

	private static function cart_recovery_action_hooks(): array {
		return array(
			Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK,
			self::ACTION_ROW_FIRE_MAIL1,
			self::ACTION_ROW_FIRE_MAIL2,
		);
	}

	private function unschedule_wp_cron(): void {
		$ts = wp_next_scheduled( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK );
		while ( $ts ) {
			if ( false === wp_unschedule_event( $ts, Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK ) ) {
				break;
			}
			$ts = wp_next_scheduled( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK );
		}
	}

	/**
	 * Main entry. Runs the maintenance and delivery sub-phases in order. Each is
	 * wrapped in try/catch so a single bad row doesn't poison the whole
	 * tick — errors are logged via `error_log` when WP_DEBUG is on.
	 */
	public function tick(): void {
		$limit = $this->batch_limit();

		$this->safely( fn() => $this->promote_pending_to_scheduled( $limit ), 'promote_pending_to_scheduled' );
		$this->safely( fn() => $this->promote_sent_mail1_ready_for_mail2( $limit ), 'promote_sent_mail1_ready_for_mail2' );
		$this->safely( fn() => $this->expire_inactive( $limit ), 'expire_inactive' );
		$this->safely( fn() => $this->fire_pending_sends( $limit ), 'fire_pending_sends' );
		$this->safely( fn() => $this->cleanup_expired_coupons( $limit ), 'cleanup_expired_coupons' );
	}

	/**
	 * Trash WC_Coupon posts for recovery rows whose coupons expired
	 * past the grace window. No-op when the coupon service isn't
	 * wired (e.g. legacy instantiation in tests). The grace window
	 * defaults to `CART_RECOVERY_COUPON_CLEANUP_GRACE_DAYS` and is
	 * overridable via `niwa_ai_cart_recovery_coupon_cleanup_grace_days`.
	 */
	public function cleanup_expired_coupons( int $limit = 100 ): int {
		if ( ! $this->coupon_service ) {
			return 0;
		}

		$grace = (int) apply_filters(
			'niwa_ai_cart_recovery_coupon_cleanup_grace_days',
			Niwa_AI_Constants::CART_RECOVERY_COUPON_CLEANUP_GRACE_DAYS
		);

		return $this->coupon_service->cleanup_expired_coupons( max( 0, $grace ), $limit );
	}

	/**
	 * Promote `pending` rows older than the mail_1 delay into `scheduled`,
	 * applying the suppression + cooldown guards. Returns the count of
	 * rows that moved to `scheduled` (suppressed/cooldown rows move to
	 * `expired` instead and are not counted here).
	 */
	public function promote_pending_to_scheduled( int $limit = 50 ): int {
		if ( ! Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail1' ) ) {
			return 0;
		}

		$delay = (int) apply_filters(
			'niwa_ai_cart_recovery_mail1_delay_minutes',
			Niwa_AI_Constants::CART_RECOVERY_MAIL1_DELAY_MIN
		);
		$cooldown_days = (int) apply_filters(
			'niwa_ai_cart_recovery_cooldown_days',
			Niwa_AI_Constants::CART_RECOVERY_COOLDOWN_DAYS
		);

		$rows     = $this->repository->get_pending_due_for_promotion( $delay, $limit );
		$promoted = 0;

		foreach ( $rows as $row ) {
			$id       = (int) ( $row['id'] ?? 0 );
			$channels = Niwa_AI_Cart_Recovery_Settings::delivery_channels_for_row( $row );

			if ( $id <= 0 || empty( $channels ) ) {
				continue;
			}

			if ( empty( $this->eligible_channels( $row, true, $cooldown_days ) ) ) {
				$this->repository->update_status(
					$id,
					Niwa_AI_Cart_Recovery_Repository::STATUS_EXPIRED
				);
				continue;
			}

			$this->repository->update_status(
				$id,
				Niwa_AI_Cart_Recovery_Repository::STATUS_SCHEDULED,
				array( 'mail1_scheduled_at' => $this->now_mysql() )
			);
			$promoted++;
		}

		return $promoted;
	}

	/**
	 * Stamp `mail2_scheduled_at` on rows where mail_1 is ≥ 24h old. The
	 * status stays `sent_mail1` — it flips to `sent_mail2` only after
	 * the Phase 5 mailer actually sends mail_2.
	 */
	public function promote_sent_mail1_ready_for_mail2( int $limit = 50 ): int {
		if ( ! Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail2' ) ) {
			return 0;
		}

		$delay_hours = (int) apply_filters(
			'niwa_ai_cart_recovery_mail2_delay_hours',
			Niwa_AI_Constants::CART_RECOVERY_MAIL2_DELAY_HOURS
		);

		$rows     = $this->repository->get_sent_mail1_due_for_mail2( $delay_hours, $limit );
		$promoted = 0;

		foreach ( $rows as $row ) {
			$id       = (int) ( $row['id'] ?? 0 );
			$channels = Niwa_AI_Cart_Recovery_Settings::delivery_channels_for_row( $row );
			if ( $id <= 0 || empty( $channels ) ) {
				continue;
			}
			if ( empty( $this->eligible_channels( $row ) ) ) {
				$this->repository->update_status( $id, Niwa_AI_Cart_Recovery_Repository::STATUS_EXPIRED );
				continue;
			}
			$this->repository->update_status(
				$id,
				Niwa_AI_Cart_Recovery_Repository::STATUS_SENT_MAIL1,
				array( 'mail2_scheduled_at' => $this->now_mysql() )
			);
			$promoted++;
		}

		return $promoted;
	}

	/**
	 * Close out rows inactive longer than the cap. No mail is sent for
	 * these — the visitor abandoned hard, bothering them now looks
	 * spammy and is far past useful-window anyway.
	 */
	public function expire_inactive( int $limit = 100 ): int {
		$cap = (int) apply_filters(
			'niwa_ai_cart_recovery_inactivity_cap_hours',
			Niwa_AI_Constants::CART_RECOVERY_INACTIVITY_CAP_H
		);

		$rows    = $this->repository->get_expired( $cap, $limit );
		$expired = 0;

		foreach ( $rows as $row ) {
			$id = (int) ( $row['id'] ?? 0 );
			if ( $id <= 0 ) {
				continue;
			}
			$this->repository->update_status(
				$id,
				Niwa_AI_Cart_Recovery_Repository::STATUS_EXPIRED
			);
			$expired++;
		}

		return $expired;
	}

	/**
	 * Fire the per-stage send actions for all rows that are due. The
	 * listener is responsible for composing + sending the mail and
	 * flipping status via the repository. We do NOT flip status here —
	 * otherwise a crash in the mailer would leave the row marked sent
	 * without an actual send.
	 */
	public function fire_pending_sends( int $limit = 50 ): int {
		// Dedicated send cap — tighter than the generic batch limit. We
		// don't want a cold install accidentally firing 50 emails in
		// the first tick; the cap is filterable for ops.
		$send_cap = (int) apply_filters(
			'niwa_ai_cart_recovery_send_max_per_tick',
			Niwa_AI_Constants::CART_RECOVERY_SEND_MAX_PER_TICK
		);
		$send_cap = max( 1, min( $limit, $send_cap ) );

		$fired = 0;

		// mail1 first — quota is shared across both stages.
		if ( Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail1' ) ) {
			foreach ( $this->repository->get_due_for_stage( 'mail1', $send_cap ) as $row ) {
				if ( $fired >= $send_cap ) {
					break;
				}
				if ( empty( $this->eligible_channels( $row ) ) ) {
					continue;
				}
				do_action( Niwa_AI_Constants::ACTION_CART_RECOVERY_SEND_MAIL1, $row );
				$fired++;
			}
		}

		// mail2 consumes whatever quota remains. Fetching only the
		// remaining slot count prevents over-reading rows we can't
		// use this tick.
		$remaining = $send_cap - $fired;
		if ( $remaining <= 0 ) {
			return $fired;
		}
		if ( Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail2' ) ) {
			foreach ( $this->repository->get_due_for_stage( 'mail2', $remaining ) as $row ) {
				if ( $fired >= $send_cap ) {
					break;
				}
				if ( empty( $this->eligible_channels( $row ) ) ) {
					continue;
				}
				do_action( Niwa_AI_Constants::ACTION_CART_RECOVERY_SEND_MAIL2, $row );
				$fired++;
			}
		}

		return $fired;
	}

	private function batch_limit(): int {
		return max(
			1,
			(int) apply_filters( 'niwa_ai_cart_recovery_tick_batch_limit', 50 )
		);
	}

	/** @return string[] */
	private function eligible_channels( array $row, bool $check_cooldown = false, int $cooldown_days = 0 ): array {
		$channels = Niwa_AI_Cart_Recovery_Settings::delivery_channels_for_row( $row );
		return array_values(
			array_filter(
				$channels,
				function ( string $channel ) use ( $row, $check_cooldown, $cooldown_days ): bool {
					if ( Niwa_AI_Cart_Recovery_Settings::CHANNEL_EMAIL === $channel ) {
						$email = (string) ( $row['email'] ?? '' );
						return ! $this->repository->is_suppressed( $email )
							&& ( ! $check_cooldown || ! $this->repository->is_in_cooldown( $email, $cooldown_days ) );
					}
					if ( Niwa_AI_Cart_Recovery_Settings::CHANNEL_WHATSAPP === $channel ) {
						$phone = (string) ( $row['phone'] ?? '' );
						return ! $this->repository->is_phone_suppressed( $phone )
							&& ( ! $check_cooldown || ! $this->repository->is_phone_in_cooldown( $phone, $cooldown_days ) );
					}
					return false;
				}
			)
		);
	}

	private function now_mysql(): string {
		if ( function_exists( 'current_time' ) ) {
			return (string) current_time( 'mysql' );
		}
		return gmdate( 'Y-m-d H:i:s' );
	}

	/* ===============================================================
	 * Event-driven per-row send pipeline
	 * ============================================================ */

	/**
	 * Queue a precise single-fire action for this row's mail1 exactly
	 * `mail1_delay_minutes` from now. Called from
	 * `Cart_Recovery_Service::handle_cart_change()` right after
	 * `upsert_snapshot` — so the visitor sees mail1 the moment their
	 * chosen delay elapses, not whenever the recurring 1-minute sweep
	 * next runs.
	 *
	 * Cancels any pre-existing schedule for the same row first —
	 * subsequent cart activity resets the inactivity clock, which is
	 * the standard semantics for abandoned-cart tools.
	 *
	 * No-op when the row isn't eligible (already past mail1, missing
	 * email, feature flag off) — those guards repeat at fire time so
	 * cancellation of a stale schedule is fine.
	 */
	public function schedule_row_mail1( int $row_id ): void {
		if ( $row_id <= 0 ) {
			return;
		}

		// Cancel any previously-queued mail1 for this row so the delay
		// clock restarts from the latest activity.
		$this->unschedule_row_action( self::ACTION_ROW_FIRE_MAIL1, $row_id );
		if ( ! Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail1' ) ) {
			return;
		}

		$row = $this->repository->find_by_id( $row_id );
		if ( ! is_array( $row ) ) {
			return;
		}

		$status = (string) ( $row['status'] ?? '' );
		if ( Niwa_AI_Cart_Recovery_Repository::STATUS_PENDING !== $status ) {
			return;
		}

		if ( empty( $this->eligible_channels( $row ) ) ) {
			return;
		}

		$delay_minutes = (int) apply_filters(
			'niwa_ai_cart_recovery_mail1_delay_minutes',
			Niwa_AI_Constants::CART_RECOVERY_MAIL1_DELAY_MIN
		);
		$delay_seconds = max( 1, $delay_minutes ) * MINUTE_IN_SECONDS;
		$when          = time() + $delay_seconds;

		$this->queue_single_action( self::ACTION_ROW_FIRE_MAIL1, $row_id, $when );
	}

	/**
	 * Schedule mail2 for a row that just finished mail1. Called from
	 * `on_mail_sent` when the Mailer announces a successful mail1 send.
	 */
	public function schedule_row_mail2( int $row_id ): void {
		if ( $row_id <= 0 ) {
			return;
		}

		$this->unschedule_row_action( self::ACTION_ROW_FIRE_MAIL2, $row_id );
		if ( ! Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail2' ) ) {
			return;
		}

		$row = $this->repository->find_by_id( $row_id );
		if ( ! is_array( $row ) ) {
			return;
		}

		$status = (string) ( $row['status'] ?? '' );
		if ( Niwa_AI_Cart_Recovery_Repository::STATUS_SENT_MAIL1 !== $status ) {
			return;
		}

		if ( empty( $this->eligible_channels( $row ) ) ) {
			return;
		}

		$delay_hours = (int) apply_filters(
			'niwa_ai_cart_recovery_mail2_delay_hours',
			Niwa_AI_Constants::CART_RECOVERY_MAIL2_DELAY_HOURS
		);
		$delay_seconds = max( 1, $delay_hours ) * HOUR_IN_SECONDS;
		$when          = time() + $delay_seconds;

		$this->queue_single_action( self::ACTION_ROW_FIRE_MAIL2, $row_id, $when );
	}

	/**
	 * AS callback for per-row mail1. Re-checks every guard the recurring
	 * tick would have applied (the row may have been recovered / expired
	 * / unsubscribed / suppressed while the schedule slot was waiting),
	 * then either flips to `scheduled` + fires `ACTION_CART_RECOVERY_SEND_MAIL1`
	 * (mailer picks it up) or short-circuits to `expired` when a guard
	 * kicks in. On any non-terminal miss the row still gets picked up
	 * by the recurring safety-net tick, so no send is ever permanently
	 * lost.
	 */
	public function fire_row_mail1( $row_id ): void {
		$row_id = (int) $row_id;
		if ( $row_id <= 0 ) {
			return;
		}

		try {
			if ( ! Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail1' ) ) {
				return;
			}
			$row = $this->repository->find_by_id( $row_id );
			if ( ! is_array( $row ) ) {
				return;
			}
			if ( Niwa_AI_Cart_Recovery_Repository::STATUS_PENDING !== (string) ( $row['status'] ?? '' ) ) {
				return;
			}
			$channels = Niwa_AI_Cart_Recovery_Settings::delivery_channels_for_row( $row );
			if ( empty( $channels ) ) {
				return;
			}

			$cooldown_days = (int) apply_filters(
				'niwa_ai_cart_recovery_cooldown_days',
				Niwa_AI_Constants::CART_RECOVERY_COOLDOWN_DAYS
			);
			if ( empty( $this->eligible_channels( $row, true, $cooldown_days ) ) ) {
				$this->repository->update_status( $row_id, Niwa_AI_Cart_Recovery_Repository::STATUS_EXPIRED );
				return;
			}

			$this->repository->update_status(
				$row_id,
				Niwa_AI_Cart_Recovery_Repository::STATUS_SCHEDULED,
				array( 'mail1_scheduled_at' => $this->now_mysql() )
			);

			// Re-read so the mailer sees the fresh status + scheduled_at.
			$refreshed = $this->repository->find_by_id( $row_id );
			if ( is_array( $refreshed ) ) {
				do_action( Niwa_AI_Constants::ACTION_CART_RECOVERY_SEND_MAIL1, $refreshed );
			}
		} catch ( \Throwable $e ) {
			if ( defined( 'WP_DEBUG' ) && WP_DEBUG ) {
				error_log( '[niwa-ai][cart-recovery][scheduler.fire_row_mail1] ' . $e->getMessage() ); // phpcs:ignore WordPress.PHP.DevelopmentFunctions.error_log_error_log
			}
		}
	}

	/**
	 * AS callback for per-row mail2. Same guard-and-fire pattern as
	 * `fire_row_mail1` but for the follow-up stage.
	 */
	public function fire_row_mail2( $row_id ): void {
		$row_id = (int) $row_id;
		if ( $row_id <= 0 ) {
			return;
		}

		try {
			if ( ! Niwa_AI_Cart_Recovery_Settings::is_stage_enabled( 'mail2' ) ) {
				return;
			}
			$row = $this->repository->find_by_id( $row_id );
			if ( ! is_array( $row ) ) {
				return;
			}
			if ( Niwa_AI_Cart_Recovery_Repository::STATUS_SENT_MAIL1 !== (string) ( $row['status'] ?? '' ) ) {
				return;
			}
			$channels = Niwa_AI_Cart_Recovery_Settings::delivery_channels_for_row( $row );
			if ( empty( $channels ) ) {
				return;
			}

			if ( empty( $this->eligible_channels( $row ) ) ) {
				$this->repository->update_status( $row_id, Niwa_AI_Cart_Recovery_Repository::STATUS_EXPIRED );
				return;
			}

			$this->repository->update_status(
				$row_id,
				Niwa_AI_Cart_Recovery_Repository::STATUS_SENT_MAIL1,
				array( 'mail2_scheduled_at' => $this->now_mysql() )
			);

			$refreshed = $this->repository->find_by_id( $row_id );
			if ( is_array( $refreshed ) ) {
				do_action( Niwa_AI_Constants::ACTION_CART_RECOVERY_SEND_MAIL2, $refreshed );
			}
		} catch ( \Throwable $e ) {
			if ( defined( 'WP_DEBUG' ) && WP_DEBUG ) {
				error_log( '[niwa-ai][cart-recovery][scheduler.fire_row_mail2] ' . $e->getMessage() ); // phpcs:ignore WordPress.PHP.DevelopmentFunctions.error_log_error_log
			}
		}
	}

	/**
	 * Listener on `niwa_ai_cart_recovery_reminder_sent`. Bridges mail1 →
	 * scheduled mail2 without waiting for the next recurring poll.
	 *
	 * @param string $stage
	 * @param array  $row
	 * @param array  $built
	 */
	public function on_reminder_sent( $stage, $row, $built, $channels = array() ): void {
		unset( $built, $channels );
		if ( Niwa_AI_Cart_Recovery_Composition_Context::STAGE_MAIL1 !== (string) $stage ) {
			return;
		}
		if ( ! is_array( $row ) ) {
			return;
		}
		$row_id = (int) ( $row['id'] ?? 0 );
		if ( $row_id <= 0 ) {
			return;
		}
		$this->schedule_row_mail2( $row_id );
	}

	/**
	 * Runtime-aware wrapper around single-action scheduling.
	 */
	private function queue_single_action( string $hook, int $row_id, int $when ): void {
		if ( $this->is_action_scheduler_available() ) {
			as_schedule_single_action( $when, $hook, array( $row_id ), self::AS_GROUP );
			return;
		}
		wp_schedule_single_event( $when, $hook, array( $row_id ) );
	}

	/**
	 * Cancel a previously-queued per-row action, if any. Safe to call
	 * when no schedule exists — both backends no-op on a miss.
	 */
	private function unschedule_row_action( string $hook, int $row_id ): void {
		if ( $this->is_action_scheduler_available() ) {
			if ( function_exists( 'as_unschedule_all_actions' ) ) {
				as_unschedule_all_actions( $hook, array( $row_id ), self::AS_GROUP );
				as_unschedule_all_actions( $hook, array( $row_id ) );
			}
			self::cancel_pending_action_scheduler_hooks( array( $hook ), array( $row_id ) );
			return;
		}
		$ts = wp_next_scheduled( $hook, array( $row_id ) );
		while ( $ts ) {
			wp_unschedule_event( $ts, $hook, array( $row_id ) );
			$ts = wp_next_scheduled( $hook, array( $row_id ) );
		}
	}

	/**
	 * Run a sub-phase and swallow exceptions so one bad batch doesn't
	 * kill the tick. Logs via `error_log` when WP_DEBUG is on.
	 */
	private function safely( callable $phase, string $label ): void {
		try {
			$phase();
		} catch ( \Throwable $e ) {
			if ( defined( 'WP_DEBUG' ) && WP_DEBUG ) {
				error_log( sprintf( // phpcs:ignore WordPress.PHP.DevelopmentFunctions.error_log_error_log
					'[niwa-ai][cart-recovery][scheduler.%s] %s',
					$label,
					$e->getMessage()
				) );
			}
		}
	}

	private function has_pending_action_scheduler_tick(): bool {
		if ( self::action_scheduler_tables() ) {
			return self::count_pending_action_scheduler_hooks( array( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK ) ) > 0;
		}

		if ( function_exists( 'as_next_scheduled_action' ) ) {
			return (bool) as_next_scheduled_action( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK, array(), self::AS_GROUP )
				|| (bool) as_next_scheduled_action( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK, array() );
		}

		return false;
	}

	private function dedupe_action_scheduler_tick(): void {
		$ids = self::pending_action_scheduler_ids( array( Niwa_AI_Constants::CRON_HOOK_CART_RECOVERY_TICK ) );
		if ( count( $ids ) <= 1 ) {
			return;
		}

		array_shift( $ids );
		self::cancel_action_scheduler_ids( $ids );
	}

	private function maybe_trim_action_scheduler_history(): void {
		$stored = (string) get_option( self::OPT_ACTION_SCHEDULER_CLEANUP_VERSION, '' );
		if ( defined( 'NIWA_AI_VERSION' ) && $stored === NIWA_AI_VERSION ) {
			return;
		}

		self::delete_action_scheduler_history(
			self::cart_recovery_action_hooks(),
			self::ACTION_SCHEDULER_HISTORY_RETENTION_DAYS
		);

		if ( defined( 'NIWA_AI_VERSION' ) ) {
			update_option( self::OPT_ACTION_SCHEDULER_CLEANUP_VERSION, NIWA_AI_VERSION, false );
		}
	}

	private static function count_pending_action_scheduler_hooks( array $hooks ): int {
		$tables = self::action_scheduler_tables();
		if ( ! $tables ) {
			return 0;
		}

		global $wpdb;
		$sql = self::prepare_in(
			"SELECT COUNT(*) FROM {$tables['actions']} WHERE hook IN (%s) AND status = 'pending'",
			$hooks
		);

		return (int) $wpdb->get_var( $sql );
	}

	private static function pending_action_scheduler_ids( array $hooks, ?array $args = null ): array {
		$tables = self::action_scheduler_tables();
		if ( ! $tables ) {
			return array();
		}

		global $wpdb;
		$args_clause = '';
		$params      = $hooks;
		if ( null !== $args ) {
			$args_clause = ' AND args = %s';
			$params[]    = function_exists( 'wp_json_encode' )
				? wp_json_encode( array_values( $args ) )
				: json_encode( array_values( $args ) );
		}

		$sql = self::prepare_in(
			"SELECT action_id FROM {$tables['actions']} WHERE hook IN (%s) AND status = 'pending'{$args_clause} ORDER BY scheduled_date_gmt ASC, action_id ASC",
			$params,
			count( $hooks )
		);

		return array_map( 'intval', (array) $wpdb->get_col( $sql ) );
	}

	private static function cancel_pending_action_scheduler_hooks( array $hooks, ?array $args = null ): void {
		self::cancel_action_scheduler_ids( self::pending_action_scheduler_ids( $hooks, $args ) );
	}

	private static function cancel_action_scheduler_ids( array $ids ): void {
		$tables = self::action_scheduler_tables();
		$ids    = array_values( array_filter( array_map( 'intval', $ids ) ) );
		if ( ! $tables || empty( $ids ) ) {
			return;
		}

		global $wpdb;
		$wpdb->query(
			"UPDATE {$tables['actions']} SET status = 'canceled' WHERE status = 'pending' AND action_id IN (" . implode( ',', $ids ) . ')'
		);
	}

	private static function delete_action_scheduler_history( array $hooks, int $older_than_days ): void {
		$tables = self::action_scheduler_tables();
		if ( ! $tables ) {
			return;
		}

		global $wpdb;
		$cutoff_clause = '';
		$params        = $hooks;
		if ( $older_than_days > 0 ) {
			$day_seconds  = defined( 'DAY_IN_SECONDS' ) ? DAY_IN_SECONDS : 86400;
			$cutoff_clause = ' AND scheduled_date_gmt < %s';
			$params[]      = gmdate( 'Y-m-d H:i:s', time() - ( $older_than_days * $day_seconds ) );
		}

		$select_sql = self::prepare_in(
			"SELECT action_id FROM {$tables['actions']} WHERE hook IN (%s) AND status IN ('complete','failed','canceled'){$cutoff_clause}",
			$params,
			count( $hooks )
		);
		$ids        = array_map( 'intval', (array) $wpdb->get_col( $select_sql ) );
		if ( empty( $ids ) ) {
			return;
		}

		$id_list = implode( ',', $ids );
		if ( ! empty( $tables['logs'] ) ) {
			$wpdb->query( "DELETE FROM {$tables['logs']} WHERE action_id IN ({$id_list})" );
		}
		$wpdb->query( "DELETE FROM {$tables['actions']} WHERE action_id IN ({$id_list})" );
	}

	private static function action_scheduler_tables(): ?array {
		if ( ! isset( $GLOBALS['wpdb'] ) || ! is_object( $GLOBALS['wpdb'] ) ) {
			return null;
		}

		global $wpdb;
		if ( empty( $wpdb->prefix ) || ! method_exists( $wpdb, 'get_var' ) || ! method_exists( $wpdb, 'prepare' ) ) {
			return null;
		}

		$actions = $wpdb->prefix . 'actionscheduler_actions';
		$logs    = $wpdb->prefix . 'actionscheduler_logs';
		$exists  = $wpdb->get_var( $wpdb->prepare( 'SHOW TABLES LIKE %s', $actions ) );
		if ( $exists !== $actions ) {
			return null;
		}

		return array(
			'actions' => $actions,
			'logs'    => $logs,
		);
	}

	private static function prepare_in( string $sql, array $params, ?int $in_count = null ): string {
		global $wpdb;

		$in_count     = null === $in_count ? count( $params ) : $in_count;
		$placeholders = implode( ',', array_fill( 0, $in_count, '%s' ) );
		$sql          = preg_replace( '/%s/', $placeholders, $sql, 1 );

		return $wpdb->prepare( $sql, $params );
	}
}
