//go:build linux

package hostexec

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

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostquarantine"
	"github.com/carlosframework/platform/internal/release"
)

type alarm struct {
	at int64
	ch chan time.Time
}
type testClock struct {
	mu     sync.Mutex
	wall   time.Time
	boot   int64
	alarms []alarm
}

func (k *testClock) Now() (time.Time, int64, error) {
	k.mu.Lock()
	defer k.mu.Unlock()
	return k.wall, k.boot, nil
}
func (k *testClock) After(d time.Duration) <-chan time.Time {
	k.mu.Lock()
	defer k.mu.Unlock()
	ch := make(chan time.Time, 1)
	if d <= 0 {
		ch <- k.wall
	} else {
		k.alarms = append(k.alarms, alarm{k.boot + int64(d), ch})
	}
	return ch
}
func (k *testClock) advance(d time.Duration) {
	k.mu.Lock()
	defer k.mu.Unlock()
	k.boot += int64(d)
	k.wall = k.wall.Add(d)
	rest := k.alarms[:0]
	for _, a := range k.alarms {
		if a.at <= k.boot {
			a.ch <- k.wall
		} else {
			rest = append(rest, a)
		}
	}
	k.alarms = rest
}

type claims struct {
	mu  sync.Mutex
	q   hostclaim.Claim
	err error
}

func (r *claims) Read(context.Context, string) (hostclaim.Observed, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	raw, _ := json.Marshal(r.q)
	var q hostclaim.Claim
	_ = json.Unmarshal(raw, &q)
	return hostclaim.Observed{Host: q.Host, Claim: &q, ETag: "claim-etag"}, r.err
}

type directory struct {
	state string
	err   error
}

func (d directory) Lookup(context.Context, string) (hostquarantine.Actor, error) {
	return hostquarantine.Actor{Enrollment: "enroll", State: d.state}, d.err
}

type delayedStore struct {
	release.Store
	mu     sync.Mutex
	calls  int
	onPut  func(int)
	failAt int
	called chan int
}

func (s *delayedStore) Put(ctx context.Context, key string, b []byte, etag string) (string, error) {
	s.mu.Lock()
	s.calls++
	n := s.calls
	s.mu.Unlock()
	if s.called != nil {
		s.called <- n
	}
	if s.onPut != nil {
		s.onPut(n)
	}
	if n == s.failAt {
		return "", errors.New("ambiguous network failure")
	}
	// Deliberately permit a late response even after cancellation.
	return s.Store.Put(context.Background(), key, b, etag)
}
func (s *delayedStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
	return release.DeleteIfMatch(ctx, s.Store, key, etag)
}

func fixture(t *testing.T) (Config, *claims, *testClock, *delayedStore, *atomic.Int32) {
	t.Helper()
	dir := t.TempDir()
	for _, name := range []string{"hostguard", ".hostexec", "locks", "exec-locks"} {
		if err := os.Mkdir(filepath.Join(dir, name), 0770); err != nil {
			t.Fatal(err)
		}
	}
	host := "shop.example.com"
	inc, _ := hostclaim.Incarnation("bab", "jam", host, time.Time{})
	q := hostclaim.Claim{V: 2, Host: host, Gen: 7, Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Incarnation: inc, Location: "instances/" + host, Stamp: 3}, Op: hostclaim.Op{Kind: hostclaim.KindDeleting, ID: "op", Phase: "stop", Exec: "box", Participants: []string{"box"}}}
	if err := q.Validate(); err != nil {
		t.Fatal(err)
	}
	r := &claims{q: q}
	clock := &testClock{wall: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), boot: int64(time.Hour)}
	store := &delayedStore{Store: release.NewDirStore(t.TempDir()), called: make(chan int, 20)}
	cleaned := &atomic.Int32{}
	cfg := Config{Binding: Binding{Box: "box", Enrollment: "enroll", Host: host, Incarnation: inc, Op: "op", BootID: "boot"}, Store: store, Guard: hostguard.Guard{DataDir: dir, LockDir: filepath.Join(dir, "locks"), Claims: r}, Directory: directory{state: "active"}, Clock: clock, LockDir: filepath.Join(dir, "exec-locks"), CloseAdmission: func() {}, CancelWork: func() {}, Cleanup: func(context.Context) error { cleaned.Add(1); return nil }}
	return cfg, r, clock, store, cleaned
}
func ready(t *testing.T, c *Controller) State {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
	defer cancel()
	s, err := c.Ready(ctx)
	if err != nil {
		t.Fatal(err)
	}
	return s
}
func closeController(t *testing.T, c *Controller) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	if err := c.Close(ctx); err != nil {
		t.Error(err)
	}
}
func until(t *testing.T, fn func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for !fn() {
		if time.Now().After(deadline) {
			t.Fatal("condition did not become true")
		}
		time.Sleep(time.Millisecond)
	}
}

func TestRenewalRequestStartAndIndependentExpiry(t *testing.T) {
	cfg, _, clock, store, cleaned := fixture(t)
	blocked := make(chan struct{})
	store.onPut = func(n int) {
		if n == 2 {
			<-blocked
		}
	}
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer closeController(t, c)
	first := ready(t, c)
	key, _ := Key(first.Host, first.Box)
	b, _, err := store.Get(t.Context(), key)
	if err != nil {
		t.Fatal(err)
	}
	var wire map[string]any
	_ = json.Unmarshal(b, &wire)
	if len(wire) != 2 || wire["op"] != "op" {
		t.Fatalf("wire %s", b)
	}
	until(t, func() bool { clock.mu.Lock(); defer clock.mu.Unlock(); return len(clock.alarms) >= 2 })
	clock.advance(Renewal)
	for {
		select {
		case n := <-store.called:
			if n == 2 {
				goto renewing
			}
		case <-time.After(2 * time.Second):
			t.Fatal("no renewal")
		}
	}
renewing:
	clock.advance(65 * time.Second)
	select {
	case <-c.Lost():
	case <-time.After(2 * time.Second):
		t.Fatal("blocked PUT blocked watchdog")
	}
	if err := CheckLocal(cfg.Guard.DataDir, first.Binding, clock, cfg.LockDir); err == nil {
		t.Fatal("writer admitted at deadline")
	}
	close(blocked)
	<-c.done
	if _, err := c.Ready(t.Context()); err == nil {
		t.Fatal("late success revived controller")
	}
	until(t, func() bool { return cleaned.Load() >= 2 })
}

func TestRenewalUsesStartAndWallRollbackCannotExtend(t *testing.T) {
	cfg, _, clock, store, _ := fixture(t)
	store.onPut = func(n int) {
		if n == 1 {
			clock.advance(9 * time.Second)
		}
	}
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer closeController(t, c)
	s := ready(t, c)
	if s.SafeUntil != int64(time.Hour+85*time.Second) {
		t.Fatalf("response-relative deadline %d", s.SafeUntil)
	}
	clock.mu.Lock()
	clock.wall = clock.wall.Add(-time.Hour)
	clock.mu.Unlock()
	clock.advance(76 * time.Second)
	if err := CheckLocal(cfg.Guard.DataDir, s.Binding, clock, cfg.LockDir); err == nil {
		t.Fatal("wall rollback extended deadline")
	}
}

func TestAmbiguousPUTLatchesAndConditionalReleasePreservesSuccessor(t *testing.T) {
	cfg, _, _, store, _ := fixture(t)
	store.failAt = 1
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = c.Ready(t.Context()); err == nil {
		t.Fatal("ambiguous PUT admitted")
	}
	closeController(t, c)
	cfg, _, _, store, _ = fixture(t)
	c, err = Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	s := ready(t, c)
	key, _ := Key(s.Host, s.Box)
	successor := []byte(`{"op":"successor","expires":"2030-01-01T00:00:00Z"}`)
	if _, err = store.Store.Put(t.Context(), key, successor, s.LeaseETag); err != nil {
		t.Fatal(err)
	}
	if err = c.Close(t.Context()); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("release = %v", err)
	}
	got, _, err := store.Get(t.Context(), key)
	if err != nil || string(got) != string(successor) {
		t.Fatal("deleted successor")
	}
}

func TestControllerExclusionBootAndCC9(t *testing.T) {
	cfg, r, clock, _, _ := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	s := ready(t, c)
	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond)
	defer cancel()
	if _, err = Open(ctx, cfg); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("second controller = %v", err)
	}
	wrong := s.Binding
	wrong.BootID = "next-boot"
	if err = CheckLocal(cfg.Guard.DataDir, wrong, clock, cfg.LockDir); err == nil {
		t.Fatal("inherited prior boot")
	}
	closeController(t, c)
	if err = ClearStale(t.Context(), cfg); !errors.Is(err, ErrInactive) {
		t.Fatalf("cleared active op: %v", err)
	}
	r.mu.Lock()
	r.q.Op = hostclaim.Op{Kind: hostclaim.KindServing, Participants: []string{"box"}}
	r.mu.Unlock()
	if err = ClearStale(t.Context(), cfg); err != nil {
		t.Fatal(err)
	}
	if _, err = os.Stat(filepath.Join(cfg.Guard.DataDir, ".hostexec", cfg.Binding.Host+".json")); !errors.Is(err, os.ErrNotExist) {
		t.Fatal("stale state remains", err)
	}
	// No lease traffic for retained serving participants, even when enabled.
	if _, err = Open(t.Context(), cfg); !errors.Is(err, ErrInactive) {
		t.Fatalf("serving controller = %v", err)
	}
}

func TestMalformedPendingMissingAndQuarantineRefuse(t *testing.T) {
	cfg, _, clock, _, _ := fixture(t)
	b := cfg.Binding
	b.ControllerID = "controller"
	if err := CheckLocal(cfg.Guard.DataDir, b, clock, cfg.LockDir); err == nil {
		t.Fatal("missing state admitted")
	}
	p := filepath.Join(cfg.Guard.DataDir, ".hostexec", b.Host+".json")
	if err := os.WriteFile(p, []byte(`{"v":1}`), 0660); err != nil {
		t.Fatal(err)
	}
	if _, err := Open(t.Context(), cfg); err == nil {
		t.Fatal("malformed state replaced")
	}
	if err := os.Remove(p); err != nil {
		t.Fatal(err)
	}
	cfg.Directory = directory{state: "quarantined"}
	if _, err := Open(t.Context(), cfg); err == nil {
		t.Fatal("quarantined box admitted")
	}
}

func TestLocalPathsPinnedAndSymlinkRefused(t *testing.T) {
	cfg, _, _, _, _ := fixture(t)
	outside := filepath.Join(t.TempDir(), "outside")
	if err := os.WriteFile(outside, []byte("untouched"), 0600); err != nil {
		t.Fatal(err)
	}
	record := filepath.Join(cfg.Guard.DataDir, ".hostexec", cfg.Binding.Host+".json")
	if err := os.Symlink(outside, record); err != nil {
		t.Fatal(err)
	}
	if _, err := Open(t.Context(), cfg); err == nil {
		t.Fatal("symlink admitted")
	}
	b, _ := os.ReadFile(outside)
	st, _ := os.Stat(outside)
	if string(b) != "untouched" || st.Mode().Perm() != 0600 {
		t.Fatal("changed symlink target")
	}
	if err := os.Remove(record); err != nil {
		t.Fatal(err)
	}
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	old := filepath.Join(cfg.Guard.DataDir, ".hostexec")
	if err = os.Rename(old, old+".pinned"); err != nil {
		t.Fatal(err)
	}
	if err = os.Mkdir(old, 0770); err != nil {
		t.Fatal(err)
	}
	s := State{V: 1, Binding: cfg.Binding, State: "pending", Expires: time.Now().Add(TTL), SafeUntil: 99}
	s.ControllerID = "controller"
	if err = d.write(s); err != nil {
		t.Fatal(err)
	}
	if _, err = os.Stat(filepath.Join(old+".pinned", s.Host+".json")); err != nil {
		t.Fatal(err)
	}
	if _, err = os.Stat(filepath.Join(old, s.Host+".json")); !errors.Is(err, os.ErrNotExist) {
		t.Fatal("followed directory replacement")
	}
	d.sync = func(int) error { return errors.New("fsync failure") }
	if err = d.write(s); err == nil {
		t.Fatal("ignored persistence failure")
	}
	entries, _ := os.ReadDir(old + ".pinned")
	for _, e := range entries {
		if strings.HasPrefix(e.Name(), ".tmp-") {
			t.Fatal("leaked temp file")
		}
	}
}

func TestSuccessfulRenewalDurablyExtendsFromItsRequestStart(t *testing.T) {
	cfg, _, clock, store, _ := fixture(t)
	c, err := Open(t.Context(), cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer closeController(t, c)
	first := 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)
	until(t, func() bool {
		c.mu.Lock()
		defer c.mu.Unlock()
		return c.state.SafeUntil == first.SafeUntil+int64(Renewal)
	})
	renewed := ready(t, c)
	if renewed.LeaseETag == first.LeaseETag || !renewed.Expires.Equal(first.Expires.Add(Renewal)) {
		t.Fatalf("renewal did not advance coherent tuple: %+v", renewed)
	}
	d, err := openLocal(cfg.Guard.DataDir)
	if err != nil {
		t.Fatal(err)
	}
	defer d.close()
	disk, err := d.read(renewed.Host)
	if err != nil || disk != renewed {
		t.Fatalf("renewal not durable: %+v %v", disk, err)
	}
	key, _ := Key(renewed.Host, renewed.Box)
	raw, etag, err := store.Get(t.Context(), key)
	if err != nil || etag != renewed.LeaseETag {
		t.Fatalf("renewed remote ownership: %q %v", etag, err)
	}
	var wire Record
	if err = json.Unmarshal(raw, &wire); err != nil || wire.Op != renewed.Op || !wire.Expires.Equal(renewed.Expires) {
		t.Fatalf("renewed wire: %+v %v", wire, err)
	}
}