feat: rpc.

database/clickhouse/batch.go

@@ -167,14 +167,15 @@ func appendStructToBatch(batch driverV2.Batch, obj interface{}, columns []string
 			field := t.Field(j)
 
 			// 检查ch标签
-			if tag := field.Tag.Get("ch"); tag == col {
+			if tag := field.Tag.Get("ch"); strings.TrimSpace(tag) == col {
 				values[i] = v.Field(j).Interface()
 				found = true
 				break
 			}
 
 			// 检查json标签
-			if tag := field.Tag.Get("json"); tag == col {
+			jsonTags := strings.Split(field.Tag.Get("json"), ",")
+			if len(jsonTags) > 0 && strings.TrimSpace(jsonTags[0]) == col {
 				values[i] = v.Field(j).Interface()
 				found = true
 				break

database/clickhouse/client.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"crypto/tls"
 	"database/sql"
+	"fmt"
 	"net/url"
+	"reflect"
+	"strings"
 
 	clickhouseV2 "github.com/ClickHouse/clickhouse-go/v2"
 	driverV2 "github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 
 	"github.com/go-kratos/kratos/v2/log"
-
 	conf "github.com/tx7do/kratos-bootstrap/api/gen/go/conf/v1"
 	"github.com/tx7do/kratos-bootstrap/utils"
 )
@@ -229,7 +231,7 @@ func (c *Client) CheckConnection(ctx context.Context) error {
 }
 
 // Query 执行查询并返回结果
-func (c *Client) Query(ctx context.Context, creator Creator, results *[]any, query string, args ...interface{}) error {
+func (c *Client) Query(ctx context.Context, creator Creator, results *[]any, query string, args ...any) error {
 	if c.conn == nil {
 		c.log.Error("clickhouse client is not initialized")
 		return ErrClientNotInitialized
@@ -269,7 +271,7 @@ func (c *Client) Query(ctx context.Context, creator Creator, results *[]any, que
 }
 
 // QueryRow 执行查询并返回单行结果
-func (c *Client) QueryRow(ctx context.Context, dest any, query string, args ...interface{}) error {
+func (c *Client) QueryRow(ctx context.Context, dest any, query string, args ...any) error {
 	row := c.conn.QueryRow(ctx, query, args...)
 	if row == nil {
 		c.log.Error("query row returned nil")
@@ -285,7 +287,7 @@ func (c *Client) QueryRow(ctx context.Context, dest any, query string, args ...i
 }
 
 // Select 封装 SELECT 子句
-func (c *Client) Select(ctx context.Context, dest any, query string, args ...interface{}) error {
+func (c *Client) Select(ctx context.Context, dest any, query string, args ...any) error {
 	if c.conn == nil {
 		c.log.Error("clickhouse client is not initialized")
 		return ErrClientNotInitialized
@@ -301,7 +303,7 @@ func (c *Client) Select(ctx context.Context, dest any, query string, args ...int
 }
 
 // Exec 执行非查询语句
-func (c *Client) Exec(ctx context.Context, query string, args ...interface{}) error {
+func (c *Client) Exec(ctx context.Context, query string, args ...any) error {
 	if c.conn == nil {
 		c.log.Error("clickhouse client is not initialized")
 		return ErrClientNotInitialized
@@ -315,23 +317,268 @@ func (c *Client) Exec(ctx context.Context, query string, args ...interface{}) er
 	return nil
 }
 
-// AsyncInsert 异步插入数据
-func (c *Client) AsyncInsert(ctx context.Context, query string, wait bool, args ...interface{}) error {
+func (c *Client) prepareInsertData(data any) (string, string, []any, error) {
+	val := reflect.ValueOf(data)
+	if val.Kind() != reflect.Ptr || val.IsNil() {
+		return "", "", nil, fmt.Errorf("data must be a non-nil pointer")
+	}
+
+	val = val.Elem()
+	typ := val.Type()
+
+	columns := make([]string, 0, typ.NumField())
+	placeholders := make([]string, 0, typ.NumField())
+	values := make([]any, 0, typ.NumField())
+
+	values = structToValueArray(data)
+
+	for i := 0; i < typ.NumField(); i++ {
+		field := typ.Field(i)
+
+		// 优先获取 `ch` 标签，其次获取 `json` 标签，最后使用字段名
+		columnName := field.Tag.Get("ch")
+		if columnName == "" {
+			jsonTag := field.Tag.Get("json")
+			if jsonTag != "" {
+				tags := strings.Split(jsonTag, ",") // 只取逗号前的部分
+				if len(tags) > 0 {
+					columnName = tags[0]
+				}
+			}
+		}
+		if columnName == "" {
+			columnName = field.Name
+		}
+		//columnName = strings.TrimSpace(columnName)
+
+		columns = append(columns, columnName)
+		placeholders = append(placeholders, "?")
+	}
+
+	return strings.Join(columns, ", "), strings.Join(placeholders, ", "), values, nil
+}
+
+// Insert 插入数据到指定表
+func (c *Client) Insert(ctx context.Context, tableName string, in any) error {
 	if c.conn == nil {
 		c.log.Error("clickhouse client is not initialized")
 		return ErrClientNotInitialized
 	}
 
-	if err := c.conn.AsyncInsert(ctx, query, wait, args...); err != nil {
-		c.log.Errorf("exec failed: %v", err)
+	columns, placeholders, values, err := c.prepareInsertData(in)
+	if err != nil {
+		c.log.Errorf("prepare insert in failed: %v", err)
+		return ErrPrepareInsertDataFailed
+	}
+
+	// 构造 SQL 语句
+	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
+		tableName,
+		columns,
+		placeholders,
+	)
+
+	// 执行插入操作
+	if err = c.conn.Exec(ctx, query, values...); err != nil {
+		c.log.Errorf("insert failed: %v", err)
+		return ErrInsertFailed
+	}
+
+	return nil
+}
+
+func (c *Client) InsertMany(ctx context.Context, tableName string, data []any) error {
+	if c.conn == nil {
+		c.log.Error("clickhouse client is not initialized")
+		return ErrClientNotInitialized
+	}
+
+	if len(data) == 0 {
+		c.log.Error("data slice is empty")
+		return ErrInvalidColumnData
+	}
+
+	var columns string
+	var placeholders []string
+	var values []any
+
+	for _, item := range data {
+		itemColumns, itemPlaceholders, itemValues, err := c.prepareInsertData(item)
+		if err != nil {
+			c.log.Errorf("prepare insert data failed: %v", err)
+			return ErrPrepareInsertDataFailed
+		}
+
+		if columns == "" {
+			columns = itemColumns
+		} else if columns != itemColumns {
+			c.log.Error("data items have inconsistent columns")
+			return ErrInvalidColumnData
+		}
+
+		placeholders = append(placeholders, fmt.Sprintf("(%s)", itemPlaceholders))
+		values = append(values, itemValues...)
+	}
+
+	// 构造 SQL 语句
+	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
+		tableName,
+		columns,
+		strings.Join(placeholders, ", "),
+	)
+
+	// 执行插入操作
+	if err := c.conn.Exec(ctx, query, values...); err != nil {
+		c.log.Errorf("insert many failed: %v", err)
+		return ErrInsertFailed
+	}
+
+	return nil
+}
+
+// AsyncInsert 异步插入数据
+func (c *Client) AsyncInsert(ctx context.Context, tableName string, data any, wait bool) error {
+	if c.conn == nil {
+		c.log.Error("clickhouse client is not initialized")
+		return ErrClientNotInitialized
+	}
+
+	// 准备插入数据
+	columns, placeholders, values, err := c.prepareInsertData(data)
+	if err != nil {
+		c.log.Errorf("prepare insert data failed: %v", err)
+		return ErrPrepareInsertDataFailed
+	}
+
+	// 构造 SQL 语句
+	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
+		tableName,
+		columns,
+		placeholders,
+	)
+
+	// 执行异步插入
+	if err = c.asyncInsert(ctx, query, wait, values...); err != nil {
+		c.log.Errorf("async insert failed: %v", err)
 		return ErrAsyncInsertFailed
 	}
 
 	return nil
 }
 
+// asyncInsert 异步插入数据
+func (c *Client) asyncInsert(ctx context.Context, query string, wait bool, args ...any) error {
+	if c.conn == nil {
+		c.log.Error("clickhouse client is not initialized")
+		return ErrClientNotInitialized
+	}
+
+	if err := c.conn.AsyncInsert(ctx, query, wait, args...); err != nil {
+		c.log.Errorf("async insert failed: %v", err)
+		return ErrAsyncInsertFailed
+	}
+
+	return nil
+}
+
+// AsyncInsertMany 批量异步插入数据
+func (c *Client) AsyncInsertMany(ctx context.Context, tableName string, data []any, wait bool) error {
+	if c.conn == nil {
+		c.log.Error("clickhouse client is not initialized")
+		return ErrClientNotInitialized
+	}
+
+	if len(data) == 0 {
+		c.log.Error("data slice is empty")
+		return ErrInvalidColumnData
+	}
+
+	// 准备插入数据的列名和占位符
+	var columns string
+	var placeholders []string
+	var values []any
+
+	for _, item := range data {
+		itemColumns, itemPlaceholders, itemValues, err := c.prepareInsertData(item)
+		if err != nil {
+			c.log.Errorf("prepare insert data failed: %v", err)
+			return ErrPrepareInsertDataFailed
+		}
+
+		if columns == "" {
+			columns = itemColumns
+		} else if columns != itemColumns {
+			c.log.Error("data items have inconsistent columns")
+			return ErrInvalidColumnData
+		}
+
+		placeholders = append(placeholders, fmt.Sprintf("(%s)", itemPlaceholders))
+		values = append(values, itemValues...)
+	}
+
+	// 构造 SQL 语句
+	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES %s",
+		tableName,
+		columns,
+		strings.Join(placeholders, ", "),
+	)
+
+	// 执行异步插入操作
+	if err := c.asyncInsert(ctx, query, wait, values...); err != nil {
+		c.log.Errorf("batch insert failed: %v", err)
+		return err
+	}
+
+	return nil
+}
+
 // BatchInsert 批量插入数据
-func (c *Client) BatchInsert(ctx context.Context, query string, data [][]interface{}) error {
+func (c *Client) BatchInsert(ctx context.Context, tableName string, data []any) error {
+	if c.conn == nil {
+		c.log.Error("clickhouse client is not initialized")
+		return ErrClientNotInitialized
+	}
+
+	if len(data) == 0 {
+		c.log.Error("data slice is empty")
+		return ErrInvalidColumnData
+	}
+
+	// 准备插入数据的列名和占位符
+	var columns string
+	var values [][]any
+
+	for _, item := range data {
+		itemColumns, _, itemValues, err := c.prepareInsertData(item)
+		if err != nil {
+			c.log.Errorf("prepare insert data failed: %v", err)
+			return ErrPrepareInsertDataFailed
+		}
+
+		if columns == "" {
+			columns = itemColumns
+		} else if columns != itemColumns {
+			c.log.Error("data items have inconsistent columns")
+			return ErrInvalidColumnData
+		}
+
+		values = append(values, itemValues)
+	}
+
+	// 构造 SQL 语句
+	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES", tableName, columns)
+
+	// 调用 batchExec 方法执行批量插入
+	if err := c.batchExec(ctx, query, values); err != nil {
+		c.log.Errorf("batch insert failed: %v", err)
+		return ErrBatchInsertFailed
+	}
+
+	return nil
+}
+
+// batchExec 执行批量操作
+func (c *Client) batchExec(ctx context.Context, query string, data [][]any) error {
 	batch, err := c.conn.PrepareBatch(ctx, query)
 	if err != nil {
 		c.log.Errorf("failed to prepare batch: %v", err)
@@ -339,8 +586,8 @@ func (c *Client) BatchInsert(ctx context.Context, query string, data [][]interfa
 	}
 
 	for _, row := range data {
-		if err := batch.Append(row...); err != nil {
-			c.log.Errorf("failed to append data: %v", err)
+		if err = batch.Append(row...); err != nil {
+			c.log.Errorf("failed to append batch data: %v", err)
 			return ErrBatchAppendFailed
 		}
 	}
@@ -352,3 +599,34 @@ func (c *Client) BatchInsert(ctx context.Context, query string, data [][]interfa
 
 	return nil
 }
+
+// BatchStructs 批量插入结构体数据
+func (c *Client) BatchStructs(ctx context.Context, query string, data []any) error {
+	if c.conn == nil {
+		c.log.Error("clickhouse client is not initialized")
+		return ErrClientNotInitialized
+	}
+
+	// 准备批量插入
+	batch, err := c.conn.PrepareBatch(ctx, query)
+	if err != nil {
+		c.log.Errorf("failed to prepare batch: %v", err)
+		return ErrBatchPrepareFailed
+	}
+
+	// 遍历数据并添加到批量插入
+	for _, row := range data {
+		if err := batch.AppendStruct(row); err != nil {
+			c.log.Errorf("failed to append batch struct data: %v", err)
+			return ErrBatchAppendFailed
+		}
+	}
+
+	// 发送批量插入
+	if err = batch.Send(); err != nil {
+		c.log.Errorf("failed to send batch: %v", err)
+		return ErrBatchSendFailed
+	}
+
+	return nil
+}

database/clickhouse/client_test.go

@@ -11,28 +11,25 @@ import (
 )
 
 type Candle struct {
-	Symbol    string    `json:"symbol" ch:"symbol"`
-	Open      float64   `json:"open" ch:"open"`
-	High      float64   `json:"high" ch:"high"`
-	Low       float64   `json:"low" ch:"low"`
-	Close     float64   `json:"close" ch:"close"`
-	Volume    float64   `json:"volume" ch:"volume"`
-	Timestamp time.Time `json:"timestamp" ch:"timestamp"`
+	Timestamp *time.Time `json:"timestamp" ch:"timestamp"`
+	Symbol    *string    `json:"symbol" ch:"symbol"`
+	Open      *float64   `json:"open" ch:"open"`
+	High      *float64   `json:"high" ch:"high"`
+	Low       *float64   `json:"low" ch:"low"`
+	Close     *float64   `json:"close" ch:"close"`
+	Volume    *float64   `json:"volume" ch:"volume"`
 }
 
 func createTestClient() *Client {
-	database := "finances"
-	username := "default"
-	password := "*Abcd123456"
 	cli, _ := NewClient(
 		log.DefaultLogger,
 		&conf.Bootstrap{
 			Data: &conf.Data{
 				Clickhouse: &conf.Data_ClickHouse{
 					Addresses: []string{"localhost:9000"},
-					Database:  &database,
-					Username:  &username,
-					Password:  &password,
+					Database:  Ptr("finances"),
+					Username:  Ptr("default"),
+					Password:  Ptr("*Abcd123456"),
 				},
 			},
 		},
@@ -44,7 +41,7 @@ func createCandlesTable(client *Client) {
 	// 创建表的 SQL 语句
 	createTableQuery := `
 		CREATE TABLE IF NOT EXISTS candles (
-			timestamp DateTime,
+			timestamp DateTime64(3),
 			symbol String,
 			open Float64,
 			high Float64,
@@ -76,32 +73,148 @@ func TestNewClient(t *testing.T) {
 	createCandlesTable(client)
 }
 
-func TestAsyncInsert(t *testing.T) {
+func TestInsertCandlesTable(t *testing.T) {
 	client := createTestClient()
 	assert.NotNil(t, client)
 
-	// 测试异步插入
-	err := client.AsyncInsert(context.Background(), "INSERT INTO test_table (id, name) VALUES (?, ?)", true, 1, "example")
-	assert.NoError(t, err, "AsyncInsert 应该成功执行")
-}
-
-func TestBatchInsert(t *testing.T) {
-	client := createTestClient()
-	assert.NotNil(t, client)
+	createCandlesTable(client)
 
 	// 测试数据
-	data := [][]interface{}{
-		{1, "example1"},
-		{2, "example2"},
-		{3, "example3"},
+	candle := &Candle{
+		Timestamp: Ptr(time.Now()),
+		Symbol:    Ptr("AAPL"),
+		Open:      Ptr(100.5),
+		High:      Ptr(105.0),
+		Low:       Ptr(99.5),
+		Close:     Ptr(102.0),
+		Volume:    Ptr(1500.0),
 	}
 
-	// 测试批量插入
-	err := client.BatchInsert(context.Background(), "INSERT INTO test_table (id, name) VALUES (?, ?)", data)
-	assert.NoError(t, err, "BatchInsert 应该成功执行")
+	// 插入数据
+	err := client.Insert(context.Background(), "candles", candle)
+	assert.NoError(t, err, "InsertCandlesTable 应该成功执行")
 }
 
-func TestInsertIntoCandlesTable(t *testing.T) {
+func TestInsertManyCandlesTable(t *testing.T) {
+	client := createTestClient()
+	assert.NotNil(t, client)
+
+	createCandlesTable(client)
+
+	// 测试数据
+	data := []any{
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("AAPL"),
+			Open:      Ptr(100.5),
+			High:      Ptr(105.0),
+			Low:       Ptr(99.5),
+			Close:     Ptr(102.0),
+			Volume:    Ptr(1500.0),
+		},
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("GOOG"),
+			Open:      Ptr(200.5),
+			High:      Ptr(205.0),
+			Low:       Ptr(199.5),
+			Close:     Ptr(202.0),
+			Volume:    Ptr(2500.0),
+		},
+	}
+
+	// 插入数据
+	err := client.InsertMany(context.Background(), "candles", data)
+	assert.NoError(t, err, "InsertManyCandlesTable 应该成功执行")
+}
+
+func TestAsyncInsertCandlesTable(t *testing.T) {
+	client := createTestClient()
+	assert.NotNil(t, client)
+
+	createCandlesTable(client)
+
+	// 测试数据
+	candle := &Candle{
+		Timestamp: Ptr(time.Now()),
+		Symbol:    Ptr("BTC/USD"),
+		Open:      Ptr(30000.0),
+		High:      Ptr(31000.0),
+		Low:       Ptr(29000.0),
+		Close:     Ptr(30500.0),
+		Volume:    Ptr(500.0),
+	}
+
+	// 异步插入数据
+	err := client.AsyncInsert(context.Background(), "candles", candle, true)
+	assert.NoError(t, err, "AsyncInsert 方法应该成功执行")
+
+	// 验证插入结果
+	query := `
+		SELECT timestamp, symbol, open, high, low, close, volume
+		FROM candles
+		WHERE symbol = ?
+	`
+	var result Candle
+	err = client.QueryRow(context.Background(), &result, query, "BTC/USD")
+	assert.NoError(t, err, "QueryRow 应该成功执行")
+	assert.Equal(t, "BTC/USD", *result.Symbol, "symbol 列值应该为 BTC/USD")
+	assert.Equal(t, 30500.0, *result.Close, "close 列值应该为 30500.0")
+	assert.Equal(t, 500.0, *result.Volume, "volume 列值应该为 500.0")
+}
+
+func TestAsyncInsertManyCandlesTable(t *testing.T) {
+	client := createTestClient()
+	assert.NotNil(t, client)
+
+	createCandlesTable(client)
+
+	// 测试数据
+	data := []any{
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("AAPL"),
+			Open:      Ptr(100.5),
+			High:      Ptr(105.0),
+			Low:       Ptr(99.5),
+			Close:     Ptr(102.0),
+			Volume:    Ptr(1500.0),
+		},
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("GOOG"),
+			Open:      Ptr(200.5),
+			High:      Ptr(205.0),
+			Low:       Ptr(199.5),
+			Close:     Ptr(202.0),
+			Volume:    Ptr(2500.0),
+		},
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("MSFT"),
+			Open:      Ptr(300.5),
+			High:      Ptr(305.0),
+			Low:       Ptr(299.5),
+			Close:     Ptr(302.0),
+			Volume:    Ptr(3500.0),
+		},
+	}
+
+	// 批量插入数据
+	err := client.AsyncInsertMany(context.Background(), "candles", data, true)
+	assert.NoError(t, err, "AsyncInsertMany 方法应该成功执行")
+
+	// 验证插入结果
+	query := `
+		SELECT timestamp, symbol, open, high, low, close, volume
+		FROM candles
+	`
+	var results []Candle
+	err = client.Select(context.Background(), &results, query)
+	assert.NoError(t, err, "查询数据应该成功执行")
+}
+
+func TestInternalBatchExecCandlesTable(t *testing.T) {
 	client := createTestClient()
 	assert.NotNil(t, client)
 
@@ -114,7 +227,121 @@ func TestInsertIntoCandlesTable(t *testing.T) {
 	`
 
 	// 测试数据
-	err := client.AsyncInsert(context.Background(), insertQuery, true,
+	data := [][]interface{}{
+		{"2023-10-01 12:00:00", "AAPL", 100.5, 105.0, 99.5, 102.0, 1500.0},
+		{"2023-10-01 12:01:00", "GOOG", 200.5, 205.0, 199.5, 202.0, 2500.0},
+		{"2023-10-01 12:02:00", "MSFT", 300.5, 305.0, 299.5, 302.0, 3500.0},
+	}
+
+	// 批量插入数据
+	err := client.batchExec(context.Background(), insertQuery, data)
+	assert.NoError(t, err, "batchExec 应该成功执行")
+}
+
+func TestBatchInsertCandlesTable(t *testing.T) {
+	client := createTestClient()
+	assert.NotNil(t, client)
+
+	createCandlesTable(client)
+
+	// 测试数据
+	data := []any{
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("AAPL"),
+			Open:      Ptr(100.5),
+			High:      Ptr(105.0),
+			Low:       Ptr(99.5),
+			Close:     Ptr(102.0),
+			Volume:    Ptr(1500.0),
+		},
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("GOOG"),
+			Open:      Ptr(200.5),
+			High:      Ptr(205.0),
+			Low:       Ptr(199.5),
+			Close:     Ptr(202.0),
+			Volume:    Ptr(2500.0),
+		},
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("MSFT"),
+			Open:      Ptr(300.5),
+			High:      Ptr(305.0),
+			Low:       Ptr(299.5),
+			Close:     Ptr(302.0),
+			Volume:    Ptr(3500.0),
+		},
+	}
+
+	// 批量插入数据
+	err := client.BatchInsert(context.Background(), "candles", data)
+	assert.NoError(t, err, "BatchInsert 方法应该成功执行")
+
+	// 验证插入结果
+	query := `
+		SELECT timestamp, symbol, open, high, low, close, volume
+		FROM candles
+	`
+	var results []Candle
+	err = client.Select(context.Background(), &results, query)
+	assert.NoError(t, err, "查询数据应该成功执行")
+}
+
+func TestBatchStructsCandlesTable(t *testing.T) {
+	client := createTestClient()
+	assert.NotNil(t, client)
+
+	createCandlesTable(client)
+
+	// 插入数据的 SQL 语句
+	insertQuery := `
+		INSERT INTO candles (timestamp, symbol, open, high, low, close, volume)
+		VALUES (?, ?, ?, ?, ?, ?, ?)
+	`
+
+	// 测试数据
+	data := []any{
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("AAPL"),
+			Open:      Ptr(100.5),
+			High:      Ptr(105.0),
+			Low:       Ptr(99.5),
+			Close:     Ptr(102.0),
+			Volume:    Ptr(1500.0),
+		},
+		&Candle{
+			Timestamp: Ptr(time.Now()),
+			Symbol:    Ptr("GOOG"),
+			Open:      Ptr(200.5),
+			High:      Ptr(205.0),
+			Low:       Ptr(199.5),
+			Close:     Ptr(202.0),
+			Volume:    Ptr(2500.0),
+		},
+	}
+
+	// 批量插入数据
+	err := client.BatchStructs(context.Background(), insertQuery, data)
+	assert.NoError(t, err, "BatchStructsCandlesTable 应该成功执行")
+}
+
+func TestInternalAsyncInsertIntoCandlesTable(t *testing.T) {
+	client := createTestClient()
+	assert.NotNil(t, client)
+
+	createCandlesTable(client)
+
+	// 插入数据的 SQL 语句
+	insertQuery := `
+		INSERT INTO candles (timestamp, symbol, open, high, low, close, volume)
+		VALUES (?, ?, ?, ?, ?, ?, ?)
+	`
+
+	// 测试数据
+	err := client.asyncInsert(context.Background(), insertQuery, true,
 		"2023-10-01 12:00:00", "AAPL", 100.5, 105.0, 99.5, 102.0, 1500.0)
 	assert.NoError(t, err, "InsertIntoCandlesTable 应该成功执行")
 }
@@ -138,6 +365,22 @@ func TestQueryCandlesTable(t *testing.T) {
 	err := client.Query(context.Background(), func() interface{} { return &Candle{} }, &results, query)
 	assert.NoError(t, err, "QueryCandlesTable 应该成功执行")
 	assert.NotEmpty(t, results, "QueryCandlesTable 应该返回结果")
+	for _, result := range results {
+		candle, ok := result.(*Candle)
+		assert.True(t, ok, "结果应该是 Candle 类型")
+		assert.NotNil(t, candle.Timestamp, "Timestamp 列不应该为 nil")
+		assert.NotNil(t, candle.Symbol, "Symbol 列不应该为 nil")
+		assert.NotNil(t, candle.Open, "Open 列不应该为 nil")
+		assert.NotNil(t, candle.High, "High 列不应该为 nil")
+		assert.NotNil(t, candle.Low, "Low 列不应该为 nil")
+		assert.NotNil(t, candle.Close, "Close 列不应该为 nil")
+		assert.NotNil(t, candle.Volume, "Volume 列不应该为 nil")
+		t.Logf("[%v] Candle: %s, Open: %f, High: %f, Low: %f, Close: %f, Volume: %f\n",
+			candle.Timestamp.String(),
+			*candle.Symbol,
+			*candle.Open, *candle.High, *candle.Low, *candle.Close, *candle.Volume,
+		)
+	}
 }
 
 func TestSelectCandlesTable(t *testing.T) {
@@ -159,6 +402,21 @@ func TestSelectCandlesTable(t *testing.T) {
 	err := client.Select(context.Background(), &results, query)
 	assert.NoError(t, err, "QueryCandlesTable 应该成功执行")
 	assert.NotEmpty(t, results, "QueryCandlesTable 应该返回结果")
+
+	for _, result := range results {
+		assert.NotNil(t, result.Timestamp, "Timestamp 列不应该为 nil")
+		assert.NotNil(t, result.Symbol, "Symbol 列不应该为 nil")
+		assert.NotNil(t, result.Open, "Open 列不应该为 nil")
+		assert.NotNil(t, result.High, "High 列不应该为 nil")
+		assert.NotNil(t, result.Low, "Low 列不应该为 nil")
+		assert.NotNil(t, result.Close, "Close 列不应该为 nil")
+		assert.NotNil(t, result.Volume, "Volume 列不应该为 nil")
+		t.Logf("[%v] Candle: %s, Open: %f, High: %f, Low: %f, Close: %f, Volume: %f\n",
+			result.Timestamp.String(),
+			*result.Symbol,
+			*result.Open, *result.High, *result.Low, *result.Close, *result.Volume,
+		)
+	}
 }
 
 func TestQueryRow(t *testing.T) {
@@ -172,7 +430,7 @@ func TestQueryRow(t *testing.T) {
 		INSERT INTO candles (timestamp, symbol, open, high, low, close, volume)
 		VALUES (?, ?, ?, ?, ?, ?, ?)
 	`
-	err := client.AsyncInsert(context.Background(), insertQuery, true,
+	err := client.asyncInsert(context.Background(), insertQuery, true,
 		"2023-10-01 12:00:00", "AAPL", 100.5, 105.0, 99.5, 102.0, 1500.0)
 	assert.NoError(t, err, "数据插入失败")
 
@@ -182,13 +440,19 @@ func TestQueryRow(t *testing.T) {
 		FROM candles
 		WHERE symbol = ?
 	`
+
 	var result Candle
 
 	err = client.QueryRow(context.Background(), &result, query, "AAPL")
 	assert.NoError(t, err, "QueryRow 应该成功执行")
-	assert.Equal(t, "AAPL", result.Symbol, "symbol 列值应该为 AAPL")
-	assert.Equal(t, 100.5, result.Open, "open 列值应该为 100.5")
-	assert.Equal(t, 1500.0, result.Volume, "volume 列值应该为 1500.0")
+	assert.Equal(t, "AAPL", *result.Symbol, "symbol 列值应该为 AAPL")
+	assert.Equal(t, 100.5, *result.Open, "open 列值应该为 100.5")
+	assert.Equal(t, 1500.0, *result.Volume, "volume 列值应该为 1500.0")
+	t.Logf("QueryRow Result: [%v] Candle: %s, Open: %f, High: %f, Low: %f, Close: %f, Volume: %f\n",
+		result.Timestamp.String(),
+		*result.Symbol,
+		*result.Open, *result.High, *result.Low, *result.Close, *result.Volume,
+	)
 }
 
 func TestDropCandlesTable(t *testing.T) {
@@ -234,23 +498,3 @@ func TestAggregateCandlesTable(t *testing.T) {
 	assert.NoError(t, err, "AggregateCandlesTable 应该成功执行")
 	assert.NotEmpty(t, results, "AggregateCandlesTable 应该返回结果")
 }
-
-func TestBatchInsertCandlesTable(t *testing.T) {
-	client := createTestClient()
-	assert.NotNil(t, client)
-
-	createCandlesTable(client)
-
-	// 测试数据
-	data := [][]interface{}{
-		{"2023-10-01 12:00:00", "AAPL", 100.5, 105.0, 99.5, 102.0, 1500.0},
-		{"2023-10-01 12:01:00", "GOOG", 200.5, 205.0, 199.5, 202.0, 2500.0},
-		{"2023-10-01 12:02:00", "MSFT", 300.5, 305.0, 299.5, 302.0, 3500.0},
-	}
-
-	// 批量插入数据
-	err := client.BatchInsert(context.Background(), `
-		INSERT INTO candles (timestamp, symbol, open, high, low, close, volume)
-		VALUES (?, ?, ?, ?, ?, ?, ?)`, data)
-	assert.NoError(t, err, "BatchInsertCandlesTable 应该成功执行")
-}

database/clickhouse/errors.go

@@ -72,9 +72,18 @@ var (
 	// ErrBatchAppendFailed is returned when appending to a batch fails.
 	ErrBatchAppendFailed = errors.InternalServer("BATCH_APPEND_FAILED", "batch append operation failed")
 
+	// ErrBatchInsertFailed is returned when a batch insert operation fails.
+	ErrBatchInsertFailed = errors.InternalServer("BATCH_INSERT_FAILED", "batch insert operation failed")
+
 	// ErrInvalidDSN is returned when the data source name (DSN) is invalid.
 	ErrInvalidDSN = errors.InternalServer("INVALID_DSN", "invalid data source name")
 
 	// ErrInvalidProxyURL is returned when the proxy URL is invalid.
 	ErrInvalidProxyURL = errors.InternalServer("INVALID_PROXY_URL", "invalid proxy URL")
+
+	// ErrPrepareInsertDataFailed is returned when preparing insert data fails.
+	ErrPrepareInsertDataFailed = errors.InternalServer("PREPARE_INSERT_DATA_FAILED", "failed to prepare insert data")
+
+	// ErrInvalidColumnData is returned when the column data type is invalid.
+	ErrInvalidColumnData = errors.InternalServer("INVALID_COLUMN_DATA", "invalid column data type")
 )

database/clickhouse/utils.go

@@ -0,0 +1,146 @@
+package clickhouse
+
+import (
+	"database/sql"
+	"reflect"
+	"time"
+
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+const (
+	timeFormat = "2006-01-02 15:04:05.000000000"
+)
+
+func structToValueArray(input any) []any {
+	// 检查是否是指针类型，如果是则解引用
+	val := reflect.ValueOf(input)
+	if val.Kind() == reflect.Ptr {
+		val = val.Elem()
+	}
+
+	// 确保输入是结构体
+	if val.Kind() != reflect.Struct {
+		return nil
+	}
+
+	var values []any
+	for i := 0; i < val.NumField(); i++ {
+		value := val.Field(i).Interface()
+
+		switch v := value.(type) {
+		case *sql.NullString:
+			if v.Valid {
+				values = append(values, v.String)
+			} else {
+				values = append(values, nil)
+			}
+		case *sql.NullInt64:
+			if v.Valid {
+				values = append(values, v.Int64)
+			} else {
+				values = append(values, nil)
+			}
+		case *sql.NullFloat64:
+			if v.Valid {
+				values = append(values, v.Float64)
+			} else {
+				values = append(values, nil)
+			}
+		case *sql.NullBool:
+			if v.Valid {
+				values = append(values, v.Bool)
+			} else {
+				values = append(values, nil)
+			}
+
+		case *sql.NullTime:
+			if v != nil && v.Valid {
+				values = append(values, v.Time.Format(timeFormat))
+			} else {
+				values = append(values, nil)
+			}
+
+		case *time.Time:
+			if v != nil {
+				values = append(values, v.Format(timeFormat))
+			} else {
+				values = append(values, nil)
+			}
+
+		case time.Time:
+			// 处理 time.Time 类型
+			if !v.IsZero() {
+				values = append(values, v.Format(timeFormat))
+			} else {
+				values = append(values, nil) // 如果时间为零值，插入 NULL
+			}
+
+		case timestamppb.Timestamp:
+			// 处理 timestamppb.Timestamp 类型
+			if !v.IsValid() {
+				values = append(values, v.AsTime().Format(timeFormat))
+			} else {
+				values = append(values, nil) // 如果时间为零值，插入 NULL
+			}
+
+		case *timestamppb.Timestamp:
+			// 处理 *timestamppb.Timestamp 类型
+			if v != nil && v.IsValid() {
+				values = append(values, v.AsTime().Format(timeFormat))
+			} else {
+				values = append(values, nil) // 如果时间为零值，插入 NULL
+			}
+
+		case durationpb.Duration:
+			// 处理 timestamppb.Duration 类型
+			if v.AsDuration() != 0 {
+				values = append(values, v.AsDuration().String())
+			} else {
+				values = append(values, nil) // 如果时间为零值，插入 NULL
+			}
+
+		case *durationpb.Duration:
+			// 处理 *timestamppb.Duration 类型
+			if v != nil && v.AsDuration() != 0 {
+				values = append(values, v.AsDuration().String())
+			} else {
+				values = append(values, nil) // 如果时间为零值，插入 NULL
+			}
+
+		case []any:
+			// 处理切片类型
+			if len(v) > 0 {
+				for _, item := range v {
+					if item == nil {
+						values = append(values, nil)
+					} else {
+						values = append(values, item)
+					}
+				}
+			} else {
+				values = append(values, nil) // 如果切片为空，插入 NULL
+			}
+
+		case [][]any:
+			// 处理二维切片类型
+			if len(v) > 0 {
+				for _, item := range v {
+					if len(item) > 0 {
+						values = append(values, item)
+					} else {
+						values = append(values, nil) // 如果子切片为空，插入 NULL
+					}
+				}
+			} else {
+				values = append(values, nil) // 如果二维切片为空，插入 NULL
+			}
+
+		default:
+			values = append(values, v)
+		}
+	}
+
+	return values
+}

database/clickhouse/utils_test.go

@@ -0,0 +1,83 @@
+package clickhouse
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// Ptr returns a pointer to the provided value.
+func Ptr[T any](v T) *T {
+	return &v
+}
+
+func TestStructToValueArrayWithCandle(t *testing.T) {
+	now := time.Now()
+
+	candle := Candle{
+		Timestamp: Ptr(now),
+		Symbol:    Ptr("AAPL"),
+		Open:      Ptr(100.5),
+		High:      Ptr(105.0),
+		Low:       Ptr(99.5),
+		Close:     Ptr(102.0),
+		Volume:    Ptr(1500.0),
+	}
+
+	values := structToValueArray(candle)
+	assert.NotNil(t, values, "Values should not be nil")
+	assert.Len(t, values, 7, "Expected 7 fields in the Candle struct")
+	assert.Equal(t, now.Format(timeFormat), values[0].(string), "Timestamp should match")
+	assert.Equal(t, *candle.Symbol, *(values[1].(*string)), "Symbol should match")
+	assert.Equal(t, *candle.Open, *values[2].(*float64), "Open price should match")
+	assert.Equal(t, *candle.High, *values[3].(*float64), "High price should match")
+	assert.Equal(t, *candle.Low, *values[4].(*float64), "Low price should match")
+	assert.Equal(t, *candle.Close, *values[5].(*float64), "Close price should match")
+	assert.Equal(t, *candle.Volume, *values[6].(*float64), "Volume should match")
+
+	t.Logf("QueryRow Result: [%v] Candle: %s, Open: %f, High: %f, Low: %f, Close: %f, Volume: %f\n",
+		values[0],
+		*(values[1].(*string)),
+		*values[2].(*float64),
+		*values[3].(*float64),
+		*values[4].(*float64),
+		*values[5].(*float64),
+		*values[6].(*float64),
+	)
+}
+
+func TestStructToValueArrayWithCandlePtr(t *testing.T) {
+	now := time.Now()
+
+	candle := &Candle{
+		Timestamp: Ptr(now),
+		Symbol:    Ptr("AAPL"),
+		Open:      Ptr(100.5),
+		High:      Ptr(105.0),
+		Low:       Ptr(99.5),
+		Close:     Ptr(102.0),
+		Volume:    Ptr(1500.0),
+	}
+
+	values := structToValueArray(candle)
+	assert.NotNil(t, values, "Values should not be nil")
+	assert.Len(t, values, 7, "Expected 7 fields in the Candle struct")
+	assert.Equal(t, now.Format(timeFormat), values[0].(string), "Timestamp should match")
+	assert.Equal(t, *candle.Symbol, *(values[1].(*string)), "Symbol should match")
+	assert.Equal(t, *candle.Open, *values[2].(*float64), "Open price should match")
+	assert.Equal(t, *candle.High, *values[3].(*float64), "High price should match")
+	assert.Equal(t, *candle.Low, *values[4].(*float64), "Low price should match")
+	assert.Equal(t, *candle.Close, *values[5].(*float64), "Close price should match")
+	assert.Equal(t, *candle.Volume, *values[6].(*float64), "Volume should match")
+
+	t.Logf("QueryRow Result: [%v] Candle: %s, Open: %f, High: %f, Low: %f, Close: %f, Volume: %f\n",
+		values[0],
+		*(values[1].(*string)),
+		*values[2].(*float64),
+		*values[3].(*float64),
+		*values[4].(*float64),
+		*values[5].(*float64),
+		*values[6].(*float64),
+	)
+}

database/elasticsearch/README.md

@@ -14,3 +14,19 @@
 - 动态映射（dynamic mapping）
 - 显式映射（explicit mapping）
 - 严格映射（strict mappings）
+
+## Docker部署
+
+```bash
+docker pull bitnami/elasticsearch:latest
+
+docker run -itd \
+    --name elasticsearch \
+    -p 9200:9200 \
+    -p 9300:9300 \
+    -e ELASTICSEARCH_USERNAME=elastic \
+    -e ELASTICSEARCH_PASSWORD=elastic \
+    -e ELASTICSEARCH_NODE_NAME=elasticsearch-node-1 \
+    -e ELASTICSEARCH_CLUSTER_NAME=elasticsearch-cluster \
+    bitnami/elasticsearch:latest
+```

rpc/go.mod

@@ -4,54 +4,49 @@ go 1.23.0
 
 toolchain go1.24.3
 
-replace (
-	github.com/bufbuild/protovalidate-go => buf.build/go/protovalidate v0.10.1
-
-	github.com/tx7do/kratos-bootstrap/api => ../api
-)
+replace github.com/tx7do/kratos-bootstrap/api => ../api
 
 require (
+	buf.build/go/protovalidate v0.14.0
 	github.com/go-kratos/aegis v0.2.0
-	github.com/go-kratos/kratos/contrib/middleware/validate/v2 v2.0.0-20250527152916-d6f5f00cf562
 	github.com/go-kratos/kratos/v2 v2.8.4
 	github.com/gorilla/handlers v1.5.2
-	github.com/tx7do/kratos-bootstrap/api v0.0.21
+	github.com/tx7do/kratos-bootstrap/api v0.0.27
 	github.com/tx7do/kratos-bootstrap/utils v0.1.3
-	google.golang.org/grpc v1.72.2
+	google.golang.org/grpc v1.73.0
+	google.golang.org/protobuf v1.36.6
 )
 
 require (
-	buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250425153114-8976f5be98c1.1 // indirect
+	buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250717185734-6c6e0d3c608e.1 // indirect
 	cel.dev/expr v0.24.0 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
-	github.com/bufbuild/protovalidate-go v0.12.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/go-playground/form/v4 v4.2.1 // indirect
-	github.com/google/cel-go v0.25.0 // indirect
+	github.com/google/cel-go v0.26.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/mux v1.8.1 // indirect
 	github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/shirou/gopsutil/v3 v3.24.5 // indirect
 	github.com/shoenig/go-m1cpu v0.1.6 // indirect
-	github.com/stoewer/go-strcase v1.3.0 // indirect
+	github.com/stoewer/go-strcase v1.3.1 // indirect
 	github.com/tklauser/go-sysconf v0.3.15 // indirect
 	github.com/tklauser/numcpus v0.10.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/otel v1.36.0 // indirect
-	go.opentelemetry.io/otel/metric v1.36.0 // indirect
-	go.opentelemetry.io/otel/trace v1.36.0 // indirect
-	golang.org/x/exp v0.0.0-20250531010427-b6e5de432a8b // indirect
-	golang.org/x/net v0.40.0 // indirect
-	golang.org/x/sync v0.14.0 // indirect
-	golang.org/x/sys v0.33.0 // indirect
-	golang.org/x/text v0.25.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20250528174236-200df99c418a // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250528174236-200df99c418a // indirect
-	google.golang.org/protobuf v1.36.6 // indirect
+	go.opentelemetry.io/otel v1.37.0 // indirect
+	go.opentelemetry.io/otel/metric v1.37.0 // indirect
+	go.opentelemetry.io/otel/trace v1.37.0 // indirect
+	golang.org/x/exp v0.0.0-20250718183923-645b1fa84792 // indirect
+	golang.org/x/net v0.42.0 // indirect
+	golang.org/x/sync v0.16.0 // indirect
+	golang.org/x/sys v0.34.0 // indirect
+	golang.org/x/text v0.27.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

rpc/go.sum

@@ -1,13 +1,13 @@
-buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250425153114-8976f5be98c1.1 h1:YhMSc48s25kr7kv31Z8vf7sPUIq5YJva9z1mn/hAt0M=
-buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250425153114-8976f5be98c1.1/go.mod h1:avRlCjnFzl98VPaeCtJ24RrV/wwHFzB8sWXhj26+n/U=
-buf.build/go/protovalidate v0.10.1 h1:2k9jAfLBLvzomB7yftRKUkpUi6Rb5TVPs5vyY+1TMYM=
-buf.build/go/protovalidate v0.10.1/go.mod h1:2NC0NSB6Lon4wR2wxisxDD6LnoJDPMB5i6BTLjD2Szw=
+buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250717185734-6c6e0d3c608e.1 h1:Lg6klmCi3v7VvpqeeLEER9/m5S8y9e9DjhqQnSCNy4k=
+buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.6-20250717185734-6c6e0d3c608e.1/go.mod h1:avRlCjnFzl98VPaeCtJ24RrV/wwHFzB8sWXhj26+n/U=
+buf.build/go/protovalidate v0.14.0 h1:kr/rC/no+DtRyYX+8KXLDxNnI1rINz0imk5K44ZpZ3A=
+buf.build/go/protovalidate v0.14.0/go.mod h1:+F/oISho9MO7gJQNYC2VWLzcO1fTPmaTA08SDYJZncA=
 cel.dev/expr v0.24.0 h1:56OvJKSH3hDGL0ml5uSxZmz3/3Pq4tJ+fb1unVLAFcY=
 cel.dev/expr v0.24.0/go.mod h1:hLPLo1W4QUmuYdA72RBX06QTs6MXw941piREPl3Yfiw=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
-github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42 h1:Om6kYQYDUk5wWbT0t0q6pvyM49i9XZAv9dDrkDA7gjk=
-github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
+github.com/cncf/xds/go v0.0.0-20250326154945-ae57f3c0d45f h1:C5bqEmzEPLsHm9Mv73lSE9e9bKV23aB1vxOsmZrkl3k=
+github.com/cncf/xds/go v0.0.0-20250326154945-ae57f3c0d45f/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -20,8 +20,6 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/go-kratos/aegis v0.2.0 h1:dObzCDWn3XVjUkgxyBp6ZeWtx/do0DPZ7LY3yNSJLUQ=
 github.com/go-kratos/aegis v0.2.0/go.mod h1:v0R2m73WgEEYB3XYu6aE2WcMwsZkJ/Rzuf5eVccm7bI=
-github.com/go-kratos/kratos/contrib/middleware/validate/v2 v2.0.0-20250527152916-d6f5f00cf562 h1:8R4XBtpXVxrTFetkbUYGZCfrZZfXULamq8MeLLsmKWY=
-github.com/go-kratos/kratos/contrib/middleware/validate/v2 v2.0.0-20250527152916-d6f5f00cf562/go.mod h1:NRPjykSmmscIXLP6Qv42b/UFhwZgxH9bD9OLgbMXMqs=
 github.com/go-kratos/kratos/v2 v2.8.4 h1:eIJLE9Qq9WSoKx+Buy2uPyrahtF/lPh+Xf4MTpxhmjs=
 github.com/go-kratos/kratos/v2 v2.8.4/go.mod h1:mq62W2101a5uYyRxe+7IdWubu7gZCGYqSNKwGFiiRcw=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
@@ -38,8 +36,8 @@ github.com/go-playground/form/v4 v4.2.1 h1:HjdRDKO0fftVMU5epjPW2SOREcZ6/wLUzEobq
 github.com/go-playground/form/v4 v4.2.1/go.mod h1:q1a2BY+AQUUzhl6xA/6hBetay6dEIhMHjgvJiGo6K7U=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
-github.com/google/cel-go v0.25.0 h1:jsFw9Fhn+3y2kBbltZR4VEz5xKkcIFRPDnuEzAGv5GY=
-github.com/google/cel-go v0.25.0/go.mod h1:hjEb6r5SuOSlhCHmFoLzu8HGCERvIsDAbxDAyNU/MmI=
+github.com/google/cel-go v0.26.0 h1:DPGjXackMpJWH680oGY4lZhYjIameYmR+/6RBdDGmaI=
+github.com/google/cel-go v0.26.0/go.mod h1:A9O8OU9rdvrK5MQyrqfIxo1a0u4g3sF8KB6PUIaryMM=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -68,8 +66,8 @@ github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFt
 github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
 github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
 github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k=
-github.com/stoewer/go-strcase v1.3.0 h1:g0eASXYtp+yvN9fK8sH94oCIk0fau9uV1/ZdJ0AVEzs=
-github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
+github.com/stoewer/go-strcase v1.3.1 h1:iS0MdW+kVTxgMoE1LAZyMiYJFKlOzLooE4MxjirtkAs=
+github.com/stoewer/go-strcase v1.3.1/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -88,35 +86,35 @@ github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo
 github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
-go.opentelemetry.io/otel v1.36.0 h1:UumtzIklRBY6cI/lllNZlALOF5nNIzJVb16APdvgTXg=
-go.opentelemetry.io/otel v1.36.0/go.mod h1:/TcFMXYjyRNh8khOAO9ybYkqaDBb/70aVwkNML4pP8E=
-go.opentelemetry.io/otel/metric v1.36.0 h1:MoWPKVhQvJ+eeXWHFBOPoBOi20jh6Iq2CcCREuTYufE=
-go.opentelemetry.io/otel/metric v1.36.0/go.mod h1:zC7Ks+yeyJt4xig9DEw9kuUFe5C3zLbVjV2PzT6qzbs=
-go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A=
-go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU=
-go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce1EK0Gyvahk=
-go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w=
-go.opentelemetry.io/otel/trace v1.36.0 h1:ahxWNuqZjpdiFAyrIoQ4GIiAIhxAunQR6MUoKrsNd4w=
-go.opentelemetry.io/otel/trace v1.36.0/go.mod h1:gQ+OnDZzrybY4k4seLzPAWNwVBBVlF2szhehOBB/tGA=
-golang.org/x/exp v0.0.0-20250531010427-b6e5de432a8b h1:QoALfVG9rhQ/M7vYDScfPdWjGL9dlsVVM5VGh7aKoAA=
-golang.org/x/exp v0.0.0-20250531010427-b6e5de432a8b/go.mod h1:U6Lno4MTRCDY+Ba7aCcauB9T60gsv5s4ralQzP72ZoQ=
-golang.org/x/net v0.40.0 h1:79Xs7wF06Gbdcg4kdCCIQArK11Z1hr5POQ6+fIYHNuY=
-golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds=
-golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
-golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
+go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
+go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
+go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
+go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY=
+go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg=
+go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o=
+go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w=
+go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
+go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
+golang.org/x/exp v0.0.0-20250718183923-645b1fa84792 h1:R9PFI6EUdfVKgwKjZef7QIwGcBKu86OEFpJ9nUEP2l4=
+golang.org/x/exp v0.0.0-20250718183923-645b1fa84792/go.mod h1:A+z0yzpGtvnG90cToK5n2tu8UJVP2XUATh+r+sfOOOc=
+golang.org/x/net v0.42.0 h1:jzkYrhi3YQWD6MLBJcsklgQsoAcw89EcZbJw8Z614hs=
+golang.org/x/net v0.42.0/go.mod h1:FF1RA5d3u7nAYA4z2TkclSCKh68eSXtiFwcWQpPXdt8=
+golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
+golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
-golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
-golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
-golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
-google.golang.org/genproto/googleapis/api v0.0.0-20250528174236-200df99c418a h1:SGktgSolFCo75dnHJF2yMvnns6jCmHFJ0vE4Vn2JKvQ=
-google.golang.org/genproto/googleapis/api v0.0.0-20250528174236-200df99c418a/go.mod h1:a77HrdMjoeKbnd2jmgcWdaS++ZLZAEq3orIOAEIKiVw=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250528174236-200df99c418a h1:v2PbRU4K3llS09c7zodFpNePeamkAwG3mPrAery9VeE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250528174236-200df99c418a/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
-google.golang.org/grpc v1.72.2 h1:TdbGzwb82ty4OusHWepvFWGLgIbNo1/SUynEN0ssqv8=
-google.golang.org/grpc v1.72.2/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM=
+golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA=
+golang.org/x/sys v0.34.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4=
+golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU=
+google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79 h1:iOye66xuaAK0WnkPuhQPUFy8eJcmwUXqGGP3om6IxX8=
+google.golang.org/genproto/googleapis/api v0.0.0-20250715232539-7130f93afb79/go.mod h1:HKJDgKsFUnv5VAGeQjz8kxcgDP0HoE0iZNp0OdZNlhE=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79 h1:1ZwqphdOdWYXsUHgMpU/101nCtf/kSp9hOrcvFsnl10=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250715232539-7130f93afb79/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
+google.golang.org/grpc v1.73.0 h1:VIWSmpI2MegBtTuFt5/JWy2oXxtjJ/e89Z70ImfD2ok=
+google.golang.org/grpc v1.73.0/go.mod h1:50sbHOUqWoCQGI8V2HQLJM0B+LMlIUjNSZmow7EVBQc=
 google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
 google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

rpc/grpc.go

@@ -14,13 +14,14 @@ import (
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/registry"
 
-	"github.com/go-kratos/kratos/contrib/middleware/validate/v2"
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/middleware/metadata"
 	midRateLimit "github.com/go-kratos/kratos/v2/middleware/ratelimit"
 	"github.com/go-kratos/kratos/v2/middleware/recovery"
 	"github.com/go-kratos/kratos/v2/middleware/tracing"
 
+	"github.com/tx7do/kratos-bootstrap/rpc/middleware/validate"
+
 	kratosGrpc "github.com/go-kratos/kratos/v2/transport/grpc"
 
 	conf "github.com/tx7do/kratos-bootstrap/api/gen/go/conf/v1"

rpc/middleware/validate/validate.go

@@ -0,0 +1,37 @@
+package validate
+
+import (
+	"context"
+
+	"github.com/go-kratos/kratos/v2/errors"
+	"github.com/go-kratos/kratos/v2/middleware"
+
+	"buf.build/go/protovalidate"
+	"google.golang.org/protobuf/proto"
+)
+
+type validator interface {
+	Validate() error
+}
+
+// ProtoValidate is a middleware that validates the request message with [protovalidate](https://github.com/bufbuild/protovalidate)
+func ProtoValidate() middleware.Middleware {
+	return func(handler middleware.Handler) middleware.Handler {
+		return func(ctx context.Context, req any) (reply any, err error) {
+			if msg, ok := req.(proto.Message); ok {
+				if err := protovalidate.Validate(msg); err != nil {
+					return nil, errors.BadRequest("VALIDATOR", err.Error()).WithCause(err)
+				}
+			}
+
+			// to compatible with the [old validator](https://github.com/envoyproxy/protoc-gen-validate)
+			if v, ok := req.(validator); ok {
+				if err := v.Validate(); err != nil {
+					return nil, errors.BadRequest("VALIDATOR", err.Error()).WithCause(err)
+				}
+			}
+
+			return handler(ctx, req)
+		}
+	}
+}

rpc/middleware/validate/validate_test.go

@@ -0,0 +1,11 @@
+package validate
+
+import (
+	"google.golang.org/protobuf/proto"
+)
+
+type testcase struct {
+	name string
+	req  proto.Message
+	err  bool
+}

rpc/rest.go

@@ -9,13 +9,14 @@ import (
 	"github.com/go-kratos/aegis/ratelimit"
 	"github.com/go-kratos/aegis/ratelimit/bbr"
 
-	"github.com/go-kratos/kratos/contrib/middleware/validate/v2"
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/middleware/metadata"
 	midRateLimit "github.com/go-kratos/kratos/v2/middleware/ratelimit"
 	"github.com/go-kratos/kratos/v2/middleware/recovery"
 	"github.com/go-kratos/kratos/v2/middleware/tracing"
 
+	"github.com/tx7do/kratos-bootstrap/rpc/middleware/validate"
+
 	kratosRest "github.com/go-kratos/kratos/v2/transport/http"
 
 	conf "github.com/tx7do/kratos-bootstrap/api/gen/go/conf/v1"

tag.bat

@@ -5,14 +5,14 @@ git tag utils/v0.1.4 --force
 git tag cache/redis/v0.0.11 --force
 git tag oss/minio/v0.0.10 --force
 git tag logger/v0.0.10 --force
-git tag rpc/v0.0.14 --force
+git tag rpc/v0.0.18 --force
 git tag tracer/v0.0.10 --force
 
 git tag database/ent/v0.0.10 --force
 git tag database/gorm/v0.0.10 --force
 git tag database/mongodb/v0.0.12 --force
 git tag database/influxdb/v0.0.12 --force
-git tag database/clickhouse/v0.0.12 --force
+git tag database/clickhouse/v0.0.14 --force
 git tag database/elasticsearch/v0.0.12 --force
 git tag database/cassandra/v0.0.10 --force