diff --git a/README.md b/README.md
index ea36bce..18b101e 100644
--- a/README.md
+++ b/README.md
@@ -35,6 +35,7 @@ Go 实现，编译为**单个静态二进制**，无运行时依赖；内置 **W
- **三站点聚合**：单站点每日额度耗尽自动切换到其他站点的同款模型，额度耗尽的站点临时降级 10 分钟
- **站点额度管理**：额度按访客身份按日计算，控制台站点卡片可一键查询当前额度、重置访客身份（等同新开隐私窗口、额度即刻回满）；请求撞额度时也会自动轮换访客身份恢复，无需重新求解
- **代理线路**：主线路 + 多条备用线路，支持 `http://` `https://` `socks5://`；slow-start 检测 + 失败降级 + 流中失败不重放
+- **代理池**：Xray 核心（分享链接 / 订阅 / `ip:port` 端点；核心可自动下载）+ 粘性选路 + 后台健康探测；每个站点可绑定独立出口，Turnstile 求解与请求共用同一出口
- **Turnstile 自动突破**：三种 Token 获取方式 — 求解服务 API（`provider="api"`）、本地浏览器自动求解（`provider="browser"`，CDP 直连 Chrome/Edge，无需 Playwright/Docker）、手动导入 Cookie（`provider="manual"`）；结果按「代理线路 × 站点」缓存，TTL 过期自动重解
- **会话自愈**：nonce / data-config 缓存 + 配额耗尽 / 验证失败时自动刷新重试
- **Web 管理台**：仪表盘、模型与站点、API Key、代理线路、Turnstile、对话调试、实时日志、设置
@@ -139,9 +140,10 @@ dsfree2api -config config.toml -check
| `[admin]` | 管理台 `enabled` / `host` / `port` / `password`（留空启动时生成随机密码） |
| `[limits]` | `max_concurrent_per_site` 每站点并发、`rate_per_minute` 每 Key 每分钟限流（0 = 不限） |
| `[proxy]` | `url` 主线路、`fallback_urls` 备用线路、`slow_start_seconds` 首事件超时 |
+| `[proxypool]` | 代理池总开关、健康检查间隔 / 超时 / URL、裸 `ip:port` 默认协议、Xray 路径 / 版本 / 自动下载；`[proxypool.entries.*]` 手动节点（分享链接或端点）、`[proxypool.subscriptions.*]` 订阅自动拉取 |
| `[upstream]` | 超时、配置缓存 TTL、会话自动刷新、`cross_site_failover` 跨站切换、`continue_rounds` 工具调用截断续写轮数（0 = 不续写、直接截断补全） |
| `[turnstile]` | **默认 `enabled = false` 且不内置任何求解服务**；`provider` 三选一：`api`（填 `api_url` / `api_key`）、`browser`（填 `browser_path` 指向本机 Chrome/Edge）、`manual`（只用控制台导入的 Cookie），另有 Cookie TTL、重试次数 |
-| `[sites.*]` | 三个站点的 `base_url` / `ajax_url` / `sitekey` / `language` |
+| `[sites.*]` | 三个站点的 `base_url` / `ajax_url` / `sitekey` / `language` / `proxies`（站点级出口绑定，按顺序粘性选路） |
| `[models.*]` | 模型到 `site` + `upstream_id` + `bot_id` / `post_id` 的映射 |
### Turnstile 说明
@@ -265,7 +267,7 @@ curl http://127.0.0.1:8000/v1/messages \
| 仪表盘 | 请求数 / 错误 / Token / P50·P95、模型用量、站点状态（含 Cookie TTL 进度条）、最近请求 |
| 模型与站点 | 编辑标签、路径、bot_id/post_id，启停模型与站点，修改站点 URL / 语言 |
| API Key | 生成 / 删除 / 复制下游 key，立即写回配置 |
-| 代理线路 | 主线路、备用线路增删、连通性测试、slow-start / 并发 / 限流 / 跨站切换 |
+| 代理线路 | 主线路、备用线路增删、连通性测试、slow-start / 并发 / 限流 / 跨站切换、代理池（节点 / 订阅 / 每站点绑定 / Xray 核心状态） |
| Turnstile | 三种 Token 获取方式的模式面板（求解服务 API / 本地浏览器自动获取 / 手动导入 Cookie）、各站 Cookie 状态与 TTL、强制刷新、求解历史 |
| 对话调试 | 选模型 / 系统提示 / tools JSON，流式或非流式，带连接计时与取消 |
| 实时日志 | 级别过滤 + SSE 实时推送 |
@@ -302,6 +304,7 @@ curl http://127.0.0.1:8000/v1/messages \
cmd/dsfree2api 入口：配置加载、日志、双 HTTP server、优雅退出
internal/config TOML 结构 + 默认值 + env 覆盖 + 校验 + 写回
internal/httpx TLS 指纹会话封装（bogdanfinn/tls-client, Chrome_120）
+internal/proxypool 代理池：Xray 核心管理 / 订阅拉取 / 分享链接解析 / 粘性选路与健康探测
internal/openai OpenAI 协议类型、prompt 拼接、tool_calls 解析
internal/anthropic Anthropic Messages 协议类型 + 与 OpenAI 消息的互转
internal/turnstile Turnstile 三种求解方式（API / 浏览器 CDP / 手动）+ 按线路/站点的 Cookie 缓存
diff --git a/cmd/dsfree2api/main.go b/cmd/dsfree2api/main.go
index 127a38e..be9b393 100644
--- a/cmd/dsfree2api/main.go
+++ b/cmd/dsfree2api/main.go
@@ -18,6 +18,7 @@ import (
"github.com/nyoungo/dsfree2api/internal/config"
"github.com/nyoungo/dsfree2api/internal/logbuf"
"github.com/nyoungo/dsfree2api/internal/metrics"
+ "github.com/nyoungo/dsfree2api/internal/proxypool"
"github.com/nyoungo/dsfree2api/internal/turnstile"
"github.com/nyoungo/dsfree2api/internal/upstream"
)
@@ -75,9 +76,18 @@ func main() {
defer met.Close()
ts := turnstile.New(cfg.Turnstile, cfg.Upstream.UserAgent)
- up := upstream.New(cfg, ts, logger)
+
+ // Proxy pool: per-site egress endpoints (Xray share links, subscriptions,
+ // plain http/socks5) with sticky selection. No-op unless enabled.
+ pool := proxypool.New(cfg, dataDir, logger)
+ poolCtx, poolCancel := context.WithCancel(context.Background())
+ defer poolCancel()
+ pool.Start(poolCtx)
+
+ up := upstream.New(cfg, ts, logger, pool)
+
apiSrv := api.New(cfg, up, met, logs, ts, logger)
- adminSrv := admin.New(cfg, up, met, logs, ts, logger)
+ adminSrv := admin.New(cfg, up, met, logs, ts, logger, pool)
mainAddr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.Port)
mainHTTP := &http.Server{
diff --git a/config.example.toml b/config.example.toml
index f9a12e2..df7bf61 100644
--- a/config.example.toml
+++ b/config.example.toml
@@ -33,6 +33,37 @@ fallback_urls = []
# 主线路允许的首事件等待秒数（>0 且存在备用线路时生效）。
slow_start_seconds = 8
+[proxypool]
+# 代理池：为每个站点绑定独立出口（Xray 分享链接 / 订阅 / ip:port 端点）。
+# 选路为粘性：优先使用绑定列表中第一个健康节点，失败才切换；未绑定或
+# 全部不健康时回退到上面的 [proxy] 主/备线路。Turnstile 求解与请求共用
+# 同一出口（Cookie 与出口 IP 绑定，换出口后会自动重新求解）。
+enabled = false
+check_interval_seconds = 120 # 健康检查间隔
+check_timeout_seconds = 10 # 单次探测超时
+check_url = "https://www.gstatic.com/generate_204"
+default_scheme = "http" # 裸 ip:port 的默认协议（http / socks5）
+# Xray 核心：留空 xray_path 且开启自动下载时，首次使用会下载到数据目录
+# （data/xray.exe；核心配置写到 data/config.json）。分享链接（vless / vmess
+# / trojan / ss）由它承载；纯 ip:port / socks5:// 端点无需核心。
+xray_path = ""
+xray_version = "" # 留空 = 最新版
+xray_auto_download = true
+
+# 手动节点（名称限字母 / 数字 / - / _）：
+# [proxypool.entries.node-a]
+# link = "vless://uuid@host:443?type=ws&security=tls&sni=example.com&path=/ws#node-a"
+# link = "socks5://user:pass@1.2.3.4:1080"
+# link = "1.2.3.4:8080" # 使用 default_scheme
+# link = "1.2.3.4:8080:user:pass"
+# enabled = true # 缺省即启用
+
+# 订阅（按 interval_minutes 自动刷新；节点名形如 <订阅名>#xxxxxxxx）：
+# [proxypool.subscriptions.home]
+# url = "https://example.com/api/v1/client/subscribe?token=…"
+# interval_minutes = 60
+# enabled = true
+
[upstream]
timeout = 60.0
stream_timeout = 300.0
@@ -80,6 +111,9 @@ base_url = "https://deepseek.de"
ajax_url = "https://deepseek.de/wp-admin/admin-ajax.php"
sitekey = "0x4AAAAAADlLZ3ljqZP6cQwq"
language = "de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7"
+# 站点级出口绑定（代理池条目名或 "sub:订阅名"，按顺序粘性选路）；
+# 留空 = 使用 [proxy] 主/备线路。三个站点可各自独立配置。
+# proxies = ["node-a", "sub:home"]
[sites.es]
base_url = "https://deepseek.es"
diff --git a/internal/admin/actions.go b/internal/admin/actions.go
index ca4ac99..199770b 100644
--- a/internal/admin/actions.go
+++ b/internal/admin/actions.go
@@ -9,12 +9,14 @@ import (
"fmt"
"io"
"net/http"
+ "regexp"
"strings"
"time"
"github.com/nyoungo/dsfree2api/internal/config"
"github.com/nyoungo/dsfree2api/internal/httpx"
"github.com/nyoungo/dsfree2api/internal/openai"
+ "github.com/nyoungo/dsfree2api/internal/proxypool"
"github.com/nyoungo/dsfree2api/internal/upstream"
)
@@ -396,6 +398,218 @@ func (s *Server) handleActions(w http.ResponseWriter, r *http.Request) {
_ = json.Unmarshal(body, &p)
writeJSON(w, http.StatusOK, s.rotateGuest(p.Site))
+ case "pool_set_config":
+ var p struct {
+ Enabled *bool `json:"enabled"`
+ CheckIntervalSeconds *int `json:"check_interval_seconds"`
+ CheckTimeoutSeconds *int `json:"check_timeout_seconds"`
+ CheckURL *string `json:"check_url"`
+ DefaultScheme *string `json:"default_scheme"`
+ XrayPath *string `json:"xray_path"`
+ XrayVersion *string `json:"xray_version"`
+ XrayAutoDownload *bool `json:"xray_auto_download"`
+ }
+ _ = json.Unmarshal(body, &p)
+ s.cfg.Lock()
+ pc := s.cfg.ProxyPool
+ if p.Enabled != nil {
+ pc.Enabled = *p.Enabled
+ }
+ if p.CheckIntervalSeconds != nil {
+ pc.CheckIntervalSeconds = clampInt(*p.CheckIntervalSeconds, 10, 3600)
+ }
+ if p.CheckTimeoutSeconds != nil {
+ pc.CheckTimeoutSeconds = clampInt(*p.CheckTimeoutSeconds, 2, 120)
+ }
+ if p.CheckURL != nil {
+ pc.CheckURL = strings.TrimSpace(*p.CheckURL)
+ }
+ if p.DefaultScheme != nil {
+ pc.DefaultScheme = strings.ToLower(strings.TrimSpace(*p.DefaultScheme))
+ }
+ if p.XrayPath != nil {
+ pc.XrayPath = strings.TrimSpace(*p.XrayPath)
+ }
+ if p.XrayVersion != nil {
+ pc.XrayVersion = strings.TrimSpace(*p.XrayVersion)
+ }
+ if p.XrayAutoDownload != nil {
+ pc.XrayAutoDownload = *p.XrayAutoDownload
+ }
+ s.cfg.ProxyPool = pc
+ s.cfg.Unlock()
+ s.poolReload()
+ s.save(w)
+
+ case "pool_add_entry":
+ var p struct {
+ Name string `json:"name"`
+ Link string `json:"link"`
+ }
+ _ = json.Unmarshal(body, &p)
+ p.Name, p.Link = strings.TrimSpace(p.Name), strings.TrimSpace(p.Link)
+ if !validPoolName(p.Name) {
+ fail(w, "name must match [A-Za-z0-9_-]{1,64}")
+ return
+ }
+ if p.Link == "" {
+ fail(w, "link is required")
+ return
+ }
+ s.cfg.RLock()
+ scheme := s.cfg.ProxyPool.DefaultScheme
+ _, exists := s.cfg.ProxyPool.Entries[p.Name]
+ s.cfg.RUnlock()
+ if exists {
+ fail(w, "entry already exists: "+p.Name)
+ return
+ }
+ if _, err := proxypool.ParseNode(p.Link, scheme); err != nil {
+ fail(w, "node parse: "+err.Error())
+ return
+ }
+ s.cfg.Lock()
+ if s.cfg.ProxyPool.Entries == nil {
+ s.cfg.ProxyPool.Entries = map[string]*config.ProxyEntry{}
+ }
+ s.cfg.ProxyPool.Entries[p.Name] = &config.ProxyEntry{Link: p.Link, Enabled: true}
+ s.cfg.Unlock()
+ s.poolReload()
+ s.save(w)
+
+ case "pool_toggle_entry", "pool_toggle_sub":
+ var p struct {
+ Name string `json:"name"`
+ Enabled bool `json:"enabled"`
+ }
+ _ = json.Unmarshal(body, &p)
+ s.cfg.Lock()
+ ok := false
+ if req.Action == "pool_toggle_entry" {
+ if e := s.cfg.ProxyPool.Entries[p.Name]; e != nil {
+ e.Enabled = p.Enabled
+ ok = true
+ }
+ } else {
+ if sc := s.cfg.ProxyPool.Subscriptions[p.Name]; sc != nil {
+ sc.Enabled = p.Enabled
+ ok = true
+ }
+ }
+ s.cfg.Unlock()
+ if !ok {
+ fail(w, "not found: "+p.Name)
+ return
+ }
+ s.poolReload()
+ s.save(w)
+
+ case "pool_delete_entry", "pool_delete_sub":
+ var p struct {
+ Name string `json:"name"`
+ }
+ _ = json.Unmarshal(body, &p)
+ ref := p.Name
+ if req.Action == "pool_delete_sub" {
+ ref = "sub:" + p.Name
+ }
+ s.cfg.Lock()
+ if req.Action == "pool_delete_entry" {
+ delete(s.cfg.ProxyPool.Entries, p.Name)
+ } else {
+ delete(s.cfg.ProxyPool.Subscriptions, p.Name)
+ }
+ for _, st := range s.cfg.Sites {
+ if st != nil {
+ st.Proxies = removeString(st.Proxies, ref)
+ }
+ }
+ s.cfg.Unlock()
+ s.poolReload()
+ s.save(w)
+
+ case "pool_add_sub":
+ var p struct {
+ Name string `json:"name"`
+ URL string `json:"url"`
+ IntervalMinutes int `json:"interval_minutes"`
+ }
+ _ = json.Unmarshal(body, &p)
+ p.Name, p.URL = strings.TrimSpace(p.Name), strings.TrimSpace(p.URL)
+ if !validPoolName(p.Name) {
+ fail(w, "name must match [A-Za-z0-9_-]{1,64}")
+ return
+ }
+ if p.URL == "" {
+ fail(w, "url is required")
+ return
+ }
+ s.cfg.Lock()
+ if s.cfg.ProxyPool.Subscriptions == nil {
+ s.cfg.ProxyPool.Subscriptions = map[string]*config.ProxySubscription{}
+ }
+ if _, exists := s.cfg.ProxyPool.Subscriptions[p.Name]; exists {
+ s.cfg.Unlock()
+ fail(w, "subscription already exists: "+p.Name)
+ return
+ }
+ if p.IntervalMinutes <= 0 {
+ p.IntervalMinutes = 60
+ }
+ s.cfg.ProxyPool.Subscriptions[p.Name] = &config.ProxySubscription{
+ URL: p.URL, Enabled: true,
+ IntervalMinutes: clampInt(p.IntervalMinutes, 5, 1440),
+ }
+ s.cfg.Unlock()
+ s.poolReload()
+ s.save(w)
+
+ case "pool_set_site":
+ var p struct {
+ Site string `json:"site"`
+ Proxies []string `json:"proxies"`
+ }
+ _ = json.Unmarshal(body, &p)
+ clean := make([]string, 0, len(p.Proxies))
+ for _, item := range p.Proxies {
+ if item = strings.TrimSpace(item); item != "" {
+ clean = append(clean, item)
+ }
+ }
+ s.cfg.Lock()
+ st, ok := s.cfg.Sites[p.Site]
+ if ok && st != nil {
+ for _, item := range clean {
+ if name, isSub := strings.CutPrefix(item, "sub:"); isSub {
+ if s.cfg.ProxyPool.Subscriptions[name] == nil {
+ s.cfg.Unlock()
+ fail(w, "unknown subscription: "+name)
+ return
+ }
+ } else if s.cfg.ProxyPool.Entries[item] == nil {
+ s.cfg.Unlock()
+ fail(w, "unknown entry: "+item)
+ return
+ }
+ }
+ st.Proxies = clean
+ }
+ s.cfg.Unlock()
+ if !ok || st == nil {
+ fail(w, "unknown site: "+p.Site)
+ return
+ }
+ s.poolReload()
+ s.save(w)
+
+ case "pool_refresh":
+ if s.pool == nil {
+ fail(w, "proxy pool is not available")
+ return
+ }
+ s.pool.RefreshNow(context.Background())
+ writeJSON(w, http.StatusOK, map[string]any{"ok": true})
+
default:
fail(w, "unknown action: "+req.Action)
}
@@ -673,3 +887,45 @@ func randomToken(n int) string {
}
return hex.EncodeToString(b)
}
+
+func clampInt(v, lo, hi int) int {
+ if v < lo {
+ return lo
+ }
+ if v > hi {
+ return hi
+ }
+ return v
+}
+
+func clampFloat(v, lo, hi float64) float64 {
+ if v < lo {
+ return lo
+ }
+ if v > hi {
+ return hi
+ }
+ return v
+}
+
+var poolNameRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`)
+
+func validPoolName(name string) bool { return poolNameRe.MatchString(name) }
+
+// removeString drops every occurrence of want from list (in place copy).
+func removeString(list []string, want string) []string {
+ out := list[:0:0]
+ for _, v := range list {
+ if v != want {
+ out = append(out, v)
+ }
+ }
+ return out
+}
+
+// poolReload pushes pool-relevant config changes into the live manager.
+func (s *Server) poolReload() {
+ if s.pool != nil {
+ s.pool.Reload()
+ }
+}
diff --git a/internal/admin/server.go b/internal/admin/server.go
index 89e568d..e298e16 100644
--- a/internal/admin/server.go
+++ b/internal/admin/server.go
@@ -15,6 +15,7 @@ import (
"github.com/nyoungo/dsfree2api/internal/config"
"github.com/nyoungo/dsfree2api/internal/logbuf"
"github.com/nyoungo/dsfree2api/internal/metrics"
+ "github.com/nyoungo/dsfree2api/internal/proxypool"
"github.com/nyoungo/dsfree2api/internal/turnstile"
"github.com/nyoungo/dsfree2api/internal/upstream"
)
@@ -31,6 +32,7 @@ type Server struct {
met *metrics.Recorder
logs *logbuf.Buffer
ts *turnstile.Solver
+ pool *proxypool.Manager
log *slog.Logger
start time.Time
@@ -39,7 +41,7 @@ type Server struct {
password string
}
-func New(cfg *config.Config, up *upstream.Client, met *metrics.Recorder, logs *logbuf.Buffer, ts *turnstile.Solver, log *slog.Logger) *Server {
+func New(cfg *config.Config, up *upstream.Client, met *metrics.Recorder, logs *logbuf.Buffer, ts *turnstile.Solver, log *slog.Logger, pool *proxypool.Manager) *Server {
if log == nil {
log = slog.Default()
}
@@ -50,7 +52,7 @@ func New(cfg *config.Config, up *upstream.Client, met *metrics.Recorder, logs *l
generated = true
}
s := &Server{
- cfg: cfg, up: up, met: met, logs: logs, ts: ts, log: log,
+ cfg: cfg, up: up, met: met, logs: logs, ts: ts, pool: pool, log: log,
start: time.Now(), tokens: map[string]time.Time{}, password: pw,
}
if generated {
@@ -185,7 +187,7 @@ func (s *Server) handleOverview(w http.ResponseWriter, _ *http.Request) {
sites = append(sites, map[string]any{
"code": code, "base_url": st.BaseURL, "ajax_url": st.AJAXURL,
"language": st.Language, "sitekey": st.SiteKey, "enabled": st.Enabled,
- "verify_action": st.VerifyAction,
+ "verify_action": st.VerifyAction, "proxies": st.Proxies,
})
}
proxyCfg := struct {
@@ -201,12 +203,17 @@ func (s *Server) handleOverview(w http.ResponseWriter, _ *http.Request) {
keys := append([]string(nil), s.cfg.Security.APIKeys...)
s.cfg.RUnlock()
+ var poolStatus any
+ if s.pool != nil {
+ poolStatus = s.pool.Status()
+ }
writeJSON(w, http.StatusOK, map[string]any{
"metrics": s.met.Snapshot(),
"models": models,
"sites": sites,
"proxy": proxyCfg,
"turnstile": map[string]any{"config": turnstileCfg, "status": s.ts.Status(siteCodes), "history": s.ts.History()},
+ "proxypool": poolStatus,
"keys": keys,
"cache": map[string]any{"chat_config_entries": s.up.ConfigCacheSize(), "routes": routeList(s.up)},
"system": map[string]any{
@@ -269,6 +276,9 @@ func (s *Server) handlePutConfig(w http.ResponseWriter, r *http.Request) {
// applyRuntime pushes a freshly edited config into the live subsystems.
func (s *Server) applyRuntime() {
s.ts.SetConfig(s.currentTurnstile())
+ if s.pool != nil {
+ s.pool.Reload()
+ }
}
func (s *Server) currentTurnstile() config.Turnstile {
diff --git a/internal/admin/web/app.js b/internal/admin/web/app.js
index f6bd42c..b4473cb 100644
--- a/internal/admin/web/app.js
+++ b/internal/admin/web/app.js
@@ -328,6 +328,7 @@ const mask = k => k.length > 16 ? k.slice(0, 7) + "…" + k.slice(-6) : k;
PAGES.proxy = async () => {
const ov = await fetchOverview();
const p = ov.proxy;
+ const pool = ov.proxypool || { xray: {}, entries: [], subscriptions: [], sticky: {} };
view.innerHTML = `
当前生效顺序：${ov.cache.routes.map(r => `${esc(r.name)} → ${esc(r.proxy)}`).join(" ")}
+
+
+
代理池 · Xray 核心
+
+
${switchHTML("pool-en", pool.enabled)}
+
+
+
+
+
+
+
+
${switchHTML("pool-xray-auto", pool.xray_auto_download)}
+
+
+
+
+
+
Xray：${pool.xray.running ? "运行中 · pid " + pool.xray.pid : "未运行"}
+ ${pool.xray.error ? ` ${esc(pool.xray.error)}` : ""}
+ ${pool.last_error ? `
${esc(pool.last_error)}` : ""}
+
+
节点与端点
+
+
+
+
+
分享链接（vless / vmess / trojan / ss）由内置 Xray 核心承载；ip:port / socks5:// 等端点直接使用。
+
+
站点绑定（按顺序，粘性选路）
+
| 站点 | 绑定（逗号分隔：名称 或 sub:订阅名） | 当前出口 | |
|---|
+
未绑定、或绑定节点全部不健康时，回退到主线路 / 备用线路。Cookie 与出口 IP 绑定：切换出口后该站点会在下一轮巡检重新求解 Turnstile。
${esc(e.name)}${e.source ? ` ← ${esc(e.source)}` : ""} |
+ ${esc(e.kind === "url" ? e.scheme : (e.scheme || "xray"))} |
+ ${esc(e.target || "")} |
+ ${health} | ${e.latency_ms ? e.latency_ms + "ms" : "–"} |
+
+ ${e.source ? "" : ``} |
`;
+ }).join("") : '| 尚无节点 —— 添加分享链接或 ip:port 端点 |
';
+ $$("[data-ptoggle]").forEach(b => b.onclick = async () => {
+ try { await api("/api/actions", { body: { action: "pool_toggle_entry", name: b.dataset.ptoggle, enabled: b.dataset.pen === "1" } }); PAGES.proxy(); }
+ catch (e) { toast(e.message, "err"); }
+ });
+ $$("[data-pdel]").forEach(b => b.onclick = async () => {
+ if (!confirm("删除节点 " + b.dataset.pdel + "？")) return;
+ try { await api("/api/actions", { body: { action: "pool_delete_entry", name: b.dataset.pdel } }); PAGES.proxy(); }
+ catch (e) { toast(e.message, "err"); }
+ });
+
+ const subs = pool.subscriptions || [];
+ $("#tb-subs").innerHTML = subs.length ? subs.map(s => `
+ ${esc(s.name)} | ${esc(s.url)} |
+ ${s.interval_minutes || 60}m | ${s.nodes || 0} |
+ ${s.fetched_at ? timeStr(s.fetched_at * 1000) : "未拉取"}${s.last_error ? ` ${esc(String(s.last_error).slice(0, 60))}` : ""} |
+
+ |
`).join("")
+ : '| 尚无订阅 |
';
+ $$("[data-stoggle]").forEach(b => b.onclick = async () => {
+ try { await api("/api/actions", { body: { action: "pool_toggle_sub", name: b.dataset.stoggle, enabled: b.dataset.sen === "1" } }); PAGES.proxy(); }
+ catch (e) { toast(e.message, "err"); }
+ });
+ $$("[data-sdel]").forEach(b => b.onclick = async () => {
+ if (!confirm("删除订阅 " + b.dataset.sdel + "？")) return;
+ try { await api("/api/actions", { body: { action: "pool_delete_sub", name: b.dataset.sdel } }); PAGES.proxy(); }
+ catch (e) { toast(e.message, "err"); }
+ });
+
+ const sticky = pool.sticky || {};
+ $("#tb-bind").innerHTML = (ov.sites || []).map(s => {
+ const cur = sticky[s.code];
+ const curEntry = entries.find(e => e.name === cur);
+ return `| ${esc(s.code.toUpperCase())} |
+ |
+ ${cur ? `${esc(cur)} → ${esc(curEntry ? curEntry.target : "?")}` : "全局线路"} |
+ |
`;
+ }).join("");
+ $$("[data-bsave]").forEach(b => b.onclick = async () => {
+ const input = $(`[data-bind="${CSS.escape(b.dataset.bsave)}"]`);
+ const list = input.value.split(",").map(x => x.trim()).filter(Boolean);
+ try { await api("/api/actions", { body: { action: "pool_set_site", site: b.dataset.bsave, proxies: list } }); toast("已保存", "ok"); PAGES.proxy(); }
+ catch (e) { toast(e.message, "err"); }
+ });
+ };
+ renderPool();
+
+ $("#pool-save").onclick = async () => {
+ try {
+ await api("/api/actions", { body: {
+ action: "pool_set_config",
+ enabled: $("#pool-en").checked,
+ check_interval_seconds: +$("#pool-check-int").value || 120,
+ check_timeout_seconds: +$("#pool-check-to").value || 10,
+ check_url: $("#pool-check-url").value.trim(),
+ default_scheme: $("#pool-scheme").value.trim() || "http",
+ xray_path: $("#pool-xray-path").value.trim(),
+ xray_version: $("#pool-xray-ver").value.trim(),
+ xray_auto_download: $("#pool-xray-auto").checked } });
+ toast("池配置已保存", "ok"); PAGES.proxy();
+ } catch (e) { toast(e.message, "err"); }
+ };
+ $("#pool-refresh").onclick = async () => {
+ const b = $("#pool-refresh"); b.disabled = true; b.textContent = "刷新中…";
+ try { await api("/api/actions", { body: { action: "pool_refresh" } }); toast("已在后台拉取订阅并检查健康，稍后重进本页查看", "ok"); PAGES.proxy(); }
+ catch (e) { toast(e.message, "err"); b.disabled = false; b.textContent = "刷新节点与健康检查"; }
+ };
+ $("#pool-add").onclick = async () => {
+ try {
+ await api("/api/actions", { body: { action: "pool_add_entry", name: $("#pool-name").value.trim(), link: $("#pool-link").value.trim() } });
+ toast("已添加", "ok"); PAGES.proxy();
+ } catch (e) { toast(e.message, "err"); }
+ };
+ $("#sub-add").onclick = async () => {
+ try {
+ await api("/api/actions", { body: { action: "pool_add_sub", name: $("#sub-name").value.trim(), url: $("#sub-url").value.trim(), interval_minutes: +$("#sub-int").value || 60 } });
+ toast("已添加", "ok"); PAGES.proxy();
+ } catch (e) { toast(e.message, "err"); }
+ };
+
$("#px-save").onclick = async () => {
try {
await api("/api/config", { method: "PUT", body: await merged(c => { c.proxy.url = $("#px-primary").value.trim(); }) });
@@ -406,6 +542,7 @@ async function merged(mutate) {
}
PAGES.turnstile = async () => {
+ if (overviewTimer) { clearInterval(overviewTimer); overviewTimer = null; }
const ov = await fetchOverview();
const t = ov.turnstile.config;
view.innerHTML = `
diff --git a/internal/config/config.go b/internal/config/config.go
index f63d1f2..fc5debb 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -59,6 +59,41 @@ type Site struct {
VerifyAction string `toml:"verify_action" json:"verify_action"`
Language string `toml:"language" json:"language"`
Enabled bool `toml:"enabled" json:"enabled"`
+ // Proxies binds this site to the proxy pool: an ordered list of entry
+ // names and/or "sub:" references. Empty = global [proxy] routes.
+ Proxies []string `toml:"proxies" json:"proxies"`
+ // WarmPoolOnly keeps the cookie warmer on the pool routes bound via
+ // Proxies; the global [proxy] primary/fallback lanes are not pre-solved
+ // (they remain available for request-time failover). Default false.
+ WarmPoolOnly bool `toml:"warm_pool_only" json:"warm_pool_only"`
+}
+
+// ProxyEntry is one manually configured pool node: an Xray share link
+// (vless/vmess/trojan/ss) or a plain http/https/socks5 endpoint.
+type ProxyEntry struct {
+ Link string `toml:"link" json:"link"`
+ Enabled bool `toml:"enabled" json:"enabled"`
+}
+
+// ProxySubscription is a remote node list refreshed on an interval.
+type ProxySubscription struct {
+ URL string `toml:"url" json:"url"`
+ Enabled bool `toml:"enabled" json:"enabled"`
+ IntervalMinutes int `toml:"interval_minutes" json:"interval_minutes"`
+}
+
+// ProxyPool configures the egress pool and its managed Xray core.
+type ProxyPool struct {
+ Enabled bool `toml:"enabled" json:"enabled"`
+ CheckIntervalSeconds int `toml:"check_interval_seconds" json:"check_interval_seconds"`
+ CheckTimeoutSeconds int `toml:"check_timeout_seconds" json:"check_timeout_seconds"`
+ CheckURL string `toml:"check_url" json:"check_url"`
+ DefaultScheme string `toml:"default_scheme" json:"default_scheme"`
+ XrayPath string `toml:"xray_path" json:"xray_path"`
+ XrayVersion string `toml:"xray_version" json:"xray_version"`
+ XrayAutoDownload bool `toml:"xray_auto_download" json:"xray_auto_download"`
+ Entries map[string]*ProxyEntry `toml:"entries" json:"entries"`
+ Subscriptions map[string]*ProxySubscription `toml:"subscriptions" json:"subscriptions"`
}
type Model struct {
@@ -101,6 +136,7 @@ type Config struct {
SlowStartSeconds float64 `toml:"slow_start_seconds" json:"slow_start_seconds"`
} `toml:"proxy" json:"proxy"`
+ ProxyPool ProxyPool `toml:"proxypool" json:"proxypool"`
Upstream struct {
Timeout float64 `toml:"timeout" json:"timeout"`
StreamTimeout float64 `toml:"stream_timeout" json:"stream_timeout"`
@@ -125,6 +161,20 @@ type Config struct {
path string `toml:"-"`
}
+// DefaultProxyPool returns the built-in pool settings (feature off, Xray
+// auto-download on).
+func DefaultProxyPool() ProxyPool {
+ return ProxyPool{
+ CheckIntervalSeconds: 120,
+ CheckTimeoutSeconds: 10,
+ CheckURL: "https://www.gstatic.com/generate_204",
+ DefaultScheme: "http",
+ XrayAutoDownload: true,
+ Entries: map[string]*ProxyEntry{},
+ Subscriptions: map[string]*ProxySubscription{},
+ }
+}
+
func DefaultTurnstile() Turnstile {
return Turnstile{
Enabled: false,
@@ -201,6 +251,7 @@ func newConfig() *Config {
c.Upstream.CrossSiteFailover = true
c.Upstream.ContinueRounds = 20
c.Runtime.DataDir = "./data"
+ c.ProxyPool = DefaultProxyPool()
c.Turnstile = DefaultTurnstile()
c.Sites = DefaultSites(c.Turnstile.SiteKey)
c.Models = DefaultModels()
@@ -300,22 +351,36 @@ func (c *Config) applyDefaults(md *toml.MetaData) {
c.Turnstile.Enabled = true
c.Upstream.AutoRefresh = true
c.Upstream.ContinueRounds = 20
+ c.ProxyPool.XrayAutoDownload = true
for _, s := range c.Sites {
s.Enabled = true
}
for _, m := range c.Models {
m.Enabled = true
}
+ for _, e := range c.ProxyPool.Entries {
+ e.Enabled = true
+ }
+ for _, sc := range c.ProxyPool.Subscriptions {
+ sc.Enabled = true
+ }
return
}
def(md.IsDefined("turnstile", "enabled"), &c.Turnstile.Enabled, true)
def(md.IsDefined("upstream", "auto_refresh"), &c.Upstream.AutoRefresh, true)
+ def(md.IsDefined("proxypool", "xray_auto_download"), &c.ProxyPool.XrayAutoDownload, true)
for code, s := range c.Sites {
def(md.IsDefined("sites", code, "enabled"), &s.Enabled, true)
}
for id, m := range c.Models {
def(md.IsDefined("models", id, "enabled"), &m.Enabled, true)
}
+ for name, e := range c.ProxyPool.Entries {
+ def(md.IsDefined("proxypool", "entries", name, "enabled"), &e.Enabled, true)
+ }
+ for name, sc := range c.ProxyPool.Subscriptions {
+ def(md.IsDefined("proxypool", "subscriptions", name, "enabled"), &sc.Enabled, true)
+ }
}
func (c *Config) applyEnv() error {
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 26dae15..a3b36a7 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -204,6 +204,78 @@ func TestTurnstileProviderEnv(t *testing.T) {
}
}
+func TestProxyPoolDefaults(t *testing.T) {
+ d := DefaultProxyPool()
+ if d.Enabled {
+ t.Error("proxy pool must be opt-in")
+ }
+ if !d.XrayAutoDownload || d.CheckIntervalSeconds != 120 || d.CheckTimeoutSeconds != 10 {
+ t.Errorf("pool defaults = %+v", d)
+ }
+ if d.DefaultScheme != "http" {
+ t.Errorf("default scheme = %q", d.DefaultScheme)
+ }
+ cfg, err := Load(examplePath)
+ if err != nil {
+ t.Fatalf("load: %v", err)
+ }
+ if !cfg.ProxyPool.XrayAutoDownload {
+ t.Error("example config should keep xray_auto_download = true")
+ }
+ if cfg.ProxyPool.CheckIntervalSeconds != 120 {
+ t.Errorf("check interval = %d", cfg.ProxyPool.CheckIntervalSeconds)
+ }
+}
+
+func TestTurnstileSolveTimeoutAndWarmDefaults(t *testing.T) {
+ d := DefaultTurnstile()
+ if d.WarmEnabled {
+ t.Error("cookie pool must be opt-in")
+ }
+ if d.WarmRatio != 0.3 || d.WarmCheckSeconds != 60 {
+ t.Errorf("warm defaults = %v / %v", d.WarmRatio, d.WarmCheckSeconds)
+ }
+ z := Turnstile{}
+ if z.SolveTimeoutValue() != 90 {
+ t.Errorf("zero timeout should normalize to 90, got %d", z.SolveTimeoutValue())
+ }
+ if z.WarmRatioValue() != 0.3 || z.WarmCheckValue() != 60 {
+ t.Errorf("zero warm values should normalize to 0.3 / 60")
+ }
+ if got := (Turnstile{TimeoutSeconds: 1}).SolveTimeoutValue(); got != 5 {
+ t.Errorf("timeout floor = %d, want 5", got)
+ }
+ if got := (Turnstile{TimeoutSeconds: 9999}).SolveTimeoutValue(); got != 600 {
+ t.Errorf("timeout ceiling = %d, want 600", got)
+ }
+ if got := (Turnstile{WarmRatio: 0.99}).WarmRatioValue(); got != 0.95 {
+ t.Errorf("warm ratio ceiling = %v, want 0.95", got)
+ }
+ if (Turnstile{}).APIStyleValue() != "sync" {
+ t.Error("zero api_style should normalize to sync")
+ }
+ if (Turnstile{APIStyle: "EzSolver"}).APIStyleValue() != "ezsolver" {
+ t.Error("api_style should normalize EzSolver")
+ }
+}
+
+func TestTurnstileSolveAndWarmEnvOverrides(t *testing.T) {
+ t.Setenv("TURNSTILE_TIMEOUT_SECONDS", "45")
+ t.Setenv("TURNSTILE_WARM_ENABLED", "true")
+ t.Setenv("TURNSTILE_WARM_RATIO", "0.5")
+ t.Setenv("TURNSTILE_WARM_CHECK_SECONDS", "120")
+ cfg, err := Load(examplePath)
+ if err != nil {
+ t.Fatalf("load: %v", err)
+ }
+ if cfg.Turnstile.TimeoutSeconds != 45 {
+ t.Errorf("timeout = %d, want 45", cfg.Turnstile.TimeoutSeconds)
+ }
+ if !cfg.Turnstile.WarmEnabled || cfg.Turnstile.WarmRatio != 0.5 || cfg.Turnstile.WarmCheckSeconds != 120 {
+ t.Errorf("warm env = %+v", cfg.Turnstile)
+ }
+}
+
func TestSaveRoundTrip(t *testing.T) {
cfg, err := Load(examplePath)
if err != nil {
diff --git a/internal/proxypool/manager.go b/internal/proxypool/manager.go
new file mode 100644
index 0000000..c738b97
--- /dev/null
+++ b/internal/proxypool/manager.go
@@ -0,0 +1,722 @@
+package proxypool
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/url"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/nyoungo/dsfree2api/internal/config"
+ "github.com/nyoungo/dsfree2api/internal/upstream"
+)
+
+// endpoint is one runtime pool slot.
+type endpoint struct {
+ name string
+ kind string // "xray" | "url" | "invalid"
+ scheme string
+ display string
+ link string
+ source string // "" = manual entry, else subscription name
+ enabled bool
+
+ parseErr string
+ outbound map[string]any
+ url string
+ proxyURL string // resolved egress URL (socks5://127.0.0.1:port or plain)
+
+ checked bool
+ healthy bool
+ latencyMs int64
+ lastErr string
+ checkedAt time.Time
+}
+
+type subState struct {
+ name string
+ nodes []string
+ lastErr string
+ fetchedAt time.Time
+}
+
+// Manager owns the proxy pool: endpoints, health, subscriptions and sticky
+// per-site selection. It is safe for concurrent use.
+type Manager struct {
+ cfg *config.Config
+ log *slog.Logger
+ dir string
+ xray *xrayCore
+
+ reloadMu sync.Mutex
+
+ mu sync.Mutex
+ endpoints map[string]*endpoint
+ subs map[string]*subState
+ sticky map[string]string
+ fetching map[string]bool
+ started bool
+ lastCheck time.Time
+ lastErr string
+}
+
+func New(cfg *config.Config, dataDir string, log *slog.Logger) *Manager {
+ if log == nil {
+ log = slog.Default()
+ }
+ return &Manager{
+ cfg: cfg,
+ log: log,
+ dir: dataDir,
+ xray: newXrayCore(dataDir, log),
+ endpoints: map[string]*endpoint{},
+ subs: map[string]*subState{},
+ sticky: map[string]string{},
+ fetching: map[string]bool{},
+ }
+}
+
+// Start runs the background loop: reload from config, refresh due
+// subscriptions, probe endpoint health.
+func (m *Manager) Start(ctx context.Context) {
+ m.mu.Lock()
+ if m.started {
+ m.mu.Unlock()
+ return
+ }
+ m.started = true
+ m.mu.Unlock()
+ go m.loop(ctx)
+}
+
+func (m *Manager) loop(ctx context.Context) {
+ ticker := time.NewTicker(5 * time.Second)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ m.xray.stop()
+ return
+ case <-ticker.C:
+ }
+ m.cycle(ctx)
+ }
+}
+
+func (m *Manager) cycle(ctx context.Context) {
+ m.Reload()
+
+ m.cfg.RLock()
+ enabled := m.cfg.ProxyPool.Enabled
+ interval := time.Duration(m.cfg.ProxyPool.CheckIntervalSeconds) * time.Second
+ subs := make(map[string]config.ProxySubscription, len(m.cfg.ProxyPool.Subscriptions))
+ for name, sc := range m.cfg.ProxyPool.Subscriptions {
+ if sc != nil {
+ subs[name] = *sc
+ }
+ }
+ m.cfg.RUnlock()
+ if !enabled {
+ return
+ }
+ if interval <= 0 {
+ interval = 120 * time.Second
+ }
+
+ m.mu.Lock()
+ due := m.lastCheck.IsZero() || time.Since(m.lastCheck) >= interval
+ m.mu.Unlock()
+ if due {
+ m.checkAll(ctx)
+ }
+
+ now := time.Now()
+ for name, sc := range subs {
+ if !sc.Enabled || strings.TrimSpace(sc.URL) == "" {
+ continue
+ }
+ m.mu.Lock()
+ st := m.subs[name]
+ busy := m.fetching[name]
+ m.mu.Unlock()
+ if busy {
+ continue
+ }
+ iv := time.Duration(sc.IntervalMinutes) * time.Minute
+ if iv <= 0 {
+ iv = time.Hour
+ }
+ if st != nil && !st.fetchedAt.IsZero() && now.Sub(st.fetchedAt) < iv {
+ continue
+ }
+ go m.fetchSub(ctx, name, sc)
+ }
+}
+
+// Reload rebuilds the runtime endpoint table from the live config plus the
+// cached subscription nodes, (re)configures the Xray core and prunes sticky
+// picks that no longer resolve.
+func (m *Manager) Reload() {
+ m.reloadMu.Lock()
+ defer m.reloadMu.Unlock()
+
+ m.cfg.RLock()
+ poolCfg := m.cfg.ProxyPool
+ entries := make(map[string]config.ProxyEntry, len(poolCfg.Entries))
+ for name, e := range poolCfg.Entries {
+ if e != nil {
+ entries[name] = *e
+ }
+ }
+ subsCfg := make(map[string]config.ProxySubscription, len(poolCfg.Subscriptions))
+ for name, sc := range poolCfg.Subscriptions {
+ if sc != nil {
+ subsCfg[name] = *sc
+ }
+ }
+ m.cfg.RUnlock()
+
+ m.mu.Lock()
+ subNodes := make(map[string][]string, len(m.subs))
+ for name, st := range m.subs {
+ subNodes[name] = append([]string(nil), st.nodes...)
+ }
+ old := m.endpoints
+ m.mu.Unlock()
+
+ desired := map[string]*endpoint{}
+ for name, e := range entries {
+ if strings.TrimSpace(e.Link) == "" {
+ continue
+ }
+ ep := newEndpoint(name, e.Link, poolCfg.DefaultScheme, "")
+ ep.enabled = e.Enabled
+ desired[name] = ep
+ }
+ for subName, nodes := range subNodes {
+ sc, ok := subsCfg[subName]
+ if !ok {
+ continue
+ }
+ for _, line := range nodes {
+ name := subName + "#" + shortHash(line)
+ if _, dup := desired[name]; dup {
+ continue
+ }
+ ep := newEndpoint(name, line, poolCfg.DefaultScheme, subName)
+ ep.enabled = sc.Enabled
+ desired[name] = ep
+ }
+ }
+ for name, ep := range desired {
+ if o := old[name]; o != nil && o.link == ep.link {
+ ep.checked = o.checked
+ ep.healthy = o.healthy
+ ep.latencyMs = o.latencyMs
+ ep.lastErr = o.lastErr
+ ep.checkedAt = o.checkedAt
+ }
+ }
+
+ nodes := make([]xrayNode, 0, len(desired))
+ for _, ep := range desired {
+ if ep.kind == "xray" && ep.enabled && ep.parseErr == "" && ep.outbound != nil {
+ nodes = append(nodes, xrayNode{Name: ep.name, Outbound: ep.outbound})
+ }
+ }
+ sort.Slice(nodes, func(i, j int) bool { return nodes[i].Name < nodes[j].Name })
+
+ if len(nodes) > 0 {
+ if err := m.xray.ensure(poolCfg.XrayPath, poolCfg.XrayAutoDownload, poolCfg.XrayVersion); err != nil {
+ m.setErr(err)
+ } else if err := m.xray.configure(nodes); err != nil {
+ m.setErr(err)
+ } else {
+ m.setErr(nil)
+ }
+ } else {
+ _ = m.xray.configure(nil)
+ m.setErr(nil)
+ }
+
+ xrayUp := m.xray.status().Running
+ for _, ep := range desired {
+ switch ep.kind {
+ case "url":
+ ep.proxyURL = ep.url
+ case "xray":
+ if ep.parseErr == "" && xrayUp {
+ if port := m.xray.port(ep.name); port > 0 {
+ ep.proxyURL = fmt.Sprintf("socks5://127.0.0.1:%d", port)
+ }
+ }
+ }
+ }
+
+ m.mu.Lock()
+ m.endpoints = desired
+ for site, name := range m.sticky {
+ ep := desired[name]
+ if ep == nil || !ep.enabled || ep.proxyURL == "" {
+ delete(m.sticky, site)
+ }
+ }
+ m.mu.Unlock()
+}
+
+// RefreshNow refreshes every subscription and probes every endpoint, then
+// reloads. Used by the console button.
+func (m *Manager) RefreshNow(ctx context.Context) {
+ go func() {
+ m.Reload()
+ m.cfg.RLock()
+ enabled := m.cfg.ProxyPool.Enabled
+ subs := make(map[string]config.ProxySubscription, len(m.cfg.ProxyPool.Subscriptions))
+ for name, sc := range m.cfg.ProxyPool.Subscriptions {
+ if sc != nil {
+ subs[name] = *sc
+ }
+ }
+ m.cfg.RUnlock()
+ for name, sc := range subs {
+ if sc.Enabled && strings.TrimSpace(sc.URL) != "" {
+ m.fetchSub(ctx, name, sc)
+ }
+ }
+ if enabled {
+ m.checkAll(ctx)
+ }
+ m.Reload()
+ }()
+}
+
+func (m *Manager) fetchSub(ctx context.Context, name string, sc config.ProxySubscription) {
+ m.mu.Lock()
+ m.fetching[name] = true
+ m.mu.Unlock()
+ defer func() {
+ m.mu.Lock()
+ delete(m.fetching, name)
+ m.mu.Unlock()
+ }()
+
+ client, err := m.fetchClient()
+ if err == nil {
+ var req *http.Request
+ req, err = http.NewRequestWithContext(ctx, http.MethodGet, sc.URL, nil)
+ if err == nil {
+ var resp *http.Response
+ resp, err = client.Do(req)
+ if err == nil {
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ err = fmt.Errorf("subscription http %d", resp.StatusCode)
+ } else {
+ var raw []byte
+ raw, err = io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+ if err == nil {
+ lines := ParseSubscriptionBody(raw)
+ if len(lines) == 0 {
+ err = fmt.Errorf("subscription returned no nodes")
+ } else {
+ err = nil
+ m.subResult(name, lines, nil)
+ m.log.Info("subscription refreshed", "name", name, "nodes", len(lines))
+ m.Reload()
+ return
+ }
+ }
+ }
+ }
+ }
+ }
+ m.subResult(name, nil, err)
+ m.log.Warn("subscription refresh failed", "name", name, "error", err)
+}
+
+func (m *Manager) subResult(name string, nodes []string, err error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ st := m.subs[name]
+ if st == nil {
+ st = &subState{name: name}
+ m.subs[name] = st
+ }
+ if err != nil {
+ st.lastErr = err.Error()
+ return
+ }
+ st.nodes = nodes
+ st.lastErr = ""
+ st.fetchedAt = time.Now()
+}
+
+func (m *Manager) fetchClient() (*http.Client, error) {
+ m.cfg.RLock()
+ proxy := strings.TrimSpace(m.cfg.Proxy.URL)
+ m.cfg.RUnlock()
+ tr := &http.Transport{
+ TLSHandshakeTimeout: 20 * time.Second,
+ ResponseHeaderTimeout: 30 * time.Second,
+ }
+ if proxy != "" {
+ u, err := url.Parse(proxy)
+ if err != nil {
+ return nil, fmt.Errorf("proxy url: %w", err)
+ }
+ tr.Proxy = http.ProxyURL(u)
+ }
+ return &http.Client{Transport: tr, Timeout: 45 * time.Second}, nil
+}
+
+func (m *Manager) checkAll(ctx context.Context) {
+ m.mu.Lock()
+ list := make([]*endpoint, 0, len(m.endpoints))
+ for _, ep := range m.endpoints {
+ list = append(list, ep)
+ }
+ m.mu.Unlock()
+
+ m.cfg.RLock()
+ timeout := time.Duration(m.cfg.ProxyPool.CheckTimeoutSeconds) * time.Second
+ checkURL := strings.TrimSpace(m.cfg.ProxyPool.CheckURL)
+ m.cfg.RUnlock()
+ if timeout <= 0 {
+ timeout = 10 * time.Second
+ }
+ if checkURL == "" {
+ checkURL = "https://www.gstatic.com/generate_204"
+ }
+
+ sem := make(chan struct{}, 4)
+ var wg sync.WaitGroup
+ for _, ep := range list {
+ if !ep.enabled || ep.proxyURL == "" {
+ continue
+ }
+ wg.Add(1)
+ sem <- struct{}{}
+ go func(ep *endpoint) {
+ defer wg.Done()
+ defer func() { <-sem }()
+ if ctx.Err() != nil {
+ return
+ }
+ ok, ms, errMsg := probeURL(ctx, ep.proxyURL, checkURL, timeout)
+ m.mu.Lock()
+ ep.checked = true
+ ep.checkedAt = time.Now()
+ ep.healthy = ok
+ ep.latencyMs = ms
+ ep.lastErr = errMsg
+ m.mu.Unlock()
+ }(ep)
+ }
+ wg.Wait()
+
+ m.mu.Lock()
+ m.lastCheck = time.Now()
+ m.mu.Unlock()
+}
+
+func probeURL(ctx context.Context, proxyURL, checkURL string, timeout time.Duration) (bool, int64, string) {
+ u, err := url.Parse(proxyURL)
+ if err != nil {
+ return false, 0, "bad proxy url"
+ }
+ tr := &http.Transport{
+ Proxy: http.ProxyURL(u),
+ TLSHandshakeTimeout: timeout,
+ ResponseHeaderTimeout: timeout,
+ DisableKeepAlives: true,
+ }
+ client := &http.Client{Transport: tr, Timeout: timeout}
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, checkURL, nil)
+ if err != nil {
+ return false, 0, err.Error()
+ }
+ start := time.Now()
+ resp, err := client.Do(req)
+ ms := time.Since(start).Milliseconds()
+ if err != nil {
+ return false, ms, trimErr(err)
+ }
+ defer resp.Body.Close()
+ _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
+ if resp.StatusCode >= 400 {
+ return false, ms, fmt.Sprintf("http %d", resp.StatusCode)
+ }
+ return true, ms, ""
+}
+
+// Candidates resolves the ordered route candidates for one site: its bound
+// entries (or subscriptions) filtered to healthy ones, sticky pick first.
+// It implements upstream.RouteSource.
+func (m *Manager) Candidates(site string) []upstream.RouteCandidate {
+ m.cfg.RLock()
+ enabled := m.cfg.ProxyPool.Enabled
+ var items []string
+ if st, ok := m.cfg.Sites[site]; ok && st != nil {
+ items = append(items, st.Proxies...)
+ }
+ m.cfg.RUnlock()
+ if !enabled || len(items) == 0 {
+ return nil
+ }
+
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ var all []*endpoint
+ for _, item := range items {
+ all = append(all, m.resolveItemLocked(item)...)
+ }
+ usable := make([]*endpoint, 0, len(all))
+ for _, ep := range all {
+ if ep.enabled && ep.proxyURL != "" && ep.parseErr == "" {
+ usable = append(usable, ep)
+ }
+ }
+ pick := make([]*endpoint, 0, len(usable))
+ for _, ep := range usable {
+ if !ep.checked || ep.healthy {
+ pick = append(pick, ep)
+ }
+ }
+ if len(pick) == 0 {
+ pick = usable
+ }
+ if len(pick) == 0 {
+ return nil
+ }
+ if chosen := m.sticky[site]; chosen != "" {
+ for i, ep := range pick {
+ if ep.name == chosen && i > 0 {
+ pick = append([]*endpoint{ep}, append(pick[:i], pick[i+1:]...)...)
+ break
+ }
+ }
+ }
+ m.sticky[site] = pick[0].name
+ out := make([]upstream.RouteCandidate, 0, len(pick))
+ for _, ep := range pick {
+ out = append(out, upstream.RouteCandidate{Name: ep.name, Proxy: ep.proxyURL})
+ }
+ return out
+}
+
+func (m *Manager) resolveItemLocked(item string) []*endpoint {
+ item = strings.TrimSpace(item)
+ if item == "" {
+ return nil
+ }
+ if name, ok := strings.CutPrefix(item, "sub:"); ok {
+ var out []*endpoint
+ for _, ep := range m.endpoints {
+ if ep.source == name {
+ out = append(out, ep)
+ }
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].name < out[j].name })
+ return out
+ }
+ if ep := m.endpoints[item]; ep != nil {
+ return []*endpoint{ep}
+ }
+ return nil
+}
+
+// Report records how a request through `proxy` went, so a failing endpoint
+// loses its sticky pick immediately and a healthy one is trusted again.
+func (m *Manager) Report(proxy string, ok bool) {
+ if m == nil || proxy == "" {
+ return
+ }
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ for _, ep := range m.endpoints {
+ if ep.proxyURL != proxy {
+ continue
+ }
+ ep.checked = true
+ ep.checkedAt = time.Now()
+ if ok {
+ ep.healthy = true
+ ep.lastErr = ""
+ } else {
+ ep.healthy = false
+ ep.lastErr = "request failed"
+ for site, name := range m.sticky {
+ if name == ep.name {
+ delete(m.sticky, site)
+ }
+ }
+ }
+ return
+ }
+}
+
+// Status is the console snapshot.
+type Status struct {
+ Enabled bool `json:"enabled"`
+ CheckIntervalS int `json:"check_interval_seconds"`
+ CheckTimeoutS int `json:"check_timeout_seconds"`
+ CheckURL string `json:"check_url"`
+ DefaultScheme string `json:"default_scheme"`
+ XrayPath string `json:"xray_path"`
+ XrayVersion string `json:"xray_version"`
+ XrayAutoDownload bool `json:"xray_auto_download"`
+ Xray XrayStatus `json:"xray"`
+ Entries []EntryStatus `json:"entries"`
+ Subscriptions []SubStatus `json:"subscriptions"`
+ Sticky map[string]string `json:"sticky"`
+ LastError string `json:"last_error,omitempty"`
+}
+
+type EntryStatus struct {
+ Name string `json:"name"`
+ Kind string `json:"kind"`
+ Scheme string `json:"scheme"`
+ Target string `json:"target"`
+ Source string `json:"source,omitempty"`
+ Enabled bool `json:"enabled"`
+ Healthy bool `json:"healthy"`
+ Checked bool `json:"checked"`
+ LatencyMs int64 `json:"latency_ms"`
+ LastErr string `json:"last_error,omitempty"`
+ ParseError string `json:"parse_error,omitempty"`
+ CheckedAt int64 `json:"checked_at"`
+}
+
+type SubStatus struct {
+ Name string `json:"name"`
+ URL string `json:"url"`
+ Enabled bool `json:"enabled"`
+ IntervalMinutes int `json:"interval_minutes"`
+ Nodes int `json:"nodes"`
+ FetchedAt int64 `json:"fetched_at"`
+ LastErr string `json:"last_error,omitempty"`
+}
+
+func (m *Manager) Status() Status {
+ m.cfg.RLock()
+ poolCfg := m.cfg.ProxyPool
+ subsCfg := make(map[string]config.ProxySubscription, len(poolCfg.Subscriptions))
+ for name, sc := range poolCfg.Subscriptions {
+ if sc != nil {
+ subsCfg[name] = *sc
+ }
+ }
+ m.cfg.RUnlock()
+
+ m.mu.Lock()
+ endpoints := make([]*endpoint, 0, len(m.endpoints))
+ for _, ep := range m.endpoints {
+ endpoints = append(endpoints, ep)
+ }
+ sticky := make(map[string]string, len(m.sticky))
+ for k, v := range m.sticky {
+ sticky[k] = v
+ }
+ subs := make(map[string]*subState, len(m.subs))
+ for name, st := range m.subs {
+ cp := *st
+ subs[name] = &cp
+ }
+ lastErr := m.lastErr
+ m.mu.Unlock()
+
+ st := Status{
+ Enabled: poolCfg.Enabled,
+ CheckIntervalS: poolCfg.CheckIntervalSeconds,
+ CheckTimeoutS: poolCfg.CheckTimeoutSeconds,
+ CheckURL: poolCfg.CheckURL,
+ DefaultScheme: poolCfg.DefaultScheme,
+ XrayPath: poolCfg.XrayPath,
+ XrayVersion: poolCfg.XrayVersion,
+ XrayAutoDownload: poolCfg.XrayAutoDownload,
+ Xray: m.xray.status(),
+ Sticky: sticky,
+ LastError: lastErr,
+ }
+ for _, ep := range endpoints {
+ es := EntryStatus{
+ Name: ep.name, Kind: ep.kind, Scheme: ep.scheme, Target: ep.display,
+ Source: ep.source, Enabled: ep.enabled, Healthy: ep.healthy,
+ Checked: ep.checked, LatencyMs: ep.latencyMs, LastErr: ep.lastErr,
+ ParseError: ep.parseErr,
+ }
+ if !ep.checkedAt.IsZero() {
+ es.CheckedAt = ep.checkedAt.Unix()
+ }
+ st.Entries = append(st.Entries, es)
+ }
+ sort.Slice(st.Entries, func(i, j int) bool {
+ a, b := st.Entries[i], st.Entries[j]
+ if (a.Source == "") != (b.Source == "") {
+ return a.Source == ""
+ }
+ if a.Source != b.Source {
+ return a.Source < b.Source
+ }
+ return a.Name < b.Name
+ })
+ for name, sc := range subsCfg {
+ ss := SubStatus{Name: name, URL: sc.URL, Enabled: sc.Enabled, IntervalMinutes: sc.IntervalMinutes}
+ if stt := subs[name]; stt != nil {
+ ss.Nodes = len(stt.nodes)
+ ss.LastErr = stt.lastErr
+ if !stt.fetchedAt.IsZero() {
+ ss.FetchedAt = stt.fetchedAt.Unix()
+ }
+ }
+ st.Subscriptions = append(st.Subscriptions, ss)
+ }
+ sort.Slice(st.Subscriptions, func(i, j int) bool { return st.Subscriptions[i].Name < st.Subscriptions[j].Name })
+ return st
+}
+
+func newEndpoint(name, link, defaultScheme, source string) *endpoint {
+ ep := &endpoint{name: name, source: source, link: strings.TrimSpace(link)}
+ node, err := ParseNode(ep.link, defaultScheme)
+ if err != nil {
+ ep.kind = "invalid"
+ ep.parseErr = err.Error()
+ return ep
+ }
+ ep.kind = node.Kind
+ ep.scheme = node.Scheme
+ ep.display = node.Display
+ ep.outbound = node.Outbound
+ ep.url = node.URL
+ return ep
+}
+
+func (m *Manager) setErr(err error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if err == nil {
+ m.lastErr = ""
+ return
+ }
+ m.lastErr = err.Error()
+}
+
+func shortHash(s string) string {
+ sum := sha256.Sum256([]byte(s))
+ return hex.EncodeToString(sum[:4])
+}
+
+func trimErr(err error) string {
+ s := err.Error()
+ if len(s) > 160 {
+ return s[:160] + "..."
+ }
+ return s
+}
diff --git a/internal/proxypool/manager_test.go b/internal/proxypool/manager_test.go
new file mode 100644
index 0000000..24aaa26
--- /dev/null
+++ b/internal/proxypool/manager_test.go
@@ -0,0 +1,76 @@
+package proxypool
+
+import (
+ "testing"
+
+ "github.com/nyoungo/dsfree2api/internal/config"
+)
+
+const examplePath = "../../config.example.toml"
+
+func TestCandidatesStickySwitchOnFailure(t *testing.T) {
+ cfg, err := config.Load(examplePath)
+ if err != nil {
+ t.Fatalf("load config: %v", err)
+ }
+ cfg.ProxyPool.Enabled = true
+ cfg.ProxyPool.Entries = map[string]*config.ProxyEntry{
+ "a": {Link: "http://127.0.0.1:9", Enabled: true},
+ "b": {Link: "socks5://127.0.0.1:10", Enabled: true},
+ }
+ cfg.Sites["de"].Proxies = []string{"a", "b"}
+
+ m := New(cfg, t.TempDir(), nil)
+ m.Reload()
+
+ cands := m.Candidates("de")
+ if len(cands) != 2 || cands[0].Name != "a" || cands[0].Proxy != "http://127.0.0.1:9" {
+ t.Fatalf("candidates = %+v", cands)
+ }
+
+ // Both healthy: the sticky pick stays a.
+ m.mu.Lock()
+ for _, ep := range m.endpoints {
+ ep.checked = true
+ ep.healthy = true
+ }
+ m.mu.Unlock()
+ cands = m.Candidates("de")
+ if cands[0].Name != "a" {
+ t.Fatalf("sticky pick changed without failure: %+v", cands)
+ }
+
+ // a fails: the request path must immediately switch to b.
+ m.Report("http://127.0.0.1:9", false)
+ cands = m.Candidates("de")
+ if len(cands) != 1 || cands[0].Name != "b" {
+ t.Fatalf("after failure = %+v", cands)
+ }
+
+ // Everything unhealthy: fall back to the bound order so requests still try.
+ m.Report("socks5://127.0.0.1:10", false)
+ cands = m.Candidates("de")
+ if len(cands) != 2 || cands[0].Name != "a" {
+ t.Fatalf("all-unhealthy fallback = %+v", cands)
+ }
+}
+
+func TestCandidatesDisabledOrUnbound(t *testing.T) {
+ cfg, err := config.Load(examplePath)
+ if err != nil {
+ t.Fatalf("load config: %v", err)
+ }
+ cfg.ProxyPool.Enabled = false
+ cfg.Sites["de"].Proxies = []string{"a"}
+ m := New(cfg, t.TempDir(), nil)
+ m.Reload()
+ if got := m.Candidates("de"); got != nil {
+ t.Fatalf("disabled pool returned %+v", got)
+ }
+
+ cfg.ProxyPool.Enabled = true
+ m.Reload()
+ if got := m.Candidates("de"); got != nil {
+ t.Fatalf("unbound/unknown entry returned %+v", got)
+ }
+}
diff --git a/internal/proxypool/parse.go b/internal/proxypool/parse.go
new file mode 100644
index 0000000..4b201cd
--- /dev/null
+++ b/internal/proxypool/parse.go
@@ -0,0 +1,517 @@
+// Package proxypool manages the egress endpoints used to reach the upstream
+// sites: Xray share links driven by a managed Xray core, plus plain
+// http/https/socks5 endpoints. Selection is sticky per site and health is
+// probed in the background.
+package proxypool
+
+import (
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net"
+ "net/url"
+ "strconv"
+ "strings"
+)
+
+// Node is a parsed proxy node.
+type Node struct {
+ Kind string // "xray" (share link) or "url" (plain endpoint)
+ Scheme string // vless / vmess / trojan / ss / http / https / socks5
+ Display string // host:port, for the console
+ Link string // original input line
+ URL string // Kind == "url": ready-to-use proxy URL
+ Outbound map[string]any // Kind == "xray": Xray outbound object (tag added later)
+}
+
+// ParseNode parses one line: an Xray share link (vless/vmess/trojan/ss) or a
+// plain endpoint. Accepted plain forms:
+//
+// scheme://user:pass@host:port
+// user:pass@host:port (scheme = defaultScheme)
+// host:port (scheme = defaultScheme)
+// host:port:user:pass (scheme = defaultScheme)
+func ParseNode(line, defaultScheme string) (*Node, error) {
+ line = strings.TrimSpace(line)
+ if line == "" {
+ return nil, errors.New("empty node")
+ }
+ switch {
+ case strings.HasPrefix(line, "vless://"):
+ return parseVless(line)
+ case strings.HasPrefix(line, "vmess://"):
+ return parseVmess(line)
+ case strings.HasPrefix(line, "trojan://"):
+ return parseTrojan(line)
+ case strings.HasPrefix(line, "ss://"):
+ return parseShadowsocks(line)
+ default:
+ return parsePlainEndpoint(line, defaultScheme)
+ }
+}
+
+func parseVless(raw string) (*Node, error) {
+ u, err := url.Parse(raw)
+ if err != nil {
+ return nil, fmt.Errorf("vless: %w", err)
+ }
+ id := u.User.Username()
+ host, port := u.Hostname(), u.Port()
+ if id == "" || host == "" || port == "" {
+ return nil, errors.New("vless: missing uuid / host / port")
+ }
+ q := u.Query()
+ user := map[string]any{"id": id, "encryption": "none"}
+ if flow := q.Get("flow"); flow != "" {
+ user["flow"] = flow
+ }
+ out := map[string]any{
+ "protocol": "vless",
+ "settings": map[string]any{
+ "vnext": []any{map[string]any{
+ "address": host, "port": portInt(port), "users": []any{user},
+ }},
+ },
+ "streamSettings": streamSettings(firstNonEmptyStr(q.Get("type"), "tcp"), q.Get("security"), q),
+ }
+ return &Node{Kind: "xray", Scheme: "vless", Display: net.JoinHostPort(host, port), Link: raw, Outbound: out}, nil
+}
+
+func parseVmess(raw string) (*Node, error) {
+ body := strings.TrimPrefix(raw, "vmess://")
+ if i := strings.IndexAny(body, "#"); i >= 0 {
+ body = body[:i]
+ }
+ dec, err := b64Decode(body)
+ if err != nil {
+ return nil, fmt.Errorf("vmess: invalid base64 payload: %w", err)
+ }
+ var v struct {
+ Add string `json:"add"`
+ Port any `json:"port"`
+ ID string `json:"id"`
+ Aid any `json:"aid"`
+ Scy string `json:"scy"`
+ Net string `json:"net"`
+ Type string `json:"type"`
+ Host string `json:"host"`
+ Path string `json:"path"`
+ TLS string `json:"tls"`
+ SNI string `json:"sni"`
+ FP string `json:"fp"`
+ ALPN string `json:"alpn"`
+ }
+ if err := json.Unmarshal([]byte(dec), &v); err != nil {
+ return nil, fmt.Errorf("vmess: bad json: %w", err)
+ }
+ port := toInt(v.Port)
+ if v.Add == "" || port == 0 || v.ID == "" {
+ return nil, errors.New("vmess: missing add / port / id")
+ }
+ security := strings.TrimSpace(v.Scy)
+ if security == "" {
+ security = "auto"
+ }
+ out := map[string]any{
+ "protocol": "vmess",
+ "settings": map[string]any{
+ "vnext": []any{map[string]any{
+ "address": v.Add, "port": port,
+ "users": []any{map[string]any{"id": v.ID, "alterId": toInt(v.Aid), "security": security}},
+ }},
+ },
+ }
+ network := strings.ToLower(strings.TrimSpace(v.Net))
+ if network == "" {
+ network = "tcp"
+ }
+ ss := map[string]any{"network": network}
+ switch network {
+ case "ws":
+ ws := map[string]any{}
+ if v.Path != "" {
+ ws["path"] = v.Path
+ }
+ if v.Host != "" {
+ ws["headers"] = map[string]any{"Host": v.Host}
+ }
+ ss["wsSettings"] = ws
+ case "h2":
+ h2 := map[string]any{}
+ if v.Path != "" {
+ h2["path"] = v.Path
+ }
+ if v.Host != "" {
+ h2["host"] = []any{v.Host}
+ }
+ ss["h2Settings"] = h2
+ case "grpc":
+ g := map[string]any{}
+ if v.Path != "" {
+ g["serviceName"] = v.Path
+ }
+ ss["grpcSettings"] = g
+ case "tcp":
+ if strings.EqualFold(v.Type, "http") {
+ req := map[string]any{"path": []any{firstNonEmptyStr(v.Path, "/")}}
+ if v.Host != "" {
+ req["headers"] = map[string]any{"Host": []any{v.Host}}
+ }
+ ss["tcpSettings"] = map[string]any{"header": map[string]any{"type": "http", "request": req}}
+ }
+ case "httpupgrade", "xhttp":
+ sub := map[string]any{}
+ if v.Path != "" {
+ sub["path"] = v.Path
+ }
+ if v.Host != "" {
+ sub["host"] = v.Host
+ }
+ ss[network+"Settings"] = sub
+ }
+ if strings.EqualFold(v.TLS, "tls") {
+ ss["security"] = "tls"
+ t := map[string]any{}
+ if sni := firstNonEmptyStr(v.SNI, v.Host); sni != "" {
+ t["serverName"] = sni
+ }
+ if v.FP != "" {
+ t["fingerprint"] = v.FP
+ }
+ if v.ALPN != "" {
+ t["alpn"] = strings.Split(v.ALPN, ",")
+ }
+ ss["tlsSettings"] = t
+ }
+ out["streamSettings"] = ss
+ return &Node{Kind: "xray", Scheme: "vmess", Display: net.JoinHostPort(v.Add, strconv.Itoa(port)), Link: raw, Outbound: out}, nil
+}
+
+func parseTrojan(raw string) (*Node, error) {
+ u, err := url.Parse(raw)
+ if err != nil {
+ return nil, fmt.Errorf("trojan: %w", err)
+ }
+ pass := u.User.Username()
+ host, port := u.Hostname(), u.Port()
+ if pass == "" || host == "" || port == "" {
+ return nil, errors.New("trojan: missing password / host / port")
+ }
+ q := u.Query()
+ security := q.Get("security")
+ if security == "" {
+ security = "tls"
+ }
+ out := map[string]any{
+ "protocol": "trojan",
+ "settings": map[string]any{
+ "servers": []any{map[string]any{"address": host, "port": portInt(port), "password": pass}},
+ },
+ "streamSettings": streamSettings(firstNonEmptyStr(q.Get("type"), "tcp"), security, q),
+ }
+ return &Node{Kind: "xray", Scheme: "trojan", Display: net.JoinHostPort(host, port), Link: raw, Outbound: out}, nil
+}
+
+func parseShadowsocks(raw string) (*Node, error) {
+ body := strings.TrimPrefix(raw, "ss://")
+ if i := strings.Index(body, "#"); i >= 0 {
+ body = body[:i]
+ }
+ var method, pass, hostport string
+ if at := strings.LastIndex(body, "@"); at >= 0 {
+ userinfo, hp := body[:at], body[at+1:]
+ if i := strings.IndexAny(hp, "?"); i >= 0 {
+ hp = hp[:i]
+ }
+ if strings.Contains(userinfo, ":") {
+ // plain method:password (uncommon)
+ } else {
+ dec, err := b64Decode(userinfo)
+ if err != nil {
+ return nil, errors.New("ss: cannot decode credentials")
+ }
+ userinfo = dec
+ }
+ mp := strings.SplitN(userinfo, ":", 2)
+ if len(mp) != 2 || mp[0] == "" || mp[1] == "" {
+ return nil, errors.New("ss: bad credentials")
+ }
+ method, pass, hostport = mp[0], mp[1], hp
+ } else {
+ // legacy: ss://base64(method:password@host:port)
+ dec, err := b64Decode(body)
+ if err != nil {
+ return nil, errors.New("ss: invalid base64 payload")
+ }
+ at := strings.LastIndex(dec, "@")
+ if at < 0 {
+ return nil, errors.New("ss: missing @host")
+ }
+ mp := strings.SplitN(dec[:at], ":", 2)
+ if len(mp) != 2 {
+ return nil, errors.New("ss: bad credentials")
+ }
+ method, pass, hostport = mp[0], mp[1], dec[at+1:]
+ }
+ host, port, err := splitHostPort(hostport)
+ if err != nil {
+ return nil, fmt.Errorf("ss: %w", err)
+ }
+ out := map[string]any{
+ "protocol": "shadowsocks",
+ "settings": map[string]any{
+ "servers": []any{map[string]any{"address": host, "port": port, "method": method, "password": pass}},
+ },
+ }
+ return &Node{Kind: "xray", Scheme: "ss", Display: net.JoinHostPort(host, strconv.Itoa(port)), Link: raw, Outbound: out}, nil
+}
+
+func parsePlainEndpoint(line, defaultScheme string) (*Node, error) {
+ if defaultScheme = strings.ToLower(strings.TrimSpace(defaultScheme)); defaultScheme == "" {
+ defaultScheme = "http"
+ }
+ build := func(scheme, host, port, user, pass string) (*Node, error) {
+ scheme = normalizeScheme(scheme)
+ if scheme == "" {
+ return nil, fmt.Errorf("unsupported proxy scheme %q (http/https/socks5)", scheme)
+ }
+ if host == "" || port == "" {
+ return nil, errors.New("missing host or port")
+ }
+ u := &url.URL{Scheme: scheme, Host: net.JoinHostPort(host, port)}
+ if user != "" {
+ u.User = url.UserPassword(user, pass)
+ }
+ return &Node{
+ Kind: "url", Scheme: scheme,
+ Display: net.JoinHostPort(host, port),
+ Link: line, URL: u.String(),
+ }, nil
+ }
+
+ switch {
+ case strings.Contains(line, "://"):
+ u, err := url.Parse(line)
+ if err != nil {
+ return nil, fmt.Errorf("bad proxy url: %w", err)
+ }
+ user, pass := "", ""
+ if u.User != nil {
+ user = u.User.Username()
+ pass, _ = u.User.Password()
+ }
+ return build(u.Scheme, u.Hostname(), u.Port(), user, pass)
+ case strings.Contains(line, "@"):
+ at := strings.LastIndex(line, "@")
+ userinfo, hostport := line[:at], line[at+1:]
+ mp := strings.SplitN(userinfo, ":", 2)
+ user, pass := mp[0], ""
+ if len(mp) == 2 {
+ pass = mp[1]
+ }
+ host, port, err := splitHostPort(hostport)
+ if err != nil {
+ return nil, err
+ }
+ return build(defaultScheme, host, strconv.Itoa(port), user, pass)
+ default:
+ parts := strings.Split(line, ":")
+ switch len(parts) {
+ case 2:
+ return build(defaultScheme, parts[0], parts[1], "", "")
+ case 4:
+ // ip:port:user:pass
+ return build(defaultScheme, parts[0], parts[1], parts[2], parts[3])
+ default:
+ return nil, fmt.Errorf("unrecognized endpoint %q", line)
+ }
+ }
+}
+
+// ParseSubscriptionBody turns a subscription response into node lines. The
+// body may be plain text (one node per line) or a base64 blob of that text.
+func ParseSubscriptionBody(raw []byte) []string {
+ s := strings.TrimSpace(string(raw))
+ if s == "" {
+ return nil
+ }
+ if !strings.Contains(s, "://") {
+ if dec, err := b64Decode(strings.ReplaceAll(s, "\n", "")); err == nil {
+ s = dec
+ }
+ }
+ out := make([]string, 0, 16)
+ for _, line := range strings.Split(s, "\n") {
+ line = strings.TrimSpace(line)
+ if line == "" || strings.HasPrefix(line, "#") {
+ continue
+ }
+ out = append(out, line)
+ }
+ return out
+}
+
+// ── helpers ──────────────────────────────────────────────────────
+
+// streamSettings builds Xray streamSettings from share-link query params.
+func streamSettings(network, security string, q url.Values) map[string]any {
+ network = strings.ToLower(strings.TrimSpace(network))
+ if network == "" {
+ network = "tcp"
+ }
+ ss := map[string]any{"network": network}
+ switch network {
+ case "ws":
+ ws := map[string]any{}
+ if p := q.Get("path"); p != "" {
+ ws["path"] = p
+ }
+ if h := q.Get("host"); h != "" {
+ ws["headers"] = map[string]any{"Host": h}
+ }
+ ss["wsSettings"] = ws
+ case "grpc":
+ g := map[string]any{}
+ if s := q.Get("serviceName"); s != "" {
+ g["serviceName"] = s
+ }
+ ss["grpcSettings"] = g
+ case "httpupgrade":
+ h := map[string]any{}
+ if p := q.Get("path"); p != "" {
+ h["path"] = p
+ }
+ if hh := q.Get("host"); hh != "" {
+ h["host"] = hh
+ }
+ ss["httpupgradeSettings"] = h
+ case "xhttp":
+ x := map[string]any{}
+ if p := q.Get("path"); p != "" {
+ x["path"] = p
+ }
+ if hh := q.Get("host"); hh != "" {
+ x["host"] = hh
+ }
+ if m := q.Get("mode"); m != "" {
+ x["mode"] = m
+ }
+ ss["xhttpSettings"] = x
+ case "tcp":
+ if strings.EqualFold(q.Get("headerType"), "http") {
+ req := map[string]any{}
+ if h := q.Get("host"); h != "" {
+ req["headers"] = map[string]any{"Host": []any{h}}
+ }
+ if p := q.Get("path"); p != "" {
+ req["path"] = []any{p}
+ }
+ ss["tcpSettings"] = map[string]any{"header": map[string]any{"type": "http", "request": req}}
+ }
+ }
+ switch strings.ToLower(security) {
+ case "tls":
+ t := map[string]any{}
+ if sni := firstNonEmptyStr(q.Get("sni"), q.Get("host")); sni != "" {
+ t["serverName"] = sni
+ }
+ if fp := q.Get("fp"); fp != "" {
+ t["fingerprint"] = fp
+ }
+ if alpn := q.Get("alpn"); alpn != "" {
+ t["alpn"] = strings.Split(alpn, ",")
+ }
+ ss["security"] = "tls"
+ ss["tlsSettings"] = t
+ case "reality":
+ r := map[string]any{}
+ if sni := q.Get("sni"); sni != "" {
+ r["serverName"] = sni
+ }
+ if fp := q.Get("fp"); fp != "" {
+ r["fingerprint"] = fp
+ }
+ if pbk := q.Get("pbk"); pbk != "" {
+ r["publicKey"] = pbk
+ }
+ if sid := q.Get("sid"); sid != "" {
+ r["shortId"] = sid
+ }
+ if spx := q.Get("spx"); spx != "" {
+ r["spiderX"] = spx
+ }
+ ss["security"] = "reality"
+ ss["realitySettings"] = r
+ }
+ return ss
+}
+
+func normalizeScheme(s string) string {
+ switch strings.ToLower(strings.TrimSpace(s)) {
+ case "http":
+ return "http"
+ case "https":
+ return "https"
+ case "socks", "socks5", "socks5h":
+ return "socks5"
+ default:
+ return ""
+ }
+}
+
+func b64Decode(s string) (string, error) {
+ s = strings.TrimSpace(s)
+ trimmed := strings.TrimRight(s, "=")
+ for _, enc := range []*base64.Encoding{
+ base64.RawURLEncoding, base64.URLEncoding,
+ base64.RawStdEncoding, base64.StdEncoding,
+ } {
+ if b, err := enc.DecodeString(trimmed); err == nil {
+ return string(b), nil
+ }
+ }
+ return "", errors.New("invalid base64")
+}
+
+func firstNonEmptyStr(vals ...string) string {
+ for _, v := range vals {
+ if strings.TrimSpace(v) != "" {
+ return v
+ }
+ }
+ return ""
+}
+
+func portInt(s string) int {
+ n, _ := strconv.Atoi(strings.TrimSpace(s))
+ return n
+}
+
+func toInt(v any) int {
+ switch x := v.(type) {
+ case float64:
+ return int(x)
+ case int:
+ return x
+ case int64:
+ return int(x)
+ case string:
+ n, _ := strconv.Atoi(strings.TrimSpace(x))
+ return n
+ default:
+ return 0
+ }
+}
+
+func splitHostPort(hostport string) (string, int, error) {
+ host, port, err := net.SplitHostPort(strings.TrimSpace(hostport))
+ if err != nil {
+ return "", 0, fmt.Errorf("bad host:port %q", hostport)
+ }
+ n, err := strconv.Atoi(port)
+ if err != nil || n < 1 || n > 65535 {
+ return "", 0, fmt.Errorf("bad port in %q", hostport)
+ }
+ return host, n, nil
+}
diff --git a/internal/proxypool/parse_test.go b/internal/proxypool/parse_test.go
new file mode 100644
index 0000000..125d92a
--- /dev/null
+++ b/internal/proxypool/parse_test.go
@@ -0,0 +1,145 @@
+package proxypool
+
+import (
+ "encoding/base64"
+ "testing"
+)
+
+func TestParseVless(t *testing.T) {
+ line := "vless://11111111-2222-3333-4444-555555555555@example.com:443" +
+ "?type=ws&security=tls&sni=sni.example.com&path=%2Fws&host=cdn.example.com&fp=chrome#node"
+ n, err := ParseNode(line, "http")
+ if err != nil {
+ t.Fatalf("parse: %v", err)
+ }
+ if n.Kind != "xray" || n.Scheme != "vless" || n.Display != "example.com:443" {
+ t.Fatalf("node = %+v", n)
+ }
+ if n.Outbound["protocol"] != "vless" {
+ t.Fatalf("protocol = %v", n.Outbound["protocol"])
+ }
+ settings := n.Outbound["settings"].(map[string]any)
+ vnext := settings["vnext"].([]any)[0].(map[string]any)
+ if vnext["address"] != "example.com" || vnext["port"] != 443 {
+ t.Fatalf("vnext = %+v", vnext)
+ }
+ users := vnext["users"].([]any)[0].(map[string]any)
+ if users["id"] != "11111111-2222-3333-4444-555555555555" || users["encryption"] != "none" {
+ t.Fatalf("users = %+v", users)
+ }
+ ss := n.Outbound["streamSettings"].(map[string]any)
+ if ss["network"] != "ws" || ss["security"] != "tls" {
+ t.Fatalf("stream = %+v", ss)
+ }
+ ws := ss["wsSettings"].(map[string]any)
+ if ws["path"] != "/ws" {
+ t.Fatalf("ws = %+v", ws)
+ }
+ tls := ss["tlsSettings"].(map[string]any)
+ if tls["serverName"] != "sni.example.com" {
+ t.Fatalf("tls = %+v", tls)
+ }
+}
+
+func TestParseVmess(t *testing.T) {
+ payload := `{"v":"2","ps":"n","add":"vm.example.com","port":"443","id":"uuid-1",` +
+ `"aid":"0","scy":"auto","net":"ws","type":"none","host":"vm.example.com",` +
+ `"path":"/vm","tls":"tls","sni":"vm.example.com"}`
+ line := "vmess://" + base64.StdEncoding.EncodeToString([]byte(payload))
+ n, err := ParseNode(line, "")
+ if err != nil {
+ t.Fatalf("parse: %v", err)
+ }
+ if n.Kind != "xray" || n.Scheme != "vmess" || n.Display != "vm.example.com:443" {
+ t.Fatalf("node = %+v", n)
+ }
+ ss := n.Outbound["streamSettings"].(map[string]any)
+ if ss["network"] != "ws" || ss["security"] != "tls" {
+ t.Fatalf("stream = %+v", ss)
+ }
+ if ss["wsSettings"].(map[string]any)["path"] != "/vm" {
+ t.Fatalf("ws = %+v", ss["wsSettings"])
+ }
+}
+
+func TestParseTrojan(t *testing.T) {
+ n, err := ParseNode("trojan://secret@tr.example.com:443?security=tls&sni=tr.example.com#t", "")
+ if err != nil {
+ t.Fatalf("parse: %v", err)
+ }
+ if n.Scheme != "trojan" || n.Display != "tr.example.com:443" {
+ t.Fatalf("node = %+v", n)
+ }
+ settings := n.Outbound["settings"].(map[string]any)
+ srv := settings["servers"].([]any)[0].(map[string]any)
+ if srv["address"] != "tr.example.com" || srv["port"] != 443 || srv["password"] != "secret" {
+ t.Fatalf("server = %+v", srv)
+ }
+}
+
+func TestParseShadowsocks(t *testing.T) {
+ // SIP002: base64(method:password)@host:port
+ ui := base64.RawURLEncoding.EncodeToString([]byte("aes-256-gcm:pass:word"))
+ n, err := ParseNode("ss://"+ui+"@ss.example.com:8388#x", "")
+ if err != nil {
+ t.Fatalf("sip002 parse: %v", err)
+ }
+ srv := n.Outbound["settings"].(map[string]any)["servers"].([]any)[0].(map[string]any)
+ if srv["method"] != "aes-256-gcm" || srv["password"] != "pass:word" || srv["address"] != "ss.example.com" {
+ t.Fatalf("server = %+v", srv)
+ }
+
+ // legacy: base64(method:password@host:port)
+ legacy := base64.RawURLEncoding.EncodeToString([]byte("aes-256-gcm:secret@ss.example.com:8388"))
+ n2, err := ParseNode("ss://"+legacy, "")
+ if err != nil {
+ t.Fatalf("legacy parse: %v", err)
+ }
+ srv2 := n2.Outbound["settings"].(map[string]any)["servers"].([]any)[0].(map[string]any)
+ if srv2["password"] != "secret" || srv2["address"] != "ss.example.com" || srv2["port"] != 8388 {
+ t.Fatalf("legacy server = %+v", srv2)
+ }
+}
+
+func TestParsePlainEndpoints(t *testing.T) {
+ cases := []struct {
+ line, def, wantURL, wantDisplay string
+ }{
+ {"socks5://u:p@1.2.3.4:1080", "http", "socks5://u:p@1.2.3.4:1080", "1.2.3.4:1080"},
+ {"http://1.2.3.4:8080", "socks5", "http://1.2.3.4:8080", "1.2.3.4:8080"},
+ {"1.2.3.4:8080", "http", "http://1.2.3.4:8080", "1.2.3.4:8080"},
+ {"1.2.3.4:8080", "socks5", "socks5://1.2.3.4:8080", "1.2.3.4:8080"},
+ {"u:p@1.2.3.4:1080", "http", "http://u:p@1.2.3.4:1080", "1.2.3.4:1080"},
+ {"1.2.3.4:8080:u:p", "http", "http://u:p@1.2.3.4:8080", "1.2.3.4:8080"},
+ }
+ for _, c := range cases {
+ n, err := ParseNode(c.line, c.def)
+ if err != nil {
+ t.Fatalf("%q: %v", c.line, err)
+ }
+ if n.Kind != "url" || n.URL != c.wantURL || n.Display != c.wantDisplay {
+ t.Errorf("%q => kind=%s url=%q display=%q", c.line, n.Kind, n.URL, n.Display)
+ }
+ }
+}
+
+func TestParseSubscriptionBody(t *testing.T) {
+ plain := "vless://a@b:1\n# comment\nsocks5://x:y@1.2.3.4:5\n\n"
+ got := ParseSubscriptionBody([]byte(plain))
+ if len(got) != 2 || got[0] != "vless://a@b:1" || got[1] != "socks5://x:y@1.2.3.4:5" {
+ t.Fatalf("plain parse = %v", got)
+ }
+ enc := base64.StdEncoding.EncodeToString([]byte(plain))
+ got2 := ParseSubscriptionBody([]byte(enc))
+ if len(got2) != 2 || got2[0] != got[0] {
+ t.Fatalf("base64 parse = %v", got2)
+ }
+}
+
+func TestParseNodeRejectsJunk(t *testing.T) {
+ for _, bad := range []string{"", "hello world", "vless://no-port", "ss://@@@", "1.2.3.4", "ftp://1.2.3.4:21"} {
+ if _, err := ParseNode(bad, "http"); err == nil {
+ t.Errorf("%q should fail to parse", bad)
+ }
+ }
+}
diff --git a/internal/proxypool/xray.go b/internal/proxypool/xray.go
new file mode 100644
index 0000000..67d4457
--- /dev/null
+++ b/internal/proxypool/xray.go
@@ -0,0 +1,498 @@
+package proxypool
+
+import (
+ "archive/zip"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net"
+ "net/http"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "sync"
+ "time"
+)
+
+// xrayNode is one Xray-backed endpoint, ready to be written into the core
+// config.
+type xrayNode struct {
+ Name string
+ Outbound map[string]any
+}
+
+// xrayCore owns the managed Xray process: one socks inbound per pool node,
+// each routed to its own outbound.
+type xrayCore struct {
+ dir string
+ log *slog.Logger
+
+ mu sync.Mutex
+ path string
+ proc *exec.Cmd
+ exited chan struct{}
+ running bool
+ pid int
+ cfgPath string
+ cfgHash string
+ ports map[string]int
+ lastErr string
+ tail *logTail
+}
+
+// XrayStatus is the console view of the core.
+type XrayStatus struct {
+ Path string `json:"path"`
+ Running bool `json:"running"`
+ PID int `json:"pid"`
+ Error string `json:"error,omitempty"`
+}
+
+func newXrayCore(dir string, log *slog.Logger) *xrayCore {
+ if log == nil {
+ log = slog.Default()
+ }
+ return &xrayCore{dir: dir, log: log, ports: map[string]int{}}
+}
+
+func (c *xrayCore) status() XrayStatus {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return XrayStatus{Path: c.path, Running: c.running, PID: c.pid, Error: c.lastErr}
+}
+
+// ensure resolves the Xray binary: explicit path, then a previously downloaded
+// copy, then (optionally) a fresh download. The download runs unlocked so
+// status reads never stall behind a slow network.
+func (c *xrayCore) ensure(explicitPath string, autoDownload bool, version string) error {
+ c.mu.Lock()
+ if c.path != "" {
+ if _, err := os.Stat(c.path); err == nil {
+ c.mu.Unlock()
+ return nil
+ }
+ c.path = ""
+ }
+ if p := strings.TrimSpace(explicitPath); p != "" {
+ if _, err := os.Stat(p); err != nil {
+ c.mu.Unlock()
+ return fmt.Errorf("xray_path: %w", err)
+ }
+ c.path = p
+ c.mu.Unlock()
+ return nil
+ }
+ target := filepath.Join(c.dir, "xray"+exeSuffix())
+ if _, err := os.Stat(target); err == nil {
+ c.path = target
+ c.mu.Unlock()
+ return nil
+ }
+ if !autoDownload {
+ c.mu.Unlock()
+ return errors.New("xray binary not found — set [proxypool].xray_path, or enable xray_auto_download")
+ }
+ c.mu.Unlock()
+
+ if err := c.download(target, version); err != nil {
+ c.mu.Lock()
+ c.lastErr = err.Error()
+ c.mu.Unlock()
+ return err
+ }
+ c.mu.Lock()
+ c.path = target
+ c.mu.Unlock()
+ return nil
+}
+
+func (c *xrayCore) download(target, version string) error {
+ if err := os.MkdirAll(c.dir, 0o755); err != nil {
+ return err
+ }
+ osName := map[string]string{"windows": "windows", "linux": "linux", "darwin": "macos"}[runtime.GOOS]
+ arch := map[string]string{"amd64": "64", "arm64": "arm64-v8a", "386": "32", "arm": "arm32-v7a"}[runtime.GOARCH]
+ if osName == "" || arch == "" {
+ return fmt.Errorf("xray: unsupported platform %s/%s", runtime.GOOS, runtime.GOARCH)
+ }
+ asset := fmt.Sprintf("Xray-%s-%s.zip", osName, arch)
+ downloadURL := "https://github.com/XTLS/Xray-core/releases/latest/download/" + asset
+ if v := strings.TrimSpace(version); v != "" {
+ if !strings.HasPrefix(v, "v") {
+ v = "v" + v
+ }
+ downloadURL = fmt.Sprintf("https://github.com/XTLS/Xray-core/releases/download/%s/%s", v, asset)
+ }
+ c.log.Info("downloading xray core", "asset", asset)
+ zipPath := target + ".download"
+ if err := httpDownload(downloadURL, zipPath, 180*time.Second); err != nil {
+ return err
+ }
+ defer os.Remove(zipPath)
+ if want, err := fetchDigest(downloadURL + ".dgst"); err != nil {
+ c.log.Warn("xray checksum unavailable, skipping verification", "error", err)
+ } else if want != "" {
+ got, err := fileSHA256(zipPath)
+ if err != nil {
+ return err
+ }
+ if !strings.EqualFold(got, want) {
+ return fmt.Errorf("xray checksum mismatch: got %s want %s", got, want)
+ }
+ }
+ if err := extractZipBinary(zipPath, target); err != nil {
+ return err
+ }
+ _ = os.Chmod(target, 0o755)
+ c.log.Info("xray core ready", "path", target)
+ return nil
+}
+
+// configure regenerates the core config for the given nodes and restarts the
+// process when something changed. Ports are kept stable across restarts so
+// cached cookies stay valid.
+func (c *xrayCore) configure(nodes []xrayNode) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if len(nodes) == 0 {
+ c.stopLocked()
+ c.cfgHash = ""
+ return nil
+ }
+ if c.path == "" {
+ return errors.New("xray binary unavailable")
+ }
+ for _, n := range nodes {
+ if c.ports[n.Name] == 0 {
+ p, err := freePort()
+ if err != nil {
+ return err
+ }
+ c.ports[n.Name] = p
+ }
+ }
+ if c.configHash(nodes) == c.cfgHash && c.running {
+ return nil
+ }
+ // A restart is due. Stop first so our own listeners do not count as
+ // "port busy", then reclaim or reassign every assigned port — another
+ // local core may have raced us for one, since the OS only guaranteed it
+ // at allocation time.
+ c.stopLocked()
+ for _, n := range nodes {
+ if !portAvailable(c.ports[n.Name]) {
+ p, err := freePort()
+ if err != nil {
+ return err
+ }
+ c.ports[n.Name] = p
+ }
+ }
+ raw, err := json.Marshal(buildXrayConfig(nodes, c.ports))
+ if err != nil {
+ return err
+ }
+ sum := sha256.Sum256(raw)
+ c.cfgHash = hex.EncodeToString(sum[:])
+ cfgPath := filepath.Join(c.dir, "config.json")
+ if err := os.MkdirAll(c.dir, 0o755); err != nil {
+ return err
+ }
+ if err := os.WriteFile(cfgPath, raw, 0o600); err != nil {
+ return err
+ }
+ c.cfgPath = cfgPath
+ return c.startLocked(nodes)
+}
+
+// configHash is the digest of the config that would be generated for nodes
+// with the current port assignments.
+func (c *xrayCore) configHash(nodes []xrayNode) string {
+ raw, err := json.Marshal(buildXrayConfig(nodes, c.ports))
+ if err != nil {
+ return ""
+ }
+ sum := sha256.Sum256(raw)
+ return hex.EncodeToString(sum[:])
+}
+
+// portAvailable reports whether 127.0.0.1:port can still be claimed.
+func portAvailable(port int) bool {
+ if port <= 0 {
+ return false
+ }
+ l, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port))
+ if err != nil {
+ return false
+ }
+ l.Close()
+ return true
+}
+
+func (c *xrayCore) startLocked(nodes []xrayNode) error {
+ cmd := exec.Command(c.path, "run", "-c", c.cfgPath)
+ c.tail = &logTail{max: 40}
+ cmd.Stdout = c.tail
+ cmd.Stderr = c.tail
+ if err := cmd.Start(); err != nil {
+ c.lastErr = "start xray: " + err.Error()
+ return errors.New(c.lastErr)
+ }
+ c.proc = cmd
+ c.pid = cmd.Process.Pid
+ c.running = true
+ c.lastErr = ""
+ exited := make(chan struct{})
+ c.exited = exited
+ go func() {
+ _ = cmd.Wait()
+ c.mu.Lock()
+ if c.proc == cmd {
+ c.running = false
+ c.pid = 0
+ if t := c.tail.Last(); t != "" {
+ c.lastErr = "xray exited: " + t
+ } else {
+ c.lastErr = "xray exited"
+ }
+ }
+ c.mu.Unlock()
+ close(exited)
+ }()
+ // Wait until the local socks inbounds answer.
+ deadline := time.Now().Add(8 * time.Second)
+ for time.Now().Before(deadline) {
+ ok := true
+ for _, n := range nodes {
+ conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", c.ports[n.Name]), 300*time.Millisecond)
+ if err != nil {
+ ok = false
+ break
+ }
+ conn.Close()
+ }
+ if ok {
+ return nil
+ }
+ if !c.running {
+ break
+ }
+ time.Sleep(200 * time.Millisecond)
+ }
+ if !c.running {
+ if c.lastErr == "" {
+ c.lastErr = "xray failed to start"
+ }
+ return errors.New(c.lastErr)
+ }
+ c.lastErr = "xray socks ports did not come up in time"
+ return errors.New(c.lastErr)
+}
+
+func (c *xrayCore) stopLocked() {
+ if c.proc != nil && c.proc.Process != nil {
+ _ = c.proc.Process.Kill()
+ }
+ if c.exited != nil {
+ select {
+ case <-c.exited:
+ case <-time.After(3 * time.Second):
+ }
+ }
+ c.proc = nil
+ c.exited = nil
+ c.running = false
+ c.pid = 0
+}
+
+func (c *xrayCore) stop() {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.stopLocked()
+}
+
+// port returns the local socks port assigned to a node name.
+func (c *xrayCore) port(name string) int {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.ports[name]
+}
+
+func buildXrayConfig(nodes []xrayNode, ports map[string]int) map[string]any {
+ inbounds := make([]any, 0, len(nodes))
+ outbounds := make([]any, 0, len(nodes)+2)
+ rules := make([]any, 0, len(nodes))
+ for i, n := range nodes {
+ inTag := fmt.Sprintf("in-%d", i)
+ outTag := fmt.Sprintf("out-%d", i)
+ inbounds = append(inbounds, map[string]any{
+ "tag": inTag, "listen": "127.0.0.1", "port": ports[n.Name],
+ "protocol": "socks",
+ "settings": map[string]any{"udp": false, "auth": "noauth"},
+ })
+ ob := make(map[string]any, len(n.Outbound)+1)
+ for k, v := range n.Outbound {
+ ob[k] = v
+ }
+ ob["tag"] = outTag
+ outbounds = append(outbounds, ob)
+ rules = append(rules, map[string]any{"type": "field", "inboundTag": []any{inTag}, "outboundTag": outTag})
+ }
+ outbounds = append(outbounds,
+ map[string]any{"tag": "direct", "protocol": "freedom"},
+ map[string]any{"tag": "blocked", "protocol": "blackhole"},
+ )
+ return map[string]any{
+ "log": map[string]any{"loglevel": "warning", "access": "none"},
+ "inbounds": inbounds,
+ "outbounds": outbounds,
+ "routing": map[string]any{"domainStrategy": "AsIs", "rules": rules},
+ }
+}
+
+// ── platform / io helpers ────────────────────────────────────────
+
+func exeSuffix() string {
+ if runtime.GOOS == "windows" {
+ return ".exe"
+ }
+ return ""
+}
+
+func freePort() (int, error) {
+ l, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ return 0, err
+ }
+ defer l.Close()
+ return l.Addr().(*net.TCPAddr).Port, nil
+}
+
+func httpDownload(url, dst string, timeout time.Duration) error {
+ client := &http.Client{Timeout: timeout}
+ resp, err := client.Get(url)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("download %s: http %d", url, resp.StatusCode)
+ }
+ f, err := os.Create(dst)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ if _, err := io.Copy(f, io.LimitReader(resp.Body, 256<<20)); err != nil {
+ return err
+ }
+ return f.Sync()
+}
+
+func fetchDigest(url string) (string, error) {
+ client := &http.Client{Timeout: 20 * time.Second}
+ resp, err := client.Get(url)
+ if err != nil {
+ return "", err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return "", fmt.Errorf("http %d", resp.StatusCode)
+ }
+ raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+ if err != nil {
+ return "", err
+ }
+ for _, line := range strings.Split(string(raw), "\n") {
+ key, val, ok := strings.Cut(line, ":")
+ if !ok {
+ continue
+ }
+ if strings.Contains(strings.ToUpper(key), "SHA2-256") || strings.Contains(strings.ToUpper(key), "SHA256") {
+ if v := strings.TrimSpace(strings.Fields(val)[0]); v != "" {
+ return v, nil
+ }
+ }
+ }
+ return "", errors.New("no sha256 digest found")
+}
+
+func fileSHA256(path string) (string, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return "", err
+ }
+ defer f.Close()
+ h := sha256.New()
+ if _, err := io.Copy(h, f); err != nil {
+ return "", err
+ }
+ return hex.EncodeToString(h.Sum(nil)), nil
+}
+
+func extractZipBinary(zipPath, target string) error {
+ r, err := zip.OpenReader(zipPath)
+ if err != nil {
+ return fmt.Errorf("open archive: %w", err)
+ }
+ defer r.Close()
+ want := "xray" + exeSuffix()
+ for _, f := range r.File {
+ if !strings.EqualFold(filepath.Base(f.Name), want) {
+ continue
+ }
+ rc, err := f.Open()
+ if err != nil {
+ return err
+ }
+ defer rc.Close()
+ out, err := os.OpenFile(target, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o755)
+ if err != nil {
+ return err
+ }
+ if _, err := io.Copy(out, rc); err != nil {
+ out.Close()
+ return err
+ }
+ return out.Close()
+ }
+ return errors.New("no xray binary inside the archive")
+}
+
+// logTail keeps the last few process log lines for status reporting.
+type logTail struct {
+ mu sync.Mutex
+ lines []string
+ max int
+}
+
+func (t *logTail) Write(p []byte) (int, error) {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ for _, line := range strings.Split(strings.TrimRight(string(p), "\r\n"), "\n") {
+ line = strings.TrimSpace(line)
+ if line == "" {
+ continue
+ }
+ t.lines = append(t.lines, line)
+ }
+ if len(t.lines) > t.max {
+ t.lines = t.lines[len(t.lines)-t.max:]
+ }
+ return len(p), nil
+}
+
+func (t *logTail) Last() string {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ if len(t.lines) == 0 {
+ return ""
+ }
+ return t.lines[len(t.lines)-1]
+}
diff --git a/internal/proxypool/xray_smoke_test.go b/internal/proxypool/xray_smoke_test.go
new file mode 100644
index 0000000..e71bed9
--- /dev/null
+++ b/internal/proxypool/xray_smoke_test.go
@@ -0,0 +1,80 @@
+package proxypool
+
+import (
+ "net"
+ "os"
+ "strings"
+ "testing"
+
+ "github.com/nyoungo/dsfree2api/internal/config"
+)
+
+// TestXrayCoreSmoke drives the managed core against the locally placed
+// binary (data/xray.exe) with a syntactically real VLESS node. It never
+// connects upstream — it only proves that ensure() picks up the local
+// binary, configure() starts the process and the local socks inbound is
+// live. Skipped unless opted in:
+//
+// $env:DSFREE_XRAY_SMOKE=1; go test ./internal/proxypool -run Smoke -v
+func TestXrayCoreSmoke(t *testing.T) {
+ if os.Getenv("DSFREE_XRAY_SMOKE") == "" {
+ t.Skip("set DSFREE_XRAY_SMOKE=1 to run the local xray smoke test")
+ }
+ cfg, err := config.Load(examplePath)
+ if err != nil {
+ t.Fatalf("load config: %v", err)
+ }
+ cfg.ProxyPool.Enabled = true
+ cfg.ProxyPool.XrayAutoDownload = false // the local binary must be used as-is
+ cfg.ProxyPool.Entries = map[string]*config.ProxyEntry{
+ "smoke": {
+ Link: "vless://11111111-2222-3333-4444-555555555555@example.com:443" +
+ "?security=tls&type=tcp&sni=example.com",
+ Enabled: true,
+ },
+ }
+ cfg.Sites["de"].Proxies = []string{"smoke"}
+
+ m := New(cfg, "../../data", nil)
+ m.Reload()
+ st := m.Status()
+ if !st.Xray.Running {
+ t.Fatalf("xray core did not start: %+v", st.Xray)
+ }
+ t.Logf("xray running: pid=%d path=%s", st.Xray.PID, st.Xray.Path)
+
+ cands := m.Candidates("de")
+ if len(cands) != 1 || !strings.HasPrefix(cands[0].Proxy, "socks5://127.0.0.1:") {
+ t.Fatalf("candidates = %+v", cands)
+ }
+ t.Logf("candidate proxy: %s", cands[0].Proxy)
+
+ m.xray.stop()
+ if m.Status().Xray.Running {
+ t.Fatal("xray still running after stop")
+ }
+
+ // Port stability: a restart re-claims the same socks port, so cached
+ // cookies (keyed by the local proxy URL) stay valid.
+ m.Reload()
+ again := m.Candidates("de")
+ if len(again) != 1 || again[0].Proxy != cands[0].Proxy {
+ t.Fatalf("port not stable across restart: %q -> %+v", cands[0].Proxy, again)
+ }
+ m.xray.stop()
+
+ // Race self-heal: steal the assigned port; the next start must move away.
+ port := strings.TrimPrefix(cands[0].Proxy, "socks5://127.0.0.1:")
+ ln, err := net.Listen("tcp", "127.0.0.1:"+port)
+ if err != nil {
+ t.Fatalf("occupy port %s: %v", port, err)
+ }
+ defer ln.Close()
+ m.Reload()
+ third := m.Candidates("de")
+ if len(third) != 1 || third[0].Proxy == cands[0].Proxy {
+ t.Fatalf("stolen port was not reassigned: %+v", third)
+ }
+ t.Logf("port moved %s -> %s", cands[0].Proxy, third[0].Proxy)
+ m.xray.stop()
+}
diff --git a/internal/upstream/client.go b/internal/upstream/client.go
index d068935..6c68c85 100644
--- a/internal/upstream/client.go
+++ b/internal/upstream/client.go
@@ -33,6 +33,19 @@ func (r Route) Key() string {
return r.Proxy
}
+// RouteCandidate is one egress route supplied by the proxy pool.
+type RouteCandidate struct {
+ Name string
+ Proxy string
+}
+
+// RouteSource resolves per-site route candidates and records request
+// outcomes. Implemented by proxypool.Manager; nil when the pool is unused.
+type RouteSource interface {
+ Candidates(site string) []RouteCandidate
+ Report(proxy string, ok bool)
+}
+
type chatConfig struct {
BotID int
PostID int
@@ -42,9 +55,10 @@ type chatConfig struct {
}
type Client struct {
- cfg *config.Config
- ts *turnstile.Solver
- log *slog.Logger
+ cfg *config.Config
+ ts *turnstile.Solver
+ log *slog.Logger
+ pool RouteSource
stateMu sync.Mutex
chatCfg map[string]chatConfig
@@ -64,7 +78,7 @@ type Client struct {
prompt string, route Route, info *ServeInfo, yield func(Event) error) error
}
-func New(cfg *config.Config, ts *turnstile.Solver, log *slog.Logger) *Client {
+func New(cfg *config.Config, ts *turnstile.Solver, log *slog.Logger, pool RouteSource) *Client {
if log == nil {
log = slog.Default()
}
@@ -72,6 +86,7 @@ func New(cfg *config.Config, ts *turnstile.Solver, log *slog.Logger) *Client {
cfg: cfg,
ts: ts,
log: log,
+ pool: pool,
chatCfg: map[string]chatConfig{},
cfgLocks: map[string]*sync.Mutex{},
gates: map[string]chan struct{}{},
@@ -239,6 +254,40 @@ func (c *Client) logCacheReject(site, modelID, prompt string) {
"head", strconv.Quote(head), "tail", strconv.Quote(tail))
}
+// routesFor returns the ordered candidate routes for one site: proxy-pool
+// candidates first (sticky, health filtered), then the global primary and
+// fallback lines as the safety net.
+func (c *Client) routesFor(siteCode string) []Route {
+ out := make([]Route, 0, 4)
+ seen := map[string]bool{}
+ add := func(r Route) {
+ key := r.Key()
+ if seen[key] {
+ return
+ }
+ seen[key] = true
+ out = append(out, r)
+ }
+ if c.pool != nil {
+ for _, cand := range c.pool.Candidates(siteCode) {
+ add(Route{Name: cand.Name, Proxy: cand.Proxy})
+ }
+ }
+ for _, r := range c.Routes() {
+ add(r)
+ }
+ return out
+}
+
+// poolReport feeds a route outcome back to the proxy pool; it is a no-op for
+// global routes and the direct connection.
+func (c *Client) poolReport(proxy string, ok bool) {
+ if c.pool == nil || proxy == "" {
+ return
+ }
+ c.pool.Report(proxy, ok)
+}
+
// Chat streams the prompt through the upstream site, honouring route
// failover, slow-start detection, session refresh and cross-site failover.
func (c *Client) Chat(ctx context.Context, modelID string, prompt string, info *ServeInfo, yield func(Event) error) error {
@@ -355,7 +404,7 @@ func (c *Client) chatWithRetries(ctx context.Context, modelID string, model conf
if !site.Enabled {
return errf("site %s is disabled", model.Site)
}
- routes := c.Routes()
+ routes := c.routesFor(site.Code)
var lastErr error
sawQuota := false
@@ -369,6 +418,7 @@ func (c *Client) chatWithRetries(ctx context.Context, modelID string, model conf
c.clearQuotaCool(site.Code)
c.probeBalanceAsync(site, model, route)
}
+ c.poolReport(route.Proxy, true)
return nil
}
lastErr = err
@@ -398,6 +448,11 @@ func (c *Client) chatWithRetries(ctx context.Context, modelID string, model conf
if isCacheEmptyProblem(err) {
return AsUpstream(err)
}
+ // Quota/session failures are site-side, not proxy-side — do not
+ // blame the egress endpoint for them.
+ if !isQuota(err) && !isSession(err) {
+ c.poolReport(route.Proxy, false)
+ }
failed = append(failed, route)
c.logUpstreamErr(site.Code, modelID, attempt, route, err)
if isQuota(err) {
diff --git a/internal/upstream/client_test.go b/internal/upstream/client_test.go
index fb445b2..8e1d3df 100644
--- a/internal/upstream/client_test.go
+++ b/internal/upstream/client_test.go
@@ -32,7 +32,7 @@ func TestRoutesKeepPrimaryFirstAndDedupe(t *testing.T) {
cfg.Proxy.URL = ""
cfg.Proxy.FallbackURLs = []string{"socks5://fallback:1", "socks5://fallback:1"}
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
got := c.Routes()
want := []Route{{Name: "primary", Proxy: ""}, {Name: "fallback-1", Proxy: "socks5://fallback:1"}}
if !reflect.DeepEqual(got, want) {
@@ -47,7 +47,7 @@ func TestSlowPrimarySwitchesToFallbackBeforeContent(t *testing.T) {
cfg.Proxy.SlowStartSeconds = 0.02
var calls []string
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model,
prompt string, route Route, info *ServeInfo, yield func(Event) error) error {
calls = append(calls, route.Name)
@@ -86,7 +86,7 @@ func TestPrimaryErrorSwitchesToFallbackBeforeContent(t *testing.T) {
cfg.Proxy.FallbackURLs = []string{"socks5://fallback:1"}
var calls []string
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model,
prompt string, route Route, info *ServeInfo, yield func(Event) error) error {
calls = append(calls, route.Name)
@@ -119,7 +119,7 @@ func TestNoFallbackAfterContentStarted(t *testing.T) {
cfg.Proxy.FallbackURLs = []string{"socks5://fallback:1"}
var calls []string
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model,
prompt string, route Route, info *ServeInfo, yield func(Event) error) error {
calls = append(calls, route.Name)
@@ -211,7 +211,7 @@ func TestTurnstileRequiredWhileDisabledFailsFast(t *testing.T) {
cfg.Upstream.RefreshRetries = 4
var calls int
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model,
prompt string, route Route, info *ServeInfo, yield func(Event) error) error {
calls++
@@ -237,7 +237,7 @@ func TestNonceFetchFailureFailsFast(t *testing.T) {
cfg.Upstream.RefreshRetries = 4
var calls int
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model,
prompt string, route Route, info *ServeInfo, yield func(Event) error) error {
calls++
@@ -273,7 +273,7 @@ func TestSSEDeltaAndDoneEvents(t *testing.T) {
func TestQuotaSiteCoolsDownAndPrefersSibling(t *testing.T) {
cfg := testConfig(t)
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
calls := map[string]int{}
c.chatOnceOverride = func(_ context.Context, site config.Site, modelID string, model config.Model,
_ string, _ Route, _ *ServeInfo, _ func(Event) error) error {
@@ -318,7 +318,7 @@ func TestCacheEmptyFailsOverToMirrorWithoutRefreshCycles(t *testing.T) {
cfg := testConfig(t)
cfg.Upstream.AutoRefresh = true
cfg.Upstream.RefreshRetries = 3
- c := New(cfg, nil, nil)
+ c := New(cfg, nil, nil, nil)
calls := map[string]int{}
c.chatOnceOverride = func(_ context.Context, site config.Site, modelID string, model config.Model,
_ string, _ Route, _ *ServeInfo, _ func(Event) error) error {
@@ -348,7 +348,7 @@ func TestQuotaRotationRecoversWithoutSolve(t *testing.T) {
if _, _, err := solver.ImportCookies("de", "cf_clearance=abc; dsgt_gid="+oldGid, ""); err != nil {
t.Fatalf("seed cookies: %v", err)
}
- c := New(cfg, solver, nil)
+ c := New(cfg, solver, nil, nil)
calls := 0
c.chatOnceOverride = func(_ context.Context, site config.Site, modelID string, model config.Model,
_ string, _ Route, _ *ServeInfo, _ func(Event) error) error {