package activator

import (
	"context"
	"reflect"
	"sync/atomic"
	"testing"
	"time"
)

// The adoption a real box performs is delegated: the sandboxed agent cannot
// write the version store, so the ROOT pass stages, repoints live/<host> and
// stamps the version — and cycles no exec child, because the children are the
// agent's. Nothing then told the agent the symlink had moved, and the awake
// child served the old build until the sweep's grace ran out (discussion #231:
// 110-200s per deploy on the flagship). ConvergeAdopted is the agent finishing
// the half it delegated: no grace, no sweep.
func TestConvergeAdoptedCyclesAStaleInstanceWithoutTheGrace(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour) // idle far away — only convergence can stop it
	var ver atomic.Value
	ver.Store("v1")
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }
	host := "one.example.com"

	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if got := a.Drifted(); len(got) != 0 {
		t.Fatalf("Drifted = %v with the symlink unmoved, want none", got)
	}

	ver.Store("v2") // the root pass repoints under the running child
	if got, want := a.Drifted(), map[string]string{host: "v2"}; !reflect.DeepEqual(got, want) {
		t.Fatalf("Drifted = %v, want %v", got, want)
	}
	if !a.ConvergeAdopted(host, "v2") {
		t.Fatal("ConvergeAdopted refused an awake instance serving v1 under a symlink that names v2")
	}
	// No clock advance and no sweep: the cycle is the call's own doing.
	waitFor(t, "the adopted-version cycle", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("stale instance's process was not stopped")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("the cycle skipped the final checkpoint / lease release")
	}

	// The next request wakes it onto v2, and a converged child is left alone.
	release, err = a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("re-wake after the cycle: %v", err)
	}
	release()
	if got := a.Drifted(); len(got) != 0 {
		t.Fatalf("Drifted = %v after re-waking onto v2, want none", got)
	}
	if a.ConvergeAdopted(host, "v2") {
		t.Fatal("ConvergeAdopted cycled a child that already runs the adopted version")
	}
	if a.entryCount() != 1 {
		t.Fatal("a converged instance was stopped")
	}
}

// Every refusal leaves the instance exactly as it was. In particular it never
// WAKES anything: a sleeper picks the new build up at its own next wake.
func TestConvergeAdoptedRefusals(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	var ver atomic.Value
	ver.Store("v1")
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }
	host := "one.example.com"

	if a.ConvergeAdopted(host, "v2") {
		t.Fatal("ConvergeAdopted acted on a host that is asleep")
	}
	if a.entryCount() != 0 {
		t.Fatal("ConvergeAdopted woke a sleeping host")
	}

	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	ver.Store("v2")

	// The caller's version and the symlink must agree. A stamp that names
	// something the symlink does not is an adoption still in flight (or
	// already superseded), and cycling now would land on whichever build the
	// symlink happens to name.
	if a.ConvergeAdopted(host, "v3") {
		t.Fatal("ConvergeAdopted cycled onto a version the live symlink does not name")
	}
	if a.ConvergeAdopted(host, "") {
		t.Fatal("ConvergeAdopted cycled for an empty version")
	}
	if a.entryCount() != 1 {
		t.Fatal("a refused ConvergeAdopted stopped the instance")
	}
}

// An unknown spawn version is "don't converge", the same rule the sweep keeps:
// with nothing to compare, a cycle could never be shown to have been needed.
func TestConvergeAdoptedLeavesAnUnknownSpawnVersionAlone(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	a.liveVer = func(string) (string, error) { return "v2", nil }
	b.execVerFn = func() string { return "" }
	host := "one.example.com"

	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if got := a.Drifted(); len(got) != 0 {
		t.Fatalf("Drifted = %v for a child whose spawn version is unknown, want none", got)
	}
	if a.ConvergeAdopted(host, "v2") {
		t.Fatal("ConvergeAdopted cycled a child whose spawn version is unknown")
	}
}

// A worker is out of this path's reach on purpose: its registry version is
// what the child REPORTS, not what adoption stamped, so "the stamp agrees with
// the symlink" means something else there. The sweep's own worker converge
// (one move, behind the grace) still covers it.
func TestConvergeAdoptedLeavesWorkersToTheSweep(t *testing.T) {
	t.Parallel()
	a, _, _, ver := shippingActivator(t, time.Hour)
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	waitFor(t, "the worker to be awake", func() bool { return a.entrySpawnVersion(host) == "v1" })
	ver.Store("v2")
	if got := a.Drifted(); len(got) != 0 {
		t.Fatalf("Drifted = %v, want no workers in it", got)
	}
	if a.ConvergeAdopted(host, "v2") {
		t.Fatal("ConvergeAdopted cycled a worker")
	}
}

// An adopt pass's restart callback fires after it has repointed live/<host>,
// so by then the symlink names the version being adopted. A child ALREADY
// exec'd from that version has nothing to pick up: stopping it is one more
// drain and wake for no change of build. That child exists whenever something
// else got there first — ConvergeAdopted finishing a root pass's adoption and
// a request re-waking the instance, while a slower pass was still mid-host.
func TestBounceAdoptedSparesAChildAlreadyOnTheLiveBuild(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	var ver atomic.Value
	ver.Store("v2")
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }
	host := "one.example.com"

	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if a.BounceAdopted(host) {
		t.Fatal("BounceAdopted stopped a child that already runs the build the symlink names")
	}
	if a.entryCount() != 1 {
		t.Fatal("the converged child was stopped")
	}

	// The ordinary case is unchanged: the symlink has moved under the child.
	ver.Store("v3")
	if !a.BounceAdopted(host) {
		t.Fatal("BounceAdopted spared a child serving v2 under a symlink that names v3")
	}
	waitFor(t, "the adoption bounce", func() bool { return a.entryCount() == 0 })
	if a.BounceAdopted(host) {
		t.Fatal("BounceAdopted acted on a host that is asleep")
	}
}

// With nothing to compare it bounces, exactly as Bounce always has: a backend
// that cannot name its build, or a symlink that cannot be read, must not turn
// an adoption into a silent no-op over a child still on the old binary.
func TestBounceAdoptedBouncesWhenTheBuildIsUnknown(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	a.liveVer = func(string) (string, error) { return "v2", nil }
	b.execVerFn = func() string { return "" }
	host := "one.example.com"

	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if !a.BounceAdopted(host) {
		t.Fatal("BounceAdopted refused a child whose spawn version is unknown")
	}
	waitFor(t, "the bounce", func() bool { return a.entryCount() == 0 })
}

// A bounce that lands while the instance is still waking is not dropped. The
// wake already read its config and execed its build, so whatever the bounce
// was for — new config, a `carlos restart` — has not reached the child, and
// nothing will ask again: config delivery bounces only when the rendered
// file changes. The bounce is held and spent the moment the wake finishes,
// before any waiting request is let through, so every request lands on a
// child started after the bounce was asked for.
func TestBounceDuringAWakeIsAppliedWhenTheWakeFinishes(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	b.startGate = make(chan struct{})
	host := "one.example.com"

	served := make(chan error, 1)
	go func() {
		release, err := a.beginRequest(context.Background(), testRoute(host))
		if err == nil {
			release()
		}
		served <- err
	}()
	waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 })

	if a.Bounce(host) {
		t.Fatal("Bounce reported a stop on an instance that is still waking")
	}
	close(b.startGate)

	if err := <-served; err != nil {
		t.Fatalf("the request waiting on the wake: %v", err)
	}
	// It was served by a SECOND child: the first was stopped as soon as its
	// wake finished, and the waiting request woke a fresh one.
	if got := b.startCount(); got != 2 {
		t.Fatalf("children started = %d, want 2: the bounce asked for during the wake was dropped", got)
	}
	if !b.procs[0].isStopped() {
		t.Fatal("the child from the wake the bounce landed in is still running")
	}
	if a.Bounce(host) != true {
		t.Fatal("an ordinary bounce of the awake replacement was refused")
	}
}