package server

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/traceid"
	"go.uber.org/zap"
)

type metricsKeyType int

const (
	metricsKey metricsKeyType = iota
	requestMetricsKey
)

const defaultTenantMetricTiDBCloudOrgID = "guest"

func eventFields(ctx context.Context, event string, kv ...any) []zap.Field {
	fields := make([]zap.Field, 0, len(kv)/2+3)
	fields = append(fields, zap.String("event", event))
	hasTiDBCloudOrgID := false
	for i := 0; i+1 < len(kv); i += 2 {
		if strings.ReplaceAll(fmt.Sprint(kv[i]), " ", "_") == "tidbcloud_org_id" {
			hasTiDBCloudOrgID = true
			break
		}
	}
	if scope := ScopeFromContext(ctx); scope != nil {
		if scope.TenantID != "" {
			fields = append(fields, zap.String("tenant_id", scope.TenantID))
		}
		if scope.APIKeyID != "" {
			fields = append(fields, zap.String("api_key_id", scope.APIKeyID))
		}
		if scope.TiDBCloudOrgID != "" && !hasTiDBCloudOrgID {
			fields = append(fields, zap.String("tidbcloud_org_id", scope.TiDBCloudOrgID))
			hasTiDBCloudOrgID = true
		}
	}
	if !hasTiDBCloudOrgID {
		_, _, _, metricOrgID := requestMetricScope(ctx)
		if metricOrgID != "" {
			fields = append(fields, zap.String("tidbcloud_org_id", metricOrgID))
		}
	}
	for i := 0; i+1 < len(kv); i += 2 {
		key := strings.ReplaceAll(fmt.Sprint(kv[i]), " ", "_")
		if key == "" {
			continue
		}
		fields = append(fields, zap.Any(key, kv[i+1]))
	}
	return fields
}

type requestMetricState struct {
	mu             sync.Mutex
	tenantID       string
	tidbCloudOrgID string
	apiKeyID       string
	provider       string
	surface        string
	action         string
	tenantInFlight bool

	tenantHTTPRequestRecorded bool
	tenantHTTPRequestRecorder func()

	bodyReadObserved bool
	bodyReadMethod   string
	bodyReadRoute    string
	bodyReadBytes    int64
	bodyReadDuration time.Duration

	// sseStreamEstablished is set by an SSE handler (handleEvents) once it has
	// written the 200 streaming headers and entered its long-lived select
	// loop. observe uses it to distinguish a real SSE connection (record
	// establishment latency, not the full connection lifetime) from a bounded
	// error response on the same route (bad auth, invalid since, non-GET)
	// that is recorded as a normal HTTP request. See the route guard in
	// observe.
	sseStreamEstablished bool
	// sseEstablishedAt is the moment handleEvents finished writing the 200
	// streaming headers (time-to-first-byte). observe records
	// (sseEstablishedAt - start) into the HTTP/tenant duration histograms so
	// /v1/events establishment latency is measured on the same footing as
	// bounded HTTP requests, without the select-loop hold-open time polluting
	// the histogram. The full connection lifetime goes into the dedicated
	// drive9_sse_connection_duration_seconds metric recorded by handleEvents.
	sseEstablishedAt time.Time
}

func withMetrics(ctx context.Context, m *serverMetrics) context.Context {
	return context.WithValue(ctx, metricsKey, m)
}

func withRequestMetricState(ctx context.Context, state *requestMetricState) context.Context {
	return context.WithValue(ctx, requestMetricsKey, state)
}

func metricsFromContext(ctx context.Context) *serverMetrics {
	v, _ := ctx.Value(metricsKey).(*serverMetrics)
	return v
}

func requestMetricStateFromContext(ctx context.Context) *requestMetricState {
	v, _ := ctx.Value(requestMetricsKey).(*requestMetricState)
	return v
}

func setRequestTenantHTTPRecorder(ctx context.Context, recorder func()) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return
	}
	state.mu.Lock()
	state.tenantHTTPRequestRecorder = recorder
	state.mu.Unlock()
}

// recordTenantHTTPRequestAtCompletion records tenant HTTP usage at most once.
// Auth middleware invokes it before releasing the final backend reference so a
// closeEntry cleanup cannot be followed by an outer observe re-creation of the
// same tenant request counter. observe invokes it as a fallback for requests
// that do not acquire a tenant backend through auth middleware.
func recordTenantHTTPRequestAtCompletion(ctx context.Context) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return
	}
	state.mu.Lock()
	if state.tenantHTTPRequestRecorded || state.tenantHTTPRequestRecorder == nil {
		state.mu.Unlock()
		return
	}
	state.tenantHTTPRequestRecorded = true
	recorder := state.tenantHTTPRequestRecorder
	state.mu.Unlock()
	recorder()
}

// markSSEStreamEstablished signals that the SSE handler has written the
// streaming response headers and entered its long-lived loop. It records
// the establishment timestamp so observe can measure the time-to-first-byte
// (request → 200 headers + flush) — a bounded, meaningful latency that
// belongs in the HTTP duration histogram — rather than the full connection
// lifetime (select-loop hold-open), which is recorded into the dedicated
// drive9_sse_connection_duration_seconds metric. Safe to call from the SSE
// handler after WriteHeader(200).
func markSSEStreamEstablished(ctx context.Context) {
	if st := requestMetricStateFromContext(ctx); st != nil {
		now := time.Now()
		st.mu.Lock()
		st.sseStreamEstablished = true
		st.sseEstablishedAt = now
		st.mu.Unlock()
	}
}

// sseStreamEstablished reports whether the SSE handler marked the response
// as an established streaming connection. observe uses this to decide whether
// to record the establishment duration (instead of the full connection
// lifetime) into the HTTP/tenant duration histograms.
func sseStreamEstablished(ctx context.Context) bool {
	if st := requestMetricStateFromContext(ctx); st != nil {
		st.mu.Lock()
		defer st.mu.Unlock()
		return st.sseStreamEstablished
	}
	return false
}

// sseEstablishmentDuration returns the time-to-first-byte for an established
// SSE stream (from request start to the markSSEStreamEstablished call). Returns
// false if the stream was not marked as established.
func sseEstablishmentDuration(ctx context.Context, start time.Time) (time.Duration, bool) {
	if st := requestMetricStateFromContext(ctx); st != nil {
		st.mu.Lock()
		defer st.mu.Unlock()
		if st.sseStreamEstablished && !st.sseEstablishedAt.IsZero() {
			return st.sseEstablishedAt.Sub(start), true
		}
	}
	return 0, false
}

func setRequestMetricScope(ctx context.Context, scope *TenantScope, class tenantRequestClass) {
	if scope == nil {
		return
	}
	setRequestMetricTenant(ctx, scope.TenantID, scope.APIKeyID, scope.Provider, scope.TiDBCloudOrgID, class)
}

// setRequestMetricOrg labels the current request with a TiDB Cloud
// organization before a tenant is known (provision pre-flight rejections that
// return before provisionTenant). Tenant-scoped metrics still require
// setRequestMetricTenant; this only fixes the org on request logs/metrics so
// first-attempt failures do not fall back to the "guest" placeholder.
func setRequestMetricOrg(ctx context.Context, tidbCloudOrgID string) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return
	}
	state.mu.Lock()
	state.tidbCloudOrgID = normalizeTenantMetricTiDBCloudOrgID(tidbCloudOrgID)
	state.mu.Unlock()
}

func setRequestMetricTenant(ctx context.Context, tenantID, apiKeyID, provider, tidbCloudOrgID string, class tenantRequestClass) {
	state := requestMetricStateFromContext(ctx)
	if state == nil || tenantID == "" {
		return
	}
	tidbCloudOrgID = normalizeTenantMetricTiDBCloudOrgID(tidbCloudOrgID)
	surface := strings.TrimSpace(class.surface)
	if surface == "" {
		surface = "other"
	}
	action := strings.TrimSpace(class.action)
	if action == "" {
		action = "other"
	}

	var oldTenantID, oldTiDBCloudOrgID, oldSurface string
	var shouldMove bool
	state.mu.Lock()
	if state.tenantInFlight {
		shouldMove = state.tenantID != tenantID || state.tidbCloudOrgID != tidbCloudOrgID || state.surface != surface
		if shouldMove {
			oldTenantID = state.tenantID
			oldTiDBCloudOrgID = state.tidbCloudOrgID
			oldSurface = state.surface
			state.tidbCloudOrgID = tidbCloudOrgID
			state.surface = surface
		}
		state.tenantID = tenantID
		state.tidbCloudOrgID = tidbCloudOrgID
		state.apiKeyID = apiKeyID
		state.provider = provider
		state.action = action
		state.mu.Unlock()
		if shouldMove {
			if m := metricsFromContext(ctx); m != nil {
				m.adjustTenantInFlight(oldTenantID, oldTiDBCloudOrgID, oldSurface, -1)
				m.adjustTenantInFlight(tenantID, tidbCloudOrgID, surface, 1)
			}
		}
		return
	}
	state.tenantID = tenantID
	state.tidbCloudOrgID = tidbCloudOrgID
	state.apiKeyID = apiKeyID
	state.provider = provider
	state.surface = surface
	state.action = action
	state.tenantInFlight = true
	state.mu.Unlock()

	if m := metricsFromContext(ctx); m != nil {
		m.adjustTenantInFlight(tenantID, tidbCloudOrgID, surface, 1)
	}
}

// setRequestMetricTenantForAuthStatus scopes tenant request metrics only after
// the auth path has established that the tenant is active. Rejected
// deleting/deleted requests must not recreate series removed by lifecycle
// cleanup outbox delivery.
func setRequestMetricTenantForAuthStatus(ctx context.Context, tenantID, apiKeyID, provider, tidbCloudOrgID string, status meta.TenantStatus, class tenantRequestClass) {
	if status != meta.TenantActive {
		return
	}
	setRequestMetricTenant(ctx, tenantID, apiKeyID, provider, tidbCloudOrgID, class)
}

func requestMetricScope(ctx context.Context) (tenantID, apiKeyID, provider, tidbCloudOrgID string) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return "", "", "", ""
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	return state.tenantID, state.apiKeyID, state.provider, state.tidbCloudOrgID
}

func requestMetricClass(ctx context.Context) (tenantRequestClass, bool) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return tenantRequestClass{}, false
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	if state.tenantID == "" || state.surface == "" || state.action == "" {
		return tenantRequestClass{}, false
	}
	return tenantRequestClass{surface: state.surface, action: state.action}, true
}

func setRequestBodyReadMetric(ctx context.Context, method, route string, bytes int64, d time.Duration) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	state.bodyReadObserved = true
	state.bodyReadMethod = method
	state.bodyReadRoute = route
	state.bodyReadBytes = bytes
	state.bodyReadDuration = d
}

func requestBodyReadMetric(ctx context.Context) (method, route string, bytes int64, d time.Duration, ok bool) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return "", "", 0, 0, false
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	if !state.bodyReadObserved {
		return "", "", 0, 0, false
	}
	return state.bodyReadMethod, state.bodyReadRoute, state.bodyReadBytes, state.bodyReadDuration, true
}

func finishRequestMetricTenant(ctx context.Context) {
	state := requestMetricStateFromContext(ctx)
	if state == nil {
		return
	}
	state.mu.Lock()
	if !state.tenantInFlight {
		state.mu.Unlock()
		return
	}
	tenantID := state.tenantID
	tidbCloudOrgID := state.tidbCloudOrgID
	surface := state.surface
	state.tenantInFlight = false
	state.mu.Unlock()

	if m := metricsFromContext(ctx); m != nil {
		m.adjustTenantInFlight(tenantID, tidbCloudOrgID, surface, -1)
	}
}

func metricEvent(ctx context.Context, event string, labels ...string) {
	m := metricsFromContext(ctx)
	if m == nil {
		return
	}
	tenantID, _, _, tidbCloudOrgID := requestMetricScope(ctx)
	m.recordEvent(tenantID, tidbCloudOrgID, event, labels...)
}

type serverMetrics struct {
	inFlight atomic.Int64
	routeMu  sync.Mutex
	routes   map[string]int64

	tenantMu              sync.Mutex
	tenantInFlight        map[string]int64
	tenantPoolBindMu      sync.Mutex
	tenantPoolBinding     map[tenantPoolBindingMetricKey]struct{}
	sharedDBPoolMu        sync.Mutex
	sharedDBPool          map[sharedDBPoolMetricKey]struct{}
	organizationStorageMu sync.Mutex
	organizationStorage   map[string]struct{}
}

func newServerMetrics() *serverMetrics {
	metrics.SetModuleAvailability("server", true)
	metrics.RecordHTTPInFlight(0)
	return &serverMetrics{
		routes:              map[string]int64{},
		tenantInFlight:      map[string]int64{},
		tenantPoolBinding:   map[tenantPoolBindingMetricKey]struct{}{},
		sharedDBPool:        map[sharedDBPoolMetricKey]struct{}{},
		organizationStorage: map[string]struct{}{},
	}
}

func (m *serverMetrics) record(method, route string, status int, d time.Duration) {
	metrics.RecordHTTPRequest(method, route, status, d)
}

func (m *serverMetrics) recordBodyRead(method, route string, status int, bodyBytes int64, d time.Duration) {
	metrics.RecordHTTPRequestBodyRead(method, route, status, bodyBytes, d)
}

func (m *serverMetrics) recordEvent(tenantID, tidbCloudOrgID, event string, labels ...string) {
	metrics.RecordTenantEventWithOrg(tenantID, tidbCloudOrgID, event, labels...)
}

func (m *serverMetrics) writePrometheus(w http.ResponseWriter) {
	metrics.WritePrometheus(w)
}

func (m *serverMetrics) adjustRouteInFlight(route string, delta int64) int64 {
	route = strings.TrimSpace(route)
	if route == "" {
		route = "other"
	}
	m.routeMu.Lock()
	defer m.routeMu.Unlock()
	next := m.routes[route] + delta
	if next <= 0 {
		delete(m.routes, route)
		return 0
	}
	m.routes[route] = next
	return next
}

func (m *serverMetrics) adjustTenantInFlight(tenantID, tidbCloudOrgID, surface string, delta int64) int64 {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return 0
	}
	tidbCloudOrgID = normalizeTenantMetricTiDBCloudOrgID(tidbCloudOrgID)
	surface = strings.TrimSpace(surface)
	if surface == "" {
		surface = "other"
	}
	key := tenantInFlightKey(tenantID, tidbCloudOrgID, surface)

	m.tenantMu.Lock()
	defer m.tenantMu.Unlock()
	next := m.tenantInFlight[key] + delta
	if next <= 0 {
		delete(m.tenantInFlight, key)
		metrics.RecordTenantInFlightWithOrg(tenantID, tidbCloudOrgID, surface, 0)
		return 0
	}
	m.tenantInFlight[key] = next
	metrics.RecordTenantInFlightWithOrg(tenantID, tidbCloudOrgID, surface, float64(next))
	return next
}

func tenantInFlightKey(tenantID, tidbCloudOrgID, surface string) string {
	return tenantID + "\x00" + tidbCloudOrgID + "\x00" + surface
}

type observedResponseWriter struct {
	http.ResponseWriter
	status int
	size   int
}

type flusherResponseWriter struct {
	*observedResponseWriter
	flusher http.Flusher
}

func (w *flusherResponseWriter) Flush() {
	w.flusher.Flush()
}

func (w *observedResponseWriter) WriteHeader(code int) {
	w.status = code
	w.ResponseWriter.WriteHeader(code)
}

func (w *observedResponseWriter) Write(p []byte) (int, error) {
	if w.status == 0 {
		w.status = http.StatusOK
	}
	n, err := w.ResponseWriter.Write(p)
	w.size += n
	return n, err
}

// requestRoute maps a request to a bounded Prometheus route label. Wildcard
// routes that multiplex several operations behind one path pattern are
// further split by an action suffix ("/v1/fs/*?grep"): the suffix comes from
// the bounded classify*Action enums, never from raw path segments or query
// values, so label cardinality stays fixed.
func requestRoute(r *http.Request) string {
	path := r.URL.Path
	switch {
	case path == "/metrics":
		return "/metrics"
	case path == "/healthz":
		return "/healthz"
	case path == "/v1/provision":
		return "/v1/provision"
	case path == "/v1/quota":
		return "/v1/quota"
	case path == "/v1/status":
		return "/v1/status"
	case path == "/v1/tenant":
		return "/v1/tenant"
	case path == "/v1/tokens" || strings.HasPrefix(path, "/v1/tokens/"):
		return "/v1/tokens/*?" + classifyTokenAction(r)
	case path == "/v1/sql":
		return "/v1/sql"
	case path == sseEventsRoute:
		return sseEventsRoute
	case path == "/v1/migration/events":
		return "/v1/migration/events"
	case path == "/v1/fork":
		return "/v1/fork"
	case path == "/v1/journals":
		return "/v1/journals"
	case strings.HasPrefix(path, "/v1/journals/"):
		// Split the bounded journal sub-resources (entries append/list,
		// verify) out of the journal item wildcard. Unknown sub-paths stay
		// in the base bucket so the label set cannot grow from raw input.
		_, sub, hasSub := strings.Cut(strings.TrimPrefix(path, "/v1/journals/"), "/")
		switch {
		case hasSub && sub == "entries":
			return "/v1/journals/*?entries"
		case hasSub && sub == "verify":
			return "/v1/journals/*?verify"
		}
		return "/v1/journals/*"
	case path == "/v1/journal-entries":
		return "/v1/journal-entries"
	case path == "/v1/git-workspaces":
		return "/v1/git-workspaces"
	case strings.HasPrefix(path, "/v1/git-workspaces/"):
		// Split the bounded workspace sub-resources out of the item
		// wildcard; unknown sub-paths stay in the base bucket. Mirrors the
		// dispatch in handleGitWorkspaceObject, including the
		// object-packs/{packID} prefix arm.
		_, sub, hasSub := strings.Cut(strings.TrimPrefix(path, "/v1/git-workspaces/"), "/")
		if hasSub {
			switch {
			case sub == "tree" || sub == "git-state" || sub == "overlay":
				return "/v1/git-workspaces/*?" + sub
			case sub == "object-packs" || strings.HasPrefix(sub, "object-packs/"):
				return "/v1/git-workspaces/*?object-packs"
			}
		}
		return "/v1/git-workspaces/*"
	case path == "/v1/layers":
		return "/v1/layers"
	case strings.HasPrefix(path, "/v1/layers/"):
		return "/v1/layers/*?" + classifyFSLayersAction(r)
	case strings.HasPrefix(path, "/v1/layer-checkpoints/"):
		return "/v1/layer-checkpoints/*"
	case path == "/v1/admin/tenant-pool":
		return "/v1/admin/tenant-pool"
	case path == "/v1/admin/tenants":
		return "/v1/admin/tenants"
	case strings.HasPrefix(path, "/v1/admin/tenants/"):
		// Split the bounded sub-resources (quota, extract-config) out of the
		// tenant item wildcard so they are visible separately from tenant
		// get/delete traffic. Reuse splitAdminTenantPath and match the exact
		// shapes adminTenantsItemHandler dispatches, so mislabeled shapes
		// (extra segments, empty segments) fall back to the base bucket with
		// their 404s.
		parts := splitAdminTenantPath(path)
		switch {
		case len(parts) == 2 && parts[1] == "quota":
			return "/v1/admin/tenants/*?quota"
		case len(parts) == 3 && parts[1] == "extract-config":
			return "/v1/admin/tenants/*?extract-config"
		}
		return "/v1/admin/tenants/*"
	case path == "/v1/auth/slock/login":
		return "/v1/auth/slock/login"
	case path == "/v1/auth/slock/callback":
		return "/v1/auth/slock/callback"
	case path == "/v1/fs:batch-stat":
		return "/v1/fs:batch-stat"
	case path == "/v1/fs:batch-read-small":
		return "/v1/fs:batch-read-small"
	case path == "/v1/fs:batch-write":
		return "/v1/fs:batch-write"
	case strings.HasPrefix(path, "/v1/fs/"):
		return "/v1/fs/*?" + classifyFSAction(r)
	case path == "/v1/uploads":
		return "/v1/uploads"
	case strings.HasPrefix(path, "/v1/uploads/"):
		return "/v1/uploads/*?" + classifyV1UploadAction(r)
	case strings.HasPrefix(path, "/v2/uploads/"):
		return "/v2/uploads/*?" + classifyV2UploadAction(r)
	case path == "/v1/vault/secrets" || strings.HasPrefix(path, "/v1/vault/secrets/"):
		return "/v1/vault/secrets/*"
	case path == "/v1/vault/tokens" || strings.HasPrefix(path, "/v1/vault/tokens/"):
		return "/v1/vault/tokens/*"
	case path == "/v1/vault/grants" || strings.HasPrefix(path, "/v1/vault/grants/"):
		return "/v1/vault/grants/*"
	case path == "/v1/vault/audit":
		return "/v1/vault/audit"
	case path == "/v1/vault/read" || strings.HasPrefix(path, "/v1/vault/read/"):
		return "/v1/vault/read/*"
	case strings.HasPrefix(path, "/s3/"):
		return "/s3/*?" + classifyS3Action(r)
	default:
		return "other"
	}
}

type tenantRequestClass struct {
	surface string
	action  string
}

func classifyTenantRequest(r *http.Request) tenantRequestClass {
	if r == nil || r.URL == nil {
		return tenantRequestClass{surface: "other", action: "other"}
	}
	path := r.URL.Path
	switch {
	case path == "/v1/fs:batch-stat":
		return tenantRequestClass{surface: "fs", action: "batch_stat"}
	case path == "/v1/fs:batch-read-small":
		return tenantRequestClass{surface: "fs", action: "batch_read_small"}
	case path == "/v1/fs:batch-write":
		return tenantRequestClass{surface: "fs", action: "batch_write"}
	case strings.HasPrefix(path, "/v1/fs/"):
		return tenantRequestClass{surface: "fs", action: classifyFSAction(r)}
	case path == "/v1/uploads":
		if r.Method == http.MethodPost {
			return tenantRequestClass{surface: "upload", action: "initiate"}
		}
		return tenantRequestClass{surface: "upload", action: "list"}
	case path == "/v1/uploads/initiate":
		return tenantRequestClass{surface: "upload", action: "initiate"}
	case strings.HasPrefix(path, "/v1/uploads/"):
		return tenantRequestClass{surface: "upload", action: classifyV1UploadAction(r)}
	case strings.HasPrefix(path, "/v2/uploads/"):
		return tenantRequestClass{surface: "upload", action: classifyV2UploadAction(r)}
	case path == "/v1/sql":
		return tenantRequestClass{surface: "sql", action: strings.ToLower(r.Method)}
	case path == sseEventsRoute:
		return tenantRequestClass{surface: "events", action: strings.ToLower(r.Method)}
	case path == "/v1/migration/events":
		return tenantRequestClass{surface: "migration", action: "event"}
	case path == "/v1/tokens" || strings.HasPrefix(path, "/v1/tokens/"):
		return tenantRequestClass{surface: "tokens", action: classifyTokenAction(r)}
	case path == "/v1/journals" || strings.HasPrefix(path, "/v1/journals/") || path == "/v1/journal-entries":
		return tenantRequestClass{surface: "journal", action: strings.ToLower(r.Method)}
	case path == "/v1/git-workspaces" || strings.HasPrefix(path, "/v1/git-workspaces/"):
		return tenantRequestClass{surface: "git_workspace", action: strings.ToLower(r.Method)}
	case strings.HasPrefix(path, "/v1/vault/"):
		return tenantRequestClass{surface: "vault", action: classifyVaultAction(r)}
	case strings.HasPrefix(path, "/s3/"):
		return tenantRequestClass{surface: "object_store", action: classifyS3Action(r)}
	case path == "/v1/status":
		return tenantRequestClass{surface: "status", action: strings.ToLower(r.Method)}
	case path == "/v1/provision":
		return tenantRequestClass{surface: "provision", action: strings.ToLower(r.Method)}
	case path == "/v1/quota":
		return tenantRequestClass{surface: "quota", action: strings.ToLower(r.Method)}
	case path == "/v1/tenant":
		return tenantRequestClass{surface: "tenant", action: strings.ToLower(r.Method)}
	case path == "/v1/fork":
		return tenantRequestClass{surface: "fork", action: classifyForkAction(r)}
	case strings.HasPrefix(path, "/v1/layer-checkpoints/"):
		return tenantRequestClass{surface: "layer", action: "checkpoint_get"}
	case path == "/v1/layers" || strings.HasPrefix(path, "/v1/layers/"):
		return tenantRequestClass{surface: "layer", action: classifyFSLayersAction(r)}
	// Admin and Slock routes normally have no TenantScope, so
	// recordTenantHTTPRequest skips tenant metrics for them. Keep their bounded
	// classes here for consistency if a scoped internal caller is added later.
	case path == "/v1/admin/tenant-pool" || path == "/v1/admin/tenants" || strings.HasPrefix(path, "/v1/admin/tenants/"):
		return tenantRequestClass{surface: "admin", action: strings.ToLower(r.Method)}
	case path == "/v1/auth/slock/login":
		return tenantRequestClass{surface: "auth", action: "slock_login"}
	case path == "/v1/auth/slock/callback":
		return tenantRequestClass{surface: "auth", action: "slock_callback"}
	default:
		return tenantRequestClass{surface: "other", action: strings.ToLower(r.Method)}
	}
}

func classifyForkAction(r *http.Request) string {
	switch r.Method {
	case http.MethodPost:
		return "create"
	case http.MethodDelete:
		return "delete"
	default:
		return strings.ToLower(r.Method)
	}
}

func classifyFSLayersAction(r *http.Request) string {
	if r.URL.Path == "/v1/layers" {
		switch r.Method {
		case http.MethodGet:
			return "list"
		case http.MethodPost:
			return "create"
		default:
			return strings.ToLower(r.Method)
		}
	}

	if strings.HasPrefix(r.URL.EscapedPath(), "/v1/layers/") {
		// Match handleFSLayerObject / handleFSLayerUploads exactly: parse the
		// raw EscapedPath remainder and compare sub-resources without Trim.
		// Slash variants (trailing/double slash) 404 in the handler and must
		// not fold into the real operation buckets.
		rest := strings.TrimPrefix(r.URL.EscapedPath(), "/v1/layers/")
		_, sub, hasSub := strings.Cut(rest, "/")
		if !hasSub {
			if r.Method == http.MethodGet {
				return "get"
			}
			return strings.ToLower(r.Method)
		}
		switch sub {
		case "diff":
			return "diff"
		case "fork":
			return "fork"
		case "chain":
			return "chain"
		case "checkpoints":
			return "checkpoint"
		case "entries":
			return "entries_" + strings.ToLower(r.Method)
		case "objects":
			return "objects_" + strings.ToLower(r.Method)
		case "events":
			return "events"
		case "rollback":
			return "rollback"
		case "commit":
			return "commit"
		case "uploads/initiate":
			return "upload_initiate"
		}
		if strings.HasPrefix(sub, "uploads/") {
			parts := strings.Split(sub, "/")
			if len(parts) == 3 && parts[1] != "" {
				switch parts[2] {
				case "presign":
					return "upload_presign"
				case "presign-batch":
					return "upload_presign_batch"
				case "complete":
					return "upload_complete"
				case "abort":
					return "upload_abort"
				}
			}
		}
		// Unknown sub-resources 404 in handleFSLayerObject; keep them out of
		// the real operation buckets.
		return "other"
	}
	return strings.ToLower(r.Method)
}

func classifyFSAction(r *http.Request) string {
	switch r.Method {
	case http.MethodGet:
		q := r.URL.Query()
		switch {
		case q.Has("stat"):
			return "stat"
		case q.Has("grep"):
			return "grep"
		case q.Has("find"):
			return "find"
		case q.Has("list"):
			return "list"
		default:
			return "read"
		}
	case http.MethodHead:
		return "stat"
	case http.MethodPut:
		return "write"
	case http.MethodPatch:
		return "patch"
	case http.MethodDelete:
		return "delete"
	case http.MethodPost:
		q := r.URL.Query()
		// Keep the key set and order in sync with handleFS's POST dispatcher.
		for _, key := range []string{"append-log", "append", "copy", "rename", "mkdir", "chmod", "setmeta", "create", "symlink", "hardlink"} {
			if q.Has(key) {
				return strings.ReplaceAll(key, "-", "_")
			}
		}
		return "post"
	default:
		return strings.ToLower(r.Method)
	}
}

func classifyV1UploadAction(r *http.Request) string {
	rest := strings.TrimPrefix(r.URL.Path, "/v1/uploads/")
	if rest == "" {
		return strings.ToLower(r.Method)
	}
	// POST /v1/uploads/initiate is a live alias for POST /v1/uploads (see
	// handleBusiness); keep it out of the junk/404 "other" bucket.
	if rest == "initiate" && r.Method == http.MethodPost {
		return "initiate"
	}
	parts := strings.SplitN(rest, "/", 2)
	if len(parts) == 1 {
		if r.Method == http.MethodDelete {
			return "abort"
		}
		return "other"
	}
	action := strings.Trim(parts[1], "/")
	if action == "" && r.Method == http.MethodDelete {
		return "abort"
	}
	switch {
	case strings.HasPrefix(action, "complete"):
		return "complete"
	case strings.HasPrefix(action, "resume"):
		return "resume"
	default:
		return "other"
	}
}

func classifyV2UploadAction(r *http.Request) string {
	rest := strings.TrimPrefix(r.URL.Path, "/v2/uploads/")
	parts := strings.SplitN(rest, "/", 2)
	if len(parts) == 0 || parts[0] == "" {
		return strings.ToLower(r.Method)
	}
	if parts[0] == "initiate" {
		return "initiate"
	}
	if len(parts) == 1 {
		return "other"
	}
	switch strings.Trim(parts[1], "/") {
	case "presign":
		return "presign"
	case "presign-batch":
		return "presign_batch"
	case "complete":
		return "complete"
	case "abort":
		return "abort"
	default:
		return "other"
	}
}

func classifyVaultAction(r *http.Request) string {
	path := r.URL.Path
	switch {
	case strings.HasPrefix(path, "/v1/vault/secrets"):
		return "secrets_" + strings.ToLower(r.Method)
	case strings.HasPrefix(path, "/v1/vault/tokens"):
		return "tokens_" + strings.ToLower(r.Method)
	case strings.HasPrefix(path, "/v1/vault/grants"):
		return "grants_" + strings.ToLower(r.Method)
	case strings.HasPrefix(path, "/v1/vault/read"):
		return "read"
	case path == "/v1/vault/audit":
		return "audit"
	default:
		return strings.ToLower(r.Method)
	}
}

func classifyTokenAction(r *http.Request) string {
	switch {
	case r.URL.Path == "/v1/tokens" && r.Method == http.MethodPost:
		return "issue"
	case r.URL.Path == "/v1/tokens" && r.Method == http.MethodGet:
		return "list"
	case r.URL.Path == "/v1/tokens/generate" && r.Method == http.MethodPost:
		return "generate"
	case r.URL.Path == "/v1/tokens/refresh" && r.Method == http.MethodPost:
		return "refresh"
	case r.URL.Path == "/v1/tokens/revoke" && r.Method == http.MethodPost:
		return "revoke_by_key"
	case strings.HasPrefix(r.URL.Path, "/v1/tokens/") && strings.HasSuffix(r.URL.Path, "/activate") && r.Method == http.MethodPost:
		return "activate"
	case strings.HasPrefix(r.URL.Path, "/v1/tokens/") && strings.HasSuffix(r.URL.Path, "/deactivate") && r.Method == http.MethodPost:
		return "deactivate"
	case strings.HasPrefix(r.URL.Path, "/v1/tokens/") && r.Method == http.MethodDelete:
		switch r.URL.Path {
		case "/v1/tokens/generate", "/v1/tokens/refresh", "/v1/tokens/revoke":
			// Reserved action paths with the wrong method are wrong-method-
			// on-known-shape, not token-ID deletes; keep them out of the
			// revoke bucket.
			return "unmatched"
		}
		return "revoke"
	default:
		// Self-identifying bucket for wrong-method-on-known-shape and
		// unknown-path cases (e.g. PUT /v1/tokens/generate vs POST /v1/tokens/bogus).
		// Avoid bare HTTP method labels that collapse both situations together.
		return "unmatched"
	}
}

func classifyS3Action(r *http.Request) string {
	if strings.Contains(r.URL.Path, "/upload/") && r.Method == http.MethodPut {
		return "upload_part"
	}
	if strings.Contains(r.URL.Path, "/objects/") && r.Method == http.MethodGet {
		return "get_object"
	}
	return strings.ToLower(r.Method)
}

func requestTenantID(r *http.Request) string {
	tenantID, _, _, _ := requestMetricScope(r.Context())
	if tenantID != "" {
		return tenantID
	}
	if r.URL != nil && strings.HasPrefix(r.URL.Path, "/s3/") {
		rest := strings.TrimPrefix(r.URL.Path, "/s3/")
		if tenant, _, ok := strings.Cut(rest, "/"); ok && tenant != "" {
			if tenant == "local" || tenant == "upload" || tenant == "objects" {
				return "local"
			}
		}
	}
	return ""
}

func requestTenantMetricScope(r *http.Request) (tenantID, tidbCloudOrgID string) {
	tenantID, _, _, tidbCloudOrgID = requestMetricScope(r.Context())
	if tenantID != "" {
		return tenantID, tidbCloudOrgID
	}
	if fallback := requestTenantID(r); fallback != "" {
		return fallback, ""
	}
	return "", ""
}

func recordTenantHTTPRequest(r *http.Request, status int, d time.Duration, responseBytes int) {
	tenantID, tidbCloudOrgID := requestTenantMetricScope(r)
	if tenantID == "" {
		return
	}
	class := classifyTenantRequest(r)
	if scopedClass, ok := requestMetricClass(r.Context()); ok {
		class = scopedClass
	}
	metrics.RecordTenantRequestWithOrg(tenantID, tidbCloudOrgID, class.surface, class.action, status, d)
}

func recordTenantFileBytes(ctx context.Context, direction, trafficPath string, bytes int64) {
	tenantID, _, _, tidbCloudOrgID := requestMetricScope(ctx)
	if tenantID == "" || bytes <= 0 {
		return
	}
	metrics.RecordTenantFileBytesWithOrg(tenantID, tidbCloudOrgID, direction, trafficPath, bytes)
}

// recordTenantFileOps counts successful file read/write operations for the
// Plan B per-ops SKUs. Unlike recordTenantFileBytes it is not guarded by byte
// count: an empty-file read/write is still one billable op. Callers should
// prefer recordTenantFileUsage so bytes and ops are always recorded together.
func recordTenantFileOps(ctx context.Context, direction, trafficPath string, ops int64) {
	tenantID, _, _, tidbCloudOrgID := requestMetricScope(ctx)
	if tenantID == "" || ops <= 0 {
		return
	}
	metrics.RecordTenantFileOpsWithOrg(tenantID, tidbCloudOrgID, direction, trafficPath, ops)
}

// recordTenantFileUsage records the billing data sources for one file
// operation: logical bytes and successful ops. A zero bytes value drops only
// the bytes sample (empty-file operations still count as ops); a zero ops
// value drops only the ops sample (failed deliveries still record the
// transferred bytes). All metered endpoints must use this single recorder so
// the bytes and ops series can never drift apart.
func recordTenantFileUsage(ctx context.Context, direction, trafficPath string, bytes, ops int64) {
	recordTenantFileBytes(ctx, direction, trafficPath, bytes)
	recordTenantFileOps(ctx, direction, trafficPath, ops)
}

// fileOpsForErr gates a billable op on delivery success: a failed copy/write
// records no op, while the caller still records the transferred bytes.
func fileOpsForErr(err error) int64 {
	if err != nil {
		return 0
	}
	return 1
}

func generateTraceID() string { return traceid.Generate() }

func (s *Server) observe(next http.Handler, w http.ResponseWriter, r *http.Request) {
	traceID := strings.TrimSpace(r.Header.Get("X-Trace-ID"))
	if traceID == "" {
		traceID = generateTraceID()
	}

	r = r.WithContext(traceid.With(r.Context(), traceID))
	r = r.WithContext(logger.WithContext(r.Context(), s.logger.With(zap.String("trace_id", traceID))))
	r = r.WithContext(withMetrics(r.Context(), s.metrics))
	r = r.WithContext(withRequestMetricState(r.Context(), &requestMetricState{}))
	w.Header().Set("X-Trace-ID", traceID)

	start := time.Now()
	route := requestRoute(r)
	ow := &observedResponseWriter{ResponseWriter: w}
	rw := http.ResponseWriter(ow)
	if f, ok := w.(http.Flusher); ok {
		rw = &flusherResponseWriter{observedResponseWriter: ow, flusher: f}
	}
	metrics.RecordHTTPInFlight(float64(s.metrics.inFlight.Add(1)))
	metrics.RecordHTTPInFlightRoute(route, float64(s.metrics.adjustRouteInFlight(route, 1)))
	defer func() {
		metrics.RecordHTTPInFlight(float64(s.metrics.inFlight.Add(-1)))
		metrics.RecordHTTPInFlightRoute(route, float64(s.metrics.adjustRouteInFlight(route, -1)))
		finishRequestMetricTenant(r.Context())
	}()

	if strings.HasPrefix(r.URL.Path, "/s3/") {
		setRequestMetricTenant(r.Context(), requestTenantID(r), "", "", "", classifyTenantRequest(r))
	}
	setRequestTenantHTTPRecorder(r.Context(), func() {
		status := ow.status
		if status == 0 {
			status = http.StatusOK
		}
		dur := time.Since(start)
		if route == sseEventsRoute && sseStreamEstablished(r.Context()) {
			if estDur, ok := sseEstablishmentDuration(r.Context(), start); ok {
				dur = estDur
			}
		}
		recordTenantHTTPRequest(r, status, dur, ow.size)
	})

	next.ServeHTTP(rw, r)
	if ow.status == 0 {
		ow.status = http.StatusOK
	}

	dur := time.Since(start)
	// An established SSE stream (/v1/events) blocks in a select loop for the
	// entire client connection lifetime — recording that lifetime into the
	// HTTP request duration histogram would pollute all HTTP P95/P99 alerts
	// (a 40s SSE connection would look like a 40s "request"). handleEvents
	// marks the stream as established (sseStreamEstablishedAt) after it has
	// written the 200 streaming headers. For an established stream, record the
	// establishment duration (request → first byte — a bounded, meaningful
	// latency) into the HTTP/tenant duration histograms so /v1/events is
	// monitored on the same footing as bounded HTTP requests. The full
	// connection lifetime goes into the dedicated
	// drive9_sse_connection_duration_seconds metric recorded by handleEvents.
	// Bounded error responses on the same route (bad auth, invalid since,
	// non-GET, backend setup failures) return before markSSEStreamEstablished
	// and fall through to the normal HTTP duration path below.
	if route == sseEventsRoute && sseStreamEstablished(r.Context()) {
		if estDur, ok := sseEstablishmentDuration(r.Context(), start); ok {
			dur = estDur
		}
		s.metrics.record(r.Method, route, ow.status, dur)
		recordTenantHTTPRequestAtCompletion(r.Context())
	} else {
		s.metrics.record(r.Method, route, ow.status, dur)
		if method, bodyRoute, bodyBytes, bodyReadDuration, ok := requestBodyReadMetric(r.Context()); ok {
			if method == "" {
				method = r.Method
			}
			if bodyRoute == "" {
				bodyRoute = route
			}
			s.metrics.recordBodyRead(method, bodyRoute, ow.status, bodyBytes, bodyReadDuration)
		}
		recordTenantHTTPRequestAtCompletion(r.Context())
	}

	tenantID, apiKeyID, _, tidbCloudOrgID := requestMetricScope(r.Context())

	fields := []zap.Field{
		zap.String("method", r.Method),
		zap.String("path", r.URL.Path),
		zap.String("route", route),
		zap.Int("status", ow.status),
		zap.Int("bytes", ow.size),
		zap.Float64("duration_ms", dur.Seconds()*1000),
		zap.String("remote", r.RemoteAddr),
		zap.String("user_agent", r.UserAgent()),
	}
	if tenantID != "" {
		fields = append(fields, zap.String("tenant_id", tenantID))
	}
	if apiKeyID != "" {
		fields = append(fields, zap.String("api_key_id", apiKeyID))
	}
	if tidbCloudOrgID != "" {
		fields = append(fields, zap.String("tidbcloud_org_id", tidbCloudOrgID))
	}
	if route == "/metrics" || route == "/healthz" {
		return
	}
	logger.Info(r.Context(), "http_request", fields...)
}

func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		errJSON(w, http.StatusMethodNotAllowed, "method not allowed")
		return
	}
	s.metrics.writePrometheus(w)
}

func normalizeTenantMetricTiDBCloudOrgID(orgID string) string {
	orgID = strings.TrimSpace(orgID)
	if orgID == "" {
		return defaultTenantMetricTiDBCloudOrgID
	}
	return orgID
}

func (s *Server) tenantMetricTiDBCloudOrgID(ctx context.Context, t *meta.Tenant) string {
	if s == nil {
		return defaultTenantMetricTiDBCloudOrgID
	}
	return tenantMetricTiDBCloudOrgIDFromMeta(ctx, s.meta, t)
}

func tenantMetricTiDBCloudOrgIDFromMeta(ctx context.Context, metaStore *meta.Store, t *meta.Tenant) string {
	if metaStore == nil || t == nil || strings.TrimSpace(t.ID) == "" || !tenant.UsesTiDBCloudNativeCredentials(t.Provider) {
		return defaultTenantMetricTiDBCloudOrgID
	}
	if tenant.IsSharedSchemaProvider(t.Provider) {
		pool, err := metaStore.GetSharedDBForTenant(ctx, t.ID)
		if err != nil {
			if !errors.Is(err, meta.ErrNotFound) {
				logger.Warn(ctx, "server_metric_shared_org_lookup_failed", zap.String("tenant_id", t.ID), zap.Error(err))
			}
			return defaultTenantMetricTiDBCloudOrgID
		}
		return normalizeTenantMetricTiDBCloudOrgID(pool.TiDBCloudOrganizationID)
	}
	if orgID := strings.TrimSpace(t.TiDBCloudOrgID); orgID != "" {
		return normalizeTenantMetricTiDBCloudOrgID(orgID)
	}
	binding, err := metaStore.GetTenantTiDBCloudOrgBinding(ctx, t.ID)
	if err != nil {
		if !errors.Is(err, meta.ErrNotFound) {
			logger.Warn(ctx, "server_metric_org_lookup_failed",
				zap.String("tenant_id", t.ID),
				zap.Error(err))
		}
		return defaultTenantMetricTiDBCloudOrgID
	}
	return normalizeTenantMetricTiDBCloudOrgID(binding.OrganizationID)
}