<?php
// ═══════════════════════════════════════════════════════════════════════════════════════════════
// PURPOSE
//   Partnership page data layer. Enumerates the nodes in the mesh, reaches each over
//   Tailscale, and reports identity, system summary, sync history and reachability — the
//   view of "who is in this partnership and are they alive".
//
// DESIGN PRINCIPLES
//   Tailscale is the transport; hostnames resolve through it.
//       Peers come from `tailscale status`, never from hardcoded IPs. A node that moves
//       networks stays reachable because nothing here records where it used to be.
//
//   The API is tried first, SSH is the fallback.
//       vv_pt_remote_system() prefers the partner's unraid-api and drops to a single
//       combined SSH call for version/uptime/load/containers when the API is unavailable.
//       One round trip either way.
//
//   Every remote read is one command.
//       Partner data is gathered in a single SSH invocation rather than several, because
//       each one pays full connection setup over a WAN link.
//
// OPERATIONAL SAFEGUARDS
//   SSH is time-boxed, non-interactive, and escaped.
//       ConnectTimeout, BatchMode=yes so it can never sit waiting for a password, and every
//       interpolated value passed through escapeshellarg(). A partner that is powered off
//       costs the configured timeout, not a hung page.
//
//   A missing or unreadable key is treated as unreachable.
//       vv_pt_ssh() returns empty immediately when the key path does not exist, rather than
//       invoking ssh and letting it fail slowly.
//
//   Unreachable partners degrade per node.
//       Each node is collected independently; one dark host leaves its own card empty and
//       affects nothing else.
//
//   Read-only over SSH. Commands issued are state reads and inventory — this file never
//   deploys, starts, or stops anything on a partner.
//
// EXPORTS
//   Config    vv_pt_config(), vv_pt_nodes(), vv_pt_ts_peers()
//   Transport vv_pt_ssh(), vv_pt_ping()
//   System    vv_pt_local_system(), vv_pt_remote_system()
//   Sync      vv_pt_sync(), vv_pt_sync_summary(), vv_pt_read_db()
//   Assembly  vv_partnership_all()
//
// CONFIGURATION
//   HOST*_SSH_KEY              per-host key used for every partner call
//   HOST*_UNRAID_API_KEY       preferred path before SSH fallback
//   PARTNERSHIP_ENABLED        whether the partnership layer is active
//   STATE_DIR / DATA_DIR       sync history and offline counters
// ═══════════════════════════════════════════════════════════════════════════════════════════════
// Partnership page data helpers

require_once __DIR__ . '/config.php';
require_once __DIR__ . '/arrs.php';   // vv_arr_known_hosts(), vv_arr_scalar()
require_once __DIR__ . '/common.php'; // vv_system_info(), vv_docker_containers(), vv_remote_hosts_stats(), vv_api_node_metrics()

// ── Config ────────────────────────────────────────────────────────────────────

function vv_pt_config(): array {
    $v = vv_conf_vars();
    // The file is named ..._days.db and does not hold days. partnership_manager.sh --check runs
    // from the 30-minute orchestrator and increments once per run, so the unit is INTERVALS, and
    // the auto-offboard threshold it is compared against is computed as days × 48. Reading the
    // raw number and printing it as days overstated the outage by 48× — a partner unreachable
    // for half an hour rendered as "unreachable 1 day · auto-offboard in 29 days".
    $offlineDays      = null;
    $offlineIntervals = null;
    $odFile = STATE_DIR . '/partnership_offline_days.db';
    if (file_exists($odFile)) {
        $raw = trim(@file_get_contents($odFile) ?: '');
        if (is_numeric($raw)) {
            $offlineIntervals = (int)$raw;
            $offlineDays      = $offlineIntervals / 48;
        }
    }
    return [
        'enabled'           => ($v['PARTNERSHIP_ENABLED']          ?? 'false') === 'true',
        'owner_host'        => $v['PARTNERSHIP_OWNER_HOST']         ?? '',
        'sync_min'          => (int)($v['PARTNERSHIP_SYNC_INTERVAL']   ?? 15),
        'grace_hours'       => (int)($v['PARTNERSHIP_GRACE_HOURS']     ?? 6),
        'offline_threshold' => (int)($v['PARTNERSHIP_OFFLINE_THRESHOLD'] ?? 30),
        'offline_days'      => $offlineDays,
        'offline_intervals' => $offlineIntervals,
        'remove_tailscale'  => ($v['PARTNERSHIP_REMOVE_TAILSCALE']  ?? 'false') === 'true',
        'folderview3'       => ($v['PARTNERSHIP_FOLDERVIEW3']       ?? 'false') === 'true',
        'tailscale_configured' => !empty($v['TAILSCALE_API_KEY']) && !empty($v['TAILSCALE_TAILNET']),
        'transfer_confirm'  => $v['PARTNERSHIP_TRANSFER_CONFIRM'] ?? 'i-understand-this-transfers-ownership',
    ];
}

// ── Mirror sync health — the partnership's actual job (rsync orchestrators) ──────

function vv_pt_sync_summary(string $logFile): string {
    if (!file_exists($logFile)) return '';
    $tail = array_filter(array_slice(file($logFile, FILE_IGNORE_NEW_LINES), -30), 'strlen');
    foreach (array_reverse(array_values($tail)) as $raw) {
        // Strip emoji / Unicode decoration for regex matching
        $line = trim(preg_replace('/[\x{1F000}-\x{1FFFF}\x{2600}-\x{27BF}\x{FE0F}]/u', '', $raw));
        $line = preg_replace('/\s+/', ' ', $line);
        // "Critical sync complete — HOST1 — 1m39s — 2 share(s)"
        if (preg_match('/Critical sync complete\s*—\s*\S+\s*—\s*([\w]+)\s*—\s*(.+)/i', $line, $m))
            return trim($m[2]) . ' in ' . $m[1];
        // "Status: all complete — 0 share(s) synced, 11 job(s) run"
        if (preg_match('/Status:\s*all complete\s*—\s*(.+)/i', $line, $m))
            return trim($m[1]);
        // "Status: X failure(s)" or "Failures: N"
        if (preg_match('/Failures:\s*(\d+)/i', $line, $m) && (int)$m[1] > 0)
            return (int)$m[1] . ' failure(s)';
    }
    return '';
}

function vv_pt_sync(): array {
    $jobs = [
        'critical' => 'Orchestrators/critical_sync_maintenance',
        'daily'    => 'Orchestrators/daily_sync_maintenance',
        'weekly'   => 'Orchestrators/weekly_sync_maintenance',
    ];
    $out = ['jobs' => []];
    foreach ($jobs as $key => $base) {
        $statFile = LOG_DIR . '/' . $base . '.json';
        $logFile  = LOG_DIR . '/' . $base . '.log';
        $s = file_exists($statFile) ? json_decode(@file_get_contents($statFile), true) : null;
        $out['jobs'][$key] = is_array($s) ? [
            'status'  => $s['status'] ?? 'unknown',
            'start'   => isset($s['start']) ? (int)$s['start'] : null,
            'end'     => isset($s['end'])   ? (int)$s['end']   : null,
            'summary' => vv_pt_sync_summary($logFile),
        ] : null;
    }
    $v = vv_conf_vars();
    $out['interval_min'] = (int)($v['PARTNERSHIP_SYNC_INTERVAL'] ?? 30);
    // Rsync gate flags (master.conf) — global Tier 1 + per-tier Tier 2.
    $out['gates'] = [
        'global'   => ['var' => 'RSYNC_ENABLED',          'on' => ($v['RSYNC_ENABLED']          ?? 'true') === 'true'],
        'critical' => ['var' => 'CRITICAL_RSYNC_ENABLED', 'on' => ($v['CRITICAL_RSYNC_ENABLED'] ?? 'true') === 'true'],
        'daily'    => ['var' => 'DAILY_RSYNC_ENABLED',    'on' => ($v['DAILY_RSYNC_ENABLED']    ?? 'true') === 'true'],
        'weekly'   => ['var' => 'WEEKLY_RSYNC_ENABLED',   'on' => ($v['WEEKLY_RSYNC_ENABLED']   ?? 'true') === 'true'],
    ];
    // Back-compat keys still used by the warning line.
    $out['rsync_enabled']    = $out['gates']['global']['on'];
    $out['critical_enabled'] = $out['gates']['critical']['on'];
    return $out;
}

// ── State file parser ─────────────────────────────────────────────────────────

function vv_pt_read_db(string $path): array {
    if (!file_exists($path)) return [];
    $out = [];
    foreach (file($path, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) as $line) {
        [$k, $v] = array_pad(explode('=', trim($line), 2), 2, '');
        if ($k) $out[trim($k)] = trim($v, '"\'');
    }
    return $out;
}

// ── Tailscale peers ───────────────────────────────────────────────────────────

function vv_pt_ts_peers(): array {
    $raw   = shell_exec('tailscale status --json 2>/dev/null') ?: '{}';
    $data  = json_decode($raw, true) ?: [];
    $peers = [];

    // Self
    $self = $data['Self'] ?? [];
    $selfLabel = strtolower(explode('.', $self['DNSName'] ?? '')[0]);
    if ($selfLabel) {
        $peers[$selfLabel] = [
            'online' => true,
            'active' => true,
            'ip'     => $self['TailscaleIPs'][0] ?? null,
        ];
    }

    // Peers
    foreach ($data['Peer'] ?? [] as $peer) {
        $label = strtolower(explode('.', $peer['DNSName'] ?? '')[0]);
        if (!$label) continue;
        $peers[$label] = [
            'online' => (bool)($peer['Online'] ?? false),
            'active' => (bool)($peer['Active'] ?? false),
            'ip'     => $peer['TailscaleIPs'][0] ?? null,
        ];
    }
    return $peers;
}

// Resolve one host's peer record out of vv_pt_ts_peers().
//
// The tailnet name and the OS hostname are not the same string, and nothing keeps them in step:
// this mesh has master.conf saying "unRAID-Jayred36" while the tailnet device is
// "unraid-jayred365". An exact-key lookup finds nothing, and every consumer that did one showed
// a live partner as dark — the Partnership cards until that was fixed inline, and the Fallback
// page for as long as it has existed, where the miss became state=UNREACHABLE.
//
// Exact key first, then a single unambiguous prefix match in either direction. One candidate or
// none — server1 must never resolve to server10 because it happens to share a prefix, and this
// is never similarity scoring. Same rule as vv_resolve_tailscale_ip(), which applies it to
// `tailscale status` text rather than to the parsed peer array.
// The matching rule itself, separated from the value it looks up, because the peer map is not
// always the same shape: include/monitor.php keys hostname => bool while everything else keys
// hostname => array. Four call sites had each written their own exact-key lookup and all four
// rendered this mesh's partner as dark. Returns the matching KEY, or null.
function vv_pt_peer_match(array $peerNames, string $hostname): ?string {
    $label = strtolower($hostname);
    if (in_array($label, $peerNames, true)) return $label;

    $cand = [];
    foreach ($peerNames as $peerName) {
        if (str_starts_with($peerName, $label) || str_starts_with($label, $peerName)) $cand[] = $peerName;
    }
    return count($cand) === 1 ? $cand[0] : null;
}

function vv_pt_peer_lookup(array $tsPeers, string $hostname): array {
    $key = vv_pt_peer_match(array_keys($tsPeers), $hostname);
    return $key !== null ? $tsPeers[$key] : ['online' => null, 'active' => false, 'ip' => null];
}

// ── SSH helper — run a single command on a remote host ────────────────────────

// Multiplexed, because this page makes several of these per render — a state read per partner,
// plus whatever a card asks for — and each one was paying a full handshake to a node across a real
// internet hop. Measured at about two seconds per call against 0.09s once a master is up.
//
// The socket lives in tmpfs so a reboot cannot inherit a stale one, and the path is kept short: a
// unix socket path is capped near 108 characters and ssh appends the user and host to this.
function vv_pt_ssh(string $ip, string $sshKey, string $cmd, int $timeout = 4): string {
    if (!$ip || !$sshKey || !file_exists($sshKey)) return '';

    $sock = rtrim(VV_CACHE_ROOT, '/') . '/ssh';
    if (!is_dir($sock)) @mkdir($sock, 0700, true);

    $full = sprintf(
        'ssh -i %s -o ConnectTimeout=%d -o StrictHostKeyChecking=no -o BatchMode=yes'
        . ' -o ControlMaster=auto -o ControlPersist=60s -o ControlPath=%s'
        . ' root@%s %s 2>/dev/null',
        escapeshellarg($sshKey), $timeout, escapeshellarg($sock . '/pt-%h'),
        escapeshellarg($ip), escapeshellarg($cmd)
    );
    return shell_exec($full) ?: '';
}

// ── System info ───────────────────────────────────────────────────────────────

// vv_system_info() (common.php) provides version, load_avg, array_state.
// /proc/uptime is the reliable uptime source (API uptime is an ISO date string, not seconds).
// vv_docker_containers() (common.php) provides the running container list.
function vv_pt_local_system(): array {
    $info      = vv_system_info();
    $uptimeSec = file_exists('/proc/uptime')
        ? (int)explode(' ', file_get_contents('/proc/uptime'))[0] : 0;
    return [
        'unraid_version' => $info['version'] ?? '',
        'uptime_sec'     => $uptimeSec,
        'load_avg'       => isset($info['load_avg']) ? $info['load_avg'][0] : null,
        'containers'     => count(vv_docker_containers()),
    ];
}

// SSH fallback for remote nodes — version/uptime/load/containers in one call.
// API stats from vv_remote_hosts_stats() take priority when available; SSH fills gaps.
function vv_pt_remote_system(string $ip, string $sshKey): array {
    $out = vv_pt_ssh($ip, $sshKey,
        'printf "%s\nUPTIME:%s\nLOAD:%s\nCONTAINERS:%s\n" ' .
        '"$(cat /etc/unraid-version 2>/dev/null)" ' .
        '"$(cat /proc/uptime 2>/dev/null)" ' .
        '"$(awk \'{print $1}\' /proc/loadavg 2>/dev/null)" ' .
        '"$(docker ps -q 2>/dev/null | wc -l)"');
    $ver = '';
    preg_match('/VERSION="([^"]+)"/', $out, $m); if ($m) $ver = $m[1];
    $uptime = 0;
    if (preg_match('/UPTIME:([\d.]+)/', $out, $m)) $uptime = (int)$m[1];
    $load = null;
    if (preg_match('/LOAD:([\d.]+)/', $out, $m)) $load = round((float)$m[1], 2);
    $containers = null;
    if (preg_match('/CONTAINERS:(\d+)/', $out, $m)) $containers = (int)$m[1];
    return [
        'unraid_version' => $ver,
        'uptime_sec'     => $uptime,
        'load_avg'       => $load,
        'containers'     => $containers,
    ];
}


// ── Mesh traffic ──────────────────────────────────────────────────────────────
//
// What has actually crossed the link to each partner, from Tailscale's own per-peer byte
// counters — not from rsync's logs.
//
// The previous version totalled data/db/bandwidth_history.db, so it could only ever describe
// rsync. Everything else using the same link — SSH, the arr APIs, conf pushes, the Unraid API,
// the webhook — was invisible to it, and it reported "no data moved" across a link carrying
// hundreds of gigabytes. Counting bytes on the wire measures the partnership rather than one
// tool's opinion of itself.
//
// Windows come from Tools/mesh_traffic_sample.php, which records the raw counters once a
// minute. A window total is the newest sample minus the oldest one still inside it.
function vv_pt_mesh_traffic(): array {
    $file = DATA_DIR . '/db/mesh_traffic.db';
    $out  = ['peers' => [], 'samples' => 0];
    if (!is_file($file)) return $out;

    $byPeer = [];
    $fh = fopen($file, 'r');
    if (!$fh) return $out;
    while (($line = fgets($fh)) !== false) {
        $p = explode('|', trim($line));
        if (count($p) < 4) continue;
        $byPeer[$p[1]][] = [(int)$p[0], (int)$p[2], (int)$p[3]];
        $out['samples']++;
    }
    fclose($fh);

    $now     = time();
    $windows = ['24h' => 86400, '7d' => 604800, '30d' => 2592000];

    foreach ($byPeer as $peer => $rows) {
        usort($rows, fn($a, $b) => $a[0] <=> $b[0]);
        $last = end($rows);

        // Live rate from the two most recent samples. Meaningless if they are far apart — a gap
        // means the sampler missed runs, and dividing by that gap reports an average over a
        // period nobody watched as though it were current.
        $live = null;
        $n = count($rows);
        if ($n >= 2) {
            $prev = $rows[$n - 2];
            $dt   = $last[0] - $prev[0];
            if ($dt > 0 && $dt <= 300 && $last[1] >= $prev[1] && $last[2] >= $prev[2]) {
                $live = ['tx_bps' => (int)(($last[1] - $prev[1]) / $dt),
                         'rx_bps' => (int)(($last[2] - $prev[2]) / $dt),
                         'age'    => $now - $last[0]];
            }
        }

        $win = [];
        foreach ($windows as $k => $span) {
            $from = $now - $span;
            // Sum forward through the window rather than subtracting endpoints, so a tailscaled
            // restart — which returns the counters to zero — costs one interval instead of
            // producing a negative total or a spike the size of the whole previous session.
            $tx = 0; $rx = 0; $seen = 0; $oldest = null;
            $prev = null;
            foreach ($rows as $r) {
                if ($r[0] < $from) { $prev = $r; continue; }
                if ($oldest === null) $oldest = $r[0];
                if ($prev !== null) {
                    $tx += ($r[1] >= $prev[1]) ? $r[1] - $prev[1] : $r[1];
                    $rx += ($r[2] >= $prev[2]) ? $r[2] - $prev[2] : $r[2];
                    $seen++;
                }
                $prev = $r;
            }
            $win[$k] = [
                'tx' => $tx, 'rx' => $rx, 'intervals' => $seen,
                // How far back the data actually reaches. A 30-day figure built from six hours of
                // samples is not a 30-day figure, and the card has to be able to say so.
                'covers' => $oldest ? $now - $oldest : 0,
            ];
        }

        $out['peers'][$peer] = [
            'live'    => $live,
            'windows' => $win,
            // Counter totals since tailscaled last started — the longest view available without
            // any history at all, and the one that is right on a fresh install.
            'session' => ['tx' => $last[1], 'rx' => $last[2]],
        ];
    }
    return $out;
}

// ── Media seed progress ───────────────────────────────────────────────────────
//
// Phase 3 dispatches Rsync/media_seed.sh detached, because a first seed of a full
// media library is a multi-week transfer and used to hold the onboard — and therefore the
// phase-2 flag, and therefore this whole page — open for the duration.
//
// Detaching it means nothing on screen would mention it at all unless something reads its
// job record, which is what this does. The seed is a push from this host to the partner, so
// it is rendered on the partner's card even though every byte of evidence for it is local.
//
// Returns [] when there has never been a seed. A stale "running" record whose pid is gone is
// reported as stopped rather than running: a record is not a process.
function vv_pt_media_seed(array $setupDb = [], string $nodeIdUpper = ''): array {
    $stat = '/var/log/varaverk/Rsync/media_seed.json';
    $log  = '/var/log/varaverk/Rsync/media_seed.log';

    // MEDIA_SEED_ENABLED is reported alongside the record rather than instead of it, because
    // the two answer different questions: the toggle says whether a seed may start, the record
    // says what the last one did. media_seed.sh exits 0 when the toggle is off — correct, it is
    // a decision and not a failure — so a record reading "ok" beside a disabled toggle would
    // otherwise render as "Seed complete" over a partner that was never seeded.
    //
    // Unset reads as enabled, matching media_seed.sh: the toggle postdates the script.
    $raw     = vv_read_conf_raw('master.conf');
    $toggle  = vv_arr_scalar($raw, 'MEDIA_SEED_ENABLED');
    $enabled = ($toggle === '' || strtolower($toggle) === 'true');

    // Phase 3 is complete only when every share landed, which media_seed.sh decides and records
    // at the end of a run that may have taken weeks. The job record cannot answer this: it goes
    // "ok" the moment the process exits, including the exits that seeded nothing.
    $phase3Done = $nodeIdUpper !== ''
        && ($setupDb[$nodeIdUpper . '_PHASE3_DONE'] ?? '') === 'true';

    if (!is_file($stat)) return ['enabled' => $enabled, 'phase3_done' => $phase3Done];

    $j = json_decode((string)file_get_contents($stat), true);
    if (!is_array($j)) return ['enabled' => $enabled, 'phase3_done' => $phase3Done];

    $status = (string)($j['status'] ?? '');
    $pid    = (int)($j['pid'] ?? 0);
    if ($status === 'running' && (!$pid || !is_dir("/proc/$pid"))) {
        $status = 'stopped';
    }

    $share = '';
    $line  = '';
    if (is_file($log)) {
        // rsync --info=progress2 rewrites one line with \r, so the tail is read as bytes and
        // split on both terminators — splitting on \n alone yields a single enormous "line"
        // holding every progress repaint of the current file.
        $fh = fopen($log, 'r');
        if ($fh) {
            fseek($fh, 0, SEEK_END);
            $size = ftell($fh);
            fseek($fh, max(0, $size - 65536));
            $tail = (string)fread($fh, 65536);
            fclose($fh);
            $parts = preg_split('/[\r\n]+/', $tail) ?: [];
            for ($i = count($parts) - 1; $i >= 0; $i--) {
                $p = trim($parts[$i]);
                if ($p === '') continue;
                if ($line === '') $line = $p;
                if (preg_match('/Seeding:\s*(\S+)/', $p, $m)) { $share = $m[1]; break; }
            }
        }
    }

    return array_filter([
        'enabled'     => $enabled,
        'phase3_done' => $phase3Done,
        'status'  => $status,
        'start'   => isset($j['start']) ? (int)$j['start'] : null,
        'end'     => isset($j['end'])   ? (int)$j['end']   : null,
        'share'   => $share,
        'line'    => mb_substr($line, 0, 160),
    ], fn($v) => $v !== null && $v !== '');
}

// ── Per-node data ─────────────────────────────────────────────────────────────

function vv_pt_nodes(): array {
    $currentHost = vv_detect_host();
    $hosts       = vv_arr_known_hosts();        // ['host1' => 'hostname', ...]
    $vars        = vv_conf_vars();
    $tsPeers     = vv_pt_ts_peers();
    // Read once: the partnership state this host records for itself, used by every remote's
    // mesh check below.
    $localPtState = vv_pt_read_db(STATE_DIR . '/partnership_' . vv_get_hostname() . '.db')['state'] ?? '';
    $ownerSlot   = strtolower($vars['PARTNERSHIP_OWNER_HOST'] ?? '');
    $setupDb     = vv_setup_state_read();

    // Remote host stats (API + 30s /tmp cache) — includes version, uptime, cpu/ram/array/temp/vms
    $remoteStats = vv_remote_hosts_stats();

    // SSH key for this host
    $myId     = strtoupper($currentHost);
    $myRaw    = vv_read_conf_raw($currentHost . '.conf');
    $mySshKey = vv_arr_scalar($myRaw, $myId . '_SSH_KEY');

    $nodes = [];
    foreach ($hosts as $slot => $hostname) {
        $isMe    = ($slot === $currentHost || $currentHost === 'unknown');
        $isOwner = (strtolower($ownerSlot) === $slot);

        // Tailscale
        //
        // Exact key first, then a single unambiguous prefix match in either direction. The
        // tailnet name and the OS hostname are not the same string and nothing keeps them in
        // step: this mesh has master.conf saying "unRAID-Jayred36" while the tailnet device is
        // "unraid-jayred365". An exact-key lookup found nothing, so every remote card rendered
        // with no IP, no online state and no container count — a partner that was answering SSH
        // the whole time displayed as though it were not there.
        //
        // Same rule as vv_resolve_tailscale_ip(), and the same refusal: one candidate or none.
        // server1 must never resolve to server10 because it happens to share a prefix.
        $ts = vv_pt_peer_lookup($tsPeers, $hostname);

        // Fallback state
        $fbState = 'UNKNOWN';
        if ($isMe) {
            $fb = vv_pt_read_db(STATE_DIR . '/fallback_state.db');
            $fbState = $fb['state'] ?? 'UNKNOWN';
        } elseif ($ts['online'] && $ts['ip'] && $mySshKey) {
            $out = vv_pt_ssh($ts['ip'], $mySshKey, vv_remote_state_cmd('fallback_state.db'));
            if ($out) {
                $fb = [];
                foreach (explode("\n", $out) as $line) {
                    [$k, $v] = array_pad(explode('=', trim($line), 2), 2, '');
                    if ($k) $fb[trim($k)] = trim($v, '"\'');
                }
                $fbState = $fb['state'] ?? 'UNKNOWN';
            }
        }

        // Partnership DB — local only (each server writes its own)
        $dbPath = STATE_DIR . "/partnership_{$hostname}.db";
        $ptDb   = vv_pt_read_db($dbPath);

        // System info
        $system = $isMe
            ? vv_pt_local_system()
            : ($ts['online'] && $ts['ip'] && $mySshKey ? vv_pt_remote_system($ts['ip'], $mySshKey) : []);

        // Onboard phase from setup.db — 0=not started, 1=SSH+conf done, 2=fully onboarded.
        //
        // Computed for self as well as for partners. The flags are keyed to the host they
        // describe, so on the mirror HOST2_PHASE*_DONE is the mirror's own progress — and the
        // mirror's action panel needs it to know whether its SSH key step is still outstanding.
        // This was forced to null for self, which left a mirror with no way to see its own
        // phase and no way to render anything but a bare button.
        $nodeIdUpper  = strtoupper($slot);
        $onboardPhase = ($setupDb[$nodeIdUpper . '_PHASE2_DONE'] ?? '') === 'true' ? 2
                      : (($setupDb[$nodeIdUpper . '_PHASE1_DONE'] ?? '') === 'true' ? 1 : 0);

        // An ACTIVE partnership means phase 2, whatever the flags say. HOST<n>_PHASE*_DONE always
        // describe the MIRROR — they are the owner's record of how far it got provisioning the
        // other side — so the owner's own slot has no such flag and never will. On the mirror
        // that left the owner's card computing phase 0 and rendering "Not provisioned · SSH keys
        // not installed", complete with an ▶ Onboard button, on the very host that had just
        // finished onboarding it. The state file is the authority on whether a partnership
        // exists; the flags only say who did what to whom.
        if ($onboardPhase < 2 && ($ptDb['state'] ?? '') === 'ACTIVE') {
            $onboardPhase = 2;
        }
        // key_ready: local key generated but not yet installed on HOST2 (SSH pending manual step)
        $keyReady  = !$isMe && ($setupDb[$nodeIdUpper . '_KEY_READY'] ?? '') === 'true';
        // For self: local setup complete flag (set by partnership_manager --onboard --local-only)
        $localDone = $isMe && ($setupDb[$nodeIdUpper . '_LOCAL_DONE'] ?? '') === 'true';

        // Unraid API key status — checks Unraid's key store directly so deletions are reflected.
        $apiKeySet     = false;
        $apiKeyPreview = '';
        if ($isMe) {
            // Cached for a minute, because `unraid-api apikey` is not a file read — it starts
            // Unraid's Node CLI, measured at 1.98s, and it ran on every render of this page. That
            // was the single largest cost in assembling it, and this page is the one that opens
            // the mesh chat, so it was two seconds in front of a conversation every time.
            //
            // Still reads the key store rather than the conf, so a key deleted in Unraid's own UI
            // is still reflected — a minute later rather than instantly, which is the trade. The
            // preview is eight characters of a key that changes when someone deliberately rotates
            // it; nobody is watching it to the second.
            $apiData = vv_cache_read('pt_apikey', 60);
            if ($apiData === null) {
                $hn      = trim((string)shell_exec("hostname -s 2>/dev/null | sed 's/^[Uu][Nn][Rr][Aa][Ii][Dd]-//'")) ?: 'Varaverk';
                $keyName = 'Varaverk ' . $hn;
                $apiOut  = shell_exec('/usr/local/sbin/unraid-api apikey --name ' . escapeshellarg($keyName) . ' --json </dev/null 2>/dev/null');
                $decoded = json_decode(trim($apiOut ?? ''), true);
                // Cached either way. A host with no key would otherwise pay the two seconds on
                // every render forever, which is the case that needs the cache most.
                $apiData = is_array($decoded) ? $decoded : [];
                vv_cache_write('pt_apikey', $apiData);
            }
            if (!empty($apiData['key'])) {
                $apiKeySet     = true;
                $apiKeyPreview = substr($apiData['key'], 0, 8) . '...' . substr($apiData['key'], -4);
            }
        } else {
            // The partner's key is in the partner's own conf, which this host holds in the RAM
            // cache that conf_sync fills — never on disk here, because sparse checkout means a
            // host only ever checks out its own host*.conf.
            //
            // Two separate facts, deliberately not merged: a key being CONFIGURED is read from
            // that conf, and the key WORKING is proven by the partner's API having answered this
            // page's own stats call. A configured key that no longer authenticates would
            // otherwise render exactly like a healthy one.
            $pConf = VV_CONF_RAM_CACHE_DIR . '/' . strtolower($nodeIdUpper) . '.conf';
            if (is_readable($pConf)) {
                $pKey = vv_arr_scalar((string)@file_get_contents($pConf), $nodeIdUpper . '_UNRAID_API_KEY');
                if ($pKey !== '') {
                    $apiKeySet     = true;
                    $apiKeyPreview = substr($pKey, 0, 8) . '...' . substr($pKey, -4);
                }
            }
        }
        // Did the partner's API actually answer? Empty for self, where the local API has its own
        // banner on the Monitor tab.
        $apiLive = !$isMe && !empty(($remoteStats[$nodeIdUpper] ?? [])['available']);

        // Live metrics: local uses vv_api_data() (cached); remote uses vv_remote_hosts_stats() (30s cache)
        if ($isMe) {
            $metrics = array_merge(
                vv_api_node_metrics(vv_api_data()),
                array_filter([
                    'load_avg'   => $system['load_avg']   ?? null,
                    'containers' => $system['containers'] ?? null,
                ], fn($v) => $v !== null)
            );
        } else {
            $rStat = $remoteStats[$nodeIdUpper] ?? [];
            // Merge API metrics from remote stats with SSH extras (load, containers)
            $metrics = array_filter([
                'cpu_pct'        => $rStat['cpu_pct']        ?? null,
                'ram_used_gb'    => $rStat['ram_used_gb']    ?? null,
                'ram_total_gb'   => $rStat['ram_total_gb']   ?? null,
                'array_used_tb'  => $rStat['array_used_tb']  ?? null,
                'array_total_tb' => $rStat['array_total_tb'] ?? null,
                'max_disk_temp'  => $rStat['max_disk_temp']  ?? null,
                'vm_count'       => $rStat['vm_count']       ?? null,
                'load_avg'       => $system['load_avg']      ?? null,
                'containers'     => $system['containers']    ?? null,
            ], fn($v) => $v !== null);
            // Fill version/uptime from API stats if SSH didn't provide them
            if (empty($system['unraid_version']) && !empty($rStat['version'])) {
                $system['unraid_version'] = $rStat['version'];
            }
            if (empty($system['uptime_sec']) && !empty($rStat['uptime_sec'])) {
                $system['uptime_sec'] = $rStat['uptime_sec'];
            }
        }

        $nodes[] = [
            'slot'            => $slot,
            'id'              => $nodeIdUpper,
            'hostname'        => $hostname,
            'is_me'           => $isMe,
            'is_owner'        => $isOwner,
            'ts_online'       => $ts['online'],
            'ts_active'       => $ts['active'],
            'ts_ip'           => $ts['ip'],
            'fallback'        => $fbState,
            'partnership'     => $ptDb,
            'system'          => $system,
            'onboard_phase'   => $onboardPhase,
            'key_ready'       => $keyReady,
            'local_done'      => $localDone,
            'api_key_set'     => $apiKeySet,
            'api_key_preview' => $apiKeyPreview,
            'api_live'        => $apiLive,
            'metrics'         => $metrics,
            'media_seed'      => $isMe ? [] : vv_pt_media_seed($setupDb, $nodeIdUpper),
            // Every condition that has to hold for this partner to actually be usable, each
            // reported separately. One rolled-up boolean would have said "not ok" for the last
            // hour without saying that the only failing part was a name lookup.
            'mesh'            => $isMe ? null : [
                'tailscale'   => $ts['online'] === true,
                'ssh'         => !empty($system['unraid_version']) || !empty($system['uptime_sec']),
                // This host's own state file, not one named after the remote. A partnership is a
                // single mutual fact, and each host writes it under its OWN hostname — so
                // partnership_<remote>.db does not exist here, and reading it made a healthy
                // partnership report as a failing leg on the owner's own page.
                'partnership' => $localPtState === 'ACTIVE',
                'onboarded'   => $onboardPhase >= 2,
            ],
        ];
    }
    return $nodes;
}

// ── Connectivity test — SSH echo with round-trip timing ─────────────────────────
function vv_pt_ping(string $slot): array {
    $slot = strtolower($slot);
    $vars = vv_conf_vars();
    $hostname = $vars[strtoupper($slot)] ?? '';
    if (!$hostname) return ['ok' => false, 'error' => 'Unknown host slot'];

    $currentHost = vv_detect_host();
    $myRaw       = vv_read_conf_raw($currentHost . '.conf');
    $sshKey      = vv_arr_scalar($myRaw, strtoupper($currentHost) . '_SSH_KEY');
    if (!$sshKey || !file_exists($sshKey)) {
        return ['ok' => false, 'error' => 'No SSH key configured on this host'];
    }

    $ip = vv_resolve_tailscale_ip($hostname);
    if (!$ip) return ['ok' => false, 'error' => "Cannot resolve Tailscale IP for $hostname"];

    $t0  = microtime(true);
    $out = vv_pt_ssh($ip, $sshKey, 'echo ok', 8);
    $ms  = (int)round((microtime(true) - $t0) * 1000);

    if (trim($out) === 'ok') {
        return ['ok' => true, 'latency_ms' => $ms, 'host' => $hostname, 'ip' => $ip];
    }
    return ['ok' => false, 'error' => "SSH to $hostname ($ip) failed or timed out", 'host' => $hostname];
}

// ── Entry point ───────────────────────────────────────────────────────────────

function vv_partnership_all(): array {
    return [
        'config' => vv_pt_config(),
        'nodes'  => vv_pt_nodes(),
        'sync'   => vv_pt_sync(),
        'xfer'   => vv_pt_mesh_traffic(),
        'ts'     => time(),
    ];
}