package monthlyusagehistory

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"sync"
	"time"

	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
)

var errProducerDone = gerror.New("run production already closed")

func producerStateError(state zset.ProducerState) error {
	if state.Status == "completed" || state.Status == "completed_with_failures" {
		return ErrRunFinished
	}
	if state.ProducerDone {
		return errProducerDone
	}
	if state.Status != "producing" && state.Status != "running" {
		return gerror.Newf("unexpected producer run status %q", state.Status)
	}
	return nil
}

// produceChannels owns all per-attempt coordination. There is one worker per
// target, NOT a pool of four long-lived channel workers. SQL tokens and the
// cancellable admission gate never span sleeps or goroutine joins.
func (p *Producer) produceChannels(ctx context.Context, message model.TrafficUsageMonthlyHistoryStartMessage, scope *model.SyncScope, scanUpperID int64, startedAt time.Time, total *productionStats) error {
	scanner := p.scanner.(channelScanner)
	metadata, err := scanner.ListChannels(ctx)
	if err != nil {
		return gerror.Wrap(err, "initialize channel scan")
	}
	state, err := p.store.ReadProducerState(ctx, message.RunID, "")
	if err != nil {
		return err
	}
	if err = producerStateError(state); err != nil {
		return err
	}
	targets := make(map[string]channelMetadata, len(metadata))
	known := make(map[string]bool, len(metadata))
	for _, channel := range metadata {
		if channel.ChanKey == "" || !scope.Allows(channel.ChanKey) {
			continue
		}
		if known[channel.ChanKey] {
			return gerror.Newf("duplicate channel metadata %q", channel.ChanKey)
		}
		known[channel.ChanKey] = true
		reason := ""
		if channel.SyncEnabled {
			reason = "sync_disabled"
		} else if channel.TempKey == "" {
			reason = "missing_channel_template"
		}
		if reason != "" {
			total.skippedChannels++
			g.Log().Warningf(ctx, "event=monthly_usage_history_channel status=skipped runId=%s chanKey=%s reason=%s skipUnit=channel", message.RunID, channel.ChanKey, reason)
			continue
		}
		targets[channel.ChanKey] = channel
	}
	if scope != nil {
		for _, chanKey := range scope.Channels {
			if !known[chanKey] {
				total.skippedChannels++
				g.Log().Warningf(ctx, "event=monthly_usage_history_channel status=skipped runId=%s chanKey=%s reason=unknown_channel skipUnit=channel", message.RunID, chanKey)
			}
		}
	}
	g.Log().Infof(ctx, "event=monthly_usage_history_channel_scan status=started runId=%s targetChannels=%d scanUpperID=%d scanConcurrency=%d pageSize=%d skipUnit=channel", message.RunID, len(targets), scanUpperID, p.config.ScanConcurrency, p.config.PageSize)
	keys := make([]string, 0, len(targets))
	for key := range targets {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	finishObservation := p.startObservation(ctx, message.RunID, keys, startedAt)
	defer finishObservation()
	var (
		workers sync.WaitGroup
		slots   = make(chan struct{}, p.config.ScanConcurrency)
		gate    = make(chan struct{}, 1)
		metrics = newScanActivity()
		results = make([]*channelScanStats, len(keys))
	)
	pool, hasPool := scanner.(interface {
		poolWait(context.Context) (int64, int64)
	})
	var poolCount, poolNanos int64
	if hasPool {
		poolCount, poolNanos = pool.poolWait(ctx)
	}
	workCtx, cancel := context.WithCancelCause(ctx)
	defer cancel(nil)
	for i, key := range keys {
		stats := &channelScanStats{chanKey: key, lastSummary: time.Now()}
		results[i] = stats
		workers.Add(1)
		go func(channel channelMetadata, stats *channelScanStats) {
			defer workers.Done()
			scanErr := p.scanOneChannel(workCtx, scanner, message, channel, scanUpperID, startedAt, slots, gate, metrics, stats)
			if scanErr != nil {
				cancel(scanErr)
			}
			reason := stats.stopReason
			if reason == "" {
				reason = "scan_complete"
			}
			if scanErr != nil {
				reason = "cancelled"
			}
			p.observe(channel.ChanKey, reason, 0, false, true)
			stats.log(workCtx, message.RunID, "ended", startedAt, scanErr)
		}(targets[key], stats)
	}
	workers.Wait()
	var summary channelScanStats
	for _, stats := range results {
		total.pages += stats.scanPages
		total.scanned += stats.eligible
		total.eligible += stats.eligible
		total.enqueued += stats.added
		total.stoppedTasks += stats.rejected
		if stats.eligible > 0 {
			total.channels[stats.chanKey] = struct{}{}
		}
		summary.merge(stats)
	}
	metrics.log(ctx, message.RunID, &summary)
	if hasPool {
		count, nanos := pool.poolWait(ctx)
		g.Log().Infof(ctx, "event=monthly_usage_history_scan_pool runId=%s sharedPoolWaitCountDelta=%d sharedPoolWaitMsDelta=%d attribution=shared_pool", message.RunID, count-poolCount, (nanos-poolNanos)/int64(time.Millisecond))
	}
	return context.Cause(workCtx)
}

func acquireContext(ctx context.Context, token chan struct{}) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case token <- struct{}{}:
		if err := ctx.Err(); err != nil {
			<-token
			return err
		}
		return nil
	}
}

func (p *Producer) scanOneChannel(ctx context.Context, scanner channelScanner, message model.TrafficUsageMonthlyHistoryStartMessage, channel channelMetadata, scanUpperID int64, startedAt time.Time, slots, gate chan struct{}, metrics *scanActivity, stats *channelScanStats) error {
	for stats.cursor < scanUpperID {
		// Detect a blocked/closed run before querying, including empty channels.
		p.observe(channel.ChanKey, "redis_read", 0, false, false)
		state, err := p.store.ReadProducerState(ctx, message.RunID, channel.ChanKey)
		if err != nil {
			return err
		}
		if err = producerStateError(state); err != nil {
			return err
		}
		if state.Blocked {
			// No page was read: the unknown unscanned remainder has no card
			// count. Never turn it into a fabricated success/failure total.
			stats.stopReason = "blocked_unscanned_remainder"
			return nil
		}
		p.observe(channel.ChanKey, "sql_slot", 0, false, false)
		waitStart := time.Now()
		if err = acquireContext(ctx, slots); err != nil {
			return err
		}
		stats.slotWait += time.Since(waitStart)
		metrics.begin()
		p.observe(channel.ChanKey, "sql_query", 0, false, false)
		queryStart := time.Now()
		cards, scanErr := scanner.ScanChannel(ctx, channel.ChanKey, stats.cursor, p.config.PageSize)
		elapsed := time.Since(queryStart)
		metrics.end()
		<-slots // release immediately after results have been materialized
		stats.sqlTime += elapsed
		stats.sqlLatency.observe(elapsed)
		stats.scanPages++
		if scanErr != nil {
			return gerror.Wrapf(scanErr, "scan channel %s after %d", channel.ChanKey, stats.cursor)
		}
		if len(cards) > p.config.PageSize {
			return gerror.New("scanner exceeded page size")
		}
		if len(cards) == 0 {
			stats.stopReason = "empty_result"
			return nil
		}
		var (
			pageCursor = stats.cursor
			overUpper  bool
			tasks      = make([]zset.Task, 0, len(cards))
		)
		for i, card := range cards {
			if card.ChanKey != channel.ChanKey || card.ID <= pageCursor {
				return gerror.New("channel scanner returned unordered or foreign cards")
			}
			if card.ID > scanUpperID {
				stats.overUpper += int64(len(cards) - i)
				overUpper = true
				break
			}
			pageCursor = card.ID
			tasks = append(tasks, zset.Task{TaskID: zset.StableTaskID(message.RunID, message.Operation, card.ID), RunID: message.RunID, Operation: message.Operation, QueryMonth: message.QueryMonth, ChanKey: channel.ChanKey, TempKey: channel.TempKey, Operator: channel.Operator, CardID: card.ID, ICCID: card.ICCID, MSISDN: card.MSISDN, IMSI: card.IMSI, PackageCode: card.PackageCode})
		}
		p.observe(channel.ChanKey, "page_ready", len(tasks), false, false)
		stats.eligible += int64(len(tasks))
		blocked, err := p.admitChannelPage(ctx, message.RunID, tasks, startedAt, gate, stats)
		if err != nil {
			return err // page cursor is not committed, even after a partial write
		}
		if blocked {
			stats.stopReason = "blocked"
			return nil
		}
		stats.cursor = pageCursor
		stats.progress(ctx, message.RunID, startedAt)
		if overUpper || stats.cursor >= scanUpperID {
			stats.stopReason = "upper_bound"
			return nil
		}
		// Do not use a short page as EOF: query-time eligibility may change.
	}
	return ctx.Err()
}

// admitChannelPage admits only new TaskIDs, in capacity-sized slices. Each
// worker retains at most this page. No capacity reservation survives the gate.
func (p *Producer) admitChannelPage(ctx context.Context, runID string, tasks []zset.Task, startedAt time.Time, gate chan struct{}, stats *channelScanStats) (bool, error) {
	for len(tasks) > 0 {
		p.observe(stats.chanKey, "admission_gate", len(tasks), false, false)
		gateStart := time.Now()
		if err := acquireContext(ctx, gate); err != nil {
			return false, err
		}
		stats.gateWait += time.Since(gateStart)
		blocked, waitKind, remaining, err := p.admitChannelSlice(ctx, runID, tasks, startedAt, stats)
		<-gate
		if err != nil || blocked {
			return blocked, err
		}
		tasks = remaining
		p.observe(stats.chanKey, "page_ready", len(tasks), false, false)
		if waitKind != "" {
			p.observe(stats.chanKey, waitKind+"_capacity", len(tasks), false, false)
			waitStart := time.Now()
			err = sleepContext(ctx, p.config.PollInterval)
			if waitKind == "global" {
				stats.globalWait += time.Since(waitStart)
			} else {
				stats.channelWait += time.Since(waitStart)
			}
			stats.progress(ctx, runID, startedAt)
			if err != nil {
				return false, err
			}
		}
	}
	return false, ctx.Err()
}

// Caller holds the short admission gate; there is no sleep in this method.
func (p *Producer) admitChannelSlice(ctx context.Context, runID string, tasks []zset.Task, startedAt time.Time, stats *channelScanStats) (blocked bool, waitKind string, remaining []zset.Task, err error) {
	p.observe(stats.chanKey, "redis_read", len(tasks), false, false)
	readStart := time.Now()
	state, err := p.store.ReadProducerState(ctx, runID, stats.chanKey)
	if err != nil {
		return false, "", tasks, err
	}
	if err = producerStateError(state); err != nil {
		return false, "", tasks, err
	}
	existing, err := p.store.ExistingRunTasks(ctx, runID, tasks)
	stats.stateTime += time.Since(readStart)
	if err != nil {
		return false, "", tasks, err
	}
	pending := tasks[:0]
	for i, exists := range existing {
		if exists {
			stats.duplicates++
		} else {
			pending = append(pending, tasks[i])
		}
	}
	tasks = pending
	p.observe(stats.chanKey, "redis_read", len(tasks), false, false)
	if state.Blocked {
		stats.rejected += int64(len(tasks))
		return true, "", nil, nil
	}
	if len(tasks) == 0 {
		return false, "", nil, nil
	}
	space := int64(len(tasks))
	if limit := p.config.GlobalHighWaterMark; limit > 0 {
		space = min(space, limit-state.ActiveCount) // inflight is already included
		if space <= 0 {
			return false, "global", tasks, nil
		}
	}
	if limit := p.config.ChannelHighWaterMark; limit > 0 {
		key, keyErr := p.store.RoundActiveKey(runID, state.CurrentRound, stats.chanKey)
		if keyErr != nil {
			return false, "", tasks, keyErr
		}
		readStart = time.Now()
		size, sizeErr := p.store.SizeKey(ctx, key)
		stats.stateTime += time.Since(readStart)
		if sizeErr != nil {
			return false, "", tasks, sizeErr
		}
		space = min(space, limit-size)
		if space <= 0 {
			return false, "channel", tasks, nil
		}
	}
	batch := tasks[:int(space)]
	p.observe(stats.chanKey, "redis_write", len(tasks), false, false)
	enqueueStart := time.Now()
	added, err := p.store.EnqueueRunBatch(ctx, runID, stats.chanKey, batch, time.Now())
	stats.enqueueTime += time.Since(enqueueStart)
	stats.enqueueBatches++
	if err != nil {
		return false, "", tasks, err // uncertain responses require idempotent rescan
	}
	if added < 0 || added > len(batch) {
		return false, "", tasks, gerror.New("invalid enqueue count")
	}
	stats.added += int64(added)
	p.observe(stats.chanKey, "redis_write", len(tasks)-len(batch), added > 0, false)
	if added > 0 && !stats.firstEnqueueRecorded {
		stats.firstEnqueueRecorded = true
		stats.firstEnqueue = time.Since(startedAt)
		stats.log(ctx, runID, "first_enqueue", startedAt, nil)
	}
	if added < len(batch) {
		readStart = time.Now()
		confirmed, checkErr := p.store.ExistingRunTasks(ctx, runID, batch)
		if checkErr != nil {
			return false, "", tasks, checkErr
		}
		state, checkErr = p.store.ReadProducerState(ctx, runID, stats.chanKey)
		stats.stateTime += time.Since(readStart)
		if checkErr != nil {
			return false, "", tasks, checkErr
		}
		if checkErr = producerStateError(state); checkErr != nil {
			return false, "", tasks, checkErr
		}
		present := 0
		for _, exists := range confirmed {
			if exists {
				present++
			}
		}
		if present < added {
			return false, "", tasks, gerror.New("enqueue count exceeds confirmed task states")
		}
		stats.duplicates += int64(present - added)
		if state.Blocked {
			stats.rejected += int64(len(tasks) - present)
			return true, "", nil, nil
		}
		if present != len(batch) {
			return false, "", tasks, gerror.Newf("unexplained enqueue rejection: added=%d submitted=%d confirmed=%d", added, len(batch), present)
		}
	}
	return false, "", tasks[len(batch):], nil
}

// Retained only to keep errors.Is semantics visible at the call site.
func isProductionClosed(err error) bool { return errors.Is(err, errProducerDone) }

func durationMS(d time.Duration) float64 { return float64(d) / float64(time.Millisecond) }

// latencyHistogram is fixed-size. Reported percentiles are bucket upper bounds,
// never exact samples or a server-side SLA; the last bucket is unbounded.
var latencyBounds = [...]time.Duration{time.Millisecond, 2 * time.Millisecond, 5 * time.Millisecond, 10 * time.Millisecond, 20 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond, 200 * time.Millisecond, 500 * time.Millisecond, time.Second, 2 * time.Second, 5 * time.Second, 10 * time.Second, 30 * time.Second, time.Minute}

type latencyHistogram struct{ counts [16]int64 }

func (h *latencyHistogram) observe(d time.Duration) {
	i := 0
	for i < len(latencyBounds) && d > latencyBounds[i] {
		i++
	}
	h.counts[i]++
}
func (h *latencyHistogram) percentile(percent int64) string {
	var total, cumulative int64
	for _, count := range h.counts {
		total += count
	}
	if total == 0 {
		return "n/a"
	}
	target := (total*percent + 99) / 100
	for i, count := range h.counts {
		cumulative += count
		if cumulative >= target {
			if i == len(latencyBounds) {
				return ">60000"
			}
			return fmt.Sprintf("%.0f", durationMS(latencyBounds[i]))
		}
	}
	return "n/a"
}

type channelScanStats struct {
	firstEnqueueRecorded                                                                       bool
	chanKey, stopReason                                                                        string
	cursor, scanPages, eligible, overUpper, enqueueBatches, added, duplicates, rejected        int64
	sqlTime, slotWait, gateWait, globalWait, channelWait, stateTime, enqueueTime, firstEnqueue time.Duration
	sqlLatency                                                                                 latencyHistogram
	lastSummary                                                                                time.Time
}

func (s *channelScanStats) log(ctx context.Context, runID, status string, startedAt time.Time, err error) {
	errorText := ""
	if err != nil {
		errorText = err.Error()
	}
	g.Log().Infof(ctx, "event=monthly_usage_history_channel_scan status=%s runId=%s chanKey=%s cursor=%d scanPages=%d eligible=%d overUpper=%d enqueueBatches=%d added=%d duplicates=%d rejected=%d sqlMs=%.3f sqlP50UpperMs=%s sqlP95UpperMs=%s slotWaitMs=%.3f admissionWaitMs=%.3f globalCapacityWaitMs=%.3f channelCapacityWaitMs=%.3f stateReadMs=%.3f enqueueMs=%.3f firstEnqueueMs=%.3f wallMs=%d stopReason=%s error=%q", status, runID, s.chanKey, s.cursor, s.scanPages, s.eligible, s.overUpper, s.enqueueBatches, s.added, s.duplicates, s.rejected, durationMS(s.sqlTime), s.sqlLatency.percentile(50), s.sqlLatency.percentile(95), durationMS(s.slotWait), durationMS(s.gateWait), durationMS(s.globalWait), durationMS(s.channelWait), durationMS(s.stateTime), durationMS(s.enqueueTime), durationMS(s.firstEnqueue), time.Since(startedAt).Milliseconds(), s.stopReason, errorText)
}
func (s *channelScanStats) progress(ctx context.Context, runID string, startedAt time.Time) {
	if time.Since(s.lastSummary) >= time.Minute {
		s.lastSummary = time.Now()
		s.log(ctx, runID, "progress", startedAt, nil)
	}
}
func (s *channelScanStats) merge(other *channelScanStats) {
	s.scanPages += other.scanPages
	s.eligible += other.eligible
	s.overUpper += other.overUpper
	s.enqueueBatches += other.enqueueBatches
	s.added += other.added
	s.duplicates += other.duplicates
	s.rejected += other.rejected
	s.sqlTime += other.sqlTime
	s.slotWait += other.slotWait
	s.gateWait += other.gateWait
	s.globalWait += other.globalWait
	s.channelWait += other.channelWait
	s.stateTime += other.stateTime
	s.enqueueTime += other.enqueueTime
	for i, count := range other.sqlLatency.counts {
		s.sqlLatency.counts[i] += count
	}
}

type scanActivity struct {
	mu               sync.Mutex
	started, changed time.Time
	running, peak    int
	area             time.Duration
}

func newScanActivity() *scanActivity {
	now := time.Now()
	return &scanActivity{started: now, changed: now}
}
func (s *scanActivity) change(delta int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := time.Now()
	s.area += time.Duration(s.running) * now.Sub(s.changed)
	s.changed = now
	s.running += delta
	s.peak = max(s.peak, s.running)
}
func (s *scanActivity) begin() { s.change(1) }
func (s *scanActivity) end()   { s.change(-1) }
func (s *scanActivity) log(ctx context.Context, runID string, stats *channelScanStats) {
	wall := time.Since(s.started)
	g.Log().Infof(ctx, "event=monthly_usage_history_scan_summary runId=%s wallMs=%d sqlInFlightPeak=%d sqlInFlightAvg=%.4f scanPages=%d enqueueBatches=%d eligible=%d overUpper=%d added=%d duplicates=%d rejected=%d sqlCumulativeMs=%.3f sqlP50UpperMs=%s sqlP95UpperMs=%s slotWaitCumulativeMs=%.3f admissionWaitCumulativeMs=%.3f globalCapacityWaitCumulativeMs=%.3f channelCapacityWaitCumulativeMs=%.3f stateReadCumulativeMs=%.3f enqueueCumulativeMs=%.3f", runID, wall.Milliseconds(), s.peak, float64(s.area)/float64(wall), stats.scanPages, stats.enqueueBatches, stats.eligible, stats.overUpper, stats.added, stats.duplicates, stats.rejected, durationMS(stats.sqlTime), stats.sqlLatency.percentile(50), stats.sqlLatency.percentile(95), durationMS(stats.slotWait), durationMS(stats.gateWait), durationMS(stats.globalWait), durationMS(stats.channelWait), durationMS(stats.stateTime), durationMS(stats.enqueueTime))
}