package host

import (
	"context"
	"errors"
	"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"
)

// plan9Fix1StopAtCommitted admits a probe at probing through the real
// runtime, then moves the stored op to probed and committed the way the moves
// controller does, under KeepWriterSession and without touching the attempt.
// The attempt's checkpoint still names probing.
func plan9Fix1StopAtCommitted(t *testing.T) (*recoveryFixture, *recoveryController, *claimWakeAttempt, hostclaim.Observed) {
	t.Helper()
	probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
	f, c, _, _, admitted := plan9D2Landing(t, hostclaim.PhaseLandingProbing, probe, true)
	a := c.wake.get(recoveryHost)
	if a == nil {
		t.Fatal("no pending attempt after admission")
	}
	for _, phase := range []string{hostclaim.PhaseLandingProbed, hostclaim.PhaseLandingCommitted} {
		if _, err := hostclaim.Update(t.Context(), f.claims, f.read(), hostclaim.KeepWriterSession("m1", func(c *hostclaim.Claim) error {
			c.Op.Phase = phase
			return nil
		})); err != nil {
			t.Fatalf("move to %s: %v", phase, err)
		}
	}
	if checkpoint, _ := a.snapshot(); checkpoint.Claim.Op.Phase != hostclaim.PhaseLandingProbing {
		t.Fatalf("checkpoint moved to %q", checkpoint.Claim.Op.Phase)
	}
	return f, c, a, admitted
}

// R1: identifying a session in order to stop it grants no write authority.
// A probe admitted at probing whose op the moves controller took to committed
// is still stopped as itself by every teardown path: root fences and stops
// that exact session at committed, the signed receipt verifies, and the
// pending lease is released so the host's next admission is not refused.
func TestPlan9Fix1TeardownStopsAProbeThatReachedCommitted(t *testing.T) {
	t.Parallel()
	for name, teardown := range map[string]func(context.Context, *recoveryController, hostclaim.Observed, hostclaim.Observed) error{
		"stop": func(ctx context.Context, c *recoveryController, checkpoint, _ hostclaim.Observed) error {
			return c.wake.Stop(ctx, checkpoint)
		},
		"close": func(ctx context.Context, c *recoveryController, _, _ hostclaim.Observed) error {
			return c.wake.Close(ctx)
		},
		"failed-admission": func(ctx context.Context, c *recoveryController, _, admitted hostclaim.Observed) error {
			return c.runtime.FailedAdmission(ctx, admitted, errors.New("restore failed"))
		},
	} {
		t.Run(name, func(t *testing.T) {
			f, c, a, admitted := plan9Fix1StopAtCommitted(t)
			ctx := t.Context()
			checkpoint, _ := a.snapshot()
			if err := teardown(ctx, c, checkpoint, admitted); err != nil {
				t.Fatalf("teardown refused the probe's own session at committed: %v", err)
			}
			h := admitted.Claim.Holder
			if len(f.requests) != 1 {
				t.Fatalf("root requests = %d, want 1", len(f.requests))
			}
			req := f.requests[0]
			if req.Recover || req.Phase != hostclaim.PhaseLandingCommitted || req.OperationID != "m1" || req.WriterSession != h.Session || req.Incarnation != h.Incarnation || req.InstallationStamp != h.Stamp || req.ObligationID != hostguard.WriterStopObligation(h.Session) || req.Box != "box-1" || req.ControllerID != c.local.ControllerID {
				t.Fatalf("root was asked to stop another tuple: %+v", req)
			}
			receipts, err := f.guard.CompletedWriterStops(ctx, recoveryHost, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}, "m1", "box-1")
			if err != nil || len(receipts) != 1 || receipts[0].Target.Session != h.Session || receipts[0].Target.Phase != hostclaim.PhaseLandingCommitted || verifyRetirementReceipt(ctx, receipts[0]) != nil {
				t.Fatalf("no verified writer-stop receipt for session %d: %+v %v", h.Session, receipts, err)
			}
			if c.wake.get(recoveryHost) != nil || !a.released || a.ctx.Err() == nil {
				t.Fatalf("pending lease kept: pending=%v released=%v", c.wake.get(recoveryHost) != nil, a.released)
			}
		})
	}
}

// A teardown never follows a session backward: a checkpoint at committed
// does not stop a claim now back at probing, and root is never asked.
func TestPlan9Fix1TeardownRefusesABackwardPhase(t *testing.T) {
	f, c, a, _ := plan9Fix1StopAtCommitted(t)
	ctx := t.Context()
	committed := f.read()
	a.mu.Lock()
	a.observed = committed
	a.mu.Unlock()
	if _, err := hostclaim.Update(ctx, f.claims, committed, func(c *hostclaim.Claim) error {
		c.Op.Phase = hostclaim.PhaseLandingProbing
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if err := c.wake.Stop(ctx, committed); !errors.Is(err, hostbox.ErrChanged) || len(f.requests) != 0 {
		t.Fatalf("stop followed the session backward: %v requests=%d", err, len(f.requests))
	}
	if err := c.wake.Close(ctx); !errors.Is(err, hostbox.ErrChanged) || len(f.requests) != 0 || c.wake.get(recoveryHost) == nil {
		t.Fatalf("close followed the session backward: %v requests=%d pending=%v", err, len(f.requests), c.wake.get(recoveryHost) != nil)
	}
}

func TestPlan9Fix1StopReleasesCheckpointThatAdvancedAfterCapture(t *testing.T) {
	f, c, a, _ := plan9Fix1StopAtCommitted(t)
	captured, _ := a.snapshot()
	// Another successful check advances the pending attempt after teardown
	// captured its identity. Root still verifies and stops the same session.
	a.mu.Lock()
	a.observed = f.read()
	a.mu.Unlock()
	if err := c.wake.Stop(t.Context(), captured); err != nil {
		t.Fatal(err)
	}
	if len(f.requests) != 1 || c.wake.get(recoveryHost) != nil || !a.released {
		t.Fatalf("stopped session kept an advanced checkpoint: requests=%d pending=%v released=%v", len(f.requests), c.wake.get(recoveryHost) != nil, a.released)
	}
}