package activator

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"slices"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// Zero-downtime plan 5 Task 12: the generations file names every live or
// pending INSTANCE generation too (spec §4.2, §8.5), and the build a staged
// instance spawn is about to exec is in a successfully written file before
// that exec, or the spawn does not happen.

// hostsOf is the generations file's hosts, or nil when it does not read.
func hostsOf(path string) map[string][]string {
	g, err := release.ReadGenerations(path)
	if err != nil || g == nil {
		return nil
	}
	return g.Hosts
}

// waitForPublications waits until the generations file at path has had n
// publications (its seq). A wake goroutine's last write comes after the
// request it served has returned; a test whose file lives in a TempDir
// waits for it, or the write races the directory's removal.
func waitForPublications(t *testing.T, path string, n uint64) *release.Generations {
	t.Helper()
	var g *release.Generations
	waitFor(t, "the wake's last publication", func() bool {
		var err error
		g, err = release.ReadGenerations(path)
		return err == nil && g != nil && g.Seq >= n
	})
	return g
}

// unwritableGenerationsFile is a path whose parent is a regular file, so
// every write fails.
func unwritableGenerationsFile(t *testing.T) string {
	t.Helper()
	blocker := filepath.Join(t.TempDir(), "not-a-dir")
	if err := os.WriteFile(blocker, nil, 0o644); err != nil {
		t.Fatal(err)
	}
	return filepath.Join(blocker, "zdd", "generations.json")
}

// breakGenerationsFile makes every later write of path fail: its directory
// is replaced by a regular file.
func breakGenerationsFile(t *testing.T, path string) {
	t.Helper()
	dir := filepath.Dir(path)
	if err := os.RemoveAll(dir); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(dir, nil, 0o644); err != nil {
		t.Fatal(err)
	}
}

func TestTheGenerationsFileFollowsEveryChange(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	good := stagedID(t, dir, "v1")
	a, _, _ := planActivator(t, WakePlan{Boot: good, Mode: ModeOverlap})
	a.GenerationsFile = release.GenerationsPath(dir)
	if err := a.persistGenerations(); err != nil {
		t.Fatal(err)
	}
	g, err := release.ReadGenerations(a.GenerationsFile)
	if err != nil || g == nil || len(g.Hosts) != 0 {
		t.Fatalf("before any spawn: %+v, %v; want an empty file, written", g, err)
	}
	done, err := a.beginRequest(context.Background(), genTestRoute(t, "one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	// The wake goroutine republishes after the request it served has
	// returned, and a reader that lands between a publication's intent and
	// its file sees an incomplete one by design (release.ReadGenerations,
	// plan 4 R19). So wait for a complete publication that names the awake
	// generation, instead of reading once.
	var lastErr error
	waitFor(t, "a complete publication naming the awake generation", func() bool {
		g, lastErr = release.ReadGenerations(a.GenerationsFile)
		return lastErr == nil && g != nil && slices.Equal(g.Hosts["one.example.com"], []string{"v1"}) &&
			slices.Equal(g.Paths["one.example.com"], []string{good.StagedPath})
	})
	a.BounceFor("one.example.com", CauseRestart)
	// Four publications: the empty file, the plan before the lease, the
	// awake generation, the stop (in either order with the third).
	if g := waitForPublications(t, a.GenerationsFile, 4); len(g.Hosts) != 0 || len(g.Paths) != 0 {
		t.Fatalf("after the stop: %+v, want no generations", g)
	}
}

func TestTheGenerationsFileNamesThePlannedBuildBeforeItSpawns(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
	a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
	a.GenerationsFile = filepath.Join(dir, "zdd", "generations.json")
	b.gate = make(chan struct{})
	woke := make(chan error, 1)
	go func() {
		done, err := a.beginRequest(context.Background(), genTestRoute(t, "one.example.com"))
		if err == nil {
			done()
		}
		woke <- err
	}()
	waitFor(t, "the spawning build and its fallback pinned before the spawn", func() bool {
		return len(b.started()) == 0 && slices.Equal(hostsOf(a.GenerationsFile)["one.example.com"], []string{"v1", "v2"})
	})
	close(b.gate)
	if err := <-woke; err != nil {
		t.Fatal(err)
	}
	waitForPublications(t, a.GenerationsFile, 2) // the plan, then the awake generation
}

// Spec §7.4, §8.1: a planned build the selection gate refuses after the
// restore is planned again, and the replanned build is published before it
// execs; when it cannot be, it is not exec'd, and that is no failure of
// either build.
func TestAReplannedBuildIsPublishedBeforeItExecs(t *testing.T) {
	t.Parallel()
	for _, broken := range []bool{false, true} {
		name := "published"
		if broken {
			name = "unpublished"
		}
		t.Run(name, func(t *testing.T) {
			dir := t.TempDir()
			v2, v3 := stagedID(t, dir, "v2"), stagedID(t, dir, "v3")
			a, b, evs := planActivator(t, WakePlan{})
			a.GenerationsFile = filepath.Join(dir, "zdd", "generations.json")
			var plans atomic.Int32
			a.Planner = func(context.Context, registry.Route) WakePlan {
				if plans.Add(1) == 1 {
					return WakePlan{Boot: v2, Mode: ModeOverlap, Check: func(build string) error {
						if build != "v2" {
							return nil
						}
						if broken {
							breakGenerationsFile(t, a.GenerationsFile)
						}
						return errors.New("v2 is held now")
					}}
				}
				return WakePlan{Boot: v3, Mode: ModeOverlap}
			}
			if broken {
				_, err := a.beginRequest(context.Background(), genTestRoute(t, "one.example.com"))
				if !errors.Is(err, errStagedBuildNotPublished) {
					t.Fatalf("err = %v, want errStagedBuildNotPublished", err)
				}
				if gens := b.started(); len(gens) != 0 {
					t.Fatalf("spawned %+v with the replanned build unpublished", gens)
				}
				if k := evs.kinds(); len(k) != 0 {
					t.Fatalf("events %v: a publication failure is no failure of the build", k)
				}
				return
			}
			b.gate = make(chan struct{})
			woke := make(chan error, 1)
			go func() {
				done, err := a.beginRequest(context.Background(), genTestRoute(t, "one.example.com"))
				if err == nil {
					done()
				}
				woke <- err
			}()
			waitFor(t, "the replanned v3 pinned before its exec", func() bool {
				return len(b.started()) == 0 && slices.Contains(hostsOf(a.GenerationsFile)["one.example.com"], "v3")
			})
			close(b.gate)
			if err := <-woke; err != nil {
				t.Fatal(err)
			}
			if gens := b.started(); len(gens) != 1 || gens[0].Build != "v3" {
				t.Fatalf("spawned %+v, want the replanned v3 alone", gens)
			}
			waitForPublications(t, a.GenerationsFile, 3) // the plan, the replan, the awake generation
		})
	}
}

// Spec §4.2, §8.5: a failed wake whose generation is not proven terminated
// keeps its lease and its process may still run (plan 3, unprovenWakes),
// though its entry is gone. Its build stays in the generations file until
// the proof holds.
func TestARetainedFailedWakeStaysPinned(t *testing.T) {
	a, _, _ := planActivator(t, WakePlan{})
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	a.mu.Lock()
	a.unprovenWakes = map[string]*unprovenWake{"one.example.com": {gen: Gen{N: 4, Build: "v2", StagedPath: "/opt/carlos/versions/v2/app"}}}
	a.mu.Unlock()
	if err := a.persistGenerations(); err != nil {
		t.Fatal(err)
	}
	g, err := release.ReadGenerations(a.GenerationsFile)
	if err != nil || g == nil || !slices.Equal(g.Hosts["one.example.com"], []string{"v2"}) ||
		!slices.Equal(g.Paths["one.example.com"], []string{"/opt/carlos/versions/v2/app"}) {
		t.Fatalf("generations %+v, %v: want the retained failed wake's v2 pinned", g, err)
	}
}

// Spec §8.5: every generation of the operation set is named, the successor
// before its exec, and the retiring one until its termination is proven.
func TestTheGenerationsFileFollowsAnOverlapSwap(t *testing.T) {
	t.Parallel()
	f := newSwapFixture(t)
	f.a.GenerationsFile = release.GenerationsPath(t.TempDir())
	gate := make(chan struct{})
	f.b.gmu.Lock()
	f.b.genGate = gate
	f.b.gmu.Unlock()
	res := make(chan SwapResult, 1)
	go func() { res <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
	waitFor(t, "the successor pinned beside the incumbent before its exec", func() bool {
		gens, _, _ := f.b.calls()
		return gens == 1 && slices.Equal(hostsOf(f.a.GenerationsFile)[f.host], []string{"v1", "v2"})
	})
	close(gate)
	if r := <-res; r.Outcome != SwapDone {
		t.Fatalf("Swap = %+v, want done", r)
	}
	waitForGenerations(t, f.a.GenerationsFile, map[string][]string{f.host: {"v2"}})
}

// Spec §4.2, §8.5: root pins only what the file names. A staged instance
// spawn whose build could not be published must not exec it. The refusal is
// no failure of the build, takes no lease, and is never down.
func TestAnUnpublishedBuildIsNeverSpawned(t *testing.T) {
	t.Parallel()
	t.Run("planned wake", func(t *testing.T) {
		dir := t.TempDir()
		good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
		a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
		a.GenerationsFile = unwritableGenerationsFile(t)
		_, err := a.beginRequest(context.Background(), genTestRoute(t, "one.example.com"))
		if !errors.Is(err, errStagedBuildNotPublished) {
			t.Fatalf("err = %v, want errStagedBuildNotPublished", err)
		}
		if gens := b.started(); len(gens) != 0 {
			t.Fatalf("spawned %+v with the generations file unwritten", gens)
		}
		if acquires, _, _ := b.counts(); acquires != 0 {
			t.Fatal("an unpublished wake took the lease")
		}
		if k := evs.kinds(); len(k) != 0 {
			t.Fatalf("events %v: a publication failure is no failure of the build", k)
		}
	})
	t.Run("overlap successor", func(t *testing.T) {
		f := newSwapFixture(t)
		f.a.GenerationsFile = unwritableGenerationsFile(t)
		res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest)
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errStagedBuildNotPublished.Error()) {
			t.Fatalf("Swap = %+v, want abandoned: unpublished", res)
		}
		if gens, raises, restores := f.b.calls(); gens != 0 || raises != restores {
			t.Fatalf("spawned %d, raises %d, restores %d", gens, raises, restores)
		}
		f.a.mu.Lock()
		members := len(f.a.entries[f.host].ops.members)
		f.a.mu.Unlock()
		if members != 0 {
			t.Fatalf("%d members left in the operation set", members)
		}
	})
	t.Run("hold's pinned wake", func(t *testing.T) {
		f := newSwapFixture(t)
		sink := &lockedSink{}
		f.a.Logs = sink
		f.a.GenerationsFile = unwritableGenerationsFile(t)
		res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest)
		if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errStagedBuildNotPublished.Error()) {
			t.Fatalf("Swap = %+v, want abandoned: unpublished", res)
		}
		if gens, _, _ := f.b.calls(); gens != 0 {
			t.Fatalf("spawned %d generations with the pinned build unpublished", gens)
		}
		if got := sink.errors(); len(got) != 0 {
			t.Fatalf("member error events %q, want none for an abandoned hold", got)
		}
	})
	t.Run("claimed planned wake", func(t *testing.T) {
		dir := t.TempDir()
		good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
		a, admits, prepares, evs := claimedPlanActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
		a.GenerationsFile = unwritableGenerationsFile(t)
		_, err := a.beginRequest(context.Background(), registry.Route{Host: "x.example", Kind: "instance", Hibernate: true})
		if !errors.Is(err, errStagedBuildNotPublished) {
			t.Fatalf("err = %v, want errStagedBuildNotPublished", err)
		}
		if admits.Load() != 0 || prepares.Load() != 0 {
			t.Fatalf("admitted %d, prepared %d: an unpublished claimed wake reached the admission", admits.Load(), prepares.Load())
		}
		if k := evs.kinds(); len(k) != 0 {
			t.Fatalf("events %v: a publication failure is no failure of the build", k)
		}
	})
}

// claimedPlanActivator is a claims-on Service whose planner answers plan,
// as TestAPlannedClaimedWakeRefusesAtTheLaunchRecordingNothing builds it,
// counting the admissions and preparations a wake reaches, and logging the
// generation events it records.
func claimedPlanActivator(t *testing.T, plan WakePlan) (*Service, *atomic.Int32, *atomic.Int32, *genLog) {
	t.Helper()
	const host = "x.example"
	obs := claimedObs(host)
	obs.Claim.Op.Exec = "box"
	obs.Claim.Holder.WriterBox = "box"
	admitted := obs
	claim := *obs.Claim
	holder := *claim.Holder
	holder.Session, holder.Generation = 2, "two"
	claim.Holder = &holder
	admitted.Claim = &claim
	work, _ := (hostwrite.Config{}).Select(context.Background(), host, "spawn")
	var (
		reads    atomic.Int32
		admits   atomic.Int32
		prepares atomic.Int32
	)
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: time.Hour, HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box",
		Read: func(context.Context, string) (hostclaim.Observed, error) {
			if reads.Add(1) == 1 {
				return obs, nil
			}
			return admitted, nil
		},
		StopForeign:     func(context.Context, string) error { return nil },
		FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil },
		Prerequisites:   func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil },
		Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
			admits.Add(1)
			return admitted, nil
		},
		Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) {
			prepares.Add(1)
			return claimedStartFor(admitted, work, p.Pin.Build), nil
		}}}, &fakeBackend{}, reg)
	evs := &genLog{}
	a.OnGen = evs.add
	a.Planner = func(context.Context, registry.Route) WakePlan { return plan }
	return a, &admits, &prepares, evs
}

// The flag-independent behaviour (rulings R17, R19): with generations off
// the file says exactly what plan 4's writer said. A worker's captured build
// is listed by version and by staged path; an instance, which runs from the
// live link, adds nothing.
func TestTheGenerationsFileIsUnchangedWithTheFlagOff(t *testing.T) {
	t.Parallel()
	a, _, _, _, _ := stagedActivator(t) // generations off, no planner
	a.GenerationsFile = release.GenerationsPath(t.TempDir())
	const bot = "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(bot)); err != nil {
		t.Fatal(err)
	}
	done, err := a.beginRequest(context.Background(), genTestRoute(t, "one.example.com"))
	if err != nil {
		t.Fatal(err)
	}
	done()
	a.mu.Lock()
	instGen := a.entries["one.example.com"].gen
	a.mu.Unlock()
	if instGen.Build == "" {
		t.Fatal("the instance's generation names no build: this test would prove nothing")
	}
	want := map[string][]string{bot: {"v1"}}
	wantPaths := map[string][]string{bot: {fakeIdentity("v1").StagedPath}}
	check := func(when string) {
		t.Helper()
		if err := a.persistGenerations(); err != nil {
			t.Fatal(err)
		}
		g := readGenerationsFile(t, a.GenerationsFile)
		if !reflect.DeepEqual(g.Hosts, want) || !reflect.DeepEqual(g.Paths, wantPaths) {
			t.Fatalf("%s: flag-off file names %v and paths %v, want %v and %v", when, g.Hosts, g.Paths, want, wantPaths)
		}
	}
	// With the instance awake, after a sweep, and after its stop: none of
	// it may add the instance.
	check("both awake")
	a.sweepOnce()
	check("after a sweep")
	a.mu.Lock()
	a.beginStopLocked(a.entries["one.example.com"], StopIdle)
	a.mu.Unlock()
	waitFor(t, "the instance's stop", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.entries["one.example.com"] == nil
	})
	check("after the instance's stop")
}

// Task 11's socketless attempt names its build and its fallback while it
// runs (its fallback above all: startWorker publishes only the build about
// to exec).
func TestAWorkerAttemptsBuildsArePublished(t *testing.T) {
	a, _, _ := planActivator(t, WakePlan{})
	a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json")
	a.mu.Lock()
	a.entries["bot.x.internal"] = &actEntry{host: "bot.x.internal", worker: true, attemptBuilds: []string{"v3", "v1"}}
	a.mu.Unlock()
	if err := a.persistGenerations(); err != nil {
		t.Fatal(err)
	}
	if got := hostsOf(a.GenerationsFile)["bot.x.internal"]; !slices.Equal(got, []string{"v1", "v3"}) {
		t.Fatalf("attempt builds published as %v, want [v1 v3]", got)
	}
}

// The index contract: the generations file is written before the agent's
// first spawn, whatever CARLOS_SWAP says. A file left by a previous
// incarnation is stale (it may over-pin, or have been cut short by a crash,
// which root reads as incomplete and answers by neither restaging nor
// sweeping), so Start replaces it with this incarnation's snapshot, which is
// empty: nothing has spawned yet.
func TestStartReplacesAStaleGenerationsFile(t *testing.T) {
	t.Parallel()
	stale := func(t *testing.T, path string) {
		t.Helper()
		if err := release.WriteGenerations(path, release.Generations{
			Hosts: map[string][]string{"gone.example.com": {"v9"}},
			Paths: map[string][]string{"gone.example.com": {"/opt/carlos/versions/v9/app"}},
		}); err != nil {
			t.Fatal(err)
		}
	}
	for _, tc := range []struct {
		name  string
		setup func(t *testing.T, path string)
	}{
		{"a complete file from a previous run", stale},
		{"an intent with the file missing", func(t *testing.T, path string) {
			stale(t, path)
			if err := os.Remove(path); err != nil {
				t.Fatal(err)
			}
		}},
		{"an intent ahead of the file's seq", func(t *testing.T, path string) {
			stale(t, path)
			if err := os.WriteFile(release.GenerationsSeqPath(path), []byte("7\n"), 0o644); err != nil {
				t.Fatal(err)
			}
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			a, _, _, _, _ := stagedActivator(t) // generations off, no planner, no instances
			a.GenerationsFile = release.GenerationsPath(t.TempDir())
			tc.setup(t, a.GenerationsFile)
			a.Start()
			t.Cleanup(func() { a.Shutdown(context.Background()) })
			g, err := release.ReadGenerations(a.GenerationsFile)
			if err != nil || g == nil {
				t.Fatalf("after Start the file reads %+v, %v; want a complete publication", g, err)
			}
			if len(g.Hosts) != 0 || len(g.Paths) != 0 {
				t.Fatalf("after Start the file names %v and %v, want nothing", g.Hosts, g.Paths)
			}
		})
	}
}

// A write that fails at Start is logged and left to the sweep: it does not
// block or fail Start.
func TestStartDoesNotFailOnAnUnwritableGenerationsFile(t *testing.T) {
	a, _, _, _, _ := stagedActivator(t)
	a.GenerationsFile = unwritableGenerationsFile(t)
	a.Start()
	t.Cleanup(func() { a.Shutdown(context.Background()) })
	a.genMu.Lock()
	dirty := a.genDirty
	a.genMu.Unlock()
	if !dirty {
		t.Fatal("a failed write at Start did not mark the publication dirty for the sweep")
	}
}