| 1 |
<?php |
| 2 |
/** |
| 3 |
* WCPOS sync store component. |
| 4 |
* |
| 5 |
* @package WCPOS\WooCommercePOS\Sync |
| 6 |
*/ |
| 7 |
|
| 8 |
namespace WCPOS\WooCommercePOS\Sync; |
| 9 |
|
| 10 |
// phpcs:disable Squiz.Commenting, Generic.Commenting -- Ported lab documentation is preserved verbatim. |
| 11 |
// phpcs:disable WordPress.DB.PreparedSQL.NotPrepared -- Queries use internal table names and generated SQL fragments. |
| 12 |
// phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Database failures are passed to exceptions, not rendered. |
| 13 |
|
| 14 |
use Automattic\WooCommerce\Utilities\OrderUtil; |
| 15 |
use WP_REST_Request; |
| 16 |
|
| 17 |
final class Sync_Journal { |
| 18 |
/** Persisted order backfill cursor. */ |
| 19 |
public const BACKFILL_OPTION = 'woocommerce_pos_sync_index_backfill'; |
| 20 |
|
| 21 |
/** Generation marker for the journal sequence space. */ |
| 22 |
public const EPOCH_OPTION = 'woocommerce_pos_sync_journal_epoch'; |
| 23 |
|
| 24 |
/** Wall-clock ms spent inside record() during the CURRENT request. */ |
| 25 |
public static float $request_write_ms = 0.0; |
| 26 |
|
| 27 |
/** Per-request dedup of identical customer lifecycle events. */ |
| 28 |
private array $recorded_this_request = array(); |
| 29 |
|
| 30 |
/** |
| 31 |
* Option-name prefix for the per-object-type lossy-prune watermarks. |
| 32 |
* |
| 33 |
* The watermark is scoped per object type for the same reason heads are |
| 34 |
* stream-scoped: the streams share one AUTO_INCREMENT space, so a single |
| 35 |
* global watermark advanced by an order prune would sit far above a quiet |
| 36 |
* catalogue stream's head. Every catalogue cursor would then read as |
| 37 |
* "below the horizon" and rebaseline on EVERY poll, forever — and |
| 38 |
* symmetrically for the order lane. A stream's horizon may only move for |
| 39 |
* rows that stream can serve. |
| 40 |
*/ |
| 41 |
const PRUNE_WATERMARK_OPTION_PREFIX = 'wcpos_change_log_prune_watermark_'; |
| 42 |
|
| 43 |
public function table_name(): string { |
| 44 |
global $wpdb; |
| 45 |
return $wpdb->prefix . Health::SYNC_JOURNAL_TABLE; |
| 46 |
} |
| 47 |
|
| 48 |
public function schema_sql( string $table_name, string $charset_collate = '' ): string { |
| 49 |
return "CREATE TABLE {$table_name} (\n" |
| 50 |
. " sequence BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,\n" |
| 51 |
. " object_type VARCHAR(20) NOT NULL,\n" |
| 52 |
. " object_id BIGINT UNSIGNED NOT NULL,\n" |
| 53 |
. " deleted TINYINT(1) NOT NULL DEFAULT 0,\n" |
| 54 |
. " revision VARCHAR(80) NOT NULL DEFAULT '',\n" |
| 55 |
. " modified_gmt DATETIME NOT NULL,\n" |
| 56 |
. " origin VARCHAR(40) NOT NULL DEFAULT 'hook',\n" |
| 57 |
. " created_gmt DATETIME NOT NULL,\n" |
| 58 |
. " PRIMARY KEY (sequence),\n" |
| 59 |
. " KEY type_sequence (object_type, sequence),\n" |
| 60 |
. " KEY type_object (object_type, object_id)\n" |
| 61 |
. ") {$charset_collate};"; |
| 62 |
} |
| 63 |
|
| 64 |
public function install(): void { |
| 65 |
global $wpdb; |
| 66 |
$table_name = $this->table_name(); |
| 67 |
$table_existed = Health::table_exists( $table_name ); |
| 68 |
if ( ! function_exists( 'dbDelta' ) ) { |
| 69 |
require_once ABSPATH . 'wp-admin/includes/upgrade.php'; |
| 70 |
} |
| 71 |
dbDelta( $this->schema_sql( $table_name, $wpdb->get_charset_collate() ) ); |
| 72 |
|
| 73 |
if ( ! $table_existed && Health::table_exists( $table_name ) ) { |
| 74 |
self::reset_prune_watermarks(); |
| 75 |
} |
| 76 |
if ( ! Health::table_exists( $table_name ) ) { |
| 77 |
return; |
| 78 |
} |
| 79 |
|
| 80 |
// The epoch marks a SEQUENCE GENERATION. Regenerate it only when the |
| 81 |
// table was actually (re)created — dbDelta on a surviving table is a |
| 82 |
// no-op and every row survives, so activation/upgrade re-runs must not |
| 83 |
// force every client (both lanes, since the epoch is journal-global) |
| 84 |
// into a needless resync-from-zero. |
| 85 |
if ( ! $table_existed ) { |
| 86 |
$this->regenerate_epoch(); |
| 87 |
} |
| 88 |
$backfill = $this->backfill_status(); |
| 89 |
$backfill_has_progress = 'idle' !== $backfill['status'] || $backfill['nextPage'] > 1 || $backfill['processed'] > 0; |
| 90 |
if ( 0 === $this->head_sequence() && $backfill_has_progress ) { |
| 91 |
delete_option( self::BACKFILL_OPTION ); |
| 92 |
} |
| 93 |
|
| 94 |
// A store with no orders has nothing to backfill: mark it complete on a |
| 95 |
// fresh table so sequence-zero pulls are journal-authoritative from the |
| 96 |
// first write (Order_Query holds the baseline on the modified-date scan |
| 97 |
// until the backfill is complete). Stores WITH history stay incomplete |
| 98 |
// until the admin backfill runs. Runs after the stale-cursor cleanup |
| 99 |
// above so a carried-over cursor cannot masquerade as completion. |
| 100 |
if ( ! $table_existed && function_exists( 'wc_get_orders' ) ) { |
| 101 |
$existing = wc_get_orders( |
| 102 |
array( |
| 103 |
'type' => 'shop_order', |
| 104 |
'limit' => 1, |
| 105 |
'return' => 'ids', |
| 106 |
) |
| 107 |
); |
| 108 |
if ( is_array( $existing ) && array() === $existing ) { |
| 109 |
update_option( |
| 110 |
self::BACKFILL_OPTION, |
| 111 |
array( |
| 112 |
'status' => 'complete', |
| 113 |
'nextPage' => 1, |
| 114 |
'pageSize' => null, |
| 115 |
'processed' => 0, |
| 116 |
'lastOrderId' => 0, |
| 117 |
'lastRunGmt' => gmdate( 'c' ), |
| 118 |
), |
| 119 |
false |
| 120 |
); |
| 121 |
} |
| 122 |
} |
| 123 |
} |
| 124 |
|
| 125 |
/** Return the stable id for the current sequence generation. */ |
| 126 |
public function ensure_epoch(): string { |
| 127 |
$epoch = (string) get_option( self::EPOCH_OPTION, '' ); |
| 128 |
if ( '' !== $epoch ) { |
| 129 |
return $epoch; |
| 130 |
} |
| 131 |
|
| 132 |
$epoch = $this->mint_epoch(); |
| 133 |
add_option( self::EPOCH_OPTION, $epoch, '', true ); |
| 134 |
return (string) get_option( self::EPOCH_OPTION, $epoch ); |
| 135 |
} |
| 136 |
|
| 137 |
/** Force a new id for the current sequence generation. */ |
| 138 |
public function regenerate_epoch(): string { |
| 139 |
$epoch = $this->mint_epoch(); |
| 140 |
update_option( self::EPOCH_OPTION, $epoch, true ); |
| 141 |
return $epoch; |
| 142 |
} |
| 143 |
|
| 144 |
private function mint_epoch(): string { |
| 145 |
return function_exists( 'wp_generate_uuid4' ) ? wp_generate_uuid4() : md5( uniqid( 'wcpos-epoch', true ) ); |
| 146 |
} |
| 147 |
|
| 148 |
/** Append a schema-upgrade customer row for every live user. */ |
| 149 |
public function append_customer_updates_for_all_users(): bool { |
| 150 |
global $wpdb; |
| 151 |
$now = gmdate( 'Y-m-d H:i:s' ); |
| 152 |
|
| 153 |
return false !== $wpdb->query( |
| 154 |
$wpdb->prepare( |
| 155 |
'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, deleted, revision, modified_gmt, origin, created_gmt)' |
| 156 |
. " SELECT 'customer', ID, 0, '', %s, 'schema-upgrade', %s FROM " . $wpdb->users, |
| 157 |
$now, |
| 158 |
$now |
| 159 |
) |
| 160 |
); |
| 161 |
} |
| 162 |
|
| 163 |
/** |
| 164 |
* Append one tombstone per catalogue post id, in a single statement. |
| 165 |
* |
| 166 |
* The per-record `record_post_deleted()` path loads a `WC_Product` for the revision stamp, which |
| 167 |
* is fine for the handful of records one settings write moves and hopeless for the whole hidden |
| 168 |
* set of a store that keeps thousands of products online-only. This is the bulk form, shaped like |
| 169 |
* `append_customer_updates_for_all_users()`: one INSERT ... SELECT, no revision (a tombstone |
| 170 |
* carries no state the client compares), and the post type read from `wp_posts` so a stale id in |
| 171 |
* the merchant's list cannot announce a change to an unrelated record. |
| 172 |
* |
| 173 |
* @param int[] $ids Product / variation post ids. |
| 174 |
*/ |
| 175 |
public function append_catalogue_tombstones( array $ids ): bool { |
| 176 |
global $wpdb; |
| 177 |
|
| 178 |
$ids = array_values( |
| 179 |
array_unique( |
| 180 |
array_filter( |
| 181 |
array_map( 'intval', $ids ), |
| 182 |
static function ( int $id ): bool { |
| 183 |
return $id > 0; |
| 184 |
} |
| 185 |
) |
| 186 |
) |
| 187 |
); |
| 188 |
if ( array() === $ids ) { |
| 189 |
return true; |
| 190 |
} |
| 191 |
|
| 192 |
$now = gmdate( 'Y-m-d H:i:s' ); |
| 193 |
$placeholders = implode( ',', array_fill( 0, count( $ids ), '%d' ) ); |
| 194 |
|
| 195 |
return false !== $wpdb->query( |
| 196 |
$wpdb->prepare( |
| 197 |
'INSERT INTO ' . $this->table_name() |
| 198 |
. ' (object_type, object_id, deleted, revision, modified_gmt, origin, created_gmt)' |
| 199 |
. " SELECT CASE p.post_type WHEN 'product_variation' THEN 'variation' ELSE 'product' END," |
| 200 |
. " p.ID, 1, '', %s, 'visibility-seed', %s" |
| 201 |
. " FROM {$wpdb->posts} p" |
| 202 |
. " WHERE p.post_type IN ('product','product_variation') AND p.ID IN ({$placeholders})" // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- %d placeholder list generated from count(); the ids are bound below. |
| 203 |
. ' ORDER BY p.ID', |
| 204 |
$now, |
| 205 |
$now, |
| 206 |
...$ids |
| 207 |
) |
| 208 |
); |
| 209 |
} |
| 210 |
|
| 211 |
public function register_hooks(): void { |
| 212 |
add_action( 'woocommerce_new_product', array( $this, 'record_product_created' ), 10, 1 ); |
| 213 |
add_action( 'woocommerce_update_product', array( $this, 'record_product_updated' ), 10, 1 ); |
| 214 |
add_action( 'woocommerce_new_product_variation', array( $this, 'record_variation_created' ), 10, 1 ); |
| 215 |
add_action( 'woocommerce_update_product_variation', array( $this, 'record_variation_updated' ), 10, 1 ); |
| 216 |
add_action( 'woocommerce_new_coupon', array( $this, 'record_coupon_created' ), 10, 1 ); |
| 217 |
add_action( 'woocommerce_update_coupon', array( $this, 'record_coupon_updated' ), 10, 1 ); |
| 218 |
add_action( 'wp_trash_post', array( $this, 'record_post_deleted' ), 10, 1 ); |
| 219 |
add_action( 'before_delete_post', array( $this, 'record_post_deleted' ), 10, 1 ); |
| 220 |
add_action( 'untrashed_post', array( $this, 'record_post_untrashed' ), 10, 1 ); |
| 221 |
add_action( 'woocommerce_tax_rate_added', array( $this, 'record_tax_rate_created' ), 10, 1 ); |
| 222 |
add_action( 'woocommerce_tax_rate_updated', array( $this, 'record_tax_rate_updated' ), 10, 1 ); |
| 223 |
add_action( 'woocommerce_tax_rate_deleted', array( $this, 'record_tax_rate_deleted' ), 10, 1 ); |
| 224 |
add_action( 'created_term', array( $this, 'record_term_created' ), 10, 3 ); |
| 225 |
add_action( 'edited_term', array( $this, 'record_term_edited' ), 10, 3 ); |
| 226 |
add_action( 'delete_term', array( $this, 'record_term_deleted' ), 10, 3 ); |
| 227 |
add_action( 'added_term_meta', array( $this, 'record_term_meta_change' ), 10, 2 ); |
| 228 |
add_action( 'updated_term_meta', array( $this, 'record_term_meta_change' ), 10, 2 ); |
| 229 |
add_action( 'deleted_term_meta', array( $this, 'record_term_meta_deleted' ), 10, 2 ); |
| 230 |
add_action( 'user_register', array( $this, 'record_customer_created' ), 10, 1 ); |
| 231 |
add_action( 'woocommerce_created_customer', array( $this, 'record_customer_created_persisted' ), 10, 1 ); |
| 232 |
add_action( 'woocommerce_new_customer', array( $this, 'record_customer_created_persisted' ), 10, 1 ); |
| 233 |
add_action( 'profile_update', array( $this, 'record_customer_profile_update' ), 10, 2 ); |
| 234 |
add_action( 'set_user_role', array( $this, 'record_customer_role_change' ), 10, 3 ); |
| 235 |
add_action( 'add_user_role', array( $this, 'record_customer_role_added' ), 10, 2 ); |
| 236 |
add_action( 'remove_user_role', array( $this, 'record_customer_role_removed' ), 10, 2 ); |
| 237 |
add_action( 'woocommerce_update_customer', array( $this, 'record_customer_updated' ), 10, 1 ); |
| 238 |
add_action( 'delete_user', array( $this, 'record_customer_deleted' ), 10, 1 ); |
| 239 |
add_action( 'woocommerce_new_order', array( $this, 'record_order_created' ), 10, 1 ); |
| 240 |
add_action( 'woocommerce_update_order', array( $this, 'record_order_updated' ), 10, 1 ); |
| 241 |
add_action( 'woocommerce_before_trash_order', array( $this, 'record_order_deleted' ), 10, 1 ); |
| 242 |
add_action( 'woocommerce_before_delete_order', array( $this, 'record_order_deleted' ), 10, 1 ); |
| 243 |
add_action( 'woocommerce_untrash_order', array( $this, 'record_cot_order_untrashed' ), 10, 1 ); |
| 244 |
} |
| 245 |
|
| 246 |
public function record_product_created( int $product_id ): void { |
| 247 |
$this->record_catalogue_object( 'product', $product_id, false ); |
| 248 |
} |
| 249 |
|
| 250 |
public function record_product_updated( int $product_id ): void { |
| 251 |
$this->record_catalogue_object( 'product', $product_id, false ); |
| 252 |
} |
| 253 |
|
| 254 |
public function record_variation_created( int $variation_id ): void { |
| 255 |
$this->record_catalogue_object( 'variation', $variation_id, false ); |
| 256 |
$this->record_variation_parent( $variation_id ); |
| 257 |
} |
| 258 |
|
| 259 |
public function record_variation_updated( int $variation_id ): void { |
| 260 |
$this->record_catalogue_object( 'variation', $variation_id, false ); |
| 261 |
$this->record_variation_parent( $variation_id ); |
| 262 |
} |
| 263 |
|
| 264 |
/** Record the representation change to a variation's parent product. */ |
| 265 |
private function record_variation_parent( int $variation_id ): void { |
| 266 |
$parent_id = function_exists( 'wp_get_post_parent_id' ) ? (int) wp_get_post_parent_id( $variation_id ) : 0; |
| 267 |
if ( $parent_id > 0 ) { |
| 268 |
$this->record_catalogue_object( 'product', $parent_id, false ); |
| 269 |
} |
| 270 |
} |
| 271 |
|
| 272 |
public function record_post_deleted( int $post_id ): void { |
| 273 |
$post_type = get_post_type( $post_id ); |
| 274 |
if ( 'product' === $post_type ) { |
| 275 |
$this->record_catalogue_object( 'product', $post_id, true ); |
| 276 |
return; |
| 277 |
} |
| 278 |
if ( 'product_variation' === $post_type ) { |
| 279 |
$this->record_catalogue_object( 'variation', $post_id, true ); |
| 280 |
$this->record_variation_parent( $post_id ); |
| 281 |
return; |
| 282 |
} |
| 283 |
if ( 'shop_coupon' === $post_type ) { |
| 284 |
$this->record( 'coupon', $post_id, true ); |
| 285 |
return; |
| 286 |
} |
| 287 |
if ( 'shop_order' === $post_type ) { |
| 288 |
$this->record_order_deleted( $post_id ); |
| 289 |
} |
| 290 |
} |
| 291 |
|
| 292 |
public function record_post_untrashed( int $post_id ): void { |
| 293 |
$post_type = get_post_type( $post_id ); |
| 294 |
if ( 'product' === $post_type ) { |
| 295 |
$this->record_product_updated( $post_id ); |
| 296 |
return; |
| 297 |
} |
| 298 |
if ( 'product_variation' === $post_type ) { |
| 299 |
$this->record_variation_updated( $post_id ); |
| 300 |
return; |
| 301 |
} |
| 302 |
if ( 'shop_coupon' === $post_type ) { |
| 303 |
$this->record_coupon_updated( $post_id ); |
| 304 |
return; |
| 305 |
} |
| 306 |
if ( 'shop_order' === $post_type ) { |
| 307 |
$this->record_order_untrashed( $post_id ); |
| 308 |
} |
| 309 |
} |
| 310 |
|
| 311 |
public function record_coupon_created( int $coupon_id ): void { |
| 312 |
$this->record( 'coupon', $coupon_id, false ); |
| 313 |
} |
| 314 |
|
| 315 |
public function record_coupon_updated( int $coupon_id ): void { |
| 316 |
$this->record( 'coupon', $coupon_id, false ); |
| 317 |
} |
| 318 |
|
| 319 |
/** Map tracked product taxonomies to journal object types. */ |
| 320 |
private static function term_taxonomy_object_types(): array { |
| 321 |
$map = array(); |
| 322 |
foreach ( Collections::with( 'identity' ) as $row ) { |
| 323 |
if ( isset( $row['identity']['taxonomy'] ) ) { |
| 324 |
$map[ $row['identity']['taxonomy'] ] = $row['object_type']; |
| 325 |
} |
| 326 |
} |
| 327 |
return $map; |
| 328 |
} |
| 329 |
|
| 330 |
public function record_term_created( int $term_id, int $tt_id, string $taxonomy ): void { |
| 331 |
$this->record_term_change( $term_id, $taxonomy, 'create' ); |
| 332 |
} |
| 333 |
|
| 334 |
public function record_term_edited( int $term_id, int $tt_id, string $taxonomy ): void { |
| 335 |
$this->record_term_change( $term_id, $taxonomy, 'update' ); |
| 336 |
} |
| 337 |
|
| 338 |
public function record_term_deleted( int $term_id, int $tt_id, string $taxonomy ): void { |
| 339 |
$this->record_term_change( $term_id, $taxonomy, 'delete' ); |
| 340 |
} |
| 341 |
|
| 342 |
private function record_term_change( int $term_id, string $taxonomy, string $change_type ): void { |
| 343 |
$object_type = self::term_taxonomy_object_types()[ $taxonomy ] ?? null; |
| 344 |
if ( null !== $object_type ) { |
| 345 |
$this->record( $object_type, $term_id, 'delete' === $change_type ); |
| 346 |
} |
| 347 |
} |
| 348 |
|
| 349 |
/** Record added or updated metadata for a tracked term. */ |
| 350 |
public function record_term_meta_change( int $meta_id, int $term_id ): void { |
| 351 |
$this->record_term_representation_change( $term_id ); |
| 352 |
} |
| 353 |
|
| 354 |
/** Record deleted metadata for a tracked term. */ |
| 355 |
public function record_term_meta_deleted( array $meta_ids, int $term_id ): void { |
| 356 |
$this->record_term_representation_change( $term_id ); |
| 357 |
} |
| 358 |
|
| 359 |
/** Resolve a meta hook's term and record it only when tracked. */ |
| 360 |
private function record_term_representation_change( int $term_id ): void { |
| 361 |
$term = get_term( $term_id ); |
| 362 |
if ( is_object( $term ) && isset( $term->taxonomy ) ) { |
| 363 |
$this->record_term_change( $term_id, (string) $term->taxonomy, 'update' ); |
| 364 |
} |
| 365 |
} |
| 366 |
|
| 367 |
public function record_tax_rate_created( int $tax_rate_id ): void { |
| 368 |
$this->record( 'tax_rate', $tax_rate_id, false ); |
| 369 |
} |
| 370 |
|
| 371 |
public function record_tax_rate_updated( int $tax_rate_id ): void { |
| 372 |
$this->record( 'tax_rate', $tax_rate_id, false ); |
| 373 |
} |
| 374 |
|
| 375 |
public function record_tax_rate_deleted( int $tax_rate_id ): void { |
| 376 |
$this->record( 'tax_rate', $tax_rate_id, true ); |
| 377 |
} |
| 378 |
|
| 379 |
public function record_customer_created( int $customer_id ): void { |
| 380 |
$this->record_customer( $customer_id, false, true, 'create' ); |
| 381 |
} |
| 382 |
|
| 383 |
/** WooCommerce create hooks for any user — definitive post-persist create with persisted dedup. */ |
| 384 |
public function record_customer_created_persisted( int $customer_id ): void { |
| 385 |
$this->record_customer( $customer_id, false, false, 'create' ); |
| 386 |
} |
| 387 |
|
| 388 |
/** Record the definitive post-persist customer update. */ |
| 389 |
public function record_customer_updated( int $customer_id ): void { |
| 390 |
$this->record_customer( $customer_id, false, false ); |
| 391 |
} |
| 392 |
|
| 393 |
/** Record a WordPress profile update as a customer update. */ |
| 394 |
public function record_customer_profile_update( int $user_id, $old_user_data = null ): void { |
| 395 |
$this->record_customer( $user_id, false ); |
| 396 |
} |
| 397 |
|
| 398 |
/** Record a customer role replacement. */ |
| 399 |
public function record_customer_role_change( int $user_id, $role = '', $old_roles = array() ): void { |
| 400 |
$this->record_customer( $user_id, false ); |
| 401 |
} |
| 402 |
|
| 403 |
/** add_user_role handler — any added role changes the served customer record. */ |
| 404 |
public function record_customer_role_added( int $user_id, $role = '' ): void { |
| 405 |
$this->record_customer( $user_id, false ); |
| 406 |
} |
| 407 |
|
| 408 |
/** Record a removed customer role as a present update. */ |
| 409 |
public function record_customer_role_removed( int $user_id, $role = '' ): void { |
| 410 |
$this->record_customer( $user_id, false ); |
| 411 |
} |
| 412 |
|
| 413 |
/** delete_user handler — deletion removes any user from the POS customer space. */ |
| 414 |
public function record_customer_deleted( int $customer_id ): void { |
| 415 |
$this->record_customer( $customer_id, true, true, 'delete' ); |
| 416 |
} |
| 417 |
|
| 418 |
public function record_order_created( int $order_id ): void { |
| 419 |
$this->record_order_change( $order_id, 'hook:create', false ); |
| 420 |
} |
| 421 |
|
| 422 |
public function record_order_updated( int $order_id ): void { |
| 423 |
$this->record_order_change( $order_id, 'hook:update', false ); |
| 424 |
} |
| 425 |
|
| 426 |
public function record_order_deleted( int $order_id ): void { |
| 427 |
$this->record_order_change( $order_id, 'hook:delete', true ); |
| 428 |
} |
| 429 |
|
| 430 |
public function record_order_untrashed( int $order_id ): void { |
| 431 |
$this->record_order_change( $order_id, 'hook:untrash', false ); |
| 432 |
} |
| 433 |
|
| 434 |
/** |
| 435 |
* Record an HPOS order's restore once the status change has settled. |
| 436 |
* |
| 437 |
* `woocommerce_untrash_order` fires BEFORE the data store restores the |
| 438 |
* status, so the row cannot be written there. The restore then performs |
| 439 |
* MORE THAN ONE object save, so arming on the first |
| 440 |
* `woocommerce_after_order_object_save` whose status is not `trash` |
| 441 |
* captures a revision from part-way through the restore — anything a later |
| 442 |
* save changes is missing from it, and the journal advertises a revision |
| 443 |
* the order does not have. |
| 444 |
* |
| 445 |
* Measured sequence for an HPOS untrash (status read from wc_orders): |
| 446 |
* |
| 447 |
* woocommerce_untrash_order stored=trash |
| 448 |
* after_order_object_save object=pending stored=wc-pending |
| 449 |
* after_order_object_save object=pending stored=wc-pending |
| 450 |
* woocommerce_order_status_changed stored=wc-pending from=trash |
| 451 |
* |
| 452 |
* `woocommerce_order_status_changed` fires once, last, with the stored |
| 453 |
* status settled — so observe that instead. CPT orders never reach here: |
| 454 |
* their restore fires only `untrashed_post` (see record_post_untrashed). |
| 455 |
* |
| 456 |
* @param int $order_id Order being restored. |
| 457 |
*/ |
| 458 |
public function record_cot_order_untrashed( int $order_id ): void { |
| 459 |
$handler = function ( $id, $from ) use ( $order_id, &$handler ): void { |
| 460 |
if ( (int) $id !== $order_id || 'trash' !== $from ) { |
| 461 |
return; |
| 462 |
} |
| 463 |
remove_action( 'woocommerce_order_status_changed', $handler ); |
| 464 |
$this->record_order_untrashed( $order_id ); |
| 465 |
}; |
| 466 |
add_action( 'woocommerce_order_status_changed', $handler, 10, 2 ); |
| 467 |
} |
| 468 |
|
| 469 |
public function record_order_change( int $order_id, string $origin, bool $deleted ): bool { |
| 470 |
global $wpdb; |
| 471 |
$order = wc_get_order( $order_id ); |
| 472 |
$modified_date = $order ? $order->get_date_modified() : null; |
| 473 |
$modified = $modified_date ? gmdate( 'Y-m-d H:i:s', $modified_date->getTimestamp() ) : gmdate( 'Y-m-d H:i:s' ); |
| 474 |
$revision = 'deleted'; |
| 475 |
|
| 476 |
if ( $order && ! $deleted ) { |
| 477 |
$serializer = new Order_Serializer(); |
| 478 |
$payload = $serializer->serialize_order( $order_id, new WP_REST_Request() ); |
| 479 |
$sync_meta = $serializer->sync_metadata( $payload, $order_id, 'custom-pull', false, 0 ); |
| 480 |
$revision = (string) $sync_meta['revision']; |
| 481 |
} |
| 482 |
|
| 483 |
$now = gmdate( 'Y-m-d H:i:s' ); |
| 484 |
return false !== $wpdb->insert( |
| 485 |
$this->table_name(), |
| 486 |
array( |
| 487 |
'object_type' => 'order', |
| 488 |
'object_id' => $order_id, |
| 489 |
'deleted' => $deleted ? 1 : 0, |
| 490 |
'revision' => $revision, |
| 491 |
'modified_gmt' => $modified, |
| 492 |
'origin' => $origin, |
| 493 |
'created_gmt' => $now, |
| 494 |
), |
| 495 |
array( '%s', '%d', '%d', '%s', '%s', '%s', '%s' ) |
| 496 |
); |
| 497 |
} |
| 498 |
|
| 499 |
private function record_catalogue_object( string $object_type, int $object_id, bool $deleted ): void { |
| 500 |
$object = function_exists( 'wc_get_product' ) ? wc_get_product( $object_id ) : null; |
| 501 |
$this->record( $object_type, $object_id, $deleted, self::object_revision( $object ) ); |
| 502 |
} |
| 503 |
|
| 504 |
private function record_customer( int $customer_id, bool $deleted, bool $dedup = true, string $dedup_namespace = 'update' ): void { |
| 505 |
$customer = class_exists( '\\WC_Customer' ) ? new \WC_Customer( $customer_id ) : null; |
| 506 |
$this->record( 'customer', $customer_id, $deleted, self::object_revision( $customer ), 'hook', $dedup, $dedup_namespace ); |
| 507 |
} |
| 508 |
|
| 509 |
private static function object_revision( $object ): string { |
| 510 |
$date = is_object( $object ) && method_exists( $object, 'get_date_modified' ) ? $object->get_date_modified() : null; |
| 511 |
return $date && method_exists( $date, 'getTimestamp' ) ? gmdate( 'Y-m-d H:i:s', $date->getTimestamp() ) : ''; |
| 512 |
} |
| 513 |
|
| 514 |
public function record( string $object_type, int $object_id, bool $deleted, string $revision = '', string $origin = 'hook', bool $dedup = true, string $dedup_namespace = '' ): void { |
| 515 |
global $wpdb; |
| 516 |
$dedup_key = null; |
| 517 |
if ( 'customer' === $object_type ) { |
| 518 |
$dedup_key = ( $dedup ? '' : 'persisted:' ) . $object_id . ':' . ( '' !== $dedup_namespace ? $dedup_namespace : ( $deleted ? 'delete' : 'update' ) ); |
| 519 |
if ( isset( $this->recorded_this_request[ $dedup_key ] ) ) { |
| 520 |
if ( $dedup ) { |
| 521 |
return; // a pre-persist duplicate carries no new state |
| 522 |
} |
| 523 |
$wpdb->delete( |
| 524 |
$this->table_name(), |
| 525 |
array( 'sequence' => (int) $this->recorded_this_request[ $dedup_key ] ), |
| 526 |
array( '%d' ) |
| 527 |
); |
| 528 |
} |
| 529 |
} |
| 530 |
$started = microtime( true ); |
| 531 |
$now = gmdate( 'Y-m-d H:i:s' ); |
| 532 |
$wpdb->insert( |
| 533 |
$this->table_name(), |
| 534 |
array( |
| 535 |
'object_type' => $object_type, |
| 536 |
'object_id' => $object_id, |
| 537 |
'deleted' => $deleted ? 1 : 0, |
| 538 |
'revision' => $revision, |
| 539 |
'modified_gmt' => $now, |
| 540 |
'origin' => $origin, |
| 541 |
'created_gmt' => $now, |
| 542 |
), |
| 543 |
array( '%s', '%d', '%d', '%s', '%s', '%s', '%s' ) |
| 544 |
); |
| 545 |
if ( null !== $dedup_key ) { |
| 546 |
$this->recorded_this_request[ $dedup_key ] = $dedup ? true : (int) $wpdb->insert_id; |
| 547 |
} |
| 548 |
self::$request_write_ms += ( microtime( true ) - $started ) * 1000; |
| 549 |
} |
| 550 |
|
| 551 |
/** Return the persisted order-backfill cursor. */ |
| 552 |
public function backfill_status(): array { |
| 553 |
$status = get_option( self::BACKFILL_OPTION, array() ); |
| 554 |
$status = is_array( $status ) ? $status : array(); |
| 555 |
|
| 556 |
return array( |
| 557 |
'status' => isset( $status['status'] ) ? (string) $status['status'] : 'idle', |
| 558 |
'nextPage' => isset( $status['nextPage'] ) ? max( 1, (int) $status['nextPage'] ) : 1, |
| 559 |
'pageSize' => isset( $status['pageSize'] ) ? max( 1, min( 250, (int) $status['pageSize'] ) ) : null, |
| 560 |
'processed' => isset( $status['processed'] ) ? max( 0, (int) $status['processed'] ) : 0, |
| 561 |
'lastOrderId' => isset( $status['lastOrderId'] ) ? max( 0, (int) $status['lastOrderId'] ) : 0, |
| 562 |
'lastRunGmt' => isset( $status['lastRunGmt'] ) ? (string) $status['lastRunGmt'] : null, |
| 563 |
); |
| 564 |
} |
| 565 |
|
| 566 |
/** Clear the full persisted order-backfill cursor. */ |
| 567 |
public function reset_backfill_state(): void { |
| 568 |
delete_option( self::BACKFILL_OPTION ); |
| 569 |
} |
| 570 |
|
| 571 |
/** Append one bounded page of existing orders to the journal. */ |
| 572 |
public function run_backfill_chunk( int $limit ): array { |
| 573 |
$requested_limit = max( 1, min( 250, $limit ) ); |
| 574 |
$status = $this->backfill_status(); |
| 575 |
if ( 'complete' === $status['status'] ) { |
| 576 |
return array_merge( $status, array( 'processedThisRun' => 0 ) ); |
| 577 |
} |
| 578 |
|
| 579 |
$page_size = null === $status['pageSize'] ? $requested_limit : (int) $status['pageSize']; |
| 580 |
$last_order_id = $status['lastOrderId']; |
| 581 |
$query_args = array( |
| 582 |
'type' => 'shop_order', |
| 583 |
'limit' => $page_size, |
| 584 |
'orderby' => 'ID', |
| 585 |
'order' => 'ASC', |
| 586 |
'return' => 'ids', |
| 587 |
); |
| 588 |
$posts_where = null; |
| 589 |
if ( $last_order_id > 0 ) { |
| 590 |
if ( class_exists( OrderUtil::class ) && OrderUtil::custom_orders_table_usage_is_enabled() ) { |
| 591 |
$query_args['field_query'] = array( |
| 592 |
array( |
| 593 |
'field' => 'id', |
| 594 |
'value' => $last_order_id, |
| 595 |
'compare' => '>', |
| 596 |
), |
| 597 |
); |
| 598 |
} else { |
| 599 |
$posts_where = static function ( string $where ) use ( $last_order_id ): string { |
| 600 |
global $wpdb; |
| 601 |
return $where . $wpdb->prepare( " AND {$wpdb->posts}.ID > %d", $last_order_id ); |
| 602 |
}; |
| 603 |
add_filter( 'posts_where', $posts_where ); |
| 604 |
} |
| 605 |
} |
| 606 |
try { |
| 607 |
/** @var array<int|numeric-string>|mixed $queried_ids The 'return' => 'ids' arg yields ids; the stub over-narrows to WC_Order[]. */ |
| 608 |
$queried_ids = wc_get_orders( $query_args ); |
| 609 |
} finally { |
| 610 |
if ( null !== $posts_where ) { |
| 611 |
remove_filter( 'posts_where', $posts_where ); |
| 612 |
} |
| 613 |
} |
| 614 |
$ids = is_array( $queried_ids ) ? array_map( 'absint', $queried_ids ) : array(); |
| 615 |
$processed_this_run = 0; |
| 616 |
$failed_this_run = 0; |
| 617 |
foreach ( $ids as $id ) { |
| 618 |
if ( $this->record_order_change( $id, 'backfill', false ) ) { |
| 619 |
$processed_this_run++; |
| 620 |
$last_order_id = $id; |
| 621 |
} else { |
| 622 |
$failed_this_run++; |
| 623 |
break; |
| 624 |
} |
| 625 |
} |
| 626 |
|
| 627 |
$all_writes_succeeded = 0 === $failed_this_run; |
| 628 |
$complete = $all_writes_succeeded && count( $ids ) < $page_size; |
| 629 |
$advance_page = $all_writes_succeeded && count( $ids ) === $page_size; |
| 630 |
$next_status = array( |
| 631 |
'status' => $complete ? 'complete' : 'running', |
| 632 |
'nextPage' => $advance_page ? $status['nextPage'] + 1 : $status['nextPage'], |
| 633 |
'pageSize' => $page_size, |
| 634 |
'processed' => $status['processed'] + $processed_this_run, |
| 635 |
'lastOrderId' => $last_order_id, |
| 636 |
'lastRunGmt' => gmdate( 'c' ), |
| 637 |
); |
| 638 |
update_option( self::BACKFILL_OPTION, $next_status, false ); |
| 639 |
|
| 640 |
return array_merge( $next_status, array( 'processedThisRun' => $processed_this_run ) ); |
| 641 |
} |
| 642 |
|
| 643 |
/** |
| 644 |
* The catalogue pointer-stream's object types, projected from the registry — |
| 645 |
* every journal-covered collection except orders (which consume the journal |
| 646 |
* via the payload-windowed pull lane). Single source for the sequence-log |
| 647 |
* `all` stream and the purge's per-stream head protection. |
| 648 |
* |
| 649 |
* @return string[] |
| 650 |
*/ |
| 651 |
public static function catalogue_object_types(): array { |
| 652 |
$types = array(); |
| 653 |
foreach ( Collections::with( 'journal' ) as $row ) { |
| 654 |
$object_type = (string) ( $row['journal']['object_type'] ?? '' ); |
| 655 |
if ( '' !== $object_type && 'order' !== $object_type ) { |
| 656 |
$types[] = $object_type; |
| 657 |
} |
| 658 |
} |
| 659 |
|
| 660 |
return $types; |
| 661 |
} |
| 662 |
|
| 663 |
/** |
| 664 |
* Head of the sequence space — the change stream's current end. |
| 665 |
* |
| 666 |
* With `$object_types`, the head is STREAM-SCOPED: the highest sequence any |
| 667 |
* row of those types holds. Every reader must serve the head of the stream |
| 668 |
* it serves — orders and catalogue share one AUTO_INCREMENT space, so the |
| 669 |
* global head moves on foreign writes. A stream-scoped head is what lets a |
| 670 |
* cursor actually reach `head` (the 304 idle condition) while the other |
| 671 |
* lane keeps writing, and is the pre-unification semantic of both lanes. |
| 672 |
* |
| 673 |
* @param string[] $object_types Empty = global head (retention clamps only). |
| 674 |
*/ |
| 675 |
public function head_sequence( array $object_types = array() ): int { |
| 676 |
global $wpdb; |
| 677 |
if ( ! $this->table_available() ) { |
| 678 |
return 0; |
| 679 |
} |
| 680 |
|
| 681 |
$types = array_values( array_filter( array_map( 'strval', $object_types ), static fn( string $t ): bool => '' !== $t ) ); |
| 682 |
if ( array() === $types ) { |
| 683 |
return (int) $wpdb->get_var( 'SELECT MAX(sequence) FROM ' . $this->table_name() ); |
| 684 |
} |
| 685 |
|
| 686 |
$placeholders = implode( ',', array_fill( 0, count( $types ), '%s' ) ); |
| 687 |
|
| 688 |
return (int) $wpdb->get_var( |
| 689 |
$wpdb->prepare( |
| 690 |
'SELECT MAX(sequence) FROM ' . $this->table_name() . " WHERE object_type IN ({$placeholders})", // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- %s placeholder list generated from count(). |
| 691 |
...$types |
| 692 |
) |
| 693 |
); |
| 694 |
} |
| 695 |
|
| 696 |
public function table_available(): bool { |
| 697 |
global $wpdb; |
| 698 |
$table = $this->table_name(); |
| 699 |
return $table === $wpdb->get_var( $wpdb->prepare( 'SHOW TABLES LIKE %s', $table ) ); |
| 700 |
} |
| 701 |
|
| 702 |
/** Return one sequence page for the order pull lane. */ |
| 703 |
public function rows_after_sequence( int $sequence, int $limit, string $object_type = 'order' ): array { |
| 704 |
global $wpdb; |
| 705 |
if ( ! $this->table_available() ) { |
| 706 |
return array(); |
| 707 |
} |
| 708 |
|
| 709 |
$rows = $wpdb->get_results( |
| 710 |
$wpdb->prepare( |
| 711 |
'SELECT sequence, object_id AS order_id, modified_gmt, revision, deleted, origin, created_gmt FROM ' . $this->table_name() |
| 712 |
. ' WHERE object_type = %s AND sequence > %d ORDER BY sequence ASC LIMIT %d', |
| 713 |
$object_type, |
| 714 |
max( 0, $sequence ), |
| 715 |
max( 1, min( 251, $limit ) ) |
| 716 |
), |
| 717 |
ARRAY_A |
| 718 |
); |
| 719 |
|
| 720 |
return is_array( $rows ) ? array_map( array( self::class, 'normalize_order_row' ), $rows ) : array(); |
| 721 |
} |
| 722 |
|
| 723 |
private static function normalize_order_row( array $row ): array { |
| 724 |
return array( |
| 725 |
'sequence' => isset( $row['sequence'] ) ? (int) $row['sequence'] : 0, |
| 726 |
'order_id' => isset( $row['order_id'] ) ? (int) $row['order_id'] : 0, |
| 727 |
'modified_gmt' => isset( $row['modified_gmt'] ) ? (string) $row['modified_gmt'] : gmdate( 'Y-m-d H:i:s' ), |
| 728 |
'revision' => isset( $row['revision'] ) ? (string) $row['revision'] : '', |
| 729 |
'deleted' => ! empty( $row['deleted'] ) ? 1 : 0, |
| 730 |
'origin' => isset( $row['origin'] ) ? (string) $row['origin'] : 'hook:update', |
| 731 |
'created_gmt' => isset( $row['created_gmt'] ) ? (string) $row['created_gmt'] : gmdate( 'Y-m-d H:i:s' ), |
| 732 |
); |
| 733 |
} |
| 734 |
|
| 735 |
/** Oldest sequence that remains available to incremental-sync clients. */ |
| 736 |
public function oldest_sequence(): int { |
| 737 |
global $wpdb; |
| 738 |
|
| 739 |
return (int) $wpdb->get_var( 'SELECT MIN(sequence) FROM ' . $this->table_name() ); |
| 740 |
} |
| 741 |
|
| 742 |
/** Delete one batch of superseded rows through the supplied sequence. */ |
| 743 |
public function compact( int $cutoff_sequence, string $cutoff_gmt, int $batch ): int { |
| 744 |
global $wpdb; |
| 745 |
if ( $cutoff_sequence <= 0 || $batch <= 0 ) { |
| 746 |
return 0; |
| 747 |
} |
| 748 |
|
| 749 |
$table = $this->table_name(); |
| 750 |
$deleted = $wpdb->query( |
| 751 |
$wpdb->prepare( |
| 752 |
'DELETE FROM ' . $table . ' WHERE sequence <= %d AND sequence IN (' |
| 753 |
. ' SELECT sequence FROM (' |
| 754 |
. ' SELECT stale.sequence FROM ' . $table . ' stale' |
| 755 |
. ' WHERE stale.sequence <= %d' |
| 756 |
. ' AND stale.created_gmt < %s' |
| 757 |
. ' AND EXISTS (' |
| 758 |
. ' SELECT 1 FROM ' . $table . ' newer' |
| 759 |
. ' WHERE newer.object_type = stale.object_type' |
| 760 |
. ' AND newer.object_id = stale.object_id' |
| 761 |
. ' AND newer.sequence > stale.sequence' |
| 762 |
. ' ) ORDER BY stale.sequence ASC LIMIT %d' |
| 763 |
. ' ) compactable' |
| 764 |
. ' )', |
| 765 |
$cutoff_sequence, |
| 766 |
$cutoff_sequence, |
| 767 |
$cutoff_gmt, |
| 768 |
$batch |
| 769 |
) |
| 770 |
); |
| 771 |
|
| 772 |
return false === $deleted ? 0 : (int) $deleted; |
| 773 |
} |
| 774 |
|
| 775 |
/** |
| 776 |
* Delete one batch of expired tombstones — the log's only LOSSY deletion. |
| 777 |
* |
| 778 |
* @param int $cutoff_sequence Highest sequence this batch may remove. |
| 779 |
* @param string $cutoff_gmt Rows created before this are expired. |
| 780 |
* @param int $batch Maximum rows to remove. |
| 781 |
* @param array $object_types Restrict to one stream's types; empty = every type. |
| 782 |
*/ |
| 783 |
public function prune_tombstones( int $cutoff_sequence, string $cutoff_gmt, int $batch, array $object_types = array() ): array { |
| 784 |
global $wpdb; |
| 785 |
$none = array( |
| 786 |
'deleted' => 0, |
| 787 |
'watermark' => 0, |
| 788 |
); |
| 789 |
if ( $cutoff_sequence <= 0 || $batch <= 0 ) { |
| 790 |
return $none; |
| 791 |
} |
| 792 |
|
| 793 |
// Each stream is pruned under its OWN cutoff (Sync_Journal_Purge), so the |
| 794 |
// batch must not reach across streams: an order tombstone is not eligible |
| 795 |
// merely because the catalogue's cutoff cleared it. |
| 796 |
$types = array_values( array_unique( array_filter( array_map( 'strval', $object_types ), static fn( string $type ): bool => '' !== $type ) ) ); |
| 797 |
$type_where = ''; |
| 798 |
$type_args = array(); |
| 799 |
if ( array() !== $types ) { |
| 800 |
$type_where = ' AND object_type IN (' . implode( ',', array_fill( 0, \count( $types ), '%s' ) ) . ')'; |
| 801 |
$type_args = $types; |
| 802 |
} |
| 803 |
|
| 804 |
$rows = $wpdb->get_results( |
| 805 |
$wpdb->prepare( |
| 806 |
'SELECT sequence, object_type FROM ' . $this->table_name() |
| 807 |
. ' WHERE deleted = 1 AND sequence <= %d AND created_gmt < %s' . $type_where // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- %s placeholder list generated from count(). |
| 808 |
. ' ORDER BY sequence ASC LIMIT %d', |
| 809 |
$cutoff_sequence, |
| 810 |
$cutoff_gmt, |
| 811 |
...array_merge( $type_args, array( $batch ) ) |
| 812 |
), |
| 813 |
ARRAY_A |
| 814 |
); |
| 815 |
if ( empty( $rows ) ) { |
| 816 |
return $none; |
| 817 |
} |
| 818 |
|
| 819 |
// Each pruned row raises ONLY its own object type's watermark: a client |
| 820 |
// reading a stream that never held these rows has missed nothing. |
| 821 |
$sequences = array(); |
| 822 |
$per_type = array(); |
| 823 |
foreach ( $rows as $row ) { |
| 824 |
$sequence = (int) $row['sequence']; |
| 825 |
$object_type = (string) $row['object_type']; |
| 826 |
$sequences[] = $sequence; |
| 827 |
$per_type[ $object_type ] = max( $per_type[ $object_type ] ?? 0, $sequence ); |
| 828 |
} |
| 829 |
$watermark = max( $sequences ); |
| 830 |
|
| 831 |
// Publish every watermark BEFORE deleting anything. A row deleted while |
| 832 |
// its type's horizon still reads below it is silently lost history — no |
| 833 |
// client would ever learn to reconcile it — so a failed write aborts the |
| 834 |
// whole batch and the next run retries it. |
| 835 |
foreach ( $per_type as $object_type => $sequence ) { |
| 836 |
$this->advance_prune_watermark( $object_type, $sequence ); |
| 837 |
if ( $this->prune_watermark( array( $object_type ) ) < $sequence ) { |
| 838 |
return $none; |
| 839 |
} |
| 840 |
} |
| 841 |
|
| 842 |
$deleted = $wpdb->query( |
| 843 |
'DELETE FROM ' . $this->table_name() |
| 844 |
. ' WHERE sequence IN (' . implode( ',', $sequences ) . ')' |
| 845 |
); |
| 846 |
|
| 847 |
return array( |
| 848 |
'deleted' => false === $deleted ? 0 : (int) $deleted, |
| 849 |
'watermark' => $watermark, |
| 850 |
); |
| 851 |
} |
| 852 |
|
| 853 |
/** Resolve a wall-clock cutoff to one stable sequence boundary. */ |
| 854 |
public function sequence_at_or_before( string $cutoff_gmt ): int { |
| 855 |
global $wpdb; |
| 856 |
|
| 857 |
return (int) $wpdb->get_var( |
| 858 |
$wpdb->prepare( |
| 859 |
'SELECT MAX(sequence) FROM ' . $this->table_name() . ' WHERE created_gmt < %s', |
| 860 |
$cutoff_gmt |
| 861 |
) |
| 862 |
); |
| 863 |
} |
| 864 |
|
| 865 |
/** Option holding one object type's lossy-prune watermark. */ |
| 866 |
public static function prune_watermark_option( string $object_type ): string { |
| 867 |
return self::PRUNE_WATERMARK_OPTION_PREFIX . $object_type; |
| 868 |
} |
| 869 |
|
| 870 |
/** |
| 871 |
* Highest sequence ever removed by lossy tombstone pruning FROM A STREAM. |
| 872 |
* |
| 873 |
* Mirrors head_sequence(): the caller names the object types its stream |
| 874 |
* serves and gets that stream's boundary. An empty list means every |
| 875 |
* registered type — the whole journal's boundary. |
| 876 |
* |
| 877 |
* @param array $object_types Object types the reading stream serves. |
| 878 |
*/ |
| 879 |
public function prune_watermark( array $object_types = array() ): int { |
| 880 |
$watermark = 0; |
| 881 |
foreach ( self::watermark_object_types( $object_types ) as $object_type ) { |
| 882 |
$watermark = max( $watermark, (int) get_option( self::prune_watermark_option( $object_type ), 0 ) ); |
| 883 |
} |
| 884 |
|
| 885 |
return $watermark; |
| 886 |
} |
| 887 |
|
| 888 |
/** |
| 889 |
* Advance one object type's persisted watermark (never moves backwards). |
| 890 |
* |
| 891 |
* @param string $object_type Journal object type the pruned rows belonged to. |
| 892 |
* @param int $sequence Highest sequence pruned for that type. |
| 893 |
*/ |
| 894 |
public function advance_prune_watermark( string $object_type, int $sequence ): void { |
| 895 |
global $wpdb; |
| 896 |
if ( $sequence <= 0 || '' === $object_type ) { |
| 897 |
return; |
| 898 |
} |
| 899 |
|
| 900 |
$option = self::prune_watermark_option( $object_type ); |
| 901 |
$wpdb->query( |
| 902 |
$wpdb->prepare( |
| 903 |
'INSERT INTO ' . $wpdb->options . " (option_name, option_value, autoload) VALUES (%s, %d, 'yes')" |
| 904 |
. ' ON DUPLICATE KEY UPDATE option_value = GREATEST(CAST(option_value AS UNSIGNED), %d)', |
| 905 |
$option, |
| 906 |
$sequence, |
| 907 |
$sequence |
| 908 |
) |
| 909 |
); |
| 910 |
wp_cache_delete( $option, 'options' ); |
| 911 |
wp_cache_delete( 'alloptions', 'options' ); |
| 912 |
wp_cache_delete( 'notoptions', 'options' ); |
| 913 |
} |
| 914 |
|
| 915 |
/** Drop every per-type watermark — the journal's history is starting over. */ |
| 916 |
public static function reset_prune_watermarks(): void { |
| 917 |
global $wpdb; |
| 918 |
|
| 919 |
// Trailing separator trimmed from the LIKE so this also clears the single |
| 920 |
// pre-stream-scoping watermark a pre-release install may still carry. |
| 921 |
$names = $wpdb->get_col( |
| 922 |
$wpdb->prepare( |
| 923 |
"SELECT option_name FROM {$wpdb->options} WHERE option_name LIKE %s", |
| 924 |
$wpdb->esc_like( rtrim( self::PRUNE_WATERMARK_OPTION_PREFIX, '_' ) ) . '%' |
| 925 |
) |
| 926 |
); |
| 927 |
foreach ( (array) $names as $name ) { |
| 928 |
delete_option( (string) $name ); |
| 929 |
} |
| 930 |
} |
| 931 |
|
| 932 |
/** |
| 933 |
* Object types a watermark read covers: the named stream, or all of them. |
| 934 |
* |
| 935 |
* @param array $object_types Object types named by the caller. |
| 936 |
*/ |
| 937 |
private static function watermark_object_types( array $object_types ): array { |
| 938 |
$types = array_values( |
| 939 |
array_filter( |
| 940 |
array_map( 'strval', $object_types ), |
| 941 |
static fn( string $type ): bool => '' !== $type |
| 942 |
) |
| 943 |
); |
| 944 |
if ( array() !== $types ) { |
| 945 |
return array_unique( $types ); |
| 946 |
} |
| 947 |
|
| 948 |
return array_unique( array_merge( array( 'order' ), self::catalogue_object_types() ) ); |
| 949 |
} |
| 950 |
|
| 951 |
/** One page of the change stream past a cursor, plus the head it was read against. */ |
| 952 |
public function page( array $object_types, int $since, int $limit ): array { |
| 953 |
global $wpdb; |
| 954 |
$types = array(); |
| 955 |
foreach ( $object_types as $object_type ) { |
| 956 |
$object_type = (string) $object_type; |
| 957 |
if ( '' !== $object_type ) { |
| 958 |
$types[] = $object_type; |
| 959 |
} |
| 960 |
} |
| 961 |
|
| 962 |
$sql = 'SELECT sequence, object_id, object_type, deleted, revision, modified_gmt FROM ' . $this->table_name() . ' WHERE '; |
| 963 |
$args = array(); |
| 964 |
if ( array() !== $types ) { |
| 965 |
$sql .= 'object_type IN (' . implode( ',', array_fill( 0, count( $types ), '%s' ) ) . ') AND '; |
| 966 |
$args = $types; |
| 967 |
} |
| 968 |
$sql .= 'sequence > %d ORDER BY sequence ASC LIMIT %d'; |
| 969 |
$args[] = max( 0, $since ); |
| 970 |
$args[] = max( 1, $limit ); |
| 971 |
|
| 972 |
$rows = $wpdb->get_results( $wpdb->prepare( $sql, ...$args ), ARRAY_A ); |
| 973 |
|
| 974 |
return array( |
| 975 |
'rows' => array_map( |
| 976 |
static function ( array $row ): array { |
| 977 |
return self::normalize_row( $row ); |
| 978 |
}, |
| 979 |
is_array( $rows ) ? $rows : array() |
| 980 |
), |
| 981 |
'head' => $this->head_sequence( $types ), |
| 982 |
); |
| 983 |
} |
| 984 |
|
| 985 |
/** Coerce a raw journal row to the served scalar types. */ |
| 986 |
private static function normalize_row( array $row ): array { |
| 987 |
return array( |
| 988 |
'sequence' => isset( $row['sequence'] ) ? (int) $row['sequence'] : 0, |
| 989 |
'object_id' => isset( $row['object_id'] ) ? (int) $row['object_id'] : 0, |
| 990 |
'object_type' => isset( $row['object_type'] ) ? (string) $row['object_type'] : '', |
| 991 |
'deleted' => ! empty( $row['deleted'] ) ? 1 : 0, |
| 992 |
'revision' => isset( $row['revision'] ) ? (string) $row['revision'] : '', |
| 993 |
'modified_gmt' => isset( $row['modified_gmt'] ) ? (string) $row['modified_gmt'] : gmdate( 'Y-m-d H:i:s' ), |
| 994 |
); |
| 995 |
} |
| 996 |
} |
| 997 |
|
