package apperr

import (
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/applog"
)

// MaxGroupsPerAppHour caps coalesced records per app per UTC hour, counting
// distinct (fp, host, release) combinations admitted within the hour, across
// every drain (spec §3.2) — not admissions since the last drain. A key once
// admitted this hour stays admitted; only genuinely new keys count against
// the cap. Per app, not per box: one tenant's storm must not suppress a
// neighbour's.
const MaxGroupsPerAppHour = 1000

// OverflowFP names the one row the excess beyond the cap folds into. Its
// Count is occurrences, and the tab labels it as such.
const OverflowFP = "_overflow"

// Sink coalesces errors between ship-loop flushes: per app, one record per
// (fp, host, release) carrying count/first/last and the first sample. An
// error storm is the moment this must work — ten thousand identical panics
// in a minute become one record. Safe for concurrent use; the edge, the
// activator and the ship loop share one Sink per agent process.
type Sink struct {
	box string
	now func() time.Time

	mu   sync.Mutex
	apps map[applog.AppKey]*appBuf
	seq  uint64
}

type groupKey struct{ fp, host, release string }

type appBuf struct {
	groups   map[groupKey]*Record
	order    []groupKey        // insertion order, for a deterministic drain
	overflow [2]*Record        // one per family, indexed by IsAvailability: Errors' excess never lands on Availability's row, nor the reverse
	hour     string            // the UTC hour `seen` counts for
	seen     map[groupKey]bool // keys admitted this hour, across every drain — the cap counter; Drain must not touch it
}

// NewSink returns a Sink stamping records with box and timestamps from now.
// seq starts at UnixMilli for the reason applog.NewSink gives: chunk keys
// embed the first record's seq, and a restart must sort after last boot.
func NewSink(box string, now func() time.Time) *Sink {
	return &Sink{box: box, now: now, apps: make(map[applog.AppKey]*appBuf), seq: uint64(now().UnixMilli())}
}

// Add buffers one occurrence for (acct, app). r is sanitised here (§3.1),
// fingerprinted when it carries no FP, and folded into its group.
func (s *Sink) Add(acct, app string, r Record) {
	now := s.now().UTC()
	r.Name = Sanitize(r.Name)
	r.Path = SanitizePath(r.Path)
	if r.Path == "/" && r.Status == 0 && r.Source != SourceEdge {
		r.Path = ""
	}
	if len(r.Stack) > MaxStack {
		r.Stack = r.Stack[:MaxStack]
	}
	r.Stack = Sanitize(r.Stack)
	if len(r.Fields) > 0 {
		f := make(map[string]string, len(r.Fields))
		for k, v := range r.Fields {
			f[k] = Sanitize(v)
		}
		r.Fields = f
	}
	if r.FP == "" {
		r.FP = Fingerprint(r.Source, r.Kind, r.Name, r.Stack)
		r.FPV = FPVersion
	}
	if len(r.FP) > 64 {
		r.FP = r.FP[:64]
	}

	s.mu.Lock()
	defer s.mu.Unlock()
	b := s.apps[applog.AppKey{Acct: acct, App: app}]
	if b == nil {
		b = &appBuf{groups: make(map[groupKey]*Record)}
		s.apps[applog.AppKey{Acct: acct, App: app}] = b
	}
	if h := applog.HourOf(now); h != b.hour {
		b.hour, b.seen = h, make(map[groupKey]bool)
	}
	k := groupKey{r.FP, r.Host, r.Release}
	if g := b.groups[k]; g != nil {
		g.Count++
		g.Last = now
		return
	}
	if !b.seen[k] && len(b.seen) >= MaxGroupsPerAppHour {
		fam := 0
		if IsAvailability(r) {
			fam = 1
		}
		if b.overflow[fam] == nil {
			o := r
			o.FP = OverflowFP
			o.TS, o.First, o.Last, o.Count = now, now, now, 0
			o.Box = s.box
			s.seq++
			o.Seq = s.seq
			b.overflow[fam] = &o
		}
		b.overflow[fam].Count++
		b.overflow[fam].Last = now
		return
	}
	b.seen[k] = true
	s.seq++
	r.TS, r.First, r.Last, r.Count = now, now, now, 1
	r.Box = s.box
	r.Seq = s.seq
	b.groups[k] = &r
	b.order = append(b.order, k)
}

// Drain returns everything buffered, keyed by app, and resets the groups
// (the per-hour cap counter `seen` and seq persist — they are the hour's and
// the process's, not the batch's; a key seen in an earlier drain this hour
// stays admitted without counting against the cap again). Empty apps are
// omitted.
func (s *Sink) Drain() map[applog.AppKey][]Record {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make(map[applog.AppKey][]Record)
	for k, b := range s.apps {
		if len(b.order) == 0 && b.overflow == [2]*Record{} {
			continue
		}
		recs := make([]Record, 0, len(b.order)+2)
		for _, gk := range b.order {
			recs = append(recs, *b.groups[gk])
		}
		for _, o := range b.overflow {
			if o != nil {
				recs = append(recs, *o)
			}
		}
		out[k] = recs
		b.groups = make(map[groupKey]*Record)
		b.order = nil
		b.overflow = [2]*Record{}
	}
	return out
}

// Known lists every app this sink has ever buffered for — the prune
// worklist, as applog.Sink.Known is for logs.
func (s *Sink) Known() []applog.AppKey {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]applog.AppKey, 0, len(s.apps))
	for k := range s.apps {
		out = append(out, k)
	}
	return out
}