opsx-vpn/backend/internal/sessionmgr/sessionmgr.go

// Package sessionmgr 管理 VPN 在线会话：同步运行时 peers → VPNSession 表，
// 提供在线列表查询和踢下线能力。跨协议统一（WireGuard peers / OpenVPN clients）。
package sessionmgr

import (
	"fmt"
	"time"

	"vpn-platform/backend/internal/core"
	"vpn-platform/backend/internal/identityprovider"
	"vpn-platform/backend/internal/model"
	"vpn-platform/backend/internal/store"
)

// ClientInfo 协议层上报的在线客户端信息（统一格式）
type ClientInfo struct {
	Username       string // 用户名或证书 CN
	RealAddress    string // 客户端公网地址 IP:Port
	VirtualAddress string // 分配的 VPN IP
	RxBytes        int64
	TxBytes        int64
}

// SyncSessions 将运行时客户端列表同步到 VPNSession 表。
// 新连接 → 创建 active 记录 + 写 ConnectionLog(connect)；
// 已不在列表中的 active 会话 → 标记 disconnected + 写 ConnectionLog(disconnect)。
func SyncSessions(tunnelID uint, protocol string, clients []ClientInfo) {
	if store.DB == nil {
		return
	}
	now := time.Now()
	subjects := make([]string, 0, len(clients))
	for _, client := range clients {
		if protocol == "openvpn" && client.Username == "" {
			continue // WebAuth reports UNDEF until the browser flow supplies a subject.
		}
		subjects = append(subjects, client.Username)
	}
	var tunnel model.Tunnel
	store.DB.First(&tunnel, tunnelID)
	names := identityprovider.ProfileNames(&tunnel, subjects)

	// 1. 收集当前在线的标识 key（username|realAddress）
	currentKeys := map[string]bool{}
	for _, c := range clients {
		if protocol == "openvpn" && c.Username == "" {
			continue // Never audit a pre-auth socket as a connected VPN identity.
		}
		key := c.Username + "|" + c.RealAddress
		currentKeys[key] = true

		// 查是否已有 active 会话
		var existing model.VPNSession
		err := store.DB.Where("tunnel_id = ? AND username = ? AND real_address = ? AND status = ?",
			tunnelID, c.Username, c.RealAddress, "active").First(&existing).Error

		if err == nil {
			// 已有会话 → 更新流量
			store.DB.Model(&existing).Updates(map[string]interface{}{
				"rx_bytes": c.RxBytes, "tx_bytes": c.TxBytes,
				"display_name": names[c.Username],
			})
		} else {
			// 新连接 → 创建会话 + 审计日志
			session := model.VPNSession{
				DisplayName:    names[c.Username],
				TunnelID:       tunnelID,
				Username:       c.Username,
				Protocol:       protocol,
				RealAddress:    c.RealAddress,
				VirtualAddress: c.VirtualAddress,
				ConnectedAt:    now,
				RxBytes:        c.RxBytes,
				TxBytes:        c.TxBytes,
				Status:         "active",
			}
			// 尝试关联 VPNUser
			var vu model.VPNUser
			if store.DB.Where("tunnel_id = ? AND username = ?", tunnelID, c.Username).First(&vu).Error == nil {
				session.VPNUserID = vu.ID
			}
			store.DB.Create(&session)
			LogConnection(tunnelID, c.Username, "connect", c.RealAddress, "")
		}
	}

	// 2. 标记已不在列表中的 active 会话为 disconnected
	var activeSessions []model.VPNSession
	store.DB.Where("tunnel_id = ? AND status = ?", tunnelID, "active").Find(&activeSessions)
	for _, s := range activeSessions {
		key := s.Username + "|" + s.RealAddress
		if !currentKeys[key] {
			store.DB.Model(&s).Updates(map[string]interface{}{
				"status":          "disconnected",
				"disconnected_at": now,
			})
			duration := now.Sub(s.ConnectedAt).Round(time.Second)
			if s.Username != "" {
				LogConnection(tunnelID, s.Username, "disconnect", s.RealAddress,
					fmt.Sprintf("时长 %s, ↓%s ↑%s", duration, fmtBytes(s.RxBytes), fmtBytes(s.TxBytes)))
			}
		}
	}

}

// SyncFromPeers 从 core.Status.Peers 同步（WireGuard 用）
func SyncFromPeers(tunnelID uint, protocol string, peers []core.PeerStatus) {
	clients := make([]ClientInfo, 0, len(peers))
	for _, p := range peers {
		if !p.Connected {
			continue
		}
		clients = append(clients, ClientInfo{
			Username:    p.Name, // WG peer 名称
			RealAddress: p.Endpoint,
			RxBytes:     p.RxBytes,
			TxBytes:     p.TxBytes,
		})
	}
	SyncSessions(tunnelID, protocol, clients)
}

// ListActive 查询所有在线会话
func ListActive() []model.VPNSession {
	var sessions []model.VPNSession
	store.DB.Where("status = ?", "active").Order("connected_at desc").Find(&sessions)
	return sessions
}

// CountActive 统计在线会话数
func CountActive() int64 {
	var count int64
	store.DB.Model(&model.VPNSession{}).Where("status = ?", "active").Count(&count)
	return count
}

// LogConnection 写连接审计日志
func LogConnection(tunnelID uint, username, event, ip, detail string) {
	if store.DB == nil {
		return
	}
	var tunnel model.Tunnel
	store.DB.First(&tunnel, tunnelID)
	names := identityprovider.ProfileNames(&tunnel, []string{username})
	store.DB.Create(&model.ConnectionLog{
		DisplayName: names[username],
		TunnelID:    tunnelID,
		Username:    username,
		Event:       event,
		IPAddress:   ip,
		Detail:      detail,
	})
}

// ListLogs 查询连接审计日志
func ListLogs(limit int) []model.ConnectionLog {
	var logs []model.ConnectionLog
	if limit <= 0 {
		limit = 200
	}
	store.DB.Order("id desc").Limit(limit).Find(&logs)
	return logs
}

func fmtBytes(n int64) string {
	if n < 1024 {
		return fmt.Sprintf("%dB", n)
	}
	if n < 1024*1024 {
		return fmt.Sprintf("%.1fKB", float64(n)/1024)
	}
	if n < 1024*1024*1024 {
		return fmt.Sprintf("%.1fMB", float64(n)/(1024*1024))
	}
	return fmt.Sprintf("%.1fGB", float64(n)/(1024*1024*1024))
}