mayfly-go/server/internal/devops/application/mongo_app.go

package application

import (
	"context"
	"mayfly-go/internal/devops/domain/entity"
	"mayfly-go/internal/devops/domain/repository"
	"mayfly-go/internal/devops/infrastructure/persistence"
	"mayfly-go/pkg/biz"
	"mayfly-go/pkg/cache"
	"mayfly-go/pkg/global"
	"mayfly-go/pkg/model"
	"time"

	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

type Mongo interface {
	// 分页获取机器脚本信息列表
	GetPageList(condition *entity.Mongo, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult

	Count(condition *entity.Mongo) int64

	// 根据条件获取
	GetBy(condition *entity.Mongo, cols ...string) error

	// 根据id获取
	GetById(id uint64, cols ...string) *entity.Mongo

	Save(entity *entity.Mongo)

	// 删除数据库信息
	Delete(id uint64)

	// 获取mongo连接client
	// @param id mongo id
	GetMongoCli(id uint64) *mongo.Client
}

type mongoAppImpl struct {
	mongoRepo repository.Mongo
}

var MongoApp Mongo = &mongoAppImpl{
	mongoRepo: persistence.MongoDao,
}

// 分页获取数据库信息列表
func (d *mongoAppImpl) GetPageList(condition *entity.Mongo, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
	return d.mongoRepo.GetList(condition, pageParam, toEntity, orderBy...)
}

func (d *mongoAppImpl) Count(condition *entity.Mongo) int64 {
	return d.mongoRepo.Count(condition)
}

// 根据条件获取
func (d *mongoAppImpl) GetBy(condition *entity.Mongo, cols ...string) error {
	return d.mongoRepo.Get(condition, cols...)
}

// 根据id获取
func (d *mongoAppImpl) GetById(id uint64, cols ...string) *entity.Mongo {
	return d.mongoRepo.GetById(id, cols...)
}

func (d *mongoAppImpl) Delete(id uint64) {
	d.mongoRepo.Delete(id)
	DeleteMongoCache(id)
}

func (d *mongoAppImpl) Save(m *entity.Mongo) {
	if m.Id == 0 {
		d.mongoRepo.Insert(m)
	} else {
		// 先关闭连接
		DeleteMongoCache(m.Id)
		d.mongoRepo.Update(m)
	}
}

func (d *mongoAppImpl) GetMongoCli(id uint64) *mongo.Client {
	cli, err := GetMongoCli(id, func(u uint64) string {
		mongo := d.GetById(id)
		biz.NotNil(mongo, "mongo信息不存在")
		return mongo.Uri
	})
	biz.ErrIsNilAppendErr(err, "连接mongo失败: %s")
	return cli
}

// -----------------------------------------------------------

//mongo客户端连接缓存，30分钟内没有访问则会被关闭
var mongoCliCache = cache.NewTimedCache(30*time.Minute, 5*time.Second).
	WithUpdateAccessTime(true).
	OnEvicted(func(key interface{}, value interface{}) {
		global.Log.Info("关闭mongo连接: id = ", key)
		value.(*mongo.Client).Disconnect(context.TODO())
	})

func GetMongoCli(mongoId uint64, getMongoUri func(uint64) string) (*mongo.Client, error) {
	cli, err := mongoCliCache.ComputeIfAbsent(mongoId, func(key interface{}) (interface{}, error) {
		c, err := connect(getMongoUri(mongoId))
		if err != nil {
			return nil, err
		}
		return c, nil
	})

	if cli != nil {
		return cli.(*mongo.Client), err
	}
	return nil, err
}

func DeleteMongoCache(mongoId uint64) {
	mongoCliCache.Delete(mongoId)
}

// 连接mongo，并返回client
func connect(uri string) (*mongo.Client, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri).SetMaxPoolSize(2))
	if err != nil {
		return nil, err
	}
	if err = client.Ping(context.TODO(), nil); err != nil {
		return nil, err
	}
	return client, err
}