//go:build linux

package host

import (
	"context"
	"errors"
	"math"
	"strconv"
	"strings"
	"testing"

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

// Fix round 1, finding 3: root bounds an agent-sent RosterRevision by
// 0 < n <= the revision of the roster it reads and verifies itself at that
// moment, before any durable fence, drain or receipt. Through the real
// instance-control assembly: rootStopCoordinator's bound is the cleanup
// enrollment's fresh read of the signed roster. A future numeral
// (math.MaxUint64, or the roster's revision + 1) is refused for an
// operation stop and for the no-process witness, with no fence, no drain
// and nothing published; the roster's own revision passes, and a retry
// after roster writes keeps the original numeral (B2). While quarantined
// the fresh read still bounds historical cleanup.
func TestStopBoundsRosterNumeralByRootsRosterRead(t *testing.T) {
	ctx := t.Context()
	w := newAssemblyWorld(t)
	w.mustStart()
	b0 := w.fresh()
	recA, routeA := w.newHost("a")
	recB, routeB := w.newHost("b")
	for _, rec := range []struct {
		host  string
		agent func() error
	}{{recA.Host, func() error { return w.agent(b0).install(recA, routeA) }}, {recB.Host, func() error { return w.agent(b0).install(recB, routeB) }}} {
		if err := rec.agent(); err != nil {
			t.Fatal(rec.host, err)
		}
	}
	drains := 0
	inner := w.root.Stop.Coordinator.Drainer
	w.root.Stop.Coordinator.Drainer = stopDrainFunc(func(ctx context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) {
		drains++
		return inner.DrainClaim(ctx, req, fence)
	})
	published := func(host string) int {
		t.Helper()
		keys, err := w.store.List(ctx, "control/hosts/"+host+"/")
		if err != nil {
			t.Fatal(err)
		}
		n := 0
		for _, k := range keys {
			if strings.HasPrefix(k, "control/hosts/"+host+"/evidence/") || strings.HasPrefix(k, "control/hosts/"+host+"/serving-evidence/") {
				n++
			}
		}
		return n
	}
	current := w.roster.read().Roster.Revision

	// Operation stop of A's writer session.
	o := w.operationSession(recA.Host)
	captured := hostguard.Captured{Incarnation: o.Claim.Holder.Incarnation, Stamp: o.Claim.Holder.Stamp}
	for _, n := range []uint64{math.MaxUint64, current + 1} {
		req := w.priorStop(o, b0.Binding)
		req.RosterRevision = n
		if _, err := w.root.Stop.Coordinator.FenceAndStop(ctx, req); !errors.Is(err, ErrClaimStopIdentity) {
			t.Fatalf("operation stop with numeral %d: %v", n, err)
		}
		if v := w.view(recA.Host, captured); v.Pending != nil || len(v.Receipts) != 0 || drains != 0 || published(recA.Host) != 0 {
			t.Fatalf("numeral %d left effects: pending %v receipts %d drains %d published %d", n, v.Pending, len(v.Receipts), drains, published(recA.Host))
		}
	}
	req := w.priorStop(o, b0.Binding)
	req.RosterRevision = current
	stopped, err := w.root.Stop.Coordinator.FenceAndStop(ctx, req)
	if err != nil {
		t.Fatal("operation stop with the roster's revision:", err)
	}
	if r := decodeStopReceipt(t, stopped.Bytes); r.DirectoryRevision != strconv.FormatUint(current, 10) {
		t.Fatalf("receipt numeral %q, want %d", r.DirectoryRevision, current)
	}
	// Roster writes, then a retry with the newer numeral: the original
	// numeral and receipt (B2).
	w.roster.holdChurn()
	newer := w.roster.read().Roster.Revision
	if newer <= current {
		t.Fatalf("hold churn did not move the roster: %d -> %d", current, newer)
	}
	req.RosterRevision = newer
	again, err := w.root.Stop.Coordinator.FenceAndStop(ctx, req)
	if err != nil || string(again.Bytes) != string(stopped.Bytes) {
		t.Fatalf("retry after roster writes: %v (same receipt %v)", err, string(again.Bytes) == string(stopped.Bytes))
	}

	// No-process witness for B (session zero).
	before := published(recB.Host)
	for _, n := range []uint64{math.MaxUint64, newer + 1} {
		if _, err := w.noProcessStop(recB.Host, b0.Binding, n); !errors.Is(err, ErrClaimStopIdentity) {
			t.Fatalf("no-process witness with numeral %d: %v", n, err)
		}
		if published(recB.Host) != before {
			t.Fatalf("numeral %d published a no-process receipt", n)
		}
	}

	// Quarantined: the fresh read still bounds historical cleanup.
	w.roster.quarantine("loss_1")
	quarantined := w.roster.read().Roster.Revision
	if _, err := w.noProcessStop(recB.Host, b0.Binding, quarantined+1); !errors.Is(err, ErrClaimStopIdentity) {
		t.Fatal("quarantined no-process witness with a future numeral:", err)
	}
	witnessed, err := w.noProcessStop(recB.Host, b0.Binding, quarantined)
	if err != nil {
		t.Fatal("quarantined no-process witness with the roster's revision:", err)
	}
	if r := decodeStopReceipt(t, witnessed.Bytes); r.DirectoryRevision != strconv.FormatUint(quarantined, 10) {
		t.Fatalf("quarantined receipt numeral %q, want %d", r.DirectoryRevision, quarantined)
	}
	cleanup := w.priorStop(o, b0.Binding)
	cleanup.Cleanup, cleanup.RosterRevision = true, quarantined
	if again, err := w.root.Stop.Coordinator.FenceAndStop(ctx, cleanup); err != nil || string(again.Bytes) != string(stopped.Bytes) {
		t.Fatalf("quarantined cleanup stop: %v", err)
	}
	cleanup.RosterRevision = math.MaxUint64
	if _, err := w.root.Stop.Coordinator.FenceAndStop(ctx, cleanup); !errors.Is(err, ErrClaimStopIdentity) {
		t.Fatal("quarantined cleanup stop with a future numeral:", err)
	}
}