package activator

import (
	"context"
	"errors"
	"testing"
	"time"

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

// These tests enter sweepOnce with the cut fence still installed. Calling
// reviewRecoveryHold directly would miss the fenced-host early continue.
func sweptProbe(t *testing.T) (*Service, *ClaimRuntime, *actEntry, *fakeProc, *fakeRep, hostclaim.Observed) {
	t.Helper()
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	admitted := probeAdmitted(obs)
	a, r, _ := probeService(t, obs)
	a.fences = map[string]uint64{probeHost: 9}
	e, proc, rep, err := heldProbe(t, a, 9, admitted, probeInputs())
	if err != nil {
		t.Fatal(err)
	}
	e.recoveryPolicy.RecoveryFence = 9
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
	return a, r, e, proc, rep, admitted
}

func waitProbe(t *testing.T, a *Service, predicate func() bool) {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for time.Now().Before(deadline) {
		a.mu.Lock()
		ok := predicate()
		a.mu.Unlock()
		if ok {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatal("probe supervision did not finish")
}

func TestRecoveryProbeSweepRetiresFailedBehindCutFence(t *testing.T) {
	t.Parallel()
	for _, mode := range []string{"exit", "lease", "join-exit", "join-lease"} {
		t.Run(mode, func(t *testing.T) {
			a, r, e, proc, rep, admitted := sweptProbe(t)
			if mode == "exit" || mode == "join-exit" {
				proc.exit()
			} else {
				rep.setErr(errors.New("lease lost"))
			}
			if mode == "join-exit" || mode == "join-lease" {
				if err := a.joinRecoveryProbe(t.Context(), e, *e.recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) {
					t.Fatalf("join: %v", err)
				}
			} else {
				a.sweepOnce()
			}
			waitProbe(t, a, func() bool { return a.entries[probeHost] == nil })
			if !proc.isStopped() || !rep.isClosed() || a.FenceGeneration(probeHost) != 9 {
				t.Fatal("failed probe was not retired with its cut fence retained")
			}
			// A retry reaches the admission/start path again, rather than joining
			// the dead entry forever. The fixture launcher deliberately refuses.
			next := probeAdmitted(admitted)
			next.Claim.Holder.Generation = "three"
			reads := 0
			r.Read = func(context.Context, string) (hostclaim.Observed, error) {
				reads++
				if reads == 1 {
					return admitted, nil
				}
				return next, nil
			}
			r.Admit = func(context.Context, registry.Route, hostclaim.Observed, StartPolicy) (hostclaim.Observed, error) {
				return next, nil
			}
			if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); !errors.Is(err, ErrClaimIsolationUnavailable) {
				t.Fatalf("retry did not reach the launcher: %v", err)
			}
		})
	}
}

func TestRecoveryProbeSweepOpensOnlyExactSessionAndFence(t *testing.T) {
	t.Parallel()
	for _, phase := range []string{hostclaim.PhaseProbing, hostclaim.PhaseProbed, hostclaim.PhaseOpened, "serving", "new-session", "foreign-executor", "unreadable"} {
		t.Run(phase, func(t *testing.T) {
			a, r, e, proc, _, admitted := sweptProbe(t)
			// The cut fence carries an owner (FenceAndStopAs at the same
			// generation): the opening's lift clears it with the fence.
			a.mu.Lock()
			a.fenceOwners = map[string]string{probeHost: "owner-1"}
			a.mu.Unlock()
			fresh := admitted
			c, h := *admitted.Claim, *admitted.Claim.Holder
			c.Holder = &h
			fresh.Claim = &c
			c.Op.Phase = phase
			switch phase {
			case "serving":
				c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
			case "new-session":
				c.Op.Phase = hostclaim.PhaseOpened
				h.Session++
			case "foreign-executor":
				c.Op.Phase = hostclaim.PhaseOpened
				c.Op.Exec = "other"
			}
			read := make(chan struct{}, 1)
			r.Read = func(context.Context, string) (hostclaim.Observed, error) {
				read <- struct{}{}
				if phase == "unreadable" {
					return hostclaim.Observed{}, errors.New("unreadable")
				}
				return fresh, nil
			}
			a.sweepOnce()
			select {
			case <-read:
			case <-time.After(time.Second):
				t.Fatal("sweep skipped held probe behind its cut fence")
			}
			waitProbe(t, a, func() bool { return !e.probeReviewing })
			opens := phase == hostclaim.PhaseOpened || phase == "serving"
			stops := phase == "new-session" || phase == "foreign-executor"
			if stops {
				waitProbe(t, a, func() bool { return a.entries[probeHost] == nil })
			}
			if opens {
				if a.Fenced(probeHost) || proc.isStopped() {
					t.Fatal("exact opening did not release the cut fence and retain its process")
				}
				a.mu.Lock()
				_, owned := a.fenceOwners[probeHost]
				a.mu.Unlock()
				if owned {
					t.Fatal("the opening lifted the cut fence but left its owner")
				}
				release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost})
				if err != nil {
					t.Fatalf("opened session cannot serve: %v", err)
				}
				release()
			} else {
				if a.FenceGeneration(probeHost) != 9 {
					t.Fatal("non-opening changed fence")
				}
				if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
					t.Fatalf("request: %v", err)
				}
				if !stops && proc.isStopped() {
					t.Fatal("a held or unreadable session was stopped")
				}
			}
		})
	}
}

func TestRecoveryProbeSweepPreservesReplacementFence(t *testing.T) {
	t.Parallel()
	for _, afterRead := range []bool{false, true} {
		t.Run(map[bool]string{false: "before-read", true: "during-read"}[afterRead], func(t *testing.T) {
			a, r, e, proc, rep, admitted := sweptProbe(t)
			read, resume := make(chan struct{}), make(chan struct{})
			r.Read = func(context.Context, string) (hostclaim.Observed, error) {
				close(read)
				<-resume
				c := *admitted.Claim
				c.Op.Phase = hostclaim.PhaseOpened
				admitted.Claim = &c
				return admitted, nil
			}
			replace := func() { a.mu.Lock(); a.fences[probeHost] = 12; a.mu.Unlock() }
			if !afterRead {
				replace()
			}
			a.sweepOnce()
			if afterRead {
				select {
				case <-read:
				case <-time.After(time.Second):
					close(resume)
					t.Fatal("sweep skipped cut fence")
				}
				replace()
			}
			close(resume)
			waitProbe(t, a, func() bool { return a.entries[probeHost] == nil && !e.probeReviewing })
			if a.FenceGeneration(probeHost) != 12 || !proc.isStopped() || !rep.isClosed() {
				t.Fatal("replacement fence or retirement lost")
			}
		})
	}
}

func TestRecoveryProbeRefusesAnotherOperationsFence(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, _, run := probeService(t, obs)
	a.fences = map[string]uint64{probeHost: 12}
	p := probePolicy(probeInputs())
	p.RecoveryFence = 9
	if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, p); !errors.Is(err, ErrFenced) {
		t.Fatalf("foreign fence: %v", err)
	}
	if len(run.seen()) != 0 || a.Owns(probeHost) || a.FenceGeneration(probeHost) != 12 {
		t.Fatal("foreign fence allowed recovery admission")
	}
}

// The moves worker may finish the rollback and lift its own fence before
// the slower activator sweep observes opened/serving. The exact session
// still owns its process; a missing fence alone is not an opening proof.
func TestRecoveryProbeSweepAcceptsOwnerLiftOnlyWhenOpened(t *testing.T) {
	t.Parallel()
	for _, phase := range []string{hostclaim.PhaseOpened, "serving", hostclaim.PhaseProbing} {
		for _, duringRead := range []bool{false, true} {
			t.Run(phase+map[bool]string{false: "/before", true: "/during"}[duringRead], func(t *testing.T) {
				a, r, e, proc, _, admitted := sweptProbe(t)
				read, resume := make(chan struct{}), make(chan struct{})
				r.Read = func(context.Context, string) (hostclaim.Observed, error) {
					close(read)
					<-resume
					c := *admitted.Claim
					c.Op.Phase = phase
					if phase == "serving" {
						c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
					}
					admitted.Claim = &c
					return admitted, nil
				}
				if !duringRead && !a.LiftFence(probeHost, 9) {
					t.Fatal("owner lift refused")
				}
				a.sweepOnce()
				select {
				case <-read:
				case <-time.After(time.Second):
					close(resume)
					t.Fatal("owner lift retired probe before reading claim")
				}
				if duringRead && !a.LiftFence(probeHost, 9) {
					close(resume)
					t.Fatal("owner lift refused")
				}
				close(resume)
				waitProbe(t, a, func() bool { return !e.probeReviewing })
				if phase == hostclaim.PhaseProbing {
					a.mu.Lock()
					held := a.entries[probeHost] == e && e.state == actAwake && e.probeHeld
					a.mu.Unlock()
					if !held || proc.isStopped() {
						t.Fatal("an owner lift with a probing read retired the session")
					}
					if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) {
						t.Fatalf("a missing fence opened a still-held probe: %v", err)
					}
					// The earlier read may precede opening. A later sweep
					// must see the same session opened before it can serve.
					r.Read = func(context.Context, string) (hostclaim.Observed, error) {
						c := *admitted.Claim
						c.Op.Phase = hostclaim.PhaseOpened
						return hostclaim.Observed{Host: admitted.Host, Claim: &c, ETag: admitted.ETag}, nil
					}
					a.sweepOnce()
					waitProbe(t, a, func() bool { return !e.probeHeld })
				}
				if proc.isStopped() || a.Fenced(probeHost) {
					t.Fatal("opened session lost its process")
				}
				release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost})
				if err != nil {
					t.Fatal(err)
				}
				release()
			})
		}
	}
}

func TestRecoveryProbeHealthyFinishAfterOwnerLift(t *testing.T) {
	obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
	a, r, _ := probeService(t, obs)
	admitted := probeAdmitted(obs)
	e, proc, _, err := heldProbe(t, a, 9, admitted, probeInputs())
	if err != nil {
		t.Fatalf("owner lift prematurely retired healthy finish: %v", err)
	}
	c := *admitted.Claim
	c.Op.Phase = hostclaim.PhaseOpened
	admitted.Claim = &c
	r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil }
	a.sweepOnce()
	waitProbe(t, a, func() bool { return !e.probeHeld })
	if proc.isStopped() {
		t.Fatal("opened process stopped")
	}
}