package activator

import (
	"context"
	"io"
	"net/http"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/schedule"
)

// The switch retires exactly the generation it switched away from (spec
// §7.10), and the retirement is latched (final review F8): the retiring
// generation's writes stay refused after it has drained and left the
// operation set, since a handler that began before the switch may still be
// running. The successor's writes apply throughout, and a hibernating stop
// after the swap retires nothing.
func TestRetiredGenNamesOnlyTheRetiringGeneration(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	duringHealth := map[string]bool{}
	f.a.genHealth = func(_ context.Context, g Gen) error {
		duringHealth["successor"] = f.a.RetiredGen(f.host, g.N)
		duringHealth["incumbent"] = f.a.RetiredGen(f.host, f.incumbent.N)
		return nil
	}
	gate := make(chan struct{})
	f.b.lastProc().setStopGate(gate) // hold the retiring generation's drain open
	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 })
	if duringHealth["successor"] || duringHealth["incumbent"] {
		t.Fatalf("before the switch: %v, want both open", duringHealth)
	}
	succ, _ := f.b.lastGen()
	if !f.a.RetiredGen(f.host, f.incumbent.N) {
		t.Fatal("the retiring generation's schedule writes are still open after the switch")
	}
	if f.a.RetiredGen(f.host, succ.N) {
		t.Fatal("the current generation's writes are refused")
	}
	if f.a.ScheduleWrite(f.host, f.incumbent.N, func() { t.Fatal("a retired generation's write ran") }) {
		t.Fatal("ScheduleWrite ran the retired generation's write")
	}
	close(gate)
	if res := <-results; res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	if _, members := f.a.genStateForTest(f.host); len(members) != 0 {
		t.Fatalf("members %v: the retiring generation should have left the set", members)
	}
	if !f.a.RetiredGen(f.host, f.incumbent.N) {
		t.Fatal("the retirement was not latched: the drained generation's writes reopened")
	}
	ran := false
	if !f.a.ScheduleWrite(f.host, succ.N, func() { ran = true }) || !ran {
		t.Fatal("the successor's write was refused")
	}
	// A hibernating stop is unchanged: the stopping generation is current.
	stopGate := make(chan struct{})
	_, cur := f.b.lastGen()
	cur.setStopGate(stopGate)
	f.clk.advance(2 * time.Hour)
	f.a.sweepOnce()
	waitFor(t, "the hibernating stop", func() bool {
		f.a.mu.Lock()
		defer f.a.mu.Unlock()
		e := f.a.entries[f.host]
		return e != nil && e.state == actStopping
	})
	if f.a.RetiredGen(f.host, succ.N) {
		t.Fatal("a hibernating generation's writes are refused")
	}
	close(stopGate)
}

// schedSwitchWaiting reports whether a switch holds host's write fence and
// is waiting for a write in flight.
func (a *Service) schedSwitchWaiting(host string) bool {
	f := a.schedFenceFor(host)
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.switching && f.writes > 0
}

// Final review F8: the switch is serialized against the schedule writes.
// A write of the incumbent that is under way when the switch comes holds
// the switch until it finishes (so it lands before the switch, while its
// generation is current), and once the switch has happened every write of
// the retired generation is refused.
func TestScheduleWriteIsSerializedWithTheSwitch(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	entered, release := make(chan struct{}), make(chan struct{})
	wrote := make(chan bool, 1)
	go func() {
		wrote <- f.a.ScheduleWrite(f.host, f.incumbent.N, func() {
			close(entered)
			<-release
			f.b.record("schedule-write")
		})
	}()
	<-entered
	results := make(chan SwapResult, 1)
	go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
	waitFor(t, "the switch waiting on the write", func() bool { return f.a.schedSwitchWaiting(f.host) })
	if ev := f.b.events(); indexPrefix(ev, "switch:") >= 0 {
		t.Fatalf("events %v: the switch ran while a schedule write was under way", ev)
	}
	close(release)
	if !<-wrote {
		t.Fatal("the incumbent's write, begun before the switch, was refused")
	}
	if res := <-results; res.Outcome != SwapDone {
		t.Fatalf("Swap = %+v", res)
	}
	ev := f.b.events()
	if w, sw := indexPrefix(ev, "schedule-write"), indexPrefix(ev, "switch:"); w < 0 || sw < 0 || w > sw {
		t.Fatalf("events %v: want the write before the switch", ev)
	}
	if f.a.ScheduleWrite(f.host, f.incumbent.N, func() { t.Fatal("a retired generation's write ran") }) {
		t.Fatal("the retired generation's write after the switch was not refused")
	}
}

// Flag off: nothing is retired and nothing is serialized. A generation the
// fence has latched (here by hand, as a switch would with swaps on) still
// writes once the switch is off.
func TestScheduleWriteIsUnfencedWithSwapOff(t *testing.T) {
	f := newSwapFixture(t)
	f.a.schedFenceFor(f.host).retire(77)
	if f.a.ScheduleWrite(f.host, 77, func() {}) || !f.a.RetiredGen(f.host, 77) {
		t.Fatal("with swaps on, a latched generation's write was not refused")
	}
	f.a.gens = false
	ran := false
	if !f.a.ScheduleWrite(f.host, 77, func() { ran = true }) || !ran || f.a.RetiredGen(f.host, 77) {
		t.Fatal("with swaps off, a write was refused or a generation reported retired")
	}
}

func TestExecBackendWriteFence(t *testing.T) {
	b := testExecBackend(t)
	if b.writeFence("h.example.com", 7) != nil {
		t.Fatal("a backend with no ScheduleFence seam fenced the schedule verbs")
	}
	var gotHost string
	var gotGen uint64
	b.ScheduleFence = func(h string, g uint64, write func()) bool { gotHost, gotGen = h, g; write(); return true }
	ran := false
	if !b.writeFence("h.example.com", 7)(func() { ran = true }) || !ran || gotHost != "h.example.com" || gotGen != 7 {
		t.Fatalf("fence asked %q %d, ran %v", gotHost, gotGen, ran)
	}
}

// The seam reaches the real control socket of a spawned child (the flat
// listener): the verbs are mounted through the fence, the fence is asked
// about this host and the generation the listener was bound for (0 for a
// plain Start), and once it refuses a write is a 409 while a read still
// answers.
func TestExecBackendStartGatesTheControlSocketWrites(t *testing.T) {
	b := testExecBackend(t)
	b.Timers = &fakeTimerStore{}
	var retired atomic.Bool
	var askedHost atomic.Value
	askedGen := uint64(99)
	var mu sync.Mutex
	b.ScheduleFence = func(host string, gen uint64, write func()) bool {
		askedHost.Store(host)
		mu.Lock()
		askedGen = gen
		mu.Unlock()
		if retired.Load() {
			return false
		}
		write()
		return true
	}
	ctl, token, stop := startAndReachControlSocket(t, b, "gated.example.com")
	defer stop()

	send := func(method, path, body string) (int, string) {
		req, err := http.NewRequest(method, "http://carlos"+path, strings.NewReader(body))
		if err != nil {
			t.Fatal(err)
		}
		req.Header.Set("Authorization", "Bearer "+token)
		resp, err := socketClient(ctl).Do(req)
		if err != nil {
			t.Fatalf("dial control socket %s: %v", ctl, err)
		}
		defer resp.Body.Close()
		var sb strings.Builder
		buf := make([]byte, 512)
		for {
			n, err := resp.Body.Read(buf)
			sb.Write(buf[:n])
			if err != nil {
				break
			}
		}
		return resp.StatusCode, sb.String()
	}
	if code, _ := send("DELETE", "/schedule/remind", ""); code != http.StatusNoContent {
		t.Fatalf("open DELETE = %d, want 204", code)
	}
	retired.Store(true)
	if code, msg := send("DELETE", "/schedule/remind", ""); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) {
		t.Fatalf("retired DELETE = %d %q, want 409", code, msg)
	}
	if code, msg := send("PUT", "/schedule/remind", `{"at":"2026-09-01T08:00:00Z","path":"/r"}`); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) {
		t.Fatalf("retired PUT = %d %q, want 409", code, msg)
	}
	if code, _ := send("GET", "/schedule", ""); code != http.StatusOK {
		t.Fatalf("retired GET = %d, want 200", code)
	}
	mu.Lock()
	gen := askedGen
	mu.Unlock()
	if askedHost.Load() != "gated.example.com" || gen != 0 {
		t.Fatalf("fence asked %v generation %d, want gated.example.com generation 0", askedHost.Load(), gen)
	}
}

// Final review F8, T8: the nested layout's control listener (prepareAt,
// which the nested generation start uses) is mounted through the same
// fence, asked for the generation it was prepared for.
func TestPreparedControlListenerIsFenced(t *testing.T) {
	b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{})
	b.GenContainment = true
	b.Timers = &fakeTimerStore{}
	var retired atomic.Bool
	var mu sync.Mutex
	asked := map[uint64]int{}
	b.ScheduleFence = func(host string, gen uint64, write func()) bool {
		mu.Lock()
		if host == rt.Host {
			asked[gen]++
		}
		mu.Unlock()
		if retired.Load() {
			return false
		}
		write()
		return true
	}
	if l, _ := b.Layout(rt); l != LayoutNested {
		t.Fatalf("layout %v, want nested", l)
	}
	p0 := b.pathsFor(rt)
	if err := os.MkdirAll(p0.runDir, 0o750); err != nil {
		t.Fatal(err)
	}
	sock := nestedGenSocketIn(p0.runDir, 7)
	p, ctl, logs, _, err := b.prepareAt(rt, instancectl.RoleRun, sock, 7)
	if err != nil {
		t.Fatalf("prepareAt: %v", err)
	}
	if logs != nil {
		defer logs.Close()
	}
	if ctl == nil {
		t.Fatal("prepareAt bound no control listener")
	}
	defer ctl.Close()
	env, err := readEnvFileBeneath(p.dataDir, rt.Host+".env")
	if err != nil {
		t.Fatal(err)
	}
	token := env["CARLOS_ADMIN_TOKEN"]
	send := func(method, path, body string) (int, string) {
		req, err := http.NewRequest(method, "http://carlos"+path, strings.NewReader(body))
		if err != nil {
			t.Fatal(err)
		}
		req.Header.Set("Authorization", "Bearer "+token)
		resp, err := socketClient(sock + ".control").Do(req)
		if err != nil {
			t.Fatalf("dial control socket: %v", err)
		}
		defer resp.Body.Close()
		buf, _ := io.ReadAll(resp.Body)
		return resp.StatusCode, string(buf)
	}
	if code, msg := send("DELETE", "/schedule/remind", ""); code != http.StatusNoContent {
		t.Fatalf("open DELETE = %d %q, want 204", code, msg)
	}
	retired.Store(true)
	if code, msg := send("PUT", "/schedule/remind", `{"at":"2026-09-02T08:00:00Z","path":"/r2"}`); code != http.StatusConflict || !strings.Contains(msg, schedule.RetiredWriteMessage) {
		t.Fatalf("retired PUT = %d %q, want 409", code, msg)
	}
	if code, _ := send("DELETE", "/schedule/remind", ""); code != http.StatusConflict {
		t.Fatalf("retired DELETE = %d, want 409", code)
	}
	if code, _ := send("GET", "/schedule", ""); code != http.StatusOK {
		t.Fatalf("retired GET = %d, want 200", code)
	}
	mu.Lock()
	defer mu.Unlock()
	if asked[7] != 3 || len(asked) != 1 {
		t.Fatalf("fence asked %v, want generation 7 for each of the three writes", asked)
	}
}