feat(2meet-data-optimizer): write flat-table data back to native meta

- TMDO_Native_Restorer copies an entity's flat-table values into its native meta table so the value each key reads today is what native returns afterwards: the owning group (by post type for posts) wins, values are stored as WP stores them, NULL removes the native row
- Resumable per-group id cursor, catch-up pass for rows updated during the run, optional demote to dual_write; CLI wp tmdo bridge-restore-native <entity> [--dry-run] [--batch=<n>] [--demote]
- Verify: NativeRestorerTest (5, user) + NativeRestorerPostTest (2, scoped vs legacy routing); core unit 652 / integration 456, HP 148, WC 10 OK; scale estimate dev21 user ~5.1M / post ~4.0M flat rows
- PR: PR-031

2meet-data-optimizer.php

@@ -259,6 +259,7 @@ require_once TMDO_PATH . 'includes/migration/phases/class-tmdo-phase-completed.p
 require_once TMDO_PATH . 'includes/migration/class-tmdo-migration-orchestrator.php';
 require_once TMDO_PATH . 'includes/migration/class-tmdo-post-migration.php';
 require_once TMDO_PATH . 'includes/migration/class-tmdo-post-group-reconciler.php';
+require_once TMDO_PATH . 'includes/migration/class-tmdo-native-restorer.php';
 add_action( TMDO_Migration_Orchestrator::CRON_HOOK, array( 'TMDO_Migration_Orchestrator', 'cron_tick' ) );
 add_action( 'wpdo_bridge_mode_changed', array( 'TMDO_Migration_Orchestrator', 'bust_attention_cache' ) );
 add_action( 'tmdo_bridge_mode_changed', array( 'TMDO_Migration_Orchestrator', 'bust_attention_cache' ) );

cli/class-tmdo-cli-v2.php

@@ -152,6 +152,47 @@ class TMDO_CLI_V2 {
 		WP_CLI::success( "Hook Bus: {$state}" );
 	}
 
+	/**
+	 * Write an entity's flat-table values back to native meta — the way out of
+	 * aeav_only. Resumable; re-run after an interruption.
+	 *
+	 * ## OPTIONS
+	 *
+	 * <entity>
+	 * : post | user | term | comment
+	 *
+	 * [--dry-run]
+	 * : Count the flat rows per group; write nothing.
+	 *
+	 * [--batch=<n>]
+	 * : Rows per batch. Default 500.
+	 *
+	 * [--demote]
+	 * : After a complete run, switch the entity to dual_write (both copies kept in step).
+	 *
+	 * ## EXAMPLES
+	 *
+	 *     wp tmdo bridge-restore-native user --dry-run
+	 *     wp tmdo bridge-restore-native user --demote
+	 */
+	public function bridge_restore_native( $args, $assoc_args ): void {
+		$entity = (string) ( $args[0] ?? '' );
+		if ( ! in_array( $entity, TMDO_Entity_Registry::VALID_ENTITY_TYPES, true ) ) {
+			WP_CLI::error( 'Usage: wp tmdo bridge-restore-native <post|user|term|comment> [--dry-run] [--batch=<n>] [--demote]' );
+		}
+		$dry_run = isset( $assoc_args['dry-run'] );
+		$report  = TMDO_Native_Restorer::run( $entity, $dry_run, (int) ( $assoc_args['batch'] ?? 500 ), isset( $assoc_args['demote'] ) );
+
+		foreach ( $report['groups'] as $group => $g ) {
+			WP_CLI::log( sprintf( '%s/%s: %d row(s)%s', $entity, $group, $g['rows'], $dry_run ? '' : sprintf( ', %d native value(s) written', $g['written'] ) ) );
+		}
+		if ( $dry_run ) {
+			WP_CLI::success( '[dry-run] Nothing written.' );
+			return;
+		}
+		WP_CLI::success( $report['demoted'] ? "Restored; {$entity} is now dual_write." : 'Restored. Demote the entity to dual_write (admin Entity Bridge tab) before its next write, or re-run with --demote.' );
+	}
+
 	/**
 	 * Audit per-module state + shadow_read flags.
 	 *
@@ -784,6 +825,8 @@ WP_CLI::add_command( 'wpdo bridge-status', array( 'TMDO_CLI_V2', 'bridge_status'
 WP_CLI::add_command( 'tmdo bridge-status', array( 'TMDO_CLI_V2', 'bridge_status' ) );
 WP_CLI::add_command( 'wpdo bridge-set', array( 'TMDO_CLI_V2', 'bridge_set' ) );
 WP_CLI::add_command( 'tmdo bridge-set', array( 'TMDO_CLI_V2', 'bridge_set' ) );
+WP_CLI::add_command( 'wpdo bridge-restore-native', array( 'TMDO_CLI_V2', 'bridge_restore_native' ) );
+WP_CLI::add_command( 'tmdo bridge-restore-native', array( 'TMDO_CLI_V2', 'bridge_restore_native' ) );
 WP_CLI::add_command( 'wpdo mode-audit', array( 'TMDO_CLI_V2', 'mode_audit' ) );
 WP_CLI::add_command( 'tmdo mode-audit', array( 'TMDO_CLI_V2', 'mode_audit' ) );
 WP_CLI::add_command( 'wpdo mode-set', array( 'TMDO_CLI_V2', 'mode_set' ) );

includes/migration/class-tmdo-native-restorer.php

@@ -0,0 +1,209 @@
+<?php
+/**
+ * TMDO_Native_Restorer — write flat-table values back to native meta.
+ *
+ * The only way out of aeav_only: in that mode the flat tables are the sole
+ * copy of every managed key, and demoting the entity (or uninstalling) would
+ * leave reads on native meta that no longer has the data.
+ *
+ * Rule: afterwards native meta returns what reads return today.
+ *   - A key is written from the group that currently owns it for that entity
+ *     (TMDO_Entity_Registry::get_field(), by post type for posts), so stale
+ *     copies in other group tables never win.
+ *   - Non-NULL values are stored the way WP stores them (maybe_serialize of
+ *     the value Type_Caster returns); a NULL value deletes the native row.
+ *   - Entities with no row in the owning group's table are not touched:
+ *     leftover native rows for them are outside the flat data.
+ *
+ * Batched by entity id with a cursor option so an interrupted run resumes; a
+ * final pass re-copies rows updated since the first run started. With
+ * $demote the entity then moves to dual_write, which keeps both copies in
+ * step from then on.
+ *
+ * @package WP_Data_Optimizer
+ */
+
+declare(strict_types=1);
+
+defined( 'ABSPATH' ) || exit;
+
+// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared,WordPress.DB.PreparedSQL.NotPrepared,WordPress.DB.DirectDatabaseQuery,WordPress.DB.PreparedSQLPlaceholders.ReplacementsWrongNumber,WordPress.DB.PreparedSQLPlaceholders.UnfinishedPrepare -- table/column names come from the entity registry / adapter; placeholders are built for spread params.
+
+/**
+ * Copies flat-table values of one entity type back into its native meta table.
+ */
+final class TMDO_Native_Restorer {
+
+	/** Option: "entity/group" → last entity id written. */
+	public const CURSOR_OPTION = 'tmdo_restore_native_cursor';
+
+	/** Option: entity → DB time its first (possibly interrupted) run started. */
+	public const STARTED_OPTION = 'tmdo_restore_native_started';
+
+	/**
+	 * Restore one entity type (or, with $dry_run, count the rows involved).
+	 *
+	 * @param string $entity_type 'post' | 'user' | 'term' | 'comment'.
+	 * @param bool   $dry_run     Count only; write nothing.
+	 * @param int    $batch       Rows per batch.
+	 * @param bool   $demote      After a complete run, switch the entity to dual_write.
+	 * @return array{groups: array<string, array{rows:int, written:int}>, demoted: bool}
+	 */
+	public static function run( string $entity_type, bool $dry_run = false, int $batch = 500, bool $demote = false ): array {
+		global $wpdb;
+
+		$report  = array(
+			'groups'  => array(),
+			'demoted' => false,
+		);
+		$adapter = TMDO_Entity_Registry::get_adapter( $entity_type );
+		if ( ! $adapter ) {
+			return $report;
+		}
+		$id_col = $adapter->get_entity_id_column();
+		$batch  = max( 1, $batch );
+
+		$started = get_option( self::STARTED_OPTION, array() );
+		$started = is_array( $started ) ? $started : array();
+		if ( ! $dry_run && ! isset( $started[ $entity_type ] ) ) {
+			$started[ $entity_type ] = (string) $wpdb->get_var( 'SELECT NOW()' );
+			update_option( self::STARTED_OPTION, $started, false );
+		}
+		$cursor = get_option( self::CURSOR_OPTION, array() );
+		$cursor = is_array( $cursor ) ? $cursor : array();
+
+		$tables = self::group_tables( $entity_type );
+		foreach ( $tables as $group => $table ) {
+			$entry = array(
+				'rows'    => 0,
+				'written' => 0,
+			);
+			$key   = "{$entity_type}/{$group}";
+			$last  = $dry_run ? 0 : (int) ( $cursor[ $key ] ?? 0 );
+			do {
+				$rows           = self::fetch( $entity_type, $table, $id_col, "t.`{$id_col}` > %d ORDER BY t.`{$id_col}` LIMIT %d", array( $last, $batch ) );
+				$fetched        = count( $rows );
+				$entry['rows'] += $fetched;
+				if ( $fetched ) {
+					$last = (int) end( $rows )[ $id_col ];
+					if ( ! $dry_run ) {
+						$entry['written'] += self::write_rows( $entity_type, $group, $adapter, $rows );
+						$cursor[ $key ]    = $last;
+						update_option( self::CURSOR_OPTION, $cursor, false );
+					}
+				}
+			} while ( $fetched === $batch );
+
+			$report['groups'][ $group ] = $entry;
+		}
+
+		if ( $dry_run ) {
+			return $report;
+		}
+
+		// Catch-up: writes that landed in the flat tables after a batch passed their row.
+		foreach ( $tables as $group => $table ) {
+			$rows = self::fetch( $entity_type, $table, $id_col, 't.updated_at >= %s', array( $started[ $entity_type ] ) );
+			self::write_rows( $entity_type, $group, $adapter, $rows );
+		}
+
+		foreach ( array_keys( $tables ) as $group ) {
+			unset( $cursor[ "{$entity_type}/{$group}" ] );
+		}
+		unset( $started[ $entity_type ] );
+		$cursor ? update_option( self::CURSOR_OPTION, $cursor, false ) : delete_option( self::CURSOR_OPTION );
+		$started ? update_option( self::STARTED_OPTION, $started, false ) : delete_option( self::STARTED_OPTION );
+
+		if ( $demote ) {
+			$report['demoted'] = true === TMDO_Mode_Manager::set( $entity_type, TMDO_Mode_Manager::MODE_DUAL_WRITE );
+		}
+
+		return $report;
+	}
+
+	/**
+	 * Registered groups of the entity whose flat table exists.
+	 *
+	 * @param string $entity_type Entity type.
+	 * @return array<string, string> Group → table name.
+	 */
+	private static function group_tables( string $entity_type ): array {
+		$tables = array();
+		foreach ( TMDO_Entity_Registry::get_groups_for_type( $entity_type ) as $group ) {
+			$table = TMDO_Schema_Manager::get_table_name( $entity_type, $group );
+			if ( TMDO_Schema_Manager::table_exists( $table ) ) {
+				$tables[ $group ] = $table;
+			}
+		}
+		return $tables;
+	}
+
+	/**
+	 * Flat rows matching $where; post rows carry their post type as __post_type.
+	 *
+	 * @param string $entity_type Entity type.
+	 * @param string $table       Flat table.
+	 * @param string $id_col      Entity id column.
+	 * @param string $where       WHERE clause with placeholders.
+	 * @param array  $params      Placeholder values.
+	 * @return array<int, array<string, mixed>>
+	 */
+	private static function fetch( string $entity_type, string $table, string $id_col, string $where, array $params ): array {
+		global $wpdb;
+		$select = 'post' === $entity_type
+			? "SELECT t.*, p.post_type AS __post_type FROM `{$table}` t LEFT JOIN `{$wpdb->posts}` p ON p.ID = t.`{$id_col}`"
+			: "SELECT t.* FROM `{$table}` t";
+
+		return (array) $wpdb->get_results( $wpdb->prepare( "{$select} WHERE {$where}", ...$params ), ARRAY_A );
+	}
+
+	/**
+	 * Replace the native meta of the keys this group owns for each row.
+	 *
+	 * @param string                        $entity_type Entity type.
+	 * @param string                        $group       Group the rows come from.
+	 * @param TMDO_Entity_Adapter_Interface $adapter     Entity adapter.
+	 * @param array                         $rows        Flat rows.
+	 * @return int Native values written.
+	 */
+	private static function write_rows( string $entity_type, string $group, TMDO_Entity_Adapter_Interface $adapter, array $rows ): int {
+		global $wpdb;
+		if ( empty( $rows ) ) {
+			return 0;
+		}
+		$meta_table = $adapter->get_native_meta_table();
+		$id_col     = $adapter->get_entity_id_column();
+
+		$deletes = array();
+		$inserts = array();
+		foreach ( $rows as $row ) {
+			$id        = (int) $row[ $id_col ];
+			$post_type = 'post' === $entity_type ? (string) ( $row['__post_type'] ?? '' ) : '';
+			foreach ( TMDO_Entity_Registry::get_group_fields( $entity_type, $group ) as $field ) {
+				$owner = TMDO_Entity_Registry::get_field( $entity_type, $field['key'], $post_type );
+				$col   = TMDO_Schema_Manager::sanitize_column_name( $field['key'] );
+				if ( null === $owner || $owner['group'] !== $group || ! array_key_exists( $col, $row ) ) {
+					continue;
+				}
+				$deletes[ $field['key'] ][] = $id;
+				if ( null !== $row[ $col ] ) {
+					$inserts[] = array( $id, $field['key'], (string) maybe_serialize( TMDO_Type_Caster::from_db( $row[ $col ], $field ) ) );
+				}
+			}
+		}
+
+		foreach ( $deletes as $meta_key => $ids ) {
+			$in = implode( ',', array_map( 'intval', $ids ) );
+			$wpdb->query( $wpdb->prepare( "DELETE FROM `{$meta_table}` WHERE meta_key = %s AND `{$id_col}` IN ({$in})", $meta_key ) );
+		}
+		if ( $inserts ) {
+			$values = implode( ', ', array_fill( 0, count( $inserts ), '(%d, %s, %s)' ) );
+			$wpdb->query( $wpdb->prepare( "INSERT INTO `{$meta_table}` (`{$id_col}`, meta_key, meta_value) VALUES {$values}", ...array_merge( ...$inserts ) ) );
+		}
+
+		foreach ( array_unique( array_map( static fn( $row ) => (int) $row[ $id_col ], $rows ) ) as $id ) {
+			wp_cache_delete( $id, "{$entity_type}_meta" );
+		}
+		return count( $inserts );
+	}
+}

tests/bootstrap.php

@@ -643,6 +643,7 @@ require_once TMDO_PATH . 'includes/migration/phases/class-tmdo-phase-completed.p
 require_once TMDO_PATH . 'includes/migration/class-tmdo-migration-orchestrator.php';
 require_once TMDO_PATH . 'includes/migration/class-tmdo-post-migration.php';
 require_once TMDO_PATH . 'includes/migration/class-tmdo-post-group-reconciler.php';
+require_once TMDO_PATH . 'includes/migration/class-tmdo-native-restorer.php';
 
 // Options Manager.
 require_once TMDO_PATH . 'modules/options/class-tmdo-options-manager.php';

tests/integration/NativeRestorerPostTest.php

@@ -0,0 +1,91 @@
+<?php
+declare(strict_types=1);
+
+use PHPUnit\Framework\TestCase;
+
+/**
+ * Post write-back follows the same routing reads use.
+ *
+ * Listing 101 has `shared_key` in both group tables: 'L' in its own
+ * (hp_listing-scoped) group, a stray 'V' in the vendor group — the legacy
+ * owner, registered last. Whichever table reads use must be the one copied.
+ */
+class NativeRestorerPostTest extends TestCase {
+
+	private const LISTING = 'wp_itest_wpdo_post_nr_listing';
+	private const VENDOR  = 'wp_itest_wpdo_post_nr_vendor';
+
+	public static function setUpBeforeClass(): void {
+		$base = WPDO_PLUGIN_DIR;
+		foreach ( array(
+			'includes/adapters/interface-entity-adapter.php',
+			'includes/engine/class-tmdo-type-caster.php',
+			'includes/engine/class-tmdo-schema-manager.php',
+			'includes/engine/class-tmdo-entity-registry.php',
+			'includes/engine/class-tmdo-mode-manager.php',
+			'includes/engine/class-tmdo-cache-orchestrator.php',
+			'includes/adapters/class-tmdo-adapter-post.php',
+			'includes/migration/class-tmdo-native-restorer.php',
+		) as $file ) {
+			require_once $base . $file;
+		}
+	}
+
+	protected function setUp(): void {
+		global $wpdb;
+		self::drop();
+		$wpdb->query( "CREATE TABLE `{$wpdb->posts}` ( ID BIGINT UNSIGNED NOT NULL, post_type VARCHAR(20) NOT NULL DEFAULT 'post', PRIMARY KEY (ID) ) ENGINE=InnoDB" );
+		$wpdb->query( "CREATE TABLE `{$wpdb->postmeta}` ( meta_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, post_id BIGINT UNSIGNED NOT NULL DEFAULT 0, meta_key VARCHAR(255) DEFAULT NULL, meta_value LONGTEXT, PRIMARY KEY (meta_id), KEY post_id (post_id), KEY meta_key (meta_key(191)) ) ENGINE=InnoDB" );
+		foreach ( array( self::LISTING, self::VENDOR ) as $table ) {
+			$wpdb->query( "CREATE TABLE `{$table}` ( id BIGINT NOT NULL AUTO_INCREMENT, post_id BIGINT NOT NULL, shared_key VARCHAR(255) DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT '2020-01-01 00:00:00', PRIMARY KEY (id), UNIQUE KEY uk_entity (post_id) ) ENGINE=InnoDB" );
+		}
+		$wpdb->insert( $wpdb->posts, array( 'ID' => 101, 'post_type' => 'hp_listing' ) );
+		$wpdb->insert( self::LISTING, array( 'post_id' => 101, 'shared_key' => 'L' ) );
+		$wpdb->insert( self::VENDOR, array( 'post_id' => 101, 'shared_key' => 'V' ) );
+
+		$GLOBALS['_wp_cache']      = array();
+		$GLOBALS['_wp_transients'] = array();
+
+		WPDO_Entity_Registry::init();
+		WPDO_Entity_Registry::register_adapter( 'post', new WPDO_Adapter_Post() );
+		WPDO_Entity_Registry::register_group( 'post', 'nr_listing', array( array( 'key' => 'shared_key', 'type' => 'text' ) ), array( 'post_types' => array( 'hp_listing' ) ) );
+		WPDO_Entity_Registry::register_group( 'post', 'nr_vendor', array( array( 'key' => 'shared_key', 'type' => 'text' ) ), array( 'post_types' => array( 'hp_vendor' ) ) );
+
+		$ref = new ReflectionProperty( WPDO_Schema_Manager::class, 'table_exists_cache' );
+		$ref->setAccessible( true );
+		$ref->setValue( null, array() );
+	}
+
+	protected function tearDown(): void {
+		self::drop();
+		WPDO_Entity_Registry::init();
+	}
+
+	private static function drop(): void {
+		global $wpdb;
+		foreach ( array( $wpdb->posts, $wpdb->postmeta, self::LISTING, self::VENDOR ) as $table ) {
+			$wpdb->query( "DROP TABLE IF EXISTS `{$table}`" );
+		}
+	}
+
+	private function native_value(): array {
+		global $wpdb;
+		return $wpdb->get_col( "SELECT meta_value FROM `{$wpdb->postmeta}` WHERE post_id = 101 AND meta_key = 'shared_key'" );
+	}
+
+	public function test_scoped_routing_writes_back_the_post_types_own_group(): void {
+		$GLOBALS['_wp_options'] = array( WPDO_Entity_Registry::SCOPED_ROUTING_OPTION => '1' );
+
+		TMDO_Native_Restorer::run( 'post' );
+
+		$this->assertSame( array( 'L' ), $this->native_value() );
+	}
+
+	public function test_legacy_routing_writes_back_the_legacy_owner(): void {
+		$GLOBALS['_wp_options'] = array( WPDO_Entity_Registry::SCOPED_ROUTING_OPTION => '0' );
+
+		TMDO_Native_Restorer::run( 'post' );
+
+		$this->assertSame( array( 'V' ), $this->native_value() );
+	}
+}

tests/integration/NativeRestorerTest.php

@@ -0,0 +1,142 @@
+<?php
+declare(strict_types=1);
+
+use PHPUnit\Framework\TestCase;
+
+/**
+ * Writing flat-table values back to native meta so an aeav_only entity can
+ * leave aeav_only.
+ *
+ * Rule: after the write-back, native meta returns what reads return today.
+ * Fixture (user entity, both groups hold `nickname`; the legacy owner is
+ * rn_extra, registered last):
+ *   rn_profile  user 1: nickname 'old', level 3, tags ["a","b"]
+ *               user 2: nickname NULL, level 5, tags NULL
+ *   rn_extra    user 1: nickname 'live'
+ *   usermeta    user 1 level '1' (stale), user 2 tags 'stale'
+ */
+class NativeRestorerTest extends TestCase {
+
+	private const PROFILE = 'wp_itest_wpdo_user_rn_profile';
+	private const EXTRA   = 'wp_itest_wpdo_user_rn_extra';
+
+	public static function setUpBeforeClass(): void {
+		$base = WPDO_PLUGIN_DIR;
+		foreach ( array(
+			'includes/adapters/interface-entity-adapter.php',
+			'includes/engine/class-tmdo-type-caster.php',
+			'includes/engine/class-tmdo-schema-manager.php',
+			'includes/engine/class-tmdo-entity-registry.php',
+			'includes/engine/class-tmdo-mode-manager.php',
+			'includes/engine/class-tmdo-cache-orchestrator.php',
+			'includes/adapters/class-tmdo-adapter-user.php',
+			'includes/migration/class-tmdo-native-restorer.php',
+		) as $file ) {
+			require_once $base . $file;
+		}
+	}
+
+	protected function setUp(): void {
+		global $wpdb;
+		self::drop();
+		$wpdb->query( "CREATE TABLE `{$wpdb->usermeta}` ( umeta_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, user_id BIGINT UNSIGNED NOT NULL DEFAULT 0, meta_key VARCHAR(255) DEFAULT NULL, meta_value LONGTEXT, PRIMARY KEY (umeta_id), KEY user_id (user_id), KEY meta_key (meta_key(191)) ) ENGINE=InnoDB" );
+		$wpdb->query( 'CREATE TABLE `' . self::PROFILE . '` ( id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT NOT NULL, nickname VARCHAR(255) DEFAULT NULL, level BIGINT(20) DEFAULT NULL, tags LONGTEXT DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_entity (user_id) ) ENGINE=InnoDB' );
+		$wpdb->query( 'CREATE TABLE `' . self::EXTRA . '` ( id BIGINT NOT NULL AUTO_INCREMENT, user_id BIGINT NOT NULL, nickname VARCHAR(255) DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_entity (user_id) ) ENGINE=InnoDB' );
+		$wpdb->query( 'INSERT INTO `' . self::PROFILE . "` (user_id, nickname, level, tags) VALUES (1, 'old', 3, '[\"a\",\"b\"]'), (2, NULL, 5, NULL)" );
+		$wpdb->insert( self::EXTRA, array( 'user_id' => 1, 'nickname' => 'live' ) );
+		$wpdb->insert( $wpdb->usermeta, array( 'user_id' => 1, 'meta_key' => 'level', 'meta_value' => '1' ) );
+		$wpdb->insert( $wpdb->usermeta, array( 'user_id' => 2, 'meta_key' => 'tags', 'meta_value' => 'stale' ) );
+		// Historical rows: older than any run, so the catch-up pass leaves them alone.
+		$wpdb->query( 'UPDATE `' . self::PROFILE . "` SET updated_at = '2020-01-01 00:00:00'" );
+		$wpdb->query( 'UPDATE `' . self::EXTRA . "` SET updated_at = '2020-01-01 00:00:00'" );
+
+		$GLOBALS['_wp_options']    = array( 'wpdo_bridge_modes' => array( 'user' => 'aeav_only' ) );
+		$GLOBALS['_wp_cache']      = array();
+		$GLOBALS['_wp_transients'] = array();
+		TMDO_Mode_Manager::reset_cache();
+
+		WPDO_Entity_Registry::init();
+		WPDO_Entity_Registry::register_adapter( 'user', new WPDO_Adapter_User() );
+		WPDO_Entity_Registry::register_group(
+			'user',
+			'rn_profile',
+			array(
+				array( 'key' => 'nickname', 'type' => 'text' ),
+				array( 'key' => 'level', 'type' => 'integer' ),
+				array( 'key' => 'tags', 'type' => 'json' ),
+			)
+		);
+		WPDO_Entity_Registry::register_group( 'user', 'rn_extra', array( array( 'key' => 'nickname', 'type' => 'text' ) ) );
+
+		$ref = new ReflectionProperty( WPDO_Schema_Manager::class, 'table_exists_cache' );
+		$ref->setAccessible( true );
+		$ref->setValue( null, array() );
+	}
+
+	protected function tearDown(): void {
+		self::drop();
+		WPDO_Entity_Registry::init();
+		TMDO_Mode_Manager::reset_cache();
+	}
+
+	private static function drop(): void {
+		global $wpdb;
+		foreach ( array( $wpdb->usermeta, self::PROFILE, self::EXTRA ) as $table ) {
+			$wpdb->query( "DROP TABLE IF EXISTS `{$table}`" );
+		}
+	}
+
+	/** @return string[] meta_value rows for (user, key). */
+	private function native( int $user_id, string $key ): array {
+		global $wpdb;
+		return $wpdb->get_col( $wpdb->prepare( "SELECT meta_value FROM `{$wpdb->usermeta}` WHERE user_id = %d AND meta_key = %s", $user_id, $key ) );
+	}
+
+	public function test_dry_run_counts_without_writing(): void {
+		$report = TMDO_Native_Restorer::run( 'user', true );
+
+		$this->assertSame( 2, $report['groups']['rn_profile']['rows'] );
+		$this->assertSame( 1, $report['groups']['rn_extra']['rows'] );
+		$this->assertSame( array( '1' ), $this->native( 1, 'level' ) );
+		$this->assertFalse( $report['demoted'] );
+	}
+
+	public function test_native_meta_ends_up_matching_current_reads(): void {
+		TMDO_Native_Restorer::run( 'user', false, 1 );
+
+		$this->assertSame( array( 'live' ), $this->native( 1, 'nickname' ), 'the group reads use (rn_extra) wins' );
+		$this->assertSame( array( '3' ), $this->native( 1, 'level' ), 'stale native value replaced' );
+		$this->assertSame( array( serialize( array( 'a', 'b' ) ) ), $this->native( 1, 'tags' ), 'json stored the way WP stores arrays' );
+		$this->assertSame( array( '5' ), $this->native( 2, 'level' ) );
+		$this->assertSame( array(), $this->native( 2, 'tags' ), 'NULL flat value removes the stale native row' );
+		$this->assertSame( array(), $this->native( 2, 'nickname' ) );
+	}
+
+	public function test_completion_clears_progress_and_demote_leaves_aeav_only(): void {
+		$report = TMDO_Native_Restorer::run( 'user', false, 500, true );
+
+		$this->assertTrue( $report['demoted'] );
+		$this->assertSame( 'dual_write', TMDO_Mode_Manager::get( 'user' ) );
+		$this->assertFalse( get_option( TMDO_Native_Restorer::CURSOR_OPTION ) );
+		$this->assertFalse( get_option( TMDO_Native_Restorer::STARTED_OPTION ) );
+	}
+
+	public function test_resumes_after_the_stored_cursor(): void {
+		update_option( TMDO_Native_Restorer::CURSOR_OPTION, array( 'user/rn_profile' => 1 ) );
+
+		TMDO_Native_Restorer::run( 'user', false );
+
+		$this->assertSame( array( '1' ), $this->native( 1, 'level' ), 'user 1 was done before the interruption' );
+		$this->assertSame( array( '5' ), $this->native( 2, 'level' ) );
+	}
+
+	public function test_catch_up_copies_rows_written_during_the_run(): void {
+		global $wpdb;
+		$wpdb->query( 'UPDATE `' . self::PROFILE . "` SET level = 9, updated_at = NOW() + INTERVAL 1 HOUR WHERE user_id = 1" );
+		update_option( TMDO_Native_Restorer::CURSOR_OPTION, array( 'user/rn_profile' => 1 ) );
+
+		TMDO_Native_Restorer::run( 'user', false );
+
+		$this->assertSame( array( '9' ), $this->native( 1, 'level' ) );
+	}
+}

tests/integration/bootstrap.php

@@ -603,6 +603,7 @@ require_once TMDO_PATH . 'includes/migration/phases/class-tmdo-phase-completed.p
 require_once TMDO_PATH . 'includes/migration/class-tmdo-migration-orchestrator.php';
 require_once TMDO_PATH . 'includes/migration/class-tmdo-post-migration.php';
 require_once TMDO_PATH . 'includes/migration/class-tmdo-post-group-reconciler.php';
+require_once TMDO_PATH . 'includes/migration/class-tmdo-native-restorer.php';
 require_once TMDO_PATH . 'modules/options/class-tmdo-options-manager.php';
 require_once TMDO_PATH . 'includes/class-tmdo-postmeta-cleaner.php';
 require_once TMDO_PATH . 'includes/class-tmdo-termmeta-cleaner.php';