woocommerce-pos / includes / Sync / Sync_Journal.php

Sync_Journal.php in WCPOS – Point of Sale (POS) plugin for WooCommerce 1.10.0, at includes/Sync/Sync_Journal.php

949 lines 35.6 KB

Raw Download Zip

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