From ffe591613c828e76579b29e72b159ce44a76824b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Efe=20G=C3=B6kdemir?= Date: Thu, 29 Oct 2026 03:11:02 +0300 Subject: [PATCH] fix(plugin): sync runtime state across replicas --- .../controller_admin/plugin_controller.go | 19 +- internal/repo/config/config_repo.go | 41 ++++ internal/service/config/config_service.go | 9 + .../new_question_notification_test.go | 6 + .../plugin_common/plugin_common_service.go | 189 ++++++++++++--- .../plugin_common_service_test.go | 228 ++++++++++++++++++ plugin/captcha.go | 12 +- plugin/cdn.go | 12 +- plugin/plugin.go | 68 +++++- 9 files changed, 512 insertions(+), 72 deletions(-) create mode 100644 internal/service/plugin_common/plugin_common_service_test.go diff --git a/internal/controller_admin/plugin_controller.go b/internal/controller_admin/plugin_controller.go index 5933f8895..760dac67f 100644 --- a/internal/controller_admin/plugin_controller.go +++ b/internal/controller_admin/plugin_controller.go @@ -20,8 +20,6 @@ package controller_admin import ( - "encoding/json" - "github.com/apache/answer/internal/base/handler" "github.com/apache/answer/internal/schema" "github.com/apache/answer/internal/service/plugin_common" @@ -148,8 +146,7 @@ func (pc *PluginController) UpdatePluginStatus(ctx *gin.Context) { return } - plugin.StatusManager.Enable(req.PluginSlugName, req.Enabled) - err := pc.pluginCommonService.UpdatePluginStatus(ctx) + err := pc.pluginCommonService.UpdatePluginStatus(ctx, req.PluginSlugName, req.Enabled) handler.HandleResponse(ctx, err, nil) } @@ -207,18 +204,6 @@ func (pc *PluginController) UpdatePluginConfig(ctx *gin.Context) { return } - configFields, _ := json.Marshal(req.ConfigFields) - err := plugin.CallConfig(func(fn plugin.Config) error { - if fn.Info().SlugName == req.PluginSlugName { - return fn.ConfigReceiver(configFields) - } - return nil - }) - if err != nil { - handler.HandleResponse(ctx, err, nil) - return - } - - err = pc.pluginCommonService.UpdatePluginConfig(ctx, req) + err := pc.pluginCommonService.UpdatePluginConfig(ctx, req) handler.HandleResponse(ctx, err, nil) } diff --git a/internal/repo/config/config_repo.go b/internal/repo/config/config_repo.go index bf0035cb6..fb3952cd7 100644 --- a/internal/repo/config/config_repo.go +++ b/internal/repo/config/config_repo.go @@ -141,3 +141,44 @@ func (cr configRepo) UpdateConfig(ctx context.Context, key string, value string) } return } + +func (cr configRepo) CompareAndSwapConfig( + ctx context.Context, key, oldValue, newValue string, +) (swapped bool, err error) { + oldConfig := &entity.Config{Key: key} + exist, err := cr.data.DB.Context(ctx).Get(oldConfig) + if err != nil { + return false, errors.InternalServer(reason.DatabaseError).WithError(err).WithStack() + } + if !exist { + return false, errors.BadRequest(reason.ObjectNotFound) + } + if oldConfig.Value != oldValue { + return false, nil + } + + affected, err := cr.data.DB.Context(ctx).ID(oldConfig.ID).Where("value = ?", oldValue). + Update(&entity.Config{Value: newValue}) + if err != nil { + return false, errors.InternalServer(reason.DatabaseError).WithError(err).WithStack() + } + if affected == 0 { + if oldValue != newValue { + return false, nil + } + current, err := cr.GetConfigByKeyFromDB(ctx, key) + if err != nil { + return false, err + } + return current.Value == newValue, nil + } + + // Invalidate both cached forms. The next reader will load the latest DB value. + if err := cr.data.Cache.Del(ctx, constant.ConfigKEY2ContentCacheKeyPrefix+key); err != nil { + log.Error(err) + } + if err := cr.data.Cache.Del(ctx, fmt.Sprintf("%s%d", constant.ConfigID2KEYCacheKeyPrefix, oldConfig.ID)); err != nil { + log.Error(err) + } + return true, nil +} diff --git a/internal/service/config/config_service.go b/internal/service/config/config_service.go index 0848d6040..006580dcc 100644 --- a/internal/service/config/config_service.go +++ b/internal/service/config/config_service.go @@ -33,6 +33,7 @@ type ConfigRepo interface { GetConfigByKey(ctx context.Context, key string) (c *entity.Config, err error) GetConfigByKeyFromDB(ctx context.Context, key string) (c *entity.Config, err error) UpdateConfig(ctx context.Context, key, value string) (err error) + CompareAndSwapConfig(ctx context.Context, key, oldValue, newValue string) (swapped bool, err error) } // ConfigService user service @@ -116,3 +117,11 @@ func (cs *ConfigService) GetIDByKey(ctx context.Context, key string) (id int, er func (cs *ConfigService) UpdateConfig(ctx context.Context, key, value string) (err error) { return cs.configRepo.UpdateConfig(ctx, key, value) } + +// CompareAndSwapConfig updates a config only when its current value still +// matches oldValue. +func (cs *ConfigService) CompareAndSwapConfig( + ctx context.Context, key, oldValue, newValue string, +) (swapped bool, err error) { + return cs.configRepo.CompareAndSwapConfig(ctx, key, oldValue, newValue) +} diff --git a/internal/service/notification/new_question_notification_test.go b/internal/service/notification/new_question_notification_test.go index 3bb6a3dd8..db6a3b0d7 100644 --- a/internal/service/notification/new_question_notification_test.go +++ b/internal/service/notification/new_question_notification_test.go @@ -674,6 +674,12 @@ func (newQuestionNotificationTestConfigRepo) UpdateConfig(context.Context, strin return nil } +func (newQuestionNotificationTestConfigRepo) CompareAndSwapConfig( + context.Context, string, string, string, +) (bool, error) { + return false, nil +} + type newQuestionNotificationTestEmailRepo struct { codesByUserID map[string][]string } diff --git a/internal/service/plugin_common/plugin_common_service.go b/internal/service/plugin_common/plugin_common_service.go index 2c1a72fda..6045adc23 100644 --- a/internal/service/plugin_common/plugin_common_service.go +++ b/internal/service/plugin_common/plugin_common_service.go @@ -22,6 +22,9 @@ package plugin_common import ( "context" "encoding/json" + "fmt" + "sync" + "time" "github.com/apache/answer/internal/base/data" "github.com/apache/answer/internal/repo/search_sync" @@ -60,8 +63,16 @@ type PluginCommonService struct { pluginUserConfigRepo PluginUserConfigRepo data *data.Data importerService *importer.ImporterService + pluginStatusMu sync.Mutex + pluginConfigMu sync.Mutex + pluginConfigValues map[string]string } +const ( + pluginConfigRefreshInterval = 10 * time.Second + pluginStatusUpdateAttempts = 10 +) + // NewPluginCommonService new report service func NewPluginCommonService( pluginConfigRepo PluginConfigRepo, @@ -76,36 +87,91 @@ func NewPluginCommonService( pluginUserConfigRepo: pluginUserConfigRepo, data: data, importerService: importerService, + pluginConfigValues: make(map[string]string), } p.initPluginData() + p.startPluginConfigRefresh() return p } // UpdatePluginStatus update plugin status -func (ps *PluginCommonService) UpdatePluginStatus(ctx context.Context) (err error) { - content, err := plugin.StatusManager.MarshalJSON() - if err != nil { - return errors.InternalServer(reason.UnknownError).WithError(err) +func (ps *PluginCommonService) UpdatePluginStatus(ctx context.Context, pluginSlugName string, enabled bool) error { + ps.pluginStatusMu.Lock() + defer ps.pluginStatusMu.Unlock() + + for range pluginStatusUpdateAttempts { + current, err := ps.configService.GetStringValueFromDB(ctx, constant.PluginStatus) + if err != nil { + return err + } + + updated, err := plugin.StatusManager.UpdateJSON([]byte(current), pluginSlugName, enabled) + if err != nil { + return errors.InternalServer(reason.UnknownError).WithError(err) + } + + swapped, err := ps.configService.CompareAndSwapConfig( + ctx, constant.PluginStatus, current, string(updated)) + if err != nil { + return err + } + if !swapped { + continue + } + + if err := plugin.StatusManager.UnmarshalJSON(updated); err != nil { + return errors.InternalServer(reason.UnknownError).WithError(err) + } + return nil } - return ps.configService.UpdateConfig(ctx, constant.PluginStatus, string(content)) + + return errors.InternalServer(reason.UnknownError).WithError( + fmt.Errorf("plugin status changed concurrently; retry the request")) } // UpdatePluginConfig update plugin config func (ps *PluginCommonService) UpdatePluginConfig(ctx context.Context, req *schema.UpdatePluginConfigReq) (err error) { - configValue, _ := json.Marshal(req.ConfigFields) + configValue, err := json.Marshal(req.ConfigFields) + if err != nil { + return errors.InternalServer(reason.UnknownError).WithError(err) + } + + ps.pluginConfigMu.Lock() + err = plugin.CallConfig(func(fn plugin.Config) error { + if fn.Info().SlugName == req.PluginSlugName { + return fn.ConfigReceiver(configValue) + } + return nil + }) + if err != nil { + ps.pluginConfigMu.Unlock() + return err + } + err = ps.pluginConfigRepo.SavePluginConfig(ctx, req.PluginSlugName, string(configValue)) if err != nil { + ps.pluginConfigMu.Unlock() return err } + if ps.pluginConfigValues == nil { + ps.pluginConfigValues = make(map[string]string) + } + ps.pluginConfigValues[req.PluginSlugName] = string(configValue) + ps.pluginConfigMu.Unlock() + ps.registerPluginConfigDependents(ctx, req.PluginSlugName) + return nil +} + +func (ps *PluginCommonService) registerPluginConfigDependents(ctx context.Context, pluginSlugName string) { _ = plugin.CallSearch(func(search plugin.Search) error { - if search.Info().SlugName == req.PluginSlugName { + if search.Info().SlugName == pluginSlugName { search.RegisterSyncer(ctx, search_sync.NewPluginSyncer(ps.data)) } return nil }) _ = plugin.CallVectorSearch(func(vs plugin.VectorSearch) error { - if vs.Info().SlugName == req.PluginSlugName { + if vs.Info().SlugName == pluginSlugName { vs.RegisterSyncer(ctx, vector_search_sync.NewPluginSyncer(ps.data)) } return nil @@ -114,7 +180,6 @@ func (ps *PluginCommonService) UpdatePluginConfig(ctx context.Context, req *sche importer.RegisterImporterFunc(ctx, ps.importerService.NewImporterFunc()) return nil }) - return nil } // UpdatePluginUserConfig update plugin config @@ -150,38 +215,14 @@ func (ps *PluginCommonService) initPluginData() { return nil }) - // init plugin status - pluginStatus, err := ps.configService.GetStringValueFromDB(context.TODO(), constant.PluginStatus) - if err != nil { - log.Error(err) - } else { - if err := plugin.StatusManager.UnmarshalJSON([]byte(pluginStatus)); err != nil { - log.Error(err) - } - } - - // init plugin config - pluginConfigs, err := ps.pluginConfigRepo.GetPluginConfigAll(context.Background()) - if err != nil { + ps.refreshPluginStatus(context.Background()) + if err := ps.refreshPluginConfigs(context.Background(), true, false); err != nil { log.Error(err) - } else { - for _, pluginConfig := range pluginConfigs { - err := plugin.CallConfig(func(fn plugin.Config) error { - if fn.Info().SlugName == pluginConfig.PluginSlugName { - return fn.ConfigReceiver([]byte(pluginConfig.Value)) - } - return nil - }) - if err != nil { - log.Errorf("parse plugin config failed: %s %v", pluginConfig.PluginSlugName, err) - } - } - - _ = plugin.CallCache(func(cache plugin.Cache) error { - ps.data.Cache = cache - return nil - }) } + _ = plugin.CallCache(func(cache plugin.Cache) error { + ps.data.Cache = cache + return nil + }) // register syncer for vector search plugins on startup _ = plugin.CallVectorSearch(func(vs plugin.VectorSearch) error { @@ -229,3 +270,73 @@ func (ps *PluginCommonService) initPluginData() { } }() } + +func (ps *PluginCommonService) startPluginConfigRefresh() { + ticker := time.NewTicker(pluginConfigRefreshInterval) + go func() { + defer ticker.Stop() + for range ticker.C { + ps.refreshPluginStatus(context.Background()) + if err := ps.refreshPluginConfigs(context.Background(), false, true); err != nil { + log.Error(err) + } + } + }() +} + +func (ps *PluginCommonService) refreshPluginStatus(ctx context.Context) { + ps.pluginStatusMu.Lock() + defer ps.pluginStatusMu.Unlock() + + pluginStatus, err := ps.configService.GetStringValueFromDB(ctx, constant.PluginStatus) + if err != nil { + log.Error(err) + return + } + if err := plugin.StatusManager.UnmarshalJSON([]byte(pluginStatus)); err != nil { + log.Error(err) + } +} + +func (ps *PluginCommonService) refreshPluginConfigs(ctx context.Context, force, registerDependents bool) error { + ps.pluginConfigMu.Lock() + pluginConfigs, err := ps.pluginConfigRepo.GetPluginConfigAll(ctx) + if err != nil { + ps.pluginConfigMu.Unlock() + return err + } + if ps.pluginConfigValues == nil { + ps.pluginConfigValues = make(map[string]string) + } + + changedPlugins := make([]string, 0, len(pluginConfigs)) + for _, pluginConfig := range pluginConfigs { + if !force && ps.pluginConfigValues[pluginConfig.PluginSlugName] == pluginConfig.Value { + continue + } + + applied := false + err := plugin.CallConfig(func(fn plugin.Config) error { + if fn.Info().SlugName != pluginConfig.PluginSlugName { + return nil + } + applied = true + return fn.ConfigReceiver([]byte(pluginConfig.Value)) + }) + if err != nil { + log.Errorf("parse plugin config failed: %s %v", pluginConfig.PluginSlugName, err) + continue + } + if applied { + ps.pluginConfigValues[pluginConfig.PluginSlugName] = pluginConfig.Value + changedPlugins = append(changedPlugins, pluginConfig.PluginSlugName) + } + } + ps.pluginConfigMu.Unlock() + if registerDependents { + for _, pluginSlugName := range changedPlugins { + ps.registerPluginConfigDependents(ctx, pluginSlugName) + } + } + return nil +} diff --git a/internal/service/plugin_common/plugin_common_service_test.go b/internal/service/plugin_common/plugin_common_service_test.go new file mode 100644 index 000000000..83280c82f --- /dev/null +++ b/internal/service/plugin_common/plugin_common_service_test.go @@ -0,0 +1,228 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package plugin_common + +import ( + "context" + "encoding/json" + "sync" + "testing" + + "github.com/apache/answer/internal/base/constant" + basedata "github.com/apache/answer/internal/base/data" + "github.com/apache/answer/internal/entity" + configRepo "github.com/apache/answer/internal/repo/config" + configService "github.com/apache/answer/internal/service/config" + "github.com/apache/answer/plugin" + "github.com/segmentfault/pacman/contrib/cache/memory" +) + +func TestUpdatePluginStatusMergesConcurrentReplicaChanges(t *testing.T) { + baseService, db := newPluginCommonTestService(t, &pluginCommonTestConfigRepo{}) + defer func() { + if err := db.DB.Close(); err != nil { + t.Errorf("close test database: %v", err) + } + }() + if _, err := baseService.configService.GetStringValue(context.Background(), constant.PluginStatus); err != nil { + t.Fatal(err) + } + + start := make(chan struct{}) + results := make(chan error, 2) + for _, slugName := range []string{"replica-a-plugin", "replica-b-plugin"} { + replicaService := &PluginCommonService{configService: baseService.configService} + go func(slugName string, service *PluginCommonService) { + <-start + results <- service.UpdatePluginStatus(context.Background(), slugName, true) + }(slugName, replicaService) + } + close(start) + for range 2 { + if err := <-results; err != nil { + t.Fatal(err) + } + } + + stored, err := baseService.configService.GetStringValueFromDB(context.Background(), constant.PluginStatus) + if err != nil { + t.Fatal(err) + } + statuses := make(map[string]bool) + if err := json.Unmarshal([]byte(stored), &statuses); err != nil { + t.Fatal(err) + } + if !statuses["replica-a-plugin"] || !statuses["replica-b-plugin"] { + t.Fatalf("status map lost a concurrent update: %s", stored) + } + cached, err := baseService.configService.GetStringValue(context.Background(), constant.PluginStatus) + if err != nil { + t.Fatal(err) + } + if cached != stored { + t.Fatalf("cached status = %s, want latest database value %s", cached, stored) + } +} + +func TestRefreshPluginStatusAndConfigFromDatabase(t *testing.T) { + registerPluginCommonConfigTestPlugin(t) + oldStatus, err := plugin.StatusManager.MarshalJSON() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = plugin.StatusManager.UnmarshalJSON(oldStatus) }) + + repo := &pluginCommonTestConfigRepo{configs: map[string]string{ + pluginCommonConfigTestSlug: `{"value":"old"}`, + }} + service, db := newPluginCommonTestService(t, repo) + defer func() { + if err := db.DB.Close(); err != nil { + t.Errorf("close test database: %v", err) + } + }() + + ctx := context.Background() + if err := plugin.StatusManager.UnmarshalJSON([]byte(`{"plugin-common-config-refresh-test":false}`)); err != nil { + t.Fatal(err) + } + if err := service.refreshPluginConfigs(ctx, true, false); err != nil { + t.Fatal(err) + } + if got := pluginCommonConfigTestInstance.configValue(); got != `{"value":"old"}` { + t.Fatalf("initial plugin config = %s, want old config", got) + } + + if err := service.configService.UpdateConfig(ctx, constant.PluginStatus, + `{"plugin-common-config-refresh-test":true}`); err != nil { + t.Fatal(err) + } + repo.set(pluginCommonConfigTestSlug, `{"value":"new"}`) + + service.refreshPluginStatus(ctx) + if err := service.refreshPluginConfigs(ctx, false, false); err != nil { + t.Fatal(err) + } + if !plugin.StatusManager.IsEnabled(pluginCommonConfigTestSlug) { + t.Fatal("plugin status was not refreshed from the database") + } + if got := pluginCommonConfigTestInstance.configValue(); got != `{"value":"new"}` { + t.Fatalf("refreshed plugin config = %s, want new config", got) + } +} + +func newPluginCommonTestService( + t *testing.T, + pluginConfigRepo PluginConfigRepo, +) (*PluginCommonService, *basedata.Data) { + t.Helper() + + db, err := basedata.NewDB(false, &basedata.Database{Driver: "sqlite", Connection: ":memory:"}) + if err != nil { + t.Fatal(err) + } + db.SetMaxOpenConns(1) + data := &basedata.Data{DB: db, Cache: memory.NewCache()} + if err := db.Sync(new(entity.Config)); err != nil { + t.Fatal(err) + } + if _, err := db.Insert(&entity.Config{Key: constant.PluginStatus, Value: `{}`}); err != nil { + t.Fatal(err) + } + + repo := configRepo.NewConfigRepo(data) + service := &PluginCommonService{ + configService: configService.NewConfigService(repo), + pluginConfigRepo: pluginConfigRepo, + pluginConfigValues: make(map[string]string), + } + return service, data +} + +const pluginCommonConfigTestSlug = "plugin-common-config-refresh-test" + +var ( + pluginCommonConfigTestOnce sync.Once + pluginCommonConfigTestInstance = &pluginCommonTestConfigPlugin{} +) + +func registerPluginCommonConfigTestPlugin(t *testing.T) { + t.Helper() + pluginCommonConfigTestOnce.Do(func() { + plugin.Register(pluginCommonConfigTestInstance) + }) +} + +type pluginCommonTestConfigPlugin struct { + mu sync.Mutex + value string +} + +func (p *pluginCommonTestConfigPlugin) Info() plugin.Info { + return plugin.Info{SlugName: pluginCommonConfigTestSlug} +} + +func (p *pluginCommonTestConfigPlugin) ConfigFields() []plugin.ConfigField { return nil } + +func (p *pluginCommonTestConfigPlugin) ConfigReceiver(config []byte) error { + p.mu.Lock() + defer p.mu.Unlock() + p.value = string(config) + return nil +} + +func (p *pluginCommonTestConfigPlugin) configValue() string { + p.mu.Lock() + defer p.mu.Unlock() + return p.value +} + +type pluginCommonTestConfigRepo struct { + mu sync.RWMutex + configs map[string]string +} + +func (r *pluginCommonTestConfigRepo) SavePluginConfig( + _ context.Context, slugName, value string, +) error { + r.set(slugName, value) + return nil +} + +func (r *pluginCommonTestConfigRepo) GetPluginConfigAll( + context.Context, +) ([]*entity.PluginConfig, error) { + r.mu.RLock() + defer r.mu.RUnlock() + configs := make([]*entity.PluginConfig, 0, len(r.configs)) + for slugName, value := range r.configs { + configs = append(configs, &entity.PluginConfig{PluginSlugName: slugName, Value: value}) + } + return configs, nil +} + +func (r *pluginCommonTestConfigRepo) set(slugName, value string) { + r.mu.Lock() + defer r.mu.Unlock() + if r.configs == nil { + r.configs = make(map[string]string) + } + r.configs[slugName] = value +} diff --git a/plugin/captcha.go b/plugin/captcha.go index 96047e67f..f0a49689c 100644 --- a/plugin/captcha.go +++ b/plugin/captcha.go @@ -64,19 +64,23 @@ func CaptchaEnabled() (enabled bool) { return } -func coordinatedCaptchaPlugins(slugName string) (enabledSlugNames []string) { +func coordinatedCaptchaPlugins(slugName string) (coordinatedSlugNames []string) { isCaptcha := false - _ = callCaptcha(func(captcha Captcha) error { + _ = CallBase(func(base Base) error { + captcha, ok := base.(Captcha) + if !ok { + return nil + } name := captcha.Info().SlugName if slugName == name { isCaptcha = true } else { - enabledSlugNames = append(enabledSlugNames, name) + coordinatedSlugNames = append(coordinatedSlugNames, name) } return nil }) if isCaptcha { - return enabledSlugNames + return coordinatedSlugNames } return nil } diff --git a/plugin/cdn.go b/plugin/cdn.go index ac1b4c135..bc93febf5 100644 --- a/plugin/cdn.go +++ b/plugin/cdn.go @@ -47,19 +47,23 @@ var ( registerCDN = MakePlugin[CDN](false) ) -func coordinatedCDNPlugins(slugName string) (enabledSlugNames []string) { +func coordinatedCDNPlugins(slugName string) (coordinatedSlugNames []string) { isCDN := false - _ = CallCDN(func(cdn CDN) error { + _ = CallBase(func(base Base) error { + cdn, ok := base.(CDN) + if !ok { + return nil + } name := cdn.Info().SlugName if slugName == name { isCDN = true } else { - enabledSlugNames = append(enabledSlugNames, name) + coordinatedSlugNames = append(coordinatedSlugNames, name) } return nil }) if isCDN { - return enabledSlugNames + return coordinatedSlugNames } return nil } diff --git a/plugin/plugin.go b/plugin/plugin.go index 3a657fc40..c397de546 100644 --- a/plugin/plugin.go +++ b/plugin/plugin.go @@ -179,29 +179,37 @@ func MakePlugin[T Base](super bool) (CallFn[T], RegisterFn[T]) { } type statusManager struct { - lock sync.Mutex + lock sync.RWMutex status map[string]bool } func (m *statusManager) Enable(name string, enabled bool) { + captchaPlugins := coordinatedCaptchaPlugins(name) + cdnPlugins := coordinatedCDNPlugins(name) + m.lock.Lock() defer m.lock.Unlock() + m.status[name] = enabled if !enabled { - m.status[name] = enabled return } - m.status[name] = enabled - for _, slugName := range coordinatedCaptchaPlugins(name) { - m.status[slugName] = false + for _, slugName := range captchaPlugins { + if m.status[slugName] { + m.status[slugName] = false + } } - for _, slugName := range coordinatedCDNPlugins(name) { - m.status[slugName] = false + for _, slugName := range cdnPlugins { + if m.status[slugName] { + m.status[slugName] = false + } } } func (m *statusManager) IsEnabled(name string) bool { + m.lock.RLock() + defer m.lock.RUnlock() if status, ok := m.status[name]; ok { return status } @@ -210,12 +218,56 @@ func (m *statusManager) IsEnabled(name string) bool { // MarshalJSON implements the json.Marshaler interface. func (m *statusManager) MarshalJSON() ([]byte, error) { + m.lock.RLock() + defer m.lock.RUnlock() return json.Marshal(m.status) } // UnmarshalJSON implements the json.Unmarshaler interface. func (m *statusManager) UnmarshalJSON(data []byte) error { - return json.Unmarshal(data, &m.status) + status := make(map[string]bool) + if err := json.Unmarshal(data, &status); err != nil { + return err + } + if status == nil { + status = make(map[string]bool) + } + + m.lock.Lock() + m.status = status + m.lock.Unlock() + return nil +} + +// UpdateJSON applies a plugin status change to a persisted status map without +// changing the in-memory status. The caller can persist the returned JSON +// before making it visible to this process. +func (m *statusManager) UpdateJSON(data []byte, name string, enabled bool) ([]byte, error) { + status := make(map[string]bool) + if len(data) > 0 { + if err := json.Unmarshal(data, &status); err != nil { + return nil, err + } + } + if status == nil { + status = make(map[string]bool) + } + + status[name] = enabled + if enabled { + for _, slugName := range coordinatedCaptchaPlugins(name) { + if status[slugName] { + status[slugName] = false + } + } + for _, slugName := range coordinatedCDNPlugins(name) { + if status[slugName] { + status[slugName] = false + } + } + } + + return json.Marshal(status) } // Translate translates the key to the current language of the context