package activator

import (
	"context"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// Index review focus 2. The agent restarts while a successor is spawning or
// while the retiring generation is still draining. Shutdown joins the whole
// operation set before the lease goes back. The next agent's first wake
// boots the installed build through the live link, and nothing of the swap
// survives the restart. This is the witness for mutations M1, M2 and M4.
func TestSwapWitnessAgentRestartMidSwap(t *testing.T) {
	t.Parallel()
	restart := func(t *testing.T, f *swapFixture) {
		t.Helper()
		f.b.execVerFn = func() string { return "v2" } // root installed v2
		gensBefore, _, _ := f.b.calls()
		a2 := New(Config{Bucket: "test-bucket", Idle: time.Hour}, f.b, f.reg)
		a2.gens = true
		arrangeAwake(t, a2, testRoute(f.host))
		if got := f.b.startCount(); got != 2 {
			t.Fatalf("Start (live link) calls = %d after the restart, want 2", got)
		}
		if gens, _, _ := f.b.calls(); gens != gensBefore {
			t.Fatal("the restarted agent spawned a generation from the old swap")
		}
		a2.mu.Lock()
		ver := a2.entries[f.host].spawnVersion
		a2.mu.Unlock()
		if ver != "v2" {
			t.Fatalf("the next wake runs %q, want the installed v2", ver)
		}
	}

	t.Run("successor spawning", func(t *testing.T) {
		f := newSwapFixture(t)
		// A spawn that outlives cancellation: StartStaged ignores its ctx
		// until the hold opens, then returns a live successor. Shutdown
		// cancels the attempt, but only the join makes it wait for that
		// process, so the incumbent may not stop and the lease may not go
		// back until the successor exists and has been stopped.
		hold := make(chan struct{})
		f.b.genHold = hold
		incumbent, rep := f.b.lastProc(), f.b.lastRep()
		var leaked atomic.Bool
		rep.setOnClose(func() {
			if _, succ := f.b.lastGen(); succ == nil || !succ.Exited() || !incumbent.Exited() {
				leaked.Store(true)
			}
		})
		results := make(chan SwapResult, 1)
		go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
		waitFor(t, "the successor's spawn", func() bool { return indexPrefix(f.b.events(), "startGen:") >= 0 })
		shut := make(chan struct{})
		go func() {
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			f.a.Shutdown(ctx)
			close(shut)
		}()
		waitFor(t, "Shutdown to reach the entry", func() bool {
			f.a.mu.Lock()
			defer f.a.mu.Unlock()
			// A missing entry means the stop already finished: the checks
			// below report that, rather than a nil dereference here.
			e := f.a.entries[f.host]
			return e == nil || e.state == actStopping
		})
		// The barrier holds the stop back while the spawn is in flight. As in
		// the draining case nothing signals the barrier itself, so the
		// negative is watched for a window; a stop that passed it would stop
		// the incumbent at once, and the poll fails then.
		for deadline := time.Now().Add(200 * time.Millisecond); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
			if incumbent.isStopped() {
				t.Fatal("the incumbent was stopped while a successor was still spawning")
			}
			if rep.isClosed() {
				t.Fatal("the lease went back while a successor was still spawning")
			}
		}
		select {
		case <-shut:
			t.Fatal("Shutdown returned with a successor still spawning")
		default:
		}
		close(hold)
		<-shut
		if res := <-results; res.Outcome != SwapAbandoned {
			t.Fatalf("Swap = %+v, want abandoned by the shutdown", res)
		}
		if _, succ := f.b.lastGen(); succ == nil || !succ.Exited() {
			t.Fatal("the successor that outlived the cancellation was never stopped")
		}
		if !rep.isClosed() || leaked.Load() {
			t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load())
		}
		if _, raises, restores := f.b.calls(); raises != restores {
			t.Fatalf("raises %d restores %d", raises, restores)
		}
		restart(t, f)
	})

	t.Run("retiring generation draining", func(t *testing.T) {
		f := newSwapFixture(t)
		incumbent, rep := f.b.lastProc(), f.b.lastRep()
		gate := make(chan struct{})
		incumbent.setStopGate(gate)
		var leaked, opened atomic.Bool
		rep.setOnClose(func() {
			_, succ := f.b.lastGen()
			if !opened.Load() || !incumbent.Exited() || succ == nil || !succ.Exited() {
				leaked.Store(true)
			}
		})
		results := make(chan SwapResult, 1)
		go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
		waitFor(t, "the switch", func() bool { return indexPrefix(f.b.events(), "switch:") >= 0 })
		shut := make(chan struct{})
		go func() {
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			f.a.Shutdown(ctx)
			close(shut)
		}()
		// Shutdown has reached the entry (it is stopping) while the swap's
		// retire of the incumbent still sits on its gate: both are observed,
		// not waited out. The lease hook above is the sound half of the
		// negative: it flags any release made before the gate opened.
		waitFor(t, "Shutdown to reach the entry", func() bool {
			f.a.mu.Lock()
			defer f.a.mu.Unlock()
			// A missing entry means the stop already finished: the checks
			// below report that, rather than a nil dereference here.
			e := f.a.entries[f.host]
			return e == nil || e.state == actStopping
		})
		waitFor(t, "the retire of the incumbent", func() bool { return incumbent.stopEntered() >= 1 })
		// The barrier holds Shutdown's own stop of the entry back until the
		// retiring generation has drained. The negative is watched for a
		// window, since nothing signals the barrier itself: a stop that
		// passed it would stop the serving successor within a few
		// instructions, and that poll fails at once.
		for deadline := time.Now().Add(200 * time.Millisecond); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
			if _, succ := f.b.lastGen(); succ.isStopped() {
				t.Fatal("Shutdown stopped the successor while the retiring generation was still draining")
			}
			if rep.isClosed() {
				t.Fatal("the lease went back while the retiring generation was still draining")
			}
		}
		select {
		case <-shut:
			t.Fatal("Shutdown returned with a generation still draining")
		default:
		}
		opened.Store(true)
		close(gate)
		<-shut
		if res := <-results; res.Outcome != SwapDone {
			t.Fatalf("Swap = %+v: the switch happened before the shutdown", res)
		}
		if !rep.isClosed() || leaked.Load() {
			t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load())
		}
		restart(t, f)
	})
}

// Index review focus 4. An app that never answers GET / below 500 fails its
// health check. The incumbent keeps serving and no request is dropped. The
// health check is the real one, over real unix sockets, under a steady load
// that dials AppSocket for every request.
func TestSwapNeverHealthyAppKeepsTheIncumbentServing(t *testing.T) {
	dir, err := os.MkdirTemp("", "zs") // TMPDIR=/tmp/k keeps socket paths short
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	serve := func(path string, code int) func() {
		ln, err := net.Listen("unix", path)
		if err != nil {
			t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/k", err)
		}
		srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(code) })}
		go srv.Serve(ln) //nolint:errcheck
		return func() { srv.Close() }
	}
	old := spawnHealthTimeout
	spawnHealthTimeout = 600 * time.Millisecond
	t.Cleanup(func() { spawnHealthTimeout = old })

	f := newSwapFixture(t)
	f.a.genHealth = nil // the real check: GET / through the generation's handle
	incumbentSock := filepath.Join(dir, "inc.sock")
	stopIncumbent := serve(incumbentSock, http.StatusOK)
	defer stopIncumbent()
	f.incumbent.Socket, f.incumbent.Control = incumbentSock, incumbentSock+".control"
	setCurrentGen(t, f.a, f.host, f.incumbent)
	f.b.genSocketDir = dir
	f.b.genServe = func(g Gen) func() { return serve(g.Socket, http.StatusServiceUnavailable) }

	var sent, bad atomic.Int64
	stop := make(chan struct{})
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-stop:
				return
			default:
			}
			sock, ok := f.a.AppSocket(f.host)
			sent.Add(1)
			if !ok {
				bad.Add(1)
				continue
			}
			resp, err := socketClient(sock).Get("http://carlos/")
			if err != nil {
				bad.Add(1)
				continue
			}
			if resp.StatusCode != http.StatusOK {
				bad.Add(1)
			}
			resp.Body.Close()
		}
	}()
	res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
	close(stop)
	wg.Wait()

	if res.Outcome != SwapFailed || !strings.Contains(res.Reason, "health") {
		t.Fatalf("Swap = %+v, want failed on health", res)
	}
	if got := f.appSocket(t); got != incumbentSock {
		t.Fatalf("AppSocket = %q, want the incumbent", got)
	}
	if bad.Load() != 0 || sent.Load() < 20 {
		t.Fatalf("%d of %d requests failed (want 0 of at least 20)", bad.Load(), sent.Load())
	}
	if _, p := f.b.lastGen(); p == nil || !p.isStopped() {
		t.Fatal("the never-healthy successor was left running")
	}
	if f.b.lastProc().isStopped() {
		t.Fatal("the incumbent was stopped")
	}
	if _, raises, restores := f.b.calls(); raises != 1 || restores != 1 {
		t.Fatalf("raises %d restores %d, want 1 and 1", raises, restores)
	}
}