//go:build linux

package hostexec

import (
	"context"
	"fmt"
	"testing"

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

func TestPlan14ExecutionReplayWindowLongSequence(t *testing.T) {
	for _, manyOps := range []bool{false, true} {
		t.Run(fmt.Sprint(manyOps), func(t *testing.T) {
			b, d, remote, _, base, peer := brokerFixture(t)
			original := b.cfg.Authority
			current := base
			b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
				if !sameExecution(q, ExecutionRecord{Request: current}) {
					return ExecutionAuthority{}, ErrLost
				}
				q.Op = base.Op
				return original(ctx, q, p)
			}
			var stale ExecutionRequest
			var staleSettle ExecutionRequest
			begin := func(q ExecutionRequest) ExecutionResponse {
				t.Helper()
				r, err := b.Execute(t.Context(), peer, q)
				if err != nil {
					t.Fatal(err)
				}
				return r
			}
			permit := begin(current)
			for i := 0; i < 4200; i++ {
				q := current
				q.Operation = "executionRenew"
				q.ID = fmt.Sprintf("renew-%d", i)
				q.Token = permit.Token
				permit = begin(q)
				if i == 0 {
					stale = q
				}
				again := begin(q)
				if again != permit {
					t.Fatal("latest request replay changed")
				}
				if manyOps {
					settle := current
					settle.Operation = "executionSettle"
					settle.ID = fmt.Sprintf("settle-%d", i)
					settle.Token = permit.Token
					begin(settle)
					if i == 0 {
						staleSettle = settle
					}
					// A different ID cannot reopen a just-settled operation.
					repeat := current
					repeat.ID = "reopen"
					if _, err := b.Execute(t.Context(), peer, repeat); err == nil {
						t.Fatal("settled operation reopened")
					}
					current.Op = fmt.Sprintf("operation-%d", i)
					current.ID = fmt.Sprintf("begin-%d", i)
					permit = begin(current)
					if _, err := b.Execute(t.Context(), peer, settle); err != nil {
						t.Fatal("previous settlement lost", err)
					}
				}
				raw, err := d.Read(current.Host + ".json")
				if err != nil {
					t.Fatal(err)
				}
				if len(raw) >= rootstate.MaxRecord/16 {
					t.Fatal("unbounded execution record", len(raw))
				}
				r, err := b.read(current.Host)
				if err != nil || len(r.Replay) > 2 {
					t.Fatal("unbounded replay", len(r.Replay), err)
				}
			}
			calls := remote.calls
			for _, q := range []ExecutionRequest{stale, base, staleSettle} {
				if q.ID == "" {
					continue
				}
				if _, err := b.Execute(t.Context(), peer, q); err == nil {
					t.Fatal("stale request accepted", q.Operation)
				}
			}
			if remote.calls != calls {
				t.Fatal("stale request made remote write")
			}
			// Persisted bounds survive reconstruction of the broker.
			if err := b.Close(t.Context()); err != nil {
				t.Fatal(err)
			}
			fresh, err := NewRootBroker(b.cfg)
			if err != nil {
				t.Fatal(err)
			}
			defer fresh.Close(t.Context())
			if r, err := fresh.read(current.Host); err != nil || len(r.Replay) > 2 {
				t.Fatal("reload lost bounds", err)
			}
		})
	}
}