package steering

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"testing"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/route53"
	r53types "github.com/aws/aws-sdk-go-v2/service/route53/types"
)

// ---------------------------------------------------------------------------
// the fake
// ---------------------------------------------------------------------------

// fakeR53 is an in-memory Route53 that records every call and APPLIES the
// changes it is handed, so a second converge against the same fake sees the
// state the first one left. No network, ever — the whole point of route53API
// being an interface.
//
// It deliberately reproduces the real API's refusals that matter to this
// converger: DELETE of a record that is not present errors (Route53's
// InvalidChangeBatch), a name+type may not hold BOTH a plain record and a
// routing-policy record set (also InvalidChangeBatch, and it takes the whole
// atomic batch with it), ListTagsForResources refuses more than 10 ids, and
// DeleteHealthCheck of a check still referenced by a record errors
// (HealthCheckInUse). A fake that is more permissive than the service would
// let a bug through — the plain-vs-policy rule is in here precisely because
// an earlier version of this file was more permissive and enshrined a zone
// state Route53 forbids.
type fakeR53 struct {
	records []r53types.ResourceRecordSet
	checks  []r53types.HealthCheck
	tags    map[string][]r53types.Tag // health check id -> tags

	pageSize int // 0 = one page for everything
	nextID   int

	// call counters — the idempotency assertions live on these
	nListRecords   int
	nChangeRecords int
	nListChecks    int
	nListTags      int
	nCreateCheck   int
	nDeleteCheck   int
	nChangeTags    int

	submitted     []r53types.Change // every change submitted, in order
	createdChecks []*route53.CreateHealthCheckInput
	deletedChecks []string
	taggedIDs     []string

	failChange error // ChangeResourceRecordSets returns this
	failTags   error // ChangeTagsForResource returns this
}

func newFake() *fakeR53 { return &fakeR53{tags: map[string][]r53types.Tag{}} }

// writes is the total number of state-changing calls — the number an
// idempotent second pass must leave at zero.
func (f *fakeR53) writes() int {
	return f.nChangeRecords + f.nCreateCheck + f.nDeleteCheck + f.nChangeTags
}

// validateZone enforces the three Route53 rules this converger can violate,
// each of which rejects the WHOLE batch:
//
//  1. a name+type holds either one simple record OR a routing-policy record
//     set, never both;
//  2. a name holding a CNAME holds no other record type;
//  3. one routing policy per name+type — a weighted set and a latency set
//     cannot share a name.
//
// Modelled on the projected post-batch state, so a batch is judged by what it
// would produce, atomically.
func validateZone(recs []r53types.ResourceRecordSet) error {
	type nt struct{ name, typ string }
	plain, policy := map[nt]bool{}, map[nt]bool{}
	policies := map[nt]map[string]bool{}
	names := map[string]map[string]bool{} // name -> set of types
	for _, rr := range recs {
		name := strings.ToLower(aws.ToString(rr.Name))
		k := nt{name, string(rr.Type)}
		if names[name] == nil {
			names[name] = map[string]bool{}
		}
		names[name][string(rr.Type)] = true
		if aws.ToString(rr.SetIdentifier) == "" {
			plain[k] = true
			continue
		}
		policy[k] = true
		if policies[k] == nil {
			policies[k] = map[string]bool{}
		}
		policies[k][fakePolicyKind(rr)] = true
	}
	for k := range plain {
		if policy[k] {
			return fmt.Errorf("fake: InvalidChangeBatch: RRSet with DNS name %s and type %s cannot be both a simple record and a routing-policy record set", k.name, k.typ)
		}
	}
	for name, types := range names {
		if types["CNAME"] && len(types) > 1 {
			return fmt.Errorf("fake: InvalidChangeBatch: RRSet with DNS name %s has a CNAME, which cannot coexist with any other record type", name)
		}
	}
	for k, kinds := range policies {
		if len(kinds) > 1 {
			return fmt.Errorf("fake: InvalidChangeBatch: RRSet with DNS name %s and type %s mixes routing policies (%v) — only one policy per name and type", k.name, k.typ, sortedKeys(kinds))
		}
	}
	return nil
}

func fakePolicyKind(rr r53types.ResourceRecordSet) string {
	switch {
	case rr.Region != "":
		return "latency"
	case rr.Weight != nil:
		return "weighted"
	case rr.Failover != "":
		return "failover"
	case rr.GeoLocation != nil:
		return "geolocation"
	case rr.GeoProximityLocation != nil:
		return "geoproximity"
	case rr.CidrRoutingConfig != nil:
		return "cidr"
	case aws.ToBool(rr.MultiValueAnswer):
		return "multivalue"
	}
	return "unknown"
}

func sortedKeys(m map[string]bool) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

func rrKey(rr r53types.ResourceRecordSet) string {
	return strings.ToLower(aws.ToString(rr.Name)) + "|" + string(rr.Type) + "|" + aws.ToString(rr.SetIdentifier)
}

func (f *fakeR53) ListResourceRecordSets(ctx context.Context, in *route53.ListResourceRecordSetsInput, _ ...func(*route53.Options)) (*route53.ListResourceRecordSetsOutput, error) {
	f.nListRecords++
	if aws.ToString(in.HostedZoneId) == "" {
		return nil, errors.New("fake: no hosted zone id")
	}
	start := 0
	if in.StartRecordName != nil {
		for i, rr := range f.records {
			if strings.EqualFold(aws.ToString(rr.Name), aws.ToString(in.StartRecordName)) &&
				rr.Type == in.StartRecordType &&
				aws.ToString(rr.SetIdentifier) == aws.ToString(in.StartRecordIdentifier) {
				start = i
				break
			}
		}
	}
	end := len(f.records)
	if f.pageSize > 0 && start+f.pageSize < end {
		end = start + f.pageSize
	}
	out := &route53.ListResourceRecordSetsOutput{
		ResourceRecordSets: append([]r53types.ResourceRecordSet(nil), f.records[start:end]...),
	}
	if end < len(f.records) {
		next := f.records[end]
		out.IsTruncated = true
		out.NextRecordName = next.Name
		out.NextRecordType = next.Type
		out.NextRecordIdentifier = next.SetIdentifier
	}
	return out, nil
}

func (f *fakeR53) ChangeResourceRecordSets(ctx context.Context, in *route53.ChangeResourceRecordSetsInput, _ ...func(*route53.Options)) (*route53.ChangeResourceRecordSetsOutput, error) {
	f.nChangeRecords++
	if f.failChange != nil {
		return nil, f.failChange
	}
	if in.ChangeBatch == nil || len(in.ChangeBatch.Changes) == 0 {
		return nil, errors.New("fake: empty change batch")
	}

	// A batch is ATOMIC: project it onto a copy, validate the result, and
	// commit only if the whole thing is legal.
	projected := append([]r53types.ResourceRecordSet(nil), f.records...)
	for _, c := range in.ChangeBatch.Changes {
		rr := *c.ResourceRecordSet
		key := rrKey(rr)
		idx := -1
		for i, have := range projected {
			if rrKey(have) == key {
				idx = i
				break
			}
		}
		switch c.Action {
		case r53types.ChangeActionUpsert:
			if idx >= 0 {
				projected[idx] = rr
			} else {
				projected = append(projected, rr)
			}
		case r53types.ChangeActionDelete:
			if idx < 0 {
				return nil, fmt.Errorf("fake: InvalidChangeBatch: no such record %s", key)
			}
			projected = append(projected[:idx], projected[idx+1:]...)
		default:
			return nil, fmt.Errorf("fake: unsupported action %q", c.Action)
		}
	}

	if err := validateZone(projected); err != nil {
		return nil, err
	}

	f.submitted = append(f.submitted, in.ChangeBatch.Changes...)
	f.records = projected
	sort.SliceStable(f.records, func(i, j int) bool { return rrKey(f.records[i]) < rrKey(f.records[j]) })
	return &route53.ChangeResourceRecordSetsOutput{}, nil
}

func (f *fakeR53) ListHealthChecks(ctx context.Context, in *route53.ListHealthChecksInput, _ ...func(*route53.Options)) (*route53.ListHealthChecksOutput, error) {
	f.nListChecks++
	start := 0
	if in.Marker != nil {
		for i, hc := range f.checks {
			if aws.ToString(hc.Id) == aws.ToString(in.Marker) {
				start = i
				break
			}
		}
	}
	end := len(f.checks)
	if f.pageSize > 0 && start+f.pageSize < end {
		end = start + f.pageSize
	}
	out := &route53.ListHealthChecksOutput{
		HealthChecks: append([]r53types.HealthCheck(nil), f.checks[start:end]...),
	}
	if end < len(f.checks) {
		out.IsTruncated = true
		out.NextMarker = f.checks[end].Id
	}
	return out, nil
}

func (f *fakeR53) ListTagsForResources(ctx context.Context, in *route53.ListTagsForResourcesInput, _ ...func(*route53.Options)) (*route53.ListTagsForResourcesOutput, error) {
	f.nListTags++
	if in.ResourceType != r53types.TagResourceTypeHealthcheck {
		return nil, fmt.Errorf("fake: unexpected resource type %q", in.ResourceType)
	}
	if len(in.ResourceIds) == 0 || len(in.ResourceIds) > 10 {
		return nil, fmt.Errorf("fake: ListTagsForResources takes 1-10 ids, got %d", len(in.ResourceIds))
	}
	out := &route53.ListTagsForResourcesOutput{}
	for _, id := range in.ResourceIds {
		out.ResourceTagSets = append(out.ResourceTagSets, r53types.ResourceTagSet{
			ResourceId:   aws.String(id),
			ResourceType: r53types.TagResourceTypeHealthcheck,
			Tags:         f.tags[id],
		})
	}
	return out, nil
}

func (f *fakeR53) CreateHealthCheck(ctx context.Context, in *route53.CreateHealthCheckInput, _ ...func(*route53.Options)) (*route53.CreateHealthCheckOutput, error) {
	f.nCreateCheck++
	f.createdChecks = append(f.createdChecks, in)
	f.nextID++
	id := fmt.Sprintf("hc-%d", f.nextID)
	hc := r53types.HealthCheck{
		Id:                aws.String(id),
		CallerReference:   in.CallerReference,
		HealthCheckConfig: in.HealthCheckConfig,
	}
	f.checks = append(f.checks, hc)
	return &route53.CreateHealthCheckOutput{HealthCheck: &hc, Location: aws.String("https://route53/" + id)}, nil
}

func (f *fakeR53) DeleteHealthCheck(ctx context.Context, in *route53.DeleteHealthCheckInput, _ ...func(*route53.Options)) (*route53.DeleteHealthCheckOutput, error) {
	f.nDeleteCheck++
	id := aws.ToString(in.HealthCheckId)
	for _, rr := range f.records {
		if aws.ToString(rr.HealthCheckId) == id {
			return nil, fmt.Errorf("fake: HealthCheckInUse: %s", id)
		}
	}
	f.deletedChecks = append(f.deletedChecks, id)
	for i, hc := range f.checks {
		if aws.ToString(hc.Id) == id {
			f.checks = append(f.checks[:i], f.checks[i+1:]...)
			break
		}
	}
	delete(f.tags, id)
	return &route53.DeleteHealthCheckOutput{}, nil
}

func (f *fakeR53) ChangeTagsForResource(ctx context.Context, in *route53.ChangeTagsForResourceInput, _ ...func(*route53.Options)) (*route53.ChangeTagsForResourceOutput, error) {
	f.nChangeTags++
	if f.failTags != nil {
		return nil, f.failTags
	}
	id := aws.ToString(in.ResourceId)
	f.taggedIDs = append(f.taggedIDs, id)
	f.tags[id] = append(f.tags[id], in.AddTags...)
	return &route53.ChangeTagsForResourceOutput{}, nil
}

// ---------------------------------------------------------------------------
// fixtures + helpers
// ---------------------------------------------------------------------------

const testZone = "Z0STEERINGTEST"

var (
	dublinRec = desiredRecord{Name: "edge.example.com.", EIP: "1.2.3.4", Box: "dublin", Region: "eu-west-1"}
	oregonRec = desiredRecord{Name: "edge.example.com.", EIP: "5.6.7.8", Box: "oregon", Region: "us-west-2"}
)

// hostRecs mints the per-host half of a desired set: one record per edge.
func hostRecs(name string, edges ...desiredRecord) []desiredRecord {
	out := make([]desiredRecord, 0, len(edges))
	for _, e := range edges {
		out = append(out, desiredRecord{Name: name, EIP: e.EIP, Box: e.Box, Region: e.Region})
	}
	return out
}

func staticA(name, ip string) r53types.ResourceRecordSet {
	return r53types.ResourceRecordSet{
		Name:            aws.String(name),
		Type:            r53types.RRTypeA,
		TTL:             aws.Int64(300),
		ResourceRecords: []r53types.ResourceRecord{{Value: aws.String(ip)}},
	}
}

func recordNames(recs []r53types.ResourceRecordSet) []string {
	out := make([]string, 0, len(recs))
	for _, rr := range recs {
		out = append(out, rrKey(rr))
	}
	sort.Strings(out)
	return out
}

func find(t *testing.T, f *fakeR53, name, setID string) r53types.ResourceRecordSet {
	t.Helper()
	for _, rr := range f.records {
		if strings.EqualFold(aws.ToString(rr.Name), name) && aws.ToString(rr.SetIdentifier) == setID {
			return rr
		}
	}
	t.Fatalf("no record %s / setid %q in zone; have %v", name, setID, recordNames(f.records))
	return r53types.ResourceRecordSet{}
}

func mustConverge(t *testing.T, f *fakeR53, want []desiredRecord) {
	t.Helper()
	if err := converge(context.Background(), f, testZone, want); err != nil {
		t.Fatalf("converge: %v", err)
	}
}

// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------

// TestConvergeCreateFromEmpty is the brief's create-from-empty case: an
// empty zone, two armed edges, one steered host.
func TestConvergeCreateFromEmpty(t *testing.T) {
	f := newFake()
	want := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, want)

	if len(f.records) != 4 {
		t.Fatalf("want 4 records, got %d: %v", len(f.records), recordNames(f.records))
	}
	if f.nCreateCheck != 2 {
		t.Fatalf("want one health check per distinct EIP (2), got %d", f.nCreateCheck)
	}
	if f.nChangeTags != 2 {
		t.Fatalf("want every created check tagged (2), got %d", f.nChangeTags)
	}

	// health check shape
	byIP := map[string]*route53.CreateHealthCheckInput{}
	for _, in := range f.createdChecks {
		byIP[aws.ToString(in.HealthCheckConfig.IPAddress)] = in
	}
	for _, ip := range []string{"1.2.3.4", "5.6.7.8"} {
		in, ok := byIP[ip]
		if !ok {
			t.Fatalf("no health check created for %s", ip)
		}
		cfg := in.HealthCheckConfig
		if cfg.Type != r53types.HealthCheckTypeHttps {
			t.Errorf("%s: type = %q, want HTTPS", ip, cfg.Type)
		}
		if aws.ToInt32(cfg.Port) != 443 {
			t.Errorf("%s: port = %d, want 443", ip, aws.ToInt32(cfg.Port))
		}
		if !aws.ToBool(cfg.EnableSNI) {
			t.Errorf("%s: EnableSNI false, want true (the SNI name is the whole point)", ip)
		}
		// FQDN = the steering hostname, dot-trimmed (it is an SNI/Host value,
		// not a Route53 record name).
		if got := aws.ToString(cfg.FullyQualifiedDomainName); got != "edge.example.com" {
			t.Errorf("%s: FQDN = %q, want the steering hostname edge.example.com", ip, got)
		}
		if !strings.HasPrefix(aws.ToString(in.CallerReference), "carlos-steering-") {
			t.Errorf("%s: CallerReference = %q, want a carlos-steering- prefix", ip, aws.ToString(in.CallerReference))
		}
		if n := len(aws.ToString(in.CallerReference)); n > 64 {
			t.Errorf("%s: CallerReference is %d chars, Route53 caps it at 64", ip, n)
		}
	}

	// every created check carries the zone-scoped tag
	for id, tags := range f.tags {
		var found bool
		for _, tg := range tags {
			if aws.ToString(tg.Key) == steeringTagKey && aws.ToString(tg.Value) == testZone {
				found = true
			}
		}
		if !found {
			t.Errorf("check %s not tagged %s=%s", id, steeringTagKey, testZone)
		}
	}

	// record shape
	rr := find(t, f, "app.example.com.", "dublin")
	if rr.Type != r53types.RRTypeA {
		t.Errorf("type = %q, want A", rr.Type)
	}
	if aws.ToInt64(rr.TTL) != 60 {
		t.Errorf("TTL = %d, want 60", aws.ToInt64(rr.TTL))
	}
	if string(rr.Region) != "eu-west-1" {
		t.Errorf("Region = %q, want eu-west-1", rr.Region)
	}
	if rr.AliasTarget != nil {
		t.Errorf("AliasTarget set; these are plain latency records to EIPs")
	}
	if len(rr.ResourceRecords) != 1 || aws.ToString(rr.ResourceRecords[0].Value) != "1.2.3.4" {
		t.Errorf("value = %v, want 1.2.3.4", rr.ResourceRecords)
	}
	if aws.ToString(rr.HealthCheckId) == "" {
		t.Errorf("record not bound to a health check")
	}
	// the record for the same EIP under a different name shares the check
	edgeRR := find(t, f, "edge.example.com.", "dublin")
	if aws.ToString(edgeRR.HealthCheckId) != aws.ToString(rr.HealthCheckId) {
		t.Errorf("one check per EIP violated: %q vs %q", aws.ToString(edgeRR.HealthCheckId), aws.ToString(rr.HealthCheckId))
	}
}

// TestConvergeIdempotentZeroWrites is the brief's headline safety property:
// a second pass with an identical want makes NO write calls at all.
func TestConvergeIdempotentZeroWrites(t *testing.T) {
	f := newFake()
	want := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, want)

	before := f.writes()
	if before == 0 {
		t.Fatalf("first pass made no writes; the test proves nothing")
	}
	mustConverge(t, f, want)
	if got := f.writes() - before; got != 0 {
		t.Fatalf("second pass made %d write calls, want 0 (records=%d creates=%d deletes=%d tags=%d)",
			got, f.nChangeRecords, f.nCreateCheck, f.nDeleteCheck, f.nChangeTags)
	}
	// a third pass, for good measure — no slow drift
	mustConverge(t, f, want)
	if got := f.writes() - before; got != 0 {
		t.Fatalf("third pass made %d write calls, want 0", got)
	}
}

// TestConvergeAddEdge: a second edge arms. Only the new records are written;
// the existing ones are left alone.
func TestConvergeAddEdge(t *testing.T) {
	f := newFake()
	one := append([]desiredRecord{dublinRec}, hostRecs("app.example.com.", dublinRec)...)
	mustConverge(t, f, one)
	base := f.writes()
	f.submitted = nil

	two := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, two)

	if f.nCreateCheck != 2 {
		t.Fatalf("want exactly one NEW check (2 total), got %d total", f.nCreateCheck)
	}
	if f.nDeleteCheck != 0 {
		t.Fatalf("adding an edge deleted %d checks", f.nDeleteCheck)
	}
	if len(f.submitted) != 2 {
		t.Fatalf("want 2 changes (the new edge's records only), got %d: %v", len(f.submitted), f.submitted)
	}
	for _, c := range f.submitted {
		if c.Action != r53types.ChangeActionUpsert {
			t.Errorf("adding an edge submitted a %s", c.Action)
		}
		if aws.ToString(c.ResourceRecordSet.SetIdentifier) != "oregon" {
			t.Errorf("rewrote an untouched record: %s", rrKey(*c.ResourceRecordSet))
		}
	}
	if len(f.records) != 4 {
		t.Fatalf("want 4 records, got %v", recordNames(f.records))
	}
	_ = base
}

// TestConvergeRemoveEdge: an edge disarms. Its records go, and the health
// check it alone used is GC'd; the surviving edge's check is retained.
func TestConvergeRemoveEdge(t *testing.T) {
	f := newFake()
	two := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, two)
	oregonCheck := aws.ToString(find(t, f, "edge.example.com.", "oregon").HealthCheckId)
	dublinCheck := aws.ToString(find(t, f, "edge.example.com.", "dublin").HealthCheckId)
	f.submitted = nil

	one := append([]desiredRecord{dublinRec}, hostRecs("app.example.com.", dublinRec)...)
	mustConverge(t, f, one)

	if len(f.records) != 2 {
		t.Fatalf("want 2 records left, got %v", recordNames(f.records))
	}
	for _, rr := range f.records {
		if aws.ToString(rr.SetIdentifier) == "oregon" {
			t.Errorf("oregon record survived: %s", rrKey(rr))
		}
	}
	for _, c := range f.submitted {
		if c.Action != r53types.ChangeActionDelete {
			t.Errorf("removing an edge submitted a %s for %s", c.Action, rrKey(*c.ResourceRecordSet))
		}
	}
	if len(f.deletedChecks) != 1 || f.deletedChecks[0] != oregonCheck {
		t.Fatalf("want the orphaned check %s GC'd, got %v", oregonCheck, f.deletedChecks)
	}
	// the surviving check is untouched and still bound
	if got := aws.ToString(find(t, f, "app.example.com.", "dublin").HealthCheckId); got != dublinCheck {
		t.Errorf("dublin's check changed: %q -> %q", dublinCheck, got)
	}
}

// TestConvergeRemoveSteering: a host stops being steered. Its records go,
// but the checks stay — another host (and the steering hostname) still uses
// the same EIPs.
func TestConvergeRemoveSteering(t *testing.T) {
	f := newFake()
	both := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)
	both = append(both, hostRecs("other.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, both)
	if len(f.records) != 6 {
		t.Fatalf("setup: want 6 records, got %v", recordNames(f.records))
	}
	checksBefore := len(f.checks)
	f.submitted = nil

	one := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("other.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, one)

	if len(f.records) != 4 {
		t.Fatalf("want 4 records left, got %v", recordNames(f.records))
	}
	for _, rr := range f.records {
		if strings.EqualFold(aws.ToString(rr.Name), "app.example.com.") {
			t.Errorf("unsteered host's record survived: %s", rrKey(rr))
		}
	}
	if len(f.deletedChecks) != 0 {
		t.Fatalf("checks still in use were GC'd: %v", f.deletedChecks)
	}
	if len(f.checks) != checksBefore {
		t.Fatalf("check count changed %d -> %d", checksBefore, len(f.checks))
	}
}

// TestConvergeStaticRecordsUntouched is the deletion-safety test. The zone is
// seeded with everything an operator's zone really holds; converge must
// submit no change that touches any of it.
func TestConvergeStaticRecordsUntouched(t *testing.T) {
	f := newFake()
	f.records = []r53types.ResourceRecordSet{
		staticA("example.com.", "9.9.9.9"),     // apex A, no SetIdentifier
		staticA("www.example.com.", "9.9.9.9"), // plain A
		staticA("app.example.com.", "9.9.9.9"), // plain A sharing a STEERED name
		{Name: aws.String("example.com."), Type: r53types.RRTypeNs, TTL: aws.Int64(172800)},
		{Name: aws.String("example.com."), Type: r53types.RRTypeSoa, TTL: aws.Int64(900)},
		{Name: aws.String("mail.example.com."), Type: r53types.RRTypeCname, TTL: aws.Int64(300)},
		{ // a WEIGHTED A with a SetIdentifier — has an identifier but is not latency
			Name: aws.String("weighted.example.com."), Type: r53types.RRTypeA,
			SetIdentifier: aws.String("dublin"), Weight: aws.Int64(10), TTL: aws.Int64(60),
			ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("8.8.8.8")}},
		},
		{ // a LATENCY A bound to a health check that is not ours
			Name: aws.String("legacy.example.com."), Type: r53types.RRTypeA,
			SetIdentifier: aws.String("dublin"), Region: "eu-west-1", TTL: aws.Int64(60),
			HealthCheckId:   aws.String("hc-someone-else"),
			ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("7.7.7.7")}},
		},
		{ // a LATENCY ALIAS record — never a shape this converger writes
			Name: aws.String("alias.example.com."), Type: r53types.RRTypeA,
			SetIdentifier: aws.String("dublin"), Region: "eu-west-1",
			AliasTarget: &r53types.AliasTarget{DNSName: aws.String("elb.example.com.")},
		},
	}
	before := recordNames(f.records)

	buf := captureLog(t)
	want := append([]desiredRecord{dublinRec}, hostRecs("app.example.com.", dublinRec)...)
	mustConverge(t, f, want)

	for _, c := range f.submitted {
		if c.Action == r53types.ChangeActionDelete {
			t.Errorf("submitted a DELETE for %s — static/foreign records are out of bounds", rrKey(*c.ResourceRecordSet))
		}
		if aws.ToString(c.ResourceRecordSet.SetIdentifier) == "" {
			t.Errorf("submitted a change for a record with no SetIdentifier: %s", rrKey(*c.ResourceRecordSet))
		}
	}
	for _, name := range before {
		var still bool
		for _, have := range recordNames(f.records) {
			if have == name {
				still = true
			}
		}
		if !still {
			t.Errorf("record %s disappeared", name)
		}
	}

	// app.example.com. already holds a PLAIN A record. Route53 forbids a
	// name from holding both a simple record and a latency record set, and
	// change batches are atomic — so writing a latency record there would
	// not merely fail for that name, it would take every other steering
	// record in the zone down with it, on every tick, forever. The name is
	// skipped entirely: no upsert, no delete, and loudly.
	for _, c := range f.submitted {
		if strings.EqualFold(aws.ToString(c.ResourceRecordSet.Name), "app.example.com.") {
			t.Errorf("submitted a change for the plain-record-conflicted name: %s %s", c.Action, rrKey(*c.ResourceRecordSet))
		}
	}
	if !strings.Contains(buf.String(), "SKIPPING app.example.com.") {
		t.Errorf("the conflict was not named in the log; log = %q", buf.String())
	}
	if !strings.Contains(buf.String(), "9.9.9.9") {
		t.Errorf("the log did not name the conflicting record's value; log = %q", buf.String())
	}
	plain := find(t, f, "app.example.com.", "")
	if aws.ToInt64(plain.TTL) != 300 || aws.ToString(plain.ResourceRecords[0].Value) != "9.9.9.9" {
		t.Errorf("the operator's app.example.com. A record was changed: TTL %d value %v", aws.ToInt64(plain.TTL), plain.ResourceRecords)
	}

	// ...and the OTHER name's record still lands. One conflicting name must
	// not wedge the whole zone.
	edgeRR := find(t, f, "edge.example.com.", "dublin")
	if aws.ToString(edgeRR.ResourceRecords[0].Value) != "1.2.3.4" {
		t.Errorf("the unconflicted name did not converge: %v", edgeRR.ResourceRecords)
	}
	if f.nChangeRecords != 1 {
		t.Errorf("want 1 change batch (the unconflicted name), got %d", f.nChangeRecords)
	}
}

// TestFakeRejectsConflictingBatches proves the fake's refusals are real, so
// the skips are fixes and not tautologies: submitting the batch the converger
// used to submit does fail, atomically, taking an unrelated and perfectly
// legal change in the same batch down with it. One row per conflict shape.
func TestFakeRejectsConflictingBatches(t *testing.T) {
	tests := []struct {
		name    string
		seeded  r53types.ResourceRecordSet
		wantErr string
	}{
		{
			name:    "plain A",
			seeded:  staticA("app.example.com.", "9.9.9.9"),
			wantErr: "cannot be both a simple record and a routing-policy record set",
		},
		{
			name: "CNAME",
			seeded: r53types.ResourceRecordSet{
				Name: aws.String("app.example.com."), Type: r53types.RRTypeCname, TTL: aws.Int64(300),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("elsewhere.example.net.")}},
			},
			wantErr: "cannot coexist with any other record type",
		},
		{
			name: "weighted A set",
			seeded: r53types.ResourceRecordSet{
				Name: aws.String("app.example.com."), Type: r53types.RRTypeA,
				SetIdentifier: aws.String("legacy"), Weight: aws.Int64(10), TTL: aws.Int64(60),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("8.8.8.8")}},
			},
			wantErr: "mixes routing policies",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			f := newFake()
			f.records = []r53types.ResourceRecordSet{tt.seeded}

			latency := r53types.ResourceRecordSet{
				Name: aws.String("app.example.com."), Type: r53types.RRTypeA,
				SetIdentifier: aws.String("dublin"), Region: "eu-west-1", TTL: aws.Int64(60),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("1.2.3.4")}},
			}
			innocent := r53types.ResourceRecordSet{
				Name: aws.String("edge.example.com."), Type: r53types.RRTypeA,
				SetIdentifier: aws.String("dublin"), Region: "eu-west-1", TTL: aws.Int64(60),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("1.2.3.4")}},
			}
			_, err := f.ChangeResourceRecordSets(context.Background(), &route53.ChangeResourceRecordSetsInput{
				HostedZoneId: aws.String(testZone),
				ChangeBatch: &r53types.ChangeBatch{Changes: []r53types.Change{
					{Action: r53types.ChangeActionUpsert, ResourceRecordSet: &innocent},
					{Action: r53types.ChangeActionUpsert, ResourceRecordSet: &latency},
				}},
			})
			if err == nil {
				t.Fatalf("the fake accepted the batch — it is more permissive than Route53")
			}
			if !strings.Contains(err.Error(), "InvalidChangeBatch") || !strings.Contains(err.Error(), tt.wantErr) {
				t.Errorf("wrong error: %v", err)
			}
			// atomic: the innocent change did NOT land either
			if len(f.records) != 1 {
				t.Fatalf("the batch was not atomic: %v", recordNames(f.records))
			}
		})
	}
}

// TestConvergeSkipsConflictingNames is FINDING A: the plain-A wedge has two
// siblings with the identical failure mode. Each must be skipped by name, and
// each must leave the OTHER steered names converging normally.
func TestConvergeSkipsConflictingNames(t *testing.T) {
	tests := []struct {
		name       string
		seeded     r53types.ResourceRecordSet
		wantLogged string
	}{
		{
			name: "a CNAME at a steered name",
			seeded: r53types.ResourceRecordSet{
				Name: aws.String("app.example.com."), Type: r53types.RRTypeCname, TTL: aws.Int64(300),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("elsewhere.example.net.")}},
			},
			wantLogged: "a name holding a CNAME may hold no other record type",
		},
		{
			name: "a weighted A set at a steered name",
			seeded: r53types.ResourceRecordSet{
				Name: aws.String("app.example.com."), Type: r53types.RRTypeA,
				SetIdentifier: aws.String("legacy"), Weight: aws.Int64(10), TTL: aws.Int64(60),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("8.8.8.8")}},
			},
			wantLogged: "only one routing policy per name and type",
		},
		{
			name: "a failover A set at a steered name",
			seeded: r53types.ResourceRecordSet{
				Name: aws.String("app.example.com."), Type: r53types.RRTypeA,
				SetIdentifier: aws.String("primary"), Failover: r53types.ResourceRecordSetFailoverPrimary,
				TTL:             aws.Int64(60),
				ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("8.8.8.8")}},
			},
			wantLogged: "only one routing policy per name and type",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			f := newFake()
			f.records = []r53types.ResourceRecordSet{tt.seeded}
			seededKey := rrKey(tt.seeded)

			buf := captureLog(t)
			want := append([]desiredRecord{dublinRec}, hostRecs("app.example.com.", dublinRec)...)
			mustConverge(t, f, want)

			// the conflicted name is skipped ENTIRELY — no upsert, no delete
			for _, c := range f.submitted {
				if strings.EqualFold(aws.ToString(c.ResourceRecordSet.Name), "app.example.com.") {
					t.Errorf("submitted a change for the conflicted name: %s %s", c.Action, rrKey(*c.ResourceRecordSet))
				}
			}
			// the operator's record is untouched
			var found bool
			for _, rr := range f.records {
				if rrKey(rr) == seededKey {
					found = true
				}
			}
			if !found {
				t.Errorf("the operator's record disappeared; zone = %v", recordNames(f.records))
			}
			// the log names the conflict and why
			logged := buf.String()
			if !strings.Contains(logged, "SKIPPING app.example.com.") {
				t.Errorf("the conflict was not named; log = %q", logged)
			}
			if !strings.Contains(logged, tt.wantLogged) {
				t.Errorf("the log did not explain the conflict (%q); log = %q", tt.wantLogged, logged)
			}
			// ...and the OTHER name still converges in the same pass
			edgeRR := find(t, f, "edge.example.com.", "dublin")
			if aws.ToString(edgeRR.ResourceRecords[0].Value) != "1.2.3.4" {
				t.Errorf("the unconflicted name did not converge: %v", edgeRR.ResourceRecords)
			}
		})
	}
}

// TestConvergeSNIDowngradeIsLogged: when the synthetic edge.<apps-domain>
// name is itself dropped for a conflict, every health check created
// afterwards silently falls back to a member hostname for SNI. That
// downgrade must be visible.
func TestConvergeSNIDowngradeIsLogged(t *testing.T) {
	f := newFake()
	f.records = []r53types.ResourceRecordSet{staticA("edge.example.com.", "9.9.9.9")}

	buf := captureLog(t)
	want := append([]desiredRecord{dublinRec}, hostRecs("app.example.com.", dublinRec)...)
	mustConverge(t, f, want)

	if len(f.createdChecks) != 1 {
		t.Fatalf("want 1 check, got %d", len(f.createdChecks))
	}
	if got := aws.ToString(f.createdChecks[0].HealthCheckConfig.FullyQualifiedDomainName); got != "app.example.com" {
		t.Fatalf("SNI = %q, want the member hostname fallback", got)
	}
	logged := buf.String()
	if !strings.Contains(logged, "fall back to a member hostname") {
		t.Errorf("the SNI downgrade was silent; log = %q", logged)
	}
	if !strings.Contains(logged, "SKIPPING edge.example.com.") {
		t.Errorf("the cause was not logged alongside it; log = %q", logged)
	}
}

// TestConvergeShrinkGuard is FINDING 1: the empty-desired-set refusal only
// catches a read that returned NOTHING. A SHORT read has a non-empty want and
// would delete every record for the members that went missing — NXDOMAIN for
// a live steered host. A pass that would remove most of what it manages must
// refuse and describe itself instead.
func TestConvergeShrinkGuard(t *testing.T) {
	// 2 names x 2 edges = 4 managed records.
	full := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)

	t.Run("deleting 3 of 4 refuses with zero writes and names them", func(t *testing.T) {
		f := newFake()
		mustConverge(t, f, full)
		before := f.writes()
		state := recordNames(f.records)

		buf := captureLog(t)
		// A short read that saw only one edge and only the synthetic name.
		mustConverge(t, f, []desiredRecord{dublinRec})

		if got := f.writes() - before; got != 0 {
			t.Fatalf("the refused pass made %d write calls, want 0", got)
		}
		if len(f.deletedChecks) != 0 {
			t.Errorf("the refused pass GC'd checks: %v", f.deletedChecks)
		}
		if got := recordNames(f.records); !reflect.DeepEqual(got, state) {
			t.Fatalf("the zone changed: %v -> %v", state, got)
		}
		logged := buf.String()
		if !strings.Contains(logged, "refusing to converge") {
			t.Errorf("the refusal was not logged; log = %q", logged)
		}
		// all three intended deletions named by (name, setid)
		for _, want := range []string{
			"would delete edge.example.com. (set oregon",
			"would delete app.example.com. (set dublin",
			"would delete app.example.com. (set oregon",
		} {
			if !strings.Contains(logged, want) {
				t.Errorf("log does not name %q; log = %q", want, logged)
			}
		}
	})

	t.Run("deleting 1 of 4 proceeds", func(t *testing.T) {
		f := newFake()
		mustConverge(t, f, full)

		// drop exactly one record: app.example.com. on oregon
		var want []desiredRecord
		for _, rec := range full {
			if rec.Name == "app.example.com." && rec.Box == "oregon" {
				continue
			}
			want = append(want, rec)
		}
		mustConverge(t, f, want)

		if len(f.records) != 3 {
			t.Fatalf("want 3 records left, got %v", recordNames(f.records))
		}
		for _, rr := range f.records {
			if strings.EqualFold(aws.ToString(rr.Name), "app.example.com.") && aws.ToString(rr.SetIdentifier) == "oregon" {
				t.Errorf("the ordinary deletion did not happen")
			}
		}
	})

	t.Run("the floor lets a small zone shrink", func(t *testing.T) {
		// 2 managed records; half of 2 is 1, but the floor allows 2.
		f := newFake()
		two := []desiredRecord{dublinRec, oregonRec}
		mustConverge(t, f, two)
		mustConverge(t, f, append(hostRecs("app.example.com.", dublinRec), dublinRec))
		if len(f.records) != 2 {
			t.Fatalf("want 2 records, got %v", recordNames(f.records))
		}
	})
}

// TestShrinkLimit pins the threshold arithmetic on its own — the zone-level
// test above can only exercise the sizes it happens to build.
func TestShrinkLimit(t *testing.T) {
	tests := []struct{ managed, want int }{
		{0, shrinkFloor}, {1, shrinkFloor}, {2, shrinkFloor}, {4, shrinkFloor},
		{6, 3}, {10, 5}, {11, 5}, {100, 50},
	}
	for _, tt := range tests {
		if got := shrinkLimit(tt.managed); got != tt.want {
			t.Errorf("shrinkLimit(%d) = %d, want %d", tt.managed, got, tt.want)
		}
	}
}

// TestConvergeHealthCheckGCIsTagScoped: a check must carry OUR tag, with OUR
// zone id, before it can ever be deleted. Everything else in the account is
// invisible to the GC.
func TestConvergeHealthCheckGCIsTagScoped(t *testing.T) {
	f := newFake()
	// three checks on an EIP nobody wants any more
	f.checks = []r53types.HealthCheck{
		{Id: aws.String("hc-untagged"), HealthCheckConfig: &r53types.HealthCheckConfig{
			Type: r53types.HealthCheckTypeHttps, Port: aws.Int32(443), IPAddress: aws.String("10.0.0.1")}},
		{Id: aws.String("hc-other-zone"), HealthCheckConfig: &r53types.HealthCheckConfig{
			Type: r53types.HealthCheckTypeHttps, Port: aws.Int32(443), IPAddress: aws.String("10.0.0.2")}},
		{Id: aws.String("hc-ours"), HealthCheckConfig: &r53types.HealthCheckConfig{
			Type: r53types.HealthCheckTypeHttps, Port: aws.Int32(443), IPAddress: aws.String("10.0.0.3")}},
		{Id: aws.String("hc-other-service"), HealthCheckConfig: &r53types.HealthCheckConfig{
			Type: r53types.HealthCheckTypeHttp, Port: aws.Int32(80), IPAddress: aws.String("10.0.0.4")}},
	}
	f.tags["hc-other-zone"] = []r53types.Tag{{Key: aws.String(steeringTagKey), Value: aws.String("ZSOMEOTHERZONE")}}
	f.tags["hc-ours"] = []r53types.Tag{{Key: aws.String(steeringTagKey), Value: aws.String(testZone)}}
	f.tags["hc-other-service"] = []r53types.Tag{{Key: aws.String("Name"), Value: aws.String("someone else's check")}}

	mustConverge(t, f, []desiredRecord{dublinRec})

	if len(f.deletedChecks) != 1 || f.deletedChecks[0] != "hc-ours" {
		t.Fatalf("GC deleted %v; only the tagged-for-THIS-zone check may ever go", f.deletedChecks)
	}
}

// TestConvergeEmptyWantRefusesTeardown: an empty desired set is a degraded
// read (zero armed edges), not an instruction to delete production DNS.
func TestConvergeEmptyWantRefusesTeardown(t *testing.T) {
	f := newFake()
	want := append([]desiredRecord{dublinRec}, hostRecs("app.example.com.", dublinRec)...)
	mustConverge(t, f, want)
	before := f.writes()
	recs := len(f.records)

	buf := captureLog(t)
	if err := converge(context.Background(), f, testZone, nil); err != nil {
		t.Fatalf("empty want should not error: %v", err)
	}
	if f.writes() != before {
		t.Fatalf("empty want made %d write calls", f.writes()-before)
	}
	if len(f.records) != recs {
		t.Fatalf("empty want removed records: %v", recordNames(f.records))
	}
	if len(f.deletedChecks) != 0 {
		t.Fatalf("empty want GC'd checks: %v", f.deletedChecks)
	}
	if !strings.Contains(buf.String(), "refusing") {
		t.Errorf("the refusal was silent; log = %q", buf.String())
	}
}

// TestConvergeRejectsBadZoneID: a malformed or empty zone id never reaches
// the API — a typo must not become a call against some other zone.
func TestConvergeRejectsBadZoneID(t *testing.T) {
	for _, zone := range []string{"", "  ", "z0abc", "/hostedzone/Z0ABC", "Z0abc", "NOTAZONE"} {
		f := newFake()
		err := converge(context.Background(), f, zone, []desiredRecord{dublinRec})
		if err == nil {
			t.Errorf("zone %q accepted", zone)
		}
		if f.nListRecords+f.nListChecks+f.writes() != 0 {
			t.Errorf("zone %q reached the API", zone)
		}
	}
}

// TestConvergeHealthCheckFQDNFallback: with no edge.<apps-domain> record in
// want, the check's SNI name falls back to a record's own name.
func TestConvergeHealthCheckFQDNFallback(t *testing.T) {
	f := newFake()
	mustConverge(t, f, hostRecs("app.example.com.", dublinRec))

	if len(f.createdChecks) != 1 {
		t.Fatalf("want 1 check, got %d", len(f.createdChecks))
	}
	if got := aws.ToString(f.createdChecks[0].HealthCheckConfig.FullyQualifiedDomainName); got != "app.example.com" {
		t.Fatalf("FQDN = %q, want the record's own name app.example.com", got)
	}
}

// TestConvergePagination: both listings paginate, and a converger that
// stopped at page one would see an empty zone and rewrite everything.
func TestConvergePagination(t *testing.T) {
	f := newFake()
	want := append([]desiredRecord{dublinRec, oregonRec}, hostRecs("app.example.com.", dublinRec, oregonRec)...)
	want = append(want, hostRecs("b.example.com.", dublinRec, oregonRec)...)
	want = append(want, hostRecs("c.example.com.", dublinRec, oregonRec)...)
	mustConverge(t, f, want)

	f.pageSize = 1 // force many pages on the second pass
	before := f.writes()
	mustConverge(t, f, want)
	if got := f.writes() - before; got != 0 {
		t.Fatalf("paginated pass made %d writes, want 0 — a page was missed", got)
	}
	if f.nListRecords < 3 || f.nListChecks < 2 {
		t.Fatalf("pagination not exercised: %d record pages, %d check pages", f.nListRecords, f.nListChecks)
	}
}

// TestConvergeTagFailureRemovesTheOrphan: a check created but not tagged can
// never be reused (untagged = not ours) and never GC'd — an unbounded leak,
// one per pass. It is deleted again before the pass fails.
func TestConvergeTagFailureRemovesTheOrphan(t *testing.T) {
	f := newFake()
	f.failTags = errors.New("boom")
	err := converge(context.Background(), f, testZone, []desiredRecord{dublinRec})
	if err == nil {
		t.Fatalf("tagging failure did not fail the pass")
	}
	if len(f.deletedChecks) != 1 {
		t.Fatalf("the untaggable check was leaked: created %d, deleted %v", f.nCreateCheck, f.deletedChecks)
	}
	if f.nChangeRecords != 0 {
		t.Errorf("records were written despite the failed pass")
	}
}

// TestConvergeChangeFailureIsLoud: a failed batch fails the pass. The
// housekeeper retries next tick; it must not be swallowed.
func TestConvergeChangeFailureIsLoud(t *testing.T) {
	f := newFake()
	f.failChange = errors.New("InvalidChangeBatch")
	err := converge(context.Background(), f, testZone, []desiredRecord{dublinRec})
	if err == nil {
		t.Fatalf("a failed change batch did not fail the pass")
	}
	if !strings.Contains(err.Error(), "InvalidChangeBatch") {
		t.Errorf("the underlying error was not wrapped: %v", err)
	}
	// nothing may be GC'd on a pass whose record changes did not land
	if len(f.deletedChecks) != 0 {
		t.Errorf("checks GC'd after a failed record batch: %v", f.deletedChecks)
	}
}

// TestConvergeBatchesChanges: a large desired set is chunked, and every
// chunk is within the batch limit.
func TestConvergeBatchesChanges(t *testing.T) {
	f := newFake()
	var want []desiredRecord
	for i := 0; i < changeBatchLimit+5; i++ {
		want = append(want, desiredRecord{
			Name: fmt.Sprintf("h%03d.example.com.", i), EIP: "1.2.3.4", Box: "dublin", Region: "eu-west-1",
		})
	}
	mustConverge(t, f, want)
	if f.nChangeRecords != 2 {
		t.Fatalf("want 2 batches for %d changes, got %d", len(want), f.nChangeRecords)
	}
	if len(f.records) != len(want) {
		t.Fatalf("want %d records, got %d", len(want), len(f.records))
	}
}

// TestConvergeAdoptsForeignLatencyRecordUnderOurKey: a latency record already
// sitting at a (name, SetIdentifier) we want is taken over by UPSERT — it is
// exactly the record we would otherwise create, and CREATE would collide.
func TestConvergeAdoptsForeignLatencyRecordUnderOurKey(t *testing.T) {
	f := newFake()
	f.records = []r53types.ResourceRecordSet{{
		Name: aws.String("edge.example.com."), Type: r53types.RRTypeA,
		SetIdentifier: aws.String("dublin"), Region: "eu-west-1", TTL: aws.Int64(300),
		ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("99.99.99.99")}},
	}}
	buf := captureLog(t)
	mustConverge(t, f, []desiredRecord{dublinRec})

	rr := find(t, f, "edge.example.com.", "dublin")
	if aws.ToString(rr.ResourceRecords[0].Value) != "1.2.3.4" || aws.ToInt64(rr.TTL) != 60 {
		t.Fatalf("record not converged: %v TTL=%d", rr.ResourceRecords, aws.ToInt64(rr.TTL))
	}
	if !strings.Contains(buf.String(), "adopting") {
		t.Errorf("the takeover was silent; log = %q", buf.String())
	}
	// ...and it must NOT also claim to be leaving that same record alone.
	// The two lines describe opposite fates for one record; only one of them
	// is true. (Regression: the "leaving alone" log lost its wanted-key
	// guard when the loop was split out of recordChanges in review round 1.)
	if strings.Contains(buf.String(), "leaving latency record") {
		t.Errorf("logged BOTH adoption and 'leaving alone' for the same record; log = %q", buf.String())
	}
}

// TestConvergeLeavingAloneLogNamesOnlyUnwantedKeys: the "leaving alone" line
// is still owed for a foreign latency record this pass genuinely does not
// touch — the FINDING B fix must not silence it everywhere.
func TestConvergeLeavingAloneLogNamesOnlyUnwantedKeys(t *testing.T) {
	f := newFake()
	f.records = []r53types.ResourceRecordSet{
		{ // a foreign latency record at a name/identifier we do NOT want
			Name: aws.String("legacy.example.com."), Type: r53types.RRTypeA,
			SetIdentifier: aws.String("dublin"), Region: "eu-west-1", TTL: aws.Int64(60),
			HealthCheckId:   aws.String("hc-someone-else"),
			ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("7.7.7.7")}},
		},
	}
	buf := captureLog(t)
	mustConverge(t, f, []desiredRecord{dublinRec})

	if !strings.Contains(buf.String(), "leaving latency record legacy.example.com.") {
		t.Errorf("the untouched foreign record was not reported; log = %q", buf.String())
	}
	if len(f.records) != 2 {
		t.Errorf("the foreign record was touched; zone = %v", recordNames(f.records))
	}
}

// TestIsManagedGates pins the deletion boundary GATE BY GATE.
//
// The zone-level test above (TestConvergeStaticRecordsUntouched) proves the
// boundary holds for the records a real zone contains, but its fixtures fail
// several gates at once — so it cannot tell which gate is doing the work, and
// stays green if one of them is deleted. This table starts from a record that
// passes EVERY gate and breaks exactly one at a time. Each row must flip the
// verdict to unmanaged on its own.
func TestIsManagedGates(t *testing.T) {
	owned := map[string]bool{"hc-ours": true}
	managed := func() r53types.ResourceRecordSet {
		return r53types.ResourceRecordSet{
			Name: aws.String("app.example.com."), Type: r53types.RRTypeA,
			SetIdentifier: aws.String("dublin"), Region: "eu-west-1", TTL: aws.Int64(60),
			HealthCheckId:   aws.String("hc-ours"),
			ResourceRecords: []r53types.ResourceRecord{{Value: aws.String("1.2.3.4")}},
		}
	}
	if !isManaged(managed(), owned) {
		t.Fatalf("the baseline record is not managed; the table below proves nothing")
	}

	tests := []struct {
		name   string
		break_ func(*r53types.ResourceRecordSet)
	}{
		{"not an A record", func(rr *r53types.ResourceRecordSet) { rr.Type = r53types.RRTypeAaaa }},
		{"no SetIdentifier (static/operator territory)", func(rr *r53types.ResourceRecordSet) { rr.SetIdentifier = nil }},
		{"empty SetIdentifier", func(rr *r53types.ResourceRecordSet) { rr.SetIdentifier = aws.String("") }},
		{"not a latency record", func(rr *r53types.ResourceRecordSet) { rr.Region = "" }},
		{"an alias record", func(rr *r53types.ResourceRecordSet) {
			rr.AliasTarget = &r53types.AliasTarget{DNSName: aws.String("elb.example.com.")}
		}},
		{"health check we do not own", func(rr *r53types.ResourceRecordSet) { rr.HealthCheckId = aws.String("hc-theirs") }},
		{"no health check at all", func(rr *r53types.ResourceRecordSet) { rr.HealthCheckId = nil }},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			rr := managed()
			tt.break_(&rr)
			if isManaged(rr, owned) {
				t.Fatalf("record is still managed — this gate does not stand on its own")
			}
		})
	}

	// and the empty-ownership case: if the tag listing found nothing ours,
	// NOTHING in the zone is deletable.
	if isManaged(managed(), map[string]bool{}) {
		t.Errorf("a record is managed with no owned health checks at all")
	}
}

// TestSteeringZoneIDFromEnv pins the dormancy contract Task 6 consumes.
func TestSteeringZoneIDFromEnv(t *testing.T) {
	tests := []struct {
		set  string
		want string
	}{
		{"", ""},
		{"  ", ""},
		{"Z0STEERINGTEST", "Z0STEERINGTEST"},
		{"  Z0STEERINGTEST  ", "Z0STEERINGTEST"},
		{"/hostedzone/Z0STEERINGTEST", ""},
		{"z0steeringtest", ""},
	}
	for _, tt := range tests {
		t.Setenv(steeringZoneEnv, tt.set)
		if got := SteeringZoneID(); got != tt.want {
			t.Errorf("SteeringZoneID() with %q = %q, want %q", tt.set, got, tt.want)
		}
	}
}