<?php
namespace Relay\Database;

use Relay\Jobs\Budget;
use Relay\Replace\Replacer;

final class TableCopier
{
    public const STATEMENT_BYTES = 50000;

    private Db $db;
    private RowReader $reader;

    public function __construct(Db $db, RowReader $reader)
    {
        $this->db = $db;
        $this->reader = $reader;
    }

    /**
     * A cursor that came back from another site: only the known fields, in the expected shape. The
     * constraints are recomputed here rather than trusted.
     */
    public static function clean_cursor(TableSchema $s, array $cursor): array
    {
        $phase = in_array($cursor['phase'] ?? 'start', ['start', 'rows', 'done'], true) ? $cursor['phase'] ?? 'start' : 'start';
        $read = [];
        if ($s->key !== [] && is_array($cursor['read']['after'] ?? null)) {
            foreach ($s->key as $col) {
                if (!array_key_exists($col, $cursor['read']['after']) || !is_scalar($cursor['read']['after'][$col])) {
                    $read = [];
                    break;
                }
                $read['after'][$col] = (string) $cursor['read']['after'][$col];
            }
        } elseif ($s->key === [] && isset($cursor['read']['offset'])) {
            $read['offset'] = max(0, (int) $cursor['read']['offset']);
        }
        return ['phase' => $phase, 'read' => $read, 'rows' => max(0, (int) ($cursor['rows'] ?? 0)), 'avg' => max(0.0, (float) ($cursor['avg'] ?? 0)), 'constraints' => null];
    }

    private static function defaults(array $cursor): array
    {
        return $cursor + ['phase' => 'start', 'read' => [], 'rows' => 0, 'avg' => 0.0, 'constraints' => null];
    }

    /**
     * Copy rows of $s into $target through $sink until done or the budget runs out. Returns the new cursor.
     * $rewrite (column, value, row) => value runs on string values before the replacer (e.g. prefix-bearing keys).
     */
    public function copy(TableSchema $s, string $target, string $final, SqlSink $sink, Budget $budget, array $cursor, ?Replacer $replacer = null, bool $with_structure = true, ?callable $rewrite = null): array
    {
        $cursor = self::defaults($cursor);
        if ($cursor['phase'] === 'start') {
            if ($with_structure) {
                $ct = CreateTable::build($this->db, $s->name, $target, $final);
                $sink->statement('DROP TABLE IF EXISTS ' . Db::ident($target));
                $sink->statement($ct['create']);
                $cursor['constraints'] = $ct['constraints'];
            }
            $cursor['phase'] = 'rows';
        }

        $head = 'INSERT INTO ' . Db::ident($target) . ' ' . RowEncoder::columns($s) . ' VALUES ';
        $buffer = '';
        while ($cursor['phase'] === 'rows' && !$budget->exhausted()) {
            $limit = $cursor['avg'] > 0 ? (int) floor(($budget->remaining_bytes() - strlen($buffer)) / $cursor['avg']) : RowReader::PAGE;
            $page = $this->reader->page($s, $cursor['read'], max(1, $limit));
            $page_bytes = 0;
            foreach ($page['rows'] as $row) {
                $tuple = RowEncoder::values($row, $s, $this->transform($replacer, $s, $row, $rewrite));
                $page_bytes += strlen($tuple);
                if ($buffer !== '' && strlen($buffer) + strlen($tuple) + 1 > self::STATEMENT_BYTES) {
                    $this->flush($sink, $budget, $buffer);
                    $buffer = '';
                }
                $buffer .= ($buffer === '' ? $head : ',') . $tuple;
                $cursor['rows']++;
            }
            if ($page['rows'] !== []) {
                $cursor['avg'] = $page_bytes / count($page['rows']);
            }
            $cursor['read'] = $page['cursor'];
            if ($page['done']) {
                $cursor['phase'] = 'done';
            }
            if (strlen($buffer) >= $budget->remaining_bytes()) {
                break;
            }
        }
        if ($buffer !== '') {
            $this->flush($sink, $budget, $buffer);
        }
        return $cursor;
    }

    /** Undo shadow-table writes made after $cursor was saved. Keyless tables restart from scratch. */
    public function recover_table(TableSchema $s, string $target, array $cursor): array
    {
        $cursor = self::defaults($cursor);
        if ($cursor['phase'] === 'start') {
            return $cursor;
        }
        if ($s->key === [] || !$this->db->table_exists($target)) {
            return ['phase' => 'start', 'read' => [], 'rows' => 0, 'avg' => 0.0, 'constraints' => null];
        }
        $where = !empty($cursor['read']['after']) ? ' WHERE ' . $this->reader->after_clause($s->key, $cursor['read']['after']) : '';
        $this->db->query('DELETE FROM ' . Db::ident($target) . $where);
        return $cursor;
    }

    private function transform(?Replacer $replacer, TableSchema $s, object $row, ?callable $rewrite): ?callable
    {
        if ($replacer !== null && !$replacer->active()) {
            $replacer = null;
        }
        if ($replacer === null && $rewrite === null) {
            return null;
        }
        $cols = $s->key !== [] ? $s->key : [$s->columns[0]];
        $ids = [];
        foreach ($cols as $c) {
            $ids[] = (string) ($row->$c ?? '');
        }
        $row_id = implode(',', $ids);
        return static function (string $column, string $value) use ($replacer, $rewrite, $s, $row_id, $row): string {
            if ($rewrite !== null) {
                $value = $rewrite($column, $value, $row);
            }
            return $replacer !== null ? $replacer->replace_value($s->name, $column, $row_id, $value) : $value;
        };
    }

    private function flush(SqlSink $sink, Budget $budget, string $sql): void
    {
        $sink->statement($sql);
        $budget->consume(strlen($sql));
    }
}