package host

import (
	"context"
	"crypto/ed25519"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"testing"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rootstate"
)

// Task 7b, the agent's side: under the claim with root's relay wired, T7's
// blocked stage and every publication are asked of root (movesstage.go,
// heldReleaseHooksVia; the probe's heldReleaseProbeRelease), and the agent
// passes only on its own reading afterwards. Root's side and the fixture
// are in movesheldrelease_test.go.

// heldReleaseRelayFunc is a relay that calls root directly, as the socket
// transport would after authenticating the agent.
type heldReleaseRelayFunc func(context.Context, heldReleaseRequest) (heldReleaseResponse, error)

func (f heldReleaseRelayFunc) HeldRelease(ctx context.Context, q heldReleaseRequest) (heldReleaseResponse, error) {
	return f(ctx, q)
}

func (f *hrFixture) relay(seen *[]heldReleaseRequest) heldReleaseRelay {
	return heldReleaseRelayFunc(func(ctx context.Context, q heldReleaseRequest) (heldReleaseResponse, error) {
		if seen != nil {
			*seen = append(*seen, q)
		}
		return f.root.Serve(ctx, hrPeer, q)
	})
}

// phase1Root is root on the target box of a Phase 1 harness, over the
// harness's own claim, record and source report.
func phase1Root(t *testing.T, h *phase1Harness, opt, data string) *movesHeldReleaseRoot {
	t.Helper()
	return &movesHeldReleaseRoot{
		OpenRelease: func() (rootstate.Store, error) { return &plan14ReleaseStore{}, nil },
		Actor: func(context.Context) (hostbox.Actor, error) {
			return hostbox.Actor{Box: "uk-box", Region: "uk", Eligible: true}, nil
		},
		Store: h.s,
		Claims: func(ctx context.Context, host string) (hostclaim.Observed, error) {
			return hostclaim.Read(ctx, h.s, host)
		},
		Records: MovesRecordsFrom(h.s),
		Route:   func(string) (*registry.Route, error) { return nil, nil },
		Works: func(context.Context, hostbox.Actor, string, string) (hostwrite.Work, error) {
			t.Fatal("a stage selected")
			return nil, nil
		},
		Selected:   hostbox.Selected,
		Pub:        hsPub,
		Ratchet:    func() *release.DigestRatchet { return hsRatchet(t) },
		Proof:      func() *release.ProofGate { return offProofGate(t) },
		OptDir:     opt,
		DataDir:    data,
		ReceiptDir: t.TempDir(),
	}
}

// blockedOnce makes the agent's version store read-only, as on a hardened
// box: a stage that would write fails as EROFS does, and a version already
// staged (by root) is read and verified as release.StageHeldVerified reads
// it.
func blockedOnce(t *testing.T) {
	t.Helper()
	prev := heldStageVerified
	t.Cleanup(func() { heldStageVerified = prev })
	heldStageVerified = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, r *release.DigestRatchet, optDir string, rt registry.Route, version string, proof *release.ProofGate) (*release.Manifest, error) {
		if _, err := os.Stat(release.VersionPath(optDir, release.RouteAccount(rt), version, rt.App)); err != nil {
			return nil, fmt.Errorf("%w: read-only file system", release.ErrStageBlocked)
		}
		return prev(ctx, s, pub, r, optDir, rt, version, proof)
	}
}

// TestHeldReleaseStagesThroughRootWithoutPublishing: on a hardened box the
// agent's stage is EROFS; under the claim T7 asks root to stage, root
// stages and verifies without publishing, and T7 passes on the agent's own
// re-verification of the staged bytes. Nothing is left for the old file
// relay, and nothing is selected.
func TestHeldReleaseStagesThroughRootWithoutPublishing(t *testing.T) {
	ctx := context.Background()
	h := newPhase1Harness(t)
	rec := h.start(t, "manual")
	shipPinnedV7(t, h.s)
	h.pass(t)
	t.Setenv("CARLOS_HOST_CLAIM", "1")
	opt, data := t.TempDir(), t.TempDir()
	root := phase1Root(t, h, opt, data)
	blockedOnce(t)
	var seen []heldReleaseRequest
	relay := heldReleaseRelayFunc(func(ctx context.Context, q heldReleaseRequest) (heldReleaseResponse, error) {
		seen = append(seen, q)
		return root.Serve(ctx, hrPeer, q)
	})
	sel := &heldSelections{}
	relayedFile := false
	ready, _ := heldReleaseHooksVia(heldReleaseVia{Relay: relay, Region: "uk"}, claimWrites(sel), testRegistry(t), opt, data, "uk-box",
		hsPub, hsRatchet(t), offProofGate(t), func(string, string, error) { relayedFile = true })
	if c := ready(ctx, h.s, rec, "v7"); !c.OK {
		t.Fatalf("T7 through root = %+v", c)
	}
	if len(seen) != 1 || seen[0].Action != heldReleaseStage || seen[0].Version != "v7" || seen[0].Kind != hostclaim.KindMoving || seen[0].RenderHash != "" {
		t.Fatalf("root was asked %+v", seen)
	}
	if _, err := os.Stat(release.VersionPath(opt, "bab", "v7", "app")); err != nil {
		t.Fatalf("v7 not staged: %v", err)
	}
	if _, err := os.Lstat(release.LiveLink(opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("root's stage published live/<host>: %v", err)
	}
	if _, err := os.Stat(release.HeldStageDir(data)); !errors.Is(err, os.ErrNotExist) || relayedFile {
		t.Fatalf("a file request was left for the old relay (%v, %v)", err, relayedFile)
	}
	if len(sel.selected) != 0 {
		t.Fatalf("T7 selected %v", sel.selected)
	}
}

// TestHeldReleaseConfiguredPublishesThroughRoot: under the claim with root's
// relay wired, configured's release hook never publishes in the agent: it
// asks root with the claim's tuple and the journalled render hash, and then
// passes on its own reading. A replay finds the release live and asks
// nothing.
func TestHeldReleaseConfiguredPublishesThroughRoot(t *testing.T) {
	ctx := context.Background()
	t.Setenv("CARLOS_HOST_CLAIM", "1")
	f := newHRFixture(t)
	var seen []heldReleaseRequest
	agentSel := &heldSelections{}
	_, live := heldReleaseHooksVia(heldReleaseVia{Relay: f.relay(&seen), Region: "uk", Claims: f.claims}, claimWrites(agentSel), testRegistry(t), f.opt, f.data, "uk-box",
		hsPub, hsRatchet(t), offProofGate(t), nil)
	if c := live(ctx, f.s, f.rec, "v7"); !c.OK {
		t.Fatalf("configured's release through root = %+v", c)
	}
	if len(seen) != 1 {
		t.Fatalf("root was asked %d times", len(seen))
	}
	q := seen[0]
	if q.Action != heldReleasePublish || q.RenderHash != hrRendered || q.HoldOp != p1ID || q.MoveID != p1ID || q.Stamp != 7 || q.Session != 3 ||
		q.Generation != p1ID || q.Kind != hostclaim.KindLanding || q.Phase != provision.LandConfigured || q.Region != "uk" || q.Box != "uk-box" {
		t.Fatalf("the agent's request = %+v", q)
	}
	if len(agentSel.selected) != 0 {
		t.Fatalf("the agent selected %v itself", agentSel.selected)
	}
	if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" {
		t.Fatalf("live = %q", v)
	}
	if c := live(ctx, f.s, f.rec, "v7"); !c.OK || len(seen) != 1 {
		t.Fatalf("a replay: %+v, asked %d times", c, len(seen))
	}

	t.Run("not yet waits for the next pass", func(t *testing.T) {
		g := newHRFixture(t)
		g.root.DataDir = t.TempDir()
		_, live := heldReleaseHooksVia(heldReleaseVia{Relay: g.relay(nil), Region: "uk", Claims: g.claims}, claimWrites(&heldSelections{}), testRegistry(t), g.opt, g.data, "uk-box",
			hsPub, hsRatchet(t), offProofGate(t), nil)
		if c := live(ctx, g.s, g.rec, "v7"); c.OK {
			t.Fatalf("published with root's journal missing: %+v", c)
		}
		if v, _ := release.LiveVersion(g.opt, p1Host); v != "" {
			t.Fatalf("live = %q", v)
		}
	})
}

// TestHeldReleaseProbeReleaseAdapter: the probe's release adapter publishes
// through root when live/<host> does not name the pinned release, and asks
// nothing when it does.
func TestHeldReleaseProbeReleaseAdapter(t *testing.T) {
	ctx := context.Background()
	f := newHRFixture(t)
	f.set(func(f *hrFixture) {
		f.claim.Op.Phase = provision.LandProbing
		f.rec.Move.LandingPhase = provision.LandProbing
	})
	var seen []heldReleaseRequest
	adapter := heldReleaseProbeRelease(heldReleaseVia{Relay: f.relay(&seen), Region: "uk", Claims: f.claims}, f.s, "uk-box", f.opt, f.data, offProofGate(t))
	rt := registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", HoldOp: p1ID}
	if err := adapter(ctx, rt, "v7", heldSelectionWork{&heldSelections{}}); err != nil {
		t.Fatalf("the probe's release: %v", err)
	}
	if len(seen) != 1 || seen[0].Phase != provision.LandProbing || seen[0].Action != heldReleasePublish {
		t.Fatalf("root was asked %+v", seen)
	}
	if err := adapter(ctx, rt, "v7", heldSelectionWork{&heldSelections{}}); err != nil || len(seen) != 1 {
		t.Fatalf("a live release: %v, asked %d times", err, len(seen))
	}
}

// TestHeldReleaseOldConsumers: a root that has no held-release route (an
// older build, or the claim off) answers unavailable, and T7 says MV-N455
// with nothing staged, published or left as a file; and the old file relay,
// run after the new path staged, finds no request and publishes nothing.
func TestHeldReleaseOldConsumers(t *testing.T) {
	ctx := context.Background()

	t.Run("an old root", func(t *testing.T) {
		h := newPhase1Harness(t)
		rec := h.start(t, "manual")
		shipPinnedV7(t, h.s)
		h.pass(t)
		t.Setenv("CARLOS_HOST_CLAIM", "1")
		opt, data := t.TempDir(), t.TempDir()
		blockedOnce(t)
		old := heldReleaseRelayFunc(func(context.Context, heldReleaseRequest) (heldReleaseResponse, error) {
			return heldReleaseResponse{}, errHeldReleaseUnavailable
		})
		ready, _ := heldReleaseHooksVia(heldReleaseVia{Relay: old, Region: "uk"}, claimWrites(&heldSelections{}), testRegistry(t), opt, data, "uk-box",
			hsPub, hsRatchet(t), offProofGate(t), nil)
		if c := ready(ctx, h.s, rec, "v7"); c.OK || c.Reason != t7NotStaged {
			t.Fatalf("T7 against an old root = %+v, want %q", c, t7NotStaged)
		}
		if _, err := os.Stat(filepath.Join(opt, "versions")); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("staged: %v", err)
		}
		if _, err := os.Stat(release.HeldStageDir(data)); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("a file request was left for the old relay: %v", err)
		}
	})

	t.Run("the old file relay publishes nothing", func(t *testing.T) {
		h := newPhase1Harness(t)
		rec := h.start(t, "manual")
		shipPinnedV7(t, h.s)
		h.pass(t)
		t.Setenv("CARLOS_HOST_CLAIM", "1")
		opt, data := t.TempDir(), t.TempDir()
		root := phase1Root(t, h, opt, data)
		blockedOnce(t)
		ready, _ := heldReleaseHooksVia(heldReleaseVia{Relay: heldReleaseRelayFunc(func(ctx context.Context, q heldReleaseRequest) (heldReleaseResponse, error) {
			return root.Serve(ctx, hrPeer, q)
		}), Region: "uk"}, claimWrites(&heldSelections{}), testRegistry(t), opt, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil)
		if c := ready(ctx, h.s, rec, "v7"); !c.OK {
			t.Fatalf("T7 = %+v", c)
		}
		t.Setenv("CARLOS_HOST_CLAIM", "")
		authorized := false
		if err := release.ServeHeldStageRequestsWithPins(ctx, h.s, hsPub, hsRatchet(t), opt, data, offProofGate(t), func(release.HeldStageRequest, func() error) error {
			authorized = true
			return nil
		}); err != nil {
			t.Fatal(err)
		}
		if authorized {
			t.Fatal("the old relay found a request")
		}
		if _, err := os.Lstat(release.LiveLink(opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("the old relay published: %v", err)
		}
	})
}

// TestHeldReleaseAgentVerifiesRootsStage: root's answer to a stage is a
// hint. T7 passes only on the agent's own reading of the staged bytes, so a
// root that answers without staging never passes T7, and a stage whose
// answer was lost passes on the next pass's reading without asking root
// again.
func TestHeldReleaseAgentVerifiesRootsStage(t *testing.T) {
	ctx := context.Background()

	t.Run("root's answer is a hint", func(t *testing.T) {
		h := newPhase1Harness(t)
		rec := h.start(t, "manual")
		shipPinnedV7(t, h.s)
		h.pass(t)
		t.Setenv("CARLOS_HOST_CLAIM", "1")
		opt, data := t.TempDir(), t.TempDir()
		blockedOnce(t)
		lying := heldReleaseRelayFunc(func(_ context.Context, q heldReleaseRequest) (heldReleaseResponse, error) {
			return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomeStaged}, nil
		})
		ready, _ := heldReleaseHooksVia(heldReleaseVia{Relay: lying, Region: "uk"}, claimWrites(&heldSelections{}), testRegistry(t), opt, data, "uk-box",
			hsPub, hsRatchet(t), offProofGate(t), nil)
		if c := ready(ctx, h.s, rec, "v7"); c.OK || c.Reason != t7NotStaged {
			t.Fatalf("T7 on root's word alone = %+v, want %q", c, t7NotStaged)
		}
	})

	t.Run("a stage whose answer is lost", func(t *testing.T) {
		h := newPhase1Harness(t)
		rec := h.start(t, "manual")
		shipPinnedV7(t, h.s)
		h.pass(t)
		t.Setenv("CARLOS_HOST_CLAIM", "1")
		opt, data := t.TempDir(), t.TempDir()
		root := phase1Root(t, h, opt, data)
		blockedOnce(t)
		asked := 0
		lost := heldReleaseRelayFunc(func(ctx context.Context, q heldReleaseRequest) (heldReleaseResponse, error) {
			asked++
			if _, err := root.Serve(ctx, hrPeer, q); err != nil {
				return heldReleaseResponse{}, err
			}
			return heldReleaseResponse{}, errors.New("the connection dropped before the answer")
		})
		ready, _ := heldReleaseHooksVia(heldReleaseVia{Relay: lost, Region: "uk"}, claimWrites(&heldSelections{}), testRegistry(t), opt, data, "uk-box",
			hsPub, hsRatchet(t), offProofGate(t), nil)
		if c := ready(ctx, h.s, rec, "v7"); c.OK {
			t.Fatalf("T7 passed on a lost answer: %+v", c)
		}
		if c := ready(ctx, h.s, rec, "v7"); !c.OK {
			t.Fatalf("T7 after the lost answer: %+v", c)
		}
		if asked != 1 {
			t.Fatalf("root asked %d times, want once: the staged bytes are verified by the agent itself", asked)
		}
	})
}

// TestHeldReleaseRecoveryProbesFailClosed (fix round 1, item 2;
// platform-1e's ruling): a clean return's probe and a revert's probe never
// run configured, so nothing journals a render hash under their hold op.
// With the recovery release missing from live/<host>, each fails closed: the
// probe's release adapter gets a typed not-yet and asks root nothing, root
// asked directly refuses the row outright (fix round 2, P2-1), and nothing
// is published. A known limitation: such a probe would need to journal its own
// render hash through RenderHeldUnderBefore at an equivalent point.
func TestHeldReleaseRecoveryProbesFailClosed(t *testing.T) {
	ctx := context.Background()
	recovery := func(f *hrFixture) {
		f.claim.Op.Exec, f.claim.Op.Participants = "uk-box", []string{"uk-box", "ie-box"}
		f.claim.Op.Intent = map[string]string{"source": "uk-box", "target": "ie-box", "to": "ie"}
		f.claim.Op.Phase = hostclaim.PhaseProbing
		f.rec.Move.From, f.rec.Move.To, f.rec.Move.LandingPhase = "uk", "ie", ""
	}
	for _, tc := range []struct {
		name string
		set  func(f *hrFixture)
		hold string
	}{
		{"a clean return's probe", func(f *hrFixture) {
			recovery(f)
			f.claim.Op.Kind = hostclaim.KindRestoring
			f.claim.Op.Intent["rollback"] = hrRB
			f.rec.Move.State = provision.MoveRollingBack
			f.rec.Move.Rollback = &provision.MoveRollback{ID: hrRB, SourcePhase: provision.RbSourceProbing, RecoveryInputs: &provision.MoveInputs{Release: "v7"}}
			f.row.HoldOp = hrRB
		}, hrRB},
		{"a revert's probe", func(f *hrFixture) {
			recovery(f)
			f.claim.Op.Kind = hostclaim.KindReverting
			f.claim.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
			f.rec.Move.State = provision.MoveReverting
			f.rec.Move.RecoveryInputs = &provision.MoveInputs{Release: "v7"}
		}, p1ID},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newHRFixture(t)
			f.set(tc.set)
			// The recovery never ran configured: no render journal at all.
			f.data = t.TempDir()
			f.root.DataDir = f.data
			var seen []heldReleaseRequest
			adapter := heldReleaseProbeRelease(heldReleaseVia{Relay: f.relay(&seen), Region: "uk", Claims: f.claims}, f.s, "uk-box", f.opt, f.data, offProofGate(t))
			rt := registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", HoldOp: tc.hold}
			if err := adapter(ctx, rt, "v7", heldSelectionWork{&heldSelections{}}); !errors.Is(err, errHeldReleaseNotYet) {
				t.Fatalf("the probe's release: %v, want errHeldReleaseNotYet", err)
			}
			if len(seen) != 0 {
				t.Fatalf("root was asked %+v", seen)
			}
			// Root asked directly, as a request carrying a render hash would:
			// since fix round 2 (P2-1) root refuses these rows outright,
			// before any journal read (TestHeldReleaseRootRefusesUnjournalledRecoveryProbes
			// covers a planted journal).
			q := f.request(heldReleasePublish)
			if q.HoldOp != tc.hold {
				t.Fatalf("request hold op %q, want %q", q.HoldOp, tc.hold)
			}
			if _, err := f.root.Serve(ctx, hrPeer, q); !errors.Is(err, release.ErrStaleHeldStage) {
				t.Fatalf("root: %v, want a refusal", err)
			}
			f.untouched(t)
		})
	}
}