<?php
/**
 * Plugin activation handler: creates database tables and sets options.
 *
 * @package Shiftora\Core
 */

declare(strict_types=1);

namespace Shiftora\Core;

use Shiftora\Backup\BackupPinStore;
use Shiftora\Backup\RestoreTable;
use Shiftora\Connection\TokenGenerator;
use Shiftora\Migration\WpChunkReceiptStore;
use Shiftora\Support\CacheFlusher;
use Shiftora\Support\Db\TableDdl;
use Shiftora\Support\HttpClient;

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

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

/**
 * Creates every Shiftora table with dbDelta() and keeps the schema current.
 */
final class Installer {

	/** Activation entry point, registered with register_activation_hook(). */
	public static function activate(): void {
		try {
			self::create_tables();
			self::stamp_schema();
		} catch ( LedgerSchemaException $e ) {
			update_option( 'shiftora_last_db_repair_error', $e->getMessage(), false );
			return;
		}
		self::set_defaults();

		// Activation rewrites options and rules, so earlier host caches are stale.
		CacheFlusher::flush_silently();
	}

	/**
	 * Upgrade check on admin_init, for administrators only and only when the stored
	 * schema revision or release differs from this build. Also covers installs made
	 * by copying files, which skip the activation hook (they carry no stored revision).
	 */
	public static function maybe_upgrade(): void {
		// admin_init also fires for admin-ajax.php and admin-post.php hits from visitors; they never run DDL.
		if ( ! current_user_can( 'manage_options' ) ) {
			return;
		}

		if ( self::installed_revision() >= SHIFTORA_SCHEMA_REV
			&& SHIFTORA_DB_VERSION === get_option( 'shiftora_db_version' )
		) {
			return;
		}

		try {
			self::converge_schema();
			// A successful repair clears the record of an earlier failed one.
			delete_option( 'shiftora_last_db_repair_error' );
		} catch ( LedgerSchemaException $e ) {
			// Leave the DB version alone so the next admin_init retries.
			update_option( 'shiftora_last_db_repair_error', $e->getMessage(), false );
		}
	}

	private static function converge_schema(): void {
		// A copied install can have no tables, and the identity guard below would
		// ALTER a missing ledger. Provision the base table first.
		self::ensure_ledger_table_exists();

		// The gates further down ALTER columns on tables a copied install may lack.
		self::ensure_gate_targets_exist();

		// Checked before the revision: a broken but unstamped install would stay broken,
		// because dbDelta cannot repair a unique-key shape.
		self::ensure_ledger_identity();

		$revision = self::installed_revision();

		// Revision zero is an older install or a lost marker; both need the whole legacy chain.
		if ( 0 === $revision ) {
			self::run_legacy_gates( (string) get_option( 'shiftora_db_version' ) );

			// Old installs defaulted chunk_size to 500. Lift it once so a low size chosen
			// for a fragile host is not overwritten later.
			$existing_chunk_size = get_option( 'shiftora_chunk_size' );
			if ( false !== $existing_chunk_size && (int) $existing_chunk_size <= 500 ) {
				update_option( 'shiftora_chunk_size', 10000, false );
			}
		}

		if ( $revision < SHIFTORA_SCHEMA_REV ) {
			self::run_revision_gates( $revision );
		}

		// create_tables() throws on a shape mismatch, so only a converged install is stamped.
		self::create_tables();
		self::set_defaults();
		self::stamp_schema();
	}
	private static function installed_revision(): int {
		return (int) get_option( 'shiftora_schema_rev', 0 );
	}

	/**
	 * Records the schema state: shiftora_schema_rev is the monotonic integer the
	 * next upgrade compares against, shiftora_db_version the release number for diagnostics.
	 */
	private static function stamp_schema(): void {
		update_option( 'shiftora_schema_rev', SHIFTORA_SCHEMA_REV, false );
		update_option( 'shiftora_db_version', SHIFTORA_DB_VERSION, false );
	}

	/**
	 * Gates keyed on a revision rather than a release number. Each runs exactly
	 * once per database and must be safe to re-run, since an admin_init can stop
	 * halfway. Bump SHIFTORA_SCHEMA_REV in shiftora.php when adding one.
	 *
	 * @return list<array{revision: int, apply: callable}>
	 */
	private static function revision_gates(): array {
		return [
			[
				'revision' => 3,
				'apply'    => static function (): void {
					// Formerly added the size-allowance columns, removed 2026-10-16 with the
					// feature. Kept as a no-op so revision numbers stay monotonic; installs
					// that ran it keep the orphan columns harmlessly.
				},
			],
			[
				'revision' => 4,
				'apply'    => static function (): void {
					// Deep uploads trees overflow 191 chars and a truncated key loses rows
					// silently. A no-op once the width reads 255.
					self::widen_column_if_narrower(
						'shiftora_file_progress',
						'virtual_path',
						'varchar(255) NOT NULL',
						255
					);
				},
			],
			[
				'revision' => 5,
				'apply'    => static function (): void {
					// The destination's request-body ceiling, which a push sizes its SQL chunks
					// under. An earlier push has no value and uses the sender's own budget.
					self::add_columns_if_missing(
						'shiftora_migrations',
						[
							'remote_max_post_bytes' => 'BIGINT UNSIGNED NULL DEFAULT NULL',
						]
					);
				},
			],
			[
				'revision' => 6,
				'apply'    => static function (): void {
					// 255 chars still truncates deep vendor paths, and a case-insensitive collation
					// folds IMG.JPG into img.jpg. Widening and going binary cannot collide.
					self::rebuild_path_column();
				},
			],
			[
				'revision' => 7,
				'apply'    => static function (): void {
					// A profile remembers which stages run and nothing else. The table pick and
					// the integrity flag columns were never read once selection and
					// verification left this plugin; dbDelta cannot drop them.
					self::drop_columns_if_present( 'shiftora_profiles', [ 'tables_json', 'all_tables', 'verify_integrity' ] );
				},
			],
			[
				'revision' => 8,
				'apply'    => static function (): void {
					// Pushes are gone: every write on a site starts from its own admin. A push
					// still in flight cannot be resumed, so it is closed instead of being picked
					// up as a pull, and the switch that allowed incoming pushes is dropped.
					self::retire_push_migrations();
					delete_option( 'shiftora_allow_push' );
				},
			],
			[
				'revision' => 9,
				'apply'    => static function (): void {
					// A connection that was push only was never allowed to read this site, and a
					// pull-only site has nothing else to give it, so it is shown as revoked.
					self::revoke_push_only_connections();
				},
			],
		];
	}

	/** @param int $installed The revision already recorded; gates at or below it must not run again. */
	private static function run_revision_gates( int $installed ): void {
		foreach ( self::revision_gates() as $gate ) {
			if ( $gate['revision'] > $installed ) {
				$gate['apply']();
			}
		}
	}

	/** Gates keyed on version strings, for installs that predate revision numbers. Every statement is guarded, so a fresh database runs them as no-ops. */
	private static function run_legacy_gates( string $installed ): void {

		// v1.3.0: file-migration tracking columns.
		if ( version_compare( $installed, '1.3.0', '<' ) ) {
			self::add_columns_if_missing(
				'shiftora_migrations',
				[
					'total_files'       => 'INT NOT NULL DEFAULT 0',
					'transferred_files' => 'INT NOT NULL DEFAULT 0',
					'bytes_total'       => 'BIGINT NOT NULL DEFAULT 0',
					'bytes_transferred' => 'BIGINT NOT NULL DEFAULT 0',
					'file_offsets_json' => 'LONGTEXT NULL DEFAULT NULL',
				]
			);
		}

		// v1.4.0: allow pull/push columns on connections, per-stage progress
		// columns on migrations.
		if ( version_compare( $installed, '1.4.0', '<' ) ) {
			self::add_columns_if_missing(
				'shiftora_connections',
				[
					'remote_allow_pull' => 'TINYINT(1) NOT NULL DEFAULT 1',
					'remote_allow_push' => 'TINYINT(1) NOT NULL DEFAULT 1',
				]
			);
			self::add_columns_if_missing(
				'shiftora_migrations',
				[
					'current_stage'  => 'VARCHAR(20) DEFAULT NULL',
					'stage_progress' => 'TINYINT UNSIGNED NOT NULL DEFAULT 0',
				]
			);
		}

		// v1.4.21: remote_site_url on connections.
		if ( version_compare( $installed, '1.4.21', '<' ) ) {
			self::add_columns_if_missing(
				'shiftora_connections',
				[ 'remote_site_url' => "VARCHAR(2083) NOT NULL DEFAULT ''" ]
			);
		}

			// v1.4.22: remote plugin version, for strict parity checks.
			if ( version_compare( $installed, '1.4.22', '<' ) ) {
				self::add_columns_if_missing(
					'shiftora_connections',
					[ 'remote_plugin_version' => "VARCHAR(64) NOT NULL DEFAULT ''" ]
				);
			}

			// v1.4.23: durable stage state so completed stages survive worker handoffs and retries.
			if ( version_compare( $installed, '1.4.23', '<' ) ) {
				self::add_columns_if_missing(
					'shiftora_migrations',
					[
						'stage_plan_json'       => 'LONGTEXT NULL DEFAULT NULL',
						'completed_stages_json' => 'LONGTEXT NULL DEFAULT NULL',
					]
				);
			}

			// v1.4.24: recovery_error.
			if ( version_compare( $installed, '1.4.24', '<' ) ) {
				self::add_columns_if_missing(
					'shiftora_migrations',
					[ 'recovery_error' => 'TEXT NULL DEFAULT NULL' ]
				);
			}

			// v1.4.25: recovery_status.
			if ( version_compare( $installed, '1.4.25', '<' ) ) {
				self::add_columns_if_missing(
					'shiftora_migrations',
					[ 'recovery_status' => 'VARCHAR(32) NULL DEFAULT NULL' ]
				);
			}

			// v1.4.26: remote_finalize columns for push rollback tracking.
			if ( version_compare( $installed, '1.4.26', '<' ) ) {
				self::add_columns_if_missing(
					'shiftora_migrations',
					[
						'remote_finalize_status'      => 'VARCHAR(32) NULL DEFAULT NULL',
						'remote_finalize_tables_json' => 'LONGTEXT NULL DEFAULT NULL',
					]
				);
			}

			// v1.5.1: sender-side lease identity so resumed pushes can prove which transfer their chunks belong to.
			if ( version_compare( $installed, '1.5.1', '<' ) ) {
				self::add_columns_if_missing(
					'shiftora_migrations',
					[ 'transfer_uuid' => 'VARCHAR(32) NULL DEFAULT NULL' ]
				);
			}

			// v1.5.7: drop the unused bytes_applied column (dbDelta never removes columns).
			if ( version_compare( $installed, '1.5.7', '<' ) ) {
				self::drop_columns_if_present( 'shiftora_migrations', [ 'bytes_applied' ] );
			}

			// v1.5.2: move the chunk-ledger transfer identity into its own column. Purge
			// the old rows and unique key; the dbDelta pass installs the composite.
		if ( version_compare( $installed, '1.5.2', '<' ) ) {
			self::apply_ledger_identity_upgrade();
		}
	}

	/**
	 * Resets the chunk-ledger identity (v1.5.2).
	 *
	 * Old rows stored `transfer_uuid|chunk_key` in a CHAR(64) column and were
	 * truncated or rejected. Hex chunk hashes never contain '|', so one purge
	 * removes exactly those rows. dbDelta cannot drop a key, so the old unique goes first.
	 */
	private static function apply_ledger_identity_upgrade(): void {
		global $wpdb;

		$ledger = $wpdb->prefix . 'shiftora_import_chunks';

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$wpdb->query( $wpdb->prepare( 'DELETE FROM %i WHERE chunk_key LIKE %s', $ledger, '%' . $wpdb->esc_like( '|' ) . '%' ) );
		TableDdl::drop_index( $wpdb, $ledger, 'chunk' );
	}

	/**
	 * Whether the chunk-ledger table has the composite identity index.
	 *
	 * Probes the real schema, not the stored revision. The column must be NOT NULL
	 * with an empty-string default and the index unique. A failed read counts as
	 * not intact, so repair never builds on metadata it could not read.
	 */
	private static function ledger_identity_intact(): bool {
		global $wpdb;

		// The column is half of the composite; an index over a missing column
		// cannot exist.
		$column = self::column_exists( $wpdb->prefix . 'shiftora_import_chunks', 'transfer_uuid' );
		if ( '' !== (string) $wpdb->last_error ) {
			return false;
		}
		if ( '0' === (string) $column || null === $column ) {
			return false;
		}

		// A legacy column can exist yet stay nullable, which the receipt INSERTs
		// assume it is not.
		$nullable = self::column_field( $wpdb->prefix . 'shiftora_import_chunks', 'transfer_uuid', 'IS_NULLABLE' );
		if ( '' !== (string) $wpdb->last_error ) {
			return false;
		}
		if ( 'NO' !== (string) $nullable ) {
			return false;
		}

		$default = self::column_field( $wpdb->prefix . 'shiftora_import_chunks', 'transfer_uuid', 'COLUMN_DEFAULT' );
		if ( '' !== (string) $wpdb->last_error ) {
			return false;
		}
		// MySQL 8.4 reports null for DEFAULT '' just as for no default, and strict mode
		// rejects an INSERT omitting a NOT NULL column with no default (1364); ask the DDL.
		if ( null === $default ) {
			// Unreadable DDL proves nothing either way, so stay conservative.
			return true === self::declares_empty_string_default(
				$wpdb->prefix . 'shiftora_import_chunks',
				'transfer_uuid'
			);
		}
		// MariaDB 10.2.7+ reports a string default as a quoted literal, so DEFAULT '' arrives as two quotes.
		if ( '' !== (string) $default && "''" !== (string) $default ) {
			return false;
		}

		// The unique key must be the exact composite, in order.
		$probe = self::ledger_index_probe();
		if ( null === $probe ) {
			return false;
		}
		if ( ! $probe['unique'] ) {
			return false;
		}

		return [ 'migration_id', 'transfer_uuid', 'chunk_key' ] === $probe['columns'];
	}

	/** Whether the table DDL declares DEFAULT '' for a column; null when the DDL could not be read. Last resort for servers that hide it from metadata. */
	private static function declares_empty_string_default( string $table, string $column ): ?bool {
		global $wpdb;

		$create = TableDdl::create_statement( $wpdb, $table );
		if ( '' !== (string) $wpdb->last_error || '' === $create ) {
			return null;
		}

		// SHOW CREATE puts one column per line; splitting on commas would break decimal(10,2).
		$pattern = '/(?:^|\R)\s*`' . preg_quote( $column, '/' ) . '`\s+([^\r\n]*)/i';
		if ( ! preg_match( $pattern, $create, $matches ) ) {
			return null;
		}

		return 1 === preg_match( "/DEFAULT\s+(''|\"\")/i", $matches[1] );
	}

	/**
	 * The chunk-index shape: ordered columns and whether every member is unique.
	 * Null means the probe failed, so no verdict is possible.
	 *
	 * @return array{columns: string[], unique: bool}|null
	 */
	private static function ledger_index_probe(): ?array {
		global $wpdb;

		$rows = TableDdl::indexes( $wpdb, $wpdb->prefix . 'shiftora_import_chunks' );
		if ( '' !== (string) $wpdb->last_error ) {
			return null;
		}

		$cols   = [];
		$unique = true;
		foreach ( $rows as $index ) {
			if ( 'chunk' === (string) $index->Key_name ) {
				$cols[ (int) $index->Seq_in_index ] = (string) $index->Column_name;
				if ( '0' !== (string) $index->Non_unique ) {
					$unique = false;
				}
			}
		}
		ksort( $cols );

		return [ 'columns' => array_values( $cols ), 'unique' => $unique ];
	}

	/**
	 * Makes the chunk-ledger table exist before the identity repair runs. Only
	 * this table: a full dbDelta pass would rebuild the legacy chunk index that
	 * the identity upgrade is about to drop.
	 */
	private static function ensure_ledger_table_exists(): void {
		global $wpdb;

		$ledger = $wpdb->prefix . 'shiftora_import_chunks';

		if ( TableDdl::exists( $wpdb, $ledger ) ) {
			return;
		}

		$ddl = WpChunkReceiptStore::table_ddl( $ledger, $wpdb->get_charset_collate() );
		// Our own schema, with this site's prefix and charset; prepare() cannot wrap a CREATE TABLE.
		// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared, PluginCheck.Security.DirectDB.UnescapedDBParameter, WordPress.DB.DirectDatabaseQuery.SchemaChange, WordPress.DB.DirectDatabaseQuery -- our own table schema, written here; DDL takes no placeholder
		$created = $wpdb->query( $ddl );
		if ( false === $created ) {
			throw new LedgerSchemaException(
				sprintf( 'Could not provision the chunk-ledger table: %s', (string) $wpdb->last_error )
			);
		}
	}

	/**
	 * Converges the ledger on the composite identity index. Runs on every
	 * admin_init, so a healthy install only pays for a read.
	 *
	 * @throws LedgerSchemaException When a repair step fails or the re-probe still disagrees.
	 */
	private static function ensure_ledger_identity(): bool {
		global $wpdb;

		if ( self::ledger_identity_intact() ) {
			return true;
		}

		// A failed read is not a missing piece; repairing from unreadable metadata could build the wrong shape.
		if ( '' !== (string) $wpdb->last_error ) {
			throw new LedgerSchemaException(
				sprintf( 'Could not verify the chunk-ledger identity: %s', (string) $wpdb->last_error )
			);
		}

		$ledger = $wpdb->prefix . 'shiftora_import_chunks';

		self::add_columns_if_missing(
			'shiftora_import_chunks',
			[ 'transfer_uuid' => "CHAR(32) NOT NULL DEFAULT ''" ]
		);

		// Legacy NULL receipts move to the empty-string namespace: MySQL treats every
		// NULL in a unique key as distinct, so a pending and a committed row could share
		// a chunk. Duplicates are purged first to avoid a 1062.
		self::purge_ledger_identity_duplicates();
		$result = self::backfill_ledger_identity();
		if ( false === $result && str_contains( (string) $wpdb->last_error, 'Duplicate entry' ) ) {
			// Purge again and retry once if the first purge missed a collision.
			self::purge_ledger_identity_duplicates();
			$result = self::backfill_ledger_identity();
		}
		if ( false === $result ) {
			throw new LedgerSchemaException(
				sprintf( 'Ledger identity backfill failed: %s', (string) $wpdb->last_error )
			);
		}

		// Tighten a nullable legacy column so receipts see '' or a 32-char token, never NULL.
		$nullable = self::column_field( $ledger, 'transfer_uuid', 'IS_NULLABLE' );
		if ( '' !== (string) $wpdb->last_error ) {
			throw new LedgerSchemaException( 'Could not confirm ledger column nullability.' );
		}
		if ( 'NO' !== (string) $nullable ) {
			$modified = TableDdl::modify_column( $wpdb, $ledger, 'transfer_uuid', "CHAR(32) NOT NULL DEFAULT ''" );
			if ( ! $modified ) {
				throw new LedgerSchemaException(
					sprintf( 'Ledger column tightening failed: %s', (string) $wpdb->last_error )
				);
			}
		} elseif ( false === self::declares_empty_string_default( $ledger, 'transfer_uuid' ) ) {
			// NOT NULL with no default fails under strict mode (1364) once a receipt INSERT
			// omits the column. Only an explicit false from the DDL check rewrites; null means unreadable.
			$modified = TableDdl::modify_column( $wpdb, $ledger, 'transfer_uuid', "CHAR(32) NOT NULL DEFAULT ''" );
			if ( ! $modified ) {
				throw new LedgerSchemaException(
					sprintf( 'Ledger default tightening failed: %s', (string) $wpdb->last_error )
				);
			}
		}

		$probe = self::ledger_index_probe();
		if ( null === $probe ) {
			throw new LedgerSchemaException(
				sprintf( 'Could not re-read the ledger index: %s', (string) $wpdb->last_error )
			);
		}
		if ( ! $probe['unique']
			|| [ 'migration_id', 'transfer_uuid', 'chunk_key' ] !== $probe['columns'] ) {
			if ( ! TableDdl::drop_index( $wpdb, $ledger, 'chunk' ) ) {
				throw new LedgerSchemaException(
					sprintf( 'Ledger index drop failed: %s', (string) $wpdb->last_error )
				);
			}
			if ( ! TableDdl::add_unique_key( $wpdb, $ledger, 'chunk', [ 'migration_id', 'transfer_uuid', 'chunk_key' ] ) ) {
				throw new LedgerSchemaException(
					sprintf( 'Ledger composite index add failed: %s', (string) $wpdb->last_error )
				);
			}
		}

		if ( ! self::ledger_identity_intact() ) {
			throw new LedgerSchemaException(
				sprintf( 'Ledger is still not converged after repair: %s', (string) $wpdb->last_error )
			);
		}

		return true;
	}

	/**
	 * Removes duplicate NULL-identity receipts before the empty-string backfill.
	 * A committed receipt outranks its pending twin; among pending-only
	 * duplicates the lowest row id wins.
	 */
	private static function purge_ledger_identity_duplicates(): void {
		global $wpdb;

		$ledger = $wpdb->prefix . 'shiftora_import_chunks';

		// The derived table avoids MySQL error 1093 on a self-referencing delete.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$wpdb->query(
			$wpdb->prepare(
				"DELETE FROM %i
				 WHERE transfer_uuid IS NULL
				   AND status = 'pending'
				   AND (migration_id, chunk_key) IN (
				       SELECT migration_id, chunk_key
				         FROM ( SELECT migration_id, chunk_key
				                  FROM %i c
				                 WHERE transfer_uuid IS NULL AND status = 'committed'
				                 GROUP BY migration_id, chunk_key ) AS committed_keys
				   )",
				$ledger,
				$ledger
			)
		);

		// Pending-only duplicates: keep the lowest id per (migration, chunk_key).
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$wpdb->query(
			$wpdb->prepare(
				"DELETE FROM %i
				 WHERE transfer_uuid IS NULL
				   AND status = 'pending'
				   AND (migration_id, chunk_key) IN (
				       SELECT migration_id, chunk_key
				         FROM ( SELECT migration_id, chunk_key, COUNT(*) AS n
				                  FROM %i p
				                 WHERE transfer_uuid IS NULL AND status = 'pending'
				                 GROUP BY migration_id, chunk_key
				                 HAVING COUNT(*) > 1 ) AS dup_pending
				   )
				   AND id NOT IN (
				       SELECT keep_id
				         FROM ( SELECT MIN(id) AS keep_id
				                  FROM %i
				                 WHERE transfer_uuid IS NULL AND status = 'pending'
				                 GROUP BY migration_id, chunk_key ) AS keep_ids
				   )",
				$ledger,
				$ledger,
				$ledger
			)
		);
	}

	/** Moves legacy NULL receipts into the empty-string identity. @return int|bool Rows changed, or false on failure. */
	private static function backfill_ledger_identity(): int|bool {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		return $wpdb->query( $wpdb->prepare( "UPDATE %i SET transfer_uuid = '' WHERE transfer_uuid IS NULL", $wpdb->prefix . 'shiftora_import_chunks' ) );
	}

	/** One COUNT of the column's metadata row: '0' or null when it is not there. */
	private static function column_exists( string $table, string $column ): mixed {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- schema metadata, a cached copy would be stale
		return $wpdb->get_var(
			$wpdb->prepare(
				'SELECT COUNT(*) FROM information_schema.columns
				  WHERE table_schema = DATABASE()
				    AND table_name   = %s
				    AND column_name  = %s
				    AND data_type    IS NOT NULL',
				$table,
				$column
			)
		);
	}

	/** One metadata field of a column (IS_NULLABLE, COLUMN_DEFAULT, CHARACTER_MAXIMUM_LENGTH); null when the column is not there. */
	private static function column_field( string $table, string $column, string $field ): mixed {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- schema metadata, a cached copy would be stale
		return $wpdb->get_var(
			$wpdb->prepare(
				'SELECT %i FROM information_schema.columns
				  WHERE table_schema = DATABASE()
				    AND table_name   = %s
				    AND column_name  = %s',
				$field,
				$table,
				$column
			)
		);
	}

	/**
	 * Adds columns to an existing table only when missing.
	 *
	 * Uses INFORMATION_SCHEMA rather than ADD COLUMN IF NOT EXISTS for MySQL 5.7
	 * and MariaDB 10.1. A failed metadata read is not "column absent", and a
	 * rejected ALTER aborts the upgrade.
	 *
	 * @param string               $bare_table Table name without DB prefix.
	 * @param array<string,string> $col_defs   Column name to column definition.
	 */
	private static function add_columns_if_missing( string $bare_table, array $col_defs ): void {
		global $wpdb;

		$table = $wpdb->prefix . $bare_table;

		foreach ( $col_defs as $column => $definition ) {
			$exists = self::column_exists( $table, $column );

			if ( '' !== (string) $wpdb->last_error ) {
				throw new LedgerSchemaException(
					sprintf( 'Could not check column %s on %s: %s', $column, $table, (string) $wpdb->last_error )
				);
			}

			if ( '0' === (string) $exists || null === $exists ) {
				if ( ! TableDdl::add_column( $wpdb, $table, $column, $definition ) ) {
					throw new LedgerSchemaException(
						sprintf( 'Failed to add column %s to %s: %s', $column, $table, (string) $wpdb->last_error )
					);
				}
			}
		}
	}

	/**
	 * Brings shiftora_file_progress.virtual_path to 512 binary-collated chars.
	 * widen_column_if_narrower() reads only width and this is also a collation
	 * change, so both are probed. A no-op once they match.
	 */
	private static function rebuild_path_column(): void {
		global $wpdb;

		$table = $wpdb->prefix . 'shiftora_file_progress';
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- schema metadata, a cached copy would be stale
		$row   = $wpdb->get_row(
			$wpdb->prepare(
				'SELECT CHARACTER_MAXIMUM_LENGTH AS width, COLLATION_NAME AS collation
				   FROM information_schema.columns
				  WHERE table_schema = DATABASE()
				    AND table_name   = %s
				    AND column_name  = %s',
				$table,
				'virtual_path'
			)
		);

		if ( '' !== (string) $wpdb->last_error ) {
			throw new LedgerSchemaException(
				sprintf( 'Could not check virtual_path on %s: %s', $table, (string) $wpdb->last_error )
			);
		}

		// No row: the table gets the new DDL from the dbDelta pass.
		if ( null === $row || ( (int) $row->width >= 512 && 'utf8mb4_bin' === (string) $row->collation ) ) {
			return;
		}

		if ( ! TableDdl::modify_column( $wpdb, $table, 'virtual_path', 'VARCHAR(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL' ) ) {
			throw new LedgerSchemaException(
				sprintf( 'Failed to rebuild virtual_path on %s: %s', $table, (string) $wpdb->last_error )
			);
		}
	}

	/**
	 * Widens a column when it is narrower than the schema asks for. Idempotent.
	 *
	 * @param string $bare_table Table name without DB prefix.
	 * @param string $definition Full column definition for the MODIFY.
	 * @param int    $min_chars  Width the column must already meet to skip.
	 */
	private static function widen_column_if_narrower( string $bare_table, string $column, string $definition, int $min_chars ): void {
		global $wpdb;

		$table = $wpdb->prefix . $bare_table;

		$width = self::column_field( $table, $column, 'CHARACTER_MAXIMUM_LENGTH' );

		if ( '' !== (string) $wpdb->last_error ) {
			throw new LedgerSchemaException(
				sprintf( 'Could not check column %s on %s: %s', $column, $table, (string) $wpdb->last_error )
			);
		}

		// A NULL width means the column is absent; missing tables are provisioned before the gates run, so treat it as satisfied.
		if ( null === $width || (int) $width >= $min_chars ) {
			return;
		}

		if ( ! TableDdl::modify_column( $wpdb, $table, $column, $definition ) ) {
			throw new LedgerSchemaException(
				sprintf( 'Failed to widen column %s on %s: %s', $column, $table, (string) $wpdb->last_error )
			);
		}
	}

	/**
	 * Removes columns the plugin no longer uses. dbDelta only adds columns, and
	 * each is probed first because DROP COLUMN fails on a server that never had it.
	 *
	 * @param string[] $columns Column names to drop.
	 */
	private static function drop_columns_if_present( string $bare_table, array $columns ): void {
		global $wpdb;

		$table = $wpdb->prefix . $bare_table;

		foreach ( $columns as $column ) {
			$exists = self::column_exists( $table, $column );

			if ( '' !== (string) $wpdb->last_error ) {
				throw new LedgerSchemaException(
					sprintf( 'Could not check column %s on %s: %s', $column, $table, (string) $wpdb->last_error )
				);
			}

			if ( '0' === (string) $exists || null === $exists ) {
				continue;
			}

			if ( ! TableDdl::drop_column( $wpdb, $table, $column ) ) {
				throw new LedgerSchemaException(
					sprintf( 'Failed to drop column %s from %s: %s', $column, $table, (string) $wpdb->last_error )
				);
			}
		}
	}

	/** Revokes the connections stored as push only, so the Connections page shows them plainly. */
	private static function revoke_push_only_connections(): void {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, one-off schema gate
		$wpdb->query(
			$wpdb->prepare(
				"UPDATE %i SET status = 'revoked', updated_at = %s WHERE direction = 'push_only' AND status <> 'revoked'",
				$wpdb->prefix . 'shiftora_connections',
				current_time( 'mysql', true )
			)
		);
	}

	/** Closes the pushes that were still in flight when pushes were removed. */
	private static function retire_push_migrations(): void {
		global $wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, one-off schema gate
		$wpdb->query(
			$wpdb->prepare(
				"UPDATE %i
				    SET status = 'failed', error_message = %s, updated_at = %s
				  WHERE direction = 'push' AND status IN ('pending', 'running', 'paused', 'cancelling')",
				$wpdb->prefix . 'shiftora_migrations',
				'This push was stopped when Shiftora became pull-only. To send this site somewhere, log into the destination and pull.',
				current_time( 'mysql', true )
			)
		);
	}

	/**
	 * Provisions the tables the column gates write to, leaving the ledger out:
	 * dbDelta would rebuild the legacy chunk index that
	 * apply_ledger_identity_upgrade() drops. ensure_ledger_table_exists() covers it.
	 */
	private static function ensure_gate_targets_exist(): void {
		self::apply_table_ddl( false );
	}

	/** @param bool $include_ledger Whether to pass the chunk ledger through dbDelta. */
	private static function apply_table_ddl( bool $include_ledger ): void {
		global $wpdb;

		require_once ABSPATH . 'wp-admin/includes/upgrade.php';

		$charset_collate = $wpdb->get_charset_collate();

		$sql_connections = "CREATE TABLE {$wpdb->prefix}shiftora_connections (
			id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
			label VARCHAR(255) NOT NULL DEFAULT '',
			remote_url VARCHAR(2083) NOT NULL,
			remote_site_url VARCHAR(2083) NOT NULL DEFAULT '',
			remote_plugin_version VARCHAR(64) NOT NULL DEFAULT '',
			direction VARCHAR(16) NOT NULL DEFAULT 'both',
			local_token_hash VARCHAR(255) NOT NULL DEFAULT '',
			remote_token_encrypted TEXT NOT NULL,
			remote_db_prefix VARCHAR(64) NOT NULL DEFAULT 'wp_',
			remote_allow_pull TINYINT(1) NOT NULL DEFAULT 1,
			remote_allow_push TINYINT(1) NOT NULL DEFAULT 1,
			status VARCHAR(32) NOT NULL DEFAULT 'pending',
			permissions LONGTEXT NULL,
			last_ping_at DATETIME NULL DEFAULT NULL,
			created_at DATETIME NOT NULL,
			updated_at DATETIME NOT NULL,
			PRIMARY KEY (id),
			KEY status (status),
			KEY remote_url (remote_url(191))
		) $charset_collate;";

		// remote_max_post_bytes is the destination's request-body ceiling, captured at
		// import/start so a push sizes SQL chunks under it.
		$sql_migrations = "CREATE TABLE {$wpdb->prefix}shiftora_migrations (
			id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
			connection_id BIGINT UNSIGNED NOT NULL,
			direction VARCHAR(16) NOT NULL,
			status VARCHAR(32) NOT NULL DEFAULT 'pending',
			tables LONGTEXT NOT NULL,
			tables_json LONGTEXT NULL,
			stages_json LONGTEXT NULL,
			table_offsets_json LONGTEXT NULL,
			staged_tables_json LONGTEXT NULL,
			dry_run TINYINT(1) NOT NULL DEFAULT 0,
			total_tables INT NOT NULL DEFAULT 0,
			processed_tables INT NOT NULL DEFAULT 0,
			total_rows BIGINT NOT NULL DEFAULT 0,
			processed_rows BIGINT NOT NULL DEFAULT 0,
			total_files INT NOT NULL DEFAULT 0,
			transferred_files INT NOT NULL DEFAULT 0,
			bytes_total BIGINT NOT NULL DEFAULT 0,
			bytes_transferred BIGINT NOT NULL DEFAULT 0,
				file_offsets_json LONGTEXT NULL DEFAULT NULL,
				current_stage VARCHAR(20) DEFAULT NULL,
				stage_progress TINYINT UNSIGNED NOT NULL DEFAULT 0,
				stage_plan_json LONGTEXT NULL DEFAULT NULL,
				completed_stages_json LONGTEXT NULL DEFAULT NULL,
				find_replace_rules LONGTEXT NULL,
			find_replace_json LONGTEXT NULL,
			backup_id BIGINT UNSIGNED NULL DEFAULT NULL,
			error_message TEXT NULL DEFAULT NULL,
			recovery_error TEXT NULL DEFAULT NULL,
			recovery_status VARCHAR(32) NULL DEFAULT NULL,
			remote_finalize_status VARCHAR(32) NULL DEFAULT NULL,
			remote_finalize_tables_json LONGTEXT NULL DEFAULT NULL,
			transfer_uuid VARCHAR(32) NULL DEFAULT NULL,
			remote_max_post_bytes BIGINT UNSIGNED NULL DEFAULT NULL,
			started_at DATETIME NULL DEFAULT NULL,
			completed_at DATETIME NULL DEFAULT NULL,
			created_at DATETIME NOT NULL,
			updated_at DATETIME NOT NULL,
			last_heartbeat_at DATETIME NULL DEFAULT NULL,
			worker_lock_owner CHAR(32) NULL DEFAULT NULL,
			worker_lock_expires DATETIME NULL DEFAULT NULL,
			PRIMARY KEY (id),
			KEY connection_id (connection_id),
			KEY status (status)
		) $charset_collate;";

		$sql_logs = "CREATE TABLE {$wpdb->prefix}shiftora_logs (
			id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
			migration_id BIGINT UNSIGNED NOT NULL,
			level VARCHAR(16) NOT NULL DEFAULT 'info',
			message TEXT NOT NULL,
			context LONGTEXT NULL,
			created_at DATETIME NOT NULL,
			PRIMARY KEY (id),
			KEY migration_id (migration_id),
			KEY level (level),
			KEY created_at (created_at)
		) $charset_collate;";

		$sql_backups = "CREATE TABLE {$wpdb->prefix}shiftora_backups (
			id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
			migration_id BIGINT UNSIGNED NULL DEFAULT NULL,
			connection_id BIGINT UNSIGNED NULL DEFAULT NULL,
			label VARCHAR(255) NOT NULL DEFAULT '',
			file_path VARCHAR(2083) NOT NULL,
			file_size BIGINT NOT NULL DEFAULT 0,
			tables LONGTEXT NOT NULL,
			tables_json LONGTEXT NULL,
			status VARCHAR(32) NOT NULL DEFAULT 'complete',
			created_at DATETIME NOT NULL,
			expires_at DATETIME NULL DEFAULT NULL,
			PRIMARY KEY (id),
			KEY migration_id (migration_id),
			KEY connection_id (connection_id),
			KEY status (status),
			KEY created_at (created_at)
		) $charset_collate;";

		// Committed-chunk ledger for idempotent SQL import replay: a retry returns the
		// saved receipt instead of re-running the INSERTs. The DDL lives in the store.
		$sql_import_chunks = WpChunkReceiptStore::table_ddl( $wpdb->prefix . 'shiftora_import_chunks', $charset_collate );

		// Per-file progress, so a status update touches one row instead of re-encoding
		// the manifest. virtual_path is 512 because deep vendor and uploads paths outgrow
		// 255 (about 2.1 KB as a utf8mb4 key, inside the 3 KB DYNAMIC limit); the binary
		// collation keeps IMG.JPG and img.jpg distinct.
		$sql_file_progress = "CREATE TABLE {$wpdb->prefix}shiftora_file_progress (
			migration_id BIGINT UNSIGNED NOT NULL,
			stage VARCHAR(20) NOT NULL,
			sequence INT UNSIGNED NOT NULL DEFAULT 0,
			virtual_path VARCHAR(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
			status VARCHAR(16) NOT NULL DEFAULT 'pending',
			byte_total BIGINT UNSIGNED NOT NULL DEFAULT 0,
			byte_done BIGINT UNSIGNED NOT NULL DEFAULT 0,
			content_hash VARCHAR(64) NOT NULL DEFAULT '',
			errors INT UNSIGNED NOT NULL DEFAULT 0,
			skip_batch TINYINT(1) NOT NULL DEFAULT 0,
			updated_at DATETIME NOT NULL,
			PRIMARY KEY (migration_id, stage, virtual_path),
			KEY sequence (migration_id, stage, status, sequence)
		) $charset_collate;";

		// What the last worker learned about the link, per migration and stage, so the
		// next run does not restart from the default batch size. Zero bytes_per_second means unmeasured.
		$sql_transfer_sizing = "CREATE TABLE {$wpdb->prefix}shiftora_transfer_sizing (
			migration_id BIGINT UNSIGNED NOT NULL,
			stage VARCHAR(20) NOT NULL,
			budget_bytes BIGINT UNSIGNED NOT NULL DEFAULT 0,
			fast_streak INT UNSIGNED NOT NULL DEFAULT 0,
			bytes_per_second BIGINT UNSIGNED NOT NULL DEFAULT 0,
			updated_at DATETIME NOT NULL,
			PRIMARY KEY (migration_id, stage)
		) $charset_collate;";

		// Staged restore session: which tables verified before a crash and whether promotion committed. DDL in RestoreTable.
		$sql_restore_ledger = RestoreTable::table_ddl( $wpdb->prefix . 'shiftora_restore_ledger', $charset_collate );

		// Restore-point pins; retention reads this rather than a transient a cache flush could wipe. DDL in BackupPinStore.
		$sql_backup_pins = BackupPinStore::table_ddl( $wpdb->prefix . 'shiftora_backup_pins', $charset_collate );

		// Saved migration profiles: named, re-runnable wizard configurations. A form-prefill convenience only, so no token or remote state.
		$sql_profiles = "CREATE TABLE {$wpdb->prefix}shiftora_profiles (
			id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
			label VARCHAR(255) NOT NULL,
			connection_id BIGINT UNSIGNED NOT NULL,
			direction VARCHAR(16) NOT NULL,
			stages_json LONGTEXT NULL,
			created_at DATETIME NOT NULL,
			updated_at DATETIME NOT NULL,
			last_used_at DATETIME NULL DEFAULT NULL,
			PRIMARY KEY (id),
			KEY connection_id (connection_id)
		) $charset_collate;";

		dbDelta( $sql_connections );
		dbDelta( $sql_migrations );
		dbDelta( $sql_logs );
		dbDelta( $sql_backups );
		if ( $include_ledger ) {
			dbDelta( $sql_import_chunks );
		}
		dbDelta( $sql_file_progress );
		dbDelta( $sql_transfer_sizing );
		dbDelta( $sql_restore_ledger );
		dbDelta( $sql_backup_pins );
		dbDelta( $sql_profiles );
	}

	/**
	 * Creates or updates every table, then verifies the ledger identity. Throws
	 * instead of returning false so the schema is stamped only after a full pass.
	 */
	private static function create_tables(): void {
		self::apply_table_ddl( true );

		if ( ! self::ensure_ledger_identity() ) {
			throw new LedgerSchemaException(
				'The chunk-ledger identity did not converge; the schema revision was not advanced.'
			);
		}
	}

	/** Sets initial plugin option defaults, only where not already set. */
	private static function set_defaults(): void {
		$defaults = [
			'shiftora_chunk_size'       => 10000,
			'shiftora_request_timeout'  => HttpClient::INSTALLED_DEFAULT_TIMEOUT,
			'shiftora_ssl_verify'       => HttpClient::TLS_AUTO,
			'shiftora_allow_pull'       => true,
		];

		foreach ( $defaults as $key => $value ) {
			if ( false === get_option( $key ) ) {
				add_option( $key, $value, '', false );
			}
		}

		// Site-wide secret other sites use to initiate connections.
		if ( ! get_option( 'shiftora_secret_key' ) ) {
			add_option( 'shiftora_secret_key', TokenGenerator::generate_secret(), '', false );
		}
	}
}