package hostops

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
	"testing"

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

type plan15StreamStore struct {
	release.Store
	bytes, closes, opens, puts int
}

func (s *plan15StreamStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.Contains(key, "control/host-origins/") {
		panic("unbounded origin Get")
	}
	return s.Store.Get(ctx, strings.TrimPrefix(key, "deployment/"))
}
func (s *plan15StreamStore) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
	s.puts++
	return "", release.ErrCASConflict
}
func (s *plan15StreamStore) Open(context.Context, string) (io.ReadCloser, string, error) {
	s.opens++
	return plan15StreamReader{s, strings.NewReader(strings.Repeat("x", 8192))}, "", nil
}

type plan15StreamReader struct {
	s *plan15StreamStore
	r io.Reader
}

func (r plan15StreamReader) Read(b []byte) (int, error) {
	n, e := r.r.Read(b)
	r.s.bytes += n
	return n, e
}
func (r plan15StreamReader) Close() error { r.s.closes++; return nil }

type plan15NoOpen struct{ release.Store }

func TestPlan15StreamingCompositions(t *testing.T) {
	for _, stream := range []bool{false, true} {
		for _, boundary := range []bool{false, true} {
			t.Run(fmt.Sprintf("stream=%v/boundary=%v", stream, boundary), func(t *testing.T) {
				w, o, _ := wakeResetFixture(t)
				o.Claim.Origins = &hostclaim.OriginHead{Digest: strings.Repeat("a", 64), Count: 1}
				key, _ := hostclaim.Key(o.Host)
				body, _ := json.Marshal(o.Claim)
				var err error
				o.ETag, err = w.raw.Put(t.Context(), key, body, o.ETag)
				if err != nil {
					t.Fatal(err)
				}
				raw := &plan15StreamStore{Store: w.raw}
				var s release.Store = raw
				if !stream {
					s = plan15NoOpen{s}
				}
				s = release.NewPrefixStore(s, "deployment")
				s = conditionalRecordStore{raw: s}
				if boundary {
					s = &mutationBoundary{raw: s, check: func(context.Context, MutationTarget) error { return nil }}
				}
				h := o.Claim.Holder
				if ok, err := hostclaim.AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "real"); ok || !errors.Is(err, hostclaim.ErrMalformed) {
					t.Fatalf("proof %v %v", ok, err)
				}
				if stream && (raw.bytes != 4097 || raw.opens != 1 || raw.closes != 1) {
					t.Fatalf("bytes=%d opens=%d closes=%d", raw.bytes, raw.opens, raw.closes)
				}
				if !stream && (raw.bytes != 0 || raw.opens != 0) {
					t.Fatal("nonstreaming store opened")
				}
			})
		}
	}
}
func TestPlan15BoundaryRefusesCandidateBeforeRead(t *testing.T) {
	for _, flag := range []string{"", "1"} {
		t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", flag)
		raw := &plan15StreamStore{}
		s := &mutationBoundary{raw: conditionalRecordStore{raw: release.NewPrefixStore(raw, "deployment")}, check: func(context.Context, MutationTarget) error { t.Fatal("origin reached grant checker"); return nil }}
		if _, err := s.Put(t.Context(), "control/host-origins/x.example/node.json", []byte("candidate"), ""); !errors.Is(err, ErrUnclassified) {
			t.Fatal(err)
		}
		if raw.puts != 0 || raw.opens != 0 {
			t.Fatal("candidate touched raw store")
		}
	}
}