[优化] 增强 WebSocket 处理逻辑，添加上下文取消支持和关闭机制

[优化] 重构 WebSocket 处理逻辑，添加消息处理接口和心跳机制

README.zh-CN.md

@@ -112,7 +112,7 @@ Docker 部署可直接运行 Agent 镜像：
 docker pull ghcr.io/rain-kl/openflare-agent:latest
 docker rm -f openflare-agent 2>/dev/null || true
 docker run -d --name openflare-agent --restart unless-stopped \
-  -p 80:80 -p 443:443 \
+  -p 80:80 -p 443:443/tcp -p 443:443/udp \
   -e OPENFLARE_SERVER_URL=http://your-server:3000 \
   -e OPENFLARE_AGENT_TOKEN=YOUR_AGENT_TOKEN \
   ghcr.io/rain-kl/openflare-agent:latest

docs/deployment/agent.md

@@ -102,7 +102,7 @@ Docker 部署时直接运行内置 OpenResty 的 Agent 镜像：
 docker pull ghcr.io/rain-kl/openflare-agent:latest
 docker rm -f openflare-agent 2>/dev/null || true
 docker run -d --name openflare-agent --restart unless-stopped \
-  -p 80:80 -p 443:443 \
+  -p 80:80 -p 443:443/tcp -p 443:443/udp \
   -e OPENFLARE_SERVER_URL=http://your-server:3000 \
   -e OPENFLARE_AGENT_TOKEN=YOUR_AGENT_TOKEN \
   ghcr.io/rain-kl/openflare-agent:latest

docs/deployment/deployment.md

@@ -168,7 +168,7 @@ Docker 部署是 Agent 推荐的部署方式。Docker 部署时直接运行 Agen
 docker pull ghcr.io/rain-kl/openflare-agent:latest
 docker rm -f openflare-agent 2>/dev/null || true
 docker run -d --name openflare-agent --restart unless-stopped \
-  -p 80:80 -p 443:443 \
+  -p 80:80 -p 443:443/tcp -p 443:443/udp \
   -v openflare-agent-data:/data \
   -v ./agent.json:/etc/openflare/agent.json:ro \
   ghcr.io/rain-kl/openflare-agent:latest
@@ -180,7 +180,7 @@ docker run -d --name openflare-agent --restart unless-stopped \
 docker pull ghcr.io/rain-kl/openflare-agent:latest
 docker rm -f openflare-agent 2>/dev/null || true
 docker run -d --name openflare-agent --restart unless-stopped \
-  -p 80:80 -p 443:443 \
+  -p 80:80 -p 443:443/tcp -p 443:443/udp \
   -e OPENFLARE_SERVER_URL=http://your-server:3000 \
   -e OPENFLARE_AGENT_TOKEN=YOUR_AGENT_TOKEN \
   ghcr.io/rain-kl/openflare-agent:latest

docs/guide/quick-start.md

@@ -119,7 +119,7 @@ Agent 部署方式推荐使用 Docker 部署（即直接运行内置 OpenResty
 docker pull ghcr.io/rain-kl/openflare-agent:latest
 docker rm -f openflare-agent 2>/dev/null || true
 docker run -d --name openflare-agent --restart unless-stopped \
-  -p 80:80 -p 443:443 \
+  -p 80:80 -p 443:443/tcp -p 443:443/udp \
   -v openflare-agent-data:/data \
   -e OPENFLARE_SERVER_URL=http://your-server:3000 \
   -e OPENFLARE_AGENT_TOKEN=YOUR_AGENT_TOKEN \

docs/guide/waf-ip-group-expr.md

@@ -12,7 +12,7 @@
   "rules": [
     {
       "name": "单 IP 404 高频扫描",
-      "expr": "request_count > 100 && status_404_ratio >= 0.8"
+      "expr": "request_count > 100 && StatusRatio(404) >= 0.8"
     }
   ]
 }
@@ -57,6 +57,13 @@ Host 是否为“通过 IP 访问”按请求日志中的 `Host` 字段判断：
 
 比例字段都是 `0` 到 `1` 之间的小数。80% 应写成 `0.8`，50% 应写成 `0.5`。
 
+### 自定义状态码匹配方法
+
+如果内置的 `status_404_count` 和 `status_404_ratio` 不能满足您的需求，您可以使用以下内置方法来匹配任意状态码的请求数与占比：
+
+* **`StatusCount(code)`**: 获取当前 IP 在回看窗口内返回指定状态码的请求数（如 `StatusCount(403) > 10`）
+* **`StatusRatio(code)`**: 获取当前 IP 在回看窗口内返回指定状态码的请求数占该 IP 总请求数的比例（如 `StatusRatio(502) >= 0.5`）
+
 ## Expr 常用写法
 
 自动 IP 组使用 Expr 语法，当前表达式必须返回布尔值。
@@ -67,12 +74,12 @@ Host 是否为“通过 IP 访问”按请求日志中的 `Host` 字段判断：
 | --- | --- | --- |
 | `>`、`>=`、`<`、`<=` | 数值比较 | `request_count > 100` |
 | `==`、`!=` | 相等或不相等 | `ip != "127.0.0.1"` |
-| `&&` | 并且 | `request_count > 100 && status_404_ratio >= 0.8` |
-| `||` | 或者 | `status_404_ratio >= 0.8 || server_error_count > 20` |
+| `&&` | 并且 | `request_count > 100 && StatusRatio(404) >= 0.8` |
+| `||` | 或者 | `StatusRatio(404) >= 0.8 || server_error_count > 20` |
 | `!` | 取反 | `!(ip == "127.0.0.1")` |
 | `in` | 判断值是否在列表中 | `ip in ["203.0.113.10", "198.51.100.20"]` |
 | `not in` | 判断值是否不在列表中 | `ip not in ["127.0.0.1"]` |
-| `()` | 分组控制优先级 | `(request_count > 100 && status_404_ratio >= 0.8) || server_error_count > 50` |
+| `()` | 分组控制优先级 | `(request_count > 100 && StatusRatio(404) >= 0.8) || server_error_count > 50` |
 
 ## 内置预设
 
@@ -81,7 +88,7 @@ Host 是否为“通过 IP 访问”按请求日志中的 `Host` 字段判断：
 ```json
 {
   "name": "单 IP 404 高频扫描",
-  "expr": "request_count > 100 && status_404_ratio >= 0.8"
+  "expr": "request_count > 100 && StatusRatio(404) >= 0.8"
 }
 ```
 
@@ -106,7 +113,7 @@ Host 是否为“通过 IP 访问”按请求日志中的 `Host` 字段判断：
   "rules": [
     {
       "name": "高频 404 扫描",
-      "expr": "request_count > 100 && status_404_ratio >= 0.8"
+      "expr": "request_count > 100 && StatusRatio(404) >= 0.8"
     }
   ]
 }
@@ -148,7 +155,7 @@ IP 直连访问异常：
   "rules": [
     {
       "name": "排除可信 IP 的 404 扫描",
-      "expr": "ip not in [\"203.0.113.10\", \"198.51.100.20\"] && request_count > 100 && status_404_ratio >= 0.8"
+      "expr": "ip not in [\"203.0.113.10\", \"198.51.100.20\"] && request_count > 100 && StatusRatio(404) >= 0.8"
     }
   ]
 }

docs/guide/waf-usage.md

@@ -48,7 +48,7 @@ IP 组是进行大批量 IP 过滤的基石。OpenFlare 提供了极富弹性的
 #### 3. 自动 IP 组 (Automatic)
 *   **用途**：**最具杀伤力的防扫描、防爆破自动通道**。
 *   **配置**：类型选择「自动」-> 编写 Expr 日志聚合逻辑。你可以直接引用系统内置的预设：
-    *   **单 IP 404 高频扫描**：`request_count > 100 && status_404_ratio >= 0.8` (单个 IP 最近一小时请求超 100 次且 404 响应占比超 80%)。
+    *   **单 IP 404 高频扫描**：`request_count > 100 && StatusRatio(404) >= 0.8` (单个 IP 最近一小时请求超 100 次且 404 响应占比超 80%)。
     *   **单 IP 直连访问异常**：`ip_host_count > 50 && ip_host_ratio > 0.5` (绕过域名直接通过 IP 地址进行高频请求)。
 *   **测试与立即执行**：保存前可点击 **「测试规则」** 按钮预览当前日志窗口被命中的 IP。保存后可点击 **「立即执行」** 直接聚合日志并生成封禁名单。
 

docs/reference/api.md

@@ -41,23 +41,7 @@ OpenFlare 的管理端 API 与 Agent API 都使用 JSON。
 | `POST` | `/api/waf/ip-groups/:id/delete` | 删除 IP 组；已被规则组引用时会拒绝 |
 | `POST` | `/api/waf/ip-groups/:id/sync` | 立即同步订阅型 IP 组或立即执行自动型 IP 组 |
 
-IP 组 `type` 支持 `manual`、`automatic`、`subscription`。自动型 IP 组的 `auto_config` 是 JSON 对象，当前支持：
-
-```json
-{
-  "lookback_minutes": 60,
-  "rules": [
-    {
-      "name": "单 IP 404 高频扫描",
-      "expr": "request_count > 100 && status_404_ratio >= 0.8"
-    },
-    {
-      "name": "单 IP 直连访问异常",
-      "expr": "ip_host_count > 50 && ip_host_ratio > 0.5"
-    }
-  ]
-}
-```
+IP 组 `type` 支持 `manual`、`automatic`、`subscription`。自动型 IP 组的 `auto_config` 是 JSON 对象
 
 自动规则使用 Expr 语法，表达式必须返回布尔值。规则按单个 IP 的请求日志聚合指标计算，可用字段包括 `ip`、`request_count`、`status_404_count`、`status_404_ratio`、`ip_host_count`、`ip_host_ratio`、`client_error_count`、`server_error_count`、`last_seen_unix`。完整语法和字段含义见 [WAF 自动 IP 组规则语法](../guide/waf-ip-group-expr.md)。订阅格式支持 `text` 与 `json`：文本格式按行解析 IP/IP 段并忽略空行和 `#` 开头的注释；JSON 格式可通过映射规则选择数组，默认读取根数组。
 

openflare_agent/internal/agent/runner.go

@@ -12,6 +12,7 @@ import (
 	"openflare-agent/internal/observability"
 	"openflare-agent/internal/protocol"
 	"openflare-agent/internal/state"
+	"openflare-agent/internal/wsclient"
 )
 
 type HeartbeatService interface {
@@ -211,53 +212,75 @@ func (r *Runner) startWebSocket(ctx context.Context, nodeID string) (<-chan erro
 	return done, nil
 }
 
+type agentWSHandler struct {
+	runner       *Runner
+	conn         protocol.WebSocketConnection
+	nodeID       string
+	statusTicker *time.Ticker
+}
+
+func (h *agentWSHandler) OnConnect(ctx context.Context) error {
+	return h.runner.sendWebSocketStatus(ctx, h.nodeID, h.conn)
+}
+
+func (h *agentWSHandler) HandleMessage(ctx context.Context, msg wsclient.WSMessage) error {
+	var payloadBytes []byte
+	if msg.Payload != nil {
+		payloadBytes = []byte(msg.Payload)
+	}
+	protoMsg := protocol.WSMessage{
+		Type:    msg.Type,
+		Payload: payloadBytes,
+	}
+	changed, err := h.runner.handleWebSocketMessage(ctx, protoMsg, h.conn)
+	if err != nil {
+		return err
+	}
+	if changed {
+		h.statusTicker.Reset(h.runner.Config.HeartbeatInterval.Duration())
+	}
+	return nil
+}
+
+func (h *agentWSHandler) OnClose(err error) {
+	slog.Error("agent ws receive failed", "error", err)
+}
+
 func (r *Runner) runWebSocket(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error {
 	slog.Debug("agent ws connected", "url", conn.URL(), "node_id", nodeID)
 	statusTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration())
 	defer statusTicker.Stop()
 
-	messages := make(chan protocol.WSMessage, 8)
-	readDone := make(chan error, 1)
+	childCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// Start status ticker sender in background
 	go func() {
 		for {
-			message, err := conn.Receive()
-			if err != nil {
-				readDone <- err
-				return
-			}
 			select {
-			case messages <- message:
-			case <-ctx.Done():
-				readDone <- ctx.Err()
+			case <-childCtx.Done():
 				return
+			case <-statusTicker.C:
+				if err := r.sendWebSocketStatus(childCtx, nodeID, conn); err != nil {
+					slog.Error("agent ws send status failed", "error", err)
+					_ = conn.Close()
+					return
+				}
 			}
 		}
 	}()
 
-	if err := r.sendWebSocketStatus(ctx, nodeID, conn); err != nil {
-		return err
+	wsConn, ok := conn.(*wsclient.Connection)
+	if !ok {
+		return errors.New("invalid websocket connection type")
 	}
 
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case err := <-readDone:
-			return err
-		case <-statusTicker.C:
-			if err := r.sendWebSocketStatus(ctx, nodeID, conn); err != nil {
-				return err
-			}
-		case message := <-messages:
-			changed, err := r.handleWebSocketMessage(ctx, message, conn)
-			if err != nil {
-				return err
-			}
-			if changed {
-				statusTicker.Reset(r.Config.HeartbeatInterval.Duration())
-			}
-		}
-	}
+	return wsConn.RunReceiveLoop(childCtx, &agentWSHandler{
+		runner:       r,
+		conn:         conn,
+		nodeID:       nodeID,
+		statusTicker: statusTicker,
+	})
 }
 
 func (r *Runner) sendWebSocketStatus(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error {

openflare_agent/internal/nginx/pow_assets.go

@@ -236,8 +236,27 @@ local challenge_info = cjson.encode({
 pow_challenges:set(challenge_id, challenge_info, challenge_ttl)
 
 local static_prefix = "/.within.website/x/cmd/anubis/static/"
-local title = "Making sure you're not a bot!"
+local accept_lang = ngx.var.http_accept_language or ""
 local lang = "en"
+if string.find(accept_lang, "zh") then
+    lang = "zh-CN"
+end
+
+local t_title = "Making sure you're not a bot!"
+local t_status = "Loading..."
+local t_protected = "This site is protected by a Proof-of-Work challenge. Your browser will solve a small puzzle before the upstream response is shown."
+local t_why = "Why am I seeing this?"
+local t_why_desc = "OpenFlare is asking your browser to complete a lightweight computation to distinguish normal browser traffic from automated abuse. This should finish automatically."
+local t_noscript = "JavaScript is required to pass this verification. Please enable JavaScript and reload."
+
+if lang == "zh-CN" then
+    t_title = "正在确认你是不是机器人！"
+    t_status = "加载中..."
+    t_protected = "本网站受工作量证明（Proof-of-Work）挑战保护。在显示源站响应之前，您的浏览器将解决一个微型谜题。"
+    t_why = "为什么我会看到这个？"
+    t_why_desc = "OpenFlare 正在要求您的浏览器完成一项轻量级计算，以区分正常的浏览器流量和自动化的恶意请求。这应该会自动完成。"
+    t_noscript = "很遗憾，您必须启用 JavaScript 才能通过这项验证。请开启 JavaScript 并刷新页面。"
+end
 
 ngx.header.content_type = "text/html; charset=utf-8"
 ngx.say([[<!DOCTYPE html>
@@ -246,7 +265,7 @@ ngx.say([[<!DOCTYPE html>
 <meta charset="utf-8">
 <meta name="viewport" content="width=device-width, initial-scale=1">
 <meta name="robots" content="noindex,nofollow">
-<title>]] .. title .. [[</title>
+<title>]] .. t_title .. [[</title>
 <link rel="stylesheet" href="]] .. static_prefix .. [[css/xess.css">
 <style>
 body,html{height:100%;display:flex;justify-content:center;align-items:center;margin-left:auto;margin-right:auto}
@@ -272,17 +291,17 @@ body,html{height:100%;display:flex;justify-content:center;align-items:center;mar
 </head>
 <body id="top">
 <main>
-<h1 id="title" class="centered-div">]] .. title .. [[</h1>
+<h1 id="title" class="centered-div">]] .. t_title .. [[</h1>
 <div class="centered-div">
 <img id="image" style="width:100%;max-width:256px;" src="]] .. static_prefix .. [[img/pensive.webp?cacheBuster=openflare-pow">
-<p id="status">Loading...</p>
-<p>This site is protected by a Proof-of-Work challenge. Your browser will solve a small puzzle before the upstream response is shown.</p>
+<p id="status">]] .. t_status .. [[</p>
+<p>]] .. t_protected .. [[</p>
 <div id="progress" role="progressbar" aria-labelledby="status"><div class="bar-inner"></div></div>
 <details>
-<summary>Why am I seeing this?</summary>
-<p>OpenFlare is asking your browser to complete a lightweight computation to distinguish normal browser traffic from automated abuse. This should finish automatically.</p>
+<summary>]] .. t_why .. [[</summary>
+<p>]] .. t_why_desc .. [[</p>
 </details>
-<noscript><p>JavaScript is required to pass this verification. Please enable JavaScript and reload.</p></noscript>
+<noscript><p>]] .. t_noscript .. [[</p></noscript>
 </div>
 </main>
 <script type="module" src="]] .. static_prefix .. [[js/main.mjs"></script>

openflare_agent/internal/wsclient/client.go

@@ -2,165 +2,81 @@ package wsclient
 
 import (
 	"context"
-	"errors"
-	"log/slog"
-	"net"
-	"net/http"
-	"net/url"
-	"strings"
 	"time"
 
-	"golang.org/x/net/websocket"
-
 	"openflare-agent/internal/protocol"
+	shared "openflare/utils/wsclient"
 )
 
+type WSMessage = shared.WSMessage
+type MessageHandler = shared.MessageHandler
+
 type Client struct {
-	baseURL string
-	token   string
-	timeout time.Duration
+	sharedClient *shared.Client
 }
 
 type Connection struct {
-	conn        *websocket.Conn
-	url         string
-	readTimeout time.Duration
+	sharedConn *shared.Connection
 }
 
 func New(baseURL string, token string, timeout time.Duration) *Client {
 	return &Client{
-		baseURL: strings.TrimRight(baseURL, "/"),
-		token:   strings.TrimSpace(token),
-		timeout: timeout,
+		sharedClient: shared.New(shared.Config{
+			BaseURL:   baseURL,
+			Token:     token,
+			Timeout:   timeout,
+			HeaderKey: "X-Agent-Token",
+			WSPath:    "/api/agent/ws",
+		}),
 	}
 }
 
 func (c *Client) SetToken(token string) {
-	c.token = strings.TrimSpace(token)
-	slog.Debug("agent ws client token updated")
+	c.sharedClient.SetToken(token)
 }
 
 func (c *Client) URL() string {
-	wsURL, err := buildWebsocketURL(c.baseURL)
-	if err != nil {
-		return ""
-	}
-	return wsURL
+	return c.sharedClient.URL()
 }
 
 func (c *Client) Connect(ctx context.Context) (protocol.WebSocketConnection, error) {
-	wsURL, err := buildWebsocketURL(c.baseURL)
+	conn, err := c.sharedClient.Connect(ctx)
 	if err != nil {
 		return nil, err
 	}
-	if strings.TrimSpace(c.token) == "" {
-		return nil, errors.New("agent ws token is empty")
-	}
-	origin := strings.TrimSpace(c.baseURL)
-	if origin == "" {
-		origin = "http://localhost"
-	}
-	config, err := websocket.NewConfig(wsURL, origin)
-	if err != nil {
-		return nil, err
-	}
-	config.Header = http.Header{}
-	config.Header.Set("X-Agent-Token", c.token)
-	if c.timeout > 0 {
-		config.Dialer = &net.Dialer{Timeout: c.timeout}
-	}
-	slog.Debug("agent ws dialing server", "url", wsURL)
-	conn, err := config.DialContext(ctx)
-	if err != nil {
-		return nil, err
-	}
-	slog.Debug("agent ws dial succeeded", "url", wsURL)
-	return &Connection{conn: conn, url: wsURL, readTimeout: websocketReadTimeout(c.timeout)}, nil
-}
-
-func buildWebsocketURL(baseURL string) (string, error) {
-	parsed, err := url.Parse(strings.TrimRight(baseURL, "/"))
-	if err != nil {
-		return "", err
-	}
-	switch parsed.Scheme {
-	case "http":
-		parsed.Scheme = "ws"
-	case "https":
-		parsed.Scheme = "wss"
-	case "ws", "wss":
-	default:
-		return "", errors.New("server_url scheme must be http, https, ws, or wss")
-	}
-	parsed.Path = strings.TrimRight(parsed.Path, "/") + "/api/agent/ws"
-	parsed.RawQuery = ""
-	parsed.Fragment = ""
-	return parsed.String(), nil
+	return &Connection{sharedConn: conn}, nil
 }
 
 func (conn *Connection) URL() string {
-	if conn == nil {
+	if conn == nil || conn.sharedConn == nil {
 		return ""
 	}
-	return conn.url
+	return conn.sharedConn.URL
 }
 
 func (conn *Connection) SendStatus(payload protocol.NodePayload) error {
-	if conn == nil || conn.conn == nil {
-		return errors.New("agent ws connection is nil")
-	}
-	slog.Debug("agent ws sending status",
-		"node_id", payload.NodeID,
-		"current_version", payload.CurrentVersion,
-		"openresty_status", payload.OpenrestyStatus,
-	)
-	return websocket.JSON.Send(conn.conn, protocol.WSOutboundMessage{
-		Type:    protocol.WSMessageTypeStatus,
-		Payload: payload,
-	})
+	return conn.sharedConn.SendMessage(protocol.WSMessageTypeStatus, payload)
 }
 
 func (conn *Connection) SendPong() error {
-	if conn == nil || conn.conn == nil {
-		return errors.New("agent ws connection is nil")
-	}
-	slog.Debug("agent ws sending pong")
-	return websocket.JSON.Send(conn.conn, protocol.WSOutboundMessage{
-		Type: protocol.WSMessageTypePong,
-	})
+	return conn.sharedConn.SendMessage(protocol.WSMessageTypePong, nil)
 }
 
 func (conn *Connection) Receive() (protocol.WSMessage, error) {
 	var message protocol.WSMessage
-	if conn == nil || conn.conn == nil {
-		return message, errors.New("agent ws connection is nil")
-	}
-	if conn.readTimeout > 0 {
-		_ = conn.conn.SetReadDeadline(time.Now().Add(conn.readTimeout))
-	}
-	err := websocket.JSON.Receive(conn.conn, &message)
-	if err != nil {
-		var netErr net.Error
-		if errors.As(err, &netErr) && netErr.Timeout() {
-			slog.Debug("agent ws receive timeout waiting for server message", "timeout", conn.readTimeout)
-		}
+	if err := conn.sharedConn.Receive(&message); err != nil {
 		return message, err
 	}
-	slog.Debug("agent ws received message", "type", message.Type)
 	return message, nil
 }
 
-func websocketReadTimeout(requestTimeout time.Duration) time.Duration {
-	timeout := requestTimeout * 6
-	if timeout < 75*time.Second {
-		return 75 * time.Second
-	}
-	return timeout
+func (conn *Connection) RunReceiveLoop(ctx context.Context, handler shared.MessageHandler) error {
+	return conn.sharedConn.RunReceiveLoop(ctx, handler)
 }
 
 func (conn *Connection) Close() error {
-	if conn == nil || conn.conn == nil {
+	if conn == nil || conn.sharedConn == nil {
 		return nil
 	}
-	return conn.conn.Close()
+	return conn.sharedConn.Close()
 }

openflare_relay/internal/frps/manager.go

@@ -102,19 +102,32 @@ func (m *Manager) UpdateConfig(cfg *service.RelayConfig) {
 		m.activeConfig.WebServerEnabled == cfg.WebServerEnabled {
 		if m.cmd == nil && !m.stopping {
 			slog.Warn("frps config unchanged but process is not running, restarting")
-			if err := m.restartProcess(); err != nil {
+			m.stopping = false
+			m.generation++
+			generation := m.generation
+			if err := m.renderConfig(cfg); err != nil {
+				slog.Error("failed to render frps config", "error", err)
 				m.status = "unhealthy"
 				m.lastError = err.Error()
-				slog.Error("failed to restart frps with unchanged config", "error", err)
+				return
 			}
+			go m.supervise(generation)
 		}
 		return
 	}
 
 	m.activeConfig = cfg
 	m.stopping = false
+	m.generation++
+	generation := m.generation
 	slog.Info("relay config updated, reloading frps")
 
+	if m.cmd != nil && m.cmd.Process != nil {
+		slog.Debug("stopping existing frps process")
+		_ = m.cmd.Process.Kill()
+		m.cmd = nil
+	}
+
 	if err := m.renderConfig(cfg); err != nil {
 		slog.Error("failed to render frps config", "error", err)
 		m.status = "unhealthy"
@@ -122,14 +135,7 @@ func (m *Manager) UpdateConfig(cfg *service.RelayConfig) {
 		return
 	}
 
-	if err := m.restartProcess(); err != nil {
-		slog.Error("failed to restart frps", "error", err)
-		m.status = "unhealthy"
-		m.lastError = err.Error()
-	} else {
-		m.status = "healthy"
-		m.lastError = ""
-	}
+	go m.supervise(generation)
 }
 
 func (m *Manager) renderConfig(cfg *service.RelayConfig) error {
@@ -166,63 +172,95 @@ func (m *Manager) renderConfig(cfg *service.RelayConfig) error {
 	return os.WriteFile(m.configPath, buf.Bytes(), 0644)
 }
 
-func (m *Manager) restartProcess() error {
-	m.generation++
-	generation := m.generation
-	if m.cmd != nil && m.cmd.Process != nil {
-		slog.Debug("stopping existing frps process")
-		_ = m.cmd.Process.Kill()
-		m.cmd = nil
-	}
-	return m.startProcessLocked(generation)
-}
+func (m *Manager) supervise(generation uint64) {
+	backoff := 1 * time.Second
+	const maxBackoff = 60 * time.Second
 
-func (m *Manager) startProcessLocked(generation uint64) error {
-	cmd := exec.Command(m.frpsPath, "-c", m.configPath)
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
-
-	if err := cmd.Start(); err != nil {
-		return err
-	}
-
-	m.cmd = cmd
-	m.status = "healthy"
-	m.lastError = ""
-
-	go func(c *exec.Cmd) {
-		err := c.Wait()
-		slog.Warn("frps process exited", "error", err)
+	for {
 		m.mu.Lock()
-		if m.cmd == c {
+		if m.stopping || m.generation != generation {
+			m.mu.Unlock()
+			return
+		}
+
+		cmd := exec.Command(m.frpsPath, "-c", m.configPath)
+		cmd.Stdout = os.Stdout
+		cmd.Stderr = os.Stderr
+
+		err := cmd.Start()
+		if err != nil {
+			m.status = "unhealthy"
+			m.lastError = fmt.Sprintf("failed to start: %v", err)
+			slog.Error("failed to start frps", "error", err, "generation", generation)
+			m.mu.Unlock()
+
+			if !m.sleepOrInterrupt(generation, backoff) {
+				return
+			}
+			backoff = backoff * 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+			continue
+		}
+
+		m.cmd = cmd
+		m.status = "healthy"
+		m.lastError = ""
+		m.mu.Unlock()
+
+		startedAt := time.Now()
+		waitErr := cmd.Wait()
+
+		m.mu.Lock()
+		if m.cmd == cmd {
 			m.cmd = nil
 			m.status = "unhealthy"
-			if err != nil {
-				m.lastError = err.Error()
+			if waitErr != nil {
+				m.lastError = fmt.Sprintf("exited with error: %v", waitErr)
 			} else {
-				m.lastError = "frps process exited"
+				m.lastError = "exited unexpectedly"
+			}
+			slog.Warn("frps process exited unexpectedly", "error", waitErr, "generation", generation)
+		}
+		shouldContinue := !m.stopping && m.generation == generation
+		m.mu.Unlock()
+
+		if !shouldContinue {
+			return
+		}
+
+		if time.Since(startedAt) >= 10*time.Second {
+			backoff = 1 * time.Second
+		}
+
+		if !m.sleepOrInterrupt(generation, backoff) {
+			return
+		}
+		backoff = backoff * 2
+		if backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}
+
+func (m *Manager) sleepOrInterrupt(generation uint64, d time.Duration) bool {
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+
+	deadline := time.Now().Add(d)
+	for time.Now().Before(deadline) {
+		select {
+		case <-ticker.C:
+			m.mu.RLock()
+			interrupted := m.stopping || m.generation != generation
+			m.mu.RUnlock()
+			if interrupted {
+				return false
 			}
 		}
-		shouldRestart := !m.stopping && m.generation == generation
-		m.mu.Unlock()
-		if !shouldRestart {
-			return
-		}
-		time.Sleep(2 * time.Second)
-		m.mu.Lock()
-		defer m.mu.Unlock()
-		if m.stopping || m.generation != generation {
-			return
-		}
-		slog.Warn("restarting frps after unexpected exit")
-		if err := m.startProcessLocked(generation); err != nil {
-			m.status = "unhealthy"
-			m.lastError = err.Error()
-			slog.Error("failed to auto restart frps", "error", err)
-		}
-	}(cmd)
-
-	return nil
+	}
+	return true
 }
 
 func (m *Manager) Stop() {

openflare_relay/internal/frps/manager_test.go

@@ -0,0 +1,295 @@
+package frps
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"openflare/service"
+)
+
+// Helper to write control file for the dummy script
+func writeControl(t *testing.T, dir string, exitCode int, delaySeconds int) {
+	controlPath := filepath.Join(dir, "control.txt")
+	content := fmt.Sprintf("%d %d\n", exitCode, delaySeconds)
+	err := os.WriteFile(controlPath, []byte(content), 0644)
+	if err != nil {
+		t.Fatalf("failed to write control file: %v", err)
+	}
+}
+
+// Setup a dummy executable script that reads control.txt to decide exit code and sleep duration
+func setupDummyScript(t *testing.T) (string, string) {
+	dir := t.TempDir()
+	scriptPath := filepath.Join(dir, "dummy_frps")
+
+	// On macOS/Linux, we write a shell script
+	scriptContent := fmt.Sprintf(`#!/bin/sh
+control_file="%s/control.txt"
+EXIT_CODE=0
+DELAY=0
+if [ -f "$control_file" ]; then
+    read -r EXIT_CODE DELAY < "$control_file"
+fi
+if [ -n "$DELAY" ] && [ "$DELAY" -gt 0 ] 2>/dev/null; then
+    sleep "$DELAY"
+fi
+exit "${EXIT_CODE:-0}"
+`, dir)
+
+	err := os.WriteFile(scriptPath, []byte(scriptContent), 0755)
+	if err != nil {
+		t.Fatalf("failed to write dummy script: %v", err)
+	}
+
+	return scriptPath, dir
+}
+
+// Helper to poll for status to eliminate timing flakiness in tests
+func assertStatusEventually(t *testing.T, m *Manager, expectedStatus string, timeout time.Duration) {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		rt := m.GetRuntimeStatus()
+		if rt.Status == expectedStatus {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	rt := m.GetRuntimeStatus()
+	t.Fatalf("expected status eventually %s, got %s (err: %s)", expectedStatus, rt.Status, rt.LastError)
+}
+
+func TestStartProcessSuccess(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	writeControl(t, dir, 0, 5) // exit code 0, sleep 5s
+
+	m := NewManager(scriptPath, dir, "agent-token")
+	defer m.Stop()
+
+	cfg := &service.RelayConfig{
+		BindPort:         7000,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+
+	m.UpdateConfig(cfg)
+
+	assertStatusEventually(t, m, "healthy", 2*time.Second)
+
+	rt := m.GetRuntimeStatus()
+	if !rt.ProcessAlive {
+		t.Error("expected process to be alive")
+	}
+}
+
+func TestStartProcessFailureAndBackoff(t *testing.T) {
+	dir := t.TempDir()
+	invalidScriptPath := filepath.Join(dir, "non_existent_frps")
+
+	m := NewManager(invalidScriptPath, dir, "agent-token")
+	defer m.Stop()
+
+	cfg := &service.RelayConfig{
+		BindPort:         7000,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+
+	m.UpdateConfig(cfg)
+
+	assertStatusEventually(t, m, "unhealthy", 2*time.Second)
+
+	rt := m.GetRuntimeStatus()
+	if !strings.Contains(rt.LastError, "failed to start") {
+		t.Errorf("expected error message containing 'failed to start', got %s", rt.LastError)
+	}
+
+	// Correct the path to dummy script
+	scriptPath, _ := setupDummyScript(t)
+	writeControl(t, filepath.Dir(scriptPath), 0, 5)
+
+	m.mu.Lock()
+	m.frpsPath = scriptPath
+	m.mu.Unlock()
+
+	// Wait for backoff retry (1s backoff)
+	assertStatusEventually(t, m, "healthy", 3*time.Second)
+
+	rt = m.GetRuntimeStatus()
+	if !rt.ProcessAlive {
+		t.Error("expected process to be alive now")
+	}
+}
+
+func TestUnexpectedExitAndAutorestart(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	// Start with immediate exit code 1
+	writeControl(t, dir, 1, 0)
+
+	m := NewManager(scriptPath, dir, "agent-token")
+	defer m.Stop()
+
+	cfg := &service.RelayConfig{
+		BindPort:         7000,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+
+	m.UpdateConfig(cfg)
+
+	assertStatusEventually(t, m, "unhealthy", 2*time.Second)
+
+	rt := m.GetRuntimeStatus()
+	if !strings.Contains(rt.LastError, "exited with error") {
+		t.Errorf("expected exit error, got %s", rt.LastError)
+	}
+
+	// Change control to be healthy (runs for 5s, exit 0)
+	writeControl(t, dir, 0, 5)
+
+	// Wait for the retry to fire (backoff was 1s)
+	assertStatusEventually(t, m, "healthy", 3*time.Second)
+}
+
+func TestBackoffReset(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	// Rapid exit to increase backoff
+	writeControl(t, dir, 1, 0)
+
+	m := NewManager(scriptPath, dir, "agent-token")
+	defer m.Stop()
+
+	cfg := &service.RelayConfig{
+		BindPort:         7000,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+
+	m.UpdateConfig(cfg)
+
+	// Crashed once, backoff is 2s
+	assertStatusEventually(t, m, "unhealthy", 2*time.Second)
+
+	// Now make it run successfully for 11 seconds (exit code 0, sleep 11s)
+	writeControl(t, dir, 0, 11)
+
+	// Wait for next retry to start running
+	assertStatusEventually(t, m, "healthy", 4*time.Second)
+
+	// Wait for process to run for 10.5 seconds to trigger backoff reset
+	time.Sleep(10500 * time.Millisecond)
+
+	// Now make it crash again (exit code 1, sleep 0s)
+	writeControl(t, dir, 1, 0)
+
+	// Wait for it to finish and crash
+	assertStatusEventually(t, m, "unhealthy", 3*time.Second)
+
+	// It crashed. Since it ran for > 10s, backoff should have been reset to 1s.
+	// We make it healthy again (exit code 0, sleep 5)
+	writeControl(t, dir, 0, 5)
+
+	// Wait 1.5 seconds. If backoff was reset to 1s, it should be healthy now.
+	assertStatusEventually(t, m, "healthy", 2*time.Second)
+}
+
+func TestImmediateRestartOnSameConfigDeadProcess(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	// Crashes immediately
+	writeControl(t, dir, 1, 0)
+
+	m := NewManager(scriptPath, dir, "agent-token")
+	defer m.Stop()
+
+	cfg := &service.RelayConfig{
+		BindPort:         7000,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+
+	m.UpdateConfig(cfg)
+
+	// Let it crash
+	assertStatusEventually(t, m, "unhealthy", 2*time.Second)
+
+	// Make it start successfully
+	writeControl(t, dir, 0, 5)
+
+	// Send same config block to trigger immediate restart bypass of backoff sleep
+	m.UpdateConfig(cfg)
+
+	// Check if it started immediately
+	assertStatusEventually(t, m, "healthy", 2*time.Second)
+}
+
+func TestSupervisorGenerationInterrupt(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	writeControl(t, dir, 0, 10)
+
+	m := NewManager(scriptPath, dir, "agent-token")
+	defer m.Stop()
+
+	cfg := &service.RelayConfig{
+		BindPort:         7000,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+
+	m.UpdateConfig(cfg)
+
+	assertStatusEventually(t, m, "healthy", 2*time.Second)
+
+	m.mu.Lock()
+	gen1 := m.generation
+	cmd1 := m.cmd
+	m.mu.Unlock()
+
+	if cmd1 == nil {
+		t.Fatal("expected active process")
+	}
+
+	// Update configuration with new bind port to trigger new generation
+	cfg2 := &service.RelayConfig{
+		BindPort:         7001,
+		VhostHTTPPort:    8080,
+		AuthToken:        "test-auth",
+		WebServerEnabled: false,
+	}
+	m.UpdateConfig(cfg2)
+
+	assertStatusEventually(t, m, "healthy", 2*time.Second)
+
+	m.mu.Lock()
+	gen2 := m.generation
+	cmd2 := m.cmd
+	m.mu.Unlock()
+
+	if gen2 <= gen1 {
+		t.Errorf("expected generation incremented, got gen1=%d gen2=%d", gen1, gen2)
+	}
+	if cmd2 == cmd1 {
+		t.Error("expected old process killed and new command started")
+	}
+
+	// Verify old process is actually killed
+	var cmd1Finished int32
+	go func() {
+		_ = cmd1.Wait()
+		atomic.StoreInt32(&cmd1Finished, 1)
+	}()
+
+	time.Sleep(200 * time.Millisecond)
+	if atomic.LoadInt32(&cmd1Finished) != 1 {
+		t.Error("expected first process to be killed")
+	}
+}

openflare_relay/internal/relay/runner.go

@@ -51,66 +51,35 @@ func (r *Runner) Run(ctx context.Context) error {
 	}
 }
 
-func (r *Runner) handleConnection(ctx context.Context, conn *wsclient.Connection) {
-	// Send pings at 2× heartbeat interval to keep the server-side read deadline
-	// from expiring (server closes the WS if no data arrives within ~30 s).
-	pingInterval := r.Config.HeartbeatInterval.Duration() * 2
-	pingTicker := time.NewTicker(pingInterval)
-	defer pingTicker.Stop()
+type relayWSHandler struct {
+	runner *Runner
+}
 
-	messages := make(chan service.WSMessage, 8)
-	readDone := make(chan error, 1)
-	go func() {
-		for {
-			msg, err := conn.Receive()
-			if err != nil {
-				readDone <- err
-				return
-			}
-			select {
-			case messages <- msg:
-			case <-ctx.Done():
-				readDone <- ctx.Err()
-				return
-			}
-		}
-	}()
+func (h *relayWSHandler) OnConnect(ctx context.Context) error {
+	return nil
+}
 
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case err := <-readDone:
-			slog.Error("relay ws receive failed", "error", err)
-			return
-		case <-pingTicker.C:
-			if err := conn.SendPing(); err != nil {
-				slog.Error("relay ws send ping failed", "error", err)
-				return
-			}
-		case msg := <-messages:
-			switch msg.Type {
-			case "ping":
-				_ = conn.SendPong()
-			case "pong":
-				slog.Debug("relay ws pong received")
-			case "relay_config":
-				payloadBytes, ok := msg.Payload.(json.RawMessage)
-				if !ok {
-					slog.Error("invalid relay_config payload type")
-					continue
-				}
-				var cfg service.RelayConfig
-				if err := json.Unmarshal(payloadBytes, &cfg); err != nil {
-					slog.Error("failed to unmarshal relay_config", "error", err)
-					continue
-				}
-				r.FrpsManager.UpdateConfig(&cfg)
-			default:
-				slog.Debug("ignored unknown ws message type", "type", msg.Type)
-			}
+func (h *relayWSHandler) HandleMessage(ctx context.Context, msg wsclient.WSMessage) error {
+	switch msg.Type {
+	case "relay_config":
+		var cfg service.RelayConfig
+		if err := json.Unmarshal(msg.Payload, &cfg); err != nil {
+			slog.Error("failed to unmarshal relay_config", "error", err)
+			return nil
 		}
+		h.runner.FrpsManager.UpdateConfig(&cfg)
+	default:
+		slog.Debug("ignored unknown ws message type", "type", msg.Type)
 	}
+	return nil
+}
+
+func (h *relayWSHandler) OnClose(err error) {
+	slog.Error("relay ws receive failed", "error", err)
+}
+
+func (r *Runner) handleConnection(ctx context.Context, conn *wsclient.Connection) {
+	_ = conn.RunReceiveLoop(ctx, &relayWSHandler{runner: r})
 }
 
 func (r *Runner) sleepContext(ctx context.Context, d time.Duration) {

openflare_relay/internal/wsclient/client.go

@@ -3,151 +3,73 @@ package wsclient
 import (
 	"context"
 	"encoding/json"
-	"errors"
-	"log/slog"
-	"net"
-	"net/http"
-	"net/url"
-	"strings"
 	"time"
 
-	"golang.org/x/net/websocket"
 	"openflare/service"
+	shared "openflare/utils/wsclient"
 )
 
+type WSMessage = shared.WSMessage
+type MessageHandler = shared.MessageHandler
+
 type Client struct {
-	baseURL string
-	token   string
-	timeout time.Duration
+	sharedClient *shared.Client
 }
 
 type Connection struct {
-	conn        *websocket.Conn
-	url         string
-	readTimeout time.Duration
+	sharedConn *shared.Connection
 }
 
 func New(baseURL string, token string, timeout time.Duration) *Client {
 	return &Client{
-		baseURL: strings.TrimRight(baseURL, "/"),
-		token:   strings.TrimSpace(token),
-		timeout: timeout,
+		sharedClient: shared.New(shared.Config{
+			BaseURL:   baseURL,
+			Token:     token,
+			Timeout:   timeout,
+			HeaderKey: "X-Agent-Token",
+			WSPath:    "/api/relay/ws",
+		}),
 	}
 }
 
 func (c *Client) SetToken(token string) {
-	c.token = strings.TrimSpace(token)
-	slog.Debug("relay ws client token updated")
+	c.sharedClient.SetToken(token)
 }
 
 func (c *Client) Connect(ctx context.Context) (*Connection, error) {
-	wsURL, err := buildWebsocketURL(c.baseURL)
+	conn, err := c.sharedClient.Connect(ctx)
 	if err != nil {
 		return nil, err
 	}
-	if strings.TrimSpace(c.token) == "" {
-		return nil, errors.New("relay ws token is empty")
-	}
-	origin := strings.TrimSpace(c.baseURL)
-	if origin == "" {
-		origin = "http://localhost"
-	}
-	config, err := websocket.NewConfig(wsURL, origin)
-	if err != nil {
-		return nil, err
-	}
-	config.Header = http.Header{}
-	config.Header.Set("X-Agent-Token", c.token)
-	if c.timeout > 0 {
-		config.Dialer = &net.Dialer{Timeout: c.timeout}
-	}
-	slog.Debug("relay ws dialing server", "url", wsURL)
-	conn, err := config.DialContext(ctx)
-	if err != nil {
-		return nil, err
-	}
-	slog.Debug("relay ws dial succeeded", "url", wsURL)
-	return &Connection{conn: conn, url: wsURL, readTimeout: websocketReadTimeout(c.timeout)}, nil
-}
-
-func buildWebsocketURL(baseURL string) (string, error) {
-	parsed, err := url.Parse(strings.TrimRight(baseURL, "/"))
-	if err != nil {
-		return "", err
-	}
-	switch parsed.Scheme {
-	case "http":
-		parsed.Scheme = "ws"
-	case "https":
-		parsed.Scheme = "wss"
-	case "ws", "wss":
-	default:
-		return "", errors.New("server_url scheme must be http, https, ws, or wss")
-	}
-	parsed.Path = strings.TrimRight(parsed.Path, "/") + "/api/relay/ws"
-	parsed.RawQuery = ""
-	parsed.Fragment = ""
-	return parsed.String(), nil
+	return &Connection{sharedConn: conn}, nil
 }
 
 func (conn *Connection) SendPing() error {
-	if conn == nil || conn.conn == nil {
-		return errors.New("relay ws connection is nil")
-	}
-	slog.Debug("relay ws sending ping")
-	return websocket.JSON.Send(conn.conn, service.WSMessage{
-		Type: "ping",
-	})
+	return conn.sharedConn.SendMessage("ping", nil)
 }
 
 func (conn *Connection) SendPong() error {
-	if conn == nil || conn.conn == nil {
-		return errors.New("relay ws connection is nil")
-	}
-	slog.Debug("relay ws sending pong")
-	return websocket.JSON.Send(conn.conn, service.WSMessage{
-		Type: "pong",
-	})
+	return conn.sharedConn.SendMessage("pong", nil)
 }
 
 func (conn *Connection) Receive() (service.WSMessage, error) {
-	var message service.WSMessage
-	if conn == nil || conn.conn == nil {
-		return message, errors.New("relay ws connection is nil")
-	}
-	if conn.readTimeout > 0 {
-		_ = conn.conn.SetReadDeadline(time.Now().Add(conn.readTimeout))
-	}
-	// Use custom json unmarshaling to handle any type
 	var raw struct {
 		Type    string          `json:"type"`
 		Payload json.RawMessage `json:"payload,omitempty"`
 	}
-	err := websocket.JSON.Receive(conn.conn, &raw)
-	if err != nil {
-		var netErr net.Error
-		if errors.As(err, &netErr) && netErr.Timeout() {
-			slog.Debug("relay ws receive timeout waiting for server message", "timeout", conn.readTimeout)
-		}
-		return message, err
+	if err := conn.sharedConn.Receive(&raw); err != nil {
+		return service.WSMessage{}, err
 	}
-	message.Type = raw.Type
-	message.Payload = raw.Payload
-	slog.Debug("relay ws received message", "type", message.Type)
-	return message, nil
+	return service.WSMessage{
+		Type:    raw.Type,
+		Payload: raw.Payload,
+	}, nil
 }
 
-func websocketReadTimeout(requestTimeout time.Duration) time.Duration {
-	timeout := requestTimeout * 6
-	if timeout < 75*time.Second {
-		return 75 * time.Second
-	}
-	return timeout
+func (conn *Connection) RunReceiveLoop(ctx context.Context, handler shared.MessageHandler) error {
+	return conn.sharedConn.RunReceiveLoop(ctx, handler)
 }
 
 func (conn *Connection) Close() error {
-	if conn == nil || conn.conn == nil {
-		return nil
-	}
-	return conn.conn.Close()
+	return conn.sharedConn.Close()
 }

openflare_server/common/constants.go

@@ -77,6 +77,7 @@ var OpenRestyProxyConnectTimeout = 3
 var OpenRestyProxySendTimeout = 60
 var OpenRestyProxyReadTimeout = 60
 var OpenRestyWebsocketEnabled = true
+var OpenRestyHTTP3Enabled = true
 var OpenRestyProxyRequestBufferingEnabled = false
 var OpenRestyProxyBufferingEnabled = true
 var OpenRestyProxyBuffers = "16 16k"

openflare_server/controller/option.go

@@ -159,6 +159,7 @@ func validateOpenRestyOption(key string, value string) error {
 		return nil
 	case "OpenRestyEventsMultiAcceptEnabled",
 		"OpenRestyWebsocketEnabled",
+		"OpenRestyHTTP3Enabled",
 		"OpenRestyProxyRequestBufferingEnabled",
 		"OpenRestyProxyBufferingEnabled",
 		"OpenRestyGzipEnabled",

openflare_server/main.go

@@ -49,6 +49,8 @@ func main() {
 		gin.SetMode(gin.ReleaseMode)
 	}
 	// Initialize SQL Database
+	defer service.ShutdownWSHubs()
+
 	err := model.InitDB()
 	if err != nil {
 		slog.Error("initialize database failed", "error", err)

openflare_server/model/option.go

@@ -70,6 +70,7 @@ func InitOptionMap() {
 	common.OptionMap["OpenRestyProxySendTimeout"] = strconv.Itoa(common.OpenRestyProxySendTimeout)
 	common.OptionMap["OpenRestyProxyReadTimeout"] = strconv.Itoa(common.OpenRestyProxyReadTimeout)
 	common.OptionMap["OpenRestyWebsocketEnabled"] = strconv.FormatBool(common.OpenRestyWebsocketEnabled)
+	common.OptionMap["OpenRestyHTTP3Enabled"] = strconv.FormatBool(common.OpenRestyHTTP3Enabled)
 	common.OptionMap["OpenRestyProxyRequestBufferingEnabled"] = strconv.FormatBool(common.OpenRestyProxyRequestBufferingEnabled)
 	common.OptionMap["OpenRestyProxyBufferingEnabled"] = strconv.FormatBool(common.OpenRestyProxyBufferingEnabled)
 	common.OptionMap["OpenRestyProxyBuffers"] = common.OpenRestyProxyBuffers
@@ -278,6 +279,8 @@ func updateOptionMap(key string, value string) {
 		}
 	case "OpenRestyWebsocketEnabled":
 		common.OpenRestyWebsocketEnabled = value == "true"
+	case "OpenRestyHTTP3Enabled":
+		common.OpenRestyHTTP3Enabled = value == "true"
 	case "OpenRestyProxyRequestBufferingEnabled":
 		common.OpenRestyProxyRequestBufferingEnabled = value == "true"
 	case "OpenRestyProxyBufferingEnabled":

openflare_server/service/config_version.go

@@ -152,6 +152,7 @@ type openRestyConfigSnapshot struct {
 	ProxySendTimeout          int    `json:"proxy_send_timeout"`
 	ProxyReadTimeout          int    `json:"proxy_read_timeout"`
 	WebsocketEnabled          bool   `json:"websocket_enabled"`
+	HTTP3Enabled              bool   `json:"http3_enabled"`
 	ProxyRequestBuffering     bool   `json:"proxy_request_buffering"`
 	ProxyBufferingEnabled     bool   `json:"proxy_buffering_enabled"`
 	ProxyBuffers              string `json:"proxy_buffers"`
@@ -952,6 +953,7 @@ func buildOpenRestyConfigSnapshot() openRestyConfigSnapshot {
 		ProxySendTimeout:          common.OpenRestyProxySendTimeout,
 		ProxyReadTimeout:          common.OpenRestyProxyReadTimeout,
 		WebsocketEnabled:          common.OpenRestyWebsocketEnabled,
+		HTTP3Enabled:              common.OpenRestyHTTP3Enabled,
 		ProxyRequestBuffering:     common.OpenRestyProxyRequestBufferingEnabled,
 		ProxyBufferingEnabled:     common.OpenRestyProxyBufferingEnabled,
 		ProxyBuffers:              common.OpenRestyProxyBuffers,
@@ -1071,6 +1073,7 @@ func diffOpenRestyOptionDetails(left openRestyConfigSnapshot, right openRestyCon
 	appendIfChanged("OpenRestyProxySendTimeout", fmt.Sprintf("%d", left.ProxySendTimeout), fmt.Sprintf("%d", right.ProxySendTimeout))
 	appendIfChanged("OpenRestyProxyReadTimeout", fmt.Sprintf("%d", left.ProxyReadTimeout), fmt.Sprintf("%d", right.ProxyReadTimeout))
 	appendIfChanged("OpenRestyWebsocketEnabled", fmt.Sprintf("%t", left.WebsocketEnabled), fmt.Sprintf("%t", right.WebsocketEnabled))
+	appendIfChanged("OpenRestyHTTP3Enabled", fmt.Sprintf("%t", left.HTTP3Enabled), fmt.Sprintf("%t", right.HTTP3Enabled))
 	appendIfChanged("OpenRestyProxyRequestBufferingEnabled", fmt.Sprintf("%t", left.ProxyRequestBuffering), fmt.Sprintf("%t", right.ProxyRequestBuffering))
 	appendIfChanged("OpenRestyProxyBufferingEnabled", fmt.Sprintf("%t", left.ProxyBufferingEnabled), fmt.Sprintf("%t", right.ProxyBufferingEnabled))
 	appendIfChanged("OpenRestyProxyBuffers", left.ProxyBuffers, right.ProxyBuffers)
@@ -1119,6 +1122,7 @@ func openRestyOptionKeys() []string {
 		"OpenRestyProxySendTimeout",
 		"OpenRestyProxyReadTimeout",
 		"OpenRestyWebsocketEnabled",
+		"OpenRestyHTTP3Enabled",
 		"OpenRestyProxyRequestBufferingEnabled",
 		"OpenRestyProxyBufferingEnabled",
 		"OpenRestyProxyBuffers",

openflare_server/service/https_phase1_test.go

@@ -489,6 +489,9 @@ func TestPublishConfigVersionRendersMultipleCertificatesForMultiDomainWebsite(t
 
 func TestPublishConfigVersionSkipsHTTPSForDomainsWithoutCertificate(t *testing.T) {
 	setupServiceTestDB(t)
+	if err := model.UpdateOption("OpenRestyHTTP3Enabled", "false"); err != nil {
+		t.Fatalf("UpdateOption failed: %v", err)
+	}
 
 	appCertPEM, appKeyPEM := generateCertificatePair(t, []string{"app.example.com"})
 	appCertificate, err := CreateTLSCertificate(TLSCertificateInput{
@@ -1338,6 +1341,80 @@ func TestOpenRestyProxyRequestBufferingDefaultsToOff(t *testing.T) {
 	}
 }
 
+func TestPreviewConfigVersionSupportsHTTP3(t *testing.T) {
+	setupServiceTestDB(t)
+
+	appCertPEM, appKeyPEM := generateCertificatePair(t, []string{"h3.example.com"})
+	appCertificate, err := CreateTLSCertificate(TLSCertificateInput{
+		Name:    "h3-cert",
+		CertPEM: appCertPEM,
+		KeyPEM:  appKeyPEM,
+	})
+	if err != nil {
+		t.Fatalf("CreateTLSCertificate failed: %v", err)
+	}
+
+	_, err = CreateProxyRoute(ProxyRouteInput{
+		SiteName:      "h3-site",
+		Domains:       []string{"h3.example.com"},
+		OriginURL:     "https://origin.internal",
+		Enabled:       true,
+		EnableHTTPS:   true,
+		CertID:        &appCertificate.ID,
+		DomainCertIDs: []uint{appCertificate.ID},
+	})
+	if err != nil {
+		t.Fatalf("CreateProxyRoute failed: %v", err)
+	}
+
+	if err := model.UpdateOption("OpenRestyHTTP3Enabled", "true"); err != nil {
+		t.Fatalf("UpdateOption OpenRestyHTTP3Enabled failed: %v", err)
+	}
+
+	preview, err := PreviewConfigVersion()
+	if err != nil {
+		t.Fatalf("PreviewConfigVersion failed: %v", err)
+	}
+
+	if !strings.Contains(preview.MainConfig, "listen 443 ssl default_server;\n        listen 443 quic reuseport default_server;") {
+		t.Fatalf("expected default server block to contain quic reuseport listener, main config: %s", preview.MainConfig)
+	}
+
+	if !strings.Contains(preview.RenderedConfig, "listen 443 quic;") {
+		t.Fatalf("expected routing server block to contain listen 443 quic, rendered config: %s", preview.RenderedConfig)
+	}
+	if !strings.Contains(preview.RenderedConfig, "add_header Alt-Svc 'h3=\":443\"; ma=86400';") {
+		t.Fatalf("expected routing server block to contain Alt-Svc header, rendered config: %s", preview.RenderedConfig)
+	}
+
+	if !strings.Contains(preview.RenderedConfig, "listen 443 ssl;") {
+		t.Fatal("expected standard listen 443 ssl to be preserved")
+	}
+	if !strings.Contains(preview.RenderedConfig, "http2 on;") {
+		t.Fatal("expected standard http2 on to be preserved")
+	}
+
+	if err := model.UpdateOption("OpenRestyHTTP3Enabled", "false"); err != nil {
+		t.Fatalf("UpdateOption OpenRestyHTTP3Enabled failed: %v", err)
+	}
+
+	previewOff, err := PreviewConfigVersion()
+	if err != nil {
+		t.Fatalf("PreviewConfigVersion failed: %v", err)
+	}
+
+	if strings.Contains(previewOff.MainConfig, "listen 443 quic reuseport default_server;") {
+		t.Fatal("expected default server block to omit quic listener when disabled")
+	}
+
+	if strings.Contains(previewOff.RenderedConfig, "listen 443 quic;") {
+		t.Fatal("expected routing server block to omit quic listener when disabled")
+	}
+	if strings.Contains(previewOff.RenderedConfig, "add_header Alt-Svc") {
+		t.Fatal("expected routing server block to omit Alt-Svc header when disabled")
+	}
+}
+
 func setupServiceTestDB(t *testing.T) {
 	t.Helper()
 	nodeAccessTokenCache.reset()

openflare_server/service/waf_ip_group.go

@@ -69,6 +69,21 @@ type wafIPGroupAutoRuleEnv struct {
 	ClientErrorCount int     `expr:"client_error_count"`
 	ServerErrorCount int     `expr:"server_error_count"`
 	LastSeenUnix     int64   `expr:"last_seen_unix"`
+	statusCounts     map[int]int
+}
+
+func (env wafIPGroupAutoRuleEnv) StatusCount(code int) int {
+	if env.statusCounts == nil {
+		return 0
+	}
+	return env.statusCounts[code]
+}
+
+func (env wafIPGroupAutoRuleEnv) StatusRatio(code int) float64 {
+	if env.RequestCount <= 0 || env.statusCounts == nil {
+		return 0.0
+	}
+	return float64(env.statusCounts[code]) / float64(env.RequestCount)
 }
 
 type wafIPGroupAutoAccumulator struct {
@@ -79,6 +94,7 @@ type wafIPGroupAutoAccumulator struct {
 	clientErrorCount int
 	serverErrorCount int
 	lastSeen         time.Time
+	statusCounts     map[int]int
 }
 
 type WAFIPGroupInput struct {
@@ -755,10 +771,14 @@ func evaluateParsedWAFIPGroupAutoConfig(config wafIPGroupAutoConfig, now time.Ti
 		}
 		acc := accumulators[ip]
 		if acc == nil {
-			acc = &wafIPGroupAutoAccumulator{ip: ip}
+			acc = &wafIPGroupAutoAccumulator{
+				ip:           ip,
+				statusCounts: make(map[int]int),
+			}
 			accumulators[ip] = acc
 		}
 		acc.requestCount++
+		acc.statusCounts[item.StatusCode]++
 		if item.StatusCode == http.StatusNotFound {
 			acc.status404Count++
 		}
@@ -800,6 +820,7 @@ func (acc *wafIPGroupAutoAccumulator) toExprEnv() wafIPGroupAutoRuleEnv {
 		IPHostCount:      acc.ipHostCount,
 		ClientErrorCount: acc.clientErrorCount,
 		ServerErrorCount: acc.serverErrorCount,
+		statusCounts:     acc.statusCounts,
 	}
 	if acc.requestCount > 0 {
 		env.Status404Ratio = float64(acc.status404Count) / float64(acc.requestCount)

openflare_server/service/waf_test.go

@@ -228,7 +228,7 @@ func TestSyncWAFIPGroupAutomaticExprRules(t *testing.T) {
 		AutoConfig: json.RawMessage(`{
 			"lookback_minutes": 60,
 			"rules": [
-				{"name":"单 IP 404 高频扫描","expr":"request_count > 100 && status_404_ratio >= 0.8"},
+				{"name":"单 IP 404 高频扫描","expr":"request_count > 100 && StatusRatio(404) >= 0.8"},
 				{"name":"单 IP 直连访问异常","expr":"ip_host_count > 50 && ip_host_ratio > 0.5"}
 			]
 		}`),
@@ -267,7 +267,7 @@ func TestWAFIPGroupAutoConfigReturnsMatchedIPs(t *testing.T) {
 		AutoConfig: json.RawMessage(`{
 			"lookback_minutes": 60,
 			"rules": [
-				{"name":"单 IP 404 高频扫描","expr":"request_count > 100 && status_404_ratio >= 0.8"},
+				{"name":"单 IP 404 高频扫描","expr":"request_count > 100 && StatusRatio(404) >= 0.8"},
 				{"name":"单 IP 直连访问异常","expr":"ip_host_count > 50 && ip_host_ratio > 0.5"}
 			]
 		}`),
@@ -383,7 +383,7 @@ func TestWAFIPGroupAutomaticTTLExpiration(t *testing.T) {
 			"lookback_minutes": 60,
 			"ttl": 10,
 			"rules": [
-				{"name":"404 Scan","expr":"request_count > 100 && status_404_ratio >= 0.8"}
+				{"name":"404 Scan","expr":"request_count > 100 && StatusRatio(404) >= 0.8"}
 			]
 		}`),
 	})
@@ -464,3 +464,50 @@ func seedWAFNodeAccessLogs(t *testing.T, loggedAt time.Time, remoteAddr string,
 		}
 	}
 }
+
+func TestSyncWAFIPGroupAutomaticCustomStatusRules(t *testing.T) {
+	setupServiceTestDB(t)
+
+	now := time.Now().UTC()
+	// Seed 10 requests from 203.0.113.50, where 3 return 403, 7 return 200
+	seedWAFNodeAccessLogsWithStatus(t, now, "203.0.113.50", "app.example.com", 7, http.StatusOK)
+	seedWAFNodeAccessLogsWithStatus(t, now, "203.0.113.50", "app.example.com", 3, http.StatusForbidden)
+
+	group, err := CreateWAFIPGroup(WAFIPGroupInput{
+		Name:    "custom status code blacklist",
+		Type:    WAFIPGroupTypeAutomatic,
+		Enabled: true,
+		AutoConfig: json.RawMessage(`{
+			"lookback_minutes": 60,
+			"rules": [
+				{"name":"高频 403 探测","expr":"StatusCount(403) >= 3 && StatusRatio(403) >= 0.3"}
+			]
+		}`),
+	})
+	if err != nil {
+		t.Fatalf("CreateWAFIPGroup failed: %v", err)
+	}
+	result, err := SyncWAFIPGroup(group.ID)
+	if err != nil {
+		t.Fatalf("SyncWAFIPGroup failed: %v", err)
+	}
+	if result.IPCount != 1 || result.Group.IPList[0] != "203.0.113.50" {
+		t.Fatalf("expected 203.0.113.50 to be matched, got %#v", result)
+	}
+}
+
+func seedWAFNodeAccessLogsWithStatus(t *testing.T, loggedAt time.Time, remoteAddr string, host string, count int, statusCode int) {
+	t.Helper()
+	for i := 0; i < count; i++ {
+		if err := model.DB.Create(&model.NodeAccessLog{
+			NodeID:     "node-waf-auto",
+			LoggedAt:   loggedAt.Add(-time.Duration(i%30) * time.Second),
+			RemoteAddr: remoteAddr,
+			Host:       host,
+			Path:       "/probe",
+			StatusCode: statusCode,
+		}).Error; err != nil {
+			t.Fatalf("failed to seed access log: %v", err)
+		}
+	}
+}

openflare_server/service/ws_hub.go

@@ -3,6 +3,7 @@ package service
 import (
 	"log/slog"
 	"sync"
+	"time"
 )
 
 type WSMessage struct {
@@ -65,13 +66,58 @@ type WSHub struct {
 	name    string
 	mu      sync.RWMutex
 	clients map[string]*WSClient
+	done    chan struct{}
 }
 
 func NewWSHub(name string) *WSHub {
-	return &WSHub{
+	h := &WSHub{
 		name:    name,
 		clients: make(map[string]*WSClient),
+		done:    make(chan struct{}),
 	}
+	go h.startPingLoop()
+	return h
+}
+
+func (h *WSHub) Close() {
+	close(h.done)
+}
+
+func (h *WSHub) startPingLoop() {
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-h.done:
+			return
+		case <-ticker.C:
+			h.mu.RLock()
+			if len(h.clients) == 0 {
+				h.mu.RUnlock()
+				continue
+			}
+			clients := make([]*WSClient, 0, len(h.clients))
+			for _, client := range h.clients {
+				clients = append(clients, client)
+			}
+			h.mu.RUnlock()
+
+			for _, client := range clients {
+				if !client.Send(WSMessage{
+					Type: "ping",
+				}) {
+					slog.Warn("ws client send ping failed, queue full, disconnecting", "hub", h.name, "id", client.id)
+					h.Disconnect(client.id)
+				}
+			}
+		}
+	}
+}
+
+func ShutdownWSHubs() {
+	DefaultAgentWSHub.Close()
+	DefaultFlaredWSHub.Close()
+	DefaultRelayWSHub.Close()
 }
 
 func (h *WSHub) Register(id string) *WSClient {

openflare_server/utils/render/openresty/render.go

@@ -305,7 +305,7 @@ func renderMainConfigTemplate(templateText string, cfg ConfigSnapshot) string {
 		"{{OpenRestyWorkerConnections}}", fmt.Sprintf("%d", cfg.WorkerConnections),
 		"{{OpenRestyWorkerRlimitNofile}}", fmt.Sprintf("%d", cfg.WorkerRlimitNofile),
 		"{{OpenRestyConnectionUpgradeMap}}", renderConnectionUpgradeMap(),
-		"{{OpenRestyDefaultServerBlock}}", renderDefaultServerBlock(cfg.DefaultServerReturnStatus),
+		"{{OpenRestyDefaultServerBlock}}", renderDefaultServerBlock(cfg.DefaultServerReturnStatus, cfg.HTTP3Enabled),
 		"{{OpenRestyAccessLogPath}}", AccessLogPlaceholder,
 		"{{OpenRestyErrorLogPath}}", ErrorLogPlaceholder,
 		"{{OpenRestyEventsUseDirective}}", renderTemplateDirective(cfg.EventsUse != "", fmt.Sprintf("use %s;", cfg.EventsUse)),
@@ -379,7 +379,13 @@ func renderHTTPRedirectServer(serverNames string) string {
 func renderHTTPSServer(serverNames string, siteName string, originURL string, originHost string, certificateID uint, customHeaders []CustomHeader, cacheConfig routeCacheConfig, limitConfig routeLimitConfig, upstreamConfig routeUpstreamConfig, powEnabled bool, basicAuthEnabled bool, basicAuthUsername string, basicAuthPassword string, cfg ConfigSnapshot) string {
 	certPath := fmt.Sprintf("%s/%d.crt", CertDirPlaceholder, certificateID)
 	keyPath := fmt.Sprintf("%s/%d.key", CertDirPlaceholder, certificateID)
-	return fmt.Sprintf("server {\n    listen 443 ssl;\n    http2 on;\n    server_name %s;\n    ssl_certificate %s;\n    ssl_certificate_key %s;\n%s%s    location / {\n%s%s%s%s%s    }\n%s}\n\n", serverNames, certPath, keyPath, renderAccessBlock(siteName, powEnabled), renderPowLocationBlocks(powEnabled), renderBasicAuthBlock(basicAuthEnabled, basicAuthUsername, basicAuthPassword), renderProxyHeaderBlock(originURL, originHost, customHeaders, upstreamConfig, cfg), renderRouteLimitBlock(limitConfig), renderRouteCacheBlock(cacheConfig, cfg), renderProxyPassBlock(originURL, upstreamConfig), renderPowStaticLocationBlock(powEnabled))
+	var h3Listen string
+	var h3Header string
+	if cfg.HTTP3Enabled {
+		h3Listen = "    listen 443 quic;\n"
+		h3Header = "    add_header Alt-Svc 'h3=\":443\"; ma=86400';\n"
+	}
+	return fmt.Sprintf("server {\n    listen 443 ssl;\n%s    http2 on;\n    server_name %s;\n    ssl_certificate %s;\n    ssl_certificate_key %s;\n%s%s%s    location / {\n%s%s%s%s%s    }\n%s}\n\n", h3Listen, serverNames, certPath, keyPath, h3Header, renderAccessBlock(siteName, powEnabled), renderPowLocationBlocks(powEnabled), renderBasicAuthBlock(basicAuthEnabled, basicAuthUsername, basicAuthPassword), renderProxyHeaderBlock(originURL, originHost, customHeaders, upstreamConfig, cfg), renderRouteLimitBlock(limitConfig), renderRouteCacheBlock(cacheConfig, cfg), renderProxyPassBlock(originURL, upstreamConfig), renderPowStaticLocationBlock(powEnabled))
 }
 
 func renderProxyHeaderBlock(originURL string, originHost string, customHeaders []CustomHeader, upstreamConfig routeUpstreamConfig, cfg ConfigSnapshot) string {
@@ -586,10 +592,14 @@ func renderConnectionUpgradeMap() string {
 	return "    map $http_upgrade $connection_upgrade {\n        default upgrade;\n        ''      \"\";\n    }\n\n"
 }
 
-func renderDefaultServerBlock(statusCode int) string {
+func renderDefaultServerBlock(statusCode int, http3Enabled bool) string {
 	if statusCode <= 0 {
 		statusCode = 421
 	}
+	var h3Default string
+	if http3Enabled {
+		h3Default = "\n        listen 443 quic reuseport default_server;"
+	}
 	return strings.Join([]string{
 		"    server {",
 		"        listen 80 default_server;",
@@ -599,7 +609,7 @@ func renderDefaultServerBlock(statusCode int) string {
 		"    }",
 		"",
 		"    server {",
-		"        listen 443 ssl default_server;",
+		fmt.Sprintf("        listen 443 ssl default_server;%s", h3Default),
 		"        server_name _;",
 		"",
 		"        ssl_reject_handshake on;",

openflare_server/utils/render/openresty/types.go

@@ -175,6 +175,7 @@ type ConfigSnapshot struct {
 	ProxySendTimeout          int    `json:"proxy_send_timeout"`
 	ProxyReadTimeout          int    `json:"proxy_read_timeout"`
 	WebsocketEnabled          bool   `json:"websocket_enabled"`
+	HTTP3Enabled              bool   `json:"http3_enabled"`
 	ProxyRequestBuffering     bool   `json:"proxy_request_buffering"`
 	ProxyBufferingEnabled     bool   `json:"proxy_buffering_enabled"`
 	ProxyBuffers              string `json:"proxy_buffers"`

openflare_server/utils/wsclient/client.go

@@ -0,0 +1,222 @@
+package wsclient
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"golang.org/x/net/websocket"
+)
+
+type Config struct {
+	BaseURL   string
+	Token     string
+	Timeout   time.Duration
+	HeaderKey string // e.g. "X-Agent-Token", "X-Tunnel-Token"
+	WSPath    string // e.g. "/api/relay/ws", "/api/agent/ws", "/api/flared/ws"
+}
+
+type Client struct {
+	cfg Config
+}
+
+type WSMessage struct {
+	Type    string          `json:"type"`
+	Payload json.RawMessage `json:"payload,omitempty"`
+}
+
+type MessageHandler interface {
+	OnConnect(ctx context.Context) error
+	HandleMessage(ctx context.Context, msg WSMessage) error
+	OnClose(err error)
+}
+
+type Connection struct {
+	Conn        *websocket.Conn
+	URL         string
+	ReadTimeout time.Duration
+}
+
+func New(cfg Config) *Client {
+	cfg.BaseURL = strings.TrimRight(cfg.BaseURL, "/")
+	cfg.Token = strings.TrimSpace(cfg.Token)
+	cfg.HeaderKey = strings.TrimSpace(cfg.HeaderKey)
+	cfg.WSPath = strings.TrimSpace(cfg.WSPath)
+	return &Client{
+		cfg: cfg,
+	}
+}
+
+func (c *Client) SetToken(token string) {
+	c.cfg.Token = strings.TrimSpace(token)
+}
+
+func (c *Client) URL() string {
+	wsURL, err := c.BuildWebsocketURL()
+	if err != nil {
+		return ""
+	}
+	return wsURL
+}
+
+func (c *Client) BuildWebsocketURL() (string, error) {
+	parsed, err := url.Parse(c.cfg.BaseURL)
+	if err != nil {
+		return "", err
+	}
+	switch parsed.Scheme {
+	case "http":
+		parsed.Scheme = "ws"
+	case "https":
+		parsed.Scheme = "wss"
+	case "ws", "wss":
+	default:
+		return "", errors.New("server_url scheme must be http, https, ws, or wss")
+	}
+
+	wsPath := c.cfg.WSPath
+	if !strings.HasPrefix(wsPath, "/") {
+		wsPath = "/" + wsPath
+	}
+	parsed.Path = strings.TrimRight(parsed.Path, "/") + wsPath
+	parsed.RawQuery = ""
+	parsed.Fragment = ""
+	return parsed.String(), nil
+}
+
+func (c *Client) Connect(ctx context.Context) (*Connection, error) {
+	wsURL, err := c.BuildWebsocketURL()
+	if err != nil {
+		return nil, err
+	}
+	if c.cfg.Token == "" {
+		return nil, errors.New("ws token is empty")
+	}
+	origin := c.cfg.BaseURL
+	if origin == "" {
+		origin = "http://localhost"
+	}
+	config, err := websocket.NewConfig(wsURL, origin)
+	if err != nil {
+		return nil, err
+	}
+	config.Header = http.Header{}
+	if c.cfg.HeaderKey != "" {
+		config.Header.Set(c.cfg.HeaderKey, c.cfg.Token)
+	}
+	if c.cfg.Timeout > 0 {
+		config.Dialer = &net.Dialer{Timeout: c.cfg.Timeout}
+	}
+	slog.Debug("ws dialing server", "url", wsURL)
+	conn, err := config.DialContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	slog.Debug("ws dial succeeded", "url", wsURL)
+	return &Connection{Conn: conn, URL: wsURL, ReadTimeout: websocketReadTimeout(c.cfg.Timeout)}, nil
+}
+
+func (conn *Connection) SendMessage(msgType string, payload any) error {
+	if conn == nil || conn.Conn == nil {
+		return errors.New("ws connection is nil")
+	}
+	slog.Debug("ws sending message", "type", msgType)
+
+	// Create the outbound message wrapper
+	message := struct {
+		Type    string `json:"type"`
+		Payload any    `json:"payload,omitempty"`
+	}{
+		Type:    msgType,
+		Payload: payload,
+	}
+
+	_ = conn.Conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
+	return websocket.JSON.Send(conn.Conn, message)
+}
+
+func (conn *Connection) Receive(target any) error {
+	if conn == nil || conn.Conn == nil {
+		return errors.New("ws connection is nil")
+	}
+	if conn.ReadTimeout > 0 {
+		_ = conn.Conn.SetReadDeadline(time.Now().Add(conn.ReadTimeout))
+	}
+	err := websocket.JSON.Receive(conn.Conn, target)
+	if err != nil {
+		var netErr net.Error
+		if errors.As(err, &netErr) && netErr.Timeout() {
+			slog.Debug("ws receive timeout waiting for server message", "timeout", conn.ReadTimeout)
+		}
+		return err
+	}
+	return nil
+}
+
+func websocketReadTimeout(requestTimeout time.Duration) time.Duration {
+	timeout := requestTimeout * 6
+	if timeout < 75*time.Second {
+		return 75 * time.Second
+	}
+	return timeout
+}
+
+func (conn *Connection) RunReceiveLoop(ctx context.Context, handler MessageHandler) error {
+	doneChan := make(chan struct{})
+	defer close(doneChan)
+
+	go func() {
+		select {
+		case <-ctx.Done():
+			_ = conn.Close()
+		case <-doneChan:
+		}
+	}()
+
+	if err := handler.OnConnect(ctx); err != nil {
+		handler.OnClose(err)
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		var raw WSMessage
+		if err := conn.Receive(&raw); err != nil {
+			handler.OnClose(err)
+			return err
+		}
+
+		switch raw.Type {
+		case "ping":
+			slog.Debug("ws received ping from server, replying with pong")
+			if err := conn.SendMessage("pong", nil); err != nil {
+				slog.Error("ws send pong response failed", "error", err)
+			}
+		case "pong":
+			slog.Debug("ws received pong response from server")
+		default:
+			if err := handler.HandleMessage(ctx, raw); err != nil {
+				slog.Error("ws handler failed to process message", "type", raw.Type, "error", err)
+				return err
+			}
+		}
+	}
+}
+
+func (conn *Connection) Close() error {
+	if conn == nil || conn.Conn == nil {
+		return nil
+	}
+	return conn.Conn.Close()
+}

openflare_server/web/features/nodes/utils.ts

@@ -211,7 +211,7 @@ export function buildNodeDockerInstallCommand(
     `docker pull ${image}`,
     `docker rm -f openflare-agent 2>/dev/null || true`,
     `docker run -d --name openflare-agent --restart unless-stopped \\`,
-    `  -p 80:80 -p 443:443 \\`,
+    `  -p 80:80 -p 443:443/tcp -p 443:443/udp \\`,
     `  -e OPENFLARE_SERVER_URL=${serverUrl} \\`,
     `  -e OPENFLARE_AGENT_TOKEN=${agentToken} \\`,
     `  ${image}`,

openflare_server/web/features/performance/components/performance-page.tsx

@@ -48,6 +48,7 @@ const defaultPerformanceFields = {
   OpenRestyProxySendTimeout: '60',
   OpenRestyProxyReadTimeout: '60',
   OpenRestyWebsocketEnabled: true,
+  OpenRestyHTTP3Enabled: true,
   OpenRestyProxyRequestBufferingEnabled: false,
   OpenRestyProxyBufferingEnabled: true,
   OpenRestyProxyBuffers: '16 16k',
@@ -96,6 +97,7 @@ const performanceFieldTooltips: Record<string, string> = {
   proxy_read_timeout: '等待上游返回响应的超时时间，单位秒。',
   websocket:
     '控制是否为反向代理规则自动注入 WebSocket 升级所需的 HTTP/1.1、Upgrade 和 Connection 头。',
+  http3: '启用后，OpenResty 将在 443 端口启用 HTTP/3 (QUIC) 协议支持。',
   proxy_request_buffering:
     '控制请求体是否先在 Nginx 侧缓冲后再转发给上游，上传和流式场景经常会用到。',
   proxy_buffering:
@@ -233,6 +235,10 @@ export function PerformancePage() {
         optionMap.OpenRestyWebsocketEnabled,
         true,
       ),
+      OpenRestyHTTP3Enabled: toBoolean(
+        optionMap.OpenRestyHTTP3Enabled,
+        false,
+      ),
       OpenRestyProxyRequestBufferingEnabled: toBoolean(
         optionMap.OpenRestyProxyRequestBufferingEnabled,
         false,
@@ -450,6 +456,10 @@ export function PerformancePage() {
             'OpenRestyWebsocketEnabled',
             String(performanceFields.OpenRestyWebsocketEnabled),
           ],
+          [
+            'OpenRestyHTTP3Enabled',
+            String(performanceFields.OpenRestyHTTP3Enabled),
+          ],
           [
             'OpenRestyProxyRequestBufferingEnabled',
             String(performanceFields.OpenRestyProxyRequestBufferingEnabled),
@@ -1008,6 +1018,17 @@ export function PerformancePage() {
                     }))
                   }
                 />
+                <ToggleField
+                  label="http3"
+                  tooltip={performanceFieldTooltips.http3}
+                  checked={performanceFields.OpenRestyHTTP3Enabled}
+                  onChange={(checked) =>
+                    setPerformanceFields((previous) => ({
+                      ...previous,
+                      OpenRestyHTTP3Enabled: checked,
+                    }))
+                  }
+                />
                 <ToggleField
                   label="proxy_request_buffering"
                   tooltip={performanceFieldTooltips.proxy_request_buffering}

openflare_server/web/features/waf/components/ip-groups-page.tsx

@@ -75,7 +75,7 @@ const typeLabels: Record<WAFIPGroupType, string> = {
 const automaticPresetRules = [
   {
     name: '单 IP 404 高频扫描',
-    expr: 'request_count > 100 && status_404_ratio >= 0.8',
+    expr: 'request_count > 100 && StatusRatio(404) >= 0.8',
   },
   {
     name: '单 IP 直连访问异常',
@@ -587,7 +587,7 @@ export function WAFIPGroupsPage() {
                 </ResourceField>
                 <ResourceField
                   label="自动配置 JSON"
-                  hint="可用字段：request_count、status_404_count、status_404_ratio、ip_host_count、ip_host_ratio。支持 ttl（秒，默认 -1 永久拉黑）。"
+                  hint="可用字段：request_count、status_404_count、status_404_ratio、ip_host_count、ip_host_ratio。方法：StatusCount(code)、StatusRatio(code)。支持 ttl（秒，默认 -1 永久拉黑）。"
                 >
                   <ResourceTextarea
                     value={draft.auto_config_text}

openflare_server/web/tests/unit/waf-ip-groups-page.test.tsx

@@ -200,7 +200,7 @@ describe('WAF IP groups', () => {
 
     const textarea = screen.getByLabelText(/自动配置 JSON/);
     const value = (textarea as HTMLTextAreaElement).value;
-    expect(value).toContain('request_count > 100 && status_404_ratio >= 0.8');
+    expect(value).toContain('request_count > 100 && StatusRatio(404) >= 0.8');
     expect(value).toContain('ip_host_count > 50 && ip_host_ratio > 0.5');
   });
 
@@ -260,7 +260,7 @@ describe('WAF IP groups', () => {
         lookback_minutes: 60,
         rules: [
           expect.objectContaining({
-            expr: 'request_count > 100 && status_404_ratio >= 0.8',
+            expr: 'request_count > 100 && StatusRatio(404) >= 0.8',
           }),
         ],
       }),

openflared/internal/flared/runner.go

@@ -49,31 +49,31 @@ func (r *Runner) Run(ctx context.Context) error {
 	}
 }
 
-func (r *Runner) handleConnection(ctx context.Context, conn *wsclient.Connection) {
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
+type flaredWSHandler struct {
+	runner *Runner
+}
 
-		msg, err := conn.Receive()
-		if err != nil {
-			slog.Error("flared ws receive failed", "error", err)
-			return
-		}
+func (h *flaredWSHandler) OnConnect(ctx context.Context) error {
+	return nil
+}
 
-		switch msg.Type {
-		case "ping":
-			_ = conn.SendPong()
-		case "active_config":
-			// Server notifies there is a new config available
-			slog.Info("received config update notification from server")
-			r.SyncService.Trigger()
-		default:
-			slog.Debug("ignored unknown ws message type", "type", msg.Type)
-		}
+func (h *flaredWSHandler) HandleMessage(ctx context.Context, msg wsclient.WSMessage) error {
+	switch msg.Type {
+	case "active_config":
+		slog.Info("received config update notification from server")
+		h.runner.SyncService.Trigger()
+	default:
+		slog.Debug("ignored unknown ws message type", "type", msg.Type)
 	}
+	return nil
+}
+
+func (h *flaredWSHandler) OnClose(err error) {
+	slog.Error("flared ws receive failed", "error", err)
+}
+
+func (r *Runner) handleConnection(ctx context.Context, conn *wsclient.Connection) {
+	_ = conn.RunReceiveLoop(ctx, &flaredWSHandler{runner: r})
 }
 
 func (r *Runner) sleepContext(ctx context.Context, d time.Duration) {

openflared/internal/frpc/manager.go

@@ -156,28 +156,57 @@ func (m *Manager) restartProcess(ctx context.Context, relayID string, configPath
 	m.processes[relayID] = proc
 
 	go func() {
+		backoff := 1 * time.Second
+		const maxBackoff = 60 * time.Second
+
 		for {
-			select {
-			case <-procCtx.Done():
+			m.mu.Lock()
+			if procCtx.Err() != nil {
+				m.mu.Unlock()
 				return
-			default:
 			}
+			m.mu.Unlock()
 
 			cmd := exec.CommandContext(procCtx, m.cfg.FrpcPath, "-c", configPath)
+
+			m.mu.Lock()
 			proc.Cmd = cmd
 			proc.Status = "running"
+			m.mu.Unlock()
 
+			startedAt := time.Now()
 			err := cmd.Run()
+
+			m.mu.Lock()
+			if procCtx.Err() != nil {
+				proc.Status = "stopped"
+				m.mu.Unlock()
+				return
+			}
+
 			if err != nil {
-				if procCtx.Err() != nil {
-					return
-				}
 				proc.LastError = err.Error()
 				proc.Status = "error"
 				slog.Error("frpc process exited unexpectedly", "relay_id", relayID, "error", err)
-				time.Sleep(5 * time.Second) // backoff
 			} else {
 				proc.Status = "stopped"
+				proc.LastError = "exited unexpectedly with code 0"
+				slog.Warn("frpc process exited unexpectedly with code 0", "relay_id", relayID)
+			}
+			m.mu.Unlock()
+
+			if time.Since(startedAt) >= 10*time.Second {
+				backoff = 1 * time.Second
+			}
+
+			select {
+			case <-procCtx.Done():
+				return
+			case <-time.After(backoff):
+				backoff = backoff * 2
+				if backoff > maxBackoff {
+					backoff = maxBackoff
+				}
 			}
 		}
 	}()

openflared/internal/frpc/manager_test.go

@@ -0,0 +1,267 @@
+package frpc
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"openflare-flared/internal/config"
+	"openflare/service"
+)
+
+// Helper to write control file for the dummy script
+func writeControl(t *testing.T, dir string, exitCode int, delaySeconds int) {
+	controlPath := filepath.Join(dir, "control.txt")
+	content := fmt.Sprintf("%d %d\n", exitCode, delaySeconds)
+	err := os.WriteFile(controlPath, []byte(content), 0644)
+	if err != nil {
+		t.Fatalf("failed to write control file: %v", err)
+	}
+}
+
+// Setup a dummy executable script that reads control.txt to decide exit code and sleep duration
+func setupDummyScript(t *testing.T) (string, string) {
+	dir := t.TempDir()
+	scriptPath := filepath.Join(dir, "dummy_frpc")
+
+	// On macOS/Linux, we write a shell script
+	scriptContent := fmt.Sprintf(`#!/bin/sh
+control_file="%s/control.txt"
+EXIT_CODE=0
+DELAY=0
+if [ -f "$control_file" ]; then
+    read -r EXIT_CODE DELAY < "$control_file"
+fi
+if [ -n "$DELAY" ] && [ "$DELAY" -gt 0 ] 2>/dev/null; then
+    sleep "$DELAY"
+fi
+exit "${EXIT_CODE:-0}"
+`, dir)
+
+	err := os.WriteFile(scriptPath, []byte(scriptContent), 0755)
+	if err != nil {
+		t.Fatalf("failed to write dummy script: %v", err)
+	}
+
+	return scriptPath, dir
+}
+
+// Helper to poll for status to eliminate timing flakiness in tests
+func assertStatusEventually(t *testing.T, m *Manager, relayID string, expectedStatus string, timeout time.Duration) {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		m.mu.RLock()
+		proc, ok := m.processes[relayID]
+		m.mu.RUnlock()
+		if ok && proc.Status == expectedStatus {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	m.mu.RLock()
+	proc, ok := m.processes[relayID]
+	var got string
+	var errStr string
+	if ok {
+		got = proc.Status
+		errStr = proc.LastError
+	} else {
+		got = "not_found"
+	}
+	m.mu.RUnlock()
+	t.Fatalf("expected status eventually %s, got %s (err: %s)", expectedStatus, got, errStr)
+}
+
+func TestStartProcessSuccess(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	writeControl(t, dir, 0, 5) // exit code 0, sleep 5s
+
+	cfg := &config.Config{
+		ServerURL:   "http://localhost:8080",
+		TunnelToken: "test-token",
+		FrpcPath:    scriptPath,
+		DataDir:     dir,
+		StatePath:   filepath.Join(dir, "flared-state.json"),
+	}
+
+	m := NewManager(cfg)
+
+	newConfig := &service.FlaredTunnelConfigResponse{
+		Version:  "1",
+		Checksum: "sum1",
+		Relays: []service.FlaredRelayInfo{
+			{
+				RelayNodeID: "relay-1",
+				Address:     "127.0.0.1:7000",
+				AuthToken:   "auth-1",
+			},
+		},
+		Proxies: nil,
+	}
+
+	err := m.UpdateConfig(context.Background(), newConfig)
+	if err != nil {
+		t.Fatalf("failed to UpdateConfig: %v", err)
+	}
+
+	assertStatusEventually(t, m, "relay-1", "running", 4*time.Second)
+
+	m.mu.RLock()
+	proc := m.processes["relay-1"]
+	m.mu.RUnlock()
+
+	proc.Cancel()
+	assertStatusEventually(t, m, "relay-1", "stopped", 4*time.Second) // wait for clean stop
+}
+
+func TestStartProcessFailureAndBackoff(t *testing.T) {
+	dir := t.TempDir()
+	invalidScriptPath := filepath.Join(dir, "non_existent_frpc")
+
+	cfg := &config.Config{
+		ServerURL:   "http://localhost:8080",
+		TunnelToken: "test-token",
+		FrpcPath:    invalidScriptPath,
+		DataDir:     dir,
+		StatePath:   filepath.Join(dir, "flared-state.json"),
+	}
+
+	m := NewManager(cfg)
+	newConfig := &service.FlaredTunnelConfigResponse{
+		Version:  "1",
+		Checksum: "sum1",
+		Relays: []service.FlaredRelayInfo{
+			{
+				RelayNodeID: "relay-1",
+				Address:     "127.0.0.1:7000",
+				AuthToken:   "auth-1",
+			},
+		},
+		Proxies: nil,
+	}
+
+	_ = m.UpdateConfig(context.Background(), newConfig)
+
+	assertStatusEventually(t, m, "relay-1", "error", 4*time.Second)
+
+	// Correct the path to dummy script
+	scriptPath, _ := setupDummyScript(t)
+	writeControl(t, filepath.Dir(scriptPath), 0, 5)
+
+	m.mu.Lock()
+	m.cfg.FrpcPath = scriptPath
+	m.mu.Unlock()
+
+	// Wait for backoff retry (1s backoff)
+	assertStatusEventually(t, m, "relay-1", "running", 4*time.Second)
+
+	m.mu.RLock()
+	proc := m.processes["relay-1"]
+	m.mu.RUnlock()
+	proc.Cancel()
+}
+
+func TestUnexpectedExit0CPUProtection(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	// Start with immediate exit code 0
+	writeControl(t, dir, 0, 0)
+
+	cfg := &config.Config{
+		ServerURL:   "http://localhost:8080",
+		TunnelToken: "test-token",
+		FrpcPath:    scriptPath,
+		DataDir:     dir,
+		StatePath:   filepath.Join(dir, "flared-state.json"),
+	}
+
+	m := NewManager(cfg)
+	newConfig := &service.FlaredTunnelConfigResponse{
+		Version:  "1",
+		Checksum: "sum1",
+		Relays: []service.FlaredRelayInfo{
+			{
+				RelayNodeID: "relay-1",
+				Address:     "127.0.0.1:7000",
+				AuthToken:   "auth-1",
+			},
+		},
+		Proxies: nil,
+	}
+
+	_ = m.UpdateConfig(context.Background(), newConfig)
+
+	assertStatusEventually(t, m, "relay-1", "stopped", 4*time.Second)
+
+	m.mu.RLock()
+	proc := m.processes["relay-1"]
+	if !strings.Contains(proc.LastError, "exited unexpectedly with code 0") {
+		t.Errorf("expected LastError to record exit status 0 warning, got %s", proc.LastError)
+	}
+	m.mu.RUnlock()
+
+	proc.Cancel()
+}
+
+func TestBackoffReset(t *testing.T) {
+	scriptPath, dir := setupDummyScript(t)
+	// Rapid exit code 1 to increase backoff
+	writeControl(t, dir, 1, 0)
+
+	cfg := &config.Config{
+		ServerURL:   "http://localhost:8080",
+		TunnelToken: "test-token",
+		FrpcPath:    scriptPath,
+		DataDir:     dir,
+		StatePath:   filepath.Join(dir, "flared-state.json"),
+	}
+
+	m := NewManager(cfg)
+	newConfig := &service.FlaredTunnelConfigResponse{
+		Version:  "1",
+		Checksum: "sum1",
+		Relays: []service.FlaredRelayInfo{
+			{
+				RelayNodeID: "relay-1",
+				Address:     "127.0.0.1:7000",
+				AuthToken:   "auth-1",
+			},
+		},
+		Proxies: nil,
+	}
+
+	_ = m.UpdateConfig(context.Background(), newConfig)
+
+	// Wait to crash
+	assertStatusEventually(t, m, "relay-1", "error", 4*time.Second)
+
+	// Now make it run successfully for 11 seconds (exit code 0, sleep 11s)
+	writeControl(t, dir, 0, 11)
+
+	// Wait for next retry to start running
+	assertStatusEventually(t, m, "relay-1", "running", 4*time.Second)
+
+	// Wait for process to run for 10.5 seconds to trigger backoff reset
+	time.Sleep(10500 * time.Millisecond)
+
+	// Now make it crash again (exit code 1, sleep 0s)
+	writeControl(t, dir, 1, 0)
+
+	// Wait for it to finish and crash
+	assertStatusEventually(t, m, "relay-1", "error", 4*time.Second)
+
+	// It crashed. Since it ran for > 10s, backoff should have been reset to 1s.
+	// We make it healthy again (exit code 0, sleep 5)
+	writeControl(t, dir, 0, 5)
+
+	// Wait 1.5 seconds. If backoff was reset to 1s, it should be running now.
+	assertStatusEventually(t, m, "relay-1", "running", 4*time.Second)
+
+	m.mu.RLock()
+	proc := m.processes["relay-1"]
+	m.mu.RUnlock()
+	proc.Cancel()
+}

openflared/internal/wsclient/client.go

@@ -3,141 +3,73 @@ package wsclient
 import (
 	"context"
 	"encoding/json"
-	"errors"
-	"log/slog"
-	"net"
-	"net/http"
-	"net/url"
-	"strings"
 	"time"
 
-	"golang.org/x/net/websocket"
 	"openflare/service"
+	shared "openflare/utils/wsclient"
 )
 
+type WSMessage = shared.WSMessage
+type MessageHandler = shared.MessageHandler
+
 type Client struct {
-	baseURL string
-	token   string
-	timeout time.Duration
+	sharedClient *shared.Client
 }
 
 type Connection struct {
-	conn        *websocket.Conn
-	url         string
-	readTimeout time.Duration
+	sharedConn *shared.Connection
 }
 
 func New(baseURL string, token string, timeout time.Duration) *Client {
 	return &Client{
-		baseURL: strings.TrimRight(baseURL, "/"),
-		token:   strings.TrimSpace(token),
-		timeout: timeout,
+		sharedClient: shared.New(shared.Config{
+			BaseURL:   baseURL,
+			Token:     token,
+			Timeout:   timeout,
+			HeaderKey: "X-Tunnel-Token",
+			WSPath:    "/api/flared/ws",
+		}),
 	}
 }
 
 func (c *Client) SetToken(token string) {
-	c.token = strings.TrimSpace(token)
-	slog.Debug("flared ws client token updated")
+	c.sharedClient.SetToken(token)
 }
 
 func (c *Client) Connect(ctx context.Context) (*Connection, error) {
-	wsURL, err := buildWebsocketURL(c.baseURL)
+	conn, err := c.sharedClient.Connect(ctx)
 	if err != nil {
 		return nil, err
 	}
-	if strings.TrimSpace(c.token) == "" {
-		return nil, errors.New("flared ws token is empty")
-	}
-	origin := strings.TrimSpace(c.baseURL)
-	if origin == "" {
-		origin = "http://localhost"
-	}
-	config, err := websocket.NewConfig(wsURL, origin)
-	if err != nil {
-		return nil, err
-	}
-	config.Header = http.Header{}
-	config.Header.Set("X-Tunnel-Token", c.token)
-	if c.timeout > 0 {
-		config.Dialer = &net.Dialer{Timeout: c.timeout}
-	}
-	slog.Debug("flared ws dialing server", "url", wsURL)
-	conn, err := config.DialContext(ctx)
-	if err != nil {
-		return nil, err
-	}
-	slog.Debug("flared ws dial succeeded", "url", wsURL)
-	return &Connection{conn: conn, url: wsURL, readTimeout: websocketReadTimeout(c.timeout)}, nil
+	return &Connection{sharedConn: conn}, nil
 }
 
-func buildWebsocketURL(baseURL string) (string, error) {
-	parsed, err := url.Parse(strings.TrimRight(baseURL, "/"))
-	if err != nil {
-		return "", err
-	}
-	switch parsed.Scheme {
-	case "http":
-		parsed.Scheme = "ws"
-	case "https":
-		parsed.Scheme = "wss"
-	case "ws", "wss":
-	default:
-		return "", errors.New("server_url scheme must be http, https, ws, or wss")
-	}
-	parsed.Path = strings.TrimRight(parsed.Path, "/") + "/api/flared/ws"
-	parsed.RawQuery = ""
-	parsed.Fragment = ""
-	return parsed.String(), nil
+func (conn *Connection) SendPing() error {
+	return conn.sharedConn.SendMessage("ping", nil)
 }
 
 func (conn *Connection) SendPong() error {
-	if conn == nil || conn.conn == nil {
-		return errors.New("flared ws connection is nil")
-	}
-	slog.Debug("flared ws sending pong")
-	return websocket.JSON.Send(conn.conn, service.WSMessage{
-		Type: "pong",
-	})
+	return conn.sharedConn.SendMessage("pong", nil)
 }
 
 func (conn *Connection) Receive() (service.WSMessage, error) {
-	var message service.WSMessage
-	if conn == nil || conn.conn == nil {
-		return message, errors.New("flared ws connection is nil")
-	}
-	if conn.readTimeout > 0 {
-		_ = conn.conn.SetReadDeadline(time.Now().Add(conn.readTimeout))
-	}
-	// Use custom json unmarshaling to handle any type
 	var raw struct {
 		Type    string          `json:"type"`
 		Payload json.RawMessage `json:"payload,omitempty"`
 	}
-	err := websocket.JSON.Receive(conn.conn, &raw)
-	if err != nil {
-		var netErr net.Error
-		if errors.As(err, &netErr) && netErr.Timeout() {
-			slog.Debug("flared ws receive timeout waiting for server message", "timeout", conn.readTimeout)
-		}
-		return message, err
+	if err := conn.sharedConn.Receive(&raw); err != nil {
+		return service.WSMessage{}, err
 	}
-	message.Type = raw.Type
-	message.Payload = raw.Payload
-	slog.Debug("flared ws received message", "type", message.Type)
-	return message, nil
+	return service.WSMessage{
+		Type:    raw.Type,
+		Payload: raw.Payload,
+	}, nil
 }
 
-func websocketReadTimeout(requestTimeout time.Duration) time.Duration {
-	timeout := requestTimeout * 6
-	if timeout < 75*time.Second {
-		return 75 * time.Second
-	}
-	return timeout
+func (conn *Connection) RunReceiveLoop(ctx context.Context, handler shared.MessageHandler) error {
+	return conn.sharedConn.RunReceiveLoop(ctx, handler)
 }
 
 func (conn *Connection) Close() error {
-	if conn == nil || conn.conn == nil {
-		return nil
-	}
-	return conn.conn.Close()
+	return conn.sharedConn.Close()
 }