package securityevents

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

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/publicerror"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)

type MetricsSnapshotProvider interface {
	SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error)
}

type DynamicMetricsSnapshotProvider interface {
	MetricsSnapshotProvider
	SecurityEventConnection(context.Context) (store.SecurityEventConnection, error)
	BillingMetrics(context.Context) (store.BillingMetricsSnapshot, error)
}

type Metrics struct {
	accepted                    atomic.Uint64
	rejected                    atomic.Uint64
	duplicate                   atomic.Uint64
	sessionsDeleted             atomic.Uint64
	watermarkRejected           atomic.Uint64
	verificationAccepted        atomic.Uint64
	heartbeatAccepted           atomic.Uint64
	heartbeatFailed             atomic.Uint64
	introspectionActive         atomic.Uint64
	introspectionInactive       atomic.Uint64
	introspectionFailed         atomic.Uint64
	jwksSSFFailed               atomic.Uint64
	jwksOIDCFailed              atomic.Uint64
	processingCount             atomic.Uint64
	processingNanos             atomic.Uint64
	processingBuckets           [6]atomic.Uint64
	billingSettlementCompleted  atomic.Uint64
	billingSettlementRetry      atomic.Uint64
	billingManualReview         atomic.Uint64
	billingEstimateFailed       atomic.Uint64
	billingIdempotentReplay     atomic.Uint64
	billingPricingUnavailable   atomic.Uint64
	asyncWorkerCapacity         atomic.Int64
	asyncWorkerDesiredCapacity  atomic.Int64
	asyncWorkerHardLimit        atomic.Int64
	asyncWorkerCapacityCapped   atomic.Int64
	asyncWorkerActiveInstances  atomic.Int64
	asyncWorkerGlobalCapacity   atomic.Int64
	asyncWorkerAllocated        atomic.Int64
	workerSafeCapacity          atomic.Int64
	workerHeavyCapacity         atomic.Int64
	workerActiveTasks           atomic.Int64
	workerPreparingTasks        atomic.Int64
	workerWaitingTasks          atomic.Int64
	workerFinalizingTasks       atomic.Int64
	workerPressureState         atomic.Int64
	providerQuotaWaitRPM        atomic.Uint64
	providerQuotaWaitTPM        atomic.Uint64
	providerQuotaWaitConcurrent atomic.Uint64
	providerQuotaWaitOther      atomic.Uint64
	asyncWorkerResizeSuccess    atomic.Uint64
	asyncWorkerRefreshFailed    atomic.Uint64
	asyncWorkerCreateFailed     atomic.Uint64
	asyncWorkerStartFailed      atomic.Uint64
	leaseRenewalSuccess         atomic.Uint64
	leaseRenewalFailure         atomic.Uint64
	leaseRenewalLost            atomic.Uint64
	taskEventDuplicate          atomic.Uint64
	taskEventUnknownType        atomic.Uint64
	taskEventBudgetExceeded     atomic.Uint64
	taskAdmissionAdmitted       atomic.Uint64
	taskAdmissionQueueFull      atomic.Uint64
	taskAdmissionTimeout        atomic.Uint64
	taskAdmissionCancelled      atomic.Uint64
	taskAdmissionExpired        atomic.Uint64
	taskAdmissionMigrated       atomic.Uint64
	asyncAdmissionAdmitted      atomic.Uint64
	asyncAdmissionGlobalWait    atomic.Uint64
	asyncAdmissionPlatformWait  atomic.Uint64
	asyncAdmissionGroupWait     atomic.Uint64
	asyncAdmissionReselect      atomic.Uint64
	asyncAdmissionDeferred      atomic.Uint64
	asyncAdmissionTerminal      atomic.Uint64
	asyncAdmissionTaskError     atomic.Uint64
	asyncAdmissionSystemError   atomic.Uint64
	candidateNormalRotation     atomic.Uint64
	candidateFullAvoided        atomic.Uint64
	candidateQuotaRaceRotated   atomic.Uint64
	candidateCooldownSkipped    atomic.Uint64
	candidateDisabledSkipped    atomic.Uint64
	candidateAllFullQueued      atomic.Uint64
	candidateRoutingOther       atomic.Uint64
	executionPoolSelected       atomic.Uint64
	executionPoolShadow         atomic.Uint64
	executionPoolUnavailable    atomic.Uint64
	executionPoolCapacityReject atomic.Uint64
	routeProbeHealthy           atomic.Uint64
	routeProbeDegraded          atomic.Uint64
	routeProbeUnreachable       atomic.Uint64
	routeProbeUnknown           atomic.Uint64
	storageAliyunWrites         atomic.Uint64
	storageAliyunFailures       atomic.Uint64
	storageAliyunWriteNanos     atomic.Uint64
	storageS3Writes             atomic.Uint64
	storageS3Failures           atomic.Uint64
	storageS3WriteNanos         atomic.Uint64
	storageServerMainWrites     atomic.Uint64
	storageServerMainFailures   atomic.Uint64
	storageServerMainWriteNanos atomic.Uint64
	storageChannelRetries       atomic.Uint64
	storageChannelFailovers     atomic.Uint64
	storageAllChannelsFailed    atomic.Uint64
	storageObjectifiedBytes     atomic.Uint64
	storageObjectReads          atomic.Uint64
	storageObjectReadFailures   atomic.Uint64
	storageObjectReadBytes      atomic.Uint64
	resultSourceUpstreamURL     atomic.Uint64
	resultSourceUploaded        atomic.Uint64
	resultPollResponses         atomic.Uint64
	resultPollResponseBytes     atomic.Uint64
	resultPollOversized         atomic.Uint64
	resultURLSignSuccess        atomic.Uint64
	resultURLSignFailure        atomic.Uint64
	syncBase64Success           atomic.Uint64
	syncBase64CapacityTimeout   atomic.Uint64
	syncBase64TooLarge          atomic.Uint64
	taskAdmissionWaitBuckets    [11]atomic.Uint64
	taskAdmissionWaitMicros     atomic.Uint64
}

var processingDurationBounds = [...]time.Duration{
	10 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond,
	500 * time.Millisecond, time.Second, 3 * time.Second,
}

var taskAdmissionWaitBounds = [...]time.Duration{
	time.Second,
	5 * time.Second,
	15 * time.Second,
	30 * time.Second,
	time.Minute,
	2 * time.Minute,
	5 * time.Minute,
	10 * time.Minute,
	30 * time.Minute,
	time.Hour,
}

func (m *Metrics) ObserveReceipt(outcome string, duration time.Duration, sessionsDeleted int64) {
	switch outcome {
	case "accepted":
		m.accepted.Add(1)
	case "duplicate":
		m.duplicate.Add(1)
	default:
		m.rejected.Add(1)
	}
	if sessionsDeleted > 0 {
		m.sessionsDeleted.Add(uint64(sessionsDeleted))
	}
	if duration < 0 {
		duration = 0
	}
	m.processingCount.Add(1)
	m.processingNanos.Add(uint64(duration))
	for index, bound := range processingDurationBounds {
		if duration <= bound {
			m.processingBuckets[index].Add(1)
		}
	}
}

func (m *Metrics) ObserveWatermarkRejection() { m.watermarkRejected.Add(1) }

func (m *Metrics) ObserveVerificationAccepted() { m.verificationAccepted.Add(1) }

func (m *Metrics) ObserveHeartbeat(outcome string) {
	if outcome == "accepted" {
		m.heartbeatAccepted.Add(1)
		return
	}
	m.heartbeatFailed.Add(1)
}

func (m *Metrics) ObserveIntrospection(outcome string) {
	switch outcome {
	case "active":
		m.introspectionActive.Add(1)
	case "inactive":
		m.introspectionInactive.Add(1)
	default:
		m.introspectionFailed.Add(1)
	}
}

func (m *Metrics) ObserveJWKSRefreshFailure(source string) {
	if source == "ssf" {
		m.jwksSSFFailed.Add(1)
		return
	}
	m.jwksOIDCFailed.Add(1)
}

func (m *Metrics) ObserveBillingEvent(event string) {
	switch event {
	case "settlement_completed":
		m.billingSettlementCompleted.Add(1)
	case "settlement_retry":
		m.billingSettlementRetry.Add(1)
	case "manual_review":
		m.billingManualReview.Add(1)
	case "estimate_failed":
		m.billingEstimateFailed.Add(1)
	case "idempotent_replay":
		m.billingIdempotentReplay.Add(1)
	case "pricing_unavailable":
		m.billingPricingUnavailable.Add(1)
	}
}

func (m *Metrics) SetAsyncWorkerCapacity(current, desired, hardLimit int, capped bool) {
	m.asyncWorkerCapacity.Store(int64(current))
	m.asyncWorkerDesiredCapacity.Store(int64(desired))
	m.asyncWorkerHardLimit.Store(int64(hardLimit))
	cappedValue := int64(0)
	if capped {
		cappedValue = 1
	}
	m.asyncWorkerCapacityCapped.Store(cappedValue)
}

func (m *Metrics) SetDistributedWorkerCapacity(activeInstances, globalCapacity, allocatedCapacity int) {
	m.asyncWorkerActiveInstances.Store(int64(activeInstances))
	m.asyncWorkerGlobalCapacity.Store(int64(globalCapacity))
	m.asyncWorkerAllocated.Store(int64(allocatedCapacity))
}

func (m *Metrics) SetWorkerLoad(activeLimit, heavyLimit, active, preparing, waiting, finalizing int, pressure string) {
	m.workerSafeCapacity.Store(int64(activeLimit))
	m.workerHeavyCapacity.Store(int64(heavyLimit))
	m.workerActiveTasks.Store(int64(active))
	m.workerPreparingTasks.Store(int64(preparing))
	m.workerWaitingTasks.Store(int64(waiting))
	m.workerFinalizingTasks.Store(int64(finalizing))
	state := int64(-1)
	switch pressure {
	case "normal":
		state = 0
	case "busy":
		state = 1
	case "critical":
		state = 2
	}
	m.workerPressureState.Store(state)
}

func (m *Metrics) ObserveProviderQuotaWait(metric string) {
	switch metric {
	case "rpm":
		m.providerQuotaWaitRPM.Add(1)
	case "tpm", "tpm_total", "tpm_input", "tpm_output":
		m.providerQuotaWaitTPM.Add(1)
	case "concurrent":
		m.providerQuotaWaitConcurrent.Add(1)
	default:
		m.providerQuotaWaitOther.Add(1)
	}
}

func (m *Metrics) ObserveTaskAdmission(event string) {
	switch event {
	case "admitted":
		m.taskAdmissionAdmitted.Add(1)
	case "queue_full":
		m.taskAdmissionQueueFull.Add(1)
	case "timeout":
		m.taskAdmissionTimeout.Add(1)
	case "cancelled":
		m.taskAdmissionCancelled.Add(1)
	case "expired":
		m.taskAdmissionExpired.Add(1)
	case "candidate_migrated":
		m.taskAdmissionMigrated.Add(1)
	}
}

func (m *Metrics) ObserveAsyncAdmissionDispatch(outcome string) {
	switch outcome {
	case "admitted":
		m.asyncAdmissionAdmitted.Add(1)
	case "global_wait":
		m.asyncAdmissionGlobalWait.Add(1)
	case "platform_wait":
		m.asyncAdmissionPlatformWait.Add(1)
	case "group_wait":
		m.asyncAdmissionGroupWait.Add(1)
	case "reselect":
		m.asyncAdmissionReselect.Add(1)
	case "deferred":
		m.asyncAdmissionDeferred.Add(1)
	case "terminal":
		m.asyncAdmissionTerminal.Add(1)
	case "task_error":
		m.asyncAdmissionTaskError.Add(1)
	case "system_error":
		m.asyncAdmissionSystemError.Add(1)
	}
}

func (m *Metrics) ObserveCandidateRouting(reason string) {
	switch reason {
	case "normal_rotation":
		m.candidateNormalRotation.Add(1)
	case "full_avoided":
		m.candidateFullAvoided.Add(1)
	case "quota_race_rotated":
		m.candidateQuotaRaceRotated.Add(1)
	case "cooldown_skipped":
		m.candidateCooldownSkipped.Add(1)
	case "disabled_skipped":
		m.candidateDisabledSkipped.Add(1)
	case "all_full_queued":
		m.candidateAllFullQueued.Add(1)
	default:
		m.candidateRoutingOther.Add(1)
	}
}

func (m *Metrics) ObserveExecutionPoolRouting(outcome string) {
	switch outcome {
	case "selected":
		m.executionPoolSelected.Add(1)
	case "shadow":
		m.executionPoolShadow.Add(1)
	case "capacity_rejected":
		m.executionPoolCapacityReject.Add(1)
	default:
		m.executionPoolUnavailable.Add(1)
	}
}

func (m *Metrics) ObserveRouteProbe(state string) {
	switch state {
	case "healthy":
		m.routeProbeHealthy.Add(1)
	case "degraded":
		m.routeProbeDegraded.Add(1)
	case "unreachable":
		m.routeProbeUnreachable.Add(1)
	default:
		m.routeProbeUnknown.Add(1)
	}
}

func (m *Metrics) ObserveObjectStorage(event string, provider string, bytes int64, duration time.Duration) {
	provider = strings.ToLower(strings.TrimSpace(provider))
	if duration < 0 {
		duration = 0
	}
	switch event {
	case "write_success", "write_failure":
		var writes, failures, nanos *atomic.Uint64
		switch provider {
		case "aliyun_oss":
			writes, failures, nanos = &m.storageAliyunWrites, &m.storageAliyunFailures, &m.storageAliyunWriteNanos
		case "s3":
			writes, failures, nanos = &m.storageS3Writes, &m.storageS3Failures, &m.storageS3WriteNanos
		case "server_main_openapi":
			writes, failures, nanos = &m.storageServerMainWrites, &m.storageServerMainFailures, &m.storageServerMainWriteNanos
		default:
			return
		}
		writes.Add(1)
		nanos.Add(uint64(duration))
		if event == "write_failure" {
			failures.Add(1)
		} else if bytes > 0 {
			m.storageObjectifiedBytes.Add(uint64(bytes))
		}
	case "retry":
		m.storageChannelRetries.Add(1)
	case "failover":
		m.storageChannelFailovers.Add(1)
	case "all_failed":
		m.storageAllChannelsFailed.Add(1)
	case "read_success", "read_failure":
		m.storageObjectReads.Add(1)
		if event == "read_failure" {
			m.storageObjectReadFailures.Add(1)
		} else if bytes > 0 {
			m.storageObjectReadBytes.Add(uint64(bytes))
		}
	}
}

func (m *Metrics) ObserveResultStorage(source string) {
	switch strings.ToLower(strings.TrimSpace(source)) {
	case "upstream_url":
		m.resultSourceUpstreamURL.Add(1)
	case "uploaded":
		m.resultSourceUploaded.Add(1)
	}
}

func (m *Metrics) ObserveResultDelivery(event string, bytes int64) {
	switch strings.ToLower(strings.TrimSpace(event)) {
	case "poll_success":
		m.resultPollResponses.Add(1)
		if bytes > 0 {
			m.resultPollResponseBytes.Add(uint64(bytes))
		}
	case "poll_oversized":
		m.resultPollOversized.Add(1)
	}
}

func (m *Metrics) ObserveResultURLSigning(outcome string) {
	switch strings.ToLower(strings.TrimSpace(outcome)) {
	case "success":
		m.resultURLSignSuccess.Add(1)
	case "failure":
		m.resultURLSignFailure.Add(1)
	}
}

func (m *Metrics) ObserveSynchronousBase64(outcome string) {
	switch strings.ToLower(strings.TrimSpace(outcome)) {
	case "success":
		m.syncBase64Success.Add(1)
	case "capacity_timeout":
		m.syncBase64CapacityTimeout.Add(1)
	case "too_large":
		m.syncBase64TooLarge.Add(1)
	}
}

func (m *Metrics) ObserveTaskAdmissionWait(wait time.Duration) {
	if wait < 0 {
		wait = 0
	}
	for index, bound := range taskAdmissionWaitBounds {
		if wait <= bound {
			m.taskAdmissionWaitBuckets[index].Add(1)
		}
	}
	m.taskAdmissionWaitBuckets[len(m.taskAdmissionWaitBuckets)-1].Add(1)
	m.taskAdmissionWaitMicros.Add(uint64(wait / time.Microsecond))
}

func (m *Metrics) ObserveAsyncWorkerResize(outcome string) {
	switch outcome {
	case "success":
		m.asyncWorkerResizeSuccess.Add(1)
	case "refresh_failed":
		m.asyncWorkerRefreshFailed.Add(1)
	case "create_failed":
		m.asyncWorkerCreateFailed.Add(1)
	case "start_failed":
		m.asyncWorkerStartFailed.Add(1)
	}
}

func (m *Metrics) ObserveConcurrencyLeaseRenewal(outcome string) {
	switch outcome {
	case "success":
		m.leaseRenewalSuccess.Add(1)
	case "lost":
		m.leaseRenewalLost.Add(1)
	default:
		m.leaseRenewalFailure.Add(1)
	}
}

func (m *Metrics) ObserveTaskEventSkip(reason string) {
	switch reason {
	case "duplicate":
		m.taskEventDuplicate.Add(1)
	case "budget_exceeded":
		m.taskEventBudgetExceeded.Add(1)
	default:
		m.taskEventUnknownType.Add(1)
	}
}

func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience string, enabled bool) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		mode := "disabled"
		var lastVerificationAt *time.Time
		if enabled {
			ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
			defer cancel()
			var err error
			mode, lastVerificationAt, err = provider.SecurityEventMetrics(ctx, issuer, audience)
			if err != nil {
				http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
				return
			}
		}
		billing := store.BillingMetricsSnapshot{}
		if billingProvider, ok := provider.(interface {
			BillingMetrics(context.Context) (store.BillingMetricsSnapshot, error)
		}); ok {
			var err error
			billing, err = billingProvider.BillingMetrics(r.Context())
			if err != nil {
				http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
				return
			}
		}
		admission := store.TaskAdmissionMetricsSnapshot{}
		if admissionProvider, ok := provider.(interface {
			TaskAdmissionMetrics(context.Context) (store.TaskAdmissionMetricsSnapshot, error)
		}); ok {
			var err error
			admission, err = admissionProvider.TaskAdmissionMetrics(r.Context())
			if err != nil {
				http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
				return
			}
		}
		postgresPool := store.PostgresPoolMetricsSnapshot{}
		if poolProvider, ok := provider.(interface {
			PostgresPoolMetrics() store.PostgresPoolMetricsSnapshot
		}); ok {
			postgresPool = poolProvider.PostgresPoolMetrics()
		}
		criticalPostgresPool := store.PostgresPoolMetricsSnapshot{}
		if poolProvider, ok := provider.(interface {
			CriticalPostgresPoolMetrics() store.PostgresPoolMetricsSnapshot
		}); ok {
			criticalPostgresPool = poolProvider.CriticalPostgresPoolMetrics()
		}
		riverPostgresPool := store.PostgresPoolMetricsSnapshot{}
		if poolProvider, ok := provider.(interface {
			RiverPostgresPoolMetrics() store.PostgresPoolMetricsSnapshot
		}); ok {
			riverPostgresPool = poolProvider.RiverPostgresPoolMetrics()
		}
		modelRateLimits := []store.ModelRateLimitStatus{}
		if rateLimitProvider, ok := provider.(interface {
			ListModelRateLimitStatuses(context.Context) ([]store.ModelRateLimitStatus, error)
		}); ok {
			var err error
			modelRateLimits, err = rateLimitProvider.ListModelRateLimitStatuses(r.Context())
			if err != nil {
				http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
				return
			}
		}
		w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
		outcomeCounters(w, "easyai_gateway_ssf_receipts_total", "Received SETs by bounded outcome.", []outcomeValue{
			{"accepted", m.accepted.Load()}, {"rejected", m.rejected.Load()}, {"duplicate", m.duplicate.Load()},
		})
		plainCounter(w, "easyai_gateway_ssf_sessions_deleted_total", "OIDC browser sessions deleted by accepted session-revoked events.", m.sessionsDeleted.Load())
		plainCounter(w, "easyai_gateway_ssf_watermark_rejections_total", "OIDC access tokens rejected by a local revocation watermark.", m.watermarkRejected.Load())
		plainCounter(w, "easyai_gateway_ssf_verifications_total", "Valid verification SETs accepted by the receiver.", m.verificationAccepted.Load())
		outcomeCounters(w, "easyai_gateway_ssf_heartbeat_requests_total", "Verification requests by bounded outcome.", []outcomeValue{
			{"accepted", m.heartbeatAccepted.Load()}, {"failed", m.heartbeatFailed.Load()},
		})
		outcomeCounters(w, "easyai_gateway_oidc_introspection_total", "OIDC introspection requests by bounded outcome.", []outcomeValue{
			{"active", m.introspectionActive.Load()}, {"inactive", m.introspectionInactive.Load()}, {"failed", m.introspectionFailed.Load()},
		})
		outcomeCounters(w, "easyai_gateway_jwks_refresh_failures_total", "JWKS refresh failures by bounded source.", []outcomeValue{
			{"ssf", m.jwksSSFFailed.Load()}, {"oidc", m.jwksOIDCFailed.Load()},
		})
		fmt.Fprintln(w, "# HELP easyai_gateway_ssf_processing_duration_seconds SSF receiver transaction processing time.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_processing_duration_seconds histogram")
		for index, bound := range processingDurationBounds {
			fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_bucket{le=\"%.2f\"} %d\n", bound.Seconds(), m.processingBuckets[index].Load())
		}
		fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_bucket{le=\"+Inf\"} %d\n", m.processingCount.Load())
		fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_sum %.6f\n", float64(m.processingNanos.Load())/float64(time.Second))
		fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_count %d\n", m.processingCount.Load())
		verificationAge := float64(0)
		if lastVerificationAt != nil {
			verificationAge = time.Since(*lastVerificationAt).Seconds()
			if verificationAge < 0 {
				verificationAge = 0
			}
		}
		fmt.Fprintln(w, "# HELP easyai_gateway_ssf_verification_age_seconds Seconds since the last matched verification SET.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_verification_age_seconds gauge")
		fmt.Fprintf(w, "easyai_gateway_ssf_verification_age_seconds %.6f\n", verificationAge)
		fmt.Fprintln(w, "# HELP easyai_gateway_ssf_mode Current SSF receiver mode as a one-hot gauge.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_mode gauge")
		for _, modeName := range []string{"disabled", "bootstrap", "push_healthy", "introspection_fallback"} {
			value := 0
			if mode == modeName {
				value = 1
			}
			fmt.Fprintf(w, "easyai_gateway_ssf_mode{mode=\"%s\"} %d\n", modeName, value)
		}
		fmt.Fprintln(w, "# HELP easyai_gateway_billing_settlement_backlog Current unsettled Outbox records.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_billing_settlement_backlog gauge")
		fmt.Fprintf(w, "easyai_gateway_billing_settlement_backlog %d\n", billing.SettlementBacklog)
		fmt.Fprintln(w, "# HELP easyai_gateway_billing_settlement_delay_seconds Age of the oldest unsettled Outbox record.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_billing_settlement_delay_seconds gauge")
		fmt.Fprintf(w, "easyai_gateway_billing_settlement_delay_seconds %.6f\n", billing.SettlementDelaySecs)
		fmt.Fprintln(w, "# HELP easyai_gateway_billing_manual_review Current billing records requiring manual review.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_billing_manual_review gauge")
		fmt.Fprintf(w, "easyai_gateway_billing_manual_review %d\n", billing.ManualReview)
		fmt.Fprintln(w, "# HELP easyai_gateway_billing_orphan_frozen Wallets whose frozen amount differs from active reservations.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_billing_orphan_frozen gauge")
		fmt.Fprintf(w, "easyai_gateway_billing_orphan_frozen %d\n", billing.OrphanFrozen)
		fmt.Fprintln(w, "# HELP easyai_gateway_billing_pricing_unavailable_tasks Current tasks rejected because no effective price was available.")
		fmt.Fprintln(w, "# TYPE easyai_gateway_billing_pricing_unavailable_tasks gauge")
		fmt.Fprintf(w, "easyai_gateway_billing_pricing_unavailable_tasks %d\n", billing.PricingUnavailable)
		plainCounter(w, "easyai_gateway_billing_settlements_completed_total", "Billing settlements completed by this process.", m.billingSettlementCompleted.Load())
		plainCounter(w, "easyai_gateway_billing_settlement_retries_total", "Billing settlement retries scheduled by this process.", m.billingSettlementRetry.Load())
		plainCounter(w, "easyai_gateway_billing_manual_review_transitions_total", "Billing records transitioned to manual review by this process.", m.billingManualReview.Load())
		plainCounter(w, "easyai_gateway_billing_estimate_failures_total", "Pricing estimate requests that failed.", m.billingEstimateFailed.Load())
		plainCounter(w, "easyai_gateway_billing_idempotent_replays_total", "Generation requests replayed idempotently.", m.billingIdempotentReplay.Load())
		plainCounter(w, "easyai_gateway_billing_pricing_unavailable_total", "Pricing requests rejected because no effective price was available.", m.billingPricingUnavailable.Load())
		plainGauge(w, "easyai_gateway_async_worker_capacity", "Current River asynchronous worker capacity.", m.asyncWorkerCapacity.Load())
		plainGauge(w, "easyai_gateway_async_worker_desired_capacity", "Uncapped asynchronous worker capacity derived from model and user-group policies.", m.asyncWorkerDesiredCapacity.Load())
		plainGauge(w, "easyai_gateway_async_worker_hard_limit", "Per-process asynchronous worker safety limit.", m.asyncWorkerHardLimit.Load())
		plainGauge(w, "easyai_gateway_async_worker_capacity_capped", "Whether desired asynchronous worker capacity is capped by the hard limit.", m.asyncWorkerCapacityCapped.Load())
		plainGauge(w, "easyai_gateway_worker_active_instances", "Active distributed worker instances with a fresh heartbeat.", m.asyncWorkerActiveInstances.Load())
		plainGauge(w, "easyai_gateway_worker_global_capacity", "Global asynchronous execution capacity before instance allocation.", m.asyncWorkerGlobalCapacity.Load())
		plainGauge(w, "easyai_gateway_worker_allocated_capacity", "Asynchronous execution capacity allocated to this worker instance.", m.asyncWorkerAllocated.Load())
		plainGauge(w, "easyai_gateway_worker_safe_capacity", "Locally resource-safe asynchronous task capacity.", m.workerSafeCapacity.Load())
		plainGauge(w, "easyai_gateway_worker_heavy_capacity", "Locally resource-safe preparing and finalizing capacity.", m.workerHeavyCapacity.Load())
		plainGauge(w, "easyai_gateway_worker_active_tasks", "Tasks currently owned by this Worker process.", m.workerActiveTasks.Load())
		plainGauge(w, "easyai_gateway_worker_preparing_tasks", "Tasks in the local preparing phase.", m.workerPreparingTasks.Load())
		plainGauge(w, "easyai_gateway_worker_waiting_upstream_tasks", "Tasks waiting for an upstream result.", m.workerWaitingTasks.Load())
		plainGauge(w, "easyai_gateway_worker_finalizing_tasks", "Tasks in the local finalizing phase.", m.workerFinalizingTasks.Load())
		plainGauge(w, "easyai_gateway_worker_pressure_state", "Local Worker pressure state: -1 unknown, 0 normal, 1 busy, 2 critical.", m.workerPressureState.Load())
		outcomeCounters(w, "easyai_gateway_provider_quota_waits_total", "Tasks delayed before an upstream call by a cluster-wide provider quota.", []outcomeValue{
			{"rpm", m.providerQuotaWaitRPM.Load()},
			{"tpm", m.providerQuotaWaitTPM.Load()},
			{"concurrent", m.providerQuotaWaitConcurrent.Load()},
			{"other", m.providerQuotaWaitOther.Load()},
		})
		outcomeCounters(w, "easyai_gateway_candidate_routing_total", "Candidate routing decisions by bounded reason.", []outcomeValue{
			{"normal_rotation", m.candidateNormalRotation.Load()},
			{"full_avoided", m.candidateFullAvoided.Load()},
			{"quota_race_rotated", m.candidateQuotaRaceRotated.Load()},
			{"cooldown_skipped", m.candidateCooldownSkipped.Load()},
			{"disabled_skipped", m.candidateDisabledSkipped.Load()},
			{"all_full_queued", m.candidateAllFullQueued.Load()},
			{"other", m.candidateRoutingOther.Load()},
		})
		outcomeCounters(w, "easyai_gateway_execution_pool_routing_total", "Execution-pool routing decisions by bounded outcome.", []outcomeValue{
			{"selected", m.executionPoolSelected.Load()},
			{"shadow", m.executionPoolShadow.Load()},
			{"unavailable", m.executionPoolUnavailable.Load()},
			{"capacity_rejected", m.executionPoolCapacityReject.Load()},
		})
		outcomeCounters(w, "easyai_gateway_route_probe_total", "Non-billable route probes by resulting state.", []outcomeValue{
			{"healthy", m.routeProbeHealthy.Load()},
			{"degraded", m.routeProbeDegraded.Load()},
			{"unreachable", m.routeProbeUnreachable.Load()},
			{"unknown", m.routeProbeUnknown.Load()},
		})
		storageProtocolCounters(w, "easyai_gateway_storage_write_attempts_total", "Object storage write attempts by bounded protocol.",
			m.storageAliyunWrites.Load(), m.storageS3Writes.Load(), m.storageServerMainWrites.Load())
		storageProtocolCounters(w, "easyai_gateway_storage_write_failures_total", "Object storage write failures by bounded protocol.",
			m.storageAliyunFailures.Load(), m.storageS3Failures.Load(), m.storageServerMainFailures.Load())
		storageProtocolDurationCounters(w,
			m.storageAliyunWriteNanos.Load(), m.storageS3WriteNanos.Load(), m.storageServerMainWriteNanos.Load())
		plainCounter(w, "easyai_gateway_storage_channel_retries_total", "Retries performed within the current storage channel.", m.storageChannelRetries.Load())
		plainCounter(w, "easyai_gateway_storage_channel_failovers_total", "Switches to the next configured storage channel.", m.storageChannelFailovers.Load())
		plainCounter(w, "easyai_gateway_storage_all_channels_failed_total", "Storage writes where every eligible channel failed.", m.storageAllChannelsFailed.Load())
		plainCounter(w, "easyai_gateway_storage_objectified_bytes_total", "Binary bytes successfully written through storage channels.", m.storageObjectifiedBytes.Load())
		plainCounter(w, "easyai_gateway_storage_object_get_attempts_total", "Object storage GET attempts, including synchronous compatibility hydration.", m.storageObjectReads.Load())
		plainCounter(w, "easyai_gateway_storage_object_get_failures_total", "Failed object storage GET attempts.", m.storageObjectReadFailures.Load())
		plainCounter(w, "easyai_gateway_storage_object_get_bytes_total", "Bytes read from object storage GET responses.", m.storageObjectReadBytes.Load())
		outcomeCounters(w, "easyai_gateway_result_storage_total", "Generated media results by canonical storage source.", []outcomeValue{
			{"upstream_url", m.resultSourceUpstreamURL.Load()},
			{"uploaded", m.resultSourceUploaded.Load()},
		})
		plainCounter(w, "easyai_gateway_result_poll_responses_total", "Successful URL-only task result responses.", m.resultPollResponses.Load())
		plainCounter(w, "easyai_gateway_result_poll_response_bytes_total", "Bytes written by successful URL-only task result responses.", m.resultPollResponseBytes.Load())
		plainCounter(w, "easyai_gateway_result_poll_oversized_total", "Task result responses rejected by the URL-only size gate.", m.resultPollOversized.Load())
		outcomeCounters(w, "easyai_gateway_result_url_signing_total", "Private result URL signing attempts by outcome.", []outcomeValue{
			{"success", m.resultURLSignSuccess.Load()},
			{"failure", m.resultURLSignFailure.Load()},
		})
		outcomeCounters(w, "easyai_gateway_sync_base64_responses_total", "Synchronous Base64 compatibility responses by bounded outcome.", []outcomeValue{
			{"success", m.syncBase64Success.Load()},
			{"capacity_timeout", m.syncBase64CapacityTimeout.Load()},
			{"too_large", m.syncBase64TooLarge.Load()},
		})
		publicErrorCounters(w, publicerror.MetricSnapshot())
		platformModelRateLimitUtilizationGauges(w, modelRateLimits)
		plainGauge(w, "easyai_gateway_postgres_pool_max_connections", "Maximum PostgreSQL connections in this process pool.", int64(postgresPool.MaxConnections))
		plainGauge(w, "easyai_gateway_postgres_pool_total_connections", "Current PostgreSQL connections in this process pool.", int64(postgresPool.TotalConnections))
		plainGauge(w, "easyai_gateway_postgres_pool_acquired_connections", "Currently acquired PostgreSQL connections in this process pool.", int64(postgresPool.AcquiredConnections))
		plainGauge(w, "easyai_gateway_postgres_pool_idle_connections", "Currently idle PostgreSQL connections in this process pool.", int64(postgresPool.IdleConnections))
		plainCounter(w, "easyai_gateway_postgres_pool_empty_acquire_total", "Pool acquires that had to wait for a PostgreSQL connection.", uint64(max(postgresPool.EmptyAcquireCount, 0)))
		plainCounter(w, "easyai_gateway_postgres_pool_canceled_acquire_total", "Canceled PostgreSQL pool acquires.", uint64(max(postgresPool.CanceledAcquireCount, 0)))
		plainGauge(w, "easyai_gateway_postgres_critical_pool_max_connections", "Maximum PostgreSQL connections reserved for leases and coordination.", int64(criticalPostgresPool.MaxConnections))
		plainGauge(w, "easyai_gateway_postgres_critical_pool_total_connections", "Current PostgreSQL connections in the leases and coordination pool.", int64(criticalPostgresPool.TotalConnections))
		plainGauge(w, "easyai_gateway_postgres_critical_pool_acquired_connections", "Currently acquired PostgreSQL connections in the leases and coordination pool.", int64(criticalPostgresPool.AcquiredConnections))
		plainGauge(w, "easyai_gateway_postgres_critical_pool_idle_connections", "Currently idle PostgreSQL connections in the leases and coordination pool.", int64(criticalPostgresPool.IdleConnections))
		plainCounter(w, "easyai_gateway_postgres_critical_pool_empty_acquire_total", "Coordination pool acquires that had to wait for a PostgreSQL connection.", uint64(max(criticalPostgresPool.EmptyAcquireCount, 0)))
		plainCounter(w, "easyai_gateway_postgres_critical_pool_canceled_acquire_total", "Canceled PostgreSQL coordination pool acquires.", uint64(max(criticalPostgresPool.CanceledAcquireCount, 0)))
		plainGauge(w, "easyai_gateway_postgres_river_pool_max_connections", "Maximum PostgreSQL connections reserved for River queue coordination.", int64(riverPostgresPool.MaxConnections))
		plainGauge(w, "easyai_gateway_postgres_river_pool_total_connections", "Current PostgreSQL connections in the River queue pool.", int64(riverPostgresPool.TotalConnections))
		plainGauge(w, "easyai_gateway_postgres_river_pool_acquired_connections", "Currently acquired PostgreSQL connections in the River queue pool.", int64(riverPostgresPool.AcquiredConnections))
		plainGauge(w, "easyai_gateway_postgres_river_pool_idle_connections", "Currently idle PostgreSQL connections in the River queue pool.", int64(riverPostgresPool.IdleConnections))
		plainCounter(w, "easyai_gateway_postgres_river_pool_empty_acquire_total", "River queue pool acquires that had to wait for a PostgreSQL connection.", uint64(max(riverPostgresPool.EmptyAcquireCount, 0)))
		plainCounter(w, "easyai_gateway_postgres_river_pool_canceled_acquire_total", "Canceled PostgreSQL River queue pool acquires.", uint64(max(riverPostgresPool.CanceledAcquireCount, 0)))
		outcomeCounters(w, "easyai_gateway_async_worker_resizes_total", "Asynchronous worker resize attempts by bounded outcome.", []outcomeValue{
			{"success", m.asyncWorkerResizeSuccess.Load()},
			{"refresh_failed", m.asyncWorkerRefreshFailed.Load()},
			{"create_failed", m.asyncWorkerCreateFailed.Load()},
			{"start_failed", m.asyncWorkerStartFailed.Load()},
		})
		outcomeCounters(w, "easyai_gateway_concurrency_lease_renewals_total", "Concurrency lease renewals by bounded outcome.", []outcomeValue{
			{"success", m.leaseRenewalSuccess.Load()},
			{"failure", m.leaseRenewalFailure.Load()},
			{"lost", m.leaseRenewalLost.Load()},
		})
		outcomeCounters(w, "easyai_gateway_task_events_skipped_total", "Task events skipped before persistence by bounded reason.", []outcomeValue{
			{"duplicate", m.taskEventDuplicate.Load()},
			{"unknown_type", m.taskEventUnknownType.Load()},
			{"budget_exceeded", m.taskEventBudgetExceeded.Load()},
		})
		plainGauge(w, "easyai_gateway_task_admission_queue_depth", "Current persistent non-text task admission queue depth.", int64(admission.QueueDepth))
		plainGauge(w, "easyai_gateway_task_admission_waiting_sync", "Current synchronous admission waiters.", int64(admission.WaitingSync))
		plainGauge(w, "easyai_gateway_task_admission_waiting_async", "Current asynchronous admission waiters.", int64(admission.WaitingAsync))
		plainGauge(w, "easyai_gateway_task_admission_deferred_async", "Asynchronous admissions quarantined until a future dispatch attempt.", int64(admission.DeferredAsync))
		plainGauge(w, "easyai_gateway_task_admission_pending_reselect", "Waiting asynchronous admissions marked for candidate reselection.", int64(admission.PendingReselect))
		plainFloatGauge(w, "easyai_gateway_task_admission_oldest_wait_seconds", "Age of the oldest persistent admission waiter.", admission.OldestWaitSeconds)
		plainFloatGauge(w, "easyai_gateway_task_admission_oldest_dispatchable_wait_seconds", "Age of the oldest asynchronous admission currently eligible for dispatch.", admission.OldestDispatchableWaitSeconds)
		plainGauge(w, "easyai_gateway_task_admission_expired_waiter_backlog", "Expired synchronous waiter leases awaiting reclamation.", int64(admission.ExpiredWaiterBacklog))
		plainGauge(w, "easyai_gateway_task_admission_expired_deadline_backlog", "Expired queue deadlines awaiting reclamation.", int64(admission.ExpiredDeadlineBacklog))
		outcomeCounters(w, "easyai_gateway_task_admissions_total", "Task admission transitions by bounded outcome.", []outcomeValue{
			{"admitted", m.taskAdmissionAdmitted.Load()},
			{"queue_full", m.taskAdmissionQueueFull.Load()},
			{"timeout", m.taskAdmissionTimeout.Load()},
			{"cancelled", m.taskAdmissionCancelled.Load()},
			{"expired", m.taskAdmissionExpired.Load()},
			{"candidate_migrated", m.taskAdmissionMigrated.Load()},
		})
		outcomeCounters(w, "easyai_gateway_async_admission_dispatch_total", "Asynchronous admission dispatcher decisions by bounded outcome.", []outcomeValue{
			{"admitted", m.asyncAdmissionAdmitted.Load()},
			{"global_wait", m.asyncAdmissionGlobalWait.Load()},
			{"platform_wait", m.asyncAdmissionPlatformWait.Load()},
			{"group_wait", m.asyncAdmissionGroupWait.Load()},
			{"reselect", m.asyncAdmissionReselect.Load()},
			{"deferred", m.asyncAdmissionDeferred.Load()},
			{"terminal", m.asyncAdmissionTerminal.Load()},
			{"task_error", m.asyncAdmissionTaskError.Load()},
			{"system_error", m.asyncAdmissionSystemError.Load()},
		})
		taskAdmissionWaitHistogram(w, m)
	})
}

func (m *Metrics) DynamicHandler(provider DynamicMetricsSnapshotProvider) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		connection, err := provider.SecurityEventConnection(r.Context())
		if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
			m.Handler(provider, "", "", false).ServeHTTP(w, r)
			return
		}
		if err != nil {
			http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
			return
		}
		if connection.Audience == nil {
			m.Handler(provider, "", "", false).ServeHTTP(w, r)
			return
		}
		m.Handler(provider, connection.TransmitterIssuer, *connection.Audience, true).ServeHTTP(w, r)
	})
}

type outcomeValue struct {
	name  string
	value uint64
}

func outcomeCounters(w http.ResponseWriter, name, help string, values []outcomeValue) {
	fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n", name, help, name)
	for _, value := range values {
		fmt.Fprintf(w, "%s{outcome=\"%s\"} %d\n", name, value.name, value.value)
	}
}

func plainCounter(w http.ResponseWriter, name, help string, value uint64) {
	fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n%s %d\n", name, help, name, name, value)
}

func plainGauge(w http.ResponseWriter, name, help string, value int64) {
	fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s gauge\n%s %d\n", name, help, name, name, value)
}

func plainFloatGauge(w http.ResponseWriter, name, help string, value float64) {
	fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s gauge\n%s %.6f\n", name, help, name, name, value)
}

func storageProtocolCounters(w http.ResponseWriter, name, help string, aliyunOSS, s3, serverMain uint64) {
	fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n", name, help, name)
	fmt.Fprintf(w, "%s{protocol=\"aliyun_oss\"} %d\n", name, aliyunOSS)
	fmt.Fprintf(w, "%s{protocol=\"s3\"} %d\n", name, s3)
	fmt.Fprintf(w, "%s{protocol=\"server_main_openapi\"} %d\n", name, serverMain)
}

func storageProtocolDurationCounters(w http.ResponseWriter, aliyunOSS, s3, serverMain uint64) {
	const name = "easyai_gateway_storage_write_duration_seconds_total"
	fmt.Fprintf(w, "# HELP %s Total time spent in object storage write attempts by bounded protocol.\n# TYPE %s counter\n", name, name)
	fmt.Fprintf(w, "%s{protocol=\"aliyun_oss\"} %.6f\n", name, float64(aliyunOSS)/float64(time.Second))
	fmt.Fprintf(w, "%s{protocol=\"s3\"} %.6f\n", name, float64(s3)/float64(time.Second))
	fmt.Fprintf(w, "%s{protocol=\"server_main_openapi\"} %.6f\n", name, float64(serverMain)/float64(time.Second))
}

func publicErrorCounters(w http.ResponseWriter, values []publicerror.MetricCount) {
	const name = "easyai_gateway_public_errors_total"
	fmt.Fprintf(w, "# HELP %s Public error responses by bounded stable code.\n# TYPE %s counter\n", name, name)
	for _, value := range values {
		fmt.Fprintf(w, "%s{code=\"%s\"} %d\n", name, value.Code, value.Count)
	}
}

func platformModelRateLimitUtilizationGauges(w http.ResponseWriter, statuses []store.ModelRateLimitStatus) {
	const name = "easyai_gateway_platform_model_rate_limit_utilization"
	fmt.Fprintf(w, "# HELP %s Cluster-wide platform model quota utilization.\n# TYPE %s gauge\n", name, name)
	for _, status := range statuses {
		platformModelID := escapePrometheusLabel(status.PlatformModelID)
		for _, metric := range []struct {
			name  string
			ratio float64
		}{
			{name: "rpm", ratio: status.RPM.Ratio},
			{name: "tpm", ratio: status.TPM.Ratio},
			{name: "concurrent", ratio: status.Concurrent.Ratio},
		} {
			fmt.Fprintf(w, "%s{platform_model_id=\"%s\",metric=\"%s\"} %.6f\n", name, platformModelID, metric.name, metric.ratio)
		}
	}
}

func escapePrometheusLabel(value string) string {
	return strings.NewReplacer("\\", "\\\\", "\n", "\\n", "\"", "\\\"").Replace(value)
}

func taskAdmissionWaitHistogram(w http.ResponseWriter, metrics *Metrics) {
	const name = "easyai_gateway_task_admission_wait_seconds"
	fmt.Fprintf(w, "# HELP %s Time spent waiting for persistent task admission.\n# TYPE %s histogram\n", name, name)
	for index, bound := range taskAdmissionWaitBounds {
		fmt.Fprintf(
			w,
			"%s_bucket{le=\"%g\"} %d\n",
			name,
			bound.Seconds(),
			metrics.taskAdmissionWaitBuckets[index].Load(),
		)
	}
	count := metrics.taskAdmissionWaitBuckets[len(metrics.taskAdmissionWaitBuckets)-1].Load()
	fmt.Fprintf(w, "%s_bucket{le=\"+Inf\"} %d\n", name, count)
	fmt.Fprintf(w, "%s_sum %.6f\n", name, float64(metrics.taskAdmissionWaitMicros.Load())/1_000_000)
	fmt.Fprintf(w, "%s_count %d\n", name, count)
}
