package serverruntime

import (
	"context"
	"database/sql"
	"sync"
	"time"

	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
)

const (
	connectorQueueObservationSourceName = "connector_event"
	connectorQueueObservationTimeout    = 2 * time.Second
)

type connectorQueueStatsQuery func(context.Context, *sql.DB) (connector_events.QueueStats, error)

// connectorQueueObservation is the API process's single owner for the last
// exact Connector queue observation. The database remains authoritative; this
// cache exists only so Prometheus and ops endpoints never query it themselves.
type connectorQueueObservation struct {
	mu          sync.RWMutex
	query       connectorQueueStatsQuery
	stats       connector_events.QueueStats
	available   bool
	stale       bool
	lastSuccess time.Time
}

func newConnectorQueueObservation() *connectorQueueObservation {
	return &connectorQueueObservation{
		query: queryConnectorLiveQueueStats,
		stale: true,
	}
}

func queryConnectorLiveQueueStats(ctx context.Context, db *sql.DB) (connector_events.QueueStats, error) {
	return connector_events.NewService(db, nil, nil).QueryCheapLiveQueueStats(ctx)
}

func (o *connectorQueueObservation) initialize() {
	connector_events.InitializeLiveQueueMetrics()
}

func (o *connectorQueueObservation) refresh(ctx context.Context, db *sql.DB) error {
	stats, err := o.query(ctx, db)
	if err != nil {
		o.mu.Lock()
		o.stale = true
		o.mu.Unlock()
		return err
	}
	lastSuccess := stats.CollectedAt.UTC()
	if lastSuccess.IsZero() {
		lastSuccess = time.Now().UTC()
	}
	o.mu.Lock()
	o.stats = cloneConnectorQueueStats(stats)
	o.available = true
	o.stale = false
	o.lastSuccess = lastSuccess
	o.mu.Unlock()
	return nil
}

func (o *connectorQueueObservation) ConnectorQueueSnapshot() (
	connector_events.QueueStats,
	bool,
	bool,
	time.Time,
) {
	if o == nil {
		return connector_events.QueueStats{}, false, true, time.Time{}
	}
	o.mu.RLock()
	defer o.mu.RUnlock()
	return cloneConnectorQueueStats(o.stats), o.available, o.stale, o.lastSuccess
}

func cloneConnectorQueueStats(stats connector_events.QueueStats) connector_events.QueueStats {
	cloned := stats
	cloned.ByQueueClass = append([]connector_events.QueueClassStats(nil), stats.ByQueueClass...)
	return cloned
}