//go:build linux

package hostexec

import (
	"context"
	"encoding/json"
	"errors"
	"golang.org/x/sys/unix"
	"os"
	"os/exec"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
)

func TestLostControllerCannotPublishStaleActiveState(t *testing.T) {
	cfg, _, clock, _, _ := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer closeController(t, c)
	active := ready(t, c)
	c.mu.Lock()
	c.loseLocked(ErrLost)
	c.mu.Unlock()
	<-c.cleanupDone
	if err = c.publish(active); !errors.Is(err, ErrLost) {
		t.Fatalf("stale publication = %v", err)
	}
	if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); err == nil {
		t.Fatal("stale active state revived cross-process admission")
	}
}

func TestWaitingPublicationDoesNotBlockWatchdogOrCleanup(t *testing.T) {
	cfg, _, clock, _, cleaned := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer closeController(t, c)
	active := ready(t, c)
	held, release := make(chan struct{}), make(chan struct{})
	guardDone := make(chan error, 1)
	go func() {
		guardDone <- cfg.Guard.WithLocal(context.Background(), active.Host, func() error {
			close(held)
			<-release
			return nil
		})
	}()
	<-held
	published := make(chan error, 1)
	go func() { published <- c.publish(active) }()
	clock.advance(TTL)
	// Ready samples the suspend-aware clock even if a watchdog wake races.
	if _, err = c.Ready(t.Context()); err == nil {
		t.Fatal("expired controller remained ready")
	}
	select {
	case <-c.Lost():
	case <-time.After(2 * time.Second):
		t.Fatal("host guard blocked watchdog")
	}
	until(t, func() bool { return cleaned.Load() >= 2 })
	close(release)
	if err = <-guardDone; err != nil {
		t.Fatal(err)
	}
	if err = <-published; !errors.Is(err, ErrLost) {
		t.Fatalf("publication after expiration = %v", err)
	}
	<-c.cleanupDone
	if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); err == nil {
		t.Fatal("expired publication admitted work")
	}
}

func TestOrphanActiveStateCannotAuthorizeAfterControllerExit(t *testing.T) {
	cfg, _, clock, _, _ := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	active := ready(t, c)
	closeController(t, c)
	// Simulate an active rename surviving a crash before its fsync outcome was
	// known. The persisted deadline is still future, but no controller lives.
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	if err = d.write(active); err != nil {
		t.Fatal(err)
	}
	if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
		t.Fatalf("orphan active state = %v", err)
	}
}

type blockedClaimRead struct {
	base    *claims
	entered chan struct{}
	release chan struct{}
}

func (r blockedClaimRead) Read(ctx context.Context, host string) (hostclaim.Observed, error) {
	close(r.entered)
	select {
	case <-ctx.Done():
		return hostclaim.Observed{}, ctx.Err()
	case <-r.release:
		return r.base.Read(ctx, host)
	}
}

func TestCC9LockDoesNotReviveOrphanDuringClaimRead(t *testing.T) {
	cfg, r, clock, _, _ := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	active := ready(t, c)
	closeController(t, c)
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	if err = d.write(active); err != nil {
		t.Fatal(err)
	}
	blocked := blockedClaimRead{base: r, entered: make(chan struct{}), release: make(chan struct{})}
	cfg.Guard.Claims = blocked
	done := make(chan error, 1)
	go func() { done <- ClearStale(t.Context(), cfg) }()
	<-blocked.entered
	checkErr := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir)
	close(blocked.release)
	if checkErr == nil {
		t.Fatal("CC9 lifetime lock revived orphan active bytes")
	}
	if err = <-done; !errors.Is(err, ErrInactive) {
		t.Fatalf("CC9 with participation retained = %v", err)
	}
}

func TestCC9SuccessorExclusionClearsOnlyExpectedOperation(t *testing.T) {
	cfg, r, _, _, _ := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	ready(t, c)
	closeController(t, c)
	r.mu.Lock()
	r.q.Op.ID = "successor-op"
	r.q.Op.Exec = "other-box"
	r.q.Op.Participants = []string{"other-box"}
	r.mu.Unlock()
	if err = ClearStale(t.Context(), cfg); err != nil {
		t.Fatalf("successor excludes box: %v", err)
	}
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	if _, err = d.read(cfg.Binding.Host); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("expected stale record removed: %v", err)
	}
	unrelated := State{V: 1, Binding: cfg.Binding, State: "lost", Expires: time.Now().Add(TTL), SafeUntil: 99}
	unrelated.ControllerID = "other-controller"
	unrelated.Op = "unrelated-op"
	if err = d.write(unrelated); err != nil {
		t.Fatal(err)
	}
	if err = ClearStale(t.Context(), cfg); !errors.Is(err, ErrState) {
		t.Fatalf("unrelated local fence: %v", err)
	}
	still, err := d.read(cfg.Binding.Host)
	if err != nil || still.Op != unrelated.Op {
		t.Fatalf("unrelated fence changed: %v, %v", still, err)
	}
}

func TestAmbiguousRenewalENOSPCBeforeRenameDeniesImmediately(t *testing.T) {
	cfg, _, clock, store, cleaned := fixture(t)
	store.failAt = 2
	var noSpace atomic.Bool
	noSpace.Store(true)
	var failures atomic.Int32
	cfg.writeLocal = func(d *localDir, s State) error {
		if s.State != "lost" || !noSpace.Load() {
			return d.write(s)
		}
		original := d.sync
		d.sync = func(int) error { return unix.ENOSPC }
		defer func() { d.sync = original }()
		err := d.write(s) // file fsync fails before rename; old active bytes remain
		if errors.Is(err, unix.ENOSPC) {
			failures.Add(1)
		}
		return err
	}
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { noSpace.Store(false); closeController(t, c) }()
	active := ready(t, c)
	until(t, func() bool {
		clock.mu.Lock()
		defer clock.mu.Unlock()
		for _, a := range clock.alarms {
			if a.at == clock.boot+int64(Renewal) {
				return true
			}
		}
		return false
	})
	clock.advance(Renewal)
	select {
	case <-c.Lost():
	case <-time.After(2 * time.Second):
		t.Fatal("ambiguous renewal did not latch loss")
	}
	// Check from the loss signal onward, before waiting for any persistence.
	if err := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
		t.Fatalf("admitted immediately after ambiguity: %v", err)
	}
	until(t, func() bool { return failures.Load() >= 2 && cleaned.Load() >= 2 })
	if _, err := c.Ready(t.Context()); err == nil {
		t.Fatal("in-memory admission survived loss")
	}
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	old, err := d.read(active.Host)
	if err != nil || old != active {
		t.Fatalf("test did not leave the old active file: %+v %v", old, err)
	}
	_, boot, _ := clock.Now()
	if boot >= active.SafeUntil {
		t.Fatal("test reached expiry instead of exercising immediate denial")
	}
	binding, _ := json.Marshal(active.Binding)
	cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^TestCheckLocalRefusesLostLiveLock$")
	cmd.Env = append(os.Environ(), "HOSTEXEC_PROBE_BINDING="+string(binding), "HOSTEXEC_PROBE_DATA="+cfg.Guard.DataDir, "HOSTEXEC_PROBE_LOCKS="+cfg.LockDir)
	if output, err := cmd.CombinedOutput(); err != nil {
		t.Fatalf("independent process admitted old active state: %v\n%s", err, output)
	}
	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond)
	defer cancel()
	entered := false
	err = cfg.Guard.WithLocal(ctx, active.Host, func() error { entered = true; return nil })
	if !errors.Is(err, context.DeadlineExceeded) || entered {
		t.Fatalf("guard released while denial write failed: %v entered=%v", err, entered)
	}
	if unlock, err := tryAcquire(cfg.LockDir, active.Host); !errors.Is(err, ErrControllerBusy) {
		if unlock != nil {
			unlock()
		}
		t.Fatalf("controller exclusion released: %v", err)
	}
	ctxClose, cancelClose := context.WithTimeout(t.Context(), 30*time.Millisecond)
	defer cancelClose()
	if err := c.Close(ctxClose); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("Close did not retain the unfinished controller: %v", err)
	}
	noSpace.Store(false)
	select {
	case <-c.cleanupDone:
	case <-time.After(2 * time.Second):
		t.Fatal("lost state was not retried after disk recovery")
	}
	lost, err := d.read(active.Host)
	if err != nil || lost.State != "lost" {
		t.Fatalf("denial not durable: %+v %v", lost, err)
	}
	if err := cfg.Guard.WithLocal(t.Context(), active.Host, func() error {
		if err := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
			t.Fatalf("disk recovery revived admission: %v", err)
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
}

func TestCheckLocalRefusesLostLiveLock(t *testing.T) {
	raw := os.Getenv("HOSTEXEC_PROBE_BINDING")
	if raw == "" {
		t.Skip("subprocess witness")
	}
	var binding Binding
	if err := json.Unmarshal([]byte(raw), &binding); err != nil {
		t.Fatal(err)
	}
	clock := &testClock{boot: int64(time.Hour + Renewal)}
	if err := CheckLocal(os.Getenv("HOSTEXEC_PROBE_DATA"), binding, clock, os.Getenv("HOSTEXEC_PROBE_LOCKS")); !errors.Is(err, ErrLost) {
		t.Fatalf("old active record authorized an independent process: %v", err)
	}
}

func TestOpenOrphanControllerPendingWriteFailureCleansBeforeReleasingExclusion(t *testing.T) {
	testOpenOrphanControllerPersistence(t, false)
}

func TestOpenBlockedPendingPersistenceStillCleansBeforeReleasingExclusion(t *testing.T) {
	testOpenOrphanControllerPersistence(t, true)
}

func testOpenOrphanControllerPersistence(t *testing.T, blocked bool) {
	cfg, _, clock, store, _ := fixture(t)
	wall, boot, _ := clock.Now()
	orphan := State{V: 1, Binding: cfg.Binding, State: "active", LeaseETag: "orphan-etag", Expires: wall.Add(TTL), SafeUntil: boot + int64(TTL-Margin)}
	orphan.ControllerID = "dead-controller"
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	if err := d.write(orphan); err != nil {
		t.Fatal(err)
	}
	// The controller's locks have vanished, but its active bytes and a host
	// process survived. Admission refusal alone cannot stop that process.
	process := exec.Command("sleep", "60")
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer process.Process.Kill()
	hostExited := make(chan struct{})
	go func() { _ = process.Wait(); close(hostExited) }()
	allowJoin := make(chan struct{})
	var joinOnce sync.Once
	defer joinOnce.Do(func() { close(allowJoin) })
	cleaned := make(chan struct{})
	cleanupResult := make(chan error, 1)
	cfg.Cleanup = func(ctx context.Context) error {
		cleanupResult <- ctx.Err()
		if err := process.Process.Kill(); err != nil {
			return err
		}
		<-hostExited
		close(cleaned)
		<-allowJoin
		return nil
	}
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	writeEntered := make(chan struct{})
	allowWrite := make(chan struct{})
	var writeOnce sync.Once
	defer writeOnce.Do(func() { close(allowWrite) })
	var noSpace atomic.Bool
	var failures atomic.Int32
	noSpace.Store(true)
	defer noSpace.Store(false)
	cfg.writeLocal = func(d *localDir, s State) error {
		if noSpace.Load() {
			if s.State == "pending" {
				cancel() // cleanup must not inherit the failed opener's lifetime
			}
			original := d.sync
			d.sync = func(int) error {
				if blocked && s.State == "pending" {
					close(writeEntered)
					<-allowWrite // model an fsync that cannot return until released
				}
				return unix.ENOSPC
			}
			defer func() { d.sync = original }()
			err := d.write(s)
			if errors.Is(err, unix.ENOSPC) {
				failures.Add(1)
			}
			return err
		}
		return d.write(s)
	}
	done := make(chan error, 1)
	go func() { _, err := Open(ctx, cfg); done <- err }()
	select {
	case <-cleaned:
	case <-time.After(2 * time.Second):
		t.Fatal("failed pending write skipped orphan process cleanup")
	}
	if err := <-cleanupResult; err != nil {
		t.Fatalf("cleanup inherited caller cancellation: %v", err)
	}

	if old, err := d.read(orphan.Host); err != nil || old != orphan {
		t.Fatalf("failed persistence did not retain orphan active bytes: %+v %v", old, err)
	}
	assertExcluded := func() {
		t.Helper()
		if unlock, err := tryAcquire(cfg.LockDir, orphan.Host); !errors.Is(err, ErrControllerBusy) {
			if unlock != nil {
				unlock()
			}
			t.Fatalf("exclusion released before cleanup joined: %v", err)
		}
		select {
		case err := <-done:
			t.Fatalf("Open returned before cleanup joined: %v", err)
		default:
		}
	}
	if blocked {
		select {
		case <-writeEntered:
		case <-time.After(2 * time.Second):
			t.Fatal("pending fsync was not reached")
		}
		if failures.Load() != 0 {
			t.Fatal("pending fsync returned before cleanup was observed")
		}
		assertExcluded()
		writeOnce.Do(func() { close(allowWrite) })
	}
	until(t, func() bool { return failures.Load() >= 1 })
	assertExcluded()
	if err := CheckLocal(cfg.Guard.DataDir, orphan.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
		t.Fatalf("orphan bytes admitted work: %v", err)
	}
	// Persistence is still unavailable when cleanup joins. No live lock was
	// acquired, so orphan bytes cannot authorize work after exclusion ends.
	joinOnce.Do(func() { close(allowJoin) })
	select {
	case err := <-done:
		if !errors.Is(err, unix.ENOSPC) {
			t.Fatalf("Open lost pending-write failure: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("Open did not join cleanup while persistence remained unavailable")
	}
	unlock, err := tryAcquire(cfg.LockDir, orphan.Host)
	if err != nil {
		t.Fatalf("exclusion retained after cleanup joined: %v", err)
	}
	unlock()
	if err := CheckLocal(cfg.Guard.DataDir, orphan.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
		t.Fatalf("released exclusion revived orphan bytes: %v", err)
	}
	store.mu.Lock()
	defer store.mu.Unlock()
	if store.calls != 0 {
		t.Fatalf("failed startup attempted %d lease writes", store.calls)
	}
}