package relay

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"testing"
	"time"
)

func TestMeterFlushAppendsDeltas(t *testing.T) {
	s := newMemStore(t)
	now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now }, nil)
	id := Identity{"abc", "meet"}
	m.Add(id, 100, 200, 5)
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}
	m.Add(id, 10, 20, 0)
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}
	body, _, _ := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
	var c MeterChunk
	json.Unmarshal(body, &c)
	if len(c.Samples) != 2 || c.Samples[0].Apps["abc/meet"].In != 100 || c.Samples[1].Apps["abc/meet"].Out != 20 {
		t.Fatalf("chunk = %+v", c)
	}
}

func TestLoadWindowSpansTwoHours(t *testing.T) {
	s := newMemStore(t)
	now := time.Date(2026, 9, 7, 12, 1, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }, nil) // 11:31
	m.Add(Identity{"abc", "meet"}, 7, 0, 0)
	m.Flush(context.Background(), s)
	samples, err := LoadWindow(context.Background(), s, "box1", now, nil)
	if err != nil || len(samples) != 1 || samples[0].Apps["abc/meet"].In != 7 {
		t.Fatalf("window = %+v %v", samples, err)
	}
}

// TestMeterFlushPreservesCorruptChunk checks that a Flush landing on a
// chunk that fails to unmarshal doesn't just clobber it: the unreadable
// bytes are preserved as a ".corrupt-<unix>" sibling, a fresh chunk is
// written in the real key's place, and the corruption is logged.
func TestMeterFlushPreservesCorruptChunk(t *testing.T) {
	s := newMemStore(t)
	now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
	key := MeterKey("box1", "2026-09-07-12")
	if _, err := s.Put(context.Background(), key, []byte("not json"), ""); err != nil {
		t.Fatal(err)
	}

	var logs []string
	m := NewMeter("box1", func() time.Time { return now }, func(format string, args ...any) {
		logs = append(logs, fmt.Sprintf(format, args...))
	})
	m.Add(Identity{"abc", "meet"}, 1, 2, 0)
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}

	body, _, err := s.Get(context.Background(), key)
	if err != nil {
		t.Fatal(err)
	}
	var c MeterChunk
	if err := json.Unmarshal(body, &c); err != nil {
		t.Fatalf("chunk at %s still unreadable: %v", key, err)
	}
	if len(c.Samples) != 1 {
		t.Fatalf("chunk = %+v", c)
	}

	keys, err := s.List(context.Background(), "metrics/relay/box1/")
	if err != nil {
		t.Fatal(err)
	}
	found := false
	for _, k := range keys {
		if strings.Contains(k, ".corrupt-") {
			found = true
		}
	}
	if !found {
		t.Fatalf("no .corrupt- sibling among %v", keys)
	}
	if len(logs) == 0 {
		t.Fatal("expected a log message about the corrupt chunk")
	}
}

// TestMeterEvictsDeadApps checks that an app with nothing left to report
// (zero deltas, zero allocations) drops out of the meter's live set — and
// out of the sample — after being flushed once, while an app still holding
// open allocations keeps being reported even with zero deltas.
func TestMeterEvictsDeadApps(t *testing.T) {
	s := newMemStore(t)
	now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now }, nil)
	a := Identity{"abc", "meet"}
	b := Identity{"abc", "docs"}
	m.Add(a, 1, 2, 0)
	m.Add(b, 3, 4, 0)
	m.SetAllocs(a, 1)
	m.SetAllocs(b, 0)

	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}

	body, _, err := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
	if err != nil {
		t.Fatal(err)
	}
	var c MeterChunk
	if err := json.Unmarshal(body, &c); err != nil {
		t.Fatal(err)
	}
	if len(c.Samples) != 2 {
		t.Fatalf("chunk = %+v", c)
	}

	second := c.Samples[1]
	appA, ok := second.Apps["abc/meet"]
	if !ok || appA.Allocs != 1 || appA.In != 0 || appA.Out != 0 || appA.Rejected != 0 {
		t.Fatalf("second sample abc/meet = %+v ok=%v", appA, ok)
	}
	if _, ok := second.Apps["abc/docs"]; ok {
		t.Fatalf("second sample still has dead app abc/docs: %+v", second.Apps)
	}
}

// TestMeterFlushOfAnIdleRelayWritesNothing: a sample naming no app carries
// no information, and a relay box is idle almost all of the time — left
// unguarded, the flush loop writes 1,440 empty rows a day per box, each one
// a read, a write, a CAS and a row every reader of the hour has to walk. An
// hour in which nothing happened is best represented by no object at all.
func TestMeterFlushOfAnIdleRelayWritesNothing(t *testing.T) {
	s := newMemStore(t)
	now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now }, nil)
	for i := 0; i < 2; i++ {
		if err := m.Flush(context.Background(), s); err != nil {
			t.Fatalf("flush %d: %v", i, err)
		}
	}
	keys, err := s.List(context.Background(), "metrics/relay/box1/")
	if err != nil {
		t.Fatal(err)
	}
	if len(keys) != 0 {
		t.Fatalf("an idle relay wrote %v, want nothing", keys)
	}
	// And the first byte that IS forwarded still lands, so this is a skip,
	// not a mute.
	m.Add(Identity{"abc", "meet"}, 1, 0, 0)
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}
	if keys, _ := s.List(context.Background(), "metrics/relay/box1/"); len(keys) != 1 {
		t.Fatalf("after traffic, keys = %v, want the hour's one chunk", keys)
	}
}

// TestSetAllocsAllZeroesDepartedApps is the leak test. SetAllocs alone can
// only ever raise a gauge, because the ids that need lowering are precisely
// the ones a census of live allocations no longer names — so an app whose
// last call ended would report a phantom allocation in every sample from
// then on and would never be evicted, growing m.cur without bound in a
// process the unit caps at 256 MiB.
func TestSetAllocsAllZeroesDepartedApps(t *testing.T) {
	s := newMemStore(t)
	now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now }, nil)
	a := Identity{"abc", "meet"}
	b := Identity{"abc", "docs"}

	// Both live and forwarding.
	m.Add(a, 10, 20, 0)
	m.Add(b, 30, 40, 0)
	m.SetAllocsAll(map[Identity]int{a: 1, b: 2})
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}
	// b's call ends part way through the next minute: it forwarded some
	// bytes and then went, so the census no longer names it.
	m.Add(b, 5, 5, 0)
	m.SetAllocsAll(map[Identity]int{a: 1})
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}
	// And a third flush finds nothing left to say about b at all.
	m.SetAllocsAll(map[Identity]int{a: 1})
	if err := m.Flush(context.Background(), s); err != nil {
		t.Fatal(err)
	}

	body, _, err := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
	if err != nil {
		t.Fatal(err)
	}
	var c MeterChunk
	if err := json.Unmarshal(body, &c); err != nil {
		t.Fatal(err)
	}
	if len(c.Samples) != 3 {
		t.Fatalf("want 3 samples, got %+v", c.Samples)
	}
	if got := c.Samples[0].Apps["abc/docs"].Allocs; got != 2 {
		t.Errorf("first sample abc/docs allocs = %d, want 2", got)
	}
	// The gauge came back DOWN. Without the census this would still read 2,
	// which is the whole finding.
	if got, ok := c.Samples[1].Apps["abc/docs"]; !ok || got.Allocs != 0 || got.In != 5 {
		t.Errorf("second sample abc/docs = %+v ok=%v, want its last 5 bytes with allocs 0", got, ok)
	}
	if _, ok := c.Samples[2].Apps["abc/docs"]; ok {
		t.Errorf("third sample still carries the departed app: %+v", c.Samples[2].Apps)
	}
	if got := c.Samples[2].Apps["abc/meet"].Allocs; got != 1 {
		t.Errorf("third sample abc/meet allocs = %d, want the live app kept at 1", got)
	}
	// And it is out of the meter's live set, not merely out of the sample:
	// the gauge that never fell was a leak as much as a lie.
	m.mu.Lock()
	n := len(m.cur)
	m.mu.Unlock()
	if n != 1 {
		t.Errorf("meter still carries %d apps, want only the live one", n)
	}
}