<?php
/**
 * Validates an archived table schema before it is executed: exactly one plain `CREATE TABLE` for a table in the
 * archive's prefix, with an allow-listed set of table options. This replaces raw SQL replay, so it fails closed.
 *
 * A tokenizer that understands quoted strings and backtick identifiers does the work, so text inside quotes (a column
 * COMMENT 'select me; -- x') is never mistaken for SQL, while anything outside quotes that is not plain DDL is refused:
 * statement separators, comments of every kind (including MySQL's executable /*! … *\/), SELECT, partitioning,
 * DATA/INDEX DIRECTORY, TABLESPACE, MERGE UNION, CONNECTION and the FEDERATED/CONNECT engines.
 *
 * Charsets, collations and engines the server lacks are normalised to utf8mb4 / utf8mb4_unicode_ci / InnoDB, and
 * MariaDB-only table options are dropped on MySQL.
 *
 * @package Mench_Migrate
 */

defined( 'ABSPATH' ) || exit;
// phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Exception messages are internal machine codes; screens and REST answers show only the escaped sentence from Mench_Migrate_Rest::message().

class Mench_Migrate_Schema {

	private const STEP = 'import-db';

	private const FALLBACK_CHARSET   = 'utf8mb4';
	private const FALLBACK_COLLATION = 'utf8mb4_unicode_ci';
	private const FALLBACK_ENGINE    = 'InnoDB';

	/** Characters allowed outside quotes besides whitespace, words, strings and identifiers. */
	private const PUNCT = '(),=.-+*/<>!%&|^~:';

	/** Bare words refused anywhere inside the column/index body. */
	private const BODY_DENY = [ 'SELECT', 'LIKE', 'DIRECTORY', 'ENGINE', 'CONNECTION', 'PARTITION', 'PARTITIONS', 'TABLESPACE', 'FEDERATED', 'CONNECT', 'UNION' ];

	/** Table options allowed after the body (CHARACTER SET is read as CHARSET). Everything else is refused. */
	private const TABLE_OPTIONS = [
		'ENGINE', 'AUTO_INCREMENT', 'CHARSET', 'COLLATE', 'ROW_FORMAT', 'COMMENT', 'STATS_PERSISTENT', 'STATS_AUTO_RECALC',
		'STATS_SAMPLE_PAGES', 'KEY_BLOCK_SIZE', 'PACK_KEYS', 'CHECKSUM', 'PAGE_CHECKSUM', 'PAGE_COMPRESSED', 'TRANSACTIONAL',
		'DELAY_KEY_WRITE', 'AVG_ROW_LENGTH', 'MAX_ROWS', 'MIN_ROWS', 'COMPRESSION', 'ENCRYPTION',
	];

	private const DENIED_ENGINES = [ 'FEDERATED', 'CONNECT' ];

	/** MariaDB-only table options (Aria and friends), dropped when the server is MySQL, which rejects them. */
	private const MARIADB_OPTIONS = [ 'PAGE_CHECKSUM', 'PAGE_COMPRESSED', 'TRANSACTIONAL' ];

	/** @var array{charsets:array<string,true>,collations:array<string,string>,engines:array<string,true>,mariadb:bool}|null */
	private static ?array $server = null;

	/**
	 * Foreign keys are removed (they would bind the temp table to live or foreign tables and break the swap); each
	 * removal is reported in 'stripped' for the preflight warnings.
	 *
	 * @return array{table:string, sql:string, stripped:string[]} The table name, the (normalised) CREATE TABLE text and notes on what was removed.
	 * @throws Mench_Migrate_Db_Exception When the text is anything but one safe CREATE TABLE in the prefix.
	 */
	public static function validate( string $create_sql, string $expected_prefix ): array {
		$t = self::tokenize( $create_sql );
		$n = count( $t );

		if ( $n < 5 || ! self::is_word( $t[0], 'CREATE' ) || ! self::is_word( $t[1], 'TABLE' ) ) {
			self::reject( 'not a CREATE TABLE statement' );
		}
		$table = self::name_of( $t[2] );
		if ( '' === $expected_prefix || 0 !== strpos( $table, $expected_prefix ) || strlen( $table ) === strlen( $expected_prefix ) ) {
			self::reject( 'table outside the archive prefix' );
		}
		if ( ! self::is_punct( $t[3], '(' ) ) {
			self::reject( 'expected the column list after the table name' );
		}

		// Body: up to the matching parenthesis. Depth-1 commas split it into clauses (one column or index each).
		$depth   = 0;
		$end     = -1;
		$clauses = [ [] ];
		$seps    = [ null ]; // Token index of the comma before each clause.
		for ( $i = 3; $i < $n; $i++ ) {
			if ( self::is_punct( $t[ $i ], '(' ) ) {
				$depth++;
			} elseif ( self::is_punct( $t[ $i ], ')' ) ) {
				if ( 0 === --$depth ) {
					$end = $i;
					break;
				}
			} elseif ( 1 === $depth && self::is_punct( $t[ $i ], ',' ) ) {
				$clauses[] = [];
				$seps[]    = $i;
				continue;
			}
			if ( 'word' === $t[ $i ]['type'] && in_array( strtoupper( $t[ $i ]['text'] ), self::BODY_DENY, true ) ) {
				self::reject( 'forbidden keyword ' . strtoupper( $t[ $i ]['text'] ) );
			}
			if ( $i > 3 ) {
				$clauses[ count( $clauses ) - 1 ][] = $i;
			}
		}
		if ( $end < 0 ) {
			self::reject( 'unbalanced column list' );
		}

		$edits    = [];
		$stripped = [];
		$server   = self::server();
		$strip    = [];
		foreach ( $clauses as $k => $clause ) {
			if ( ! $clause ) {
				self::reject( 'empty column definition' );
			}
			$fk = self::foreign_key_name( $t, $clause );
			if ( null !== $fk ) {
				$strip[ $k ] = true;
				$stripped[]  = "foreign key $fk on $table removed";
				continue;
			}
			foreach ( $clause as $i ) {
				if ( self::is_word( $t[ $i ], 'REFERENCES' ) ) {
					self::reject( 'inline REFERENCES' );
				}
			}
			self::normalise_clause( $t, $clause, $server, $edits );
		}
		if ( $strip ) {
			$kept = array_diff( array_keys( $clauses ), array_keys( $strip ) );
			if ( ! $kept ) {
				self::reject( 'no columns' );
			}
			$first = min( $kept );
			foreach ( array_keys( $strip ) as $k ) {
				$last_tok = $t[ end( $clauses[ $k ] ) ];
				$clause_end = $last_tok['pos'] + strlen( $last_tok['text'] );
				if ( $k < $first ) {
					// Before the first kept clause: remove the clause and the comma after it.
					$from = $t[ $clauses[ $k ][0] ]['pos'];
					$to   = $t[ $seps[ $k + 1 ] ]['pos'] + 1;
				} else {
					// After it: remove the comma before the clause and the clause.
					$from = $t[ $seps[ $k ] ]['pos'];
					$to   = $clause_end;
				}
				$edits[] = [ $from, $to - $from, '' ];
			}
		}

		// Table options: an allow-list of `[DEFAULT] KEY [=] value`, optionally comma-separated.
		for ( $i = $end + 1; $i < $n; $i++ ) {
			if ( self::is_punct( $t[ $i ], ',' ) ) {
				continue;
			}
			$first   = $t[ $i ]['pos'];
			$default = self::is_word( $t[ $i ], 'DEFAULT' );
			$i      += $default ? 1 : 0;
			if ( $i >= $n || 'word' !== $t[ $i ]['type'] ) {
				self::reject( 'unsupported table option' );
			}
			$key = strtoupper( $t[ $i ]['text'] );
			if ( 'CHARACTER' === $key ) {
				$i++;
				if ( $i >= $n || ! self::is_word( $t[ $i ], 'SET' ) ) {
					self::reject( 'unsupported table option' );
				}
				$key = 'CHARSET';
			}
			if ( ! in_array( $key, self::TABLE_OPTIONS, true ) || ( $default && 'CHARSET' !== $key && 'COLLATE' !== $key ) ) {
				self::reject( 'unsupported table option ' . $key );
			}
			$i++;
			if ( $i < $n && self::is_punct( $t[ $i ], '=' ) ) {
				$i++;
			}
			if ( $i >= $n || ! in_array( $t[ $i ]['type'], [ 'word', 'string' ], true ) ) {
				self::reject( 'missing value for table option ' . $key );
			}
			if ( 'ENGINE' === $key ) {
				$engine = strtoupper( $t[ $i ]['value'] );
				if ( in_array( $engine, self::DENIED_ENGINES, true ) ) {
					self::reject( 'forbidden engine ' . $engine );
				}
				if ( ! isset( $server['engines'][ $engine ] ) ) {
					$edits[] = [ $t[ $i ]['pos'], strlen( $t[ $i ]['text'] ), self::FALLBACK_ENGINE ];
				}
			}
			if ( ! $server['mariadb'] && in_array( $key, self::MARIADB_OPTIONS, true ) ) {
				$edits[] = [ $first, $t[ $i ]['pos'] + strlen( $t[ $i ]['text'] ) - $first, '' ];
			}
		}
		if ( $end + 1 < $n ) {
			self::normalise_clause( $t, range( $end + 1, $n - 1 ), $server, $edits );
		}

		return [ 'table' => $table, 'sql' => self::apply( $create_sql, $edits ), 'stripped' => $stripped ];
	}

	/** Changes the table name of a validated CREATE TABLE text, leaving everything else byte-identical. */
	public static function rename( string $sql, string $from_table, string $to_table ): string {
		$t = self::tokenize( $sql );
		if ( count( $t ) < 3 || ! self::is_word( $t[0], 'CREATE' ) || ! self::is_word( $t[1], 'TABLE' ) || self::name_of( $t[2] ) !== $from_table ) {
			throw new Mench_Migrate_Db_Exception( 'schema_rejected: the schema is not for ' . $from_table, self::STEP );
		}
		self::check_name( $to_table );
		return self::apply( $sql, [ [ $t[2]['pos'], strlen( $t[2]['text'] ), '`' . $to_table . '`' ] ] );
	}

	/**
	 * Splits SQL into word, string, ident and punct tokens with byte offsets; refuses comments, ';' and stray bytes.
	 *
	 * @return array<int,array{type:string,text:string,pos:int,value:string}>
	 */
	private static function tokenize( string $sql ): array {
		$tokens = [];
		$len    = strlen( $sql );
		for ( $i = 0; $i < $len; ) {
			$c = $sql[ $i ];
			if ( ' ' === $c || "\t" === $c || "\n" === $c || "\r" === $c ) {
				$i++;
				continue;
			}
			$next = $i + 1 < $len ? $sql[ $i + 1 ] : '';
			if ( ';' === $c ) {
				self::reject( 'statement separator' );
			}
			if ( '#' === $c || ( '/' === $c && '*' === $next ) || ( '-' === $c && '-' === $next ) ) {
				self::reject( 'comment' );
			}
			if ( "'" === $c || '"' === $c || '`' === $c ) {
				$j     = $i + 1;
				$value = '';
				while ( true ) {
					if ( $j >= $len ) {
						self::reject( 'unterminated quote' );
					}
					$d = $sql[ $j ];
					if ( '\\' === $d && '`' !== $c ) {
						$value .= $sql[ $j + 1 ] ?? '';
						$j     += 2;
						continue;
					}
					if ( $d === $c ) {
						if ( $j + 1 < $len && $sql[ $j + 1 ] === $c ) {
							$value .= $c;
							$j     += 2;
							continue;
						}
						break;
					}
					$value .= $d;
					$j++;
				}
				$tokens[] = [ 'type' => '`' === $c ? 'ident' : 'string', 'text' => substr( $sql, $i, $j + 1 - $i ), 'pos' => $i, 'value' => $value ];
				$i        = $j + 1;
				continue;
			}
			if ( preg_match( '/\G[A-Za-z0-9_$]+/', $sql, $m, 0, $i ) ) {
				$tokens[] = [ 'type' => 'word', 'text' => $m[0], 'pos' => $i, 'value' => $m[0] ];
				$i       += strlen( $m[0] );
				continue;
			}
			if ( false !== strpos( self::PUNCT, $c ) ) {
				$tokens[] = [ 'type' => 'punct', 'text' => $c, 'pos' => $i, 'value' => $c ];
				$i++;
				continue;
			}
			self::reject( 'unexpected character' );
		}
		return $tokens;
	}

	/**
	 * Normalises the CHARACTER SET / CHARSET / COLLATE values in one clause (a column, or the table options).
	 *
	 * @param int[]                  $idx   Token indexes of the clause.
	 * @param array<int,array{0:int,1:int,2:string}> $edits Collected byte edits.
	 */
	private static function normalise_clause( array $t, array $idx, array $server, array &$edits ): void {
		$charsets   = []; // Value token indexes.
		$collations = []; // [ keyword token index, value token index ].
		$count      = count( $idx );
		for ( $k = 0; $k < $count; $k++ ) {
			$i = $idx[ $k ];
			if ( 'word' !== $t[ $i ]['type'] ) {
				continue;
			}
			$w = strtoupper( $t[ $i ]['text'] );
			$v = null;
			if ( 'CHARSET' === $w || ( 'CHARACTER' === $w && isset( $idx[ $k + 1 ] ) && self::is_word( $t[ $idx[ $k + 1 ] ], 'SET' ) ) ) {
				$k += 'CHARACTER' === $w ? 1 : 0;
				$v  = self::value_after( $t, $idx, $k );
				if ( null !== $v ) {
					$charsets[] = $v;
				}
			} elseif ( 'COLLATE' === $w ) {
				$v = self::value_after( $t, $idx, $k );
				if ( null !== $v ) {
					$collations[] = [ $i, $v ];
				}
			}
		}

		$effective = null;
		foreach ( $charsets as $v ) {
			$cs = self::canonical_charset( strtolower( $t[ $v ]['value'] ), $server );
			if ( null === $cs ) {
				$edits[] = [ $t[ $v ]['pos'], strlen( $t[ $v ]['text'] ), self::FALLBACK_CHARSET ];
				$cs      = self::FALLBACK_CHARSET;
			}
			$effective = $cs;
		}
		foreach ( $collations as [ $kw, $v ] ) {
			$co    = self::canonical_collation( strtolower( $t[ $v ]['value'] ), $server );
			$co_cs = null === $co ? null : $server['collations'][ $co ];
			if ( null !== $co && ( null === $effective || $co_cs === $effective ) ) {
				continue;
			}
			$edits[] = [ $t[ $v ]['pos'], strlen( $t[ $v ]['text'] ), self::FALLBACK_COLLATION ];
			if ( null === $effective ) {
				// No charset in this clause: pin one, so the fallback collation never meets an inherited latin1.
				$edits[] = [ $t[ $kw ]['pos'], 0, 'CHARACTER SET ' . self::FALLBACK_CHARSET . ' ' ];
			} elseif ( self::FALLBACK_CHARSET !== $effective ) {
				foreach ( $charsets as $c ) {
					$edits[] = [ $t[ $c ]['pos'], strlen( $t[ $c ]['text'] ), self::FALLBACK_CHARSET ];
				}
				$effective = self::FALLBACK_CHARSET;
			}
		}
	}

	/** Index of the value token after a keyword at position $k of $idx (skipping one '='), advancing $k past it. */
	private static function value_after( array $t, array $idx, int &$k ): ?int {
		$j = $k + 1;
		if ( isset( $idx[ $j ] ) && self::is_punct( $t[ $idx[ $j ] ], '=' ) ) {
			$j++;
		}
		if ( ! isset( $idx[ $j ] ) || ! in_array( $t[ $idx[ $j ] ]['type'], [ 'word', 'string' ], true ) ) {
			return null;
		}
		$k = $j;
		return $idx[ $j ];
	}

	private static function canonical_charset( string $cs, array $server ): ?string {
		if ( isset( $server['charsets'][ $cs ] ) ) {
			return $cs;
		}
		if ( 'utf8' === $cs && isset( $server['charsets']['utf8mb3'] ) ) {
			return 'utf8mb3';
		}
		return null;
	}

	private static function canonical_collation( string $co, array $server ): ?string {
		if ( isset( $server['collations'][ $co ] ) ) {
			return $co;
		}
		if ( 0 === strpos( $co, 'utf8_' ) && isset( $server['collations'][ 'utf8mb3_' . substr( $co, 5 ) ] ) ) {
			return 'utf8mb3_' . substr( $co, 5 );
		}
		return null;
	}

	/** @return array{charsets:array<string,true>,collations:array<string,string>,engines:array<string,true>,mariadb:bool} */
	private static function server(): array {
		if ( null === self::$server ) {
			global $wpdb;
			$s = [ 'charsets' => [], 'collations' => [], 'engines' => [], 'mariadb' => false !== stripos( (string) $wpdb->get_var( 'SELECT VERSION()' ), 'mariadb' ) ];
			foreach ( (array) $wpdb->get_results( 'SHOW CHARACTER SET', ARRAY_A ) as $r ) {
				$s['charsets'][ strtolower( (string) $r['Charset'] ) ] = true;
			}
			foreach ( (array) $wpdb->get_results( 'SHOW COLLATION', ARRAY_A ) as $r ) {
				$s['collations'][ strtolower( (string) $r['Collation'] ) ] = strtolower( (string) $r['Charset'] );
			}
			foreach ( (array) $wpdb->get_results( 'SHOW ENGINES', ARRAY_A ) as $r ) {
				if ( in_array( strtoupper( (string) $r['Support'] ), [ 'YES', 'DEFAULT' ], true ) ) {
					$s['engines'][ strtoupper( (string) $r['Engine'] ) ] = true;
				}
			}
			if ( ! $s['collations'] || ! $s['engines'] ) {
				throw new Mench_Migrate_Db_Exception( 'Could not read the server charsets, collations and engines.', self::STEP );
			}
			self::$server = $s;
		}
		return self::$server;
	}

	/** @param array<int,array{0:int,1:int,2:string}> $edits [ offset, length, replacement ], non-overlapping. */
	private static function apply( string $sql, array $edits ): string {
		usort( $edits, static fn( $a, $b ) => $b[0] <=> $a[0] ?: $b[1] <=> $a[1] );
		foreach ( $edits as [ $pos, $len, $text ] ) {
			$sql = substr_replace( $sql, $text, $pos, $len );
		}
		return $sql;
	}

	private static function name_of( array $token ): string {
		if ( ! in_array( $token['type'], [ 'ident', 'word' ], true ) ) {
			self::reject( 'expected a table name' );
		}
		self::check_name( $token['value'] );
		return $token['value'];
	}

	private static function check_name( string $name ): void {
		if ( ! self::is_valid_name( $name ) ) {
			self::reject( 'invalid table name' );
		}
	}

	/** Table names this plugin will handle: 1-64 of [A-Za-z0-9_$-]. */
	public static function is_valid_name( string $name ): bool {
		return '' !== $name && strlen( $name ) <= 64 && 1 === preg_match( '/^[A-Za-z0-9_$\-]+$/D', $name );
	}

	/**
	 * Name of the foreign key a body clause declares (`[CONSTRAINT [name]] FOREIGN KEY …`), 'unnamed' when it has none,
	 * or null when the clause is something else (a column, an index, a CHECK constraint).
	 *
	 * @param int[] $clause Token indexes.
	 */
	private static function foreign_key_name( array $t, array $clause ): ?string {
		$j    = 0;
		$name = 'unnamed';
		if ( self::is_word( $t[ $clause[0] ], 'CONSTRAINT' ) ) {
			$j = 1;
			if ( isset( $clause[1] ) && ( 'ident' === $t[ $clause[1] ]['type'] || ( 'word' === $t[ $clause[1] ]['type'] && ! in_array( strtoupper( $t[ $clause[1] ]['text'] ), [ 'FOREIGN', 'CHECK', 'PRIMARY', 'UNIQUE' ], true ) ) ) ) {
				$name = $t[ $clause[1] ]['value'];
				$j    = 2;
			}
		}
		if ( isset( $clause[ $j ] ) && self::is_word( $t[ $clause[ $j ] ], 'FOREIGN' ) ) {
			return $name;
		}
		return null;
	}

	private static function is_word( array $token, string $word ): bool {
		return 'word' === $token['type'] && 0 === strcasecmp( $token['text'], $word );
	}

	private static function is_punct( array $token, string $c ): bool {
		return 'punct' === $token['type'] && $c === $token['text'];
	}

	private static function reject( string $why ): never {
		throw new Mench_Migrate_Db_Exception( 'schema_rejected: ' . $why, self::STEP );
	}
}