package server

import (
	"context"
	"crypto/subtle"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"

	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/tenant/token"
)

type quotaRequest struct {
	TenantID   string `json:"tenant_id"`
	PublicKey  string `json:"public_key"`
	PrivateKey string `json:"private_key"`
	quotaFields
}

type quotaFields struct {
	MaxStorageSize         *int64 `json:"max_storage_size,omitempty"`
	MaxFileSize            *int64 `json:"max_file_size,omitempty"`
	MaxFileCount           *int64 `json:"max_file_count,omitempty"`
	MaxMediaLLMFiles       *int64 `json:"max_media_llm_files,omitempty"`
	MaxVideoLLMFiles       *int64 `json:"max_video_llm_files,omitempty"`
	TiDBCloudSpendingLimit *int64 `json:"tidbcloud_spending_limit,omitempty"`
}

func (f quotaFields) anySet() bool {
	return f.MaxStorageSize != nil ||
		f.MaxFileSize != nil ||
		f.MaxFileCount != nil ||
		f.MaxMediaLLMFiles != nil ||
		f.MaxVideoLLMFiles != nil ||
		f.TiDBCloudSpendingLimit != nil
}

type quotaResponse struct {
	TenantID       string              `json:"tenant_id"`
	Provider       string              `json:"provider"`
	Status         string              `json:"status"`
	SupportsUpdate bool                `json:"supports_update"`
	Config         quotaConfigResponse `json:"config"`
	Usage          quotaUsageResponse  `json:"usage"`
}

type quotaConfigResponse struct {
	MaxStorageSize         int64   `json:"max_storage_size"`
	MaxFileSize            int64   `json:"max_file_size"`
	MaxFileCount           int64   `json:"max_file_count"`
	MaxMediaLLMFiles       int64   `json:"max_media_llm_files"`
	MaxVideoLLMFiles       int64   `json:"max_video_llm_files"`
	TiDBCloudSpendingLimit **int64 `json:"tidbcloud_spending_limit,omitempty"`
}

type quotaUsageResponse struct {
	StorageBytes   int64 `json:"storage_bytes"`
	ReservedBytes  int64 `json:"reserved_bytes"`
	FileCount      int64 `json:"file_count"`
	MediaFileCount int64 `json:"media_file_count"`
	VideoFileCount int64 `json:"video_file_count"`
}

const (
	quotaTenantIDQueryParam = "tenant_id"
	quotaTenantIDHeader     = "X-Drive9-Tenant-ID"
	quotaPublicKeyHeader    = "X-TiDBCloud-Public-Key"
	quotaPrivateKeyHeader   = "X-TiDBCloud-Private-Key"
)

func (s *Server) quotaRootHandler(_ Config) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.Method {
		case http.MethodGet:
			s.handleQuotaGet(w, r)
		case http.MethodPost:
			s.handleQuotaSet(w, r)
		default:
			writeControlPlaneStatusError(w, http.StatusMethodNotAllowed, "method not allowed")
		}
	})
}

func (s *Server) handleQuotaGet(w http.ResponseWriter, r *http.Request) {
	if s.meta == nil || s.provisioner == nil {
		writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "quota query not enabled")
		return
	}
	req := decodeQuotaGetRequest(r)
	t, ok := s.quotaTenant(w, r.Context(), req.TenantID)
	if !ok {
		return
	}
	if !tenant.IsSharedSchemaProvider(t.Provider) && strings.TrimSpace(t.ClusterID) == "" {
		writeControlPlaneError(w, http.StatusNotFound, "quota_backend_not_found", quotaBackendNotFoundMessage)
		return
	}
	if bearerToken(r) != "" {
		apiKeyTenant, apiKeyErr := s.resolveQuotaAPIKey(r.Context(), r)
		if apiKeyErr == nil && apiKeyTenant != nil && apiKeyTenant.Tenant.ID == t.ID {
			setRequestMetricTenant(r.Context(), t.ID, apiKeyTenant.APIKey.ID, t.Provider, apiKeyTenant.TiDBCloudOrgID, classifyTenantRequest(r))
			s.writeQuotaResponse(w, r, t)
			return
		}
		if apiKeyErr != nil {
			writeQuotaAPIKeyError(w, r.Context(), apiKeyErr)
			return
		}
		writeControlPlaneStatusError(w, http.StatusForbidden, "API key tenant does not match requested tenant")
		return
	}
	if t.Provider == tenant.ProviderTiDBCloudNativeShared {
		cred, err := controlPlaneCredentials(r, req.PublicKey, req.PrivateKey)
		if err != nil {
			writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
			return
		}
		physical, err := s.authorizeSharedQuotaCredentials(r.Context(), t, cred, "quota_get")
		if err != nil {
			writeQuotaCredentialError(w, r.Context(), err, "query")
			return
		}
		setRequestMetricTenant(r.Context(), t.ID, "", t.Provider, physical.TiDBCloudOrganizationID, classifyTenantRequest(r))
		s.writeQuotaResponse(w, r, t)
		return
	}
	if t.Provider != tenant.ProviderTiDBCloudNative {
		writeControlPlaneStatusError(w, http.StatusConflict, "tidbcloud credential quota query is only supported for tidb_cloud_native tenants")
		return
	}
	cred, err := controlPlaneCredentials(r, req.PublicKey, req.PrivateKey)
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	binding, err := s.authorizeNativeTenantCredentials(r.Context(), t, cred, "quota_get")
	if err != nil {
		writeQuotaCredentialError(w, r.Context(), err, "query")
		return
	}
	setRequestMetricTenant(r.Context(), t.ID, "", t.Provider, binding.OrganizationID, classifyTenantRequest(r))
	s.writeQuotaResponse(w, r, t)
}

var (
	errQuotaAPIKeyAuthNotEnabled = errors.New("API key quota auth not enabled")
	errQuotaAPIKeyMissing        = errors.New("missing or malformed Authorization header")
	errQuotaAPIKeyInvalid        = errors.New("invalid API key")
)

func (s *Server) resolveQuotaAPIKey(ctx context.Context, r *http.Request) (*meta.TenantWithAPIKey, error) {
	if s.meta == nil || s.pool == nil || len(s.tokenSecret) == 0 {
		return nil, errQuotaAPIKeyAuthNotEnabled
	}
	tok := bearerToken(r)
	if tok == "" {
		return nil, errQuotaAPIKeyMissing
	}
	hash := token.HashToken(tok)
	resolved, err := s.meta.ResolveByAPIKeyHash(ctx, hash)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			return nil, errQuotaAPIKeyInvalid
		}
		return nil, fmt.Errorf("resolve API key: %w", err)
	}
	if subtle.ConstantTimeCompare([]byte(hash), []byte(resolved.APIKey.JWTHash)) != 1 {
		return nil, errQuotaAPIKeyInvalid
	}
	if resolved.APIKey.Status != meta.APIKeyActive {
		return nil, errQuotaAPIKeyInvalid
	}
	plain, err := poolDecryptToken(ctx, s.pool, resolved.APIKey.JWTCiphertext)
	if err != nil {
		return nil, fmt.Errorf("decrypt API key: %w", err)
	}
	if subtle.ConstantTimeCompare([]byte(tok), plain) != 1 {
		return nil, errQuotaAPIKeyInvalid
	}
	claims, err := token.ParseAndVerifyToken(s.tokenSecret, tok)
	if err != nil {
		return nil, errQuotaAPIKeyInvalid
	}
	if claims.TenantID != resolved.Tenant.ID || claims.TokenVersion != resolved.APIKey.TokenVersion {
		return nil, errQuotaAPIKeyInvalid
	}
	return resolved, nil
}

func writeQuotaAPIKeyError(w http.ResponseWriter, ctx context.Context, err error) {
	switch {
	case errors.Is(err, errQuotaAPIKeyMissing), errors.Is(err, errQuotaAPIKeyInvalid):
		writeControlPlaneError(w, http.StatusUnauthorized, controlPlaneErrorCode(http.StatusUnauthorized, err), errQuotaAPIKeyInvalid.Error())
	case errors.Is(err, errQuotaAPIKeyAuthNotEnabled):
		writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "quota query not enabled")
	default:
		logger.Warn(ctx, "quota_api_key_auth_failed", zap.Error(err))
		status := backendErrorStatus(ctx, err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "auth backend unavailable")
	}
}

func decodeQuotaGetRequest(r *http.Request) quotaRequest {
	q := r.URL.Query()
	return quotaRequest{
		TenantID:   strings.TrimSpace(firstNonEmpty(q.Get(quotaTenantIDQueryParam), r.Header.Get(quotaTenantIDHeader))),
		PublicKey:  strings.TrimSpace(r.Header.Get(quotaPublicKeyHeader)),
		PrivateKey: strings.TrimSpace(r.Header.Get(quotaPrivateKeyHeader)),
	}
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}

func (s *Server) handleQuotaSet(w http.ResponseWriter, r *http.Request) {
	if s.meta == nil || s.provisioner == nil {
		writeControlPlaneError(w, http.StatusNotFound, "not_enabled", "quota setting not enabled")
		return
	}
	req, err := decodeQuotaRequest(w, r)
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	cred, err := controlPlaneCredentials(r, req.PublicKey, req.PrivateKey)
	if err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	access, err := s.authorizeTiDBCloudQuotaMutation(r.Context(), cred, "quota_set")
	if err != nil {
		writeQuotaSetError(w, r.Context(), err, "authorize")
		return
	}
	if err := s.validateQuotaSetRequestWithoutSpendingLimit(req); err != nil {
		writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
		return
	}
	t, ok := s.quotaTenant(w, r.Context(), req.TenantID)
	if !ok {
		return
	}
	if t.Provider != tenant.ProviderTiDBCloudNative && t.Provider != tenant.ProviderTiDBCloudNativeShared {
		writeControlPlaneStatusError(w, http.StatusConflict, "quota setting is only supported for tidb_cloud_native tenants")
		return
	}
	if err := s.rejectQuotaSetForTenantStatus(t); err != nil {
		writeControlPlaneError(w, http.StatusConflict, controlPlaneErrorCode(http.StatusConflict, err), err.Error())
		return
	}
	var sharedPhysical *meta.SharedDB
	if t.Provider == tenant.ProviderTiDBCloudNativeShared {
		req = *quotaRequestForShared(&req)
		physical, err := s.authorizeSharedQuotaOrganization(r.Context(), t, access.OrganizationID)
		if err != nil {
			writeQuotaSetError(w, r.Context(), err, "authorize")
			return
		}
		if err := s.applySharedQuotaSet(r.Context(), t, req); err != nil {
			writeQuotaSetError(w, r.Context(), err, "update")
			return
		}
		sharedPhysical = physical
	} else {
		if err := validateTiDBCloudSpendingLimit(req.TiDBCloudSpendingLimit); err != nil {
			writeControlPlaneError(w, http.StatusBadRequest, controlPlaneErrorCode(http.StatusBadRequest, err), err.Error())
			return
		}
		if _, err := s.authorizeNativeTenantOrganization(r.Context(), t, access.OrganizationID); err != nil {
			writeQuotaSetError(w, r.Context(), err, "authorize")
			return
		}
		if err := s.applyQuotaSet(r.Context(), t, cred, req); err != nil {
			writeQuotaSetError(w, r.Context(), err, "update")
			return
		}
	}
	metricOrgID := ""
	if t.Provider == tenant.ProviderTiDBCloudNativeShared {
		// The shared authorization path already resolved the physical pool;
		// reuse its organization label instead of issuing another meta query.
		metricOrgID = sharedPhysical.TiDBCloudOrganizationID
	} else {
		metricOrgID = s.tenantMetricTiDBCloudOrgID(r.Context(), t)
	}
	setRequestMetricTenant(r.Context(), t.ID, "", t.Provider, metricOrgID, classifyTenantRequest(r))
	s.writeQuotaResponse(w, r, t)
}

func (s *Server) authorizeSharedQuotaCredentials(ctx context.Context, t *meta.Tenant, cred tenant.CredentialProvisionRequest, metricPath string) (*meta.SharedDB, error) {
	physical, err := s.meta.GetSharedDBForTenant(ctx, t.ID)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			return nil, tenant.ErrQuotaBackendNotFound
		}
		return nil, fmt.Errorf("get shared DB for tenant: %w", err)
	}
	identity, err := s.resolveTiDBCloudIdentity(ctx, cred, metricPath)
	if err != nil {
		return nil, err
	}
	if !tiDBCloudOrganizationMatches(identity.OrganizationID, physical.TiDBCloudOrganizationID) {
		return nil, tenant.ErrQuotaPermissionDenied
	}
	return physical, nil
}

func (s *Server) authorizeSharedQuotaOrganization(ctx context.Context, t *meta.Tenant, organizationID string) (*meta.SharedDB, error) {
	physical, err := s.meta.GetSharedDBForTenant(ctx, t.ID)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			return nil, tenant.ErrQuotaBackendNotFound
		}
		return nil, fmt.Errorf("get shared DB for tenant: %w", err)
	}
	if !tiDBCloudOrganizationMatches(organizationID, physical.TiDBCloudOrganizationID) {
		return nil, tenant.ErrQuotaPermissionDenied
	}
	return physical, nil
}

func (s *Server) applySharedQuotaSet(ctx context.Context, t *meta.Tenant, req quotaRequest) error {
	req = *quotaRequestForShared(&req)
	if !req.anySet() {
		return nil
	}
	patch, err := quotaConfigPatchFromRequest(req)
	if err != nil {
		return err
	}
	if err := s.meta.UpdateSharedTenantQuotaConfig(ctx, t.ID, patch); err != nil {
		return err
	}
	if err := s.meta.EnsureQuotaUsageRow(ctx, t.ID); err != nil {
		return fmt.Errorf("%w: quota usage initialization failed: %w", errQuotaLocalUpdateFailed, err)
	}
	return nil
}

func decodeQuotaRequest(w http.ResponseWriter, r *http.Request) (quotaRequest, error) {
	var req quotaRequest
	dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxCredentialProvisionBodyBytes))
	dec.DisallowUnknownFields()
	if err := dec.Decode(&req); err != nil {
		return quotaRequest{}, fmt.Errorf("invalid JSON body: %w", err)
	}
	var extra struct{}
	if err := dec.Decode(&extra); !errors.Is(err, io.EOF) {
		return quotaRequest{}, fmt.Errorf("invalid JSON body: trailing data")
	}
	req.TenantID = strings.TrimSpace(req.TenantID)
	req.PublicKey = strings.TrimSpace(req.PublicKey)
	req.PrivateKey = strings.TrimSpace(req.PrivateKey)
	return req, nil
}

func quotaCredentials(req quotaRequest) (tenant.CredentialProvisionRequest, error) {
	if req.TenantID == "" {
		return tenant.CredentialProvisionRequest{}, fmt.Errorf("tenant_id is required")
	}
	if req.PublicKey == "" && req.PrivateKey == "" {
		return tenant.CredentialProvisionRequest{}, tenant.ErrCredentialsRequired
	}
	if req.PublicKey == "" || req.PrivateKey == "" {
		return tenant.CredentialProvisionRequest{}, tenant.ErrPartialCredentials
	}
	return tenant.CredentialProvisionRequest{PublicKey: req.PublicKey, PrivateKey: req.PrivateKey}, nil
}

func (s *Server) validateQuotaSetRequest(req quotaRequest) error {
	if err := s.validateQuotaSetRequestWithoutSpendingLimit(req); err != nil {
		return err
	}
	return validateTiDBCloudSpendingLimit(req.TiDBCloudSpendingLimit)
}

func (s *Server) validateQuotaSetRequestWithoutSpendingLimit(req quotaRequest) error {
	if !req.anySet() {
		return fmt.Errorf("at least one quota limit is required")
	}
	if req.MaxStorageSize != nil && *req.MaxStorageSize <= 0 {
		return fmt.Errorf("max_storage_size must be positive")
	}
	if req.MaxStorageSize != nil {
		if _, err := quotaStorageSizeToBytes(*req.MaxStorageSize); err != nil {
			return err
		}
	}
	if req.MaxFileSize != nil && *req.MaxFileSize <= 0 {
		return fmt.Errorf("max_file_size must be positive")
	}
	if req.MaxFileSize != nil {
		maxFileSizeBytes, err := quotaFileSizeToBytes(*req.MaxFileSize)
		if err != nil {
			return err
		}
		if s.maxUploadBytes > 0 && maxFileSizeBytes > s.maxUploadBytes {
			return fmt.Errorf("max_file_size must be less than or equal to server max upload size (%dMi)", s.maxUploadBytes/quotaStorageSizeBytes)
		}
	}
	if req.MaxFileCount != nil && *req.MaxFileCount < 0 {
		return fmt.Errorf("max_file_count must be non-negative")
	}
	if req.MaxMediaLLMFiles != nil && *req.MaxMediaLLMFiles < 0 {
		return fmt.Errorf("max_media_llm_files must be non-negative")
	}
	if req.MaxVideoLLMFiles != nil && *req.MaxVideoLLMFiles < 0 {
		return fmt.Errorf("max_video_llm_files must be non-negative")
	}
	return nil
}

func validateTiDBCloudSpendingLimit(limit *int64) error {
	if limit != nil && *limit < 0 {
		return fmt.Errorf("%w: tidbcloud_spending_limit must be non-negative", errTiDBCloudSpendingLimitInvalid)
	}
	if limit != nil && *limit > 0 && *limit < 10 {
		return fmt.Errorf("%w: tidbcloud_spending_limit must be 0 or at least 10 RMB", errTiDBCloudSpendingLimitInvalid)
	}
	if limit != nil && *limit > meta.MaxTiDBCloudSpendingLimit {
		return fmt.Errorf("%w: tidbcloud_spending_limit is too large", errTiDBCloudSpendingLimitInvalid)
	}
	return nil
}

var (
	errQuotaSettingNotEnabled        = errors.New("quota setting not enabled")
	errQuotaLocalUpdateFailed        = errors.New("quota local update failed")
	errTiDBCloudSpendingLimitInvalid = errors.New("invalid TiDB Cloud spending limit")
)

func quotaRequestForShared(req *quotaRequest) *quotaRequest {
	if req == nil || req.TiDBCloudSpendingLimit == nil {
		return req
	}
	out := *req
	out.TiDBCloudSpendingLimit = nil
	return &out
}

func (s *Server) rejectQuotaSetForTenantStatus(t *meta.Tenant) error {
	if t == nil {
		return nil
	}
	switch t.Status {
	case meta.TenantProvisioning:
		return fmt.Errorf("tenant is still provisioning")
	case meta.TenantDeleting:
		return fmt.Errorf("tenant is deleting")
	default:
		return nil
	}
}

func (s *Server) applyQuotaSet(ctx context.Context, t *meta.Tenant, cred tenant.CredentialProvisionRequest, req quotaRequest) error {
	if t == nil {
		return fmt.Errorf("tenant is required")
	}
	if strings.TrimSpace(t.ClusterID) == "" {
		return tenant.ErrQuotaBackendNotFound
	}
	quotaPatch, err := quotaConfigPatchFromRequest(req)
	if err != nil {
		return err
	}
	if err := s.meta.ValidateQuotaExtractConfigs(ctx, t.ID, quotaPatch); err != nil {
		return err
	}
	updater, ok := s.provisioner.(tenant.QuotaUpdater)
	if !ok {
		return errQuotaSettingNotEnabled
	}
	_, err = updater.MarkQuotaUpdateStarted(ctx, clusterInfoFromTenant(t), cred)
	if err != nil {
		return err
	}
	if req.TiDBCloudSpendingLimit != nil {
		cloudCfg, err := updater.UpdateQuota(ctx, clusterInfoFromTenant(t), cred, tenant.QuotaUpdateOptions{
			TiDBCloudSpendingLimitMonthly: req.TiDBCloudSpendingLimit,
		})
		if err != nil {
			return err
		}
		if cloudCfg == nil || cloudCfg.TiDBCloudSpendingLimitMonthly == nil {
			return fmt.Errorf("tidbcloud spending limit update returned no value")
		}
		req.TiDBCloudSpendingLimit = cloudCfg.TiDBCloudSpendingLimitMonthly
	}
	quotaPatch.TiDBCloudSpendingLimit = req.TiDBCloudSpendingLimit
	if quotaPatch.AnySet() {
		if err := s.meta.SetQuotaConfigPatch(ctx, t.ID, quotaPatch); err != nil {
			if errors.Is(err, meta.ErrTenantExtractConfigNotEnabled) {
				return err
			}
			return fmt.Errorf("%w: quota config update failed: %w", errQuotaLocalUpdateFailed, err)
		}
		if err := s.meta.EnsureQuotaUsageRow(ctx, t.ID); err != nil {
			return fmt.Errorf("%w: quota usage initialization failed: %w", errQuotaLocalUpdateFailed, err)
		}
	}
	return nil
}

func (s *Server) applyQuotaLocalConfig(ctx context.Context, tenantID string, req quotaRequest) error {
	quotaPatch, err := quotaConfigPatchFromRequest(req)
	if err != nil {
		return err
	}
	if !quotaPatch.AnySet() {
		return nil
	}
	if err := s.meta.SetQuotaConfigPatch(ctx, tenantID, quotaPatch); err != nil {
		return fmt.Errorf("%w: quota config update failed: %w", errQuotaLocalUpdateFailed, err)
	}
	if err := s.meta.EnsureQuotaUsageRow(ctx, tenantID); err != nil {
		return fmt.Errorf("%w: quota usage initialization failed: %w", errQuotaLocalUpdateFailed, err)
	}
	return nil
}

func quotaConfigPatchFromRequest(req quotaRequest) (meta.QuotaConfigPatch, error) {
	var patch meta.QuotaConfigPatch
	if req.MaxStorageSize != nil {
		maxStorageBytes, err := quotaStorageSizeToBytes(*req.MaxStorageSize)
		if err != nil {
			return patch, err
		}
		patch.MaxStorageBytes = &maxStorageBytes
	}
	if req.MaxFileSize != nil {
		maxFileSizeBytes, err := quotaFileSizeToBytes(*req.MaxFileSize)
		if err != nil {
			return patch, err
		}
		patch.MaxFileSizeBytes = &maxFileSizeBytes
	}
	if req.MaxFileCount != nil {
		patch.MaxFileCount = req.MaxFileCount
	}
	if req.MaxMediaLLMFiles != nil {
		patch.MaxMediaLLMFiles = req.MaxMediaLLMFiles
	}
	if req.MaxVideoLLMFiles != nil {
		patch.MaxVideoLLMFiles = req.MaxVideoLLMFiles
	}
	if req.TiDBCloudSpendingLimit != nil {
		patch.TiDBCloudSpendingLimit = req.TiDBCloudSpendingLimit
	}
	return patch, nil
}

func (s *Server) quotaTenant(w http.ResponseWriter, ctx context.Context, tenantID string) (*meta.Tenant, bool) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		writeControlPlaneStatusError(w, http.StatusBadRequest, "tenant_id is required")
		return nil, false
	}
	t, err := s.meta.GetTenant(ctx, tenantID)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
			return nil, false
		}
		status := backendErrorStatus(ctx, err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "tenant lookup failed")
		return nil, false
	}
	if t.Status == meta.TenantDeleted {
		writeControlPlaneStatusError(w, http.StatusNotFound, "tenant not found")
		return nil, false
	}
	return t, true
}

func (s *Server) writeQuotaResponse(w http.ResponseWriter, r *http.Request, t *meta.Tenant) {
	cfg, err := s.meta.GetQuotaConfig(r.Context(), t.ID)
	if err != nil {
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "quota config lookup failed")
		return
	}
	usage, err := s.meta.GetQuotaUsage(r.Context(), t.ID)
	if err != nil {
		status := backendErrorStatus(r.Context(), err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), "quota usage lookup failed")
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(quotaResponse{
		TenantID:       t.ID,
		Provider:       t.Provider,
		Status:         string(t.Status),
		SupportsUpdate: tenant.UsesTiDBCloudNativeCredentials(t.Provider),
		Config:         quotaConfigResponseFromValues(t.Provider, *cfg),
		Usage: quotaUsageResponse{
			StorageBytes:   usage.StorageBytes,
			ReservedBytes:  usage.ReservedBytes,
			FileCount:      usage.FileCount,
			MediaFileCount: usage.MediaFileCount,
			VideoFileCount: usage.VideoFileCount,
		},
	})
}

func quotaConfigResponseFromValues(provider string, cfg meta.QuotaConfig) quotaConfigResponse {
	out := quotaConfigResponse{
		MaxStorageSize:   quotaStorageBytesToSize(cfg.MaxStorageBytes),
		MaxFileSize:      quotaStorageBytesToSize(cfg.MaxFileSizeBytes),
		MaxFileCount:     cfg.MaxFileCount,
		MaxMediaLLMFiles: cfg.MaxMediaLLMFiles,
		MaxVideoLLMFiles: cfg.MaxVideoLLMFiles,
	}
	if provider == tenant.ProviderTiDBCloudNative {
		spendingLimit := cfg.TiDBCloudSpendingLimit
		out.TiDBCloudSpendingLimit = &spendingLimit
	}
	return out
}

const quotaStorageSizeBytes int64 = 1024 * 1024

func quotaStorageSizeToBytes(sizeMi int64) (int64, error) {
	return quotaSizeToBytes("max_storage_size", sizeMi)
}

func quotaFileSizeToBytes(sizeMi int64) (int64, error) {
	return quotaSizeToBytes("max_file_size", sizeMi)
}

func quotaSizeToBytes(field string, sizeMi int64) (int64, error) {
	const maxInt64 = int64(1<<63 - 1)
	if sizeMi > maxInt64/quotaStorageSizeBytes {
		return 0, fmt.Errorf("%s is too large", field)
	}
	return sizeMi * quotaStorageSizeBytes, nil
}

func quotaStorageBytesToSize(sizeBytes int64) int64 {
	if sizeBytes <= 0 {
		return 0
	}
	return (sizeBytes + quotaStorageSizeBytes - 1) / quotaStorageSizeBytes
}

const quotaBackendNotFoundMessage = "backend service exception; please check TiDB Cloud starter/native cluster status"

func writeQuotaCredentialError(w http.ResponseWriter, ctx context.Context, err error, action string) {
	if errors.Is(err, tenant.ErrQuotaBackendNotFound) {
		writeControlPlaneError(w, http.StatusNotFound, "quota_backend_not_found", quotaBackendNotFoundMessage)
		return
	}
	logger.Warn(ctx, "tidbcloud_quota_failed", zap.String("action", action), zap.Error(err))
	status, msg := clientFacingErrorResponse(http.StatusBadGateway, "tidbcloud quota "+action+" failed", err)
	writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
}

func writeQuotaSetError(w http.ResponseWriter, ctx context.Context, err error, action string) {
	if isTiDBCloudBillingLookupError(err) {
		status, msg := tiDBCloudBillingErrorResponse(err)
		writeControlPlaneError(w, status, controlPlaneErrorCode(status, err), msg)
		return
	}
	if status, msg, ok := quotaSetErrorStatusMessage(err, action); ok {
		if errors.Is(err, errQuotaLocalUpdateFailed) {
			logger.Warn(ctx, "drive9_quota_update_failed", zap.String("action", action), zap.Error(err))
		}
		code := controlPlaneErrorCode(status, err)
		if errors.Is(err, errQuotaSettingNotEnabled) {
			code = "not_enabled"
		}
		writeControlPlaneError(w, status, code, msg)
		return
	}
	writeQuotaCredentialError(w, ctx, err, action)
}

func quotaSetErrorStatusMessage(err error, action string) (int, string, bool) {
	switch {
	case errors.Is(err, tenant.ErrTiDBCloudFreeQuotaMutationForbidden):
		return http.StatusPaymentRequired, freeQuotaMutationForbiddenMessage(), true
	case errors.Is(err, meta.ErrTenantExtractConfigNotEnabled):
		return http.StatusConflict, err.Error(), true
	case errors.Is(err, errQuotaSettingNotEnabled):
		return http.StatusNotFound, "quota setting not enabled", true
	case errors.Is(err, errQuotaLocalUpdateFailed):
		return http.StatusInternalServerError, "quota " + action + " failed", true
	default:
		return 0, "", false
	}
}