refactor: code review

server/internal/db/dbm/dialect_mysql.go

@@ -0,0 +1,183 @@
+package dbm
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	machineapp "mayfly-go/internal/machine/application"
+	"mayfly-go/pkg/errorx"
+	"mayfly-go/pkg/utils/anyx"
+	"net"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+func getMysqlDB(d *DbInfo) (*sql.DB, error) {
+	// SSH Conect
+	if d.SshTunnelMachineId > 0 {
+		sshTunnelMachine, err := machineapp.GetMachineApp().GetSshTunnelMachine(d.SshTunnelMachineId)
+		if err != nil {
+			return nil, err
+		}
+		mysql.RegisterDialContext(d.Network, func(ctx context.Context, addr string) (net.Conn, error) {
+			return sshTunnelMachine.GetDialConn("tcp", addr)
+		})
+	}
+	// 设置dataSourceName  -> 更多参数参考：https://github.com/go-sql-driver/mysql#dsn-data-source-name
+	dsn := fmt.Sprintf("%s:%s@%s(%s:%d)/%s?timeout=8s", d.Username, d.Password, d.Network, d.Host, d.Port, d.Database)
+	if d.Params != "" {
+		dsn = fmt.Sprintf("%s&%s", dsn, d.Params)
+	}
+	return sql.Open(string(d.Type), dsn)
+}
+
+// ---------------------------------- mysql元数据 -----------------------------------
+const (
+	MYSQL_META_FILE      = "metasql/mysql_meta.sql"
+	MYSQL_TABLE_INFO_KEY = "MYSQL_TABLE_INFO"
+	MYSQL_INDEX_INFO_KEY = "MYSQL_INDEX_INFO"
+	MYSQL_COLUMN_MA_KEY  = "MYSQL_COLUMN_MA"
+)
+
+type MysqlDialect struct {
+	dc *DbConn
+}
+
+func (md *MysqlDialect) GetDbNames() ([]string, error) {
+	_, res, err := md.dc.Query("SELECT SCHEMA_NAME AS dbname FROM SCHEMATA")
+	if err != nil {
+		return nil, err
+	}
+
+	databases := make([]string, 0)
+	for _, re := range res {
+		databases = append(databases, anyx.ConvString(re["dbname"]))
+	}
+
+	return databases, nil
+}
+
+// 获取表基础元信息, 如表名等
+func (md *MysqlDialect) GetTables() ([]Table, error) {
+	_, res, err := md.dc.Query(GetLocalSql(MYSQL_META_FILE, MYSQL_TABLE_INFO_KEY))
+	if err != nil {
+		return nil, err
+	}
+
+	tables := make([]Table, 0)
+	for _, re := range res {
+		tables = append(tables, Table{
+			TableName:    re["tableName"].(string),
+			TableComment: anyx.ConvString(re["tableComment"]),
+			CreateTime:   anyx.ConvString(re["createTime"]),
+			TableRows:    anyx.ConvInt(re["tableRows"]),
+			DataLength:   anyx.ConvInt64(re["dataLength"]),
+			IndexLength:  anyx.ConvInt64(re["indexLength"]),
+		})
+	}
+	return tables, nil
+}
+
+// 获取列元信息, 如列名等
+func (md *MysqlDialect) GetColumns(tableNames ...string) ([]Column, error) {
+	tableName := ""
+	for i := 0; i < len(tableNames); i++ {
+		if i != 0 {
+			tableName = tableName + ", "
+		}
+		tableName = tableName + "'" + tableNames[i] + "'"
+	}
+
+	_, res, err := md.dc.Query(fmt.Sprintf(GetLocalSql(MYSQL_META_FILE, MYSQL_COLUMN_MA_KEY), tableName))
+	if err != nil {
+		return nil, err
+	}
+
+	columns := make([]Column, 0)
+	for _, re := range res {
+		columns = append(columns, Column{
+			TableName:     re["tableName"].(string),
+			ColumnName:    re["columnName"].(string),
+			ColumnType:    anyx.ConvString(re["columnType"]),
+			ColumnComment: anyx.ConvString(re["columnComment"]),
+			Nullable:      anyx.ConvString(re["nullable"]),
+			ColumnKey:     anyx.ConvString(re["columnKey"]),
+			ColumnDefault: anyx.ConvString(re["columnDefault"]),
+			NumScale:      anyx.ConvString(re["numScale"]),
+		})
+	}
+	return columns, nil
+}
+
+// 获取表主键字段名，不存在主键标识则默认第一个字段
+func (md *MysqlDialect) GetPrimaryKey(tablename string) (string, error) {
+	columns, err := md.GetColumns(tablename)
+	if err != nil {
+		return "", err
+	}
+	if len(columns) == 0 {
+		return "", errorx.NewBiz("[%s] 表不存在", tablename)
+	}
+
+	for _, v := range columns {
+		if v.ColumnKey == "PRI" {
+			return v.ColumnName, nil
+		}
+	}
+
+	return columns[0].ColumnName, nil
+}
+
+// 获取表索引信息
+func (md *MysqlDialect) GetTableIndex(tableName string) ([]Index, error) {
+	_, res, err := md.dc.Query(fmt.Sprintf(GetLocalSql(MYSQL_META_FILE, MYSQL_INDEX_INFO_KEY), tableName))
+	if err != nil {
+		return nil, err
+	}
+
+	indexs := make([]Index, 0)
+	for _, re := range res {
+		indexs = append(indexs, Index{
+			IndexName:    re["indexName"].(string),
+			ColumnName:   anyx.ConvString(re["columnName"]),
+			IndexType:    anyx.ConvString(re["indexType"]),
+			IndexComment: anyx.ConvString(re["indexComment"]),
+			NonUnique:    anyx.ConvInt(re["nonUnique"]),
+			SeqInIndex:   anyx.ConvInt(re["seqInIndex"]),
+		})
+	}
+	// 把查询结果以索引名分组，索引字段以逗号连接
+	result := make([]Index, 0)
+	key := ""
+	for _, v := range indexs {
+		// 当前的索引名
+		in := v.IndexName
+		if key == in {
+			// 索引字段已根据名称和顺序排序，故取最后一个即可
+			i := len(result) - 1
+			// 同索引字段以逗号连接
+			result[i].ColumnName = result[i].ColumnName + "," + v.ColumnName
+		} else {
+			key = in
+			result = append(result, v)
+		}
+	}
+	return result, nil
+}
+
+// 获取建表ddl
+func (md *MysqlDialect) GetCreateTableDdl(tableName string) (string, error) {
+	_, res, err := md.dc.Query(fmt.Sprintf("show create table `%s` ", tableName))
+	if err != nil {
+		return "", err
+	}
+	return res[0]["Create Table"].(string) + ";", nil
+}
+
+func (md *MysqlDialect) GetTableRecord(tableName string, pageNum, pageSize int) ([]string, []map[string]any, error) {
+	return md.dc.Query(fmt.Sprintf("SELECT * FROM %s LIMIT %d, %d", tableName, (pageNum-1)*pageSize, pageSize))
+}
+
+func (md *MysqlDialect) WalkTableRecord(tableName string, walk func(record map[string]any, columns []string)) error {
+	return md.dc.WalkTableRecord(fmt.Sprintf("SELECT * FROM %s", tableName), walk)
+}