Api/database/clickhouse.go

package database

import (
	"context"
	"crypto/tls"
	"fmt"
	"github.com/ClickHouse/clickhouse-go/v2"
	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
	"sync"
	"time"
)

type _ch struct {
	Database[driver.Rows, driver.Batch]

	config *configCH
	conn   clickhouse.Conn
}

//goland:noinspection GoUnusedExportedFunction
func NewClickhouse(cfg *Config, debug bool) (Database[driver.Rows, driver.Batch], error) {
	if cfg == nil || cfg.Clickhouse == nil {
		return nil, fmt.Errorf("illegal state: `*config.ClickhouseConfig` is nil")
	}

	var tlcConfig *tls.Config
	if cfg.Clickhouse.Params.Secure {
		tlcConfig = &tls.Config{
			InsecureSkipVerify: true,
		}
	}

	session, err := clickhouse.Open(&clickhouse.Options{
		Protocol: clickhouse.Native,
		TLS:      tlcConfig,
		Addr: []string{
			fmt.Sprintf("%s:%d", cfg.Clickhouse.Host, cfg.Clickhouse.Port),
		},
		Auth: clickhouse.Auth{
			Database: cfg.Clickhouse.Database,
			Username: cfg.Clickhouse.Username,
			Password: cfg.Clickhouse.Password,
		},
		Debug: debug,
		Compression: &clickhouse.Compression{
			Method: cfg.Clickhouse.Params.CompressionMethod(),
		},
		DialTimeout: time.Duration(cfg.Clickhouse.Params.Timeout) * time.Second,
		ReadTimeout: time.Duration(cfg.Clickhouse.Params.ReadTimeout) * time.Second,
	})
	if err != nil {
		return nil, err
	}

	s := &_ch{
		config: cfg.Clickhouse,
		conn:   session,
	}

	if err = s.Ping(context.TODO()); err != nil {
		return nil, err
	}

	return s, nil
}

func (db *_ch) String() string {
	return "clickhouse"
}

func (db *_ch) Close() error {
	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()

		if db.conn != nil {
			_ = db.conn.Close()
		}
	}()

	wg.Wait()
	return nil
}

func (db *_ch) Ping(ctx context.Context) error {
	return db.conn.Ping(ctx)
}

func (db *_ch) Query(ctx context.Context, query string, args ...any) (driver.Rows, error) {
	return db.conn.Query(ctx, query, args...)
}