database/wrapper_db_client.go

// Package database ...
//
// Description : define ...
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 2024-08-20 17:36
package database

import (
	"context"
	"errors"
	"fmt"
	"git.zhangdeman.cn/zhangdeman/consts"
	"git.zhangdeman.cn/zhangdeman/database/define"
	"git.zhangdeman.cn/zhangdeman/logger/wrapper"
	"go.uber.org/zap"
	"gorm.io/driver/mysql"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	gormLogger "gorm.io/gorm/logger"
	"strings"
	"sync"
	"time"
)

// DBClient 包装日志实例
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 3:09 PM 2021/12/24
type DBClient struct {
	DbFlag                    string                            // 数据库标识
	LoggerInstance            *zap.Logger                       // 日志实例
	master                    *gorm.DB                          // 主库
	slave                     *gorm.DB                          // 从库
	ExtraFieldList            []string                          // 提取的字段
	Cfg                       define.Driver                     // 数据库配置
	cacheTableStructureConfig *define.CacheTableStructureConfig // 缓存配置
	lock                      *sync.RWMutex                     // 操作锁
	tableStructureCache       map[string][]*define.ColumnConfig // 表结构缓存
}

// Init 初始化客户端
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 17:44 2024/8/20
func (dc *DBClient) Init(databaseConfig *define.Database, cacheTableStructureConfig *define.CacheTableStructureConfig) error {
	dc.lock = &sync.RWMutex{}
	dc.cacheTableStructureConfig = cacheTableStructureConfig
	var err error
	if dc.master, err = dc.GetDatabaseClient(databaseConfig.Master, dc.LoggerInstance); nil != err {
		return err
	}
	if dc.slave, err = dc.GetDatabaseClient(databaseConfig.Slave, dc.LoggerInstance); nil != err {
		return err
	}
	if err = dc.syncDbTableStructure(false); nil != err { // 同步缓存表结构
		return err
	}
	// 启动异步任务
	go func() {
		if !dc.CacheDataTableStructureConfig().Enable {
			// 未启用
			return
		}
		for {
			select {
			case <-time.After(time.Second * time.Duration(dc.CacheDataTableStructureConfig().SyncTimeInterval)):
				_ = dc.syncDbTableStructure(true)
			}
		}
	}()
	return nil
}

// GetMaster 获取主库连接
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 3:28 PM 2021/12/24
func (dc *DBClient) GetMaster(ctx context.Context) *gorm.DB {
	session := dc.master.Session(&gorm.Session{})
	session.Logger = dc.getLogger(ctx, session, "master")
	return session
}

// GetSlave 获取从库链接
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 3:29 PM 2021/12/24
func (dc *DBClient) GetSlave(ctx context.Context) *gorm.DB {
	session := dc.slave.Session(&gorm.Session{})
	session.Logger = dc.getLogger(ctx, session, "slave")
	return session
}

// getLogger 获取日志实例
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 3:45 PM 2021/12/24
func (dc *DBClient) getLogger(ctx context.Context, dbClient *gorm.DB, node string) gormLogger.Interface {
	return wrapper.NewGormLoggerWithInstance(ctx, dbClient, dc.LoggerInstance, dc.DbFlag+"|"+node, dc.ExtraFieldList)
}

// GetDatabaseClient 获取数据库连接
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 18:41 2022/6/11
func (dc *DBClient) GetDatabaseClient(conf *define.Driver, logInstance *zap.Logger) (*gorm.DB, error) {
	var (
		instance *gorm.DB
		err      error
	)

	if nil == logInstance {
		logInstance = dc.LoggerInstance
	}

	if conf.DBType == consts.DatabaseDriverMysql {
		if instance, err = gorm.Open(mysql.Open(dc.buildConnectionDSN(conf)), &gorm.Config{}); nil != err {
			return nil, err
		}
	} else if conf.DBType == consts.DatabaseDriverSqlite3 {
		if instance, err = gorm.Open(sqlite.Open(dc.buildConnectionDSN(conf)), &gorm.Config{}); nil != err {
			return nil, err
		}
	} else {
		return nil, fmt.Errorf("%v : db driver is not support", conf.DBType)
	}

	instance.Logger = wrapper.NewGormLoggerWithInstance(nil, instance, logInstance, "", nil)

	return instance, nil
}

// buildConnectionDSN 构建连接信息
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 18:42 2022/6/11
func (dc *DBClient) buildConnectionDSN(conf *define.Driver) string {
	if conf.DBType == consts.DatabaseDriverSqlite3 {
		// 兼容sqlite3
		return conf.Host
	}
	return fmt.Sprintf(
		"%s:%s@tcp(%s:%d)/%s?charset=%s&parseTime=True&loc=%s",
		conf.Username,
		conf.Password,
		conf.Host,
		conf.Port,
		conf.Database,
		conf.Charset,
		conf.Timezone,
	)
}

// CacheDataTableStructureConfig ...
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:03 2024/8/21
func (dc *DBClient) CacheDataTableStructureConfig() *define.CacheTableStructureConfig {
	if nil == dc.cacheTableStructureConfig {
		return &define.CacheTableStructureConfig{Enable: false, SyncTimeInterval: 3600}
	}
	if dc.cacheTableStructureConfig.SyncTimeInterval <= 0 {
		dc.cacheTableStructureConfig.SyncTimeInterval = 3600
	}
	return dc.cacheTableStructureConfig
}

// GetTableFieldList 获取表结构配置
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:07 2024/8/21
func (dc *DBClient) GetTableFieldList(tableName string) ([]*define.ColumnConfig, error) {
	if !dc.CacheDataTableStructureConfig().Enable {
		// 未启用缓存, 返回空list
		return make([]*define.ColumnConfig, 0), nil
	}
	dc.lock.RLock()
	defer dc.lock.RUnlock()
	if _, exist := dc.tableStructureCache[tableName]; !exist {
		return nil, errors.New(tableName + " : cache result not found")
	}
	return dc.tableStructureCache[tableName], nil
}

// syncDbTableStructure 缓存表结构
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:17 2024/8/21
func (dc *DBClient) syncDbTableStructure(ignoreErr bool) error {
	if !dc.CacheDataTableStructureConfig().Enable {
		// 自动同步不可用
		return nil
	}
	var (
		err       error
		tableList []string
	)

	c := dc.GetMaster(context.Background())
	systemDao := &SystemDao{}
	if tableList, err = systemDao.GetTableList(c); nil != err {
		return err
	}
	tableStructCache := make(map[string][]*define.ColumnConfig)
	for _, itemTableName := range tableList {
		fieldList, loadTableErr := systemDao.GetTableInfo(c, dc.Cfg.Database, itemTableName)
		if nil != loadTableErr {
			if ignoreErr {
				continue
			}
			return loadTableErr
		}
		tableStructCache[itemTableName] = make([]*define.ColumnConfig, 0)
		for _, itemColumn := range fieldList {
			fieldType := "string"
			if strings.Contains(itemColumn.ColumnType, "int") {
				if strings.Contains(itemColumn.ColumnType, "unsigned") {
					fieldType = "uint"
				} else {
					fieldType = "int"
				}
			}
			tableStructCache[itemTableName] = append(tableStructCache[itemTableName], &define.ColumnConfig{
				Column: itemColumn.ColumnName,
				Alias:  itemColumn.ColumnName,
				Type:   fieldType,
			})
		}
	}
	// 更新缓存结果
	dc.lock.Lock()
	defer dc.lock.Unlock()
	dc.tableStructureCache = tableStructCache
	return nil
}

// SetTableStructure 设置表结构, 一旦调用人工设置, 则将终止自动同步
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 17:06 2024/8/23
func (dc *DBClient) SetTableStructure(tableConfigTable map[string][]*define.ColumnConfig) {
	if nil != dc.cacheTableStructureConfig {
		// 关闭自动同步
		dc.cacheTableStructureConfig.Enable = false
	}
	dc.lock.Lock()
	if nil == dc.tableStructureCache {
		dc.tableStructureCache = make(map[string][]*define.ColumnConfig)
	}
	for table, columnConfig := range tableConfigTable {
		dc.tableStructureCache[table] = columnConfig
	}
	dc.lock.Unlock()
}