<?php
/**
 * Transfer lease registry for destination-side concurrency isolation.
 *
 * @package Shiftora\Migration
 */

declare(strict_types=1);

namespace Shiftora\Migration;

use InvalidArgumentException;
use RuntimeException;
use Shiftora\Support\Db\TableDdl;
use Shiftora\Support\TableNameScoper;

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

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

/**
 * One active transfer per connection, tracked in a self-provisioned table.
 *
 * The connection_id primary key makes the claim atomic: a second start either
 * reclaims a dead lease or fails the insert. The row doubles as the commit lock.
 * Finalize flips active -> finalizing before any RENAME, and rollback/cleanup are
 * refused while finalizing, so a cancel can't land mid-rename.
 */
final class TransferRegistry {

	public const TTL = 90;

	/**
	 * How long a lease may sit mid-finalize, mid-rollback or mid-cancel without a write
	 * before it counts as stranded. Those states hold the commit intent and rollback
	 * checkpoint, so the window is generous: reclaiming early would erase the only
	 * record of how to undo a slow constraint pass.
	 */
	public const STRANDED_AFTER = 1800;

	/** Cached existence check to avoid re-running DDL on every request. */
	private bool $table_ready = false;

	/** @var callable(): string */
	private $clock;

	public function __construct(
		private readonly \wpdb $wpdb,
		?callable $clock = null
	) {
		$this->clock = $clock ?? static fn(): string => current_time( 'mysql', true );
	}

	/**
	 * Override the clock (inject a fixed timestamp; production keeps
	 * current_time).
	 */
	public function set_clock( callable $clock ): void {
		$this->clock = $clock;
	}

	public function now(): string {
		return ( $this->clock )();
	}

	private function table(): string {
		return $this->wpdb->prefix . 'shiftora_transfers';
	}

	private function ensure_table(): void {
		$wpdb = $this->wpdb;

		if ( $this->table_ready ) {
			return;
		}

		$table   = $this->table();
		$collate = $wpdb->get_charset_collate();

		$ddl = "CREATE TABLE IF NOT EXISTS `{$table}` (
			connection_id BIGINT UNSIGNED NOT NULL,
			sender_migration_id BIGINT UNSIGNED NOT NULL,
			transfer_uuid CHAR(32) NOT NULL,
			status VARCHAR(16) NOT NULL DEFAULT 'active',
			tables_json LONGTEXT NOT NULL,
			prefix VARCHAR(64) NOT NULL DEFAULT '',
			schemas_json LONGTEXT NULL DEFAULT NULL,
			commit_receipt_json LONGTEXT NULL DEFAULT NULL,
			commit_intent_json LONGTEXT NULL DEFAULT NULL,
			rollback_state_json LONGTEXT NULL DEFAULT NULL,
			staging_names_json LONGTEXT NULL DEFAULT NULL,
			owner_token CHAR(32) NOT NULL,
			expires_at DATETIME NOT NULL,
			created_at DATETIME NOT NULL,
			finalized_at DATETIME NULL DEFAULT NULL,
			updated_at DATETIME NOT NULL,
			PRIMARY KEY (connection_id)
		) {$collate}";

		// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared, PluginCheck.Security.DirectDB.UnescapedDBParameter, WordPress.DB.DirectDatabaseQuery -- this plugin's own CREATE TABLE IF NOT EXISTS, built from the site prefix; DDL cannot be bound; the transfer reads and writes the site's own tables, a cached copy would be stale
		$result = $wpdb->query( $ddl );

		if ( false === $result ) {
			throw new RuntimeException(
				"Failed to create transfer lease table `{$table}`: {$wpdb->last_error}"
			);
		}

		// Create includes the columns for fresh installs; older installs get
		// an idempotent ALTER per missing column. This table is upgraded here
		// rather than by the plugin Installer.
		$this->ensure_columns();

		$this->table_ready = true;
	}

	private function ensure_columns(): void {
		// Column name => whether it may be NULL. Every one is LONGTEXT.
		$columns = [
			'tables_json'         => false,
			'schemas_json'        => true,
			'commit_receipt_json' => true,
			'commit_intent_json'  => true,
			'rollback_state_json' => true,
			'staging_names_json'  => true,
		];
		$wpdb  = $this->wpdb;
		$table = $this->table();
		foreach ( $columns as $name => $nullable ) {
			$data_type = $this->column_data_type( $table, $name );
			if ( null === $data_type ) {
				$result = $nullable
					// 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( 'ALTER TABLE %i ADD COLUMN %i LONGTEXT NULL DEFAULT NULL', $table, $name ) )
					// 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( 'ALTER TABLE %i ADD COLUMN %i LONGTEXT NOT NULL', $table, $name ) );
				if ( false === $result ) {
					throw new RuntimeException(
						"Failed to add column {$name} on `{$table}`: {$wpdb->last_error}"
					);
				}
				continue;
			}
			// Older installs created these JSON columns as TEXT. A manifest can
			// exceed 64KB, so a narrow column is promoted to LONGTEXT.
			if ( 'longtext' !== strtolower( (string) $data_type ) ) {
				$result = $nullable
					// 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( 'ALTER TABLE %i MODIFY COLUMN %i LONGTEXT NULL DEFAULT NULL', $table, $name ) )
					// 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( 'ALTER TABLE %i MODIFY COLUMN %i LONGTEXT NOT NULL', $table, $name ) );
				if ( false === $result ) {
					throw new RuntimeException(
						"Failed to promote column {$name} on `{$table}`: {$wpdb->last_error}"
					);
				}
			}
		}

		// Re-read once after the pass. An ALTER that reported success but left
		// the column missing (racing DDL, replication lag) must not be trusted.
		foreach ( $columns as $name => $nullable ) {
			if ( null === $this->column_data_type( $table, $name ) ) {
				throw new RuntimeException(
					"Column {$name} is still missing from `{$table}` after the schema pass."
				);
			}
		}
	}

	/**
	 * The information_schema DATA_TYPE for a transfer-table column, or null when it
	 * doesn't exist yet. A failed read throws so a server problem isn't masked by a
	 * bogus ALTER.
	 */
	private function column_data_type( string $table, string $name ): ?string {
		$wpdb = $this->wpdb;

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- server metadata, nothing to cache
		$type = $wpdb->get_var(
			$wpdb->prepare(
				'SELECT DATA_TYPE FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = %s AND column_name = %s',
				$table,
				$name
			)
		);
		if ( '' !== (string) $wpdb->last_error ) {
			throw new RuntimeException(
				"Could not read metadata for column {$name} on `{$table}`: {$wpdb->last_error}"
			);
		}
		return null === $type ? null : (string) $type;
	}

	/**
	 * Claim the one lease a connection may hold.
	 *
	 * One INSERT does the whole job. MySQL applies ON DUPLICATE assignments left to
	 * right and re-reads the row after each, so a guard repeated per column would see
	 * the status already flipped to 'active' and leave other columns stale. The guard
	 * is captured once into @reclaim on the first assignment, which keeps the row
	 * atomic and portable to MySQL 5.7 and MariaDB. A live lease stays byte-identical
	 * and reports 0 affected rows ("busy"); a created or reclaimed one reports 1 or 2.
	 *
	 * Expiry only reclaims a still-active lease. A finalizing or finalized row keeps
	 * its recovery context however old it is, since a pause can outlive the TTL and
	 * retry and rollback depend on it. Reclaiming a dead lease wipes its recovery
	 * columns so the new transfer inherits no stale checkpoint.
	 *
	 * @param string[]    $bare_tables    Bare (non-prefixed) table names.
	 * @param string      $prefix         Remote DB prefix.
	 * @param string|null $requested_uuid Replay identity from an earlier start; rejected when malformed, reused on reclaim.
	 *
	 * @return array{transfer_id: string, expires_in: int}
	 * @throws TransferAlreadyActiveException When a live lease exists.
	 */
	public function claim(
		int $connection_id,
		int $sender_migration_id,
		array $bare_tables,
		string $prefix = '',
		?string $requested_uuid = null
	): array {
		$wpdb = $this->wpdb;

		if ( null !== $requested_uuid && ! preg_match( '/^[0-9a-f]{32}$/', $requested_uuid ) ) {
			throw new InvalidArgumentException( sprintf(
				'Invalid transfer uuid: %s',
				$requested_uuid
			) );
		}

		$this->ensure_table();

		$now     = $this->now();
		$expires = gmdate( 'Y-m-d H:i:s', strtotime( $now ) + self::TTL );
		$uuid    = $requested_uuid ?? bin2hex( random_bytes( 16 ) );
		$tables  = wp_json_encode( array_values( $bare_tables ) ) ?: '[]';
		// A lease nobody will finish must not hold the slot forever. `active` lapses on
		// its own and settled statuses are free at once. In-flight states hold the commit
		// intent and checkpoint, so they are reclaimed only after STRANDED_AFTER.
		$stranded = gmdate( 'Y-m-d H:i:s', strtotime( $now ) - self::STRANDED_AFTER );

		// Staging names are fixed at claim time so the finalize discovery pass
		// reconciles against what this transfer actually created.
		$staging_plan = [];
		if ( [] !== $bare_tables ) {
			$token = substr( md5( $uuid ), 0, 16 );
			foreach ( array_values( $bare_tables ) as $bare ) {
				$bare                    = (string) $bare;
				$staging_plan[ $bare ] = TableNameScoper::scoped_table_name( (string) $wpdb->prefix, 'shiftora_stg_', $token, $bare );
			}
		}
		$staging_names = wp_json_encode( $staging_plan ) ?: '[]';

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$affected = (int) $wpdb->query(
			$wpdb->prepare(
				"INSERT INTO %i
					(connection_id, sender_migration_id, transfer_uuid, status, tables_json, prefix, staging_names_json, owner_token, expires_at, created_at, updated_at)
				 VALUES (%d, %d, %s, 'active', %s, %s, %s, %s, %s, %s, %s)
				 ON DUPLICATE KEY UPDATE
					sender_migration_id = IF(
						(@reclaim := (
							status NOT IN ('active','finalizing','rolling_back','cancelling')
							OR (status = 'active' AND expires_at < %s)
							OR (status IN ('finalizing','rolling_back','cancelling') AND updated_at < %s)
						)),
						VALUES(sender_migration_id),
						sender_migration_id
					),
					transfer_uuid = IF(@reclaim, VALUES(transfer_uuid), transfer_uuid),
					status = IF(@reclaim, 'active', status),
					tables_json = IF(@reclaim, VALUES(tables_json), tables_json),
					prefix = IF(@reclaim, VALUES(prefix), prefix),
					staging_names_json = IF(@reclaim, VALUES(staging_names_json), staging_names_json),
					owner_token = IF(@reclaim, VALUES(owner_token), owner_token),
					expires_at = IF(@reclaim, VALUES(expires_at), expires_at),
					schemas_json = IF(@reclaim, NULL, schemas_json),
					commit_receipt_json = IF(@reclaim, NULL, commit_receipt_json),
					commit_intent_json = IF(@reclaim, NULL, commit_intent_json),
					rollback_state_json = IF(@reclaim, NULL, rollback_state_json),
					finalized_at = IF(@reclaim, NULL, finalized_at),
					created_at = IF(@reclaim, VALUES(created_at), created_at),
					updated_at = IF(@reclaim, VALUES(updated_at), updated_at)",
				$this->table(),
				$connection_id,
				$sender_migration_id,
				$uuid,
				$tables,
				$prefix,
				$staging_names,
				$uuid,
				$expires,
				// created_at, updated_at, then the two guard bounds: expiry for an
				// active lease and the stranded cut-off for an in-flight one. The
				// count must match exactly or wpdb::prepare rejects the call.
				$now, $now, $now, $stranded
			)
		);

		if ( 0 === $affected ) {
			// The INSERT above matched the live lease and changed nothing. If the caller's
			// uuid is the active transfer's, this is a start replayed after a lost response,
			// so hand back that lease. Anything else is a real busy rejection.
			if ( null !== $requested_uuid ) {
				$active = $this->find_active( $connection_id, $sender_migration_id );
				if ( is_array( $active ) && $active['transfer_uuid'] === $requested_uuid ) {
					return [ 'transfer_id' => $requested_uuid, 'expires_in' => self::TTL ];
				}
			}

			throw new TransferAlreadyActiveException( sprintf(
				'[shiftora:busy] A transfer is already active for connection %d.',
				$connection_id
			) );
		}

		// Taking the slot means whatever the previous holder staged is garbage.
		// A process that died never ran mark_failed, and sweeping here lets the
		// new transfer start on an empty namespace instead of failing its own
		// finalize check on the old debris.
		$this->prune_dead_staging();

		return [ 'transfer_id' => $uuid, 'expires_in' => self::TTL ];
	}

	/**
	 * The staging-name plan recorded at claim time, as bare => full name. Empty for
	 * legacy rows, where finalize re-derives names from the token.
	 *
	 * @return array<string, string>
	 */
	public function staging_plan( int $connection_id ): array {
		$wpdb = $this->wpdb;

		$this->ensure_table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				'SELECT staging_names_json FROM %i WHERE connection_id = %d LIMIT 1',
				$this->table(),
				$connection_id
			),
			ARRAY_A
		);
		$raw = (string) ( $rows[0]['staging_names_json'] ?? '' );
		if ( '' === $raw ) {
			return [];
		}
		$plan = json_decode( $raw, true );

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

	/**
	 * The staging names owned by every other live lease in the destination.
	 *
	 * A concurrent transfer on another connection shares the shiftora_stg_ namespace,
	 * and this union lets finalize tell a live sibling from an orphan. The optional
	 * uuid excludes the caller's own lease.
	 *
	 * @return list<string> Full staging table names.
	 */
	public function live_staging_names( string $except_transfer_uuid = '' ): array {
		$this->ensure_table();

		$wpdb = $this->wpdb;
		$rows = '' === $except_transfer_uuid
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
			? $wpdb->get_results(
				$wpdb->prepare( "SELECT staging_names_json FROM %i WHERE status IN ('active','cancelling','finalizing','rolling_back')", $this->table() ),
				ARRAY_A
			)
			: $wpdb->get_results(
				$wpdb->prepare(
					"SELECT staging_names_json FROM %i WHERE status IN ('active','cancelling','finalizing','rolling_back') AND transfer_uuid <> %s",
					$this->table(),
					$except_transfer_uuid
				),
				ARRAY_A
			);
		$names = [];
		foreach ( $rows as $row ) {
			$plan = json_decode( (string) ( $row['staging_names_json'] ?? '' ), true );
			if ( ! is_array( $plan ) ) {
				continue;
			}
			foreach ( $plan as $full ) {
				if ( is_string( $full ) && '' !== $full ) {
					$names[] = $full;
				}
			}
		}

		return array_values( array_unique( $names ) );
	}

	/**
	 * Renew the lease on the connection's active transfer. False when the row isn't
	 * an active match.
	 *
	 * The optional uuid is the ownership key, so a stale request with a foreign
	 * identity can't keep a lease alive.
	 */
	public function renew( int $connection_id, int $sender_migration_id, ?string $transfer_uuid = null ): bool {
		$this->ensure_table();

		$now     = $this->now();
		$expires = gmdate( 'Y-m-d H:i:s', strtotime( $now ) + self::TTL );

		$wpdb = $this->wpdb;

		$affected = (int) ( null === $transfer_uuid
			// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
			? $wpdb->query(
				$wpdb->prepare(
					"UPDATE %i
						SET expires_at = %s, updated_at = %s
					 WHERE connection_id = %d AND sender_migration_id = %d AND status = 'active'",
					$this->table(),
					$expires,
					$now,
					$connection_id,
					$sender_migration_id
				)
			)
			: $wpdb->query(
				$wpdb->prepare(
					"UPDATE %i
						SET expires_at = %s, updated_at = %s
					 WHERE connection_id = %d AND sender_migration_id = %d AND transfer_uuid = %s AND status = 'active'",
					$this->table(),
					$expires,
					$now,
					$connection_id,
					$sender_migration_id,
					$transfer_uuid
				)
			)
		);

		return $affected > 0;
	}

	/**
	 * The connection's active lease for the sender migration id.
	 *
	 * An absent row or mismatched id means "not a leased transfer"; callers must
	 * refuse the mutation.
	 *
	 * @return array<string, string>|null
	 */
	public function find_active( int $connection_id, int $sender_migration_id ): ?array {
		$wpdb = $this->wpdb;

		$this->ensure_table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT transfer_uuid, sender_migration_id, status, tables_json, prefix, owner_token, expires_at,
				        schemas_json, commit_receipt_json, commit_intent_json, rollback_state_json, staging_names_json
				   FROM %i
				  WHERE connection_id = %d AND status = 'active'
				  LIMIT 1",
				$this->table(),
				$connection_id
			),
			ARRAY_A
		);

		if ( empty( $rows ) || (int) $rows[0]['sender_migration_id'] !== $sender_migration_id ) {
			return null;
		}

		return $rows[0];
	}

	/**
	 * The connection's lease for a migration regardless of status, so terminal rows
	 * survive for rollback/cleanup token resolution.
	 *
	 * @return array<string, string>|null
	 */
	public function find( int $connection_id, int $sender_migration_id ): ?array {
		$wpdb = $this->wpdb;

		$this->ensure_table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				'SELECT transfer_uuid, sender_migration_id, status, tables_json, prefix, owner_token, expires_at,
				        schemas_json, commit_receipt_json, commit_intent_json, rollback_state_json, staging_names_json
				   FROM %i
				  WHERE connection_id = %d AND sender_migration_id = %d
				  LIMIT 1',
				$this->table(),
				$connection_id,
				$sender_migration_id
			),
			ARRAY_A
		);

		return $rows[0] ?? null;
	}

	/**
	 * Record a dead transfer so the row is reclaimable.
	 *
	 * Guarded only by the owner token: a finalize failure and a post-commit rollback
	 * both end here. A rollback passes $clear_receipt so the receipt can't answer
	 * "already finalized" for a transfer that was undone, and $rollback_state so a
	 * repeated rollback replies from what happened rather than inferring ownership
	 * from current tables.
	 *
	 * @param array{status: int, restored: string[], failed: array{table: string, reason: string}[]}|null $rollback_state
	 */
	public function mark_failed( int $connection_id, string $owner_token, bool $clear_receipt = false, ?array $rollback_state = null ): bool {
		$wpdb = $this->wpdb;

		$this->ensure_table();

		$now = $this->now();

		// The receipt is only cleared and the rollback state only written when asked;
		// IF() keeps the stored value otherwise.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$affected = (int) $wpdb->query(
			$wpdb->prepare(
				"UPDATE %i SET status = 'failed', finalized_at = %s, updated_at = %s,
					commit_receipt_json = IF(%d, NULL, commit_receipt_json),
					rollback_state_json = IF(%d, %s, rollback_state_json)
				 WHERE connection_id = %d AND owner_token = %s",
				$this->table(),
				$now,
				$now,
				$clear_receipt ? 1 : 0,
				null !== $rollback_state ? 1 : 0,
				null !== $rollback_state ? ( wp_json_encode( $rollback_state ) ?: '{"status":500}' ) : '',
				$connection_id,
				$owner_token
			)
		);

		if ( $affected > 0 ) {
			$this->prune_dead_staging();
		}

		return $affected > 0;
	}

	/**
	 * Drop staging tables that no live lease owns.
	 *
	 * A dead transfer keeps its staging tables, and finalize would read those leftovers
	 * as leaks and block the connection's next push. Ownership is the lease table:
	 * anything an active/cancelling/finalizing/rolling_back lease plans belongs to a
	 * live sibling. Run after the caller has flipped its own row.
	 *
	 * @return int Number of tables dropped.
	 */
	public function prune_dead_staging(): int {
		$this->ensure_table();

		// A local pull holds no lease, so its record is what keeps its tables. One that
		// has not recorded them yet cannot be proven dead, so nothing is dropped.
		$live = $this->protected_staging_names();
		if ( null === $live ) {
			return 0;
		}

		$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_' ) . '%'
			)
		);

		$dropped = 0;
		foreach ( array_diff( $observed, $live ) as $table ) {
			$table = (string) $table;
			if ( '' === $table ) {
				continue;
			}
			$dropped += (int) TableDdl::drop( $wpdb, $table );
		}

		return $dropped;
	}

	/**
	 * Staging names that must survive any sweep: what live leases plan and what local
	 * pulls in progress have staged. Null when a pull has not recorded its tables yet,
	 * which means nothing can be called dead.
	 *
	 * @return string[]|null
	 */
	public function protected_staging_names(): ?array {
		$pulls = $this->in_progress_pull_staging();

		return null === $pulls ? null : array_values( array_unique( array_merge( $this->live_staging_names(), $pulls ) ) );
	}

	/**
	 * Drops day-old staging tables that nothing live owns. Called when an import
	 * starts, so debris from a crashed transfer does not pile up on a full disk.
	 *
	 * Live means planned by a lease, or staged by a local pull still in progress: a
	 * pull holds no lease, so its own record is the proof. A pull that has not
	 * recorded what it staged cannot be proven dead, so the sweep waits.
	 *
	 * @return string[] Dropped names.
	 */
	public function sweep_stale_staging(): array {
		$protected = $this->protected_staging_names();

		return null === $protected ? [] : ( new StagingSweeper( $this->wpdb ) )->sweep_stale( $protected );
	}

	/**
	 * Staging names of the local pulls that are still going, or null when one of them
	 * has not recorded its tables yet.
	 *
	 * @return string[]|null
	 */
	private function in_progress_pull_staging(): ?array {
		$wpdb = $this->wpdb;
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$rows = (array) $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, direction, staged_tables_json FROM %i
				  WHERE status IN ('pending','running','paused','cancelling') AND direction = 'pull'",
				$wpdb->prefix . 'shiftora_migrations'
			)
		);

		$names = [];
		foreach ( $rows as $row ) {
			$staged = json_decode( (string) ( $row->staged_tables_json ?? '' ), true );
			if ( ! is_array( $staged ) || [] === $staged ) {
				return null;
			}

			foreach ( $staged as $bare ) {
				// A pull stages under unscoped names, so there is no token to compare.
				$names[] = TableNameScoper::scoped_table_name( (string) $this->wpdb->prefix, 'shiftora_stg_', '', (string) $bare );
			}
		}

		return $names;
	}

	/**
	 * Delete every lease that has not reached a terminal state.
	 *
	 * The operator flush uses it to free the connection. Terminal rows stay as the
	 * history diagnostics and rollback read. This discards recovery state (commit
	 * intent, rollback state), so an interrupted finalize can't be resumed. That is why
	 * it is its own method, not a flag on normal expiry.
	 *
	 * @return int Rows removed.
	 */
public function purge_live_leases(): int {
		$wpdb = $this->wpdb;

		$this->ensure_table();

		// A finalized row records a completed transfer and is kept, so a
		// connection holding one is not wedged by the button. claim() reclaims
		// it, since it takes any status that is not still active.
		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		return (int) $wpdb->query(
			$wpdb->prepare(
				"DELETE FROM %i
				  WHERE status IN ('active','cancelling','finalizing','rolling_back')",
				$this->table()
			)
		);
	}

	/**
	 * Whether a live destination operation holds the connection slot apart from the
	 * given migration.
	 *
	 * A local pull finalizes and rolls back staging on the destination, so overlapping
	 * pulls of one connection could both reach the commit rename. The engine checks
	 * this before finalize and rollback.
	 *
	 * @return bool True when another migration's lease is active or finalizing.
	 */
	public function destination_held_by_other( int $connection_id, int $excluding_migration_id ): bool {
		$wpdb = $this->wpdb;

		$this->ensure_table();

		// phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required
		$rows = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT connection_id FROM %i
				 WHERE connection_id = %d
				   AND status IN ('active','finalizing')
				   AND sender_migration_id <> %d
				 LIMIT 1",
				$this->table(),
				$connection_id,
				$excluding_migration_id
			),
			ARRAY_A
		);

		return ! empty( $rows );
	}

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