// Package metrics provides process-wide metrics with an OpenTelemetry-like
// meter/instrument/attribute model and a Prometheus text exporter.
package metrics

import (
	"context"
	"database/sql/driver"
	"errors"
	"net/http"
	"strconv"
	"strings"
	"time"
)

var operationDurationBounds = []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600}
var httpDurationBounds = []float64{0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, 15, 20, 30, 60}
var tenantPoolMetadataResumeWaitDurationBounds = []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 240, 480, 600, 900, 1200}

const tenantMetricGuestTiDBCloudOrgID = "guest"

// sseConnectionDurationBounds covers SSE connection lifetimes, which can
// range from sub-second probes to long-lived mounts. Buckets extend well
// beyond httpDurationBounds so long-lived connections are visible without
// piling into +Inf. These are recorded into a dedicated histogram
// (drive9_sse_connection_duration_seconds), NOT drive9_http_request_duration_seconds.
var sseConnectionDurationBounds = []float64{0.005, 0.05, 0.1, 0.5, 1, 5, 10, 30, 60, 300, 600, 1800, 3600}

// eventBusQueryDurationBounds covers fs_events DB query latencies. These
// should normally be low (PK range scan); a shift right indicates DB
// pressure or table growth.
var eventBusQueryDurationBounds = []float64{0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}

// ssePhase1DurationBounds covers the Phase-1 replay/reset cost of a new SSE
// connection (one EventsSince call + stream of buffered events). Used to
// distinguish slow first-replay from slow Phase-2 live streaming.
var ssePhase1DurationBounds = []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}

var globalRegistry = NewRegistry()
var globalMeterProvider = NewMeterProvider(globalRegistry)

var serviceMeter = globalMeterProvider.Meter("service")
var httpMeter = globalMeterProvider.Meter("http")
var eventMeter = globalMeterProvider.Meter("event")
var sseMeter = globalMeterProvider.Meter("sse")
var tenantMeter = globalMeterProvider.Meter("tenant")

var serviceOperationsTotal = serviceMeter.Int64Counter("drive9_service_operations_total", "Service operations by component/operation/result")
var tenantOperationFailuresTotal = serviceMeter.Int64Counter("drive9_tenant_operation_failures_total", "Tenant-scoped service operation failures by component/operation/result/tenant/tidbcloud_org")
var serviceOperationDuration = serviceMeter.Float64Histogram("drive9_service_operation_duration_seconds", "Service operation duration histogram by component/operation/result", operationDurationBounds)
var deleteReconcileOrphansFoundTotal = serviceMeter.Int64Counter("drive9_delete_reconcile_orphans_found_total", "Delete-orphan reconciliation candidates found per round by type/mode; dry-run rounds recount the same still-present candidates each round")
var serviceGauge = serviceMeter.Float64Gauge("drive9_service_gauge", "Service gauges by component/name")
var featureEnabled = serviceMeter.Float64Gauge("drive9_feature_enabled", "Configured Drive9 feature state")
var tenantPoolMetadataResumeWaitTotal = serviceMeter.Int64Counter("drive9_tenant_pool_metadata_resume_wait_total", "Tenant pool metadata resume wait attempts by pool_id/organization_id/scope/result")
var tenantPoolMetadataResumeWaitDuration = serviceMeter.Float64Histogram("drive9_tenant_pool_metadata_resume_wait_duration_seconds", "Tenant pool metadata resume wait duration by pool_id/organization_id/scope/result", tenantPoolMetadataResumeWaitDurationBounds)
var tenantPoolCapacity = serviceMeter.Float64Gauge("drive9_tenant_pool_capacity", "Tenant pool capacity by pool_id/organization_id/state")
var tenantPoolBindings = serviceMeter.Float64Gauge("drive9_tenant_pool_bindings", "Tenant pool binding count by pool_id/tidbcloud_org_id/status")
var sharedDBPoolTotal = serviceMeter.Float64Gauge("drive9_shared_db_pool_total", "Physical shared DB-pool presence by tidbcloud_org_id/db_pool_id/db_pool_uuid/status")
var sharedDBPoolCapacity = serviceMeter.Float64Gauge("drive9_shared_db_pool_capacity", "Physical shared DB-pool capacity by tidbcloud_org_id/db_pool_id/db_pool_uuid/type")
var sharedDBPoolTenants = serviceMeter.Float64Gauge("drive9_shared_db_pool_tenants", "Physical shared DB-pool tenant count by tidbcloud_org_id/db_pool_id/db_pool_uuid/state")
var sharedDBPoolSpendingLimit = serviceMeter.Float64Gauge("drive9_shared_db_pool_spending_limit", "Physical shared DB-pool spending limit by tidbcloud_org_id/db_pool_id/db_pool_uuid/type")
var sharedDBPoolCacheHandles = serviceMeter.Float64Gauge("drive9_shared_db_pool_cache_handles", "Per-pod cached physical shared DB handles by tidbcloud_org_id/db_pool_id/db_pool_uuid")
var sharedDBPoolCacheTenants = serviceMeter.Float64Gauge("drive9_shared_db_pool_cache_tenants", "Per-pod active tenant backend refs on a cached shared DB handle by tidbcloud_org_id/db_pool_id/db_pool_uuid")
var tidbCloudRBACCacheRequestsTotal = serviceMeter.Int64Counter("drive9_tidbcloud_rbac_cache_requests_total", "TiDB Cloud API key to cluster RBAC cache requests by path/scope/result")
var tidbCloudBillingCacheRequestsTotal = serviceMeter.Int64Counter("drive9_tidbcloud_billing_cache_requests_total", "TiDB Cloud non-free Billing plan cache requests by path/result")
var tidbCloudInnerAuthCacheRequestsTotal = serviceMeter.Int64Counter("drive9_tidbcloud_inner_auth_cache_requests_total", "TiDB Cloud inner authentication cache requests by operation/result")
var tidbCloudOpenAPIRequestsTotal = serviceMeter.Int64Counter("drive9_tidbcloud_openapi_requests_total", "TiDB Cloud API requests by api/operation/result")
var tidbCloudOpenAPIRequestDurationBounds = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}
var tidbCloudOpenAPIRequestDuration = serviceMeter.Float64Histogram("drive9_tidbcloud_openapi_request_duration_seconds", "TiDB Cloud API logical request duration by api/operation", tidbCloudOpenAPIRequestDurationBounds)

var httpRequestsTotal = httpMeter.Int64Counter("drive9_http_requests_total", "Total HTTP requests by method/route/status")
var httpRequestDuration = httpMeter.Float64Histogram("drive9_http_request_duration_seconds", "HTTP request duration histogram by method/route", httpDurationBounds)
var httpRequestBodyReadDuration = httpMeter.Float64Histogram("drive9_http_request_body_read_duration_seconds", "HTTP request body read duration histogram by method/route/status_class/body_size_bucket", httpDurationBounds)
var httpInflight = httpMeter.Float64Gauge("drive9_http_inflight_requests", "Current in-flight HTTP requests")

var businessEventsTotal = eventMeter.Int64Counter("drive9_business_events_total", "Business lifecycle events")

var tenantRequestsTotal = tenantMeter.Int64Counter("drive9_tenant_requests_total", "Organization-scoped requests by tidbcloud_org/surface/action/status_class, with tenant attribution on data-plane requests and errors")
var tenantRequestDuration = tenantMeter.Float64Histogram("drive9_tenant_request_duration_seconds", "Tenant request duration histogram by surface/status_class", httpDurationBounds)
var tenantInflight = tenantMeter.Float64Gauge("drive9_tenant_inflight_requests", "Current in-flight tenant-scoped requests by tenant/tidbcloud_org/surface")
var tenantFileBytes = tenantMeter.Int64Counter("drive9_tenant_file_bytes_total", "Tenant-scoped logical file bytes by tenant/tidbcloud_org/direction/traffic_path")
var tenantFileOps = tenantMeter.Int64Counter("drive9_tenant_file_ops_total", "Tenant-scoped successful file read/write operations by tenant/tidbcloud_org/direction/traffic_path; billing unit for the Plan B file read/write ops SKUs")
var tenantCount = tenantMeter.Float64Gauge("drive9_tenant_count", "Tenant count by status")
var organizationStorageUsageBytes = tenantMeter.Float64Gauge("drive9_organization_storage_usage_bytes", "Total confirmed and reserved storage bytes by TiDB Cloud organization across all non-deleted tenants")
var organizationStorageConfirmedBytes = tenantMeter.Float64Gauge("drive9_organization_storage_confirmed_bytes", "Confirmed storage bytes by TiDB Cloud organization across all non-deleted tenants")
var organizationStorageReservedBytes = tenantMeter.Float64Gauge("drive9_organization_storage_reserved_bytes", "Reserved storage bytes by TiDB Cloud organization across all non-deleted tenants")
var organizationStorageLimitBytes = tenantMeter.Float64Gauge("drive9_organization_storage_limit_bytes", "Configured storage limit bytes by TiDB Cloud organization across all non-deleted tenants")
var organizationTenantCount = tenantMeter.Float64Gauge("drive9_organization_tenant_count", "Non-deleted tenant count by TiDB Cloud organization, independent of tenant pool bindings")
var organizationFileCount = tenantMeter.Float64Gauge("drive9_organization_file_count", "Confirmed file count by TiDB Cloud organization across all non-deleted tenants")
var tenantStorageBytes = tenantMeter.Float64Gauge("drive9_tenant_storage_bytes", "Tenant storage bytes by tenant/tidbcloud_org/state/storage_type")
var tenantFileCount = tenantMeter.Float64Gauge("drive9_tenant_file_count", "Tenant confirmed file count by tenant/tidbcloud_org/state/storage_type")

// SSE-specific instruments. SSE connection lifetimes are recorded here, not
// in drive9_http_request_duration_seconds (which would pollute HTTP latency
// alerts — SSE connections live as long as the client stays subscribed).
var sseConnectionsTotal = sseMeter.Int64Counter("drive9_sse_connections_total", "SSE /v1/events connections opened by tenant_id/tidbcloud_org_id/reason")
var sseConnectionDuration = sseMeter.Float64Histogram("drive9_sse_connection_duration_seconds", "SSE connection lifetime by tenant_id/tidbcloud_org_id (client stay-open duration, not request processing time)", sseConnectionDurationBounds)
var sseInflight = sseMeter.Float64Gauge("drive9_sse_inflight_connections", "Active SSE /v1/events connections by tenant_id/tidbcloud_org_id")
var ssePhase1Duration = sseMeter.Float64Histogram("drive9_sse_phase1_duration_seconds", "SSE Phase-1 replay/reset duration by tenant_id/tidbcloud_org_id (one EventsSince call + buffered stream)", ssePhase1DurationBounds)
var sseEventsSentTotal = sseMeter.Int64Counter("drive9_sse_events_sent_total", "SSE events sent to clients by type/tenant_id/tidbcloud_org_id")
var sseResetsSentTotal = sseMeter.Int64Counter("drive9_sse_resets_sent_total", "SSE reset events sent by reason/tenant_id/tidbcloud_org_id")
var sseHeartbeatsSentTotal = sseMeter.Int64Counter("drive9_sse_heartbeats_sent_total", "SSE heartbeat events sent by tenant_id/tidbcloud_org_id")

// Event-bus query instruments. Covers all fs_events DB reads (events_since,
// poll, latest, oldest) so DB pressure and table growth on the events path
// are observable without direct DB access.
var eventBusQueryDuration = serviceMeter.Float64Histogram("drive9_event_bus_query_duration_seconds", "Event-bus fs_events query duration by operation/result", eventBusQueryDurationBounds)

// fs_events table instruments. Compensates for the lack of direct TiDB
// access: row count and prune volume are reported by the server itself.
var fsEventsRows = sseMeter.Float64Gauge("drive9_fs_events_rows", "fs_events table row count by tenant_id/tidbcloud_org_id (capped at 100000)")
var fsEventsPrunedTotal = sseMeter.Int64Counter("drive9_fs_events_pruned_total", "fs_events rows pruned by retention cleanup")

// File GC backlog instruments. Sparse: only tenants with claimable > 0 keep a
// series (zero deletes the labels). Updated on worker paths that already hold
// the tenant store — not via a full-tenant scan. drive9_file_gc_claimable is
// remaining depth after a drain that could claim work; it is omitted when
// Claim returns empty even if COUNT still sees due rows (mismatch).
var fileGCClaimable = serviceMeter.Float64Gauge("drive9_file_gc_claimable", "Claimable file_gc_tasks remaining after a drain that could claim work, by tenant_id/tidbcloud_org_id; series omitted when zero or when claim returns empty")
var fileGCOldestClaimableAgeSeconds = serviceMeter.Float64Gauge("drive9_file_gc_oldest_claimable_age_seconds", "Age in seconds of the oldest claimable file_gc task by tenant_id/tidbcloud_org_id; series omitted when claimable is zero")
var fileGCDrainHitCap = serviceMeter.Float64Gauge("drive9_file_gc_drain_hit_cap", "1 when the last FileGC drain for this tenant filled its batch (more work likely remains); series omitted otherwise")
var fileGCClaimStalled = serviceMeter.Float64Gauge("drive9_file_gc_claim_stalled", "1 when this shard owner exhausted FileGC mismatch re-kicks while COUNT still sees due rows (SKIP LOCKED stall); series omitted otherwise")

// Event retry-buffer instruments (see pkg/server/event_retry.go). The dropped
// counter is the hard-loss SLI for the buffer: entries evicted by the
// per-tenant or global cap, or expired before a durable insert.
var sseEventRetryDroppedTotal = sseMeter.Int64Counter("drive9_sse_event_retry_dropped_total", "SSE event retry-buffer entries dropped before durable insert by tenant_id/tidbcloud_org_id/reason")
var sseEventRetryBufferDepth = sseMeter.Float64Gauge("drive9_sse_event_retry_buffer_depth", "Current number of buffered SSE events awaiting retry insert")

func RegisterModule(module string) {
	globalRegistry.RegisterModule(module)
}

func SetModuleAvailability(module string, up bool) {
	globalRegistry.SetModuleAvailability(module, up)
}

func SetFeatureEnabled(feature string, enabled bool) {
	value := float64(0)
	if enabled {
		value = 1
	}
	featureEnabled.Set(value, Attr("feature", cleanMetricValue(feature, "unknown")))
}

func RecordOperation(component, operation, result string, d time.Duration) {
	RecordTenantOperation("", component, operation, result, d)
}

// RecordDeleteReconcileOrphansFound counts delete-orphan reconciliation
// candidates found in one round. typ is "dentry" or "inode"; mode is
// "dry_run" or "repair". A sustained non-zero rate in repair mode is the
// designated signal to prioritize the delete-in-progress marker (design doc
// recursive-delete-batched-design.md §4.2.1). Dry-run audits recount the
// same still-present candidates every round, so the counter is a per-round
// audit volume, not unique candidates.
func RecordDeleteReconcileOrphansFound(typ, mode string, count int64) {
	if count <= 0 {
		return
	}
	RegisterModule("datastore")
	deleteReconcileOrphansFoundTotal.Add(count,
		Attr("type", cleanMetricValue(typ, "unknown")),
		Attr("mode", cleanMetricValue(mode, "unknown")))
}

// RecordOperationCount records count occurrences of a service operation
// without a duration sample (batched variant of RecordOperation for
// discard/flush-style counters).
func RecordOperationCount(component, operation, result string, count int64) {
	if count <= 0 {
		return
	}
	component = cleanMetricValue(component, "unknown")
	operation = cleanMetricValue(operation, "unknown")
	result = cleanMetricValue(result, "unknown")
	RegisterModule(component)
	serviceOperationsTotal.Add(count,
		Attr("component", component),
		Attr("operation", operation),
		Attr("result", result),
	)
}

func RecordTenantOperation(tenantID, component, operation, result string, d time.Duration) {
	RecordTenantOperationWithOrg(tenantID, "", component, operation, result, d)
}

func RecordTenantOperationWithOrg(tenantID, tidbCloudOrgID, component, operation, result string, d time.Duration) {
	component = cleanMetricValue(component, "unknown")
	operation = cleanMetricValue(operation, "unknown")
	result = cleanMetricValue(result, "unknown")
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	RegisterModule(component)
	attrs := []Attribute{
		Attr("component", component),
		Attr("operation", operation),
		Attr("result", result),
	}
	serviceOperationsTotal.Add(1, attrs...)
	recordTenantOperationFailure(tenantID, tidbCloudOrgID, result, attrs)
	if d <= 0 {
		return
	}
	serviceOperationDuration.Observe(d.Seconds(), attrs...)
}

func RecordTenantPoolMetadataResumeWait(poolID, organizationID, scope, result string, d time.Duration) {
	poolID = cleanMetricValue(poolID, "unknown")
	organizationID = cleanMetricValue(organizationID, "unknown")
	scope = cleanMetricValue(scope, "unknown")
	result = cleanMetricValue(result, "unknown")
	attrs := []Attribute{
		Attr("pool_id", poolID),
		Attr("organization_id", organizationID),
		Attr("scope", scope),
		Attr("result", result),
	}
	tenantPoolMetadataResumeWaitTotal.Add(1, attrs...)
	tenantPoolMetadataResumeWaitDuration.Observe(d.Seconds(), attrs...)
}

func RecordTenantPoolCapacity(poolID, organizationID, state string, value float64) {
	RegisterModule("admin_tenant_pool")
	tenantPoolCapacity.Set(value,
		Attr("pool_id", cleanMetricValue(poolID, "unknown")),
		Attr("organization_id", cleanMetricValue(organizationID, "unknown")),
		Attr("state", cleanMetricValue(state, "unknown")),
	)
}

func RecordTenantPoolBindings(poolID, tidbCloudOrgID, poolStatus string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("admin_tenant_pool")
	tenantPoolBindings.Set(float64(count),
		Attr("pool_id", cleanMetricValue(poolID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("status", cleanMetricValue(poolStatus, "unknown")),
	)
}

func DeleteTenantPoolBindings(poolID, tidbCloudOrgID, poolStatus string) {
	tenantPoolBindings.Delete(
		Attr("pool_id", cleanMetricValue(poolID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("status", cleanMetricValue(poolStatus, "unknown")),
	)
}

func RecordSharedDBPoolTotal(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, status string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("shared_db_pool")
	sharedDBPoolTotal.Set(float64(count), sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "status", status)...)
}

func DeleteSharedDBPoolTotal(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, status string) {
	sharedDBPoolTotal.Delete(sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "status", status)...)
}

func RecordSharedDBPoolCapacity(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, capacityType string, value int64) {
	RegisterModule("shared_db_pool")
	sharedDBPoolCapacity.Set(float64(value), sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "type", capacityType)...)
}

func DeleteSharedDBPoolCapacity(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, capacityType string) {
	sharedDBPoolCapacity.Delete(sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "type", capacityType)...)
}

func RecordSharedDBPoolTenants(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, state string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("shared_db_pool")
	sharedDBPoolTenants.Set(float64(count), sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "state", state)...)
}

func DeleteSharedDBPoolTenants(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, state string) {
	sharedDBPoolTenants.Delete(sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "state", state)...)
}

func RecordSharedDBPoolSpendingLimit(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, limitType string, value int64) {
	RegisterModule("shared_db_pool")
	sharedDBPoolSpendingLimit.Set(float64(value), sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "type", limitType)...)
}

func DeleteSharedDBPoolSpendingLimit(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, limitType string) {
	sharedDBPoolSpendingLimit.Delete(sharedDBPoolAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID, "type", limitType)...)
}

func sharedDBPoolAttrs(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID, dimension, value string) []Attribute {
	return []Attribute{
		Attr("tidbcloud_org_id", cleanMetricValue(tidbCloudOrgID, "unknown")),
		Attr("db_pool_id", strconv.FormatInt(dbPoolID, 10)),
		Attr("db_pool_uuid", cleanMetricValue(dbPoolUUID, "unknown")),
		Attr(dimension, cleanMetricValue(value, "unknown")),
	}
}

func RecordSharedDBPoolCacheHandles(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("shared_db_pool")
	sharedDBPoolCacheHandles.Set(float64(count), sharedDBPoolCacheAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID)...)
}

func DeleteSharedDBPoolCacheHandles(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID string) {
	sharedDBPoolCacheHandles.Delete(sharedDBPoolCacheAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID)...)
}

func RecordSharedDBPoolCacheTenants(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("shared_db_pool")
	sharedDBPoolCacheTenants.Set(float64(count), sharedDBPoolCacheAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID)...)
}

func DeleteSharedDBPoolCacheTenants(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID string) {
	sharedDBPoolCacheTenants.Delete(sharedDBPoolCacheAttrs(tidbCloudOrgID, dbPoolID, dbPoolUUID)...)
}

func sharedDBPoolCacheAttrs(tidbCloudOrgID string, dbPoolID int64, dbPoolUUID string) []Attribute {
	return []Attribute{
		Attr("tidbcloud_org_id", cleanMetricValue(tidbCloudOrgID, "unknown")),
		Attr("db_pool_id", strconv.FormatInt(dbPoolID, 10)),
		Attr("db_pool_uuid", cleanMetricValue(dbPoolUUID, "unknown")),
	}
}

func DeleteTenantCounters(tenantID string) {
	tenantID = cleanMetricValue(tenantID, "unknown")
	if tenantID == "unknown" {
		return
	}
	globalRegistry.DeleteTenantSeries(tenantID)
}

func RecordOrganizationStorageUsage(tidbCloudOrgID string, bytes int64) {
	if bytes < 0 {
		return
	}
	RegisterModule("tenant_usage")
	organizationStorageUsageBytes.Set(float64(bytes),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	)
}

// OrganizationUsageSnapshot is the fleet observer's aggregated usage snapshot
// for one organization. Passing a struct keeps the record call self-describing;
// pkg/metrics cannot import pkg/meta types.
type OrganizationUsageSnapshot struct {
	ConfirmedBytes    int64
	ReservedBytes     int64
	StorageLimitBytes int64
	TenantCount       int64
	FileCount         int64
}

func RecordOrganizationUsageSnapshot(tidbCloudOrgID string, snapshot OrganizationUsageSnapshot) {
	if snapshot.ConfirmedBytes < 0 || snapshot.ReservedBytes < 0 || snapshot.StorageLimitBytes < 0 ||
		snapshot.TenantCount < 0 || snapshot.FileCount < 0 {
		return
	}
	orgID := cleanTiDBCloudOrgID(tidbCloudOrgID)
	RegisterModule("tenant_usage")
	organizationStorageUsageBytes.Set(float64(snapshot.ConfirmedBytes+snapshot.ReservedBytes), Attr("tidbcloud_org_id", orgID))
	organizationStorageConfirmedBytes.Set(float64(snapshot.ConfirmedBytes), Attr("tidbcloud_org_id", orgID))
	organizationStorageReservedBytes.Set(float64(snapshot.ReservedBytes), Attr("tidbcloud_org_id", orgID))
	organizationStorageLimitBytes.Set(float64(snapshot.StorageLimitBytes), Attr("tidbcloud_org_id", orgID))
	organizationTenantCount.Set(float64(snapshot.TenantCount), Attr("tidbcloud_org_id", orgID))
	organizationFileCount.Set(float64(snapshot.FileCount), Attr("tidbcloud_org_id", orgID))
}

func DeleteOrganizationStorageUsage(tidbCloudOrgID string) {
	orgAttr := Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID))
	organizationStorageUsageBytes.Delete(orgAttr)
	organizationStorageConfirmedBytes.Delete(orgAttr)
	organizationStorageReservedBytes.Delete(orgAttr)
	organizationStorageLimitBytes.Delete(orgAttr)
	organizationTenantCount.Delete(orgAttr)
	organizationFileCount.Delete(orgAttr)
}

func RecordTiDBCloudRBACCacheRequest(path, scope, result string) {
	RegisterModule("tidbcloud_quota")
	tidbCloudRBACCacheRequestsTotal.Add(1,
		Attr("path", cleanMetricValue(path, "unknown")),
		Attr("scope", cleanMetricValue(scope, "unknown")),
		Attr("result", cleanMetricValue(result, "unknown")),
	)
}

func RecordTiDBCloudBillingCacheRequest(path, result string) {
	RegisterModule("tidbcloud_billing")
	tidbCloudBillingCacheRequestsTotal.Add(1,
		Attr("path", cleanMetricValue(path, "unknown")),
		Attr("result", cleanMetricValue(result, "unknown")),
	)
}

func RecordTiDBCloudOpenAPIRequest(api, operation, result string) {
	RegisterModule("tidbcloud_openapi")
	tidbCloudOpenAPIRequestsTotal.Add(1,
		Attr("api", cleanMetricValue(api, "unknown")),
		Attr("operation", cleanMetricValue(operation, "unknown")),
		Attr("result", cleanMetricValue(result, "unknown")),
	)
}

// RecordTiDBCloudOpenAPIRequestDuration records one logical TiDB Cloud API request.
// OpenAPI durations include a possible digest-auth challenge/retry round trip.
func RecordTiDBCloudOpenAPIRequestDuration(api, operation string, d time.Duration) {
	RegisterModule("tidbcloud_openapi")
	if d <= 0 {
		return
	}
	tidbCloudOpenAPIRequestDuration.Observe(d.Seconds(),
		Attr("api", cleanMetricValue(api, "unknown")),
		Attr("operation", cleanMetricValue(operation, "unknown")),
	)
}

func RecordTiDBCloudInnerAuthCacheRequest(operation, result string) {
	RegisterModule("tidbcloud_openapi")
	tidbCloudInnerAuthCacheRequestsTotal.Add(1,
		Attr("operation", cleanMetricValue(operation, "unknown")),
		Attr("result", cleanMetricValue(result, "unknown")),
	)
}

// ResultForError returns a stable metric result label for common infrastructure
// errors. Callers should use this when recording generic worker/DB failures so
// transient bad connections do not get bucketed with semantic operation errors.
// Keep mysqlutil.dbResult delegated here so DB and worker labels stay aligned.
func ResultForError(err error) string {
	switch {
	case err == nil:
		return "ok"
	case errors.Is(err, context.Canceled):
		return "canceled"
	case errors.Is(err, context.DeadlineExceeded):
		return "deadline_exceeded"
	case errors.Is(err, driver.ErrBadConn):
		return "bad_conn"
	case strings.Contains(strings.ToLower(err.Error()), "invalid connection"):
		return "bad_conn"
	default:
		return "error"
	}
}

// RecordTenantOperationCount records an operation outcome without adding a
// duration sample. Use it for per-entry counters when one wall-clock duration
// would be duplicated across many logical items in the same batch.
func RecordTenantOperationCount(tenantID, component, operation, result string) {
	RecordTenantOperationCountWithOrg(tenantID, "", component, operation, result)
}

func RecordTenantOperationCountWithOrg(tenantID, tidbCloudOrgID, component, operation, result string) {
	component = cleanMetricValue(component, "unknown")
	operation = cleanMetricValue(operation, "unknown")
	result = cleanMetricValue(result, "unknown")
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	RegisterModule(component)
	attrs := []Attribute{
		Attr("component", component),
		Attr("operation", operation),
		Attr("result", result),
	}
	serviceOperationsTotal.Add(1, attrs...)
	recordTenantOperationFailure(tenantID, tidbCloudOrgID, result, attrs)
}

func recordTenantOperationFailure(tenantID, tidbCloudOrgID, result string, attrs []Attribute) {
	if tenantID == "unknown" || !isOperationFailureResult(result) {
		return
	}
	failureAttrs := append([]Attribute(nil), attrs...)
	failureAttrs = append(failureAttrs,
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
	)
	tenantOperationFailuresTotal.Add(1, failureAttrs...)
}

func isOperationFailureResult(result string) bool {
	switch result {
	case "error", "bad_conn", "deadline_exceeded", "failed",
		"fail_open", "busy_fail_open",
		"runtime_not_configured", "handler_missing", "unsupported",
		"lease_lost", "dead_lettered",
		// Worker-loop / processKicked panics: rare, must hit the tenant failure
		// series for alerting without adding success-path tenant labels.
		"panic":
		return true
	default:
		return strings.HasSuffix(result, "_error")
	}
}

func tenantAttributedSurface(surface, statusClass string) bool {
	switch surface {
	case "fs", "upload", "object_store", "sql", "events", "journal", "git_workspace", "fork", "layer", "migration":
		return true
	default:
		return statusClass == "4xx" || statusClass == "5xx"
	}
}

func RecordGauge(component, name string, value float64) {
	RecordTenantGauge("", component, name, value)
}

func RecordTenantGauge(tenantID, component, name string, value float64) {
	RecordTenantGaugeWithOrg(tenantID, "", component, name, value)
}

func RecordTenantGaugeWithOrg(tenantID, tidbCloudOrgID, component, name string, value float64) {
	component = cleanMetricValue(component, "unknown")
	name = cleanMetricValue(name, "unknown")
	tenantID = strings.TrimSpace(tenantID)
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	RegisterModule(component)
	serviceGauge.Set(value,
		Attr("component", component),
		Attr("name", name),
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
	)
}

func DeleteTenantGauge(tenantID, component, name string) {
	DeleteTenantGaugeWithOrg(tenantID, "", component, name)
}

func DeleteTenantGaugeWithOrg(tenantID, tidbCloudOrgID, component, name string) {
	serviceGauge.Delete(
		Attr("component", cleanMetricValue(component, "unknown")),
		Attr("name", cleanMetricValue(name, "unknown")),
		Attr("tenant_id", strings.TrimSpace(tenantID)),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	)
}

func RecordHTTPRequest(method, route string, status int, d time.Duration) {
	RegisterModule("server")
	httpRequestsTotal.Add(1,
		Attr("method", cleanMetricValue(method, "UNKNOWN")),
		Attr("route", cleanMetricValue(route, "other")),
		Attr("status", strconv.Itoa(status)),
	)
	httpRequestDuration.Observe(d.Seconds(),
		Attr("method", cleanMetricValue(method, "UNKNOWN")),
		Attr("route", cleanMetricValue(route, "other")),
	)
}

func RecordHTTPRequestBodyRead(method, route string, status int, bodyBytes int64, d time.Duration) {
	RegisterModule("server")
	httpRequestBodyReadDuration.Observe(d.Seconds(),
		Attr("method", cleanMetricValue(method, "UNKNOWN")),
		Attr("route", cleanMetricValue(route, "other")),
		Attr("status_class", statusClass(status)),
		Attr("body_size_bucket", bodySizeBucket(bodyBytes)),
	)
}

// RecordHTTPRequestCount records only the request counter, not the duration
// histogram. Used for SSE/streaming routes whose handler blocks for the
// connection lifetime (which is not request processing time and would
// pollute HTTP latency alerts if recorded into the duration histogram).
func RecordHTTPRequestCount(method, route string, status int) {
	RegisterModule("server")
	httpRequestsTotal.Add(1,
		Attr("method", cleanMetricValue(method, "UNKNOWN")),
		Attr("route", cleanMetricValue(route, "other")),
		Attr("status", strconv.Itoa(status)),
	)
}

// RecordTenantRequestCount records only the request counter, not the
// duration histogram. Companion to RecordHTTPRequestCount for SSE routes.
func RecordTenantRequestCount(tenantID, surface, action string, status int) {
	RecordTenantRequestCountWithOrg(tenantID, "", surface, action, status)
}

func RecordTenantRequestCountWithOrg(tenantID, tidbCloudOrgID, surface, action string, status int) {
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	surface = cleanMetricValue(surface, "other")
	action = cleanMetricValue(action, "other")
	statusClass := statusClass(status)
	RegisterModule("tenant_usage")
	attrs := []Attribute{
		Attr("surface", surface),
		Attr("action", action),
		Attr("status_class", statusClass),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
	}
	if tenantAttributedSurface(surface, statusClass) {
		attrs = append(attrs, Attr("tenant_id", tenantID))
	}
	tenantRequestsTotal.Add(1, attrs...)
}

func RecordHTTPInFlight(value float64) {
	RegisterModule("server")
	httpInflight.Set(value)
}

func RecordHTTPInFlightRoute(route string, value float64) {
	RegisterModule("server")
	httpInflight.Set(value, Attr("route", cleanMetricValue(route, "other")))
}

func RecordEvent(event string, labels ...string) {
	RecordTenantEvent("", event, labels...)
}

func RecordTenantEvent(tenantID, event string, labels ...string) {
	RecordTenantEventWithOrg(tenantID, "", event, labels...)
}

func RecordTenantEventWithOrg(tenantID, tidbCloudOrgID, event string, labels ...string) {
	RegisterModule("server")
	attrs := make([]Attribute, 0, len(labels)/2+2)
	attrs = append(attrs, Attr("event", cleanMetricValue(event, "unknown")))
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	if tenantID != "unknown" && tenantAttributedBusinessEvent(event, metricLabelValue(labels, "result")) {
		attrs = append(attrs, Attr("tenant_id", tenantID), Attr("tidbcloud_org_id", tidbCloudOrgID))
	}
	for i := 0; i+1 < len(labels); i += 2 {
		attrs = append(attrs, Attr(labels[i], labels[i+1]))
	}
	businessEventsTotal.Add(1, attrs...)
}

func tenantAttributedBusinessEvent(event, result string) bool {
	switch event {
	case "tenant_provision":
		return isOperationFailureResult(result)
	case "auth", "tenant_schema_init", "tenant_status", "api_key_manage", "token_refresh":
		return true
	case "fs_invalid_path":
		// Failure-only and rare (client sent an illegal path). Tenant
		// attribution is what separates "one tenant misbehaving" from a
		// cross-tenant server regression, and the volume cannot blow up
		// cardinality.
		return true
	default:
		return false
	}
}

func metricLabelValue(labels []string, key string) string {
	for i := 0; i+1 < len(labels); i += 2 {
		if labels[i] == key {
			return cleanMetricValue(labels[i+1], "unknown")
		}
	}
	return "unknown"
}

func RecordTenantRequest(tenantID, surface, action string, status int, d time.Duration) {
	RecordTenantRequestWithOrg(tenantID, "", surface, action, status, d)
}

func RecordTenantRequestWithOrg(tenantID, tidbCloudOrgID, surface, action string, status int, d time.Duration) {
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	surface = cleanMetricValue(surface, "other")
	action = cleanMetricValue(action, "other")
	statusClass := "unknown"
	if status > 0 {
		statusClass = strconv.Itoa(status/100) + "xx"
	}
	RegisterModule("tenant_usage")
	attrs := []Attribute{
		Attr("surface", surface),
		Attr("action", action),
		Attr("status_class", statusClass),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
	}
	if tenantAttributedSurface(surface, statusClass) {
		attrs = append(attrs, Attr("tenant_id", tenantID))
	}
	tenantRequestsTotal.Add(1, attrs...)
	if d <= 0 {
		return
	}
	tenantRequestDuration.Observe(d.Seconds(),
		Attr("surface", surface),
		Attr("status_class", statusClass),
	)
}

func RecordTenantFileBytes(tenantID, direction, trafficPath string, bytes int64) {
	RecordTenantFileBytesWithOrg(tenantID, "", direction, trafficPath, bytes)
}

func RecordTenantFileBytesWithOrg(tenantID, tidbCloudOrgID, direction, trafficPath string, bytes int64) {
	if bytes <= 0 || (direction != "read" && direction != "write") ||
		(trafficPath != "db9" && trafficPath != "s3") {
		return
	}
	RegisterModule("tenant_usage")
	tenantFileBytes.Add(bytes,
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("direction", direction),
		Attr("traffic_path", trafficPath),
	)
}

// RecordTenantFileOps counts successful file read/write operations for the
// Plan B per-ops SKUs. Unlike RecordTenantFileBytes it is not guarded by a
// positive byte count: reading or writing an empty file is still one
// billable op. Batch endpoints pass the number of successful items.
func RecordTenantFileOps(tenantID, direction, trafficPath string, ops int64) {
	RecordTenantFileOpsWithOrg(tenantID, "", direction, trafficPath, ops)
}

func RecordTenantFileOpsWithOrg(tenantID, tidbCloudOrgID, direction, trafficPath string, ops int64) {
	if ops <= 0 || (direction != "read" && direction != "write") ||
		(trafficPath != "db9" && trafficPath != "s3") {
		return
	}
	RegisterModule("tenant_usage")
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	tenantFileOps.Add(ops,
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
		Attr("direction", direction),
		Attr("traffic_path", trafficPath),
	)
	// Mirror the absolute series value for the billing delta flush: the
	// billing worker reads these deltas from the central meta DB instead of
	// depending on an external metrics store.
	fileOpsFlush.add(tenantID, tidbCloudOrgID, direction, trafficPath, ops)
}

func RecordTenantInFlight(tenantID, surface string, value float64) {
	RecordTenantInFlightWithOrg(tenantID, "", surface, value)
}

func RecordTenantInFlightWithOrg(tenantID, tidbCloudOrgID, surface string, value float64) {
	RegisterModule("tenant_usage")
	attrs := []Attribute{
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("surface", cleanMetricValue(surface, "other")),
	}
	if value <= 0 {
		tenantInflight.Delete(attrs...)
		return
	}
	tenantInflight.Set(value, attrs...)
}

func RecordTenantCount(status string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("tenant_usage")
	tenantCount.Set(float64(count),
		Attr("status", cleanMetricValue(status, "unknown")),
	)
}

func RecordTenantStorageBytes(tenantID, state string, bytes int64) {
	RecordTenantStorageBytesWithOrg(tenantID, "", state, bytes)
}

func RecordTenantStorageBytesWithOrg(tenantID, tidbCloudOrgID, state string, bytes int64) {
	if bytes < 0 || (state != "confirmed" && state != "reserved" && state != "limit") {
		return
	}
	RegisterModule("tenant_usage")
	tenantStorageBytes.Set(float64(bytes),
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("state", cleanMetricValue(state, "unknown")),
		Attr("storage_type", "all"),
	)
}

// ReplaceTenantQuotaSnapshotByStorageTierWithOrg atomically replaces storage
// quota gauges while using an explicit readiness bit for the storage split.
// Pointer presence alone is not sufficient evidence that a full scan and its
// mutation watermark completed.
func ReplaceTenantQuotaSnapshotByStorageTierWithOrg(
	tenantID, tidbCloudOrgID string,
	storageBytes int64,
	dbStorageBytes, s3StorageBytes *int64,
	storageTierReady bool,
	reservedBytes, storageLimitBytes int64,
	replaceLimits bool,
) {
	tenantID = cleanMetricValue(tenantID, "unknown")
	if tenantID == "unknown" {
		return
	}
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	RegisterModule("tenant_usage")

	globalRegistry.mu.Lock()
	defer globalRegistry.mu.Unlock()
	tenantState := globalRegistry.admitTenantSeriesLocked(tenantID, globalRegistry.tenantRetention.now())
	previous := globalRegistry.tenantQuotaSnapshots[tenantID]
	if replaceLimits {
		previous.storageLimitBytes = storageLimitBytes
	}
	if previous.orgID != "" && previous.orgID != tidbCloudOrgID {
		deleteTenantQuotaSnapshotLabelsLocked(globalRegistry, tenantState, tenantID, previous.orgID)
	}
	previous.orgID = tidbCloudOrgID
	globalRegistry.tenantQuotaSnapshots[tenantID] = previous

	replaceTenantStorageGaugesLocked(globalRegistry, tenantState, tenantID, tidbCloudOrgID,
		storageBytes, dbStorageBytes, s3StorageBytes, storageTierReady, reservedBytes, previous.storageLimitBytes)
}

// ReplaceTenantFileCountSnapshotWithOrg atomically replaces the tenant's
// drive9_tenant_file_count gauge. Background refreshers call this at a low
// cadence (tenant maintenance loop); the db9/s3 split follows the same
// readiness and reconciliation contract as the storage-bytes split: it is
// exported only when the tier scan is ready and the per-tier counts add up
// to the authoritative file_count, otherwise a single storage_type="all"
// series is exported.
func ReplaceTenantFileCountSnapshotWithOrg(
	tenantID, tidbCloudOrgID string,
	fileCount int64,
	dbFileCount, s3FileCount *int64,
	storageTierReady bool,
) {
	tenantID = cleanMetricValue(tenantID, "unknown")
	if tenantID == "unknown" {
		return
	}
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	RegisterModule("tenant_usage")

	globalRegistry.mu.Lock()
	defer globalRegistry.mu.Unlock()
	tenantState := globalRegistry.admitTenantSeriesLocked(tenantID, globalRegistry.tenantRetention.now())
	previous := globalRegistry.tenantQuotaSnapshots[tenantID]
	if previous.fileCountOrgID != "" && previous.fileCountOrgID != tidbCloudOrgID {
		for _, storageType := range []string{"all", "db9", "s3"} {
			deleteTenantFileCountGaugeLocked(globalRegistry, tenantState, tenantID, previous.fileCountOrgID, "confirmed", storageType)
		}
	}
	previous.fileCountOrgID = tidbCloudOrgID
	globalRegistry.tenantQuotaSnapshots[tenantID] = previous

	replaceTenantFileCountGaugesLocked(globalRegistry, tenantState, tenantID, tidbCloudOrgID,
		fileCount, dbFileCount, s3FileCount, storageTierReady)
}

func replaceTenantStorageGaugesLocked(
	r *Registry,
	tenantState *tenantSeriesState,
	tenantID, tidbCloudOrgID string,
	storageBytes int64,
	dbStorageBytes, s3StorageBytes *int64,
	storageTierReady bool,
	reservedBytes, storageLimitBytes int64,
) {
	deleteTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "all")
	deleteTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "db9")
	deleteTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "s3")
	if storageTierReady && dbStorageBytes != nil && s3StorageBytes != nil &&
		*dbStorageBytes >= 0 && *s3StorageBytes >= 0 && *dbStorageBytes+*s3StorageBytes == storageBytes {
		setTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "db9", *dbStorageBytes)
		setTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "s3", *s3StorageBytes)
	} else {
		setTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "all", storageBytes)
	}
	setTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "reserved", "all", reservedBytes)
	if storageLimitBytes > 0 {
		setTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "limit", "all", storageLimitBytes)
	} else {
		deleteTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "limit", "all")
	}
}

func setTenantStorageGaugeLocked(r *Registry, tenantState *tenantSeriesState, tenantID, tidbCloudOrgID, state, storageType string, value int64) {
	if value < 0 {
		return
	}
	labels := labelsKey([]Attribute{
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
		Attr("state", state),
		Attr("storage_type", storageType),
	})
	inst := r.gauges[tenantStorageBytes.name]
	_, exists := inst.values[labels]
	inst.values[labels] = float64(value)
	if !exists {
		indexTenantSeriesLocked(tenantState, tenantSeriesRef{kind: instrumentGauge, name: tenantStorageBytes.name, labels: labels})
	}
}

func deleteTenantStorageGaugeLocked(r *Registry, tenantState *tenantSeriesState, tenantID, tidbCloudOrgID, state, storageType string) {
	labels := labelsKey([]Attribute{
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
		Attr("state", state),
		Attr("storage_type", storageType),
	})
	delete(r.gauges[tenantStorageBytes.name].values, labels)
	removeTenantSeriesRefFromStateLocked(tenantState, tenantSeriesRef{kind: instrumentGauge, name: tenantStorageBytes.name, labels: labels})
}

// replaceTenantFileCountGaugesLocked mirrors the storage split for the file
// count: the db9/s3 split is exported only when the storage tier scan is
// ready and the per-tier counts reconcile with the authoritative file_count
// counter; anything else falls back to a single storage_type="all" series.
func replaceTenantFileCountGaugesLocked(
	r *Registry,
	tenantState *tenantSeriesState,
	tenantID, tidbCloudOrgID string,
	fileCount int64,
	dbFileCount, s3FileCount *int64,
	storageTierReady bool,
) {
	deleteTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "all")
	deleteTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "db9")
	deleteTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "s3")
	if storageTierReady && dbFileCount != nil && s3FileCount != nil &&
		*dbFileCount >= 0 && *s3FileCount >= 0 && *dbFileCount+*s3FileCount == fileCount {
		setTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "db9", *dbFileCount)
		setTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "s3", *s3FileCount)
	} else {
		setTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", "all", fileCount)
	}
}

func setTenantFileCountGaugeLocked(r *Registry, tenantState *tenantSeriesState, tenantID, tidbCloudOrgID, state, storageType string, value int64) {
	if value < 0 {
		return
	}
	labels := labelsKey([]Attribute{
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
		Attr("state", state),
		Attr("storage_type", storageType),
	})
	inst := r.gauges[tenantFileCount.name]
	_, exists := inst.values[labels]
	inst.values[labels] = float64(value)
	if !exists {
		indexTenantSeriesLocked(tenantState, tenantSeriesRef{kind: instrumentGauge, name: tenantFileCount.name, labels: labels})
	}
}

func deleteTenantFileCountGaugeLocked(r *Registry, tenantState *tenantSeriesState, tenantID, tidbCloudOrgID, state, storageType string) {
	labels := labelsKey([]Attribute{
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
		Attr("state", state),
		Attr("storage_type", storageType),
	})
	delete(r.gauges[tenantFileCount.name].values, labels)
	removeTenantSeriesRefFromStateLocked(tenantState, tenantSeriesRef{kind: instrumentGauge, name: tenantFileCount.name, labels: labels})
}

func deleteTenantQuotaSnapshotLabelsLocked(r *Registry, tenantState *tenantSeriesState, tenantID, tidbCloudOrgID string) {
	for _, dimension := range []struct{ state, storageType string }{
		{"confirmed", "all"}, {"confirmed", "db9"}, {"confirmed", "s3"},
		{"reserved", "all"}, {"limit", "all"},
	} {
		deleteTenantStorageGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, dimension.state, dimension.storageType)
	}
	for _, storageType := range []string{"all", "db9", "s3"} {
		deleteTenantFileCountGaugeLocked(r, tenantState, tenantID, tidbCloudOrgID, "confirmed", storageType)
	}
}

// TenantFileCountTenantIDs lists tenants that currently export a
// drive9_tenant_file_count series on this process. Used to drop leftover
// series after shard ownership moves off this pod.
func TenantFileCountTenantIDs() []string {
	return globalRegistry.namedGaugeTenantIDs(tenantFileCount.name)
}

// ClearTenantFileCount drops this process's drive9_tenant_file_count series
// for tenantID (any org label). Used when the tenant's shard ownership moves
// to another pod: per-tenant retention cannot be relied on here because
// request-path metrics on this pod keep the tenant's series state fresh, so
// a stale file-count value would keep winning cross-pod aggregations.
func ClearTenantFileCount(tenantID string) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	cleanTenantID := cleanMetricValue(tenantID, "unknown")
	if cleanTenantID == "unknown" {
		return
	}
	RegisterModule("tenant_usage")
	globalRegistry.deleteNamedGaugesByTenant(tenantFileCount.name, cleanTenantID)
}

// RecordSSEConnection records an SSE connection lifecycle: it increments the
// connection counter and observes the connection lifetime (time from accept
// to client disconnect) into the dedicated SSE histogram. This duration must
// NOT be recorded into drive9_http_request_duration_seconds — SSE connections
// live as long as the client stays subscribed, so mixing them into the HTTP
// latency histogram would make all HTTP P95/P99 alerts meaningless.
func RecordSSEConnection(tenantID, reason string, d time.Duration) {
	RecordSSEConnectionWithOrg(tenantID, "", reason, d)
}

func RecordSSEConnectionWithOrg(tenantID, tidbCloudOrgID, reason string, d time.Duration) {
	RegisterModule("sse")
	tenantID = cleanMetricValue(tenantID, "unknown")
	tidbCloudOrgID = cleanTiDBCloudOrgID(tidbCloudOrgID)
	sseConnectionsTotal.Add(1,
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
		Attr("reason", cleanMetricValue(reason, "unknown")),
	)
	if d <= 0 {
		return
	}
	sseConnectionDuration.Observe(d.Seconds(),
		Attr("tenant_id", tenantID),
		Attr("tidbcloud_org_id", tidbCloudOrgID),
	)
}

// RecordSSEInFlight sets the active SSE connection count for a tenant. The
// caller maintains the absolute count (incremented on subscribe, decremented
// on unsubscribe); this records the current value so the gauge reflects the
// true number of live SSE connections per tenant.
func RecordSSEInFlight(tenantID string, count float64) {
	RecordSSEInFlightWithOrg(tenantID, "", count)
}

func RecordSSEInFlightWithOrg(tenantID, tidbCloudOrgID string, count float64) {
	RegisterModule("sse")
	attrs := []Attribute{
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	}
	if count <= 0 {
		sseInflight.Delete(attrs...)
		return
	}
	sseInflight.Set(count, attrs...)
}

// RecordSSEPhase1 records the duration of the SSE Phase-1 replay/reset stage.
func RecordSSEPhase1(tenantID string, d time.Duration) {
	RecordSSEPhase1WithOrg(tenantID, "", d)
}

func RecordSSEPhase1WithOrg(tenantID, tidbCloudOrgID string, d time.Duration) {
	if d <= 0 {
		return
	}
	RegisterModule("sse")
	ssePhase1Duration.Observe(d.Seconds(),
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	)
}

// RecordSSEEventSent records a single SSE file_changed event delivery.
func RecordSSEEventSent(tenantID, op string) {
	RecordSSEEventSentWithOrg(tenantID, "", op)
}

func RecordSSEEventSentWithOrg(tenantID, tidbCloudOrgID, op string) {
	RegisterModule("sse")
	sseEventsSentTotal.Add(1,
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("op", cleanMetricValue(op, "unknown")),
	)
}

// RecordSSEResetSent records an SSE reset event delivery.
func RecordSSEResetSent(tenantID, reason string) {
	RecordSSEResetSentWithOrg(tenantID, "", reason)
}

func RecordSSEResetSentWithOrg(tenantID, tidbCloudOrgID, reason string) {
	RegisterModule("sse")
	sseResetsSentTotal.Add(1,
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("reason", cleanMetricValue(reason, "unknown")),
	)
}

// RecordSSEHeartbeatSent records an SSE heartbeat delivery.
func RecordSSEHeartbeatSent(tenantID string) {
	RecordSSEHeartbeatSentWithOrg(tenantID, "")
}

func RecordSSEHeartbeatSentWithOrg(tenantID, tidbCloudOrgID string) {
	RegisterModule("sse")
	sseHeartbeatsSentTotal.Add(1,
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	)
}

// RecordEventBusQuery records the duration of an fs_events DB query. Unlike
// the error-only counters, this records every query (ok and error) so DB
// pressure and table growth on the events path are observable.
func RecordEventBusQuery(tenantID, operation, result string, d time.Duration) {
	if d <= 0 {
		return
	}
	RegisterModule("sse")
	eventBusQueryDuration.Observe(d.Seconds(),
		Attr("operation", cleanMetricValue(operation, "unknown")),
		Attr("result", cleanMetricValue(result, "unknown")),
	)
}

// ObserveFileGCClaimableBacklog records sparse File GC backlog gauges for a
// tenant. When claimable is 0 the series are deleted (zero is not exported).
// oldestAge should be time.Since(oldestAvailableAt) for claimable rows; ignored
// when claimable is 0. Negative ages are clamped to 0.
func ObserveFileGCClaimableBacklog(tenantID, tidbCloudOrgID string, claimable int64, oldestAge time.Duration) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	RegisterModule("file_gc")
	cleanTenantID := cleanMetricValue(tenantID, "unknown")
	if cleanTenantID == "unknown" {
		return
	}
	orgID := cleanTiDBCloudOrgID(tidbCloudOrgID)
	if claimable <= 0 {
		deleteFileGCBacklogGauges(cleanTenantID, orgID)
		return
	}
	ageSec := oldestAge.Seconds()
	if ageSec < 0 {
		ageSec = 0
	}
	fileGCClaimable.Set(float64(claimable),
		Attr("tenant_id", cleanTenantID),
		Attr("tidbcloud_org_id", orgID),
	)
	fileGCOldestClaimableAgeSeconds.Set(ageSec,
		Attr("tenant_id", cleanTenantID),
		Attr("tidbcloud_org_id", orgID),
	)
}

// RecordFSEventsRows records the current fs_events row count for a tenant.
// This compensates for the lack of direct TiDB access: the leader cleanup
// goroutine samples the count and reports it here.
func RecordFSEventsRows(tenantID string, count int64) {
	RecordFSEventsRowsWithOrg(tenantID, "", count)
}

func RecordFSEventsRowsWithOrg(tenantID, tidbCloudOrgID string, count int64) {
	if count < 0 {
		return
	}
	RegisterModule("sse")
	cleanTenantID := cleanMetricValue(tenantID, "unknown")
	// Maintenance observations must not re-admit or refresh a tenant: doing so
	// every five minutes would keep every warm tenant's full metric set alive.
	fsEventsRows.setForExistingTenant(cleanTenantID, float64(count),
		Attr("tenant_id", cleanTenantID),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	)
}

func DeleteFSEventsRowsWithOrg(tenantID, tidbCloudOrgID string) {
	fsEventsRows.Delete(
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
	)
}

// RecordFSEventsPruned records the number of fs_events rows deleted by
// retention cleanup.
func RecordFSEventsPruned(count int64) {
	if count <= 0 {
		return
	}
	RegisterModule("sse")
	fsEventsPrunedTotal.Add(count)
}

// RecordSSEEventRetryDropped records one retry-buffer entry dropped before its
// durable insert (hard event loss). reason is one of tenant_cap, global_cap,
// expired, stopped (enqueued after shutdown began), shutdown (still buffered
// when the shutdown flush budget ran out), or tenant_gone (tenant deleted
// before the flush landed). Alert on any non-zero rate.
func RecordSSEEventRetryDropped(tenantID, tidbCloudOrgID, reason string) {
	RegisterModule("sse")
	sseEventRetryDroppedTotal.Add(1,
		Attr("tenant_id", cleanMetricValue(tenantID, "unknown")),
		Attr("tidbcloud_org_id", cleanTiDBCloudOrgID(tidbCloudOrgID)),
		Attr("reason", cleanMetricValue(reason, "unknown")),
	)
}

// RecordSSEEventRetryBufferDepth records the current retry-buffer depth.
func RecordSSEEventRetryBufferDepth(depth int) {
	if depth < 0 {
		return
	}
	RegisterModule("sse")
	sseEventRetryBufferDepth.Set(float64(depth))
}

func WritePrometheus(w http.ResponseWriter) {
	if w == nil {
		return
	}
	w.Header().Set("Content-Type", "text/plain; version=0.0.4")
	SweepStaleFileGCDrainHitCap()
	globalRegistry.WritePrometheus(w)
	writeDBPrometheus(w)
	writeRuntimeMetrics(w)
	writeProcessMetrics(w)
}

func cleanMetricValue(value, fallback string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return fallback
	}
	return value
}

func cleanTiDBCloudOrgID(orgID string) string {
	return cleanMetricValue(orgID, tenantMetricGuestTiDBCloudOrgID)
}

func statusClass(status int) string {
	if status <= 0 {
		return "unknown"
	}
	return strconv.Itoa(status/100) + "xx"
}

func bodySizeBucket(bytes int64) string {
	switch {
	case bytes < 0:
		return "unknown"
	case bytes == 0:
		return "0"
	case bytes <= 1<<10:
		return "le_1KiB"
	case bytes <= 10<<10:
		return "le_10KiB"
	case bytes <= 100<<10:
		return "le_100KiB"
	case bytes <= 1<<20:
		return "le_1MiB"
	case bytes <= 10<<20:
		return "le_10MiB"
	default:
		return "gt_10MiB"
	}
}