//go:build linux

package activator

import (
	"context"
	"encoding/json"
	"errors"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/instancefs"
	"golang.org/x/sys/unix"
)

type fakeRoleRoot struct {
	mu         sync.Mutex
	events     []string
	launches   []hostbox.RoleLaunchRequest
	stops      []hostbox.RoleLaunchRequest
	grant      func(hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error)
	stopErr    error
	capability string
}

func (f *fakeRoleRoot) Capability(context.Context) (string, error) { return f.capability, nil }

func (f *fakeRoleRoot) Launch(_ context.Context, q hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
	f.mu.Lock()
	f.events = append(f.events, "launch")
	f.launches = append(f.launches, q)
	grant := f.grant
	f.mu.Unlock()
	return grant(q)
}

func (f *fakeRoleRoot) Stop(_ context.Context, q hostbox.RoleLaunchRequest) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.events = append(f.events, "stop")
	f.stops = append(f.stops, q)
	return f.stopErr
}

func selfStart(t *testing.T) string {
	t.Helper()
	start, err := roleProcessStart(os.Getpid())
	if err != nil {
		t.Fatal(err)
	}
	return start
}

type rootLaunchFixture struct {
	launch     RoleLaunch
	root       *fakeRoleRoot
	attempt    hostclaim.WakeRetryAttempt
	build      func() *exec.Cmd
	exe        string
	exeID      hostbox.LaunchFile
	launchDir  string
	runDir     string
	logSocket  string
	beforeRuns int
}

func newRootLaunchFixture(t *testing.T) *rootLaunchFixture {
	t.Helper()
	target := hostguard.FenceTarget{Region: "eu", Host: "tenant.test", Incarnation: "inc", Stamp: 2, Session: 3, Box: "box", Enrollment: "enroll", BootID: "boot", ControllerID: "ctrl", ClaimETag: "etag", DirectoryRevision: "7"}
	work, err := (hostwrite.Config{}).Select(t.Context(), target.Host, "spawn")
	if err != nil {
		t.Fatal(err)
	}
	base := shortTempDir(t)
	f := &rootLaunchFixture{launchDir: filepath.Join(base, "data"), runDir: filepath.Join(base, "run"), logSocket: filepath.Join(base, "logs", "t.run.sock"), exe: filepath.Join(base, "app")}
	for _, d := range []string{f.launchDir, filepath.Dir(f.logSocket)} {
		if err := os.MkdirAll(d, 0o700); err != nil {
			t.Fatal(err)
		}
	}
	if err := os.WriteFile(f.exe, []byte("build"), 0o555); err != nil {
		t.Fatal(err)
	}
	var st unix.Stat_t
	if err := unix.Stat(f.exe, &st); err != nil {
		t.Fatal(err)
	}
	f.exeID = hostbox.LaunchFile{Device: uint64(st.Dev), Inode: st.Ino}
	f.attempt = hostclaim.WakeRetryAttempt{ID: "attempt-1", InvocationID: "inv-1", LaunchInputs: "inputs-1", Session: 3, Box: "box", Enrollment: "enroll", BootID: "boot", ControllerID: "ctrl"}
	start := selfStart(t)
	f.root = &fakeRoleRoot{capability: hostbox.RoleLaunchProtocol}
	f.root.grant = func(q hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
		return hostbox.RoleLaunchGrant{ReservationID: q.Reservation.ID, Unit: "carlos-instance@tenant.test.service", UnitInvocationID: "unit-1", PID: os.Getpid(), ProcessStart: start, Group: hostbox.LaunchGroup{RelativePath: q.Reservation.Group.RelativePath, Device: 3, Inode: 4, InvocationID: "unit-1"}}, nil
	}
	f.launch = RoleLaunch{Work: work, Target: target, Root: f.root, LaunchDir: f.launchDir, RunDir: f.runDir, LogSocket: f.logSocket, MemoryMax: cgroup.TierDefault.MemoryMax, Executable: func(context.Context, *exec.Cmd) (*os.File, error) { return os.Open(f.exe) }}
	f.build = func() *exec.Cmd {
		cmd := exec.Command("/opt/carlos/live/tenant.test", "--socket", filepath.Join(f.runDir, "app.sock"), "--db", filepath.Join(f.launchDir, "app.db"))
		cmd.Env = []string{"CARLOS_ADMIN_TOKEN=secret", "A=1"}
		return cmd
	}
	return f
}

func (f *rootLaunchFixture) before() error {
	f.root.mu.Lock()
	defer f.root.mu.Unlock()
	f.root.events = append(f.root.events, "before")
	f.beforeRuns++
	return nil
}

func TestRootRoleLaunchCarriesTheReservation(t *testing.T) {
	f := newRootLaunchFixture(t)
	finished := 0
	f.launch.RegisterProducer = func(_ context.Context, role string) (RoleProducer, error) {
		if role != "main" {
			t.Fatalf("producer role %q", role)
		}
		return RoleProducer{Context: t.Context(), Finish: func(error) { finished++ }}, nil
	}
	cmd, prepared, settle, producer, err := f.launch.Start(t.Context(), "main", f.attempt, f.build, f.before)
	if err != nil {
		t.Fatal(err)
	}
	if cmd == nil || cmd.Process == nil || cmd.Process.Pid != os.Getpid() || prepared != nil || settle == nil || producer.Finish == nil {
		t.Fatalf("start returned cmd=%v prepared=%v settle=%v", cmd, prepared, settle != nil)
	}
	if got := strings.Join(f.root.events, ","); got != "before,launch" {
		t.Fatalf("order=%s", got)
	}
	q := f.root.launches[0]
	r := q.Reservation
	if q.V != hostbox.RoleLaunchV || q.Op != hostbox.RoleLaunchStart || q.Executable != f.exe || q.MemoryMax != cgroup.TierDefault.MemoryMax || r == nil {
		t.Fatalf("request=%+v", q)
	}
	want := hostbox.LaunchReservation{ID: "attempt-1:main", Role: "main", InvocationID: "inv-1", Inputs: "inputs-1", Target: f.launch.Target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup("tenant.test")}, Executable: f.exeID}
	if *r != want {
		t.Fatalf("reservation=%+v want %+v", *r, want)
	}
	raw, err := instancefs.ReadBeneath(f.launchDir, instancectl.LaunchFileName(instancectl.RoleRun))
	if err != nil {
		t.Fatal(err)
	}
	var file instancectl.Launch
	if err := json.Unmarshal(raw, &file); err != nil {
		t.Fatal(err)
	}
	if file.Schema != instancectl.LaunchSchema || file.Host != "tenant.test" || strings.Join(file.Argv, " ") != "/opt/carlos/live/tenant.test --socket "+filepath.Join(f.runDir, "app.sock")+" --db "+filepath.Join(f.launchDir, "app.db") || strings.Join(file.Env, ",") != "CARLOS_ADMIN_TOKEN=secret,A=1" {
		t.Fatalf("launch file=%+v", file)
	}
	if info, err := os.Stat(filepath.Join(f.launchDir, instancectl.LaunchFileName(instancectl.RoleRun))); err != nil || info.Mode().Perm() != 0o600 {
		t.Fatalf("launch file mode=%v %v", info, err)
	}
	// The unit's stdout socket listens before root starts it.
	conn, err := net.Dial("unix", f.logSocket)
	if err != nil {
		t.Fatalf("log listener: %v", err)
	}
	conn.Close()
	if err := settle(t.Context()); err != nil {
		t.Fatal(err)
	}
	if len(f.root.stops) != 1 || f.root.stops[0].Op != hostbox.RoleLaunchStop || *f.root.stops[0].Reservation != want {
		t.Fatalf("stops=%+v", f.root.stops)
	}
	if _, err := net.Dial("unix", f.logSocket); err == nil {
		t.Fatal("log listener outlived the settled role")
	}
}

func TestRootRoleLaunchRefusesShapesItCannotCarry(t *testing.T) {
	for name, mutate := range map[string]func(*rootLaunchFixture, *string){
		"seed role":    func(_ *rootLaunchFixture, role *string) { *role = "seed" },
		"sidecar role": func(_ *rootLaunchFixture, role *string) { *role = "sidecar" },
		"socket elsewhere": func(f *rootLaunchFixture, _ *string) {
			f.build = argsBuild("--socket", "/run/carlos/other.sock", "--db", filepath.Join(f.launchDir, "app.db"))
		},
		"db elsewhere": func(f *rootLaunchFixture, _ *string) {
			f.build = argsBuild("--socket", filepath.Join(f.runDir, "app.sock"), "--db", "/data/other.db")
		},
		"no socket": func(f *rootLaunchFixture, _ *string) {
			f.build = argsBuild("--db", filepath.Join(f.launchDir, "app.db"))
		},
		"extra flag": func(f *rootLaunchFixture, _ *string) {
			f.build = argsBuild("--socket", filepath.Join(f.runDir, "app.sock"), "--db", filepath.Join(f.launchDir, "app.db"), "--exec", "/bin/sh")
		},
		"no launch dir":      func(f *rootLaunchFixture, _ *string) { f.launch.LaunchDir = "" },
		"no log socket":      func(f *rootLaunchFixture, _ *string) { f.launch.LogSocket = "relative.sock" },
		"no tier":            func(f *rootLaunchFixture, _ *string) { f.launch.MemoryMax = 0 },
		"attempt session":    func(f *rootLaunchFixture, _ *string) { f.attempt.Session = 4 },
		"attempt controller": func(f *rootLaunchFixture, _ *string) { f.attempt.ControllerID = "ctrl-2" },
		"attempt id":         func(f *rootLaunchFixture, _ *string) { f.attempt.ID = "" },
	} {
		f := newRootLaunchFixture(t)
		role := "main"
		mutate(f, &role)
		cmd, _, settle, _, err := f.launch.Start(t.Context(), role, f.attempt, f.build, f.before)
		if err == nil || cmd != nil || settle != nil {
			t.Fatalf("%s started: %v", name, err)
		}
		if len(f.root.launches) != 0 || f.beforeRuns != 0 {
			t.Fatalf("%s reached root or the before-start hook", name)
		}
		if _, err := os.Stat(filepath.Join(f.launchDir, instancectl.LaunchFileName(instancectl.RoleRun))); err == nil {
			t.Fatalf("%s wrote a launch file", name)
		}
	}
}

func argsBuild(args ...string) func() *exec.Cmd {
	return func() *exec.Cmd { return exec.Command("/opt/carlos/live/tenant.test", args...) }
}

func TestRootRoleLaunchBurnsEveryLaunchItDoesNotHold(t *testing.T) {
	// A refusal or an unanswered launch is followed by a stop of the same
	// reservation, so a late or uncertain start can never run unowned.
	f := newRootLaunchFixture(t)
	finished := []error{}
	f.launch.RegisterProducer = func(context.Context, string) (RoleProducer, error) {
		return RoleProducer{Context: t.Context(), Finish: func(err error) { finished = append(finished, err) }}, nil
	}
	f.root.grant = func(hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
		return hostbox.RoleLaunchGrant{}, errors.New("connection reset")
	}
	cmd, _, settle, _, err := f.launch.Start(t.Context(), "main", f.attempt, f.build, f.before)
	if err == nil || cmd != nil || settle != nil || len(f.root.stops) != 1 || *f.root.stops[0].Reservation != *f.root.launches[0].Reservation {
		t.Fatalf("unanswered launch: err=%v settle=%v stops=%d", err, settle != nil, len(f.root.stops))
	}
	if len(finished) != 1 || finished[0] != nil {
		t.Fatalf("settled producer retained a launch error: %v", finished)
	}
	// When the burning stop also fails, the caller keeps a settle to retry.
	f = newRootLaunchFixture(t)
	f.root.grant = func(hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
		return hostbox.RoleLaunchGrant{}, errors.New("connection reset")
	}
	finished = nil
	f.launch.RegisterProducer = func(context.Context, string) (RoleProducer, error) {
		return RoleProducer{Context: t.Context(), Finish: func(err error) { finished = append(finished, err) }}, nil
	}
	f.root.stopErr = errors.New("root unreachable")
	_, _, settle, _, err = f.launch.Start(t.Context(), "main", f.attempt, f.build, f.before)
	if err == nil || settle == nil {
		t.Fatalf("uncertain launch dropped its cleanup: %v", err)
	}
	if len(finished) != 0 {
		t.Fatal("unproven unit finished its producer")
	}
	f.root.stopErr = nil
	if err := settle(t.Context()); err != nil || len(finished) != 1 || finished[0] != nil {
		t.Fatalf("settle retry did not complete producer: %v %v", err, finished)
	}
	// A grant for another process or reservation is never adopted.
	for name, bad := range map[string]func(*hostbox.RoleLaunchGrant){
		"other reservation": func(g *hostbox.RoleLaunchGrant) { g.ReservationID = "attempt-2:main" },
		"reused pid":        func(g *hostbox.RoleLaunchGrant) { g.ProcessStart = "1" },
		"other group":       func(g *hostbox.RoleLaunchGrant) { g.Group.RelativePath = "carlos.slice/x.service" },
	} {
		f := newRootLaunchFixture(t)
		good := f.root.grant
		f.root.grant = func(q hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
			g, err := good(q)
			bad(&g)
			return g, err
		}
		cmd, _, _, _, err := f.launch.Start(t.Context(), "main", f.attempt, f.build, f.before)
		if err == nil || cmd != nil || len(f.root.stops) != 1 {
			t.Fatalf("%s adopted: %v stops=%d", name, err, len(f.root.stops))
		}
	}
}

func TestRootClaimProcessStopsThroughRootAndWatchesExit(t *testing.T) {
	child := exec.Command("sleep", "30")
	if err := child.Start(); err != nil {
		t.Skip("no sleep binary:", err)
	}
	reaped := make(chan struct{})
	go func() { _ = child.Wait(); close(reaped) }()
	proc, err := os.FindProcess(child.Process.Pid)
	if err != nil {
		t.Fatal(err)
	}
	cmd := &exec.Cmd{Path: "/opt/carlos/live/tenant.test", Process: proc}
	p := newClaimProcessFor(cmd, nil, true)
	if p.Exited() {
		t.Fatal("running root process reported exited")
	}
	stopped := 0
	p.settle = func(context.Context) error {
		stopped++
		return child.Process.Signal(syscall.SIGKILL)
	}
	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
	defer cancel()
	if err := p.Stop(ctx); err != nil {
		t.Fatal(err)
	}
	<-reaped
	if stopped != 1 || !p.Exited() {
		t.Fatalf("stopped=%d exited=%v", stopped, p.Exited())
	}
	if err := p.Stop(ctx); err != nil || stopped != 1 {
		t.Fatalf("second stop=%v stopped=%d", err, stopped)
	}
	// Without a root settle there is no way to stop it: refuse, never assume.
	other := exec.Command("sleep", "30")
	if err := other.Start(); err != nil {
		t.Fatal(err)
	}
	defer func() { _ = other.Process.Kill(); _ = other.Wait() }()
	proc, _ = os.FindProcess(other.Process.Pid)
	q := newClaimProcessFor(&exec.Cmd{Process: proc}, nil, true)
	if err := q.Stop(ctx); !errors.Is(err, hostwrite.ErrUnavailable) {
		t.Fatalf("unsettleable root process: %v", err)
	}
}

func TestRootRoleHealthAndPeerReadTheUnitReservation(t *testing.T) {
	raw, err := os.ReadFile("/proc/self/cgroup")
	if err != nil || !strings.HasPrefix(string(raw), "0::/") {
		t.Skip("no unified cgroup membership")
	}
	rel := strings.TrimSpace(strings.TrimPrefix(strings.SplitN(string(raw), "\n", 2)[0], "0::/"))
	if !hostbox.UnitGroupPath(rel) {
		t.Skip("test process is not in a service cgroup:", rel)
	}
	var st unix.Stat_t
	if err := unix.Stat(filepath.Join("/sys/fs/cgroup", rel), &st); err != nil {
		t.Skip("cgroup tree not visible:", err)
	}
	ctx := t.Context()
	journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launch.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer journal.Close()
	f := newRootLaunchFixture(t)
	target := f.launch.Target
	if err := journal.Prelaunch(ctx, hostbox.LaunchInventory{ControllerPID: os.Getpid(), ProcessStart: selfStart(t), Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/parent", Device: 1, Inode: 2}}, NoState: true}); err != nil {
		t.Fatal(err)
	}
	r := hostbox.LaunchReservation{ID: "attempt-1:main", Role: "main", InvocationID: "inv-1", Inputs: "inputs-1", Target: target, Group: hostbox.LaunchGroup{RelativePath: rel}, Executable: f.exeID}
	if err := journal.IntendUnit(ctx, r); err != nil {
		t.Fatal(err)
	}
	if err := journal.StartedUnit(ctx, r, hostbox.LaunchGroup{RelativePath: rel, Device: uint64(st.Dev), Inode: st.Ino, InvocationID: "unit-1"}, os.Getpid(), selfStart(t)); err != nil {
		t.Fatal(err)
	}
	proc, _ := os.FindProcess(os.Getpid())
	cmd := &exec.Cmd{Process: proc}
	launch := RoleLaunch{Journal: journal, Target: target, CgroupRoot: "/sys/fs/cgroup", Root: f.root}
	if err := launch.ExactHealth(ctx, cmd, f.attempt); err != nil {
		t.Fatalf("exact health: %v", err)
	}
	a, b, err := socketPair()
	if err != nil {
		t.Fatal(err)
	}
	defer a.Close()
	defer b.Close()
	if err := launch.PeerInRole(ctx, a, cmd); err != nil {
		t.Fatalf("peer in role: %v", err)
	}
	if err := (RoleLaunch{Journal: journal, Target: target, CgroupRoot: "/sys/fs/cgroup"}).PeerInRole(ctx, a, cmd); err == nil {
		t.Fatal("peer accepted without the isolated root protocol")
	}
	other := f.attempt
	other.InvocationID = "inv-2"
	if err := launch.ExactHealth(ctx, cmd, other); err == nil {
		t.Fatal("other attempt reported healthy")
	}
}

func socketPair() (net.Conn, net.Conn, error) {
	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0)
	if err != nil {
		return nil, nil, err
	}
	fa, fb := os.NewFile(uintptr(fds[0]), "a"), os.NewFile(uintptr(fds[1]), "b")
	defer fa.Close()
	defer fb.Close()
	a, err := net.FileConn(fa)
	if err != nil {
		return nil, nil, err
	}
	b, err := net.FileConn(fb)
	if err != nil {
		a.Close()
		return nil, nil, err
	}
	return a, b, nil
}

// shortTempDir keeps unix socket paths under the kernel's length limit.
func shortTempDir(t *testing.T) string {
	t.Helper()
	dir, err := os.MkdirTemp("", "rl")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	return dir
}

type plan14LaunchWork struct {
	held  bool
	calls int
	stale bool
}

func (w *plan14LaunchWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	if w.held {
		return hostwrite.ErrUnavailable
	}
	w.calls++
	if w.stale {
		return hostwrite.ErrUnavailable
	}
	w.held = true
	defer func() { w.held = false }()
	return fn()
}
func TestPlan14LaunchInitializationOutsideLock(t *testing.T) {
	for _, stale := range []bool{false, true} {
		t.Run(strconv.FormatBool(stale), func(t *testing.T) {
			f := newRootLaunchFixture(t)
			work := &plan14LaunchWork{}
			f.launch.Work = work
			calls := 0
			f.launch.InitializeUnlocked = func(context.Context) error {
				if work.held {
					t.Fatal("root RPC under agent lock")
				}
				calls++
				work.stale = stale
				return nil
			}
			_, _, settle, _, err := f.launch.Start(t.Context(), "main", f.attempt, f.build, f.before)
			if stale {
				if err == nil || len(f.root.launches) != 0 {
					t.Fatal("stale initialization applied")
				}
			} else {
				if err != nil {
					t.Fatal(err)
				}
				if err = settle(t.Context()); err != nil {
					t.Fatal(err)
				}
			}
			if calls != 1 || work.calls != 1 {
				t.Fatal(calls, work.calls)
			}
		})
	}
}