geo/server/internal/shared/config/runtime/config.go

package runtime

import (
	"context"
	"errors"
	"reflect"
	"sync"
	"time"
)

var ErrNotFound = errors.New("key not found")

type Observer func(string, Value)

type Config interface {
	Load() error
	Scan(any) error
	Value(string) Value
	Watch(string, Observer) error
	Close() error
}

type config struct {
	opts      options
	reader    Reader
	cached    sync.Map
	observers sync.Map
	watchers  []Watcher
	closeOnce sync.Once
}

func New(opts ...Option) Config {
	o := options{
		decoder:  defaultDecoder,
		resolver: defaultResolver,
		merge:    mergeMap,
	}
	for _, opt := range opts {
		opt(&o)
	}
	return &config{
		opts:   o,
		reader: newReader(o),
	}
}

func (c *config) Load() error {
	for _, src := range c.opts.sources {
		kvs, err := src.Load()
		if err != nil {
			return err
		}
		if err = c.reader.Merge(kvs...); err != nil {
			return err
		}
		w, err := src.Watch()
		if err != nil {
			return err
		}
		c.watchers = append(c.watchers, w)
		go c.watch(w)
	}
	return c.reader.Resolve()
}

func (c *config) Value(key string) Value {
	if v, ok := c.cached.Load(key); ok {
		return v.(Value)
	}
	if v, ok := c.reader.Value(key); ok {
		c.cached.Store(key, v)
		return v
	}
	return &errValue{err: ErrNotFound}
}

func (c *config) Scan(v any) error {
	data, err := c.reader.Source()
	if err != nil {
		return err
	}
	return unmarshalJSON(data, v)
}

func (c *config) Watch(key string, o Observer) error {
	if v := c.Value(key); v.Load() == nil {
		return ErrNotFound
	}
	c.observers.Store(key, o)
	return nil
}

func (c *config) Close() error {
	var closeErr error
	c.closeOnce.Do(func() {
		for _, w := range c.watchers {
			if err := w.Stop(); err != nil && closeErr == nil {
				closeErr = err
			}
		}
	})
	return closeErr
}

func (c *config) watch(w Watcher) {
	for {
		kvs, err := w.Next()
		if err != nil {
			if errors.Is(err, context.Canceled) {
				return
			}
			time.Sleep(time.Second)
			continue
		}
		if err := c.reader.Merge(kvs...); err != nil {
			c.notifyRootObserver()
			continue
		}
		if err := c.reader.Resolve(); err != nil {
			c.notifyRootObserver()
			continue
		}
		c.notifyObservers()
	}
}

func (c *config) notifyRootObserver() {
	if observer, ok := c.observers.Load(""); ok {
		observer.(Observer)("", c.Value(""))
	}
}

func (c *config) notifyObservers() {
	if observer, ok := c.observers.Load(""); ok {
		next, ok := c.reader.Value("")
		if !ok {
			return
		}
		current := c.Value("")
		current.Store(next.Load())
		observer.(Observer)("", current)
	}

	c.cached.Range(func(key, value any) bool {
		path := key.(string)
		if path == "" {
			return true
		}
		previous := value.(Value)
		next, ok := c.reader.Value(path)
		if !ok {
			return true
		}
		if reflect.TypeOf(next.Load()) == reflect.TypeOf(previous.Load()) && !reflect.DeepEqual(next.Load(), previous.Load()) {
			previous.Store(next.Load())
			if observer, ok := c.observers.Load(path); ok {
				observer.(Observer)(path, previous)
			}
		}
		return true
	})
}

func mergeMap(dst, src any) error {
	dstMap, ok := dst.(*map[string]any)
	if !ok {
		return errors.New("merge destination must be *map[string]any")
	}
	srcMap, ok := src.(map[string]any)
	if !ok {
		return errors.New("merge source must be map[string]any")
	}
	deepMerge(*dstMap, srcMap)
	return nil
}

func deepMerge(dst, src map[string]any) {
	for key, srcValue := range src {
		if srcChild, ok := srcValue.(map[string]any); ok {
			if dstChild, ok := dst[key].(map[string]any); ok {
				deepMerge(dstChild, srcChild)
				continue
			}
		}
		dst[key] = srcValue
	}
}