fix(backend): track traffic stats for federation port-forward tunnels

Federation mode panel peering with port-forward tunnel type (tunnelType=1)
was not updating peer_share.current_flow because regular forward traffic
uses different service name pattern than federation tunnel traffic.

Added parsePeerShareIDFromFederationTunnelName() to extract shareID from
tunnel names matching 'Share-{shareId}-Port-{port}' pattern, and
processPeerShareFlowFromForward() to update peer_share flow stats when
the forward belongs to a federation port-forward tunnel.
This commit is contained in:
sagitchu
2026-02-24 20:25:04 +08:00
parent ee6bc8c50e
commit a97484cd9b
2 changed files with 125 additions and 0 deletions
@@ -40,6 +40,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, item)
if userTunnelID > 0 {
h.enforceFlowPolicies(userID, userTunnelID)
@@ -87,6 +88,23 @@ func parsePeerShareRuntimeServiceID(serviceName string) (int64, bool) {
return runtimeID, true
}
func parsePeerShareIDFromFederationTunnelName(tunnelName string) (int64, bool) {
tunnelName = strings.TrimSpace(tunnelName)
if !strings.HasPrefix(tunnelName, "Share-") {
return 0, false
}
raw := strings.TrimPrefix(tunnelName, "Share-")
idx := strings.Index(raw, "-Port-")
if idx <= 0 {
return 0, false
}
shareID, err := strconv.ParseInt(raw[:idx], 10, 64)
if err != nil || shareID <= 0 {
return 0, false
}
return shareID, true
}
func (h *Handler) processPeerShareFlow(runtimeID int64, item flowItem) {
if h == nil || h.repo == nil || runtimeID <= 0 {
return
@@ -113,6 +131,44 @@ func (h *Handler) processPeerShareFlow(runtimeID int64, item flowItem) {
h.enforcePeerShareFlowLimit(share.ID)
}
func (h *Handler) processPeerShareFlowFromForward(forwardID int64, item flowItem) {
if h == nil || h.repo == nil || forwardID <= 0 {
return
}
forward, err := h.getForwardRecord(forwardID)
if err != nil || forward == nil {
return
}
tunnel, err := h.getTunnelRecord(forward.TunnelID)
if err != nil || tunnel == nil {
return
}
tunnelName, err := h.repo.GetTunnelName(forward.TunnelID)
if err != nil {
return
}
shareID, ok := parsePeerShareIDFromFederationTunnelName(tunnelName)
if !ok {
return
}
delta := item.D + item.U
if delta <= 0 {
return
}
_ = h.repo.AddPeerShareCurrentFlow(shareID, delta)
share, err := h.repo.GetPeerShare(shareID)
if err != nil || share == nil {
return
}
if !isPeerShareFlowExceeded(share) {
return
}
h.enforcePeerShareFlowLimit(share.ID)
}
func (h *Handler) enforcePeerShareFlowLimit(shareID int64) {
if h == nil || h.repo == nil || shareID <= 0 {
return