package console

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rolloutwords"
	"github.com/carlosframework/platform/internal/status"
)

// writeFleetHeldBox registers fleet to owner and writes, under the fleet's
// own prefix, a box whose report holds host back on build for this
// fixture's account and app. It returns the box's label.
func writeFleetHeldBox(t *testing.T, fx *consoleFixture, fleet, owner, host, app, build string) string {
	t.Helper()
	reg, err := registry.Open(fx.regPath)
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	if err := reg.UpsertFleet(registry.Fleet{Name: fleet, Account: owner, Bucket: fleet + "-bucket"}); err != nil {
		t.Fatal(err)
	}
	box := status.CollectBox(nil, status.Probes{MachineID: fleet + "-box"}, time.Now())
	box.Metrics = &status.BoxMetrics{Rollout: &status.RolloutReport{Hosts: map[string]status.HostRollout{
		host: {Account: fx.sqid, App: app, Installed: build, Serving: "v1", Good: "v1", Awake: true, HeldBy: build, Reason: status.ReasonHealth,
			HeldEpoch: heldEpoch, HeldBook: heldBook, RetryV: status.HostRetryV},
	}}}
	if err := status.WriteBox(context.Background(), release.NewPrefixStore(fx.rel, "fleets/"+fleet+"/"), box); err != nil {
		t.Fatal(err)
	}
	return box.Label
}

// listRollout is the list's answer for host: its rollout view, and whether
// the answer said rollout_incomplete.
func listRollout(t *testing.T, fx *consoleFixture, token, app, host string) (*rolloutView, bool) {
	t.Helper()
	code, _, out := rawReq(t, http.MethodGet, fx.ts.URL+"/api/cli/apps/"+fx.sqid+"/"+app+"/instances", token, "")
	if code != http.StatusOK {
		t.Fatalf("list = %d %s", code, out)
	}
	var list struct {
		Instances []struct {
			Host    string       `json:"host"`
			Rollout *rolloutView `json:"rollout"`
		} `json:"instances"`
		RolloutIncomplete bool `json:"rollout_incomplete"`
	}
	if err := json.Unmarshal([]byte(out), &list); err != nil {
		t.Fatal(err)
	}
	for _, r := range list.Instances {
		if r.Host == host {
			return r.Rollout, list.RolloutIncomplete
		}
	}
	t.Fatalf("list lacks %s: %s", host, out)
	return nil, false
}

// placedRecord writes app's instance record for host, placed at region
// ("ie", or "fleet/<name>" for a customer fleet).
func placedRecord(t *testing.T, fx *consoleFixture, app, host, region string) {
	t.Helper()
	if err := provision.CreateRecord(context.Background(), fx.rel, provision.Record{
		Schema: 1, Host: host, Slug: strings.SplitN(host, ".", 2)[0],
		App: app, Account: release.OpsAccount, Region: region, Channel: "stable",
		CreatedAt: time.Now().UTC(), CreatedBy: "paul@keymail.dev",
	}); err != nil {
		t.Fatal(err)
	}
}

// Astra plan 6 R1 #6: a customer-fleet box's report counts. Its held host
// shows held on the list and on the instance page, and the retry door takes
// the retry, through the one loader all three share. The record is placed on
// the fleet, so an unreadable snapshot there makes the list incomplete.
func TestFleetBoxHoldShowsOnTheListPageAndRetryDoor(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.addRoute(t, registry.Route{Host: heldHost, App: "kass", Account: fx.sqid, Kind: "instance", Status: "ready", Version: "v2", Channel: "stable"})
	placedRecord(t, fx, "kass", heldHost, registry.PlacementTarget("pi"))
	label := writeFleetHeldBox(t, fx, "pi", fx.sqid, heldHost, "kass", "v2")
	token := cliToken(t, fx, "paul@keymail.dev", "code-fleet-held")

	ro, incomplete := listRollout(t, fx, token, "kass", heldHost)
	if ro == nil || ro.Box != label || ro.HeldBy != "v2" || incomplete {
		t.Fatalf("list rollout = %+v, incomplete %v: want the fleet box's hold, complete", ro, incomplete)
	}
	c := signInPaul(t, fx)
	if code, _, body := get(t, c, fx.u("/apps/kass/instances/"+heldHost)); code != http.StatusOK ||
		!strings.Contains(body, fmt.Sprintf(swapCopy.HoldText, "v2", rolloutwords.Reason(status.ReasonHealth), "v1")) {
		t.Fatalf("page = %d, lacks the fleet box's hold", code)
	}
	code, _, body := rawReq(t, http.MethodPost, fx.ts.URL+"/api/cli/apps/"+fx.sqid+"/kass/instances/"+heldHost+"/retry", token, `{}`)
	if code != http.StatusOK || !strings.Contains(body, `"version":"v2"`) {
		t.Fatalf("retry = %d %s, want 200 for v2", code, body)
	}

	// A snapshot that cannot be read on the fleet the record is placed on
	// makes the join incomplete.
	if _, err := fx.rel.Put(context.Background(), "fleets/pi/status/boxes/broken.json", []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	if _, incomplete := listRollout(t, fx, token, "kass", heldHost); !incomplete {
		t.Fatal("an unreadable fleet box snapshot left the join complete")
	}
}

// Astra plan 6 R1 #6: only this account's fleets are read. Another
// account's fleet box that names this account's host is never joined,
// whatever account its report claims: no hold on the list or the page, and
// the retry door refuses.
func TestAnotherAccountsFleetBoxIsNeverJoined(t *testing.T) {
	fx := newConsoleFixture(t)
	fx.addRoute(t, registry.Route{Host: heldHost, App: "kass", Account: fx.sqid, Kind: "instance", Status: "ready", Version: "v2", Channel: "stable"})
	sibling(t, fx, "kass", heldHost, "stable")
	writeFleetHeldBox(t, fx, "stranger", "zzz", heldHost, "kass", "v2")
	token := cliToken(t, fx, "paul@keymail.dev", "code-fleet-stranger")

	if ro, _ := listRollout(t, fx, token, "kass", heldHost); ro != nil {
		t.Fatalf("list joined another account's fleet box: %+v", ro)
	}
	c := signInPaul(t, fx)
	if _, _, body := get(t, c, fx.u("/apps/kass/instances/"+heldHost)); strings.Contains(body, swapCopy.Installed) {
		t.Fatal("page joined another account's fleet box")
	}
	code, _, body := rawReq(t, http.MethodPost, fx.ts.URL+"/api/cli/apps/"+fx.sqid+"/kass/instances/"+heldHost+"/retry", token, `{}`)
	if code != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(apiRetryNotHeld, heldHost)) {
		t.Fatalf("retry = %d %s, want 409", code, body)
	}
}

// A platform box's report wins over a fleet box's for the same host: the
// operator's boxes are the ones the console trusts most.
func TestJoinFleetRolloutsNeverOverrideAPlatformBox(t *testing.T) {
	now := time.Now()
	view := func(label, build string) status.Box {
		return status.Box{GeneratedAt: now, Label: label, Metrics: &status.BoxMetrics{Rollout: &status.RolloutReport{
			Hosts: map[string]status.HostRollout{"h": {Account: "bab", App: "shop", Installed: build}},
		}}}
	}
	j := joinBoxes([]status.Box{view("platform", "v1")}, now)
	j.joinFleetRollouts([]status.Box{view("fleet", "v9")}, now)
	if got := j.rollout("h", "bab", "shop"); got == nil || got.Box != "platform" || got.Installed != "v1" {
		t.Fatalf("h = %+v, want the platform box's report", got)
	}
	var empty boxJoin
	empty.joinFleetRollouts([]status.Box{view("fleet", "v9")}, now)
	if got := empty.rollout("h", "bab", "shop"); got == nil || got.Box != "fleet" {
		t.Fatalf("h on an empty join = %+v, want the fleet box's report", got)
	}
}

// Astra plan 6 R2 (new Medium): completeness covers only the sources that
// could hold this app's reports. The account has a fleet whose snapshot is
// corrupt, but none of the app's instances is placed there; the app runs on
// healthy platform boxes with no reports (flag off). The list must not say
// rollout_incomplete, so the deploy watch falls back to today's URL watch.
func TestAnUnrelatedUnreadableFleetDoesNotMakeTheListIncomplete(t *testing.T) {
	fx := newConsoleFixture(t)
	placedRecord(t, fx, "kass", heldHost, "ie")
	reg, err := registry.Open(fx.regPath)
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	if err := reg.UpsertFleet(registry.Fleet{Name: "spare", Account: fx.sqid, Bucket: "spare-bucket"}); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), "fleets/spare/status/boxes/broken.json", []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	box := status.CollectBox(nil, status.Probes{MachineID: "box1"}, time.Now())
	box.Metrics = &status.BoxMetrics{RouteVersions: map[string]string{heldHost: "v1"}}
	if err := status.WriteBox(context.Background(), fx.rel, box); err != nil {
		t.Fatal(err)
	}
	token := cliToken(t, fx, "paul@keymail.dev", "code-fleet-unrelated")
	ro, incomplete := listRollout(t, fx, token, "kass", heldHost)
	if ro != nil || incomplete {
		t.Fatalf("rollout %+v, incomplete %v: an unrelated fleet's unreadable snapshot blocked the legacy watch", ro, incomplete)
	}
}

// Astra plan 6 R3: an unreadable record must never drop its fleet from the
// completeness check. The record's host sits on fleet/pi, whose only
// snapshot is unreadable too. The readable record is on the platform, so
// without the fix no fleet is relevant and the list would claim to be
// complete, sending the deploy watch to poll URLs it must not poll.
func TestAnUnreadableRecordKeepsEveryFleetInTheCompletenessCheck(t *testing.T) {
	fx := newConsoleFixture(t)
	const piHost = "pi-host.example.test"
	placedRecord(t, fx, "kass", heldHost, "ie")
	placedRecord(t, fx, "kass", piHost, registry.PlacementTarget("pi"))
	reg, err := registry.Open(fx.regPath)
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	if err := reg.UpsertFleet(registry.Fleet{Name: "pi", Account: fx.sqid, Bucket: "pi-bucket"}); err != nil {
		t.Fatal(err)
	}
	box := status.CollectBox(nil, status.Probes{MachineID: "box1"}, time.Now())
	box.Metrics = &status.BoxMetrics{RouteVersions: map[string]string{heldHost: "v1"}}
	if err := status.WriteBox(context.Background(), fx.rel, box); err != nil {
		t.Fatal(err)
	}
	token := cliToken(t, fx, "paul@keymail.dev", "code-unreadable-record")

	// Both readable and complete first: the platform record's fleet set is
	// empty, but the pi record is placed on pi, whose snapshot is fine.
	if _, incomplete := listRollout(t, fx, token, "kass", heldHost); incomplete {
		t.Fatal("a healthy app was reported incomplete before anything was broken")
	}

	// Now the pi record and the pi snapshot both go unreadable.
	key := provision.RecordKey(release.OpsAccount, "kass", piHost)
	_, etag, err := fx.rel.Get(context.Background(), key)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), key, []byte("{not json"), etag); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), "fleets/pi/status/boxes/broken.json", []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	if _, incomplete := listRollout(t, fx, token, "kass", heldHost); !incomplete {
		t.Fatal("an unreadable record dropped its fleet from the completeness check")
	}
}

// The fleets that could hold an app's reports: where its records are placed,
// and both ends of a move in flight. A platform region names none.
func TestRecordFleetsNamesPlacementsAndMoveEnds(t *testing.T) {
	got := recordFleets([]provision.Record{
		{Host: "a", Region: "ie"},
		{Host: "b", Region: registry.PlacementTarget("pi")},
		{Host: "c", Region: "ie", Move: &provision.MoveState{From: "ie", To: registry.PlacementTarget("zero")}},
	})
	if len(got) != 2 || !got["pi"] || !got["zero"] {
		t.Fatalf("fleets = %v, want pi and zero", got)
	}
	if got := recordFleets(nil); got == nil || len(got) != 0 {
		t.Fatalf("no records = %v, want an empty, non-nil set", got)
	}
}