<?php

namespace classes;

use Aws\S3\S3Client;
use mysqli;
use Predis\Client as PredisClient;
use RuntimeException;
use Throwable;

class replication_manager
{
    private const KIND_DATABASE = 'database';
    private const KIND_REDIS = 'redis';
    private const KIND_MINIO = 'minio';
    private const MINIO_DEFAULT_BUCKETS = ['attachments', 'backups', 'invoices', 'pdfs', 'uploads', 'truckwashdev'];
    private const MINIO_SPACE_HEADROOM_PERCENT = 20.0;
    private const MARIADB_SEED_BATCH_ROWS = 500;
    private const MARIADB_SEED_STEP_SECONDS = 3;
    private const MARIADB_SEED_FREEZE_TTL_SECONDS = 900;
    private const MARIADB_SCHEMA_ONLY_TABLES = [
        'logs',
        'edge_gateway_log_entries',
        'edge_gateway_audit_logs',
        'edge_gateway_shell_sessions',
        'replication_status_snapshots',
        'replication_operations',
        'replication_audit_logs',
    ];

    public function __construct()
    {
        replication_schema_bootstrap::ensureTables();
    }

    public function summary(bool $refresh = false): array
    {
        $this->ensureEnvironmentPrimaryRows();
        if ($refresh) {
            $this->refreshStatuses();
        }

        $databaseHosts = $this->listHosts(self::KIND_DATABASE);
        $redisHosts = $this->listHosts(self::KIND_REDIS);
        $minioHosts = $this->listHosts(self::KIND_MINIO);

        return [
            'generated_at' => date('c'),
            'database' => [
                'primary' => $this->publicHost($this->primaryHost(self::KIND_DATABASE)),
                'hosts' => array_map(fn(array $host): array => $this->publicHost($host), $databaseHosts),
                'replication' => $this->buildReplicationSummary(self::KIND_DATABASE, $databaseHosts),
            ],
            'redis' => [
                'primary' => $this->publicHost($this->primaryHost(self::KIND_REDIS)),
                'hosts' => array_map(fn(array $host): array => $this->publicHost($host), $redisHosts),
                'replication' => $this->buildReplicationSummary(self::KIND_REDIS, $redisHosts),
            ],
            'minio' => [
                'primary' => $this->publicHost($this->primaryHost(self::KIND_MINIO)),
                'hosts' => array_map(fn(array $host): array => $this->publicHost($host), $minioHosts),
                'replication' => $this->buildReplicationSummary(self::KIND_MINIO, $minioHosts),
            ],
            'write_freeze' => application_write_freeze::state(),
        ];
    }

    public function dependencyReplication(string $kind): array
    {
        $kind = self::normalizeKind($kind);
        $this->ensureEnvironmentPrimaryRows();
        return $this->buildReplicationSummary($kind, $this->listHosts($kind));
    }

    public function addHost(string $kind, array $input, ?int $actorUserId = null): array
    {
        $kind = self::normalizeKind($kind);
        $host = $this->normalizeHostInput($kind, $input);

        $this->execute(
            "INSERT INTO replication_hosts (
                kind, label, host, port, database_name, database_index, username, password_secret,
                admin_username, admin_password_secret, replication_username, replication_password_secret,
                role, status, ssl_mode, options_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'replica', 'unknown', ?, ?)",
            'sssissssssssss',
            [
                $kind,
                $host['label'],
                $host['host'],
                $host['port'],
                $host['database_name'],
                $host['database_index'],
                $host['username'],
                $host['password_secret'],
                $host['admin_username'],
                $host['admin_password_secret'],
                $host['replication_username'],
                $host['replication_password_secret'],
                $host['ssl_mode'],
                self::jsonEncode($host['options']),
            ]
        );

        $id = $this->insertId();
        $this->audit($kind, $id, 'host_added', $actorUserId, 'info', [
            'label' => $host['label'],
            'host' => $host['host'],
            'port' => $host['port'],
        ]);
        $this->writeBootstrapSnapshot();

        return $this->publicHost($this->getHost($kind, $id));
    }

    public function testHost(string $kind, int $id, ?int $actorUserId = null): array
    {
        $kind = self::normalizeKind($kind);
        $host = $this->getHost($kind, $id);

        $status = match ($kind) {
            self::KIND_DATABASE => $this->testDatabaseHost($host),
            self::KIND_REDIS => $this->testRedisHost($host),
            self::KIND_MINIO => $this->testMinioHost($host),
        };

        $this->storeStatus($host, $status);
        $this->audit($kind, $id, 'host_tested', $actorUserId, $status['blockers'] === [] ? 'info' : 'warning', [
            'status' => $status['status'],
            'replication_percent' => $status['replication_percent'],
            'blockers' => $status['blockers'],
        ]);

        return [
            'host' => $this->publicHost($this->getHost($kind, $id)),
            'status' => $status,
        ];
    }

    public function testCredentials(string $kind, array $input): array
    {
        $kind = self::normalizeKind($kind);
        $host = $this->transientHost($kind, $input);
        $options = $this->decodeOptions($host);
        if ($kind === self::KIND_DATABASE && !empty($options['allow_preseeded_replica'])) {
            $host['connect_without_database'] = true;
        }
        $status = match ($kind) {
            self::KIND_DATABASE => $this->testDatabaseHost($host),
            self::KIND_REDIS => $this->testRedisHost($host),
            self::KIND_MINIO => $this->testMinioHost($host),
        };

        if ($kind === self::KIND_DATABASE
            && ($host['role'] ?? '') !== 'primary'
            && !empty($options['allow_preseeded_replica'])
            && $status['status'] !== 'down') {
            try {
                $target = $this->databaseConnection($host, true);
                try {
                    $seedBlockers = $this->databaseReplicaSeedBlockers($this->primaryHost(self::KIND_DATABASE), $host, $target);
                } finally {
                    $target->close();
                }
            } catch (Throwable $throwable) {
                $seedBlockers = [$throwable->getMessage()];
            }

            if ($seedBlockers !== []) {
                $status['blockers'] = array_values(array_unique(array_merge($status['blockers'], $seedBlockers)));
                $status['status'] = 'degraded';
                if ((float)$status['replication_percent'] >= 100.0) {
                    $status['replication_percent'] = 99.99;
                }
            }
        }

        return [
            'ok' => $status['status'] === 'ok',
            'host' => $this->publicHost($host),
            'status' => $status,
        ];
    }

    public function provisionHost(string $kind, int $id, ?int $actorUserId = null): array
    {
        $kind = self::normalizeKind($kind);
        $host = $this->getHost($kind, $id);
        $operationId = $this->activeOperationId($kind, $id, 'provision')
            ?? $this->startOperation($kind, $id, 'provision', $actorUserId);

        try {
            $result = match ($kind) {
                self::KIND_DATABASE => $this->provisionDatabaseHost($host, $operationId),
                self::KIND_REDIS => $this->provisionRedisHost($host, $operationId),
                self::KIND_MINIO => $this->provisionMinioHost($host, $operationId),
            };

            if (($result['operation']['status'] ?? null) === 'running') {
                $this->audit($kind, $id, 'host_provision_progress', $actorUserId, 'info', $result);
                return $result;
            }

            $status = $result['ok'] ? 'completed' : 'blocked';
            $this->finishOperation($operationId, $status, (float)($result['replication_percent'] ?? 0), $result['message'] ?? null, $result['blockers'] ?? []);
            $this->audit($kind, $id, 'host_provisioned', $actorUserId, $result['ok'] ? 'info' : 'warning', $result);
            return $result;
        } catch (Throwable $throwable) {
            $this->finishOperation($operationId, 'failed', 0, null, [$throwable->getMessage()]);
            $this->audit($kind, $id, 'host_provision_failed', $actorUserId, 'error', ['error' => $throwable->getMessage()]);
            throw $throwable;
        }
    }

    public function promoteHost(string $kind, int $id, ?int $actorUserId = null): array
    {
        $kind = self::normalizeKind($kind);
        $host = $this->getHost($kind, $id);
        if (($host['role'] ?? '') === 'primary') {
            return [
                'ok' => true,
                'message' => 'Host is already primary.',
                'host' => $this->publicHost($host),
                'blockers' => [],
            ];
        }

        $operationId = $this->startOperation($kind, $id, 'promote', $actorUserId);
        $owner = 'replication-promote-' . $kind . '-' . $id . '-' . bin2hex(random_bytes(4));
        $lockHandle = $this->acquirePromotionLock();

        try {
            application_write_freeze::freeze('Replication promotion in progress.', $owner, 600);
            $result = match ($kind) {
                self::KIND_DATABASE => $this->promoteDatabaseHost($host),
                self::KIND_REDIS => $this->promoteRedisHost($host),
                self::KIND_MINIO => $this->promoteMinioHost($host),
            };

            $this->finishOperation($operationId, 'completed', 100, $result['message'] ?? null, []);
            $this->audit($kind, $id, 'host_promoted', $actorUserId, 'critical', $result);
            return $result;
        } catch (Throwable $throwable) {
            $this->finishOperation($operationId, 'failed', 0, null, [$throwable->getMessage()]);
            $this->audit($kind, $id, 'host_promotion_failed', $actorUserId, 'error', ['error' => $throwable->getMessage()]);
            throw $throwable;
        } finally {
            application_write_freeze::unfreeze($owner);
            $this->releasePromotionLock($lockHandle);
        }
    }

    public function removeHost(string $kind, int $id, ?int $actorUserId = null): array
    {
        $kind = self::normalizeKind($kind);
        $host = $this->getHost($kind, $id);
        if (!self::replicationHostCanBeRemoved($host)) {
            if (($host['role'] ?? '') === 'primary') {
                throw new RuntimeException('Primary hosts cannot be removed. Promote a healthy replica first.');
            }
            throw new RuntimeException('Only inactive prior hosts or unhealthy replicas can be removed.');
        }
        $this->execute(
            "UPDATE replication_hosts SET deleted_at = NOW(), status = 'removed' WHERE id = ? AND kind = ?",
            'is',
            [$id, $kind]
        );
        $this->audit($kind, $id, 'host_removed', $actorUserId, 'warning', [
            'label' => $host['label'] ?? '',
            'host' => $host['host'] ?? '',
        ]);
        $this->writeBootstrapSnapshot();

        return [
            'ok' => true,
            'message' => 'Replication host removed.',
            'id' => $id,
            'kind' => $kind,
        ];
    }

    public static function replicationHostCanBeRemoved(array $host): bool
    {
        $role = (string)($host['role'] ?? '');
        if ($role === 'primary') {
            return false;
        }
        if ($role === 'inactive') {
            return true;
        }

        $status = (string)($host['status'] ?? 'unknown');
        return $role === 'replica' && in_array($status, ['degraded', 'down', 'unknown', 'not_configured'], true);
    }

    public static function composeTemplate(array $input): array
    {
        $kind = self::normalizeKind((string)($input['kind'] ?? self::KIND_DATABASE));
        $role = self::normalizeComposeRole((string)($input['role'] ?? 'replica'));

        return match ($kind) {
            self::KIND_DATABASE => self::databaseComposeTemplate($input, $role),
            self::KIND_REDIS => self::redisComposeTemplate($input, $role),
            self::KIND_MINIO => self::minioComposeTemplate($input, $role),
        };
    }

    public static function normalizeKind(string $kind): string
    {
        $kind = strtolower(trim($kind));
        if (in_array($kind, ['database', 'databases', 'mysql', 'db'], true)) {
            return self::KIND_DATABASE;
        }
        if ($kind === self::KIND_REDIS) {
            return self::KIND_REDIS;
        }
        if (in_array($kind, ['minio', 's3', 'object-storage', 'object_storage'], true)) {
            return self::KIND_MINIO;
        }
        throw new RuntimeException('Unsupported replication kind.');
    }

    private static function databaseComposeTemplate(array $input, string $role): array
    {
        $serviceName = self::composeIdentifier($input['service_name'] ?? null, 'mariadb-' . $role);
        $volumeName = self::composeIdentifier($input['volume_name'] ?? null, $serviceName . '-data');
        $database = self::composeScalar($input['database'] ?? null, 'nnks_db');
        $username = self::composeScalar($input['username'] ?? null, 'nnks_db_user');
        $image = self::composeImage($input['image'] ?? null, 'mariadb:11');
        $hostPort = self::boundedInt($input['host_port'] ?? null, $role === 'primary' ? 3306 : 3307, 1, 65535);
        $serverId = self::boundedInt($input['server_id'] ?? null, $role === 'primary' ? 1 : 2, 1, 4294967295);
        $rootPassword = self::composePassword($input['admin_password'] ?? null);
        $applicationPassword = self::composePassword($input['password'] ?? null);
        $replicationUsername = self::composeScalar($input['replication_username'] ?? null, 'replication');
        $replicationPassword = self::composePassword($input['replication_password'] ?? null);
        $primaryHost = self::composeScalar($input['primary_host'] ?? null, '<primary-host>');
        $primaryPort = self::boundedInt($input['primary_port'] ?? null, 3306, 1, 65535);

        $command = [
            'mariadbd',
            '--server-id=' . $serverId,
            '--log-bin=/var/lib/mysql/mariadb-bin',
            '--binlog-format=ROW',
            '--gtid-strict-mode=ON',
            '--expire-logs-days=7',
        ];
        if ($role === 'replica') {
            $command[] = '--read-only=ON';
            foreach (self::MARIADB_SCHEMA_ONLY_TABLES as $tableName) {
                $command[] = '--replicate-ignore-table=' . $database . '.' . $tableName;
            }
        }

        $lines = [
            'services:',
            '  ' . $serviceName . ':',
            '    image: ' . self::yamlQuote($image),
            '    restart: unless-stopped',
            '    environment:',
            '      MARIADB_ROOT_PASSWORD: "${MARIADB_ROOT_PASSWORD:?set MARIADB_ROOT_PASSWORD}"',
            '      MARIADB_DATABASE: ' . self::yamlQuote($database),
            '      MARIADB_USER: ' . self::yamlQuote($username),
            '      MARIADB_PASSWORD: "${MARIADB_PASSWORD:?set MARIADB_PASSWORD}"',
            '    command:',
        ];

        foreach ($command as $argument) {
            $lines[] = '      - ' . self::yamlQuote($argument);
        }

        $lines = array_merge($lines, [
            '    volumes:',
            '      - ' . $volumeName . ':/var/lib/mysql',
            '    ports:',
            '      - ' . self::yamlQuote($hostPort . ':3306'),
            '    healthcheck:',
            '      test:',
            '        - "CMD-SHELL"',
            '        - "mariadb-admin ping -h 127.0.0.1 -uroot -p$${MARIADB_ROOT_PASSWORD} --silent"',
            '      interval: 10s',
            '      timeout: 5s',
            '      retries: 12',
        ]);

        if ($role === 'replica') {
            $seedServiceName = self::composeIdentifier($serviceName . '-seed', 'mariadb-replica-seed');
            $seedScript = [
                'marker="/var/lib/mysql/.truckwash-replica-seeded"',
                'if [ -f "$marker" ]; then',
                '  echo "Replica already seeded."',
                '  exit 0',
                'fi',
                'echo "Waiting for local replica..."',
                'until mariadb-admin ping -h ' . self::shellArg($serviceName) . ' -uroot -p"$MARIADB_ROOT_PASSWORD" --silent; do sleep 2; done',
                'echo "Importing seed from primary..."',
                'mariadb-dump --host="$MARIADB_PRIMARY_HOST" --port="$MARIADB_PRIMARY_PORT" --user="$MARIADB_PRIMARY_ADMIN_USER" --password="$MARIADB_PRIMARY_ADMIN_PASSWORD" --single-transaction --quick --routines --triggers --events --gtid --master-data=2 ' . self::mariaDbSchemaOnlyDumpIgnoreArgs('$MARIADB_SEED_DATABASE') . ' --databases "$MARIADB_SEED_DATABASE" | mariadb --host=' . self::shellArg($serviceName) . ' --user=root --password="$MARIADB_ROOT_PASSWORD"',
                'for table in ' . implode(' ', self::MARIADB_SCHEMA_ONLY_TABLES) . '; do',
                '  mariadb-dump --host="$MARIADB_PRIMARY_HOST" --port="$MARIADB_PRIMARY_PORT" --user="$MARIADB_PRIMARY_ADMIN_USER" --password="$MARIADB_PRIMARY_ADMIN_PASSWORD" --single-transaction --quick --no-data "$MARIADB_SEED_DATABASE" "$table" | mariadb --host=' . self::shellArg($serviceName) . ' --user=root --password="$MARIADB_ROOT_PASSWORD" "$MARIADB_SEED_DATABASE" || true',
                'done',
                'touch "$marker"',
                'echo "Replica seed completed."',
            ];

            $lines = array_merge($lines, [
                '  ' . $seedServiceName . ':',
                '    image: ' . self::yamlQuote($image),
                '    restart: "no"',
                '    depends_on:',
                '      ' . $serviceName . ':',
                '        condition: service_healthy',
                '    environment:',
                '      MARIADB_ROOT_PASSWORD: "${MARIADB_ROOT_PASSWORD:?set MARIADB_ROOT_PASSWORD}"',
                '      MARIADB_PRIMARY_HOST: "${MARIADB_PRIMARY_HOST:?set MARIADB_PRIMARY_HOST}"',
                '      MARIADB_PRIMARY_PORT: "${MARIADB_PRIMARY_PORT:-3306}"',
                '      MARIADB_PRIMARY_ADMIN_USER: "${MARIADB_PRIMARY_ADMIN_USER:-root}"',
                '      MARIADB_PRIMARY_ADMIN_PASSWORD: "${MARIADB_PRIMARY_ADMIN_PASSWORD:?set MARIADB_PRIMARY_ADMIN_PASSWORD}"',
                '      MARIADB_SEED_DATABASE: ' . self::yamlQuote($database),
                '    volumes:',
                '      - ' . $volumeName . ':/var/lib/mysql',
                '    entrypoint:',
                '      - /bin/sh',
                '      - -ec',
                '      - |',
            ]);
            foreach ($seedScript as $scriptLine) {
                $lines[] = '        ' . $scriptLine;
            }
        }

        $lines = array_merge($lines, [
            'volumes:',
            '  ' . $volumeName . ':',
        ]);

        $steps = [
            'Deploy this compose file as a normal Docker Compose or Coolify compose service.',
            'Keep server-id unique across the MariaDB primary and every replica.',
            'Create or store a replication user on the primary with REPLICATION SLAVE privileges.',
        ];
        if ($role === 'replica') {
            $steps[] = 'Fill MARIADB_PRIMARY_ADMIN_PASSWORD in the generated .env file.';
            $steps[] = 'Deploy the compose file and wait for the seed service to complete successfully.';
            $steps[] = 'Test the connection, then save and provision the replica.';
        } else {
            $steps[] = 'Add the primary credentials in the superuser UI after the service is reachable.';
        }

        $seedCommand = implode(' ', [
            'mariadb-dump',
            '--host=' . self::shellArg($primaryHost),
            '--port=' . $primaryPort,
            '--user=<primary-admin-user>',
            '--password',
            '--single-transaction',
            '--quick',
            '--routines',
            '--triggers',
            '--events',
            '--gtid',
            '--master-data=2',
            ...self::mariaDbSchemaOnlySeedCommandIgnoreArgs($database),
            '--databases',
            self::shellArg($database),
            '|',
            'mariadb',
            '--host=<replica-host>',
            '--port=' . $hostPort,
            '--user=root',
            '--password',
        ]);
        $envLines = [
            'MARIADB_ROOT_PASSWORD=' . $rootPassword,
            'MARIADB_PASSWORD=' . $applicationPassword,
        ];
        if ($role === 'replica') {
            $envLines[] = 'MARIADB_PRIMARY_HOST=' . $primaryHost;
            $envLines[] = 'MARIADB_PRIMARY_PORT=' . $primaryPort;
            $envLines[] = 'MARIADB_PRIMARY_ADMIN_USER=root';
            $envLines[] = 'MARIADB_PRIMARY_ADMIN_PASSWORD=';
        }

        return [
            'kind' => self::KIND_DATABASE,
            'engine' => 'mariadb',
            'role' => $role,
            'service_name' => $serviceName,
            'host_port' => $hostPort,
            'server_id' => $serverId,
            'compose' => implode("\n", $lines) . "\n",
            'env' => implode("\n", $envLines) . "\n",
            'seed_command' => $role === 'replica' ? $seedCommand : '',
            'credentials' => [
                'label' => $serviceName,
                'host' => '',
                'port' => $hostPort,
                'database' => $database,
                'username' => $username,
                'password' => $applicationPassword,
                'admin_username' => 'root',
                'admin_password' => $rootPassword,
                'replication_username' => $replicationUsername,
                'replication_password' => $replicationPassword,
                'ssl_mode' => 'DISABLED',
                'allow_preseeded_replica' => $role === 'replica',
            ],
            'steps' => $steps,
        ];
    }

    private static function redisComposeTemplate(array $input, string $role): array
    {
        $serviceName = self::composeIdentifier($input['service_name'] ?? null, 'redis-' . $role);
        $volumeName = self::composeIdentifier($input['volume_name'] ?? null, $serviceName . '-data');
        $image = self::composeImage($input['image'] ?? null, 'redis:7');
        $hostPort = self::boundedInt($input['host_port'] ?? null, $role === 'primary' ? 6379 : 6380, 1, 65535);
        $primaryHost = self::composeScalar($input['primary_host'] ?? null, 'redis-primary');
        $primaryPort = self::boundedInt($input['primary_port'] ?? null, 6379, 1, 65535);
        $redisPassword = self::composePassword($input['password'] ?? null);
        $primaryPassword = self::composeScalar($input['primary_password'] ?? null, '');
        $primaryUsername = self::composeScalar($input['primary_username'] ?? null, '');

        $script = [
            'if [ ! -f /data/redis.conf ]; then',
            '  {',
            '    echo "appendonly yes"',
            '    echo "requirepass $$REDIS_PASSWORD"',
        ];
        if ($role === 'replica') {
            $script[] = '    echo "replicaof $$REDIS_PRIMARY_HOST $${REDIS_PRIMARY_PORT:-6379}"';
            $script[] = '    echo "masterauth $$REDIS_PRIMARY_PASSWORD"';
            $script[] = '    if [ -n "$${REDIS_PRIMARY_USERNAME:-}" ] && [ "$${REDIS_PRIMARY_USERNAME}" != "default" ]; then';
            $script[] = '      echo "masteruser $$REDIS_PRIMARY_USERNAME"';
            $script[] = '    fi';
        }
        $script = array_merge($script, [
            '  } > /data/redis.conf',
            'fi',
            'exec redis-server /data/redis.conf',
        ]);

        $lines = [
            'services:',
            '  ' . $serviceName . ':',
            '    image: ' . self::yamlQuote($image),
            '    restart: unless-stopped',
            '    environment:',
            '      REDIS_PASSWORD: "${REDIS_PASSWORD:?set REDIS_PASSWORD}"',
        ];
        if ($role === 'replica') {
            $lines[] = '      REDIS_PRIMARY_HOST: "${REDIS_PRIMARY_HOST:?set REDIS_PRIMARY_HOST}"';
            $lines[] = '      REDIS_PRIMARY_PORT: "${REDIS_PRIMARY_PORT:-6379}"';
            $lines[] = '      REDIS_PRIMARY_PASSWORD: "${REDIS_PRIMARY_PASSWORD:?set REDIS_PRIMARY_PASSWORD}"';
            $lines[] = '      REDIS_PRIMARY_USERNAME: "${REDIS_PRIMARY_USERNAME:-}"';
        }
        $lines[] = '    command:';
        $lines[] = '      - /bin/sh';
        $lines[] = '      - -ec';
        $lines[] = '      - |';
        foreach ($script as $scriptLine) {
            $lines[] = '        ' . $scriptLine;
        }

        $lines = array_merge($lines, [
            '    volumes:',
            '      - ' . $volumeName . ':/data',
            '    ports:',
            '      - ' . self::yamlQuote($hostPort . ':6379'),
            '    healthcheck:',
            '      test:',
            '        - "CMD-SHELL"',
            '        - "redis-cli --no-auth-warning -a \"$${REDIS_PASSWORD}\" ping | grep PONG"',
            '      interval: 10s',
            '      timeout: 5s',
            '      retries: 12',
            'volumes:',
            '  ' . $volumeName . ':',
        ]);

        $steps = [
            'Deploy this compose file as a normal Docker Compose or Coolify compose service.',
            'Add the Redis credentials in the superuser UI after the service is reachable.',
        ];
        if ($role === 'replica') {
            $steps[] = 'Use the current Redis primary host and password for REDIS_PRIMARY_HOST and REDIS_PRIMARY_PASSWORD, then run Test in the superuser UI.';
        }

        $envLines = [
            'REDIS_PASSWORD=' . $redisPassword,
        ];
        if ($role === 'replica') {
            $envLines[] = 'REDIS_PRIMARY_HOST=' . $primaryHost;
            $envLines[] = 'REDIS_PRIMARY_PORT=' . $primaryPort;
            $envLines[] = 'REDIS_PRIMARY_PASSWORD=' . $primaryPassword;
            $envLines[] = 'REDIS_PRIMARY_USERNAME=' . $primaryUsername;
        }

        return [
            'kind' => self::KIND_REDIS,
            'engine' => 'redis',
            'role' => $role,
            'service_name' => $serviceName,
            'host_port' => $hostPort,
            'compose' => implode("\n", $lines) . "\n",
            'env' => implode("\n", $envLines) . "\n",
            'credentials' => [
                'label' => $serviceName,
                'host' => '',
                'port' => $hostPort,
                'database' => 0,
                'username' => '',
                'password' => $redisPassword,
            ],
            'steps' => $steps,
        ];
    }

    private static function minioComposeTemplate(array $input, string $role): array
    {
        $serviceName = self::composeIdentifier($input['service_name'] ?? null, 'minio-' . $role);
        $volumeName = self::composeIdentifier($input['volume_name'] ?? null, $serviceName . '-data');
        $image = self::composeImage($input['image'] ?? null, 'minio/minio:latest');
        $mcImage = self::composeImage($input['mc_image'] ?? null, 'minio/mc:latest');
        $hostPort = self::boundedInt($input['host_port'] ?? null, $role === 'primary' ? 9000 : 9010, 1, 65535);
        $consolePort = self::boundedInt($input['console_port'] ?? null, $role === 'primary' ? 9001 : 9011, 1, 65535);
        $rootUser = self::composeScalar($input['username'] ?? $input['access_key'] ?? null, 'truckwash-minio');
        $rootPassword = self::composePassword($input['password'] ?? $input['secret_key'] ?? null);
        $buckets = self::normalizeMinioBuckets($input['buckets'] ?? self::MINIO_DEFAULT_BUCKETS);

        $setupScript = [
            'until mc alias set local http://' . self::shellArg($serviceName) . ':9000 "$MINIO_ROOT_USER" "$MINIO_ROOT_PASSWORD"; do sleep 2; done',
        ];
        foreach ($buckets as $bucket) {
            $bucketArg = self::shellArg('local/' . $bucket);
            $setupScript[] = 'mc mb --ignore-existing ' . $bucketArg;
            $setupScript[] = 'mc version enable ' . $bucketArg . ' || true';
        }

        $lines = [
            'services:',
            '  ' . $serviceName . ':',
            '    image: ' . self::yamlQuote($image),
            '    restart: unless-stopped',
            '    command:',
            '      - server',
            '      - /data',
            '      - --console-address',
            '      - ":9001"',
            '    environment:',
            '      MINIO_ROOT_USER: "${MINIO_ROOT_USER:?set MINIO_ROOT_USER}"',
            '      MINIO_ROOT_PASSWORD: "${MINIO_ROOT_PASSWORD:?set MINIO_ROOT_PASSWORD}"',
            '    volumes:',
            '      - ' . $volumeName . ':/data',
            '    ports:',
            '      - ' . self::yamlQuote($hostPort . ':9000'),
            '      - ' . self::yamlQuote($consolePort . ':9001'),
            '    healthcheck:',
            '      test:',
            '        - "CMD"',
            '        - "curl"',
            '        - "-f"',
            '        - "http://127.0.0.1:9000/minio/health/live"',
            '      interval: 10s',
            '      timeout: 5s',
            '      retries: 12',
            '  ' . $serviceName . '-setup:',
            '    image: ' . self::yamlQuote($mcImage),
            '    restart: "no"',
            '    depends_on:',
            '      ' . $serviceName . ':',
            '        condition: service_healthy',
            '    environment:',
            '      MINIO_ROOT_USER: "${MINIO_ROOT_USER:?set MINIO_ROOT_USER}"',
            '      MINIO_ROOT_PASSWORD: "${MINIO_ROOT_PASSWORD:?set MINIO_ROOT_PASSWORD}"',
        ];
        if ($role === 'replica') {
            $lines[] = '      MINIO_PRIMARY_ENDPOINT: "${MINIO_PRIMARY_ENDPOINT:-}"';
            $lines[] = '      MINIO_PRIMARY_ACCESS_KEY: "${MINIO_PRIMARY_ACCESS_KEY:-}"';
            $lines[] = '      MINIO_PRIMARY_SECRET_KEY: "${MINIO_PRIMARY_SECRET_KEY:-}"';
        }
        $lines = array_merge($lines, [
            '    entrypoint:',
            '      - /bin/sh',
            '      - -ec',
            '      - |',
        ]);
        foreach ($setupScript as $scriptLine) {
            $lines[] = '        ' . $scriptLine;
        }
        $lines = array_merge($lines, [
            'volumes:',
            '  ' . $volumeName . ':',
        ]);

        $envLines = [
            'MINIO_ROOT_USER=' . $rootUser,
            'MINIO_ROOT_PASSWORD=' . $rootPassword,
            'MINIO_BUCKETS=' . implode(',', $buckets),
        ];
        if ($role === 'replica') {
            $envLines[] = 'MINIO_PRIMARY_ENDPOINT=';
            $envLines[] = 'MINIO_PRIMARY_ACCESS_KEY=';
            $envLines[] = 'MINIO_PRIMARY_SECRET_KEY=';
        }

        $steps = [
            'Deploy this compose file as a normal Docker Compose or Coolify compose service.',
            'The setup service creates required buckets and enables bucket versioning.',
            'Add the MinIO credentials in the superuser UI after the API endpoint is reachable.',
        ];
        if ($role === 'replica') {
            $steps[] = 'Fill the MINIO_PRIMARY_* .env values for reference; managed bucket replication is configured from the superuser UI.';
            $steps[] = 'Test the connection, then save and provision the replica.';
        }

        return [
            'kind' => self::KIND_MINIO,
            'engine' => 'minio',
            'role' => $role,
            'service_name' => $serviceName,
            'host_port' => $hostPort,
            'console_port' => $consolePort,
            'compose' => implode("\n", $lines) . "\n",
            'env' => implode("\n", $envLines) . "\n",
            'credentials' => [
                'label' => $serviceName,
                'host' => '',
                'port' => $hostPort,
                'scheme' => 'http',
                'endpoint' => '',
                'buckets' => $buckets,
                'console_port' => $consolePort,
                'username' => $rootUser,
                'password' => $rootPassword,
                'space_headroom_percent' => self::MINIO_SPACE_HEADROOM_PERCENT,
            ],
            'steps' => $steps,
        ];
    }

    private static function normalizeComposeRole(string $role): string
    {
        $role = strtolower(trim($role));
        if (in_array($role, ['primary', 'replica'], true)) {
            return $role;
        }
        throw new RuntimeException('Unsupported compose role.');
    }

    private static function composeIdentifier(mixed $value, string $fallback): string
    {
        $identifier = strtolower(trim((string)$value));
        $identifier = (string)preg_replace('/[^a-z0-9_.-]+/', '-', $identifier);
        $identifier = trim($identifier, '-_.');
        return $identifier !== '' ? $identifier : $fallback;
    }

    private static function composeScalar(mixed $value, string $fallback): string
    {
        $scalar = trim((string)$value);
        return $scalar !== '' ? $scalar : $fallback;
    }

    private static function composePassword(mixed $value): string
    {
        $password = trim((string)$value);
        if ($password !== '') {
            return $password;
        }

        return self::generateSecret(24);
    }

    private static function generateSecret(int $bytes): string
    {
        return rtrim(strtr(base64_encode(random_bytes($bytes)), '+/', '-_'), '=');
    }

    private static function composeImage(mixed $value, string $fallback): string
    {
        $image = trim((string)$value);
        if ($image === '' || preg_match('/^[a-zA-Z0-9._:\/-]+$/', $image) !== 1) {
            return $fallback;
        }
        return $image;
    }

    private static function boundedInt(mixed $value, int $fallback, int $min, int $max): int
    {
        if (filter_var($value, FILTER_VALIDATE_INT) === false) {
            return $fallback;
        }
        return max($min, min($max, (int)$value));
    }

    private static function yamlQuote(mixed $value): string
    {
        $encoded = json_encode((string)$value, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);
        return is_string($encoded) ? $encoded : '""';
    }

    private static function shellArg(string $value): string
    {
        return "'" . str_replace("'", "'\"'\"'", $value) . "'";
    }

    private static function mariaDbSchemaOnlyDumpIgnoreArgs(string $databaseExpression): string
    {
        return implode(' ', array_map(
            static fn(string $tableName): string => '--ignore-table="' . $databaseExpression . '.' . $tableName . '"',
            self::MARIADB_SCHEMA_ONLY_TABLES
        ));
    }

    private static function mariaDbSchemaOnlySeedCommandIgnoreArgs(string $database): array
    {
        return array_map(
            static fn(string $tableName): string => '--ignore-table=' . self::shellArg($database . '.' . $tableName),
            self::MARIADB_SCHEMA_ONLY_TABLES
        );
    }

    private static function quoteIdentifier(string $identifier): string
    {
        return '`' . str_replace('`', '``', $identifier) . '`';
    }

    private static function sqlString(mysqli $connection, string $value): string
    {
        return "'" . $connection->real_escape_string($value) . "'";
    }

    public static function mysqlGtidIntervalCount(string $gtidSet): int
    {
        $count = 0;
        foreach (self::parseMysqlGtidSet($gtidSet) as $intervals) {
            foreach ($intervals as [$start, $end]) {
                $count += max(0, $end - $start + 1);
            }
        }
        return $count;
    }

    public static function mysqlGtidCoveragePercent(string $sourceSet, string $executedSet): float
    {
        $source = self::parseMysqlGtidSet($sourceSet);
        $executed = self::parseMysqlGtidSet($executedSet);
        $total = 0;
        $covered = 0;

        foreach ($source as $uuid => $sourceIntervals) {
            foreach ($sourceIntervals as [$sourceStart, $sourceEnd]) {
                $total += max(0, $sourceEnd - $sourceStart + 1);
                foreach ($executed[$uuid] ?? [] as [$executedStart, $executedEnd]) {
                    $start = max($sourceStart, $executedStart);
                    $end = min($sourceEnd, $executedEnd);
                    if ($end >= $start) {
                        $covered += $end - $start + 1;
                    }
                }
            }
        }

        if ($total === 0) {
            return 100.0;
        }

        return round(min(100, max(0, ($covered / $total) * 100)), 2);
    }

    public static function redisOffsetPercent(int $primaryOffset, int $replicaOffset): float
    {
        if ($primaryOffset <= 0) {
            return 100.0;
        }

        return round(min(100, max(0, ($replicaOffset / $primaryOffset) * 100)), 2);
    }

    public static function minioRequiredFreeBytes(int $sourceBytes, float $headroomPercent = self::MINIO_SPACE_HEADROOM_PERCENT): int
    {
        return (int)ceil(max(0, $sourceBytes) * (1 + max(0.0, $headroomPercent) / 100));
    }

    public static function minioByteReplicationPercent(int $sourceBytes, int $replicaBytes): float
    {
        if ($sourceBytes <= 0) {
            return 100.0;
        }

        return round(min(100, max(0, ($replicaBytes / $sourceBytes) * 100)), 2);
    }

    public static function minioSpaceBlockers(?int $availableBytes, int $requiredBytes): array
    {
        if ($availableBytes === null) {
            return ['MinIO target free space could not be determined.'];
        }
        if ($availableBytes < $requiredBytes) {
            return ['MinIO target does not have enough free space. Required ' . $requiredBytes . ' bytes, available ' . $availableBytes . ' bytes.'];
        }

        return [];
    }

    public static function normalizeMinioBuckets(mixed $value): array
    {
        if (is_string($value)) {
            $value = preg_split('/[\s,]+/', $value);
        }
        if (!is_array($value)) {
            $value = self::MINIO_DEFAULT_BUCKETS;
        }

        $buckets = [];
        foreach ($value as $bucket) {
            $bucket = strtolower(trim((string)$bucket));
            if ($bucket === '' || preg_match('/^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$/', $bucket) !== 1) {
                continue;
            }
            $buckets[] = $bucket;
        }

        $buckets = array_values(array_unique($buckets));
        return $buckets !== [] ? $buckets : self::MINIO_DEFAULT_BUCKETS;
    }

    public static function mariadbGtidCoveragePercent(string $sourceSet, string $replicaSet): float
    {
        $source = self::parseMariaDbGtidSet($sourceSet);
        $replica = self::parseMariaDbGtidSet($replicaSet);
        $total = array_sum($source);
        if ($total <= 0) {
            return 100.0;
        }

        $covered = 0;
        foreach ($source as $domain => $sourceSequence) {
            $covered += min($sourceSequence, $replica[$domain] ?? 0);
        }

        return round(min(100, max(0, ($covered / $total) * 100)), 2);
    }

    private static function parseMariaDbGtidSet(string $gtidSet): array
    {
        $positions = [];
        foreach (explode(',', trim($gtidSet)) as $gtid) {
            $gtid = trim($gtid);
            if ($gtid === '') {
                continue;
            }

            $parts = explode('-', $gtid);
            if (count($parts) !== 3) {
                continue;
            }

            [$domain, , $sequence] = array_map('intval', $parts);
            if ($sequence <= 0) {
                continue;
            }

            $positions[$domain] = max($positions[$domain] ?? 0, $sequence);
        }

        return $positions;
    }

    private static function parseMysqlGtidSet(string $gtidSet): array
    {
        $parsed = [];
        foreach (explode(',', trim($gtidSet)) as $uuidSet) {
            $uuidSet = trim($uuidSet);
            if ($uuidSet === '') {
                continue;
            }

            $parts = explode(':', $uuidSet);
            if (count($parts) < 2) {
                continue;
            }

            $uuid = strtolower(array_shift($parts));
            foreach ($parts as $interval) {
                if (str_contains($interval, '-')) {
                    [$start, $end] = array_map('intval', explode('-', $interval, 2));
                } else {
                    $start = $end = (int)$interval;
                }
                if ($start <= 0 || $end <= 0) {
                    continue;
                }
                if ($end < $start) {
                    [$start, $end] = [$end, $start];
                }
                $parsed[$uuid][] = [$start, $end];
            }
        }

        foreach ($parsed as $uuid => $intervals) {
            usort($intervals, static fn(array $a, array $b): int => $a[0] <=> $b[0]);
            $merged = [];
            foreach ($intervals as [$start, $end]) {
                $lastIndex = count($merged) - 1;
                if ($lastIndex >= 0 && $start <= $merged[$lastIndex][1] + 1) {
                    $merged[$lastIndex][1] = max($merged[$lastIndex][1], $end);
                    continue;
                }
                $merged[] = [$start, $end];
            }
            $parsed[$uuid] = $merged;
        }

        return $parsed;
    }

    private function provisionDatabaseHost(array $host, int $operationId): array
    {
        $primary = $this->primaryHost(self::KIND_DATABASE);
        if ($primary === null) {
            throw new RuntimeException('No database primary is registered.');
        }

        $options = $this->decodeOptions($host);
        $usePreseededReplica = !empty($options['allow_preseeded_replica']);
        $targetStatus = $this->testDatabaseHost(array_merge($host, [
            'test_connectivity_only' => true,
            'connect_without_database' => $usePreseededReplica,
        ]));
        $primaryStatus = $this->testDatabaseHost($primary);
        $blockers = array_merge($targetStatus['blockers'], $primaryStatus['blockers']);
        $targetEngine = self::databaseEngine($targetStatus['raw'] ?? []);
        $primaryEngine = self::databaseEngine($primaryStatus['raw'] ?? []);

        if (($targetStatus['raw']['server_id'] ?? null) !== null
            && ($primaryStatus['raw']['server_id'] ?? null) !== null
            && (int)$targetStatus['raw']['server_id'] === (int)$primaryStatus['raw']['server_id']) {
            $blockers[] = 'Database replica must have a unique server_id.';
        }
        if ($targetEngine !== $primaryEngine) {
            $blockers[] = 'Database primary and replica must use the same engine family.';
        }

        $cloneReady = (bool)($targetStatus['raw']['clone_plugin_active'] ?? false);
        $primaryCloneReady = (bool)($primaryStatus['raw']['clone_plugin_active'] ?? false);
        if ($targetEngine === 'mariadb' && !$usePreseededReplica) {
            $blockers[] = 'MariaDB replicas must be safely seeded before managed replication can be configured.';
        }
        if ($targetEngine === 'mysql' && !$usePreseededReplica) {
            if (!$cloneReady) {
                $blockers[] = 'MySQL Clone plugin is not active on the target. Set allow_preseeded_replica only after the target has been safely seeded.';
            }
            if (!$primaryCloneReady) {
                $blockers[] = 'MySQL Clone plugin is not active on the primary donor.';
            }
        }

        if ($blockers !== []) {
            $this->storeStatus($host, array_replace($targetStatus, ['blockers' => array_values(array_unique($blockers))]));
            return [
                'ok' => false,
                'message' => 'Database replica provisioning is blocked.',
                'blockers' => array_values(array_unique($blockers)),
                'replication_percent' => $targetStatus['replication_percent'],
                'host' => $this->publicHost($host),
            ];
        }

        $target = $this->databaseConnection($host, true, $usePreseededReplica);
        try {
            $primaryCredentials = $this->credentials($primary);
            $hostCredentials = $this->credentials($host);
            $replicationUser = $hostCredentials['replication_username']
                ?: ($primaryCredentials['replication_username'] ?: $primaryCredentials['username']);
            $replicationPassword = $hostCredentials['replication_password']
                ?: ($primaryCredentials['replication_password'] ?: $primaryCredentials['password']);
            $shouldManageReplicationUser = ($hostCredentials['replication_username'] ?? '') !== ''
                && ($hostCredentials['replication_password'] ?? '') !== '';

            if ($usePreseededReplica && $targetEngine === 'mariadb') {
                $seedContext = $this->operationContext($operationId);
                $seedInProgress = ($seedContext['phase'] ?? '') !== '' && ($seedContext['phase'] ?? '') !== 'complete';
                $seedBlockers = $this->databaseReplicaSeedBlockers($primary, $host, $target);
                if ($seedInProgress || $seedBlockers !== []) {
                    $seedResult = $this->advanceMariaDbReplicaSeed($operationId, $primary, $host, $target);
                    $targetStatus = $seedResult['status'];

                    if (($seedResult['running'] ?? false) === true) {
                        $this->storeStatus($host, $targetStatus);
                        return [
                            'ok' => true,
                            'message' => $seedResult['message'],
                            'blockers' => [],
                            'replication_percent' => $targetStatus['replication_percent'],
                            'operation' => [
                                'id' => $operationId,
                                'status' => 'running',
                                'progress_percent' => $targetStatus['replication_percent'],
                                'message' => $seedResult['message'],
                            ],
                            'host' => $this->publicHost($host),
                        ];
                    }
                }
            } elseif ($usePreseededReplica) {
                $seedBlockers = $this->databaseReplicaSeedBlockers($primary, $host, $target);
                if ($seedBlockers !== []) {
                    $combinedBlockers = array_values(array_unique(array_merge($targetStatus['blockers'], $seedBlockers)));
                    $this->storeStatus($host, array_replace($targetStatus, [
                        'status' => 'degraded',
                        'replication_percent' => min(99.99, (float)$targetStatus['replication_percent']),
                        'blockers' => $combinedBlockers,
                    ]));
                    return [
                        'ok' => false,
                        'message' => 'Database replica provisioning is blocked.',
                        'blockers' => $combinedBlockers,
                        'replication_percent' => min(99.99, (float)$targetStatus['replication_percent']),
                        'host' => $this->publicHost($host),
                    ];
                }
            }

            if ($shouldManageReplicationUser) {
                $this->ensureDatabaseReplicationUser($primary, $replicationUser, $replicationPassword);
            }

            if ($targetEngine === 'mysql' && !$usePreseededReplica) {
                $this->runMysqlClone($target, $primary, $replicationUser, $replicationPassword);
                $target->close();
                $target = $this->waitForDatabaseConnection($host, true, 120);
            }

            if ($targetEngine === 'mariadb') {
                $this->configureMariaDbReplication($target, $primary, $host, $replicationUser, $replicationPassword);
            } else {
                $this->configureMySqlReplication($target, $primary, $replicationUser, $replicationPassword);
            }
        } finally {
            $target->close();
        }

        $this->execute(
            "UPDATE replication_hosts SET replication_source_id = ?, status = 'provisioning' WHERE id = ?",
            'ii',
            [(int)$primary['id'], (int)$host['id']]
        );

        $status = $this->testDatabaseHost($this->getHost(self::KIND_DATABASE, (int)$host['id']));
        $this->storeStatus($host, $status);

        return [
            'ok' => true,
            'healthy' => $status['blockers'] === [],
            'message' => $targetEngine === 'mariadb'
                ? 'MariaDB replication was configured with GTID slave_pos.'
                : 'Database replication was configured with GTID auto-positioning.',
            'blockers' => $status['blockers'],
            'replication_percent' => $status['replication_percent'],
            'host' => $this->publicHost($this->getHost(self::KIND_DATABASE, (int)$host['id'])),
        ];
    }

    private function ensureDatabaseReplicationUser(array $primary, string $replicationUser, string $replicationPassword): void
    {
        if (trim($replicationUser) === '' || trim($replicationPassword) === '') {
            throw new RuntimeException('Replication username and password are required.');
        }

        $connection = $this->databaseConnection($primary, true);
        try {
            $account = sprintf(
                "'%s'@'%%'",
                $connection->real_escape_string($replicationUser)
            );
            $password = $connection->real_escape_string($replicationPassword);

            $this->mysqliExec($connection, "CREATE USER IF NOT EXISTS " . $account . " IDENTIFIED BY '" . $password . "'");
            $this->mysqliExec($connection, "ALTER USER " . $account . " IDENTIFIED BY '" . $password . "'");
            $this->mysqliExec($connection, "GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO " . $account);
            $this->mysqliExec($connection, 'FLUSH PRIVILEGES');
        } catch (Throwable $throwable) {
            throw new RuntimeException(
                'Could not create or update the replication user on the primary database. Add primary admin credentials or create the replication user manually: ' . $throwable->getMessage(),
                0,
                $throwable
            );
        } finally {
            $connection->close();
        }
    }

    private function configureMySqlReplication(mysqli $target, array $primary, string $replicationUser, string $replicationPassword): void
    {
        try {
            $this->mysqliExec($target, 'STOP REPLICA');
        } catch (Throwable) {
        }
        $sql = sprintf(
            "CHANGE REPLICATION SOURCE TO SOURCE_HOST = '%s', SOURCE_PORT = %d, SOURCE_USER = '%s', SOURCE_PASSWORD = '%s', SOURCE_AUTO_POSITION = 1",
            $target->real_escape_string((string)$primary['host']),
            (int)$primary['port'],
            $target->real_escape_string($replicationUser),
            $target->real_escape_string($replicationPassword)
        );
        $this->mysqliExec($target, $sql);
        $this->mysqliExec($target, 'START REPLICA');
    }

    private function configureMariaDbReplication(mysqli $target, array $primary, array $host, string $replicationUser, string $replicationPassword): void
    {
        foreach (['STOP SLAVE', 'RESET SLAVE ALL'] as $statement) {
            try {
                $this->mysqliExec($target, $statement);
            } catch (Throwable) {
            }
        }
        $this->configureMariaDbReplicationFilters($target, $primary, $host);
        $sql = sprintf(
            "CHANGE MASTER TO MASTER_HOST = '%s', MASTER_PORT = %d, MASTER_USER = '%s', MASTER_PASSWORD = '%s', MASTER_USE_GTID = slave_pos",
            $target->real_escape_string((string)$primary['host']),
            (int)$primary['port'],
            $target->real_escape_string($replicationUser),
            $target->real_escape_string($replicationPassword)
        );
        $this->mysqliExec($target, $sql);
        $this->mysqliExec($target, 'START SLAVE');
    }

    private function configureMariaDbReplicationFilters(mysqli $target, array $primary, array $host): void
    {
        $existing = $this->mysqliSelectOne($target, "SHOW GLOBAL VARIABLES LIKE 'replicate_ignore_table'");
        $filters = array_values(array_filter(array_map(
            static fn(string $filter): string => trim($filter),
            explode(',', (string)($existing['Value'] ?? ''))
        )));

        foreach ([(string)($primary['database_name'] ?? ''), (string)($host['database_name'] ?? '')] as $database) {
            $database = trim($database);
            if ($database !== '') {
                foreach (self::MARIADB_SCHEMA_ONLY_TABLES as $tableName) {
                    $filters[] = $database . '.' . $tableName;
                }
            }
        }

        $filters = array_values(array_unique($filters));
        if ($filters === []) {
            return;
        }

        try {
            $this->mysqliExec($target, 'SET GLOBAL replicate_ignore_table = ' . self::sqlString($target, implode(',', $filters)));
        } catch (Throwable $throwable) {
            throw new RuntimeException(
                'Could not configure MariaDB replica schema-only table filters: ' . $throwable->getMessage(),
                0,
                $throwable
            );
        }
    }

    private function runMysqlClone(mysqli $target, array $primary, string $cloneUser, string $clonePassword): void
    {
        $donor = $target->real_escape_string((string)$primary['host'] . ':' . (int)$primary['port']);
        $this->mysqliExec($target, "SET GLOBAL clone_valid_donor_list = '" . $donor . "'");

        $sql = sprintf(
            "CLONE INSTANCE FROM '%s'@'%s':%d IDENTIFIED BY '%s'",
            $target->real_escape_string($cloneUser),
            $target->real_escape_string((string)$primary['host']),
            (int)$primary['port'],
            $target->real_escape_string($clonePassword)
        );

        try {
            $this->mysqliExec($target, $sql);
        } catch (Throwable $throwable) {
            $message = strtolower($throwable->getMessage());
            if (!str_contains($message, 'lost connection') && !str_contains($message, 'server has gone away')) {
                throw $throwable;
            }
        }
    }

    private function waitForDatabaseConnection(array $host, bool $admin, int $timeoutSeconds): mysqli
    {
        $deadline = time() + max(1, $timeoutSeconds);
        $lastError = null;

        do {
            try {
                return $this->databaseConnection($host, $admin);
            } catch (Throwable $throwable) {
                $lastError = $throwable;
                sleep(2);
            }
        } while (time() < $deadline);

        throw new RuntimeException('Database target did not reconnect after MySQL Clone: ' . ($lastError?->getMessage() ?? 'timeout'));
    }

    private function provisionRedisHost(array $host, int $operationId): array
    {
        $primary = $this->primaryHost(self::KIND_REDIS);
        if ($primary === null) {
            throw new RuntimeException('No Redis primary is registered.');
        }

        $targetStatus = $this->testRedisHost(array_merge($host, ['test_connectivity_only' => true]));
        $primaryStatus = $this->testRedisHost($primary);
        $blockers = array_merge($targetStatus['blockers'], $primaryStatus['blockers']);
        if ($blockers !== []) {
            $this->storeStatus($host, array_replace($targetStatus, ['blockers' => array_values(array_unique($blockers))]));
            return [
                'ok' => false,
                'message' => 'Redis replica provisioning is blocked.',
                'blockers' => array_values(array_unique($blockers)),
                'replication_percent' => $targetStatus['replication_percent'],
                'host' => $this->publicHost($host),
            ];
        }

        $context = $this->operationContext($operationId);
        if (($context['phase'] ?? '') !== 'configured') {
            try {
                $client = $this->redisClient($host);
                $primaryCredentials = $this->credentials($primary);
                if (($primaryCredentials['username'] ?? '') !== '' && ($primaryCredentials['username'] ?? '') !== 'default') {
                    $client->executeRaw(['CONFIG', 'SET', 'masteruser', (string)$primaryCredentials['username']]);
                }
                $client->executeRaw(['CONFIG', 'SET', 'masterauth', (string)($primaryCredentials['password'] ?? '')]);
                $client->executeRaw(['REPLICAOF', (string)$primary['host'], (string)$primary['port']]);
                $client->executeRaw(['CONFIG', 'REWRITE']);
            } catch (Throwable $throwable) {
                $blockers = [$throwable->getMessage()];
                $this->storeStatus($host, array_replace($targetStatus, [
                    'status' => 'degraded',
                    'replication_percent' => 0,
                    'blockers' => $blockers,
                ]));
                return [
                    'ok' => false,
                    'message' => 'Redis replica provisioning is blocked.',
                    'blockers' => $blockers,
                    'replication_percent' => 0,
                    'host' => $this->publicHost($host),
                ];
            }

            $context = [
                'phase' => 'configured',
                'configured_at' => date('c'),
                'primary_host' => (string)$primary['host'],
                'primary_port' => (int)$primary['port'],
            ];
            $this->updateOperationProgress(
                $operationId,
                5.0,
                'Redis replication was configured; waiting for the replica to catch up.',
                $context
            );
        }

        $this->execute(
            "UPDATE replication_hosts SET replication_source_id = ?, status = 'provisioning' WHERE id = ?",
            'ii',
            [(int)$primary['id'], (int)$host['id']]
        );

        $host = $this->getHost(self::KIND_REDIS, (int)$host['id']);
        $status = $this->testRedisHost($host);
        $this->storeStatus($host, $status);
        $progress = min(99.99, max(5.0, (float)$status['replication_percent']));
        $syncBlockers = array_values(array_intersect($status['blockers'], [
            'Redis host is not currently a replica.',
            'Redis replica link to primary is not up.',
        ]));
        $onlySyncBlockers = $status['blockers'] === []
            || ($syncBlockers !== [] && count($syncBlockers) === count($status['blockers']));

        if ($onlySyncBlockers && ((float)$status['replication_percent'] < 100.0 || $syncBlockers !== [])) {
            $message = $syncBlockers !== []
                ? 'Redis replica is waiting for the primary link.'
                : 'Redis replica is syncing from the primary.';
            $this->updateOperationProgress($operationId, $progress, $message, $context);
            return [
                'ok' => true,
                'message' => $message,
                'blockers' => $status['blockers'],
                'replication_percent' => $progress,
                'operation' => [
                    'id' => $operationId,
                    'status' => 'running',
                    'progress_percent' => $progress,
                    'message' => $message,
                ],
                'host' => $this->publicHost($host),
            ];
        }

        if ($status['blockers'] !== []) {
            return [
                'ok' => false,
                'message' => 'Redis replica provisioning is blocked.',
                'blockers' => $status['blockers'],
                'replication_percent' => $status['replication_percent'],
                'host' => $this->publicHost($host),
            ];
        }

        return [
            'ok' => true,
            'healthy' => $status['blockers'] === [],
            'message' => 'Redis replication was configured.',
            'blockers' => $status['blockers'],
            'replication_percent' => $status['replication_percent'],
            'host' => $this->publicHost($this->getHost(self::KIND_REDIS, (int)$host['id'])),
        ];
    }

    private function provisionMinioHost(array $host, int $operationId): array
    {
        $primary = $this->primaryHost(self::KIND_MINIO);
        if ($primary === null) {
            throw new RuntimeException('No MinIO primary is registered.');
        }

        $targetStatus = $this->testMinioHost(array_merge($host, ['test_connectivity_only' => true]));
        $primaryStatus = $this->testMinioHost($primary);
        $blockers = array_values(array_unique(array_merge($targetStatus['blockers'], $primaryStatus['blockers'])));
        if ($blockers !== []) {
            $this->storeStatus($host, array_replace($targetStatus, ['blockers' => $blockers]));
            return [
                'ok' => false,
                'message' => 'MinIO replica provisioning is blocked.',
                'blockers' => $blockers,
                'replication_percent' => $targetStatus['replication_percent'],
                'host' => $this->publicHost($host),
            ];
        }

        $context = $this->operationContext($operationId);
        if (($context['phase'] ?? '') !== 'configured') {
            try {
                $this->configureMinioReplication($primary, $host);
            } catch (Throwable $throwable) {
                $blockers = [$throwable->getMessage()];
                $this->storeStatus($host, array_replace($targetStatus, [
                    'status' => 'degraded',
                    'replication_percent' => 0,
                    'blockers' => $blockers,
                ]));
                return [
                    'ok' => false,
                    'message' => 'MinIO replica provisioning is blocked.',
                    'blockers' => $blockers,
                    'replication_percent' => 0,
                    'host' => $this->publicHost($host),
                ];
            }

            $context = [
                'phase' => 'configured',
                'configured_at' => date('c'),
                'primary_endpoint' => self::minioEndpoint($primary),
            ];
            $this->updateOperationProgress(
                $operationId,
                5.0,
                'MinIO bucket replication was configured; waiting for buckets to catch up.',
                $context
            );
        }

        $this->execute(
            "UPDATE replication_hosts SET replication_source_id = ?, status = 'provisioning' WHERE id = ?",
            'ii',
            [(int)$primary['id'], (int)$host['id']]
        );

        $host = $this->getHost(self::KIND_MINIO, (int)$host['id']);
        $status = $this->testMinioHost($host);
        $this->storeStatus($host, $status);
        $progress = min(99.99, max(5.0, (float)$status['replication_percent']));
        $syncBlockers = array_values(array_intersect($status['blockers'], [
            'MinIO replica has not caught up.',
            'MinIO bucket replication is not configured.',
        ]));
        $onlySyncBlockers = $status['blockers'] === []
            || ($syncBlockers !== [] && count($syncBlockers) === count($status['blockers']));

        if ($onlySyncBlockers && ((float)$status['replication_percent'] < 100.0 || $syncBlockers !== [])) {
            $message = 'MinIO replica is syncing from the primary.';
            $this->updateOperationProgress($operationId, $progress, $message, $context);
            return [
                'ok' => true,
                'message' => $message,
                'blockers' => $status['blockers'],
                'replication_percent' => $progress,
                'operation' => [
                    'id' => $operationId,
                    'status' => 'running',
                    'progress_percent' => $progress,
                    'message' => $message,
                ],
                'host' => $this->publicHost($host),
            ];
        }

        if ($status['blockers'] !== []) {
            return [
                'ok' => false,
                'message' => 'MinIO replica provisioning is blocked.',
                'blockers' => $status['blockers'],
                'replication_percent' => $status['replication_percent'],
                'host' => $this->publicHost($host),
            ];
        }

        return [
            'ok' => true,
            'healthy' => $status['blockers'] === [],
            'message' => 'MinIO bucket replication was configured.',
            'blockers' => $status['blockers'],
            'replication_percent' => $status['replication_percent'],
            'host' => $this->publicHost($this->getHost(self::KIND_MINIO, (int)$host['id'])),
        ];
    }

    private function promoteDatabaseHost(array $host): array
    {
        $status = $this->testDatabaseHost($host);
        if ($status['blockers'] !== [] || (float)$status['replication_percent'] < 100.0) {
            throw new RuntimeException('Database promotion blocked: ' . implode(' ', $status['blockers'] ?: ['Replica is not caught up.']));
        }

        $oldPrimary = $this->primaryHost(self::KIND_DATABASE);
        if ($oldPrimary === null) {
            throw new RuntimeException('No current database primary is registered.');
        }

        $oldPrimaryConn = null;
        $targetConn = null;
        $metadataSwitched = false;

        try {
            $oldPrimaryConn = $this->databaseConnection($oldPrimary, true);
            $oldPrimaryStatus = $this->databaseServerStatus($oldPrimaryConn);
            $this->setDatabaseReadOnly($oldPrimaryConn, $oldPrimaryStatus, true);

            $targetConn = $this->databaseConnection($host, true);
            $targetStatus = $this->databaseServerStatus($targetConn);
            $this->stopDatabaseReplication($targetConn, $targetStatus);
            $this->setDatabaseReadOnly($targetConn, $targetStatus, false);

            $this->switchPrimary(self::KIND_DATABASE, (int)$host['id'], (int)$oldPrimary['id']);
            $metadataSwitched = true;
            $this->writeBootstrapSnapshot();
        } catch (Throwable $throwable) {
            if (!$metadataSwitched && $oldPrimaryConn instanceof mysqli) {
                try {
                    $this->setDatabaseReadOnly($oldPrimaryConn, $oldPrimaryStatus ?? [], false);
                } catch (Throwable) {
                }
            }
            throw $throwable;
        } finally {
            if ($targetConn instanceof mysqli) {
                $targetConn->close();
            }
            if ($oldPrimaryConn instanceof mysqli) {
                $oldPrimaryConn->close();
            }
        }

        return [
            'ok' => true,
            'message' => 'Database replica promoted to primary.',
            'primary' => $this->publicHost($this->getHost(self::KIND_DATABASE, (int)$host['id'])),
            'prior_primary' => $this->publicHost($this->getHost(self::KIND_DATABASE, (int)$oldPrimary['id'], true)),
            'blockers' => [],
        ];
    }

    private function setDatabaseReadOnly(mysqli $connection, array $status, bool $readOnly): void
    {
        $value = $readOnly ? 'ON' : 'OFF';
        if (array_key_exists('super_read_only', $status)) {
            try {
                $this->mysqliExec($connection, 'SET GLOBAL super_read_only = ' . $value);
            } catch (Throwable) {
            }
        }
        $this->mysqliExec($connection, 'SET GLOBAL read_only = ' . $value);
    }

    private function stopDatabaseReplication(mysqli $connection, array $status): void
    {
        if (self::databaseEngine($status) === 'mariadb') {
            $this->mysqliExec($connection, 'STOP SLAVE');
            return;
        }

        $this->mysqliExec($connection, 'STOP REPLICA');
    }

    private function promoteRedisHost(array $host): array
    {
        $status = $this->testRedisHost($host);
        if ($status['blockers'] !== [] || (float)$status['replication_percent'] < 100.0) {
            throw new RuntimeException('Redis promotion blocked: ' . implode(' ', $status['blockers'] ?: ['Replica is not caught up.']));
        }

        $oldPrimary = $this->primaryHost(self::KIND_REDIS);
        if ($oldPrimary === null) {
            throw new RuntimeException('No current Redis primary is registered.');
        }

        $client = $this->redisClient($host);
        $client->executeRaw(['REPLICAOF', 'NO', 'ONE']);
        try {
            $client->executeRaw(['CONFIG', 'REWRITE']);
        } catch (Throwable) {
        }

        $this->switchPrimary(self::KIND_REDIS, (int)$host['id'], (int)$oldPrimary['id']);
        $this->writeBootstrapSnapshot();

        return [
            'ok' => true,
            'message' => 'Redis replica promoted to primary.',
            'primary' => $this->publicHost($this->getHost(self::KIND_REDIS, (int)$host['id'])),
            'prior_primary' => $this->publicHost($this->getHost(self::KIND_REDIS, (int)$oldPrimary['id'], true)),
            'blockers' => [],
        ];
    }

    private function promoteMinioHost(array $host): array
    {
        $status = $this->testMinioHost($host);
        if ($status['blockers'] !== [] || (float)$status['replication_percent'] < 100.0) {
            throw new RuntimeException('MinIO promotion blocked: ' . implode(' ', $status['blockers'] ?: ['Replica is not caught up.']));
        }

        $oldPrimary = $this->primaryHost(self::KIND_MINIO);
        if ($oldPrimary === null) {
            throw new RuntimeException('No current MinIO primary is registered.');
        }

        $this->switchPrimary(self::KIND_MINIO, (int)$host['id'], (int)$oldPrimary['id']);
        $this->writeBootstrapSnapshot();

        return [
            'ok' => true,
            'message' => 'MinIO replica promoted to primary.',
            'primary' => $this->publicHost($this->getHost(self::KIND_MINIO, (int)$host['id'])),
            'prior_primary' => $this->publicHost($this->getHost(self::KIND_MINIO, (int)$oldPrimary['id'], true)),
            'blockers' => [],
        ];
    }

    private function testDatabaseHost(array $host): array
    {
        $blockers = [];
        $raw = [];
        $percent = (float)(($host['role'] ?? '') === 'primary' ? 100 : 0);
        $lagSeconds = null;
        $reachable = true;

        try {
            $connection = $this->databaseConnection($host, true, !empty($host['connect_without_database']));
            try {
                $raw = $this->databaseServerStatus($connection);
                $blockers = array_merge($blockers, self::databasePrerequisiteBlockers($raw));

                if (($host['role'] ?? '') !== 'primary' && empty($host['test_connectivity_only'])) {
                    $primary = $this->primaryHost(self::KIND_DATABASE);
                    if ($primary === null) {
                        $blockers[] = 'No database primary is registered.';
                    } else {
                        $sourceConnection = $this->databaseConnection($primary, true);
                        try {
                            $source = $this->databaseServerStatus($sourceConnection);
                            $replica = $this->showReplicaStatus($connection);
                            $sourceEngine = self::databaseEngine($source);
                            $replicaEngine = self::databaseEngine($raw);
                            $raw['source_gtid_executed'] = self::databaseGtidPosition($source);
                            $raw['replica_status'] = $replica;

                            if ($sourceEngine !== $replicaEngine) {
                                $blockers[] = 'Database primary and replica must use the same engine family.';
                            }
                            $percent = $sourceEngine === 'mariadb'
                                ? self::mariadbGtidCoveragePercent(
                                    self::databaseGtidPosition($source),
                                    (string)($replica['Gtid_IO_Pos'] ?? $raw['gtid_slave_pos'] ?? $raw['gtid_current_pos'] ?? '')
                                )
                                : self::mysqlGtidCoveragePercent(
                                    (string)($source['gtid_executed'] ?? ''),
                                    (string)($replica['Executed_Gtid_Set'] ?? $raw['gtid_executed'] ?? '')
                                );
                            $lagSeconds = isset($replica['Seconds_Behind_Source'])
                                ? (int)$replica['Seconds_Behind_Source']
                                : (isset($replica['Seconds_Behind_Master']) ? (int)$replica['Seconds_Behind_Master'] : null);

                            $ioRunning = false;
                            $sqlRunning = false;
                            if ($replica === []) {
                                $blockers[] = 'Database replica status is not configured.';
                            } else {
                                $ioRunning = strtoupper((string)($replica['Replica_IO_Running'] ?? $replica['Slave_IO_Running'] ?? '')) === 'YES';
                                $sqlRunning = strtoupper((string)($replica['Replica_SQL_Running'] ?? $replica['Slave_SQL_Running'] ?? '')) === 'YES';
                                if (!$ioRunning || !$sqlRunning) {
                                    $blockers[] = 'Database replication IO and SQL threads must both be running.';
                                }
                                if (!$ioRunning) {
                                    $blockers[] = 'Database replication IO thread is not running.';
                                }
                                if (!$sqlRunning) {
                                    $blockers[] = 'Database replication SQL thread is not running.';
                                }
                                if ($sourceEngine === 'mariadb' && isset($replica['Using_Gtid']) && strtoupper((string)$replica['Using_Gtid']) === 'NO') {
                                    $blockers[] = 'MariaDB replication must use GTID mode.';
                                }
                                foreach (['Last_IO_Error', 'Last_SQL_Error', 'Last_Error'] as $errorKey) {
                                    $error = trim((string)($replica[$errorKey] ?? ''));
                                    if ($error !== '') {
                                        $blockers[] = $error;
                                    }
                                }
                            }
                            if (($blockers !== [] || !$ioRunning || !$sqlRunning) && $percent >= 100.0) {
                                $percent = 99.99;
                            }
                        } finally {
                            $sourceConnection->close();
                        }
                    }
                }
            } finally {
                $connection->close();
            }
        } catch (Throwable $throwable) {
            $reachable = false;
            $blockers[] = $throwable->getMessage();
        }

        $blockers = array_values(array_unique(array_filter($blockers)));
        return [
            'status' => !$reachable ? 'down' : ($blockers === [] ? 'ok' : 'degraded'),
            'replication_percent' => round($percent, 2),
            'lag_seconds' => $lagSeconds,
            'blockers' => $blockers,
            'raw' => $raw,
            'checked_at' => date('c'),
        ];
    }

    private function testRedisHost(array $host): array
    {
        $blockers = [];
        $raw = [];
        $percent = (float)(($host['role'] ?? '') === 'primary' ? 100 : 0);
        $reachable = true;

        try {
            $client = $this->redisClient($host);
            $ping = (string)$client->ping();
            if (stripos($ping, 'PONG') === false && stripos($ping, 'OK') === false) {
                $blockers[] = 'Redis PING did not return PONG.';
            }

            $role = $client->executeRaw(['ROLE']);
            $info = $this->redisInfo($client);
            $raw = [
                'role' => $role,
                'replication' => $info,
            ];

            try {
                $client->executeRaw(['CONFIG', 'GET', 'appendonly']);
            } catch (Throwable $throwable) {
                $blockers[] = 'Redis ACL must allow CONFIG GET/SET/REWRITE for durable replication changes.';
            }

            if (($host['role'] ?? '') !== 'primary' && empty($host['test_connectivity_only'])) {
                $primary = $this->primaryHost(self::KIND_REDIS);
                if ($primary === null) {
                    $blockers[] = 'No Redis primary is registered.';
                } else {
                    $primaryClient = $this->redisClient($primary);
                    $primaryInfo = $this->redisInfo($primaryClient);
                    $primaryOffset = (int)($primaryInfo['master_repl_offset'] ?? 0);
                    $replicaOffset = (int)($info['slave_repl_offset'] ?? $info['master_repl_offset'] ?? 0);
                    $percent = self::redisOffsetPercent($primaryOffset, $replicaOffset);
                    $raw['primary_replication'] = $primaryInfo;

                    if (!in_array(strtolower((string)($info['role'] ?? '')), ['slave', 'replica'], true)) {
                        $blockers[] = 'Redis host is not currently a replica.';
                    }
                    if (strtolower((string)($info['master_link_status'] ?? '')) !== 'up') {
                        $blockers[] = 'Redis replica link to primary is not up.';
                    }
                    if ($blockers !== [] && $percent >= 100.0) {
                        $percent = 99.99;
                    }
                }
            }
        } catch (Throwable $throwable) {
            $reachable = false;
            $blockers[] = $throwable->getMessage();
        }

        $blockers = array_values(array_unique(array_filter($blockers)));
        return [
            'status' => !$reachable ? 'down' : ($blockers === [] ? 'ok' : 'degraded'),
            'replication_percent' => round($percent, 2),
            'lag_seconds' => null,
            'blockers' => $blockers,
            'raw' => $raw,
            'checked_at' => date('c'),
        ];
    }

    private function testMinioHost(array $host): array
    {
        $blockers = [];
        $raw = [];
        $percent = (float)(($host['role'] ?? '') === 'primary' ? 100 : 0);
        $reachable = true;
        $options = $this->decodeOptions($host);
        $buckets = self::normalizeMinioBuckets($options['buckets'] ?? self::MINIO_DEFAULT_BUCKETS);

        try {
            $client = $this->minioS3Client($host);
            $client->listBuckets();

            if (($host['role'] ?? '') === 'primary') {
                $sourceStats = $this->minioBucketStats($client, $buckets, true, 'MinIO source bucket');
                $blockers = array_merge($blockers, $sourceStats['blockers']);
                $raw['buckets'] = $sourceStats['buckets'];
                $raw['storage'] = [
                    'source_bytes' => $sourceStats['bytes'],
                    'source_objects' => $sourceStats['objects'],
                ];
            } else {
                $targetStats = $this->minioBucketStats($client, $buckets, empty($host['test_connectivity_only']), 'MinIO target bucket');
                $blockers = array_merge($blockers, $targetStats['blockers']);
                $raw['target_buckets'] = $targetStats['buckets'];
                $raw['storage'] = [
                    'target_bytes' => $targetStats['bytes'],
                    'target_objects' => $targetStats['objects'],
                ];

                $primary = $this->primaryHost(self::KIND_MINIO);
                if ($primary === null) {
                    $blockers[] = 'No MinIO primary is registered.';
                } else {
                    $sourceStats = $this->minioBucketStats($this->minioS3Client($primary), $buckets, true, 'MinIO source bucket');
                    $blockers = array_merge($blockers, $sourceStats['blockers']);
                    $headroom = (float)($options['space_headroom_percent'] ?? self::MINIO_SPACE_HEADROOM_PERCENT);
                    $requiredBytes = self::minioRequiredFreeBytes((int)$sourceStats['bytes'], $headroom);
                    $availableBytes = $this->minioTargetFreeBytes($host);
                    $spaceBlockers = self::minioSpaceBlockers($availableBytes, $requiredBytes);
                    $blockers = array_merge($blockers, $spaceBlockers);
                    $percent = self::minioByteReplicationPercent((int)$sourceStats['bytes'], (int)$targetStats['bytes']);
                    if (empty($host['test_connectivity_only'])) {
                        $replicationConfigured = $this->minioReplicationConfigured($primary, $buckets);
                        if (!$replicationConfigured) {
                            $blockers[] = 'MinIO bucket replication is not configured.';
                        } elseif ($percent < 100.0) {
                            $blockers[] = 'MinIO replica has not caught up.';
                        }
                    }
                    $raw['source_buckets'] = $sourceStats['buckets'];
                    $raw['storage'] = array_merge($raw['storage'], [
                        'source_bytes' => $sourceStats['bytes'],
                        'source_objects' => $sourceStats['objects'],
                        'required_free_bytes' => $requiredBytes,
                        'available_free_bytes' => $availableBytes,
                        'space_headroom_percent' => $headroom,
                        'space_ok' => $spaceBlockers === [],
                    ]);
                }
            }
        } catch (Throwable $throwable) {
            $reachable = false;
            $blockers[] = $throwable->getMessage();
        }

        $blockers = array_values(array_unique(array_filter($blockers)));
        return [
            'status' => !$reachable ? 'down' : ($blockers === [] ? 'ok' : 'degraded'),
            'replication_percent' => round($percent, 2),
            'lag_seconds' => null,
            'blockers' => $blockers,
            'raw' => $raw,
            'checked_at' => date('c'),
        ];
    }

    public static function databasePrerequisiteBlockers(array $status): array
    {
        if (self::databaseEngine($status) === 'mariadb') {
            return self::mariaDbPrerequisiteBlockers($status);
        }

        $blockers = [];
        if (strtoupper((string)($status['gtid_mode'] ?? '')) !== 'ON') {
            $blockers[] = isset($status['gtid_mode'])
                ? 'MySQL GTID mode must be ON.'
                : 'MySQL GTID mode is unavailable. Managed replication requires Oracle MySQL 8.x with GTID enabled.';
        }
        if (!self::mysqlBooleanEnabled($status['log_bin'] ?? null)) {
            $blockers[] = isset($status['log_bin'])
                ? 'MySQL binary logging must be enabled.'
                : 'MySQL binary logging status is unavailable.';
        }
        if ((int)($status['server_id'] ?? 0) <= 0) {
            $blockers[] = 'MySQL server_id must be configured.';
        }
        if (trim((string)($status['server_uuid'] ?? '')) === '') {
            $blockers[] = 'MySQL server_uuid must be available. Managed replication requires Oracle MySQL 8.x.';
        }
        $serverVersion = (string)($status['server_version'] ?? '');
        if (!str_starts_with($serverVersion, '8.') || stripos($serverVersion, 'mariadb') !== false) {
            $blockers[] = $serverVersion !== ''
                ? 'Oracle MySQL 8.x is required for managed replication. Current server reports ' . $serverVersion . '.'
                : 'Oracle MySQL 8.x is required for managed replication.';
        }

        return $blockers;
    }

    public static function missingDatabaseTables(array $sourceTables, array $replicaTables): array
    {
        $source = array_values(array_unique(array_filter(array_map(
            static fn(mixed $table): string => trim((string)$table),
            $sourceTables
        ))));
        $replicaLookup = array_flip(array_values(array_unique(array_filter(array_map(
            static fn(mixed $table): string => trim((string)$table),
            $replicaTables
        )))));

        return array_values(array_filter(
            $source,
            static fn(string $table): bool => !isset($replicaLookup[$table])
        ));
    }

    private static function mariaDbPrerequisiteBlockers(array $status): array
    {
        $blockers = [];
        if (!self::mysqlBooleanEnabled($status['log_bin'] ?? null)) {
            $blockers[] = isset($status['log_bin'])
                ? 'MariaDB binary logging must be enabled.'
                : 'MariaDB binary logging status is unavailable.';
        }
        if ((int)($status['server_id'] ?? 0) <= 0) {
            $blockers[] = 'MariaDB server_id must be configured.';
        }
        if (!self::mariaDbGtidPositionAvailable($status)) {
            $blockers[] = 'MariaDB GTID position must be available.';
        }

        $serverVersion = (string)($status['server_version'] ?? '');
        if (!self::mariaDbVersionSupported($serverVersion)) {
            $blockers[] = $serverVersion !== ''
                ? 'MariaDB 10.6 or newer is required for managed replication. Current server reports ' . $serverVersion . '.'
                : 'MariaDB 10.6 or newer is required for managed replication.';
        }

        return $blockers;
    }

    private static function databaseEngine(array $status): string
    {
        return stripos((string)($status['server_version'] ?? ''), 'mariadb') !== false ? 'mariadb' : 'mysql';
    }

    private static function databaseGtidPosition(array $status): string
    {
        if (self::databaseEngine($status) === 'mariadb') {
            return trim((string)($status['gtid_binlog_pos'] ?? $status['gtid_current_pos'] ?? $status['gtid_slave_pos'] ?? ''));
        }

        return trim((string)($status['gtid_executed'] ?? ''));
    }

    private static function mariaDbGtidPositionAvailable(array $status): bool
    {
        foreach (['gtid_binlog_pos', 'gtid_current_pos', 'gtid_slave_pos'] as $key) {
            if (array_key_exists($key, $status) && $status[$key] !== null) {
                return true;
            }
        }

        return false;
    }

    private static function mariaDbVersionSupported(string $serverVersion): bool
    {
        if (!preg_match('/(\d+)\.(\d+)/', $serverVersion, $matches)) {
            return false;
        }

        $major = (int)$matches[1];
        $minor = (int)$matches[2];
        return $major > 10 || ($major === 10 && $minor >= 6);
    }

    private static function mysqlBooleanEnabled(mixed $value): bool
    {
        $normalized = strtoupper(trim((string)$value));
        return in_array($normalized, ['1', 'ON', 'YES', 'TRUE'], true);
    }

    private function databaseServerStatus(mysqli $connection): array
    {
        $row = $this->mysqliSelectOne($connection, "SELECT VERSION() AS server_version");
        $variables = $connection->query(
            "SHOW GLOBAL VARIABLES WHERE Variable_name IN (
                'gtid_mode',
                'log_bin',
                'server_id',
                'server_uuid',
                'read_only',
                'super_read_only',
                'gtid_executed',
                'gtid_binlog_pos',
                'gtid_current_pos',
                'gtid_slave_pos',
                'gtid_strict_mode'
            )"
        );
        if ($variables !== false) {
            while ($variable = $variables->fetch_assoc()) {
                $name = strtolower((string)($variable['Variable_name'] ?? ''));
                if ($name !== '') {
                    $row[$name] = $variable['Value'] ?? null;
                }
            }
        }

        $plugin = $this->mysqliSelectOne(
            $connection,
            "SELECT PLUGIN_STATUS AS plugin_status FROM information_schema.PLUGINS WHERE PLUGIN_NAME = 'clone' LIMIT 1"
        );
        $row['clone_plugin_active'] = strtoupper((string)($plugin['plugin_status'] ?? '')) === 'ACTIVE';

        return $row;
    }

    private function showReplicaStatus(mysqli $connection): array
    {
        try {
            $status = $this->mysqliSelectOne($connection, 'SHOW REPLICA STATUS');
            if ($status !== []) {
                return $status;
            }
        } catch (Throwable) {
        }

        return $this->mysqliSelectOne($connection, 'SHOW SLAVE STATUS');
    }

    private function databaseConnection(array $host, bool $admin = false, bool $connectWithoutDatabase = false): mysqli
    {
        $credentials = $this->credentials($host);
        $username = $admin && $credentials['admin_username'] !== ''
            ? $credentials['admin_username']
            : $credentials['username'];
        $password = $admin && $credentials['admin_password'] !== ''
            ? $credentials['admin_password']
            : $credentials['password'];

        mysqli_report(MYSQLI_REPORT_ERROR | MYSQLI_REPORT_STRICT);
        $connection = new mysqli(
            (string)$host['host'],
            $username,
            $password,
            $connectWithoutDatabase ? '' : (string)($host['database_name'] ?? ''),
            (int)$host['port']
        );
        $connection->set_charset('utf8mb4');
        return $connection;
    }

    private function databaseReplicaSeedBlockers(?array $primary, array $host, mysqli $target): array
    {
        if ($primary === null) {
            return ['No database primary is registered.'];
        }

        $primaryDatabase = trim((string)($primary['database_name'] ?? ''));
        $targetDatabase = trim((string)($host['database_name'] ?? ''));
        if ($primaryDatabase === '' || $targetDatabase === '') {
            return [];
        }

        $primaryConnection = $this->databaseConnection($primary, true);
        try {
            $primaryTables = $this->databaseTableNames($primaryConnection, $primaryDatabase);
            $targetTables = $this->databaseTableNames($target, $targetDatabase);
        } finally {
            $primaryConnection->close();
        }

        if ($primaryTables === []) {
            return [];
        }

        $missingTables = self::missingDatabaseTables($primaryTables, $targetTables);
        if ($missingTables === []) {
            $schemaOnlyTablesWithRows = $this->databaseSchemaOnlyTablesWithRows($target, $targetDatabase);
            if ($schemaOnlyTablesWithRows === []) {
                return [];
            }

            return [self::schemaOnlyTablesContainRowsBlocker($targetDatabase, $schemaOnlyTablesWithRows)];
        }

        return [self::missingDatabaseTablesBlocker($targetDatabase, $missingTables)];
    }

    private function advanceMariaDbReplicaSeed(int $operationId, array $primary, array $host, mysqli $target): array
    {
        $owner = 'replication-seed:' . $operationId;
        $context = $this->operationContext($operationId);
        $freezeState = application_write_freeze::state();
        if (($context['phase'] ?? '') !== '' && ($freezeState['owner'] ?? null) !== $owner) {
            $context = [];
        }
        if (self::mariaDbSeedContextRequiresFilterReset($context)) {
            $context = [];
        }
        application_write_freeze::freeze('MariaDB replica seed is copying data.', $owner, self::MARIADB_SEED_FREEZE_TTL_SECONDS);

        $source = $this->databaseConnection($primary, true);
        $deadline = microtime(true) + self::MARIADB_SEED_STEP_SECONDS;

        try {
            $this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 0');
            $this->mysqliExec($target, 'SET SESSION UNIQUE_CHECKS = 0');

            if (($context['phase'] ?? '') === '') {
                $context = $this->initializeMariaDbSeedContext($source, $target, $primary, $host);
            }

            while (microtime(true) < $deadline && ($context['phase'] ?? '') !== 'complete') {
                if (($context['phase'] ?? '') === 'schema') {
                    $context = $this->advanceMariaDbSeedSchema($source, $target, $context);
                    continue;
                }

                if (($context['phase'] ?? '') === 'copy') {
                    $context = $this->advanceMariaDbSeedRows($source, $target, $context);
                    continue;
                }

                break;
            }

            $progress = self::mariaDbSeedProgress($context);
            $message = self::mariaDbSeedMessage($context);
            $this->updateOperationProgress($operationId, $progress, $message, $context);

            if (($context['phase'] ?? '') === 'complete') {
                application_write_freeze::unfreeze($owner);
                return [
                    'running' => false,
                    'message' => 'MariaDB replica seed completed.',
                    'status' => [
                        'status' => 'provisioning',
                        'replication_percent' => 90.0,
                        'lag_seconds' => null,
                        'blockers' => [],
                        'raw' => ['seed' => $context],
                        'checked_at' => date('c'),
                    ],
                ];
            }

            return [
                'running' => true,
                'message' => $message,
                'status' => [
                    'status' => 'provisioning',
                    'replication_percent' => $progress,
                    'lag_seconds' => null,
                    'blockers' => ['MariaDB replica seed is running.'],
                    'raw' => ['seed' => $context],
                    'checked_at' => date('c'),
                ],
            ];
        } catch (Throwable $throwable) {
            application_write_freeze::unfreeze($owner);
            throw $throwable;
        } finally {
            $source->close();
        }
    }

    private function initializeMariaDbSeedContext(mysqli $source, mysqli $target, array $primary, array $host): array
    {
        $primaryDatabase = trim((string)($primary['database_name'] ?? ''));
        $targetDatabase = trim((string)($host['database_name'] ?? ''));
        if ($primaryDatabase === '' || $targetDatabase === '') {
            throw new RuntimeException('Database names are required before a MariaDB replica can be seeded.');
        }

        $sourceGtid = $this->mariaDbCurrentGtid($source);
        $this->prepareMariaDbReplicaTarget($target, $source, $primaryDatabase, $targetDatabase);

        $tables = [];
        foreach ($this->databaseTableNames($source, $primaryDatabase) as $tableName) {
            $skipData = self::databaseReplicaSeedSkipsTableData($tableName);
            $tables[] = [
                'name' => $tableName,
                'rows' => $skipData ? 0 : $this->estimatedDatabaseTableRows($source, $primaryDatabase, $tableName),
                'copied' => 0,
                'schema_created' => false,
                'skip_data' => $skipData,
                'skip_reason' => $skipData ? 'excluded from managed replication' : '',
            ];
        }

        return [
            'phase' => 'schema',
            'source_database' => $primaryDatabase,
            'target_database' => $targetDatabase,
            'source_gtid' => $sourceGtid,
            'schema_index' => 0,
            'copy_index' => 0,
            'tables' => $tables,
            'started_at' => date('c'),
            'updated_at' => date('c'),
        ];
    }

    private function advanceMariaDbSeedSchema(mysqli $source, mysqli $target, array $context): array
    {
        $tables = $context['tables'] ?? [];
        $index = (int)($context['schema_index'] ?? 0);
        if (!isset($tables[$index])) {
            $context['phase'] = 'copy';
            $context['copy_index'] = 0;
            $context['updated_at'] = date('c');
            return $context;
        }

        $table = $tables[$index];
        $this->createMariaDbReplicaTable(
            $source,
            $target,
            (string)$context['source_database'],
            (string)$context['target_database'],
            (string)$table['name']
        );

        $context['tables'][$index]['schema_created'] = true;
        $context['schema_index'] = $index + 1;
        $context['updated_at'] = date('c');
        return $context;
    }

    private function advanceMariaDbSeedRows(mysqli $source, mysqli $target, array $context): array
    {
        $tables = $context['tables'] ?? [];
        $index = (int)($context['copy_index'] ?? 0);
        if (!isset($tables[$index])) {
            if (trim((string)($context['source_gtid'] ?? '')) !== '') {
                $this->mysqliExec($target, 'SET GLOBAL gtid_slave_pos = ' . self::sqlString($target, (string)$context['source_gtid']));
            }
            $this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 1');
            $this->mysqliExec($target, 'SET SESSION UNIQUE_CHECKS = 1');
            $context['phase'] = 'complete';
            $context['updated_at'] = date('c');
            $context['completed_at'] = date('c');
            return $context;
        }

        $tableName = (string)($tables[$index]['name'] ?? '');
        $copied = (int)($tables[$index]['copied'] ?? 0);
        if (!empty($tables[$index]['skip_data'])) {
            $context['tables'][$index]['copied'] = (int)($tables[$index]['rows'] ?? 0);
            $context['copy_index'] = $index + 1;
            $context['updated_at'] = date('c');
            return $context;
        }

        $copiedNow = $this->copyMariaDbReplicaTableRowsChunk(
            $source,
            $target,
            (string)$context['source_database'],
            (string)$context['target_database'],
            $tableName,
            $copied,
            self::MARIADB_SEED_BATCH_ROWS
        );

        $context['tables'][$index]['copied'] = $copied + $copiedNow;
        if ($copiedNow < self::MARIADB_SEED_BATCH_ROWS) {
            $context['copy_index'] = $index + 1;
        }
        $context['updated_at'] = date('c');
        return $context;
    }

    private static function mariaDbSeedProgress(array $context): float
    {
        $tables = is_array($context['tables'] ?? null) ? $context['tables'] : [];
        if (($context['phase'] ?? '') === 'complete') {
            return 90.0;
        }
        if ($tables === []) {
            return 5.0;
        }

        $schemaCount = count($tables);
        $schemaDone = min($schemaCount, (int)($context['schema_index'] ?? 0));
        $schemaProgress = $schemaCount > 0 ? ($schemaDone / $schemaCount) * 20.0 : 20.0;

        $totalRows = 0;
        $copiedRows = 0;
        foreach ($tables as $table) {
            if (!empty($table['skip_data'])) {
                continue;
            }
            $rows = max(1, (int)($table['rows'] ?? 0));
            $totalRows += $rows;
            $copiedRows += min($rows, (int)($table['copied'] ?? 0));
        }
        $copyProgress = $totalRows > 0 ? ($copiedRows / $totalRows) * 65.0 : 0.0;

        return round(min(89.0, 5.0 + $schemaProgress + $copyProgress), 2);
    }

    private static function mariaDbSeedMessage(array $context): string
    {
        $tables = is_array($context['tables'] ?? null) ? $context['tables'] : [];
        if (($context['phase'] ?? '') === 'schema') {
            return 'Creating replica schema ' . min(count($tables), (int)($context['schema_index'] ?? 0)) . ' of ' . count($tables) . '.';
        }
        if (($context['phase'] ?? '') === 'copy') {
            $index = (int)($context['copy_index'] ?? 0);
            $table = $tables[$index]['name'] ?? 'table data';
            if (!empty($tables[$index]['skip_data'])) {
                return 'Skipping replica data for ' . $table . '.';
            }
            return 'Copying replica data for ' . $table . '.';
        }
        if (($context['phase'] ?? '') === 'complete') {
            return 'Replica seed completed.';
        }
        return 'Preparing replica seed.';
    }

    private function seedMariaDbReplicaFromPrimary(array $primary, array $host, mysqli $target): void
    {
        $primaryDatabase = trim((string)($primary['database_name'] ?? ''));
        $targetDatabase = trim((string)($host['database_name'] ?? ''));
        if ($primaryDatabase === '' || $targetDatabase === '') {
            throw new RuntimeException('Database names are required before a MariaDB replica can be seeded.');
        }

        $source = $this->databaseConnection($primary, true);
        $readLockAcquired = false;
        $transactionStarted = false;

        try {
            $source->query('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
            $source->query('FLUSH TABLES WITH READ LOCK');
            $readLockAcquired = true;
            $source->query('START TRANSACTION WITH CONSISTENT SNAPSHOT');
            $transactionStarted = true;
            $sourceGtid = $this->mariaDbCurrentGtid($source);
            $source->query('UNLOCK TABLES');
            $readLockAcquired = false;

            $this->prepareMariaDbReplicaTarget($target, $source, $primaryDatabase, $targetDatabase);
            foreach ($this->databaseTableNames($source, $primaryDatabase) as $tableName) {
                $this->createMariaDbReplicaTable($source, $target, $primaryDatabase, $targetDatabase, $tableName);
                $this->copyMariaDbReplicaTableRows($source, $target, $primaryDatabase, $targetDatabase, $tableName);
            }

            if ($sourceGtid !== '') {
                $this->mysqliExec($target, 'SET GLOBAL gtid_slave_pos = ' . self::sqlString($target, $sourceGtid));
            }
            $this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 1');

            $source->query('COMMIT');
            $transactionStarted = false;
        } catch (Throwable $throwable) {
            if ($readLockAcquired) {
                try {
                    $source->query('UNLOCK TABLES');
                } catch (Throwable) {
                }
            }
            if ($transactionStarted) {
                try {
                    $source->query('ROLLBACK');
                } catch (Throwable) {
                }
            }
            throw new RuntimeException('MariaDB replica seed failed: ' . $throwable->getMessage(), 0, $throwable);
        } finally {
            $source->close();
        }
    }

    private function mariaDbCurrentGtid(mysqli $source): string
    {
        foreach (['gtid_binlog_pos', 'gtid_current_pos'] as $variable) {
            $row = $this->mysqliSelectOne($source, "SELECT @@GLOBAL.$variable AS value");
            $value = trim((string)($row['value'] ?? ''));
            if ($value !== '') {
                return $value;
            }
        }

        return '';
    }

    private function prepareMariaDbReplicaTarget(mysqli $target, mysqli $source, string $primaryDatabase, string $targetDatabase): void
    {
        foreach (['STOP SLAVE', 'RESET SLAVE ALL'] as $statement) {
            try {
                $this->mysqliExec($target, $statement);
            } catch (Throwable) {
            }
        }

        $this->mysqliExec($target, 'SET SESSION FOREIGN_KEY_CHECKS = 0');
        $this->mysqliExec($target, 'DROP DATABASE IF EXISTS ' . self::quoteIdentifier($targetDatabase));
        $this->mysqliExec($target, $this->createDatabaseSql($source, $primaryDatabase, $targetDatabase));
        $this->mysqliExec($target, 'USE ' . self::quoteIdentifier($targetDatabase));

        try {
            $this->mysqliExec($target, 'RESET MASTER');
        } catch (Throwable) {
        }
        try {
            $this->mysqliExec($target, "SET GLOBAL gtid_slave_pos = ''");
        } catch (Throwable) {
        }
    }

    private function createDatabaseSql(mysqli $source, string $primaryDatabase, string $targetDatabase): string
    {
        $stmt = $source->prepare(
            'SELECT DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME
             FROM information_schema.SCHEMATA
             WHERE SCHEMA_NAME = ?
             LIMIT 1'
        );
        if ($stmt === false) {
            throw new RuntimeException('Could not prepare database schema lookup.');
        }

        $stmt->bind_param('s', $primaryDatabase);
        $stmt->execute();
        $result = $stmt->get_result();
        $row = $result ? $result->fetch_assoc() : null;
        $stmt->close();

        $charset = preg_replace('/[^a-zA-Z0-9_]/', '', (string)($row['DEFAULT_CHARACTER_SET_NAME'] ?? 'utf8mb4')) ?: 'utf8mb4';
        $collation = preg_replace('/[^a-zA-Z0-9_]/', '', (string)($row['DEFAULT_COLLATION_NAME'] ?? 'utf8mb4_unicode_ci')) ?: 'utf8mb4_unicode_ci';

        return 'CREATE DATABASE ' . self::quoteIdentifier($targetDatabase)
            . ' CHARACTER SET ' . $charset
            . ' COLLATE ' . $collation;
    }

    private function createMariaDbReplicaTable(mysqli $source, mysqli $target, string $primaryDatabase, string $targetDatabase, string $tableName): void
    {
        $create = $this->mysqliSelectOne(
            $source,
            'SHOW CREATE TABLE ' . self::quoteIdentifier($primaryDatabase) . '.' . self::quoteIdentifier($tableName)
        );
        $createSql = (string)($create['Create Table'] ?? '');
        if ($createSql === '') {
            throw new RuntimeException('Could not read CREATE TABLE for ' . $primaryDatabase . '.' . $tableName . '.');
        }

        $this->mysqliExec($target, 'USE ' . self::quoteIdentifier($targetDatabase));
        $this->mysqliExec($target, $createSql);
    }

    private static function mariaDbSeedContextRequiresFilterReset(array $context): bool
    {
        if (($context['phase'] ?? '') === '' || ($context['phase'] ?? '') === 'complete') {
            return false;
        }

        foreach (($context['tables'] ?? []) as $table) {
            if (self::databaseReplicaSeedSkipsTableData((string)($table['name'] ?? ''))
                && empty($table['skip_data'])) {
                return true;
            }
        }

        return false;
    }

    private function copyMariaDbReplicaTableRows(mysqli $source, mysqli $target, string $primaryDatabase, string $targetDatabase, string $tableName): void
    {
        if (self::databaseReplicaSeedSkipsTableData($tableName)) {
            return;
        }

        $offset = 0;
        do {
            $copied = $this->copyMariaDbReplicaTableRowsChunk(
                $source,
                $target,
                $primaryDatabase,
                $targetDatabase,
                $tableName,
                $offset,
                self::MARIADB_SEED_BATCH_ROWS
            );
            $offset += $copied;
        } while ($copied >= self::MARIADB_SEED_BATCH_ROWS);
    }

    private function copyMariaDbReplicaTableRowsChunk(
        mysqli $source,
        mysqli $target,
        string $primaryDatabase,
        string $targetDatabase,
        string $tableName,
        int $offset,
        int $limit
    ): int {
        $columnNames = $this->databaseWritableColumnNames($source, $primaryDatabase, $tableName);
        if ($columnNames === []) {
            return 0;
        }

        $quotedColumns = array_map(static fn(string $column): string => self::quoteIdentifier($column), $columnNames);
        $primaryKeyColumns = $this->databasePrimaryKeyColumnNames($source, $primaryDatabase, $tableName);
        $orderSql = $primaryKeyColumns !== []
            ? ' ORDER BY ' . implode(', ', array_map(static fn(string $column): string => self::quoteIdentifier($column), $primaryKeyColumns))
            : '';
        $result = $source->query(
            'SELECT ' . implode(', ', $quotedColumns)
                . ' FROM ' . self::quoteIdentifier($primaryDatabase) . '.' . self::quoteIdentifier($tableName)
                . $orderSql
                . ' LIMIT ' . max(0, $offset) . ', ' . max(1, $limit),
            MYSQLI_USE_RESULT
        );
        if ($result === false) {
            throw new RuntimeException('Could not read rows from ' . $primaryDatabase . '.' . $tableName . '.');
        }

        $fields = $result->fetch_fields();
        $insertPrefix = 'INSERT INTO ' . self::quoteIdentifier($targetDatabase) . '.' . self::quoteIdentifier($tableName)
            . ' (' . implode(', ', $quotedColumns) . ') VALUES ';
        $rows = [];
        $batchSize = 200;
        $copied = 0;

        try {
            $target->begin_transaction();
            while (true) {
                $row = $result->fetch_assoc();
                if (!is_array($row)) {
                    break;
                }

                $values = [];
                foreach ($fields as $field) {
                    $value = $row[$field->name] ?? null;
                    $values[] = $value === null ? 'NULL' : self::sqlString($target, (string)$value);
                }
                $rows[] = '(' . implode(', ', $values) . ')';
                $copied++;

                if (count($rows) >= $batchSize) {
                    $this->mysqliExec($target, $insertPrefix . implode(', ', $rows));
                    $rows = [];
                }
            }

            if ($rows !== []) {
                $this->mysqliExec($target, $insertPrefix . implode(', ', $rows));
            }
            $target->commit();
        } catch (Throwable $throwable) {
            try {
                $target->rollback();
            } catch (Throwable) {
            }
            throw $throwable;
        } finally {
            $result->free();
        }

        return $copied;
    }

    private function estimatedDatabaseTableRows(mysqli $connection, string $database, string $tableName): int
    {
        $stmt = $connection->prepare(
            "SELECT TABLE_ROWS
             FROM information_schema.TABLES
             WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND TABLE_TYPE = 'BASE TABLE'
             LIMIT 1"
        );
        if ($stmt === false) {
            return 1;
        }

        $stmt->bind_param('ss', $database, $tableName);
        $stmt->execute();
        $result = $stmt->get_result();
        $row = $result ? $result->fetch_assoc() : null;
        $stmt->close();

        return max(1, (int)($row['TABLE_ROWS'] ?? 1));
    }

    private static function databaseReplicaSeedSkipsTableData(string $tableName): bool
    {
        return in_array(strtolower($tableName), self::MARIADB_SCHEMA_ONLY_TABLES, true);
    }

    private function databaseWritableColumnNames(mysqli $connection, string $database, string $tableName): array
    {
        $stmt = $connection->prepare(
            "SELECT COLUMN_NAME
             FROM information_schema.COLUMNS
             WHERE TABLE_SCHEMA = ?
               AND TABLE_NAME = ?
               AND EXTRA NOT LIKE '%GENERATED%'
             ORDER BY ORDINAL_POSITION"
        );
        if ($stmt === false) {
            throw new RuntimeException('Could not prepare table column lookup.');
        }

        $stmt->bind_param('ss', $database, $tableName);
        $stmt->execute();
        $result = $stmt->get_result();
        $rows = $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
        $stmt->close();

        return array_values(array_filter(array_map(
            static fn(array $row): string => (string)($row['COLUMN_NAME'] ?? ''),
            $rows
        )));
    }

    private function databasePrimaryKeyColumnNames(mysqli $connection, string $database, string $tableName): array
    {
        $stmt = $connection->prepare(
            "SELECT COLUMN_NAME
             FROM information_schema.KEY_COLUMN_USAGE
             WHERE TABLE_SCHEMA = ?
               AND TABLE_NAME = ?
               AND CONSTRAINT_NAME = 'PRIMARY'
             ORDER BY ORDINAL_POSITION"
        );
        if ($stmt === false) {
            return [];
        }

        $stmt->bind_param('ss', $database, $tableName);
        $stmt->execute();
        $result = $stmt->get_result();
        $rows = $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
        $stmt->close();

        return array_values(array_filter(array_map(
            static fn(array $row): string => (string)($row['COLUMN_NAME'] ?? ''),
            $rows
        )));
    }

    private function databaseTableNames(mysqli $connection, string $database): array
    {
        $stmt = $connection->prepare(
            "SELECT TABLE_NAME FROM information_schema.TABLES
             WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE'
             ORDER BY TABLE_NAME"
        );
        if ($stmt === false) {
            throw new RuntimeException('Could not prepare database table comparison query.');
        }

        $stmt->bind_param('s', $database);
        $stmt->execute();
        $result = $stmt->get_result();
        $rows = $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
        $stmt->close();

        return array_values(array_map(
            static fn(array $row): string => (string)($row['TABLE_NAME'] ?? ''),
            $rows
        ));
    }

    private function databaseSchemaOnlyTablesWithRows(mysqli $connection, string $database): array
    {
        $existingTables = [];
        foreach ($this->databaseTableNames($connection, $database) as $tableName) {
            $existingTables[strtolower($tableName)] = $tableName;
        }

        $tablesWithRows = [];
        foreach (self::MARIADB_SCHEMA_ONLY_TABLES as $schemaOnlyTable) {
            $actualTable = $existingTables[strtolower($schemaOnlyTable)] ?? null;
            if ($actualTable === null) {
                continue;
            }

            $row = $this->mysqliSelectOne(
                $connection,
                'SELECT 1 AS has_rows FROM ' . self::quoteIdentifier($database) . '.' . self::quoteIdentifier($actualTable) . ' LIMIT 1'
            );
            if (($row['has_rows'] ?? null) !== null) {
                $tablesWithRows[] = $actualTable;
            }
        }

        return $tablesWithRows;
    }

    private static function missingDatabaseTablesBlocker(string $database, array $missingTables): string
    {
        $shownTables = array_slice($missingTables, 0, 3);
        $qualifiedTables = array_map(
            static fn(string $table): string => $database . '.' . $table,
            $shownTables
        );
        $tableWord = count($missingTables) === 1 ? 'table' : 'tables';
        $sample = $qualifiedTables !== [] ? ': ' . implode(', ', $qualifiedTables) : '';

        return 'Replica seed is incomplete. Missing ' . count($missingTables) . ' database ' . $tableWord . ' on replica' . $sample . '.';
    }

    private static function schemaOnlyTablesContainRowsBlocker(string $database, array $tables): string
    {
        $shownTables = array_slice($tables, 0, 3);
        $qualifiedTables = array_map(
            static fn(string $table): string => $database . '.' . $table,
            $shownTables
        );
        $sample = $qualifiedTables !== [] ? ': ' . implode(', ', $qualifiedTables) : '';

        return 'Replica seed includes data for schema-only tables' . $sample . '. Re-run provisioning to rebuild the replica without operational log data.';
    }

    private function minioS3Client(array $host): S3Client
    {
        $credentials = $this->credentials($host);
        return new S3Client([
            'version' => 'latest',
            'region' => 'us-east-1',
            'endpoint' => self::minioEndpoint($host),
            'use_path_style_endpoint' => true,
            'credentials' => [
                'key' => $credentials['username'],
                'secret' => $credentials['password'],
            ],
        ]);
    }

    private function minioBucketStats(S3Client $client, array $buckets, bool $requireExists, string $missingPrefix): array
    {
        $stats = [
            'bytes' => 0,
            'objects' => 0,
            'buckets' => [],
            'blockers' => [],
        ];

        foreach ($buckets as $bucket) {
            try {
                $exists = (bool)$client->doesBucketExist($bucket);
                if (!$exists) {
                    $stats['buckets'][] = [
                        'name' => $bucket,
                        'status' => 'missing',
                        'bytes' => 0,
                        'objects' => 0,
                    ];
                    if ($requireExists) {
                        $stats['blockers'][] = $missingPrefix . ' ' . $bucket . ' is missing.';
                    }
                    continue;
                }

                $bucketBytes = 0;
                $bucketObjects = 0;
                $token = null;
                do {
                    $args = ['Bucket' => $bucket];
                    if ($token !== null) {
                        $args['ContinuationToken'] = $token;
                    }
                    $result = $client->listObjectsV2($args);
                    foreach (($result['Contents'] ?? []) as $object) {
                        $bucketBytes += (int)($object['Size'] ?? 0);
                        $bucketObjects++;
                    }
                    $token = isset($result['NextContinuationToken']) ? (string)$result['NextContinuationToken'] : null;
                } while ($token !== null);

                $stats['bytes'] += $bucketBytes;
                $stats['objects'] += $bucketObjects;
                $stats['buckets'][] = [
                    'name' => $bucket,
                    'status' => 'ok',
                    'bytes' => $bucketBytes,
                    'objects' => $bucketObjects,
                ];
            } catch (Throwable $throwable) {
                $stats['buckets'][] = [
                    'name' => $bucket,
                    'status' => 'down',
                    'bytes' => 0,
                    'objects' => 0,
                    'error' => $throwable->getMessage(),
                ];
                $stats['blockers'][] = $missingPrefix . ' ' . $bucket . ' could not be inspected: ' . $throwable->getMessage();
            }
        }

        $stats['blockers'] = array_values(array_unique($stats['blockers']));
        return $stats;
    }

    private function configureMinioReplication(array $primary, array $target): void
    {
        $primaryOptions = $this->decodeOptions($primary);
        $targetOptions = $this->decodeOptions($target);
        $buckets = self::normalizeMinioBuckets($targetOptions['buckets'] ?? $primaryOptions['buckets'] ?? self::MINIO_DEFAULT_BUCKETS);
        $configDir = $this->createMinioConfigDir();

        try {
            $this->prepareMinioAlias($configDir, 'source', $primary);
            $this->prepareMinioAlias($configDir, 'target', $target);

            foreach ($buckets as $bucket) {
                $this->runMinioClient($configDir, ['mb', '--ignore-existing', 'target/' . $bucket]);
                $this->runMinioClient($configDir, ['version', 'enable', 'source/' . $bucket]);
                $this->runMinioClient($configDir, ['version', 'enable', 'target/' . $bucket]);
                try {
                    $this->runMinioClient($configDir, [
                        'replicate',
                        'add',
                        '--remote-bucket',
                        'target/' . $bucket,
                        '--replicate',
                        'delete,delete-marker,existing-objects',
                        'source/' . $bucket,
                    ]);
                } catch (Throwable $throwable) {
                    $message = strtolower($throwable->getMessage());
                    if (!str_contains($message, 'already') && !str_contains($message, 'exists')) {
                        throw $throwable;
                    }
                }
            }
        } finally {
            $this->removeDirectory($configDir);
        }
    }

    private function minioReplicationConfigured(array $primary, array $buckets): bool
    {
        $configDir = $this->createMinioConfigDir();
        try {
            $this->prepareMinioAlias($configDir, 'source', $primary);
            foreach ($buckets as $bucket) {
                try {
                    $result = $this->runMinioClient($configDir, ['replicate', 'ls', '--json', 'source/' . $bucket]);
                } catch (Throwable) {
                    return false;
                }
                if (trim((string)$result['stdout']) === '') {
                    return false;
                }
            }

            return true;
        } finally {
            $this->removeDirectory($configDir);
        }
    }

    private function minioTargetFreeBytes(array $host): ?int
    {
        $configDir = $this->createMinioConfigDir();
        try {
            $this->prepareMinioAlias($configDir, 'target', $host);
            $result = $this->runMinioClient($configDir, ['admin', 'info', '--json', 'target']);
            $decoded = self::decodeMinioJsonOutput((string)$result['stdout']);
            return self::minioAvailableBytesFromAdminInfo($decoded);
        } catch (Throwable) {
            return null;
        } finally {
            $this->removeDirectory($configDir);
        }
    }

    private function prepareMinioAlias(string $configDir, string $alias, array $host): void
    {
        $credentials = $this->credentials($host);
        $this->runMinioClient($configDir, [
            'alias',
            'set',
            $alias,
            self::minioEndpoint($host),
            $credentials['username'],
            $credentials['password'],
        ]);
    }

    private function runMinioClient(string $configDir, array $arguments): array
    {
        $command = array_merge(['mc', '--config-dir', $configDir], array_map('strval', $arguments));
        $pipes = [];
        $process = @proc_open($command, [
            1 => ['pipe', 'w'],
            2 => ['pipe', 'w'],
        ], $pipes);
        if (!is_resource($process)) {
            throw new RuntimeException('MinIO Client (mc) is not available.');
        }

        $stdout = stream_get_contents($pipes[1]);
        fclose($pipes[1]);
        $stderr = stream_get_contents($pipes[2]);
        fclose($pipes[2]);
        $exitCode = proc_close($process);
        if ($exitCode !== 0) {
            $message = trim((string)$stderr) ?: trim((string)$stdout) ?: 'MinIO Client command failed.';
            throw new RuntimeException($message);
        }

        return [
            'stdout' => (string)$stdout,
            'stderr' => (string)$stderr,
            'exit_code' => $exitCode,
        ];
    }

    private function createMinioConfigDir(): string
    {
        $dir = rtrim(sys_get_temp_dir(), DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . 'truckwash-mc-' . bin2hex(random_bytes(6));
        if (!mkdir($dir, 0700, true) && !is_dir($dir)) {
            throw new RuntimeException('Could not create MinIO Client config directory.');
        }
        return $dir;
    }

    private function removeDirectory(string $dir): void
    {
        if (!is_dir($dir)) {
            return;
        }
        $entries = scandir($dir);
        if (!is_array($entries)) {
            @rmdir($dir);
            return;
        }
        foreach ($entries as $entry) {
            if ($entry === '.' || $entry === '..') {
                continue;
            }
            $path = $dir . DIRECTORY_SEPARATOR . $entry;
            if (is_dir($path)) {
                $this->removeDirectory($path);
            } else {
                @unlink($path);
            }
        }
        @rmdir($dir);
    }

    private static function decodeMinioJsonOutput(string $output): mixed
    {
        $trimmed = trim($output);
        if ($trimmed === '') {
            return null;
        }

        $decoded = json_decode($trimmed, true);
        if (json_last_error() === JSON_ERROR_NONE) {
            return $decoded;
        }

        $items = [];
        foreach (preg_split('/\R/', $trimmed) ?: [] as $line) {
            $line = trim($line);
            if ($line === '') {
                continue;
            }
            $decodedLine = json_decode($line, true);
            if (json_last_error() === JSON_ERROR_NONE) {
                $items[] = $decodedLine;
            }
        }

        return $items !== [] ? $items : null;
    }

    public static function minioAvailableBytesFromAdminInfo(mixed $value): ?int
    {
        $values = [];
        self::collectMinioAvailableByteValues($value, $values);
        if ($values === []) {
            return null;
        }

        return max($values);
    }

    private static function collectMinioAvailableByteValues(mixed $value, array &$values): void
    {
        if (!is_array($value)) {
            return;
        }

        foreach ($value as $key => $entry) {
            $normalizedKey = strtolower((string)$key);
            if (in_array($normalizedKey, ['available', 'availablebytes', 'available_bytes', 'avail', 'availspace', 'avail_space', 'availablespace', 'available_space', 'free', 'freebytes', 'free_bytes', 'freespace', 'free_space'], true)) {
                $bytes = self::parseMinioByteValue($entry);
                if ($bytes !== null && $bytes >= 0) {
                    $values[] = $bytes;
                }
            }
            if (is_array($entry)) {
                self::collectMinioAvailableByteValues($entry, $values);
            }
        }
    }

    private static function parseMinioByteValue(mixed $value): ?int
    {
        if (is_int($value)) {
            return $value;
        }
        if (is_float($value)) {
            return (int)$value;
        }

        $text = trim((string)$value);
        if ($text === '') {
            return null;
        }
        if (ctype_digit($text)) {
            return (int)$text;
        }
        if (preg_match('/^([0-9]+(?:\.[0-9]+)?)\s*([kmgtp]?i?b?|bytes?)$/i', $text, $matches) !== 1) {
            return null;
        }

        $number = (float)$matches[1];
        $unit = strtolower($matches[2]);
        $multipliers = [
            'b' => 1,
            'byte' => 1,
            'bytes' => 1,
            'kb' => 1000,
            'kib' => 1024,
            'mb' => 1000 ** 2,
            'mib' => 1024 ** 2,
            'gb' => 1000 ** 3,
            'gib' => 1024 ** 3,
            'tb' => 1000 ** 4,
            'tib' => 1024 ** 4,
            'pb' => 1000 ** 5,
            'pib' => 1024 ** 5,
        ];

        return (int)floor($number * ($multipliers[$unit] ?? 1));
    }

    private function redisClient(array $host): PredisClient
    {
        $credentials = $this->credentials($host);
        $params = [
            'scheme' => 'tcp',
            'host' => (string)$host['host'],
            'port' => (int)$host['port'],
            'database' => (int)($host['database_index'] ?? 0),
            'password' => $credentials['password'],
        ];

        if (($credentials['username'] ?? '') !== '' && $credentials['username'] !== 'default') {
            $params['username'] = $credentials['username'];
        }

        return new PredisClient($params);
    }

    private function redisInfo(PredisClient $client): array
    {
        $info = $client->info('replication');
        if (is_array($info)) {
            return isset($info['Replication']) && is_array($info['Replication'])
                ? $info['Replication']
                : $info;
        }

        $parsed = [];
        foreach (explode("\n", (string)$info) as $line) {
            $line = trim($line);
            if ($line === '' || str_starts_with($line, '#') || !str_contains($line, ':')) {
                continue;
            }
            [$key, $value] = explode(':', $line, 2);
            $parsed[$key] = trim($value);
        }
        return $parsed;
    }

    private function mysqliExec(mysqli $connection, string $sql): void
    {
        $connection->query($sql);
    }

    private function mysqliSelectOne(mysqli $connection, string $sql): array
    {
        $result = $connection->query($sql);
        if ($result === false) {
            return [];
        }
        $row = $result->fetch_assoc();
        return is_array($row) ? $row : [];
    }

    private function refreshStatuses(): void
    {
        foreach ($this->listHosts() as $host) {
            if ($this->activeOperation((string)$host['kind'], (int)$host['id']) !== null) {
                continue;
            }

            try {
                $status = match ((string)$host['kind']) {
                    self::KIND_DATABASE => $this->testDatabaseHost($host),
                    self::KIND_REDIS => $this->testRedisHost($host),
                    self::KIND_MINIO => $this->testMinioHost($host),
                    default => throw new RuntimeException('Unsupported replication kind.'),
                };
                $this->storeStatus($host, $status);
            } catch (Throwable $throwable) {
                $this->storeStatus($host, [
                    'status' => 'degraded',
                    'replication_percent' => 0,
                    'lag_seconds' => null,
                    'blockers' => [$throwable->getMessage()],
                    'raw' => [],
                    'checked_at' => date('c'),
                ]);
            }
        }
    }

    private function storeStatus(array $host, array $status): void
    {
        $publicStatus = [
            'status' => (string)($status['status'] ?? 'unknown'),
            'replication_percent' => round((float)($status['replication_percent'] ?? 0), 2),
            'lag_seconds' => $status['lag_seconds'] ?? null,
            'blockers' => array_values(array_filter($status['blockers'] ?? [])),
            'raw' => $status['raw'] ?? [],
            'checked_at' => (string)($status['checked_at'] ?? date('c')),
        ];

        $this->execute(
            "UPDATE replication_hosts SET status = ?, last_status_json = ?, last_checked_at = NOW() WHERE id = ?",
            'ssi',
            [$publicStatus['status'], self::jsonEncode($publicStatus), (int)$host['id']]
        );
        $this->execute(
            "INSERT INTO replication_status_snapshots
                (host_id, kind, status, replication_percent, lag_seconds, blockers_json, raw_status_json)
             VALUES (?, ?, ?, ?, ?, ?, ?)",
            'issdiss',
            [
                (int)$host['id'],
                (string)$host['kind'],
                $publicStatus['status'],
                (float)$publicStatus['replication_percent'],
                $publicStatus['lag_seconds'],
                self::jsonEncode($publicStatus['blockers']),
                self::jsonEncode($publicStatus['raw']),
            ]
        );
    }

    private function buildReplicationSummary(string $kind, array $hosts): array
    {
        $replicas = [];
        $blockers = [];
        $percents = [];
        $statuses = [];

        foreach ($hosts as $host) {
            if (($host['role'] ?? '') === 'primary') {
                continue;
            }

            $public = $this->publicHost($host);
            $replicas[] = $public;
            $status = is_array($public['last_status'] ?? null) ? $public['last_status'] : [];
            $activeProgress = $public['active_operation']['progress_percent'] ?? null;
            $percent = is_numeric($activeProgress) && (float)$activeProgress > 0
                ? (float)$activeProgress
                : (float)($status['replication_percent'] ?? $public['replication_percent'] ?? 0);
            $percents[] = $percent;
            $statuses[] = (string)($status['status'] ?? $public['status'] ?? 'unknown');
            foreach ($status['blockers'] ?? [] as $blocker) {
                $blockers[] = (string)$blocker;
            }
        }

        if ($replicas === []) {
            return [
                'status' => 'not_configured',
                'min_percent' => 0.0,
                'average_percent' => 0.0,
                'replicas' => [],
                'blockers' => ['No ' . $kind . ' replicas configured.'],
            ];
        }

        $min = min($percents);
        $average = array_sum($percents) / max(1, count($percents));
        $status = ($min >= 100.0 && $blockers === []) ? 'ok' : 'degraded';
        if (in_array('down', $statuses, true)) {
            $status = 'down';
        }

        return [
            'status' => $status,
            'min_percent' => round($min, 2),
            'average_percent' => round($average, 2),
            'replicas' => $replicas,
            'blockers' => array_values(array_unique($blockers)),
        ];
    }

    private function publicHost(?array $host): ?array
    {
        if ($host === null) {
            return null;
        }

        $lastStatus = self::jsonDecode($host['last_status_json'] ?? null);
        $credentials = $this->credentials($host);
        $options = $this->decodeOptions($host);
        $activeOperation = isset($host['id'])
            ? $this->activeOperation((string)$host['kind'], (int)$host['id'])
            : null;
        $database = match ((string)$host['kind']) {
            self::KIND_DATABASE => (string)($host['database_name'] ?? ''),
            self::KIND_REDIS => (int)($host['database_index'] ?? 0),
            self::KIND_MINIO => null,
            default => null,
        };

        return [
            'id' => (int)$host['id'],
            'kind' => (string)$host['kind'],
            'label' => (string)$host['label'],
            'host' => (string)$host['host'],
            'port' => (int)$host['port'],
            'database' => $database,
            'endpoint' => (string)($options['endpoint'] ?? (((string)$host['kind'] === self::KIND_MINIO) ? self::minioEndpoint($host) : '')),
            'scheme' => $options['scheme'] ?? null,
            'buckets' => ((string)$host['kind'] === self::KIND_MINIO) ? self::normalizeMinioBuckets($options['buckets'] ?? []) : [],
            'console_port' => ((string)$host['kind'] === self::KIND_MINIO) ? (int)($options['console_port'] ?? 9001) : null,
            'space_headroom_percent' => ((string)$host['kind'] === self::KIND_MINIO) ? (float)($options['space_headroom_percent'] ?? self::MINIO_SPACE_HEADROOM_PERCENT) : null,
            'role' => (string)$host['role'],
            'status' => (string)$host['status'],
            'replication_source_id' => isset($host['replication_source_id']) ? (int)$host['replication_source_id'] : null,
            'ssl_mode' => $host['ssl_mode'] ?? null,
            'replication_percent' => round((float)($lastStatus['replication_percent'] ?? (($host['role'] ?? '') === 'primary' ? 100 : 0)), 2),
            'last_status' => $lastStatus,
            'active_operation' => $activeOperation,
            'last_checked_at' => $host['last_checked_at'] ?? null,
            'credential_summary' => [
                'username' => $credentials['username'] !== '' ? replication_secret_box::mask($credentials['username']) : '',
                'password_set' => $credentials['password'] !== '',
                'admin_username' => $credentials['admin_username'] !== '' ? replication_secret_box::mask($credentials['admin_username']) : '',
                'admin_password_set' => $credentials['admin_password'] !== '',
                'replication_username' => $credentials['replication_username'] !== '' ? replication_secret_box::mask($credentials['replication_username']) : '',
                'replication_password_set' => $credentials['replication_password'] !== '',
            ],
            'created_at' => $host['created_at'] ?? null,
            'updated_at' => $host['updated_at'] ?? null,
            'deleted_at' => $host['deleted_at'] ?? null,
        ];
    }

    private static function normalizeMinioAddress(string $host, mixed $port, mixed $scheme): array
    {
        $raw = trim($host);
        $hasScheme = preg_match('/^https?:\/\//i', $raw) === 1;
        $parsed = parse_url($hasScheme ? $raw : 'http://' . $raw);
        if (!is_array($parsed) || empty($parsed['host'])) {
            throw new RuntimeException('MinIO endpoint host is invalid.');
        }

        $normalizedScheme = strtolower(trim((string)($scheme ?: ($parsed['scheme'] ?? 'http'))));
        if (!in_array($normalizedScheme, ['http', 'https'], true)) {
            throw new RuntimeException('MinIO scheme must be http or https.');
        }

        $normalizedHost = trim((string)$parsed['host']);
        $portValue = ($port !== null && trim((string)$port) !== '')
            ? $port
            : ($parsed['port'] ?? ($normalizedScheme === 'https' ? 443 : 9000));
        $normalizedPort = (int)$portValue;

        return [$normalizedHost, $normalizedPort, $normalizedScheme];
    }

    private static function minioEndpointFromParts(string $scheme, string $host, int $port): string
    {
        return strtolower($scheme) . '://' . $host . ':' . $port;
    }

    private static function minioEndpoint(array $host): string
    {
        $options = isset($host['options']) && is_array($host['options'])
            ? $host['options']
            : self::jsonDecode($host['options_json'] ?? null);
        $endpoint = trim((string)($options['endpoint'] ?? ''));
        if ($endpoint !== '') {
            return $endpoint;
        }

        return self::minioEndpointFromParts((string)($options['scheme'] ?? 'http'), (string)$host['host'], (int)$host['port']);
    }

    private function normalizeHostInput(string $kind, array $input): array
    {
        $host = trim((string)($input['host'] ?? $input['endpoint'] ?? ''));
        if ($host === '') {
            throw new RuntimeException('Host is required.');
        }

        $scheme = null;
        $defaultPort = match ($kind) {
            self::KIND_DATABASE => 3306,
            self::KIND_REDIS => 6379,
            self::KIND_MINIO => 9000,
        };
        if ($kind === self::KIND_MINIO) {
            [$host, $port, $scheme] = self::normalizeMinioAddress($host, $input['port'] ?? null, $input['scheme'] ?? null);
        } else {
            $port = (int)($input['port'] ?? $defaultPort);
        }
        if ($port <= 0 || $port > 65535) {
            throw new RuntimeException('Port must be between 1 and 65535.');
        }

        $label = trim((string)($input['label'] ?? ''));
        if ($label === '') {
            $label = $host . ':' . $port;
        }

        $username = trim((string)($input['username'] ?? $input['access_key'] ?? $input['user'] ?? ''));
        $password = (string)($input['password'] ?? $input['secret_key'] ?? '');
        $databaseName = null;
        $databaseIndex = null;
        if ($kind === self::KIND_DATABASE) {
            $databaseName = trim((string)($input['database'] ?? $input['database_name'] ?? ''));
            if ($databaseName === '' || $username === '') {
                throw new RuntimeException('Database name and username are required for database replication hosts.');
            }
        } elseif ($kind === self::KIND_REDIS) {
            $databaseIndex = (int)($input['database'] ?? $input['database_index'] ?? 0);
            if ($databaseIndex < 0) {
                throw new RuntimeException('Redis database index must be zero or greater.');
            }
        } else {
            if ($username === '' || $password === '') {
                throw new RuntimeException('Access key and secret key are required for MinIO replication hosts.');
            }
        }

        $options = is_array($input['options'] ?? null) ? $input['options'] : [];
        if ($kind === self::KIND_MINIO) {
            $headroom = (float)($input['space_headroom_percent'] ?? $options['space_headroom_percent'] ?? self::MINIO_SPACE_HEADROOM_PERCENT);
            $options = array_replace($options, [
                'scheme' => $scheme ?: 'http',
                'endpoint' => self::minioEndpointFromParts($scheme ?: 'http', $host, $port),
                'buckets' => self::normalizeMinioBuckets($input['buckets'] ?? $options['buckets'] ?? self::MINIO_DEFAULT_BUCKETS),
                'console_port' => (int)($input['console_port'] ?? $options['console_port'] ?? 9001),
                'space_headroom_percent' => max(0.0, $headroom),
            ]);
        }

        return [
            'label' => $label,
            'host' => $host,
            'port' => $port,
            'database_name' => $databaseName,
            'database_index' => $databaseIndex,
            'username' => $username,
            'password_secret' => replication_secret_box::encrypt($password),
            'admin_username' => trim((string)($input['admin_username'] ?? '')),
            'admin_password_secret' => replication_secret_box::encrypt((string)($input['admin_password'] ?? '')),
            'replication_username' => trim((string)($input['replication_username'] ?? '')),
            'replication_password_secret' => replication_secret_box::encrypt((string)($input['replication_password'] ?? '')),
            'ssl_mode' => strtoupper(trim((string)($input['ssl_mode'] ?? 'DISABLED'))) ?: 'DISABLED',
            'options' => $options,
        ];
    }

    private function transientHost(string $kind, array $input): array
    {
        $normalized = $this->normalizeHostInput($kind, $input);
        $role = strtolower(trim((string)($input['role'] ?? 'replica')));
        if (!in_array($role, ['primary', 'replica'], true)) {
            $role = 'replica';
        }

        return array_merge($normalized, [
            'id' => 0,
            'kind' => $kind,
            'role' => $role,
            'status' => 'unknown',
            'replication_source_id' => null,
            'last_status_json' => null,
            'last_checked_at' => null,
            'created_at' => null,
            'updated_at' => null,
            'deleted_at' => null,
            'test_connectivity_only' => true,
        ]);
    }

    private function ensureEnvironmentPrimaryRows(): void
    {
        if ($this->primaryHost(self::KIND_DATABASE) === null && isset($GLOBALS['CONFIG_DB']) && is_array($GLOBALS['CONFIG_DB'])) {
            $config = $GLOBALS['CONFIG_DB'];
            if (!empty($config['host']) && !empty($config['database']) && !empty($config['user'])) {
                $this->insertEnvironmentPrimary(self::KIND_DATABASE, [
                    'label' => 'Current database primary',
                    'host' => (string)$config['host'],
                    'port' => (int)($config['port'] ?? 3306),
                    'database_name' => (string)$config['database'],
                    'database_index' => null,
                    'username' => (string)$config['user'],
                    'password_secret' => replication_secret_box::encrypt((string)($config['password'] ?? '')),
                    'ssl_mode' => (string)($config['ssl_mode'] ?? 'DISABLED'),
                ]);
            }
        }

        if ($this->primaryHost(self::KIND_REDIS) === null && isset($GLOBALS['REDIS_CONFIG']) && is_array($GLOBALS['REDIS_CONFIG'])) {
            $config = $GLOBALS['REDIS_CONFIG'];
            if (!empty($config['host'])) {
                $this->insertEnvironmentPrimary(self::KIND_REDIS, [
                    'label' => 'Current Redis primary',
                    'host' => (string)$config['host'],
                    'port' => (int)($config['port'] ?? 6379),
                    'database_name' => null,
                    'database_index' => (int)($config['database'] ?? 0),
                    'username' => (string)($config['user'] ?? ''),
                    'password_secret' => replication_secret_box::encrypt((string)($config['password'] ?? '')),
                    'ssl_mode' => null,
                ]);
            }
        }

        if ($this->primaryHost(self::KIND_MINIO) === null && isset($GLOBALS['MINIO']) && is_array($GLOBALS['MINIO'])) {
            $config = $GLOBALS['MINIO'];
            $endpoint = trim((string)($config['endpoint'] ?? ''));
            $accessKey = trim((string)($config['access_key'] ?? ''));
            if ($endpoint !== '' && $accessKey !== '') {
                [$host, $port, $scheme] = self::normalizeMinioAddress($endpoint, null, null);
                $buckets = self::normalizeMinioBuckets($config['buckets'] ?? self::MINIO_DEFAULT_BUCKETS);
                $this->insertEnvironmentPrimary(self::KIND_MINIO, [
                    'label' => 'Current MinIO primary',
                    'host' => $host,
                    'port' => $port,
                    'database_name' => null,
                    'database_index' => null,
                    'username' => $accessKey,
                    'password_secret' => replication_secret_box::encrypt((string)($config['secret_key'] ?? '')),
                    'ssl_mode' => null,
                    'options' => [
                        'source' => 'environment',
                        'scheme' => $scheme,
                        'endpoint' => self::minioEndpointFromParts($scheme, $host, $port),
                        'buckets' => $buckets,
                        'console_port' => (int)($config['console_port'] ?? 9001),
                        'space_headroom_percent' => self::MINIO_SPACE_HEADROOM_PERCENT,
                    ],
                ]);
            }
        }
    }

    private function insertEnvironmentPrimary(string $kind, array $host): void
    {
        $this->execute(
            "INSERT INTO replication_hosts (
                kind, label, host, port, database_name, database_index, username, password_secret,
                role, status, ssl_mode, options_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'primary', 'unknown', ?, ?)",
            'sssissssss',
            [
                $kind,
                $host['label'],
                $host['host'],
                $host['port'],
                $host['database_name'],
                $host['database_index'],
                $host['username'],
                $host['password_secret'],
                $host['ssl_mode'],
                self::jsonEncode(array_replace(['source' => 'environment'], is_array($host['options'] ?? null) ? $host['options'] : [])),
            ]
        );
    }

    private function writeBootstrapSnapshot(): void
    {
        $databasePrimary = $this->primaryHost(self::KIND_DATABASE);
        $redisPrimary = $this->primaryHost(self::KIND_REDIS);
        $minioPrimary = $this->primaryHost(self::KIND_MINIO);
        $active = [];

        if ($databasePrimary !== null) {
            $credentials = $this->credentials($databasePrimary);
            $active['database'] = [
                'host' => (string)$databasePrimary['host'],
                'port' => (int)$databasePrimary['port'],
                'database' => (string)$databasePrimary['database_name'],
                'user' => $credentials['username'],
                'password_secret' => $databasePrimary['password_secret'] ?? '',
                'ssl_mode' => (string)($databasePrimary['ssl_mode'] ?? 'DISABLED'),
            ];
        }

        if ($redisPrimary !== null) {
            $credentials = $this->credentials($redisPrimary);
            $active['redis'] = [
                'host' => (string)$redisPrimary['host'],
                'port' => (int)$redisPrimary['port'],
                'database' => (int)($redisPrimary['database_index'] ?? 0),
                'user' => $credentials['username'],
                'password_secret' => $redisPrimary['password_secret'] ?? '',
            ];
        }

        if ($minioPrimary !== null) {
            $credentials = $this->credentials($minioPrimary);
            $options = $this->decodeOptions($minioPrimary);
            $active['minio'] = [
                'endpoint' => self::minioEndpoint($minioPrimary),
                'access_key' => $credentials['username'],
                'secret_key_secret' => $minioPrimary['password_secret'] ?? '',
                'buckets' => self::normalizeMinioBuckets($options['buckets'] ?? self::MINIO_DEFAULT_BUCKETS),
            ];
        }

        replication_bootstrap_config::writeSnapshot([
            'version' => 1,
            'generated_at' => date('c'),
            'active' => $active,
        ]);
    }

    private function switchPrimary(string $kind, int $newPrimaryId, int $oldPrimaryId): void
    {
        $this->execute(
            "UPDATE replication_hosts SET role = 'inactive', status = 'inactive' WHERE kind = ? AND role = 'primary' AND id <> ?",
            'si',
            [$kind, $newPrimaryId]
        );
        $this->execute(
            "UPDATE replication_hosts SET role = 'primary', status = 'ok', replication_source_id = NULL WHERE kind = ? AND id = ?",
            'si',
            [$kind, $newPrimaryId]
        );
        $this->execute(
            "UPDATE replication_hosts SET replication_source_id = ? WHERE kind = ? AND role = 'replica'",
            'is',
            [$newPrimaryId, $kind]
        );
    }

    private function credentials(array $host): array
    {
        return [
            'username' => (string)($host['username'] ?? ''),
            'password' => replication_secret_box::decrypt($host['password_secret'] ?? ''),
            'admin_username' => (string)($host['admin_username'] ?? ''),
            'admin_password' => replication_secret_box::decrypt($host['admin_password_secret'] ?? ''),
            'replication_username' => (string)($host['replication_username'] ?? ''),
            'replication_password' => replication_secret_box::decrypt($host['replication_password_secret'] ?? ''),
        ];
    }

    private function decodeOptions(array $host): array
    {
        if (isset($host['options']) && is_array($host['options'])) {
            return $host['options'];
        }

        return self::jsonDecode($host['options_json'] ?? null);
    }

    private function listHosts(?string $kind = null, bool $includeDeleted = false): array
    {
        $where = [];
        $types = '';
        $params = [];
        if ($kind !== null) {
            $where[] = 'kind = ?';
            $types .= 's';
            $params[] = $kind;
        }
        if (!$includeDeleted) {
            $where[] = 'deleted_at IS NULL';
        }

        $sql = 'SELECT * FROM replication_hosts';
        if ($where !== []) {
            $sql .= ' WHERE ' . implode(' AND ', $where);
        }
        $sql .= " ORDER BY FIELD(role, 'primary', 'replica', 'inactive'), id";

        return $this->selectRows($sql, $types, $params);
    }

    private function primaryHost(string $kind): ?array
    {
        return $this->selectOne(
            "SELECT * FROM replication_hosts WHERE kind = ? AND role = 'primary' AND deleted_at IS NULL ORDER BY id DESC LIMIT 1",
            's',
            [$kind]
        );
    }

    private function getHost(string $kind, int $id, bool $includeDeleted = false): array
    {
        $sql = 'SELECT * FROM replication_hosts WHERE kind = ? AND id = ?';
        if (!$includeDeleted) {
            $sql .= ' AND deleted_at IS NULL';
        }
        $host = $this->selectOne($sql . ' LIMIT 1', 'si', [$kind, $id]);
        if ($host === null) {
            throw new RuntimeException('Replication host was not found.');
        }
        return $host;
    }

    private function startOperation(string $kind, int $hostId, string $operation, ?int $actorUserId): int
    {
        $this->execute(
            "INSERT INTO replication_operations (kind, host_id, operation, status, actor_user_id)
             VALUES (?, ?, ?, 'running', ?)",
            'sisi',
            [$kind, $hostId, $operation, $actorUserId]
        );
        return $this->insertId();
    }

    private function activeOperationId(string $kind, int $hostId, string $operation): ?int
    {
        $operationRow = $this->selectOne(
            "SELECT id FROM replication_operations
             WHERE kind = ? AND host_id = ? AND operation = ? AND status = 'running'
             ORDER BY id DESC
             LIMIT 1",
            'sis',
            [$kind, $hostId, $operation]
        );

        return $operationRow !== null ? (int)$operationRow['id'] : null;
    }

    private function activeOperation(string $kind, int $hostId): ?array
    {
        $operation = $this->selectOne(
            "SELECT id, operation, status, progress_percent, message, error_message, started_at, updated_at
             FROM replication_operations
             WHERE kind = ? AND host_id = ? AND status = 'running'
             ORDER BY id DESC
             LIMIT 1",
            'si',
            [$kind, $hostId]
        );

        if ($operation === null) {
            return null;
        }

        return [
            'id' => (int)$operation['id'],
            'operation' => (string)$operation['operation'],
            'status' => (string)$operation['status'],
            'progress_percent' => round((float)$operation['progress_percent'], 2),
            'message' => $operation['message'] ?? null,
            'error_message' => $operation['error_message'] ?? null,
            'started_at' => $operation['started_at'] ?? null,
            'updated_at' => $operation['updated_at'] ?? null,
        ];
    }

    private function operationContext(int $operationId): array
    {
        $operation = $this->selectOne(
            'SELECT context_json FROM replication_operations WHERE id = ? LIMIT 1',
            'i',
            [$operationId]
        );

        return self::jsonDecode($operation['context_json'] ?? null);
    }

    private function updateOperationProgress(int $operationId, float $progress, string $message, array $context): void
    {
        $this->execute(
            "UPDATE replication_operations
             SET progress_percent = ?, message = ?, context_json = ?
             WHERE id = ?",
            'dssi',
            [max(0, min(100, $progress)), $message, self::jsonEncode($context), $operationId]
        );
    }

    private function finishOperation(int $operationId, string $status, float $progress, ?string $message, array $errors): void
    {
        $this->execute(
            "UPDATE replication_operations
             SET status = ?, progress_percent = ?, message = ?, error_message = ?, completed_at = NOW()
             WHERE id = ?",
            'sdssi',
            [$status, $progress, $message, implode("\n", $errors), $operationId]
        );
    }

    private function audit(string $kind, ?int $hostId, string $action, ?int $actorUserId, string $severity, array $context): void
    {
        $this->execute(
            "INSERT INTO replication_audit_logs (kind, host_id, action, actor_user_id, severity, context_json)
             VALUES (?, ?, ?, ?, ?, ?)",
            'sisiss',
            [$kind, $hostId, $action, $actorUserId, $severity, self::jsonEncode($context)]
        );
    }

    private function acquirePromotionLock()
    {
        $path = (defined('WD') ? WD : dirname(__DIR__)) . DIRECTORY_SEPARATOR . 'storage' . DIRECTORY_SEPARATOR . 'replication-promotion.lock';
        $dir = dirname($path);
        if (!is_dir($dir) && !mkdir($dir, 0770, true) && !is_dir($dir)) {
            throw new RuntimeException('Could not create promotion lock directory.');
        }
        $handle = fopen($path, 'c');
        if ($handle === false || !flock($handle, LOCK_EX | LOCK_NB)) {
            throw new RuntimeException('Another replication promotion is already running.');
        }
        return $handle;
    }

    private function releasePromotionLock($handle): void
    {
        if (is_resource($handle)) {
            flock($handle, LOCK_UN);
            fclose($handle);
        }
    }

    private function selectOne(string $sql, string $types = '', array $params = []): ?array
    {
        $rows = $this->selectRows($sql, $types, $params);
        return $rows[0] ?? null;
    }

    private function selectRows(string $sql, string $types = '', array $params = []): array
    {
        global $db;
        if ($types === '') {
            $result = $db->query($sql);
            return $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
        }

        $stmt = $db->prepare($sql);
        if ($stmt === false) {
            throw new RuntimeException('Could not prepare replication query.');
        }
        $stmt->bind_param($types, ...$params);
        $stmt->execute();
        $result = $stmt->get_result();
        return $result ? $result->fetch_all(MYSQLI_ASSOC) : [];
    }

    private function execute(string $sql, string $types = '', array $params = []): void
    {
        global $db;
        if ($types === '') {
            $db->query($sql);
            return;
        }

        $stmt = $db->prepare($sql);
        if ($stmt === false) {
            throw new RuntimeException('Could not prepare replication statement.');
        }
        $stmt->bind_param($types, ...$params);
        $stmt->execute();
    }

    private function insertId(): int
    {
        global $db;
        return (int)$db->insert_id();
    }

    private static function jsonEncode(mixed $value): string
    {
        $json = json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        if ($json === false) {
            throw new RuntimeException('Could not encode replication JSON payload.');
        }
        return $json;
    }

    private static function jsonDecode(mixed $value): array
    {
        if (!is_string($value) || trim($value) === '') {
            return [];
        }
        $decoded = json_decode($value, true);
        return is_array($decoded) ? $decoded : [];
    }
}