woocommerce-pos / includes / Sync / Integrity_Digest.php

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

849 lines 34.9 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	* Detach THE digest stamper from both served read lanes.
299	*
300	* The teardown twin of {@see register_proxy_digest_stampers()}, matching the
301	* `unregister_*` seams {@see Revision} and {@see Proxy_Uuid_Stamper} already
302	* expose. `Augmentation_Pipeline::reset()` only removes the projections the
303	* pipeline itself installed, so without this a caller that installs the real
304	* pipeline — a test wiring the production read lane — cannot unwind it and
305	* leaks this filter into everything that runs after it.
306	*/
307	public static function unregister_proxy_digest_stampers(): void {
308	remove_filter( 'woocommerce_pos_sync_proxy_response', array( __CLASS__, 'stamp_digests' ), 10 );
309	remove_filter( 'woocommerce_pos_sync_order_pull_payloads', array( __CLASS__, 'stamp_digests' ), 10 );
310	}
311
312	/**
313	* Attach each served record's stored 64-bit digest as a top-level `_rxdb_digest`
314	* string, so the client seeds its existence-reconcile manifest (ADR 0014 Leg 3)
315	* as records flow through the NORMAL pull — no separate fetch. The client reads
316	* it into the sidecar manifest; it is NOT persisted into the document. A record
317	* with no stored digest yet simply carries no `_rxdb_digest`.
318	*
319	* The lane's resource slug picks the id-space (the registry owns the mapping,
320	* including the slug traps), and a resource with no digest group — or none at
321	* all — returns the payload untouched.
322	*
323	* @param mixed $data Served list of records.
324	* @param mixed $resource Lane resource slug.
325	* @param mixed $request Request context.
326	*
327	* @return mixed
328	*/
329	public static function stamp_digests( $data, $resource = '', $request = null ) {
330	if ( ! \is_array( $data ) \|\| ! \is_string( $resource ) \|\| '' === $resource ) {
331	return $data;
332	}
333	$row = Collections::by_proxy_slug( $resource );
334	if ( null === $row \|\| ! isset( $row['digest'] ) ) {
335	return $data;
336	}
337	$ids = array();
338	foreach ( $data as $record ) {
339	if ( \is_array( $record ) && isset( $record['id'] ) ) {
340	$ids[] = (int) $record['id'];
341	}
342	}
343	if ( array() === $ids ) {
344	return $data;
345	}
346	$digests = ( new Digest_Index() )->read_digests( $row['_collection'], $ids );
347	foreach ( $data as $index => $record ) {
348	if ( \is_array( $record ) && isset( $record['id'] ) && isset( $digests[ (int) $record['id'] ] ) ) {
349	$data[ $index ]['_rxdb_digest'] = $digests[ (int) $record['id'] ];
350	}
351	}
352
353	return $data;
354	}
355
356	/**
357	* Canonical per-CUSTOMER digest SELECT (ADR 0015, Leg-3 phase 7).
358	*
359	* @deprecated Use {@see Digest_Index::customer_digest_select_sql()}.
360	*/
361	public function customer_digest_select_sql( string $where_sql = '' ): string {
362	return $this->index->customer_digest_select_sql( $where_sql );
363	}
364
365	/**
366	* Customer digest maintenance (ADR 0015, Leg-3 phase 7) — every WordPress
367	* user is a POS customer, so saves and role changes always upsert.
368	*/
369	/** Owe the customer's digest; it is written once, on flush (see $pending_digests). */
370	public function record_customer_saved( int $user_id ): void {
371	$this->defer( 'customer', $user_id );
372	}
373
374	/**
375	* Queue one digest upsert, or write it now if the boundary has passed.
376	*
377	* @param string $type 'order' or 'customer'.
378	* @param int $id Record id.
379	*/
380	private function defer( string $type, int $id ): void {
381	if ( self::$shutdown_flushed ) {
382	// A save triggered by another shutdown handler (WooCommerce saves the
383	// customer at priority 10): nothing will flush again, so write now.
384	$this->upsert_pending( $type, $id );
385	return;
386	}
387	if ( null === self::$flusher ) {
388	self::$flusher = $this;
389	}
390	$blog = get_current_blog_id();
391	self::$pending_digests[ self::pending_key( $type, $id ) ] = array( $blog, $type, $id );
392	if ( \count( self::$pending_digests ) >= self::PENDING_DIGEST_FLUSH_THRESHOLD ) {
393	self::flush_pending_digests();
394	}
395	}
396
397	private static function pending_key( string $type, int $id ): string {
398	return get_current_blog_id() . ':' . $type . ':' . $id;
399	}
400
401	/** One queued upsert, under the observer's fail-open posture. */
402	private function upsert_pending( string $type, int $id ): void {
403	$this->observe(
404	function () use ( $type, $id ): void {
405	if ( 'customer' === $type ) {
406	$this->upsert_customer_digest( $id );
407	} elseif ( 'order' === $type ) {
408	$this->upsert_order_digest( $id );
409	} else {
410	// 'post' (product or variation): the SQL derives the stored type from the row.
411	$this->upsert_digest( $id );
412	}
413	}
414	);
415	}
416
417	/**
418	* Write every owed digest.
419	*
420	* Called from the shutdown flush, from {@see Digest_Index::read_digests()}
421	* before it reads, and when the queue reaches its threshold. Writes go
422	* through the instance that first queued (so an injected Digest_Index is
423	* honoured) and under the blog each entry was recorded on. Each upsert keeps
424	* the observer's fail-open posture: a failure is logged and the scan
425	* self-heals. Safe to call repeatedly — a flushed digest is no longer pending.
426	*/
427	public static function flush_pending_digests(): void {
428	if ( array() === self::$pending_digests ) {
429	return;
430	}
431	$pending = self::$pending_digests;
432	self::$pending_digests = array();
433	$digest = self::$flusher ?? new self();
434	foreach ( $pending as $entry ) {
435	list( $blog, $type, $id ) = $entry;
436	$switch = is_multisite() && get_current_blog_id() !== (int) $blog;
437	if ( $switch ) {
438	switch_to_blog( (int) $blog );
439	}
440	try {
441	$digest->upsert_pending( (string) $type, (int) $id );
442	} finally {
443	if ( $switch ) {
444	restore_current_blog();
445	}
446	}
447	}
448	}
449
450	/**
451	* The `shutdown` callback: flush, then write every later save immediately.
452	*/
453	public static function flush_pending_digests_at_shutdown(): void {
454	self::$shutdown_flushed = true;
455	self::flush_pending_digests();
456	}
457
458	/**
459	* Discard per-request coalescing state. Tests only: the PHPUnit process
460	* never reaches `shutdown`, so the static queue, flusher and flag would
461	* leak between test cases otherwise.
462	*
463	* @internal
464	*/
465	public static function reset_request_state(): void {
466	self::$pending_digests = array();
467	self::$flusher = null;
468	self::$shutdown_flushed = false;
469	}
470
471	public function record_customer_deleted( int $user_id ): void {
472	// A pending upsert for a record that is leaving must not be written after the fact.
473	unset( self::$pending_digests[ self::pending_key( 'customer', $user_id ) ] );
474	$this->observe(
475	function () use ( $user_id ): void {
476	$this->delete_customer_digest( $user_id );
477	}
478	);
479	}
480
481	/**
482	* Observation hooks must never break the host write that fired them: a
483	* broken or missing digest store is a sync problem (the integrity scan and
484	* the health gate surface it), not a reason to fatal a WooCommerce save.
485	* The ops paths (rebuild/prune) keep throwing — they run on demand and
486	* want the loudness.
487	*
488	* @param callable $observer The digest write to attempt.
489	*/
490	private function observe( callable $observer ): void {
491	try {
492	$observer();
493	} catch ( \Throwable $e ) {
494	Logger::error( 'Sync digest observer failed (sync will self-heal via scan/rebuild): ' . $e->getMessage() );
495	}
496	}
497
498	private function delete_customer_digest( int $user_id ): void {
499	global $wpdb;
500	$deleted = $wpdb->delete(
501	$this->table_name(),
502	array(
503	'object_type' => 'customer',
504	'object_id' => $user_id,
505	),
506	array( '%s', '%d' )
507	);
508	if ( false === $deleted ) {
509	throw new RuntimeException( 'delete stored customer digest failed: ' . $wpdb->last_error );
510	}
511	}
512
513	/**
514	* Order digest maintenance (ADR 0015, Leg-3 phase 7). The WC order hooks are storage-agnostic (fire
515	* under HPOS AND CPT); the digest SQL's `type='shop_order'` filter makes the upsert a no-op for any
516	* non-order, so no type re-check is needed here.
517	*/
518	/** Owe the order's digest; it is written once, on flush (see $pending_digests). */
519	public function record_order_saved( int $order_id ): void {
520	$this->defer( 'order', $order_id );
521	}
522
523	public function record_order_deleted( int $order_id ): void {
524	// A pending upsert for a record that is leaving must not be written after the fact.
525	unset( self::$pending_digests[ self::pending_key( 'order', $order_id ) ] );
526	$this->observe(
527	function () use ( $order_id ): void {
528	$this->delete_order_digest( $order_id );
529	}
530	);
531	}
532
533	private function delete_order_digest( int $order_id ): void {
534	global $wpdb;
535	$deleted = $wpdb->delete(
536	$this->table_name(),
537	array(
538	'object_type' => 'order',
539	'object_id' => $order_id,
540	),
541	array( '%s', '%d' )
542	);
543	if ( false === $deleted ) {
544	throw new RuntimeException( 'delete stored order digest failed: ' . $wpdb->last_error );
545	}
546	}
547
548	/** Order analogue of {@see upsert_customer_digest}: compute + store one order's digest (HPOS or CPT). */
549	public function upsert_order_digest( int $order_id ): void {
550	global $wpdb;
551	$started = microtime( true );
552	$this->index->raise_group_concat_max_len();
553	$result = $wpdb->query(
554	$wpdb->prepare(
555	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
556	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
557	. ' FROM (' . $this->index->order_digest_select_sql( '{id} = %d' ) . ') t'
558	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
559	$order_id
560	)
561	);
562	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
563	if ( false === $result ) {
564	throw new RuntimeException( 'upsert stored order digest failed: ' . $wpdb->last_error );
565	}
566	}
567
568	/**
569	* Owe the product's or variation's digest; it is written once, on flush (see
570	* $pending_digests). The queue type is 'post' for both: the upsert's SQL
571	* derives the stored object_type from the row, so nothing here needs to.
572	*/
573	public function record_post_saved( int $post_id ): void {
574	$this->defer( 'post', $post_id );
575	}
576
577	public function record_post_untrashed( int $post_id ): void {
578	$post_type = get_post_type( $post_id );
579	if ( 'shop_order' === $post_type ) {
580	$this->record_order_saved( $post_id );
581	return;
582	}
583	if ( in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
584	$this->record_post_saved( $post_id );
585	}
586	}
587
588	public function record_post_deleted( int $post_id ): void {
589	$post_type = get_post_type( $post_id );
590	if ( ! in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
591	return;
592	}
593	// A pending upsert for a record that is leaving must not be written after the fact.
594	unset( self::$pending_digests[ self::pending_key( 'post', $post_id ) ] );
595	$this->observe(
596	function () use ( $post_id, $post_type ): void {
597	$this->delete_post_digest( $post_id, $post_type );
598	}
599	);
600	}
601
602	/**
603	* Remove a product/variation digest row after a hooked delete.
604	*
605	* A hooked delete removes the stored row so stored == current again.
606	* Only a hook-BYPASSING delete leaves an orphan digest behind, which
607	* the scan reports as a mismatch (stored side carries a row the
608	* current side lacks) and the drill-down labels status=deleted.
609	*
610	* @param int $post_id The deleted post id.
611	* @param string $post_type Its post type (product \| product_variation).
612	*/
613	private function delete_post_digest( int $post_id, string $post_type ): void {
614	global $wpdb;
615	$started = microtime( true );
616	$deleted = $wpdb->delete(
617	$this->table_name(),
618	array(
619	'object_type' => 'product_variation' === $post_type ? 'variation' : 'product',
620	'object_id' => $post_id,
621	),
622	array( '%s', '%d' )
623	);
624	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
625	if ( false === $deleted ) {
626	throw new RuntimeException( 'delete stored digest failed: ' . $wpdb->last_error );
627	}
628	}
629
630	/**
631	* One statement: the digest is computed in SQL from the raw row and
632	* upserted in the same statement — PHP never materializes the value.
633	* No-op for rows outside the live predicate (the delete hook owns those).
634	*/
635	public function upsert_digest( int $post_id ): void {
636	global $wpdb;
637	// Time from BEFORE the session setup so timing.digest_ms covers ALL digest hook work
638	// (the raise runs inside the save hook — codex P3).
639	$started = microtime( true );
640	$this->index->raise_group_concat_max_len();
641	$this->query_with_retry(
642	$wpdb->prepare(
643	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
644	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
645	. ' FROM (' . $this->index->row_digest_select_sql( 'p.ID = %d' ) . ') t'
646	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
647	$post_id
648	),
649	'upsert stored digest failed: ',
650	$started
651	);
652	}
653
654	/**
655	* Customer analogue of {@see upsert_digest} (ADR 0015, Leg-3 phase 7):
656	* compute and store one WordPress user's customer digest in a single
657	* INSERT…SELECT. Only the delete hook removes it.
658	*/
659	public function upsert_customer_digest( int $user_id ): void {
660	global $wpdb;
661	$started = microtime( true );
662	$this->index->raise_group_concat_max_len();
663	$this->query_with_retry(
664	$wpdb->prepare(
665	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
666	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
667	. ' FROM (' . $this->index->customer_digest_select_sql( 'u.ID = %d' ) . ') t'
668	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
669	$user_id
670	),
671	'upsert stored customer digest failed: ',
672	$started
673	);
674	}
675
676	/**
677	* MySQL/MariaDB error numbers a second attempt can clear: 1020 ER_CHECKREAD
678	* ("Record has changed since last read"), 1205 ER_LOCK_WAIT_TIMEOUT, 1213
679	* ER_LOCK_DEADLOCK. Two requests upserting the same digest row race on
680	* the `INSERT … ON DUPLICATE KEY UPDATE`; the retry reads the updated row.
681	*/
682	private const TRANSIENT_CONTENTION_ERRNOS = array( 1020, 1205, 1213 );
683
684	/**
685	* Message fallback for the same three errors, used only when the driver's
686	* error number is unavailable (a wpdb without a live mysqli handle).
687	*/
688	private const TRANSIENT_CONTENTION_MESSAGES = array(
689	'Record has changed since last read',
690	'Lock wait timeout',
691	'Deadlock found',
692	);
693
694	/** Retry a contended upsert once, including both attempts in the hook timing. */
695	private function query_with_retry( string $sql, string $error_message, float $started ): void {
696	global $wpdb;
697	$result = $wpdb->query( $sql );
698	if ( false === $result && $this->is_transient_contention( $wpdb ) ) {
699	$result = $wpdb->query( $sql );
700	}
701	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
702	if ( false === $result ) {
703	throw new RuntimeException( $error_message . $wpdb->last_error );
704	}
705	}
706
707	/**
708	* The error number is authoritative: server messages are localised
709	* (`lc_messages`), so the English text is only a fallback for a handle-less
710	* wpdb. `$wpdb->dbh` is reachable through wpdb's magic getter.
711	*/
712	private function is_transient_contention( \wpdb $wpdb ): bool {
713	$dbh = $wpdb->__get( 'dbh' );
714	if ( $dbh instanceof \mysqli ) {
715	$errno = mysqli_errno( $dbh ); // phpcs:ignore WordPress.DB.RestrictedFunctions -- reads the driver's last error number; no query is issued.
716	if ( 0 !== $errno ) {
717	return in_array( $errno, self::TRANSIENT_CONTENTION_ERRNOS, true );
718	}
719	}
720	foreach ( self::TRANSIENT_CONTENTION_MESSAGES as $message ) {
721	if ( false !== strpos( $wpdb->last_error, $message ) ) {
722	return true;
723	}
724	}
725	return false;
726	}
727
728	/**
729	* Backfill/repair: prune orphans, then digest every live row in one
730	* INSERT…SELECT pass. Pre-existing catalogs (the 10k seed) become fully
731	* digestable in one call; measured timing is returned so the lab can
732	* report the backfill price.
733	*
734	* @param bool $products_only Whether to stop after rebuilding product digests.
735	*/
736	public function rebuild( bool $products_only = false ): array {
737	global $wpdb;
738	$this->index->raise_group_concat_max_len();
739	$started = microtime( true );
740
741	$orphans_deleted = $wpdb->query(
742	'DELETE FROM ' . $this->table_name()
743	. ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
744	. ' AND NOT ' . $this->index->live_row_exists_sql( 'object_id' )
745	);
746	if ( false === $orphans_deleted ) {
747	throw new RuntimeException( 'prune orphan stored digests failed: ' . $wpdb->last_error );
748	}
749
750	// Affected-rows semantics of ON DUPLICATE KEY: 1 per insert, 2 per
751	// update, 0 per already-matching row — reported raw as "writes".
752	// updated_gmt is assigned FIRST and only when the digest actually
753	// changed (assignments evaluate left-to-right, so the IF must read
754	// the pre-update digest before the digest assignment overwrites it).
755	// Otherwise a repeated rebuild rewrites UTC_TIMESTAMP() into every
756	// row, counts the whole table as writes, and inflates the
757	// hash-checksum baseline cost (codex review).
758	$writes = $wpdb->query(
759	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
760	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
761	. ' FROM (' . $this->index->row_digest_select_sql() . ') t'
762	. ' ON DUPLICATE KEY UPDATE'
763	. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
764	. ' digest = VALUES(digest)'
765	);
766	if ( false === $writes ) {
767	throw new RuntimeException( 'rebuild stored digests failed: ' . $wpdb->last_error );
768	}
769	update_option( Digest_Index::FORMULA_FP_OPTION, Digest_Index::digest_formula_fingerprint(), false );
770
771	if ( $products_only ) {
772	$stored_total = (int) $wpdb->get_var(
773	'SELECT COUNT(*) FROM ' . $this->table_name() . ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
774	);
775
776	return array(
777	'writes' => (int) $writes,
778	'orphans_deleted' => (int) $orphans_deleted,
779	'stored_total' => $stored_total,
780	'duration_ms' => round( ( microtime( true ) - $started ) * 1000, 3 ),
781	);
782	}
783
784	// Leg-3 phase 7 (ADR 0015): customers share the digest table via their own 'customer' rows —
785	// the same prune-orphans + INSERT…SELECT pass, over the customer predicate + id-space. A stored
786	// customer whose user vanished or lost the customer role is an orphan (a role removal never fires
787	// before_delete_post, so the rebuild is the backstop that reconciles it).
788	$customer_orphans = $wpdb->query(
789	'DELETE FROM ' . $this->table_name()
790	. " WHERE object_type = 'customer'"
791	. ' AND NOT ' . $this->index->customer_live_row_exists_sql( 'object_id' )
792	);
793	if ( false === $customer_orphans ) {
794	throw new RuntimeException( 'prune orphan customer digests failed: ' . $wpdb->last_error );
795	}
796	$customer_writes = $wpdb->query(
797	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
798	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
799	. ' FROM (' . $this->index->customer_digest_select_sql() . ') t'
800	. ' ON DUPLICATE KEY UPDATE'
801	. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
802	. ' digest = VALUES(digest)'
803	);
804	if ( false === $customer_writes ) {
805	throw new RuntimeException( 'rebuild customer digests failed: ' . $wpdb->last_error );
806	}
807
808	// Leg-3 phase 7 (ADR 0015): orders share the digest table via their own 'order' rows (HPOS or CPT).
809	// Same prune-orphans + INSERT…SELECT pass; Digest_Index::order_digest_select_sql() emits the storage-correct SQL
810	// (the CPT path GROUP BYs, the HPOS path does not — both valid as an INSERT…SELECT source).
811	$order_orphans = $wpdb->query(
812	'DELETE FROM ' . $this->table_name()
813	. " WHERE object_type = 'order'"
814	. ' AND NOT ' . $this->index->order_live_row_exists_sql( 'object_id' )
815	);
816	if ( false === $order_orphans ) {
817	throw new RuntimeException( 'prune orphan order digests failed: ' . $wpdb->last_error );
818	}
819	$order_writes = $wpdb->query(
820	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
821	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
822	. ' FROM (' . $this->index->order_digest_select_sql() . ') t'
823	. ' ON DUPLICATE KEY UPDATE'
824	. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
825	. ' digest = VALUES(digest)'
826	);
827	if ( false === $order_writes ) {
828	throw new RuntimeException( 'rebuild order digests failed: ' . $wpdb->last_error );
829	}
830
831	$stored_total = (int) $wpdb->get_var(
832	'SELECT COUNT(*) FROM ' . $this->table_name() . ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
833	);
834	$customer_stored_total = (int) $wpdb->get_var(
835	'SELECT COUNT(*) FROM ' . $this->table_name() . " WHERE object_type = 'customer'"
836	);
837	$order_stored_total = (int) $wpdb->get_var(
838	'SELECT COUNT(*) FROM ' . $this->table_name() . " WHERE object_type = 'order'"
839	);
840
841	return array(
842	'writes' => (int) $writes + (int) $customer_writes + (int) $order_writes,
843	'orphans_deleted' => (int) $orphans_deleted + (int) $customer_orphans + (int) $order_orphans,
844	'stored_total' => $stored_total + $customer_stored_total + $order_stored_total,
845	'duration_ms' => round( ( microtime( true ) - $started ) * 1000, 3 ),
846	);
847	}
848	}
849