package main

import (
	"context"
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"flag"
	"log"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"
)

type Config struct {
	// Logs: absolute/relative file paths to tail.
	Logs []string `json:"logs"`

	// Endpoint: HTTP endpoint to POST batched metrics to.
	Endpoint string `json:"endpoint"`

	// DashboardAddr: HTTP listen address for dashboard, e.g. ":8080"
	DashboardAddr string `json:"dashboard_addr"`

	// StartAtEnd: when true, start tailing from end of file (recommended for live monitoring).
	StartAtEnd bool `json:"start_at_end"`

	// PollIntervalMs: polling interval for tailer checks (rotation / new data).
	PollIntervalMs int `json:"poll_interval_ms"`

	// LineQueueSize: central bounded queue for log lines (backpressure).
	LineQueueSize int `json:"line_queue_size"`

	// BatchIntervalSec: how often to emit a batch payload and update time series.
	BatchIntervalSec int `json:"batch_interval_sec"`

	// BatchQueueSize: bounded queue for outgoing metric batches.
	BatchQueueSize int `json:"batch_queue_size"`

	// HTTPTimeoutMs: timeout for sending a batch.
	HTTPTimeoutMs int `json:"http_timeout_ms"`

	// RetryMax: maximum retries per batch.
	RetryMax int `json:"retry_max"`

	// RetryBaseMs: base backoff in milliseconds.
	RetryBaseMs int `json:"retry_base_ms"`

	// RetryMaxMs: cap backoff in milliseconds.
	RetryMaxMs int `json:"retry_max_ms"`
}

func defaultConfig() Config {
	return Config{
		Logs:             []string{},
		Endpoint:         "http://localhost:9000/ingest",
		DashboardAddr:    ":8080",
		StartAtEnd:       true,
		PollIntervalMs:   250,
		LineQueueSize:    5000,
		BatchIntervalSec: 5,
		BatchQueueSize:   200,
		HTTPTimeoutMs:    5000,
		RetryMax:         5,
		RetryBaseMs:      500,
		RetryMaxMs:       30000,
	}
}

func loadConfig(path string) (Config, error) {
	cfg := defaultConfig()
	if path == "" {
		return cfg, nil
	}
	b, err := os.ReadFile(path)
	if err != nil {
		return cfg, err
	}
	if err := json.Unmarshal(b, &cfg); err != nil {
		return cfg, err
	}
	return cfg, nil
}

func main() {
	var (
		configPath = flag.String("config", "", "path to JSON config file (optional)")
		logsCSV    = flag.String("logs", "", "comma-separated log paths (override config.logs)")
		endpoint   = flag.String("endpoint", "", "metrics endpoint (override config.endpoint)")
		addr       = flag.String("addr", "", "dashboard address (override config.dashboard_addr)")
	)
	flag.Parse()

	cfg, err := loadConfig(*configPath)
	if err != nil {
		log.Fatalf("config load error: %v", err)
	}

	if strings.TrimSpace(*logsCSV) != "" {
		cfg.Logs = splitCSV(*logsCSV)
	}
	if strings.TrimSpace(*endpoint) != "" {
		cfg.Endpoint = *endpoint
	}
	if strings.TrimSpace(*addr) != "" {
		cfg.DashboardAddr = *addr
	}
	if len(cfg.Logs) == 0 {
		log.Println("no logs configured. Use -logs or config.json (field: logs).")
		log.Println("example: -logs \"./examples/app.log\"")
	}

	// Normalize log paths (keep original if it doesn't exist yet).
	for i, p := range cfg.Logs {
		if abs, e := filepath.Abs(p); e == nil {
			cfg.Logs[i] = abs
		}
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Handle Ctrl+C / termination.
	sigCh := make(chan os.Signal, 2)
	signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
	go func() {
		<-sigCh
		cancel()
		<-sigCh
		os.Exit(1)
	}()

	hostname, _ := os.Hostname()
	collectorID := stableID(hostname, cfg.Logs, cfg.Endpoint)

	lines := make(chan string, max(1, cfg.LineQueueSize))
	batches := make(chan BatchPayload, max(1, cfg.BatchQueueSize))

	m := NewMetrics(SeriesWindow{
		Step:     time.Duration(max(1, cfg.BatchIntervalSec)) * time.Second,
		Capacity: 240, // 20 minutes @ 5s step
	})

	// Start tailers.
	var wg sync.WaitGroup
	for _, path := range cfg.Logs {
		p := path
		wg.Add(1)
		go func() {
			defer wg.Done()
			TailFile(ctx, TailOptions{
				Path:          p,
				StartAtEnd:    cfg.StartAtEnd,
				PollInterval:  time.Duration(max(50, cfg.PollIntervalMs)) * time.Millisecond,
				MaxLineBytes:  1024 * 1024, // 1MB
				Emit:          func(line string) { nonBlockingSendLine(lines, line, m) },
				Logf:          func(format string, args ...any) { log.Printf("[tail] "+format, args...) },
				OnRotateEvent: func() { m.IncRotate(1) },
			})
		}()
	}

	// Aggregator: parse -> update metrics -> emit periodic batches (and time series points).
	wg.Add(1)
	go func() {
		defer wg.Done()
		Aggregate(ctx, AggregateOptions{
			Lines:        lines,
			Metrics:      m,
			Every:        time.Duration(max(1, cfg.BatchIntervalSec)) * time.Second,
			OutBatches:   batches,
			CollectorID:  collectorID,
			Hostname:     hostname,
			Endpoint:     cfg.Endpoint,
			DropOnFull:   true,
			Logf:         func(format string, args ...any) { log.Printf("[agg] "+format, args...) },
			MaxTopErrors: 8,
		})
	}()

	// Sender: POST batches with retry.
	wg.Add(1)
	go func() {
		defer wg.Done()
		SendBatches(ctx, SendOptions{
			In:          batches,
			Metrics:     m,
			Endpoint:    cfg.Endpoint,
			HTTPTimeout: time.Duration(max(500, cfg.HTTPTimeoutMs)) * time.Millisecond,
			RetryMax:    max(0, cfg.RetryMax),
			RetryBase:   time.Duration(max(50, cfg.RetryBaseMs)) * time.Millisecond,
			RetryCap:    time.Duration(max(100, cfg.RetryMaxMs)) * time.Millisecond,
			Logf:        func(format string, args ...any) { log.Printf("[send] "+format, args...) },
		})
	}()

	// Dashboard server.
	srv := &http.Server{
		Addr:              cfg.DashboardAddr,
		Handler:           NewDashboardHandler(m, cfg, collectorID),
		ReadHeaderTimeout: 3 * time.Second,
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		log.Printf("dashboard: http://localhost%s", cfg.DashboardAddr)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Printf("dashboard server error: %v", err)
			cancel()
		}
	}()

	<-ctx.Done()
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer shutdownCancel()
	_ = srv.Shutdown(shutdownCtx)
	close(lines)   // allow aggregator to drain
	close(batches) // allow sender to finish best effort
	wg.Wait()
	log.Println("stopped")
}

func splitCSV(s string) []string {
	parts := strings.Split(s, ",")
	out := make([]string, 0, len(parts))
	for _, p := range parts {
		p = strings.TrimSpace(p)
		if p != "" {
			out = append(out, p)
		}
	}
	return out
}

func stableID(hostname string, logs []string, endpoint string) string {
	h := sha1.New()
	_, _ = h.Write([]byte(hostname))
	_, _ = h.Write([]byte{0})
	_, _ = h.Write([]byte(strings.Join(logs, "|")))
	_, _ = h.Write([]byte{0})
	_, _ = h.Write([]byte(endpoint))
	return hex.EncodeToString(h.Sum(nil))[:12]
}

func nonBlockingSendLine(ch chan<- string, line string, m *Metrics) {
	select {
	case ch <- line:
	default:
		// Backpressure: drop line and count it.
		m.IncDroppedLines(1)
	}
}

func max(a, b int) int {
	if a > b {
		return a
	}
	return b
}
