package runtimeworker

import (
	"sync"
	"sync/atomic"
	"time"
)

const (
	workerExecutionClassPayload  = "payload"
	workerExecutionClassMetadata = "metadata"

	startupStageDatabase                  = "database"
	startupStagePermissionState           = "permission_state"
	startupStageCatalogBind               = "catalog_bind"
	startupStageMutationGraphBuild        = "mutation_graph_build"
	startupStageMutationGraphPrepare      = "mutation_graph_prepare"
	startupStageDurableProducerActivation = "durable_producer_activation"
	startupStageRolePrepare               = "role_prepare"
)

var workerStartupStages = []string{
	startupStageDatabase,
	startupStagePermissionState,
	startupStageCatalogBind,
	startupStageMutationGraphBuild,
	startupStageMutationGraphPrepare,
	startupStageDurableProducerActivation,
	startupStageRolePrepare,
}

type workerStartupSnapshot struct {
	Duration time.Duration
	Stages   map[string]time.Duration
}

type workerStartupObservation struct {
	started time.Time

	mu        sync.RWMutex
	completed time.Time
	stages    map[string]time.Duration
}

func newWorkerStartupObservation() *workerStartupObservation {
	return &workerStartupObservation{started: time.Now(), stages: make(map[string]time.Duration)}
}

func (o *workerStartupObservation) observe(stage string, started time.Time) {
	if o == nil {
		return
	}
	o.record(stage, time.Since(started))
}

func (o *workerStartupObservation) record(stage string, duration time.Duration) {
	if o == nil {
		return
	}
	o.mu.Lock()
	o.stages[stage] += duration
	o.mu.Unlock()
}

func (o *workerStartupObservation) complete() {
	if o == nil {
		return
	}
	o.mu.Lock()
	if o.completed.IsZero() {
		o.completed = time.Now()
	}
	o.mu.Unlock()
}

func (o *workerStartupObservation) snapshot() workerStartupSnapshot {
	if o == nil {
		return workerStartupSnapshot{}
	}
	o.mu.RLock()
	defer o.mu.RUnlock()
	completed := o.completed
	if completed.IsZero() {
		completed = time.Now()
	}
	stages := make(map[string]time.Duration, len(o.stages))
	for stage, duration := range o.stages {
		stages[stage] = duration
	}
	return workerStartupSnapshot{
		Duration: completed.Sub(o.started), Stages: stages,
	}
}

type authorityCheckSnapshot struct {
	Checks      uint64
	Failures    uint64
	Duration    time.Duration
	LastSuccess time.Time
}

type workLoopSnapshot struct {
	PollChecks    uint64
	ConfirmedIdle uint64
	HintFailures  uint64
	ClaimAttempts uint64
	EmptyClaims   uint64
	Claimed       uint64
	PollDuration  time.Duration
	ClaimDuration time.Duration
}

// workLoopObservation describes existing admission and authoritative work
// attempts. It never controls whether a hint or claim runs.
type workLoopObservation struct {
	pollChecks        atomic.Uint64
	confirmedIdle     atomic.Uint64
	hintFailures      atomic.Uint64
	claimAttempts     atomic.Uint64
	emptyClaims       atomic.Uint64
	claimed           atomic.Uint64
	pollDurationNano  atomic.Int64
	claimDurationNano atomic.Int64
}

func (o *workLoopObservation) observePoll(started time.Time, pending bool, err error) {
	if o == nil {
		return
	}
	o.pollChecks.Add(1)
	o.pollDurationNano.Add(time.Since(started).Nanoseconds())
	if err != nil {
		o.hintFailures.Add(1)
		return
	}
	if !pending {
		o.confirmedIdle.Add(1)
	}
}

func (o *workLoopObservation) observeClaim(started time.Time, claimed uint64, empty bool) {
	if o == nil {
		return
	}
	o.claimAttempts.Add(1)
	o.claimDurationNano.Add(time.Since(started).Nanoseconds())
	if empty {
		o.emptyClaims.Add(1)
	}
	if claimed > 0 {
		o.claimed.Add(claimed)
	}
}

func (o *workLoopObservation) snapshot() workLoopSnapshot {
	if o == nil {
		return workLoopSnapshot{}
	}
	return workLoopSnapshot{
		PollChecks: o.pollChecks.Load(), ConfirmedIdle: o.confirmedIdle.Load(),
		HintFailures: o.hintFailures.Load(), ClaimAttempts: o.claimAttempts.Load(),
		EmptyClaims: o.emptyClaims.Load(), Claimed: o.claimed.Load(),
		PollDuration:  time.Duration(o.pollDurationNano.Load()),
		ClaimDuration: time.Duration(o.claimDurationNano.Load()),
	}
}

// authorityCheckObservation contains only process-local counters. The metrics
// collector never invokes the durable check it describes.
type authorityCheckObservation struct {
	checks       atomic.Uint64
	failures     atomic.Uint64
	durationNano atomic.Int64
	lastSuccess  atomic.Int64
}

func (o *authorityCheckObservation) observe(started time.Time, success bool) {
	if o == nil {
		return
	}
	o.checks.Add(1)
	o.durationNano.Add(time.Since(started).Nanoseconds())
	if !success {
		o.failures.Add(1)
		return
	}
	o.lastSuccess.Store(time.Now().UnixNano())
}

func (o *authorityCheckObservation) snapshot() authorityCheckSnapshot {
	if o == nil {
		return authorityCheckSnapshot{}
	}
	result := authorityCheckSnapshot{
		Checks: o.checks.Load(), Failures: o.failures.Load(),
		Duration: time.Duration(o.durationNano.Load()),
	}
	if value := o.lastSuccess.Load(); value > 0 {
		result.LastSuccess = time.Unix(0, value)
	}
	return result
}