package steering

// route53.go — the converger. This is the ONLY file in the tree that holds
// write credentials to a production DNS zone, so its whole design is about
// what it REFUSES to touch. Read the managed-set rule (isManaged) and the
// health-check ownership rule (ownedHealthChecks) before changing anything
// here: everything else is bookkeeping around those two boundaries.
//
// converge is UNEXPORTED: ConvergeDeployment (deployment.go) is this
// package's only entry point, so no caller outside it can converge a desired
// set the fold did not build. It is reached, from Task 6's housekeeper duty
// (internal/host/steering_loop.go) — but only on a deployment that sets
// CARLOS_STEERING_ZONE_ID, which no box does today: unset keeps the duty
// dormant, zero API calls, zero writes.

import (
	"context"
	"fmt"
	"log"
	"os"
	"regexp"
	"sort"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/route53"
	r53types "github.com/aws/aws-sdk-go-v2/service/route53/types"
	"github.com/carlosframework/platform/internal/rootaws"
)

const (
	// steeringZoneEnv names the deployment's steering hosted zone. Unset =
	// dormant, which is every box today.
	steeringZoneEnv = "CARLOS_STEERING_ZONE_ID"

	// steeringTagKey / its value (the zone id) is the ONLY thing that makes
	// a Route53 health check eligible for deletion by this code. An untagged
	// check, or one tagged for another deployment's zone, is invisible to
	// the GC — see ownedHealthChecks.
	steeringTagKey = "carlos-steering"

	// recordTTL is short on purpose: the health check withdraws a dead
	// edge's answer in ~90s, and a long TTL would keep resolvers handing it
	// out for far longer than that.
	recordTTL = 60

	// health check shape. HTTPS on 443 with SNI: the name in
	// FullyQualifiedDomainName is what the edge sees in the handshake, so it
	// must be a name the edge holds a certificate for.
	healthCheckPort            = 443
	healthCheckPath            = "/"
	healthCheckInterval        = 30 // seconds
	healthCheckFailureThresh   = 3  // 3 x 30s ~= the spec's 90s withdrawal
	healthCheckTagBatch        = 10 // ListTagsForResources' own cap
	healthCheckCallerRefPrefix = "carlos-steering-"

	// changeBatchLimit chunks ChangeResourceRecordSets well under Route53's
	// own 1000-change cap. Smaller batches mean a failure retries less work
	// and names a smaller blast radius in the log.
	changeBatchLimit = 100

	// shrinkFloor is the number of managed records a pass may always remove,
	// however small the zone. Without it, the proportional guard below would
	// make a 2-record zone unable to lose a single record. See shrinkLimit.
	shrinkFloor = 2
)

// steeringZonePattern is the region-host tofu module's own zone-id shape
// (infra/modules/region-host/variables.tf: `^Z[A-Z0-9]+$`) — a BARE id, never
// the "/hostedzone/Z..." ARN-ish form the API returns in some places. A typo
// that slips this gate would be a write against some other zone, so the gate
// runs before any client call, not after.
var steeringZonePattern = regexp.MustCompile(`^Z[A-Z0-9]+$`)

// route53API is the slice of the Route53 client this converger uses — an
// interface so tests substitute an in-memory fake and never reach AWS
// (route53_test.go). Same seam as internal/edge/route53dns.go's own
// route53API.
type route53API interface {
	ListResourceRecordSets(ctx context.Context, in *route53.ListResourceRecordSetsInput, optFns ...func(*route53.Options)) (*route53.ListResourceRecordSetsOutput, error)
	ChangeResourceRecordSets(ctx context.Context, in *route53.ChangeResourceRecordSetsInput, optFns ...func(*route53.Options)) (*route53.ChangeResourceRecordSetsOutput, error)
	ListHealthChecks(ctx context.Context, in *route53.ListHealthChecksInput, optFns ...func(*route53.Options)) (*route53.ListHealthChecksOutput, error)
	ListTagsForResources(ctx context.Context, in *route53.ListTagsForResourcesInput, optFns ...func(*route53.Options)) (*route53.ListTagsForResourcesOutput, error)
	CreateHealthCheck(ctx context.Context, in *route53.CreateHealthCheckInput, optFns ...func(*route53.Options)) (*route53.CreateHealthCheckOutput, error)
	DeleteHealthCheck(ctx context.Context, in *route53.DeleteHealthCheckInput, optFns ...func(*route53.Options)) (*route53.DeleteHealthCheckOutput, error)
	ChangeTagsForResource(ctx context.Context, in *route53.ChangeTagsForResourceInput, optFns ...func(*route53.Options)) (*route53.ChangeTagsForResourceOutput, error)
}

// route53Methods satisfies route53API from method values of one route53 client.
// Storing the *route53.Client itself in the interface would make the linker keep every exported Route53 method once html/template is linked in (discussion #275); method values keep only the calls made here.
type route53Methods struct {
	listResourceRecordSets   func(ctx context.Context, in *route53.ListResourceRecordSetsInput, optFns ...func(*route53.Options)) (*route53.ListResourceRecordSetsOutput, error)
	changeResourceRecordSets func(ctx context.Context, in *route53.ChangeResourceRecordSetsInput, optFns ...func(*route53.Options)) (*route53.ChangeResourceRecordSetsOutput, error)
	listHealthChecks         func(ctx context.Context, in *route53.ListHealthChecksInput, optFns ...func(*route53.Options)) (*route53.ListHealthChecksOutput, error)
	listTagsForResources     func(ctx context.Context, in *route53.ListTagsForResourcesInput, optFns ...func(*route53.Options)) (*route53.ListTagsForResourcesOutput, error)
	createHealthCheck        func(ctx context.Context, in *route53.CreateHealthCheckInput, optFns ...func(*route53.Options)) (*route53.CreateHealthCheckOutput, error)
	deleteHealthCheck        func(ctx context.Context, in *route53.DeleteHealthCheckInput, optFns ...func(*route53.Options)) (*route53.DeleteHealthCheckOutput, error)
	changeTagsForResource    func(ctx context.Context, in *route53.ChangeTagsForResourceInput, optFns ...func(*route53.Options)) (*route53.ChangeTagsForResourceOutput, error)
}

func (m route53Methods) ListResourceRecordSets(ctx context.Context, in *route53.ListResourceRecordSetsInput, opts ...func(*route53.Options)) (*route53.ListResourceRecordSetsOutput, error) {
	return m.listResourceRecordSets(ctx, in, opts...)
}

func (m route53Methods) ChangeResourceRecordSets(ctx context.Context, in *route53.ChangeResourceRecordSetsInput, opts ...func(*route53.Options)) (*route53.ChangeResourceRecordSetsOutput, error) {
	return m.changeResourceRecordSets(ctx, in, opts...)
}

func (m route53Methods) ListHealthChecks(ctx context.Context, in *route53.ListHealthChecksInput, opts ...func(*route53.Options)) (*route53.ListHealthChecksOutput, error) {
	return m.listHealthChecks(ctx, in, opts...)
}

func (m route53Methods) ListTagsForResources(ctx context.Context, in *route53.ListTagsForResourcesInput, opts ...func(*route53.Options)) (*route53.ListTagsForResourcesOutput, error) {
	return m.listTagsForResources(ctx, in, opts...)
}

func (m route53Methods) CreateHealthCheck(ctx context.Context, in *route53.CreateHealthCheckInput, opts ...func(*route53.Options)) (*route53.CreateHealthCheckOutput, error) {
	return m.createHealthCheck(ctx, in, opts...)
}

func (m route53Methods) DeleteHealthCheck(ctx context.Context, in *route53.DeleteHealthCheckInput, opts ...func(*route53.Options)) (*route53.DeleteHealthCheckOutput, error) {
	return m.deleteHealthCheck(ctx, in, opts...)
}

func (m route53Methods) ChangeTagsForResource(ctx context.Context, in *route53.ChangeTagsForResourceInput, opts ...func(*route53.Options)) (*route53.ChangeTagsForResourceOutput, error) {
	return m.changeTagsForResource(ctx, in, opts...)
}

func newRoute53Methods(c *route53.Client) route53Methods {
	return route53Methods{
		listResourceRecordSets:   c.ListResourceRecordSets,
		changeResourceRecordSets: c.ChangeResourceRecordSets,
		listHealthChecks:         c.ListHealthChecks,
		listTagsForResources:     c.ListTagsForResources,
		createHealthCheck:        c.CreateHealthCheck,
		deleteHealthCheck:        c.DeleteHealthCheck,
		changeTagsForResource:    c.ChangeTagsForResource,
	}
}

// SteeringZoneID returns the deployment's validated steering hosted-zone id,
// or "" when the deployment is dormant (env unset, blank, or malformed —
// malformed is logged, because a typo'd zone id is a configuration error
// somebody needs to see, not a quiet no-op).
func SteeringZoneID() string {
	raw := strings.TrimSpace(os.Getenv(steeringZoneEnv))
	if raw == "" {
		return ""
	}
	if !steeringZonePattern.MatchString(raw) {
		log.Printf("steering: %s=%q is not a bare hosted-zone id (want e.g. Z0ABCDEF1234) — staying dormant", steeringZoneEnv, raw)
		return ""
	}
	return raw
}

// NewClient builds a Route53 client from the default AWS credential chain —
// the box instance profile in production, exactly like
// internal/edge/route53dns.go's newRoute53DNS. Route53 is a global service
// but the SDK still needs a signing region, so default to us-east-1 when the
// environment names none; an explicit AWS_REGION still wins.
func NewClient(ctx context.Context) (route53API, error) {
	var opts []func(*awsconfig.LoadOptions) error
	if strings.TrimSpace(os.Getenv("AWS_REGION")) == "" && strings.TrimSpace(os.Getenv("AWS_DEFAULT_REGION")) == "" {
		opts = append(opts, awsconfig.WithRegion("us-east-1"))
	}
	cfg, err := rootaws.LoadDefaultConfig(ctx, opts...)
	if err != nil {
		return nil, fmt.Errorf("steering: load aws config: %w", err)
	}
	return newRoute53Methods(route53.NewFromConfig(cfg)), nil
}

// ownedCheck is one Route53 health check this deployment owns: tagged
// carlos-steering=<zoneID>, and therefore the only kind this code will ever
// delete.
type ownedCheck struct {
	id  string
	eip string
}

// converge makes the zone's carlos-managed latency records match want.
//
// The order is deliberate and is a safety property, not an implementation
// detail:
//
//  1. read ownership (which health checks are OURS) — a failure here aborts
//     before anything is written, and an empty result means the managed
//     record set is empty, which means NOTHING is deletable;
//  2. read the zone;
//  3. drop names Route53 will not let us write (plainConflicts) and decide
//     what the pass would delete — BOTH REFUSALS RUN BEFORE THE FIRST WRITE,
//     so a refused pass is genuinely zero-write, not merely zero-delete;
//  4. CREATE any missing health check — capacity before records need it;
//  5. UPSERT wanted records, DELETE managed-but-unwanted ones;
//  6. GC health checks nothing references any more — after the records that
//     referenced them are gone, never before.
//
// A failure at any step returns loudly and leaves the rest of the pass
// undone; the housekeeper's next tick re-runs the whole thing from a fresh
// read. Every step is individually idempotent, so a half-applied pass is a
// state this function converges from, not a state it gets stuck in.
func converge(ctx context.Context, client route53API, zoneID string, want []desiredRecord) error {
	if !steeringZonePattern.MatchString(strings.TrimSpace(zoneID)) {
		return fmt.Errorf("steering: refusing to converge: %q is not a bare hosted-zone id", zoneID)
	}
	zoneID = strings.TrimSpace(zoneID)

	// An EMPTY desired set is never an instruction to tear down production
	// DNS. desiredState only returns nothing when zero edges are armed (or
	// there are no steered hosts AND no apps domain), and the overwhelmingly
	// likelier cause of that on a live deployment is a degraded read — a
	// control-bucket blip, a listing that came back short — not an operator
	// deciding every edge should stop answering. Deleting the whole managed
	// set on that signal would turn a transient read failure into a global
	// outage. So: refuse, loudly, and let the next tick decide again.
	// A genuine full teardown is an operator action.
	//
	// This is the degenerate case of the proportional guard below; both
	// exist because "delete everything" and "delete most things" are the
	// same mistake at different sizes.
	if len(want) == 0 {
		log.Printf("steering: empty desired set for zone %s — refusing to remove any record (a wholesale teardown is an operator action, not a converge outcome)", zoneID)
		return nil
	}

	owned, err := ownedHealthChecks(ctx, client, zoneID)
	if err != nil {
		return err
	}
	ownedIDs := make(map[string]bool, len(owned))
	checkByEIP := make(map[string]string, len(owned))
	for _, c := range owned {
		ownedIDs[c.id] = true
		// Deterministic winner when a leak has left two checks on one EIP:
		// lowest id. The loser is unreferenced after this pass and the GC
		// below collects it — the leak self-heals.
		if have, ok := checkByEIP[c.eip]; !ok || c.id < have {
			checkByEIP[c.eip] = c.id
		}
	}

	existing, err := listRecords(ctx, client, zoneID)
	if err != nil {
		return err
	}

	// 3a. names Route53 forbids us from writing at all.
	conflicted := conflictedNames(want, existing)
	if len(conflicted) > 0 {
		want = dropNames(want, conflicted)
		if len(want) == 0 {
			log.Printf("steering: every steered name in zone %s collides with an existing record — refusing to change anything", zoneID)
			return nil
		}
	}

	// 3b. the proportional shrink guard.
	managed, unwanted := partitionManaged(want, existing, ownedIDs, conflicted)
	if limit := shrinkLimit(managed); len(unwanted) > limit {
		for _, rr := range unwanted {
			log.Printf("steering: would delete %s (set %s, value %s)", aws.ToString(rr.Name), aws.ToString(rr.SetIdentifier), firstValue(rr))
		}
		log.Printf("steering: refusing to converge zone %s: the pass would delete %d of %d managed records (limit %d) — a shrink that large is far likelier a short read of edges or routes than an intent to withdraw those names, and a genuine teardown that size is an operator action, exactly like the empty-desired-set case; the intended deletions are listed above",
			zoneID, len(unwanted), managed, limit)
		return nil
	}

	// 4. health checks for every wanted EIP.
	sni := steeringHostname(want)
	loggedSNIFallback := false
	for _, eip := range distinctEIPs(want) {
		if _, ok := checkByEIP[eip]; ok {
			continue
		}
		fqdnFor := sni
		if fqdnFor == "" {
			fqdnFor = fallbackHostname(want, eip)
			// The shared steering hostname is the RIGHT SNI name because
			// every edge holds a certificate for it — a check on it tests
			// the path every visitor takes. Falling back to one member's
			// hostname is a real downgrade: the check now depends on that
			// one member's name and certificate, so if the member later
			// moves or its route is removed, the check starts failing for
			// reasons that have nothing to do with the edge's health.
			//
			// Note the interaction worth seeing in a log: if the synthetic
			// edge.<apps-domain> name was DROPPED for a record conflict just
			// above, this is where that shows up — a name-level conflict
			// silently degrading every health check created afterwards.
			if !loggedSNIFallback {
				log.Printf("steering: no edge.<apps-domain> record in the desired set — new health checks fall back to a member hostname (%s) for SNI instead of the shared steering hostname; if that name was skipped for a record conflict above, THAT is the cause, and the check now rides one member's name and certificate", fqdnFor)
				loggedSNIFallback = true
			}
		}
		id, err := createHealthCheck(ctx, client, zoneID, eip, fqdnFor)
		if err != nil {
			return err
		}
		checkByEIP[eip] = id
		ownedIDs[id] = true
		owned = append(owned, ownedCheck{id: id, eip: eip})
	}

	// 5. records.
	changes, err := recordChanges(want, existing, checkByEIP, ownedIDs, unwanted)
	if err != nil {
		return err
	}
	if err := applyChanges(ctx, client, zoneID, changes); err != nil {
		return err
	}

	// 6. GC. Runs only once the record batches have landed.
	return gcHealthChecks(ctx, client, owned, want, existing, changes)
}

// shrinkLimit is how many managed records one pass may delete, given how many
// it currently manages: half of them, but never fewer than shrinkFloor.
//
// FINDING 1 (review, 2026-08-22). The empty-desired-set refusal above only
// catches a read that returned NOTHING. A SHORT read — three of five edges'
// files listed, or a routes query that came back missing a steered host — has
// a non-empty want and sails straight past it, and the converger would
// dutifully delete every record belonging to the members that went missing.
// For a live steered host that is NXDOMAIN until the next tick: not a slow
// failover, a hard outage, caused by a transient read rather than by anything
// an operator did.
//
// The durable fix is fail-closed reads upstream (Task 6's half: a partial
// listing must produce an ERROR, not a smaller slice). This is the half that
// belongs here — a blast-radius cap that holds even if a caller one day hands
// us a quietly-truncated set. A pass that would remove most of what it manages
// stops and describes itself instead.
//
// The floor keeps small zones workable: a 2-record zone losing one record is
// ordinary, and a purely proportional rule would refuse it forever. The floor
// is small enough that the cap still bites at any size worth protecting.
func shrinkLimit(managed int) int {
	if half := managed / 2; half > shrinkFloor {
		return half
	}
	return shrinkFloor
}

// ---------------------------------------------------------------------------
// ownership
// ---------------------------------------------------------------------------

// ownedHealthChecks returns the account's health checks that THIS deployment
// owns, i.e. that carry carlos-steering=<zoneID>.
//
// Two narrowings, both in the safe direction:
//
//   - the shape prefilter (HTTPS on 443 with an IPAddress, and not owned by
//     another AWS service) only ever SHRINKS the set of checks that can be
//     deleted. Every check this code creates has exactly that shape, so a
//     tagged check of some other shape is one we did not create — ignoring it
//     costs a stranded check; deleting it could cost someone else's
//     monitoring.
//   - the tag is then checked for real, per candidate. Shape alone never
//     confers ownership: the tag is the authority.
//
// A listing or tag call that errors aborts the whole pass. That matters: if
// this function returned a partial set, checks it failed to see would look
// unowned (so records bound to them would look unmanaged — harmless) but,
// worse, their EIPs would look uncovered and a duplicate check would be
// created. Failing loudly is the only correct response.
func ownedHealthChecks(ctx context.Context, client route53API, zoneID string) ([]ownedCheck, error) {
	type candidate struct{ id, eip string }
	var candidates []candidate

	var marker *string
	for {
		out, err := client.ListHealthChecks(ctx, &route53.ListHealthChecksInput{Marker: marker})
		if err != nil {
			return nil, fmt.Errorf("steering: list health checks: %w", err)
		}
		for _, hc := range out.HealthChecks {
			cfg := hc.HealthCheckConfig
			if cfg == nil || hc.LinkedService != nil {
				continue
			}
			if cfg.Type != r53types.HealthCheckTypeHttps || aws.ToInt32(cfg.Port) != healthCheckPort {
				continue
			}
			eip := aws.ToString(cfg.IPAddress)
			if eip == "" {
				continue
			}
			candidates = append(candidates, candidate{id: aws.ToString(hc.Id), eip: eip})
		}
		if !out.IsTruncated || aws.ToString(out.NextMarker) == "" {
			break
		}
		marker = out.NextMarker
	}
	if len(candidates) == 0 {
		return nil, nil
	}
	sort.Slice(candidates, func(i, j int) bool { return candidates[i].id < candidates[j].id })

	eipByID := make(map[string]string, len(candidates))
	ids := make([]string, 0, len(candidates))
	for _, c := range candidates {
		eipByID[c.id] = c.eip
		ids = append(ids, c.id)
	}

	var owned []ownedCheck
	for start := 0; start < len(ids); start += healthCheckTagBatch {
		end := start + healthCheckTagBatch
		if end > len(ids) {
			end = len(ids)
		}
		out, err := client.ListTagsForResources(ctx, &route53.ListTagsForResourcesInput{
			ResourceType: r53types.TagResourceTypeHealthcheck,
			ResourceIds:  ids[start:end],
		})
		if err != nil {
			return nil, fmt.Errorf("steering: list health-check tags: %w", err)
		}
		for _, set := range out.ResourceTagSets {
			id := aws.ToString(set.ResourceId)
			for _, tg := range set.Tags {
				if aws.ToString(tg.Key) == steeringTagKey && aws.ToString(tg.Value) == zoneID {
					owned = append(owned, ownedCheck{id: id, eip: eipByID[id]})
					break
				}
			}
		}
	}
	sort.Slice(owned, func(i, j int) bool { return owned[i].id < owned[j].id })
	return owned, nil
}

// createHealthCheck mints one check for eip and tags it for this zone.
//
// Create-then-tag is two calls and Route53 offers no atomic form. An
// UNTAGGED check is the worst outcome available: this code would never reuse
// it (untagged is not ours) and never GC it (the GC is tag-scoped), so every
// subsequent pass would mint another one — an unbounded leak at the
// housekeeper's cadence. So a tagging failure best-effort deletes the check
// it just made before failing the pass. If that delete ALSO fails, the check
// is named in the log so an operator can find it.
func createHealthCheck(ctx context.Context, client route53API, zoneID, eip, fqdn string) (string, error) {
	// CallerReference must be unique per creation: Route53 remembers it for
	// deleted checks for a while, so a name derived only from the EIP would
	// make a delete-then-recreate (edge disarms, then rearms) fail with
	// HealthCheckAlreadyExists. The timestamp keeps it under the 64-char cap
	// while staying unique. Ownership is NEVER read from this string — the
	// tag is the authority — so its only job is uniqueness plus a human hint
	// in the console.
	ref := fmt.Sprintf("%s%s-%d", healthCheckCallerRefPrefix, eip, time.Now().UnixNano())
	out, err := client.CreateHealthCheck(ctx, &route53.CreateHealthCheckInput{
		CallerReference: aws.String(ref),
		HealthCheckConfig: &r53types.HealthCheckConfig{
			Type:                     r53types.HealthCheckTypeHttps,
			Port:                     aws.Int32(healthCheckPort),
			IPAddress:                aws.String(eip),
			FullyQualifiedDomainName: aws.String(fqdn),
			ResourcePath:             aws.String(healthCheckPath),
			EnableSNI:                aws.Bool(true),
			RequestInterval:          aws.Int32(healthCheckInterval),
			FailureThreshold:         aws.Int32(healthCheckFailureThresh),
			MeasureLatency:           aws.Bool(false),
		},
	})
	if err != nil {
		return "", fmt.Errorf("steering: create health check for %s: %w", eip, err)
	}
	id := ""
	if out.HealthCheck != nil {
		id = aws.ToString(out.HealthCheck.Id)
	}
	if id == "" {
		return "", fmt.Errorf("steering: create health check for %s: no id returned", eip)
	}

	if _, err := client.ChangeTagsForResource(ctx, &route53.ChangeTagsForResourceInput{
		ResourceType: r53types.TagResourceTypeHealthcheck,
		ResourceId:   aws.String(id),
		AddTags:      []r53types.Tag{{Key: aws.String(steeringTagKey), Value: aws.String(zoneID)}},
	}); err != nil {
		if _, delErr := client.DeleteHealthCheck(ctx, &route53.DeleteHealthCheckInput{HealthCheckId: aws.String(id)}); delErr != nil {
			log.Printf("steering: LEAKED health check %s for %s: tagging failed (%v) and so did the cleanup delete (%v) — delete it by hand", id, eip, err, delErr)
		}
		return "", fmt.Errorf("steering: tag health check %s for %s: %w", id, eip, err)
	}
	log.Printf("steering: created health check %s for %s (HTTPS:%d, SNI %s), tagged %s=%s", id, eip, healthCheckPort, fqdn, steeringTagKey, zoneID)
	return id, nil
}

// gcHealthChecks deletes owned checks that nothing references any more.
//
// Two independent gates, both of which must pass:
//
//   - the check is OURS (it is in owned, which is tag-derived — see
//     ownedHealthChecks). This is the gate that makes it structurally
//     impossible to delete a check belonging to another deployment, another
//     tool, or a person.
//   - no record SURVIVING this pass references it: not one of the records we
//     just wrote, and not one of the zone's records we left alone (which
//     includes every static and foreign record). Route53 would refuse a
//     delete of an in-use check anyway, but failing the whole pass on that
//     is worse than skipping the one check and saying so.
//
// A delete that fails is logged, not fatal: the records are already correct,
// and next tick will try again.
func gcHealthChecks(ctx context.Context, client route53API, owned []ownedCheck, want []desiredRecord, existing []r53types.ResourceRecordSet, applied []r53types.Change) error {
	if len(owned) == 0 {
		return nil
	}

	wantedEIP := make(map[string]bool, len(want))
	for _, rec := range want {
		wantedEIP[rec.EIP] = true
	}

	// Every check id still referenced once this pass has landed.
	deleted := make(map[string]bool)
	for _, c := range applied {
		if c.Action == r53types.ChangeActionDelete && c.ResourceRecordSet != nil {
			deleted[recordKey(*c.ResourceRecordSet)] = true
		}
	}
	referenced := make(map[string]bool)
	for _, rr := range existing {
		if deleted[recordKey(rr)] {
			continue
		}
		if id := aws.ToString(rr.HealthCheckId); id != "" {
			referenced[id] = true
		}
	}
	for _, c := range applied {
		if c.Action == r53types.ChangeActionUpsert && c.ResourceRecordSet != nil {
			if id := aws.ToString(c.ResourceRecordSet.HealthCheckId); id != "" {
				referenced[id] = true
			}
		}
	}

	for _, c := range owned {
		if wantedEIP[c.eip] && referenced[c.id] {
			continue
		}
		if referenced[c.id] {
			log.Printf("steering: health check %s (%s) is no longer wanted but a surviving record still references it — leaving it in place", c.id, c.eip)
			continue
		}
		if _, err := client.DeleteHealthCheck(ctx, &route53.DeleteHealthCheckInput{HealthCheckId: aws.String(c.id)}); err != nil {
			log.Printf("steering: could not delete unreferenced health check %s (%s): %v — retrying next pass", c.id, c.eip, err)
			continue
		}
		log.Printf("steering: deleted unreferenced health check %s (%s)", c.id, c.eip)
	}
	return nil
}

// ---------------------------------------------------------------------------
// records
// ---------------------------------------------------------------------------

// listRecords pages the whole zone. Every page matters: a converger that
// stopped early would see records it manages as absent and rewrite them, and
// would see managed-but-unwanted ones as already gone.
func listRecords(ctx context.Context, client route53API, zoneID string) ([]r53types.ResourceRecordSet, error) {
	var out []r53types.ResourceRecordSet
	in := &route53.ListResourceRecordSetsInput{HostedZoneId: aws.String(zoneID)}
	for {
		page, err := client.ListResourceRecordSets(ctx, in)
		if err != nil {
			return nil, fmt.Errorf("steering: list record sets in %s: %w", zoneID, err)
		}
		out = append(out, page.ResourceRecordSets...)
		if !page.IsTruncated || aws.ToString(page.NextRecordName) == "" {
			return out, nil
		}
		in = &route53.ListResourceRecordSetsInput{
			HostedZoneId:          aws.String(zoneID),
			StartRecordName:       page.NextRecordName,
			StartRecordType:       page.NextRecordType,
			StartRecordIdentifier: page.NextRecordIdentifier,
		}
	}
}

// recordKey is a record set's identity: Route53 keys a resource record set on
// (name, type, set identifier), and so does this converger. Names are
// lowercased because DNS is case-insensitive and desiredRecord.Name is
// already lowercase by contract (steering.go's fqdn) — the listing side is
// the only place a stray case could enter.
func recordKey(rr r53types.ResourceRecordSet) string {
	return strings.ToLower(aws.ToString(rr.Name)) + "|" + string(rr.Type) + "|" + aws.ToString(rr.SetIdentifier)
}

func wantKey(rec desiredRecord) string {
	return rec.Name + "|" + string(r53types.RRTypeA) + "|" + rec.Box
}

// isManaged is THE deletion boundary. A record in the steering zone is
// carlos-managed — and therefore something this code may delete — only if ALL
// of the following hold:
//
//   - it is an A record. Nothing else is ever written here.
//   - it carries a SetIdentifier. A record WITHOUT one is a plain,
//     single-answer record: the zone's apex, its www, its MX/NS/SOA, an
//     operator's hand-made host. Those are STATIC / OPERATOR TERRITORY and
//     are never listed as deletable, full stop. This is the rule the brief
//     names and the one that must never be relaxed.
//   - it is a LATENCY record (Region set). A weighted, failover, geo or
//     multivalue record also carries a SetIdentifier and is somebody else's
//     routing policy.
//   - it is not an alias. This converger writes plain A records to EIPs;
//     an alias record under our key was made by something else.
//   - it is bound to a health check WE own (tag-derived, ownedHealthChecks).
//     This is the strongest of the five: every record this code writes is
//     bound to one of our tagged checks, so the binding is a creation
//     signature. It is what makes the answer to "can converge delete a
//     record it did not create?" a structural no rather than a naming
//     convention.
//
// The last gate has a cost, taken deliberately: a managed record whose health
// check was deleted out from under us becomes invisible to this code and will
// linger. It is logged by name every pass (see recordChanges) so it is loud
// rather than silent, and a stale record is a far cheaper failure than a
// wrongly-deleted one.
func isManaged(rr r53types.ResourceRecordSet, ownedIDs map[string]bool) bool {
	if rr.Type != r53types.RRTypeA {
		return false
	}
	if aws.ToString(rr.SetIdentifier) == "" {
		return false
	}
	if rr.Region == "" {
		return false
	}
	if rr.AliasTarget != nil {
		return false
	}
	return ownedIDs[aws.ToString(rr.HealthCheckId)]
}

// desiredRRSet renders one desiredRecord as the record set Route53 should
// hold. TTL 60, type A, plain value (never an AliasTarget), latency routing
// keyed on the box label, bound to the box's EIP health check.
func desiredRRSet(rec desiredRecord, checkID string) r53types.ResourceRecordSet {
	return r53types.ResourceRecordSet{
		Name:            aws.String(rec.Name),
		Type:            r53types.RRTypeA,
		SetIdentifier:   aws.String(rec.Box),
		Region:          r53types.ResourceRecordSetRegion(rec.Region),
		TTL:             aws.Int64(recordTTL),
		ResourceRecords: []r53types.ResourceRecord{{Value: aws.String(rec.EIP)}},
		HealthCheckId:   aws.String(checkID),
	}
}

// sameRRSet reports whether the zone already holds exactly what we want. It
// is what makes a steady-state pass write NOTHING: every field this converger
// sets is compared, and every field it never sets is required to be absent
// (an unexpected Weight or GeoLocation on our key means somebody changed the
// routing policy underneath us, and we should rewrite it).
func sameRRSet(have, want r53types.ResourceRecordSet) bool {
	if !strings.EqualFold(aws.ToString(have.Name), aws.ToString(want.Name)) {
		return false
	}
	if have.Type != want.Type || aws.ToString(have.SetIdentifier) != aws.ToString(want.SetIdentifier) {
		return false
	}
	if have.Region != want.Region || aws.ToInt64(have.TTL) != aws.ToInt64(want.TTL) {
		return false
	}
	if aws.ToString(have.HealthCheckId) != aws.ToString(want.HealthCheckId) {
		return false
	}
	if have.AliasTarget != nil || have.Weight != nil || have.GeoLocation != nil ||
		have.GeoProximityLocation != nil || have.CidrRoutingConfig != nil ||
		have.MultiValueAnswer != nil || have.Failover != "" {
		return false
	}
	if len(have.ResourceRecords) != len(want.ResourceRecords) {
		return false
	}
	for i := range have.ResourceRecords {
		if aws.ToString(have.ResourceRecords[i].Value) != aws.ToString(want.ResourceRecords[i].Value) {
			return false
		}
	}
	return true
}

// conflictedNames returns the wanted names this converger cannot write a
// latency A record at, lowercased with the trailing dot, because an existing
// record there makes the write illegal.
//
// FINDING 2 (review round 1) + FINDING A (round 2, 2026-08-22). Three record
// shapes at a wanted name are all rejected by Route53, and change batches are
// ATOMIC — so the rejection takes the whole batch with it. A single such
// record would wedge EVERY steering record in the zone, on every tick,
// forever: the pass would fail, retry identically, and fail again, while
// unrelated names never got their records. conflictReason enumerates the
// three; they share one failure mode and therefore one cure.
//
// The cure is to skip the conflicting NAME entirely — no upsert, no delete —
// and say so loudly, naming what is in the way. Never the alternative of
// deleting the offending record to make room: it is an operator's record
// (a plain record has no SetIdentifier, which is static/operator territory
// under the same rule isManaged enforces; a CNAME or a weighted set is
// somebody's deliberate routing), and taking a name away from its owner
// because a member typed it into a steering field is not a call this code
// gets to make.
//
// Scope note: this is a NAME-level refusal, and deliberately so — all three
// shapes make the whole name unwritable. It is not the right shape for a
// KEY-level conflict such as an existing ALIAS latency record at exactly one
// (name, SetIdentifier) we want, which Route53 also refuses to convert in
// place via UPSERT; that one is left open on the report rather than guessed
// at here, because dropping the whole name for it would refuse to steer names
// that are in fact steerable.
func conflictedNames(want []desiredRecord, existing []r53types.ResourceRecordSet) map[string]bool {
	wantNames := make(map[string]bool, len(want))
	for _, rec := range want {
		wantNames[strings.ToLower(rec.Name)] = true
	}
	conflicted := map[string]bool{}
	for _, rr := range existing {
		name := strings.ToLower(aws.ToString(rr.Name))
		if !wantNames[name] || conflicted[name] {
			continue
		}
		reason := conflictReason(rr)
		if reason == "" {
			continue
		}
		conflicted[name] = true
		log.Printf("steering: SKIPPING %s entirely: %s — steering this name needs that record changed by its owner first; every other steered name in this zone is unaffected",
			aws.ToString(rr.Name), reason)
	}
	return conflicted
}

// conflictReason names why an existing record makes its name unwritable by
// this converger, or "" if it does not. Three shapes, all with the SAME
// failure mode — the write is rejected and takes the whole atomic batch with
// it — and therefore all with the same cure: skip the name.
func conflictReason(rr r53types.ResourceRecordSet) string {
	setID := aws.ToString(rr.SetIdentifier)
	switch {
	// A CNAME is EXCLUSIVE in DNS: a name holding one may hold no other
	// record type at all. A member steering a hostname that already CNAMEs
	// somewhere in the steering zone is not exotic — it is the ordinary
	// custom-domain shape the spec itself recommends (CNAME to
	// edge.<apps-domain>), pointed at a name inside the same zone.
	case rr.Type == r53types.RRTypeCname:
		return fmt.Sprintf("the zone already holds a CNAME there (-> %s, TTL %d), and a name holding a CNAME may hold no other record type",
			firstValue(rr), aws.ToInt64(rr.TTL))

	// A plain (simple) A record and a routing-policy record set cannot share
	// a name+type.
	case rr.Type == r53types.RRTypeA && setID == "":
		return fmt.Sprintf("the zone already holds a plain A record there (value %s, TTL %d), and Route53 forbids a name from holding both a simple record and a routing-policy record set",
			firstValue(rr), aws.ToInt64(rr.TTL))

	// One routing POLICY per name+type: a weighted, failover, geo, geo-
	// proximity, CIDR or multivalue A set at this name cannot coexist with a
	// latency set. Latency records are the one policy we write, and they are
	// identified by Region being set.
	case rr.Type == r53types.RRTypeA && setID != "" && rr.Region == "":
		return fmt.Sprintf("the zone already holds a NON-LATENCY A record set there (set %q, %s), and Route53 allows only one routing policy per name and type",
			setID, otherPolicyName(rr))
	}
	return ""
}

// otherPolicyName describes which non-latency routing policy a record set
// uses, for the conflict log — an operator reading it should not have to go
// look the record up to know what they are being told about.
func otherPolicyName(rr r53types.ResourceRecordSet) string {
	switch {
	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-answer"
	}
	return "a routing policy this converger does not write"
}

// dropNames removes every desired record whose Name is in the set.
func dropNames(want []desiredRecord, names map[string]bool) []desiredRecord {
	out := make([]desiredRecord, 0, len(want))
	for _, rec := range want {
		if names[strings.ToLower(rec.Name)] {
			continue
		}
		out = append(out, rec)
	}
	return out
}

// partitionManaged counts the zone's carlos-managed records and collects the
// ones no desired key wants — the pass's would-be deletions. It is separated
// from recordChanges so converge can weigh the deletions BEFORE it creates a
// single health check, which is what makes the shrink guard's refusal
// genuinely zero-write.
//
// Records at a conflicted name are excluded from both halves: the pass has
// already promised not to touch that name. That exclusion is a STRUCTURAL
// BACKSTOP and is deliberately untested — every conflict shape
// (conflictReason) is one Route53 forbids from coexisting with a latency
// record set at that name, so a MANAGED record at a conflicted name is a
// state the service cannot produce. It is kept, in the same spirit as
// dedupeRecords in steering.go, so that a misread listing fails by leaving
// records alone rather than by deleting our own.
func partitionManaged(want []desiredRecord, existing []r53types.ResourceRecordSet, ownedIDs, conflicted map[string]bool) (managed int, unwanted []r53types.ResourceRecordSet) {
	wanted := make(map[string]bool, len(want))
	for _, rec := range want {
		wanted[wantKey(rec)] = true
	}
	for _, rr := range existing {
		if conflicted[strings.ToLower(aws.ToString(rr.Name))] {
			continue
		}
		if !isManaged(rr, ownedIDs) {
			// Say so when it LOOKS like ours but fails the ownership gate —
			// a latency A record with a SetIdentifier whose health check we
			// do not own is either an operator's or an orphan of ours whose
			// check went missing. Either way we do not touch it, and either
			// way somebody should know it is there.
			//
			// NOT when this pass WANTS that key, though: recordChanges is
			// about to adopt it, so "leaving it alone" would be a lie about
			// the one record here that does get taken over. (The wanted
			// check was lost when this loop was split out of recordChanges
			// in review round 1; round 2 caught the regression.)
			if rr.Type == r53types.RRTypeA && aws.ToString(rr.SetIdentifier) != "" && rr.Region != "" && rr.AliasTarget == nil &&
				!wanted[recordKey(rr)] {
				log.Printf("steering: leaving latency record %s (set %s) alone: health check %q is not one this deployment owns",
					aws.ToString(rr.Name), aws.ToString(rr.SetIdentifier), aws.ToString(rr.HealthCheckId))
			}
			continue
		}
		managed++
		if !wanted[recordKey(rr)] {
			unwanted = append(unwanted, rr)
		}
	}
	return managed, unwanted
}

// firstValue renders a record set's first value for a log line.
func firstValue(rr r53types.ResourceRecordSet) string {
	if len(rr.ResourceRecords) == 0 {
		return "?"
	}
	return aws.ToString(rr.ResourceRecords[0].Value)
}

// recordChanges diffs want against the zone. The deletions were decided
// already, by partitionManaged, and are passed in — this function only turns
// them into changes.
//
// UPSERTs come first in the returned slice and DELETEs last, so that when the
// changes are chunked, capacity is added before any is taken away.
//
// A wanted key that already holds a record we do NOT manage (a foreign
// latency record at exactly the (name, SetIdentifier) we would create) is
// ADOPTED by the UPSERT — CREATE would collide with it and there is no third
// option. It is logged, because taking over a record is the one write here
// that changes something we did not make. Note the difference from the
// plain-record conflict above: adoption replaces a record of the SAME routing
// policy at the SAME key, which Route53 permits and which is the record we
// would otherwise have created; the plain-record case is a different policy
// at the same name, which Route53 refuses outright.
func recordChanges(want []desiredRecord, existing []r53types.ResourceRecordSet, checkByEIP map[string]string, ownedIDs map[string]bool, unwanted []r53types.ResourceRecordSet) ([]r53types.Change, error) {
	byKey := make(map[string]r53types.ResourceRecordSet, len(existing))
	for _, rr := range existing {
		byKey[recordKey(rr)] = rr
	}

	var upserts []r53types.Change
	for _, rec := range want {
		checkID, ok := checkByEIP[rec.EIP]
		if !ok || checkID == "" {
			// Unreachable: converge creates a check for every distinct EIP
			// before calling this. Loud rather than silently minting a
			// record with no health check, which would never withdraw.
			return nil, fmt.Errorf("steering: no health check for %s (record %s / %s)", rec.EIP, rec.Name, rec.Box)
		}
		desired := desiredRRSet(rec, checkID)
		have, present := byKey[wantKey(rec)]
		if present && sameRRSet(have, desired) {
			continue
		}
		if present && !isManaged(have, ownedIDs) {
			log.Printf("steering: adopting existing record %s (set %s) — it sits at a name/identifier this deployment steers", rec.Name, rec.Box)
		}
		upserts = append(upserts, r53types.Change{Action: r53types.ChangeActionUpsert, ResourceRecordSet: &desired})
	}

	deletes := make([]r53types.Change, 0, len(unwanted))
	for _, rr := range unwanted {
		set := rr // DELETE must resubmit the record set verbatim
		deletes = append(deletes, r53types.Change{Action: r53types.ChangeActionDelete, ResourceRecordSet: &set})
	}

	return append(upserts, deletes...), nil
}

// applyChanges submits the changes in chunks.
//
// Partial failure is explicit: chunk N lands, chunk N+1 fails, the pass
// returns the error and stops. That is safe because each chunk is
// independently valid — an UPSERT is idempotent and a DELETE is of a record
// nothing wants — so the next housekeeper tick re-reads the zone and issues
// exactly the changes still outstanding. The one thing that must NOT happen
// after a failed batch is the health-check GC, and converge returns before
// reaching it.
func applyChanges(ctx context.Context, client route53API, zoneID string, changes []r53types.Change) error {
	if len(changes) == 0 {
		return nil
	}
	for start := 0; start < len(changes); start += changeBatchLimit {
		end := start + changeBatchLimit
		if end > len(changes) {
			end = len(changes)
		}
		chunk := changes[start:end]
		if _, err := client.ChangeResourceRecordSets(ctx, &route53.ChangeResourceRecordSetsInput{
			HostedZoneId: aws.String(zoneID),
			ChangeBatch:  &r53types.ChangeBatch{Changes: chunk},
		}); err != nil {
			return fmt.Errorf("steering: change batch %d-%d of %d in %s: %w", start, end, len(changes), zoneID, err)
		}
		for _, c := range chunk {
			log.Printf("steering: %s %s (set %s)", c.Action, aws.ToString(c.ResourceRecordSet.Name), aws.ToString(c.ResourceRecordSet.SetIdentifier))
		}
	}
	return nil
}

// ---------------------------------------------------------------------------
// naming
// ---------------------------------------------------------------------------

// distinctEIPs returns want's EIPs, sorted, so health checks are created in a
// stable order.
func distinctEIPs(want []desiredRecord) []string {
	seen := make(map[string]bool, len(want))
	out := make([]string, 0, len(want))
	for _, rec := range want {
		if rec.EIP == "" || seen[rec.EIP] {
			continue
		}
		seen[rec.EIP] = true
		out = append(out, rec.EIP)
	}
	sort.Strings(out)
	return out
}

// steeringHostname finds the synthetic edge.<apps-domain> name in want and
// returns it dot-trimmed, for use as a health check's
// FullyQualifiedDomainName — an SNI/Host value, not a Route53 record name, so
// the trailing dot comes off.
//
// It is the right SNI name because every edge holds a certificate covering
// it, so the check exercises the same TLS path a real visitor takes rather
// than one host's certificate in particular.
//
// A member route literally NAMED edge.<apps-domain> produces the identical
// desired records, and desiredState dedupes on (Name, Box) — so this
// converger only ever sees one such name and needs no special case: whether
// it came from the synthetic mint or from a real route, it is the same record
// and the same SNI name.
//
// The prefix-only match here relies on steeredHosts' in-zone check
// (steering.go): every per-host record in want already lies under
// appsDomain, so an "edge."-prefixed Name reaching this function can only
// ever be the one true edge.<apps-domain> hostname, never some other
// zone's "edge.something-else.com" riding along by coincidence of prefix.
// That in-zone filter is what makes this match order-independent; it was
// added for a different reason (a steered host outside the zone wedges
// Route53's whole batch) but closes this hostname's own would-be
// ambiguity as a side effect.
func steeringHostname(want []desiredRecord) string {
	for _, rec := range want {
		if strings.HasPrefix(rec.Name, "edge.") {
			return strings.TrimSuffix(rec.Name, ".")
		}
	}
	return ""
}

// fallbackHostname picks an SNI name for eip when want carries no
// edge.<apps-domain> record (no apps domain configured — steering.go logs
// that case): the first wanted record's own name for that EIP. want's order
// is deterministic (desiredState sorts and dedupes), so the choice is stable
// across passes.
func fallbackHostname(want []desiredRecord, eip string) string {
	for _, rec := range want {
		if rec.EIP == eip {
			return strings.TrimSuffix(rec.Name, ".")
		}
	}
	return ""
}