package host

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

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

func logshipFixture(t *testing.T) (release.Store, *applog.Sink) {
	t.Helper()
	s := release.NewDirStore(t.TempDir())
	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
	return s, applog.NewSink("box1", func() time.Time { return now })
}

// TestLogshipFlushWritesOneChunkPerApp: a drain with two apps lands as two
// immutable chunks, each under its own app prefix, decodable and complete.
func TestLogshipFlushWritesOneChunkPerApp(t *testing.T) {
	s, sink := logshipFixture(t)
	sink.App("bde", "ledgerpay", "ledgerpay.bde.oncarlos.com", "hello")
	sink.Event("bab", "kass", "kass.example.com", applog.LevelInfo, "instance woke")

	pending := make(map[applog.AppKey][]applog.Record)
	logshipGather(sink, pending)
	logshipFlush(context.Background(), s, pending)
	if len(pending) != 0 {
		t.Fatalf("flushed apps still pending: %v", pending)
	}

	for _, app := range []applog.AppKey{{Acct: "bde", App: "ledgerpay"}, {Acct: "bab", App: "kass"}} {
		keys, err := s.List(context.Background(), mustKey(t)(applog.Prefix(app.Acct, app.App)))
		if err != nil || len(keys) != 1 {
			t.Fatalf("chunks for %v: %v (%v)", app, keys, err)
		}
		body, _, err := s.Get(context.Background(), keys[0])
		if err != nil {
			t.Fatal(err)
		}
		recs, err := applog.DecodeChunk(body)
		if err != nil || len(recs) != 1 {
			t.Fatalf("chunk decode for %v: %d recs, %v", app, len(recs), err)
		}
		// The key carries the real byte size — retention counts with it.
		if n, ok := applog.ChunkKeySize(keys[0]); !ok || n != int64(len(body)) {
			t.Fatalf("key size %d/%v, body %d: %s", n, ok, len(body), keys[0])
		}
	}
}

// TestLogshipFlushKeepsRecordsWhenTheBucketRefuses: a failed Put must hold
// the app's records for the next round, not lose them — and the pending cap
// bounds that holding.
func TestLogshipFlushKeepsRecordsWhenTheBucketRefuses(t *testing.T) {
	_, sink := logshipFixture(t)
	sink.App("bde", "ledgerpay", "h", "precious")
	pending := make(map[applog.AppKey][]applog.Record)
	logshipGather(sink, pending)
	logshipFlush(context.Background(), failingStore{}, pending)
	k := applog.AppKey{Acct: "bde", App: "ledgerpay"}
	if len(pending[k]) != 1 || pending[k][0].Msg != "precious" {
		t.Fatalf("records lost on failed flush: %v", pending)
	}
	// Cap: a long outage trims oldest, keeps the freshest tail.
	for i := 0; i < logshipMaxPending+10; i++ {
		sink.App("bde", "ledgerpay", "h", "line")
	}
	logshipGather(sink, pending)
	if len(pending[k]) != logshipMaxPending {
		t.Fatalf("pending cap not enforced: %d", len(pending[k]))
	}
	if pending[k][0].Msg == "precious" {
		t.Fatal("cap kept oldest instead of freshest")
	}
}

// TestLogshipPrune: hours past the age floor go; then oldest-first until
// the key-embedded sizes fit the byte cap; the box only touches its own
// subtree.
func TestLogshipPrune(t *testing.T) {
	t.Parallel()
	s := release.NewDirStore(t.TempDir())
	ctx := context.Background()
	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
	k := applog.AppKey{Acct: "bde", App: "ledgerpay"}

	put := func(box, hour string, seq uint64, size int) string {
		key := mustKey(t)(applog.ChunkKey(k.Acct, k.App, box, hour, seq, size))
		if _, err := s.Put(ctx, key, make([]byte, size), ""); err != nil {
			t.Fatal(err)
		}
		return key
	}
	oldKey := put("box1", "2026-08-01-10", 1, 10)     // over the 7d floor
	bigKey := put("box1", "2026-08-10-10", 2, 60<<20) // pushes total over the byte cap
	liveKey := put("box1", "2026-08-10-11", 3, 10)
	otherBox := put("box2", "2026-08-01-10", 4, 10) // NOT ours to prune

	chunkPrune(ctx, s, applog.FamilyLogs, "box1", k, now)

	for key, want := range map[string]bool{oldKey: false, bigKey: false, liveKey: true, otherBox: true} {
		_, _, err := s.Get(ctx, key)
		if got := err == nil; got != want {
			t.Errorf("%s survived=%v, want %v", key, got, want)
		}
	}
}

// failingStore refuses writes — the bucket-outage fixture.
type failingStore struct{ release.Store }

func (failingStore) Put(context.Context, string, []byte, string) (string, error) {
	return "", context.DeadlineExceeded
}

// TestLogshipPruneKeepsRecentUnderCap: nothing under both limits is touched.
func TestLogshipPruneKeepsRecentUnderCap(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	ctx := context.Background()
	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
	k := applog.AppKey{Acct: "bde", App: "ledgerpay"}
	key := mustKey(t)(applog.ChunkKey(k.Acct, k.App, "box1", "2026-08-10-11", 1, 10))
	if _, err := s.Put(ctx, key, make([]byte, 10), ""); err != nil {
		t.Fatal(err)
	}
	chunkPrune(ctx, s, applog.FamilyLogs, "box1", k, now)
	if _, _, err := s.Get(ctx, key); err != nil {
		t.Fatalf("live chunk pruned: %v", err)
	}
	if !strings.HasPrefix(key, "apps/bde/ledgerpay/logs/") {
		t.Fatalf("key layout drifted: %s", key)
	}
}

// must unwraps a key builder's (key, error) for fixtures whose names are
// known good: mustKey(t)(ChunkKey(...)).
func mustKey(t testing.TB) func(string, error) string {
	return func(k string, err error) string {
		t.Helper()
		if err != nil {
			t.Fatal(err)
		}
		return k
	}
}