package batcher

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"sync"
	"testing"
	"time"

	"internetpressure.io/probe-agent/internal/client"
	"internetpressure.io/probe-agent/internal/protocol"
	"internetpressure.io/probe-agent/internal/spool"
)

// fakePoster records batches and can be switched to failing.
type fakePoster struct {
	mu      sync.Mutex
	batches []protocol.Batch
	fail    error
	calls   int
}

func (p *fakePoster) PostBatch(_ context.Context, gz []byte, _ time.Duration) (*protocol.BatchResponse, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.calls++
	if p.fail != nil {
		return nil, p.fail
	}
	zr, err := gzip.NewReader(bytes.NewReader(gz))
	if err != nil {
		return nil, err
	}
	var b protocol.Batch
	if err := json.NewDecoder(zr).Decode(&b); err != nil {
		return nil, err
	}
	p.batches = append(p.batches, b)
	return &protocol.BatchResponse{Accepted: len(b.Measurements), ConfigVersion: "v1", ServerTime: protocol.FormatTime(time.Now())}, nil
}

func (p *fakePoster) setFail(err error) { p.mu.Lock(); p.fail = err; p.mu.Unlock() }
func (p *fakePoster) count() int        { p.mu.Lock(); defer p.mu.Unlock(); return len(p.batches) }

func newBatcher(t *testing.T, p Poster, hooks Hooks) (*Batcher, *spool.Spool) {
	sp, err := spool.Open(t.TempDir(), 0)
	if err != nil {
		t.Fatal(err)
	}
	return New("ca-qc-01", "0.1.0", p, sp, hooks, slog.New(slog.DiscardHandler)), sp
}

func meas(n int) []protocol.Measurement {
	out := make([]protocol.Measurement, n)
	for i := range out {
		out[i] = protocol.Measurement{TS: protocol.FormatTime(time.Now()), TargetID: "t", Kind: "http", OK: true}
	}
	return out
}

func TestBatchSizeLimit(t *testing.T) {
	p := &fakePoster{}
	b, _ := newBatcher(t, p, Hooks{})
	b.Configure(10, 500)
	b.Add(meas(1200)...)
	b.AddTraceroute(protocol.Traceroute{TargetID: "t"})
	b.Flush(context.Background())
	if b.Buffered() != 0 {
		t.Fatalf("queue should be drained, has %d", b.Buffered())
	}
	if p.count() != 3 {
		t.Fatalf("1200 measurements should give 3 batches, got %d", p.count())
	}
	for i, batch := range p.batches {
		if len(batch.Measurements) > 500 {
			t.Fatalf("batch %d has %d > max_batch", i, len(batch.Measurements))
		}
		if batch.ProbeID != "ca-qc-01" || batch.AgentVersion != "0.1.0" || batch.SentAt == "" {
			t.Fatalf("envelope: %+v", batch)
		}
	}
	if len(p.batches[0].Measurements) != 500 || len(p.batches[2].Measurements) != 200 || len(p.batches[0].Traceroutes) != 1 {
		t.Fatalf("split: %d %d %d, trs %d", len(p.batches[0].Measurements), len(p.batches[1].Measurements), len(p.batches[2].Measurements), len(p.batches[0].Traceroutes))
	}
}

func TestHealthOncePerMinute(t *testing.T) {
	p := &fakePoster{}
	calls := 0
	b, _ := newBatcher(t, p, Hooks{Health: func() *protocol.Health { calls++; return &protocol.Health{AgentVersion: "0.1.0"} }})
	b.Add(meas(1)...)
	b.Flush(context.Background())
	b.Add(meas(1)...)
	b.Flush(context.Background())
	if calls != 1 || p.batches[0].Health == nil || p.batches[1].Health != nil {
		t.Fatalf("health should be attached once: calls=%d", calls)
	}
	// Empty queue and health not due → no POST at all.
	n := p.calls
	b.Flush(context.Background())
	if p.calls != n {
		t.Fatal("empty flush must not POST")
	}
}

func TestSpoolOnFailureAndDrain(t *testing.T) {
	p := &fakePoster{}
	failures := 0
	b, sp := newBatcher(t, p, Hooks{OnFlushFailure: func() { failures++ }})
	p.setFail(&client.HTTPError{Status: 503, Body: "down"})
	b.Add(meas(10)...)
	b.Flush(context.Background())
	if n, _ := sp.Stats(); n != 1 || failures != 1 || !b.InBackoff() {
		t.Fatalf("expected 1 spooled file + backoff, got %d files, failures=%d", n, failures)
	}
	// During backoff new batches go straight to the spool without touching the network.
	calls := p.calls
	b.Add(meas(5)...)
	b.Flush(context.Background())
	if p.calls != calls {
		t.Fatal("must not POST during backoff")
	}
	if n, _ := sp.Stats(); n != 2 {
		t.Fatalf("expected 2 spooled files, got %d", n)
	}
	// Server back: force backoff expiry, drain oldest first, then live.
	p.setFail(nil)
	b.nextTry = time.Now().Add(-time.Second)
	b.Add(meas(3)...)
	b.Flush(context.Background())
	if n, _ := sp.Stats(); n != 0 {
		t.Fatalf("spool should be drained, %d left", n)
	}
	if p.count() != 3 || len(p.batches[0].Measurements) != 10 || len(p.batches[1].Measurements) != 5 || len(p.batches[2].Measurements) != 3 {
		t.Fatalf("delivery order wrong: %d batches", p.count())
	}
	if b.InBackoff() {
		t.Fatal("backoff should be cleared after success")
	}
}

func TestBackoffGrowsAndCaps(t *testing.T) {
	b, _ := newBatcher(t, &fakePoster{}, Hooks{})
	want := []time.Duration{5 * time.Second, 10 * time.Second, 20 * time.Second, 40 * time.Second, 80 * time.Second, 160 * time.Second, 300 * time.Second, 300 * time.Second}
	for i, w := range want {
		b.fail(errors.New("x"))
		if b.backoff != w {
			t.Fatalf("step %d: backoff %v want %v", i, b.backoff, w)
		}
	}
}

func TestAuthFailureDrops(t *testing.T) {
	p := &fakePoster{}
	b, sp := newBatcher(t, p, Hooks{})
	p.setFail(&client.HTTPError{Status: 401, Body: `{"detail":"bad signature"}`})
	b.Add(meas(2)...)
	b.Flush(context.Background())
	if n, _ := sp.Stats(); n != 0 || b.InBackoff() || b.Buffered() != 0 {
		t.Fatalf("401 must drop without spool/backoff: files=%d backoff=%v buffered=%d", n, b.InBackoff(), b.Buffered())
	}
	// 401 skew → retried once.
	p.setFail(&client.HTTPError{Status: 401, Body: `{"detail":"timestamp skew"}`})
	calls := p.calls
	b.Add(meas(2)...)
	b.Flush(context.Background())
	if p.calls-calls != 2 {
		t.Fatalf("skew should retry exactly once, got %d calls", p.calls-calls)
	}
}

func TestStopSpoolsRemainder(t *testing.T) {
	p := &fakePoster{}
	b, sp := newBatcher(t, p, Hooks{})
	b.Configure(10, 5)
	p.setFail(errors.New("dial tcp: connection refused"))
	b.Add(meas(12)...)
	b.Stop()
	if b.Buffered() != 0 {
		t.Fatalf("queue not emptied on stop: %d", b.Buffered())
	}
	if n, _ := sp.Stats(); n != 3 {
		t.Fatalf("expected 3 spooled files on stop, got %d", n)
	}
	if p.calls != 1 {
		t.Fatalf("stop should try the network once, tried %d", p.calls)
	}
}

func TestMaxBatchKicks(t *testing.T) {
	p := &fakePoster{}
	b, _ := newBatcher(t, p, Hooks{})
	b.Configure(3600, 4)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go b.Run(ctx)
	b.Add(meas(4)...)
	deadline := time.Now().Add(2 * time.Second)
	for p.count() == 0 && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	if p.count() != 1 {
		t.Fatal("reaching max_batch should flush immediately")
	}
}