243 lines
8.2 KiB
Go
243 lines
8.2 KiB
Go
package mongo
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
mutil "git.sendico.io/tech/sxchnge/shared/config"
|
|
"git.sendico.io/tech/sxchnge/shared/database/internal/transaction"
|
|
"git.sendico.io/tech/sxchnge/shared/database/source"
|
|
"git.sendico.io/tech/sxchnge/shared/database/transaction"
|
|
dbvault "git.sendico.io/tech/sxchnge/shared/database/vault"
|
|
"git.sendico.io/tech/sxchnge/shared/logging"
|
|
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
|
|
"github.com/go-viper/mapstructure/v2"
|
|
"go.mongodb.org/mongo-driver/v2/mongo"
|
|
"go.mongodb.org/mongo-driver/v2/mongo/readpref"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// Config represents configuration
|
|
type Config struct {
|
|
Vault *dbvault.Config `mapstructure:"vault"`
|
|
Port *string `mapstructure:"port"`
|
|
PortEnv *string `mapstructure:"port_env"`
|
|
User *string `mapstructure:"user"`
|
|
UserEnv *string `mapstructure:"user_env"`
|
|
PasswordEnv string `mapstructure:"password_env"`
|
|
Database *string `mapstructure:"database"`
|
|
DatabaseEnv *string `mapstructure:"database_env"`
|
|
Host *string `mapstructure:"host"`
|
|
HostEnv *string `mapstructure:"host_env"`
|
|
Hosts []string `mapstructure:"hosts,omitempty"`
|
|
HostsEnvPrefix *string `mapstructure:"hosts_env_prefix,omitempty"`
|
|
HostsEnvPrefixEnv *string `mapstructure:"hosts_env_prefix_env,omitempty"`
|
|
PortsEnvPrefix *string `mapstructure:"ports_env_prefix,omitempty"`
|
|
PortsEnvPrefixEnv *string `mapstructure:"ports_env_prefix_env,omitempty"`
|
|
URI *string `mapstructure:"uri,omitempty"`
|
|
URIEnv *string `mapstructure:"uri_env,omitempty"`
|
|
AuthSource *string `mapstructure:"auth_source,omitempty"`
|
|
AuthSourceEnv *string `mapstructure:"auth_source_env,omitempty"`
|
|
AuthMechanism *string `mapstructure:"auth_mechanism,omitempty"`
|
|
AuthMechanismEnv *string `mapstructure:"auth_mechanism_env,omitempty"`
|
|
ReplicaSet *string `mapstructure:"replica_set,omitempty"`
|
|
ReplicaSetEnv *string `mapstructure:"replica_set_env,omitempty"`
|
|
}
|
|
|
|
type DBSettings struct {
|
|
Host string
|
|
Hosts []string
|
|
Port string
|
|
User string
|
|
Password string
|
|
Database string
|
|
AuthSource string
|
|
AuthMechanism string
|
|
ReplicaSet string
|
|
URI string
|
|
}
|
|
|
|
func Config2DBSettings(logger mlogger.Logger, config *Config) *DBSettings {
|
|
p := new(DBSettings)
|
|
p.Port = mutil.GetConfigValue(logger, "port", "port_env", config.Port, config.PortEnv)
|
|
p.Database = mutil.GetConfigValue(logger, "database", "database_env", config.Database, config.DatabaseEnv)
|
|
p.Password = os.Getenv(config.PasswordEnv)
|
|
p.User = mutil.GetConfigValue(logger, "user", "user_env", config.User, config.UserEnv)
|
|
p.Host = mutil.GetConfigValue(logger, "host", "host_env", config.Host, config.HostEnv)
|
|
p.AuthSource = mutil.GetConfigValue(logger, "auth_source", "auth_source_env", config.AuthSource, config.AuthSourceEnv)
|
|
p.AuthMechanism = mutil.GetConfigValue(logger, "auth_mechanism", "auth_mechanism_env", config.AuthMechanism, config.AuthMechanismEnv)
|
|
p.ReplicaSet = mutil.GetConfigValue(logger, "replica_set", "replica_set_env", config.ReplicaSet, config.ReplicaSetEnv)
|
|
p.URI = mutil.GetConfigValue(logger, "uri", "uri_env", config.URI, config.URIEnv)
|
|
|
|
hostPrefix := mutil.GetConfigValue(logger, "hosts_env_prefix", "hosts_env_prefix_env", config.HostsEnvPrefix, config.HostsEnvPrefixEnv)
|
|
portPrefix := mutil.GetConfigValue(logger, "ports_env_prefix", "ports_env_prefix_env", config.PortsEnvPrefix, config.PortsEnvPrefixEnv)
|
|
|
|
if hostPrefix == "" && p.ReplicaSet != "" {
|
|
hostPrefix = "MONGO_HOSTS_"
|
|
}
|
|
if portPrefix == "" && p.ReplicaSet != "" {
|
|
portPrefix = "MONGO_PORTS_"
|
|
}
|
|
|
|
p.Hosts = collectReplicaHosts(config.Hosts, p.ReplicaSet, p.Port, hostPrefix, portPrefix)
|
|
return p
|
|
}
|
|
|
|
func decodeConfig(logger mlogger.Logger, settings sharedmodel.SettingsT) (*Config, *DBSettings, error) {
|
|
var config Config
|
|
if err := mapstructure.Decode(settings, &config); err != nil {
|
|
logger.Warn("Failed to decode settings", zap.Error(err))
|
|
return nil, nil, err
|
|
}
|
|
dbSettings := Config2DBSettings(logger, &config)
|
|
return &config, dbSettings, nil
|
|
}
|
|
|
|
func dialMongo(logger mlogger.Logger, dbSettings *DBSettings) (*mongo.Client, error) {
|
|
opts := buildOptions(dbSettings)
|
|
|
|
client, err := mongo.Connect(opts)
|
|
if err != nil {
|
|
logger.Error("Unable to connect to database", zap.Error(err))
|
|
return nil, err
|
|
}
|
|
|
|
if dbSettings.URI != "" {
|
|
logger.Info("Connected successfully", zap.Bool("uri_provided", true))
|
|
} else {
|
|
logger.Info("Connected successfully", zap.String("user", dbSettings.User),
|
|
zap.Strings("hosts", opts.Hosts), zap.String("port", dbSettings.Port),
|
|
zap.String("database", dbSettings.Database), zap.String("replica_set", dbSettings.ReplicaSet))
|
|
}
|
|
|
|
if err := client.Ping(context.Background(), readpref.Primary()); err != nil {
|
|
logger.Error("Unable to ping database", zap.Error(err))
|
|
_ = client.Disconnect(context.Background())
|
|
return nil, err
|
|
}
|
|
|
|
return client, nil
|
|
}
|
|
|
|
// DB represents the structure of the database
|
|
type DB struct {
|
|
logger mlogger.Logger
|
|
database source.Source
|
|
close func(context.Context) error
|
|
}
|
|
|
|
func (db *DB) MongoDatabase() source.Source { return db.database }
|
|
func (db *DB) TransactionFactory() transaction.Factory {
|
|
return transactionimp.CreateSourceFactory(db.database)
|
|
}
|
|
|
|
// OpenSource owns either a fixed connection or a rotating Vault-backed source.
|
|
func OpenSource(logger mlogger.Logger, settings sharedmodel.SettingsT) (source.Source, func(context.Context) error, error) {
|
|
config, dbSettings, err := decodeConfig(logger, settings)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if config.Vault != nil {
|
|
issuer, err := dbvault.New(*config.Vault, dbSettings.URI)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
defer cancel()
|
|
manager, err := source.New(ctx, logger.Named("credentials"), dbSettings.Database, issuer)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return manager, manager.Close, nil
|
|
}
|
|
client, err := dialMongo(logger, dbSettings)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
return source.Fixed(client.Database(dbSettings.Database)), client.Disconnect, nil
|
|
}
|
|
|
|
func collectReplicaHosts(configuredHosts []string, replicaSet, defaultPort, hostPrefix, portPrefix string) []string {
|
|
normalize := func(host, port string) string {
|
|
host = strings.TrimSpace(host)
|
|
port = strings.TrimSpace(port)
|
|
if host == "" {
|
|
return ""
|
|
}
|
|
// If host already has a port, keep it; otherwise apply provided/default port.
|
|
if _, _, err := net.SplitHostPort(host); err == nil {
|
|
return host
|
|
}
|
|
if port != "" {
|
|
return net.JoinHostPort(host, port)
|
|
}
|
|
if defaultPort != "" {
|
|
return net.JoinHostPort(host, defaultPort)
|
|
}
|
|
return host
|
|
}
|
|
|
|
appendHost := func(list []string, host, port string) []string {
|
|
if normalized := normalize(host, port); normalized != "" {
|
|
return append(list, normalized)
|
|
}
|
|
return list
|
|
}
|
|
|
|
var hosts []string
|
|
for _, h := range configuredHosts {
|
|
hosts = appendHost(hosts, h, "")
|
|
}
|
|
|
|
if replicaSet == "" || hostPrefix == "" {
|
|
return hosts
|
|
}
|
|
|
|
index := 0
|
|
for {
|
|
hostEnv := os.Getenv(fmt.Sprintf("%s%d", hostPrefix, index))
|
|
portEnv := ""
|
|
if portPrefix != "" {
|
|
portEnv = os.Getenv(fmt.Sprintf("%s%d", portPrefix, index))
|
|
}
|
|
|
|
if hostEnv == "" && index == 0 {
|
|
hostEnv = os.Getenv(fmt.Sprintf("%s%d", hostPrefix, 1))
|
|
if portPrefix != "" {
|
|
portEnv = os.Getenv(fmt.Sprintf("%s%d", portPrefix, 1))
|
|
}
|
|
if hostEnv == "" {
|
|
break
|
|
}
|
|
index = 1
|
|
} else if hostEnv == "" {
|
|
break
|
|
}
|
|
|
|
hosts = appendHost(hosts, hostEnv, portEnv)
|
|
index++
|
|
}
|
|
|
|
return hosts
|
|
}
|
|
|
|
func (db *DB) CloseConnection() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
defer cancel()
|
|
if err := db.close(ctx); err != nil {
|
|
db.logger.Warn("Failed to drain database connection", zap.Error(err))
|
|
}
|
|
}
|
|
|
|
func NewConnection(logger mlogger.Logger, settings sharedmodel.SettingsT) (*DB, error) {
|
|
database, close, err := OpenSource(logger, settings)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &DB{logger: logger, database: database, close: close}, nil
|
|
}
|