customer/pkg/mongo/connection.go

package mongo

import (
	"context"
	"fmt"
	"log"
	"net"
	"net/url"
	"time"

	"go.mongodb.org/mongo-driver/event"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

type ConnectDeps struct {
	Configuration *Configuration
	Timeout       time.Duration
}

func Connect(ctx context.Context, deps *ConnectDeps) (*mongo.Database, error) {
	if deps == nil {
		return nil, ErrEmptyArgs
	}

	mongoURI := &url.URL{
		Scheme: "mongodb",
		Host:   net.JoinHostPort(deps.Configuration.Host, deps.Configuration.Port),
	}

	cmdMonitor := &event.CommandMonitor{
		Started: func(_ context.Context, evt *event.CommandStartedEvent) {
			log.Println(evt.Command)
		},
		Succeeded: func(_ context.Context, evt *event.CommandSucceededEvent) {
			log.Println(evt.Reply)
		},
		Failed: func(_ context.Context, evt *event.CommandFailedEvent) {
			log.Println(evt.Failure)
		},
	}

	connectionOptions := options.Client().
		ApplyURI(mongoURI.String()).
		SetAuth(options.Credential{
			AuthMechanism: "SCRAM-SHA-1",
			AuthSource:    deps.Configuration.Auth,
			Username:      deps.Configuration.User,
			Password:      deps.Configuration.Password,
		}).
		SetMonitor(cmdMonitor)

	fmt.Println(connectionOptions.GetURI())

	ticker := time.NewTicker(1 * time.Second)
	timeoutExceeded := time.After(deps.Timeout)

	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			connection, err := mongo.Connect(ctx, connectionOptions)
			if err == nil {
				return connection.Database(deps.Configuration.DatabaseName), nil
			}

			log.Printf("failed to connect to db <%s>: %s", mongoURI.String(), err.Error())
		case <-timeoutExceeded:
			return nil, fmt.Errorf("db connection <%s> failed after %d timeout", mongoURI.String(), deps.Timeout)
		default:
			time.Sleep(1 * time.Second)
		}
	}
}