<?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 -- Bulk repository queries operate only on plugin-owned custom tables with normalized IDs/status lists and prepared runtime placeholder lists.

final class Bulk_Repo {
    public const SCHEMA_VERSION = 4;
    private const SCHEMA_VERSION_OPTION = 'fotogenic_ai_bulk_schema_version';
    private const WORKFLOWS = [ 'product_images', 'campaign_set', 'increase_resolution' ];
    private const SUBJECT_TYPES = [ 'product', 'media', 'post', 'page' ];
    private const RUN_CREATE_COLUMNS = [
        'run_uuid',
        'user_id',
        'workflow',
        'request_fingerprint',
        'status',
        'created_at',
        'started_at',
        'updated_at',
        'finished_at',
        'last_heartbeat_at',
        'status_reason',
        'filters_snapshot',
        'selected_count',
        'completed_count',
        'failed_count',
        'last_error',
        'lock_owner',
        'lock_expires_at',
    ];
    private const RUN_MUTABLE_COLUMNS = [
        'workflow',
        'request_fingerprint',
        'status',
        'started_at',
        'updated_at',
        'finished_at',
        'last_heartbeat_at',
        'status_reason',
        'filters_snapshot',
        'selected_count',
        'completed_count',
        'failed_count',
        'last_error',
        'lock_owner',
        'lock_expires_at',
    ];
    private const MUTABLE_ITEM_COLUMNS = [
        'status',
        'attempts',
        'run_id_single',
        'job_ids',
        'attachment_id',
        'attachment_ids',
        'generation_started_at',
        'last_polled_at',
        'poll_count',
        'started_at',
        'finished_at',
        'updated_at',
        'error_code',
        'error_message',
        'meta',
        'claim_owner',
        'claim_expires_at',
    ];
    private const ATTACHMENT_REFERENCE_BATCH_SIZE = 250;
    private const HOUSEKEEPING_MAX_BATCH_SIZE = 100;
    private const TERMINAL_RUN_STATUSES = [ 'completed', 'stopped', 'failed' ];
    private const TERMINAL_ITEM_STATUSES = [ 'completed', 'partial', 'failed', 'skipped', 'uncertain' ];

    private static ?Bulk_Repo $instance = null;
    private bool $db_ready = false;
	private bool $db_failed = false;
	private string $runs_table;
	private string $items_table;
	/** @var callable */
	private $admission_transaction_query;

	public function __construct( ?callable $admission_transaction_query = null ) {
		global $wpdb;
		$this->runs_table = $wpdb->prefix . 'fotogenic_ai_bulk_runs';
		$this->items_table = $wpdb->prefix . 'fotogenic_ai_bulk_run_items';
		$this->admission_transaction_query = $admission_transaction_query ?? static function ( string $query ) {
			global $wpdb;
			return $wpdb->query( $query );
		};
    }

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

    /**
     * Create or upgrade the shared Bulk control-plane tables.
     */
    public function ensure_tables(): bool {
        if ( $this->db_ready ) {
			return $this->transactional_storage_status()['available'];
        }
        if ( $this->db_failed ) {
            return false;
        }
        $stored_version = (int) get_option( self::SCHEMA_VERSION_OPTION, 0 );
        if ( $stored_version > self::SCHEMA_VERSION ) {
            $this->db_failed = true;
            return false;
        }
        if ( $this->schema_is_current() ) {
			if ( ! $this->transactional_storage_status()['available'] ) {
				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;
        }
        $this->db_failed = true;
        return false;
    }

    /**
     * Install additive columns before the central cutover fills immutable identities.
     */
    public function prepare_schema_for_migration(): bool {
        $stored_version = (int) get_option( self::SCHEMA_VERSION_OPTION, 0 );
        if ( $stored_version > self::SCHEMA_VERSION ) {
            return false;
        }

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

        $runs_sql = "CREATE TABLE {$this->runs_table} (
            id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
            run_uuid char(36) NOT NULL,
            user_id bigint(20) unsigned NOT NULL,
            workflow varchar(32) NOT NULL DEFAULT 'product_images',
            request_fingerprint char(64) NULL,
            status varchar(32) NOT NULL DEFAULT 'created',
            created_at datetime NOT NULL,
            started_at datetime NULL,
            updated_at datetime NOT NULL,
            finished_at datetime NULL,
            last_heartbeat_at datetime NULL,
            status_reason varchar(64) NULL,
            filters_snapshot longtext NULL,
            selected_count int(11) NOT NULL DEFAULT 0,
            completed_count int(11) NOT NULL DEFAULT 0,
            failed_count int(11) NOT NULL DEFAULT 0,
            last_error text NULL,
            lock_owner varchar(64) NULL,
            lock_expires_at datetime NULL,
            PRIMARY KEY  (id),
            UNIQUE KEY run_uuid (run_uuid),
            KEY user_status_created (user_id, status, created_at),
            KEY user_workflow_status (user_id, workflow, status)
        ) ENGINE=InnoDB {$charset};";

        $items_sql = "CREATE TABLE {$this->items_table} (
            id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
            run_id bigint(20) unsigned NOT NULL,
            run_item_uuid char(36) NULL,
            campaign_group_uuid char(36) NULL,
            item_ordinal int(11) unsigned NOT NULL DEFAULT 0,
            subject_type varchar(16) NOT NULL DEFAULT 'product',
            subject_id bigint(20) unsigned NOT NULL DEFAULT 0,
            target_type varchar(20) NOT NULL DEFAULT 'unknown',
            parent_product_id bigint(20) unsigned NOT NULL DEFAULT 0,
            variation_id bigint(20) unsigned NOT NULL DEFAULT 0,
            target_snapshot longtext NULL,
            source_roles_snapshot longtext NULL,
            destination_snapshot longtext NULL,
            status varchar(32) NOT NULL DEFAULT 'pending',
            attempts int(11) NOT NULL DEFAULT 0,
            run_id_single char(36) NULL,
            request_id char(36) NULL,
            request_fingerprint char(64) NULL,
            command_snapshot longtext NULL,
            generation_key char(40) NOT NULL,
            job_ids longtext NULL,
            attachment_id bigint(20) unsigned NULL,
            attachment_ids longtext NULL,
            generation_started_at datetime NULL,
            last_polled_at datetime NULL,
            poll_count int(11) NOT NULL DEFAULT 0,
            started_at datetime NULL,
            finished_at datetime NULL,
            updated_at datetime NOT NULL,
            error_code varchar(64) NULL,
            error_message text NULL,
            meta longtext NULL,
            claim_owner varchar(64) NULL,
            claim_expires_at datetime NULL,
            PRIMARY KEY  (id),
            UNIQUE KEY run_item (run_id, run_item_uuid),
            UNIQUE KEY run_generation_key (run_id, generation_key),
            KEY run_target_lookup (run_id, target_type, subject_type, subject_id),
            KEY campaign_group (campaign_group_uuid),
            KEY run_status_id (run_id, status, id),
            KEY request_id (request_id),
            KEY attachment_id (attachment_id)
        ) ENGINE=InnoDB {$charset};";

        dbDelta( $runs_sql );
        dbDelta( $items_sql );
        $this->db_failed = false;
        return $this->schema_has_required_columns() && $this->transactional_storage_status()['available'];
    }

	/**
	 * Live storage-engine status for the exact Bulk transaction boundary.
	 *
	 * @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->runs_table, $this->items_table ]
		);
	}

    /**
     * Persist a universal Bulk run and its immutable subject graph.
     *
     * Every item must already contain its stable request UUID, canonical request
     * fingerprint, and normalized command snapshot. The graph is written before
     * the run is activated, so a worker can never observe a partially admitted
     * billable command.
     *
     * @return array{ok: bool, run_id: int, created: bool, activated: bool, run: array|null, reason: string}
     */
    public function admit_run( array $data, array $items ): array {
        if ( ! $this->ensure_tables() ) {
			$storage = $this->transactional_storage_status();
			$reason = 'transactional_storage_required' === $storage['code']
				? 'transactional_storage_required'
				: 'schema_unavailable';
			return $this->admission_result( false, 0, false, false, null, $reason );
        }
        $run_uuid = strtolower( trim( (string) ( $data['run_uuid'] ?? '' ) ) );
        $user_id = absint( $data['user_id'] ?? 0 );
        $workflow = sanitize_key( (string) ( $data['workflow'] ?? '' ) );
        $request_fingerprint = $this->normalize_fingerprint( $data['request_fingerprint'] ?? null );
        $final_status = sanitize_key( (string) ( $data['status'] ?? 'running' ) );
        $final_status = in_array( $final_status, [ 'running', 'paused' ], true ) ? $final_status : 'running';
        $final_status_reason = 'paused' === $final_status
            ? sanitize_key( (string) ( $data['status_reason'] ?? 'not_integrated' ) )
            : '';
        $final_last_error = 'paused' === $final_status ? (string) ( $data['last_error'] ?? '' ) : '';
        $items = $this->normalize_universal_items( $items, true, $workflow );
        if (
            $user_id <= 0
            || ! wp_is_uuid( $run_uuid )
            || ! in_array( $workflow, self::WORKFLOWS, true )
            || null === $request_fingerprint
            || empty( $items )
        ) {
            return $this->admission_result( false, 0, false, false, null, 'invalid_admission' );
        }

        global $wpdb;
		if ( false === ( $this->admission_transaction_query )( 'START TRANSACTION' ) ) {
			return $this->admission_result( false, 0, false, false, null, 'transaction_start_failed' );
		}
		$transaction_started = true;
        $created = false;
        $activated = false;
        $run_id = 0;
        $owned_run = false;
        $failure_reason = 'admission_failed';

        try {
            $run = $this->get_run_by_uuid( $run_uuid, $user_id );
            if ( null === $run ) {
                $data['run_uuid'] = $run_uuid;
                $data['user_id'] = $user_id;
                $data['workflow'] = $workflow;
                $data['request_fingerprint'] = $request_fingerprint;
                $data['status'] = 'admitting';
                $data['status_reason'] = 'admitting';
                $data['started_at'] = null;
                $data['finished_at'] = null;
                $now = current_time( 'mysql', true );
                $payload = array_intersect_key(
                    array_merge(
                        [
                            'created_at'       => $now,
                            'updated_at'       => $now,
                            'filters_snapshot' => null,
                            'selected_count'   => 0,
                            'completed_count'  => 0,
                            'failed_count'     => 0,
                            'last_error'       => null,
                        ],
                        $data
                    ),
                    array_flip( self::RUN_CREATE_COLUMNS )
                );
                $inserted = $wpdb->insert( $this->runs_table, $payload );
                $run_id = false !== $inserted ? (int) $wpdb->insert_id : 0;
                if ( $run_id <= 0 ) {
                    throw new \RuntimeException( 'run_persistence_failed' );
                }
                $created = true;
                $run = $this->get_run_by_id( $run_id, $user_id );
            }

            if ( ! is_array( $run ) || $workflow !== (string) ( $run['workflow'] ?? '' ) ) {
                $failure_reason = 'workflow_conflict';
                throw new \RuntimeException( $failure_reason );
            }
            $stored_fingerprint = $this->normalize_fingerprint( $run['request_fingerprint'] ?? null );
            if ( null === $stored_fingerprint || ! hash_equals( $stored_fingerprint, $request_fingerprint ) ) {
                $failure_reason = 'request_fingerprint_conflict';
                throw new \RuntimeException( $failure_reason );
            }

            $run_id = (int) ( $run['id'] ?? 0 );
            $owned_run = $run_id > 0;
            if ( $run_id <= 0 || (int) ( $run['selected_count'] ?? 0 ) !== count( $items ) ) {
                throw new \RuntimeException( 'item_graph_mismatch' );
            }
            if ( ! $this->ensure_item_graph( $run_id, $user_id, $workflow, $items ) ) {
                throw new \RuntimeException( 'item_persistence_failed' );
            }

            $status = sanitize_key( (string) ( $run['status'] ?? '' ) );
            $status_reason = sanitize_key( (string) ( $run['status_reason'] ?? '' ) );
            $activated = $created
                || 'admitting' === $status
                || in_array( $status_reason, [ 'admitting', 'admission_incomplete', 'item_persistence_failed' ], true );
            if ( $activated ) {
                $started_at = 'running' === $final_status ? trim( (string) ( $data['started_at'] ?? '' ) ) : '';
                if ( 'running' === $final_status && '' === $started_at ) {
                    $started_at = current_time( 'mysql', true );
                }
                if ( ! $this->update_run(
                    $run_id,
                    $user_id,
                    [
                        'status'        => $final_status,
                        'status_reason' => $final_status_reason,
                        'started_at'    => '' !== $started_at ? $started_at : null,
                        'finished_at'   => null,
                        'last_error'    => '' !== $final_last_error ? $final_last_error : null,
                    ],
                    $workflow
                ) ) {
                    throw new \RuntimeException( 'run_activation_failed' );
                }
            }

			if ( $transaction_started && false === ( $this->admission_transaction_query )( 'COMMIT' ) ) {
				throw new \RuntimeException( 'transaction_commit_failed' );
			}
            $run = $this->get_run_by_id( $run_id, $user_id );
            return $this->admission_result( true, $run_id, $created, $activated, $run, '' );
		} catch ( \Throwable $error ) {
			$failure_reason = sanitize_key( $error->getMessage() );
			if ( $transaction_started ) {
				( $this->admission_transaction_query )( 'ROLLBACK' );
			}
            if ( $owned_run && ! in_array( $failure_reason, [ 'workflow_conflict', 'request_fingerprint_conflict' ], true ) ) {
                $persisted = $this->get_run_by_id( $run_id, $user_id );
                if ( is_array( $persisted ) && $workflow === (string) ( $persisted['workflow'] ?? '' ) ) {
                    $this->update_run(
                        $run_id,
                        $user_id,
                        [
                            'status'        => 'paused',
                            'status_reason' => 'admission_incomplete',
                            'last_error'    => 'Bulk item graph admission is incomplete.',
                        ],
                        $workflow
                    );
                }
            }
            return $this->admission_result( false, $run_id, $created, false, null, $failure_reason );
        }
    }

    public function run_has_exact_graph( int $run_id, int $user_id, string $workflow, array $items ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        $items = $this->normalize_universal_items( $items, true, sanitize_key( $workflow ) );
        return ! empty( $items ) && $this->item_graph_matches( $run_id, $user_id, $workflow, $items );
    }

    /**
     * Persist a Campaign Set run and its complete product graph as one recoverable admission.
     *
     * The transaction is the fast path on InnoDB. The explicit `admitting` state and
     * idempotent graph verification are the crash-recovery path for interrupted or
     * non-transactional installs.
     *
     * @return array{ok: bool, run_id: int, created: bool, activated: bool, run: array|null, reason: string}
     */

    public function update_run( int $run_id, int $user_id, array $data, string $workflow = '' ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        if ( $run_id <= 0 || $user_id <= 0 || empty( $data ) ) {
            return false;
        }
        $run = $this->get_run_by_id( $run_id, $user_id );
        if ( null === $run ) {
            return false;
        }
        global $wpdb;
        unset( $data['id'], $data['user_id'], $data['run_uuid'] );
        if ( ! empty( array_diff( array_keys( $data ), self::RUN_MUTABLE_COLUMNS ) ) ) {
            return false;
        }
        if ( isset( $data['workflow'] ) ) {
            $data['workflow'] = $this->normalize_workflow( (string) $data['workflow'] );
        }
        $data['updated_at'] = current_time( 'mysql', true );
        $where = [
            'id'      => $run_id,
            'user_id' => $user_id,
        ];
        if ( '' !== $workflow ) {
            $workflow = sanitize_key( $workflow );
            if ( ! in_array( $workflow, self::WORKFLOWS, true ) ) {
                return false;
            }
            if ( $workflow !== (string) ( $run['workflow'] ?? '' ) ) {
                return false;
            }
            $where['workflow'] = $workflow;
        }
        $updated = $wpdb->update(
            $this->runs_table,
            $data,
            $where
        );
        if ( is_int( $updated ) && $updated > 0 ) {
            $this->fire_run_state_changed( $run, $data );
        }
        return false !== $updated;
    }

    /**
     * Atomically update a run only while it remains in the expected state.
     */
    public function update_run_if_status( int $run_id, int $user_id, string $expected_status, array $data, string $workflow = '' ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        $expected_status = sanitize_key( $expected_status );
        if ( $run_id <= 0 || $user_id <= 0 || '' === $expected_status || empty( $data ) ) {
            return false;
        }
        $run = $this->get_run_by_id( $run_id, $user_id );
        if ( null === $run ) {
            return false;
        }
        $workflow = sanitize_key( $workflow );
        if ( '' !== $workflow && ( ! in_array( $workflow, self::WORKFLOWS, true ) || $workflow !== (string) ( $run['workflow'] ?? '' ) ) ) {
            return false;
        }

        global $wpdb;
        unset( $data['id'], $data['user_id'], $data['run_uuid'], $data['workflow'] );
        if ( ! empty( array_diff( array_keys( $data ), self::RUN_MUTABLE_COLUMNS ) ) ) {
            return false;
        }
        $data['updated_at'] = current_time( 'mysql', true );
        $where = [
            'id'      => $run_id,
            'user_id' => $user_id,
            'status'  => $expected_status,
        ];
        if ( '' !== $workflow ) {
            $where['workflow'] = $workflow;
        }
        $updated = $wpdb->update( $this->runs_table, $data, $where );
        if ( 1 === $updated ) {
            $this->fire_run_state_changed( $run, $data );
            return true;
        }
        return false;
    }

    /**
     * Emit a bounded transition event only after a durable status change.
     */
    private function fire_run_state_changed( array $run, array $data ): void {
        if ( ! function_exists( 'do_action' ) || ! array_key_exists( 'status', $data ) ) {
            return;
        }
        $run_uuid = strtolower( trim( (string) ( $run['run_uuid'] ?? '' ) ) );
        $previous_status = sanitize_key( (string) ( $run['status'] ?? '' ) );
        $status = sanitize_key( (string) $data['status'] );
        if ( ! wp_is_uuid( $run_uuid ) || '' === $status || $status === $previous_status ) {
            return;
        }
        $workflow = sanitize_key( (string) ( $data['workflow'] ?? $run['workflow'] ?? '' ) );
        if ( ! in_array( $workflow, self::WORKFLOWS, true ) ) {
            return;
        }

        /**
         * Fires after a Bulk run status transition is durably committed.
         *
         * @param array $event Safe immutable run-state metadata.
         */
        do_action(
            'fotogenic_ai_bulk_run_state_changed',
            [
                'runUuid'        => $run_uuid,
                'workflow'       => $workflow,
                'previousStatus' => $previous_status,
                'status'         => $status,
            ]
        );
    }

    public function get_run_by_uuid( string $run_uuid, int $user_id ): ?array {
        if ( ! $this->ensure_tables() ) {
            return null;
        }
        $run_uuid = strtolower( trim( $run_uuid ) );
        if ( '' === $run_uuid || $user_id <= 0 ) {
            return null;
        }
        global $wpdb;
        $row = $wpdb->get_row(
            $wpdb->prepare(
                "SELECT * FROM {$this->runs_table} WHERE run_uuid = %s AND user_id = %d LIMIT 1",
                $run_uuid,
                $user_id
            ),
            ARRAY_A
        );
        return is_array( $row ) ? $row : null;
    }

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


    public function update_item( int $item_id, int $user_id, array $data, string $workflow = '' ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        if ( $item_id <= 0 || $user_id <= 0 || empty( $data ) ) {
            return false;
        }
        $item = $this->get_item_by_id( $item_id, $user_id );
        if ( null === $item ) {
            return false;
        }
        if ( '' !== $workflow ) {
            $workflow = sanitize_key( $workflow );
            if ( ! in_array( $workflow, self::WORKFLOWS, true ) || $workflow !== (string) ( $item['workflow'] ?? '' ) ) {
                return false;
            }
        }
        global $wpdb;
        unset( $data['id'], $data['run_id'], $data['run_item_uuid'], $data['campaign_group_uuid'], $data['item_ordinal'], $data['subject_type'], $data['subject_id'], $data['target_type'], $data['parent_product_id'], $data['variation_id'], $data['target_snapshot'], $data['source_roles_snapshot'], $data['destination_snapshot'], $data['request_fingerprint'], $data['command_snapshot'] );
        if ( ! empty( $item['command_snapshot'] ) ) {
            unset( $data['request_id'], $data['generation_key'] );
        }
        $data['updated_at'] = current_time( 'mysql', true );
        return false !== $wpdb->update(
            $this->items_table,
            $data,
            [
                'id'     => $item_id,
                'run_id' => (int) $item['run_id'],
            ]
        );
    }

    /**
     * Atomically update an item only while it remains in the expected state.
     */
    public function update_item_if_status( int $item_id, int $user_id, string $expected_status, array $data, string $workflow = '' ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        $expected_status = sanitize_key( $expected_status );
        if ( $item_id <= 0 || $user_id <= 0 || '' === $expected_status || empty( $data ) ) {
            return false;
        }
        $item = $this->get_item_by_id( $item_id, $user_id );
        if ( null === $item ) {
            return false;
        }
        $workflow = sanitize_key( $workflow );
        if ( '' !== $workflow && ( ! in_array( $workflow, self::WORKFLOWS, true ) || $workflow !== (string) ( $item['workflow'] ?? '' ) ) ) {
            return false;
        }

        global $wpdb;
        unset( $data['id'], $data['run_id'], $data['run_item_uuid'], $data['campaign_group_uuid'], $data['item_ordinal'], $data['subject_type'], $data['subject_id'], $data['target_type'], $data['parent_product_id'], $data['variation_id'], $data['target_snapshot'], $data['source_roles_snapshot'], $data['destination_snapshot'], $data['request_fingerprint'], $data['command_snapshot'], $data['workflow'] );
        if ( ! empty( $item['command_snapshot'] ) ) {
            unset( $data['request_id'], $data['generation_key'] );
        }
        $data['updated_at'] = current_time( 'mysql', true );
        return 1 === $wpdb->update(
            $this->items_table,
            $data,
            [
                'id'     => $item_id,
                'run_id' => (int) $item['run_id'],
                'status' => $expected_status,
            ]
        );
    }

    /**
     * Atomically persist one bulk item's durable identity before its first billable request.
     */
    public function admit_item_start( int $item_id, int $user_id, string $workflow, string $request_id, string $started_at, string $claim_owner = '' ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        $workflow = sanitize_key( $workflow );
        $request_id = strtolower( trim( $request_id ) );
        $started_at = trim( $started_at );
        $claim_owner = substr( trim( $claim_owner ), 0, 64 );
        if (
            $item_id <= 0
            || $user_id <= 0
            || ! in_array( $workflow, self::WORKFLOWS, true )
            || ! wp_is_uuid( $request_id )
            || '' === $started_at
        ) {
            return false;
        }

        global $wpdb;
        $updated_at = current_time( 'mysql', true );
        // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Both table names are plugin-owned; all values are prepared below.
        $claim_sql = '' !== $claim_owner
            ? $wpdb->prepare( ' AND i.claim_owner = %s AND i.claim_expires_at >= %s', $claim_owner, $updated_at )
            : '';
        $sql = "UPDATE {$this->items_table} i
            INNER JOIN {$this->runs_table} r ON r.id = i.run_id
            SET i.request_id = %s, i.started_at = %s, i.updated_at = %s
            WHERE i.id = %d AND r.user_id = %d AND r.workflow = %s
                AND r.status = 'running' AND i.status = 'starting'
				AND (i.request_id IS NULL OR i.request_id = '' OR i.request_id = %s)
				AND i.started_at IS NULL{$claim_sql}";
        // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching -- This compare-and-update is the bulk billing admission boundary.
        return 1 === $wpdb->query( $wpdb->prepare( $sql, $request_id, $started_at, $updated_at, $item_id, $user_id, $workflow, $request_id ) );
    }

    /**
     * Delete one bounded page of old terminal Product Images and Campaign Set runs.
     *
     * Only plugin control-plane rows are removed. Media Library attachments are
     * intentionally preserved and remain independently owned WordPress posts.
     *
     * @return array{processed: int, deleted: int, next_cursor: int}
     */
    public function delete_expired_terminal_runs( int $cutoff_timestamp, int $limit, int $after_run_id = 0 ): array {
        if ( ! $this->ensure_tables() ) {
            return [
                'processed'   => 0,
                'deleted'     => 0,
                'next_cursor' => 0,
            ];
        }
        $limit = max( 1, min( self::HOUSEKEEPING_MAX_BATCH_SIZE, $limit ) );
        $after_run_id = max( 0, $after_run_id );
        if ( $cutoff_timestamp <= 0 ) {
            return [
                'processed'   => 0,
                'deleted'     => 0,
                'next_cursor' => 0,
            ];
        }

        global $wpdb;
        $cutoff = gmdate( 'Y-m-d H:i:s', $cutoff_timestamp );
        $terminal_placeholders = implode( ',', array_fill( 0, count( self::TERMINAL_RUN_STATUSES ), '%s' ) );
        $terminal_item_placeholders = implode( ',', array_fill( 0, count( self::TERMINAL_ITEM_STATUSES ), '%s' ) );
        $workflow_placeholders = implode( ',', array_fill( 0, count( self::WORKFLOWS ), '%s' ) );
        $params = array_merge(
            [ $after_run_id ],
            self::WORKFLOWS,
            self::TERMINAL_RUN_STATUSES,
            [ $cutoff ],
            self::TERMINAL_ITEM_STATUSES,
            [ $limit ]
        );
        $sql = "SELECT r.id, r.user_id, r.workflow
            FROM {$this->runs_table} r
            WHERE r.id > %d
              AND r.workflow IN ({$workflow_placeholders})
              AND r.status IN ({$terminal_placeholders})
              AND GREATEST(r.updated_at, COALESCE(NULLIF(r.finished_at, '0000-00-00 00:00:00'), r.created_at)) < %s
              AND NOT EXISTS (
                  SELECT 1 FROM {$this->items_table} nonterminal
                  WHERE nonterminal.run_id = r.id
                    AND nonterminal.status NOT IN ({$terminal_item_placeholders})
              )
            ORDER BY r.id ASC
            LIMIT %d";
        // The cursor and limit bound this plugin-table scan; each candidate is owner-revalidated under a row lock before deletion.
        $rows = $wpdb->get_results( $wpdb->prepare( $sql, ...$params ), ARRAY_A );
        $rows = is_array( $rows ) ? $rows : [];
        $deleted = 0;
        $last_run_id = 0;
        foreach ( $rows as $row ) {
            $run_id = absint( $row['id'] ?? 0 );
            $last_run_id = max( $last_run_id, $run_id );
            if ( $this->delete_terminal_run_if_eligible( $row, $cutoff ) ) {
                ++$deleted;
            }
        }

        return [
            'processed'   => count( $rows ),
            'deleted'     => $deleted,
            'next_cursor' => count( $rows ) < $limit ? 0 : $last_run_id,
        ];
    }

    /**
     * Clear references after a generated attachment is deleted.
     *
     * This is a system cleanup operation, so it discovers each row owner before
     * performing the owner-scoped item mutation.
     */
    public function clear_attachment_references( int $attachment_id ): int {
        if ( ! $this->ensure_tables() ) {
            return 0;
        }
        $attachment_id = absint( $attachment_id );
        if ( $attachment_id <= 0 ) {
            return 0;
        }
        global $wpdb;
        $updated = 0;
        $last_id = 0;
        $attachment_like = '%' . $wpdb->esc_like( (string) $attachment_id ) . '%';
        do {
            $rows = (array) $wpdb->get_results(
                $wpdb->prepare(
                    "SELECT i.id, i.attachment_id, i.attachment_ids, i.meta, r.user_id, r.workflow
                    FROM {$this->items_table} i
                    INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                    WHERE i.id > %d AND (i.attachment_id = %d OR i.attachment_ids LIKE %s)
                    ORDER BY i.id ASC
                    LIMIT %d",
                    $last_id,
                    $attachment_id,
                    $attachment_like,
                    self::ATTACHMENT_REFERENCE_BATCH_SIZE
                ),
                ARRAY_A
            );
            foreach ( $rows as $row ) {
                $item_id = (int) ( $row['id'] ?? 0 );
                $last_id = max( $last_id, $item_id );
                $user_id = (int) ( $row['user_id'] ?? 0 );
                $workflow = sanitize_key( (string) ( $row['workflow'] ?? '' ) );
                if ( $item_id <= 0 || $user_id <= 0 || ! in_array( $workflow, self::WORKFLOWS, true ) ) {
                    continue;
                }
                $attachment_ids = $this->decode_id_list( $row['attachment_ids'] ?? [] );
                $contains_attachment = (int) ( $row['attachment_id'] ?? 0 ) === $attachment_id
                    || in_array( $attachment_id, $attachment_ids, true );
                if ( ! $contains_attachment ) {
                    continue;
                }
                $attachment_ids = array_values( array_diff( $attachment_ids, [ $attachment_id ] ) );
                $meta = $this->decode_json_object( $row['meta'] ?? '' );
                unset( $meta['attachment_url'] );
                $payload = [
                    'attachment_id'  => (int) ( $row['attachment_id'] ?? 0 ) === $attachment_id
                        ? ( ! empty( $attachment_ids ) ? (int) $attachment_ids[0] : null )
                        : $row['attachment_id'],
                    'attachment_ids' => ! empty( $attachment_ids ) ? wp_json_encode( $attachment_ids ) : null,
                    'meta'           => ! empty( $meta ) ? wp_json_encode( $meta ) : null,
                ];
                if ( $this->update_item( $item_id, $user_id, $payload, $workflow ) ) {
                    ++$updated;
                }
            }
        } while ( count( $rows ) === self::ATTACHMENT_REFERENCE_BATCH_SIZE );
        return $updated;
    }

    /**
     * Serialize all Bulk start admission for one owner.
     *
     * This lock is deliberately short-lived and must always be released by the
     * caller. A zero timeout makes overlapping requests fail closed immediately.
     */
    public function acquire_admission_lock( int $user_id ): bool {
        if ( $user_id <= 0 ) {
            return false;
        }
        global $wpdb;
        $acquired = $wpdb->get_var(
            $wpdb->prepare( 'SELECT GET_LOCK(%s, %d)', $this->admission_lock_name( $user_id ), 0 )
        );
        return '1' === (string) $acquired;
    }

    public function release_admission_lock( int $user_id ): void {
        if ( $user_id <= 0 ) {
            return;
        }
        global $wpdb;
        $wpdb->get_var(
            $wpdb->prepare( 'SELECT RELEASE_LOCK(%s)', $this->admission_lock_name( $user_id ) )
        );
    }

    public function acquire_run_lock( int $run_id, int $user_id, string $owner, int $ttl_seconds ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        $owner = substr( trim( $owner ), 0, 64 );
        if ( $run_id <= 0 || $user_id <= 0 || '' === $owner ) {
            return false;
        }
        global $wpdb;
        $now = current_time( 'mysql', true );
        $expires = gmdate( 'Y-m-d H:i:s', time() + max( 10, $ttl_seconds ) );
        $sql = "UPDATE {$this->runs_table}
            SET lock_owner = %s, lock_expires_at = %s, updated_at = %s
            WHERE id = %d AND user_id = %d
              AND (lock_expires_at IS NULL OR lock_expires_at < %s OR lock_owner = '')";
        $updated = $wpdb->query( $wpdb->prepare( $sql, $owner, $expires, $now, $run_id, $user_id, $now ) );
        return is_numeric( $updated ) && $updated > 0;
    }

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

    public function get_generating_item( int $run_id, int $user_id ): ?array {
        if ( ! $this->ensure_tables() ) {
            return null;
        }
        if ( $run_id <= 0 || $user_id <= 0 ) {
            return null;
        }
        global $wpdb;
        $row = $wpdb->get_row(
            $wpdb->prepare(
                "SELECT i.* FROM {$this->items_table} i
                INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                WHERE i.run_id = %d AND r.user_id = %d
                  AND i.status IN ('starting','generating','cancel_requested')
                ORDER BY i.id ASC LIMIT 1",
                $run_id,
                $user_id
            ),
            ARRAY_A
        );
        return is_array( $row ) ? $this->normalize_item_row( $row ) : null;
    }

    public function get_item_by_id( int $item_id, int $user_id ): ?array {
        if ( ! $this->ensure_tables() ) {
            return null;
        }
        if ( $item_id <= 0 || $user_id <= 0 ) {
            return null;
        }
        global $wpdb;
        $row = $wpdb->get_row(
            $wpdb->prepare(
                "SELECT i.*, r.workflow AS workflow FROM {$this->items_table} i
                INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                WHERE i.id = %d AND r.user_id = %d LIMIT 1",
                $item_id,
                $user_id
            ),
            ARRAY_A
        );
        return is_array( $row ) ? $this->normalize_item_row( $row ) : null;
    }

    public function list_items_for_progress( int $run_id, int $user_id, array $statuses, int $limit = 50 ): array {
        if ( ! $this->ensure_tables() ) {
            return [];
        }
        $limit = max( 1, min( 100, $limit ) );
        $statuses = $this->normalize_statuses( $statuses );
        if ( $run_id <= 0 || $user_id <= 0 || empty( $statuses ) ) {
            return [];
        }
        global $wpdb;
        $placeholders = implode( ',', array_fill( 0, count( $statuses ), '%s' ) );
        $params = array_merge( [ $run_id, $user_id ], $statuses );
        $sql = "SELECT i.id, i.run_item_uuid, i.campaign_group_uuid, i.item_ordinal, i.target_snapshot,
                i.subject_type, i.subject_id, i.status, i.attachment_id, i.attachment_ids,
                i.error_code, i.error_message, i.updated_at, i.started_at, i.finished_at, i.meta,
                i.command_snapshot
            FROM {$this->items_table} i
            INNER JOIN {$this->runs_table} r ON r.id = i.run_id
            WHERE i.run_id = %d AND r.user_id = %d AND i.status IN ($placeholders)
            ORDER BY i.id DESC LIMIT %d";
        $params[] = $limit;
        return array_map( [ $this, 'normalize_item_row' ], (array) $wpdb->get_results( $wpdb->prepare( $sql, ...$params ), ARRAY_A ) );
    }

    public function list_items_by_statuses( int $run_id, int $user_id, array $statuses, int $limit = 50, string $order_by = 'id', string $order_dir = 'ASC' ): array {
        if ( ! $this->ensure_tables() ) {
            return [];
        }
        $limit = max( 1, min( 100, $limit ) );
        $statuses = $this->normalize_statuses( $statuses );
        if ( $run_id <= 0 || $user_id <= 0 || empty( $statuses ) ) {
            return [];
        }
        global $wpdb;
        $placeholders = implode( ',', array_fill( 0, count( $statuses ), '%s' ) );
        $order_by = in_array( $order_by, [ 'id', 'last_polled_at', 'started_at', 'updated_at' ], true ) ? $order_by : 'id';
        $order_dir = 'DESC' === strtoupper( $order_dir ) ? 'DESC' : 'ASC';
        $params = array_merge( [ $run_id, $user_id ], $statuses );
        $sql = "SELECT i.* FROM {$this->items_table} i
            INNER JOIN {$this->runs_table} r ON r.id = i.run_id
            WHERE i.run_id = %d AND r.user_id = %d AND i.status IN ($placeholders)
            ORDER BY i.{$order_by} {$order_dir} LIMIT %d";
        $params[] = $limit;
        return array_map( [ $this, 'normalize_item_row' ], (array) $wpdb->get_results( $wpdb->prepare( $sql, ...$params ), ARRAY_A ) );
    }

    public function claim_pending_items( int $run_id, int $user_id, int $limit = 1, string $owner = '', int $ttl_seconds = 120 ): array {
        if ( ! $this->ensure_tables() ) {
            return [];
        }
        $limit = max( 1, min( 50, $limit ) );
        $owner = substr( trim( $owner ), 0, 64 );
        if ( '' === $owner ) {
            $owner = 'item-' . wp_generate_uuid4();
        }
        if ( $run_id <= 0 || $user_id <= 0 || '' === $owner ) {
            return [];
        }
        global $wpdb;
        $ids = (array) $wpdb->get_col(
            $wpdb->prepare(
                "SELECT i.id FROM {$this->items_table} i
                INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                WHERE i.run_id = %d AND r.user_id = %d AND r.status = 'running' AND i.status = 'pending'
                ORDER BY i.id ASC LIMIT %d",
                $run_id,
                $user_id,
                $limit
            )
        );
        $ids = array_values( array_unique( array_filter( array_map( 'absint', $ids ) ) ) );
        if ( empty( $ids ) ) {
            return [];
        }
        $now = current_time( 'mysql', true );
        $expires = gmdate( 'Y-m-d H:i:s', time() + max( 30, $ttl_seconds ) );
        $placeholders = implode( ',', array_fill( 0, count( $ids ), '%d' ) );
        $params = array_merge( [ $owner, $expires, $now, $run_id ], $ids );
        $sql = "UPDATE {$this->items_table} i
            INNER JOIN {$this->runs_table} r ON r.id = i.run_id
            SET i.status = 'starting', i.attempts = i.attempts + 1, i.poll_count = 0,
                i.generation_started_at = NULL, i.started_at = NULL,
                i.last_polled_at = NULL, i.claim_owner = %s, i.claim_expires_at = %s, i.updated_at = %s
            WHERE i.run_id = %d AND r.user_id = %d AND r.status = 'running'
              AND i.status = 'pending' AND i.id IN ({$placeholders})";
        array_splice( $params, 4, 0, [ $user_id ] );
        $wpdb->query( $wpdb->prepare( $sql, ...$params ) );
        $claimed_params = array_merge( [ $run_id, $user_id, $owner ], $ids );
        $claimed_sql = "SELECT i.* FROM {$this->items_table} i
            INNER JOIN {$this->runs_table} r ON r.id = i.run_id
            WHERE i.run_id = %d AND r.user_id = %d AND i.status = 'starting'
              AND i.claim_owner = %s AND i.id IN ({$placeholders})
            ORDER BY i.id ASC";
        return array_map( [ $this, 'normalize_item_row' ], (array) $wpdb->get_results( $wpdb->prepare( $claimed_sql, ...$claimed_params ), ARRAY_A ) );
    }

    /**
     * Acquire or renew the lease for a persisted starting item.
     */
    public function claim_starting_item( int $item_id, int $user_id, string $workflow, string $owner, int $ttl_seconds = 120 ): ?array {
        if ( ! $this->ensure_tables() ) {
            return null;
        }
        $workflow = sanitize_key( $workflow );
        $owner = substr( trim( $owner ), 0, 64 );
        if ( $item_id <= 0 || $user_id <= 0 || ! in_array( $workflow, self::WORKFLOWS, true ) || '' === $owner ) {
            return null;
        }

        global $wpdb;
        $now = current_time( 'mysql', true );
        $expires = gmdate( 'Y-m-d H:i:s', time() + max( 30, $ttl_seconds ) );
        $updated = $wpdb->query(
            $wpdb->prepare(
                "UPDATE {$this->items_table} i
                INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                SET i.claim_owner = %s, i.claim_expires_at = %s, i.updated_at = %s
                WHERE i.id = %d AND r.user_id = %d AND r.workflow = %s
                  AND r.status = 'running' AND i.status = 'starting'
                  AND (i.claim_owner = %s OR i.claim_owner IS NULL OR i.claim_owner = ''
                       OR i.claim_expires_at IS NULL OR i.claim_expires_at < %s)",
                $owner,
                $expires,
                $now,
                $item_id,
                $user_id,
                $workflow,
                $owner,
                $now
            )
        );
        if ( ! is_numeric( $updated ) || $updated <= 0 ) {
            return null;
        }
        $item = $this->get_item_by_id( $item_id, $user_id );
        return is_array( $item ) && hash_equals( $owner, (string) ( $item['claim_owner'] ?? '' ) ) ? $item : null;
    }

    /**
     * Compare-and-swap a leased item and clear the lease after start dispatch.
     */
    public function update_claimed_item_if_status(
        int $item_id,
        int $user_id,
        string $workflow,
        string $expected_status,
        string $owner,
        array $data
    ): bool {
        if ( ! $this->ensure_tables() ) {
            return false;
        }
        $workflow = sanitize_key( $workflow );
        $expected_status = sanitize_key( $expected_status );
        $owner = substr( trim( $owner ), 0, 64 );
        if (
            $item_id <= 0
            || $user_id <= 0
            || ! in_array( $workflow, self::WORKFLOWS, true )
            || '' === $expected_status
            || '' === $owner
            || empty( $data )
        ) {
            return false;
        }

        global $wpdb;
        unset( $data['id'], $data['run_id'], $data['run_item_uuid'], $data['campaign_group_uuid'], $data['item_ordinal'], $data['subject_type'], $data['subject_id'], $data['target_type'], $data['parent_product_id'], $data['variation_id'], $data['target_snapshot'], $data['source_roles_snapshot'], $data['destination_snapshot'], $data['request_id'], $data['request_fingerprint'], $data['command_snapshot'], $data['generation_key'], $data['workflow'], $data['claim_owner'], $data['claim_expires_at'] );
        $data['updated_at'] = current_time( 'mysql', true );
        if ( 'starting' !== (string) ( $data['status'] ?? $expected_status ) ) {
            $data['claim_owner'] = null;
            $data['claim_expires_at'] = null;
        }
        $set = [];
        $values = [];
        foreach ( $data as $column => $value ) {
            $column = sanitize_key( (string) $column );
            if ( '' === $column || ! in_array( $column, self::MUTABLE_ITEM_COLUMNS, true ) ) {
                return false;
            }
            $set[] = "i.{$column} = " . ( null === $value ? 'NULL' : '%s' );
            if ( null !== $value ) {
                $values[] = is_scalar( $value ) ? (string) $value : '';
            }
        }
        $values = array_merge( $values, [ $item_id, $user_id, $workflow, $expected_status, $owner, current_time( 'mysql', true ) ] );
        $sql = "UPDATE {$this->items_table} i
            INNER JOIN {$this->runs_table} r ON r.id = i.run_id
            SET " . implode( ', ', $set ) . "
            WHERE i.id = %d AND r.user_id = %d AND r.workflow = %s
              AND i.status = %s AND i.claim_owner = %s AND i.claim_expires_at >= %s";
        return 1 === $wpdb->query( $wpdb->prepare( $sql, ...$values ) );
    }

    public function count_items_by_status( int $run_id, int $user_id ): array {
        if ( ! $this->ensure_tables() ) {
            return [];
        }
        if ( $run_id <= 0 || $user_id <= 0 ) {
            return [];
        }
        global $wpdb;
        $rows = $wpdb->get_results(
            $wpdb->prepare(
                "SELECT i.status, COUNT(*) AS total FROM {$this->items_table} i
                INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                WHERE i.run_id = %d AND r.user_id = %d GROUP BY i.status",
                $run_id,
                $user_id
            ),
            ARRAY_A
        );
        $counts = [];
        foreach ( $rows as $row ) {
            $status = (string) ( $row['status'] ?? '' );
            if ( '' !== $status ) {
                $counts[ $status ] = (int) ( $row['total'] ?? 0 );
            }
        }
        return $counts;
    }

    private function normalize_workflow( string $workflow ): string {
        $workflow = sanitize_key( $workflow );
        return in_array( $workflow, self::WORKFLOWS, true ) ? $workflow : 'product_images';
    }

    /**
     * Revalidate and remove one terminal run while its owner row and item graph are locked.
     */
    private function delete_terminal_run_if_eligible( array $candidate, string $cutoff ): bool {
        $run_id = absint( $candidate['id'] ?? 0 );
        $user_id = absint( $candidate['user_id'] ?? 0 );
        $workflow = sanitize_key( (string) ( $candidate['workflow'] ?? '' ) );
        if ( $run_id <= 0 || $user_id <= 0 || ! in_array( $workflow, self::WORKFLOWS, true ) ) {
            return false;
        }

        global $wpdb;
        // A per-run transaction keeps the item graph and owner row deletion together on transactional tables.
        if ( false === $wpdb->query( 'START TRANSACTION' ) ) {
            return false;
        }

        try {
            $run = $wpdb->get_row(
                $wpdb->prepare(
                    "SELECT id, user_id, workflow, status,
                        GREATEST(updated_at, COALESCE(NULLIF(finished_at, '0000-00-00 00:00:00'), created_at)) AS retention_at
                    FROM {$this->runs_table}
                    WHERE id = %d AND user_id = %d AND workflow = %s
                    LIMIT 1 FOR UPDATE",
                    $run_id,
                    $user_id,
                    $workflow
                ),
                ARRAY_A
            );
            if (
                ! is_array( $run )
                || ! in_array( (string) ( $run['status'] ?? '' ), self::TERMINAL_RUN_STATUSES, true )
                || strcmp( (string) ( $run['retention_at'] ?? '' ), $cutoff ) >= 0
            ) {
                $wpdb->query( 'ROLLBACK' );
                return false;
            }

            $item_statuses = $wpdb->get_col(
                $wpdb->prepare(
                    "SELECT status FROM {$this->items_table} WHERE run_id = %d FOR UPDATE",
                    $run_id
                )
            );
            if ( array_diff( array_map( 'strval', (array) $item_statuses ), self::TERMINAL_ITEM_STATUSES ) ) {
                $wpdb->query( 'ROLLBACK' );
                return false;
            }

            // Owner/workflow predicates prevent a housekeeping candidate from crossing tenant or workflow boundaries.
            $items_deleted = $wpdb->query(
                $wpdb->prepare(
                    "DELETE i FROM {$this->items_table} i
                    INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                    WHERE i.run_id = %d AND r.id = %d AND r.user_id = %d AND r.workflow = %s",
                    $run_id,
                    $run_id,
                    $user_id,
                    $workflow
                )
            );
            if ( false === $items_deleted ) {
                $wpdb->query( 'ROLLBACK' );
                return false;
            }

            $terminal_placeholders = implode( ',', array_fill( 0, count( self::TERMINAL_RUN_STATUSES ), '%s' ) );
            $params = array_merge(
                [ $run_id, $user_id, $workflow ],
                self::TERMINAL_RUN_STATUSES,
                [ $cutoff ]
            );
            $deleted = $wpdb->query(
                // phpcs:ignore WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber -- Fixed internal terminal statuses expand into prepared placeholders at runtime.
                $wpdb->prepare(
                    "DELETE FROM {$this->runs_table}
                    WHERE id = %d AND user_id = %d AND workflow = %s
                      AND status IN ({$terminal_placeholders})
                      AND GREATEST(updated_at, COALESCE(NULLIF(finished_at, '0000-00-00 00:00:00'), created_at)) < %s",
                    ...$params
                )
            );
            if ( 1 !== (int) $deleted || false === $wpdb->query( 'COMMIT' ) ) {
                $wpdb->query( 'ROLLBACK' );
                return false;
            }
            return true;
        } catch ( \Throwable $error ) {
            $wpdb->query( 'ROLLBACK' );
            if ( Logger::instance()->enabled() ) {
                Logger::instance()->warn( 'bulk_retention_delete_failed', [
                    'run_id'        => 'bulk-retention:' . $run_id,
                    'bulk_run_id'   => $run_id,
                    'owner_user_id' => $user_id,
                    'workflow'      => $workflow,
                    'error_message' => $error->getMessage(),
                ] );
            }
            return false;
        }
    }

    /**
     * Normalize a newly admitted immutable item graph.
     *
     * @return array<int, array>|array{}
     */
    private function normalize_universal_items( array $items, bool $require_command, string $workflow = '' ): array {
        $normalized = [];
        $run_item_uuids = [];
        $campaign_group_uuids = [];
        $ordinals = [];
        $generation_keys = [];
        $request_ids = [];
        foreach ( $items as $item ) {
            if ( ! is_array( $item ) || fotogenic_ai_array_is_list( $item ) ) {
                return [];
            }
            $raw_subject_type = sanitize_key( (string) ( $item['subject_type'] ?? '' ) );
            if ( ! in_array( $raw_subject_type, self::SUBJECT_TYPES, true ) ) {
                return [];
            }
            $subject_type = $raw_subject_type;
            $subject_id = absint( $item['subject_id'] ?? 0 );
            $generation_key = strtolower( trim( (string) ( $item['generation_key'] ?? '' ) ) );
            $request_id = strtolower( trim( (string) ( $item['request_id'] ?? '' ) ) );
            $request_fingerprint = $this->normalize_fingerprint( $item['request_fingerprint'] ?? null );
            $command_snapshot = $this->normalize_command_snapshot( $item['command_snapshot'] ?? null );
            $run_item_uuid = $this->normalize_uuid( $item['run_item_uuid'] ?? '' );
            $campaign_group_uuid = $this->normalize_uuid( $item['campaign_group_uuid'] ?? '' );
            $item_ordinal = (int) ( $item['item_ordinal'] ?? 0 );
            $target_snapshot = $this->normalize_target_snapshot( $item['target_snapshot'] ?? null );
            $source_roles_snapshot = $this->normalize_json_snapshot( $item['source_roles_snapshot'] ?? null );
            $destination_snapshot = $this->normalize_json_snapshot( $item['destination_snapshot'] ?? null );
            if (
                $subject_id <= 0
                || null === $run_item_uuid
                || $item_ordinal <= 0
                || null === $target_snapshot
                || $subject_type !== (string) $target_snapshot['sourceType']
                || $subject_id !== (int) $target_snapshot['targetId']
                || null === $source_roles_snapshot
                || null === $destination_snapshot
                || ( 'campaign_set' === $workflow && null === $campaign_group_uuid )
                || ( 'campaign_set' !== $workflow && null !== $campaign_group_uuid )
                || 1 !== preg_match( '/^[a-f0-9]{40}$/', $generation_key )
                || ! wp_is_uuid( $request_id )
                || null === $request_fingerprint
                || ( $require_command && null === $command_snapshot )
                || isset( $run_item_uuids[ $run_item_uuid ] )
                || isset( $ordinals[ $item_ordinal ] )
                || ( null !== $campaign_group_uuid && isset( $campaign_group_uuids[ $campaign_group_uuid ] ) )
                || isset( $generation_keys[ $generation_key ] )
                || isset( $request_ids[ $request_id ] )
            ) {
                return [];
            }
            $run_item_uuids[ $run_item_uuid ] = true;
            $ordinals[ $item_ordinal ] = true;
            if ( null !== $campaign_group_uuid ) {
                $campaign_group_uuids[ $campaign_group_uuid ] = true;
            }
            $generation_keys[ $generation_key ] = true;
            $request_ids[ $request_id ] = true;
            $normalized[] = [
                'run_item_uuid'       => $run_item_uuid,
                'campaign_group_uuid' => $campaign_group_uuid,
                'item_ordinal'        => $item_ordinal,
                'subject_type'        => $subject_type,
                'subject_id'          => $subject_id,
                'target_type'         => (string) $target_snapshot['targetType'],
                'parent_product_id'   => (int) $target_snapshot['parentProductId'],
                'variation_id'        => (int) $target_snapshot['variationId'],
                'target_snapshot'     => wp_json_encode( $target_snapshot, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE ),
                'source_roles_snapshot'=> $source_roles_snapshot,
                'destination_snapshot'=> $destination_snapshot,
                'generation_key'      => $generation_key,
                'request_id'          => $request_id,
                'request_fingerprint' => $request_fingerprint,
                'command_snapshot'    => $command_snapshot,
            ];
        }
        return $normalized;
    }

    private function ensure_item_graph( int $run_id, int $user_id, string $workflow, array $items ): bool {
        global $wpdb;
        $existing = $this->get_item_graph( $run_id, $user_id, $workflow );
        $by_run_item = [];
        foreach ( $existing as $row ) {
            $row = $this->normalize_item_row( $row );
            $by_run_item[ (string) $row['run_item_uuid'] ] = $row;
        }

        foreach ( $items as $item ) {
            $run_item_uuid = (string) $item['run_item_uuid'];
            if ( isset( $by_run_item[ $run_item_uuid ] ) ) {
                if ( ! $this->immutable_item_matches( $by_run_item[ $run_item_uuid ], $item ) ) {
                    return false;
                }
                continue;
            }
            $inserted = $wpdb->insert(
                $this->items_table,
                [
                    'run_id'              => $run_id,
                    'run_item_uuid'       => (string) $item['run_item_uuid'],
                    'campaign_group_uuid' => $item['campaign_group_uuid'],
                    'item_ordinal'        => (int) $item['item_ordinal'],
                    'subject_type'        => (string) $item['subject_type'],
                    'subject_id'          => (int) $item['subject_id'],
                    'target_type'         => (string) $item['target_type'],
                    'parent_product_id'   => (int) $item['parent_product_id'],
                    'variation_id'        => (int) $item['variation_id'],
                    'target_snapshot'     => (string) $item['target_snapshot'],
                    'source_roles_snapshot'=> (string) $item['source_roles_snapshot'],
                    'destination_snapshot'=> (string) $item['destination_snapshot'],
                    'status'              => 'pending',
                    'attempts'            => 0,
                    'request_id'          => (string) $item['request_id'],
                    'request_fingerprint' => (string) $item['request_fingerprint'],
                    'command_snapshot'    => (string) $item['command_snapshot'],
                    'generation_key'      => (string) $item['generation_key'],
                    'poll_count'          => 0,
                    'updated_at'          => current_time( 'mysql', true ),
                ]
            );
            if ( false === $inserted ) {
                return false;
            }
        }

        return $this->item_graph_matches( $run_id, $user_id, $workflow, $items );
    }

    private function item_graph_matches( int $run_id, int $user_id, string $workflow, array $items ): bool {
        $persisted = $this->get_item_graph( $run_id, $user_id, $workflow );
        if ( count( $persisted ) !== count( $items ) ) {
            return false;
        }
        $expected = [];
        foreach ( $items as $item ) {
            $expected[ (string) $item['run_item_uuid'] ] = $item;
        }
        foreach ( $persisted as $row ) {
            $row = $this->normalize_item_row( $row );
            $key = (string) $row['run_item_uuid'];
            if ( ! isset( $expected[ $key ] ) || ! $this->immutable_item_matches( $row, $expected[ $key ] ) ) {
                return false;
            }
        }
        return true;
    }

    private function get_item_graph( int $run_id, int $user_id, string $workflow ): array {
        $workflow = sanitize_key( $workflow );
        if ( $run_id <= 0 || $user_id <= 0 || ! in_array( $workflow, self::WORKFLOWS, true ) ) {
            return [];
        }
        global $wpdb;
        return (array) $wpdb->get_results(
            $wpdb->prepare(
                "SELECT i.run_item_uuid, i.campaign_group_uuid, i.item_ordinal,
                    i.subject_type, i.subject_id, i.target_type,
                    i.parent_product_id, i.variation_id, i.target_snapshot,
                    i.source_roles_snapshot, i.destination_snapshot, i.generation_key,
                    i.request_id, i.request_fingerprint, i.command_snapshot
                FROM {$this->items_table} i
                INNER JOIN {$this->runs_table} r ON r.id = i.run_id
                WHERE i.run_id = %d AND r.user_id = %d AND r.workflow = %s
                ORDER BY i.id ASC",
                $run_id,
                $user_id,
                $workflow
            ),
            ARRAY_A
        );
    }

    private function immutable_item_matches( array $stored, array $expected ): bool {
        foreach ( [ 'run_item_uuid', 'generation_key', 'request_id', 'request_fingerprint', 'command_snapshot', 'target_snapshot', 'source_roles_snapshot', 'destination_snapshot' ] as $field ) {
            $left = (string) ( $stored[ $field ] ?? '' );
            $right = (string) ( $expected[ $field ] ?? '' );
            if ( '' === $left || '' === $right || ! hash_equals( $left, $right ) ) {
                return false;
            }
        }
        foreach ( [ 'campaign_group_uuid', 'item_ordinal', 'subject_type', 'subject_id', 'target_type', 'parent_product_id', 'variation_id' ] as $field ) {
            if ( (string) ( $stored[ $field ] ?? '' ) !== (string) ( $expected[ $field ] ?? '' ) ) {
                return false;
            }
        }
        return true;
    }

    private function schema_has_required_columns(): bool {
        global $wpdb;
        $run_columns = array_map(
            static fn( array $row ): string => (string) ( $row['Field'] ?? '' ),
            (array) $wpdb->get_results( "SHOW COLUMNS FROM {$this->runs_table}", ARRAY_A )
        );
        $item_columns = array_map(
            static fn( array $row ): string => (string) ( $row['Field'] ?? '' ),
            (array) $wpdb->get_results( "SHOW COLUMNS FROM {$this->items_table}", ARRAY_A )
        );
        return empty(
            array_diff(
                [ 'id', 'run_uuid', 'user_id', 'workflow', 'request_fingerprint', 'status', 'created_at', 'started_at', 'updated_at', 'finished_at', 'last_heartbeat_at', 'status_reason', 'filters_snapshot', 'selected_count', 'completed_count', 'failed_count', 'last_error', 'lock_owner', 'lock_expires_at' ],
                $run_columns
            )
        )
            && empty(
                array_diff(
                    [ 'id', 'run_id', 'run_item_uuid', 'campaign_group_uuid', 'item_ordinal', 'subject_type', 'subject_id', 'target_type', 'parent_product_id', 'variation_id', 'target_snapshot', 'source_roles_snapshot', 'destination_snapshot', 'status', 'attempts', 'run_id_single', 'request_id', 'request_fingerprint', 'command_snapshot', 'generation_key', 'job_ids', 'attachment_id', 'attachment_ids', 'generation_started_at', 'last_polled_at', 'poll_count', 'started_at', 'finished_at', 'updated_at', 'error_code', 'error_message', 'meta', 'claim_owner', 'claim_expires_at' ],
                    $item_columns
                )
            );
    }

    private function schema_is_current(): bool {
        if ( ! $this->schema_has_required_columns() ) {
            return false;
        }
        global $wpdb;
        $run_columns = array_map(
            static fn( array $row ): string => (string) ( $row['Field'] ?? '' ),
            (array) $wpdb->get_results( "SHOW COLUMNS FROM {$this->runs_table}", ARRAY_A )
        );
        $item_columns = array_map(
            static fn( array $row ): string => (string) ( $row['Field'] ?? '' ),
            (array) $wpdb->get_results( "SHOW COLUMNS FROM {$this->items_table}", ARRAY_A )
        );
        $current_run_columns = [ 'id', 'run_uuid', 'user_id', 'workflow', 'request_fingerprint', 'status', 'created_at', 'started_at', 'updated_at', 'finished_at', 'last_heartbeat_at', 'status_reason', 'filters_snapshot', 'selected_count', 'completed_count', 'failed_count', 'last_error', 'lock_owner', 'lock_expires_at' ];
        $current_item_columns = [ 'id', 'run_id', 'run_item_uuid', 'campaign_group_uuid', 'item_ordinal', 'subject_type', 'subject_id', 'target_type', 'parent_product_id', 'variation_id', 'target_snapshot', 'source_roles_snapshot', 'destination_snapshot', 'status', 'attempts', 'run_id_single', 'request_id', 'request_fingerprint', 'command_snapshot', 'generation_key', 'job_ids', 'attachment_id', 'attachment_ids', 'generation_started_at', 'last_polled_at', 'poll_count', 'started_at', 'finished_at', 'updated_at', 'error_code', 'error_message', 'meta', 'claim_owner', 'claim_expires_at' ];
        sort( $run_columns );
        sort( $item_columns );
        sort( $current_run_columns );
        sort( $current_item_columns );
        if ( $current_run_columns !== $run_columns || $current_item_columns !== $item_columns ) {
            return false;
        }
        $target_column = $wpdb->get_row(
            $wpdb->prepare( "SHOW COLUMNS FROM {$this->items_table} LIKE %s", 'target_type' ),
            ARRAY_A
        );
        if ( ! is_array( $target_column ) || 'unknown' !== (string) ( $target_column['Default'] ?? '' ) ) {
            return false;
        }
        $required_item_indexes = [
            'run_item'           => [ [ 'run_id', 'run_item_uuid' ], true ],
            'run_generation_key' => [ [ 'run_id', 'generation_key' ], true ],
            'run_target_lookup'  => [ [ 'run_id', 'target_type', 'subject_type', 'subject_id' ], false ],
            'campaign_group'     => [ [ 'campaign_group_uuid' ], false ],
            'run_status_id'      => [ [ 'run_id', 'status', 'id' ], false ],
            'request_id'         => [ [ 'request_id' ], false ],
            'attachment_id'      => [ [ 'attachment_id' ], false ],
        ];
        foreach ( $required_item_indexes as $name => [ $columns, $unique ] ) {
            if ( ! $this->index_matches( $this->get_index_definition( $name ), $columns, $unique ) ) {
                return false;
            }
        }
        $item_index_names = array_values( array_unique( array_map( 'strval', (array) $wpdb->get_col( "SHOW INDEX FROM {$this->items_table}", 2 ) ) ) );
        $run_index_names  = array_values( array_unique( array_map( 'strval', (array) $wpdb->get_col( "SHOW INDEX FROM {$this->runs_table}", 2 ) ) ) );
        $current_item_indexes = [ 'PRIMARY', 'run_item', 'run_generation_key', 'run_target_lookup', 'campaign_group', 'run_status_id', 'request_id', 'attachment_id' ];
        $current_run_indexes  = [ 'PRIMARY', 'run_uuid', 'user_status_created', 'user_workflow_status' ];
        sort( $item_index_names );
        sort( $run_index_names );
        sort( $current_item_indexes );
        sort( $current_run_indexes );
        if ( $current_item_indexes !== $item_index_names || $current_run_indexes !== $run_index_names ) {
            return false;
        }
        return $this->index_matches( $this->get_table_index_definition( $this->runs_table, 'run_uuid' ), [ 'run_uuid' ], true )
            && $this->index_matches( $this->get_table_index_definition( $this->runs_table, 'user_status_created' ), [ 'user_id', 'status', 'created_at' ], false )
            && $this->index_matches( $this->get_table_index_definition( $this->runs_table, 'user_workflow_status' ), [ 'user_id', 'workflow', 'status' ], false );
    }

    private function get_index_definition( string $name ): array {
        return $this->get_table_index_definition( $this->items_table, $name );
    }

    private function get_table_index_definition( string $table, string $name ): array {
        global $wpdb;
        $rows = $wpdb->get_results(
            $wpdb->prepare( "SHOW INDEX FROM {$table} WHERE Key_name = %s", $name ),
            ARRAY_A
        );
        if ( ! is_array( $rows ) ) {
            return [];
        }
        usort(
            $rows,
            static fn( array $left, array $right ): int => (int) ( $left['Seq_in_index'] ?? 0 ) <=> (int) ( $right['Seq_in_index'] ?? 0 )
        );
        return $rows;
    }

    private function index_matches( array $definition, array $columns, bool $unique ): bool {
        if ( empty( $definition ) ) {
            return false;
        }
        $actual_columns = array_map(
            static fn( array $row ): string => (string) ( $row['Column_name'] ?? '' ),
            $definition
        );
        return $columns === $actual_columns
            && ( $unique ? 0 : 1 ) === (int) ( $definition[0]['Non_unique'] ?? 1 );
    }

    private function normalize_item_row( array $row ): array {
        $subject_type = $this->normalize_subject_type( $row['subject_type'] ?? 'product' );
        $subject_id = absint( $row['subject_id'] ?? 0 );
        $row['subject_type'] = $subject_type;
        $row['subject_id'] = $subject_id;
        $row['run_item_uuid'] = (string) ( $row['run_item_uuid'] ?? '' );
        $row['campaign_group_uuid'] = (string) ( $row['campaign_group_uuid'] ?? '' );
        $row['item_ordinal'] = max( 0, (int) ( $row['item_ordinal'] ?? 0 ) );
        $row['target_type'] = sanitize_key( (string) ( $row['target_type'] ?? 'unknown' ) );
        if ( '' === $row['target_type'] ) {
            $row['target_type'] = 'unknown';
        }
        $row['parent_product_id'] = absint( $row['parent_product_id'] ?? 0 );
        $row['variation_id'] = absint( $row['variation_id'] ?? 0 );
        return $row;
    }

    private function normalize_subject_type( $subject_type ): string {
        $subject_type = sanitize_key( (string) $subject_type );
        return in_array( $subject_type, self::SUBJECT_TYPES, true ) ? $subject_type : 'product';
    }

    private function normalize_fingerprint( $fingerprint ): ?string {
        $fingerprint = strtolower( trim( (string) $fingerprint ) );
        return 1 === preg_match( '/^[a-f0-9]{64}$/', $fingerprint ) ? $fingerprint : null;
    }

    private function normalize_command_snapshot( $snapshot ): ?string {
        if ( is_string( $snapshot ) ) {
            $decoded = json_decode( $snapshot, true );
            $snapshot = is_array( $decoded ) ? $decoded : null;
        }
        if ( ! is_array( $snapshot ) || fotogenic_ai_array_is_list( $snapshot ) || empty( $snapshot ) ) {
            return null;
        }
        $encoded = wp_json_encode( $snapshot, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE );
        return is_string( $encoded ) && '' !== $encoded ? $encoded : null;
    }

    /** @param mixed $value Candidate value. */
    private function normalize_uuid( $value ): ?string {
        $value = strtolower( trim( (string) $value ) );
        return wp_is_uuid( $value ) ? $value : null;
    }

    /** @param mixed $snapshot Stored snapshot. */
    private function normalize_target_snapshot( $snapshot ): ?array {
        if ( is_string( $snapshot ) ) {
            $decoded = json_decode( $snapshot, true );
            $snapshot = is_array( $decoded ) ? $decoded : null;
        }
        $normalized = Workflow_Item_Identity::normalize_target_snapshot( $snapshot );
        return is_wp_error( $normalized ) || ! in_array( (string) ( $normalized['targetType'] ?? '' ), [ 'product', 'variation', 'media', 'post', 'page' ], true )
            ? null
            : $normalized;
    }

    /** @param mixed $snapshot Stored snapshot. */
    private function normalize_json_snapshot( $snapshot ): ?string {
        if ( $snapshot instanceof \stdClass ) {
            $snapshot = get_object_vars( $snapshot );
        } elseif ( is_string( $snapshot ) ) {
            $trimmed = trim( $snapshot );
            $decoded = json_decode( $trimmed, true );
            if ( ! is_array( $decoded ) || JSON_ERROR_NONE !== json_last_error() || ! fotogenic_ai_str_starts_with( $trimmed, '{' ) ) {
                return null;
            }
            $snapshot = $decoded;
        }
        if ( ! is_array( $snapshot ) || ( ! empty( $snapshot ) && fotogenic_ai_array_is_list( $snapshot ) ) ) {
            return null;
        }
        $encoded = wp_json_encode( (object) $snapshot, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE );
        return is_string( $encoded ) && '' !== $encoded ? $encoded : null;
    }

    private function admission_result(
        bool $ok,
        int $run_id,
        bool $created,
        bool $activated,
        ?array $run,
        string $reason
    ): array {
        return [
            'ok'        => $ok,
            'run_id'    => $run_id,
            'created'   => $created,
            'activated' => $activated,
            'run'       => $run,
            'reason'    => $reason,
        ];
    }

    private function admission_lock_name( int $user_id ): string {
        return 'fotogenic-ai-' . sha1( $this->runs_table . ':' . $user_id . ':bulk-admission' );
    }

    private function normalize_statuses( array $statuses ): array {
        return array_values( array_unique( array_filter( array_map( 'sanitize_key', $statuses ) ) ) );
    }

    /**
     * @param mixed $value JSON or list value.
     */
    private function decode_id_list( $value ): array {
        if ( is_string( $value ) ) {
            $decoded = json_decode( $value, true );
            $value = is_array( $decoded ) ? $decoded : [];
        }
        return is_array( $value )
            ? array_values( array_unique( array_filter( array_map( 'absint', $value ) ) ) )
            : [];
    }

    /**
     * @param mixed $value JSON or object value.
     */
    private function decode_json_object( $value ): array {
        if ( is_string( $value ) && '' !== $value ) {
            $decoded = json_decode( $value, true );
            return is_array( $decoded ) ? $decoded : [];
        }
        return is_array( $value ) ? $value : [];
    }
}
