config.go

package aastro

import (
	"errors"
	"fmt"
	"io"
	"os"
	"reflect"
	"regexp"
	"strings"
	"time"

	"github.com/creasty/defaults"
	"github.com/go-playground/validator/v10"
	"gopkg.in/yaml.v3"
)

type Config struct {
	Schema  string        `yaml:"schema" validate:"required,oneof=v1"`
	Debug   bool          `yaml:"debug"`
	Gateway GatewayConfig `yaml:"gateway" validate:"required"`
}

func (c *Config) Marshal() ([]byte, error) {
	return yaml.Marshal(c)
}

func (c *Config) WriteTo(w io.Writer) (int64, error) {
	data, err := c.Marshal()
	if err != nil {
		return 0, fmt.Errorf("marshal config: %w", err)
	}

	n, err := w.Write(data)

	return int64(n), err
}

type GatewayConfig struct {
	Service       ServiceConfig       `yaml:"service"`
	Server        ServerConfig        `yaml:"server"  validate:"required"`
	Admin         AdminConfig         `yaml:"admin" validate:"required"`
	Observability ObservabilityConfig `yaml:"observability"`
	Routing       RoutingConfig       `yaml:"routing" validate:"required"`
}

type ServiceConfig struct {
	Name string `yaml:"name" default:"aastro"`
}

type ServerConfig struct {
	Port          int             `yaml:"port"    validate:"required,min=1,max=65535"`
	Timeout       time.Duration   `yaml:"timeout" default:"5s"`
	HeaderTimeout time.Duration   `yaml:"header_timeout" default:"5s"`
	TLS           ServerTLSConfig `yaml:"tls"`
}

type AdminConfig struct {
	Port          int           `yaml:"port"    validate:"required,min=1,max=65535"`
	BindAddr      string        `yaml:"bind_addr" default:"127.0.0.1"`
	Timeout       time.Duration `yaml:"timeout"   default:"5m"`
	HeaderTimeout time.Duration `yaml:"header_timeout" default:"5s"`
	EnablePprof   bool          `yaml:"enable_pprof" default:"false"`
}

type ObservabilityConfig struct {
	Tracing TracingConfig `yaml:"tracing"`
	Metrics MetricsConfig `yaml:"metrics"`
}

type MetricsConfig struct {
	Enabled  bool       `yaml:"enabled"`
	Exporter string     `yaml:"exporter" validate:"required_if=Enabled true,omitempty,oneof=otlp prometheus"`
	OTLP     OTLPConfig `yaml:"otlp"`
}

type TracingConfig struct {
	Enabled       bool       `yaml:"enabled"`
	Exporter      string     `yaml:"exporter" validate:"required_if=Enabled true,omitempty,oneof=otlp"`
	SamplingRatio float64    `yaml:"sampling_ratio" default:"1.0" validate:"min=0,max=1"`
	OTLP          OTLPConfig `yaml:"otlp"`
}

type OTLPConfig struct {
	Endpoint string        `yaml:"endpoint"`
	Insecure bool          `yaml:"insecure"`
	Interval time.Duration `yaml:"interval"`
}

type ServerTLSConfig struct {
	Enabled      bool   `yaml:"enabled"`
	CertFile     string `yaml:"cert_file"      validate:"required_if=Enabled true"`
	KeyFile      string `yaml:"key_file"       validate:"required_if=Enabled true"`
	MinVersion   string `yaml:"min_version"    default:"1.2" validate:"omitempty,oneof=1.2 1.3"`
	ClientAuth   string `yaml:"client_auth"    default:"none" validate:"omitempty,oneof=require optional none"`
	ClientCAFile string `yaml:"client_ca_file" validate:"required_unless=ClientAuth none"`
}

type RoutingConfig struct {
	RateLimiter    RateLimiterConfig `yaml:"rate_limiter" validate:"omitempty"`
	TrustedProxies []string          `yaml:"trusted_proxies"`
	Flows          []FlowConfig      `yaml:"flows" validate:"min=1,dive,required"`
}

type RateLimiterConfig struct {
	Enabled bool                   `yaml:"enabled"`
	Config  map[string]interface{} `yaml:"config" validate:"required"`
}

type FlowConfig struct {
	Path      string `yaml:"path"   validate:"required,startswith=/"`
	Method    string `yaml:"method" validate:"required,oneof=GET POST PUT PATCH DELETE HEAD OPTIONS QUERY"`
	Streaming bool   `yaml:"streaming"`

	// Aggregation is required only for flows with more than one upstream -
	// a single-upstream flow is proxied directly and never aggregates
	// (enforced in validateFlows, since that depends on len(Upstreams)).
	Aggregation *AggregationConfig `yaml:"aggregation"`
	Upstreams   []UpstreamConfig   `yaml:"upstreams"    validate:"required,min=1,dive,required"`
	Plugins     []PluginConfig     `yaml:"plugins"      validate:"omitempty,dive"`
	Middlewares []MiddlewareConfig `yaml:"middlewares"  validate:"omitempty,dive"`
}

type AggregationConfig struct {
	BestEffort bool              `yaml:"best_effort"`
	Strategy   string            `yaml:"strategy"    validate:"required,oneof=array merge namespace"`
	OnConflict *OnConflictConfig `yaml:"on_conflict" validate:"required_if=Strategy merge"`
}

type OnConflictConfig struct {
	Policy   string `yaml:"policy"          validate:"oneof=overwrite error first prefer"`
	Upstream string `yaml:"prefer_upstream" validate:"required_if=Policy prefer"`
}

type UpstreamConfig struct {
	Name    string        `yaml:"name" validate:"required"`
	Hosts   AddrList      `yaml:"hosts" validate:"min=1,dive"`
	Path    string        `yaml:"path"`
	Method  string        `yaml:"method"`
	Timeout time.Duration `yaml:"timeout" default:"3s"`

	ForwardHeaders []string `yaml:"forward_headers"`
	ForwardQueries []string `yaml:"forward_queries"`
	ForwardParams  []string `yaml:"forward_params"`

	Policy    PolicyConfig    `yaml:"policy"`
	Transport TransportConfig `yaml:"transport"`
	TLS       TLSConfig       `yaml:"tls"`
}

type TLSConfig struct {
	Enabled            bool   `yaml:"enabled"`
	CertFile           string `yaml:"cert_file" validate:"required_with=KeyFile"`
	KeyFile            string `yaml:"key_file" validate:"required_with=CertFile"`
	CAFile             string `yaml:"ca_file"`
	ServerName         string `yaml:"server_name"`
	InsecureSkipVerify bool   `yaml:"insecure_skip_verify"`
	MinVersion         string `yaml:"min_version" default:"1.2" validate:"omitempty,oneof=1.2 1.3"`
}

type TransportConfig struct {
	MaxIdleConns        int           `yaml:"max_idle_conns"         default:"100"`
	MaxIdleConnsPerHost int           `yaml:"max_idle_conns_per_host" default:"50"`
	IdleConnTimeout     time.Duration `yaml:"idle_conn_timeout"      default:"90s"`
}

type PluginConfig struct {
	Name   string                 `yaml:"name"   validate:"required"`
	Source string                 `yaml:"source" validate:"required,oneof=builtin file"`
	Path   string                 `yaml:"path"   validate:"required_if=Source file"`
	Config map[string]interface{} `yaml:"config"`
}

type MiddlewareConfig struct {
	Name   string                 `yaml:"name"   validate:"required"`
	Source string                 `yaml:"source" validate:"required,oneof=builtin file"`
	Path   string                 `yaml:"path"   validate:"required_if=Source file,omitempty"`
	Config map[string]interface{} `yaml:"config"`
}

type PolicyConfig struct {
	HeaderBlacklist     []string `yaml:"header_blacklist"`
	RequireBody         bool     `yaml:"require_body"`
	MaxResponseBodySize int64    `yaml:"max_response_body_size"`
	FollowRedirects     bool     `yaml:"follow_redirects"`

	RetryConfig          RetryConfig          `yaml:"retry"`
	CircuitBreakerConfig CircuitBreakerConfig `yaml:"circuit_breaker"`
	LoadBalancingConfig  LoadBalancingConfig  `yaml:"load_balancing"`
}

type RetryConfig struct {
	MaxRetries      int           `yaml:"max_retries"`
	RetryOnStatuses []int         `yaml:"retry_on_statuses"`
	BackoffDelay    time.Duration `yaml:"backoff_delay"`
}

type CircuitBreakerConfig struct {
	Enabled      bool          `yaml:"enabled"`
	MaxFailures  int           `yaml:"max_failures"`
	ResetTimeout time.Duration `yaml:"reset_timeout"`
}

type LoadBalancingConfig struct {
	Mode string `yaml:"mode"`
}

type AddrList []string

func (a *AddrList) UnmarshalYAML(value *yaml.Node) error {
	switch value.Kind {
	case yaml.ScalarNode:
		var addr string
		if err := value.Decode(&addr); err != nil {
			return err
		}

		*a = []string{addr}

		return nil
	case yaml.SequenceNode:
		var addrs []string
		if err := value.Decode(&addrs); err != nil {
			return err
		}

		*a = addrs

		return nil
	case yaml.DocumentNode, yaml.MappingNode, yaml.AliasNode:
		return fmt.Errorf("expects a string or sequence, got %v", value.Kind)
	default:
		return fmt.Errorf("unexpected yaml node kind for AddrList: %v", value.Kind)
	}
}

// LoadConfig reads, parses, applies defaults, validates, and returns the config.
func LoadConfig(path string) (Config, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return Config{}, fmt.Errorf("cannot read configuration file: %w", err)
	}

	var cfg Config

	if err = yaml.Unmarshal(data, &cfg); err != nil {
		return Config{}, fmt.Errorf("cannot parse configuration file: %w", err)
	}

	if err = ValidateConfig(&cfg); err != nil {
		return Config{}, err
	}

	return cfg, nil
}

func DefaultUpstreamConfig() UpstreamConfig {
	var u UpstreamConfig
	_ = defaults.Set(&u)

	return u
}

func ValidateConfig(cfg *Config) error {
	if err := defaults.Set(cfg); err != nil {
		return fmt.Errorf("cannot apply configuration defaults: %w", err)
	}

	v := newValidator()

	if err := v.Struct(cfg); err != nil {
		return fmt.Errorf("invalid configuration:\n%w", formatValidationError(err))
	}

	if cfg.Gateway.Server.Port == cfg.Gateway.Admin.Port {
		return errors.New("server.port and admin.port must differ")
	}

	if err := validateFlows(*cfg); err != nil {
		return fmt.Errorf("invalid flow configuration: %w", err)
	}

	return nil
}

func newValidator() *validator.Validate {
	v := validator.New()

	v.RegisterTagNameFunc(func(fld reflect.StructField) string {
		name := fld.Tag.Get("yaml")
		if name == "" || name == "-" {
			return strings.ToLower(fld.Name)
		}

		return strings.ToLower(strings.Split(name, ",")[0])
	})

	return v
}

var pathParamPattern = regexp.MustCompile(`\{([^}]+)\}`)

func validateFlows(cfg Config) error {
	for _, f := range cfg.Gateway.Routing.Flows {
		flowParams := extractPathParams(f.Path)

		for _, u := range f.Upstreams {
			if err := validateUpstreamParams(u, flowParams, f.Path); err != nil {
				return err
			}
		}

		if f.Streaming && len(f.Upstreams) > 1 {
			return fmt.Errorf("flow %q: streaming flow must have only one upstream", f.Path)
		}

		if !f.Streaming && len(f.Upstreams) > 1 && f.Aggregation == nil {
			return fmt.Errorf("flow %q: aggregation is required for flows with more than one upstream", f.Path)
		}
	}

	return nil
}

func extractPathParams(path string) map[string]struct{} {
	params := make(map[string]struct{})

	for _, match := range pathParamPattern.FindAllStringSubmatch(path, -1) {
		params[match[1]] = struct{}{}
	}

	return params
}

func validateUpstreamParams(u UpstreamConfig, flowParams map[string]struct{}, flowPath string) error {
	for _, match := range pathParamPattern.FindAllStringSubmatch(u.Path, -1) {
		if _, ok := flowParams[match[1]]; !ok {
			return fmt.Errorf(
				"upstream %q: path param '{%s}' not declared in flow path %q",
				u.Name, match[1], flowPath,
			)
		}
	}

	for _, param := range u.ForwardParams {
		if param == "*" {
			continue
		}

		if _, ok := flowParams[param]; !ok {
			return fmt.Errorf(
				"upstream %q: forward_param %q not declared in flow path %q",
				u.Name, param, flowPath,
			)
		}
	}

	return nil
}

func formatValidationError(err error) error {
	var ves validator.ValidationErrors
	if !errors.As(err, &ves) {
		return err
	}

	messages := make([]string, 0, len(ves))

	for _, fe := range ves {
		path := strings.TrimPrefix(fe.Namespace(), "Config.")
		messages = append(messages, fmt.Sprintf("  %s: %s", path, validationMessage(fe)))
	}

	return errors.New(strings.Join(messages, "\n"))
}

func validationMessage(fe validator.FieldError) string {
	switch fe.Tag() {
	case "required":
		return "field is required"
	case "min":
		return fmt.Sprintf("must have at least %s item(s)", fe.Param())
	case "oneof":
		return fmt.Sprintf("must be one of [%s]", fe.Param())
	case "hosts":
		return "must be a valid URL"
	case "required_if":
		if fe.Field() == "path" {
			return "is required when source is 'file'"
		}

		return fe.Error()
	default:
		return fmt.Sprintf("validation failed on %q", fe.Tag())
	}
}