package activator

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"strings"
	"testing"
	"time"

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

// readGenerations is the generations file's hosts, or nil when it is absent.
func readGenerations(t *testing.T, path string) map[string][]string {
	t.Helper()
	g, err := release.ReadGenerations(path)
	if err != nil {
		t.Fatalf("ReadGenerations: %v", err)
	}
	if g == nil {
		return nil
	}
	return g.Hosts
}

func waitForGenerations(t *testing.T, path string, want map[string][]string) {
	t.Helper()
	waitFor(t, "generations file naming "+stringOf(want), func() bool {
		g, err := release.ReadGenerations(path)
		return err == nil && g != nil && reflect.DeepEqual(g.Hosts, want)
	})
}

func stringOf(m map[string][]string) string {
	var b strings.Builder
	for h, vs := range m {
		b.WriteString(h + "=" + strings.Join(vs, ",") + " ")
	}
	return b.String()
}

// Ruling R17: the file names a worker's captured build after its wake, keeps
// naming it across a crash recycle while live/<host> names another build,
// follows a converge, and drops it once the worker stops.
func TestWorkerGenerationsFileFollowsTheCapturedBuild(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	path := release.GenerationsPath(t.TempDir())
	a.GenerationsFile = path
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := readGenerations(t, path); !reflect.DeepEqual(got, map[string][]string{host: {"v1"}}) {
		t.Fatalf("after the wake the file names %v; want %s: [v1]", got, host)
	}

	ver.Store("v2") // root installs v2 under the running child
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b.fakeBackend, host, 2)
	if calls := b.calls(); calls[1] == nil || *calls[1] != fakeIdentity("v1") {
		t.Fatalf("crash recycle asked for %v; want the captured v1", calls)
	}
	waitForGenerations(t, path, map[string][]string{host: {"v1"}})

	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
	waitForGenerations(t, path, map[string][]string{host: {"v2"}})

	// The stop every path takes (a hibernation, a stand-down, a shutdown).
	a.mu.Lock()
	a.beginStopLocked(a.entries[host], StopIdle)
	a.mu.Unlock()
	waitFor(t, "the worker's stop", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.entries[host] == nil
	})
	waitForGenerations(t, path, map[string][]string{})
}

// A captured build with no version, or a version name the store could never
// hold, is not listed by version, which would make the file unwritable; it
// is listed by its staged path instead (ruling R19 item 2).
func TestWorkerGenerationsListOnlyVersionStoreBuilds(t *testing.T) {
	t.Parallel()
	a, _, _, _, ver := stagedActivator(t)
	path := release.GenerationsPath(t.TempDir())
	a.GenerationsFile = path
	ver.Store("")
	if err := a.WakeWorker(alwaysOnWorkerRoute("empty.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("not/a/version")
	if err := a.WakeWorker(alwaysOnWorkerRoute("odd.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v3")
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := readGenerations(t, path); !reflect.DeepEqual(got, map[string][]string{"bot.x.internal": {"v3"}}) {
		t.Fatalf("file names %v; want only bot.x.internal: [v3]", got)
	}
}

// realWorkerBackend is the fake's lease, restore and replication around a
// REAL ExecBackend worker spawn: the child is a process exec'd by staged path,
// and a recycle runs release.VerifyStaged on real bytes.
type realWorkerBackend struct {
	*fakeBackend
	exec *ExecBackend
}

func (b *realWorkerBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
	return b.exec.StartWorkerStaged(ctx, rt, env, onto)
}

// captureStdout runs fn with os.Stdout redirected, and returns what it
// printed: the adoption pass reports its refusals there.
func captureStdout(t *testing.T, fn func()) string {
	t.Helper()
	r, w, err := os.Pipe()
	if err != nil {
		t.Fatal(err)
	}
	orig := os.Stdout
	os.Stdout = w
	done := make(chan struct{})
	var buf bytes.Buffer
	go func() {
		buf.ReadFrom(r)
		close(done)
	}()
	func() {
		defer func() { os.Stdout = orig }()
		fn()
	}()
	w.Close()
	<-done
	return buf.String()
}

// Ruling R17, end to end (spec §4.2 "a staged copy that something refers to
// never changes underneath it"). A worker is captured on build A (v1). Root
// installs B and then C on its host, so the live link names C and the row
// says Version C, PrevVersion B: only the generations file still names A.
// A is then re-shipped under the same name with different bytes, and a pass
// wants it for another host. The pass refuses with ErrVersionReused and A's
// staged bytes do not change, so the worker's crash recycle onto A passes
// VerifyStaged and it keeps running.
func TestWorkerCapturedBuildSurvivesARestageOfItsReusedName(t *testing.T) {
	ctx := context.Background()
	eb := testExecBackend(t)
	opt, dataDir := eb.OptDir, t.TempDir()
	fb := &fakeBackend{}
	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}, &realWorkerBackend{fakeBackend: fb, exec: eb}, reg)
	a.liveVer = eb.LiveVersion
	a.GenerationsFile = release.GenerationsPath(dataDir)
	t.Cleanup(func() { a.Shutdown(context.Background()) })

	store := release.NewDirStore(t.TempDir())
	seed, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := release.ParseSigningKey(seed, "test")
	if err != nil {
		t.Fatal(err)
	}
	pub, err := release.ParsePublicKey(pubB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	plat := runtime.GOOS + "-" + runtime.GOARCH
	argv0 := filepath.Join(t.TempDir(), "argv0")
	build := func(name string) []byte {
		return []byte("#!/bin/sh\n# build " + name + "\nprintf '%s\\n' \"$0\" > '" + argv0 + ".tmp' && mv '" + argv0 + ".tmp' '" + argv0 + "'\nexec sleep 30\n")
	}
	ship := func(ver string, body []byte) {
		t.Helper()
		m := release.Manifest{App: "app", Version: ver, Kind: "binary", Artifacts: map[string]release.Artifact{plat: {}}}
		if err := release.Ship(ctx, store, "bab", m, map[string][]byte{plat: body}); err != nil {
			t.Fatalf("Ship %s: %v", ver, err)
		}
	}
	promote := func(channel, ver string) {
		t.Helper()
		if err := release.Promote(ctx, store, priv, "bab", "app", channel, ver); err != nil {
			t.Fatalf("Promote %s %s: %v", channel, ver, err)
		}
	}
	pass := func() string {
		t.Helper()
		return captureStdout(t, func() {
			if err := release.Pass(ctx, reg, store, pub, opt, func(string) error { return nil }, release.BakeDir(dataDir)); err != nil {
				t.Fatalf("Pass: %v", err)
			}
		})
	}

	host := "bot.example.com"
	rt := registry.Route{Host: host, Kind: "sidecar", App: "app", Channel: "canary/x", Args: `["sidecar","run"]`, Status: "ready"}
	if err := reg.Upsert(rt); err != nil {
		t.Fatal(err)
	}
	stagedA := release.VersionPath(opt, "bab", "v1", "app")

	// A (v1) is installed and the worker is captured on it.
	buildA := build("A")
	ship("v1", buildA)
	promote("canary/x", "v1")
	pass()
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	// The child's own startup write, waited out so it cannot land after the
	// removal below and pass for the recycle's.
	if got := strings.TrimSpace(waitForFile(t, argv0, 5*time.Second)); got != stagedA {
		t.Fatalf("the wake exec'd %q; want A's staged path %s", got, stagedA)
	}
	captured := a.entryServing(host)
	if captured.Version != "v1" || captured.StagedPath != stagedA {
		t.Fatalf("captured %+v; want v1 at %s", captured, stagedA)
	}
	if got := readGenerations(t, a.GenerationsFile); !reflect.DeepEqual(got, map[string][]string{host: {"v1"}}) {
		t.Fatalf("generations file names %v; want %s: [v1]", got, host)
	}

	// B (v2), then C (v3): the link and the row move on.
	ship("v2", build("B"))
	promote("canary/x", "v2")
	pass()
	ship("v3", build("C"))
	promote("canary/x", "v3")
	pass()
	row, err := reg.Get(host)
	if err != nil || row == nil || row.Version != "v3" || row.PrevVersion != "v2" {
		t.Fatalf("row = %+v, %v; want Version v3, PrevVersion v2", row, err)
	}
	if lv, _ := release.LiveVersion(opt, host); lv != "v3" {
		t.Fatalf("live = %q, want v3", lv)
	}
	rows, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	if pins, err := release.BoxPins(opt, rows); err != nil || pins.Has(stagedA) {
		t.Fatalf("BoxPins = %v, %v; want A unpinned by links and rows, so only the generations file protects it", pins, err)
	}

	// A is pruned from the store and re-shipped under the same name with
	// different bytes, and another host wants it.
	for _, key := range []string{release.ManifestKey("bab", "app", "v1"), release.ArtifactKey("bab", "app", "v1", plat)} {
		if err := store.Delete(ctx, key); err != nil {
			t.Fatalf("prune %s: %v", key, err)
		}
	}
	ship("v1", build("A, re-shipped"))
	// Hibernating, so the sweep's always-on arm never starts it: only the
	// pass acts on it here.
	other := registry.Route{Host: "bot2.example.com", Kind: "sidecar", App: "app", Channel: "canary/y", Args: `["sidecar","run"]`, Status: "ready", Hibernate: true}
	if err := reg.Upsert(other); err != nil {
		t.Fatal(err)
	}
	promote("canary/y", "v1")
	out := pass()
	if !strings.Contains(out, release.ErrVersionReused.Error()) {
		t.Fatalf("pass printed %q; want the refusal %q", out, release.ErrVersionReused)
	}
	if b, err := os.ReadFile(stagedA); err != nil || !bytes.Equal(b, buildA) {
		t.Fatalf("A's staged copy = %q, %v; want A's bytes untouched", b, err)
	}
	if r2, _ := reg.Get(other.Host); r2 == nil || r2.Version == "v1" {
		t.Fatalf("%s row = %+v; the reused name must not install", other.Host, r2)
	}

	// The worker crashes, and its recycle onto A verifies and runs.
	os.Remove(argv0)
	a.mu.Lock()
	old := a.entries[host].proc
	a.mu.Unlock()
	if err := old.Stop(ctx); err != nil {
		t.Fatalf("crash the worker: %v", err)
	}
	a.sweepOnce()
	if got := strings.TrimSpace(waitForFile(t, argv0, 5*time.Second)); got != stagedA {
		t.Fatalf("the recycle exec'd %q; want A's staged path %s", got, stagedA)
	}
	waitFor(t, "the recycle to land", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		e := a.entries[host]
		return e != nil && !e.workerRespawning && e.proc != old
	})
	a.mu.Lock()
	e := a.entries[host]
	serving, proc := e.workerServing, e.proc
	a.mu.Unlock()
	if serving != captured {
		t.Fatalf("serving after the recycle = %+v; want the captured A %+v", serving, captured)
	}
	time.Sleep(100 * time.Millisecond)
	if proc.Exited() {
		t.Fatal("the worker recycled onto A is not running")
	}
}

// readGenerationsFile is the whole generations file; it fails the test when
// the file does not read.
func readGenerationsFile(t *testing.T, path string) *release.Generations {
	t.Helper()
	g, err := release.ReadGenerations(path)
	if err != nil {
		t.Fatalf("ReadGenerations: %v", err)
	}
	if g == nil {
		return &release.Generations{}
	}
	return g
}

func describeGenerations(gs []*release.Generations) string {
	var b strings.Builder
	for i, g := range gs {
		fmt.Fprintf(&b, "[%d] hosts %v paths %v ", i, g.Hosts, g.Paths)
	}
	return b.String()
}

// Ruling R19 item 3 (Astra r3 F1(c)): the build a spawn is about to exec is
// in the generations file BEFORE the exec, for a wake (from the live link)
// and for a converge onto a newly installed build. Once the spawn has
// installed, the captured build replaces the pending one.
func TestWorkerSpawnPublishesItsBuildBeforeTheExec(t *testing.T) {
	t.Parallel()
	a, b, clk, _, ver := stagedActivator(t)
	path := release.GenerationsPath(t.TempDir())
	a.GenerationsFile = path
	host := "bot.x.internal"
	var atExec []*release.Generations
	b.workerStartHook = func(registry.Route) {
		g, err := release.ReadGenerations(path)
		if err != nil || g == nil {
			t.Errorf("at the exec the generations file = %+v, %v; want it written", g, err)
			g = &release.Generations{}
		}
		atExec = append(atExec, g)
	}
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if len(atExec) != 1 || !reflect.DeepEqual(atExec[0].Hosts[host], []string{"v1"}) || !reflect.DeepEqual(atExec[0].Paths[host], []string{fakeIdentity("v1").StagedPath}) {
		t.Fatalf("at the wake's exec the file named %s; want %s: v1 by version and path", describeGenerations(atExec), host)
	}

	ver.Store("v2")
	a.sweepOnce() // the drift clock starts
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
	if len(atExec) != 2 || !reflect.DeepEqual(atExec[1].Hosts[host], []string{"v1", "v2"}) {
		t.Fatalf("at the converge's exec the file named %s; want %s: the serving v1 and the pending v2", describeGenerations(atExec), host)
	}
	waitForGenerations(t, path, map[string][]string{host: {"v2"}})
	if g := readGenerationsFile(t, path); !reflect.DeepEqual(g.Paths, map[string][]string{host: {fakeIdentity("v2").StagedPath}}) {
		t.Fatalf("after the converge the file's paths = %v; want only v2's", g.Paths)
	}
}

// Ruling R19 items 3 and 4: when the build cannot be published first, the
// spawn does not happen, for a wake and for a crash recycle. Once the file
// can be written again, the sweep's retry publishes the set without the
// refused build.
func TestWorkerSpawnRefusedWhenItsBuildCannotBePublished(t *testing.T) {
	t.Parallel()
	a, b, _, _, _ := stagedActivator(t)
	data := filepath.Join(t.TempDir(), "data")
	if err := os.WriteFile(data, []byte("a file where the data dir should be"), 0o644); err != nil {
		t.Fatal(err)
	}
	path := release.GenerationsPath(data)
	a.GenerationsFile = path
	if err := a.WakeWorker(alwaysOnWorkerRoute("first.x.internal")); err == nil || !errors.Is(err, ErrBuildNotPublished) {
		t.Fatalf("WakeWorker with an unwritable generations file = %v; want ErrBuildNotPublished", err)
	}
	if n := b.workerStarts(); n != 0 {
		t.Fatalf("workerStarts = %d; a build that could not be published must not be exec'd", n)
	}

	// The file becomes writable; a worker wakes. Then publication breaks
	// again and the child crashes: the recycle is not spawned either.
	if err := os.Remove(data); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(data, 0o755); err != nil {
		t.Fatal(err)
	}
	host := "bot.x.internal"
	if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d after the wake; want 1", n)
	}
	zdd := filepath.Dir(path)
	if err := os.RemoveAll(zdd); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(zdd, []byte("a file where the zdd dir should be"), 0o644); err != nil {
		t.Fatal(err)
	}
	asked := len(b.calls())
	b.lastWorker().kill()
	a.sweepOnce()
	waitFor(t, "the recycle attempt to finish", func() bool { return len(b.calls()) == asked+1 && !a.entryRespawning(host) })
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d after the recycle; a build that could not be published must not be exec'd", n)
	}

	if err := os.Remove(zdd); err != nil {
		t.Fatal(err)
	}
	a.sweepOnce()
	waitForGenerations(t, path, map[string][]string{host: {"v1"}})
}

// Ruling R19 item 2 (Astra r3 F1(b)), with a real ExecBackend: a worker
// exec'd through a chained live link (live/<host> -> operator-current ->
// versions/v1/app) has Version "", and is published by its staged path,
// which root's pin set then holds without any registry row.
func TestWorkerFromAChainedLinkIsPublishedByPath(t *testing.T) {
	t.Parallel()
	eb := testExecBackend(t)
	fb := &fakeBackend{}
	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}, &realWorkerBackend{fakeBackend: fb, exec: eb}, reg)
	a.liveVer = eb.LiveVersion
	a.GenerationsFile = release.GenerationsPath(t.TempDir())
	t.Cleanup(func() { a.Shutdown(context.Background()) })

	host := "bot.example.com"
	argv0 := filepath.Join(t.TempDir(), "argv0")
	staged := stageArgv0Worker(t, eb, host, "v1", argv0)
	current := filepath.Join(t.TempDir(), "operator-current")
	if err := os.Symlink(staged, current); err != nil {
		t.Fatal(err)
	}
	if err := os.Remove(eb.liveLink(host)); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(current, eb.liveLink(host)); err != nil {
		t.Fatal(err)
	}
	rt := registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`, Status: "ready"}
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := strings.TrimSpace(waitForFile(t, argv0, 5*time.Second)); got != staged {
		t.Fatalf("the wake exec'd %q; want %s", got, staged)
	}
	if id := a.entryServing(host); id.Version != "" || id.StagedPath != staged {
		t.Fatalf("captured %+v; want no Version and the staged path %s", id, staged)
	}
	g := readGenerationsFile(t, a.GenerationsFile)
	if len(g.Hosts) != 0 || !reflect.DeepEqual(g.Paths, map[string][]string{host: {staged}}) {
		t.Fatalf("generations file = hosts %v, paths %v; want only %s by path", g.Hosts, g.Paths, staged)
	}
	pins, err := release.GenerationPins(a.GenerationsFile, func() ([]registry.Route, error) { return nil, nil }).StagedPins(context.Background(), eb.OptDir)
	if err != nil || !pins.Has(staged) {
		t.Fatalf("root's generation pins = %v, %v; want %s pinned with no row", pins, err, staged)
	}
}