测试通过

routes/heath.go → functions/heath.go

@@ -1,20 +1,19 @@
-package routes
+package functions
 
 import (
-	"database/sql"
 	"time"
 
 	"github.com/gin-gonic/gin"
 
 	"git.x2erp.com/qdy/go-base/types"
-	"git.x2erp.com/qdy/go-db/factory"
+	"git.x2erp.com/qdy/go-db/factory/database"
 )
 
 // HealthHandler 返回一个处理函数
-func HealthHandler(db *sql.DB, dbType string) gin.HandlerFunc {
+func HealthHandler(dbFactory *database.DBFactory, dbType string) gin.HandlerFunc {
 	return func(c *gin.Context) {
 
-		err := factory.TestConnection(db, dbType)
+		err := dbFactory.TestConnection(dbType)
 		success := err == nil
 
 		status := "DOWN"

routes/info.go → functions/info.go

@@ -1,18 +1,19 @@
-package routes
+package functions
 
 import (
 	"time"
 
 	"github.com/gin-gonic/gin"
 
+	"git.x2erp.com/qdy/go-base/config"
 	"git.x2erp.com/qdy/go-base/types"
-	"git.x2erp.com/qdy/go-db/factory"
+	"git.x2erp.com/qdy/go-db/factory/database"
 )
 
 // InfoHandler 数据库信息
-func InfoHandler(dbFactory *factory.DBFactory) gin.HandlerFunc {
+func InfoHandler(dbFactory *database.DBFactory) gin.HandlerFunc {
 	return func(c *gin.Context) {
-		config := dbFactory.GetConfig()
+		config := config.GetConfig()
 		drivers := dbFactory.GetAvailableDrivers()
 
 		c.JSON(200, &types.QueryResult{

functions/init_table.go

@@ -0,0 +1,283 @@
+package functions
+
+import (
+	"log"
+
+	"git.x2erp.com/qdy/go-base/types"
+	"git.x2erp.com/qdy/go-db/factory/database"
+	"github.com/gin-gonic/gin"
+)
+
+// 定义表结构
+type tableDDL struct {
+	TableName string
+	SQL       string
+}
+
+// 最简单的使用方式 - 直接在代码中维护
+func ExecuteDDLHandler(dbFactory *database.DBFactory) func(c *gin.Context) {
+	return func(c *gin.Context) {
+
+		tables := getTableDDLs()
+
+		success := []string{}
+		failures := map[string]string{}
+
+		// 直接遍历 tables 切片
+		for _, table := range tables {
+			if err := dbFactory.ExecuteDDL(table.SQL); err != nil {
+				failures[table.TableName] = err.Error()
+				log.Printf("[%s] 失败: %v", table.TableName, err)
+			} else {
+				success = append(success, table.TableName)
+				log.Printf("[%s] 成功", table.TableName)
+			}
+		}
+
+		successState := true
+		if len(failures) > 0 {
+			successState = false // 这里用 = 赋值，不是 :=
+		}
+
+		c.JSON(200, types.QueryResult{
+			Success: successState,
+
+			Data: gin.H{
+				"success":  success,
+				"failures": failures,
+			},
+		})
+	}
+}
+
+// 获取所有表定义
+func getTableDDLs() []tableDDL {
+	return []tableDDL{
+		{
+			TableName: "etl_service",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_service (
+				 	service_id VARCHAR(128) PRIMARY KEY,
+                    service_name VARCHAR(128) NOT NULL,
+                    service_ip VARCHAR(50) NOT NULL,
+                    service_port INTEGER NOT NULL,
+                    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+		{
+			TableName: "etl_agent",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_agent (
+                    agent_id  VARCHAR(50) PRIMARY KEY,
+                    agent_name VARCHAR(128) NOT NULL,
+                    agent_url VARCHAR(512)  NOT NULL,
+					agent_token VARCHAR(512)  NOT NULL,
+					agent_database_type VARCHAR(50)  NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+		{
+			TableName: "etl_table",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_table (
+                    table_id  VARCHAR(50) PRIMARY KEY,
+                    table_tag VARCHAR(512) NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+		{
+			TableName: "etl_table_field",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_table_field (
+                    system_id  VARCHAR(128) PRIMARY KEY,
+					table_id  VARCHAR(50) NOT NULL,
+					field_id  VARCHAR(50) NOT NULL,
+					field_type  VARCHAR(50) NOT NULL,
+					field_default  VARCHAR(50) NOT NULL,
+                    field_tag VARCHAR(512) NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		{
+			TableName: "etl_sql",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_sql (
+					sql_id  VARCHAR(128) PRIMARY KEY,
+					agent_id  VARCHAR(50) NOT NULL,
+					table_id  VARCHAR(50) NOT NULL,
+					where_field_name  VARCHAR(50) NOT NULL,
+					where_field_type  VARCHAR(50) NOT NULL,
+                    sql_tag VARCHAR(512)  NULL,
+            		data_sql  VARCHAR NOT NULL,
+					cunt_sql  VARCHAR NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//参数配置表
+		{
+			TableName: "etl_config",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_config (
+					config_id  VARCHAR(50) PRIMARY KEY,
+					config_type  VARCHAR(50) NOT NULL,
+					config_value  VARCHAR(128) NOT NULL,
+					config_default_value  VARCHAR(128) NOT NULL,
+            		config_tag  VARCHAR(128) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-count
+		{
+			TableName: "etl_task_count",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_count (
+					task_id  VARCHAR(38) PRIMARY KEY,
+					sql_id  VARCHAR(50) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-data
+		{
+			TableName: "etl_task_data",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_data (
+					task_id  VARCHAR(38) PRIMARY KEY,
+					sql_id  VARCHAR(50) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-count_log
+		{
+			TableName: "etl_task_count_log",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_count_log (
+				    log_id  VARCHAR(50) PRIMARY KEY,
+					task_id  VARCHAR(38) NOT NULL,
+					sql_id  VARCHAR(50) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-data_log
+		{
+			TableName: "etl_task_data_log",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_data_log (
+				    log_id  VARCHAR(50) PRIMARY KEY,
+					task_id  VARCHAR(38) NOT NULL,
+					sql_id  VARCHAR(50) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-count_queue
+		{
+			TableName: "etl_task_queue",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_queue (
+				    system_id  VARCHAR(50) PRIMARY KEY,
+					task_id  VARCHAR(38) NOT NULL,
+					queue_id  VARCHAR(38) NOT NULL,
+					sql_id  VARCHAR(50) NOT NULL,
+					task_type  INTEGER NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-count_queue
+		{
+			TableName: "etl_task_queue_log",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_queue_log (
+					system_id  VARCHAR(50) PRIMARY KEY,
+				    task_queue_id  VARCHAR(50) PRIMARY KEY,
+					task_id  VARCHAR(38) NOT NULL,
+					queue_id  VARCHAR(38) NOT NULL,
+					sql_id  VARCHAR(50) NOT NULL,
+					task_type  INTEGER NOT NULL,
+					task_data  VARCHAR NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		//任务表-count_queue
+		{
+			TableName: "etl_task_queue_success_log",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_queue_success_log (
+					system_id  VARCHAR(50) PRIMARY KEY,
+				    task_queue_id  VARCHAR(50) PRIMARY KEY,
+					task_id  VARCHAR(38) NOT NULL,
+					queue_id  VARCHAR(38) NOT NULL,
+					sql_id  VARCHAR(50) NOT NULL,
+					task_type  INTEGER NOT NULL,
+					task_data  VARCHAR NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+		//任务表-count_queue
+		{
+			TableName: "etl_task_queue_failure_log",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_task_queue_failure_log (
+					system_id  VARCHAR(50) PRIMARY KEY,
+				    task_queue_id  VARCHAR(50) PRIMARY KEY,
+					task_id  VARCHAR(38) NOT NULL,
+					queue_id  VARCHAR(38) NOT NULL,
+					sql_id  VARCHAR(50) NOT NULL,
+					task_type  INTEGER NOT NULL,
+					task_data  VARCHAR NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		// 定时日志Timer
+		{
+			TableName: "etl_timer",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_timer (
+				    timer_id  VARCHAR(50) PRIMARY KEY,
+					timer_name  VARCHAR(50) NOT NULL,
+					timer_cron  VARCHAR(50) NOT NULL,
+					timer_tag  VARCHAR(50) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+
+		// 定时日志Timer
+		{
+			TableName: "etl_timer_log",
+			SQL: `
+                CREATE TABLE IF NOT EXISTS etl_timer_log (
+					system_id  VARCHAR(50) PRIMARY KEY,
+				    timer_id  VARCHAR(50) NOT NULL,
+					timer_name  VARCHAR(50) NOT NULL,
+					timer_cron  VARCHAR(50) NOT NULL,
+					timer_tag  VARCHAR(50) NOT NULL,
+			        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );
+            `,
+		},
+		// 添加新表在这里插入新的 TableDDL
+	}
+}

main.go

@@ -1,75 +1,36 @@
 package main
 
 import (
-	"database/sql"
 	"fmt"
 	"log"
 	"net/http"
+	"os"
+	"os/signal"
 	"strings"
+	"syscall"
 	"time"
 
+	"git.x2erp.com/qdy/go-base/config"
 	"git.x2erp.com/qdy/go-base/types"
-	"git.x2erp.com/qdy/go-db/factory"
-	"git.x2erp.com/qdy/go-service-job/auth"
-	"git.x2erp.com/qdy/go-service-job/routes"
+	"git.x2erp.com/qdy/go-db/factory/database"
+	"git.x2erp.com/qdy/go-service-agent/auth"
+	"git.x2erp.com/qdy/go-service-agent/functions"
 	"github.com/gin-gonic/gin"
 )
 
-// 单例实例
-var (
-	dbFactory *factory.DBFactory
-	db        *sql.DB
-)
-
-// initDB 初始化数据库连接（单例）
-func initDB() error {
-	var err error
-
-	// 创建数据库工厂（单例）
-	if dbFactory == nil {
-		dbFactory, err = factory.NewDBFactory()
-		if err != nil {
-			return fmt.Errorf("failed to create DB factory: %v", err)
-		}
-	}
-
-	// 创建数据库连接（单例）
-	if db == nil {
-		db, err = dbFactory.CreateDB()
-		if err != nil {
-			return fmt.Errorf("failed to create database connection: %v", err)
-		}
-	}
-
-	// 测试连接
-	config := dbFactory.GetConfig()
-	if err := factory.TestConnection(db, config.GetDatabase().Type); err != nil {
-		return fmt.Errorf("database connection test failed: %v", err)
-	}
-
-	return nil
-}
-
 func main() {
-	// 1. 初始化数据库（单例）
-	if err := initDB(); err != nil {
-		log.Fatalf("Database initialization failed: %v", err)
-	}
 
-	// 2. 显示基础信息
-	drivers := dbFactory.GetAvailableDrivers()
-	config := dbFactory.GetConfig()
+	cfg := config.GetConfig()
 
-	log.Printf("Service Port: %d", config.GetService().Port)
-	log.Printf("Service IdleTimeout: %d", config.GetService().IdleTimeout)
-	log.Printf("Service ReadTimeout: %d", config.GetService().ReadTimeout)
-	log.Printf("Service WriteTimeout: %d", config.GetService().WriteTimeout)
-	log.Printf("Service TrustedProxies: %s", config.GetService().TrustedProxies)
+	log.Printf("Service Port: %d", cfg.GetService().Port)
+	log.Printf("Service IdleTimeout: %d", cfg.GetService().IdleTimeout)
+	log.Printf("Service ReadTimeout: %d", cfg.GetService().ReadTimeout)
+	log.Printf("Service WriteTimeout: %d", cfg.GetService().WriteTimeout)
+	log.Printf("Service TrustedProxies: %s", cfg.GetService().TrustedProxies)
 
-	log.Printf("Available database drivers: %v", drivers)
-	log.Printf("Using database type: %s", config.GetDatabase().Type)
-	log.Printf("Database host: %s:%d", config.GetDatabase().Host, config.GetDatabase().Port)
-	log.Printf("Database name: %s", config.GetDatabase().Database)
+	log.Printf("Using database type: %s", cfg.GetDatabase().Type)
+	log.Printf("Database host: %s:%d", cfg.GetDatabase().Host, cfg.GetDatabase().Port)
+	log.Printf("Database name: %s", cfg.GetDatabase().Database)
 	log.Println("Database connection test passed!")
 
 	// 3. 启动Gin HTTP服务
@@ -80,14 +41,26 @@ func main() {
 func startHTTPServer() {
 	//建立路由
 	router := gin.Default()
-	config := dbFactory.GetConfig()
-	serviceConfig := config.GetService()
+	cfg := config.GetConfig()
+	serviceConfig := cfg.GetService()
+
+	dbFactory, err := database.GetDBFactory()
+	if err != nil {
+		log.Fatalf("Failed to create DB factory: %v", err)
+	}
+	// 设置优雅关闭
+	setupGracefulShutdown(dbFactory)
+	defer func() {
+		dbFactory.Close()
+		log.Println("Database connection closed")
+	}()
 
 	// 核心路由
-	router.GET("/api/health", routes.HealthHandler(db, config.GetDatabase().Type))
-	router.POST("/api/init/table", auth.AuthMiddleware(), routes.ExecuteDDLHandler(db))
-	//router.POST("/api/query/csv", auth.AuthMiddleware(), withQueryRequest(routes.QueryHandlerCSV(db)))
-	router.GET("/api/info", routes.InfoHandler(dbFactory))
+	router.GET("/api/health", functions.HealthHandler(dbFactory, cfg.GetDatabase().Type))
+	router.POST("/api/query", auth.AuthMiddleware(), withQueryRequest(functions.QueryToJSON(dbFactory)))
+	router.POST("/api/query/csv", auth.AuthMiddleware(), withQueryRequest(functions.QueryToCSV(dbFactory)))
+	router.POST("/api/query/csv/param", auth.AuthMiddleware(), withQueryRequest(functions.QueryPositionalToCSV(dbFactory)))
+	router.GET("/api/info", functions.InfoHandler(dbFactory))
 
 	// 日志输出配置信息
 	log.Printf("Service Port: %d", serviceConfig.Port)
@@ -99,12 +72,6 @@ func startHTTPServer() {
 	// 设置可信代理
 	setupTrustedProxies(router, serviceConfig.TrustedProxies)
 
-	// 启动服务
-	log.Println("POST /api/query  - Execute SQL query to JSON")
-	log.Println("POST /api/query/csv  - Execute SQL query to CSV")
-	log.Println("GET  /api/health - Health check")
-	log.Println("GET  /api/info   - Database info")
-
 	// 创建HTTP服务器配置
 	server := &http.Server{
 		Addr:         fmt.Sprintf(":%d", serviceConfig.Port),
@@ -118,6 +85,7 @@ func startHTTPServer() {
 	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 		log.Fatalf("Failed to start server: %v", err)
 	}
+
 }
 
 // 参数绑定包装器
@@ -172,3 +140,16 @@ func setupTrustedProxiesRouter(router *gin.Engine, trimmedProxies []string) {
 	}
 
 }
+
+func setupGracefulShutdown(dbFactory *database.DBFactory) {
+	signalCh := make(chan os.Signal, 1)
+	signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		<-signalCh
+		log.Println("\nReceived shutdown signal, closing database connection...")
+		dbFactory.Close()
+		log.Println("Database connection closed gracefully")
+		os.Exit(0)
+	}()
+}

routes/init_table.go

@@ -1,83 +0,0 @@
-package routes
-
-import (
-	"database/sql"
-	"log"
-
-	"git.x2erp.com/qdy/go-base/types"
-	"git.x2erp.com/qdy/go-db/factory"
-	"github.com/gin-gonic/gin"
-)
-
-// 定义表结构
-type tableDDL struct {
-	TableName string
-	SQL       string
-}
-
-// 最简单的使用方式 - 直接在代码中维护
-func ExecuteDDLHandler(db *sql.DB) func(c *gin.Context) {
-	return func(c *gin.Context) {
-
-		tables := getTableDDLs()
-
-		success := []string{}
-		failures := map[string]string{}
-
-		// 直接遍历 tables 切片
-		for _, table := range tables {
-			if err := factory.ExecuteDDL(db, table.SQL); err != nil {
-				failures[table.TableName] = err.Error()
-				log.Printf("[%s] 失败: %v", table.TableName, err)
-			} else {
-				success = append(success, table.TableName)
-				log.Printf("[%s] 成功", table.TableName)
-			}
-		}
-
-		successState := true
-		if len(failures) > 0 {
-			successState = false // 这里用 = 赋值，不是 :=
-		}
-
-		c.JSON(200, types.QueryResult{
-			Success: successState,
-
-			Data: gin.H{
-				"success":  success,
-				"failures": failures,
-			},
-		})
-	}
-}
-
-// 获取所有表定义
-func getTableDDLs() []tableDDL {
-	return []tableDDL{
-		{
-			TableName: "etl_service",
-			SQL: `
-                CREATE TABLE IF NOT EXISTS etl_service (
-                    etl_service_name VARCHAR(255) PRIMARY KEY,
-                    etl_ip VARCHAR(50) NOT NULL,
-                    etl_port INTEGER NOT NULL,
-                    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-                );
-            `,
-		},
-		{
-			TableName: "etl_agent",
-			SQL: `
-                CREATE TABLE IF NOT EXISTS etl_agent (
-                    agent_id  VARCHAR(50) PRIMARY KEY,
-                    agent_name VARCHAR(100) NOT NULL,
-                    agent_url VARCHAR(512)  NOT NULL,
-					agent_token VARCHAR(512)  NOT NULL,
-                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-                );
-            `,
-		},
-
-		// 添加新表在这里插入新的 TableDDL
-	}
-}