package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/stopevidence"
)

type writersJournalFake struct {
	mu      sync.Mutex
	inv     hostbox.LaunchInventory
	settled []string
	reads   int
	err     error
}

func (j *writersJournalFake) Read(_ context.Context, t hostguard.FenceTarget) (hostbox.LaunchInventory, error) {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.reads++
	if j.err != nil {
		return hostbox.LaunchInventory{}, j.err
	}
	if !hostbox.SameLaunchOwner(j.inv.Target, t) {
		return hostbox.LaunchInventory{}, hostbox.ErrChanged
	}
	out := j.inv
	out.Reservations = slices.Clone(j.inv.Reservations)
	return out, nil
}

func (j *writersJournalFake) Settled(_ context.Context, r hostbox.LaunchReservation) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	for i := range j.inv.Reservations {
		if j.inv.Reservations[i].ID == r.ID {
			j.inv.Reservations[i].State = "settled"
			j.settled = append(j.settled, r.ID)
			return nil
		}
	}
	return hostbox.ErrChanged
}

type writersCgroupFake struct {
	mu        sync.Mutex
	populated bool
	kills     int
	killErr   error
}

func (c *writersCgroupFake) Populated() (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.populated, nil
}

func (c *writersCgroupFake) KillAndEmpty(context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.kills++
	if c.killErr != nil {
		return c.killErr
	}
	c.populated = false
	return nil
}

// producerProbe is one registered producer: done closes when it finished.
type producerProbe struct {
	ctx  context.Context
	done chan struct{}
}

// runProducer registers id and runs it until its context ends, as a role
// launch, a timer delivery or the replicator does.
func runProducer(t *testing.T, p *ClaimProducers, target hostguard.FenceTarget, id string) producerProbe {
	t.Helper()
	ctx, finish, err := p.Register(context.Background(), target, id)
	if err != nil {
		t.Fatalf("register %s: %v", id, err)
	}
	done := make(chan struct{})
	go func() {
		<-ctx.Done()
		finish(nil)
		close(done)
	}()
	return producerProbe{ctx: ctx, done: done}
}

type writersFixture struct {
	wake     *ClaimWakeAssembly
	attempt  *claimWakeAttempt
	target   hostguard.FenceTarget
	journal  *writersJournalFake
	cgroup   *writersCgroupFake
	main     producerProbe
	timer    producerProbe
	rep      producerProbe
	pub      producerProbe
	producer *ClaimProducers
}

func newWritersFixture(t *testing.T) *writersFixture {
	t.Helper()
	target := hostguard.FenceTarget{Host: "shop.example", Incarnation: "inc-1", Stamp: 11, Session: 13, Box: "box-1", Enrollment: "enrollment-1", BootID: "boot-1", ControllerID: "controller-1", DirectoryRevision: "7"}
	claim := &hostclaim.Claim{V: 2, Host: target.Host, Gen: 4,
		Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 11, Session: 13, WriterBox: "box-1", Generation: "g-13", Location: "loc"},
		Op:     hostclaim.Op{Kind: hostclaim.KindServing}}
	observed := hostclaim.Observed{Host: target.Host, Claim: claim, ETag: "etag-1"}
	producers, err := NewClaimProducers(target, "start-1", func(context.Context, hostguard.FenceTarget, string) error { return nil })
	if err != nil {
		t.Fatal(err)
	}
	journal := &writersJournalFake{inv: hostbox.LaunchInventory{Target: target, Reservations: []hostbox.LaunchReservation{
		{ID: "attempt-1:main", Role: "main", State: "started", Target: target, PID: 77},
		{ID: "attempt-0:main", Role: "main", State: "settled", Target: target},
	}}}
	cg := &writersCgroupFake{populated: true}
	wake := &ClaimWakeAssembly{Producers: func(context.Context, hostclaim.Observed) (*ClaimProducers, error) { return producers, nil }}
	live, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	a := &claimWakeAttempt{owner: wake, observed: observed, inputs: ClaimLaunchInputs{DBPath: "/var/lib/carlos/shop.example/data.db"},
		prelaunch: &activator.RoleLaunch{Target: target, ParentPath: "carlos/claim-1.local"}, writersJournal: journal, writersCgroup: cg,
		ctx: live, cancel: cancel, done: make(chan struct{})}
	f := &writersFixture{wake: wake, attempt: a, target: target, journal: journal, cgroup: cg, producer: producers}
	f.main = runProducer(t, producers, target, "main")
	f.timer = runProducer(t, producers, target, "timer-1")
	f.rep = runProducer(t, producers, target, ClaimProducerReplicator)
	f.pub = runProducer(t, producers, target, ClaimProducerPublisher)
	return f
}

func closed(c <-chan struct{}) bool {
	select {
	case <-c:
		return true
	default:
		return false
	}
}

func TestStopWritersDrainsWritersAndKeepsTheReplicator(t *testing.T) {
	f := newWritersFixture(t)
	drained, err := f.attempt.StopWriters(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if !closed(f.main.done) || !closed(f.timer.done) {
		t.Fatal("the role launch or the timer delivery was not joined")
	}
	if f.rep.ctx.Err() != nil || f.pub.ctx.Err() != nil || f.attempt.ctx.Err() != nil {
		t.Fatal("the replicator, the publisher or the attempt was cancelled")
	}
	if !drained.FenceClosed || !drained.SpawnsSettled || !drained.WorkDrained || !drained.LaunchJoined || !drained.CgroupEmpty {
		t.Fatalf("facts = %+v", drained)
	}
	if drained.Target != f.target || drained.Generation != "g-13" || drained.Cgroup != "carlos/claim-1.local" || !slices.Equal(drained.DBPaths, []string{"/var/lib/carlos/shop.example/data.db"}) {
		t.Fatalf("witness = %+v", drained)
	}
	if !slices.Equal(drained.Drained, []string{"main", "timer-1"}) || !slices.Equal(drained.Retained, []string{ClaimProducerPublisher, ClaimProducerReplicator}) {
		t.Fatalf("drained %v retained %v", drained.Drained, drained.Retained)
	}
	if !slices.Equal(f.journal.settled, []string{"attempt-1:main"}) || f.cgroup.kills != 1 {
		t.Fatalf("settled %v kills %d", f.journal.settled, f.cgroup.kills)
	}
	if err := drained.For(f.attempt.observed); err != nil {
		t.Fatalf("witness does not bind its own session: %v", err)
	}
}

func TestStopWritersClosesAdmission(t *testing.T) {
	f := newWritersFixture(t)
	if _, err := f.attempt.StopWriters(t.Context()); err != nil {
		t.Fatal(err)
	}
	for _, id := range []string{"main-2", "timer-2"} {
		if _, _, err := f.producer.Register(context.Background(), f.target, id); !errors.Is(err, ErrClaimWritersStopped) {
			t.Fatalf("%s registered after the writers stopped: %v", id, err)
		}
		if err := f.producer.Launch(context.Background(), f.target, id, func(context.Context) error { return nil }); !errors.Is(err, ErrClaimWritersStopped) {
			t.Fatalf("%s launched after the writers stopped: %v", id, err)
		}
	}
	// The files publisher is retained: it may still register (L13).
	if _, finish, err := f.producer.Register(context.Background(), f.target, ClaimProducerFiles); err != nil {
		t.Fatalf("files publisher refused: %v", err)
	} else {
		finish(nil)
	}
	if err := f.attempt.admitsWriters(); !errors.Is(err, ErrClaimWritersStopped) {
		t.Fatalf("the attempt still admits writers: %v", err)
	}
}

func TestStopWritersIsIdempotent(t *testing.T) {
	f := newWritersFixture(t)
	first, err := f.attempt.StopWriters(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	reads, kills := f.journal.reads, f.cgroup.kills
	second, err := f.attempt.StopWriters(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(first.Drained, second.Drained) || first.Target != second.Target || first.CheckedAt != second.CheckedAt {
		t.Fatalf("second = %+v first = %+v", second, first)
	}
	if f.journal.reads != reads || f.cgroup.kills != kills {
		t.Fatal("a stopped attempt repeated its effects")
	}
}

func TestStopWritersRetriesAfterAFailure(t *testing.T) {
	f := newWritersFixture(t)
	f.cgroup.killErr = errors.New("cgroup busy")
	if _, err := f.attempt.StopWriters(t.Context()); err == nil {
		t.Fatal("a cgroup that would not empty passed")
	}
	if !closed(f.main.done) {
		t.Fatal("the role was not joined before the cgroup step")
	}
	f.cgroup.killErr = nil
	drained, err := f.attempt.StopWriters(t.Context())
	if err != nil || !drained.CgroupEmpty {
		t.Fatalf("retry = %+v %v", drained, err)
	}
}

func TestStopWritersRefuses(t *testing.T) {
	t.Run("a pending role intention", func(t *testing.T) {
		f := newWritersFixture(t)
		f.journal.inv.Reservations = append(f.journal.inv.Reservations, hostbox.LaunchReservation{ID: "attempt-2:main", State: "intended", Target: f.target})
		if _, err := f.attempt.StopWriters(t.Context()); !errors.Is(err, ErrClaimStopInventory) {
			t.Fatalf("got %v", err)
		}
	})
	t.Run("a populated cgroup after the kill", func(t *testing.T) {
		f := newWritersFixture(t)
		f.cgroup.killErr = nil
		f.attempt.writersCgroup = stuckCgroup{}
		if _, err := f.attempt.StopWriters(t.Context()); !errors.Is(err, ErrClaimStopInventory) {
			t.Fatalf("got %v", err)
		}
	})
	t.Run("another session", func(t *testing.T) {
		f := newWritersFixture(t)
		next := *f.attempt.observed.Claim
		holder := *next.Holder
		holder.Session = 14
		next.Holder = &holder
		f.attempt.observed.Claim = &next
		if _, err := f.attempt.StopWriters(t.Context()); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("got %v", err)
		}
		if closed(f.main.done) {
			t.Fatal("a mismatched session drained the writers")
		}
	})
	t.Run("no launch record", func(t *testing.T) {
		f := newWritersFixture(t)
		f.attempt.prelaunch = nil
		if _, err := f.attempt.StopWriters(t.Context()); err == nil {
			t.Fatal("an attempt with no launch record stopped writers")
		}
	})
	t.Run("after the full drain", func(t *testing.T) {
		f := newWritersFixture(t)
		req := StopRequest{Host: f.target.Host, Incarnation: f.target.Incarnation, InstallationStamp: f.target.Stamp, WriterSession: f.target.Session, Box: f.target.Box, Enrollment: f.target.Enrollment, BootID: f.target.BootID, ControllerID: f.target.ControllerID}
		if _, err := f.producer.DrainClaim(t.Context(), req, "fence-1"); err != nil {
			t.Fatal(err)
		}
		if _, err := f.attempt.StopWriters(t.Context()); !errors.Is(err, ErrClaimStopIdentity) {
			t.Fatalf("got %v", err)
		}
	})
}

type stuckCgroup struct{}

func (stuckCgroup) Populated() (bool, error)           { return true, nil }
func (stuckCgroup) KillAndEmpty(context.Context) error { return nil }

func TestWritersWitnessBindsTheExactSession(t *testing.T) {
	f := newWritersFixture(t)
	drained, err := f.attempt.StopWriters(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	for name, change := range map[string]func(*hostclaim.Holder){
		"session":     func(h *hostclaim.Holder) { h.Session++ },
		"stamp":       func(h *hostclaim.Holder) { h.Stamp++ },
		"incarnation": func(h *hostclaim.Holder) { h.Incarnation = "inc-2" },
		"generation":  func(h *hostclaim.Holder) { h.Generation = "g-14" },
		"writer box":  func(h *hostclaim.Holder) { h.WriterBox = "box-2" },
	} {
		o := f.attempt.observed
		c := *o.Claim
		h := *c.Holder
		change(&h)
		c.Holder = &h
		o.Claim = &c
		if err := drained.For(o); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("%s: %v", name, err)
		}
	}
	if err := (ClaimWritersDrained{}).For(f.attempt.observed); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("a zero witness passed: %v", err)
	}
}

func TestOwnedReplicatorOutlivesTheActivatorCloseOnceWritersStop(t *testing.T) {
	f := newWritersFixture(t)
	if _, err := f.attempt.StopWriters(t.Context()); err != nil {
		t.Fatal(err)
	}
	rep := &claimOwnedRep{attempt: f.attempt, owner: f.wake, host: f.target.Host}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	if err := rep.Close(ctx); err != nil {
		t.Fatal(err)
	}
	if f.attempt.ctx.Err() != nil || f.rep.ctx.Err() != nil {
		t.Fatal("the activator's close released a writers-stopped attempt")
	}
}

// R3's agent half: the replicator's descriptors are the difference between
// this process's own table before and after it opens.
func TestRecordReplicatorHintsKeepsOnlyTheNewDescriptors(t *testing.T) {
	dir := t.TempDir()
	db := filepath.Join(dir, "data.db")
	if err := os.WriteFile(db, nil, 0o600); err != nil {
		t.Fatal(err)
	}
	earlier, err := os.Open(db)
	if err != nil {
		t.Fatal(err)
	}
	defer earlier.Close()
	var opened *os.File
	defer func() {
		if opened != nil {
			opened.Close()
		}
	}()
	hints, err := recordReplicatorHints(t.Context(), "/proc", os.Getpid(), db, func() error {
		var err error
		opened, err = os.Open(db)
		return err
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(hints) != 1 || hints[0].FD != int(opened.Fd()) {
		t.Fatalf("hints = %+v, want only fd %d", hints, opened.Fd())
	}
	failed := errors.New("open failed")
	if _, err := recordReplicatorHints(t.Context(), "/proc", os.Getpid(), db, func() error { return failed }); !errors.Is(err, failed) {
		t.Fatalf("got %v", err)
	}
}

// The agent's whole step: StopWriters, then root's observation of the
// witnessed session at the current claim, carrying the recorded hint.
func TestClaimAssemblyWritersStoppedAsksRootForTheWitnessedSession(t *testing.T) {
	f := newWritersFixture(t)
	f.attempt.replicatorFDs = []stopevidence.DescriptorHint{{FD: 7, Device: 1, Inode: 2}}
	f.wake.pending = map[string]*claimWakeAttempt{f.target.Host: f.attempt}
	current := f.attempt.observed
	a := ClaimAssembly{Wake: f.wake, stopLocal: func(context.Context) (claimStopLocal, error) {
		return claimStopLocal{Actor: hostbox.Actor{Box: "box-1", Enrollment: "enrollment-1", Region: "eu-west-1", Revision: 9, Binding: "9"}, BootID: "boot-1", ControllerID: "controller-1", PID: 42, ProcessStart: "start-42"}, nil
	}}
	read := func(context.Context, string) (hostclaim.Observed, error) { return current, nil }
	var got WritersStoppedRequest
	collect := func(_ context.Context, r WritersStoppedRequest) (WritersStoppedEvidence, error) {
		got = r
		return WritersStoppedEvidence{Host: r.Host, WriterSession: r.WriterSession}, nil
	}
	drained, ev, err := a.writersStopped(t.Context(), f.target.Host, read, collect)
	if err != nil {
		t.Fatal(err)
	}
	if !drained.Valid() || ev.Host != f.target.Host || f.rep.ctx.Err() != nil {
		t.Fatalf("drained %+v ev %+v", drained, ev)
	}
	if got.WriterSession != 13 || got.ExpectedClaimETag != "etag-1" || got.ExpectedDirectoryRevision != "7" || got.Region != "eu-west-1" || got.BootID != "boot-1" || len(got.Replicator) != 1 {
		t.Fatalf("request = %+v", got)
	}
	t.Run("a later session", func(t *testing.T) {
		c := *current.Claim
		h := *c.Holder
		h.Session = 14
		c.Holder = &h
		current.Claim = &c
		if _, _, err := a.writersStopped(t.Context(), f.target.Host, read, collect); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("got %v", err)
		}
	})
	t.Run("another controller", func(t *testing.T) {
		current = f.attempt.observed
		b := a
		b.stopLocal = func(context.Context) (claimStopLocal, error) {
			return claimStopLocal{Actor: hostbox.Actor{Box: "box-1", Enrollment: "enrollment-1"}, BootID: "boot-1", ControllerID: "controller-2"}, nil
		}
		if _, _, err := b.writersStopped(t.Context(), f.target.Host, read, collect); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("got %v", err)
		}
	})
	t.Run("no attempt", func(t *testing.T) {
		if _, _, err := a.writersStopped(t.Context(), "other.example", read, collect); !errors.Is(err, ErrClaimStopUnavailable) {
			t.Fatalf("got %v", err)
		}
	})
}

func (j *writersJournalFake) CloseWriters(_ context.Context, target hostguard.FenceTarget) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.err != nil {
		return j.err
	}
	if !hostbox.SameLaunchOwner(j.inv.Target, target) {
		return hostbox.ErrChanged
	}
	j.inv.WritersClosed = true
	return nil
}