package host

import (
	"bytes"
	"context"
	"crypto"
	"crypto/rand"
	"crypto/rsa"
	"crypto/sha256"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/base64"
	"encoding/json"
	"errors"
	"math/big"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/boxid"
	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// carlos system box publish-record, through the production wiring: the
// real S3Store (openDeployStoreEnv over CARLOS_DEPLOYMENT_BUCKET and
// CARLOS_S3_ENDPOINT) against leasetest's S3 server, which enforces
// If-None-Match: * itself, behind the dev bucket's box_records policy;
// the real IMDS reader over fakeIMDS (boxrecord_test.go); and enrollment's
// own verifier over documents a test AWS signs.

const (
	publishMachineID = "0123456789abcdef0123456789abcdef"
	publishBucket    = "deploy"
	publishAgentUID  = 999
	publishAccount   = "111122223333"
	publishRegion    = "eu-west-1"
	publishInstance  = "i-0abc"
)

// testAWS stands in for AWS's instance-identity signer the way boxid's
// own tests do (boxid/internal/signtest, which this package cannot
// import): one RSA key, a self-signed certificate for it, and PKCS #1
// v1.5 SHA-256 signatures over identity documents.
type testAWS struct {
	key  *rsa.PrivateKey
	cert *x509.Certificate
}

func newTestAWS(t testing.TB) *testAWS {
	t.Helper()
	key, err := rsa.GenerateKey(rand.Reader, 2048)
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "test"},
		NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour)}
	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	if err != nil {
		t.Fatal(err)
	}
	cert, err := x509.ParseCertificate(der)
	if err != nil {
		t.Fatal(err)
	}
	return &testAWS{key: key, cert: cert}
}

// identityDoc is an identity document as IMDS serves one, with the two
// fields that change over an instance's life.
func identityDoc(account, region, instance, pendingTime, instanceType string) string {
	return "{\n  \"accountId\" : \"" + account + "\",\n  \"instanceId\" : \"" + instance + "\",\n  \"instanceType\" : \"" + instanceType +
		"\",\n  \"pendingTime\" : \"" + pendingTime + "\",\n  \"region\" : \"" + region + "\"\n}\n"
}

func (a *testAWS) sign(t testing.TB, doc string) string {
	t.Helper()
	sum := sha256.Sum256([]byte(doc))
	sig, err := rsa.SignPKCS1v15(rand.Reader, a.key, crypto.SHA256, sum[:])
	if err != nil {
		t.Fatal(err)
	}
	return base64.StdEncoding.EncodeToString(sig)
}

// certs certifies eu-west-1, this box's region, and eu-west-2, so a record
// signed in another supported region verifies and is still refused.
func (a *testAWS) certs() (map[string]*x509.Certificate, error) {
	return map[string]*x509.Certificate{"eu-west-1": a.cert, "eu-west-2": a.cert}, nil
}

func (a *testAWS) record(t testing.TB, box, doc, writtenAt string) boxid.Record {
	return boxid.Record{V: 1, Box: box, IdentityDocument: base64.StdEncoding.EncodeToString([]byte(doc)),
		IdentitySignature: a.sign(t, doc), WrittenAt: writtenAt}
}

// s3Request is one request the fake S3 saw.
type s3Request struct{ Method, Key, IfNoneMatch, IfMatch string }

// boxRecordsBucket is the fake deployment bucket: leasetest's versioned S3
// server, behind the dev bucket's deny-only box_records policy. Under
// control/boxes/ a PUT without If-None-Match: * is AccessDenied
// (BoxRecordsAreCreateOnly) and so is any DELETE
// (BoxRecordsAreNeverDeleted). Every request is recorded. aws and doc
// are the box's AWS and the identity document IMDS serves it now.
//
// Hooks, each used once: afterGet runs after the first GET has been
// answered (a writer arriving between the verb's read and
// publishBoxRecord's); beforePut runs before the first PUT reaches the
// server (a writer arriving just before the create). commitThenFail lists
// statuses for successive PUTs: each such PUT is applied and then answered
// with that status instead, as when S3 commits a write whose answer is
// lost.
type boxRecordsBucket struct {
	model *leasetest.VersionedBucket
	mu    sync.Mutex
	seen  []s3Request
	aws   *testAWS
	doc   string

	afterGet       func()
	beforePut      func()
	commitThenFail []int
}

func (b *boxRecordsBucket) requests() []s3Request {
	b.mu.Lock()
	defer b.mu.Unlock()
	return append([]s3Request(nil), b.seen...)
}

func (b *boxRecordsBucket) handler() http.Handler {
	next, _ := leasetest.NewS3Server(b.model)
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, key, _ := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/")
		b.mu.Lock()
		b.seen = append(b.seen, s3Request{r.Method, key, r.Header.Get("If-None-Match"), r.Header.Get("If-Match")})
		var afterGet, beforePut func()
		fail := 0
		switch r.Method {
		case http.MethodGet:
			afterGet, b.afterGet = b.afterGet, nil
		case http.MethodPut:
			beforePut, b.beforePut = b.beforePut, nil
			if len(b.commitThenFail) > 0 {
				fail, b.commitThenFail = b.commitThenFail[0], b.commitThenFail[1:]
			}
		}
		b.mu.Unlock()
		if strings.HasPrefix(key, boxid.Prefix) {
			if r.Method == http.MethodDelete || (r.Method == http.MethodPut && r.Header.Get("If-None-Match") != "*") {
				s3Error(w, http.StatusForbidden, "AccessDenied")
				return
			}
		}
		if beforePut != nil {
			beforePut()
		}
		if fail != 0 {
			next.ServeHTTP(httptest.NewRecorder(), r)
			s3Error(w, fail, http.StatusText(fail))
			return
		}
		next.ServeHTTP(w, r)
		if afterGet != nil {
			afterGet()
		}
	})
}

func s3Error(w http.ResponseWriter, status int, code string) {
	w.Header().Set("Content-Type", "application/xml")
	w.WriteHeader(status)
	_, _ = w.Write([]byte("<Error><Code>" + strings.ReplaceAll(code, " ", "") + "</Code><Message>" + code + "</Message></Error>"))
}

// publishFixture points the process's AWS and CARLOS_* environment at a
// fresh fake bucket, as host.env would on a box, and makes the box's AWS.
func publishFixture(t *testing.T) *boxRecordsBucket {
	t.Helper()
	b := &boxRecordsBucket{model: leasetest.NewVersionedBucket(), aws: newTestAWS(t),
		doc: identityDoc(publishAccount, publishRegion, publishInstance, "2026-10-06T08:00:00Z", "t4g.small")}
	b.model.AddBucket(publishBucket)
	server := httptest.NewServer(b.handler())
	t.Cleanup(server.Close)
	for k, v := range map[string]string{
		"AWS_ACCESS_KEY_ID": "test-key", "AWS_SECRET_ACCESS_KEY": "test-secret", "AWS_SESSION_TOKEN": "",
		"AWS_REGION": "eu-west-1", "AWS_EC2_METADATA_DISABLED": "true",
		"AWS_CONFIG_FILE": "/nonexistent", "AWS_SHARED_CREDENTIALS_FILE": "/nonexistent",
		"CARLOS_S3_ENDPOINT": server.URL, "CARLOS_DEPLOYMENT_BUCKET": publishBucket,
		"CARLOS_DEPLOYMENT_DIR": "", "CARLOS_FLEET": "", "CARLOS_FLEET_BOX": "", "CARLOS_BOX_REFERENCE": "",
		"CARLOS_REGION_HA": "",
	} {
		t.Setenv(k, v)
	}
	return b
}

// imdsRead is the real IMDS reader over fakeIMDS serving b's current
// document and AWS's signature over it.
func (b *boxRecordsBucket) imdsRead(t *testing.T, written time.Time) func(string) (boxid.Record, error) {
	client := imdsClient(&fakeIMDS{doc: b.doc, sig: b.aws.sign(t, b.doc)})
	return func(box string) (boxid.Record, error) {
		return readIdentityRecord(client, "http://imds.test", box, written)
	}
}

// publishRun is the verb's production wiring with the test's seams: the
// agent user's uid, the fixed machine id, the fake IMDS and the test
// AWS's certificates.
func (b *boxRecordsBucket) publishRun(t *testing.T) boxPublishRun {
	return boxPublishRun{getenv: os.Getenv, geteuid: func() int { return publishAgentUID },
		machineID: func() string { return publishMachineID }, open: openDeployStoreEnv, read: b.imdsRead(t, recordNow),
		certs: b.aws.certs}
}

func (b *boxRecordsBucket) run(t *testing.T) (string, error) {
	t.Helper()
	return b.publishRun(t).publish(context.Background())
}

// thisBoxRecord is the record this box would publish now, written at
// writtenAt.
func (b *boxRecordsBucket) thisBoxRecord(t *testing.T, writtenAt string) boxid.Record {
	return b.aws.record(t, status.Label(publishMachineID), b.doc, writtenAt)
}

// assertCreateOnly: no DELETE, no request off this box's key, and every
// PUT create-only. Overwrites would also have been denied by the fake's
// policy, but the verb must never even ask.
func assertCreateOnly(t *testing.T, reqs []s3Request) (puts int) {
	t.Helper()
	key := boxid.Key(status.Label(publishMachineID))
	for _, r := range reqs {
		if r.Key != key {
			t.Errorf("%s %s: publish-record touched another key", r.Method, r.Key)
		}
		switch r.Method {
		case http.MethodPut:
			puts++
			if r.IfNoneMatch != "*" || r.IfMatch != "" {
				t.Errorf("PUT %s with If-None-Match %q If-Match %q; want create-only", r.Key, r.IfNoneMatch, r.IfMatch)
			}
		case http.MethodGet:
		default:
			t.Errorf("publish-record sent %s %s", r.Method, r.Key)
		}
	}
	return puts
}

func inPlaceLine(box string) string {
	return "the record at control/boxes/" + box + ".json is valid AWS-signed evidence for " + box +
		" (instance i-0abc), so nothing was written. enrollment still makes its own checks before it accepts the box"
}

func TestPublishRecordWritesAFreshBoxsRecord(t *testing.T) {
	b := publishFixture(t)
	box := status.Label(publishMachineID)
	line, err := b.run(t)
	if err != nil {
		t.Fatal(err)
	}
	if want := "published the identity record for " + box + " (instance i-0abc) to control/boxes/" + box + ".json"; line != want {
		t.Fatalf("line = %q\nwant   %q", line, want)
	}
	raw, ok := b.model.Current(publishBucket, "control/boxes/"+box+".json")
	if !ok {
		t.Fatal("no record at control/boxes/<box>.json")
	}
	var rec map[string]any
	if err := json.Unmarshal(raw, &rec); err != nil {
		t.Fatal(err)
	}
	want := b.thisBoxRecord(t, "2026-10-06T09:00:00Z")
	if len(rec) != 5 || rec["v"] != float64(1) || rec["box"] != box || rec["identity_document"] != want.IdentityDocument ||
		rec["written_at"] != "2026-10-06T09:00:00Z" {
		t.Fatalf("record = %v", rec)
	}
	// The record the verb wrote is evidence enrollment accepts.
	certs, _ := b.aws.certs()
	if _, err := (boxid.Verifier{Account: publishAccount, Certs: certs}).Enroll(raw, boxid.Expect{Box: box, AWSRegion: publishRegion}); err != nil {
		t.Fatalf("the published record does not enroll: %v", err)
	}
	if puts := assertCreateOnly(t, b.requests()); puts != 1 {
		t.Fatalf("%d PUTs, want one", puts)
	}
}

// R3: there is no region HA gate. Run after the loop has published (its
// publishBoxRecord, store, label and IMDS read, an hour earlier), the
// verb keeps that record and changes nothing.
func TestPublishRecordAfterTheLoopIsANoOp(t *testing.T) {
	b := publishFixture(t)
	ctx := context.Background()
	store, err := openDeployStoreEnv(ctx, os.Getenv)
	if err != nil {
		t.Fatal(err)
	}
	box := metricsBoxLabel(os.Getenv, publishMachineID)
	loopRead := b.imdsRead(t, recordNow.Add(-time.Hour))
	if done, err := publishBoxRecord(ctx, store, box, func() (boxid.Record, error) { return loopRead(box) }); !done || err != nil {
		t.Fatalf("the loop's publish: done=%v err=%v", done, err)
	}
	before, _ := b.model.Current(publishBucket, boxid.Key(box))
	seenBefore := len(b.requests())
	line, err := b.run(t)
	if err != nil {
		t.Fatal(err)
	}
	if want := inPlaceLine(box); line != want {
		t.Fatalf("line = %q\nwant   %q", line, want)
	}
	if after, _ := b.model.Current(publishBucket, boxid.Key(box)); string(after) != string(before) {
		t.Fatalf("the loop's record was rewritten:\n%s\n%s", before, after)
	}
	if puts := assertCreateOnly(t, b.requests()[seenBefore:]); puts != 0 {
		t.Fatalf("%d PUTs, want none", puts)
	}
}

// Astra, round 2: an older record AWS signed for this same instance, in
// this account and region, is evidence enrollment accepts, although its
// document's mutable fields have changed since (a stop and start, a
// resize). It is kept byte for byte, never replaced, and nothing is
// written.
func TestPublishRecordKeepsAnOlderValidRecord(t *testing.T) {
	b := publishFixture(t)
	box := status.Label(publishMachineID)
	older := identityDoc(publishAccount, publishRegion, publishInstance, "2026-09-01T00:00:00Z", "t4g.micro")
	if older == b.doc {
		t.Fatal("the two documents are the same")
	}
	before := mustJSON(t, b.aws.record(t, box, older, "2026-09-01T00:05:00Z"))
	b.model.Put(publishBucket, boxid.Key(box), before)
	line, err := b.run(t)
	if err != nil {
		t.Fatal(err)
	}
	if want := inPlaceLine(box); line != want {
		t.Fatalf("line = %q\nwant   %q", line, want)
	}
	if after, _ := b.model.Current(publishBucket, boxid.Key(box)); string(after) != string(before) {
		t.Fatalf("the older record was replaced:\n%s\n%s", before, after)
	}
	if puts := assertCreateOnly(t, b.requests()); puts != 0 {
		t.Fatalf("%d PUTs, want none", puts)
	}
}

// Anything enrollment's verifier refuses is refused, named, and left
// alone.
func TestPublishRecordRefusesWhatEnrollmentWouldRefuse(t *testing.T) {
	box := status.Label(publishMachineID)
	prefix := "refused: the record already under " + box + " is not valid evidence for this box: "
	type held struct {
		box, doc string
		forger   bool // signed by another key
		edit     func(*boxid.Record)
		raw      func([]byte) []byte
	}
	cases := map[string]struct {
		h    held
		want string
	}{
		"different account": {held{doc: identityDoc("999999999999", publishRegion, publishInstance, "2026-09-01T00:00:00Z", "t4g.small")}, "it is signed for account 999999999999, and this box is in account 111122223333"},
		"different region":  {held{doc: identityDoc(publishAccount, "eu-west-2", publishInstance, "2026-09-01T00:00:00Z", "t4g.small")}, "it is signed in region eu-west-2, and this box is in eu-west-1"},
		"another instance":  {held{doc: identityDoc(publishAccount, publishRegion, "i-0zzz", "2026-09-01T00:00:00Z", "t4g.small")}, "it is signed for instance i-0zzz, and this box is i-0abc"},
		"bad signature":     {held{forger: true}, "it does not verify as AWS-signed evidence: signature: crypto/rsa: verification error"},
		"missing signature": {held{edit: func(r *boxid.Record) { r.IdentitySignature = "" }}, "it does not verify as AWS-signed evidence: signature: crypto/rsa: verification error"},
		"altered document": {held{edit: func(r *boxid.Record) {
			r.IdentityDocument = base64.StdEncoding.EncodeToString([]byte(strings.Replace(identityDocFor(r), "t4g.small", "m7g.large", 1)))
		}}, "it does not verify as AWS-signed evidence: signature: crypto/rsa: verification error"},
		"different box":     {held{box: "bert-sawing"}, `it is for box "bert-sawing"`},
		"version 2":         {held{edit: func(r *boxid.Record) { r.V = 2 }}, "its version is 2, not 1"},
		"unknown field":     {held{raw: func(b []byte) []byte { return append(bytes.TrimSuffix(b, []byte("}")), []byte(`,"extra":1}`)...) }}, "it cannot be read as an identity record"},
		"trailing data":     {held{raw: func(b []byte) []byte { return append(b, []byte(`{}`)...) }}, "it cannot be read as an identity record"},
		"not json":          {held{raw: func([]byte) []byte { return []byte("not json") }}, "it cannot be read as an identity record"},
		"document not json": {held{edit: func(r *boxid.Record) { r.IdentityDocument = base64.StdEncoding.EncodeToString([]byte("nope")) }}, "it cannot be read as an identity record"},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			b := publishFixture(t)
			signer, doc, hbox := b.aws, b.doc, box
			if tc.h.forger {
				signer = newTestAWS(t)
			}
			if tc.h.doc != "" {
				doc = tc.h.doc
			}
			if tc.h.box != "" {
				hbox = tc.h.box
			}
			rec := signer.record(t, hbox, doc, "2026-09-01T00:05:00Z")
			if tc.h.edit != nil {
				tc.h.edit(&rec)
			}
			before := mustJSON(t, rec)
			if tc.h.raw != nil {
				before = tc.h.raw(before)
			}
			b.model.Put(publishBucket, boxid.Key(box), before)
			line, err := b.run(t)
			if want := prefix + tc.want + ". nothing was changed"; err == nil || err.Error() != want {
				t.Fatalf("line=%q err=%v\nwant %q", line, err, want)
			}
			if after, _ := b.model.Current(publishBucket, boxid.Key(box)); string(after) != string(before) {
				t.Fatalf("the record was changed:\n%s\n%s", before, after)
			}
			if puts := assertCreateOnly(t, b.requests()); puts != 0 {
				t.Fatalf("%d PUTs, want none", puts)
			}
		})
	}
}

// identityDocFor is r's decoded identity document.
func identityDocFor(r *boxid.Record) string {
	doc, _ := base64.StdEncoding.DecodeString(r.IdentityDocument)
	return string(doc)
}

// A record that arrives after the verb's read and before
// publishBoxRecord's is held to the same verifier: publishBoxRecord alone
// would take any record naming this instance.
func TestPublishRecordHoldsALateRecordToTheSameVerifier(t *testing.T) {
	box := status.Label(publishMachineID)
	for name, tc := range map[string]struct {
		forged bool
		want   string
	}{
		"valid, older":                    {false, inPlaceLine(box)},
		"same instance, forged signature": {true, "refused: the record already under " + box + " is not valid evidence for this box: it does not verify as AWS-signed evidence: signature: crypto/rsa: verification error. nothing was changed"},
	} {
		t.Run(name, func(t *testing.T) {
			b := publishFixture(t)
			signer := b.aws
			if tc.forged {
				signer = newTestAWS(t)
			}
			late := mustJSON(t, signer.record(t, box, identityDoc(publishAccount, publishRegion, publishInstance, "2026-09-01T00:00:00Z", "t4g.micro"), "2026-09-01T00:05:00Z"))
			b.afterGet = func() { b.model.Put(publishBucket, boxid.Key(box), late) }
			line, err := b.run(t)
			got := line
			if err != nil {
				got = err.Error()
			}
			if got != tc.want || (err != nil) != tc.forged {
				t.Fatalf("line=%q err=%v\nwant %q", line, err, tc.want)
			}
			if after, _ := b.model.Current(publishBucket, boxid.Key(box)); string(after) != string(late) {
				t.Fatalf("the late record was changed:\n%s", after)
			}
			if puts := assertCreateOnly(t, b.requests()); puts != 0 {
				t.Fatalf("%d PUTs, want none", puts)
			}
		})
	}
}

// The conditional create is the bucket's: a record another instance
// creates just before the verb's write is not overwritten, because the
// fake answers the If-None-Match: * PUT with 412.
func TestPublishRecordLosesARaceWithoutOverwriting(t *testing.T) {
	b := publishFixture(t)
	box := status.Label(publishMachineID)
	racer := mustJSON(t, b.aws.record(t, box, identityDoc(publishAccount, publishRegion, "i-0zzz", "2026-10-06T08:00:00Z", "t4g.small"), "2026-10-01T00:00:00Z"))
	b.beforePut = func() { b.model.Put(publishBucket, boxid.Key(box), racer) }
	want := "refused: another record already holds " + box + ": boxid: " + box + " already has a different record: store: compare-and-swap conflict. nothing was changed"
	if line, err := b.run(t); err == nil || err.Error() != want {
		t.Fatalf("line=%q err=%v\nwant %q", line, err, want)
	}
	if after, _ := b.model.Current(publishBucket, boxid.Key(box)); string(after) != string(racer) {
		t.Fatalf("the racing record was overwritten:\n%s", after)
	}
	if puts := assertCreateOnly(t, b.requests()); puts != 1 {
		t.Fatalf("%d PUTs, want the one conditional create", puts)
	}
}

// S3 commits the create but its answer is lost (500). The SDK retries,
// the retry is answered 412, and boxid.Publish finds exactly this run's
// bytes. That is this run's record: published, although the store never
// saw a successful Put.
func TestPublishRecordReportsARetriedCreateAsPublished(t *testing.T) {
	b := publishFixture(t)
	box := status.Label(publishMachineID)
	b.commitThenFail = []int{http.StatusInternalServerError}
	line, err := b.run(t)
	if err != nil {
		t.Fatal(err)
	}
	if want := "published the identity record for " + box + " (instance i-0abc) to control/boxes/" + box + ".json"; line != want {
		t.Fatalf("line = %q\nwant   %q", line, want)
	}
	got, _ := b.model.Current(publishBucket, boxid.Key(box))
	var rec boxid.Record
	if json.Unmarshal(got, &rec) != nil || rec.WrittenAt != "2026-10-06T09:00:00Z" || rec.IdentityDocument != base64.StdEncoding.EncodeToString([]byte(b.doc)) {
		t.Fatalf("stored %s", got)
	}
	if puts := assertCreateOnly(t, b.requests()); puts != 2 {
		t.Fatalf("%d PUTs, want the create and its retry", puts)
	}
}

// Once a create has reached the bucket, a failure does not claim nothing
// was written. Here S3 committed it and answered 400.
func TestPublishRecordSaysAnUncertainWriteMayHaveHappened(t *testing.T) {
	b := publishFixture(t)
	box := status.Label(publishMachineID)
	b.commitThenFail = []int{http.StatusBadRequest}
	_, err := b.run(t)
	if err == nil || !strings.HasPrefix(err.Error(), "the identity record for "+box+" may or may not have been written: ") ||
		!strings.HasSuffix(err.Error(), ". it is safe to run again, which only ever creates the record and says what is there") {
		t.Fatalf("err = %v", err)
	}
	if _, ok := b.model.Current(publishBucket, boxid.Key(box)); !ok {
		t.Fatal("the fake did not commit the create")
	}
	// The second run finds valid evidence and says so.
	if line, err := b.run(t); err != nil || line != inPlaceLine(box) {
		t.Fatalf("second run: line=%q err=%v", line, err)
	}
}

// Failures before any create reached the bucket say nothing was written.
func TestPublishRecordReportsFailuresBeforeAnyWriteAsRetryable(t *testing.T) {
	box := status.Label(publishMachineID)
	for name, tc := range map[string]struct {
		imdsDown bool
		doc      string // the document IMDS serves, when not the usual one
		reason   string
	}{
		"IMDS down":               {imdsDown: true, reason: "IMDS down"},
		"own identity incomplete": {doc: "{\"accountId\" : \"111122223333\", \"region\" : \"eu-west-1\"}\n", reason: "the identity document AWS served this box names no account, region or instance"},
	} {
		t.Run(name, func(t *testing.T) {
			b := publishFixture(t)
			if tc.doc != "" {
				b.doc = tc.doc
			}
			run := b.publishRun(t)
			if tc.imdsDown {
				run.read = func(string) (boxid.Record, error) { return boxid.Record{}, errors.New("IMDS down") }
			}
			_, err := run.publish(context.Background())
			want := "the identity record for " + box + " was not published: " + tc.reason + ". nothing was written, so it is safe to run again"
			if err == nil || err.Error() != want {
				t.Fatalf("err = %v\nwant  %q", err, want)
			}
			if puts := assertCreateOnly(t, b.requests()); puts != 0 {
				t.Fatalf("%d PUTs, want none", puts)
			}
		})
	}
}

// refusedEarly runs the verb with every later seam tripwired: it must
// refuse before the machine id, the store, IMDS or the certificates are
// touched.
func refusedEarly(t *testing.T, b *boxRecordsBucket, run boxPublishRun, want string) {
	t.Helper()
	touched := []string{}
	run.machineID = func() string { touched = append(touched, "machine id"); return publishMachineID }
	run.open = func(context.Context, func(string) string) (release.Store, error) {
		touched = append(touched, "store")
		return nil, errors.New("opened")
	}
	run.read = func(string) (boxid.Record, error) { touched = append(touched, "IMDS"); return boxid.Record{}, nil }
	run.certs = func() (map[string]*x509.Certificate, error) { touched = append(touched, "certs"); return nil, nil }
	_, err := run.publish(context.Background())
	if err == nil || err.Error() != want {
		t.Fatalf("err = %v\nwant  %q", err, want)
	}
	if len(touched) != 0 || len(b.requests()) != 0 {
		t.Fatalf("read %v and sent %d requests before refusing", touched, len(b.requests()))
	}
}

// Census ruling R1: root is refused before any store, IMDS or host read.
func TestPublishRecordRefusesRoot(t *testing.T) {
	b := publishFixture(t)
	run := b.publishRun(t)
	run.geteuid = func() int { return 0 }
	refusedEarly(t, b, run, "carlos system box publish-record runs as the agent user, not root. run it with systemd-run -p User=carlos.")
}

func TestPublishRecordRefusesAFleetBox(t *testing.T) {
	b := publishFixture(t)
	t.Setenv("CARLOS_FLEET", "acme")
	t.Setenv("CARLOS_FLEET_BOX", "kass-1")
	refusedEarly(t, b, b.publishRun(t), boxRecordFleetRefusal)
}

// A deployment directory cannot refuse an overwrite (DirStore.Put checks,
// then renames), so only an S3 bucket will do.
func TestPublishRecordRequiresAnS3Bucket(t *testing.T) {
	b := publishFixture(t)
	dir := t.TempDir()
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
	t.Setenv("CARLOS_DEPLOYMENT_DIR", dir)
	refusedEarly(t, b, b.publishRun(t), boxRecordNoBucket)
	if entries, _ := os.ReadDir(dir); len(entries) != 0 {
		t.Fatalf("the deployment directory was written: %v", entries)
	}
	// Belt and braces: a store that is not S3 is refused even with a
	// bucket named.
	t.Setenv("CARLOS_DEPLOYMENT_BUCKET", publishBucket)
	run := b.publishRun(t)
	run.open = func(context.Context, func(string) string) (release.Store, error) {
		return release.NewDirStore(dir), nil
	}
	if _, err := run.publish(context.Background()); err == nil || err.Error() != boxRecordNoBucket {
		t.Fatalf("err = %v", err)
	}
	if entries, _ := os.ReadDir(dir); len(entries) != 0 {
		t.Fatalf("the deployment directory was written: %v", entries)
	}
}

// A box its region's group launched publishes under its group reference,
// the label boxRecordLoop takes (metricsBoxLabel).
func TestPublishRecordUsesTheGroupReference(t *testing.T) {
	b := publishFixture(t)
	t.Setenv("CARLOS_BOX_REFERENCE", "group")
	withGroupIdentity(t, publishAccount, publishInstance, nil)
	ref := hostactors.GroupBoxReference(publishAccount, publishInstance)
	if loop := metricsBoxLabel(os.Getenv, publishMachineID); loop != ref {
		t.Fatalf("the loop's label is %q, want the group reference %q", loop, ref)
	}
	line, err := b.run(t)
	if err != nil {
		t.Fatal(err)
	}
	if want := "published the identity record for " + ref + " (instance i-0abc) to control/boxes/" + ref + ".json"; line != want {
		t.Fatalf("line = %q\nwant   %q", line, want)
	}
	raw, ok := b.model.Current(publishBucket, boxid.Key(ref))
	var rec boxid.Record
	if !ok || json.Unmarshal(raw, &rec) != nil || rec.Box != ref {
		t.Fatalf("record under the group reference: %s", raw)
	}
	if _, ok := b.model.Current(publishBucket, boxid.Key(status.Label(publishMachineID))); ok {
		t.Fatal("a group box published under its machine-id label")
	}
}

// The verb's certificates are enrollment's: boxid.EmbeddedCerts covers
// the dev region, so a real box's record can verify.
func TestPublishRecordProductionCertsCoverTheDevRegion(t *testing.T) {
	certs, err := boxid.EmbeddedCerts()
	if err != nil || certs[publishRegion] == nil {
		t.Fatalf("embedded certs: %v (eu-west-1 present: %v)", err, certs[publishRegion] != nil)
	}
}

// Census ruling R4: the store publish-record hands publishBoxRecord sends
// the bucket nothing for a delete, a list, an overwrite or any other key.
func TestBoxRecordStoreOnlyReadsAndCreatesItsOwnKey(t *testing.T) {
	b := publishFixture(t)
	ctx := context.Background()
	inner, err := openDeployStoreEnv(ctx, os.Getenv)
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := inner.(*release.S3Store); !ok {
		t.Fatalf("store is %T, want the S3Store", inner)
	}
	s := &boxRecordStore{inner: inner, key: boxid.Key("annie-welding")}
	if _, err := s.Put(ctx, boxid.Key("annie-welding"), []byte("{}"), `"etag"`); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("overwrite (replace-only Put): %v", err)
	}
	if _, err := s.Put(ctx, boxid.Key("bert-sawing"), []byte("{}"), ""); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("Put on another box's key: %v", err)
	}
	if _, err := s.Put(ctx, "control/serial", []byte("{}"), ""); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("Put on another key: %v", err)
	}
	if _, _, err := s.Get(ctx, boxid.Key("bert-sawing")); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("Get on another box's key: %v", err)
	}
	if err := s.Delete(ctx, boxid.Key("annie-welding")); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("Delete: %v", err)
	}
	if _, err := s.List(ctx, boxid.Prefix); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("List: %v", err)
	}
	if _, err := s.ListDirs(ctx, "control/"); !errors.Is(err, errBoxRecordStore) {
		t.Errorf("ListDirs: %v", err)
	}
	if reqs := b.requests(); len(reqs) != 0 || s.attempted || s.wrote {
		t.Fatalf("refused calls reached S3: %v (attempted=%v wrote=%v)", reqs, s.attempted, s.wrote)
	}
	if _, err := s.Put(ctx, boxid.Key("annie-welding"), []byte("{}"), ""); err != nil || !s.wrote {
		t.Fatalf("create: err=%v wrote=%v", err, s.wrote)
	}
	if _, err := s.Put(ctx, boxid.Key("annie-welding"), []byte("{}"), ""); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("a second create: %v, want the bucket's 412 as ErrCASConflict", err)
	}
	if reqs := b.requests(); len(reqs) != 2 || reqs[0].IfNoneMatch != "*" || reqs[1].IfNoneMatch != "*" {
		t.Fatalf("requests %v, want two create-only PUTs", reqs)
	}
}

// boxRecordLoop's gate is unchanged: with CARLOS_REGION_HA off, or on a
// fleet box, it returns at once and never reaches the bucket. Past the
// gate it would retry each minute, so a broken gate shows as a timeout.
func TestBoxRecordLoopGateIsUnchanged(t *testing.T) {
	for name, env := range map[string]map[string]string{
		"region HA off":           {"CARLOS_REGION_HA": ""},
		"region HA unparseable":   {"CARLOS_REGION_HA": "yes"},
		"fleet box, region HA on": {"CARLOS_REGION_HA": "1", "CARLOS_FLEET": "acme", "CARLOS_FLEET_BOX": "kass-1"},
	} {
		t.Run(name, func(t *testing.T) {
			b := publishFixture(t)
			for k, v := range env {
				t.Setenv(k, v)
			}
			done := make(chan struct{})
			go func() { boxRecordLoop(); close(done) }()
			select {
			case <-done:
			case <-time.After(10 * time.Second):
				t.Fatal("boxRecordLoop did not return: its region HA gate let it run")
			}
			if n := len(b.requests()); n != 0 {
				t.Fatalf("boxRecordLoop sent %d requests with its gate shut", n)
			}
		})
	}
}

// Box publication reaches the deployment bucket through openDeployStoreEnv
// and requires the *release.S3Store it returns (boxpublish.go's
// opened.(*release.S3Store)). The change index must not cost it that type:
// the opener hands back the same concrete store with the index enabled,
// against the real S3 wiring, and publication still writes only its own
// control/boxes/ key (a key outside every change scope, so no index traffic).
// Not parallel: publishFixture sets the process environment.
func TestBoxPublishThroughDeploymentOpener(t *testing.T) {
	b := publishFixture(t)
	ctx := context.Background()
	opened, err := openDeployStoreEnv(ctx, os.Getenv)
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := opened.(*release.S3Store); !ok {
		t.Fatalf("openDeployStoreEnv returned %T; box publication needs the *release.S3Store", opened)
	}
	// The opener enabled the index: a scoped write bumps its generation.
	if _, err := opened.Put(ctx, "accounts/shop/manifest.json", []byte("{}"), ""); err != nil {
		t.Fatal(err)
	}
	idx, _, err := release.ReadChangeIndex(ctx, opened)
	if err != nil {
		t.Fatal(err)
	}
	if got := idx.Entries["accounts/shop"]; got != 1 {
		t.Fatalf("accounts/shop generation = %d after one scoped Put, want 1 (entries %v)", got, idx.Entries)
	}
	// Publication itself is unchanged: one create-only PUT on its own key.
	before := len(b.requests())
	if _, err := b.run(t); err != nil {
		t.Fatal(err)
	}
	if puts := assertCreateOnly(t, b.requests()[before:]); puts != 1 {
		t.Fatalf("%d PUTs, want one", puts)
	}
}