package connector_events

import (
	"context"
	"time"

	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

type DrainThrottleState struct {
	State  string `json:"state"`
	Reason string `json:"reason,omitempty"`
}

// Keep the Event service's published response types and JSON field names
// stable while delegating their mutable lifecycle values to WorkItem.
type QueueBacklogStats struct {
	Total                   int64 `json:"total"`
	Pending                 int64 `json:"pending"`
	Due                     int64 `json:"due"`
	FailedDue               int64 `json:"failed_due"`
	FailedNotDue            int64 `json:"failed_not_due"`
	Processing              int64 `json:"processing"`
	StaleProcessing         int64 `json:"stale_processing"`
	Processed               int64 `json:"processed"`
	Dead                    int64 `json:"dead"`
	RetryAttempts           int64 `json:"retry_attempts"`
	OldestPendingAgeSeconds int64 `json:"oldest_pending_age_seconds"`
	OldestDueAgeSeconds     int64 `json:"oldest_due_age_seconds"`
	OldestFailedAgeSeconds  int64 `json:"oldest_failed_age_seconds"`
	OldestProcessingSeconds int64 `json:"oldest_processing_age_seconds"`
}

type QueueClassStats struct {
	Namespace       string            `json:"namespace"`
	InstalledPackID string            `json:"installed_pack_id"`
	QueueClass      string            `json:"queue_class"`
	Stats           QueueBacklogStats `json:"stats"`
}

type QueueStats struct {
	CollectedAt   time.Time          `json:"collected_at"`
	Global        QueueBacklogStats  `json:"global"`
	ByQueueClass  []QueueClassStats  `json:"by_queue_class"`
	DrainThrottle DrainThrottleState `json:"drain_throttle"`
}

func (s *Service) QueryQueueStats(ctx context.Context) (QueueStats, error) {
	return s.queryQueueStats(ctx, connector_workqueue.ConnectorQueueStatsAll)
}

func (s *Service) QueryLiveQueueStats(ctx context.Context) (QueueStats, error) {
	stats, err := s.queryQueueStats(ctx, connector_workqueue.ConnectorQueueStatsLive)
	if err != nil {
		return QueueStats{}, err
	}
	observeConnectorLiveQueueStats(stats.Global)
	return stats, nil
}

// QueryCheapLiveQueueStats refreshes low-cardinality queue gauges while
// avoiding the correlated lane-head claimability predicate.
func (s *Service) QueryCheapLiveQueueStats(ctx context.Context) (QueueStats, error) {
	stats, err := s.queryQueueStats(ctx, connector_workqueue.ConnectorQueueStatsCheapLive)
	if err != nil {
		return QueueStats{}, err
	}
	observeConnectorLiveQueueStats(stats.Global)
	return stats, nil
}

func (s *Service) queryQueueStats(
	ctx context.Context,
	mode connector_workqueue.ConnectorQueueStatsMode,
) (QueueStats, error) {
	now := s.now().UTC()
	byQueueClass, err := s.workqueue.QueryConnectorEventQueueStats(
		ctx,
		now,
		now.Add(-StaleProcessingSeconds*time.Second),
		mode,
	)
	if err != nil {
		return QueueStats{}, err
	}
	stats := QueueStats{
		CollectedAt: now,
		DrainThrottle: DrainThrottleState{
			State:  "unwired",
			Reason: "drain limiter snapshot provider not configured",
		},
	}
	for _, item := range byQueueClass {
		converted := QueueClassStats{
			Namespace:       item.Namespace,
			InstalledPackID: item.InstalledPackID,
			QueueClass:      item.QueueClass,
			Stats:           queueBacklogStatsFromWorkItem(item.Stats),
		}
		stats.ByQueueClass = append(stats.ByQueueClass, converted)
		mergeQueueBacklogStats(&stats.Global, converted.Stats)
	}
	return stats, nil
}

func queueBacklogStatsFromWorkItem(src connector_workqueue.QueueBacklogStats) QueueBacklogStats {
	return QueueBacklogStats{
		Total:                   src.Total,
		Pending:                 src.Pending,
		Due:                     src.Due,
		FailedDue:               src.FailedDue,
		FailedNotDue:            src.FailedNotDue,
		Processing:              src.Processing,
		StaleProcessing:         src.StaleProcessing,
		Processed:               src.Processed,
		Dead:                    src.Dead,
		RetryAttempts:           src.RetryAttempts,
		OldestPendingAgeSeconds: src.OldestPendingAgeSeconds,
		OldestDueAgeSeconds:     src.OldestDueAgeSeconds,
		OldestFailedAgeSeconds:  src.OldestFailedAgeSeconds,
		OldestProcessingSeconds: src.OldestProcessingSeconds,
	}
}

func mergeQueueBacklogStats(dst *QueueBacklogStats, src QueueBacklogStats) {
	dst.Total += src.Total
	dst.Pending += src.Pending
	dst.Due += src.Due
	dst.FailedDue += src.FailedDue
	dst.FailedNotDue += src.FailedNotDue
	dst.Processing += src.Processing
	dst.StaleProcessing += src.StaleProcessing
	dst.Processed += src.Processed
	dst.Dead += src.Dead
	dst.RetryAttempts += src.RetryAttempts
	dst.OldestPendingAgeSeconds = maxInt64(dst.OldestPendingAgeSeconds, src.OldestPendingAgeSeconds)
	dst.OldestDueAgeSeconds = maxInt64(dst.OldestDueAgeSeconds, src.OldestDueAgeSeconds)
	dst.OldestFailedAgeSeconds = maxInt64(dst.OldestFailedAgeSeconds, src.OldestFailedAgeSeconds)
	dst.OldestProcessingSeconds = maxInt64(dst.OldestProcessingSeconds, src.OldestProcessingSeconds)
}

func maxInt64(a, b int64) int64 {
	if b > a {
		return b
	}
	return a
}