crawler/service/db.go

package service

import (
	"crawler/config"
	"strings"
	"time"

	"database/sql"
	"fmt"

	"gorm.io/driver/mysql"
	"gorm.io/gorm"
)

var (
	DB_MOVIEW  DB
	DB_CRAWLER DB
	DB_PLAYR   DB
)

type DB struct {
	SQLDB  *sql.DB
	GORMDB *gorm.DB
	DSN    string
	Err    error
}

// 2025.08.08 비동기 처리를 위한 수정
type logItem struct {
	Action  int
	Query   string
	Comment string
}

var (
	logChannel     = make(chan logItem, 10000)
	flushN         = 200                    // 200개 쌓이면 flush
	flushT         = 250 * time.Millisecond // 250ms마다 flush
	stopChannel    = make(chan struct{})
	stoppedChannel = make(chan struct{})
)

func InitLogWorker(db *sql.DB) {
	go func() {
		defer close(stoppedChannel)
		buf := make([]logItem, 0, flushN)
		ticker := time.NewTicker(flushT)
		defer ticker.Stop()

		for {
			select {
			case item := <-logChannel:
				buf = append(buf, item)
				if len(buf) >= flushN {
					flushLogs(db, buf)
					buf = buf[:0]
				}
			case <-ticker.C:
				if len(buf) > 0 {
					flushLogs(db, buf)
					buf = buf[:0]
				}
			case <-stopChannel:
				for {
					select {
					case it := <-logChannel:
						buf = append(buf, it)
					default:
						if len(buf) > 0 {
							flushLogs(db, buf)
						}
						return
					}
				}
			}
		}
	}()
}

func StopLogWorker() {
	close(stopChannel) // stopChannel를 닫아 더 이상 logItem을 받지 않도록 함
	<-stoppedChannel   // flush 끝날 때까지 대기
}

// 다중 입력 SQL 생성
func flushLogs(db *sql.DB, items []logItem) {
	var sb strings.Builder
	sb.WriteString(`INSERT INTO tb_general_log (action, query, comment, created_at) VALUES `)

	args := make([]any, 0, len(items)*3)

	for i, it := range items {
		if i > 0 {
			sb.WriteByte(',')
		}
		sb.WriteString("(?, ?, ?, NOW())")
		args = append(args, it.Action, it.Query, it.Comment)
	}

	if _, err := db.Exec(sb.String(), args...); err != nil {
		fmt.Println("log flush error:", err)
	}
}

func SetConfig() config.DBAccount {
	var (
		env     = config.Env
		db      = config.DB
		account = config.DBAccount{}
	)

	// DB 환경설정
	switch env.DeveloperEnv {
	case config.LOCAL:
		account = db.Local
	case config.DEV:
		account = db.Dev
	default:
		fmt.Println("DB 설정이 잘못되었습니다.")
	}

	//fmt.Printf("\n%#v\n", account)

	return account
}

// DB 연결
func SetDatabase(dbName string) DB {
	var (
		env     = config.Env
		db      = config.DB
		account = SetConfig()
		err     error
	)

	if dbName != "" {
		account.Name = dbName
	}

	/*
	 * sqlDB
	 */
	dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s", account.User, account.Password, account.Address, account.Name)
	fmt.Printf("\nEnv : %s\n", env.DeveloperEnv)
	fmt.Printf("%s\n", dsn)
	SQLConnd, err := sql.Open(account.Driver, dsn)

	if err != nil || SQLConnd.Ping() != nil {
		fmt.Printf("%s\n", account.Driver)
		fmt.Printf("%s\n", dsn)
		fmt.Printf("%v\n", SQLConnd)
		fmt.Printf("sqlDB에 연결에 실패하였습니다.\n %v", err.Error())
		fmt.Println(err)
		SQLConnd.Close()
	}

	SQLConnd.SetConnMaxLifetime(time.Duration(db.MaxLifetime) * time.Second)
	SQLConnd.SetConnMaxIdleTime(time.Duration(db.MaxIdleTime) * time.Second)
	SQLConnd.SetMaxIdleConns(db.MaxIdleConn)
	SQLConnd.SetMaxOpenConns(db.MaxOpenConn)

	/*
	 * GormDB
	 */
	GORMConnd, err := gorm.Open(mysql.New(mysql.Config{
		Conn: SQLConnd,
	}), &gorm.Config{})
	if err != nil {
		fmt.Printf("GormDB에 연결에 실패하였습니다.\n %v", err.Error())
		fmt.Println(err)
	}

	new := new(DB)
	new.SQLDB = SQLConnd
	new.GORMDB = GORMConnd
	new.DSN = dsn
	new.Err = err

	fmt.Println("Database was opened successfully !!")

	return *new
}

// DB 연결
func (db *DB) Connection(dbName string) DB {
	return SetDatabase(dbName)
}

// gorm 커넥션 연결
func (db *DB) GConnection(dbName string) DB {
	return SetDatabase(dbName)
}

// DB 오류 입력
func (db *DB) SetErrorLog(err error, query string) {
	var (
		conn         = DB_CRAWLER.SQLDB
		sql          = `CALL SP_ERROR_LOG(?, ?, ?);`
		errorMessage = strings.TrimSpace(strings.Split(err.Error(), ":")[1])
		errorNo      = strings.TrimSpace(strings.Replace(strings.Split(err.Error(), ":")[0], "Error ", "", 1))
	)

	conn.Exec(sql, errorNo, errorMessage, query)
	switch errorMessage {
	case "sql: no rows in result set":
	case "":
	default:
		fmt.Printf("DB error msg : %s\n", err.Error())
	}

	fmt.Printf("Exe error query : %s\n", query)
}

/*
 * DB SQL 입력
 * 2025.08.08 비동기 처리를 위한 수정으로 미사용
 */
/*
func (db *DB) SetGeneralLog(action int, query, comment string) {
	var (
		conn = DB_CRAWLER.SQLDB
		sql  = `
			INSERT INTO tb_general_log
			SET
				action = ?,
				query = ?,
				comment = ?,
				created_at = NOW();
		`
	)

	_, err := conn.Exec(sql, action, query, comment)
	if err != nil {
		db.SetErrorLog(err, sql)
	}
}
*/

func (db *DB) SetGeneralLog(action int, query, comment string) {
	if len(query) > 4000 {
		query = query[:4000] + "…(truncated)" // 너무 긴 쿼리는 잘라서 저장 (행 크기/페이지 split 완화)
	}

	select {
	case logChannel <- logItem{action, query, comment}:
	default:
		// 큐가 가득하면 발생
	}
}

// DB 연결 종료
func (db *DB) Close() {
	defer func() {
		if err := recover(); err != nil {
			fmt.Printf("Database recovered message: %s\n", err)
		}
	}()

	if DB_MOVIEW != (DB{}) {
		defer DB_MOVIEW.SQLDB.Close()
		DB_MOVIEW.SQLDB = nil
		DB_MOVIEW.GORMDB = nil
		DB_MOVIEW.DSN = ""
		DB_MOVIEW.Err = nil
	}

	if DB_CRAWLER != (DB{}) {
		defer DB_CRAWLER.SQLDB.Close()
		DB_CRAWLER.SQLDB = nil
		DB_CRAWLER.GORMDB = nil
		DB_CRAWLER.DSN = ""
		DB_CRAWLER.Err = nil
	}

	if DB_PLAYR != (DB{}) {
		defer DB_PLAYR.SQLDB.Close()
		DB_PLAYR.SQLDB = nil
		DB_PLAYR.GORMDB = nil
		DB_PLAYR.DSN = ""
		DB_PLAYR.Err = nil
	}

	fmt.Println("DB closed")
}