package hostclaim

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"testing"

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

type plan15BoundedStore struct {
	*release.DirStore
	oversized, conflict       bool
	read, closed, opens, gets int
}

func (s *plan15BoundedStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.Contains(key, "control/host-origins/") {
		panic("unbounded node Get")
	}
	s.gets++
	return s.DirStore.Get(ctx, key)
}
func (s *plan15BoundedStore) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
	if s.conflict && strings.Contains(key, "control/host-origins/") {
		return "", release.ErrCASConflict
	}
	return s.DirStore.Put(ctx, key, b, tag)
}

type plan15Reader struct {
	io.Reader
	store *plan15BoundedStore
}

func (r plan15Reader) Read(b []byte) (int, error) {
	n, e := r.Reader.Read(b)
	r.store.read += n
	return n, e
}
func (r plan15Reader) Close() error { r.store.closed++; return nil }
func (s *plan15BoundedStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
	s.opens++
	if s.oversized {
		return plan15Reader{strings.NewReader(strings.Repeat("x", 1<<20)), s}, "oversized", nil
	}
	return s.DirStore.Open(ctx, key)
}
func TestPlan15BoundedReachableAndConflicting(t *testing.T) {
	for _, prefixed := range []bool{false, true} {
		for _, conflict := range []bool{false, true} {
			t.Run(fmt.Sprintf("prefix=%v/conflict=%v", prefixed, conflict), func(t *testing.T) {
				plan15Enable(t)
				counter := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
				var s release.Store = counter
				if prefixed {
					s = release.NewPrefixStore(release.NewPrefixStore(s, "outer"), "inner")
				}
				o := servingClaim(t, t.Context(), s)
				if !conflict {
					o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "real", "ie"))
				}
				counter.oversized, counter.conflict = true, conflict
				counter.opens, counter.closed, counter.read = 0, 0, 0
				var err error
				if conflict {
					_, err = Update(t.Context(), s, o, plan15Admit(o, "candidate", "ie"))
				} else {
					h := o.Claim.Holder
					_, err = AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "real")
				}
				if !errors.Is(err, ErrMalformed) || counter.read != 4097 || counter.closed != 1 || counter.opens != 1 {
					t.Fatalf("err=%v bytes=%d closes=%d opens=%d", err, counter.read, counter.closed, counter.opens)
				}
			})
		}
	}
}
func TestPlan15CacheBindings(t *testing.T) {
	plan15Enable(t)
	s := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
	o := servingClaim(t, t.Context(), s)
	o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "first", "ie"))
	h := o.Claim.Holder
	first := OriginTuple{h.Incarnation, h.Location, "ie", "first"}
	cache := NewOriginProofCache(s)
	prove := func(obs Observed, set []OriginTuple, want bool) {
		t.Helper()
		result, err := cache.Proven(t.Context(), obs, set)
		if err != nil || result[set[0]] != want {
			t.Fatalf("%v %v", result, err)
		}
	}
	s.opens, s.gets = 0, 0
	prove(o, []OriginTuple{first, first}, true)
	prove(o, []OriginTuple{first}, true)
	if s.opens != int(o.Claim.Origins.Count) || s.gets != 2 {
		t.Fatalf("cache opens=%d reads=%d", s.opens, s.gets)
	}
	different := first
	different.Region = "uk"
	prove(o, []OriginTuple{different}, false)
	if s.opens != 2*int(o.Claim.Origins.Count) {
		t.Fatal("tuple set reused")
	}
	other := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
	key, _ := Key(o.Host)
	body, _, err := s.Get(t.Context(), key)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := other.Put(t.Context(), key, body, ""); err != nil {
		t.Fatal(err)
	}
	if got, err := NewOriginProofCache(other).Proven(t.Context(), o, []OriginTuple{different}); got != nil || err == nil {
		t.Fatal("foreign store reused")
	}
	prove(o, []OriginTuple{first}, true)
	edited := o
	c := *o.Claim
	edited.Claim = &c
	c.Origins = &OriginHead{Digest: strings.Repeat("f", 64), Count: o.Claim.Origins.Count}
	if got, err := cache.Proven(t.Context(), edited, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
		t.Fatal("edited head reused")
	}
	prove(o, []OriginTuple{first}, true)
	newer := updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "second", "ie"))
	if got, err := cache.Proven(t.Context(), o, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
		t.Fatal("stale observation reused")
	}
	s.opens = 0
	prove(newer, []OriginTuple{first}, true)
	if s.opens != int(newer.Claim.Origins.Count) {
		t.Fatal("changed head did not traverse")
	}
	// No partial positives escape a failed batch, including a corrupt tail.
	if err := s.Delete(t.Context(), originKey(o.Host, o.Claim.Origins.Digest)); err != nil {
		t.Fatal(err)
	}
	second := first
	second.Generation = "second"
	if got, err := cache.Proven(t.Context(), newer, []OriginTuple{first, second}); got != nil || err == nil {
		t.Fatal("partial proof escaped")
	}
}