package schedule

import (
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

// gateFence is a WriteFence over a retired flag, serialized against
// retire() the way the activator's is against its switch. inside is true
// while a write runs in it.
type gateFence struct {
	mu      sync.RWMutex
	retired bool
	inside  atomic.Bool
}

func (f *gateFence) fence(write func()) bool {
	f.mu.RLock()
	defer f.mu.RUnlock()
	if f.retired {
		return false
	}
	f.inside.Store(true)
	defer f.inside.Store(false)
	write()
	return true
}

func (f *gateFence) retire() {
	f.mu.Lock()
	f.retired = true
	f.mu.Unlock()
}

// A retiring generation's writes are refused with 409 once the switch has
// happened (zero-downtime spec §7.10). The successor's writes apply. Reads
// stay open, and the bearer is still checked first.
func TestMountGatedRefusesARetiredGenerationsWrites(t *testing.T) {
	reg, _ := controlFixture(t)
	f := &gateFence{}
	mux := http.NewServeMux()
	MountGated(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil },
		func() time.Time { return at("2026-08-23T13:07:00Z") }, f.fence)
	body := `{"at":"2026-09-01T08:00:00Z","path":"/jobs/remind"}`
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", body); rec.Code != http.StatusOK {
		t.Fatalf("current generation's write = %d %s, want 200", rec.Code, rec.Body)
	}
	f.retire()
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", body); rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), RetiredWriteMessage) {
		t.Fatalf("retired PUT = %d %q, want 409 with RetiredWriteMessage", rec.Code, rec.Body)
	}
	if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusConflict {
		t.Fatalf("retired DELETE = %d, want 409", rec.Code)
	}
	if row, err := reg.GetTimer("a.bab.oncarlos.com", "remind"); err != nil || row == nil {
		t.Fatalf("a refused DELETE removed the timer: %v", err)
	}
	if rec := do(mux, "GET", "/schedule", "Bearer tok", ""); rec.Code != http.StatusOK {
		t.Fatalf("retired GET = %d, want 200", rec.Code)
	}
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer wrong", body); rec.Code != http.StatusUnauthorized {
		t.Fatalf("a retired generation's bad bearer = %d, want 401", rec.Code)
	}
	if rec := do(mux, "DELETE", "/schedule/remind", "Bearer wrong", ""); rec.Code != http.StatusUnauthorized {
		t.Fatalf("a retired generation's bad bearer on DELETE = %d, want 401", rec.Code)
	}
}

// A refused PUT changes nothing: the row the successor wrote survives a
// retiring generation's attempt to overwrite it.
func TestMountGatedRefusedPutLeavesTheRowAlone(t *testing.T) {
	reg, _ := controlFixture(t)
	f := &gateFence{}
	mux := http.NewServeMux()
	MountGated(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil },
		func() time.Time { return at("2026-08-23T13:07:00Z") }, f.fence)
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/jobs/new"}`); rec.Code != http.StatusOK {
		t.Fatalf("PUT = %d %s", rec.Code, rec.Body)
	}
	f.retire()
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-02T08:00:00Z","path":"/jobs/old"}`); rec.Code != http.StatusConflict {
		t.Fatalf("retired PUT = %d, want 409", rec.Code)
	}
	row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
	if err != nil || row.Path != "/jobs/new" {
		t.Fatalf("a refused PUT changed the row: %+v, %v", row, err)
	}
}

// A nil gate is Mount exactly: both writing verbs apply.
func TestMountGatedNilIsMount(t *testing.T) {
	reg, _ := controlFixture(t)
	mux := http.NewServeMux()
	MountGated(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil },
		func() time.Time { return at("2026-08-23T13:07:00Z") }, nil)
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); rec.Code != http.StatusOK {
		t.Fatalf("PUT = %d", rec.Code)
	}
	if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
		t.Fatalf("DELETE = %d, want 204", rec.Code)
	}
	if _, err := reg.GetTimer("a.bab.oncarlos.com", "remind"); err == nil {
		t.Fatal("a nil gate's DELETE left the timer in place")
	}
}

// stallBody is a request body whose first Read signals entered and then
// blocks until release closes: a client stalled mid-body.
type stallBody struct {
	r       *strings.Reader
	entered chan struct{}
	release chan struct{}
	first   atomic.Bool
}

func (b *stallBody) Read(p []byte) (int, error) {
	if b.first.CompareAndSwap(false, true) {
		close(b.entered)
		<-b.release
	}
	return b.r.Read(p)
}
func (b *stallBody) Close() error { return nil }

// Final review F8: a retiring generation's PUT that began before the
// switch and stalled in its body cannot overwrite the successor's write.
// The retirement decision is made inside the fence, after the body, and
// the switch is serialized against it.
func TestRetiredPutStalledInItsBodyCannotOverwriteTheSuccessor(t *testing.T) {
	reg, _ := controlFixture(t)
	now := func() time.Time { return at("2026-08-23T13:07:00Z") }
	tok := func() (string, error) { return "tok", nil }
	f := &gateFence{}
	oldMux, newMux := http.NewServeMux(), http.NewServeMux()
	MountGated(oldMux, "a.bab.oncarlos.com", reg, tok, now, f.fence)
	MountGated(newMux, "a.bab.oncarlos.com", reg, tok, now, (&gateFence{}).fence)
	body := &stallBody{r: strings.NewReader(`{"at":"2026-09-02T08:00:00Z","path":"/jobs/old"}`), entered: make(chan struct{}), release: make(chan struct{})}
	req := httptest.NewRequest("PUT", "/schedule/remind", nil)
	req.Body = body
	req.Header.Set("Authorization", "Bearer tok")
	rec := httptest.NewRecorder()
	done := make(chan struct{})
	go func() { oldMux.ServeHTTP(rec, req); close(done) }()
	<-body.entered
	f.retire() // the switch: it does not wait on a request still reading its body
	if r := do(newMux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/jobs/new"}`); r.Code != http.StatusOK {
		t.Fatalf("successor PUT = %d %s", r.Code, r.Body)
	}
	close(body.release)
	<-done
	if rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), RetiredWriteMessage) {
		t.Errorf("the retired generation's stalled PUT = %d %q, want 409", rec.Code, rec.Body)
	}
	if row, err := reg.GetTimer("a.bab.oncarlos.com", "remind"); err != nil || row.Path != "/jobs/new" {
		t.Fatalf("the successor's timer was overwritten: %+v, %v", row, err)
	}
}

// fenceCheckStore fails the test when a writing verb touches the store
// outside its generation's fence.
type fenceCheckStore struct {
	TimerStore
	t *testing.T
	f *gateFence
}

func (s fenceCheckStore) in(op string) {
	if !s.f.inside.Load() {
		s.t.Errorf("%s ran outside the write fence", op)
	}
}
func (s fenceCheckStore) UpsertTimer(t registry.Timer) error {
	s.in("UpsertTimer")
	return s.TimerStore.UpsertTimer(t)
}
func (s fenceCheckStore) DeleteTimer(h, n string) error {
	s.in("DeleteTimer")
	return s.TimerStore.DeleteTimer(h, n)
}
func (s fenceCheckStore) GetTimer(h, n string) (*registry.Timer, error) {
	s.in("GetTimer")
	return s.TimerStore.GetTimer(h, n)
}
func (s fenceCheckStore) CountTimers(h, k string) (int, error) {
	s.in("CountTimers")
	return s.TimerStore.CountTimers(h, k)
}

// Final review F8: PUT's and DELETE's store checks and mutations all run
// inside the fence, so the final retirement decision and the write are one
// step against the switch (DELETE's check/use race included).
func TestMountGatedWritesOnlyInsideTheFence(t *testing.T) {
	reg, _ := controlFixture(t)
	f := &gateFence{}
	mux := http.NewServeMux()
	MountGated(mux, "a.bab.oncarlos.com", fenceCheckStore{TimerStore: reg, t: t, f: f}, func() (string, error) { return "tok", nil },
		func() time.Time { return at("2026-08-23T13:07:00Z") }, f.fence)
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); rec.Code != http.StatusOK {
		t.Fatalf("PUT = %d %s", rec.Code, rec.Body)
	}
	if rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"2026-09-02T08:00:00Z","path":"/r2"}`); rec.Code != http.StatusOK {
		t.Fatalf("second PUT = %d %s", rec.Code, rec.Body)
	}
	if rec := do(mux, "DELETE", "/schedule/remind", "Bearer tok", ""); rec.Code != http.StatusNoContent {
		t.Fatalf("DELETE = %d %s", rec.Code, rec.Body)
	}
}