107 lines
2.9 KiB
Go
107 lines
2.9 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
)
|
|
|
|
// EngineWithMigrations extends Engine with migration support
|
|
type EngineWithMigrations struct {
|
|
readDB *sql.DB
|
|
writeDB *sql.DB
|
|
}
|
|
|
|
// NewWithMigrations creates a new Engine with read/write separation and migration support
|
|
func NewWithMigrations(readURI, writeURI string) (*EngineWithMigrations, error) {
|
|
readDB, err := connect(readURI)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to connect to read database: %w", err)
|
|
}
|
|
|
|
writeDB, err := connect(writeURI)
|
|
if err != nil {
|
|
readDB.Close()
|
|
return nil, fmt.Errorf("failed to connect to write database: %w", err)
|
|
}
|
|
|
|
// Initialize schema version table if it doesn't exist
|
|
if err := initSchemaVersionTable(writeDB); err != nil {
|
|
readDB.Close()
|
|
writeDB.Close()
|
|
return nil, fmt.Errorf("failed to initialize schema version table: %w", err)
|
|
}
|
|
|
|
return &EngineWithMigrations{
|
|
readDB: readDB,
|
|
writeDB: writeDB,
|
|
}, nil
|
|
}
|
|
|
|
// WriterDB returns the write database connection
|
|
func (e *EngineWithMigrations) WriterDB() *sql.DB {
|
|
return e.writeDB
|
|
}
|
|
|
|
// ReaderDB returns the read database connection
|
|
func (e *EngineWithMigrations) ReaderDB() *sql.DB {
|
|
return e.readDB
|
|
}
|
|
|
|
// GetDatabaseSchemaVersion returns the current database schema version
|
|
func (e *EngineWithMigrations) GetDatabaseSchemaVersion(ctx context.Context) (string, error) {
|
|
var version string
|
|
err := e.readDB.QueryRowContext(ctx, "SELECT version FROM schema_version ORDER BY id DESC LIMIT 1").Scan(&version)
|
|
if err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return "", nil // No version means fresh database
|
|
}
|
|
return "", fmt.Errorf("failed to get schema version: %w", err)
|
|
}
|
|
return version, nil
|
|
}
|
|
|
|
// SetDatabaseSchemaVersion sets the database schema version
|
|
func (e *EngineWithMigrations) SetDatabaseSchemaVersion(ctx context.Context, version string) error {
|
|
_, err := e.writeDB.ExecContext(ctx, "INSERT INTO schema_version (version) VALUES (?)", version)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to set schema version: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Close closes all database connections
|
|
func (e *EngineWithMigrations) Close() error {
|
|
var errs []error
|
|
if e.readDB != nil {
|
|
if err := e.readDB.Close(); err != nil {
|
|
errs = append(errs, fmt.Errorf("failed to close read connection: %w", err))
|
|
}
|
|
}
|
|
if e.writeDB != nil {
|
|
if err := e.writeDB.Close(); err != nil {
|
|
errs = append(errs, fmt.Errorf("failed to close write connection: %w", err))
|
|
}
|
|
}
|
|
if len(errs) > 0 {
|
|
return fmt.Errorf("errors closing connections: %v", errs)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// initSchemaVersionTable creates the schema_version table if it doesn't exist
|
|
func initSchemaVersionTable(db *sql.DB) error {
|
|
query := `
|
|
CREATE TABLE IF NOT EXISTS schema_version (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
version TEXT NOT NULL,
|
|
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
)`
|
|
|
|
_, err := db.Exec(query)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create schema_version table: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|