qdr.app.woo.plugins/studiou-wc-mail-queue/includes/class-wcmq-worker.php

<?php
/**
 * Action Scheduler worker: claims a row and sends it.
 *
 * @package studiou-wc-mail-queue
 * @copyright 2026 QUADARAX
 * @author    Dalibor Votruba <dvotruba@quadarax.com>
 * @license   GPL-2.0-or-later
 */

defined( 'ABSPATH' ) || exit;

/**
 * Class Studiou_WCMQ_Worker
 */
class Studiou_WCMQ_Worker {

	/**
	 * Last WP_Error captured from wp_mail_failed, for the row in flight.
	 *
	 * @var string|null
	 */
	private $last_error = null;

	/**
	 * Constructor.
	 *
	 * The wp_mail_failed listener is registered ONCE here, never per send.
	 * Under Action Scheduler many send() calls run in one PHP process; a
	 * per-send add_action leaks a closure per row, and an early `return` that
	 * skips the paired remove_action leaves a stale WP_Error in the buffer to be
	 * misattributed to the next row. Resetting the buffer before each send (in
	 * send(), step 3) makes both failure modes structurally impossible.
	 */
	public function __construct() {
		add_action( Studiou_WCMQ_Queue::HOOK, array( $this, 'send' ), 10, 1 );
		add_action( 'wp_mail_failed', array( $this, 'capture' ), 10, 1 );
	}

	/**
	 * Capture a mail error.
	 *
	 * @param WP_Error $error Error.
	 */
	public function capture( $error ) {
		if ( is_wp_error( $error ) ) {
			$this->last_error = $error->get_error_message();
		}
	}

	/**
	 * Send one queued mail.
	 *
	 * @param int $row_id Queue row id.
	 */
	public function send( $row_id ) {
		$row_id = (int) $row_id;
		if ( ! $row_id ) {
			return;
		}

		// The reaper runs before we claim OUR row, which looks unsafe: it hunts
		// pending rows with no scheduled action, and our row is pending and about
		// to run. It does not misfire, but only by a coincidence of Action
		// Scheduler internals — as_has_scheduled_action() matches STATUS_RUNNING
		// as well as STATUS_PENDING (packages/action-scheduler/functions.php:404),
		// and AS calls log_execution() (which sets RUNNING) before invoking this
		// hook (ActionScheduler_Abstract_QueueRunner.php:102). So our own row is
		// seen as running, not as an orphan. If AS ever changes that ordering,
		// this reaper call would duplicate the running row's action — revisit here.
		Studiou_WCMQ_Reaper::run();

		// Atomic claim. Zero affected rows means someone else has it, or it is
		// already sent. This is the duplicate-send guard.
		if ( ! Studiou_WCMQ_DB::claim( $row_id ) ) {
			return;
		}

		$row = Studiou_WCMQ_DB::get_row( $row_id );
		if ( ! $row ) {
			return;
		}

		$payload = json_decode( (string) $row->payload, true );
		if ( ! is_array( $payload ) || JSON_ERROR_NONE !== json_last_error()
			|| ! isset( $payload['to'], $payload['subject'], $payload['message'] ) ) {
			// Not retriable. Do not burn an attempt on it.
			$affected = Studiou_WCMQ_DB::update_row_if_state(
				$row_id,
				array(
					'state'      => Studiou_WCMQ_DB::STATE_FAILED,
					'claimed_at' => null,
					'last_error' => 'corrupt payload',
				),
				Studiou_WCMQ_DB::STATE_SENDING
			);
			Studiou_WCMQ_DB::log( 'error', 'Corrupt payload; row failed without sending.', $row_id );
			// This is a terminal state. Without the note the order records nothing
			// at all: the customer got no email and the audit trail is silent.
			if ( 1 === (int) $affected ) {
				Studiou_WCMQ_Notifier::annotate( $row, false, 'corrupt payload' );
			}
			Studiou_WCMQ_State::maybe_settle();
			return;
		}

		$this->last_error = null;

		$sent = false;
		try {
			$sent = $this->dispatch( $payload );
		} catch ( Throwable $e ) {
			$this->last_error = $e->getMessage();
			$sent             = false;
		}

		$error            = $this->last_error;
		$this->last_error = null;

		// Completed send attempts, not crashes. Never bumped at claim, and folded
		// into the same UPDATE that commits the outcome so there is no window in
		// which the row has been attempted but not resolved. We hold the claim, so
		// nobody else is writing this column.
		$attempts = (int) $row->attempts + 1;

		if ( $sent ) {
			$this->on_success( $row, $row_id, $attempts );
		} else {
			$this->on_failure( $row, $row_id, $attempts, $error );
		}

		Studiou_WCMQ_State::maybe_settle();
	}

	/**
	 * Call wp_mail() with the stored payload, exactly once.
	 *
	 * Two hooks have to be neutralised for the duration:
	 *
	 * 1. `pre_wp_mail` — our own interceptor, or the mail would be re-queued
	 *    forever. Handled by the $sending flag.
	 *
	 * 2. `wp_mail` — WordPress applies this filter to $atts *before* it applies
	 *    `pre_wp_mail`, so the payload we snapshotted at enqueue has already been
	 *    through it. Calling wp_mail() again here would run every `wp_mail` filter
	 *    on the site a second time, against its own prior output: an unsubscribe
	 *    footer appended twice, a BCC-archive header duplicated. Queued mail would
	 *    differ from synchronous mail in ways nobody would connect to this plugin.
	 *
	 * Detaching the whole `wp_mail` hook and restoring it in `finally` is the only
	 * way to suppress third-party callbacks we do not control. `wp_mail_from` and
	 * friends need no such treatment — they are idempotent, and the resolved
	 * sender is already baked into the stored headers.
	 *
	 * @param array $payload Decoded mail payload.
	 * @return bool
	 */
	private function dispatch( array $payload ) {
		global $wp_filter;

		$headers     = isset( $payload['headers'] ) ? $payload['headers'] : '';
		$attachments = isset( $payload['attachments'] ) ? (array) $payload['attachments'] : array();

		$saved_wp_mail = isset( $wp_filter['wp_mail'] ) ? $wp_filter['wp_mail'] : null;
		if ( null !== $saved_wp_mail ) {
			unset( $wp_filter['wp_mail'] );
		}

		Studiou_WCMQ_Interceptor::$sending = true;

		try {
			return (bool) wp_mail( $payload['to'], $payload['subject'], $payload['message'], $headers, $attachments );
		} finally {
			// Never leave either of these in place: under Action Scheduler many
			// sends share one PHP process, and a stuck flag or a missing hook
			// would corrupt every later mail in the batch.
			Studiou_WCMQ_Interceptor::$sending = false;
			if ( null !== $saved_wp_mail ) {
				$wp_filter['wp_mail'] = $saved_wp_mail;
			}
		}
	}

	/**
	 * Commit the send, then annotate.
	 *
	 * Order matters. Once wp_mail() has returned true, nothing may throw back
	 * into the send path: a fatal here would leave the row in `sending` for the
	 * reaper to rescue and send a second time.
	 *
	 * @param object $row      Row snapshot from before the send.
	 * @param int    $row_id   Row id.
	 * @param int    $attempts Attempts including this one.
	 */
	private function on_success( $row, $row_id, $attempts ) {
		// State-guarded: if the row was deleted from the admin, or reclaimed by
		// the reaper, between our claim and now, this reports 0 rather than
		// silently "succeeding" and letting us delete attachments and write an
		// order note for a row that no longer exists.
		$affected = Studiou_WCMQ_DB::update_row_if_state(
			$row_id,
			array(
				'state'      => Studiou_WCMQ_DB::STATE_SENT,
				'sent_at'    => time(),
				'claimed_at' => null,
				'last_error' => null,
				'attempts'   => (int) $attempts,
			),
			Studiou_WCMQ_DB::STATE_SENDING
		);

		if ( 1 !== (int) $affected ) {
			// The mail went out but we could not record it. Either the DB errored
			// (row stays `sending`, the reaper will reclaim and resend — a
			// duplicate we cannot prevent from here) or the row vanished under us.
			// Either way: do not delete the attachment directory, so a resend
			// still carries its invoice, and write no note claiming success.
			Studiou_WCMQ_DB::log(
				'error',
				'Mail was sent but the row could not be committed to `sent` (it may have been deleted or reclaimed). It may be re-sent.',
				$row_id
			);
			return;
		}

		Studiou_WCMQ_Queue::delete_attachment_dir( $row->attachment_dir );
		Studiou_WCMQ_Notifier::annotate( $row, true, null );

		Studiou_WCMQ_DB::log(
			'info',
			sprintf( 'Sent %s to %s.', $row->context, $row->recipient ),
			$row_id
		);
	}

	/**
	 * Retry with backoff, or give up.
	 *
	 * @param object      $row      Row snapshot.
	 * @param int         $row_id   Row id.
	 * @param int         $attempts Attempts after this one.
	 * @param string|null $error    Captured error.
	 */
	private function on_failure( $row, $row_id, $attempts, $error ) {
		$max_attempts = (int) Studiou_WCMQ_Settings::get( 'max_attempts' );

		if ( $attempts >= $max_attempts ) {
			$affected = Studiou_WCMQ_DB::update_row_if_state(
				$row_id,
				array(
					'state'      => Studiou_WCMQ_DB::STATE_FAILED,
					'claimed_at' => null,
					'last_error' => $error,
					'attempts'   => (int) $attempts,
				),
				Studiou_WCMQ_DB::STATE_SENDING
			);
			// The attachment directory is deliberately KEPT: failed rows are
			// manually retriable from the admin, and a retry without its
			// attachments sends a broken email that looks fine.
			if ( 1 === (int) $affected ) {
				Studiou_WCMQ_Notifier::annotate( $row, false, $error );
			}
			Studiou_WCMQ_DB::log(
				'error',
				sprintf( 'Send failed permanently after %d attempts: %s', $attempts, (string) $error ),
				$row_id
			);
			return;
		}

		$interval = Studiou_WCMQ_Settings::interval();
		// `**`, not `^`. In PHP `^` is bitwise XOR and binds looser than `*`.
		$backoff = min( HOUR_IN_SECONDS, $interval * ( 2 ** $attempts ) );

		list( $slot, $affected ) = Studiou_WCMQ_Queue::with_slot_lock(
			time() + $backoff,
			function ( $slot ) use ( $row_id, $error, $attempts ) {
				return Studiou_WCMQ_DB::update_row_if_state(
					$row_id,
					array(
						'state'        => Studiou_WCMQ_DB::STATE_PENDING,
						'claimed_at'   => null,
						'scheduled_at' => $slot,
						'last_error'   => $error,
						'attempts'     => (int) $attempts,
					),
					Studiou_WCMQ_DB::STATE_SENDING
				);
			}
		);

		if ( 1 !== (int) $affected ) {
			// The row did not move back to `pending`. Scheduling an action for it
			// would wake a worker that finds nothing to claim. The row is still
			// `sending`, so the reaper will reclaim it after claim_timeout.
			Studiou_WCMQ_DB::log( 'error', 'Could not return row to pending for retry; leaving it to the reaper.', $row_id );
			return;
		}

		$action_id = as_schedule_single_action( $slot, Studiou_WCMQ_Queue::HOOK, array( (int) $row_id ), Studiou_WCMQ_Queue::GROUP );
		if ( ! $action_id ) {
			// The row is pending with no action — an invariant violation. The
			// reaper repairs it, so this degrades to "recovered late", not "lost".
			Studiou_WCMQ_DB::log( 'error', 'Could not schedule retry; reaper will repair.', $row_id );
			return;
		}

		Studiou_WCMQ_DB::log(
			'warning',
			sprintf( 'Send failed (attempt %d/%d); retrying at %d. %s', $attempts, $max_attempts, $slot, (string) $error ),
			$row_id
		);
	}
}