package connectors

import (
	"context"
	"errors"
	"log"
	"net/http"
	"strconv"
	"time"

	"github.com/gin-gonic/gin"

	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/interfaces/rest/internal/hostctx"
)

// ConnectorEventsAdminService is the operator-facing subset of the
// connector_events service used by the admin replay surface.
type ConnectorEventsAdminService interface {
	GetDeadEvent(ctx context.Context, perms connector_events.PermissionChecker, eventID string) (*connector_events.ConnectorEvent, error)
	ListDeadEvents(ctx context.Context, perms connector_events.PermissionChecker, namespace string, page connector_events.Page) ([]connector_events.ConnectorEvent, error)
	ResolveActivePack(ctx context.Context, namespace string) (packInstallID string, err error)
	ReplayDeadEvent(ctx context.Context, perms connector_events.PermissionChecker, eventID string, packInstallID string, routeVersion int64) (newEventID string, err error)
}

// connectorEventsAdminWorkItemService is an additive capability implemented by
// the in-tree service. Keeping it separate preserves the published
// ConnectorEventsAdminService interface for pinned consumers and custom mocks.
type connectorEventsAdminWorkItemService interface {
	GetDeadWorkItem(ctx context.Context, perms connector_events.PermissionChecker, eventID string) (*connector_events.DeadWorkItem, error)
	ListDeadWorkItems(ctx context.Context, perms connector_events.PermissionChecker, namespace string, page connector_events.Page) ([]connector_events.DeadWorkItem, error)
}

// connectorEventsRouteResolver is the minimal subset of connector_inbound_routing.RouteResolver
// used by the admin handler.
type connectorEventsRouteResolver interface {
	ResolveFlow(ctx context.Context, namespace, packInstallID, eventKind string) (version int64, flowDeveloperName string, err error)
}

type connectorEventSystemPerms map[string]bool

func (p connectorEventSystemPerms) Has(key string) bool {
	return p[key]
}

// connectorEventAdminResponse is metadata-only: it deliberately OMITS the raw
// webhook Payload, LastError text, FailureClass, and FailureSummary, which can
// carry secrets, upstream response bodies, or tenant data when custom callers
// persist them. The list endpoint returns many rows at the read-permission
// level, so exposing caller-authored diagnostics would be a batch secret leak.
// Only fixed-shape counters, ownership metadata, and timestamps are returned.
type connectorEventAdminResponse struct {
	ID               string     `json:"id"`
	WorkID           string     `json:"workId,omitempty"`
	Namespace        string     `json:"namespace"`
	InstalledPackID  string     `json:"installedPackId"`
	EventKind        string     `json:"eventKind"`
	IdempotencyKey   string     `json:"idempotencyKey"`
	Status           string     `json:"status"`
	State            string     `json:"state"`
	RetryCount       int        `json:"retryCount"`
	FailureCount     int        `json:"failureCount"`
	LastErrorAt      *time.Time `json:"lastErrorAt,omitempty"`
	LastFailureAt    *time.Time `json:"lastFailureAt,omitempty"`
	TerminalAt       *time.Time `json:"terminalAt,omitempty"`
	ProcessedAt      *time.Time `json:"processedAt,omitempty"`
	ReplayedAt       *time.Time `json:"replayedAt,omitempty"`
	ReplayOfID       string     `json:"replayOfId,omitempty"`
	ReplayOfWorkID   string     `json:"replayOfWorkId,omitempty"`
	ExecutionOwner   string     `json:"executionOwner"`
	CreatedDate      time.Time  `json:"createdDate"`
	LastModifiedDate time.Time  `json:"lastModifiedDate"`
}

// ConnectorEventsAdminHandler serves the operator-facing dead-letter list and
// replay APIs through the WorkItem-owned execution projection.
type ConnectorEventsAdminHandler struct {
	svc    ConnectorEventsAdminService
	routes connectorEventsRouteResolver
}

// NewConnectorEventsAdminHandler constructs a handler with a service and a
// route resolver for the current active incarnation lookup during replay.
func NewConnectorEventsAdminHandler(svc ConnectorEventsAdminService, routes connectorEventsRouteResolver) *ConnectorEventsAdminHandler {
	return &ConnectorEventsAdminHandler{svc: svc, routes: routes}
}

// ListDeadEvents handles GET /:namespace/events/dead.
func (h *ConnectorEventsAdminHandler) ListDeadEvents(c *gin.Context) {
	namespace := c.Param("namespace")

	limit, ok := parseConnectorEventsLimit(c)
	if !ok {
		return
	}
	offset, ok := parseConnectorEventsOffset(c)
	if !ok {
		return
	}

	ctx := c.Request.Context()
	perms := systemPermissionsFromContext(c)
	page := connector_events.Page{Limit: limit, Offset: offset}
	var out []connectorEventAdminResponse
	if workItems, ok := h.svc.(connectorEventsAdminWorkItemService); ok {
		items, err := workItems.ListDeadWorkItems(ctx, perms, namespace, page)
		if err != nil {
			respondConnectorEventsAdminError(c, "ListDeadEvents", namespace, err)
			return
		}
		out = make([]connectorEventAdminResponse, 0, len(items))
		for _, item := range items {
			out = append(out, connectorDeadWorkItemToAdminResponse(item))
		}
	} else {
		events, err := h.svc.ListDeadEvents(ctx, perms, namespace, page)
		if err != nil {
			respondConnectorEventsAdminError(c, "ListDeadEvents", namespace, err)
			return
		}
		out = make([]connectorEventAdminResponse, 0, len(events))
		for _, event := range events {
			out = append(out, connectorEventToAdminResponse(event))
		}
	}

	c.JSON(http.StatusOK, gin.H{
		"namespace": namespace,
		"limit":     limit,
		"offset":    offset,
		"events":    out,
	})
}

// ReplayDeadEvent handles POST /events/:id/replay.
// It performs a 4-step resolve: load effective dead execution state → ResolveActivePack →
// ResolveFlow → ReplayDeadEvent, so the replay is always bound to the
// current active pack incarnation and route version.
func (h *ConnectorEventsAdminHandler) ReplayDeadEvent(c *gin.Context) {
	ctx := c.Request.Context()
	perms := systemPermissionsFromContext(c)
	eventID := c.Param("id")

	// Step 1: prefer the WorkItem-owned view when the service advertises the
	// additive capability. Pinned services and custom mocks retain the
	// published Event-shaped interface and compatibility behavior.
	var namespace, eventKind string
	if workItems, ok := h.svc.(connectorEventsAdminWorkItemService); ok {
		item, err := workItems.GetDeadWorkItem(ctx, perms, eventID)
		if err != nil {
			respondConnectorEventsAdminError(c, "GetDeadWorkItem", eventID, err)
			return
		}
		namespace = item.Namespace
		eventKind = item.EventKind
	} else {
		event, err := h.svc.GetDeadEvent(ctx, perms, eventID)
		if err != nil {
			respondConnectorEventsAdminError(c, "GetDeadEvent", eventID, err)
			return
		}
		namespace = event.Namespace
		eventKind = event.EventKind
	}

	// Step 2: resolve the current active pack incarnation.
	packInstallID, err := h.svc.ResolveActivePack(ctx, namespace)
	if err != nil {
		respondConnectorEventsAdminError(c, "ResolveActivePack", namespace, err)
		return
	}

	// Step 3: resolve the current route version. Never pass 0 to ReplayDeadEvent.
	routeVersion, _, err := h.routes.ResolveFlow(ctx, namespace, packInstallID, eventKind)
	if err != nil {
		respondConnectorEventsAdminError(c, "ResolveFlow", namespace, err)
		return
	}

	// Step 4: create the replay row bound to the fresh provider incarnation +
	// route version. Cross-pack companion note: ReplayDeadEvent intentionally
	// copies the dead event's enqueue-time companion snapshot verbatim
	// (CompanionPackInstallId / CompanionFlowVersionId) per the #1777 design
	// (see service.go ReplayDeadEvent). Staleness is caught at drain by the
	// worker's companion incarnation re-check (fail-closed), so the admin path
	// neither can (the signature has no companion parameter) nor should
	// re-resolve the companion here.
	newID, err := h.svc.ReplayDeadEvent(ctx, perms, eventID, packInstallID, routeVersion)
	if err != nil {
		respondConnectorEventsAdminError(c, "ReplayDeadEvent", eventID, err)
		return
	}

	c.JSON(http.StatusOK, gin.H{
		"eventId": newID,
	})
}

// RegisterReadRoutes mounts the read-only dead-event list endpoint.
// Callers should gate the group on PermViewConnectorEvents before passing it.
func (h *ConnectorEventsAdminHandler) RegisterReadRoutes(group *gin.RouterGroup) {
	group.GET("/:namespace/events/dead", h.ListDeadEvents)
}

// RegisterReplayRoutes mounts the replay endpoint.
// Callers should gate the group on PermReplayConnectorEvents before passing it.
func (h *ConnectorEventsAdminHandler) RegisterReplayRoutes(group *gin.RouterGroup) {
	group.POST("/events/:id/replay", h.ReplayDeadEvent)
}

func connectorDeadWorkItemToAdminResponse(item connector_events.DeadWorkItem) connectorEventAdminResponse {
	executionOwner := "work_item"
	if item.LegacyEvent {
		executionOwner = "legacy_event"
	}
	return connectorEventAdminResponse{
		ID:               item.EventID,
		WorkID:           item.WorkID,
		Namespace:        item.Namespace,
		InstalledPackID:  item.InstalledPackID,
		EventKind:        item.EventKind,
		IdempotencyKey:   item.IdempotencyKey,
		Status:           item.State,
		State:            item.State,
		RetryCount:       item.FailureCount,
		FailureCount:     item.FailureCount,
		LastErrorAt:      item.LastFailureAt,
		LastFailureAt:    item.LastFailureAt,
		TerminalAt:       item.TerminalAt,
		ProcessedAt:      item.ProcessedAt,
		ReplayedAt:       item.ReplayedAt,
		ReplayOfID:       item.ReplayOfEventID,
		ReplayOfWorkID:   item.ReplayOfWorkID,
		ExecutionOwner:   executionOwner,
		CreatedDate:      item.EventCreatedDate,
		LastModifiedDate: item.LastModifiedDate,
	}
}

func connectorEventToAdminResponse(event connector_events.ConnectorEvent) connectorEventAdminResponse {
	return connectorEventAdminResponse{
		ID:               event.ID,
		Namespace:        event.Namespace,
		InstalledPackID:  event.InstalledPackID,
		EventKind:        event.EventKind,
		IdempotencyKey:   event.IdempotencyKey,
		Status:           event.Status,
		State:            event.Status,
		RetryCount:       event.RetryCount,
		FailureCount:     event.RetryCount,
		LastErrorAt:      event.LastErrorAt,
		LastFailureAt:    event.LastErrorAt,
		ProcessedAt:      event.ProcessedAt,
		ReplayedAt:       event.ReplayedAt,
		ReplayOfID:       event.ReplayOfID,
		ExecutionOwner:   "legacy_event",
		CreatedDate:      event.CreatedDate,
		LastModifiedDate: event.LastModifiedDate,
	}
}

// parseConnectorEventsLimit parses the "limit" query param (default 50, cap 500).
// Returns false and writes a 400 response when the value is present but invalid.
func parseConnectorEventsLimit(c *gin.Context) (int, bool) {
	const defaultLimit = 50
	const maxLimit = 500
	raw := c.Query("limit")
	if raw == "" {
		return defaultLimit, true
	}
	n, err := strconv.Atoi(raw)
	if err != nil || n <= 0 {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid limit"})
		return 0, false
	}
	if n > maxLimit {
		n = maxLimit
	}
	return n, true
}

// parseConnectorEventsOffset parses the "offset" query param (default 0).
// Returns false and writes a 400 response when the value is present but invalid.
func parseConnectorEventsOffset(c *gin.Context) (int, bool) {
	raw := c.Query("offset")
	if raw == "" {
		return 0, true
	}
	n, err := strconv.Atoi(raw)
	if err != nil || n < 0 {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid offset"})
		return 0, false
	}
	return n, true
}

func systemPermissionsFromContext(c *gin.Context) connector_events.PermissionChecker {
	perms := hostctx.SystemPermissions(c)
	if perms == nil {
		return connectorEventSystemPerms{}
	}
	return connectorEventSystemPerms(perms)
}

func respondConnectorEventsAdminError(c *gin.Context, operation, target string, err error) bool {
	switch {
	case errors.Is(err, connector_events.ErrPermissionDenied):
		c.JSON(http.StatusForbidden, gin.H{"error": "forbidden"})
		return true
	case errors.Is(err, connector_events.ErrEventNotFound):
		c.JSON(http.StatusNotFound, gin.H{"error": "connector event not found"})
		return true
	case errors.Is(err, connector_events.ErrEventNotDead):
		c.JSON(http.StatusConflict, gin.H{"error": "connector event is not in dead state"})
		return true
	case errors.Is(err, connector_events.ErrReplayBlocked):
		c.JSON(http.StatusConflict, gin.H{"error": "replay is blocked by an active source projection"})
		return true
	case errors.Is(err, connector_events.ErrInactivePack):
		c.JSON(http.StatusNotFound, gin.H{"error": "connector pack not available"})
		return true
	case errors.Is(err, connector_events.ErrInvalidReplayRequest):
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid replay request"})
		return true
	case errors.Is(err, connector_inbound_routing.ErrRouteNotDeclared):
		c.JSON(http.StatusConflict, gin.H{"error": "no active route declared for event kind"})
		return true
	case errors.Is(err, connector_events.ErrPackIncarnationChanged):
		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "pack incarnation changed; retry"})
		return true
	case errors.Is(err, connector_admission.ErrIngressQuiesced), errors.Is(err, connector_admission.ErrIngressSchemaUnavailable):
		var quiesced *connector_admission.QuiescedError
		if errors.As(err, &quiesced) && quiesced.Fence.RetryAfterSeconds > 0 {
			c.Header("Retry-After", strconv.Itoa(quiesced.Fence.RetryAfterSeconds))
		}
		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "connector ingress temporarily unavailable; retry"})
		return true
	default:
		log.Printf("connector events admin: %s target=%q internal error: %v", operation, target, err)
		c.JSON(http.StatusInternalServerError, gin.H{"error": "internal server error"})
		return true
	}
}