From d502c26f352f19201f063777c3872559a9850041 Mon Sep 17 00:00:00 2001 From: Alex Standiford Date: Sun, 20 Sep 2026 19:31:53 -0400 Subject: [PATCH] Add optional coordinated core wpdb strategy --- .github/workflows/phpunit.yml | 46 +- lib/Database/ClauseBuilder.php | 112 +++- lib/Database/QueryBuilder.php | 35 +- lib/Strategies/CoordinatedQueryStrategy.php | 557 ++++++++++++++++++ lib/Strategies/PinnedQueryStrategy.php | 258 ++++++++ lib/Strategies/WordPressInitializer.php | 9 +- .../RealWpdbCoordinationContractTest.php | 234 ++++++++ .../Database/RealWpdbQueryContractTest.php | 74 +++ 8 files changed, 1294 insertions(+), 31 deletions(-) create mode 100644 lib/Strategies/CoordinatedQueryStrategy.php create mode 100644 lib/Strategies/PinnedQueryStrategy.php create mode 100644 tests/Integration/Database/RealWpdbCoordinationContractTest.php diff --git a/.github/workflows/phpunit.yml b/.github/workflows/phpunit.yml index 38d772b..ac02604 100644 --- a/.github/workflows/phpunit.yml +++ b/.github/workflows/phpunit.yml @@ -18,4 +18,48 @@ jobs: with: php_version: "8.3" version: "9.6" - configuration: phpunit.xml \ No newline at end of file + configuration: phpunit.xml + + integration: + runs-on: ubuntu-latest + + services: + mysql: + image: mysql:8.0 + env: + MYSQL_ROOT_PASSWORD: root + MYSQL_DATABASE: nomad_coordination_fixture + ports: + - 3306:3306 + options: >- + --health-cmd="mysqladmin ping -h 127.0.0.1 -uroot -proot" + --health-interval=10s + --health-timeout=5s + --health-retries=12 + + steps: + - uses: actions/checkout@v4 + + - uses: shivammathur/setup-php@v2 + with: + php-version: "8.3" + extensions: mysqli + coverage: none + + - name: Install dependencies + run: composer install --no-interaction --no-progress --prefer-dist + + - name: Install official WordPress 6.8.3 + run: | + curl --fail --silent --show-error --location https://wordpress.org/wordpress-6.8.3.tar.gz \ + | tar -xz -C "$RUNNER_TEMP" + + - name: PHPUnit integration tests + env: + WORDPRESS_ROOT: ${{ runner.temp }}/wordpress + MYSQL_HOST: 127.0.0.1 + MYSQL_PORT: 3306 + MYSQL_USER: root + MYSQL_PASSWORD: root + MYSQL_DATABASE: nomad_coordination_fixture + run: vendor/bin/phpunit -c phpunit-integration.xml --colors=never --fail-on-skipped diff --git a/lib/Database/ClauseBuilder.php b/lib/Database/ClauseBuilder.php index 6286ca1..2bc61c3 100644 --- a/lib/Database/ClauseBuilder.php +++ b/lib/Database/ClauseBuilder.php @@ -27,6 +27,38 @@ public function __construct(?wpdb $database = null) $this->database = $database; } + /** + * Return an operation-local copy that prepares through the supplied wpdb. + * + * The ordinary builder remains global-state compatible. Coordinated + * operations use this seam so a caller's mutable clause is not rebound in + * place while another request is using it. + */ + public function forDatabase(wpdb $database): static + { + $clone = clone $this; + $clone->database = $database; + + foreach ($clone->clauses as $index => $clause) { + if (!is_array($clause) || ($clause['type'] ?? null) !== 'group') { + continue; + } + + $bound = []; + foreach ($clause['clauses'] as $groupClause) { + if (!$groupClause instanceof self) { + throw new \PHPNomad\Database\Exceptions\UnsupportedCoordinationException( + 'Coordinated WordPress queries require official WordPress grouped clause builders.' + ); + } + $bound[] = $groupClause->forDatabase($database); + } + $clone->clauses[$index]['clauses'] = $bound; + } + + return $clone; + } + /** @inheritDoc */ public function where($field, string $operator, ...$values) { @@ -64,7 +96,7 @@ public function andGroup(string $logic, ClauseBuilderInterface ...$clauses) $this->clauses[] = 'AND'; } - $this->clauses[] = ['logic' => $logic, 'clauses' => $clauses]; + $this->clauses[] = ['type' => 'group', 'logic' => $logic, 'clauses' => $clauses]; return $this; } @@ -77,7 +109,7 @@ public function orGroup(string $logic, ClauseBuilderInterface ...$clauses) $this->clauses[] = 'OR'; } - $this->clauses[] = ['logic' => $logic, 'clauses' => $clauses]; + $this->clauses[] = ['type' => 'group', 'logic' => $logic, 'clauses' => $clauses]; return $this; } @@ -140,34 +172,18 @@ protected function addCondition($field, string $operator, array $values, ?string $fieldString = $this->getFieldString($field); $values = $this->normalizeValues($field, $operator, $values); - $placeholder = $this->generatePlaceholder($field, $values, $operator); - $condition = "{$fieldString} {$operator}" . ($placeholder === '' ? '' : " {$placeholder}"); - - $preparedValues = []; - foreach ($values as $value) { - if (is_array($value)) { - foreach ($value as $tupleValue) { - if ($tupleValue !== null) { - $preparedValues[] = $tupleValue; - } - } - } elseif ($value !== null) { - $preparedValues[] = $value; - } - } - - if ($preparedValues !== []) { - $condition = $this->wpdb()->prepare($condition, ...$preparedValues); - if (!is_string($condition) || $condition === '') { - throw new QueryBuilderException('WordPress could not prepare a condition.'); - } - } if ($this->clauses !== [] && $logic !== null) { $this->clauses[] = $logic; } - $this->clauses[] = $condition; + $this->clauses[] = [ + 'type' => 'condition', + 'field' => $fieldString, + 'placeholderField' => $field, + 'operator' => $operator, + 'values' => $values, + ]; return $this; } @@ -182,6 +198,11 @@ public function build(): string continue; } + if (($clause['type'] ?? null) === 'condition') { + $queryParts[] = $this->buildCondition($clause); + continue; + } + $groupParts = []; foreach ($clause['clauses'] as $groupClause) { $builtClause = $groupClause->build(); @@ -204,6 +225,16 @@ public function build(): string /** @inheritDoc */ public function reset() { + foreach ($this->clauses as $clause) { + if (!is_array($clause) || ($clause['type'] ?? null) !== 'group') { + continue; + } + foreach ($clause['clauses'] as $groupClause) { + if ($groupClause instanceof ClauseBuilderInterface) { + $groupClause->reset(); + } + } + } $this->clauses = []; $this->preparedValues = []; return $this; @@ -301,6 +332,7 @@ private function normalizeValues($field, string $operator, array $values): array private function appendGroup(string $logic, array $clauses): void { $this->clauses[] = [ + 'type' => 'group', 'logic' => $this->validateGroup($logic, $clauses), 'clauses' => $clauses, ]; @@ -332,4 +364,34 @@ private function wpdb() return $wpdb; } + + /** @param array{field:string, placeholderField:string|string[], operator:string, values:array} $clause */ + private function buildCondition(array $clause): string + { + $values = $clause['values']; + $preparedValues = []; + foreach ($values as $value) { + if (is_array($value)) { + foreach ($value as $tupleValue) { + if ($tupleValue !== null) { + $preparedValues[] = $tupleValue; + } + } + } elseif ($value !== null) { + $preparedValues[] = $value; + } + } + + $placeholder = $this->generatePlaceholder($clause['placeholderField'], $values, $clause['operator']); + $condition = $clause['field'] . ' ' . $clause['operator'] + . ($placeholder === '' ? '' : ' ' . $placeholder); + if ($preparedValues !== []) { + $condition = $this->wpdb()->prepare($condition, ...$preparedValues); + if (!is_string($condition) || $condition === '') { + throw new QueryBuilderException('WordPress could not prepare a condition.'); + } + } + + return $condition; + } } diff --git a/lib/Database/QueryBuilder.php b/lib/Database/QueryBuilder.php index 110370d..022f508 100644 --- a/lib/Database/QueryBuilder.php +++ b/lib/Database/QueryBuilder.php @@ -3,12 +3,13 @@ namespace PHPNomad\Integrations\WordPress\Database; use PHPNomad\Database\Exceptions\QueryBuilderException; -use PHPNomad\Database\Interfaces\ClauseBuilder; +use PHPNomad\Database\Interfaces\ClauseBuilder as ClauseBuilderInterface; use PHPNomad\Database\Interfaces\HasQueryTables; use PHPNomad\Database\Interfaces\QueryBuilder as QueryBuilderInterface; use PHPNomad\Database\Interfaces\Table; use PHPNomad\Database\Traits\WithPrependedFields; use PHPNomad\Integrations\WordPress\Traits\CanGetDataFormats; +use PHPNomad\Integrations\WordPress\Database\ClauseBuilder as WordPressClauseBuilder; use PHPNomad\Utils\Helpers\Arr; use wpdb; @@ -41,7 +42,7 @@ class QueryBuilder implements QueryBuilderInterface, HasQueryTables protected array $orderBy = []; - protected ?ClauseBuilder $clauseBuilder = null; + protected ?ClauseBuilderInterface $clauseBuilder = null; protected array $groupBy = []; private ?Table $rootTable = null; @@ -56,6 +57,31 @@ public function __construct(?wpdb $database = null) $this->database = $database; } + /** + * Return an operation-local copy bound to one wpdb resource. + * + * Query builders are mutable and may be shared by the regular database + * provider. Coordinated calls clone instead of changing that provider's + * builder or the process-global wpdb object. + */ + public function forDatabase(wpdb $database): static + { + $clone = clone $this; + $clone->database = $database; + + if ($clone->clauseBuilder !== null) { + $clauseBuilder = $clone->clauseBuilder; + if (!$clauseBuilder instanceof WordPressClauseBuilder) { + throw new \PHPNomad\Database\Exceptions\UnsupportedCoordinationException( + 'Coordinated WordPress queries require the official WordPress clause builder.' + ); + } + $clone->clauseBuilder = $clauseBuilder->forDatabase($database); + } + + return $clone; + } + /** @inheritDoc */ public function select(string $field, string ...$fields) { @@ -86,7 +112,7 @@ public function from(Table $table) } /** @inheritDoc */ - public function where(?ClauseBuilder $clauseBuilder) + public function where(?ClauseBuilderInterface $clauseBuilder) { $this->clauseBuilder = $clauseBuilder->useTable($this->table); @@ -307,6 +333,9 @@ public function getReferencedTables(): array /** @inheritDoc */ public function reset() { + if ($this->clauseBuilder !== null) { + $this->clauseBuilder->reset(); + } $this->select = []; $this->clauseBuilder = null; $this->from = []; diff --git a/lib/Strategies/CoordinatedQueryStrategy.php b/lib/Strategies/CoordinatedQueryStrategy.php new file mode 100644 index 0000000..cb93e2c --- /dev/null +++ b/lib/Strategies/CoordinatedQueryStrategy.php @@ -0,0 +1,557 @@ +database !== null || $this->connection !== null) { + throw new UnsupportedCoordinationException('A coordinated operation is already active on this strategy.'); + } + $names = []; + $operationStrategy = null; + + try { + $participants = $this->validateInput($coordinationTable, $identity, $participants, $names); + [$database, $connection] = $this->pinCoreWpdb(); + $this->database = $database; + $this->connection = $connection; + $state = $this->sessionState(); + $schema = $state['schema']; + $this->validateGrantProfile($schema, $names); + $coordinationPrimary = $this->primaryFields($schema, $coordinationTable->getName()); + $this->validateIdentity($identity, $coordinationPrimary); + } catch (Throwable $failure) { + $this->clearAttempt(); + $this->reportFailure('validation', $names, 'unchanged', false, $failure); + throw $failure; + } + + try { + $this->raw('START TRANSACTION'); + } catch (Throwable $failure) { + $this->clearAttempt(); + $this->reportFailure('coordination', $names, 'unchanged', false, $failure); + throw $failure; + } + + try { + try { + $this->guard($schema, $coordinationTable->getName(), $coordinationPrimary, $identity); + $coordinationPrimary = $this->primaryFields($schema, $coordinationTable->getName()); + $this->validateIdentity($identity, $coordinationPrimary); + $this->validateParticipants($schema, $participants, $coordinationPrimary); + } catch (Throwable $failure) { + $this->abort($names, 'coordination', $failure); + } + + try { + $delegate = new PinnedQueryStrategy( + $this->database, + $this->connection, + $this->connectionId, + function (): void { + $this->assertPinned(); + }, + function (string $table) use ($schema): array { + return $this->primaryFields($schema, $table); + } + ); + $operationStrategy = new OperationQueryStrategy($delegate, $participants); + $result = $operation($operationStrategy); + } catch (Throwable $failure) { + $this->abort($names, 'callback', $failure); + } + + $this->assertPinned(); + if (!$this->inTransaction()) { + $failure = new CoordinatedOperationOutcomeUnknownException( + 'The coordinated WordPress operation outcome is unknown.', + 0, + new DatastoreErrorException('The transaction ended before commit.') + ); + $this->reportFailure('commit', $names, 'unknown', false, $failure); + throw $failure; + } + + try { + $this->raw('COMMIT'); + } catch (Throwable $failure) { + $this->handleCommitFailure($names, $failure); + } + + return $result; + } finally { + if ($operationStrategy instanceof OperationQueryStrategy) { + $operationStrategy->close(); + } + $this->clearAttempt(); + } + } + + private function clearAttempt(): void + { + $this->database = null; + $this->connection = null; + $this->connectionId = 0; + $this->transactionActive = false; + } + + /** @param list $names */ + /** @return non-empty-list */ + private function validateInput(Table $coordinationTable, array $identity, array $participants, array &$names): array + { + if ($participants === [] || !$this->isList($participants)) { + throw new \InvalidArgumentException('Participants must be a nonempty list of tables.'); + } + foreach ($participants as $participant) { + if (!$participant instanceof Table) { + throw new \InvalidArgumentException('Every participant must be a table descriptor.'); + } + $name = $participant->getName(); + $names[] = $name; + $this->identifier($name); + } + $coordinationName = $coordinationTable->getName(); + $this->identifier($coordinationName); + if (!in_array($coordinationName, $names, true)) { + throw new \InvalidArgumentException('The coordination table must be a participant.'); + } + if ($identity === []) { + throw new \InvalidArgumentException('The coordination identity must be nonempty.'); + } + foreach ($identity as $field => $value) { + if (!is_string($field)) { + throw new \InvalidArgumentException('Coordination identity fields must be strings.'); + } + $this->identifier($field); + if (!is_int($value) && !is_string($value)) { + throw new \InvalidArgumentException('Coordination identity values must be integers or strings.'); + } + } + + $ordered = []; + foreach ($participants as $participant) { + if ($participant->getName() === $coordinationName) { + $ordered[] = $participant; + break; + } + } + foreach ($participants as $participant) { + if ($participant->getName() !== $coordinationName) { + $ordered[] = $participant; + } + } + + return $ordered; + } + + /** @return array{0: wpdb, 1: mysqli} */ + private function pinCoreWpdb(): array + { + global $wpdb; + if (!isset($wpdb) || !($wpdb instanceof wpdb) || get_class($wpdb) !== wpdb::class) { + throw new UnsupportedCoordinationException('Coordination supports only the official core wpdb class.'); + } + $connection = $wpdb->dbh; + if (!$connection instanceof mysqli || !$wpdb->ready) { + throw new UnsupportedCoordinationException('Coordination requires a ready core wpdb mysqli session.'); + } + $this->connectionId = mysqli_thread_id($connection); + if ($this->connectionId <= 0) { + throw new UnsupportedCoordinationException('The core wpdb session identity could not be established.'); + } + + return [$wpdb, $connection]; + } + + /** @return array{schema: string, isolation: string} */ + private function sessionState(): array + { + try { + $rows = $this->rows( + 'SELECT VERSION() AS server_version, @@autocommit AS autocommit, ' + . '@@session.transaction_isolation AS isolation, DATABASE() AS schema_name, ' + . 'CURRENT_ROLE() AS current_role, @@global.partial_revokes AS partial_revokes' + ); + } catch (Throwable $failure) { + throw new UnsupportedCoordinationException('The MySQL 8 coordination session profile could not be established.', 0, $failure); + } + $state = $rows[0] ?? []; + if (!is_string($state['server_version'] ?? null) || version_compare($state['server_version'], '8.0.0', '<')) { + throw new UnsupportedCoordinationException('Coordination requires MySQL 8.0 or newer.'); + } + if ((string) ($state['autocommit'] ?? '') !== '1' || $this->hasAmbientTransaction()) { + throw new UnsupportedCoordinationException('Coordination cannot join an ambient or disabled-autocommit transaction.'); + } + $isolation = strtoupper((string) ($state['isolation'] ?? '')); + if (!in_array($isolation, ['READ-COMMITTED', 'REPEATABLE-READ'], true)) { + throw new UnsupportedCoordinationException('The selected transaction isolation is unsupported.'); + } + if (($state['current_role'] ?? null) !== 'NONE') { + throw new UnsupportedCoordinationException('Active MySQL roles are unsupported for coordination.'); + } + if (!in_array(strtoupper((string) ($state['partial_revokes'] ?? '')), ['OFF', '0'], true)) { + throw new UnsupportedCoordinationException('MySQL partial privilege revokes are unsupported for coordination.'); + } + $schema = (string) ($state['schema_name'] ?? ''); + if ($schema === '') { + throw new UnsupportedCoordinationException('Coordination requires a selected database.'); + } + + return ['schema' => $schema, 'isolation' => $isolation]; + } + + /** @param list $tables */ + private function validateGrantProfile(string $schema, array $tables): void + { + $rows = $this->rows('SHOW GRANTS FOR CURRENT_USER()'); + $grants = array_map(static fn (array $row): string => (string) array_values($row)[0], $rows); + foreach ($tables as $table) { + $visible = false; + foreach ($grants as $grant) { + if ($this->grantAllowsTriggers($grant, $schema, $table)) { + $visible = true; + break; + } + } + if (!$visible) { + throw new UnsupportedCoordinationException('Direct trigger visibility is required for every participant.'); + } + } + } + + private function grantAllowsTriggers(string $grant, string $schema, string $table): bool + { + if (stripos($grant, 'REVOKE ') === 0 || !preg_match('/^GRANT (.+) ON (.+) TO /i', $grant, $match)) { + return false; + } + $privileges = array_map('trim', explode(',', strtoupper($match[1]))); + if (!in_array('TRIGGER', $privileges, true) && !in_array('ALL PRIVILEGES', $privileges, true)) { + return false; + } + $resource = trim($match[2]); + if ($resource === '*.*') { + return true; + } + if (preg_match('/^`((?:``|[^`])*)`\.\*$/D', $resource, $scope)) { + return str_replace('``', '`', $scope[1]) === $schema; + } + if (preg_match('/^`((?:``|[^`])*)`\.`((?:``|[^`])*)`$/D', $resource, $scope)) { + return str_replace('``', '`', $scope[1]) === $schema + && str_replace('``', '`', $scope[2]) === $table; + } + + return false; + } + + /** @return list */ + private function primaryFields(string $schema, string $table): array + { + $rows = $this->rows( + 'SELECT COLUMN_NAME FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = ' + . $this->literal($schema) . ' AND TABLE_NAME = ' . $this->literal($table) + . " AND INDEX_NAME = 'PRIMARY' ORDER BY SEQ_IN_INDEX" + ); + + return array_values(array_filter(array_map(static fn (array $row): mixed => $row['COLUMN_NAME'] ?? null, $rows), 'is_string')); + } + + /** @param list $primary */ + private function validateIdentity(array $identity, array $primary): void + { + if ($primary === []) { + throw new UnsupportedCoordinationException('The coordination table must have a primary key.'); + } + $fields = array_keys($identity); + if (count($fields) !== count($primary) || array_diff($fields, $primary) !== [] || array_diff($primary, $fields) !== []) { + throw new \InvalidArgumentException('The coordination identity must contain every primary field exactly once.'); + } + } + + /** @param list $primary */ + private function guard(string $schema, string $table, array $primary, array $identity): void + { + $conditions = []; + foreach ($primary as $field) { + $conditions[] = $this->identifier($field) . ' = ' . $this->literal($identity[$field]); + } + $rows = $this->rows( + 'SELECT 1 FROM ' . $this->identifier($table) . ' WHERE ' . implode(' AND ', $conditions) . ' FOR UPDATE' + ); + if ($rows === []) { + throw new RecordNotFoundException('The coordination record does not exist.'); + } + } + + /** @param list
$participants @param list $coordinationPrimary */ + private function validateParticipants(string $schema, array $participants, array $coordinationPrimary): void + { + foreach ($participants as $index => $participant) { + $table = $participant->getName(); + $this->raw('SELECT 1 FROM ' . $this->identifier($table) . ' WHERE 1 = 0 FOR UPDATE'); + $createRows = $this->rows('SHOW CREATE TABLE ' . $this->identifier($table)); + $definition = $createRows[0] ?? []; + if (isset($definition['Create View']) || preg_match('/^CREATE\s+TEMPORARY\s+TABLE\b/i', (string) ($definition['Create Table'] ?? ''))) { + throw new UnsupportedCoordinationException('Temporary tables and views are unsupported participants.'); + } + + $metadata = $this->rows( + 'SELECT TABLE_TYPE, ENGINE FROM information_schema.TABLES WHERE TABLE_SCHEMA = ' + . $this->literal($schema) . ' AND TABLE_NAME = ' . $this->literal($table) + )[0] ?? []; + if (($metadata['TABLE_TYPE'] ?? null) !== 'BASE TABLE' || strtoupper((string) ($metadata['ENGINE'] ?? '')) !== 'INNODB') { + throw new UnsupportedCoordinationException('Every participant must be an InnoDB base table.'); + } + + $primary = $this->primaryFields($schema, $table); + if ($primary === []) { + throw new UnsupportedCoordinationException('Every participant must have a primary key.'); + } + if ($table === $participants[0]->getName() && $primary !== $coordinationPrimary) { + throw new \InvalidArgumentException('The coordination identity must match the stable primary key.'); + } + + $triggers = $this->rows( + 'SELECT TRIGGER_NAME FROM information_schema.TRIGGERS WHERE TRIGGER_SCHEMA = ' + . $this->literal($schema) . ' AND EVENT_OBJECT_TABLE = ' . $this->literal($table) + ); + if ($triggers !== []) { + throw new UnsupportedCoordinationException('Trigger-bearing tables are unsupported participants.'); + } + } + } + + /** + * @param list $tables + * @return never + */ + private function abort(array $tables, string $phase, Throwable $failure): void + { + try { + $this->assertPinned(); + if (!$this->inTransaction() || !$this->raw('ROLLBACK')) { + throw new DatastoreErrorException('The coordinated rollback was not acknowledged.'); + } + } catch (Throwable $cleanup) { + $composite = new CoordinatedOperationCleanupFailedException($failure, $cleanup); + $this->reportFailure('rollback', $tables, 'unknown', false, $composite, $cleanup); + throw $composite; + } + + $operationFailure = $this->isContention($failure) + ? new CoordinatedOperationConflictException('The coordinated WordPress operation conflicted.', 0, $failure) + : $failure; + $this->reportFailure($phase, $tables, 'rolled_back', $operationFailure instanceof CoordinatedOperationConflictException, $operationFailure, $failure); + throw $operationFailure; + } + + /** + * @param list $tables + * @return never + */ + private function handleCommitFailure(array $tables, Throwable $failure): void + { + try { + if (!$this->inTransaction()) { + $unknown = new CoordinatedOperationOutcomeUnknownException('The coordinated WordPress operation outcome is unknown.', 0, $failure); + $this->reportFailure('commit', $tables, 'unknown', false, $unknown, $failure); + throw $unknown; + } + if (!$this->raw('ROLLBACK')) { + throw new DatastoreErrorException('The coordinated rollback was not acknowledged.'); + } + } catch (CoordinatedOperationOutcomeUnknownException $unknown) { + throw $unknown; + } catch (Throwable $cleanup) { + $composite = new CoordinatedOperationCleanupFailedException($failure, $cleanup); + $this->reportFailure('rollback', $tables, 'unknown', false, $composite, $cleanup); + throw $composite; + } + + $failed = new DatastoreErrorException('The coordinated WordPress commit failed.', 0, $failure); + $this->reportFailure('commit', $tables, 'rolled_back', false, $failed, $failure); + throw $failed; + } + + /** @param list $tables */ + private function reportFailure(string $phase, array $tables, string $outcome, bool $retryable, Throwable $operationFailure, ?Throwable $cause = null): void + { + if ($this->logger === null) { + return; + } + $cause ??= $operationFailure; + try { + $this->logger->error('Coordinated database operation failed.', [ + 'phase' => $phase, + 'tables' => $tables, + 'outcome' => $outcome, + 'retryable' => $retryable, + 'causeClass' => get_class($cause), + 'sqlState' => $this->sqlState($cause), + 'driverCode' => $this->driverCode($cause), + ]); + } catch (Throwable $reportingFailure) { + throw new CoordinatedOperationReportingFailedException($operationFailure, $reportingFailure); + } + } + + private function isContention(Throwable $failure): bool + { + $code = $this->driverCode($failure); + $state = $this->sqlState($failure); + return ($state === '40001' && $code === 1213) || ($state === 'HY000' && $code === 1205) || in_array($code, [1205, 1213], true); + } + + private function sqlState(Throwable $failure): ?string + { + for ($node = $failure; $node !== null; $node = $node->getPrevious()) { + if (property_exists($node, 'sqlstate') && is_string($node->sqlstate)) { + return $node->sqlstate; + } + } + return null; + } + + private function driverCode(Throwable $failure): ?int + { + for ($node = $failure; $node !== null; $node = $node->getPrevious()) { + if ($node->getCode() !== 0) { + return (int) $node->getCode(); + } + } + return null; + } + + private function inTransaction(): bool + { + if (!$this->transactionActive) { + return false; + } + $this->assertPinned(); + $probe = mysqli_query($this->connection, 'SAVEPOINT __nomad_coordination_probe'); + if ($probe === false) { + return false; + } + $rollback = mysqli_query($this->connection, 'ROLLBACK TO SAVEPOINT __nomad_coordination_probe'); + + return $rollback !== false; + } + + private function hasAmbientTransaction(): bool + { + $this->assertPinned(); + $probe = mysqli_query($this->connection, 'SAVEPOINT __nomad_coordination_ambient_probe'); + if ($probe === false) { + return false; + } + $rollback = mysqli_query($this->connection, 'ROLLBACK TO SAVEPOINT __nomad_coordination_ambient_probe'); + + return $rollback !== false; + } + + private function raw(string $sql): bool + { + $this->assertPinned(); + $result = mysqli_query($this->connection, $sql); + $this->assertPinned(); + if ($result === false) { + throw new DatastoreErrorException('WordPress database command failed.', mysqli_errno($this->connection), new \RuntimeException(mysqli_error($this->connection))); + } + if ($result instanceof \mysqli_result) { + mysqli_free_result($result); + } + if (preg_match('/^START\s+TRANSACTION\b/i', $sql)) { + $this->transactionActive = true; + } elseif (preg_match('/^(COMMIT|ROLLBACK)\b/i', $sql)) { + $this->transactionActive = false; + } + return true; + } + + /** @return list> */ + private function rows(string $sql): array + { + $this->assertPinned(); + $result = mysqli_query($this->connection, $sql); + $this->assertPinned(); + if ($result === false) { + throw new DatastoreErrorException('WordPress database query failed.', mysqli_errno($this->connection), new \RuntimeException(mysqli_error($this->connection))); + } + if (!$result instanceof \mysqli_result) { + return []; + } + $rows = []; + while ($row = mysqli_fetch_assoc($result)) { + $rows[] = $row; + } + mysqli_free_result($result); + return $rows; + } + + private function assertPinned(): void + { + global $wpdb; + if ($this->database === null || $this->connection === null || $wpdb !== $this->database || $wpdb->dbh !== $this->connection) { + throw new CoordinatedOperationOutcomeUnknownException('The core wpdb resource changed during coordination.'); + } + if (mysqli_thread_id($this->connection) !== $this->connectionId) { + throw new CoordinatedOperationOutcomeUnknownException('The core wpdb session changed during coordination.'); + } + } + + private function identifier(string $identifier): string + { + if (!preg_match('/^[A-Za-z0-9_$]+$/', $identifier)) { + throw new \InvalidArgumentException('Invalid WordPress database identifier.'); + } + return '`' . $identifier . '`'; + } + + private function literal(string $value): string + { + return "'" . mysqli_real_escape_string($this->connection, $value) . "'"; + } + + private function isList(array $value): bool + { + return $value === [] || array_keys($value) === range(0, count($value) - 1); + } +} diff --git a/lib/Strategies/PinnedQueryStrategy.php b/lib/Strategies/PinnedQueryStrategy.php new file mode 100644 index 0000000..0bd4a28 --- /dev/null +++ b/lib/Strategies/PinnedQueryStrategy.php @@ -0,0 +1,258 @@ +assertResource)(); + + return new WordPressQueryBuilder($this->database); + } + + /** Create a fresh clause builder already bound to this operation's wpdb. */ + public function createClauseBuilder(): WordPressClauseBuilder + { + ($this->assertResource)(); + + return new WordPressClauseBuilder($this->database); + } + + /** @inheritDoc */ + public function query(QueryBuilder $builder): array + { + ($this->assertResource)(); + try { + $queryBuilder = $builder instanceof WordPressQueryBuilder + ? $builder->forDatabase($this->database) + : throw new \PHPNomad\Database\Exceptions\UnsupportedCoordinationException( + 'Coordinated WordPress queries require the official WordPress query builder.' + ); + $sql = $queryBuilder->build(); + } catch (QueryBuilderException $e) { + throw new DatastoreErrorException('Get results failed: invalid query.', 500, $e); + } finally { + // The supplied builder belongs to the caller's mutable provider. + // Consume it exactly once without retaining state for a later + // operation, including when binding/building fails. + $builder->reset(); + } + + $result = $this->execute($sql); + if (!$result instanceof \mysqli_result) { + throw new DatastoreErrorException('Get results failed.'); + } + + $rows = []; + while ($row = mysqli_fetch_assoc($result)) { + $rows[] = $row; + } + mysqli_free_result($result); + + if ($rows === []) { + throw new RecordNotFoundException('No records found for the query.'); + } + + return $rows; + } + + /** @inheritDoc */ + public function insert(Table $table, array $data): array + { + ($this->assertResource)(); + $name = $this->identifier($table->getName()); + + if ($data === []) { + $sql = 'INSERT INTO ' . $name . ' () VALUES ()'; + } else { + $fields = []; + $values = []; + foreach ($data as $field => $value) { + $fields[] = $this->identifier((string) $field); + $values[] = $this->literal($value); + } + $sql = 'INSERT INTO ' . $name . ' (' . implode(', ', $fields) . ') VALUES (' + . implode(', ', $values) . ')'; + } + + $this->execute($sql); + $identity = []; + $primaryFields = ($this->primaryFields)($table->getName()); + foreach ($primaryFields as $field) { + if (array_key_exists($field, $data)) { + $identity[$field] = $data[$field]; + } + } + + if (count($identity) === count($primaryFields)) { + return $identity; + } + + $insertId = mysqli_insert_id($this->connection); + if (count($primaryFields) !== count($identity) + 1 || $insertId === 0) { + throw new DatastoreErrorException('Insert identity could not be established.'); + } + + foreach ($primaryFields as $field) { + if (!array_key_exists($field, $identity)) { + $identity[$field] = $insertId; + break; + } + } + + return $identity; + } + + /** @inheritDoc */ + public function delete(Table $table, array $ids): void + { + ($this->assertResource)(); + $this->execute($this->mutationSql('DELETE FROM', $table, [], $ids)); + } + + /** @inheritDoc */ + public function update(Table $table, array $where, array $data): void + { + if ($data === []) { + throw new DatastoreErrorException('Update failed - no update data provided.'); + } + + ($this->assertResource)(); + $set = []; + foreach ($data as $field => $value) { + $set[] = $this->identifier((string) $field) . ' = ' . $this->literal($value); + } + $this->execute($this->mutationSql('UPDATE', $table, $set, $where)); + if (mysqli_affected_rows($this->connection) === 0 && !$this->exists($table, $where)) { + throw new RecordNotFoundException('Update failed because the record does not exist.'); + } + } + + /** @inheritDoc */ + public function estimatedCount(Table $table): int + { + ($this->assertResource)(); + $result = $this->execute('SELECT COUNT(*) AS `count` FROM ' . $this->identifier($table->getName())); + if (!$result instanceof \mysqli_result) { + throw new DatastoreErrorException('Count query failed.'); + } + $row = mysqli_fetch_assoc($result); + mysqli_free_result($result); + + return (int) ($row['count'] ?? 0); + } + + private function mutationSql(string $verb, Table $table, array $set, array $where): string + { + $name = $this->identifier($table->getName()); + $whereSql = $this->conditions($where); + + if ($verb === 'UPDATE') { + return 'UPDATE ' . $name . ' SET ' . implode(', ', $set) . ' WHERE ' . $whereSql; + } + + return $verb . ' ' . $name . ' WHERE ' . $whereSql; + } + + private function exists(Table $table, array $where): bool + { + $result = $this->execute('SELECT 1 FROM ' . $this->identifier($table->getName()) . ' WHERE ' + . $this->conditions($where) . ' LIMIT 1'); + if (!$result instanceof \mysqli_result) { + return false; + } + $exists = mysqli_fetch_row($result) !== null; + mysqli_free_result($result); + + return $exists; + } + + private function conditions(array $where): string + { + $conditions = []; + foreach ($where as $field => $value) { + if ($value === null) { + $conditions[] = $this->identifier((string) $field) . ' IS NULL'; + continue; + } + $conditions[] = $this->identifier((string) $field) . ' = ' . $this->literal($value); + } + if ($conditions === []) { + throw new \InvalidArgumentException('A coordinated mutation requires a nonempty identity.'); + } + + return implode(' AND ', $conditions); + } + + private function execute(string $sql): \mysqli_result|bool + { + ($this->assertResource)(); + $result = mysqli_query($this->connection, $sql); + ($this->assertResource)(); + if ($result === false) { + throw new DatastoreErrorException( + 'WordPress database query failed.', + mysqli_errno($this->connection), + new \RuntimeException(mysqli_error($this->connection)) + ); + } + + return $result; + } + + private function identifier(string $identifier): string + { + if (!preg_match('/^[A-Za-z0-9_$]+$/', $identifier)) { + throw new \InvalidArgumentException('Invalid WordPress database identifier.'); + } + + return '`' . $identifier . '`'; + } + + private function literal(mixed $value): string + { + if ($value === null) { + return 'NULL'; + } + if (is_int($value) || is_float($value)) { + return (string) $value; + } + if (!is_string($value)) { + throw new \InvalidArgumentException('WordPress mutation values must be scalar.'); + } + + return "'" . mysqli_real_escape_string($this->connection, $value) . "'"; + } +} diff --git a/lib/Strategies/WordPressInitializer.php b/lib/Strategies/WordPressInitializer.php index 1e26a9e..2eb9e56 100644 --- a/lib/Strategies/WordPressInitializer.php +++ b/lib/Strategies/WordPressInitializer.php @@ -26,6 +26,7 @@ use PHPNomad\Database\Interfaces\CanConvertDatabaseStringToDateTime; use PHPNomad\Database\Interfaces\CanConvertToDatabaseDateString; use PHPNomad\Database\Interfaces\ClauseBuilder as CoreClauseBuilder; +use PHPNomad\Database\Interfaces\CoordinatedQueryStrategy as CoreCoordinatedQueryStrategy; use PHPNomad\Database\Interfaces\HasCharsetProvider; use PHPNomad\Database\Interfaces\HasCollateProvider; use PHPNomad\Database\Interfaces\HasGlobalDatabasePrefix; @@ -93,7 +94,11 @@ public function getClassDefinitions(): array ActionBindingStrategy::class => CoreActionBindingStrategy::class, ObjectCacheStrategy::class => CacheStrategy::class, CachePolicy::class => CoreCachePolicy::class, - QueryStrategy::class => CoreQueryStrategy::class, + // Opting into the coordinated capability must not create a second + // strategy. DatabaseServiceProvider and both public aliases share + // one resource-owning instance, while inherited CRUD remains the + // ordinary WordPress implementation. + CoordinatedQueryStrategy::class => [CoreQueryStrategy::class, CoreCoordinatedQueryStrategy::class], DefaultCacheTtlProvider::class => HasDefaultTtl::class, TableCreateStrategy::class => CoreTableCreateStrategyAlias::class, TableUpdateStrategy::class => CoreTableUpdateStrategy::class, @@ -108,7 +113,7 @@ public function getClassDefinitions(): array CurrentContextResolverStrategy::class => CurrentContextResolverStrategyInterface::class, CurrentUserResolverStrategy::class => CurrentUserResolverStrategyInterface::class, PostAuthorResolver::class => PageAuthorResolver::class, - DatabaseProvider::class => [HasDefaultTtl::class, HasGlobalDatabasePrefix::class, HasCollateProvider::class, HasCharsetProvider::class], + DatabaseProvider::class => [HasGlobalDatabasePrefix::class, HasCollateProvider::class, HasCharsetProvider::class], DatabaseDateAdapter::class => [CanConvertToDatabaseDateString::class, CanConvertDatabaseStringToDateTime::class], AssetStrategy::class => AssetStrategyInterface::class, TrackingPermissionStrategy::class => TrackingPermissionStrategyInterface::class, diff --git a/tests/Integration/Database/RealWpdbCoordinationContractTest.php b/tests/Integration/Database/RealWpdbCoordinationContractTest.php new file mode 100644 index 0000000..a12f03f --- /dev/null +++ b/tests/Integration/Database/RealWpdbCoordinationContractTest.php @@ -0,0 +1,234 @@ +suppress_errors(false); + $GLOBALS['wpdb'] = self::$wpdb; + self::$parent = new ContractTable(self::PARENT, 'coordination_parent', [new Column('id', 'INT', null, 'PRIMARY KEY')], ['id']); + self::$child = new ContractTable(self::CHILD, 'coordination_child', [new Column('id', 'INT', null, 'PRIMARY KEY'), new Column('value', 'VARCHAR', [64])], ['id']); + self::query('DROP TABLE IF EXISTS ' . self::CHILD); + self::query('DROP TABLE IF EXISTS ' . self::PARENT); + self::query('CREATE TABLE ' . self::PARENT . ' (id INT PRIMARY KEY) ENGINE=InnoDB'); + self::query('CREATE TABLE ' . self::CHILD . ' (id INT PRIMARY KEY, value VARCHAR(64) NULL) ENGINE=InnoDB'); + } + + protected function setUp(): void + { + self::query('DELETE FROM ' . self::CHILD); + self::query('DELETE FROM ' . self::PARENT); + self::query('INSERT INTO ' . self::PARENT . ' (id) VALUES (1)'); + } + + public static function tearDownAfterClass(): void + { + if (isset(self::$wpdb)) { + self::query('DROP TABLE IF EXISTS ' . self::CHILD); + self::query('DROP TABLE IF EXISTS ' . self::PARENT); + unset($GLOBALS['wpdb']); + } + } + + public function testSuccessCommitsOnceAndSupportsNonFirstCoordinationTable(): void + { + $calls = 0; + $strategy = new WordPressCoordinatedQueryStrategy(); + self::assertInstanceOf(CoordinatedQueryStrategy::class, $strategy); + $strategy->coordinate(self::$parent, ['id' => 1], [self::$child, self::$parent], function ($operation) use (&$calls): void { + $calls++; + $operation->insert(self::$child, ['id' => 10, 'value' => 'committed']); + }); + + self::assertSame(1, $calls); + self::assertSame('committed', self::$wpdb->get_var('SELECT value FROM ' . self::CHILD . ' WHERE id = 10')); + } + + public function testInitializerSharesTheCoordinatorWithTheOrdinaryProvider(): void + { + $container = new Container(); + (new Bootstrapper($container, new WordPressInitializer()))->load(); + $ordinary = $container->get(QueryStrategy::class); + $coordinated = $container->get(CoordinatedQueryStrategy::class); + self::assertSame($ordinary, $coordinated); + } + + public function testCallbackFailureRollsBackAndIsNotRetried(): void + { + $calls = 0; + $strategy = new WordPressCoordinatedQueryStrategy(); + try { + $strategy->coordinate(self::$parent, ['id' => 1], [self::$parent, self::$child], function ($operation) use (&$calls): void { + $calls++; + $operation->insert(self::$child, ['id' => 11, 'value' => 'rolled-back']); + throw new RuntimeException('callback failed'); + }); + self::fail('Expected callback failure.'); + } catch (RuntimeException $failure) { + self::assertSame('callback failed', $failure->getMessage()); + } + self::assertSame(1, $calls); + self::assertFalse((bool) self::$wpdb->get_var('SELECT 1 FROM ' . self::CHILD . ' WHERE id = 11')); + } + + public function testDuplicateWriteRollsBackWithoutCallbackReplay(): void + { + self::query("INSERT INTO " . self::CHILD . " (id, value) VALUES (12, 'existing')"); + $calls = 0; + try { + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 1], [self::$parent, self::$child], function ($operation) use (&$calls): void { + $calls++; + $operation->insert(self::$child, ['id' => 12, 'value' => 'duplicate']); + }); + self::fail('Expected duplicate write failure.'); + } catch (DatastoreErrorException) { + self::assertSame(1, $calls); + } + self::assertSame('existing', self::$wpdb->get_var('SELECT value FROM ' . self::CHILD . ' WHERE id = 12')); + } + + public function testMissingUpdateIdentityIsRecordNotFoundAndRollsBack(): void + { + try { + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 1], [self::$parent, self::$child], function ($operation): void { + $operation->update(self::$child, ['id' => 404], ['value' => 'missing']); + }); + self::fail('Expected a missing update identity.'); + } catch (\PHPNomad\Datastore\Exceptions\RecordNotFoundException) { + self::assertTrue(true); + } + } + + public function testNullMutationPredicateUsesIsNull(): void + { + self::query("INSERT INTO " . self::CHILD . " (id, value) VALUES (13, NULL)"); + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 1], [self::$parent, self::$child], function ($operation): void { + $operation->update(self::$child, ['value' => null], ['value' => 'updated']); + }); + self::assertSame('updated', self::$wpdb->get_var('SELECT value FROM ' . self::CHILD . ' WHERE id = 13')); + } + + public function testMissingCoordinationRecordDoesNotInvokeCallback(): void + { + $calls = 0; + try { + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 999], [self::$parent], function () use (&$calls): void { + $calls++; + }); + self::fail('Expected missing coordination record.'); + } catch (Throwable) { + self::assertSame(0, $calls); + } + } + + public function testGlobalWpdbReplacementIsDetectedBeforePinnedQueryRuns(): void + { + $original = $GLOBALS['wpdb']; + try { + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 1], [self::$parent], function ($operation) use ($original): void { + $GLOBALS['wpdb'] = new class () { + }; + try { + $operation->estimatedCount(self::$parent); + } finally { + $GLOBALS['wpdb'] = $original; + } + }); + self::fail('Expected resource replacement failure.'); + } catch (CoordinatedOperationOutcomeUnknownException|RuntimeException) { + self::assertSame($original, $GLOBALS['wpdb']); + } + } + + public function testCustomWpdbSubclassIsRefusedBeforeCallback(): void + { + $original = $GLOBALS['wpdb']; + $calls = 0; + $GLOBALS['wpdb'] = new class () extends wpdb { + public function __construct() + { + } + }; + try { + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 1], [self::$parent], function () use (&$calls): void { + $calls++; + }); + self::fail('Expected custom wpdb refusal.'); + } catch (Throwable) { + self::assertSame(0, $calls); + } finally { + $GLOBALS['wpdb'] = $original; + } + } + + public function testOperationBuilderIsConsumedAndFreshReadUsesPinnedSession(): void + { + $builder = (new QueryBuilder()) + ->select('*') + ->from(self::$parent) + ->where((new ClauseBuilder())->useTable(self::$parent)->where('id', '=', 1)); + $rows = []; + (new WordPressCoordinatedQueryStrategy())->coordinate(self::$parent, ['id' => 1], [self::$parent], function ($operation) use ($builder, &$rows): void { + $rows = $operation->query($builder); + }); + self::assertCount(1, $rows); + try { + $builder->build(); + self::fail('Expected the operation to consume the builder.'); + } catch (QueryBuilderException) { + self::assertTrue(true); + } + } + + private static function query(string $sql): void + { + if (self::$wpdb->query($sql) === false) { + throw new RuntimeException(self::$wpdb->last_error); + } + } +} diff --git a/tests/Integration/Database/RealWpdbQueryContractTest.php b/tests/Integration/Database/RealWpdbQueryContractTest.php index b4ef123..876952d 100644 --- a/tests/Integration/Database/RealWpdbQueryContractTest.php +++ b/tests/Integration/Database/RealWpdbQueryContractTest.php @@ -5,7 +5,9 @@ namespace PHPNomad\Integrations\WordPress\Tests\Integration\Database; use PHPNomad\Database\Exceptions\QueryBuilderException; +use PHPNomad\Database\Exceptions\UnsupportedCoordinationException; use PHPNomad\Database\Factories\Column; +use PHPNomad\Database\Interfaces\ClauseBuilder as CoreClauseBuilder; use PHPNomad\Database\Interfaces\QueryStrategy as CoreQueryStrategy; use PHPNomad\Datastore\Exceptions\DatastoreErrorException; use PHPNomad\Datastore\Exceptions\RecordNotFoundException; @@ -756,6 +758,78 @@ public function preparedLimit(int $limit): self self::assertCount(4, self::rawSelect($resetSql)); } + public function testClausePreparedForAnotherSessionIsRepreparedOnBoundWpdb(): void + { + $alternate = new class ( + self::environment('MYSQL_USER', 'root'), + self::environment('MYSQL_PASSWORD', ''), + self::environment('MYSQL_DATABASE', 'phpnomad_wordpress_integration_test'), + self::environment('MYSQL_HOST', '127.0.0.1') . ':' . self::environment('MYSQL_PORT', '3306') + ) extends wpdb { + public function prepare($query, ...$args) + { + throw new \RuntimeException('The alternate session must not prepare operation clauses.'); + } + }; + + try { + self::assertNotSame(self::$wpdb->dbh, $alternate->dbh); + $clause = (new ClauseBuilder($alternate)) + ->useTable(self::$predicateTable) + ->where('label', '=', "session-bound O'Reilly"); + + $sql = self::selectQuery($clause)->forDatabase(self::$wpdb)->build(); + + self::assertStringContainsString("predicates.label = 'session-bound O\\'Reilly'", $sql); + + $tupleClause = (new ClauseBuilder($alternate)) + ->useTable(self::$compoundTable) + ->where(['leftId', 'rightId'], 'IN', [1, 10]); + $tupleSql = (new QueryBuilder()) + ->from(self::$compoundTable) + ->select('*') + ->where($tupleClause) + ->forDatabase(self::$wpdb) + ->build(); + + self::assertStringContainsString( + "compound_records.leftId, compound_records.rightId) IN (('1', '10'))", + $tupleSql + ); + } finally { + $alternate->close(); + } + } + + public function testForDatabaseRefusesACustomClauseBuilder(): void + { + $custom = $this->createMock(CoreClauseBuilder::class); + $custom->method('useTable')->willReturnSelf(); + $query = (new QueryBuilder()) + ->from(self::$predicateTable) + ->select('*') + ->where($custom); + + $this->expectException(UnsupportedCoordinationException::class); + $query->forDatabase(self::$wpdb); + } + + public function testQueryResetClearsHeldClauseAndGroupedChildren(): void + { + $child = (new ClauseBuilder()) + ->useTable(self::$predicateTable) + ->where('id', '=', 50); + $clause = (new ClauseBuilder()) + ->useTable(self::$predicateTable) + ->group('AND', $child); + $query = self::selectQuery($clause); + + $query->reset(); + + self::assertSame('', $clause->build()); + self::assertSame('', $child->build()); + } + private static function selectQuery(ClauseBuilder $clause): QueryBuilder { return (new QueryBuilder())