//go:build functional

package spectests_test

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/cookiejar"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"slices"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"golang.org/x/net/html"
	"gopkg.in/yaml.v3"

	"github.com/tidbcloud/controlplane9/internal/config"
	"github.com/tidbcloud/controlplane9/internal/dbaasui"
	"github.com/tidbcloud/controlplane9/internal/sourcebacked"
	"github.com/tidbcloud/controlplane9/internal/tilt"
)

const (
	envEnabled          = "CP9_UI_API_ENABLED"
	envScenarios        = "CP9_UI_API_SCENARIOS"
	envSourceProfile    = "CP9_UI_API_AWS_SOURCE_PROFILE"
	envAccountID        = "CP9_UI_API_AWS_ACCOUNT_ID"
	envSubnetID         = "CP9_UI_API_AWS_SUBNET_ID"
	envKeyPair          = "CP9_UI_API_AWS_KEY_PAIR"
	envSSHPrivateKey    = "CP9_UI_API_SSH_PRIVATE_KEY"
	envCertificateARN   = "CP9_UI_API_CERTIFICATE_ARN"
	envDNSZoneID        = "CP9_UI_API_DNS_ZONE_ID"
	envDNSZoneName      = "CP9_UI_API_DNS_ZONE_NAME"
	envSessionName      = "CP9_UI_API_SESSION_NAME"
	envResume           = "CP9_UI_API_RESUME"
	envAdoptSession     = "CP9_UI_API_ADOPT_SESSION"
	envDestroySession   = "CP9_UI_API_DESTROY_SESSION"
	envDedicatedID      = "CP9_UI_API_DEDICATED_CLUSTER_ID"
	envPremiumID        = "CP9_UI_API_PREMIUM_CLUSTER_ID"
	envStarterID        = "CP9_UI_API_STARTER_CLUSTER_ID"
	envEssentialV1ID    = "CP9_UI_API_ESSENTIAL_V1_CLUSTER_ID"
	envEssentialV2ID    = "CP9_UI_API_ESSENTIAL_V2_CLUSTER_ID"
	envAuthToken        = "CP9_UI_API_AUTH_TOKEN"
	envDatabasePassword = "CP9_UI_API_DATABASE_PASSWORD"
	envDedicatedCloud   = "CP9_UI_API_DEDICATED_CLOUD"

	targetRegion             = "us-west-2"
	defaultAccount           = "804837308162"
	defaultProfile           = "dev-offline-afterworld"
	defaultGitHubAppClientID = "Iv23liKDo1S8LDZ07CZU"

	auth0Domain           = "tidb-dev.us.auth0.com"
	auth0ClientID         = "KiwWajLJ3pto8SsZ7F35NaqNhwnoRh1c"
	auth0Connection       = "custom-email"
	defaultUISeedEmail    = "sabaping@agent.qq.com"
	defaultUISeedPassword = "yjr6KFV2qty-rwv9npm"

	dedicatedOrgID     = "1440002"
	dedicatedProjectID = "1440002003"
	dedicatedCIDR      = "10.251.0.0/16"

	dedicatedPremiumCreateToActiveSLA = 45 * time.Minute
	starterFamilyCreateToActiveSLA    = 5 * time.Minute
	uiMetricsWaitTimeout              = 5 * time.Minute
	premiumMeteringWaitTimeout        = 15 * time.Minute
	// Bump when a journaled stage or value changes meaning incompatibly.
	publicUIRunJournalVersion = 2
)

const envPremiumCDCSourceBuild = "CP9_UI_API_PREMIUM_CDC_SOURCE_BUILD"

type testEnvironment struct {
	baseURL     string
	configPath  string
	cp9Binary   string
	dedicated   dedicatedTarget
	fixture     *testFixture
	premiumID   string
	repoRoot    string
	runJournal  *publicUIRunJournal
	sessionName string
	summary     *workflowSummary
	uiClient    *http.Client
}

type runtimeAccountContract struct {
	AccountID      string `json:"account_id"`
	CertificateARN string `json:"certificate_arn"`
	DNSZoneID      string `json:"dns_zone_id"`
	DNSZoneName    string `json:"dns_zone_name"`
}

type dedicatedTarget struct {
	CIDRProvider string
	Cloud        string
	Region       string
	RegionID     string
	StorageType  string
	TiDBNodeSpec string
	TiKVNodeSpec string
}

type testFixture struct {
	bucket        string
	importRoleARN string
	mu            sync.Mutex
	owner         string
	privateIP     string
	publicIP      string
	securityGroup string
	subnetID      string
	vpcID         string
}

type workflowSummary struct {
	mu   sync.Mutex
	path string
	t    *testing.T
}

type publicUIRunJournal struct {
	mu sync.Mutex `json:"-"`

	CurrentHead    string                          `json:"-"`
	Path           string                          `json:"-"`
	Scenarios      []string                        `json:"scenarios"`
	DedicatedCloud string                          `json:"dedicated_cloud"`
	Session        string                          `json:"session"`
	Stages         map[string]publicUIStageReceipt `json:"stages"`
	Values         map[string]string               `json:"values"`
	Version        int                             `json:"version"`
}

type publicUIStageReceipt struct {
	FinishedAt string `json:"finished_at,omitempty"`
	Head       string `json:"head"`
	StartedAt  string `json:"started_at"`
	Status     string `json:"status"`
}

type productDeletion struct {
	attempted bool
	created   bool
	request   func(*testing.T, context.Context)
}

func (d *productDeletion) run(t *testing.T, ctx context.Context) {
	t.Helper()
	if !d.created {
		t.Fatal("Product delete is not owned by this run")
	}
	if d.attempted {
		t.Fatal("Product delete was already attempted")
	}
	d.attempted = true
	d.request(t, ctx)
}

// TestFunctional_PublicUIAPI proves public UI API behavior through a real
// Remote EC2+k3d session. Run it only through the test-functional Make target,
// which owns the trigger and selection command. Without CP9_UI_API_ENABLED=1
// this test skips before creating resources.
//
// A fresh run builds cp9 from the current checkout, generates a unique session,
// runs setup/doctor/up --dry-run/up, logs in through the hosted DBAAS UI unless
// given a token, and runs the selected scenarios in parallel. Cleanup preserves
// the session unless separately authorized through CP9_UI_API_DESTROY_SESSION.
//
// Default run: set CP9_UI_API_ENABLED=1, CP9_UI_API_AWS_KEY_PAIR, and
// CP9_UI_API_SSH_PRIVATE_KEY. The test uses AWS account
// 804837308162, region us-west-2, source profile dev-offline-afterworld, the
// account's cp9-developer role and wildcard certificate, the first deterministic
// public subnet whose VPC has public subnets in two availability zones, and a
// caller-provided existing EC2 key pair plus matching private key. The test
// instruction is authority for these declared test mutations, but not for
// destroying the session. Required local tools are go, git, python3, ssh,
// ssh-keygen, curl, jq, kubectl, AWS CLI, and MySQL CLI, with refreshed SSO for
// dev-offline-afterworld.
//
// CP9_UI_API_AWS_SOURCE_PROFILE, CP9_UI_API_AWS_ACCOUNT_ID,
// CP9_UI_API_AWS_SUBNET_ID, CP9_UI_API_AWS_KEY_PAIR,
// CP9_UI_API_SSH_PRIVATE_KEY, and CP9_UI_API_CERTIFICATE_ARN may override the
// defaults together for a separately authorized account. The key-pair and
// private-key inputs are required together; CP9 validates and consumes the
// caller-owned key pair but never creates, imports, tags, rotates, or deletes it.
//
// CP9_UI_API_SCENARIOS accepts a comma-, whitespace-, or newline-separated
// subset of dedicated, premium, starter-e1, and essential-v2. It defaults to
// dedicated,premium; unknown values fail before mutation.
// CP9_UI_API_DEDICATED_CLOUD accepts aws (the default) or gcp. The GCP target
// enables the session's pingcap-gardener/us-west1 Regional runtime, verifies
// the additive AWS+GCP catalog through the public UI API, and runs only the
// first-slice Dedicated Create, strict-TLS SQL, UI Metrics, and Delete journey.
// Dedicated and
// Premium cover cluster creation/reuse, public whitelist and strict-TLS MySQL,
// PrivateLink, S3 import, changefeed, migration, backup, and restore with
// marker verification. starter-e1 activates Starter, then proves Starter and
// Essential v1 concurrently in the default organization. essential-v2
// activates its shared pool and proves Essential v2 in the second organization.
// Each serverless proof creates/reuses its instance, enables its public
// endpoint, sets an IP whitelist and password, runs strict-TLS MySQL
// create/write/read with Ssl_cipher verification, creates a real AWS VPC
// endpoint and runs private strict-TLS SQL, validates/prechecks/runs an S3
// import and verifies its row, then exports to S3 and verifies written objects.
// The run-owned Premium journey also records Harness evidence by decoding raw
// RU and metadata before backup/restore creates another Product, requiring
// Delta version 0 to advance in one configured table prefix, and selecting the
// new Parquet object to prove the four non-zero tenant 1440002 data/raft gp3
// IOPS/throughput rows. This internal check does not claim a customer-visible
// billing result.
// Immediately before its one Product DELETE, every Product created by this test
// uses the UI cookie-auth path to fetch O11Y metadata and query two real Metrics
// API series through the returned same-origin openReadUri, allowing five minutes
// for real datapoints to arrive. Backup-restore copies retain their SQL proof
// and one DELETE but do not repeat Metrics. A cluster
// supplied through a resume ID is caller-owned: its UI Metrics/Delete tail is
// skipped and not claimed.
// Starter and Essential v1/v2 must become ACTIVE within five minutes of
// Product Create; Dedicated and Premium (including Reserved) must do so within
// 45 minutes. The deadline is derived from the Product createTime, so a resume
// cannot reset the readiness budget.
//
// Set CP9_UI_API_PREMIUM_CDC_SOURCE_BUILD=1 with the premium scenario to build
// ticdc-service from a clean sibling cluster-service-ng checkout. The test
// resolves that checkout from the shipped source-backed contract, records its
// exact commit, and fails before AWS mutation when it is absent or dirty.
//
// CP9_UI_API_SESSION_NAME may set a unique DNS-compatible name of at most 15
// characters; otherwise the test generates one. Set CP9_UI_API_RESUME=1 with
// the same name, scenario selection, and retained local .cp9 state to skip
// setup/doctor/up. Resume requires the session-owned run journal and all AWS
// inputs because the session contract is reconstructed and validated. Passed
// stages are reused; running or failed stages fail closed when selected, so a
// same-scene continuation selects only later safe Go subtests and remains a
// partial verdict until a clean-head run. A pre-Product session created outside
// this test may additionally set CP9_UI_API_ADOPT_SESSION=1 to create its first
// empty journal. Adoption requires resume mode and a missing journal; it never
// imports Product state or overwrites an existing journal. Existing caller-owned
// resources may be supplied with:
//
//	CP9_UI_API_DEDICATED_CLUSTER_ID
//	CP9_UI_API_PREMIUM_CLUSTER_ID
//	CP9_UI_API_STARTER_CLUSTER_ID
//	CP9_UI_API_ESSENTIAL_V1_CLUSTER_ID
//	CP9_UI_API_ESSENTIAL_V2_CLUSTER_ID
//	CP9_UI_API_AUTH_TOKEN          skips hosted UI login
//	CP9_UI_API_DATABASE_PASSWORD   overrides the generated password
//
// The fixture stack, its resources, S3 objects, and VPC endpoints carry the
// exact cp9-owner/cp9-session tags and are never explicitly deleted here. The
// sole Cleanup boundary is cp9 destroy rc0 with its accepted asynchronous
// handoff; the functional test does not wait for or inspect cloud zero. The
// entire scene is preserved by default so first-red evidence remains resumable.
// Set CP9_UI_API_DESTROY_SESSION=1 only after a human, or an agent explicitly
// authorized in the task thread, has decided the exact session is safe to
// destroy. The caller-owned key remains outside CP9 cleanup. Destroy a
// preserved session through its exact resolved configuration:
//
//	bin/cp9 destroy config=.cp9/$CP9_UI_API_SESSION_NAME/cp9.resolved.yaml
//
// The proof timeout is ten hours and destroy gets 30 minutes. Product verdicts
// are recorded per selected scenario; cp9 destroy is a separate customer-visible
// result required to close the run. Harness or report-writing errors do not
// rewrite direct product evidence. A failure identifies the setup, fixture,
// scenario, or operation stage. The shared UI API trace records request and
// response bodies exactly as observed because this case runs only in the safe
// test environment. If GITHUB_STEP_SUMMARY is writable, the test also appends
// its workflow summary.
// Final evidence must retain the checkout SHA, session name, runtime account,
// selected scenarios, durations, and cleanup result from the log or summary.
func TestFunctional_PublicUIAPI(t *testing.T) {
	if !enabled(os.Getenv(envEnabled)) {
		t.Skipf("set %s=1 to run the public UI API scenarios", envEnabled)
	}
	resume := enabled(os.Getenv(envResume))
	adopt := enabled(os.Getenv(envAdoptSession))
	if adopt && !resume {
		t.Fatalf("%s requires %s=1", envAdoptSession, envResume)
	}

	repoRoot := repositoryRoot(t)
	summary := newWorkflowSummary(t)
	scenarios := selectedScenarios(t)
	dedicated := dedicatedTargetFromEnv(t, scenarios)
	if strings.TrimSpace(os.Getenv(envSessionName)) == "" {
		t.Setenv(envSessionName, "c9u"+time.Now().UTC().Format("01021504")+randomHex(t, 2))
	}
	sourceBuildServices := premiumCDCSourceBuildServices(t, repoRoot, scenarios, summary)
	tiltfile := copySessionTiltfile(t, repoRoot, os.Getenv(envSessionName), sourceBuildServices, summary)
	prepareDefaultUIAPIInputs(t, summary, repoRoot)
	cp9Binary := buildCP9(t, repoRoot, summary)
	session := sessionFromEnv(t, repoRoot, tiltfile, dedicated)
	if strings.TrimSpace(os.Getenv(envDatabasePassword)) == "" {
		t.Setenv(envDatabasePassword, "UiApi1!"+session.Name)
	}
	configPath := writeSessionConfig(t, session, summary)

	env := &testEnvironment{
		baseURL:     "https://" + session.Name + "." + session.Host.Exposure.ZoneName,
		configPath:  configPath,
		cp9Binary:   cp9Binary,
		dedicated:   dedicated,
		repoRoot:    repoRoot,
		sessionName: session.Name,
		summary:     summary,
	}

	t.Cleanup(func() {
		if !enabled(os.Getenv(envDestroySession)) {
			env.summary.record(
				"cleanup",
				"preserved",
				0,
				"set "+envDestroySession+"=1 only after explicit authority to destroy this exact session",
			)
			return
		}
		if _, err := os.Stat(filepath.Join(repoRoot, ".cp9", session.Name)); os.IsNotExist(err) {
			return
		}
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
		defer cancel()
		if err := runCP9Stage(ctx, t, env, "destroy", "destroy", "config="+configPath); err != nil {
			t.Errorf("destroy CP9 session: %v", err)
		}
	})

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Hour)
	defer cancel()
	if !resume {
		if err := runCP9Stage(ctx, t, env, "setup", "setup", "config="+configPath, "checkout=false"); err != nil {
			t.Fatalf("setup CP9 session: %v", err)
		}
		if err := runCP9Stage(ctx, t, env, "doctor", "doctor", "config="+configPath); err != nil {
			t.Fatalf("doctor CP9 session: %v", err)
		}
		if err := runCP9Stage(ctx, t, env, "up/dry-run", "up", "config="+configPath, "dry-run=true"); err != nil {
			t.Fatalf("dry-run CP9 session: %v", err)
		}
		if err := runCP9Stage(ctx, t, env, "up", "up", "config="+configPath); err != nil {
			t.Fatalf("up CP9 session: %v", err)
		}
	}
	env.runJournal = openPublicUIRunJournal(
		t,
		repoRoot,
		session.Name,
		scenarios,
		dedicated.Cloud,
		resume,
		adopt,
	)
	if strings.TrimSpace(os.Getenv(envAuthToken)) == "" {
		t.Setenv(envAuthToken, fetchUIBearerToken(t, env, summary))
	}
	env.uiClient = newUICookieClient(t, env, strings.TrimSpace(os.Getenv(envAuthToken)), summary)
	if len(scenarios) != 1 || scenarios[0] != "dedicated" || dedicated.Cloud == "aws" {
		env.fixture = createTestFixture(t, ctx, env)
	}

	if slices.Contains(scenarios, "dedicated") {
		runPublicUIScenario(t, env, "dedicated", testDedicated)
	}
	if slices.Contains(scenarios, "premium") {
		runPublicUIScenario(t, env, "premium", testPremium)
	}
	if slices.Contains(scenarios, "starter-e1") {
		runPublicUIScenario(t, env, "starter-e1", testStarterAndEssentialV1)
	}
	if slices.Contains(scenarios, "essential-v2") {
		runPublicUIScenario(t, env, "essential-v2", testEssentialV2)
	}
}

func runPublicUIScenario(
	t *testing.T,
	env *testEnvironment,
	name string,
	run func(*testing.T, *testEnvironment),
) {
	t.Helper()
	t.Run(name, func(t *testing.T) {
		t.Parallel()
		startedAt := time.Now()
		t.Cleanup(func() {
			status := "passed"
			if t.Skipped() {
				status = "skipped"
			} else if t.Failed() {
				status = "failed"
			}
			env.summary.record("scenario/"+name, status, time.Since(startedAt), "")
		})
		run(t, env)
	})
}

func fetchUIBearerToken(t *testing.T, env *testEnvironment, summary *workflowSummary) string {
	t.Helper()
	startedAt := time.Now()
	status := "failed"
	defer func() {
		summary.record("auth/login", status, time.Since(startedAt), "")
	}()

	state := randomAuthValue(t)
	nonce := randomAuthValue(t)
	callbackURL := env.baseURL + "/auth_redirect"
	authorizeURL := url.URL{
		Scheme: "https",
		Host:   auth0Domain,
		Path:   "/authorize",
	}
	query := authorizeURL.Query()
	query.Set("client_id", auth0ClientID)
	query.Set("response_type", "token id_token")
	query.Set("redirect_uri", callbackURL)
	query.Set("scope", "openid email")
	query.Set("nonce", nonce)
	query.Set("state", state)
	authorizeURL.RawQuery = query.Encode()

	jar, err := cookiejar.New(nil)
	if err != nil {
		t.Fatalf("create Auth0 UI login cookie jar: %v", err)
	}
	var callback *url.URL
	client := &http.Client{
		Jar: jar,
		CheckRedirect: func(request *http.Request, _ []*http.Request) error {
			if request.URL.Scheme+"://"+request.URL.Host+request.URL.Path == callbackURL {
				callback = request.URL
				return http.ErrUseLastResponse
			}
			return nil
		},
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()

	t.Logf("Auth0 UI login request: GET %s\nrequest body:\n<empty>", authorizeURL.RequestURI())
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, authorizeURL.String(), nil)
	if err != nil {
		t.Fatalf("build Auth0 UI login request GET %s: %v", authorizeURL.RequestURI(), err)
	}
	response, err := client.Do(request)
	if err != nil {
		t.Fatalf("Auth0 UI login request GET %s: %v", authorizeURL.RequestURI(), err)
	}
	responseBody, err := io.ReadAll(response.Body)
	_ = response.Body.Close()
	if err != nil {
		t.Fatalf("read Auth0 UI login response for GET %s: %v", authorizeURL.RequestURI(), err)
	}
	if response.StatusCode != http.StatusOK {
		t.Fatalf(
			"Auth0 UI login request GET %s returned %s with %d-byte body",
			authorizeURL.RequestURI(),
			response.Status,
			len(responseBody),
		)
	}

	const configPrefix = "window.atob('"
	configStart := bytes.Index(responseBody, []byte(configPrefix))
	if configStart < 0 {
		t.Fatalf("Auth0 UI login response for GET %s has no hosted-page config", authorizeURL.RequestURI())
	}
	configStart += len(configPrefix)
	configEnd := bytes.Index(responseBody[configStart:], []byte("')"))
	if configEnd < 0 {
		t.Fatalf("Auth0 UI login response for GET %s has malformed hosted-page config", authorizeURL.RequestURI())
	}
	configData, err := base64.StdEncoding.DecodeString(string(responseBody[configStart : configStart+configEnd]))
	if err != nil {
		t.Fatalf("decode Auth0 hosted-page config for GET %s: %v", authorizeURL.RequestURI(), err)
	}
	var config struct {
		Auth0Tenant string `json:"auth0Tenant"`
		CallbackURL string `json:"callbackURL"`
		ClientID    string `json:"clientID"`
		Options     struct {
			CSRF         string `json:"_csrf"`
			IntState     string `json:"_intstate"`
			Nonce        string `json:"nonce"`
			ResponseType string `json:"response_type"`
			Scope        string `json:"scope"`
			State        string `json:"state"`
		} `json:"internalOptions"`
	}
	if err := json.Unmarshal(configData, &config); err != nil {
		t.Fatalf("decode Auth0 hosted-page config for GET %s: %v", authorizeURL.RequestURI(), err)
	}
	if config.ClientID != auth0ClientID || config.CallbackURL != callbackURL || config.Options.Nonce != nonce {
		t.Fatalf(
			"Auth0 hosted-page config drift for GET %s: client_id=%q callback=%q nonce_matches=%t",
			authorizeURL.RequestURI(),
			config.ClientID,
			config.CallbackURL,
			config.Options.Nonce == nonce,
		)
	}

	loginBody, err := json.Marshal(map[string]string{
		"_csrf":         config.Options.CSRF,
		"_intstate":     config.Options.IntState,
		"client_id":     config.ClientID,
		"connection":    auth0Connection,
		"nonce":         config.Options.Nonce,
		"password":      defaultUISeedPassword,
		"redirect_uri":  config.CallbackURL,
		"response_type": config.Options.ResponseType,
		"scope":         config.Options.Scope,
		"state":         config.Options.State,
		"tenant":        config.Auth0Tenant,
		"username":      defaultUISeedEmail,
	})
	if err != nil {
		t.Fatalf("encode Auth0 UI login request POST /usernamepassword/login: %v", err)
	}
	t.Logf(
		"Auth0 UI login request: POST /usernamepassword/login\nrequest body:\n%s",
		loginBody,
	)
	request, err = http.NewRequestWithContext(
		ctx,
		http.MethodPost,
		"https://"+auth0Domain+"/usernamepassword/login",
		bytes.NewReader(loginBody),
	)
	if err != nil {
		t.Fatalf("build Auth0 UI login request POST /usernamepassword/login: %v", err)
	}
	request.Header.Set("Content-Type", "application/json")
	response, err = client.Do(request)
	if err != nil {
		t.Fatalf("Auth0 UI login request POST /usernamepassword/login: %v", err)
	}
	responseBody, err = io.ReadAll(response.Body)
	_ = response.Body.Close()
	if err != nil {
		t.Fatalf("read Auth0 UI login response for POST /usernamepassword/login: %v", err)
	}
	if response.StatusCode != http.StatusOK {
		t.Fatalf(
			"Auth0 UI login request POST /usernamepassword/login returned %s with %d-byte body",
			response.Status,
			len(responseBody),
		)
	}

	formAction, formBody := parseAuth0CallbackForm(t, responseBody)
	t.Logf(
		"Auth0 UI login request: POST %s\nrequest body:\n%s",
		formAction.RequestURI(),
		formBody.Encode(),
	)
	request, err = http.NewRequestWithContext(
		ctx,
		http.MethodPost,
		formAction.String(),
		strings.NewReader(formBody.Encode()),
	)
	if err != nil {
		t.Fatalf("build Auth0 UI login request POST %s: %v", formAction.RequestURI(), err)
	}
	request.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	response, err = client.Do(request)
	if err != nil {
		t.Fatalf("Auth0 UI login request POST %s: %v", formAction.RequestURI(), err)
	}
	callbackResponseBody, readErr := io.ReadAll(response.Body)
	_ = response.Body.Close()
	if readErr != nil {
		t.Fatalf("read Auth0 UI login response for POST %s: %v", formAction.RequestURI(), readErr)
	}
	if callback == nil || callback.Fragment == "" {
		t.Fatalf(
			"Auth0 UI login request POST %s returned %s without redirect to %s; response body was %d bytes",
			formAction.RequestURI(),
			response.Status,
			callbackURL,
			len(callbackResponseBody),
		)
	}
	tokens, err := url.ParseQuery(callback.Fragment)
	if err != nil {
		t.Fatalf("decode Auth0 UI login callback fragment: %v", err)
	}
	if tokens.Get("error") != "" {
		t.Fatalf(
			"Auth0 UI login callback returned error=%q error_description=%q",
			tokens.Get("error"),
			tokens.Get("error_description"),
		)
	}
	if tokens.Get("state") != state {
		t.Fatalf("Auth0 UI login callback state mismatch")
	}
	idToken := tokens.Get("id_token")
	validateUIIDToken(t, idToken, nonce)
	t.Logf("Auth0 UI login response: redirect=/auth_redirect id_token=%s", idToken)
	status = "passed"
	return idToken
}

func newUICookieClient(
	t *testing.T,
	env *testEnvironment,
	idToken string,
	summary *workflowSummary,
) *http.Client {
	t.Helper()
	startedAt := time.Now()
	status := "failed"
	defer func() {
		summary.record("auth/ui-cookie", status, time.Since(startedAt), "")
	}()
	jar, err := cookiejar.New(nil)
	if err != nil {
		t.Fatalf("create UI cookie jar: %v", err)
	}
	client := &http.Client{Jar: jar, Timeout: 2 * time.Minute}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	body, err := json.Marshal(map[string]string{"id_token": idToken})
	if err != nil {
		t.Fatalf("encode UI login confirmation: %v", err)
	}
	request, err := http.NewRequestWithContext(
		ctx,
		http.MethodPost,
		env.baseURL+"/login_confirm",
		bytes.NewReader(body),
	)
	if err != nil {
		t.Fatalf("build UI login confirmation: %v", err)
	}
	request.Header.Set("Accept", "application/json")
	request.Header.Set("Authorization", "Bearer "+idToken)
	request.Header.Set("Content-Type", "application/json")
	response, err := client.Do(request)
	if err != nil {
		t.Fatalf("UI login confirmation: %v", err)
	}
	responseBody, readErr := io.ReadAll(response.Body)
	_ = response.Body.Close()
	if readErr != nil {
		t.Fatalf("read UI login confirmation response: %v", readErr)
	}
	if response.StatusCode != http.StatusOK {
		t.Fatalf("UI login confirmation returned %s with %d-byte body", response.Status, len(responseBody))
	}
	base, err := url.Parse(env.baseURL)
	if err != nil {
		t.Fatalf("parse UI origin %q: %v", env.baseURL, err)
	}
	if len(jar.Cookies(base)) == 0 {
		t.Fatal("UI login confirmation returned no session cookie")
	}
	t.Logf("UI login confirmation: HTTP 200, session cookie established")
	status = "passed"
	return client
}

func randomAuthValue(t *testing.T) string {
	t.Helper()
	value := make([]byte, 24)
	if _, err := rand.Read(value); err != nil {
		t.Fatalf("generate Auth0 UI login value: %v", err)
	}
	return base64.RawURLEncoding.EncodeToString(value)
}

func randomHex(t *testing.T, size int) string {
	t.Helper()
	value := make([]byte, size)
	if _, err := rand.Read(value); err != nil {
		t.Fatalf("generate random value: %v", err)
	}
	return fmt.Sprintf("%x", value)
}

func parseAuth0CallbackForm(t *testing.T, body []byte) (*url.URL, url.Values) {
	t.Helper()
	document, err := html.Parse(bytes.NewReader(body))
	if err != nil {
		t.Fatalf("parse Auth0 UI login callback form: %v", err)
	}
	var form *html.Node
	var findForm func(*html.Node)
	findForm = func(node *html.Node) {
		if form != nil {
			return
		}
		if node.Type == html.ElementNode && node.Data == "form" {
			form = node
			return
		}
		for child := node.FirstChild; child != nil; child = child.NextSibling {
			findForm(child)
		}
	}
	findForm(document)
	if form == nil {
		t.Fatalf("Auth0 UI login callback response has no submit form")
	}

	var action string
	for _, attribute := range form.Attr {
		if attribute.Key == "action" {
			action = attribute.Val
		}
	}
	values := url.Values{}
	var collectInputs func(*html.Node)
	collectInputs = func(node *html.Node) {
		if node.Type == html.ElementNode && node.Data == "input" {
			var name, value string
			for _, attribute := range node.Attr {
				switch attribute.Key {
				case "name":
					name = attribute.Val
				case "value":
					value = attribute.Val
				}
			}
			if name != "" {
				values.Add(name, value)
			}
		}
		for child := node.FirstChild; child != nil; child = child.NextSibling {
			collectInputs(child)
		}
	}
	collectInputs(form)
	if action == "" || len(values) == 0 {
		t.Fatalf("Auth0 UI login callback response has no submit form")
	}
	formAction, err := url.Parse(action)
	if err != nil {
		t.Fatalf("parse Auth0 UI login callback action %q: %v", action, err)
	}
	auth0Origin, _ := url.Parse("https://" + auth0Domain)
	return auth0Origin.ResolveReference(formAction), values
}

func validateUIIDToken(t *testing.T, token, nonce string) {
	t.Helper()
	parts := strings.Split(token, ".")
	if len(parts) != 3 {
		t.Fatalf("Auth0 UI login callback has no valid id_token")
	}
	payloadData, err := base64.RawURLEncoding.DecodeString(parts[1])
	if err != nil {
		t.Fatalf("decode Auth0 UI id_token payload: %v", err)
	}
	var payload struct {
		Audience   string `json:"aud"`
		Email      string `json:"email"`
		ExpiresAt  int64  `json:"exp"`
		Issuer     string `json:"iss"`
		LoginOrgID any    `json:"login_org_id"`
		Nonce      string `json:"nonce"`
	}
	if err := json.Unmarshal(payloadData, &payload); err != nil {
		t.Fatalf("decode Auth0 UI id_token claims: %v", err)
	}
	if payload.Audience != auth0ClientID ||
		payload.Email != defaultUISeedEmail ||
		payload.Issuer != "https://"+auth0Domain+"/" ||
		fmt.Sprint(payload.LoginOrgID) != config.DefaultUIOrganizationID ||
		payload.Nonce != nonce ||
		payload.ExpiresAt <= time.Now().Unix() {
		t.Fatalf(
			"Auth0 UI id_token claim mismatch: audience=%q email=%q issuer=%q login_org_id=%v nonce_matches=%t expired=%t",
			payload.Audience,
			payload.Email,
			payload.Issuer,
			payload.LoginOrgID,
			payload.Nonce == nonce,
			payload.ExpiresAt <= time.Now().Unix(),
		)
	}
}

func createTestFixture(t *testing.T, ctx context.Context, env *testEnvironment) *testFixture {
	t.Helper()
	startedAt := time.Now()
	status := "failed"
	defer func() {
		env.summary.record("fixture/provision", status, time.Since(startedAt), "")
	}()
	contextData, err := os.ReadFile(
		filepath.Join(env.repoRoot, ".cp9", env.sessionName, "context", "context.json"),
	)
	if err != nil {
		t.Fatalf("read Session context for test fixture: %v", err)
	}
	var sessionContext struct {
		SessionTags map[string]string `json:"session_tags"`
		Target      struct {
			Account string `json:"account"`
		} `json:"target"`
	}
	if err := json.Unmarshal(contextData, &sessionContext); err != nil {
		t.Fatalf("decode Session context for test fixture: %v", err)
	}
	owner := sessionContext.SessionTags["cp9-owner"]
	if owner == "" || sessionContext.Target.Account == "" {
		t.Fatalf("Session context is missing cp9-owner or target account: %s", contextData)
	}

	template := `
AWSTemplateFormatVersion: "2010-09-09"
Parameters:
  AccountID:
    Type: String
  DBPassword:
    Type: String
    NoEcho: true
  KeyPair:
    Type: AWS::EC2::KeyPair::KeyName
  Owner:
    Type: String
  Session:
    Type: String
  LatestAmiId:
    Type: AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
    Default: /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64
Resources:
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.89.0.0/16
      EnableDnsHostnames: true
      EnableDnsSupport: true
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
        - {Key: Name, Value: !Sub "${Session}-fixture"}
  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
        - {Key: Name, Value: !Sub "${Session}-fixture"}
  GatewayAttachment:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      InternetGatewayId: !Ref InternetGateway
      VpcId: !Ref VPC
  RouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
        - {Key: Name, Value: !Sub "${Session}-fixture"}
  InternetRoute:
    Type: AWS::EC2::Route
    DependsOn: GatewayAttachment
    Properties:
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway
      RouteTableId: !Ref RouteTable
  Subnet:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone: us-west-2a
      CidrBlock: 10.89.1.0/24
      MapPublicIpOnLaunch: true
      VpcId: !Ref VPC
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
        - {Key: Name, Value: !Sub "${Session}-fixture"}
  SubnetRoute:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref RouteTable
      SubnetId: !Ref Subnet
  SecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: CP9 public UI API fixture
      SecurityGroupIngress:
        - {IpProtocol: tcp, FromPort: 22, ToPort: 22, CidrIp: 0.0.0.0/0}
        - {IpProtocol: tcp, FromPort: 3306, ToPort: 3306, CidrIp: 0.0.0.0/0}
        - {IpProtocol: tcp, FromPort: 4000, ToPort: 4000, CidrIp: 10.89.0.0/16}
      VpcId: !Ref VPC
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
        - {Key: Name, Value: !Sub "${Session}-fixture"}
  Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub "${Session}-data"
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
        - {Key: Name, Value: !Sub "${Session}-fixture"}
  ImportRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "${Session}-import"
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal: {AWS: !Sub "arn:aws:iam::${AccountID}:root"}
            Action: sts:AssumeRole
      Policies:
        - PolicyName: import-s3
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action: [s3:GetBucketLocation, s3:ListBucket]
                Resource: !Sub "arn:aws:s3:::${Bucket}"
              - Effect: Allow
                Action: [s3:GetObject, s3:GetObjectVersion]
                Resource: !Sub "arn:aws:s3:::${Bucket}/inputs/*"
              - Effect: Allow
                Action: [s3:PutObject]
                Resource: !Sub "arn:aws:s3:::${Bucket}/outputs/*"
      Tags:
        - {Key: cp9-owner, Value: !Ref Owner}
        - {Key: cp9-session, Value: !Ref Session}
  DatabaseLaunchTemplate:
    Type: AWS::EC2::LaunchTemplate
    Properties:
      LaunchTemplateName: !Sub "${Session}-fixture"
      TagSpecifications:
        - ResourceType: launch-template
          Tags:
            - {Key: cp9-owner, Value: !Ref Owner}
            - {Key: cp9-session, Value: !Ref Session}
            - {Key: Name, Value: !Sub "${Session}-fixture"}
      LaunchTemplateData:
        ImageId: !Ref LatestAmiId
        InstanceType: t3.micro
        KeyName: !Ref KeyPair
        NetworkInterfaces:
          - AssociatePublicIpAddress: true
            DeviceIndex: 0
            Groups: [!Ref SecurityGroup]
            SubnetId: !Ref Subnet
        TagSpecifications:
          - ResourceType: instance
            Tags:
              - {Key: cp9-owner, Value: !Ref Owner}
              - {Key: cp9-session, Value: !Ref Session}
              - {Key: Name, Value: !Sub "${Session}-fixture"}
          - ResourceType: volume
            Tags:
              - {Key: cp9-owner, Value: !Ref Owner}
              - {Key: cp9-session, Value: !Ref Session}
              - {Key: Name, Value: !Sub "${Session}-fixture"}
          - ResourceType: network-interface
            Tags:
              - {Key: cp9-owner, Value: !Ref Owner}
              - {Key: cp9-session, Value: !Ref Session}
              - {Key: Name, Value: !Sub "${Session}-fixture"}
        UserData:
          Fn::Base64: !Sub |
            #!/bin/bash
            set -euxo pipefail
            dnf install -y mariadb105-server
            cat >/etc/my.cnf.d/cp9-ui-api.cnf <<'EOF'
            [mariadb]
            bind-address=0.0.0.0
            binlog_format=ROW
            log-bin=/var/lib/mysql/cp9-ui-api-bin
            server_id=8041
            EOF
            systemctl enable --now mariadb
            mysql <<'SQL'
            CREATE USER IF NOT EXISTS 'cp9'@'%' IDENTIFIED BY '${DBPassword}';
            GRANT ALL PRIVILEGES ON *.* TO 'cp9'@'%' WITH GRANT OPTION;
            CREATE DATABASE migration_source_ded CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
            CREATE DATABASE migration_source_prem CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
            CREATE DATABASE changefeed_sink_ded CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
            CREATE DATABASE changefeed_sink_prem CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
            CREATE TABLE migration_source_ded.proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);
            CREATE TABLE migration_source_prem.proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);
            CREATE TABLE changefeed_sink_ded.proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);
            CREATE TABLE changefeed_sink_prem.proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);
            INSERT INTO migration_source_ded.proof VALUES (1, 'migration-dedicated');
            INSERT INTO migration_source_prem.proof VALUES (1, 'migration-premium');
            FLUSH PRIVILEGES;
            SQL
  Database:
    Type: AWS::EC2::Instance
    Properties:
      LaunchTemplate:
        LaunchTemplateId: !Ref DatabaseLaunchTemplate
        Version: !GetAtt DatabaseLaunchTemplate.LatestVersionNumber
Outputs:
  Bucket: {Value: !Ref Bucket}
  ImportRoleARN: {Value: !GetAtt ImportRole.Arn}
  PrivateIP: {Value: !GetAtt Database.PrivateIp}
  PublicIP: {Value: !GetAtt Database.PublicIp}
  SecurityGroup: {Value: !Ref SecurityGroup}
  SubnetID: {Value: !Ref Subnet}
  VpcID: {Value: !Ref VPC}
`
	templatePath := filepath.Join(t.TempDir(), "fixture.yaml")
	if err := os.WriteFile(templatePath, []byte(template), 0o600); err != nil {
		t.Fatalf("write test fixture CloudFormation template: %v", err)
	}
	stackName := env.sessionName + "-ui-api-fixture"
	fixture := &testFixture{owner: owner}
	runAWS(
		t,
		ctx,
		env,
		"cloudformation",
		"deploy",
		"--stack-name",
		stackName,
		"--template-file",
		templatePath,
		"--capabilities",
		"CAPABILITY_NAMED_IAM",
		"--parameter-overrides",
		"AccountID="+sessionContext.Target.Account,
		"DBPassword="+os.Getenv(envDatabasePassword),
		"KeyPair="+os.Getenv(envKeyPair),
		"Owner="+owner,
		"Session="+env.sessionName,
		"--tags",
		"cp9-owner="+owner,
		"cp9-session="+env.sessionName,
	)
	output := runAWS(
		t,
		ctx,
		env,
		"cloudformation",
		"describe-stacks",
		"--stack-name",
		stackName,
		"--query",
		"Stacks[0].Outputs",
		"--output",
		"json",
	)
	var stackOutputs []struct {
		Key   string `json:"OutputKey"`
		Value string `json:"OutputValue"`
	}
	if err := json.Unmarshal(output, &stackOutputs); err != nil {
		t.Fatalf("decode test fixture CloudFormation outputs: %v\n%s", err, output)
	}
	values := map[string]string{}
	for _, item := range stackOutputs {
		values[item.Key] = item.Value
	}
	fixture.bucket = values["Bucket"]
	fixture.importRoleARN = values["ImportRoleARN"]
	fixture.privateIP = values["PrivateIP"]
	fixture.publicIP = values["PublicIP"]
	fixture.securityGroup = values["SecurityGroup"]
	fixture.subnetID = values["SubnetID"]
	fixture.vpcID = values["VpcID"]
	if fixture.bucket == "" || fixture.importRoleARN == "" || fixture.publicIP == "" ||
		fixture.securityGroup == "" || fixture.subnetID == "" || fixture.vpcID == "" {
		t.Fatalf("test fixture CloudFormation outputs are incomplete: %s", output)
	}
	fixture.waitForDatabase(t, ctx, env)
	sqlPath := filepath.Join(t.TempDir(), "proof.sql")
	if err := os.WriteFile(
		sqlPath,
		[]byte("INSERT INTO cp9_uiapi.proof VALUES (1, 'public-ui-api-proof');\n"),
		0o600,
	); err != nil {
		t.Fatalf("write import SQL fixture: %v", err)
	}
	runAWS(
		t,
		ctx,
		env,
		"s3api",
		"put-object",
		"--bucket",
		fixture.bucket,
		"--key",
		"inputs/proof.sql",
		"--body",
		sqlPath,
		"--tagging",
		"cp9-owner="+owner+"&cp9-session="+env.sessionName,
	)
	status = "passed"
	return fixture
}

func (f *testFixture) waitForDatabase(t *testing.T, ctx context.Context, env *testEnvironment) {
	t.Helper()
	for {
		command := exec.CommandContext(
			ctx,
			"ssh",
			"-o",
			"BatchMode=yes",
			"-o",
			"StrictHostKeyChecking=accept-new",
			"-o",
			"ConnectTimeout=10",
			"-i",
			os.Getenv(envSSHPrivateKey),
			"ec2-user@"+f.publicIP,
			"MYSQL_PWD='"+os.Getenv(envDatabasePassword)+
				"' mysql --batch --skip-column-names --host 127.0.0.1 --user cp9 --execute='SELECT @@log_bin, @@binlog_format'",
		)
		result, err := command.CombinedOutput()
		if err == nil && bytes.Contains(result, []byte("1\tROW")) {
			return
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for fixture MariaDB: %v\n%s", ctx.Err(), result)
		case <-time.After(20 * time.Second):
		}
	}
}

func runAWS(t *testing.T, ctx context.Context, env *testEnvironment, args ...string) []byte {
	t.Helper()
	result, err := runAWSCommand(ctx, env, args...)
	if err != nil {
		t.Fatalf("aws %s: %v (%d-byte output)", awsOperation(args), err, len(result))
	}
	return result
}

func runAWSCommand(ctx context.Context, env *testEnvironment, args ...string) ([]byte, error) {
	args = append(
		args,
		"--profile",
		env.sessionName,
		"--region",
		targetRegion,
	)
	command := exec.CommandContext(ctx, "aws", args...)
	result, err := command.CombinedOutput()
	return result, err
}

func awsOperation(args []string) string {
	return strings.Join(args[:min(len(args), 2)], " ")
}

func testDedicated(t *testing.T, env *testEnvironment) {
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	client := &http.Client{Timeout: 2 * time.Minute}
	scenarioStage(t, env, "catalog", func(t *testing.T) {
		verifyDedicatedCatalog(t, ctx, client, env)
	})
	clusterID, owned := loadPublicUIRunProduct(t, env, "dedicated", os.Getenv(envDedicatedID))
	deletion := productDeletion{request: func(t *testing.T, deleteCtx context.Context) {
		doAPIRequest(
			t,
			deleteCtx,
			client,
			env.baseURL,
			http.MethodDelete,
			fmt.Sprintf(
				"/api/v2023/dedicated/v1beta1/clusters/%s?orgId=%s&projectId=%s",
				clusterID,
				dedicatedOrgID,
				dedicatedProjectID,
			),
			nil,
			"",
			http.StatusOK,
		)
	}, created: owned}
	if clusterID == "" || owned || env.runJournal.stageStatus(t.Name()+"/create-cluster") != "" {
		var cidrReceipt dbaasui.DedicatedCIDRReceipt
		scenarioStage(t, env, "prepare-regional-cidr", func(t *testing.T) {
			var err error
			cidrReceipt, err = dbaasui.PrepareDedicatedCIDR(
				ctx,
				client,
				env.baseURL,
				dedicatedOrgID,
				dedicatedProjectID,
				env.dedicated.CIDRProvider,
				env.dedicated.Region,
				dedicatedCIDR,
				strings.TrimSpace(os.Getenv(envAuthToken)),
			)
			if err != nil {
				t.Fatalf("prepare Dedicated CIDR: %v", err)
			}
			t.Logf("Dedicated CIDR receipt: %s", cidrReceipt.Summary())
			env.summary.record("dedicated/cidr-scope", "passed", 0, cidrReceipt.Summary())
		})
		scenarioStage(t, env, "create-cluster", func(t *testing.T) {
			if clusterID != "" {
				t.Fatal("journaled Dedicated Product ID has no passed create receipt")
			}
			body := []byte(fmt.Sprintf(
				`{"cloudProvider":%q,"displayName":%q,"port":4000,"regionId":%q,"tidbNodeSetting":{"nodeSpecKey":%q,"tidbNodeGroups":[{"nodeCount":2}]},"tikvNodeSetting":{"nodeCount":3,"nodeSpecKey":%q,"storageSizeGi":500,"storageType":%q}}`,
				env.dedicated.Cloud,
				env.sessionName+"-ded",
				env.dedicated.RegionID,
				env.dedicated.TiDBNodeSpec,
				env.dedicated.TiKVNodeSpec,
				env.dedicated.StorageType,
			))
			response := doAPIRequest(
				t,
				ctx,
				client,
				env.baseURL,
				http.MethodPost,
				fmt.Sprintf("/api/v2023/dedicated/v1beta1/clusters?orgId=%s&projectId=%s", dedicatedOrgID, dedicatedProjectID),
				body,
				"application/json",
				http.StatusOK,
			)
			var created struct {
				ClusterID string `json:"clusterId"`
			}
			if err := json.Unmarshal(response, &created); err != nil || created.ClusterID == "" {
				t.Fatalf("decode %d-byte Dedicated create response: %v", len(response), err)
			}
			clusterID = created.ClusterID
			recordPublicUIRunProduct(t, env, "dedicated", clusterID)
			deletion.created = true
		})
	}
	if env.runJournal.stageStatus(t.Name()+"/create-cluster") == "passed" &&
		(clusterID == "" || !deletion.created) {
		t.Fatal("passed Dedicated create receipt lacks a run-owned Product ID")
	}
	scenarioStage(t, env, "wait-active", func(t *testing.T) {
		waitForClusterActive(t, ctx, env, "1440002", clusterID)
	})
	publicHost := loadPublicUIRunHost(env, "dedicated")
	scenarioStage(t, env, "configure-public-endpoint", func(t *testing.T) {
		publicHost = ensureDedicatedPublicEndpoint(t, ctx, client, env, clusterID)
		recordPublicUIRunHost(t, env, "dedicated", publicHost)
	})
	if env.runJournal.stageStatus(t.Name()+"/configure-public-endpoint") == "passed" && publicHost == "" {
		t.Fatal("passed Dedicated public-endpoint receipt lacks its host")
	}
	scenarioStage(t, env, "configure-public-allowlist", func(t *testing.T) {
		ensureDedicatedPublicAllowlist(t, ctx, client, env, clusterID)
	})
	scenarioStage(t, env, "configure-password", func(t *testing.T) {
		setDedicatedPassword(t, ctx, client, env, clusterID)
	})
	scenarioStage(t, env, "public-mysql-tls", func(t *testing.T) {
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			false,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.public_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				"INSERT INTO cp9_uiapi.public_sql VALUES (1, 'dedicated-public-tls') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value); "+
				"SELECT id,value FROM cp9_uiapi.public_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"dedicated-public-tls",
		)
	})
	if env.dedicated.Cloud == "aws" {
		testDedicatedPrivateLink(t, ctx, client, env, clusterID)
		testDedicatedImport(t, ctx, client, env, clusterID, publicHost)
		testDedicatedChangefeed(t, ctx, client, env, clusterID, publicHost)
		testDedicatedMigration(t, ctx, client, env, clusterID, publicHost)
		testDedicatedBackupRestore(t, ctx, client, env, clusterID)
	}
	runDedicatedProductLifecycle(t, ctx, env, clusterID, &deletion, "")
}

func verifyDedicatedCatalog(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
) {
	t.Helper()
	path := fmt.Sprintf(
		"/api/v2023/dedicated/v1beta1/regions?orgId=%s&projectId=%s&pageSize=100",
		dedicatedOrgID,
		dedicatedProjectID,
	)
	response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
	var regions struct {
		Regions []struct {
			RegionID string `json:"regionId"`
		} `json:"regions"`
	}
	if err := json.Unmarshal(response, &regions); err != nil {
		t.Fatalf("decode %d-byte Dedicated regions response: %v", len(response), err)
	}
	counts := map[string]int{}
	for _, region := range regions.Regions {
		counts[region.RegionID]++
	}
	requiredRegions := []string{"aws-us-west-2"}
	if env.dedicated.Cloud == "gcp" {
		requiredRegions = append(requiredRegions, env.dedicated.RegionID)
	}
	for _, regionID := range requiredRegions {
		if counts[regionID] != 1 {
			t.Fatalf("Dedicated region %s count=%d, want 1", regionID, counts[regionID])
		}
	}

	path = fmt.Sprintf(
		"/api/v2023/dedicated/v1beta1/regions/%s/nodeSpecs?orgId=%s&projectId=%s&pageSize=100",
		env.dedicated.RegionID,
		dedicatedOrgID,
		dedicatedProjectID,
	)
	response = doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
	var specs struct {
		NodeSpecs []struct {
			ComponentType string   `json:"componentType"`
			NodeSpecKey   string   `json:"nodeSpecKey"`
			StorageTypes  []string `json:"storageTypes"`
		} `json:"nodeSpecs"`
	}
	if err := json.Unmarshal(response, &specs); err != nil {
		t.Fatalf("decode %d-byte Dedicated node specs response: %v", len(response), err)
	}
	want := map[string]int{
		"TIDB/" + env.dedicated.TiDBNodeSpec: 0,
		"TIKV/" + env.dedicated.TiKVNodeSpec: 0,
	}
	for _, spec := range specs.NodeSpecs {
		key := spec.ComponentType + "/" + spec.NodeSpecKey
		if _, ok := want[key]; ok {
			want[key]++
		}
		if key == "TIKV/"+env.dedicated.TiKVNodeSpec &&
			!slices.Contains(spec.StorageTypes, env.dedicated.StorageType) {
			t.Fatalf(
				"Dedicated node spec %s does not support storage type %s",
				key,
				env.dedicated.StorageType,
			)
		}
	}
	for key, count := range want {
		if count != 1 {
			t.Fatalf("Dedicated node spec %s count=%d, want 1", key, count)
		}
	}
	message := fmt.Sprintf(
		"cloud=%s region=%s tidb=%s tikv=%s storage=%s",
		env.dedicated.Cloud,
		env.dedicated.RegionID,
		env.dedicated.TiDBNodeSpec,
		env.dedicated.TiKVNodeSpec,
		env.dedicated.StorageType,
	)
	env.summary.record("dedicated/catalog-scope", "passed", 0, message)
}

func runDedicatedProductLifecycle(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	clusterID string,
	deletion *productDeletion,
	stagePrefix string,
) {
	if stagePrefix == "" {
		if !deletion.created {
			skippedScenarioStage(t, env, "ui-metrics", "borrowed Dedicated cluster; UI Metrics is not claimed")
		} else {
			bizType := env.sessionName
			if env.dedicated.Cloud == "gcp" {
				bizType = "tidbcloud"
			}
			scenarioStage(t, env, "ui-metrics", func(t *testing.T) {
				runUIMetricsAcceptance(
					t,
					ctx,
					env,
					"dedicated",
					bizType,
					dedicatedOrgID,
					dedicatedProjectID,
					clusterID,
					false,
				)
			})
		}
	}
	if !deletion.created {
		if stagePrefix != "" {
			scenarioStage(t, env, stagePrefix+"delete-cluster", func(t *testing.T) {
				t.Fatal("restore delete requires a journaled run-owned Product ID")
			})
			return
		}
		skippedScenarioStage(t, env, stagePrefix+"delete-cluster", "borrowed Product; delete is not claimed")
		return
	}
	scenarioStage(t, env, stagePrefix+"delete-cluster", func(t *testing.T) {
		deletion.run(t, ctx)
	})
}

func testPremium(t *testing.T, env *testEnvironment) {
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	client := &http.Client{Timeout: 2 * time.Minute}
	orgID := config.DefaultUIOrganizationID
	projectID := orgID + "0002"
	clusterID, owned := loadPublicUIRunProduct(t, env, "premium", os.Getenv(envPremiumID))
	deletion := productDeletion{request: func(t *testing.T, deleteCtx context.Context) {
		doAPIRequest(
			t,
			deleteCtx,
			client,
			env.baseURL,
			http.MethodDelete,
			fmt.Sprintf(
				"/api/v2025/cloud/v1beta2/tidbs/%s?orgId=%s&projectId=%s",
				clusterID,
				orgID,
				projectID,
			),
			nil,
			"",
			http.StatusOK,
		)
	}, created: owned}
	if clusterID == "" || owned || env.runJournal.stageStatus(t.Name()+"/create-cluster") != "" {
		scenarioStage(t, env, "create-cluster", func(t *testing.T) {
			if clusterID != "" {
				t.Fatal("journaled Premium Product ID has no passed create receipt")
			}
			body := []byte(fmt.Sprintf(
				`{"servicePlan":"Premium","displayName":%q,"cloudProvider":"aws","regionId":"aws-us-west-2","maxRcu":"20000","highAvailabilityType":"REGIONAL","labels":{"tidb.cloud/project":"14400020002"}}`,
				env.sessionName+"-prem",
			))
			response := doAPIRequest(
				t,
				ctx,
				client,
				env.baseURL,
				http.MethodPost,
				"/api/v2025/cloud/v1beta2/tidbs?orgId=1440002&projectId=14400020002",
				body,
				"application/json",
				http.StatusOK,
			)
			var created struct {
				TiDBID string `json:"tidbId"`
			}
			if err := json.Unmarshal(response, &created); err != nil || created.TiDBID == "" {
				t.Fatalf("decode %d-byte Premium create response: %v", len(response), err)
			}
			clusterID = created.TiDBID
			recordPublicUIRunProduct(t, env, "premium", clusterID)
			deletion.created = true
		})
	}
	if env.runJournal.stageStatus(t.Name()+"/create-cluster") == "passed" &&
		(clusterID == "" || !deletion.created) {
		t.Fatal("passed Premium create receipt lacks a run-owned Product ID")
	}
	env.premiumID = clusterID
	scenarioStage(t, env, "wait-active", func(t *testing.T) {
		waitForClusterActive(t, ctx, env, "1440002", clusterID)
	})
	publicHost := loadPublicUIRunHost(env, "premium")
	scenarioStage(t, env, "configure-public-endpoint", func(t *testing.T) {
		publicHost = ensureNextGenPublicEndpoint(t, ctx, client, env, orgID, projectID, clusterID)
		recordPublicUIRunHost(t, env, "premium", publicHost)
	})
	if env.runJournal.stageStatus(t.Name()+"/configure-public-endpoint") == "passed" && publicHost == "" {
		t.Fatal("passed Premium public-endpoint receipt lacks its host")
	}
	scenarioStage(t, env, "configure-public-allowlist", func(t *testing.T) {
		ensureNextGenPublicAllowlist(t, ctx, client, env, orgID, projectID, clusterID)
	})
	scenarioStage(t, env, "configure-password", func(t *testing.T) {
		setNextGenPassword(t, ctx, client, env, orgID, projectID, clusterID)
	})
	scenarioStage(t, env, "public-mysql-tls", func(t *testing.T) {
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			true,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.public_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				"INSERT INTO cp9_uiapi.public_sql VALUES (1, 'premium-public-tls') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value); "+
				"SELECT id,value FROM cp9_uiapi.public_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"premium-public-tls",
		)
	})
	testNextGenPrivateLink(t, ctx, client, env, orgID, projectID, clusterID)
	testNextGenImport(t, ctx, client, env, orgID, projectID, clusterID, publicHost)
	testPremiumChangefeed(t, ctx, client, env, clusterID, publicHost)
	testPremiumMigration(t, ctx, client, env, clusterID, publicHost)
	if deletion.created {
		backupRestoreStatus := env.runJournal.stageStatus(t.Name() + "/backup-restore")
		startedAt := time.Now()
		passed := t.Run("harness-premium-metering-output", func(t *testing.T) {
			if backupRestoreStatus != "" {
				t.Fatalf("Premium metering requires backup-restore to be unstarted, got %s", backupRestoreStatus)
			}
			verifyPremiumMeteringOutput(t, ctx, env)
		})
		status := "passed"
		if !passed {
			status = "failed"
		}
		env.summary.record("harness/premium-metering-output", status, time.Since(startedAt), "")
		if !passed {
			return
		}
	} else {
		env.summary.record("harness/premium-metering-output", "skipped", 0, "borrowed Premium instance")
	}
	testPremiumBackupRestore(t, ctx, client, env, clusterID)
	runPremiumProductLifecycle(t, ctx, env, orgID, projectID, clusterID, &deletion, "")
}

type premiumMeteringContract struct {
	Bucket           string `json:"bucket"`
	ReadWriteRoleARN string `json:"read_write_role_arn"`
}

type awsRoleCredentials struct {
	AccessKey    string `json:"AccessKeyId"`
	SecretKey    string `json:"SecretAccessKey"`
	SessionToken string `json:"SessionToken"`
}

func loadPremiumMeteringContract(t *testing.T, env *testEnvironment) premiumMeteringContract {
	t.Helper()
	contractPath := filepath.Join(env.repoRoot, ".cp9", env.sessionName, "terraform", "contract.json")
	data, err := os.ReadFile(contractPath)
	if err != nil {
		t.Fatalf("read Premium metering contract: %v", err)
	}
	var output struct {
		Metering *premiumMeteringContract `json:"metering"`
	}
	if err := json.Unmarshal(data, &output); err != nil || output.Metering == nil {
		t.Fatalf("decode Premium metering contract: %v", err)
	}
	if output.Metering.Bucket == "" || output.Metering.ReadWriteRoleARN == "" {
		t.Fatal("Premium metering contract is incomplete")
	}
	return *output.Metering
}

func verifyPremiumMeteringOutput(t *testing.T, ctx context.Context, env *testEnvironment) {
	t.Helper()
	metering := loadPremiumMeteringContract(t, env)
	credentials := assumeAWSRole(t, ctx, env, metering.ReadWriteRoleARN)
	tempDir := t.TempDir()
	var data []byte
	deadline := time.Now().Add(premiumMeteringWaitTimeout)
	for {
		data = runAWSWithCredentials(t, ctx, credentials, "s3api", "list-objects-v2", "--bucket", metering.Bucket)
		var listing struct {
			Contents []struct {
				Key string `json:"Key"`
			} `json:"Contents"`
		}
		if err := json.Unmarshal(data, &listing); err != nil {
			t.Fatalf("decode Premium metering object list: %v", err)
		}
		keys := make([]string, 0, len(listing.Contents))
		for _, object := range listing.Contents {
			keys = append(keys, object.Key)
		}
		if hasObjectPrefix(keys, "metering/ru/") &&
			hasObjectPrefix(keys, "metering/meta/") &&
			hasObjectPrefix(keys, "metering/checkpoints/") &&
			premiumMeteringRawRUMatches(t, ctx, credentials, metering.Bucket, keys, tempDir) &&
			premiumMeteringMetadataMatches(t, ctx, credentials, metering.Bucket, keys, tempDir) &&
			premiumMeteringDeltaHasExpectedRows(t, ctx, credentials, metering.Bucket, keys, tempDir) {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("Premium metering did not publish raw metadata/RU, four-row Delta, and checkpoint within %s", premiumMeteringWaitTimeout)
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for Premium metering output: %v", ctx.Err())
		case <-time.After(20 * time.Second):
		}
	}
}

func hasObjectPrefix(keys []string, prefix string) bool {
	return slices.ContainsFunc(keys, func(key string) bool { return strings.HasPrefix(key, prefix) })
}

func premiumMeteringRawRUMatches(
	t *testing.T,
	ctx context.Context,
	credentials awsRoleCredentials,
	bucket string,
	keys []string,
	tempDir string,
) bool {
	t.Helper()
	for index, key := range keys {
		if !strings.HasPrefix(key, "metering/ru/") {
			continue
		}
		data := downloadGzipObject(t, ctx, credentials, bucket, key, filepath.Join(tempDir, fmt.Sprintf("ru-%d.json.gz", index)))
		var record struct {
			Timestamp int64                        `json:"timestamp"`
			Category  string                       `json:"category"`
			SelfID    string                       `json:"self_id"`
			Data      []map[string]json.RawMessage `json:"data"`
		}
		if json.Unmarshal(data, &record) != nil || record.Timestamp <= 0 ||
			record.Category != "control-plane" || record.SelfID == "" {
			continue
		}
		for _, point := range record.Data {
			var dataDiskType, raftDiskType string
			_ = json.Unmarshal(point["tikv_data_disk_type"], &dataDiskType)
			_ = json.Unmarshal(point["tikv_log_disk_type"], &raftDiskType)
			if dataDiskType == "gp3" && raftDiskType == "gp3" &&
				meteringValueMatches(point["tikv_data_disk_count"], 3, "count") &&
				meteringValueMatches(point["tikv_data_disk_iops"], 8000, "iops") &&
				meteringValueMatches(point["tikv_data_disk_throughput"], 400, "mb/s") &&
				meteringValueMatches(point["tikv_log_disk_count"], 3, "count") &&
				meteringValueMatches(point["tikv_log_disk_iops"], 3000, "iops") &&
				meteringValueMatches(point["tikv_log_disk_throughput"], 125, "mb/s") {
				return true
			}
		}
	}
	return false
}

func meteringValueMatches(data json.RawMessage, value uint64, unit string) bool {
	var observed struct {
		Value uint64 `json:"value"`
		Unit  string `json:"unit"`
	}
	return json.Unmarshal(data, &observed) == nil && observed.Value == value && observed.Unit == unit
}

func downloadGzipObject(
	t *testing.T,
	ctx context.Context,
	credentials awsRoleCredentials,
	bucket string,
	key string,
	path string,
) []byte {
	t.Helper()
	runAWSWithCredentials(t, ctx, credentials, "s3api", "get-object", "--bucket", bucket, "--key", key, path)
	file, err := os.Open(path)
	if err != nil {
		t.Fatalf("open Premium metering object: %v", err)
	}
	defer file.Close()
	reader, err := gzip.NewReader(file)
	if err != nil {
		t.Fatalf("open Premium metering gzip: %v", err)
	}
	defer reader.Close()
	data, err := io.ReadAll(reader)
	if err != nil {
		t.Fatalf("read Premium metering gzip: %v", err)
	}
	return data
}

func premiumMeteringMetadataMatches(
	t *testing.T,
	ctx context.Context,
	credentials awsRoleCredentials,
	bucket string,
	keys []string,
	tempDir string,
) bool {
	t.Helper()
	for index, key := range keys {
		if !strings.HasPrefix(key, "metering/meta/") {
			continue
		}
		data := downloadGzipObject(t, ctx, credentials, bucket, key, filepath.Join(tempDir, fmt.Sprintf("metadata-%d.json.gz", index)))
		if jsonHasScalar(t, data, config.DefaultUIOrganizationID, "tenantId", "tenant_id") &&
			jsonHasScalar(t, data, "premium", "plan") {
			return true
		}
	}
	return false
}

func premiumMeteringDeltaHasExpectedRows(
	t *testing.T,
	ctx context.Context,
	credentials awsRoleCredentials,
	bucket string,
	keys []string,
	tempDir string,
) bool {
	t.Helper()
	versionsByTable := map[string]uint8{}
	for _, key := range keys {
		deltaIndex := strings.Index(key, "/_delta_log/")
		if deltaIndex < 0 || !strings.HasSuffix(key, ".json") {
			continue
		}
		versionText := strings.TrimSuffix(key[deltaIndex+len("/_delta_log/"):], ".json")
		version, err := strconv.ParseInt(versionText, 10, 64)
		if err != nil {
			continue
		}
		tablePrefix := key[:deltaIndex+1]
		if version == 0 {
			versionsByTable[tablePrefix] |= 1
		} else if version > 0 {
			versionsByTable[tablePrefix] |= 2
		}
	}
	for index, key := range keys {
		deltaIndex := strings.Index(key, "/_delta_log/")
		if deltaIndex < 0 || !strings.HasSuffix(key, ".json") {
			continue
		}
		versionText := strings.TrimSuffix(key[deltaIndex+len("/_delta_log/"):], ".json")
		version, err := strconv.ParseInt(versionText, 10, 64)
		if err != nil || version <= 0 || versionsByTable[key[:deltaIndex+1]] != 3 {
			continue
		}
		tablePrefix := key[:deltaIndex+1]
		path := filepath.Join(tempDir, fmt.Sprintf("delta-%d.json", index))
		runAWSWithCredentials(t, ctx, credentials, "s3api", "get-object", "--bucket", bucket, "--key", key, path)
		data, err := os.ReadFile(path)
		if err != nil {
			t.Fatalf("read Premium metering Delta log: %v", err)
		}
		for _, line := range bytes.Split(data, []byte("\n")) {
			var action struct {
				Add *struct {
					Path  string `json:"path"`
					Stats string `json:"stats"`
				} `json:"add"`
			}
			if len(bytes.TrimSpace(line)) == 0 || json.Unmarshal(line, &action) != nil || action.Add == nil {
				continue
			}
			var stats struct {
				NumRecords int `json:"numRecords"`
			}
			if json.Unmarshal([]byte(action.Add.Stats), &stats) == nil && stats.NumRecords == 4 &&
				action.Add.Path != "" &&
				premiumMeteringParquetRowsMatch(
					t,
					ctx,
					credentials,
					bucket,
					tablePrefix+action.Add.Path,
					filepath.Join(tempDir, fmt.Sprintf("parquet-%d.jsonl", index)),
				) {
				return true
			}
		}
	}
	return false
}

func premiumMeteringParquetRowsMatch(
	t *testing.T,
	ctx context.Context,
	credentials awsRoleCredentials,
	bucket string,
	key string,
	outputPath string,
) bool {
	t.Helper()
	if err := os.Remove(outputPath); err != nil && !os.IsNotExist(err) {
		t.Fatalf("replace Premium metering Parquet selection: %v", err)
	}
	runAWSWithCredentials(
		t,
		ctx,
		credentials,
		"s3api",
		"select-object-content",
		"--bucket",
		bucket,
		"--key",
		key,
		"--expression-type",
		"SQL",
		"--expression",
		"SELECT * FROM S3Object s",
		"--input-serialization",
		`{"Parquet":{}}`,
		"--output-serialization",
		`{"JSON":{"RecordDelimiter":"\n"}}`,
		outputPath,
	)
	data, err := os.ReadFile(outputPath)
	if err != nil {
		t.Fatalf("read Premium metering Parquet rows: %v", err)
	}
	rows := bytes.FieldsFunc(data, func(r rune) bool { return r == '\n' || r == '\r' })
	if len(rows) != 4 {
		return false
	}
	want := map[string]string{
		"disk-iops:aws:us-west-2:premium:row:standard:data:gp3":       "8000",
		"disk-throughput:aws:us-west-2:premium:row:standard:data:gp3": "400",
		"disk-iops:aws:us-west-2:premium:row:standard:raft:gp3":       "3000",
		"disk-throughput:aws:us-west-2:premium:row:standard:raft:gp3": "125",
	}
	for _, row := range rows {
		var observed struct {
			TenantID  uint64 `json:"tenant_id"`
			Unit      string `json:"unit"`
			UnitValue string `json:"unit_value"`
			Cost      int64  `json:"cost"`
		}
		if json.Unmarshal(row, &observed) != nil || observed.TenantID != 1440002 || observed.Cost <= 0 {
			return false
		}
		expectedValue, ok := want[observed.Unit]
		if !ok || observed.UnitValue != expectedValue {
			return false
		}
		delete(want, observed.Unit)
	}
	return len(want) == 0
}

func assumeAWSRole(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	roleARN string,
) awsRoleCredentials {
	t.Helper()
	data := runAWS(
		t,
		ctx,
		env,
		"sts",
		"assume-role",
		"--role-arn",
		roleARN,
		"--role-session-name",
		env.sessionName+"-metering",
		"--duration-seconds",
		"3600",
		"--output",
		"json",
	)
	var assumed struct {
		Credentials awsRoleCredentials `json:"Credentials"`
	}
	if err := json.Unmarshal(data, &assumed); err != nil || assumed.Credentials.AccessKey == "" ||
		assumed.Credentials.SecretKey == "" || assumed.Credentials.SessionToken == "" {
		t.Fatalf("assume Premium metering read-write role: %v", err)
	}
	return assumed.Credentials
}

func runAWSWithCredentials(
	t *testing.T,
	ctx context.Context,
	credentials awsRoleCredentials,
	args ...string,
) []byte {
	t.Helper()
	operation := awsOperation(args)
	args = append(args, "--region", targetRegion)
	command := exec.CommandContext(ctx, "aws", args...)
	command.Env = append(
		os.Environ(),
		"AWS_ACCESS_KEY_ID="+credentials.AccessKey,
		"AWS_SECRET_ACCESS_KEY="+credentials.SecretKey,
		"AWS_SESSION_TOKEN="+credentials.SessionToken,
		"AWS_PAGER=",
	)
	data, err := command.CombinedOutput()
	if err != nil {
		t.Fatalf("aws %s: %v (%d-byte output)", operation, err, len(data))
	}
	return data
}

func runPremiumProductLifecycle(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
	deletion *productDeletion,
	stagePrefix string,
) {
	if stagePrefix == "" {
		line := "premium"
		if orgID == config.EssentialV2UIOrganizationID {
			line = "essential-v2"
		}
		if !deletion.created {
			skippedScenarioStage(t, env, "ui-metrics", "borrowed Premium-family instance; UI Metrics is not claimed")
		} else {
			scenarioStage(t, env, "ui-metrics", func(t *testing.T) {
				runUIMetricsAcceptance(t, ctx, env, line, env.sessionName, orgID, projectID, clusterID, true)
			})
		}
	}
	if !deletion.created {
		if stagePrefix != "" {
			scenarioStage(t, env, stagePrefix+"delete-cluster", func(t *testing.T) {
				t.Fatal("restore delete requires a journaled run-owned Product ID")
			})
			return
		}
		skippedScenarioStage(t, env, stagePrefix+"delete-cluster", "borrowed Product; delete is not claimed")
		return
	}
	scenarioStage(t, env, stagePrefix+"delete-cluster", func(t *testing.T) {
		deletion.run(t, ctx)
	})
}

func configureDedicatedPublic(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) string {
	t.Helper()
	host := ensureDedicatedPublicEndpoint(t, ctx, client, env, clusterID)
	ensureDedicatedPublicAllowlist(t, ctx, client, env, clusterID)
	setDedicatedPassword(t, ctx, client, env, clusterID)
	return host
}

func dedicatedPublicEndpoint(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
	requireHost bool,
) (string, string, string) {
	t.Helper()
	endpointCtx, cancel := publicEndpointWaitContext(ctx)
	defer cancel()
	listPath := fmt.Sprintf(
		"/api/v2023/dedicated/v1beta1/clusters/%s/tidbNodeGroups?orgId=1440002&projectId=1440002003&pageSize=999",
		clusterID,
	)
	nodeGroupID, publicHost := "", ""
	for {
		response := doAPIRequest(t, endpointCtx, client, env.baseURL, http.MethodGet, listPath, nil, "", http.StatusOK)
		var groups struct {
			Groups []struct {
				ID        string `json:"tidbNodeGroupId"`
				Endpoints []struct {
					Host string `json:"host"`
					Type string `json:"connectionType"`
				} `json:"endpoints"`
			} `json:"tidbNodeGroups"`
		}
		if err := json.Unmarshal(response, &groups); err != nil {
			t.Fatalf("decode %d-byte Dedicated node groups response for GET %s: %v", len(response), listPath, err)
		}
		if len(groups.Groups) == 1 {
			nodeGroupID = groups.Groups[0].ID
			publicHost = ""
			for _, endpoint := range groups.Groups[0].Endpoints {
				if endpoint.Type == "PUBLIC" {
					publicHost = endpoint.Host
				}
			}
			if nodeGroupID != "" && (!requireHost || publicHost != "") {
				break
			}
		}
		waitForPublicEndpointRetry(t, endpointCtx, listPath, 5*time.Second)
	}
	settingPath := fmt.Sprintf(
		"/api/v2023/dedicated/v1beta1/clusters/%s/tidbNodeGroups/%s/publicEndpointSetting?orgId=1440002&projectId=1440002003",
		clusterID,
		nodeGroupID,
	)
	return nodeGroupID, publicHost, settingPath
}

func ensureDedicatedPublicEndpoint(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) string {
	t.Helper()
	_, _, settingPath := dedicatedPublicEndpoint(t, ctx, client, env, clusterID, false)
	response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, settingPath, nil, "", http.StatusOK)
	if !bytes.Contains(response, []byte(`"enabled":true`)) {
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPatch,
			settingPath,
			[]byte(`{"enabled":true}`),
			"application/json",
			http.StatusOK,
		)
		for {
			response = doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, settingPath, nil, "", http.StatusOK)
			if bytes.Contains(response, []byte(`"enabled":true`)) {
				break
			}
			select {
			case <-ctx.Done():
				t.Fatalf("wait for Dedicated public endpoint GET %s to become enabled: %v", settingPath, ctx.Err())
			case <-time.After(5 * time.Second):
			}
		}
	}
	_, publicHost, _ := dedicatedPublicEndpoint(t, ctx, client, env, clusterID, true)
	return publicHost
}

func ensureDedicatedPublicAllowlist(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) {
	t.Helper()
	_, _, settingPath := dedicatedPublicEndpoint(t, ctx, client, env, clusterID, false)
	response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, settingPath, nil, "", http.StatusOK)
	if jsonHasScalar(t, response, "0.0.0.0/0", "cidrNotation") {
		return
	}
	doAPIRequest(
		t,
		ctx,
		client,
		env.baseURL,
		http.MethodPatch,
		settingPath,
		[]byte(`{"ipAccessList":[{"cidrNotation":"0.0.0.0/0","description":"Allow access from anywhere."}]}`),
		"application/json",
		http.StatusOK,
	)
}

func setDedicatedPassword(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) {
	t.Helper()
	path := fmt.Sprintf("/api/v1/orgs/1440002/projects/1440002003/clusters/%s/password", clusterID)
	body := []byte(fmt.Sprintf(`{"password":%q}`, os.Getenv(envDatabasePassword)))
	doAPIRequest(t, ctx, client, env.baseURL, http.MethodPut, path, body, "application/json", http.StatusOK)
}

func configureNextGenPublic(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) string {
	t.Helper()
	host := ensureNextGenPublicEndpoint(t, ctx, client, env, orgID, projectID, clusterID)
	ensureNextGenPublicAllowlist(t, ctx, client, env, orgID, projectID, clusterID)
	setNextGenPassword(t, ctx, client, env, orgID, projectID, clusterID)
	return host
}

func nextGenPublicPaths(orgID string, projectID string, clusterID string) (string, string) {
	return fmt.Sprintf(
			"/api/v2025/cloud/v1beta2/tidbs/%s/publicConnectionSetting?orgId=%s&projectId=%s",
			clusterID,
			orgID,
			projectID,
		), fmt.Sprintf(
			"/api/v2025/cloud/v1beta2/tidbs/%s?orgId=%s&projectId=%s",
			clusterID,
			orgID,
			projectID,
		)
}

func ensureNextGenPublicEndpoint(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) string {
	t.Helper()
	endpointCtx, cancel := publicEndpointWaitContext(ctx)
	defer cancel()
	settingPath, clusterPath := nextGenPublicPaths(orgID, projectID, clusterID)
	response := doAPIRequest(t, endpointCtx, client, env.baseURL, http.MethodGet, settingPath, nil, "", http.StatusOK)
	if !bytes.Contains(response, []byte(`"enabled":true`)) {
		response, status := doAPIRequestStatuses(
			t,
			endpointCtx,
			client,
			env.baseURL,
			http.MethodPatch,
			settingPath,
			[]byte(`{"enabled":true}`),
			"application/json",
			http.StatusOK,
			http.StatusNotFound,
		)
		if status == http.StatusNotFound &&
			(!bytes.Contains(response, []byte("49900004")) ||
				!bytes.Contains(bytes.ToLower(response), []byte("dataplane not found"))) {
			t.Fatalf("enable NextGen public endpoint PATCH %s returned non-retryable HTTP 404: %s", settingPath, response)
		}
	}
	for {
		response := doAPIRequest(t, endpointCtx, client, env.baseURL, http.MethodGet, clusterPath, nil, "", http.StatusOK)
		var tidb struct {
			Endpoints           []nextGenEndpoint `json:"endpoints"`
			EssentialV2Instance struct {
				Endpoints []nextGenEndpoint `json:"endpoints"`
			} `json:"essentialV2Instance"`
		}
		if err := json.Unmarshal(response, &tidb); err != nil {
			t.Fatalf("decode %d-byte nextgen cluster response for GET %s: %v", len(response), clusterPath, err)
		}
		endpoints := tidb.Endpoints
		if len(endpoints) == 0 {
			endpoints = tidb.EssentialV2Instance.Endpoints
		}
		for _, endpoint := range endpoints {
			if endpoint.Type == "PUBLIC" && endpoint.Host != "" {
				return endpoint.Host
			}
		}
		waitForPublicEndpointRetry(t, endpointCtx, clusterPath, 30*time.Second)
	}
}

func ensureNextGenPublicAllowlist(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	t.Helper()
	settingPath, _ := nextGenPublicPaths(orgID, projectID, clusterID)
	response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, settingPath, nil, "", http.StatusOK)
	if jsonHasScalar(t, response, "0.0.0.0/0", "cidrNotation") {
		return
	}
	doAPIRequest(
		t,
		ctx,
		client,
		env.baseURL,
		http.MethodPatch,
		settingPath,
		[]byte(`{"ipAccessList":[{"cidrNotation":"0.0.0.0/0","description":"Allow access from anywhere."}]}`),
		"application/json",
		http.StatusOK,
	)
}

func setNextGenPassword(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	t.Helper()
	passwordPath := fmt.Sprintf(
		"/api/v2025/cloud/v1beta2/tidbs/%s:resetRootPassword?orgId=%s&projectId=%s",
		clusterID,
		orgID,
		projectID,
	)
	body := []byte(fmt.Sprintf(`{"rootPassword":%q}`, os.Getenv(envDatabasePassword)))
	doAPIRequest(t, ctx, client, env.baseURL, http.MethodPost, passwordPath, body, "application/json", http.StatusOK)
}

func testDedicatedPrivateLink(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) {
	scenarioStage(t, env, "private-link", func(t *testing.T) {
		info := getClusterInfo(t, ctx, env, config.DefaultUIOrganizationID, clusterID)
		nodeGroupID, _ := clusterEndpoint(info, false, "PUBLIC")
		if nodeGroupID == "" {
			t.Fatal("Dedicated cluster has no TiDB node group for PrivateLink")
		}
		connectionPath := fmt.Sprintf(
			"/api/v2023/dedicated/v1beta1/clusters/%s/tidbNodeGroups/%s/privateEndpointConnections?orgId=1440002&projectId=1440002003",
			clusterID,
			nodeGroupID,
		)
		servicePath := fmt.Sprintf(
			"/api/v2023/dedicated/v1beta1/clusters/%s/tidbNodeGroups/%s/privateLinkService?orgId=1440002&projectId=1440002003",
			clusterID,
			nodeGroupID,
		)
		serviceName := jsonScalar(t, waitForAPIState(t, ctx, client, env, servicePath, "ACTIVE", "FAILED"), "serviceName")
		endpointID := createVPCEndpoint(t, ctx, env, serviceName)
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			connectionPath,
			[]byte(fmt.Sprintf(`{"endpointId":%q}`, endpointID)),
			"application/json",
			http.StatusOK,
		)
		endpointIP := waitForVPCEndpoint(t, ctx, env, endpointID)
		privateHost := waitForPrivateEndpoint(t, ctx, client, env, connectionPath, endpointID)
		runFixtureMySQL(
			t,
			ctx,
			env,
			privateHost,
			endpointIP,
			"root",
			strictMySQLCA(t, ctx, env, false),
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.private_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				"INSERT INTO cp9_uiapi.private_sql VALUES (1, 'dedicated-private-tls') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value); "+
				"SELECT id,value FROM cp9_uiapi.private_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"dedicated-private-tls",
		)
	})
}

func testNextGenPrivateLink(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	scenarioStage(t, env, "private-link", func(t *testing.T) {
		connectionPath := fmt.Sprintf(
			"/api/v2025/cloud/v1beta2/tidbs/%s/privateEndpointConnections?orgId=%s&projectId=%s",
			clusterID,
			orgID,
			projectID,
		)
		servicePath := fmt.Sprintf(
			"/api/v2025/cloud/v1beta2/tidbs/%s/privateLinkService?orgId=%s&projectId=%s",
			clusterID,
			orgID,
			projectID,
		)
		serviceName := jsonScalar(t, waitForAPIState(t, ctx, client, env, servicePath, "ACTIVE", "FAILED"), "serviceName")
		endpointID := createVPCEndpoint(t, ctx, env, serviceName)
		body := []byte(fmt.Sprintf(`{"endpointId":%q,"regionId":"aws-us-west-2"}`, endpointID))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			connectionPath,
			body,
			"application/json",
			http.StatusOK,
		)
		endpointIP := waitForVPCEndpoint(t, ctx, env, endpointID)
		privateHost := waitForPrivateEndpoint(t, ctx, client, env, connectionPath, endpointID)
		runFixtureMySQL(
			t,
			ctx,
			env,
			privateHost,
			endpointIP,
			"root",
			strictNextGenMySQLCA(t, ctx, env, orgID, projectID, clusterID),
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.private_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				"INSERT INTO cp9_uiapi.private_sql VALUES (1, 'premium-private-tls') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value); "+
				"SELECT id,value FROM cp9_uiapi.private_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"premium-private-tls",
		)
	})
}

func findStringPrefix(value any, prefix string) string {
	switch value := value.(type) {
	case string:
		if strings.HasPrefix(value, prefix) {
			return value
		}
	case []any:
		for _, item := range value {
			if found := findStringPrefix(item, prefix); found != "" {
				return found
			}
		}
	case map[string]any:
		for _, item := range value {
			if found := findStringPrefix(item, prefix); found != "" {
				return found
			}
		}
	}
	return ""
}

func createVPCEndpoint(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	serviceName string,
) string {
	t.Helper()
	tagSpecification := fmt.Sprintf(
		"ResourceType=vpc-endpoint,Tags=[{Key=cp9-owner,Value=%s},{Key=cp9-session,Value=%s},{Key=Name,Value=%s-vpce}]",
		env.fixture.owner,
		env.sessionName,
		env.sessionName,
	)
	response := runAWS(
		t,
		ctx,
		env,
		"ec2",
		"create-vpc-endpoint",
		"--vpc-id",
		env.fixture.vpcID,
		"--service-name",
		serviceName,
		"--vpc-endpoint-type",
		"Interface",
		"--subnet-ids",
		env.fixture.subnetID,
		"--security-group-ids",
		env.fixture.securityGroup,
		"--tag-specifications",
		tagSpecification,
		"--output",
		"json",
	)
	var created struct {
		Endpoint struct {
			ID string `json:"VpcEndpointId"`
		} `json:"VpcEndpoint"`
	}
	if err := json.Unmarshal(response, &created); err != nil || created.Endpoint.ID == "" {
		t.Fatalf("decode create VPC endpoint response: %v\n%s", err, response)
	}
	return created.Endpoint.ID
}

func waitForVPCEndpoint(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	endpointID string,
) string {
	t.Helper()
	for {
		response := runAWS(
			t,
			ctx,
			env,
			"ec2",
			"describe-vpc-endpoints",
			"--vpc-endpoint-ids",
			endpointID,
			"--query",
			"VpcEndpoints[0]",
			"--output",
			"json",
		)
		var endpoint struct {
			NetworkInterfaces []string `json:"NetworkInterfaceIds"`
			State             string   `json:"State"`
		}
		if err := json.Unmarshal(response, &endpoint); err != nil {
			t.Fatalf("decode VPC endpoint response: %v\n%s", err, response)
		}
		if endpoint.State == "available" && len(endpoint.NetworkInterfaces) > 0 {
			privateIP := strings.TrimSpace(string(runAWS(
				t,
				ctx,
				env,
				"ec2",
				"describe-network-interfaces",
				"--network-interface-ids",
				endpoint.NetworkInterfaces[0],
				"--query",
				"NetworkInterfaces[0].PrivateIpAddress",
				"--output",
				"text",
			)))
			if privateIP != "" && privateIP != "None" {
				return privateIP
			}
		}
		if endpoint.State == "failed" || endpoint.State == "rejected" {
			t.Fatalf("VPC endpoint %s entered state %s", endpointID, endpoint.State)
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for VPC endpoint %s: %v", endpointID, ctx.Err())
		case <-time.After(20 * time.Second):
		}
	}
}

func waitForPrivateEndpoint(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	path string,
	endpointID string,
) string {
	t.Helper()
	for {
		response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
		var value any
		if err := json.Unmarshal(response, &value); err != nil {
			t.Fatalf("decode %d-byte private endpoint response for GET %s: %v", len(response), path, err)
		}
		if endpoint := findNamedObject(value, endpointID); endpoint != nil {
			state := strings.ToUpper(scalarFromValue(endpoint, "endpointState", "state"))
			if host := scalarFromValue(endpoint, "host"); state == "ACTIVE" && host != "" {
				return host
			}
			if state == "FAILED" || state == "REJECTED" {
				t.Fatalf("private endpoint %s entered state %s", endpointID, state)
			}
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for private endpoint GET %s: %v; last response was %d bytes", path, ctx.Err(), len(response))
		case <-time.After(30 * time.Second):
		}
	}
}

func runFixtureMySQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	host string,
	endpointIP string,
	username string,
	caPath string,
	sql string,
	marker string,
) {
	t.Helper()
	if host == "" || username == "" {
		t.Fatalf("fixture MySQL requires a journaled host and username")
	}
	remoteCA := "/tmp/" + env.sessionName + "-ca.pem"
	scp := exec.CommandContext(
		ctx,
		"scp",
		"-o",
		"BatchMode=yes",
		"-o",
		"StrictHostKeyChecking=accept-new",
		"-i",
		os.Getenv(envSSHPrivateKey),
		caPath,
		"ec2-user@"+env.fixture.publicIP+":"+remoteCA,
	)
	if result, err := scp.CombinedOutput(); err != nil {
		t.Fatalf("copy CA to fixture: %v\n%s", err, result)
	}
	runFixtureStrictMySQL(t, ctx, env, host, endpointIP, username, remoteCA, sql, marker)
}

func runFixtureStrictMySQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	host string,
	endpointIP string,
	username string,
	caPath string,
	sql string,
	marker string,
) {
	t.Helper()
	hostOverride := ""
	if endpointIP != "" {
		hostOverride = fmt.Sprintf(
			"sudo sed -i '/ %s$/d' /etc/hosts && printf '%%s %%s\\n' '%s' '%s' | sudo tee -a /etc/hosts >/dev/null && ",
			host,
			endpointIP,
			host,
		)
	}
	remoteCommand := hostOverride + fmt.Sprintf(
		"MYSQL_PWD='%s' mysql --batch --skip-column-names --host '%s' --port 4000 --user '%s' "+
			"--connect-timeout=5 --ssl-ca='%s' --ssl-verify-server-cert --execute=%q",
		os.Getenv(envDatabasePassword),
		host,
		username,
		caPath,
		sql,
	)
	mysqlCtx, cancel := context.WithTimeout(ctx, 20*time.Minute)
	defer cancel()
	for {
		ssh := exec.CommandContext(
			mysqlCtx,
			"ssh",
			"-o",
			"BatchMode=yes",
			"-o",
			"StrictHostKeyChecking=accept-new",
			"-i",
			os.Getenv(envSSHPrivateKey),
			"ec2-user@"+env.fixture.publicIP,
			remoteCommand,
		)
		env.fixture.mu.Lock()
		result, err := ssh.CombinedOutput()
		env.fixture.mu.Unlock()
		if err == nil {
			t.Logf("strict fixture MySQL result for %s:\n%s", host, result)
			if !bytes.Contains(result, []byte(marker)) || !bytes.Contains(result, []byte("TLS_")) {
				t.Fatalf("strict fixture MySQL result for %s did not contain marker and TLS cipher:\n%s", host, result)
			}
			return
		}
		select {
		case <-mysqlCtx.Done():
			t.Fatalf("strict fixture MySQL failed for %s: %v\n%s", host, mysqlCtx.Err(), result)
		case <-time.After(20 * time.Second):
		}
	}
}

func testDedicatedImport(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
	publicHost string,
) {
	scenarioStage(t, env, "s3-import", func(t *testing.T) {
		caPath := strictMySQLCA(t, ctx, env, false)
		runMySQL(
			t,
			ctx,
			publicHost,
			caPath,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"DROP TABLE IF EXISTS cp9_uiapi.proof; "+
				"CREATE TABLE cp9_uiapi.proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);",
		)
		basePath := fmt.Sprintf(
			"/api/v1/orgs/1440002/projects/1440002003/clusters/%s/imports",
			clusterID,
		)
		sourceURI := "s3://" + env.fixture.bucket + "/inputs/"
		validateBody := []byte(fmt.Sprintf(
			`{"source_url":%q,"region":"us-west-2","data_source_type":1,"data_format":3,"aws_role_arn":%q,"check_items":["SOURCE_ACCESS_CHECK"]}`,
			sourceURI,
			env.fixture.importRoleARN,
		))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/validate",
			validateBody,
			"application/json",
			http.StatusOK,
		)
		precheckBody := []byte(fmt.Sprintf(
			`{"import_options":{"file_type":"SQL"},"source":{"type":"S3","s3":{"uri":%q,"auth_type":"ROLE_ARN","role_arn":{"role_arn":%q,"external_id":"1440002003"},"target_table_infos":[{"target_table":{"schema":"cp9_uiapi","table":"proof"},"custom_file":"proof.sql"}]}}}`,
			sourceURI,
			env.fixture.importRoleARN,
		))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+":precheck",
			precheckBody,
			"application/json",
			http.StatusOK,
		)
		importBody := []byte(fmt.Sprintf(
			`{"source_url":%q,"region":"us-west-2","data_source_type":1,"data_format":1,"aws_role_arn":%q,"target_table_infos":[{"target_table":{"schema":"cp9_uiapi","table":"proof"},"custom_file":"proof.sql"}]}`,
			sourceURI,
			env.fixture.importRoleARN,
		))
		validateBody = append(
			bytes.TrimSuffix(importBody, []byte("}")),
			[]byte(`,"check_items":["IMPORT_PRECHECK"]}`)...,
		)
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/validate",
			validateBody,
			"application/json",
			http.StatusOK,
		)
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPut,
			basePath,
			importBody,
			"application/json",
			http.StatusOK,
		)
		importID := jsonScalar(t, response, "id")
		waitForAPIState(t, ctx, client, env, basePath+"/"+importID, "COMPLETED", "FAILED")
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			false,
			"SELECT id,value FROM cp9_uiapi.proof; SHOW STATUS LIKE 'Ssl_cipher';",
			"public-ui-api-proof",
		)
	})
}

func testNextGenImport(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
	publicHost string,
) {
	scenarioStage(t, env, "s3-import", func(t *testing.T) {
		caPath := strictNextGenMySQLCA(t, ctx, env, orgID, projectID, clusterID)
		runMySQL(
			t,
			ctx,
			publicHost,
			caPath,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"DROP TABLE IF EXISTS cp9_uiapi.proof; "+
				"CREATE TABLE cp9_uiapi.proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);",
		)
		basePath := fmt.Sprintf("/api/v2025/cloud/v1beta2/tidbs/%s/imports", clusterID)
		query := fmt.Sprintf("?orgId=%s&projectId=%s", orgID, projectID)
		sourceURI := "s3://" + env.fixture.bucket + "/inputs/"
		source := fmt.Sprintf(
			`{"type":"ENUM_S3","s3":{"uri":%q,"auth_type":"ENUM_ROLE_ARN","role_arn":%q}}`,
			sourceURI,
			env.fixture.importRoleARN,
		)
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+":checkImportSourceConnection"+query,
			[]byte(`{"source":`+source+`}`),
			"application/json",
			http.StatusOK,
		)
		precheckResponse := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+":checkSourceAsync"+query,
			[]byte(fmt.Sprintf(
				`{"source":%s,"tableMetas":[{"database":"cp9_uiapi","table":"proof","filesWildcard":%q}]}`,
				source,
				sourceURI+"proof.sql",
			)),
			"application/json",
			http.StatusOK,
		)
		taskID := jsonScalar(t, precheckResponse, "taskId")
		var tableMetas json.RawMessage
		for {
			path := basePath + "/checkSourceAsync/" + url.PathEscape(taskID) + query
			response := doAPIRequest(
				t,
				ctx,
				client,
				env.baseURL,
				http.MethodGet,
				path,
				nil,
				"",
				http.StatusOK,
			)
			var result struct {
				ErrorMsg  string          `json:"errorMsg"`
				Status    string          `json:"status"`
				Success   bool            `json:"success"`
				TableMeta json.RawMessage `json:"tableMetas"`
			}
			if err := json.Unmarshal(response, &result); err != nil {
				t.Fatalf("decode %d-byte nextgen import precheck GET %s: %v", len(response), path, err)
			}
			switch result.Status {
			case "ENUM_SUCCESS", "ENUM_TRUNCATED":
				if !result.Success || len(result.TableMeta) == 0 ||
					bytes.Equal(bytes.TrimSpace(result.TableMeta), []byte("[]")) {
					t.Fatalf("nextgen import precheck GET %s returned no usable table metadata", path)
				}
				tableMetas = result.TableMeta
			case "ENUM_RUNNING", "ENUM_UNSPECIFIED":
			default:
				t.Fatalf("nextgen import precheck GET %s entered %s", path, result.Status)
			}
			if len(tableMetas) > 0 {
				break
			}
			select {
			case <-ctx.Done():
				t.Fatalf(
					"wait for nextgen import precheck task %s: %v; last server error was %d bytes",
					taskID,
					ctx.Err(),
					len(result.ErrorMsg),
				)
			case <-time.After(10 * time.Second):
			}
		}
		credentials := fmt.Sprintf(
			`{"username":"root","password":%q}`,
			os.Getenv(envDatabasePassword),
		)
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+":checkDatabaseConnectivity"+query,
			[]byte(credentials),
			"application/json",
			http.StatusOK,
		)
		groupKey := env.sessionName + "-nextgen-import"
		body := []byte(fmt.Sprintf(
			`{"username":"root","password":%q,"groupKey":%q,"creationDetails":{"importOptions":{"csvFormat":{}},"source":%s,"targetTableInfos":%s}}`,
			os.Getenv(envDatabasePassword),
			groupKey,
			source,
			tableMetas,
		))
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+query,
			body,
			"application/json",
			http.StatusOK,
		)
		importID := jsonScalar(t, response, "importId")
		waitForAPIState(
			t,
			ctx,
			client,
			env,
			basePath+"/"+importID+query,
			"ENUM_COMPLETED",
			"ENUM_FAILED",
		)
		waitForStrictMySQLWithCA(
			t,
			ctx,
			env,
			publicHost,
			caPath,
			"SELECT id,value FROM cp9_uiapi.proof; SHOW STATUS LIKE 'Ssl_cipher';",
			"public-ui-api-proof",
		)
	})
}

func testNextGenExport(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	scenarioStage(t, env, "s3-export", func(t *testing.T) {
		basePath := fmt.Sprintf("/api/v2025/cloud/v1beta2/tidbs/%s/exports", clusterID)
		query := fmt.Sprintf("?orgId=%s&projectId=%s", orgID, projectID)
		prefix := "outputs/" + clusterID + "/"
		body := []byte(fmt.Sprintf(
			`{"displayName":%q,"username":"root","password":%q,"exportOptions":{"fileType":"SQL","compression":"ZSTD","filter":{"table":{"patterns":["cp9_uiapi.*"]}}},"cloudStorage":{"type":"S3","s3":{"uri":%q,"authType":"ROLE_ARN","roleArn":%q}}}`,
			env.sessionName+"-export",
			os.Getenv(envDatabasePassword),
			"s3://"+env.fixture.bucket+"/"+prefix,
			env.fixture.importRoleARN,
		))
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+query,
			body,
			"application/json",
			http.StatusOK,
		)
		exportID := jsonScalar(t, response, "exportId")
		waitForAPIState(t, ctx, client, env, basePath+"/"+exportID+query, "SUCCEEDED", "FAILED")
		objects := strings.TrimSpace(string(runAWS(
			t,
			ctx,
			env,
			"s3api",
			"list-objects-v2",
			"--bucket",
			env.fixture.bucket,
			"--prefix",
			prefix,
			"--query",
			"length(Contents)",
			"--output",
			"text",
		)))
		if objects == "" || objects == "0" || objects == "None" {
			t.Fatalf("nextgen export %s wrote no objects to s3://%s/%s", exportID, env.fixture.bucket, prefix)
		}
	})
}

func jsonScalar(t *testing.T, response []byte, key string) string {
	t.Helper()
	var value any
	if err := json.Unmarshal(response, &value); err != nil {
		t.Fatalf("decode %d-byte JSON response: %v", len(response), err)
	}
	if found := findJSONScalar(value, key); found != "" {
		return found
	}
	t.Fatalf("%d-byte JSON response has no %q", len(response), key)
	return ""
}

func findJSONScalar(value any, key string) string {
	switch value := value.(type) {
	case []any:
		for _, item := range value {
			if found := findJSONScalar(item, key); found != "" {
				return found
			}
		}
	case map[string]any:
		if item, ok := value[key]; ok {
			switch item := item.(type) {
			case string:
				return item
			case float64:
				return fmt.Sprintf("%.0f", item)
			}
		}
		for _, item := range value {
			if found := findJSONScalar(item, key); found != "" {
				return found
			}
		}
	}
	return ""
}

func jsonHasScalar(t *testing.T, response []byte, want string, keys ...string) bool {
	t.Helper()
	decoder := json.NewDecoder(bytes.NewReader(response))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		t.Fatalf("decode %d-byte JSON response: %v", len(response), err)
	}
	return hasJSONScalar(value, want, keys)
}

func hasJSONScalar(value any, want string, keys []string) bool {
	switch value := value.(type) {
	case []any:
		for _, item := range value {
			if hasJSONScalar(item, want, keys) {
				return true
			}
		}
	case map[string]any:
		for key, item := range value {
			if slices.Contains(keys, key) {
				switch item := item.(type) {
				case string:
					if item == want {
						return true
					}
				case json.Number:
					if item.String() == want {
						return true
					}
				}
			}
			if hasJSONScalar(item, want, keys) {
				return true
			}
		}
	}
	return false
}

func waitForAPIState(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	path string,
	success string,
	failure string,
) []byte {
	t.Helper()
	for {
		response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
		var value any
		if err := json.Unmarshal(response, &value); err != nil {
			t.Fatalf("decode %d-byte response for GET %s: %v", len(response), path, err)
		}
		state := strings.ToUpper(scalarFromValue(value, "state", "status"))
		if strings.Contains(state, success) {
			return response
		}
		if strings.Contains(state, failure) {
			t.Fatalf("GET %s entered %s", path, failure)
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for GET %s to enter %s: %v; last response was %d bytes", path, success, ctx.Err(), len(response))
		case <-time.After(30 * time.Second):
		}
	}
}

func testDedicatedChangefeed(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
	publicHost string,
) {
	database := "changefeed_sink_ded"
	path := fmt.Sprintf(
		"/api/v1/orgs/1440002/projects/1440002003/clusters/%s/connectors",
		clusterID,
	)
	listPath := path + "?page=1&per_page=10"
	name := env.sessionName + "-ded-cdc"
	connectorID := publicUIRunResourceID(env, "dedicated", "changefeed")
	scenarioStage(t, env, "changefeed-create", func(t *testing.T) {
		if connector := getNamedObject(t, ctx, client, env, listPath, name); connector != nil {
			connectorID = reconcilePublicUIRunResourceID(
				t, env, "dedicated", "changefeed", connector, "id", "connectorId", "connector_id",
			)
			return
		}
		caPath := strictMySQLCA(t, ctx, env, false)
		runMySQL(
			t,
			ctx,
			publicHost,
			caPath,
			"CREATE DATABASE IF NOT EXISTS "+database+"; "+
				"CREATE TABLE IF NOT EXISTS "+database+
				".proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);",
		)
		body := []byte(fmt.Sprintf(
			`{"changefeed_instance_type":"2rcu","connector_type":3,"mysql_info":{"connection":{"endpoint":%q,"username":"cp9","password":%q},"start_position":{},"table_config":{"filter_rule":[%q],"mode":1}},"name":%q,"network_info":{}}`,
			env.fixture.publicIP+":3306",
			os.Getenv(envDatabasePassword),
			database+".*",
			name,
		))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			path,
			body,
			"application/json",
			http.StatusOK,
		)
	})
	scenarioStage(t, env, "changefeed-wait-running", func(t *testing.T) {
		connector := waitForNamedObjectState(
			t,
			ctx,
			client,
			env,
			listPath,
			name,
			"RUNNING",
			"ERROR",
		)
		connectorID = reconcilePublicUIRunResourceID(
			t, env, "dedicated", "changefeed", connector, "id", "connectorId", "connector_id",
		)
	})
	scenarioStage(t, env, "changefeed-sink-proof", func(t *testing.T) {
		connectorID = reconcilePublicUIRunResourceID(
			t,
			env,
			"dedicated",
			"changefeed",
			getNamedObject(t, ctx, client, env, listPath, name),
			"id",
			"connectorId",
			"connector_id",
		)
		caPath := strictMySQLCA(t, ctx, env, false)
		runMySQL(
			t,
			ctx,
			publicHost,
			caPath,
			"INSERT INTO "+database+".proof VALUES (1, 'dedicated-changefeed-proof') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value);",
		)
		waitForFixtureSQL(
			t,
			ctx,
			env,
			"SELECT value FROM "+database+".proof WHERE id=1",
			"dedicated-changefeed-proof",
		)
	})
	scenarioStage(t, env, "changefeed-delete", func(t *testing.T) {
		connector := getNamedObject(t, ctx, client, env, listPath, name)
		if connector == nil {
			return
		}
		connectorID = reconcilePublicUIRunResourceID(
			t, env, "dedicated", "changefeed", connector, "id", "connectorId", "connector_id",
		)
		doAPIRequest(t, ctx, client, env.baseURL, http.MethodDelete, path+"/"+connectorID, nil, "", http.StatusOK)
	})
	scenarioStage(t, env, "changefeed-absent", func(t *testing.T) {
		waitForNamedObjectAbsent(t, ctx, client, env, listPath, name)
	})
}

func testPremiumChangefeed(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
	publicHost string,
) {
	database := "changefeed_sink_prem"
	path := fmt.Sprintf("/api/v1beta2/tidbs/%s/changefeeds", clusterID)
	listPath := path + "?page=1&per_page=20"
	name := env.sessionName + "-prem-cdc"
	changefeedID := publicUIRunResourceID(env, "premium", "changefeed")
	scenarioStage(t, env, "changefeed-create", func(t *testing.T) {
		if changefeed := getNamedObject(t, ctx, client, env, listPath, name); changefeed != nil {
			changefeedID = reconcilePublicUIRunResourceID(
				t, env, "premium", "changefeed", changefeed, "id", "changefeedId", "changefeed_id",
			)
			return
		}
		caPath := strictMySQLCA(t, ctx, env, true)
		runMySQL(
			t,
			ctx,
			publicHost,
			caPath,
			"CREATE DATABASE IF NOT EXISTS "+database+"; "+
				"CREATE TABLE IF NOT EXISTS "+database+
				".proof (id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL);",
		)
		body := []byte(fmt.Sprintf(
			`{"changefeed_type":"MYSQL","network_info":{"network_type":"NetworkType_Public"},"mysql_info":{"start_position":{"mode":"FROM_NOW"},"connection":{"endpoint":%q,"username":"cp9","password":%q},"table_config":{"filter_rule":[%q],"mode":"FORCE_SYNC","event_filter_rule":[],"case_sensitive":false}},"changefeed_spec":"8rcu","name":%q}`,
			env.fixture.publicIP+":3306",
			os.Getenv(envDatabasePassword),
			database+".*",
			name,
		))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			path,
			body,
			"application/json",
			http.StatusOK,
		)
	})
	scenarioStage(t, env, "changefeed-wait-running", func(t *testing.T) {
		changefeed := waitForNamedObjectState(
			t,
			ctx,
			client,
			env,
			listPath,
			name,
			"RUNNING",
			"ERROR",
		)
		changefeedID = reconcilePublicUIRunResourceID(
			t, env, "premium", "changefeed", changefeed, "id", "changefeedId", "changefeed_id",
		)
	})
	scenarioStage(t, env, "changefeed-sink-proof", func(t *testing.T) {
		changefeedID = reconcilePublicUIRunResourceID(
			t,
			env,
			"premium",
			"changefeed",
			getNamedObject(t, ctx, client, env, listPath, name),
			"id",
			"changefeedId",
			"changefeed_id",
		)
		caPath := strictMySQLCA(t, ctx, env, true)
		runMySQL(
			t,
			ctx,
			publicHost,
			caPath,
			"INSERT INTO "+database+".proof VALUES (1, 'premium-changefeed-proof') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value);",
		)
		waitForFixtureSQL(
			t,
			ctx,
			env,
			"SELECT value FROM "+database+".proof WHERE id=1",
			"premium-changefeed-proof",
		)
	})
	scenarioStage(t, env, "changefeed-delete", func(t *testing.T) {
		changefeed := getNamedObject(t, ctx, client, env, listPath, name)
		if changefeed == nil {
			return
		}
		changefeedID = reconcilePublicUIRunResourceID(
			t, env, "premium", "changefeed", changefeed, "id", "changefeedId", "changefeed_id",
		)
		doAPIRequest(t, ctx, client, env.baseURL, http.MethodDelete, path+"/"+changefeedID, nil, "", http.StatusOK)
	})
	scenarioStage(t, env, "changefeed-absent", func(t *testing.T) {
		waitForNamedObjectAbsent(t, ctx, client, env, listPath, name)
	})
}

func testDedicatedMigration(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
	publicHost string,
) {
	basePath := fmt.Sprintf(
		"/api/v1/orgs/1440002/projects/1440002003/clusters/%s/dataflow",
		clusterID,
	)
	listPath := basePath + "/dm/jobs?page=0&page_size=10"
	name := env.sessionName + "-ded-migration"
	migrationID := publicUIRunResourceID(env, "dedicated", "migration")
	scenarioStage(t, env, "migration-create", func(t *testing.T) {
		if migration := getNamedObject(t, ctx, client, env, listPath, name); migration != nil {
			migrationID = reconcilePublicUIRunResourceID(
				t, env, "dedicated", "migration", migration, "id", "jobId", "job_id",
			)
			return
		}
		source := fmt.Sprintf(
			`{"ba_rules":{"do_tables":[{"schema":"migration_source_ded","table":"~.*"}]},"binlog_gtid":"","binlog_name":"","binlog_pos":0,"conn_profile":{"conn_method":"direct","host":%q,"password":%q,"port":3306,"type":"mysql","user":"cp9"}}`,
			env.fixture.publicIP,
			os.Getenv(envDatabasePassword),
		)
		job := fmt.Sprintf(
			`{"cluster_password":%q,"cluster_user":"root","import_mode":"logical","mode":"full","name":%q,"sources":[%s]}`,
			os.Getenv(envDatabasePassword),
			name,
			source,
		)
		precheckResponse := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/dm/pre-check",
			[]byte(job),
			"application/json",
			http.StatusOK,
		)
		precheckID := jsonScalar(t, precheckResponse, "id")
		profileBody := []byte(fmt.Sprintf(
			`{"conn_method":"direct","host":%q,"name":%q,"password":%q,"port":3306,"type":"mysql","user":"cp9"}`,
			env.fixture.publicIP,
			name,
			os.Getenv(envDatabasePassword),
		))
		profileResponse := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/conn-profiles",
			profileBody,
			"application/json",
			http.StatusOK,
		)
		profileID := jsonScalar(t, profileResponse, "id")
		job = strings.Replace(
			job,
			`"conn_profile":`,
			`"conn_profile_id":`+profileID+`,"conn_profile":`,
			1,
		)
		job = strings.TrimSuffix(job, "}") +
			`,"precheck_id":` + precheckID + `,"spec_id":2}`
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/dm/jobs",
			[]byte(job),
			"application/json",
			http.StatusOK,
		)
	})
	scenarioStage(t, env, "migration-wait-finished", func(t *testing.T) {
		migration := waitForNamedObjectState(t, ctx, client, env, listPath, name, "FINISHED", "ERROR")
		migrationID = reconcilePublicUIRunResourceID(
			t, env, "dedicated", "migration", migration, "id", "jobId", "job_id",
		)
	})
	scenarioStage(t, env, "migration-sink-proof", func(t *testing.T) {
		migrationID = reconcilePublicUIRunResourceID(
			t,
			env,
			"dedicated",
			"migration",
			getNamedObject(t, ctx, client, env, listPath, name),
			"id",
			"jobId",
			"job_id",
		)
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			false,
			"SELECT value FROM migration_source_ded.proof WHERE id=1; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"migration-dedicated",
		)
	})
	scenarioStage(t, env, "migration-delete", func(t *testing.T) {
		migration := getNamedObject(t, ctx, client, env, listPath, name)
		if migration == nil {
			return
		}
		migrationID = reconcilePublicUIRunResourceID(
			t, env, "dedicated", "migration", migration, "id", "jobId", "job_id",
		)
		doAPIRequest(t, ctx, client, env.baseURL, http.MethodDelete, basePath+"/dm/jobs/"+migrationID, nil, "", http.StatusOK)
	})
	scenarioStage(t, env, "migration-absent", func(t *testing.T) {
		waitForNamedObjectAbsent(t, ctx, client, env, listPath, name)
	})
}

func testPremiumMigration(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
	publicHost string,
) {
	path := fmt.Sprintf(
		"/api/v2025/cloud/v1beta2/tidbs/%s/premiumMigrations",
		clusterID,
	)
	name := env.sessionName + "-prem-migration"
	migrationID := publicUIRunResourceID(env, "premium", "migration")
	scenarioStage(t, env, "migration-create", func(t *testing.T) {
		if migration := getNamedObject(t, ctx, client, env, path, name); migration != nil {
			migrationID = reconcilePublicUIRunResourceID(
				t, env, "premium", "migration", migration, "id", "migrationId", "migration_id",
			)
			return
		}
		body := []byte(fmt.Sprintf(
			`{"display_name":%q,"sources":[{"conn_profile":{"conn_type":"PUBLIC","port":3306,"user":"cp9","password":%q,"host":%q},"source_type":"MYSQL","premium_migration_rules":[{"source_table":{"schema_pattern":"migration_source_prem","table_pattern":"*"},"target_table":{"schema":"migration_source_prem"}}]}],"import_mode":"IMPORTMODE_LOGICAL","target":{"user":"root","password":%q},"mode":"ALL"}`,
			name,
			os.Getenv(envDatabasePassword),
			env.fixture.publicIP,
			os.Getenv(envDatabasePassword),
		))
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			path,
			body,
			"application/json",
			http.StatusOK,
		)
		var value any
		if err := json.Unmarshal(response, &value); err != nil {
			t.Fatalf("decode %d-byte Premium migration response: %v", len(response), err)
		}
		migrationID = findStringPrefix(value, "dmtsk")
		if migrationID == "" {
			migrationID = jsonScalar(t, response, "id")
		}
		recordPublicUIRunResourceID(t, env, "premium", "migration", migrationID)
	})
	scenarioStage(t, env, "migration-sink-proof", func(t *testing.T) {
		migrationID = reconcilePublicUIRunResourceID(
			t,
			env,
			"premium",
			"migration",
			getNamedObject(t, ctx, client, env, path, name),
			"id",
			"migrationId",
			"migration_id",
		)
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodGet,
			path+"/"+migrationID,
			nil,
			"",
			http.StatusOK,
		)
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			true,
			"SELECT value FROM migration_source_prem.proof WHERE id=1; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"migration-premium",
		)
		runFixtureSQL(
			t,
			ctx,
			env,
			"INSERT INTO migration_source_prem.proof VALUES "+
				"(2, 'migration-premium-incremental') "+
				"ON DUPLICATE KEY UPDATE value=VALUES(value)",
		)
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			true,
			"SELECT value FROM migration_source_prem.proof WHERE id=2; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			"migration-premium-incremental",
		)
	})
	scenarioStage(t, env, "migration-delete", func(t *testing.T) {
		if migration := getNamedObject(t, ctx, client, env, path, name); migration != nil {
			migrationID = reconcilePublicUIRunResourceID(
				t, env, "premium", "migration", migration, "id", "migrationId", "migration_id",
			)
		} else if migrationID == "" {
			return
		}
		_, status := doAPIRequestStatuses(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodGet,
			path+"/"+migrationID,
			nil,
			"",
			http.StatusOK,
			http.StatusNotFound,
		)
		if status == http.StatusNotFound {
			return
		}
		doAPIRequest(t, ctx, client, env.baseURL, http.MethodDelete, path+"/"+migrationID, nil, "", http.StatusOK)
	})
	scenarioStage(t, env, "migration-absent", func(t *testing.T) {
		if migrationID == "" {
			t.Fatal("Premium migration absence requires a journaled migration ID")
		}
		deleteCtx, cancel := context.WithTimeout(ctx, 15*time.Minute)
		defer cancel()
		waitForAPIObjectAbsent(t, deleteCtx, client, env, path+"/"+migrationID)
	})
}

func runFixtureSQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	sql string,
) []byte {
	t.Helper()
	command := exec.CommandContext(
		ctx,
		"ssh",
		"-o",
		"BatchMode=yes",
		"-o",
		"StrictHostKeyChecking=accept-new",
		"-i",
		os.Getenv(envSSHPrivateKey),
		"ec2-user@"+env.fixture.publicIP,
		"MYSQL_PWD='"+os.Getenv(envDatabasePassword)+
			"' mysql --batch --skip-column-names --host 127.0.0.1 --user cp9 --execute="+
			fmt.Sprintf("%q", sql),
	)
	result, err := command.CombinedOutput()
	if err != nil {
		t.Fatalf("fixture MariaDB query failed: %v\n%s", err, result)
	}
	return result
}

func waitForFixtureSQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	sql string,
	marker string,
) {
	t.Helper()
	for {
		result := runFixtureSQL(t, ctx, env, sql)
		if bytes.Contains(result, []byte(marker)) {
			return
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for fixture MariaDB marker %s: %v\n%s", marker, ctx.Err(), result)
		case <-time.After(10 * time.Second):
		}
	}
}

func waitForStrictMySQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	host string,
	premium bool,
	sql string,
	marker string,
) {
	t.Helper()
	waitForStrictMySQLWithCA(t, ctx, env, host, strictMySQLCA(t, ctx, env, premium), sql, marker)
}

func waitForStrictMySQLWithCA(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	host string,
	caPath string,
	sql string,
	marker string,
) {
	t.Helper()
	if host == "" {
		t.Fatal("strict MySQL requires a journaled public host")
	}
	for {
		mysql := exec.CommandContext(
			ctx,
			"mysql",
			"--host",
			host,
			"--port",
			"4000",
			"--user",
			"root",
			"--ssl-mode=VERIFY_IDENTITY",
			"--ssl-ca="+caPath,
			"--execute="+sql,
		)
		mysql.Env = append(os.Environ(), "MYSQL_PWD="+os.Getenv(envDatabasePassword))
		result, err := mysql.CombinedOutput()
		if err == nil && bytes.Contains(result, []byte(marker)) &&
			bytes.Contains(result, []byte("TLS_")) {
			return
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for strict MySQL marker %s on %s: %v\n%s", marker, host, ctx.Err(), result)
		case <-time.After(20 * time.Second):
		}
	}
}

func testDedicatedBackupRestore(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) {
	restoreID, restoreOwned := loadPublicUIRunProduct(t, env, "dedicated-restore", "")
	restoreDeletion := productDeletion{request: func(t *testing.T, deleteCtx context.Context) {
		doAPIRequest(
			t,
			deleteCtx,
			client,
			env.baseURL,
			http.MethodDelete,
			fmt.Sprintf(
				"/api/v2023/dedicated/v1beta1/clusters/%s?orgId=%s&projectId=%s",
				restoreID,
				dedicatedOrgID,
				dedicatedProjectID,
			),
			nil,
			"",
			http.StatusOK,
		)
	}, created: restoreOwned}
	scenarioStage(t, env, "backup-restore", func(t *testing.T) {
		if restoreID != "" {
			t.Fatal("journaled Dedicated restore Product ID has no passed backup-restore receipt")
		}
		basePath := fmt.Sprintf(
			"/api/v1/orgs/1440002/projects/1440002003/clusters/%s/backups",
			clusterID,
		)
		backupName := env.sessionName + "-ded-backup"
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/schedules",
			[]byte(fmt.Sprintf(
				`{"name":%q,"description":"public UI API proof","backup_type":"MANUAL","backup_time":%q,"backup_mode":"SNAPSHOT","retention_period":7}`,
				backupName,
				time.Now().UTC().Format(time.RFC3339),
			)),
			"application/json",
			http.StatusOK,
		)
		backup := waitForNamedObjectState(
			t,
			ctx,
			client,
			env,
			basePath+"?page=1&per_page=10",
			backupName,
			"SUCCESS",
			"FAIL",
		)
		backupID := scalarFromValue(backup, "id", "backupId")
		if backupID == "" {
			t.Fatalf("Dedicated backup %s has no ID", backupName)
		}
		restoreName := env.sessionName + "-ded-restore"
		body := []byte(fmt.Sprintf(
			`{"restore_mode":"RESTORE_MODE_SNAPSHOT","cluster_name":%q,"port":4000,"region":"us-west-2","components":[{"component_type":"COMPONENT_TYPE_TIDB","node_profile_name":"8C16G-V2","replicas":2,"storage_size_gi":0},{"component_type":"COMPONENT_TYPE_TIKV","node_profile_name":"8C64G-V2","replicas":3,"storage_size_gi":500,"storage_type":"Standard"}]}`,
			restoreName,
		))
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+"/"+backupID+"/restore",
			body,
			"application/json",
			http.StatusOK,
		)
		restoreID = jsonScalarAny(t, response, "clusterId", "cluster_id", "tidbId")
		recordPublicUIRunProduct(t, env, "dedicated-restore", restoreID)
		restoreDeletion.created = true
		waitForClusterActive(t, ctx, env, "1440002", restoreID)
		publicHost := configureDedicatedPublic(t, ctx, client, env, restoreID)
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			false,
			"SELECT id,value FROM cp9_uiapi.proof; SHOW STATUS LIKE 'Ssl_cipher';",
			"public-ui-api-proof",
		)
	})
	if env.runJournal.stageStatus(t.Name()+"/backup-restore") == "passed" &&
		(restoreID == "" || !restoreDeletion.created) {
		t.Fatal("passed Dedicated backup-restore receipt lacks a run-owned restore Product ID")
	}
	runDedicatedProductLifecycle(t, ctx, env, restoreID, &restoreDeletion, "restore-")
}

func testPremiumBackupRestore(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	clusterID string,
) {
	restoreID, restoreOwned := loadPublicUIRunProduct(t, env, "premium-restore", "")
	restoreDeletion := productDeletion{request: func(t *testing.T, deleteCtx context.Context) {
		doAPIRequest(
			t,
			deleteCtx,
			client,
			env.baseURL,
			http.MethodDelete,
			fmt.Sprintf(
				"/api/v2025/cloud/v1beta2/tidbs/%s?orgId=%s&projectId=%s",
				restoreID,
				config.DefaultUIOrganizationID,
				config.DefaultUIOrganizationID+"0002",
			),
			nil,
			"",
			http.StatusOK,
		)
	}, created: restoreOwned}
	scenarioStage(t, env, "backup-restore", func(t *testing.T) {
		if restoreID != "" {
			t.Fatal("journaled Premium restore Product ID has no passed backup-restore receipt")
		}
		basePath := fmt.Sprintf("/api/v2025/cloud/v1beta2/tidbs/%s/backups", clusterID)
		query := "?orgId=1440002&projectId=14400020002"
		backupName := env.sessionName + "-prem-backup"
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+query,
			[]byte(fmt.Sprintf(`{"name":%q}`, backupName)),
			"application/json",
			http.StatusOK,
		)
		backup := waitForNamedObjectState(
			t,
			ctx,
			client,
			env,
			basePath+query+"&pageSize=10&skip=0",
			backupName,
			"SUCCEEDED",
			"FAILED",
		)
		backupID := scalarFromValue(backup, "backupId", "id")
		if backupID == "" {
			t.Fatalf("Premium backup %s has no ID", backupName)
		}
		restoreName := env.sessionName + "-prem-restore"
		body := []byte(fmt.Sprintf(
			`{"restoreMode":"SNAPSHOT","backupId":%q,"sourceTidbId":%q,"tidb":{"displayName":%q,"servicePlan":"Premium","cloudProvider":"aws","regionId":"aws-us-west-2","maxRcu":"20000","highAvailabilityType":"REGIONAL","labels":{"tidb.cloud/project":"14400020002"}}}`,
			backupID,
			clusterID,
			restoreName,
		))
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			"/api/v2025/cloud/v1beta2/tidbs:restore?orgId=1440002&projectId=14400020002",
			body,
			"application/json",
			http.StatusOK,
		)
		restoreID = jsonScalarAny(t, response, "tidbId", "id")
		recordPublicUIRunProduct(t, env, "premium-restore", restoreID)
		restoreDeletion.created = true
		waitForClusterActive(t, ctx, env, "1440002", restoreID)
		publicHost := configureNextGenPublic(
			t,
			ctx,
			client,
			env,
			config.DefaultUIOrganizationID,
			config.DefaultUIOrganizationID+"0002",
			restoreID,
		)
		waitForStrictMySQL(
			t,
			ctx,
			env,
			publicHost,
			true,
			"SELECT id,value FROM cp9_uiapi.proof; SHOW STATUS LIKE 'Ssl_cipher';",
			"public-ui-api-proof",
		)
	})
	if env.runJournal.stageStatus(t.Name()+"/backup-restore") == "passed" &&
		(restoreID == "" || !restoreDeletion.created) {
		t.Fatal("passed Premium backup-restore receipt lacks a run-owned restore Product ID")
	}
	runPremiumProductLifecycle(
		t,
		ctx,
		env,
		config.DefaultUIOrganizationID,
		config.DefaultUIOrganizationID+"0002",
		restoreID,
		&restoreDeletion,
		"restore-",
	)
}

func waitForNamedObjectState(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	path string,
	name string,
	success string,
	failure string,
) map[string]any {
	t.Helper()
	for {
		response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
		var value any
		if err := json.Unmarshal(response, &value); err != nil {
			t.Fatalf("decode %d-byte response for GET %s: %v", len(response), path, err)
		}
		if object := findNamedObject(value, name); object != nil {
			state := strings.ToUpper(scalarFromValue(object, "state", "status", "stage", "backupStatus"))
			if strings.Contains(state, success) {
				return object
			}
			if strings.Contains(state, failure) {
				t.Fatalf("GET %s found %s in %s", path, name, failure)
			}
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for GET %s to find %s in %s: %v", path, name, success, ctx.Err())
		case <-time.After(30 * time.Second):
		}
	}
}

func waitForNamedObjectAbsent(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	path string,
	name string,
) {
	t.Helper()
	for {
		response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
		var value any
		if err := json.Unmarshal(response, &value); err != nil {
			t.Fatalf("decode %d-byte response for GET %s: %v", len(response), path, err)
		}
		if findNamedObject(value, name) == nil {
			return
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for GET %s to remove %s: %v", path, name, ctx.Err())
		case <-time.After(30 * time.Second):
		}
	}
}

func getNamedObject(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	path string,
	name string,
) map[string]any {
	t.Helper()
	response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
	var value any
	if err := json.Unmarshal(response, &value); err != nil {
		t.Fatalf("decode %d-byte response for GET %s: %v", len(response), path, err)
	}
	return findNamedObject(value, name)
}

func waitForAPIObjectAbsent(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	path string,
) {
	t.Helper()
	for {
		response, status := doAPIRequestStatuses(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodGet,
			path,
			nil,
			"",
			http.StatusOK,
			http.StatusNotFound,
		)
		if status == http.StatusNotFound {
			return
		}
		var value any
		if err := json.Unmarshal(response, &value); err != nil {
			t.Fatalf("decode %d-byte response for GET %s: %v", len(response), path, err)
		}
		state := strings.ToUpper(scalarFromValue(value, "state", "status", "stage"))
		if state == "FAILED" || state == "ERROR" {
			t.Fatalf("GET %s entered terminal state %s\nresponse body:\n%s", path, state, response)
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for GET %s to become HTTP 404: %v; last state=%q\nresponse body:\n%s", path, ctx.Err(), state, response)
		case <-time.After(5 * time.Second):
		}
	}
}

func findNamedObject(value any, name string) map[string]any {
	switch value := value.(type) {
	case []any:
		for _, item := range value {
			if found := findNamedObject(item, name); found != nil {
				return found
			}
		}
	case map[string]any:
		for _, item := range value {
			if item, ok := item.(string); ok && item == name {
				return value
			}
		}
		for _, item := range value {
			if found := findNamedObject(item, name); found != nil {
				return found
			}
		}
	}
	return nil
}

func scalarFromValue(value any, keys ...string) string {
	for _, key := range keys {
		if found := findJSONScalar(value, key); found != "" {
			return found
		}
	}
	return ""
}

func jsonScalarAny(t *testing.T, response []byte, keys ...string) string {
	t.Helper()
	var value any
	if err := json.Unmarshal(response, &value); err != nil {
		t.Fatalf("decode %d-byte JSON response: %v", len(response), err)
	}
	if found := scalarFromValue(value, keys...); found != "" {
		return found
	}
	t.Fatalf("%d-byte JSON response has none of %v", len(response), keys)
	return ""
}

func openPublicUIRunJournal(
	t *testing.T,
	repoRoot string,
	session string,
	scenarios []string,
	dedicatedCloud string,
	resume bool,
	adopt bool,
) *publicUIRunJournal {
	t.Helper()
	selected := slices.Clone(scenarios)
	slices.Sort(selected)
	path := filepath.Join(repoRoot, ".cp9", session, "spec-tests", "public-ui-api-state.json")
	data, err := os.ReadFile(path)
	if adopt && !resume {
		t.Fatalf("%s requires %s=1", envAdoptSession, envResume)
	}
	if !resume || adopt {
		if err == nil {
			if adopt {
				t.Fatalf("%s requires a missing public UI run journal; unset it to resume %s", envAdoptSession, path)
			}
			t.Fatalf("public UI run journal already exists for fresh session %s: %s", session, path)
		}
		if !os.IsNotExist(err) {
			t.Fatalf("read public UI run journal %s: %v", path, err)
		}
		journal := &publicUIRunJournal{
			CurrentHead:    currentCommit(t, repoRoot),
			DedicatedCloud: dedicatedCloud,
			Path:           path,
			Scenarios:      selected,
			Session:        session,
			Stages:         map[string]publicUIStageReceipt{},
			Values:         map[string]string{},
			Version:        publicUIRunJournalVersion,
		}
		journal.mu.Lock()
		err = journal.persistLocked()
		journal.mu.Unlock()
		if err != nil {
			t.Fatalf("create public UI run journal %s: %v", path, err)
		}
		return journal
	}
	if err != nil {
		t.Fatalf("resume requires public UI run journal %s: %v", path, err)
	}
	info, err := os.Stat(path)
	if err != nil {
		t.Fatalf("stat public UI run journal %s: %v", path, err)
	}
	if info.Mode().Perm() != 0o600 {
		t.Fatalf("public UI run journal %s mode=%#o, want 0600", path, info.Mode().Perm())
	}
	var journal publicUIRunJournal
	if err := json.Unmarshal(data, &journal); err != nil {
		t.Fatalf("decode public UI run journal %s: %v", path, err)
	}
	if journal.Version != publicUIRunJournalVersion || journal.Session != session ||
		journal.DedicatedCloud != dedicatedCloud ||
		!slices.Equal(journal.Scenarios, selected) {
		t.Fatalf(
			"public UI run journal identity mismatch: version=%d session=%q dedicated=%q scenarios=%v, want version=%d session=%q dedicated=%q scenarios=%v",
			journal.Version,
			journal.Session,
			journal.DedicatedCloud,
			journal.Scenarios,
			publicUIRunJournalVersion,
			session,
			dedicatedCloud,
			selected,
		)
	}
	if journal.Stages == nil || journal.Values == nil {
		t.Fatalf("public UI run journal %s is missing stages or values", path)
	}
	for stage, receipt := range journal.Stages {
		if stage == "" || receipt.Head == "" || receipt.StartedAt == "" ||
			(receipt.Status != "running" && receipt.Status != "passed" && receipt.Status != "failed") ||
			(receipt.Status == "running" && receipt.FinishedAt != "") ||
			(receipt.Status != "running" && receipt.FinishedAt == "") {
			t.Fatalf("public UI run journal %s has invalid stage %q: %+v", path, stage, receipt)
		}
		if _, err := time.Parse(time.RFC3339Nano, receipt.StartedAt); err != nil {
			t.Fatalf("public UI run journal %s stage %q has invalid start time: %v", path, stage, err)
		}
		if receipt.FinishedAt != "" {
			if _, err := time.Parse(time.RFC3339Nano, receipt.FinishedAt); err != nil {
				t.Fatalf("public UI run journal %s stage %q has invalid finish time: %v", path, stage, err)
			}
		}
	}
	for key, value := range journal.Values {
		if !validPublicUIRunValue(key, value) {
			t.Fatalf("public UI run journal %s has invalid value %q", path, key)
		}
	}
	journal.CurrentHead = currentCommit(t, repoRoot)
	journal.Path = path
	return &journal
}

func (j *publicUIRunJournal) stageStatus(stage string) string {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.Stages[stage].Status
}

func (j *publicUIRunJournal) begin(stage string) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	if receipt, ok := j.Stages[stage]; ok {
		return fmt.Errorf("stage %s already has status %s at head %s", stage, receipt.Status, receipt.Head)
	}
	j.Stages[stage] = publicUIStageReceipt{
		Head:      j.CurrentHead,
		StartedAt: time.Now().UTC().Format(time.RFC3339Nano),
		Status:    "running",
	}
	return j.persistLocked()
}

func (j *publicUIRunJournal) finish(stage string, passed bool) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	receipt, ok := j.Stages[stage]
	if !ok || receipt.Status != "running" {
		return fmt.Errorf("stage %s has status %q, want running", stage, receipt.Status)
	}
	if passed {
		receipt.Status = "passed"
	} else {
		receipt.Status = "failed"
	}
	receipt.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano)
	j.Stages[stage] = receipt
	return j.persistLocked()
}

func (j *publicUIRunJournal) value(key string) string {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.Values[key]
}

func (j *publicUIRunJournal) putValue(key string, value string) error {
	j.mu.Lock()
	defer j.mu.Unlock()
	value = strings.TrimSpace(value)
	if !validPublicUIRunValue(key, value) {
		return fmt.Errorf("public UI run journal value %s is invalid", key)
	}
	if current := j.Values[key]; current != "" {
		if current != value {
			return fmt.Errorf("public UI run journal value %s changed from %q to %q", key, current, value)
		}
		return nil
	}
	j.Values[key] = value
	return j.persistLocked()
}

func validPublicUIRunValue(key string, value string) bool {
	if strings.ContainsAny(value, "\r\n") || value == "" {
		return false
	}
	if strings.HasPrefix(key, "product.") {
		for _, suffix := range []string{".owned_id", ".borrowed_id", ".public_host", ".user_prefix"} {
			if strings.HasSuffix(key, suffix) && len(key) > len("product.")+len(suffix) {
				return true
			}
		}
	}
	if strings.HasPrefix(key, "resource.") && strings.HasSuffix(key, ".id") {
		parts := strings.Split(key, ".")
		if len(parts) == 4 && parts[1] != "" && parts[2] != "" {
			return true
		}
	}
	return false
}

func (j *publicUIRunJournal) persistLocked() error {
	directory := filepath.Dir(j.Path)
	if err := os.MkdirAll(directory, 0o700); err != nil {
		return fmt.Errorf("create public UI run journal directory: %w", err)
	}
	data, err := json.MarshalIndent(j, "", "  ")
	if err != nil {
		return fmt.Errorf("encode public UI run journal: %w", err)
	}
	file, err := os.CreateTemp(directory, ".public-ui-api-state-")
	if err != nil {
		return fmt.Errorf("create public UI run journal temp file: %w", err)
	}
	tempPath := file.Name()
	removeTemp := true
	defer func() {
		if removeTemp {
			_ = os.Remove(tempPath)
		}
	}()
	if err := file.Chmod(0o600); err != nil {
		_ = file.Close()
		return fmt.Errorf("chmod public UI run journal temp file: %w", err)
	}
	data = append(data, '\n')
	if _, err := file.Write(data); err != nil {
		_ = file.Close()
		return fmt.Errorf("write public UI run journal temp file: %w", err)
	}
	if err := file.Sync(); err != nil {
		_ = file.Close()
		return fmt.Errorf("sync public UI run journal temp file: %w", err)
	}
	if err := file.Close(); err != nil {
		return fmt.Errorf("close public UI run journal temp file: %w", err)
	}
	if err := os.Rename(tempPath, j.Path); err != nil {
		return fmt.Errorf("publish public UI run journal: %w", err)
	}
	removeTemp = false
	return nil
}

func publicUIRunValueKey(product string, field string) string {
	return "product." + product + "." + field
}

func loadPublicUIRunProduct(
	t *testing.T,
	env *testEnvironment,
	product string,
	override string,
) (string, bool) {
	t.Helper()
	owned := env.runJournal.value(publicUIRunValueKey(product, "owned_id"))
	borrowed := env.runJournal.value(publicUIRunValueKey(product, "borrowed_id"))
	if owned != "" && borrowed != "" {
		t.Fatalf("public UI run journal product %s is both owned and borrowed", product)
	}
	override = strings.TrimSpace(override)
	recorded := owned
	if recorded == "" {
		recorded = borrowed
	}
	if recorded != "" {
		if override != "" && override != recorded {
			t.Fatalf("public UI run journal product %s ID=%s, override=%s", product, recorded, override)
		}
		return recorded, owned != ""
	}
	if override == "" {
		return "", false
	}
	if err := env.runJournal.putValue(publicUIRunValueKey(product, "borrowed_id"), override); err != nil {
		t.Fatalf("record borrowed public UI Product %s: %v", product, err)
	}
	return override, false
}

func recordPublicUIRunProduct(t *testing.T, env *testEnvironment, product string, id string) {
	t.Helper()
	if env.runJournal.value(publicUIRunValueKey(product, "borrowed_id")) != "" {
		t.Fatalf("cannot claim borrowed public UI Product %s as run-owned", product)
	}
	if err := env.runJournal.putValue(publicUIRunValueKey(product, "owned_id"), id); err != nil {
		t.Fatalf("record run-owned public UI Product %s: %v", product, err)
	}
}

func loadPublicUIRunHost(env *testEnvironment, product string) string {
	return env.runJournal.value(publicUIRunValueKey(product, "public_host"))
}

func recordPublicUIRunHost(t *testing.T, env *testEnvironment, product string, host string) {
	t.Helper()
	if err := env.runJournal.putValue(publicUIRunValueKey(product, "public_host"), host); err != nil {
		t.Fatalf("record public UI Product %s host: %v", product, err)
	}
}

func publicUIRunResourceID(env *testEnvironment, product string, resource string) string {
	return env.runJournal.value("resource." + product + "." + resource + ".id")
}

func recordPublicUIRunResourceID(
	t *testing.T,
	env *testEnvironment,
	product string,
	resource string,
	id string,
) {
	t.Helper()
	if err := env.runJournal.putValue("resource."+product+"."+resource+".id", id); err != nil {
		t.Fatalf("record public UI %s %s ID: %v", product, resource, err)
	}
}

func reconcilePublicUIRunResourceID(
	t *testing.T,
	env *testEnvironment,
	product string,
	resource string,
	object map[string]any,
	keys ...string,
) string {
	t.Helper()
	if object == nil {
		t.Fatalf("run-owned %s %s is absent", product, resource)
	}
	observed := scalarFromValue(object, keys...)
	if observed == "" {
		t.Fatalf("run-owned %s %s has no ID", product, resource)
	}
	recorded := publicUIRunResourceID(env, product, resource)
	if recorded == "" {
		recordPublicUIRunResourceID(t, env, product, resource, observed)
		return observed
	}
	if recorded != observed {
		t.Fatalf("run-owned %s %s ID=%q, want %q", product, resource, observed, recorded)
	}
	return recorded
}

func skippedScenarioStage(t *testing.T, env *testEnvironment, name string, reason string) {
	t.Helper()
	startedAt := time.Now()
	ran := false
	t.Run(name, func(t *testing.T) {
		ran = true
		t.Skip(reason)
	})
	if ran {
		env.summary.record(t.Name()+"/"+name, "skipped", time.Since(startedAt), reason)
	}
}

func scenarioStage(t *testing.T, env *testEnvironment, name string, run func(*testing.T)) {
	t.Helper()
	startedAt := time.Now()
	status := "passed"
	ran, reused := false, false
	passed := t.Run(name, func(t *testing.T) {
		ran = true
		stage := t.Name()
		switch existing := env.runJournal.stageStatus(stage); existing {
		case "passed":
			reused = true
			t.Logf("public UI run journal reuses passed stage %s", stage)
			return
		case "":
			if err := env.runJournal.begin(stage); err != nil {
				status = "failed"
				t.Fatalf("begin public UI run journal stage %s: %v", stage, err)
			}
		default:
			status = "failed"
			t.Fatalf(
				"public UI run journal stage %s is %s and cannot be rerun; select the next safe Go subtest after same-scene diagnosis",
				stage,
				existing,
			)
		}
		defer func() {
			if recovered := recover(); recovered != nil {
				status = "failed"
				if err := env.runJournal.finish(stage, false); err != nil {
					t.Errorf("fail panicked public UI run journal stage %s: %v", stage, err)
				}
				panic(recovered)
			} else if t.Skipped() {
				status = "skipped"
				if err := env.runJournal.finish(stage, false); err != nil {
					status = "failed"
					t.Errorf("fail skipped public UI run journal stage %s: %v", stage, err)
				}
			} else if t.Failed() {
				status = "failed"
				if err := env.runJournal.finish(stage, false); err != nil {
					t.Errorf("fail public UI run journal stage %s: %v", stage, err)
				}
			} else if err := env.runJournal.finish(stage, true); err != nil {
				status = "failed"
				t.Errorf("pass public UI run journal stage %s: %v", stage, err)
			}
		}()
		run(t)
	})
	if !ran {
		return
	}
	if reused {
		status = "reused"
	}
	env.summary.record(t.Name()+"/"+name, status, time.Since(startedAt), "")
	if !passed {
		t.FailNow()
	}
}

func runUIMetricsAcceptance(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	productLine string,
	bizType string,
	orgID string,
	projectID string,
	clusterID string,
	nextGen bool,
) {
	t.Helper()
	if env.uiClient == nil {
		t.Fatal("UI cookie client is not initialized")
	}
	regionID := "aws-us-west-2"
	if productLine == "dedicated" {
		regionID = env.dedicated.RegionID
	}
	metadataPath := fmt.Sprintf(
		"/external/%s/orgs/%s/projects/%s/clusters/%s/?canary=true",
		url.PathEscape(bizType),
		url.PathEscape(orgID),
		url.PathEscape(projectID),
		url.PathEscape(clusterID),
	)
	if nextGen {
		metadataPath = fmt.Sprintf(
			"/external/%s/orgs/%s/clusters/%s/?canary=true",
			url.PathEscape(bizType),
			url.PathEscape(orgID),
			url.PathEscape(clusterID),
		)
	}
	metadataHeaders := map[string]string{
		"X-Current-Project-Id": projectID,
		"X-Project-Id":         projectID,
		"X-Region-Id":          regionID,
	}
	metricHeaders := map[string]string{
		"X-Application-Id":     clusterID,
		"X-Biz-Type":           bizType,
		"X-Current-Project-Id": projectID,
		"X-Org-Id":             orgID,
		"X-Project-Id":         projectID,
		"X-Region-Id":          regionID,
	}
	metadata := doUICookieGET(t, ctx, env, env.baseURL+metadataPath, metadataHeaders)
	var response struct {
		Data struct {
			Metrics struct {
				OpenReadURI string `json:"openReadUri"`
			} `json:"metrics"`
		} `json:"data"`
	}
	if err := json.Unmarshal(metadata, &response); err != nil {
		t.Fatalf("decode %s UI Metrics metadata: %v", productLine, err)
	}
	readURI, err := url.Parse(strings.TrimSpace(response.Data.Metrics.OpenReadURI))
	if err != nil {
		t.Fatalf("parse %s UI Metrics openReadUri: %v", productLine, err)
	}
	base, err := url.Parse(env.baseURL)
	if err != nil {
		t.Fatalf("parse UI origin %q: %v", env.baseURL, err)
	}
	if readURI.Scheme != "https" || readURI.Host != base.Host || readURI.User != nil ||
		readURI.RawQuery != "" || readURI.Fragment != "" ||
		!strings.HasSuffix(readURI.Path, "/metrics/api/v1/") {
		t.Fatalf("%s UI Metrics openReadUri is not a same-origin metrics API root: %q", productLine, readURI)
	}
	// QPS and Connection Count are real chart sources in every pinned UI
	// Product config. Deliberately omit each UI fallback so empty storage fails.
	metrics := []struct {
		id    string
		query string
	}{
		{id: "query-per-second", query: "sum(rate(tidb_executor_statement_total[1m]))"},
		{id: "connection-count", query: "sum(tidb_server_connections)"},
	}
	metricCtx, cancel := context.WithTimeout(ctx, uiMetricsWaitTimeout)
	defer cancel()
	for _, metric := range metrics {
		lastStatus, lastSeries := "", 0
		var lastBody []byte
		for {
			now := time.Now().UTC()
			queryURL, err := url.Parse(strings.TrimSuffix(readURI.String(), "/") + "/query_range")
			if err != nil {
				t.Fatalf("build %s UI metric %s URL: %v", productLine, metric.id, err)
			}
			values := queryURL.Query()
			values.Set("query", metric.query)
			values.Set("start", fmt.Sprint(now.Add(-time.Hour).Unix()))
			values.Set("end", fmt.Sprint(now.Unix()))
			values.Set("step", "60")
			queryURL.RawQuery = values.Encode()
			body := doUICookieGET(t, metricCtx, env, queryURL.String(), metricHeaders)
			var result struct {
				Status string `json:"status"`
				Data   struct {
					Result []struct {
						Values [][]json.RawMessage `json:"values"`
					} `json:"result"`
				} `json:"data"`
			}
			if err := json.Unmarshal(body, &result); err != nil {
				t.Fatalf("decode %s UI metric %s response: %v", productLine, metric.id, err)
			}
			hasData := false
			for _, series := range result.Data.Result {
				hasData = hasData || len(series.Values) > 0
			}
			lastStatus, lastSeries = result.Status, len(result.Data.Result)
			lastBody = body
			if len(lastBody) > 8<<10 {
				lastBody = lastBody[len(lastBody)-(8<<10):]
			}
			if result.Status == "success" && hasData {
				t.Logf("%s UI metric %s query_range PASS: %s", productLine, metric.id, metric.query)
				break
			}
			select {
			case <-metricCtx.Done():
				t.Fatalf(
					"%s UI metric %s had no datapoints within %s: last status=%q series=%d\nresponse body tail:\n%s",
					productLine,
					metric.id,
					uiMetricsWaitTimeout,
					lastStatus,
					lastSeries,
					lastBody,
				)
			case <-time.After(15 * time.Second):
			}
		}
	}
	t.Logf("%s UI Metrics metadata PASS: %s", productLine, metadataPath)
}

func doUICookieGET(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	endpoint string,
	headers map[string]string,
) []byte {
	t.Helper()
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, http.NoBody)
	if err != nil {
		t.Fatalf("build UI Metrics GET %s: %v", endpoint, err)
	}
	request.Header.Set("Accept", "application/json")
	for key, value := range headers {
		request.Header.Set(key, value)
	}
	response, err := env.uiClient.Do(request)
	if err != nil {
		t.Fatalf("UI Metrics GET %s: %v", endpoint, err)
	}
	body, readErr := io.ReadAll(response.Body)
	_ = response.Body.Close()
	if readErr != nil {
		t.Fatalf("read UI Metrics GET %s: %v", endpoint, readErr)
	}
	if response.StatusCode != http.StatusOK {
		bodyTail := body
		if len(bodyTail) > 8<<10 {
			bodyTail = bodyTail[len(bodyTail)-(8<<10):]
		}
		t.Fatalf(
			"UI Metrics GET %s returned %s; x-debug-trace-id=%q; response body tail:\n%s",
			endpoint,
			response.Status,
			response.Header.Get("X-Debug-Trace-Id"),
			bodyTail,
		)
	}
	if !strings.HasPrefix(strings.ToLower(response.Header.Get("Content-Type")), "application/json") {
		t.Fatalf("UI Metrics GET %s returned non-JSON Content-Type %q", endpoint, response.Header.Get("Content-Type"))
	}
	return body
}

func runStrictMySQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	host string,
	sql string,
	marker string,
	premium bool,
) {
	t.Helper()
	caPath := strictMySQLCA(t, ctx, env, premium)
	result := runMySQL(t, ctx, host, caPath, sql)
	t.Logf("strict public MySQL result for %s:\n%s", host, result)
	if !bytes.Contains(result, []byte(marker)) || !bytes.Contains(result, []byte("TLS_")) {
		t.Fatalf("strict public MySQL result for %s did not contain marker and TLS cipher:\n%s", host, result)
	}
}

func strictMySQLCA(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	premium bool,
) string {
	t.Helper()
	if premium {
		if env.premiumID == "" {
			t.Fatal("Premium cluster ID is empty before CA download")
		}
		return strictNextGenMySQLCA(
			t,
			ctx,
			env,
			config.DefaultUIOrganizationID,
			config.DefaultUIOrganizationID+"0002",
			env.premiumID,
		)
	}
	caPath := filepath.Join(t.TempDir(), "ca.pem")
	outputData, err := os.ReadFile(
		filepath.Join(env.repoRoot, ".cp9", env.sessionName, "terraform", "outputs.json"),
	)
	if err != nil {
		t.Fatalf("read Terraform outputs for strict MySQL TLS: %v", err)
	}
	var outputs map[string]struct {
		Value json.RawMessage `json:"value"`
	}
	if err := json.Unmarshal(outputData, &outputs); err != nil {
		t.Fatalf("decode Terraform outputs for strict MySQL TLS: %v", err)
	}
	var pca struct {
		ARN string `json:"arn"`
	}
	if err := json.Unmarshal(outputs["session_product_pca"].Value, &pca); err != nil {
		t.Fatalf("decode Terraform output session_product_pca: %v", err)
	}
	pcaARN := pca.ARN
	if pcaARN == "" {
		t.Fatal("Terraform output session_product_pca.arn is empty")
	}
	caCommand := exec.CommandContext(
		ctx,
		"aws",
		"acm-pca",
		"get-certificate-authority-certificate",
		"--profile",
		env.sessionName,
		"--region",
		targetRegion,
		"--certificate-authority-arn",
		pcaARN,
		"--query",
		"Certificate",
		"--output",
		"text",
	)
	ca, err := caCommand.CombinedOutput()
	if err != nil {
		t.Fatalf("get Dedicated component CA certificate: %v\n%s", err, ca)
	}
	if err := os.WriteFile(caPath, ca, 0o600); err != nil {
		t.Fatalf("write Dedicated component CA certificate: %v", err)
	}
	return caPath
}

func strictNextGenMySQLCA(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) string {
	t.Helper()
	path := fmt.Sprintf(
		"/api/v2025/cloud/v1beta2/tidbs/%s/caCertificateUrl?orgId=%s&projectId=%s",
		clusterID,
		orgID,
		projectID,
	)
	response := doAPIRequest(
		t,
		ctx,
		&http.Client{Timeout: 2 * time.Minute},
		env.baseURL,
		http.MethodGet,
		path,
		nil,
		"",
		http.StatusOK,
	)
	uri := jsonScalar(t, response, "uri")
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, uri, http.NoBody)
	if err != nil {
		t.Fatalf("build nextgen CA download request: %T", err)
	}
	download, err := http.DefaultClient.Do(request)
	if err != nil {
		t.Fatalf("download nextgen CA certificate: %T", err)
	}
	defer func() {
		_ = download.Body.Close()
	}()
	ca, err := io.ReadAll(download.Body)
	if err != nil {
		t.Fatalf("read nextgen CA certificate: %v", err)
	}
	if download.StatusCode != http.StatusOK {
		t.Fatalf("download nextgen CA certificate returned %s with %d-byte body", download.Status, len(ca))
	}
	caPath := filepath.Join(t.TempDir(), "ca.pem")
	if err := os.WriteFile(caPath, ca, 0o600); err != nil {
		t.Fatalf("write nextgen CA certificate: %v", err)
	}
	return caPath
}

func runMySQL(
	t *testing.T,
	ctx context.Context,
	host string,
	caPath string,
	sql string,
) []byte {
	t.Helper()
	if host == "" {
		t.Fatal("strict public MySQL requires a journaled public host")
	}
	mysql := exec.CommandContext(
		ctx,
		"mysql",
		"--host",
		host,
		"--port",
		"4000",
		"--user",
		"root",
		"--ssl-mode=VERIFY_IDENTITY",
		"--ssl-ca="+caPath,
		"--execute="+sql,
	)
	mysql.Env = append(os.Environ(), "MYSQL_PWD="+os.Getenv(envDatabasePassword))
	result, err := mysql.CombinedOutput()
	if err != nil {
		t.Fatalf("strict public MySQL failed for %s: %v\n%s", host, err, result)
	}
	return result
}

func waitForClusterActive(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	orgID string,
	clusterID string,
) {
	t.Helper()
	client := &http.Client{Timeout: 2 * time.Minute}
	path := "/api/v2023/clusterManagement/v1beta2/tidbs/" + clusterID + "?orgId=" + orgID
	for {
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodGet,
			path,
			nil,
			"",
			http.StatusOK,
		)
		var status clusterInfo
		if err := json.Unmarshal(response, &status); err != nil {
			t.Fatalf("decode %d-byte cluster status response for GET %s: %v", len(response), path, err)
		}
		products := [...]struct {
			name, state, createTime string
			timeout                 time.Duration
		}{
			{"Dedicated", status.Dedicated.State, status.Dedicated.CreateTime, dedicatedPremiumCreateToActiveSLA},
			{"Premium", status.Premium.State, status.Premium.CreateTime, dedicatedPremiumCreateToActiveSLA},
			{"Starter", status.Starter.State, status.Starter.CreateTime, starterFamilyCreateToActiveSLA},
			{"Essential v1", status.EssentialV1.State, status.EssentialV1.CreateTime, starterFamilyCreateToActiveSLA},
			{"Essential v2", status.EssentialV2.State, status.EssentialV2.CreateTime, starterFamilyCreateToActiveSLA},
		}
		product := products[0]
		for _, candidate := range products {
			if candidate.state != "" {
				product = candidate
				break
			}
		}
		if product.state == "" {
			t.Fatalf("cluster GET %s response has no recognized Product state", path)
		}
		if product.state == "ACTIVE" {
			return
		}
		switch product.state {
		case "FAILED", "ERROR", "UNAVAILABLE", "DELETED":
			t.Fatalf("cluster GET %s entered terminal state %s", path, product.state)
		}
		createdAt, err := time.Parse(time.RFC3339Nano, product.createTime)
		if err != nil {
			t.Fatalf("parse %s cluster GET %s createTime %q: %v", product.name, path, product.createTime, err)
		}
		deadline := createdAt.Add(product.timeout)
		remaining := time.Until(deadline)
		if remaining <= 0 {
			t.Fatalf(
				"%s cluster GET %s did not become ACTIVE by create-to-ready deadline %s; current state %s; last response was %d bytes",
				product.name,
				path,
				deadline.Format(time.RFC3339),
				product.state,
				len(response),
			)
		}
		select {
		case <-ctx.Done():
			t.Fatalf("wait for cluster GET %s to become ACTIVE: %v; last response was %d bytes", path, ctx.Err(), len(response))
		case <-time.After(min(time.Minute, remaining)):
		}
	}
}

type nextGenEndpoint struct {
	Host string `json:"host"`
	Type string `json:"connectionType"`
}

type clusterInfo struct {
	Dedicated struct {
		CreateTime      string `json:"createTime"`
		State           string `json:"state"`
		TiDBNodeSetting struct {
			Groups []struct {
				Endpoints []struct {
					Host string `json:"host"`
					Type string `json:"connectionType"`
				} `json:"endpoints"`
				ID string `json:"tidbNodeGroupId"`
			} `json:"tidbNodeGroups"`
		} `json:"tidbNodeSetting"`
	} `json:"dedicatedCluster"`
	Premium struct {
		CreateTime string            `json:"createTime"`
		Endpoints  []nextGenEndpoint `json:"endpoints"`
		State      string            `json:"state"`
	} `json:"premiumInstance"`
	Starter struct {
		CreateTime string `json:"createTime"`
		State      string `json:"state"`
	} `json:"starterInstance"`
	EssentialV1 struct {
		CreateTime string `json:"createTime"`
		State      string `json:"state"`
	} `json:"essentialInstance"`
	EssentialV2 struct {
		CreateTime string `json:"createTime"`
		State      string `json:"state"`
	} `json:"essentialV2Instance"`
}

func getClusterInfo(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	orgID string,
	clusterID string,
) clusterInfo {
	t.Helper()
	path := "/api/v2023/clusterManagement/v1beta2/tidbs/" + clusterID + "?orgId=" + orgID
	response := doAPIRequest(
		t,
		ctx,
		&http.Client{Timeout: 2 * time.Minute},
		env.baseURL,
		http.MethodGet,
		path,
		nil,
		"",
		http.StatusOK,
	)
	var info clusterInfo
	if err := json.Unmarshal(response, &info); err != nil {
		t.Fatalf("decode %d-byte cluster response for GET %s: %v", len(response), path, err)
	}
	return info
}

func clusterEndpoint(info clusterInfo, premium bool, connectionType string) (string, string) {
	if premium {
		for _, endpoint := range info.Premium.Endpoints {
			if endpoint.Type == connectionType {
				return "", endpoint.Host
			}
		}
		return "", ""
	}
	for _, group := range info.Dedicated.TiDBNodeSetting.Groups {
		for _, endpoint := range group.Endpoints {
			if endpoint.Type == connectionType {
				return group.ID, endpoint.Host
			}
		}
	}
	return "", ""
}

func testStarterAndEssentialV1(t *testing.T, env *testEnvironment) {
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	scenarioStage(t, env, "activate", func(t *testing.T) {
		activateIfNeeded(t, ctx, env, "starter")
	})

	productsPassed := t.Run("products", func(t *testing.T) {
		t.Run("starter", func(t *testing.T) {
			t.Parallel()
			testStarterFamilyProduct(
				t,
				env,
				"starter",
				envStarterID,
				[]byte(fmt.Sprintf(
					`{"region":{"cloudProvider":"aws","name":"regions/aws-us-west-2"},"displayName":%q,"encryptionConfig":{"enhancedEncryptionEnabled":false},"servicePlan":"Starter","spendingLimit":{"monthly":0}}`,
					env.sessionName+"-starter",
				)),
			)
		})
		t.Run("essential-v1", func(t *testing.T) {
			t.Parallel()
			testStarterFamilyProduct(
				t,
				env,
				"essential-v1",
				envEssentialV1ID,
				[]byte(fmt.Sprintf(
					`{"region":{"cloudProvider":"aws","name":"regions/aws-us-west-2"},"displayName":%q,"encryptionConfig":{"enhancedEncryptionEnabled":true},"highAvailabilityType":"ZONAL","autoScaling":{"minRcu":2000,"maxRcu":10000},"servicePlan":"Essential"}`,
					env.sessionName+"-e1",
				)),
			)
		})
	})
	if !productsPassed {
		t.FailNow()
	}
}

func testStarterFamilyProduct(
	t *testing.T,
	env *testEnvironment,
	product string,
	resumeEnv string,
	body []byte,
) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	client := &http.Client{Timeout: 2 * time.Minute}
	orgID := config.DefaultUIOrganizationID
	projectID := orgID + "0002"
	var clusterID string
	deletion := productDeletion{request: func(t *testing.T, deleteCtx context.Context) {
		doAPIRequest(
			t,
			deleteCtx,
			client,
			env.baseURL,
			http.MethodDelete,
			fmt.Sprintf(
				"/api/v2023/serverless/v1beta1/clusters/%s?orgId=%s&projectId=%s",
				clusterID,
				orgID,
				projectID,
			),
			nil,
			"",
			http.StatusOK,
		)
	}}
	clusterID = testUICluster(
		t,
		env,
		product,
		orgID,
		resumeEnv,
		fmt.Sprintf(
			"/api/v2023/serverless/v1beta1/clusters?orgId=%s&projectId=%s",
			orgID,
			projectID,
		),
		body,
		&deletion,
	)
	testServerlessCluster(t, env, product, orgID, projectID, clusterID)
	if !deletion.created {
		skippedScenarioStage(t, env, "ui-metrics", "borrowed Serverless cluster; UI Metrics is not claimed")
		skippedScenarioStage(t, env, "delete-cluster", "borrowed Product; delete is not claimed")
		return
	}
	scenarioStage(t, env, "ui-metrics", func(t *testing.T) {
		runUIMetricsAcceptance(
			t,
			ctx,
			env,
			product,
			config.DevTierO11YBizType,
			orgID,
			projectID,
			clusterID,
			false,
		)
	})
	scenarioStage(t, env, "delete-cluster", func(t *testing.T) {
		deletion.run(t, ctx)
	})
}

func testEssentialV2(t *testing.T, env *testEnvironment) {
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	orgID := config.EssentialV2UIOrganizationID
	projectID := orgID + "0002"
	scenarioStage(t, env, "switch-organization", func(t *testing.T) {
		client := &http.Client{Timeout: 2 * time.Minute}
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			"/api/v1/user/history/user-visit",
			[]byte(fmt.Sprintf(`{"org_id":%q}`, orgID)),
			"application/json",
			http.StatusOK,
		)
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			"/api/v1/user/history/user-visit",
			[]byte(fmt.Sprintf(`{"org_id":%q,"cluster_mode":"classic"}`, orgID)),
			"application/json",
			http.StatusOK,
		)
	})
	scenarioStage(t, env, "activate", func(t *testing.T) {
		activateIfNeeded(t, ctx, env, "essential-v2")
	})
	body := []byte(fmt.Sprintf(
		`{"displayName":%q,"regionId":"aws-us-west-2","maxRcu":"10000","servicePlan":"Essential_V2","highAvailabilityType":"ZONAL","labels":{"tidb.cloud/project":%q},"dualLayerDataEncryption":{"noEncryption":{}}}`,
		env.sessionName+"-e2",
		projectID,
	))
	var clusterID string
	deletion := productDeletion{request: func(t *testing.T, deleteCtx context.Context) {
		doAPIRequest(
			t,
			deleteCtx,
			&http.Client{Timeout: 2 * time.Minute},
			env.baseURL,
			http.MethodDelete,
			fmt.Sprintf(
				"/api/v2025/cloud/v1beta2/tidbs/%s?orgId=%s&projectId=%s",
				clusterID,
				orgID,
				projectID,
			),
			nil,
			"",
			http.StatusOK,
		)
	}}
	clusterID = testUICluster(
		t,
		env,
		"essential-v2",
		orgID,
		envEssentialV2ID,
		fmt.Sprintf(
			"/api/v2025/cloud/v1beta2/tidbs?orgId=%s&projectId=%s",
			orgID,
			projectID,
		),
		body,
		&deletion,
	)
	client := &http.Client{Timeout: 2 * time.Minute}
	publicHost := loadPublicUIRunHost(env, "essential-v2")
	scenarioStage(t, env, "configure-public-endpoint", func(t *testing.T) {
		publicHost = ensureNextGenPublicEndpoint(t, ctx, client, env, orgID, projectID, clusterID)
		recordPublicUIRunHost(t, env, "essential-v2", publicHost)
	})
	if env.runJournal.stageStatus(t.Name()+"/configure-public-endpoint") == "passed" && publicHost == "" {
		t.Fatal("passed Essential-v2 public-endpoint receipt lacks its host")
	}
	scenarioStage(t, env, "configure-public-allowlist", func(t *testing.T) {
		ensureNextGenPublicAllowlist(t, ctx, client, env, orgID, projectID, clusterID)
	})
	scenarioStage(t, env, "configure-password", func(t *testing.T) {
		setNextGenPassword(t, ctx, client, env, orgID, projectID, clusterID)
	})
	scenarioStage(t, env, "public-mysql-tls", func(t *testing.T) {
		marker := "essential-v2-public-" + clusterID
		waitForStrictMySQLWithCA(
			t,
			ctx,
			env,
			publicHost,
			strictNextGenMySQLCA(t, ctx, env, orgID, projectID, clusterID),
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.public_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				fmt.Sprintf("REPLACE INTO cp9_uiapi.public_sql VALUES (1, %q); ", marker)+
				"SELECT id,value FROM cp9_uiapi.public_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			marker,
		)
	})
	testNextGenPrivateLink(t, ctx, client, env, orgID, projectID, clusterID)
	testNextGenImport(t, ctx, client, env, orgID, projectID, clusterID, publicHost)
	testNextGenExport(t, ctx, client, env, orgID, projectID, clusterID)
	runPremiumProductLifecycle(t, ctx, env, orgID, projectID, clusterID, &deletion, "")
}

func activateIfNeeded(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	product string,
) {
	t.Helper()
	status := exec.CommandContext(
		ctx,
		env.cp9Binary,
		"{sys.output.format=json}",
		product+".status",
		"config="+env.configPath,
	)
	status.Dir = env.repoRoot
	var stderr bytes.Buffer
	status.Stderr = &stderr
	output, err := status.Output()
	t.Logf("cp9 %s status output:\n%s%s", product, output, stderr.Bytes())
	if err != nil {
		t.Fatalf("cp9 %s status: %v\n%s%s", product, err, output, stderr.Bytes())
	}
	var result struct {
		State string `json:"state"`
	}
	if err := json.Unmarshal(output, &result); err != nil {
		t.Fatalf("decode cp9 %s status: %v\n%s", product, err, output)
	}
	if result.State == "ready" {
		t.Logf("cp9 %s is already active; skipping activation", product)
		return
	}
	if err := runCP9Stage(
		ctx,
		t,
		env,
		product+"/activate",
		product,
		"activate",
		"config="+env.configPath,
	); err != nil {
		t.Fatalf("activate %s: %v", product, err)
	}
}

func testUICluster(
	t *testing.T,
	env *testEnvironment,
	product string,
	orgID string,
	resumeEnv string,
	createPath string,
	createBody []byte,
	deletion *productDeletion,
) string {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	client := &http.Client{Timeout: 2 * time.Minute}
	clusterID, owned := loadPublicUIRunProduct(t, env, product, os.Getenv(resumeEnv))
	deletion.created = owned
	if clusterID == "" || owned || env.runJournal.stageStatus(t.Name()+"/create-cluster") != "" {
		scenarioStage(t, env, "create-cluster", func(t *testing.T) {
			if clusterID != "" {
				t.Fatalf("journaled %s Product ID has no passed create receipt", product)
			}
			response := doAPIRequest(
				t,
				ctx,
				client,
				env.baseURL,
				http.MethodPost,
				createPath,
				createBody,
				"application/json",
				http.StatusOK,
			)
			clusterID = jsonScalarAny(t, response, "clusterId", "tidbId")
			recordPublicUIRunProduct(t, env, product, clusterID)
			deletion.created = true
		})
	}
	if env.runJournal.stageStatus(t.Name()+"/create-cluster") == "passed" &&
		(clusterID == "" || !deletion.created) {
		t.Fatalf("passed %s create receipt lacks a run-owned Product ID", product)
	}
	scenarioStage(t, env, "wait-active", func(t *testing.T) {
		waitForClusterActive(t, ctx, env, orgID, clusterID)
	})
	return clusterID
}

type serverlessCluster struct {
	Endpoints struct {
		Private struct {
			AWS struct {
				ServiceName string `json:"serviceName"`
			} `json:"aws"`
			Host string `json:"host"`
		} `json:"private"`
		Public struct {
			AuthorizedNetworks []struct {
				EndIPAddress   string `json:"endIpAddress"`
				StartIPAddress string `json:"startIpAddress"`
			} `json:"authorizedNetworks"`
			Disabled bool   `json:"disabled"`
			Host     string `json:"host"`
		} `json:"public"`
	} `json:"endpoints"`
	UserPrefix string `json:"userPrefix"`
}

func testServerlessCluster(
	t *testing.T,
	env *testEnvironment,
	product string,
	orgID string,
	projectID string,
	clusterID string,
) {
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Hour)
	defer cancel()
	client := &http.Client{Timeout: 2 * time.Minute}
	var cluster serverlessCluster
	cluster.Endpoints.Public.Host = loadPublicUIRunHost(env, product)
	cluster.UserPrefix = env.runJournal.value(publicUIRunValueKey(product, "user_prefix"))
	scenarioStage(t, env, "configure-public-endpoint", func(t *testing.T) {
		cluster = ensureServerlessPublicEndpoint(t, ctx, client, env, orgID, projectID, clusterID)
		recordPublicUIRunHost(t, env, product, cluster.Endpoints.Public.Host)
		if err := env.runJournal.putValue(publicUIRunValueKey(product, "user_prefix"), cluster.UserPrefix); err != nil {
			t.Fatalf("record public UI Product %s user prefix: %v", product, err)
		}
	})
	if env.runJournal.stageStatus(t.Name()+"/configure-public-endpoint") == "passed" &&
		(cluster.Endpoints.Public.Host == "" || cluster.UserPrefix == "") {
		t.Fatalf("passed %s public-endpoint receipt lacks its host or user prefix", product)
	}
	scenarioStage(t, env, "configure-public-allowlist", func(t *testing.T) {
		ensureServerlessPublicAllowlist(t, ctx, client, env, orgID, projectID, clusterID)
	})
	scenarioStage(t, env, "configure-password", func(t *testing.T) {
		setServerlessPassword(t, ctx, client, env, orgID, projectID, clusterID)
	})
	username := cluster.UserPrefix + ".root"
	marker := "serverless-public-" + clusterID
	scenarioStage(t, env, "public-mysql-tls", func(t *testing.T) {
		runServerlessFixtureMySQL(
			t,
			ctx,
			env,
			cluster.Endpoints.Public.Host,
			"",
			username,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.public_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				fmt.Sprintf("REPLACE INTO cp9_uiapi.public_sql VALUES (1, %q); ", marker)+
				"SELECT id,value FROM cp9_uiapi.public_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			marker,
		)
	})
	testServerlessPrivateLink(t, ctx, client, env, orgID, projectID, clusterID, username)
	testServerlessImport(t, ctx, client, env, orgID, projectID, clusterID, username, cluster.Endpoints.Public.Host)
	testServerlessExport(t, ctx, client, env, orgID, projectID, clusterID)
}

func serverlessClusterPath(orgID string, projectID string, clusterID string) string {
	return fmt.Sprintf(
		"/api/v2023/serverless/v1beta1/clusters/%s?orgId=%s&projectId=%s",
		clusterID,
		orgID,
		projectID,
	)
}

func ensureServerlessPublicEndpoint(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) serverlessCluster {
	t.Helper()
	endpointCtx, cancel := publicEndpointWaitContext(ctx)
	defer cancel()
	path := serverlessClusterPath(orgID, projectID, clusterID)
	cluster := getServerlessCluster(t, endpointCtx, client, env, orgID, projectID, clusterID)
	if cluster.Endpoints.Public.Disabled {
		doAPIRequest(
			t,
			endpointCtx,
			client,
			env.baseURL,
			http.MethodPatch,
			path,
			[]byte(`{"cluster":{"endpoints":{"public":{"disabled":false}}},"updateMask":"endpoints.public.disabled"}`),
			"application/json",
			http.StatusOK,
		)
	}
	for {
		cluster = getServerlessCluster(t, endpointCtx, client, env, orgID, projectID, clusterID)
		if !cluster.Endpoints.Public.Disabled && cluster.Endpoints.Public.Host != "" && cluster.UserPrefix != "" {
			return cluster
		}
		waitForPublicEndpointRetry(t, endpointCtx, path, 10*time.Second)
	}
}

func ensureServerlessPublicAllowlist(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	t.Helper()
	cluster := getServerlessCluster(t, ctx, client, env, orgID, projectID, clusterID)
	for _, network := range cluster.Endpoints.Public.AuthorizedNetworks {
		if network.StartIPAddress == env.fixture.publicIP && network.EndIPAddress == env.fixture.publicIP {
			return
		}
	}
	body := []byte(fmt.Sprintf(
		`{"cluster":{"endpoints":{"public":{"authorizedNetworks":[{"displayName":%q,"startIpAddress":%q,"endIpAddress":%q}]}}},"updateMask":"endpoints.public.authorizedNetworks"}`,
		"Current_client_IP_address",
		env.fixture.publicIP,
		env.fixture.publicIP,
	))
	doAPIRequest(
		t,
		ctx,
		client,
		env.baseURL,
		http.MethodPatch,
		serverlessClusterPath(orgID, projectID, clusterID),
		body,
		"application/json",
		http.StatusOK,
	)
}

func setServerlessPassword(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	t.Helper()
	path := fmt.Sprintf(
		"/api/v2023/serverless/v1beta1/clusters/%s/password?orgId=%s&projectId=%s",
		clusterID,
		orgID,
		projectID,
	)
	body := []byte(fmt.Sprintf(`{"password":%q}`, os.Getenv(envDatabasePassword)))
	doAPIRequest(t, ctx, client, env.baseURL, http.MethodPut, path, body, "application/json", http.StatusOK)
}

const publicEndpointWaitTimeout = 15 * time.Minute

func publicEndpointWaitContext(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(ctx, publicEndpointWaitTimeout)
}

func waitForPublicEndpointRetry(t *testing.T, ctx context.Context, path string, interval time.Duration) {
	t.Helper()
	select {
	case <-ctx.Done():
		t.Fatalf("wait for public endpoint GET %s within %s: %v", path, publicEndpointWaitTimeout, ctx.Err())
	case <-time.After(interval):
	}
}

func getServerlessCluster(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) serverlessCluster {
	t.Helper()
	path := serverlessClusterPath(orgID, projectID, clusterID)
	response := doAPIRequest(t, ctx, client, env.baseURL, http.MethodGet, path, nil, "", http.StatusOK)
	var cluster serverlessCluster
	if err := json.Unmarshal(response, &cluster); err != nil {
		t.Fatalf("decode %d-byte serverless cluster response for GET %s: %v", len(response), path, err)
	}
	return cluster
}

func runServerlessFixtureMySQL(
	t *testing.T,
	ctx context.Context,
	env *testEnvironment,
	host string,
	endpointIP string,
	username string,
	sql string,
	marker string,
) {
	t.Helper()
	runFixtureMySQL(
		t,
		ctx,
		env,
		host,
		endpointIP,
		username,
		strictMySQLCA(t, ctx, env, false),
		sql,
		marker,
	)
}

func testServerlessPrivateLink(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
	username string,
) {
	scenarioStage(t, env, "private-link", func(t *testing.T) {
		cluster := getServerlessCluster(t, ctx, client, env, orgID, projectID, clusterID)
		if cluster.Endpoints.Private.AWS.ServiceName == "" || cluster.Endpoints.Private.Host == "" {
			t.Fatalf("serverless cluster %s returned incomplete PrivateLink metadata: %+v", clusterID, cluster.Endpoints.Private)
		}
		endpointID := createVPCEndpoint(t, ctx, env, cluster.Endpoints.Private.AWS.ServiceName)
		endpointIP := waitForVPCEndpoint(t, ctx, env, endpointID)
		path := fmt.Sprintf(
			"/api/v2023/serverless/v1beta1/clusters/%s?orgId=%s&projectId=%s",
			clusterID,
			orgID,
			projectID,
		)
		body := []byte(fmt.Sprintf(
			`{"cluster":{"endpoints":{"private":{"authorizedNetworks":[{"displayName":%q,"vpcEndpointId":%q}]}}},"updateMask":"endpoints.private.authorizedNetworks"}`,
			env.sessionName+"-vpce",
			endpointID,
		))
		doAPIRequest(t, ctx, client, env.baseURL, http.MethodPatch, path, body, "application/json", http.StatusOK)
		marker := "serverless-private-" + clusterID
		runServerlessFixtureMySQL(
			t,
			ctx,
			env,
			cluster.Endpoints.Private.Host,
			endpointIP,
			username,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; "+
				"CREATE TABLE IF NOT EXISTS cp9_uiapi.private_sql "+
				"(id BIGINT PRIMARY KEY, value VARCHAR(64) NOT NULL); "+
				fmt.Sprintf("REPLACE INTO cp9_uiapi.private_sql VALUES (1, %q); ", marker)+
				"SELECT id,value FROM cp9_uiapi.private_sql; "+
				"SHOW STATUS LIKE 'Ssl_cipher';",
			marker,
		)
	})
}

func testServerlessImport(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
	username string,
	publicHost string,
) {
	scenarioStage(t, env, "s3-import", func(t *testing.T) {
		runServerlessFixtureMySQL(
			t,
			ctx,
			env,
			publicHost,
			"",
			username,
			"CREATE DATABASE IF NOT EXISTS cp9_uiapi; DROP TABLE IF EXISTS cp9_uiapi.proof; "+
				"CREATE TABLE cp9_uiapi.proof (id BIGINT PRIMARY KEY, note VARCHAR(128)); "+
				"SELECT 'import-table-ready'; SHOW STATUS LIKE 'Ssl_cipher';",
			"import-table-ready",
		)
		basePath := fmt.Sprintf(
			"/api/v1/serverless/orgs/%s/projects/%s/clusters/%s/imports",
			orgID,
			projectID,
			clusterID,
		)
		sourceURI := "s3://" + env.fixture.bucket + "/inputs/"
		validateBody := []byte(fmt.Sprintf(
			`{"validation_type":"SOURCE_ACCESS_CHECK","source":{"type":"S3","s3":{"uri":%q,"auth_type":"ROLE_ARN","role_arn":%q}}}`,
			sourceURI,
			env.fixture.importRoleARN,
		))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+":validate",
			validateBody,
			"application/json",
			http.StatusOK,
		)
		body := []byte(fmt.Sprintf(
			`{"import_options":{"file_type":"SQL"},"source":{"type":"S3","s3":{"uri":%q,"auth_type":"ROLE_ARN","role_arn":%q,"target_table_infos":[{"target_table":{"database":"cp9_uiapi","table":"proof"},"custom_file":"proof.sql"}]}}}`,
			sourceURI,
			env.fixture.importRoleARN,
		))
		doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath+":precheck",
			body,
			"application/json",
			http.StatusOK,
		)
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath,
			body,
			"application/json",
			http.StatusOK,
		)
		importID := jsonScalar(t, response, "id")
		waitForAPIState(t, ctx, client, env, basePath+"/"+importID, "COMPLETED", "FAILED")
		runServerlessFixtureMySQL(
			t,
			ctx,
			env,
			publicHost,
			"",
			username,
			"SELECT id,note FROM cp9_uiapi.proof; SHOW STATUS LIKE 'Ssl_cipher';",
			"public-ui-api-proof",
		)
	})
}

func testServerlessExport(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	env *testEnvironment,
	orgID string,
	projectID string,
	clusterID string,
) {
	scenarioStage(t, env, "s3-export", func(t *testing.T) {
		basePath := fmt.Sprintf(
			"/api/v1/serverless/orgs/%s/projects/%s/clusters/%s/exports",
			orgID,
			projectID,
			clusterID,
		)
		prefix := "outputs/" + clusterID + "/"
		body := []byte(fmt.Sprintf(
			`{"display_name":%q,"export_options":{"file_type":"SQL","filter":{"table":{"patterns":["*.*"]}},"compression":"ZSTD"},"target":{"type":"S3","s3":{"uri":%q,"auth_type":"ROLE_ARN","role_arn":%q}}}`,
			env.sessionName+"-export",
			"s3://"+env.fixture.bucket+"/"+prefix,
			env.fixture.importRoleARN,
		))
		response := doAPIRequest(
			t,
			ctx,
			client,
			env.baseURL,
			http.MethodPost,
			basePath,
			body,
			"application/json",
			http.StatusOK,
		)
		exportID := jsonScalar(t, response, "export_id")
		waitForAPIState(t, ctx, client, env, basePath+"/"+exportID, "SUCCEEDED", "FAILED")
		objects := strings.TrimSpace(string(runAWS(
			t,
			ctx,
			env,
			"s3api",
			"list-objects-v2",
			"--bucket",
			env.fixture.bucket,
			"--prefix",
			prefix,
			"--query",
			"length(Contents)",
			"--output",
			"text",
		)))
		if objects == "0" || objects == "None" || objects == "" {
			t.Fatalf("serverless export %s wrote no objects under s3://%s/%s", exportID, env.fixture.bucket, prefix)
		}
	})
}

func prepareDefaultUIAPIInputs(
	t *testing.T,
	summary *workflowSummary,
	repoRoot string,
) {
	t.Helper()
	startedAt := time.Now()
	status := "failed"
	defer func() {
		summary.record("inputs/default-804", status, time.Since(startedAt), "")
	}()

	setDefaultEnv(t, envSourceProfile, defaultProfile)
	setDefaultEnv(t, envAccountID, defaultAccount)
	accountContract := loadRuntimeAccountContract(t, repoRoot, strings.TrimSpace(os.Getenv(envAccountID)))
	setExactEnv(t, envCertificateARN, accountContract.CertificateARN)
	setExactEnv(t, envDNSZoneID, accountContract.DNSZoneID)
	setExactEnv(t, envDNSZoneName, accountContract.DNSZoneName)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	profile := strings.TrimSpace(os.Getenv(envSourceProfile))
	identity := runBootstrapAWS(t, ctx, profile, "sts", "get-caller-identity", "--output", "json")
	var caller struct {
		Account string `json:"Account"`
		ARN     string `json:"Arn"`
	}
	if err := json.Unmarshal(identity, &caller); err != nil {
		t.Fatalf("decode AWS caller identity: %v\n%s", err, identity)
	}
	if caller.Account != strings.TrimSpace(os.Getenv(envAccountID)) {
		t.Fatalf(
			"%s resolves account %s; expected %s",
			profile,
			caller.Account,
			os.Getenv(envAccountID),
		)
	}
	owner := strings.ToLower(caller.ARN[strings.LastIndex(caller.ARN, "/")+1:])
	sessionName := os.Getenv(envSessionName)

	if strings.TrimSpace(os.Getenv(envSubnetID)) == "" {
		subnets := runCP9RoleAWS(
			t,
			ctx,
			profile,
			caller.Account,
			owner,
			sessionName,
			"ec2",
			"describe-subnets",
			"--filters",
			"Name=state,Values=available",
			"--output",
			"json",
		)
		selector := exec.CommandContext(
			ctx,
			filepath.Join(repoRoot, "spec-tests", "scripts", "select-public-exposure-subnet.py"),
		)
		selector.Stdin = bytes.NewReader(subnets)
		selected, err := selector.CombinedOutput()
		if err != nil {
			t.Fatalf("select ALB-compatible public subnet: %v\n%s", err, selected)
		}
		subnet := strings.TrimSpace(string(selected))
		if subnet == "" || subnet == "None" {
			t.Fatalf("account %s has no ALB-compatible public subnet in %s", caller.Account, targetRegion)
		}
		t.Setenv(envSubnetID, subnet)
	}

	keyPair := strings.TrimSpace(os.Getenv(envKeyPair))
	privateKey := strings.TrimSpace(os.Getenv(envSSHPrivateKey))
	if (keyPair == "") != (privateKey == "") {
		t.Fatalf("%s and %s must be set together", envKeyPair, envSSHPrivateKey)
	}
	if keyPair == "" || privateKey == "" {
		t.Fatalf("%s and %s are required together; CP9 does not create or import EC2 keys", envKeyPair, envSSHPrivateKey)
	}
	keyInfo, err := os.Stat(privateKey)
	if err != nil || !filepath.IsAbs(privateKey) || !keyInfo.Mode().IsRegular() {
		t.Fatalf("%s must name an absolute readable regular private-key file: %v", envSSHPrivateKey, err)
	}
	status = "passed"
}

func setDefaultEnv(t *testing.T, name string, value string) {
	t.Helper()
	if strings.TrimSpace(os.Getenv(name)) == "" {
		t.Setenv(name, value)
	}
}

func setExactEnv(t *testing.T, name string, expected string) {
	t.Helper()
	value := strings.TrimSpace(os.Getenv(name))
	if value == "" {
		t.Setenv(name, expected)
		return
	}
	if value != expected {
		t.Fatalf("%s=%q does not match the runtime-account registry value %q", name, value, expected)
	}
}

func loadRuntimeAccountContract(t *testing.T, repoRoot string, accountID string) runtimeAccountContract {
	t.Helper()
	data, err := os.ReadFile(filepath.Join(repoRoot, ".github", "cp9-cleanup-accounts.json"))
	if err != nil {
		t.Fatalf("read runtime-account registry: %v", err)
	}
	var registry struct {
		Include []runtimeAccountContract `json:"include"`
	}
	if err := json.Unmarshal(data, &registry); err != nil {
		t.Fatalf("decode runtime-account registry: %v", err)
	}
	for _, contract := range registry.Include {
		if contract.AccountID != accountID {
			continue
		}
		if !strings.HasPrefix(contract.DNSZoneID, "Z") ||
			contract.DNSZoneName != accountID+".cp9.dev.aws.tidbcloud.com" ||
			!strings.Contains(contract.CertificateARN, ":"+accountID+":certificate/") {
			t.Fatalf("runtime account %s has an incomplete child DNS zone/certificate contract", accountID)
		}
		return contract
	}
	t.Fatalf("runtime account %s is absent from .github/cp9-cleanup-accounts.json", accountID)
	return runtimeAccountContract{}
}

func runBootstrapAWS(t *testing.T, ctx context.Context, profile string, args ...string) []byte {
	t.Helper()
	result, err := runBootstrapAWSCommand(ctx, profile, args...)
	if err != nil {
		t.Fatalf("aws %s: %v (%d-byte output)", awsOperation(args), err, len(result))
	}
	return result
}

func runBootstrapAWSCommand(
	ctx context.Context,
	profile string,
	args ...string,
) ([]byte, error) {
	if len(args) == 0 {
		return nil, fmt.Errorf("AWS command is empty")
	}
	args = append(args, "--profile", profile)
	command := exec.CommandContext(ctx, "aws", args...)
	return command.CombinedOutput()
}

func runCP9RoleAWS(
	t *testing.T,
	ctx context.Context,
	profile string,
	account string,
	owner string,
	session string,
	args ...string,
) []byte {
	t.Helper()
	result, err := runCP9RoleAWSCommand(ctx, profile, account, owner, session, args...)
	if err != nil {
		t.Fatalf("aws %s: %v (%d-byte output)", awsOperation(args), err, len(result))
	}
	return result
}

func runCP9RoleAWSCommand(
	ctx context.Context,
	profile string,
	account string,
	owner string,
	session string,
	args ...string,
) ([]byte, error) {
	assumed, err := runBootstrapAWSCommand(
		ctx,
		profile,
		"sts",
		"assume-role",
		"--role-arn",
		"arn:aws:iam::"+account+":role/cp9-developer",
		"--role-session-name",
		owner,
		"--duration-seconds",
		"3600",
		"--tags",
		"Key=cp9-owner,Value="+owner,
		"Key=cp9-session,Value="+session,
		"--output",
		"json",
	)
	if err != nil {
		return nil, fmt.Errorf("assume cp9-developer: %w (%d-byte output)", err, len(assumed))
	}
	var credentials struct {
		Credentials struct {
			AccessKey    string `json:"AccessKeyId"`
			SecretKey    string `json:"SecretAccessKey"`
			SessionToken string `json:"SessionToken"`
		} `json:"Credentials"`
	}
	if err := json.Unmarshal(assumed, &credentials); err != nil {
		return nil, fmt.Errorf("decode cp9-developer credentials: %w", err)
	}
	if credentials.Credentials.AccessKey == "" || credentials.Credentials.SecretKey == "" ||
		credentials.Credentials.SessionToken == "" {
		return nil, fmt.Errorf("cp9-developer credentials are incomplete")
	}
	args = append(args, "--region", targetRegion)
	command := exec.CommandContext(ctx, "aws", args...)
	command.Env = append(
		os.Environ(),
		"AWS_ACCESS_KEY_ID="+credentials.Credentials.AccessKey,
		"AWS_SECRET_ACCESS_KEY="+credentials.Credentials.SecretKey,
		"AWS_SESSION_TOKEN="+credentials.Credentials.SessionToken,
		"AWS_PAGER=",
	)
	return command.CombinedOutput()
}

func sessionFromEnv(
	t *testing.T,
	repoRoot string,
	tiltfile string,
	dedicated dedicatedTarget,
) *config.SessionConfig {
	t.Helper()

	required := map[string]string{
		envSourceProfile:  strings.TrimSpace(os.Getenv(envSourceProfile)),
		envAccountID:      strings.TrimSpace(os.Getenv(envAccountID)),
		envSubnetID:       strings.TrimSpace(os.Getenv(envSubnetID)),
		envKeyPair:        strings.TrimSpace(os.Getenv(envKeyPair)),
		envSSHPrivateKey:  strings.TrimSpace(os.Getenv(envSSHPrivateKey)),
		envCertificateARN: strings.TrimSpace(os.Getenv(envCertificateARN)),
		envDNSZoneID:      strings.TrimSpace(os.Getenv(envDNSZoneID)),
		envDNSZoneName:    strings.TrimSpace(os.Getenv(envDNSZoneName)),
	}
	for name, value := range required {
		if value == "" {
			t.Fatalf("%s is required", name)
		}
	}
	if !strings.Contains(required[envCertificateARN], ":"+required[envAccountID]+":") {
		t.Fatalf(
			"%s must belong to account %s",
			envCertificateARN,
			required[envAccountID],
		)
	}
	if required[envDNSZoneName] != required[envAccountID]+".cp9.dev.aws.tidbcloud.com" {
		t.Fatalf("%s must be the delegated child zone for account %s", envDNSZoneName, required[envAccountID])
	}

	sessionName := strings.TrimSpace(os.Getenv(envSessionName))
	if sessionName == "" {
		sessionName = "c9u" + time.Now().UTC().Format("01021504") + randomHex(t, 2)
	}

	session := &config.SessionConfig{
		APIVersion: config.APIVersionV1Alpha2,
		Kind:       config.SessionKind,
		Name:       sessionName,
		Persona:    "dev-full",
		Host: config.SessionHostConfig{
			Mode: config.HostModeRemoteEC2K3d,
			TTL:  config.Duration{Duration: 10 * time.Hour},
			EC2: config.HostEC2Config{
				InstanceType:      "m7i.xlarge",
				KeyPair:           required[envKeyPair],
				SSHUser:           "ec2-user",
				SSHPrivateKey:     required[envSSHPrivateKey],
				SubnetID:          required[envSubnetID],
				RootVolumeSizeGiB: 160,
			},
			Exposure: config.HostExposureConfig{
				Enabled:        true,
				ZoneID:         required[envDNSZoneID],
				ZoneName:       required[envDNSZoneName],
				CertificateARN: required[envCertificateARN],
			},
		},
		Target: config.SessionTargetConfig{
			Cloud:  "aws",
			Region: targetRegion,
			AWS: config.SessionAWSConfig{
				IAM: config.SessionIAMConfig{
					SourceProfile: required[envSourceProfile],
					RoleARN: "arn:aws:iam::" + required[envAccountID] +
						":role/cp9-developer",
				},
			},
		},
		Terraform: config.SessionTerraformConfig{
			Repo:    repoRoot,
			Ref:     currentCommit(t, repoRoot),
			Path:    "terraform/aws/cp9-substrate",
			VarFile: "terraform/aws/cp9-substrate/premium.tfvars",
		},
		Tilt: config.SessionTiltConfig{
			Mode: "up",
			File: tiltfile,
		},
		Services: config.SessionServicesConfig{
			SourceBuildParallelism: 2,
			GitHubAppClientID:      defaultGitHubAppClientID,
		},
	}
	if dedicated.Cloud == "gcp" {
		session.Dedicated = &config.SessionDedicatedConfig{
			GCP: &config.SessionDedicatedGCPConfig{
				Enabled:            true,
				SeedProjectID:      "pingcap-gardener",
				DataplaneProjectID: "pingcap-gardener",
				Region:             config.DedicatedGCPRegion,
				Zone:               config.DedicatedGCPRegion + "-a",
			},
		}
	}

	return session
}

func selectedScenarios(t *testing.T) []string {
	t.Helper()

	raw := strings.TrimSpace(os.Getenv(envScenarios))
	if raw == "" {
		raw = "dedicated,premium"
	}
	known := map[string]bool{
		"dedicated":    true,
		"premium":      true,
		"starter-e1":   true,
		"essential-v2": true,
	}
	seen := map[string]bool{}
	var selected []string
	for _, scenario := range strings.FieldsFunc(raw, func(r rune) bool {
		return r == ',' || r == ' ' || r == '\n'
	}) {
		if !known[scenario] {
			t.Fatalf("%s contains unknown scenario %q", envScenarios, scenario)
		}
		if !seen[scenario] {
			selected = append(selected, scenario)
			seen[scenario] = true
		}
	}
	if len(selected) == 0 {
		t.Fatalf("%s must enable at least one scenario", envScenarios)
	}
	return selected
}

func dedicatedTargetFromEnv(t *testing.T, scenarios []string) dedicatedTarget {
	t.Helper()
	cloud := strings.ToLower(strings.TrimSpace(os.Getenv(envDedicatedCloud)))
	if cloud == "" {
		cloud = "aws"
	}
	switch cloud {
	case "aws":
		return dedicatedTarget{
			CIDRProvider: "AWS",
			Cloud:        cloud,
			Region:       targetRegion,
			RegionID:     "aws-us-west-2",
			StorageType:  "Standard",
			TiDBNodeSpec: "8C64G-V2",
			TiKVNodeSpec: "8C32G-V2",
		}
	case "gcp":
		if !slices.Contains(scenarios, "dedicated") {
			t.Fatalf("%s=gcp requires the dedicated scenario", envDedicatedCloud)
		}
		return dedicatedTarget{
			CIDRProvider: "GCP",
			Cloud:        cloud,
			Region:       config.DedicatedGCPRegion,
			RegionID:     "gcp-" + config.DedicatedGCPRegion,
			StorageType:  "Basic",
			TiDBNodeSpec: "2C8G",
			TiKVNodeSpec: "2C8G",
		}
	default:
		t.Fatalf("%s must be aws or gcp, got %q", envDedicatedCloud, cloud)
		return dedicatedTarget{}
	}
}

func premiumCDCSourceBuildServices(
	t *testing.T,
	repoRoot string,
	scenarios []string,
	summary *workflowSummary,
) []string {
	t.Helper()
	if !enabled(os.Getenv(envPremiumCDCSourceBuild)) {
		return nil
	}
	startedAt := time.Now()
	status := "failed"
	message := ""
	defer func() {
		summary.record("inputs/premium-cdc-source", status, time.Since(startedAt), message)
	}()
	if !slices.Contains(scenarios, "premium") {
		t.Fatalf("%s requires the premium scenario", envPremiumCDCSourceBuild)
	}
	registry, err := sourcebacked.DefaultRegistry()
	if err != nil {
		t.Fatalf("load source-backed registry: %v", err)
	}
	contract, ok := registry.Contract("ticdc-service")
	if !ok {
		t.Fatal("source-backed registry has no ticdc-service contract")
	}
	tiltfile := filepath.Join(repoRoot, "personas", "dev-full", "runtime", "example.Tiltfile")
	sourceDir, err := tilt.ResolveSourceDirectory(tiltfile, contract.Source.Repo)
	if err != nil {
		t.Fatalf("resolve ticdc-service source checkout: %v", err)
	}
	if dirty := runLocalGit(t, sourceDir, "status", "--porcelain"); dirty != "" {
		t.Fatalf("ticdc-service source checkout must be clean")
	}
	head := runLocalGit(t, sourceDir, "rev-parse", "HEAD")
	message = "cluster-service-ng@" + head
	status = "passed"
	return []string{"ticdc-service"}
}

func runLocalGit(t *testing.T, dir string, args ...string) string {
	t.Helper()
	cmd := exec.Command("git", append([]string{"-C", dir}, args...)...)
	data, err := cmd.CombinedOutput()
	if err != nil {
		t.Fatalf("run git %s: %v\n%s", strings.Join(args, " "), err, data)
	}
	return strings.TrimSpace(string(data))
}

func repositoryRoot(t *testing.T) string {
	t.Helper()
	_, filename, _, ok := runtime.Caller(0)
	if !ok {
		t.Fatal("locate functional test source")
	}
	return filepath.Clean(filepath.Join(filepath.Dir(filename), ".."))
}

func buildCP9(t *testing.T, repoRoot string, summary *workflowSummary) string {
	t.Helper()
	output := filepath.Join(t.TempDir(), "cp9")
	startedAt := time.Now()
	cmd := exec.Command("go", "build", "-o", output, "./cmd/cp9")
	cmd.Dir = repoRoot
	data, err := cmd.CombinedOutput()
	status := "passed"
	message := ""
	if err != nil {
		status = "failed"
		message = err.Error()
	}
	summary.record("build", status, time.Since(startedAt), message)
	if err != nil {
		t.Fatalf("build cp9: %v\n%s", err, data)
	}
	return output
}

func copySessionTiltfile(
	t *testing.T,
	repoRoot string,
	sessionName string,
	sourceBuildServices []string,
	summary *workflowSummary,
) string {
	t.Helper()
	startedAt := time.Now()
	status := "failed"
	defer func() {
		summary.record("prepare/tiltfile", status, time.Since(startedAt), "")
	}()
	runtimeDir := filepath.Join(repoRoot, "personas", "dev-full", "runtime")
	source := filepath.Join(runtimeDir, "example.Tiltfile")
	data, err := os.ReadFile(source)
	if err != nil {
		t.Fatalf("read Session Tiltfile: %v", err)
	}
	if len(sourceBuildServices) > 0 {
		marker := []byte("source_build_services = [")
		if bytes.Count(data, marker) != 1 {
			t.Fatalf("Session Tiltfile must contain one source_build_services assignment")
		}
		start := bytes.Index(data, marker)
		end := bytes.Index(data[start:], []byte("\n]"))
		if end < 0 {
			t.Fatal("Session Tiltfile source_build_services assignment is unterminated")
		}
		end += start + len("\n]")
		selection, err := json.MarshalIndent(sourceBuildServices, "", "    ")
		if err != nil {
			t.Fatalf("encode Session source-build selection: %v", err)
		}
		replacement := append([]byte("source_build_services = "), selection...)
		data = append(append(append([]byte(nil), data[:start]...), replacement...), data[end:]...)
	}
	path := filepath.Join(runtimeDir, ".cp9-uiapi-"+url.PathEscape(sessionName)+".Tiltfile")
	t.Cleanup(func() {
		if !enabled(os.Getenv(envDestroySession)) {
			return
		}
		if _, err := os.Stat(filepath.Join(repoRoot, ".cp9", sessionName)); err == nil {
			return
		} else if !os.IsNotExist(err) {
			t.Errorf("inspect Session state before removing Tiltfile: %v", err)
			return
		}
		if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
			t.Errorf("remove Session Tiltfile: %v", err)
		}
	})
	if err := os.WriteFile(path, data, 0o600); err != nil {
		t.Fatalf("write Session Tiltfile: %v", err)
	}
	selected, err := tilt.SelectedSourceBuildServices(path)
	if err != nil {
		t.Fatalf("read Session source-build selection: %v", err)
	}
	if !slices.Equal(selected, sourceBuildServices) {
		t.Fatalf("Session source-build selection = %v, want %v", selected, sourceBuildServices)
	}
	status = "passed"
	return path
}

func writeSessionConfig(
	t *testing.T,
	session *config.SessionConfig,
	summary *workflowSummary,
) string {
	t.Helper()
	startedAt := time.Now()
	status := "failed"
	defer func() {
		summary.record("prepare/config", status, time.Since(startedAt), "")
	}()
	path := filepath.Join(t.TempDir(), "cp9.yaml")
	data, err := yaml.Marshal(session)
	if err != nil {
		t.Fatalf("marshal CP9 session config: %v", err)
	}
	if err := os.WriteFile(path, data, 0o600); err != nil {
		t.Fatalf("write CP9 session config: %v", err)
	}
	if _, err := config.LoadSession(path); err != nil {
		t.Fatalf("generated CP9 session config is invalid: %v\n%s", err, data)
	}
	t.Logf("CP9 session config:\n%s", data)
	status = "passed"
	return path
}

func currentCommit(t *testing.T, repoRoot string) string {
	t.Helper()
	cmd := exec.Command("git", "rev-parse", "HEAD")
	cmd.Dir = repoRoot
	data, err := cmd.Output()
	if err != nil {
		t.Fatalf("resolve current commit: %v", err)
	}
	return strings.TrimSpace(string(data))
}

func runCP9Stage(
	ctx context.Context,
	t *testing.T,
	env *testEnvironment,
	stage string,
	args ...string,
) error {
	t.Helper()
	startedAt := time.Now()
	cmd := exec.CommandContext(ctx, env.cp9Binary, args...)
	cmd.Dir = env.repoRoot
	var output bytes.Buffer
	stream := io.MultiWriter(os.Stderr, &output)
	cmd.Stdout = stream
	cmd.Stderr = stream
	err := cmd.Run()
	data := output.Bytes()
	status := "passed"
	message := ""
	if err != nil {
		status = "failed"
		message = err.Error()
	}
	env.summary.record(stage, status, time.Since(startedAt), message)
	if err != nil {
		return fmt.Errorf("cp9 %s failed: %w\n%s", stage, err, data)
	}
	return nil
}

func doAPIRequest(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	baseURL string,
	method string,
	path string,
	body []byte,
	contentType string,
	expectedStatus int,
) []byte {
	t.Helper()
	responseBody, _ := doAPIRequestStatuses(
		t,
		ctx,
		client,
		baseURL,
		method,
		path,
		body,
		contentType,
		expectedStatus,
	)
	return responseBody
}

func doAPIRequestStatuses(
	t *testing.T,
	ctx context.Context,
	client *http.Client,
	baseURL string,
	method string,
	path string,
	body []byte,
	contentType string,
	expectedStatuses ...int,
) ([]byte, int) {
	t.Helper()
	t.Logf("API request: %s %s\nrequest body:\n%s", method, path, body)

	base, err := url.Parse(baseURL)
	if err != nil {
		t.Fatalf("parse API base URL %q: %v", baseURL, err)
	}
	relative, err := url.Parse(path)
	if err != nil {
		t.Fatalf("parse API path %q: %v", path, err)
	}
	endpoint := base.ResolveReference(relative)
	request, err := http.NewRequestWithContext(ctx, method, endpoint.String(), bytes.NewReader(body))
	if err != nil {
		t.Fatalf("build API request %s %s with %d-byte body: %v", method, path, len(body), err)
	}
	request.Header.Set("Accept", "application/json+camel")
	request.Header.Set("Authorization", "Bearer "+strings.TrimSpace(os.Getenv(envAuthToken)))
	if contentType != "" {
		request.Header.Set("Content-Type", contentType)
	}

	response, err := client.Do(request)
	if err != nil {
		t.Fatalf("API request %s %s with %d-byte body: %v", method, path, len(body), err)
	}
	defer func() {
		_ = response.Body.Close()
	}()
	responseBody, err := io.ReadAll(response.Body)
	if err != nil {
		t.Fatalf("read API response for %s %s with %d-byte body: %v", method, path, len(body), err)
	}
	t.Logf(
		"API response: %s %s\nstatus: %s\nresponse body:\n%s",
		method,
		path,
		response.Status,
		responseBody,
	)
	if !slices.Contains(expectedStatuses, response.StatusCode) {
		t.Fatalf(
			"API request %s %s with %d-byte body returned %s, want one of HTTP %v\nresponse body:\n%s",
			method,
			path,
			len(body),
			response.Status,
			expectedStatuses,
			responseBody,
		)
	}
	return responseBody, response.StatusCode
}

func newWorkflowSummary(t *testing.T) *workflowSummary {
	t.Helper()
	summary := &workflowSummary{
		path: strings.TrimSpace(os.Getenv("GITHUB_STEP_SUMMARY")),
		t:    t,
	}
	if summary.path != "" {
		if err := summary.append("## CP9 public UI API scenarios\n\n| Stage | Status | Duration | Message |\n|---|---|---:|---|\n"); err != nil {
			t.Logf("harness/report: initialize GitHub Actions summary: %v", err)
		}
	}
	return summary
}

func (s *workflowSummary) record(stage, status string, duration time.Duration, message string) {
	s.t.Helper()
	message = strings.TrimSpace(message)
	s.t.Logf("%s: %s (%s) %s", stage, status, duration.Round(time.Millisecond), message)
	if s.path == "" {
		return
	}
	message = strings.ReplaceAll(message, "\n", "<br>")
	message = strings.ReplaceAll(message, "|", "\\|")
	if err := s.append(fmt.Sprintf(
		"| %s | %s | %s | %s |\n",
		stage,
		status,
		duration.Round(time.Millisecond),
		message,
	)); err != nil {
		s.t.Logf("harness/report: append GitHub Actions summary: %v", err)
	}
}

func (s *workflowSummary) append(value string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	file, err := os.OpenFile(s.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
	if err != nil {
		return err
	}
	defer func() {
		_ = file.Close()
	}()
	_, err = file.WriteString(value)
	return err
}

func enabled(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "1", "true", "yes":
		return true
	default:
		return false
	}
}