package host

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path"
	"path/filepath"
	"reflect"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
)

// The cut's files preflight (#236; Paul's ruling of 2026-10-15): a source
// whose files manifest a recovery could not restore, because an entry path
// is past the readers' cap, is refused at S1 before anything is staged,
// stopped or cut.

// capPath is a clean relative path of exactly n bytes: 200-byte segments
// joined by "/", as hostfiles' own tests build them.
func capPath(n int) string {
	var b strings.Builder
	for b.Len() < n {
		if b.Len() > 0 {
			b.WriteByte('/')
		}
		b.WriteString(strings.Repeat("x", min(200, n-b.Len())))
	}
	return b.String()
}

// capCase is one manifest shape: its one entry's kind (live, tombstone or
// skipped) at a path of the given length.
type capCase struct {
	kind string
	n    int
}

// capCases are the shapes every test here runs: an ordinary manifest, a
// live entry at the cap, and each kind of entry one byte past it.
var capCases = map[string]capCase{
	"an ordinary manifest":    {"live", len("a.txt")},
	"a live entry at the cap": {"live", hostfiles.MaxPathBytes},
	"a live entry past it":    {"live", hostfiles.MaxPathBytes + 1},
	"a tombstone past it":     {"tombstone", hostfiles.MaxPathBytes + 1},
	"a skipped entry past it": {"skipped", hostfiles.MaxPathBytes + 1},
}

func (c capCase) path() string {
	if c.n == len("a.txt") {
		return "a.txt"
	}
	return capPath(c.n)
}

// manifest is the case's manifest for generation gen at location loc: a
// live entry names an object under gen's files prefix, as the publisher
// writes it.
func (c capCase) manifest(loc, gen string) hostfiles.Manifest {
	var e hostfiles.Entry
	switch c.kind {
	case "live":
		sum := sha([]byte("A"))
		e = hostfiles.Entry{Object: movesFilesPrefix(loc, gen) + fmt.Sprintf("objects/%016d-%s", 1, sum), SHA256: sum, Size: 1, Mode: 0o600, Seq: 1}
	case "tombstone":
		e = hostfiles.Entry{Seq: 1, Deleted: true}
	case "skipped":
		e = hostfiles.Entry{Seq: 1, Skipped: "too-large"}
	}
	return hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 1, Entries: map[string]hostfiles.Entry{c.path(): e}, Platform: map[string]hostfiles.Entry{}}
}

// readerRefuses is what the files readers do with m today: ValidateManifest
// (checkManifest, the establishment's reader) refuses it as ErrTooDeep. The
// tests hold the cut to this verdict rather than to a fixed answer per
// kind: today the readers cap every entry, tombstones and skipped entries
// included, so the cut refuses all three past the cap. If region HA
// relaxes the readers for tombstones and skipped entries (its part of
// #236), RestorableWithinCap follows, and the cut with it.
func readerRefuses(m hostfiles.Manifest, loc, gen string) bool {
	return errors.Is(hostfiles.ValidateManifest(m, []string{movesFilesPrefix(loc, gen)}), hostfiles.ErrTooDeep)
}

// putP1Manifest stores m as the current version of gen's files manifest in
// the Phase 1 harness's data bucket, written directly (a manifest from
// before the cap), never through the staging mirror.
func putP1Manifest(t *testing.T, h *phase1Harness, loc, gen string, m hostfiles.Manifest) {
	t.Helper()
	raw, err := json.Marshal(m)
	if err != nil {
		t.Fatal(err)
	}
	key := hostfiles.ManifestKey(movesFilesPrefix(loc, gen))
	h.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "manifest-" + gen, Size: int64(len(raw))}, string(raw)})
	h.f.setVersion("data-ie", key, "ver-manifest-"+gen)
}

// A source whose files manifest has an entry past the cap is refused at S1
// with MV-N501, and nothing is staged: no staging request, no touch of the
// target in the claim, the move still preparing. A manifest the readers
// accept passes, and the round stages as before.
func TestPreparingRefusesAFilesManifestPastTheCap(t *testing.T) {
	t.Parallel()
	for name, c := range capCases {
		t.Run(name, func(t *testing.T) {
			h := newPhase1Harness(t)
			h.start(t, "manual")
			loc := "instances/" + p1Host + "/" + h.inc
			m := c.manifest(loc, "cur")
			putP1Manifest(t, h, loc, "cur", m)
			touched := h.claim(t).Touched
			h.pass(t)
			sr := h.sourceReport(t)
			if sr == nil {
				t.Fatal("no source report")
			}
			refuses := readerRefuses(m, loc, "cur")
			if c.kind == "live" && refuses != (c.n > hostfiles.MaxPathBytes) {
				t.Fatalf("the reader's verdict on a live entry of %d bytes is %v", c.n, refuses)
			}
			if !refuses {
				if !sr.S1.OK || len(h.f.stagingCalls()) == 0 {
					t.Fatalf("a manifest the readers accept: S1 %+v, %d staging calls", sr.S1, len(h.f.stagingCalls()))
				}
				return
			}
			if sr.S1.OK || sr.S1.Reason != s1FilesPastCap(p1Host) {
				t.Fatalf("S1 = %+v, want %q", sr.S1, s1FilesPastCap(p1Host))
			}
			if calls := h.f.stagingCalls(); len(calls) != 0 {
				t.Fatalf("staged despite the refusal: %+v", calls)
			}
			if got := h.claim(t).Touched; !reflect.DeepEqual(got, touched) {
				t.Fatalf("the claim was touched: %+v, was %+v", got, touched)
			}
			if st := h.read(t, p1Host).Move.State; st != provision.MovePreparing {
				t.Fatalf("the move is %s", st)
			}
		})
	}
}

// A files manifest that cannot be read ends the round as the other read
// errors do: no source report, nothing staged, the reason in the log.
func TestPreparingEndsTheRoundWhenTheFilesManifestCannotBeRead(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.vs.deny = func(op, _, key string) error {
		if op == "list-versions" && strings.Contains(key, "/files/g-") {
			return errors.New("injected 403")
		}
		return nil
	}
	logs := captureLog(t)
	h.pass(t)
	if sr := h.sourceReport(t); sr != nil {
		t.Fatalf("a report was written: %+v", sr)
	}
	if calls := h.f.stagingCalls(); len(calls) != 0 {
		t.Fatalf("staged: %+v", calls)
	}
	if !strings.Contains(logs(), "read the files manifest") {
		t.Fatalf("no log of the read:\n%s", logs())
	}
}

// stallingVersioned is the harness's versioned reader with the files
// manifest's download stalled: GetVersion on a manifest key blocks until
// its context ends, as an S3 body that stops arriving does, and answers
// the context's error.
type stallingVersioned struct {
	*fakeVersioned
	stalled chan struct{} // closed when the first stalled read starts
	once    sync.Once
}

func (v *stallingVersioned) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
	if strings.Contains(key, "/files/g-") && strings.HasSuffix(key, "/manifest.json") {
		v.once.Do(func() { close(v.stalled) })
		<-ctx.Done()
		return nil, ctx.Err()
	}
	return v.fakeVersioned.GetVersion(ctx, bucket, key, versionID)
}

// Fix round 1, P2 1: the preflight's reads are bounded
// (movesFilesPreflightTimeout). A stalled manifest download is cancelled at
// the bound, even under a pass with no deadline of its own; the round ends
// with no report and nothing staged; the staging lease is released and the
// pass mutex is free, so the next pass runs.
func TestPreparingBoundsAStalledFilesManifestRead(t *testing.T) {
	const bound = 50 * time.Millisecond
	orig := movesFilesPreflightBound
	movesFilesPreflightBound = bound
	defer func() { movesFilesPreflightBound = orig }()
	if movesFilesPreflightTimeout <= bound || movesFilesPreflightTimeout > hostbox.RequestLimit {
		t.Fatalf("the production bound %v is not a short one inside the claim's request limit", movesFilesPreflightTimeout)
	}

	h := newPhase1Harness(t)
	h.start(t, "manual")
	loc := "instances/" + p1Host + "/" + h.inc
	putP1Manifest(t, h, loc, "cur", capCases["an ordinary manifest"].manifest(loc, "cur"))
	sv := &stallingVersioned{fakeVersioned: h.vs, stalled: make(chan struct{})}
	h.cfg.Versioned = func(context.Context, string, string) (lease.VersionedStore, error) { return sv, nil }
	logs := captureLog(t)

	done := make(chan error, 1)
	start := time.Now()
	go func() { done <- movesPass(context.Background(), h.s, h.w()) }()
	select {
	case <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("the pass is still blocked on the stalled read")
	}
	if elapsed := time.Since(start); elapsed < bound {
		t.Fatalf("the pass ended after %v, before the bound %v", elapsed, bound)
	}
	select {
	case <-sv.stalled:
	default:
		t.Fatal("the manifest download was never asked for")
	}
	if sr := h.sourceReport(t); sr != nil {
		t.Fatalf("a report was written: %+v", sr)
	}
	if calls := h.f.stagingCalls(); len(calls) != 0 {
		t.Fatalf("staged: %+v", calls)
	}
	if !strings.Contains(logs(), context.DeadlineExceeded.Error()) {
		t.Fatalf("no log of the cancelled read:\n%s", logs())
	}
	h.lease.mu.Lock()
	acquired, released := h.lease.acquired, h.lease.released
	h.lease.mu.Unlock()
	if acquired == 0 || acquired != released {
		t.Fatalf("staging lease acquired %d, released %d", acquired, released)
	}
	if !h.w().pass.TryLock() {
		t.Fatal("the pass mutex is still held")
	}
	h.w().pass.Unlock()

	// The download recovers: the next pass reports and stages.
	h.cfg.Versioned = func(context.Context, string, string) (lease.VersionedStore, error) { return h.vs, nil }
	h.pass(t)
	if sr := h.sourceReport(t); sr == nil || !sr.S1.OK || len(h.f.stagingCalls()) == 0 {
		t.Fatalf("the next pass: %+v, %d staging calls", sr, len(h.f.stagingCalls()))
	}
}

// Fix round 1, P2 2: the files directory itself is held to the tree's cap.
// A file whose path inside it is 4096 bytes (the archive format's own
// bound, which used to pass) is refused at S1 with MV-N502 before anything
// is staged; one of 4095 bytes passes and its archive is staged. The
// files directory is under the data directory, so both absolute paths are
// past PATH_MAX: the archive reads through an os.Root.
func TestPreparingRefusesAFilePathPastTheTreeCap(t *testing.T) {
	for _, n := range []int{hostfiles.MaxPathBytes + 1, hostfiles.MaxPathBytes} {
		t.Run(fmt.Sprint(n), func(t *testing.T) {
			h := newPhase1Harness(t)
			h.start(t, "manual")
			root, err := os.OpenRoot(filepath.Join(h.cfg.DataDir, p1Host))
			if err != nil {
				t.Fatal(err)
			}
			defer root.Close()
			leaf := capPath(n)
			if err := root.MkdirAll(path.Dir(leaf), 0o755); err != nil {
				t.Fatal(err)
			}
			if err := root.WriteFile(leaf, []byte("deep"), 0o644); err != nil {
				t.Fatal(err)
			}
			h.pass(t)
			sr := h.sourceReport(t)
			if sr == nil {
				t.Fatal("no source report")
			}
			if n > hostfiles.MaxPathBytes {
				if sr.S1.OK || sr.S1.Reason != s1FilePastCap(p1Host) {
					t.Fatalf("S1 = %+v, want %q", sr.S1, s1FilePastCap(p1Host))
				}
				if calls := h.f.stagingCalls(); slices.ContainsFunc(calls, func(c stagingCall) bool { return strings.HasPrefix(c.Key, moves.FilesPrefix(p1Host, p1ID)) }) {
					t.Fatalf("an archive was staged: %+v", calls)
				}
				return
			}
			if !sr.S1.OK || sr.Archive == nil {
				t.Fatalf("a path at the cap: S1 %+v, archive %+v", sr.S1, sr.Archive)
			}
		})
	}
}

// Fix round 1, P2 2(b): a path at the tree's cap goes end to end. The
// source's archive carries it, the target's Extract places it, and the
// prepared files manifest build stages it (hostfiles.Prepared.Stage), with
// the extracted tree's absolute paths past PATH_MAX: a 4095-byte file, a
// deep tree of files near it, and an ordinary one. The build reads through
// an os.Root over the files directory, every name relative to it.
func TestATreeCapPathReachesThePreparedManifest(t *testing.T) {
	ctx := context.Background()
	x := newFilesFixture(t)
	src := filepath.Join(t.TempDir(), "h")
	if err := os.Mkdir(src, 0o755); err != nil {
		t.Fatal(err)
	}
	root, err := os.OpenRoot(src)
	if err != nil {
		t.Fatal(err)
	}
	defer root.Close()
	files := map[string]string{"a.txt": "A"}
	files[capPath(hostfiles.MaxPathBytes)] = "at the cap"
	for _, n := range []int{hostfiles.MaxPathBytes - 1, 4000, 3900} {
		files[capPath(n)+"-"+fmt.Sprint(n)] = fmt.Sprint(n) // under one deep tree
	}
	for rel, body := range files {
		if len(rel) > hostfiles.MaxPathBytes {
			rel = rel[:hostfiles.MaxPathBytes]
		}
		if err := root.MkdirAll(path.Dir(rel), 0o755); err != nil {
			t.Fatal(err)
		}
		if err := root.WriteFile(rel, []byte(body), 0o644); err != nil {
			t.Fatal(err)
		}
	}
	var tarz bytes.Buffer
	if _, _, _, err := archive.Build(src, "", &tarz); err != nil {
		t.Fatalf("the source's archive: %v", err)
	}
	dst := filepath.Join(x.dir, filesHost)
	if len(dst)+1+hostfiles.MaxPathBytes <= 4096 {
		t.Fatalf("%s does not put the tree past PATH_MAX", dst)
	}
	if err := archive.Extract(bytes.NewReader(tarz.Bytes()), dst, x.envPath()); err != nil {
		t.Fatalf("the target's extract: %v", err)
	}
	if _, err := x.f.BuildPrepared(ctx, filesLoc, "g1", "g1", 7, dst, x.envPath(), false, hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}}); err != nil {
		t.Fatalf("the prepared manifest: %v", err)
	}
	m, found := x.manifest("g1")
	if !found {
		t.Fatal("no prepared manifest")
	}
	staged := 0
	err = fs.WalkDir(root.FS(), ".", func(rel string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return err
		}
		body, err := root.ReadFile(rel)
		if err != nil {
			return err
		}
		if e, ok := m.Entries[rel]; !ok || e.SHA256 != sha(body) || e.Skipped != "" {
			t.Errorf("%d-byte path: entry %+v, %v", len(rel), e, ok)
		}
		staged++
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if staged != len(files) || len(m.Entries) != len(files) {
		t.Fatalf("%d files, %d walked, %d entries", len(files), staged, len(m.Entries))
	}
	if err := movesFilesRestorable(m); err != nil {
		t.Fatalf("the prepared manifest is not restorable: %v", err)
	}
}

// The cut's requested round runs the same source round (G13): a manifest
// past the cap that appeared after readiness sends the cut back to ready
// before anything holds, fences or stops, with MV-N501 in S1, and the move
// is not pinned again while it stands. Once an operator replaces the
// manifest, a passing round pins it again and the cut admits.
func TestCutRefusesAFilesManifestPastTheCap(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	h.ev.stopped = false // stop at stopping, to look at the admission
	past := capCases["a live entry past it"]
	putP1Manifest(t, h.phase1Harness, cutRoot, "cur", past.manifest(cutRoot, "cur"))
	h.cut(t)
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.CutAt != nil {
		t.Fatalf("not back to ready: %+v", m)
	}
	if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 {
		t.Fatalf("something stopped: %d evidence calls, holds %v, %d fences", h.ev.calls, h.holds, h.fences)
	}
	if sr := h.sourceReport(t); sr == nil || sr.S1.OK || sr.S1.Reason != s1FilesPastCap(p1Host) {
		t.Fatalf("S1 = %+v, want %q", sr, s1FilesPastCap(p1Host))
	}
	h.targetPass(t, h.target())
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveReady || m.Inputs != nil {
		t.Fatalf("pinned again while the manifest stands: %+v", m)
	}

	// The operator replaces the manifest.
	putP1Manifest(t, h.phase1Harness, cutRoot, "cur", capCases["an ordinary manifest"].manifest(cutRoot, "cur"))
	// A refused report carries no archive, so the source reports once, the
	// target checks that report, and the next round pins.
	h.pass(t)
	h.targetPass(t, h.target())
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveReady || m.Inputs == nil {
		t.Fatalf("not pinned again after the replacement: %+v", m)
	}
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping {
		t.Fatalf("the cut did not admit after the replacement: %+v", m)
	}
}

// The cut's preflight and the recovery's establishment agree on every
// manifest shape, so they cannot drift: the source's round before the flip
// (sourceFilesRefusal over the holder at src) refuses exactly the
// manifests a recovery's Establish refuses as ErrTooDeep, through the
// src_gen bridge of a clean return and through a revert's own lineage, and
// both agree with the readers (readerRefuses). Both call
// movesFilesRestorable; a manifest the cut passes, the recovery
// establishes from.
func TestCutPreflightAndRecoveryAgreeOnTheCap(t *testing.T) {
	ctx := context.Background()
	recoveries := map[string]func(o *hostclaim.Observed){
		"a clean return": func(*hostclaim.Observed) {},
		"a revert": func(o *hostclaim.Observed) {
			o.Claim.Op.Kind = hostclaim.KindReverting
			delete(o.Claim.Op.Intent, "rollback")
			o.Claim.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "src", Region: "uk", Predecessor: "@root", PredecessorRegion: "uk", Mode: "cold", Session: 3, Box: "uk-box"},
				{Generation: "r1", Region: "uk", Predecessor: "src", PredecessorRegion: "uk", Mode: "cold", Session: 3, Box: "uk-box"}}
		},
	}
	for name, c := range capCases {
		for rname, shape := range recoveries {
			t.Run(name+", "+rname, func(t *testing.T) {
				x := newFilesFixture(t)
				x.writeEnv("K=1\n")
				m := c.manifest(filesLoc, "src")
				raw, err := json.Marshal(m)
				if err != nil {
					t.Fatal(err)
				}
				x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "src")), raw)

				// The cut's side: the source's round before the flip, its
				// holder at src.
				w := &movesWorker{cfg: moveCfg{Region: "uk", OwnBucket: "data", OwnBucketRegion: "eu-west-2",
					Versioned: func(context.Context, string, string) (lease.VersionedStore, error) { return x.bucket, nil }}}
				pre := servingObserved("src", "src:uk:@root:uk")
				why, err := w.sourceFilesRefusal(ctx, provision.Record{Host: filesHost}, pre.Claim.Holder)
				if err != nil {
					t.Fatalf("the preflight: %v", err)
				}
				if why != "" && why != s1FilesPastCap(filesHost) {
					t.Fatalf("the preflight's reason %q", why)
				}

				// The recovery's side: the first recovery admission after
				// the cut, whose src_gen is the holder's generation then.
				o := recoveryObserved(hostclaim.KindRestoring, "src", "mv1", "r1")
				shape(&o)
				o = x.observe(o, "src", "r1")
				est := x.f.Establish(ctx, o, launchInputs("r1"), x)
				if est != nil && !errors.Is(est, hostfiles.ErrTooDeep) {
					t.Fatalf("the establishment failed for another reason: %v", est)
				}

				cut, recovery, reader := why != "", est != nil, readerRefuses(m, filesLoc, "src")
				if cut != recovery || cut != reader {
					t.Fatalf("the cut refuses %v, the recovery refuses %v (%v), the readers refuse %v", cut, recovery, est, reader)
				}
			})
		}
	}
}

// The establishment applies the cap before any origin proof: a manifest
// past it is ErrTooDeep with no Proven call, by the same function the
// cut's preflight calls.
func TestEstablishRefusesPastTheCapBeforeTheProof(t *testing.T) {
	ctx := context.Background()
	x := newFilesFixture(t)
	x.writeEnv("K=1\n")
	m := capCases["a live entry past it"].manifest(filesLoc, "src")
	if err := movesFilesRestorable(m); !errors.Is(err, hostfiles.ErrTooDeep) {
		t.Fatalf("movesFilesRestorable: %v", err)
	}
	proof := &countingProof{proven: map[string]bool{"src": true, "r1": true}}
	x.f.Proof = proof
	if _, err := x.f.movesFilesAllowed(ctx, servingObserved("src"), launchInputs("r1").Admission, filesLoc, "r1", m); !errors.Is(err, hostfiles.ErrTooDeep) {
		t.Fatalf("movesFilesAllowed: %v", err)
	}
	if len(proof.calls) != 0 {
		t.Fatalf("the proof was asked %v", proof.calls)
	}
}

// The preflight reads by exact version through the data bucket's versioned
// reader; a holder with no location, or a chain with no manifest, has
// nothing to check.
func TestSourceFilesRefusalNeedsAManifestToRefuse(t *testing.T) {
	ctx := context.Background()
	x := newFilesFixture(t)
	w := &movesWorker{cfg: moveCfg{Region: "uk", OwnBucket: "data", OwnBucketRegion: "eu-west-2",
		Versioned: func(context.Context, string, string) (lease.VersionedStore, error) { return x.bucket, nil }}}
	pre := servingObserved("src", "src:uk:@root:uk")
	if why, err := w.sourceFilesRefusal(ctx, provision.Record{Host: filesHost}, pre.Claim.Holder); why != "" || err != nil {
		t.Fatalf("no manifest: %q, %v", why, err)
	}
	noLoc := *pre.Claim.Holder
	noLoc.Location = ""
	if why, err := w.sourceFilesRefusal(ctx, provision.Record{Host: filesHost}, &noLoc); why != "" || err != nil {
		t.Fatalf("no location: %q, %v", why, err)
	}
	w.cfg.Versioned = nil
	if _, err := w.sourceFilesRefusal(ctx, provision.Record{Host: filesHost}, pre.Claim.Holder); err == nil {
		t.Fatal("no versioned reader passed")
	}
}