package analytics

// TopKSize bounds a top-keys summary (pages, referers, events) per app
// per day per box (spec §3.3). Misra–Gries: exact below the cap, a lower
// bound within the summary's error above it, mergeable, with the evicted
// mass kept in OtherKey so totals are conserved.
const TopKSize = 500

// OtherKey is the fold-over row for keys beyond the cap.
const OtherKey = "_other"

// MaxKeyLen caps a key's bytes.
const MaxKeyLen = 256

// TopK is a Misra–Gries summary with TopKSize counters.
type TopK struct {
	counts map[string]int64
	other  int64
}

// NewTopK returns an empty summary.
func NewTopK() *TopK { return &TopK{counts: make(map[string]int64)} }

// Add counts n occurrences of key. When the summary is full and key is
// new, every counter is decremented by the smallest amount that frees a
// slot, and the decremented mass moves to other.
func (t *TopK) Add(key string, n int64) {
	if len(key) > MaxKeyLen {
		key = key[:MaxKeyLen]
	}
	if n <= 0 {
		return
	}
	if _, ok := t.counts[key]; ok || len(t.counts) < TopKSize {
		t.counts[key] += n
		return
	}
	// Full and new: take the smallest counter as the decrement.
	min := int64(-1)
	for _, v := range t.counts {
		if min < 0 || v < min {
			min = v
		}
	}
	if n <= min {
		t.other += n
		return
	}
	for k, v := range t.counts {
		if v <= min {
			delete(t.counts, k)
			t.other += v
		} else {
			t.counts[k] = v - min
			t.other += min
		}
	}
	t.counts[key] = n - min
	t.other += min
}

// Merge folds o into t: counters add, other adds, then t re-folds to the
// cap so a merge of two full summaries stays bounded.
func (t *TopK) Merge(o *TopK) {
	merged := NewTopK()
	merged.other = t.other + o.other
	sum := make(map[string]int64, len(t.counts)+len(o.counts))
	for k, v := range t.counts {
		sum[k] += v
	}
	for k, v := range o.counts {
		sum[k] += v
	}
	for k, v := range sum { // Add re-applies the cap
		merged.Add(k, v)
	}
	t.counts, t.other = merged.counts, merged.other
}

// Fold returns the counters plus OtherKey when any mass fell over.
func (t *TopK) Fold() map[string]int64 {
	out := make(map[string]int64, len(t.counts)+1)
	for k, v := range t.counts {
		out[k] = v
	}
	if t.other > 0 {
		out[OtherKey] = t.other
	}
	return out
}

// TopKFrom rebuilds a summary from a folded map (the day file's shape).
func TopKFrom(m map[string]int64) *TopK {
	t := NewTopK()
	for k, v := range m {
		if k == OtherKey {
			t.other += v
			continue
		}
		t.Add(k, v)
	}
	return t
}

// Approximate reports whether counters are lower bounds (mass fell over).
func (t *TopK) Approximate() bool { return t.other > 0 }