package main

import (
	"context"
	"time"
)

type BatchPayload struct {
	CollectorID string            `json:"collector_id"`
	Hostname    string            `json:"hostname"`
	Endpoint    string            `json:"endpoint"`
	Timestamp   string            `json:"timestamp"`
	Levels      map[string]uint64 `json:"levels"`
	Counters    map[string]uint64 `json:"counters"`
	HTTP        map[string]uint64 `json:"http"`
	TopErrors   []KV              `json:"top_errors"`
}

type AggregateOptions struct {
	Lines      <-chan string
	Metrics    *Metrics
	Every      time.Duration
	OutBatches chan<- BatchPayload

	CollectorID string
	Hostname    string
	Endpoint    string

	DropOnFull bool
	Logf       func(string, ...any)

	MaxTopErrors int
}

func Aggregate(ctx context.Context, opt AggregateOptions) {
	if opt.Every <= 0 {
		opt.Every = 5 * time.Second
	}
	logf := opt.Logf
	if logf == nil {
		logf = func(string, ...any) {}
	}

	// Rolling counters per interval for time series.
	intervalLevels := map[string]uint64{}

	ticker := time.NewTicker(opt.Every)
	defer ticker.Stop()

	flush := func(ts time.Time) {
		// snapshot interval series point
		opt.Metrics.PushSeriesPoint(ts, intervalLevels)
		intervalLevels = map[string]uint64{}

		// build outgoing payload
		snap := opt.Metrics.Snapshot(opt.CollectorID, opt.Hostname, opt.Endpoint, opt.MaxTopErrors)
		payload := BatchPayload{
			CollectorID: opt.CollectorID,
			Hostname:    opt.Hostname,
			Endpoint:    opt.Endpoint,
			Timestamp:   time.Now().UTC().Format(time.RFC3339Nano),
			Levels:      snap.Levels,
			Counters:    snap.Counters,
			HTTP:        snap.HTTP,
			TopErrors:   snap.TopErrors,
		}

		select {
		case opt.OutBatches <- payload:
		default:
			if opt.DropOnFull {
				opt.Metrics.IncBatchDropped(1)
				logf("batch queue full -> dropped one payload")
			} else {
				// block if not dropping
				opt.OutBatches <- payload
			}
		}
	}

	for {
		select {
		case <-ctx.Done():
			return

		case line, ok := <-opt.Lines:
			if !ok {
				return
			}
			opt.Metrics.IncLines(1)

			parsed, ok := ParseLine(line)
			if !ok {
				opt.Metrics.IncParseErrors(1)
				continue
			}
			opt.Metrics.IncLevel(parsed.Level, 1)
			intervalLevels[parsed.Level]++

			if parsed.HTTPCode > 0 {
				opt.Metrics.IncHTTP(parsed.HTTPCode, 1)
			}
			if len(parsed.ErrKW) > 0 {
				opt.Metrics.IncErrKW(parsed.ErrKW)
			}

		case t := <-ticker.C:
			flush(t)
		}
	}
}
