main.go

package main

import (
	"flag"
	"fmt"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"syscall"

	"my-broker/broker"
	"my-broker/config"
	"my-broker/logger"
	"my-broker/server"
)

func main() {
	var (
		configPath    string
		nodeID        string
		grpcPort      string
		raftPort      string
		dataPort      string
		metricsPort   string
		dataDir       string
		join          string
		bindAddr      string
		advertiseAddr string
		logLevel      string
		isController  bool
	)

	flag.StringVar(&configPath, "config", "config.yaml", "Path to YAML config file (optional)")
	flag.StringVar(&nodeID, "id", "", "Node ID (required)")
	flag.StringVar(&grpcPort, "grpc-port", "8080", "gRPC port for client connections")
	flag.StringVar(&raftPort, "raft-port", "9080", "Raft port for controller communication")
	flag.StringVar(&dataPort, "data-port", "9180", "Port for inter-broker replication")
	flag.StringVar(&metricsPort, "metrics-port", "9090", "HTTP port for Prometheus metrics")
	flag.StringVar(&dataDir, "data-dir", "", "Data directory (default: ./data/<node-id>)")
	flag.StringVar(&join, "join", "", "Address of node to join (e.g., localhost:8080)")
	flag.StringVar(&bindAddr, "bind-addr", "", "Address to bind listeners to (default: the advertise address)")
	flag.StringVar(&advertiseAddr, "advertise-addr", "", "Address to advertise to other nodes (default: localhost)")
	flag.StringVar(&logLevel, "log-level", "info", "Log level: debug, info, warn, error")
	flag.BoolVar(&isController, "controller", true, "Run as a controller node (default: true)")
	flag.Parse()

	// Precedence: defaults < config file < environment < explicit CLI flags.
	cfg, err := config.LoadConfigWithEnvOverrides(configPath)
	if err != nil {
		fmt.Printf("Error loading config: %v\n", err)
		os.Exit(1)
	}

	explicit := make(map[string]bool)
	flag.Visit(func(f *flag.Flag) { explicit[f.Name] = true })

	if explicit["id"] || cfg.Node.ID == "" {
		cfg.Node.ID = nodeID
	}
	if explicit["grpc-port"] || cfg.Network.GRPCPort == 0 {
		cfg.Network.GRPCPort, _ = strconv.Atoi(grpcPort)
	}
	if explicit["raft-port"] || cfg.Network.RaftPort == 0 {
		cfg.Network.RaftPort, _ = strconv.Atoi(raftPort)
	}
	if explicit["data-port"] || cfg.Network.DataPort == 0 {
		cfg.Network.DataPort, _ = strconv.Atoi(dataPort)
	}
	if explicit["metrics-port"] || cfg.Network.MetricsPort == 0 {
		cfg.Network.MetricsPort, _ = strconv.Atoi(metricsPort)
	}
	if explicit["data-dir"] {
		cfg.Node.DataDir = dataDir
	}
	if explicit["join"] {
		cfg.Cluster.JoinAddr = join
	}
	if explicit["bind-addr"] {
		cfg.Network.BindAddr = bindAddr
	}
	if explicit["advertise-addr"] {
		cfg.Network.AdvertiseAddr = advertiseAddr
	}
	if explicit["log-level"] {
		cfg.Logging.Level = logLevel
	}
	if explicit["controller"] {
		cfg.Cluster.Controller = isController
	}

	if cfg.Node.ID == "" {
		fmt.Println("Error: node ID is required (-id flag or node.id in config)")
		flag.Usage()
		os.Exit(1)
	}
	if err := cfg.Validate(); err != nil {
		fmt.Printf("Invalid configuration: %v\n", err)
		os.Exit(1)
	}

	if cfg.Node.DataDir == "" || cfg.Node.DataDir == "./data" {
		cfg.Node.DataDir = filepath.Join("data", cfg.Node.ID)
	}

	logger.Init(cfg.Logging.Level, cfg.Node.ID)

	advertiseHost := cfg.Network.AdvertiseAddr
	if advertiseHost == "" {
		advertiseHost = "localhost"
	}
	// Listeners bind wide by default; the advertise host is what peers dial.
	bindHost := cfg.Network.BindAddr
	if bindHost == "" {
		bindHost = "0.0.0.0"
	}

	logger.Info("Starting distributed broker node",
		"grpc_addr", fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.GRPCPort),
		"raft_addr", fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.RaftPort),
		"data_addr", fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.DataPort),
		"metrics_addr", fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.MetricsPort),
		"data_dir", cfg.Node.DataDir,
		"is_controller", cfg.Cluster.Controller,
	)

	bootstrap := cfg.Cluster.JoinAddr == ""

	brokerCfg := broker.DistributedBrokerConfig{
		NodeID:            cfg.Node.ID,
		GRPCAddr:          fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.GRPCPort),
		RaftAddr:          fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.RaftPort),
		RaftBindAddr:      fmt.Sprintf("%s:%d", bindHost, cfg.Network.RaftPort),
		DataAddr:          fmt.Sprintf("%s:%d", advertiseHost, cfg.Network.DataPort),
		DataBindAddr:      fmt.Sprintf("%s:%d", bindHost, cfg.Network.DataPort),
		DataDir:           cfg.Node.DataDir,
		Bootstrap:         bootstrap,
		IsController:      cfg.Cluster.Controller,
		MaxSegmentBytes:   cfg.Storage.MaxSegmentBytes,
		DisableSyncWrites: !cfg.Storage.SyncWrites,
		RetentionMaxAge:   cfg.Retention.DefaultMaxAge,
		RetentionMaxBytes: cfg.Retention.DefaultMaxBytes,
		RetentionInterval: cfg.Retention.CleanupInterval,
	}

	b, err := broker.NewDistributedBroker(brokerCfg)
	if err != nil {
		logger.Error("Failed to create broker", "error", err)
		os.Exit(1)
	}
	defer b.Close()

	serverCfg := server.Config{
		NodeID:        cfg.Node.ID,
		GRPCPort:      strconv.Itoa(cfg.Network.GRPCPort),
		RaftPort:      strconv.Itoa(cfg.Network.RaftPort),
		MetricsPort:   strconv.Itoa(cfg.Network.MetricsPort),
		JoinAddr:      cfg.Cluster.JoinAddr,
		BindAddr:      cfg.Network.BindAddr,
		AdvertiseAddr: cfg.Network.AdvertiseAddr,
	}

	srv := server.NewDistributedServer(b, serverCfg)

	if err := srv.Start(); err != nil {
		logger.Error("Failed to start server", "error", err)
		os.Exit(1)
	}

	logger.Info("Distributed broker ready",
		"bootstrap", bootstrap,
		"controller", cfg.Cluster.Controller,
	)

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	<-sigCh

	logger.Info("Shutting down...")
	srv.Stop()
}