package connector_inbound

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/rs/zerolog"
	zlog "github.com/rs/zerolog/log"
)

var (
	connectorInboundSchedulerClaimsInflight = promauto.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "nexuscrm_connector_inbound_scheduler_claims_inflight",
			Help: "Candidate claim calls currently executing, excluding event processing and empty queue polls.",
		},
		[]string{"work_type", "queue_class"},
	)
	connectorInboundSchedulerClaimDuration = promauto.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "nexuscrm_connector_inbound_scheduler_claim_duration_seconds",
			Help:    "Candidate claim call duration by bounded work type, queue class, and outcome.",
			Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600},
		},
		[]string{"work_type", "queue_class", "outcome"},
	)
)

const schedulerSlowClaimThreshold = 5 * time.Second

// IDs are diagnostic log fields, never metric labels. One timer per active
// candidate emits at most one warning, even if a database call never returns.
// This observer does not cancel, retry, or otherwise control the claim.
type schedulerClaimObservation struct {
	mu                         sync.Mutex
	done, activeWarningEmitted bool
	started                    time.Time
	timer                      *time.Timer
	logger                     zerolog.Logger
	workType, queueClass       string
	eventID, workID            string
	workerID, installedPackID  string
}

func startSchedulerClaimObservation(cfg WorkerConfig, candidate schedulerClaimCandidate) *schedulerClaimObservation {
	o := &schedulerClaimObservation{
		started: time.Now(), logger: zlog.Logger,
		workType: candidate.workType, queueClass: candidate.queueClass,
		workID: candidate.workID, workerID: cfg.WorkerID, installedPackID: cfg.PackInstallID,
	}
	if candidate.event != nil {
		o.eventID = candidate.event.ID
		if o.queueClass == "" {
			o.queueClass = candidate.event.QueueClass
		}
	}
	switch o.workType {
	case "connector_event", "workflow_action_apply":
	default:
		o.workType = "unknown"
	}
	switch o.queueClass {
	case "interactive", "default", "bulk":
	default:
		o.queueClass = "unknown"
	}
	connectorInboundSchedulerClaimsInflight.WithLabelValues(o.workType, o.queueClass).Inc()
	o.timer = time.AfterFunc(schedulerSlowClaimThreshold, o.warnStillActive)
	return o
}

func (o *schedulerClaimObservation) warnStillActive() {
	o.mu.Lock()
	defer o.mu.Unlock()
	if o.done || o.activeWarningEmitted {
		return
	}
	o.activeWarningEmitted = true
	o.warn("inflight", time.Since(o.started))
}

func (o *schedulerClaimObservation) finish(outcome string) {
	o.mu.Lock()
	if o.done {
		o.mu.Unlock()
		return
	}
	o.done = true
	o.timer.Stop()
	elapsed := time.Since(o.started)
	o.mu.Unlock()
	connectorInboundSchedulerClaimsInflight.WithLabelValues(o.workType, o.queueClass).Dec()
	connectorInboundSchedulerClaimDuration.WithLabelValues(o.workType, o.queueClass, outcome).Observe(elapsed.Seconds())
	if elapsed >= schedulerSlowClaimThreshold {
		o.warn(outcome, elapsed)
	}
}

func (o *schedulerClaimObservation) warn(outcome string, elapsed time.Duration) {
	o.logger.Warn().
		Str("event", "connector_inbound_claim_slow").
		Str("connector_event_id", o.eventID).
		Str("work_item_id", o.workID).
		Str("worker_id", o.workerID).
		Str("installed_pack_id", o.installedPackID).
		Str("work_type", o.workType).
		Str("queue_class", o.queueClass).
		Time("started_at", o.started.UTC()).
		Int64("elapsed_ms", elapsed.Milliseconds()).
		Str("outcome", outcome).
		Msg("Connector inbound candidate claim exceeded slow threshold")
}

func schedulerClaimOutcome(claimed bool, err error) string {
	switch {
	case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
		return "canceled"
	case err != nil:
		return "error"
	case claimed:
		return "claimed"
	default:
		return "cas_lost"
	}
}