logcorrelator/internal/config/config.go

package config

import (
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"gopkg.in/yaml.v3"
)

// Config holds the complete application configuration.
type Config struct {
	Inputs      InputsConfig      `yaml:"inputs"`
	Outputs     OutputsConfig     `yaml:"outputs"`
	Correlation CorrelationConfig `yaml:"correlation"`
}

// ServiceConfig holds service-level configuration.
type ServiceConfig struct {
	Name     string `yaml:"name"`
	Language string `yaml:"language"`
}

// InputsConfig holds input sources configuration.
type InputsConfig struct {
	UnixSockets []UnixSocketConfig `yaml:"unix_sockets"`
}

// UnixSocketConfig holds a Unix socket source configuration.
type UnixSocketConfig struct {
	Name            string `yaml:"name"`
	Path            string `yaml:"path"`
	Format          string `yaml:"format"`
	SourceType      string `yaml:"source_type"`       // "A" for Apache/HTTP, "B" for Network
	SocketPermissions string `yaml:"socket_permissions"` // octal string, e.g., "0660", "0666"
}

// OutputsConfig holds output sinks configuration.
type OutputsConfig struct {
	File       FileOutputConfig       `yaml:"file"`
	ClickHouse ClickHouseOutputConfig `yaml:"clickhouse"`
	Stdout     bool                   `yaml:"stdout"`
}

// FileOutputConfig holds file sink configuration.
type FileOutputConfig struct {
	Path string `yaml:"path"`
}

// ClickHouseOutputConfig holds ClickHouse sink configuration.
type ClickHouseOutputConfig struct {
	Enabled         bool   `yaml:"enabled"`
	DSN             string `yaml:"dsn"`
	Table           string `yaml:"table"`
	BatchSize       int    `yaml:"batch_size"`
	FlushIntervalMs int    `yaml:"flush_interval_ms"`
	MaxBufferSize   int    `yaml:"max_buffer_size"`
	DropOnOverflow  bool   `yaml:"drop_on_overflow"`
	AsyncInsert     bool   `yaml:"async_insert"`
	TimeoutMs       int    `yaml:"timeout_ms"`
}

// StdoutOutputConfig holds stdout sink configuration.
// Deprecated: stdout is now a boolean flag in OutputsConfig.
type StdoutOutputConfig struct {
	Enabled bool `yaml:"enabled"`
}

// CorrelationConfig holds correlation configuration.
type CorrelationConfig struct {
	TimeWindowS  int  `yaml:"time_window_s"`
	EmitOrphans  bool `yaml:"emit_orphans"`
}

// Load loads configuration from a YAML file.
func Load(path string) (*Config, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, fmt.Errorf("failed to read config file: %w", err)
	}

	cfg := defaultConfig()

	if err := yaml.Unmarshal(data, cfg); err != nil {
		return nil, fmt.Errorf("failed to parse config file: %w", err)
	}

	if err := cfg.Validate(); err != nil {
		return nil, fmt.Errorf("invalid config: %w", err)
	}

	return cfg, nil
}

// defaultConfig returns a Config with default values.
func defaultConfig() *Config {
	return &Config{
		Inputs: InputsConfig{
			UnixSockets: make([]UnixSocketConfig, 0),
		},
		Outputs: OutputsConfig{
			File: FileOutputConfig{
				Path: "/var/log/logcorrelator/correlated.log",
			},
			ClickHouse: ClickHouseOutputConfig{
				Enabled:         false,
				BatchSize:       500,
				FlushIntervalMs: 200,
				MaxBufferSize:   5000,
				DropOnOverflow:  true,
				AsyncInsert:     true,
				TimeoutMs:       1000,
			},
			Stdout: false,
		},
		Correlation: CorrelationConfig{
			TimeWindowS: 1,
			EmitOrphans: true,
		},
	}
}

// Validate validates the configuration.
func (c *Config) Validate() error {
	if len(c.Inputs.UnixSockets) < 2 {
		return fmt.Errorf("at least two unix socket inputs are required")
	}

	seenNames := make(map[string]struct{}, len(c.Inputs.UnixSockets))
	seenPaths := make(map[string]struct{}, len(c.Inputs.UnixSockets))

	for i, input := range c.Inputs.UnixSockets {
		if strings.TrimSpace(input.Name) == "" {
			return fmt.Errorf("inputs.unix_sockets[%d].name is required", i)
		}
		if strings.TrimSpace(input.Path) == "" {
			return fmt.Errorf("inputs.unix_sockets[%d].path is required", i)
		}

		if _, exists := seenNames[input.Name]; exists {
			return fmt.Errorf("duplicate unix socket input name: %s", input.Name)
		}
		seenNames[input.Name] = struct{}{}

		if _, exists := seenPaths[input.Path]; exists {
			return fmt.Errorf("duplicate unix socket input path: %s", input.Path)
		}
		seenPaths[input.Path] = struct{}{}
	}

	// At least one output must be enabled
	hasOutput := false
	if c.Outputs.File.Path != "" {
		hasOutput = true
	}
	if c.Outputs.ClickHouse.Enabled {
		hasOutput = true
	}
	if c.Outputs.Stdout {
		hasOutput = true
	}

	if !hasOutput {
		return fmt.Errorf("at least one output must be enabled (file, clickhouse, or stdout)")
	}

	if c.Outputs.ClickHouse.Enabled {
		if strings.TrimSpace(c.Outputs.ClickHouse.DSN) == "" {
			return fmt.Errorf("clickhouse DSN is required when enabled")
		}
		if strings.TrimSpace(c.Outputs.ClickHouse.Table) == "" {
			return fmt.Errorf("clickhouse table is required when enabled")
		}
		if c.Outputs.ClickHouse.BatchSize <= 0 {
			return fmt.Errorf("clickhouse batch_size must be > 0")
		}
		if c.Outputs.ClickHouse.MaxBufferSize <= 0 {
			return fmt.Errorf("clickhouse max_buffer_size must be > 0")
		}
		if c.Outputs.ClickHouse.TimeoutMs <= 0 {
			return fmt.Errorf("clickhouse timeout_ms must be > 0")
		}
	}

	if c.Correlation.TimeWindowS <= 0 {
		return fmt.Errorf("correlation.time_window_s must be > 0")
	}

	return nil
}

// GetTimeWindow returns the time window as a duration.
func (c *CorrelationConfig) GetTimeWindow() time.Duration {
	value := c.TimeWindowS
	if value <= 0 {
		value = 1
	}
	return time.Duration(value) * time.Second
}

// GetSocketPermissions returns the socket permissions as os.FileMode.
// Default is 0660 (owner + group read/write).
func (c *UnixSocketConfig) GetSocketPermissions() os.FileMode {
	if c.SocketPermissions == "" {
		return 0660
	}

	// Parse octal string (e.g., "0660", "660", "0666")
	perms, err := strconv.ParseUint(strings.TrimPrefix(c.SocketPermissions, "0"), 8, 32)
	if err != nil {
		return 0660
	}

	return os.FileMode(perms)
}