<?php
/**
 * Composite-key keyset cursor support.
 *
 * @package Shiftora\Migration
 */

declare(strict_types=1);

namespace Shiftora\Migration;

use InvalidArgumentException;

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

/**
 * Encodes, decodes and renders keyset cursors for tables whose primary key spans several columns (e.g. term_relationships).
 *
 * A single-column cursor (`WHERE pk > %s`) silently skips rows that share the
 * first key value once the cursor moves past it, so resumption needs a
 * lexicographic comparison over the whole key tuple. The cursor travels as a
 * JSON-encoded tuple in the existing `last_pk` string field, so callers that pass
 * it through REST payloads or migration records are untouched.
 */
final class KeyCursor {

	/**
	 * Encode a tuple of key values into a transport string.
	 *
	 * @param list<string>|null $values Ordered key values, or null for "no cursor yet".
	 *
	 * @return string|null Null when there is no cursor.
	 */
	public static function encode( ?array $values ): ?string {
		if ( null === $values || [] === $values ) {
			return null;
		}

		// json_encode() fails on invalid UTF-8 (binary or latin1 keys), which would yield an empty cursor and restart the table every page. Such tuples go under a base64 key.
		foreach ( $values as $value ) {
			if ( 1 !== preg_match( '//u', (string) $value ) ) {
				return (string) json_encode( [ 'b64' => array_map( static fn( $v ): string => base64_encode( (string) $v ), array_values( $values ) ) ] );
			}
		}

		return (string) json_encode( $values, JSON_UNESCAPED_UNICODE );
	}

	/**
	 * Decode a transport string back into a tuple. Accepts a JSON tuple or a plain scalar (single-column cursor).
	 *
	 * @return list<string>|null Null when there is no cursor.
	 */
	public static function decode( ?string $encoded ): ?array {
		if ( null === $encoded || '' === $encoded ) {
			return null;
		}

		$as_array = json_decode( $encoded, true );
		if ( is_array( $as_array ) && isset( $as_array['b64'] ) && is_array( $as_array['b64'] ) ) {
			return array_map( static fn( $v ): string => (string) base64_decode( (string) $v, true ), array_values( $as_array['b64'] ) );
		}
		if ( is_array( $as_array ) ) {
			return array_map( 'strval', array_values( $as_array ) );
		}

		// Plain scalar: wrap it in a one-element tuple.
		return [ $encoded ];
	}

	/**
	 * Reads one page of a table in a stable order: by key column when there is one,
	 * by an offset when the run began with offset paging or has no key.
	 *
	 * Without a cursor the page starts at the top, or $offset rows in. With one, the
	 * WHERE is a lexicographic "greater than" over the whole key tuple. For columns
	 * [a, b, c] and values [1, 2, 3] that is
	 *   (a > ?) OR (a = ? AND b > ?) OR (a = ? AND b = ? AND c > ?)
	 * so rows sharing a first key value are not skipped once the cursor moves on.
	 *
	 * @param list<string>                        $columns    Ordered column names, read from the database; empty pages unordered.
	 * @param list<string>|null                   $cursor     Key values of the previous page's last row, or null.
	 * @param list<array{0: string, 1: string, 2: string}> $exclusions Rows to leave out as [column, 'NOT LIKE' or '<>', value].
	 *
	 * @return array<int, array<string, mixed>>|null Rows as get_results() reports them, null when the query failed.
	 *
	 * @throws InvalidArgumentException When the cursor does not match the columns or an exclusion is malformed.
	 */
	public static function fetch_page( \wpdb $wpdb, string $table, array $columns, ?array $cursor, int $limit, int $offset = 0, array $exclusions = [] ): ?array {
		if ( null !== $cursor && [] === $columns ) {
			throw new InvalidArgumentException( 'KeyCursor requires at least one key column.' );
		}
		if ( null !== $cursor && count( $columns ) !== count( $cursor ) ) {
			throw new InvalidArgumentException( 'KeyCursor value count must match key column count.' );
		}

		$args       = [ $table ];
		$conditions = [];

		foreach ( $exclusions as [ $column, $operator, $value ] ) {
			if ( ! in_array( $operator, [ 'NOT LIKE', '<>' ], true ) ) {
				throw new InvalidArgumentException( 'KeyCursor exclusions only take NOT LIKE or <>.' );
			}
			$conditions[] = '%i ' . $operator . ' %s';
			$args[]       = $column;
			$args[]       = $value;
		}

		if ( null !== $cursor ) {
			$groups = [];
			foreach ( $columns as $index => $column ) {
				$clauses = [];
				for ( $eq = 0; $eq < $index; $eq++ ) {
					$clauses[] = '%i = %s';
					$args[]    = $columns[ $eq ];
					$args[]    = $cursor[ $eq ];
				}
				$clauses[] = '%i > %s';
				$args[]    = $column;
				$args[]    = $cursor[ $index ];
				$groups[]  = '(' . implode( ' AND ', $clauses ) . ')';
			}
			// Exclusions are ANDed in, so the OR of groups needs its own parentheses then.
			$conditions[] = [] === $exclusions ? implode( ' OR ', $groups ) : '(' . implode( ' OR ', $groups ) . ')';
		}

		$format = 'SELECT * FROM %i';
		if ( [] !== $conditions ) {
			$format .= ' WHERE ' . implode( ' AND ', $conditions );
		}
		if ( [] !== $columns ) {
			$format .= ' ORDER BY ' . implode( ', ', array_fill( 0, count( $columns ), '%i' ) );
			$args    = [ ...$args, ...$columns ];
		}

		$format .= ' LIMIT %d';
		$args[]  = $limit;

		if ( null === $cursor && ( $offset > 0 || [] === $columns ) ) {
			$format .= ' OFFSET %d';
			$args[]  = $offset;
		}

		// phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared, PluginCheck.Security.DirectDB.UnescapedDBParameter, WordPress.DB.DirectDatabaseQuery -- the format holds only %i, %s and %d placeholders, so every name and value is bound by prepare(); the transfer reads and writes the site's own tables, a cached copy would be stale
		return $wpdb->get_results( $wpdb->prepare( $format, $args ), ARRAY_A );
	}

	/**
	 * Pick the best keyset column tuple from raw SHOW INDEX rows.
	 *
	 * Row order is not guaranteed across engines and MariaDB rejects ORDER BY on it
	 * (error 1064), so the rows are sorted here. A key with a prefix segment
	 * (Sub_part) or expression column (Column_name "%", or NULL for functional
	 * indexes) is not unique over full values, so the whole key is rejected. PRIMARY
	 * wins; otherwise the first unique, fully NOT NULL index.
	 *
	 * @param list<array<string, mixed>> $indexes SHOW INDEX result rows.
	 *
	 * @return list<string> Ordered key column names, empty when none usable.
	 */
	public static function pick_keyset_columns( array $indexes ): array {
		if ( [] === $indexes ) {
			return [];
		}

		usort( $indexes, static function ( array $a, array $b ): int {
			$cmp = strcmp( (string) ( $a['Key_name'] ?? '' ), (string) ( $b['Key_name'] ?? '' ) );
			return 0 !== $cmp
				? $cmp
				: ( (int) ( $a['Seq_in_index'] ?? 0 ) <=> (int) ( $b['Seq_in_index'] ?? 0 ) );
		} );

		/** @var array<string, array{unique: bool, null_ok: bool, columns: list<string>} > $candidates */
		$candidates = [];
		/** @var array<string, true> $unusable */
		$unusable = [];

		foreach ( $indexes as $row ) {
			$key_name   = (string) ( $row['Key_name'] ?? '' );
			$raw_column = $row['Column_name'] ?? null;
			$column     = null === $raw_column ? '' : (string) $raw_column;

			if ( '' === $key_name ) {
				continue;
			}

			// A NULL Column_name is an expression segment (functional index) and, like "%" or a prefix, makes the whole index unusable.
			if ( null === $raw_column || '%' === $column || null !== ( $row['Sub_part'] ?? null ) ) {
				$unusable[ $key_name ] = true;
				continue;
			}

			if ( '' === $column ) {
				continue;
			}

			if ( ! isset( $candidates[ $key_name ] ) ) {
				$candidates[ $key_name ] = [
					'unique'  => '0' === (string) ( $row['Non_unique'] ?? '1' ),
					'null_ok' => 'YES' === (string) ( $row['Null'] ?? 'NO' ),
					'columns' => [],
				];
			}
			$candidates[ $key_name ]['columns'][] = $column;
			$candidates[ $key_name ]['null_ok']   = $candidates[ $key_name ]['null_ok'] || 'YES' === (string) ( $row['Null'] ?? 'NO' );
		}

		if ( isset( $candidates['PRIMARY'] ) && ! isset( $unusable['PRIMARY'] ) ) {
			return $candidates['PRIMARY']['columns'];
		}

		foreach ( $candidates as $name => $key ) {
			if ( isset( $unusable[ $name ] ) || 'PRIMARY' === $name ) {
				continue;
			}
			if ( $key['unique'] && ! $key['null_ok'] ) {
				return $key['columns'];
			}
		}

		return [];
	}
}