package metrics

import (
	"context"
	"database/sql/driver"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"
)

func hasMetricLineWith(text, metric string, fragments ...string) bool {
	for _, line := range strings.Split(text, "\n") {
		if !strings.HasPrefix(line, metric+"{") && !strings.HasPrefix(line, metric+" ") {
			continue
		}
		matches := true
		for _, fragment := range fragments {
			if !strings.Contains(line, fragment) {
				matches = false
				break
			}
		}
		if matches {
			return true
		}
	}
	return false
}

func TestResultForError(t *testing.T) {
	tests := []struct {
		name string
		err  error
		want string
	}{
		{name: "nil", err: nil, want: "ok"},
		{name: "context canceled", err: context.Canceled, want: "canceled"},
		{name: "context deadline", err: fmt.Errorf("query: %w", context.DeadlineExceeded), want: "deadline_exceeded"},
		{name: "bad conn sentinel", err: fmt.Errorf("query: %w", driver.ErrBadConn), want: "bad_conn"},
		{name: "invalid connection string", err: errors.New("driver: invalid connection"), want: "bad_conn"},
		{name: "invalid connection mixed case", err: errors.New("driver: Invalid Connection"), want: "bad_conn"},
		{name: "generic", err: errors.New("quota outbox lease mismatch"), want: "error"},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := ResultForError(tt.err); got != tt.want {
				t.Fatalf("ResultForError(%v) = %q, want %q", tt.err, got, tt.want)
			}
		})
	}
}

func TestRecordTenantOperationCountDoesNotRecordDuration(t *testing.T) {
	const component = "counter_only_test_component"
	const operation = "counter_only_test_operation"

	RecordTenantOperationCountWithOrg("tenant-a", "org-counter-only", component, operation, "ok")

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="`+component+`",operation="`+operation+`",result="ok"} 1`) {
		t.Fatalf("missing aggregate counter-only operation total:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_service_operations_total", `tenant_id="tenant-a"`) {
		t.Fatalf("aggregate operation total unexpectedly carried tenant labels:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_tenant_operation_failures_total", `tenant_id="tenant-a"`) {
		t.Fatalf("successful operation unexpectedly created a tenant failure series:\n%s", text)
	}
	if strings.Contains(text, `drive9_service_operation_duration_seconds_count{component="`+component+`",operation="`+operation+`",result="ok",tenant_id="tenant-a"}`) {
		t.Fatalf("counter-only operation unexpectedly recorded a duration:\n%s", text)
	}
}

func TestRecordTenantOperationZeroDurationDoesNotRecordDuration(t *testing.T) {
	const component = "zero_duration_test_component"
	const operation = "zero_duration_test_operation"

	RecordTenantOperationWithOrg("tenant-zero", "org-zero-duration", component, operation, "ok", 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="`+component+`",operation="`+operation+`",result="ok"} 1`) {
		t.Fatalf("missing zero-duration operation total:\n%s", text)
	}
	if strings.Contains(text, `drive9_service_operation_duration_seconds_count{component="`+component+`",operation="`+operation+`",result="ok",tenant_id="tenant-zero"}`) {
		t.Fatalf("zero-duration operation unexpectedly recorded a duration:\n%s", text)
	}
}

func TestRecordTenantOperationOmitsTenantFromDurationHistograms(t *testing.T) {
	const operation = "duration_tenant_omit_test_operation"

	for _, component := range []string{"backend", "central_quota", "file_gc", "quota_config_cache", "server_quota", "user_db_access", "vault"} {
		t.Run(component, func(t *testing.T) {
			tenantID := "tenant-duration-omit-" + component
			RecordTenantOperationWithOrg(tenantID, "org-"+component, component, operation, "ok", time.Second)

			rec := httptest.NewRecorder()
			WritePrometheus(rec)
			text := rec.Body.String()
			if !strings.Contains(text, `drive9_service_operations_total{component="`+component+`",operation="`+operation+`",result="ok"} 1`) {
				t.Fatalf("missing aggregate operation total:\n%s", text)
			}
			if hasMetricLineWith(text, "drive9_service_operations_total", `tenant_id="`+tenantID+`"`) {
				t.Fatalf("aggregate operation total unexpectedly carried tenant labels:\n%s", text)
			}
			if !strings.Contains(text, `drive9_service_operation_duration_seconds_count{component="`+component+`",operation="`+operation+`",result="ok"} 1`) {
				t.Fatalf("missing tenant-omitted duration histogram:\n%s", text)
			}
			if strings.Contains(text, `drive9_service_operation_duration_seconds_count{component="`+component+`",operation="`+operation+`",result="ok",tenant_id="`+tenantID+`"}`) {
				t.Fatalf("duration histogram unexpectedly carried tenant_id:\n%s", text)
			}
		})
	}
}

func TestRecordTenantRequestOmitsHighCardinalityLabels(t *testing.T) {
	const (
		tenantID = "tenant-request-duration-omit"
		surface  = "fs"
		action   = "request_duration_action"
	)

	RecordTenantRequestWithOrg(tenantID, "org-request-duration", surface, action, 201, time.Second)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_requests_total{action="`+action+`",status_class="2xx",surface="`+surface+`",tenant_id="`+tenantID+`",tidbcloud_org_id="org-request-duration"} 1`) {
		t.Errorf("missing tenant-scoped request total:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_request_duration_seconds_count{status_class="2xx",surface="`+surface+`"} 1`) {
		t.Errorf("missing reduced request duration histogram:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_tenant_requests_total", `status="201"`) {
		t.Errorf("tenant request total unexpectedly carried raw status label:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_tenant_requests_total", `result=`) {
		t.Errorf("tenant request total unexpectedly carried derived result label:\n%s", text)
	}
	for _, unexpected := range []string{
		`action="` + action + `"`,
		`result="ok"`,
		`tenant_id="` + tenantID + `"`,
		`tidbcloud_org_id="org-request-duration"`,
	} {
		if hasMetricLineWith(text, "drive9_tenant_request_duration_seconds_count", unexpected) {
			t.Errorf("tenant request duration unexpectedly carried label %q:\n%s", unexpected, text)
		}
	}
}

func TestRecordTenantRequestZeroDurationDoesNotRecordDuration(t *testing.T) {
	const tenantID = "tenant-zero-request"

	RecordTenantRequest(tenantID, "object_store", "zero_action", 200, 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_requests_total{action="zero_action",status_class="2xx",surface="object_store",tenant_id="`+tenantID+`",tidbcloud_org_id="guest"} 1`) {
		t.Errorf("missing zero-duration tenant request total:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_tenant_request_duration_seconds_count", `surface="object_store"`, `status_class="2xx"`) {
		t.Errorf("zero-duration tenant request unexpectedly recorded a duration:\n%s", text)
	}
}

func TestRecordSSEConnectionZeroDurationDoesNotRecordDuration(t *testing.T) {
	const tenantID = "tenant-zero-sse-connection"

	RecordSSEConnectionWithOrg(tenantID, "org-sse-zero", "closed", 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_sse_connections_total{reason="closed",tenant_id="`+tenantID+`",tidbcloud_org_id="org-sse-zero"} 1`) {
		t.Fatalf("missing zero-duration SSE connection total:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_sse_connection_duration_seconds_count", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("zero-duration SSE connection unexpectedly recorded a duration:\n%s", text)
	}
}

func TestRecordTenantDurationMetricsSkipZeroDuration(t *testing.T) {
	const (
		phaseTenant    = "tenant-zero-sse-phase1"
		eventBusTenant = "tenant-zero-event-bus"
	)

	RecordSSEPhase1(phaseTenant, 0)
	RecordEventBusQuery(eventBusTenant, "events_since", "ok", 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if hasMetricLineWith(text, "drive9_sse_phase1_duration_seconds_count", `tenant_id="`+phaseTenant+`"`) {
		t.Fatalf("zero-duration SSE phase1 unexpectedly recorded a duration:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_event_bus_query_duration_seconds_count", `operation="events_since"`, `result="ok"`) {
		t.Fatalf("zero-duration event bus query unexpectedly recorded a duration:\n%s", text)
	}
}

func TestRecordEventBusQueryDurationOmitsTenant(t *testing.T) {
	const tenantID = "tenant-event-bus-duration-omit"

	RecordEventBusQuery(tenantID, "event_bus_duration_omit", "ok", time.Second)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_event_bus_query_duration_seconds_count{operation="event_bus_duration_omit",result="ok"} 1`) {
		t.Errorf("missing tenant-omitted event bus duration:\n%s", text)
	}
	if strings.Contains(text, `drive9_event_bus_query_duration_seconds_count{operation="event_bus_duration_omit",result="ok",tenant_id="`+tenantID+`"}`) {
		t.Errorf("event bus duration unexpectedly carried tenant_id:\n%s", text)
	}
}

func TestRecordTenantInFlightDeletesZeroValue(t *testing.T) {
	const (
		tenantID = "tenant-inflight-delete"
		surface  = "inflight_delete_surface"
		action   = "inflight_delete_action"
	)

	RecordTenantInFlightWithOrg(tenantID, "org-inflight-delete", surface, 1)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_inflight_requests{surface="`+surface+`",tenant_id="`+tenantID+`",tidbcloud_org_id="org-inflight-delete"} 1.000000`) {
		t.Errorf("missing tenant in-flight gauge:\n%s", text)
	}

	RecordTenantInFlightWithOrg(tenantID, "org-inflight-delete", surface, 0)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if strings.Contains(text, `drive9_tenant_inflight_requests{surface="`+surface+`",tenant_id="`+tenantID+`",tidbcloud_org_id="org-inflight-delete"}`) {
		t.Errorf("tenant in-flight gauge should be deleted at zero:\n%s", text)
	}
}

func TestDeleteTenantGaugeAndSSEInFlightRemoveSeries(t *testing.T) {
	const tenantID = "tenant-gauge-delete"
	const orgID = "org-gauge-delete"

	RecordTenantGaugeWithOrg(tenantID, orgID, "semantic_worker", "queue_lag_seconds", 12)
	RecordSSEInFlightWithOrg(tenantID, orgID, 2)
	DeleteTenantGaugeWithOrg(tenantID, orgID, "semantic_worker", "queue_lag_seconds")
	RecordSSEInFlightWithOrg(tenantID, orgID, 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if hasMetricLineWith(text, "drive9_service_gauge", `tenant_id="`+tenantID+`"`, `name="queue_lag_seconds"`) {
		t.Fatalf("deleted tenant gauge remains exported:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_sse_inflight_connections", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("zero SSE in-flight series remains exported:\n%s", text)
	}
}

func TestReplaceTenantQuotaSnapshotRemovesOnlyAbsentLimitSeries(t *testing.T) {
	const tenantID = "tenant-quota-limit-delete"
	const orgID = "org-quota-limit-delete"

	ReplaceTenantQuotaSnapshotByStorageTierWithOrg(tenantID, orgID, 10, nil, nil, false, 1, 100, true)
	ReplaceTenantQuotaSnapshotByStorageTierWithOrg(tenantID, orgID, 11, nil, nil, false, 1, 0, true)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !hasMetricLineWith(text, "drive9_tenant_storage_bytes", `tenant_id="`+tenantID+`"`, `state="confirmed"`) {
		t.Fatalf("confirmed usage was deleted while limits were replaced:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_tenant_storage_bytes", `tenant_id="`+tenantID+`"`, `state="limit"`) {
		t.Fatalf("stale storage limit remains:\n%s", text)
	}
}

func TestReplaceTenantQuotaStorageSnapshotSwitchesFromAllToFixedTierSeries(t *testing.T) {
	const tenantID = "tenant-quota-tier-switch"
	const orgID = "org-quota-tier-switch"
	dbBytes := int64(10)
	s3Bytes := int64(0)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	ReplaceTenantQuotaSnapshotByStorageTierWithOrg(tenantID, orgID, 10, nil, nil, false, 2, 100, true)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	for _, want := range []string{
		`drive9_tenant_storage_bytes{state="confirmed",storage_type="all",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 10.000000`,
		`drive9_tenant_storage_bytes{state="reserved",storage_type="all",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 2.000000`,
		`drive9_tenant_storage_bytes{state="limit",storage_type="all",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 100.000000`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing pre-readiness storage series %q:\n%s", want, text)
		}
	}
	if hasMetricLineWith(text, "drive9_tenant_storage_bytes", `tenant_id="`+tenantID+`"`, `storage_type="db9"`) ||
		hasMetricLineWith(text, "drive9_tenant_storage_bytes", `tenant_id="`+tenantID+`"`, `storage_type="s3"`) {
		t.Fatalf("tier series exported before readiness:\n%s", text)
	}

	ReplaceTenantQuotaSnapshotByStorageTierWithOrg(tenantID, orgID, 10, &dbBytes, &s3Bytes, true, 3, 100, true)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if hasMetricLineWith(text, "drive9_tenant_storage_bytes", `tenant_id="`+tenantID+`"`, `state="confirmed"`, `storage_type="all"`) {
		t.Fatalf("confirmed/all remained after readiness switch:\n%s", text)
	}
	for _, want := range []string{
		`drive9_tenant_storage_bytes{state="confirmed",storage_type="db9",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 10.000000`,
		`drive9_tenant_storage_bytes{state="confirmed",storage_type="s3",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 0.000000`,
		`drive9_tenant_storage_bytes{state="reserved",storage_type="all",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 3.000000`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing ready storage series %q:\n%s", want, text)
		}
	}
	series := 0
	for _, line := range strings.Split(text, "\n") {
		if strings.HasPrefix(line, "drive9_tenant_storage_bytes{") && strings.Contains(line, `tenant_id="`+tenantID+`"`) {
			series++
		}
	}
	if series != 4 {
		t.Fatalf("ready tenant storage series = %d, want exactly 4:\n%s", series, text)
	}
}

func TestReplaceTenantFileCountSnapshotSwitchesFromAllToFixedTierSeries(t *testing.T) {
	const tenantID = "tenant-quota-file-count-tier-switch"
	const orgID = "org-quota-file-count-tier-switch"
	dbFiles := int64(4)
	s3Files := int64(2)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	// Before the tier scan is ready the split counters are absent: only the
	// storage_type="all" series may be exported.
	ReplaceTenantFileCountSnapshotWithOrg(tenantID, orgID, 6, nil, nil, false)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	want := `drive9_tenant_file_count{state="confirmed",storage_type="all",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 6.000000`
	if !strings.Contains(text, want) {
		t.Fatalf("missing pre-readiness file count series %q:\n%s", want, text)
	}
	if hasMetricLineWith(text, "drive9_tenant_file_count", `tenant_id="`+tenantID+`"`, `storage_type="db9"`) ||
		hasMetricLineWith(text, "drive9_tenant_file_count", `tenant_id="`+tenantID+`"`, `storage_type="s3"`) {
		t.Fatalf("file count tier series exported before readiness:\n%s", text)
	}

	// Ready and reconciled (db9+s3 == total): export only the tier series.
	ReplaceTenantFileCountSnapshotWithOrg(tenantID, orgID, 6, &dbFiles, &s3Files, true)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if hasMetricLineWith(text, "drive9_tenant_file_count", `tenant_id="`+tenantID+`"`, `storage_type="all"`) {
		t.Fatalf("file count all series remained after readiness switch:\n%s", text)
	}
	for _, want := range []string{
		`drive9_tenant_file_count{state="confirmed",storage_type="db9",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 4.000000`,
		`drive9_tenant_file_count{state="confirmed",storage_type="s3",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 2.000000`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing ready file count series %q:\n%s", want, text)
		}
	}

	// A split that does not reconcile with file_count (e.g. active upload
	// reservations inflating the counter) falls back to "all".
	s3Files = 1
	ReplaceTenantFileCountSnapshotWithOrg(tenantID, orgID, 6, &dbFiles, &s3Files, true)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if !strings.Contains(text, want) {
		t.Fatalf("missing fallback file count series %q:\n%s", want, text)
	}
	if hasMetricLineWith(text, "drive9_tenant_file_count", `tenant_id="`+tenantID+`"`, `storage_type="db9"`) {
		t.Fatalf("unreconciled file count tier series exported:\n%s", text)
	}
}

func TestReplaceTenantFileCountSnapshotOrgChangeRemovesOldSeries(t *testing.T) {
	const tenantID = "tenant-quota-file-count-org-change"
	dbFiles := int64(4)
	s3Files := int64(2)
	t.Cleanup(func() { DeleteTenantCounters(tenantID) })

	ReplaceTenantFileCountSnapshotWithOrg(tenantID, "org-old", 6, &dbFiles, &s3Files, true)
	ReplaceTenantFileCountSnapshotWithOrg(tenantID, "org-new", 6, &dbFiles, &s3Files, true)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	for _, line := range strings.Split(rec.Body.String(), "\n") {
		if strings.HasPrefix(line, "drive9_tenant_file_count{") &&
			strings.Contains(line, `tenant_id="`+tenantID+`"`) &&
			strings.Contains(line, `tidbcloud_org_id="org-old"`) {
			t.Fatalf("old org file count series remains exported: %s", line)
		}
	}
}

func TestOrganizationStorageUsageMetricUsesOnlyOrganizationLabel(t *testing.T) {
	const orgID = "org-storage-usage-metric"
	RecordOrganizationStorageUsage(orgID, 123)
	t.Cleanup(func() { DeleteOrganizationStorageUsage(orgID) })

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	line := `drive9_organization_storage_usage_bytes{tidbcloud_org_id="` + orgID + `"} 123.000000`
	if !strings.Contains(rec.Body.String(), line) {
		t.Fatalf("missing organization storage usage metric %q:\n%s", line, rec.Body.String())
	}
	if hasMetricLineWith(rec.Body.String(), "drive9_organization_storage_usage_bytes", "tenant_id=") {
		t.Fatalf("organization storage usage metric must not carry tenant_id:\n%s", rec.Body.String())
	}
}

func TestOrganizationUsageSnapshotUsesOnlyOrganizationLabel(t *testing.T) {
	const orgID = "org-storage-breakdown"
	DeleteOrganizationStorageUsage(orgID)
	t.Cleanup(func() { DeleteOrganizationStorageUsage(orgID) })

	RecordOrganizationUsageSnapshot(orgID, OrganizationUsageSnapshot{
		ConfirmedBytes:    100,
		ReservedBytes:     25,
		StorageLimitBytes: 500,
		TenantCount:       3,
		FileCount:         42,
	})
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	for _, want := range []string{
		`drive9_organization_storage_usage_bytes{tidbcloud_org_id="` + orgID + `"} 125.000000`,
		`drive9_organization_storage_confirmed_bytes{tidbcloud_org_id="` + orgID + `"} 100.000000`,
		`drive9_organization_storage_reserved_bytes{tidbcloud_org_id="` + orgID + `"} 25.000000`,
		`drive9_organization_storage_limit_bytes{tidbcloud_org_id="` + orgID + `"} 500.000000`,
		`drive9_organization_tenant_count{tidbcloud_org_id="` + orgID + `"} 3.000000`,
		`drive9_organization_file_count{tidbcloud_org_id="` + orgID + `"} 42.000000`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing organization storage breakdown metric %q:\n%s", want, text)
		}
	}
	for _, metric := range []string{
		"drive9_organization_storage_usage_bytes",
		"drive9_organization_storage_confirmed_bytes",
		"drive9_organization_storage_reserved_bytes",
		"drive9_organization_storage_limit_bytes",
		"drive9_organization_tenant_count",
		"drive9_organization_file_count",
	} {
		if hasMetricLineWith(text, metric, "tenant_id=") {
			t.Fatalf("organization storage metric %s must not carry tenant_id:\n%s", metric, text)
		}
	}
}

func TestTenantIDMetricsIncludeTiDBCloudOrgID(t *testing.T) {
	const (
		tenantID = "tenant-org-label-test"
		orgID    = "org-label-test"
	)

	RecordTenantGaugeWithOrg(tenantID, orgID, "component_org_label_test", "gauge", 1)
	RecordTenantEventWithOrg(tenantID, orgID, "event_org_label_test", "result", "ok")
	RecordTenantRequestCountWithOrg(tenantID, orgID, "fs", "action", 200)
	RecordTenantFileBytesWithOrg(tenantID, orgID, "read", "db9", 20)
	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "db9", 1)
	RecordTenantStorageBytesWithOrg(tenantID, orgID, "confirmed", 30)
	RecordSSEInFlightWithOrg(tenantID, orgID, 1)
	RecordSSEPhase1WithOrg(tenantID, orgID, time.Second)
	RecordSSEEventSentWithOrg(tenantID, orgID, "write")
	RecordSSEResetSentWithOrg(tenantID, orgID, "seq_too_old")
	RecordSSEHeartbeatSentWithOrg(tenantID, orgID)
	RecordFSEventsRowsWithOrg(tenantID, orgID, 50)
	RecordFSEventsPruned(60)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	for _, want := range []string{
		`drive9_service_gauge{component="component_org_label_test",name="gauge",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1.000000`,
		`drive9_business_events_total{event="event_org_label_test",result="ok"} 1`,
		`drive9_tenant_requests_total{action="action",status_class="2xx",surface="fs",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_tenant_file_bytes_total{direction="read",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `",traffic_path="db9"} 20`,
		`drive9_tenant_file_ops_total{direction="read",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `",traffic_path="db9"} 1`,
		`drive9_tenant_storage_bytes{state="confirmed",storage_type="all",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 30.000000`,
		`drive9_sse_inflight_connections{tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1.000000`,
		`drive9_sse_phase1_duration_seconds_count{tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_sse_events_sent_total{op="write",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_sse_resets_sent_total{reason="seq_too_old",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_sse_heartbeats_sent_total{tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_fs_events_rows{tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 50.000000`,
		`drive9_fs_events_pruned_total 60`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing org-scoped tenant metric %q:\n%s", want, text)
		}
	}
}

func TestRecordTenantFileBytesAllowsOnlyBillingDimensions(t *testing.T) {
	const (
		tenantID = "tenant-file-byte-dimensions"
		orgID    = "org-file-byte-dimensions"
	)

	for _, direction := range []string{"read", "write"} {
		for _, trafficPath := range []string{"db9", "s3"} {
			RecordTenantFileBytesWithOrg(tenantID, orgID, direction, trafficPath, 1)
		}
	}
	RecordTenantFileBytesWithOrg(tenantID, orgID, "copy", "db9", 1)
	RecordTenantFileBytesWithOrg(tenantID, orgID, "read", "proxy", 1)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	series := 0
	for _, line := range strings.Split(text, "\n") {
		if !strings.HasPrefix(line, "drive9_tenant_file_bytes_total{") ||
			!strings.Contains(line, `tenant_id="`+tenantID+`"`) {
			continue
		}
		series++
		if strings.Contains(line, `direction="copy"`) || strings.Contains(line, `traffic_path="proxy"`) {
			t.Fatalf("invalid billing dimension was exported: %s", line)
		}
	}
	if series != 4 {
		t.Fatalf("tenant file byte series = %d, want exactly 4:\n%s", series, text)
	}
}

func TestRecordTenantFileOpsAllowsOnlyBillingDimensions(t *testing.T) {
	const (
		tenantID = "tenant-file-op-dimensions"
		orgID    = "org-file-op-dimensions"
	)

	for _, direction := range []string{"read", "write"} {
		for _, trafficPath := range []string{"db9", "s3"} {
			RecordTenantFileOpsWithOrg(tenantID, orgID, direction, trafficPath, 1)
		}
	}
	// Empty-file operations still count as one op.
	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "db9", 1)
	RecordTenantFileOpsWithOrg(tenantID, orgID, "copy", "db9", 1)
	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "proxy", 1)
	RecordTenantFileOpsWithOrg(tenantID, orgID, "read", "db9", 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	series := 0
	for _, line := range strings.Split(text, "\n") {
		if !strings.HasPrefix(line, "drive9_tenant_file_ops_total{") ||
			!strings.Contains(line, `tenant_id="`+tenantID+`"`) {
			continue
		}
		series++
		if strings.Contains(line, `direction="copy"`) || strings.Contains(line, `traffic_path="proxy"`) {
			t.Fatalf("invalid billing dimension was exported: %s", line)
		}
	}
	if series != 4 {
		t.Fatalf("tenant file ops series = %d, want exactly 4:\n%s", series, text)
	}
	if !strings.Contains(text, `drive9_tenant_file_ops_total{direction="read",tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`",traffic_path="db9"} 2`) {
		t.Fatalf("expected read/db9 ops to total 2 (empty-file op included, zero-op call dropped):\n%s", text)
	}
}

func TestDeleteFSEventsRowsRemovesHealthySeries(t *testing.T) {
	const tenantID = "tenant-fs-events-rows-delete"
	const orgID = "org-fs-events-rows-delete"
	RecordFSEventsRowsWithOrg(tenantID, orgID, 12)
	DeleteFSEventsRowsWithOrg(tenantID, orgID)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	if hasMetricLineWith(rec.Body.String(), "drive9_fs_events_rows", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("deleted fs_events row gauge remains exported:\n%s", rec.Body.String())
	}
}

func TestObserveFileGCClaimableBacklogSparseGauge(t *testing.T) {
	const tenantID = "tenant-file-gc-backlog"
	const orgID = "org-file-gc-backlog"

	ObserveFileGCClaimableBacklog(tenantID, orgID, 3, 2*time.Hour)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"} 3`) {
		t.Fatalf("missing claimable gauge:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_oldest_claimable_age_seconds{tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"} 7200`) {
		t.Fatalf("missing oldest age gauge:\n%s", text)
	}

	// Zero deletes the sparse series.
	ObserveFileGCClaimableBacklog(tenantID, orgID, 0, 0)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatalf("claimable gauge should be deleted at zero:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_oldest_claimable_age_seconds{tenant_id="`+tenantID+`"`) {
		t.Fatalf("oldest age gauge should be deleted at zero:\n%s", text)
	}
}

func TestClearFileGCClaimableBacklog(t *testing.T) {
	const tenantID = "tenant-file-gc-clear"
	const orgID = "org-file-gc-clear"
	ObserveFileGCClaimableBacklog(tenantID, orgID, 7, time.Hour)
	ObserveFileGCDrainHitCap(tenantID, orgID, true)
	ObserveFileGCClaimStalled(tenantID, orgID, true)
	ClearFileGCClaimableBacklog(tenantID)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatalf("ClearFileGCClaimableBacklog left claimable series:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_drain_hit_cap{tenant_id="`+tenantID+`"`) {
		t.Fatalf("ClearFileGCClaimableBacklog left drain_hit_cap series:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_oldest_claimable_age_seconds{tenant_id="`+tenantID+`"`) {
		t.Fatalf("ClearFileGCClaimableBacklog left oldest age series:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_claim_stalled{tenant_id="`+tenantID+`"`) {
		t.Fatalf("ClearFileGCClaimableBacklog left claim_stalled series:\n%s", text)
	}
}

func TestObserveFileGCClaimStalledSparseGauge(t *testing.T) {
	const tenantID = "tenant-file-gc-claim-stalled"
	const orgID = "org-file-gc-claim-stalled"
	ObserveFileGCClaimStalled(tenantID, orgID, true)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	if !hasMetricLineWith(rec.Body.String(), "drive9_file_gc_claim_stalled", `tenant_id="`+tenantID+`"`, `tidbcloud_org_id="`+orgID+`"`) {
		t.Fatalf("missing claim_stalled gauge:\n%s", rec.Body.String())
	}
	ObserveFileGCClaimStalled(tenantID, orgID, false)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	if hasMetricLineWith(rec.Body.String(), "drive9_file_gc_claim_stalled", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("claim_stalled should be deleted when false:\n%s", rec.Body.String())
	}
}

func TestFileGCDrainHitCapExpiresIndependentOfTenantActivity(t *testing.T) {
	const tenantID = "tenant-file-gc-hit-cap-ttl"
	const orgID = "org-file-gc-hit-cap-ttl"
	now := time.Unix(1_700_000_000, 0).UTC()
	fileGCBacklogClock = func() time.Time { return now }
	t.Cleanup(func() { fileGCBacklogClock = time.Now })

	ObserveFileGCClaimableBacklog(tenantID, orgID, 5, time.Hour)
	ObserveFileGCDrainHitCap(tenantID, orgID, true)
	RecordTenantRequestCountWithOrg(tenantID, orgID, "fs", "read", 200)

	now = now.Add(time.Minute)
	RecordTenantRequestCountWithOrg(tenantID, orgID, "fs", "read", 200)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !hasMetricLineWith(text, "drive9_file_gc_drain_hit_cap", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("hit_cap expired before TTL despite tenant activity:\n%s", text)
	}
	if !hasMetricLineWith(text, "drive9_file_gc_claimable", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("claimable missing while hit_cap still live:\n%s", text)
	}

	now = now.Add(defaultFileGCDrainHitCapTTL)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if hasMetricLineWith(text, "drive9_file_gc_drain_hit_cap", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("hit_cap should expire on FileGC last-touch TTL, not tenant activity:\n%s", text)
	}
	if !hasMetricLineWith(text, "drive9_file_gc_claimable", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("claimable must remain after hit_cap TTL so leftover can page:\n%s", text)
	}
}

func TestFileGCClaimStalledExpiresIndependentOfTenantActivity(t *testing.T) {
	const tenantID = "tenant-file-gc-claim-stalled-ttl"
	const orgID = "org-file-gc-claim-stalled-ttl"
	now := time.Unix(1_700_200_000, 0).UTC()
	fileGCBacklogClock = func() time.Time { return now }
	t.Cleanup(func() { fileGCBacklogClock = time.Now })

	ObserveFileGCClaimStalled(tenantID, orgID, true)
	RecordTenantRequestCountWithOrg(tenantID, orgID, "fs", "read", 200)
	now = now.Add(defaultFileGCDrainHitCapTTL)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	if hasMetricLineWith(rec.Body.String(), "drive9_file_gc_claim_stalled", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("claim_stalled should expire on FileGC last-touch TTL:\n%s", rec.Body.String())
	}
}

func TestTouchFileGCDrainHitCapExtendsTTL(t *testing.T) {
	const tenantID = "tenant-file-gc-hit-cap-touch"
	const orgID = "org-file-gc-hit-cap-touch"
	now := time.Unix(1_700_100_000, 0).UTC()
	fileGCBacklogClock = func() time.Time { return now }
	t.Cleanup(func() { fileGCBacklogClock = time.Now })

	ObserveFileGCDrainHitCap(tenantID, orgID, true)
	now = now.Add(defaultFileGCDrainHitCapTTL - time.Second)
	TouchFileGCDrainHitCap(tenantID, orgID)
	now = now.Add(defaultFileGCDrainHitCapTTL - time.Second)
	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	if !hasMetricLineWith(rec.Body.String(), "drive9_file_gc_drain_hit_cap", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("in-flight drain touch must keep hit_cap inside TTL:\n%s", rec.Body.String())
	}
	now = now.Add(defaultFileGCDrainHitCapTTL)
	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	if hasMetricLineWith(rec.Body.String(), "drive9_file_gc_drain_hit_cap", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("hit_cap should expire after last drain touch:\n%s", rec.Body.String())
	}
}

func TestFileGCBacklogTenantIDs(t *testing.T) {
	const a = "tenant-file-gc-list-a"
	const b = "tenant-file-gc-list-b"
	const org = "org-file-gc-list"
	ObserveFileGCClaimableBacklog(a, org, 1, time.Minute)
	ObserveFileGCDrainHitCap(b, org, true)
	ids := FileGCBacklogTenantIDs()
	haveA, haveB := false, false
	for _, id := range ids {
		if id == a {
			haveA = true
		}
		if id == b {
			haveB = true
		}
	}
	if !haveA || !haveB {
		t.Fatalf("FileGCBacklogTenantIDs=%v, want %s and %s", ids, a, b)
	}
	ClearFileGCClaimableBacklog(a)
	ClearFileGCClaimableBacklog(b)
	for _, id := range FileGCBacklogTenantIDs() {
		if id == a || id == b {
			t.Fatalf("cleared tenant %s still listed: %v", id, FileGCBacklogTenantIDs())
		}
	}
}

func TestTenantOperationFailuresUseSeparateFailureOnlyMetric(t *testing.T) {
	const tenantID = "tenant-operation-failure-contract"
	const orgID = "org-operation-failure-contract"

	RecordTenantOperationCountWithOrg(tenantID, orgID, "contract", "work", "queued")
	RecordTenantOperationCountWithOrg(tenantID, orgID, "contract", "work", "write_error")
	RecordTenantOperationCountWithOrg(tenantID, orgID, "contract", "quota", "fail_open")
	RecordTenantOperationCountWithOrg(tenantID, orgID, "contract", "extract", "runtime_not_configured")
	// Worker panic path: classified as failure for tenant attribution without
	// success-path tenant labels (see isOperationFailureResult "panic").
	RecordTenantOperationCountWithOrg(tenantID, orgID, "tenant_worker", "process_kicked", "panic")

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="contract",operation="work",result="queued"} 1`) ||
		!strings.Contains(text, `drive9_service_operations_total{component="contract",operation="work",result="write_error"} 1`) {
		t.Fatalf("missing aggregate operation totals:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_service_operations_total", `tenant_id="`+tenantID+`"`) {
		t.Fatalf("aggregate operation total carried tenant labels:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_operation_failures_total{component="contract",operation="work",result="write_error",tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"} 1`) {
		t.Fatalf("missing tenant failure attribution metric:\n%s", text)
	}
	for _, want := range []string{
		`drive9_tenant_operation_failures_total{component="contract",operation="quota",result="fail_open",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_tenant_operation_failures_total{component="contract",operation="extract",result="runtime_not_configured",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
		`drive9_tenant_operation_failures_total{component="tenant_worker",operation="process_kicked",result="panic",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing explicit tenant failure attribution %q:\n%s", want, text)
		}
	}
	if hasMetricLineWith(text, "drive9_tenant_operation_failures_total", `result="queued"`) {
		t.Fatalf("non-failure result created tenant failure series:\n%s", text)
	}
}

func TestBusinessEventTenantLabelsAreLimitedToFailures(t *testing.T) {
	RecordTenantEventWithOrg("tenant-event-contract", "org-event-contract", "fs_write", "result", "ok")
	RecordTenantEventWithOrg("tenant-event-contract", "org-event-contract", "tenant_provision", "provider", "local", "result", "accepted")
	RecordTenantEventWithOrg("tenant-event-contract", "org-event-contract", "tenant_provision", "provider", "local", "result", "cluster_error")
	RecordTenantEventWithOrg("tenant-event-contract", "org-event-contract", "tenant_schema_init", "provider", "local", "result", "error")
	RecordTenantEventWithOrg("tenant-event-contract", "org-event-contract", "fs_invalid_path")

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_business_events_total{event="fs_write",result="ok"} 1`) {
		t.Fatalf("hot-path event was not aggregated:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_business_events_total", `event="fs_write"`, `tenant_id=`) {
		t.Fatalf("hot-path event carried tenant labels:\n%s", text)
	}
	if !strings.Contains(text, `drive9_business_events_total{event="tenant_provision",provider="local",result="accepted"} 1`) {
		t.Fatalf("successful provisioning event was not aggregated:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_business_events_total", `event="tenant_provision"`, `result="accepted"`, `tenant_id=`) {
		t.Fatalf("successful provisioning event carried tenant labels:\n%s", text)
	}
	if !strings.Contains(text, `drive9_business_events_total{event="tenant_provision",provider="local",result="cluster_error",tenant_id="tenant-event-contract",tidbcloud_org_id="org-event-contract"} 1`) {
		t.Fatalf("failed provisioning event lost tenant attribution:\n%s", text)
	}
	if !strings.Contains(text, `drive9_business_events_total{event="tenant_schema_init",provider="local",result="error",tenant_id="tenant-event-contract",tidbcloud_org_id="org-event-contract"} 1`) {
		t.Fatalf("schema-init event lost tenant attribution:\n%s", text)
	}
	if !strings.Contains(text, `drive9_business_events_total{event="fs_invalid_path",tenant_id="tenant-event-contract",tidbcloud_org_id="org-event-contract"} 1`) {
		t.Fatalf("invalid-path event lost tenant attribution:\n%s", text)
	}
}

func TestTenantRequestAlwaysIncludesOrgAndLimitsTenantLabel(t *testing.T) {
	RecordTenantRequestWithOrg("tenant-request-contract", "org-request-contract", "provision", "post", http.StatusAccepted, time.Second)
	RecordTenantRequestWithOrg("tenant-request-contract", "org-request-contract", "provision", "post", http.StatusBadRequest, time.Second)
	RecordTenantRequestWithOrg("tenant-request-contract", "org-request-contract", "fs", "write", http.StatusAccepted, time.Second)
	RecordTenantRequestWithOrg("tenant-request-contract", "org-request-contract", "fork", "create", http.StatusAccepted, time.Second)
	RecordTenantRequestWithOrg("tenant-request-contract", "org-request-contract", "layer", "get", http.StatusOK, time.Second)
	RecordTenantRequestWithOrg("tenant-request-contract", "org-request-contract", "migration", "event", http.StatusNoContent, time.Second)
	RecordTenantRequestCountWithOrg("tenant-request-contract", "org-request-contract", "status", "get", http.StatusOK)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_requests_total{action="post",status_class="2xx",surface="provision",tidbcloud_org_id="org-request-contract"} 1`) {
		t.Errorf("successful control-plane request did not retain organization label:\n%s", text)
	}
	if hasMetricLineWith(text, "drive9_tenant_requests_total", `surface="provision"`, `status_class="2xx"`, `tenant_id=`) {
		t.Errorf("successful control-plane request carried tenant labels:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_requests_total{action="get",status_class="2xx",surface="status",tidbcloud_org_id="org-request-contract"} 1`) {
		t.Errorf("counter-only control-plane request did not retain organization label:\n%s", text)
	}
	for _, want := range []string{
		`drive9_tenant_requests_total{action="post",status_class="4xx",surface="provision",tenant_id="tenant-request-contract",tidbcloud_org_id="org-request-contract"} 1`,
		`drive9_tenant_requests_total{action="write",status_class="2xx",surface="fs",tenant_id="tenant-request-contract",tidbcloud_org_id="org-request-contract"} 1`,
		`drive9_tenant_requests_total{action="create",status_class="2xx",surface="fork",tenant_id="tenant-request-contract",tidbcloud_org_id="org-request-contract"} 1`,
		`drive9_tenant_requests_total{action="get",status_class="2xx",surface="layer",tenant_id="tenant-request-contract",tidbcloud_org_id="org-request-contract"} 1`,
		`drive9_tenant_requests_total{action="event",status_class="2xx",surface="migration",tenant_id="tenant-request-contract",tidbcloud_org_id="org-request-contract"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Errorf("missing tenant-scoped request metric %q:\n%s", want, text)
		}
	}
}

func TestRecordTenantCountUsesStatus(t *testing.T) {
	RecordTenantCount("pending_test", 1)
	RecordTenantCount("provisioning_test", 2)
	RecordTenantCount("active_test", 3)
	RecordTenantCount("failed_test", 4)
	RecordTenantCount("suspended_test", 5)
	RecordTenantCount("deleting_test", 6)
	RecordTenantCount("deleted_test", 7)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	for _, want := range []string{
		`drive9_tenant_count{status="pending_test"} 1.000000`,
		`drive9_tenant_count{status="provisioning_test"} 2.000000`,
		`drive9_tenant_count{status="active_test"} 3.000000`,
		`drive9_tenant_count{status="failed_test"} 4.000000`,
		`drive9_tenant_count{status="suspended_test"} 5.000000`,
		`drive9_tenant_count{status="deleting_test"} 6.000000`,
		`drive9_tenant_count{status="deleted_test"} 7.000000`,
	} {
		if !strings.Contains(text, want) {
			t.Errorf("missing tenant count metric %q:\n%s", want, text)
		}
	}
	if strings.Contains(text, `drive9_tenant_non_deleted_count`) || strings.Contains(text, `total_non_deleted`) {
		t.Fatalf("tenant count metrics must only use real status values:\n%s", text)
	}
}

func TestRecordTenantPoolMetadataResumeWaitIncludesPoolAndOrganizationID(t *testing.T) {
	const poolID = "pool-metadata-resume-metrics-test"
	const organizationID = "org-metadata-resume-metrics-test"

	RecordTenantPoolMetadataResumeWait(poolID, organizationID, "group", "ok", 10*time.Minute)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_pool_metadata_resume_wait_total{organization_id="`+organizationID+`",pool_id="`+poolID+`",result="ok",scope="group"} 1`) {
		t.Fatalf("missing tenant pool metadata resume wait total with pool_id and organization_id:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_metadata_resume_wait_duration_seconds_count{organization_id="`+organizationID+`",pool_id="`+poolID+`",result="ok",scope="group"} 1`) {
		t.Fatalf("missing tenant pool metadata resume wait duration with pool_id and organization_id:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_metadata_resume_wait_duration_seconds_bucket{organization_id="`+organizationID+`",pool_id="`+poolID+`",result="ok",scope="group",le="480"} 0`) {
		t.Fatalf("missing 480s tenant pool metadata resume wait duration bucket:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_metadata_resume_wait_duration_seconds_bucket{organization_id="`+organizationID+`",pool_id="`+poolID+`",result="ok",scope="group",le="600"} 1`) {
		t.Fatalf("missing 600s tenant pool metadata resume wait duration bucket:\n%s", text)
	}
}

func TestRecordHTTPRequestIncludesLongWriteLatencyBuckets(t *testing.T) {
	const route = "/v1/fs/http-long-bucket-test"

	RecordHTTPRequest("POST", route, 200, 25*time.Second)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_http_request_duration_seconds_bucket{method="POST",route="`+route+`",le="20"} 0`) {
		t.Fatalf("missing 20s HTTP duration bucket:\n%s", text)
	}
	if !strings.Contains(text, `drive9_http_request_duration_seconds_bucket{method="POST",route="`+route+`",le="30"} 1`) {
		t.Fatalf("missing 30s HTTP duration bucket:\n%s", text)
	}
}

func TestRecordOperationIncludesLongAdminTenantPoolUpdateBuckets(t *testing.T) {
	RecordOperation("admin_tenant_pool", "update", "ok", 5*time.Minute)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operation_duration_seconds_bucket{component="admin_tenant_pool",operation="update",result="ok",le="300"} 1`) {
		t.Fatalf("missing 300s service operation duration bucket:\n%s", text)
	}
	if !strings.Contains(text, `drive9_service_operation_duration_seconds_bucket{component="admin_tenant_pool",operation="update",result="ok",le="600"} 1`) {
		t.Fatalf("missing 600s service operation duration bucket:\n%s", text)
	}
}

func TestRecordTenantPoolCapacityIncludesPoolOrganizationAndState(t *testing.T) {
	const poolID = "pool-capacity-metrics-test"
	const organizationID = "org-capacity-metrics-test"

	RecordTenantPoolCapacity(poolID, organizationID, "size", 10)
	RecordTenantPoolCapacity(poolID, organizationID, "free", 1)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_pool_capacity{organization_id="`+organizationID+`",pool_id="`+poolID+`",state="size"} 10`) {
		t.Fatalf("missing tenant pool size capacity gauge:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_capacity{organization_id="`+organizationID+`",pool_id="`+poolID+`",state="free"} 1`) {
		t.Fatalf("missing tenant pool free capacity gauge:\n%s", text)
	}
}

func TestRecordTenantPoolBindingsIncludesPoolTiDBCloudOrgAndStatus(t *testing.T) {
	const poolID = "pool-binding-metrics-test"
	const orgID = "org-binding-metrics-test"

	RecordTenantPoolBindings(poolID, orgID, "free", 3)
	RecordTenantPoolBindings(poolID, orgID, "used", 2)
	RecordTenantPoolBindings(poolID+"-guest", "", "free", 1)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_tenant_pool_bindings{pool_id="`+poolID+`",status="free",tidbcloud_org_id="`+orgID+`"} 3`) {
		t.Fatalf("missing tenant pool free binding gauge:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_bindings{pool_id="`+poolID+`",status="used",tidbcloud_org_id="`+orgID+`"} 2`) {
		t.Fatalf("missing tenant pool used binding gauge:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_bindings{pool_id="`+poolID+`-guest",status="free",tidbcloud_org_id="guest"} 1`) {
		t.Fatalf("missing tenant pool guest org binding gauge:\n%s", text)
	}
	if strings.Contains(text, `pool_status=`) {
		t.Fatalf("tenant pool binding metrics must use status label, not pool_status:\n%s", text)
	}

	DeleteTenantPoolBindings(poolID, orgID, "free")

	rec = httptest.NewRecorder()
	WritePrometheus(rec)
	text = rec.Body.String()
	if strings.Contains(text, `drive9_tenant_pool_bindings{pool_id="`+poolID+`",status="free",tidbcloud_org_id="`+orgID+`"}`) {
		t.Fatalf("tenant pool free binding gauge was not deleted:\n%s", text)
	}
	if !strings.Contains(text, `drive9_tenant_pool_bindings{pool_id="`+poolID+`",status="used",tidbcloud_org_id="`+orgID+`"} 2`) {
		t.Fatalf("tenant pool used binding gauge should remain after deleting free gauge:\n%s", text)
	}
}

func TestTiDBCloudQuotaMetricsOmitTenantID(t *testing.T) {
	RecordTiDBCloudRBACCacheRequest("quota_get", "cluster", "hit")
	RecordTiDBCloudBillingCacheRequest("provision_cache_test", "miss")
	RecordTiDBCloudOpenAPIRequest("cluster", "list_clusters", "ok")
	RecordTiDBCloudOpenAPIRequestDuration("cluster", "list_clusters", 125*time.Millisecond)
	RecordTiDBCloudOpenAPIRequest("iam", "user_authn", "denied")
	RecordTiDBCloudOpenAPIRequestDuration("iam", "user_authn", 25*time.Millisecond)
	RecordTiDBCloudInnerAuthCacheRequest("api_key_authn", "hit")

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	for _, want := range []string{
		`drive9_tidbcloud_rbac_cache_requests_total{path="quota_get",result="hit",scope="cluster"} 1`,
		`drive9_tidbcloud_billing_cache_requests_total{path="provision_cache_test",result="miss"} 1`,
		`drive9_tidbcloud_openapi_requests_total{api="cluster",operation="list_clusters",result="ok"} 1`,
		`drive9_tidbcloud_openapi_request_duration_seconds_count{api="cluster",operation="list_clusters"} 1`,
		`drive9_tidbcloud_openapi_requests_total{api="iam",operation="user_authn",result="denied"} 1`,
		`drive9_tidbcloud_openapi_request_duration_seconds_count{api="iam",operation="user_authn"} 1`,
		`drive9_tidbcloud_inner_auth_cache_requests_total{operation="api_key_authn",result="hit"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("missing TiDB Cloud quota metric %q:\n%s", want, text)
		}
	}
	for _, line := range strings.Split(text, "\n") {
		if strings.HasPrefix(line, "drive9_tidbcloud_") && strings.Contains(line, `tenant_id=`) {
			t.Fatalf("TiDB Cloud quota metrics must not carry tenant_id:\n%s", text)
		}
		if strings.HasPrefix(line, "drive9_tidbcloud_openapi_requests_total") && strings.Contains(line, `path=`) {
			t.Fatalf("TiDB Cloud OpenAPI metrics must use api, not path:\n%s", text)
		}
	}
}

func TestRecordDBOperationOmitsTenantID(t *testing.T) {
	const tenantID = "tenant-db-operation-test"

	RecordDBOperation("user", "query", "ok", 0)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_db_operations_total{operation="query",result="ok",role="user"} 1`) {
		t.Fatalf("missing db operation total:\n%s", text)
	}
	if strings.Contains(text, `tenant_id="`+tenantID+`"`) {
		t.Fatalf("db operation metrics must not carry tenant_id:\n%s", text)
	}
}

func TestDeleteTenantCounters(t *testing.T) {
	const tenantID = "tenant-delete-counter-test"
	const tidbCloudOrgID = "org-delete-counter"

	RecordTenantOperationCountWithOrg(tenantID, tidbCloudOrgID, "cmp", "op", "ok")
	RecordTenantRequestCountWithOrg(tenantID, tidbCloudOrgID, "api", "read", 200)
	RecordTenantFileBytesWithOrg(tenantID, tidbCloudOrgID, "read", "db9", 10)
	RecordTenantFileOpsWithOrg(tenantID, tidbCloudOrgID, "write", "db9", 1)
	RecordSSEConnectionWithOrg(tenantID, tidbCloudOrgID, "mount", time.Second)
	RecordSSEEventSentWithOrg(tenantID, tidbCloudOrgID, "write")
	RecordTenantInFlightWithOrg(tenantID, tidbCloudOrgID, "api", 3)

	rec := httptest.NewRecorder()
	WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `tenant_id="`+tenantID+`"`) {
		t.Fatalf("tenant_id not found before delete:\n%s", text)
	}

	DeleteTenantCounters(tenantID)

	rec2 := httptest.NewRecorder()
	WritePrometheus(rec2)
	text2 := rec2.Body.String()

	counterNames := []string{
		"drive9_service_operations_total",
		"drive9_tenant_requests_total",
		"drive9_tenant_file_bytes_total",
		"drive9_tenant_file_ops_total",
		"drive9_sse_connections_total",
		"drive9_sse_events_sent_total",
	}
	histogramNames := []string{
		"drive9_sse_connection_duration_seconds_bucket",
		"drive9_sse_connection_duration_seconds_count",
		"drive9_sse_connection_duration_seconds_sum",
	}
	gaugeNames := []string{
		"drive9_tenant_inflight_requests",
	}
	var failures int
	for _, line := range strings.Split(text2, "\n") {
		if !strings.Contains(line, `tenant_id="`+tenantID+`"`) {
			continue
		}
		for _, name := range counterNames {
			if strings.HasPrefix(line, name) {
				t.Errorf("counter line with tenant_id still present: %s", line)
				failures++
			}
		}
		for _, name := range histogramNames {
			if strings.HasPrefix(line, name) {
				t.Errorf("histogram line with tenant_id still present: %s", line)
				failures++
			}
		}
		for _, name := range gaugeNames {
			if strings.HasPrefix(line, name) {
				t.Errorf("gauge line with tenant_id still present: %s", line)
				failures++
			}
		}
	}
	if failures > 0 {
		t.Fatalf("%d tenant-scoped lines leaked after DeleteTenantCounters", failures)
	}
}

func TestDeleteTenantCountersEmptyIsNoop(t *testing.T) {
	DeleteTenantCounters("")
}