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
|
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package sqlstore
import (
"database/sql"
"fmt"
"net/http"
"github.com/mattermost/mattermost-server/model"
"github.com/mattermost/mattermost-server/store"
)
const (
DEFAULT_PLUGIN_KEY_FETCH_LIMIT = 10
)
type SqlPluginStore struct {
SqlStore
}
func NewSqlPluginStore(sqlStore SqlStore) store.PluginStore {
s := &SqlPluginStore{sqlStore}
for _, db := range sqlStore.GetAllConns() {
table := db.AddTableWithName(model.PluginKeyValue{}, "PluginKeyValueStore").SetKeys(false, "PluginId", "Key")
table.ColMap("PluginId").SetMaxSize(190)
table.ColMap("Key").SetMaxSize(50)
table.ColMap("Value").SetMaxSize(8192)
}
return s
}
func (ps SqlPluginStore) CreateIndexesIfNotExists() {
}
func (ps SqlPluginStore) SaveOrUpdate(kv *model.PluginKeyValue) store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
if result.Err = kv.IsValid(); result.Err != nil {
return
}
if ps.DriverName() == model.DATABASE_DRIVER_POSTGRES {
// Unfortunately PostgreSQL pre-9.5 does not have an atomic upsert, so we use
// separate update and insert queries to accomplish our upsert
if rowsAffected, err := ps.GetMaster().Update(kv); err != nil {
result.Err = model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return
} else if rowsAffected == 0 {
// No rows were affected by the update, so let's try an insert
if err := ps.GetMaster().Insert(kv); err != nil {
// If the error is from unique constraints violation, it's the result of a
// valid race and we can report success. Otherwise we have a real error and
// need to return it
if !IsUniqueConstraintError(err, []string{"PRIMARY", "PluginId", "Key", "PKey"}) {
result.Err = model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return
}
}
}
} else if ps.DriverName() == model.DATABASE_DRIVER_MYSQL {
if _, err := ps.GetMaster().Exec("INSERT INTO PluginKeyValueStore (PluginId, PKey, PValue, ExpireAt) VALUES(:PluginId, :Key, :Value, :ExpireAt) ON DUPLICATE KEY UPDATE PValue = :Value, ExpireAt = :ExpireAt", map[string]interface{}{"PluginId": kv.PluginId, "Key": kv.Key, "Value": kv.Value, "ExpireAt": kv.ExpireAt}); err != nil {
result.Err = model.NewAppError("SqlPluginStore.SaveOrUpdate", "store.sql_plugin_store.save.app_error", nil, err.Error(), http.StatusInternalServerError)
return
}
}
result.Data = kv
})
}
func (ps SqlPluginStore) Get(pluginId, key string) store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
var kv *model.PluginKeyValue
currentTime := model.GetMillis()
if err := ps.GetReplica().SelectOne(&kv, "SELECT * FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key AND (ExpireAt = 0 OR ExpireAt > :CurrentTime)", map[string]interface{}{"PluginId": pluginId, "Key": key, "CurrentTime": currentTime}); err != nil {
if err == sql.ErrNoRows {
result.Err = model.NewAppError("SqlPluginStore.Get", "store.sql_plugin_store.get.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", pluginId, key, err.Error()), http.StatusNotFound)
} else {
result.Err = model.NewAppError("SqlPluginStore.Get", "store.sql_plugin_store.get.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", pluginId, key, err.Error()), http.StatusInternalServerError)
}
} else {
result.Data = kv
}
})
}
func (ps SqlPluginStore) Delete(pluginId, key string) store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
if _, err := ps.GetMaster().Exec("DELETE FROM PluginKeyValueStore WHERE PluginId = :PluginId AND PKey = :Key", map[string]interface{}{"PluginId": pluginId, "Key": key}); err != nil {
result.Err = model.NewAppError("SqlPluginStore.Delete", "store.sql_plugin_store.delete.app_error", nil, fmt.Sprintf("plugin_id=%v, key=%v, err=%v", pluginId, key, err.Error()), http.StatusInternalServerError)
} else {
result.Data = true
}
})
}
func (ps SqlPluginStore) DeleteAllForPlugin(pluginId string) store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
if _, err := ps.GetMaster().Exec("DELETE FROM PluginKeyValueStore WHERE PluginId = :PluginId", map[string]interface{}{"PluginId": pluginId}); err != nil {
result.Err = model.NewAppError("SqlPluginStore.Delete", "store.sql_plugin_store.delete.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
} else {
result.Data = true
}
})
}
func (ps SqlPluginStore) DeleteAllExpired() store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
currentTime := model.GetMillis()
if _, err := ps.GetMaster().Exec("DELETE FROM PluginKeyValueStore WHERE ExpireAt != 0 AND ExpireAt < :CurrentTime", map[string]interface{}{"CurrentTime": currentTime}); err != nil {
result.Err = model.NewAppError("SqlPluginStore.Delete", "store.sql_plugin_store.delete.app_error", nil, fmt.Sprintf("current_time=%v, err=%v", currentTime, err.Error()), http.StatusInternalServerError)
} else {
result.Data = true
}
})
}
func (ps SqlPluginStore) List(pluginId string, offset int, limit int) store.StoreChannel {
if limit <= 0 {
limit = DEFAULT_PLUGIN_KEY_FETCH_LIMIT
}
if offset <= 0 {
offset = 0
}
return store.Do(func(result *store.StoreResult) {
var keys []string
_, err := ps.GetReplica().Select(&keys, "SELECT PKey FROM PluginKeyValueStore WHERE PluginId = :PluginId order by PKey limit :Limit offset :Offset", map[string]interface{}{"PluginId": pluginId, "Limit": limit, "Offset": offset})
if err != nil {
result.Err = model.NewAppError("SqlPluginStore.List", "store.sql_plugin_store.list.app_error", nil, fmt.Sprintf("plugin_id=%v, err=%v", pluginId, err.Error()), http.StatusInternalServerError)
} else {
result.Data = keys
}
})
}
|
