ahdan.zeeda.0103/api_antrean
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updat erubhan besar query builder

Browse Source
This commit is contained in:
meninjar
2025-11-02 03:08:38 +00:00
parent 0002cf26be
commit 19324041b8
13 changed files with 2916 additions and 842 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/database.go
+228 -71
View File
@@ -2,24 +2,22 @@ package database
import (
"context"
"crypto/tls"
"database/sql"
"fmt"
"log" // Import runtime package
// Import debug package
"log"
"strconv"
"sync"
"time"
"api-service/internal/config"
_ "github.com/jackc/pgx/v5" // Import pgx driver
_ "github.com/jackc/pgx/v5"
"github.com/jmoiron/sqlx"
"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
_ "gorm.io/driver/mysql"
_ "gorm.io/driver/postgres"
_ "gorm.io/driver/sqlserver"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
@@ -40,27 +38,31 @@ const (
type Service interface {
Health() map[string]map[string]string
GetDB(name string) (*sql.DB, error)
GetSQLXDB(name string) (*sqlx.DB, error) // Tambahkan metode ini
GetMongoClient(name string) (*mongo.Client, error)
GetReadDB(name string) (*sql.DB, error) // For read replicas
GetReadDB(name string) (*sql.DB, error)
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
GetPrimaryDB(name string) (*sql.DB, error)
ExecuteQuery(ctx context.Context, dbName string, query string, args ...interface{}) (*sql.Rows, error)
ExecuteQueryRow(ctx context.Context, dbName string, query string, args ...interface{}) *sql.Row
Exec(ctx context.Context, dbName string, query string, args ...interface{}) (sql.Result, error)
}
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
sqlDatabases map[string]*sql.DB
sqlxDatabases map[string]*sqlx.DB // Tambahkan map untuk sqlx.DB
mongoClients map[string]*mongo.Client
readReplicas map[string][]*sql.DB
configs map[string]config.DatabaseConfig
readConfigs map[string][]config.DatabaseConfig
mu sync.RWMutex
readBalancer map[string]int
listeners map[string]*pq.Listener
listenersMu sync.RWMutex
}
var (
@@ -72,18 +74,17 @@ var (
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),
sqlDatabases: make(map[string]*sql.DB),
sqlxDatabases: make(map[string]*sqlx.DB), // Inisialisasi map sqlx
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
log.Println("Initializing database service...")
dbManager.loadFromConfig(cfg)
// Initialize all databases
@@ -125,14 +126,17 @@ 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)
// Check for duplicate database connections
for existingName, existingConfig := range s.configs {
if existingName != name &&
existingConfig.Host == config.Host &&
existingConfig.Port == config.Port &&
existingConfig.Database == config.Database &&
existingConfig.Type == config.Type {
log.Printf("⚠️ Database %s appears to be a duplicate of %s (same host:port:database), skipping connection", name, existingName)
return nil
}
}
var db *sql.DB
var err error
@@ -156,12 +160,11 @@ func (s *service) addDatabase(name string, config config.DatabaseConfig) error {
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)
return s.configureSQLDB(name, db, config)
}
func (s *service) addReadReplica(name string, index int, config config.DatabaseConfig) error {
@@ -206,19 +209,32 @@ func (s *service) addReadReplica(name string, index int, config config.DatabaseC
}
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,
// Build connection string with security parameters
// Convert timeout durations to seconds for pgx
connectTimeoutSec := int(config.ConnectTimeout.Seconds())
statementTimeoutSec := int(config.StatementTimeout.Seconds())
connStr := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s connect_timeout=%d statement_timeout=%d",
config.Host,
config.Port,
config.Username,
config.Password,
config.Database,
config.SSLMode,
connectTimeoutSec,
statementTimeoutSec,
)
if config.Schema != "" {
connStr += "&search_path=" + config.Schema
connStr += " search_path=" + config.Schema
}
// Add SSL configuration if required
if config.RequireSSL {
connStr += " sslcert=" + config.SSLCert + " sslkey=" + config.SSLKey + " sslrootcert=" + config.SSLRootCert
}
// Open connection using standard database/sql interface
db, err := sql.Open("pgx", connStr)
if err != nil {
return nil, fmt.Errorf("failed to open PostgreSQL connection: %w", err)
@@ -228,14 +244,33 @@ func (s *service) openPostgresConnection(config config.DatabaseConfig) (*sql.DB,
}
func (s *service) openMySQLConnection(config config.DatabaseConfig) (*sql.DB, error) {
connStr := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true",
// Build connection string with security parameters
connStr := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true&timeout=%s&readTimeout=%s&writeTimeout=%s",
config.Username,
config.Password,
config.Host,
config.Port,
config.Database,
config.Timeout,
config.ReadTimeout,
config.WriteTimeout,
)
// Add SSL configuration if required
if config.RequireSSL {
connStr += "&tls=true"
if config.SSLRootCert != "" {
connStr += "&ssl-ca=" + config.SSLRootCert
}
if config.SSLCert != "" {
connStr += "&ssl-cert=" + config.SSLCert
}
if config.SSLKey != "" {
connStr += "&ssl-key=" + config.SSLKey
}
}
// Open connection
db, err := sql.Open("mysql", connStr)
if err != nil {
return nil, fmt.Errorf("failed to open MySQL connection: %w", err)
@@ -245,14 +280,30 @@ func (s *service) openMySQLConnection(config config.DatabaseConfig) (*sql.DB, er
}
func (s *service) openSQLServerConnection(config config.DatabaseConfig) (*sql.DB, error) {
connStr := fmt.Sprintf("sqlserver://%s:%s@%s:%d?database=%s",
// Build connection string with security parameters
// Convert timeout to seconds for SQL Server
connectTimeoutSec := int(config.ConnectTimeout.Seconds())
connStr := fmt.Sprintf("sqlserver://%s:%s@%s:%d?database=%s&connection timeout=%d",
config.Username,
config.Password,
config.Host,
config.Port,
config.Database,
connectTimeoutSec,
)
// Add SSL configuration if required
if config.RequireSSL {
connStr += "&encrypt=true"
if config.SSLRootCert != "" {
connStr += "&trustServerCertificate=false"
} else {
connStr += "&trustServerCertificate=true"
}
}
// Open connection
db, err := sql.Open("sqlserver", connStr)
if err != nil {
return nil, fmt.Errorf("failed to open SQL Server connection: %w", err)
@@ -262,23 +313,26 @@ func (s *service) openSQLServerConnection(config config.DatabaseConfig) (*sql.DB
}
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)
// Open connection
db, err := sql.Open("sqlite3", config.Path)
if err != nil {
return nil, fmt.Errorf("failed to open SQLite connection: %w", err)
}
// Enable foreign key constraints and WAL mode for better security and performance
_, err = db.Exec("PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL;")
if err != nil {
return nil, fmt.Errorf("failed to configure SQLite: %w", err)
}
return db, nil
}
func (s *service) addMongoDB(name string, config config.DatabaseConfig) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
ctx, cancel := context.WithTimeout(context.Background(), config.Timeout)
defer cancel()
// Build MongoDB URI with authentication and TLS options
uri := fmt.Sprintf("mongodb://%s:%s@%s:%d/%s",
config.Username,
config.Password,
@@ -287,23 +341,45 @@ func (s *service) addMongoDB(name string, config config.DatabaseConfig) error {
config.Database,
)
client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
// Configure client options with security settings
clientOptions := options.Client().ApplyURI(uri)
// Set TLS configuration if needed
if config.RequireSSL {
clientOptions.SetTLSConfig(&tls.Config{
InsecureSkipVerify: config.SSLMode == "require",
MinVersion: tls.VersionTLS12,
})
}
// Set connection timeout
clientOptions.SetConnectTimeout(config.ConnectTimeout)
clientOptions.SetServerSelectionTimeout(config.Timeout)
client, err := mongo.Connect(ctx, clientOptions)
if err != nil {
return fmt.Errorf("failed to connect to MongoDB: %w", err)
}
// Ping to verify connection
if err := client.Ping(ctx, nil); err != nil {
return fmt.Errorf("failed to ping 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)
func (s *service) configureSQLDB(name string, db *sql.DB, config config.DatabaseConfig) error {
// Set connection pool limits
db.SetMaxOpenConns(config.MaxOpenConns)
db.SetMaxIdleConns(config.MaxIdleConns)
db.SetConnMaxLifetime(config.ConnMaxLifetime)
db.SetConnMaxIdleTime(config.MaxIdleTime)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
ctx, cancel := context.WithTimeout(context.Background(), config.Timeout)
defer cancel()
if err := db.PingContext(ctx); err != nil {
@@ -312,6 +388,28 @@ func (s *service) configureSQLDB(name string, db *sql.DB, maxOpenConns, maxIdleC
}
s.sqlDatabases[name] = db
// PERUBAHAN: Tambahkan pembuatan sqlx.DB dari sql.DB yang sudah ada
dbType := DatabaseType(config.Type)
var driverName string
switch dbType {
case Postgres:
driverName = "pgx"
case MySQL:
driverName = "mysql"
case SQLServer:
driverName = "sqlserver"
case SQLite:
driverName = "sqlite3"
default:
return fmt.Errorf("unsupported database type for sqlx: %s", config.Type)
}
// Buat sqlx.DB dari sql.DB yang sudah ada
sqlxDB := sqlx.NewDb(db, driverName)
s.sqlxDatabases[name] = sqlxDB
log.Printf("Successfully connected to SQL database: %s", name)
return nil
@@ -439,26 +537,27 @@ func (s *service) Health() map[string]map[string]string {
// 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)
return db, nil
}
// PERUBAHAN: Tambahkan metode GetSQLXDB
// GetSQLXDB returns a specific SQLX database connection by name
func (s *service) GetSQLXDB(name string) (*sqlx.DB, error) {
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)
// }
db, exists := s.sqlxDatabases[name]
if !exists {
return nil, fmt.Errorf("database %s not found", name)
}
return db, nil
}
@@ -537,6 +636,13 @@ func (s *service) Close() error {
var errs []error
// Close listeners first
for name, listener := range s.listeners {
if err := listener.Close(); err != nil {
errs = append(errs, fmt.Errorf("failed to close listener for %s: %w", name, err))
}
}
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))
@@ -566,10 +672,12 @@ func (s *service) Close() error {
}
s.sqlDatabases = make(map[string]*sql.DB)
s.sqlxDatabases = make(map[string]*sqlx.DB) // Reset map sqlx
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)
s.listeners = make(map[string]*pq.Listener)
if len(errs) > 0 {
return fmt.Errorf("errors closing databases: %v", errs)
@@ -583,6 +691,51 @@ func (s *service) GetPrimaryDB(name string) (*sql.DB, error) {
return s.GetDB(name)
}
// ExecuteQuery executes a query with parameters and returns rows
func (s *service) ExecuteQuery(ctx context.Context, dbName string, query string, args ...interface{}) (*sql.Rows, error) {
db, err := s.GetDB(dbName)
if err != nil {
return nil, fmt.Errorf("failed to get database %s: %w", dbName, err)
}
// Use parameterized queries to prevent SQL injection
rows, err := db.QueryContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("failed to execute query: %w", err)
}
return rows, nil
}
// ExecuteQueryRow executes a query with parameters and returns a single row
func (s *service) ExecuteQueryRow(ctx context.Context, dbName string, query string, args ...interface{}) *sql.Row {
db, err := s.GetDB(dbName)
if err != nil {
// Return an empty row with error
row := &sql.Row{}
return row
}
// Use parameterized queries to prevent SQL injection
return db.QueryRowContext(ctx, query, args...)
}
// Exec executes a query with parameters and returns the result
func (s *service) Exec(ctx context.Context, dbName string, query string, args ...interface{}) (sql.Result, error) {
db, err := s.GetDB(dbName)
if err != nil {
return nil, fmt.Errorf("failed to get database %s: %w", dbName, err)
}
// Use parameterized queries to prevent SQL injection
result, err := db.ExecContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("failed to execute query: %w", err)
}
return result, nil
}
// 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()
@@ -599,13 +752,17 @@ func (s *service) ListenForChanges(ctx context.Context, dbName string, channels
}
// Create connection string for listener
connStr := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=%s",
// Convert timeout to seconds for pq
connectTimeoutSec := int(config.ConnectTimeout.Seconds())
connStr := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=%s&connect_timeout=%d",
config.Username,
config.Password,
config.Host,
config.Port,
config.Database,
config.SSLMode,
connectTimeoutSec,
)
// Create listener
@@ -687,7 +844,7 @@ func (s *service) NotifyChange(dbName, channel, payload string) error {
return fmt.Errorf("NOTIFY only supported for PostgreSQL databases")
}
// Execute NOTIFY
// Execute NOTIFY with parameterized query to prevent SQL injection
query := "SELECT pg_notify($1, $2)"
_, err = db.Exec(query, channel, payload)
if err != nil {