* @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; } }