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

<?php
/**
 * Schema, migrations and row access.
 *
 * @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_DB
 */
class Studiou_WCMQ_DB {

	const DB_VERSION_OPTION = 'studiou_wcmq_db_version';
	const DB_VERSION        = '1.0.0';

	const STATE_PENDING = 'pending';
	const STATE_SENDING = 'sending';
	const STATE_SENT    = 'sent';
	const STATE_FAILED  = 'failed';

	/**
	 * Queue table name.
	 *
	 * @return string
	 */
	public static function queue_table() {
		global $wpdb;
		return $wpdb->prefix . 'studiou_wcmq_queue';
	}

	/**
	 * Log table name.
	 *
	 * @return string
	 */
	public static function log_table() {
		global $wpdb;
		return $wpdb->prefix . 'studiou_wcmq_log';
	}

	/**
	 * Create/upgrade tables when the stored schema version lags the constant.
	 *
	 * mu-plugins get no activation hook, so this runs on load. dbDelta is
	 * idempotent; the option check keeps it to one comparison per request.
	 */
	public static function maybe_upgrade_db() {
		if ( get_option( self::DB_VERSION_OPTION ) === self::DB_VERSION ) {
			return;
		}
		self::create_tables();
		update_option( self::DB_VERSION_OPTION, self::DB_VERSION, true );
	}

	/**
	 * Run dbDelta for both tables.
	 *
	 * All timestamp columns are SIGNED bigint. Unsigned subtraction underflows
	 * into MySQL error 1690 inside the slot allocator's range comparison.
	 */
	private static function create_tables() {
		global $wpdb;

		require_once ABSPATH . 'wp-admin/includes/upgrade.php';

		$charset_collate = $wpdb->get_charset_collate();
		$queue           = self::queue_table();
		$log             = self::log_table();

		$sql_queue = "CREATE TABLE {$queue} (
	id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
	context varchar(100) NOT NULL DEFAULT '',
	order_id bigint(20) unsigned NOT NULL DEFAULT 0,
	recipient varchar(320) NOT NULL DEFAULT '',
	subject text NOT NULL,
	payload longtext NOT NULL,
	attachment_dir char(32) NULL,
	priority smallint(6) NOT NULL DEFAULT 10,
	state varchar(20) NOT NULL DEFAULT 'pending',
	attempts tinyint(3) unsigned NOT NULL DEFAULT 0,
	reclaims tinyint(3) unsigned NOT NULL DEFAULT 0,
	last_error text NULL,
	scheduled_at bigint(20) NOT NULL DEFAULT 0,
	claimed_at bigint(20) NULL,
	sent_at bigint(20) NULL,
	created_at bigint(20) NOT NULL DEFAULT 0,
	PRIMARY KEY  (id),
	KEY state_sched (state,scheduled_at),
	KEY state_claim (state,claimed_at),
	KEY created_at (created_at),
	KEY order_id (order_id)
) {$charset_collate};";

		$sql_log = "CREATE TABLE {$log} (
	id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
	level varchar(10) NOT NULL DEFAULT 'info',
	queue_id bigint(20) unsigned NULL,
	message text NOT NULL,
	created_at bigint(20) NOT NULL DEFAULT 0,
	PRIMARY KEY  (id),
	KEY created_at (created_at),
	KEY level (level)
) {$charset_collate};";

		dbDelta( $sql_queue );
		dbDelta( $sql_log );
	}

	/* ---------------------------------------------------------------------
	 * Queue rows
	 * ------------------------------------------------------------------ */

	/**
	 * Insert a queue row.
	 *
	 * @param array $data Column => value.
	 * @return int Row id, or 0 on failure.
	 */
	public static function insert_row( array $data ) {
		global $wpdb;
		$ok = $wpdb->insert( self::queue_table(), $data ); // phpcs:ignore WordPress.DB.DirectDatabaseQuery
		if ( false === $ok ) {
			return 0;
		}
		return (int) $wpdb->insert_id;
	}

	/**
	 * Fetch one queue row.
	 *
	 * @param int $row_id Row id.
	 * @return object|null
	 */
	public static function get_row( $row_id ) {
		global $wpdb;
		$table = self::queue_table();
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		return $wpdb->get_row( $wpdb->prepare( "SELECT * FROM {$table} WHERE id = %d", (int) $row_id ) );
	}

	/**
	 * Update a queue row. Null values are written as SQL NULL by $wpdb.
	 *
	 * Returns $wpdb->update()'s raw result — rows CHANGED, or false on a DB
	 * error. Callers must decide what they need, because the two are different
	 * questions and conflating them hides real failures:
	 *
	 *   false === $r   a DB error occurred
	 *   0    === $r    no row changed: the WHERE matched nothing, or the values
	 *                  were already what we asked for (WordPress does not set
	 *                  CLIENT_FOUND_ROWS, so matched-but-unchanged reads as 0)
	 *   >0             rows changed
	 *
	 * Do not "fix" a caller by testing `> 0` on an update that may legitimately
	 * be a no-op. Where a caller needs to know the row is still in the state it
	 * expects, use update_row_if_state() instead.
	 *
	 * @param int   $row_id Row id.
	 * @param array $data   Column => value.
	 * @return int|false
	 */
	public static function update_row( $row_id, array $data ) {
		global $wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		return $wpdb->update( self::queue_table(), $data, array( 'id' => (int) $row_id ) );
	}

	/**
	 * Update a queue row only while it is still in the expected state.
	 *
	 * The state is part of the WHERE clause, so a row deleted from the admin,
	 * or reclaimed by the reaper, or already resolved by another worker, comes
	 * back as 0 rows rather than as a silent success. Every terminal write in
	 * the worker and the reaper uses this: the alternative is deleting a row's
	 * attachments and writing its order note after the row has ceased to exist.
	 *
	 * All these callers change `state` (or `scheduled_at` to a future slot), so
	 * a matched row always reports >= 1 changed. 0 unambiguously means "not in
	 * the state I expected".
	 *
	 * @param int    $row_id         Row id.
	 * @param array  $data           Column => value.
	 * @param string $expected_state State the row must currently be in.
	 * @return int|false Rows changed, or false on DB error.
	 */
	public static function update_row_if_state( $row_id, array $data, $expected_state ) {
		global $wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		return $wpdb->update(
			self::queue_table(),
			$data,
			array(
				'id'    => (int) $row_id,
				'state' => (string) $expected_state,
			)
		);
	}

	/**
	 * Atomically claim a pending row for sending.
	 *
	 * This is the duplicate-send guard. Never replace it with SELECT-then-UPDATE.
	 *
	 * It does NOT touch `attempts`. The worker folds `attempts` into the same
	 * UPDATE that commits the outcome, so no row is ever "attempted but not
	 * resolved". Crashes are counted separately, in `reclaims`, by the reaper.
	 *
	 * @param int $row_id Row id.
	 * @return bool True when this process won the claim.
	 */
	public static function claim( $row_id ) {
		global $wpdb;
		$table = self::queue_table();
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		$affected = $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table} SET state = %s, claimed_at = %d WHERE id = %d AND state = %s",
				self::STATE_SENDING,
				time(),
				(int) $row_id,
				self::STATE_PENDING
			)
		);
		return 1 === (int) $affected;
	}

	/**
	 * Count rows still in flight.
	 *
	 * @return int
	 */
	public static function count_unfinished() {
		global $wpdb;
		$table = self::queue_table();
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		return (int) $wpdb->get_var(
			$wpdb->prepare(
				"SELECT COUNT(*) FROM {$table} WHERE state IN (%s, %s)",
				self::STATE_PENDING,
				self::STATE_SENDING
			)
		);
	}

	/**
	 * Row counts keyed by state.
	 *
	 * @return array
	 */
	public static function counts_by_state() {
		global $wpdb;
		$table = self::queue_table();
		$out   = array(
			self::STATE_PENDING => 0,
			self::STATE_SENDING => 0,
			self::STATE_SENT    => 0,
			self::STATE_FAILED  => 0,
		);
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		$rows = $wpdb->get_results( "SELECT state, COUNT(*) AS c FROM {$table} GROUP BY state" );
		if ( is_array( $rows ) ) {
			foreach ( $rows as $row ) {
				$out[ $row->state ] = (int) $row->c;
			}
		}
		return $out;
	}

	/**
	 * Paginated queue rows for the admin view.
	 *
	 * @param string $state    State filter, or '' for all.
	 * @param int    $per_page Rows per page.
	 * @param int    $offset   Offset.
	 * @return array
	 */
	public static function get_rows( $state, $per_page, $offset ) {
		global $wpdb;
		$table    = self::queue_table();
		$per_page = max( 1, (int) $per_page );
		$offset   = max( 0, (int) $offset );

		if ( $state ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
			return (array) $wpdb->get_results(
				$wpdb->prepare(
					"SELECT * FROM {$table} WHERE state = %s ORDER BY scheduled_at ASC, id ASC LIMIT %d OFFSET %d",
					$state,
					$per_page,
					$offset
				)
			);
		}
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		return (array) $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM {$table} ORDER BY scheduled_at ASC, id ASC LIMIT %d OFFSET %d",
				$per_page,
				$offset
			)
		);
	}

	/**
	 * Count rows, optionally filtered by state.
	 *
	 * @param string $state State filter, or '' for all.
	 * @return int
	 */
	public static function count_rows( $state = '' ) {
		global $wpdb;
		$table = self::queue_table();
		if ( $state ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
			return (int) $wpdb->get_var( $wpdb->prepare( "SELECT COUNT(*) FROM {$table} WHERE state = %s", $state ) );
		}
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		return (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$table}" );
	}

	/**
	 * Delete one queue row.
	 *
	 * @param int $row_id Row id.
	 * @return bool
	 */
	public static function delete_row( $row_id ) {
		global $wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		return false !== $wpdb->delete( self::queue_table(), array( 'id' => (int) $row_id ), array( '%d' ) );
	}

	/* ---------------------------------------------------------------------
	 * Log rows
	 * ------------------------------------------------------------------ */

	/**
	 * Append a log row.
	 *
	 * `debug` rows are written only when the debug setting is on.
	 *
	 * @param string   $level    debug|info|warning|error.
	 * @param string   $message  Message.
	 * @param int|null $queue_id Related queue row.
	 */
	public static function log( $level, $message, $queue_id = null ) {
		if ( ! in_array( $level, array( 'debug', 'info', 'warning', 'error' ), true ) ) {
			$level = 'info';
		}
		if ( 'debug' === $level && ! Studiou_WCMQ_Settings::debug() ) {
			return;
		}

		global $wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery
		$wpdb->insert(
			self::log_table(),
			array(
				'level'      => $level,
				'queue_id'   => $queue_id ? (int) $queue_id : null,
				'message'    => (string) $message,
				'created_at' => time(),
			)
		);

		if ( in_array( $level, array( 'warning', 'error' ), true ) ) {
			Studiou_WCMQ_Utils_Log::log( strtoupper( $level ) . ': ' . $message );
		}
	}

	/**
	 * Paginated log rows.
	 *
	 * @param string $level    Level filter, or '' for all.
	 * @param int    $per_page Rows per page.
	 * @param int    $offset   Offset.
	 * @return array
	 */
	public static function get_log_rows( $level, $per_page, $offset ) {
		global $wpdb;
		$table    = self::log_table();
		$per_page = max( 1, (int) $per_page );
		$offset   = max( 0, (int) $offset );

		if ( $level ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
			return (array) $wpdb->get_results(
				$wpdb->prepare(
					"SELECT * FROM {$table} WHERE level = %s ORDER BY id DESC LIMIT %d OFFSET %d",
					$level,
					$per_page,
					$offset
				)
			);
		}
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		return (array) $wpdb->get_results(
			$wpdb->prepare( "SELECT * FROM {$table} ORDER BY id DESC LIMIT %d OFFSET %d", $per_page, $offset )
		);
	}

	/**
	 * Count log rows.
	 *
	 * @param string $level Level filter, or '' for all.
	 * @return int
	 */
	public static function count_log_rows( $level = '' ) {
		global $wpdb;
		$table = self::log_table();
		if ( $level ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
			return (int) $wpdb->get_var( $wpdb->prepare( "SELECT COUNT(*) FROM {$table} WHERE level = %s", $level ) );
		}
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		return (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$table}" );
	}

	/**
	 * Empty the log.
	 *
	 * Batched DELETE, never TRUNCATE: TRUNCATE requires the DROP privilege,
	 * which the shared hosts this plugin targets frequently withhold.
	 *
	 * @return int Rows deleted.
	 */
	public static function clear_log() {
		global $wpdb;
		$table   = self::log_table();
		$deleted = 0;
		for ( $i = 0; $i < 200; $i++ ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
			$n = $wpdb->query( "DELETE FROM {$table} LIMIT 5000" );
			if ( ! $n ) {
				break;
			}
			$deleted += (int) $n;
		}
		return $deleted;
	}
}