//go:build e2e

package lb

import (
	"bytes"
	"context"
	"crypto/x509"
	_ "embed"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/service/acm"
	"github.com/aws/aws-sdk-go/service/ec2"
	"github.com/aws/aws-sdk-go/service/elbv2"
	"github.com/mulgadc/spinifex/tests/e2e/harness"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

//go:embed testdata/app-userdata.sh
var appUserData string

const (
	lbVPCCIDR    = "10.200.0.0/16"
	lbSubnetCIDR = "10.200.1.0/24"
	lbKeyName    = "lb-e2e-key"
	httpPort     = 80
	tcpPort      = 9000
	udpPort      = 9001
	triggerPort  = 9090
	probesPerRun = 20
)

// LB kind: ALB or NLB. Used to parameterise the per-suite path.
type lbKind string

const (
	kindALB lbKind = "ALB"
	kindNLB lbKind = "NLB"
)

// TestLoadBalancer runs 8 LB scenarios (ALB/NLB × internet-facing/internal, plus
// HTTPS/UDP/ModifyListener/ListenerRules variants) against a shared VPC and app-instance
// fixture. Each scenario stands up its own dedicated LB (+ target group) — never shared
// across scenarios, since ModifyListener and ListenerRules mutate their LB's listener/rules
// in place and must not race a sibling reading the same LB.
//
// On multinode clusters (peer != "", i.e. the cluster has the node2 the internet-facing
// scenarios drive traffic from — this suite's multinode target is a 3-node cluster) there is
// enough spare capacity to stand up every scenario's dedicated LB at once, so subtests run
// concurrently via t.Parallel() instead of serially tearing one LB down before the next is
// created. Single-node dev boxes lack that headroom and keep the original serial
// create -> verify -> teardown path, where createActiveLB's retry/backoff absorbs the wait
// for the previous LB's sys.micro slot to be reclaimed.
func TestLoadBalancer(t *testing.T) {
	env := harness.LoadEnv(t)
	skipIfDevNetworking(t, env)
	harness.RequireDNSEnabled(t, env)

	// Resolve peer availability before building the shared fixture so the "skip" message
	// appears immediately rather than after minutes of VPC/VM setup.
	peer := pickPeer(env)
	var ssh *harness.PeerSSH
	if peer != "" {
		ssh = harness.NewPeerSSH()
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		if err := ssh.Ping(ctx, peer); err != nil {
			t.Logf("cannot SSH to peer %s: %v (internet-facing suites will skip)", peer, err)
			peer = ""
			ssh = nil
		}
	}
	if peer == "" {
		t.Logf("no peer node available — internet-facing subtests will skip; building shared fixture for internal subtests")
	}

	artifacts := harness.ArtifactDir(t, env)
	client := harness.NewAWSClient(t, env)
	fixture := setupSharedFixture(t, client, artifacts)

	// A live peer implies the 3-node multinode cluster this restructure targets, which has
	// the spare capacity to stand up every scenario's LB concurrently (see doc comment
	// above). Single-node dev boxes (peer == "") fall through to the serial path below.
	parallelizeLBs := peer != ""

	// Internal and internet-facing LB records share one zone object: ELBChanges
	// sets Zone to the base domain for both, and the names differ only by an
	// "internal-" prefix. Concurrent registrations used to clobber each other
	// there, which is why these ran serially; the writer now takes a per-zone lock,
	// so all of them register in parallel.
	t.Run("InternetFacing_ALB", func(t *testing.T) {
		if peer == "" {
			t.Skip("no peer node available")
		}
		if parallelizeLBs {
			t.Parallel()
		}
		runLBSuite(t, client, fixture, kindALB, "internet-facing", ssh, peer)
	})
	t.Run("InternetFacing_NLB", func(t *testing.T) {
		if peer == "" {
			t.Skip("no peer node available")
		}
		if parallelizeLBs {
			t.Parallel()
		}
		runLBSuite(t, client, fixture, kindNLB, "internet-facing", ssh, peer)
	})
	t.Run("InternetFacing_ALB_HTTPS", func(t *testing.T) {
		if peer == "" {
			// HTTPS handshake is driven from the test runner against the LB
			// public IP; gate on the same peer-available signal the sibling
			// internet-facing subtests use, where driver→LB reachability holds.
			t.Skip("no peer node available")
		}
		if parallelizeLBs {
			t.Parallel()
		}
		runHTTPSCertSuite(t, client, fixture)
	})

	if parallelizeLBs {
		// Multinode: each internal subtest stands up its own independent LB and they start
		// together with the internet-facing group above, so there is no serial queue for one
		// stuck LB to back up behind.
		// A single LB never reaching active only fails that one subtest — it can no longer
		// cascade into 5x sequential 5-minute timeouts, so the serial path's fail-fast gate
		// (below) is unnecessary here: concurrency itself bounds the worst-case wall time to
		// one timeout window instead of five.
		t.Run("Internal_ALB", func(t *testing.T) {
			t.Parallel()
			runLBSuite(t, client, fixture, kindALB, "internal", nil, "")
		})
		t.Run("Internal_NLB", func(t *testing.T) {
			t.Parallel()
			runLBSuite(t, client, fixture, kindNLB, "internal", nil, "")
		})
		t.Run("Internal_NLB_UDP", func(t *testing.T) {
			t.Parallel()
			runUDPNLBSuite(t, client, fixture)
		})
		t.Run("Internal_ALB_ModifyListener", func(t *testing.T) {
			t.Parallel()
			runModifyListenerSuite(t, client, fixture)
		})
		t.Run("Internal_ALB_ListenerRules", func(t *testing.T) {
			t.Parallel()
			runListenerRulesSuite(t, client, fixture)
		})
		return
	}

	// Single-node: gate remaining internal subtests on Internal_ALB — if the LB never
	// reaches active, the rest will time out identically, so fail fast instead of burning
	// ~5min each. No fixed inter-subtest sleeps: createActiveLB retries with backoff when
	// the previous LB's sys.micro slot is still being reclaimed.
	albOK := t.Run("Internal_ALB", func(t *testing.T) {
		runLBSuite(t, client, fixture, kindALB, "internal", nil, "")
	})
	skipIfInternalBroken := func(t *testing.T) {
		if !albOK {
			t.Skip("Internal_ALB failed (LB never reached active) — skipping remaining internal subtests to fail fast")
		}
	}
	t.Run("Internal_NLB", func(t *testing.T) {
		skipIfInternalBroken(t)
		runLBSuite(t, client, fixture, kindNLB, "internal", nil, "")
	})
	t.Run("Internal_NLB_UDP", func(t *testing.T) {
		skipIfInternalBroken(t)
		runUDPNLBSuite(t, client, fixture)
	})
	t.Run("Internal_ALB_ModifyListener", func(t *testing.T) {
		skipIfInternalBroken(t)
		runModifyListenerSuite(t, client, fixture)
	})
	t.Run("Internal_ALB_ListenerRules", func(t *testing.T) {
		skipIfInternalBroken(t)
		runListenerRulesSuite(t, client, fixture)
	})
}

// --- Fixture: shared VPC, subnet, IGW, SG, app instances ----------------

type sharedFixture struct {
	VPCID            string
	SubnetID         string
	IGWID            string
	MainRouteTableID string
	SecurityGroup    string
	AMIID            string
	InstanceType     string
	AppInstanceIDs   []string
	ClientID         string
	ClientPublicIP   string
}

func setupSharedFixture(t *testing.T, c *harness.AWSClient, artifacts string) *sharedFixture {
	t.Helper()
	f := &sharedFixture{}

	harness.Phase(t, "Discovering cluster capacity")
	f.InstanceType = discoverNanoInstanceType(t, c)
	f.AMIID = discoverAMI(t, c)

	harness.Phase(t, "Ensuring SSH key pair %q", lbKeyName)
	ensureKeyPair(t, c)
	t.Cleanup(func() { deleteKeyPair(t, c) })

	harness.Phase(t, "Creating shared VPC topology (%s)", lbVPCCIDR)
	createVPC(t, c, f)
	t.Cleanup(func() { deleteVPC(t, c, f) })
	createIGW(t, c, f)
	t.Cleanup(func() { deleteIGW(t, c, f) })
	createSubnet(t, c, f)
	t.Cleanup(func() { deleteSubnet(t, c, f) })
	addPublicRoute(t, c, f)
	t.Cleanup(func() { deletePublicRoute(t, c, f) })
	configureDefaultSG(t, c, f)

	harness.Phase(t, "Launching app instances (2× %s)", f.InstanceType)
	launchAppInstances(t, c, f)
	t.Cleanup(func() { terminateInstances(t, c, f.AppInstanceIDs) })

	harness.Phase(t, "Launching probe client")
	launchSharedProbeClient(t, c, f)
	t.Cleanup(func() { terminateInstances(t, c, []string{f.ClientID}) })

	harness.OnFailure(t, func() {
		dumpDaemonLogs(t, artifacts, "setup")
	})
	return f
}

//go:embed testdata/client-userdata.sh
var clientUserData string

// --- Discovery helpers ---------------------------------------------------

func skipIfDevNetworking(t *testing.T, env *harness.Env) {
	t.Helper()
	cfg := os.ExpandEnv("$HOME/spinifex/config/spinifex.toml")
	if env.ConfigDir != "" {
		cfg = env.ConfigDir + "/spinifex.toml"
	}
	raw, err := os.ReadFile(cfg)
	if err != nil {
		return
	}
	if bytes.Contains(raw, []byte("dev_networking = true")) {
		t.Skip("dev_networking enabled in spinifex.toml; LB E2E requires pool mode w/ external IPAM")
	}
}

func pickPeer(env *harness.Env) string {
	if len(env.NodeIPs) < 2 {
		return ""
	}
	return env.NodeIPs[1]
}

func discoverNanoInstanceType(t *testing.T, c *harness.AWSClient) string {
	t.Helper()
	out, err := c.EC2.DescribeInstanceTypes(&ec2.DescribeInstanceTypesInput{})
	require.NoError(t, err)
	for _, it := range out.InstanceTypes {
		name := aws.StringValue(it.InstanceType)
		if strings.Contains(name, "nano") {
			t.Logf("instance type: %s", name)
			return name
		}
	}
	t.Fatal("no nano instance type available")
	return ""
}

func discoverAMI(t *testing.T, c *harness.AWSClient) string {
	t.Helper()
	out, err := c.EC2.DescribeImages(&ec2.DescribeImagesInput{})
	require.NoError(t, err)
	var fallback, nonAlpine, ubuntu string
	for _, img := range out.Images {
		id := aws.StringValue(img.ImageId)
		name := aws.StringValue(img.Name)
		if fallback == "" {
			fallback = id
		}
		if !strings.Contains(strings.ToLower(name), "alpine") && nonAlpine == "" {
			nonAlpine = id
		}
		if strings.HasPrefix(name, "ami-ubuntu") {
			ubuntu = id
			break
		}
	}
	for _, candidate := range []string{ubuntu, nonAlpine, fallback} {
		if candidate != "" {
			t.Logf("AMI: %s", candidate)
			return candidate
		}
	}
	t.Fatal("no AMIs available")
	return ""
}

func ensureKeyPair(t *testing.T, c *harness.AWSClient) {
	t.Helper()
	_, _ = c.EC2.DeleteKeyPair(&ec2.DeleteKeyPairInput{KeyName: aws.String(lbKeyName)})
	_, err := c.EC2.CreateKeyPair(&ec2.CreateKeyPairInput{KeyName: aws.String(lbKeyName)})
	require.NoError(t, err, "create key pair %s", lbKeyName)
}

func deleteKeyPair(t *testing.T, c *harness.AWSClient) {
	t.Helper()
	if _, err := c.EC2.DeleteKeyPair(&ec2.DeleteKeyPairInput{KeyName: aws.String(lbKeyName)}); err != nil {
		t.Logf("delete key pair: %v", err)
	}
}

// --- VPC / Subnet / IGW / SG --------------------------------------------

func createVPC(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	out, err := c.EC2.CreateVpc(&ec2.CreateVpcInput{CidrBlock: aws.String(lbVPCCIDR)})
	require.NoError(t, err)
	f.VPCID = aws.StringValue(out.Vpc.VpcId)
	t.Logf("VPC: %s", f.VPCID)
}

func deleteVPC(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	if f.VPCID == "" {
		return
	}
	if _, err := c.EC2.DeleteVpc(&ec2.DeleteVpcInput{VpcId: aws.String(f.VPCID)}); err != nil {
		t.Logf("delete VPC %s: %v", f.VPCID, err)
	}
}

func createIGW(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	out, err := c.EC2.CreateInternetGateway(&ec2.CreateInternetGatewayInput{})
	require.NoError(t, err)
	f.IGWID = aws.StringValue(out.InternetGateway.InternetGatewayId)
	_, err = c.EC2.AttachInternetGateway(&ec2.AttachInternetGatewayInput{
		InternetGatewayId: aws.String(f.IGWID),
		VpcId:             aws.String(f.VPCID),
	})
	require.NoError(t, err)
	t.Logf("IGW: %s (attached to %s)", f.IGWID, f.VPCID)
}

func deleteIGW(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	if f.IGWID == "" {
		return
	}
	_, _ = c.EC2.DetachInternetGateway(&ec2.DetachInternetGatewayInput{
		InternetGatewayId: aws.String(f.IGWID),
		VpcId:             aws.String(f.VPCID),
	})
	if _, err := c.EC2.DeleteInternetGateway(&ec2.DeleteInternetGatewayInput{
		InternetGatewayId: aws.String(f.IGWID),
	}); err != nil {
		t.Logf("delete IGW: %v", err)
	}
}

func createSubnet(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	out, err := c.EC2.CreateSubnet(&ec2.CreateSubnetInput{
		VpcId:     aws.String(f.VPCID),
		CidrBlock: aws.String(lbSubnetCIDR),
	})
	require.NoError(t, err)
	f.SubnetID = aws.StringValue(out.Subnet.SubnetId)
	_, err = c.EC2.ModifySubnetAttribute(&ec2.ModifySubnetAttributeInput{
		SubnetId:            aws.String(f.SubnetID),
		MapPublicIpOnLaunch: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
	})
	require.NoError(t, err)
	t.Logf("subnet: %s (MapPublicIpOnLaunch)", f.SubnetID)
}

func deleteSubnet(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	if f.SubnetID == "" {
		return
	}
	if _, err := c.EC2.DeleteSubnet(&ec2.DeleteSubnetInput{SubnetId: aws.String(f.SubnetID)}); err != nil {
		t.Logf("delete subnet: %v", err)
	}
}

// addPublicRoute installs 0.0.0.0/0 → IGW on the VPC's main route table so the daemon
// classifies the subnet as public and does not install an OVN LR egress DROP.
func addPublicRoute(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	out, err := c.EC2.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
		Filters: []*ec2.Filter{
			{Name: aws.String("vpc-id"), Values: []*string{aws.String(f.VPCID)}},
			{Name: aws.String("association.main"), Values: []*string{aws.String("true")}},
		},
	})
	require.NoError(t, err, "describe main route table")
	require.NotEmpty(t, out.RouteTables, "VPC %s has no main route table", f.VPCID)
	f.MainRouteTableID = aws.StringValue(out.RouteTables[0].RouteTableId)
	_, err = c.EC2.CreateRoute(&ec2.CreateRouteInput{
		RouteTableId:         aws.String(f.MainRouteTableID),
		DestinationCidrBlock: aws.String("0.0.0.0/0"),
		GatewayId:            aws.String(f.IGWID),
	})
	require.NoError(t, err, "create-route 0.0.0.0/0 -> IGW")
	t.Logf("public route: %s 0.0.0.0/0 -> %s", f.MainRouteTableID, f.IGWID)
}

func deletePublicRoute(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	if f.MainRouteTableID == "" {
		return
	}
	if _, err := c.EC2.DeleteRoute(&ec2.DeleteRouteInput{
		RouteTableId:         aws.String(f.MainRouteTableID),
		DestinationCidrBlock: aws.String("0.0.0.0/0"),
	}); err != nil {
		t.Logf("delete public route: %v", err)
	}
}

func configureDefaultSG(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	out, err := c.EC2.DescribeSecurityGroups(&ec2.DescribeSecurityGroupsInput{
		Filters: []*ec2.Filter{
			{Name: aws.String("vpc-id"), Values: []*string{aws.String(f.VPCID)}},
			{Name: aws.String("group-name"), Values: []*string{aws.String("default")}},
		},
	})
	require.NoError(t, err)
	require.NotEmpty(t, out.SecurityGroups, "VPC default SG missing")
	f.SecurityGroup = aws.StringValue(out.SecurityGroups[0].GroupId)
	t.Logf("default SG: %s", f.SecurityGroup)

	// Use structured IpPermissions form — vpcd ignores the top-level shorthand fields
	// so the OVN ACL would never be installed otherwise.
	for _, port := range []int64{httpPort, tcpPort, triggerPort} {
		_, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
			GroupId: aws.String(f.SecurityGroup),
			IpPermissions: []*ec2.IpPermission{{
				IpProtocol: aws.String("tcp"),
				FromPort:   aws.Int64(port),
				ToPort:     aws.Int64(port),
				IpRanges:   []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
			}},
		})
		if err != nil {
			var aerr awserr.Error
			if !errors.As(err, &aerr) || aerr.Code() != "InvalidPermission.Duplicate" {
				t.Fatalf("authorize tcp/%d: %v", port, err)
			}
		}
	}

	// UDP data plane for the NLB UDP suite (health-checked over TCP/9000).
	if _, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
		GroupId: aws.String(f.SecurityGroup),
		IpPermissions: []*ec2.IpPermission{{
			IpProtocol: aws.String("udp"),
			FromPort:   aws.Int64(udpPort),
			ToPort:     aws.Int64(udpPort),
			IpRanges:   []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
		}},
	}); err != nil {
		var aerr awserr.Error
		if !errors.As(err, &aerr) || aerr.Code() != "InvalidPermission.Duplicate" {
			t.Fatalf("authorize udp/%d: %v", udpPort, err)
		}
	}
}

// --- App instances -------------------------------------------------------

func launchAppInstances(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	for i := 0; i < 2; i++ {
		out, err := c.EC2.RunInstances(&ec2.RunInstancesInput{
			ImageId:      aws.String(f.AMIID),
			InstanceType: aws.String(f.InstanceType),
			KeyName:      aws.String(lbKeyName),
			SubnetId:     aws.String(f.SubnetID),
			MinCount:     aws.Int64(1),
			MaxCount:     aws.Int64(1),
			UserData:     aws.String(base64Encode(appUserData)),
		})
		require.NoErrorf(t, err, "run-instances app %d", i+1)
		require.NotEmpty(t, out.Instances)
		id := aws.StringValue(out.Instances[0].InstanceId)
		f.AppInstanceIDs = append(f.AppInstanceIDs, id)
		t.Logf("app instance %d: %s", i+1, id)
	}

	for _, id := range f.AppInstanceIDs {
		harness.WaitForInstanceRunning(t, c, id, 120*time.Second)
	}
}

func terminateInstances(t *testing.T, c *harness.AWSClient, ids []string) {
	t.Helper()
	if len(ids) == 0 {
		return
	}
	awsIDs := make([]*string, len(ids))
	for i, id := range ids {
		awsIDs[i] = aws.String(id)
	}
	if _, err := c.EC2.TerminateInstances(&ec2.TerminateInstancesInput{InstanceIds: awsIDs}); err != nil {
		t.Logf("terminate instances: %v", err)
		return
	}
	harness.WaitForInstanceTerminated(t, c, ids, 60*time.Second)
}

// --- LB suite: one LB (ALB or NLB) × one scheme (internet-facing/internal)

// runLBSuite creates one TG + LB + listener, asserts scheme/DNS/ENI invariants, runs
// traffic tests, then tears down before returning. Self-contained so subtests don't
// accumulate LB system instances on capacity-constrained dev nodes.
func runLBSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture, kind lbKind, scheme string, ssh *harness.PeerSSH, peer string) {
	t.Helper()
	suffix := "int"
	if scheme == "internet-facing" {
		suffix = "inet"
	}
	lbName := strings.ToLower(string(kind))
	label := fmt.Sprintf("%s %s", kind, scheme)

	var proto, hcPath, eniDescPrefix, lbType string
	var port int64
	if kind == kindALB {
		proto, port, hcPath, eniDescPrefix, lbType = "HTTP", httpPort, "/index.html", "app", "application"
	} else {
		proto, port, hcPath, eniDescPrefix, lbType = "TCP", tcpPort, "", "net", "network"
	}

	tgArn := createTargetGroup(t, c, f, fmt.Sprintf("lb-e2e-%s-%s-tg", lbName, suffix), proto, port, hcPath)
	t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })

	registerTargets(t, c, tgArn, f.AppInstanceIDs)
	t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })

	// ALB scenarios get their own LB-facing SG so concurrent scenarios (parallel
	// internal group) never share one SG; NLBs auto-isolate via their managed SG.
	var sgIDs []string
	if kind == kindALB {
		sgIDs = []string{createScenarioSG(t, c, f, fmt.Sprintf("lb-e2e-%s-%s-sg", lbName, suffix), port)}
	}

	lb, _ := createActiveLB(t, c, f, fmt.Sprintf("lb-e2e-%s-%s", lbName, suffix), lbType, scheme, proto, port, tgArn, label, sgIDs)

	assert.Equal(t, scheme, lb.Scheme, label+" scheme")
	assert.Equal(t, lbType, lb.Type, label+" type")
	if kind == kindNLB {
		assert.Contains(t, lb.ARN, "/net/", label+" ARN must contain /net/")
	}

	if kind == kindNLB {
		captureLBConsoleOnFailure(t, c, eniDescPrefix, lb)
	}
	harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)

	eni := lbENI(t, c, eniDescPrefix, lb)

	if scheme == "internet-facing" {
		ip := publicIP(eni)
		require.NotEmpty(t, ip, label+" needs public IP")
		assertLBDNSResolves(t, lb.DNSName, ip, label)
		runInternetFacingTrafficSingle(t, kind, ssh, peer, ip)
		if kind == kindNLB {
			runNLBDeregisterDraining(t, c, tgArn, f.AppInstanceIDs[0])
		}
		return
	}

	// internal
	assert.Empty(t, publicIP(eni), label+" must not have public IP")
	priv := privateIP(eni)
	require.NotEmpty(t, priv, label+" needs private IP")
	assertInternalDNS(t, c, lb.ARN, label)
	assertLBDNSResolves(t, lb.DNSName, priv, label)
	runInternalTrafficViaClient(t, c, f, kind, priv)
}

// runHTTPSCertSuite exercises the ACM → HTTPS listener path end-to-end:
// ImportCertificate a self-signed leaf → create an internet-facing ALB HTTPS
// listener referencing the cert ARN → complete a TLS handshake against the LB
// and assert HAProxy serves exactly the imported cert → drive one HTTPS GET
// that terminates TLS at the LB and forwards to a healthy backend →
// DeleteCertificate. The cert is minted for the LB's public IP (known only
// after the LB is active) so the end-to-end GET can verify the chain by IP SAN.
func runHTTPSCertSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	const label = "ALB internet-facing HTTPS (ACM)"
	const httpsPort int64 = 443

	sgIDs := []string{createScenarioSG(t, c, f, "lb-e2e-https-sg", httpsPort)}

	tgArn := createTargetGroup(t, c, f, "lb-e2e-https-tg", "HTTP", httpPort, "/index.html")
	t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })
	registerTargets(t, c, tgArn, f.AppInstanceIDs)
	t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })

	lb, _ := createActiveLB(t, c, f, "lb-e2e-https", "application", "internet-facing", "", 0, "", label, sgIDs)
	eni := lbENI(t, c, "app", lb)
	pubIP := publicIP(eni)
	require.NotEmpty(t, pubIP, label+" needs a public IP for the TLS handshake")

	// Mint a self-signed leaf for the LB's public IP and import it into ACM.
	certPEM, keyPEM, leaf, err := harness.GenerateSelfSignedCertPEM(pubIP)
	require.NoError(t, err, "generate self-signed cert")
	imp, err := c.ACM.ImportCertificate(&acm.ImportCertificateInput{
		Certificate: certPEM,
		PrivateKey:  keyPEM,
	})
	require.NoError(t, err, "acm import-certificate")
	certArn := aws.StringValue(imp.CertificateArn)
	require.NotEmpty(t, certArn, "ImportCertificate returned empty CertificateArn")
	require.Contains(t, certArn, ":certificate/", "cert ARN must be an ACM certificate ARN")
	t.Cleanup(func() {
		if _, derr := c.ACM.DeleteCertificate(&acm.DeleteCertificateInput{
			CertificateArn: aws.String(certArn),
		}); derr != nil {
			t.Logf("delete certificate %s: %v", certArn, derr)
		}
	})
	t.Logf("imported cert: %s", certArn)

	listener := createHTTPSListener(t, c, lb.ARN, httpsPort, tgArn, certArn)
	t.Cleanup(func() { deleteListener(t, c, listener) })

	harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)

	// TLS handshake: HAProxy must present exactly the imported leaf. The
	// lb-agent picks up the new listener cert on its reconcile tick, so poll
	// the handshake until the served cert matches before asserting.
	var served *x509.Certificate
	harness.EventuallyErr(t, func() error {
		got, ferr := harness.FetchServedCert(pubIP, int(httpsPort), 5*time.Second)
		if ferr != nil {
			return ferr
		}
		if !harness.FingerprintMatches(got, leaf) {
			return fmt.Errorf("served cert does not match imported cert yet")
		}
		served = got
		return nil
	}, 90*time.Second, 3*time.Second)
	require.NotNil(t, served, label+" never served the imported cert")
	t.Logf("TLS handshake ok — HAProxy served the imported cert (CN=%s)", served.Subject.CommonName)

	version, _, err := harness.FetchTLSPosture(pubIP, int(httpsPort), 5*time.Second)
	require.NoError(t, err, "fetch TLS posture")
	assert.GreaterOrEqualf(t, version, uint16(0x0303), "negotiated TLS version must be >= 1.2, got 0x%04x", version)

	// End-to-end: one HTTPS GET that terminates TLS at the LB and forwards to a
	// backend. Verify the chain against the imported cert (IP SAN matches pubIP).
	pool := x509.NewCertPool()
	require.True(t, pool.AppendCertsFromPEM(certPEM), "build CA pool from imported cert")
	url := fmt.Sprintf("https://%s:%d/index.html", pubIP, httpsPort)
	harness.EventuallyErr(t, func() error {
		status, body, gerr := harness.HTTPSGet(url, pool, 10*time.Second)
		if gerr != nil {
			return gerr
		}
		if status != http.StatusOK {
			return fmt.Errorf("status %d (body=%q)", status, string(body))
		}
		return nil
	}, 90*time.Second, 3*time.Second)
	t.Logf("HTTPS GET %s -> 200 (TLS terminated at LB, forwarded to backend)", url)
}

// createHTTPSListener creates an HTTPS listener referencing certArn. Distinct
// from createListener because the ELBv2 handler requires Certificates for HTTPS
// and rejects them for HTTP.
func createHTTPSListener(t *testing.T, c *harness.AWSClient, lbArn string, port int64, tgArn, certArn string) string {
	t.Helper()
	// e2e:allow-create — the HTTPS listener is the subject under test (ACM cert -> termination).
	out, err := c.ELBv2.CreateListener(&elbv2.CreateListenerInput{
		LoadBalancerArn: aws.String(lbArn),
		Protocol:        aws.String("HTTPS"),
		Port:            aws.Int64(port),
		Certificates:    []*elbv2.Certificate{{CertificateArn: aws.String(certArn)}},
		DefaultActions: []*elbv2.Action{{
			Type:           aws.String("forward"),
			TargetGroupArn: aws.String(tgArn),
		}},
	})
	require.NoError(t, err, "create-listener HTTPS")
	require.NotEmpty(t, out.Listeners, "create-listener HTTPS returned no listeners")
	arn := aws.StringValue(out.Listeners[0].ListenerArn)
	t.Logf("listener HTTPS:%d -> %s (cert %s)", port, arn, certArn)
	return arn
}

// createScenarioSG creates a dedicated security group for one ALB scenario's
// frontend/listener ports and authorizes tcp/0.0.0.0/0 on each of ports. Giving
// each ALB scenario its own SG means concurrent scenarios (see the parallel
// internal group in TestLoadBalancer) never authorize into, or otherwise
// mutate, the same SG object.
//
// This only isolates the LB-FACING side. The shared app-instance/probe-client
// SG (f.SecurityGroup, set up by configureDefaultSG) is left untouched and
// keeps permitting LB->target health-check and data traffic: SG ingress rules
// gate inbound traffic at the DESTINATION ENI's own SG regardless of which SG
// the source ENI carries, so targets never needed to share the LB's SG in the
// first place.
//
// NLBs don't need this: CreateLoadBalancer with no explicit SecurityGroups
// already gives every NLB its own auto-managed SG with listener ports opened
// automatically, so only ALB scenarios (which otherwise fall back onto the one
// shared VPC default SG) require an explicit dedicated SG here.
func createScenarioSG(t *testing.T, c *harness.AWSClient, f *sharedFixture, name string, ports ...int64) string {
	t.Helper()
	out, err := c.EC2.CreateSecurityGroup(&ec2.CreateSecurityGroupInput{ // e2e:allow-create — a dedicated per-scenario LB SG is the isolation under test; a shared Ensure* fixture would defeat it.
		VpcId:       aws.String(f.VPCID),
		GroupName:   aws.String(name),
		Description: aws.String("lb-e2e dedicated LB-facing security group"),
	})
	require.NoErrorf(t, err, "create-security-group %s", name)
	sgID := aws.StringValue(out.GroupId)
	t.Cleanup(func() { deleteSecurityGroup(t, c, sgID) })

	for _, port := range ports {
		_, err := c.EC2.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{
			GroupId: aws.String(sgID),
			IpPermissions: []*ec2.IpPermission{{
				IpProtocol: aws.String("tcp"),
				FromPort:   aws.Int64(port),
				ToPort:     aws.Int64(port),
				IpRanges:   []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
			}},
		})
		require.NoErrorf(t, err, "authorize tcp/%d on %s", port, name)
	}
	t.Logf("scenario SG %s: %s (tcp ports=%v)", name, sgID, ports)
	return sgID
}

// deleteSecurityGroup best-effort deletes a scenario SG created by
// createScenarioSG. Registered via t.Cleanup after the LB's own teardown
// (LIFO order means the LB, and the ENI referencing this SG, is gone first).
func deleteSecurityGroup(t *testing.T, c *harness.AWSClient, sgID string) {
	if sgID == "" {
		return
	}
	if _, err := c.EC2.DeleteSecurityGroup(&ec2.DeleteSecurityGroupInput{GroupId: aws.String(sgID)}); err != nil {
		t.Logf("delete security group %s: %v", sgID, err)
	}
}

// runUDPNLBSuite exercises an internal NLB with a UDP listener. The lb-agent probes
// targets over TCP/9000 (nginx stream has no active health-check); once healthy, a
// UDP datagram round-trip through the VIP proves the L4 data plane.
func runUDPNLBSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	const label = "NLB internal UDP"

	tgArn := createUDPTargetGroup(t, c, f, "lb-e2e-nlb-udp-tg")
	t.Cleanup(func() { deleteTargetGroup(t, c, tgArn) })

	registerTargets(t, c, tgArn, f.AppInstanceIDs)
	t.Cleanup(func() { deregisterTargets(t, c, tgArn, f.AppInstanceIDs) })

	// NLB — no dedicated SG needed; CreateLoadBalancer auto-provisions one.
	lb, _ := createActiveLB(t, c, f, "lb-e2e-nlb-udp", "network", "internal", "UDP", udpPort, tgArn, label, nil)

	assert.Equal(t, "network", lb.Type, label+" type")
	assert.Contains(t, lb.ARN, "/net/", label+" ARN must contain /net/")

	captureLBConsoleOnFailure(t, c, "net", lb)
	// The key assertion: NLB targets reach healthy via the agent's active
	// prober. Pre-feature this timed out (nginx reported empty server lists).
	harness.WaitForTargetsHealthy(t, c, tgArn, 2, label, 2*time.Minute)
	// Targets only reach healthy if the lb-agent authenticated to the gateway;
	// with static keys dropped from user-data that auth is IMDS-only, so assert
	// the cred provenance explicitly (no baked keys, assumed-role principal).
	assertLBAgentAuthViaIMDS(t)

	eni := lbENI(t, c, "net", lb)
	priv := privateIP(eni)
	require.NotEmpty(t, priv, label+" needs private IP")

	runInternalUDPViaClient(t, f, priv, label)
}

// runInternalUDPViaClient drives the shared client to send UDP datagrams at the
// NLB VIP and asserts the echoed hostnames round-robin across both backends.
func runInternalUDPViaClient(t *testing.T, f *sharedFixture, lbIP, label string) {
	t.Helper()
	resultsFile := fmt.Sprintf("nlb-udp-%d.txt", time.Now().UnixNano())
	order := map[string]any{
		"proto":    "udp",
		"ip":       lbIP,
		"count":    probesPerRun,
		"outfile":  resultsFile,
		"udp_port": udpPort,
	}
	body, err := json.Marshal(order)
	require.NoError(t, err)

	triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
	client := &http.Client{Timeout: 120 * time.Second}
	resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
	require.NoErrorf(t, err, "trigger %s probe", label)
	resp.Body.Close()
	require.Equalf(t, 200, resp.StatusCode, "trigger %s probe HTTP status", label)

	results, err := plainHTTPGet(
		fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
		10*time.Second)
	require.NoErrorf(t, err, "fetch %s", resultsFile)
	harness.AssertRoundRobin(t,
		harness.VerifyResultsLines(results, "udp"),
		1, probesPerRun/2, label)
}

// --- LB CRUD helpers -----------------------------------------------------

func createTargetGroup(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, proto string, port int64, hcPath string) string {
	t.Helper()
	in := &elbv2.CreateTargetGroupInput{
		Name:                       aws.String(name),
		Protocol:                   aws.String(proto),
		Port:                       aws.Int64(port),
		VpcId:                      aws.String(f.VPCID),
		HealthCheckIntervalSeconds: aws.Int64(5),
		HealthyThresholdCount:      aws.Int64(2),
		UnhealthyThresholdCount:    aws.Int64(2),
	}
	if hcPath != "" {
		in.HealthCheckPath = aws.String(hcPath)
	} else {
		in.HealthCheckProtocol = aws.String("TCP")
		in.HealthCheckIntervalSeconds = aws.Int64(10)
	}
	out, err := c.ELBv2.CreateTargetGroup(in)
	require.NoErrorf(t, err, "create-target-group %s", name)
	arn := aws.StringValue(out.TargetGroups[0].TargetGroupArn)
	t.Logf("TG %s: %s", name, arn)
	return arn
}

// createUDPTargetGroup creates a UDP TG with a TCP health check (UDP can't be probed),
// exercising the nginx agent's active TCP prober that advances NLB targets to healthy.
func createUDPTargetGroup(t *testing.T, c *harness.AWSClient, f *sharedFixture, name string) string {
	t.Helper()
	out, err := c.ELBv2.CreateTargetGroup(&elbv2.CreateTargetGroupInput{
		Name:                       aws.String(name),
		Protocol:                   aws.String("UDP"),
		Port:                       aws.Int64(udpPort),
		VpcId:                      aws.String(f.VPCID),
		HealthCheckProtocol:        aws.String("TCP"),
		HealthCheckPort:            aws.String(fmt.Sprintf("%d", tcpPort)),
		HealthCheckIntervalSeconds: aws.Int64(10),
		HealthyThresholdCount:      aws.Int64(2),
		UnhealthyThresholdCount:    aws.Int64(2),
	})
	require.NoErrorf(t, err, "create-target-group %s (UDP)", name)
	arn := aws.StringValue(out.TargetGroups[0].TargetGroupArn)
	t.Logf("UDP TG %s: %s (HC TCP:%d)", name, arn, tcpPort)
	return arn
}

func deleteTargetGroup(t *testing.T, c *harness.AWSClient, arn string) {
	if arn == "" {
		return
	}
	if _, err := c.ELBv2.DeleteTargetGroup(&elbv2.DeleteTargetGroupInput{TargetGroupArn: aws.String(arn)}); err != nil {
		t.Logf("delete TG %s: %v", arn, err)
	}
}

func registerTargets(t *testing.T, c *harness.AWSClient, tgArn string, instanceIDs []string) {
	t.Helper()
	targets := make([]*elbv2.TargetDescription, len(instanceIDs))
	for i, id := range instanceIDs {
		targets[i] = &elbv2.TargetDescription{Id: aws.String(id)}
	}
	_, err := c.ELBv2.RegisterTargets(&elbv2.RegisterTargetsInput{
		TargetGroupArn: aws.String(tgArn),
		Targets:        targets,
	})
	require.NoError(t, err, "register-targets")
}

func deregisterTargets(t *testing.T, c *harness.AWSClient, tgArn string, instanceIDs []string) {
	if tgArn == "" || len(instanceIDs) == 0 {
		return
	}
	targets := make([]*elbv2.TargetDescription, len(instanceIDs))
	for i, id := range instanceIDs {
		targets[i] = &elbv2.TargetDescription{Id: aws.String(id)}
	}
	if _, err := c.ELBv2.DeregisterTargets(&elbv2.DeregisterTargetsInput{
		TargetGroupArn: aws.String(tgArn),
		Targets:        targets,
	}); err != nil {
		t.Logf("deregister-targets: %v", err)
	}
}

type lbInfo struct {
	ARN, ID, Scheme, Type, DNSName string
}

// createLB builds the CreateLoadBalancer request. sgIDs, when non-empty, is
// passed as the LB's SecurityGroups — for ALB scenarios this is a dedicated
// per-scenario SG (see createScenarioSG); nil for NLBs, which get their own
// managed SG automatically. Nil/empty SecurityGroups on an ALB falls back to
// the VPC's shared default SG, which is the multi-scenario contention this
// per-scenario SG threading avoids.
func createLB(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, lbType, scheme string, sgIDs []string) lbInfo {
	t.Helper()
	in := &elbv2.CreateLoadBalancerInput{
		Name:    aws.String(name),
		Subnets: []*string{aws.String(f.SubnetID)},
		Scheme:  aws.String(scheme),
	}
	if lbType == "network" {
		in.Type = aws.String("network")
	}
	if len(sgIDs) > 0 {
		in.SecurityGroups = aws.StringSlice(sgIDs)
	}
	out, err := c.ELBv2.CreateLoadBalancer(in)
	require.NoErrorf(t, err, "create-load-balancer %s", name)
	require.NotEmpty(t, out.LoadBalancers)
	lb := out.LoadBalancers[0]
	arn := aws.StringValue(lb.LoadBalancerArn)
	parts := strings.Split(arn, "/")
	info := lbInfo{
		ARN:     arn,
		ID:      parts[len(parts)-1],
		Scheme:  aws.StringValue(lb.Scheme),
		Type:    aws.StringValue(lb.Type),
		DNSName: aws.StringValue(lb.DNSName),
	}
	t.Logf("LB %s: %s (scheme=%s type=%s dns=%s)", name, info.ARN, info.Scheme, info.Type, info.DNSName)
	return info
}

// lbCreateAttempts bounds retries when an LB lands in terminal state failed.
// On single-node dev boxes (serial subtests, one LB torn down before the next
// is created) this is usually the previous suite's sys.micro VM still being
// deallocated. On multinode clusters (subtests run concurrently, each owning
// an independent LB — see TestLoadBalancer) there is no such reclaim wait, so
// a terminal failure here reflects a genuine transient issue; the same retry
// serves as a defensive fallback either way.
const lbCreateAttempts = 3

// createActiveLB creates an LB (plus a listener when proto is non-empty) and
// waits for state=active. A terminal state failed is retried with backoff
// (see lbCreateAttempts) rather than failing the suite outright. Cleanups for
// the surviving LB/listener are registered on success.
func createActiveLB(t *testing.T, c *harness.AWSClient, f *sharedFixture, name, lbType, scheme, proto string, port int64, tgArn, label string, sgIDs []string) (lbInfo, string) {
	t.Helper()
	var lastErr error
	for attempt := 1; attempt <= lbCreateAttempts; attempt++ {
		if attempt > 1 {
			wait := time.Duration(attempt-1) * 15 * time.Second
			t.Logf("%s: waiting %s before retry (serial-mode capacity reclaim, or a transient failure)", label, wait)
			time.Sleep(wait)
		}
		lb := createLB(t, c, f, name, lbType, scheme, sgIDs)
		// Register idempotent teardown up-front so no exit path (fatal, timeout,
		// or success) can leak the LB/listener into the shared VPC. Deletes are
		// idempotent server-side, so a retry's explicit teardown below is safe.
		t.Cleanup(func() { deleteLB(t, c, lb) })
		listener := ""
		if proto != "" {
			listener = createListener(t, c, lb.ARN, proto, port, tgArn)
			t.Cleanup(func() { deleteListener(t, c, listener) })
		}
		lastErr = harness.WaitForLBActiveErr(t, c, lb.ARN, label, 5*time.Minute)
		if lastErr == nil {
			return lb, listener
		}
		if !errors.Is(lastErr, harness.ErrLBTerminalFailed) && !errors.Is(lastErr, harness.ErrLBProvisioningTimeout) {
			t.Fatalf("%s: %v", label, lastErr)
		}
		t.Logf("%s: attempt %d/%d: %v — tearing down and retrying", label, attempt, lbCreateAttempts, lastErr)
		deleteListener(t, c, listener)
		deleteLB(t, c, lb)
	}
	t.Fatalf("%s: LB never reached active after %d attempts: %v", label, lbCreateAttempts, lastErr)
	return lbInfo{}, ""
}

func deleteLB(t *testing.T, c *harness.AWSClient, lb lbInfo) {
	if lb.ARN == "" {
		return
	}
	prefix := "app"
	if lb.Type == "network" {
		prefix = "net"
	}
	parts := strings.Split(lb.ARN, "/")
	require.GreaterOrEqualf(t, len(parts), 3, "deleteLB: malformed LB ARN %q", lb.ARN)
	lbName := parts[len(parts)-2]
	filter := fmt.Sprintf("ELB %s/%s/%s", prefix, lbName, lb.ID)

	// Capture the underlying sys.micro VM id before deleting so we can wait
	// for it to actually terminate. ELBv2.DeleteLoadBalancer returns once
	// the LB resource is gone, but the system VM termination is async and
	// holds the vCPU/memory allocation until reaped — without this wait,
	// the next suite's createLB can race the capacity reclaim and trip the
	// reserve, ending up in terminal state=failed.
	var sysInstanceID string
	if eniOut, err := c.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
		Filters: []*ec2.Filter{{
			Name:   aws.String("description"),
			Values: []*string{aws.String(filter)},
		}},
	}); err == nil && len(eniOut.NetworkInterfaces) > 0 && eniOut.NetworkInterfaces[0].Attachment != nil {
		sysInstanceID = aws.StringValue(eniOut.NetworkInterfaces[0].Attachment.InstanceId)
	}

	if _, err := c.ELBv2.DeleteLoadBalancer(&elbv2.DeleteLoadBalancerInput{
		LoadBalancerArn: aws.String(lb.ARN),
	}); err != nil {
		t.Logf("delete LB %s: %v", lb.ARN, err)
		return
	}
	// Block until describe reports LoadBalancerNotFound — the daemon doesn't
	// mark the LB gone until the underlying sys.micro VM has been torn down
	// and its vCPU/memory deallocated. Polling this avoids the capacity race
	// where the next LB createLB fires before deallocation completes (sys
	// instances are filtered from DescribeInstances so WaitForInstanceTerminated
	// is a no-op for them).
	waitForLBGone(t, c, lb.ARN, 60*time.Second)
	harness.WaitForENICleanup(t, c, filter, lb.ARN, 30*time.Second)
	if sysInstanceID != "" {
		harness.WaitForInstanceTerminated(t, c, []string{sysInstanceID}, 60*time.Second)
	}
}

// waitForLBGone polls until DescribeLoadBalancers returns LoadBalancerNotFound,
// indicating the underlying system VM has terminated and released its capacity.
func waitForLBGone(t *testing.T, c *harness.AWSClient, arn string, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for {
		out, err := c.ELBv2.DescribeLoadBalancers(&elbv2.DescribeLoadBalancersInput{
			LoadBalancerArns: []*string{aws.String(arn)},
		})
		if err != nil {
			var aerr awserr.Error
			if errors.As(err, &aerr) && aerr.Code() == "LoadBalancerNotFound" {
				return
			}
		}
		if err == nil && len(out.LoadBalancers) == 0 {
			return
		}
		if time.Now().After(deadline) {
			t.Logf("waitForLBGone: %s still visible after %s (continuing)", arn, timeout)
			return
		}
		time.Sleep(2 * time.Second)
	}
}

func createListener(t *testing.T, c *harness.AWSClient, lbArn, proto string, port int64, tgArn string) string {
	t.Helper()
	out, err := c.ELBv2.CreateListener(&elbv2.CreateListenerInput{
		LoadBalancerArn: aws.String(lbArn),
		Protocol:        aws.String(proto),
		Port:            aws.Int64(port),
		DefaultActions: []*elbv2.Action{{
			Type:           aws.String("forward"),
			TargetGroupArn: aws.String(tgArn),
		}},
	})
	require.NoError(t, err, "create-listener")
	arn := aws.StringValue(out.Listeners[0].ListenerArn)
	t.Logf("listener %s:%d -> %s", proto, port, arn)
	return arn
}

func deleteListener(t *testing.T, c *harness.AWSClient, arn string) {
	if arn == "" {
		return
	}
	if _, err := c.ELBv2.DeleteListener(&elbv2.DeleteListenerInput{ListenerArn: aws.String(arn)}); err != nil {
		t.Logf("delete listener: %v", err)
	}
}

// modifyListenerPort changes the listener's port in place. Verifies that
// HAProxy is reconciled without the listener being deleted (no traffic outage).
func modifyListenerPort(t *testing.T, c *harness.AWSClient, listenerArn string, port int64) {
	t.Helper()
	_, err := c.ELBv2.ModifyListener(&elbv2.ModifyListenerInput{
		ListenerArn: aws.String(listenerArn),
		Port:        aws.Int64(port),
	})
	require.NoErrorf(t, err, "modify-listener port=%d", port)
	t.Logf("listener %s: port → %d", listenerArn, port)
}

// modifyListenerDefaultTG swaps the listener's default target group.
func modifyListenerDefaultTG(t *testing.T, c *harness.AWSClient, listenerArn, tgArn string) {
	t.Helper()
	_, err := c.ELBv2.ModifyListener(&elbv2.ModifyListenerInput{
		ListenerArn: aws.String(listenerArn),
		DefaultActions: []*elbv2.Action{{
			Type:           aws.String("forward"),
			TargetGroupArn: aws.String(tgArn),
		}},
	})
	require.NoError(t, err, "modify-listener default actions")
	t.Logf("listener %s: default TG → %s", listenerArn, tgArn)
}

// describeListenerPort fetches the current port of a listener — used to
// assert that ModifyListener actually persisted before probing.
func describeListenerPort(t *testing.T, c *harness.AWSClient, listenerArn string) int64 {
	t.Helper()
	out, err := c.ELBv2.DescribeListeners(&elbv2.DescribeListenersInput{
		ListenerArns: []*string{aws.String(listenerArn)},
	})
	require.NoError(t, err, "describe-listeners")
	require.NotEmpty(t, out.Listeners)
	return aws.Int64Value(out.Listeners[0].Port)
}

// runModifyListenerSuite verifies in-place listener edits: port change and DefaultActions swap.
// Regression gate: the listener must stay alive across edits (not delete+recreate).
func runModifyListenerSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	const label = "ALB internal ModifyListener"
	const altPort int64 = 8090

	tgA := createTargetGroup(t, c, f, "lb-e2e-mod-tg-a", "HTTP", httpPort, "/index.html")
	t.Cleanup(func() { deleteTargetGroup(t, c, tgA) })

	tgB := createTargetGroup(t, c, f, "lb-e2e-mod-tg-b", "HTTP", httpPort, "/index.html")
	t.Cleanup(func() { deleteTargetGroup(t, c, tgB) })

	registerTargets(t, c, tgA, f.AppInstanceIDs)
	t.Cleanup(func() { deregisterTargets(t, c, tgA, f.AppInstanceIDs) })
	registerTargets(t, c, tgB, f.AppInstanceIDs)
	t.Cleanup(func() { deregisterTargets(t, c, tgB, f.AppInstanceIDs) })

	// Both the original listener port and the port it's modified to (altPort)
	// must be open on this scenario's own dedicated SG up front.
	sgIDs := []string{createScenarioSG(t, c, f, "lb-e2e-mod-sg", httpPort, altPort)}

	lb, listener := createActiveLB(t, c, f, "lb-e2e-mod", "application", "internal", "HTTP", httpPort, tgA, label, sgIDs)
	harness.WaitForTargetsHealthy(t, c, tgA, 2, label+" tgA", 2*time.Minute)

	eni := lbENI(t, c, "app", lb)
	priv := privateIP(eni)
	require.NotEmpty(t, priv, label+" needs private IP")

	// Baseline: traffic on the original port.
	probeAtPort(t, f, priv, httpPort, label+" before modify (port 80)")

	// In-place port change.
	modifyListenerPort(t, c, listener, altPort)
	assert.Equal(t, altPort, describeListenerPort(t, c, listener), "listener port persisted")
	// lb-agent picks up the new HAProxy config on its next 5s tick, not
	// synchronously with the ModifyListener API. Poll until the listener
	// actually answers on altPort before driving the assertive probe.
	waitForListenerServing(t, f, priv, altPort, label+" port 8090", 60*time.Second)
	probeAtPort(t, f, priv, altPort, label+" after port-modify (port 8090)")

	// Defaults: in-place TG swap on same port.
	modifyListenerDefaultTG(t, c, listener, tgB)
	harness.WaitForTargetsHealthy(t, c, tgB, 2, label+" tgB (post-swap)", 2*time.Minute)
	probeAtPort(t, f, priv, altPort, label+" after TG-swap (still port 8090, now tgB)")
}

// runListenerRulesSuite exercises ALB listener rules: CreateRule, ModifyRule, DescribeRules,
// and DeleteRule. Two TGs with disjoint backends so probes confirm which TG served.
// Also tests host-header condition to verify a second ACL field reaches HAProxy.
func runListenerRulesSuite(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	const label = "ALB internal ListenerRules"

	require.Len(t, f.AppInstanceIDs, 2, label+" needs 2 app instances")
	appA, appB := f.AppInstanceIDs[0], f.AppInstanceIDs[1]

	tgA := createTargetGroup(t, c, f, "lb-e2e-rul-tg-a", "HTTP", httpPort, "/index.html")
	t.Cleanup(func() { deleteTargetGroup(t, c, tgA) })
	tgB := createTargetGroup(t, c, f, "lb-e2e-rul-tg-b", "HTTP", httpPort, "/alpha/index.html")
	t.Cleanup(func() { deleteTargetGroup(t, c, tgB) })

	registerTargets(t, c, tgA, []string{appA})
	t.Cleanup(func() { deregisterTargets(t, c, tgA, []string{appA}) })
	registerTargets(t, c, tgB, []string{appB})
	t.Cleanup(func() { deregisterTargets(t, c, tgB, []string{appB}) })

	sgIDs := []string{createScenarioSG(t, c, f, "lb-e2e-rul-sg", httpPort)}

	lb, listener := createActiveLB(t, c, f, "lb-e2e-rul", "application", "internal", "HTTP", httpPort, tgA, label, sgIDs)
	harness.WaitForTargetsHealthy(t, c, tgA, 1, label+" tgA", 2*time.Minute)

	eni := lbENI(t, c, "app", lb)
	priv := privateIP(eni)
	require.NotEmpty(t, priv, label+" needs private IP")

	waitForPathServing(t, f, priv, httpPort, "/", "", label+" baseline", 60*time.Second)
	base := probeAtPath(t, f, priv, httpPort, "/", "", label+" default -> tgA")
	require.Equal(t, 1, base.Unique(), label+" default expects 1 responder (tgA single backend)")
	appAHost := singleResponder(base)

	// CreateRule wires tgB to the listener; only then does spinifex begin
	// health-checking tgB. Wait for healthy AFTER rule creation.
	ruleArn := createPathRule(t, c, listener, 10, "/alpha*", tgB)
	ruleCleanup := func() {
		if ruleArn == "" {
			return
		}
		deleteRule(t, c, ruleArn)
	}
	t.Cleanup(ruleCleanup)

	harness.WaitForTargetsHealthy(t, c, tgB, 1, label+" tgB (post-rule)", 2*time.Minute)
	waitForPathRoutedAway(t, f, priv, httpPort, "/alpha/", appAHost, label+" wait rule active", 60*time.Second)
	ruleResp := probeAtPath(t, f, priv, httpPort, "/alpha/", "", label+" path /alpha/ -> tgB")
	require.Equal(t, 1, ruleResp.Unique(), label+" rule probe expects 1 responder")
	appBHost := singleResponder(ruleResp)
	require.NotEqual(t, appAHost, appBHost, label+" /alpha/ must route to tgB, not default")

	// Default route unaffected by rule.
	stillDefault := probeAtPath(t, f, priv, httpPort, "/", "", label+" default unchanged")
	require.Equal(t, 1, stillDefault.Unique())
	assert.Equal(t, appAHost, singleResponder(stillDefault), label+" / must still hit tgA")

	// DescribeRules: rule + default present.
	rules := describeRules(t, c, listener)
	assert.GreaterOrEqual(t, len(rules), 2, label+" expect rule + default")
	assert.True(t, hasRule(rules, ruleArn), label+" CreateRule arn must appear in DescribeRules")

	// ModifyRule: /alpha* -> /beta*. /alpha now falls back to default;
	// /beta* now reaches tgB (appB).
	modifyPathRule(t, c, ruleArn, "/beta*")
	waitForPathRoutedTo(t, f, priv, httpPort, "/beta/", appBHost, label+" wait modify active", 60*time.Second)
	modified := probeAtPath(t, f, priv, httpPort, "/alpha/", "", label+" /alpha/ falls to default after modify")
	require.Equal(t, 1, modified.Unique())
	assert.Equal(t, appAHost, singleResponder(modified), label+" /alpha/ should hit default after pattern change")

	// /beta/ now routes to tgB.
	betaResp := probeAtPath(t, f, priv, httpPort, "/beta/", "", label+" /beta/ -> tgB")
	require.Equal(t, 1, betaResp.Unique())
	assert.Equal(t, appBHost, singleResponder(betaResp), label+" /beta/ must hit tgB after modify")

	// DeleteRule: /beta also falls to default.
	deleteRule(t, c, ruleArn)
	ruleArn = ""
	waitForPathRoutedAway(t, f, priv, httpPort, "/beta/", appBHost, label+" wait delete active", 60*time.Second)
	afterDelete := probeAtPath(t, f, priv, httpPort, "/beta/", "", label+" /beta/ after delete")
	require.Equal(t, 1, afterDelete.Unique())
	assert.Equal(t, appAHost, singleResponder(afterDelete), label+" /beta/ must hit default after delete")
}

// createPathRule adds a path-pattern listener rule. Returns the rule ARN.
func createPathRule(t *testing.T, c *harness.AWSClient, listenerArn string, priority int64, pattern, tgArn string) string {
	t.Helper()
	out, err := c.ELBv2.CreateRule(&elbv2.CreateRuleInput{
		ListenerArn: aws.String(listenerArn),
		Priority:    aws.Int64(priority),
		Conditions: []*elbv2.RuleCondition{{
			Field: aws.String("path-pattern"),
			PathPatternConfig: &elbv2.PathPatternConditionConfig{
				Values: []*string{aws.String(pattern)},
			},
		}},
		Actions: []*elbv2.Action{{
			Type:           aws.String("forward"),
			TargetGroupArn: aws.String(tgArn),
		}},
	})
	require.NoErrorf(t, err, "create-rule priority=%d pattern=%s", priority, pattern)
	require.NotEmpty(t, out.Rules)
	arn := aws.StringValue(out.Rules[0].RuleArn)
	t.Logf("rule %s: priority=%d path=%s -> %s", arn, priority, pattern, tgArn)
	return arn
}

func modifyPathRule(t *testing.T, c *harness.AWSClient, ruleArn, pattern string) {
	t.Helper()
	_, err := c.ELBv2.ModifyRule(&elbv2.ModifyRuleInput{
		RuleArn: aws.String(ruleArn),
		Conditions: []*elbv2.RuleCondition{{
			Field: aws.String("path-pattern"),
			PathPatternConfig: &elbv2.PathPatternConditionConfig{
				Values: []*string{aws.String(pattern)},
			},
		}},
	})
	require.NoErrorf(t, err, "modify-rule pattern=%s", pattern)
	t.Logf("rule %s: path -> %s", ruleArn, pattern)
}

func deleteRule(t *testing.T, c *harness.AWSClient, ruleArn string) {
	if ruleArn == "" {
		return
	}
	if _, err := c.ELBv2.DeleteRule(&elbv2.DeleteRuleInput{RuleArn: aws.String(ruleArn)}); err != nil {
		t.Logf("delete rule %s: %v", ruleArn, err)
	}
}

func describeRules(t *testing.T, c *harness.AWSClient, listenerArn string) []*elbv2.Rule {
	t.Helper()
	out, err := c.ELBv2.DescribeRules(&elbv2.DescribeRulesInput{
		ListenerArn: aws.String(listenerArn),
	})
	require.NoError(t, err, "describe-rules")
	return out.Rules
}

func hasRule(rules []*elbv2.Rule, arn string) bool {
	for _, r := range rules {
		if aws.StringValue(r.RuleArn) == arn {
			return true
		}
	}
	return false
}

func singleResponder(r harness.TrafficResult) string {
	for k := range r.Distribution {
		return k
	}
	return ""
}

// probeAtPath drives the shared client to issue probesPerRun probes at lbIP:port with the
// given path and optional Host header. Asserts at least half succeed and returns the
// distribution so callers can assert which backend(s) served.
func probeAtPath(t *testing.T, f *sharedFixture, lbIP string, port int64, path, host, label string) harness.TrafficResult {
	t.Helper()
	r, err := probeOnceAt(f, lbIP, port, path, host, probesPerRun)
	require.NoErrorf(t, err, "trigger probe %s", label)
	for inst, count := range r.Distribution {
		t.Logf("  %s: %s -> %d", label, inst, count)
	}
	t.Logf("  %s: %d/%d successful, %d unique", label, r.Successful, r.Total, r.Unique())
	require.GreaterOrEqualf(t, r.Successful, probesPerRun/2, "%s probes succeeded", label)
	return r
}

// waitForPathServing polls until the path/host combination returns any
// successful response. Use after CreateRule/ModifyRule/DeleteRule to bridge
// the lb-agent reconcile window.
func waitForPathServing(t *testing.T, f *sharedFixture, lbIP string, port int64, path, host, label string, timeout time.Duration) {
	t.Helper()
	harness.EventuallyErr(t, func() error {
		r, err := probeOnceAt(f, lbIP, port, path, host, 2)
		if err != nil {
			return err
		}
		if r.Successful == 0 {
			return fmt.Errorf("%s: 0/%d successful", label, r.Total)
		}
		return nil
	}, timeout, 2*time.Second)
}

// waitForPathRoutedAway polls a path until responses stop coming from awayFrom.
// Use after CreateRule/ModifyRule/DeleteRule to wait out lb-agent reconcile.
// Requires at least one successful probe to avoid passing on a transient outage.
func waitForPathRoutedAway(t *testing.T, f *sharedFixture, lbIP string, port int64, path, awayFrom, label string, timeout time.Duration) {
	t.Helper()
	harness.EventuallyErr(t, func() error {
		r, err := probeOnceAt(f, lbIP, port, path, "", 2)
		if err != nil {
			return err
		}
		if r.Successful == 0 {
			return fmt.Errorf("%s: 0/%d successful", label, r.Total)
		}
		if r.Distribution[awayFrom] > 0 {
			return fmt.Errorf("%s: still seeing %s", label, awayFrom)
		}
		return nil
	}, timeout, 2*time.Second)
}

// waitForPathRoutedTo polls a path until responses come from wantHost.
// Use when the caller already knows the expected backend hostname.
func waitForPathRoutedTo(t *testing.T, f *sharedFixture, lbIP string, port int64, path, wantHost, label string, timeout time.Duration) {
	t.Helper()
	harness.EventuallyErr(t, func() error {
		r, err := probeOnceAt(f, lbIP, port, path, "", 2)
		if err != nil {
			return err
		}
		if r.Distribution[wantHost] == 0 {
			return fmt.Errorf("%s: %s not yet observed", label, wantHost)
		}
		return nil
	}, timeout, 2*time.Second)
}

// probeOnceAt is probeOnce with explicit path + Host-header support. Returns
// errors rather than failing so callers can poll.
func probeOnceAt(f *sharedFixture, lbIP string, port int64, path, host string, count int) (harness.TrafficResult, error) {
	resultsFile := fmt.Sprintf("rul-%d-%d.txt", port, time.Now().UnixNano())
	order := map[string]any{
		"proto":     "http",
		"ip":        lbIP,
		"count":     count,
		"outfile":   resultsFile,
		"http_port": port,
		"http_path": path,
		"host":      host,
		"tcp_port":  tcpPort,
	}
	body, err := json.Marshal(order)
	if err != nil {
		return harness.TrafficResult{}, err
	}
	triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
	client := &http.Client{Timeout: 90 * time.Second}
	resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
	if err != nil {
		return harness.TrafficResult{}, fmt.Errorf("trigger: %w", err)
	}
	resp.Body.Close()
	if resp.StatusCode != 200 {
		return harness.TrafficResult{}, fmt.Errorf("trigger status %d", resp.StatusCode)
	}
	results, err := plainHTTPGet(
		fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
		10*time.Second)
	if err != nil {
		return harness.TrafficResult{}, fmt.Errorf("fetch %s: %w", resultsFile, err)
	}
	return harness.VerifyResultsLines(results, "http"), nil
}

// waitForListenerServing polls the shared client until at least one probe returns
// a successful instance_id payload. Used after ModifyListener to avoid racing
// the lb-agent's poll-driven HAProxy reload (5s tick + reload latency).
func waitForListenerServing(t *testing.T, f *sharedFixture, lbIP string, port int64, label string, timeout time.Duration) {
	t.Helper()
	harness.EventuallyErr(t, func() error {
		r, err := probeOnce(f, lbIP, port, 2)
		if err != nil {
			return err
		}
		if r.Successful == 0 {
			return fmt.Errorf("%s: listener not serving on :%d (0/%d successful)", label, port, r.Total)
		}
		return nil
	}, timeout, 2*time.Second)
}

// probeOnce drives the shared client to issue `count` HTTP probes at
// lbIP:port and returns the parsed TrafficResult. Errors are returned rather
// than failing the test so callers can poll.
func probeOnce(f *sharedFixture, lbIP string, port int64, count int) (harness.TrafficResult, error) {
	resultsFile := fmt.Sprintf("mod-%d-%d.txt", port, time.Now().UnixNano())
	order := map[string]any{
		"proto":     "http",
		"ip":        lbIP,
		"count":     count,
		"outfile":   resultsFile,
		"http_port": port,
		"tcp_port":  tcpPort,
	}
	body, err := json.Marshal(order)
	if err != nil {
		return harness.TrafficResult{}, err
	}
	triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
	client := &http.Client{Timeout: 90 * time.Second}
	resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
	if err != nil {
		return harness.TrafficResult{}, fmt.Errorf("trigger: %w", err)
	}
	resp.Body.Close()
	if resp.StatusCode != 200 {
		return harness.TrafficResult{}, fmt.Errorf("trigger status %d", resp.StatusCode)
	}
	results, err := plainHTTPGet(
		fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
		10*time.Second)
	if err != nil {
		return harness.TrafficResult{}, fmt.Errorf("fetch %s: %w", resultsFile, err)
	}
	return harness.VerifyResultsLines(results, "http"), nil
}

// probeAtPort runs round-robin probes from the shared client to lbIP:port and
// asserts both app instances respond. Mirrors runInternalTrafficViaClient but
// takes an explicit port so ModifyListener edits can be verified.
func probeAtPort(t *testing.T, f *sharedFixture, lbIP string, port int64, label string) {
	t.Helper()
	resultsFile := fmt.Sprintf("mod-%d-%d.txt", port, time.Now().UnixNano())
	order := map[string]any{
		"proto":     "http",
		"ip":        lbIP,
		"count":     probesPerRun,
		"outfile":   resultsFile,
		"http_port": port,
		"tcp_port":  tcpPort,
	}
	body, err := json.Marshal(order)
	require.NoError(t, err)

	triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
	client := &http.Client{Timeout: 90 * time.Second}
	resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
	require.NoErrorf(t, err, "trigger probe %s", label)
	resp.Body.Close()
	require.Equalf(t, 200, resp.StatusCode, "trigger probe %s status", label)

	results, err := plainHTTPGet(
		fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
		10*time.Second)
	require.NoErrorf(t, err, "fetch %s", resultsFile)
	harness.AssertRoundRobin(t,
		harness.VerifyResultsLines(results, "http"),
		1, probesPerRun/2, label)
}

func lbENI(t *testing.T, c *harness.AWSClient, prefix string, lb lbInfo) *ec2.NetworkInterface {
	t.Helper()
	parts := strings.Split(lb.ARN, "/")
	require.GreaterOrEqualf(t, len(parts), 3, "lbENI: malformed LB ARN %q", lb.ARN)
	lbName := parts[len(parts)-2]
	desc := fmt.Sprintf("ELB %s/%s/%s", prefix, lbName, lb.ID)
	var eni *ec2.NetworkInterface
	harness.EventuallyErr(t, func() error {
		out, err := c.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
			Filters: []*ec2.Filter{{
				Name:   aws.String("description"),
				Values: []*string{aws.String(desc)},
			}},
		})
		if err != nil {
			return err
		}
		if len(out.NetworkInterfaces) == 0 {
			return fmt.Errorf("no ENI for %s", desc)
		}
		eni = out.NetworkInterfaces[0]
		return nil
	}, 30*time.Second, 2*time.Second)
	return eni
}

// captureLBConsoleOnFailure registers an on-failure dump of the NLB VM's serial console.
// nginx activation errors land on the guest console, not the host journal — so an
// "0/N healthy" timeout is otherwise undiagnosable from CI artifacts.
func captureLBConsoleOnFailure(t *testing.T, c *harness.AWSClient, eniDescPrefix string, lb lbInfo) {
	t.Helper()
	eni := lbENI(t, c, eniDescPrefix, lb)
	if eni.Attachment == nil {
		return
	}
	instanceID := aws.StringValue(eni.Attachment.InstanceId)
	if instanceID == "" {
		return
	}
	dir := harness.ArtifactDir(t, harness.LoadEnv(t))
	harness.OnFailure(t, func() {
		harness.DumpInstanceConsole(t, c, instanceID, dir, "lb-vm-console.log")
	})
}

func publicIP(eni *ec2.NetworkInterface) string {
	if eni == nil || eni.Association == nil {
		return ""
	}
	return aws.StringValue(eni.Association.PublicIp)
}

func privateIP(eni *ec2.NetworkInterface) string {
	if eni == nil {
		return ""
	}
	return aws.StringValue(eni.PrivateIpAddress)
}

func assertInternalDNS(t *testing.T, c *harness.AWSClient, lbArn, label string) {
	t.Helper()
	out, err := c.ELBv2.DescribeLoadBalancers(&elbv2.DescribeLoadBalancersInput{
		LoadBalancerArns: []*string{aws.String(lbArn)},
	})
	require.NoError(t, err)
	require.NotEmpty(t, out.LoadBalancers)
	dns := aws.StringValue(out.LoadBalancers[0].DNSName)
	assert.True(t, strings.HasPrefix(dns, "internal-"), "%s internal DNS missing internal- prefix: %s", label, dns)
}

// assertLBDNSResolves confirms the SDK-returned DNSName resolves to the LB's
// frontend IP through the host resolver (the same path an AWS SDK/CLI client
// uses). The suite requires Northstar, so an empty or legacy name is a failure.
// Retries because the control-plane writer publishes the record asynchronously
// (best-effort + reconcile).
func assertLBDNSResolves(t *testing.T, dnsName, wantIP, label string) {
	t.Helper()
	require.NotEmptyf(t, dnsName, "%s: load balancer returned no DNS name despite required Northstar DNS", label)
	require.Falsef(t, strings.HasSuffix(dnsName, ".spinifex.local"),
		"%s: load balancer returned legacy DNS name %q despite required Northstar DNS", label, dnsName)
	harness.Step(t, "resolve LB DNS %s → %s (northstar path)", dnsName, wantIP)
	deadline := time.Now().Add(90 * time.Second)
	var last []string
	for time.Now().Before(deadline) {
		addrs, err := net.LookupHost(dnsName)
		if err == nil {
			last = addrs
			for _, a := range addrs {
				if a == wantIP {
					return
				}
			}
		}
		time.Sleep(3 * time.Second)
	}
	t.Fatalf("%s: LB DNS %q never resolved to frontend IP %s within 90s (last=%v) — northstar did not serve the A record",
		label, dnsName, wantIP, last)
}

// --- Internet-facing traffic ---------------------------------------------

// runInternetFacingTrafficSingle drives one LB (ALB or NLB) over its public
// IP. Runs probes locally from the test driver, and if a peer node is wired
// up, also runs the same probes from there to exercise inter-node routing.
func runInternetFacingTrafficSingle(t *testing.T, kind lbKind, ssh *harness.PeerSSH, peer, ip string) {
	t.Helper()
	if kind == kindALB {
		url := fmt.Sprintf("http://%s:%d", ip, httpPort)
		harness.AssertRoundRobin(t,
			harness.HTTPRoundRobin(url, probesPerRun, 5*time.Second),
			2, probesPerRun/2, "ALB inet (local)")
		if ssh != nil && peer != "" {
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
			defer cancel()
			harness.AssertRoundRobin(t,
				remoteHTTPRoundRobin(t, ctx, ssh, peer, url, probesPerRun),
				2, probesPerRun/2, "ALB inet (remote)")
		}
		return
	}
	harness.AssertRoundRobin(t,
		harness.TCPRoundRobin(ip, tcpPort, probesPerRun, 5*time.Second),
		1, probesPerRun/2, "NLB inet (local)")
	if ssh != nil && peer != "" {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		harness.AssertRoundRobin(t,
			remoteTCPRoundRobin(t, ctx, ssh, peer, ip, tcpPort, probesPerRun),
			1, probesPerRun/2, "NLB inet (remote)")
	}
}

func remoteHTTPRoundRobin(t *testing.T, ctx context.Context, ssh *harness.PeerSSH, peer, url string, n int) harness.TrafficResult {
	t.Helper()
	var lines []string
	for i := 0; i < n; i++ {
		out, err := ssh.Run(ctx, peer, fmt.Sprintf("curl -s --max-time 5 '%s/'", url))
		if err != nil {
			continue
		}
		lines = append(lines, strings.TrimSpace(string(out)))
	}
	return harness.VerifyResultsLines(strings.Join(lines, "\n"), "http")
}

func remoteTCPRoundRobin(t *testing.T, ctx context.Context, ssh *harness.PeerSSH, peer, host string, port, n int) harness.TrafficResult {
	t.Helper()
	var lines []string
	for i := 0; i < n; i++ {
		out, err := ssh.Run(ctx, peer, fmt.Sprintf("echo '' | nc -w5 '%s' %d", host, port))
		if err != nil {
			continue
		}
		lines = append(lines, strings.TrimSpace(string(out)))
	}
	return harness.VerifyResultsLines(strings.Join(lines, "\n"), "tcp")
}

func runNLBDeregisterDraining(t *testing.T, c *harness.AWSClient, tgArn, targetID string) {
	t.Helper()
	_, err := c.ELBv2.DeregisterTargets(&elbv2.DeregisterTargetsInput{
		TargetGroupArn: aws.String(tgArn),
		Targets:        []*elbv2.TargetDescription{{Id: aws.String(targetID)}},
	})
	require.NoError(t, err)

	time.Sleep(3 * time.Second)
	out, err := c.ELBv2.DescribeTargetHealth(&elbv2.DescribeTargetHealthInput{TargetGroupArn: aws.String(tgArn)})
	require.NoError(t, err)

	remaining := len(out.TargetHealthDescriptions)
	draining := 0
	for _, th := range out.TargetHealthDescriptions {
		if aws.StringValue(th.TargetHealth.State) == "draining" {
			draining++
		}
	}
	t.Logf("NLB deregister: %d remaining, %d draining", remaining, draining)
	assert.True(t, remaining == 1 || draining >= 1, "NLB deregister: expected 1 remaining or >=1 draining")
}

// --- Shared probe client + per-suite trigger ----------------------------

// launchSharedProbeClient launches one client VM whose user-data exposes http.server on :80
// and a JSON trigger endpoint on :9090. Both internal suites share the same client
// to avoid the ~60s cold-boot cost of a per-suite client.
func launchSharedProbeClient(t *testing.T, c *harness.AWSClient, f *sharedFixture) {
	t.Helper()
	out, err := c.EC2.RunInstances(&ec2.RunInstancesInput{
		ImageId:      aws.String(f.AMIID),
		InstanceType: aws.String(f.InstanceType),
		KeyName:      aws.String(lbKeyName),
		SubnetId:     aws.String(f.SubnetID),
		MinCount:     aws.Int64(1),
		MaxCount:     aws.Int64(1),
		UserData:     aws.String(base64Encode(clientUserData)),
	})
	require.NoError(t, err, "run-instances probe client")
	f.ClientID = aws.StringValue(out.Instances[0].InstanceId)
	t.Logf("probe client: %s", f.ClientID)

	harness.WaitForInstanceRunning(t, c, f.ClientID, 120*time.Second)
	eni := harness.InstanceENI(t, c, f.ClientID)
	f.ClientPublicIP = publicIP(eni)
	require.NotEmpty(t, f.ClientPublicIP, "probe client needs public IP")
	t.Logf("probe client public IP: %s", f.ClientPublicIP)

	// Wait until the trigger server is accepting connections before the
	// first suite starts firing probes at it.
	triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
	harness.Eventually(t, func() bool {
		req, _ := http.NewRequest(http.MethodGet, triggerURL, nil)
		client := &http.Client{Timeout: 3 * time.Second}
		resp, err := client.Do(req)
		if err != nil {
			return false
		}
		resp.Body.Close()
		return true
	}, 5*time.Minute, 5*time.Second, "probe client trigger server not ready")
}

// runInternalTrafficViaClient POSTs a probe order to the shared client's
// trigger server, then fetches the results file it wrote and parses it.
func runInternalTrafficViaClient(t *testing.T, _ *harness.AWSClient, f *sharedFixture, kind lbKind, lbIP string) {
	t.Helper()
	var proto, resultsFile string
	if kind == kindALB {
		proto, resultsFile = "http", fmt.Sprintf("alb-%d.txt", time.Now().UnixNano())
	} else {
		proto, resultsFile = "tcp", fmt.Sprintf("nlb-%d.txt", time.Now().UnixNano())
	}

	order := map[string]any{
		"proto":     proto,
		"ip":        lbIP,
		"count":     probesPerRun,
		"outfile":   resultsFile,
		"http_port": httpPort,
		"tcp_port":  tcpPort,
	}
	body, err := json.Marshal(order)
	require.NoError(t, err)

	triggerURL := fmt.Sprintf("http://%s:%d/", f.ClientPublicIP, triggerPort)
	client := &http.Client{Timeout: 90 * time.Second}
	resp, err := client.Post(triggerURL, "application/json", bytes.NewReader(body))
	require.NoErrorf(t, err, "trigger %s probe", kind)
	resp.Body.Close()
	require.Equalf(t, 200, resp.StatusCode, "trigger %s probe HTTP status", kind)

	results, err := plainHTTPGet(
		fmt.Sprintf("http://%s:%d/%s", f.ClientPublicIP, httpPort, resultsFile),
		10*time.Second)
	require.NoErrorf(t, err, "fetch %s", resultsFile)
	harness.AssertRoundRobin(t,
		harness.VerifyResultsLines(results, proto),
		1, probesPerRun/2, string(kind)+" internal")
}

// plainHTTPGet fetches a plain-HTTP URL (no TLS). The probe client serves
// over port 80 with no cert, so the harness HTTPSGet helper would refuse.
func plainHTTPGet(url string, timeout time.Duration) (string, error) {
	client := &http.Client{Timeout: timeout}
	resp, err := client.Get(url)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("GET %s: status %d", url, resp.StatusCode)
	}
	buf := new(bytes.Buffer)
	if _, err := buf.ReadFrom(resp.Body); err != nil {
		return "", err
	}
	return buf.String(), nil
}

func dumpDaemonLogs(t *testing.T, dir, label string) {
	t.Helper()
	harness.DumpCmd(t, dir, fmt.Sprintf("daemon-%s.log", label),
		"journalctl", "-u", "spinifex-daemon", "--no-pager", "-n", "200")
}

// assertLBAgentAuthViaIMDS proves the LB VM authenticated to the gateway with
// IMDS-served instance-role credentials rather than baked static keys. The LB VM
// is a system-account instance plugged into a customer-account ENI, so it is
// invisible to the customer AWS API; the gateway journal on the cluster nodes is
// the only surface that observes its principal. The awsgw hosts the in-process
// IMDS STS responder, which logs "AssumeRoleForInstance success" at INFO when the
// LB VM's IMDS request mints assumed-role (ASIA) creds for the spinifex-lb-agent
// role — a regression to the three boot-path defects strands the agent on an
// empty role list, so no recent mint appears and this fails. Keying on that INFO
// line, not the gateway's per-request SigV4 debug log, keeps the check
// independent of the awsgw debug level. Skips when no node is SSH-reachable.
//
// The window is time-bounded on purpose: console logs accumulate across every VM
// the node has ever run (including pre-fix failures), so a flat-file scan for the
// old error would false-fail; the gateway's recent successes do not have that
// problem.
func assertLBAgentAuthViaIMDS(t *testing.T) {
	t.Helper()
	env := harness.LoadEnv(t)
	if len(env.NodeIPs) == 0 {
		t.Skip("no node IPs in env; cannot observe the LB VM gateway principal")
	}
	ssh := harness.NewPeerSSH()
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	sawAssumedRole, anyReachable := false, false
	for _, ip := range env.NodeIPs {
		gw, err := ssh.Run(ctx, ip,
			"sudo journalctl -u spinifex-awsgw --no-pager --since '5 minutes ago' 2>/dev/null | "+
				`grep -F 'AssumeRoleForInstance success' | grep -F 'role/spinifex-lb-agent' | grep -F '"akid":"ASIA' | tail -1`)
		if err != nil {
			continue // node unreachable; another node carries the heartbeat
		}
		anyReachable = true
		if strings.TrimSpace(string(gw)) != "" {
			sawAssumedRole = true
		}
	}
	if !anyReachable {
		t.Skip("no node SSH-reachable; cannot observe the LB VM gateway principal")
	}
	require.True(t, sawAssumedRole,
		"no assumed-role (ASIA) gateway auth on any node — LB agent did not authenticate via IMDS creds")
}

func base64Encode(s string) string {
	return base64.StdEncoding.EncodeToString([]byte(s))
}