feat: tunnel state reporting to control plane

.codegraph/.gitignore

@@ -0,0 +1,16 @@
+# CodeGraph data files
+# These are local to each machine and should not be committed
+
+# Database
+*.db
+*.db-wal
+*.db-shm
+
+# Cache
+cache/
+
+# Logs
+*.log
+
+# Hook markers
+.dirty

lib/servercfg/config_control_plane_compat_test.go

@@ -0,0 +1,147 @@
+package servercfg
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestControlPlaneFlatConfigParsing verifies that the flat INI format
+// produced by nps_enhanced/renderPushModeConfig is correctly parsed.
+func TestControlPlaneFlatConfigParsing(t *testing.T) {
+	resetTestState(t)
+
+	// This is the exact format produced by nps_enhanced renderPushModeConfig
+	// (flat keys in the default section, no [common] block).
+	path := writeConfig(t, "nps.conf", strings.Join([]string{
+		"platform_ids=nps-enhanced-42",
+		"platform_tokens=test-platform-token-abc",
+		"platform_scopes=full",
+		"platform_connect_modes=reverse",
+		"platform_reverse_ws_urls=ws://ctrl.example.com:18081/node/ws",
+		"platform_reverse_enabled=true",
+		"platform_reverse_heartbeat_seconds=30",
+		"platform_callback_urls=https://ctrl.example.com:18081/node/callback",
+		"platform_callback_enabled=true",
+		"platform_callback_signing_keys=test-signing-key-xyz",
+		"# Legacy native keys (ignored by data-plane)",
+		"run_mode=node",
+		"platform_url=https://ctrl.example.com:18081",
+		"platform_token=test-platform-token-abc",
+		"callback_signing_key=test-signing-key-xyz",
+		"traffic_report=true",
+	}, "\n")+"\n")
+
+	if err := Load(path); err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+
+	cfg := Current()
+	if cfg == nil {
+		t.Fatal("Current() returned nil")
+	}
+
+	platforms := cfg.Runtime.ManagementPlatforms
+	if len(platforms) != 1 {
+		t.Fatalf("expected 1 management platform, got %d", len(platforms))
+	}
+
+	p := platforms[0]
+	if p.PlatformID != "nps-enhanced-42" {
+		t.Errorf("PlatformID = %q, want nps-enhanced-42", p.PlatformID)
+	}
+	if p.Token != "test-platform-token-abc" {
+		t.Errorf("Token = %q, want test-platform-token-abc", p.Token)
+	}
+	if p.ControlScope != "full" {
+		t.Errorf("ControlScope = %q, want full", p.ControlScope)
+	}
+	if p.ConnectMode != "reverse" {
+		t.Errorf("ConnectMode = %q, want reverse", p.ConnectMode)
+	}
+	if p.ReverseWSURL != "ws://ctrl.example.com:18081/node/ws" {
+		t.Errorf("ReverseWSURL = %q, want ws://ctrl.example.com:18081/node/ws", p.ReverseWSURL)
+	}
+	if !p.ReverseEnabled {
+		t.Error("ReverseEnabled = false, want true")
+	}
+	if p.ReverseHeartbeatSeconds != 30 {
+		t.Errorf("ReverseHeartbeatSeconds = %d, want 30", p.ReverseHeartbeatSeconds)
+	}
+	if p.CallbackURL != "https://ctrl.example.com:18081/node/callback" {
+		t.Errorf("CallbackURL = %q, want https://ctrl.example.com:18081/node/callback", p.CallbackURL)
+	}
+	if !p.CallbackEnabled {
+		t.Error("CallbackEnabled = false, want true")
+	}
+	if p.CallbackSigningKey != "test-signing-key-xyz" {
+		t.Errorf("CallbackSigningKey = %q, want test-signing-key-xyz", p.CallbackSigningKey)
+	}
+}
+
+// TestControlPlaneMergedConfigParsing verifies that a merged config
+// (push block inserted at top, before [common]) is correctly parsed.
+func TestControlPlaneMergedConfigParsing(t *testing.T) {
+	resetTestState(t)
+
+	path := writeConfig(t, "nps.conf", strings.Join([]string{
+		"# --- nps-enhanced push mode config START ---",
+		"platform_ids=nps-enhanced-1",
+		"platform_tokens=tok-1",
+		"platform_scopes=full",
+		"platform_connect_modes=reverse",
+		"platform_reverse_ws_urls=ws://ctrl.example.com:18081/node/ws",
+		"platform_reverse_enabled=true",
+		"platform_reverse_heartbeat_seconds=30",
+		"platform_callback_urls=https://ctrl.example.com:18081/node/callback",
+		"platform_callback_enabled=true",
+		"platform_callback_signing_keys=key-1",
+		"# --- nps-enhanced push mode config END ---",
+		"",
+		"[common]",
+		"web_port=8080",
+		"run_mode=server",
+	}, "\n")+"\n")
+
+	if err := Load(path); err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+
+	cfg := Current()
+	platforms := cfg.Runtime.ManagementPlatforms
+	if len(platforms) != 1 {
+		t.Fatalf("expected 1 platform, got %d", len(platforms))
+	}
+	if platforms[0].PlatformID != "nps-enhanced-1" {
+		t.Errorf("PlatformID = %q, want nps-enhanced-1", platforms[0].PlatformID)
+	}
+	if platforms[0].ConnectMode != "reverse" {
+		t.Errorf("ConnectMode = %q, want reverse", platforms[0].ConnectMode)
+	}
+}
+
+// TestCommonSectionPlatformKeysIgnored verifies that platform keys inside
+// [common] are NOT visible to the data-plane parser (they become
+// common_platform_ids after flattening). This confirms why mergePushConfig
+// must place push keys in the default section (before any [section]).
+func TestCommonSectionPlatformKeysIgnored(t *testing.T) {
+	resetTestState(t)
+
+	path := writeConfig(t, "nps.conf", strings.Join([]string{
+		"[common]",
+		"platform_ids=old-master",
+		"platform_tokens=old-tok",
+		"platform_connect_modes=reverse",
+		"platform_reverse_ws_urls=ws://old.example/ws",
+		"platform_reverse_enabled=true",
+	}, "\n")+"\n")
+
+	if err := Load(path); err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+
+	cfg := Current()
+	platforms := cfg.Runtime.ManagementPlatforms
+	if len(platforms) != 0 {
+		t.Fatalf("expected 0 platforms (keys inside [common] are invisible), got %d", len(platforms))
+	}
+}

server/client_host_list.go

@@ -562,6 +562,7 @@ func snapshotTunnelForList(tunnel *file.Tunnel) *file.Tunnel {
 		ServiceTraffic: cloneTrafficStatsForList(tunnel.ServiceTraffic),
 		ServiceMeter:   tunnel.ServiceMeter.Clone(),
 		NowConn:        tunnel.NowConn,
+		MaxConn:        tunnel.MaxConn,
 		Password:       tunnel.Password,
 		Remark:         tunnel.Remark,
 		TargetAddr:     tunnel.TargetAddr,

server/engine_startup.go

@@ -184,7 +184,9 @@ func wrapBackgroundLoopStart(start func()) func() {
 	}
 	var once sync.Once
 	return func() {
-		once.Do(start)
+		once.Do(func() {
+			go start()
+		})
 	}
 }
 

server/server.go

@@ -26,6 +26,11 @@ var (
 	Bridge         *bridge.Bridge
 	RunList        sync.Map //map[int]interface{}
 	HttpProxyCache = index.NewAnyIntIndex()
+
+	// ManagementEventHook is called when tunnel/client/host resources are mutated.
+	// It is set by web/routers/state.go during node initialization.
+	// The hook receives: eventName, resource, action, fields.
+	ManagementEventHook func(eventName, resource, action string, fields map[string]interface{})
 )
 
 func ClearProxyCache() {

web/api/tunnels.go

@@ -187,7 +187,8 @@ func (a *App) sanitizeNodeTunnel(actor *Actor, scope webservice.NodeAccessScope,
 }
 
 func (a *App) finishNodeTunnelMutation(c Context, action, eventName string, id int, tunnel *file.Tunnel, overrides map[string]interface{}) {
-	a.emitNodeResourceMutationEvent(c, eventName, "tunnel", action, nodeResourceMutationFields(id, tunnelEventFields(tunnel), overrides))
+	// Event emission is now handled exclusively by the service layer (web/service/resources.go).
+	// a.emitNodeResourceMutationEvent(c, eventName, "tunnel", action, nodeResourceMutationFields(id, tunnelEventFields(tunnel), overrides))
 	a.respondCompletedNodeMutation(c, "tunnel", action, id, a.nodeTunnelResource(tunnel))
 }
 

web/routers/node_ws.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/djylb/nps/lib/logs"
 	"github.com/djylb/nps/lib/servercfg"
+	"github.com/djylb/nps/server"
 	webapi "github.com/djylb/nps/web/api"
 	webservice "github.com/djylb/nps/web/service"
 	"github.com/gin-gonic/gin"
@@ -2018,6 +2019,25 @@ func (m *nodeReverseManager) serveConn(platform servercfg.ManagementPlatformConf
 		)
 	}(platform.ReverseHeartbeatSeconds)
 
+	// Periodic full tunnel state reporting (60s)
+	backgroundWG.Add(1)
+	go func() {
+		defer backgroundWG.Done()
+		ticker := time.NewTicker(60 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+				report := buildTunnelFullReportFrame(m.state, platform.PlatformID)
+				if report != nil {
+					_ = writeFrame(*report)
+				}
+			}
+		}
+	}()
+
 	reverseHost := reverseWSHost(platform.ReverseWSURL)
 	base := nodeWSDispatchBase{
 		Context:    m.ctx,
@@ -2246,3 +2266,73 @@ func (m *nodeReverseManager) runtimeStatus() webservice.ManagementPlatformRuntim
 	}
 	return webservice.NewInMemoryManagementPlatformRuntimeStatusStore()
 }
+
+
+// buildTunnelFullReportFrame collects all tunnels and builds a WS event frame.
+func buildTunnelFullReportFrame(state *State, platformID string) *nodeWSFrame {
+	if state == nil {
+		return nil
+	}
+	allTunnels, _ := server.GetTunnel(0, 99999, "", 0, "", "", "")
+	if len(allTunnels) == 0 {
+		return nil
+	}
+	tunnelSnapshots := make([]map[string]interface{}, 0, len(allTunnels))
+	for _, t := range allTunnels {
+		targetAddr := ""
+		if t.Target != nil {
+			targetAddr = t.Target.TargetStr
+		}
+		snap := map[string]interface{}{
+			"id":          t.Id,
+			"port":        t.Port,
+			"mode":        t.Mode,
+			"status":      t.Status,
+			"run_status":  t.RunStatus,
+			"remark":      t.Remark,
+			"target":      targetAddr,
+			"target_type": t.TargetType,
+			"now_conn":    t.NowConn,
+			"max_conn":    t.MaxConn,
+			"expire_at":   t.ExpireAt,
+			"rate_limit":  t.RateLimit,
+		}
+		if t.Client != nil {
+			snap["client_id"] = t.Client.Id
+		}
+		if t.Flow != nil {
+			snap["flow_in"] = t.Flow.InletFlow
+			snap["flow_out"] = t.Flow.ExportFlow
+			snap["flow_limit"] = t.Flow.FlowLimit
+		}
+		if t.ServiceTraffic != nil {
+			snap["service_traffic_in"] = t.ServiceTraffic.InletBytes
+			snap["service_traffic_out"] = t.ServiceTraffic.ExportBytes
+		}
+		tunnelSnapshots = append(tunnelSnapshots, snap)
+	}
+
+	event := webapi.Event{
+		Name:     "tunnel.full_report",
+		Resource: "tunnel",
+		Action:   "report",
+		Fields: map[string]interface{}{
+			"platform_id": platformID,
+			"tunnels":     tunnelSnapshots,
+			"reported_at": time.Now().Unix(),
+			"count":       len(tunnelSnapshots),
+		},
+	}
+	body, err := json.Marshal(event)
+	if err != nil {
+		logs.Warn("[rev-ws] failed to marshal tunnel full report: %v", err)
+		return nil
+	}
+	frame := &nodeWSFrame{
+		Type:      "event",
+		ID:        "tunnel-report-" + strconv.FormatInt(time.Now().UnixNano(), 10),
+		Timestamp: time.Now().Unix(),
+		Body:      body,
+	}
+	return frame
+}

web/routers/state.go

@@ -16,6 +16,7 @@ import (
 	"github.com/djylb/nps/lib/common"
 	"github.com/djylb/nps/lib/logs"
 	"github.com/djylb/nps/lib/servercfg"
+	"github.com/djylb/nps/server"
 	webapi "github.com/djylb/nps/web/api"
 	webservice "github.com/djylb/nps/web/service"
 )
@@ -606,6 +607,17 @@ func NewStateWithApp(app *webapi.App) *State {
 		hooks.webhooks = state.NodeWebhooks
 		app.Hooks = hooks
 	}
+	// Register the global management event hook so web/service can emit events.
+	// Must be done after state is fully initialized to avoid nil pointer dereferences
+	// if an event is emitted concurrently during startup.
+	server.ManagementEventHook = func(eventName, resource, action string, fields map[string]interface{}) {
+		emitNodeManagementEvent(state, state.BaseContext(), webapi.Event{
+			Name:     eventName,
+			Resource: resource,
+			Action:   action,
+			Fields:   fields,
+		})
+	}
 	return state
 }