package client import ( "encoding/json" "fmt" "io" "log" "net/http" "strings" "sync" "time" ) // Client is the config center client type Client struct { config *Config cache *Cache listener *Listener configs map[string]string secret string globalSecret string lastTime int64 mu sync.RWMutex httpClient *http.Client stopCh chan struct{} stopOnce sync.Once started bool } // New creates a new config client func New(config *Config) *Client { if config.Version == "" { config.Version = "latest" } if config.SyncInterval <= 0 { config.SyncInterval = 60 } if config.CacheDir == "" { config.CacheDir = "." } return &Client{ config: config, cache: NewCache(config.CacheDir), listener: NewListener(), configs: make(map[string]string), httpClient: &http.Client{Timeout: 10 * time.Second}, stopCh: make(chan struct{}), } } // OnChange registers a config change handler func (c *Client) OnChange(handler ChangeHandler) { c.listener.OnChange(handler) } // Get returns a config value by key func (c *Client) Get(key string) string { c.mu.RLock() defer c.mu.RUnlock() return c.configs[key] } // GetAll returns all configs func (c *Client) GetAll() map[string]string { c.mu.RLock() defer c.mu.RUnlock() result := make(map[string]string, len(c.configs)) for k, v := range c.configs { result[k] = v } return result } // Start initializes the client and starts sync loop func (c *Client) Start() error { c.mu.Lock() if c.started { c.mu.Unlock() return nil } c.started = true c.mu.Unlock() // fetch all configs if err := c.fetchAll(); err != nil { log.Printf("[ConfigClient] fetch all configs error: %v, trying cache", err) if cached, cacheErr := c.cache.Read(); cacheErr == nil { c.applyConfigs(cached) log.Printf("[ConfigClient] loaded %d configs from cache", len(c.configs)) } else { return fmt.Errorf("fetch configs failed and no cache: %w", err) } } // start sync loop go c.syncLoop() return nil } // Stop stops the sync loop func (c *Client) Stop() { c.stopOnce.Do(func() { close(c.stopCh) }) } func (c *Client) fetchAll() error { url := fmt.Sprintf("%s/api/fetchAllConfigs?appName=%s&env=%s&version=%s&ignoreGlobal=%v", c.config.BaseURL, c.config.AppName, c.config.Env, c.config.Version, c.config.IgnoreGlobal) req, err := http.NewRequest("GET", url, nil) if err != nil { return err } resp, err := c.httpClient.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return fmt.Errorf("server returned HTTP %d", resp.StatusCode) } body, err := io.ReadAll(resp.Body) if err != nil { return err } var result map[string]interface{} if err := json.Unmarshal(body, &result); err != nil { return err } // check error response if code, ok := result["code"]; ok { codeInt, _ := code.(float64) if codeInt != 0 { msg, _ := result["msg"].(string) return fmt.Errorf("server error: %s", msg) } } // extract secrets if s, ok := result["kenaito.config.encrypt-secret"]; ok { c.secret, _ = s.(string) } if s, ok := result["kenaito.config.global-encrypt-secret"]; ok { c.globalSecret, _ = s.(string) } // extract timestamp if ts, ok := result["latest.fetch.timestamp"]; ok { if tsStr, ok := ts.(string); ok { fmt.Sscanf(tsStr, "%d", &c.lastTime) } } // save to cache c.cache.Write(result) // apply configs c.applyConfigs(result) log.Printf("[ConfigClient] fetched %d configs", len(c.configs)) return nil } func (c *Client) fetchChanged() error { url := fmt.Sprintf("%s/api/fetchChangedConfigs", c.config.BaseURL) payload := map[string]interface{}{ "appName": c.config.AppName, "env": c.config.Env, "version": c.config.Version, "lastTime": c.lastTime, "ignoreGlobal": c.config.IgnoreGlobal, } payloadBytes, _ := json.Marshal(payload) req, err := http.NewRequest("POST", url, strings.NewReader(string(payloadBytes))) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := c.httpClient.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return fmt.Errorf("server returned HTTP %d", resp.StatusCode) } body, err := io.ReadAll(resp.Body) if err != nil { return err } var wrapper struct { Code int `json:"code"` Data map[string]string `json:"data"` } if err := json.Unmarshal(body, &wrapper); err != nil { return err } if wrapper.Code != 0 { return fmt.Errorf("server error code: %d", wrapper.Code) } // update timestamp if ts, ok := wrapper.Data["latest.fetch.timestamp"]; ok { fmt.Sscanf(ts, "%d", &c.lastTime) delete(wrapper.Data, "latest.fetch.timestamp") } if len(wrapper.Data) > 0 { // decrypt values changed := make(map[string]string) for k, v := range wrapper.Data { decrypted, err := decryptConfig(c.secret, v) if err != nil && c.globalSecret != "" { decrypted, err = decryptConfig(c.globalSecret, v) } if err != nil { decrypted = v } changed[k] = decrypted } // update local configs c.mu.Lock() for k, v := range changed { c.configs[k] = v } c.mu.Unlock() // notify listeners c.listener.Notify(changed) log.Printf("[ConfigClient] %d configs changed", len(changed)) } return nil } func (c *Client) applyConfigs(data map[string]interface{}) { c.mu.Lock() defer c.mu.Unlock() for k, v := range data { // skip meta keys if strings.HasPrefix(k, "kenaito.") || k == "latest.fetch.timestamp" { continue } strVal, ok := v.(string) if !ok { continue } // decrypt if needed - try app secret first, then global secret decrypted, err := decryptConfig(c.secret, strVal) if err != nil && c.globalSecret != "" { decrypted, err = decryptConfig(c.globalSecret, strVal) } if err != nil { decrypted = strVal } c.configs[k] = decrypted } } func (c *Client) syncLoop() { ticker := time.NewTicker(time.Duration(c.config.SyncInterval) * time.Second) defer ticker.Stop() for { select { case <-ticker.C: if err := c.fetchChanged(); err != nil { log.Printf("[ConfigClient] sync error: %v", err) } case <-c.stopCh: return } } }