package handlers

import (
	"context"
	stderrors "errors"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/clustercontext"
	"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/ticdc"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type tiCDCMetadataService interface {
	List(context.Context, uint64, uint64, uint64, ticdc.Architecture) (*ticdc.ListResponse, error)
	Detail(context.Context, uint64, uint64, uint64, string, ticdc.Architecture) (*ticdc.DetailResponse, error)
}

type TiCDCHandler struct {
	ownership clustercontext.Repository
	service   tiCDCMetadataService
}

func NewTiCDCHandler(ownership clustercontext.Repository, service tiCDCMetadataService) *TiCDCHandler {
	return &TiCDCHandler{ownership: ownership, service: service}
}

func (h *TiCDCHandler) RegisterRoutes(r *gin.Engine) {
	internal := r.Group("/data-proxy/api/v1/orgs/:orgId/clusters/:clusterId/ticdc")
	internal.GET("/changefeeds", h.ListInternal)
	internal.GET("/changefeeds/:changefeedId", h.DetailInternal)

	public := r.Group("/api/v1/orgs/:orgId/data-proxy/clusters/:clusterId/ticdc")
	public.GET("/changefeeds", h.ListPublic)
	public.GET("/changefeeds/:changefeedId", h.DetailPublic)
}

// ListInternal returns changefeeds for a trusted internal request.
//
//	@Summary	List TiCDC changefeeds (internal)
//	@Tags		ticdc
//	@Produce	json
//	@Param		orgId		path		integer	true	"TiDB Cloud org ID"
//	@Param		clusterId	path		integer	true	"TiDB Cloud cluster ID"
//	@Success	200			{object}	ticdc.ListResponse
//	@Failure	400			{object}	errors.APIError
//	@Failure	403			{object}	errors.APIError
//	@Failure	404			{object}	errors.APIError
//	@Failure	502			{object}	errors.APIError
//	@Failure	503			{object}	errors.APIError
//	@Failure	504			{object}	errors.APIError
//	@Router		/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/ticdc/changefeeds [get]
func (h *TiCDCHandler) ListInternal(c *gin.Context) { h.list(c, false) }

// ListPublic returns Dedicated changefeeds after Gateway authorization.
//
//	@Summary	List TiCDC changefeeds
//	@Tags		ticdc
//	@Produce	json
//	@Param		orgId		path		integer	true	"TiDB Cloud org ID"
//	@Param		clusterId	path		integer	true	"TiDB Cloud cluster ID"
//	@Success	200			{object}	ticdc.ListResponse
//	@Failure	400			{object}	errors.APIError
//	@Failure	401			{object}	errors.APIError
//	@Failure	403			{object}	errors.APIError
//	@Failure	404			{object}	errors.APIError
//	@Failure	502			{object}	errors.APIError
//	@Failure	503			{object}	errors.APIError
//	@Failure	504			{object}	errors.APIError
//	@Security	BearerAuth
//	@Router		/api/v1/orgs/{orgId}/data-proxy/clusters/{clusterId}/ticdc/changefeeds [get]
func (h *TiCDCHandler) ListPublic(c *gin.Context) { h.list(c, true) }

// DetailInternal returns one changefeed after its source cluster is verified.
//
//	@Summary	Get TiCDC changefeed (internal)
//	@Tags		ticdc
//	@Produce	json
//	@Param		orgId			path		integer	true	"TiDB Cloud org ID"
//	@Param		clusterId		path		integer	true	"TiDB Cloud cluster ID"
//	@Param		changefeedId	path		string	true	"Changefeed ID"
//	@Success	200				{object}	ticdc.DetailResponse
//	@Failure	400				{object}	errors.APIError
//	@Failure	403				{object}	errors.APIError
//	@Failure	404				{object}	errors.APIError
//	@Failure	502				{object}	errors.APIError
//	@Failure	503				{object}	errors.APIError
//	@Failure	504				{object}	errors.APIError
//	@Router		/data-proxy/api/v1/orgs/{orgId}/clusters/{clusterId}/ticdc/changefeeds/{changefeedId} [get]
func (h *TiCDCHandler) DetailInternal(c *gin.Context) { h.detail(c, false) }

// DetailPublic returns one Dedicated changefeed after its source cluster is verified.
//
//	@Summary	Get TiCDC changefeed
//	@Tags		ticdc
//	@Produce	json
//	@Param		orgId			path		integer	true	"TiDB Cloud org ID"
//	@Param		clusterId		path		integer	true	"TiDB Cloud cluster ID"
//	@Param		changefeedId	path		string	true	"Changefeed ID"
//	@Success	200				{object}	ticdc.DetailResponse
//	@Failure	400				{object}	errors.APIError
//	@Failure	401				{object}	errors.APIError
//	@Failure	403				{object}	errors.APIError
//	@Failure	404				{object}	errors.APIError
//	@Failure	502				{object}	errors.APIError
//	@Failure	503				{object}	errors.APIError
//	@Failure	504				{object}	errors.APIError
//	@Security	BearerAuth
//	@Router		/api/v1/orgs/{orgId}/data-proxy/clusters/{clusterId}/ticdc/changefeeds/{changefeedId} [get]
func (h *TiCDCHandler) DetailPublic(c *gin.Context) { h.detail(c, true) }

func (h *TiCDCHandler) list(c *gin.Context, public bool) {
	start, record := beginTiCDCRequest(c, "ticdc_changefeed_list", public)
	record.QueryType = "list"
	defer func() { observability.RecordDataProxy(c, start, record) }()
	resolved, ok := h.authorizeAndResolve(c, record, public)
	if !ok {
		return
	}
	logger := log.WithContextL(c.Request.Context()).With(
		zap.String("operation", "ticdc_changefeed_list"),
		zap.String("architecture", string(resolved.architecture)),
		zap.Bool("public", public),
	)
	logger.Info("TiCDC metadata upstream request started")
	response, err := h.service.List(c.Request.Context(), resolved.orgID, resolved.projectID, resolved.clusterID, resolved.architecture)
	if err != nil {
		logger.Warn("TiCDC metadata upstream request failed", zap.String("status_class", tiCDCErrorClass(err)))
		writeTiCDCError(c, record, err)
		return
	}
	if response == nil {
		logger.Error("TiCDC metadata upstream response rejected", zap.String("reason", "nil_list_response"))
		writeTiCDCError(c, record, ticdc.ErrUpstream)
		return
	}
	logger.Info("TiCDC metadata request completed", zap.Int("result_count", len(response.Changefeeds)))
	c.JSON(http.StatusOK, response)
}

func (h *TiCDCHandler) detail(c *gin.Context, public bool) {
	start, record := beginTiCDCRequest(c, "ticdc_changefeed_detail", public)
	record.QueryType = "detail"
	defer func() { observability.RecordDataProxy(c, start, record) }()
	resolved, ok := h.authorizeAndResolve(c, record, public)
	if !ok {
		return
	}
	changefeedID := strings.TrimSpace(c.Param("changefeedId"))
	if changefeedID == "" || (resolved.architecture != ticdc.ArchitectureEssential && !isPositiveUint(changefeedID)) {
		record.ErrCode = "INVALID_CHANGEFEED_ID"
		log.WithContextL(c.Request.Context()).Info("TiCDC metadata request validation failed",
			zap.String("operation", "ticdc_changefeed_detail"), zap.String("reason", "invalid_changefeed_id"))
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "changefeedId is invalid for this cluster type", nil)
		return
	}
	logger := log.WithContextL(c.Request.Context()).With(
		zap.String("operation", "ticdc_changefeed_detail"),
		zap.String("architecture", string(resolved.architecture)),
		zap.Bool("public", public),
	)
	logger.Info("TiCDC metadata upstream request started")
	response, err := h.service.Detail(c.Request.Context(), resolved.orgID, resolved.projectID, resolved.clusterID, changefeedID, resolved.architecture)
	if err != nil {
		logger.Warn("TiCDC metadata upstream request failed", zap.String("status_class", tiCDCErrorClass(err)))
		writeTiCDCError(c, record, err)
		return
	}
	if response == nil || response.Changefeed == nil {
		logger.Error("TiCDC metadata upstream response rejected", zap.String("reason", "nil_detail_response"))
		writeTiCDCError(c, record, ticdc.ErrUpstream)
		return
	}
	logger.Info("TiCDC metadata request completed", zap.Int("result_count", 1))
	c.JSON(http.StatusOK, response)
}

type resolvedTiCDCRequest struct {
	orgID        uint64
	projectID    uint64
	clusterID    uint64
	architecture ticdc.Architecture
}

func (h *TiCDCHandler) authorizeAndResolve(
	c *gin.Context,
	record *observability.DataProxyRecord,
	public bool,
) (resolvedTiCDCRequest, bool) {
	logger := log.WithContextL(c.Request.Context()).With(
		zap.String("operation", "ticdc_authorize_and_resolve"),
		zap.Bool("public", public),
	)
	if h == nil || h.ownership == nil || h.service == nil {
		record.ErrCode = "TICDC_UNAVAILABLE"
		logger.Error("TiCDC metadata request rejected", zap.String("reason", "service_unavailable"))
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "TiCDC metadata service is unavailable", nil)
		return resolvedTiCDCRequest{}, false
	}
	orgID, err := parsePositiveUint(c.Param("orgId"))
	if err != nil {
		record.ErrCode = "INVALID_ORG_ID"
		logger.Info("TiCDC metadata request validation failed", zap.String("reason", "invalid_org_id"))
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "orgId must be a positive integer", nil)
		return resolvedTiCDCRequest{}, false
	}
	clusterID, err := parsePositiveUint(c.Param("clusterId"))
	if err != nil {
		record.ErrCode = "INVALID_CLUSTER_ID"
		logger.Info("TiCDC metadata request validation failed", zap.String("reason", "invalid_cluster_id"))
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "clusterId must be a positive integer", nil)
		return resolvedTiCDCRequest{}, false
	}

	clusterIDString := strconv.FormatUint(clusterID, 10)
	cluster, err := h.ownership.GetByOrgAndClusterID(c.Request.Context(), orgID, clusterIDString)
	if err != nil {
		logger.Warn("TiCDC metadata ownership resolution failed", zap.String("status_class", tiCDCErrorClass(err)))
		writeTiCDCOwnershipError(c, record, err)
		return resolvedTiCDCRequest{}, false
	}
	if cluster == nil || cluster.ClusterID != clusterIDString || cluster.OrgID != strconv.FormatUint(orgID, 10) {
		record.ErrCode = "OWNERSHIP_IDENTITY_MISMATCH"
		logger.Error("TiCDC metadata ownership response rejected", zap.String("reason", "identity_mismatch"))
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster ownership service returned an invalid response", nil)
		return resolvedTiCDCRequest{}, false
	}
	architecture, err := ticdc.ResolveArchitecture(cluster.DeployType, cluster.DeployTypeV2)
	if err != nil {
		record.ErrCode = "UNSUPPORTED_DEPLOY_TYPE"
		logger.Info("TiCDC metadata request rejected", zap.String("reason", "unsupported_deploy_type"))
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "TiCDC metadata is not supported for this cluster type", nil)
		return resolvedTiCDCRequest{}, false
	}
	if public && architecture != ticdc.ArchitectureDedicated {
		record.ErrCode = "UNSUPPORTED_PUBLIC_DEPLOY_TYPE"
		logger.Info("TiCDC metadata request authorization denied", zap.String("reason", "public_deploy_type"))
		writeResourceError(c, http.StatusForbidden, "PermissionDenied", "public TiCDC metadata supports Dedicated clusters only", nil)
		return resolvedTiCDCRequest{}, false
	}
	projectID, err := ticdc.ParseProjectID(cluster.DiagnosticContext.ProjectID)
	if err != nil {
		record.ErrCode = "INVALID_PROJECT_METADATA"
		logger.Error("TiCDC metadata canonical metadata rejected", zap.String("reason", "invalid_project_id"))
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "cluster metadata contains an invalid project ID", nil)
		return resolvedTiCDCRequest{}, false
	}
	record.Provider = strings.ToLower(strings.TrimSpace(cluster.Provider))
	record.Region = cluster.Region
	record.DeployTypeV2 = cluster.DeployTypeV2
	record.Engine = string(architecture)
	logger.Info("TiCDC metadata request authorized", zap.String("architecture", string(architecture)))
	return resolvedTiCDCRequest{orgID: orgID, projectID: projectID, clusterID: clusterID, architecture: architecture}, true
}

func beginTiCDCRequest(c *gin.Context, endpoint string, public bool) (time.Time, *observability.DataProxyRecord) {
	if !public {
		endpoint += "_internal"
	}
	return time.Now(), &observability.DataProxyRecord{Endpoint: endpoint, Caller: observability.CallerFromGin(c)}
}

func parsePositiveUint(value string) (uint64, error) {
	parsed, err := strconv.ParseUint(strings.TrimSpace(value), 10, 64)
	if err != nil || parsed == 0 {
		return 0, ticdc.ErrInvalidArgument
	}
	return parsed, nil
}

func isPositiveUint(value string) bool {
	_, err := parsePositiveUint(value)
	return err == nil
}

func writeTiCDCOwnershipError(c *gin.Context, record *observability.DataProxyRecord, err error) {
	switch {
	case stderrors.Is(err, clustercontext.ErrNotFound):
		record.ErrCode = "CLUSTER_NOT_FOUND"
		writeResourceError(c, http.StatusNotFound, "NotFound", "cluster not found", nil)
	case stderrors.Is(err, clustercontext.ErrTimeout):
		record.ErrCode = "METADATA_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "cluster metadata request timed out", nil)
	case stderrors.Is(err, clustercontext.ErrUnavailable):
		record.ErrCode = "METADATA_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "cluster metadata is temporarily unavailable", nil)
	case stderrors.Is(err, clustercontext.ErrUnauthenticated):
		record.ErrCode = "METADATA_UNAUTHENTICATED"
		writeResourceError(c, http.StatusUnauthorized, "Unauthenticated", "cluster metadata request was not authenticated", nil)
	case stderrors.Is(err, clustercontext.ErrPermissionDenied):
		record.ErrCode = "METADATA_FORBIDDEN"
		writeResourceError(c, http.StatusForbidden, "PermissionDenied", "cluster access was denied", nil)
	default:
		record.ErrCode = "METADATA_QUERY_FAILED"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "failed to resolve cluster metadata", nil)
	}
}

func writeTiCDCError(c *gin.Context, record *observability.DataProxyRecord, err error) {
	grpcCode := status.Code(err)
	switch {
	case stderrors.Is(err, context.Canceled), grpcCode == codes.Canceled:
		record.ErrCode = "REQUEST_CANCELED"
		writeResourceError(c, 499, "Canceled", "request was canceled", nil)
	case stderrors.Is(err, context.DeadlineExceeded), grpcCode == codes.DeadlineExceeded:
		record.ErrCode = "TICDC_TIMEOUT"
		writeResourceError(c, http.StatusGatewayTimeout, "DeadlineExceeded", "TiCDC metadata request timed out", nil)
	case stderrors.Is(err, ticdc.ErrNotFound), grpcCode == codes.NotFound:
		record.ErrCode = "CHANGEFEED_NOT_FOUND"
		writeResourceError(c, http.StatusNotFound, "NotFound", "changefeed not found", nil)
	case stderrors.Is(err, ticdc.ErrForbidden), grpcCode == codes.PermissionDenied:
		record.ErrCode = "TICDC_FORBIDDEN"
		writeResourceError(c, http.StatusForbidden, "PermissionDenied", "TiCDC metadata request was denied", nil)
	case stderrors.Is(err, ticdc.ErrInvalidArgument), grpcCode == codes.InvalidArgument:
		record.ErrCode = "INVALID_TICDC_REQUEST"
		writeResourceError(c, http.StatusBadRequest, "InvalidArgument", "invalid TiCDC metadata request", nil)
	case stderrors.Is(err, ticdc.ErrUnavailable), grpcCode == codes.Unavailable:
		record.ErrCode = "TICDC_UNAVAILABLE"
		writeResourceError(c, http.StatusServiceUnavailable, "Unavailable", "TiCDC metadata service is temporarily unavailable", nil)
	default:
		record.ErrCode = "TICDC_UPSTREAM_FAILURE"
		writeResourceError(c, http.StatusBadGateway, "BadGateway", "TiCDC metadata upstream request failed", nil)
	}
}

func tiCDCErrorClass(err error) string {
	if err == nil {
		return "success"
	}
	if code := status.Code(err); code != codes.Unknown {
		return strings.ToLower(code.String())
	}
	switch {
	case stderrors.Is(err, ticdc.ErrNotFound):
		return "not_found"
	case stderrors.Is(err, ticdc.ErrForbidden):
		return "forbidden"
	case stderrors.Is(err, ticdc.ErrUnavailable):
		return "unavailable"
	case stderrors.Is(err, context.DeadlineExceeded):
		return "deadline_exceeded"
	default:
		return "upstream_failure"
	}
}