-
Notifications
You must be signed in to change notification settings - Fork 7.1k
/
database.go
351 lines (294 loc) · 10 KB
/
database.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package config
import (
"bytes"
"database/sql"
"encoding/json"
"strings"
"github.com/jmoiron/sqlx"
"github.com/pkg/errors"
"github.com/mattermost/mattermost-server/v5/mlog"
"github.com/mattermost/mattermost-server/v5/model"
// Load the MySQL driver
_ "github.com/go-sql-driver/mysql"
// Load the Postgres driver
_ "github.com/lib/pq"
)
// MaxWriteLength defines the maximum length accepted for write to the Configurations or
// ConfigurationFiles table.
//
// It is imposed by MySQL's default max_allowed_packet value of 4Mb.
const MaxWriteLength = 4 * 1024 * 1024
// DatabaseStore is a config store backed by a database.
// Not to be used directly. Only to be used as a backing store for config.Store
type DatabaseStore struct {
originalDsn string
driverName string
dataSourceName string
db *sqlx.DB
}
// NewDatabaseStore creates a new instance of a config store backed by the given database.
func NewDatabaseStore(dsn string) (ds *DatabaseStore, err error) {
driverName, dataSourceName, err := parseDSN(dsn)
if err != nil {
return nil, errors.Wrap(err, "invalid DSN")
}
db, err := sqlx.Open(driverName, dataSourceName)
if err != nil {
return nil, errors.Wrapf(err, "failed to connect to %s database", driverName)
}
db.SetMaxIdleConns(0)
defer func() {
if err != nil {
db.Close()
}
}()
ds = &DatabaseStore{
driverName: driverName,
originalDsn: dsn,
dataSourceName: dataSourceName,
db: db,
}
if err = initializeConfigurationsTable(ds.db); err != nil {
err = errors.Wrap(err, "failed to initialize")
return nil, err
}
return ds, nil
}
// initializeConfigurationsTable ensures the requisite tables in place to form the backing store.
//
// Uses MEDIUMTEXT on MySQL, and TEXT on sane databases.
func initializeConfigurationsTable(db *sqlx.DB) error {
mysqlCharset := ""
if db.DriverName() == "mysql" {
mysqlCharset = "DEFAULT CHARACTER SET utf8mb4"
}
_, err := db.Exec(`
CREATE TABLE IF NOT EXISTS Configurations (
Id VARCHAR(26) PRIMARY KEY,
Value TEXT NOT NULL,
CreateAt BIGINT NOT NULL,
Active BOOLEAN NULL UNIQUE
)
` + mysqlCharset)
if err != nil {
return errors.Wrap(err, "failed to create Configurations table")
}
_, err = db.Exec(`
CREATE TABLE IF NOT EXISTS ConfigurationFiles (
Name VARCHAR(64) PRIMARY KEY,
Data TEXT NOT NULL,
CreateAt BIGINT NOT NULL,
UpdateAt BIGINT NOT NULL
)
` + mysqlCharset)
if err != nil {
return errors.Wrap(err, "failed to create ConfigurationFiles table")
}
// Change from TEXT (65535 limit) to MEDIUM TEXT (16777215) on MySQL. This is a
// backwards-compatible migration for any existing schema.
// Also fix using the wrong encoding initially
if db.DriverName() == "mysql" {
_, err = db.Exec(`ALTER TABLE Configurations MODIFY Value MEDIUMTEXT`)
if err != nil {
return errors.Wrap(err, "failed to alter Configurations table")
}
_, err = db.Exec(`ALTER TABLE Configurations CONVERT TO CHARACTER SET utf8mb4`)
if err != nil {
return errors.Wrap(err, "failed to alter Configurations table character set")
}
_, err = db.Exec(`ALTER TABLE ConfigurationFiles MODIFY Data MEDIUMTEXT`)
if err != nil {
return errors.Wrap(err, "failed to alter ConfigurationFiles table")
}
_, err = db.Exec(`ALTER TABLE ConfigurationFiles CONVERT TO CHARACTER SET utf8mb4`)
if err != nil {
return errors.Wrap(err, "failed to alter ConfigurationFiles table character set")
}
}
return nil
}
// parseDSN splits up a connection string into a driver name and data source name.
//
// For example:
// mysql://mmuser:mostest@localhost:5432/mattermost_test
// returns
// driverName = mysql
// dataSourceName = mmuser:mostest@localhost:5432/mattermost_test
//
// By contrast, a Postgres DSN is returned unmodified.
func parseDSN(dsn string) (string, string, error) {
// Treat the DSN as the URL that it is.
s := strings.SplitN(dsn, "://", 2)
if len(s) != 2 {
return "", "", errors.New("failed to parse DSN as URL")
}
scheme := s[0]
switch scheme {
case "mysql":
// Strip off the mysql:// for the dsn with which to connect.
dsn = s[1]
case "postgres":
// No changes required
default:
return "", "", errors.Errorf("unsupported scheme %s", scheme)
}
return scheme, dsn, nil
}
// Set replaces the current configuration in its entirety and updates the backing store.
func (ds *DatabaseStore) Set(newCfg *model.Config) error {
return ds.persist(newCfg)
}
// maxLength identifies the maximum length of a configuration or configuration file
func (ds *DatabaseStore) checkLength(length int) error {
if ds.db.DriverName() == "mysql" && length > MaxWriteLength {
return errors.Errorf("value is too long: %d > %d bytes", length, MaxWriteLength)
}
return nil
}
// persist writes the configuration to the configured database.
func (ds *DatabaseStore) persist(cfg *model.Config) error {
b, err := marshalConfig(cfg)
if err != nil {
return errors.Wrap(err, "failed to serialize")
}
id := model.NewId()
value := string(b)
createAt := model.GetMillis()
err = ds.checkLength(len(value))
if err != nil {
return errors.Wrap(err, "marshalled configuration failed length check")
}
tx, err := ds.db.Beginx()
if err != nil {
return errors.Wrap(err, "failed to begin transaction")
}
defer func() {
// Rollback after Commit just returns sql.ErrTxDone.
if err := tx.Rollback(); err != nil && err != sql.ErrTxDone {
mlog.Error("Failed to rollback configuration transaction", mlog.Err(err))
}
}()
params := map[string]interface{}{
"id": id,
"value": value,
"create_at": createAt,
"key": "ConfigurationId",
}
// Skip the persist altogether if we're effectively writing the same configuration.
var oldValue []byte
row := ds.db.QueryRow("SELECT Value FROM Configurations WHERE Active")
if err := row.Scan(&oldValue); err != nil && err != sql.ErrNoRows {
return errors.Wrap(err, "failed to query active configuration")
}
if bytes.Equal(oldValue, b) {
return nil
}
if _, err := tx.Exec("UPDATE Configurations SET Active = NULL WHERE Active"); err != nil {
return errors.Wrap(err, "failed to deactivate current configuration")
}
if _, err := tx.NamedExec("INSERT INTO Configurations (Id, Value, CreateAt, Active) VALUES (:id, :value, :create_at, TRUE)", params); err != nil {
return errors.Wrap(err, "failed to record new configuration")
}
if err := tx.Commit(); err != nil {
return errors.Wrap(err, "failed to commit transaction")
}
return nil
}
// Load updates the current configuration from the backing store.
func (ds *DatabaseStore) Load() ([]byte, error) {
var configurationData []byte
row := ds.db.QueryRow("SELECT Value FROM Configurations WHERE Active")
if err := row.Scan(&configurationData); err != nil && err != sql.ErrNoRows {
return nil, errors.Wrap(err, "failed to query active configuration")
}
// Initialize from the default config if no active configuration could be found.
if len(configurationData) == 0 {
configWithDB := model.Config{}
configWithDB.SqlSettings.DriverName = model.NewString(ds.driverName)
configWithDB.SqlSettings.DataSource = model.NewString(ds.dataSourceName)
return json.Marshal(configWithDB)
}
return configurationData, nil
}
// GetFile fetches the contents of a previously persisted configuration file.
func (ds *DatabaseStore) GetFile(name string) ([]byte, error) {
query, args, err := sqlx.Named("SELECT Data FROM ConfigurationFiles WHERE Name = :name", map[string]interface{}{
"name": name,
})
if err != nil {
return nil, err
}
var data []byte
row := ds.db.QueryRowx(ds.db.Rebind(query), args...)
if err = row.Scan(&data); err != nil {
return nil, errors.Wrapf(err, "failed to scan data from row for %s", name)
}
return data, nil
}
// SetFile sets or replaces the contents of a configuration file.
func (ds *DatabaseStore) SetFile(name string, data []byte) error {
err := ds.checkLength(len(data))
if err != nil {
return errors.Wrap(err, "file data failed length check")
}
params := map[string]interface{}{
"name": name,
"data": data,
"create_at": model.GetMillis(),
"update_at": model.GetMillis(),
}
result, err := ds.db.NamedExec("UPDATE ConfigurationFiles SET Data = :data, UpdateAt = :update_at WHERE Name = :name", params)
if err != nil {
return errors.Wrapf(err, "failed to update row for %s", name)
}
count, err := result.RowsAffected()
if err != nil {
return errors.Wrapf(err, "failed to count rows affected for %s", name)
} else if count > 0 {
return nil
}
_, err = ds.db.NamedExec("INSERT INTO ConfigurationFiles (Name, Data, CreateAt, UpdateAt) VALUES (:name, :data, :create_at, :update_at)", params)
if err != nil {
return errors.Wrapf(err, "failed to insert row for %s", name)
}
return nil
}
// HasFile returns true if the given file was previously persisted.
func (ds *DatabaseStore) HasFile(name string) (bool, error) {
query, args, err := sqlx.Named("SELECT COUNT(*) FROM ConfigurationFiles WHERE Name = :name", map[string]interface{}{
"name": name,
})
if err != nil {
return false, err
}
var count int64
row := ds.db.QueryRowx(ds.db.Rebind(query), args...)
if err = row.Scan(&count); err != nil {
return false, errors.Wrapf(err, "failed to scan count of rows for %s", name)
}
return count != 0, nil
}
// RemoveFile remoevs a previously persisted configuration file.
func (ds *DatabaseStore) RemoveFile(name string) error {
_, err := ds.db.NamedExec("DELETE FROM ConfigurationFiles WHERE Name = :name", map[string]interface{}{
"name": name,
})
if err != nil {
return errors.Wrapf(err, "failed to remove row for %s", name)
}
return nil
}
// String returns the path to the database backing the config, masking the password.
func (ds *DatabaseStore) String() string {
return stripPassword(ds.originalDsn, ds.driverName)
}
// Close cleans up resources associated with the store.
func (ds *DatabaseStore) Close() error {
return ds.db.Close()
}
// Watch nothing on memory store
func (ds *DatabaseStore) Watch(_ func()) error {
return nil
}