package host

import (
	"context"
	"encoding/json"
	"errors"
	"net"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/registry"
)

func retirementRecoveryRow(t *testing.T, f *recoveryFixture) registry.Route {
	t.Helper()
	o := f.read()
	var err error
	o, err = hostclaim.Update(t.Context(), f.claims, o, hostclaim.BeginDelete("delete-1", "console", `"record-etag"`, f.now, nil, nil, ""))
	if err != nil {
		t.Fatal(err)
	}
	o, err = hostclaim.Update(t.Context(), f.claims, o, hostclaim.ToParking("delete-1", nil, nil, f.newID))
	if err != nil {
		t.Fatal(err)
	}

	h := o.Claim.Holder
	return registry.Route{Host: o.Host, ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: h.Incarnation, ClaimStamp: h.Stamp, ClaimOwner: "installation", ClaimRowID: "route-1"}}
}

func recordRetirementSuccessor(t *testing.T, f *recoveryFixture, admitted hostclaim.Observed, actor hostbox.Actor) (hostguard.FenceTarget, hostbox.LaunchReservation) {
	t.Helper()
	successor := recoveryTuple(admitted, claimStopLocal{Actor: actor, BootID: "boot-1", ControllerID: "controller-2"})
	successor.Session++
	successor.Stamp++
	successor.DirectoryRevision = "8"
	if err := f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: 4202, ProcessStart: "start-2", Target: successor, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/retirement-successor", Device: 1, Inode: 99}}, NoState: true}); err != nil {
		t.Fatal(err)
	}
	reservation := hostbox.LaunchReservation{ID: "successor:main", Role: "main", InvocationID: "successor-invocation", Inputs: "inputs", Target: successor, Group: hostbox.LaunchGroup{RelativePath: "carlos/retirement-successor/role", Device: 1, Inode: 100, InvocationID: "successor-invocation"}, Executable: hostbox.LaunchFile{Device: 1, Inode: 101}}
	if err := f.journal.Reserve(t.Context(), reservation); err != nil {
		t.Fatal(err)
	}
	return successor, reservation
}

func retirementRecoveryReceipts(t *testing.T, f *recoveryFixture) []StopReceipt {
	t.Helper()
	keys, err := f.evidence.List(t.Context(), "control/hosts/"+recoveryHost+"/evidence/delete-1/box-1/")
	if err != nil {
		t.Fatal(err)
	}
	out := make([]StopReceipt, 0, len(keys))
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		body, _, err := f.evidence.Get(t.Context(), key)
		if err != nil {
			t.Fatal(err)
		}
		var receipt StopReceipt
		if err := json.Unmarshal(body, &receipt); err != nil {
			t.Fatal(err)
		}
		out = append(out, receipt)
	}
	return out
}

func startRetirementRecoveryTransport(t *testing.T, handler ClaimRootHTTP) string {
	t.Helper()
	directory, err := os.MkdirTemp("", "retire-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(directory) })
	socket := filepath.Join(directory, "s")
	listener, err := net.Listen("unix", socket)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(t.Context())
	done := make(chan error, 1)
	server := instancectl.Server{ClaimStop: handler, PeerOK: func(net.Conn) error { return nil }}
	go func() { done <- server.Serve(ctx, listener) }()
	t.Cleanup(func() {
		cancel()
		_ = listener.Close()
		if err := <-done; err != nil {
			t.Errorf("retirement transport: %v", err)
		}
	})
	return socket
}

func TestClaimRetirementRecoversPersistedLaunchOwner(t *testing.T) {
	t.Parallel()
	f := newRecoveryFixture(t)
	former := f.controller("boot-1", "controller-1", 4101, "start-1")
	admitted := former.admit(t, f)
	successor, reservation := recordRetirementSuccessor(t, f, admitted, former.local.Actor)
	row := retirementRecoveryRow(t, f)

	f.revision = 8
	restarted := f.controller("boot-1", "controller-2", 4202, "start-2")
	f.kernel.alive = true
	if _, err := restarted.assembly.collectRetirementStop(t.Context(), row); !errors.Is(err, ErrClaimStopControllerAlive) {
		t.Fatalf("live launch owner retired = %v", err)
	}
	if len(f.requests) != 1 {
		t.Fatalf("stop requests before death proof = %+v", f.requests)
	}
	request := f.requests[0]
	if !request.Recover || request.BootID != "boot-1" || request.ControllerID != "controller-1" || request.ExpectedDirectoryRevision != "7" {
		t.Fatalf("retirement substituted the restarted controller tuple: %+v", request)
	}
	if got := retirementRecoveryReceipts(t, f); len(got) != 0 {
		t.Fatalf("receipt before root death proof: %+v", got)
	}

	f.kernel.alive = false
	target, err := restarted.assembly.collectRetirementStop(t.Context(), row)
	if err != nil {
		t.Fatal(err)
	}
	if target.BootID != "boot-1" || target.ControllerID != "controller-1" || target.DirectoryRevision != "7" || target.Session != admitted.Claim.Holder.Session {
		t.Fatalf("retirement target = %+v", target)
	}
	if len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 2 || f.kernel.recovered[0].FenceID != f.kernel.recovered[1].FenceID {
		t.Fatalf("verified=%v recovered=%v", f.kernel.verified, f.kernel.recovered)
	}
	other, err := f.journal.Read(t.Context(), successor)
	expected := reservation
	expected.State = "reserved"
	if err != nil || len(other.Reservations) != 1 || other.Reservations[0] != expected {
		t.Fatalf("retirement changed successor: %+v %v", other, err)
	}
	if got := retirementRecoveryReceipts(t, f); len(got) != 1 || got[0].ControllerID != "controller-1" || got[0].DirectoryRevision != "7" || got[0].FenceID != target.FenceID {
		t.Fatalf("receipts = %+v", got)
	}

	// A bookkeeping CAS changes the claim ETag but not the stop authority.
	// Recollection must replay the immutable receipt and retain its old tuple.
	current := f.read()
	if _, err = hostclaim.Update(t.Context(), f.claims, current, func(*hostclaim.Claim) error { return nil }); err != nil {
		t.Fatal(err)
	}
	recoveredBefore := len(f.kernel.recovered)
	replayed, err := restarted.assembly.collectRetirementStop(t.Context(), row)
	if err != nil {
		t.Fatalf("receipt replay after bookkeeping CAS: %v", err)
	}
	if replayed != target || len(f.kernel.recovered) != recoveredBefore {
		t.Fatalf("replay target=%+v recovered=%d, want %+v and %d", replayed, len(f.kernel.recovered), target, recoveredBefore)
	}

	// Retirement acknowledgement reads the claim from the evidence store in
	// production. Mirror the fixture's claim there before exercising root HTTP.
	claimKey, err := hostclaim.Key(recoveryHost)
	if err != nil {
		t.Fatal(err)
	}
	claimBody, _, err := f.claims.Get(t.Context(), claimKey)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = f.evidence.Put(t.Context(), claimKey, claimBody, ""); err != nil {
		t.Fatal(err)
	}

	base := t.TempDir()
	inventory, err := openLegacyInventory(t, filepath.Join(base, "inventory.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { inventory.Close() })
	reg, err := registry.Open(filepath.Join(base, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	owner := row.ClaimOwnership
	in := hostbox.StopInventory{Complete: true, Host: row.Host, Incarnation: owner.ClaimIncarnation, InstallationStamp: owner.ClaimStamp, Box: "box-1", Enrollment: "enrollment-1", DirectoryRevision: "7", Entries: []hostbox.StopInventoryEntry{{Kind: "route", ArtifactID: owner.ClaimRowID, ManagedRoot: "registry", RelativePath: row.Host, RowIdentity: owner.ClaimRowID, Ownership: owner.ClaimOwner, Materialization: "materialized"}}}
	if err = inventory.WriteStopInventory(t.Context(), in); err != nil {
		t.Fatal(err)
	}
	successorRow := row
	successorRow.ClaimStamp++
	successorRow.ClaimOwner = "successor"
	successorRow.ClaimRowID = "route-2"
	if err = reg.Upsert(row); err != nil {
		t.Fatal(err)
	}
	coordinator := f.root(recoveryStopDrain{})
	root := ClaimRootHTTP{
		Stop:  ClaimStopHTTP{Coordinator: coordinator, Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return f.peer, nil }},
		Roots: map[string]string{"data": base},
		Reconcile: hostbox.ReconcileAdapter{
			Box: "box-1", Enrollment: "enrollment-1", Inventory: inventory, Registry: reg,
			Retirement: hostbox.TypedRetirement{JournalDir: filepath.Join(base, "retire")},
		},
	}
	socket := startRetirementRecoveryTransport(t, root)
	unknown := target
	unknown.ControllerID = "unknown-controller"
	if err = RetireClaimOverSocket(t.Context(), socket, unknown); !errors.Is(err, ErrClaimStopReceipt) {
		t.Fatalf("unknown old target = %v", err)
	}
	if got, e := reg.Get(row.Host); e != nil || got == nil || got.ClaimRowID != row.ClaimRowID {
		t.Fatalf("unknown target changed row: %+v %v", got, e)
	}
	if err = RetireClaimOverSocket(t.Context(), socket, target); err != nil {
		t.Fatalf("retire recovered target over HTTP: %v", err)
	}
	if got, e := reg.Get(row.Host); e != nil || got != nil {
		t.Fatalf("retirement left original route: %+v %v", got, e)
	}
	if err = reg.Upsert(successorRow); err != nil {
		t.Fatal(err)
	}
	if err = RetireClaimOverSocket(t.Context(), socket, target); err != nil {
		t.Fatalf("replay recovered retirement over HTTP: %v", err)
	}
	if got, e := reg.Get(row.Host); e != nil || got == nil || got.ClaimRowID != successorRow.ClaimRowID {
		t.Fatalf("recovered retirement changed successor: %+v %v", got, e)
	}
}

func TestClaimRetirementRefusesMissingOrAmbiguousLaunchJournal(t *testing.T) {
	t.Run("missing", func(t *testing.T) {
		f := newRecoveryFixture(t)
		f.recordPrelaunch = false
		former := f.controller("boot-1", "controller-1", 4101, "start-1")
		former.admit(t, f)
		row := retirementRecoveryRow(t, f)
		f.revision = 8
		restarted := f.controller("boot-1", "controller-2", 4202, "start-2")
		if _, err := restarted.assembly.collectRetirementStop(t.Context(), row); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("missing launch journal = %v", err)
		}
		if len(f.requests) != 0 || len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 0 {
			t.Fatalf("missing journal caused stop effects: requests=%+v verified=%v recovered=%v", f.requests, f.kernel.verified, f.kernel.recovered)
		}
		if got := retirementRecoveryReceipts(t, f); len(got) != 0 {
			t.Fatalf("missing journal produced evidence: %+v", got)
		}
	})

	t.Run("ambiguous", func(t *testing.T) {
		f := newRecoveryFixture(t)
		former := f.controller("boot-1", "controller-1", 4101, "start-1")
		admitted := former.admit(t, f)
		second := recoveryTuple(admitted, claimStopLocal{Actor: former.local.Actor, BootID: "boot-1", ControllerID: "controller-other"})
		if err := f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: 4199, ProcessStart: "start-other", Target: second, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/ambiguous", Device: 1, Inode: 98}}, NoState: true}); err != nil {
			t.Fatal(err)
		}
		row := retirementRecoveryRow(t, f)
		f.revision = 8
		restarted := f.controller("boot-1", "controller-2", 4202, "start-2")
		if _, err := restarted.assembly.collectRetirementStop(t.Context(), row); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("ambiguous launch journal = %v", err)
		}
		if len(f.requests) != 0 || len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 0 {
			t.Fatalf("ambiguous journal caused partial stop: requests=%+v verified=%v recovered=%v", f.requests, f.kernel.verified, f.kernel.recovered)
		}
		if got := retirementRecoveryReceipts(t, f); len(got) != 0 {
			t.Fatalf("ambiguous journal produced evidence: %+v", got)
		}
	})
}

func TestClaimRetirementAfterServingStopSameController(t *testing.T) {
	f := newRecoveryFixture(t)
	controller := f.controller("boot-1", "controller-1", 4101, "start-1")
	admitted := controller.admit(t, f)
	if err := controller.wake.Stop(t.Context(), admitted); err != nil {
		t.Fatal("serving stop", err)
	}
	row := retirementRecoveryRow(t, f)
	target, err := controller.assembly.collectRetirementStop(t.Context(), row)
	if err != nil {
		t.Fatal("retirement after serving stop", err)
	}
	if target.Op != "delete-1" || target.Phase != "park" || target.ObligationID == "" || target.Session != admitted.Claim.Holder.Session {
		t.Fatalf("retirement tuple: %+v", target)
	}
}

func TestClaimRetirementCannotReplaceUnfinishedDrain(t *testing.T) {
	target := hostguard.FenceTarget{Host: "app.example", Incarnation: "inc", Stamp: 1, Session: 1, Box: "box", Enrollment: "en", BootID: "boot", ControllerID: "controller"}
	p, err := NewClaimProducers(target, "start", func(context.Context, hostguard.FenceTarget, string) error { return nil })
	if err != nil {
		t.Fatal(err)
	}
	finish := make(chan struct{})
	defer func() {
		select {
		case <-finish:
		default:
			close(finish)
		}
	}()
	if err = p.Launch(t.Context(), target, "writer", func(ctx context.Context) error { <-ctx.Done(); <-finish; return ctx.Err() }); err != nil {
		t.Fatal(err)
	}
	req := StopRequest{Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: 1, WriterSession: 1, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID}
	short, cancel := context.WithTimeout(t.Context(), 10*time.Millisecond)
	_, err = p.DrainClaim(short, req, "serving-fence")
	cancel()
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatal("unfinished drain", err)
	}
	short, cancel = context.WithTimeout(t.Context(), 20*time.Millisecond)
	_, err = p.DrainClaim(short, req, "retirement-fence")
	cancel()
	if !errors.Is(err, ErrClaimStopIdentity) {
		t.Fatal("replaced unfinished drain", err)
	}
	close(finish)
	if _, err = p.DrainClaim(t.Context(), req, "serving-fence"); err != nil {
		t.Fatal(err)
	}
	ack, err := p.DrainClaim(t.Context(), req, "retirement-fence")
	if err != nil || ack.FenceID != "retirement-fence" {
		t.Fatal("completed drain retirement", ack, err)
	}
	req.WriterSession++
	if _, err = p.DrainClaim(t.Context(), req, "foreign-fence"); !errors.Is(err, ErrClaimStopIdentity) {
		t.Fatal("foreign session rebound", err)
	}
	if err = p.Launch(t.Context(), target, "new-writer", func(context.Context) error { return nil }); err == nil {
		t.Fatal("retirement reopened producer admission")
	}
}