Feature: add bun as ORM and test

Author: aopoltorzhicky

database/bun.go

@@ -0,0 +1,112 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/dipdup-net/go-lib/config"
+	"github.com/pkg/errors"
+	"github.com/uptrace/bun"
+	"github.com/uptrace/bun/dialect/pgdialect"
+	"github.com/uptrace/bun/driver/pgdriver"
+)
+
+// Bun -
+type Bun struct {
+	sqldb *sql.DB
+	conn  *bun.DB
+}
+
+// NewBun -
+func NewBun() *Bun {
+	return new(Bun)
+}
+
+// DB -
+func (db *Bun) DB() *bun.DB {
+	return db.conn
+}
+
+// Connect -
+func (db *Bun) Connect(ctx context.Context, cfg config.Database) error {
+	if cfg.Kind != config.DBKindPostgres {
+		return errors.Wrap(ErrUnsupportedDatabaseType, cfg.Kind)
+	}
+	if cfg.Path != "" {
+		db.sqldb = sql.OpenDB(pgdriver.NewConnector(pgdriver.WithDSN(cfg.Path)))
+		db.conn = bun.NewDB(db.sqldb, pgdialect.New())
+	} else {
+		db.sqldb = sql.OpenDB(pgdriver.NewConnector(
+			pgdriver.WithAddr(fmt.Sprintf("%s:%d", cfg.Host, cfg.Port)),
+			pgdriver.WithDatabase(cfg.Database),
+			pgdriver.WithPassword(cfg.Password),
+			pgdriver.WithUser(cfg.User),
+			pgdriver.WithInsecure(true),
+		))
+		db.conn = bun.NewDB(db.sqldb, pgdialect.New())
+	}
+	return nil
+}
+
+// Close -
+func (db *Bun) Close() error {
+	if err := db.conn.Close(); err != nil {
+		return err
+	}
+	return db.sqldb.Close()
+}
+
+// Ping -
+func (db *Bun) Ping(ctx context.Context) error {
+	if db.conn == nil {
+		return ErrConnectionIsNotInitialized
+	}
+	return db.conn.PingContext(ctx)
+}
+
+// State -
+func (db *Bun) State(ctx context.Context, indexName string) (*State, error) {
+	var s State
+	err := db.conn.NewSelect().Model(&s).Where("index_name = ?", indexName).Limit(1).Scan(ctx)
+	return &s, err
+}
+
+// CreateState -
+func (db *Bun) CreateState(ctx context.Context, s *State) error {
+	_, err := db.conn.NewInsert().Model(s).Exec(ctx)
+	return err
+}
+
+// UpdateState -
+func (db *Bun) UpdateState(ctx context.Context, s *State) error {
+	_, err := db.conn.NewUpdate().Model(s).Where("index_name = ?", s.IndexName).Exec(ctx)
+	return err
+}
+
+// DeleteState -
+func (db *Bun) DeleteState(ctx context.Context, s *State) error {
+	_, err := db.conn.NewDelete().Model(s).Where("index_name = ?", s.IndexName).Exec(ctx)
+	return err
+}
+
+// MakeTableComment -
+func (db *Bun) MakeTableComment(ctx context.Context, name string, comment string) error {
+	_, err := db.conn.ExecContext(ctx,
+		`COMMENT ON TABLE ? IS ?`,
+		bun.Ident(name),
+		comment)
+
+	return err
+}
+
+// MakeColumnComment -
+func (db *Bun) MakeColumnComment(ctx context.Context, tableName string, columnName string, comment string) error {
+	_, err := db.conn.ExecContext(ctx,
+		`COMMENT ON COLUMN ?.? IS ?`,
+		bun.Ident(tableName),
+		bun.Ident(columnName),
+		comment)
+
+	return err
+}

database/comment.go

@@ -2,12 +2,14 @@ package database
 
 import (
 	"context"
-	"github.com/dipdup-net/go-lib/hasura"
-	"github.com/pkg/errors"
 	"reflect"
 	"strings"
+
+	"github.com/dipdup-net/go-lib/hasura"
+	"github.com/pkg/errors"
 )
 
+// MakeComments -
 func MakeComments(ctx context.Context, sc SchemeCommenter, models ...interface{}) error {
 	if models == nil {
 		return nil

database/container.go

@@ -0,0 +1,79 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/docker/go-connections/nat"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+type (
+	PostgreSQLContainer struct {
+		testcontainers.Container
+		cfg        PostgreSQLContainerConfig
+		mappedPort nat.Port
+	}
+
+	PostgreSQLContainerConfig struct {
+		User     string
+		Password string
+		Database string
+		Host     string
+		Port     int
+		Image    string
+	}
+)
+
+// NewPostgreSQLContainer -
+func NewPostgreSQLContainer(ctx context.Context, cfg PostgreSQLContainerConfig) (*PostgreSQLContainer, error) {
+	if cfg.Port == 0 {
+		cfg.Port = 5432
+	}
+	port := fmt.Sprintf("%d/tcp", cfg.Port)
+
+	if cfg.Image == "" {
+		cfg.Image = "postgres:15"
+	}
+
+	container, err := postgres.RunContainer(ctx,
+		testcontainers.WithImage(cfg.Image),
+		postgres.WithDatabase(cfg.Database),
+		postgres.WithUsername(cfg.User),
+		postgres.WithPassword(cfg.Password),
+		testcontainers.WithWaitStrategy(wait.ForLog("database system is ready to accept connections").WithOccurrence(2).WithStartupTimeout(5*time.Second)),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	host, err := container.Host(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("getting host for: %w", err)
+	}
+	cfg.Host = host
+
+	mappedPort, err := container.MappedPort(ctx, nat.Port(port))
+	if err != nil {
+		return nil, fmt.Errorf("getting mapped port for %s: %w", port, err)
+	}
+
+	return &PostgreSQLContainer{
+		Container:  container,
+		cfg:        cfg,
+		mappedPort: mappedPort,
+	}, nil
+}
+
+// GetDSN -
+func (c PostgreSQLContainer) GetDSN() string {
+	return fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable", c.cfg.User, c.cfg.Password, c.cfg.Host, c.mappedPort.Port(), c.cfg.Database)
+}
+
+// MappedPort -
+func (c PostgreSQLContainer) MappedPort() nat.Port {
+	return c.mappedPort
+}

database/db.go

@@ -11,11 +11,18 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+// SchemeCommenter -
+type SchemeCommenter interface {
+	MakeTableComment(ctx context.Context, name string, comment string) error
+	MakeColumnComment(ctx context.Context, tableName string, columnName string, comment string) error
+}
+
 // Database -
 type Database interface {
 	Connect(ctx context.Context, cfg config.Database) error
 
 	StateRepository
+	SchemeCommenter
 
 	driver.Pinger
 	io.Closer