diff --git a/go-backend/tests/contract/federation_forward_flow_linkage_contract_test.go b/go-backend/tests/contract/federation_forward_flow_linkage_contract_test.go index aec6e10..049d05f 100644 --- a/go-backend/tests/contract/federation_forward_flow_linkage_contract_test.go +++ b/go-backend/tests/contract/federation_forward_flow_linkage_contract_test.go @@ -386,6 +386,221 @@ func TestFederationForwardCardFlowLinkageContractSplitShareFlowAcrossMultipleFor } } +func TestFederationForwardCardFlowLinkageContractResolvesShareByTunnelBindingWhenTunnelNameIsCustom(t *testing.T) { + secret := "federation-forward-binding-flow-contract-jwt" + router, r := setupContractRouter(t, secret) + + adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret) + if err != nil { + t.Fatalf("generate admin token: %v", err) + } + + now := time.Now().UnixMilli() + remoteShareID := int64(901) + remoteShareFlow := int64(5000) + + 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(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + "flow-binding-remote-node", "flow-binding-remote-secret", "10.31.41.51", "10.31.41.51", "", "33000-33020", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0, 1, "", "", fmt.Sprintf(`{"shareId":%d,"maxBandwidth":0,"currentFlow":%d,"portRangeStart":33000,"portRangeEnd":33020}`, remoteShareID, remoteShareFlow), + ).Error; err != nil { + t.Fatalf("insert remote node: %v", err) + } + remoteNodeID := mustLastInsertID(t, r, "flow-binding-remote-node") + + if err := r.DB().Exec(` + INSERT INTO tunnel(name, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?) + `, "federation-port-forward-custom-name", 1, "tcp", 1, now, now, 1, "", 0).Error; err != nil { + t.Fatalf("insert custom tunnel: %v", err) + } + tunnelID := mustLastInsertID(t, r, "flow-binding-custom-tunnel") + + if err := r.DB().Exec(` + INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, 1, "admin_user", "flow-binding-forward", tunnelID, "1.1.1.1:443", "fifo", 0, 0, now, now, 1, 0).Error; err != nil { + t.Fatalf("insert forward: %v", err) + } + forwardID := mustLastInsertID(t, r, "flow-binding-forward") + + if err := r.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, remoteNodeID, 33001).Error; err != nil { + t.Fatalf("insert forward_port: %v", err) + } + + forwardOut := requestContractEnvelope(t, router, adminToken, "/api/v1/forward/list", nil) + if forwardOut.Code != 0 { + t.Fatalf("forward list failed: code=%d msg=%q", forwardOut.Code, forwardOut.Msg) + } + forwardRows := mustContractSlice(t, forwardOut.Data, "forward list data") + + shareOut := requestContractEnvelope(t, router, adminToken, "/api/v1/federation/share/list", nil) + if shareOut.Code != 0 { + t.Fatalf("share list failed: code=%d msg=%q", shareOut.Code, shareOut.Msg) + } + localShareRows := mustContractSlice(t, shareOut.Data, "share list data") + + remoteUsageOut := requestContractEnvelope(t, router, adminToken, "/api/v1/federation/share/remote-usage/list", nil) + if remoteUsageOut.Code != 0 { + t.Fatalf("remote usage list failed: code=%d msg=%q", remoteUsageOut.Code, remoteUsageOut.Msg) + } + remoteUsageRows := mustContractSlice(t, remoteUsageOut.Data, "remote usage data") + if len(remoteUsageRows) == 0 { + t.Fatalf("expected non-empty remote usage rows") + } + + findForward := func(id int64) map[string]interface{} { + t.Helper() + for _, row := range forwardRows { + m, ok := row.(map[string]interface{}) + if !ok { + continue + } + if contractValueAsInt64(m["id"]) == id { + return m + } + } + t.Fatalf("forward %d not found in /forward/list response", id) + return nil + } + + flowByShare := make(map[int64]int64) + shareIDsByTunnel := make(map[int64]map[int64]struct{}) + + for _, row := range remoteUsageRows { + m, ok := row.(map[string]interface{}) + if !ok { + continue + } + shareID := contractValueAsInt64(m["shareId"]) + currentFlow := contractValueAsInt64(m["currentFlow"]) + if shareID > 0 && currentFlow > 0 { + if currentFlow > flowByShare[shareID] { + flowByShare[shareID] = currentFlow + } + } + + bindings, _ := m["bindings"].([]interface{}) + for _, bindingRaw := range bindings { + binding, ok := bindingRaw.(map[string]interface{}) + if !ok { + continue + } + tunnelIDVal := contractValueAsInt64(binding["tunnelId"]) + chainType := contractValueAsInt64(binding["chainType"]) + if shareID <= 0 || tunnelIDVal <= 0 { + continue + } + if chainType != 1 { + continue + } + setByTunnel, ok := shareIDsByTunnel[tunnelIDVal] + if !ok { + setByTunnel = make(map[int64]struct{}) + shareIDsByTunnel[tunnelIDVal] = setByTunnel + } + setByTunnel[shareID] = struct{}{} + } + } + + for _, row := range localShareRows { + m, ok := row.(map[string]interface{}) + if !ok { + continue + } + shareID := contractValueAsInt64(m["id"]) + currentFlow := contractValueAsInt64(m["currentFlow"]) + if shareID > 0 && currentFlow > 0 { + if currentFlow > flowByShare[shareID] { + flowByShare[shareID] = currentFlow + } + } + } + + targetForward := findForward(forwardID) + parsedByName := contractParseShareIDFromTunnelName(contractValueAsString(targetForward["tunnelName"])) + if parsedByName != 0 { + t.Fatalf("expected custom tunnel name cannot be parsed as Share-*-Port-*, got %d", parsedByName) + } + + resolveShareIDForForward := func(forward map[string]interface{}) int64 { + candidates := make(map[int64]struct{}) + + shareIDFromName := contractParseShareIDFromTunnelName(contractValueAsString(forward["tunnelName"])) + if shareIDFromName > 0 { + candidates[shareIDFromName] = struct{}{} + } + + tunnelIDVal := contractValueAsInt64(forward["tunnelId"]) + if setByTunnel, ok := shareIDsByTunnel[tunnelIDVal]; ok { + for sid := range setByTunnel { + candidates[sid] = struct{}{} + } + } + + var bestShareID int64 + bestFlow := int64(0) + for sid := range candidates { + flow := flowByShare[sid] + if flow > bestFlow { + bestFlow = flow + bestShareID = sid + } + } + return bestShareID + } + + resolvedShareID := resolveShareIDForForward(targetForward) + if resolvedShareID != remoteShareID { + t.Fatalf("expected resolved shareID=%d via tunnel binding, got %d", remoteShareID, resolvedShareID) + } + + forwardCountByShare := make(map[int64]int) + resolvedByForwardID := make(map[int64]int64) + for _, row := range forwardRows { + m, ok := row.(map[string]interface{}) + if !ok { + continue + } + fid := contractValueAsInt64(m["id"]) + sid := resolveShareIDForForward(m) + if sid > 0 { + resolvedByForwardID[fid] = sid + } + if sid > 0 && flowByShare[sid] > 0 { + forwardCountByShare[sid] = forwardCountByShare[sid] + 1 + } + } + + directFlow := contractValueAsInt64(targetForward["inFlow"]) + contractValueAsInt64(targetForward["outFlow"]) + if directFlow != 0 { + t.Fatalf("fixture expectation failed: directFlow should be 0, got %d", directFlow) + } + + shareFlow := flowByShare[resolvedByForwardID[forwardID]] + if shareFlow <= 0 { + t.Fatalf("expected merged share flow > 0 for resolved share %d", resolvedByForwardID[forwardID]) + } + + count := forwardCountByShare[resolvedByForwardID[forwardID]] + if count <= 0 { + count = 1 + } + estimated := shareFlow / int64(count) + if estimated < 1 { + estimated = 1 + } + + displayFlow := estimated + if displayFlow <= 0 { + t.Fatalf("expected displayFlow > 0 after tunnel-binding-based merge, got %d", displayFlow) + } + if displayFlow != remoteShareFlow { + t.Fatalf("expected displayFlow=%d, got %d", remoteShareFlow, displayFlow) + } +} + func requestContractEnvelope(t *testing.T, router http.Handler, token string, path string, body interface{}) response.R { t.Helper() diff --git a/vite-frontend/src/pages/forward.tsx b/vite-frontend/src/pages/forward.tsx index c9c5300..67fb061 100644 --- a/vite-frontend/src/pages/forward.tsx +++ b/vite-frontend/src/pages/forward.tsx @@ -244,17 +244,7 @@ export default function ForwardPage() { const mergeFederationShareFlow = async ( forwardsData: Forward[], ): Promise => { - const shareIds = new Set(); - - forwardsData.forEach((forward) => { - const shareId = parseShareIdFromTunnelName(forward.tunnelName || ""); - - if (shareId) { - shareIds.add(shareId); - } - }); - - if (shareIds.size === 0) { + if (forwardsData.length === 0) { return forwardsData; } @@ -265,6 +255,7 @@ export default function ForwardPage() { ]); const flowByShare = new Map(); + const shareIdsByTunnel = new Map>(); if (usageRes.code === 0 && Array.isArray(usageRes.data)) { usageRes.data.forEach((item: Record) => { @@ -281,6 +272,34 @@ export default function ForwardPage() { flowByShare.set(shareId, Math.max(prev, currentFlow)); } + + if (Number.isFinite(shareId) && shareId > 0) { + const bindings = Array.isArray(item.bindings) + ? (item.bindings as Array>) + : []; + + bindings.forEach((binding) => { + const tunnelId = Number(binding.tunnelId || 0); + const chainType = Number(binding.chainType || 0); + + if (!Number.isFinite(tunnelId) || tunnelId <= 0) { + return; + } + + if (Number.isFinite(chainType) && chainType !== 1) { + return; + } + + let shareSet = shareIdsByTunnel.get(tunnelId); + + if (!shareSet) { + shareSet = new Set(); + shareIdsByTunnel.set(tunnelId, shareSet); + } + + shareSet.add(shareId); + }); + } }); } @@ -306,10 +325,58 @@ export default function ForwardPage() { return forwardsData; } + const resolveShareIdForForward = (forward: Forward): number | null => { + const candidates = new Set(); + const shareIdFromName = parseShareIdFromTunnelName(forward.tunnelName || ""); + + if (shareIdFromName) { + candidates.add(shareIdFromName); + } + + const tunnelId = Number(forward.tunnelId || 0); + const shareSetByTunnel = shareIdsByTunnel.get(tunnelId); + + if (shareSetByTunnel && shareSetByTunnel.size > 0) { + shareSetByTunnel.forEach((shareId) => { + if (Number.isFinite(shareId) && shareId > 0) { + candidates.add(shareId); + } + }); + } + + if (candidates.size === 0) { + return null; + } + + let bestShareId: number | null = null; + let bestFlow = 0; + + candidates.forEach((shareId) => { + const shareFlow = flowByShare.get(shareId) || 0; + + if (shareFlow > bestFlow) { + bestFlow = shareFlow; + bestShareId = shareId; + } + }); + + return bestShareId; + }; + + const resolvedShareByForwardId = new Map(); + + forwardsData.forEach((forward) => { + const shareId = resolveShareIdForForward(forward); + + if (shareId) { + resolvedShareByForwardId.set(forward.id, shareId); + } + }); + const forwardCountByShare = new Map(); forwardsData.forEach((forward) => { - const shareId = parseShareIdFromTunnelName(forward.tunnelName || ""); + const shareId = resolvedShareByForwardId.get(forward.id) || null; if (!shareId || !flowByShare.has(shareId)) { return; @@ -322,7 +389,7 @@ export default function ForwardPage() { }); return forwardsData.map((forward) => { - const shareId = parseShareIdFromTunnelName(forward.tunnelName || ""); + const shareId = resolvedShareByForwardId.get(forward.id) || null; if (!shareId) { return { ...forward, federationShareFlow: undefined };