package activator

import (
	"context"
	"errors"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

// blockedOnGen is an OnGen that blocks on events of kind until release
// closes, as a scheduler whose failure transition is slow would, and marks
// recorded once that event's handling has returned. Only the first such
// event is the one a test waits on.
type blockedOnGen struct {
	kind     GenEventKind
	once     sync.Once
	entered  chan struct{}
	release  chan struct{}
	recorded atomic.Bool
	evs      genLog
}

func newBlockedOnGen(kind GenEventKind) *blockedOnGen {
	return &blockedOnGen{kind: kind, entered: make(chan struct{}), release: make(chan struct{})}
}

func (o *blockedOnGen) on(ev GenEvent) {
	o.evs.add(ev)
	if ev.Kind != o.kind {
		return
	}
	o.once.Do(func() { close(o.entered) })
	<-o.release
	o.recorded.Store(true)
}

func (o *blockedOnGen) waitEntered(t *testing.T) {
	t.Helper()
	select {
	case <-o.entered:
	case <-time.After(5 * time.Second):
		t.Fatalf("OnGen never received %s; saw %v", o.kind, o.evs.kinds())
	}
}

// secondPlanner answers the first plan with first, and records, for every
// later plan, whether the blocked outcome had been recorded when it was
// asked. A later plan refuses once it has, as the scheduler's planner does
// for a held-down build.
func secondPlanner(o *blockedOnGen, first WakePlan, saw chan<- bool) func(context.Context, registry.Route) WakePlan {
	var n atomic.Int32
	return func(context.Context, registry.Route) WakePlan {
		if n.Add(1) == 1 {
			return first
		}
		rec := o.recorded.Load()
		saw <- rec
		if rec {
			return WakePlan{Refuse: "held down"}
		}
		return first
	}
}

// Final review F3 (spec §8.1, O7): a failed first wake's outcome is
// delivered before admission reopens. A request that arrives after the
// failed wake's entry is gone, while OnGen is still recording the failure,
// starts a wake that waits for that delivery and only then plans, with the
// record in place. Before the fence the second wake planned B again in the
// gap, before the hold-down existed.
func TestANewWakeWaitsForAFailedFirstWakesOutcome(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	next := stagedID(t, dir, "v2")
	a, b, _ := planActivator(t, WakePlan{})
	const host = "one.example.com"
	a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
	o := newBlockedOnGen(GenFailed)
	a.OnGen = o.on
	saw := make(chan bool, 4)
	a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true}, saw)
	if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
		t.Fatal("the failed hold first wake served")
	}
	o.waitEntered(t)
	if a.Owns(host) {
		t.Fatal("fixture: the failed wake's entry is still there")
	}
	done := make(chan error, 1)
	go func() {
		_, err := a.beginRequest(context.Background(), testRoute(host))
		done <- err
	}()
	select {
	case rec := <-saw:
		close(o.release)
		t.Fatalf("the second wake planned while the first wake's failure was still being delivered (record in place: %v)", rec)
	case <-time.After(300 * time.Millisecond):
	}
	close(o.release)
	select {
	case rec := <-saw:
		if !rec {
			t.Fatal("the second wake planned before the failure was recorded")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the second wake never planned")
	}
	if err := <-done; err == nil || !strings.Contains(err.Error(), "held down") {
		t.Fatalf("second wake = %v, want refused by the recorded hold-down", err)
	}
	if gens := b.started(); len(gens) != 1 {
		t.Fatalf("spawned %+v: B ran again before its failure was recorded", gens)
	}
}

// Final review wave 2 (N1): the fence fails closed. The wait is bounded
// (outcomeFenceWait), the restriction is not. A wake still behind an
// undelivered outcome when the bound passes is refused before it plans:
// no Planner call, nothing launched, and the member reads the maintenance
// fence's existing line. The fence stays, so the next request is refused
// the same way. Once the outcome is delivered a wake proceeds to plan
// (here the planner, with the record in place, refuses the held build).
// Before the fix the wake planned after the bound and could boot B again
// without its record.
func TestTheOutcomeFenceFailsClosed(t *testing.T) {
	old := outcomeFenceWait
	outcomeFenceWait = 200 * time.Millisecond
	t.Cleanup(func() { outcomeFenceWait = old })
	dir := t.TempDir()
	next := stagedID(t, dir, "v2")
	a, b, _ := planActivator(t, WakePlan{})
	const host = "one.example.com"
	a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") }
	o := newBlockedOnGen(GenFailed)
	var releaseOnce sync.Once
	release := func() { releaseOnce.Do(func() { close(o.release) }) }
	t.Cleanup(release)
	a.OnGen = o.on
	sink := &lockedSink{}
	a.Logs = sink
	saw := make(chan bool, 4)
	a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true}, saw)
	if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil {
		t.Fatal("the failed hold first wake served")
	}
	o.waitEntered(t)
	for i := 1; i <= 2; i++ {
		start := time.Now()
		_, err := a.beginRequest(context.Background(), testRoute(host))
		if !errors.Is(err, errOutcomeUndelivered) {
			t.Fatalf("wake %d behind the undelivered outcome = %v, want refused (errOutcomeUndelivered)", i, err)
		}
		if took := time.Since(start); took < 150*time.Millisecond {
			t.Fatalf("wake %d refused after %v: it did not wait for the fence", i, took)
		}
		select {
		case rec := <-saw:
			t.Fatalf("wake %d planned behind the undelivered outcome (record in place: %v)", i, rec)
		default:
		}
		if gens := b.started(); len(gens) != 1 {
			t.Fatalf("spawned %+v: a wake behind the undelivered outcome launched something", gens)
		}
	}
	want := memberSafeWakeReason(ErrFenced)
	got := sink.errors()
	if len(got) != 3 || got[1] != want || got[2] != want {
		t.Fatalf("member error events %q: want the failed wake's, then the maintenance fence's line (%q) for each refusal", got, want)
	}
	release()
	_, err := a.beginRequest(context.Background(), testRoute(host))
	select {
	case rec := <-saw:
		if !rec {
			t.Fatal("the wake after the delivery planned before the failure was recorded")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the wake after the delivery never planned: the fence outlived the outcome")
	}
	if err == nil || !strings.Contains(err.Error(), "held down") {
		t.Fatalf("wake after the delivery = %v, want refused by the recorded hold-down", err)
	}
	if gens := b.started(); len(gens) != 1 {
		t.Fatalf("spawned %+v: B ran again", gens)
	}
}

// Final review F3, the observer's half: a first wake's 10 s observation
// delivers its terminal event (here GenFailed, an exit inside the window)
// BEFORE it clears firstWatch, and holds the host's outcome fence until
// the delivery returns. While OnGen records the failure, Current still
// hides the generation, and a request that stops the dead entry and wakes
// it again waits for the record before planning. Before the fix the
// observer cleared firstWatch first and nothing held the next wake back.
func TestAFirstWakeObserverDeliversItsOutcomeBeforeAdmissionReopens(t *testing.T) {
	shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second)
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, _ := planActivator(t, WakePlan{})
	const host = "one.example.com"
	o := newBlockedOnGen(GenFailed)
	a.OnGen = o.on
	saw := make(chan bool, 4)
	a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}, saw)
	done, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatal(err)
	}
	done()
	o.evs.wait(t, GenHealthy)
	b.procOf(b.started()[0].N).crash()
	o.waitEntered(t)
	if _, ok := a.Current(host); ok {
		close(o.release)
		t.Fatal("Current showed the first wake while its failure was still being delivered: firstWatch was cleared first")
	}
	second := make(chan error, 1)
	go func() {
		_, err := a.beginRequest(context.Background(), testRoute(host))
		second <- err
	}()
	select {
	case rec := <-saw:
		close(o.release)
		t.Fatalf("the next wake planned while the observer's failure was still being delivered (record in place: %v)", rec)
	case <-time.After(300 * time.Millisecond):
	}
	close(o.release)
	select {
	case rec := <-saw:
		if !rec {
			t.Fatal("the next wake planned before the failure was recorded")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the next wake never planned")
	}
	if err := <-second; err == nil || !strings.Contains(err.Error(), "held down") {
		t.Fatalf("next wake = %v, want refused by the recorded failure", err)
	}
	if k := o.evs.kinds(); !slices.Equal(k, []GenEventKind{GenHealthy, GenFailed}) {
		t.Fatalf("events %v, want healthy then failed", k)
	}
}