From b153051e58956683220fbc80fb0e6b818c27025e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E6=80=A7=E9=A9=8A?= Date: Wed, 4 Nov 2026 19:30:32 +0800 Subject: [PATCH] Lasting connection to the account server, per-item sync, files and models via the brain - Each AnyBot keeps one WebSocket to the account server (pinged, reopened within a second when cut). Requests come down it at once; calls to other AnyBots of the account (brain sync, hands) go up it. Frames carry raw bytes, compressed when that saves space. HTTP long-poll stays as fallback. - The account server knows at once when an AnyBot goes away. - Edits to lists and maps go item by item, so two computers editing one list keep both changes. - Screens fetch chat files from the brain, get model lists through the brain's keys, and show the brain's computer as the brain's. - Moving this computer's bots skips starter bots the brain has and bots already moved. - The phone goes to the brain instead of a computer that shows it. - deploy/linux/hands.sh: a Linux computer (with or without a screen) as hands, kept running by systemd. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Uirha3oEtWeHh58zA7ErHa --- Cargo.lock | 3 + Cargo.toml | 4 + deploy/linux/hands.sh | 103 +++++++++ deploy/server/docker-compose.yml | 6 + mobile/index.html | 15 +- relay/Caddyfile | 5 +- relay/Cargo.lock | 161 +++++++++++++- relay/Cargo.toml | 1 + relay/src/main.rs | 252 ++++++++++++++++++--- src/app.rs | 22 +- src/app/attach_ui.rs | 2 +- src/app/computers_ui.rs | 11 +- src/app/mirror_ui.rs | 233 +++++++++++++++++++- src/httpd.rs | 13 ++ src/main.rs | 2 + src/remote.rs | 117 ++++++++-- src/tunnel.rs | 365 +++++++++++++++++++++++++++++++ src/wire.rs | 42 ++++ 18 files changed, 1291 insertions(+), 66 deletions(-) create mode 100755 deploy/linux/hands.sh create mode 100644 src/tunnel.rs create mode 100644 src/wire.rs diff --git a/Cargo.lock b/Cargo.lock index 3b8845a..aa4bb9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -345,6 +345,7 @@ dependencies = [ "cpal", "dirs", "eframe", + "flate2", "futures-lite", "hayro", "iana-time-zone", @@ -356,6 +357,7 @@ dependencies = [ "resvg", "ring", "rodio", + "rustls", "serde", "serde_json", "tungstenite", @@ -365,6 +367,7 @@ dependencies = [ "wayland-protocols", "wayland-protocols-misc", "wayland-protocols-wlr", + "webpki-roots", "x11rb 0.13.2", "zbus", "zhconv", diff --git a/Cargo.toml b/Cargo.toml index 4f94769..68baaf6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,10 @@ path = "src/main.rs" [dependencies] eframe = "0.37" ureq = { version = "3", features = ["json"] } +# 長連線 to the account server (WebSocket over the same TLS ureq uses), and compressing what goes over it. +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } +webpki-roots = "1" +flate2 = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" dirs = "7" diff --git a/deploy/linux/hands.sh b/deploy/linux/hands.sh new file mode 100755 index 0000000..bb77f70 --- /dev/null +++ b/deploy/linux/hands.sh @@ -0,0 +1,103 @@ +#!/bin/sh +# Make this Linux computer a pair of hands for your AnyBot brain: the brain's bots can see and use +# its screen, files and browser (asking you first, as usual). Works on a desktop and on a computer +# without a screen (a virtual screen is made for it). Run from the app/ folder: +# +# ./deploy/linux/hands.sh --deps also install what a computer without a screen needs (sudo) +# ./deploy/linux/hands.sh set up and start it (asks for your account the first time) +# ./deploy/linux/hands.sh --stop stop it and don't start it with the computer any more +set -e +cd "$(dirname "$0")/../.." +UNIT="$HOME/.config/systemd/user/anybot-hands.service" +ENVF="$HOME/.config/anybot/hands.env" +RUN="$HOME/.local/bin/anybot-hands" + +if [ "$1" = "--stop" ]; then + systemctl --user disable --now anybot-hands.service 2>/dev/null || true + rm -f "$UNIT" + systemctl --user daemon-reload + echo "Stopped. (Your account details stay in $ENVF; delete it if you like.)" + exit 0 +fi + +if [ "$1" = "--deps" ]; then + # A virtual screen, a window manager, the accessibility bus (for reading apps' buttons), a session bus. + if command -v apt-get >/dev/null; then + sudo apt-get update && sudo apt-get install -y xvfb openbox at-spi2-core libatk-adaptor dbus-x11 xauth fonts-noto-cjk + elif command -v dnf >/dev/null; then + sudo dnf install -y xorg-x11-server-Xvfb openbox at-spi2-core at-spi2-atk dbus-x11 xorg-x11-xauth google-noto-sans-cjk-fonts + elif command -v pacman >/dev/null; then + sudo pacman -S --needed xorg-server-xvfb openbox at-spi2-core dbus xorg-xauth noto-fonts-cjk + else + echo "Install with your package manager: Xvfb, openbox, at-spi2-core, dbus, xauth" + fi +fi + +command -v systemctl >/dev/null || { echo "This needs systemd (to keep AnyBot running)."; exit 1; } +[ -x "$HOME/.local/bin/anybot" ] || sh deploy/linux/install.sh + +if [ ! -f "$ENVF" ]; then + echo "Your AnyBot account (the same one your brain uses):" + printf " Account server address (e.g. https://anybot.example.com): "; read -r URL + printf " Account name: "; read -r USERNAME + printf " Password: "; stty -echo 2>/dev/null; read -r PASSWORD; stty echo 2>/dev/null; echo + printf " Name for this computer [%s]: " "$(hostname)"; read -r NAME + printf " Brain's name (empty = only lend hands, show no bots here): "; read -r BRAIN + mkdir -p "$(dirname "$ENVF")" + umask 077 + # Each value in single quotes, so a password with $ or spaces stays as typed. + q() { printf "%s='%s'\n" "$1" "$(printf %s "$2" | sed "s/'/'\\\\''/g")"; } + { + q ANYBOT_ACCOUNT "${URL%/}|$USERNAME" + q ANYBOT_ACCOUNT_PASSWORD "$PASSWORD" + q ANYBOT_HOST_NAME "${NAME:-$(hostname)}" + if [ -n "$BRAIN" ]; then q ANYBOT_BRAIN "$BRAIN"; fi + } > "$ENVF" + chmod 600 "$ENVF" +fi + +mkdir -p "$(dirname "$RUN")" "$(dirname "$UNIT")" +cat > "$RUN" <<'SH' +#!/bin/sh +# Started by systemd (anybot-hands.service). +set -a; . "$HOME/.config/anybot/hands.env"; set +a +export ANYBOT_DEVICE_LINK=1 +export LANG="${LANG:-C.UTF-8}" +# The accessibility bus, so bots can read apps' buttons instead of only looking at pixels. +export NO_AT_BRIDGE=0 GTK_A11Y=atspi QT_LINUX_ACCESSIBILITY_ALWAYS_ON=1 +if [ -z "$DISPLAY" ] && [ -z "$WAYLAND_DISPLAY" ]; then + # No screen: a virtual one with a window manager, and a session bus for the accessibility bus. + exec xvfb-run -a -s "-screen 0 1440x900x24" dbus-run-session -- sh -c 'openbox >/dev/null 2>&1 & exec "$HOME/.local/bin/anybot" --serve' +fi +exec "$HOME/.local/bin/anybot" --serve +SH +chmod 755 "$RUN" + +cat > "$UNIT" </dev/null | grep -q "Linger=yes"; then + echo "To keep it running after you log out and after a restart, run once:" + echo " sudo loginctl enable-linger $USER" +fi diff --git a/deploy/server/docker-compose.yml b/deploy/server/docker-compose.yml index a09616e..70ad677 100644 --- a/deploy/server/docker-compose.yml +++ b/deploy/server/docker-compose.yml @@ -5,6 +5,12 @@ # 4. On the phone open https://your-domain, create the account (first time), and use it. # AI keys: set them in .env, or set ANYBOT_RESTORE_FROM_ACCOUNT=1 to start from the newest backup # your Mac made (bots, chats, routines and the keys you set there). +# +# 大腦: this AnyBot is the brain of the account. It keeps the bots, chats, memory and routines, +# and works even while your computers sleep. On each of your computers: 設定 › 手機與帳號 › +# 大腦 › 「改用它當大腦」. Those computers then show these bots, and the bots can use them as hands +# (screen, files, browser), asking you first as usual. A Linux computer without a screen can be +# hands too: deploy/linux/hands.sh. services: relay: build: diff --git a/mobile/index.html b/mobile/index.html index 464e3f1..00f5580 100644 --- a/mobile/index.html +++ b/mobile/index.html @@ -298,11 +298,20 @@ $("b-out1").onclick = () => signOut(); async function pickHost() { const me = await api("relay/me"); - const hosts = me.hosts || []; + const all = me.hosts || []; + // A computer that shows another one's bots (大腦) leads to that brain. + const byId = Object.fromEntries(all.map((h) => [h.id, h])); + const target = (id) => (byId[id] && byId[id].uses && byId[byId[id].uses]) ? byId[id].uses : id; + const hosts = all.filter((h) => !(h.uses && byId[h.uses])); + const sharers = (id) => all.filter((h) => h.uses === id).map((h) => h.name); const live = hosts.filter((h) => h.online); - if (S.host && hosts.some((h) => h.id === S.host && h.online)) return start(); + if (S.host) { + const t = target(S.host); + if (byId[t] && byId[t].online) { S.host = t; localStorage.setItem("anybot.host", t); return start(); } + } if (live.length === 1) { S.host = live[0].id; localStorage.setItem("anybot.host", S.host); return start(); } - $("hosts").innerHTML = hosts.length ? hosts.map((h) => `
${esc((h.name || "?")[0])}
${esc(h.name)}
${h.online ? "在線上" : "離線（" + esc(h.last_seen_text || "很久沒連線") + "）"}
`).join("") + const note = (h) => { const n = sharers(h.id); return n.length ? " · " + esc(n.join("、")) + " 也用這個大腦" : ""; }; + $("hosts").innerHTML = hosts.length ? hosts.map((h) => `
${esc((h.name || "?")[0])}
${esc(h.name)}
${h.online ? "在線上" : "離線（" + esc(h.last_seen_text || "很久沒連線") + "）"}${note(h)}
`).join("") : `
這個帳號還沒有連上任何 AnyBot。
在電腦上打開 AnyBot › 設定 › 手機與帳號，登入同一個帳號。
`; document.querySelectorAll("#hosts .row").forEach((r) => r.onclick = () => { S.host = r.dataset.h; localStorage.setItem("anybot.host", S.host); start(); }); show("hosts"); diff --git a/relay/Caddyfile b/relay/Caddyfile index 0487d13..0be4b12 100644 --- a/relay/Caddyfile +++ b/relay/Caddyfile @@ -3,9 +3,12 @@ anybot.example.com { encode gzip reverse_proxy relay:8080 { - # The phone's "anything new?" request waits up to 35 s. + # The phone's "anything new?" request waits up to 35 s. Each AnyBot also keeps one + # WebSocket open here (pinged every 15–20 s, so it never sits idle for 60 s). transport http { read_timeout 60s } + # Reloading this file doesn't cut those connections at once. + stream_close_delay 5m } } diff --git a/relay/Cargo.lock b/relay/Cargo.lock index 16a817c..056ab2d 100644 --- a/relay/Cargo.lock +++ b/relay/Cargo.lock @@ -26,6 +26,7 @@ dependencies = [ "ring", "serde", "serde_json", + "tungstenite", "ureq", ] @@ -47,6 +48,15 @@ version = "0.23.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -75,6 +85,17 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + [[package]] name = "chrono" version = "0.4.45" @@ -88,12 +109,27 @@ dependencies = [ "windows-link", ] +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "core-foundation-sys" version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + [[package]] name = "crc32fast" version = "1.5.4" @@ -103,6 +139,32 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", +] + [[package]] name = "find-msvc-tools" version = "0.1.15" @@ -155,6 +217,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "rand_core", +] + [[package]] name = "http" version = "1.5.0" @@ -171,6 +245,15 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -285,6 +368,29 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "ring" version = "0.17.14" @@ -293,7 +399,7 @@ checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" dependencies = [ "cc", "cfg-if", - "getrandom", + "getrandom 0.2.17", "libc", "untrusted", "windows-sys", @@ -383,6 +489,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "2.0.1" @@ -429,6 +546,48 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "tungstenite" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e48ac77174b19c110a50ab2128b24215ac9cb40e0e12e093fb602d175c569d22" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand", + "sha1", + "thiserror", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.27" diff --git a/relay/Cargo.toml b/relay/Cargo.toml index b969f28..28f04dc 100644 --- a/relay/Cargo.toml +++ b/relay/Cargo.toml @@ -12,6 +12,7 @@ ring = "0.17" base64 = "0.22" chrono = "0.4" ureq = "3" +tungstenite = { version = "0.30", default-features = false, features = ["handshake"] } [profile.release] diff --git a/relay/src/main.rs b/relay/src/main.rs index 8d2e88f..8c95480 100644 --- a/relay/src/main.rs +++ b/relay/src/main.rs @@ -14,6 +14,8 @@ #[path = "../../src/httpd.rs"] mod httpd; +#[path = "../../src/wire.rs"] +mod wire; mod push; use httpd::{Req, Resp}; @@ -22,6 +24,7 @@ use serde_json::{json, Value}; use std::collections::{HashMap, VecDeque}; use std::path::PathBuf; use std::sync::mpsc::{channel, Sender}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; @@ -35,6 +38,11 @@ const HOST_POLL_SECS: u64 = 25; const FORWARD_SECS: u64 = 35; /// How long a request may wait to be picked up before the host counts as gone. const PICKUP_SECS: u64 = 8; +/// The lasting connection: a ping this often, and given up after this long without a word. +const WS_PING_SECS: u64 = 20; +const WS_DEAD_SECS: u64 = 60; +const GONE: &str = "那台 AnyBot 斷線了（網路中斷，或 AnyBot 關掉了）"; +const OFFLINE: &str = "那台 AnyBot 目前離線（電腦可能關機、睡眠、斷網，或 AnyBot 沒開）"; /// A host counts as online when it asked within this many seconds. const ONLINE_SECS: u64 = 60; /// Sessions last this long without use. @@ -124,6 +132,51 @@ self.addEventListener("notificationclick", (e) => { }); "#; +/// A forwarded request's answer. +struct Answer { + status: u16, + ctype: String, + headers: Vec<(String, String)>, + body: Vec, + /// The body is deflate-compressed (only when the asker said it can read that). + z: bool, +} + +impl Answer { + fn error(status: u16, msg: &str) -> Answer { + Answer { status, ctype: "application/json; charset=utf-8".into(), headers: vec![], body: json!({"error": msg}).to_string().into_bytes(), z: false } + } + fn from_json(v: &Value) -> Answer { + use base64::Engine; + Answer { + status: v["status"].as_u64().unwrap_or(502) as u16, + ctype: v["ctype"].as_str().unwrap_or("application/octet-stream").to_string(), + headers: v["headers"].as_array().into_iter().flatten().filter_map(|h| Some((h[0].as_str()?.to_string(), h[1].as_str()?.to_string()))).collect(), + body: base64::engine::general_purpose::STANDARD.decode(v["body"].as_str().unwrap_or("")).unwrap_or_default(), + z: v["z"].as_bool().unwrap_or(false), + } + } + fn into_resp(self) -> Resp { + let mut resp = Resp::bytes(&self.ctype, self.body); + resp.status = self.status; + for (k, v) in self.headers { + // Only harmless headers pass through. + if k.eq_ignore_ascii_case("x-size") { + resp.headers.push((k, v)); + } + } + resp + } +} + +/// An AnyBot's lasting connection: frames to write to it. +struct Sock { + tx: Sender>, + serial: u64, +} + +static SERIAL: AtomicU64 = AtomicU64::new(1); + struct Relay { dir: PathBuf, data: Mutex, @@ -131,7 +184,12 @@ struct Relay { queues: Mutex>>, arrived: Condvar, /// Where each forwarded request's answer goes. - waiting: Mutex>>, + waiting: Mutex>>, + /// AnyBots connected the lasting way (WebSocket), and which requests went over which. + sockets: Mutex>, + flights: Mutex>, + /// The brain each AnyBot shows (when it is a screen of another). + uses: Mutex>, /// When each host last asked (in memory; saved now and then). seen: Mutex>, fails: Mutex>, @@ -155,6 +213,9 @@ impl Relay { queues: Mutex::new(HashMap::new()), arrived: Condvar::new(), waiting: Mutex::new(HashMap::new()), + sockets: Mutex::new(HashMap::new()), + flights: Mutex::new(HashMap::new()), + uses: Mutex::new(HashMap::new()), seen: Mutex::new(HashMap::new()), fails: Mutex::new(HashMap::new()), invite: std::env::var("ANYBOT_RELAY_INVITE").unwrap_or_default(), @@ -201,7 +262,7 @@ impl Relay { } fn online(&self, id: &str) -> bool { - self.seen.lock().unwrap().get(id).map(|t| t.elapsed().as_secs() < ONLINE_SECS).unwrap_or(false) + self.sockets.lock().unwrap().contains_key(id) || self.seen.lock().unwrap().get(id).map(|t| t.elapsed().as_secs() < ONLINE_SECS).unwrap_or(false) } fn too_many_fails(&self, key: &str) -> bool { @@ -221,14 +282,36 @@ impl Relay { /// Hand a request to a host and wait for its answer. fn forward(&self, host_id: &str, method: &str, path: &str, query: &HashMap, body: &[u8], who: (&str, &str)) -> Resp { + self.forward_raw(host_id, method, path, query, body, false, who, false).into_resp() + } + + /// `z`: the body is compressed; `z_ok`: the asker reads a compressed answer. + #[allow(clippy::too_many_arguments)] + fn forward_raw(&self, host_id: &str, method: &str, path: &str, query: &HashMap, body: &[u8], z: bool, who: (&str, &str), z_ok: bool) -> Answer { use base64::Engine; if !self.online(host_id) { - return Resp::error(503, "那台 AnyBot 目前離線（電腦關機、睡眠，或 AnyBot 沒開）"); + return Answer::error(503, OFFLINE); } let id = httpd::random_hex(12); let (tx, rx) = channel(); self.waiting.lock().unwrap().insert(id.clone(), tx); - let job = json!({"id": id, "method": method, "path": path, "query": query, "body": base64::engine::general_purpose::STANDARD.encode(body), "user": who.0, "role": who.1}); + let head = json!({"t": "job", "id": id, "method": method, "path": path, "query": query, "user": who.0, "role": who.1, "z": z, "z_ok": z_ok}); + // Connected the lasting way: straight down that connection. + let sock = self.sockets.lock().unwrap().get(host_id).map(|s| (s.tx.clone(), s.serial)); + if let Some((stx, serial)) = sock { + self.flights.lock().unwrap().insert(id.clone(), (host_id.to_string(), serial)); + let sent = stx.send(wire::pack(&head, body)).is_ok(); + let r = if sent { rx.recv_timeout(Duration::from_secs(FORWARD_SECS)).ok() } else { None }; + self.waiting.lock().unwrap().remove(&id); + self.flights.lock().unwrap().remove(&id); + return match r { + Some(a) => a, + None if !sent => Answer::error(503, GONE), + None => Answer::error(504, "那台 AnyBot 收到了，但太久沒有回應（可能太忙，或網路很慢）"), + }; + } + let mut job = head; + job["body"] = json!(base64::engine::general_purpose::STANDARD.encode(body)); self.queues.lock().unwrap().entry(host_id.to_string()).or_default().push_back(job); self.arrived.notify_all(); // A running AnyBot always has a worker waiting, so it picks a request up within a moment. @@ -236,35 +319,146 @@ impl Relay { let mut r = rx.recv_timeout(Duration::from_secs(PICKUP_SECS)); if r.is_err() { let queued = self.queues.lock().unwrap().get(host_id).map(|q| q.iter().any(|j| j["id"].as_str() == Some(id.as_str()))).unwrap_or(false); - if queued { - self.waiting.lock().unwrap().remove(&id); - if let Some(q) = self.queues.lock().unwrap().get_mut(host_id) { - q.retain(|j| j["id"].as_str() != Some(id.as_str())); - } - return Resp::error(503, "那台 AnyBot 目前離線（電腦可能關機、睡眠、斷網，或 AnyBot 沒開）"); + if !queued { + r = rx.recv_timeout(Duration::from_secs(FORWARD_SECS - PICKUP_SECS)); } - r = rx.recv_timeout(Duration::from_secs(FORWARD_SECS - PICKUP_SECS)); } self.waiting.lock().unwrap().remove(&id); // Not picked up in time: drop it so it doesn't run late. - if let Some(q) = self.queues.lock().unwrap().get_mut(host_id) { + let was_queued = self.queues.lock().unwrap().get_mut(host_id).map(|q| { + let n = q.len(); q.retain(|j| j["id"].as_str() != Some(id.as_str())); - } + n != q.len() + }); match r { - Ok(v) => { - let mut resp = Resp::bytes(v["ctype"].as_str().unwrap_or("application/octet-stream"), base64::engine::general_purpose::STANDARD.decode(v["body"].as_str().unwrap_or("")).unwrap_or_default()); - resp.status = v["status"].as_u64().unwrap_or(502) as u16; - for h in v["headers"].as_array().into_iter().flatten() { - if let (Some(k), Some(val)) = (h[0].as_str(), h[1].as_str()) { - // Only harmless headers pass through. - if k.eq_ignore_ascii_case("x-size") { - resp.headers.push((k.into(), val.into())); - } - } + Ok(a) => a, + Err(_) if was_queued == Some(true) => Answer::error(503, OFFLINE), + Err(_) => Answer::error(504, "那台 AnyBot 收到了，但太久沒有回應（可能太忙，或網路很慢）"), + } + } + + /// 長連線: an AnyBot keeps one WebSocket open; requests for it come down it at once, and its + /// own requests to the account's other AnyBots go up it. + fn ws(self: &Arc, req: Req, mut c: std::net::TcpStream) { + use std::io::Write; + use tungstenite::{protocol::Role, Message, WebSocket}; + if req.path != "/relay/host/ws" { + return httpd::write_response(&mut c, &Resp::error(404, "沒有這個網址")); + } + let Some(h) = self.host(&req) else { return httpd::write_response(&mut c, &Resp::error(401, "這台電腦的金鑰無效")) }; + let key = req.header("sec-websocket-key").to_string(); + if key.is_empty() { + return httpd::write_response(&mut c, &Resp::error(400, "bad request")); + } + let accept = tungstenite::handshake::derive_accept_key(key.as_bytes()); + if c.write_all(format!("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\n\r\n").as_bytes()).is_err() { + return; + } + let _ = c.set_read_timeout(Some(Duration::from_millis(40))); + let _ = c.set_nodelay(true); + let cfg = tungstenite::protocol::WebSocketConfig::default().max_message_size(Some(httpd::MAX_BODY * 2)).max_frame_size(Some(httpd::MAX_BODY * 2)); + let mut ws = WebSocket::from_raw_socket(c, Role::Server, Some(cfg)); + let serial = SERIAL.fetch_add(1, Ordering::SeqCst); + let (tx, rx) = channel::>(); + self.sockets.lock().unwrap().insert(h.id.clone(), Sock { tx: tx.clone(), serial }); + self.seen.lock().unwrap().insert(h.id.clone(), Instant::now()); + let (mut heard, mut pinged, mut marked) = (Instant::now(), Instant::now(), Instant::now()); + 'live: loop { + match ws.read() { + Ok(Message::Binary(b)) => { + heard = Instant::now(); + self.on_frame(&h, &b, &tx); } - resp + Ok(Message::Close(_)) => break, + Ok(_) => heard = Instant::now(), + Err(tungstenite::Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {} + Err(_) => break, } - Err(_) => Resp::error(504, "那台 AnyBot 收到了，但太久沒有回應（可能太忙，或網路很慢）"), + while let Ok(f) = rx.try_recv() { + if ws.write(Message::Binary(f.into())).is_err() { + break 'live; + } + } + if pinged.elapsed().as_secs() >= WS_PING_SECS { + pinged = Instant::now(); + if ws.write(Message::Ping(Vec::new().into())).is_err() { + break; + } + } + match ws.flush() { + Err(tungstenite::Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {} + Err(_) => break, + Ok(()) => {} + } + if heard.elapsed().as_secs() >= WS_DEAD_SECS { + break; + } + if marked.elapsed().as_secs() >= 10 { + marked = Instant::now(); + self.seen.lock().unwrap().insert(h.id.clone(), Instant::now()); + } + } + // Gone: whoever waits on it hears so now, not after a timeout. + { + let mut socks = self.sockets.lock().unwrap(); + if socks.get(&h.id).map(|s| s.serial) == Some(serial) { + socks.remove(&h.id); + self.seen.lock().unwrap().remove(&h.id); + } + } + let lost: Vec = self.flights.lock().unwrap().iter().filter(|(_, (host, s))| *host == h.id && *s == serial).map(|(id, _)| id.clone()).collect(); + for id in lost { + if let Some(w) = self.waiting.lock().unwrap().remove(&id) { + let _ = w.send(Answer::error(503, GONE)); + } + } + } + + fn on_frame(self: &Arc, h: &Host, b: &[u8], tx: &Sender>) { + let Some((head, body)) = wire::unpack(b) else { return }; + let s = |k: &str| head[k].as_str().unwrap_or("").to_string(); + match head["t"].as_str().unwrap_or("") { + "reply" => { + if let Some(w) = self.waiting.lock().unwrap().remove(&s("id")) { + let headers = head["headers"].as_array().into_iter().flatten().filter_map(|x| Some((x[0].as_str()?.to_string(), x[1].as_str()?.to_string()))).collect(); + let _ = w.send(Answer { status: head["status"].as_u64().unwrap_or(502) as u16, ctype: s("ctype"), headers, body, z: head["z"].as_bool().unwrap_or(false) }); + } + } + // One AnyBot asks another of the same account (大腦 sync, 裝置連結). + "call" => { + let (id, to, path) = (s("id"), s("to"), s("path")); + let role = if path == "/api/device/run" { + "device" + } else if path == "/api/sync" || path.starts_with("/api/sync/") { + "mirror" + } else { + let a = Answer::error(404, "沒有這個網址"); + let _ = tx.send(wire::pack(&json!({"t": "ret", "id": id, "status": a.status, "ctype": a.ctype}), &a.body)); + return; + }; + let same = self.data.lock().unwrap().hosts.iter().any(|x| x.id == to && x.user == h.user); + if !same || to == h.id { + let a = Answer::error(403, "不是同一個帳號的電腦"); + let _ = tx.send(wire::pack(&json!({"t": "ret", "id": id, "status": a.status, "ctype": a.ctype}), &a.body)); + return; + } + let query: HashMap = head["query"].as_object().map(|o| o.iter().map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string())).collect()).unwrap_or_default(); + let (me, tx, user, method, z) = (self.clone(), tx.clone(), h.user.clone(), s("method"), head["z"].as_bool().unwrap_or(false)); + std::thread::spawn(move || { + let a = me.forward_raw(&to, &method, &path, &query, &body, z, (&user, role), true); + let _ = tx.send(wire::pack(&json!({"t": "ret", "id": id, "status": a.status, "ctype": a.ctype, "z": a.z}), &a.body)); + }); + } + "hello" => { + let brain = s("brain"); + let mut u = self.uses.lock().unwrap(); + if brain.is_empty() { + u.remove(&h.id); + } else { + u.insert(h.id.clone(), brain); + } + } + _ => {} } } @@ -389,7 +583,7 @@ impl Relay { return Resp::error(401, "這台電腦的金鑰無效"); } if let Some(tx) = self.waiting.lock().unwrap().remove(*id) { - let _ = tx.send(req.json()); + let _ = tx.send(Answer::from_json(&req.json())); } Resp::json(&json!({"ok": true})) } @@ -542,7 +736,7 @@ impl Relay { fn host_json(&self, h: &Host) -> Value { let online = self.online(&h.id); - json!({"id": h.id, "name": h.name, "online": online, "last_seen_text": if online { "在線上".to_string() } else if h.last_seen > 0 { chrono::DateTime::from_timestamp(h.last_seen, 0).map(|t| t.format("%m/%d %H:%M UTC").to_string()).unwrap_or_default() } else { "從未連線".into() }}) + json!({"id": h.id, "name": h.name, "online": online, "uses": self.uses.lock().unwrap().get(&h.id).cloned().unwrap_or_default(), "last_seen_text": if online { "在線上".to_string() } else if h.last_seen > 0 { chrono::DateTime::from_timestamp(h.last_seen, 0).map(|t| t.format("%m/%d %H:%M UTC").to_string()).unwrap_or_default() } else { "從未連線".into() }}) } fn signup(&self, req: &Req) -> Resp { @@ -630,8 +824,8 @@ fn main() { std::thread::sleep(Duration::from_secs(60)); r2.save_seen(); }); - let r = relay.clone(); - httpd::serve(listener, Arc::new(move |req| r.handle(req)), Arc::new(|| true)); + let (r, w) = (relay.clone(), relay.clone()); + httpd::serve_upgrade(listener, Arc::new(move |req| r.handle(req)), Some(Arc::new(move |req, c| w.ws(req, c))), Arc::new(|| true)); } #[cfg(test)] diff --git a/src/app.rs b/src/app.rs index f46266d..a01eeab 100644 --- a/src/app.rs +++ b/src/app.rs @@ -518,6 +518,11 @@ impl App { } } app.apply_theme(ctx); + // Real-data screenshots can open 設定 › 手機與帳號 (to press its buttons with ANYBOT_SHOT_CLICK). + if app.screenshot.is_some() && !app.demo && std::env::var("ANYBOT_SHOT_VIEW").as_deref() == Ok("remote") { + app.show_settings = true; + app.settings_tab = SettingsTab::Remote; + } app } @@ -1245,6 +1250,20 @@ impl App { } fn fetch_models(&mut self, kind: ProviderKind) { + // A screen of a brain: the brain has the key, so it asks. + if self.mir.on { + let (tx, rx) = channel(); + let s = self.state.remote.clone(); + std::thread::spawn(move || { + let name = serde_json::to_value(kind).unwrap_or_default(); + let r = crate::remote::sync_post(&s, "models", &serde_json::json!({"provider": name})) + .and_then(|v| v["error"].as_str().map(|e| Err(e.to_string())).unwrap_or_else(|| Ok(v["models"].as_array().into_iter().flatten().filter_map(|m| m.as_str().map(String::from)).collect()))); + let _ = tx.send(r); + }); + self.model_fetch.insert(kind, rx); + self.fetch_status.insert(kind, "正在請大腦取得模型清單…".into()); + return; + } let ps = self.state.provider(kind); let base = providers::resolve_base(kind, &ps.base_url); let key = providers::resolve_key(kind, &ps.api_key); @@ -3008,6 +3027,7 @@ impl App { // A screen of another AnyBot's brain: its bots work there, not here. if self.mir.on { self.remote_tick(ctx); + self.poll_fetches(); return; } self.drain_events(ctx); @@ -3187,7 +3207,7 @@ impl eframe::App for App { /// Screenshot mode only: ANYBOT_SHOT_CLICK="x,y@frame[;x,y@frame…][:right]" clicks there at that /// frame (to check buttons really respond). fn raw_input_hook(&mut self, _ctx: &egui::Context, raw: &mut egui::RawInput) { - if !self.demo { + if self.screenshot.is_none() { return; } // ANYBOT_SHOT_DRAG="x1,y1>x2,y2@frame": press, move over 12 frames, release. diff --git a/src/app/attach_ui.rs b/src/app/attach_ui.rs index 032bb61..97b72c6 100644 --- a/src/app/attach_ui.rs +++ b/src/app/attach_ui.rs @@ -8,7 +8,7 @@ use crate::attach::{self, Attachment}; pub struct AttachUi { pub pending: HashMap>, picked: Option)>>, - thumbs: HashMap>, + pub(super) thumbs: HashMap>, pub error: Option, voice: Option<(u64, crate::voice::Recording)>, voice_rx: Option)>>, diff --git a/src/app/computers_ui.rs b/src/app/computers_ui.rs index 1820a05..6a5be14 100644 --- a/src/app/computers_ui.rs +++ b/src/app/computers_ui.rs @@ -351,7 +351,14 @@ impl App { // ── AnyBot Computer Use: seeing and using this computer's screen ── self.mac_permissions_card(ui); - // ── Bot 的電腦 ── + // ── Bot 的電腦 ── (on a screen of a brain, it is the brain's) + if self.mir.on { + sec_title(ui, "Bot 的電腦", p, |_| {}); + sec_card(ui, p, |ui| { + let name = self.state.remote.brain_name.clone(); + sec_row(ui, p, "在大腦上", &format!("Bot 用的電腦在大腦「{name}」上，要在那台管理。這台電腦是大腦的手腳：上面的權限決定 Bot 能不能看這台的畫面、操作這台。"), |_| {}); + }); + } else { sec_title(ui, "Bot 的電腦", p, |ui| status_pill(ui, &run, p)); #[derive(PartialEq)] enum Act { @@ -656,6 +663,8 @@ impl App { } } + } + // ── 示範教學、瀏覽器 ── ui.add_space(SEC_GAP * 0.5); ui.spacing_mut().item_spacing.y = 6.0; diff --git a/src/app/mirror_ui.rs b/src/app/mirror_ui.rs index 19099ab..685a15d 100644 --- a/src/app/mirror_ui.rs +++ b/src/app/mirror_ui.rs @@ -115,6 +115,81 @@ fn hash_value(v: &Value) -> u64 { h.finish() } +/// What names an item in a list: its id, else its name, else (plain values) the value itself. +fn item_key(x: &Value) -> Option { + match x { + Value::Object(o) => o.get("id").filter(|v| !v.is_null()).or_else(|| o.get("name").filter(|v| v.is_string())).cloned(), + Value::Array(_) | Value::Null => None, + v => Some(v.clone()), + } +} + +/// Every item has its own key, and no two share one. +fn keyed(a: &[Value]) -> bool { + let mut seen = std::collections::HashSet::new(); + a.iter().all(|x| item_key(x).map(|k| seen.insert(k.to_string())).unwrap_or(false)) +} + +/// The smallest change from `old` to `new`: lists of named things and maps change item by item, +/// so two computers editing different items of one list don't undo each other. +pub(super) fn patch_of(old: &Value, new: &Value) -> Value { + match (old, new) { + (Value::Object(o), Value::Object(n)) => { + let put: serde_json::Map = n.iter().filter(|(k, v)| o.get(*k) != Some(*v)).map(|(k, v)| (k.clone(), v.clone())).collect(); + let del: Vec<&String> = o.keys().filter(|k| !n.contains_key(*k)).collect(); + json!({"map": {"put": put, "del": del}}) + } + (Value::Array(o), Value::Array(n)) if keyed(o) && keyed(n) => { + let key = |x: &Value| item_key(x).unwrap_or(Value::Null); + let put: Vec<&Value> = n.iter().filter(|x| !o.iter().any(|y| key(y) == key(x) && y == *x)).collect(); + let del: Vec = o.iter().map(key).filter(|k| !n.iter().any(|x| key(x) == *k)).collect(); + let common_old: Vec = o.iter().map(key).filter(|k| n.iter().any(|x| key(x) == *k)).collect(); + let common_new: Vec = n.iter().map(key).filter(|k| o.iter().any(|x| key(x) == *k)).collect(); + let mut p = json!({"list": {"put": put, "del": del}}); + if common_old != common_new { + p["list"]["order"] = Value::Array(n.iter().map(key).collect()); + } + p + } + _ => json!({"value": new}), + } +} + +/// Apply a change made by `patch_of` to whatever the value is now. +pub(super) fn apply_patch(cur: &mut Value, p: &Value) { + if let Some(v) = p.get("value") { + *cur = v.clone(); + } else if let Some(m) = p.get("map") { + if !cur.is_object() { + *cur = json!({}); + } + let o = cur.as_object_mut().expect("object"); + for (k, v) in m["put"].as_object().into_iter().flatten() { + o.insert(k.clone(), v.clone()); + } + for k in m["del"].as_array().into_iter().flatten().filter_map(Value::as_str) { + o.remove(k); + } + } else if let Some(l) = p.get("list") { + if !cur.is_array() { + *cur = json!([]); + } + let a = cur.as_array_mut().expect("array"); + let key = |x: &Value| item_key(x).unwrap_or(Value::Null); + for item in l["put"].as_array().into_iter().flatten() { + match a.iter_mut().find(|x| key(x) == key(item)) { + Some(x) => *x = item.clone(), + None => a.push(item.clone()), + } + } + let del: Vec = l["del"].as_array().cloned().unwrap_or_default(); + a.retain(|x| !del.contains(&key(x))); + if let Some(order) = l["order"].as_array() { + a.sort_by_key(|x| order.iter().position(|k| *k == key(x)).unwrap_or(usize::MAX)); + } + } +} + /// The person's edits since `base`, as operations the brain applies one by one. pub(super) fn diff_ops(base: &Value, cur: &Value) -> Vec { let mut ops = vec![]; @@ -124,7 +199,8 @@ pub(super) fn diff_ops(base: &Value, cur: &Value) -> Vec { continue; } if b.get(k) != Some(v) { - ops.push(json!({"op": "set", "key": k, "value": v})); + let p = patch_of(b.get(k).unwrap_or(&Value::Null), v); + ops.push(if p.get("value").is_some() { json!({"op": "set", "key": k, "value": v}) } else { json!({"op": "patch", "key": k, "patch": p}) }); } } let id = |x: &Value| x["id"].as_u64().unwrap_or(0); @@ -146,7 +222,8 @@ pub(super) fn diff_ops(base: &Value, cur: &Value) -> Vec { continue; } if ob.get(k) != Some(v) { - ops.push(json!({"op": "bot_set", "id": id(nb), "key": k, "value": v})); + let p = patch_of(ob.get(k).unwrap_or(&Value::Null), v); + ops.push(if p.get("value").is_some() { json!({"op": "bot_set", "id": id(nb), "key": k, "value": v}) } else { json!({"op": "bot_patch", "id": id(nb), "key": k, "patch": p}) }); } } } @@ -179,6 +256,16 @@ pub(super) fn apply_ops(cfg: &mut Value, ops: &[Value]) { cfg[k] = op["value"].clone(); } } + "patch" => { + if let Some(k) = op["key"].as_str() { + apply_patch(&mut cfg[k], &op["patch"]); + } + } + "bot_patch" => { + if let (Some(b), Some(k)) = (cfg["bots"].as_array_mut().and_then(|a| a.iter_mut().find(|b| id(b) == target)), op["key"].as_str()) { + apply_patch(&mut b[k], &op["patch"]); + } + } "bot_add" => { if let Some(a) = cfg["bots"].as_array_mut() { if !a.iter().any(|b| id(b) == id(&op["bot"])) { @@ -254,6 +341,8 @@ struct Seen { pub enum Incoming { Pull(Value), Offline(String), + /// A file from the brain arrived (its local copy's path). + File(String), } /// What a screen sends to the brain, one at a time and in order. @@ -298,6 +387,15 @@ pub struct Mirror { /// Why the brain can't be reached (shown above the chat), and since when. pub(super) reason: String, down_since: Option, + /// Files from the brain being fetched (by local path). + fetching: std::collections::HashSet, +} + +/// Where a screen keeps its copy of a file the brain has. +fn local_copy(brain_path: &str, name: &str) -> PathBuf { + let h = crate::httpd::sha256_hex(brain_path); + let name: String = name.chars().filter(|c| !matches!(c, '/' | '\\' | '\0' | ':')).take(80).collect(); + State::path().with_file_name("brain-files").join(format!("{}-{name}", &h[..16])) } fn cache_path() -> PathBuf { @@ -314,6 +412,25 @@ impl App { for r in reqs { let v = match r.kind { SyncKind::Pull { client, since } => self.brain_pull(&client, since), + SyncKind::Models(name) => { + let kind = serde_json::from_value::(json!(name)).ok(); + match kind { + Some(kind) => { + let ps = self.state.provider(kind); + let (base, key) = (crate::providers::resolve_base(kind, &ps.base_url), crate::providers::resolve_key(kind, &ps.api_key)); + let reply = r.reply; + // The list can take a while; the app doesn't wait for it. + std::thread::spawn(move || { + let _ = reply.send(match crate::providers::list_models(kind, &base, &key) { + Ok(models) => json!({"models": models}), + Err(e) => json!({"error": e}), + }); + }); + continue; + } + None => json!({"error": "大腦不認得這個模型服務（兩邊 AnyBot 版本不同？）"}), + } + } SyncKind::Edit(body) => { if self.mir.on { json!({"error": "這台 AnyBot 本身連著別的大腦"}) @@ -430,6 +547,24 @@ impl App { } } "merge" => self.brain_merge_top(op["key"].as_str().unwrap_or(""), &op["value"]), + "patch" => { + let key = op["key"].as_str().unwrap_or(""); + let mut v = self.top_value(key); + apply_patch(&mut v, &op["patch"]); + self.brain_set_top(key, v) + } + "bot_patch" => { + let (id, key) = (op["id"].as_u64().unwrap_or(0), op["key"].as_str().unwrap_or("")); + match self.state.bot(id).map(|b| serde_json::to_value(b).ok().map(|v| v[key].clone()).unwrap_or(Value::Null)) { + Some(mut v) => { + let old = v.clone(); + apply_patch(&mut v, &op["patch"]); + unredact(&mut v, &old); + self.brain_set_bot(id, key, v) + } + None => false, + } + } "bot_set" => self.brain_set_bot(op["id"].as_u64().unwrap_or(0), op["key"].as_str().unwrap_or(""), op["value"].clone()), "bot_read" => match self.state.bot_mut(op["id"].as_u64().unwrap_or(0)) { Some(b) => { @@ -467,6 +602,14 @@ impl App { json!({"ok": true, "applied": applied}) } + /// One top-level setting as JSON (with the brain's real keys). + fn top_value(&mut self, key: &str) -> Value { + let bots = std::mem::take(&mut self.state.bots); + let v = serde_json::to_value(&self.state).map(|t| t[key].clone()).unwrap_or(Value::Null); + self.state.bots = bots; + v + } + fn brain_set_top(&mut self, key: &str, mut value: Value) -> bool { if key.is_empty() || key == "bots" || LOCAL_KEYS.contains(&key) { return false; @@ -583,6 +726,7 @@ impl App { *self.mir.status.lock().unwrap() = "正在連到大腦…".into(); self.mirror_worker(); self.mirror_outbox(); + crate::tunnel::hello(&self.state.remote.brain_id); } /// Back to this computer's own bots. @@ -610,6 +754,7 @@ impl App { } self.state.save(); self.dirty = false; + crate::tunnel::hello(""); } fn keep_local_fields(&mut self, from: &State) { @@ -706,6 +851,10 @@ impl App { *self.mir.status.lock().unwrap() = format!("連不到大腦：{e}"); self.mir.reason = e; } + Incoming::File(path) => { + self.mir.fetching.remove(&path); + self.att.thumbs.remove(&path); + } Incoming::Pull(v) => { if let Some(e) = v["error"].as_str() { self.mir.online = false; @@ -764,7 +913,8 @@ impl App { for d in v["bots"].as_array().into_iter().flatten() { let id = d["id"].as_u64().unwrap_or(0); let from = d["from"].as_u64().unwrap_or(0) as usize; - let msgs: Vec = serde_json::from_value(d["messages"].clone()).unwrap_or_default(); + let mut msgs: Vec = serde_json::from_value(d["messages"].clone()).unwrap_or_default(); + self.mirror_files(&mut msgs); if let Some(b) = self.state.bot_mut(id) { b.messages.truncate(from); b.messages.extend(msgs); @@ -824,6 +974,35 @@ impl App { self.mirror_save_copy(); } + /// Files in the brain's chats live on the brain: show local copies, fetched once. + fn mirror_files(&mut self, msgs: &mut [Message]) { + for a in msgs.iter_mut().flat_map(|m| m.attachments.iter_mut()) { + if std::path::Path::new(&a.path).exists() { + continue; + } + let local = local_copy(&a.path, &a.name); + let lp = local.display().to_string(); + let brain_path = std::mem::replace(&mut a.path, lp.clone()); + if local.exists() || !self.mir.fetching.insert(lp.clone()) { + continue; + } + let (s, tx, repaint) = (self.state.remote.clone(), self.mir.tx.clone(), self.repaint_fn()); + std::thread::spawn(move || { + if let Ok(bytes) = remote::sync_file(&s, &brain_path) { + let _ = std::fs::create_dir_all(local.parent().unwrap_or(std::path::Path::new("."))); + let tmp = local.with_extension("part"); + if std::fs::write(&tmp, bytes).is_ok() { + let _ = std::fs::rename(&tmp, &local); + } + } + if let Some(tx) = tx { + let _ = tx.send(Incoming::File(lp)); + } + repaint(); + }); + } + } + fn mirror_save_copy(&mut self) { if let Ok(s) = serde_json::to_string(&self.state) { let p = cache_path(); @@ -909,7 +1088,8 @@ impl App { } Out::Move { ops, n } => { *note.lock().unwrap() = match remote::sync_post(&s, "edit", &json!({"ops": ops})) { - Ok(v) => format!("已把 {n} 個 Bot 搬到大腦（{} 項）", v["applied"].as_u64().unwrap_or(0)), + Ok(_) if n == 0 => "大腦上已經有這台的 Bot 了；模型金鑰、技能和連接器有缺的已補上".into(), + Ok(_) => format!("已把 {n} 個 Bot（連同聊天）搬到大腦"), Err(e) => format!("搬移失敗：{e}"), }; busy.store(false, Ordering::SeqCst); @@ -957,7 +1137,15 @@ impl App { return; } let mut ops: Vec = vec![]; - for b in &local.bots { + // Skip what the brain already has: the same starter bot never used here, or a bot moved + // before (same name, same chat). + let last = |b: &Bot| b.messages.last().map(|m| (m.at, m.text.clone())); + let moving: Vec<&Bot> = local + .bots + .iter() + .filter(|b| !self.state.bots.iter().any(|o| o.name == b.name && (b.messages.is_empty() || (o.messages.len() == b.messages.len() && last(o) == last(b))))) + .collect(); + for b in moving.iter().copied() { let mut v = serde_json::to_value(b).unwrap_or(Value::Null); v["id"] = json!(0); ops.push(json!({"op": "bot_add", "bot": v})); @@ -966,8 +1154,8 @@ impl App { for k in ["providers", "skills", "mcp_servers"] { ops.push(json!({"op": "merge", "key": k, "value": top[k]})); } + let n = moving.len(); *self.mir.move_note.lock().unwrap() = "搬移中…".into(); - let n = local.bots.len(); self.mirror_out(Out::Move { ops, n }); } @@ -1019,7 +1207,7 @@ mod tests { let cur = json!({"theme": "dark", "folders": ["a", "b"], "bots": [{"id": 1, "name": "A2", "unread": false}, {"id": 3, "name": "C"}]}); let ops = diff_ops(&base, &cur); // theme is this computer's own; folders changed. - assert!(ops.iter().any(|o| o["op"] == "set" && o["key"] == "folders")); + assert!(ops.iter().any(|o| o["op"] == "patch" && o["key"] == "folders" && o["patch"]["list"]["put"] == json!(["b"]))); assert!(!ops.iter().any(|o| o["key"] == "theme")); assert!(ops.iter().any(|o| o["op"] == "bot_set" && o["key"] == "name" && o["value"] == "A2")); assert!(ops.iter().any(|o| o["op"] == "bot_read" && o["id"] == 1)); @@ -1087,4 +1275,35 @@ mod tests { assert!(remote::why_refused(401, &Value::Null, "大腦").contains("重新登入")); assert!(remote::why_unreachable(&ureq::Error::HostNotFound).contains("網路")); } + + #[test] + fn two_computers_edit_one_list_without_undoing_each_other() { + let base = json!({"skills": [{"name": "a", "on": true}, {"name": "b", "on": true}], "providers": {"xai": {"model": "1"}, "openai": {"model": "1"}}, "bots": [{"id": 1, "memories": ["x", "y"]}]}); + // Laptop turns skill a off, changes xai, forgets memory x. + let mut laptop = base.clone(); + laptop["skills"][0]["on"] = json!(false); + laptop["providers"]["xai"]["model"] = json!("2"); + laptop["bots"][0]["memories"] = json!(["y"]); + // Meanwhile the desk computer turned skill b off, changed openai, and the bot learned z. + let mut brain = base.clone(); + brain["skills"][1]["on"] = json!(false); + brain["skills"].as_array_mut().unwrap().push(json!({"name": "c", "on": true})); + brain["providers"]["openai"]["model"] = json!("3"); + brain["bots"][0]["memories"] = json!(["x", "y", "z"]); + let ops = diff_ops(&base, &laptop); + assert!(ops.iter().all(|o| o["op"] != "set" || o["key"] != "skills"), "{ops:?}"); + apply_ops(&mut brain, &ops); + assert_eq!(brain["skills"], json!([{"name": "a", "on": false}, {"name": "b", "on": false}, {"name": "c", "on": true}])); + assert_eq!(brain["providers"], json!({"xai": {"model": "2"}, "openai": {"model": "3"}})); + assert_eq!(brain["bots"][0]["memories"], json!(["y", "z"])); + } + + #[test] + fn lists_without_names_are_sent_whole() { + let p = patch_of(&json!([[1], [2]]), &json!([[2]])); + assert_eq!(p, json!({"value": [[2]]})); + let mut cur = json!([{"id": 1}, {"id": 2}]); + apply_patch(&mut cur, &patch_of(&json!([{"id": 1}, {"id": 2}]), &json!([{"id": 2}, {"id": 1}]))); + assert_eq!(cur, json!([{"id": 2}, {"id": 1}])); + } } diff --git a/src/httpd.rs b/src/httpd.rs index 37141f9..461b3e9 100644 --- a/src/httpd.rs +++ b/src/httpd.rs @@ -178,9 +178,16 @@ pub fn write_response(c: &mut TcpStream, r: &Resp) { } pub type Handler = Arc Resp + Send + Sync>; +/// Takes over a connection that asked to become a WebSocket (it answers the handshake itself). +pub type Upgrade = Arc; /// Answer every connection on this listener with `handler` (each on its own thread). pub fn serve(listener: TcpListener, handler: Handler, alive: Arc bool + Send + Sync>) { + serve_upgrade(listener, handler, None, alive) +} + +/// Like `serve`, and connections asking for a WebSocket go to `upgrade`. +pub fn serve_upgrade(listener: TcpListener, handler: Handler, upgrade: Option, alive: Arc bool + Send + Sync>) { listener.set_nonblocking(true).ok(); loop { if !alive() { @@ -190,7 +197,13 @@ pub fn serve(listener: TcpListener, handler: Handler, alive: Arc boo Ok((mut c, _)) => { c.set_nonblocking(false).ok(); let h = handler.clone(); + let up = upgrade.clone(); std::thread::spawn(move || match read_request(&mut c) { + Ok(req) if req.header("upgrade").eq_ignore_ascii_case("websocket") && up.is_some() => { + if let Some(u) = up { + u(req, c) + } + } Ok(req) => { let resp = h(req); write_response(&mut c, &resp); diff --git a/src/main.rs b/src/main.rs index 27b4ef6..590c12a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -46,7 +46,9 @@ mod triggers; mod vault; mod vault_store; mod voice; +mod tunnel; mod web; +mod wire; use eframe::egui; use std::path::PathBuf; diff --git a/src/remote.rs b/src/remote.rs index b3fd06c..6bf32df 100644 --- a/src/remote.rs +++ b/src/remote.rs @@ -144,6 +144,8 @@ pub struct Hub { pub enum SyncKind { Pull { client: String, since: u64 }, Edit(Value), + /// A provider's model list, fetched with the brain's own key (screens don't have it). + Models(String), } pub struct SyncReq { @@ -169,12 +171,15 @@ impl Hub { } /// Hand a sync request to the app's thread and wait for its answer. fn ask_sync(&self, kind: SyncKind) -> Result { + self.ask_sync_for(kind, 10) + } + fn ask_sync_for(&self, kind: SyncKind, secs: u64) -> Result { let (tx, rx) = std::sync::mpsc::channel(); self.sync.lock().unwrap().push(SyncReq { kind, reply: tx }); if let Some(r) = self.repaint.lock().unwrap().as_ref() { r(); } - rx.recv_timeout(Duration::from_secs(10)).map_err(|_| "大腦太忙，沒有及時回應".to_string()) + rx.recv_timeout(Duration::from_secs(secs)).map_err(|_| "大腦太忙，沒有及時回應".to_string()) } pub fn version(&self) -> u64 { self.view.lock().unwrap().version @@ -383,6 +388,22 @@ pub fn handle(req: Req, via_relay: bool) -> Resp { Err(e) => Resp::error(503, &e), }, ("POST", ["api", "sync", "upload", id]) => upload(id.parse().unwrap_or(0), &req), + ("POST", ["api", "sync", "models"]) => match hub().ask_sync_for(SyncKind::Models(req.json()["provider"].as_str().unwrap_or("").to_string()), 45) { + Ok(v) => Resp::json(&v), + Err(e) => Resp::error(503, &e), + }, + // A file shown in a chat (only what AnyBot keeps as attachments). + ("GET", ["api", "sync", "file"]) => { + let want = std::path::PathBuf::from(req.query.get("path").cloned().unwrap_or_default()); + let root = crate::model::State::path().parent().map(|p| p.join("attachments")).and_then(|p| p.canonicalize().ok()); + match (want.canonicalize().ok(), root) { + (Some(f), Some(root)) if f.starts_with(&root) && f.is_file() => match std::fs::read(&f) { + Ok(b) => Resp::bytes("application/octet-stream", b), + Err(_) => Resp::error(404, "大腦上找不到這個檔案"), + }, + _ => Resp::error(404, "大腦上找不到這個檔案"), + } + } ("POST", ["api", "sync", "act"]) => { let b = req.json(); let bot = b["bot"].as_u64().unwrap_or(0); @@ -642,8 +663,17 @@ pub fn addresses(port: u16) -> Vec { static RELAY_GEN: AtomicU64 = AtomicU64::new(0); +/// One agent per timeout, kept: its connections stay open between requests (no new TCP and TLS +/// handshake each time). fn agent(secs: u64) -> ureq::Agent { - ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(secs))).http_status_as_error(false).build().into() + static AGENTS: OnceLock>> = OnceLock::new(); + AGENTS + .get_or_init(Mutex::default) + .lock() + .unwrap() + .entry(secs) + .or_insert_with(|| ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(secs))).http_status_as_error(false).build().into()) + .clone() } /// Why a request never got an answer, in plain words (never an error code). @@ -746,11 +776,17 @@ pub fn relay_restart(s: &Settings) { } let (url, token) = (s.relay_url.clone(), s.host_token.clone()); hub().set_relay_status("連線中…"); - // A few workers, so a slow request (a long-poll) doesn't block the others. - for _ in 0..4 { - let (url, token) = (url.clone(), token.clone()); - std::thread::spawn(move || relay_worker(&url, &token, gen)); - } + std::thread::spawn(move || { + // 長連線 first; an older account server without it gets asked over and over instead. + if crate::tunnel::run(&url, &token, &|| alive(gen), &|m| hub().set_relay_status(m)) || !alive(gen) { + return; + } + // A few workers, so a slow request (a long-poll) doesn't block the others. + for _ in 0..4 { + let (url, token) = (url.clone(), token.clone()); + std::thread::spawn(move || relay_worker(&url, &token, gen)); + } + }); } fn alive(gen: u64) -> bool { @@ -808,7 +844,14 @@ pub fn req_from_json(job: &Value) -> Req { query: job["query"].as_object().map(|o| o.iter().map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string())).collect()).unwrap_or_default(), // Who asked, as the relay vouches for it (團隊: owner / admin / member; "device" for 裝置連結). headers: [("x-anybot-user", job["user"].as_str().unwrap_or("")), ("x-anybot-role", job["role"].as_str().unwrap_or("owner"))].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(), - body: base64::engine::general_purpose::STANDARD.decode(job["body"].as_str().unwrap_or("")).unwrap_or_default(), + body: { + let b = base64::engine::general_purpose::STANDARD.decode(job["body"].as_str().unwrap_or("")).unwrap_or_default(); + if job["z"].as_bool().unwrap_or(false) { + crate::tunnel::unzip(&b).unwrap_or_default() + } else { + b + } + }, peer: "relay".into(), } } @@ -866,6 +909,10 @@ pub fn run_on_device(host_id: &str, tool: &str, args: &str, who: &str) -> Result if s.relay_url.is_empty() { return Err("沒有登入帳號，連不到其他裝置".into()); } + let body = json!({"tool": tool, "args": args, "who": who}).to_string(); + if let Some(r) = crate::tunnel::call(host_id, "POST", "/api/device/run", &[], body.as_bytes(), 50) { + return r.map_err(|e| format!("連不到那台電腦：{e}")).and_then(|a| answer_value(a.status, &a.body, "那台電腦")); + } let url = format!("{}/relay/hosts/{host_id}/api/device/run", s.relay_url); let mut resp = agent(45) .post(&url) @@ -894,40 +941,66 @@ fn brain_url(s: &Settings, rest: &str) -> Result { Ok(format!("{}/relay/hosts/{}/api/sync{rest}", s.relay_url, s.brain_id)) } -fn read_reply(r: Result, ureq::Error>) -> Result { - let mut resp = r.map_err(|e| format!("連不到帳號伺服器：{}", why_unreachable(&e)))?; - let st = resp.status().as_u16(); - let text = resp.body_mut().with_config().limit(200 * 1024 * 1024).read_to_string().map_err(|_| "和大腦的連線到一半斷掉了".to_string())?; - let v: Value = serde_json::from_str(&text).unwrap_or(Value::Null); +/// A JSON answer, or what went wrong in words. +fn answer_value(st: u16, bytes: &[u8], whom: &str) -> Result { + let v: Value = serde_json::from_slice(bytes).unwrap_or(Value::Null); if !(200..300).contains(&st) { - return Err(why_refused(st, &v, "大腦")); + return Err(why_refused(st, &v, whom)); } if v.is_null() { - return Err("大腦的回應看不懂，請把兩邊的 AnyBot 都更新到最新版".into()); + return Err(format!("{whom}的回應看不懂，請把兩邊的 AnyBot 都更新到最新版")); } Ok(v) } +/// Ask the brain: over the lasting connection when it is open, else an ordinary request. +fn to_brain(s: &Settings, method: &str, rest: &str, query: &[(&str, String)], body: &[u8], secs: u64) -> Result { + let (st, bytes) = to_brain_raw(s, method, rest, query, body, secs)?; + answer_value(st, &bytes, "大腦") +} + +fn to_brain_raw(s: &Settings, method: &str, rest: &str, query: &[(&str, String)], body: &[u8], secs: u64) -> Result<(u16, Vec), String> { + if s.brain_id.is_empty() { + return Err("沒有選大腦".into()); + } + if let Some(r) = crate::tunnel::call(&s.brain_id, method, &format!("/api/sync{rest}"), query, body, secs) { + return r.map(|a| (a.status, a.body)); + } + let q: Vec = query.iter().map(|(k, v)| format!("{k}={}", url_escape(v))).collect(); + let url = brain_url(s, &format!("{rest}{}{}", if q.is_empty() { "" } else { "?" }, q.join("&")))?; + let auth = format!("Bearer {}", s.host_token); + let r = if method == "GET" { agent(secs).get(&url).header("Authorization", &auth).call() } else { agent(secs).post(&url).header("Authorization", &auth).header("Content-Type", "application/octet-stream").send(body) }; + let mut resp = r.map_err(|e| format!("連不到帳號伺服器：{}", why_unreachable(&e)))?; + let st = resp.status().as_u16(); + let bytes = resp.body_mut().with_config().limit(200 * 1024 * 1024).read_to_vec().map_err(|_| "和大腦的連線到一半斷掉了".to_string())?; + Ok((st, bytes)) +} + /// What changed at the brain since `since` (waits up to LONG_POLL_SECS when nothing has). pub fn sync_pull(s: &Settings, since: u64, client: &str) -> Result { - let url = brain_url(s, &format!("?since={since}&client={client}"))?; - read_reply(agent(LONG_POLL_SECS + 25).get(&url).header("Authorization", &format!("Bearer {}", s.host_token)).call()) + to_brain(s, "GET", "", &[("since", since.to_string()), ("client", client.to_string())], &[], LONG_POLL_SECS + 25) } /// Send edits ("edit") or a chat action ("act") to the brain. pub fn sync_post(s: &Settings, what: &str, body: &Value) -> Result { - let url = brain_url(s, &format!("/{what}"))?; - read_reply(agent(60).post(&url).header("Authorization", &format!("Bearer {}", s.host_token)).header("Content-Type", "application/json").send(body.to_string().as_bytes())) + to_brain(s, "POST", &format!("/{what}"), &[], body.to_string().as_bytes(), 60) } /// A file for the next message, kept at the brain; gives back its path there. pub fn sync_upload(s: &Settings, bot: u64, name: &str, bytes: &[u8]) -> Result { - let q: String = url_escape(name); - let url = brain_url(s, &format!("/upload/{bot}?name={q}"))?; - let v = read_reply(agent(120).post(&url).header("Authorization", &format!("Bearer {}", s.host_token)).header("Content-Type", "application/octet-stream").send(bytes))?; + let v = to_brain(s, "POST", &format!("/upload/{bot}"), &[("name", name.to_string())], bytes, 120)?; v["path"].as_str().map(String::from).ok_or("大腦沒有收下檔案".into()) } +/// A file the brain keeps (an attachment shown in a chat), as bytes. +pub fn sync_file(s: &Settings, path: &str) -> Result, String> { + let (st, bytes) = to_brain_raw(s, "GET", "/file", &[("path", path.to_string())], &[], 120)?; + if !(200..300).contains(&st) { + return Err(why_refused(st, &serde_json::from_slice(&bytes).unwrap_or(Value::Null), "大腦")); + } + Ok(bytes) +} + fn url_escape(s: &str) -> String { s.bytes() .map(|b| match b { diff --git a/src/tunnel.rs b/src/tunnel.rs new file mode 100644 index 0000000..4a2d6f3 --- /dev/null +++ b/src/tunnel.rs @@ -0,0 +1,365 @@ +// 長連線: one lasting WebSocket from this AnyBot to the account server. Requests for this AnyBot +// (the phone, a screen of this brain, a brain using this computer's hands) come down it at once; +// this AnyBot's own requests to the account's other AnyBots go up it. Big bodies are compressed. +// A ping every 15 s keeps it (and any proxy on the way) awake; 45 s without a word and it is +// opened again. If the account server is older and has no WebSocket, the old way (asking +// "anything for me?" over and over) is used instead. + +use crate::httpd::Req; +use crate::wire; +use serde_json::{json, Value}; +use std::collections::HashMap; +use std::io::{ErrorKind, Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::mpsc::{channel, Sender}; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, Instant}; +use tungstenite::{Message, WebSocket}; + +const PING_SECS: u64 = 15; +const DEAD_SECS: u64 = 45; +/// Bodies bigger than this are compressed. +const ZIP_OVER: usize = 1024; +const CUT: &str = "和帳號伺服器的連線斷了，正在重新連線"; + +/// Another AnyBot's answer to a call. +pub struct Answer { + pub status: u16, + pub body: Vec, +} + +type Calls = Arc>>>>; + +struct Link { + tx: Sender>, + calls: Calls, + serial: u64, +} + +fn link() -> &'static Mutex> { + static L: OnceLock>> = OnceLock::new(); + L.get_or_init(|| Mutex::new(None)) +} + + +pub fn zip(b: &[u8]) -> Vec { + let mut e = flate2::write::DeflateEncoder::new(Vec::with_capacity(b.len() / 3), flate2::Compression::fast()); + let _ = e.write_all(b); + e.finish().unwrap_or_default() +} + +/// The body to send and whether it is compressed: only when that saves a good part of it +/// (pictures and PDFs are compressed already). +fn maybe_zip(b: Vec) -> (Vec, bool) { + if b.len() <= ZIP_OVER { + return (b, false); + } + let z = zip(&b); + if z.len() * 10 < b.len() * 9 { + (z, true) + } else { + (b, false) + } +} + +pub fn unzip(b: &[u8]) -> Result, String> { + let mut out = vec![]; + flate2::read::DeflateDecoder::new(b).take(512 * 1024 * 1024).read_to_end(&mut out).map_err(|_| "收到的資料壞掉了".to_string())?; + Ok(out) +} + +// ---------- the connection ---------- + +enum Stream { + Plain(TcpStream), + Tls(Box>), +} + +impl Stream { + fn tcp(&self) -> &TcpStream { + match self { + Stream::Plain(t) => t, + Stream::Tls(s) => &s.sock, + } + } +} + +impl Read for Stream { + fn read(&mut self, b: &mut [u8]) -> std::io::Result { + match self { + Stream::Plain(t) => t.read(b), + Stream::Tls(s) => s.read(b), + } + } +} + +impl Write for Stream { + fn write(&mut self, b: &[u8]) -> std::io::Result { + match self { + Stream::Plain(t) => t.write(b), + Stream::Tls(s) => s.write(b), + } + } + fn flush(&mut self) -> std::io::Result<()> { + match self { + Stream::Plain(t) => t.flush(), + Stream::Tls(s) => s.flush(), + } + } +} + +fn tls_config() -> Arc { + static C: OnceLock> = OnceLock::new(); + C.get_or_init(|| { + let roots = rustls::RootCertStore { roots: webpki_roots::TLS_SERVER_ROOTS.to_vec() }; + Arc::new( + rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .expect("TLS versions") + .with_root_certificates(roots) + .with_no_client_auth(), + ) + }) + .clone() +} + +fn would_block(e: &tungstenite::Error) -> bool { + matches!(e, tungstenite::Error::Io(io) if matches!(io.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut)) +} + +/// Why it couldn't connect; `old` = the account server has no WebSocket (use the old way). +struct NoLink { + old: bool, + why: String, +} + +fn open(url: &str, token: &str) -> Result, NoLink> { + use tungstenite::client::IntoClientRequest; + let fail = |why: &str| NoLink { old: false, why: why.to_string() }; + let (tls, rest) = match (url.strip_prefix("https://"), url.strip_prefix("http://")) { + (Some(r), _) => (true, r), + (_, Some(r)) => (false, r), + _ => return Err(fail("帳號伺服器的網址不對")), + }; + let hostport = rest.split('/').next().unwrap_or(""); + let (host, port) = match hostport.rsplit_once(':') { + Some((h, p)) if p.parse::().is_ok() => (h, p.parse().unwrap_or(443)), + _ => (hostport, if tls { 443 } else { 80 }), + }; + let addrs: Vec<_> = (host, port).to_socket_addrs().map_err(|_| fail("找不到伺服器：這台電腦可能沒連上網路，或伺服器網址打錯了"))?.collect(); + // Each address in turn ("localhost" is both ::1 and 127.0.0.1). + let mut last = None; + let mut tcp = None; + for a in &addrs { + match TcpStream::connect_timeout(a, Duration::from_secs(10)) { + Ok(t) => { + tcp = Some(t); + break; + } + Err(e) => last = Some(e.kind()), + } + } + let tcp = tcp.ok_or_else(|| { + fail(match last { + None => "找不到伺服器", + Some(ErrorKind::ConnectionRefused) => "伺服器沒有開（連線被拒絕）", + Some(ErrorKind::TimedOut | ErrorKind::WouldBlock) => "伺服器沒有回應：可能沒開機，或網路把連線擋住了", + _ => "這台電腦沒有連上網路", + }) + })?; + let _ = tcp.set_nodelay(true); + let _ = tcp.set_read_timeout(Some(Duration::from_secs(15))); + let _ = tcp.set_write_timeout(Some(Duration::from_secs(30))); + let stream = if tls { + let name = rustls::pki_types::ServerName::try_from(host.to_string()).map_err(|_| fail("帳號伺服器的網址不對"))?; + let conn = rustls::ClientConnection::new(tls_config(), name).map_err(|_| fail("和伺服器的安全連線失敗"))?; + Stream::Tls(Box::new(rustls::StreamOwned::new(conn, tcp))) + } else { + Stream::Plain(tcp) + }; + let mut req = format!("{}://{hostport}/relay/host/ws", if tls { "wss" } else { "ws" }).into_client_request().map_err(|_| fail("帳號伺服器的網址不對"))?; + if let Ok(v) = format!("Bearer {token}").parse() { + req.headers_mut().insert("Authorization", v); + } + let cfg = tungstenite::protocol::WebSocketConfig::default().max_message_size(Some(512 << 20)).max_frame_size(Some(512 << 20)); + match tungstenite::client::client_with_config(req, stream, Some(cfg)) { + Ok((ws, _)) => Ok(ws), + Err(tungstenite::HandshakeError::Failure(tungstenite::Error::Http(r))) => match r.status().as_u16() { + 401 | 403 => Err(fail("這台電腦的帳號登入失效了，請到「設定 › 手機與帳號」重新登入")), + 400 | 404 | 405 | 426 => Err(NoLink { old: true, why: String::new() }), + _ => Err(fail("帳號伺服器暫時無法使用，稍後會自動再試")), + }, + Err(tungstenite::HandshakeError::Failure(e)) if e.to_string().to_lowercase().contains("certificate") || e.to_string().to_lowercase().contains("tls") => Err(fail("和伺服器的安全連線失敗（伺服器的憑證有問題）")), + Err(_) => Err(fail("連線到一半斷掉了")), + } +} + +/// Keep the connection for as long as `alive()`. Returns false when the account server has no +/// WebSocket (the caller then uses the old way). +pub fn run(url: &str, token: &str, alive: &dyn Fn() -> bool, status: &dyn Fn(&str)) -> bool { + static SERIAL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + let mut backoff = 1u64; + while alive() { + match open(url, token) { + Err(NoLink { old: true, .. }) => return false, + Err(e) => { + status(&format!("連不上帳號伺服器：{}", e.why)); + // Wait, but notice at once when told to stop. + let until = Instant::now() + Duration::from_secs(backoff); + while Instant::now() < until && alive() { + std::thread::sleep(Duration::from_millis(200)); + } + backoff = (backoff * 2).min(30); + } + Ok(ws) => { + backoff = 1; + status("已連線"); + let serial = SERIAL.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + keep(ws, serial, alive); + if alive() { + status("連線中斷，正在重新連線…"); + } + } + } + } + true +} + +fn keep(mut ws: WebSocket, serial: u64, alive: &dyn Fn() -> bool) { + let _ = ws.get_ref().tcp().set_read_timeout(Some(Duration::from_millis(40))); + let (tx, rx) = channel::>(); + let calls: Calls = Arc::default(); + *link().lock().unwrap() = Some(Link { tx: tx.clone(), calls: calls.clone(), serial }); + let _ = tx.send(wire::pack(&json!({"t": "hello", "brain": crate::remote::hub().settings().brain_id}), &[])); + let (mut heard, mut pinged) = (Instant::now(), Instant::now()); + 'live: while alive() { + match ws.read() { + Ok(Message::Binary(b)) => { + heard = Instant::now(); + if let Some((head, body)) = wire::unpack(&b) { + on_frame(head, body, &tx, &calls); + } + } + Ok(Message::Close(_)) => break, + Ok(_) => heard = Instant::now(), + Err(e) if would_block(&e) => {} + Err(_) => break, + } + while let Ok(f) = rx.try_recv() { + if ws.write(Message::Binary(f.into())).is_err() { + break 'live; + } + } + if pinged.elapsed().as_secs() >= PING_SECS { + pinged = Instant::now(); + if ws.write(Message::Ping(Vec::new().into())).is_err() { + break; + } + } + match ws.flush() { + Err(e) if !would_block(&e) => break, + _ => {} + } + if heard.elapsed().as_secs() >= DEAD_SECS { + break; + } + } + let _ = ws.close(None); + let _ = ws.flush(); + let mut l = link().lock().unwrap(); + if l.as_ref().map(|l| l.serial) == Some(serial) { + *l = None; + } + drop(l); + // Whoever waits on an answer hears now, not after a timeout. + for (_, w) in calls.lock().unwrap().drain() { + let _ = w.send(Err(CUT.into())); + } +} + +fn on_frame(head: Value, body: Vec, tx: &Sender>, calls: &Calls) { + match head["t"].as_str().unwrap_or("") { + "job" => { + let tx = tx.clone(); + std::thread::spawn(move || { + let s = |k: &str| head[k].as_str().unwrap_or("").to_string(); + let id = s("id"); + let body = if head["z"].as_bool().unwrap_or(false) { unzip(&body).unwrap_or_default() } else { body }; + let req = Req { + method: s("method"), + path: s("path"), + query: head["query"].as_object().map(|o| o.iter().map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string())).collect()).unwrap_or_default(), + headers: [("x-anybot-user", s("user")), ("x-anybot-role", if head["role"].is_string() { s("role") } else { "owner".into() })].into_iter().map(|(k, v)| (k.to_string(), v)).collect(), + body, + peer: "relay".into(), + }; + let resp = crate::remote::handle(req, true); + let (out, z) = if head["z_ok"].as_bool().unwrap_or(false) { maybe_zip(resp.body) } else { (resp.body, false) }; + let _ = tx.send(wire::pack(&json!({"t": "reply", "id": id, "status": resp.status, "ctype": resp.ctype, "headers": resp.headers, "z": z}), &out)); + }); + } + "ret" => { + if let Some(w) = calls.lock().unwrap().remove(head["id"].as_str().unwrap_or("")) { + let body = if head["z"].as_bool().unwrap_or(false) { unzip(&body) } else { Ok(body) }; + let _ = w.send(body.map(|body| Answer { status: head["status"].as_u64().unwrap_or(502) as u16, body })); + } + } + _ => {} + } +} + +/// Tell the account server which brain this AnyBot shows (so the phone can go there). +pub fn hello(brain: &str) { + if let Some(l) = link().lock().unwrap().as_ref() { + let _ = l.tx.send(wire::pack(&json!({"t": "hello", "brain": brain}), &[])); + } +} + +/// Ask another AnyBot of this account over the lasting connection. None when it isn't open +/// (the caller then uses an ordinary request). +pub fn call(to: &str, method: &str, path: &str, query: &[(&str, String)], body: &[u8], secs: u64) -> Option> { + let (tx, calls) = { + let l = link().lock().unwrap(); + let l = l.as_ref()?; + (l.tx.clone(), l.calls.clone()) + }; + let id = crate::httpd::random_hex(8); + let (rtx, rrx) = channel(); + calls.lock().unwrap().insert(id.clone(), rtx); + let (out, z) = maybe_zip(body.to_vec()); + let q: serde_json::Map = query.iter().map(|(k, v)| (k.to_string(), json!(v))).collect(); + if tx.send(wire::pack(&json!({"t": "call", "id": id, "to": to, "method": method, "path": path, "query": q, "z": z}), &out)).is_err() { + calls.lock().unwrap().remove(&id); + return None; + } + Some(match rrx.recv_timeout(Duration::from_secs(secs)) { + Ok(r) => r, + Err(_) => { + calls.lock().unwrap().remove(&id); + Err("等太久沒有回應：網路很慢或斷斷續續".into()) + } + }) +} + +#[cfg(test)] +mod tests { + #[test] + fn compression_round_trips_and_shrinks_text() { + let text = "你好，這是一段會重複的聊天內容。".repeat(400); + let z = super::zip(text.as_bytes()); + assert!(z.len() * 5 < text.len()); + assert_eq!(super::unzip(&z).unwrap(), text.as_bytes()); + // Already-compressed bytes go as they are. + let mut x = 88172645463325252u64; + let noise: Vec = (0..5000) + .map(|_| { + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + (x >> 24) as u8 + }) + .collect(); + assert!(!super::maybe_zip(noise).1); + assert!(super::maybe_zip(text.into_bytes()).1); + } +} diff --git a/src/wire.rs b/src/wire.rs new file mode 100644 index 0000000..c83ab97 --- /dev/null +++ b/src/wire.rs @@ -0,0 +1,42 @@ +// (Shared with relay/.) +#![allow(dead_code)] +// One frame on the lasting connection between an AnyBot and the account server (a WebSocket): +// a small JSON header, then the body as raw bytes (no base64), in one binary message. +// +// Headers ("t" says which): +// server → AnyBot job {id, method, path, query, user, role, z, z_ok} a request for this AnyBot +// AnyBot → server reply {id, status, ctype, headers, z} its answer +// AnyBot → server call {id, to, method, path, query, z} a request for another AnyBot of the account +// server → AnyBot ret {id, status, ctype, z} that one's answer +// AnyBot → server hello {brain} what this AnyBot shows +// "z": the body is deflate-compressed. "z_ok": the asker can read a compressed answer. + +use serde_json::Value; + +pub fn pack(head: &Value, body: &[u8]) -> Vec { + let h = head.to_string(); + let mut out = Vec::with_capacity(4 + h.len() + body.len()); + out.extend_from_slice(&(h.len() as u32).to_be_bytes()); + out.extend_from_slice(h.as_bytes()); + out.extend_from_slice(body); + out +} + +pub fn unpack(b: &[u8]) -> Option<(Value, Vec)> { + let n = u32::from_be_bytes(b.get(..4)?.try_into().ok()?) as usize; + let head = serde_json::from_slice(b.get(4..4 + n)?).ok()?; + Some((head, b[4 + n..].to_vec())) +} + +#[cfg(test)] +mod tests { + #[test] + fn frames_round_trip() { + let head = serde_json::json!({"t": "job", "id": "a"}); + let f = super::pack(&head, b"\x00\x01body"); + let (h, b) = super::unpack(&f).unwrap(); + assert_eq!(h, head); + assert_eq!(b, b"\x00\x01body"); + assert!(super::unpack(&f[..3]).is_none()); + } +}