<?php
namespace Relay\Remote;

use Relay\Database\BufferSink;
use Relay\Database\CreateTable;
use Relay\Database\Db;
use Relay\Database\Naming;
use Relay\Database\SiteSwap;
use Relay\Database\TableCopier;
use Relay\Database\TableSchema;
use Relay\Files\FileDiff;
use Relay\Files\FileParts;
use Relay\Files\FilePlacer;
use Relay\Files\Fs;
use Relay\Files\Groups;
use Relay\Files\Inventory;
use Relay\Files\JobFiles;
use Relay\Files\ManifestBuilder;
use Relay\Files\Packer;
use Relay\Jobs\Budget;
use Relay\Jobs\Job;
use Relay\Jobs\JobRepository;
use Relay\Jobs\Runner;
use Relay\Multisite\Scope;
use Relay\Multisite\ShapeOut;
use Relay\Multisite\Subsites;
use Relay\Multisite\SubsiteFiles;
use Relay\Multisite\SubsiteTables;
use Relay\Multisite\UserMerge;
use Relay\Jobs\Steps\SubsiteCreateStep;
use Relay\Jobs\Steps\UserMergeStep;
use Relay\Replace\MapPair;
use Relay\Replace\PairFactory;
use Relay\Replace\Replacer;
use Relay\Restore\SqlGate;
use Relay\Security\KeyRepository;
use Relay\Security\Permissions;
use Relay\Security\SiteMode;
use Relay\Settings;
use Relay\Support\Limits;
use Relay\Support\Paths;
use Relay\Support\RelayError;
use Relay\Support\SiteInfo;

/**
 * What this site does when another site pulls from it or pushes to it (spec §2.5). The other site drives
 * the migration; this site keeps a "receiver" job in its History and does its own swap on a push.
 */
final class Operations
{
    public const PROTOCOL = 1;
    /**
     * What this version can do beyond protocol 1's database operations ("subsites" only on a network, features()).
     * "tables_together": db.write takes several whole tables in one request and db.read answers with the whole
     * small tables after the one asked for, up to TOGETHER_MAX of them.
     */
    public const FEATURES = ['files', 'tables_together'];
    public const TOGETHER_MAX = 100;
    private const READ_MAX = 8388608;
    /** One file-list page; how many file numbers files.read takes; how many needed numbers one files.plan answer holds. */
    private const LIST_PAGE = 4194304;
    private const MAX_WANT = 1000;
    private const NEEDED_PAGE = 20000;

    private Db $db;
    private JobRepository $jobs;
    private TableCopier $copier;
    private Settings $settings;
    private SiteSwap $site;
    private KeyRepository $keys;
    private Runner $runner;

    public function __construct(Db $db, JobRepository $jobs, TableCopier $copier, Settings $settings, SiteSwap $site, KeyRepository $keys, Runner $runner)
    {
        $this->db = $db;
        $this->jobs = $jobs;
        $this->copier = $copier;
        $this->settings = $settings;
        $this->site = $site;
        $this->keys = $keys;
        $this->runner = $runner;
    }

    public function hello(array $key): array
    {
        return [
            'protocol' => self::PROTOCOL,
            'site' => SiteInfo::current($this->db),
            'mode' => SiteMode::current(),
            'permissions' => Permissions::effective((string) $key['permissions'], SiteMode::current()),
            'key' => ['label' => $key['label'], 'permission' => $key['permissions']],
            'features' => self::features(),
            'limits' => ['max_body' => Limits::max_body()],
        ];
    }

    /** What this site can do: a network can also choose one of its subsites to send or receive (spec 2026-10-16 §8). */
    public static function features(): array
    {
        return is_multisite() ? array_merge(self::FEATURES, ['subsites']) : self::FEATURES;
    }

    /**
     * Tables and file inventory: this site's, or on a network those of the subsite $meta['subsite'] in single-site
     * shape (its users as the base prefix's users tables) together with that subsite's details. A network asked
     * without a subsite lists its subsites instead; with $meta['only'] = 'site' it gives just the subsite's details.
     */
    public function describe(array $meta = []): array
    {
        if (!is_multisite()) {
            return ['site' => SiteInfo::current($this->db), 'tables' => $this->table_list($this->db->tables()), 'files' => Inventory::current()];
        }
        $id = is_int($meta['subsite'] ?? null) ? $meta['subsite'] : 0;
        if ($id <= 0) {
            return ['site' => SiteInfo::current($this->db), 'tables' => [], 'files' => null, 'subsites' => $this->subsites()['subsites']];
        }
        $scope = Scope::subsite($id);
        if (($meta['only'] ?? '') === 'site') {
            return ['site' => SiteInfo::of($this->db, $scope, $scope->base_prefix)]; // a pull opening: no sizes or inventory
        }
        $sources = ShapeOut::sources($this->db, $scope);
        $status = $this->db->table_status(array_values(array_filter($sources)));
        $tables = [];
        foreach ($sources as $name => $source) {
            $s = $source !== null ? ($status[$source] ?? ['rows' => 0, 'bytes' => 0]) : ShapeOut::count_users($this->db, $scope, $name);
            $tables[] = ['name' => $name, 'rows' => $s['rows'], 'bytes' => $s['bytes']];
        }
        return ['site' => SiteInfo::of($this->db, $scope, $scope->base_prefix), 'tables' => $tables, 'files' => Inventory::current($scope)];
    }

    /** The network's subsites (none on a single site): id, name, address, whether it is the main site and can be used. */
    public function subsites(): array
    {
        $out = [];
        foreach (Subsites::list() as $s) {
            $out[] = ['id' => $s['id'], 'name' => $s['name'], 'home_url' => $s['url'], 'main' => $s['main'], 'available' => $s['available']];
        }
        return ['subsites' => $out];
    }

    /**
     * Who would leave with what is pulled (spec 2026-10-16 §5.5): this site's users, or on a network those of the
     * subsite $meta['subsite'] — logins and emails only, for the other site to show who joins it.
     */
    public function users_list(array $meta): array
    {
        $scope = is_multisite() ? $this->subsite_of($meta) : Scope::site();
        return ShapeOut::user_list($this->db, $scope);
    }

    /**
     * What a push of these people ({login, email}) into the subsite $meta['subsite'] (or a new one) would do to this
     * network's accounts, changing nothing.
     */
    public function users_preview(array $meta): array
    {
        if (!is_multisite()) {
            return ['merged' => [], 'renamed' => [], 'created' => 0, 'removed' => [], 'replaced' => true];
        }
        $scope = is_array($meta['new_subsite'] ?? null) ? null : $this->subsite_of($meta);
        $users = array_slice(array_values(array_filter((array) ($meta['users'] ?? []), 'is_array')), 0, 5000);
        return UserMerge::preview($this->db, $scope, $users);
    }

    /** @param string[] $names */
    private function table_list(array $names): array
    {
        $tables = [];
        foreach ($this->db->table_status($names) as $name => $s) {
            $tables[] = ['name' => $name, 'rows' => $s['rows'], 'bytes' => $s['bytes']];
        }
        usort($tables, static fn ($a, $b) => strcmp($a['name'], $b['name']));
        return $tables;
    }

    public function open(array $key, array $meta): array
    {
        $direction = (string) ($meta['direction'] ?? '');
        $peer = SiteInfo::from_peer($meta['site'] ?? null);
        $client_job = is_string($meta['client_job'] ?? null) ? $meta['client_job'] : '';
        $active = $this->jobs->active();
        if ($active !== null) {
            // The same migration asking again (its first answer was lost) gets the job it already opened.
            if ($client_job !== '' && $active->role === 'receiver' && $active->key_id === $key['id'] && ($active->settings['client_job'] ?? '') === $client_job) {
                return $this->opened($active);
            }
            /* translators: %s: site URL */
            throw new RelayError('job.active_exists', esc_html(sprintf(__('%s is busy with another SiteRelay job. Try again when it has finished.', 'siterelay'), home_url())), '', true);
        }
        $tables = array_values(array_unique(array_filter((array) ($meta['tables'] ?? []), 'is_string')));
        $parts = FileParts::normalize($meta['files'] ?? null);
        if ($tables === [] && !FileParts::any($parts)) {
            throw new RelayError('remote.invalid_tables', esc_html__('Nothing was chosen to copy.', 'siterelay'));
        }
        if (!in_array($direction, ['pull', 'push'], true)) {
            throw new RelayError('remote.bad_request', esc_html__('Unknown migration direction.', 'siterelay'));
        }
        $mine = $this->db->prefix();
        // On a network the migration is one subsite's (spec 2026-10-16 §8); a push may create it.
        $new = is_multisite() && $direction === 'push' && is_array($meta['new_subsite'] ?? null) ? $this->new_subsite_of($meta['new_subsite']) : null;
        $scope = is_multisite() && $new === null ? $this->subsite_of($meta) : null;
        $skipped = [];
        if ($direction === 'pull') {
            // A subsite leaves in single-site shape: its tables under the base prefix, its users from a copy.
            $sources = $scope !== null ? ShapeOut::sources($this->db, $scope) : array_combine($this->db->tables(), $this->db->tables());
            $unknown = array_diff($tables, array_keys($sources));
            if ($unknown !== []) {
                /* translators: %s: comma-separated table names */
                throw new RelayError('remote.invalid_tables', esc_html(sprintf(__('These tables do not exist on the other site: %s', 'siterelay'), implode(', ', array_slice($unknown, 0, 10)))));
            }
            $names = [];
            foreach ($tables as $t) {
                $names[$t] = $peer['prefix'] . substr($t, strlen($mine));
            }
            $settings = [
                'tables' => $tables,
                'names' => $names,
                'pairs' => self::pairs($meta['pairs'] ?? []),
                'map' => array_filter((array) ($meta['map'] ?? []), static fn ($v, $k) => is_string($v) && is_string($k) && $k !== '', ARRAY_FILTER_USE_BOTH),
                'replace_guids' => !empty($meta['replace_guids']),
                'dest_prefix' => $peer['prefix'],
            ] + ($scope !== null ? ['sources' => array_intersect_key($sources, array_flip($tables))] : []);
        } else {
            if (is_multisite()) {
                // A subsite takes no must-use plugins or other wp-content (spec 2026-10-16 §6.2).
                [$parts, $skipped] = SubsiteFiles::receiving($parts);
                if ($tables === [] && !FileParts::any($parts)) {
                    throw new RelayError('remote.invalid_tables', esc_html__('Nothing was chosen that a subsite takes: must-use plugins and other wp-content are never copied into a network.', 'siterelay'));
                }
            }
            // A new subsite exists from here on; if the push can't be opened after all, it goes again.
            $scope = $new !== null ? Scope::subsite(Subsites::create($new['address'], $new['title'], self::network_owner())) : $scope;
            try {
                $names = $this->push_names($tables, $peer['prefix'], $scope !== null ? $scope->prefix : $mine, $scope);
                if ($tables !== []) {
                    $this->db->check_privileges(substr(md5($client_job . microtime()), 0, 6));
                }
            } catch (\Throwable $e) {
                if ($new !== null) {
                    wp_delete_site($scope->blog_id);
                }
                throw $e;
            }
            $settings = ['tables' => $tables, 'names' => $names, 'src_prefix' => $peer['prefix']] + ($skipped !== [] ? ['files_skipped' => $skipped] : []) + ($new !== null ? ['new_subsite' => $new] : []);
        }
        if ($scope !== null) {
            $settings += ['subsite' => $scope->blog_id, 'subsite_name' => $scope->name, 'subsite_url' => $scope->home_url];
        }
        try {
            $job = $this->jobs->create($direction, $settings + ['files' => $parts, 'client_job' => $client_job], 'receiver');
        } catch (\Throwable $e) {
            if ($new !== null) {
                wp_delete_site($scope->blog_id); // no job will ever undo or cancel it
            }
            throw $e;
        }
        if ($new !== null) {
            update_site_meta($scope->blog_id, SubsiteCreateStep::META, $job->id);
            $job->set_cursor('subsite:create', ['created' => $scope->blog_id]); // undo, cancel and failure delete it
        }
        if ($scope !== null && in_array(null, (array) ($settings['sources'] ?? []), true)) {
            ShapeOut::materialize($this->db, $scope, $job->tag()); // the subsite's own users, read by db.read
        }
        $job->direction = $direction;
        $job->peer_url = $peer['home_url'];
        $job->key_id = (string) $key['id'];
        $job->key_label = (string) $key['label'];
        $job->status = Job::RUNNING;
        $job->started_at = gmdate('Y-m-d H:i:s');
        $job->last_tick_at = $job->started_at;
        $job->progress['total'] = 1;
        $job->progress['current'] = $direction === 'pull' ? __('Being pulled by the other site', 'siterelay') : __('Receiving a push from the other site', 'siterelay');
        $this->jobs->save($job);
        $this->jobs->event(
            $job->id,
            'info',
            'incoming.' . $direction,
            $direction === 'pull'
                /* translators: 1: other site's URL, 2: access key label */
                ? sprintf(__('Incoming pull from %1$s with the key "%2$s".', 'siterelay'), $peer['home_url'], $key['label'])
                /* translators: 1: other site's URL, 2: access key label */
                : sprintf(__('Incoming push from %1$s with the key "%2$s".', 'siterelay'), $peer['home_url'], $key['label'])
        );
        if ($key['max_uses'] !== null) {
            $this->keys->bind((string) $key['id'], $job->id);
        }
        $this->keys->record_use((string) $key['id'], $peer['home_url']);
        return $this->opened($job);
    }

    /**
     * One chunk of a table (pull), with the destination's replacements applied here at the source. When the table
     * is done and $meta['more'] names the tables after it, the whole small ones among them that fit come along:
     * "bytes" is then the asked table's part of the payload, and "more" the others' (table, rows, bytes).
     *
     * @return array{0: array, 1: string}
     */
    public function read(Job $job, array $meta): array
    {
        $table = (string) ($meta['table'] ?? '');
        if (!in_array($table, $job->settings['tables'], true)) {
            throw self::out_of_scope();
        }
        $budget = new Budget(10.0, max(65536, min(self::READ_MAX, (int) ($meta['bytes'] ?? $this->settings->get('request_size')))));
        $pairs = PairFactory::from_settings($job->settings['pairs']);
        if (!empty($job->settings['map'])) {
            array_unshift($pairs, new MapPair($job->settings['map']));
        }
        $replacer = $pairs !== [] ? new Replacer($pairs, null, !empty($job->settings['replace_guids'])) : null;
        $sink = new BufferSink();
        $cursor = $this->read_table($job, $table, $sink, $budget, (array) ($meta['cursor'] ?? []), $replacer);
        $answer = ['cursor' => $cursor, 'done' => $cursor['phase'] === 'done', 'rows' => $cursor['rows']];
        $payload = $sink->contents();
        if ($answer['done'] && is_array($meta['more'] ?? null)) {
            $answer['bytes'] = strlen($payload);
            $answer['more'] = [];
            $more = array_values(array_intersect(array_slice(array_values(array_filter($meta['more'], 'is_string')), 0, self::TOGETHER_MAX), $job->settings['tables']));
            $sizes = $this->db->table_status(array_map(static fn (string $t): string => self::source($job, $t), $more));
            $misses = 0;
            foreach ($more as $next) {
                $room = $budget->remaining_bytes();
                if ($budget->exhausted() || $room < 1024 || $misses >= 3) {
                    break;
                }
                if (($sizes[self::source($job, $next)]['bytes'] ?? 0) > 8 * $room) {
                    continue; // clearly too big to come whole: it is read on its own
                }
                $piece = new BufferSink();
                $part = $budget->with_bytes($room);
                if ($this->read_table($job, $next, $piece, $part, [], $replacer)['phase'] !== 'done' || $piece->bytes() > $room) {
                    $misses++;
                    continue;
                }
                $budget->consume($part->used());
                $answer['more'][] = ['table' => $next, 'rows' => (int) $job->stats['tables'][$next]['rows'], 'bytes' => $piece->bytes()];
                $payload .= $piece->contents();
            }
        }
        $this->touch($job);
        return [$answer, $payload];
    }

    /** Copies $table's rows from $cursor into $sink as far as $budget goes, with its constraints once it is done. */
    private function read_table(Job $job, string $table, BufferSink $sink, Budget $budget, array $cursor, ?Replacer $replacer): array
    {
        $source = self::source($job, $table);
        $schema = TableSchema::load($this->db, $source);
        $rewrite = ShapeOut::rewrite(Scope::for_settings($job->settings), $job->settings['names'][$table], $job->settings['dest_prefix']);
        $cursor = $this->copier->copy($schema, $table, $table, $sink, $budget, TableCopier::clean_cursor($schema, $cursor), $replacer, true, $rewrite);
        if ($cursor['phase'] === 'done') {
            $constraints = CreateTable::build($this->db, $source, $table, $table)['constraints'];
            if ($constraints !== null) {
                $sink->statement($constraints);
            }
        }
        $cursor['constraints'] = null;
        $job->stats['tables'][$table] = ['rows' => $cursor['rows']];
        return $cursor;
    }

    /** Where a pulled table's rows are: the table itself, a subsite's own table, or the job's copy of its users. */
    private static function source(Job $job, string $table): string
    {
        $sources = (array) ($job->settings['sources'] ?? []);
        if (!array_key_exists($table, $sources)) {
            return $table;
        }
        return is_string($sources[$table]) ? $sources[$table] : Naming::shape($table, $job->tag());
    }

    /**
     * One chunk of SQL for a table (push), or several whole small tables ($meta['pieces']), written into this
     * site's shadow tables through the SQL gate. Chunks are numbered; a chunk that was already applied (its answer
     * got lost) is acknowledged without running it again, and a chunk's rows are committed together with its number.
     */
    public function write(Job $job, array $meta, string $payload): array
    {
        if (!array_key_exists('pieces', $meta)) {
            return [$this->write_piece($job, (string) ($meta['table'] ?? ''), (int) ($meta['seq'] ?? 0), (int) ($meta['rows'] ?? 0), $payload), ''];
        }
        // Several tables in one request ("tables_together"): each piece its table's SQL, in turn.
        $pieces = is_array($meta['pieces']) ? array_values($meta['pieces']) : [];
        $lengths = array_map(static fn ($p): int => is_array($p) && is_int($p['bytes'] ?? null) ? $p['bytes'] : -1, $pieces);
        if ($pieces === [] || count($pieces) > self::TOGETHER_MAX + 1 || min($lengths) < 0 || array_sum($lengths) !== strlen($payload)) {
            throw new RelayError('remote.bad_pieces', esc_html__('The other site sent tables whose sizes do not add up.', 'siterelay'), '', false, ['status' => 400]);
        }
        $answers = [];
        $at = 0;
        foreach ($pieces as $i => $p) {
            $answers[] = $this->write_piece($job, (string) ($p['table'] ?? ''), (int) ($p['seq'] ?? 0), (int) ($p['rows'] ?? 0), substr($payload, $at, $lengths[$i]));
            $at += $lengths[$i];
        }
        return [['pieces' => $answers], ''];
    }

    /** @return array{received: int, duplicate?: bool} */
    private function write_piece(Job $job, string $table, int $seq, int $rows, string $payload): array
    {
        if (!isset($job->settings['names'][$table])) {
            throw self::out_of_scope();
        }
        $last = (int) ($job->progress['seq'][$table] ?? 0);
        if ($seq <= $last) {
            // Already applied (its answer got lost). The resent piece may be cut differently, so the other site
            // is told, and sends the table again from the start.
            return ['received' => $last, 'duplicate' => true];
        }
        if ($seq !== $last + 1) {
            throw new RelayError('remote.out_of_order', esc_html__('A piece of the push arrived out of order.', 'siterelay'), '', true, ['status' => 409]);
        }
        $import = [];
        foreach ($job->settings['names'] as $source => $dest) {
            $import[$source] = Naming::temp($dest);
        }
        $gate = new SqlGate($import, $job->settings['names'], $this->db->collations(), true);
        $in_transaction = false;
        $this->db->raw('SET FOREIGN_KEY_CHECKS = 0');
        try {
            foreach (SqlGate::split($payload) as $statement) {
                $checked = $gate->check($statement);
                $own = $checked['kind'] === 'constraint' ? $job->settings['names'][$table] : $table;
                if ($checked['table'] !== null && $checked['table'] !== $own) {
                    /* translators: %s: start of the rejected SQL statement */
                    throw new RelayError('restore.statement_rejected', esc_html(sprintf(__('The other site sent SQL SiteRelay will not run: %s', 'siterelay'), substr($statement, 0, 100))));
                }
                if ($checked['kind'] === 'constraint') {
                    $job->progress['constraints'][$own] = $checked['sql'];
                    continue;
                }
                if ($checked['kind'] === 'insert' && !$in_transaction) {
                    $this->db->raw('START TRANSACTION');
                    $in_transaction = true;
                }
                $this->db->raw($checked['sql']);
            }
            $job->progress['seq'][$table] = $seq;
            $job->stats['tables'][$table] = ['rows' => max(0, $rows)];
            $this->touch($job);
            if ($in_transaction) {
                $this->db->raw('COMMIT');
                $in_transaction = false;
            }
        } finally {
            if ($in_transaction) {
                $this->db->raw('ROLLBACK');
            }
            $this->db->raw('SET FOREIGN_KEY_CHECKS = 1');
        }
        return ['received' => $seq];
    }

    /**
     * Pull: the other site has everything, the job is closed. Push: received uploads go live first (this may
     * take several calls, answered with done: false), then themes, plugins and the tables in one go; the site
     * is loaded and put back if it doesn't load.
     */
    /** @param array<string, mixed> $meta "copied": the files the other site copied, for this site's History */
    public function finalize(Job $job, array $meta = []): array
    {
        if (is_array($meta['copied'] ?? null)) {
            $job->stats['transfer']['copied'] = ['count' => max(0, (int) ($meta['copied']['count'] ?? 0)), 'bytes' => max(0, (int) ($meta['copied']['bytes'] ?? 0)), 'gone' => max(0, (int) ($meta['copied']['gone'] ?? 0))];
        }
        if ($job->direction === 'pull') {
            $this->complete($job);
            self::discard($this->db, $job);
            return ['done' => true];
        }
        $names = array_values($job->settings['names']);
        foreach ($names as $dest) {
            if (!$this->db->table_exists(Naming::temp($dest))) {
                /* translators: %s: table name */
                throw new RelayError('remote.incomplete', esc_html(sprintf(__('The push is not complete: %s was never received.', 'siterelay'), $dest)));
            }
        }
        $parts = self::parts_of($job);
        $placer = null;
        if (FileParts::any($parts)) {
            if (empty($job->progress['diff']['done'])) {
                throw new RelayError('remote.incomplete', esc_html__('The push is not complete: the files were never received.', 'siterelay'));
            }
            $placer = JobFiles::placer($job);
            $cursor = $job->progress['place'] ?? FilePlacer::start();
            if (empty($cursor['done'])) {
                $cursor = $placer->uploads($cursor, self::work_budget(), $parts['low_disk']);
                $job->progress['place'] = $cursor;
                $this->touch($job);
                if (empty($cursor['done'])) {
                    return ['done' => false, 'placed' => $cursor['placed']];
                }
            }
        }
        // Into a subsite (spec 2026-10-16 §3.3, §5.2): it keeps its address, and the pushed users join the network
        // right before the swap; only the subsite's own tables go live.
        $scope = Scope::for_settings($job->settings);
        $db = $this->db;
        $put_back = static function () use ($job, $db): void {
            JobFiles::put_back($job);
            UserMergeStep::put_back($db, $job);
        };
        try {
            $this->site->carry_options($scope->prefix, $names);
            if ($scope->is_subsite()) {
                $this->site->keep_address($scope->prefix, $names, $scope);
                $names = array_values(UserMergeStep::merge($this->db, $job, $scope, $job->settings['names']));
            }
            $moved = $placer !== null ? $placer->code() : 0;
            $point = $this->site->swap()->swap_in($names, $job->tag(), (array) ($job->progress['constraints'] ?? []));
            $warning = $this->site->check($job, $point, $put_back, __('A plugin or theme in what was pushed probably needs a newer PHP version or a server setting this site lacks.', 'siterelay'), $scope);
        } catch (\Throwable $e) {
            // Any error, not only Relay's, fails the job now: left running, it would wait for the daily sweep.
            $e = $e instanceof RelayError ? $e : new RelayError(
                'job.unexpected',
                esc_html($e->getMessage()),
                esc_html__('This looks like a bug in SiteRelay. Download the job report and share it with the developers.', 'siterelay'),
                false,
                ['where' => basename($e->getFile()) . ':' . $e->getLine()]
            );
            if ($e->error_code() !== 'restore.site_broken') {
                $put_back();
            }
            $this->fail($job, $e);
            self::discard($this->db, $job); // what it received and, on a network, a subsite it created
            throw $e;
        }
        SubsiteFiles::enable_theme($scope); // after the check, which cleared the subsite's cached options
        $files = JobFiles::point($job);
        if ($files !== null) {
            $point['files'] = $files;
        }
        $this->site->settle($job, SubsiteCreateStep::settle($job, UserMergeStep::settle($this->db, $job, $point)), $scope);
        if ($placer !== null) {
            Fs::remove_tree(Paths::staging_dir($job->id));
        }
        $locked = SiteMode::after_incoming_push();
        $job->stats['swapped'] = $names;
        if ($placer !== null) {
            $job->stats['transfer']['placed'] = (int) ($job->progress['place']['placed'] ?? 0) + $moved;
        }
        if ($warning !== null) {
            $job->stats['warnings'][] = $warning;
        }
        if ($locked) {
            $job->stats['locked_after'] = true;
        }
        $this->complete($job);
        return ['done' => true, 'restore_point' => true, 'files' => isset($point['files']), 'tables' => count($names), 'placed' => (int) ($job->stats['transfer']['placed'] ?? 0), 'warning' => $warning, 'locked' => $locked]
            + (isset($job->stats['users']) ? ['users' => $job->stats['users']] : [])
            + (!empty($job->restore_point['created_subsite']) ? ['created_subsite' => ['id' => (int) $job->restore_point['created_subsite'], 'url' => (string) ($job->settings['subsite_url'] ?? '')]] : []);
    }

    public function cancel(Job $job): array
    {
        if ($job->is_finished()) {
            return ['status' => $job->status];
        }
        self::discard($this->db, $job);
        $job->status = Job::CANCELLED;
        $job->finished_at = gmdate('Y-m-d H:i:s');
        $job->progress['current'] = '';
        $this->jobs->save($job);
        $this->jobs->event($job->id, 'info', 'job.cancelled', __('Cancelled by the other site. Temporary data was removed.', 'siterelay'));
        do_action('relay_job_finished', $job);
        return ['status' => $job->status];
    }

    /** Undo an incoming push from the pushing site's History (spec §3.3). */
    public function restore(array $key, array $meta): array
    {
        $source = $this->jobs->find(is_string($meta['job_id'] ?? null) ? $meta['job_id'] : '');
        if ($source === null || $source->role !== 'receiver' || $source->direction !== 'push' || $source->key_id !== $key['id']) {
            throw self::out_of_scope();
        }
        if ($this->jobs->active() !== null) {
            throw new RelayError('job.active_exists', esc_html__('This site is busy with another SiteRelay job.', 'siterelay'), '', true);
        }
        $job = $this->jobs->create('restore', ['source_job' => $source->id]);
        $job->peer_url = $source->peer_url;
        $job->key_id = $source->key_id;
        $job->key_label = $source->key_label;
        $this->jobs->save($job);
        for ($i = 0; $i < 3 && !$job->is_finished(); $i++) {
            $job = $this->runner->tick($job, new Budget(60.0, PHP_INT_MAX));
        }
        return ['status' => $job->status, 'error' => $job->error, 'restored' => $job->stats['restored'] ?? []] + (isset($job->stats['deleted_subsite']) ? ['deleted_subsite' => $job->stats['deleted_subsite']] : []);
    }

    /**
     * Throw away what a receiver job received and was not finalized: shadow tables, uploads already placed
     * (their moves are undone), staged files and the job's working files. Used on cancel and abandon.
     */
    public static function discard(Db $db, Job $job): void
    {
        if ($job->direction === 'push') {
            foreach ((array) ($job->settings['names'] ?? []) as $dest) {
                $db->raw('DROP TABLE IF EXISTS ' . Db::ident(Naming::temp((string) $dest)));
            }
            if ($job->restore_point === null) {
                if (JobFiles::any($job)) {
                    JobFiles::put_back($job);
                }
                // Nothing went live: users merged at finalize leave the network, a subsite created for it goes.
                UserMergeStep::put_back($db, $job);
                $created = (int) ($job->cursor('subsite:create')['created'] ?? 0);
                if ($created > 0 && get_site($created) !== null) {
                    wp_delete_site($created);
                }
            }
        }
        if ($job->direction === 'pull' && !empty($job->settings['sources'])) {
            ShapeOut::drop($db, $job->tag());
        }
        if (JobFiles::any($job)) {
            Fs::remove_tree(Paths::staging_dir($job->id));
        }
    }

    /** Pull: this site lists its files, as many requests as it takes; then the list is sent page by page. @return array{0: array, 1: string} */
    public function files_manifest(Job $job, array $meta): array
    {
        $parts = self::parts_of($job, true);
        $scope = Scope::for_settings($job->settings);
        $builder = new ManifestBuilder(Groups::roots($scope), $parts, Paths::job_dir($job->id), $scope);
        $cursor = $job->progress['manifest'] ?? ManifestBuilder::start();
        if (empty($cursor['done'])) {
            $cursor = $builder->run($cursor, self::work_budget());
            $job->progress['manifest'] = $cursor;
            $this->touch($job);
            if (empty($cursor['done'])) {
                return [['done' => false, 'count' => $cursor['count']], ''];
            }
        }
        $list = $builder->list();
        $page = $list->read(max(0, (int) ($meta['from'] ?? 0)), self::LIST_PAGE);
        return [['done' => true, 'total' => $list->count(), 'next' => $page['next'], 'files' => $cursor['files'], 'bytes' => $cursor['bytes'], 'skipped' => $cursor['skipped']], $page['lines']];
    }

    /** Pull: a batch of listed files, asked for by number only. @return array{0: array, 1: string} */
    public function files_read(Job $job, array $meta): array
    {
        self::parts_of($job, true);
        if (empty($job->progress['manifest']['done'])) {
            throw self::out_of_scope();
        }
        $list = JobFiles::list($job);
        $count = $list->count();
        $want = [];
        foreach ((array) ($meta['want'] ?? []) as $i) {
            if (!is_int($i) || $i < 0 || $i >= $count) {
                throw self::out_of_scope();
            }
            $want[] = $i;
        }
        if ($want === [] || count($want) > self::MAX_WANT) {
            throw new RelayError('remote.bad_request', esc_html__('The request asked for no files or too many.', 'siterelay'));
        }
        $max = max(65536, min(Limits::max_body(), self::READ_MAX, (int) ($meta['max'] ?? 1048576))); // an answer in JSON is a third bigger: 8 MB of files keeps it light on memory
        $batch = Packer::pack($list, Groups::roots(Scope::for_settings($job->settings)), $want, max(0, (int) ($meta['off'] ?? 0)), $max);
        $job->stats['transfer']['bytes'] = (int) ($job->stats['transfer']['bytes'] ?? 0) + strlen($batch['payload']);
        $this->touch($job);
        return [['sent' => $batch['sent'], 'off' => $batch['off']], $batch['payload']];
    }

    /**
     * Push, before any file is sent: the other site's file list arrives page by page ("manifest"), this site
     * works out what it needs ("diff", as many calls as it takes, then checks its free disk space) and
     * returns the needed numbers page by page ("needed"). @return array{0: array, 1: string}
     */
    public function files_plan(Job $job, array $meta, string $payload): array
    {
        $parts = self::parts_of($job, true);
        $list = JobFiles::list($job);
        $diff = JobFiles::diff($job);
        switch ((string) ($meta['phase'] ?? '')) {
            case 'manifest':
                if (isset($job->progress['diff'])) {
                    throw self::out_of_order();
                }
                $count = $list->import($payload, $parts, Groups::protected_items(JobFiles::roots($job)));
                $this->touch($job);
                return [['count' => $count], ''];
            case 'diff':
                if ((int) ($meta['total'] ?? -1) !== $list->count()) {
                    throw self::out_of_order();
                }
                $cursor = $job->progress['diff'] ?? FileDiff::start();
                if (empty($cursor['done'])) {
                    $cursor = $diff->run($cursor, self::work_budget());
                    $job->progress['diff'] = $cursor;
                    if (!empty($cursor['done'])) {
                        Limits::check_disk_space((int) $cursor['bytes'], !empty($parts['low_disk']));
                    }
                    $this->touch($job);
                }
                $kept = [];
                foreach ((array) ($cursor['kept'] ?? []) as $group => $items) {
                    $kept[$group] = array_map('strval', array_keys((array) $items)); // a subsite keeps the network's copies
                }
                return [['done' => !empty($cursor['done']), 'needed' => $cursor['needed'], 'bytes' => $cursor['bytes'], 'extras' => $cursor['extras'], 'kept' => $kept], ''];
            case 'needed':
                if (empty($job->progress['diff']['done'])) {
                    throw self::out_of_order();
                }
                $from = max(0, (int) ($meta['from'] ?? 0));
                $indexes = $diff->needed()->slice($from, self::NEEDED_PAGE);
                return [['indexes' => $indexes, 'next' => $from + count($indexes), 'total' => (int) $job->progress['diff']['needed']], ''];
            default:
                throw new RelayError('remote.bad_request', esc_html__('The request is not a valid SiteRelay request.', 'siterelay'));
        }
    }

    /** Push: a batch of files into this job's staging folder (uploads in place in low-disk mode). @return array{0: array, 1: string} */
    public function files_write(Job $job, string $payload): array
    {
        self::parts_of($job, true);
        if (empty($job->progress['diff']['done'])) {
            throw self::out_of_order();
        }
        $result = JobFiles::unpacker($job)->unpack($payload);
        $job->stats['transfer']['count'] = (int) ($job->stats['transfer']['count'] ?? 0) + count($result['done']);
        $job->stats['transfer']['bytes'] = (int) ($job->stats['transfer']['bytes'] ?? 0) + $result['bytes'];
        $this->touch($job);
        return [$result, ''];
    }

    /** The job's file parts; with $required, a job without files refuses file operations. */
    private static function parts_of(Job $job, bool $required = false): array
    {
        $parts = FileParts::normalize($job->settings['files'] ?? null);
        if ($required && !FileParts::any($parts)) {
            throw self::out_of_scope();
        }
        return $parts;
    }

    /** Time for one request's file work, well inside the other site's 30-second wait. */
    private static function work_budget(): Budget
    {
        $max = (int) ini_get('max_execution_time');
        return new Budget($max > 0 ? min(10.0, $max * 0.5) : 10.0, PHP_INT_MAX);
    }

    private function opened(Job $job): array
    {
        $out = ['job_id' => $job->id, 'names' => $job->settings['names'], 'site' => SiteInfo::current($this->db)];
        if (!empty($job->settings['subsite'])) {
            $scope = Scope::subsite((int) $job->settings['subsite']);
            // A push lands under the subsite's own prefix; a pull leaves in single-site shape.
            $out['site'] = SiteInfo::of($this->db, $scope, $job->direction === 'push' ? $scope->prefix : $scope->base_prefix);
            $out['files'] = FileParts::normalize($job->settings['files'] ?? null);
            $out['files_skipped'] = (array) ($job->settings['files_skipped'] ?? []);
        }
        return $out;
    }

    /**
     * Where pushed tables land: this site's prefix (on a network, the subsite's) in place of the other site's. Relay's
     * own tables are never written, nor — into a subsite — another subsite's or the network's.
     *
     * @return array<string, string>
     */
    private function push_names(array $tables, string $from, string $to, ?Scope $scope): array
    {
        $names = [];
        foreach ($tables as $t) {
            $dest = $to . substr($t, strlen($from));
            if (!preg_match('/^[A-Za-z0-9_$-]{1,64}$/', $t) || strpos($t, $from) !== 0 || Naming::is_internal($dest) || strlen($dest) > 64 || ($scope !== null && !SubsiteTables::accepts($scope, $dest))) {
                /* translators: %s: table name */
                throw new RelayError('remote.invalid_tables', esc_html(sprintf(__('SiteRelay will not write the table %s here.', 'siterelay'), $t)));
            }
            $names[$t] = $dest;
        }
        return $names;
    }

    /**
     * A new subsite asked for by a push: its address checked now (valid here and free).
     *
     * @return array{address: string, title: string}
     */
    private function new_subsite_of(array $new): array
    {
        $where = Subsites::parse_address(is_string($new['address'] ?? null) ? $new['address'] : '');
        return ['address' => $where['domain'] . $where['path'], 'title' => mb_substr(sanitize_text_field(is_string($new['title'] ?? null) ? $new['title'] : ''), 0, 190)];
    }

    /** Who owns a subsite another site's push creates: the network's first super admin. */
    private static function network_owner(): int
    {
        foreach ((array) get_super_admins() as $login) {
            $user = get_user_by('login', (string) $login);
            if ($user !== false) {
                return (int) $user->ID;
            }
        }
        return 0;
    }

    /** The subsite a migration on this network is for: one that exists and can be used. */
    private function subsite_of(array $meta): Scope
    {
        $id = is_int($meta['subsite'] ?? null) ? $meta['subsite'] : 0;
        if ($id <= 0) {
            throw new RelayError('subsite.required', esc_html__('This site is a network: choose the subsite to copy from or to.', 'siterelay'));
        }
        return Scope::subsite($id);
    }

    private function touch(Job $job): void
    {
        $job->last_tick_at = gmdate('Y-m-d H:i:s');
        $this->jobs->save($job);
    }

    private function complete(Job $job): void
    {
        $job->status = Job::COMPLETED;
        $job->finished_at = gmdate('Y-m-d H:i:s');
        $job->progress['step'] = 1;
        $job->progress['percent'] = 100;
        $job->progress['current'] = '';
        $this->jobs->save($job);
        $this->jobs->event($job->id, 'info', 'job.completed', __('Job completed.', 'siterelay'));
        do_action('relay_job_finished', $job);
    }

    private function fail(Job $job, RelayError $e): void
    {
        $job->status = Job::FAILED;
        $job->error = $e->to_array();
        $job->finished_at = gmdate('Y-m-d H:i:s');
        $job->progress['current'] = '';
        $this->jobs->save($job);
        $this->jobs->event($job->id, 'error', $e->error_code(), $e->text());
        do_action('relay_job_finished', $job);
    }

    /**
     * Find & replace pairs sent by the other site, checked like local ones (invalid regex is refused).
     *
     * @param mixed $rows as the other site sent them
     */
    private static function pairs($rows): array
    {
        $clean = [];
        foreach (is_array($rows) ? $rows : [] as $row) {
            if (!is_array($row) || !is_string($row['find'] ?? null) || $row['find'] === '') {
                continue;
            }
            $clean[] = ['find' => $row['find'], 'replace' => (string) ($row['replace'] ?? ''), 'regex' => !empty($row['regex']), 'case_sensitive' => !empty($row['case_sensitive'])];
        }
        PairFactory::from_settings($clean);
        return $clean;
    }

    private static function out_of_scope(): RelayError
    {
        return new RelayError('remote.out_of_scope', esc_html__('That request is outside the migration the other site opened.', 'siterelay'), '', false, ['status' => 403]);
    }

    private static function out_of_order(): RelayError
    {
        return new RelayError('remote.out_of_order', esc_html__('A piece of the migration arrived out of order.', 'siterelay'), '', false, ['status' => 409]);
    }
}