upstream-corteza/server/store/adapters/rdbms/connect.go

package rdbms

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/cortezaproject/corteza/server/pkg/sentry"
	"github.com/jmoiron/sqlx"
	"go.uber.org/zap"
)

const (
	// TxRetryHardLimit is the absolute maximum retries we'll allow
	TxRetryHardLimit = 100

	DefaultSliceCapacity = 1000

	MinEnsureFetchLimit = 10

	// Upper bound for the widening fetch window used when the check fn
	// rejects entire batches
	MaxEnsureFetchLimit = 1000
	MaxRefetches        = 100

	MaxLimit = 1000
)

// Connect is rdbms' package connector
//
// Function is called from (store) driver's required connection function
// to open connection and bind it with
func Connect(ctx context.Context, log *zap.Logger, cfg *ConnConfig) (db *sqlx.DB, err error) {
	var (
		connErrCh = make(chan error, 1)
		patience  = time.Now().Add(cfg.ConnTryPatience)
		base      *sql.DB
	)

	log = log.Named("store")

	if base, err = sql.Open(cfg.DriverName, cfg.DataSourceName); err != nil {
		return
	}

	db = sqlx.NewDb(base, cfg.DriverName)
	log.Debug(
		"setting database connection parameters",
		zap.Int("MaxOpenConns", cfg.MaxOpenConns),
		zap.Duration("MaxLifetime", cfg.ConnMaxLifetime),
		zap.Int("MaxIdleConns", cfg.MaxIdleConns),

		// log DSN with masked username and password
		zap.String("DSN", cfg.MaskedDSN),
	)

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

	go func() {
		defer sentry.Recover()

		var (
			err error
			try = 0

			// Make a small adjustment when
			// collecting callers from the callstack for this
			log = log.WithOptions(zap.AddCallerSkip(-2))
		)

		for {
			try++

			if cfg.ConnTryMax <= try {
				connErrCh <- fmt.Errorf("could not connect in %d tries", try)
				return
			}

			if err = db.PingContext(ctx); err != nil {

				if time.Now().After(patience) {
					// don't make too much fuss
					// if we're in patience mode
					log.Warn(
						"could not connect to the database",
						zap.Error(err),
						zap.Int("try", try),
						zap.Float64("delay", cfg.ConnTryBackoffDelay.Seconds()),
					)
				}

				select {
				case <-ctx.Done():
					// Forced break
					break
				case <-time.After(cfg.ConnTryBackoffDelay):
					//	Wait before next try
					continue
				}
			}

			log.Debug("connected to the database")
			break
		}

		connErrCh <- err
	}()

	to := cfg.ConnTryTimeout * time.Duration(cfg.ConnTryMax*2)
	select {
	case err = <-connErrCh:
		return
	case <-time.After(to):
		// Wait before next try
		return nil, fmt.Errorf("timedout after %.2fs", to.Seconds())
	case <-ctx.Done():
		return nil, fmt.Errorf("connection cancelled")
	}
}

// Connect is called from the adapter's Connect function
//
// It is intentionally not compatible with store.ConnectorFn
// and can not be used to register
//func Connect(ctx context.Context, cfg *ConnConfig) (s *Store, err error) {
//	return s, func() error {
//		if err = cfg.ParseExtra(); err != nil {
//			return err
//		}
//
//		cfg.SetDefaults()
//		s = &Store{
//			config: cfg,
//			//schema: ddl.SchemaAPI(s.DB(), ddl.NewCommonDialect()),
//		}
//
//		return s.Connect(ctx)
//	}()
//}

func dbHealthcheck(db *sqlx.DB) func(ctx context.Context) error {
	return db.PingContext
}