package activator

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

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

// Issue #213: a box that cannot provide an instance's config must not take
// its lease. The gate runs inside ExecBackend.Acquire, before lease.Acquire,
// so every wake source (request, tick, sweeper, worker boot) is covered by
// the one place a lease is taken.

// The gate is asked about the route being woken, and a refusal means the
// lease is never attempted. Cfg.Bucket is empty here, so a call that fell
// through to lease.Acquire would fail with a different error.
func TestExecBackendAcquireRefusesBeforeTheLease(t *testing.T) {
	var asked []registry.Route
	boom := errors.New("cannot open sealed config")
	b := &ExecBackend{ConfigGate: func(_ context.Context, rt registry.Route) error {
		asked = append(asked, rt)
		return boom
	}}
	rt := registry.Route{Host: "kpmg.example.com", Kind: "instance", App: "shop", Hibernate: true}

	lease, err := b.Acquire(context.Background(), rt)
	if lease != nil {
		t.Fatalf("lease = %v, want nil", lease)
	}
	var refused *errConfigRefused
	if !errors.As(err, &refused) {
		t.Fatalf("err = %v, want *errConfigRefused", err)
	}
	if !errors.Is(err, boom) {
		t.Errorf("err %v does not wrap the gate's reason", err)
	}
	if len(asked) != 1 || asked[0].Host != rt.Host {
		t.Errorf("gate asked about %v, want exactly %s", asked, rt.Host)
	}
}

// The refusal is the member's 503 with a Retry-After, answered quietly (the
// guard logs once per host; a log line per request would drown it), and the
// failed wake leaves no entry behind so the next request asks again.
func TestActivatorConfigRefusal503(t *testing.T) {
	a, b, _, reg := testActivator(t, time.Minute)
	b.acquireErr = &errConfigRefused{err: errors.New("no key")}

	if err := reg.Upsert(registry.Route{Host: "nokey.example.com", Kind: "instance", Socket: "/tmp/nokey.sock", Hibernate: true}); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	front := httptest.NewServer(rt)
	t.Cleanup(front.Close)

	req, _ := http.NewRequest("GET", front.URL+"/", nil)
	req.Host = "nokey.example.com"
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status %d, want 503", resp.StatusCode)
	}
	if ra := resp.Header.Get("Retry-After"); ra != "30" {
		t.Fatalf("Retry-After %q, want 30", ra)
	}
	if a.entryCount() != 0 {
		t.Fatal("refused wake left an entry behind")
	}
	// Nothing past Acquire ran: no restore, no spawn.
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.restores != 0 || b.starts != 0 {
		t.Fatalf("refused wake went on: restores=%d starts=%d", b.restores, b.starts)
	}
}

// A worker respawn keeps its lease and calls StartWorker directly, so it
// bypasses Acquire. The config gate has to be asked again there, and a refusal
// has to stand the worker down (stop it, release the lease) rather than leave
// a childless entry holding the lease from a sibling that could run it.
func TestWorkerRespawnIsRefusedByTheConfigGateAndStandsDown(t *testing.T) {
	t.Parallel()
	for _, why := range []string{"crash", "max_run"} {
		t.Run(why, func(t *testing.T) {
			a, b, clk, _ := testActivator(t, time.Hour)
			rt := alwaysOnWorkerRoute("bot.x.internal")
			rt.MaxRunSeconds = 30
			if err := a.WakeWorker(rt); err != nil {
				t.Fatalf("WakeWorker: %v", err)
			}
			if b.workerStarts() != 1 {
				t.Fatalf("workerStarts = %d, want 1", b.workerStarts())
			}
			first := b.lastWorker()

			// Secrets arrive after the worker started; this box cannot open them.
			b.mu.Lock()
			b.admitErr = errors.New("no key")
			b.mu.Unlock()

			if why == "crash" {
				first.kill()
			} else {
				clk.advance(31 * time.Second)
			}
			a.sweepOnce()

			waitFor(t, "the refused worker to stand down", func() bool { return a.entryCount() == 0 })
			if n := b.workerStarts(); n != 1 {
				t.Fatalf("workerStarts = %d: a refused respawn still started a child", n)
			}
			if !first.isStopped() && why == "max_run" {
				t.Error("the old child was left running")
			}
			if n := b.releaseCount(); n != 1 {
				t.Fatalf("lease releases = %d, want 1: a refused worker must give its lease back", n)
			}
			if !b.lastRep().isClosed() {
				t.Error("replication was left open on a refused worker")
			}
		})
	}
}

// Build convergence shares the respawn helper but is its own trigger: a ship
// repoints live/<host>, the drift outlasts ConvergeGrace, and the always-on
// worker would be cycled onto the new build. With the gate refusing, it stands
// down instead of starting the new build without its config.
func TestWorkerConvergeRespawnIsRefusedByTheConfigGateAndStandsDown(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	first := b.lastWorker()
	b.mu.Lock()
	b.admitErr = errors.New("cannot open sealed config")
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce() // first sighting of the drift
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce() // the converge respawn, refused

	waitFor(t, "the refused convergence to stand down", func() bool { return a.entryCount() == 0 })
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d: a refused convergence still started the new build", n)
	}
	if !first.isStopped() {
		t.Error("the old child was left running")
	}
	if n := b.releaseCount(); n != 1 {
		t.Fatalf("lease releases = %d, want 1", n)
	}
	if !b.lastRep().isClosed() {
		t.Error("replication was left open")
	}
}

// And a gate that allows it changes nothing: the respawn runs in place and
// keeps its lease, as before.
func TestWorkerRespawnProceedsWhenTheConfigGateAllows(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 2)
	if n := b.releaseCount(); n != 0 {
		t.Fatalf("lease releases = %d, want 0", n)
	}
	b.mu.Lock()
	admits := b.admits
	b.mu.Unlock()
	if admits == 0 {
		t.Error("the respawn never asked the config gate")
	}
}

// The gate follows dispatch, which is Backing alone (SplitBackend.pick). A
// socket-backed hibernating instance may carry a Unit with an empty Backing
// and still reaches ExecBackend.Acquire, so it is gated; only Backing == unit
// is exempt.
func TestSplitBackendGatesWhatDispatchSendsToExec(t *testing.T) {
	var asked []string
	exec := &ExecBackend{ConfigGate: func(_ context.Context, rt registry.Route) error {
		asked = append(asked, rt.Host)
		return errors.New("no key")
	}}
	b := &SplitBackend{Exec: exec, Unit: &UnitBackend{}}

	unitNamed := registry.Route{Host: "named.example.com", Kind: "instance", App: "shop",
		Socket: "/run/named.sock", Unit: "x.service", Hibernate: true}
	if _, err := b.Acquire(context.Background(), unitNamed); err == nil {
		t.Fatal("Acquire of a Unit-named, empty-Backing instance was not refused")
	}
	if err := b.AdmitConfig(context.Background(), unitNamed); err == nil {
		t.Fatal("AdmitConfig of a Unit-named, empty-Backing instance was not refused")
	}
	unitBacked := unitNamed
	unitBacked.Host, unitBacked.Backing = "backed.example.com", registry.BackingUnit
	if err := b.AdmitConfig(context.Background(), unitBacked); err != nil {
		t.Fatalf("a Backing=unit route must not be gated: %v", err)
	}
	if len(asked) != 2 || asked[0] != "named.example.com" || asked[1] != "named.example.com" {
		t.Fatalf("gate asked %v, want only the Unit-named exec route, twice", asked)
	}
}