diff --git a/go-backend/internal/http/handler/federation.go b/go-backend/internal/http/handler/federation.go index 2c34c68..97c7228 100644 --- a/go-backend/internal/http/handler/federation.go +++ b/go-backend/internal/http/handler/federation.go @@ -1268,6 +1268,8 @@ func (h *Handler) federationRuntimeCommand(w http.ResponseWriter, r *http.Reques } if strings.EqualFold(cmd, "addservice") || strings.EqualFold(cmd, "updateservice") { h.bindPeerShareForwardRuntimeServices(share, req.Data) + } else if strings.EqualFold(cmd, "deleteservice") { + h.releasePeerShareForwardRuntimeServices(share, req.Data) } response.WriteJSON(w, response.OK(res)) } @@ -1326,6 +1328,45 @@ func parseFederationForwardServiceBindings(data interface{}) []federationForward return bindings } +func parseFederationForwardServiceNamesForRelease(data interface{}) []string { + names := make(map[string]struct{}) + appendName := func(raw string) { + name := normalizeForwardRuntimeServiceName(raw) + if name == "" { + return + } + if _, _, _, ok := parseFlowServiceIDs(name); !ok { + return + } + names[name] = struct{}{} + } + + for _, svcMap := range extractFederationServiceEntries(data) { + appendName(asString(svcMap["name"])) + } + + if dataMap, ok := data.(map[string]interface{}); ok { + for _, item := range asAnySlice(dataMap["services"]) { + appendName(asString(item)) + } + } + + for _, item := range asAnySlice(data) { + appendName(asString(item)) + } + + if len(names) == 0 { + return nil + } + + out := make([]string, 0, len(names)) + for name := range names { + out = append(out, name) + } + sort.Strings(out) + return out +} + func (h *Handler) bindPeerShareForwardRuntimeServices(share *repo.PeerShare, data interface{}) { if h == nil || h.repo == nil || share == nil { return @@ -1338,13 +1379,66 @@ func (h *Handler) bindPeerShareForwardRuntimeServices(share *repo.PeerShare, dat now := time.Now().UnixMilli() for _, binding := range bindings { runtime, err := h.repo.GetActiveForwardPeerShareRuntimeByPort(share.ID, binding.Port) - if err != nil || runtime == nil || runtime.Status != 1 { + if err != nil { continue } - if runtime.ServiceName == binding.Name && runtime.Applied == 1 { + if runtime == nil { + runtime, err = h.repo.GetActiveForwardPeerShareRuntimeByServiceName(share.ID, binding.Name) + if err != nil { + continue + } + } + if runtime == nil { + _ = h.repo.CreatePeerShareRuntime(&repo.PeerShareRuntime{ + ShareID: share.ID, + NodeID: share.NodeID, + ReservationID: randomToken(24), + ResourceKey: fmt.Sprintf("forward-runtime:%d:%s:%d:%s", share.ID, binding.Name, binding.Port, randomToken(8)), + BindingID: "", + Role: "forward", + ChainName: "", + ServiceName: binding.Name, + Protocol: "tcp", + Strategy: "fifo", + Port: binding.Port, + Target: "", + Applied: 1, + Status: 1, + CreatedTime: now, + UpdatedTime: now, + }) continue } - _ = h.repo.UpdatePeerShareRuntimeServiceName(runtime.ID, binding.Name, now) + if runtime.ServiceName == binding.Name && runtime.Applied == 1 && runtime.Port == binding.Port && runtime.Status == 1 { + continue + } + runtime.ServiceName = binding.Name + runtime.Port = binding.Port + runtime.Applied = 1 + runtime.Status = 1 + runtime.UpdatedTime = now + if strings.TrimSpace(runtime.Protocol) == "" { + runtime.Protocol = "tcp" + } + if strings.TrimSpace(runtime.Strategy) == "" { + runtime.Strategy = "fifo" + } + _ = h.repo.UpdatePeerShareRuntime(runtime) + } +} + +func (h *Handler) releasePeerShareForwardRuntimeServices(share *repo.PeerShare, data interface{}) { + if h == nil || h.repo == nil || share == nil { + return + } + names := parseFederationForwardServiceNamesForRelease(data) + if len(names) == 0 { + return + } + + now := time.Now().UnixMilli() + for _, name := range names { + _ = h.repo.MarkForwardPeerShareRuntimeReleasedByServiceName(share.ID, name, now) } } diff --git a/go-backend/internal/http/handler/federation_share_test.go b/go-backend/internal/http/handler/federation_share_test.go index 2071155..dfbccc0 100644 --- a/go-backend/internal/http/handler/federation_share_test.go +++ b/go-backend/internal/http/handler/federation_share_test.go @@ -722,6 +722,123 @@ func TestBindPeerShareForwardRuntimeServicesAcceptsTopLevelServiceArray(t *testi } } +func TestBindPeerShareForwardRuntimeServicesCreatesRuntimeWhenMissing(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel-bind-create-runtime.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = r.Close() }) + + h := New(r, "test-jwt-secret") + now := time.Now().UnixMilli() + + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "bind-create-runtime-share", + NodeID: 1, + Token: "bind-create-runtime-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 26300, + PortRangeEnd: 26320, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share: %v", err) + } + share, err := r.GetPeerShareByToken("bind-create-runtime-token") + if err != nil || share == nil { + t.Fatalf("load share: %v", err) + } + + h.bindPeerShareForwardRuntimeServices(share, map[string]interface{}{ + "services": []interface{}{ + map[string]interface{}{"name": "55_2_10_tcp", "addr": "[::]:26301"}, + }, + }) + + var count int64 + if err := r.DB().Raw(`SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND role = ? AND status = 1`, share.ID, "forward").Scan(&count).Error; err != nil { + t.Fatalf("query runtime count: %v", err) + } + if count != 1 { + t.Fatalf("expected 1 active forward runtime row, got %d", count) + } + + var serviceName string + var port int + var applied int + if err := r.DB().Raw(`SELECT service_name, port, applied FROM peer_share_runtime WHERE share_id = ? AND role = ? ORDER BY id DESC LIMIT 1`, share.ID, "forward").Row().Scan(&serviceName, &port, &applied); err != nil { + t.Fatalf("query created runtime: %v", err) + } + if serviceName != "55_2_10" { + t.Fatalf("expected service_name=55_2_10, got %q", serviceName) + } + if port != 26301 { + t.Fatalf("expected port=26301, got %d", port) + } + if applied != 1 { + t.Fatalf("expected applied=1, got %d", applied) + } +} + +func TestReleasePeerShareForwardRuntimeServicesMarksRuntimeReleased(t *testing.T) { + r, err := repo.Open(filepath.Join(t.TempDir(), "panel-release-runtime.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = r.Close() }) + + h := New(r, "test-jwt-secret") + now := time.Now().UnixMilli() + + if err := r.CreatePeerShare(&repo.PeerShare{ + Name: "release-runtime-share", + NodeID: 1, + Token: "release-runtime-token", + MaxBandwidth: 0, + CurrentFlow: 0, + PortRangeStart: 26400, + PortRangeEnd: 26420, + IsActive: 1, + CreatedTime: now, + UpdatedTime: now, + }); err != nil { + t.Fatalf("create share: %v", err) + } + share, err := r.GetPeerShareByToken("release-runtime-token") + if err != nil || share == nil { + t.Fatalf("load 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, share.NodeID, "release-r1", "release-rk1", "", "forward", "", "77_2_10", "tcp", "fifo", 26401, "", 1, 1, now, now).Error; err != nil { + t.Fatalf("insert runtime: %v", err) + } + + h.releasePeerShareForwardRuntimeServices(share, map[string]interface{}{ + "services": []interface{}{"77_2_10_tcp"}, + }) + + var status int + var applied int + var serviceName string + if err := r.DB().Raw(`SELECT status, applied, service_name FROM peer_share_runtime WHERE share_id = ? AND role = ? ORDER BY id DESC LIMIT 1`, share.ID, "forward").Row().Scan(&status, &applied, &serviceName); err != nil { + t.Fatalf("query released runtime: %v", err) + } + if status != 0 { + t.Fatalf("expected status=0 after release, got %d", status) + } + if applied != 0 { + t.Fatalf("expected applied=0 after release, got %d", applied) + } + if serviceName != "" { + t.Fatalf("expected service_name cleared after release, got %q", serviceName) + } +} + func TestValidateFederationCommandPortsAcceptsTopLevelServiceArray(t *testing.T) { share := &repo.PeerShare{ PortRangeStart: 26200, diff --git a/go-backend/internal/http/handler/flow_policy.go b/go-backend/internal/http/handler/flow_policy.go index 460ee57..0850cf5 100644 --- a/go-backend/internal/http/handler/flow_policy.go +++ b/go-backend/internal/http/handler/flow_policy.go @@ -2,9 +2,12 @@ package handler import ( "encoding/json" + "log" "strconv" "strings" "time" + + "go-backend/internal/store/model" ) const bytesPerGB int64 = 1024 * 1024 * 1024 @@ -30,7 +33,7 @@ type namedConfigItem struct { Name string `json:"name"` } -func (h *Handler) processFlowItem(item flowItem) { +func (h *Handler) processFlowItem(nodeID int64, item flowItem) { serviceName := strings.TrimSpace(item.N) if serviceName == "" || serviceName == "web_api" { return @@ -40,7 +43,7 @@ func (h *Handler) processFlowItem(item flowItem) { if ok { inFlow, outFlow := h.scaleFlowByTunnel(forwardID, item.D, item.U) _ = h.repo.AddFlow(forwardID, userID, userTunnelID, inFlow, outFlow) - h.processPeerShareFlowFromForward(forwardID, serviceName, item) + h.processPeerShareFlowFromForward(forwardID, nodeID, serviceName, item) if userTunnelID > 0 { h.enforceFlowPolicies(userID, userTunnelID) @@ -153,7 +156,7 @@ func (h *Handler) processPeerShareFlow(runtimeID int64, item flowItem) { h.enforcePeerShareFlowLimit(share.ID) } -func (h *Handler) processPeerShareFlowFromForward(forwardID int64, serviceName string, item flowItem) { +func (h *Handler) processPeerShareFlowFromForward(forwardID int64, nodeID int64, serviceName string, item flowItem) { if h == nil || h.repo == nil || forwardID <= 0 { return } @@ -167,22 +170,22 @@ func (h *Handler) processPeerShareFlowFromForward(forwardID int64, serviceName s if err != nil || forward == nil { // Forward not found in local database - might be a federation port-forward // Try to find by service name in peer_share_runtime - h.processPeerShareFlowByServiceName(serviceName, item) + h.processPeerShareFlowByServiceName(nodeID, serviceName, item) return } tunnelName, err := h.repo.GetTunnelName(forward.TunnelID) if err != nil { - h.processPeerShareFlowByServiceName(serviceName, item) + h.processPeerShareFlowByServiceName(nodeID, serviceName, item) return } shareID, ok := parsePeerShareIDFromFederationTunnelName(tunnelName) if !ok { - h.processPeerShareFlowByServiceName(serviceName, item) + h.processPeerShareFlowByServiceName(nodeID, serviceName, item) return } if err := h.repo.AddPeerShareCurrentFlow(shareID, delta); err != nil { - h.processPeerShareFlowByServiceName(serviceName, item) + h.processPeerShareFlowByServiceName(nodeID, serviceName, item) return } @@ -207,7 +210,7 @@ func normalizeForwardRuntimeServiceName(serviceName string) string { return name } -func (h *Handler) processPeerShareFlowByServiceName(serviceName string, item flowItem) { +func (h *Handler) processPeerShareFlowByServiceName(nodeID int64, serviceName string, item flowItem) { if h == nil || h.repo == nil || strings.TrimSpace(serviceName) == "" { return } @@ -218,17 +221,41 @@ func (h *Handler) processPeerShareFlowByServiceName(serviceName string, item flo } normalized := normalizeForwardRuntimeServiceName(serviceName) - runtimes, err := h.repo.ListActiveForwardPeerShareRuntimesByServiceName(normalized) - if err != nil { - return - } - if len(runtimes) == 0 && normalized != serviceName { - runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(serviceName) + var runtimes []model.PeerShareRuntime + var err error + + // Try node-scoped query first if nodeID is valid + if nodeID > 0 { + runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByNodeAndServiceName(nodeID, normalized) if err != nil { return } + if len(runtimes) == 0 && normalized != serviceName { + runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByNodeAndServiceName(nodeID, serviceName) + if err != nil { + return + } + } } + + // Fallback to global query if node-scoped query returned nothing or nodeID is invalid + if len(runtimes) == 0 { + runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(normalized) + if err != nil { + return + } + if len(runtimes) == 0 && normalized != serviceName { + runtimes, err = h.repo.ListActiveForwardPeerShareRuntimesByServiceName(serviceName) + if err != nil { + return + } + } + } + if len(runtimes) != 1 { + if len(runtimes) > 1 { + log.Printf("WARN: ambiguous peer share runtime match for service=%s nodeID=%d count=%d", serviceName, nodeID, len(runtimes)) + } return } runtime := runtimes[0] diff --git a/go-backend/internal/http/handler/flow_policy_federation_test.go b/go-backend/internal/http/handler/flow_policy_federation_test.go index ce80980..c2a3bfc 100644 --- a/go-backend/internal/http/handler/flow_policy_federation_test.go +++ b/go-backend/internal/http/handler/flow_policy_federation_test.go @@ -44,7 +44,7 @@ func TestProcessFlowItemTracksPeerShareFlowAndEnforcesLimit(t *testing.T) { } h := &Handler{repo: r} - h.processFlowItem(flowItem{N: "fed_svc_17", U: 1200, D: 900}) + h.processFlowItem(1, flowItem{N: "fed_svc_17", U: 1200, D: 900}) updatedShare, err := r.GetPeerShare(share.ID) if err != nil || updatedShare == nil { @@ -120,7 +120,7 @@ func TestProcessFlowItemTracksPeerShareFlowForFederationPortForward(t *testing.T } h := &Handler{repo: r} - h.processFlowItem(flowItem{N: "20_2_10", U: 120, D: 80}) + h.processFlowItem(1, flowItem{N: "20_2_10", U: 120, D: 80}) updatedShare, err := r.GetPeerShare(share.ID) if err != nil || updatedShare == nil { @@ -166,7 +166,7 @@ func TestProcessFlowItemTracksPeerShareFlowByForwardServiceName(t *testing.T) { } h := &Handler{repo: r} - h.processFlowItem(flowItem{N: "20_2_10_tcp", U: 120, D: 80}) + h.processFlowItem(1, flowItem{N: "20_2_10_tcp", U: 120, D: 80}) updatedShare, err := r.GetPeerShare(share.ID) if err != nil || updatedShare == nil { @@ -226,7 +226,7 @@ func TestProcessFlowItemFallsBackToServiceNameWhenForwardIDCollidesAcrossPanels( } h := &Handler{repo: r} - h.processFlowItem(flowItem{N: "20_2_10_tcp", U: 120, D: 80}) + h.processFlowItem(1, flowItem{N: "20_2_10_tcp", U: 120, D: 80}) updatedShare, err := r.GetPeerShare(share.ID) if err != nil || updatedShare == nil { @@ -288,7 +288,7 @@ func TestProcessFlowItemSkipsPeerShareFlowWhenServiceNameIsAmbiguous(t *testing. } h := &Handler{repo: r} - h.processFlowItem(flowItem{N: "99_2_10_tcp", U: 120, D: 80}) + h.processFlowItem(1, flowItem{N: "99_2_10_tcp", U: 120, D: 80}) updatedA, _ := r.GetPeerShare(shareA.ID) updatedB, _ := r.GetPeerShare(shareB.ID) diff --git a/go-backend/internal/http/handler/handler.go b/go-backend/internal/http/handler/handler.go index de94a63..0759b37 100644 --- a/go-backend/internal/http/handler/handler.go +++ b/go-backend/internal/http/handler/handler.go @@ -704,7 +704,8 @@ func (h *Handler) flowConfig(w http.ResponseWriter, r *http.Request) { func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) { secret := r.URL.Query().Get("secret") - if ok, _ := h.repo.NodeExistsBySecret(secret); !ok { + node, _ := h.repo.GetNodeBySecret(secret) + if node == nil { w.Header().Set("Content-Type", "text/plain; charset=utf-8") _, _ = w.Write([]byte("ok")) return @@ -715,7 +716,7 @@ func (h *Handler) flowUpload(w http.ResponseWriter, r *http.Request) { var items []flowItem if json.Unmarshal([]byte(raw), &items) == nil { for _, item := range items { - h.processFlowItem(item) + h.processFlowItem(node.ID, item) } } } diff --git a/go-backend/internal/store/repo/repository.go b/go-backend/internal/store/repo/repository.go index 0113be9..c1d2b04 100644 --- a/go-backend/internal/store/repo/repository.go +++ b/go-backend/internal/store/repo/repository.go @@ -1287,6 +1287,28 @@ func (r *Repository) ListActiveForwardPeerShareRuntimesByServiceName(serviceName return items, nil } +func (r *Repository) ListActiveForwardPeerShareRuntimesByNodeAndServiceName(nodeID int64, serviceName string) ([]model.PeerShareRuntime, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + serviceName = strings.TrimSpace(serviceName) + if serviceName == "" { + return []model.PeerShareRuntime{}, nil + } + var items []model.PeerShareRuntime + err := r.db.Where("node_id = ? AND service_name = ? AND status = 1 AND role = ?", nodeID, serviceName, "forward"). + Order("id ASC"). + Find(&items).Error + if err != nil { + return nil, err + } + if items == nil { + items = make([]model.PeerShareRuntime, 0) + } + return items, nil +} + + func (r *Repository) ListActiveForwardPeerShareRuntimeServiceNamesByNode(nodeID int64) ([]string, error) { if r == nil || r.db == nil { return nil, errors.New("repository not initialized") @@ -1333,6 +1355,27 @@ func (r *Repository) GetActiveForwardPeerShareRuntimeByPort(shareID int64, port return &item, nil } +func (r *Repository) GetActiveForwardPeerShareRuntimeByServiceName(shareID int64, serviceName string) (*model.PeerShareRuntime, error) { + if r == nil || r.db == nil { + return nil, errors.New("repository not initialized") + } + serviceName = strings.TrimSpace(serviceName) + if shareID <= 0 || serviceName == "" { + return nil, nil + } + var item model.PeerShareRuntime + err := r.db.Where("share_id = ? AND service_name = ? AND status = 1 AND role = ?", shareID, serviceName, "forward"). + Order("id ASC"). + First(&item).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &item, nil +} + func (r *Repository) ExistsActivePeerShareRuntimeOnNodePort(nodeID int64, port int) (bool, error) { if r == nil || r.db == nil { return false, errors.New("repository not initialized") @@ -1376,6 +1419,27 @@ func (r *Repository) MarkPeerShareRuntimeReleasedByPort(shareID int64, port int, }).Error } +func (r *Repository) MarkForwardPeerShareRuntimeReleasedByServiceName(shareID int64, serviceName string, updatedTime int64) error { + if r == nil || r.db == nil { + return errors.New("repository not initialized") + } + serviceName = strings.TrimSpace(serviceName) + if shareID <= 0 || serviceName == "" { + return nil + } + if updatedTime <= 0 { + updatedTime = unixMilliNow() + } + return r.db.Model(&model.PeerShareRuntime{}). + Where("share_id = ? AND status = 1 AND role = ? AND service_name = ?", shareID, "forward", serviceName). + Updates(map[string]interface{}{ + "status": 0, + "applied": 0, + "service_name": "", + "updated_time": updatedTime, + }).Error +} + // ─── FederationTunnelBinding ───────────────────────────────────────── func (r *Repository) UpsertFederationTunnelBinding(item *model.FederationTunnelBinding) error { diff --git a/go-backend/tests/contract/federation_dual_panel_contract_test.go b/go-backend/tests/contract/federation_dual_panel_contract_test.go index 00aa449..3e9cbfb 100644 --- a/go-backend/tests/contract/federation_dual_panel_contract_test.go +++ b/go-backend/tests/contract/federation_dual_panel_contract_test.go @@ -166,7 +166,7 @@ func TestFederationDualPanelMiddleExitAutoPortContract(t *testing.T) { assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1 AND applied = 1`, middleShareID, 1) assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1 AND applied = 1`, exitShareID, 1) - assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ?`, entryShareID, 0) + assertCount(t, providerRepo, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND status = 1 AND applied = 1`, entryShareID, 1) } func TestFederationDualPanelRemoteDiagnosisContract(t *testing.T) {