package counting

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/alicebob/miniredis/v2"
	"github.com/redis/go-redis/v9"

	"migadu_counters/internal/store"
)

func TestWarnThreshold(t *testing.T) {
	tests := []struct {
		limit, percent, want int64
	}{
		{1, 80, 0}, // ceil(0.8) = 1 is the limit itself: no warning
		{2, 80, 0}, // ceil(1.6) = 2 is the limit itself: no warning
		{5, 80, 4},
		{100, 80, 80},
		{101, 80, 81},
		{100, 0, 0}, // disabled
		{0, 80, 0},  // no limit, no warning
		{-5, 80, 0},
		{1 << 62, 50, 1 << 61}, // no overflow
	}
	for _, tt := range tests {
		if got := WarnThreshold(tt.limit, tt.percent); got != tt.want {
			t.Errorf("WarnThreshold(%d, %d) = %d, want %d", tt.limit, tt.percent, got, tt.want)
		}
	}
}

func warnCounter(t *testing.T) (*Counter, *miniredis.Miniredis) {
	t.Helper()
	c, m := newCounter(t)
	c.Planner.WarnPercent = 80
	return c, m
}

func TestPlan_SetsWarnAt(t *testing.T) {
	p := newPlanner(false)
	p.WarnPercent = 80
	plan, err := p.PlanOut("w", "noreply@example.com", "", "ext@gmail.com")
	if err != nil {
		t.Fatal(err)
	}
	got := []int64{plan.Increments[0].WarnAt, plan.Increments[1].WarnAt, plan.Increments[2].WarnAt}
	if got[0] != 4 || got[1] != 1760 || got[2] != 1760 {
		t.Errorf("WarnAt = %v, want [4 1760 1760]", got)
	}
	p.WarnPercent = 0
	plan, _ = p.PlanOut("w", "noreply@example.com", "", "ext@gmail.com")
	if plan.Increments[0].WarnAt != 0 {
		t.Errorf("WarnAt with warnings off = %d, want 0", plan.Increments[0].WarnAt)
	}
}

// noreply@example.com has A/o=5, so 80% is the 4th message.
func TestCounter_WarnsOnceThenBreachesOnce(t *testing.T) {
	c, _ := warnCounter(t)
	ctx := context.Background()
	for i := 1; i <= 7; i++ {
		out, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com")
		if err != nil {
			t.Fatal(err)
		}
		a := out.Results[0]
		if a.Warned != (i == 4) {
			t.Errorf("message %d: warned = %v", i, a.Warned)
		}
		if a.Reached != (i == 5) {
			t.Errorf("message %d: reached = %v", i, a.Reached)
		}
	}
}

// Markers must stay out of the cnt/ namespace: the web app scans cnt/*<address>*.
func TestCounter_ConfirmedCrossingWritesMarker(t *testing.T) {
	c, m := warnCounter(t)
	ctx := context.Background()
	for i := 0; i < 5; i++ {
		if _, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com"); err != nil {
			t.Fatal(err)
		}
	}
	for _, kind := range []string{store.MarkerLimit, store.MarkerWarn} {
		key := store.MarkerKey("A", "noreply@example.com", day, "o", kind)
		if !strings.HasPrefix(key, "ntf/") {
			t.Errorf("marker key %q must not share the cnt/ namespace", key)
		}
		if !m.Exists(key) {
			t.Errorf("marker %s not written; keys = %v", key, m.Keys())
		}
		if ttl := m.TTL(key); ttl != store.MarkerTTL {
			t.Errorf("marker %s TTL = %v, want %v", key, ttl, store.MarkerTTL)
		}
	}
}

// lostReplyCounter counts through a proxy that delays the next EXEC reply past
// the client's read timeout, so Redis applies the transaction but the reply is lost.
func lostReplyCounter(t *testing.T, m *miniredis.Miniredis) *Counter {
	t.Helper()
	proxy := delayProxy(t, m.Addr(), 1, 800*time.Millisecond)
	st := store.NewWithClients(redis.NewClient(&redis.Options{Addr: proxy, ReadTimeout: 200 * time.Millisecond, PoolSize: 2}), nil)
	t.Cleanup(func() { st.Close() })
	p := newPlanner(false)
	p.WarnPercent = 80
	return &Counter{Planner: p, Store: st}
}

// A counter that already crossed and notified must stay silent on every later
// lost reply.
func TestCounter_LostReplyAfterBreachDoesNotNotifyAgain(t *testing.T) {
	c, m := warnCounter(t)
	ctx := context.Background()
	for i := 0; i < 5; i++ {
		if _, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com"); err != nil {
			t.Fatal(err)
		}
	}
	out, err := lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com")
	if err != nil {
		t.Fatal(err)
	}
	if out.Unknown == nil {
		t.Fatal("reply should have been lost")
	}
	if a := out.Results[0]; a.Count != 6 || a.Reached || a.Warned {
		t.Errorf("A = %+v, want count 6, no reached, no warned", a)
	}
}

// When the crossing increment itself loses its reply it notifies exactly once,
// however many later replies are lost too.
func TestCounter_LostReplyOnCrossingNotifiesOnce(t *testing.T) {
	m := miniredis.RunT(t)
	key := store.Key("A", "noreply@example.com", day, "o")
	m.Set(key, "4") // A/o=5: the next increment crosses
	ctx := context.Background()
	reached := 0
	for i := 0; i < 2; i++ {
		out, err := lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com")
		if err != nil {
			t.Fatal(err)
		}
		if out.Unknown == nil {
			t.Fatalf("round %d: reply should have been lost", i)
		}
		if out.Results[0].Reached {
			reached++
		}
	}
	if reached != 1 {
		t.Errorf("reached %d times over two lost replies, want 1", reached)
	}
}

func TestCounter_LostReplyWarnsOnce(t *testing.T) {
	m := miniredis.RunT(t)
	key := store.Key("A", "noreply@example.com", day, "o")
	m.Set(key, "3") // 80% of A/o=5 is 4: the next increment warns
	ctx := context.Background()
	out, err := lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com")
	if err != nil {
		t.Fatal(err)
	}
	if a := out.Results[0]; a.Count != 4 || !a.Warned || a.Reached {
		t.Errorf("A = %+v, want count 4 warned", a)
	}
	m.Set(key, "3") // second lost reply lands on the same value
	out, err = lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com")
	if err != nil {
		t.Fatal(err)
	}
	if a := out.Results[0]; a.Warned {
		t.Errorf("A = %+v, must not warn twice", a)
	}
}

// Crossed says a limit of 0 never fires; the lost-reply path must agree.
func TestCounter_LostReplyZeroLimitNeverFires(t *testing.T) {
	m := miniredis.RunT(t)
	c := lostReplyCounter(t, m)
	plan := &Plan{Route: RouteOut, Date: day, Increments: []Increment{
		{Level: "A", ID: "x@example.com", Flow: "o", Key: store.Key("A", "x@example.com", day, "o"), Limited: true, Limit: 0},
	}}
	out, err := c.Execute(context.Background(), plan)
	if err != nil {
		t.Fatal(err)
	}
	if out.Unknown == nil {
		t.Fatal("reply should have been lost")
	}
	if out.Results[0].Reached {
		t.Errorf("limit 0 fired after a lost reply: %+v", out.Results[0])
	}
}

// A transaction Redis aborted (READONLY during a failover) applied nothing, so
// it is safe to run it once more.
func TestCounter_AbortedTransactionIsRetried(t *testing.T) {
	c, m := newCounter(t)
	c.AbortRetryDelay = 150 * time.Millisecond
	m.SetError("READONLY You can't write against a read only replica.")
	go func() {
		time.Sleep(30 * time.Millisecond)
		m.SetError("")
	}()
	out, err := c.CountOut(context.Background(), "fo", "noreply@example.com", "", "ext@gmail.com")
	if err != nil {
		t.Fatalf("retry should have counted: %v", err)
	}
	if out.Unknown != nil || out.Results[0].Err != nil || out.Results[0].Count != 1 {
		t.Errorf("outcome = %+v unknown=%v", out.Results[0], out.Unknown)
	}
	if v, _ := m.Get(store.Key("A", "noreply@example.com", day, "o")); v != "1" {
		t.Errorf("redis A/o = %q, want exactly 1", v)
	}
}

func TestCounter_AbortedTwiceIsAWholeFailure(t *testing.T) {
	c, m := newCounter(t)
	c.AbortRetryDelay = time.Millisecond
	m.SetError("READONLY You can't write against a read only replica.")
	out, err := c.CountOut(context.Background(), "fo", "noreply@example.com", "", "ext@gmail.com")
	if err == nil || !errors.Is(err, store.ErrNotApplied) {
		t.Fatalf("err = %v, outcome = %+v; want ErrNotApplied", err, out)
	}
	m.SetError("")
	if keys := m.Keys(); len(keys) != 0 {
		t.Errorf("nothing should be written, keys = %v", keys)
	}
}

// A job that waited in the queue across midnight UTC counts on the day it arrived.
func TestPlan_DayIsWhenTheRequestArrived(t *testing.T) {
	p := newPlanner(false)
	received := fixedNow.Add(-24 * time.Hour)
	plan, err := p.Plan(Job{Route: RouteIn, Recipient: "y@example.com", Received: received})
	if err != nil {
		t.Fatal(err)
	}
	if want := store.DayKey(received); plan.Date != want {
		t.Errorf("plan date = %s, want %s", plan.Date, want)
	}
	plan, _ = p.Plan(Job{Route: RouteIn, Recipient: "y@example.com"})
	if plan.Date != day {
		t.Errorf("plan date without Received = %s, want %s", plan.Date, day)
	}
}