//go:build unix

package hostbox

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/superfly/ltx"
)

// restorePlan is a restore plan whose one selected file is the snapshot
// [1, max] of generation gen ("" is the location's own files) under
// location: what Select hands a loss admission.
func restorePlan(location, gen string, max uint64) *genselect.Plan {
	base := strings.TrimSuffix(location, "/") + "/"
	if gen != "" && gen != hostclaim.RootGeneration {
		base += "g-" + gen + "/"
	}
	f := genselect.File{Key: base + "0009/" + ltx.FormatFilename(1, ltx.TXID(max)), VersionID: "v1", Level: 9, MinTXID: 1, MaxTXID: max, Size: 100}
	g := gen
	if g == "" {
		g = hostclaim.RootGeneration
	}
	return &genselect.Plan{Generation: g, Files: []genselect.File{f}, Root: &f, MaxTXID: max}
}

// moves Task 4 (platform-1e's ruling on #236): a loss admission's restore
// point is consistent with its own plan: the admission's RemoteMaxTXID is
// the plan's maximum, which genselect.GenerationMax reads from the plan's
// selected files of its own generation, never a fresh listing. A plan that
// carries only a maximum, or whose files do not reach it, is refused before
// the loss sources run and before the claim CAS; no bucket is read, so a
// writer with none still admits a consistent plan.
func TestLossRestorePointIsTheGenerationMax(t *testing.T) {
	for _, path := range []string{"declared", "fenced"} {
		for _, mode := range []string{"only a maximum", "files short of it", "another input", "consistent, no bucket"} {
			t.Run(path+"/"+mode, func(t *testing.T) {
				_, store, o := workFixture(t)
				o, err := hostclaim.Update(t.Context(), store, o, func(c *hostclaim.Claim) error {
					c.Op = hostclaim.Op{Kind: hostclaim.KindRestoring, ID: "op", Phase: hostclaim.PhaseProbing, Exec: "box", Participants: []string{"box"}, Intent: map[string]string{"target": "target"}}
					return nil
				})
				if err != nil {
					t.Fatal(err)
				}
				w := Writer{Claims: store, Box: "box", Region: "ie"}
				in := hostclaim.Admission{Box: "box", Mode: "cold", Region: "ie", RemoteMaxTXID: 3}
				baseline := restorePlan(o.Claim.Holder.Location, "src", 3)
				switch mode {
				case "only a maximum":
					baseline = &genselect.Plan{Generation: "src", MaxTXID: 3}
				case "files short of it":
					baseline = restorePlan(o.Claim.Holder.Location, "src", 2)
					baseline.MaxTXID = 3
				case "another input":
					in.RemoteMaxTXID = 4
				}
				err = w.restorePoint(t.Context(), o, in, baseline)
				if mode == "consistent, no bucket" {
					if err != nil {
						t.Fatalf("a consistent plan was refused: %v", err)
					}
					return
				}
				if !errors.Is(err, ErrChanged) {
					t.Fatalf("refused with %v", err)
				}
				sources := 0
				w.Loss = &LossAdmission{PublicKey: make([]byte, 32), Owner: &plan12LossOwner{}, Sources: func(context.Context, hostclaim.Observed, hostclaim.LossRecord, hostclaim.Admission, *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) {
					sources++
					return nil, nil
				}}
				if path == "declared" {
					_, err = w.admitLoss(t.Context(), o, in, baseline, "loss_1")
				} else {
					_, err = w.admitFencedLoss(t.Context(), o, in, baseline, hostclaim.LossRecord{})
				}
				if !errors.Is(err, ErrChanged) || sources != 0 {
					t.Fatalf("%s: %v, the loss sources ran %d times", path, err, sources)
				}
			})
		}
	}
}

// Fix round 1, item 2: the restore point honours the authorized pin, as
// Select does. An ordinary selection whose only usable root is the marker's
// pinned snapshot, surviving only as a historical version (its current
// version a retention tombstone), is accepted at the pin's maximum.
func TestLossRestorePointHonoursThePinnedRoot(t *testing.T) {
	_, store, o := workFixture(t)
	o, err := hostclaim.Update(t.Context(), store, o, func(c *hostclaim.Claim) error {
		c.Holder.Generation = "src"
		c.Op = hostclaim.Op{Kind: hostclaim.KindRestoring, ID: "op", Phase: hostclaim.PhaseProbing, Exec: "box", Participants: []string{"box"}, Intent: map[string]string{"target": "target"}}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	h := o.Claim.Holder
	data := leasetest.NewVersionedBucket()
	data.AddBucket("data")
	header, err := (&ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: 1, MaxTXID: 5}).MarshalBinary()
	if err != nil {
		t.Fatal(err)
	}
	root := strings.TrimSuffix(h.Location, "/") + "/g-src/0009/" + ltx.FormatFilename(1, 5)
	version := data.Put("data", root, header)
	marker, err := json.Marshal(genselect.Marker{Op: "src", Incarnation: h.Incarnation, Host: o.Host, Box: "box", InstalledAt: time.Unix(1, 0).UTC(),
		Pin: genselect.Pin{SnapshotKey: root, SnapshotVersion: version, SnapshotTXID: 5}})
	if err != nil {
		t.Fatal(err)
	}
	data.Put("data", strings.TrimSuffix(h.Location, "/")+"/generation.json", marker)
	data.BareDelete("data", root)
	w := Writer{Claims: store, Box: "box", Region: "ie", Bucket: "data", Data: data}
	plan, err := w.Baseline(t.Context(), o)
	if err != nil {
		t.Fatal(err)
	}
	if plan.Root == nil || plan.Root.VersionID != version || plan.MaxTXID != 5 {
		t.Fatalf("the selection is not the pinned historical root: %+v", plan)
	}
	if err := w.restorePoint(t.Context(), o, hostclaim.Admission{RemoteMaxTXID: plan.MaxTXID}, plan); err != nil {
		t.Fatalf("a valid pinned restore point was refused: %v", err)
	}
}