woocommerce-pos / includes / Sync / Integrity_Digest.php

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

834 lines 34.1 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	use WCPOS\WooCommercePOS\Logger;
11
12	// phpcs:disable Squiz.Commenting, Generic.Commenting -- Ported lab documentation is preserved verbatim.
13	// phpcs:disable WordPress.DB.PreparedSQL.NotPrepared -- Queries use internal table names and generated SQL fragments.
14	// phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Database failures are passed to exceptions, not rendered.
15
16	use RuntimeException;
17
18	/**
19	* Hash-backed range-checksum support: stored per-record content digests.
20	*
21	* STORES a digest of each product/variation's raw DB row, marked dirty by the
22	* same save/delete hooks class-change-log.php uses and written at the request
23	* boundary (see $pending_digests), so the integrity scan
24	* can compare — entirely in SQL — the aggregate of CURRENT raw-row digests
25	* against the aggregate of STORED digests per id-range bucket. If hooks
26	* fired for every write, stored == current (and sequence-log already
27	* reported the change); a bucket mismatch therefore means exactly "content
28	* changed without hooks firing" — the sql-bypass signature — at GROUP BY
29	* prices instead of revision-hash's full-hydration prices. Because the write
30	* lands at the boundary, a hook-less write later in the SAME request is
31	* absorbed into that request's digest; the scan catches bypasses between
32	* requests, which is where they happen (a direct SQL job, an importer).
33	*
34	* The digest basis is deliberately the RAW DB ROW, NOT the filtered REST
35	* payload: this signal is detection-only (discovery of WHERE drift
36	* happened, ADR 0003 "discovery, never values"); hydration of anything the
37	* POS trusts still goes through the filtered REST path. The flip side is
38	* documented too: a raw-row digest cannot see a plugin changing the served
39	* representation without touching the row — that staleness case remains
40	* revision-hash territory.
41	*
42	* This class is the WRITE half. The READ half — every question the REST read
43	* surface asks of the store, plus the canonical digest SQL both halves share —
44	* lives in {@see Digest_Index}. The SQL-fragment accessors that used to hang off
45	* this class are kept as deprecated delegates so existing callers keep working.
46	*/
47	final class Integrity_Digest {
48
49	/**
50	* Wall-clock ms spent inside the digest write hooks during the CURRENT
51	* request. Read (and reset) by the product-edit fixture for the
52	* hook-overhead bench's per-component breakdown. Two microtime() calls
53	* per hook fire — negligible against the INSERT…SELECT it wraps.
54	*/
55	public static float $request_write_ms = 0.0;
56
57	/**
58	* @see Digest_Index::DIGESTED_META_KEYS The BASELINE key set.
59	* The formula the digest actually uses is Digest_Index::digested_meta_keys(),
60	* which folds in the configured barcode key (mono#1234).
61	*/
62	public const DIGESTED_META_KEYS = Digest_Index::DIGESTED_META_KEYS;
63
64	/** @see Digest_Index::CUSTOMER_DIGESTED_META_KEYS The digest formula's home. */
65	public const CUSTOMER_DIGESTED_META_KEYS = Digest_Index::CUSTOMER_DIGESTED_META_KEYS;
66
67	/** @see Digest_Index::ORDER_DIGESTED_META_KEYS The digest formula's home. */
68	public const ORDER_DIGESTED_META_KEYS = Digest_Index::ORDER_DIGESTED_META_KEYS;
69
70	/** @see Digest_Index::OBJECT_TYPES_SQL The product-space object types. */
71	public const OBJECT_TYPES_SQL = Digest_Index::OBJECT_TYPES_SQL;
72
73	public const REBUILD_HOOK = 'wcpos_integrity_digest_rebuild';
74	public const REBUILD_LOCK = 'wcpos_integrity_digest_rebuild_lock';
75	public const REBUILD_LOCK_TTL = 300;
76
77	/**
78	* The read half + the canonical digest SQL. The write statements below compose
79	* their INSERT…SELECT sources from it, so stored and current digests are
80	* computed by ONE expression — the invariant the whole scan rests on.
81	*/
82	private Digest_Index $index;
83
84	/**
85	* Order and customer digests owed but not yet written, keyed "type:id".
86	*
87	* A stored digest is a pure function of the settled record, so only the
88	* LAST upsert in a request carries information — yet one Store API checkout
89	* ran the order INSERT…SELECT eleven times (35 ms) and, with account
90	* creation, the customer one six more times (measured 2026-09-03 on
91	* dev-next). Upserts land on {@see flush_pending_digests()}: at `shutdown`
92	* (last, after WooCommerce's own customer save at 10 and session save at 20,
93	* whose `woocommerce_update_customer` would otherwise queue after the only
94	* flush), before any {@see Digest_Index::read_digests()} so the pull lane
95	* never stamps a stale `_rxdb_digest`, and whenever the queue reaches
96	* {@see PENDING_DIGEST_FLUSH_THRESHOLD} distinct records (a bulk import
97	* coalesces nothing, so it must not accumulate). Once the shutdown flush
98	* has run, later saves write immediately. Static so the read path can
99	* flush without holding the observer instance; each entry keeps its blog id
100	* so a multisite `switch_to_blog()` between save and flush still writes the
101	* originating site's table. Product and variation digests ride the same
102	* queue: WooCommerce saves a product more than once per request too
103	* (`wc_reduce_stock_levels()` saves the quantity, then the stock status —
104	* two INSERT…SELECT statements per purchased product at checkout, measured
105	* 2026-09-03), and the v2 write lane reads digests back through
106	* {@see Digest_Index::read_digests()}, which flushes first, so the
107	* serializer still stamps the fresh `_rxdb_digest` in the same request.
108	*
109	* @var array<string, array{0: int, 1: string, 2: int}> "blog:type:id" => [blog, type, id]
110	*/
111	private static array $pending_digests = array();
112
113	/**
114	* Flush the queue when it holds this many distinct records. Sized for the
115	* realistic per-request maximum (a checkout touches an order and a customer;
116	* a REST batch a few dozen records) while keeping a WP-CLI import's deferred
117	* SQL and memory bounded.
118	*/
119	public const PENDING_DIGEST_FLUSH_THRESHOLD = 50;
120
121	/** The instance that first queued a digest; the flush writes through its Digest_Index. */
122	private static ?Integrity_Digest $flusher = null;
123
124	/** Set by the shutdown flush; afterwards saves are written immediately. */
125	private static bool $shutdown_flushed = false;
126
127	public function __construct( ?Digest_Index $index = null ) {
128	$this->index = $index ?? new Digest_Index();
129	}
130
131	public function table_name(): string {
132	return $this->index->table_name();
133	}
134
135	/**
136	* Separate current-state table rather than a column on the change-log:
137	* the change-log is an append-only event journal (many rows per object,
138	* tombstones included) while the stored digest is exactly one row per
139	* live object — different cardinality and lifecycle. Folding the digest
140	* into the log would force a latest-row-per-object subquery on every
141	* scan, destroying the GROUP BY price this design exists for.
142	*
143	* digest is BIGINT UNSIGNED holding a 64-bit value (top 16 hex of MD5): integer
144	* storage keeps the BIT_XOR bucket aggregate a pure integer fold with
145	* constant-size state, where a CHAR hash would need GROUP_CONCAT (and
146	* its max_len truncation hazard) to aggregate.
147	*/
148	public function schema_sql( string $table_name, string $charset_collate = '' ): string {
149	return "CREATE TABLE {$table_name} (\n"
150	. " object_type VARCHAR(20) NOT NULL,\n"
151	. " object_id BIGINT UNSIGNED NOT NULL,\n"
152	. " digest BIGINT UNSIGNED NOT NULL,\n"
153	. " updated_gmt DATETIME NOT NULL,\n"
154	. " PRIMARY KEY (object_type, object_id),\n"
155	. " KEY object_id (object_id)\n"
156	. ") {$charset_collate};";
157	}
158
159	public function install(): void {
160	global $wpdb;
161	if ( ! function_exists( 'dbDelta' ) ) {
162	require_once ABSPATH . 'wp-admin/includes/upgrade.php';
163	}
164	dbDelta( $this->schema_sql( $this->table_name(), $wpdb->get_charset_collate() ) );
165	}
166
167
168	/**
169	* Same save/delete hooks the change-log listens to (products and
170	* variations only — tax rates live in their own table outside the
171	* wp_posts id space this scan buckets; they stay covered by the plain
172	* range-checksum candidate, whose checksum covers the full rate row).
173	*/
174	public function register_hooks(): void {
175	add_action( 'woocommerce_new_product', array( $this, 'record_post_saved' ), 10, 1 );
176	add_action( 'woocommerce_update_product', array( $this, 'record_post_saved' ), 10, 1 );
177	add_action( 'woocommerce_new_product_variation', array( $this, 'record_post_saved' ), 10, 1 );
178	add_action( 'woocommerce_update_product_variation', array( $this, 'record_post_saved' ), 10, 1 );
179	// Untrash does not reliably re-fire woocommerce_update_product; the
180	// upsert is a no-op for non-live rows, so hooking it is free.
181	add_action( 'untrashed_post', array( $this, 'record_post_untrashed' ), 10, 1 );
182	add_action( 'wp_trash_post', array( $this, 'record_post_deleted' ), 10, 1 );
183	add_action( 'before_delete_post', array( $this, 'record_post_deleted' ), 10, 1 );
184
185	// Leg-3 phase 7 (ADR 0015): ALL WordPress users are POS customers under
186	// #1379 (1.9 parity). Saves and role changes idempotently upsert their
187	// digest; only delete_user removes it.
188	add_action( 'user_register', array( $this, 'record_customer_saved' ), 10, 1 );
189	add_action( 'profile_update', array( $this, 'record_customer_saved' ), 10, 1 );
190	add_action( 'woocommerce_created_customer', array( $this, 'record_customer_saved' ), 10, 1 );
191	add_action( 'woocommerce_new_customer', array( $this, 'record_customer_saved' ), 10, 1 );
192	add_action( 'woocommerce_update_customer', array( $this, 'record_customer_saved' ), 10, 1 );
193	add_action( 'set_user_role', array( $this, 'record_customer_saved' ), 10, 1 );
194	// add_role()/remove_role() fire ONLY add_user_role/remove_user_role, so
195	// register both to capture membership changes in the served record.
196	add_action( 'add_user_role', array( $this, 'record_customer_saved' ), 10, 1 );
197	add_action( 'remove_user_role', array( $this, 'record_customer_saved' ), 10, 1 );
198	add_action( 'delete_user', array( $this, 'record_customer_deleted' ), 10, 1 );
199
200	// Leg-3 phase 7 (ADR 0015): order digest maintenance. Storage-agnostic WC order hooks (fire under
201	// HPOS AND CPT), matching the sync-index's order hooks. upsert/delete are idempotent (no dedup).
202	add_action( 'woocommerce_new_order', array( $this, 'record_order_saved' ), 10, 1 );
203	add_action( 'woocommerce_update_order', array( $this, 'record_order_saved' ), 10, 1 );
204	add_action( 'woocommerce_before_trash_order', array( $this, 'record_order_deleted' ), 10, 1 );
205	add_action( 'woocommerce_before_delete_order', array( $this, 'record_order_deleted' ), 10, 1 );
206	// Untrash recreation: `untrashed_post` (handled by record_post_untrashed)
207	// never fires for COT orders — without the HPOS twin hook a restored
208	// order's digest is never recreated and integrity scans treat it as
209	// deleted forever.
210	add_action( 'woocommerce_untrash_order', array( $this, 'record_order_untrashed' ), 10, 1 );
211	// Request boundary for the coalesced digest upserts (see
212	// $pending_digests). LAST on shutdown: WooCommerce saves the customer at
213	// 10 and the session at 20. Zero accepted args: do_action( 'shutdown' )
214	// passes an empty string otherwise.
215	add_action( 'shutdown', array( __CLASS__, 'flush_pending_digests_at_shutdown' ), PHP_INT_MAX, 0 );
216	}
217
218	/**
219	* Recreate a COT order's digest once its restore completes.
220	*
221	* `woocommerce_untrash_order` fires BEFORE the data store restores the
222	* status, and the restore's internal save fires no observer hook we bind
223	* (verified: `woocommerce_update_order` does not fire there) — so an
224	* immediate upsert would read a still-trashed row and write nothing. Arm a
225	* one-shot on the order's first object save after it leaves the trash and
226	* upsert then.
227	*
228	* @param int $order_id Order being restored.
229	*/
230	public function record_order_untrashed( int $order_id ): void {
231	$handler = function ( $order ) use ( $order_id, &$handler ): void {
232	if ( ! \is_object( $order ) \|\| ! method_exists( $order, 'get_id' ) \|\| ! method_exists( $order, 'get_status' ) \|\| (int) $order->get_id() !== $order_id \|\| 'trash' === $order->get_status() ) {
233	return;
234	}
235	remove_action( 'woocommerce_after_order_object_save', $handler );
236	$this->record_order_saved( $order_id );
237	};
238	add_action( 'woocommerce_after_order_object_save', $handler );
239	}
240
241	/**
242	* Cron entry point for rebuilding unexpectedly empty or stale product digests.
243	*/
244	public static function run_scheduled_rebuild(): void {
245	$lease = get_transient( self::REBUILD_LOCK );
246	try {
247	( new self() )->rebuild( true );
248	} catch ( \Throwable $exception ) {
249	Logger::error( 'WCPOS sync: scheduled integrity digest rebuild failed: ' . $exception->getMessage() );
250	} finally {
251	self::release_rebuild_lock( $lease );
252	}
253	}
254
255	/**
256	* Release the rebuild lease only if this run still owns it — a rebuild that
257	* outlived the lock TTL must not delete a successor's fresh lease.
258	*
259	* @param mixed $lease The lease value captured when this run started.
260	*/
261	public static function release_rebuild_lock( $lease ): void {
262	if ( get_transient( self::REBUILD_LOCK ) === $lease ) {
263	delete_transient( self::REBUILD_LOCK );
264	}
265	}
266
267	/**
268	* Wire THE digest stamper onto both served read lanes (#421 increment 3).
269	*
270	* ONE named static serves every digest id-space: it resolves the registry row
271	* from the lane's resource slug, so a collection that gains a digest group is
272	* stamped by adding a row and nothing else. The composed callback name this
273	* used to build (`stamp_proxy_{object_type}_digests`) could name a method that
274	* did not exist — add_filter() does not validate callables, so the miss only
275	* surfaced as a fatal at apply_filters() time, on a catalogue proxy read.
276	*
277	* Both public filter names stay live and both are registered here, so the order
278	* pull lane is wired by the same call as the proxy lane instead of by hand in
279	* Init. Returns the digest-and-proxy collections (the wiring golden pins them).
280	*
281	* @return string[] Collections whose served records carry a stored digest.
282	*/
283	public static function register_proxy_digest_stampers(): array {
284	$registered = array();
285	foreach ( Collections::with( 'digest' ) as $collection => $row ) {
286	if ( ! isset( $row['proxy'] ) ) {
287	continue;
288	}
289	$registered[] = $collection;
290	}
291	add_filter( 'woocommerce_pos_sync_proxy_response', array( __CLASS__, 'stamp_digests' ), 10, 3 );
292	add_filter( 'woocommerce_pos_sync_order_pull_payloads', array( __CLASS__, 'stamp_digests' ), 10, 3 );
293
294	return $registered;
295	}
296
297	/**
298	* Attach each served record's stored 64-bit digest as a top-level `_rxdb_digest`
299	* string, so the client seeds its existence-reconcile manifest (ADR 0014 Leg 3)
300	* as records flow through the NORMAL pull — no separate fetch. The client reads
301	* it into the sidecar manifest; it is NOT persisted into the document. A record
302	* with no stored digest yet simply carries no `_rxdb_digest`.
303	*
304	* The lane's resource slug picks the id-space (the registry owns the mapping,
305	* including the slug traps), and a resource with no digest group — or none at
306	* all — returns the payload untouched.
307	*
308	* @param mixed $data Served list of records.
309	* @param mixed $resource Lane resource slug.
310	* @param mixed $request Request context.
311	*
312	* @return mixed
313	*/
314	public static function stamp_digests( $data, $resource = '', $request = null ) {
315	if ( ! \is_array( $data ) \|\| ! \is_string( $resource ) \|\| '' === $resource ) {
316	return $data;
317	}
318	$row = Collections::by_proxy_slug( $resource );
319	if ( null === $row \|\| ! isset( $row['digest'] ) ) {
320	return $data;
321	}
322	$ids = array();
323	foreach ( $data as $record ) {
324	if ( \is_array( $record ) && isset( $record['id'] ) ) {
325	$ids[] = (int) $record['id'];
326	}
327	}
328	if ( array() === $ids ) {
329	return $data;
330	}
331	$digests = ( new Digest_Index() )->read_digests( $row['_collection'], $ids );
332	foreach ( $data as $index => $record ) {
333	if ( \is_array( $record ) && isset( $record['id'] ) && isset( $digests[ (int) $record['id'] ] ) ) {
334	$data[ $index ]['_rxdb_digest'] = $digests[ (int) $record['id'] ];
335	}
336	}
337
338	return $data;
339	}
340
341	/**
342	* Canonical per-CUSTOMER digest SELECT (ADR 0015, Leg-3 phase 7).
343	*
344	* @deprecated Use {@see Digest_Index::customer_digest_select_sql()}.
345	*/
346	public function customer_digest_select_sql( string $where_sql = '' ): string {
347	return $this->index->customer_digest_select_sql( $where_sql );
348	}
349
350	/**
351	* Customer digest maintenance (ADR 0015, Leg-3 phase 7) — every WordPress
352	* user is a POS customer, so saves and role changes always upsert.
353	*/
354	/** Owe the customer's digest; it is written once, on flush (see $pending_digests). */
355	public function record_customer_saved( int $user_id ): void {
356	$this->defer( 'customer', $user_id );
357	}
358
359	/**
360	* Queue one digest upsert, or write it now if the boundary has passed.
361	*
362	* @param string $type 'order' or 'customer'.
363	* @param int $id Record id.
364	*/
365	private function defer( string $type, int $id ): void {
366	if ( self::$shutdown_flushed ) {
367	// A save triggered by another shutdown handler (WooCommerce saves the
368	// customer at priority 10): nothing will flush again, so write now.
369	$this->upsert_pending( $type, $id );
370	return;
371	}
372	if ( null === self::$flusher ) {
373	self::$flusher = $this;
374	}
375	$blog = get_current_blog_id();
376	self::$pending_digests[ self::pending_key( $type, $id ) ] = array( $blog, $type, $id );
377	if ( \count( self::$pending_digests ) >= self::PENDING_DIGEST_FLUSH_THRESHOLD ) {
378	self::flush_pending_digests();
379	}
380	}
381
382	private static function pending_key( string $type, int $id ): string {
383	return get_current_blog_id() . ':' . $type . ':' . $id;
384	}
385
386	/** One queued upsert, under the observer's fail-open posture. */
387	private function upsert_pending( string $type, int $id ): void {
388	$this->observe(
389	function () use ( $type, $id ): void {
390	if ( 'customer' === $type ) {
391	$this->upsert_customer_digest( $id );
392	} elseif ( 'order' === $type ) {
393	$this->upsert_order_digest( $id );
394	} else {
395	// 'post' (product or variation): the SQL derives the stored type from the row.
396	$this->upsert_digest( $id );
397	}
398	}
399	);
400	}
401
402	/**
403	* Write every owed digest.
404	*
405	* Called from the shutdown flush, from {@see Digest_Index::read_digests()}
406	* before it reads, and when the queue reaches its threshold. Writes go
407	* through the instance that first queued (so an injected Digest_Index is
408	* honoured) and under the blog each entry was recorded on. Each upsert keeps
409	* the observer's fail-open posture: a failure is logged and the scan
410	* self-heals. Safe to call repeatedly — a flushed digest is no longer pending.
411	*/
412	public static function flush_pending_digests(): void {
413	if ( array() === self::$pending_digests ) {
414	return;
415	}
416	$pending = self::$pending_digests;
417	self::$pending_digests = array();
418	$digest = self::$flusher ?? new self();
419	foreach ( $pending as $entry ) {
420	list( $blog, $type, $id ) = $entry;
421	$switch = is_multisite() && get_current_blog_id() !== (int) $blog;
422	if ( $switch ) {
423	switch_to_blog( (int) $blog );
424	}
425	try {
426	$digest->upsert_pending( (string) $type, (int) $id );
427	} finally {
428	if ( $switch ) {
429	restore_current_blog();
430	}
431	}
432	}
433	}
434
435	/**
436	* The `shutdown` callback: flush, then write every later save immediately.
437	*/
438	public static function flush_pending_digests_at_shutdown(): void {
439	self::$shutdown_flushed = true;
440	self::flush_pending_digests();
441	}
442
443	/**
444	* Discard per-request coalescing state. Tests only: the PHPUnit process
445	* never reaches `shutdown`, so the static queue, flusher and flag would
446	* leak between test cases otherwise.
447	*
448	* @internal
449	*/
450	public static function reset_request_state(): void {
451	self::$pending_digests = array();
452	self::$flusher = null;
453	self::$shutdown_flushed = false;
454	}
455
456	public function record_customer_deleted( int $user_id ): void {
457	// A pending upsert for a record that is leaving must not be written after the fact.
458	unset( self::$pending_digests[ self::pending_key( 'customer', $user_id ) ] );
459	$this->observe(
460	function () use ( $user_id ): void {
461	$this->delete_customer_digest( $user_id );
462	}
463	);
464	}
465
466	/**
467	* Observation hooks must never break the host write that fired them: a
468	* broken or missing digest store is a sync problem (the integrity scan and
469	* the health gate surface it), not a reason to fatal a WooCommerce save.
470	* The ops paths (rebuild/prune) keep throwing — they run on demand and
471	* want the loudness.
472	*
473	* @param callable $observer The digest write to attempt.
474	*/
475	private function observe( callable $observer ): void {
476	try {
477	$observer();
478	} catch ( \Throwable $e ) {
479	Logger::error( 'Sync digest observer failed (sync will self-heal via scan/rebuild): ' . $e->getMessage() );
480	}
481	}
482
483	private function delete_customer_digest( int $user_id ): void {
484	global $wpdb;
485	$deleted = $wpdb->delete(
486	$this->table_name(),
487	array(
488	'object_type' => 'customer',
489	'object_id' => $user_id,
490	),
491	array( '%s', '%d' )
492	);
493	if ( false === $deleted ) {
494	throw new RuntimeException( 'delete stored customer digest failed: ' . $wpdb->last_error );
495	}
496	}
497
498	/**
499	* Order digest maintenance (ADR 0015, Leg-3 phase 7). The WC order hooks are storage-agnostic (fire
500	* under HPOS AND CPT); the digest SQL's `type='shop_order'` filter makes the upsert a no-op for any
501	* non-order, so no type re-check is needed here.
502	*/
503	/** Owe the order's digest; it is written once, on flush (see $pending_digests). */
504	public function record_order_saved( int $order_id ): void {
505	$this->defer( 'order', $order_id );
506	}
507
508	public function record_order_deleted( int $order_id ): void {
509	// A pending upsert for a record that is leaving must not be written after the fact.
510	unset( self::$pending_digests[ self::pending_key( 'order', $order_id ) ] );
511	$this->observe(
512	function () use ( $order_id ): void {
513	$this->delete_order_digest( $order_id );
514	}
515	);
516	}
517
518	private function delete_order_digest( int $order_id ): void {
519	global $wpdb;
520	$deleted = $wpdb->delete(
521	$this->table_name(),
522	array(
523	'object_type' => 'order',
524	'object_id' => $order_id,
525	),
526	array( '%s', '%d' )
527	);
528	if ( false === $deleted ) {
529	throw new RuntimeException( 'delete stored order digest failed: ' . $wpdb->last_error );
530	}
531	}
532
533	/** Order analogue of {@see upsert_customer_digest}: compute + store one order's digest (HPOS or CPT). */
534	public function upsert_order_digest( int $order_id ): void {
535	global $wpdb;
536	$started = microtime( true );
537	$this->index->raise_group_concat_max_len();
538	$result = $wpdb->query(
539	$wpdb->prepare(
540	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
541	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
542	. ' FROM (' . $this->index->order_digest_select_sql( '{id} = %d' ) . ') t'
543	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
544	$order_id
545	)
546	);
547	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
548	if ( false === $result ) {
549	throw new RuntimeException( 'upsert stored order digest failed: ' . $wpdb->last_error );
550	}
551	}
552
553	/**
554	* Owe the product's or variation's digest; it is written once, on flush (see
555	* $pending_digests). The queue type is 'post' for both: the upsert's SQL
556	* derives the stored object_type from the row, so nothing here needs to.
557	*/
558	public function record_post_saved( int $post_id ): void {
559	$this->defer( 'post', $post_id );
560	}
561
562	public function record_post_untrashed( int $post_id ): void {
563	$post_type = get_post_type( $post_id );
564	if ( 'shop_order' === $post_type ) {
565	$this->record_order_saved( $post_id );
566	return;
567	}
568	if ( in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
569	$this->record_post_saved( $post_id );
570	}
571	}
572
573	public function record_post_deleted( int $post_id ): void {
574	$post_type = get_post_type( $post_id );
575	if ( ! in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
576	return;
577	}
578	// A pending upsert for a record that is leaving must not be written after the fact.
579	unset( self::$pending_digests[ self::pending_key( 'post', $post_id ) ] );
580	$this->observe(
581	function () use ( $post_id, $post_type ): void {
582	$this->delete_post_digest( $post_id, $post_type );
583	}
584	);
585	}
586
587	/**
588	* Remove a product/variation digest row after a hooked delete.
589	*
590	* A hooked delete removes the stored row so stored == current again.
591	* Only a hook-BYPASSING delete leaves an orphan digest behind, which
592	* the scan reports as a mismatch (stored side carries a row the
593	* current side lacks) and the drill-down labels status=deleted.
594	*
595	* @param int $post_id The deleted post id.
596	* @param string $post_type Its post type (product \| product_variation).
597	*/
598	private function delete_post_digest( int $post_id, string $post_type ): void {
599	global $wpdb;
600	$started = microtime( true );
601	$deleted = $wpdb->delete(
602	$this->table_name(),
603	array(
604	'object_type' => 'product_variation' === $post_type ? 'variation' : 'product',
605	'object_id' => $post_id,
606	),
607	array( '%s', '%d' )
608	);
609	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
610	if ( false === $deleted ) {
611	throw new RuntimeException( 'delete stored digest failed: ' . $wpdb->last_error );
612	}
613	}
614
615	/**
616	* One statement: the digest is computed in SQL from the raw row and
617	* upserted in the same statement — PHP never materializes the value.
618	* No-op for rows outside the live predicate (the delete hook owns those).
619	*/
620	public function upsert_digest( int $post_id ): void {
621	global $wpdb;
622	// Time from BEFORE the session setup so timing.digest_ms covers ALL digest hook work
623	// (the raise runs inside the save hook — codex P3).
624	$started = microtime( true );
625	$this->index->raise_group_concat_max_len();
626	$this->query_with_retry(
627	$wpdb->prepare(
628	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
629	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
630	. ' FROM (' . $this->index->row_digest_select_sql( 'p.ID = %d' ) . ') t'
631	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
632	$post_id
633	),
634	'upsert stored digest failed: ',
635	$started
636	);
637	}
638
639	/**
640	* Customer analogue of {@see upsert_digest} (ADR 0015, Leg-3 phase 7):
641	* compute and store one WordPress user's customer digest in a single
642	* INSERT…SELECT. Only the delete hook removes it.
643	*/
644	public function upsert_customer_digest( int $user_id ): void {
645	global $wpdb;
646	$started = microtime( true );
647	$this->index->raise_group_concat_max_len();
648	$this->query_with_retry(
649	$wpdb->prepare(
650	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
651	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
652	. ' FROM (' . $this->index->customer_digest_select_sql( 'u.ID = %d' ) . ') t'
653	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
654	$user_id
655	),
656	'upsert stored customer digest failed: ',
657	$started
658	);
659	}
660
661	/**
662	* MySQL/MariaDB error numbers a second attempt can clear: 1020 ER_CHECKREAD
663	* ("Record has changed since last read"), 1205 ER_LOCK_WAIT_TIMEOUT, 1213
664	* ER_LOCK_DEADLOCK. Two requests upserting the same digest row race on
665	* the `INSERT … ON DUPLICATE KEY UPDATE`; the retry reads the updated row.
666	*/
667	private const TRANSIENT_CONTENTION_ERRNOS = array( 1020, 1205, 1213 );
668
669	/**
670	* Message fallback for the same three errors, used only when the driver's
671	* error number is unavailable (a wpdb without a live mysqli handle).
672	*/
673	private const TRANSIENT_CONTENTION_MESSAGES = array(
674	'Record has changed since last read',
675	'Lock wait timeout',
676	'Deadlock found',
677	);
678
679	/** Retry a contended upsert once, including both attempts in the hook timing. */
680	private function query_with_retry( string $sql, string $error_message, float $started ): void {
681	global $wpdb;
682	$result = $wpdb->query( $sql );
683	if ( false === $result && $this->is_transient_contention( $wpdb ) ) {
684	$result = $wpdb->query( $sql );
685	}
686	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
687	if ( false === $result ) {
688	throw new RuntimeException( $error_message . $wpdb->last_error );
689	}
690	}
691
692	/**
693	* The error number is authoritative: server messages are localised
694	* (`lc_messages`), so the English text is only a fallback for a handle-less
695	* wpdb. `$wpdb->dbh` is reachable through wpdb's magic getter.
696	*/
697	private function is_transient_contention( \wpdb $wpdb ): bool {
698	$dbh = $wpdb->__get( 'dbh' );
699	if ( $dbh instanceof \mysqli ) {
700	$errno = mysqli_errno( $dbh ); // phpcs:ignore WordPress.DB.RestrictedFunctions -- reads the driver's last error number; no query is issued.
701	if ( 0 !== $errno ) {
702	return in_array( $errno, self::TRANSIENT_CONTENTION_ERRNOS, true );
703	}
704	}
705	foreach ( self::TRANSIENT_CONTENTION_MESSAGES as $message ) {
706	if ( false !== strpos( $wpdb->last_error, $message ) ) {
707	return true;
708	}
709	}
710	return false;
711	}
712
713	/**
714	* Backfill/repair: prune orphans, then digest every live row in one
715	* INSERT…SELECT pass. Pre-existing catalogs (the 10k seed) become fully
716	* digestable in one call; measured timing is returned so the lab can
717	* report the backfill price.
718	*
719	* @param bool $products_only Whether to stop after rebuilding product digests.
720	*/
721	public function rebuild( bool $products_only = false ): array {
722	global $wpdb;
723	$this->index->raise_group_concat_max_len();
724	$started = microtime( true );
725
726	$orphans_deleted = $wpdb->query(
727	'DELETE FROM ' . $this->table_name()
728	. ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
729	. ' AND NOT ' . $this->index->live_row_exists_sql( 'object_id' )
730	);
731	if ( false === $orphans_deleted ) {
732	throw new RuntimeException( 'prune orphan stored digests failed: ' . $wpdb->last_error );
733	}
734
735	// Affected-rows semantics of ON DUPLICATE KEY: 1 per insert, 2 per
736	// update, 0 per already-matching row — reported raw as "writes".
737	// updated_gmt is assigned FIRST and only when the digest actually
738	// changed (assignments evaluate left-to-right, so the IF must read
739	// the pre-update digest before the digest assignment overwrites it).
740	// Otherwise a repeated rebuild rewrites UTC_TIMESTAMP() into every
741	// row, counts the whole table as writes, and inflates the
742	// hash-checksum baseline cost (codex review).
743	$writes = $wpdb->query(
744	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
745	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
746	. ' FROM (' . $this->index->row_digest_select_sql() . ') t'
747	. ' ON DUPLICATE KEY UPDATE'
748	. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
749	. ' digest = VALUES(digest)'
750	);
751	if ( false === $writes ) {
752	throw new RuntimeException( 'rebuild stored digests failed: ' . $wpdb->last_error );
753	}
754	update_option( Digest_Index::FORMULA_FP_OPTION, Digest_Index::digest_formula_fingerprint(), false );
755
756	if ( $products_only ) {
757	$stored_total = (int) $wpdb->get_var(
758	'SELECT COUNT(*) FROM ' . $this->table_name() . ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
759	);
760
761	return array(
762	'writes' => (int) $writes,
763	'orphans_deleted' => (int) $orphans_deleted,
764	'stored_total' => $stored_total,
765	'duration_ms' => round( ( microtime( true ) - $started ) * 1000, 3 ),
766	);
767	}
768
769	// Leg-3 phase 7 (ADR 0015): customers share the digest table via their own 'customer' rows —
770	// the same prune-orphans + INSERT…SELECT pass, over the customer predicate + id-space. A stored
771	// customer whose user vanished or lost the customer role is an orphan (a role removal never fires
772	// before_delete_post, so the rebuild is the backstop that reconciles it).
773	$customer_orphans = $wpdb->query(
774	'DELETE FROM ' . $this->table_name()
775	. " WHERE object_type = 'customer'"
776	. ' AND NOT ' . $this->index->customer_live_row_exists_sql( 'object_id' )
777	);
778	if ( false === $customer_orphans ) {
779	throw new RuntimeException( 'prune orphan customer digests failed: ' . $wpdb->last_error );
780	}
781	$customer_writes = $wpdb->query(
782	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
783	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
784	. ' FROM (' . $this->index->customer_digest_select_sql() . ') t'
785	. ' ON DUPLICATE KEY UPDATE'
786	. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
787	. ' digest = VALUES(digest)'
788	);
789	if ( false === $customer_writes ) {
790	throw new RuntimeException( 'rebuild customer digests failed: ' . $wpdb->last_error );
791	}
792
793	// Leg-3 phase 7 (ADR 0015): orders share the digest table via their own 'order' rows (HPOS or CPT).
794	// Same prune-orphans + INSERT…SELECT pass; Digest_Index::order_digest_select_sql() emits the storage-correct SQL
795	// (the CPT path GROUP BYs, the HPOS path does not — both valid as an INSERT…SELECT source).
796	$order_orphans = $wpdb->query(
797	'DELETE FROM ' . $this->table_name()
798	. " WHERE object_type = 'order'"
799	. ' AND NOT ' . $this->index->order_live_row_exists_sql( 'object_id' )
800	);
801	if ( false === $order_orphans ) {
802	throw new RuntimeException( 'prune orphan order digests failed: ' . $wpdb->last_error );
803	}
804	$order_writes = $wpdb->query(
805	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
806	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
807	. ' FROM (' . $this->index->order_digest_select_sql() . ') t'
808	. ' ON DUPLICATE KEY UPDATE'
809	. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
810	. ' digest = VALUES(digest)'
811	);
812	if ( false === $order_writes ) {
813	throw new RuntimeException( 'rebuild order digests failed: ' . $wpdb->last_error );
814	}
815
816	$stored_total = (int) $wpdb->get_var(
817	'SELECT COUNT(*) FROM ' . $this->table_name() . ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
818	);
819	$customer_stored_total = (int) $wpdb->get_var(
820	'SELECT COUNT(*) FROM ' . $this->table_name() . " WHERE object_type = 'customer'"
821	);
822	$order_stored_total = (int) $wpdb->get_var(
823	'SELECT COUNT(*) FROM ' . $this->table_name() . " WHERE object_type = 'order'"
824	);
825
826	return array(
827	'writes' => (int) $writes + (int) $customer_writes + (int) $order_writes,
828	'orphans_deleted' => (int) $orphans_deleted + (int) $customer_orphans + (int) $order_orphans,
829	'stored_total' => $stored_total + $customer_stored_total + $order_stored_total,
830	'duration_ms' => round( ( microtime( true ) - $started ) * 1000, 3 ),
831	);
832	}
833	}
834