sirius-press/plugins/sirius-press-sia-export/includes/class-spe-queue.php

<?php
/**
 * The export queue.
 *
 * Publishing a post must not wait on a network round trip to a gateway that
 * might be slow, rate limited or down. If it did, the first flaky minute the
 * gateway has would show up as WordPress hanging on "Publish", and the site
 * owner would reasonably conclude the fork is broken.
 *
 * So publishing writes a row here and returns. Something else — cron, an
 * admin screen, WP-CLI, or a browser in manual-signing mode — drains it. The
 * queue is also what makes retries, bulk exports and manual signing the same
 * mechanism instead of three.
 *
 * Rows are keyed by path, not by post. A post and its archive page and the
 * home page all move when one post changes, and several posts changing in a
 * minute should not queue the home page five times.
 *
 * @package SiriusPress
 */

defined( 'ABSPATH' ) || exit;

final class SPE_Queue {

	const TABLE   = 'sirius_export_queue';
	const VERSION = 1;

	const STATUS_PENDING = 'pending';
	const STATUS_DONE    = 'done';
	const STATUS_FAILED  = 'failed';

	/** Give up after this many attempts and leave the row for a human. */
	const MAX_ATTEMPTS = 5;

	public static function table() {
		global $wpdb;
		return $wpdb->prefix . self::TABLE;
	}

	public static function install() {
		global $wpdb;
		require_once ABSPATH . 'wp-admin/includes/upgrade.php';
		$table   = self::table();
		$collate = $wpdb->get_charset_collate();
		dbDelta(
			"CREATE TABLE {$table} (
				id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
				path varchar(200) NOT NULL,
				source_url text NOT NULL,
				kind varchar(20) NOT NULL DEFAULT 'page',
				status varchar(12) NOT NULL DEFAULT 'pending',
				attempts smallint(5) unsigned NOT NULL DEFAULT 0,
				last_error text NULL,
				content_hash char(64) NOT NULL DEFAULT '',
				queued_at datetime NOT NULL,
				done_at datetime NULL,
				PRIMARY KEY  (id),
				UNIQUE KEY path (path),
				KEY status (status)
			) {$collate};"
		);
		update_option( 'sirius_press_export_db_version', self::VERSION, false );
	}

	/**
	 * Add or refresh a path.
	 *
	 * An existing row is reset to pending rather than duplicated, so a path
	 * that changes ten times before the queue drains is exported once, from
	 * its final state.
	 *
	 * @param string $path       Path inside the bucket.
	 * @param string $source_url The URL to render.
	 * @param string $kind       'page', 'asset', 'feed' — for the admin UI only.
	 */
	public static function add( $path, $source_url, $kind = 'page' ) {
		global $wpdb;
		$path = SP_Gateway::clean_path( $path );
		if ( '' === $path ) {
			return false;
		}
		$table = self::table();
		// A single statement so two concurrent publishes cannot both decide
		// the row is missing and then collide on the unique index.
		$sql = $wpdb->prepare(
			"INSERT INTO {$table} (path, source_url, kind, status, attempts, queued_at)
			 VALUES (%s, %s, %s, %s, 0, %s)
			 ON DUPLICATE KEY UPDATE
			   source_url = VALUES(source_url),
			   kind       = VALUES(kind),
			   status     = VALUES(status),
			   attempts   = 0,
			   last_error = NULL,
			   queued_at  = VALUES(queued_at)", // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			$path,
			(string) $source_url,
			$kind,
			self::STATUS_PENDING,
			current_time( 'mysql', true )
		);
		return false !== $wpdb->query( $sql ); // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared
	}

	/** @return array Rows waiting to go out. */
	public static function claim( $limit = 10 ) {
		global $wpdb;
		$table = self::table();
		return (array) $wpdb->get_results(
			$wpdb->prepare(
				"SELECT * FROM {$table} WHERE status = %s AND attempts < %d ORDER BY id ASC LIMIT %d", // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
				self::STATUS_PENDING,
				self::MAX_ATTEMPTS,
				(int) $limit
			)
		);
	}

	public static function mark_done( $id, $hash ) {
		global $wpdb;
		$wpdb->update(
			self::table(),
			array(
				'status'       => self::STATUS_DONE,
				'done_at'      => current_time( 'mysql', true ),
				'content_hash' => $hash,
				'last_error'   => null,
			),
			array( 'id' => (int) $id ),
			array( '%s', '%s', '%s', '%s' ),
			array( '%d' )
		);
	}

	public static function mark_failed( $id, $error ) {
		global $wpdb;
		$table = self::table();
		$wpdb->query(
			$wpdb->prepare(
				"UPDATE {$table} SET attempts = attempts + 1, last_error = %s,
				 status = CASE WHEN attempts + 1 >= %d THEN %s ELSE %s END
				 WHERE id = %d", // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
				mb_substr( (string) $error, 0, 1000 ),
				self::MAX_ATTEMPTS,
				self::STATUS_FAILED,
				self::STATUS_PENDING,
				(int) $id
			)
		);
	}

	/** Put failed rows back in the queue — the "try again" button. */
	public static function retry_failed() {
		global $wpdb;
		return (int) $wpdb->update(
			self::table(),
			array(
				'status'   => self::STATUS_PENDING,
				'attempts' => 0,
			),
			array( 'status' => self::STATUS_FAILED ),
			array( '%s', '%d' ),
			array( '%s' )
		);
	}

	public static function counts() {
		global $wpdb;
		$table = self::table();
		$rows  = (array) $wpdb->get_results( "SELECT status, COUNT(*) AS n FROM {$table} GROUP BY status" ); // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		$out   = array(
			self::STATUS_PENDING => 0,
			self::STATUS_DONE    => 0,
			self::STATUS_FAILED  => 0,
		);
		foreach ( $rows as $row ) {
			$out[ $row->status ] = (int) $row->n;
		}
		return $out;
	}

	public static function recent( $limit = 50, $status = '' ) {
		global $wpdb;
		$table = self::table();
		if ( '' !== $status ) {
			return (array) $wpdb->get_results(
				$wpdb->prepare( "SELECT * FROM {$table} WHERE status = %s ORDER BY id DESC LIMIT %d", $status, (int) $limit ) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
			);
		}
		return (array) $wpdb->get_results(
			$wpdb->prepare( "SELECT * FROM {$table} ORDER BY id DESC LIMIT %d", (int) $limit ) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	/**
	 * The hash of what was last successfully published at a path.
	 *
	 * Used to skip an upload whose bytes have not changed — the common case
	 * when a post edit re-renders the home page identically, and the
	 * difference between a queue that costs bandwidth and one that does not.
	 */
	public static function hash_at( $path ) {
		global $wpdb;
		$table = self::table();
		// Deliberately not filtered by status: re-queueing a path resets it to
		// pending but leaves the hash of what is actually sitting in the
		// bucket, which is the thing worth comparing against.
		return (string) $wpdb->get_var(
			$wpdb->prepare( "SELECT content_hash FROM {$table} WHERE path = %s", $path ) // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared
		);
	}

	public static function clear_done() {
		global $wpdb;
		return (int) $wpdb->delete( self::table(), array( 'status' => self::STATUS_DONE ), array( '%s' ) );
	}
}