be.ems/features/event/event.go

package event

import (
	"fmt"
	"net/http"
	"strings"
	"time"

	"be.ems/lib/config"
	"be.ems/lib/core/ctx"
	"be.ems/lib/dborm"
	"be.ems/lib/log"
	"be.ems/lib/services"
	"be.ems/src/framework/utils/parse"
	neService "be.ems/src/modules/network_element/service"
	wsService "be.ems/src/modules/ws/service"
	"github.com/gin-gonic/gin"
)

var (
	// 走Gin
	UriUEEventAMF = "/upload-ue/v1/:eventType"
	// 走Mux
	UriUEEvent       = config.DefaultUriPrefix + "/logManagement/v1/elementType/{elementTypeValue}/objectType/ueEvent"
	CustomUriUEEvent = config.UriPrefix + "/logManagement/v1/elementType/{elementTypeValue}/objectType/ueEvent"
)

type UEEvent struct {
	NeType    string         `json:"neType" xorm:"ne_type"`
	NeName    string         `json:"neName" xorm:"ne_name"`
	RmUID     string         `json:"rmUID" xorm:"rm_uid"`
	Timestamp int64          `json:"timestamp" xorm:"timestamp"`
	EventType string         `json:"eventType" xorm:"event_type"`
	EventJson map[string]any `json:"eventJSON" xorm:"event_json"`
}

// 旧AMF上报处理
func PostUEEventFromAMF(c *gin.Context) {
	log.Info("PostUEEventFromAMF processing... ")

	eventType, ok := c.Params.Get("eventType")
	if !ok || eventType == "" {
		log.Error("eventType is empty")
		services.ResponseNotFound404UriNotExist(c.Writer, c.Request)
		return
	}
	var body map[string]any
	if err := c.ShouldBindBodyWithJSON(&body); err != nil {
		log.Error("Failed to Unmarshal ueEvent:", err)
		services.ResponseInternalServerError500ProcessError(c.Writer, err)
		return
	}

	ueEvent := UEEvent{
		NeType:    "AMF",
		Timestamp: time.Now().Unix(),
		EventType: eventType,
	}
	// 从eventJson中获取rmUID
	if v, ok := body["rmUID"]; ok {
		ueEvent.RmUID = fmt.Sprint(v)
	} else {
		ueEvent.RmUID = "4400HXAMF001"
	}
	if v, ok := body["neName"]; ok {
		ueEvent.NeName = fmt.Sprint(v)
	} else {
		ueEvent.NeName = "AMF_001"
	}

	// 统一格式
	eventJson := map[string]any{"cellID": 0, "gNBID": "", "imsi": "", "onlineNumber": 0, "result": "", "tacID": 0, "timestamp": 0, "time": 0, "type": eventType}
	switch eventType {
	case "auth-result":
		// {"authCode":"200","authMessage":"成功","authTime":"2024-12-07 16:48:37","cellID":"3","gNBID":"1","imsi":"460002082100000","onlineNumber":1,"tacID":"81"}
		if v, ok := body["imsi"]; ok {
			eventJson["imsi"] = fmt.Sprint(v)
		}
		if v, ok := body["cellID"]; ok {
			eventJson["cellID"] = fmt.Sprint(v)
		}
		if v, ok := body["gNBID"]; ok {
			eventJson["gNBID"] = fmt.Sprint(v)
		}
		if v, ok := body["tacID"]; ok {
			eventJson["tacID"] = fmt.Sprint(v)
		}
		if v, ok := body["onlineNumber"]; ok {
			eventJson["onlineNumber"] = parse.Number(v)
		}
		if v, ok := body["authCode"]; ok {
			eventJson["result"] = fmt.Sprint(v)
		}
		if v, ok := body["authTime"]; ok {
			eventJson["timestamp"] = ueEvent.Timestamp
			eventJson["time"] = fmt.Sprint(v)
		}
	case "detach":
		// {"detachResult":0,"detachTime":"2024-12-07 18:00:47","imsi":"460002082100000"}
		if v, ok := body["imsi"]; ok {
			eventJson["imsi"] = fmt.Sprint(v)
		}
		if v, ok := body["detachResult"]; ok {
			if fmt.Sprint(v) == "0" {
				eventJson["result"] = "200"
			} else {
				eventJson["result"] = "500"
			}
		}
		if v, ok := body["detachTime"]; ok {
			eventJson["timestamp"] = ueEvent.Timestamp
			eventJson["time"] = fmt.Sprint(v)
		}
	case "cm-state":
		// {"changeTime":"2024-12-07 17:07:52","imsi":"460002082100000","onlineNumber":1,"status":2}
		if v, ok := body["imsi"]; ok {
			eventJson["imsi"] = fmt.Sprint(v)
		}
		if v, ok := body["onlineNumber"]; ok {
			eventJson["onlineNumber"] = parse.Number(v)
		}
		if v, ok := body["status"]; ok {
			eventJson["result"] = fmt.Sprint(v)
		}
		if v, ok := body["changeTime"]; ok {
			eventJson["timestamp"] = ueEvent.Timestamp
			eventJson["time"] = fmt.Sprint(v)
		}
	}
	ueEvent.EventJson = eventJson

	affected, err := dborm.XormInsertTableOne("ue_event_amf", ueEvent)
	if err != nil && affected <= 0 {
		log.Error("Failed to insert ue_event_amf:", err)
		services.ResponseInternalServerError500ProcessError(c.Writer, err)
		return
	}

	// 发送到匹配的网元
	neInfo := neService.NewNeInfo.SelectNeInfoByRmuid(ueEvent.RmUID)
	if neInfo.RmUID == ueEvent.RmUID {
		// 推送到ws订阅组
		if ueEvent.NeType == "AMF" {
			wsService.NewWSSend.ByGroupID(wsService.GROUP_AMF_UE, ueEvent)
			wsService.NewWSSend.ByGroupID(wsService.GROUP_AMF_UE+"_"+neInfo.NeId, ueEvent)
		}
	}

	services.ResponseStatusOK204NoContent(c.Writer)
}

// UE上报处理
func PostUEEvent(w http.ResponseWriter, r *http.Request) {
	log.Info("PostUEEvent processing... ")

	neType := ctx.GetParam(r, "elementTypeValue")
	var ueEvent UEEvent
	if err := ctx.ShouldBindJSON(r, &ueEvent); err != nil {
		services.ResponseInternalServerError500ProcessError(w, err)
		return
	}

	ueEvent.NeType = strings.ToUpper(neType)
	tableName := fmt.Sprintf("ue_event_%s", strings.ToLower(neType))
	affected, err := dborm.XormInsertTableOne(tableName, ueEvent)
	if err != nil && affected <= 0 {
		log.Error("Failed to insert "+tableName, err)
		services.ResponseInternalServerError500ProcessError(w, err)
		return
	}

	// 发送到匹配的网元
	neInfo := neService.NewNeInfo.SelectNeInfoByRmuid(ueEvent.RmUID)
	if neInfo.RmUID == ueEvent.RmUID {
		// 推送到ws订阅组
		if ueEvent.NeType == "MME" {
			wsService.NewWSSend.ByGroupID(wsService.GROUP_MME_UE, ueEvent)
			wsService.NewWSSend.ByGroupID(wsService.GROUP_MME_UE+"_"+neInfo.NeId, ueEvent)
		}
		if ueEvent.NeType == "AMF" {
			wsService.NewWSSend.ByGroupID(wsService.GROUP_AMF_UE, ueEvent)
			wsService.NewWSSend.ByGroupID(wsService.GROUP_AMF_UE+"_"+neInfo.NeId, ueEvent)
		}
	}

	services.ResponseStatusOK204NoContent(w)
}