package instancectl

import (
	"context"
	"io"
	"net"
	"net/http"
	"strings"
	"testing"
	"time"
)

// Astra round 3 P1: connection 1 (keep-alive) is accepted first and later
// carries the /stop; a delayed /start arrives on connection 2 and has not
// registered when the stop fences the host. The fence is the accepted
// connection watermark at that moment, not the stop's own number, so the
// delayed start is refused when it finally registers.
func TestADelayedStartAcceptedBeforeTheFenceIsRefusedWhenItRegisters(t *testing.T) {
	host := "watermark.example.com"
	s, _ := newServer(t, &fakeCtl{})
	_ = s.nextConnSeq()          // connection 1: will carry the /stop
	startConn := s.nextConnSeq() // connection 2: the delayed /start
	release, err := s.fenceLegacyStarts(context.Background(), host)
	if err != nil {
		t.Fatalf("fence: %v", err)
	}
	release()
	if _, finish, err := s.beginLegacyStart(context.Background(), host, startConn); err == nil {
		finish()
		t.Fatal("a start accepted before the fence registered after it and was admitted")
	}
}

// Astra round 3 P2: a client that reuses its connection gets a fresh stamp
// for every request, because the server answers each on its own connection.
// A start after a stop, from the same keep-alive client, runs.
func TestAKeepAliveClientsStartAfterAStopRuns(t *testing.T) {
	host := "reuse.example.com"
	ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
	_, sock := serveOnSocket(t, ctl)
	var dials int
	c := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
		dials++
		return (&net.Dialer{}).DialContext(ctx, "unix", sock)
	}}}
	for i, path := range []string{"/start?host=" + host + "&role=run&memory_max=536870912", "/stop?host=" + host + "&role=run", "/start?host=" + host + "&role=run&memory_max=536870912"} {
		resp, err := c.Post("http://unix"+path, "", nil)
		if err != nil {
			t.Fatalf("POST %s: %v", path, err)
		}
		body, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		if resp.StatusCode/100 != 2 {
			t.Fatalf("request %d %s: status %d: %s", i, path, resp.StatusCode, body)
		}
	}
	if dials != 3 {
		t.Fatalf("dials = %d, want one connection per request", dials)
	}
}

// Liveness 1: the fence covers only starts accepted before the stop. A
// /start accepted after the /stop has returned runs normally, and nothing
// of a fence outlives the server: a restarted instance-control has none.
func TestTheFenceCoversOnlyStartsAcceptedBeforeTheStop(t *testing.T) {
	host := "after.example.com"
	ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
	s, c := newServer(t, ctl)
	stop := post(t, c, "/stop?host="+host+"&role=run")
	stop.Body.Close()
	if stop.StatusCode != http.StatusNoContent {
		t.Fatalf("/stop status %d", stop.StatusCode)
	}
	start := post(t, c, "/start?host="+host+"&role=run&memory_max=536870912")
	start.Body.Close()
	if start.StatusCode != http.StatusOK {
		t.Fatalf("a start accepted after the stop returned: status %d, want 200", start.StatusCode)
	}
	s.legacyMu.Lock()
	stopping := s.legacyStopping[host]
	s.legacyMu.Unlock()
	if stopping != 0 {
		t.Fatalf("the stop's refusal of new starts outlived it (%d)", stopping)
	}
	// A restarted service: the fence lived in memory only.
	_, c2 := newServer(t, &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv2\n"}})
	again := post(t, c2, "/start?host="+host+"&role=run&memory_max=536870912")
	again.Body.Close()
	if again.StatusCode != http.StatusOK {
		t.Fatalf("a restarted instance-control refused a start: status %d", again.StatusCode)
	}
}

// heldStart parks host's legacy /start before its tier step until release
// is closed, and returns once the handler is parked there. The handler does
// not see its own cancellation (unseenCancel).
func heldStart(t *testing.T, ctl *fakeCtl, c *http.Client, host string) (release chan struct{}) {
	t.Helper()
	entered, gate := make(chan struct{}), make(chan struct{})
	ctl.mu.Lock()
	if ctl.before == nil {
		ctl.before = map[string]func(){}
	}
	ctl.before["set-property"] = func() {
		close(entered)
		<-gate
	}
	ctl.unseenCancel = true
	ctl.mu.Unlock()
	postAsync(c, "/start?host="+host+"&role=run&memory_max=536870912")
	select {
	case <-entered:
	case <-time.After(5 * time.Second):
		t.Fatal("the held start never reached its tier step")
	}
	return gate
}

// Liveness 2: the join is per host. Host A's attempt will not join; /start
// and /stop of host B answer at once while A's /stop is still joining.
func TestAStuckJoinDelaysNoOtherHost(t *testing.T) {
	ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
	_, c := newServer(t, ctl)
	gate := heldStart(t, ctl, c, "stuck.example.com")
	stopA := postAsync(c, "/stop?host=stuck.example.com&role=run")
	defer func() { close(gate); <-stopA }()
	time.Sleep(50 * time.Millisecond) // A's stop is joining
	for _, path := range []string{"/start?host=free.example.com&role=run&memory_max=536870912", "/stop?host=free.example.com&role=run"} {
		done := postAsync(c, path)
		select {
		case code := <-done:
			if code/100 != 2 {
				t.Fatalf("%s: status %d", path, code)
			}
		case <-time.After(2 * time.Second):
			t.Fatalf("%s waited behind another host's stuck join", path)
		}
	}
	select {
	case <-stopA:
		t.Fatal("A's stop answered while its attempt was still stuck")
	default:
	}
}

// Liveness 3: a stuck stop answers once, within the bound, and the next
// /stop of the host does not hang behind the stuck attempt: it answers the
// same error within the bound, and joins it once it returns.
func TestAStuckStopAnswersAndTheNextStopDoesNotHang(t *testing.T) {
	host := "wedged.example.com"
	ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
	_, c := newServer(t, ctl, func(s *Server) { s.legacyJoin = 200 * time.Millisecond })
	gate := heldStart(t, ctl, c, host)
	stopOnce := func() (int, string, time.Duration) {
		begun := time.Now()
		resp := post(t, c, "/stop?host="+host+"&role=run")
		body, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		return resp.StatusCode, string(body), time.Since(begun)
	}
	for i := 0; i < 2; i++ {
		code, body, took := stopOnce()
		if code != http.StatusBadGateway || !strings.Contains(body, "still in flight") {
			t.Fatalf("stop %d with the attempt stuck: %d %q, want 502 naming the start in flight", i+1, code, body)
		}
		if took > 2*time.Second {
			t.Fatalf("stop %d took %s, past its bound", i+1, took)
		}
	}
	close(gate)
	time.Sleep(50 * time.Millisecond)
	if code, body, _ := stopOnce(); code != http.StatusNoContent {
		t.Fatalf("stop once the attempt returned: %d %q, want 204", code, body)
	}
}

// Liveness 4: the other verbs pass through unchanged during a legacy stop.
// While host A's legacy /stop is joining a stuck attempt, A's sidecar
// /start and seed /stop answer at once.
func TestOtherVerbsPassThroughDuringALegacyStop(t *testing.T) {
	host := "verbs.example.com"
	ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
	_, c := newServer(t, ctl)
	gate := heldStart(t, ctl, c, host)
	stop := postAsync(c, "/stop?host="+host+"&role=run")
	defer func() { close(gate); <-stop }()
	time.Sleep(50 * time.Millisecond) // the legacy stop is joining
	for _, path := range []string{"/start?host=" + host + "&role=sidecar&memory_max=536870912", "/stop?host=" + host + "&role=seed"} {
		done := postAsync(c, path)
		select {
		case code := <-done:
			if code/100 != 2 {
				t.Fatalf("%s during a legacy stop: status %d", path, code)
			}
		case <-time.After(2 * time.Second):
			t.Fatalf("%s waited behind a legacy stop", path)
		}
	}
}

// postAsync POSTs path from its own goroutine, never touching t, and
// delivers the status (0 on a transport error).
func postAsync(c *http.Client, path string) <-chan int {
	done := make(chan int, 1)
	go func() {
		resp, err := c.Post("http://unix"+path, "", nil)
		if err != nil {
			done <- 0
			return
		}
		resp.Body.Close()
		done <- resp.StatusCode
	}()
	return done
}