2meet-data-optimizer/includes/migration/class-tmdo-migration-orchestrator.php

<?php
// phpcs:ignore WPDO.AntiEAV -- platform migration tool: WPDO v1->v2 legacy meta orchestration
/**
 * One-click User Entity Migration Orchestrator.
 *
 * Drives the full demote → backfill → verify → promote → cleanup flow as a
 * single state machine. Designed for both small (sync, <30s) and large
 * (async via cron, minutes-to-hours) sites with auto-detection.
 *
 * Fast mode: bulk SQL pivot (INSERT...SELECT...GROUP BY) for text-only groups
 * is 10-50× faster than per-row PHP loops. Per-row safe_unserialize fallback
 * applies only to groups containing `json` fields (currently only `hp_user`).
 *
 * @package WP_Data_Optimizer
 * @since 2.8.0
 */

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

// phpcs:disable Squiz.Commenting.FunctionComment,Squiz.Commenting.VariableComment,Squiz.Commenting.ClassComment,WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.Security.EscapeOutput.ExceptionNotEscaped,WordPress.PHP.YodaConditions,Squiz.Commenting.InlineComment.InvalidEndChar -- Internal state-machine helpers and Throwable messages — not user output. SQL composed with sanitize_column_name() + $wpdb->prepare() values.

/**
 * One-click User Migration Orchestrator — drives the demote → backfill →
 * verify → promote → cleanup state machine for the user entity.
 *
 * @since 2.8.0
 */
final class TMDO_Migration_Orchestrator {

	const OPT_JOB             = 'wpdo_migration_job';
	const OPT_LOCK            = 'wpdo_migration_lock';
	const TRANS_PROGRESS      = 'wpdo_migration_progress';
	const TRANS_NEEDS         = 'wpdo_migration_needs_attention';
	const CRON_HOOK           = 'wpdo_migration_tick';
	const BACKUP_DIR_REL      = 'wpdo-backups';
	const MAX_LOG_LINES       = 50;
	const SYNC_THRESHOLD_SEC  = 30;
	const SYNC_DEADLINE_SEC   = 110;
	const VERIFY_SAMPLE_MIN   = 500;
	const VERIFY_SAMPLE_RATIO = 0.10;
	const LOCK_TTL_SEC        = 1800;
	const NEEDS_TTL_SEC       = 300;
	const ENTITY_TYPE         = 'user';

	/**
	 * Phase order. Each phase is idempotent — re-entering returns ok if
	 * the desired state is already achieved.
	 */
	private const PHASES = array(
		'diagnose',
		'backup',
		'demote',
		'install_schema',
		'backfill_bulk',
		'backfill_unserialize',
		'promote_shadow',
		'verify_sample',
		'promote_aeav',
		'cleanup',
		'completed',
	);

	// ─────────────────────────────────────────────────────────────────────
	// Public API
	// ─────────────────────────────────────────────────────────────────────

	/**
	 * Read-only inspection — returns what `start()` would do.
	 *
	 * @return array{managed_keys:array,eav_rows:int,users:int,usermeta:int,ratio:float,mode:string,groups:array,estimated_strategy:string,estimated_sec:float}
	 */
	public static function preflight(): array {
		global $wpdb;

		$managed_keys = self::get_managed_keys();
		$eav_rows     = self::count_eav_residue( $managed_keys );
		$users        = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$wpdb->users}" );
		$usermeta     = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$wpdb->usermeta}" );
		$mode         = TMDO_Mode_Manager::get( self::ENTITY_TYPE );

		$groups = array();
		foreach ( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) as $group ) {
			$keys             = TMDO_Entity_Registry::get_group_keys( self::ENTITY_TYPE, $group );
			$has_json         = self::group_has_json_field( $group );
			$residue          = $keys ? self::count_eav_residue( $keys ) : 0;
			$flat_table       = TMDO_Schema_Manager::get_table_name( self::ENTITY_TYPE, $group );
			$flat_rows        = TMDO_Schema_Manager::table_exists( $flat_table )
				? (int) $wpdb->get_var( "SELECT COUNT(*) FROM `{$flat_table}`" )
				: 0;
			$groups[ $group ] = array(
				'keys'      => $keys,
				'residue'   => $residue,
				'flat_rows' => $flat_rows,
				'has_json'  => $has_json,
				'strategy'  => $has_json ? 'row_by_row' : 'bulk_pivot',
			);
		}

		// Heuristic: 5ms per bulk row, 80ms per per-row PHP entity.
		$bulk_rows = 0;
		$row_rows  = 0;
		foreach ( $groups as $g ) {
			if ( $g['has_json'] ) {
				$row_rows += $g['residue'];
			} else {
				$bulk_rows += $g['residue'];
			}
		}
		$estimated_sec = ( $bulk_rows * 0.005 ) + ( $row_rows * 0.080 ) + 2.0;
		$strategy      = $estimated_sec <= self::SYNC_THRESHOLD_SEC ? 'sync' : 'async';

		return array(
			'managed_keys'       => $managed_keys,
			'eav_rows'           => $eav_rows,
			'users'              => $users,
			'usermeta'           => $usermeta,
			'ratio'              => $users > 0 ? round( $usermeta / $users, 2 ) : 0,
			'mode'               => $mode,
			'groups'             => $groups,
			'estimated_strategy' => $strategy,
			'estimated_sec'      => round( $estimated_sec, 1 ),
		);
	}

	/**
	 * Cached attention summary for dashboard widget + tab-nav red dot.
	 *
	 * Hits a single SELECT against wp_usermeta + an in-process ratio
	 * calculation; cached for NEEDS_TTL_SEC (5 min) so dashboard widgets
	 * stay snappy. Bust the cache after migrations or mode changes via
	 * `bust_attention_cache()`.
	 *
	 * @return array{needs:bool,eav_rows:int,groups_with_residue:int,ratio:float,mode:string,job_state:string}
	 */
	public static function needs_attention(): array {
		$cached = get_transient( self::TRANS_NEEDS );
		if ( is_array( $cached ) ) {
			return $cached;
		}

		global $wpdb;
		$managed_keys = self::get_managed_keys();
		$eav_rows     = self::count_eav_residue( $managed_keys );

		$groups_with_residue = 0;
		foreach ( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) as $group ) {
			$keys = TMDO_Entity_Registry::get_group_keys( self::ENTITY_TYPE, $group );
			if ( $keys && self::count_eav_residue( $keys ) > 0 ) {
				++$groups_with_residue;
			}
		}

		$users    = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$wpdb->users}" );
		$usermeta = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$wpdb->usermeta}" );
		$mode     = TMDO_Mode_Manager::get( self::ENTITY_TYPE );

		$job_state = 'idle';
		$job       = get_option( self::OPT_JOB );
		if ( is_array( $job ) && ! empty( $job['state'] ) ) {
			$job_state = (string) $job['state'];
		}

		$result = array(
			'needs'               => $eav_rows > 0,
			'eav_rows'            => $eav_rows,
			'groups_with_residue' => $groups_with_residue,
			'ratio'               => $users > 0 ? round( $usermeta / $users, 2 ) : 0,
			'mode'                => $mode,
			'job_state'           => $job_state,
		);

		set_transient( self::TRANS_NEEDS, $result, self::NEEDS_TTL_SEC );
		return $result;
	}

	/**
	 * Bust the attention cache — call after migrations, mode changes,
	 * group registrations, etc.
	 */
	public static function bust_attention_cache(): void {
		delete_transient( self::TRANS_NEEDS );
	}

	/**
	 * Begin a migration job. If small, runs to completion in this request.
	 * Otherwise schedules cron-driven ticks and returns immediately.
	 *
	 * @param array{verify_strict?:bool,verify_24h?:bool,auto_backup?:bool,force_async?:bool,dry_run?:bool} $options
	 * @return array{ok:bool,job_id?:string,strategy?:string,error?:string,reason?:string}
	 */
	public static function start( array $options = array() ): array {
		if ( ! self::acquire_lock() ) {
			return array(
				'ok'    => false,
				'error' => __( '另一個 migration job 正在執行；請先取消或等候完成。', '2meet-data-optimizer' ),
			);
		}

		$preflight = self::preflight();

		// Idempotency: nothing to do.
		if ( $preflight['eav_rows'] === 0 && TMDO_Mode_Manager::MODE_AEAV_ONLY === $preflight['mode'] ) {
			self::release_lock();
			self::bust_attention_cache();
			return array(
				'ok'     => false,
				'reason' => 'nothing_to_do',
				'error'  => sprintf(
					/* translators: %s: ratio. */
					__( '所有 entity group 已 aeav_only / 0 EAV 殘留。當前 ratio %s。', '2meet-data-optimizer' ),
					(string) $preflight['ratio']
				),
			);
		}

		$options = wp_parse_args(
			$options,
			array(
				'verify_strict' => true,
				'verify_24h'    => false,
				'auto_backup'   => true,
				'force_async'   => false,
				'dry_run'       => false,
			)
		);

		$job = array(
			'job_id'           => 'mig_' . wp_generate_password( 12, false ),
			'started_at'       => time(),
			'updated_at'       => time(),
			'phase'            => 'diagnose',
			'phase_index'      => 0,
			'phase_progress'   => 0,
			'overall_progress' => 0,
			'log'              => array(),
			'options'          => $options,
			'metrics'          => array(
				'mode_start'     => $preflight['mode'],
				'ratio_start'    => $preflight['ratio'],
				'ratio_now'      => $preflight['ratio'],
				'eav_rows_start' => $preflight['eav_rows'],
				'eav_rows_now'   => $preflight['eav_rows'],
				'users'          => $preflight['users'],
			),
			'strategy'         => $options['force_async'] ? 'async' : $preflight['estimated_strategy'],
			'state'            => 'running',
			'errors'           => array(),
			'backup_path'      => '',
		);

		self::log(
			$job,
			sprintf(
				'Job started — strategy=%s, residue=%d rows across %d groups, mode=%s',
				$job['strategy'],
				$preflight['eav_rows'],
				count( array_filter( $preflight['groups'], fn( $g ) => $g['residue'] > 0 ) ),
				$preflight['mode']
			)
		);

		self::persist_job( $job );

		if ( 'sync' === $job['strategy'] ) {
			self::run_sync_loop( $job );
		} else {
			self::schedule_next_tick();
		}

		return array(
			'ok'       => true,
			'job_id'   => $job['job_id'],
			'strategy' => $job['strategy'],
		);
	}

	/**
	 * Advance one phase. Used by cron and by sync inline loop.
	 *
	 * @return array{done:bool,phase:string,error?:string}
	 * @throws \RuntimeException When a phase callback throws or returns non-ok status; caught internally and converted to an `error` array entry.
	 */
	public static function tick(): array {
		$job = get_option( self::OPT_JOB );
		if ( ! is_array( $job ) || empty( $job['job_id'] ) ) {
			return array(
				'done'  => true,
				'phase' => 'idle',
				'error' => 'No active job',
			);
		}
		if ( 'running' !== ( $job['state'] ?? '' ) ) {
			return array(
				'done'  => true,
				'phase' => $job['phase'] ?? 'unknown',
			);
		}

		$current = $job['phase'];
		$method  = 'phase_' . $current;

		try {
			if ( ! method_exists( __CLASS__, $method ) ) {
				throw new \RuntimeException( "Unknown phase: {$current}" );
			}

			$result = self::{$method}( $job );

			if ( 'in_progress' === ( $result['status'] ?? '' ) ) {
				self::persist_job( $job );
				if ( 'async' === $job['strategy'] ) {
					self::schedule_next_tick();
				}
				return array(
					'done'  => false,
					'phase' => $current,
				);
			}

			if ( 'ok' !== ( $result['status'] ?? '' ) ) {
				throw new \RuntimeException( $result['message'] ?? 'Phase returned non-ok status' );
			}

			$next = self::next_phase( $current );
			self::log( $job, sprintf( '✓ %s (%.2fs)', $current, microtime( true ) - ( $result['_started_at'] ?? microtime( true ) ) ) );
			$job['phase']            = $next;
			$job['phase_index']      = array_search( $next, self::PHASES, true );
			$job['phase_progress']   = 0;
			$job['overall_progress'] = (int) round( ( $job['phase_index'] / ( count( self::PHASES ) - 1 ) ) * 100 );
			$job['updated_at']       = time();

			if ( 'completed' === $next ) {
				$job['state']            = 'completed';
				$job['completed_at']     = time();
				$job['overall_progress'] = 100;
				self::log(
					$job,
					sprintf(
						'✅ Migration complete — ratio %s → %s (-%.0f%%)',
						$job['metrics']['ratio_start'],
						$job['metrics']['ratio_now'],
						( $job['metrics']['ratio_start'] - $job['metrics']['ratio_now'] ) / max( $job['metrics']['ratio_start'], 0.01 ) * 100
					)
				);
				self::persist_job( $job );
				self::release_lock();
				self::bust_attention_cache();
				return array(
					'done'  => true,
					'phase' => 'completed',
				);
			}

			self::persist_job( $job );

			if ( 'async' === $job['strategy'] ) {
				self::schedule_next_tick();
			}

			return array(
				'done'  => false,
				'phase' => $next,
			);

		} catch ( \Throwable $e ) {
			self::handle_failure( $job, $e );
			return array(
				'done'  => true,
				'phase' => 'failed',
				'error' => $e->getMessage(),
			);
		}
	}

	/**
	 * Inspection — read-only snapshot for the polling UI.
	 */
	public static function get_status(): array {
		$cached = get_transient( self::TRANS_PROGRESS );
		if ( is_array( $cached ) ) {
			return $cached;
		}
		$job = get_option( self::OPT_JOB );
		if ( ! is_array( $job ) ) {
			return array( 'state' => 'idle' );
		}
		return self::project_status( $job );
	}

	public static function cancel(): bool {
		$job = get_option( self::OPT_JOB );
		if ( ! is_array( $job ) || 'running' !== ( $job['state'] ?? '' ) ) {
			return false;
		}
		// Auto-rollback to safe state (dual_write) before clearing.
		try {
			$current_mode = TMDO_Mode_Manager::get( self::ENTITY_TYPE );
			if ( TMDO_Mode_Manager::MODE_AEAV_ONLY === $current_mode ) {
				TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_SHADOW_READ );
				TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_DUAL_WRITE );
			} elseif ( TMDO_Mode_Manager::MODE_SHADOW_READ === $current_mode ) {
				TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_DUAL_WRITE );
			}
		} catch ( \Throwable $e ) {
			// Logged but non-fatal.
			TMDO_Logger::warning( 'migration_cancel_rollback_failed', array( 'error' => $e->getMessage() ) );
		}

		$job['state']        = 'cancelled';
		$job['cancelled_at'] = time();
		self::log( $job, '⚠ Cancelled by operator — rolled back to dual_write.' );
		self::persist_job( $job );
		self::release_lock();
		self::bust_attention_cache();
		wp_clear_scheduled_hook( self::CRON_HOOK );
		return true;
	}

	public static function resume(): array {
		$job = get_option( self::OPT_JOB );
		if ( ! is_array( $job ) ) {
			return array(
				'ok'    => false,
				'error' => 'No job to resume',
			);
		}
		if ( ! in_array( $job['state'] ?? '', array( 'failed', 'paused' ), true ) ) {
			return array(
				'ok'    => false,
				'error' => 'Job is not in a resumable state',
			);
		}
		$job['state']      = 'running';
		$job['updated_at'] = time();
		self::log( $job, '↻ Resumed by operator.' );
		self::persist_job( $job );
		self::acquire_lock();

		if ( 'sync' === $job['strategy'] ) {
			self::run_sync_loop( $job );
		} else {
			self::schedule_next_tick();
		}
		return array( 'ok' => true );
	}

	// ─────────────────────────────────────────────────────────────────────
	// Sync inline loop — for sites where total work fits within
	// SYNC_DEADLINE_SEC. Heartbeats progress via transient on every phase.
	// ─────────────────────────────────────────────────────────────────────

	private static function run_sync_loop( array $job ): void {
		set_time_limit( self::SYNC_DEADLINE_SEC + 10 );
		$deadline = microtime( true ) + self::SYNC_DEADLINE_SEC;

		while ( microtime( true ) < $deadline ) {
			$result = self::tick();
			if ( $result['done'] ) {
				return;
			}
			// Tiny pause to let DB breathe and avoid 100% CPU pegs.
			usleep( 5000 );
			$job = get_option( self::OPT_JOB );
			if ( ! is_array( $job ) || 'running' !== ( $job['state'] ?? '' ) ) {
				return;
			}
		}

		// Deadline reached but not done — convert to async.
		$job             = get_option( self::OPT_JOB );
		$job['strategy'] = 'async';
		self::log( $job, '⏱ Sync deadline reached — switching to async (cron-driven) for remaining phases.' );
		self::persist_job( $job );
		self::schedule_next_tick();
	}

	// ─────────────────────────────────────────────────────────────────────
	// Phases — each returns ['status' => 'ok'|'in_progress'|'error', 'message' => string]
	// ─────────────────────────────────────────────────────────────────────

	private static function phase_diagnose( array &$job ): array {
		$preflight                      = self::preflight();
		$job['metrics']['eav_rows_now'] = $preflight['eav_rows'];
		$job['metrics']['ratio_now']    = $preflight['ratio'];

		self::log(
			$job,
			sprintf(
				'Diagnose: %d EAV residue rows, %d users, ratio %s, mode %s',
				$preflight['eav_rows'],
				$preflight['users'],
				(string) $preflight['ratio'],
				$preflight['mode']
			)
		);

		return array(
			'status'  => 'ok',
			'message' => 'Diagnose complete',
		);
	}

	private static function phase_backup( array &$job ): array {
		if ( empty( $job['options']['auto_backup'] ) ) {
			self::log( $job, '↪ Backup skipped (auto_backup=false)' );
			return array( 'status' => 'ok' );
		}

		$upload_dir = wp_upload_dir();
		$backup_dir = trailingslashit( $upload_dir['basedir'] ) . self::BACKUP_DIR_REL;

		if ( ! wp_mkdir_p( $backup_dir ) ) {
			throw new \RuntimeException( "Cannot create backup directory: {$backup_dir}" );
		}

		// HTTP-level deny for Apache/IIS. Note: nginx silently ignores .htaccess —
		// operators on nginx must add a `location ~ /wp-content/uploads/wpdo-backups/
		// { deny all; }` block (logged in admin notice).
		$htaccess = $backup_dir . '/.htaccess';
		if ( ! file_exists( $htaccess ) ) {
			file_put_contents( $htaccess, "Require all denied\n" );
			@chmod( $htaccess, 0644 );
		}
		$webconfig = $backup_dir . '/web.config';
		if ( ! file_exists( $webconfig ) ) {
			file_put_contents(
				$webconfig,
				"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<configuration><system.webServer><authorization><deny users=\"*\"/></authorization></system.webServer></configuration>\n"
			);
			@chmod( $webconfig, 0644 );
		}
		$index = $backup_dir . '/index.php';
		if ( ! file_exists( $index ) ) {
			file_put_contents( $index, "<?php // Silence is golden.\n" );
			@chmod( $index, 0644 );
		}

		$filename    = sprintf(
			'wp_usermeta_%s_%s.sql',
			gmdate( 'Ymd_His' ),
			$job['job_id']
		);
		$backup_path = $backup_dir . '/' . $filename;

		global $wpdb;
		$rows = $wpdb->get_results(
			"SELECT umeta_id, user_id, meta_key, meta_value FROM {$wpdb->usermeta} ORDER BY umeta_id ASC",
			ARRAY_A
		);

		$fp = fopen( $backup_path, 'wb' );
		if ( ! $fp ) {
			throw new \RuntimeException( "Cannot open backup file for writing: {$backup_path}" );
		}
		// Owner-only read/write — backup contains user PII (emails, billing
		// addresses, OAuth tokens, session blobs). Default umask leaks to
		// other system users on shared hosts.
		@chmod( $backup_path, 0600 );

		fwrite( $fp, "-- WPDO migration backup of {$wpdb->usermeta}\n" );
		fwrite( $fp, '-- Job: ' . $job['job_id'] . "\n" );
		fwrite( $fp, '-- Generated: ' . gmdate( 'c' ) . "\n" );
		fwrite( $fp, "-- Restore: mysql ... < this_file.sql\n\n" );
		fwrite( $fp, "/*!40101 SET NAMES utf8mb4 */;\n" );
		fwrite( $fp, "/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;\n" );
		fwrite( $fp, "SET FOREIGN_KEY_CHECKS=0;\n" );
		fwrite( $fp, "LOCK TABLES `{$wpdb->usermeta}` WRITE;\n" );

		$batch = array();
		$count = 0;
		foreach ( $rows as $row ) {
			// UNHEX-encode meta_value: avoids ALL escape edge cases (binary,
			// embedded NULs, non-UTF8, sql_mode mismatches at restore time).
			// meta_key is %-bound through esc_sql which is sufficient for an
			// ASCII-only key universe.
			$batch[] = sprintf(
				"(%d,%d,'%s',UNHEX('%s'))",
				(int) $row['umeta_id'],
				(int) $row['user_id'],
				esc_sql( (string) $row['meta_key'] ),
				bin2hex( (string) ( $row['meta_value'] ?? '' ) )
			);
			++$count;
			if ( count( $batch ) >= 500 ) {
				fwrite( $fp, "INSERT INTO `{$wpdb->usermeta}` (umeta_id,user_id,meta_key,meta_value) VALUES\n" . implode( ",\n", $batch ) . ";\n" );
				$batch = array();
			}
		}
		if ( $batch ) {
			fwrite( $fp, "INSERT INTO `{$wpdb->usermeta}` (umeta_id,user_id,meta_key,meta_value) VALUES\n" . implode( ",\n", $batch ) . ";\n" );
		}
		fwrite( $fp, "UNLOCK TABLES;\n" );
		fwrite( $fp, "SET FOREIGN_KEY_CHECKS=1;\n" );
		fwrite( $fp, "/*!40101 SET SQL_MODE=@OLD_SQL_MODE */;\n" );
		fclose( $fp );

		$job['backup_path'] = $backup_path;
		self::log(
			$job,
			sprintf(
				'Backup written: %s (%d rows, %s, chmod 0600)',
				$filename,
				$count,
				size_format( filesize( $backup_path ) )
			)
		);

		// Best-effort web-server detection — log nginx warning so operator
		// can add the manual location block (htaccess/web.config don't apply).
		$server = isset( $_SERVER['SERVER_SOFTWARE'] ) ? strtolower( sanitize_text_field( wp_unslash( (string) $_SERVER['SERVER_SOFTWARE'] ) ) ) : '';
		if ( str_contains( $server, 'nginx' ) ) {
			self::log( $job, '⚠ nginx detected: add `location ~ /wp-content/uploads/wpdo-backups/ { deny all; }` to your nginx config — .htaccess does not apply.' );
		}

		return array( 'status' => 'ok' );
	}

	private static function phase_demote( array &$job ): array {
		$current = TMDO_Mode_Manager::get( self::ENTITY_TYPE );
		if ( TMDO_Mode_Manager::MODE_AEAV_ONLY !== $current ) {
			self::log( $job, "↪ Demote skipped (mode is {$current}, not aeav_only)" );
			return array( 'status' => 'ok' );
		}

		// Two-step demotion: aeav_only → shadow_read → dual_write.
		// (Safe transition: demotion always allowed.)
		$r = TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_SHADOW_READ );
		if ( is_wp_error( $r ) ) {
			throw new \RuntimeException( 'Demote step 1 failed: ' . $r->get_error_message() );
		}
		$r = TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_DUAL_WRITE );
		if ( is_wp_error( $r ) ) {
			throw new \RuntimeException( 'Demote step 2 failed: ' . $r->get_error_message() );
		}

		self::log( $job, 'Mode: aeav_only → dual_write (reads now go to EAV)' );
		return array( 'status' => 'ok' );
	}

	private static function phase_install_schema( array &$job ): array {
		do_action( 'wpdo_register_entity_fields', TMDO_Entity_Registry::class );
		TMDO_Schema_Manager::process_pending_migrations();

		// Verify all 9 expected tables exist.
		$missing = array();
		foreach ( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) as $group ) {
			$table = TMDO_Schema_Manager::get_table_name( self::ENTITY_TYPE, $group );
			if ( ! TMDO_Schema_Manager::table_exists( $table ) ) {
				$missing[] = $group;
			}
		}
		if ( $missing ) {
			throw new \RuntimeException( 'Schema migration failed; missing tables: ' . implode( ',', $missing ) );
		}

		$count = count( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) );
		self::log( $job, "Schema migration ok ({$count} flat tables verified)" );
		return array( 'status' => 'ok' );
	}

	private static function phase_backfill_bulk( array &$job ): array {
		if ( ! empty( $job['options']['dry_run'] ) ) {
			self::log( $job, '↪ Backfill bulk skipped (dry_run)' );
			return array( 'status' => 'ok' );
		}

		$total_groups = 0;
		$total_rows   = 0;
		foreach ( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) as $group ) {
			if ( self::group_has_json_field( $group ) ) {
				continue; // handled in phase_backfill_unserialize
			}
			$rows        = self::execute_bulk_pivot( $group );
			$total_rows += $rows;
			++$total_groups;
			self::log( $job, sprintf( '  • bulk pivot %s: %d row(s)', $group, $rows ) );
		}
		self::log( $job, sprintf( 'Bulk backfill: %d groups, %d rows total', $total_groups, $total_rows ) );

		// Update live ratio.
		$preflight                      = self::preflight();
		$job['metrics']['eav_rows_now'] = $preflight['eav_rows'];
		$job['metrics']['ratio_now']    = $preflight['ratio'];

		return array( 'status' => 'ok' );
	}

	private static function phase_backfill_unserialize( array &$job ): array {
		if ( ! empty( $job['options']['dry_run'] ) ) {
			self::log( $job, '↪ Backfill unserialize skipped (dry_run)' );
			return array( 'status' => 'ok' );
		}

		$total_rows = 0;
		foreach ( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) as $group ) {
			if ( ! self::group_has_json_field( $group ) ) {
				continue;
			}
			$result = TMDO_Entity_Migration_Engine::migrate_group(
				self::ENTITY_TYPE,
				$group,
				array( 'sleep_ms' => 0 )
			);
			if ( ! empty( $result['error'] ) ) {
				throw new \RuntimeException( "Row-by-row backfill {$group} failed: " . $result['error'] );
			}
			if ( ( $result['errors'] ?? 0 ) > 0 && 0 === ( $result['migrated'] ?? 0 ) ) {
				throw new \RuntimeException( "All rows failed during {$group} backfill — see error_log" );
			}
			$total_rows += (int) ( $result['migrated'] ?? 0 );
			self::log( $job, sprintf( '  • row-by-row %s: %d row(s)', $group, $result['migrated'] ?? 0 ) );
		}
		self::log( $job, sprintf( 'Row-by-row backfill: %d rows total', $total_rows ) );
		return array( 'status' => 'ok' );
	}

	private static function phase_promote_shadow( array &$job ): array {
		$current = TMDO_Mode_Manager::get( self::ENTITY_TYPE );
		if ( TMDO_Mode_Manager::MODE_DUAL_WRITE === $current ) {
			$r = TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_SHADOW_READ );
			if ( is_wp_error( $r ) ) {
				throw new \RuntimeException( 'Promote to shadow_read failed: ' . $r->get_error_message() );
			}
			self::log( $job, 'Mode: dual_write → shadow_read (verifying flat reads)' );
		} else {
			self::log( $job, "↪ Already at {$current}; skip promote_shadow" );
		}
		return array( 'status' => 'ok' );
	}

	private static function phase_verify_sample( array &$job ): array {
		global $wpdb;

		if ( ! empty( $job['options']['verify_24h'] ) ) {
			// Operator opted into the 24-hour shadow-read window — pause here.
			$elapsed = time() - ( $job['phase_started_at'] ?? time() );
			if ( ! isset( $job['phase_started_at'] ) ) {
				$job['phase_started_at'] = time();
				self::log( $job, '⏸ 24h shadow_read window started — wizard will resume after window elapses.' );
				return array( 'status' => 'in_progress' );
			}
			if ( $elapsed < DAY_IN_SECONDS ) {
				return array( 'status' => 'in_progress' );
			}
			self::log( $job, '⏳ 24h shadow_read window complete; running sample compare' );
		}

		$strict      = ! empty( $job['options']['verify_strict'] );
		$users_total = (int) $wpdb->get_var( "SELECT COUNT(*) FROM {$wpdb->users}" );
		$sample_size = $strict
			? (int) max( self::VERIFY_SAMPLE_MIN, ceil( $users_total * self::VERIFY_SAMPLE_RATIO ) )
			: 100;
		$sample_size = min( $sample_size, $users_total );
		$sample_ids  = $wpdb->get_col(
			$wpdb->prepare(
				"SELECT ID FROM {$wpdb->users} ORDER BY RAND() LIMIT %d",
				$sample_size
			)
		);

		$diffs        = 0;
		$compared     = 0;
		$managed_keys = self::get_managed_keys();

		foreach ( $sample_ids as $uid ) {
			foreach ( $managed_keys as $key ) {
				$eav = $wpdb->get_var(
					$wpdb->prepare(
						"SELECT meta_value FROM {$wpdb->usermeta} WHERE user_id = %d AND meta_key = %s LIMIT 1",
						$uid,
						$key
					)
				);

				// Skip: no EAV source-of-truth to verify against (key already cleaned
				// or never existed). The wizard's invariant only flags as DIFF the
				// case where EAV has data but flat doesn't match it — that is the
				// only real "backfill error".
				if ( null === $eav || '' === $eav ) {
					continue;
				}

				++$compared;
				$flat = TMDO_Hook_Bus::direct_read( self::ENTITY_TYPE, (int) $uid, $key );

				if ( ! self::values_loose_equal( $flat, $eav ) ) {
					++$diffs;
					if ( $diffs <= 3 ) {
						self::log(
							$job,
							sprintf(
								'  ! diff uid=%d key=%s flat=%s eav=%s',
								$uid,
								$key,
								is_scalar( $flat ) ? (string) $flat : gettype( $flat ),
								is_scalar( $eav ) ? (string) $eav : gettype( $eav )
							)
						);
					}
				}
			}
		}

		self::log(
			$job,
			sprintf(
				'Verify: sampled %d users × %d keys = %d compares, %d diff(s)',
				count( $sample_ids ),
				count( $managed_keys ),
				$compared,
				$diffs
			)
		);

		if ( $diffs > 0 ) {
			throw new \RuntimeException(
				sprintf(
					'Verification found %d divergence(s) across %d compares — aborting before destructive cleanup.',
					$diffs,
					$compared
				)
			);
		}

		return array( 'status' => 'ok' );
	}

	private static function phase_promote_aeav( array &$job ): array {
		$current = TMDO_Mode_Manager::get( self::ENTITY_TYPE );
		if ( TMDO_Mode_Manager::MODE_AEAV_ONLY === $current ) {
			self::log( $job, '↪ Already aeav_only' );
			return array( 'status' => 'ok' );
		}
		$r = TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_AEAV_ONLY );
		if ( is_wp_error( $r ) ) {
			throw new \RuntimeException( 'Promote to aeav_only failed: ' . $r->get_error_message() );
		}
		self::log( $job, 'Mode: shadow_read → aeav_only (cutover complete)' );
		return array( 'status' => 'ok' );
	}

	private static function phase_cleanup( array &$job ): array {
		if ( ! empty( $job['options']['dry_run'] ) ) {
			self::log( $job, '↪ Cleanup skipped (dry_run)' );
			return array( 'status' => 'ok' );
		}

		// Hard guard: must be in aeav_only AND backup must exist (if requested).
		$mode = TMDO_Mode_Manager::get( self::ENTITY_TYPE );
		if ( TMDO_Mode_Manager::MODE_AEAV_ONLY !== $mode ) {
			throw new \RuntimeException( "Refusing cleanup — mode is {$mode}, must be aeav_only" );
		}
		if ( ! empty( $job['options']['auto_backup'] ) && empty( $job['backup_path'] ) ) {
			throw new \RuntimeException( 'Refusing cleanup — auto_backup requested but no backup_path on record' );
		}

		global $wpdb;
		$keys = self::get_managed_keys();
		if ( empty( $keys ) ) {
			self::log( $job, '↪ No managed keys to clean' );
			return array( 'status' => 'ok' );
		}

		$placeholders = implode( ',', array_fill( 0, count( $keys ), '%s' ) );
		$deleted      = (int) $wpdb->query(
			$wpdb->prepare(
				"DELETE FROM {$wpdb->usermeta} WHERE meta_key IN ({$placeholders})",
				...$keys
			)
		);

		// Refresh metrics.
		$preflight                      = self::preflight();
		$job['metrics']['eav_rows_now'] = $preflight['eav_rows'];
		$job['metrics']['ratio_now']    = $preflight['ratio'];

		self::log( $job, sprintf( 'Cleanup: deleted %d EAV row(s); ratio now %s', $deleted, (string) $preflight['ratio'] ) );
		return array( 'status' => 'ok' );
	}

	// ─────────────────────────────────────────────────────────────────────
	// Helpers
	// ─────────────────────────────────────────────────────────────────────

	/** Returns all meta_keys registered as entity fields for `user`. */
	private static function get_managed_keys(): array {
		$keys = array();
		foreach ( TMDO_Entity_Registry::get_groups_for_type( self::ENTITY_TYPE ) as $group ) {
			$keys = array_merge( $keys, TMDO_Entity_Registry::get_group_keys( self::ENTITY_TYPE, $group ) );
		}
		return array_values( array_unique( $keys ) );
	}

	private static function group_has_json_field( string $group ): bool {
		foreach ( TMDO_Entity_Registry::get_group_fields( self::ENTITY_TYPE, $group ) as $field ) {
			if ( 'json' === ( $field['type'] ?? '' ) ) {
				return true;
			}
		}
		return false;
	}

	private static function count_eav_residue( array $keys = array() ): int {
		global $wpdb;
		if ( empty( $keys ) ) {
			$keys = self::get_managed_keys();
		}
		if ( empty( $keys ) ) {
			return 0;
		}
		$placeholders = implode( ',', array_fill( 0, count( $keys ), '%s' ) );
		return (int) $wpdb->get_var(
			$wpdb->prepare(
				"SELECT COUNT(*) FROM {$wpdb->usermeta} WHERE meta_key IN ({$placeholders})",
				...$keys
			)
		);
	}

	/**
	 * Single-statement bulk pivot for a text-only group.
	 *
	 * @param string $group Entity group name.
	 * @return int Affected rows.
	 * @throws \RuntimeException If $wpdb->query() fails.
	 */
	private static function execute_bulk_pivot( string $group ): int {
		global $wpdb;

		$fields = TMDO_Entity_Registry::get_group_fields( self::ENTITY_TYPE, $group );
		if ( empty( $fields ) ) {
			return 0;
		}

		$adapter = TMDO_Entity_Registry::get_adapter( self::ENTITY_TYPE );
		$id_col  = $adapter->get_entity_id_column();
		$table   = TMDO_Schema_Manager::get_table_name( self::ENTITY_TYPE, $group );

		$select_cases = array();
		$update_cols  = array();
		$col_names    = array();
		$keys         = array();
		foreach ( $fields as $f ) {
			$key         = $f['key'];
			$col         = TMDO_Schema_Manager::sanitize_column_name( $key );
			$keys[]      = $key;
			$col_names[] = "`{$col}`";
			// MAX(CASE WHEN meta_key='...' THEN meta_value END)
			$select_cases[] = $wpdb->prepare(
				"MAX(CASE WHEN um.meta_key = %s THEN um.meta_value END) AS `{$col}`",
				$key
			);
			$update_cols[]  = "`{$col}` = COALESCE(VALUES(`{$col}`), `{$col}`)";
		}

		$placeholders = implode( ',', array_fill( 0, count( $keys ), '%s' ) );

		$sql = sprintf(
			'INSERT INTO `%s` (`%s`, %s)
			 SELECT um.user_id, %s
			 FROM `%s` um
			 WHERE um.meta_key IN (%s)
			 GROUP BY um.user_id
			 ON DUPLICATE KEY UPDATE %s',
			$table,
			$id_col,
			implode( ', ', $col_names ),
			implode( ', ', $select_cases ),
			$wpdb->usermeta,
			$placeholders,
			implode( ', ', $update_cols )
		);

		// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- pivot SQL composed from registry-validated identifiers + prepared values inside CASE/IN.
		$result = $wpdb->query( $wpdb->prepare( $sql, ...$keys ) );

		if ( false === $result ) {
			throw new \RuntimeException( "Bulk pivot failed for group {$group}: " . $wpdb->last_error );
		}
		return (int) $result;
	}

	private static function values_loose_equal( $a, $b ): bool {
		if ( null === $a && null === $b ) {
			return true;
		}
		if ( null === $a || null === $b ) {
			$other = null === $a ? $b : $a;
			return '' === $other || array() === $other || 0 === $other || '0' === $other;
		}
		if ( is_array( $a ) || is_array( $b ) ) {
			return wp_json_encode( $a ) === wp_json_encode( $b );
		}
		return (string) $a === (string) $b;
	}

	private static function next_phase( string $current ): string {
		$idx = array_search( $current, self::PHASES, true );
		if ( false === $idx || $idx + 1 >= count( self::PHASES ) ) {
			return 'completed';
		}
		return self::PHASES[ $idx + 1 ];
	}

	private static function log( array &$job, string $message ): void {
		$line         = '[' . gmdate( 'H:i:s' ) . '] ' . $message;
		$job['log'][] = $line;
		if ( count( $job['log'] ) > self::MAX_LOG_LINES ) {
			$job['log'] = array_slice( $job['log'], -self::MAX_LOG_LINES );
		}
		$job['updated_at'] = time();
	}

	private static function persist_job( array $job ): void {
		update_option( self::OPT_JOB, $job, false );
		set_transient( self::TRANS_PROGRESS, self::project_status( $job ), 60 );
	}

	/** Public-safe projection of job state for the polling UI. */
	private static function project_status( array $job ): array {
		return array(
			'job_id'           => $job['job_id'] ?? '',
			'state'            => $job['state'] ?? 'idle',
			'phase'            => $job['phase'] ?? 'idle',
			'phase_index'      => $job['phase_index'] ?? 0,
			'phase_total'      => count( self::PHASES ) - 1,
			'overall_progress' => $job['overall_progress'] ?? 0,
			'log'              => $job['log'] ?? array(),
			'metrics'          => $job['metrics'] ?? array(),
			'strategy'         => $job['strategy'] ?? 'sync',
			'started_at'       => $job['started_at'] ?? 0,
			'updated_at'       => $job['updated_at'] ?? 0,
			'completed_at'     => $job['completed_at'] ?? null,
			'cancelled_at'     => $job['cancelled_at'] ?? null,
			'errors'           => $job['errors'] ?? array(),
			'backup_path'      => isset( $job['backup_path'] ) ? basename( $job['backup_path'] ) : '',
		);
	}

	private static function acquire_lock(): bool {
		if ( false === add_option( self::OPT_LOCK, time(), '', false ) ) {
			$existing = (int) get_option( self::OPT_LOCK, 0 );
			if ( $existing > 0 && time() - $existing > self::LOCK_TTL_SEC ) {
				delete_option( self::OPT_LOCK );
				return add_option( self::OPT_LOCK, time(), '', false );
			}
			return false;
		}
		return true;
	}

	private static function release_lock(): void {
		delete_option( self::OPT_LOCK );
	}

	private static function schedule_next_tick(): void {
		if ( ! wp_next_scheduled( self::CRON_HOOK ) ) {
			wp_schedule_single_event( time() + 1, self::CRON_HOOK );
		}
	}

	private static function handle_failure( array &$job, \Throwable $e ): void {
		$job['state']    = 'failed';
		$job['errors'][] = array(
			'phase'   => $job['phase'] ?? 'unknown',
			'message' => $e->getMessage(),
			'at'      => time(),
		);
		self::log( $job, sprintf( '✗ %s failed: %s', $job['phase'] ?? '?', $e->getMessage() ) );

		// Auto-rollback: only if mode is currently aeav_only AND we haven't reached cleanup yet.
		try {
			$current     = TMDO_Mode_Manager::get( self::ENTITY_TYPE );
			$phase       = $job['phase'] ?? '';
			$pre_cleanup = ! in_array( $phase, array( 'cleanup', 'completed' ), true );
			if ( $pre_cleanup && TMDO_Mode_Manager::MODE_AEAV_ONLY === $current ) {
				TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_SHADOW_READ );
				TMDO_Mode_Manager::set( self::ENTITY_TYPE, TMDO_Mode_Manager::MODE_DUAL_WRITE );
				self::log( $job, '↩ Auto-rolled back to dual_write (reads safe)' );
			}
		} catch ( \Throwable $inner ) {
			TMDO_Logger::warning( 'migration_rollback_failed', array( 'error' => $inner->getMessage() ) );
		}

		TMDO_Logger::warning(
			'migration_phase_failed',
			array(
				'phase' => $job['phase'] ?? '?',
				'job'   => $job['job_id'] ?? '?',
				'error' => $e->getMessage(),
			)
		);

		self::persist_job( $job );
		self::release_lock();
		self::bust_attention_cache();
	}

	/**
	 * Cron callback — wires CRON_HOOK to tick().
	 */
	public static function cron_tick(): void {
		self::tick();
	}
}