package claimactivation

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"errors"
	"os"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostfloor"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// backfillDeployment is a dev-shaped deployment at step 8: a published
// roster at home region ie with a verified floor, leo in pool cohort1 with
// its census key bound, sophie in pool console (with or without a census
// key), and three instance records.
type backfillDeployment struct {
	r       Runner
	keys    map[string]ed25519.PrivateKey
	records map[string]hostops.Record
}

func newBackfillDeployment(t *testing.T, sophieKey bool) *backfillDeployment {
	t.Helper()
	return newBackfillDeploymentAt(t, sophieKey, true)
}

func newBackfillDeploymentAt(t *testing.T, sophieKey, published bool) *backfillDeployment {
	t.Helper()
	ctx := t.Context()
	b := &backfillDeployment{r: runnerFixture(t), keys: map[string]ed25519.PrivateKey{}, records: map[string]hostops.Record{}}
	d := b.r.Directory
	if err := d.Initialize(ctx, b.r.Signer); err != nil {
		t.Fatal(err)
	}
	o, err := d.Read(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := hostfloor.Publish(ctx, d.Store, b.r.Signer, "bucket", hostfloor.Floor{Min: 1, MinRelease: "v1.0.0"}); err != nil {
		t.Fatal(err)
	}
	f, err := hostfloor.Read(ctx, d.Store, d.PublicKey, "bucket")
	if err != nil {
		t.Fatal(err)
	}
	next := hostactors.Clone(o.Roster)
	next.RequestLimitNS = int64(hostbox.RequestLimit)
	next.Regions = map[string]hostactors.Region{"ie": {Account: "123456789012", AWSRegion: "eu-west-1", Namespaces: []string{"platform"}, Pools: []string{"cohort1", "console"}}}
	next.HomeRegion = "ie"
	if published {
		next.Publication = "published"
		next.FloorSHA256 = f.Digest
	}
	for i, box := range []struct{ ref, pool string }{{"leo-climbing", "cohort1"}, {"sophie-knitting", "console"}} {
		row := hostactors.Box{Enrollment: strings.Repeat(string(rune('a'+i)), 32), Namespace: "platform", BoxID: box.ref, Label: box.ref, Region: "ie", Pool: box.pool,
			Cloud: hostactors.Cloud{Provider: "aws", Account: "123456789012", Region: "eu-west-1", InstanceID: "i-" + box.ref}, EvidenceSHA256: strings.Repeat("b", 64), State: "active", LossIDs: []string{}}
		if box.ref == "leo-climbing" || sophieKey {
			pub, key, _ := ed25519.GenerateKey(rand.Reader)
			row.CensusPublicKey, b.keys[box.ref] = pub, key
		}
		next.Boxes[box.ref] = row
	}
	if _, err := d.Replace(ctx, b.r.Signer, o, next); err != nil {
		t.Fatal(err)
	}
	for _, rec := range []provision.Record{
		{Schema: 1, Account: "bab", App: "claimtrial", Host: "claimtrial.example.com", Slug: "claimtrial", Channel: "edge", Pool: "cohort1", Acquisition: "acq-claimtrial"},
		{Schema: 1, Account: "bab", App: "rollcheck", Host: "rollcheck.example.com", Slug: "rollcheck", Channel: "edge", Acquisition: "acq-rollcheck"},
		{Schema: 1, Account: "bab", App: "wakecheck", Host: "wakecheck.example.com", Slug: "wakecheck", Channel: "edge", Acquisition: "acq-wakecheck"},
	} {
		b.record(t, rec)
	}
	return b
}

func (b *backfillDeployment) record(t *testing.T, rec provision.Record) {
	t.Helper()
	r, err := hostops.PrepareInstance(rec, time.Date(2026, 10, 18, 9, 0, 0, 0, time.UTC), "ie")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := b.r.Directory.Store.Put(t.Context(), r.Holder.Key, r.Body, ""); err != nil {
		t.Fatal(err)
	}
	if r, err = hostops.ReadRecord(t.Context(), b.r.Directory.Store, r.Holder.Key); err != nil {
		t.Fatal(err)
	}
	b.records[rec.Host] = r
}

func (b *backfillDeployment) revision(t *testing.T) uint64 {
	t.Helper()
	o, err := b.r.Directory.Read(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	return o.Roster.Revision
}

func (b *backfillDeployment) plan(t *testing.T, phase string) Plan {
	return Plan{Phase: phase, Bucket: "bucket", ExpectedRevision: b.revision(t), Backfill: &BackfillInput{ID: "cohort1-20261021"}}
}

// report publishes box's signed census report for the reservation, naming
// hosts as its route installations: what journal-migrate's census publishes.
func (b *backfillDeployment) report(t *testing.T, box string, hosts ...string) {
	t.Helper()
	o, err := b.r.Directory.Read(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	report := hostops.CensusReport{Box: box, Enrollment: o.Roster.Boxes[box].Enrollment, BootID: "boot", Reservation: "cohort1-20261021", DirectoryRevision: o.Roster.Revision,
		CapturedAt: time.Now().UTC(), Complete: true, Installations: []hostops.CensusInstallation{}, Seeds: &hostops.CensusSeeds{Journaled: []string{}, Gated: []string{}}}
	for _, host := range hosts {
		r := b.records[host]
		inc, err := r.Incarnation()
		if err != nil {
			t.Fatal(err)
		}
		report.Installations = append(report.Installations, hostops.CensusInstallation{Host: host, Account: r.Holder.Account, App: r.Holder.App, Family: r.Holder.Family, Key: r.Holder.Key, Incarnation: inc, Route: true})
	}
	if err := hostops.PublishCensusReport(t.Context(), b.r.Directory.Store, b.r.Directory, report, b.keys[box]); err != nil {
		t.Fatal(err)
	}
}

func readPlan(t *testing.T, r Runner) hostops.BackfillPlan {
	t.Helper()
	p, _, err := readSignedPlan(t.Context(), r.Directory)
	if err != nil {
		t.Fatal(err)
	}
	return p
}

// backfill-collect freezes every record the bucket lists, create-only, and
// prints the plan on a dry run without writing it. backfill-adopt, once each
// roster box has published its census, advances it under If-Match, takes
// the boxes from the census, and leaves unreported hosts without one.
func TestBackfillPhases(t *testing.T) {
	ctx := t.Context()
	b := newBackfillDeployment(t, true)
	dry, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), true)
	if err != nil || len(dry.Writes) != 1 || dry.Writes[0].Key != hostops.BackfillKey || dry.Writes[0].BeforeETag != "" || dry.Backfill == nil || len(dry.Backfill.Hosts) != 3 {
		t.Fatalf("collect dry run = %+v, %v", dry, err)
	}
	if _, _, err := b.r.Directory.Store.Get(ctx, hostops.BackfillKey); !errors.Is(err, os.ErrNotExist) {
		t.Fatal("dry run wrote the plan", err)
	}
	got, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false)
	if err != nil || got.Backfill.Phase != "collecting" || len(got.Backfill.Hosts) != 3 {
		t.Fatalf("collect = %+v, %v", got, err)
	}
	p := readPlan(t, b.r)
	for host, r := range b.records {
		h := p.Hosts[host]
		if h.Key != r.Holder.Key || h.ETag != r.ETag || h.Acquisition != r.Holder.Acquisition || h.Box != "" || h.Region != "" {
			t.Fatalf("frozen %s = %+v", host, h)
		}
	}
	if p.Phase != "collecting" || p.DirectoryRevision != b.revision(t) || p.ID != "cohort1-20261021" {
		t.Fatalf("plan = %+v", p)
	}
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); !errors.Is(err, ErrBackfillExists) {
		t.Fatalf("second collect = %v", err)
	}
	// Adopt waits for every roster box's census report.
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); err == nil {
		t.Fatal("adopted without census reports")
	}
	b.report(t, "leo-climbing", "claimtrial.example.com")
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); err == nil {
		t.Fatal("adopted without sophie's report")
	}
	b.report(t, "sophie-knitting", "wakecheck.example.com")
	wrong := b.plan(t, "backfill-adopt")
	wrong.Backfill.ID = "other"
	if _, err := b.r.Run(ctx, wrong, false); !errors.Is(err, ErrBackfillPlan) {
		t.Fatalf("other reservation = %v", err)
	}
	dry, err = b.r.Run(ctx, b.plan(t, "backfill-adopt"), true)
	if err != nil || len(dry.Writes) != 1 || dry.Backfill.Phase != "adopting" {
		t.Fatalf("adopt dry run = %+v, %v", dry, err)
	}
	if readPlan(t, b.r).Phase != "collecting" {
		t.Fatal("dry run advanced the plan")
	}
	got, err = b.r.Run(ctx, b.plan(t, "backfill-adopt"), false)
	if err != nil || got.Backfill.Phase != "adopting" || len(got.Backfill.Reports) != 2 || len(got.Backfill.Unowned) != 1 || got.Backfill.Unowned[0] != "rollcheck.example.com" {
		t.Fatalf("adopt = %+v, %v", got.Backfill, err)
	}
	if got.Writes[0].BeforeETag == "" {
		t.Fatal("adopt wrote without If-Match")
	}
	p = readPlan(t, b.r)
	if h := p.Hosts["claimtrial.example.com"]; h.Box != "leo-climbing" || h.Region != "ie" {
		t.Fatalf("claimtrial = %+v", h)
	}
	if h := p.Hosts["wakecheck.example.com"]; h.Box != "sophie-knitting" {
		t.Fatalf("wakecheck = %+v", h)
	}
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); !errors.Is(err, ErrBackfillPlan) {
		t.Fatalf("second adopt = %v", err)
	}
	// done is only ever from collecting.
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillPlan) {
		t.Fatalf("done over an adopting plan = %v", err)
	}

	// backfill-done, the empty cohort (claim trial Option B): records are
	// deleted before the window (all of them, as the trial does; or only
	// leo's, the two-host shape), sophie has no census key, leo reports
	// nothing. done writes done under If-Match and lifts the create freeze;
	// a dry run writes nothing; a second done is refused.
	for name, keep := range map[string][]string{"zero hosts": nil, "two hosts": {"rollcheck.example.com", "wakecheck.example.com"}} {
		t.Run(name, func(t *testing.T) {
			ctx := t.Context()
			e := newBackfillDeployment(t, false)
			e.deleteRecordsExcept(t, keep...)
			dry, err := e.r.Run(ctx, e.plan(t, "backfill-collect"), true)
			if err != nil || len(dry.Writes) != 1 || len(dry.Backfill.Hosts) != len(keep) {
				t.Fatalf("collect dry run = %+v, %v", dry, err)
			}
			if len(keep) == 0 && !bytes.Contains(dry.Writes[0].Payload, []byte(`"hosts":{}`)) {
				t.Fatalf("zero-host plan payload = %s; want \"hosts\":{}", dry.Writes[0].Payload)
			}
			if _, err := e.r.Run(ctx, e.plan(t, "backfill-collect"), false); err != nil {
				t.Fatal(err)
			}
			collecting := readPlan(t, e.r)
			if collecting.Hosts == nil || len(collecting.Hosts) != len(keep) {
				t.Fatalf("collecting hosts = %#v", collecting.Hosts)
			}
			_, collectingTag, err := e.r.Directory.Store.Get(ctx, hostops.BackfillKey)
			if err != nil {
				t.Fatal(err)
			}
			// Leo has not reported yet.
			if _, err := e.r.Run(ctx, e.plan(t, "backfill-done"), false); !errors.Is(err, os.ErrNotExist) {
				t.Fatalf("done before leo's report = %v", err)
			}
			e.report(t, "leo-climbing")
			wrong := e.plan(t, "backfill-done")
			wrong.Backfill.ID = "other"
			if _, err := e.r.Run(ctx, wrong, false); !errors.Is(err, ErrBackfillPlan) {
				t.Fatalf("done for another reservation = %v", err)
			}
			dry, err = e.r.Run(ctx, e.plan(t, "backfill-done"), true)
			if err != nil || len(dry.Writes) != 1 || dry.Writes[0].Key != hostops.BackfillKey || dry.Writes[0].BeforeETag != collectingTag || dry.Backfill.Phase != "done" {
				t.Fatalf("done dry run = %+v, %v", dry, err)
			}
			if readPlan(t, e.r).Phase != "collecting" {
				t.Fatal("dry run moved the plan")
			}
			if err := hostops.CheckCreateFreeze(ctx, e.r.Directory.Store, e.r.Directory.PublicKey); !errors.Is(err, hostops.ErrPending) {
				t.Fatalf("freeze after a dry run = %v", err)
			}
			got, err := e.r.Run(ctx, e.plan(t, "backfill-done"), false)
			if err != nil || got.Backfill.Phase != "done" || len(got.Backfill.Hosts) != len(keep) || len(got.Backfill.Reports) != 1 || got.Backfill.Reports["leo-climbing"] == "" ||
				len(got.Backfill.Unreported) != 1 || got.Backfill.Unreported[0] != "sophie-knitting" || len(got.Backfill.Unowned) != 0 {
				t.Fatalf("done = %+v, %v", got.Backfill, err)
			}
			if len(got.Writes) != 1 || got.Writes[0].BeforeETag != collectingTag {
				t.Fatalf("done wrote %+v; want one If-Match write over the collecting plan", got.Writes)
			}
			done := readPlan(t, e.r)
			if done.Phase != "done" || done.Hosts == nil || len(done.Hosts) != len(keep) {
				t.Fatalf("done plan = %#v", done)
			}
			if err := hostops.CheckCreateFreeze(ctx, e.r.Directory.Store, e.r.Directory.PublicKey); err != nil {
				t.Fatalf("freeze after done = %v", err)
			}
			if _, err := e.r.Run(ctx, e.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillPlan) {
				t.Fatalf("second done = %v", err)
			}
		})
	}
}

func (b *backfillDeployment) deleteRecordsExcept(t *testing.T, keep ...string) {
	t.Helper()
	for host, r := range b.records {
		if slices.Contains(keep, host) {
			continue
		}
		if err := release.DeleteIfMatch(t.Context(), b.r.Directory.Store, r.Holder.Key, r.ETag); err != nil {
			t.Fatal(err)
		}
		delete(b.records, host)
	}
}

// backfill-done is fail-closed on installations, with no override: leo's
// report still lists claimtrial (B1 skipped), so done refuses with the
// empty-cohort refusal and the plan stays collecting. A record that changed
// since collect refuses with the records refusal.
func TestBackfillDoneRefusesAPopulatedCohort(t *testing.T) {
	ctx := t.Context()
	b := newBackfillDeployment(t, false)
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); err != nil {
		t.Fatal(err)
	}
	b.report(t, "leo-climbing", "claimtrial.example.com")
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), true); !errors.Is(err, ErrBackfillInstallations) || !errors.Is(err, hostops.ErrDoneInstallations) {
		t.Fatalf("done dry run over an installation = %v", err)
	}
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillInstallations) {
		t.Fatalf("done over an installation = %v", err)
	}
	if readPlan(t, b.r).Phase != "collecting" {
		t.Fatal("plan moved")
	}

	m := newBackfillDeployment(t, false)
	if _, err := m.r.Run(ctx, m.plan(t, "backfill-collect"), false); err != nil {
		t.Fatal(err)
	}
	m.report(t, "leo-climbing")
	r := m.records["wakecheck.example.com"]
	if _, err := m.r.Directory.Store.Put(ctx, r.Holder.Key, append(append([]byte(nil), r.Body...), '\n'), r.ETag); err != nil {
		t.Fatal(err)
	}
	if _, err := m.r.Run(ctx, m.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillRecords) {
		t.Fatalf("done over a changed record = %v", err)
	}
	if readPlan(t, m.r).Phase != "collecting" {
		t.Fatal("plan moved")
	}
}

// The dev roster at step 8: sophie is a roster box with no census key, so
// no census report of hers can verify, and backfill-adopt refuses. The plan
// stays collecting.
func TestBackfillAdoptRefusesABoxWithoutCensusKey(t *testing.T) {
	ctx := t.Context()
	b := newBackfillDeployment(t, false)
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); err != nil {
		t.Fatal(err)
	}
	b.report(t, "leo-climbing", "claimtrial.example.com")
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); err == nil {
		t.Fatal("adopted with a roster box that cannot report")
	}
	if readPlan(t, b.r).Phase != "collecting" {
		t.Fatal("plan moved")
	}
}

// Refusals up front: a roster that is not published, or has no home region
// or no verified floor; a reservation id that is not one safe segment; a
// record with no acquisition.
func TestBackfillCollectRefusesUpFront(t *testing.T) {
	ctx := t.Context()
	if _, err := newBackfillDeploymentAt(t, true, false).r.Run(ctx, Plan{Phase: "backfill-collect", Bucket: "bucket", ExpectedRevision: 2, Backfill: &BackfillInput{ID: "cohort1-20261021"}}, true); !errors.Is(err, ErrBackfillState) {
		t.Fatalf("collect on an unpublished roster = %v", err)
	}
	gone := newBackfillDeployment(t, true)
	for _, key := range []string{hostfloor.Key, hostfloor.AuthKey} {
		if err := gone.r.Directory.Store.Delete(ctx, key); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := gone.r.Run(ctx, gone.plan(t, "backfill-collect"), true); !errors.Is(err, ErrBackfillState) {
		t.Fatalf("collect with no verified floor = %v", err)
	}
	b := newBackfillDeployment(t, true)
	for _, id := range []string{"", "a/b", "..", "UPPER", strings.Repeat("a", 65)} {
		p := b.plan(t, "backfill-collect")
		p.Backfill.ID = id
		if _, err := b.r.Run(ctx, p, true); !errors.Is(err, ErrActivation) {
			t.Fatalf("id %q = %v", id, err)
		}
	}
	p := b.plan(t, "backfill-collect")
	p.Backfill = nil
	if _, err := b.r.Run(ctx, p, true); !errors.Is(err, ErrActivation) {
		t.Fatalf("no backfill input = %v", err)
	}
	b.record(t, provision.Record{Schema: 1, Account: "bab", App: "legacy", Host: "legacy.example.com", Slug: "legacy", Channel: "edge"})
	if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), true); !errors.Is(err, ErrBackfillRecords) {
		t.Fatalf("record without acquisition = %v", err)
	}
}

// A record that moves while the plan is being written leaves a plan the box
// census would refuse; collect reports it rather than succeeding.
type movingRecord struct {
	release.Store
	key  string
	body []byte
	done bool
}

func (s *movingRecord) Put(ctx context.Context, key string, raw []byte, before string) (string, error) {
	tag, err := s.Store.Put(ctx, key, raw, before)
	if err == nil && key == hostops.BackfillKey && !s.done {
		s.done = true
		_, old, _ := s.Store.Get(ctx, s.key)
		if _, e := s.Store.Put(ctx, s.key, s.body, old); e != nil {
			return "", e
		}
	}
	return tag, err
}

func TestBackfillCollectRefusesARecordThatMoved(t *testing.T) {
	b := newBackfillDeployment(t, true)
	r := b.records["claimtrial.example.com"]
	body := append([]byte(nil), r.Body...)
	b.r.Directory.Store = &movingRecord{Store: b.r.Directory.Store, key: r.Holder.Key, body: append(body, '\n')}
	if _, err := b.r.Run(t.Context(), b.plan(t, "backfill-collect"), false); !errors.Is(err, ErrBackfillRecords) {
		t.Fatalf("collect over a moving record = %v", err)
	}
}

// windowStore runs afterGet once on the first Get of getKey and afterPut
// once after the first Put of putKey: a console record writer racing the
// claims run's backfill-done.
type windowStore struct {
	release.Store
	getKey, putKey     string
	afterGet, afterPut func()
}

func (s *windowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	b, tag, err := s.Store.Get(ctx, key)
	if err == nil && key == s.getKey && s.afterGet != nil {
		f := s.afterGet
		s.afterGet = nil
		f()
	}
	return b, tag, err
}

func (s *windowStore) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
	got, err := s.Store.Put(ctx, key, b, tag)
	if err == nil && key == s.putKey && s.afterPut != nil {
		f := s.afterPut
		s.afterPut = nil
		f()
	}
	return got, err
}

// A record created during the window reaches the operator as the records
// refusal: while leo's report is read, done refuses before writing and the
// plan stays collecting; after done's write, the refusal carries
// ErrDoneRecordsChanged and the plan is done.
func TestBackfillDoneRecordCreatedDuringTheWindow(t *testing.T) {
	for _, after := range []bool{false, true} {
		name := map[bool]string{false: "during report retrieval", true: "after the write"}[after]
		t.Run(name, func(t *testing.T) {
			ctx := t.Context()
			b := newBackfillDeployment(t, false)
			if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); err != nil {
				t.Fatal(err)
			}
			b.report(t, "leo-climbing")
			fired := false
			create := func() {
				fired = true
				b.record(t, provision.Record{Schema: 1, Account: "bab", App: "late", Host: "late.example.com", Slug: "late", Channel: "edge", Acquisition: "acq-late"})
			}
			s := &windowStore{Store: b.r.Directory.Store}
			if after {
				s.putKey, s.afterPut = hostops.BackfillKey, create
			} else {
				s.getKey, s.afterGet = "control/host-census/cohort1-20261021/leo-climbing.json", create
			}
			b.r.Directory.Store = s
			_, err := b.r.Run(ctx, b.plan(t, "backfill-done"), false)
			if !fired {
				t.Fatal("the window's hook never ran")
			}
			if !errors.Is(err, ErrBackfillRecords) || errors.Is(err, hostops.ErrDoneRecordsChanged) != after {
				t.Fatalf("done = %v", err)
			}
			if want := map[bool]string{false: "collecting", true: "done"}[after]; readPlan(t, b.r).Phase != want {
				t.Fatalf("plan phase = %s; want %s", readPlan(t, b.r).Phase, want)
			}
		})
	}
}