Varaverk/Plugin/unraid/include/node_chat.php

<?php
// ═══════════════════════════════════════════════════════════════════════════════════════════════
// PURPOSE
//   The mesh's message store. A place for the people who run these servers to talk to each
//   other — "mine is down for a week", "who has seen this error", "the key you wanted is X" —
//   in the one window that already knows which machines are in the partnership.
//
// OPERATIONAL MODEL
//   Every node keeps its own copy. Sending is: append here, then push a copy to each recipient
//   over the SSH trust the onboard already established. There is no server and no election —
//   the store is a log, and a log replicated by append is the shape that survives a partner
//   being switched off.
//
//     data/db/node_chat/global.jsonl          every HOST* in master.conf
//     data/db/node_chat/dm-host1-host3.jsonl  one pair, names sorted so both sides agree
//     data/db/node_chat/.spool/<host>/*.json  undelivered, retried by the minute job
//
// DESIGN PRINCIPLES
//   Never two hosts.
//       Channels are derived from vv_known_hosts(), so a third machine gets a global channel
//       that includes it and a DM channel with each existing node, with no code change. The
//       pair name is sorted rather than "mine-theirs" precisely so both ends compute the same
//       filename without having to agree on who spoke first.
//
//   Delete is local, and that is the feature.
//       Removing a message everywhere needs tombstones, ordering and a story for a node that
//       was offline when the delete happened. Removing it from this machine is one file write.
//       The UI says "delete here" so nobody expects otherwise.
//
//   Sender identity comes from the transport, not the payload's word for it.
//       The record carries `from`, but anything writing to this store already had root over
//       SSH — the trust boundary is the key, and a machine that holds it can say anything
//       regardless of what this file validates.
//
// OPERATIONAL SAFEGUARDS
//   Messages are stored PLAINTEXT on every recipient. Tailscale encrypts the wire; the file on
//   disk is readable by anyone with the box. The card says so, because the first thing anyone
//   wants to send a partner is a credential.
//
//   Channel names are derived, never taken from the caller — a name is rebuilt from known host
//   ids, so no request can address a path outside the store.
//
//   Retention is per channel and enforced on write, so a busy channel cannot fill the flash.
//
//   Delivery failures spool rather than throw. A partner that is asleep is the normal case for
//   the message "my server is going down".
//
// EXPORTS
//   vv_nc_channels()        the channels this host participates in
//   vv_nc_read()            messages for one channel, newest last
//   vv_nc_append()          store one record locally
//   vv_nc_send()            store + deliver to the channel's other members
//   vv_nc_delete_local()    forget one message on this machine
//   vv_nc_flush_spool()     retry undelivered
// ═══════════════════════════════════════════════════════════════════════════════════════════════
require_once __DIR__ . '/config.php';

define('VV_NC_DIR',   DATA_DIR . '/db/node_chat');
define('VV_NC_SPOOL', VV_NC_DIR . '/.spool');
define('VV_NC_KEEP',  300);          // messages retained per channel, per node

function vv_nc_me(): string { return strtolower(vv_detect_host()); }

// One validator, used by both the sender and the receiver.
//
// Both ends must agree on what a style is, and they had two copies of the rules — the sender's
// and the receiver's. Two copies of a whitelist is one that drifts, and the half that drifts is
// the half a partner running newer code writes into. Every value here ends up interpolated into
// a style attribute, so the allowed set is closed rather than sanitised.
function vv_nc_clean_style(array $s): array {
    return [
        'color'     => preg_match('/^#[0-9a-f]{6}$/i', $s['color'] ?? '') ? strtolower($s['color']) : '',
        'font'      => in_array($s['font'] ?? '', ['mono', 'sans', 'serif'], true) ? $s['font'] : '',
        'size'      => in_array($s['size'] ?? '', ['sm', 'lg'], true) ? $s['size'] : '',
        'bold'      => !empty($s['bold']),
        'italic'    => !empty($s['italic']),
        'underline' => !empty($s['underline']),
    ];
}

// host1/host2/... → the channels this host can see. Global plus one DM per other node.
function vv_nc_channels(): array {
    $me    = vv_nc_me();
    $hosts = array_keys(vv_known_hosts());
    $out   = [['id' => 'global', 'label' => 'Global', 'members' => $hosts]];
    foreach ($hosts as $h) {
        if ($h === $me) continue;
        $pair = [$me, $h];
        sort($pair);
        $out[] = ['id' => 'dm-' . implode('-', $pair), 'label' => strtoupper($h), 'members' => $pair];
    }
    return $out;
}

// Rebuilt from the known channel list rather than trusted — the caller names a channel, and the
// only names that resolve are ones this host is actually a member of.
function vv_nc_channel(string $id): ?array {
    foreach (vv_nc_channels() as $c) if ($c['id'] === $id) return $c;
    return null;
}

function vv_nc_path(string $channelId): string {
    return VV_NC_DIR . '/' . $channelId . '.jsonl';
}

function vv_nc_read(string $channelId, int $limit = 200): array {
    $p = vv_nc_path($channelId);
    if (!is_file($p)) return [];
    $lines = @file($p, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) ?: [];
    if (count($lines) > $limit) $lines = array_slice($lines, -$limit);
    $out = [];
    foreach ($lines as $l) {
        $j = json_decode($l, true);
        if (is_array($j) && !empty($j['id'])) $out[] = $j;
    }
    return $out;
}

function vv_nc_append(string $channelId, array $msg): bool {
    if (!vv_nc_channel($channelId)) return false;
    @mkdir(VV_NC_DIR, 0755, true);
    $p = vv_nc_path($channelId);

    // Idempotent on id: a spool retry that actually landed the first time must not double-post.
    foreach (vv_nc_read($channelId, VV_NC_KEEP) as $m) {
        if (($m['id'] ?? '') === ($msg['id'] ?? '')) return true;
    }
    if (@file_put_contents($p, json_encode($msg, JSON_UNESCAPED_SLASHES) . "\n",
                           FILE_APPEND | LOCK_EX) === false) return false;

    $lines = @file($p, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) ?: [];
    if (count($lines) > VV_NC_KEEP) {
        $tmp = $p . '.tmp';
        if (@file_put_contents($tmp, implode("\n", array_slice($lines, -VV_NC_KEEP)) . "\n") !== false)
            @rename($tmp, $p);
    }
    return true;
}

function vv_nc_delete_local(string $channelId, string $msgId): bool {
    $p = vv_nc_path($channelId);
    if (!vv_nc_channel($channelId) || !is_file($p)) return false;
    $keep = [];
    foreach (@file($p, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) ?: [] as $l) {
        $j = json_decode($l, true);
        if (is_array($j) && ($j['id'] ?? '') === $msgId) continue;
        $keep[] = $l;
    }
    $tmp = $p . '.tmp';
    if (@file_put_contents($tmp, $keep ? implode("\n", $keep) . "\n" : '') === false) return false;
    return @rename($tmp, $p);
}

// Deliver one record to one host. Returns false on any failure so the caller can spool.
function vv_nc_deliver(string $host, string $channelId, array $msg): bool {
    $vars   = vv_conf_vars();
    $me     = strtoupper(vv_nc_me());
    $sshKey = $vars[$me . '_SSH_KEY'] ?? '';
    if (!$sshKey || !is_file($sshKey)) return false;

    $hostname = $vars[strtoupper($host)] ?? '';
    if ($hostname === '') return false;
    $ip = vv_resolve_tailscale_ip($hostname);
    if (!$ip) return false;

    // The receiving script lives wherever that host installed Varaverk, which is not necessarily
    // where this one did — read from the partner's varaverk.cfg, which stays on flash whatever
    // layout it uses. Assuming our own path is the bug that broke Step 1b on an appdata mirror.
    $sshBase = 'ssh -i ' . escapeshellarg($sshKey)
             . ' -o ConnectTimeout=8 -o BatchMode=yes -o StrictHostKeyChecking=no root@' . escapeshellarg($ip);
    $cfg = trim((string)shell_exec($sshBase
        . ' ' . escapeshellarg('grep -oP \'(?<=SCRIPTS_DIR=")[^"]+\' /boot/config/plugins/varaverk/varaverk.cfg 2>/dev/null')));
    $sd = $cfg !== '' ? $cfg : '/boot/config/plugins/varaverk';
    $remote = $sd . '/Plugin/unraid/Tools/node_chat_receive.php';

    $cmd = $sshBase . ' ' . escapeshellarg('[ -f ' . $remote . ' ] || exit 127; php ' . $remote
                                . ' --channel=' . escapeshellarg($channelId));
    $desc = [0 => ['pipe','r'], 1 => ['pipe','w'], 2 => ['pipe','w']];
    $pr = @proc_open($cmd, $desc, $pipes);
    if (!is_resource($pr)) return false;
    fwrite($pipes[0], json_encode($msg, JSON_UNESCAPED_SLASHES));
    fclose($pipes[0]);
    stream_get_contents($pipes[1]); fclose($pipes[1]);
    stream_get_contents($pipes[2]); fclose($pipes[2]);
    return proc_close($pr) === 0;
}

function vv_nc_spool(string $host, string $channelId, array $msg): void {
    $dir = VV_NC_SPOOL . '/' . $host;
    @mkdir($dir, 0755, true);
    @file_put_contents($dir . '/' . $msg['id'] . '.json',
        json_encode(['channel' => $channelId, 'msg' => $msg], JSON_UNESCAPED_SLASHES));
}

// Store locally, then deliver to everyone else in the channel. Always returns the record: a
// message that reached no one is still a message this operator wrote and must be able to see.
function vv_nc_send(string $channelId, string $text, array $style = []): ?array {
    $ch = vv_nc_channel($channelId);
    if (!$ch || trim($text) === '') return null;

    $msg = [
        'id'    => bin2hex(random_bytes(8)),
        'ts'    => time(),
        'from'  => vv_nc_me(),
        // Reserved from day one so a board or an announcement is a filter later rather than a
        // migration. Everything written today is a message.
        'kind'  => 'msg',
        'text'  => mb_substr(trim($text), 0, 4000),
        'style' => vv_nc_clean_style($style),
    ];
    vv_nc_append($channelId, $msg);

    foreach ($ch['members'] as $h) {
        if ($h === vv_nc_me()) continue;
        if (!vv_nc_deliver($h, $channelId, $msg)) vv_nc_spool($h, $channelId, $msg);
    }
    return $msg;
}

function vv_nc_flush_spool(): int {
    $sent = 0;
    foreach (glob(VV_NC_SPOOL . '/*', GLOB_ONLYDIR) ?: [] as $dir) {
        $host = basename($dir);
        foreach (glob($dir . '/*.json') ?: [] as $f) {
            $j = json_decode((string)@file_get_contents($f), true);
            if (!is_array($j) || empty($j['msg'])) { @unlink($f); continue; }
            if (vv_nc_deliver($host, (string)$j['channel'], $j['msg'])) { @unlink($f); $sent++; }
        }
    }
    return $sent;
}