Skip to content

Commit 7893443

Browse files
cschleidencschleiden
authored
Merge branch 'main' into sqlite-vacuum
2 parents c099e23 + dbca48e commit 7893443

10 files changed

Lines changed: 566 additions & 32 deletions

File tree

‎backend/mysql/mysql.go‎

Lines changed: 54 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -52,10 +52,42 @@ func NewMysqlBackend(host string, port int, user, password, database string, opt
5252
}
5353

5454
b := &mysqlBackend{
55-
dsn: dsn,
56-
db: db,
57-
workerName: getWorkerName(options),
58-
options: options,
55+
dsn: dsn,
56+
db: db,
57+
workerName: getWorkerName(options),
58+
options: options,
59+
ownsConnection: true,
60+
}
61+
62+
if options.ApplyMigrations {
63+
if err := b.Migrate(); err != nil {
64+
panic(err)
65+
}
66+
}
67+
68+
return b
69+
}
70+
71+
// NewMysqlBackendWithDB creates a new MySQL backend using an existing database connection.
72+
// When using this constructor, the backend will not close the database connection when Close() is called.
73+
// Migrations are disabled by default; to enable them, use WithApplyMigrations(true) along with
74+
// WithMigrationDSN to provide a DSN that supports multi-statement queries.
75+
func NewMysqlBackendWithDB(db *sql.DB, opts ...option) *mysqlBackend {
76+
options := &options{
77+
Options: backend.ApplyOptions(),
78+
ApplyMigrations: false,
79+
}
80+
81+
for _, opt := range opts {
82+
opt(options)
83+
}
84+
85+
b := &mysqlBackend{
86+
dsn: "",
87+
db: db,
88+
workerName: getWorkerName(options),
89+
options: options,
90+
ownsConnection: false,
5991
}
6092

6193
if options.ApplyMigrations {
@@ -68,23 +100,36 @@ func NewMysqlBackend(host string, port int, user, password, database string, opt
68100
}
69101

70102
type mysqlBackend struct {
71-
dsn string
72-
db *sql.DB
73-
workerName string
74-
options *options
103+
dsn string
104+
db *sql.DB
105+
workerName string
106+
options *options
107+
ownsConnection bool
75108
}
76109

77110
func (mb *mysqlBackend) FeatureSupported(feature backend.Feature) bool {
78111
return true
79112
}
80113

81114
func (mb *mysqlBackend) Close() error {
115+
if !mb.ownsConnection {
116+
return nil
117+
}
82118
return mb.db.Close()
83119
}
84120

85121
// Migrate applies any pending database migrations.
86122
func (mb *mysqlBackend) Migrate() error {
87-
schemaDsn := mb.dsn + "&multiStatements=true"
123+
// Determine which DSN to use for migrations
124+
var schemaDsn string
125+
if mb.options.MigrationDSN != "" {
126+
schemaDsn = mb.options.MigrationDSN
127+
} else if mb.dsn != "" {
128+
schemaDsn = mb.dsn + "&multiStatements=true"
129+
} else {
130+
return errors.New("cannot apply migrations: no DSN available; use WithMigrationDSN option or apply migrations externally")
131+
}
132+
88133
db, err := sql.Open("mysql", schemaDsn)
89134
if err != nil {
90135
return fmt.Errorf("opening schema database: %w", err)

‎backend/mysql/mysql_test.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -216,3 +216,126 @@ func Test_MysqlBackend_WorkerName(t *testing.T) {
216216
}
217217
})
218218
}
219+
220+
func Test_MysqlBackendWithDB(t *testing.T) {
221+
if testing.Short() {
222+
t.Skip()
223+
}
224+
225+
t.Run("UsesProvidedConnection", func(t *testing.T) {
226+
// Create database for test
227+
adminDB, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
228+
if err != nil {
229+
t.Fatal(err)
230+
}
231+
232+
dbName := "test_withdb_" + strings.ReplaceAll(uuid.NewString(), "-", "")
233+
if _, err := adminDB.Exec("CREATE DATABASE " + dbName); err != nil {
234+
t.Fatal(err)
235+
}
236+
defer func() {
237+
adminDB.Exec("DROP DATABASE IF EXISTS " + dbName)
238+
adminDB.Close()
239+
}()
240+
241+
// Create our own connection to the test database
242+
dsn := fmt.Sprintf("%s:%s@tcp(localhost:3306)/%s?parseTime=true&interpolateParams=true", testUser, testPassword, dbName)
243+
db, err := sql.Open("mysql", dsn)
244+
if err != nil {
245+
t.Fatal(err)
246+
}
247+
defer db.Close()
248+
249+
// Create backend with existing connection and migration DSN
250+
migrationDSN := dsn + "&multiStatements=true"
251+
backend := NewMysqlBackendWithDB(db,
252+
WithApplyMigrations(true),
253+
WithMigrationDSN(migrationDSN),
254+
)
255+
256+
// Verify the backend uses our connection
257+
if backend.db != db {
258+
t.Error("Backend should use provided db connection")
259+
}
260+
if backend.ownsConnection {
261+
t.Error("Backend should not own the connection")
262+
}
263+
264+
// Close backend - should NOT close our connection
265+
if err := backend.Close(); err != nil {
266+
t.Fatal(err)
267+
}
268+
269+
// Verify our connection is still usable
270+
if err := db.Ping(); err != nil {
271+
t.Errorf("Connection should still be open after backend.Close(): %v", err)
272+
}
273+
})
274+
275+
t.Run("MigrationsDisabledByDefault", func(t *testing.T) {
276+
// Create database for test
277+
adminDB, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
278+
if err != nil {
279+
t.Fatal(err)
280+
}
281+
282+
dbName := "test_withdb2_" + strings.ReplaceAll(uuid.NewString(), "-", "")
283+
if _, err := adminDB.Exec("CREATE DATABASE " + dbName); err != nil {
284+
t.Fatal(err)
285+
}
286+
defer func() {
287+
adminDB.Exec("DROP DATABASE IF EXISTS " + dbName)
288+
adminDB.Close()
289+
}()
290+
291+
dsn := fmt.Sprintf("%s:%s@tcp(localhost:3306)/%s?parseTime=true&interpolateParams=true", testUser, testPassword, dbName)
292+
db, err := sql.Open("mysql", dsn)
293+
if err != nil {
294+
t.Fatal(err)
295+
}
296+
defer db.Close()
297+
298+
// Create backend without enabling migrations
299+
backend := NewMysqlBackendWithDB(db)
300+
defer backend.Close()
301+
302+
// Tables should not exist since migrations weren't applied
303+
_, err = db.Exec("SELECT 1 FROM instances LIMIT 1")
304+
if err == nil {
305+
t.Error("Expected error because table should not exist")
306+
}
307+
})
308+
309+
t.Run("MigrationFailsWithoutDSN", func(t *testing.T) {
310+
// Create database for test
311+
adminDB, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
312+
if err != nil {
313+
t.Fatal(err)
314+
}
315+
316+
dbName := "test_withdb3_" + strings.ReplaceAll(uuid.NewString(), "-", "")
317+
if _, err := adminDB.Exec("CREATE DATABASE " + dbName); err != nil {
318+
t.Fatal(err)
319+
}
320+
defer func() {
321+
adminDB.Exec("DROP DATABASE IF EXISTS " + dbName)
322+
adminDB.Close()
323+
}()
324+
325+
dsn := fmt.Sprintf("%s:%s@tcp(localhost:3306)/%s?parseTime=true&interpolateParams=true", testUser, testPassword, dbName)
326+
db, err := sql.Open("mysql", dsn)
327+
if err != nil {
328+
t.Fatal(err)
329+
}
330+
defer db.Close()
331+
332+
// Create backend without migration DSN - should panic when trying to migrate
333+
defer func() {
334+
if r := recover(); r == nil {
335+
t.Error("Expected panic when ApplyMigrations=true without MigrationDSN")
336+
}
337+
}()
338+
339+
NewMysqlBackendWithDB(db, WithApplyMigrations(true))
340+
})
341+
}

‎backend/mysql/options.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,11 @@ type options struct {
1313

1414
// ApplyMigrations automatically applies database migrations on startup.
1515
ApplyMigrations bool
16+
17+
// MigrationDSN is an optional DSN to use for running migrations. This is useful when
18+
// using NewMysqlBackendWithDB where no DSN is available. The DSN must support
19+
// multi-statement queries (e.g., include &multiStatements=true).
20+
MigrationDSN string
1621
}
1722

1823
type option func(*options)
@@ -30,6 +35,15 @@ func WithMySQLOptions(f func(db *sql.DB)) option {
3035
}
3136
}
3237

38+
// WithMigrationDSN sets the DSN to use for running migrations. This is required when
39+
// using NewMysqlBackendWithDB with ApplyMigrations enabled. The DSN should support
40+
// multi-statement queries.
41+
func WithMigrationDSN(dsn string) option {
42+
return func(o *options) {
43+
o.MigrationDSN = dsn
44+
}
45+
}
46+
3347
// WithBackendOptions allows to pass generic backend options.
3448
func WithBackendOptions(opts ...backend.BackendOption) option {
3549
return func(o *options) {

‎backend/postgres/postgres.go‎

Lines changed: 61 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -51,10 +51,42 @@ func NewPostgresBackend(host string, port int, user, password, database string,
5151
}
5252

5353
b := &postgresBackend{
54-
dsn: dsn,
55-
db: db,
56-
workerName: getWorkerName(options),
57-
options: options,
54+
dsn: dsn,
55+
db: db,
56+
workerName: getWorkerName(options),
57+
options: options,
58+
ownsConnection: true,
59+
}
60+
61+
if options.ApplyMigrations {
62+
if err := b.Migrate(); err != nil {
63+
panic(err)
64+
}
65+
}
66+
67+
return b
68+
}
69+
70+
// NewPostgresBackendWithDB creates a new Postgres backend using an existing database connection.
71+
// When using this constructor, the backend will not close the database connection when Close() is called.
72+
// Migrations can still be applied using WithApplyMigrations(true) as Postgres does not require
73+
// special connection settings for migrations.
74+
func NewPostgresBackendWithDB(db *sql.DB, opts ...option) *postgresBackend {
75+
options := &options{
76+
Options: backend.ApplyOptions(),
77+
ApplyMigrations: false,
78+
}
79+
80+
for _, opt := range opts {
81+
opt(options)
82+
}
83+
84+
b := &postgresBackend{
85+
dsn: "",
86+
db: db,
87+
workerName: getWorkerName(options),
88+
options: options,
89+
ownsConnection: false,
5890
}
5991

6092
if options.ApplyMigrations {
@@ -67,26 +99,39 @@ func NewPostgresBackend(host string, port int, user, password, database string,
6799
}
68100

69101
type postgresBackend struct {
70-
dsn string
71-
db *sql.DB
72-
workerName string
73-
options *options
102+
dsn string
103+
db *sql.DB
104+
workerName string
105+
options *options
106+
ownsConnection bool
74107
}
75108

76109
func (pb *postgresBackend) FeatureSupported(feature backend.Feature) bool {
77110
return true
78111
}
79112

80113
func (pb *postgresBackend) Close() error {
114+
if !pb.ownsConnection {
115+
return nil
116+
}
81117
return pb.db.Close()
82118
}
83119

84120
// Migrate applies any pending database migrations.
85121
func (pb *postgresBackend) Migrate() error {
86-
schemaDsn := pb.dsn
87-
db, err := sql.Open("pgx", schemaDsn)
88-
if err != nil {
89-
return fmt.Errorf("opening schema database: %w", err)
122+
var db *sql.DB
123+
var needsClose bool
124+
125+
if pb.dsn != "" {
126+
var err error
127+
db, err = sql.Open("pgx", pb.dsn)
128+
if err != nil {
129+
return fmt.Errorf("opening schema database: %w", err)
130+
}
131+
needsClose = true
132+
} else {
133+
db = pb.db
134+
needsClose = false
90135
}
91136

92137
dbi, err := postgres.WithInstance(db, &postgres.Config{})
@@ -110,8 +155,10 @@ func (pb *postgresBackend) Migrate() error {
110155
}
111156
}
112157

113-
if err := db.Close(); err != nil {
114-
return fmt.Errorf("closing schema database: %w", err)
158+
if needsClose {
159+
if err := db.Close(); err != nil {
160+
return fmt.Errorf("closing schema database: %w", err)
161+
}
115162
}
116163

117164
return nil

0 commit comments

Comments
 (0)