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