// Package agent 实现 host-metrics-collector 的 Agent 端：定时采集、文件
// 日志 tail、本地 WAL、HTTP/protobuf 上报与断点续传。
package agent

import (
	"context"
	"log"
	"sync"
	"time"

	"google.golang.org/protobuf/proto"

	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/checkpoint"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/collect"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/config"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/model"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/internal/wal"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/collectorpb"
	"git.opencomputing.cn/yumoqing/host-metrics-collector/pkg/convert"
)

// 检查点键：指标与日志使用独立的 seq 空间，均以 host_id 为前缀。
const (
	metricsSeqKey = "metrics:seq:"
	logsSeqKey    = "logs:seq:"
)

// Agent 聚合采集、缓冲、WAL 与上报。
type Agent struct {
	cfg       config.AgentConfig
	store     checkpoint.Store
	collector *collect.Collector
	reporter  *Reporter

	metricsWAL *wal.WAL
	logsWAL    *wal.WAL

	metricsCh chan model.Sample
	logBuf    *logBuffer
	errCh     chan error

	mu         sync.Mutex
	metricsSeq int64
	logsSeq    int64
}

// New 根据配置构建 Agent。
func New(cfg config.AgentConfig) (*Agent, error) {
	store, err := checkpoint.New(cfg.CheckpointDriver, cfg.CheckpointFile, cfg.RedisURL)
	if err != nil {
		return nil, err
	}

	metricsWAL, err := wal.Open(cfg.CheckpointFile + ".metrics.wal")
	if err != nil {
		return nil, err
	}
	logsWAL, err := wal.Open(cfg.CheckpointFile + ".logs.wal")
	if err != nil {
		return nil, err
	}

	a := &Agent{
		cfg:        cfg,
		store:      store,
		collector:  collect.NewCollector(nil, cfg.HostID, cfg.Service),
		reporter:   NewReporter(cfg.GatewayURL, cfg.AgentToken, cfg.HostID, cfg.AgentVersion),
		metricsWAL: metricsWAL,
		logsWAL:    logsWAL,
		metricsCh:  make(chan model.Sample, 4096),
		logBuf:     newLogBuffer(),
		errCh:      make(chan error, 16),
	}
	if err := a.initSeq(); err != nil {
		return nil, err
	}
	return a, nil
}

// Run 启动 Agent 并阻塞直到 ctx 取消；取消时会尽力 flush 剩余数据。
func (a *Agent) Run(ctx context.Context) error {
	// 恢复未确认批次（断点续传）。
	if err := a.drainMetrics(ctx); err != nil {
		log.Printf("[agent] initial metrics drain: %v", err)
	}
	if err := a.drainLogs(ctx); err != nil {
		log.Printf("[agent] initial logs drain: %v", err)
	}

	go a.heartbeatLoop(ctx)
	go a.logErrors(ctx)

	// 指标采集：核心 15s / 磁盘 30s / 网络 30s / 进程 60s。
	go runGroup(ctx, a.cfg.CoreInterval, a.cfg.CollectTimeout, "core", a.collector.CollectCore, a.metricsCh, a.errCh)
	go runGroup(ctx, a.cfg.DiskInterval, a.cfg.CollectTimeout, "disk", a.collector.CollectDisk, a.metricsCh, a.errCh)
	go runGroup(ctx, a.cfg.NetworkInterval, a.cfg.CollectTimeout, "network", a.collector.CollectNetwork, a.metricsCh, a.errCh)
	go runGroup(ctx, a.cfg.ProcessInterval, a.cfg.CollectTimeout, "process", a.collector.CollectProcess, a.metricsCh, a.errCh)

	// 日志采集：每个配置的日志文件一个 tail goroutine。
	for _, file := range a.cfg.LogFiles {
		file := file
		tail := collect.NewLogTail(file, a.store, time.Second, a.logBuf.Add)
		go tail.Run(ctx)
	}

	// 上报循环。
	go a.flushMetricsLoop(ctx)
	go a.flushLogsLoop(ctx)

	<-ctx.Done()

	// 最后 flush：使用独立的超时上下文，避免因父 ctx 取消立即失败。
	finalCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	select {
	case s := <-a.metricsCh:
		_ = s // 通道在关闭时不会再投递，这里仅吞掉可能的残余。
	default:
	}
	if err := a.flushMetrics(finalCtx, nil); err != nil {
		log.Printf("[agent] final metrics flush: %v", err)
	}
	if err := a.flushLogs(finalCtx, nil); err != nil {
		log.Printf("[agent] final logs flush: %v", err)
	}
	return nil
}

// Close 释放 WAL 与检查点存储。
func (a *Agent) Close() error {
	if err := a.metricsWAL.Close(); err != nil {
		return err
	}
	if err := a.logsWAL.Close(); err != nil {
		return err
	}
	return a.store.Close()
}

func (a *Agent) heartbeatLoop(ctx context.Context) {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if _, err := a.reporter.Heartbeat(ctx, 0); err != nil {
				log.Printf("[agent] heartbeat: %v", err)
			}
		}
	}
}

func (a *Agent) logErrors(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case err := <-a.errCh:
			log.Printf("[agent] collect: %v", err)
		}
	}
}

// initSeq 从检查点与 WAL 恢复下一个可分配的 seq。
func (a *Agent) initSeq() error {
	a.mu.Lock()
	defer a.mu.Unlock()

	ackedM, err := a.store.GetAckedSeq(metricsSeqKey + a.cfg.HostID)
	if err != nil {
		ackedM = 0
	}
	a.metricsSeq = ackedM
	if recs, err := a.metricsWAL.Load(); err == nil {
		for _, r := range recs {
			if r.Seq > a.metricsSeq {
				a.metricsSeq = r.Seq
			}
		}
	}

	ackedL, err := a.store.GetAckedSeq(logsSeqKey + a.cfg.HostID)
	if err != nil {
		ackedL = 0
	}
	a.logsSeq = ackedL
	if recs, err := a.logsWAL.Load(); err == nil {
		for _, r := range recs {
			if r.Seq > a.logsSeq {
				a.logsSeq = r.Seq
			}
		}
	}
	return nil
}

func (a *Agent) nextMetricsSeq() int64 {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.metricsSeq++
	return a.metricsSeq
}

func (a *Agent) nextLogsSeq() int64 {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.logsSeq++
	return a.logsSeq
}

// flushMetrics 将内存缓冲样本写入 WAL 并尝试上报全部未确认批次。
func (a *Agent) flushMetrics(ctx context.Context, samples []model.Sample) error {
	if len(samples) > 0 {
		seq := a.nextMetricsSeq()
		batch := &collectorpb.MetricBatch{
			HostId:       a.cfg.HostID,
			AgentVersion: a.cfg.AgentVersion,
			Seq:          seq,
			SentAt:       time.Now().Unix(),
			Samples:      make([]*collectorpb.MetricSample, 0, len(samples)),
		}
		for _, s := range samples {
			batch.Samples = append(batch.Samples, convert.SampleToPB(s))
		}
		payload, err := proto.Marshal(batch)
		if err != nil {
			return err
		}
		if err := a.metricsWAL.Append(seq, payload); err != nil {
			return err
		}
	}
	return a.drainMetrics(ctx)
}

func (a *Agent) drainMetrics(ctx context.Context) error {
	key := metricsSeqKey + a.cfg.HostID
	acked, err := a.store.GetAckedSeq(key)
	if err != nil {
		acked = 0
	}
	recs, err := a.metricsWAL.Load()
	if err != nil {
		return err
	}

	for _, r := range recs {
		if r.Seq <= acked {
			continue
		}
		var batch collectorpb.MetricBatch
		if err := proto.Unmarshal(r.Payload, &batch); err != nil {
			return err
		}
		if _, err := a.reporter.SendMetrics(ctx, &batch); err != nil {
			return err
		}
		acked = r.Seq
		if err := a.store.SetAckedSeq(key, acked); err != nil {
			return err
		}
	}
	return a.metricsWAL.TruncateBefore(acked + 1)
}

func (a *Agent) flushMetricsLoop(ctx context.Context) {
	ticker := time.NewTicker(a.cfg.FlushInterval)
	defer ticker.Stop()

	buf := make([]model.Sample, 0, a.cfg.BatchSize)
	for {
		select {
		case s := <-a.metricsCh:
			buf = append(buf, s)
			if len(buf) >= a.cfg.BatchSize {
				if err := a.flushMetrics(ctx, buf); err != nil {
					log.Printf("[agent] flush metrics: %v", err)
				}
				buf = buf[:0]
			}
		case <-ticker.C:
			if err := a.flushMetrics(ctx, buf); err != nil {
				log.Printf("[agent] flush metrics: %v", err)
			}
			buf = buf[:0]
		case <-ctx.Done():
			return
		}
	}
}

// flushLogs 将内存缓冲日志写入 WAL 并尝试上报全部未确认批次。
func (a *Agent) flushLogs(ctx context.Context, records []model.LogRecord) error {
	if len(records) > 0 {
		seq := a.nextLogsSeq()
		batch := &collectorpb.LogBatch{
			HostId:  a.cfg.HostID,
			Seq:     seq,
			Entries: make([]*collectorpb.LogEntry, 0, len(records)),
		}
		for _, r := range records {
			batch.Entries = append(batch.Entries, convert.LogToPB(r))
		}
		payload, err := proto.Marshal(batch)
		if err != nil {
			return err
		}
		if err := a.logsWAL.Append(seq, payload); err != nil {
			return err
		}
	}
	return a.drainLogs(ctx)
}

func (a *Agent) drainLogs(ctx context.Context) error {
	key := logsSeqKey + a.cfg.HostID
	acked, err := a.store.GetAckedSeq(key)
	if err != nil {
		acked = 0
	}
	recs, err := a.logsWAL.Load()
	if err != nil {
		return err
	}

	for _, r := range recs {
		if r.Seq <= acked {
			continue
		}
		var batch collectorpb.LogBatch
		if err := proto.Unmarshal(r.Payload, &batch); err != nil {
			return err
		}
		if _, err := a.reporter.SendLogs(ctx, &batch); err != nil {
			return err
		}
		acked = r.Seq
		if err := a.store.SetAckedSeq(key, acked); err != nil {
			return err
		}
	}
	return a.logsWAL.TruncateBefore(acked + 1)
}

func (a *Agent) flushLogsLoop(ctx context.Context) {
	ticker := time.NewTicker(a.cfg.FlushInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			records := a.logBuf.TakeAll()
			if err := a.flushLogs(ctx, records); err != nil {
				log.Printf("[agent] flush logs: %v", err)
			}
		case <-ctx.Done():
			return
		}
	}
}

// logBuffer 是日志记录的内存缓冲，避免 tail goroutine 在通道满时阻塞或丢日志。
type logBuffer struct {
	mu    sync.Mutex
	items []model.LogRecord
}

func newLogBuffer() *logBuffer {
	return &logBuffer{items: make([]model.LogRecord, 0, 4096)}
}

func (b *logBuffer) Add(r model.LogRecord) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.items = append(b.items, r)
}

func (b *logBuffer) TakeAll() []model.LogRecord {
	b.mu.Lock()
	defer b.mu.Unlock()
	out := b.items
	b.items = make([]model.LogRecord, 0, 4096)
	return out
}