package applog

import (
	"strings"
	"testing"
	"time"
)

func testClock(start time.Time) (func() time.Time, func(d time.Duration)) {
	now := start
	return func() time.Time { return now }, func(d time.Duration) { now = now.Add(d) }
}

func TestSinkStampsAndSequences(t *testing.T) {
	now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
	sk := NewSink("box1", now)
	sk.App("bde", "ledgerpay", "ledgerpay.bde.oncarlos.com", "hello")
	sk.Event("bde", "ledgerpay", "ledgerpay.bde.oncarlos.com", LevelError, "wake failed: seed timed out")

	got := sk.Drain()
	if len(got) != 1 {
		t.Fatalf("apps drained: %d, want 1", len(got))
	}
	recs := got[AppKey{"bde", "ledgerpay"}]
	if len(recs) != 2 {
		t.Fatalf("records: %d, want 2", len(recs))
	}
	if recs[0].Box != "box1" || recs[0].Stream != StreamApp || recs[0].TS.IsZero() || recs[0].Host != "ledgerpay.bde.oncarlos.com" {
		t.Fatalf("app record not stamped: %+v", recs[0])
	}
	if recs[1].Stream != StreamPlatform || recs[1].Level != LevelError {
		t.Fatalf("event record wrong: %+v", recs[1])
	}
	if recs[1].Seq != recs[0].Seq+1 {
		t.Fatalf("seq not monotonic: %d then %d", recs[0].Seq, recs[1].Seq)
	}
	// Drain resets; seq does not.
	sk.App("bde", "ledgerpay", "h", "again")
	recs2 := sk.Drain()[AppKey{"bde", "ledgerpay"}]
	if len(recs2) != 1 || recs2[0].Seq != recs[1].Seq+1 {
		t.Fatalf("after drain: %+v", recs2)
	}
}

// TestSinkAppLinesCarryParsedJSON: the sink is where ParseAppLine is
// applied, so a structured app line arrives in the bucket with fields and
// an honored level rather than as opaque text.
func TestSinkAppLinesCarryParsedJSON(t *testing.T) {
	now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
	sk := NewSink("box1", now)
	sk.App("bde", "ledgerpay", "h", `{"level":"warn","msg":"slow","request_id":"r1"}`)
	rec := sk.Drain()[AppKey{"bde", "ledgerpay"}][0]
	if rec.Msg != "slow" || rec.Level != LevelWarn || rec.Fields["request_id"] != "r1" {
		t.Fatalf("structured line not parsed at the sink: %+v", rec)
	}
}

// TestSinkRingCapDropsOldestAndSaysSo: over the per-app byte cap the oldest
// lines go first, and the drain carries one synthetic platform record
// naming the count — never silent thinning (spec §4).
func TestSinkRingCapDropsOldestAndSaysSo(t *testing.T) {
	now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
	sk := NewSink("box1", now)
	sk.MaxBytesPerApp = 600 // tiny for the test
	sk.Burst = 1 << 20      // rate cap out of the way
	for i := 0; i < 50; i++ {
		sk.App("bde", "ledgerpay", "h", strings.Repeat("x", 40))
	}
	recs := sk.Drain()[AppKey{"bde", "ledgerpay"}]
	if len(recs) >= 50 {
		t.Fatalf("nothing evicted: %d records", len(recs))
	}
	last := recs[len(recs)-1]
	if last.Stream != StreamPlatform || last.Level != LevelWarn ||
		!strings.Contains(last.Msg, "dropped") {
		t.Fatalf("no honest drop record; last = %+v", last)
	}
	// The synthetic record is accounted once per drain, not once per drop.
	for _, r := range recs[:len(recs)-1] {
		if strings.Contains(r.Msg, "dropped") {
			t.Fatalf("drop record duplicated mid-stream: %+v", r)
		}
	}
}

// TestSinkRateCapRefills: the token bucket drops lines while exhausted and
// admits again after time passes; drops surface in the same honest record.
func TestSinkRateCapRefills(t *testing.T) {
	now, advance := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
	sk := NewSink("box1", now)
	sk.RatePerMin = 300
	sk.Burst = 300
	sk.App("bde", "ledgerpay", "h", strings.Repeat("a", 200)) // fits burst
	sk.App("bde", "ledgerpay", "h", strings.Repeat("b", 200)) // bucket empty → dropped
	recs := sk.Drain()[AppKey{"bde", "ledgerpay"}]
	if len(recs) != 2 { // the admitted line + the drop notice
		t.Fatalf("got %d records: %+v", len(recs), recs)
	}
	if !strings.Contains(recs[1].Msg, "dropped 1") {
		t.Fatalf("drop notice wrong: %+v", recs[1])
	}
	advance(time.Minute)
	sk.App("bde", "ledgerpay", "h", strings.Repeat("c", 200))
	if got := sk.Drain()[AppKey{"bde", "ledgerpay"}]; len(got) != 1 || !strings.HasPrefix(got[0].Msg, "ccc") {
		t.Fatalf("bucket did not refill: %+v", got)
	}
}

// TestSinkEventsBypassTheRateCap: platform events are the diagnosis — a
// chatty app must not be able to crowd out its own wake-failure line.
func TestSinkEventsBypassTheRateCap(t *testing.T) {
	now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
	sk := NewSink("box1", now)
	sk.RatePerMin = 10
	sk.Burst = 10
	sk.App("bde", "ledgerpay", "h", strings.Repeat("a", 400)) // exhausts the bucket
	sk.Event("bde", "ledgerpay", "h", LevelError, "wake failed")
	recs := sk.Drain()[AppKey{"bde", "ledgerpay"}]
	var sawEvent bool
	for _, r := range recs {
		if r.Stream == StreamPlatform && r.Msg == "wake failed" {
			sawEvent = true
		}
	}
	if !sawEvent {
		t.Fatalf("event lost to the app's own rate cap: %+v", recs)
	}
}

func TestSinkDrainEmptyIsEmpty(t *testing.T) {
	now, _ := testClock(time.Now())
	sk := NewSink("box1", now)
	if got := sk.Drain(); len(got) != 0 {
		t.Fatalf("empty sink drained %v", got)
	}
}

// TestIsDropNotice: the rate-cap line Drain appends is recognised, and only
// it — Availability leaves it out by this test, and a platform event that
// merely starts with "dropped" must not vanish with it.
func TestIsDropNotice(t *testing.T) {
	now, _ := testClock(time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC))
	sk := NewSink("box1", now)
	sk.RatePerMin, sk.Burst = 300, 300
	sk.App("bab", "web", "h", strings.Repeat("a", 200)) // fits burst
	sk.App("bab", "web", "h", strings.Repeat("b", 200)) // bucket empty → dropped
	var notice *Record
	for _, r := range sk.Drain()[AppKey{"bab", "web"}] {
		if r.Stream == StreamPlatform {
			notice = &r
		}
	}
	if notice == nil || !IsDropNotice(*notice) {
		t.Fatalf("drain's rate-cap line not recognised: %+v", notice)
	}
	for _, r := range []Record{
		{Stream: StreamPlatform, Level: LevelWarn, Msg: "dropped connection to the database"},
		{Stream: StreamApp, Msg: notice.Msg},
	} {
		if IsDropNotice(r) {
			t.Errorf("IsDropNotice(%+v) = true", r)
		}
	}
}