mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-10-03 01:06:36 +08:00
fix(backend): keep shared federation port-forward listeners alive (#205)
## Summary - prevent shared federation forward services from being misclassified as orphaned during node config cleanup - harden peer-share runtime mapping for service-name based flow accounting and node/port conflict checks - add regression tests for listener persistence, cleanup guards, runtime binding, and federation port-forward lifecycle ## Verification - go test ./internal/http/handler -run \"TestCleanOrphanedServicesSkipsActiveSharedForwardRuntimeServices|TestCleanOrphanedServicesSkipsFederationServicePrefix|TestProcessFlowItemTracksPeerShareFlowByForwardServiceName|TestProcessFlowItemSkipsPeerShareFlowWhenServiceNameIsAmbiguous|TestFederationTunnelCreateRejectsOccupiedPort\" - go test ./... - make build
This commit is contained in:
@@ -766,6 +766,37 @@ func (h *Handler) federationTunnelCreate(w http.ResponseWriter, r *http.Request)
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
usedPorts, err := h.repo.ListUsedPortsOnNode(share.NodeID)
|
||||||
|
if err != nil {
|
||||||
|
response.WriteJSON(w, response.Err(-2, err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for _, port := range usedPorts {
|
||||||
|
if port == req.RemotePort {
|
||||||
|
response.WriteJSON(w, response.Err(403, "Port already in use"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
runtimeOnPort, err := h.repo.GetActiveForwardPeerShareRuntimeByPort(share.ID, req.RemotePort)
|
||||||
|
if err != nil {
|
||||||
|
response.WriteJSON(w, response.Err(-2, err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if runtimeOnPort != nil {
|
||||||
|
response.WriteJSON(w, response.Err(403, "Port already in use"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
existsOnNodePort, err := h.repo.ExistsActivePeerShareRuntimeOnNodePort(share.NodeID, req.RemotePort)
|
||||||
|
if err != nil {
|
||||||
|
response.WriteJSON(w, response.Err(-2, err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if existsOnNodePort {
|
||||||
|
response.WriteJSON(w, response.Err(403, "Port already in use"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
now := time.Now().UnixMilli()
|
now := time.Now().UnixMilli()
|
||||||
tunnelID, err := h.repo.CreateFederationTunnel(
|
tunnelID, err := h.repo.CreateFederationTunnel(
|
||||||
fmt.Sprintf("Share-%d-Port-%d", share.ID, req.RemotePort),
|
fmt.Sprintf("Share-%d-Port-%d", share.ID, req.RemotePort),
|
||||||
@@ -780,6 +811,30 @@ func (h *Handler) federationTunnelCreate(w http.ResponseWriter, r *http.Request)
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
runtime := &repo.PeerShareRuntime{
|
||||||
|
ShareID: share.ID,
|
||||||
|
NodeID: share.NodeID,
|
||||||
|
ReservationID: randomToken(24),
|
||||||
|
ResourceKey: fmt.Sprintf("federation-forward-%d-%d-%d", share.ID, tunnelID, req.RemotePort),
|
||||||
|
BindingID: "",
|
||||||
|
Role: "forward",
|
||||||
|
ChainName: "",
|
||||||
|
ServiceName: "",
|
||||||
|
Protocol: defaultString(req.Protocol, "tcp"),
|
||||||
|
Strategy: "fifo",
|
||||||
|
Port: req.RemotePort,
|
||||||
|
Target: strings.TrimSpace(req.Target),
|
||||||
|
Applied: 0,
|
||||||
|
Status: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}
|
||||||
|
if err := h.repo.CreatePeerShareRuntime(runtime); err != nil {
|
||||||
|
_ = h.deleteTunnelByID(tunnelID)
|
||||||
|
response.WriteJSON(w, response.Err(-2, err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
h.wsServer.SendCommand(share.NodeID, "reload", nil, time.Second*5)
|
h.wsServer.SendCommand(share.NodeID, "reload", nil, time.Second*5)
|
||||||
|
|
||||||
response.WriteJSON(w, response.OK(map[string]interface{}{
|
response.WriteJSON(w, response.OK(map[string]interface{}{
|
||||||
@@ -1211,9 +1266,83 @@ func (h *Handler) federationRuntimeCommand(w http.ResponseWriter, r *http.Reques
|
|||||||
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if strings.EqualFold(cmd, "addservice") || strings.EqualFold(cmd, "updateservice") {
|
||||||
|
h.bindPeerShareForwardRuntimeServices(share, req.Data)
|
||||||
|
}
|
||||||
response.WriteJSON(w, response.OK(res))
|
response.WriteJSON(w, response.OK(res))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type federationForwardServiceBinding struct {
|
||||||
|
Name string
|
||||||
|
Port int
|
||||||
|
}
|
||||||
|
|
||||||
|
func parseFederationForwardServiceBindings(data interface{}) []federationForwardServiceBinding {
|
||||||
|
dataMap, ok := data.(map[string]interface{})
|
||||||
|
if !ok {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
rawServices, ok := dataMap["services"]
|
||||||
|
if !ok {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
serviceList, ok := rawServices.([]interface{})
|
||||||
|
if !ok {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings := make([]federationForwardServiceBinding, 0, len(serviceList))
|
||||||
|
for _, item := range serviceList {
|
||||||
|
svcMap, ok := item.(map[string]interface{})
|
||||||
|
if !ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
name := normalizeForwardRuntimeServiceName(asString(svcMap["name"]))
|
||||||
|
if name == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, _, _, ok := parseFlowServiceIDs(name); !ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
addr := strings.TrimSpace(asString(svcMap["addr"]))
|
||||||
|
if addr == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_, portStr, err := net.SplitHostPort(addr)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
port, err := strconv.Atoi(portStr)
|
||||||
|
if err != nil || port <= 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
bindings = append(bindings, federationForwardServiceBinding{Name: name, Port: port})
|
||||||
|
}
|
||||||
|
return bindings
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *Handler) bindPeerShareForwardRuntimeServices(share *repo.PeerShare, data interface{}) {
|
||||||
|
if h == nil || h.repo == nil || share == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
bindings := parseFederationForwardServiceBindings(data)
|
||||||
|
if len(bindings) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
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 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if runtime.ServiceName == binding.Name && runtime.Applied == 1 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_ = h.repo.UpdatePeerShareRuntimeServiceName(runtime.ID, binding.Name, now)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func isFederationRuntimeCommandAllowed(commandType string) bool {
|
func isFederationRuntimeCommandAllowed(commandType string) bool {
|
||||||
switch strings.ToLower(strings.TrimSpace(commandType)) {
|
switch strings.ToLower(strings.TrimSpace(commandType)) {
|
||||||
case "addservice", "updateservice", "deleteservice", "pauseservice", "resumeservice", "addchains", "deletechains", "addlimiters", "deletelimiters", "tcpping", "reload":
|
case "addservice", "updateservice", "deleteservice", "pauseservice", "resumeservice", "addchains", "deletechains", "addlimiters", "deletelimiters", "tcpping", "reload":
|
||||||
|
|||||||
@@ -414,6 +414,262 @@ func TestFederationShareResetFlow(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestFederationTunnelCreateCreatesPeerShareRuntime(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.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.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, is_remote, remote_url, remote_token, remote_config)
|
||||||
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "federation-forward-node", "federation-forward-secret", "10.90.80.70", "10.90.80.70", "", "24000-24020", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 0, "", "", "").Error; err != nil {
|
||||||
|
t.Fatalf("insert node: %v", err)
|
||||||
|
}
|
||||||
|
nodeID := mustLastInsertID(t, r, "federation-forward-node")
|
||||||
|
|
||||||
|
if err := r.CreatePeerShare(&repo.PeerShare{
|
||||||
|
Name: "federation-forward-share",
|
||||||
|
NodeID: nodeID,
|
||||||
|
Token: "federation-forward-token",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 24000,
|
||||||
|
PortRangeEnd: 24020,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create share: %v", err)
|
||||||
|
}
|
||||||
|
share, err := r.GetPeerShareByToken("federation-forward-token")
|
||||||
|
if err != nil || share == nil {
|
||||||
|
t.Fatalf("load share: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
body, err := json.Marshal(federationTunnelRequest{
|
||||||
|
Protocol: "tcp",
|
||||||
|
RemotePort: 24001,
|
||||||
|
Target: "1.1.1.1:443",
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal request: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/federation/tunnel/create", bytes.NewReader(body))
|
||||||
|
req.Header.Set("Authorization", "Bearer "+share.Token)
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
res := httptest.NewRecorder()
|
||||||
|
|
||||||
|
h.federationTunnelCreate(res, req)
|
||||||
|
|
||||||
|
if res.Code != http.StatusOK {
|
||||||
|
t.Fatalf("expected status %d, got %d", http.StatusOK, res.Code)
|
||||||
|
}
|
||||||
|
|
||||||
|
var payload response.R
|
||||||
|
if err := json.NewDecoder(res.Body).Decode(&payload); err != nil {
|
||||||
|
t.Fatalf("decode response: %v", err)
|
||||||
|
}
|
||||||
|
if payload.Code != 0 {
|
||||||
|
t.Fatalf("expected response code 0, got %d (%s)", payload.Code, payload.Msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
runtimeCount := mustQueryInt(t, r, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND port = ? AND status = 1`, share.ID, 24001)
|
||||||
|
if runtimeCount != 1 {
|
||||||
|
t.Fatalf("expected 1 runtime row for new federation forward tunnel, got %d", runtimeCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestFederationTunnelCreateRejectsOccupiedPort(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.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.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, is_remote, remote_url, remote_token, remote_config)
|
||||||
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`, "federation-port-check-node", "federation-port-check-secret", "10.91.80.70", "10.91.80.70", "", "24100-24120", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 0, "", "", "").Error; err != nil {
|
||||||
|
t.Fatalf("insert node: %v", err)
|
||||||
|
}
|
||||||
|
nodeID := mustLastInsertID(t, r, "federation-port-check-node")
|
||||||
|
|
||||||
|
if err := r.CreatePeerShare(&repo.PeerShare{
|
||||||
|
Name: "federation-port-check-share",
|
||||||
|
NodeID: nodeID,
|
||||||
|
Token: "federation-port-check-token",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 24100,
|
||||||
|
PortRangeEnd: 24120,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create share: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
create := func() response.R {
|
||||||
|
body, err := json.Marshal(federationTunnelRequest{Protocol: "tcp", RemotePort: 24101, Target: "1.1.1.1:443"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal request: %v", err)
|
||||||
|
}
|
||||||
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/federation/tunnel/create", bytes.NewReader(body))
|
||||||
|
req.Header.Set("Authorization", "Bearer federation-port-check-token")
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
res := httptest.NewRecorder()
|
||||||
|
h.federationTunnelCreate(res, req)
|
||||||
|
if res.Code != http.StatusOK {
|
||||||
|
t.Fatalf("expected status %d, got %d", http.StatusOK, res.Code)
|
||||||
|
}
|
||||||
|
var payload response.R
|
||||||
|
if err := json.NewDecoder(res.Body).Decode(&payload); err != nil {
|
||||||
|
t.Fatalf("decode response: %v", err)
|
||||||
|
}
|
||||||
|
return payload
|
||||||
|
}
|
||||||
|
|
||||||
|
first := create()
|
||||||
|
if first.Code != 0 {
|
||||||
|
t.Fatalf("expected first create success, got %d (%s)", first.Code, first.Msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
second := create()
|
||||||
|
if second.Code != 403 {
|
||||||
|
t.Fatalf("expected second create to be rejected with 403, got %d (%s)", second.Code, second.Msg)
|
||||||
|
}
|
||||||
|
if second.Msg != "Port already in use" {
|
||||||
|
t.Fatalf("expected occupied port message, got %q", second.Msg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeleteTunnelReleasesFederationForwardRuntimeByPort(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.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: "delete-forward-share",
|
||||||
|
NodeID: 1,
|
||||||
|
Token: "delete-forward-token",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 25000,
|
||||||
|
PortRangeEnd: 25020,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create share: %v", err)
|
||||||
|
}
|
||||||
|
share, err := r.GetPeerShareByToken("delete-forward-token")
|
||||||
|
if err != nil || share == nil {
|
||||||
|
t.Fatalf("load share: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
tunnelName := fmt.Sprintf("Share-%d-Port-%d", share.ID, 25001)
|
||||||
|
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, tunnelName, 1.0, 1, "tcp", 1, now, now, 1, nil, 0).Error; err != nil {
|
||||||
|
t.Fatalf("insert tunnel: %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, "del-r1", "del-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 25001, "", 1, 1, now, now).Error; err != nil {
|
||||||
|
t.Fatalf("insert runtime: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := h.deleteTunnelByID(1); err != nil {
|
||||||
|
t.Fatalf("delete tunnel: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
activeCount := mustQueryInt(t, r, `SELECT COUNT(1) FROM peer_share_runtime WHERE share_id = ? AND port = ? AND status = 1`, share.ID, 25001)
|
||||||
|
if activeCount != 0 {
|
||||||
|
t.Fatalf("expected runtime released after tunnel delete, active rows=%d", activeCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBindPeerShareForwardRuntimeServicesOnlyBindsForwardRole(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.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-forward-role-share",
|
||||||
|
NodeID: 1,
|
||||||
|
Token: "bind-forward-role-token",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 26000,
|
||||||
|
PortRangeEnd: 26020,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create share: %v", err)
|
||||||
|
}
|
||||||
|
share, err := r.GetPeerShareByToken("bind-forward-role-token")
|
||||||
|
if err != nil || share == nil {
|
||||||
|
t.Fatalf("load share: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := r.DB().Exec(`
|
||||||
|
INSERT INTO peer_share_runtime(id, 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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),
|
||||||
|
(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`,
|
||||||
|
1, share.ID, share.NodeID, "bind-r1", "bind-rk1", "", "forward", "", "", "tcp", "fifo", 26001, "", 0, 1, now, now,
|
||||||
|
2, share.ID, share.NodeID, "bind-r2", "bind-rk2", "", "middle", "", "", "tcp", "round", 26002, "", 0, 1, now, now,
|
||||||
|
).Error; err != nil {
|
||||||
|
t.Fatalf("insert runtimes: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
h.bindPeerShareForwardRuntimeServices(share, map[string]interface{}{
|
||||||
|
"services": []interface{}{
|
||||||
|
map[string]interface{}{"name": "77_2_10_tcp", "addr": "[::]:26001"},
|
||||||
|
map[string]interface{}{"name": "88_2_10_tcp", "addr": "[::]:26002"},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
forwardServiceName := ""
|
||||||
|
middleServiceName := ""
|
||||||
|
if err := r.DB().Raw(`SELECT service_name FROM peer_share_runtime WHERE id = 1`).Scan(&forwardServiceName).Error; err != nil {
|
||||||
|
t.Fatalf("load forward runtime service name: %v", err)
|
||||||
|
}
|
||||||
|
if err := r.DB().Raw(`SELECT service_name FROM peer_share_runtime WHERE id = 2`).Scan(&middleServiceName).Error; err != nil {
|
||||||
|
t.Fatalf("load middle runtime service name: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if forwardServiceName != "77_2_10" {
|
||||||
|
t.Fatalf("expected forward runtime service name bound, got %q", forwardServiceName)
|
||||||
|
}
|
||||||
|
if middleServiceName != "" {
|
||||||
|
t.Fatalf("expected non-forward runtime unchanged, got %q", middleServiceName)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestFederationRemoteUsageList(t *testing.T) {
|
func TestFederationRemoteUsageList(t *testing.T) {
|
||||||
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db"))
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel.db"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ func (h *Handler) processFlowItem(item flowItem) {
|
|||||||
if ok {
|
if ok {
|
||||||
inFlow, outFlow := h.scaleFlowByTunnel(forwardID, item.D, item.U)
|
inFlow, outFlow := h.scaleFlowByTunnel(forwardID, item.D, item.U)
|
||||||
_ = h.repo.AddFlow(forwardID, userID, userTunnelID, inFlow, outFlow)
|
_ = h.repo.AddFlow(forwardID, userID, userTunnelID, inFlow, outFlow)
|
||||||
h.processPeerShareFlowFromForward(forwardID, item)
|
h.processPeerShareFlowFromForward(forwardID, serviceName, item)
|
||||||
|
|
||||||
if userTunnelID > 0 {
|
if userTunnelID > 0 {
|
||||||
h.enforceFlowPolicies(userID, userTunnelID)
|
h.enforceFlowPolicies(userID, userTunnelID)
|
||||||
@@ -88,6 +88,28 @@ func parsePeerShareRuntimeServiceID(serviceName string) (int64, bool) {
|
|||||||
return runtimeID, true
|
return runtimeID, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func parsePeerShareInfoFromFederationTunnelName(tunnelName string) (int64, int, bool) {
|
||||||
|
tunnelName = strings.TrimSpace(tunnelName)
|
||||||
|
if !strings.HasPrefix(tunnelName, "Share-") {
|
||||||
|
return 0, 0, false
|
||||||
|
}
|
||||||
|
raw := strings.TrimPrefix(tunnelName, "Share-")
|
||||||
|
idx := strings.Index(raw, "-Port-")
|
||||||
|
if idx <= 0 {
|
||||||
|
return 0, 0, false
|
||||||
|
}
|
||||||
|
shareID, err := strconv.ParseInt(raw[:idx], 10, 64)
|
||||||
|
if err != nil || shareID <= 0 {
|
||||||
|
return 0, 0, false
|
||||||
|
}
|
||||||
|
portValue := strings.TrimSpace(raw[idx+len("-Port-"):])
|
||||||
|
port, err := strconv.Atoi(portValue)
|
||||||
|
if err != nil || port <= 0 {
|
||||||
|
return 0, 0, false
|
||||||
|
}
|
||||||
|
return shareID, port, true
|
||||||
|
}
|
||||||
|
|
||||||
func parsePeerShareIDFromFederationTunnelName(tunnelName string) (int64, bool) {
|
func parsePeerShareIDFromFederationTunnelName(tunnelName string) (int64, bool) {
|
||||||
tunnelName = strings.TrimSpace(tunnelName)
|
tunnelName = strings.TrimSpace(tunnelName)
|
||||||
if !strings.HasPrefix(tunnelName, "Share-") {
|
if !strings.HasPrefix(tunnelName, "Share-") {
|
||||||
@@ -131,12 +153,15 @@ func (h *Handler) processPeerShareFlow(runtimeID int64, item flowItem) {
|
|||||||
h.enforcePeerShareFlowLimit(share.ID)
|
h.enforcePeerShareFlowLimit(share.ID)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) processPeerShareFlowFromForward(forwardID int64, item flowItem) {
|
func (h *Handler) processPeerShareFlowFromForward(forwardID int64, serviceName string, item flowItem) {
|
||||||
if h == nil || h.repo == nil || forwardID <= 0 {
|
if h == nil || h.repo == nil || forwardID <= 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
forward, err := h.getForwardRecord(forwardID)
|
forward, err := h.getForwardRecord(forwardID)
|
||||||
if err != nil || forward == nil {
|
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)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
tunnel, err := h.getTunnelRecord(forward.TunnelID)
|
tunnel, err := h.getTunnelRecord(forward.TunnelID)
|
||||||
@@ -169,6 +194,54 @@ func (h *Handler) processPeerShareFlowFromForward(forwardID int64, item flowItem
|
|||||||
h.enforcePeerShareFlowLimit(share.ID)
|
h.enforcePeerShareFlowLimit(share.ID)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func normalizeForwardRuntimeServiceName(serviceName string) string {
|
||||||
|
name := strings.TrimSpace(serviceName)
|
||||||
|
if strings.HasSuffix(name, "_tcp") {
|
||||||
|
return strings.TrimSuffix(name, "_tcp")
|
||||||
|
}
|
||||||
|
if strings.HasSuffix(name, "_udp") {
|
||||||
|
return strings.TrimSuffix(name, "_udp")
|
||||||
|
}
|
||||||
|
return name
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *Handler) processPeerShareFlowByServiceName(serviceName string, item flowItem) {
|
||||||
|
if h == nil || h.repo == nil || strings.TrimSpace(serviceName) == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
delta := item.D + item.U
|
||||||
|
if delta <= 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
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)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(runtimes) != 1 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
runtime := runtimes[0]
|
||||||
|
|
||||||
|
_ = h.repo.AddPeerShareCurrentFlow(runtime.ShareID, delta)
|
||||||
|
|
||||||
|
matchedShare, err := h.repo.GetPeerShare(runtime.ShareID)
|
||||||
|
if err != nil || matchedShare == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if isPeerShareFlowExceeded(matchedShare) {
|
||||||
|
h.enforcePeerShareFlowLimit(matchedShare.ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (h *Handler) enforcePeerShareFlowLimit(shareID int64) {
|
func (h *Handler) enforcePeerShareFlowLimit(shareID int64) {
|
||||||
if h == nil || h.repo == nil || shareID <= 0 {
|
if h == nil || h.repo == nil || shareID <= 0 {
|
||||||
return
|
return
|
||||||
@@ -327,11 +400,34 @@ func (h *Handler) cleanNodeConfigs(nodeID int64, rawConfig string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) cleanOrphanedServices(nodeID int64, services []namedConfigItem) {
|
func (h *Handler) cleanOrphanedServices(nodeID int64, services []namedConfigItem) {
|
||||||
|
runtimeServiceNames, err := h.repo.ListActiveForwardPeerShareRuntimeServiceNamesByNode(nodeID)
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
runtimeServiceSet := make(map[string]struct{}, len(runtimeServiceNames))
|
||||||
|
for _, serviceName := range runtimeServiceNames {
|
||||||
|
serviceName = strings.TrimSpace(serviceName)
|
||||||
|
if serviceName == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
runtimeServiceSet[serviceName] = struct{}{}
|
||||||
|
}
|
||||||
|
|
||||||
for _, item := range services {
|
for _, item := range services {
|
||||||
name := strings.TrimSpace(item.Name)
|
name := strings.TrimSpace(item.Name)
|
||||||
if name == "" || name == "web_api" {
|
if name == "" || name == "web_api" {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
if strings.HasPrefix(name, "fed_svc_") {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
normalizedName := normalizeForwardRuntimeServiceName(name)
|
||||||
|
if _, ok := runtimeServiceSet[normalizedName]; ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, ok := runtimeServiceSet[name]; ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
parts := strings.Split(name, "_")
|
parts := strings.Split(name, "_")
|
||||||
if len(parts) >= 3 {
|
if len(parts) >= 3 {
|
||||||
|
|||||||
@@ -130,3 +130,172 @@ func TestProcessFlowItemTracksPeerShareFlowForFederationPortForward(t *testing.T
|
|||||||
t.Fatalf("expected current_flow=200, got %d", updatedShare.CurrentFlow)
|
t.Fatalf("expected current_flow=200, got %d", updatedShare.CurrentFlow)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestProcessFlowItemTracksPeerShareFlowByForwardServiceName(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel-forward-service.db"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("open repo: %v", err)
|
||||||
|
}
|
||||||
|
defer r.Close()
|
||||||
|
|
||||||
|
now := time.Now().UnixMilli()
|
||||||
|
if err := r.CreatePeerShare(&repo.PeerShare{
|
||||||
|
Name: "forward-service-share",
|
||||||
|
NodeID: 1,
|
||||||
|
Token: "forward-service-token",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 31000,
|
||||||
|
PortRangeEnd: 31010,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create peer share: %v", err)
|
||||||
|
}
|
||||||
|
share, err := r.GetPeerShareByToken("forward-service-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, share.NodeID, "svc-r1", "svc-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 31001, "", 1, 1, now, now).Error; err != nil {
|
||||||
|
t.Fatalf("insert peer_share_runtime: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
h := &Handler{repo: r}
|
||||||
|
h.processFlowItem(flowItem{N: "20_2_10_tcp", U: 120, D: 80})
|
||||||
|
|
||||||
|
updatedShare, err := r.GetPeerShare(share.ID)
|
||||||
|
if err != nil || updatedShare == nil {
|
||||||
|
t.Fatalf("reload share: %v", err)
|
||||||
|
}
|
||||||
|
if updatedShare.CurrentFlow != 200 {
|
||||||
|
t.Fatalf("expected current_flow=200, got %d", updatedShare.CurrentFlow)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestProcessFlowItemSkipsPeerShareFlowWhenServiceNameIsAmbiguous(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel-forward-ambiguous.db"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("open repo: %v", err)
|
||||||
|
}
|
||||||
|
defer r.Close()
|
||||||
|
|
||||||
|
now := time.Now().UnixMilli()
|
||||||
|
if err := r.CreatePeerShare(&repo.PeerShare{
|
||||||
|
Name: "ambiguous-share-a",
|
||||||
|
NodeID: 1,
|
||||||
|
Token: "ambiguous-token-a",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 31100,
|
||||||
|
PortRangeEnd: 31110,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create share A: %v", err)
|
||||||
|
}
|
||||||
|
if err := r.CreatePeerShare(&repo.PeerShare{
|
||||||
|
Name: "ambiguous-share-b",
|
||||||
|
NodeID: 1,
|
||||||
|
Token: "ambiguous-token-b",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 31200,
|
||||||
|
PortRangeEnd: 31210,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create share B: %v", err)
|
||||||
|
}
|
||||||
|
shareA, _ := r.GetPeerShareByToken("ambiguous-token-a")
|
||||||
|
shareB, _ := r.GetPeerShareByToken("ambiguous-token-b")
|
||||||
|
|
||||||
|
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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),
|
||||||
|
(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
`,
|
||||||
|
shareA.ID, 1, "amb-r1", "amb-rk1", "", "forward", "", "99_2_10", "tcp", "fifo", 31101, "", 1, 1, now, now,
|
||||||
|
shareB.ID, 1, "amb-r2", "amb-rk2", "", "forward", "", "99_2_10", "tcp", "fifo", 31201, "", 1, 1, now, now,
|
||||||
|
).Error; err != nil {
|
||||||
|
t.Fatalf("insert ambiguous runtimes: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
h := &Handler{repo: r}
|
||||||
|
h.processFlowItem(flowItem{N: "99_2_10_tcp", U: 120, D: 80})
|
||||||
|
|
||||||
|
updatedA, _ := r.GetPeerShare(shareA.ID)
|
||||||
|
updatedB, _ := r.GetPeerShare(shareB.ID)
|
||||||
|
if updatedA.CurrentFlow != 0 || updatedB.CurrentFlow != 0 {
|
||||||
|
t.Fatalf("expected ambiguous service flow to be skipped, got shareA=%d shareB=%d", updatedA.CurrentFlow, updatedB.CurrentFlow)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanOrphanedServicesSkipsActiveSharedForwardRuntimeServices(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel-cleanup-runtime.db"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("open repo: %v", err)
|
||||||
|
}
|
||||||
|
defer r.Close()
|
||||||
|
|
||||||
|
now := time.Now().UnixMilli()
|
||||||
|
if err := r.CreatePeerShare(&repo.PeerShare{
|
||||||
|
Name: "cleanup-runtime-share",
|
||||||
|
NodeID: 1,
|
||||||
|
Token: "cleanup-runtime-token",
|
||||||
|
MaxBandwidth: 0,
|
||||||
|
CurrentFlow: 0,
|
||||||
|
PortRangeStart: 31300,
|
||||||
|
PortRangeEnd: 31310,
|
||||||
|
IsActive: 1,
|
||||||
|
CreatedTime: now,
|
||||||
|
UpdatedTime: now,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create peer share: %v", err)
|
||||||
|
}
|
||||||
|
share, err := r.GetPeerShareByToken("cleanup-runtime-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, share.NodeID, "cleanup-r1", "cleanup-rk1", "", "forward", "", "20_2_10", "tcp", "fifo", 31301, "", 1, 1, now, now).Error; err != nil {
|
||||||
|
t.Fatalf("insert peer_share_runtime: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
h := &Handler{repo: r}
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
if rec := recover(); rec != nil {
|
||||||
|
t.Fatalf("cleanOrphanedServices should skip active shared runtime service; got panic: %v", rec)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
h.cleanOrphanedServices(share.NodeID, []namedConfigItem{{Name: "20_2_10_tcp"}})
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanOrphanedServicesSkipsFederationServicePrefix(t *testing.T) {
|
||||||
|
r, err := repo.Open(filepath.Join(t.TempDir(), "panel-cleanup-fed-svc.db"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("open repo: %v", err)
|
||||||
|
}
|
||||||
|
defer r.Close()
|
||||||
|
|
||||||
|
h := &Handler{repo: r}
|
||||||
|
|
||||||
|
defer func() {
|
||||||
|
if rec := recover(); rec != nil {
|
||||||
|
t.Fatalf("cleanOrphanedServices should skip fed_svc_ service names; got panic: %v", rec)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
h.cleanOrphanedServices(1, []namedConfigItem{{Name: "fed_svc_999_tcp"}})
|
||||||
|
}
|
||||||
|
|||||||
@@ -2794,7 +2794,21 @@ func (h *Handler) deleteNodeByID(id int64) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) deleteTunnelByID(id int64) error {
|
func (h *Handler) deleteTunnelByID(id int64) error {
|
||||||
return h.repo.DeleteTunnelCascade(id)
|
if h == nil || h.repo == nil {
|
||||||
|
return errors.New("repository not initialized")
|
||||||
|
}
|
||||||
|
|
||||||
|
tunnelName, _ := h.repo.GetTunnelName(id)
|
||||||
|
if err := h.repo.DeleteTunnelCascade(id); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
shareID, port, ok := parsePeerShareInfoFromFederationTunnelName(tunnelName)
|
||||||
|
if !ok {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return h.repo.MarkPeerShareRuntimeReleasedByPort(shareID, port, time.Now().UnixMilli())
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) deleteForwardByID(id int64) error {
|
func (h *Handler) deleteForwardByID(id int64) error {
|
||||||
|
|||||||
@@ -1270,6 +1270,98 @@ func (r *Repository) ListActivePeerShareRuntimePorts(shareID int64, nodeID int64
|
|||||||
return ports, nil
|
return ports, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (r *Repository) ListActiveForwardPeerShareRuntimesByServiceName(serviceName string) ([]model.PeerShareRuntime, error) {
|
||||||
|
if r == nil || r.db == nil {
|
||||||
|
return nil, errors.New("repository not initialized")
|
||||||
|
}
|
||||||
|
var items []model.PeerShareRuntime
|
||||||
|
err := r.db.Where("service_name = ? AND status = 1 AND role = ?", 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")
|
||||||
|
}
|
||||||
|
var names []string
|
||||||
|
err := r.db.Model(&model.PeerShareRuntime{}).
|
||||||
|
Where("node_id = ? AND status = 1 AND role = ? AND service_name <> ''", nodeID, "forward").
|
||||||
|
Pluck("service_name", &names).Error
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if names == nil {
|
||||||
|
names = make([]string, 0)
|
||||||
|
}
|
||||||
|
return names, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *Repository) GetActiveForwardPeerShareRuntimeByPort(shareID int64, port int) (*model.PeerShareRuntime, error) {
|
||||||
|
if r == nil || r.db == nil {
|
||||||
|
return nil, errors.New("repository not initialized")
|
||||||
|
}
|
||||||
|
var item model.PeerShareRuntime
|
||||||
|
err := r.db.Where("share_id = ? AND port = ? AND status = 1 AND role = ?", shareID, port, "forward").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")
|
||||||
|
}
|
||||||
|
var count int64
|
||||||
|
err := r.db.Model(&model.PeerShareRuntime{}).
|
||||||
|
Where("node_id = ? AND port = ? AND status = 1", nodeID, port).
|
||||||
|
Count(&count).Error
|
||||||
|
if err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
return count > 0, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *Repository) UpdatePeerShareRuntimeServiceName(id int64, serviceName string, updatedTime int64) error {
|
||||||
|
if r == nil || r.db == nil {
|
||||||
|
return errors.New("repository not initialized")
|
||||||
|
}
|
||||||
|
return r.db.Model(&model.PeerShareRuntime{}).Where("id = ?", id).Updates(map[string]interface{}{
|
||||||
|
"service_name": serviceName,
|
||||||
|
"applied": 1,
|
||||||
|
"updated_time": updatedTime,
|
||||||
|
}).Error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *Repository) MarkPeerShareRuntimeReleasedByPort(shareID int64, port int, updatedTime int64) error {
|
||||||
|
if r == nil || r.db == nil {
|
||||||
|
return errors.New("repository not initialized")
|
||||||
|
}
|
||||||
|
if shareID <= 0 || port <= 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if updatedTime <= 0 {
|
||||||
|
updatedTime = unixMilliNow()
|
||||||
|
}
|
||||||
|
return r.db.Model(&model.PeerShareRuntime{}).Where("share_id = ? AND port = ? AND status = 1", shareID, port).Updates(map[string]interface{}{
|
||||||
|
"status": 0,
|
||||||
|
"applied": 0,
|
||||||
|
"service_name": "",
|
||||||
|
"updated_time": updatedTime,
|
||||||
|
}).Error
|
||||||
|
}
|
||||||
|
|
||||||
// ─── FederationTunnelBinding ─────────────────────────────────────────
|
// ─── FederationTunnelBinding ─────────────────────────────────────────
|
||||||
|
|
||||||
func (r *Repository) UpsertFederationTunnelBinding(item *model.FederationTunnelBinding) error {
|
func (r *Repository) UpsertFederationTunnelBinding(item *model.FederationTunnelBinding) error {
|
||||||
|
|||||||
Reference in New Issue
Block a user