package handlers

import (
	"fmt"
	"io"
	"net/http"
	"net/url"
	"time"

	"github.com/gin-gonic/gin"
	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/observability"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/service"
	"go.uber.org/zap"
)

// LokiHandler handles Loki API proxy requests
type LokiHandler struct {
	clinicClient *clinic_internal_api.Client
	service      *service.LokiProxyService
}

// NewLokiHandler creates a new LokiHandler instance
func NewLokiHandler(clinicAPIBase string, svc *service.LokiProxyService) *LokiHandler {
	return &LokiHandler{
		clinicClient: clinic_internal_api.NewClientWithTimeout(clinicAPIBase, 2*time.Minute),
		service:      svc,
	}
}

// RegisterRoutes registers Loki proxy routes
func (h *LokiHandler) RegisterRoutes(r *gin.Engine) {
	loki := r.Group("/data-proxy/loki")
	{
		loki.GET("/:cluster_id/api/v1/query_range", h.ProxyQueryRange)
		loki.GET("/:cluster_id/api/v1/labels", h.ProxyLabels)
		loki.GET("/:cluster_id/api/v1/label/:name/values", h.ProxyLabelValues)
		loki.GET("/:cluster_id/api/v1/series", h.ProxySeries)
	}
	log.PlainInfo("Loki proxy routes registered", zap.String("path", "/data-proxy/loki"))

	systemLogs := r.Group("/data-proxy/system-logs")
	{
		systemLogs.GET("/api/v1/datasources", h.ProxySystemDatasources)
		systemLogs.GET("/api/v1/query_range", h.ProxySystemQueryRange)
		systemLogs.GET("/api/v1/labels", h.ProxySystemLabels)
		systemLogs.GET("/api/v1/label/:name/values", h.ProxySystemLabelValues)
		systemLogs.GET("/api/v1/series", h.ProxySystemSeries)
	}
	log.PlainInfo("System logs proxy routes registered", zap.String("path", "/data-proxy/system-logs"))

	serverlessParentLogs := r.Group("/data-proxy/serverless-parent-logs")
	{
		serverlessParentLogs.GET("/api/v1/query_range", h.ProxyServerlessParentLogsQueryRange)
		serverlessParentLogs.GET("/api/v1/labels", h.ProxyServerlessParentLogsLabels)
		serverlessParentLogs.GET("/api/v1/label/:name/values", h.ProxyServerlessParentLogsLabelValues)
		serverlessParentLogs.GET("/api/v1/series", h.ProxyServerlessParentLogsSeries)
	}
	log.PlainInfo("Serverless parent logs proxy routes registered", zap.String("path", "/data-proxy/serverless-parent-logs"))
}

// ProxyQueryRange handles /data-proxy/loki/{cluster_id}/query_range
func (h *LokiHandler) ProxyQueryRange(c *gin.Context) {
	h.proxyLokiRequest(c, "/loki/api/v1/query_range", "loki_query_range")
}

// ProxyLabels handles /data-proxy/loki/{cluster_id}/labels
func (h *LokiHandler) ProxyLabels(c *gin.Context) {
	// labels 端点也支持 query 参数来过滤流
	h.proxyLokiRequest(c, "/loki/api/v1/labels", "loki_labels")
}

// ProxyLabelValues handles /data-proxy/loki/{cluster_id}/label/{name}/values
func (h *LokiHandler) ProxyLabelValues(c *gin.Context) {
	labelName := c.Param("name")
	// label values 端点也支持 query 参数来过滤流
	h.proxyLokiRequest(c, fmt.Sprintf("/loki/api/v1/label/%s/values", labelName), "loki_label_values")
}

// ProxySeries handles /data-proxy/loki/{cluster_id}/series
func (h *LokiHandler) ProxySeries(c *gin.Context) {
	// series 端点支持 match[] 参数来选择流
	h.proxyLokiRequest(c, "/loki/api/v1/series", "loki_series")
}

// ProxySystemDatasources returns configured control-plane log datasources.
func (h *LokiHandler) ProxySystemDatasources(c *gin.Context) {
	start := time.Now()
	rec := &observability.DataProxyRecord{
		Endpoint:  "system_logs_datasources",
		Caller:    observability.CallerFromGin(c),
		QueryType: "system_logs",
	}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	c.JSON(http.StatusOK, gin.H{
		"status": "success",
		"data":   h.service.ListSystemLogDatasources(),
	})
}

// ProxySystemQueryRange handles control-plane log range queries without a business cluster id.
func (h *LokiHandler) ProxySystemQueryRange(c *gin.Context) {
	h.proxySystemLogsRequest(c, "/loki/api/v1/query_range", "system_logs_query_range")
}

// ProxySystemLabels handles control-plane Loki label discovery.
func (h *LokiHandler) ProxySystemLabels(c *gin.Context) {
	h.proxySystemLogsRequest(c, "/loki/api/v1/labels", "system_logs_labels")
}

// ProxySystemLabelValues handles control-plane Loki label values.
func (h *LokiHandler) ProxySystemLabelValues(c *gin.Context) {
	labelName := c.Param("name")
	h.proxySystemLogsRequest(c, fmt.Sprintf("/loki/api/v1/label/%s/values", labelName), "system_logs_label_values")
}

// ProxySystemSeries handles control-plane Loki series discovery.
func (h *LokiHandler) ProxySystemSeries(c *gin.Context) {
	h.proxySystemLogsRequest(c, "/loki/api/v1/series", "system_logs_series")
}

// ProxyServerlessParentLogsQueryRange handles parent/control-plane log range queries by control_plane_info.
func (h *LokiHandler) ProxyServerlessParentLogsQueryRange(c *gin.Context) {
	h.proxyServerlessParentLogsRequest(c, "/loki/api/v1/query_range", "serverless_parent_logs_query_range")
}

// ProxyServerlessParentLogsLabels handles parent/control-plane Loki label discovery.
func (h *LokiHandler) ProxyServerlessParentLogsLabels(c *gin.Context) {
	h.proxyServerlessParentLogsRequest(c, "/loki/api/v1/labels", "serverless_parent_logs_labels")
}

// ProxyServerlessParentLogsLabelValues handles parent/control-plane Loki label values.
func (h *LokiHandler) ProxyServerlessParentLogsLabelValues(c *gin.Context) {
	labelName := c.Param("name")
	h.proxyServerlessParentLogsRequest(c, fmt.Sprintf("/loki/api/v1/label/%s/values", labelName), "serverless_parent_logs_label_values")
}

// ProxyServerlessParentLogsSeries handles parent/control-plane Loki series discovery.
func (h *LokiHandler) ProxyServerlessParentLogsSeries(c *gin.Context) {
	h.proxyServerlessParentLogsRequest(c, "/loki/api/v1/series", "serverless_parent_logs_series")
}

// proxyLokiRequest is the common handler for all Loki proxy requests
func (h *LokiHandler) proxyLokiRequest(c *gin.Context, lokiPath string, endpointName string) {
	start := time.Now()
	rec := &observability.DataProxyRecord{
		Endpoint:  endpointName,
		Caller:    observability.CallerFromGin(c),
		QueryType: "loki",
	}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	ctx := c.Request.Context()
	clusterID := c.Param("cluster_id")

	if clusterID == "" {
		rec.ErrCode = "CLUSTER_ID_REQUIRED"
		c.JSON(http.StatusBadRequest, gin.H{
			"error": "cluster_id is required",
		})
		return
	}

	// 1. Fetch cluster Loki URL and filters from clinic API
	clusterURLs, err := h.clinicClient.GetClusterDataURLs(ctx, clusterID)
	if err != nil {
		rec.ErrCode = "CLUSTER_INFO_FETCH_FAILED"
		log.Error(ctx, "failed to fetch cluster data urls",
			zap.String("cluster_id", clusterID),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{
			"error": fmt.Sprintf("failed to fetch cluster info: %v", err),
		})
		return
	}

	rec.DeployTypeV2 = clusterURLs.DeployTypeV2
	rec.Provider = clusterURLs.Provider
	rec.Region = clusterURLs.Region

	if clusterURLs.LokiURL == "" {
		rec.ErrCode = "LOKI_URL_NOT_CONFIGURED"
		log.Warn(ctx, "loki url not available for cluster", zap.String("cluster_id", clusterID))
		c.JSON(http.StatusNotFound, gin.H{
			"error": "loki url not available for this cluster",
		})
		return
	}

	// 3. Build cluster info for LogQL label injection by deployTypeV2
	info := &service.LokiClusterInfo{
		ClusterID:         clusterURLs.ClusterID,
		ParentID:          clusterURLs.ParentID,
		DeployTypeV2:      clusterURLs.DeployTypeV2,
		ProjectID:         clusterURLs.ProjectID,
		ClusterMetricsURL: clusterURLs.ClusterMetricsURL,
		LokiURL:           clusterURLs.LokiURL,
	}

	// 4. Use service to proxy the request (injects labels into query when needed)
	var resp *http.Response
	if service.IsBYOCLogsCluster(clusterURLs.DeployType, clusterURLs.DeployTypeV2) {
		info.TenantID = clusterURLs.ClinicTenantID
		info.Provider = clusterURLs.Provider
		info.Region = clusterURLs.Region
		info.DeployType = clusterURLs.DeployType
		resp, err = h.service.ProxyBYOCRequest(
			ctx,
			lokiPath,
			c.Request.URL,
			c.Request.Method,
			c.Request.Body,
			c.Request.Header,
			info,
		)
	} else {
		resp, err = h.service.ProxyRequest(
			ctx,
			lokiPath,
			c.Request.URL,
			c.Request.Method,
			c.Request.Body,
			c.Request.Header,
			info,
		)
	}
	if err != nil {
		rec.ErrCode = "LOKI_PROXY_FAILED"
		log.Error(ctx, "failed to proxy loki request",
			zap.String("cluster_id", clusterID),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{
			"error": err.Error(),
		})
		return
	}
	defer resp.Body.Close()

	if err := h.writeLokiResponse(c, resp); err != nil {
		rec.ErrCode = "LOKI_RESPONSE_COPY_FAILED"
		log.Error(ctx, "failed to copy loki response",
			zap.String("cluster_id", clusterID),
			zap.Error(err))
	}
}

func (h *LokiHandler) proxySystemLogsRequest(c *gin.Context, lokiPath string, endpointName string) {
	start := time.Now()
	rec := &observability.DataProxyRecord{
		Endpoint:  endpointName,
		Caller:    observability.CallerFromGin(c),
		QueryType: "system_logs",
		Provider:  c.Query("provider"),
		Region:    c.Query("region"),
	}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	ctx := c.Request.Context()
	datasource := firstNonEmpty(c.Query("datasource"), c.Query("datasource_uid"), c.Query("datasourceUid"), c.Query("uid"))
	lokiURL, err := h.service.ResolveSystemLogsURL(datasource)
	if err != nil {
		rec.ErrCode = "SYSTEM_LOGS_DATASOURCE_NOT_FOUND"
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	proxyURL := stripSystemLogsRoutingParams(c.Request.URL)
	resp, err := h.service.ProxySystemRequest(
		ctx,
		lokiPath,
		proxyURL,
		c.Request.Method,
		c.Request.Body,
		c.Request.Header,
		lokiURL,
	)
	if err != nil {
		rec.ErrCode = "SYSTEM_LOGS_PROXY_FAILED"
		log.Error(ctx, "failed to proxy system logs request",
			zap.String("datasource", datasource),
			zap.String("provider", c.Query("provider")),
			zap.String("region", c.Query("region")),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
		return
	}
	defer resp.Body.Close()

	if err := h.writeLokiResponse(c, resp); err != nil {
		rec.ErrCode = "SYSTEM_LOGS_RESPONSE_COPY_FAILED"
		log.Error(ctx, "failed to copy system logs response", zap.Error(err))
	}
}

func (h *LokiHandler) proxyServerlessParentLogsRequest(c *gin.Context, lokiPath string, endpointName string) {
	start := time.Now()
	rec := &observability.DataProxyRecord{
		Endpoint:     endpointName,
		Caller:       observability.CallerFromGin(c),
		QueryType:    "serverless_parent_logs",
		DeployTypeV2: "serverless_parent",
	}
	defer func() { observability.RecordDataProxy(c, start, rec) }()

	ctx := c.Request.Context()
	info, err := service.ParseServerlessParentControlPlaneInfo(controlPlaneInfoFromRequest(c))
	if err != nil {
		rec.ErrCode = "SERVERLESS_PARENT_LOGS_CONTROL_PLANE_INFO_INVALID"
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}
	rec.Provider = info.Vendor
	rec.Region = info.Region

	datasource := firstNonEmpty(c.Query("datasource"), c.Query("datasource_uid"), c.Query("datasourceUid"), c.Query("uid"))
	lokiURL, err := h.service.ResolveServerlessParentLogsURL(datasource, info)
	if err != nil {
		rec.ErrCode = "SERVERLESS_PARENT_LOGS_DATASOURCE_NOT_FOUND"
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	proxyURL := stripServerlessParentLogsRoutingParams(c.Request.URL)
	proxyURL, err = h.service.BuildServerlessParentLogsProxyURL(lokiPath, proxyURL, info)
	if err != nil {
		rec.ErrCode = "SERVERLESS_PARENT_LOGS_QUERY_REWRITE_FAILED"
		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
		return
	}

	resp, err := h.service.ProxySystemRequest(
		ctx,
		lokiPath,
		proxyURL,
		c.Request.Method,
		c.Request.Body,
		c.Request.Header,
		lokiURL,
	)
	if err != nil {
		rec.ErrCode = "SERVERLESS_PARENT_LOGS_PROXY_FAILED"
		log.Error(ctx, "failed to proxy serverless parent logs request",
			zap.String("datasource", datasource),
			zap.String("provider", info.Vendor),
			zap.String("region", info.Region),
			zap.Error(err))
		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
		return
	}
	defer resp.Body.Close()

	if err := h.writeLokiResponse(c, resp); err != nil {
		rec.ErrCode = "SERVERLESS_PARENT_LOGS_RESPONSE_COPY_FAILED"
		log.Error(ctx, "failed to copy serverless parent logs response", zap.Error(err))
	}
}

func (h *LokiHandler) writeLokiResponse(c *gin.Context, resp *http.Response) error {
	for key, values := range resp.Header {
		for _, value := range values {
			c.Header(key, value)
		}
	}
	c.Header("X-Loki-Proxy", "clinic-data-service")
	c.Status(resp.StatusCode)
	_, err := io.Copy(c.Writer, resp.Body)
	return err
}

func stripSystemLogsRoutingParams(original *url.URL) *url.URL {
	copied := *original
	q := copied.Query()
	for _, key := range []string{"provider", "region", "datasource", "datasource_uid", "datasourceUid", "uid"} {
		q.Del(key)
	}
	copied.RawQuery = q.Encode()
	return &copied
}

func stripServerlessParentLogsRoutingParams(original *url.URL) *url.URL {
	copied := *stripSystemLogsRoutingParams(original)
	q := copied.Query()
	q.Del("control_plane_info")
	copied.RawQuery = q.Encode()
	return &copied
}

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