package waf import ( "context" "errors" "fmt" v1 "github.com/go-nunu/nunu-layout-advanced/api/v1" "github.com/go-nunu/nunu-layout-advanced/internal/model" "github.com/go-nunu/nunu-layout-advanced/internal/repository" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" "time" ) type UdpForWardingRepository interface { GetUdpForWarding(ctx context.Context, id int64) (*model.UdpForWarding, error) AddUdpForwarding(ctx context.Context, req *model.UdpForWarding) (int, error) EditUdpForwarding(ctx context.Context, req *model.UdpForWarding) error DeleteUdpForwarding(ctx context.Context, id int64) error GetUdpForwardingWafUdpIdById(ctx context.Context, id int) (int, error) GetUdpForwardingPortCountByHostId(ctx context.Context, hostId int) (int64, error) GetUdpForwardingWafUdpAllIds(ctx context.Context, hostId int) ([]int, error) AddUdpForwardingIps(ctx context.Context, req model.UdpForwardingRule) (primitive.ObjectID, error) EditUdpForwardingIps(ctx context.Context, req model.UdpForwardingRule) error GetUdpForwardingIpsByID(ctx context.Context, udpId int) (*model.UdpForwardingRule, error) DeleteUdpForwardingIpsById(ctx context.Context, udpId int) error // 获取ip数量等于1的ip GetIpCountByIp(ctx context.Context, ips []string) ([]v1.IpCountResult, error) // 获取端口数量 GetPortCount(ctx context.Context, hostId int64, port string) (int64, error) GetUdpAll(ctx context.Context, hostIds []int) ([]int, error) } func NewUdpForWardingRepository( repository *repository.Repository, ) UdpForWardingRepository { return &udpForWardingRepository{ Repository: repository, } } type udpForWardingRepository struct { *repository.Repository } func (r *udpForWardingRepository) GetUdpForWarding(ctx context.Context, id int64) (*model.UdpForWarding, error) { var udpForWarding model.UdpForWarding if err := r.Db.Where("id = ?", id).First(&udpForWarding).Error; err != nil { return nil, err } return &udpForWarding, nil } func (r *udpForWardingRepository) AddUdpForwarding(ctx context.Context, req *model.UdpForWarding) (int, error) { if err := r.Db.WithContext(ctx).Create(req).Error; err != nil { return 0, err } return req.Id, nil } func (r *udpForWardingRepository) EditUdpForwarding(ctx context.Context, req *model.UdpForWarding) error { data := map[string]interface{}{ "proxy" : req.Proxy, } if err := r.Db.Updates(req).Updates(data).Error; err != nil { return err } return nil } func (r *udpForWardingRepository) DeleteUdpForwarding(ctx context.Context, id int64) error { if err := r.Db.Where("id = ?", id).Delete(&model.UdpForWarding{}).Error; err != nil { return err } return nil } func (r *udpForWardingRepository) GetUdpForwardingWafUdpIdById(ctx context.Context, id int) (int, error) { var WafUdpId int if err := r.Db.Model(&model.UdpForWarding{}).Where("id = ?", id).Select("waf_udp_id").Find(&WafUdpId).Error; err != nil { return 0, err } return WafUdpId, nil } func (r *udpForWardingRepository) GetUdpForwardingPortCountByHostId(ctx context.Context, hostId int) (int64, error) { var count int64 if err := r.Db.Model(&model.UdpForWarding{}).Where("host_id = ?", hostId).Count(&count).Error; err != nil { return 0, err } return count, nil } func (r *udpForWardingRepository) GetUdpForwardingWafUdpAllIds(ctx context.Context, hostId int) ([]int, error) { var res []int if err:= r.Db.WithContext(ctx).Model(&model.UdpForWarding{}).Where("host_id = ?", hostId).Select("id").Find(&res).Error; err != nil { return nil, err } return res, nil } // mongodb 插入 func (r *udpForWardingRepository) AddUdpForwardingIps(ctx context.Context, req model.UdpForwardingRule) (primitive.ObjectID, error) { collection := r.MongoDB.Collection("udp_forwarding_rules") req.CreatedAt = time.Now() result, err := collection.InsertOne(ctx, req) if err != nil { return primitive.NilObjectID, fmt.Errorf("插入MongoDB失败: %w", err) } // 返回插入文档的ID return result.InsertedID.(primitive.ObjectID), nil } func (r *udpForWardingRepository) EditUdpForwardingIps(ctx context.Context, req model.UdpForwardingRule) error { collection := r.MongoDB.Collection("udp_forwarding_rules") updateData := bson.M{} if req.Uid != 0 { updateData["uid"] = req.Uid } if req.HostId != 0 { updateData["host_id"] = req.HostId } if req.UdpId != 0 { updateData["udp_id"] = req.UdpId } if len(req.BackendList) > 0 { updateData["backend_list"] = req.BackendList } updateData["cdn_origin_ids"] = req.CdnOriginIds // 始终更新更新时间 updateData["updated_at"] = time.Now() // 如果没有任何字段需要更新,则直接返回 if len(updateData) == 0 { return nil } // 执行更新 update := bson.M{"$set": updateData} err := collection.UpdateOne(ctx, bson.M{"udp_id": req.UdpId}, update) if err != nil { return fmt.Errorf("更新MongoDB文档失败: %w", err) } return nil } func (r *udpForWardingRepository) GetUdpForwardingIpsByID(ctx context.Context, udpId int) (*model.UdpForwardingRule, error) { collection := r.MongoDB.Collection("udp_forwarding_rules") var result model.UdpForwardingRule err := collection.Find(ctx, bson.M{"udp_id": udpId}).One(&result) if err != nil { if errors.Is(err, mongo.ErrNoDocuments) { return nil, nil } return nil, fmt.Errorf("从MongoDB中获取文档失败: %w", err) } return &result, nil } func (r *udpForWardingRepository) DeleteUdpForwardingIpsById(ctx context.Context, udpId int) error { collection := r.MongoDB.Collection("udp_forwarding_rules") err := collection.Remove(ctx, bson.M{"udp_id": udpId}) if err != nil { if errors.Is(err, mongo.ErrNoDocuments) { return fmt.Errorf("记录不存在") } return fmt.Errorf("删除MongoDB文档失败: %w", err) } return nil } // 获取IP数量等于1的IP func (r *udpForWardingRepository) GetIpCountByIp(ctx context.Context, ips []string) ([]v1.IpCountResult, error) { if len(ips) == 0 { return []v1.IpCountResult{}, nil } // 管道逻辑与 TCP 版本完全相同 pipeline := []bson.M{ { "$unwind": "$backend_list", }, { "$addFields": bson.M{ "extracted_ip": bson.M{ "$arrayElemAt": []interface{}{ bson.M{"$split": []string{"$backend_list", ":"}}, 0, }, }, }, }, { "$match": bson.M{ "extracted_ip": bson.M{"$in": ips}, }, }, { "$group": bson.M{ "_id": "$extracted_ip", "count": bson.M{"$sum": 1}, }, }, { "$project": bson.M{ "_id": 0, "ip": "$_id", "count": 1, }, }, } var results []v1.IpCountResult err := r.MongoDB.Collection("udp_forwarding_rules").Aggregate(ctx, pipeline).All(&results) if err != nil { return nil, fmt.Errorf("聚合查询 udp_forwarding_rules 失败: %w", err) } return results, nil } func (r *udpForWardingRepository) GetPortCount(ctx context.Context, hostId int64, port string) (int64, error) { var count int64 if err := r.Db.WithContext(ctx).Model(&model.UdpForWarding{}).Where("host_id = ? AND port = ?", hostId, port).Count(&count).Error; err != nil { return 0, err } return count, nil } func (r *udpForWardingRepository) GetUdpAll(ctx context.Context, hostIds []int) ([]int, error) { var res []int if err:= r.Db.WithContext(ctx).Model(&model.UdpForWarding{}).Where("host_id IN ?", hostIds).Select("cdn_web_id").Scan(&res).Error; err != nil { return nil, err } return res, nil }