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

<?php
/**
 * Slot allocation, enqueue, attachment custody.
 *
 * @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_Queue
 */
class Studiou_WCMQ_Queue {

	const HOOK       = 'studiou_wcmq_send';
	const GROUP      = 'studiou-wcmq';
	const LOCK_NAME  = 'studiou_wcmq_slot';
	const LOCK_WAIT  = 5;
	const ATTACH_DIR = 'studiou-wcmq-attachments';

	/**
	 * Allocate a slot and publish it, under the advisory lock.
	 *
	 * Returns [ $slot, $publish_result ]. Callers need the slot AFTER the lock is
	 * released, to hand to as_schedule_single_action(); a helper that returned
	 * only $publish's result would leave them with no timestamp, and
	 * as_schedule_single_action(null, ...) coerces to 0, which Action Scheduler
	 * treats as "run immediately" — draining the whole queue at once.
	 *
	 * The publish callback must INSERT or UPDATE the row: with the slot derived
	 * from the table, the write is what publishes the allocation. A lock that
	 * spans only the SELECT closes no race.
	 *
	 * @param int      $earliest Floor timestamp.
	 * @param callable $publish  fn( int $slot ): mixed.
	 * @return array [ int $slot, mixed $result ]
	 */
	public static function with_slot_lock( $earliest, callable $publish ) {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.NotPrepared
		$locked = (int) $wpdb->get_var( $wpdb->prepare( 'SELECT GET_LOCK(%s, %d)', self::LOCK_NAME, self::LOCK_WAIT ) );

		if ( 1 !== $locked ) {
			// Overshooting the rate limit is bad; losing a customer's order
			// confirmation is worse. Proceed unlocked and say so.
			Studiou_WCMQ_DB::log( 'warning', 'Slot lock unavailable; allocating without it.' );
		}

		try {
			$slot   = self::allocate_slot( (int) $earliest );
			$result = call_user_func( $publish, $slot );
			return array( $slot, $result );
		} finally {
			if ( 1 === $locked ) {
				// phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.NotPrepared
				$wpdb->query( $wpdb->prepare( 'SELECT RELEASE_LOCK(%s)', self::LOCK_NAME ) );
			}
		}
	}

	/**
	 * First free slot at or after $earliest.
	 *
	 * The smallest t >= max(now, $earliest) such that no pending/sending row sits
	 * within `interval` of t. A monotonic "next slot" counter cannot express this
	 * once retries exist: a row backed off an hour would drag every subsequent
	 * enqueue behind it.
	 *
	 * The collision test uses a two-sided range comparison rather than
	 * ABS(scheduled_at - candidate). MySQL evaluates a subtraction with any
	 * UNSIGNED operand as unsigned, so the moment an occupied row precedes the
	 * candidate — mail #2 of every bulk — it underflows into error 1690. The
	 * timestamp columns are signed for the same reason.
	 *
	 * @param int $earliest Floor timestamp.
	 * @return int
	 */
	public static function allocate_slot( $earliest ) {
		global $wpdb;

		$table    = Studiou_WCMQ_DB::queue_table();
		$interval = Studiou_WCMQ_Settings::interval();
		$earliest = (int) $earliest;

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		$slot = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT c.candidate FROM (
					SELECT GREATEST(%d, UNIX_TIMESTAMP()) AS candidate
					UNION ALL
					SELECT q.scheduled_at + %d
					  FROM {$table} q
					 WHERE q.state IN ('pending','sending')
					   AND q.scheduled_at + %d >= GREATEST(%d, UNIX_TIMESTAMP())
				) AS c
				WHERE NOT EXISTS (
					SELECT 1 FROM {$table} q2
					 WHERE q2.state IN ('pending','sending')
					   AND q2.scheduled_at >  c.candidate - %d
					   AND q2.scheduled_at <  c.candidate + %d
				)
				ORDER BY c.candidate ASC
				LIMIT 1",
				$earliest,
				$interval,
				$interval,
				$earliest,
				$interval,
				$interval
			)
		);
		// phpcs:enable

		// Cannot happen: the floor is always a candidate, and the largest
		// candidate always survives NOT EXISTS. Defensive only.
		if ( null === $slot ) {
			return max( time(), $earliest );
		}

		return (int) $slot;
	}

	/**
	 * Persist a mail and schedule its send.
	 *
	 * Returns the row id, or false on ANY failure. The interceptor treats false
	 * as "not handled" and lets wp_mail() send synchronously.
	 *
	 * @param array $ctx     Context from the probe.
	 * @param array $payload Mail snapshot.
	 * @return int|false
	 */
	public static function enqueue( array $ctx, array $payload ) {
		try {
			$dir_token = bin2hex( random_bytes( 16 ) );
		} catch ( Throwable $e ) {
			Studiou_WCMQ_DB::log( 'error', 'Could not generate attachment token: ' . $e->getMessage() );
			return false;
		}

		// Copy attachments BEFORE the row becomes durable.
		//
		// Attachments are file paths. Invoice plugins write them to temp dirs and
		// delete them at end of request; by send time they are gone. If the row
		// were inserted first and the paths rewritten afterwards, a PHP fatal in
		// between would leave a permanent `pending` row whose payload points at
		// files that no longer exist — and rollback() cannot run after a fatal.
		// The reaper would then faithfully resurrect a mail that can never send.
		//
		// Copying first means the row is never durable with paths we do not own.
		// The only cost is an orphaned directory if we bail before the INSERT,
		// which the retention sweep collects.
		$copied = self::copy_attachments( $dir_token, $payload['attachments'] );
		if ( null === $copied ) {
			self::delete_attachment_dir( $dir_token );
			Studiou_WCMQ_DB::log( 'error', 'Could not copy mail attachments; sending synchronously.' );
			return false;
		}
		$payload['attachments'] = $copied;

		// wp_json_encode returns false on unencodable input (e.g. invalid UTF-8
		// in a rendered order email) WITHOUT throwing. An exception-only guard
		// would miss it and the mail would be lost.
		$json = wp_json_encode( $payload );
		if ( ! is_string( $json ) || '' === $json ) {
			self::delete_attachment_dir( $dir_token );
			Studiou_WCMQ_DB::log( 'error', 'Could not encode mail payload; sending synchronously.' );
			return false;
		}

		$recipient = is_array( $payload['to'] ) ? implode( ',', $payload['to'] ) : (string) $payload['to'];
		$now       = time();

		$row = array(
			'context'        => (string) $ctx['id'],
			'order_id'       => (int) $ctx['order_id'],
			'recipient'      => substr( $recipient, 0, 320 ),
			'subject'        => (string) $payload['subject'],
			'payload'        => $json,
			'attachment_dir' => $dir_token,
			'priority'       => 10,
			'state'          => Studiou_WCMQ_DB::STATE_PENDING,
			'attempts'       => 0,
			'reclaims'       => 0,
			'scheduled_at'   => 0,
			'created_at'     => $now,
		);

		try {
			list( $slot, $row_id ) = self::with_slot_lock(
				$now,
				function ( $slot ) use ( $row ) {
					$row['scheduled_at'] = $slot;
					return Studiou_WCMQ_DB::insert_row( $row );
				}
			);
		} catch ( Throwable $e ) {
			self::delete_attachment_dir( $dir_token );
			Studiou_WCMQ_DB::log( 'error', 'Enqueue insert failed: ' . $e->getMessage() );
			return false;
		}

		$row_id = (int) $row_id;
		if ( ! $row_id ) {
			self::delete_attachment_dir( $dir_token );
			Studiou_WCMQ_DB::log( 'error', 'Could not insert queue row; sending synchronously.' );
			return false;
		}

		// Scheduled last, and checked: as_schedule_single_action() returns 0 on
		// failure without throwing. A fatal between the INSERT and here leaves a
		// `pending` row with no action — but with a payload that is already
		// self-contained, so the reaper can safely reschedule it.
		$action_id = as_schedule_single_action( $slot, self::HOOK, array( (int) $row_id ), self::GROUP );
		if ( ! $action_id ) {
			self::rollback( $row_id, $dir_token );
			Studiou_WCMQ_DB::log( 'error', 'Could not schedule send action; sending synchronously.' );
			return false;
		}

		Studiou_WCMQ_DB::log(
			'debug',
			sprintf( 'Queued %s for %s at %d.', $ctx['id'], $recipient, $slot ),
			$row_id
		);

		return $row_id;
	}

	/**
	 * Put an existing row back into `pending` on a fresh future slot.
	 *
	 * Used by the admin Retry action. Takes the same lock as enqueue, because
	 * the UPDATE publishes the allocation exactly as the INSERT does, and guards
	 * the schedule call the same way.
	 *
	 * **$expected_state is mandatory and closes a double-send race.** The admin
	 * reads the row, then calls this — and in that window a worker's Action
	 * Scheduler action can fire and claim the row (`pending → sending`). An
	 * unguarded UPDATE would then force the mid-send row back to `pending`: the
	 * worker's own state-guarded commit fails (0 rows), so it keeps the
	 * attachments and writes no note — but the mail already went out via
	 * `wp_mail()`, and the fresh action we schedule here delivers a SECOND copy.
	 * Guarding on the state the caller observed means a row a worker has since
	 * claimed comes back as 0-changed → we do not schedule, and the worker's send
	 * stands as the single delivery.
	 *
	 * @param int    $row_id         Row id.
	 * @param int    $earliest       Floor timestamp, 0 for now.
	 * @param array  $extra          Extra columns to set.
	 * @param string $expected_state State the row must still be in (what the
	 *                               caller observed at read time).
	 * @return bool
	 */
	public static function reschedule( $row_id, $earliest, array $extra, $expected_state ) {
		$row_id = (int) $row_id;
		if ( ! $row_id || ! function_exists( 'as_schedule_single_action' ) ) {
			return false;
		}
		$earliest = $earliest ? (int) $earliest : time();

		list( $slot, $updated ) = self::with_slot_lock(
			$earliest,
			function ( $slot ) use ( $row_id, $extra, $expected_state ) {
				$data = array_merge(
					$extra,
					array(
						'state'        => Studiou_WCMQ_DB::STATE_PENDING,
						'claimed_at'   => null,
						'scheduled_at' => $slot,
					)
				);
				return Studiou_WCMQ_DB::update_row_if_state( $row_id, $data, $expected_state );
			}
		);

		// 0 (or false) means the row was no longer in $expected_state — most
		// importantly, a worker claimed it out from under us. Do NOT schedule an
		// action: the worker is mid-send and a fresh action would double-deliver.
		// Report failure so the admin sees "couldn't process", not "re-queued".
		if ( ! $updated ) {
			Studiou_WCMQ_DB::log( 'error', 'Reschedule changed no row (state moved under us); not scheduling.', $row_id );
			return false;
		}

		// The row is now ours in `pending`. Drop any stale action for it before
		// adding the new one, so AS never holds two actions for one row (a row
		// retried while still pending already owns an action at its old slot; a
		// running action cannot be unscheduled, but that path is closed by the
		// state guard above). Both fire → the claim guard makes it a single send,
		// but unscheduling keeps it on the intended new slot.
		if ( function_exists( 'as_unschedule_all_actions' ) ) {
			as_unschedule_all_actions( self::HOOK, array( (int) $row_id ), self::GROUP );
		}

		$action_id = as_schedule_single_action( $slot, self::HOOK, array( (int) $row_id ), self::GROUP );
		if ( ! $action_id ) {
			Studiou_WCMQ_DB::log( 'error', 'Could not schedule action while rescheduling; reaper will repair.', $row_id );
			return false;
		}
		return true;
	}

	/**
	 * Undo a partial enqueue.
	 *
	 * Cannot run after a PHP fatal — that is what the reaper is for. This only
	 * covers the failures we can actually catch.
	 *
	 * @param int    $row_id    Row id.
	 * @param string $dir_token Attachment directory token.
	 */
	private static function rollback( $row_id, $dir_token ) {
		self::delete_attachment_dir( $dir_token );
		Studiou_WCMQ_DB::delete_row( $row_id );
	}

	/* ---------------------------------------------------------------------
	 * Attachments
	 * ------------------------------------------------------------------ */

	/**
	 * Base directory for attachment copies, created and guarded on first use.
	 *
	 * @return string Absolute path, or '' on failure.
	 */
	public static function attachments_basedir() {
		$uploads = wp_upload_dir();
		if ( ! empty( $uploads['error'] ) || empty( $uploads['basedir'] ) ) {
			return '';
		}
		$base = trailingslashit( $uploads['basedir'] ) . self::ATTACH_DIR;

		if ( ! file_exists( $base ) && ! wp_mkdir_p( $base ) ) {
			return '';
		}

		// Write the guards whenever they are missing, not only when we just
		// created the directory — a backup restore or a security scanner can
		// remove them while the directory survives, and an unguarded directory
		// full of customer invoices is exactly what these prevent on Apache.
		// Nginx ignores .htaccess entirely; there the random 32-hex directory
		// token is the real protection, and the readme documents the location deny.
		$index = $base . '/index.php';
		if ( ! file_exists( $index ) ) {
			@file_put_contents( $index, '<?php // Silence is golden.' ); // phpcs:ignore
		}
		$htaccess = $base . '/.htaccess';
		if ( ! file_exists( $htaccess ) ) {
			// Both syntaxes: Apache 2.4 dropped Order/Deny.
			$rules = "<IfModule mod_authz_core.c>\n\tRequire all denied\n</IfModule>\n"
				. "<IfModule !mod_authz_core.c>\n\tOrder deny,allow\n\tDeny from all\n</IfModule>\n";
			@file_put_contents( $htaccess, $rules ); // phpcs:ignore
		}

		return $base;
	}

	/**
	 * Copy attachments into this row's private directory.
	 *
	 * Array keys are preserved: wp_mail() treats a string key as the attachment
	 * filename.
	 *
	 * @param string $dir_token   Directory token.
	 * @param array  $attachments Original paths.
	 * @return array|null Copied paths, or null on failure.
	 */
	private static function copy_attachments( $dir_token, $attachments ) {
		$attachments = (array) $attachments;
		if ( empty( $attachments ) ) {
			return array();
		}

		$base = self::attachments_basedir();
		if ( '' === $base ) {
			return null;
		}

		$dir = $base . '/' . $dir_token;
		if ( ! wp_mkdir_p( $dir ) ) {
			return null;
		}

		$out = array();
		foreach ( $attachments as $key => $path ) {
			if ( ! is_string( $path ) || '' === $path || ! is_readable( $path ) ) {
				return null;
			}
			$dest = $dir . '/' . wp_unique_filename( $dir, basename( $path ) );
			if ( ! @copy( $path, $dest ) ) { // phpcs:ignore
				return null;
			}
			$out[ $key ] = $dest;
		}

		return $out;
	}

	/**
	 * Remove a row's attachment directory.
	 *
	 * @param string $dir_token Directory token.
	 */
	public static function delete_attachment_dir( $dir_token ) {
		if ( ! is_string( $dir_token ) || ! preg_match( '/^[a-f0-9]{32}$/', $dir_token ) ) {
			return;
		}
		$base = self::attachments_basedir();
		if ( '' === $base ) {
			return;
		}
		self::rmdir_recursive( $base . '/' . $dir_token );
	}

	/**
	 * Delete a directory and its files, one level deep.
	 *
	 * @param string $dir Absolute path.
	 */
	public static function rmdir_recursive( $dir ) {
		if ( ! is_dir( $dir ) ) {
			return;
		}
		$items = @scandir( $dir ); // phpcs:ignore
		if ( is_array( $items ) ) {
			foreach ( $items as $item ) {
				if ( '.' === $item || '..' === $item ) {
					continue;
				}
				$path = $dir . '/' . $item;
				if ( is_dir( $path ) ) {
					self::rmdir_recursive( $path );
				} else {
					@unlink( $path ); // phpcs:ignore
				}
			}
		}
		@rmdir( $dir ); // phpcs:ignore
	}
}