package dbreads

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

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

// The identity check (spec §8.2): the record as the envelope named it, or a
// code. T9's stages are Task 11's; this is the check itself.
func TestCheckIsTheRecordAsTheEnvelopeNamedIt(t *testing.T) {
	t.Parallel()
	no := false
	for name, c := range map[string]struct {
		change func(fx *fixture)
		want   string
	}{
		"unchanged":               {func(*fixture) {}, ""},
		"moved to another region": {func(fx *fixture) { r := fx.record(); r.Region = "ie"; fx.putRecord(r) }, CodeRecordChanged},
		"moving": {func(fx *fixture) {
			r := fx.record()
			r.Move = &provision.MoveState{ID: "mv_x", To: "ie", State: "requested"}
			fx.putRecord(r)
		}, CodeMoving},
		"recreated in the account": {func(fx *fixture) { r := fx.record(); r.CreatedAt = r.CreatedAt.Add(time.Hour); fx.putRecord(r) }, CodeRecordChanged},
		"recreated in another account": {func(fx *fixture) {
			created := fx.record().CreatedAt
			fx.store.Delete(context.Background(), provision.RecordKey(fxAccount, fxApp, fxHost))
			fx.putRecordUnder("bac", provision.Record{Region: "uk", CreatedAt: created})
		}, CodeRecordChanged},
		"no longer sleeps":  {func(fx *fixture) { r := fx.record(); r.Hibernate = &no; fx.putRecord(r) }, CodeRecordChanged},
		"pooled":            {func(fx *fixture) { r := fx.record(); r.Pool = "shared"; fx.putRecord(r) }, CodeRecordChanged},
		"placed on a fleet": {func(fx *fixture) { r := fx.record(); r.Region = "fleet/acme"; fx.putRecord(r) }, CodeRecordChanged},
	} {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			fx := newFixture(t, 0)
			req := fx.request("SELECT 1", "")
			c.change(fx)
			a, _ := fx.agent("uk-1")
			_, code, err := a.check(context.Background(), req)
			if err != nil || code != c.want {
				t.Fatalf("check = %q, %v; want %q", code, err, c.want)
			}
		})
	}
}

// T5: two capable boxes in one region; exactly one owns each read, across
// 1,000 reads with random jitter.
func TestElectionOneOwnerAcrossTwoAgents(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 0)
	a, _ := fx.agent("uk-1")
	b, _ := fx.agent("uk-2")
	var mu sync.Mutex
	owners := map[string]int{}
	var wg sync.WaitGroup
	for i := 0; i < 1000; i++ {
		id, err := MintID()
		if err != nil {
			t.Fatal(err)
		}
		for _, ag := range []*Agent{a, b} {
			wg.Add(1)
			go func(ag *Agent) {
				defer wg.Done()
				time.Sleep(rand.N(2 * time.Millisecond))
				if _, ok := ag.elect(context.Background(), id); ok {
					mu.Lock()
					owners[id]++
					mu.Unlock()
				}
			}(ag)
		}
	}
	wg.Wait()
	if len(owners) != 1000 {
		t.Fatalf("%d reads have an owner, want 1000", len(owners))
	}
	for id, n := range owners {
		if n != 1 {
			t.Fatalf("%s has %d owners", id, n)
		}
	}
}

// lossyStore commits some Puts and then answers an error: a response lost
// on the way back (T5, codex 16).
type lossyStore struct {
	release.Store
	lose   func(key string) bool
	refuse func(key string) bool
}

func (s lossyStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if s.refuse != nil && s.refuse(key) {
		return "", errors.New("connection reset before the write")
	}
	etag, err := s.Store.Put(ctx, key, body, ifMatch)
	if err == nil && s.lose != nil && s.lose(key) {
		return "", errors.New("connection reset after the write")
	}
	return etag, err
}

// A box whose picked create commits but answers an error reads it back and
// owns the read.
func TestElectionReadsBackALostAnswer(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 0)
	a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
		c.Store = lossyStore{Store: fx.store, lose: func(string) bool { return true }}
	})
	if _, ok := a.elect(context.Background(), "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"); !ok {
		t.Fatal("a lost answer to a create that landed was not owned")
	}
}

// A box whose create answers an error while another box owns the read steps
// away; so does one whose read-back fails.
func TestElectionStepsAwayFromAnotherOwner(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 0)
	a, _ := fx.agent("uk-1")
	id := "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"
	if _, ok := a.elect(context.Background(), id); !ok {
		t.Fatal("the first create did not own")
	}
	b, _ := fx.agent("uk-2", func(c *AgentConfig, _ *inProcWorkers) {
		c.Store = lossyStore{Store: fx.store, refuse: func(string) bool { return true }}
	})
	if _, ok := b.elect(context.Background(), id); ok {
		t.Fatal("a box owned a read another box had picked")
	}
	c, _ := fx.agent("uk-3", func(cfg *AgentConfig, _ *inProcWorkers) {
		cfg.Store = failingGets{lossyStore{Store: fx.store, refuse: func(string) bool { return true }}}
	})
	if _, ok := c.elect(context.Background(), "dbr_bbbbbbbbbbbbbbbbbbbbbbbbbb"); ok {
		t.Fatal("a box owned a read it could not read back")
	}
}

type failingGets struct{ release.Store }

func (failingGets) Get(context.Context, string) ([]byte, string, error) {
	return nil, "", errors.New("timeout")
}