<?php

namespace Fotogenic_AI;

if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

// phpcs:disable WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared,PluginCheck.Security.DirectDB.UnescapedDBParameter -- Store Queue queries operate only on plugin-owned custom tables with normalized identifiers and prepared values.

/**
 * Durable parent queue control plane for bounded Product Bulk children.
 */
final class Store_Queue_Repo {
	public const SCHEMA_VERSION = 1;
	private const SCHEMA_VERSION_OPTION = 'fotogenic_ai_store_queue_schema_version';
	private const QUEUE_STATUSES = [ 'preparing', 'reviewing', 'ready', 'running', 'paused', 'stopping', 'completed', 'partial', 'failed', 'stopped', 'uncertain', 'expired' ];
	private const ITEM_STATUSES = [ 'ready', 'ineligible', 'reserved', 'admitted', 'review_required', 'completed', 'partial', 'failed', 'skipped', 'uncertain' ];
	private const BATCH_STATUSES = [ 'created', 'admitted', 'running', 'paused', 'completed', 'partial', 'failed', 'stopped', 'uncertain' ];

	private static ?Store_Queue_Repo $instance = null;
	private bool $db_ready = false;
	private bool $db_failed = false;
	private string $queues_table;
	private string $items_table;
	private string $batches_table;

	public function __construct() {
		global $wpdb;
		$this->queues_table  = $wpdb->prefix . 'fotogenic_ai_store_queues';
		$this->items_table   = $wpdb->prefix . 'fotogenic_ai_store_queue_items';
		$this->batches_table = $wpdb->prefix . 'fotogenic_ai_store_queue_batches';
	}

	public static function instance(): Store_Queue_Repo {
		if ( null === self::$instance ) {
			self::$instance = new self();
		}

		return self::$instance;
	}

	public function ensure_tables(): bool {
		if ( $this->db_ready ) {
			return $this->transactional_storage_status()['available'];
		}
		if ( $this->db_failed || (int) get_option( self::SCHEMA_VERSION_OPTION, 0 ) > self::SCHEMA_VERSION ) {
			return false;
		}
		if ( ! $this->schema_is_current() && ! $this->prepare_schema_for_migration() ) {
			$this->db_failed = true;
			return false;
		}
		if ( ! $this->schema_is_current() || ! $this->transactional_storage_status()['available'] ) {
			$this->db_failed = true;
			return false;
		}
		if ( ! update_option( self::SCHEMA_VERSION_OPTION, self::SCHEMA_VERSION, false )
			&& self::SCHEMA_VERSION !== (int) get_option( self::SCHEMA_VERSION_OPTION, 0 ) ) {
			$this->db_failed = true;
			return false;
		}

		$this->db_ready = true;
		return true;
	}

	public function prepare_schema_for_migration(): bool {
		if ( (int) get_option( self::SCHEMA_VERSION_OPTION, 0 ) > self::SCHEMA_VERSION ) {
			return false;
		}

		global $wpdb;
		require_once ABSPATH . 'wp-admin/includes/upgrade.php';
		$charset = $wpdb->get_charset_collate();

		$queues_sql = "CREATE TABLE {$this->queues_table} (
			id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
			queue_uuid char(36) NOT NULL,
			user_id bigint(20) unsigned NOT NULL,
			workflow varchar(32) NOT NULL DEFAULT 'product_images',
			source_type varchar(16) NOT NULL DEFAULT 'product',
			schema_version int(11) unsigned NOT NULL DEFAULT 1,
			request_fingerprint char(64) NOT NULL,
			snapshot_fingerprint char(64) NULL,
			status varchar(32) NOT NULL DEFAULT 'preparing',
			status_reason varchar(64) NULL,
			revision bigint(20) unsigned NOT NULL DEFAULT 1,
			history_label varchar(191) NOT NULL DEFAULT '',
			history_search_text varchar(191) NOT NULL DEFAULT '',
			selection_snapshot longtext NULL,
			draft_snapshot longtext NULL,
			consent_snapshot longtext NULL,
			scan_high_water_id bigint(20) unsigned NOT NULL DEFAULT 0,
			scan_after_id bigint(20) unsigned NOT NULL DEFAULT 0,
			scan_ordinal bigint(20) unsigned NOT NULL DEFAULT 0,
			matched_count bigint(20) unsigned NOT NULL DEFAULT 0,
			selected_count bigint(20) unsigned NOT NULL DEFAULT 0,
			ineligible_count bigint(20) unsigned NOT NULL DEFAULT 0,
			expected_output_count bigint(20) unsigned NOT NULL DEFAULT 0,
			estimated_credits bigint(20) unsigned NOT NULL DEFAULT 0,
			completed_count bigint(20) unsigned NOT NULL DEFAULT 0,
			partial_count bigint(20) unsigned NOT NULL DEFAULT 0,
			failed_count bigint(20) unsigned NOT NULL DEFAULT 0,
			skipped_count bigint(20) unsigned NOT NULL DEFAULT 0,
			attention_count bigint(20) unsigned NOT NULL DEFAULT 0,
			next_batch_ordinal int(11) unsigned NOT NULL DEFAULT 1,
			created_at datetime NOT NULL,
			started_at datetime NULL,
			updated_at datetime NOT NULL,
			finished_at datetime NULL,
			last_heartbeat_at datetime NULL,
			lock_owner varchar(64) NULL,
			lock_expires_at datetime NULL,
			last_error text NULL,
			PRIMARY KEY  (id),
			UNIQUE KEY queue_uuid (queue_uuid),
			KEY user_request_created (user_id, request_fingerprint, created_at),
			KEY user_status_created (user_id, status, created_at),
			KEY status_updated (status, updated_at)
		) ENGINE=InnoDB {$charset};";

		$items_sql = "CREATE TABLE {$this->items_table} (
			id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
			queue_id bigint(20) unsigned NOT NULL,
			item_ordinal bigint(20) unsigned NOT NULL,
			subject_type varchar(16) NOT NULL DEFAULT 'product',
			subject_id bigint(20) unsigned NOT NULL,
			search_label varchar(191) NOT NULL DEFAULT '',
			sku varchar(100) NOT NULL DEFAULT '',
			status varchar(32) NOT NULL DEFAULT 'ready',
			status_reason varchar(64) NULL,
			summary_snapshot longtext NULL,
			target_snapshot longtext NULL,
			source_snapshot longtext NULL,
			source_fingerprint char(64) NULL,
			batch_id bigint(20) unsigned NULL,
			bulk_item_id bigint(20) unsigned NULL,
			claim_owner varchar(64) NULL,
			claim_expires_at datetime NULL,
			created_at datetime NOT NULL,
			updated_at datetime NOT NULL,
			finished_at datetime NULL,
			error_code varchar(64) NULL,
			error_message text NULL,
			PRIMARY KEY  (id),
			UNIQUE KEY queue_ordinal (queue_id, item_ordinal),
			UNIQUE KEY queue_subject (queue_id, subject_type, subject_id),
			KEY queue_status_ordinal (queue_id, status, item_ordinal),
			KEY batch_ordinal (batch_id, item_ordinal),
			KEY search_queue (search_label, queue_id),
			KEY sku_queue (sku, queue_id)
		) ENGINE=InnoDB {$charset};";

		$batches_sql = "CREATE TABLE {$this->batches_table} (
			id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
			queue_id bigint(20) unsigned NOT NULL,
			batch_ordinal int(11) unsigned NOT NULL,
			bulk_run_uuid char(36) NOT NULL,
			status varchar(32) NOT NULL DEFAULT 'created',
			status_reason varchar(64) NULL,
			selected_count int(11) unsigned NOT NULL DEFAULT 0,
			pricing_fingerprint char(64) NULL,
			configuration_fingerprint char(64) NULL,
			created_at datetime NOT NULL,
			started_at datetime NULL,
			updated_at datetime NOT NULL,
			finished_at datetime NULL,
			error_code varchar(64) NULL,
			error_message text NULL,
			PRIMARY KEY  (id),
			UNIQUE KEY queue_batch (queue_id, batch_ordinal),
			UNIQUE KEY bulk_run_uuid (bulk_run_uuid),
			KEY queue_status_batch (queue_id, status, batch_ordinal)
		) ENGINE=InnoDB {$charset};";

		dbDelta( $queues_sql );
		dbDelta( $items_sql );
		dbDelta( $batches_sql );
		$this->db_failed = false;

		return $this->schema_is_current() && $this->transactional_storage_status()['available'];
	}

	/** @return array{available: bool, code: string, message: string, engines: array<string, string>} */
	public function transactional_storage_status(): array {
		return Platform_Capabilities::instance()->transactional_storage_status(
			[ $this->queues_table, $this->items_table, $this->batches_table ]
		);
	}

	/**
	 * @return array{ok: bool, created: bool, conflict: bool, reason: string, queue: array|null}
	 */
	public function admit_queue( array $data ): array {
		if ( ! $this->ensure_tables() ) {
			return $this->admission_result( false, false, false, 'schema_unavailable', null );
		}
		$queue_uuid = strtolower( trim( (string) ( $data['queue_uuid'] ?? '' ) ) );
		$user_id    = absint( $data['user_id'] ?? 0 );
		$fingerprint = $this->normalize_fingerprint( $data['request_fingerprint'] ?? '' );
		if ( $user_id <= 0 || ! wp_is_uuid( $queue_uuid ) || null === $fingerprint ) {
			return $this->admission_result( false, false, false, 'invalid_admission', null );
		}

		global $wpdb;
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return $this->admission_result( false, false, false, 'transaction_start_failed', null );
		}
		try {
			$existing = $wpdb->get_row(
				$wpdb->prepare( "SELECT * FROM {$this->queues_table} WHERE queue_uuid = %s FOR UPDATE", $queue_uuid ),
				ARRAY_A
			);
			if ( is_array( $existing ) ) {
				$matches = $user_id === (int) $existing['user_id']
					&& hash_equals( (string) $existing['request_fingerprint'], $fingerprint );
				$wpdb->query( 'COMMIT' );
				return $matches
					? $this->admission_result( true, false, false, '', $existing )
					: $this->admission_result( false, false, true, 'request_fingerprint_conflict', null );
			}

			$now    = current_time( 'mysql', true );
			$status = sanitize_key( (string) ( $data['status'] ?? 'preparing' ) );
			$status = in_array( $status, [ 'preparing', 'ready' ], true ) ? $status : 'preparing';
			$inserted = $wpdb->insert(
				$this->queues_table,
				[
					'queue_uuid'             => $queue_uuid,
					'user_id'                => $user_id,
					'workflow'               => 'product_images',
					'source_type'            => 'product',
					'schema_version'         => self::SCHEMA_VERSION,
					'request_fingerprint'    => $fingerprint,
					'status'                 => $status,
					'status_reason'          => $this->nullable_key( $data['status_reason'] ?? '' ),
					'history_label'          => $this->bounded_text( $data['history_label'] ?? '', 191 ),
					'history_search_text'    => $this->bounded_text( $data['history_search_text'] ?? '', 191 ),
					'selection_snapshot'     => $this->normalize_json( $data['selection_snapshot'] ?? [] ),
					'draft_snapshot'         => $this->normalize_json( $data['draft_snapshot'] ?? [] ),
					'scan_high_water_id'     => absint( $data['scan_high_water_id'] ?? 0 ),
					'created_at'             => $now,
					'updated_at'             => $now,
				]
			);
			if ( false === $inserted || (int) $wpdb->insert_id <= 0 ) {
				throw new \RuntimeException( 'queue_persistence_failed' );
			}
			$queue = $this->get_queue_by_id( (int) $wpdb->insert_id, $user_id );
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				throw new \RuntimeException( 'transaction_commit_failed' );
			}

			return $this->admission_result( true, true, false, '', $queue );
		} catch ( \Throwable $error ) {
			$wpdb->query( 'ROLLBACK' );
			return $this->admission_result( false, false, false, sanitize_key( $error->getMessage() ), null );
		}
	}

	public function get_queue_by_uuid( string $queue_uuid, int $user_id ): ?array {
		if ( ! $this->ensure_tables() || $user_id <= 0 || ! wp_is_uuid( $queue_uuid ) ) {
			return null;
		}
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare( "SELECT * FROM {$this->queues_table} WHERE queue_uuid = %s AND user_id = %d", strtolower( $queue_uuid ), $user_id ),
			ARRAY_A
		);

		return is_array( $row ) ? $row : null;
	}

	public function get_queue_by_id( int $queue_id, int $user_id ): ?array {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 ) {
			return null;
		}
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare( "SELECT * FROM {$this->queues_table} WHERE id = %d AND user_id = %d", $queue_id, $user_id ),
			ARRAY_A
		);

		return is_array( $row ) ? $row : null;
	}

	public function get_active_queue_for_user( int $user_id ): ?array {
		if ( ! $this->ensure_tables() || $user_id <= 0 ) {
			return null;
		}
		global $wpdb;
		$placeholders = implode( ',', array_fill( 0, 6, '%s' ) );
		$params       = array_merge( [ $user_id ], [ 'preparing', 'reviewing', 'ready', 'running', 'paused', 'stopping' ] );
		$row          = $wpdb->get_row(
			$wpdb->prepare(
				"SELECT * FROM {$this->queues_table} WHERE user_id = %d AND status IN ({$placeholders}) ORDER BY created_at DESC, id DESC LIMIT 1",
				$params
			),
			ARRAY_A
		);

		return is_array( $row ) ? $row : null;
	}

	/**
	 * Append one committed keyset page and advance its cursor atomically.
	 *
	 * @return array{ok: bool, replay: bool, inserted: int, queue: array|null, reason: string}
	 */
	public function append_scan_page( int $queue_id, int $user_id, int $expected_after_id, array $items, int $next_after_id, int $high_water_id, bool $done ): array {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 || count( $items ) > 100 ) {
			return [ 'ok' => false, 'replay' => false, 'inserted' => 0, 'queue' => null, 'reason' => 'invalid_scan_page' ];
		}

		global $wpdb;
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return [ 'ok' => false, 'replay' => false, 'inserted' => 0, 'queue' => null, 'reason' => 'transaction_start_failed' ];
		}
		try {
			$queue = $wpdb->get_row(
				$wpdb->prepare( "SELECT * FROM {$this->queues_table} WHERE id = %d AND user_id = %d FOR UPDATE", $queue_id, $user_id ),
				ARRAY_A
			);
			if ( ! is_array( $queue ) || ! in_array( (string) $queue['status'], [ 'preparing', 'reviewing' ], true ) ) {
				throw new \RuntimeException( 'queue_not_preparing' );
			}
			$current_after = (int) $queue['scan_after_id'];
			if ( $current_after >= $next_after_id && $next_after_id > $expected_after_id ) {
				$wpdb->query( 'COMMIT' );
				return [ 'ok' => true, 'replay' => true, 'inserted' => 0, 'queue' => $queue, 'reason' => '' ];
			}
			if ( $current_after !== $expected_after_id || $next_after_id < $expected_after_id ) {
				throw new \RuntimeException( 'scan_cursor_conflict' );
			}

			$ordinal    = (int) $queue['scan_ordinal'];
			$inserted   = 0;
			$eligible   = 0;
			$ineligible = 0;
			$now        = current_time( 'mysql', true );
			foreach ( $items as $item ) {
				$normalized = $this->normalize_item( $item );
				if ( null === $normalized ) {
					throw new \RuntimeException( 'invalid_scan_item' );
				}
				++$ordinal;
				$normalized['queue_id']    = $queue_id;
				$normalized['item_ordinal'] = $ordinal;
				$normalized['created_at']  = $now;
				$normalized['updated_at']  = $now;
				$result = $wpdb->insert( $this->items_table, $normalized );
				if ( false === $result ) {
					throw new \RuntimeException( 'scan_item_persistence_failed' );
				}
				++$inserted;
				'ineligible' === $normalized['status'] ? ++$ineligible : ++$eligible;
			}

			$updated = $wpdb->update(
				$this->queues_table,
				[
					'scan_high_water_id' => $high_water_id,
					'scan_after_id'      => $next_after_id,
					'scan_ordinal'       => $ordinal,
					'matched_count'      => (int) $queue['matched_count'] + $inserted,
					'selected_count'     => (int) $queue['selected_count'] + $eligible,
					'ineligible_count'   => (int) $queue['ineligible_count'] + $ineligible,
					'status'             => $done ? 'reviewing' : 'preparing',
					'updated_at'         => $now,
					'last_heartbeat_at'  => $now,
				],
				[ 'id' => $queue_id, 'user_id' => $user_id, 'scan_after_id' => $expected_after_id ]
			);
			if ( 1 !== $updated ) {
				throw new \RuntimeException( 'scan_cursor_update_failed' );
			}
			if ( false === $wpdb->query( 'COMMIT' ) ) {
				throw new \RuntimeException( 'transaction_commit_failed' );
			}

			return [ 'ok' => true, 'replay' => false, 'inserted' => $inserted, 'queue' => $this->get_queue_by_id( $queue_id, $user_id ), 'reason' => '' ];
		} catch ( \Throwable $error ) {
			$wpdb->query( 'ROLLBACK' );
			return [ 'ok' => false, 'replay' => false, 'inserted' => 0, 'queue' => null, 'reason' => sanitize_key( $error->getMessage() ) ];
		}
	}

	/** @return array<int, array> */
	public function claim_ready_items( int $queue_id, int $user_id, int $limit, string $owner, int $ttl = 120 ): array {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 || '' === trim( $owner ) ) {
			return [];
		}
		$limit = min( 12, max( 1, $limit ) );
		$owner = $this->bounded_text( sanitize_key( $owner ), 64 );
		global $wpdb;
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return [];
		}
		try {
			$queue = $wpdb->get_row(
				$wpdb->prepare( "SELECT id FROM {$this->queues_table} WHERE id = %d AND user_id = %d AND status = 'running' FOR UPDATE", $queue_id, $user_id ),
				ARRAY_A
			);
			if ( ! is_array( $queue ) ) {
				throw new \RuntimeException( 'queue_not_running' );
			}
			$ids = array_map(
				'absint',
				(array) $wpdb->get_col(
					$wpdb->prepare(
						"SELECT id FROM {$this->items_table} WHERE queue_id = %d AND status = 'ready' ORDER BY item_ordinal ASC LIMIT %d FOR UPDATE",
						$queue_id,
						$limit
					)
				)
			);
			if ( empty( $ids ) ) {
				$wpdb->query( 'COMMIT' );
				return [];
			}
			$placeholders = implode( ',', array_fill( 0, count( $ids ), '%d' ) );
			$expires      = gmdate( 'Y-m-d H:i:s', time() + max( 30, min( 300, $ttl ) ) );
			$params       = array_merge( [ $owner, $expires, current_time( 'mysql', true ) ], $ids, [ $queue_id ] );
			$updated      = $wpdb->query(
				// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- The prepared replacement array matches the runtime-sized item placeholder list plus four fixed placeholders.
				$wpdb->prepare(
					"UPDATE {$this->items_table} SET status = 'reserved', claim_owner = %s, claim_expires_at = %s, updated_at = %s WHERE id IN ({$placeholders}) AND queue_id = %d AND status = 'ready'",
					$params
				)
			);
			if ( count( $ids ) !== (int) $updated ) {
				throw new \RuntimeException( 'item_claim_conflict' );
			}
			$rows = $wpdb->get_results(
				$wpdb->prepare(
					"SELECT * FROM {$this->items_table} WHERE id IN ({$placeholders}) AND queue_id = %d ORDER BY item_ordinal ASC",
					array_merge( $ids, [ $queue_id ] )
				),
				ARRAY_A
			);
			$wpdb->query( 'COMMIT' );
			return is_array( $rows ) ? $rows : [];
		} catch ( \Throwable $error ) {
			$wpdb->query( 'ROLLBACK' );
			return [];
		}
	}

	/** @return array{ok: bool, created: bool, reason: string, batch: array|null} */
	public function create_batch( int $queue_id, int $user_id, array $data ): array {
		$batch_ordinal = absint( $data['batch_ordinal'] ?? 0 );
		$bulk_run_uuid = strtolower( trim( (string) ( $data['bulk_run_uuid'] ?? '' ) ) );
		$selected_count = min( 12, max( 1, absint( $data['selected_count'] ?? 0 ) ) );
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 || $batch_ordinal <= 0 || ! wp_is_uuid( $bulk_run_uuid ) ) {
			return [ 'ok' => false, 'created' => false, 'reason' => 'invalid_batch', 'batch' => null ];
		}
		global $wpdb;
		if ( null === $this->get_queue_by_id( $queue_id, $user_id ) ) {
			return [ 'ok' => false, 'created' => false, 'reason' => 'queue_not_found', 'batch' => null ];
		}
		$existing = $this->get_batch( $queue_id, $batch_ordinal );
		if ( is_array( $existing ) ) {
			$matches = hash_equals( (string) $existing['bulk_run_uuid'], $bulk_run_uuid )
				&& $selected_count === (int) $existing['selected_count'];
			return [ 'ok' => $matches, 'created' => false, 'reason' => $matches ? '' : 'batch_conflict', 'batch' => $matches ? $existing : null ];
		}
		$now      = current_time( 'mysql', true );
		$inserted = $wpdb->insert(
			$this->batches_table,
			[
				'queue_id'                  => $queue_id,
				'batch_ordinal'             => $batch_ordinal,
				'bulk_run_uuid'             => $bulk_run_uuid,
				'status'                    => 'created',
				'selected_count'            => $selected_count,
				'pricing_fingerprint'       => $this->normalize_fingerprint( $data['pricing_fingerprint'] ?? '' ),
				'configuration_fingerprint' => $this->normalize_fingerprint( $data['configuration_fingerprint'] ?? '' ),
				'created_at'                => $now,
				'updated_at'                => $now,
			]
		);
		$batch = false !== $inserted ? $this->get_batch( $queue_id, $batch_ordinal ) : null;
		return [ 'ok' => is_array( $batch ), 'created' => is_array( $batch ), 'reason' => is_array( $batch ) ? '' : 'batch_persistence_failed', 'batch' => $batch ];
	}

	public function get_batch( int $queue_id, int $batch_ordinal ): ?array {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $batch_ordinal <= 0 ) {
			return null;
		}
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare( "SELECT * FROM {$this->batches_table} WHERE queue_id = %d AND batch_ordinal = %d", $queue_id, $batch_ordinal ),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	public function get_current_batch( int $queue_id ): ?array {
		if ( ! $this->ensure_tables() || $queue_id <= 0 ) {
			return null;
		}
		global $wpdb;
		$placeholders = implode( ',', array_fill( 0, 5, '%s' ) );
		$params       = array_merge( [ $queue_id ], [ 'created', 'admitted', 'running', 'paused', 'uncertain' ] );
		$row          = $wpdb->get_row(
			$wpdb->prepare( "SELECT * FROM {$this->batches_table} WHERE queue_id = %d AND status IN ({$placeholders}) ORDER BY batch_ordinal DESC LIMIT 1", $params ),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	public function get_latest_batch( int $queue_id ): ?array {
		if ( ! $this->ensure_tables() || $queue_id <= 0 ) {
			return null;
		}
		global $wpdb;
		$row = $wpdb->get_row(
			$wpdb->prepare( "SELECT * FROM {$this->batches_table} WHERE queue_id = %d ORDER BY batch_ordinal DESC LIMIT 1", $queue_id ),
			ARRAY_A
		);
		return is_array( $row ) ? $row : null;
	}

	/** @param array<int, array{queueItemId: int, bulkItemId: int}> $mappings */
	public function map_batch_items( int $queue_id, int $batch_id, array $mappings ): bool {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $batch_id <= 0 || empty( $mappings ) || count( $mappings ) > 12 ) {
			return false;
		}
		global $wpdb;
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return false;
		}
		try {
			foreach ( $mappings as $mapping ) {
				$item_id      = absint( $mapping['queueItemId'] ?? 0 );
				$bulk_item_id = absint( $mapping['bulkItemId'] ?? 0 );
				if ( $item_id <= 0 || $bulk_item_id <= 0 || 1 !== $wpdb->update(
					$this->items_table,
					[
						'batch_id'        => $batch_id,
						'bulk_item_id'    => $bulk_item_id,
						'status'          => 'admitted',
						'claim_owner'     => null,
						'claim_expires_at'=> null,
						'updated_at'      => current_time( 'mysql', true ),
					],
					[ 'id' => $item_id, 'queue_id' => $queue_id, 'status' => 'reserved' ]
				) ) {
					throw new \RuntimeException( 'batch_item_mapping_failed' );
				}
			}
			$wpdb->query( 'COMMIT' );
			return true;
		} catch ( \Throwable $error ) {
			$wpdb->query( 'ROLLBACK' );
			return false;
		}
	}

	public function release_reserved_items( int $queue_id, string $owner ): int {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || '' === trim( $owner ) ) {
			return 0;
		}
		global $wpdb;
		$updated = $wpdb->update(
			$this->items_table,
			[ 'status' => 'ready', 'claim_owner' => null, 'claim_expires_at' => null, 'updated_at' => current_time( 'mysql', true ) ],
			[ 'queue_id' => $queue_id, 'status' => 'reserved', 'claim_owner' => $this->bounded_text( sanitize_key( $owner ), 64 ) ]
		);
		return false === $updated ? 0 : (int) $updated;
	}

	public function mark_item_attention( int $queue_id, int $item_id, string $reason ): bool {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $item_id <= 0 ) {
			return false;
		}
		global $wpdb;
		$result = $wpdb->update(
			$this->items_table,
			[
				'status'           => 'review_required',
				'status_reason'    => $this->nullable_key( $reason ),
				'claim_owner'      => null,
				'claim_expires_at' => null,
				'updated_at'       => current_time( 'mysql', true ),
			],
			[ 'id' => $item_id, 'queue_id' => $queue_id ]
		);
		return false !== $result;
	}

	public function update_batch_status( int $queue_id, int $batch_id, string $status, string $reason = '', array $extra = [] ): bool {
		$status = sanitize_key( $status );
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $batch_id <= 0 || ! in_array( $status, self::BATCH_STATUSES, true ) ) {
			return false;
		}
		$allowed = [ 'started_at', 'finished_at', 'error_code', 'error_message' ];
		$data    = array_intersect_key( $extra, array_flip( $allowed ) );
		$data['status']        = $status;
		$data['status_reason'] = $this->nullable_key( $reason );
		$data['updated_at']    = current_time( 'mysql', true );
		global $wpdb;
		$result = $wpdb->update( $this->batches_table, $data, [ 'id' => $batch_id, 'queue_id' => $queue_id ] );
		return false !== $result;
	}

	/**
	 * Project authoritative terminal Bulk item states into their queue rows.
	 *
	 * @param array<int, array{id:int,status:string,error_code?:string,error_message?:string,finished_at?:string}> $bulk_items
	 */
	public function project_batch_items( int $queue_id, int $batch_id, array $bulk_items ): bool {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $batch_id <= 0 || count( $bulk_items ) > 12 ) {
			return false;
		}
		$status_map = [
			'completed' => 'completed',
			'partial'   => 'partial',
			'failed'    => 'failed',
			'skipped'   => 'skipped',
			'uncertain' => 'uncertain',
		];
		global $wpdb;
		if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
			return false;
		}
		try {
			foreach ( $bulk_items as $item ) {
				$bulk_item_id = absint( $item['id'] ?? 0 );
				$bulk_status  = sanitize_key( (string) ( $item['status'] ?? '' ) );
				if ( $bulk_item_id <= 0 || ! isset( $status_map[ $bulk_status ] ) ) {
					continue;
				}
				$result = $wpdb->update(
					$this->items_table,
					[
						'status'        => $status_map[ $bulk_status ],
						'status_reason' => $this->nullable_key( $item['error_code'] ?? '' ),
						'finished_at'   => $item['finished_at'] ?? current_time( 'mysql', true ),
						'updated_at'    => current_time( 'mysql', true ),
						'error_code'    => $this->nullable_key( $item['error_code'] ?? '' ),
						'error_message' => $this->bounded_text( $item['error_message'] ?? '', 500 ),
					],
					[ 'queue_id' => $queue_id, 'batch_id' => $batch_id, 'bulk_item_id' => $bulk_item_id ]
				);
				if ( false === $result ) {
					throw new \RuntimeException( 'batch_projection_failed' );
				}
			}
			$wpdb->query( 'COMMIT' );
			return true;
		} catch ( \Throwable $error ) {
			$wpdb->query( 'ROLLBACK' );
			return false;
		}
	}

	public function skip_unadmitted_items( int $queue_id ): int {
		if ( ! $this->ensure_tables() || $queue_id <= 0 ) {
			return 0;
		}
		global $wpdb;
		$now     = current_time( 'mysql', true );
		$updated = $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$this->items_table} SET status = 'skipped', status_reason = 'user_stop', claim_owner = NULL, claim_expires_at = NULL, finished_at = %s, updated_at = %s WHERE queue_id = %d AND status IN ('ready', 'reserved')",
				$now,
				$now,
				$queue_id
			)
		);
		return false === $updated ? 0 : (int) $updated;
	}

	/**
	 * @return array{items: array, page: int, perPage: int, totalItems: int, totalPages: int}
	 */
	public function page_items( int $queue_id, int $user_id, int $page = 1, int $per_page = 24, string $status = '', string $search = '' ): array {
		$empty = [ 'items' => [], 'page' => 1, 'perPage' => 24, 'totalItems' => 0, 'totalPages' => 1 ];
		if ( ! $this->ensure_tables() || null === $this->get_queue_by_id( $queue_id, $user_id ) ) {
			return $empty;
		}
		$page     = max( 1, $page );
		$per_page = min( 48, max( 1, $per_page ) );
		$status   = sanitize_key( $status );
		$status   = in_array( $status, self::ITEM_STATUSES, true ) ? $status : '';
		$search   = $this->bounded_text( $search, 100 );
		$where    = 'queue_id = %d';
		$params   = [ $queue_id ];
		if ( '' !== $status ) {
			$where   .= ' AND status = %s';
			$params[] = $status;
		}
		if ( '' !== $search ) {
			$where   .= ' AND (search_label LIKE %s OR sku = %s OR subject_id = %d)';
			global $wpdb;
			$params[] = $wpdb->esc_like( $search ) . '%';
			$params[] = $search;
			$params[] = ctype_digit( $search ) ? absint( $search ) : 0;
		}
		global $wpdb;
		// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- The bounded WHERE fragment and its matching values are assembled only from internal allowlisted filters above.
		$total = (int) $wpdb->get_var( $wpdb->prepare( "SELECT COUNT(*) FROM {$this->items_table} WHERE {$where}", $params ) );
		$offset = ( $page - 1 ) * $per_page;
		$rows = $wpdb->get_results(
			// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- The values array includes the bounded filter values plus LIMIT and OFFSET.
			$wpdb->prepare(
				"SELECT * FROM {$this->items_table} WHERE {$where} ORDER BY item_ordinal ASC LIMIT %d OFFSET %d",
				array_merge( $params, [ $per_page, $offset ] )
			),
			ARRAY_A
		);

		return [
			'items'       => is_array( $rows ) ? $rows : [],
			'page'        => $page,
			'perPage'     => $per_page,
			'totalItems'  => $total,
			'totalPages'  => max( 1, (int) ceil( $total / $per_page ) ),
		];
	}

	/** @return array<string, int> */
	public function aggregate_counts( int $queue_id, int $user_id ): array {
		$counts = array_fill_keys( self::ITEM_STATUSES, 0 );
		if ( ! $this->ensure_tables() || null === $this->get_queue_by_id( $queue_id, $user_id ) ) {
			return $counts;
		}
		global $wpdb;
		$rows = $wpdb->get_results(
			$wpdb->prepare( "SELECT status, COUNT(*) AS item_count FROM {$this->items_table} WHERE queue_id = %d GROUP BY status", $queue_id ),
			ARRAY_A
		);
		foreach ( (array) $rows as $row ) {
			$status = sanitize_key( (string) ( $row['status'] ?? '' ) );
			if ( array_key_exists( $status, $counts ) ) {
				$counts[ $status ] = (int) $row['item_count'];
			}
		}
		return $counts;
	}

	/** @return array{queues:array,total:int} */
	public function list_history_queues( int $user_id, int $page, int $per_page, string $search = '', string $public_status = 'all' ): array {
		if ( ! $this->ensure_tables() || $user_id <= 0 ) {
			return [ 'queues' => [], 'total' => 0 ];
		}
		$page     = max( 1, $page );
		$per_page = min( 48, max( 1, $per_page ) );
		$search   = $this->bounded_text( $search, 200 );
		$status_map = [
			'queued'     => [ 'preparing', 'reviewing', 'ready' ],
			'processing' => [ 'running' ],
			'paused'     => [ 'paused' ],
			'stopping'   => [ 'stopping' ],
			'completed'  => [ 'completed' ],
			'partial'    => [ 'partial' ],
			'failed'     => [ 'failed' ],
			'stopped'    => [ 'stopped' ],
			'uncertain'  => [ 'uncertain' ],
		];
		$where  = 'q.user_id = %d';
		$params = [ $user_id ];
		if ( 'all' !== $public_status && isset( $status_map[ $public_status ] ) ) {
			$statuses     = $status_map[ $public_status ];
			$placeholders = implode( ',', array_fill( 0, count( $statuses ), '%s' ) );
			$where       .= " AND q.status IN ({$placeholders})";
			$params       = array_merge( $params, $statuses );
		}
		if ( '' !== $search ) {
			$where   .= " AND (q.queue_uuid = %s OR q.history_label LIKE %s OR q.history_search_text LIKE %s OR EXISTS (SELECT 1 FROM {$this->items_table} qi WHERE qi.queue_id = q.id AND (qi.search_label LIKE %s OR qi.sku = %s OR qi.subject_id = %d)))";
			global $wpdb;
			$prefix   = $wpdb->esc_like( $search ) . '%';
			$params[] = strtolower( $search );
			$params[] = '%' . $wpdb->esc_like( $search ) . '%';
			$params[] = '%' . $wpdb->esc_like( $search ) . '%';
			$params[] = $prefix;
			$params[] = $search;
			$params[] = ctype_digit( $search ) ? absint( $search ) : 0;
		}
		global $wpdb;
		// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- The bounded WHERE fragment is assembled only from internal status/search clauses and a matching values array.
		$total  = (int) $wpdb->get_var( $wpdb->prepare( "SELECT COUNT(*) FROM {$this->queues_table} q WHERE {$where}", $params ) );
		$offset = ( $page - 1 ) * $per_page;
		$rows   = $wpdb->get_results(
			// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- The values array contains the bounded WHERE values plus LIMIT and OFFSET.
			$wpdb->prepare(
				"SELECT q.* FROM {$this->queues_table} q WHERE {$where} ORDER BY q.created_at DESC, q.id DESC LIMIT %d OFFSET %d",
				array_merge( $params, [ $per_page, $offset ] )
			),
			ARRAY_A
		);
		return [ 'queues' => is_array( $rows ) ? $rows : [], 'total' => $total ];
	}

	public function bulk_run_belongs_to_queue( string $bulk_run_uuid, int $user_id ): bool {
		if ( ! $this->ensure_tables() || ! wp_is_uuid( $bulk_run_uuid ) || $user_id <= 0 ) {
			return false;
		}
		global $wpdb;
		$found = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT b.id FROM {$this->batches_table} b INNER JOIN {$this->queues_table} q ON q.id = b.queue_id WHERE b.bulk_run_uuid = %s AND q.user_id = %d LIMIT 1",
				strtolower( $bulk_run_uuid ),
				$user_id
			)
		);
		return null !== $found;
	}

	/**
	 * Remove abandoned non-billable drafts after seven days and terminal queue records after thirty days.
	 *
	 * @return array{drafts:int,terminal:int}
	 */
	public function purge_expired(): array {
		if ( ! $this->ensure_tables() ) {
			return [ 'drafts' => 0, 'terminal' => 0 ];
		}
		global $wpdb;
		$draft_cutoff    = gmdate( 'Y-m-d H:i:s', time() - 7 * DAY_IN_SECONDS );
		$terminal_cutoff = gmdate( 'Y-m-d H:i:s', time() - 30 * DAY_IN_SECONDS );
		$groups = [
			'drafts'   => [ [ 'preparing', 'reviewing', 'ready' ], $draft_cutoff ],
			'terminal' => [ [ 'completed', 'partial', 'failed', 'stopped', 'uncertain', 'expired' ], $terminal_cutoff ],
		];
		$deleted = [ 'drafts' => 0, 'terminal' => 0 ];
		foreach ( $groups as $key => [ $statuses, $cutoff ] ) {
			$placeholders = implode( ',', array_fill( 0, count( $statuses ), '%s' ) );
			$params       = array_merge( $statuses, [ $cutoff ] );
			$queue_ids    = array_map(
				'absint',
				(array) $wpdb->get_col(
					// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- Fixed internal status list plus one cutoff value.
					$wpdb->prepare( "SELECT id FROM {$this->queues_table} WHERE status IN ({$placeholders}) AND updated_at < %s ORDER BY id ASC LIMIT 250", $params )
				)
			);
			if ( empty( $queue_ids ) ) {
				continue;
			}
			$id_placeholders = implode( ',', array_fill( 0, count( $queue_ids ), '%d' ) );
			if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
				continue;
			}
			try {
				// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber,WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- One placeholder is emitted for every normalized queue ID.
				$wpdb->query( $wpdb->prepare( "DELETE FROM {$this->items_table} WHERE queue_id IN ({$id_placeholders})", $queue_ids ) );
				// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber,WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- One placeholder is emitted for every normalized queue ID.
				$wpdb->query( $wpdb->prepare( "DELETE FROM {$this->batches_table} WHERE queue_id IN ({$id_placeholders})", $queue_ids ) );
				// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber,WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- One placeholder is emitted for every normalized queue ID.
				$count = $wpdb->query( $wpdb->prepare( "DELETE FROM {$this->queues_table} WHERE id IN ({$id_placeholders})", $queue_ids ) );
				if ( false === $count || false === $wpdb->query( 'COMMIT' ) ) {
					throw new \RuntimeException( 'store_queue_retention_failed' );
				}
				$deleted[ $key ] = (int) $count;
			} catch ( \Throwable $error ) {
				unset( $error );
				$wpdb->query( 'ROLLBACK' );
			}
		}
		return $deleted;
	}

	public function update_queue_cas( int $queue_id, int $user_id, int $expected_revision, array $data, bool $increment_revision = false ): bool {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 || $expected_revision <= 0 || empty( $data ) ) {
			return false;
		}
		$allowed = [ 'status', 'status_reason', 'snapshot_fingerprint', 'consent_snapshot', 'expected_output_count', 'estimated_credits', 'completed_count', 'partial_count', 'failed_count', 'skipped_count', 'attention_count', 'next_batch_ordinal', 'started_at', 'finished_at', 'last_heartbeat_at', 'last_error' ];
		$update  = array_intersect_key( $data, array_flip( $allowed ) );
		if ( isset( $update['status'] ) ) {
			$update['status'] = sanitize_key( (string) $update['status'] );
			if ( ! in_array( $update['status'], self::QUEUE_STATUSES, true ) ) {
				return false;
			}
		}
		if ( isset( $update['status_reason'] ) ) {
			$update['status_reason'] = $this->nullable_key( $update['status_reason'] );
		}
		if ( isset( $update['consent_snapshot'] ) && is_array( $update['consent_snapshot'] ) ) {
			$update['consent_snapshot'] = $this->normalize_json( $update['consent_snapshot'] );
		}
		$update['updated_at'] = current_time( 'mysql', true );
		if ( $increment_revision ) {
			global $wpdb;
			$sets   = [];
			$params = [];
			foreach ( $update as $column => $value ) {
				$sets[]  = "{$column} = %s";
				$params[] = $value;
			}
			$params = array_merge( $params, [ $queue_id, $user_id, $expected_revision ] );
			return 1 === $wpdb->query(
				// phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- The mutation columns are selected from the fixed allowlist above and each has one matching prepared value.
				$wpdb->prepare(
					'UPDATE ' . $this->queues_table . ' SET ' . implode( ', ', $sets ) . ', revision = revision + 1 WHERE id = %d AND user_id = %d AND revision = %d',
					$params
				)
			);
		}
		global $wpdb;
		return 1 === $wpdb->update( $this->queues_table, $update, [ 'id' => $queue_id, 'user_id' => $user_id, 'revision' => $expected_revision ] );
	}

	public function acquire_lease( int $queue_id, int $user_id, string $owner, int $ttl = 120 ): bool {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 || '' === trim( $owner ) ) {
			return false;
		}
		global $wpdb;
		$now     = current_time( 'mysql', true );
		$expires = gmdate( 'Y-m-d H:i:s', time() + max( 30, min( 300, $ttl ) ) );
		return 1 === $wpdb->query(
			$wpdb->prepare(
				"UPDATE {$this->queues_table} SET lock_owner = %s, lock_expires_at = %s, last_heartbeat_at = %s, updated_at = %s WHERE id = %d AND user_id = %d AND (lock_owner IS NULL OR lock_owner = '' OR lock_expires_at < %s OR lock_owner = %s)",
				$this->bounded_text( sanitize_key( $owner ), 64 ),
				$expires,
				$now,
				$now,
				$queue_id,
				$user_id,
				$now,
				$this->bounded_text( sanitize_key( $owner ), 64 )
			)
		);
	}

	public function release_lease( int $queue_id, int $user_id, string $owner ): void {
		if ( ! $this->ensure_tables() || $queue_id <= 0 || $user_id <= 0 ) {
			return;
		}
		global $wpdb;
		$wpdb->update(
			$this->queues_table,
			[ 'lock_owner' => null, 'lock_expires_at' => null, 'updated_at' => current_time( 'mysql', true ) ],
			[ 'id' => $queue_id, 'user_id' => $user_id, 'lock_owner' => $this->bounded_text( sanitize_key( $owner ), 64 ) ]
		);
	}

	private function schema_is_current(): bool {
		global $wpdb;
		foreach (
			[
				$this->queues_table  => [ 'queue_uuid', 'revision', 'scan_after_id', 'next_batch_ordinal' ],
				$this->items_table   => [ 'queue_id', 'item_ordinal', 'source_fingerprint', 'bulk_item_id' ],
				$this->batches_table => [ 'queue_id', 'batch_ordinal', 'bulk_run_uuid', 'status' ],
			]
			as $table => $required
		) {
			if ( $table !== (string) $wpdb->get_var( $wpdb->prepare( 'SHOW TABLES LIKE %s', $table ) ) ) {
				return false;
			}
			$columns = array_map( 'strval', (array) $wpdb->get_col( "SHOW COLUMNS FROM {$table}", 0 ) );
			if ( array_diff( $required, $columns ) ) {
				return false;
			}
		}
		return true;
	}

	private function normalize_item( array $item ): ?array {
		$subject_type = sanitize_key( (string) ( $item['subject_type'] ?? 'product' ) );
		$subject_id   = absint( $item['subject_id'] ?? 0 );
		$status       = sanitize_key( (string) ( $item['status'] ?? 'ready' ) );
		$fingerprint  = $this->normalize_fingerprint( $item['source_fingerprint'] ?? '' );
		if ( 'product' !== $subject_type || $subject_id <= 0 || ! in_array( $status, [ 'ready', 'ineligible' ], true ) ) {
			return null;
		}
		if ( 'ready' === $status && null === $fingerprint ) {
			return null;
		}
		return [
			'subject_type'      => $subject_type,
			'subject_id'        => $subject_id,
			'search_label'      => $this->bounded_text( $item['search_label'] ?? '', 191 ),
			'sku'               => $this->bounded_text( $item['sku'] ?? '', 100 ),
			'status'            => $status,
			'status_reason'     => $this->nullable_key( $item['status_reason'] ?? '' ),
			'summary_snapshot'  => $this->normalize_json( $item['summary_snapshot'] ?? [] ),
			'target_snapshot'   => $this->normalize_json( $item['target_snapshot'] ?? [] ),
			'source_snapshot'   => $this->normalize_json( $item['source_snapshot'] ?? [] ),
			'source_fingerprint'=> $fingerprint,
		];
	}

	private function normalize_fingerprint( $value ): ?string {
		$value = strtolower( trim( (string) $value ) );
		return 64 === strlen( $value ) && ctype_xdigit( $value ) ? $value : null;
	}

	private function normalize_json( $value ): string {
		$encoded = wp_json_encode( is_array( $value ) ? $value : [] );
		return is_string( $encoded ) ? $encoded : '[]';
	}

	private function nullable_key( $value ): ?string {
		$value = sanitize_key( (string) $value );
		return '' !== $value ? $this->bounded_text( $value, 64 ) : null;
	}

	private function bounded_text( $value, int $length ): string {
		$value = sanitize_text_field( (string) $value );
		return function_exists( 'mb_substr' ) ? mb_substr( $value, 0, $length ) : substr( $value, 0, $length );
	}

	/** @return array{ok: bool, created: bool, conflict: bool, reason: string, queue: array|null} */
	private function admission_result( bool $ok, bool $created, bool $conflict, string $reason, ?array $queue ): array {
		return compact( 'ok', 'created', 'conflict', 'reason', 'queue' );
	}
}
