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

<?php
/**
 * Abstract base for all WPDO data migrations.
 *
 * @package WP_Data_Optimizer
 */

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

/**
 * Abstract base for all WPDO data migrations.
 *
 * Each subclass implements:
 *  - get_module()    — module name (TMDO_Feature_Flags key)
 *  - get_zone()      — zone identifier ('' for HPCT modules, 'hot'/'warm'/'cold'/'archive' for zone modules)
 *  - count_source()  — total number of native rows to migrate
 *  - migrate_batch() — migrate one batch starting at $offset, returns rows processed
 *  - verify_counts() — returns true when custom table count >= native count
 *
 * Migration record is tracked in wpdo_migrations table.
 * Batch size: 500 rows. Timeout: 28 seconds per run.
 */
abstract class TMDO_Migration_Base {

	protected const BATCH_SIZE = 500;
	protected const TIMEOUT    = 28; // Seconds.

	/**
	 * Returns the module identifier for this migration.
	 *
	 * @return string Module name.
	 */
	abstract public function get_module(): string;

	/**
	 * Zone identifier. Empty string for HPCT-inherited modules.
	 */
	public function get_zone(): string {
		return '';
	}

	/**
	 * Returns the total number of source rows to migrate.
	 *
	 * @return int Total row count.
	 */
	abstract protected function count_source(): int;

	/**
	 * Migrate one batch of rows.
	 *
	 * @param int $offset Starting row offset.
	 * @return int Number of rows processed in this batch.
	 */
	abstract protected function migrate_batch( int $offset ): int;

	/**
	 * Verify that the custom table count is >= native count.
	 */
	abstract public function verify_counts(): bool;

	// ── Migration record helpers ──────────────────────────────────────────

	/**
	 * Retrieves the current migration record from the database.
	 *
	 * @return array|null Migration record array, or null if not found.
	 */
	public function get_record(): ?array {
		global $wpdb;
		$table = TMDO_DB::table( 'wpdo_migrations' );
		$row   = $wpdb->get_row(
			$wpdb->prepare( "SELECT * FROM `{$table}` WHERE module = %s", $this->get_module() ), // phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name from TMDO_DB::table().
			ARRAY_A
		);

		return $row ?: null;
	}

	/**
	 * Initialize or reset the migration record.
	 */
	public function init_record(): void {
		global $wpdb;
		$table = TMDO_DB::table( 'wpdo_migrations' );
		$now   = TMDO_DB::now();
		$total = $this->count_source();

		$existing = $this->get_record();

		if ( $existing ) {
			$wpdb->update(
				$table,
				array(
					'state'          => 'backfill',
					'zone'           => $this->get_zone(),
					'total_rows'     => $total,
					'processed_rows' => 0,
					'last_offset'    => 0,
					'error_count'    => 0,
					'started_at'     => $now,
					'completed_at'   => null,
					'updated_at'     => $now,
				),
				array( 'module' => $this->get_module() ),
				array( '%s', '%s', '%d', '%d', '%d', '%d', '%s', '%s', '%s' ),
				array( '%s' )
			);
		} else {
			$wpdb->insert(
				$table,
				array(
					'module'         => $this->get_module(),
					'zone'           => $this->get_zone(),
					'state'          => 'backfill',
					'total_rows'     => $total,
					'processed_rows' => 0,
					'last_offset'    => 0,
					'error_count'    => 0,
					'started_at'     => $now,
					'created_at'     => $now,
					'updated_at'     => $now,
				),
				array( '%s', '%s', '%s', '%d', '%d', '%d', '%d', '%s', '%s', '%s' )
			);
		}
	}

	/**
	 * Resume an existing migration (does NOT reset processed_rows).
	 */
	public function resume_record(): void {
		global $wpdb;
		$table = TMDO_DB::table( 'wpdo_migrations' );

		$wpdb->update(
			$table,
			array(
				'state'      => 'backfill',
				'updated_at' => TMDO_DB::now(),
			),
			array( 'module' => $this->get_module() ),
			array( '%s', '%s' ),
			array( '%s' )
		);
	}

	/**
	 * Update progress after each batch.
	 *
	 * @param int $processed   Number of rows processed in this batch.
	 * @param int $last_offset Last row offset processed.
	 * @return void
	 */
	public function update_progress( int $processed, int $last_offset ): void {
		global $wpdb;
		$table = TMDO_DB::table( 'wpdo_migrations' );

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name from TMDO_DB::table().
		$wpdb->query(
			$wpdb->prepare(
				"UPDATE `{$table}` SET
					processed_rows = processed_rows + %d,
					last_offset    = %d,
					updated_at     = %s
				WHERE module = %s",
				$processed,
				$last_offset,
				TMDO_DB::now(),
				$this->get_module()
			)
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared
	}

	/**
	 * Mark migration as completed (state → verify).
	 */
	public function mark_complete(): void {
		global $wpdb;
		$table = TMDO_DB::table( 'wpdo_migrations' );
		$now   = TMDO_DB::now();

		$wpdb->update(
			$table,
			array(
				'state'        => 'verify',
				'completed_at' => $now,
				'updated_at'   => $now,
			),
			array( 'module' => $this->get_module() ),
			array( '%s', '%s', '%s' ),
			array( '%s' )
		);
	}

	/**
	 * Increment error count.
	 */
	public function increment_errors(): void {
		global $wpdb;
		$table = TMDO_DB::table( 'wpdo_migrations' );

		// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name from TMDO_DB::table().
		$wpdb->query(
			$wpdb->prepare(
				"UPDATE `{$table}` SET error_count = error_count + 1, updated_at = %s WHERE module = %s",
				TMDO_DB::now(),
				$this->get_module()
			)
		);
		// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared
	}

	/**
	 * Run the migration in batches. Stops after ~28 seconds.
	 *
	 * @param bool          $resume            If true, resumes from last_offset.
	 * @param callable|null $progress_callback Called after each batch with (processed, total).
	 * @return bool True if migration completed, false if timed out (resume needed).
	 */
	public function run( bool $resume = false, ?callable $progress_callback = null ): bool {
		// Set feature flag to dual_write before starting.
		TMDO_Feature_Flags::set( $this->get_module(), 'dual_write' );

		if ( $resume ) {
			$record = $this->get_record();
			$offset = (int) ( $record['last_offset'] ?? 0 );
			$this->resume_record();
		} else {
			$offset = 0;
			$this->init_record();
		}

		// Advance to backfill state.
		TMDO_Feature_Flags::set( $this->get_module(), 'backfill' );

		$start = microtime( true );

		do {
			try {
				$batch_count = $this->migrate_batch( $offset );
			} catch ( \Throwable $e ) {
				$this->increment_errors();
				TMDO_Logger::error(
					$this->get_module(),
					'migrate_batch',
					$e->getMessage(),
					array(
						'offset' => $offset,
						'zone'   => $this->get_zone(),
					)
				);
				$batch_count = 0;

				// Check if too many errors.
				$record = $this->get_record();
				if ( $record && (int) $record['error_count'] >= 10 ) {
					TMDO_Feature_Flags::reset( $this->get_module() );
					return false;
				}
			}

			if ( $batch_count > 0 ) {
				$this->update_progress( $batch_count, $offset + $batch_count );
				$offset += $batch_count;
			}

			if ( $progress_callback ) {
				$record = $this->get_record();
				$progress_callback( (int) $record['processed_rows'], (int) $record['total_rows'] );
			}

			if ( ( microtime( true ) - $start ) > self::TIMEOUT ) {
				return false; // Timed out — next run will resume.
			}
		} while ( $batch_count >= self::BATCH_SIZE );

		// Backfill done → move to verify.
		$this->mark_complete();
		TMDO_Feature_Flags::set( $this->get_module(), 'verify' );

		return true;
	}
}