<?php
/**
 * Database importer, receives chunks and inserts them into the local DB.
 *
 * @package Shiftora\Migration
 */

declare(strict_types=1);

namespace Shiftora\Migration;

use Shiftora\Support\Db\PeerSql;
use Shiftora\Support\Db\PeerSqlRefused;
use Shiftora\Support\Db\PrefixedKeys;
use Shiftora\Support\Db\ServerDdl;
use Shiftora\Support\Db\SqlLiteral;
use Shiftora\Support\Db\TableDdl;
use Shiftora\Support\SessionSqlMode;
use Shiftora\Support\TableNameScoper;
use Shiftora\Logging\MigrationLogger;
use RuntimeException;

// phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages are not rendered as HTML output.

// Prevent direct file access.
if ( ! defined( 'ABSPATH' ) ) {
	exit;
}

/**
 * Receives exported chunk payloads, applies find/replace rules and inserts rows.
 *
 * Staging mode (the default) writes to `shiftora_stg_*` tables and leaves the real
 * tables untouched until finalize_staging() renames them in; rollback_staging()
 * drops them on failure.
 *
 * Any `*options` table keeps a set of local runtime and Shiftora rows through an
 * import. Otherwise siteurl/home would take the remote URL and the local
 * encryption key would be lost mid-migration.
 */
class DatabaseImporter {

	/** Explicitly listed option names are always kept local (see ProtectedOptions). */
	private const PROTECTED_OPTIONS = ProtectedOptions::NAMES;

	/**
	 * Shiftora's own tables, never overwritten by a migration.
	 *
	 * The exporter filters them at the source; this is the second check at the
	 * destination. They hold install-specific data that cannot come from another site.
	 */
	private const SKIPPED_TABLES = [
		'shiftora_connections',
		'shiftora_migrations',
		'shiftora_migration_logs',
		'shiftora_logs',
		'shiftora_backups',
		'shiftora_import_chunks',
	];

	public function __construct(
		private readonly \wpdb $wpdb
	) {}

	/**
	 * Token-scoped staging name for a bare table.
	 *
	 * Public so the file-import endpoint targets the same tables that
	 * finalize_staging() and rollback_staging() later rename or drop.
	 */
	public function staging_name_for( string $bare, string $token ): string {
		return $this->staging_table_name( $bare, $token );
	}

	/**
	 * Whether a bare table name is Shiftora infrastructure that must never be imported.
	 */
	public function is_skipped_table( string $bare ): bool {
		return in_array( $bare, self::SKIPPED_TABLES, true );
	}

	/**
	 * Re-cut one exported multi-row INSERT into statements no longer than $limit.
	 *
	 * Only safe on exporter output: the head is the first line, each row group is its
	 * own line joined by ",\n", and newlines inside values are always escaped.
	 *
	 * @return list<string>
	 */
	private static function split_insert( string $stmt, int $limit, string $bare_table ): array {
		$break = strpos( $stmt, "\n" );
		if ( false === $break || ! str_starts_with( strtoupper( $stmt ), 'INSERT INTO' ) ) {
			throw new RuntimeException(
				"SQL statement for table {$bare_table} is larger than this server's max_allowed_packet ({$limit} bytes) and cannot be split."
			);
		}

		$head   = substr( $stmt, 0, $break + 1 );
		$pieces = [];
		$batch  = [];
		$size   = strlen( $head );

		foreach ( explode( ",\n", substr( $stmt, $break + 1 ) ) as $row ) {
			if ( strlen( $head ) + strlen( $row ) > $limit ) {
				throw new RuntimeException( sprintf(
					'A row in table %s is %d bytes, more than this server\'s max_allowed_packet of %d bytes. Raise max_allowed_packet on this server and run the migration again.',
					$bare_table,
					strlen( $row ),
					$limit
				) );
			}

			if ( [] !== $batch && $size + 2 + strlen( $row ) > $limit ) {
				$pieces[] = $head . implode( ",\n", $batch );
				$batch    = [];
				$size     = strlen( $head );
			}

			$batch[] = $row;
			$size   += strlen( $row ) + 2;
		}

		if ( [] !== $batch ) {
			$pieces[] = $head . implode( ",\n", $batch );
		}

		return $pieces;
	}

	/** Cached max_allowed_packet value (bytes). Fetched once per importer instance. */
	private ?int $max_packet_bytes = null;

	/**
	 * Server max_allowed_packet in bytes, cached per instance. Falls back to 4 MB.
	 */
	private function server_max_packet_bytes(): int {
		if ( null === $this->max_packet_bytes ) {
			$raw                    = (int) $this->wpdb->get_var( 'SELECT @@max_allowed_packet' );
			$this->max_packet_bytes = $raw > 0 ? $raw : ( 4 * 1024 * 1024 );
		}
		return $this->max_packet_bytes;
	}

	/**
	 * Import a gzipped raw-SQL chunk, skipping JSON encoding.
	 *
	 * `__TARGET__` in the SQL is swapped for the real staging (or local) table name.
	 * The first chunk also creates the staging table from $schema.
	 *
	 * @param string      $sql_gz        Base64-encoded gzipped SQL.
	 * @param string|null $schema        CREATE TABLE SQL (first chunk only), on `__TARGET__`.
	 * @param string|null $transfer_uuid Receiver-owned identity scoping receipt rows.
	 * @param string|null $page_identity Pagination position; scopes the replay ledger so
	 *                                   pages with identical content do not alias.
	 * @param string      $source_prefix The other site's table prefix, which a foreign key may reference.
	 *
	 * @return int Rows (or simulated rows in dry-run mode).
	 *
	 * @throws PeerSqlRefused When the payload holds anything but row inserts and one table definition.
	 * @throws RuntimeException On SQL execution failure.
	 */
	public function import_sql_chunk(
		string $sql_gz,
		string $bare_table,
		?string $schema = null,
		bool $use_staging = true,
		bool $dry_run = false,
		int $migration_id = 0,
		string $staging_token = '',
		?string $transfer_uuid = null,
		?string $page_identity = null,
		string $source_prefix = ''
	): int {
		if ( $dry_run ) {
			// Decode to validate the payload is well-formed without executing it.
			$decoded = base64_decode( $sql_gz, true );
			if ( false === $decoded ) {
				throw new RuntimeException( 'Failed to decode base64 SQL payload.' );
			}
			// Malformed payloads are the point of this dry-run guard. gzuncompress
			// warns on garbage input, which the throw below already surfaces.
			$raw_sql = @gzuncompress( $decoded ); // phpcs:ignore WordPress.PHP.NoSilencedErrors -- Probes a possibly corrupt gzip payload; a failed decompress returns false and is rejected below.
			if ( false === $raw_sql ) {
				throw new RuntimeException( 'Failed to decompress gzipped SQL payload.' );
			}
			$statements = array_filter(
				array_map( 'trim', explode( ";\n", $raw_sql ) ),
				static fn( string $s ): bool => '' !== $s
			);
			// Count INSERT statements as simulated rows.
			return count( array_filter(
				$statements,
				static fn( string $s ): bool => str_starts_with( strtoupper( $s ), 'INSERT' )
			) );
		}
		$local_table   = $this->wpdb->prefix . $bare_table;
		$staging_table = $this->staging_table_name( $bare_table, $staging_token );
		$target_table  = $use_staging ? $staging_table : $local_table;

		if ( in_array( $bare_table, self::SKIPPED_TABLES, true ) ) {
			return 0;
		}

		// Nothing from the other site reaches the database before it has been read: the
		// schema and every statement must be a row insert or table definition.
		$decoded = base64_decode( $sql_gz, true );
		if ( false === $decoded ) {
			throw new RuntimeException( 'Failed to decode base64 SQL payload.' );
		}

		$raw_sql = gzuncompress( $decoded );
		if ( false === $raw_sql ) {
			throw new RuntimeException( 'Failed to decompress gzipped SQL payload.' );
		}

		$peer_statements = [];
		$this->guard_peer_sql(
			function () use ( $raw_sql, $schema, $source_prefix, &$peer_statements ): void {
				$prefixes = $this->reference_prefixes( $source_prefix );
				if ( null !== $schema ) {
					PeerSql::assert_schema( $schema, PeerSql::TARGET, $prefixes );
				}
				$peer_statements = PeerSql::split( $raw_sql );
				foreach ( $peer_statements as $statement ) {
					PeerSql::assert_statement( $statement, PeerSql::TARGET, $prefixes );
				}
			},
			$migration_id,
			$bare_table
		);

		// Settle an exact page-0 replay before any DDL. Preparing a schema
		// rebuilds the staging table, so a page 0 that arrives again after it
		// already committed (the sender never saw the response) would drop the
		// rows later pages have staged and commit an empty table.
		if ( null !== $schema && $use_staging && $migration_id > 0 ) {
			$receipt = ( new WpChunkReceiptStore( $this->wpdb ) )->committed_rows(
				$migration_id,
				SqlImportChunkIds::chunk_key( $sql_gz, $bare_table, $page_identity ?? '' ),
				$transfer_uuid
			);
			if ( null !== $receipt ) {
				return $receipt;
			}
		}

		// Ensure the staging table exists when a schema arrived (first chunk).
		// The source schema can carry a zero-date default that a strict
		// session refuses at CREATE, and dbDelta is where that lands.
		if ( null !== $schema ) {
			$schema_mode = SessionSqlMode::relax( $this->wpdb );
			try {
				$this->preflight_schema( $schema, $bare_table );
				if ( $use_staging ) {
					$this->prepare_staging_table( $bare_table, $schema, $staging_token, $migration_id, $transfer_uuid );
				} else {
					$this->ensure_table( $local_table, $this->retarget_sql_head( $schema, $local_table ), false );
				}
			} finally {
				SessionSqlMode::restore( $this->wpdb, $schema_mode );
			}
		}

		// Redirect the __TARGET__ head. The exporter writes the placeholder in exactly
		// two statement heads (CREATE TABLE and INSERT INTO); anchoring there leaves
		// column values that contain the literal "__TARGET__" untouched.
		$statements = array_map(
			fn( string $stmt ): string => $this->retarget_sql_head( $stmt, $target_table ),
			$peer_statements
		);

		// A statement over the receiver's max_allowed_packet fails the whole query. The
		// sender can't see this server's limit, so an oversized INSERT is re-cut here.
		// Only a single row too big for the packet is refused.
		$max_packet = $this->server_max_packet_bytes();
		$fitted     = [];
		foreach ( $statements as $stmt ) {
			if ( strlen( $stmt ) + 1 <= $max_packet ) {
				$fitted[] = $stmt;
				continue;
			}
			$pieces = self::split_insert( $stmt, $max_packet - 1, $bare_table );
			// The cut is made on line breaks, which a string value may contain, so each
			// piece is read again before it can run.
			$this->guard_peer_sql(
				static function () use ( $pieces, $target_table ): void {
					foreach ( $pieces as $piece ) {
						PeerSql::assert_statement( $piece, $target_table );
					}
				},
				$migration_id,
				$bare_table
			);
			array_push( $fitted, ...$pieces );
		}
		$statements = $fitted;

		// Execute the data statements and return the row count. CREATE TABLE
		// was already handled above and is skipped here.
		$execute = function () use ( $statements, $bare_table, $raw_sql, $migration_id ): int {
			$inserted = 0;
			foreach ( $statements as $stmt ) {
				if ( str_starts_with( strtoupper( $stmt ), 'CREATE TABLE' ) ) {
					continue;
				}

				// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared, PluginCheck.Security.DirectDB.UnescapedDBParameter -- a row insert PeerSql already checked, which cannot be bound.
				$result = $this->wpdb->query( $stmt . ';' );
				if ( false === $result ) {
					throw new RuntimeException(
						"SQL import failed for table {$bare_table}: {$this->wpdb->last_error}"
					);
				}
				$inserted += max( 0, (int) $result );
			}

			return $inserted;
		};

		// A staged table already has its foreign keys when sibling rows arrive, so
		// parent-first ordering can't be relied on. Load with checks off, as a SQL dump
		// does, and bind constraints at the swap. Outside the transaction because SET
		// can't safely sit inside an atomic one.
		$fk_before = (string) ( $this->wpdb->get_var( 'SELECT @@foreign_key_checks' ) ?? '1' );
		$sql_mode_before = SessionSqlMode::relax( $this->wpdb );
		TableDdl::set_foreign_key_checks( $this->wpdb, '0' );
		try {
			// With a migration id, a committed-receipt ledger makes replay idempotent: rows
			// and receipt share a transaction, so a failed chunk rolls back both and a retry
			// runs it exactly once.
			if ( $migration_id > 0 ) {
				$chunk_key = SqlImportChunkIds::chunk_key( $sql_gz, $bare_table, $page_identity ?? '' );
				$store     = new WpChunkReceiptStore( $this->wpdb );
				$guard     = new SqlChunkReplayGuard( $store );

				// Provision the ledger before opening the transaction. MySQL
				// implicitly commits at CREATE TABLE, so DDL inside the
				// transaction would end it and turn the later ROLLBACK into a
				// no-op that leaves imported rows behind.
				$store->ensure_table();

				// wpdb returns the affected row count (0) for a successful
				// START TRANSACTION, so only a hard false means it failed.
				if ( false === $this->wpdb->query( 'START TRANSACTION' ) ) {
					// Without a transaction the data and the receipt would commit
					// separately and a failed half-import could not be rolled
					// back, so fail the chunk instead of running unprotected.
					throw new RuntimeException(
						"SQL import transaction could not be started for table {$bare_table}: {$this->wpdb->last_error}"
					);
				}
				$in_tx = true;

				try {
					$rows = $guard->run( $migration_id, $chunk_key, $bare_table, $execute, $transfer_uuid );
					if ( $in_tx ) {
						// A lost COMMIT must not report success with the receipt
						// uncommitted; the error path rolls back what it can and the
						// retry re-runs the chunk.
						if ( false === $this->wpdb->query( 'COMMIT' ) ) {
							throw new RuntimeException(
								"SQL import commit failed for table {$bare_table}: {$this->wpdb->last_error}"
							);
						}
					}

					return $rows;
				} catch ( \Throwable $e ) {
					if ( $in_tx ) {
						try {
							$this->wpdb->query( 'ROLLBACK' );
						} catch ( \Throwable $rollback_error ) {
							// Best-effort rollback; the guard still clears the pending
							// receipt so a retry re-runs the chunk.
						}
					}
					throw $e;
				}
			}

			return $execute();
		} finally {
			TableDdl::set_foreign_key_checks( $this->wpdb, $fk_before );
			SessionSqlMode::restore( $this->wpdb, $sql_mode_before );
		}
	}

	/**
	 * Atomically swap all staging tables into production.
	 *
	 * One `RENAME TABLE` covers every pair, so a failure cannot leave a partial swap.
	 * A commit receipt makes a retry after a completed rename a no-op.
	 *
	 * With a `$verification` snapshot (leased push), the staging set must match the
	 * claimed manifest and recorded schemas before any rename runs. The return value
	 * is the audit context the endpoint persists as the commit receipt.
	 *
	 * @param string[] $bare_tables   Bare table names (without DB prefix).
	 * @param int      $migration_id  When > 0 a commit receipt is recorded and consulted on retry.
	 * @param string   $src_prefix    Source prefix; if it differs from the local one, role keys
	 *                                on staging *_options / *_usermeta tables are rewritten first.
	 * @param array{manifest: string[], schemas: array<string, string>}|null $verification
	 *
	 * @return array{renamed: array<int, array{bare: string, local: string, staging: string, old: null|string}>, verified: array<string, int>}
	 *
	 * @throws RuntimeException If a staging table is absent, verification fails or the RENAME fails.
	 */
	public function finalize_staging(
		array $bare_tables,
		int $migration_id = 0,
		string $src_prefix = '',
		string $staging_token = '',
		?array $verification = null,
		?array $commit_intent = null,
		?string $transfer_uuid = null,
		?array $staging_plan = null,
		?array $foreign_owned = null
	): array {
		// The source prefix travelled over the wire and only word-format
		// characters are legal in a table prefix. Anything else would silently
		// disable the reconciliation UPDATEs below.
		$src_prefix = (string) preg_replace( '/[^a-zA-Z0-9_]/', '', $src_prefix );

		$receipt_store = null;
		$verified      = [];
		if ( $migration_id > 0 ) {
			$receipt_store = new WpChunkReceiptStore( $this->wpdb );

			// A commit receipt means a completed attempt already renamed the staging tables
			// away, so their absence is expected. Rebuild the verified set from recorded
			// schemas. Leased transfers scope the marker to their identity so another
			// transfer reusing the numeric id can't satisfy it.
			if ( $receipt_store->committed_rows( $migration_id, WpChunkReceiptStore::FINALIZE_MARKER, $transfer_uuid ) !== null ) {
				return $this->receipt_context(
					$bare_tables,
					$staging_token,
					$this->verified_from_schemas( $verification )
				);
			}
		}

		// Pre-rename gate for leased pushes: the staging set must match the claim
		// manifest and every table its applied schema, or nothing is renamed. Unleased
		// transfers have no snapshot to compare against.
		if ( null !== $verification ) {
			$verified = $this->verify_staging_set( $bare_tables, $staging_token, $verification, $staging_plan, $foreign_owned ?? [] );
		}

		$rename_pairs = [];
		$missing      = [];
		$renamed      = [];

		// The durable commit intent (when provided) is the source of truth for
		// the rename shape: which destination existed and the checkpoint name
		// the swap assigns it. Trusting it over a re-read keeps the executed
		// plan identical to the one recorded on the lease before the flip.
		$intent_for = [];
		if ( null !== $commit_intent ) {
			foreach ( $commit_intent as $entry ) {
				if ( is_array( $entry ) && isset( $entry['bare'] ) ) {
					$intent_for[ (string) $entry['bare'] ] = $entry;
				}
			}
		}

		foreach ( $bare_tables as $bare ) {
			// Shiftora's own infrastructure tables are never staged.
			if ( in_array( $bare, self::SKIPPED_TABLES, true ) ) {
				continue;
			}

			$local   = $this->wpdb->prefix . $bare;
			$staging = $this->staging_table_name( $bare, $staging_token );
			$old     = $this->old_table_name( $bare, $staging_token );
			$entry   = $intent_for[ $bare ] ?? null;

			if ( ! TableDdl::exists( $this->wpdb, $staging ) ) {
				// The exporter always sends a CREATE TABLE schema on the first
				// chunk, so an absent staging table means the import never
				// created it. Fail instead of reporting a table as finalized
				// that was never replaced.
				$missing[] = $bare;
				continue;
			}

				// Rewrite prefix-bound option and usermeta keys when prefixes differ, so
				// imported users keep roles and settings. Runs before the protected-option
				// restore, which upserts so the destination's own runtime state wins.
			if ( '' !== $src_prefix && $src_prefix !== $this->wpdb->prefix ) {
				$this->reconcile_prefixed_keys( $staging, $bare, $src_prefix );
			}

			// Inject local protected rows into *_options staging tables before
			// the rename, so siteurl/home survive the swap.
			if ( str_ends_with( $bare, 'options' ) ) {
				if ( $this->intent_says_live( $entry, $local ) ) {
					$snapshot = $this->snapshot_protected_options( $local );
					if ( ! empty( $snapshot ) ) {
						$this->restore_protected_options( $staging, $snapshot );
					}
				}
			}

			// Build the rename pair. The intent's existed flag (when present)
			// decides the swap shape; skipping a re-read keeps the plan from
			// drifting from what was recorded before the commit lock.
			$local_exists = $this->intent_says_live( $entry, $local );
			$old_name     = is_array( $entry ) && isset( $entry['old'] ) && is_string( $entry['old'] ) && '' !== $entry['old']
				? $entry['old']
				: $old;

			if ( $local_exists ) {
				// Drop any previous shiftora_old_* table so the rename pair won't fail.
				$this->drop_foreign_key_dependent_table( $old_name );
				// Two-step swap: move live table to _old, move staging to live.
				$rename_pairs[] = [ $local, $old_name ];
				$rename_pairs[] = [ $staging, $local ];
				$renamed[]      = [
					'bare'    => $bare,
					'local'   => $local,
					'staging' => $staging,
					'old'     => $old_name,
					'existed' => true,
				];
			} else {
				// New table that did not exist locally: just rename staging.
				$rename_pairs[] = [ $staging, $local ];
				$renamed[]      = [
					'bare'    => $bare,
					'local'   => $local,
					'staging' => $staging,
					'old'     => null,
					'existed' => false,
				];
			}
		}

		if ( ! empty( $missing ) ) {
			throw new RuntimeException(
				'Finalization aborted: staging table(s) missing for: ' . implode( ', ', $missing )
			);
		}

		$finalized_count = count( $bare_tables );

		if ( empty( $rename_pairs ) ) {
			// Nothing was renamed (e.g. only Shiftora's own tables selected).
			// Record the receipt anyway so a retry is accepted as completed.
			if ( $receipt_store ) {
				$receipt_store->set_committed( $migration_id, WpChunkReceiptStore::FINALIZE_MARKER, $finalized_count, $transfer_uuid );
			}
			return [ 'renamed' => $renamed, 'verified' => $verified ];
		}

		// One atomic RENAME TABLE across all pairs.
		$result = TableDdl::rename( $this->wpdb, $rename_pairs );

		if ( false === $result ) {
			throw new RuntimeException(
				"Atomic table finalization failed: {$this->wpdb->last_error}"
			);
		}

		// Table constraints arrive scoped so staging could hold them, and the
		// swap has just put the incoming table under its real name. Rename them
		// back now, while the live name is the table meant to keep them and
		// every referenced table is already in place.
		$freed = $this->apply_constraints( $renamed, $src_prefix );

		// Record the commit receipt so a retry is provably already-completed.
		if ( $receipt_store ) {
			$receipt_store->set_committed( $migration_id, WpChunkReceiptStore::FINALIZE_MARKER, $finalized_count, $transfer_uuid );
		}
		// The shiftora_old_* tables are kept as the rollback checkpoint until
		// complete_migration() confirms the migration and calls drop_old_tables().

		return [ 'renamed' => $renamed, 'verified' => $verified, 'freed_constraints' => $freed ];
	}

	/**
	 * Put the source's constraint names back on the tables just swapped in.
	 *
	 * Staging holds the definitions under a reserved suffix (see
	 * SchemaFingerprint::scope_constraint()), so push, pull and SQL import all
	 * restore them the same way without a schema payload at commit time.
	 *
	 * @param array<int, array{bare: string, local: string, old?: ?string}> $renamed
	 *
	 * @return array<string, string[]> Bare table => checkpoint clauses that gave up their names.
	 */
	private function apply_constraints( array $renamed, string $src_prefix ): array {
		$work = [];
		foreach ( $renamed as $entry ) {
			$local = (string) ( $entry['local'] ?? '' );
			if ( '' === $local ) {
				continue;
			}
			$ddl = TableDdl::create_statement( $this->wpdb, $local );
			if ( '' === $ddl ) {
				continue;
			}
			$staged = array_values( array_filter(
				SchemaFingerprint::constraint_clauses( $ddl ),
				static fn( string $clause ): bool => SchemaFingerprint::is_staged_constraint( $clause )
			) );
			if ( [] !== $staged ) {
				$work[] = [ 'entry' => $entry, 'staged' => $staged ];
			}
		}
		if ( [] === $work ) {
			return [];
		}

		$fk_before = (string) ( $this->wpdb->get_var( 'SELECT @@foreign_key_checks' ) ?? '1' );
		TableDdl::set_foreign_key_checks( $this->wpdb, '0' );

		// RENAME TABLE rewrites foreign keys aimed at a renamed parent to follow it, so a
		// staged clause that named the live parent would end up pointing at the
		// checkpoint. The map below points those targets back at the current names.
		$checkpoint_live = [];
		foreach ( $renamed as $entry ) {
			if ( ! empty( $entry['old'] ) && is_string( $entry['old'] ) && ! empty( $entry['local'] ) ) {
				$checkpoint_live[ $entry['old'] ] = (string) $entry['local'];
			}
		}

		$freed = [];
		try {
			foreach ( $work as $item ) {
				$entry = $item['entry'];
				$own   = array_map(
					static fn( string $clause ): string => SchemaFingerprint::unscope_constraint( $clause ),
					$item['staged']
				);

				// The checkpoint carried these names before the swap. Freeing
				// them lets the incoming table keep the source's own constraint
				// names instead of a scoped stand-in.
				if ( ! empty( $entry['old'] ) && is_string( $entry['old'] ) ) {
					$got = $this->drop_named_constraints( (string) $entry['old'], $own );
					if ( [] !== $got ) {
						$freed[ (string) $entry['bare'] ] = $got;
					}
				}

				foreach ( $item['staged'] as $clause ) {
					$name = SchemaFingerprint::constraint_name( $clause );
					if ( '' === $name ) {
						continue;
					}
					if ( ! TableDdl::drop_constraint( $this->wpdb, (string) $entry['local'], $name, $clause ) ) {
						throw new RuntimeException( sprintf(
							'The transfer swapped `%s` into place but its staged constraint `%s` could not be renamed: %s',
							$entry['local'],
							$name,
							(string) $this->wpdb->last_error
						) );
					}

					$incoming = self::restore_checkpoint_reference(
						$this->retarget_reference_prefix(
							SchemaFingerprint::unscope_constraint( $clause ),
							$src_prefix
						),
						$checkpoint_live
					);
					if ( ! TableDdl::add_constraint( $this->wpdb, (string) $entry['local'], $incoming ) ) {
						throw new RuntimeException( sprintf(
							'The transfer swapped `%s` into place but its table constraint could not be added: %s',
							$entry['local'],
							(string) $this->wpdb->last_error
						) );
					}
				}
			}
		} finally {
			TableDdl::set_foreign_key_checks( $this->wpdb, $fk_before );
		}

		return $freed;
	}

	/**
	 * Drop the given constraint names from a table that still carries them.
	 *
	 * Only names colliding with an incoming constraint are looked up. The return value
	 * is the table's own definition of what was removed, not the incoming text, since
	 * rollback must restore what the destination had (ON DELETE rules and CHECK
	 * expressions can differ).
	 *
	 * @param string[] $clauses Full constraint clauses in source order.
	 *
	 * @return string[] The clauses that were dropped, in table order.
	 */
	private function drop_named_constraints( string $table, array $clauses ): array {
		$wanted = [];
		foreach ( $clauses as $clause ) {
			$name = SchemaFingerprint::constraint_name( $clause );
			if ( '' !== $name ) {
				$wanted[] = $name;
			}
		}
		if ( [] === $wanted ) {
			return [];
		}

		$own = TableDdl::create_statement( $this->wpdb, $table );
		if ( '' === $own ) {
			return [];
		}

		$dropped = [];
		foreach ( SchemaFingerprint::constraint_clauses( $own ) as $clause ) {
			$name = SchemaFingerprint::constraint_name( $clause );
			if ( '' === $name || ! in_array( $name, $wanted, true ) ) {
				continue;
			}
			if ( ! TableDdl::drop_constraint( $this->wpdb, $table, $name, $clause ) ) {
				continue;
			}
			$dropped[] = $clause;
		}

		return $dropped;
	}

	/**
	 * Point a constraint's REFERENCES at the receiver's table prefix.
	 *
	 * Without it the clause would bind to a table the transfer never created.
	 */
	private function retarget_reference_prefix( string $clause, string $src_prefix ): string {
		$local = (string) $this->wpdb->prefix;
		if ( '' === $src_prefix || $src_prefix === $local ) {
			return $clause;
		}

		return (string) preg_replace_callback(
			'/REFERENCES\s+(`[^`]+`|[a-zA-Z0-9_$]+)/i',
			static function ( array $m ) use ( $src_prefix, $local ): string {
				$backticked = str_contains( $m[1], '`' );
				$target     = trim( $m[1], '`' );
				if ( ! str_starts_with( $target, $src_prefix ) ) {
					return $m[0];
				}
				$target = $local . substr( $target, strlen( $src_prefix ) );

				return 'REFERENCES ' . ( $backticked ? "`{$target}`" : $target );
			},
			$clause
		);
	}

	/**
	 * Point a constraint back at the table the swap moved out of the way.
	 *
	 * @param array<string, string> $checkpoint_live Checkpoint name => live name.
	 */
	private static function restore_checkpoint_reference( string $clause, array $checkpoint_live ): string {
		if ( [] === $checkpoint_live ) {
			return $clause;
		}

		return (string) preg_replace_callback(
			'/REFERENCES\s+(`[^`]+`|[a-zA-Z0-9_$]+)/i',
			static function ( array $m ) use ( $checkpoint_live ): string {
				$backticked = str_contains( $m[1], '`' );
				$target     = trim( $m[1], '`' );
				if ( ! isset( $checkpoint_live[ $target ] ) ) {
					return $m[0];
				}
				$live = $checkpoint_live[ $target ];

				return 'REFERENCES ' . ( $backticked ? "`{$live}`" : $live );
			},
			$clause
		);
	}

	/**
	 * Drop a table another table may hold a foreign key against.
	 *
	 * MySQL refuses to drop a referenced table. Checkpoint pairs reference each other
	 * by name, so checks are lowered for the one statement and restored right after.
	 *
	 * @return bool Whether the statement was accepted.
	 */
	private function drop_foreign_key_dependent_table( string $table ): bool {
		$before = (string) ( $this->wpdb->get_var( 'SELECT @@foreign_key_checks' ) ?? '1' );
		TableDdl::set_foreign_key_checks( $this->wpdb, '0' );
		try {
			return false !== TableDdl::drop( $this->wpdb, $table );
		} finally {
			TableDdl::set_foreign_key_checks( $this->wpdb, $before );
		}
	}

	/**
	 * Whether a live destination should be assumed to exist for a table being
	 * finalized. The recorded intent wins; without one the disk is read once.
	 *
	 * @param array<string, mixed>|null $intent_entry
	 */
	private function intent_says_live( ?array $intent_entry, string $local ): bool {
		if ( is_array( $intent_entry ) ) {
			return (bool) ( $intent_entry['existed'] ?? false );
		}
		return TableDdl::exists( $this->wpdb, $local );
	}

	/**
	 * The renamed-list for a replay, rebuilt from bare names and the token alone.
	 *
	 * `old` is null because a backup table can't be confirmed without a disk read.
	 *
	 * @param string[]           $bare_tables
	 * @param array<string, int> $verified
	 *
	 * @return array{renamed: array<int, array{bare: string, local: string, staging: string, old: null}>, verified: array<string, int>}
	 */
	private function receipt_context( array $bare_tables, string $staging_token, array $verified ): array {
		$renamed = [];
		foreach ( $bare_tables as $bare ) {
			if ( in_array( $bare, self::SKIPPED_TABLES, true ) ) {
				continue;
			}
			$local = $this->wpdb->prefix . $bare;
			$renamed[] = [
				'bare'    => $bare,
				'local'   => $local,
				'staging' => $this->staging_table_name( $bare, $staging_token ),
				'old'     => null,
			];
		}
		return [ 'renamed' => $renamed, 'verified' => $verified ];
	}

	/**
	 * Column counts for a committed replay, from the recorded schema fingerprints and
	 * never from disk. Bad or missing schemas are excluded.
	 *
	 * @return array<string, int>
	 */
	private function verified_from_schemas( ?array $verification ): array {
		$verified = [];
		if ( null === $verification || ! isset( $verification['schemas'] ) || ! is_array( $verification['schemas'] ) ) {
			return $verified;
		}
		foreach ( $verification['schemas'] as $bare => $create ) {
			if ( ! is_string( $create ) || in_array( $bare, self::SKIPPED_TABLES, true ) ) {
				continue;
			}
			$verified[ $bare ] = SchemaFingerprint::column_count( SchemaFingerprint::fingerprint( $create ) );
		}
		return $verified;
	}

	/**
	 * Prove the staging set matches the claim-time manifest before the commit RENAME.
	 *
	 * The finalize payload, the claimed manifest and the recorded-schema tables must
	 * agree after the same SKIPPED_TABLES subtraction. Each staging table must match
	 * its recorded schema, and no unexpected table may exist in the token namespace.
	 *
	 * @param array{manifest: string[], schemas: array<string, string>} $verification
	 * @param array<string, string>|null $staging_plan Staging names per bare table; legacy
	 *                                                 rows fall back to re-derivation.
	 *
	 * @return array<string, int> Verified column counts per bare table.
	 *
	 * @throws RuntimeException On any mismatch.
	 */
	private function verify_staging_set( array $bare_tables, string $staging_token, array $verification, ?array $staging_plan = null, array $foreign_owned = [] ): array {
		$skipped    = self::SKIPPED_TABLES;
		$expected   = array_values( array_unique( array_diff( $bare_tables, $skipped ) ) );
		$manifest   = array_values( array_unique( array_diff( array_map( 'strval', $verification['manifest'] ?? [] ), $skipped ) ) );
		$schema_set = array_keys( $verification['schemas'] ?? [] );
		$schema_set = array_values( array_diff( $schema_set, $skipped ) );
		sort( $expected );
		sort( $manifest );
		sort( $schema_set );

		if ( $expected !== $manifest ) {
			throw new RuntimeException( sprintf(
				'Finalize set mismatch: payload and manifest disagree (missing: %s; unexpected: %s).',
				implode( ', ', array_diff( $manifest, $expected ) ),
				implode( ', ', array_diff( $expected, $manifest ) )
			) );
		}
		if ( $expected !== $schema_set ) {
			throw new RuntimeException( sprintf(
				'Finalize set mismatch: schema snapshots disagree (missing: %s; unexpected: %s).',
				implode( ', ', array_diff( $schema_set, $expected ) ),
				implode( ', ', array_diff( $expected, $schema_set ) )
			) );
		}

		$verified = [];
		$failures = [];
		$expected_staging = [];
		foreach ( $expected as $bare ) {
			$recorded = $staging_plan[ $bare ] ?? null;
			$staging_name = is_string( $recorded ) && '' !== $recorded
				? $recorded
				: $this->staging_table_name( $bare, $staging_token );
			$expected_staging[] = $staging_name;

			if ( ! TableDdl::exists( $this->wpdb, $staging_name ) ) {
				$failures[] = "staging table `{$staging_name}` is missing";
				continue;
			}
			if ( ! isset( $verification['schemas'][ $bare ] ) || ! is_string( $verification['schemas'][ $bare ] ) ) {
				$failures[] = "no schema snapshot for `{$bare}`";
				continue;
			}

			// Constraints are masked on both sides: the staging table is built
			// without them (the outgoing table still holds their names) and
			// they are verified against the live table after the swap.
			$expected_fp = SchemaFingerprint::without_constraints(
				SchemaFingerprint::fingerprint( $verification['schemas'][ $bare ] )
			);

			$create = TableDdl::create_statement( $this->wpdb, $staging_name ) ?: null;
			if ( null === $create ) {
				$failures[] = "could not read schema of `{$staging_name}`";
				continue;
			}
			$actual_fp = SchemaFingerprint::without_constraints( SchemaFingerprint::fingerprint( $create ) );
			if ( ! SchemaFingerprint::matches( $expected_fp, $actual_fp ) ) {
				$failures[] = sprintf(
					'`%s` schema mismatch: %s',
					$bare,
					implode( '; ', SchemaFingerprint::describe( $expected_fp, $actual_fp ) )
				);
				continue;
			}
			$verified[ $bare ] = SchemaFingerprint::column_count( $actual_fp );
		}

		// Extra tables in the shared shiftora_stg_ namespace are either a sibling
		// transfer under a live lease (ignored) or a leak from a bad import (stops the
		// commit). Scanning the whole namespace lets shortened names verify.
		$wpdb     = $this->wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- server metadata, nothing to cache
		$observed = (array) $wpdb->get_col(
			$wpdb->prepare(
				'SHOW TABLES LIKE %s',
				$wpdb->esc_like( (string) $wpdb->prefix . 'shiftora_stg_' ) . '%'
			)
		);
		$unexpected = array_map(
			static fn( string $t ): string => '`' . $t . '`',
			array_values( array_diff( $observed, $expected_staging, $foreign_owned ) )
		);
		if ( ! empty( $unexpected ) ) {
			$failures[] = 'unexpected staging table(s): ' . implode( ', ', $unexpected );
		}

		if ( ! empty( $failures ) ) {
			throw new RuntimeException( 'Staging-set verification failed: ' . implode( ' | ', $failures ) );
		}

		return $verified;
	}

	/**
	 * How many tables the commit renamed into place, or null when nothing is committed.
	 *
	 * Unlike the boolean marker, this shows the RENAME ran, not just that the worker
	 * believed it was finished.
	 *
	 * @return int|null
	 */
	public function committed_finalize_tables( int $migration_id, ?string $transfer_uuid = null ): ?int {
		if ( $migration_id <= 0 ) {
			return null;
		}
		$store = new WpChunkReceiptStore( $this->wpdb );
		return $store->committed_rows( $migration_id, WpChunkReceiptStore::FINALIZE_MARKER, $transfer_uuid );
	}

	/**
	 * Drop the staging tables created for a migration. The real tables are untouched.
	 *
	 * @param int $migration_id Purges its chunk receipts so a same-record retry re-imports.
	 */
	public function rollback_staging( array $bare_tables, int $migration_id = 0, string $staging_token = '', ?string $transfer_uuid = null ): void {
		foreach ( $bare_tables as $bare ) {
			$staging = $this->staging_table_name( $bare, $staging_token );
			TableDdl::drop( $this->wpdb, $staging );
		}

		if ( $migration_id > 0 ) {
			try {
				( new WpChunkReceiptStore( $this->wpdb ) )->clear_migration( $migration_id, $transfer_uuid );
			} catch ( \Throwable $e ) {
				// Best-effort: stale receipts merely cause a redundant re-import.
			}
		}
	}

	/**
	 * Drop shiftora_old_* checkpoints once migration completion is confirmed.
	 *
	 * Idempotent, so a replayed cleanup is a no-op. A table surviving the DROP throws,
	 * since it would break the next finalize's rename pair.
	 *
	 * @param string[] $bare_tables
	 *
	 * @return int Names verified gone (already-absent names count).
	 *
	 * @throws RuntimeException When a table still exists after the drop.
	 */
	public function drop_old_tables( array $bare_tables, string $staging_token = '' ): int {
		$cleaned = 0;
		foreach ( $bare_tables as $bare ) {
			$old = $this->old_table_name( $bare, $staging_token );
			$this->drop_foreign_key_dependent_table( $old );
			if ( TableDdl::exists( $this->wpdb, $old ) ) {
				throw new RuntimeException( "Old table still present after drop: {$old}" );
			}
			++$cleaned;
		}

		return $cleaned;
	}

	/**
	 * Decide whether a staging table holds this operation's own rows, and ready it.
	 *
	 * A staging name identifies its transfer only with a token, and the engine's SQL
	 * ingest passes none, so one name is shared by every migration on the site. An
	 * untokened table is reused only when this migration's ledger says it already
	 * staged rows there. Otherwise it is rebuilt: a failed first chunk is
	 * recoverable, merging in rows from a dead migration is not.
	 *
	 * @param string      $staging_token Per-transfer discriminator, or '' when shared.
	 * @param int         $migration_id  Owning migration, or 0 when unknown.
	 * @param string|null $transfer_uuid Owning transfer, or null when unknown.
	 *
	 * @return string The staging table name.
	 */
	private function prepare_staging_table(
		string $bare_table,
		string $schema,
		string $staging_token,
		int $migration_id,
		?string $transfer_uuid
	): string {
		$staging_table = $this->staging_table_name( $bare_table, $staging_token );
		$staging_sql   = $this->retarget_sql_head( $schema, $staging_table );
		$exists        = TableDdl::exists( $this->wpdb, $staging_table );
		$staged_rows   = $exists ? $this->staging_row_count( $staging_table ) : 0;

		// A schema only arrives with page 0, so the sender is restarting this table.
		// Existing rows come from a run that never finalized, and appending would hit
		// duplicate primary keys. A real mid-table resume has no schema and skips this.
		if ( $exists ) {
			TableDdl::drop( $this->wpdb, $staging_table );
		}

		if ( $staged_rows > 0 ) {
			// A rebuild is routine whenever the sender restarts a table, but a
			// discarded staging table is worth seeing when a transfer goes on to
			// fail, so it stays in the migration's log.
			( new MigrationLogger( $migration_id ) )->warning(
				"Rebuilt staging table {$bare_table}, discarding {$staged_rows} row(s) from an earlier attempt.",
				[
					'migration_id'  => $migration_id,
					'table'         => $bare_table,
					'staging_table' => $staging_table,
					'transfer_uuid' => $transfer_uuid,
				]
			);

			if ( function_exists( 'do_action' ) ) {
				/**
				 * Fires when foreign staging rows were discarded before a chunk.
				 *
				 * Those rows belonged to no identifiable migration, so this is the only trace.
				 *
				 * @param array $context Migration id, table and staging table name.
				 */
				do_action(
					'shiftora_staging_rows_discarded',
					[
						'migration_id'  => $migration_id,
						'table'         => $bare_table,
						'staging_table' => $staging_table,
						'transfer_uuid' => $transfer_uuid,
					]
				);
			}
		}

		$this->ensure_table( $staging_table, $staging_sql );

		return $staging_table;
	}

	private function staging_row_count( string $staging_table ): int {
		$wpdb = $this->wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- the transfer reads and writes the site's own tables, a cached copy would be stale
		return (int) $wpdb->get_var( $wpdb->prepare( 'SELECT COUNT(*) FROM %i', $staging_table ) );
	}

	/**
	 * Ensure the table exists with the correct structure.
	 *
	 * Runs dbDelta on the schema with constraints it can't carry stripped. For a
	 * staging table those are re-added under reserved names.
	 *
	 * @param bool $stage_constraints Carry table constraints across the swap.
	 *
	 * @throws RuntimeException If dbDelta is unavailable.
	 */
	private function ensure_table( string $table, string $schema, bool $stage_constraints = true ): void {
		if ( ! function_exists( 'dbDelta' ) ) {
			require_once ABSPATH . 'wp-admin/includes/upgrade.php';
		}

		// dbDelta reads a definition line that starts with CONSTRAINT as a
		// column called `constraint` and tries to ADD it. The clauses are lifted
		// out, put on the staging table under a reserved name so they travel
		// with it, and renamed back by finalize_staging() after the swap.
		$constraints = SchemaFingerprint::constraint_clauses( $schema );
		$schema      = SchemaFingerprint::strip_constraints( $schema );

		$local_schema = preg_replace(
			'/CREATE TABLE `[^`]+`/',
			"CREATE TABLE `{$table}`",
			$schema
		);

		if ( ! $local_schema ) {
			throw new RuntimeException( "Failed to rewrite schema for table {$table}" );
		}

		// Ensure a trailing semicolon.
		$local_schema = rtrim( $local_schema, ';' ) . ';';

		// Map the source server's collation names to ones this server has, and drop
		// MySQL 8 clauses MariaDB cannot run.
		$local_schema = ServerDdl::adapt( $this->wpdb, $local_schema );

		// dbDelta rewrites DDL line by line and mangles generated columns and
		// expression defaults. Staging tables are dropped before this runs, so
		// there is never an existing table to diff; for those schemas run the
		// source CREATE as-is instead. dbDelta still handles the common case.
		if ( $this->schema_needs_raw_create( $local_schema ) ) {
			// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared, PluginCheck.Security.DirectDB.UnescapedDBParameter -- the source's CREATE TABLE, already allow-listed by PeerSql; DDL text cannot be bound.
			$this->wpdb->query( $local_schema );
		} else {
			dbDelta( $local_schema );
		}

		// Verify dbDelta created (or already had) the table. It can return an
		// empty diff when it cannot interpret the DDL, and an empty-table
		// transfer has no later INSERT that would surface the failure.
		if ( ! TableDdl::exists( $this->wpdb, $table ) ) {
			throw new RuntimeException(
				"Failed to create table `{$table}`: dbDelta did not create it ({$this->wpdb->last_error})"
			);
		}

		if ( $stage_constraints ) {
			$this->stage_constraints( $table, $constraints );
		}
	}

	/**
	 * Whether a CREATE TABLE has syntax dbDelta can't reproduce (generated column,
	 * MySQL 8 expression or bit-literal DEFAULT) and must run verbatim.
	 *
	 * Matches on DDL text without regard to string literals, so it errs toward the raw path.
	 */
	private function schema_needs_raw_create( string $schema ): bool {
		return 1 === preg_match( '/\bGENERATED\s+ALWAYS\b/i', $schema )
			|| 1 === preg_match( '/\bAS\s*\(/i', $schema )
			|| 1 === preg_match( '/\bDEFAULT\s*\(/i', $schema )
			|| 1 === preg_match( "/\bDEFAULT\s+b'[01]+'/i", $schema );
	}

	/**
	 * Put the source's foreign keys and checks on a staging table under scoped names.
	 *
	 * The clauses travel with the table because the commit step is reached through
	 * push, pull and SQL import, and only push is handed the source schema later.
	 * Re-running is harmless: an existing clause came from an earlier chunk of this
	 * transfer.
	 *
	 * @param string[] $clauses Constraints the source declared.
	 */
	private function stage_constraints( string $table, array $clauses ): void {
		if ( [] === $clauses ) {
			return;
		}

		$own      = TableDdl::create_statement( $this->wpdb, $table );
		$on_table = array_map(
			[ SchemaFingerprint::class, 'constraint_name' ],
			SchemaFingerprint::constraint_clauses( $own )
		);

		foreach ( $clauses as $clause ) {
			$staged = SchemaFingerprint::scope_constraint( $clause );
			if ( null === $staged ) {
				// No name to reserve. MySQL only generates names from the table
				// they sit on, and a table cannot carry a foreign key named after
				// a name it will no longer have, so an unnamed clause came from
				// hand-written DDL, not from SHOW CREATE TABLE.
				continue;
			}

			if ( in_array( SchemaFingerprint::constraint_name( $staged ), $on_table, true ) ) {
				continue;
			}

			$kind = SchemaFingerprint::constraint_drop_keyword( $staged, $this->wpdb );
			// A staged definition may point at a table this selection is not
			// transferring, so it is bound under lowered checks, the same way
			// the dump importer loads one.
			$fk_before = (string) ( $this->wpdb->get_var( 'SELECT @@foreign_key_checks' ) ?? '1' );
			TableDdl::set_foreign_key_checks( $this->wpdb, '0' );
			$ok = TableDdl::add_constraint( $this->wpdb, $table, $staged );
			TableDdl::set_foreign_key_checks( $this->wpdb, $fk_before );

			if ( ! $ok ) {
				throw new RuntimeException( sprintf(
					'Table `%s` was created but its %s constraint could not be staged: %s',
					$table,
					$kind,
					(string) $this->wpdb->last_error
				) );
			}
		}
	}

	/**
	 * Snapshot protected option rows before the table is truncated.
	 *
	 * Captures PROTECTED_OPTIONS matches and all shiftora_* rows; transients are
	 * skipped. option_id is left out so the restore doesn't collide with the
	 * auto-increment IDs given to remote rows.
	 *
	 * @return array<int, array{option_name: string, option_value: string, autoload: string}>
	 */
	private function snapshot_protected_options( string $table ): array {
		$wpdb     = $this->wpdb;
		$names    = self::PROTECTED_OPTIONS;
		$prefixes = ProtectedOptions::prefixes();

		// The names and prefixes of one IN list and its LIKE arms, then the _user_roles
		// suffix, whose underscores are literal.
		$args = array_merge(
			[ $table ],
			$names,
			array_map( static fn( string $prefix ): string => $wpdb->esc_like( $prefix ) . '%', $prefixes ),
			[ '%' . $wpdb->esc_like( '_user_roles' ) ]
		);

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- the transfer reads and writes the site's own tables, a cached copy would be stale
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				'SELECT option_name, option_value, autoload FROM %i
				 WHERE option_name IN (' . implode( ', ', array_fill( 0, count( $names ), '%s' ) ) . ')'
				. implode( '', array_fill( 0, count( $prefixes ), ' OR option_name LIKE %s' ) )
				. ' OR option_name LIKE %s',
				$args
			),
			ARRAY_A
		);

		return is_array( $rows ) ? $rows : [];
	}

	/**
	 * Re-insert the snapshotted protected option rows after the remote data is in.
	 *
	 * Upserts, so the local value wins even if a protected row got past
	 * filter_protected_options.
	 *
	 * @param array<int, array{option_name: string, option_value: string, autoload: string}> $rows
	 */
	private function restore_protected_options( string $table, array $rows ): void {
		if ( empty( $rows ) ) {
			return;
		}

		// The literals and the statement that carries them must meet in a session that
		// reads a backslash as an escape, whatever the host's own sql_mode is. This runs
		// after the options table was emptied, so a refusal here would lose siteurl and home.
		$sql_mode_before = SessionSqlMode::relax( $this->wpdb );

		try {
			$this->write_protected_options( $table, $rows );
		} finally {
			SessionSqlMode::restore( $this->wpdb, $sql_mode_before );
		}
	}

	/**
	 * @param array<int, array{option_name: string, option_value: string, autoload: string}> $rows
	 */
	private function write_protected_options( string $table, array $rows ): void {
		$value_groups = [];

		foreach ( $rows as $row ) {
			$value_groups[] = '(' . implode(
				', ',
				[
					SqlLiteral::value( $row['option_name'] ),
					SqlLiteral::value( $row['option_value'] ),
					SqlLiteral::value( $row['autoload'] ?? 'yes' ),
				]
			) . ')';
		}

		$sql = 'INSERT INTO ' . SqlLiteral::identifier( $table ) . ' (option_name, option_value, autoload)
			 VALUES ' . implode( ', ', $value_groups ) . '
			 ON DUPLICATE KEY UPDATE
			   option_value = VALUES(option_value),
			   autoload     = VALUES(autoload)';

		// One bulk INSERT is far cheaper than N prepare()d rows; the values are escaped by SqlLiteral.
		// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared, PluginCheck.Security.DirectDB.UnescapedDBParameter -- a variable number of rows, escaped by SqlLiteral, which prepare() cannot bind.
		$this->wpdb->query( $sql );
	}

	/**
	 * Remove rows that must never be imported into any *_options table.
	 *
	 * Covers core options that identify this install, Shiftora options (encryption
	 * key, connection metadata, settings) and transients.
	 *
	 * @param array<int, array<string, mixed>> $rows Associative option rows.
	 *
	 * @return array<int, array<string, mixed>>
	 */
	private function filter_protected_options( array $rows ): array {
		return array_values(
			array_filter(
				$rows,
				static fn( array $row ): bool => ! ProtectedOptions::is_protected( (string) ( $row['option_name'] ?? '' ) )
			)
		);
	}

	/**
	 * Rewrite install-bound role keys on a staging table in place.
	 *
	 * The SQL path ships rows verbatim, so staging may still carry {src}user_roles
	 * and every {src}* usermeta key (and multisite variants). The key rule is
	 * PrefixedKeys, which the row path uses too.
	 *
	 * @param string $src_prefix Word-safe source table prefix.
	 */
	private function reconcile_prefixed_keys( string $staging, string $bare_table, string $src_prefix ): void {
		$wpdb         = $this->wpdb;
		$local_prefix = (string) $wpdb->prefix;
		$length       = strlen( $src_prefix ) + 1;

		if ( str_ends_with( $bare_table, 'options' ) ) {
			// Only user_roles carries the blog prefix in the options table; every
			// other option name is global.
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- the transfer reads and writes the site's own tables, a cached copy would be stale
			$wpdb->query(
				$wpdb->prepare(
					'UPDATE %i
					 SET option_name = CONCAT( %s, SUBSTRING( option_name, %d ) )
					 WHERE option_name = %s
					    OR option_name REGEXP %s',
					$staging,
					$local_prefix,
					$length,
					$src_prefix . 'user_roles',
					'^' . $src_prefix . '[0-9]+_user_roles$'
				)
			);
			return;
		}

		if ( str_ends_with( $bare_table, 'usermeta' ) ) {
			// Every usermeta key starting with the blog prefix is blog-scoped by WordPress
			// convention, not just roles, so rewrite the whole set. The prefix is escaped
			// for LIKE so its underscore is literal.
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- the transfer reads and writes the site's own tables, a cached copy would be stale
			$wpdb->query(
				$wpdb->prepare(
					'UPDATE %i SET meta_key = CONCAT( %s, SUBSTRING( meta_key, %d ) ) WHERE meta_key LIKE %s',
					$staging,
					$local_prefix,
					$length,
					$wpdb->esc_like( $src_prefix ) . '%'
				)
			);
		}
	}

	/**
	 * Parse a source CREATE against the fingerprint contract before any staging DDL.
	 *
	 * Constructs the receiver can't rebuild (generated columns, expression defaults,
	 * template markers) fail here while aborting is still cheap. Table constraints
	 * are in the contract, so they don't stop a transfer.
	 */
	private function preflight_schema( string $schema, string $bare_table ): void {
		try {
			SchemaFingerprint::fingerprint( $schema );
		} catch ( RuntimeException $e ) {
			throw new RuntimeException(
				sprintf( 'Schema preflight rejected `%s`: %s', $bare_table, $e->getMessage() ),
				0,
				$e
			);
		}
	}

	/**
	 * Build a staging/rollback table name namespaced by a transfer token.
	 *
	 * The token sits between the stem and the bare name. If that would pass MySQL's
	 * 64-char cap, an md5 suffix over the prefixed bare name and the full token
	 * replaces the bare name, and the token is truncated to the remaining budget.
	 * The hash covers the full token so transfers sharing a truncated slice can't
	 * collide. If even a minimum discriminator doesn't fit, the name is rejected.
	 * An empty token gives the unscoped names.
	 */
	private function scoped_table_name( string $stem, string $bare, string $token ): string {
		return TableNameScoper::scoped_table_name( (string) $this->wpdb->prefix, $stem, $token, $bare );
	}

	/**
	 * Create a staging table from a dump's CREATE TABLE, the way a transfer does.
	 *
	 * A dump's constraint names are unique per schema, so a dump of this same site
	 * would collide with its live tables, and its REFERENCES keep the source prefix.
	 * ensure_table() stages constraints under scoped names that finalize_staging()
	 * binds after the swap, and maps collations this server lacks.
	 *
	 * @param string $staging_table Staging table the statement already names.
	 */
	public function create_staging_from_dump( string $staging_table, string $create_sql ): void {
		// dbDelta reads the word after CREATE TABLE as the table name.
		$create_sql = (string) preg_replace( '/^(\s*CREATE\s+TABLE)\s+IF\s+NOT\s+EXISTS\b/i', '$1', $create_sql );

		$this->ensure_table( $staging_table, $create_sql );
	}

	/**
	 * The staging table a transfer would stage `$bare` under.
	 *
	 * Public so finalize diagnostics name tables the same way the commit does.
	 */
	public function staging_table_name( string $bare, string $token ): string {
		return $this->scoped_table_name( 'shiftora_stg_', $bare, $token );
	}

	/**
	 * Runs a check on the other site's SQL and logs a refusal before it propagates.
	 *
	 * The reason stays in the log row; the exception message is the fixed refusal text.
	 *
	 * @throws PeerSqlRefused
	 */
	private function guard_peer_sql( callable $check, int $migration_id, string $bare_table ): void {
		try {
			$check();
		} catch ( PeerSqlRefused $e ) {
			if ( $migration_id > 0 ) {
				( new MigrationLogger( $migration_id ) )->error( $e->getMessage(), [ 'table' => $bare_table, 'reason' => $e->reason() ] );
			}
			throw $e;
		}
	}

	/**
	 * Prefixes a foreign key from the other site may point at: this site's and the sender's.
	 *
	 * @return list<string>
	 */
	private function reference_prefixes( string $source_prefix ): array {
		return array_values( array_unique( array_filter(
			[ (string) $this->wpdb->prefix, $source_prefix ],
			static fn( string $prefix ): bool => '' !== $prefix
		) ) );
	}

	/**
	 * Redirect an emitted __TARGET__ head to the real table name.
	 *
	 * Only the statement head (CREATE TABLE / INSERT INTO) is replaced, so a value
	 * containing "__TARGET__" is left alone.
	 */
	private function retarget_sql_head( string $sql, string $target_table ): string {
		return (string) preg_replace_callback(
			'/^((?:CREATE TABLE|INSERT INTO)\s+)`__TARGET__`/i',
			static fn( array $m ): string => $m[1] . '`' . $target_table . '`',
			$sql
		);
	}

	private function old_table_name( string $bare, string $token ): string {
		return $this->scoped_table_name( 'shiftora_old_', $bare, $token );
	}

// phpcs:enable WordPress.Security.EscapeOutput.ExceptionNotEscaped
}