* MM-31094: Adds tooling to develop and test using a MySQL instance with replication lag. Adds some lazy lookups to fallback to master if results are not found. * MM-31094: Removes mysql-read-replica from default docker services. * MM-31094: Switches (store..SessionStore).Get and (store.TeamStore).GetMember to using context.Context. * MM-31094: Updates (store.UsersStore).Get to use context. * MM-31094: Updates (store.PostStore).Get to use context. * MM-31094: Removes feature flag and config setting. * MM-31094: Rolls back some master reads. * MM-31094: Rolls a non-cache read. * MM-31094: Removes feature flag from the store. * MM-31094: Removes unused constant and struct field. * MM-31094: Removes some old feature flag references. * MM-31094: Fixes some tests. * MM-31094: App layers fix. * MM-31094: Fixes mocks. * MM-31094: Don't reparse flag. * MM-31094: No reparse. * MM-31094: Removed unused FeatureFlags field. * MM-31094: Removes unnecessary feature flags variable declarations. * MM-31094: Fixes copy-paste error. * MM-31094: Fixes logical error. * MM-30194: Removes test method from store. * Revert "MM-30194: Removes test method from store." This reverts commit d5a6e8529bd5f4d993824c828e239d009b05e567. * MM-31094: Conforming to make's strange syntax. * MM-31094: Configures helper for read replica with option. * MM-31094: Adds some missing ctx's. * MM-31094: WIP * MM-31094: Updates test names. * MM-31094: WIP * MM-31094: Removes unnecessary master reads. * MM-31094: ID case changes out of scope. * MM-31094: Removes unused context. * MM-31094: Switches to a helper. Removes some var naming changes. Fixes a merge error. * MM-31094: Removes SQLITE db driver ref. * MM-31094: Layer generate fix. * MM-31094: Removes unnecessary changes. * MM-31094: Moves test method. * MM-31094: Re-add previous fix. * MM-31094: Removes make command for dev. * MM-31094: Fix for login. Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
1425 строки
42 KiB
Go
1425 строки
42 KiB
Go
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
package sqlstore
|
|
|
|
import (
|
|
"context"
|
|
dbsql "database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
sq "github.com/Masterminds/squirrel"
|
|
"github.com/dyatlov/go-opengraph/opengraph"
|
|
"github.com/go-sql-driver/mysql"
|
|
"github.com/golang-migrate/migrate/v4"
|
|
"github.com/golang-migrate/migrate/v4/database"
|
|
mysqlmigrate "github.com/golang-migrate/migrate/v4/database/mysql"
|
|
"github.com/golang-migrate/migrate/v4/database/postgres"
|
|
_ "github.com/golang-migrate/migrate/v4/source/file"
|
|
bindata "github.com/golang-migrate/migrate/v4/source/go_bindata"
|
|
"github.com/lib/pq"
|
|
_ "github.com/lib/pq"
|
|
"github.com/mattermost/gorp"
|
|
"github.com/pkg/errors"
|
|
|
|
"github.com/mattermost/mattermost-server/v5/db/migrations"
|
|
"github.com/mattermost/mattermost-server/v5/einterfaces"
|
|
"github.com/mattermost/mattermost-server/v5/model"
|
|
"github.com/mattermost/mattermost-server/v5/shared/i18n"
|
|
"github.com/mattermost/mattermost-server/v5/shared/mlog"
|
|
"github.com/mattermost/mattermost-server/v5/store"
|
|
)
|
|
|
|
type migrationDirection string
|
|
|
|
const (
|
|
IndexTypeFullText = "full_text"
|
|
IndexTypeDefault = "default"
|
|
PGDupTableErrorCode = "42P07" // see https://github.com/lib/pq/blob/master/error.go#L268
|
|
MySQLDupTableErrorCode = uint16(1050) // see https://dev.mysql.com/doc/mysql-errors/5.7/en/server-error-reference.html#error_er_table_exists_error
|
|
DBPingAttempts = 18
|
|
DBPingTimeoutSecs = 10
|
|
// This is a numerical version string by postgres. The format is
|
|
// 2 characters for major, minor, and patch version prior to 10.
|
|
// After 10, it's major and minor only.
|
|
// 10.1 would be 100001.
|
|
// 9.6.3 would be 90603.
|
|
MinimumRequiredPostgresVersion = 100000
|
|
|
|
migrationsDirectionUp migrationDirection = "up"
|
|
migrationsDirectionDown migrationDirection = "down"
|
|
)
|
|
|
|
const (
|
|
ExitGenericFailure = 1
|
|
ExitCreateTable = 100
|
|
ExitDBOpen = 101
|
|
ExitPing = 102
|
|
ExitNoDriver = 103
|
|
ExitTableExists = 104
|
|
ExitTableExistsMySQL = 105
|
|
ExitColumnExists = 106
|
|
ExitDoesColumnExistsPostgres = 107
|
|
ExitDoesColumnExistsMySQL = 108
|
|
ExitDoesColumnExistsMissing = 109
|
|
ExitCreateColumnPostgres = 110
|
|
ExitCreateColumnMySQL = 111
|
|
ExitCreateColumnMissing = 112
|
|
ExitRemoveColumn = 113
|
|
ExitRenameColumn = 114
|
|
ExitMaxColumn = 115
|
|
ExitAlterColumn = 116
|
|
ExitCreateIndexPostgres = 117
|
|
ExitCreateIndexMySQL = 118
|
|
ExitCreateIndexFullMySQL = 119
|
|
ExitCreateIndexMissing = 120
|
|
ExitRemoveIndexPostgres = 121
|
|
ExitRemoveIndexMySQL = 122
|
|
ExitRemoveIndexMissing = 123
|
|
ExitRemoveTable = 134
|
|
ExitAlterPrimaryKey = 139
|
|
)
|
|
|
|
type SqlStoreStores struct {
|
|
team store.TeamStore
|
|
channel store.ChannelStore
|
|
post store.PostStore
|
|
thread store.ThreadStore
|
|
user store.UserStore
|
|
bot store.BotStore
|
|
audit store.AuditStore
|
|
cluster store.ClusterDiscoveryStore
|
|
compliance store.ComplianceStore
|
|
session store.SessionStore
|
|
oauth store.OAuthStore
|
|
system store.SystemStore
|
|
webhook store.WebhookStore
|
|
command store.CommandStore
|
|
commandWebhook store.CommandWebhookStore
|
|
preference store.PreferenceStore
|
|
license store.LicenseStore
|
|
token store.TokenStore
|
|
emoji store.EmojiStore
|
|
status store.StatusStore
|
|
fileInfo store.FileInfoStore
|
|
uploadSession store.UploadSessionStore
|
|
reaction store.ReactionStore
|
|
job store.JobStore
|
|
userAccessToken store.UserAccessTokenStore
|
|
plugin store.PluginStore
|
|
channelMemberHistory store.ChannelMemberHistoryStore
|
|
role store.RoleStore
|
|
scheme store.SchemeStore
|
|
TermsOfService store.TermsOfServiceStore
|
|
productNotices store.ProductNoticesStore
|
|
group store.GroupStore
|
|
UserTermsOfService store.UserTermsOfServiceStore
|
|
linkMetadata store.LinkMetadataStore
|
|
}
|
|
|
|
type SqlStore struct {
|
|
// rrCounter and srCounter should be kept first.
|
|
// See https://github.com/mattermost/mattermost-server/v5/pull/7281
|
|
rrCounter int64
|
|
srCounter int64
|
|
master *gorp.DbMap
|
|
Replicas []*gorp.DbMap
|
|
searchReplicas []*gorp.DbMap
|
|
stores SqlStoreStores
|
|
settings *model.SqlSettings
|
|
lockedToMaster bool
|
|
context context.Context
|
|
license *model.License
|
|
licenseMutex sync.RWMutex
|
|
}
|
|
|
|
type TraceOnAdapter struct{}
|
|
|
|
func (t *TraceOnAdapter) Printf(format string, v ...interface{}) {
|
|
originalString := fmt.Sprintf(format, v...)
|
|
newString := strings.ReplaceAll(originalString, "\n", " ")
|
|
newString = strings.ReplaceAll(newString, "\t", " ")
|
|
newString = strings.ReplaceAll(newString, "\"", "")
|
|
mlog.Debug(newString)
|
|
}
|
|
|
|
func New(settings model.SqlSettings, metrics einterfaces.MetricsInterface) *SqlStore {
|
|
store := &SqlStore{
|
|
rrCounter: 0,
|
|
srCounter: 0,
|
|
settings: &settings,
|
|
}
|
|
|
|
store.initConnection()
|
|
|
|
err := store.migrate(migrationsDirectionUp)
|
|
if err != nil {
|
|
mlog.Critical("Failed to apply database migrations.", mlog.Err(err))
|
|
os.Exit(ExitGenericFailure)
|
|
}
|
|
|
|
store.stores.team = newSqlTeamStore(store)
|
|
store.stores.channel = newSqlChannelStore(store, metrics)
|
|
store.stores.post = newSqlPostStore(store, metrics)
|
|
store.stores.user = newSqlUserStore(store, metrics)
|
|
store.stores.bot = newSqlBotStore(store, metrics)
|
|
store.stores.audit = newSqlAuditStore(store)
|
|
store.stores.cluster = newSqlClusterDiscoveryStore(store)
|
|
store.stores.compliance = newSqlComplianceStore(store)
|
|
store.stores.session = newSqlSessionStore(store)
|
|
store.stores.oauth = newSqlOAuthStore(store)
|
|
store.stores.system = newSqlSystemStore(store)
|
|
store.stores.webhook = newSqlWebhookStore(store, metrics)
|
|
store.stores.command = newSqlCommandStore(store)
|
|
store.stores.commandWebhook = newSqlCommandWebhookStore(store)
|
|
store.stores.preference = newSqlPreferenceStore(store)
|
|
store.stores.license = newSqlLicenseStore(store)
|
|
store.stores.token = newSqlTokenStore(store)
|
|
store.stores.emoji = newSqlEmojiStore(store, metrics)
|
|
store.stores.status = newSqlStatusStore(store)
|
|
store.stores.fileInfo = newSqlFileInfoStore(store, metrics)
|
|
store.stores.uploadSession = newSqlUploadSessionStore(store)
|
|
store.stores.thread = newSqlThreadStore(store)
|
|
store.stores.job = newSqlJobStore(store)
|
|
store.stores.userAccessToken = newSqlUserAccessTokenStore(store)
|
|
store.stores.channelMemberHistory = newSqlChannelMemberHistoryStore(store)
|
|
store.stores.plugin = newSqlPluginStore(store)
|
|
store.stores.TermsOfService = newSqlTermsOfServiceStore(store, metrics)
|
|
store.stores.UserTermsOfService = newSqlUserTermsOfServiceStore(store)
|
|
store.stores.linkMetadata = newSqlLinkMetadataStore(store)
|
|
store.stores.reaction = newSqlReactionStore(store)
|
|
store.stores.role = newSqlRoleStore(store)
|
|
store.stores.scheme = newSqlSchemeStore(store)
|
|
store.stores.group = newSqlGroupStore(store)
|
|
store.stores.productNotices = newSqlProductNoticesStore(store)
|
|
err = store.GetMaster().CreateTablesIfNotExists()
|
|
|
|
if err != nil {
|
|
if IsDuplicate(err) {
|
|
mlog.Warn("Duplicate key error occurred; assuming table already created and proceeding.", mlog.Err(err))
|
|
} else {
|
|
mlog.Critical("Error creating database tables.", mlog.Err(err))
|
|
os.Exit(ExitCreateTable)
|
|
}
|
|
}
|
|
|
|
err = upgradeDatabase(store, model.CurrentVersion)
|
|
if err != nil {
|
|
mlog.Critical("Failed to upgrade database.", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitGenericFailure)
|
|
}
|
|
|
|
store.stores.channel.(*SqlChannelStore).createIndexesIfNotExists()
|
|
store.stores.post.(*SqlPostStore).createIndexesIfNotExists()
|
|
store.stores.thread.(*SqlThreadStore).createIndexesIfNotExists()
|
|
store.stores.user.(*SqlUserStore).createIndexesIfNotExists()
|
|
store.stores.bot.(*SqlBotStore).createIndexesIfNotExists()
|
|
store.stores.audit.(*SqlAuditStore).createIndexesIfNotExists()
|
|
store.stores.compliance.(*SqlComplianceStore).createIndexesIfNotExists()
|
|
store.stores.session.(*SqlSessionStore).createIndexesIfNotExists()
|
|
store.stores.oauth.(*SqlOAuthStore).createIndexesIfNotExists()
|
|
store.stores.system.(*SqlSystemStore).createIndexesIfNotExists()
|
|
store.stores.webhook.(*SqlWebhookStore).createIndexesIfNotExists()
|
|
store.stores.command.(*SqlCommandStore).createIndexesIfNotExists()
|
|
store.stores.commandWebhook.(*SqlCommandWebhookStore).createIndexesIfNotExists()
|
|
store.stores.preference.(*SqlPreferenceStore).createIndexesIfNotExists()
|
|
store.stores.license.(*SqlLicenseStore).createIndexesIfNotExists()
|
|
store.stores.token.(*SqlTokenStore).createIndexesIfNotExists()
|
|
store.stores.emoji.(*SqlEmojiStore).createIndexesIfNotExists()
|
|
store.stores.status.(*SqlStatusStore).createIndexesIfNotExists()
|
|
store.stores.fileInfo.(*SqlFileInfoStore).createIndexesIfNotExists()
|
|
store.stores.uploadSession.(*SqlUploadSessionStore).createIndexesIfNotExists()
|
|
store.stores.job.(*SqlJobStore).createIndexesIfNotExists()
|
|
store.stores.userAccessToken.(*SqlUserAccessTokenStore).createIndexesIfNotExists()
|
|
store.stores.plugin.(*SqlPluginStore).createIndexesIfNotExists()
|
|
store.stores.TermsOfService.(SqlTermsOfServiceStore).createIndexesIfNotExists()
|
|
store.stores.productNotices.(SqlProductNoticesStore).createIndexesIfNotExists()
|
|
store.stores.UserTermsOfService.(SqlUserTermsOfServiceStore).createIndexesIfNotExists()
|
|
store.stores.linkMetadata.(*SqlLinkMetadataStore).createIndexesIfNotExists()
|
|
store.stores.group.(*SqlGroupStore).createIndexesIfNotExists()
|
|
store.stores.scheme.(*SqlSchemeStore).createIndexesIfNotExists()
|
|
store.stores.preference.(*SqlPreferenceStore).deleteUnusedFeatures()
|
|
|
|
return store
|
|
}
|
|
|
|
func setupConnection(connType string, dataSource string, settings *model.SqlSettings) *gorp.DbMap {
|
|
db, err := dbsql.Open(*settings.DriverName, dataSource)
|
|
if err != nil {
|
|
mlog.Critical("Failed to open SQL connection to err.", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitDBOpen)
|
|
}
|
|
|
|
for i := 0; i < DBPingAttempts; i++ {
|
|
mlog.Info("Pinging SQL", mlog.String("database", connType))
|
|
ctx, cancel := context.WithTimeout(context.Background(), DBPingTimeoutSecs*time.Second)
|
|
defer cancel()
|
|
err = db.PingContext(ctx)
|
|
if err == nil {
|
|
break
|
|
} else {
|
|
if i == DBPingAttempts-1 {
|
|
mlog.Critical("Failed to ping DB, server will exit.", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitPing)
|
|
} else {
|
|
mlog.Error("Failed to ping DB", mlog.Err(err), mlog.Int("retrying in seconds", DBPingTimeoutSecs))
|
|
time.Sleep(DBPingTimeoutSecs * time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
db.SetMaxIdleConns(*settings.MaxIdleConns)
|
|
db.SetMaxOpenConns(*settings.MaxOpenConns)
|
|
db.SetConnMaxLifetime(time.Duration(*settings.ConnMaxLifetimeMilliseconds) * time.Millisecond)
|
|
db.SetConnMaxIdleTime(time.Duration(*settings.ConnMaxIdleTimeMilliseconds) * time.Millisecond)
|
|
|
|
var dbmap *gorp.DbMap
|
|
|
|
connectionTimeout := time.Duration(*settings.QueryTimeout) * time.Second
|
|
|
|
if *settings.DriverName == model.DATABASE_DRIVER_MYSQL {
|
|
dbmap = &gorp.DbMap{Db: db, TypeConverter: mattermConverter{}, Dialect: gorp.MySQLDialect{Engine: "InnoDB", Encoding: "UTF8MB4"}, QueryTimeout: connectionTimeout}
|
|
} else if *settings.DriverName == model.DATABASE_DRIVER_POSTGRES {
|
|
dbmap = &gorp.DbMap{Db: db, TypeConverter: mattermConverter{}, Dialect: gorp.PostgresDialect{}, QueryTimeout: connectionTimeout}
|
|
} else {
|
|
mlog.Critical("Failed to create dialect specific driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitNoDriver)
|
|
}
|
|
|
|
if settings.Trace != nil && *settings.Trace {
|
|
dbmap.TraceOn("sql-trace:", &TraceOnAdapter{})
|
|
}
|
|
|
|
return dbmap
|
|
}
|
|
|
|
func (ss *SqlStore) SetContext(context context.Context) {
|
|
ss.context = context
|
|
}
|
|
|
|
func (ss *SqlStore) Context() context.Context {
|
|
return ss.context
|
|
}
|
|
|
|
func (ss *SqlStore) initConnection() {
|
|
ss.master = setupConnection("master", *ss.settings.DataSource, ss.settings)
|
|
|
|
if len(ss.settings.DataSourceReplicas) > 0 {
|
|
ss.Replicas = make([]*gorp.DbMap, len(ss.settings.DataSourceReplicas))
|
|
for i, replica := range ss.settings.DataSourceReplicas {
|
|
ss.Replicas[i] = setupConnection(fmt.Sprintf("replica-%v", i), replica, ss.settings)
|
|
}
|
|
}
|
|
|
|
if len(ss.settings.DataSourceSearchReplicas) > 0 {
|
|
ss.searchReplicas = make([]*gorp.DbMap, len(ss.settings.DataSourceSearchReplicas))
|
|
for i, replica := range ss.settings.DataSourceSearchReplicas {
|
|
ss.searchReplicas[i] = setupConnection(fmt.Sprintf("search-replica-%v", i), replica, ss.settings)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) DriverName() string {
|
|
return *ss.settings.DriverName
|
|
}
|
|
|
|
func (ss *SqlStore) GetCurrentSchemaVersion() string {
|
|
version, _ := ss.GetMaster().SelectStr("SELECT Value FROM Systems WHERE Name='Version'")
|
|
return version
|
|
}
|
|
|
|
// GetDbVersion returns the version of the database being used.
|
|
// If numerical is set to true, it attempts to return a numerical version string
|
|
// that can be parsed by callers.
|
|
func (ss *SqlStore) GetDbVersion(numerical bool) (string, error) {
|
|
var sqlVersion string
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
if numerical {
|
|
sqlVersion = `SHOW server_version_num`
|
|
} else {
|
|
sqlVersion = `SHOW server_version`
|
|
}
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
sqlVersion = `SELECT version()`
|
|
} else {
|
|
return "", errors.New("Not supported driver")
|
|
}
|
|
|
|
version, err := ss.GetReplica().SelectStr(sqlVersion)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
return version, nil
|
|
|
|
}
|
|
|
|
func (ss *SqlStore) GetMaster() *gorp.DbMap {
|
|
return ss.master
|
|
}
|
|
|
|
func (ss *SqlStore) GetSearchReplica() *gorp.DbMap {
|
|
ss.licenseMutex.RLock()
|
|
license := ss.license
|
|
ss.licenseMutex.RUnlock()
|
|
if license == nil {
|
|
return ss.GetMaster()
|
|
}
|
|
|
|
if len(ss.settings.DataSourceSearchReplicas) == 0 {
|
|
return ss.GetReplica()
|
|
}
|
|
|
|
rrNum := atomic.AddInt64(&ss.srCounter, 1) % int64(len(ss.searchReplicas))
|
|
return ss.searchReplicas[rrNum]
|
|
}
|
|
|
|
func (ss *SqlStore) GetReplica() *gorp.DbMap {
|
|
ss.licenseMutex.RLock()
|
|
license := ss.license
|
|
ss.licenseMutex.RUnlock()
|
|
if len(ss.settings.DataSourceReplicas) == 0 || ss.lockedToMaster || license == nil {
|
|
return ss.GetMaster()
|
|
}
|
|
|
|
rrNum := atomic.AddInt64(&ss.rrCounter, 1) % int64(len(ss.Replicas))
|
|
return ss.Replicas[rrNum]
|
|
}
|
|
|
|
func (ss *SqlStore) TotalMasterDbConnections() int {
|
|
return ss.GetMaster().Db.Stats().OpenConnections
|
|
}
|
|
|
|
func (ss *SqlStore) TotalReadDbConnections() int {
|
|
if len(ss.settings.DataSourceReplicas) == 0 {
|
|
return 0
|
|
}
|
|
|
|
count := 0
|
|
for _, db := range ss.Replicas {
|
|
count = count + db.Db.Stats().OpenConnections
|
|
}
|
|
|
|
return count
|
|
}
|
|
|
|
func (ss *SqlStore) TotalSearchDbConnections() int {
|
|
if len(ss.settings.DataSourceSearchReplicas) == 0 {
|
|
return 0
|
|
}
|
|
|
|
count := 0
|
|
for _, db := range ss.searchReplicas {
|
|
count = count + db.Db.Stats().OpenConnections
|
|
}
|
|
|
|
return count
|
|
}
|
|
|
|
func (ss *SqlStore) MarkSystemRanUnitTests() {
|
|
props, err := ss.System().Get()
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
unitTests := props[model.SYSTEM_RAN_UNIT_TESTS]
|
|
if unitTests == "" {
|
|
systemTests := &model.System{Name: model.SYSTEM_RAN_UNIT_TESTS, Value: "1"}
|
|
ss.System().Save(systemTests)
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) DoesTableExist(tableName string) bool {
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
count, err := ss.GetMaster().SelectInt(
|
|
`SELECT count(relname) FROM pg_class WHERE relname=$1`,
|
|
strings.ToLower(tableName),
|
|
)
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to check if table exists", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitTableExists)
|
|
}
|
|
|
|
return count > 0
|
|
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
|
|
count, err := ss.GetMaster().SelectInt(
|
|
`SELECT
|
|
COUNT(0) AS table_exists
|
|
FROM
|
|
information_schema.TABLES
|
|
WHERE
|
|
TABLE_SCHEMA = DATABASE()
|
|
AND TABLE_NAME = ?
|
|
`,
|
|
tableName,
|
|
)
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to check if table exists", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitTableExistsMySQL)
|
|
}
|
|
|
|
return count > 0
|
|
|
|
} else {
|
|
mlog.Critical("Failed to check if column exists because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitColumnExists)
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) DoesColumnExist(tableName string, columnName string) bool {
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
count, err := ss.GetMaster().SelectInt(
|
|
`SELECT COUNT(0)
|
|
FROM pg_attribute
|
|
WHERE attrelid = $1::regclass
|
|
AND attname = $2
|
|
AND NOT attisdropped`,
|
|
strings.ToLower(tableName),
|
|
strings.ToLower(columnName),
|
|
)
|
|
|
|
if err != nil {
|
|
if err.Error() == "pq: relation \""+strings.ToLower(tableName)+"\" does not exist" {
|
|
return false
|
|
}
|
|
|
|
mlog.Critical("Failed to check if column exists", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitDoesColumnExistsPostgres)
|
|
}
|
|
|
|
return count > 0
|
|
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
|
|
count, err := ss.GetMaster().SelectInt(
|
|
`SELECT
|
|
COUNT(0) AS column_exists
|
|
FROM
|
|
information_schema.COLUMNS
|
|
WHERE
|
|
TABLE_SCHEMA = DATABASE()
|
|
AND TABLE_NAME = ?
|
|
AND COLUMN_NAME = ?`,
|
|
tableName,
|
|
columnName,
|
|
)
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to check if column exists", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitDoesColumnExistsMySQL)
|
|
}
|
|
|
|
return count > 0
|
|
|
|
} else {
|
|
mlog.Critical("Failed to check if column exists because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitDoesColumnExistsMissing)
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) DoesTriggerExist(triggerName string) bool {
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
count, err := ss.GetMaster().SelectInt(`
|
|
SELECT
|
|
COUNT(0)
|
|
FROM
|
|
pg_trigger
|
|
WHERE
|
|
tgname = $1
|
|
`, triggerName)
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to check if trigger exists", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitGenericFailure)
|
|
}
|
|
|
|
return count > 0
|
|
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
count, err := ss.GetMaster().SelectInt(`
|
|
SELECT
|
|
COUNT(0)
|
|
FROM
|
|
information_schema.triggers
|
|
WHERE
|
|
trigger_schema = DATABASE()
|
|
AND trigger_name = ?
|
|
`, triggerName)
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to check if trigger exists", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitGenericFailure)
|
|
}
|
|
|
|
return count > 0
|
|
|
|
} else {
|
|
mlog.Critical("Failed to check if column exists because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitGenericFailure)
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) CreateColumnIfNotExists(tableName string, columnName string, mySqlColType string, postgresColType string, defaultValue string) bool {
|
|
|
|
if ss.DoesColumnExist(tableName, columnName) {
|
|
return false
|
|
}
|
|
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
_, err := ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " ADD " + columnName + " " + postgresColType + " DEFAULT '" + defaultValue + "'")
|
|
if err != nil {
|
|
mlog.Critical("Failed to create column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateColumnPostgres)
|
|
}
|
|
|
|
return true
|
|
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
_, err := ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " ADD " + columnName + " " + mySqlColType + " DEFAULT '" + defaultValue + "'")
|
|
if err != nil {
|
|
mlog.Critical("Failed to create column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateColumnMySQL)
|
|
}
|
|
|
|
return true
|
|
|
|
} else {
|
|
mlog.Critical("Failed to create column because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateColumnMissing)
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) CreateColumnIfNotExistsNoDefault(tableName string, columnName string, mySqlColType string, postgresColType string) bool {
|
|
|
|
if ss.DoesColumnExist(tableName, columnName) {
|
|
return false
|
|
}
|
|
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
_, err := ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " ADD " + columnName + " " + postgresColType)
|
|
if err != nil {
|
|
mlog.Critical("Failed to create column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateColumnPostgres)
|
|
}
|
|
|
|
return true
|
|
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
_, err := ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " ADD " + columnName + " " + mySqlColType)
|
|
if err != nil {
|
|
mlog.Critical("Failed to create column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateColumnMySQL)
|
|
}
|
|
|
|
return true
|
|
|
|
} else {
|
|
mlog.Critical("Failed to create column because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateColumnMissing)
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) RemoveColumnIfExists(tableName string, columnName string) bool {
|
|
|
|
if !ss.DoesColumnExist(tableName, columnName) {
|
|
return false
|
|
}
|
|
|
|
_, err := ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " DROP COLUMN " + columnName)
|
|
if err != nil {
|
|
mlog.Critical("Failed to drop column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRemoveColumn)
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) RemoveTableIfExists(tableName string) bool {
|
|
if !ss.DoesTableExist(tableName) {
|
|
return false
|
|
}
|
|
|
|
_, err := ss.GetMaster().ExecNoTimeout("DROP TABLE " + tableName)
|
|
if err != nil {
|
|
mlog.Critical("Failed to drop table", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRemoveTable)
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) RenameColumnIfExists(tableName string, oldColumnName string, newColumnName string, colType string) bool {
|
|
if !ss.DoesColumnExist(tableName, oldColumnName) {
|
|
return false
|
|
}
|
|
|
|
var err error
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
_, err = ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " CHANGE " + oldColumnName + " " + newColumnName + " " + colType)
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
_, err = ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " RENAME COLUMN " + oldColumnName + " TO " + newColumnName)
|
|
}
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to rename column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRenameColumn)
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) GetMaxLengthOfColumnIfExists(tableName string, columnName string) string {
|
|
if !ss.DoesColumnExist(tableName, columnName) {
|
|
return ""
|
|
}
|
|
|
|
var result string
|
|
var err error
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
result, err = ss.GetMaster().SelectStr("SELECT CHARACTER_MAXIMUM_LENGTH FROM information_schema.columns WHERE table_name = '" + tableName + "' AND COLUMN_NAME = '" + columnName + "'")
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
result, err = ss.GetMaster().SelectStr("SELECT character_maximum_length FROM information_schema.columns WHERE table_name = '" + strings.ToLower(tableName) + "' AND column_name = '" + strings.ToLower(columnName) + "'")
|
|
}
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to get max length of column", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitMaxColumn)
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
func (ss *SqlStore) AlterColumnTypeIfExists(tableName string, columnName string, mySqlColType string, postgresColType string) bool {
|
|
if !ss.DoesColumnExist(tableName, columnName) {
|
|
return false
|
|
}
|
|
|
|
var err error
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
_, err = ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " MODIFY " + columnName + " " + mySqlColType)
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
_, err = ss.GetMaster().ExecNoTimeout("ALTER TABLE " + strings.ToLower(tableName) + " ALTER COLUMN " + strings.ToLower(columnName) + " TYPE " + postgresColType)
|
|
}
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to alter column type", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitAlterColumn)
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) AlterColumnDefaultIfExists(tableName string, columnName string, mySqlColDefault *string, postgresColDefault *string) bool {
|
|
if !ss.DoesColumnExist(tableName, columnName) {
|
|
return false
|
|
}
|
|
|
|
var defaultValue string
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
// Some column types in MySQL cannot have defaults, so don't try to configure anything.
|
|
if mySqlColDefault == nil {
|
|
return true
|
|
}
|
|
|
|
defaultValue = *mySqlColDefault
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
// Postgres doesn't have the same limitation, but preserve the interface.
|
|
if postgresColDefault == nil {
|
|
return true
|
|
}
|
|
|
|
tableName = strings.ToLower(tableName)
|
|
columnName = strings.ToLower(columnName)
|
|
defaultValue = *postgresColDefault
|
|
} else {
|
|
mlog.Critical("Failed to alter column default because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitGenericFailure)
|
|
return false
|
|
}
|
|
|
|
var err error
|
|
if defaultValue == "" {
|
|
_, err = ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " ALTER COLUMN " + columnName + " DROP DEFAULT")
|
|
} else {
|
|
_, err = ss.GetMaster().ExecNoTimeout("ALTER TABLE " + tableName + " ALTER COLUMN " + columnName + " SET DEFAULT " + defaultValue)
|
|
}
|
|
|
|
if err != nil {
|
|
mlog.Critical("Failed to alter column", mlog.String("table", tableName), mlog.String("column", columnName), mlog.String("default value", defaultValue), mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitGenericFailure)
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) AlterPrimaryKey(tableName string, columnNames []string) bool {
|
|
var currentPrimaryKey string
|
|
var err error
|
|
// get the current primary key as a comma separated list of columns
|
|
switch ss.DriverName() {
|
|
case model.DATABASE_DRIVER_MYSQL:
|
|
query := `
|
|
SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index) AS PK
|
|
FROM
|
|
information_schema.statistics
|
|
WHERE
|
|
table_schema = DATABASE()
|
|
AND table_name = ?
|
|
AND index_name = 'PRIMARY'
|
|
GROUP BY
|
|
index_name`
|
|
currentPrimaryKey, err = ss.GetMaster().SelectStr(query, tableName)
|
|
case model.DATABASE_DRIVER_POSTGRES:
|
|
query := `
|
|
SELECT string_agg(a.attname, ',') AS pk
|
|
FROM
|
|
pg_constraint AS c
|
|
CROSS JOIN
|
|
(SELECT unnest(conkey) FROM pg_constraint WHERE conrelid='` + strings.ToLower(tableName) + `'::REGCLASS AND contype='p') AS cols(colnum)
|
|
INNER JOIN
|
|
pg_attribute AS a ON a.attrelid = c.conrelid
|
|
AND cols.colnum = a.attnum
|
|
WHERE
|
|
c.contype = 'p'
|
|
AND c.conrelid = '` + strings.ToLower(tableName) + `'::REGCLASS`
|
|
currentPrimaryKey, err = ss.GetMaster().SelectStr(query)
|
|
}
|
|
if err != nil {
|
|
mlog.Critical("Failed to get current primary key", mlog.String("table", tableName), mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitAlterPrimaryKey)
|
|
}
|
|
|
|
primaryKey := strings.Join(columnNames, ",")
|
|
if strings.EqualFold(currentPrimaryKey, primaryKey) {
|
|
return false
|
|
}
|
|
// alter primary key
|
|
var alterQuery string
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
alterQuery = "ALTER TABLE " + tableName + " DROP PRIMARY KEY, ADD PRIMARY KEY (" + primaryKey + ")"
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
alterQuery = "ALTER TABLE " + tableName + " DROP CONSTRAINT " + strings.ToLower(tableName) + "_pkey, ADD PRIMARY KEY (" + strings.ToLower(primaryKey) + ")"
|
|
}
|
|
_, err = ss.GetMaster().ExecNoTimeout(alterQuery)
|
|
if err != nil {
|
|
mlog.Critical("Failed to alter primary key", mlog.String("table", tableName), mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitAlterPrimaryKey)
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) CreateUniqueIndexIfNotExists(indexName string, tableName string, columnName string) bool {
|
|
return ss.createIndexIfNotExists(indexName, tableName, []string{columnName}, IndexTypeDefault, true)
|
|
}
|
|
|
|
func (ss *SqlStore) CreateIndexIfNotExists(indexName string, tableName string, columnName string) bool {
|
|
return ss.createIndexIfNotExists(indexName, tableName, []string{columnName}, IndexTypeDefault, false)
|
|
}
|
|
|
|
func (ss *SqlStore) CreateCompositeIndexIfNotExists(indexName string, tableName string, columnNames []string) bool {
|
|
return ss.createIndexIfNotExists(indexName, tableName, columnNames, IndexTypeDefault, false)
|
|
}
|
|
|
|
func (ss *SqlStore) CreateUniqueCompositeIndexIfNotExists(indexName string, tableName string, columnNames []string) bool {
|
|
return ss.createIndexIfNotExists(indexName, tableName, columnNames, IndexTypeDefault, true)
|
|
}
|
|
|
|
func (ss *SqlStore) CreateFullTextIndexIfNotExists(indexName string, tableName string, columnName string) bool {
|
|
return ss.createIndexIfNotExists(indexName, tableName, []string{columnName}, IndexTypeFullText, false)
|
|
}
|
|
|
|
func (ss *SqlStore) createIndexIfNotExists(indexName string, tableName string, columnNames []string, indexType string, unique bool) bool {
|
|
|
|
uniqueStr := ""
|
|
if unique {
|
|
uniqueStr = "UNIQUE "
|
|
}
|
|
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
_, errExists := ss.GetMaster().SelectStr("SELECT $1::regclass", indexName)
|
|
// It should fail if the index does not exist
|
|
if errExists == nil {
|
|
return false
|
|
}
|
|
|
|
query := ""
|
|
if indexType == IndexTypeFullText {
|
|
if len(columnNames) != 1 {
|
|
mlog.Critical("Unable to create multi column full text index")
|
|
os.Exit(ExitCreateIndexPostgres)
|
|
}
|
|
columnName := columnNames[0]
|
|
postgresColumnNames := convertMySQLFullTextColumnsToPostgres(columnName)
|
|
query = "CREATE INDEX " + indexName + " ON " + tableName + " USING gin(to_tsvector('english', " + postgresColumnNames + "))"
|
|
} else {
|
|
query = "CREATE " + uniqueStr + "INDEX " + indexName + " ON " + tableName + " (" + strings.Join(columnNames, ", ") + ")"
|
|
}
|
|
|
|
_, err := ss.GetMaster().ExecNoTimeout(query)
|
|
if err != nil {
|
|
mlog.Critical("Failed to create index", mlog.Err(errExists), mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateIndexPostgres)
|
|
}
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
|
|
count, err := ss.GetMaster().SelectInt("SELECT COUNT(0) AS index_exists FROM information_schema.statistics WHERE TABLE_SCHEMA = DATABASE() and table_name = ? AND index_name = ?", tableName, indexName)
|
|
if err != nil {
|
|
mlog.Critical("Failed to check index", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateIndexMySQL)
|
|
}
|
|
|
|
if count > 0 {
|
|
return false
|
|
}
|
|
|
|
fullTextIndex := ""
|
|
if indexType == IndexTypeFullText {
|
|
fullTextIndex = " FULLTEXT "
|
|
}
|
|
|
|
_, err = ss.GetMaster().ExecNoTimeout("CREATE " + uniqueStr + fullTextIndex + " INDEX " + indexName + " ON " + tableName + " (" + strings.Join(columnNames, ", ") + ")")
|
|
if err != nil {
|
|
mlog.Critical("Failed to create index", mlog.String("table", tableName), mlog.String("index_name", indexName), mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateIndexFullMySQL)
|
|
}
|
|
} else {
|
|
mlog.Critical("Failed to create index because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitCreateIndexMissing)
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (ss *SqlStore) RemoveIndexIfExists(indexName string, tableName string) bool {
|
|
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
_, err := ss.GetMaster().SelectStr("SELECT $1::regclass", indexName)
|
|
// It should fail if the index does not exist
|
|
if err != nil {
|
|
return false
|
|
}
|
|
|
|
_, err = ss.GetMaster().ExecNoTimeout("DROP INDEX " + indexName)
|
|
if err != nil {
|
|
mlog.Critical("Failed to remove index", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRemoveIndexPostgres)
|
|
}
|
|
|
|
return true
|
|
} else if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
|
|
count, err := ss.GetMaster().SelectInt("SELECT COUNT(0) AS index_exists FROM information_schema.statistics WHERE TABLE_SCHEMA = DATABASE() and table_name = ? AND index_name = ?", tableName, indexName)
|
|
if err != nil {
|
|
mlog.Critical("Failed to check index", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRemoveIndexMySQL)
|
|
}
|
|
|
|
if count <= 0 {
|
|
return false
|
|
}
|
|
|
|
_, err = ss.GetMaster().ExecNoTimeout("DROP INDEX " + indexName + " ON " + tableName)
|
|
if err != nil {
|
|
mlog.Critical("Failed to remove index", mlog.Err(err))
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRemoveIndexMySQL)
|
|
}
|
|
} else {
|
|
mlog.Critical("Failed to create index because of missing driver")
|
|
time.Sleep(time.Second)
|
|
os.Exit(ExitRemoveIndexMissing)
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func IsUniqueConstraintError(err error, indexName []string) bool {
|
|
unique := false
|
|
if pqErr, ok := err.(*pq.Error); ok && pqErr.Code == "23505" {
|
|
unique = true
|
|
}
|
|
|
|
if mysqlErr, ok := err.(*mysql.MySQLError); ok && mysqlErr.Number == 1062 {
|
|
unique = true
|
|
}
|
|
|
|
field := false
|
|
for _, contain := range indexName {
|
|
if strings.Contains(err.Error(), contain) {
|
|
field = true
|
|
break
|
|
}
|
|
}
|
|
|
|
return unique && field
|
|
}
|
|
|
|
func (ss *SqlStore) GetAllConns() []*gorp.DbMap {
|
|
all := make([]*gorp.DbMap, len(ss.Replicas)+1)
|
|
copy(all, ss.Replicas)
|
|
all[len(ss.Replicas)] = ss.master
|
|
return all
|
|
}
|
|
|
|
// RecycleDBConnections closes active connections by setting the max conn lifetime
|
|
// to d, and then resets them back to their original duration.
|
|
func (ss *SqlStore) RecycleDBConnections(d time.Duration) {
|
|
// Get old time.
|
|
originalDuration := time.Duration(*ss.settings.ConnMaxLifetimeMilliseconds) * time.Millisecond
|
|
// Set the max lifetimes for all connections.
|
|
for _, conn := range ss.GetAllConns() {
|
|
conn.Db.SetConnMaxLifetime(d)
|
|
}
|
|
// Wait for that period with an additional 2 seconds of scheduling delay.
|
|
time.Sleep(d + 2*time.Second)
|
|
// Reset max lifetime back to original value.
|
|
for _, conn := range ss.GetAllConns() {
|
|
conn.Db.SetConnMaxLifetime(originalDuration)
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) Close() {
|
|
ss.master.Db.Close()
|
|
for _, replica := range ss.Replicas {
|
|
replica.Db.Close()
|
|
}
|
|
|
|
for _, replica := range ss.searchReplicas {
|
|
replica.Db.Close()
|
|
}
|
|
}
|
|
|
|
func (ss *SqlStore) LockToMaster() {
|
|
ss.lockedToMaster = true
|
|
}
|
|
|
|
func (ss *SqlStore) UnlockFromMaster() {
|
|
ss.lockedToMaster = false
|
|
}
|
|
|
|
func (ss *SqlStore) Team() store.TeamStore {
|
|
return ss.stores.team
|
|
}
|
|
|
|
func (ss *SqlStore) Channel() store.ChannelStore {
|
|
return ss.stores.channel
|
|
}
|
|
|
|
func (ss *SqlStore) Post() store.PostStore {
|
|
return ss.stores.post
|
|
}
|
|
|
|
func (ss *SqlStore) User() store.UserStore {
|
|
return ss.stores.user
|
|
}
|
|
|
|
func (ss *SqlStore) Bot() store.BotStore {
|
|
return ss.stores.bot
|
|
}
|
|
|
|
func (ss *SqlStore) Session() store.SessionStore {
|
|
return ss.stores.session
|
|
}
|
|
|
|
func (ss *SqlStore) Audit() store.AuditStore {
|
|
return ss.stores.audit
|
|
}
|
|
|
|
func (ss *SqlStore) ClusterDiscovery() store.ClusterDiscoveryStore {
|
|
return ss.stores.cluster
|
|
}
|
|
|
|
func (ss *SqlStore) Compliance() store.ComplianceStore {
|
|
return ss.stores.compliance
|
|
}
|
|
|
|
func (ss *SqlStore) OAuth() store.OAuthStore {
|
|
return ss.stores.oauth
|
|
}
|
|
|
|
func (ss *SqlStore) System() store.SystemStore {
|
|
return ss.stores.system
|
|
}
|
|
|
|
func (ss *SqlStore) Webhook() store.WebhookStore {
|
|
return ss.stores.webhook
|
|
}
|
|
|
|
func (ss *SqlStore) Command() store.CommandStore {
|
|
return ss.stores.command
|
|
}
|
|
|
|
func (ss *SqlStore) CommandWebhook() store.CommandWebhookStore {
|
|
return ss.stores.commandWebhook
|
|
}
|
|
|
|
func (ss *SqlStore) Preference() store.PreferenceStore {
|
|
return ss.stores.preference
|
|
}
|
|
|
|
func (ss *SqlStore) License() store.LicenseStore {
|
|
return ss.stores.license
|
|
}
|
|
|
|
func (ss *SqlStore) Token() store.TokenStore {
|
|
return ss.stores.token
|
|
}
|
|
|
|
func (ss *SqlStore) Emoji() store.EmojiStore {
|
|
return ss.stores.emoji
|
|
}
|
|
|
|
func (ss *SqlStore) Status() store.StatusStore {
|
|
return ss.stores.status
|
|
}
|
|
|
|
func (ss *SqlStore) FileInfo() store.FileInfoStore {
|
|
return ss.stores.fileInfo
|
|
}
|
|
|
|
func (ss *SqlStore) UploadSession() store.UploadSessionStore {
|
|
return ss.stores.uploadSession
|
|
}
|
|
|
|
func (ss *SqlStore) Reaction() store.ReactionStore {
|
|
return ss.stores.reaction
|
|
}
|
|
|
|
func (ss *SqlStore) Job() store.JobStore {
|
|
return ss.stores.job
|
|
}
|
|
|
|
func (ss *SqlStore) UserAccessToken() store.UserAccessTokenStore {
|
|
return ss.stores.userAccessToken
|
|
}
|
|
|
|
func (ss *SqlStore) ChannelMemberHistory() store.ChannelMemberHistoryStore {
|
|
return ss.stores.channelMemberHistory
|
|
}
|
|
|
|
func (ss *SqlStore) Plugin() store.PluginStore {
|
|
return ss.stores.plugin
|
|
}
|
|
|
|
func (ss *SqlStore) Thread() store.ThreadStore {
|
|
return ss.stores.thread
|
|
}
|
|
|
|
func (ss *SqlStore) Role() store.RoleStore {
|
|
return ss.stores.role
|
|
}
|
|
|
|
func (ss *SqlStore) TermsOfService() store.TermsOfServiceStore {
|
|
return ss.stores.TermsOfService
|
|
}
|
|
|
|
func (ss *SqlStore) ProductNotices() store.ProductNoticesStore {
|
|
return ss.stores.productNotices
|
|
}
|
|
|
|
func (ss *SqlStore) UserTermsOfService() store.UserTermsOfServiceStore {
|
|
return ss.stores.UserTermsOfService
|
|
}
|
|
|
|
func (ss *SqlStore) Scheme() store.SchemeStore {
|
|
return ss.stores.scheme
|
|
}
|
|
|
|
func (ss *SqlStore) Group() store.GroupStore {
|
|
return ss.stores.group
|
|
}
|
|
|
|
func (ss *SqlStore) LinkMetadata() store.LinkMetadataStore {
|
|
return ss.stores.linkMetadata
|
|
}
|
|
|
|
func (ss *SqlStore) DropAllTables() {
|
|
ss.master.TruncateTables()
|
|
}
|
|
|
|
func (ss *SqlStore) getQueryBuilder() sq.StatementBuilderType {
|
|
builder := sq.StatementBuilder.PlaceholderFormat(sq.Question)
|
|
if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
|
|
builder = builder.PlaceholderFormat(sq.Dollar)
|
|
}
|
|
return builder
|
|
}
|
|
|
|
func (ss *SqlStore) CheckIntegrity() <-chan model.IntegrityCheckResult {
|
|
results := make(chan model.IntegrityCheckResult)
|
|
go CheckRelationalIntegrity(ss, results)
|
|
return results
|
|
}
|
|
|
|
func (ss *SqlStore) UpdateLicense(license *model.License) {
|
|
ss.licenseMutex.Lock()
|
|
defer ss.licenseMutex.Unlock()
|
|
ss.license = license
|
|
}
|
|
|
|
func (ss *SqlStore) GetLicense() *model.License {
|
|
return ss.license
|
|
}
|
|
|
|
func (ss *SqlStore) migrate(direction migrationDirection) error {
|
|
var driver database.Driver
|
|
var err error
|
|
|
|
// When WithInstance is used in golang-migrate, the underlying driver connections are not tracked.
|
|
// So we will have to open a fresh connection for migrations and explicitly close it when all is done.
|
|
dataSource := ss.appendMultipleStatementsFlag(*ss.settings.DataSource)
|
|
conn := setupConnection("migrations", dataSource, ss.settings)
|
|
defer conn.Db.Close()
|
|
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
driver, err = mysqlmigrate.WithInstance(conn.Db, &mysqlmigrate.Config{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
} else {
|
|
driver, err = postgres.WithInstance(conn.Db, &postgres.Config{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
var assetNamesForDriver []string
|
|
for _, assetName := range migrations.AssetNames() {
|
|
if strings.HasPrefix(assetName, ss.DriverName()) {
|
|
assetNamesForDriver = append(assetNamesForDriver, filepath.Base(assetName))
|
|
}
|
|
}
|
|
|
|
source := bindata.Resource(assetNamesForDriver, func(name string) ([]byte, error) {
|
|
return migrations.Asset(filepath.Join(ss.DriverName(), name))
|
|
})
|
|
|
|
sourceDriver, err := bindata.WithInstance(source)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
migrations, err := migrate.NewWithInstance("go-bindata",
|
|
sourceDriver,
|
|
ss.DriverName(),
|
|
driver)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer migrations.Close()
|
|
|
|
switch direction {
|
|
case migrationsDirectionUp:
|
|
err = migrations.Up()
|
|
case migrationsDirectionDown:
|
|
err = migrations.Down()
|
|
default:
|
|
return errors.New(fmt.Sprintf("unsupported migration direction %s", direction))
|
|
}
|
|
|
|
if err != nil && err != migrate.ErrNoChange && !errors.Is(err, os.ErrNotExist) {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (ss *SqlStore) appendMultipleStatementsFlag(dataSource string) string {
|
|
// We need to tell the MySQL driver that we want to use multiStatements
|
|
// in order to make migrations work.
|
|
if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
|
|
u, err := url.Parse(dataSource)
|
|
if err != nil {
|
|
mlog.Critical("Invalid database url found", mlog.Err(err))
|
|
os.Exit(ExitGenericFailure)
|
|
}
|
|
q := u.Query()
|
|
q.Set("multiStatements", "true")
|
|
u.RawQuery = q.Encode()
|
|
return u.String()
|
|
}
|
|
|
|
return dataSource
|
|
}
|
|
|
|
type mattermConverter struct{}
|
|
|
|
func (me mattermConverter) ToDb(val interface{}) (interface{}, error) {
|
|
|
|
switch t := val.(type) {
|
|
case model.StringMap:
|
|
return model.MapToJson(t), nil
|
|
case map[string]string:
|
|
return model.MapToJson(model.StringMap(t)), nil
|
|
case model.StringArray:
|
|
return model.ArrayToJson(t), nil
|
|
case model.StringInterface:
|
|
return model.StringInterfaceToJson(t), nil
|
|
case map[string]interface{}:
|
|
return model.StringInterfaceToJson(model.StringInterface(t)), nil
|
|
case JSONSerializable:
|
|
return t.ToJson(), nil
|
|
case *opengraph.OpenGraph:
|
|
return json.Marshal(t)
|
|
}
|
|
|
|
return val, nil
|
|
}
|
|
|
|
func (me mattermConverter) FromDb(target interface{}) (gorp.CustomScanner, bool) {
|
|
switch target.(type) {
|
|
case *model.StringMap:
|
|
binder := func(holder, target interface{}) error {
|
|
s, ok := holder.(*string)
|
|
if !ok {
|
|
return errors.New(i18n.T("store.sql.convert_string_map"))
|
|
}
|
|
b := []byte(*s)
|
|
return json.Unmarshal(b, target)
|
|
}
|
|
return gorp.CustomScanner{Holder: new(string), Target: target, Binder: binder}, true
|
|
case *map[string]string:
|
|
binder := func(holder, target interface{}) error {
|
|
s, ok := holder.(*string)
|
|
if !ok {
|
|
return errors.New(i18n.T("store.sql.convert_string_map"))
|
|
}
|
|
b := []byte(*s)
|
|
return json.Unmarshal(b, target)
|
|
}
|
|
return gorp.CustomScanner{Holder: new(string), Target: target, Binder: binder}, true
|
|
case *model.StringArray:
|
|
binder := func(holder, target interface{}) error {
|
|
s, ok := holder.(*string)
|
|
if !ok {
|
|
return errors.New(i18n.T("store.sql.convert_string_array"))
|
|
}
|
|
b := []byte(*s)
|
|
return json.Unmarshal(b, target)
|
|
}
|
|
return gorp.CustomScanner{Holder: new(string), Target: target, Binder: binder}, true
|
|
case *model.StringInterface:
|
|
binder := func(holder, target interface{}) error {
|
|
s, ok := holder.(*string)
|
|
if !ok {
|
|
return errors.New(i18n.T("store.sql.convert_string_interface"))
|
|
}
|
|
b := []byte(*s)
|
|
return json.Unmarshal(b, target)
|
|
}
|
|
return gorp.CustomScanner{Holder: new(string), Target: target, Binder: binder}, true
|
|
case *map[string]interface{}:
|
|
binder := func(holder, target interface{}) error {
|
|
s, ok := holder.(*string)
|
|
if !ok {
|
|
return errors.New(i18n.T("store.sql.convert_string_interface"))
|
|
}
|
|
b := []byte(*s)
|
|
return json.Unmarshal(b, target)
|
|
}
|
|
return gorp.CustomScanner{Holder: new(string), Target: target, Binder: binder}, true
|
|
}
|
|
|
|
return gorp.CustomScanner{}, false
|
|
}
|
|
|
|
type JSONSerializable interface {
|
|
ToJson() string
|
|
}
|
|
|
|
func convertMySQLFullTextColumnsToPostgres(columnNames string) string {
|
|
columns := strings.Split(columnNames, ", ")
|
|
concatenatedColumnNames := ""
|
|
for i, c := range columns {
|
|
concatenatedColumnNames += c
|
|
if i < len(columns)-1 {
|
|
concatenatedColumnNames += " || ' ' || "
|
|
}
|
|
}
|
|
|
|
return concatenatedColumnNames
|
|
}
|
|
|
|
// IsDuplicate checks whether an error is a duplicate key error, which comes when processes are competing on creating the same
|
|
// tables in the database.
|
|
func IsDuplicate(err error) bool {
|
|
var pqErr *pq.Error
|
|
var mysqlErr *mysql.MySQLError
|
|
switch {
|
|
case errors.As(errors.Cause(err), &pqErr):
|
|
if pqErr.Code == PGDupTableErrorCode {
|
|
return true
|
|
}
|
|
case errors.As(errors.Cause(err), &mysqlErr):
|
|
if mysqlErr.Number == MySQLDupTableErrorCode {
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// VersionString converts an integer representation of a DB version
|
|
// to a pretty-printed string.
|
|
// Postgres doesn't follow three-part version numbers from 10.0 onwards:
|
|
// https://www.postgresql.org/docs/13/libpq-status.html#LIBPQ-PQSERVERVERSION.
|
|
func VersionString(v int) string {
|
|
minor := v % 10000
|
|
major := v / 10000
|
|
return strconv.Itoa(major) + "." + strconv.Itoa(minor)
|
|
}
|