# woocommerce-pos/1.10.18/includes/Sync/Integrity_Digest.php

WCPOS – Point of Sale (POS) plugin for WooCommerce, version 1.10.18. 755 lines.

- Page: https://pluginprobe.com/plugins/woocommerce-pos/1.10.18/code/includes/Sync/Integrity_Digest.php
- Raw: https://pluginprobe.com/plugins/woocommerce-pos/1.10.18/raw/includes/Sync/Integrity_Digest.php
- Modified: 2026-09-18T11:11:16+00:00

Line numbers below start at 1. Link to a line or a range by appending a fragment to the
page URL, for example `https://pluginprobe.com/plugins/woocommerce-pos/1.10.18/code/includes/Sync/Integrity_Digest.php#L10-L20`.

```php
<?php
/**
 * WCPOS sync store component.
 *
 * @package WCPOS\WooCommercePOS\Sync
 */

namespace WCPOS\WooCommercePOS\Sync;

use WCPOS\WooCommercePOS\Logger;

// phpcs:disable Squiz.Commenting, Generic.Commenting -- Ported lab documentation is preserved verbatim.
// phpcs:disable WordPress.DB.PreparedSQL.NotPrepared -- Queries use internal table names and generated SQL fragments.
// phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Database failures are passed to exceptions, not rendered.

use RuntimeException;

/**
 * Hash-backed range-checksum support: stored per-record content digests.
 *
 * STORES a digest of each product/variation's raw DB row, marked dirty by the
 * same save/delete hooks class-change-log.php uses and written at the request
 * boundary (see $pending), so the integrity scan
 * can compare — entirely in SQL — the aggregate of CURRENT raw-row digests
 * against the aggregate of STORED digests per id-range bucket. If hooks
 * fired for every write, stored == current (and sequence-log already
 * reported the change); a bucket mismatch therefore means exactly "content
 * changed without hooks firing" — the sql-bypass signature — at GROUP BY
 * prices instead of revision-hash's full-hydration prices. Because the write
 * lands at the boundary, a hook-less write later in the SAME request is
 * absorbed into that request's digest; the scan catches bypasses between
 * requests, which is where they happen (a direct SQL job, an importer).
 *
 * The digest basis is deliberately the RAW DB ROW, NOT the filtered REST
 * payload: this signal is detection-only (discovery of WHERE drift
 * happened, ADR 0003 "discovery, never values"); hydration of anything the
 * POS trusts still goes through the filtered REST path. The flip side is
 * documented too: a raw-row digest cannot see a plugin changing the served
 * representation without touching the row — that staleness case remains
 * revision-hash territory.
 *
 * This class is the WRITE half. The READ half — every question the REST read
 * surface asks of the store, plus the canonical digest SQL both halves share —
 * lives in {@see Digest_Index}. The SQL-fragment accessors that used to hang off
 * this class are kept as deprecated delegates so existing callers keep working.
 */
final class Integrity_Digest {

	/**
	 * Wall-clock ms spent inside the digest write hooks during the CURRENT
	 * request. Read (and reset) by the product-edit fixture for the
	 * hook-overhead bench's per-component breakdown. Two microtime() calls
	 * per hook fire — negligible against the INSERT…SELECT it wraps.
	 */
	public static float $request_write_ms = 0.0;

	/**
	 * @see Digest_Index::DIGESTED_META_KEYS The BASELINE key set.
	 * The formula the digest actually uses is Digest_Index::digested_meta_keys(),
	 * which folds in the configured barcode key (mono#1234).
	 */
	public const DIGESTED_META_KEYS = Digest_Index::DIGESTED_META_KEYS;

	/** @see Digest_Index::CUSTOMER_DIGESTED_META_KEYS The digest formula's home. */
	public const CUSTOMER_DIGESTED_META_KEYS = Digest_Index::CUSTOMER_DIGESTED_META_KEYS;

	/** @see Digest_Index::ORDER_DIGESTED_META_KEYS The digest formula's home. */
	public const ORDER_DIGESTED_META_KEYS = Digest_Index::ORDER_DIGESTED_META_KEYS;

	/** @see Digest_Index::OBJECT_TYPES_SQL The product-space object types. */
	public const OBJECT_TYPES_SQL = Digest_Index::OBJECT_TYPES_SQL;

	public const REBUILD_HOOK     = 'wcpos_integrity_digest_rebuild';
	public const REBUILD_LOCK     = 'wcpos_integrity_digest_rebuild_lock';
	public const REBUILD_LOCK_TTL = 300;

	/**
	 * The read half + the canonical digest SQL. The write statements below compose
	 * their INSERT…SELECT sources from it, so stored and current digests are
	 * computed by ONE expression — the invariant the whole scan rests on.
	 */
	private Digest_Index $index;

	/**
	 * Digests owed but not yet written, keyed by blog, type and id.
	 *
	 * A stored digest is a pure function of the settled record, so only the
	 * LAST upsert in a request carries information — yet one Store API checkout
	 * ran the order INSERT…SELECT eleven times (35 ms) and, with account
	 * creation, the customer one six more times (measured 2026-09-03 on
	 * dev-next). Upserts land on {@see flush_pending_digests()}: at shutdown
	 * and before {@see Digest_Index::read_digests()}, so the pull lane never
	 * stamps a stale `_rxdb_digest`. Product and variation digests ride the
	 * same queue: `wc_reduce_stock_levels()` saves the quantity, then the stock
	 * status — two INSERT…SELECT statements per purchased product at checkout
	 * (measured 2026-09-03). The v2 write lane reads digests through that same
	 * read method, so its serializer still stamps a fresh `_rxdb_digest`.
	 *
	 * Static so the read path can flush without holding the observer. The first
	 * queuing instance binds the writer, including after shutdown; all instances
	 * write the same table. An empty shutdown uses a default instance instead.
	 * See Request_Write_Queue for the queue mechanics.
	 */
	private static ?Request_Write_Queue $pending = null;

	/**
	 * The most distinct records held; the next distinct record flushes them first.
	 * Sized for the
	 * realistic per-request maximum (a checkout touches an order and a customer;
	 * a REST batch a few dozen records) while keeping a WP-CLI import's deferred
	 * SQL and memory bounded.
	 */
	public const PENDING_DIGEST_FLUSH_THRESHOLD = 50;

	public function __construct( ?Digest_Index $index = null ) {
		$this->index = $index ?? new Digest_Index();
	}

	public function table_name(): string {
		return $this->index->table_name();
	}

	/**
	 * Separate current-state table rather than a column on the change-log:
	 * the change-log is an append-only event journal (many rows per object,
	 * tombstones included) while the stored digest is exactly one row per
	 * live object — different cardinality and lifecycle. Folding the digest
	 * into the log would force a latest-row-per-object subquery on every
	 * scan, destroying the GROUP BY price this design exists for.
	 *
	 * digest is BIGINT UNSIGNED holding a 64-bit value (top 16 hex of MD5): integer
	 * storage keeps the BIT_XOR bucket aggregate a pure integer fold with
	 * constant-size state, where a CHAR hash would need GROUP_CONCAT (and
	 * its max_len truncation hazard) to aggregate.
	 */
	public function schema_sql( string $table_name, string $charset_collate = '' ): string {
		return "CREATE TABLE {$table_name} (\n"
			. "  object_type VARCHAR(20) NOT NULL,\n"
			. "  object_id BIGINT UNSIGNED NOT NULL,\n"
			. "  digest BIGINT UNSIGNED NOT NULL,\n"
			. "  updated_gmt DATETIME NOT NULL,\n"
			. "  PRIMARY KEY  (object_type, object_id),\n"
			. "  KEY object_id (object_id)\n"
			. ") {$charset_collate};";
	}

	public function install(): void {
		global $wpdb;
		if ( ! function_exists( 'dbDelta' ) ) {
			require_once ABSPATH . 'wp-admin/includes/upgrade.php';
		}
		dbDelta( $this->schema_sql( $this->table_name(), $wpdb->get_charset_collate() ) );
	}


	/**
	 * Same save/delete hooks the change-log listens to (products and
	 * variations only — tax rates live in their own table outside the
	 * wp_posts id space this scan buckets; they stay covered by the plain
	 * range-checksum candidate, whose checksum covers the full rate row).
	 */
	public function register_hooks(): void {
		add_action( 'woocommerce_new_product', array( $this, 'record_post_saved' ), 10, 1 );
		add_action( 'woocommerce_update_product', array( $this, 'record_post_saved' ), 10, 1 );
		add_action( 'woocommerce_new_product_variation', array( $this, 'record_post_saved' ), 10, 1 );
		add_action( 'woocommerce_update_product_variation', array( $this, 'record_post_saved' ), 10, 1 );
		// Untrash does not reliably re-fire woocommerce_update_product; the
		// upsert is a no-op for non-live rows, so hooking it is free.
		add_action( 'untrashed_post', array( $this, 'record_post_untrashed' ), 10, 1 );
		add_action( 'wp_trash_post', array( $this, 'record_post_deleted' ), 10, 1 );
		add_action( 'before_delete_post', array( $this, 'record_post_deleted' ), 10, 1 );

		// Leg-3 phase 7 (ADR 0015): ALL WordPress users are POS customers under
		// #1379 (1.9 parity). Saves and role changes idempotently upsert their
		// digest; only delete_user removes it.
		add_action( 'user_register', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'profile_update', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'woocommerce_created_customer', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'woocommerce_new_customer', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'woocommerce_update_customer', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'set_user_role', array( $this, 'record_customer_saved' ), 10, 1 );
		// add_role()/remove_role() fire ONLY add_user_role/remove_user_role, so
		// register both to capture membership changes in the served record.
		add_action( 'add_user_role', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'remove_user_role', array( $this, 'record_customer_saved' ), 10, 1 );
		add_action( 'delete_user', array( $this, 'record_customer_deleted' ), 10, 1 );

		// Leg-3 phase 7 (ADR 0015): order digest maintenance. Storage-agnostic WC order hooks (fire under
		// HPOS AND CPT), matching the sync-index's order hooks. upsert/delete are idempotent (no dedup).
		add_action( 'woocommerce_new_order', array( $this, 'record_order_saved' ), 10, 1 );
		add_action( 'woocommerce_update_order', array( $this, 'record_order_saved' ), 10, 1 );
		add_action( 'woocommerce_before_trash_order', array( $this, 'record_order_deleted' ), 10, 1 );
		add_action( 'woocommerce_before_delete_order', array( $this, 'record_order_deleted' ), 10, 1 );
		// Untrash recreation: `untrashed_post` (handled by record_post_untrashed)
		// never fires for COT orders — without the HPOS twin hook a restored
		// order's digest is never recreated and integrity scans treat it as
		// deleted forever.
		add_action( 'woocommerce_untrash_order', array( $this, 'record_order_untrashed' ), 10, 1 );
		// Request boundary for the coalesced digest upserts (see
		// $pending). LAST on shutdown: WooCommerce saves the customer at
		// 10 and the session at 20. Zero accepted args: do_action( 'shutdown' )
		// passes an empty string otherwise.
		add_action( 'shutdown', array( __CLASS__, 'flush_pending_digests_at_shutdown' ), PHP_INT_MAX, 0 );
	}

	/**
	 * Recreate a COT order's digest once its restore completes.
	 *
	 * `woocommerce_untrash_order` fires BEFORE the data store restores the
	 * status, and the restore's internal save fires no observer hook we bind
	 * (verified: `woocommerce_update_order` does not fire there) — so an
	 * immediate upsert would read a still-trashed row and write nothing. Arm a
	 * one-shot on the order's first object save after it leaves the trash and
	 * upsert then.
	 *
	 * @param int $order_id Order being restored.
	 */
	public function record_order_untrashed( int $order_id ): void {
		$handler = function ( $order ) use ( $order_id, &$handler ): void {
			if ( ! \is_object( $order ) || ! method_exists( $order, 'get_id' ) || ! method_exists( $order, 'get_status' ) || (int) $order->get_id() !== $order_id || 'trash' === $order->get_status() ) {
				return;
			}
			remove_action( 'woocommerce_after_order_object_save', $handler );
			$this->record_order_saved( $order_id );
		};
		add_action( 'woocommerce_after_order_object_save', $handler );
	}

	/**
	 * Cron entry point for rebuilding unexpectedly empty or stale product digests.
	 */
	public static function run_scheduled_rebuild(): void {
		$lease = get_transient( self::REBUILD_LOCK );
		try {
			( new self() )->rebuild( true );
		} catch ( \Throwable $exception ) {
			Logger::error( 'WCPOS sync: scheduled integrity digest rebuild failed: ' . $exception->getMessage() );
		} finally {
			self::release_rebuild_lock( $lease );
		}
	}

	/**
	 * Release the rebuild lease only if this run still owns it — a rebuild that
	 * outlived the lock TTL must not delete a successor's fresh lease.
	 *
	 * @param mixed $lease The lease value captured when this run started.
	 */
	public static function release_rebuild_lock( $lease ): void {
		if ( get_transient( self::REBUILD_LOCK ) === $lease ) {
			delete_transient( self::REBUILD_LOCK );
		}
	}

	/**
	 * Wire THE digest stamper onto both served read lanes (#421 increment 3).
	 *
	 * ONE named static serves every digest id-space: it resolves the registry row
	 * from the lane's resource slug, so a collection that gains a digest group is
	 * stamped by adding a row and nothing else. The composed callback name this
	 * used to build (`stamp_proxy_{object_type}_digests`) could name a method that
	 * did not exist — add_filter() does not validate callables, so the miss only
	 * surfaced as a fatal at apply_filters() time, on a catalogue proxy read.
	 *
	 * Both public filter names stay live and both are registered here, so the order
	 * pull lane is wired by the same call as the proxy lane instead of by hand in
	 * Init. Returns the digest-and-proxy collections (the wiring golden pins them).
	 *
	 * @return string[] Collections whose served records carry a stored digest.
	 */
	public static function register_proxy_digest_stampers(): array {
		$registered = array();
		foreach ( Collections::with( 'digest' ) as $collection => $row ) {
			if ( ! isset( $row['proxy'] ) ) {
				continue;
			}
			$registered[] = $collection;
		}
		add_filter( 'woocommerce_pos_sync_proxy_response', array( __CLASS__, 'stamp_digests' ), 10, 3 );
		add_filter( 'woocommerce_pos_sync_order_pull_payloads', array( __CLASS__, 'stamp_digests' ), 10, 3 );

		return $registered;
	}

	/**
	 * Detach THE digest stamper from both served read lanes.
	 *
	 * The teardown twin of {@see register_proxy_digest_stampers()}, matching the
	 * `unregister_*` seams {@see Revision} and {@see Proxy_Uuid_Stamper} already
	 * expose. `Augmentation_Pipeline::reset()` only removes the projections the
	 * pipeline itself installed, so without this a caller that installs the real
	 * pipeline — a test wiring the production read lane — cannot unwind it and
	 * leaks this filter into everything that runs after it.
	 */
	public static function unregister_proxy_digest_stampers(): void {
		remove_filter( 'woocommerce_pos_sync_proxy_response', array( __CLASS__, 'stamp_digests' ), 10 );
		remove_filter( 'woocommerce_pos_sync_order_pull_payloads', array( __CLASS__, 'stamp_digests' ), 10 );
	}

	/**
	 * Attach each served record's stored 64-bit digest as a top-level `_rxdb_digest`
	 * string, so the client seeds its existence-reconcile manifest (ADR 0014 Leg 3)
	 * as records flow through the NORMAL pull — no separate fetch. The client reads
	 * it into the sidecar manifest; it is NOT persisted into the document. A record
	 * with no stored digest yet simply carries no `_rxdb_digest`.
	 *
	 * The lane's resource slug picks the id-space (the registry owns the mapping,
	 * including the slug traps), and a resource with no digest group — or none at
	 * all — returns the payload untouched.
	 *
	 * @param mixed $data     Served list of records.
	 * @param mixed $resource Lane resource slug.
	 * @param mixed $request  Request context.
	 *
	 * @return mixed
	 */
	public static function stamp_digests( $data, $resource = '', $request = null ) {
		if ( ! \is_array( $data ) || ! \is_string( $resource ) || '' === $resource ) {
			return $data;
		}
		$row = Collections::by_proxy_slug( $resource );
		if ( null === $row || ! isset( $row['digest'] ) ) {
			return $data;
		}
		$ids = array();
		foreach ( $data as $record ) {
			if ( \is_array( $record ) && isset( $record['id'] ) ) {
				$ids[] = (int) $record['id'];
			}
		}
		if ( array() === $ids ) {
			return $data;
		}
		$digests = ( new Digest_Index() )->read_digests( $row['_collection'], $ids );
		foreach ( $data as $index => $record ) {
			if ( \is_array( $record ) && isset( $record['id'] ) && isset( $digests[ (int) $record['id'] ] ) ) {
				$data[ $index ]['_rxdb_digest'] = $digests[ (int) $record['id'] ];
			}
		}

		return $data;
	}

	/**
	 * Canonical per-CUSTOMER digest SELECT (ADR 0015, Leg-3 phase 7).
	 *
	 * @deprecated Use {@see Digest_Index::customer_digest_select_sql()}.
	 */
	public function customer_digest_select_sql( string $where_sql = '' ): string {
		return $this->index->customer_digest_select_sql( $where_sql );
	}

	/**
	 * Customer digest maintenance (ADR 0015, Leg-3 phase 7) — every WordPress
	 * user is a POS customer, so saves and role changes always upsert.
	 */
	/** Owe the customer's digest; it is written once, on flush (see $pending). */
	public function record_customer_saved( int $user_id ): void {
		$this->defer( self::pending_type( 'customers' ), $user_id );
	}

	/**
	 * Queue one digest upsert, or write it now if the boundary has passed.
	 *
	 * @param string $type Registry digest object-types key.
	 * @param int    $id   Record id.
	 */
	private function defer( string $type, int $id ): void {
		self::queue( $this )->owe( $type, $id );
	}

	/** Bind the first queuing instance, or a default for an empty shutdown. */
	private static function queue( ?self $writer = null ): Request_Write_Queue {
		if ( null === self::$pending ) {
			$writer        = $writer ?? new self();
			self::$pending = new Request_Write_Queue(
				self::PENDING_DIGEST_FLUSH_THRESHOLD,
				function ( $type, $id ) use ( $writer ): void {
					$writer->upsert_pending( $type, $id );
				}
			);
		}
		return self::$pending;
	}

	/** One queue discriminator per digest id-space, including shared product/variation ids. */
	private static function pending_type( string $collection ): string {
		return implode( ',', Collections::row( $collection )['digest']['object_types'] );
	}

	/** One queued upsert, under the observer's fail-open posture. */
	private function upsert_pending( string $type, int $id ): void {
		$this->observe(
			function () use ( $type, $id ): void {
				foreach ( Collections::with( 'digest' ) as $collection => $row ) {
					if ( self::pending_type( $collection ) === $type ) {
						$this->upsert_for( $row['digest']['id_space'], $id );
						return;
					}
				}
				Logger::warning( 'WCPOS sync: no digest collection matches queued type: ' . $type );
			}
		);
	}

	/**
	 * Write every owed digest.
	 *
	 * Called from the shutdown flush, from {@see Digest_Index::read_digests()}
	 * before it reads, and before a new record exceeds queue capacity. Writes go
	 * through the instance that first queued (so an injected Digest_Index is
	 * honoured) and under the blog each entry was recorded on. Each upsert keeps
	 * the observer's fail-open posture: a failure is logged and the scan
	 * self-heals. Safe to call repeatedly — a flushed digest is no longer pending.
	 */
	public static function flush_pending_digests(): void {
		if ( null !== self::$pending ) {
			self::$pending->flush();
		}
	}

	/**
	 * The `shutdown` callback: flush, then write every later save immediately.
	 */
	public static function flush_pending_digests_at_shutdown(): void {
		self::queue()->flush_at_shutdown();
	}

	/**
	 * Discard per-request coalescing state. Tests only: the PHPUnit process
	 * never reaches `shutdown`, so the static queue would
	 * leak between test cases otherwise.
	 *
	 * @internal
	 */
	public static function reset_request_state(): void {
		self::$pending = null;
	}

	public function record_customer_deleted( int $user_id ): void {
		$this->delete_for( 'customers', $user_id );
	}

	/**
	 * Observation hooks must never break the host write that fired them: a
	 * broken or missing digest store is a sync problem (the integrity scan and
	 * the health gate surface it), not a reason to fatal a WooCommerce save.
	 * The ops paths (rebuild/prune) keep throwing — they run on demand and
	 * want the loudness.
	 *
	 * @param callable $observer The digest write to attempt.
	 */
	private function observe( callable $observer ): void {
		try {
			$observer();
		} catch ( \Throwable $e ) {
			Logger::error( 'Sync digest observer failed (sync will self-heal via scan/rebuild): ' . $e->getMessage() );
		}
	}

	/**
	 * Order digest maintenance (ADR 0015, Leg-3 phase 7). The WC order hooks are storage-agnostic (fire
	 * under HPOS AND CPT); the digest SQL's `type='shop_order'` filter makes the upsert a no-op for any
	 * non-order, so no type re-check is needed here.
	 */
	/** Owe the order's digest; it is written once, on flush (see $pending). */
	public function record_order_saved( int $order_id ): void {
		$this->defer( self::pending_type( 'orders' ), $order_id );
	}

	public function record_order_deleted( int $order_id ): void {
		$this->delete_for( 'orders', $order_id );
	}

	/**
	 * Order analogue of {@see upsert_customer_digest} (HPOS or CPT).
	 *
	 * @deprecated Use record_order_saved().
	 */
	public function upsert_order_digest( int $order_id ): void {
		$this->upsert_for( 'orders', $order_id );
	}

	/**
	 * Owe the product's or variation's digest; it is written once, on flush (see
	 * $pending). Both share the registry's queue key: the upsert's SQL
	 * derives the stored object_type from the row, so nothing here needs to.
	 */
	public function record_post_saved( int $post_id ): void {
		$this->defer( self::pending_type( 'products' ), $post_id );
	}

	public function record_post_untrashed( int $post_id ): void {
		$post_type = get_post_type( $post_id );
		if ( 'shop_order' === $post_type ) {
			$this->record_order_saved( $post_id );
			return;
		}
		if ( in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
			$this->record_post_saved( $post_id );
		}
	}

	public function record_post_deleted( int $post_id ): void {
		$post_type = get_post_type( $post_id );
		if ( ! in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
			return;
		}
		$this->delete_for( 'products', $post_id, 'product_variation' === $post_type );
	}

	/** Cancel an owed upsert and remove the registry-selected stored row. */
	private function delete_for( string $collection, int $id, bool $child = false ): void {
		if ( null !== self::$pending ) {
			self::$pending->drop( self::pending_type( $collection ), $id );
		}
		$this->observe(
			function () use ( $collection, $id, $child ): void {
				global $wpdb;
				$row     = Collections::row( $collection );
				$digest  = $row['digest'];
				$started = microtime( true );
				$deleted = $wpdb->delete(
					$this->table_name(),
					array(
						'object_type' => $child ? $digest['child_type'] : $row['object_type'],
						'object_id'   => $id,
					),
					array( '%s', '%d' )
				);
				if ( 'products' === $digest['id_space'] ) {
					self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
				}
				if ( false === $deleted ) {
					$label = $digest['label'];
					throw new RuntimeException( 'delete stored ' . $label . 'digest failed: ' . $wpdb->last_error );
				}
			}
		);
	}

	/**
	 * One statement: the digest is computed in SQL from the raw row and
	 * upserted in the same statement — PHP never materializes the value.
	 * No-op for rows outside the live predicate (the delete hook owns those).
	 * @deprecated Use record_post_saved().
	 */
	public function upsert_digest( int $post_id ): void {
		$this->upsert_for( 'products', $post_id );
	}

	/**
	 * Customer analogue of {@see upsert_digest} (ADR 0015, Leg-3 phase 7):
	 * compute and store one WordPress user's customer digest in a single
	 * INSERT…SELECT. Only the delete hook removes it.
	 * @deprecated Use record_customer_saved().
	 */
	public function upsert_customer_digest( int $user_id ): void {
		$this->upsert_for( 'customers', $user_id );
	}

	/** Compute and store one row using its id-space's canonical SELECT and retry policy. */
	private function upsert_for( string $collection, int $id ): void {
		global $wpdb;
		$digest  = Collections::row( $collection )['digest'];
		$label   = $digest['label'];
		$started = microtime( true );
		$this->index->raise_group_concat_max_len();
		$this->query_with_retry(
			$wpdb->prepare(
				'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
				. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
				. ' FROM (' . $this->index->{$digest['select']}( $digest['id_column'] . ' = %d' ) . ') t'
				. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
				$id
			),
			'upsert stored ' . $label . 'digest failed: ',
			$started
		);
	}

	/**
	 * MySQL/MariaDB error numbers a second attempt can clear: 1020 ER_CHECKREAD
	 * ("Record has changed since last read"), 1205 ER_LOCK_WAIT_TIMEOUT, 1213
	 * ER_LOCK_DEADLOCK. Two requests upserting the same digest row race on
	 * the `INSERT … ON DUPLICATE KEY UPDATE`; the retry reads the updated row.
	 */
	private const TRANSIENT_CONTENTION_ERRNOS = array( 1020, 1205, 1213 );

	/**
	 * Message fallback for the same three errors, used only when the driver's
	 * error number is unavailable (a wpdb without a live mysqli handle).
	 */
	private const TRANSIENT_CONTENTION_MESSAGES = array(
		'Record has changed since last read',
		'Lock wait timeout',
		'Deadlock found',
	);

	/** Retry a contended upsert once, including both attempts in the hook timing. */
	private function query_with_retry( string $sql, string $error_message, float $started ): void {
		global $wpdb;
		$result = $wpdb->query( $sql );
		if ( false === $result && $this->is_transient_contention( $wpdb ) ) {
			$result = $wpdb->query( $sql );
		}
		self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
		if ( false === $result ) {
			throw new RuntimeException( $error_message . $wpdb->last_error );
		}
	}

	/**
	 * The error number is authoritative: server messages are localised
	 * (`lc_messages`), so the English text is only a fallback for a handle-less
	 * wpdb. `$wpdb->dbh` is reachable through wpdb's magic getter.
	 */
	private function is_transient_contention( \wpdb $wpdb ): bool {
		$dbh = $wpdb->__get( 'dbh' );
		if ( $dbh instanceof \mysqli ) {
			$errno = mysqli_errno( $dbh ); // phpcs:ignore WordPress.DB.RestrictedFunctions -- reads the driver's last error number; no query is issued.
			if ( 0 !== $errno ) {
				return in_array( $errno, self::TRANSIENT_CONTENTION_ERRNOS, true );
			}
		}
		foreach ( self::TRANSIENT_CONTENTION_MESSAGES as $message ) {
			if ( false !== strpos( $wpdb->last_error, $message ) ) {
				return true;
			}
		}
		return false;
	}

	/**
	 * Backfill/repair: prune orphans, then digest every live row in one
	 * INSERT…SELECT pass. Pre-existing catalogs (the 10k seed) become fully
	 * digestable in one call; measured timing is returned so the lab can
	 * report the backfill price.
	 *
	 * @param bool $products_only Whether to stop after rebuilding product digests.
	 */
	public function rebuild( bool $products_only = false ): array {
		global $wpdb;
		$this->index->raise_group_concat_max_len();
		$started = microtime( true );

		$orphans_deleted = $wpdb->query(
			'DELETE FROM ' . $this->table_name()
			. ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
			. ' AND NOT ' . $this->index->live_row_exists_sql( 'object_id' )
		);
		if ( false === $orphans_deleted ) {
			throw new RuntimeException( 'prune orphan stored digests failed: ' . $wpdb->last_error );
		}

		// Affected-rows semantics of ON DUPLICATE KEY: 1 per insert, 2 per
		// update, 0 per already-matching row — reported raw as "writes".
		// updated_gmt is assigned FIRST and only when the digest actually
		// changed (assignments evaluate left-to-right, so the IF must read
		// the pre-update digest before the digest assignment overwrites it).
		// Otherwise a repeated rebuild rewrites UTC_TIMESTAMP() into every
		// row, counts the whole table as writes, and inflates the
		// hash-checksum baseline cost (codex review).
		$writes = $wpdb->query(
			'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
			. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
			. ' FROM (' . $this->index->row_digest_select_sql() . ') t'
			. ' ON DUPLICATE KEY UPDATE'
			. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
			. ' digest = VALUES(digest)'
		);
		if ( false === $writes ) {
			throw new RuntimeException( 'rebuild stored digests failed: ' . $wpdb->last_error );
		}
		update_option( Digest_Index::FORMULA_FP_OPTION, Digest_Index::digest_formula_fingerprint(), false );

		if ( $products_only ) {
			$stored_total = (int) $wpdb->get_var(
				'SELECT COUNT(*) FROM ' . $this->table_name() . ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
			);

			return array(
				'writes' => (int) $writes,
				'orphans_deleted' => (int) $orphans_deleted,
				'stored_total' => $stored_total,
				'duration_ms' => round( ( microtime( true ) - $started ) * 1000, 3 ),
			);
		}

		// Leg-3 phase 7 (ADR 0015): customers share the digest table via their own 'customer' rows —
		// the same prune-orphans + INSERT…SELECT pass, over the customer predicate + id-space. A stored
		// customer whose user vanished or lost the customer role is an orphan (a role removal never fires
		// before_delete_post, so the rebuild is the backstop that reconciles it).
		$customer_orphans = $wpdb->query(
			'DELETE FROM ' . $this->table_name()
			. " WHERE object_type = 'customer'"
			. ' AND NOT ' . $this->index->customer_live_row_exists_sql( 'object_id' )
		);
		if ( false === $customer_orphans ) {
			throw new RuntimeException( 'prune orphan customer digests failed: ' . $wpdb->last_error );
		}
		$customer_writes = $wpdb->query(
			'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
			. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
			. ' FROM (' . $this->index->customer_digest_select_sql() . ') t'
			. ' ON DUPLICATE KEY UPDATE'
			. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
			. ' digest = VALUES(digest)'
		);
		if ( false === $customer_writes ) {
			throw new RuntimeException( 'rebuild customer digests failed: ' . $wpdb->last_error );
		}

		// Leg-3 phase 7 (ADR 0015): orders share the digest table via their own 'order' rows (HPOS or CPT).
		// Same prune-orphans + INSERT…SELECT pass; Digest_Index::order_digest_select_sql() emits the storage-correct SQL
		// (the CPT path GROUP BYs, the HPOS path does not — both valid as an INSERT…SELECT source).
		$order_orphans = $wpdb->query(
			'DELETE FROM ' . $this->table_name()
			. " WHERE object_type = 'order'"
			. ' AND NOT ' . $this->index->order_live_row_exists_sql( 'object_id' )
		);
		if ( false === $order_orphans ) {
			throw new RuntimeException( 'prune orphan order digests failed: ' . $wpdb->last_error );
		}
		$order_writes = $wpdb->query(
			'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
			. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
			. ' FROM (' . $this->index->order_digest_select_sql() . ') t'
			. ' ON DUPLICATE KEY UPDATE'
			. ' updated_gmt = IF(digest <=> VALUES(digest), updated_gmt, VALUES(updated_gmt)),'
			. ' digest = VALUES(digest)'
		);
		if ( false === $order_writes ) {
			throw new RuntimeException( 'rebuild order digests failed: ' . $wpdb->last_error );
		}

		$stored_total = (int) $wpdb->get_var(
			'SELECT COUNT(*) FROM ' . $this->table_name() . ' WHERE object_type IN ' . self::OBJECT_TYPES_SQL
		);
		$customer_stored_total = (int) $wpdb->get_var(
			'SELECT COUNT(*) FROM ' . $this->table_name() . " WHERE object_type = 'customer'"
		);
		$order_stored_total = (int) $wpdb->get_var(
			'SELECT COUNT(*) FROM ' . $this->table_name() . " WHERE object_type = 'order'"
		);

		return array(
			'writes' => (int) $writes + (int) $customer_writes + (int) $order_writes,
			'orphans_deleted' => (int) $orphans_deleted + (int) $customer_orphans + (int) $order_orphans,
			'stored_total' => $stored_total + $customer_stored_total + $order_stored_total,
			'duration_ms' => round( ( microtime( true ) - $started ) * 1000, 3 ),
		);
	}
}

```