package activator

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"
)

// restoreInFlightForTest reports whether host's entry has a parent restore
// retry in flight.
func (a *Service) restoreInFlightForTest(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	return e != nil && e.ops.restoring
}

// owedRestore leaves f's entry with a parent restore owed: an overlap whose
// last member's restore failed (F11), so the parent stays marked raised.
func owedRestore(t *testing.T, f *swapFixture) {
	t.Helper()
	f.b.gmu.Lock()
	f.b.restoreFails = 1
	f.b.gmu.Unlock()
	if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	if !f.a.parentRaisedForTest(f.host) {
		t.Fatal("fixture: the failed restore should leave the parent marked raised")
	}
}

// gateRestores holds every later RestoreGenParent before its write and
// returns the gate and the channel each held call signals on.
func gateRestores(f *swapFixture) (gate, entered chan struct{}) {
	gate, entered = make(chan struct{}), make(chan struct{}, 8)
	f.b.gmu.Lock()
	f.b.restoreGate, f.b.restoreEntered = gate, entered
	f.b.gmu.Unlock()
	return gate, entered
}

// joinSignal installs the barrier's join hook and returns the channel it
// signals on when a stop waits on the operation set.
func joinSignal(f *swapFixture) chan struct{} {
	joining := make(chan struct{}, 1)
	f.a.mu.Lock()
	f.a.joinWaiting = func(_, what string) {
		if what == "operation set" {
			select {
			case joining <- struct{}{}:
			default:
			}
		}
	}
	f.a.mu.Unlock()
	return joining
}

// Final review R2-2: a parent restore retry is a lifecycle operation of its
// entry. A terminal stop joins one in flight before its lease release; a
// stopping entry launches none; and a hold's stop joins it before its
// pinned replacement is installed, so the old retry can never write the
// old entry's single-generation tier after a later overlap raised the
// host's parent for two children.
func TestAParentRestoreRetryIsJoinedByTheEntrysStop(t *testing.T) {
	t.Parallel()
	t.Run("a terminal stop joins the retry in flight", func(t *testing.T) {
		f := newSwapFixture(t)
		owedRestore(t, f)
		gate, entered := gateRestores(f)
		f.a.sweepOnce()
		<-entered // the retry is past its launch, before its write
		joining := joinSignal(f)
		rep := f.b.lastRep()
		f.a.mu.Lock()
		f.a.beginStopLocked(f.a.entries[f.host], StopIdle)
		f.a.mu.Unlock()
		waitFor(t, "the stop to join the retry or finish", func() bool {
			select {
			case <-joining:
				return true
			default:
			}
			return !f.a.entryPresent(f.host)
		})
		if !f.a.entryPresent(f.host) || rep.isClosed() {
			t.Errorf("the stop finished (entry present %v, lease released %v) while the parent restore retry was in flight", f.a.entryPresent(f.host), rep.isClosed())
		}
		close(gate)
		waitFor(t, "the stop to finish once the retry is done", func() bool { return !f.a.entryPresent(f.host) && rep.isClosed() })
	})

	t.Run("a stopping entry launches no retry", func(t *testing.T) {
		f := newSwapFixture(t)
		owedRestore(t, f)
		gate, _ := gateRestores(f)
		defer close(gate)
		stopGate := make(chan struct{})
		f.a.mu.Lock()
		e := f.a.entries[f.host]
		cur := e.proc.(*fakeProc)
		f.a.mu.Unlock()
		cur.setStopGate(stopGate)
		f.a.mu.Lock()
		f.a.beginStopLocked(e, StopIdle)
		f.a.mu.Unlock()
		// Past the barrier, inside the current generation's Stop.
		waitFor(t, "the stop to reach the current generation", func() bool { return cur.stopEntered() > 0 })
		f.a.sweepOnce()
		if f.a.restoreInFlightForTest(f.host) {
			t.Error("the sweep launched a parent restore retry for a stopping entry")
		}
		close(stopGate)
		waitFor(t, "the stop to finish", func() bool { return !f.a.entryPresent(f.host) })
	})

	t.Run("a hold, its replacement and a later overlap", func(t *testing.T) {
		f := newSwapFixture(t)
		owedRestore(t, f)
		gate, entered := gateRestores(f)
		f.a.sweepOnce()
		<-entered
		joining := joinSignal(f)
		f.a.mu.Lock()
		oldE := f.a.entries[f.host]
		f.a.mu.Unlock()
		f.rev.Add(1)
		results := make(chan SwapResult, 1)
		go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) }()
		joined := false
		waitFor(t, "the hold's stop to join the retry or install its replacement", func() bool {
			select {
			case <-joining:
				joined = true
				return true
			default:
			}
			f.a.mu.Lock()
			defer f.a.mu.Unlock()
			return f.a.entries[f.host] != oldE
		})
		var once sync.Once
		release := func() { once.Do(func() { close(gate) }) }
		defer release()
		if !joined {
			t.Error("the hold's stop installed its pinned replacement while the old entry's parent restore retry was in flight")
		} else {
			release()
		}
		select {
		case res := <-results:
			if res.Outcome != SwapDone {
				t.Fatalf("hold Swap = %+v", res)
			}
		case <-time.After(10 * time.Second):
			t.Fatal("the hold never finished")
		}
		waitFor(t, "the replacement to be awake", func() bool { return f.a.entryAwake(f.host) })
		// A later overlap on the replacement raises the host's parent for two
		// children. An old retry still held lands now, if it can.
		_, _, before := f.b.calls()
		f.b.gmu.Lock()
		f.b.onRaise = func() {
			if joined {
				return
			}
			release()
			deadline := time.Now().Add(5 * time.Second)
			for time.Now().Before(deadline) {
				if _, _, n := f.b.calls(); n > before {
					return
				}
				time.Sleep(time.Millisecond)
			}
		}
		f.b.gmu.Unlock()
		f.rev.Add(1)
		if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone {
			t.Fatalf("later overlap Swap = %+v", res)
		}
		ev := f.b.events()
		raise := -1
		for i, x := range ev {
			if strings.HasPrefix(x, "raise:") {
				raise = i
			}
		}
		for _, x := range ev[raise+1:] {
			if strings.HasPrefix(x, "switch:") {
				break
			}
			if strings.HasPrefix(x, "restore:") {
				t.Fatalf("a parent restore (%s) was written after the later overlap raised the parent and before its switch, while two generations ran: %v", x, ev[raise:])
			}
		}
	})
}