refactor: 后端包结构重构、去除无用的文件

server/internal/devops/application/db_app.go

@@ -0,0 +1,419 @@
+package application
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"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"
+	"mayfly-go/pkg/utils"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+type Db interface {
+	// 分页获取机器脚本信息列表
+	GetPageList(condition *entity.Db, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	Count(condition *entity.Db) int64
+
+	// 根据条件获取
+	GetDbBy(condition *entity.Db, cols ...string) error
+
+	// 根据id获取
+	GetById(id uint64, cols ...string) *entity.Db
+
+	Save(entity *entity.Db)
+
+	// 删除数据库信息
+	Delete(id uint64)
+
+	// 获取数据库连接实例
+	// @param id 数据库实例id
+	// @param db 数据库
+	GetDbInstance(id uint64, db string) *DbInstance
+}
+
+type dbAppImpl struct {
+	dbRepo    repository.Db
+	dbSqlRepo repository.DbSql
+}
+
+var DbApp Db = &dbAppImpl{
+	dbRepo:    persistence.DbDao,
+	dbSqlRepo: persistence.DbSqlDao,
+}
+
+// 分页获取数据库信息列表
+func (d *dbAppImpl) GetPageList(condition *entity.Db, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return d.dbRepo.GetDbList(condition, pageParam, toEntity, orderBy...)
+}
+
+func (d *dbAppImpl) Count(condition *entity.Db) int64 {
+	return d.dbRepo.Count(condition)
+}
+
+// 根据条件获取
+func (d *dbAppImpl) GetDbBy(condition *entity.Db, cols ...string) error {
+	return d.dbRepo.GetDb(condition, cols...)
+}
+
+// 根据id获取
+func (d *dbAppImpl) GetById(id uint64, cols ...string) *entity.Db {
+	return d.dbRepo.GetById(id, cols...)
+}
+
+func (d *dbAppImpl) Save(dbEntity *entity.Db) {
+	// 默认tcp连接
+	dbEntity.Network = "tcp"
+	// 测试连接
+	if dbEntity.Password != "" {
+		TestConnection(*dbEntity)
+	}
+
+	// 查找是否存在该库
+	oldDb := &entity.Db{Host: dbEntity.Host, Port: dbEntity.Port, EnvId: dbEntity.EnvId}
+	err := d.GetDbBy(oldDb)
+
+	if dbEntity.Id == 0 {
+		biz.NotEmpty(dbEntity.Password, "密码不能为空")
+		biz.IsTrue(err != nil, "该数据库实例已存在")
+		d.dbRepo.Insert(dbEntity)
+		return
+	}
+
+	// 如果存在该库，则校验修改的库是否为该库
+	if err == nil {
+		biz.IsTrue(oldDb.Id == dbEntity.Id, "该数据库实例已存在")
+	}
+
+	dbId := dbEntity.Id
+	old := d.GetById(dbId)
+
+	var oldDbs []interface{}
+	for _, v := range strings.Split(old.Database, " ") {
+		oldDbs = append(oldDbs, v)
+	}
+
+	var newDbs []interface{}
+	for _, v := range strings.Split(dbEntity.Database, " ") {
+		newDbs = append(newDbs, v)
+	}
+	// 比较新旧数据库列表，需要将移除的数据库相关联的信息删除
+	_, delDb, _ := utils.ArrayCompare(newDbs, oldDbs, func(i1, i2 interface{}) bool {
+		return i1.(string) == i2.(string)
+	})
+	for _, v := range delDb {
+		// 先关闭数据库连接
+		CloseDb(dbEntity.Id, v.(string))
+		// 删除该库关联的所有sql记录
+		d.dbSqlRepo.DeleteBy(&entity.DbSql{DbId: dbId, Db: v.(string)})
+	}
+
+	d.dbRepo.Update(dbEntity)
+
+}
+
+func (d *dbAppImpl) Delete(id uint64) {
+	db := d.GetById(id)
+	dbs := strings.Split(db.Database, " ")
+	for _, v := range dbs {
+		// 关闭连接
+		CloseDb(id, v)
+	}
+	d.dbRepo.Delete(id)
+	// 删除该库下用户保存的所有sql信息
+	d.dbSqlRepo.DeleteBy(&entity.DbSql{DbId: id})
+}
+
+var mutex sync.Mutex
+
+func (da *dbAppImpl) GetDbInstance(id uint64, db string) *DbInstance {
+	mutex.Lock()
+	defer mutex.Unlock()
+	// Id不为0，则为需要缓存
+	needCache := id != 0
+	if needCache {
+		load, ok := dbCache.Get(GetDbCacheKey(id, db))
+		if ok {
+			return load.(*DbInstance)
+		}
+	}
+
+	d := da.GetById(id)
+	biz.NotNil(d, "数据库信息不存在")
+	biz.IsTrue(strings.Contains(d.Database, db), "未配置该库的操作权限")
+	global.Log.Infof("连接db: %s:%d/%s", d.Host, d.Port, db)
+
+	// 将数据库替换为要访问的数据库，原本数据库为空格拼接的所有库
+	d.Database = db
+	DB, err := sql.Open(d.Type, getDsn(d))
+	biz.ErrIsNil(err, fmt.Sprintf("Open %s failed, err:%v\n", d.Type, err))
+	perr := DB.Ping()
+	if perr != nil {
+		global.Log.Errorf("连接db失败: %s:%d/%s", d.Host, d.Port, db)
+		panic(biz.NewBizErr(fmt.Sprintf("数据库连接失败: %s", perr.Error())))
+	}
+
+	// 最大连接周期，超过时间的连接就close
+	// DB.SetConnMaxLifetime(100 * time.Second)
+	// 设置最大连接数
+	DB.SetMaxOpenConns(2)
+	// 设置闲置连接数
+	DB.SetMaxIdleConns(1)
+
+	cacheKey := GetDbCacheKey(id, db)
+	dbi := &DbInstance{Id: cacheKey, Type: d.Type, ProjectId: d.ProjectId, db: DB}
+	if needCache {
+		dbCache.Put(cacheKey, dbi)
+	}
+	return dbi
+}
+
+//------------------------------------------------------------------------------
+
+// 客户端连接缓存，30分钟内没有访问则会被关闭, key为数据库实例id:数据库
+var dbCache = cache.NewTimedCache(30*time.Minute, 5*time.Second).
+	WithUpdateAccessTime(true).
+	OnEvicted(func(key interface{}, value interface{}) {
+		global.Log.Info(fmt.Sprintf("删除db连接缓存 id = %s", key))
+		value.(*DbInstance).Close()
+	})
+
+func GetDbCacheKey(dbId uint64, db string) string {
+	return fmt.Sprintf("%d:%s", dbId, db)
+}
+
+func GetDbInstanceByCache(id string) *DbInstance {
+	if load, ok := dbCache.Get(id); ok {
+		return load.(*DbInstance)
+	}
+	return nil
+}
+
+func TestConnection(d entity.Db) {
+	// 验证第一个库是否可以连接即可
+	d.Database = strings.Split(d.Database, " ")[0]
+	DB, err := sql.Open(d.Type, getDsn(&d))
+	biz.ErrIsNil(err, "Open %s failed, err:%v\n", d.Type, err)
+	defer DB.Close()
+	perr := DB.Ping()
+	biz.ErrIsNilAppendErr(perr, "数据库连接失败: %s")
+}
+
+// db实例
+type DbInstance struct {
+	Id        string
+	Type      string
+	ProjectId uint64
+	db        *sql.DB
+}
+
+// 执行查询语句
+// 依次返回 列名数组，结果map，错误
+func (d *DbInstance) SelectData(sql string) ([]string, []map[string]string, error) {
+	sql = strings.Trim(sql, " ")
+	isSelect := strings.HasPrefix(sql, "SELECT") || strings.HasPrefix(sql, "select")
+	isShow := strings.HasPrefix(sql, "show")
+
+	if !isSelect && !isShow {
+		return nil, nil, errors.New("该sql非查询语句")
+	}
+	// 没加limit，则默认限制50条
+	if isSelect && !strings.Contains(sql, "limit") && !strings.Contains(sql, "LIMIT") {
+		sql = sql + " LIMIT 50"
+	}
+
+	rows, err := d.db.Query(sql)
+	if err != nil {
+		return nil, nil, err
+	}
+	// rows对象一定要close掉，如果出错，不关掉则会很迅速的达到设置最大连接数，
+	// 后面的链接过来直接报错或拒绝，实际上也没有起效果
+	defer func() {
+		if rows != nil {
+			rows.Close()
+		}
+	}()
+	cols, _ := rows.Columns()
+	// 这里表示一行填充数据
+	scans := make([]interface{}, len(cols))
+	// 这里表示一行所有列的值，用[]byte表示
+	vals := make([][]byte, len(cols))
+	// 这里scans引用vals，把数据填充到[]byte里
+	for k := range vals {
+		scans[k] = &vals[k]
+	}
+
+	result := make([]map[string]string, 0)
+	// 列名
+	colNames := make([]string, 0)
+	// 是否第一次遍历，列名数组只需第一次遍历时加入
+	isFirst := true
+	for rows.Next() {
+		// 不Scan也会导致等待，该链接实际处于未工作的状态，然后也会导致连接数迅速达到最大
+		err := rows.Scan(scans...)
+		if err != nil {
+			return nil, nil, err
+		}
+		// 每行数据
+		rowData := make(map[string]string)
+		// 把vals中的数据复制到row中
+		for k, v := range vals {
+			key := cols[k]
+			// 如果是密码字段，则脱敏显示
+			if key == "password" {
+				v = []byte("******")
+			}
+			if isFirst {
+				colNames = append(colNames, key)
+			}
+			// 这里把[]byte数据转成string
+			rowData[key] = string(v)
+		}
+		//放入结果集
+		result = append(result, rowData)
+		isFirst = false
+	}
+	return colNames, result, nil
+}
+
+// 执行 update, insert, delete，建表等sql
+// 返回影响条数和错误
+func (d *DbInstance) Exec(sql string) (int64, error) {
+	res, err := d.db.Exec(sql)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
+// 关闭连接
+func (d *DbInstance) Close() {
+	d.db.Close()
+}
+
+// 获取dataSourceName
+func getDsn(d *entity.Db) string {
+	if d.Type == "mysql" {
+		return fmt.Sprintf("%s:%s@%s(%s:%d)/%s?timeout=8s", d.Username, d.Password, d.Network, d.Host, d.Port, d.Database)
+	}
+	return ""
+}
+
+// 删除db缓存并关闭该数据库所有连接
+func CloseDb(dbId uint64, db string) {
+	dbCache.Delete(GetDbCacheKey(dbId, db))
+}
+
+//-----------------------------------元数据-------------------------------------------
+
+const (
+	// mysql 表信息元数据
+	MYSQL_TABLE_MA = `SELECT table_name tableName, engine, table_comment tableComment, 
+	create_time createTime from information_schema.tables
+	WHERE table_schema = (SELECT database())`
+
+	// mysql 表信息
+	MYSQL_TABLE_INFO = `SELECT table_name tableName, table_comment tableComment, table_rows tableRows,
+	data_length dataLength, index_length indexLength, create_time createTime 
+	FROM information_schema.tables 
+    WHERE table_schema = (SELECT database())`
+
+	// mysql 索引信息
+	MYSQL_INDEX_INFO = `SELECT index_name indexName, column_name columnName, index_type indexType,
+	SEQ_IN_INDEX seqInIndex, INDEX_COMMENT indexComment
+	FROM information_schema.STATISTICS 
+    WHERE table_schema = (SELECT database()) AND table_name = '%s'`
+
+	// 默认每次查询列元信息数量
+	DEFAULT_COLUMN_SIZE = 2000
+
+	// mysql 列信息元数据
+	MYSQL_COLOUMN_MA = `SELECT table_name tableName, column_name columnName, column_type columnType,
+	column_comment columnComment, column_key columnKey, extra, is_nullable nullable from information_schema.columns
+	WHERE table_name in (%s) AND table_schema = (SELECT database()) ORDER BY tableName, ordinal_position limit %d, %d`
+
+	// mysql 列信息元数据总数
+	MYSQL_COLOUMN_MA_COUNT = `SELECT COUNT(*) maNum from information_schema.columns
+	WHERE table_name in (%s) AND table_schema = (SELECT database())`
+)
+
+func (d *DbInstance) GetTableMetedatas() []map[string]string {
+	var sql string
+	if d.Type == "mysql" {
+		sql = MYSQL_TABLE_MA
+	}
+	_, res, _ := d.SelectData(sql)
+	return res
+}
+
+func (d *DbInstance) GetColumnMetadatas(tableNames ...string) []map[string]string {
+	var sql, tableName string
+	for i := 0; i < len(tableNames); i++ {
+		if i != 0 {
+			tableName = tableName + ", "
+		}
+		tableName = tableName + "'" + tableNames[i] + "'"
+	}
+
+	var countSqlTmp string
+	var sqlTmp string
+	if d.Type == "mysql" {
+		countSqlTmp = MYSQL_COLOUMN_MA_COUNT
+		sqlTmp = MYSQL_COLOUMN_MA
+	}
+
+	countSql := fmt.Sprintf(countSqlTmp, tableName)
+	_, countRes, _ := d.SelectData(countSql)
+	// 查询出所有列信息总数，手动分页获取所有数据
+	maCount, _ := strconv.Atoi(countRes[0]["maNum"])
+	// 计算需要查询的页数
+	pageNum := maCount / DEFAULT_COLUMN_SIZE
+	if maCount%DEFAULT_COLUMN_SIZE > 0 {
+		pageNum++
+	}
+
+	res := make([]map[string]string, 0)
+	for index := 0; index < pageNum; index++ {
+		sql = fmt.Sprintf(sqlTmp, tableName, index*DEFAULT_COLUMN_SIZE, DEFAULT_COLUMN_SIZE)
+		_, result, err := d.SelectData(sql)
+		biz.ErrIsNilAppendErr(err, "获取数据库列信息失败: %s")
+		res = append(res, result...)
+	}
+	return res
+}
+
+func (d *DbInstance) GetTableInfos() []map[string]string {
+	var sql string
+	if d.Type == "mysql" {
+		sql = MYSQL_TABLE_INFO
+	}
+	_, res, _ := d.SelectData(sql)
+	return res
+}
+
+func (d *DbInstance) GetTableIndex(tableName string) []map[string]string {
+	var sql string
+	if d.Type == "mysql" {
+		sql = fmt.Sprintf(MYSQL_INDEX_INFO, tableName)
+	}
+	_, res, _ := d.SelectData(sql)
+	return res
+}
+
+func (d *DbInstance) GetCreateTableDdl(tableName string) []map[string]string {
+	var sql string
+	if d.Type == "mysql" {
+		sql = fmt.Sprintf("show create table %s ", tableName)
+	}
+	_, res, _ := d.SelectData(sql)
+	return res
+}

server/internal/devops/application/machine_app.go

@@ -0,0 +1,128 @@
+package application
+
+import (
+	"mayfly-go/internal/devops/domain/entity"
+	"mayfly-go/internal/devops/domain/repository"
+	"mayfly-go/internal/devops/infrastructure/machine"
+	"mayfly-go/internal/devops/infrastructure/persistence"
+	"mayfly-go/pkg/biz"
+	"mayfly-go/pkg/model"
+
+	"gorm.io/gorm"
+)
+
+type Machine interface {
+	// 根据条件获取账号信息
+	GetMachine(condition *entity.Machine, cols ...string) error
+
+	Save(entity *entity.Machine)
+
+	// 调整机器状态
+	ChangeStatus(id uint64, status int8)
+
+	Count(condition *entity.Machine) int64
+
+	Delete(id uint64)
+
+	// 根据id获取
+	GetById(id uint64, cols ...string) *entity.Machine
+
+	// 分页获取机器信息列表
+	GetMachineList(condition *entity.Machine, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	// 获取机器连接
+	GetCli(id uint64) *machine.Cli
+}
+
+type machineAppImpl struct {
+	machineRepo repository.Machine
+}
+
+var MachineApp Machine = &machineAppImpl{machineRepo: persistence.MachineDao}
+
+// 分页获取机器信息列表
+func (m *machineAppImpl) GetMachineList(condition *entity.Machine, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return m.machineRepo.GetMachineList(condition, pageParam, toEntity, orderBy...)
+}
+
+func (m *machineAppImpl) Count(condition *entity.Machine) int64 {
+	return m.machineRepo.Count(condition)
+}
+
+// 根据条件获取机器信息
+func (m *machineAppImpl) Save(me *entity.Machine) {
+	// ’修改机器信息且密码不为空‘ or ‘新增’需要测试是否可连接
+	if (me.Id != 0 && me.Password != "") || me.Id == 0 {
+		biz.ErrIsNilAppendErr(machine.TestConn(me), "该机器无法连接: %s")
+	}
+
+	oldMachine := &entity.Machine{Ip: me.Ip, Port: me.Port, Username: me.Username}
+	err := m.GetMachine(oldMachine)
+
+	if me.Id != 0 {
+		// 如果存在该库，则校验修改的库是否为该库
+		if err == nil {
+			biz.IsTrue(oldMachine.Id == me.Id, "该机器信息已存在")
+		}
+		// 关闭连接
+		machine.DeleteCli(me.Id)
+		m.machineRepo.UpdateById(me)
+	} else {
+		biz.IsTrue(err != nil, "该机器信息已存在")
+		// 新增机器，默认启用状态
+		me.Status = entity.MachineStatusEnable
+		m.machineRepo.Create(me)
+	}
+}
+
+func (m *machineAppImpl) ChangeStatus(id uint64, status int8) {
+	if status == entity.MachineStatusDisable {
+		// 关闭连接
+		machine.DeleteCli(id)
+	}
+	machine := new(entity.Machine)
+	machine.Id = id
+	machine.Status = status
+	m.machineRepo.UpdateById(machine)
+}
+
+// 根据条件获取机器信息
+func (m *machineAppImpl) Delete(id uint64) {
+	// 关闭连接
+	machine.DeleteCli(id)
+	model.Tx(
+		func(db *gorm.DB) error {
+			// 删除machine表信息
+			return db.Delete(new(entity.Machine), "id = ?", id).Error
+		},
+		func(db *gorm.DB) error {
+			// 删除machine_file
+			machineFile := &entity.MachineFile{MachineId: id}
+			return db.Where(machineFile).Delete(machineFile).Error
+		},
+		func(db *gorm.DB) error {
+			// 删除machine_script
+			machineScript := &entity.MachineScript{MachineId: id}
+			return db.Where(machineScript).Delete(machineScript).Error
+		},
+	)
+}
+
+// 根据条件获取机器信息
+func (m *machineAppImpl) GetMachine(condition *entity.Machine, cols ...string) error {
+	return m.machineRepo.GetMachine(condition, cols...)
+}
+
+func (m *machineAppImpl) GetById(id uint64, cols ...string) *entity.Machine {
+	return m.machineRepo.GetById(id, cols...)
+}
+
+func (m *machineAppImpl) GetCli(id uint64) *machine.Cli {
+	cli, err := machine.GetCli(id, func(machineId uint64) *entity.Machine {
+		machine := m.GetById(machineId)
+		biz.IsTrue(machine.Status == entity.MachineStatusEnable, "该机器已被停用")
+		return machine
+	})
+	biz.ErrIsNilAppendErr(err, "获取客户端错误: %s")
+	return cli
+}

server/internal/devops/application/machine_file_app.go

@@ -0,0 +1,174 @@
+package application
+
+import (
+	"io"
+	"io/fs"
+	"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/model"
+	"os"
+	"strings"
+
+	"github.com/pkg/sftp"
+)
+
+type MachineFile interface {
+	// 分页获取机器文件信息列表
+	GetPageList(condition *entity.MachineFile, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	// 根据条件获取
+	GetMachineFile(condition *entity.MachineFile, cols ...string) error
+
+	// 根据id获取
+	GetById(id uint64, cols ...string) *entity.MachineFile
+
+	Save(entity *entity.MachineFile)
+
+	Delete(id uint64)
+
+	/**  sftp 相关操作 **/
+
+	// 读取目录
+	ReadDir(fid uint64, path string) []fs.FileInfo
+
+	// 读取文件内容
+	ReadFile(fileId uint64, path string) *sftp.File
+
+	// 写文件
+	WriteFileContent(fileId uint64, path string, content []byte)
+
+	// 文件上传
+	UploadFile(fileId uint64, path, filename string, reader io.Reader)
+
+	// 移除文件
+	RemoveFile(fileId uint64, path string)
+}
+
+type machineFileAppImpl struct {
+	machineFileRepo repository.MachineFile
+	machineRepo     repository.Machine
+}
+
+// 实现类单例
+var MachineFileApp MachineFile = &machineFileAppImpl{
+	machineRepo:     persistence.MachineDao,
+	machineFileRepo: persistence.MachineFileDao,
+}
+
+// 分页获取机器脚本信息列表
+func (m *machineFileAppImpl) GetPageList(condition *entity.MachineFile, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return m.machineFileRepo.GetPageList(condition, pageParam, toEntity, orderBy...)
+}
+
+// 根据条件获取
+func (m *machineFileAppImpl) GetMachineFile(condition *entity.MachineFile, cols ...string) error {
+	return m.machineFileRepo.GetMachineFile(condition, cols...)
+}
+
+// 根据id获取
+func (m *machineFileAppImpl) GetById(id uint64, cols ...string) *entity.MachineFile {
+	return m.machineFileRepo.GetById(id, cols...)
+}
+
+// 保存机器文件配置
+func (m *machineFileAppImpl) Save(entity *entity.MachineFile) {
+	biz.NotNil(m.machineRepo.GetById(entity.MachineId, "Name"), "该机器不存在")
+
+	if entity.Id != 0 {
+		model.UpdateById(entity)
+	} else {
+		model.Insert(entity)
+	}
+}
+
+// 根据id删除
+func (m *machineFileAppImpl) Delete(id uint64) {
+	m.machineFileRepo.Delete(id)
+}
+
+func (m *machineFileAppImpl) ReadDir(fid uint64, path string) []fs.FileInfo {
+	path, machineId := m.checkAndReturnPathMid(fid, path)
+	if !strings.HasSuffix(path, "/") {
+		path = path + "/"
+	}
+
+	sftpCli := m.getSftpCli(machineId)
+	fis, err := sftpCli.ReadDir(path)
+	biz.ErrIsNilAppendErr(err, "读取目录失败: %s")
+	return fis
+}
+
+func (m *machineFileAppImpl) ReadFile(fileId uint64, path string) *sftp.File {
+	path, machineId := m.checkAndReturnPathMid(fileId, path)
+	sftpCli := m.getSftpCli(machineId)
+	// 读取文件内容
+	fc, err := sftpCli.Open(path)
+	biz.ErrIsNilAppendErr(err, "打开文件失败: %s")
+	return fc
+}
+
+// 写文件内容
+func (m *machineFileAppImpl) WriteFileContent(fileId uint64, path string, content []byte) {
+	_, machineId := m.checkAndReturnPathMid(fileId, path)
+
+	sftpCli := m.getSftpCli(machineId)
+	f, err := sftpCli.OpenFile(path, os.O_WRONLY|os.O_TRUNC|os.O_CREATE|os.O_RDWR)
+	biz.ErrIsNilAppendErr(err, "打开文件失败: %s")
+	defer f.Close()
+
+	fi, _ := f.Stat()
+	biz.IsTrue(!fi.IsDir(), "该路径不是文件")
+	f.Write(content)
+}
+
+// 上传文件
+func (m *machineFileAppImpl) UploadFile(fileId uint64, path, filename string, reader io.Reader) {
+	path, machineId := m.checkAndReturnPathMid(fileId, path)
+	if !strings.HasSuffix(path, "/") {
+		path = path + "/"
+	}
+
+	sftpCli := m.getSftpCli(machineId)
+	createfile, err := sftpCli.Create(path + filename)
+	biz.ErrIsNilAppendErr(err, "创建文件失败: %s")
+	defer createfile.Close()
+
+	io.Copy(createfile, reader)
+}
+
+// 删除文件
+func (m *machineFileAppImpl) RemoveFile(fileId uint64, path string) {
+	path, machineId := m.checkAndReturnPathMid(fileId, path)
+
+	sftpCli := m.getSftpCli(machineId)
+	file, err := sftpCli.Open(path)
+	biz.ErrIsNilAppendErr(err, "打开文件失败: %s")
+	fi, _ := file.Stat()
+	if fi.IsDir() {
+		err = sftpCli.RemoveDirectory(path)
+	} else {
+		err = sftpCli.Remove(path)
+	}
+	biz.ErrIsNilAppendErr(err, "删除文件失败: %s")
+}
+
+// 获取sftp client
+func (m *machineFileAppImpl) getSftpCli(machineId uint64) *sftp.Client {
+	return MachineApp.GetCli(machineId).GetSftpCli()
+}
+
+// 校验并返回实际可访问的文件path
+func (m *machineFileAppImpl) checkAndReturnPathMid(fid uint64, inputPath string) (string, uint64) {
+	biz.IsTrue(fid != 0, "文件id不能为空")
+	mf := m.GetById(uint64(fid))
+	biz.NotNil(mf, "文件不存在")
+	if inputPath != "" {
+		// 接口传入的地址需为配置路径的子路径
+		biz.IsTrue(strings.HasPrefix(inputPath, mf.Path), "无权访问该目录或文件")
+		return inputPath, mf.MachineId
+	} else {
+		return mf.Path, mf.MachineId
+	}
+}

server/internal/devops/application/machine_script_app.go

@@ -0,0 +1,70 @@
+package application
+
+import (
+	"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/model"
+)
+
+type MachineScript interface {
+	// 分页获取机器脚本信息列表
+	GetPageList(condition *entity.MachineScript, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	// 根据条件获取
+	GetMachineScript(condition *entity.MachineScript, cols ...string) error
+
+	// 根据id获取
+	GetById(id uint64, cols ...string) *entity.MachineScript
+
+	Save(entity *entity.MachineScript)
+
+	Delete(id uint64)
+}
+
+type machineScriptAppImpl struct {
+	machineScriptRepo repository.MachineScript
+	machineRepo       repository.Machine
+}
+
+const Common_Script_Machine_Id = 9999999
+
+// 实现类单例
+var MachineScriptApp MachineScript = &machineScriptAppImpl{
+	machineRepo:       persistence.MachineDao,
+	machineScriptRepo: persistence.MachineScriptDao}
+
+// 分页获取机器脚本信息列表
+func (m *machineScriptAppImpl) GetPageList(condition *entity.MachineScript, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return m.machineScriptRepo.GetPageList(condition, pageParam, toEntity, orderBy...)
+}
+
+// 根据条件获取
+func (m *machineScriptAppImpl) GetMachineScript(condition *entity.MachineScript, cols ...string) error {
+	return m.machineScriptRepo.GetMachineScript(condition, cols...)
+}
+
+// 根据id获取
+func (m *machineScriptAppImpl) GetById(id uint64, cols ...string) *entity.MachineScript {
+	return m.machineScriptRepo.GetById(id, cols...)
+}
+
+// 保存机器脚本
+func (m *machineScriptAppImpl) Save(entity *entity.MachineScript) {
+	// 如果机器id不为公共脚本id，则校验机器是否存在
+	if machineId := entity.MachineId; machineId != Common_Script_Machine_Id {
+		biz.NotNil(m.machineRepo.GetById(machineId, "Name"), "该机器不存在")
+	}
+
+	if entity.Id != 0 {
+		model.UpdateById(entity)
+	} else {
+		model.Insert(entity)
+	}
+}
+
+// 根据id删除
+func (m *machineScriptAppImpl) Delete(id uint64) {
+	m.machineScriptRepo.Delete(id)
+}

server/internal/devops/application/mongo_app.go

@@ -0,0 +1,133 @@
+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
+}

server/internal/devops/application/project_app.go

@@ -0,0 +1,128 @@
+package application
+
+import (
+	"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/model"
+)
+
+type Project interface {
+	// 分页获取项目信息列表
+	GetPageList(condition *entity.Project, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	Count(condition *entity.Project) int64
+
+	ListProjectByIds(ids []uint64, toEntity interface{}, orderBy ...string)
+
+	SaveProject(project *entity.Project)
+
+	DelProject(id uint64)
+
+	// 根据项目id获取所有该项目下的环境信息列表
+	ListEnvByProjectId(projectId uint64, listPtr interface{})
+
+	// 保存项目环境信息
+	SaveProjectEnv(projectEnv *entity.ProjectEnv)
+
+	// 根据条件获取项目成员信息
+	ListMember(condition *entity.ProjectMember, toEntity interface{}, orderBy ...string)
+
+	SaveProjectMember(pm *entity.ProjectMember)
+
+	// 根据条件获取项目成员信息
+	GetMemberPage(condition *entity.ProjectMember, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	DeleteMember(projectId, accountId uint64)
+
+	// 账号是否有权限访问该项目关联的资源信息
+	CanAccess(accountId, projectId uint64) error
+}
+
+type projectAppImpl struct {
+	projectRepo       repository.Project
+	projectEnvRepo    repository.ProjectEnv
+	projectMemberRepo repository.ProjectMemeber
+	machineRepo       repository.Machine
+	redisRepo         repository.Redis
+	dbRepo            repository.Db
+}
+
+var ProjectApp Project = &projectAppImpl{
+	projectRepo:       persistence.ProjectRepo,
+	projectEnvRepo:    persistence.ProjectEnvRepo,
+	projectMemberRepo: persistence.ProjectMemberRepo,
+	machineRepo:       persistence.MachineDao,
+	redisRepo:         persistence.RedisDao,
+	dbRepo:            persistence.DbDao,
+}
+
+// 分页获取项目信息列表
+func (p *projectAppImpl) GetPageList(condition *entity.Project, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return p.projectRepo.GetPageList(condition, pageParam, toEntity, orderBy...)
+}
+
+func (p *projectAppImpl) Count(condition *entity.Project) int64 {
+	return p.projectRepo.Count(condition)
+}
+
+func (p *projectAppImpl) ListProjectByIds(ids []uint64, toEntity interface{}, orderBy ...string) {
+	p.projectRepo.GetByIdIn(ids, toEntity, orderBy...)
+}
+
+func (p *projectAppImpl) SaveProject(project *entity.Project) {
+	if project.Id == 0 {
+		p.projectRepo.Save(project)
+	} else {
+		// 防止误传导致项目名更新
+		project.Name = ""
+		p.projectRepo.Update(project)
+	}
+}
+
+func (p *projectAppImpl) DelProject(id uint64) {
+	biz.IsTrue(p.machineRepo.Count(&entity.Machine{ProjectId: id}) == 0, "请先删除该项目关联的机器信息")
+	biz.IsTrue(p.redisRepo.Count(&entity.Redis{ProjectId: id}) == 0, "请先删除该项目关联的redis信息")
+	biz.IsTrue(p.dbRepo.Count(&entity.Db{ProjectId: id}) == 0, "请先删除该项目关联的数据库信息")
+	p.projectRepo.Delete(id)
+	p.projectEnvRepo.DeleteEnvs(id)
+	p.projectMemberRepo.DeleteMems(id)
+}
+
+// 根据项目id获取所有该项目下的环境信息列表
+func (p *projectAppImpl) ListEnvByProjectId(projectId uint64, listPtr interface{}) {
+	p.projectEnvRepo.ListEnv(&entity.ProjectEnv{ProjectId: projectId}, listPtr)
+}
+
+// 保存项目环境信息
+func (p *projectAppImpl) SaveProjectEnv(projectEnv *entity.ProjectEnv) {
+	p.projectEnvRepo.Save(projectEnv)
+}
+
+// 根据条件获取项目成员信息
+func (p *projectAppImpl) ListMember(condition *entity.ProjectMember, toEntity interface{}, orderBy ...string) {
+	p.projectMemberRepo.ListMemeber(condition, toEntity, orderBy...)
+}
+
+func (p *projectAppImpl) SaveProjectMember(pm *entity.ProjectMember) {
+	pms := new([]entity.ProjectMember)
+	p.ListMember(&entity.ProjectMember{ProjectId: pm.ProjectId, AccountId: pm.AccountId}, pms)
+	biz.IsTrue(len(*pms) == 0, "该成员已存在")
+	p.projectMemberRepo.Save(pm)
+}
+
+func (p *projectAppImpl) GetMemberPage(condition *entity.ProjectMember, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return p.projectMemberRepo.GetPageList(condition, pageParam, toEntity, orderBy...)
+}
+
+func (p *projectAppImpl) DeleteMember(projectId, accountId uint64) {
+	p.projectMemberRepo.DeleteByPidMid(projectId, accountId)
+}
+
+func (p *projectAppImpl) CanAccess(accountId, projectId uint64) error {
+	if p.projectMemberRepo.IsExist(projectId, accountId) {
+		return nil
+	}
+	return biz.NewBizErr("您无权操作该资源")
+}

server/internal/devops/application/redis_app.go

@@ -0,0 +1,161 @@
+package application
+
+import (
+	"fmt"
+	"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"
+
+	"github.com/go-redis/redis"
+)
+
+type Redis interface {
+	// 分页获取机器脚本信息列表
+	GetPageList(condition *entity.Redis, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult
+
+	Count(condition *entity.Redis) int64
+
+	// 根据id获取
+	GetById(id uint64, cols ...string) *entity.Redis
+
+	// 根据条件获取
+	GetRedisBy(condition *entity.Redis, cols ...string) error
+
+	Save(entity *entity.Redis)
+
+	// 删除数据库信息
+	Delete(id uint64)
+
+	// 获取数据库连接实例
+	GetRedisInstance(id uint64) *RedisInstance
+}
+
+type redisAppImpl struct {
+	redisRepo repository.Redis
+}
+
+var RedisApp Redis = &redisAppImpl{
+	redisRepo: persistence.RedisDao,
+}
+
+// 分页获取机器脚本信息列表
+func (r *redisAppImpl) GetPageList(condition *entity.Redis, pageParam *model.PageParam, toEntity interface{}, orderBy ...string) *model.PageResult {
+	return r.redisRepo.GetRedisList(condition, pageParam, toEntity, orderBy...)
+}
+
+func (r *redisAppImpl) Count(condition *entity.Redis) int64 {
+	return r.redisRepo.Count(condition)
+}
+
+// 根据id获取
+func (r *redisAppImpl) GetById(id uint64, cols ...string) *entity.Redis {
+	return r.redisRepo.GetById(id, cols...)
+}
+
+// 根据条件获取
+func (r *redisAppImpl) GetRedisBy(condition *entity.Redis, cols ...string) error {
+	return r.redisRepo.GetRedis(condition, cols...)
+}
+
+func (r *redisAppImpl) Save(re *entity.Redis) {
+	TestRedisConnection(re)
+
+	// 查找是否存在该库
+	oldRedis := &entity.Redis{Host: re.Host, Db: re.Db}
+	err := r.GetRedisBy(oldRedis)
+
+	if re.Id == 0 {
+		biz.IsTrue(err != nil, "该库已存在")
+		r.redisRepo.Insert(re)
+	} else {
+		// 如果存在该库，则校验修改的库是否为该库
+		if err == nil {
+			biz.IsTrue(oldRedis.Id == re.Id, "该库已存在")
+		}
+		// 先关闭数据库连接
+		CloseRedis(re.Id)
+		r.redisRepo.Update(re)
+	}
+}
+
+// 删除Redis信息
+func (r *redisAppImpl) Delete(id uint64) {
+	CloseRedis(id)
+	r.redisRepo.Delete(id)
+}
+
+// 获取数据库连接实例
+func (r *redisAppImpl) GetRedisInstance(id uint64) *RedisInstance {
+	// Id不为0，则为需要缓存
+	needCache := id != 0
+	if needCache {
+		load, ok := redisCache.Get(id)
+		if ok {
+			return load.(*RedisInstance)
+		}
+	}
+	// 缓存不存在，则回调获取redis信息
+	re := r.GetById(id)
+	biz.NotNil(re, "redis信息不存在")
+	global.Log.Infof("连接redis: %s", re.Host)
+
+	rcli := redis.NewClient(&redis.Options{
+		Addr:     re.Host,
+		Password: re.Password, // no password set
+		DB:       re.Db,       // use default DB
+	})
+	// 测试连接
+	_, e := rcli.Ping().Result()
+	biz.ErrIsNilAppendErr(e, "redis连接失败: %s")
+
+	ri := &RedisInstance{Id: id, ProjectId: re.ProjectId, Cli: rcli}
+	if needCache {
+		redisCache.Put(re.Id, ri)
+	}
+	return ri
+}
+
+//------------------------------------------------------------------------------
+
+// redis客户端连接缓存，30分钟内没有访问则会被关闭
+var redisCache = cache.NewTimedCache(30*time.Minute, 5*time.Second).
+	WithUpdateAccessTime(true).
+	OnEvicted(func(key interface{}, value interface{}) {
+		global.Log.Info(fmt.Sprintf("删除redis连接缓存 id = %d", key))
+		value.(*RedisInstance).Cli.Close()
+	})
+
+// redis实例
+type RedisInstance struct {
+	Id        uint64
+	ProjectId uint64
+	Cli       *redis.Client
+}
+
+// 移除redis连接缓存并关闭redis连接
+func CloseRedis(id uint64) {
+	redisCache.Delete(id)
+}
+
+func TestRedisConnection(re *entity.Redis) {
+	rcli := redis.NewClient(&redis.Options{
+		Addr:     re.Host,
+		Password: re.Password, // no password set
+		DB:       re.Db,       // use default DB
+	})
+	defer rcli.Close()
+	// 测试连接
+	_, e := rcli.Ping().Result()
+	biz.ErrIsNilAppendErr(e, "Redis连接失败: %s")
+}
+
+func (r *RedisInstance) Scan(cursor uint64, match string, count int64) ([]string, uint64) {
+	keys, newcursor, err := r.Cli.Scan(cursor, match, count).Result()
+	biz.ErrIsNilAppendErr(err, "scan失败: %s")
+	return keys, newcursor
+}