fix: reduce reconnect redeploy and metrics load (#476)

* fix: reduce reconnect redeploy and metrics load

Throttle node-online redeploy retries and lower the agent metric cadence so brief reconnect churn no longer fans out into repeated runtime syncs and backend connection pressure.

* docs: add follow-up implementation design notes

Document the planned flow upload batching work and the local remote-address toggle so the next changesets can implement them against an agreed design.

docs/superpowers/plans/2026-04-26-allow-local-remote-addr.md

@@ -0,0 +1,171 @@
+# Allow Local Remote Address Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add a global settings toggle that allows non-admin forward rules to target local/private addresses when explicitly enabled.
+
+**Architecture:** Keep the existing remote-address safety validator as the default path for non-admin rule changes, but gate its use behind a single backend config lookup in forward create/update handlers. Surface the toggle through the existing `vite_config` settings page and prove behavior with backend contract tests first.
+
+**Tech Stack:** Go `net/http` + GORM backend, React + TypeScript frontend settings page, Go contract tests.
+
+---
+
+### Task 1: Backend Contract Coverage
+
+**Files:**
+- Modify: `go-backend/tests/contract/forward_contract_test.go`
+
+- [ ] **Step 1: Write the failing tests**
+
+Add contract tests that prove the desired behavior:
+
+```go
+t.Run("local remote address is rejected when toggle is off", func(t *testing.T) {
+	createPayload := map[string]interface{}{
+		"name":       "deny-local-remote",
+		"tunnelId":   tunnelID,
+		"remoteAddr": "127.0.0.1:8080",
+		"strategy":   "fifo",
+	}
+	createBody, _ := json.Marshal(createPayload)
+	createReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(createBody))
+	createReq.Header.Set("Authorization", adminToken)
+	createReq.Header.Set("Content-Type", "application/json")
+	createRes := httptest.NewRecorder()
+	router.ServeHTTP(createRes, createReq)
+
+	var out response.R
+	_ = json.NewDecoder(createRes.Body).Decode(&out)
+	if out.Code == 0 {
+		t.Fatalf("expected local remote address to be rejected when toggle is off")
+	}
+})
+
+t.Run("local remote address is allowed when toggle is on", func(t *testing.T) {
+	if err := repo.DB().Exec(`
+		INSERT INTO vite_config(name, value, time)
+		VALUES(?, ?, ?)
+		ON CONFLICT(name) DO UPDATE SET value = excluded.value, time = excluded.time
+	`, "allow_local_remote_addr", "1", time.Now().UnixMilli()).Error; err != nil {
+		t.Fatalf("enable allow_local_remote_addr: %v", err)
+	}
+
+	createPayload := map[string]interface{}{
+		"name":       "allow-local-remote",
+		"tunnelId":   tunnelID,
+		"remoteAddr": "127.0.0.1:8080",
+		"strategy":   "fifo",
+	}
+	createBody, _ := json.Marshal(createPayload)
+	createReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(createBody))
+	createReq.Header.Set("Authorization", adminToken)
+	createReq.Header.Set("Content-Type", "application/json")
+	createRes := httptest.NewRecorder()
+	router.ServeHTTP(createRes, createReq)
+	assertCode(t, createRes, 0)
+})
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `go test ./tests/contract/... -run 'TestForwardContracts|local remote address'`
+Expected: FAIL because backend still rejects local/private addresses unconditionally.
+
+- [ ] **Step 3: Commit**
+
+Do not commit yet; combine with Task 2 after implementation passes.
+
+### Task 2: Backend Toggle Implementation
+
+**Files:**
+- Modify: `go-backend/internal/http/handler/mutations.go`
+
+- [ ] **Step 1: Add a tiny config helper**
+
+Add a helper near other handler helpers:
+
+```go
+func (h *Handler) allowLocalRemoteAddr() bool {
+	if h == nil || h.repo == nil {
+		return false
+	}
+	cfg, err := h.repo.GetConfigByName("allow_local_remote_addr")
+	if err != nil || cfg == nil {
+		return false
+	}
+	return strings.TrimSpace(cfg.Value) == "1"
+}
+```
+
+- [ ] **Step 2: Gate create/update validation behind the helper**
+
+Replace the unconditional checks with:
+
+```go
+if !h.allowLocalRemoteAddr() {
+	if err := IsSafeRemoteAddr(remoteAddr); err != nil {
+		response.WriteJSON(w, response.Err(403, err.Error()))
+		return
+	}
+}
+```
+
+- [ ] **Step 3: Run contract tests to verify they pass**
+
+Run: `go test ./tests/contract/... -run 'TestForwardContracts|local remote address'`
+Expected: PASS
+
+- [ ] **Step 4: Run full backend tests**
+
+Run: `go test ./...`
+Expected: PASS
+
+### Task 3: Settings Page Toggle
+
+**Files:**
+- Modify: `vite-frontend/src/pages/config.tsx`
+
+- [ ] **Step 1: Add the config item to the settings schema**
+
+Add a switch-style item for `allow_local_remote_addr` with warning copy about reduced safety.
+
+- [ ] **Step 2: Ensure the key is included in config loading/saving paths**
+
+Add `allow_local_remote_addr` anywhere the page enumerates config keys or groups persisted config values.
+
+- [ ] **Step 3: Run frontend build**
+
+Run: `pnpm run build`
+Expected: PASS
+
+- [ ] **Step 4: Run frontend lint**
+
+Run: `pnpm run lint`
+Expected: 0 errors; existing warnings may remain.
+
+### Task 4: Final Verification
+
+**Files:**
+- Verify only
+
+- [ ] **Step 1: Re-run backend contracts for the toggle**
+
+Run: `go test ./tests/contract/... -run 'TestForwardContracts|local remote address'`
+Expected: PASS
+
+- [ ] **Step 2: Re-run full backend tests**
+
+Run: `go test ./...`
+Expected: PASS
+
+- [ ] **Step 3: Re-run frontend build/lint**
+
+Run: `pnpm run build && pnpm run lint`
+Expected: Build passes, lint has no errors.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add go-backend/internal/http/handler/mutations.go go-backend/tests/contract/forward_contract_test.go vite-frontend/src/pages/config.tsx docs/superpowers/specs/2026-04-26-allow-local-remote-addr-design.md docs/superpowers/plans/2026-04-26-allow-local-remote-addr.md
+git commit -m "feat: add allow-local-remote-address toggle"
+```

docs/superpowers/plans/2026-04-26-flow-upload-batch-optimization.md

@@ -0,0 +1,849 @@
+# flow/upload Batch Optimization Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Reduce `POST /flow/upload` database pressure by converting the hot path from per-item queries and per-item transactions to per-request aggregation, batched metadata reads, and batched writes, while preserving immediate quota disable / forward pause behavior inside the same upload.
+
+**Architecture:** Parse one upload into a batch object in the handler layer, fetch one shared `forward+tunnel` metadata map, then reuse that map for flow accounting and tunnel metric aggregation. Replace `AddFlow` and `AddUserQuotaUsage` per-item transactions with one batched flow transaction and one batched quota transaction; run policy enforcement, orphan cleanup, and peer-share flow handling once per affected target instead of once per item.
+
+**Tech Stack:** Go, net/http, GORM, SQLite/PostgreSQL, existing backend contract tests.
+
+---
+
+## File Map
+
+- Create: `go-backend/internal/http/handler/flow_upload_batch.go`
+  Responsibility: request-scoped parsing, aggregation, and application of one `/flow/upload` batch.
+- Create: `go-backend/internal/http/handler/flow_upload_batch_test.go`
+  Responsibility: unit coverage for batch aggregation semantics.
+- Create: `go-backend/internal/store/repo/repository_flow_batch_test.go`
+  Responsibility: unit coverage for batched flow and quota persistence.
+- Create: `go-backend/tests/contract/flow_upload_batch_contract_test.go`
+  Responsibility: contract coverage that repeated items still accumulate correctly and still disable quota immediately.
+- Modify: `go-backend/internal/http/handler/handler.go`
+  Responsibility: switch `/flow/upload` entrypoint to the new batch pipeline.
+- Modify: `go-backend/internal/http/handler/tunnel_metrics_ingestion.go`
+  Responsibility: accept pre-aggregated forward deltas plus shared forward metadata instead of reparsing the raw items.
+- Modify: `go-backend/internal/store/repo/repository.go`
+  Responsibility: add batched flow persistence primitives near the existing flow update code.
+- Modify: `go-backend/internal/store/repo/repository_flow.go`
+  Responsibility: add shared flow-upload metadata query helpers.
+- Modify: `go-backend/internal/store/repo/repository_user_quota.go`
+  Responsibility: add batched quota usage persistence that still returns normalized quota views for immediate enforcement.
+
+---
+
+### Task 1: Add Failing Tests For Batched flow/upload Semantics
+
+**Files:**
+- Create: `go-backend/internal/http/handler/flow_upload_batch_test.go`
+- Create: `go-backend/tests/contract/flow_upload_batch_contract_test.go`
+
+- [ ] **Step 1: Write the failing handler unit test**
+
+Create `go-backend/internal/http/handler/flow_upload_batch_test.go` with a unit test that locks in the new aggregation contract.
+
+```go
+package handler
+
+import (
+	"testing"
+
+	"go-backend/internal/store/repo"
+)
+
+func TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets(t *testing.T) {
+	h := &Handler{}
+	metas := map[int64]repo.FlowUploadForwardMeta{
+		20: {
+			ForwardID:     20,
+			TunnelID:      1,
+			TrafficRatio:  2,
+			TunnelFlow:    3,
+		},
+	}
+
+	batch := h.buildFlowUploadBatch([]flowItem{
+		{N: "20_2_10", U: 70, D: 50},
+		{N: "20_2_10_tcp", U: 40, D: 30},
+		{N: "99_2_10", U: 12, D: 8},
+		{N: "fed_svc_17", U: 9, D: 1},
+	}, metas)
+
+	if len(batch.flowDeltas) != 1 {
+		t.Fatalf("expected 1 flow delta, got %d", len(batch.flowDeltas))
+	}
+	delta := batch.flowDeltas[0]
+	if delta.ForwardID != 20 || delta.UserID != 2 || delta.UserTunnelID != 10 {
+		t.Fatalf("unexpected flow delta identity: %#v", delta)
+	}
+	if delta.InFlow != 480 || delta.OutFlow != 660 {
+		t.Fatalf("expected scaled flow in=480 out=660, got in=%d out=%d", delta.InFlow, delta.OutFlow)
+	}
+	if batch.quotaUsage[2] != 1140 {
+		t.Fatalf("expected quota usage 1140, got %d", batch.quotaUsage[2])
+	}
+	if len(batch.policyTargets) != 1 {
+		t.Fatalf("expected 1 policy target, got %d", len(batch.policyTargets))
+	}
+	if batch.policyTargets[0].UserID != 2 || batch.policyTargets[0].UserTunnelID != 10 {
+		t.Fatalf("unexpected policy target: %#v", batch.policyTargets[0])
+	}
+	traffic := batch.forwardTraffic[20]
+	if traffic.bytesIn != 80 || traffic.bytesOut != 110 {
+		t.Fatalf("expected raw traffic in=80 out=110, got in=%d out=%d", traffic.bytesIn, traffic.bytesOut)
+	}
+	if _, ok := batch.orphanServices["99_2_10"]; !ok {
+		t.Fatalf("expected orphan service cleanup target for 99_2_10")
+	}
+	if item, ok := batch.peerShareForwardItems["20_2_10"]; !ok || item.U != 110 || item.D != 80 {
+		t.Fatalf("expected merged peer-share forward item, got %#v ok=%v", item, ok)
+	}
+	if item, ok := batch.peerShareRuntimeItems[17]; !ok || item.U != 9 || item.D != 1 {
+		t.Fatalf("expected merged peer-share runtime item, got %#v ok=%v", item, ok)
+	}
+}
+```
+
+- [ ] **Step 2: Run the handler unit test to verify RED**
+
+Run:
+
+```bash
+go test ./internal/http/handler -run TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets -v
+```
+
+Expected: FAIL because `FlowUploadForwardMeta`, `buildFlowUploadBatch`, and the new batch fields do not exist yet.
+
+- [ ] **Step 3: Write the contract test that guards current behavior**
+
+Create `go-backend/tests/contract/flow_upload_batch_contract_test.go` so the optimization cannot weaken same-request quota enforcement.
+
+```go
+package contract_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"go-backend/internal/store/model"
+)
+
+func TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately(t *testing.T) {
+	secret := "monitoring-jwt-secret"
+	router, repo := setupContractRouter(t, secret)
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	dayKey := int64(now.Year()*10000 + int(now.Month())*100 + now.Day())
+	monthKey := int64(now.Year()*100 + int(now.Month()))
+	const bytesPerGB = int64(1024 * 1024 * 1024)
+
+	node := &model.Node{Name: "node-1", Secret: "node-secret", ServerIP: "127.0.0.1", Port: "10000-10010", TCPListenAddr: "[::]", UDPListenAddr: "[::]", CreatedTime: nowMs, Status: 1}
+	if err := repo.DB().Create(node).Error; err != nil {
+		t.Fatalf("seed node: %v", err)
+	}
+	if err := repo.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'flow_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	tunnel := &model.Tunnel{Name: "tunnel-1", TrafficRatio: 1.0, Type: 1, Protocol: "tls", Flow: 1, CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}
+	if err := repo.DB().Create(tunnel).Error; err != nil {
+		t.Fatalf("seed tunnel: %v", err)
+	}
+	if err := repo.DB().Exec(`INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(10, 2, ?, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)`, tunnel.ID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+	forward := &model.Forward{ID: 20, UserID: 2, UserName: "flow_user", Name: "forward-20", TunnelID: tunnel.ID, RemoteAddr: "1.1.1.1:80", Strategy: "fifo", CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}
+	if err := repo.DB().Create(forward).Error; err != nil {
+		t.Fatalf("seed forward: %v", err)
+	}
+	if err := repo.DB().Exec(`INSERT INTO user_quota(user_id, daily_limit_gb, monthly_limit_gb, daily_used_bytes, monthly_used_bytes, day_key, month_key, disabled_by_quota, disabled_at, paused_forward_ids, created_time, updated_time) VALUES(2, 1, 0, ?, ?, ?, ?, 0, 0, '', ?, ?)`, bytesPerGB-100, bytesPerGB-100, dayKey, monthKey, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user_quota: %v", err)
+	}
+
+	body, err := json.Marshal([]map[string]interface{}{
+		{"n": "20_2_10", "u": 70, "d": 50},
+		{"n": "20_2_10_tcp", "u": 40, "d": 30},
+	})
+	if err != nil {
+		t.Fatalf("marshal body: %v", err)
+	}
+	req := httptest.NewRequest(http.MethodPost, "/flow/upload?secret="+node.Secret, bytes.NewReader(body))
+	res := httptest.NewRecorder()
+	router.ServeHTTP(res, req)
+
+	if res.Code != http.StatusOK {
+		t.Fatalf("expected status 200, got %d", res.Code)
+	}
+	if got := mustQueryInt(t, repo, `SELECT status FROM forward WHERE id = 20`); got != 0 {
+		t.Fatalf("expected forward paused immediately, got status=%d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT disabled_by_quota FROM user_quota WHERE user_id = 2`); got != 1 {
+		t.Fatalf("expected quota disabled flag=1, got %d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT in_flow FROM forward WHERE id = 20`); got != 80 {
+		t.Fatalf("expected forward in_flow=80, got %d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT out_flow FROM forward WHERE id = 20`); got != 110 {
+		t.Fatalf("expected forward out_flow=110, got %d", got)
+	}
+	metrics, err := repo.GetTunnelMetrics(tunnel.ID, 0, nowMs+60_000)
+	if err != nil {
+		t.Fatalf("get tunnel metrics: %v", err)
+	}
+	if len(metrics) != 1 || metrics[0].BytesIn != 80 || metrics[0].BytesOut != 110 {
+		t.Fatalf("expected one aggregated metric row, got %#v", metrics)
+	}
+}
+```
+
+- [ ] **Step 4: Run the contract test to verify the same-request guard stays green or reveals an existing regression**
+
+Run:
+
+```bash
+go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -v
+```
+
+Expected: this test may already PASS before the refactor because it locks in existing external behavior. Keep it either way; it is the guardrail for the optimization.
+
+- [ ] **Step 5: Optional commit if the user explicitly requested commits**
+
+```bash
+git add go-backend/internal/http/handler/flow_upload_batch_test.go go-backend/tests/contract/flow_upload_batch_contract_test.go
+git commit -m "test: cover flow upload batch semantics"
+```
+
+---
+
+### Task 2: Add Batched Repository Primitives
+
+**Files:**
+- Modify: `go-backend/internal/store/repo/repository_flow.go`
+- Modify: `go-backend/internal/store/repo/repository.go`
+- Modify: `go-backend/internal/store/repo/repository_user_quota.go`
+- Create: `go-backend/internal/store/repo/repository_flow_batch_test.go`
+
+- [ ] **Step 1: Write the failing repository tests**
+
+Create `go-backend/internal/store/repo/repository_flow_batch_test.go` with coverage for both the shared metadata query and the batched counter/quota writes.
+
+```go
+package repo
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch(t *testing.T) {
+	r, err := Open(filepath.Join(t.TempDir(), "flow-batch.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now().UnixMilli()
+	if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) VALUES(1, 't1', 2.0, 1, 'tls', 3, ?, ?, 1, NULL, 0)`, now, now).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(10, 2, 1, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)`).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx) VALUES(20, 2, 'u2', 'f20', 1, '1.1.1.1:80', 'fifo', 0, 0, ?, ?, 1, 0)`, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+
+	metas, err := r.GetFlowUploadForwardMetas([]int64{20, 99})
+	if err != nil {
+		t.Fatalf("get metas: %v", err)
+	}
+	if metas[20].TunnelID != 1 || metas[20].TrafficRatio != 2 || metas[20].TunnelFlow != 3 {
+		t.Fatalf("unexpected meta for forward 20: %#v", metas[20])
+	}
+	if _, ok := metas[99]; ok {
+		t.Fatalf("did not expect meta for missing forward 99")
+	}
+
+	err = r.ApplyFlowUploadDeltasBatch([]FlowUploadCounterDelta{{ForwardID: 20, UserID: 2, UserTunnelID: 10, InFlow: 480, OutFlow: 660}})
+	if err != nil {
+		t.Fatalf("apply flow batch: %v", err)
+	}
+	if got := mustFlowBatchCount(t, r, `SELECT in_flow FROM forward WHERE id = 20`); got != 480 {
+		t.Fatalf("expected forward in_flow=480, got %d", got)
+	}
+	if got := mustFlowBatchCount(t, r, `SELECT out_flow FROM user WHERE id = 2`); got != 660 {
+		t.Fatalf("expected user out_flow=660, got %d", got)
+	}
+	if got := mustFlowBatchCount(t, r, `SELECT in_flow FROM user_tunnel WHERE id = 10`); got != 480 {
+		t.Fatalf("expected user_tunnel in_flow=480, got %d", got)
+	}
+}
+
+func TestAddUserQuotaUsageBatchReturnsNormalizedViews(t *testing.T) {
+	r, err := Open(filepath.Join(t.TempDir(), "quota-batch.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	views, err := r.AddUserQuotaUsageBatch(map[int64]int64{2: 1140}, now)
+	if err != nil {
+		t.Fatalf("batch quota update: %v", err)
+	}
+	if views[2] == nil || views[2].DailyUsedBytes != 1140 || views[2].MonthlyUsedBytes != 1140 {
+		t.Fatalf("unexpected quota view: %#v", views[2])
+	}
+}
+
+func mustFlowBatchCount(t *testing.T, r *Repository, query string, args ...interface{}) int64 {
+	t.Helper()
+	var value int64
+	if err := r.DB().Raw(query, args...).Row().Scan(&value); err != nil {
+		t.Fatalf("query %q failed: %v", query, err)
+	}
+	return value
+}
+```
+
+- [ ] **Step 2: Run the repository tests to verify RED**
+
+Run:
+
+```bash
+go test ./internal/store/repo -run 'TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch|TestAddUserQuotaUsageBatchReturnsNormalizedViews' -v
+```
+
+Expected: FAIL because `GetFlowUploadForwardMetas`, `ApplyFlowUploadDeltasBatch`, `FlowUploadCounterDelta`, and `AddUserQuotaUsageBatch` do not exist yet.
+
+- [ ] **Step 3: Implement shared flow-upload metadata and batched persistence**
+
+Update `go-backend/internal/store/repo/repository_flow.go`, `repository.go`, and `repository_user_quota.go` with the following concrete APIs. Add `sort` to the `repository_user_quota.go` import list.
+
+```go
+// repository_flow.go
+type FlowUploadForwardMeta struct {
+	ForwardID    int64
+	TunnelID     int64
+	TrafficRatio float64
+	TunnelFlow   int64
+}
+
+func (r *Repository) GetFlowUploadForwardMetas(forwardIDs []int64) (map[int64]FlowUploadForwardMeta, error) {
+	if r == nil || r.db == nil {
+		return nil, errors.New("repository not initialized")
+	}
+	if len(forwardIDs) == 0 {
+		return map[int64]FlowUploadForwardMeta{}, nil
+	}
+	ids := make([]int64, 0, len(forwardIDs))
+	seen := make(map[int64]struct{}, len(forwardIDs))
+	for _, id := range forwardIDs {
+		if id <= 0 {
+			continue
+		}
+		if _, ok := seen[id]; ok {
+			continue
+		}
+		seen[id] = struct{}{}
+		ids = append(ids, id)
+	}
+	type row struct {
+		ForwardID    int64   `gorm:"column:forward_id"`
+		TunnelID     int64   `gorm:"column:tunnel_id"`
+		TrafficRatio float64 `gorm:"column:traffic_ratio"`
+		TunnelFlow   int64   `gorm:"column:tunnel_flow"`
+	}
+	var rows []row
+	err := r.db.Table("forward AS f").
+		Select("f.id AS forward_id, f.tunnel_id AS tunnel_id, t.traffic_ratio AS traffic_ratio, t.flow AS tunnel_flow").
+		Joins("JOIN tunnel t ON t.id = f.tunnel_id").
+		Where("f.id IN ?", ids).
+		Scan(&rows).Error
+	if err != nil {
+		return nil, err
+	}
+	out := make(map[int64]FlowUploadForwardMeta, len(rows))
+	for _, row := range rows {
+		if row.TunnelFlow <= 0 {
+			row.TunnelFlow = 1
+		}
+		if row.TrafficRatio <= 0 {
+			row.TrafficRatio = 1
+		}
+		out[row.ForwardID] = FlowUploadForwardMeta{ForwardID: row.ForwardID, TunnelID: row.TunnelID, TrafficRatio: row.TrafficRatio, TunnelFlow: row.TunnelFlow}
+	}
+	return out, nil
+}
+```
+
+```go
+// repository.go
+type FlowUploadCounterDelta struct {
+	ForwardID    int64
+	UserID       int64
+	UserTunnelID int64
+	InFlow       int64
+	OutFlow      int64
+}
+
+func (r *Repository) ApplyFlowUploadDeltasBatch(deltas []FlowUploadCounterDelta) error {
+	if r == nil || r.db == nil {
+		return errors.New("repository not initialized")
+	}
+	if len(deltas) == 0 {
+		return nil
+	}
+	forwardTotals := make(map[int64][2]int64, len(deltas))
+	userTotals := make(map[int64][2]int64, len(deltas))
+	userTunnelTotals := make(map[int64][2]int64, len(deltas))
+	for _, delta := range deltas {
+		if delta.ForwardID > 0 {
+			current := forwardTotals[delta.ForwardID]
+			current[0] += delta.InFlow
+			current[1] += delta.OutFlow
+			forwardTotals[delta.ForwardID] = current
+		}
+		if delta.UserID > 0 {
+			current := userTotals[delta.UserID]
+			current[0] += delta.InFlow
+			current[1] += delta.OutFlow
+			userTotals[delta.UserID] = current
+		}
+		if delta.UserTunnelID > 0 {
+			current := userTunnelTotals[delta.UserTunnelID]
+			current[0] += delta.InFlow
+			current[1] += delta.OutFlow
+			userTunnelTotals[delta.UserTunnelID] = current
+		}
+	}
+	return r.db.Transaction(func(tx *gorm.DB) error {
+		for forwardID, total := range forwardTotals {
+			if err := tx.Model(&model.Forward{}).Where("id = ?", forwardID).UpdateColumns(map[string]interface{}{"in_flow": gorm.Expr("in_flow + ?", total[0]), "out_flow": gorm.Expr("out_flow + ?", total[1])}).Error; err != nil {
+				return err
+			}
+		}
+		for userID, total := range userTotals {
+			if err := tx.Model(&model.User{}).Where("id = ?", userID).UpdateColumns(map[string]interface{}{"in_flow": gorm.Expr("in_flow + ?", total[0]), "out_flow": gorm.Expr("out_flow + ?", total[1])}).Error; err != nil {
+				return err
+			}
+		}
+		for userTunnelID, total := range userTunnelTotals {
+			if err := tx.Model(&model.UserTunnel{}).Where("id = ?", userTunnelID).UpdateColumns(map[string]interface{}{"in_flow": gorm.Expr("in_flow + ?", total[0]), "out_flow": gorm.Expr("out_flow + ?", total[1])}).Error; err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+```
+
+```go
+// repository_user_quota.go
+func (r *Repository) AddUserQuotaUsageBatch(usages map[int64]int64, now time.Time) (map[int64]*model.UserQuotaView, error) {
+	if r == nil || r.db == nil {
+		return nil, errors.New("repository not initialized")
+	}
+	if len(usages) == 0 {
+		return map[int64]*model.UserQuotaView{}, nil
+	}
+	result := make(map[int64]*model.UserQuotaView, len(usages))
+	err := r.db.Transaction(func(tx *gorm.DB) error {
+		userIDs := make([]int64, 0, len(usages))
+		for userID := range usages {
+			if userID > 0 {
+				userIDs = append(userIDs, userID)
+			}
+		}
+		sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] })
+		for _, userID := range userIDs {
+			q, err := r.loadOrCreateUserQuotaTx(tx, userID, now)
+			if err != nil {
+				return err
+			}
+			applyUserQuotaWindowRoll(q, now)
+			if usages[userID] > 0 {
+				q.DailyUsedBytes += usages[userID]
+				q.MonthlyUsedBytes += usages[userID]
+			}
+			q.UpdatedTime = now.UnixMilli()
+			if err := tx.Model(&model.UserQuota{}).Where("user_id = ?", userID).Updates(map[string]interface{}{"daily_used_bytes": q.DailyUsedBytes, "monthly_used_bytes": q.MonthlyUsedBytes, "day_key": q.DayKey, "month_key": q.MonthKey, "updated_time": q.UpdatedTime}).Error; err != nil {
+				return err
+			}
+			result[userID] = normalizeUserQuotaView(cloneUserQuotaView(*q), now)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+```
+
+- [ ] **Step 4: Run the repository tests to verify GREEN**
+
+Run:
+
+```bash
+go test ./internal/store/repo -run 'TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch|TestAddUserQuotaUsageBatchReturnsNormalizedViews' -v
+```
+
+Expected: PASS.
+
+- [ ] **Step 5: Optional commit if the user explicitly requested commits**
+
+```bash
+git add go-backend/internal/store/repo/repository.go go-backend/internal/store/repo/repository_flow.go go-backend/internal/store/repo/repository_user_quota.go go-backend/internal/store/repo/repository_flow_batch_test.go
+git commit -m "refactor: batch flow upload persistence"
+```
+
+---
+
+### Task 3: Refactor flow/upload To Use One Parsed Batch
+
+**Files:**
+- Create: `go-backend/internal/http/handler/flow_upload_batch.go`
+- Modify: `go-backend/internal/http/handler/handler.go`
+- Modify: `go-backend/internal/http/handler/tunnel_metrics_ingestion.go`
+- Modify: `go-backend/internal/http/handler/flow_upload_batch_test.go`
+- Modify: `go-backend/tests/contract/flow_upload_batch_contract_test.go`
+
+- [ ] **Step 1: Write the new handler batch implementation**
+
+Create `go-backend/internal/http/handler/flow_upload_batch.go` and move the request-scoped aggregation there.
+
+```go
+package handler
+
+import (
+	"log"
+	"sort"
+	"strings"
+	"time"
+
+	"go-backend/internal/store/repo"
+)
+
+type flowPolicyTarget struct {
+	UserID       int64
+	UserTunnelID int64
+}
+
+type flowUploadBatch struct {
+	flowDeltas           []repo.FlowUploadCounterDelta
+	quotaUsage           map[int64]int64
+	policyTargets        []flowPolicyTarget
+	forwardTraffic       map[int64]tunnelTrafficDelta
+	orphanServices       map[string]struct{}
+	peerShareForwardItems map[string]flowItem
+	peerShareRuntimeItems map[int64]flowItem
+}
+
+func (h *Handler) buildFlowUploadBatch(items []flowItem, metas map[int64]repo.FlowUploadForwardMeta) flowUploadBatch {
+	batch := flowUploadBatch{
+		quotaUsage:            make(map[int64]int64),
+		forwardTraffic:        make(map[int64]tunnelTrafficDelta),
+		orphanServices:        make(map[string]struct{}),
+		peerShareForwardItems: make(map[string]flowItem),
+		peerShareRuntimeItems: make(map[int64]flowItem),
+	}
+	policySeen := map[flowPolicyTarget]struct{}{}
+	flowSeen := map[int64]int{}
+
+	for _, item := range items {
+		serviceName := strings.TrimSpace(item.N)
+		if serviceName == "" || serviceName == "web_api" {
+			continue
+		}
+		if runtimeID, ok := parsePeerShareRuntimeServiceID(serviceName); ok {
+			merged := batch.peerShareRuntimeItems[runtimeID]
+			merged.N = serviceName
+			merged.U += item.U
+			merged.D += item.D
+			batch.peerShareRuntimeItems[runtimeID] = merged
+			continue
+		}
+		forwardID, userID, userTunnelID, ok := parseFlowServiceIDs(serviceName)
+		if !ok {
+			continue
+		}
+		meta, exists := metas[forwardID]
+		if !exists {
+			batch.orphanServices[serviceName] = struct{}{}
+			continue
+		}
+		raw := batch.forwardTraffic[forwardID]
+		raw.bytesIn += item.D
+		raw.bytesOut += item.U
+		batch.forwardTraffic[forwardID] = raw
+
+		scaledIn := int64(float64(item.D)*meta.TrafficRatio) * meta.TunnelFlow
+		scaledOut := int64(float64(item.U)*meta.TrafficRatio) * meta.TunnelFlow
+		if idx, ok := flowSeen[forwardID]; ok {
+			batch.flowDeltas[idx].InFlow += scaledIn
+			batch.flowDeltas[idx].OutFlow += scaledOut
+		} else {
+			flowSeen[forwardID] = len(batch.flowDeltas)
+			batch.flowDeltas = append(batch.flowDeltas, repo.FlowUploadCounterDelta{ForwardID: forwardID, UserID: userID, UserTunnelID: userTunnelID, InFlow: scaledIn, OutFlow: scaledOut})
+		}
+		batch.quotaUsage[userID] += scaledIn + scaledOut
+		target := flowPolicyTarget{UserID: userID, UserTunnelID: userTunnelID}
+		if _, seen := policySeen[target]; !seen {
+			policySeen[target] = struct{}{}
+			batch.policyTargets = append(batch.policyTargets, target)
+		}
+		merged := batch.peerShareForwardItems[normalizeForwardRuntimeServiceName(serviceName)]
+		merged.N = normalizeForwardRuntimeServiceName(serviceName)
+		merged.U += item.U
+		merged.D += item.D
+		batch.peerShareForwardItems[normalizeForwardRuntimeServiceName(serviceName)] = merged
+	}
+
+	sort.Slice(batch.policyTargets, func(i, j int) bool {
+		if batch.policyTargets[i].UserID == batch.policyTargets[j].UserID {
+			return batch.policyTargets[i].UserTunnelID < batch.policyTargets[j].UserTunnelID
+		}
+		return batch.policyTargets[i].UserID < batch.policyTargets[j].UserID
+	})
+	return batch
+}
+
+func (h *Handler) applyFlowUploadBatch(nodeID int64, batch flowUploadBatch, now time.Time) {
+	if h == nil || h.repo == nil {
+		return
+	}
+	if err := h.repo.ApplyFlowUploadDeltasBatch(batch.flowDeltas); err != nil {
+		log.Printf("flow upload write failed op=flow.batch_apply node_id=%d err=%v", nodeID, err)
+		return
+	}
+	quotaViews, err := h.repo.AddUserQuotaUsageBatch(batch.quotaUsage, now)
+	if err != nil {
+		log.Printf("flow upload write failed op=quota.batch_apply node_id=%d err=%v", nodeID, err)
+		return
+	}
+	for userID, quota := range quotaViews {
+		h.enforceUserQuotaIfNeeded(userID, quota)
+	}
+	for _, target := range batch.policyTargets {
+		if target.UserID <= 0 || target.UserTunnelID <= 0 {
+			continue
+		}
+		h.enforceFlowPolicies(target.UserID, target.UserTunnelID)
+	}
+	for serviceName := range batch.orphanServices {
+		h.sendDeleteOrphanedForwardService(nodeID, serviceName)
+	}
+	for serviceName, item := range batch.peerShareForwardItems {
+		forwardID, _, _, ok := parseFlowServiceIDs(serviceName)
+		if ok {
+			h.processPeerShareFlowFromForward(forwardID, nodeID, serviceName, item)
+		}
+	}
+	for runtimeID, item := range batch.peerShareRuntimeItems {
+		h.processPeerShareFlow(runtimeID, item)
+	}
+}
+```
+
+- [ ] **Step 2: Switch the `/flow/upload` entrypoint and tunnel metric ingestion to the shared batch**
+
+Modify `handler.go` and `tunnel_metrics_ingestion.go` so the raw JSON is parsed once and the same forward metadata powers both flow counters and tunnel metrics.
+
+```go
+// handler.go
+func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) {
+	secret := r.URL.Query().Get("secret")
+	node, _ := h.repo.GetNodeBySecret(secret)
+	if node == nil {
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		_, _ = w.Write([]byte("ok"))
+		return
+	}
+
+	raw, err := readAndDecryptFlowBody(r.Body, secret)
+	if err == nil && strings.TrimSpace(raw) != "" {
+		var items []flowItem
+		if json.Unmarshal([]byte(raw), &items) == nil {
+			now := time.Now()
+			forwardIDs := collectFlowUploadForwardIDs(items)
+			metas, metaErr := h.repo.GetFlowUploadForwardMetas(forwardIDs)
+			if metaErr != nil {
+				log.Printf("flow upload metadata lookup failed node_id=%d err=%v", node.ID, metaErr)
+				metas = map[int64]repo.FlowUploadForwardMeta{}
+			}
+			batch := h.buildFlowUploadBatch(items, metas)
+			h.recordTunnelMetricsFromForwardBatch(node.ID, batch.forwardTraffic, metas, now.UnixMilli())
+			h.applyFlowUploadBatch(node.ID, batch, now)
+		}
+	}
+
+	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	_, _ = w.Write([]byte("ok"))
+}
+```
+
+```go
+// tunnel_metrics_ingestion.go
+func collectFlowUploadForwardIDs(items []flowItem) []int64 {
+	ids := make([]int64, 0, len(items))
+	seen := make(map[int64]struct{}, len(items))
+	for _, item := range items {
+		forwardID, _, _, ok := parseFlowServiceIDs(strings.TrimSpace(item.N))
+		if !ok || forwardID <= 0 {
+			continue
+		}
+		if _, exists := seen[forwardID]; exists {
+			continue
+		}
+		seen[forwardID] = struct{}{}
+		ids = append(ids, forwardID)
+	}
+	return ids
+}
+
+func (h *Handler) recordTunnelMetricsFromForwardBatch(nodeID int64, forwardDeltas map[int64]tunnelTrafficDelta, metas map[int64]repo.FlowUploadForwardMeta, nowMs int64) {
+	if h == nil || h.repo == nil || nodeID <= 0 || len(forwardDeltas) == 0 {
+		return
+	}
+	bucketTs := unixMilliBucketMinute(nowMs)
+	if bucketTs <= 0 {
+		return
+	}
+	tunnelAgg := make(map[int64]tunnelTrafficDelta)
+	for forwardID, delta := range forwardDeltas {
+		meta, ok := metas[forwardID]
+		if !ok || meta.TunnelID <= 0 {
+			continue
+		}
+		current := tunnelAgg[meta.TunnelID]
+		current.bytesIn += delta.bytesIn
+		current.bytesOut += delta.bytesOut
+		tunnelAgg[meta.TunnelID] = current
+	}
+	metrics := make([]*model.TunnelMetric, 0, len(tunnelAgg))
+	for tunnelID, delta := range tunnelAgg {
+		if delta.bytesIn == 0 && delta.bytesOut == 0 {
+			continue
+		}
+		metrics = append(metrics, &model.TunnelMetric{TunnelID: tunnelID, NodeID: nodeID, Timestamp: bucketTs, BytesIn: delta.bytesIn, BytesOut: delta.bytesOut})
+	}
+	if len(metrics) == 0 {
+		return
+	}
+	if err := h.repo.UpsertTunnelMetricBuckets(metrics); err != nil {
+		log.Printf("monitoring write failed op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d err=%v", nodeID, bucketTs, len(metrics), err)
+		return
+	}
+	log.Printf("monitoring ok op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d", nodeID, bucketTs, len(metrics))
+}
+```
+
+- [ ] **Step 3: Run focused handler and contract tests to verify GREEN**
+
+Run:
+
+```bash
+go test ./internal/http/handler -run TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets -v
+go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -v
+```
+
+Expected: PASS.
+
+- [ ] **Step 4: Run the full backend suite**
+
+Run:
+
+```bash
+go test ./...
+```
+
+Expected: PASS across the backend module.
+
+- [ ] **Step 5: Optional commit if the user explicitly requested commits**
+
+```bash
+git add go-backend/internal/http/handler/handler.go go-backend/internal/http/handler/tunnel_metrics_ingestion.go go-backend/internal/http/handler/flow_upload_batch.go go-backend/internal/http/handler/flow_upload_batch_test.go go-backend/tests/contract/flow_upload_batch_contract_test.go
+git commit -m "refactor: batch flow upload processing"
+```
+
+---
+
+### Task 4: Final Verification And Performance Sanity Check
+
+**Files:**
+- Modify: `go-backend/tests/contract/flow_upload_batch_contract_test.go`
+
+- [ ] **Step 1: Add a same-batch duplicate-item stress assertion**
+
+Extend the contract test with a second request that repeats the same service name multiple times and assert the counters advance by exactly the summed amount.
+
+```go
+body, err = json.Marshal([]map[string]interface{}{
+	{"n": "20_2_10", "u": 10, "d": 20},
+	{"n": "20_2_10", "u": 10, "d": 20},
+	{"n": "20_2_10_tcp", "u": 10, "d": 20},
+})
+if err != nil {
+	t.Fatalf("marshal body: %v", err)
+}
+req = httptest.NewRequest(http.MethodPost, "/flow/upload?secret="+node.Secret, bytes.NewReader(body))
+res = httptest.NewRecorder()
+router.ServeHTTP(res, req)
+
+if got := mustQueryInt(t, repo, `SELECT in_flow FROM forward WHERE id = 20`); got != 140 {
+	t.Fatalf("expected forward in_flow=140 after second request, got %d", got)
+}
+if got := mustQueryInt(t, repo, `SELECT out_flow FROM forward WHERE id = 20`); got != 140 {
+	t.Fatalf("expected forward out_flow=140 after second request, got %d", got)
+}
+```
+
+- [ ] **Step 2: Run the targeted contract test again**
+
+Run:
+
+```bash
+go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -v
+```
+
+Expected: PASS.
+
+- [ ] **Step 3: Re-run the full backend suite before claiming completion**
+
+Run:
+
+```bash
+go test ./...
+```
+
+Expected: PASS.
+
+- [ ] **Step 4: Optional local profiling sanity check**
+
+Run a short local comparison before and after the change with the same repeated flow payload.
+
+```bash
+go test ./tests/contract/... -run TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately -count=10
+```
+
+Expected: the test remains stable across repeated runs and does not introduce flakiness.
+
+- [ ] **Step 5: Optional commit if the user explicitly requested commits**
+
+```bash
+git add go-backend/tests/contract/flow_upload_batch_contract_test.go
+git commit -m "test: harden flow upload batch regression coverage"
+```

docs/superpowers/specs/2026-04-26-allow-local-remote-addr-design.md

@@ -0,0 +1,133 @@
+# 允许转发到本地地址开关设计
+
+**日期**: 2026-04-26
+**状态**: 待审核
+**作者**: AI Assistant
+
+## 概述
+
+新增一个全局设置开关，控制规则目标地址是否允许指向本地/内网地址。默认关闭，保持当前安全策略不变；开启后，规则创建和编辑时允许将目标地址设置为 `127.0.0.1`、`10.x.x.x`、`172.16-31.x.x`、`192.168.x.x` 等本地或私网地址。
+
+## 背景
+
+当前后端在规则创建和编辑时会调用 `IsSafeRemoteAddr()`，统一禁止目标地址指向本地/内网地址，用来降低 SSRF / 开放代理风险。这一行为是全局硬编码的，无法按部署场景调整。
+
+有些用户需要把规则转发到本机或内网服务，因此需要一个显式、全局的开关来放宽这条限制。
+
+## 目标
+
+1. 在设置页提供一个全局开关控制该行为。
+2. 默认关闭，不改变现有安全默认值。
+3. 开启后，规则创建和编辑允许本地/内网目标地址。
+4. 不影响其他安全校验和其他业务流程。
+
+## 影响范围
+
+### 后端
+- `go-backend/internal/http/handler/security_utils.go`
+- `go-backend/internal/http/handler/mutations.go`
+- `go-backend/internal/http/handler/handler.go`
+
+### 前端
+- `vite-frontend/src/pages/config.tsx`
+
+### 测试
+- `go-backend/tests/contract/forward_contract_test.go` 或新增独立 contract test
+
+## 详细设计
+
+### 1. 配置存储
+
+使用现有 `vite_config` 表新增一个配置项：
+
+| name | value | 说明 |
+|------|-------|------|
+| `allow_local_remote_addr` | `"1"` / `"0"` | 是否允许规则目标地址指向本地/内网地址 |
+
+约定：
+- 未配置时按 `"0"` 处理
+- `"1"` 表示允许
+- 其他值一律按关闭处理
+
+### 2. 后端行为
+
+新增一个轻量辅助函数，用于读取该配置开关：
+
+```go
+func (h *Handler) allowLocalRemoteAddr() bool {
+    if h == nil || h.repo == nil {
+        return false
+    }
+    cfg, err := h.repo.GetConfigByName("allow_local_remote_addr")
+    if err != nil || cfg == nil {
+        return false
+    }
+    return strings.TrimSpace(cfg.Value) == "1"
+}
+```
+
+在以下路径中应用：
+- `forwardCreate`
+- `forwardUpdate`
+
+行为改为：
+- 当开关关闭时，继续执行 `IsSafeRemoteAddr(remoteAddr)`
+- 当开关开启时，跳过这条“本地/内网地址禁止”校验
+
+这样可以把改动范围限定在规则创建/编辑，不改变其他依赖 `IsSafeRemoteAddr()` 的场景。
+
+### 3. 前端设置页
+
+在 `vite-frontend/src/pages/config.tsx` 增加一个全局开关配置项。
+
+建议文案：
+
+- 标签：`允许转发到本地地址`
+- 描述：`开启后，规则目标地址可指向 127.0.0.1、10.x.x.x、172.16-31.x.x、192.168.x.x 等本地或内网地址。默认关闭以降低开放代理风险。`
+
+控件类型：
+- 使用现有设置页的布尔开关模式
+
+默认显示策略：
+- 不依赖其他配置项
+- 直接显示在设置页的网络/安全相关区域；若现有页面没有单独分区，则先按现有配置项组织方式加入即可
+
+### 4. 错误与兼容性
+
+关闭开关时：
+- 保持现有错误行为，继续阻止本地/内网地址
+
+开启开关时：
+- 仅放开“本地/内网地址禁止”这条限制
+- 仍保留地址格式解析失败等其他错误
+
+### 5. 测试
+
+需要补两类后端契约测试：
+
+1. 开关关闭时拒绝本地/内网地址
+- 创建规则时使用本地/内网地址
+- 断言接口返回非 0 code
+
+2. 开关开启时允许本地/内网地址
+- 先写入 `vite_config(name=allow_local_remote_addr, value=1)`
+- 创建或更新规则时使用相同地址
+- 断言接口成功
+
+建议至少覆盖：
+- create 路径
+- update 路径
+- 多目标地址输入（逗号或换行分隔）中包含本地地址时的行为
+
+## 风险与约束
+
+1. 该开关会降低默认安全防护，应明确标注风险。
+2. 这是全局开关，不做用户级或规则级细分控制。
+3. 该开关只影响规则目标地址校验，不影响其他独立的安全策略。
+
+## 推荐实施顺序
+
+1. 先补失败的后端契约测试
+2. 实现后端配置读取与创建/更新分支控制
+3. 在设置页增加开关
+4. 跑后端测试与前端构建验证

go-backend/internal/http/handler/flow_upload_batch.go

@@ -0,0 +1,183 @@
+package handler
+
+import (
+	"log"
+	"sort"
+	"strings"
+	"time"
+
+	"go-backend/internal/store/model"
+	"go-backend/internal/store/repo"
+)
+
+type flowPolicyTarget struct {
+	UserID       int64
+	UserTunnelID int64
+}
+
+type flowUploadBatch struct {
+	flowDeltas            []repo.FlowUploadCounterDelta
+	quotaUsage            map[int64]int64
+	policyTargets         []flowPolicyTarget
+	forwardTraffic        map[int64]tunnelTrafficDelta
+	orphanServices        map[string]struct{}
+	peerShareForwardItems map[string]flowItem
+	peerShareRuntimeItems map[int64]flowItem
+}
+
+func (h *Handler) buildFlowUploadBatch(items []flowItem, metas map[int64]repo.FlowUploadForwardMeta) flowUploadBatch {
+	batch := flowUploadBatch{
+		quotaUsage:            make(map[int64]int64),
+		forwardTraffic:        make(map[int64]tunnelTrafficDelta),
+		orphanServices:        make(map[string]struct{}),
+		peerShareForwardItems: make(map[string]flowItem),
+		peerShareRuntimeItems: make(map[int64]flowItem),
+	}
+	policySeen := map[flowPolicyTarget]struct{}{}
+	flowSeen := map[int64]int{}
+
+	for _, item := range items {
+		serviceName := strings.TrimSpace(item.N)
+		if serviceName == "" || serviceName == "web_api" {
+			continue
+		}
+		if runtimeID, ok := parsePeerShareRuntimeServiceID(serviceName); ok {
+			merged := batch.peerShareRuntimeItems[runtimeID]
+			merged.N = serviceName
+			merged.U += item.U
+			merged.D += item.D
+			batch.peerShareRuntimeItems[runtimeID] = merged
+			continue
+		}
+		forwardID, userID, userTunnelID, ok := parseFlowServiceIDs(serviceName)
+		if !ok {
+			continue
+		}
+		normalized := normalizeForwardRuntimeServiceName(serviceName)
+		merged := batch.peerShareForwardItems[normalized]
+		merged.N = normalized
+		merged.U += item.U
+		merged.D += item.D
+		batch.peerShareForwardItems[normalized] = merged
+
+		meta, exists := metas[forwardID]
+		if !exists {
+			batch.orphanServices[serviceName] = struct{}{}
+			continue
+		}
+
+		raw := batch.forwardTraffic[forwardID]
+		raw.bytesIn += item.D
+		raw.bytesOut += item.U
+		batch.forwardTraffic[forwardID] = raw
+
+		scaledIn := int64(float64(item.D)*meta.TrafficRatio) * meta.TunnelFlow
+		scaledOut := int64(float64(item.U)*meta.TrafficRatio) * meta.TunnelFlow
+		if idx, ok := flowSeen[forwardID]; ok {
+			batch.flowDeltas[idx].InFlow += scaledIn
+			batch.flowDeltas[idx].OutFlow += scaledOut
+		} else {
+			flowSeen[forwardID] = len(batch.flowDeltas)
+			batch.flowDeltas = append(batch.flowDeltas, repo.FlowUploadCounterDelta{
+				ForwardID:    forwardID,
+				UserID:       userID,
+				UserTunnelID: userTunnelID,
+				InFlow:       scaledIn,
+				OutFlow:      scaledOut,
+			})
+		}
+		batch.quotaUsage[userID] += scaledIn + scaledOut
+
+		target := flowPolicyTarget{UserID: userID, UserTunnelID: userTunnelID}
+		if _, seen := policySeen[target]; !seen {
+			policySeen[target] = struct{}{}
+			batch.policyTargets = append(batch.policyTargets, target)
+		}
+
+	}
+
+	sort.Slice(batch.policyTargets, func(i, j int) bool {
+		if batch.policyTargets[i].UserID == batch.policyTargets[j].UserID {
+			return batch.policyTargets[i].UserTunnelID < batch.policyTargets[j].UserTunnelID
+		}
+		return batch.policyTargets[i].UserID < batch.policyTargets[j].UserID
+	})
+
+	return batch
+}
+
+func (h *Handler) applyFlowUploadBatch(nodeID int64, batch flowUploadBatch, now time.Time) {
+	if h == nil || h.repo == nil {
+		return
+	}
+	h.applyFlowDeltasWithFallback(nodeID, batch.flowDeltas)
+	for userID, quota := range h.applyQuotaUsageWithFallback(nodeID, batch.quotaUsage, now) {
+		h.enforceUserQuotaIfNeeded(userID, quota)
+	}
+	for _, target := range batch.policyTargets {
+		if target.UserID <= 0 || target.UserTunnelID <= 0 {
+			continue
+		}
+		h.enforceFlowPolicies(target.UserID, target.UserTunnelID)
+	}
+	for serviceName := range batch.orphanServices {
+		h.sendDeleteOrphanedForwardService(nodeID, serviceName)
+	}
+	for serviceName, item := range batch.peerShareForwardItems {
+		forwardID, _, _, ok := parseFlowServiceIDs(serviceName)
+		if ok {
+			h.processPeerShareFlowFromForward(forwardID, nodeID, serviceName, item)
+		}
+	}
+	for runtimeID, item := range batch.peerShareRuntimeItems {
+		h.processPeerShareFlow(runtimeID, item)
+	}
+}
+
+func (h *Handler) applyFlowDeltasWithFallback(nodeID int64, deltas []repo.FlowUploadCounterDelta) {
+	if h == nil || h.repo == nil || len(deltas) == 0 {
+		return
+	}
+	if err := h.repo.ApplyFlowUploadDeltasBatch(deltas); err == nil {
+		return
+	} else {
+		log.Printf("flow upload write failed op=flow.batch_apply node_id=%d err=%v", nodeID, err)
+	}
+	for _, delta := range deltas {
+		if err := h.repo.AddFlow(delta.ForwardID, delta.UserID, delta.UserTunnelID, delta.InFlow, delta.OutFlow); err != nil {
+			log.Printf("flow upload write failed op=flow.single_apply node_id=%d forward_id=%d user_id=%d user_tunnel_id=%d err=%v", nodeID, delta.ForwardID, delta.UserID, delta.UserTunnelID, err)
+		}
+	}
+}
+
+func (h *Handler) applyQuotaUsageWithFallback(nodeID int64, usages map[int64]int64, now time.Time) map[int64]*model.UserQuotaView {
+	if h == nil || h.repo == nil || len(usages) == 0 {
+		return map[int64]*model.UserQuotaView{}
+	}
+	quotaViews, err := h.repo.AddUserQuotaUsageBatch(usages, now)
+	if err == nil {
+		return quotaViews
+	}
+	log.Printf("flow upload write failed op=quota.batch_apply node_id=%d err=%v", nodeID, err)
+
+	userIDs := make([]int64, 0, len(usages))
+	for userID := range usages {
+		if userID > 0 {
+			userIDs = append(userIDs, userID)
+		}
+	}
+	sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] })
+
+	quotaViews = make(map[int64]*model.UserQuotaView, len(userIDs))
+	for _, userID := range userIDs {
+		quota, singleErr := h.repo.AddUserQuotaUsage(userID, usages[userID], now)
+		if singleErr != nil {
+			log.Printf("flow upload write failed op=quota.single_apply node_id=%d user_id=%d err=%v", nodeID, userID, singleErr)
+			continue
+		}
+		if quota != nil {
+			quotaViews[userID] = quota
+		}
+	}
+	return quotaViews
+}

go-backend/internal/http/handler/flow_upload_batch_test.go

@@ -0,0 +1,254 @@
+package handler
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+
+	"go-backend/internal/store/model"
+	"go-backend/internal/store/repo"
+)
+
+func TestBuildFlowUploadBatchAggregatesForwardQuotaPeerShareAndCleanupTargets(t *testing.T) {
+	h := &Handler{}
+	metas := map[int64]repo.FlowUploadForwardMeta{
+		20: {
+			ForwardID:    20,
+			TunnelID:     1,
+			TrafficRatio: 2,
+			TunnelFlow:   3,
+		},
+	}
+
+	batch := h.buildFlowUploadBatch([]flowItem{
+		{N: "20_2_10", U: 70, D: 50},
+		{N: "20_2_10_tcp", U: 40, D: 30},
+		{N: "99_2_10", U: 12, D: 8},
+		{N: "fed_svc_17", U: 9, D: 1},
+	}, metas)
+
+	if len(batch.flowDeltas) != 1 {
+		t.Fatalf("expected 1 flow delta, got %d", len(batch.flowDeltas))
+	}
+	delta := batch.flowDeltas[0]
+	if delta.ForwardID != 20 || delta.UserID != 2 || delta.UserTunnelID != 10 {
+		t.Fatalf("unexpected flow delta identity: %#v", delta)
+	}
+	if delta.InFlow != 480 || delta.OutFlow != 660 {
+		t.Fatalf("expected scaled flow in=480 out=660, got in=%d out=%d", delta.InFlow, delta.OutFlow)
+	}
+	if batch.quotaUsage[2] != 1140 {
+		t.Fatalf("expected quota usage 1140, got %d", batch.quotaUsage[2])
+	}
+	if len(batch.policyTargets) != 1 {
+		t.Fatalf("expected 1 policy target, got %d", len(batch.policyTargets))
+	}
+	if batch.policyTargets[0].UserID != 2 || batch.policyTargets[0].UserTunnelID != 10 {
+		t.Fatalf("unexpected policy target: %#v", batch.policyTargets[0])
+	}
+	traffic := batch.forwardTraffic[20]
+	if traffic.bytesIn != 80 || traffic.bytesOut != 110 {
+		t.Fatalf("expected raw traffic in=80 out=110, got in=%d out=%d", traffic.bytesIn, traffic.bytesOut)
+	}
+	if _, ok := batch.orphanServices["99_2_10"]; !ok {
+		t.Fatalf("expected orphan service cleanup target for 99_2_10")
+	}
+	if item, ok := batch.peerShareForwardItems["99_2_10"]; !ok || item.U != 12 || item.D != 8 {
+		t.Fatalf("expected orphan forward to remain eligible for peer-share accounting, got %#v ok=%v", item, ok)
+	}
+	if item, ok := batch.peerShareForwardItems["20_2_10"]; !ok || item.U != 110 || item.D != 80 {
+		t.Fatalf("expected merged peer-share forward item, got %#v ok=%v", item, ok)
+	}
+	if item, ok := batch.peerShareRuntimeItems[17]; !ok || item.U != 9 || item.D != 1 {
+		t.Fatalf("expected merged peer-share runtime item, got %#v ok=%v", item, ok)
+	}
+}
+
+func TestApplyFlowUploadBatchContinuesPolicyAndPeerShareSideEffectsWhenQuotaBatchFails(t *testing.T) {
+	r, err := repo.Open(filepath.Join(t.TempDir(), "flow-upload-batch-quota-fail.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	if err := r.DB().Create(&model.User{ID: 2, User: "flow-user", Pwd: "pwd", RoleID: 1, ExpTime: 2727251700000, Flow: 99999, Num: 99999, CreatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed user: %v", err)
+	}
+	if err := r.DB().Create(&model.Tunnel{ID: 1, Name: "tunnel-1", TrafficRatio: 1, Type: 1, Protocol: "tls", Flow: 1, CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed tunnel: %v", err)
+	}
+	if err := r.DB().Create(&model.UserTunnel{ID: 10, UserID: 2, TunnelID: 1, Num: 99999, Flow: 0, ExpTime: 2727251700000, Status: 1}).Error; err != nil {
+		t.Fatalf("seed user tunnel: %v", err)
+	}
+	if err := r.DB().Create(&model.Forward{ID: 20, UserID: 2, UserName: "flow-user", Name: "forward-20", TunnelID: 1, RemoteAddr: "1.1.1.1:80", Strategy: "fifo", CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed forward: %v", err)
+	}
+	if err := r.CreatePeerShare(&repo.PeerShare{Name: "share", NodeID: 1, Token: "token", MaxBandwidth: 0, CurrentFlow: 0, PortRangeStart: 31000, PortRangeEnd: 31010, IsActive: 1, CreatedTime: nowMs, UpdatedTime: nowMs}); err != nil {
+		t.Fatalf("create peer share: %v", err)
+	}
+	share, err := r.GetPeerShareByToken("token")
+	if err != nil || share == nil {
+		t.Fatalf("load peer share: %v", err)
+	}
+	if err := r.DB().Exec(`
+		INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, share.ID, 1, "svc-r1", "svc-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 31001, "", 1, 1, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert peer share runtime: %v", err)
+	}
+	if err := r.DB().Exec(`
+		CREATE TRIGGER fail_user_quota_insert
+		BEFORE INSERT ON user_quota
+		BEGIN
+			SELECT RAISE(FAIL, 'quota insert blocked for test');
+		END;
+	`).Error; err != nil {
+		t.Fatalf("create quota failure trigger: %v", err)
+	}
+
+	h := &Handler{repo: r}
+	h.applyFlowUploadBatch(1, flowUploadBatch{
+		flowDeltas:            []repo.FlowUploadCounterDelta{{ForwardID: 20, UserID: 2, UserTunnelID: 10, InFlow: 80, OutFlow: 120}},
+		quotaUsage:            map[int64]int64{2: 200},
+		policyTargets:         []flowPolicyTarget{{UserID: 2, UserTunnelID: 10}},
+		peerShareForwardItems: map[string]flowItem{"20_2_10": {N: "20_2_10", U: 120, D: 80}},
+	}, now)
+
+	if got := mustQueryInt(t, r, `SELECT status FROM forward WHERE id = 20`); got != 0 {
+		t.Fatalf("expected flow-policy enforcement to pause forward after quota failure, got status=%d", got)
+	}
+	updatedShare, err := r.GetPeerShare(share.ID)
+	if err != nil || updatedShare == nil {
+		t.Fatalf("reload peer share: %v", err)
+	}
+	if updatedShare.CurrentFlow != 200 {
+		t.Fatalf("expected peer-share flow accounting to continue after quota failure, got %d", updatedShare.CurrentFlow)
+	}
+}
+
+func TestApplyFlowUploadBatchContinuesPeerShareSideEffectsWhenFlowBatchFails(t *testing.T) {
+	r, err := repo.Open(filepath.Join(t.TempDir(), "flow-upload-batch-flow-fail.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	if err := r.DB().Create(&model.User{ID: 2, User: "flow-user", Pwd: "pwd", RoleID: 1, ExpTime: 2727251700000, Flow: 99999, Num: 99999, CreatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed user: %v", err)
+	}
+	if err := r.DB().Create(&model.Tunnel{ID: 1, Name: "tunnel-1", TrafficRatio: 1, Type: 1, Protocol: "tls", Flow: 1, CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed tunnel: %v", err)
+	}
+	if err := r.DB().Create(&model.UserTunnel{ID: 10, UserID: 2, TunnelID: 1, Num: 99999, Flow: 0, ExpTime: 2727251700000, Status: 1}).Error; err != nil {
+		t.Fatalf("seed user tunnel: %v", err)
+	}
+	if err := r.DB().Create(&model.Forward{ID: 20, UserID: 2, UserName: "flow-user", Name: "forward-20", TunnelID: 1, RemoteAddr: "1.1.1.1:80", Strategy: "fifo", CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed forward: %v", err)
+	}
+	if err := r.DB().Create(&model.Forward{ID: 21, UserID: 2, UserName: "flow-user", Name: "forward-21", TunnelID: 1, RemoteAddr: "1.1.1.1:81", Strategy: "fifo", CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}).Error; err != nil {
+		t.Fatalf("seed second forward: %v", err)
+	}
+	if err := r.CreatePeerShare(&repo.PeerShare{Name: "share", NodeID: 1, Token: "token", MaxBandwidth: 0, CurrentFlow: 0, PortRangeStart: 31000, PortRangeEnd: 31010, IsActive: 1, CreatedTime: nowMs, UpdatedTime: nowMs}); err != nil {
+		t.Fatalf("create peer share: %v", err)
+	}
+	share, err := r.GetPeerShareByToken("token")
+	if err != nil || share == nil {
+		t.Fatalf("load peer share: %v", err)
+	}
+	if err := r.DB().Exec(`
+		INSERT INTO peer_share_runtime(share_id, node_id, reservation_id, resource_key, binding_id, role, chain_name, service_name, protocol, strategy, port, target, applied, status, created_time, updated_time)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, share.ID, 1, "svc-r1", "svc-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 31001, "", 1, 1, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert peer share runtime: %v", err)
+	}
+	if err := r.DB().Exec(`
+		CREATE TRIGGER fail_forward_flow_update
+		BEFORE UPDATE ON forward
+		WHEN NEW.id = 21 AND (NEW.in_flow != OLD.in_flow OR NEW.out_flow != OLD.out_flow)
+		BEGIN
+			SELECT RAISE(FAIL, 'forward flow update blocked for test');
+		END;
+	`).Error; err != nil {
+		t.Fatalf("create flow failure trigger: %v", err)
+	}
+
+	h := &Handler{repo: r}
+	h.applyFlowUploadBatch(1, flowUploadBatch{
+		flowDeltas: []repo.FlowUploadCounterDelta{
+			{ForwardID: 20, UserID: 2, UserTunnelID: 10, InFlow: 80, OutFlow: 120},
+			{ForwardID: 21, UserID: 2, UserTunnelID: 10, InFlow: 30, OutFlow: 40},
+		},
+		quotaUsage:            map[int64]int64{2: 200},
+		policyTargets:         []flowPolicyTarget{{UserID: 2, UserTunnelID: 10}},
+		peerShareForwardItems: map[string]flowItem{"20_2_10": {N: "20_2_10", U: 120, D: 80}},
+	}, now)
+
+	if got := mustQueryInt(t, r, `SELECT status FROM forward WHERE id = 20`); got != 0 {
+		t.Fatalf("expected flow-policy enforcement to pause forward after flow batch failure, got status=%d", got)
+	}
+	updatedShare, err := r.GetPeerShare(share.ID)
+	if err != nil || updatedShare == nil {
+		t.Fatalf("reload peer share: %v", err)
+	}
+	if updatedShare.CurrentFlow != 200 {
+		t.Fatalf("expected peer-share flow accounting to continue after flow batch failure, got %d", updatedShare.CurrentFlow)
+	}
+	if got := mustQueryInt(t, r, `SELECT in_flow FROM forward WHERE id = 20`); got != 80 {
+		t.Fatalf("expected flow fallback to persist forward 20 in_flow=80, got %d", got)
+	}
+	if got := mustQueryInt(t, r, `SELECT in_flow FROM forward WHERE id = 21`); got != 0 {
+		t.Fatalf("expected failed forward 21 delta to remain unapplied, got %d", got)
+	}
+	if got := mustQueryInt(t, r, `SELECT in_flow FROM user WHERE id = 2`); got != 80 {
+		t.Fatalf("expected flow fallback to preserve successful user totals, got %d", got)
+	}
+	if got := mustQueryInt(t, r, `SELECT in_flow FROM user_tunnel WHERE id = 10`); got != 80 {
+		t.Fatalf("expected flow fallback to preserve successful user_tunnel totals, got %d", got)
+	}
+}
+
+func TestApplyFlowUploadBatchFallsBackToPerUserQuotaUpdates(t *testing.T) {
+	r, err := repo.Open(filepath.Join(t.TempDir(), "flow-upload-batch-quota-fallback.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	dayKey := int64(now.Year()*10000 + int(now.Month())*100 + now.Day())
+	monthKey := int64(now.Year()*100 + int(now.Month()))
+	if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user 2: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(3, 'u3', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user 3: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO user_quota(user_id, daily_limit_gb, monthly_limit_gb, daily_used_bytes, monthly_used_bytes, day_key, month_key, disabled_by_quota, disabled_at, paused_forward_ids, created_time, updated_time) VALUES(2, 0, 0, 0, 0, ?, ?, 0, 0, '', ?, ?), (3, 0, 0, 0, 0, ?, ?, 0, 0, '', ?, ?)`, dayKey, monthKey, nowMs, nowMs, dayKey, monthKey, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user quotas: %v", err)
+	}
+	if err := r.DB().Exec(`
+		CREATE TRIGGER fail_user_3_quota_update
+		BEFORE UPDATE ON user_quota
+		WHEN NEW.user_id = 3 AND (NEW.daily_used_bytes != OLD.daily_used_bytes OR NEW.monthly_used_bytes != OLD.monthly_used_bytes)
+		BEGIN
+			SELECT RAISE(FAIL, 'quota update blocked for user 3');
+		END;
+	`).Error; err != nil {
+		t.Fatalf("create quota fallback trigger: %v", err)
+	}
+
+	h := &Handler{repo: r}
+	h.applyFlowUploadBatch(1, flowUploadBatch{quotaUsage: map[int64]int64{2: 200, 3: 300}}, now)
+
+	if got := mustQueryInt(t, r, `SELECT daily_used_bytes FROM user_quota WHERE user_id = 2`); got != 200 {
+		t.Fatalf("expected quota fallback to persist user 2 usage, got %d", got)
+	}
+	if got := mustQueryInt(t, r, `SELECT daily_used_bytes FROM user_quota WHERE user_id = 3`); got != 0 {
+		t.Fatalf("expected failed user 3 quota delta to remain unapplied, got %d", got)
+	}
+}

go-backend/internal/http/handler/handler.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"net/http"
 	"net/url"
 	"sort"
@@ -43,13 +44,17 @@ type Handler struct {
 	jobsStarted bool
 	jobsWG      sync.WaitGroup
 
-	upgradeMu              sync.Mutex
-	pendingUpgradeRedeploy map[int64]struct{}
+	upgradeMu                sync.Mutex
+	pendingUpgradeRedeploy   map[int64]struct{}
+	nodeOnlineRedeployAt     map[int64]time.Time
+	nodeOnlineRedeployQueued map[int64]struct{}
+	nodeOnlineRedeploying    map[int64]struct{}
 
 	qualityProber *tunnelQualityProber
 }
 
 const monitorTunnelQualityEnabledConfigKey = "monitor_tunnel_quality_enabled"
+const allowLocalRemoteAddrConfigKey = "allow_local_remote_addr"
 
 type loginRequest struct {
 	Username  string `json:"username"`
@@ -95,13 +100,16 @@ const (
 
 func New(repo *repo.Repository, jwtSecret string) *Handler {
 	h := &Handler{
-		repo:                   repo,
-		jwtSecret:              jwtSecret,
-		wsServer:               ws.NewServer(repo, jwtSecret),
-		metrics:                metrics.NewIngestionService(repo),
-		healthCheck:            nil,
-		captchaTokens:          make(map[string]int64),
-		pendingUpgradeRedeploy: make(map[int64]struct{}),
+		repo:                     repo,
+		jwtSecret:                jwtSecret,
+		wsServer:                 ws.NewServer(repo, jwtSecret),
+		metrics:                  metrics.NewIngestionService(repo),
+		healthCheck:              nil,
+		captchaTokens:            make(map[string]int64),
+		pendingUpgradeRedeploy:   make(map[int64]struct{}),
+		nodeOnlineRedeployAt:     make(map[int64]time.Time),
+		nodeOnlineRedeployQueued: make(map[int64]struct{}),
+		nodeOnlineRedeploying:    make(map[int64]struct{}),
 	}
 	h.healthCheck = health.NewChecker(repo, h.wsServer)
 	h.qualityProber = newTunnelQualityProber(h)
@@ -796,11 +804,16 @@ func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) {
 	if err == nil && strings.TrimSpace(raw) != "" {
 		var items []flowItem
 		if json.Unmarshal([]byte(raw), &items) == nil {
-			nowMs := time.Now().UnixMilli()
-			h.recordTunnelMetricsFromFlowItems(node.ID, items, nowMs)
-			for _, item := range items {
-				h.processFlowItem(node.ID, item)
+			now := time.Now()
+			forwardIDs := collectFlowUploadForwardIDs(items)
+			metas, metaErr := h.repo.GetFlowUploadForwardMetas(forwardIDs)
+			if metaErr != nil {
+				log.Printf("flow upload metadata lookup failed node_id=%d err=%v", node.ID, metaErr)
+				metas = map[int64]repo.FlowUploadForwardMeta{}
 			}
+			batch := h.buildFlowUploadBatch(items, metas)
+			h.recordTunnelMetricsFromForwardBatch(node.ID, batch.forwardTraffic, metas, now.UnixMilli())
+			h.applyFlowUploadBatch(node.ID, batch, now)
 		}
 	}
 
@@ -1041,6 +1054,19 @@ func (h *Handler) isTunnelQualityMonitoringEnabled() bool {
 	return strings.TrimSpace(strings.ToLower(cfg.Value)) != "false"
 }
 
+func (h *Handler) allowLocalRemoteAddr() bool {
+	if h == nil || h.repo == nil {
+		return false
+	}
+
+	cfg, err := h.repo.GetConfigByName(allowLocalRemoteAddrConfigKey)
+	if err != nil || cfg == nil {
+		return false
+	}
+
+	return strings.TrimSpace(strings.ToLower(cfg.Value)) == "true"
+}
+
 func (h *Handler) userPackage(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		response.WriteJSON(w, response.ErrDefault("请求失败"))

go-backend/internal/http/handler/mutations.go

@@ -1726,11 +1726,13 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
 		response.WriteJSON(w, response.ErrDefault("转发名称和目标地址不能为空"))
 		return
 	}
-	if roleID != 0 {
+	if roleID != 0 && !h.allowLocalRemoteAddr() {
 		if err := IsSafeRemoteAddr(remoteAddr); err != nil {
 			response.WriteJSON(w, response.Err(403, err.Error()))
 			return
 		}
+	}
+	if roleID != 0 {
 		if speedIDVal, ok := req["speedId"]; ok && speedIDVal != nil {
 			response.WriteJSON(w, response.Err(-1, "普通用户无法设置限速规则"))
 			return
@@ -1853,7 +1855,7 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
 	if remoteAddr == "" {
 		remoteAddr = forward.RemoteAddr
 	}
-	if actorRole != 0 {
+	if actorRole != 0 && !h.allowLocalRemoteAddr() {
 		if err := IsSafeRemoteAddr(remoteAddr); err != nil {
 			response.WriteJSON(w, response.Err(403, err.Error()))
 			return

go-backend/internal/http/handler/tunnel_metrics_ingestion.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"go-backend/internal/store/model"
+	"go-backend/internal/store/repo"
 )
 
 type tunnelTrafficDelta struct {
@@ -21,75 +22,42 @@ func unixMilliBucketMinute(nowMs int64) int64 {
 	return nowMs - (nowMs % minuteMs)
 }
 
-func (h *Handler) recordTunnelMetricsFromFlowItems(nodeID int64, items []flowItem, nowMs int64) {
-	if h == nil || h.repo == nil {
-		return
-	}
-	if nodeID <= 0 || len(items) == 0 {
-		return
+func collectFlowUploadForwardIDs(items []flowItem) []int64 {
+	ids := make([]int64, 0, len(items))
+	seen := make(map[int64]struct{}, len(items))
+	for _, item := range items {
+		forwardID, _, _, ok := parseFlowServiceIDs(strings.TrimSpace(item.N))
+		if !ok || forwardID <= 0 {
+			continue
+		}
+		if _, exists := seen[forwardID]; exists {
+			continue
+		}
+		seen[forwardID] = struct{}{}
+		ids = append(ids, forwardID)
 	}
+	return ids
+}
 
+func (h *Handler) recordTunnelMetricsFromForwardBatch(nodeID int64, forwardDeltas map[int64]tunnelTrafficDelta, metas map[int64]repo.FlowUploadForwardMeta, nowMs int64) {
+	if h == nil || h.repo == nil || nodeID <= 0 || len(forwardDeltas) == 0 {
+		return
+	}
 	bucketTs := unixMilliBucketMinute(nowMs)
 	if bucketTs <= 0 {
 		return
 	}
 
-	forwardDeltas := make(map[int64]tunnelTrafficDelta)
-	var skippedParse, skippedZero int
-	for _, item := range items {
-		name := strings.TrimSpace(item.N)
-		if name == "" || name == "web_api" {
-			continue
-		}
-		forwardID, _, _, ok := parseFlowServiceIDs(name)
-		if !ok {
-			skippedParse++
-			continue
-		}
-		if item.D == 0 && item.U == 0 {
-			skippedZero++
-			continue
-		}
-		d := forwardDeltas[forwardID]
-		d.bytesIn += item.D
-		d.bytesOut += item.U
-		forwardDeltas[forwardID] = d
-	}
-	if len(forwardDeltas) == 0 {
-		if len(items) > 0 {
-			log.Printf("monitoring debug op=tunnel_metric.no_forward_deltas node_id=%d items=%d skipped_parse=%d skipped_zero=%d", nodeID, len(items), skippedParse, skippedZero)
-		}
-		return
-	}
-
-	forwardIDs := make([]int64, 0, len(forwardDeltas))
-	for id := range forwardDeltas {
-		forwardIDs = append(forwardIDs, id)
-	}
-
-	forwardTunnelMap, err := h.repo.MapForwardIDsToTunnelIDs(forwardIDs)
-	if err != nil {
-		log.Printf("monitoring write skipped op=tunnel_metric.map_forward_to_tunnel node_id=%d err=%v", nodeID, err)
-		return
-	}
-	if len(forwardTunnelMap) == 0 {
-		log.Printf("monitoring debug op=tunnel_metric.no_tunnel_map node_id=%d forward_ids=%v", nodeID, forwardIDs)
-		return
-	}
-
 	tunnelAgg := make(map[int64]tunnelTrafficDelta)
 	for forwardID, delta := range forwardDeltas {
-		tunnelID := forwardTunnelMap[forwardID]
-		if tunnelID <= 0 {
+		meta, ok := metas[forwardID]
+		if !ok || meta.TunnelID <= 0 {
 			continue
 		}
-		a := tunnelAgg[tunnelID]
-		a.bytesIn += delta.bytesIn
-		a.bytesOut += delta.bytesOut
-		tunnelAgg[tunnelID] = a
-	}
-	if len(tunnelAgg) == 0 {
-		return
+		current := tunnelAgg[meta.TunnelID]
+		current.bytesIn += delta.bytesIn
+		current.bytesOut += delta.bytesOut
+		tunnelAgg[meta.TunnelID] = current
 	}
 
 	metrics := make([]*model.TunnelMetric, 0, len(tunnelAgg))
@@ -98,14 +66,11 @@ func (h *Handler) recordTunnelMetricsFromFlowItems(nodeID int64, items []flowIte
 			continue
 		}
 		metrics = append(metrics, &model.TunnelMetric{
-			TunnelID:     tunnelID,
-			NodeID:       nodeID,
-			Timestamp:    bucketTs,
-			BytesIn:      delta.bytesIn,
-			BytesOut:     delta.bytesOut,
-			Connections:  0,
-			Errors:       0,
-			AvgLatencyMs: 0,
+			TunnelID:  tunnelID,
+			NodeID:    nodeID,
+			Timestamp: bucketTs,
+			BytesIn:   delta.bytesIn,
+			BytesOut:  delta.bytesOut,
 		})
 	}
 	if len(metrics) == 0 {
@@ -114,7 +79,7 @@ func (h *Handler) recordTunnelMetricsFromFlowItems(nodeID int64, items []flowIte
 
 	if err := h.repo.UpsertTunnelMetricBuckets(metrics); err != nil {
 		log.Printf("monitoring write failed op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d err=%v", nodeID, bucketTs, len(metrics), err)
-	} else {
-		log.Printf("monitoring ok op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d", nodeID, bucketTs, len(metrics))
+		return
 	}
+	log.Printf("monitoring ok op=tunnel_metric.upsert_buckets node_id=%d bucket_ts=%d count=%d", nodeID, bucketTs, len(metrics))
 }

go-backend/internal/http/handler/upgrade.go

@@ -39,6 +39,8 @@ var (
 	testKeywordPattern   = regexp.MustCompile(`(?i)(alpha|beta|rc)`)
 )
 
+const nodeOnlineRedeployCooldown = 30 * time.Second
+
 type githubRelease struct {
 	TagName     string `json:"tag_name"`
 	Name        string `json:"name"`
@@ -396,23 +398,120 @@ func (h *Handler) consumeNodePendingUpgradeRedeploy(nodeID int64) bool {
 }
 
 func (h *Handler) onNodeOnline(nodeID int64) {
-	h.consumeNodePendingUpgradeRedeploy(nodeID)
-	// Always redeploy rules on reconnection, not just for pending upgrade nodes.
-	// This handles cases where the node restarted and lost its in-memory config
-	// before persistence had time to flush, or if the panel also restarted.
-	h.redeployNodeRuntimeAfterUpgrade(nodeID)
+	if !h.startNodeOnlineRedeploy(nodeID, time.Now()) {
+		return
+	}
+	defer h.finishNodeOnlineRedeploy(nodeID)
+
+	// Reconcile node runtime on the first reconnect, but suppress rapid flapping
+	// so websocket churn does not trigger repeated full redeploy storms.
+	if !h.redeployNodeRuntimeAfterUpgrade(nodeID) {
+		h.markNodePendingUpgradeRedeploy(nodeID)
+	}
 }
 
-func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) {
+func (h *Handler) startNodeOnlineRedeploy(nodeID int64, now time.Time) bool {
+	if h == nil || nodeID <= 0 {
+		return false
+	}
+	if now.IsZero() {
+		now = time.Now()
+	}
+
+	h.upgradeMu.Lock()
+	defer h.upgradeMu.Unlock()
+	if h.pendingUpgradeRedeploy == nil {
+		h.pendingUpgradeRedeploy = make(map[int64]struct{})
+	}
+	if h.nodeOnlineRedeployAt == nil {
+		h.nodeOnlineRedeployAt = make(map[int64]time.Time)
+	}
+	if h.nodeOnlineRedeployQueued == nil {
+		h.nodeOnlineRedeployQueued = make(map[int64]struct{})
+	}
+	if h.nodeOnlineRedeploying == nil {
+		h.nodeOnlineRedeploying = make(map[int64]struct{})
+	}
+
+	_, pendingUpgrade := h.pendingUpgradeRedeploy[nodeID]
+	lastRedeployAt := h.nodeOnlineRedeployAt[nodeID]
+	_, inFlight := h.nodeOnlineRedeploying[nodeID]
+	if fireAt, start := nextNodeOnlineRedeployFireAt(lastRedeployAt, now, pendingUpgrade, inFlight); !start {
+		h.queueNodeOnlineRedeployLocked(nodeID, fireAt)
+		return false
+	}
+
+	delete(h.pendingUpgradeRedeploy, nodeID)
+	h.nodeOnlineRedeployAt[nodeID] = now
+	h.nodeOnlineRedeploying[nodeID] = struct{}{}
+	return true
+}
+
+func nextNodeOnlineRedeployFireAt(lastRedeployAt, now time.Time, pendingUpgrade bool, inFlight bool) (time.Time, bool) {
+	if now.IsZero() {
+		now = time.Now()
+	}
+	if inFlight {
+		fireAt := now.Add(nodeOnlineRedeployCooldown)
+		if !lastRedeployAt.IsZero() {
+			cooldownAt := lastRedeployAt.Add(nodeOnlineRedeployCooldown)
+			if cooldownAt.After(now) {
+				fireAt = cooldownAt
+			}
+		}
+		return fireAt, false
+	}
+	if !pendingUpgrade && !lastRedeployAt.IsZero() && now.Sub(lastRedeployAt) < nodeOnlineRedeployCooldown {
+		return lastRedeployAt.Add(nodeOnlineRedeployCooldown), false
+	}
+	return time.Time{}, true
+}
+
+func (h *Handler) queueNodeOnlineRedeployLocked(nodeID int64, fireAt time.Time) {
+	if h == nil || nodeID <= 0 {
+		return
+	}
+	if h.nodeOnlineRedeployQueued == nil {
+		h.nodeOnlineRedeployQueued = make(map[int64]struct{})
+	}
+	if _, queued := h.nodeOnlineRedeployQueued[nodeID]; queued {
+		return
+	}
+	if fireAt.IsZero() {
+		fireAt = time.Now().Add(nodeOnlineRedeployCooldown)
+	}
+	delay := time.Until(fireAt)
+	if delay < 0 {
+		delay = 0
+	}
+	h.nodeOnlineRedeployQueued[nodeID] = struct{}{}
+	time.AfterFunc(delay, func() {
+		h.upgradeMu.Lock()
+		delete(h.nodeOnlineRedeployQueued, nodeID)
+		h.upgradeMu.Unlock()
+		h.onNodeOnline(nodeID)
+	})
+}
+
+func (h *Handler) finishNodeOnlineRedeploy(nodeID int64) {
+	if h == nil || nodeID <= 0 {
+		return
+	}
+	h.upgradeMu.Lock()
+	delete(h.nodeOnlineRedeploying, nodeID)
+	h.upgradeMu.Unlock()
+}
+
+func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) bool {
 	tunnelIDs, err := h.repo.ListActiveTunnelIDsByNode(nodeID)
 	if err != nil {
 		fmt.Printf("post-upgrade redeploy: list tunnels for node %d failed: %v\n", nodeID, err)
-		return
+		return false
 	}
 	forwardIDs, err := h.repo.ListForwardIDsByNode(nodeID)
 	if err != nil {
 		fmt.Printf("post-upgrade redeploy: list forwards for node %d failed: %v\n", nodeID, err)
-		return
+		return false
 	}
 
 	// First pass: deploy everything
@@ -442,7 +541,7 @@ func (h *Handler) redeployNodeRuntimeAfterUpgrade(nodeID int64) {
 	}
 
 	// Retry failed items with exponential backoff (max 3 attempts)
-	h.retryFailedRedeploys(nodeID, tunnelFailed, failedForwards)
+	return h.retryFailedRedeploys(nodeID, tunnelFailed, failedForwards)
 }
 
 // isRetryableError returns true if the error looks transient and worth retrying.
@@ -463,9 +562,9 @@ func isRetryableError(err error) bool {
 }
 
 // retryFailedRedeploys retries failed tunnels and forwards with exponential backoff.
-func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]struct{}, failedForwards []failedForward) {
+func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]struct{}, failedForwards []failedForward) bool {
 	if len(tunnelFailed) == 0 && len(failedForwards) == 0 {
-		return
+		return true
 	}
 
 	const maxRetries = 3
@@ -507,7 +606,7 @@ func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]stru
 
 		if len(tunnelFailed) == 0 && len(failedForwards) == 0 {
 			fmt.Printf("post-upgrade redeploy retry: all items recovered on node %d\n", nodeID)
-			return
+			return true
 		}
 	}
 
@@ -518,4 +617,5 @@ func (h *Handler) retryFailedRedeploys(nodeID int64, tunnelFailed map[int64]stru
 	for _, ff := range failedForwards {
 		fmt.Printf("post-upgrade redeploy: forward %d permanently failed on node %d after retries\n", ff.id, nodeID)
 	}
+	return false
 }

go-backend/internal/http/handler/upgrade_test.go

@@ -0,0 +1,111 @@
+package handler
+
+import (
+	"testing"
+	"time"
+)
+
+func TestStartNodeOnlineRedeploySkipsRecentReconnects(t *testing.T) {
+	h := &Handler{
+		pendingUpgradeRedeploy:   map[int64]struct{}{},
+		nodeOnlineRedeployAt:     map[int64]time.Time{},
+		nodeOnlineRedeployQueued: map[int64]struct{}{},
+		nodeOnlineRedeploying:    map[int64]struct{}{},
+	}
+	now := time.Unix(1_777_176_720, 0)
+
+	if !h.startNodeOnlineRedeploy(54, now) {
+		t.Fatalf("expected first reconnect to redeploy")
+	}
+	h.finishNodeOnlineRedeploy(54)
+
+	if h.startNodeOnlineRedeploy(54, now.Add(5*time.Second)) {
+		t.Fatalf("expected recent reconnect to skip redeploy")
+	}
+	if h.consumeNodePendingUpgradeRedeploy(54) {
+		t.Fatalf("did not expect pending upgrade marker to be consumed")
+	}
+}
+
+func TestStartNodeOnlineRedeployAllowsPendingUpgradeDuringCooldown(t *testing.T) {
+	h := &Handler{
+		pendingUpgradeRedeploy:   map[int64]struct{}{},
+		nodeOnlineRedeployAt:     map[int64]time.Time{},
+		nodeOnlineRedeployQueued: map[int64]struct{}{},
+		nodeOnlineRedeploying:    map[int64]struct{}{},
+	}
+	now := time.Unix(1_777_176_720, 0)
+
+	if !h.startNodeOnlineRedeploy(54, now) {
+		t.Fatalf("expected first reconnect to redeploy")
+	}
+	h.finishNodeOnlineRedeploy(54)
+	h.markNodePendingUpgradeRedeploy(54)
+
+	if !h.startNodeOnlineRedeploy(54, now.Add(5*time.Second)) {
+		t.Fatalf("expected pending upgrade reconnect to bypass cooldown")
+	}
+	if h.consumeNodePendingUpgradeRedeploy(54) {
+		t.Fatalf("expected pending upgrade marker to be consumed during redeploy")
+	}
+}
+
+func TestStartNodeOnlineRedeployQueuesCooldownReconnect(t *testing.T) {
+	h := &Handler{
+		pendingUpgradeRedeploy:   map[int64]struct{}{},
+		nodeOnlineRedeployAt:     map[int64]time.Time{},
+		nodeOnlineRedeployQueued: map[int64]struct{}{},
+		nodeOnlineRedeploying:    map[int64]struct{}{},
+	}
+	now := time.Unix(1_777_176_720, 0)
+
+	if !h.startNodeOnlineRedeploy(54, now) {
+		t.Fatalf("expected first reconnect to redeploy")
+	}
+	h.finishNodeOnlineRedeploy(54)
+
+	if h.startNodeOnlineRedeploy(54, now.Add(5*time.Second)) {
+		t.Fatalf("expected cooldown reconnect to skip immediate redeploy")
+	}
+	if _, queued := h.nodeOnlineRedeployQueued[54]; !queued {
+		t.Fatalf("expected cooldown reconnect to queue a follow-up redeploy")
+	}
+}
+
+func TestStartNodeOnlineRedeployKeepsPendingUpgradeWhileInFlight(t *testing.T) {
+	h := &Handler{
+		pendingUpgradeRedeploy:   map[int64]struct{}{},
+		nodeOnlineRedeployAt:     map[int64]time.Time{},
+		nodeOnlineRedeployQueued: map[int64]struct{}{},
+		nodeOnlineRedeploying:    map[int64]struct{}{},
+	}
+	now := time.Unix(1_777_176_720, 0)
+
+	if !h.startNodeOnlineRedeploy(54, now) {
+		t.Fatalf("expected first reconnect to redeploy")
+	}
+	h.markNodePendingUpgradeRedeploy(54)
+
+	if h.startNodeOnlineRedeploy(54, now.Add(time.Second)) {
+		t.Fatalf("expected in-flight redeploy to suppress parallel restart")
+	}
+	if !h.consumeNodePendingUpgradeRedeploy(54) {
+		t.Fatalf("expected pending upgrade marker to remain for the next retry")
+	}
+	h.finishNodeOnlineRedeploy(54)
+}
+
+func TestNextNodeOnlineRedeployFireAtDefersExpiredInFlightReconnect(t *testing.T) {
+	now := time.Unix(1_777_176_720, 0)
+	last := now.Add(-nodeOnlineRedeployCooldown - 5*time.Second)
+
+	fireAt, start := nextNodeOnlineRedeployFireAt(last, now, false, true)
+	if start {
+		t.Fatalf("expected in-flight reconnect to queue instead of starting immediately")
+	}
+
+	want := now.Add(nodeOnlineRedeployCooldown)
+	if !fireAt.Equal(want) {
+		t.Fatalf("expected queued reconnect at %s, got %s", want, fireAt)
+	}
+}

go-backend/internal/store/repo/repository.go

@@ -22,6 +22,13 @@ import (
 	"go-backend/internal/store/model"
 )
 
+const (
+	defaultPostgresMaxOpenConns = 32
+	defaultPostgresMaxIdleConns = 8
+	defaultPostgresConnMaxIdle  = 5 * time.Minute
+	defaultPostgresConnMaxLife  = 30 * time.Minute
+)
+
 // ─── Type aliases for backward compatibility ─────────────────────────
 // Handlers still reference repo.User, repo.BackupData, etc.
 
@@ -61,6 +68,14 @@ type Repository struct {
 	db *gorm.DB
 }
 
+type FlowUploadCounterDelta struct {
+	ForwardID    int64
+	UserID       int64
+	UserTunnelID int64
+	InFlow       int64
+	OutFlow      int64
+}
+
 func (r *Repository) DB() *gorm.DB {
 	if r == nil {
 		return nil
@@ -68,6 +83,79 @@ func (r *Repository) DB() *gorm.DB {
 	return r.db
 }
 
+func sortedFlowUploadTargetIDs(totals map[int64][2]int64) []int64 {
+	ids := make([]int64, 0, len(totals))
+	for id := range totals {
+		ids = append(ids, id)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	return ids
+}
+
+func (r *Repository) ApplyFlowUploadDeltasBatch(deltas []FlowUploadCounterDelta) error {
+	if r == nil || r.db == nil {
+		return errors.New("repository not initialized")
+	}
+	if len(deltas) == 0 {
+		return nil
+	}
+
+	forwardTotals := make(map[int64][2]int64, len(deltas))
+	userTotals := make(map[int64][2]int64, len(deltas))
+	userTunnelTotals := make(map[int64][2]int64, len(deltas))
+	for _, delta := range deltas {
+		if delta.ForwardID > 0 {
+			current := forwardTotals[delta.ForwardID]
+			current[0] += delta.InFlow
+			current[1] += delta.OutFlow
+			forwardTotals[delta.ForwardID] = current
+		}
+		if delta.UserID > 0 {
+			current := userTotals[delta.UserID]
+			current[0] += delta.InFlow
+			current[1] += delta.OutFlow
+			userTotals[delta.UserID] = current
+		}
+		if delta.UserTunnelID > 0 {
+			current := userTunnelTotals[delta.UserTunnelID]
+			current[0] += delta.InFlow
+			current[1] += delta.OutFlow
+			userTunnelTotals[delta.UserTunnelID] = current
+		}
+	}
+
+	return r.db.Transaction(func(tx *gorm.DB) error {
+		for _, forwardID := range sortedFlowUploadTargetIDs(forwardTotals) {
+			total := forwardTotals[forwardID]
+			if err := tx.Model(&model.Forward{}).Where("id = ?", forwardID).UpdateColumns(map[string]interface{}{
+				"in_flow":  gorm.Expr("in_flow + ?", total[0]),
+				"out_flow": gorm.Expr("out_flow + ?", total[1]),
+			}).Error; err != nil {
+				return err
+			}
+		}
+		for _, userID := range sortedFlowUploadTargetIDs(userTotals) {
+			total := userTotals[userID]
+			if err := tx.Model(&model.User{}).Where("id = ?", userID).UpdateColumns(map[string]interface{}{
+				"in_flow":  gorm.Expr("in_flow + ?", total[0]),
+				"out_flow": gorm.Expr("out_flow + ?", total[1]),
+			}).Error; err != nil {
+				return err
+			}
+		}
+		for _, userTunnelID := range sortedFlowUploadTargetIDs(userTunnelTotals) {
+			total := userTunnelTotals[userTunnelID]
+			if err := tx.Model(&model.UserTunnel{}).Where("id = ?", userTunnelID).UpdateColumns(map[string]interface{}{
+				"in_flow":  gorm.Expr("in_flow + ?", total[0]),
+				"out_flow": gorm.Expr("out_flow + ?", total[1]),
+			}).Error; err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 // ─── Open / Close ────────────────────────────────────────────────────
 
 func Open(path string) (*Repository, error) {
@@ -129,6 +217,7 @@ func OpenPostgres(dsn string) (*Repository, error) {
 	if err != nil {
 		return nil, err
 	}
+	configurePostgresPool(sqlDB)
 	if err := sqlDB.Ping(); err != nil {
 		_ = sqlDB.Close()
 		return nil, err
@@ -154,6 +243,16 @@ func OpenPostgres(dsn string) (*Repository, error) {
 	return &Repository{db: db}, nil
 }
 
+func configurePostgresPool(sqlDB *sql.DB) {
+	if sqlDB == nil {
+		return
+	}
+	sqlDB.SetMaxOpenConns(defaultPostgresMaxOpenConns)
+	sqlDB.SetMaxIdleConns(defaultPostgresMaxIdleConns)
+	sqlDB.SetConnMaxIdleTime(defaultPostgresConnMaxIdle)
+	sqlDB.SetConnMaxLifetime(defaultPostgresConnMaxLife)
+}
+
 func (r *Repository) Close() error {
 	if r == nil || r.db == nil {
 		return nil
@@ -690,11 +789,11 @@ func (r *Repository) ListNodes() ([]map[string]interface{}, error) {
 			"version":       nullableString(n.Version),
 			"http":          n.HTTP, "tls": n.TLS, "socks": n.Socks,
 			"status": n.Status, "isRemote": n.IsRemote,
-			"remoteUrl":                 nullableString(n.RemoteURL),
-			"remoteToken":               nullableString(n.RemoteToken),
-			"remoteConfig":              nullableString(n.RemoteConfig),
-			"expiryReminderDismissed":   n.ExpiryReminderDismissed,
-			"interfaceName":             nullableString(n.InterfaceName),
+			"remoteUrl":               nullableString(n.RemoteURL),
+			"remoteToken":             nullableString(n.RemoteToken),
+			"remoteConfig":            nullableString(n.RemoteConfig),
+			"expiryReminderDismissed": n.ExpiryReminderDismissed,
+			"interfaceName":           nullableString(n.InterfaceName),
 		})
 	}
 	return items, nil
@@ -725,7 +824,7 @@ func (r *Repository) ListUsers() ([]map[string]interface{}, error) {
 			"flowResetTime": u.FlowResetTime, "createdTime": u.CreatedTime,
 			"updatedTime": nullableInt64(u.UpdatedTime),
 			"inFlow":      u.InFlow, "outFlow": u.OutFlow,
-			"maxConn":     u.MaxConn,
+			"maxConn": u.MaxConn,
 		}
 		if quota := quotaMap[u.ID]; quota != nil {
 			item["dailyQuotaGB"] = quota.DailyLimitGB
@@ -766,22 +865,22 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
 	}
 
 	type fwdRow struct {
-		ID           int64
-		UserID       int64
-		UserName     string
-		Name         string
-		TunnelID     int64
-		TunnelName   string
-		TrafficRatio float64
-		RemoteAddr   string
-		Strategy     string
-		InFlow       int64
-		OutFlow      int64
-		CreatedTime  int64
-		Status       int
-		Inx          int
-		SpeedID      sql.NullInt64
-		MaxConn      int
+		ID            int64
+		UserID        int64
+		UserName      string
+		Name          string
+		TunnelID      int64
+		TunnelName    string
+		TrafficRatio  float64
+		RemoteAddr    string
+		Strategy      string
+		InFlow        int64
+		OutFlow       int64
+		CreatedTime   int64
+		Status        int
+		Inx           int
+		SpeedID       sql.NullInt64
+		MaxConn       int
 		ProxyProtocol int
 	}
 
@@ -809,7 +908,7 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) {
 			"remoteAddr": row.RemoteAddr, "strategy": row.Strategy,
 			"inFlow": row.InFlow, "outFlow": row.OutFlow,
 			"createdTime": row.CreatedTime, "status": row.Status, "inx": int64(row.Inx),
-			"maxConn": row.MaxConn,
+			"maxConn":       row.MaxConn,
 			"proxyProtocol": row.ProxyProtocol,
 		}
 		if row.SpeedID.Valid {
@@ -2388,19 +2487,19 @@ func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int
 	count := 0
 	for _, f := range forwards {
 		item := model.Forward{
-			ID:          f.ID,
-			UserID:      f.UserID,
-			UserName:    f.UserName,
-			Name:        f.Name,
-			TunnelID:    f.TunnelID,
-			RemoteAddr:  f.RemoteAddr,
-			Strategy:    f.Strategy,
-			InFlow:      f.InFlow,
-			OutFlow:     f.OutFlow,
-			CreatedTime: f.CreatedTime,
-			UpdatedTime: now,
-			Status:      f.Status,
-			Inx:         f.Inx,
+			ID:            f.ID,
+			UserID:        f.UserID,
+			UserName:      f.UserName,
+			Name:          f.Name,
+			TunnelID:      f.TunnelID,
+			RemoteAddr:    f.RemoteAddr,
+			Strategy:      f.Strategy,
+			InFlow:        f.InFlow,
+			OutFlow:       f.OutFlow,
+			CreatedTime:   f.CreatedTime,
+			UpdatedTime:   now,
+			Status:        f.Status,
+			Inx:           f.Inx,
 			ProxyProtocol: f.ProxyProtocol,
 		}
 		err := tx.Clauses(clause.OnConflict{
@@ -3355,7 +3454,7 @@ func (r *Repository) GetNodeMetrics(nodeID int64, startMs, endMs int64) ([]model
 
 	rangeMs := endMs - startMs
 	const maxRawRangeMs = int64(60 * 60 * 1000) // 1 hour — return raw data for short ranges
-	const targetPoints = 500                     // target number of chart points for downsampled data
+	const targetPoints = 500                    // target number of chart points for downsampled data
 
 	// For short ranges, return raw data (full resolution).
 	if rangeMs <= maxRawRangeMs {

go-backend/internal/store/repo/repository_flow.go

@@ -9,6 +9,90 @@ import (
 	"go-backend/internal/store/model"
 )
 
+type FlowUploadForwardMeta struct {
+	ForwardID    int64
+	TunnelID     int64
+	TrafficRatio float64
+	TunnelFlow   int64
+}
+
+const flowUploadForwardMetaChunkSize = 500
+
+func chunkFlowUploadForwardIDs(ids []int64) [][]int64 {
+	if len(ids) == 0 {
+		return nil
+	}
+	chunks := make([][]int64, 0, (len(ids)+flowUploadForwardMetaChunkSize-1)/flowUploadForwardMetaChunkSize)
+	for start := 0; start < len(ids); start += flowUploadForwardMetaChunkSize {
+		end := start + flowUploadForwardMetaChunkSize
+		if end > len(ids) {
+			end = len(ids)
+		}
+		chunks = append(chunks, ids[start:end])
+	}
+	return chunks
+}
+
+func (r *Repository) GetFlowUploadForwardMetas(forwardIDs []int64) (map[int64]FlowUploadForwardMeta, error) {
+	if r == nil || r.db == nil {
+		return nil, errors.New("repository not initialized")
+	}
+	if len(forwardIDs) == 0 {
+		return map[int64]FlowUploadForwardMeta{}, nil
+	}
+
+	ids := make([]int64, 0, len(forwardIDs))
+	seen := make(map[int64]struct{}, len(forwardIDs))
+	for _, id := range forwardIDs {
+		if id <= 0 {
+			continue
+		}
+		if _, ok := seen[id]; ok {
+			continue
+		}
+		seen[id] = struct{}{}
+		ids = append(ids, id)
+	}
+	if len(ids) == 0 {
+		return map[int64]FlowUploadForwardMeta{}, nil
+	}
+
+	type row struct {
+		ForwardID    int64   `gorm:"column:forward_id"`
+		TunnelID     int64   `gorm:"column:tunnel_id"`
+		TrafficRatio float64 `gorm:"column:traffic_ratio"`
+		TunnelFlow   int64   `gorm:"column:tunnel_flow"`
+	}
+
+	out := make(map[int64]FlowUploadForwardMeta, len(ids))
+	for _, chunk := range chunkFlowUploadForwardIDs(ids) {
+		var rows []row
+		err := r.db.Table("forward AS f").
+			Select("f.id AS forward_id, f.tunnel_id AS tunnel_id, t.traffic_ratio AS traffic_ratio, t.flow AS tunnel_flow").
+			Joins("LEFT JOIN tunnel t ON t.id = f.tunnel_id").
+			Where("f.id IN ?", chunk).
+			Scan(&rows).Error
+		if err != nil {
+			return nil, err
+		}
+		for _, row := range rows {
+			if row.TunnelFlow <= 0 {
+				row.TunnelFlow = 1
+			}
+			if row.TrafficRatio <= 0 {
+				row.TrafficRatio = 1
+			}
+			out[row.ForwardID] = FlowUploadForwardMeta{
+				ForwardID:    row.ForwardID,
+				TunnelID:     row.TunnelID,
+				TrafficRatio: row.TrafficRatio,
+				TunnelFlow:   row.TunnelFlow,
+			}
+		}
+	}
+	return out, nil
+}
+
 func (r *Repository) UpdateForwardStatus(forwardID int64, status int, now int64) error {
 	if r == nil || r.db == nil {
 		return errors.New("repository not initialized")

go-backend/internal/store/repo/repository_flow_batch_test.go

@@ -0,0 +1,142 @@
+package repo
+
+import (
+	"path/filepath"
+	"reflect"
+	"testing"
+	"time"
+)
+
+func TestChunkFlowUploadForwardIDs(t *testing.T) {
+	ids := make([]int64, 0, 1001)
+	for i := int64(1); i <= 1001; i++ {
+		ids = append(ids, i)
+	}
+
+	chunks := chunkFlowUploadForwardIDs(ids)
+	if len(chunks) != 3 {
+		t.Fatalf("expected 3 chunks, got %d", len(chunks))
+	}
+	if len(chunks[0]) != 500 || len(chunks[1]) != 500 || len(chunks[2]) != 1 {
+		t.Fatalf("unexpected chunk sizes: %d, %d, %d", len(chunks[0]), len(chunks[1]), len(chunks[2]))
+	}
+	if chunks[0][0] != 1 || chunks[1][0] != 501 || chunks[2][0] != 1001 {
+		t.Fatalf("unexpected chunk boundaries: %#v %#v %#v", chunks[0][:1], chunks[1][:1], chunks[2][:1])
+	}
+}
+
+func TestSortedFlowUploadTargetIDs(t *testing.T) {
+	totals := map[int64][2]int64{
+		9: {1, 1},
+		2: {1, 1},
+		7: {1, 1},
+	}
+
+	got := sortedFlowUploadTargetIDs(totals)
+	want := []int64{2, 7, 9}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("expected sorted ids %v, got %v", want, got)
+	}
+}
+
+func TestGetFlowUploadForwardMetasAndApplyFlowUploadDeltasBatch(t *testing.T) {
+	r, err := Open(filepath.Join(t.TempDir(), "flow-batch.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now().UnixMilli()
+	if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) VALUES(1, 't1', 2.0, 1, 'tls', 3, ?, ?, 1, NULL, 0)`, now, now).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(10, 2, 1, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)`).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+	if err := r.DB().Exec(`INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx) VALUES(20, 2, 'u2', 'f20', 1, '1.1.1.1:80', 'fifo', 0, 0, ?, ?, 1, 0)`, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+
+	metas, err := r.GetFlowUploadForwardMetas([]int64{20, 99})
+	if err != nil {
+		t.Fatalf("get metas: %v", err)
+	}
+	if metas[20].TunnelID != 1 || metas[20].TrafficRatio != 2 || metas[20].TunnelFlow != 3 {
+		t.Fatalf("unexpected meta for forward 20: %#v", metas[20])
+	}
+	if _, ok := metas[99]; ok {
+		t.Fatalf("did not expect meta for missing forward 99")
+	}
+
+	err = r.ApplyFlowUploadDeltasBatch([]FlowUploadCounterDelta{{ForwardID: 20, UserID: 2, UserTunnelID: 10, InFlow: 480, OutFlow: 660}})
+	if err != nil {
+		t.Fatalf("apply flow batch: %v", err)
+	}
+	if got := mustFlowBatchCount(t, r, `SELECT in_flow FROM forward WHERE id = 20`); got != 480 {
+		t.Fatalf("expected forward in_flow=480, got %d", got)
+	}
+	if got := mustFlowBatchCount(t, r, `SELECT out_flow FROM user WHERE id = 2`); got != 660 {
+		t.Fatalf("expected user out_flow=660, got %d", got)
+	}
+	if got := mustFlowBatchCount(t, r, `SELECT in_flow FROM user_tunnel WHERE id = 10`); got != 480 {
+		t.Fatalf("expected user_tunnel in_flow=480, got %d", got)
+	}
+}
+
+func TestGetFlowUploadForwardMetasKeepsForwardsWhenTunnelRowMissing(t *testing.T) {
+	r, err := Open(filepath.Join(t.TempDir(), "flow-batch-missing-tunnel.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now().UnixMilli()
+	if err := r.DB().Exec(`INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx) VALUES(25, 2, 'u2', 'f25', 99, '1.1.1.1:80', 'fifo', 0, 0, ?, ?, 1, 0)`, now, now).Error; err != nil {
+		t.Fatalf("insert forward: %v", err)
+	}
+
+	metas, err := r.GetFlowUploadForwardMetas([]int64{25})
+	if err != nil {
+		t.Fatalf("get metas: %v", err)
+	}
+	meta, ok := metas[25]
+	if !ok {
+		t.Fatalf("expected metadata for forward with missing tunnel row")
+	}
+	if meta.ForwardID != 25 || meta.TunnelID != 99 || meta.TrafficRatio != 1 || meta.TunnelFlow != 1 {
+		t.Fatalf("unexpected fallback meta: %#v", meta)
+	}
+}
+
+func TestAddUserQuotaUsageBatchReturnsNormalizedViews(t *testing.T) {
+	r, err := Open(filepath.Join(t.TempDir(), "quota-batch.db"))
+	if err != nil {
+		t.Fatalf("open repo: %v", err)
+	}
+	defer r.Close()
+
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	if err := r.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'u2', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	views, err := r.AddUserQuotaUsageBatch(map[int64]int64{2: 1140}, now)
+	if err != nil {
+		t.Fatalf("batch quota update: %v", err)
+	}
+	if views[2] == nil || views[2].DailyUsedBytes != 1140 || views[2].MonthlyUsedBytes != 1140 {
+		t.Fatalf("unexpected quota view: %#v", views[2])
+	}
+}
+
+func mustFlowBatchCount(t *testing.T, r *Repository, query string, args ...interface{}) int64 {
+	t.Helper()
+	var value int64
+	if err := r.DB().Raw(query, args...).Row().Scan(&value); err != nil {
+		t.Fatalf("query %q failed: %v", query, err)
+	}
+	return value
+}

go-backend/internal/store/repo/repository_pool_test.go

@@ -0,0 +1,32 @@
+package repo
+
+import (
+	"testing"
+
+	gsqlite "github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+)
+
+func TestConfigurePostgresPoolSetsMaxOpenConnections(t *testing.T) {
+	db, err := gorm.Open(gsqlite.Open(":memory:"), &gorm.Config{
+		Logger: logger.Default.LogMode(logger.Silent),
+	})
+	if err != nil {
+		t.Fatalf("open sqlite: %v", err)
+	}
+
+	sqlDB, err := db.DB()
+	if err != nil {
+		t.Fatalf("db handle: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = sqlDB.Close()
+	})
+
+	configurePostgresPool(sqlDB)
+
+	if got := sqlDB.Stats().MaxOpenConnections; got != defaultPostgresMaxOpenConns {
+		t.Fatalf("expected max open conns %d, got %d", defaultPostgresMaxOpenConns, got)
+	}
+}

go-backend/internal/store/repo/repository_user_quota.go

@@ -3,6 +3,7 @@ package repo
 import (
 	"errors"
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -255,6 +256,54 @@ func (r *Repository) AddUserQuotaUsage(userID int64, usedBytes int64, now time.T
 	return normalizeUserQuotaView(result, now), nil
 }
 
+func (r *Repository) AddUserQuotaUsageBatch(usages map[int64]int64, now time.Time) (map[int64]*model.UserQuotaView, error) {
+	if r == nil || r.db == nil {
+		return nil, errors.New("repository not initialized")
+	}
+	if len(usages) == 0 {
+		return map[int64]*model.UserQuotaView{}, nil
+	}
+
+	result := make(map[int64]*model.UserQuotaView, len(usages))
+	err := r.db.Transaction(func(tx *gorm.DB) error {
+		userIDs := make([]int64, 0, len(usages))
+		for userID := range usages {
+			if userID > 0 {
+				userIDs = append(userIDs, userID)
+			}
+		}
+		sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] })
+
+		for _, userID := range userIDs {
+			q, err := r.loadOrCreateUserQuotaTx(tx, userID, now)
+			if err != nil {
+				return err
+			}
+			applyUserQuotaWindowRoll(q, now)
+			if usages[userID] > 0 {
+				q.DailyUsedBytes += usages[userID]
+				q.MonthlyUsedBytes += usages[userID]
+			}
+			q.UpdatedTime = now.UnixMilli()
+			if err := tx.Model(&model.UserQuota{}).Where("user_id = ?", userID).Updates(map[string]interface{}{
+				"daily_used_bytes":   q.DailyUsedBytes,
+				"monthly_used_bytes": q.MonthlyUsedBytes,
+				"day_key":            q.DayKey,
+				"month_key":          q.MonthKey,
+				"updated_time":       q.UpdatedTime,
+			}).Error; err != nil {
+				return err
+			}
+			result[userID] = normalizeUserQuotaView(cloneUserQuotaView(*q), now)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+
 func (r *Repository) MarkUserQuotaDisabled(userID int64, pausedForwardIDs []int64, now int64) error {
 	if r == nil || r.db == nil {
 		return errors.New("repository not initialized")

go-backend/tests/contract/flow_upload_batch_contract_test.go

@@ -0,0 +1,107 @@
+package contract_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"go-backend/internal/store/model"
+)
+
+func TestFlowUploadAggregatesRepeatedItemsAndDisablesQuotaImmediately(t *testing.T) {
+	secret := "monitoring-jwt-secret"
+	router, repo := setupContractRouter(t, secret)
+	now := time.Now()
+	nowMs := now.UnixMilli()
+	dayKey := int64(now.Year()*10000 + int(now.Month())*100 + now.Day())
+	monthKey := int64(now.Year()*100 + int(now.Month()))
+	const bytesPerGB = int64(1024 * 1024 * 1024)
+
+	node := &model.Node{Name: "node-1", Secret: "node-secret", ServerIP: "127.0.0.1", Port: "10000-10010", TCPListenAddr: "[::]", UDPListenAddr: "[::]", CreatedTime: nowMs, Status: 1}
+	if err := repo.DB().Create(node).Error; err != nil {
+		t.Fatalf("seed node: %v", err)
+	}
+	if err := repo.DB().Exec(`INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) VALUES(2, 'flow_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)`, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+	tunnel := &model.Tunnel{Name: "tunnel-1", TrafficRatio: 1.0, Type: 1, Protocol: "tls", Flow: 1, CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}
+	if err := repo.DB().Create(tunnel).Error; err != nil {
+		t.Fatalf("seed tunnel: %v", err)
+	}
+	if err := repo.DB().Exec(`INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(10, 2, ?, NULL, 99999, 99999, 0, 0, 1, 2727251700000, 1)`, tunnel.ID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+	forward := &model.Forward{ID: 20, UserID: 2, UserName: "flow_user", Name: "forward-20", TunnelID: tunnel.ID, RemoteAddr: "1.1.1.1:80", Strategy: "fifo", CreatedTime: nowMs, UpdatedTime: nowMs, Status: 1}
+	if err := repo.DB().Create(forward).Error; err != nil {
+		t.Fatalf("seed forward: %v", err)
+	}
+	if err := repo.DB().Exec(`INSERT INTO user_quota(user_id, daily_limit_gb, monthly_limit_gb, daily_used_bytes, monthly_used_bytes, day_key, month_key, disabled_by_quota, disabled_at, paused_forward_ids, created_time, updated_time) VALUES(2, 1, 0, ?, ?, ?, ?, 0, 0, '', ?, ?)`, bytesPerGB-100, bytesPerGB-100, dayKey, monthKey, nowMs, nowMs).Error; err != nil {
+		t.Fatalf("insert user_quota: %v", err)
+	}
+
+	body, err := json.Marshal([]map[string]interface{}{
+		{"n": "20_2_10", "u": 70, "d": 50},
+		{"n": "20_2_10_tcp", "u": 40, "d": 30},
+	})
+	if err != nil {
+		t.Fatalf("marshal body: %v", err)
+	}
+	req := httptest.NewRequest(http.MethodPost, "/flow/upload?secret="+node.Secret, bytes.NewReader(body))
+	res := httptest.NewRecorder()
+	router.ServeHTTP(res, req)
+
+	if res.Code != http.StatusOK {
+		t.Fatalf("expected status 200, got %d", res.Code)
+	}
+	if got := mustQueryInt(t, repo, `SELECT status FROM forward WHERE id = 20`); got != 0 {
+		t.Fatalf("expected forward paused immediately, got status=%d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT disabled_by_quota FROM user_quota WHERE user_id = 2`); got != 1 {
+		t.Fatalf("expected quota disabled flag=1, got %d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT in_flow FROM forward WHERE id = 20`); got != 80 {
+		t.Fatalf("expected forward in_flow=80, got %d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT out_flow FROM forward WHERE id = 20`); got != 110 {
+		t.Fatalf("expected forward out_flow=110, got %d", got)
+	}
+	metrics, err := repo.GetTunnelMetrics(tunnel.ID, 0, nowMs+60_000)
+	if err != nil {
+		t.Fatalf("get tunnel metrics: %v", err)
+	}
+	if len(metrics) != 1 || metrics[0].BytesIn != 80 || metrics[0].BytesOut != 110 {
+		t.Fatalf("expected one aggregated metric row, got %#v", metrics)
+	}
+
+	body, err = json.Marshal([]map[string]interface{}{
+		{"n": "20_2_10", "u": 10, "d": 20},
+		{"n": "20_2_10", "u": 10, "d": 20},
+		{"n": "20_2_10_tcp", "u": 10, "d": 20},
+	})
+	if err != nil {
+		t.Fatalf("marshal body: %v", err)
+	}
+	req = httptest.NewRequest(http.MethodPost, "/flow/upload?secret="+node.Secret, bytes.NewReader(body))
+	res = httptest.NewRecorder()
+	router.ServeHTTP(res, req)
+
+	if res.Code != http.StatusOK {
+		t.Fatalf("expected second request status 200, got %d", res.Code)
+	}
+	if got := mustQueryInt(t, repo, `SELECT in_flow FROM forward WHERE id = 20`); got != 140 {
+		t.Fatalf("expected forward in_flow=140 after second request, got %d", got)
+	}
+	if got := mustQueryInt(t, repo, `SELECT out_flow FROM forward WHERE id = 20`); got != 140 {
+		t.Fatalf("expected forward out_flow=140 after second request, got %d", got)
+	}
+	metrics, err = repo.GetTunnelMetrics(tunnel.ID, 0, nowMs+60_000)
+	if err != nil {
+		t.Fatalf("get tunnel metrics after second request: %v", err)
+	}
+	if len(metrics) != 1 || metrics[0].BytesIn != 140 || metrics[0].BytesOut != 140 {
+		t.Fatalf("expected one aggregated metric row after second request, got %#v", metrics)
+	}
+}

go-backend/tests/contract/forward_local_remote_addr_contract_test.go

@@ -0,0 +1,219 @@
+package contract_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"go-backend/internal/auth"
+	"go-backend/internal/http/handler"
+	"go-backend/internal/http/response"
+)
+
+func TestForwardLocalRemoteAddrToggleContracts(t *testing.T) {
+	handler.DisableSafeRemoteAddrCheckForTesting = false
+	t.Cleanup(func() {
+		handler.DisableSafeRemoteAddrCheckForTesting = true
+	})
+
+	secret := "contract-jwt-secret"
+	router, repo := setupContractRouter(t, secret)
+	server := httptest.NewServer(router)
+	defer server.Close()
+
+	now := time.Now().UnixMilli()
+	if err := repo.DB().Exec(`
+		INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
+		VALUES(2, 'local_remote_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
+	`, now, now).Error; err != nil {
+		t.Fatalf("insert user: %v", err)
+	}
+
+	if err := repo.DB().Exec(`
+		INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, "local-remote-tunnel", 1.0, 1, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	tunnelID := mustLastInsertID(t, repo, "local-remote-tunnel")
+
+	if err := repo.DB().Exec(`
+		INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, "local-remote-entry", "local-remote-secret", "10.60.0.1", "10.60.0.1", "", "31000-31010", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0).Error; err != nil {
+		t.Fatalf("insert node: %v", err)
+	}
+	entryNodeID := mustLastInsertID(t, repo, "local-remote-entry")
+
+	if err := repo.DB().Exec(`
+		INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
+		VALUES(?, 1, ?, 31001, 'round', 1, 'tls')
+	`, tunnelID, entryNodeID).Error; err != nil {
+		t.Fatalf("insert chain_tunnel: %v", err)
+	}
+
+	if err := repo.DB().Exec(`
+		INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+		VALUES(601, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
+	`, tunnelID).Error; err != nil {
+		t.Fatalf("insert user_tunnel: %v", err)
+	}
+
+	userToken, err := auth.GenerateToken(2, "local_remote_user", 1, secret)
+	if err != nil {
+		t.Fatalf("generate user token: %v", err)
+	}
+
+	stopNode := startMockNodeSession(t, server.URL, "local-remote-secret")
+	defer stopNode()
+	waitNodeStatus(t, repo, entryNodeID, 1)
+
+	t.Run("local remote address is rejected on create when toggle is off", func(t *testing.T) {
+		createPayload := map[string]interface{}{
+			"name":       "deny-local-create",
+			"tunnelId":   tunnelID,
+			"remoteAddr": "127.0.0.1:8080",
+			"strategy":   "fifo",
+		}
+		createBody, err := json.Marshal(createPayload)
+		if err != nil {
+			t.Fatalf("marshal create payload: %v", err)
+		}
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(createBody))
+		req.Header.Set("Authorization", userToken)
+		req.Header.Set("Content-Type", "application/json")
+		res := httptest.NewRecorder()
+		router.ServeHTTP(res, req)
+
+		var out response.R
+		if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
+			t.Fatalf("decode response: %v", err)
+		}
+		if out.Code == 0 {
+			t.Fatalf("expected local remote address to be rejected when toggle is off")
+		}
+		if !strings.Contains(out.Msg, "internal IP") && !strings.Contains(out.Msg, "内部") {
+			t.Fatalf("expected internal IP error, got code=%d msg=%q", out.Code, out.Msg)
+		}
+	})
+
+	t.Run("local remote address is allowed on create when toggle is on", func(t *testing.T) {
+		if err := repo.DB().Exec(`
+			INSERT INTO vite_config(name, value, time)
+			VALUES(?, ?, ?)
+			ON CONFLICT(name) DO UPDATE SET value = excluded.value, time = excluded.time
+		`, "allow_local_remote_addr", "true", time.Now().UnixMilli()).Error; err != nil {
+			t.Fatalf("enable allow_local_remote_addr: %v", err)
+		}
+
+		createPayload := map[string]interface{}{
+			"name":       "allow-local-create",
+			"tunnelId":   tunnelID,
+			"remoteAddr": "127.0.0.1:8080",
+			"strategy":   "fifo",
+		}
+		createBody, err := json.Marshal(createPayload)
+		if err != nil {
+			t.Fatalf("marshal create payload: %v", err)
+		}
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(createBody))
+		req.Header.Set("Authorization", userToken)
+		req.Header.Set("Content-Type", "application/json")
+		res := httptest.NewRecorder()
+		router.ServeHTTP(res, req)
+		assertCode(t, res, 0)
+	})
+
+	t.Run("local remote address is rejected on update when toggle is off", func(t *testing.T) {
+		if err := repo.DB().Exec(`
+			INSERT INTO vite_config(name, value, time)
+			VALUES(?, ?, ?)
+			ON CONFLICT(name) DO UPDATE SET value = excluded.value, time = excluded.time
+		`, "allow_local_remote_addr", "false", time.Now().UnixMilli()).Error; err != nil {
+			t.Fatalf("disable allow_local_remote_addr: %v", err)
+		}
+
+		createPayload := map[string]interface{}{
+			"name":       "safe-remote-before-update",
+			"tunnelId":   tunnelID,
+			"remoteAddr": "8.8.8.8:53",
+			"strategy":   "fifo",
+		}
+		createBody, err := json.Marshal(createPayload)
+		if err != nil {
+			t.Fatalf("marshal safe create payload: %v", err)
+		}
+		createReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(createBody))
+		createReq.Header.Set("Authorization", userToken)
+		createReq.Header.Set("Content-Type", "application/json")
+		createRes := httptest.NewRecorder()
+		router.ServeHTTP(createRes, createReq)
+		assertCode(t, createRes, 0)
+
+		forwardID := mustLastInsertID(t, repo, "safe-remote-before-update")
+		updatePayload := map[string]interface{}{
+			"id":         forwardID,
+			"name":       "safe-remote-before-update",
+			"tunnelId":   tunnelID,
+			"remoteAddr": "127.0.0.1:8081",
+			"strategy":   "fifo",
+		}
+		updateBody, err := json.Marshal(updatePayload)
+		if err != nil {
+			t.Fatalf("marshal update payload: %v", err)
+		}
+		updateReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/update", bytes.NewReader(updateBody))
+		updateReq.Header.Set("Authorization", userToken)
+		updateReq.Header.Set("Content-Type", "application/json")
+		updateRes := httptest.NewRecorder()
+		router.ServeHTTP(updateRes, updateReq)
+
+		var out response.R
+		if err := json.NewDecoder(updateRes.Body).Decode(&out); err != nil {
+			t.Fatalf("decode update response: %v", err)
+		}
+		if out.Code == 0 {
+			t.Fatalf("expected local remote address to be rejected on update when toggle is off")
+		}
+		if !strings.Contains(out.Msg, "internal IP") && !strings.Contains(out.Msg, "内部") {
+			t.Fatalf("expected internal IP error on update, got code=%d msg=%q", out.Code, out.Msg)
+		}
+	})
+
+	t.Run("local remote address is allowed on update when toggle is on", func(t *testing.T) {
+		if err := repo.DB().Exec(`
+			INSERT INTO vite_config(name, value, time)
+			VALUES(?, ?, ?)
+			ON CONFLICT(name) DO UPDATE SET value = excluded.value, time = excluded.time
+		`, "allow_local_remote_addr", "true", time.Now().UnixMilli()).Error; err != nil {
+			t.Fatalf("enable allow_local_remote_addr: %v", err)
+		}
+
+		var forwardID int64
+		if err := repo.DB().Raw(`SELECT id FROM forward WHERE name = ? ORDER BY id DESC LIMIT 1`, "safe-remote-before-update").Row().Scan(&forwardID); err != nil {
+			t.Fatalf("query forward id: %v", err)
+		}
+
+		updatePayload := map[string]interface{}{
+			"id":         forwardID,
+			"name":       "safe-remote-before-update",
+			"tunnelId":   tunnelID,
+			"remoteAddr": "127.0.0.1:8081",
+			"strategy":   "fifo",
+		}
+		updateBody, err := json.Marshal(updatePayload)
+		if err != nil {
+			t.Fatalf("marshal update payload: %v", err)
+		}
+		updateReq := httptest.NewRequest(http.MethodPost, "/api/v1/forward/update", bytes.NewReader(updateBody))
+		updateReq.Header.Set("Authorization", userToken)
+		updateReq.Header.Set("Content-Type", "application/json")
+		updateRes := httptest.NewRecorder()
+		router.ServeHTTP(updateRes, updateReq)
+		assertCode(t, updateRes, 0)
+	})
+}

go-gost/x/socket/websocket_reporter.go

@@ -145,11 +145,12 @@ type ServiceMonitorCheckResult struct {
 }
 
 const (
-	reporterReadWait  = 60 * time.Second
-	reporterWriteWait = 5 * time.Second
-	wsPingInterval    = 20 * time.Second // 独立 WebSocket ping 间隔
-	initialBackoff    = 2 * time.Second  // 重连初始退避
-	maxBackoff        = 2 * time.Minute  // 重连最大退避
+	reporterReadWait            = 60 * time.Second
+	reporterWriteWait           = 5 * time.Second
+	wsPingInterval              = 20 * time.Second // 独立 WebSocket ping 间隔
+	initialBackoff              = 2 * time.Second  // 重连初始退避
+	maxBackoff                  = 2 * time.Minute  // 重连最大退避
+	defaultMetricReportInterval = 5 * time.Second
 )
 
 type WebSocketReporter struct {
@@ -189,9 +190,9 @@ func NewWebSocketReporter(serverURL string, secret string) *WebSocketReporter {
 
 	return &WebSocketReporter{
 		url:            serverURL,
-		curBackoff:     initialBackoff,   // 当前退避间隔
-		pingInterval:   1 * time.Second,  // 指标上报间隔（每秒采集）
-		configInterval: 10 * time.Minute, // 配置上报间隔
+		curBackoff:     initialBackoff,              // 当前退避间隔
+		pingInterval:   defaultMetricReportInterval, // 指标上报间隔
+		configInterval: 10 * time.Minute,            // 配置上报间隔
 		ctx:            ctx,
 		cancel:         cancel,
 		connected:      false,

go-gost/x/socket/websocket_reporter_test.go

@@ -110,6 +110,14 @@ func TestSanitizeWebSocketURL(t *testing.T) {
 	}
 }
 
+func TestNewWebSocketReporterUsesReducedMetricInterval(t *testing.T) {
+	reporter := NewWebSocketReporter("panel.example.com:443", "abc")
+
+	if reporter.pingInterval != defaultMetricReportInterval {
+		t.Fatalf("expected metric interval %s, got %s", defaultMetricReportInterval, reporter.pingInterval)
+	}
+}
+
 func TestFormatWebSocketDialErrorIncludesHTTPStatus(t *testing.T) {
 	err := errors.New("websocket: bad handshake")
 	resp := &http.Response{

vite-frontend/src/pages/config.tsx

@@ -185,6 +185,13 @@ const CONFIG_ITEMS: ConfigItem[] = [
     dependsOn: "github_proxy_enabled",
     dependsValue: "true",
   },
+  {
+    key: "allow_local_remote_addr",
+    label: "允许转发到本地地址",
+    description:
+      "开启后，普通用户创建或编辑规则时可将目标地址指向 127.0.0.1、10.x.x.x、172.16-31.x.x、192.168.x.x 等本地或内网地址。默认关闭以降低开放代理风险。",
+    type: "switch",
+  },
 ];
 
 const BACKUP_TYPE_OPTIONS = [
@@ -219,6 +226,7 @@ const getInitialConfigs = (): Record<string, string> => {
     "app_favicon",
     "github_proxy_enabled",
     "github_proxy_url",
+    "allow_local_remote_addr",
   ];
   const initialConfigs: Record<string, string> = {};
 

vite-frontend/src/pages/dashboard/use-dashboard-data.ts

@@ -360,11 +360,7 @@ export const useDashboardData = (): DashboardDataState => {
           if (updateTime > storedTime) {
             setIsAnnouncementModalOpen(true);
           }
-        } catch (err) {
-          console.warn(
-            "Failed to read localStorage for announcement state",
-            err,
-          );
+        } catch {
           setIsAnnouncementModalOpen(true);
         }
       } else {
@@ -385,8 +381,8 @@ export const useDashboardData = (): DashboardDataState => {
           "flvx_announcement_seen_time",
           announcement.update_time.toString(),
         );
-      } catch (err) {
-        console.warn("Failed to set localStorage for announcement state", err);
+      } catch {
+        // Ignore localStorage write failures and keep the modal dismissed.
       }
     }
   }, [announcement]);

vite-frontend/src/pages/forward.tsx

@@ -2600,7 +2600,7 @@ export default function ForwardPage() {
         try {
           document.execCommand("copy");
           toast.success(`已复制${label}`);
-        } catch (err) {
+        } catch {
           toast.error("复制失败");
         }
         document.body.removeChild(textArea);
@@ -4884,6 +4884,7 @@ export default function ForwardPage() {
                     <AccordionItem
                       key="advanced"
                       aria-label="高级设置"
+                      className="border-b-0 [&_[data-slot=accordion-trigger]]:no-underline [&_[data-slot=accordion-trigger]]:hover:no-underline"
                       title={
                         <span className="text-small text-default-500 font-medium">
                           高级设置

vite-frontend/src/themes/loader.ts

@@ -10,7 +10,6 @@
  */
 
 import { registerTheme } from "./registry";
-
 // ── Built-in themes ──────────────────────────────────────────────────────────
 import defaultTheme from "./default";
 import cyberpunkTheme from "./example-cyberpunk";

vite-frontend/src/themes/registry.ts

@@ -117,8 +117,6 @@ export function activateTheme(id: string): void {
   const pkg = installed.get(id);
 
   if (!pkg) {
-    console.warn(`[FLVX themes] Theme "${id}" is not registered.`);
-
     return;
   }