package provision

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

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

// overlapStore holds each record Get open briefly and records the most Gets
// it saw in flight at once.
type overlapStore struct {
	release.Store
	mu        sync.Mutex
	inFlight  int
	maxFlight int
	reads     int
	failKey   string
}

func (s *overlapStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.Contains(key, "/instances/") {
		s.mu.Lock()
		s.inFlight++
		s.reads++
		if s.inFlight > s.maxFlight {
			s.maxFlight = s.inFlight
		}
		s.mu.Unlock()
		time.Sleep(20 * time.Millisecond)
		s.mu.Lock()
		s.inFlight--
		s.mu.Unlock()
		if key == s.failKey {
			return nil, "", errors.New("read refused")
		}
	}
	return s.Store.Get(ctx, key)
}

func seedRecords(t *testing.T, n int) (release.Store, string) {
	t.Helper()
	inner, sqid := setup(t)
	for i := 0; i < n; i++ {
		rec := acmeRecord(sqid)
		slug := "r" + string(rune('a'+i))
		rec.Host, rec.Slug = slug+".go.tito.io", slug
		if err := CreateRecord(ctx, inner, rec); err != nil {
			t.Fatal(err)
		}
	}
	return inner, sqid
}

// The record readers read records several at a time. One at a time, a few
// hundred records cost the fast loop most of a 27-second pass on every
// serial bump (#231).
func TestRecordReadersReadRecordsConcurrently(t *testing.T) {
	inner, _ := seedRecords(t, 12)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	for _, c := range []struct {
		name string
		run  func(release.Store) error
	}{
		{"Reconcile", func(s release.Store) error { return Reconcile(ctx, s, regWith(t), cfg) }},
		{"ListAllRecords", func(s release.Store) error { _, err := ListAllRecords(ctx, s); return err }},
	} {
		s := &overlapStore{Store: inner}
		if err := c.run(s); err != nil {
			t.Fatalf("%s: %v", c.name, err)
		}
		if s.maxFlight < 2 || s.maxFlight > 8 {
			t.Fatalf("%s read up to %d records at a time, want several and never more than 8", c.name, s.maxFlight)
		}
		if s.reads != 12 {
			t.Fatalf("%s made %d record reads for 12 records, want one each", c.name, s.reads)
		}
	}
}

// A read that fails is still that record's failure, judged where it always
// was: reconcile keeps the routes it could not read, and mints the rest.
func TestReconcileStillJudgesAFailedReadPerRecord(t *testing.T) {
	inner, sqid := seedRecords(t, 4)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	reg := regWith(t)
	if err := Reconcile(ctx, inner, reg, cfg); err != nil {
		t.Fatal(err)
	}
	s := &overlapStore{Store: inner, failKey: RecordKey(sqid, "tito", "rb.go.tito.io")}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	// A record created since the first pass must still be minted in a pass
	// where another record's read fails: the failure is rb's alone.
	fresh := acmeRecord(sqid)
	fresh.Host, fresh.Slug = "re.go.tito.io", "re"
	if err := CreateRecord(ctx, inner, fresh); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	for _, h := range []string{"ra.go.tito.io", "rb.go.tito.io", "rc.go.tito.io", "rd.go.tito.io", "re.go.tito.io"} {
		if rt, err := reg.Get(h); err != nil || rt == nil {
			t.Fatalf("%s missing after a pass where only rb's read failed: %+v, %v", h, rt, err)
		}
	}
	_, err := ListAllRecords(ctx, s)
	if err == nil || !strings.Contains(err.Error(), "rb.go.tito.io") {
		t.Fatalf("ListAllRecords with rb unreadable = %v, want an error naming rb", err)
	}
}

// A loop that stops at its first bad read has read at most one chunk ahead,
// and a large record set is never prefetched whole.
func TestListAllRecordsStopsWithinAChunkOfABadRead(t *testing.T) {
	inner, sqid := setup(t)
	const total = 200 // more than one chunk, fixed so the test does not move with readChunk
	for i := 0; i < total; i++ {
		rec := acmeRecord(sqid)
		rec.Host, rec.Slug = fmt.Sprintf("r%04d.go.tito.io", i), fmt.Sprintf("r%04d", i)
		if err := CreateRecord(ctx, inner, rec); err != nil {
			t.Fatal(err)
		}
	}
	s := &countingGets{Store: inner, failKey: RecordKey(sqid, "tito", "r0000.go.tito.io")}
	if _, err := ListAllRecords(ctx, s); err == nil {
		t.Fatal("ListAllRecords read past an unreadable first record without an error")
	}
	if s.n > readChunk || s.n >= total {
		t.Fatalf("read %d of %d records before stopping at the first, want at most one chunk (%d)", s.n, total, readChunk)
	}
}

// A cancelled context reads nothing, and every key reports why.
func TestGetEachOnACancelledContextReadsNothing(t *testing.T) {
	inner, sqid := setup(t)
	s := &countingGets{Store: inner}
	cctx, cancel := context.WithCancel(ctx)
	cancel()
	key := RecordKey(sqid, "tito", "acme.go.tito.io")
	got := getEach(cctx, s, []string{key})
	if s.n != 0 {
		t.Fatalf("made %d reads on a cancelled context", s.n)
	}
	if !errors.Is(got[key].err, context.Canceled) {
		t.Fatalf("key reported %v, want context.Canceled", got[key].err)
	}
}

type countingGets struct {
	release.Store
	mu      sync.Mutex
	n       int
	failKey string
}

func (s *countingGets) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.Contains(key, "/instances/") {
		s.mu.Lock()
		s.n++
		s.mu.Unlock()
		if key == s.failKey {
			return nil, "", errors.New("read refused")
		}
	}
	return s.Store.Get(ctx, key)
}

// appOverlapStore holds each app-level read (an app's instances manifest and
// its trash marker) open briefly and records the most in flight at once.
type appOverlapStore struct {
	release.Store
	mu        sync.Mutex
	inFlight  int
	maxFlight int
	reads     map[string]int
}

func (s *appOverlapStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.HasSuffix(key, "/instances.json") || strings.HasSuffix(key, "/deleted.json") {
		s.mu.Lock()
		s.inFlight++
		if s.inFlight > s.maxFlight {
			s.maxFlight = s.inFlight
		}
		if s.reads == nil {
			s.reads = map[string]int{}
		}
		s.reads[key]++
		s.mu.Unlock()
		time.Sleep(20 * time.Millisecond)
		s.mu.Lock()
		s.inFlight--
		s.mu.Unlock()
	}
	return s.Store.Get(ctx, key)
}

// Reconcile reads each app's manifest and trash marker several apps at a
// time, and each one once.
func TestReconcileReadsAppStateConcurrentlyAndOnce(t *testing.T) {
	inner, sqid := setup(t)
	want := map[string]bool{}
	for i := 0; i < 10; i++ {
		app := "app" + string(rune('a'+i))
		want[release.AppPrefix(sqid, app)+"instances.json"] = true
		want[release.AppTombstoneKey(sqid, app)] = true
		if err := PutManifest(ctx, inner, sqid, app, Manifest{Schema: 1, App: app, Domain: app + ".example", Pool: "tito"}); err != nil {
			t.Fatal(err)
		}
		rec := acmeRecord(sqid)
		rec.App, rec.Host, rec.Slug = app, "x."+app+".example", "x"
		if err := CreateRecord(ctx, inner, rec); err != nil {
			t.Fatal(err)
		}
	}
	s := &appOverlapStore{Store: inner}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, regWith(t), cfg); err != nil {
		t.Fatal(err)
	}
	if s.maxFlight < 2 || s.maxFlight > 8 {
		t.Fatalf("app state read up to %d at a time, want several and never more than 8", s.maxFlight)
	}
	for key := range want {
		if s.reads[key] != 1 {
			t.Fatalf("%s read %d times in one pass, want once", key, s.reads[key])
		}
	}
}

// An app in the trash is never minted for, so its manifest is not read: the
// order the record loop always asked in, trash first.
func TestReconcileDoesNotReadATrashedAppsManifest(t *testing.T) {
	inner, sqid := setup(t)
	tomb := fmt.Sprintf(`{"schema":1,"account":%q,"app":"tito","deleted_at":"2026-10-05T00:00:00Z","actor":"test"}`, sqid)
	if _, err := inner.Put(ctx, release.AppTombstoneKey(sqid, "tito"), []byte(tomb), ""); err != nil {
		t.Fatal(err)
	}
	if got, err := release.ReadAppTombstone(ctx, inner, sqid, "tito"); err != nil || got == nil {
		t.Fatalf("fixture: the trash marker does not read as one: %+v, %v", got, err)
	}
	if err := CreateRecord(ctx, inner, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	s := &appOverlapStore{Store: inner}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, regWith(t), cfg); err != nil {
		t.Fatal(err)
	}
	if n := s.reads[release.AppPrefix(sqid, "tito")+"instances.json"]; n != 0 {
		t.Fatalf("a trashed app's manifest was read %d times", n)
	}
}

// Every chunk is walked: 130 records, three chunks, each record minted and
// listed, each read once.
func TestRecordReadersWalkEveryChunk(t *testing.T) {
	inner, sqid := setup(t)
	const total = 130
	hosts := map[string]bool{}
	for i := 0; i < total; i++ {
		rec := acmeRecord(sqid)
		rec.Host, rec.Slug = fmt.Sprintf("r%04d.go.tito.io", i), fmt.Sprintf("r%04d", i)
		hosts[rec.Host] = true
		if err := CreateRecord(ctx, inner, rec); err != nil {
			t.Fatal(err)
		}
	}
	s := &perKeyGets{Store: inner}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	for h := range hosts {
		if rt, err := reg.Get(h); err != nil || rt == nil {
			t.Fatalf("%s not minted: %+v, %v", h, rt, err)
		}
		if n := s.count(RecordKey(sqid, "tito", h)); n != 1 {
			t.Fatalf("%s read %d times by Reconcile, want once", h, n)
		}
	}
	recs, err := ListAllRecords(ctx, s)
	if err != nil {
		t.Fatal(err)
	}
	if len(recs) != total {
		t.Fatalf("ListAllRecords = %d records, want %d", len(recs), total)
	}
}

type perKeyGets struct {
	release.Store
	mu sync.Mutex
	n  map[string]int
}

func (s *perKeyGets) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.mu.Lock()
	if s.n == nil {
		s.n = map[string]int{}
	}
	s.n[key]++
	s.mu.Unlock()
	return s.Store.Get(ctx, key)
}

func (s *perKeyGets) count(key string) int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.n[key]
}