pkgdash/internal/database/database.go

package database

import (
	"context"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"time"

	appconfig "git.unistack.org/unistack-org/pkgdash/internal/config"
	"github.com/golang-migrate/migrate/v4"
	"github.com/golang-migrate/migrate/v4/database"
	mpgx "github.com/golang-migrate/migrate/v4/database/pgx"
	msqlite "github.com/golang-migrate/migrate/v4/database/sqlite"
	"github.com/golang-migrate/migrate/v4/source/iofs"
	"github.com/jackc/pgx/v4"
	"github.com/jackc/pgx/v4/stdlib"
	"github.com/jmoiron/sqlx"
	"go.unistack.org/micro/v4/logger"
	_ "modernc.org/sqlite"
)

func ParseDSN(cfg *appconfig.DatabaseConfig) error {
	var err error

	u, err := url.Parse(cfg.DSN)
	if err != nil {
		return err
	}

	values := u.Query()
	var value string

	if value = values.Get("conn_max"); value != "" {
		values.Del("conn_max")
		maxOpenConns, err := strconv.Atoi(value)
		if err != nil {
			return err
		}
		cfg.MaxOpenConns = maxOpenConns
		cfg.MaxIdleConns = maxOpenConns / 2
	}

	if value = values.Get("conn_maxidle"); value != "" {
		values.Del("conn_maxidle")
		maxIdleConns, err := strconv.Atoi(value)
		if err != nil {
			return err
		}
		cfg.MaxIdleConns = maxIdleConns
	}

	if value = values.Get("conn_lifetime"); value != "" {
		values.Del("conn_lifetime")
		connMaxLifetime, err := time.ParseDuration(value)
		if err != nil {
			return err
		}
		cfg.ConnMaxLifetime = connMaxLifetime
	}

	if value = values.Get("conn_maxidletime"); value != "" {
		values.Del("conn_maxidletime")
		connMaxIdleTime, err := time.ParseDuration(value)
		if err != nil {
			return err
		}
		cfg.ConnMaxIdleTime = connMaxIdleTime
	}

	if mtype := values.Get("migrate"); mtype != "" {
		values.Del("migrate")
		cfg.Migrate = mtype
	}

	switch u.Scheme {
	case "postgres", "pgsql", "postgresql":
		u.Scheme = "postgres"
	case "sqlite", "sqlite3":
		u.Scheme = "sqlite"
	default:
		return fmt.Errorf("unknown database %s", u.Scheme)
	}

	cfg.Type = u.Scheme
	u.RawQuery = values.Encode()

	cfg.ConnStr = u.String()

	return nil
}

func connect(ctx context.Context, cfg *appconfig.DatabaseConfig, log logger.Logger) (*sqlx.DB, error) {
	var db *sqlx.DB
	var err error

	log.Infof(ctx, "connect to %s", cfg.Type)
	switch cfg.Type {
	case "postgres", "pgsql", "postgresql":
		db, err = connectPostgres(ctx, cfg.ConnStr)
		cfg.Type = "postgres"
	case "sqlite", "sqlite3":
		db, err = connectSqlite(ctx, cfg.ConnStr)
		cfg.Type = "sqlite"
	default:
		return nil, fmt.Errorf("unknown database type %s", cfg.Type)
	}

	if err != nil {
		return nil, err
	}

	return db, nil
}

func Connect(ctx context.Context, cfg *appconfig.DatabaseConfig, log logger.Logger) (*sqlx.DB, error) {
	db, err := connect(ctx, cfg, log)
	if err != nil {
		return nil, err
	}
	m, err := migratePrepare(ctx, db, log, cfg.Type)
	if err != nil {
		return nil, err
	}

	switch cfg.Migrate {
	case "":
		break
	case "up":
		logger.Infof(ctx, "migrate up")
		err = m.Up()
	case "down":
		logger.Infof(ctx, "migrate down")
		err = m.Down()
	case "seed":
		logger.Infof(ctx, "migrate seed")
		if err = m.Drop(); err == nil {
			err = m.Up()
		}
	default:
		logger.Infof(ctx, "migrate version")
		v, verr := strconv.ParseUint(cfg.Type, 10, 64)
		if verr != nil {
			return nil, err
		}
		err = m.Migrate(uint(v))
	}

	if err == nil || err == migrate.ErrNoChange {
		srcerr, dberr := m.Close()
		if srcerr != nil {
			err = srcerr
		} else if dberr != nil {
			err = dberr
		} else {
			err = nil
		}
	}

	if err == nil {
		db, err = connect(ctx, cfg, log)
	}

	if err != nil {
		return nil, err
	}

	db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime)
	db.SetConnMaxLifetime(cfg.ConnMaxLifetime)
	db.SetMaxIdleConns(cfg.MaxIdleConns)
	db.SetMaxOpenConns(cfg.MaxOpenConns)

	return db, nil
}

func connectSqlite(ctx context.Context, connstr string) (*sqlx.DB, error) {
	if !strings.Contains(connstr, ":memory:") {
		return sqlx.ConnectContext(ctx, "sqlite", "file:"+connstr[9:])
	}
	return sqlx.ConnectContext(ctx, "sqlite", connstr[9:])
}

func connectPostgres(ctx context.Context, connstr string) (*sqlx.DB, error) {
	// parse connection string
	dbConf, err := pgx.ParseConfig(connstr)
	if err != nil {
		return nil, err
	}

	// needed for pgbouncer
	dbConf.RuntimeParams = map[string]string{
		"standard_conforming_strings": "on",
		"application_name":            "authn",
	}

	// may be needed for pbbouncer, needs to check
	// dbConf.PreferSimpleProtocol = true
	// register pgx conn
	connStr := stdlib.RegisterConnConfig(dbConf)

	db, err := sqlx.ConnectContext(ctx, "pgx", connStr)
	if err != nil {
		return nil, err
	}

	return db, nil
}

func migratePrepare(ctx context.Context, db *sqlx.DB, log logger.Logger, dbtype string) (*migrate.Migrate, error) {
	var driver database.Driver
	var err error

	switch dbtype {
	case "postgres":
		driver, err = mpgx.WithInstance(db.DB, &mpgx.Config{
			DatabaseName:    "pkgdash",
			MigrationsTable: "schema_migrations",
		})
	case "sqlite":
		driver, err = msqlite.WithInstance(db.DB, &msqlite.Config{
			DatabaseName:    "pkgdash",
			MigrationsTable: "schema_migrations",
		})
	}

	if err != nil {
		return nil, err
	}
	source, err := iofs.New(assets, "migrations/"+dbtype)
	if err != nil {
		return nil, err
	}
	m, err := migrate.NewWithInstance("fs", source, "apigw", driver)
	if err != nil {
		return nil, err
	}
	m.Log = &mLog{ctx: ctx, l: log}
	return m, nil
}

type mLog struct {
	ctx context.Context
	l   logger.Logger
}

func (l *mLog) Verbose() bool {
	return l.l.V(logger.DebugLevel)
}

func (l *mLog) Printf(format string, v ...interface{}) {
	l.l.Infof(l.ctx, format, v...)
}