package applog

import (
	"context"
	"errors"
	"fmt"
	"math/rand/v2"
	"slices"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

// seedWindow writes chunks for two boxes across three hours.
func seedWindow(t *testing.T, s release.Store) {
	t.Helper()
	ctx := context.Background()
	write := func(box, hour string, seq uint64, recs []Record) {
		t.Helper()
		body, err := EncodeChunk(recs)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := s.Put(ctx, must(t)(ChunkKey("bde", "ledgerpay", box, hour, seq, len(body))), body, ""); err != nil {
			t.Fatal(err)
		}
	}
	ts := func(h, m int) time.Time { return time.Date(2026, 8, 10, h, m, 0, 0, time.UTC) }
	write("box1", "2026-08-10-10", 100, []Record{
		{TS: ts(10, 5), Box: "box1", Seq: 100, Stream: StreamApp, Host: "h1", Msg: "old line"},
	})
	write("box1", "2026-08-10-11", 200, []Record{
		{TS: ts(11, 10), Box: "box1", Seq: 200, Stream: StreamApp, Host: "h1", Msg: "b1 app"},
		{TS: ts(11, 30), Box: "box1", Seq: 201, Stream: StreamPlatform, Level: LevelError, Host: "h1", Msg: "wake failed: x"},
	})
	write("box2", "2026-08-10-11", 500, []Record{
		{TS: ts(11, 20), Box: "box2", Seq: 500, Stream: StreamApp, Host: "h2", Msg: "b2 app"},
	})
}

func TestReadWindowMergesAndClipsHours(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	seedWindow(t, s)
	ctx := context.Background()

	// Window covering hour 11 only: the hour-10 chunk is never even listed.
	f := Filter{Since: time.Date(2026, 8, 10, 11, 0, 0, 0, time.UTC)}
	recs, trimmed, err := ReadWindow(ctx, s, "bde", "ledgerpay", f, 1000)
	if err != nil || trimmed {
		t.Fatalf("read: %v trimmed=%v", err, trimmed)
	}
	want := []string{"b1 app", "b2 app", "wake failed: x"} // (ts, box, seq) order
	if len(recs) != len(want) {
		t.Fatalf("got %d records: %+v", len(recs), recs)
	}
	for i, m := range want {
		if recs[i].Msg != m {
			t.Fatalf("order[%d] = %q, want %q", i, recs[i].Msg, m)
		}
	}

	// Stream filter composes.
	recs, _, err = ReadWindow(ctx, s, "bde", "ledgerpay", Filter{Stream: StreamPlatform}, 1000)
	if err != nil || len(recs) != 1 || recs[0].Level != LevelError {
		t.Fatalf("platform filter: %v %+v", err, recs)
	}

	// Unbounded window sees all three hours.
	recs, _, err = ReadWindow(ctx, s, "bde", "ledgerpay", Filter{}, 1000)
	if err != nil || len(recs) != 4 {
		t.Fatalf("unbounded: %v, %d records", err, len(recs))
	}

	// collectCap keeps the TAIL and says it trimmed.
	recs, trimmed, err = ReadWindow(ctx, s, "bde", "ledgerpay", Filter{}, 2)
	if err != nil || !trimmed || len(recs) != 2 {
		t.Fatalf("capped: %v trimmed=%v n=%d", err, trimmed, len(recs))
	}
	if recs[len(recs)-1].Msg != "wake failed: x" {
		t.Fatalf("cap dropped the tail: %+v", recs)
	}

	// An app with no logs at all is an empty answer, not an error.
	recs, _, err = ReadWindow(ctx, s, "bde", "ghost", Filter{}, 1000)
	if err != nil || len(recs) != 0 {
		t.Fatalf("empty app: %v %+v", err, recs)
	}
}

func TestHourIntersects(t *testing.T) {
	since := time.Date(2026, 8, 10, 11, 0, 0, 0, time.UTC)
	for hour, want := range map[string]bool{
		"2026-08-10-10": false, // ends exactly at since — nothing at/after since lives there
		"2026-08-10-11": true,
		"2026-08-10-12": true,
		"not-an-hour":   false,
	} {
		if got := hourIntersects(hour, since, time.Time{}); got != want {
			t.Errorf("hourIntersects(%q, since) = %v, want %v", hour, got, want)
		}
	}
	until := time.Date(2026, 8, 10, 11, 30, 0, 0, time.UTC)
	if hourIntersects("2026-08-10-12", time.Time{}, until) {
		t.Error("an hour starting after until intersects")
	}
	if !hourIntersects("2026-08-10-11", since, until) {
		t.Error("the window's own hour should intersect")
	}
}

func TestReadChunksWalksAnyFamily(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	type ev struct {
		TS   time.Time `json:"ts"`
		Name string    `json:"name"`
	}
	now := time.Date(2026, 9, 7, 10, 30, 0, 0, time.UTC)
	body, _ := EncodeJSONL([]ev{{TS: now, Name: "boom"}, {TS: now.Add(time.Minute), Name: "skip"}})
	key := must(t)(FamilyChunkKey(FamilyErrors, "bab", "web", "box1", HourOf(now), 1, len(body)))
	if _, err := s.Put(ctx, key, body, ""); err != nil {
		t.Fatal(err)
	}
	perBox, trimmed, err := ReadChunks[ev](ctx, s, must(t)(FamilyPrefix(FamilyErrors, "bab", "web")),
		now.Add(-time.Hour), now.Add(time.Hour), func(e ev) bool { return e.Name != "skip" }, 100)
	if err != nil || trimmed {
		t.Fatalf("ReadChunks: %v trimmed=%v", err, trimmed)
	}
	if len(perBox) != 1 || len(perBox[0]) != 1 || perBox[0][0].Name != "boom" {
		t.Fatalf("perBox = %+v", perBox)
	}
	// An hour outside the window is never listed.
	perBox, _, err = ReadChunks[ev](ctx, s, must(t)(FamilyPrefix(FamilyErrors, "bab", "web")),
		now.Add(2*time.Hour), time.Time{}, func(ev) bool { return true }, 100)
	if err != nil || len(perBox) != 1 || len(perBox[0]) != 0 {
		t.Fatalf("out-of-window read = %+v, %v", perBox, err)
	}
}

// slowStore delays every Get by a random few milliseconds, so fetches
// finish out of order, counts how many are in flight at once, and can fail
// one key.
type slowStore struct {
	release.Store
	mu       sync.Mutex
	inFlight int
	peak     int
	fail     string
}

func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.mu.Lock()
	s.inFlight++
	if s.inFlight > s.peak {
		s.peak = s.inFlight
	}
	s.mu.Unlock()
	defer func() { s.mu.Lock(); s.inFlight--; s.mu.Unlock() }()
	time.Sleep(time.Duration(rand.IntN(4)) * time.Millisecond)
	if key == s.fail {
		return nil, "", errors.New("bucket had a bad day")
	}
	return s.Store.Get(ctx, key)
}

// TestReadChunksConcurrentIsSerialOrder: fetched in parallel and finishing
// in any order, the answer is still exactly the serial walk's — each box's
// records in hour then seq order, the cap keeping the tail — with no more
// than readConcurrency GETs in flight, and one failed chunk fails the read.
func TestReadChunksConcurrentIsSerialOrder(t *testing.T) {
	ctx := context.Background()
	dir := release.NewDirStore(t.TempDir())
	base := time.Date(2026, 9, 24, 0, 0, 0, 0, time.UTC)
	want := map[string][]string{}
	var someKey string
	for _, box := range []string{"box1", "box2", "box3"} {
		for h := 0; h < 5; h++ {
			hour := base.Add(time.Duration(h) * time.Hour)
			for c := 0; c < 8; c++ {
				seq := uint64(1000 + h*100 + c)
				msg := fmt.Sprintf("%s h%d c%d", box, h, c)
				body, err := EncodeChunk([]Record{{TS: hour.Add(time.Duration(c) * time.Minute), Box: box, Seq: seq, Msg: msg}})
				if err != nil {
					t.Fatal(err)
				}
				key := must(t)(ChunkKey("bde", "ledgerpay", box, HourOf(hour), seq, len(body)))
				if _, err := dir.Put(ctx, key, body, ""); err != nil {
					t.Fatal(err)
				}
				want[box] = append(want[box], msg)
				someKey = key
			}
		}
	}
	s := &slowStore{Store: dir}
	all := func(Record) bool { return true }
	perBox, trimmed, err := ReadChunks[Record](ctx, s, must(t)(Prefix("bde", "ledgerpay")), time.Time{}, time.Time{}, all, 1000)
	if err != nil || trimmed || len(perBox) != 3 {
		t.Fatalf("read: %v trimmed=%v boxes=%d", err, trimmed, len(perBox))
	}
	for i, box := range []string{"box1", "box2", "box3"} {
		var got []string
		for _, r := range perBox[i] {
			got = append(got, r.Msg)
		}
		if !slices.Equal(got, want[box]) {
			t.Fatalf("%s order:\n got %v\nwant %v", box, got, want[box])
		}
	}
	if s.peak < 2 || s.peak > readConcurrency {
		t.Fatalf("peak in-flight GETs = %d, want 2..%d", s.peak, readConcurrency)
	}

	perBox, trimmed, err = ReadChunks[Record](ctx, s, must(t)(Prefix("bde", "ledgerpay")), time.Time{}, time.Time{}, all, 5)
	if err != nil || !trimmed || len(perBox[0]) != 5 || perBox[0][4].Msg != "box1 h4 c7" || perBox[0][0].Msg != "box1 h4 c3" {
		t.Fatalf("capped read kept %+v (err %v, trimmed %v)", perBox[0], err, trimmed)
	}

	s.fail = someKey
	if _, _, err := ReadChunks[Record](ctx, s, must(t)(Prefix("bde", "ledgerpay")), time.Time{}, time.Time{}, all, 1000); err == nil {
		t.Fatal("a failed chunk must fail the read, never return a partial answer")
	}
}