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