<?php

declare(strict_types=1);

namespace PpPhp\Core\Databases;

/**
 * Logical cloning of a managed database — read it here, write it there.
 *
 * This is the same technique the licence tables already use, applied to user
 * data. NATIVE MySQL replication is impossible on shared hosting (binlog,
 * `CHANGE MASTER TO`, `SUPER`); this needs none of it — just a login that can
 * read the source and create tables in the target.
 *
 * THREE RULES THIS FILE ENFORCES, because getting them wrong loses data:
 *
 *  1. **Time decides WHAT to copy, never WHO wins.** `fingerprint()` exists so an
 *     unchanged database can be skipped. It must never be used to pick a winner
 *     between two diverged copies — this project runs PHP in UTC and MySQL at
 *     +5h45m (see memory `php-mysql-clock-skew`), so cross-host timestamp
 *     comparison would silently discard the "older" side's real work.
 *
 *  2. **A restore never lands on a live database.** `restore()` refuses any
 *     target that is not `replica_`-prefixed. The replica is a restore target,
 *     not a second master, which is what makes split-brain impossible here
 *     rather than merely unlikely.
 *
 *  3. **Reads are consistent.** The dump runs in a repeatable-read transaction so
 *     a database written mid-dump yields a coherent snapshot, not a torn one.
 */
final class DbCloner
{
    /** Every restore target carries this prefix. Non-negotiable — see rule 2. */
    public const REPLICA_PREFIX = 'replica_';

    /** Rows per INSERT. Big enough to be fast, small enough to stay under max_allowed_packet. */
    private const BATCH = 200;

    // ── Change detection ──────────────────────────────────────────────

    /**
     * A cheap signal of "has anything changed in here".
     *
     * One `information_schema` query per database — no table scan, so this is
     * safe to run over the whole estate on a schedule.
     *
     * `update_time` alone is unreliable on InnoDB (it can be NULL or stale),
     * which is exactly why it is folded together with total size and row count
     * rather than trusted on its own. Any real write moves at least one of them.
     *
     * @param array{host:string,port:int|string,dbname:string,user:string,pass:string} $conn
     * @return string 64-char hash, or '' when the database cannot be read
     */
    public static function fingerprint(array $conn): string
    {
        try {
            $pdo = self::connect($conn);
            $st = $pdo->prepare(
                'SELECT COALESCE(SUM(data_length + index_length), 0) AS bytes,
                        COALESCE(SUM(table_rows), 0)                 AS rws,
                        COALESCE(MAX(update_time), 0)                AS touched,
                        COUNT(*)                                     AS tbls
                   FROM information_schema.TABLES WHERE table_schema = ?'
            );
            $st->execute([$conn['dbname']]);
            $r = $st->fetch(\PDO::FETCH_ASSOC) ?: [];
            return hash('sha256', implode('|', [
                (string) ($r['bytes'] ?? ''), (string) ($r['rws'] ?? ''),
                (string) ($r['touched'] ?? ''), (string) ($r['tbls'] ?? ''),
            ]));
        } catch (\Throwable $e) {
            return '';
        }
    }

    // ── Dump ──────────────────────────────────────────────────────────

    /**
     * Serialise a whole database to SQL.
     *
     * @param array<string,mixed> $conn
     * @return array{success:bool,message:string,sql?:string,tables?:int,rows?:int,bytes?:int}
     */
    public static function dump(array $conn, ?array $only = null, array $drop = []): array
    {
        try {
            $pdo = self::connect($conn);
        } catch (\Throwable $e) {
            return ['success' => false, 'message' => 'Could not read the source database: ' . $e->getMessage()];
        }

        try {
            // Repeatable read across the whole dump: a database being written
            // while we read must still produce a coherent snapshot.
            $pdo->exec('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
            $pdo->exec('START TRANSACTION WITH CONSISTENT SNAPSHOT');

            $tables = $pdo->query('SHOW FULL TABLES WHERE Table_type = \'BASE TABLE\'')->fetchAll(\PDO::FETCH_NUM);

            // Incremental mode: only the tables the caller asked for. Anything not
            // listed is left untouched on the replica, which is the whole point —
            // an unchanged table is neither read here nor rewritten there.
            if ($only !== null) {
                $want = array_flip($only);
                $tables = array_values(array_filter($tables,
                    static fn(array $t): bool => isset($want[(string) $t[0]])));
            }

            $out = [
                '-- PPPHP logical clone of `' . $conn['dbname'] . '`',
                '-- Taken ' . gmdate('c') . ' (UTC). Restore target must be ' . self::REPLICA_PREFIX . '*',
                'SET FOREIGN_KEY_CHECKS=0;',
                'SET NAMES utf8mb4;',
                '',
            ];
            $rowTotal = 0;

            foreach ($tables as $t) {
                $name = (string) $t[0];
                $create = $pdo->query('SHOW CREATE TABLE `' . str_replace('`', '', $name) . '`')->fetch(\PDO::FETCH_NUM);
                $out[] = 'DROP TABLE IF EXISTS `' . $name . '`;';
                $out[] = ((string) ($create[1] ?? '')) . ';';
                $out[] = '';

                $rowTotal += self::dumpRows($pdo, $name, $out);
            }

            // Tables deleted on the source must be removed from the replica too,
            // or it silently accumulates tables that no longer exist.
            foreach ($drop as $d) {
                $out[] = 'DROP TABLE IF EXISTS `' . str_replace('`', '', (string) $d) . '`;';
            }

            $out[] = 'SET FOREIGN_KEY_CHECKS=1;';
            $pdo->commit();

            $sql = implode("\n", $out);
            return ['success' => true, 'message' => 'Dumped ' . count($tables) . ' table(s).',
                    'sql' => $sql, 'tables' => count($tables), 'rows' => $rowTotal, 'bytes' => strlen($sql)];
        } catch (\Throwable $e) {
            if ($pdo->inTransaction()) { try { $pdo->rollBack(); } catch (\Throwable $x) {} }
            return ['success' => false, 'message' => 'Dump failed: ' . $e->getMessage()];
        }
    }

    /**
     * Stream one table's rows into $out as batched INSERTs.
     * Unbuffered so a large table does not have to fit in memory at once.
     */
    private static function dumpRows(\PDO $pdo, string $table, array &$out): int
    {
        $safe = str_replace('`', '', $table);
        $pdo->setAttribute(\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, false);
        $st = $pdo->query('SELECT * FROM `' . $safe . '`');

        $batch = [];
        $cols = null;
        $count = 0;

        while ($row = $st->fetch(\PDO::FETCH_ASSOC)) {
            if ($cols === null) $cols = array_keys($row);
            $vals = [];
            foreach ($row as $v) {
                $vals[] = $v === null ? 'NULL' : $pdo->quote((string) $v);
            }
            $batch[] = '(' . implode(',', $vals) . ')';
            $count++;

            if (count($batch) >= self::BATCH) {
                $out[] = self::insertStmt($safe, $cols, $batch);
                $batch = [];
            }
        }
        if ($batch !== [] && $cols !== null) $out[] = self::insertStmt($safe, $cols, $batch);
        if ($count > 0) $out[] = '';

        $st->closeCursor();
        $pdo->setAttribute(\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, true);
        return $count;
    }

    /** @param array<int,string> $cols @param array<int,string> $rows */
    private static function insertStmt(string $table, array $cols, array $rows): string
    {
        return 'INSERT INTO `' . $table . '` (`' . implode('`,`', $cols) . '`) VALUES '
             . implode(',', $rows) . ';';
    }

    // ── Restore ───────────────────────────────────────────────────────

    /**
     * Apply a dump into a REPLICA database.
     *
     * @param array<string,mixed> $conn connection whose `dbname` is the target
     * @return array{success:bool,message:string,statements?:int}
     */
    public static function restore(array $conn, string $sql): array
    {
        $target = (string) ($conn['dbname'] ?? '');

        // Rule 2, enforced rather than documented: a clone must never be able to
        // land on a live database, however it is called.
        if (!self::isReplicaTarget($target)) {
            return ['success' => false, 'message' =>
                'Refusing to restore into "' . $target . '". A clone may only be written to a database whose '
                . 'name starts with "' . self::REPLICA_PREFIX . '" — the replica is a restore target, never a '
                . 'live database.'];
        }
        if (trim($sql) === '') {
            return ['success' => false, 'message' => 'Refusing to restore an empty dump.'];
        }
        // NOTE: a partial (incremental) dump is legitimate and may contain only a
        // couple of tables — emptiness is the only thing refused here, not size.

        try {
            $pdo = self::connect($conn);
        } catch (\Throwable $e) {
            return ['success' => false, 'message' => 'Could not reach the replica: ' . $e->getMessage()];
        }

        $ran = 0;
        try {
            foreach (self::statements($sql) as $stmt) {
                $pdo->exec($stmt);
                $ran++;
            }
            return ['success' => true, 'message' => 'Restored ' . $ran . ' statement(s) into ' . $target . '.',
                    'statements' => $ran];
        } catch (\Throwable $e) {
            return ['success' => false, 'statements' => $ran,
                    'message' => 'Restore failed after ' . $ran . ' statement(s): ' . $e->getMessage()];
        }
    }

    /**
     * Split a dump into executable statements.
     *
     * Deliberately simple: this only ever parses dumps THIS class produced, where
     * every statement ends with ";\n" and no statement body contains a newline
     * followed by nothing. It is not a general-purpose SQL parser and must not be
     * pointed at arbitrary files.
     *
     * @return array<int,string>
     */
    private static function statements(string $sql): array
    {
        $out = [];
        foreach (preg_split('/;\s*\n/', $sql) ?: [] as $s) {
            $s = trim($s);
            if ($s === '' || str_starts_with($s, '--')) continue;
            $out[] = rtrim($s, ';');
        }
        return $out;
    }

    // ── Helpers ───────────────────────────────────────────────────────


    /**
     * Is this name a legitimate restore target?
     *
     * The marker is looked for ANYWHERE in the name, not just at the start,
     * because a control panel silently prepends the account name: ask cPanel for
     * `replica_orders` and you get `ppradhan_replica_orders`. Requiring a leading
     * match would make every panel-created replica unrestorable.
     *
     * This is still safe: a live managed database is named `pp_db_*` or
     * `<account>_<label>`, and nothing in the provisioning path can produce the
     * `replica_` segment — `provision()` builds names from `slugTag()`, which
     * strips to `[a-z0-9]` and could only collide if a user literally labelled a
     * database "replica".  Guarded below.
     */
    public static function isReplicaTarget(string $name): bool
    {
        return str_contains($name, self::REPLICA_PREFIX);
    }

    /** MySQL's hard limit on a database name. */
    private const MAX_DB_NAME = 64;

    /**
     * The replica name for a source database.
     *
     * `$reserve` is how many characters the target control panel will prepend
     * (cPanel/DirectAdmin add `<account>_`). It matters because the source name
     * usually ALREADY carries the other side's account prefix, so a clone can
     * carry two of them — `ppradhan_replica_ppradhan_database2olq`. That is ugly
     * but harmless; what is NOT harmless is silently exceeding MySQL's 64-char
     * limit, which would fail the clone outright.
     *
     * When the natural name does not fit, it is compressed to a readable head
     * plus a digest of the FULL source name. The digest keeps it deterministic —
     * the same source must always map to the same replica, or every run would
     * create a new database instead of refreshing the existing one.
     */
    public static function replicaName(string $dbName, int $reserve = 0): string
    {
        $clean  = (string) preg_replace('/[^A-Za-z0-9_]/', '', $dbName);
        $name   = self::REPLICA_PREFIX . $clean;
        $budget = self::MAX_DB_NAME - max(0, $reserve);

        if (strlen($name) <= $budget) return $name;

        $hash = substr(sha1($dbName), 0, 8);
        $head = substr($clean, 0, max(1, $budget - strlen(self::REPLICA_PREFIX) - 9));
        return self::REPLICA_PREFIX . $head . '_' . $hash;
    }

    /** @param array<string,mixed> $conn */
    private static function connect(array $conn): \PDO
    {
        $dsn = sprintf('mysql:host=%s;port=%d;dbname=%s;charset=utf8mb4',
            (string) ($conn['host'] ?? '127.0.0.1'),
            (int) ($conn['port'] ?? 3306),
            (string) ($conn['dbname'] ?? ''));
        return new \PDO($dsn, (string) ($conn['user'] ?? ''), (string) ($conn['pass'] ?? ''), [
            \PDO::ATTR_ERRMODE => \PDO::ERRMODE_EXCEPTION,
            \PDO::ATTR_TIMEOUT => 15,
        ]);
    }
}
