first commit

2025-09-24 19:13:54 +07:00
commit d29f72a40b
46 changed files with 13616 additions and 0 deletions
@@ -0,0 +1,699 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log" // Import runtime package
+
+	// Import debug package
+	"strconv"
+	"sync"
+	"time"
+
+	"api-service/internal/config"
+
+	_ "github.com/jackc/pgx/v5" // Import pgx driver
+	"github.com/lib/pq"
+	_ "gorm.io/driver/postgres" // Import GORM PostgreSQL driver
+
+	_ "github.com/go-sql-driver/mysql" // MySQL driver for database/sql
+	_ "gorm.io/driver/mysql"           // GORM MySQL driver
+	_ "gorm.io/driver/sqlserver"       // GORM SQL Server driver
+
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+// DatabaseType represents supported database types
+type DatabaseType string
+
+const (
+	Postgres  DatabaseType = "postgres"
+	MySQL     DatabaseType = "mysql"
+	SQLServer DatabaseType = "sqlserver"
+	SQLite    DatabaseType = "sqlite"
+	MongoDB   DatabaseType = "mongodb"
+)
+
+// Service represents a service that interacts with multiple databases
+type Service interface {
+	Health() map[string]map[string]string
+	GetDB(name string) (*sql.DB, error)
+	GetMongoClient(name string) (*mongo.Client, error)
+	GetReadDB(name string) (*sql.DB, error) // For read replicas
+	Close() error
+	ListDBs() []string
+	GetDBType(name string) (DatabaseType, error)
+	// Tambahkan method untuk WebSocket notifications
+	ListenForChanges(ctx context.Context, dbName string, channels []string, callback func(string, string)) error
+	NotifyChange(dbName, channel, payload string) error
+	GetPrimaryDB(name string) (*sql.DB, error) // Helper untuk get primary DB
+}
+
+type service struct {
+	sqlDatabases map[string]*sql.DB
+	mongoClients map[string]*mongo.Client
+	readReplicas map[string][]*sql.DB // Read replicas for load balancing
+	configs      map[string]config.DatabaseConfig
+	readConfigs  map[string][]config.DatabaseConfig
+	mu           sync.RWMutex
+	readBalancer map[string]int          // Round-robin counter for read replicas
+	listeners    map[string]*pq.Listener // Tambahkan untuk tracking listeners
+	listenersMu  sync.RWMutex
+}
+
+var (
+	dbManager *service
+	once      sync.Once
+)
+
+// New creates a new database service with multiple connections
+func New(cfg *config.Config) Service {
+	once.Do(func() {
+		dbManager = &service{
+			sqlDatabases: make(map[string]*sql.DB),
+			mongoClients: make(map[string]*mongo.Client),
+			readReplicas: make(map[string][]*sql.DB),
+			configs:      make(map[string]config.DatabaseConfig),
+			readConfigs:  make(map[string][]config.DatabaseConfig),
+			readBalancer: make(map[string]int),
+			listeners:    make(map[string]*pq.Listener),
+		}
+
+		log.Println("Initializing database service...") // Log when the initialization starts
+		// log.Printf("Current Goroutine ID: %d", runtime.NumGoroutine()) // Log the number of goroutines
+		// log.Printf("Stack Trace: %s", debug.Stack()) // Log the stack trace
+		dbManager.loadFromConfig(cfg)
+
+		// Initialize all databases
+		for name, dbConfig := range dbManager.configs {
+			if err := dbManager.addDatabase(name, dbConfig); err != nil {
+				log.Printf("Failed to connect to database %s: %v", name, err)
+			}
+		}
+
+		// Initialize read replicas
+		for name, replicaConfigs := range dbManager.readConfigs {
+			for i, replicaConfig := range replicaConfigs {
+				if err := dbManager.addReadReplica(name, i, replicaConfig); err != nil {
+					log.Printf("Failed to connect to read replica %s[%d]: %v", name, i, err)
+				}
+			}
+		}
+	})
+
+	return dbManager
+}
+
+func (s *service) loadFromConfig(cfg *config.Config) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Load primary databases
+	for name, dbConfig := range cfg.Databases {
+		s.configs[name] = dbConfig
+	}
+
+	// Load read replicas
+	for name, replicaConfigs := range cfg.ReadReplicas {
+		s.readConfigs[name] = replicaConfigs
+	}
+}
+
+func (s *service) addDatabase(name string, config config.DatabaseConfig) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	log.Printf("=== Database Connection Debug ===")
+	// log.Printf("Database: %s", name)
+	// log.Printf("Type: %s", config.Type)
+	// log.Printf("Host: %s", config.Host)
+	// log.Printf("Port: %d", config.Port)
+	// log.Printf("Database: %s", config.Database)
+	// log.Printf("Username: %s", config.Username)
+	// log.Printf("SSLMode: %s", config.SSLMode)
+
+	var db *sql.DB
+	var err error
+
+	dbType := DatabaseType(config.Type)
+
+	switch dbType {
+	case Postgres:
+		db, err = s.openPostgresConnection(config)
+	case MySQL:
+		db, err = s.openMySQLConnection(config)
+	case SQLServer:
+		db, err = s.openSQLServerConnection(config)
+	case SQLite:
+		db, err = s.openSQLiteConnection(config)
+	case MongoDB:
+		return s.addMongoDB(name, config)
+	default:
+		return fmt.Errorf("unsupported database type: %s", config.Type)
+	}
+
+	if err != nil {
+		log.Printf("❌ Error connecting to database %s: %v", name, err)
+		log.Printf("   Database: %s@%s:%d/%s", config.Username, config.Host, config.Port, config.Database)
+		return err
+	}
+
+	log.Printf("✅ Successfully connected to database: %s", name)
+	return s.configureSQLDB(name, db, config.MaxOpenConns, config.MaxIdleConns, config.ConnMaxLifetime)
+}
+
+func (s *service) addReadReplica(name string, index int, config config.DatabaseConfig) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var db *sql.DB
+	var err error
+
+	dbType := DatabaseType(config.Type)
+
+	switch dbType {
+	case Postgres:
+		db, err = s.openPostgresConnection(config)
+	case MySQL:
+		db, err = s.openMySQLConnection(config)
+	case SQLServer:
+		db, err = s.openSQLServerConnection(config)
+	case SQLite:
+		db, err = s.openSQLiteConnection(config)
+	default:
+		return fmt.Errorf("unsupported database type for read replica: %s", config.Type)
+	}
+
+	if err != nil {
+		return err
+	}
+
+	if s.readReplicas[name] == nil {
+		s.readReplicas[name] = make([]*sql.DB, 0)
+	}
+
+	// Ensure we have enough slots
+	for len(s.readReplicas[name]) <= index {
+		s.readReplicas[name] = append(s.readReplicas[name], nil)
+	}
+
+	s.readReplicas[name][index] = db
+	log.Printf("Successfully connected to read replica %s[%d]", name, index)
+
+	return nil
+}
+
+func (s *service) openPostgresConnection(config config.DatabaseConfig) (*sql.DB, error) {
+	connStr := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=%s",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database,
+		config.SSLMode,
+	)
+
+	if config.Schema != "" {
+		connStr += "&search_path=" + config.Schema
+	}
+
+	db, err := sql.Open("pgx", connStr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open PostgreSQL connection: %w", err)
+	}
+
+	return db, nil
+}
+
+func (s *service) openMySQLConnection(config config.DatabaseConfig) (*sql.DB, error) {
+	connStr := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database,
+	)
+
+	db, err := sql.Open("mysql", connStr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open MySQL connection: %w", err)
+	}
+
+	return db, nil
+}
+
+func (s *service) openSQLServerConnection(config config.DatabaseConfig) (*sql.DB, error) {
+	connStr := fmt.Sprintf("sqlserver://%s:%s@%s:%d?database=%s",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database,
+	)
+
+	db, err := sql.Open("sqlserver", connStr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open SQL Server connection: %w", err)
+	}
+
+	return db, nil
+}
+
+func (s *service) openSQLiteConnection(config config.DatabaseConfig) (*sql.DB, error) {
+	dbPath := config.Path
+	if dbPath == "" {
+		dbPath = fmt.Sprintf("./data/%s.db", config.Database)
+	}
+
+	db, err := sql.Open("sqlite3", dbPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open SQLite connection: %w", err)
+	}
+
+	return db, nil
+}
+
+func (s *service) addMongoDB(name string, config config.DatabaseConfig) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	uri := fmt.Sprintf("mongodb://%s:%s@%s:%d/%s",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database,
+	)
+
+	client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
+	if err != nil {
+		return fmt.Errorf("failed to connect to MongoDB: %w", err)
+	}
+
+	s.mongoClients[name] = client
+	log.Printf("Successfully connected to MongoDB: %s", name)
+
+	return nil
+}
+
+func (s *service) configureSQLDB(name string, db *sql.DB, maxOpenConns, maxIdleConns int, connMaxLifetime time.Duration) error {
+	db.SetMaxOpenConns(maxOpenConns)
+	db.SetMaxIdleConns(maxIdleConns)
+	db.SetConnMaxLifetime(connMaxLifetime)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	if err := db.PingContext(ctx); err != nil {
+		db.Close()
+		return fmt.Errorf("failed to ping database: %w", err)
+	}
+
+	s.sqlDatabases[name] = db
+	log.Printf("Successfully connected to SQL database: %s", name)
+
+	return nil
+}
+
+// Health checks the health of all database connections by pinging each database.
+func (s *service) Health() map[string]map[string]string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	result := make(map[string]map[string]string)
+
+	// Check SQL databases
+	for name, db := range s.sqlDatabases {
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+
+		stats := make(map[string]string)
+
+		err := db.PingContext(ctx)
+		if err != nil {
+			stats["status"] = "down"
+			stats["error"] = fmt.Sprintf("db down: %v", err)
+			stats["type"] = "sql"
+			stats["role"] = "primary"
+			result[name] = stats
+			continue
+		}
+
+		stats["status"] = "up"
+		stats["message"] = "It's healthy"
+		stats["type"] = "sql"
+		stats["role"] = "primary"
+
+		dbStats := db.Stats()
+		stats["open_connections"] = strconv.Itoa(dbStats.OpenConnections)
+		stats["in_use"] = strconv.Itoa(dbStats.InUse)
+		stats["idle"] = strconv.Itoa(dbStats.Idle)
+		stats["wait_count"] = strconv.FormatInt(dbStats.WaitCount, 10)
+		stats["wait_duration"] = dbStats.WaitDuration.String()
+		stats["max_idle_closed"] = strconv.FormatInt(dbStats.MaxIdleClosed, 10)
+		stats["max_lifetime_closed"] = strconv.FormatInt(dbStats.MaxLifetimeClosed, 10)
+
+		if dbStats.OpenConnections > 40 {
+			stats["message"] = "The database is experiencing heavy load."
+		}
+
+		if dbStats.WaitCount > 1000 {
+			stats["message"] = "The database has a high number of wait events, indicating potential bottlenecks."
+		}
+
+		if dbStats.MaxIdleClosed > int64(dbStats.OpenConnections)/2 {
+			stats["message"] = "Many idle connections are being closed, consider revising the connection pool settings."
+		}
+
+		if dbStats.MaxLifetimeClosed > int64(dbStats.OpenConnections)/2 {
+			stats["message"] = "Many connections are being closed due to max lifetime, consider increasing max lifetime or revising the connection usage pattern."
+		}
+
+		result[name] = stats
+	}
+
+	// Check read replicas
+	for name, replicas := range s.readReplicas {
+		for i, db := range replicas {
+			if db == nil {
+				continue
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+			defer cancel()
+
+			replicaName := fmt.Sprintf("%s_replica_%d", name, i)
+			stats := make(map[string]string)
+
+			err := db.PingContext(ctx)
+			if err != nil {
+				stats["status"] = "down"
+				stats["error"] = fmt.Sprintf("read replica down: %v", err)
+				stats["type"] = "sql"
+				stats["role"] = "replica"
+				result[replicaName] = stats
+				continue
+			}
+
+			stats["status"] = "up"
+			stats["message"] = "Read replica healthy"
+			stats["type"] = "sql"
+			stats["role"] = "replica"
+
+			dbStats := db.Stats()
+			stats["open_connections"] = strconv.Itoa(dbStats.OpenConnections)
+			stats["in_use"] = strconv.Itoa(dbStats.InUse)
+			stats["idle"] = strconv.Itoa(dbStats.Idle)
+
+			result[replicaName] = stats
+		}
+	}
+
+	// Check MongoDB connections
+	for name, client := range s.mongoClients {
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+
+		stats := make(map[string]string)
+
+		err := client.Ping(ctx, nil)
+		if err != nil {
+			stats["status"] = "down"
+			stats["error"] = fmt.Sprintf("mongodb down: %v", err)
+			stats["type"] = "mongodb"
+			result[name] = stats
+			continue
+		}
+
+		stats["status"] = "up"
+		stats["message"] = "It's healthy"
+		stats["type"] = "mongodb"
+
+		result[name] = stats
+	}
+
+	return result
+}
+
+// GetDB returns a specific SQL database connection by name
+func (s *service) GetDB(name string) (*sql.DB, error) {
+	log.Printf("Attempting to get database connection for: %s", name)
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	db, exists := s.sqlDatabases[name]
+	if !exists {
+		log.Printf("Error: database %s not found", name) // Log the error
+		return nil, fmt.Errorf("database %s not found", name)
+	}
+
+	log.Printf("Current connection pool state for %s: Open: %d, In Use: %d, Idle: %d",
+		name, db.Stats().OpenConnections, db.Stats().InUse, db.Stats().Idle)
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	// db, exists := s.sqlDatabases[name]
+	// if !exists {
+	// 	log.Printf("Error: database %s not found", name) // Log the error
+	// 	return nil, fmt.Errorf("database %s not found", name)
+	// }
+
+	return db, nil
+}
+
+// GetReadDB returns a read replica connection using round-robin load balancing
+func (s *service) GetReadDB(name string) (*sql.DB, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	replicas, exists := s.readReplicas[name]
+	if !exists || len(replicas) == 0 {
+		// Fallback to primary if no replicas available
+		return s.GetDB(name)
+	}
+
+	// Round-robin load balancing
+	s.readBalancer[name] = (s.readBalancer[name] + 1) % len(replicas)
+	selected := replicas[s.readBalancer[name]]
+
+	if selected == nil {
+		// Fallback to primary if replica is nil
+		return s.GetDB(name)
+	}
+
+	return selected, nil
+}
+
+// GetMongoClient returns a specific MongoDB client by name
+func (s *service) GetMongoClient(name string) (*mongo.Client, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	client, exists := s.mongoClients[name]
+	if !exists {
+		return nil, fmt.Errorf("MongoDB client %s not found", name)
+	}
+
+	return client, nil
+}
+
+// ListDBs returns list of available database names
+func (s *service) ListDBs() []string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	names := make([]string, 0, len(s.sqlDatabases)+len(s.mongoClients))
+
+	for name := range s.sqlDatabases {
+		names = append(names, name)
+	}
+
+	for name := range s.mongoClients {
+		names = append(names, name)
+	}
+
+	return names
+}
+
+// GetDBType returns the type of a specific database
+func (s *service) GetDBType(name string) (DatabaseType, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	config, exists := s.configs[name]
+	if !exists {
+		return "", fmt.Errorf("database %s not found", name)
+	}
+
+	return DatabaseType(config.Type), nil
+}
+
+// Close closes all database connections
+func (s *service) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var errs []error
+
+	for name, db := range s.sqlDatabases {
+		if err := db.Close(); err != nil {
+			errs = append(errs, fmt.Errorf("failed to close database %s: %w", name, err))
+		} else {
+			log.Printf("Disconnected from SQL database: %s", name)
+		}
+	}
+
+	for name, replicas := range s.readReplicas {
+		for i, db := range replicas {
+			if db != nil {
+				if err := db.Close(); err != nil {
+					errs = append(errs, fmt.Errorf("failed to close read replica %s[%d]: %w", name, i, err))
+				} else {
+					log.Printf("Disconnected from read replica: %s[%d]", name, i)
+				}
+			}
+		}
+	}
+
+	for name, client := range s.mongoClients {
+		if err := client.Disconnect(context.Background()); err != nil {
+			errs = append(errs, fmt.Errorf("failed to disconnect MongoDB client %s: %w", name, err))
+		} else {
+			log.Printf("Disconnected from MongoDB: %s", name)
+		}
+	}
+
+	s.sqlDatabases = make(map[string]*sql.DB)
+	s.mongoClients = make(map[string]*mongo.Client)
+	s.readReplicas = make(map[string][]*sql.DB)
+	s.configs = make(map[string]config.DatabaseConfig)
+	s.readConfigs = make(map[string][]config.DatabaseConfig)
+
+	if len(errs) > 0 {
+		return fmt.Errorf("errors closing databases: %v", errs)
+	}
+
+	return nil
+}
+
+// GetPrimaryDB returns primary database connection
+func (s *service) GetPrimaryDB(name string) (*sql.DB, error) {
+	return s.GetDB(name)
+}
+
+// ListenForChanges implements PostgreSQL LISTEN/NOTIFY for real-time updates
+func (s *service) ListenForChanges(ctx context.Context, dbName string, channels []string, callback func(string, string)) error {
+	s.mu.RLock()
+	config, exists := s.configs[dbName]
+	s.mu.RUnlock()
+
+	if !exists {
+		return fmt.Errorf("database %s not found", dbName)
+	}
+
+	// Only support PostgreSQL for LISTEN/NOTIFY
+	if DatabaseType(config.Type) != Postgres {
+		return fmt.Errorf("LISTEN/NOTIFY only supported for PostgreSQL databases")
+	}
+
+	// Create connection string for listener
+	connStr := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=%s",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database,
+		config.SSLMode,
+	)
+
+	// Create listener
+	listener := pq.NewListener(
+		connStr,
+		10*time.Second,
+		time.Minute,
+		func(ev pq.ListenerEventType, err error) {
+			if err != nil {
+				log.Printf("Database listener (%s) error: %v", dbName, err)
+			}
+		},
+	)
+
+	// Store listener for cleanup
+	s.listenersMu.Lock()
+	s.listeners[dbName] = listener
+	s.listenersMu.Unlock()
+
+	// Listen to specified channels
+	for _, channel := range channels {
+		err := listener.Listen(channel)
+		if err != nil {
+			listener.Close()
+			return fmt.Errorf("failed to listen to channel %s: %w", channel, err)
+		}
+		log.Printf("Listening to database channel: %s on %s", channel, dbName)
+	}
+
+	// Start listening loop
+	go func() {
+		defer func() {
+			listener.Close()
+			s.listenersMu.Lock()
+			delete(s.listeners, dbName)
+			s.listenersMu.Unlock()
+			log.Printf("Database listener for %s stopped", dbName)
+		}()
+
+		for {
+			select {
+			case n := <-listener.Notify:
+				if n != nil {
+					callback(n.Channel, n.Extra)
+				}
+			case <-ctx.Done():
+				return
+			case <-time.After(90 * time.Second):
+				// Send ping to keep connection alive
+				go func() {
+					if err := listener.Ping(); err != nil {
+						log.Printf("Listener ping failed for %s: %v", dbName, err)
+					}
+				}()
+			}
+		}
+	}()
+
+	return nil
+}
+
+// NotifyChange sends a notification to a PostgreSQL channel
+func (s *service) NotifyChange(dbName, channel, payload string) error {
+	db, err := s.GetDB(dbName)
+	if err != nil {
+		return fmt.Errorf("failed to get database %s: %w", dbName, err)
+	}
+
+	// Check if it's PostgreSQL
+	s.mu.RLock()
+	config, exists := s.configs[dbName]
+	s.mu.RUnlock()
+
+	if !exists {
+		return fmt.Errorf("database %s configuration not found", dbName)
+	}
+
+	if DatabaseType(config.Type) != Postgres {
+		return fmt.Errorf("NOTIFY only supported for PostgreSQL databases")
+	}
+
+	// Execute NOTIFY
+	query := "SELECT pg_notify($1, $2)"
+	_, err = db.Exec(query, channel, payload)
+	if err != nil {
+		return fmt.Errorf("failed to send notification: %w", err)
+	}
+
+	log.Printf("Sent notification to channel %s on %s: %s", channel, dbName, payload)
+	return nil
+}