* @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, '\n\tRequire all denied\n\n" . "\n\tOrder deny,allow\n\tDeny from all\n\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 } }