diff --git a/README.md b/README.md index 08a0f01..bbee06f 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@ | `OFFLINE_THRESHOLD_SECONDS` | 可选 | 离线判定秒数,未设置时使用 180,范围 30-3600 | | `HISTORY_RETENTION_DAYS` | 可选 | 历史保留天数,未设置时使用 30,范围 1-365 | | `CF_USAGE_ACCOUNT_ID` | 可选 | Cloudflare 用量查询的账户 ID,不设置则不启用用量查询 | -| `CF_USAGE_API_TOKEN` | 可选 | 用量查询 Token,勾选“加密”;需要 Account Analytics Read 权限 | +| `CF_USAGE_API_TOKEN` | 可选 | 用量查询 Token,勾选“加密”;需要 Account Analytics: Read,并授权对应账户 | ## API @@ -45,6 +45,7 @@ | `GET` | `/api/history/:id?hours=24` | 节点历史 | | `GET` | `/api/latency/:id?hours=24` | 节点延迟任务历史 | | `GET` | `/api/ws` | 实时 WebSocket | +| `GET` | `/api/agent/live` | Agent 实时指标 WebSocket | | `POST` | `/api/agent/report` | 探针上报,节点 Bearer token | | `POST` | `/api/admin/login` | 管理登录 | | `GET/POST` | `/api/admin/servers` | 管理节点 | @@ -64,7 +65,7 @@ | `GET` | `/api/admin/theme-settings` | 读取当前前端提供的主题设置描述 | | `POST` | `/api/turnstile/verify` | 校验全站 Turnstile 令牌 | | `GET` | `/api/admin/database` | 查询 D1 节点、历史和数据库统计 | -| `GET` | `/api/admin/cloudflare-usage` | 查询 UTC 今日/昨日的 D1 与 Workers 用量 | +| `GET` | `/api/admin/cloudflare-usage` | 查询 UTC 今日/昨日的 D1、Workers 与 Durable Objects 用量 | | `POST` | `/api/admin/exchange-rates/refresh` | 立即拉取并更新 D1 汇率快照 | | `DELETE` | `/api/admin/history` | 清理历史指标 | | `POST` | `/api/admin/notifications/test` | 发送一条通知测试消息 | diff --git a/agent/Cargo.lock b/agent/Cargo.lock index e8d65bc..dcf40aa 100644 --- a/agent/Cargo.lock +++ b/agent/Cargo.lock @@ -51,6 +51,17 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + [[package]] name = "clap" version = "4.6.6" @@ -143,6 +154,12 @@ 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 = "deranged" version = "0.5.8" @@ -212,6 +229,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 = "hashbrown" version = "0.17.1" @@ -240,6 +269,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.14" @@ -403,10 +438,12 @@ name = "nodeflare-agent" version = "0.0.1" dependencies = [ "clap", + "httpdate", "serde", "serde_json", "sha2", "sysinfo", + "tungstenite", "ureq", "url", ] @@ -490,6 +527,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.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +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" @@ -498,7 +558,7 @@ checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" dependencies = [ "cc", "cfg-if", - "getrandom", + "getrandom 0.2.17", "libc", "untrusted", "windows-sys", @@ -582,6 +642,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 = "sha2" version = "0.11.0" @@ -670,6 +741,26 @@ dependencies = [ "windows", ] +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "time" version = "0.3.55" @@ -710,6 +801,25 @@ dependencies = [ "zerovec", ] +[[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", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror", + "webpki-roots 0.26.11", +] + [[package]] name = "typenum" version = "1.20.1" @@ -744,7 +854,7 @@ dependencies = [ "serde_json", "ureq-proto", "utf8-zero", - "webpki-roots", + "webpki-roots 1.0.9", ] [[package]] @@ -795,6 +905,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.9", +] + [[package]] name = "webpki-roots" version = "1.0.9" diff --git a/agent/Cargo.toml b/agent/Cargo.toml index 55d33ac..7565970 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -11,8 +11,10 @@ clap = { version = "4.6", default-features = false, features = ["derive", "std", serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.11" +httpdate = "1" ureq = { version = "3.4", default-features = false, features = ["json", "rustls"] } url = "2.5" +tungstenite = { version = "0.30", default-features = false, features = ["handshake", "rustls-tls-webpki-roots"] } [target.'cfg(any(target_os = "windows", target_os = "macos"))'.dependencies] sysinfo = { version = "0.39.6", default-features = false, features = ["disk", "network", "system"] } diff --git a/agent/agent.sh b/agent/agent.sh index 910028c..a206a44 100644 --- a/agent/agent.sh +++ b/agent/agent.sh @@ -8,9 +8,28 @@ SERVICE_FILE="/etc/systemd/system/$SERVICE_NAME.service" OPENRC_FILE="/etc/init.d/$SERVICE_NAME" usage() { - echo "Usage: agent.sh -e ENDPOINT -t TOKEN [-i 60]" - echo " agent.sh --uninstall" - echo " agent.sh --status" + cat <<'EOF' +NodeFlare Agent 安装脚本 + +用法: + agent.sh -e -t [-i <上报间隔>] + agent.sh --status + agent.sh --uninstall + +参数: + -e NodeFlare Worker 地址(必填) + -t 后台生成的 Agent Token(必填,请勿泄露) + -i 初始上报间隔,15-3600 秒(默认 60) +EOF +} + +log() { + printf '[NodeFlare] %s\n' "$1" +} + +fail() { + printf '[NodeFlare] 错误:%s\n' "$1" >&2 + exit 1 } safe_value() { @@ -31,7 +50,7 @@ detect_init_system() { ensure_curl() { command -v curl >/dev/null 2>&1 && return - echo "curl not found; installing it..." + log "未找到 curl,正在通过系统包管理器安装" if command -v apt-get >/dev/null 2>&1; then apt-get update DEBIAN_FRONTEND=noninteractive apt-get install -y curl @@ -42,30 +61,29 @@ ensure_curl() { elif command -v apk >/dev/null 2>&1; then apk add --no-cache curl else - echo "curl is required and no supported package manager was found" >&2 - exit 1 + fail "缺少 curl,且未找到受支持的包管理器" fi - command -v curl >/dev/null 2>&1 || { echo "Failed to install curl" >&2; exit 1; } + command -v curl >/dev/null 2>&1 || fail "curl 安装失败" } explain_exec_failure() { target="$1" if [ ! -f "$target" ]; then - echo "Downloaded Agent disappeared before validation: $target" >&2 + echo "下载文件在校验前丢失:$target" >&2 elif [ ! -x "$target" ]; then - echo "Downloaded Agent is not executable: $target" >&2 + echo "下载文件不可执行:$target" >&2 elif [ "$(dd if="$target" bs=4 count=1 2>/dev/null)" != "$(printf '\177ELF')" ]; then - echo "Downloaded asset is not a Linux ELF executable" >&2 + echo "下载文件不是有效的 Linux ELF 可执行文件" >&2 else - echo "Downloaded Agent could not run on this system." >&2 - echo "Use a release built with the current static Linux targets, then retry installation." >&2 + echo "下载的 Agent 无法在当前系统运行,请确认系统架构与 Release 文件匹配" >&2 fi } install_agent() { - [ "$(id -u)" -eq 0 ] || { echo "Run install as root" >&2; exit 1; } + [ "$(id -u)" -eq 0 ] || fail "请使用 root 权限运行安装命令" + log "正在检查运行环境" ensure_curl - command -v sha256sum >/dev/null || { echo "sha256sum is required" >&2; exit 1; } + command -v sha256sum >/dev/null || fail "缺少 sha256sum,无法校验下载文件" token="" endpoint="" interval=60 @@ -74,46 +92,46 @@ install_agent() { option="$1" case "$option" in -t|-e|-i) - [ "$#" -ge 2 ] || { echo "Missing value for $option" >&2; usage; exit 1; } + [ "$#" -ge 2 ] || { echo "参数 $option 缺少值" >&2; usage; exit 1; } value="$2" shift 2 ;; - *) echo "Unknown argument: $option" >&2; usage; exit 1 ;; + *) echo "未知参数:$option" >&2; usage; exit 1 ;; esac case "$option" in - -t) [ -z "$token" ] || { echo "Duplicate argument: $option" >&2; exit 1; }; token="$value" ;; - -e) [ -z "$endpoint" ] || { echo "Duplicate argument: $option" >&2; exit 1; }; endpoint="$value" ;; - -i) [ "$interval_set" = false ] || { echo "Duplicate argument: $option" >&2; exit 1; }; interval="$value"; interval_set=true ;; + -t) [ -z "$token" ] || fail "参数 $option 重复"; token="$value" ;; + -e) [ -z "$endpoint" ] || fail "参数 $option 重复"; endpoint="$value" ;; + -i) [ "$interval_set" = false ] || fail "参数 $option 重复"; interval="$value"; interval_set=true ;; esac done [ -n "$token" ] && [ -n "$endpoint" ] || { usage; exit 1; } endpoint=${endpoint%/} - [ ${#token} -le 512 ] && [ ${#endpoint} -le 2048 ] || { echo "Install argument is too long" >&2; exit 1; } - safe_value "$token" && safe_value "$endpoint" || { echo "Invalid install argument" >&2; exit 1; } + [ ${#token} -le 512 ] && [ ${#endpoint} -le 2048 ] || fail "安装参数长度超出限制" + safe_value "$token" && safe_value "$endpoint" || fail "Worker 地址或 Agent Token 格式无效" case "$endpoint" in https://?*|http://localhost|http://localhost/*|http://localhost:*|http://127.0.0.1|http://127.0.0.1/*|http://127.0.0.1:*) ;; - *) echo "Worker URL must use HTTPS; HTTP is only allowed for loopback development" >&2; exit 1 ;; + *) fail "Worker 地址必须使用 HTTPS;仅本机调试可使用 HTTP" ;; esac - case "$endpoint" in *@*) echo "Endpoint must not contain user information" >&2; exit 1 ;; esac - case "$interval" in ''|*[!0-9]*) echo "Invalid interval" >&2; exit 1 ;; esac - [ "$interval" -ge 15 ] && [ "$interval" -le 3600 ] || { echo "Interval must be between 15 and 3600 seconds" >&2; exit 1; } - case "$(uname -m)" in x86_64|amd64) arch="x86_64" ;; aarch64|arm64) arch="aarch64" ;; *) echo "Unsupported architecture" >&2; exit 1 ;; esac + case "$endpoint" in *@*) fail "Worker 地址不能包含用户信息" ;; esac + case "$interval" in ''|*[!0-9]*) fail "上报间隔必须是整数" ;; esac + [ "$interval" -ge 15 ] && [ "$interval" -le 3600 ] || fail "上报间隔必须在 15-3600 秒之间" + case "$(uname -m)" in x86_64|amd64) arch="x86_64" ;; aarch64|arm64) arch="aarch64" ;; *) fail "暂不支持当前 CPU 架构:$(uname -m)" ;; esac init_system=$(detect_init_system) - [ "$init_system" != "unknown" ] || { echo "A running systemd or OpenRC installation is required" >&2; exit 1; } + [ "$init_system" != "unknown" ] || fail "未检测到正在运行的 systemd 或 OpenRC" mkdir -p "$INSTALL_DIR" temporary="$INSTALL_DIR/.agent.$$.download" trap 'rm -f "$temporary"' EXIT HUP INT TERM artifact="agent-linux-$arch" release_api="https://api.github.com/repos/imengying/NodeFlare/releases/latest" + log "正在获取 GitHub 最新正式版本($artifact)" release_json=$(curl --fail --location --silent --show-error --max-time 30 \ -H 'Accept: application/vnd.github+json' \ -H 'User-Agent: nodeflare-installer' \ "$release_api") release_tag=$(printf '%s\n' "$release_json" | tr ',' '\n' | sed -n 's/.*"tag_name"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' | head -n 1) printf '%s\n' "$release_tag" | grep -Eq '^v[0-9]+\.[0-9]+\.[0-9]+$' || { - echo "GitHub latest release has an invalid tag: ${release_tag:-missing}" >&2 - exit 1 + fail "GitHub 最新 Release 标签无效:${release_tag:-未找到}" } expected=$(printf '%s\n' "$release_json" | tr '{' '\n' | awk -v name="$artifact" ' { @@ -133,13 +151,15 @@ install_agent() { } } ') - [ -n "$expected" ] || { echo "GitHub release does not contain a SHA-256 digest for $artifact" >&2; exit 1; } + [ -n "$expected" ] || fail "Release 缺少 $artifact 的 SHA-256 摘要" release_base="https://github.com/imengying/NodeFlare/releases/download/$release_tag" + log "正在下载 NodeFlare Agent $release_tag" curl --fail --location --silent --show-error --max-time 120 \ "$release_base/$artifact" \ -o "$temporary" actual=$(sha256sum "$temporary" | awk '{ print $1 }') - [ -n "$expected" ] && [ "$actual" = "$expected" ] || { echo "Agent checksum verification failed" >&2; exit 1; } + [ -n "$expected" ] && [ "$actual" = "$expected" ] || fail "Agent SHA-256 校验失败,已停止安装" + log "下载校验通过,正在验证可执行文件" chmod 755 "$temporary" if ! installed_version=$("$temporary" --version); then explain_exec_failure "$temporary" @@ -147,12 +167,10 @@ install_agent() { fi installed_version=${installed_version##* } printf '%s\n' "$installed_version" | grep -Eq '^[0-9]+\.[0-9]+\.[0-9]+$' || { - echo "Downloaded Agent returned an invalid version: $installed_version" >&2 - exit 1 + fail "Agent 返回了无效版本号:$installed_version" } [ "$installed_version" = "${release_tag#v}" ] || { - echo "Release $release_tag contains Agent version $installed_version; refusing to install it" >&2 - exit 1 + fail "Release $release_tag 与 Agent 版本 $installed_version 不一致" } case "$init_system" in systemd) systemctl stop "$SERVICE_NAME" 2>/dev/null || true ;; @@ -160,6 +178,7 @@ install_agent() { esac mv "$temporary" "$AGENT_FILE" trap - EXIT HUP INT TERM + log "正在配置并启动 $init_system 服务" if [ "$init_system" = "systemd" ]; then printf '%s\n' \ '[Unit]' \ @@ -184,8 +203,7 @@ install_agent() { systemctl enable --now "$SERVICE_NAME" systemctl is-active --quiet "$SERVICE_NAME" || { systemctl status "$SERVICE_NAME" --no-pager >&2 || true - echo "NodeFlare Agent service failed to start" >&2 - exit 1 + fail "NodeFlare 服务启动失败,请查看上方状态信息" } elif [ "$init_system" = "openrc" ]; then printf '%s\n' \ @@ -202,11 +220,17 @@ install_agent() { rc-service "$SERVICE_NAME" restart rc-service "$SERVICE_NAME" status >/dev/null || { rc-service "$SERVICE_NAME" status >&2 || true - echo "NodeFlare Agent service failed to start" >&2 - exit 1 + fail "NodeFlare 服务启动失败,请查看上方状态信息" } fi - echo "NodeFlare Agent $installed_version installed with $init_system." + printf '\nNodeFlare Agent 安装完成\n' + printf ' 版本:%s\n' "$installed_version" + printf ' 服务:%s(%s)\n' "$SERVICE_NAME" "$init_system" + if [ "$init_system" = "systemd" ]; then + printf ' 查看状态:systemctl status %s\n' "$SERVICE_NAME" + else + printf ' 查看状态:rc-service %s status\n' "$SERVICE_NAME" + fi } status_agent() { @@ -218,12 +242,13 @@ status_agent() { rc-service "$SERVICE_NAME" status return $? fi - echo "NodeFlare agent service is not installed." >&2 + echo "未检测到 NodeFlare Agent 服务" >&2 return 1 } uninstall_agent() { - [ "$(id -u)" -eq 0 ] || { echo "Run uninstall as root" >&2; exit 1; } + [ "$(id -u)" -eq 0 ] || fail "请使用 root 权限执行卸载" + log "正在停止并移除 NodeFlare Agent" init_system=$(detect_init_system) case "$init_system" in systemd) systemctl disable --now "$SERVICE_NAME" 2>/dev/null || true ;; @@ -236,7 +261,7 @@ uninstall_agent() { rm -f "$OPENRC_FILE" [ "$init_system" != "systemd" ] || systemctl daemon-reload 2>/dev/null || true rm -rf "$INSTALL_DIR" - echo "NodeFlare agent removed." + echo "NodeFlare Agent 已卸载" } case "${1:-}" in diff --git a/agent/install-macos.sh b/agent/install-macos.sh index e5dea44..4dd84f9 100644 --- a/agent/install-macos.sh +++ b/agent/install-macos.sh @@ -6,35 +6,58 @@ INSTALL_DIR="/usr/local/libexec/nodeflare" AGENT_FILE="$INSTALL_DIR/agent" PLIST_FILE="/Library/LaunchDaemons/$LABEL.plist" +log() { + printf '[NodeFlare] %s\n' "$1" +} + +fail() { + printf '[NodeFlare] 错误:%s\n' "$1" >&2 + exit 1 +} + +usage() { + cat <<'EOF' +NodeFlare Agent macOS 安装脚本 + +用法: + install-macos.sh -e -t [-i <上报间隔>] + install-macos.sh --status + install-macos.sh --uninstall + +仅支持 Apple Silicon(arm64)。Agent Token 请勿泄露。 +EOF +} + safe_value() { case "$1" in *[!A-Za-z0-9_./:@-]*|'') return 1 ;; esac } if [ "${1:-}" = "--uninstall" ]; then - [ "$#" -eq 1 ] || { echo "--uninstall does not accept additional arguments" >&2; exit 1; } - [ "$(id -u)" -eq 0 ] || { echo "Run as root" >&2; exit 1; } + [ "$#" -eq 1 ] || fail "--uninstall 不接受其它参数" + [ "$(id -u)" -eq 0 ] || fail "请使用 root 权限执行卸载" + log "正在停止并移除 NodeFlare Agent" launchctl bootout system "$PLIST_FILE" 2>/dev/null || true rm -f "$PLIST_FILE" "$AGENT_FILE" rmdir "$INSTALL_DIR" 2>/dev/null || true - echo "NodeFlare Agent removed." + echo "NodeFlare Agent 已卸载" exit 0 fi if [ "${1:-}" = "--status" ]; then - [ "$#" -eq 1 ] || { echo "--status does not accept additional arguments" >&2; exit 1; } + [ "$#" -eq 1 ] || fail "--status 不接受其它参数" launchctl print "system/$LABEL" exit $? fi [ "${1:-}" != "-h" ] && [ "${1:-}" != "--help" ] && [ "$#" -gt 0 ] || { - echo "Usage: install-macos.sh -e ENDPOINT -t TOKEN [-i 60]" - echo " install-macos.sh --uninstall|--status" + usage exit 0 } -[ "$(id -u)" -eq 0 ] || { echo "Run as root" >&2; exit 1; } -[ "$(uname -m)" = "arm64" ] || { echo "Only Apple Silicon (arm64) is supported" >&2; exit 1; } -command -v curl >/dev/null || { echo "curl is required" >&2; exit 1; } -command -v shasum >/dev/null || { echo "shasum is required" >&2; exit 1; } +[ "$(id -u)" -eq 0 ] || fail "请使用 root 权限运行安装" +[ "$(uname -m)" = "arm64" ] || fail "仅支持 Apple Silicon(arm64)" +command -v curl >/dev/null || fail "缺少 curl" +command -v shasum >/dev/null || fail "缺少 shasum,无法校验下载文件" +log "正在检查运行环境" token="" endpoint="" interval=60 @@ -43,41 +66,42 @@ while [ "$#" -gt 0 ]; do option="$1" case "$option" in -t|-e|-i) - [ "$#" -ge 2 ] || { echo "Missing value for $option" >&2; exit 1; } + [ "$#" -ge 2 ] || fail "参数 $option 缺少值" value="$2" shift 2 ;; - *) echo "Unknown argument: $option" >&2; exit 1 ;; + *) fail "未知参数:$option" ;; esac case "$option" in - -t) [ -z "$token" ] || { echo "Duplicate argument: $option" >&2; exit 1; }; token="$value" ;; - -e) [ -z "$endpoint" ] || { echo "Duplicate argument: $option" >&2; exit 1; }; endpoint="$value" ;; - -i) [ "$interval_set" = false ] || { echo "Duplicate argument: $option" >&2; exit 1; }; interval="$value"; interval_set=true ;; + -t) [ -z "$token" ] || fail "参数 $option 重复"; token="$value" ;; + -e) [ -z "$endpoint" ] || fail "参数 $option 重复"; endpoint="$value" ;; + -i) [ "$interval_set" = false ] || fail "参数 $option 重复"; interval="$value"; interval_set=true ;; esac done -[ -n "$token" ] && [ -n "$endpoint" ] || { echo "Missing required argument" >&2; exit 1; } +[ -n "$token" ] && [ -n "$endpoint" ] || { usage; exit 1; } endpoint=${endpoint%/} -[ ${#token} -le 512 ] && [ ${#endpoint} -le 2048 ] || { echo "Install argument is too long" >&2; exit 1; } -safe_value "$token" && safe_value "$endpoint" || { echo "Invalid argument" >&2; exit 1; } +[ ${#token} -le 512 ] && [ ${#endpoint} -le 2048 ] || fail "安装参数长度超出限制" +safe_value "$token" && safe_value "$endpoint" || fail "Worker 地址或 Agent Token 格式无效" case "$endpoint" in https://?*|http://localhost|http://localhost/*|http://localhost:*|http://127.0.0.1|http://127.0.0.1/*|http://127.0.0.1:*) ;; - *) echo "Worker URL must use HTTPS; HTTP is only allowed for loopback development" >&2; exit 1 ;; + *) fail "Worker 地址必须使用 HTTPS;仅本机调试可使用 HTTP" ;; esac -case "$endpoint" in *@*) echo "Endpoint must not contain user information" >&2; exit 1 ;; esac -case "$interval" in ''|*[!0-9]*) echo "Invalid interval" >&2; exit 1 ;; esac -[ "$interval" -ge 15 ] && [ "$interval" -le 3600 ] || { echo "Interval must be between 15 and 3600 seconds" >&2; exit 1; } +case "$endpoint" in *@*) fail "Worker 地址不能包含用户信息" ;; esac +case "$interval" in ''|*[!0-9]*) fail "上报间隔必须是整数" ;; esac +[ "$interval" -ge 15 ] && [ "$interval" -le 3600 ] || fail "上报间隔必须在 15-3600 秒之间" mkdir -p "$INSTALL_DIR" temporary="$INSTALL_DIR/.agent.$$.download" trap 'rm -f "$temporary"' EXIT HUP INT TERM artifact="agent-macos-aarch64" release_api="https://api.github.com/repos/imengying/NodeFlare/releases/latest" +log "正在获取 GitHub 最新正式版本($artifact)" release_json=$(curl --fail --location --silent --show-error --max-time 30 \ -H 'Accept: application/vnd.github+json' \ -H 'User-Agent: nodeflare-installer' \ "$release_api") release_tag=$(printf '%s\n' "$release_json" | tr ',' '\n' | sed -n 's/.*"tag_name"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' | head -n 1) -printf '%s\n' "$release_tag" | grep -Eq '^v[0-9]+\.[0-9]+\.[0-9]+$' || { echo "GitHub latest release has an invalid tag" >&2; exit 1; } +printf '%s\n' "$release_tag" | grep -Eq '^v[0-9]+\.[0-9]+\.[0-9]+$' || fail "GitHub 最新 Release 标签无效:${release_tag:-未找到}" expected=$(printf '%s\n' "$release_json" | tr '{' '\n' | awk -v name="$artifact" ' { compact = $0 @@ -96,17 +120,20 @@ expected=$(printf '%s\n' "$release_json" | tr '{' '\n' | awk -v name="$artifact" } } ') -[ -n "$expected" ] || { echo "GitHub release does not contain a SHA-256 digest for $artifact" >&2; exit 1; } +[ -n "$expected" ] || fail "Release 缺少 $artifact 的 SHA-256 摘要" release_base="https://github.com/imengying/NodeFlare/releases/download/$release_tag" +log "正在下载 NodeFlare Agent $release_tag" curl --fail --location --silent --show-error --max-time 120 \ "$release_base/$artifact" \ -o "$temporary" actual=$(shasum -a 256 "$temporary" | awk '{ print $1 }') -[ -n "$expected" ] && [ "$actual" = "$expected" ] || { echo "Agent checksum verification failed" >&2; exit 1; } +[ -n "$expected" ] && [ "$actual" = "$expected" ] || fail "Agent SHA-256 校验失败,已停止安装" chmod 755 "$temporary" -installed_version=$("$temporary" --version) || { echo "Downloaded Agent cannot run on this macOS system" >&2; exit 1; } +log "下载校验通过,正在验证可执行文件" +installed_version=$("$temporary" --version) || fail "下载的 Agent 无法在当前 macOS 运行" installed_version=${installed_version##* } -[ "$installed_version" = "${release_tag#v}" ] || { echo "Release $release_tag contains Agent version $installed_version" >&2; exit 1; } +[ "$installed_version" = "${release_tag#v}" ] || fail "Release $release_tag 与 Agent 版本 $installed_version 不一致" +log "正在配置并启动 macOS LaunchDaemon 服务" launchctl bootout system "$PLIST_FILE" 2>/dev/null || true mv "$temporary" "$AGENT_FILE" trap - EXIT HUP INT TERM @@ -131,5 +158,8 @@ for _ in 1 2 3 4 5 6 7 8 9 10; do break fi done -[ "$started" = true ] || { echo "NodeFlare Agent failed to start; inspect /var/log/nodeflare-agent.log" >&2; exit 1; } -echo "NodeFlare Agent $installed_version installed and started." +[ "$started" = true ] || fail "NodeFlare 服务启动失败,请查看 /var/log/nodeflare-agent.log" +printf '\nNodeFlare Agent 安装完成\n' +printf ' 版本:%s\n' "$installed_version" +printf ' 服务:%s(LaunchDaemon)\n' "$LABEL" +printf ' 查看状态:sudo launchctl print system/%s\n' "$LABEL" diff --git a/agent/install.ps1 b/agent/install.ps1 index defd4c8..af2ca74 100644 --- a/agent/install.ps1 +++ b/agent/install.ps1 @@ -12,9 +12,17 @@ $TaskName = "NodeFlare Agent" $InstallDir = Join-Path $env:ProgramData "NodeFlare" $AgentFile = Join-Path $InstallDir "nodeflare-agent.exe" +function Write-Step([string]$Message) { + Write-Host "[NodeFlare] $Message" +} + +function Write-InstallError([string]$Message) { + throw "[NodeFlare] 错误:$Message" +} + function Assert-Safe([string]$Name, [string]$Value) { if ([string]::IsNullOrWhiteSpace($Value) -or $Value -notmatch '^[A-Za-z0-9_./:@-]+$') { - throw "Invalid $Name" + Write-InstallError "$Name 格式无效" } } @@ -35,22 +43,23 @@ function Assert-Endpoint([string]$Value) { -not [string]::IsNullOrEmpty($Parsed.Query) -or -not [string]::IsNullOrEmpty($Parsed.Fragment) ) { - throw "Endpoint must use HTTPS; HTTP is only allowed for loopback development" + Write-InstallError "Worker 地址必须使用 HTTPS;仅本机调试可使用 HTTP" } } if ($Uninstall) { + Write-Step "正在停止并移除 NodeFlare Agent" Stop-ScheduledTask -TaskName $TaskName -ErrorAction SilentlyContinue Unregister-ScheduledTask -TaskName $TaskName -Confirm:$false -ErrorAction SilentlyContinue Remove-Item -LiteralPath $InstallDir -Recurse -Force -ErrorAction SilentlyContinue - Write-Host "NodeFlare Agent removed." + Write-Host "NodeFlare Agent 已卸载" exit 0 } if ($Status) { $Task = Get-ScheduledTask -TaskName $TaskName -ErrorAction SilentlyContinue if ($null -eq $Task) { - Write-Error "NodeFlare Agent is not installed." + Write-Error "未检测到 NodeFlare Agent 服务" exit 1 } $Task @@ -58,17 +67,18 @@ if ($Status) { } if (-not ([Security.Principal.WindowsPrincipal][Security.Principal.WindowsIdentity]::GetCurrent()).IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)) { - throw "Run PowerShell as Administrator" + Write-InstallError "请使用管理员身份运行 PowerShell" } $NativeArchitecture = if ($env:PROCESSOR_ARCHITEW6432) { $env:PROCESSOR_ARCHITEW6432 } else { $env:PROCESSOR_ARCHITECTURE } if ($NativeArchitecture -ne "AMD64") { - throw "Only Windows x86_64 is supported" + Write-InstallError "仅支持 Windows x86_64" } +Write-Step "正在检查运行环境" Assert-Safe "Token" $Token Assert-Endpoint $Endpoint $TokenLength = $Token.Length if ($TokenLength -gt 512) { - throw "Install argument is too long" + Write-InstallError "安装参数长度超出限制" } $Endpoint = $Endpoint.TrimEnd('/') @@ -78,30 +88,33 @@ $Temporary = "$AgentFile.$PID.download.exe" $ReleaseApi = "https://api.github.com/repos/imengying/NodeFlare/releases/latest" $Artifact = "agent-windows-x86_64.exe" try { + Write-Step "正在获取 GitHub 最新正式版本($Artifact)" $Release = Invoke-RestMethod -Uri $ReleaseApi -Headers @{ Accept = "application/vnd.github+json"; "User-Agent" = "nodeflare-installer" } -TimeoutSec 30 $ReleaseAsset = $Release.assets | Where-Object { $_.name -eq $Artifact } | Select-Object -First 1 if ($Release.tag_name -notmatch '^v[0-9]+\.[0-9]+\.[0-9]+$' -or $null -eq $ReleaseAsset) { - throw "GitHub latest release is invalid or does not contain $Artifact" + Write-InstallError "GitHub 最新 Release 无效,或缺少 $Artifact" } $DigestMatch = [regex]::Match([string]$ReleaseAsset.digest, '^sha256:([0-9a-fA-F]{64})$') if (-not $DigestMatch.Success) { - throw "GitHub release does not contain a SHA-256 digest for $Artifact" + Write-InstallError "Release 缺少 $Artifact 的 SHA-256 摘要" } $ExpectedChecksum = $DigestMatch.Groups[1].Value $DownloadUrl = "https://github.com/imengying/NodeFlare/releases/download/$($Release.tag_name)/$Artifact" + Write-Step "正在下载 NodeFlare Agent $($Release.tag_name)" Invoke-WebRequest -Uri $DownloadUrl -OutFile $Temporary -TimeoutSec 120 $ActualChecksum = (Get-FileHash -LiteralPath $Temporary -Algorithm SHA256).Hash if ($ActualChecksum -ne $ExpectedChecksum) { - throw "Agent checksum verification failed" + Write-InstallError "Agent SHA-256 校验失败,已停止安装" } + Write-Step "下载校验通过,正在验证可执行文件" $InstalledVersion = (& $Temporary --version | Out-String).Trim() if ($LASTEXITCODE -ne 0) { - throw "Downloaded Agent failed its version check" + Write-InstallError "下载的 Agent 无法在当前 Windows 运行" } $InstalledVersion = ($InstalledVersion -split '\s+')[-1] $ExpectedVersion = $Release.tag_name.Substring(1) if ($InstalledVersion -ne $ExpectedVersion) { - throw "Release $($Release.tag_name) contains Agent version $InstalledVersion" + Write-InstallError "Release $($Release.tag_name) 与 Agent 版本 $InstalledVersion 不一致" } Stop-ScheduledTask -TaskName $TaskName -ErrorAction SilentlyContinue Move-Item -LiteralPath $Temporary -Destination $AgentFile -Force @@ -110,6 +123,7 @@ try { } $TaskArguments = "-e $Endpoint -t $Token -i $Interval" +Write-Step "正在注册并启动 Windows 计划任务" $TaskAction = New-ScheduledTaskAction -Execute $AgentFile -Argument $TaskArguments $Trigger = New-ScheduledTaskTrigger -AtStartup $Settings = New-ScheduledTaskSettingsSet -RestartCount 999 -RestartInterval (New-TimeSpan -Minutes 1) -ExecutionTimeLimit (New-TimeSpan -Days 3650) @@ -123,6 +137,10 @@ for ($Attempt = 0; $Attempt -lt 10; $Attempt++) { if ($Task.State -eq "Running") { break } } if ($Task.State -ne "Running") { - throw "NodeFlare Agent scheduled task failed to start (state: $($Task.State))" + Write-InstallError "NodeFlare 服务启动失败(状态:$($Task.State))" } -Write-Host "NodeFlare Agent $InstalledVersion installed and started." +Write-Host "" +Write-Host "NodeFlare Agent 安装完成" +Write-Host " 版本:$InstalledVersion" +Write-Host " 服务:$TaskName(Windows 计划任务)" +Write-Host " 查看状态:Get-ScheduledTask -TaskName '$TaskName'" diff --git a/agent/src/main.rs b/agent/src/main.rs index f167c93..462f2c3 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -1,7 +1,7 @@ use std::collections::{HashMap, HashSet}; use std::env; use std::fs; -use std::io::{self, Read, Write}; +use std::io::{self, ErrorKind, Read, Write}; use std::net::{IpAddr, SocketAddr, TcpStream, ToSocketAddrs}; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; @@ -11,7 +11,7 @@ use std::path::Path; use std::process::Command; #[cfg(target_os = "windows")] use std::process::Stdio; -use std::sync::{mpsc, Arc, Mutex}; +use std::sync::{mpsc, Arc, Condvar, Mutex}; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -20,6 +20,8 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; #[cfg(any(target_os = "windows", target_os = "macos"))] use sysinfo::{Disks, Networks, System}; +use tungstenite::client::IntoClientRequest; +use tungstenite::{connect, Error as WebSocketError, Message}; #[cfg(not(any(target_os = "linux", target_os = "windows", target_os = "macos")))] compile_error!("nodeflare-agent supports Linux, Windows, and macOS"); @@ -40,6 +42,16 @@ const REPORT_RETRY_MIN: Duration = Duration::from_secs(5); const REPORT_RETRY_MAX: Duration = Duration::from_secs(300); const UPDATE_CHECK_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60); const UPDATE_RETRY_INTERVAL: Duration = Duration::from_secs(30 * 60); +const LIVE_RECONNECT_DELAY: Duration = Duration::from_secs(3); +const LIVE_QUEUE_CAPACITY: usize = 720; +// Keep each WebSocket frame comfortably below the Worker 1 MiB validation limit; +// the queue preserves the remaining samples for the next frame. +const LIVE_BATCH_CAPACITY: usize = 32; +// D1 persistence can take a few hundred milliseconds; allow the ACK to arrive +// before falling back to the configured interval. +const LIVE_ACK_READ_TIMEOUT: Duration = Duration::from_millis(1_000); +const CLOCK_CALIBRATION_MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60); +const CLOCK_CALIBRATION_MIN_CHANGE_MS: i64 = 20_000; type Error = Box; type Result = std::result::Result; @@ -187,6 +199,75 @@ struct GithubReleaseAsset { struct SubmitResult { config: Option, config_hash: String, + clock_offset_ms: Option, +} + +#[derive(Debug, Default)] +struct ClockCalibration { + offset_ms: i64, + calibrated_at: Option, +} + +impl ClockCalibration { + fn observe(&mut self, offset_ms: i64) -> bool { + let should_update = self.calibrated_at.is_none() + || self + .calibrated_at + .is_some_and(|at| at.elapsed() >= CLOCK_CALIBRATION_MAX_AGE) + || self.offset_ms.abs_diff(offset_ms) >= CLOCK_CALIBRATION_MIN_CHANGE_MS as u64; + if should_update { + self.offset_ms = offset_ms; + self.calibrated_at = Some(Instant::now()); + } + should_update + } +} + +type SharedClock = Arc>; + +struct LiveSender { + pending: Arc<(Mutex>, Condvar)>, + report_interval: Arc>, +} + +impl LiveSender { + fn start(config: &RuntimeConfig, clock: SharedClock) -> Result { + let pending = Arc::new((Mutex::new(Vec::new()), Condvar::new())); + let report_interval = Arc::new(Mutex::new(Duration::from_secs( + config.report_interval.clamp(15, 3600), + ))); + let endpoint = live_endpoint(&config.endpoint)?; + let token = config.token.clone(); + let sender_state = Arc::clone(&pending); + let sender_interval = Arc::clone(&report_interval); + thread::Builder::new() + .name("nodeflare-live".to_string()) + .spawn(move || { + live_sender_loop(&endpoint, &token, sender_state, sender_interval, clock) + })?; + Ok(Self { + pending, + report_interval, + }) + } + + fn send(&self, report: &Report) { + let (pending, ready) = &*self.pending; + if let Ok(mut pending) = pending.lock() { + if pending.len() >= LIVE_QUEUE_CAPACITY { + pending.remove(0); + } + pending.push(report.clone()); + ready.notify_one(); + } + } + + fn set_report_interval(&self, seconds: u64) { + if let Ok(mut interval) = self.report_interval.lock() { + *interval = Duration::from_secs(seconds.clamp(15, 3600)); + } + self.pending.1.notify_one(); + } } #[derive(Debug, Serialize)] @@ -216,7 +297,7 @@ struct GpuMetric { memory_total: i64, } -#[derive(Debug, Default, Serialize)] +#[derive(Debug, Default, Clone, Serialize)] struct Report { timestamp: i64, cpu: f64, @@ -675,6 +756,52 @@ fn unix_timestamp() -> i64 { .as_secs() as i64 } +fn unix_timestamp_millis() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(i64::MAX as u128) as i64 +} + +fn clock_offset_from_http_date(value: &str, started_ms: i64, ended_ms: i64) -> Option { + let server_ms: i64 = httpdate::parse_http_date(value) + .ok()? + .duration_since(UNIX_EPOCH) + .ok()? + .as_millis() + .try_into() + .ok()?; + let midpoint_ms = started_ms.saturating_add(ended_ms.saturating_sub(started_ms) / 2); + Some(server_ms.saturating_sub(midpoint_ms)) +} + +fn clock_offset_from_server_seconds(value: &str, started_ms: i64, ended_ms: i64) -> Option { + let server_ms = value.trim().parse::().ok()?.saturating_mul(1_000); + let midpoint_ms = started_ms.saturating_add(ended_ms.saturating_sub(started_ms) / 2); + Some(server_ms.saturating_sub(midpoint_ms)) +} + +fn corrected_timestamp(timestamp: i64, offset_ms: i64) -> i64 { + timestamp + .saturating_mul(1_000) + .saturating_add(offset_ms) + .div_euclid(1_000) +} + +fn shared_clock_offset(clock: &SharedClock) -> i64 { + clock.lock().map_or(0, |calibration| calibration.offset_ms) +} + +fn observe_clock(clock: &SharedClock, offset_ms: Option) { + let Some(offset_ms) = offset_ms else { + return; + }; + if let Ok(mut calibration) = clock.lock() { + calibration.observe(offset_ms); + } +} + fn execute_latency_task(task: LatencyTask) -> LatencyResult { let (latency_ms, packet_loss) = match task.task_type.as_str() { "tcp" => tcp_latency_probe(&task.target, task.port), @@ -894,7 +1021,7 @@ fn u64_to_i64(value: u64) -> i64 { } #[cfg(target_os = "linux")] -fn collect(config: &RuntimeConfig, latency_results: Vec) -> Report { +fn collect(config: &RuntimeConfig, latency_results: Vec, timestamp: i64) -> Report { let cpu_before = cpu_sample(); let io_before = io_sample(&config.network_interface); thread::sleep(Duration::from_secs(1)); @@ -955,7 +1082,7 @@ fn collect(config: &RuntimeConfig, latency_results: Vec) -> Repor .saturating_add(io_after.write_millis.saturating_sub(io_before.write_millis)); Report { - timestamp: unix_timestamp(), + timestamp, cpu, load1: loads.first().copied().unwrap_or(0.0), load5: loads.get(1).copied().unwrap_or(0.0), @@ -1009,7 +1136,7 @@ fn collect(config: &RuntimeConfig, latency_results: Vec) -> Repor } #[cfg(any(target_os = "windows", target_os = "macos"))] -fn collect(config: &RuntimeConfig, latency_results: Vec) -> Report { +fn collect(config: &RuntimeConfig, latency_results: Vec, timestamp: i64) -> Report { let mut system = System::new_all(); let mut networks = Networks::new_with_refreshed_list(); thread::sleep(Duration::from_secs(1)); @@ -1066,7 +1193,7 @@ fn collect(config: &RuntimeConfig, latency_results: Vec) -> Repor let (tcp_connections, udp_connections) = connection_counts(); Report { - timestamp: unix_timestamp(), + timestamp, cpu: system.global_cpu_usage() as f64, load1: load.one, load5: load.five, @@ -1143,12 +1270,26 @@ fn submit( config_hash: &str, ) -> Result { let batch = ReportBatch { samples: reports }; + let started_ms = unix_timestamp_millis(); let mut response = agent .post(&format!("{}/api/agent/report", config.endpoint)) .header("Authorization", format!("Bearer {}", config.token)) .header("User-Agent", format!("nodeflare-agent/{VERSION}")) .header("X-Agent-Config-Sha256", config_hash) .send_json(batch)?; + let ended_ms = unix_timestamp_millis(); + let clock_offset_ms = response + .headers() + .get("x-nodeflare-server-time") + .and_then(|value| value.to_str().ok()) + .and_then(|value| clock_offset_from_server_seconds(value, started_ms, ended_ms)) + .or_else(|| { + response + .headers() + .get("date") + .and_then(|value| value.to_str().ok()) + .and_then(|value| clock_offset_from_http_date(value, started_ms, ended_ms)) + }); let next_hash = response .headers() .get("X-Agent-Config-Sha256") @@ -1163,9 +1304,253 @@ fn submit( Ok(SubmitResult { config: remote, config_hash: next_hash, + clock_offset_ms, }) } +fn fetch_clock_offset(agent: &ureq::Agent, endpoint: &str) -> Result> { + let started_ms = unix_timestamp_millis(); + let response = agent + .get(&format!("{endpoint}/api/config")) + .header("User-Agent", format!("nodeflare-agent/{VERSION}")) + .call()?; + let ended_ms = unix_timestamp_millis(); + Ok(response + .headers() + .get("x-nodeflare-server-time") + .and_then(|value| value.to_str().ok()) + .and_then(|value| clock_offset_from_server_seconds(value, started_ms, ended_ms)) + .or_else(|| { + response + .headers() + .get("date") + .and_then(|value| value.to_str().ok()) + .and_then(|value| clock_offset_from_http_date(value, started_ms, ended_ms)) + })) +} + +fn live_endpoint(endpoint: &str) -> Result { + let mut url = url::Url::parse(endpoint)?; + url.set_scheme(if url.scheme() == "https" { "wss" } else { "ws" }) + .map_err(|_| "unsupported live endpoint scheme")?; + let path = format!("{}/api/agent/live", url.path().trim_end_matches('/')); + url.set_path(&path); + Ok(url.to_string()) +} + +type LiveSocket = tungstenite::WebSocket>; + +fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option)> { + let mut request = endpoint.into_client_request()?; + request + .headers_mut() + .insert("Authorization", format!("Bearer {token}").parse()?); + request + .headers_mut() + .insert("User-Agent", format!("nodeflare-agent/{VERSION}").parse()?); + let started_ms = unix_timestamp_millis(); + let (socket, response) = connect(request)?; + let ended_ms = unix_timestamp_millis(); + let clock_offset_ms = response + .headers() + .get("date") + .and_then(|value| value.to_str().ok()) + .and_then(|value| clock_offset_from_http_date(value, started_ms, ended_ms)); + Ok((socket, clock_offset_ms)) +} + +fn set_live_read_timeout(socket: &mut LiveSocket, timeout: Option) -> io::Result<()> { + match socket.get_mut() { + tungstenite::stream::MaybeTlsStream::Plain(stream) => stream.set_read_timeout(timeout), + tungstenite::stream::MaybeTlsStream::Rustls(stream) => { + stream.sock.set_read_timeout(timeout) + } + _ => Ok(()), + } +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct LiveAck { + #[serde(rename = "type")] + message_type: String, + ts: i64, + persisted: bool, + #[serde(rename = "nextD1WriteAfterMs")] + next_d1_write_after_ms: u64, + #[serde(rename = "nextWssReportAfterMs")] + next_wss_report_after_ms: Option, +} + +fn ack_interval(ack: &LiveAck, fallback: Duration) -> Duration { + ack.next_wss_report_after_ms + .map(Duration::from_millis) + .map(|interval| interval.clamp(Duration::from_secs(2), Duration::from_secs(3600))) + .unwrap_or(fallback) +} + +fn read_live_ack(socket: &mut LiveSocket, current_interval: Duration) -> Result> { + match socket.read() { + Ok(Message::Text(text)) => { + let Ok(ack) = serde_json::from_str::(text.as_ref()) else { + return Ok(None); + }; + if ack.message_type != "ack" || ack.ts <= 0 { + return Ok(None); + } + let _ = (ack.persisted, ack.next_d1_write_after_ms); + Ok(Some(ack_interval(&ack, current_interval))) + } + Ok(Message::Ping(payload)) => { + socket.send(Message::Pong(payload))?; + Ok(None) + } + Ok(Message::Close(_)) => Ok(Some(Duration::ZERO)), + Ok(_) => Ok(None), + Err(WebSocketError::Io(error)) + if matches!(error.kind(), ErrorKind::TimedOut | ErrorKind::WouldBlock) => + { + Ok(None) + } + Err(error) => Err(error.into()), + } +} + +fn live_update_payload(reports: &[Report]) -> Result { + Ok(serde_json::to_string(&serde_json::json!({ + "type": "update", + "samples": reports, + }))?) +} + +fn requeue_reports(pending: &Arc<(Mutex>, Condvar)>, reports: Vec) { + if reports.is_empty() { + return; + } + let (queue, ready) = &**pending; + if let Ok(mut queue) = queue.lock() { + let mut merged = reports; + merged.append(&mut queue); + if merged.len() > LIVE_QUEUE_CAPACITY { + merged.truncate(LIVE_QUEUE_CAPACITY); + } + *queue = merged; + ready.notify_one(); + } +} + +fn live_sender_loop( + endpoint: &str, + token: &str, + pending: Arc<(Mutex>, Condvar)>, + configured_interval: Arc>, + clock: SharedClock, +) { + let mut socket: Option = None; + let mut send_interval = configured_interval + .lock() + .map(|interval| *interval) + .unwrap_or(Duration::from_secs(60)); + let mut next_send_at = Instant::now(); + loop { + let (queue, ready) = &*pending; + let mut queue = match queue.lock() { + Ok(queue) => queue, + Err(_) => return, + }; + while queue.is_empty() || Instant::now() < next_send_at { + let wait = if queue.is_empty() { + Duration::from_millis(250) + } else { + next_send_at + .saturating_duration_since(Instant::now()) + .min(Duration::from_millis(250)) + }; + queue = match ready.wait_timeout(queue, wait) { + Ok((queue, _)) => queue, + Err(_) => return, + }; + } + let count = queue.len().min(LIVE_BATCH_CAPACITY); + let batch = queue.drain(..count).collect::>(); + drop(queue); + + if let Ok(interval) = configured_interval.lock() { + send_interval = (*interval).min(Duration::from_secs(3600)); + } + if socket.is_none() { + match connect_live(endpoint, token) { + Ok((mut connected, offset_ms)) => { + observe_clock(&clock, offset_ms); + if set_live_read_timeout(&mut connected, Some(LIVE_ACK_READ_TIMEOUT)).is_err() { + socket = None; + thread::sleep(LIVE_RECONNECT_DELAY); + requeue_reports(&pending, batch); + continue; + } + socket = Some(connected); + } + Err(error) => { + eprintln!("live connection failed: {error}"); + requeue_reports(&pending, batch); + thread::sleep(LIVE_RECONNECT_DELAY); + continue; + } + } + } + + let payload = match live_update_payload(&batch) { + Ok(payload) => payload, + Err(error) => { + eprintln!("live payload encode failed: {error}"); + requeue_reports(&pending, batch); + continue; + } + }; + let sent = socket + .as_mut() + .is_some_and(|socket| socket.send(Message::Text(payload.clone().into())).is_ok()); + if !sent { + socket = None; + thread::sleep(LIVE_RECONNECT_DELAY); + let mut resent = false; + if let Ok((mut connected, offset_ms)) = connect_live(endpoint, token) { + observe_clock(&clock, offset_ms); + if set_live_read_timeout(&mut connected, Some(LIVE_ACK_READ_TIMEOUT)).is_ok() + && connected.send(Message::Text(payload.into())).is_ok() + { + socket = Some(connected); + resent = true; + } + } + if !resent { + requeue_reports(&pending, batch); + } + continue; + } + + next_send_at = Instant::now() + send_interval; + let mut drop_socket = false; + if let Some(socket) = socket.as_mut() { + match read_live_ack(socket, send_interval) { + Ok(Some(Duration::ZERO)) => drop_socket = true, + Ok(Some(interval)) => { + send_interval = interval; + next_send_at = Instant::now() + send_interval; + } + Ok(None) => {} + Err(error) => { + eprintln!("live ACK read failed: {error}"); + drop_socket = true; + } + } + } + if drop_socket { + socket = None; + } + } +} + fn apply_remote(config: &mut RuntimeConfig, remote: &RemoteConfig) -> bool { if (15..=3600).contains(&remote.report_interval) { config.report_interval = remote.report_interval; @@ -1480,6 +1865,16 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { let mut report_failures = 0_u32; let mut next_update_check = Instant::now(); let mut config_hash = String::new(); + let clock = Arc::new(Mutex::new(ClockCalibration::default())); + if !print_only { + match fetch_clock_offset(&agent, &config.endpoint) { + Ok(offset_ms) => observe_clock(&clock, offset_ms), + Err(error) => eprintln!("server time calibration failed: {error}"), + } + } + let live = (!once && !print_only) + .then(|| LiveSender::start(&config, Arc::clone(&clock))) + .transpose()?; loop { pending_results.extend(latency_executor.drain()); let current = Instant::now(); @@ -1505,17 +1900,26 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { } if Instant::now() >= next_collect { + let offset_ms = shared_clock_offset(&clock); let mut latest_results = HashMap::new(); - for result in std::mem::take(&mut pending_results) { + for mut result in std::mem::take(&mut pending_results) { + result.timestamp = corrected_timestamp(result.timestamp, offset_ms); latest_results.insert(result.task_id.clone(), result); } - let report = collect(&config, latest_results.into_values().collect()); + let mut report = collect(&config, latest_results.into_values().collect(), 0); + report.timestamp = corrected_timestamp(unix_timestamp(), offset_ms); if print_only { println!("{}", serde_json::to_string_pretty(&report)?); return Ok(()); } + if let Some(live) = &live { + live.send(&report); + } pending_samples.push(report); - prune_report_samples(&mut pending_samples, unix_timestamp()); + prune_report_samples( + &mut pending_samples, + corrected_timestamp(unix_timestamp(), offset_ms), + ); if pending_samples.len() > 720 { let overflow = pending_samples.len() - 720; pending_samples.drain(..overflow); @@ -1526,6 +1930,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { if Instant::now() >= next_report && !pending_samples.is_empty() { match submit(&agent, &config, &pending_samples, &config_hash) { Ok(result) => { + observe_clock(&clock, result.clock_offset_ms); pending_samples.clear(); report_failures = 0; config_hash = result.config_hash; @@ -1533,6 +1938,9 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { if apply_remote(&mut config, &remote) { next_latency.clear(); } + if let Some(live) = &live { + live.set_report_interval(config.report_interval); + } } if !once && config.auto_update && Instant::now() >= next_update_check { match update(&agent) { @@ -1592,15 +2000,17 @@ mod tests { use clap::Parser; use std::net::TcpListener; use std::thread; - use std::time::Duration; + use std::time::{Duration, Instant}; use super::{ - agent_artifact_name, executable_format_valid, gpu_name_from_uevent, is_public_probe_ip, - median, normalized_version, parse_lspci_gpu_names, parse_probe_target, + ack_interval, agent_artifact_name, clock_offset_from_http_date, corrected_timestamp, + executable_format_valid, gpu_name_from_uevent, is_public_probe_ip, live_endpoint, + live_update_payload, median, normalized_version, parse_lspci_gpu_names, parse_probe_target, parse_system_profiler_gpu_names, ping_latency, prune_report_samples, release_asset_sha256, report_retry_delay, sanitize_latency_tasks, selected_interface, tcp_latency_probe_address, - valid_endpoint, version_triplet, CliOptions, GithubReleaseAsset, LatencyResult, - LatencyTask, Report, MAX_LATENCY_TASKS, MAX_PENDING_LATENCY_RESULTS, PROBE_ATTEMPTS, + valid_endpoint, version_triplet, CliOptions, ClockCalibration, GithubReleaseAsset, + LatencyResult, LatencyTask, LiveAck, Report, CLOCK_CALIBRATION_MAX_AGE, MAX_LATENCY_TASKS, + MAX_PENDING_LATENCY_RESULTS, PROBE_ATTEMPTS, }; #[cfg(any(target_os = "windows", target_os = "macos"))] @@ -1693,6 +2103,89 @@ mod tests { assert!(!valid_endpoint("https://bad host.example")); } + #[test] + fn builds_live_websocket_endpoints() { + assert_eq!( + live_endpoint("https://monitor.example.com").unwrap(), + "wss://monitor.example.com/api/agent/live" + ); + assert_eq!( + live_endpoint("https://monitor.example.com/base/").unwrap(), + "wss://monitor.example.com/base/api/agent/live" + ); + assert_eq!( + live_endpoint("http://127.0.0.1:8787").unwrap(), + "ws://127.0.0.1:8787/api/agent/live" + ); + } + + #[test] + fn accepts_server_realtime_ack_interval() { + let ack: LiveAck = serde_json::from_str( + r#"{"type":"ack","ts":100,"persisted":false,"nextD1WriteAfterMs":60000,"nextWssReportAfterMs":5000}"#, + ) + .unwrap(); + assert_eq!( + ack_interval(&ack, Duration::from_secs(60)), + Duration::from_secs(5) + ); + let slow: LiveAck = serde_json::from_str( + r#"{"type":"ack","ts":100,"persisted":true,"nextD1WriteAfterMs":60000,"nextWssReportAfterMs":1}"#, + ) + .unwrap(); + assert_eq!( + ack_interval(&slow, Duration::from_secs(60)), + Duration::from_secs(2) + ); + } + + #[test] + fn encodes_realtime_samples_as_a_batch() { + let payload = live_update_payload(&[ + Report { + timestamp: 10, + cpu: 20.0, + ..Report::default() + }, + Report { + timestamp: 15, + cpu: 30.0, + ..Report::default() + }, + ]) + .unwrap(); + let value: serde_json::Value = serde_json::from_str(&payload).unwrap(); + assert_eq!(value["type"], "update"); + assert_eq!(value["samples"].as_array().unwrap().len(), 2); + assert_eq!(value["samples"][1]["cpu"], 30.0); + } + + #[test] + fn calibrates_timestamps_from_http_date() { + let server_ms = 784_111_777_000_i64; + assert_eq!( + clock_offset_from_http_date( + "Sun, 06 Nov 1994 08:49:37 GMT", + server_ms - 1_000, + server_ms + 1_000, + ), + Some(0), + ); + assert_eq!(corrected_timestamp(100, 2_500), 102); + assert_eq!(corrected_timestamp(100, -2_500), 97); + } + + #[test] + fn ignores_small_clock_jitter_until_the_calibration_expires() { + let mut calibration = ClockCalibration::default(); + assert!(calibration.observe(1_000)); + assert!(!calibration.observe(2_000)); + assert_eq!(calibration.offset_ms, 1_000); + assert!(calibration.observe(25_000)); + calibration.calibrated_at = Some(Instant::now() - CLOCK_CALIBRATION_MAX_AGE); + assert!(calibration.observe(25_001)); + } + #[test] fn parses_cli_options() { let parsed = CliOptions::try_parse_from([ diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 4a100d2..511543d 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,18 +1,45 @@ -import { Megaphone, Moon, Search, Sun, UserCircle } from "lucide-react"; -import { lazy, Suspense, useCallback, useEffect, useMemo, useRef, useState } from "react"; -import { api, ApiError } from "./api"; +import { KeyRound, Megaphone, Moon, Search, Sun, UserCircle } from "lucide-react"; +import { lazy, Suspense, useCallback, useEffect, useMemo, useRef, useState, type FormEvent } from "react"; +import { api, ApiError, setToken } from "./api"; import { NodeCard } from "./components/NodeCard"; import { StatsBar } from "./components/StatsBar"; import { TurnstileWidget } from "./components/TurnstileWidget"; import { demoConfig, demoExchangeRates, demoServers } from "./demo"; -import type { Config, ExchangeRates, Server } from "./types"; +import type { Config, ExchangeRates, LatencySample, LiveLatencyResult, Server } from "./types"; import { resolveBackground, themeToggle } from "./theme"; import { ui } from "./locale"; +import { derivePassword } from "./password"; const defaultConfig: Config = { ...demoConfig, site_description: "", site_name: "" }; const demoMode = import.meta.env.DEV && new URLSearchParams(window.location.search).has("demo"); const NodeDetails = lazy(() => import("./components/NodeDetails").then((module) => ({ default: module.NodeDetails }))); +interface LiveMetrics { + timestamp: number; + metrics: Partial; + latencyResults?: LiveLatencyResult[]; +} + +interface LiveSample { + ts: number; + data: Partial & { latency_results?: LiveLatencyResult[] }; +} + +function mergeLiveLatency(server: Server, results: LiveLatencyResult[]): LatencySample[] { + const definitions = new Map(server.latency.map((point) => [point.task_id, point])); + return results.flatMap((result) => { + const definition = definitions.get(result.task_id); + if (!definition || !Number.isFinite(result.timestamp)) return []; + return [{ + ...definition, + server_id: server.id, + timestamp: result.timestamp, + latency_ms: result.latency_ms, + packet_loss: result.packet_loss, + }]; + }); +} + function routeServerId() { const match = window.location.pathname.match(/^\/instance\/([^/]+)\/?$/); if (!match) return null; @@ -23,6 +50,7 @@ export default function App() { const [config, setConfig] = useState(defaultConfig); const [configReady, setConfigReady] = useState(demoMode); const [servers, setServers] = useState([]); + const [liveMetrics, setLiveMetrics] = useState>({}); const [exchangeRates, setExchangeRates] = useState(null); const [loading, setLoading] = useState(true); const [error, setError] = useState(""); @@ -31,15 +59,28 @@ export default function App() { const [selectedId, setSelectedId] = useState(routeServerId); const [needsVerification, setNeedsVerification] = useState(false); const [verificationBusy, setVerificationBusy] = useState(false); + const [needsLogin, setNeedsLogin] = useState(false); + const [loginUsername, setLoginUsername] = useState(""); + const [loginPassword, setLoginPassword] = useState(""); + const [loginTurnstileToken, setLoginTurnstileToken] = useState(""); + const [loginTurnstileReset, setLoginTurnstileReset] = useState(0); + const [loginBusy, setLoginBusy] = useState(false); const [appearance, setAppearance] = useState<"light" | "dark" | null>(() => { const stored = localStorage.getItem("nodeflare-theme"); return stored === "light" || stored === "dark" ? stored : null; }); const [systemDark, setSystemDark] = useState(() => matchMedia("(prefers-color-scheme: dark)").matches); const wsRef = useRef([]); + const liveConnectedRef = useRef(false); const dark = appearance ? appearance === "dark" : config.default_theme === "system" ? systemDark : config.default_theme === "dark"; const background = resolveBackground(config.background_url, dark); const blur = themeToggle(config, "enableBlur"); + const liveServers = useMemo(() => servers.map((server) => { + const live = liveMetrics[server.id]; + if (!live || live.timestamp <= (server.timestamp ?? 0)) return server; + const latency = live.latencyResults?.length ? mergeLiveLatency(server, live.latencyResults) : server.latency; + return { ...server, ...live.metrics, latency, timestamp: live.timestamp }; + }), [liveMetrics, servers]); const load = useCallback(async (quiet = false) => { if (!quiet) setLoading(true); @@ -52,26 +93,61 @@ export default function App() { setLoading(false); return; } + let dashboardPublic = true; try { const nextConfig = await api.config(); + dashboardPublic = nextConfig.public_dashboard; setConfig(nextConfig); setConfigReady(true); const [serverResult, ratesResult] = await Promise.all([api.servers(), api.exchangeRates()]); setServers(serverResult.servers); setExchangeRates(ratesResult); setNeedsVerification(false); + setNeedsLogin(false); setError(""); } catch (reason) { - if (reason instanceof ApiError && reason.status === 403) { - setNeedsVerification(true); + if (reason instanceof ApiError && (reason.status === 401 || (!dashboardPublic && reason.status === 403))) { + setNeedsLogin(true); + setNeedsVerification(false); setServers([]); + setLiveMetrics({}); + setError(""); + } else if (reason instanceof ApiError && reason.status === 403) { + setNeedsVerification(true); + setNeedsLogin(false); + setServers([]); + setLiveMetrics({}); + setError(reason.message); + } else { + setError(reason instanceof Error ? reason.message : ui(config.locale, "无法加载节点状态", "Unable to load server status")); } - setError(reason instanceof Error ? reason.message : ui(config.locale, "无法加载节点状态", "Unable to load server status")); } finally { setLoading(false); } }, []); + async function loginDashboard(event: FormEvent) { + event.preventDefault(); + if (loginBusy) return; + setLoginBusy(true); + setError(""); + try { + const passwordDerived = await derivePassword(loginPassword, config.password_client_salt); + const result = await api.login(loginUsername.trim(), loginPassword, passwordDerived, loginTurnstileToken); + setToken(result.token); + setLoginPassword(""); + setLoginTurnstileToken(""); + setNeedsLogin(false); + await load(); + } catch (reason) { + setError(reason instanceof Error ? reason.message : ui(config.locale, "登录失败", "Unable to sign in")); + setLoginTurnstileToken(""); + setLoginTurnstileReset((value) => value + 1); + } finally { + setLoginBusy(false); + } + } + const verifyDashboard = useCallback(async (token: string) => { if (!token || verificationBusy) return; setVerificationBusy(true); @@ -89,10 +165,18 @@ export default function App() { useEffect(() => { void load(); - const timer = window.setInterval(() => void load(true), 60_000); - return () => clearInterval(timer); }, [load]); + useEffect(() => { + if (needsLogin) return; + let ticks = 0; + const timer = window.setInterval(() => { + ticks += 1; + if (!liveConnectedRef.current || ticks % 4 === 0) void load(true); + }, 15_000); + return () => clearInterval(timer); + }, [load, needsLogin]); + useEffect(() => { document.documentElement.classList.toggle("dark", dark); document.documentElement.style.colorScheme = dark ? "dark" : "light"; @@ -125,12 +209,12 @@ export default function App() { useEffect(() => { if (!configReady) return; - const selected = servers.find((server) => server.id === selectedId); + const selected = liveServers.find((server) => server.id === selectedId); document.title = selected ? `${selected.name} · ${config.site_name}` : config.site_name; - }, [config.site_name, configReady, selectedId, servers]); + }, [config.site_name, configReady, liveServers, selectedId]); useEffect(() => { - if (loading || needsVerification || !config.public_dashboard || demoMode) return; + if (loading || needsLogin || needsVerification || demoMode) return; const reconnects: number[] = []; let cancelled = false; const connect = () => { @@ -142,19 +226,67 @@ export default function App() { if (selectedId) endpoint.searchParams.set("server_id", selectedId); const socket = new WebSocket(endpoint); wsRef.current.push(socket); - socket.onopen = () => void load(true); + socket.onopen = () => { + liveConnectedRef.current = true; + void load(true); + }; socket.onclose = () => { wsRef.current = wsRef.current.filter((current) => current !== socket); + liveConnectedRef.current = wsRef.current.some((current) => current.readyState === WebSocket.OPEN); if (!cancelled) reconnects.push(window.setTimeout(connect, 3000)); }; + socket.onerror = () => { + liveConnectedRef.current = false; + }; socket.onmessage = (event) => { if (event.data === "pong") return; try { const message = JSON.parse(event.data); - if (message.type !== "server" || !message.server?.id) return; - setServers((current) => current.map((server) => server.id === message.server.id - ? message.server - : server)); + if (message.type === "server" && message.server?.id) { + setServers((current) => current.map((server) => server.id === message.server.id + ? message.server + : server)); + return; + } + if (message.type === "batchUpdate" && Array.isArray(message.updates)) { + setLiveMetrics((current) => { + const next = { ...current }; + for (const update of message.updates as Array<{ serverId?: string; samples?: LiveSample[] }>) { + if (!update.serverId || !Array.isArray(update.samples) || !update.samples.length) continue; + const samples = update.samples + .filter((sample) => Number.isFinite(sample.ts) && sample.data && typeof sample.data === "object") + .sort((left, right) => left.ts - right.ts); + const latest = samples.at(-1); + if (!latest) continue; + const previous = next[update.serverId]; + if (previous && previous.timestamp >= latest.ts) continue; + const results = latest.data.latency_results; + next[update.serverId] = { + timestamp: latest.ts, + metrics: latest.data, + latencyResults: Array.isArray(results) ? results : previous?.latencyResults, + }; + } + return next; + }); + return; + } + if (message.type !== "metrics" || !message.server_id || !message.metrics) return; + if (!Number.isFinite(message.timestamp)) return; + setLiveMetrics((current) => { + const previous = current[message.server_id]; + if (previous && previous.timestamp >= message.timestamp) return current; + return { + ...current, + [message.server_id]: { + timestamp: message.timestamp, + metrics: message.metrics, + latencyResults: Array.isArray(message.metrics.latency_results) + ? message.metrics.latency_results + : previous?.latencyResults, + }, + }; + }); } catch { /* Ignore non-protocol messages. */ } }; }; @@ -162,21 +294,22 @@ export default function App() { const heartbeat = window.setInterval(() => wsRef.current.forEach((socket) => socket.readyState === WebSocket.OPEN && socket.send("ping")), 30_000); return () => { cancelled = true; + liveConnectedRef.current = false; clearInterval(heartbeat); reconnects.forEach(clearTimeout); wsRef.current.forEach((socket) => socket.close()); wsRef.current = []; }; - }, [config.public_dashboard, load, loading, needsVerification, selectedId]); + }, [load, loading, needsLogin, needsVerification, selectedId]); - const groups = useMemo(() => ["__all__", ...Array.from(new Set(servers.map((server) => server.group_name || "默认")))], [servers]); - const visible = useMemo(() => servers.filter((server) => { + const groups = useMemo(() => ["__all__", ...Array.from(new Set(liveServers.map((server) => server.group_name || "默认")))], [liveServers]); + const visible = useMemo(() => liveServers.filter((server) => { const text = `${server.name} ${server.region} ${server.tags} ${server.group_name}`.toLowerCase(); const groupMatches = !config.show_groups || group === "__all__" || (server.group_name || "默认") === group; const queryMatches = !config.show_search || text.includes(query.trim().toLowerCase()); return groupMatches && queryMatches; - }), [servers, query, group, config.show_groups, config.show_search]); - const selected = servers.find((server) => server.id === selectedId) ?? null; + }), [liveServers, query, group, config.show_groups, config.show_search]); + const selected = liveServers.find((server) => server.id === selectedId) ?? null; function goHome() { window.history.pushState({}, "", demoMode ? "/?demo=1" : "/"); @@ -220,7 +353,19 @@ export default function App() {
{config.site_announcement ?
{config.site_announcement}
: null} - {needsVerification ? ( + {needsLogin ? ( +
+ +

{ui(config.locale, "登录仪表盘", "Sign in to dashboard")}

{ui(config.locale, "此仪表盘仅限登录后访问", "This dashboard requires an administrator sign-in")}

+
void loginDashboard(event)}> + + + {config.turnstile_login_enabled ?
: null} + {error ?

{error}

: null} + +
+
+ ) : needsVerification ? (

{ui(config.locale, "访问验证", "Access verification")}

{config.site_name}

@@ -229,18 +374,18 @@ export default function App() { {error ?

{error}

: null}
) : selectedId ? selected ? ( - {ui(config.locale, "正在加载节点", "Loading server")}}> + {ui(config.locale, "正在加载节点", "Loading server")}}> ) : loading ?
{ui(config.locale, "正在加载节点", "Loading server")}
:
{ui(config.locale, "节点不存在或已隐藏", "Server not found or hidden")}
:
- + {config.show_search || config.show_groups ?
{config.show_search ?
setQuery(event.target.value)} />
: null} {config.show_groups ?
{groups.map((item) => )}
: null} {ui(config.locale, `${visible.length} 个节点`, `${visible.length} servers`)}
: null} {error ?
{error}
: null} - {loading && !servers.length ?
{Array.from({ length: 8 }).map((_, index) => )}
: visible.length ? ( + {loading && !liveServers.length ?
{Array.from({ length: 8 }).map((_, index) => )}
: visible.length ? (
{visible.map((server) => openServer(server)} />)}
- ) : !error ?
{servers.length ? ui(config.locale, "没有匹配的节点", "No matching servers") : ui(config.locale, "尚未添加节点", "No servers added")}
: null} + ) : !error ?
{liveServers.length ? ui(config.locale, "没有匹配的节点", "No matching servers") : ui(config.locale, "尚未添加节点", "No servers added")}
: null}
}
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index e0ed4a7..3bf9b70 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -54,6 +54,7 @@ export const api = { method: "POST", body: JSON.stringify({ username, password, password_derived: passwordDerived, turnstile_token: turnstileToken }), }), + logout: () => request("/api/admin/logout", { method: "POST" }), settings: () => request("/api/admin/settings", {}, true), latencyTasks: () => request<{ tasks: LatencyTask[] }>("/api/admin/latency-tasks", {}, true), createLatencyTask: (input: LatencyTaskInput) => diff --git a/frontend/src/components/AdminPanel.tsx b/frontend/src/components/AdminPanel.tsx index 4e65614..6183895 100644 --- a/frontend/src/components/AdminPanel.tsx +++ b/frontend/src/components/AdminPanel.tsx @@ -3,7 +3,9 @@ import { ArrowLeft, ChevronDown, ChevronUp, + CircleAlert, Cloud, + CircleCheck, Coins, Copy, Database, @@ -203,6 +205,12 @@ export function AdminPanel({ return () => window.clearTimeout(timer); }, [notice]); + useEffect(() => { + if (!error) return; + const timer = window.setTimeout(() => setError(""), 6000); + return () => window.clearTimeout(timer); + }, [error]); + async function login(event: FormEvent) { event.preventDefault(); setBusy(true); @@ -278,6 +286,14 @@ export function AdminPanel({ } } + async function copyInstallCommands() { + try { + await navigator.clipboard.writeText(commands.map((item) => `# ${item.name}\n${item.command}`).join("\n\n")); + } catch { + setError("复制失败,请手动选择命令复制"); + } + } + async function move(index: number, offset: number) { const target = index + offset; if (target < 0 || target >= servers.length) return; @@ -438,8 +454,11 @@ export function AdminPanel({ finally { setBusy(false); } } - function logout() { - setToken(""); setAuthenticated(false); setServers([]); setSettings(null); setSelectedIds([]); + async function logout() { + try { await api.logout(); } + finally { + setToken(""); setAuthenticated(false); setServers([]); setSettings(null); setSelectedIds([]); + } } const commands = useMemo(() => install.map((server) => { @@ -459,7 +478,7 @@ export function AdminPanel({ } return { ...server, - command: `wget -qO- ${installer}/agent.sh | sudo sh -s -- -e ${shellLiteral(origin)} -t ${shellLiteral(server.agent_token)}`, + command: `curl -fsSL ${installer}/agent.sh | sudo sh -s -- -e ${shellLiteral(origin)} -t ${shellLiteral(server.agent_token)}`, }; }), [install, installPlatform]); @@ -479,6 +498,8 @@ export function AdminPanel({ return (
+ {error ?
+ : notice ?
: null} {!authenticated ?
@@ -488,7 +509,6 @@ export function AdminPanel({ {config.turnstile_login_enabled || config.turnstile_enabled ?
: null} - {error ?

{error}

: null}
:
@@ -500,7 +520,7 @@ export function AdminPanel({
- +
@@ -520,8 +540,6 @@ export function AdminPanel({

{adminPages[tab].title}

{adminPages[tab].description}

- {error ?

{error}

: null} - {notice ?

{notice}

: null} {tab === "servers" ? (

监控节点

{servers.length} 个节点 · 可拖动上下排序
@@ -532,7 +550,7 @@ export function AdminPanel({ toggleSelected(server.id)} ariaLabel={`选择 ${server.name}`} /> -
{server.name}{server.group_name} · {server.region || "未设置地区"} · {server.last_ip || "尚未上报 IP"} · {server.report_interval}s
+
{server.name}{server.group_name} · {server.region || "未设置地区"} · {server.last_ip || "尚未上报 IP"} · {server.agent_version ? `Agent v${server.agent_version}` : "Agent 未上报版本"} · {server.report_interval}s
{server.traffic_limit > 0 ? formatBytes(server.traffic_limit) : "不限流量"}
@@ -603,7 +621,7 @@ export function AdminPanel({
当前密码状态:{settings.admin_password_configured ? "已使用 D1 加盐哈希" : "使用 ADMIN_PASSWORD 初始化密钥"}
updateSettings("turnstile_enabled", value)} /> updateSettings("turnstile_login_enabled", value)} />
-
+

留空时读取 Worker 的 TURNSTILE_SITE_KEY 和 TURNSTILE_SECRET_KEY,后台配置优先。

: null} @@ -616,9 +634,9 @@ export function AdminPanel({
Cloudflare 用量

统计周期使用 UTC

-
-

留空时读取 Worker Secret 中的同名变量。

- {cloudflareUsage ?
周期D1 读取D1 写入Workers 请求
:
尚未读取
} +
+

Token 需要账户级 Account Analytics: Read 权限,并授权对应账户;留空读取 Worker Secret。

+ {cloudflareUsage ? <>
周期D1 读取D1 写入Workers 请求DO 请求(估算)DO 时长 (GB-s)
:
尚未读取
}

清空历史不会删除节点、密钥或最新状态。

@@ -642,7 +660,7 @@ export function AdminPanel({
: null} - {commands.length ?
setInstall([])}>
event.stopPropagation()}>
Agent 部署

安装命令

{commands.map((item) =>
{item.name}
{item.command}
)}
: null} + {commands.length ?
setInstall([])}>
event.stopPropagation()}>
Agent 部署

安装命令

{commands.map((item) =>
{item.name}
{item.command}
)}
: null} ); } @@ -672,5 +690,9 @@ function DataStat({ label, value }: { label: string; value: number | string }) { } function UsageRow({ label, usage }: { label: string; usage: CloudflareUsage["today"] }) { - return {label}{usage.rows_read.toLocaleString()}{usage.rows_written.toLocaleString()}{usage.workers_requests.toLocaleString()}; + return {label}{usage.rows_read.toLocaleString()}{usage.rows_written.toLocaleString()}{usage.workers_requests.toLocaleString()}{usage.durable_objects_requests.toLocaleString()}{usage.durable_objects_duration.toLocaleString(undefined, { maximumFractionDigits: 2 })}; +} + +function UsageDoBreakdown({ label, usage }: { label: string; usage: CloudflareUsage["today"] }) { + return
{label}HTTP {usage.durable_objects_http_requests.toLocaleString()}休眠唤醒 {usage.durable_objects_hibernation_wakeups.toLocaleString()}WS 入站 {usage.durable_objects_inbound_websocket_messages.toLocaleString()}WS 出站 {usage.durable_objects_outbound_websocket_messages.toLocaleString()}每 {usage.durable_objects_request_billing_ratio} 条入站消息折算 1 次请求
; } diff --git a/frontend/src/components/NodeCard.tsx b/frontend/src/components/NodeCard.tsx index ef6e2a2..3d42a1a 100644 --- a/frontend/src/components/NodeCard.tsx +++ b/frontend/src/components/NodeCard.tsx @@ -74,7 +74,7 @@ export function NodeCard({ server, config, onOpen }: { server: Server; config: C const locale = config.locale; const price = formatPrice(server, locale); const remainingValue = remainingAssetValue(server.price, server.billing_cycle, server.expires_at); - const showExpiryPanel = config.show_expiry && (server.price !== 0 || server.expires_at !== null); + const showExpiryPanel = config.show_expiry || config.show_price; return ( ); diff --git a/frontend/src/components/NodeDetails.tsx b/frontend/src/components/NodeDetails.tsx index 7c51404..2d8b8fe 100644 --- a/frontend/src/components/NodeDetails.tsx +++ b/frontend/src/components/NodeDetails.tsx @@ -3,13 +3,12 @@ import { ArrowLeft, Box, Cpu, - HardDrive, Network, Radio, RadioTower, ServerCog, } from "lucide-react"; -import { useEffect, useMemo, useState } from "react"; +import { useEffect, useMemo, useRef, useState } from "react"; import { Area, AreaChart, @@ -25,7 +24,7 @@ import { import { api } from "../api"; import { demoHistory, demoLatencyHistory, demoLatencyTasks } from "../demo"; import { displayGpuDevices, formatBytes, formatCpuName, formatSpeed, formatUptime, isOnline, number } from "../format"; -import type { HistoryPoint, LatencySample, LatencyTestPoint, Server } from "../types"; +import type { HistoryPoint, LatencySample, LatencyTestPoint, LiveLatencyResult, Server } from "../types"; import { Flag, regionDisplayName } from "./Flag"; import { OSIcon } from "./OSIcon"; import { ui } from "../locale"; @@ -33,6 +32,48 @@ import { ui } from "../locale"; type ChartType = "load" | "latency"; const LATENCY_COLORS = ["#2563eb", "#db2777", "#ea7b1b", "#0f766e", "#7c3aed", "#0891b2"]; +const REALTIME_WINDOW_SECONDS = 60 * 60; +const MAX_REALTIME_POINTS = 720; + +function historyPointFromServer(server: Server): HistoryPoint | null { + if (!Number.isFinite(server.timestamp) || number(server.timestamp) <= 0) return null; + return { + timestamp: number(server.timestamp), + cpu: number(server.cpu), + load1: number(server.load1), + load5: number(server.load5), + load15: number(server.load15), + mem_used: number(server.mem_used), + mem_total: number(server.mem_total), + swap_used: number(server.swap_used), + swap_total: number(server.swap_total), + disk_used: number(server.disk_used), + disk_total: number(server.disk_total), + net_in: number(server.net_in), + net_out: number(server.net_out), + net_rx_total: number(server.net_rx_total), + net_tx_total: number(server.net_tx_total), + processes: number(server.processes), + tcp_connections: number(server.tcp_connections), + udp_connections: number(server.udp_connections), + gpu_usage: number(server.gpu_usage), + disk_read_bps: number(server.disk_read_bps), + disk_write_bps: number(server.disk_write_bps), + disk_read_iops: number(server.disk_read_iops), + disk_write_iops: number(server.disk_write_iops), + disk_await_ms: number(server.disk_await_ms), + disk_utilization: number(server.disk_utilization), + }; +} + +function appendRealtimePoint(points: HistoryPoint[], point: HistoryPoint | null): HistoryPoint[] { + if (!point) return points; + const cutoff = point.timestamp - REALTIME_WINDOW_SECONDS; + const next = points.filter((current) => current.timestamp >= cutoff && current.timestamp !== point.timestamp); + next.push(point); + next.sort((left, right) => left.timestamp - right.timestamp); + return next.slice(-MAX_REALTIME_POINTS); +} function InfoItem({ icon, label, value }: { icon: React.ReactNode; label: string; value: React.ReactNode }) { return
{icon}
{label}{value}
; @@ -62,49 +103,125 @@ const tooltipStyle = { fontSize: 12, }; -export function NodeDetails({ server, threshold, retentionDays, locale, demo = false, onClose }: { +export function NodeDetails({ server, liveLatencyResults, threshold, retentionDays, locale, demo = false, onClose }: { server: Server; + liveLatencyResults: LiveLatencyResult[]; threshold: number; retentionDays: number; locale: "zh-CN" | "en"; demo?: boolean; onClose: () => void; }) { - const [hours, setHours] = useState(0); + const [loadHours, setLoadHours] = useState(0); + const [latencyHours, setLatencyHours] = useState(1); const [chartType, setChartType] = useState("load"); const [points, setPoints] = useState([]); const [latencyPoints, setLatencyPoints] = useState([]); const [latencyTasks, setLatencyTasks] = useState([]); - const [loading, setLoading] = useState(true); - const [error, setError] = useState(""); + const [loadLoading, setLoadLoading] = useState(true); + const [latencyLoading, setLatencyLoading] = useState(true); + const [loadError, setLoadError] = useState(""); + const [latencyError, setLatencyError] = useState(""); + const latestServerRef = useRef(server); + latestServerRef.current = server; useEffect(() => { - setLoading(true); - setError(""); - const requestHours = hours === 0 ? 1 : hours; + setLoadLoading(true); + setLoadError(""); + const requestHours = loadHours === 0 ? 1 : loadHours; if (demo) { - const taskIds = new Set(server.latency.map((point) => point.task_id)); - const tasks = demoLatencyTasks.filter((task) => taskIds.has(task.id)); setPoints(demoHistory(server.id, requestHours)); - setLatencyTasks(tasks); - setLatencyPoints(tasks.length ? demoLatencyHistory(server.id, requestHours).filter((point) => taskIds.has(point.task_id)) : []); - setLoading(false); + setLoadLoading(false); return; } let active = true; - Promise.all([api.history(server.id, requestHours), api.latencyHistory(server.id, requestHours)]) - .then(([history, latency]) => { if (active) { setPoints(history.points); setLatencyTasks(latency.tasks); setLatencyPoints(latency.points); } }) - .catch((reason) => { if (active) { setPoints([]); setLatencyTasks([]); setLatencyPoints([]); setError(reason instanceof Error ? reason.message : ui(locale, "历史数据加载失败", "Unable to load history")); } }) - .finally(() => { if (active) setLoading(false); }); + api.history(server.id, requestHours) + .then((history) => { + if (active) setPoints(loadHours === 0 + ? appendRealtimePoint(history.points, historyPointFromServer(latestServerRef.current)) + : history.points); + }) + .catch((reason) => { if (active) { setPoints([]); setLoadError(reason instanceof Error ? reason.message : ui(locale, "历史数据加载失败", "Unable to load history")); } }) + .finally(() => { if (active) setLoadLoading(false); }); return () => { active = false; }; - }, [server.id, hours, demo, locale]); + }, [server.id, loadHours, demo, locale]); + + useEffect(() => { + if (demo || loadHours !== 0) return; + const point = historyPointFromServer(server); + if (point) setPoints((current) => appendRealtimePoint(current, point)); + }, [demo, loadHours, server]); + + useEffect(() => { + setLatencyLoading(true); + setLatencyError(""); + if (demo) { + const taskIds = new Set(server.latency.map((point) => point.task_id)); + const tasks = demoLatencyTasks.filter((task) => taskIds.has(task.id)); + setLatencyTasks(tasks); + setLatencyPoints(tasks.length ? demoLatencyHistory(server.id, latencyHours).filter((point) => taskIds.has(point.task_id)) : []); + setLatencyLoading(false); + return; + } + let active = true; + api.latencyHistory(server.id, latencyHours) + .then((latency) => { if (active) { setLatencyTasks(latency.tasks); setLatencyPoints(latency.points); } }) + .catch((reason) => { if (active) { setLatencyPoints([]); setLatencyError(reason instanceof Error ? reason.message : ui(locale, "延迟数据加载失败", "Unable to load latency")); } }) + .finally(() => { if (active) setLatencyLoading(false); }); + return () => { active = false; }; + }, [server.id, latencyHours, demo, locale]); + + useEffect(() => { + if (demo || !server.latency.length) return; + const latest = Math.max(...server.latency.map((point) => point.timestamp).filter(Number.isFinite)); + if (!Number.isFinite(latest)) return; + const cutoff = latest - latencyHours * 3600; + setLatencyPoints((current) => { + const next = new Map(current + .filter((point) => point.timestamp >= cutoff) + .map((point) => [`${point.task_id}:${point.timestamp}`, point])); + for (const point of server.latency) { + if (point.timestamp >= cutoff) { + next.set(`${point.task_id}:${point.timestamp}`, point); + } + } + return Array.from(next.values()).sort((left, right) => left.timestamp - right.timestamp); + }); + }, [demo, latencyHours, server.latency]); + + useEffect(() => { + if (demo || !liveLatencyResults.length || !latencyTasks.length) return; + const definitions = new Map(latencyTasks.map((task) => [task.id, task])); + const samples = liveLatencyResults.flatMap((result): LatencySample[] => { + const task = definitions.get(result.task_id); + if (!task || !Number.isFinite(result.timestamp)) return []; + return [{ + task_id: task.id, + server_id: server.id, + name: task.name, + task_type: task.task_type, + target: task.target, + port: task.port, + timestamp: result.timestamp, + latency_ms: result.latency_ms, + packet_loss: result.packet_loss, + }]; + }); + if (!samples.length) return; + const latest = Math.max(...samples.map((point) => point.timestamp)); + const cutoff = latest - latencyHours * 3600; + setLatencyPoints((current) => { + const next = new Map(current + .filter((point) => point.timestamp >= cutoff) + .map((point) => [`${point.task_id}:${point.timestamp}`, point])); + for (const point of samples) next.set(`${point.task_id}:${point.timestamp}`, point); + return Array.from(next.values()).sort((left, right) => left.timestamp - right.timestamp); + }); + }, [demo, latencyHours, latencyTasks, liveLatencyResults, server.id]); const online = isOnline(server, threshold); const gpuDevices = useMemo(() => displayGpuDevices(server.gpus), [server.gpus]); const gpuNames = gpuDevices.map((gpu) => gpu.model).join(" · "); - const gpuMemory = gpuDevices.map((gpu) => gpu.memory_total > 0 - ? `${formatBytes(gpu.memory_used)} / ${formatBytes(gpu.memory_total)}` - : ui(locale, "共享内存", "Shared memory")).join(" · "); const loadRanges = [ { value: 0, label: ui(locale, "实时", "Live") }, { value: 1, label: ui(locale, "1 小时", "1 hour") }, @@ -114,6 +231,7 @@ export function NodeDetails({ server, threshold, retentionDays, locale, demo = f { value: 720, label: ui(locale, "30 天", "30 days") }, ].filter((range) => range.value <= Math.max(24, retentionDays * 24)); const ranges = chartType === "load" ? loadRanges : loadRanges.filter((range) => range.value > 0); + const hours = chartType === "load" ? loadHours : latencyHours; const data = useMemo(() => points.map((point) => ({ ...point, time: new Date(point.timestamp * 1000).toLocaleString(locale, hours >= 24 @@ -151,11 +269,6 @@ export function NodeDetails({ server, threshold, retentionDays, locale, demo = f const last = data[data.length - 1]; const pingEnabled = latencyTasks.length > 0; - function selectChart(next: ChartType) { - setChartType(next); - if (next === "latency" && hours === 0) setHours(1); - } - return (
@@ -171,35 +284,31 @@ export function NodeDetails({ server, threshold, retentionDays, locale, demo = f }> } label="CPU" value={formatCpuName(server.cpu_model, server.cpu_cores)} /> } label={ui(locale, "架构", "Architecture")} value={server.arch || "--"} /> - {gpuDevices.length ? <> - } label="GPU" value={gpuNames} /> - } label={ui(locale, "GPU 显存", "GPU Memory")} value={gpuMemory} /> - : null} - } label={ui(locale, "存储设备", "Storage")} value={server.disks.length ? server.disks.map((disk) => `${disk.mount_point} ${formatBytes(disk.used)} / ${formatBytes(disk.total)}`).join(" · ") : `${formatBytes(server.disk_used)} / ${formatBytes(server.disk_total)}`} /> + } label="GPU" value={gpuNames || ui(locale, "未检测到", "Not detected")} /> } label={ui(locale, "虚拟化", "Virtualization")} value={server.virtualization || "--"} /> }> } label={ui(locale, "操作系统", "Operating system")} value={server.os || "--"} /> - } label={ui(locale, "运行时间", "Uptime")} value={online ? formatUptime(server.uptime, locale) : "--"} /> + } label={ui(locale, "运行时间", "Uptime")} value={formatUptime(server.uptime, locale)} /> } label={ui(locale, "内核版本", "Kernel")} value={server.kernel || "--"} /> - } label={ui(locale, "进程 / 连接", "Processes / connections")} value={online ? `${server.processes || 0} / ${number(server.tcp_connections) + number(server.udp_connections)}` : "--"} /> - } label={ui(locale, "磁盘 IO", "Disk IO")} value={`${number(server.disk_await_ms).toFixed(1)} ms · ${number(server.disk_utilization).toFixed(1)}%`} /> + } label={ui(locale, "进程 / 连接", "Processes / connections")} value={`${number(server.processes)} / ${number(server.tcp_connections) + number(server.udp_connections)}`} />
- - {pingEnabled ? : null} + + {pingEnabled ? : null}
- {ranges.map((range) => )} + {ranges.map((range) => )}
- {error ?
{error}
: null} - {loading ?
{ui(locale, "正在读取历史数据", "Loading history")}
: chartType === "load" && !data.length ?
{ui(locale, "暂无历史数据", "No history")}
: chartType === "latency" && !(latencyData.length || data.length) ?
{ui(locale, "暂无延迟数据", "No latency data")}
: chartType === "load" ? ( + {chartType === "load" && loadError ?
{loadError}
: null} + {chartType === "latency" && latencyError ?
{latencyError}
: null} + {chartType === "load" && loadLoading ?
{ui(locale, "正在读取历史数据", "Loading history")}
: chartType === "latency" && latencyLoading ?
{ui(locale, "正在读取延迟数据", "Loading latency")}
: chartType === "load" && !data.length ?
{ui(locale, "暂无历史数据", "No history")}
: chartType === "load" ? (
[`${Number(value).toFixed(1)}%`, "CPU"]} /> @@ -224,7 +333,7 @@ export function NodeDetails({ server, threshold, retentionDays, locale, demo = f
)}
- + {latencyData.length ? @@ -233,7 +342,7 @@ export function NodeDetails({ server, threshold, retentionDays, locale, demo = f {latencySeries.map((series) => )} - + :
{ui(locale, "暂无延迟数据", "No latency data")}
}
)} diff --git a/frontend/src/styles.css b/frontend/src/styles.css index 4a9de98..8f489b1 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -177,10 +177,6 @@ html[data-blur="off"] .glass-panel { background: color-mix(in srgb, var(--card-s .quality-bar:hover { z-index: 20; } .quality-bar:hover .quality-tooltip { visibility: visible; opacity: 1; transform: translateY(0); } .quality-bar:hover .quality-tooltip.center { transform: translate(-50%, 0); } -.offline-layer { position: absolute; z-index: 3; inset: 0; display: flex; align-items: center; justify-content: center; flex-direction: column; border-radius: 0 0 12px 12px; background: var(--node-offline-bg); backdrop-filter: blur(2px); } -html[data-blur="off"] .offline-layer { backdrop-filter: none; } -.offline-layer strong { color: var(--danger); font-size: 13px; } -.offline-layer span { margin-top: 4px; color: var(--muted-foreground); font-size: 9px; } .error-band { display: flex; align-items: center; justify-content: space-between; border: 1px solid color-mix(in srgb, var(--danger) 32%, var(--border)); border-radius: 7px; padding: 12px 14px; color: var(--danger); background: color-mix(in srgb, var(--danger) 7%, transparent); font-size: 11px; } .error-band button { font-weight: 700; } @@ -188,6 +184,16 @@ html[data-blur="off"] .offline-layer { backdrop-filter: none; } .verification-gate { display: flex; width: min(430px, 100%); min-height: 380px; align-items: center; justify-content: center; flex-direction: column; gap: 14px; margin: 12vh auto 0; border: 1px solid var(--border); border-radius: 8px; padding: 30px; background: var(--card); box-shadow: var(--shadow); text-align: center; } .verification-gate h1 { margin: 0; font-size: 19px; } .verification-gate p { margin: 4px 0 0; color: var(--muted-foreground); font-size: 10px; } +.dashboard-login-gate { display: flex; width: min(420px, 100%); align-items: center; flex-direction: column; gap: 16px; margin: 9vh auto 0; border-radius: 8px; padding: 28px; } +.dashboard-login-copy { text-align: center; } +.dashboard-login-copy h1 { margin: 0; font-size: 19px; } +.dashboard-login-copy p { margin: 6px 0 0; color: var(--muted-foreground); font-size: 11px; line-height: 1.5; } +.dashboard-login-form { display: flex; width: 100%; flex-direction: column; gap: 13px; } +.dashboard-login-form label > span { font-size: 11px; } +.dashboard-login-form input { min-height: 42px; font-size: 13px; } +.dashboard-login-turnstile { display: flex; width: 100%; justify-content: center; } +.dashboard-login-turnstile .turnstile-widget { width: 100%; } +.dashboard-login-submit { width: 100%; min-height: 42px; } .verification-status { color: var(--primary); font-size: 10px; } .turnstile-widget { width: min(100%, 300px); min-height: 66px; } .turnstile-widget.testing { height: 51px; min-height: 51px; overflow: hidden; } @@ -238,6 +244,7 @@ html[data-blur="off"] .offline-layer { backdrop-filter: none; } .latency-summary strong { display: block; margin-top: 6px; font-size: 22px; font-weight: 650; font-variant-numeric: tabular-nums; letter-spacing: 0; } .latency-summary small { display: block; margin-top: 5px; color: var(--muted-foreground); font-size: 11px; font-variant-numeric: tabular-nums; } .latency-line-chart { width: 100%; height: 320px; padding: 16px 18px 10px 10px; } +.latency-empty { display: grid; width: 100%; height: 100%; place-items: center; color: var(--muted-foreground); font-size: 12px; } .modal-backdrop, .submodal-backdrop { position: fixed; z-index: 80; inset: 0; display: flex; align-items: center; justify-content: center; padding: 24px; background: rgb(5 10 12 / .56); backdrop-filter: blur(6px); } .submodal-backdrop { z-index: 90; background: rgb(5 10 12 / .42); } @@ -287,12 +294,9 @@ html[data-blur="off"] .offline-layer { backdrop-filter: none; } .login-turnstile:has(.turnstile-widget.testing) { min-height: 51px; } .login-turnstile:has(.turnstile-widget.verified) { min-height: 0; } .login-submit { width: 100%; min-height: 40px; } -.login-error { border: 1px solid color-mix(in srgb, var(--danger) 32%, var(--border)); border-radius: 6px; padding: 8px 10px; background: color-mix(in srgb, var(--danger) 6%, transparent); } label { display: flex; min-width: 0; flex-direction: column; gap: 6px; } label > span { color: var(--muted-foreground); font-size: 9px; } .form-error { margin: 0; color: var(--danger); font-size: 9px; } -.form-notice { margin: 0 21px 9px; border: 1px solid color-mix(in srgb, var(--success) 32%, var(--border)); border-radius: 6px; padding: 8px 10px; color: var(--success); background: color-mix(in srgb, var(--success) 7%, transparent); font-size: 9px; } -.admin-error { margin: 0 21px 10px; border: 1px solid color-mix(in srgb, var(--danger) 32%, var(--border)); border-radius: 6px; padding: 8px 10px; background: color-mix(in srgb, var(--danger) 6%, transparent); line-height: 1.5; } .settings-form, .editor-modal { display: flex; flex-direction: column; gap: 14px; } .form-grid { display: grid; gap: 12px; grid-template-columns: 1fr 1fr; } .form-grid.three { grid-template-columns: repeat(3, 1fr); } @@ -328,6 +332,11 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .usage-table thead th { color: var(--muted-foreground); background: color-mix(in srgb, var(--foreground) 3%, transparent); font-weight: 600; } .usage-table tbody th { font-weight: 650; } .usage-table tbody tr:last-child > * { border-bottom: 0; } +.cloudflare-usage-table { min-width: 720px; } +.usage-do-breakdown { display: grid; gap: 7px; grid-template-columns: 1fr 1fr; } +.usage-do-breakdown > div { display: flex; min-width: 0; align-items: center; flex-wrap: wrap; gap: 5px 10px; border: 1px solid var(--border); border-radius: 7px; padding: 9px 10px; color: var(--muted-foreground); font-size: 9px; } +.usage-do-breakdown strong { color: var(--foreground); } +.usage-do-breakdown small { flex-basis: 100%; color: var(--muted-foreground); font-size: inherit; } .usage-empty { display: flex; min-height: 48px; align-items: center; justify-content: center; border: 1px dashed var(--border); border-radius: 7px; color: var(--muted-foreground); font-size: 9px; } .data-actions { border-top: 1px solid var(--border); padding-top: 14px; } .theme-store-page { display: grid; gap: 16px; } @@ -451,7 +460,6 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .admin-tabs button { min-height: 33px; flex: 0 0 auto; } .admin-content { height: calc(100vh - 112px); } .admin-section, .settings-form { padding: 14px; } - .admin-error { margin-inline: 14px; } .section-head { align-items: flex-start; } .server-row { padding-block: 6px; grid-template-columns: 18px 20px 9px minmax(0, 1fr) auto; } .server-usage { display: none; } @@ -461,6 +469,7 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .theme-card-icon { width: 32px; height: 32px; } .theme-card-actions { grid-column: 2; justify-content: flex-start; } .data-stat-grid { grid-template-columns: 1fr 1fr 1fr; } + .usage-do-breakdown { grid-template-columns: 1fr; } .usage-table th, .usage-table td { padding-inline: 5px; font-size: 8px; } .editor-modal, .install-modal { padding: 15px 13px; } .editor-toggles { grid-template-columns: 1fr 1fr; } @@ -525,7 +534,6 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .login-form img { width: 44px; height: 44px; } .login-copy h1 { color: var(--admin-text); font-size: 22px; } .login-copy p { color: var(--admin-muted); font-size: 13px; } -.login-error, .admin-error { font-size: 13px; } .admin-shell { display: grid; width: 100%; max-width: 100vw; height: 100dvh; min-width: 0; min-height: 0; overflow: hidden; border: 0; border-radius: 0; background: var(--admin-surface); grid-template-rows: 64px minmax(0, 1fr); } .admin-topbar { z-index: 2; display: flex; width: 100%; max-width: 100vw; min-width: 0; align-items: center; justify-content: space-between; gap: 24px; overflow: hidden; border-bottom: 1px solid var(--admin-border); padding: 0 24px; background: var(--admin-surface); } .admin-brand { display: flex; min-width: 0; align-items: center; gap: 10px; padding: 0; color: var(--admin-text); text-align: left; text-decoration: none; } @@ -546,6 +554,11 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .admin-content-header { padding: 25px 28px 17px; } .admin-content-header h1 { margin: 0; color: var(--admin-text); font-size: 22px; line-height: 1.25; } .admin-content-header p { margin: 6px 0 0; color: var(--admin-muted); font-size: 13px; } +.admin-toast { position: fixed; z-index: 120; top: 80px; right: 24px; display: flex; width: min(420px, calc(100vw - 32px)); min-height: 64px; align-items: center; gap: 11px; border: 1px solid var(--admin-border); border-radius: 8px; padding: 14px 17px; color: var(--admin-text); background: var(--admin-surface); box-shadow: 0 12px 34px rgb(15 23 42 / .18); font-size: 14px; font-weight: 600; line-height: 1.45; animation: admin-toast-in .18s ease-out; } +.admin-toast svg { width: 21px; height: 21px; flex: 0 0 auto; color: var(--success); stroke-width: 2.5; } +.admin-toast.error { border-color: color-mix(in srgb, var(--danger) 38%, var(--admin-border)); } +.admin-toast.error svg { color: var(--danger); } +@keyframes admin-toast-in { from { opacity: 0; transform: translateY(-8px); } to { opacity: 1; transform: translateY(0); } } .admin-section, .settings-form { margin: 0 28px 28px; border: 1px solid var(--admin-border); border-radius: 7px; padding: 21px; color: var(--admin-text); background: var(--admin-surface); } .section-head { margin-bottom: 16px; } .section-head h3 { color: var(--admin-text); font-size: 17px; } @@ -558,8 +571,6 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .server-row .icon-btn { width: 32px; height: 32px; } .drag-handle { color: var(--admin-muted); } .list-empty { color: var(--admin-muted); font-size: 13px; } -.form-notice { margin: 0 28px 14px; padding: 10px 12px; font-size: 13px; } -.admin-error { margin: 0 28px 14px; padding: 10px 12px; } .settings-form, .editor-modal { gap: 17px; } .form-grid { gap: 15px; } .settings-toggles { gap: 12px; } @@ -573,6 +584,8 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .data-stat span { color: var(--admin-muted); font-size: 12px; } .usage-table th, .usage-table td { height: 40px; color: var(--admin-text); font-size: 12px; } .usage-table thead th { color: var(--admin-muted); background: var(--admin-bg); } +.admin-page .usage-do-breakdown > div { border-color: var(--admin-border); color: var(--admin-muted); font-size: 12px; } +.admin-page .usage-do-breakdown strong { color: var(--admin-text); } .usage-empty { color: var(--admin-muted); font-size: 13px; } .admin-page .theme-store-card { border-color: var(--admin-border); background: var(--admin-surface); } .admin-page .theme-store-card.active { border-color: var(--admin-accent-border); box-shadow: 0 0 0 1px color-mix(in srgb, var(--admin-accent) 18%, transparent); } @@ -634,7 +647,7 @@ label > span { color: var(--muted-foreground); font-size: 9px; } .admin-content-header h1 { font-size: 20px; } .admin-content-header p { font-size: 12px; } .admin-section, .settings-form { margin: 0 12px 18px; padding: 17px 15px; } - .form-notice, .admin-error { margin: 0 12px 12px; } + .admin-toast { top: 70px; right: 16px; min-height: 58px; padding: 12px 14px; font-size: 13px; } .section-head { min-width: 0; align-items: flex-start; flex-wrap: wrap; } .server-row { grid-template-columns: 20px 20px 9px minmax(0, 1fr) auto; } .server-row .row-actions { grid-column: 4 / 6; } diff --git a/frontend/src/types.ts b/frontend/src/types.ts index fc7b0eb..090f043 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -109,6 +109,13 @@ export interface LatencySample { packet_loss: number; } +export interface LiveLatencyResult { + task_id: string; + timestamp: number; + latency_ms: number; + packet_loss: number; +} + export interface LatencyTestPoint { id: string; name: string; @@ -295,6 +302,15 @@ interface CloudflareUsagePeriod { rows_read: number; rows_written: number; workers_requests: number; + durable_objects_requests: number; + durable_objects_http_requests: number; + durable_objects_hibernation_wakeups: number; + durable_objects_inbound_websocket_messages: number; + durable_objects_outbound_websocket_messages: number; + durable_objects_raw_requests: number; + durable_objects_requests_estimated: boolean; + durable_objects_request_billing_ratio: number; + durable_objects_duration: number; } export interface CloudflareUsage { diff --git a/src/auth.rs b/src/auth.rs index 0bd3848..30807e4 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -9,7 +9,7 @@ use worker::{ Date, Env, Request, }; -const SESSION_SECONDS: i64 = 7 * 24 * 60 * 60; +pub const ADMIN_SESSION_SECONDS: i64 = 7 * 24 * 60 * 60; const TURNSTILE_PROOF_SECONDS: i64 = 60 * 60; const THEME_PREVIEW_SECONDS: i64 = 10 * 60; const SERVER_PASSWORD_ROUNDS: u32 = 10_000; @@ -137,7 +137,7 @@ fn create_admin_jwt_with_secret(secret: &str, issued_at: i64) -> Option serde_json::to_vec(&AdminClaims { sub: "admin".to_string(), iat: issued_at, - exp: issued_at + SESSION_SECONDS, + exp: issued_at + ADMIN_SESSION_SECONDS, }) .ok()?, ); @@ -177,7 +177,7 @@ fn verify_admin_jwt(token: &str, secret: &str, now: i64) -> bool { claims.sub == "admin" && claims.iat <= now + 60 && claims.exp >= now - && claims.exp == claims.iat + SESSION_SECONDS + && claims.exp == claims.iat + ADMIN_SESSION_SECONDS } pub fn create_admin_jwt(env: &Env, password_hash: &str) -> Option { @@ -189,14 +189,33 @@ pub fn is_admin(req: &Request, env: &Env, password_hash: &str) -> bool { let Some(secret) = signing_secret(env, password_hash) else { return false; }; - let Ok(Some(header)) = req.headers().get("Authorization") else { - return false; - }; - let Some(token) = header.strip_prefix("Bearer ") else { - return false; - }; let now = Date::now().as_millis() as i64 / 1000; - verify_admin_jwt(token, &secret, now) + let bearer_valid = req + .headers() + .get("Authorization") + .ok() + .flatten() + .is_some_and(|header| { + header + .strip_prefix("Bearer ") + .is_some_and(|token| verify_admin_jwt(token, &secret, now)) + }); + if bearer_valid { + return true; + } + req.headers() + .get("Cookie") + .ok() + .flatten() + .is_some_and(|cookies| { + cookies + .split(';') + .find_map(|entry| { + let (name, value) = entry.trim().split_once('=')?; + (name == "nodeflare_admin").then_some(value) + }) + .is_some_and(|token| verify_admin_jwt(token, &secret, now)) + }) } pub fn create_turnstile_proof(env: &Env, password_hash: &str) -> Option { @@ -273,7 +292,7 @@ pub fn bearer_token(req: &Request) -> Option { mod tests { use super::{ create_admin_jwt_with_secret, hash_password, verify_admin_jwt, verify_password_hash, - SESSION_SECONDS, + ADMIN_SESSION_SECONDS, }; #[test] @@ -303,12 +322,12 @@ mod tests { assert!(verify_admin_jwt( &token, "d1-password-hash", - issued_at + SESSION_SECONDS + issued_at + ADMIN_SESSION_SECONDS )); assert!(!verify_admin_jwt( &token, "d1-password-hash", - issued_at + SESSION_SECONDS + 1 + issued_at + ADMIN_SESSION_SECONDS + 1 )); assert!(!verify_admin_jwt(&token, "different-secret", issued_at)); } diff --git a/src/cloudflare.rs b/src/cloudflare.rs index 74ca9d0..8c1565f 100644 --- a/src/cloudflare.rs +++ b/src/cloudflare.rs @@ -2,6 +2,7 @@ use serde::{Deserialize, Serialize}; use worker::{wasm_bindgen::JsValue, Fetch, Method, Request, RequestInit, Result}; const GRAPHQL_URL: &str = "https://api.cloudflare.com/client/v4/graphql"; +const DURABLE_OBJECTS_WEBSOCKET_MESSAGE_BILLING_RATIO: i64 = 20; const USAGE_QUERY: &str = r#" query CloudflareUsage( $accountTag: string!, @@ -24,6 +25,19 @@ query CloudflareUsage( ) { sum { requests } } + durableObjectsInvocationsAdaptiveGroups( + limit: 10000, + filter: { date_geq: $start, date_leq: $end } + ) { + sum { requests } + dimensions { type } + } + durableObjectsPeriodicGroups( + limit: 10000, + filter: { date_geq: $start, date_leq: $end } + ) { + sum { duration inboundWebsocketMsgCount outboundWebsocketMsgCount } + } } } } @@ -34,6 +48,15 @@ pub struct UsagePeriod { pub rows_read: i64, pub rows_written: i64, pub workers_requests: i64, + pub durable_objects_requests: i64, + pub durable_objects_http_requests: i64, + pub durable_objects_hibernation_wakeups: i64, + pub durable_objects_inbound_websocket_messages: i64, + pub durable_objects_outbound_websocket_messages: i64, + pub durable_objects_raw_requests: i64, + pub durable_objects_requests_estimated: bool, + pub durable_objects_request_billing_ratio: i64, + pub durable_objects_duration: f64, } #[derive(Debug, Serialize)] @@ -61,8 +84,7 @@ struct GraphQlVariables<'a> { #[derive(Deserialize)] struct GraphQlResponse { data: Option, - #[serde(default)] - errors: Vec, + errors: Option>, } #[derive(Deserialize)] @@ -87,6 +109,10 @@ struct AccountUsage { d1_analytics_adaptive_groups: Vec, #[serde(default)] workers_invocations_adaptive: Vec, + #[serde(default)] + durable_objects_invocations_adaptive_groups: Vec, + #[serde(default)] + durable_objects_periodic_groups: Vec, } #[derive(Deserialize)] @@ -114,6 +140,104 @@ struct WorkersSum { requests: i64, } +#[derive(Deserialize)] +struct DurableObjectsInvocationGroup { + sum: DurableObjectsInvocationSum, + dimensions: DurableObjectsInvocationDimensions, +} + +#[derive(Deserialize)] +struct DurableObjectsInvocationSum { + #[serde(default)] + requests: i64, +} + +#[derive(Deserialize)] +struct DurableObjectsInvocationDimensions { + #[serde(rename = "type", default)] + invocation_type: String, +} + +#[derive(Deserialize)] +struct DurableObjectsPeriodicGroup { + sum: DurableObjectsPeriodicSum, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct DurableObjectsPeriodicSum { + #[serde(default)] + duration: f64, + #[serde(default)] + inbound_websocket_msg_count: i64, + #[serde(default)] + outbound_websocket_msg_count: i64, +} + +#[derive(Default)] +struct DurableObjectsSummary { + http_requests: i64, + hibernation_wakeups: i64, + inbound_websocket_messages: i64, + outbound_websocket_messages: i64, + raw_requests: i64, + billable_requests: i64, + duration: f64, +} + +fn is_hibernation_invocation_type(value: &str) -> bool { + let normalized = value + .chars() + .filter(|character| character.is_ascii_alphanumeric()) + .flat_map(char::to_lowercase) + .collect::(); + normalized.contains("hibernation") + || (normalized.contains("websocket") && normalized.contains("message")) +} + +fn ceil_div_nonnegative(value: i64, divisor: i64) -> i64 { + if value <= 0 { + 0 + } else { + value.saturating_add(divisor - 1) / divisor + } +} + +fn summarize_durable_objects( + invocation_groups: Vec, + periodic_groups: Vec, +) -> DurableObjectsSummary { + let mut summary = DurableObjectsSummary::default(); + for group in invocation_groups { + let requests = group.sum.requests.max(0); + summary.raw_requests = summary.raw_requests.saturating_add(requests); + if is_hibernation_invocation_type(&group.dimensions.invocation_type) { + summary.hibernation_wakeups = summary.hibernation_wakeups.saturating_add(requests); + } else { + summary.http_requests = summary.http_requests.saturating_add(requests); + } + } + for group in periodic_groups { + summary.inbound_websocket_messages = summary + .inbound_websocket_messages + .saturating_add(group.sum.inbound_websocket_msg_count.max(0)); + summary.outbound_websocket_messages = summary + .outbound_websocket_messages + .saturating_add(group.sum.outbound_websocket_msg_count.max(0)); + if group.sum.duration.is_finite() && group.sum.duration > 0.0 { + summary.duration += group.sum.duration; + } + } + summary.billable_requests = summary + .http_requests + .saturating_add(summary.hibernation_wakeups) + .saturating_add(ceil_div_nonnegative( + summary.inbound_websocket_messages, + DURABLE_OBJECTS_WEBSOCKET_MESSAGE_BILLING_RATIO, + )); + summary +} + pub fn date_from_unix_days(days: i64) -> String { let z = days + 719_468; let era = if z >= 0 { z } else { z - 146_096 } / 146_097; @@ -155,9 +279,9 @@ async fn query_period(token: &str, account_id: &str, date: &str) -> Result>() @@ -187,18 +311,33 @@ async fn query_period(token: &str, account_id: &str, date: &str) -> Result Result { let today = date_from_unix_days(now.div_euclid(86_400)); let yesterday = date_from_unix_days(now.div_euclid(86_400) - 1); - let today_usage = query_period(token, account_id, &today).await?; - let yesterday_usage = query_period(token, account_id, &yesterday).await?; + let (today_usage, yesterday_usage) = futures_util::try_join!( + query_period(token, account_id, &today), + query_period(token, account_id, &yesterday) + )?; Ok(CloudflareUsage { today: today_usage, yesterday: yesterday_usage, @@ -207,11 +346,67 @@ pub async fn usage(token: &str, account_id: &str, now: i64) -> Result i64 { } } -fn traffic_cycle_key(timestamp: i64, reset_day: i64) -> i64 { +pub(crate) fn traffic_cycle_key(timestamp: i64, reset_day: i64) -> i64 { let (year, month, day) = civil_date_from_unix_days(timestamp.div_euclid(86_400)); let boundary = reset_day.clamp(1, 31).min(days_in_month(year, month)); let current_month = year * 12 + month - 1; @@ -257,6 +274,38 @@ pub async fn get_agent_identity(db: &D1Database, token: &str) -> Result Result> { + let mut context: Option = db + .prepare( + r#"SELECT + s.report_interval, + s.collect_interval, + s.reset_day, + COALESCE(c.cycle_key, 0) AS cycle_key, + COALESCE(c.raw_rx, m.net_rx_total, 0) AS raw_rx, + COALESCE(c.raw_tx, m.net_tx_total, 0) AS raw_tx, + COALESCE(c.used_rx, m.net_rx_total, 0) AS used_rx, + COALESCE(c.used_tx, m.net_tx_total, 0) AS used_tx, + s.rx_correction, + s.tx_correction, + COALESCE(m.timestamp, 0) AS last_persisted_at + FROM servers s + LEFT JOIN latest_metrics m ON m.server_id = s.id + LEFT JOIN traffic_cycles c ON c.server_id = s.id + WHERE s.id = ?1 + LIMIT 1"#, + ) + .bind(&[text(id)])? + .first(None) + .await?; + if let Some(context) = context.as_mut() { + if context.cycle_key == 0 { + context.cycle_key = traffic_cycle_key(crate::now(), context.reset_day); + } + } + Ok(context) +} + pub async fn get_agent_token(db: &D1Database, id: &str) -> Result> { db.prepare("SELECT token FROM servers WHERE id = ?1 LIMIT 1") .bind(&[text(id)])? @@ -1122,7 +1171,11 @@ pub async fn update_settings( if value { "true" } else { "false" } ); } - if let Some(value) = input.turnstile_site_key.as_deref() { + if let Some(value) = input + .turnstile_site_key + .as_deref() + .filter(|value| value.trim() != SECRET_MASK) + { push_setting!("turnstile_site_key", value.trim()); } if let Some(value) = input @@ -1153,14 +1206,13 @@ pub async fn update_settings( let value = value.clamp(0, 365).to_string(); push_setting!("expiry_alert_days", &value); } - macro_rules! push_trimmed { - ($field:expr, $key:expr) => { - if let Some(value) = $field.as_deref() { - push_setting!($key, value.trim()); - } - }; + if let Some(value) = input + .cloudflare_account_id + .as_deref() + .filter(|value| value.trim() != SECRET_MASK) + { + push_setting!("cloudflare_account_id", value.trim()); } - push_trimmed!(input.cloudflare_account_id, "cloudflare_account_id"); if let Some(value) = input .cloudflare_api_token .as_deref() diff --git a/src/lib.rs b/src/lib.rs index a658510..37c0637 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,7 +19,7 @@ use worker::*; use crate::auth::{ bearer_token, create_admin_jwt, create_theme_preview_proof, create_turnstile_proof, hash_password, is_admin, random_salt, sha256_hex, verify_credentials, - verify_theme_preview_proof, verify_turnstile_proof, + verify_theme_preview_proof, verify_turnstile_proof, ADMIN_SESSION_SECONDS, }; use crate::models::{ AgentDiskMetric, AgentGpuMetric, AgentReport, AgentReportBatch, AlertRuleInput, ApiError, @@ -90,7 +90,7 @@ struct PublicServer { latency: Vec, } -fn now() -> i64 { +pub(crate) fn now() -> i64 { Date::now().as_millis() as i64 / 1000 } @@ -115,6 +115,20 @@ fn env_secret_text(env: &Env, key: &str) -> String { .unwrap_or_default() } +fn settings_for_admin_response(mut settings: db::SettingsView, env: &Env) -> db::SettingsView { + for (field, variable) in [ + (&mut settings.turnstile_site_key, "TURNSTILE_SITE_KEY"), + (&mut settings.turnstile_secret_key, "TURNSTILE_SECRET_KEY"), + (&mut settings.cloudflare_account_id, "CF_USAGE_ACCOUNT_ID"), + (&mut settings.cloudflare_api_token, "CF_USAGE_API_TOKEN"), + ] { + if field.trim().is_empty() && !env_secret_text(env, variable).trim().is_empty() { + *field = db::SECRET_MASK.to_string(); + } + } + settings +} + fn json(value: &T, status: u16) -> Result { let mut response = Response::from_json(value)?.with_status(status); set_api_headers(&mut response)?; @@ -139,6 +153,25 @@ fn no_content() -> Result { Ok(response) } +fn set_admin_session_cookie( + response: &mut Response, + req: &Request, + token: Option<&str>, +) -> Result<()> { + let secure = req.url().ok().is_some_and(|url| url.scheme() == "https"); + let (value, max_age) = token + .map(|value| (value, ADMIN_SESSION_SECONDS)) + .unwrap_or(("", 0)); + response.headers_mut().append( + "Set-Cookie", + &format!( + "nodeflare_admin={value}; Path=/; Max-Age={max_age}; HttpOnly{}; SameSite=Strict", + if secure { "; Secure" } else { "" } + ), + )?; + Ok(()) +} + async fn request_json(req: &mut Request, limit: usize) -> Result { if req .headers() @@ -427,7 +460,8 @@ fn public_probe_ipv4(address: Ipv4Addr) -> bool { fn valid_cloudflare_account_id(value: &str) -> bool { let value = value.trim(); - value.is_empty() + value == db::SECRET_MASK + || value.is_empty() || (value.len() == 32 && value.chars().all(|character| character.is_ascii_hexdigit())) } @@ -597,7 +631,7 @@ fn validate_server(input: &ServerInput) -> Option<&'static str> { None } -fn validate_report(report: &AgentReport) -> Option<&'static str> { +pub(crate) fn validate_report(report: &AgentReport) -> Option<&'static str> { let floats = [ report.cpu, report.load1, @@ -783,7 +817,9 @@ fn client_ip(req: &Request) -> Option { fn rate_limit_binding(method: Method, path: &str) -> Option<&'static str> { if method == Method::Post && matches!(path, "/api/admin/login" | "/api/turnstile/verify") { Some("AUTH_RATE_LIMITER") - } else if method == Method::Post && path == "/api/agent/report" { + } else if (method == Method::Post && path == "/api/agent/report") + || (method == Method::Get && path == "/api/agent/live") + { Some("AGENT_RATE_LIMITER") } else if path.starts_with("/api/") { Some("API_RATE_LIMITER") @@ -991,6 +1027,9 @@ async fn handle_agent_report( response .headers_mut() .set("X-Agent-Config-Sha256", &config_hash)?; + response + .headers_mut() + .set("X-NodeFlare-Server-Time", &received_at.to_string())?; response.headers_mut().set("Cache-Control", "no-store")?; return Ok(response); } @@ -998,6 +1037,9 @@ async fn handle_agent_report( response .headers_mut() .set("X-Agent-Config-Sha256", &config_hash)?; + response + .headers_mut() + .set("X-NodeFlare-Server-Time", &received_at.to_string())?; Ok(response) } @@ -1028,6 +1070,19 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { if method == Method::Post && path == "/api/agent/report" { return handle_agent_report(req, env, ctx, &database).await; } + if method == Method::Get && path == "/api/agent/live" { + let token = match bearer_token(&req) { + Some(value) => value, + None => return error("缺少探针凭据", 401), + }; + let Some(identity) = db::get_agent_identity(&database, &token).await? else { + return error("探针凭据无效", 401); + }; + let Some(context) = db::agent_live_context(&database, &identity.id).await? else { + return error("节点不存在", 404); + }; + return live::upgrade_agent(req, &env, &identity.id, identity.hidden != 0, context).await; + } let default_name = env_text(&env, "SITE_NAME", "NodeFlare"); let default_threshold = env_number(&env, "OFFLINE_THRESHOLD_SECONDS", 180).clamp(30, 3600); @@ -1059,7 +1114,7 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { let login_protection_enabled = settings.turnstile_login_enabled && turnstile_configured; if method == Method::Get && path == "/api/config" { - return json( + let mut response = json( &serde_json::json!({ "site_name": settings.site_name, "site_description": settings.site_description, @@ -1089,7 +1144,11 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { "password_client_salt": settings.password_client_salt }), 200, - ); + )?; + response + .headers_mut() + .set("X-NodeFlare-Server-Time", &now().to_string())?; + return Ok(response); } if method == Method::Post && path == "/api/admin/login" { @@ -1145,7 +1204,9 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { let Some(token) = create_admin_jwt(&env, &session_password_hash) else { return error("会话密钥未配置", 503); }; - return json(&serde_json::json!({ "token": token }), 200); + let mut response = json(&serde_json::json!({ "token": token }), 200)?; + set_admin_session_cookie(&mut response, &req, Some(&token))?; + return Ok(response); } if method == Method::Post && path == "/api/turnstile/verify" { @@ -1188,6 +1249,12 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { verify_turnstile_proof(&value, &env, &settings.admin_password_hash) }); + if method == Method::Post && path == "/api/admin/logout" { + let mut response = no_content()?; + set_admin_session_cookie(&mut response, &req, None)?; + return Ok(response); + } + if method == Method::Get && path == "/api/exchange-rates" { if !turnstile_verified { return error("请先完成 Cloudflare 人机验证", 403); @@ -1497,6 +1564,9 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { return error("批量删除列表无效", 400); } db::delete_servers(&database, &input.ids).await?; + if let Err(error) = live::disconnect_agents(&env, &input.ids).await { + console_error!("failed to disconnect deleted Agents: {error}"); + } return no_content(); } @@ -1561,11 +1631,13 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { return error("节点 ID 无效", 400); }; if method == Method::Delete { - return if db::delete_server(&database, &id).await? { - no_content() - } else { - error("节点不存在", 404) - }; + if !db::delete_server(&database, &id).await? { + return error("节点不存在", 404); + } + if let Err(error) = live::disconnect_agents(&env, std::slice::from_ref(&id)).await { + console_error!("failed to disconnect deleted Agent: {error}"); + } + return no_content(); } let input: ServerInput = match request_json(&mut req, API_JSON_MAX_BYTES).await { Ok(value) => value, @@ -1574,15 +1646,17 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { if let Some(message) = validate_server(&input) { return error(message, 400); } - return if db::update_server(&database, &id, &input).await? { - no_content() - } else { - error("节点不存在", 404) - }; + if !db::update_server(&database, &id, &input).await? { + return error("节点不存在", 404); + } + if let Err(error) = live::disconnect_agents(&env, std::slice::from_ref(&id)).await { + console_error!("failed to reconnect updated Agent: {error}"); + } + return no_content(); } if method == Method::Get && path == "/api/admin/settings" { - return json(&settings, 200); + return json(&settings_for_admin_response(settings.clone(), &env), 200); } if method == Method::Get && path == "/api/admin/themes" { @@ -1759,7 +1833,8 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { Ok(usage) => json(&usage, 200), Err(err) => { console_error!("cloudflare usage query failed: {err}"); - error("Cloudflare 用量查询失败,请检查 Token 权限和账户 ID", 502) + let detail = err.to_string().chars().take(240).collect::(); + error(&format!("Cloudflare 用量查询失败:{detail}"), 502) } }; } @@ -1874,11 +1949,10 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { { return error("新密码派生值无效", 400); } - let configured_site_key = input - .turnstile_site_key - .as_deref() - .unwrap_or(&settings.turnstile_site_key) - .trim(); + let configured_site_key = submitted_secret( + input.turnstile_site_key.as_deref(), + &settings.turnstile_site_key, + ); let configured_secret_key = submitted_secret( input.turnstile_secret_key.as_deref(), &settings.turnstile_secret_key, @@ -1970,10 +2044,15 @@ async fn handle(mut req: Request, env: Env, ctx: Context) -> Result { } else { None }; - return json( - &serde_json::json!({ "settings": updated, "token": token }), + let response_settings = settings_for_admin_response(updated, &env); + let mut response = json( + &serde_json::json!({ "settings": response_settings, "token": token }), 200, - ); + )?; + if let Some(token) = token.as_deref() { + set_admin_session_cookie(&mut response, &req, Some(token))?; + } + return Ok(response); } if path.starts_with("/api/") { @@ -2121,6 +2200,10 @@ mod tests { rate_limit_binding(Method::Post, "/api/agent/report"), Some("AGENT_RATE_LIMITER") ); + assert_eq!( + rate_limit_binding(Method::Get, "/api/agent/live"), + Some("AGENT_RATE_LIMITER") + ); assert_eq!( rate_limit_binding(Method::Get, "/api/history/node-a"), Some("API_RATE_LIMITER") @@ -2219,6 +2302,7 @@ mod tests { #[test] fn validates_cloudflare_usage_credentials() { assert!(valid_cloudflare_account_id("")); + assert!(valid_cloudflare_account_id(SECRET_MASK)); assert!(valid_cloudflare_account_id( "0123456789abcdef0123456789ABCDEF" )); diff --git a/src/live.rs b/src/live.rs index 8d2f7c9..44e97da 100644 --- a/src/live.rs +++ b/src/live.rs @@ -1,20 +1,207 @@ +use serde::{Deserialize, Serialize}; use worker::*; +use crate::db::AgentLiveContext; +use crate::models::{AgentLatencyResult, AgentReport}; + +const MAX_LIVE_SAMPLES: usize = 720; +const MAX_LIVE_LATENCY_RESULTS: usize = 4096; + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct AgentLiveBatch { + #[serde(rename = "type")] + message_type: String, + samples: Vec, +} + +#[derive(Debug, Deserialize, Serialize)] +struct TrafficAttachment { + reset_day: i64, + cycle_key: i64, + raw_rx: i64, + raw_tx: i64, + used_rx: i64, + used_tx: i64, + rx_correction: i64, + tx_correction: i64, +} + +#[derive(Debug, Deserialize, Serialize)] +struct SocketAttachment { + role: String, + server_id: Option, + hidden: bool, + report_interval: i64, + collect_interval: i64, + last_d1_write_at: i64, + traffic: Option, +} + +fn active_report_interval_ms(report_interval: i64, collect_interval: i64) -> i64 { + let report_interval = report_interval.clamp(15, 3600); + let collect_interval = collect_interval.clamp(2, 60).min(report_interval); + ((report_interval + 14) / 15) + .max(collect_interval) + .clamp(2, 60) + * 1000 +} + +fn apply_live_traffic(report: &mut AgentReport, traffic: &mut TrafficAttachment) { + let cycle_key = crate::db::traffic_cycle_key(report.timestamp, traffic.reset_day); + if cycle_key != traffic.cycle_key { + traffic.cycle_key = cycle_key; + traffic.raw_rx = report.net_rx_total; + traffic.raw_tx = report.net_tx_total; + traffic.used_rx = 0; + traffic.used_tx = 0; + } else { + let rx_delta = if report.net_rx_total >= traffic.raw_rx { + report.net_rx_total - traffic.raw_rx + } else { + report.net_rx_total + }; + let tx_delta = if report.net_tx_total >= traffic.raw_tx { + report.net_tx_total - traffic.raw_tx + } else { + report.net_tx_total + }; + traffic.used_rx = traffic.used_rx.saturating_add(rx_delta); + traffic.used_tx = traffic.used_tx.saturating_add(tx_delta); + traffic.raw_rx = report.net_rx_total; + traffic.raw_tx = report.net_tx_total; + } + report.net_rx_total = traffic.used_rx.saturating_add(traffic.rx_correction); + report.net_tx_total = traffic.used_tx.saturating_add(traffic.tx_correction); +} + +fn batch_update_payload( + server_id: &str, + reports: &[AgentReport], + traffic: &mut TrafficAttachment, +) -> Result { + batch_update_payload_at(server_id, reports, traffic, crate::now()) +} + +fn batch_update_payload_at( + server_id: &str, + reports: &[AgentReport], + traffic: &mut TrafficAttachment, + envelope_timestamp: i64, +) -> Result { + let mut reports = reports.to_vec(); + reports.sort_by_key(|report| report.timestamp); + let mut samples = Vec::with_capacity(reports.len()); + for mut report in reports { + apply_live_traffic(&mut report, traffic); + let timestamp = report.timestamp; + let mut data = serde_json::to_value(report)?; + if let Some(data) = data.as_object_mut() { + data.remove("timestamp"); + } + samples.push(serde_json::json!({ "ts": timestamp, "data": data })); + } + Ok(serde_json::to_string(&serde_json::json!({ + "type": "batchUpdate", + "ts": envelope_timestamp, + "updates": [{ + "serverId": server_id, + "samples": samples + }] + }))?) +} + +fn agent_attachment( + server_id: String, + hidden: bool, + context: AgentLiveContext, +) -> SocketAttachment { + SocketAttachment { + role: "agent".to_string(), + server_id: Some(server_id), + hidden, + report_interval: context.report_interval.clamp(15, 3600), + collect_interval: context.collect_interval.clamp(2, 60), + last_d1_write_at: context.last_persisted_at, + traffic: Some(TrafficAttachment { + reset_day: context.reset_day.clamp(1, 31), + cycle_key: context.cycle_key, + raw_rx: context.raw_rx, + raw_tx: context.raw_tx, + used_rx: context.used_rx, + used_tx: context.used_tx, + rx_correction: context.rx_correction, + tx_correction: context.tx_correction, + }), + } +} + +fn required_i64_header(req: &Request, name: &str) -> Result { + req.headers() + .get(name)? + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| Error::RustError(format!("missing or invalid {name}"))) +} + +fn live_context_from_headers(req: &Request) -> Result { + Ok(AgentLiveContext { + report_interval: required_i64_header(req, "X-Live-Report-Interval")?, + collect_interval: required_i64_header(req, "X-Live-Collect-Interval")?, + reset_day: required_i64_header(req, "X-Live-Reset-Day")?, + cycle_key: required_i64_header(req, "X-Live-Cycle-Key")?, + raw_rx: required_i64_header(req, "X-Live-Raw-Rx")?, + raw_tx: required_i64_header(req, "X-Live-Raw-Tx")?, + used_rx: required_i64_header(req, "X-Live-Used-Rx")?, + used_tx: required_i64_header(req, "X-Live-Used-Tx")?, + rx_correction: required_i64_header(req, "X-Live-Rx-Correction")?, + tx_correction: required_i64_header(req, "X-Live-Tx-Correction")?, + last_persisted_at: required_i64_header(req, "X-Live-Last-Persisted-At")?, + }) +} + +fn dashboard_attachment() -> SocketAttachment { + SocketAttachment { + role: "dashboard".to_string(), + server_id: None, + hidden: false, + report_interval: 0, + collect_interval: 0, + last_d1_write_at: 0, + traffic: None, + } +} + #[durable_object(websocket)] pub struct LiveHub { state: State, + env: Env, } impl DurableObject for LiveHub { - fn new(state: State, _env: Env) -> Self { + fn new(state: State, env: Env) -> Self { if let Ok(pair) = WebSocketRequestResponsePair::new("ping", "pong") { state.set_websocket_auto_response(&pair); } - Self { state } + Self { state, env } } async fn fetch(&self, mut req: Request) -> Result { if req.method() == Method::Post { + if req.headers().get("X-Live-Action")?.as_deref() == Some("disconnect-agents") { + let server_ids = req.json::>().await?; + for server_id in server_ids { + if server_id.is_empty() || server_id.len() > 80 || server_id.contains('/') { + continue; + } + for socket in self + .state + .get_websockets_with_tag(&format!("agent:{server_id}")) + { + let _ = socket.close(Some(1008), Some("server configuration changed")); + } + } + return Response::empty(); + } let payload = req.text().await?; let server_id = req.headers().get("X-Server-ID")?.unwrap_or_default(); let mut sockets = self.state.get_websockets_with_tag("all"); @@ -42,21 +229,154 @@ impl DurableObject for LiveHub { } let pair = WebSocketPair::new()?; + if let Some(server_id) = req.headers().get("X-Live-Agent-ID")? { + if server_id.is_empty() || server_id.len() > 80 || server_id.contains('/') { + return Response::error("Invalid live Agent identity", 400); + } + let hidden = req + .headers() + .get("X-Live-Agent-Hidden")? + .is_some_and(|value| value == "1"); + let context = match live_context_from_headers(&req) { + Ok(context) => context, + Err(_) => return Response::error("Invalid live Agent context", 400), + }; + pair.server.serialize_attachment(agent_attachment( + server_id.clone(), + hidden, + context, + ))?; + self.state + .accept_websocket_with_tags(&pair.server, &[&format!("agent:{server_id}")]); + return Response::from_websocket(pair.client); + } let server_id = req .url()? .query_pairs() .find_map(|(key, value)| (key == "server_id").then(|| value.to_string())) .filter(|value| !value.is_empty() && value.len() <= 80 && !value.contains('/')); let tag = server_id.map_or_else(|| "all".to_string(), |id| format!("server:{id}")); + pair.server.serialize_attachment(dashboard_attachment())?; self.state.accept_websocket_with_tags(&pair.server, &[&tag]); Response::from_websocket(pair.client) } async fn websocket_message( &self, - _ws: WebSocket, - _message: WebSocketIncomingMessage, + ws: WebSocket, + message: WebSocketIncomingMessage, ) -> Result<()> { + let Some(mut attachment) = ws.deserialize_attachment::()? else { + return Ok(()); + }; + if attachment.role != "agent" { + return Ok(()); + } + let WebSocketIncomingMessage::String(message) = message else { + return Ok(()); + }; + if message.len() > 1024 * 1024 { + return ws.close(Some(1009), Some("live metric payload too large")); + } + let Ok(batch) = serde_json::from_str::(&message) else { + return ws.close(Some(1007), Some("invalid live metric payload")); + }; + if batch.message_type != "update" + || batch.samples.is_empty() + || batch.samples.len() > MAX_LIVE_SAMPLES + { + return ws.close(Some(1007), Some("invalid live metric batch")); + } + let received_at = crate::now(); + let mut latency_results = Vec::::new(); + for report in &batch.samples { + if report.timestamp <= 0 + || (report.timestamp - received_at).abs() > 7200 + || crate::validate_report(report).is_some() + { + return ws.close(Some(1007), Some("invalid live metric sample")); + } + latency_results.extend(report.latency_results.iter().cloned()); + } + if latency_results.len() > MAX_LIVE_LATENCY_RESULTS { + return ws.close(Some(1009), Some("too many live latency results")); + } + let Some(server_id) = attachment.server_id.clone() else { + return ws.close(Some(1008), Some("missing server identity")); + }; + let mut traffic = attachment + .traffic + .take() + .ok_or_else(|| Error::RustError("missing live traffic state".to_string()))?; + let payload = batch_update_payload(&server_id, &batch.samples, &mut traffic)?; + attachment.traffic = Some(traffic); + + let report_interval = attachment.report_interval.clamp(15, 3600); + let due_for_d1 = received_at.saturating_sub(attachment.last_d1_write_at) >= report_interval; + let mut persisted = false; + if due_for_d1 { + let database = self.env.d1("DB")?; + if let Err(error) = crate::db::save_reports(&database, &server_id, &batch.samples).await + { + console_warn!("live D1 metrics write failed for {server_id}: {error:?}"); + } else if let Err(error) = + crate::latency::save_results(&database, &server_id, &latency_results, received_at) + .await + { + console_warn!("live D1 latency write failed for {server_id}: {error:?}"); + } else { + attachment.last_d1_write_at = received_at; + persisted = true; + } + } + + if !attachment.hidden { + let mut sockets = self.state.get_websockets_with_tag("all"); + sockets.extend( + self.state + .get_websockets_with_tag(&format!("server:{server_id}")), + ); + let has_dashboard = !sockets.is_empty(); + for dashboard in sockets { + if dashboard.send_with_str(&payload).is_err() { + let _ = dashboard.close(Some(1011), Some("send failed")); + } + } + let next_wss_ms = if has_dashboard { + active_report_interval_ms(attachment.report_interval, attachment.collect_interval) + } else { + report_interval * 1000 + }; + let next_d1_ms = if persisted { + report_interval * 1000 + } else { + let due_at = attachment.last_d1_write_at.saturating_add(report_interval); + due_at.saturating_sub(received_at).saturating_mul(1000) + }; + ws.serialize_attachment(attachment)?; + ws.send_with_str(&serde_json::to_string(&serde_json::json!({ + "type": "ack", + "ts": received_at, + "persisted": persisted, + "nextD1WriteAfterMs": next_d1_ms, + "nextWssReportAfterMs": next_wss_ms + }))?)?; + } else { + let next_d1_ms = if persisted { + report_interval * 1000 + } else { + let due_at = attachment.last_d1_write_at.saturating_add(report_interval); + due_at.saturating_sub(received_at).saturating_mul(1000) + }; + ws.serialize_attachment(attachment)?; + ws.send_with_str(&serde_json::to_string(&serde_json::json!({ + "type": "ack", + "ts": received_at, + "persisted": persisted, + "nextD1WriteAfterMs": next_d1_ms, + "nextWssReportAfterMs": report_interval * 1000 + }))?)?; + } Ok(()) } @@ -87,8 +407,162 @@ pub async fn broadcast(env: &Env, server_id: &str, payload: &str) -> Result<()> Ok(()) } +pub async fn disconnect_agents(env: &Env, server_ids: &[String]) -> Result<()> { + if server_ids.is_empty() { + return Ok(()); + } + let namespace = env.durable_object("LIVE_HUB")?; + let stub = namespace.id_from_name("dashboard")?.get_stub()?; + let mut init = RequestInit::new(); + init.with_method(Method::Post) + .with_body(Some(worker::wasm_bindgen::JsValue::from_str( + &serde_json::to_string(server_ids)?, + ))); + let req = Request::new_with_init("https://live.internal/disconnect-agents", &init)?; + req.headers().set("Content-Type", "application/json")?; + req.headers().set("X-Live-Action", "disconnect-agents")?; + stub.fetch_with_request(req).await?; + Ok(()) +} + pub async fn upgrade(req: Request, env: &Env) -> Result { let namespace = env.durable_object("LIVE_HUB")?; let stub = namespace.id_from_name("dashboard")?.get_stub()?; + req.headers().delete("X-Live-Agent-ID")?; + req.headers().delete("X-Live-Agent-Hidden")?; + for header in [ + "X-Live-Report-Interval", + "X-Live-Collect-Interval", + "X-Live-Reset-Day", + "X-Live-Cycle-Key", + "X-Live-Raw-Rx", + "X-Live-Raw-Tx", + "X-Live-Used-Rx", + "X-Live-Used-Tx", + "X-Live-Rx-Correction", + "X-Live-Tx-Correction", + "X-Live-Last-Persisted-At", + ] { + req.headers().delete(header)?; + } stub.fetch_with_request(req).await } + +pub async fn upgrade_agent( + req: Request, + env: &Env, + server_id: &str, + hidden: bool, + context: AgentLiveContext, +) -> Result { + let namespace = env.durable_object("LIVE_HUB")?; + let stub = namespace.id_from_name("dashboard")?.get_stub()?; + req.headers().set("X-Live-Agent-ID", server_id)?; + req.headers() + .set("X-Live-Agent-Hidden", if hidden { "1" } else { "0" })?; + req.headers().set( + "X-Live-Report-Interval", + &context.report_interval.to_string(), + )?; + req.headers().set( + "X-Live-Collect-Interval", + &context.collect_interval.to_string(), + )?; + req.headers() + .set("X-Live-Reset-Day", &context.reset_day.to_string())?; + req.headers() + .set("X-Live-Cycle-Key", &context.cycle_key.to_string())?; + req.headers() + .set("X-Live-Raw-Rx", &context.raw_rx.to_string())?; + req.headers() + .set("X-Live-Raw-Tx", &context.raw_tx.to_string())?; + req.headers() + .set("X-Live-Used-Rx", &context.used_rx.to_string())?; + req.headers() + .set("X-Live-Used-Tx", &context.used_tx.to_string())?; + req.headers() + .set("X-Live-Rx-Correction", &context.rx_correction.to_string())?; + req.headers() + .set("X-Live-Tx-Correction", &context.tx_correction.to_string())?; + req.headers().set( + "X-Live-Last-Persisted-At", + &context.last_persisted_at.to_string(), + )?; + req.headers().delete("Authorization")?; + stub.fetch_with_request(req).await +} + +#[cfg(test)] +mod tests { + use super::{active_report_interval_ms, batch_update_payload_at, TrafficAttachment}; + use crate::models::AgentReport; + + #[test] + fn live_payload_contains_cycle_adjusted_traffic_and_latency() { + let report = AgentReport { + timestamp: 1_234, + cpu: 12.5, + load1: 0.5, + load5: 0.4, + load15: 0.3, + mem_used: 1, + mem_total: 2, + swap_used: 0, + swap_total: 0, + disk_used: 3, + disk_total: 4, + net_in: 100.0, + net_out: 200.0, + net_rx_total: 300, + net_tx_total: 400, + uptime: 500, + processes: 10, + tcp_connections: 2, + udp_connections: 1, + cpu_cores: 2, + cpu_model: "CPU".to_string(), + os: "Linux".to_string(), + kernel: "6.0".to_string(), + arch: "x86_64".to_string(), + virtualization: "kvm".to_string(), + gpu_usage: 0.0, + gpu_model: String::new(), + agent_version: "test".to_string(), + disk_read_bps: 0.0, + disk_write_bps: 0.0, + disk_read_iops: 0.0, + disk_write_iops: 0.0, + disk_await_ms: 0.0, + disk_utilization: 0.0, + disks: vec![], + gpus: vec![], + latency_results: vec![], + }; + let mut traffic = TrafficAttachment { + reset_day: 1, + cycle_key: crate::db::traffic_cycle_key(1_234, 1), + raw_rx: 100, + raw_tx: 100, + used_rx: 500, + used_tx: 700, + rx_correction: 10, + tx_correction: 20, + }; + let payload: serde_json::Value = serde_json::from_str( + &batch_update_payload_at("node-a", &[report], &mut traffic, 1_235).unwrap(), + ) + .unwrap(); + assert_eq!(payload["type"], "batchUpdate"); + assert_eq!(payload["updates"][0]["serverId"], "node-a"); + assert_eq!(payload["updates"][0]["samples"][0]["ts"], 1_234); + assert_eq!(payload["updates"][0]["samples"][0]["data"]["cpu"], 12.5); + assert_eq!( + payload["updates"][0]["samples"][0]["data"]["net_rx_total"], + 710 + ); + assert!(payload["updates"][0]["samples"][0]["data"] + .get("timestamp") + .is_none()); + assert_eq!(active_report_interval_ms(60, 5), 5_000); + } +}